Fix swarm credential inheritance and completion updates
- commit
- f3d8eb665923e218f8dff1eeaa795c138721db91
- author
- patriceckhart · · 2026-07-31 21:12 UTC
- parent
- b58450d
- signature
- unsigned
- refs
- tag: v0.3.29
13 files · +291 -54
M README.md +1 -1
| @@ -370,7 +370,7 @@ Background subagents that run alongside your main session. Each one is a separat | |||
| 370 | 370 | ||
| 371 | 371 | **`/session export` does NOT bundle subagents.** A `.zotsession` is just the main chat transcript; per-agent state (session file, unix-socket inbox) is machine-local and doesn't round-trip through a JSONL file. To share what an agent said, copy it out of the transcript view manually. | |
| 372 | 372 | ||
| 373 | - | **Auto-swarm.** With `/settings` -> auto-swarm on, the main agent gets a built-in `swarm_spawn` tool and a system-prompt nudge to use it. It can then fork sub-agents on its own when a request naturally splits into independent parallel work ("implement A and B", "investigate three files"). Each spawn returns the sub-agent id immediately and the main turn keeps going. When every sub-agent the agent spawned in that batch finishes its initial task, zot injects one `[auto-swarm update]` message back into the main chat recapping each agent's status, task, and transcript tail; the main agent then writes a short follow-up summary referencing the agents by id. Off by default; toggle from `/settings`. | |
| 373 | + | **Auto-swarm.** With `/settings` -> auto-swarm on, the main agent gets a built-in `swarm_spawn` tool and a system-prompt nudge to use it. It can then fork sub-agents on its own when a request naturally splits into independent parallel work ("implement A and B", "investigate three files"). Each spawn returns the sub-agent id immediately and the main turn keeps going. When every sub-agent the agent spawned in that batch finishes its initial task, zot injects one `[auto-swarm update]` message back into the main chat recapping each agent's status, task, and complete final response. If an agent produced no assistant response, zot includes a truncated transcript tail for diagnostics instead. The main agent then writes a short follow-up summary referencing the agents by id. Off by default; toggle from `/settings`. | |
| 374 | 374 | ||
| 375 | 375 | ### `/settings` | |
| 376 | 376 | ||
| @@ -370,7 +370,7 @@ Background subagents that run alongside your main session. Each one is a separat | |||||
| 370 | 370 | ||||
| 371 | **`/session export` does NOT bundle subagents.** A `.zotsession` is just the main chat transcript; per-agent state (session file, unix-socket inbox) is machine-local and doesn't round-trip through a JSONL file. To share what an agent said, copy it out of the transcript view manually. | 371 | **`/session export` does NOT bundle subagents.** A `.zotsession` is just the main chat transcript; per-agent state (session file, unix-socket inbox) is machine-local and doesn't round-trip through a JSONL file. To share what an agent said, copy it out of the transcript view manually. | ||
| 372 | 372 | ||||
| 373 | - | **Auto-swarm.** With `/settings` -> auto-swarm on, the main agent gets a built-in `swarm_spawn` tool and a system-prompt nudge to use it. It can then fork sub-agents on its own when a request naturally splits into independent parallel work ("implement A and B", "investigate three files"). Each spawn returns the sub-agent id immediately and the main turn keeps going. When every sub-agent the agent spawned in that batch finishes its initial task, zot injects one `[auto-swarm update]` message back into the main chat recapping each agent's status, task, and transcript tail; the main agent then writes a short follow-up summary referencing the agents by id. Off by default; toggle from `/settings`. | 373 | + | **Auto-swarm.** With `/settings` -> auto-swarm on, the main agent gets a built-in `swarm_spawn` tool and a system-prompt nudge to use it. It can then fork sub-agents on its own when a request naturally splits into independent parallel work ("implement A and B", "investigate three files"). Each spawn returns the sub-agent id immediately and the main turn keeps going. When every sub-agent the agent spawned in that batch finishes its initial task, zot injects one `[auto-swarm update]` message back into the main chat recapping each agent's status, task, and complete final response. If an agent produced no assistant response, zot includes a truncated transcript tail for diagnostics instead. The main agent then writes a short follow-up summary referencing the agents by id. Off by default; toggle from `/settings`. |
| 374 | 374 | ||||
| 375 | ### `/settings` | 375 | ### `/settings` | ||
| 376 | 376 | ||||
M packages/agent/args.go +6 -0
| @@ -35,6 +35,12 @@ type Args struct { | |||
| 35 | 35 | Model string | |
| 36 | 36 | APIKey string | |
| 37 | 37 | ||
| 38 | + | // inheritedCredential is populated only by swarm-agent mode from | |
| 39 | + | // the supervisor's stdin. It is never accepted as a CLI argument. | |
| 40 | + | inheritedCredential string | |
| 41 | + | inheritedAuthMethod string | |
| 42 | + | inheritedAccountID string | |
| 43 | + | ||
| 38 | 44 | BaseURL string // override provider base URL (for tests/self-hosted) | |
| 39 | 45 | SystemPrompt string | |
| 40 | 46 | AppendSystemPrompt []string | |
| @@ -35,6 +35,12 @@ type Args struct { | |||||
| 35 | Model string | 35 | Model string | ||
| 36 | APIKey string | 36 | APIKey string | ||
| 37 | 37 | ||||
| 38 | + | // inheritedCredential is populated only by swarm-agent mode from | |||
| 39 | + | // the supervisor's stdin. It is never accepted as a CLI argument. | |||
| 40 | + | inheritedCredential string | |||
| 41 | + | inheritedAuthMethod string | |||
| 42 | + | inheritedAccountID string | |||
| 43 | + | ||||
| 38 | BaseURL string // override provider base URL (for tests/self-hosted) | 44 | BaseURL string // override provider base URL (for tests/self-hosted) | ||
| 39 | SystemPrompt string | 45 | SystemPrompt string | ||
| 40 | AppendSystemPrompt []string | 46 | AppendSystemPrompt []string | ||
M packages/agent/build.go +5 -1
| @@ -351,7 +351,11 @@ func Resolve(args Args, requireCred bool) (Resolved, error) { | |||
| 351 | 351 | accountID string | |
| 352 | 352 | credErr error | |
| 353 | 353 | ) | |
| 354 | - | if provName == "ollama" { | |
| 354 | + | if args.inheritedCredential != "" { | |
| 355 | + | cred = args.inheritedCredential | |
| 356 | + | method = args.inheritedAuthMethod | |
| 357 | + | accountID = args.inheritedAccountID | |
| 358 | + | } else if provName == "ollama" { | |
| 355 | 359 | cred = firstNonEmpty(args.APIKey, "ollama") | |
| 356 | 360 | method = "apikey" | |
| 357 | 361 | } else { | |
| @@ -351,7 +351,11 @@ func Resolve(args Args, requireCred bool) (Resolved, error) { | |||||
| 351 | accountID string | 351 | accountID string | ||
| 352 | credErr error | 352 | credErr error | ||
| 353 | ) | 353 | ) | ||
| 354 | - | if provName == "ollama" { | 354 | + | if args.inheritedCredential != "" { |
| 355 | + | cred = args.inheritedCredential | |||
| 356 | + | method = args.inheritedAuthMethod | |||
| 357 | + | accountID = args.inheritedAccountID | |||
| 358 | + | } else if provName == "ollama" { | |||
| 355 | cred = firstNonEmpty(args.APIKey, "ollama") | 359 | cred = firstNonEmpty(args.APIKey, "ollama") | ||
| 356 | method = "apikey" | 360 | method = "apikey" | ||
| 357 | } else { | 361 | } else { | ||
M packages/agent/build_test.go +19 -0
| @@ -257,6 +257,25 @@ func TestResolveOllamaUsesModelBaseURLBeforeDefault(t *testing.T) { | |||
| 257 | 257 | } | |
| 258 | 258 | } | |
| 259 | 259 | ||
| 260 | + | func TestResolveUsesInheritedSwarmCredential(t *testing.T) { | |
| 261 | + | t.Setenv("ZOT_HOME", t.TempDir()) | |
| 262 | + | t.Setenv("OPENAI_API_KEY", "") | |
| 263 | + | ||
| 264 | + | r, err := Resolve(Args{ | |
| 265 | + | Provider: "openai", | |
| 266 | + | Model: "gpt-5", | |
| 267 | + | inheritedCredential: "inherited-key", | |
| 268 | + | inheritedAuthMethod: "apikey", | |
| 269 | + | inheritedAccountID: "account-id", | |
| 270 | + | }, true) | |
| 271 | + | if err != nil { | |
| 272 | + | t.Fatal(err) | |
| 273 | + | } | |
| 274 | + | if r.Credential != "inherited-key" || r.AuthMethod != "apikey" || r.AccountID != "account-id" { | |
| 275 | + | t.Fatalf("inherited credential was not preserved: %+v", r) | |
| 276 | + | } | |
| 277 | + | } | |
| 278 | + | ||
| 260 | 279 | func TestResolveLlamaCPPUsesRouterInferenceURL(t *testing.T) { | |
| 261 | 280 | t.Setenv("ZOT_HOME", t.TempDir()) | |
| 262 | 281 | t.Setenv("LLAMA_BASE_URL", "http://127.0.0.1:8080/v1/") | |
| @@ -257,6 +257,25 @@ func TestResolveOllamaUsesModelBaseURLBeforeDefault(t *testing.T) { | |||||
| 257 | } | 257 | } | ||
| 258 | } | 258 | } | ||
| 259 | 259 | ||||
| 260 | + | func TestResolveUsesInheritedSwarmCredential(t *testing.T) { | |||
| 261 | + | t.Setenv("ZOT_HOME", t.TempDir()) | |||
| 262 | + | t.Setenv("OPENAI_API_KEY", "") | |||
| 263 | + | ||||
| 264 | + | r, err := Resolve(Args{ | |||
| 265 | + | Provider: "openai", | |||
| 266 | + | Model: "gpt-5", | |||
| 267 | + | inheritedCredential: "inherited-key", | |||
| 268 | + | inheritedAuthMethod: "apikey", | |||
| 269 | + | inheritedAccountID: "account-id", | |||
| 270 | + | }, true) | |||
| 271 | + | if err != nil { | |||
| 272 | + | t.Fatal(err) | |||
| 273 | + | } | |||
| 274 | + | if r.Credential != "inherited-key" || r.AuthMethod != "apikey" || r.AccountID != "account-id" { | |||
| 275 | + | t.Fatalf("inherited credential was not preserved: %+v", r) | |||
| 276 | + | } | |||
| 277 | + | } | |||
| 278 | + | ||||
| 260 | func TestResolveLlamaCPPUsesRouterInferenceURL(t *testing.T) { | 279 | func TestResolveLlamaCPPUsesRouterInferenceURL(t *testing.T) { | ||
| 261 | t.Setenv("ZOT_HOME", t.TempDir()) | 280 | t.Setenv("ZOT_HOME", t.TempDir()) | ||
| 262 | t.Setenv("LLAMA_BASE_URL", "http://127.0.0.1:8080/v1/") | 281 | t.Setenv("LLAMA_BASE_URL", "http://127.0.0.1:8080/v1/") | ||
M packages/agent/cli.go +15 -0
| @@ -456,6 +456,21 @@ func runInteractive(ctx context.Context, args Args, version string) error { | |||
| 456 | 456 | swarmMgr = swarm.New(swarm.Config{ | |
| 457 | 457 | Root: filepath.Join(ZotHome(), "swarm"), | |
| 458 | 458 | RepoRoot: r.CWD, | |
| 459 | + | ResolveCredential: func(ctx context.Context, providerID string) (swarm.Credential, error) { | |
| 460 | + | if providerID == "ollama" { | |
| 461 | + | return swarm.Credential{Value: "ollama", Method: "apikey"}, nil | |
| 462 | + | } | |
| 463 | + | credential, method, accountID, err := ResolveCredentialFullContext(ctx, providerID, "") | |
| 464 | + | if err != nil { | |
| 465 | + | // Providers backed only by a local endpoint do not need a key. | |
| 466 | + | // Leave stdin untouched and let the child resolve that endpoint. | |
| 467 | + | if !CredentialAvailable(providerID) { | |
| 468 | + | return swarm.Credential{}, nil | |
| 469 | + | } | |
| 470 | + | return swarm.Credential{}, err | |
| 471 | + | } | |
| 472 | + | return swarm.Credential{Value: credential, Method: method, AccountID: accountID}, nil | |
| 473 | + | }, | |
| 459 | 474 | }) | |
| 460 | 475 | // Pull any previously-spawned agents off disk so the dashboard | |
| 461 | 476 | // shows them as detached and the user can resume / remove them. | |
| @@ -456,6 +456,21 @@ func runInteractive(ctx context.Context, args Args, version string) error { | |||||
| 456 | swarmMgr = swarm.New(swarm.Config{ | 456 | swarmMgr = swarm.New(swarm.Config{ | ||
| 457 | Root: filepath.Join(ZotHome(), "swarm"), | 457 | Root: filepath.Join(ZotHome(), "swarm"), | ||
| 458 | RepoRoot: r.CWD, | 458 | RepoRoot: r.CWD, | ||
| 459 | + | ResolveCredential: func(ctx context.Context, providerID string) (swarm.Credential, error) { | |||
| 460 | + | if providerID == "ollama" { | |||
| 461 | + | return swarm.Credential{Value: "ollama", Method: "apikey"}, nil | |||
| 462 | + | } | |||
| 463 | + | credential, method, accountID, err := ResolveCredentialFullContext(ctx, providerID, "") | |||
| 464 | + | if err != nil { | |||
| 465 | + | // Providers backed only by a local endpoint do not need a key. | |||
| 466 | + | // Leave stdin untouched and let the child resolve that endpoint. | |||
| 467 | + | if !CredentialAvailable(providerID) { | |||
| 468 | + | return swarm.Credential{}, nil | |||
| 469 | + | } | |||
| 470 | + | return swarm.Credential{}, err | |||
| 471 | + | } | |||
| 472 | + | return swarm.Credential{Value: credential, Method: method, AccountID: accountID}, nil | |||
| 473 | + | }, | |||
| 459 | }) | 474 | }) | ||
| 460 | // Pull any previously-spawned agents off disk so the dashboard | 475 | // Pull any previously-spawned agents off disk so the dashboard | ||
| 461 | // shows them as detached and the user can resume / remove them. | 476 | // shows them as detached and the user can resume / remove them. | ||
M packages/agent/modes/auto_swarm_test.go +25 -0
| @@ -11,6 +11,31 @@ import ( | |||
| 11 | 11 | "github.com/patriceckhart/zot/packages/core" | |
| 12 | 12 | ) | |
| 13 | 13 | ||
| 14 | + | func TestAutoSwarmSummaryIncludesCompleteFinalResponseAfterLongTask(t *testing.T) { | |
| 15 | + | response := "first response line\n" + strings.Repeat("result ", 150) + "final marker" | |
| 16 | + | mgr := swarm.New(swarm.Config{ | |
| 17 | + | Root: t.TempDir(), | |
| 18 | + | RepoRoot: t.TempDir(), | |
| 19 | + | NewRunner: func(*swarm.Agent) swarm.Runner { | |
| 20 | + | return swarm.RunnerFunc(func(_ context.Context, sink swarm.Sink) error { | |
| 21 | + | sink.Transcript(response) | |
| 22 | + | return nil | |
| 23 | + | }) | |
| 24 | + | }, | |
| 25 | + | }) | |
| 26 | + | a, err := mgr.Spawn(context.Background(), strings.Repeat("long task ", 150)) | |
| 27 | + | if err != nil { | |
| 28 | + | t.Fatal(err) | |
| 29 | + | } | |
| 30 | + | ||
| 31 | + | iv := newQueuedAutoSwarmInteractive() | |
| 32 | + | iv.TrackSwarmAgent(a, a.Task) | |
| 33 | + | update := waitForQueuedPrompt(t, iv) | |
| 34 | + | if !strings.Contains(update, "final response:\n"+response) { | |
| 35 | + | t.Fatalf("update missing complete final response: %q", update) | |
| 36 | + | } | |
| 37 | + | } | |
| 38 | + | ||
| 14 | 39 | func TestTrackSwarmAgentReportsStartupFailure(t *testing.T) { | |
| 15 | 40 | mgr := swarm.New(swarm.Config{ | |
| 16 | 41 | Root: t.TempDir(), | |
| @@ -11,6 +11,31 @@ import ( | |||||
| 11 | "github.com/patriceckhart/zot/packages/core" | 11 | "github.com/patriceckhart/zot/packages/core" | ||
| 12 | ) | 12 | ) | ||
| 13 | 13 | ||||
| 14 | + | func TestAutoSwarmSummaryIncludesCompleteFinalResponseAfterLongTask(t *testing.T) { | |||
| 15 | + | response := "first response line\n" + strings.Repeat("result ", 150) + "final marker" | |||
| 16 | + | mgr := swarm.New(swarm.Config{ | |||
| 17 | + | Root: t.TempDir(), | |||
| 18 | + | RepoRoot: t.TempDir(), | |||
| 19 | + | NewRunner: func(*swarm.Agent) swarm.Runner { | |||
| 20 | + | return swarm.RunnerFunc(func(_ context.Context, sink swarm.Sink) error { | |||
| 21 | + | sink.Transcript(response) | |||
| 22 | + | return nil | |||
| 23 | + | }) | |||
| 24 | + | }, | |||
| 25 | + | }) | |||
| 26 | + | a, err := mgr.Spawn(context.Background(), strings.Repeat("long task ", 150)) | |||
| 27 | + | if err != nil { | |||
| 28 | + | t.Fatal(err) | |||
| 29 | + | } | |||
| 30 | + | ||||
| 31 | + | iv := newQueuedAutoSwarmInteractive() | |||
| 32 | + | iv.TrackSwarmAgent(a, a.Task) | |||
| 33 | + | update := waitForQueuedPrompt(t, iv) | |||
| 34 | + | if !strings.Contains(update, "final response:\n"+response) { | |||
| 35 | + | t.Fatalf("update missing complete final response: %q", update) | |||
| 36 | + | } | |||
| 37 | + | } | |||
| 38 | + | ||||
| 14 | func TestTrackSwarmAgentReportsStartupFailure(t *testing.T) { | 39 | func TestTrackSwarmAgentReportsStartupFailure(t *testing.T) { | ||
| 15 | mgr := swarm.New(swarm.Config{ | 40 | mgr := swarm.New(swarm.Config{ | ||
| 16 | Root: t.TempDir(), | 41 | Root: t.TempDir(), | ||
M packages/agent/modes/interactive.go +5 -1
| @@ -6295,7 +6295,11 @@ func (i *Interactive) flushSwarmSummary(batch []*swarmWatchEntry) { | |||
| 6295 | 6295 | } else if e.err != "" { | |
| 6296 | 6296 | fmt.Fprintf(&sb, " turn error: %s\n", truncateForSummary(e.err, 240)) | |
| 6297 | 6297 | } | |
| 6298 | - | if tail := strings.TrimSpace(snap.Tail); tail != "" { | |
| 6298 | + | if snap.LastAssistant != "" { | |
| 6299 | + | sb.WriteString(" final response:\n") | |
| 6300 | + | sb.WriteString(snap.LastAssistant) | |
| 6301 | + | sb.WriteString("\n") | |
| 6302 | + | } else if tail := strings.TrimSpace(snap.Tail); tail != "" { | |
| 6299 | 6303 | fmt.Fprintf(&sb, " tail: %s\n", truncateForSummary(tail, 600)) | |
| 6300 | 6304 | } | |
| 6301 | 6305 | sb.WriteString("\n") | |
| @@ -6295,7 +6295,11 @@ func (i *Interactive) flushSwarmSummary(batch []*swarmWatchEntry) { | |||||
| 6295 | } else if e.err != "" { | 6295 | } else if e.err != "" { | ||
| 6296 | fmt.Fprintf(&sb, " turn error: %s\n", truncateForSummary(e.err, 240)) | 6296 | fmt.Fprintf(&sb, " turn error: %s\n", truncateForSummary(e.err, 240)) | ||
| 6297 | } | 6297 | } | ||
| 6298 | - | if tail := strings.TrimSpace(snap.Tail); tail != "" { | 6298 | + | if snap.LastAssistant != "" { |
| 6299 | + | sb.WriteString(" final response:\n") | |||
| 6300 | + | sb.WriteString(snap.LastAssistant) | |||
| 6301 | + | sb.WriteString("\n") | |||
| 6302 | + | } else if tail := strings.TrimSpace(snap.Tail); tail != "" { | |||
| 6299 | fmt.Fprintf(&sb, " tail: %s\n", truncateForSummary(tail, 600)) | 6303 | fmt.Fprintf(&sb, " tail: %s\n", truncateForSummary(tail, 600)) | ||
| 6300 | } | 6304 | } | ||
| 6301 | sb.WriteString("\n") | 6305 | sb.WriteString("\n") | ||
M packages/agent/swarm/agent.go +24 -7
| @@ -64,12 +64,13 @@ type Agent struct { | |||
| 64 | 64 | // caller redialing. | |
| 65 | 65 | inbox *Inbox | |
| 66 | 66 | ||
| 67 | - | mu sync.Mutex | |
| 68 | - | status Status | |
| 69 | - | activity string | |
| 70 | - | transcript []string | |
| 71 | - | finished time.Time | |
| 72 | - | lastErr error | |
| 67 | + | mu sync.Mutex | |
| 68 | + | status Status | |
| 69 | + | activity string | |
| 70 | + | transcript []string | |
| 71 | + | lastAssistant string | |
| 72 | + | finished time.Time | |
| 73 | + | lastErr error | |
| 73 | 74 | ||
| 74 | 75 | // OnTurnEnd, if set, fires once per prompt-level turn_end event | |
| 75 | 76 | // emitted by the swarm daemon wrapper. Provider/tool-loop | |
| @@ -154,13 +155,29 @@ func (a *Agent) setActivity(msg string) { | |||
| 154 | 155 | } | |
| 155 | 156 | ||
| 156 | 157 | func (a *Agent) appendTranscript(chunk string) { | |
| 158 | + | a.appendTranscriptLocked(chunk, "", false) | |
| 159 | + | } | |
| 160 | + | ||
| 161 | + | func (a *Agent) appendUserMessage(text string) { | |
| 162 | + | a.appendTranscriptLocked(text, "user: ", false) | |
| 163 | + | } | |
| 164 | + | ||
| 165 | + | func (a *Agent) appendAssistantMessage(text string) { | |
| 166 | + | a.appendTranscriptLocked(text, "", true) | |
| 167 | + | } | |
| 168 | + | ||
| 169 | + | func (a *Agent) appendTranscriptLocked(chunk, linePrefix string, assistant bool) { | |
| 170 | + | message := chunk | |
| 157 | 171 | chunk = strings.TrimRight(chunk, "\n") | |
| 158 | 172 | if chunk == "" { | |
| 159 | 173 | return | |
| 160 | 174 | } | |
| 161 | 175 | a.mu.Lock() | |
| 162 | 176 | for _, line := range strings.Split(chunk, "\n") { | |
| 163 | - | a.transcript = append(a.transcript, line) | |
| 177 | + | a.transcript = append(a.transcript, linePrefix+line) | |
| 178 | + | } | |
| 179 | + | if assistant { | |
| 180 | + | a.lastAssistant = message | |
| 164 | 181 | } | |
| 165 | 182 | // Bound the transcript so dashboards don't hold gigabytes for | |
| 166 | 183 | // long-running agents. 2000 lines is plenty for inspection; | |
| @@ -64,12 +64,13 @@ type Agent struct { | |||||
| 64 | // caller redialing. | 64 | // caller redialing. | ||
| 65 | inbox *Inbox | 65 | inbox *Inbox | ||
| 66 | 66 | ||||
| 67 | - | mu sync.Mutex | 67 | + | mu sync.Mutex |
| 68 | - | status Status | 68 | + | status Status |
| 69 | - | activity string | 69 | + | activity string |
| 70 | - | transcript []string | 70 | + | transcript []string |
| 71 | - | finished time.Time | 71 | + | lastAssistant string |
| 72 | - | lastErr error | 72 | + | finished time.Time |
| 73 | + | lastErr error | |||
| 73 | 74 | ||||
| 74 | // OnTurnEnd, if set, fires once per prompt-level turn_end event | 75 | // OnTurnEnd, if set, fires once per prompt-level turn_end event | ||
| 75 | // emitted by the swarm daemon wrapper. Provider/tool-loop | 76 | // emitted by the swarm daemon wrapper. Provider/tool-loop | ||
| @@ -154,13 +155,29 @@ func (a *Agent) setActivity(msg string) { | |||||
| 154 | } | 155 | } | ||
| 155 | 156 | ||||
| 156 | func (a *Agent) appendTranscript(chunk string) { | 157 | func (a *Agent) appendTranscript(chunk string) { | ||
| 158 | + | a.appendTranscriptLocked(chunk, "", false) | |||
| 159 | + | } | |||
| 160 | + | ||||
| 161 | + | func (a *Agent) appendUserMessage(text string) { | |||
| 162 | + | a.appendTranscriptLocked(text, "user: ", false) | |||
| 163 | + | } | |||
| 164 | + | ||||
| 165 | + | func (a *Agent) appendAssistantMessage(text string) { | |||
| 166 | + | a.appendTranscriptLocked(text, "", true) | |||
| 167 | + | } | |||
| 168 | + | ||||
| 169 | + | func (a *Agent) appendTranscriptLocked(chunk, linePrefix string, assistant bool) { | |||
| 170 | + | message := chunk | |||
| 157 | chunk = strings.TrimRight(chunk, "\n") | 171 | chunk = strings.TrimRight(chunk, "\n") | ||
| 158 | if chunk == "" { | 172 | if chunk == "" { | ||
| 159 | return | 173 | return | ||
| 160 | } | 174 | } | ||
| 161 | a.mu.Lock() | 175 | a.mu.Lock() | ||
| 162 | for _, line := range strings.Split(chunk, "\n") { | 176 | for _, line := range strings.Split(chunk, "\n") { | ||
| 163 | - | a.transcript = append(a.transcript, line) | 177 | + | a.transcript = append(a.transcript, linePrefix+line) |
| 178 | + | } | |||
| 179 | + | if assistant { | |||
| 180 | + | a.lastAssistant = message | |||
| 164 | } | 181 | } | ||
| 165 | // Bound the transcript so dashboards don't hold gigabytes for | 182 | // Bound the transcript so dashboards don't hold gigabytes for | ||
| 166 | // long-running agents. 2000 lines is plenty for inspection; | 183 | // long-running agents. 2000 lines is plenty for inspection; | ||
M packages/agent/swarm/persist.go +8 -12
| @@ -242,27 +242,23 @@ func replayEventsIntoAgent(a *Agent, evs []Event) { | |||
| 242 | 242 | terminal := false | |
| 243 | 243 | for _, ev := range evs { | |
| 244 | 244 | switch ev.Type { | |
| 245 | - | case "assistant_message": | |
| 245 | + | case "assistant_message", "user_message": | |
| 246 | + | var text []string | |
| 246 | 247 | if c, ok := ev.Data["content"].([]any); ok { | |
| 247 | 248 | for _, blk := range c { | |
| 248 | 249 | m, _ := blk.(map[string]any) | |
| 249 | 250 | if t, _ := m["type"].(string); t == "text" { | |
| 250 | 251 | if txt, _ := m["text"].(string); txt != "" { | |
| 251 | - | a.appendTranscript(txt) | |
| 252 | + | text = append(text, txt) | |
| 252 | 253 | } | |
| 253 | 254 | } | |
| 254 | 255 | } | |
| 255 | 256 | } | |
| 256 | - | case "user_message": | |
| 257 | - | if c, ok := ev.Data["content"].([]any); ok { | |
| 258 | - | for _, blk := range c { | |
| 259 | - | m, _ := blk.(map[string]any) | |
| 260 | - | if t, _ := m["type"].(string); t == "text" { | |
| 261 | - | if txt, _ := m["text"].(string); txt != "" { | |
| 262 | - | a.appendTranscript("user: " + txt) | |
| 263 | - | } | |
| 264 | - | } | |
| 265 | - | } | |
| 257 | + | message := strings.Join(text, "\n") | |
| 258 | + | if ev.Type == "assistant_message" { | |
| 259 | + | a.appendAssistantMessage(message) | |
| 260 | + | } else { | |
| 261 | + | a.appendUserMessage(message) | |
| 266 | 262 | } | |
| 267 | 263 | case "stdout": | |
| 268 | 264 | if txt, _ := ev.Data["text"].(string); txt != "" { | |
| @@ -242,27 +242,23 @@ func replayEventsIntoAgent(a *Agent, evs []Event) { | |||||
| 242 | terminal := false | 242 | terminal := false | ||
| 243 | for _, ev := range evs { | 243 | for _, ev := range evs { | ||
| 244 | switch ev.Type { | 244 | switch ev.Type { | ||
| 245 | - | case "assistant_message": | 245 | + | case "assistant_message", "user_message": |
| 246 | + | var text []string | |||
| 246 | if c, ok := ev.Data["content"].([]any); ok { | 247 | if c, ok := ev.Data["content"].([]any); ok { | ||
| 247 | for _, blk := range c { | 248 | for _, blk := range c { | ||
| 248 | m, _ := blk.(map[string]any) | 249 | m, _ := blk.(map[string]any) | ||
| 249 | if t, _ := m["type"].(string); t == "text" { | 250 | if t, _ := m["type"].(string); t == "text" { | ||
| 250 | if txt, _ := m["text"].(string); txt != "" { | 251 | if txt, _ := m["text"].(string); txt != "" { | ||
| 251 | - | a.appendTranscript(txt) | 252 | + | text = append(text, txt) |
| 252 | } | 253 | } | ||
| 253 | } | 254 | } | ||
| 254 | } | 255 | } | ||
| 255 | } | 256 | } | ||
| 256 | - | case "user_message": | 257 | + | message := strings.Join(text, "\n") |
| 257 | - | if c, ok := ev.Data["content"].([]any); ok { | 258 | + | if ev.Type == "assistant_message" { |
| 258 | - | for _, blk := range c { | 259 | + | a.appendAssistantMessage(message) |
| 259 | - | m, _ := blk.(map[string]any) | 260 | + | } else { |
| 260 | - | if t, _ := m["type"].(string); t == "text" { | 261 | + | a.appendUserMessage(message) |
| 261 | - | if txt, _ := m["text"].(string); txt != "" { | |||
| 262 | - | a.appendTranscript("user: " + txt) | |||
| 263 | - | } | |||
| 264 | - | } | |||
| 265 | - | } | |||
| 266 | } | 262 | } | ||
| 267 | case "stdout": | 263 | case "stdout": | ||
| 268 | if txt, _ := ev.Data["text"].(string); txt != "" { | 264 | if txt, _ := ev.Data["text"].(string); txt != "" { | ||
M packages/agent/swarm/runner.go +46 -14
| @@ -2,6 +2,7 @@ package swarm | |||
| 2 | 2 | ||
| 3 | 3 | import ( | |
| 4 | 4 | "bufio" | |
| 5 | + | "bytes" | |
| 5 | 6 | "context" | |
| 6 | 7 | "encoding/json" | |
| 7 | 8 | "fmt" | |
| @@ -33,7 +34,8 @@ import ( | |||
| 33 | 34 | // for the parent's own agents. /swarm open in a separate zot would | |
| 34 | 35 | // read the log directly. | |
| 35 | 36 | type execRunner struct { | |
| 36 | - | agent *Agent | |
| 37 | + | agent *Agent | |
| 38 | + | resolveCredential func(context.Context, string) (Credential, error) | |
| 37 | 39 | ||
| 38 | 40 | // Command overrides the default `zot --swarm-agent ...` | |
| 39 | 41 | // invocation. Tests set this to a fake binary (or `go run` | |
| @@ -174,6 +176,20 @@ func (r *execRunner) Run(ctx context.Context, sink Sink) error { | |||
| 174 | 176 | "ZOT_SWARM_AGENT_ID="+r.agent.ID, | |
| 175 | 177 | "ZOT_SWARM_EVENT_LOG="+logPath, | |
| 176 | 178 | ) | |
| 179 | + | if r.resolveCredential != nil { | |
| 180 | + | credential, resolveErr := r.resolveCredential(ctx, r.agent.Provider) | |
| 181 | + | if resolveErr != nil { | |
| 182 | + | return fmt.Errorf("resolve swarm credential for %s: %w", r.agent.Provider, resolveErr) | |
| 183 | + | } | |
| 184 | + | if credential.Value != "" { | |
| 185 | + | encoded, encodeErr := json.Marshal(credential) | |
| 186 | + | if encodeErr != nil { | |
| 187 | + | return fmt.Errorf("encode swarm credential: %w", encodeErr) | |
| 188 | + | } | |
| 189 | + | cmd.Stdin = bytes.NewReader(encoded) | |
| 190 | + | cmd.Env = append(cmd.Env, "ZOT_SWARM_CREDENTIAL_STDIN=1") | |
| 191 | + | } | |
| 192 | + | } | |
| 177 | 193 | ||
| 178 | 194 | stdout, err := cmd.StdoutPipe() | |
| 179 | 195 | if err != nil { | |
| @@ -319,30 +335,46 @@ func notifyPromptTurnEnd(a *Agent, ev Event) { | |||
| 319 | 335 | // few event types are interpreted; the rest still land in the | |
| 320 | 336 | // durable log via the caller. | |
| 321 | 337 | func applyEventToSink(ev Event, sink Sink) { | |
| 322 | - | switch ev.Type { | |
| 323 | - | case "assistant_message": | |
| 324 | - | if c, ok := ev.Data["content"].([]any); ok { | |
| 325 | - | for _, blk := range c { | |
| 326 | - | m, _ := blk.(map[string]any) | |
| 327 | - | if t, _ := m["type"].(string); t == "text" { | |
| 328 | - | if txt, _ := m["text"].(string); txt != "" { | |
| 329 | - | sink.Transcript(txt) | |
| 330 | - | } | |
| 331 | - | } | |
| 338 | + | type roleSink interface { | |
| 339 | + | userMessage(string) | |
| 340 | + | assistantMessage(string) | |
| 341 | + | } | |
| 342 | + | appendMessage := func(text string, assistant bool) { | |
| 343 | + | if text == "" { | |
| 344 | + | return | |
| 345 | + | } | |
| 346 | + | if roles, ok := sink.(roleSink); ok { | |
| 347 | + | if assistant { | |
| 348 | + | roles.assistantMessage(text) | |
| 349 | + | } else { | |
| 350 | + | roles.userMessage(text) | |
| 332 | 351 | } | |
| 352 | + | return | |
| 333 | 353 | } | |
| 334 | - | sink.Activity("idle") | |
| 335 | - | case "user_message": | |
| 354 | + | if assistant { | |
| 355 | + | sink.Transcript(text) | |
| 356 | + | } else { | |
| 357 | + | sink.Transcript("user: " + text) | |
| 358 | + | } | |
| 359 | + | } | |
| 360 | + | ||
| 361 | + | switch ev.Type { | |
| 362 | + | case "assistant_message", "user_message": | |
| 363 | + | var text []string | |
| 336 | 364 | if c, ok := ev.Data["content"].([]any); ok { | |
| 337 | 365 | for _, blk := range c { | |
| 338 | 366 | m, _ := blk.(map[string]any) | |
| 339 | 367 | if t, _ := m["type"].(string); t == "text" { | |
| 340 | 368 | if txt, _ := m["text"].(string); txt != "" { | |
| 341 | - | sink.Transcript("user: " + txt) | |
| 369 | + | text = append(text, txt) | |
| 342 | 370 | } | |
| 343 | 371 | } | |
| 344 | 372 | } | |
| 345 | 373 | } | |
| 374 | + | appendMessage(strings.Join(text, "\n"), ev.Type == "assistant_message") | |
| 375 | + | if ev.Type == "assistant_message" { | |
| 376 | + | sink.Activity("idle") | |
| 377 | + | } | |
| 346 | 378 | case "turn_start": | |
| 347 | 379 | sink.Activity("thinking") | |
| 348 | 380 | case "tool_call": | |
| @@ -2,6 +2,7 @@ package swarm | |||||
| 2 | 2 | ||||
| 3 | import ( | 3 | import ( | ||
| 4 | "bufio" | 4 | "bufio" | ||
| 5 | + | "bytes" | |||
| 5 | "context" | 6 | "context" | ||
| 6 | "encoding/json" | 7 | "encoding/json" | ||
| 7 | "fmt" | 8 | "fmt" | ||
| @@ -33,7 +34,8 @@ import ( | |||||
| 33 | // for the parent's own agents. /swarm open in a separate zot would | 34 | // for the parent's own agents. /swarm open in a separate zot would | ||
| 34 | // read the log directly. | 35 | // read the log directly. | ||
| 35 | type execRunner struct { | 36 | type execRunner struct { | ||
| 36 | - | agent *Agent | 37 | + | agent *Agent |
| 38 | + | resolveCredential func(context.Context, string) (Credential, error) | |||
| 37 | 39 | ||||
| 38 | // Command overrides the default `zot --swarm-agent ...` | 40 | // Command overrides the default `zot --swarm-agent ...` | ||
| 39 | // invocation. Tests set this to a fake binary (or `go run` | 41 | // invocation. Tests set this to a fake binary (or `go run` | ||
| @@ -174,6 +176,20 @@ func (r *execRunner) Run(ctx context.Context, sink Sink) error { | |||||
| 174 | "ZOT_SWARM_AGENT_ID="+r.agent.ID, | 176 | "ZOT_SWARM_AGENT_ID="+r.agent.ID, | ||
| 175 | "ZOT_SWARM_EVENT_LOG="+logPath, | 177 | "ZOT_SWARM_EVENT_LOG="+logPath, | ||
| 176 | ) | 178 | ) | ||
| 179 | + | if r.resolveCredential != nil { | |||
| 180 | + | credential, resolveErr := r.resolveCredential(ctx, r.agent.Provider) | |||
| 181 | + | if resolveErr != nil { | |||
| 182 | + | return fmt.Errorf("resolve swarm credential for %s: %w", r.agent.Provider, resolveErr) | |||
| 183 | + | } | |||
| 184 | + | if credential.Value != "" { | |||
| 185 | + | encoded, encodeErr := json.Marshal(credential) | |||
| 186 | + | if encodeErr != nil { | |||
| 187 | + | return fmt.Errorf("encode swarm credential: %w", encodeErr) | |||
| 188 | + | } | |||
| 189 | + | cmd.Stdin = bytes.NewReader(encoded) | |||
| 190 | + | cmd.Env = append(cmd.Env, "ZOT_SWARM_CREDENTIAL_STDIN=1") | |||
| 191 | + | } | |||
| 192 | + | } | |||
| 177 | 193 | ||||
| 178 | stdout, err := cmd.StdoutPipe() | 194 | stdout, err := cmd.StdoutPipe() | ||
| 179 | if err != nil { | 195 | if err != nil { | ||
| @@ -319,30 +335,46 @@ func notifyPromptTurnEnd(a *Agent, ev Event) { | |||||
| 319 | // few event types are interpreted; the rest still land in the | 335 | // few event types are interpreted; the rest still land in the | ||
| 320 | // durable log via the caller. | 336 | // durable log via the caller. | ||
| 321 | func applyEventToSink(ev Event, sink Sink) { | 337 | func applyEventToSink(ev Event, sink Sink) { | ||
| 322 | - | switch ev.Type { | 338 | + | type roleSink interface { |
| 323 | - | case "assistant_message": | 339 | + | userMessage(string) |
| 324 | - | if c, ok := ev.Data["content"].([]any); ok { | 340 | + | assistantMessage(string) |
| 325 | - | for _, blk := range c { | 341 | + | } |
| 326 | - | m, _ := blk.(map[string]any) | 342 | + | appendMessage := func(text string, assistant bool) { |
| 327 | - | if t, _ := m["type"].(string); t == "text" { | 343 | + | if text == "" { |
| 328 | - | if txt, _ := m["text"].(string); txt != "" { | 344 | + | return |
| 329 | - | sink.Transcript(txt) | 345 | + | } |
| 330 | - | } | 346 | + | if roles, ok := sink.(roleSink); ok { |
| 331 | - | } | 347 | + | if assistant { |
| 348 | + | roles.assistantMessage(text) | |||
| 349 | + | } else { | |||
| 350 | + | roles.userMessage(text) | |||
| 332 | } | 351 | } | ||
| 352 | + | return | |||
| 333 | } | 353 | } | ||
| 334 | - | sink.Activity("idle") | 354 | + | if assistant { |
| 335 | - | case "user_message": | 355 | + | sink.Transcript(text) |
| 356 | + | } else { | |||
| 357 | + | sink.Transcript("user: " + text) | |||
| 358 | + | } | |||
| 359 | + | } | |||
| 360 | + | ||||
| 361 | + | switch ev.Type { | |||
| 362 | + | case "assistant_message", "user_message": | |||
| 363 | + | var text []string | |||
| 336 | if c, ok := ev.Data["content"].([]any); ok { | 364 | if c, ok := ev.Data["content"].([]any); ok { | ||
| 337 | for _, blk := range c { | 365 | for _, blk := range c { | ||
| 338 | m, _ := blk.(map[string]any) | 366 | m, _ := blk.(map[string]any) | ||
| 339 | if t, _ := m["type"].(string); t == "text" { | 367 | if t, _ := m["type"].(string); t == "text" { | ||
| 340 | if txt, _ := m["text"].(string); txt != "" { | 368 | if txt, _ := m["text"].(string); txt != "" { | ||
| 341 | - | sink.Transcript("user: " + txt) | 369 | + | text = append(text, txt) |
| 342 | } | 370 | } | ||
| 343 | } | 371 | } | ||
| 344 | } | 372 | } | ||
| 345 | } | 373 | } | ||
| 374 | + | appendMessage(strings.Join(text, "\n"), ev.Type == "assistant_message") | |||
| 375 | + | if ev.Type == "assistant_message" { | |||
| 376 | + | sink.Activity("idle") | |||
| 377 | + | } | |||
| 346 | case "turn_start": | 378 | case "turn_start": | ||
| 347 | sink.Activity("thinking") | 379 | sink.Activity("thinking") | ||
| 348 | case "tool_call": | 380 | case "tool_call": | ||
M packages/agent/swarm/runner_test.go +81 -0
| @@ -1,6 +1,11 @@ | |||
| 1 | 1 | package swarm | |
| 2 | 2 | ||
| 3 | 3 | import ( | |
| 4 | + | "context" | |
| 5 | + | "encoding/json" | |
| 6 | + | "fmt" | |
| 7 | + | "os" | |
| 8 | + | "path/filepath" | |
| 4 | 9 | "strings" | |
| 5 | 10 | "testing" | |
| 6 | 11 | "time" | |
| @@ -25,6 +30,82 @@ import ( | |||
| 25 | 30 | // The test asserts the load-bearing pieces are present in plausible | |
| 26 | 31 | // positions. If a flag is renamed, update both the runner and this | |
| 27 | 32 | // test so we notice immediately. | |
| 33 | + | func TestCredentialStdinHelperProcess(t *testing.T) { | |
| 34 | + | if os.Getenv("ZOT_SWARM_CREDENTIAL_HELPER") != "1" { | |
| 35 | + | return | |
| 36 | + | } | |
| 37 | + | if os.Getenv("ZOT_SWARM_CREDENTIAL_STDIN") != "1" { | |
| 38 | + | os.Exit(2) | |
| 39 | + | } | |
| 40 | + | var credential Credential | |
| 41 | + | if err := json.NewDecoder(os.Stdin).Decode(&credential); err != nil { | |
| 42 | + | os.Exit(3) | |
| 43 | + | } | |
| 44 | + | if credential.Value != "inherited-secret" || credential.Method != "apikey" { | |
| 45 | + | os.Exit(4) | |
| 46 | + | } | |
| 47 | + | for _, arg := range os.Args { | |
| 48 | + | if strings.Contains(arg, credential.Value) { | |
| 49 | + | os.Exit(5) | |
| 50 | + | } | |
| 51 | + | } | |
| 52 | + | for _, env := range os.Environ() { | |
| 53 | + | if strings.Contains(env, credential.Value) { | |
| 54 | + | os.Exit(6) | |
| 55 | + | } | |
| 56 | + | } | |
| 57 | + | fmt.Println(`{"type":"agent_stopped","data":{"reason":"completed"}}`) | |
| 58 | + | os.Exit(0) | |
| 59 | + | } | |
| 60 | + | ||
| 61 | + | func TestExecRunnerTransfersCredentialOnlyOnStdin(t *testing.T) { | |
| 62 | + | t.Setenv("ZOT_SWARM_CREDENTIAL_HELPER", "1") | |
| 63 | + | root := t.TempDir() | |
| 64 | + | a := &Agent{ | |
| 65 | + | ID: "credential-test", | |
| 66 | + | Dir: root, | |
| 67 | + | Provider: "custom", | |
| 68 | + | SessionPath: filepath.Join(root, "session.json"), | |
| 69 | + | InboxPath: filepath.Join(root, "inbox"), | |
| 70 | + | EventLogPath: filepath.Join(root, "events.jsonl"), | |
| 71 | + | } | |
| 72 | + | calls := 0 | |
| 73 | + | r := &execRunner{ | |
| 74 | + | agent: a, | |
| 75 | + | Command: []string{os.Args[0], "-test.run=^TestCredentialStdinHelperProcess$"}, | |
| 76 | + | resolveCredential: func(context.Context, string) (Credential, error) { | |
| 77 | + | calls++ | |
| 78 | + | return Credential{Value: "inherited-secret", Method: "apikey"}, nil | |
| 79 | + | }, | |
| 80 | + | } | |
| 81 | + | if err := r.Run(context.Background(), agentSink{a: a}); err != nil { | |
| 82 | + | t.Fatal(err) | |
| 83 | + | } | |
| 84 | + | if calls != 1 { | |
| 85 | + | t.Fatalf("credential resolver called %d times, want 1", calls) | |
| 86 | + | } | |
| 87 | + | } | |
| 88 | + | ||
| 89 | + | func TestApplyEventPreservesMultilineRolesAndFinalAssistant(t *testing.T) { | |
| 90 | + | a := &Agent{} | |
| 91 | + | sink := agentSink{a: a} | |
| 92 | + | applyEventToSink(NewEvent("user_message", map[string]any{"content": []any{ | |
| 93 | + | map[string]any{"type": "text", "text": "first line\n\nlast line"}, | |
| 94 | + | }}), sink) | |
| 95 | + | applyEventToSink(NewEvent("assistant_message", map[string]any{"content": []any{ | |
| 96 | + | map[string]any{"type": "text", "text": "complete\nanswer"}, | |
| 97 | + | }}), sink) | |
| 98 | + | ||
| 99 | + | snap := a.Snapshot() | |
| 100 | + | wantLines := []string{"user: first line", "user: ", "user: last line", "complete", "answer"} | |
| 101 | + | if strings.Join(snap.Lines, "\n") != strings.Join(wantLines, "\n") { | |
| 102 | + | t.Fatalf("transcript = %#v, want %#v", snap.Lines, wantLines) | |
| 103 | + | } | |
| 104 | + | if snap.LastAssistant != "complete\nanswer" { | |
| 105 | + | t.Fatalf("last assistant = %q", snap.LastAssistant) | |
| 106 | + | } | |
| 107 | + | } | |
| 108 | + | ||
| 28 | 109 | func TestSwarmAgentArgs(t *testing.T) { | |
| 29 | 110 | args := swarmAgentArgs(swarmAgentArgsOpts{ | |
| 30 | 111 | Exe: "/path/to/zot", | |
| @@ -1,6 +1,11 @@ | |||||
| 1 | package swarm | 1 | package swarm | ||
| 2 | 2 | ||||
| 3 | import ( | 3 | import ( | ||
| 4 | + | "context" | |||
| 5 | + | "encoding/json" | |||
| 6 | + | "fmt" | |||
| 7 | + | "os" | |||
| 8 | + | "path/filepath" | |||
| 4 | "strings" | 9 | "strings" | ||
| 5 | "testing" | 10 | "testing" | ||
| 6 | "time" | 11 | "time" | ||
| @@ -25,6 +30,82 @@ import ( | |||||
| 25 | // The test asserts the load-bearing pieces are present in plausible | 30 | // The test asserts the load-bearing pieces are present in plausible | ||
| 26 | // positions. If a flag is renamed, update both the runner and this | 31 | // positions. If a flag is renamed, update both the runner and this | ||
| 27 | // test so we notice immediately. | 32 | // test so we notice immediately. | ||
| 33 | + | func TestCredentialStdinHelperProcess(t *testing.T) { | |||
| 34 | + | if os.Getenv("ZOT_SWARM_CREDENTIAL_HELPER") != "1" { | |||
| 35 | + | return | |||
| 36 | + | } | |||
| 37 | + | if os.Getenv("ZOT_SWARM_CREDENTIAL_STDIN") != "1" { | |||
| 38 | + | os.Exit(2) | |||
| 39 | + | } | |||
| 40 | + | var credential Credential | |||
| 41 | + | if err := json.NewDecoder(os.Stdin).Decode(&credential); err != nil { | |||
| 42 | + | os.Exit(3) | |||
| 43 | + | } | |||
| 44 | + | if credential.Value != "inherited-secret" || credential.Method != "apikey" { | |||
| 45 | + | os.Exit(4) | |||
| 46 | + | } | |||
| 47 | + | for _, arg := range os.Args { | |||
| 48 | + | if strings.Contains(arg, credential.Value) { | |||
| 49 | + | os.Exit(5) | |||
| 50 | + | } | |||
| 51 | + | } | |||
| 52 | + | for _, env := range os.Environ() { | |||
| 53 | + | if strings.Contains(env, credential.Value) { | |||
| 54 | + | os.Exit(6) | |||
| 55 | + | } | |||
| 56 | + | } | |||
| 57 | + | fmt.Println(`{"type":"agent_stopped","data":{"reason":"completed"}}`) | |||
| 58 | + | os.Exit(0) | |||
| 59 | + | } | |||
| 60 | + | ||||
| 61 | + | func TestExecRunnerTransfersCredentialOnlyOnStdin(t *testing.T) { | |||
| 62 | + | t.Setenv("ZOT_SWARM_CREDENTIAL_HELPER", "1") | |||
| 63 | + | root := t.TempDir() | |||
| 64 | + | a := &Agent{ | |||
| 65 | + | ID: "credential-test", | |||
| 66 | + | Dir: root, | |||
| 67 | + | Provider: "custom", | |||
| 68 | + | SessionPath: filepath.Join(root, "session.json"), | |||
| 69 | + | InboxPath: filepath.Join(root, "inbox"), | |||
| 70 | + | EventLogPath: filepath.Join(root, "events.jsonl"), | |||
| 71 | + | } | |||
| 72 | + | calls := 0 | |||
| 73 | + | r := &execRunner{ | |||
| 74 | + | agent: a, | |||
| 75 | + | Command: []string{os.Args[0], "-test.run=^TestCredentialStdinHelperProcess$"}, | |||
| 76 | + | resolveCredential: func(context.Context, string) (Credential, error) { | |||
| 77 | + | calls++ | |||
| 78 | + | return Credential{Value: "inherited-secret", Method: "apikey"}, nil | |||
| 79 | + | }, | |||
| 80 | + | } | |||
| 81 | + | if err := r.Run(context.Background(), agentSink{a: a}); err != nil { | |||
| 82 | + | t.Fatal(err) | |||
| 83 | + | } | |||
| 84 | + | if calls != 1 { | |||
| 85 | + | t.Fatalf("credential resolver called %d times, want 1", calls) | |||
| 86 | + | } | |||
| 87 | + | } | |||
| 88 | + | ||||
| 89 | + | func TestApplyEventPreservesMultilineRolesAndFinalAssistant(t *testing.T) { | |||
| 90 | + | a := &Agent{} | |||
| 91 | + | sink := agentSink{a: a} | |||
| 92 | + | applyEventToSink(NewEvent("user_message", map[string]any{"content": []any{ | |||
| 93 | + | map[string]any{"type": "text", "text": "first line\n\nlast line"}, | |||
| 94 | + | }}), sink) | |||
| 95 | + | applyEventToSink(NewEvent("assistant_message", map[string]any{"content": []any{ | |||
| 96 | + | map[string]any{"type": "text", "text": "complete\nanswer"}, | |||
| 97 | + | }}), sink) | |||
| 98 | + | ||||
| 99 | + | snap := a.Snapshot() | |||
| 100 | + | wantLines := []string{"user: first line", "user: ", "user: last line", "complete", "answer"} | |||
| 101 | + | if strings.Join(snap.Lines, "\n") != strings.Join(wantLines, "\n") { | |||
| 102 | + | t.Fatalf("transcript = %#v, want %#v", snap.Lines, wantLines) | |||
| 103 | + | } | |||
| 104 | + | if snap.LastAssistant != "complete\nanswer" { | |||
| 105 | + | t.Fatalf("last assistant = %q", snap.LastAssistant) | |||
| 106 | + | } | |||
| 107 | + | } | |||
| 108 | + | ||||
| 28 | func TestSwarmAgentArgs(t *testing.T) { | 109 | func TestSwarmAgentArgs(t *testing.T) { | ||
| 29 | args := swarmAgentArgs(swarmAgentArgsOpts{ | 110 | args := swarmAgentArgs(swarmAgentArgsOpts{ | ||
| 30 | Exe: "/path/to/zot", | 111 | Exe: "/path/to/zot", | ||
M packages/agent/swarm/swarm.go +42 -18
| @@ -63,10 +63,22 @@ type Config struct { | |||
| 63 | 63 | // here. | |
| 64 | 64 | NewRunner func(a *Agent) Runner | |
| 65 | 65 | ||
| 66 | + | // ResolveCredential resolves the credential inherited by a child | |
| 67 | + | // process. The default runner transfers it over the child's stdin, | |
| 68 | + | // keeping command-backed credentials out of argv and the environment. | |
| 69 | + | ResolveCredential func(ctx context.Context, provider string) (Credential, error) | |
| 70 | + | ||
| 66 | 71 | // Now is a clock seam for tests; defaults to time.Now. | |
| 67 | 72 | Now func() time.Time | |
| 68 | 73 | } | |
| 69 | 74 | ||
| 75 | + | // Credential is the resolved authentication state inherited by a swarm child. | |
| 76 | + | type Credential struct { | |
| 77 | + | Value string `json:"value"` | |
| 78 | + | Method string `json:"method"` | |
| 79 | + | AccountID string `json:"account_id,omitempty"` | |
| 80 | + | } | |
| 81 | + | ||
| 70 | 82 | // Runner executes one agent task. Run blocks until the task finishes, | |
| 71 | 83 | // is cancelled via ctx, or hits an unrecoverable error. | |
| 72 | 84 | // | |
| @@ -117,7 +129,9 @@ func New(cfg Config) *Swarm { | |||
| 117 | 129 | cfg.Now = time.Now | |
| 118 | 130 | } | |
| 119 | 131 | if cfg.NewRunner == nil { | |
| 120 | - | cfg.NewRunner = func(a *Agent) Runner { return &execRunner{agent: a} } | |
| 132 | + | cfg.NewRunner = func(a *Agent) Runner { | |
| 133 | + | return &execRunner{agent: a, resolveCredential: cfg.ResolveCredential} | |
| 134 | + | } | |
| 121 | 135 | } | |
| 122 | 136 | return &Swarm{ | |
| 123 | 137 | cfg: cfg, | |
| @@ -418,16 +432,17 @@ func (f *Swarm) Remove(id string) error { | |||
| 418 | 432 | // Snapshot returns a read-only view of one agent. Safe for the TUI | |
| 419 | 433 | // goroutine to call repeatedly; never blocks on the Runner. | |
| 420 | 434 | type AgentSnapshot struct { | |
| 421 | - | ID string | |
| 422 | - | Task string | |
| 423 | - | Dir string | |
| 424 | - | Status Status | |
| 425 | - | Activity string | |
| 426 | - | Started time.Time | |
| 427 | - | Finished time.Time | |
| 428 | - | Err string | |
| 429 | - | Tail string // last few transcript lines, joined with "\n" | |
| 430 | - | Lines []string // full transcript (already capped by Agent.appendTranscript) | |
| 435 | + | ID string | |
| 436 | + | Task string | |
| 437 | + | Dir string | |
| 438 | + | Status Status | |
| 439 | + | Activity string | |
| 440 | + | Started time.Time | |
| 441 | + | Finished time.Time | |
| 442 | + | Err string | |
| 443 | + | Tail string // last few transcript lines, joined with "\n" | |
| 444 | + | Lines []string // full transcript (already capped by Agent.appendTranscript) | |
| 445 | + | LastAssistant string // complete final assistant message, without role formatting | |
| 431 | 446 | ||
| 432 | 447 | // Model and Provider expose the per-agent overrides set at | |
| 433 | 448 | // Spawn time (empty when the agent inherits the child's default | |
| @@ -461,11 +476,12 @@ func (a *Agent) Snapshot() AgentSnapshot { | |||
| 461 | 476 | Status: a.status, Activity: a.activity, | |
| 462 | 477 | Started: a.Started, Finished: a.finished, | |
| 463 | 478 | Err: errStr, Tail: tail, Lines: lines, | |
| 464 | - | Model: a.Model, | |
| 465 | - | Provider: a.Provider, | |
| 466 | - | InboxPath: a.InboxPath, | |
| 467 | - | EventLogPath: a.EventLogPath, | |
| 468 | - | SessionPath: a.SessionPath, | |
| 479 | + | LastAssistant: a.lastAssistant, | |
| 480 | + | Model: a.Model, | |
| 481 | + | Provider: a.Provider, | |
| 482 | + | InboxPath: a.InboxPath, | |
| 483 | + | EventLogPath: a.EventLogPath, | |
| 484 | + | SessionPath: a.SessionPath, | |
| 469 | 485 | } | |
| 470 | 486 | } | |
| 471 | 487 | ||
| @@ -501,8 +517,16 @@ func (f *Swarm) SnapshotAll() []AgentSnapshot { | |||
| 501 | 517 | // agentSink is the Sink the Swarm hands to each Runner. | |
| 502 | 518 | type agentSink struct{ a *Agent } | |
| 503 | 519 | ||
| 504 | - | func (s agentSink) Activity(msg string) { s.a.setActivity(msg) } | |
| 505 | - | func (s agentSink) Transcript(chunk string) { s.a.appendTranscript(chunk) } | |
| 520 | + | func (s agentSink) Activity(msg string) { s.a.setActivity(msg) } | |
| 521 | + | func (s agentSink) Transcript(chunk string) { | |
| 522 | + | if strings.HasPrefix(chunk, "stderr: ") || strings.HasPrefix(chunk, "error: ") { | |
| 523 | + | s.a.appendTranscript(chunk) | |
| 524 | + | return | |
| 525 | + | } | |
| 526 | + | s.a.appendAssistantMessage(chunk) | |
| 527 | + | } | |
| 528 | + | func (s agentSink) userMessage(text string) { s.a.appendUserMessage(text) } | |
| 529 | + | func (s agentSink) assistantMessage(text string) { s.a.appendAssistantMessage(text) } | |
| 506 | 530 | ||
| 507 | 531 | func truncate(s string, n int) string { | |
| 508 | 532 | if len(s) <= n { | |
| @@ -63,10 +63,22 @@ type Config struct { | |||||
| 63 | // here. | 63 | // here. | ||
| 64 | NewRunner func(a *Agent) Runner | 64 | NewRunner func(a *Agent) Runner | ||
| 65 | 65 | ||||
| 66 | + | // ResolveCredential resolves the credential inherited by a child | |||
| 67 | + | // process. The default runner transfers it over the child's stdin, | |||
| 68 | + | // keeping command-backed credentials out of argv and the environment. | |||
| 69 | + | ResolveCredential func(ctx context.Context, provider string) (Credential, error) | |||
| 70 | + | ||||
| 66 | // Now is a clock seam for tests; defaults to time.Now. | 71 | // Now is a clock seam for tests; defaults to time.Now. | ||
| 67 | Now func() time.Time | 72 | Now func() time.Time | ||
| 68 | } | 73 | } | ||
| 69 | 74 | ||||
| 75 | + | // Credential is the resolved authentication state inherited by a swarm child. | |||
| 76 | + | type Credential struct { | |||
| 77 | + | Value string `json:"value"` | |||
| 78 | + | Method string `json:"method"` | |||
| 79 | + | AccountID string `json:"account_id,omitempty"` | |||
| 80 | + | } | |||
| 81 | + | ||||
| 70 | // Runner executes one agent task. Run blocks until the task finishes, | 82 | // Runner executes one agent task. Run blocks until the task finishes, | ||
| 71 | // is cancelled via ctx, or hits an unrecoverable error. | 83 | // is cancelled via ctx, or hits an unrecoverable error. | ||
| 72 | // | 84 | // | ||
| @@ -117,7 +129,9 @@ func New(cfg Config) *Swarm { | |||||
| 117 | cfg.Now = time.Now | 129 | cfg.Now = time.Now | ||
| 118 | } | 130 | } | ||
| 119 | if cfg.NewRunner == nil { | 131 | if cfg.NewRunner == nil { | ||
| 120 | - | cfg.NewRunner = func(a *Agent) Runner { return &execRunner{agent: a} } | 132 | + | cfg.NewRunner = func(a *Agent) Runner { |
| 133 | + | return &execRunner{agent: a, resolveCredential: cfg.ResolveCredential} | |||
| 134 | + | } | |||
| 121 | } | 135 | } | ||
| 122 | return &Swarm{ | 136 | return &Swarm{ | ||
| 123 | cfg: cfg, | 137 | cfg: cfg, | ||
| @@ -418,16 +432,17 @@ func (f *Swarm) Remove(id string) error { | |||||
| 418 | // Snapshot returns a read-only view of one agent. Safe for the TUI | 432 | // Snapshot returns a read-only view of one agent. Safe for the TUI | ||
| 419 | // goroutine to call repeatedly; never blocks on the Runner. | 433 | // goroutine to call repeatedly; never blocks on the Runner. | ||
| 420 | type AgentSnapshot struct { | 434 | type AgentSnapshot struct { | ||
| 421 | - | ID string | 435 | + | ID string |
| 422 | - | Task string | 436 | + | Task string |
| 423 | - | Dir string | 437 | + | Dir string |
| 424 | - | Status Status | 438 | + | Status Status |
| 425 | - | Activity string | 439 | + | Activity string |
| 426 | - | Started time.Time | 440 | + | Started time.Time |
| 427 | - | Finished time.Time | 441 | + | Finished time.Time |
| 428 | - | Err string | 442 | + | Err string |
| 429 | - | Tail string // last few transcript lines, joined with "\n" | 443 | + | Tail string // last few transcript lines, joined with "\n" |
| 430 | - | Lines []string // full transcript (already capped by Agent.appendTranscript) | 444 | + | Lines []string // full transcript (already capped by Agent.appendTranscript) |
| 445 | + | LastAssistant string // complete final assistant message, without role formatting | |||
| 431 | 446 | ||||
| 432 | // Model and Provider expose the per-agent overrides set at | 447 | // Model and Provider expose the per-agent overrides set at | ||
| 433 | // Spawn time (empty when the agent inherits the child's default | 448 | // Spawn time (empty when the agent inherits the child's default | ||
| @@ -461,11 +476,12 @@ func (a *Agent) Snapshot() AgentSnapshot { | |||||
| 461 | Status: a.status, Activity: a.activity, | 476 | Status: a.status, Activity: a.activity, | ||
| 462 | Started: a.Started, Finished: a.finished, | 477 | Started: a.Started, Finished: a.finished, | ||
| 463 | Err: errStr, Tail: tail, Lines: lines, | 478 | Err: errStr, Tail: tail, Lines: lines, | ||
| 464 | - | Model: a.Model, | 479 | + | LastAssistant: a.lastAssistant, |
| 465 | - | Provider: a.Provider, | 480 | + | Model: a.Model, |
| 466 | - | InboxPath: a.InboxPath, | 481 | + | Provider: a.Provider, |
| 467 | - | EventLogPath: a.EventLogPath, | 482 | + | InboxPath: a.InboxPath, |
| 468 | - | SessionPath: a.SessionPath, | 483 | + | EventLogPath: a.EventLogPath, |
| 484 | + | SessionPath: a.SessionPath, | |||
| 469 | } | 485 | } | ||
| 470 | } | 486 | } | ||
| 471 | 487 | ||||
| @@ -501,8 +517,16 @@ func (f *Swarm) SnapshotAll() []AgentSnapshot { | |||||
| 501 | // agentSink is the Sink the Swarm hands to each Runner. | 517 | // agentSink is the Sink the Swarm hands to each Runner. | ||
| 502 | type agentSink struct{ a *Agent } | 518 | type agentSink struct{ a *Agent } | ||
| 503 | 519 | ||||
| 504 | - | func (s agentSink) Activity(msg string) { s.a.setActivity(msg) } | 520 | + | func (s agentSink) Activity(msg string) { s.a.setActivity(msg) } |
| 505 | - | func (s agentSink) Transcript(chunk string) { s.a.appendTranscript(chunk) } | 521 | + | func (s agentSink) Transcript(chunk string) { |
| 522 | + | if strings.HasPrefix(chunk, "stderr: ") || strings.HasPrefix(chunk, "error: ") { | |||
| 523 | + | s.a.appendTranscript(chunk) | |||
| 524 | + | return | |||
| 525 | + | } | |||
| 526 | + | s.a.appendAssistantMessage(chunk) | |||
| 527 | + | } | |||
| 528 | + | func (s agentSink) userMessage(text string) { s.a.appendUserMessage(text) } | |||
| 529 | + | func (s agentSink) assistantMessage(text string) { s.a.appendAssistantMessage(text) } | |||
| 506 | 530 | ||||
| 507 | func truncate(s string, n int) string { | 531 | func truncate(s string, n int) string { | ||
| 508 | if len(s) <= n { | 532 | if len(s) <= n { | ||
M packages/agent/swarm_agent.go +14 -0
| @@ -4,6 +4,7 @@ import ( | |||
| 4 | 4 | "context" | |
| 5 | 5 | "encoding/json" | |
| 6 | 6 | "fmt" | |
| 7 | + | "io" | |
| 7 | 8 | "os" | |
| 8 | 9 | "strings" | |
| 9 | 10 | "sync" | |
| @@ -36,6 +37,19 @@ func runSwarmAgentMode(ctx context.Context, args Args, version string) error { | |||
| 36 | 37 | if args.SwarmAgent == "" { | |
| 37 | 38 | return fmt.Errorf("--swarm-agent requires a socket path") | |
| 38 | 39 | } | |
| 40 | + | if os.Getenv("ZOT_SWARM_CREDENTIAL_STDIN") == "1" { | |
| 41 | + | var inherited swarm.Credential | |
| 42 | + | dec := json.NewDecoder(io.LimitReader(os.Stdin, 1<<20)) | |
| 43 | + | if err := dec.Decode(&inherited); err != nil { | |
| 44 | + | return fmt.Errorf("read inherited swarm credential: %w", err) | |
| 45 | + | } | |
| 46 | + | if inherited.Value == "" || inherited.Method == "" { | |
| 47 | + | return fmt.Errorf("read inherited swarm credential: missing value or method") | |
| 48 | + | } | |
| 49 | + | args.inheritedCredential = inherited.Value | |
| 50 | + | args.inheritedAuthMethod = inherited.Method | |
| 51 | + | args.inheritedAccountID = inherited.AccountID | |
| 52 | + | } | |
| 39 | 53 | ||
| 40 | 54 | r, err := Resolve(args, true) | |
| 41 | 55 | if err != nil { | |
| @@ -4,6 +4,7 @@ import ( | |||||
| 4 | "context" | 4 | "context" | ||
| 5 | "encoding/json" | 5 | "encoding/json" | ||
| 6 | "fmt" | 6 | "fmt" | ||
| 7 | + | "io" | |||
| 7 | "os" | 8 | "os" | ||
| 8 | "strings" | 9 | "strings" | ||
| 9 | "sync" | 10 | "sync" | ||
| @@ -36,6 +37,19 @@ func runSwarmAgentMode(ctx context.Context, args Args, version string) error { | |||||
| 36 | if args.SwarmAgent == "" { | 37 | if args.SwarmAgent == "" { | ||
| 37 | return fmt.Errorf("--swarm-agent requires a socket path") | 38 | return fmt.Errorf("--swarm-agent requires a socket path") | ||
| 38 | } | 39 | } | ||
| 40 | + | if os.Getenv("ZOT_SWARM_CREDENTIAL_STDIN") == "1" { | |||
| 41 | + | var inherited swarm.Credential | |||
| 42 | + | dec := json.NewDecoder(io.LimitReader(os.Stdin, 1<<20)) | |||
| 43 | + | if err := dec.Decode(&inherited); err != nil { | |||
| 44 | + | return fmt.Errorf("read inherited swarm credential: %w", err) | |||
| 45 | + | } | |||
| 46 | + | if inherited.Value == "" || inherited.Method == "" { | |||
| 47 | + | return fmt.Errorf("read inherited swarm credential: missing value or method") | |||
| 48 | + | } | |||
| 49 | + | args.inheritedCredential = inherited.Value | |||
| 50 | + | args.inheritedAuthMethod = inherited.Method | |||
| 51 | + | args.inheritedAccountID = inherited.AccountID | |||
| 52 | + | } | |||
| 39 | 53 | ||||
| 40 | r, err := Resolve(args, true) | 54 | r, err := Resolve(args, true) | ||
| 41 | if err != nil { | 55 | if err != nil { | ||