package main import ( "bytes" "context" "encoding/json" "errors" "fmt" "io" "log" "net/http" "net/url" "os" "os/exec" "os/signal" "path/filepath" "sort" "strconv" "strings" "sync" "syscall" "time" ) const skillProtocol = "jarvis.skill.v1" const workerProtocol = "jarvis.skill.worker.v1" const invokeProtocol = "jarvis.skill.invoke.v1" type runtimeSpec struct { Type string `json:"type"` Command string `json:"command"` Args []string `json:"args,omitempty"` TimeoutMS int `json:"timeout_ms,omitempty"` Env map[string]string `json:"env,omitempty"` EnvFrom []string `json:"env_from,omitempty"` } type permissions struct { Network bool `json:"network,omitempty"` SystemExec bool `json:"system_exec,omitempty"` } type actionManifest struct { Name string `json:"name"` ToolName string `json:"tool_name,omitempty"` Description string `json:"description"` InputSchema map[string]any `json:"input_schema"` OutputSchema map[string]any `json:"output_schema,omitempty"` Mutates bool `json:"mutates,omitempty"` RequiresConfirmation bool `json:"requires_confirmation,omitempty"` Triggers []string `json:"triggers,omitempty"` } type manifest struct { Protocol string `json:"protocol"` ID string `json:"id"` Name string `json:"name"` Version string `json:"version"` Description string `json:"description"` Enabled *bool `json:"enabled,omitempty"` Runtime runtimeSpec `json:"runtime"` Permissions permissions `json:"permissions,omitempty"` Actions []actionManifest `json:"actions"` } type skillRef struct { Manifest manifest Action actionManifest Dir string } type publicSkill struct { Protocol string `json:"protocol"` ID string `json:"id"` Name string `json:"name"` Version string `json:"version"` Description string `json:"description,omitempty"` Actions []actionManifest `json:"actions"` } type invokeContext struct { TraceID string `json:"trace_id,omitempty"` Now string `json:"now"` Timezone string `json:"timezone"` } type invokeRequest struct { Protocol string `json:"protocol"` RequestID string `json:"request_id"` SkillID string `json:"skill_id"` Action string `json:"action"` Input map[string]any `json:"input"` Context invokeContext `json:"context"` } type wireError struct { Code string `json:"code"` Message string `json:"message"` } type invokeResponse struct { Protocol string `json:"protocol"` Success bool `json:"success"` Data any `json:"data,omitempty"` Message string `json:"message,omitempty"` Error *wireError `json:"error,omitempty"` Warnings []string `json:"warnings,omitempty"` Mutated *bool `json:"mutated,omitempty"` Entity any `json:"entity,omitempty"` } type worker struct { mu sync.RWMutex skills map[string]manifest actions map[string]skillRef accessToken string workerID string lease time.Duration master string enrollment string name string runtime string runtimeVersion string publicURL string skillsDir string workDir string allowSystemExec bool maxOutput int client *http.Client } func main() { w := &worker{master: trimURL(env("JARVIS_MASTER_URL", "http://jarvis:8080")), enrollment: strings.TrimSpace(os.Getenv("JARVIS_ENROLLMENT_TOKEN")), name: env("JARVIS_WORKER_NAME", "skill-worker"), runtime: env("JARVIS_WORKER_RUNTIME", "generic"), runtimeVersion: env("JARVIS_WORKER_RUNTIME_VERSION", ""), publicURL: trimURL(env("JARVIS_WORKER_PUBLIC_URL", "http://localhost:8090")), skillsDir: env("JARVIS_SKILLS_DIR", "/skills"), allowSystemExec: envBool("JARVIS_WORKER_ALLOW_SYSTEM_EXEC", true), maxOutput: envInt("JARVIS_WORKER_MAX_OUTPUT_KB", 1024) * 1024, client: &http.Client{Timeout: 15 * time.Second}, skills: map[string]manifest{}, actions: map[string]skillRef{}, lease: 30 * time.Second} if err := w.reload(); err != nil { log.Printf("Skill load warning: %v", err) } mux := http.NewServeMux() mux.HandleFunc("GET /health", w.health) mux.HandleFunc("GET /v1/skills", w.listSkills) mux.HandleFunc("POST /v1/invoke", w.invoke) mux.HandleFunc("POST /v1/reload", w.reloadHTTP) srv := &http.Server{Addr: env("JARVIS_WORKER_ADDR", ":8090"), Handler: mux, ReadHeaderTimeout: 5 * time.Second, ReadTimeout: 60 * time.Second, WriteTimeout: 180 * time.Second} go func() { log.Printf("JARVIS Skill Worker %s runtime=%s listening %s", w.name, w.runtime, srv.Addr) if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed { log.Fatal(err) } }() ctx, cancel := context.WithCancel(context.Background()) go w.controlLoop(ctx) ch := make(chan os.Signal, 1) signal.Notify(ch, syscall.SIGINT, syscall.SIGTERM) <-ch cancel() w.deregister() sd, sdCancel := context.WithTimeout(context.Background(), 5*time.Second) defer sdCancel() _ = srv.Shutdown(sd) } func (w *worker) controlLoop(ctx context.Context) { for { if ctx.Err() != nil { return } if err := w.enrollAndRegister(ctx); err != nil { log.Printf("Enrollment: %v", err) if !sleepCtx(ctx, 5*time.Second) { return } continue } break } for { w.mu.RLock() lease := w.lease w.mu.RUnlock() wait := lease / 3 if wait < 5*time.Second { wait = 5 * time.Second } if !sleepCtx(ctx, wait) { return } if err := w.heartbeat(ctx); err != nil { log.Printf("Heartbeat: %v; erneutes Enrollment", err) w.mu.Lock() w.accessToken = "" w.workerID = "" w.mu.Unlock() for { if ctx.Err() != nil { return } if err := w.enrollAndRegister(ctx); err == nil { break } if !sleepCtx(ctx, 5*time.Second) { return } } } } } func sleepCtx(ctx context.Context, d time.Duration) bool { select { case <-ctx.Done(): return false case <-time.After(d): return true } } func (w *worker) enrollAndRegister(ctx context.Context) error { if w.enrollment == "" { return errors.New("JARVIS_ENROLLMENT_TOKEN fehlt") } in := map[string]any{"protocol": workerProtocol, "enrollment_token": w.enrollment, "worker": map[string]any{"name": w.name, "runtime": w.runtime, "runtime_version": w.runtimeVersion, "endpoint": w.publicURL, "labels": map[string]string{"runtime": w.runtime}}} var out struct { Protocol string `json:"protocol"` WorkerID string `json:"worker_id"` AccessToken string `json:"access_token"` LeaseSeconds int `json:"lease_seconds"` } if err := w.doJSON(ctx, http.MethodPost, w.master+"/api/mesh/enroll", "", in, &out); err != nil { return err } if out.WorkerID == "" || out.AccessToken == "" { return errors.New("Master lieferte keine Worker-ID/Token") } w.mu.Lock() w.workerID = out.WorkerID w.accessToken = out.AccessToken if out.LeaseSeconds > 0 { w.lease = time.Duration(out.LeaseSeconds) * time.Second } w.mu.Unlock() return w.register(ctx) } func (w *worker) register(ctx context.Context) error { w.mu.RLock() idv, tok := w.workerID, w.accessToken skills := w.publicSkillsLocked() w.mu.RUnlock() in := map[string]any{"protocol": workerProtocol, "skills": skills} var out map[string]any if err := w.doJSON(ctx, http.MethodPut, w.master+"/api/mesh/workers/"+url.PathEscape(idv)+"/skills", tok, in, &out); err != nil { return err } if ok, exists := out["ok"].(bool); exists && !ok { b, _ := json.Marshal(out["issues"]) return fmt.Errorf("Master hat Skill-Registrierung abgelehnt: %s", string(b)) } return nil } func (w *worker) heartbeat(ctx context.Context) error { w.mu.RLock() idv, tok := w.workerID, w.accessToken skillCount := len(w.skills) w.mu.RUnlock() if idv == "" || tok == "" { return errors.New("not enrolled") } var out map[string]any return w.doJSON(ctx, http.MethodPost, w.master+"/api/mesh/workers/"+url.PathEscape(idv)+"/heartbeat", tok, map[string]any{"protocol": workerProtocol, "status": map[string]any{"skills": skillCount}}, &out) } func (w *worker) deregister() { w.mu.RLock() idv, tok := w.workerID, w.accessToken w.mu.RUnlock() if idv == "" || tok == "" { return } ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) defer cancel() _ = w.doJSON(ctx, http.MethodDelete, w.master+"/api/mesh/workers/"+url.PathEscape(idv), tok, nil, nil) } func (w *worker) doJSON(ctx context.Context, method, endpoint, token string, in, out any) error { var rd io.Reader if in != nil { b, _ := json.Marshal(in) rd = bytes.NewReader(b) } req, err := http.NewRequestWithContext(ctx, method, endpoint, rd) if err != nil { return err } if in != nil { req.Header.Set("Content-Type", "application/json") } if token != "" { req.Header.Set("Authorization", "Bearer "+token) } resp, err := w.client.Do(req) if err != nil { return err } defer resp.Body.Close() raw, _ := io.ReadAll(io.LimitReader(resp.Body, 2<<20)) if resp.StatusCode < 200 || resp.StatusCode >= 300 { return fmt.Errorf("HTTP %d: %s", resp.StatusCode, clip(string(raw), 800)) } if out != nil && len(raw) > 0 { return json.Unmarshal(raw, out) } return nil } func (w *worker) reload() error { entries, err := os.ReadDir(w.skillsDir) if err != nil { if os.IsNotExist(err) { _ = os.MkdirAll(w.skillsDir, 0o755) return nil } return err } type candidate struct{ name, dir string } candidates := []candidate{} if st, err := os.Stat(filepath.Join(w.skillsDir, "skill.json")); err == nil && !st.IsDir() { candidates = append(candidates, candidate{name: "root", dir: w.skillsDir}) } for _, e := range entries { if !e.IsDir() || strings.HasPrefix(e.Name(), ".") || strings.HasPrefix(e.Name(), "_") { continue } dir := filepath.Join(w.skillsDir, e.Name()) if st, err := os.Stat(filepath.Join(dir, "skill.json")); err != nil || st.IsDir() { continue } candidates = append(candidates, candidate{name: e.Name(), dir: dir}) } skills := map[string]manifest{} actions := map[string]skillRef{} issues := []string{} for _, c := range candidates { dir := c.dir execDir := dir if strings.TrimSpace(w.workDir) != "" { target := filepath.Join(w.workDir, c.name) _ = os.RemoveAll(target) if err := copySkillDir(dir, target); err != nil { issues = append(issues, c.name+": workdir copy: "+err.Error()) continue } execDir = target } b, err := os.ReadFile(filepath.Join(dir, "skill.json")) if err != nil { continue } var m manifest dec := json.NewDecoder(bytes.NewReader(b)) dec.DisallowUnknownFields() if err := dec.Decode(&m); err != nil { issues = append(issues, c.name+": "+err.Error()) continue } if m.Protocol == "" { m.Protocol = skillProtocol } if m.Protocol != skillProtocol || m.ID == "" || m.Name == "" || len(m.Actions) == 0 { issues = append(issues, c.name+": manifest ungültig") continue } if m.Enabled != nil && !*m.Enabled { continue } if m.Runtime.Type == "" { m.Runtime.Type = "process" } if m.Runtime.Type != "process" || strings.TrimSpace(m.Runtime.Command) == "" { issues = append(issues, m.ID+": runtime.command fehlt") continue } good := true for i := range m.Actions { a := &m.Actions[i] if a.Name == "" { good = false break } if a.InputSchema == nil { a.InputSchema = map[string]any{"type": "object", "properties": map[string]any{}, "additionalProperties": false} } key := m.ID + "\x00" + a.Name actions[key] = skillRef{Manifest: m, Action: *a, Dir: execDir} } if good { skills[m.ID] = m } } w.mu.Lock() w.skills = skills w.actions = actions w.mu.Unlock() if len(issues) > 0 { return errors.New(strings.Join(issues, "; ")) } return nil } func (w *worker) publicSkillsLocked() []publicSkill { out := make([]publicSkill, 0, len(w.skills)) for _, m := range w.skills { acts := make([]actionManifest, len(m.Actions)) copy(acts, m.Actions) for i := range acts { acts[i].ToolName = strings.TrimSpace(acts[i].ToolName) } out = append(out, publicSkill{Protocol: skillProtocol, ID: m.ID, Name: m.Name, Version: m.Version, Description: m.Description, Actions: acts}) } sort.Slice(out, func(i, j int) bool { return out[i].ID < out[j].ID }) return out } func (w *worker) health(rw http.ResponseWriter, r *http.Request) { w.mu.RLock() out := map[string]any{"ok": true, "protocol": workerProtocol, "worker_id": w.workerID, "name": w.name, "runtime": w.runtime, "runtime_version": w.runtimeVersion, "enrolled": w.accessToken != "", "skills": len(w.skills), "actions": len(w.actions)} w.mu.RUnlock() writeJSON(rw, 200, out) } func (w *worker) listSkills(rw http.ResponseWriter, r *http.Request) { w.mu.RLock() out := w.publicSkillsLocked() w.mu.RUnlock() writeJSON(rw, 200, map[string]any{"protocol": workerProtocol, "skills": out}) } func (w *worker) reloadHTTP(rw http.ResponseWriter, r *http.Request) { if !w.authorized(r) { http.Error(rw, "unauthorized", 401) return } err := w.reload() if err == nil { ctx, cancel := context.WithTimeout(r.Context(), 10*time.Second) _ = w.register(ctx) cancel() } writeJSON(rw, 200, map[string]any{"ok": err == nil, "error": errString(err)}) } func (w *worker) authorized(r *http.Request) bool { w.mu.RLock() tok := w.accessToken w.mu.RUnlock() return tok != "" && strings.TrimSpace(strings.TrimPrefix(r.Header.Get("Authorization"), "Bearer ")) == tok && strings.HasPrefix(r.Header.Get("Authorization"), "Bearer ") } func (w *worker) invoke(rw http.ResponseWriter, r *http.Request) { if !w.authorized(r) { http.Error(rw, "unauthorized", 401) return } var in invokeRequest if err := json.NewDecoder(io.LimitReader(r.Body, 2<<20)).Decode(&in); err != nil { http.Error(rw, "invalid json", 400) return } if in.Protocol != "" && in.Protocol != invokeProtocol && in.Protocol != skillProtocol { writeJSON(rw, 400, fail(in.RequestID, "PROTOCOL_MISMATCH", "unsupported protocol")) return } key := in.SkillID + "\x00" + in.Action w.mu.RLock() ref, ok := w.actions[key] w.mu.RUnlock() if !ok { writeJSON(rw, 404, fail(in.RequestID, "SKILL_NOT_FOUND", "Skill action not found")) return } if in.Input == nil { in.Input = map[string]any{} } if err := validate(in.Input, ref.Action.InputSchema, "input"); err != nil { writeJSON(rw, 400, fail(in.RequestID, "INVALID_SKILL_INPUT", err.Error())) return } res := w.execute(r.Context(), ref, in) writeJSON(rw, 200, res) } func (w *worker) execute(ctx context.Context, ref skillRef, in invokeRequest) invokeResponse { timeout := 5 * time.Second if ref.Manifest.Runtime.TimeoutMS > 0 { timeout = time.Duration(ref.Manifest.Runtime.TimeoutMS) * time.Millisecond } if timeout > 5*time.Minute { timeout = 5 * time.Minute } call, cancel := context.WithTimeout(ctx, timeout) defer cancel() cmdPath, err := w.resolveCommand(ref) if err != nil { return fail(in.RequestID, "SKILL_EXEC_DENIED", err.Error()) } payload, _ := json.Marshal(in) cmd := exec.CommandContext(call, cmdPath, ref.Manifest.Runtime.Args...) cmd.Dir = ref.Dir cmd.Stdin = bytes.NewReader(payload) cmd.Env = w.environment(ref, in) stdout := &limitBuffer{max: w.maxOutput} stderr := &limitBuffer{max: min(w.maxOutput/4, 256*1024)} cmd.Stdout = stdout cmd.Stderr = stderr err = cmd.Run() if call.Err() == context.DeadlineExceeded { return fail(in.RequestID, "SKILL_TIMEOUT", "Skill timed out") } if stdout.overflow { return fail(in.RequestID, "SKILL_OUTPUT_LIMIT", "Skill output too large") } if err != nil { return fail(in.RequestID, "SKILL_PROCESS_FAILED", err.Error()+optionalStderr(stderr.String())) } raw := strings.TrimSpace(stdout.String()) if raw == "" { return fail(in.RequestID, "SKILL_EMPTY_OUTPUT", "Skill returned empty output") } var res invokeResponse if err := json.Unmarshal([]byte(raw), &res); err != nil { return fail(in.RequestID, "SKILL_INVALID_OUTPUT", err.Error()) } if res.Protocol == "" { res.Protocol = invokeProtocol } if res.Success && ref.Action.OutputSchema != nil { if err := validate(res.Data, ref.Action.OutputSchema, "data"); err != nil { return fail(in.RequestID, "INVALID_SKILL_OUTPUT", err.Error()) } } if res.Success && ref.Action.Mutates { if res.Mutated == nil { v := true res.Mutated = &v } } return res } func optionalStderr(s string) string { s = strings.TrimSpace(s) if s == "" { return "" } return ": " + clip(s, 1200) } func (w *worker) resolveCommand(ref skillRef) (string, error) { raw := strings.TrimSpace(ref.Manifest.Runtime.Command) if filepath.IsAbs(raw) { if !w.allowSystemExec || !ref.Manifest.Permissions.SystemExec { return "", errors.New("absolute system exec denied") } return raw, nil } if strings.Contains(raw, "/") || strings.HasPrefix(raw, ".") { p := filepath.Clean(filepath.Join(ref.Dir, raw)) base, _ := filepath.Abs(ref.Dir) abs, _ := filepath.Abs(p) if abs != base && !strings.HasPrefix(abs, base+string(os.PathSeparator)) { return "", errors.New("command escapes skill dir") } return p, nil } local := filepath.Join(ref.Dir, raw) if st, err := os.Stat(local); err == nil && !st.IsDir() { return local, nil } if !w.allowSystemExec || !ref.Manifest.Permissions.SystemExec { return "", fmt.Errorf("system executable %q denied", raw) } return exec.LookPath(raw) } func (w *worker) environment(ref skillRef, in invokeRequest) []string { m := map[string]string{"PATH": os.Getenv("PATH"), "LANG": "C.UTF-8", "TZ": in.Context.Timezone, "JARVIS_SKILL_PROTOCOL": skillProtocol, "JARVIS_SKILL_ID": ref.Manifest.ID, "JARVIS_SKILL_ACTION": ref.Action.Name, "JARVIS_SKILL_REQUEST_ID": in.RequestID, "JARVIS_SKILL_TRACE_ID": in.Context.TraceID, "JARVIS_WORKER_RUNTIME": w.runtime} for _, key := range []string{"HOME", "TMPDIR", "GOCACHE", "GOMODCACHE", "CARGO_HOME", "CARGO_TARGET_DIR", "DOTNET_CLI_HOME", "NUGET_PACKAGES", "npm_config_cache", "PYTHONPATH", "PYTHONUNBUFFERED"} { if v := os.Getenv(key); v != "" { m[key] = v } } for k, v := range ref.Manifest.Runtime.Env { if strings.HasPrefix(strings.ToUpper(k), "JARVIS_") { continue } m[k] = v } // Secrets/config stay in the worker container and are only exposed to a // skill when the manifest explicitly allowlists the variable name. for _, rawKey := range ref.Manifest.Runtime.EnvFrom { key := strings.TrimSpace(rawKey) if key == "" || strings.HasPrefix(strings.ToUpper(key), "JARVIS_") { continue } if v, ok := os.LookupEnv(key); ok { m[key] = v } } ks := make([]string, 0, len(m)) for k := range m { ks = append(ks, k) } sort.Strings(ks) out := []string{} for _, k := range ks { out = append(out, k+"="+m[k]) } return out } func copySkillDir(src, dst string) error { if err := os.MkdirAll(dst, 0o755); err != nil { return err } return filepath.WalkDir(src, func(path string, d os.DirEntry, err error) error { if err != nil { return err } rel, err := filepath.Rel(src, path) if err != nil { return err } if rel == "." { return nil } target := filepath.Join(dst, rel) info, err := d.Info() if err != nil { return err } if info.Mode()&os.ModeSymlink != 0 { return fmt.Errorf("symlinks sind im Skill-Verzeichnis nicht erlaubt: %s", rel) } if d.IsDir() { return os.MkdirAll(target, info.Mode().Perm()) } if !info.Mode().IsRegular() { return fmt.Errorf("nicht reguläre Datei im Skill: %s", rel) } in, err := os.Open(path) if err != nil { return err } defer in.Close() out, err := os.OpenFile(target, os.O_CREATE|os.O_TRUNC|os.O_WRONLY, info.Mode().Perm()) if err != nil { return err } _, cpErr := io.Copy(out, in) closeErr := out.Close() if cpErr != nil { return cpErr } return closeErr }) } type limitBuffer struct { buf bytes.Buffer max int overflow bool } func (b *limitBuffer) Write(p []byte) (int, error) { remain := b.max - b.buf.Len() if remain <= 0 { b.overflow = true return len(p), nil } if len(p) > remain { b.buf.Write(p[:remain]) b.overflow = true return len(p), nil } return b.buf.Write(p) } func (b *limitBuffer) String() string { return b.buf.String() } func validate(v any, s map[string]any, path string) error { if s == nil { return nil } typ, _ := s["type"].(string) switch typ { case "object": m, ok := v.(map[string]any) if !ok { return fmt.Errorf("%s muss object sein", path) } req, _ := s["required"].([]any) for _, x := range req { k, _ := x.(string) if _, ok := m[k]; !ok { return fmt.Errorf("%s.%s fehlt", path, k) } } if rr, ok := s["required"].([]string); ok { for _, k := range rr { if _, ok := m[k]; !ok { return fmt.Errorf("%s.%s fehlt", path, k) } } } props, _ := s["properties"].(map[string]any) for k, val := range m { raw, exists := props[k] if !exists { if ap, ok := s["additionalProperties"].(bool); ok && !ap { return fmt.Errorf("%s.%s ist nicht erlaubt", path, k) } continue } child, _ := raw.(map[string]any) if err := validate(val, child, path+"."+k); err != nil { return err } } case "array": arr, ok := v.([]any) if !ok { return fmt.Errorf("%s muss array sein", path) } if item, ok := s["items"].(map[string]any); ok { for i, x := range arr { if err := validate(x, item, fmt.Sprintf("%s[%d]", path, i)); err != nil { return err } } } case "string": if _, ok := v.(string); !ok { return fmt.Errorf("%s muss string sein", path) } case "boolean": if _, ok := v.(bool); !ok { return fmt.Errorf("%s muss boolean sein", path) } case "integer": f, ok := number(v) if !ok || f != float64(int64(f)) { return fmt.Errorf("%s muss integer sein", path) } case "number": if _, ok := number(v); !ok { return fmt.Errorf("%s muss number sein", path) } } if en, ok := s["enum"].([]any); ok { found := false for _, x := range en { if fmt.Sprint(x) == fmt.Sprint(v) { found = true break } } if !found { return fmt.Errorf("%s außerhalb enum", path) } } return nil } func number(v any) (float64, bool) { switch x := v.(type) { case float64: return x, true case float32: return float64(x), true case int: return float64(x), true case int64: return float64(x), true case json.Number: f, e := x.Float64() return f, e == nil } return 0, false } func fail(idv, code, msg string) invokeResponse { return invokeResponse{Protocol: invokeProtocol, Success: false, Error: &wireError{Code: code, Message: msg}} } func writeJSON(w http.ResponseWriter, status int, v any) { w.Header().Set("Content-Type", "application/json") w.WriteHeader(status) _ = json.NewEncoder(w).Encode(v) } func env(k, d string) string { if v := strings.TrimSpace(os.Getenv(k)); v != "" { return v } return d } func envInt(k string, d int) int { v, err := strconv.Atoi(strings.TrimSpace(os.Getenv(k))) if err == nil { return v } return d } func envBool(k string, d bool) bool { v := strings.TrimSpace(strings.ToLower(os.Getenv(k))) if v == "" { return d } return v == "1" || v == "true" || v == "yes" || v == "on" } func trimURL(s string) string { return strings.TrimRight(strings.TrimSpace(s), "/") } func clip(s string, n int) string { if len(s) <= n { return s } return s[:n] + "…" } func errString(e error) string { if e == nil { return "" } return e.Error() } func min(a, b int) int { if a < b { return a } return b }