fix(subagents): address trace review findings
- commit
- a753677bdd4c10a5ab010510d57ad263c76320e8
- author
- brice · · 2026-08-12 06:45 UTC
- parent
- 895a89c
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
14 files · +305 -112
M README.md +1 -1
| @@ -419,7 +419,7 @@ Background subagents that run alongside your main session. Each one is a separat | |||
| 419 | 419 | ||
| 420 | 420 | **Session scoping** — each agent is stamped with the host session that spawned it and only shows up in that session's dashboard. Swap sessions with `/sessions` and the dashboard re-narrows accordingly. Agents from other sessions keep running in the background and reappear when you switch back. | |
| 421 | 421 | ||
| 422 | - | **Persistence across zut restarts** — every spawn writes durable metadata, a session file, and structured `result.json` under `$ZUT_HOME/subagents/agents/<id>/`. A local execution trace is an optional separate diagnostic bundle. On the next `zut` launch, retained session and result data can be used with `R` or `/subagents resume-session <id>` to continue the existing session. `/subagents restart-task <id>` is the explicit operation that intentionally replays the stored task. | |
| 422 | + | **Persistence across zut restarts** — every spawn writes durable metadata and a session file under `$ZUT_HOME/subagents/agents/<id>/`; a completed or terminated worker writes structured `result.json`. A local execution trace is an optional separate diagnostic bundle. On the next `zut` launch, retained session and result data can be used with `R` or `/subagents resume-session <id>` to continue the existing session. `/subagents restart-task <id>` is the explicit operation that intentionally replays the stored task. | |
| 423 | 423 | ||
| 424 | 424 | **Where state lives** — per-agent metadata, session files, results, and patches live under `$ZUT_HOME/subagents/agents/<id>/`; inbox sockets are runtime-only and permission-restricted. When enabled, trace bundles live below `ZUT_SUBAGENT_TRACE_DIR`; `zut debug trace <bundle>` reports the open operation, terminal fact, or last event. Shared-mode edits land directly in the repo. Worktree-mode edits are captured as durable patches and changed-file summaries before cleanup. | |
| 425 | 425 | ||
| @@ -419,7 +419,7 @@ Background subagents that run alongside your main session. Each one is a separat | |||||
| 419 | 419 | ||||
| 420 | **Session scoping** — each agent is stamped with the host session that spawned it and only shows up in that session's dashboard. Swap sessions with `/sessions` and the dashboard re-narrows accordingly. Agents from other sessions keep running in the background and reappear when you switch back. | 420 | **Session scoping** — each agent is stamped with the host session that spawned it and only shows up in that session's dashboard. Swap sessions with `/sessions` and the dashboard re-narrows accordingly. Agents from other sessions keep running in the background and reappear when you switch back. | ||
| 421 | 421 | ||||
| 422 | - | **Persistence across zut restarts** — every spawn writes durable metadata, a session file, and structured `result.json` under `$ZUT_HOME/subagents/agents/<id>/`. A local execution trace is an optional separate diagnostic bundle. On the next `zut` launch, retained session and result data can be used with `R` or `/subagents resume-session <id>` to continue the existing session. `/subagents restart-task <id>` is the explicit operation that intentionally replays the stored task. | 422 | + | **Persistence across zut restarts** — every spawn writes durable metadata and a session file under `$ZUT_HOME/subagents/agents/<id>/`; a completed or terminated worker writes structured `result.json`. A local execution trace is an optional separate diagnostic bundle. On the next `zut` launch, retained session and result data can be used with `R` or `/subagents resume-session <id>` to continue the existing session. `/subagents restart-task <id>` is the explicit operation that intentionally replays the stored task. |
| 423 | 423 | ||||
| 424 | **Where state lives** — per-agent metadata, session files, results, and patches live under `$ZUT_HOME/subagents/agents/<id>/`; inbox sockets are runtime-only and permission-restricted. When enabled, trace bundles live below `ZUT_SUBAGENT_TRACE_DIR`; `zut debug trace <bundle>` reports the open operation, terminal fact, or last event. Shared-mode edits land directly in the repo. Worktree-mode edits are captured as durable patches and changed-file summaries before cleanup. | 424 | **Where state lives** — per-agent metadata, session files, results, and patches live under `$ZUT_HOME/subagents/agents/<id>/`; inbox sockets are runtime-only and permission-restricted. When enabled, trace bundles live below `ZUT_SUBAGENT_TRACE_DIR`; `zut debug trace <bundle>` reports the open operation, terminal fact, or last event. Shared-mode edits land directly in the repo. Worktree-mode edits are captured as durable patches and changed-file summaries before cleanup. | ||
| 425 | 425 | ||||
M docs/subagents.md +1 -1
| @@ -164,7 +164,7 @@ When the relevant manager tool is permitted by launch-time policy, the model can | |||
| 164 | 164 | ||
| 165 | 165 | ## Local execution trace | |
| 166 | 166 | ||
| 167 | - | Set `ZUT_SUBAGENT_TRACE_DIR` to a private local directory before launching zut to retain an execution-trace bundle for each supervisor. Each bundle contains `manifest.json`, ordered `trace.jsonl`, and (only in detailed mode) private `payloads/` files. The default normal mode records redacted metadata. `ZUT_SUBAGENT_TRACE_MODE=detailed` retains unredacted prompts, tool arguments, and tool output in local trace files; enable it only for an appropriate investigation. Trace bundles are not uploaded or deleted automatically, and should be removed manually after the investigation. | |
| 167 | + | Set `ZUT_SUBAGENT_TRACE_DIR` to a private local directory before launching zut to retain an execution-trace bundle for each supervisor. Each bundle contains `manifest.json`, ordered `trace.jsonl`, and (only in detailed mode) private `payloads/` files. The default normal mode records redacted metadata. `ZUT_SUBAGENT_TRACE_MODE=detailed` retains unredacted prompts, tool arguments, and tool output that can contain credentials, secrets, personal data, or private source code. Enable it only when the local bundle can be protected and manually deleted afterwards; do not commit, upload, or otherwise expose trace data. Bundles request restrictive local permissions, but access control remains platform-dependent.\n\nExplicit runtime `TraceDir` and `TraceMode` configuration takes precedence over these environment variables. Environment mode values are case-insensitive (`normal` or `detailed`); an invalid value is ignored with a stderr warning. Detailed mode requires a non-empty trace directory, otherwise tracing remains disabled. | |
| 168 | 168 | ||
| 169 | 169 | Inspect a retained bundle with: | |
| 170 | 170 | ||
| @@ -164,7 +164,7 @@ When the relevant manager tool is permitted by launch-time policy, the model can | |||||
| 164 | 164 | ||||
| 165 | ## Local execution trace | 165 | ## Local execution trace | ||
| 166 | 166 | ||||
| 167 | - | Set `ZUT_SUBAGENT_TRACE_DIR` to a private local directory before launching zut to retain an execution-trace bundle for each supervisor. Each bundle contains `manifest.json`, ordered `trace.jsonl`, and (only in detailed mode) private `payloads/` files. The default normal mode records redacted metadata. `ZUT_SUBAGENT_TRACE_MODE=detailed` retains unredacted prompts, tool arguments, and tool output in local trace files; enable it only for an appropriate investigation. Trace bundles are not uploaded or deleted automatically, and should be removed manually after the investigation. | 167 | + | Set `ZUT_SUBAGENT_TRACE_DIR` to a private local directory before launching zut to retain an execution-trace bundle for each supervisor. Each bundle contains `manifest.json`, ordered `trace.jsonl`, and (only in detailed mode) private `payloads/` files. The default normal mode records redacted metadata. `ZUT_SUBAGENT_TRACE_MODE=detailed` retains unredacted prompts, tool arguments, and tool output that can contain credentials, secrets, personal data, or private source code. Enable it only when the local bundle can be protected and manually deleted afterwards; do not commit, upload, or otherwise expose trace data. Bundles request restrictive local permissions, but access control remains platform-dependent.\n\nExplicit runtime `TraceDir` and `TraceMode` configuration takes precedence over these environment variables. Environment mode values are case-insensitive (`normal` or `detailed`); an invalid value is ignored with a stderr warning. Detailed mode requires a non-empty trace directory, otherwise tracing remains disabled. |
| 168 | 168 | ||||
| 169 | Inspect a retained bundle with: | 169 | Inspect a retained bundle with: | ||
| 170 | 170 | ||||
M packages/agent/modes/subagent_indicator.go +1 -1
| @@ -40,7 +40,7 @@ func renderSubagentActivityLine(th tui.Theme, spinnerGlyph string, snapshot suba | |||
| 40 | 40 | if spinnerGlyph == "" { | |
| 41 | 41 | spinnerGlyph = "." | |
| 42 | 42 | } | |
| 43 | - | activity := strings.TrimSuffix(operation.Type, ".started") + " open" | |
| 43 | + | activity := operation.Label() | |
| 44 | 44 | age := formatSubagentActivityAge(operation.StartedAt, now) | |
| 45 | 45 | plain, layout := fitSubagentActivityLine(spinnerGlyph, name, activity, age, width-2) | |
| 46 | 46 | if plain == "" { | |
| @@ -40,7 +40,7 @@ func renderSubagentActivityLine(th tui.Theme, spinnerGlyph string, snapshot suba | |||||
| 40 | if spinnerGlyph == "" { | 40 | if spinnerGlyph == "" { | ||
| 41 | spinnerGlyph = "." | 41 | spinnerGlyph = "." | ||
| 42 | } | 42 | } | ||
| 43 | - | activity := strings.TrimSuffix(operation.Type, ".started") + " open" | 43 | + | activity := operation.Label() |
| 44 | age := formatSubagentActivityAge(operation.StartedAt, now) | 44 | age := formatSubagentActivityAge(operation.StartedAt, now) | ||
| 45 | plain, layout := fitSubagentActivityLine(spinnerGlyph, name, activity, age, width-2) | 45 | plain, layout := fitSubagentActivityLine(spinnerGlyph, name, activity, age, width-2) | ||
| 46 | if plain == "" { | 46 | if plain == "" { | ||
M packages/agent/modes/subagents_dialog.go +11 -10
| @@ -344,7 +344,7 @@ func (d *subagentsDialog) transcriptEditorCursorRow(width, popupRows, editorRowO | |||
| 344 | 344 | } | |
| 345 | 345 | ||
| 346 | 346 | row += popupRows // @-file-suggest popup, if active | |
| 347 | - | if d.agentHasOpenOperation(a.ID) { | |
| 347 | + | if a.Status == subagents.StatusRunning && d.send != nil && d.agentHasOpenOperation(a.ID) { | |
| 348 | 348 | row += 2 // blank + spinner row inserted above the editor | |
| 349 | 349 | } | |
| 350 | 350 | row++ // blank above editor (appendTranscriptEditor) | |
| @@ -1462,7 +1462,7 @@ func (d *subagentsDialog) traceSummary(id string) string { | |||
| 1462 | 1462 | return view.Terminal | |
| 1463 | 1463 | } | |
| 1464 | 1464 | if len(view.OpenOperations) != 0 { | |
| 1465 | - | return strings.TrimSuffix(view.OpenOperations[0].Type, ".started") + " open" | |
| 1465 | + | return view.OpenOperations[0].Label() | |
| 1466 | 1466 | } | |
| 1467 | 1467 | if view.LastEvent.Type != "" { | |
| 1468 | 1468 | return "last event " + view.LastEvent.Type | |
| @@ -1618,13 +1618,7 @@ func (d *subagentsDialog) renderPromptEditor(th tui.Theme, width int, out []stri | |||
| 1618 | 1618 | return out | |
| 1619 | 1619 | } | |
| 1620 | 1620 | ||
| 1621 | - | // formatSupervisorRow is the one-line summary shown per agent. | |
| 1622 | - | // | |
| 1623 | - | // Layout (fixed-width columns, then free-form activity): | |
| 1624 | - | // | |
| 1625 | - | // OPERATION ID AGE TASK | |
| 1626 | - | // tool open 3m fix-login-12345 3m inspect main.go | |
| 1627 | - | // completed write-tests-67890 1h add coverage | |
| 1621 | + | // traceView retrieves the latest shared factual trace projection for id. | |
| 1628 | 1622 | func (d *subagentsDialog) traceView(id string) subagents.AgentTraceView { | |
| 1629 | 1623 | if d != nil && d.traceViews != nil { | |
| 1630 | 1624 | return d.traceViews()[id] | |
| @@ -1632,13 +1626,20 @@ func (d *subagentsDialog) traceView(id string) subagents.AgentTraceView { | |||
| 1632 | 1626 | return subagents.AgentTraceView{AgentID: id} | |
| 1633 | 1627 | } | |
| 1634 | 1628 | ||
| 1629 | + | // formatSupervisorRow is the one-line summary shown per agent. | |
| 1630 | + | // | |
| 1631 | + | // Layout (fixed-width columns, then free-form task): | |
| 1632 | + | // | |
| 1633 | + | // OPERATION ID AGE TASK | |
| 1634 | + | // tool open 3m fix-login-12345 3m inspect main.go | |
| 1635 | + | // completed write-tests-67890 1h add coverage | |
| 1635 | 1636 | func formatSupervisorRow(r subagents.AgentSnapshot, view subagents.AgentTraceView, maxWidth int) string { | |
| 1636 | 1637 | fact := "no observable operation" | |
| 1637 | 1638 | if view.Terminal != "" { | |
| 1638 | 1639 | fact = view.Terminal | |
| 1639 | 1640 | } else if len(view.OpenOperations) != 0 { | |
| 1640 | 1641 | operation := view.OpenOperations[0] | |
| 1641 | - | fact = strings.TrimSuffix(operation.Type, ".started") + " open " + formatAge(operation.StartedAt) | |
| 1642 | + | fact = operation.Label() + " " + formatAge(operation.StartedAt) | |
| 1642 | 1643 | } else if view.LastEvent.Type != "" { | |
| 1643 | 1644 | fact = "last " + view.LastEvent.Type + " " + formatAge(view.LastEvent.Timestamp) | |
| 1644 | 1645 | } | |
| @@ -344,7 +344,7 @@ func (d *subagentsDialog) transcriptEditorCursorRow(width, popupRows, editorRowO | |||||
| 344 | } | 344 | } | ||
| 345 | 345 | ||||
| 346 | row += popupRows // @-file-suggest popup, if active | 346 | row += popupRows // @-file-suggest popup, if active | ||
| 347 | - | if d.agentHasOpenOperation(a.ID) { | 347 | + | if a.Status == subagents.StatusRunning && d.send != nil && d.agentHasOpenOperation(a.ID) { |
| 348 | row += 2 // blank + spinner row inserted above the editor | 348 | row += 2 // blank + spinner row inserted above the editor | ||
| 349 | } | 349 | } | ||
| 350 | row++ // blank above editor (appendTranscriptEditor) | 350 | row++ // blank above editor (appendTranscriptEditor) | ||
| @@ -1462,7 +1462,7 @@ func (d *subagentsDialog) traceSummary(id string) string { | |||||
| 1462 | return view.Terminal | 1462 | return view.Terminal | ||
| 1463 | } | 1463 | } | ||
| 1464 | if len(view.OpenOperations) != 0 { | 1464 | if len(view.OpenOperations) != 0 { | ||
| 1465 | - | return strings.TrimSuffix(view.OpenOperations[0].Type, ".started") + " open" | 1465 | + | return view.OpenOperations[0].Label() |
| 1466 | } | 1466 | } | ||
| 1467 | if view.LastEvent.Type != "" { | 1467 | if view.LastEvent.Type != "" { | ||
| 1468 | return "last event " + view.LastEvent.Type | 1468 | return "last event " + view.LastEvent.Type | ||
| @@ -1618,13 +1618,7 @@ func (d *subagentsDialog) renderPromptEditor(th tui.Theme, width int, out []stri | |||||
| 1618 | return out | 1618 | return out | ||
| 1619 | } | 1619 | } | ||
| 1620 | 1620 | ||||
| 1621 | - | // formatSupervisorRow is the one-line summary shown per agent. | 1621 | + | // traceView retrieves the latest shared factual trace projection for id. |
| 1622 | - | // | |||
| 1623 | - | // Layout (fixed-width columns, then free-form activity): | |||
| 1624 | - | // | |||
| 1625 | - | // OPERATION ID AGE TASK | |||
| 1626 | - | // tool open 3m fix-login-12345 3m inspect main.go | |||
| 1627 | - | // completed write-tests-67890 1h add coverage | |||
| 1628 | func (d *subagentsDialog) traceView(id string) subagents.AgentTraceView { | 1622 | func (d *subagentsDialog) traceView(id string) subagents.AgentTraceView { | ||
| 1629 | if d != nil && d.traceViews != nil { | 1623 | if d != nil && d.traceViews != nil { | ||
| 1630 | return d.traceViews()[id] | 1624 | return d.traceViews()[id] | ||
| @@ -1632,13 +1626,20 @@ func (d *subagentsDialog) traceView(id string) subagents.AgentTraceView { | |||||
| 1632 | return subagents.AgentTraceView{AgentID: id} | 1626 | return subagents.AgentTraceView{AgentID: id} | ||
| 1633 | } | 1627 | } | ||
| 1634 | 1628 | ||||
| 1629 | + | // formatSupervisorRow is the one-line summary shown per agent. | |||
| 1630 | + | // | |||
| 1631 | + | // Layout (fixed-width columns, then free-form task): | |||
| 1632 | + | // | |||
| 1633 | + | // OPERATION ID AGE TASK | |||
| 1634 | + | // tool open 3m fix-login-12345 3m inspect main.go | |||
| 1635 | + | // completed write-tests-67890 1h add coverage | |||
| 1635 | func formatSupervisorRow(r subagents.AgentSnapshot, view subagents.AgentTraceView, maxWidth int) string { | 1636 | func formatSupervisorRow(r subagents.AgentSnapshot, view subagents.AgentTraceView, maxWidth int) string { | ||
| 1636 | fact := "no observable operation" | 1637 | fact := "no observable operation" | ||
| 1637 | if view.Terminal != "" { | 1638 | if view.Terminal != "" { | ||
| 1638 | fact = view.Terminal | 1639 | fact = view.Terminal | ||
| 1639 | } else if len(view.OpenOperations) != 0 { | 1640 | } else if len(view.OpenOperations) != 0 { | ||
| 1640 | operation := view.OpenOperations[0] | 1641 | operation := view.OpenOperations[0] | ||
| 1641 | - | fact = strings.TrimSuffix(operation.Type, ".started") + " open " + formatAge(operation.StartedAt) | 1642 | + | fact = operation.Label() + " " + formatAge(operation.StartedAt) |
| 1642 | } else if view.LastEvent.Type != "" { | 1643 | } else if view.LastEvent.Type != "" { | ||
| 1643 | fact = "last " + view.LastEvent.Type + " " + formatAge(view.LastEvent.Timestamp) | 1644 | fact = "last " + view.LastEvent.Type + " " + formatAge(view.LastEvent.Timestamp) | ||
| 1644 | } | 1645 | } | ||
M packages/agent/subagent_runtime.go +18 -6
| @@ -2,6 +2,8 @@ package agent | |||
| 2 | 2 | ||
| 3 | 3 | import ( | |
| 4 | 4 | "context" | |
| 5 | + | "errors" | |
| 6 | + | "fmt" | |
| 5 | 7 | "os" | |
| 6 | 8 | "path/filepath" | |
| 7 | 9 | "strings" | |
| @@ -104,8 +106,19 @@ func newSubagentRuntime(cfg subagentRuntimeConfig) *subagentRuntime { | |||
| 104 | 106 | if cfg.TraceDir == "" { | |
| 105 | 107 | cfg.TraceDir = strings.TrimSpace(os.Getenv("ZUT_SUBAGENT_TRACE_DIR")) | |
| 106 | 108 | } | |
| 107 | - | if mode := strings.TrimSpace(os.Getenv("ZUT_SUBAGENT_TRACE_MODE")); mode == string(subagents.TraceModeDetailed) { | |
| 108 | - | cfg.TraceMode = subagents.TraceModeDetailed | |
| 109 | + | if cfg.TraceMode == "" { | |
| 110 | + | switch strings.ToLower(strings.TrimSpace(os.Getenv("ZUT_SUBAGENT_TRACE_MODE"))) { | |
| 111 | + | case "", string(subagents.TraceModeNormal): | |
| 112 | + | cfg.TraceMode = subagents.TraceModeNormal | |
| 113 | + | case string(subagents.TraceModeDetailed): | |
| 114 | + | cfg.TraceMode = subagents.TraceModeDetailed | |
| 115 | + | default: | |
| 116 | + | fmt.Fprintf(os.Stderr, "zut: ignoring invalid ZUT_SUBAGENT_TRACE_MODE\n") | |
| 117 | + | cfg.TraceMode = subagents.TraceModeNormal | |
| 118 | + | } | |
| 119 | + | } | |
| 120 | + | if cfg.TraceDir == "" { | |
| 121 | + | cfg.TraceMode = subagents.TraceModeNormal | |
| 109 | 122 | } | |
| 110 | 123 | rt := &subagentRuntime{ | |
| 111 | 124 | args: cfg.Args, | |
| @@ -460,10 +473,9 @@ func (rt *subagentRuntime) Close(ctx context.Context) error { | |||
| 460 | 473 | if rt.closed { | |
| 461 | 474 | return nil | |
| 462 | 475 | } | |
| 463 | - | if err := rt.supervisor.StopAllContext(ctx); err != nil { | |
| 464 | - | return err | |
| 465 | - | } | |
| 466 | - | if err := rt.supervisor.Close(); err != nil { | |
| 476 | + | stopErr := rt.supervisor.StopAllContext(ctx) | |
| 477 | + | closeErr := rt.supervisor.CloseContext(ctx) | |
| 478 | + | if err := errors.Join(stopErr, closeErr); err != nil { | |
| 467 | 479 | return err | |
| 468 | 480 | } | |
| 469 | 481 | rt.closed = true | |
| @@ -2,6 +2,8 @@ package agent | |||||
| 2 | 2 | ||||
| 3 | import ( | 3 | import ( | ||
| 4 | "context" | 4 | "context" | ||
| 5 | + | "errors" | |||
| 6 | + | "fmt" | |||
| 5 | "os" | 7 | "os" | ||
| 6 | "path/filepath" | 8 | "path/filepath" | ||
| 7 | "strings" | 9 | "strings" | ||
| @@ -104,8 +106,19 @@ func newSubagentRuntime(cfg subagentRuntimeConfig) *subagentRuntime { | |||||
| 104 | if cfg.TraceDir == "" { | 106 | if cfg.TraceDir == "" { | ||
| 105 | cfg.TraceDir = strings.TrimSpace(os.Getenv("ZUT_SUBAGENT_TRACE_DIR")) | 107 | cfg.TraceDir = strings.TrimSpace(os.Getenv("ZUT_SUBAGENT_TRACE_DIR")) | ||
| 106 | } | 108 | } | ||
| 107 | - | if mode := strings.TrimSpace(os.Getenv("ZUT_SUBAGENT_TRACE_MODE")); mode == string(subagents.TraceModeDetailed) { | 109 | + | if cfg.TraceMode == "" { |
| 108 | - | cfg.TraceMode = subagents.TraceModeDetailed | 110 | + | switch strings.ToLower(strings.TrimSpace(os.Getenv("ZUT_SUBAGENT_TRACE_MODE"))) { |
| 111 | + | case "", string(subagents.TraceModeNormal): | |||
| 112 | + | cfg.TraceMode = subagents.TraceModeNormal | |||
| 113 | + | case string(subagents.TraceModeDetailed): | |||
| 114 | + | cfg.TraceMode = subagents.TraceModeDetailed | |||
| 115 | + | default: | |||
| 116 | + | fmt.Fprintf(os.Stderr, "zut: ignoring invalid ZUT_SUBAGENT_TRACE_MODE\n") | |||
| 117 | + | cfg.TraceMode = subagents.TraceModeNormal | |||
| 118 | + | } | |||
| 119 | + | } | |||
| 120 | + | if cfg.TraceDir == "" { | |||
| 121 | + | cfg.TraceMode = subagents.TraceModeNormal | |||
| 109 | } | 122 | } | ||
| 110 | rt := &subagentRuntime{ | 123 | rt := &subagentRuntime{ | ||
| 111 | args: cfg.Args, | 124 | args: cfg.Args, | ||
| @@ -460,10 +473,9 @@ func (rt *subagentRuntime) Close(ctx context.Context) error { | |||||
| 460 | if rt.closed { | 473 | if rt.closed { | ||
| 461 | return nil | 474 | return nil | ||
| 462 | } | 475 | } | ||
| 463 | - | if err := rt.supervisor.StopAllContext(ctx); err != nil { | 476 | + | stopErr := rt.supervisor.StopAllContext(ctx) |
| 464 | - | return err | 477 | + | closeErr := rt.supervisor.CloseContext(ctx) |
| 465 | - | } | 478 | + | if err := errors.Join(stopErr, closeErr); err != nil { |
| 466 | - | if err := rt.supervisor.Close(); err != nil { | |||
| 467 | return err | 479 | return err | ||
| 468 | } | 480 | } | ||
| 469 | rt.closed = true | 481 | rt.closed = true | ||
M packages/agent/subagents/projection.go +28 -28
| @@ -2,37 +2,36 @@ package subagents | |||
| 2 | 2 | ||
| 3 | 3 | import ( | |
| 4 | 4 | "sort" | |
| 5 | + | "strings" | |
| 5 | 6 | "time" | |
| 6 | 7 | ) | |
| 7 | 8 | ||
| 8 | - | // Operation is a factual operation inferred from paired trace boundaries. An | |
| 9 | - | // empty FinishedAt means that no terminal boundary has been observed yet. | |
| 9 | + | // Operation is a currently open factual operation inferred from its start | |
| 10 | + | // boundary. Completed operations are deliberately not retained in the public | |
| 11 | + | // projection: their terminal trace fact remains available as LastEvent. | |
| 10 | 12 | type Operation struct { | |
| 11 | - | Type string | |
| 12 | - | AgentID string | |
| 13 | - | TurnID string | |
| 14 | - | StartedAt time.Time | |
| 15 | - | FinishedAt time.Time | |
| 16 | - | Outcome string | |
| 13 | + | Type string | |
| 14 | + | AgentID string | |
| 15 | + | TurnID string | |
| 16 | + | CallID string | |
| 17 | + | StartedAt time.Time | |
| 17 | 18 | } | |
| 18 | 19 | ||
| 19 | - | // Open reports whether the trace contains a start boundary without a matching | |
| 20 | - | // terminal boundary. | |
| 21 | - | func (o Operation) Open() bool { return o.FinishedAt.IsZero() } | |
| 20 | + | // Open is always true because the projection exposes only unmatched starts. | |
| 21 | + | func (o Operation) Open() bool { return true } | |
| 22 | 22 | ||
| 23 | - | // Duration measures an observed operation. Open operations are measured at | |
| 24 | - | // now; callers supply now to make rendering and tests deterministic. | |
| 23 | + | // Duration measures this open operation at now; callers supply now to make | |
| 24 | + | // rendering and tests deterministic. | |
| 25 | 25 | func (o Operation) Duration(now time.Time) time.Duration { | |
| 26 | - | end := o.FinishedAt | |
| 27 | - | if end.IsZero() { | |
| 28 | - | end = now | |
| 29 | - | } | |
| 30 | - | if end.Before(o.StartedAt) { | |
| 26 | + | if now.Before(o.StartedAt) { | |
| 31 | 27 | return 0 | |
| 32 | 28 | } | |
| 33 | - | return end.Sub(o.StartedAt) | |
| 29 | + | return now.Sub(o.StartedAt) | |
| 34 | 30 | } | |
| 35 | 31 | ||
| 32 | + | // Label returns the shared concise label for an open operation. | |
| 33 | + | func (o Operation) Label() string { return strings.TrimSuffix(o.Type, ".started") + " open" } | |
| 34 | + | ||
| 36 | 35 | // AgentTraceView is the only user-facing execution state. It intentionally | |
| 37 | 36 | // contains operations and terminal facts rather than lifecycle guesses such as | |
| 38 | 37 | // working, idle, or alive. | |
| @@ -60,17 +59,15 @@ func ProjectTrace(events []TraceEvent) map[string]AgentTraceView { | |||
| 60 | 59 | if open[event.AgentID] == nil { | |
| 61 | 60 | open[event.AgentID] = make(map[string]Operation) | |
| 62 | 61 | } | |
| 63 | - | open[event.AgentID][key] = Operation{Type: event.Type, AgentID: event.AgentID, TurnID: event.TurnID, StartedAt: event.Timestamp} | |
| 62 | + | callID, _ := event.Data["call_id"].(string) | |
| 63 | + | open[event.AgentID][key] = Operation{Type: event.Type, AgentID: event.AgentID, TurnID: event.TurnID, CallID: callID, StartedAt: event.Timestamp} | |
| 64 | 64 | } | |
| 65 | 65 | if ends { | |
| 66 | - | if operation, ok := open[event.AgentID][key]; ok { | |
| 67 | - | operation.FinishedAt = event.Timestamp | |
| 68 | - | operation.Outcome = event.Type | |
| 69 | - | delete(open[event.AgentID], key) | |
| 70 | - | } | |
| 66 | + | delete(open[event.AgentID], key) | |
| 71 | 67 | } | |
| 72 | 68 | if terminal != "" { | |
| 73 | 69 | view.Terminal = terminal | |
| 70 | + | delete(open, event.AgentID) | |
| 74 | 71 | } | |
| 75 | 72 | views[event.AgentID] = view | |
| 76 | 73 | } | |
| @@ -87,7 +84,10 @@ func ProjectTrace(events []TraceEvent) map[string]AgentTraceView { | |||
| 87 | 84 | if left.Type != right.Type { | |
| 88 | 85 | return left.Type < right.Type | |
| 89 | 86 | } | |
| 90 | - | return left.TurnID < right.TurnID | |
| 87 | + | if left.TurnID != right.TurnID { | |
| 88 | + | return left.TurnID < right.TurnID | |
| 89 | + | } | |
| 90 | + | return left.CallID < right.CallID | |
| 91 | 91 | }) | |
| 92 | 92 | views[agentID] = view | |
| 93 | 93 | } | |
| @@ -100,11 +100,11 @@ func traceBoundary(event TraceEvent) (key string, starts, ends bool, terminal st | |||
| 100 | 100 | switch event.Type { | |
| 101 | 101 | case "turn.started": | |
| 102 | 102 | return "turn:" + event.TurnID, true, false, "" | |
| 103 | - | case "turn.finished", "turn.failed": | |
| 103 | + | case "turn.finished", "turn.failed", "turn.cancelled": | |
| 104 | 104 | return "turn:" + event.TurnID, false, true, "" | |
| 105 | 105 | case "tool.started": | |
| 106 | 106 | return "tool:" + event.TurnID + ":" + callID, true, false, "" | |
| 107 | - | case "tool.finished": | |
| 107 | + | case "tool.finished", "tool.failed", "tool.cancelled": | |
| 108 | 108 | return "tool:" + event.TurnID + ":" + callID, false, true, "" | |
| 109 | 109 | case "agent.finished": | |
| 110 | 110 | return key, false, false, "completed" | |
| @@ -2,37 +2,36 @@ package subagents | |||||
| 2 | 2 | ||||
| 3 | import ( | 3 | import ( | ||
| 4 | "sort" | 4 | "sort" | ||
| 5 | + | "strings" | |||
| 5 | "time" | 6 | "time" | ||
| 6 | ) | 7 | ) | ||
| 7 | 8 | ||||
| 8 | - | // Operation is a factual operation inferred from paired trace boundaries. An | 9 | + | // Operation is a currently open factual operation inferred from its start |
| 9 | - | // empty FinishedAt means that no terminal boundary has been observed yet. | 10 | + | // boundary. Completed operations are deliberately not retained in the public |
| 11 | + | // projection: their terminal trace fact remains available as LastEvent. | |||
| 10 | type Operation struct { | 12 | type Operation struct { | ||
| 11 | - | Type string | 13 | + | Type string |
| 12 | - | AgentID string | 14 | + | AgentID string |
| 13 | - | TurnID string | 15 | + | TurnID string |
| 14 | - | StartedAt time.Time | 16 | + | CallID string |
| 15 | - | FinishedAt time.Time | 17 | + | StartedAt time.Time |
| 16 | - | Outcome string | |||
| 17 | } | 18 | } | ||
| 18 | 19 | ||||
| 19 | - | // Open reports whether the trace contains a start boundary without a matching | 20 | + | // Open is always true because the projection exposes only unmatched starts. |
| 20 | - | // terminal boundary. | 21 | + | func (o Operation) Open() bool { return true } |
| 21 | - | func (o Operation) Open() bool { return o.FinishedAt.IsZero() } | |||
| 22 | 22 | ||||
| 23 | - | // Duration measures an observed operation. Open operations are measured at | 23 | + | // Duration measures this open operation at now; callers supply now to make |
| 24 | - | // now; callers supply now to make rendering and tests deterministic. | 24 | + | // rendering and tests deterministic. |
| 25 | func (o Operation) Duration(now time.Time) time.Duration { | 25 | func (o Operation) Duration(now time.Time) time.Duration { | ||
| 26 | - | end := o.FinishedAt | 26 | + | if now.Before(o.StartedAt) { |
| 27 | - | if end.IsZero() { | |||
| 28 | - | end = now | |||
| 29 | - | } | |||
| 30 | - | if end.Before(o.StartedAt) { | |||
| 31 | return 0 | 27 | return 0 | ||
| 32 | } | 28 | } | ||
| 33 | - | return end.Sub(o.StartedAt) | 29 | + | return now.Sub(o.StartedAt) |
| 34 | } | 30 | } | ||
| 35 | 31 | ||||
| 32 | + | // Label returns the shared concise label for an open operation. | |||
| 33 | + | func (o Operation) Label() string { return strings.TrimSuffix(o.Type, ".started") + " open" } | |||
| 34 | + | ||||
| 36 | // AgentTraceView is the only user-facing execution state. It intentionally | 35 | // AgentTraceView is the only user-facing execution state. It intentionally | ||
| 37 | // contains operations and terminal facts rather than lifecycle guesses such as | 36 | // contains operations and terminal facts rather than lifecycle guesses such as | ||
| 38 | // working, idle, or alive. | 37 | // working, idle, or alive. | ||
| @@ -60,17 +59,15 @@ func ProjectTrace(events []TraceEvent) map[string]AgentTraceView { | |||||
| 60 | if open[event.AgentID] == nil { | 59 | if open[event.AgentID] == nil { | ||
| 61 | open[event.AgentID] = make(map[string]Operation) | 60 | open[event.AgentID] = make(map[string]Operation) | ||
| 62 | } | 61 | } | ||
| 63 | - | open[event.AgentID][key] = Operation{Type: event.Type, AgentID: event.AgentID, TurnID: event.TurnID, StartedAt: event.Timestamp} | 62 | + | callID, _ := event.Data["call_id"].(string) |
| 63 | + | open[event.AgentID][key] = Operation{Type: event.Type, AgentID: event.AgentID, TurnID: event.TurnID, CallID: callID, StartedAt: event.Timestamp} | |||
| 64 | } | 64 | } | ||
| 65 | if ends { | 65 | if ends { | ||
| 66 | - | if operation, ok := open[event.AgentID][key]; ok { | 66 | + | delete(open[event.AgentID], key) |
| 67 | - | operation.FinishedAt = event.Timestamp | |||
| 68 | - | operation.Outcome = event.Type | |||
| 69 | - | delete(open[event.AgentID], key) | |||
| 70 | - | } | |||
| 71 | } | 67 | } | ||
| 72 | if terminal != "" { | 68 | if terminal != "" { | ||
| 73 | view.Terminal = terminal | 69 | view.Terminal = terminal | ||
| 70 | + | delete(open, event.AgentID) | |||
| 74 | } | 71 | } | ||
| 75 | views[event.AgentID] = view | 72 | views[event.AgentID] = view | ||
| 76 | } | 73 | } | ||
| @@ -87,7 +84,10 @@ func ProjectTrace(events []TraceEvent) map[string]AgentTraceView { | |||||
| 87 | if left.Type != right.Type { | 84 | if left.Type != right.Type { | ||
| 88 | return left.Type < right.Type | 85 | return left.Type < right.Type | ||
| 89 | } | 86 | } | ||
| 90 | - | return left.TurnID < right.TurnID | 87 | + | if left.TurnID != right.TurnID { |
| 88 | + | return left.TurnID < right.TurnID | |||
| 89 | + | } | |||
| 90 | + | return left.CallID < right.CallID | |||
| 91 | }) | 91 | }) | ||
| 92 | views[agentID] = view | 92 | views[agentID] = view | ||
| 93 | } | 93 | } | ||
| @@ -100,11 +100,11 @@ func traceBoundary(event TraceEvent) (key string, starts, ends bool, terminal st | |||||
| 100 | switch event.Type { | 100 | switch event.Type { | ||
| 101 | case "turn.started": | 101 | case "turn.started": | ||
| 102 | return "turn:" + event.TurnID, true, false, "" | 102 | return "turn:" + event.TurnID, true, false, "" | ||
| 103 | - | case "turn.finished", "turn.failed": | 103 | + | case "turn.finished", "turn.failed", "turn.cancelled": |
| 104 | return "turn:" + event.TurnID, false, true, "" | 104 | return "turn:" + event.TurnID, false, true, "" | ||
| 105 | case "tool.started": | 105 | case "tool.started": | ||
| 106 | return "tool:" + event.TurnID + ":" + callID, true, false, "" | 106 | return "tool:" + event.TurnID + ":" + callID, true, false, "" | ||
| 107 | - | case "tool.finished": | 107 | + | case "tool.finished", "tool.failed", "tool.cancelled": |
| 108 | return "tool:" + event.TurnID + ":" + callID, false, true, "" | 108 | return "tool:" + event.TurnID + ":" + callID, false, true, "" | ||
| 109 | case "agent.finished": | 109 | case "agent.finished": | ||
| 110 | return key, false, false, "completed" | 110 | return key, false, false, "completed" | ||
M packages/agent/subagents/projection_test.go +24 -2
| @@ -36,10 +36,32 @@ func TestProjectTraceClosesPairedOperationsAndKeepsTerminalFact(t *testing.T) { | |||
| 36 | 36 | {Seq: 6, Timestamp: started.Add(5 * time.Second), Type: "agent.finished", AgentID: "agent-1"}, | |
| 37 | 37 | }) | |
| 38 | 38 | view := views["agent-1"] | |
| 39 | - | if len(view.OpenOperations) != 1 || view.OpenOperations[0].Type != "tool.started" { | |
| 40 | - | t.Fatalf("open operations = %#v, want unmatched tool", view.OpenOperations) | |
| 39 | + | if len(view.OpenOperations) != 0 { | |
| 40 | + | t.Fatalf("open operations = %#v, want none after terminal agent event", view.OpenOperations) | |
| 41 | 41 | } | |
| 42 | 42 | if view.Terminal != "completed" { | |
| 43 | 43 | t.Fatalf("terminal = %q, want completed", view.Terminal) | |
| 44 | 44 | } | |
| 45 | 45 | } | |
| 46 | + | ||
| 47 | + | func TestProjectTraceClosesFailedToolByCallID(t *testing.T) { | |
| 48 | + | started := time.Date(2026, 8, 12, 7, 0, 0, 0, time.UTC) | |
| 49 | + | view := ProjectTrace([]TraceEvent{ | |
| 50 | + | {Timestamp: started, Type: "tool.started", AgentID: "agent-1", TurnID: "turn-1", Data: map[string]any{"call_id": "one"}}, | |
| 51 | + | {Timestamp: started, Type: "tool.started", AgentID: "agent-1", TurnID: "turn-1", Data: map[string]any{"call_id": "two"}}, | |
| 52 | + | {Timestamp: started.Add(time.Second), Type: "tool.failed", AgentID: "agent-1", TurnID: "turn-1", Data: map[string]any{"call_id": "one"}}, | |
| 53 | + | })["agent-1"] | |
| 54 | + | if len(view.OpenOperations) != 1 || view.OpenOperations[0].CallID != "two" { | |
| 55 | + | t.Fatalf("open operations = %#v, want only call two", view.OpenOperations) | |
| 56 | + | } | |
| 57 | + | } | |
| 58 | + | ||
| 59 | + | func TestProjectTraceFailedAgentClosesOpenOperations(t *testing.T) { | |
| 60 | + | view := ProjectTrace([]TraceEvent{ | |
| 61 | + | {Type: "tool.started", AgentID: "agent-1", TurnID: "turn-1", Data: map[string]any{"call_id": "one"}}, | |
| 62 | + | {Type: "agent.failed", AgentID: "agent-1"}, | |
| 63 | + | })["agent-1"] | |
| 64 | + | if len(view.OpenOperations) != 0 || view.Terminal != "failed" { | |
| 65 | + | t.Fatalf("view = %#v, want failed terminal with no open operations", view) | |
| 66 | + | } | |
| 67 | + | } | |
| @@ -36,10 +36,32 @@ func TestProjectTraceClosesPairedOperationsAndKeepsTerminalFact(t *testing.T) { | |||||
| 36 | {Seq: 6, Timestamp: started.Add(5 * time.Second), Type: "agent.finished", AgentID: "agent-1"}, | 36 | {Seq: 6, Timestamp: started.Add(5 * time.Second), Type: "agent.finished", AgentID: "agent-1"}, | ||
| 37 | }) | 37 | }) | ||
| 38 | view := views["agent-1"] | 38 | view := views["agent-1"] | ||
| 39 | - | if len(view.OpenOperations) != 1 || view.OpenOperations[0].Type != "tool.started" { | 39 | + | if len(view.OpenOperations) != 0 { |
| 40 | - | t.Fatalf("open operations = %#v, want unmatched tool", view.OpenOperations) | 40 | + | t.Fatalf("open operations = %#v, want none after terminal agent event", view.OpenOperations) |
| 41 | } | 41 | } | ||
| 42 | if view.Terminal != "completed" { | 42 | if view.Terminal != "completed" { | ||
| 43 | t.Fatalf("terminal = %q, want completed", view.Terminal) | 43 | t.Fatalf("terminal = %q, want completed", view.Terminal) | ||
| 44 | } | 44 | } | ||
| 45 | } | 45 | } | ||
| 46 | + | ||||
| 47 | + | func TestProjectTraceClosesFailedToolByCallID(t *testing.T) { | |||
| 48 | + | started := time.Date(2026, 8, 12, 7, 0, 0, 0, time.UTC) | |||
| 49 | + | view := ProjectTrace([]TraceEvent{ | |||
| 50 | + | {Timestamp: started, Type: "tool.started", AgentID: "agent-1", TurnID: "turn-1", Data: map[string]any{"call_id": "one"}}, | |||
| 51 | + | {Timestamp: started, Type: "tool.started", AgentID: "agent-1", TurnID: "turn-1", Data: map[string]any{"call_id": "two"}}, | |||
| 52 | + | {Timestamp: started.Add(time.Second), Type: "tool.failed", AgentID: "agent-1", TurnID: "turn-1", Data: map[string]any{"call_id": "one"}}, | |||
| 53 | + | })["agent-1"] | |||
| 54 | + | if len(view.OpenOperations) != 1 || view.OpenOperations[0].CallID != "two" { | |||
| 55 | + | t.Fatalf("open operations = %#v, want only call two", view.OpenOperations) | |||
| 56 | + | } | |||
| 57 | + | } | |||
| 58 | + | ||||
| 59 | + | func TestProjectTraceFailedAgentClosesOpenOperations(t *testing.T) { | |||
| 60 | + | view := ProjectTrace([]TraceEvent{ | |||
| 61 | + | {Type: "tool.started", AgentID: "agent-1", TurnID: "turn-1", Data: map[string]any{"call_id": "one"}}, | |||
| 62 | + | {Type: "agent.failed", AgentID: "agent-1"}, | |||
| 63 | + | })["agent-1"] | |||
| 64 | + | if len(view.OpenOperations) != 0 || view.Terminal != "failed" { | |||
| 65 | + | t.Fatalf("view = %#v, want failed terminal with no open operations", view) | |||
| 66 | + | } | |||
| 67 | + | } | |||
M packages/agent/subagents/runner_e2e_test.go +1 -1
| @@ -214,7 +214,7 @@ func TestRunnerEndToEndWithStubChild(t *testing.T) { | |||
| 214 | 214 | // stdout is no longer duplicated into a per-agent durable event log. | |
| 215 | 215 | waitFor := func(want string) { | |
| 216 | 216 | t.Helper() | |
| 217 | - | deadline := time.Now().Add(5 * time.Second) | |
| 217 | + | deadline := time.Now().Add(runnerProcessTimeout) | |
| 218 | 218 | for time.Now().Before(deadline) { | |
| 219 | 219 | if strings.Contains(strings.Join(a.Transcript(), "\n"), want) { | |
| 220 | 220 | return | |
| @@ -214,7 +214,7 @@ func TestRunnerEndToEndWithStubChild(t *testing.T) { | |||||
| 214 | // stdout is no longer duplicated into a per-agent durable event log. | 214 | // stdout is no longer duplicated into a per-agent durable event log. | ||
| 215 | waitFor := func(want string) { | 215 | waitFor := func(want string) { | ||
| 216 | t.Helper() | 216 | t.Helper() | ||
| 217 | - | deadline := time.Now().Add(5 * time.Second) | 217 | + | deadline := time.Now().Add(runnerProcessTimeout) |
| 218 | for time.Now().Before(deadline) { | 218 | for time.Now().Before(deadline) { | ||
| 219 | if strings.Contains(strings.Join(a.Transcript(), "\n"), want) { | 219 | if strings.Contains(strings.Join(a.Transcript(), "\n"), want) { | ||
| 220 | return | 220 | return | ||
M packages/agent/subagents/supervisor.go +5 -2
| @@ -648,11 +648,14 @@ func (f *Supervisor) TraceViews() map[string]AgentTraceView { | |||
| 648 | 648 | ||
| 649 | 649 | // Close flushes and closes the optional trace bundle. It does not stop agents; | |
| 650 | 650 | // their lifetime remains controlled by the supervisor context and Stop. | |
| 651 | - | func (f *Supervisor) Close() error { | |
| 651 | + | func (f *Supervisor) Close() error { return f.CloseContext(context.Background()) } | |
| 652 | + | ||
| 653 | + | // CloseContext flushes and closes the optional trace bundle within ctx. | |
| 654 | + | func (f *Supervisor) CloseContext(ctx context.Context) error { | |
| 652 | 655 | if f == nil || f.trace == nil { | |
| 653 | 656 | return nil | |
| 654 | 657 | } | |
| 655 | - | return f.trace.Close() | |
| 658 | + | return f.trace.CloseContext(ctx) | |
| 656 | 659 | } | |
| 657 | 660 | ||
| 658 | 661 | func (f *Supervisor) CancelTurn(id string) error { | |
| @@ -648,11 +648,14 @@ func (f *Supervisor) TraceViews() map[string]AgentTraceView { | |||||
| 648 | 648 | ||||
| 649 | // Close flushes and closes the optional trace bundle. It does not stop agents; | 649 | // Close flushes and closes the optional trace bundle. It does not stop agents; | ||
| 650 | // their lifetime remains controlled by the supervisor context and Stop. | 650 | // their lifetime remains controlled by the supervisor context and Stop. | ||
| 651 | - | func (f *Supervisor) Close() error { | 651 | + | func (f *Supervisor) Close() error { return f.CloseContext(context.Background()) } |
| 652 | + | ||||
| 653 | + | // CloseContext flushes and closes the optional trace bundle within ctx. | |||
| 654 | + | func (f *Supervisor) CloseContext(ctx context.Context) error { | |||
| 652 | if f == nil || f.trace == nil { | 655 | if f == nil || f.trace == nil { | ||
| 653 | return nil | 656 | return nil | ||
| 654 | } | 657 | } | ||
| 655 | - | return f.trace.Close() | 658 | + | return f.trace.CloseContext(ctx) |
| 656 | } | 659 | } | ||
| 657 | 660 | ||||
| 658 | func (f *Supervisor) CancelTurn(id string) error { | 661 | func (f *Supervisor) CancelTurn(id string) error { | ||
M packages/agent/subagents/trace.go +112 -37
| @@ -7,6 +7,7 @@ import ( | |||
| 7 | 7 | "fmt" | |
| 8 | 8 | "os" | |
| 9 | 9 | "path/filepath" | |
| 10 | + | "reflect" | |
| 10 | 11 | "strings" | |
| 11 | 12 | "sync" | |
| 12 | 13 | "time" | |
| @@ -89,6 +90,7 @@ type TraceWriter struct { | |||
| 89 | 90 | payloadDir string | |
| 90 | 91 | memory bool | |
| 91 | 92 | sequence uint64 | |
| 93 | + | dropped uint64 | |
| 92 | 94 | events []TraceEvent | |
| 93 | 95 | ||
| 94 | 96 | errorMu sync.Mutex | |
| @@ -103,6 +105,7 @@ type traceRecord struct { | |||
| 103 | 105 | ||
| 104 | 106 | const ( | |
| 105 | 107 | traceQueueInitialCapacity = 64 | |
| 108 | + | traceQueueLimit = 1024 | |
| 106 | 109 | memoryTraceEventLimit = 4096 | |
| 107 | 110 | ) | |
| 108 | 111 | ||
| @@ -144,6 +147,11 @@ func NewTraceWriter(dir string, modes ...TraceMode) (*TraceWriter, error) { | |||
| 144 | 147 | if err := os.Chmod(absolute, 0o700); err != nil { | |
| 145 | 148 | return nil, fmt.Errorf("trace writer directory permissions: %w", err) | |
| 146 | 149 | } | |
| 150 | + | if _, err := os.Stat(filepath.Join(absolute, "trace.jsonl")); err == nil { | |
| 151 | + | return nil, fmt.Errorf("trace writer: bundle already exists") | |
| 152 | + | } else if !errors.Is(err, os.ErrNotExist) { | |
| 153 | + | return nil, fmt.Errorf("trace writer trace file: %w", err) | |
| 154 | + | } | |
| 147 | 155 | payloadDir := filepath.Join(absolute, "payloads") | |
| 148 | 156 | if err := os.MkdirAll(payloadDir, 0o700); err != nil { | |
| 149 | 157 | return nil, fmt.Errorf("trace writer payload directory: %w", err) | |
| @@ -244,6 +252,18 @@ func (w *TraceWriter) Record(event TraceEvent) { | |||
| 244 | 252 | } | |
| 245 | 253 | w.sequence++ | |
| 246 | 254 | record.event.Seq = w.sequence | |
| 255 | + | if len(w.pending) >= traceQueueLimit { | |
| 256 | + | for index, pending := range w.pending { | |
| 257 | + | if pending.barrier == nil { | |
| 258 | + | copy(w.pending[index:], w.pending[index+1:]) | |
| 259 | + | w.pending[len(w.pending)-1] = traceRecord{} | |
| 260 | + | w.pending = w.pending[:len(w.pending)-1] | |
| 261 | + | w.dropped++ | |
| 262 | + | w.setError(errors.New("trace event queue overflow")) | |
| 263 | + | break | |
| 264 | + | } | |
| 265 | + | } | |
| 266 | + | } | |
| 247 | 267 | w.pending = append(w.pending, record) | |
| 248 | 268 | w.cond.Signal() | |
| 249 | 269 | } | |
| @@ -303,11 +323,7 @@ func (w *TraceWriter) write(record traceRecord) { | |||
| 303 | 323 | event := record.event | |
| 304 | 324 | if w.memory { | |
| 305 | 325 | w.mu.Lock() | |
| 306 | - | w.events = append(w.events, event) | |
| 307 | - | if overflow := len(w.events) - memoryTraceEventLimit; overflow > 0 { | |
| 308 | - | copy(w.events, w.events[overflow:]) | |
| 309 | - | w.events = w.events[:memoryTraceEventLimit] | |
| 310 | - | } | |
| 326 | + | w.appendEvent(event) | |
| 311 | 327 | w.mu.Unlock() | |
| 312 | 328 | return | |
| 313 | 329 | } | |
| @@ -347,17 +363,24 @@ func (w *TraceWriter) write(record traceRecord) { | |||
| 347 | 363 | w.setError(fmt.Errorf("trace event sync: %w", err)) | |
| 348 | 364 | } | |
| 349 | 365 | w.mu.Lock() | |
| 350 | - | w.events = append(w.events, event) | |
| 366 | + | w.appendEvent(event) | |
| 351 | 367 | w.mu.Unlock() | |
| 352 | 368 | } | |
| 353 | 369 | ||
| 354 | - | // Events returns a stable copy of observations durably written by this writer. | |
| 355 | - | // It supports the in-process projection used by status and the dashboard. | |
| 370 | + | func (w *TraceWriter) appendEvent(event TraceEvent) { | |
| 371 | + | w.events = append(w.events, event) | |
| 372 | + | if overflow := len(w.events) - memoryTraceEventLimit; overflow > 0 { | |
| 373 | + | copy(w.events, w.events[overflow:]) | |
| 374 | + | w.events = w.events[:memoryTraceEventLimit] | |
| 375 | + | } | |
| 376 | + | } | |
| 377 | + | ||
| 378 | + | // Events returns a stable copy of events already written by this writer. It | |
| 379 | + | // never waits for queued filesystem work, keeping status and UI reads cheap. | |
| 356 | 380 | func (w *TraceWriter) Events() []TraceEvent { | |
| 357 | 381 | if w == nil { | |
| 358 | 382 | return nil | |
| 359 | 383 | } | |
| 360 | - | _ = w.Flush() | |
| 361 | 384 | w.mu.Lock() | |
| 362 | 385 | defer w.mu.Unlock() | |
| 363 | 386 | return append([]TraceEvent(nil), w.events...) | |
| @@ -375,48 +398,66 @@ func (w *TraceWriter) setError(err error) { | |||
| 375 | 398 | } | |
| 376 | 399 | ||
| 377 | 400 | // Flush waits until all events recorded before the call have been written. | |
| 378 | - | // It is useful to readers that need a live, consistent view; agent recording | |
| 379 | - | // itself remains asynchronous. | |
| 380 | - | func (w *TraceWriter) Flush() error { | |
| 401 | + | func (w *TraceWriter) Flush() error { return w.FlushContext(context.Background()) } | |
| 402 | + | ||
| 403 | + | // FlushContext honors cancellation while waiting for the trace writer. | |
| 404 | + | func (w *TraceWriter) FlushContext(ctx context.Context) error { | |
| 381 | 405 | if w == nil { | |
| 382 | 406 | return nil | |
| 383 | 407 | } | |
| 408 | + | if ctx == nil { | |
| 409 | + | ctx = context.Background() | |
| 410 | + | } | |
| 384 | 411 | w.mu.Lock() | |
| 412 | + | wait := w.finished | |
| 385 | 413 | if !w.closed { | |
| 386 | 414 | barrier := make(chan struct{}) | |
| 387 | 415 | w.pending = append(w.pending, traceRecord{barrier: barrier}) | |
| 388 | 416 | w.cond.Signal() | |
| 389 | - | w.mu.Unlock() | |
| 390 | - | <-barrier | |
| 391 | - | } else { | |
| 392 | - | w.mu.Unlock() | |
| 393 | - | <-w.finished | |
| 417 | + | wait = barrier | |
| 418 | + | } | |
| 419 | + | w.mu.Unlock() | |
| 420 | + | select { | |
| 421 | + | case <-wait: | |
| 422 | + | return w.recordingError() | |
| 423 | + | case <-ctx.Done(): | |
| 424 | + | return fmt.Errorf("flush trace: %w", ctx.Err()) | |
| 394 | 425 | } | |
| 395 | - | return w.recordingError() | |
| 396 | 426 | } | |
| 397 | 427 | ||
| 398 | - | // Close drains the queue, syncs each event, and closes the bundle. It is safe | |
| 399 | - | // to call more than once. Close is the only operation that reports persistence | |
| 400 | - | // errors; Record remains best effort. | |
| 401 | - | func (w *TraceWriter) Close() error { | |
| 428 | + | // Close drains the queue, syncs each event, and closes the bundle. | |
| 429 | + | func (w *TraceWriter) Close() error { return w.CloseContext(context.Background()) } | |
| 430 | + | ||
| 431 | + | // CloseContext starts closing once and honors cancellation while waiting. A | |
| 432 | + | // later caller can retry the wait without repeating file-close work. | |
| 433 | + | func (w *TraceWriter) CloseContext(ctx context.Context) error { | |
| 402 | 434 | if w == nil { | |
| 403 | 435 | return nil | |
| 404 | 436 | } | |
| 437 | + | if ctx == nil { | |
| 438 | + | ctx = context.Background() | |
| 439 | + | } | |
| 405 | 440 | w.closeOnce.Do(func() { | |
| 406 | - | w.mu.Lock() | |
| 407 | - | w.closed = true | |
| 408 | - | w.cond.Broadcast() | |
| 409 | - | w.mu.Unlock() | |
| 410 | - | <-w.finished | |
| 411 | - | if !w.memory { | |
| 412 | - | if err := w.traceFile.Close(); err != nil { | |
| 413 | - | w.setError(fmt.Errorf("trace file close: %w", err)) | |
| 441 | + | go func() { | |
| 442 | + | w.mu.Lock() | |
| 443 | + | w.closed = true | |
| 444 | + | w.cond.Broadcast() | |
| 445 | + | w.mu.Unlock() | |
| 446 | + | <-w.finished | |
| 447 | + | if !w.memory { | |
| 448 | + | if err := w.traceFile.Close(); err != nil { | |
| 449 | + | w.setError(fmt.Errorf("trace file close: %w", err)) | |
| 450 | + | } | |
| 414 | 451 | } | |
| 415 | - | } | |
| 416 | - | close(w.closeDone) | |
| 452 | + | close(w.closeDone) | |
| 453 | + | }() | |
| 417 | 454 | }) | |
| 418 | - | <-w.closeDone | |
| 419 | - | return w.recordingError() | |
| 455 | + | select { | |
| 456 | + | case <-w.closeDone: | |
| 457 | + | return w.recordingError() | |
| 458 | + | case <-ctx.Done(): | |
| 459 | + | return fmt.Errorf("close trace: %w", ctx.Err()) | |
| 460 | + | } | |
| 420 | 461 | } | |
| 421 | 462 | ||
| 422 | 463 | func (w *TraceWriter) recordingError() error { | |
| @@ -502,7 +543,7 @@ func redactTraceMap(data map[string]any) map[string]any { | |||
| 502 | 543 | } | |
| 503 | 544 | out := make(map[string]any, len(data)) | |
| 504 | 545 | for key, value := range data { | |
| 505 | - | if _, sensitive := sensitiveTraceKeys[normalizeTraceKey(key)]; sensitive { | |
| 546 | + | if sensitiveTraceKey(normalizeTraceKey(key)) { | |
| 506 | 547 | out[key] = "[REDACTED]" | |
| 507 | 548 | continue | |
| 508 | 549 | } | |
| @@ -511,7 +552,19 @@ func redactTraceMap(data map[string]any) map[string]any { | |||
| 511 | 552 | return out | |
| 512 | 553 | } | |
| 513 | 554 | ||
| 555 | + | func sensitiveTraceKey(key string) bool { | |
| 556 | + | for fragment := range sensitiveTraceKeys { | |
| 557 | + | if strings.Contains(key, fragment) { | |
| 558 | + | return true | |
| 559 | + | } | |
| 560 | + | } | |
| 561 | + | return false | |
| 562 | + | } | |
| 563 | + | ||
| 514 | 564 | func redactTraceValue(value any) any { | |
| 565 | + | if value == nil { | |
| 566 | + | return nil | |
| 567 | + | } | |
| 515 | 568 | switch value := value.(type) { | |
| 516 | 569 | case map[string]any: | |
| 517 | 570 | return redactTraceMap(value) | |
| @@ -521,9 +574,31 @@ func redactTraceValue(value any) any { | |||
| 521 | 574 | out[i] = redactTraceValue(value[i]) | |
| 522 | 575 | } | |
| 523 | 576 | return out | |
| 524 | - | default: | |
| 525 | - | return value | |
| 526 | 577 | } | |
| 578 | + | v := reflect.ValueOf(value) | |
| 579 | + | if v.Kind() == reflect.Map && v.Type().Key().Kind() == reflect.String { | |
| 580 | + | out := make(map[string]any, v.Len()) | |
| 581 | + | iter := v.MapRange() | |
| 582 | + | for iter.Next() { | |
| 583 | + | key := iter.Key().String() | |
| 584 | + | if sensitiveTraceKey(normalizeTraceKey(key)) { | |
| 585 | + | out[key] = "[REDACTED]" | |
| 586 | + | } else { | |
| 587 | + | out[key] = redactTraceValue(iter.Value().Interface()) | |
| 588 | + | } | |
| 589 | + | } | |
| 590 | + | return out | |
| 591 | + | } | |
| 592 | + | if v.Kind() == reflect.Struct { | |
| 593 | + | encoded, err := json.Marshal(value) | |
| 594 | + | if err == nil { | |
| 595 | + | var data map[string]any | |
| 596 | + | if json.Unmarshal(encoded, &data) == nil { | |
| 597 | + | return redactTraceMap(data) | |
| 598 | + | } | |
| 599 | + | } | |
| 600 | + | } | |
| 601 | + | return value | |
| 527 | 602 | } | |
| 528 | 603 | ||
| 529 | 604 | func cloneTraceMap(data map[string]any) map[string]any { | |
| @@ -7,6 +7,7 @@ import ( | |||||
| 7 | "fmt" | 7 | "fmt" | ||
| 8 | "os" | 8 | "os" | ||
| 9 | "path/filepath" | 9 | "path/filepath" | ||
| 10 | + | "reflect" | |||
| 10 | "strings" | 11 | "strings" | ||
| 11 | "sync" | 12 | "sync" | ||
| 12 | "time" | 13 | "time" | ||
| @@ -89,6 +90,7 @@ type TraceWriter struct { | |||||
| 89 | payloadDir string | 90 | payloadDir string | ||
| 90 | memory bool | 91 | memory bool | ||
| 91 | sequence uint64 | 92 | sequence uint64 | ||
| 93 | + | dropped uint64 | |||
| 92 | events []TraceEvent | 94 | events []TraceEvent | ||
| 93 | 95 | ||||
| 94 | errorMu sync.Mutex | 96 | errorMu sync.Mutex | ||
| @@ -103,6 +105,7 @@ type traceRecord struct { | |||||
| 103 | 105 | ||||
| 104 | const ( | 106 | const ( | ||
| 105 | traceQueueInitialCapacity = 64 | 107 | traceQueueInitialCapacity = 64 | ||
| 108 | + | traceQueueLimit = 1024 | |||
| 106 | memoryTraceEventLimit = 4096 | 109 | memoryTraceEventLimit = 4096 | ||
| 107 | ) | 110 | ) | ||
| 108 | 111 | ||||
| @@ -144,6 +147,11 @@ func NewTraceWriter(dir string, modes ...TraceMode) (*TraceWriter, error) { | |||||
| 144 | if err := os.Chmod(absolute, 0o700); err != nil { | 147 | if err := os.Chmod(absolute, 0o700); err != nil { | ||
| 145 | return nil, fmt.Errorf("trace writer directory permissions: %w", err) | 148 | return nil, fmt.Errorf("trace writer directory permissions: %w", err) | ||
| 146 | } | 149 | } | ||
| 150 | + | if _, err := os.Stat(filepath.Join(absolute, "trace.jsonl")); err == nil { | |||
| 151 | + | return nil, fmt.Errorf("trace writer: bundle already exists") | |||
| 152 | + | } else if !errors.Is(err, os.ErrNotExist) { | |||
| 153 | + | return nil, fmt.Errorf("trace writer trace file: %w", err) | |||
| 154 | + | } | |||
| 147 | payloadDir := filepath.Join(absolute, "payloads") | 155 | payloadDir := filepath.Join(absolute, "payloads") | ||
| 148 | if err := os.MkdirAll(payloadDir, 0o700); err != nil { | 156 | if err := os.MkdirAll(payloadDir, 0o700); err != nil { | ||
| 149 | return nil, fmt.Errorf("trace writer payload directory: %w", err) | 157 | return nil, fmt.Errorf("trace writer payload directory: %w", err) | ||
| @@ -244,6 +252,18 @@ func (w *TraceWriter) Record(event TraceEvent) { | |||||
| 244 | } | 252 | } | ||
| 245 | w.sequence++ | 253 | w.sequence++ | ||
| 246 | record.event.Seq = w.sequence | 254 | record.event.Seq = w.sequence | ||
| 255 | + | if len(w.pending) >= traceQueueLimit { | |||
| 256 | + | for index, pending := range w.pending { | |||
| 257 | + | if pending.barrier == nil { | |||
| 258 | + | copy(w.pending[index:], w.pending[index+1:]) | |||
| 259 | + | w.pending[len(w.pending)-1] = traceRecord{} | |||
| 260 | + | w.pending = w.pending[:len(w.pending)-1] | |||
| 261 | + | w.dropped++ | |||
| 262 | + | w.setError(errors.New("trace event queue overflow")) | |||
| 263 | + | break | |||
| 264 | + | } | |||
| 265 | + | } | |||
| 266 | + | } | |||
| 247 | w.pending = append(w.pending, record) | 267 | w.pending = append(w.pending, record) | ||
| 248 | w.cond.Signal() | 268 | w.cond.Signal() | ||
| 249 | } | 269 | } | ||
| @@ -303,11 +323,7 @@ func (w *TraceWriter) write(record traceRecord) { | |||||
| 303 | event := record.event | 323 | event := record.event | ||
| 304 | if w.memory { | 324 | if w.memory { | ||
| 305 | w.mu.Lock() | 325 | w.mu.Lock() | ||
| 306 | - | w.events = append(w.events, event) | 326 | + | w.appendEvent(event) |
| 307 | - | if overflow := len(w.events) - memoryTraceEventLimit; overflow > 0 { | |||
| 308 | - | copy(w.events, w.events[overflow:]) | |||
| 309 | - | w.events = w.events[:memoryTraceEventLimit] | |||
| 310 | - | } | |||
| 311 | w.mu.Unlock() | 327 | w.mu.Unlock() | ||
| 312 | return | 328 | return | ||
| 313 | } | 329 | } | ||
| @@ -347,17 +363,24 @@ func (w *TraceWriter) write(record traceRecord) { | |||||
| 347 | w.setError(fmt.Errorf("trace event sync: %w", err)) | 363 | w.setError(fmt.Errorf("trace event sync: %w", err)) | ||
| 348 | } | 364 | } | ||
| 349 | w.mu.Lock() | 365 | w.mu.Lock() | ||
| 350 | - | w.events = append(w.events, event) | 366 | + | w.appendEvent(event) |
| 351 | w.mu.Unlock() | 367 | w.mu.Unlock() | ||
| 352 | } | 368 | } | ||
| 353 | 369 | ||||
| 354 | - | // Events returns a stable copy of observations durably written by this writer. | 370 | + | func (w *TraceWriter) appendEvent(event TraceEvent) { |
| 355 | - | // It supports the in-process projection used by status and the dashboard. | 371 | + | w.events = append(w.events, event) |
| 372 | + | if overflow := len(w.events) - memoryTraceEventLimit; overflow > 0 { | |||
| 373 | + | copy(w.events, w.events[overflow:]) | |||
| 374 | + | w.events = w.events[:memoryTraceEventLimit] | |||
| 375 | + | } | |||
| 376 | + | } | |||
| 377 | + | ||||
| 378 | + | // Events returns a stable copy of events already written by this writer. It | |||
| 379 | + | // never waits for queued filesystem work, keeping status and UI reads cheap. | |||
| 356 | func (w *TraceWriter) Events() []TraceEvent { | 380 | func (w *TraceWriter) Events() []TraceEvent { | ||
| 357 | if w == nil { | 381 | if w == nil { | ||
| 358 | return nil | 382 | return nil | ||
| 359 | } | 383 | } | ||
| 360 | - | _ = w.Flush() | |||
| 361 | w.mu.Lock() | 384 | w.mu.Lock() | ||
| 362 | defer w.mu.Unlock() | 385 | defer w.mu.Unlock() | ||
| 363 | return append([]TraceEvent(nil), w.events...) | 386 | return append([]TraceEvent(nil), w.events...) | ||
| @@ -375,48 +398,66 @@ func (w *TraceWriter) setError(err error) { | |||||
| 375 | } | 398 | } | ||
| 376 | 399 | ||||
| 377 | // Flush waits until all events recorded before the call have been written. | 400 | // Flush waits until all events recorded before the call have been written. | ||
| 378 | - | // It is useful to readers that need a live, consistent view; agent recording | 401 | + | func (w *TraceWriter) Flush() error { return w.FlushContext(context.Background()) } |
| 379 | - | // itself remains asynchronous. | 402 | + | |
| 380 | - | func (w *TraceWriter) Flush() error { | 403 | + | // FlushContext honors cancellation while waiting for the trace writer. |
| 404 | + | func (w *TraceWriter) FlushContext(ctx context.Context) error { | |||
| 381 | if w == nil { | 405 | if w == nil { | ||
| 382 | return nil | 406 | return nil | ||
| 383 | } | 407 | } | ||
| 408 | + | if ctx == nil { | |||
| 409 | + | ctx = context.Background() | |||
| 410 | + | } | |||
| 384 | w.mu.Lock() | 411 | w.mu.Lock() | ||
| 412 | + | wait := w.finished | |||
| 385 | if !w.closed { | 413 | if !w.closed { | ||
| 386 | barrier := make(chan struct{}) | 414 | barrier := make(chan struct{}) | ||
| 387 | w.pending = append(w.pending, traceRecord{barrier: barrier}) | 415 | w.pending = append(w.pending, traceRecord{barrier: barrier}) | ||
| 388 | w.cond.Signal() | 416 | w.cond.Signal() | ||
| 389 | - | w.mu.Unlock() | 417 | + | wait = barrier |
| 390 | - | <-barrier | 418 | + | } |
| 391 | - | } else { | 419 | + | w.mu.Unlock() |
| 392 | - | w.mu.Unlock() | 420 | + | select { |
| 393 | - | <-w.finished | 421 | + | case <-wait: |
| 422 | + | return w.recordingError() | |||
| 423 | + | case <-ctx.Done(): | |||
| 424 | + | return fmt.Errorf("flush trace: %w", ctx.Err()) | |||
| 394 | } | 425 | } | ||
| 395 | - | return w.recordingError() | |||
| 396 | } | 426 | } | ||
| 397 | 427 | ||||
| 398 | - | // Close drains the queue, syncs each event, and closes the bundle. It is safe | 428 | + | // Close drains the queue, syncs each event, and closes the bundle. |
| 399 | - | // to call more than once. Close is the only operation that reports persistence | 429 | + | func (w *TraceWriter) Close() error { return w.CloseContext(context.Background()) } |
| 400 | - | // errors; Record remains best effort. | 430 | + | |
| 401 | - | func (w *TraceWriter) Close() error { | 431 | + | // CloseContext starts closing once and honors cancellation while waiting. A |
| 432 | + | // later caller can retry the wait without repeating file-close work. | |||
| 433 | + | func (w *TraceWriter) CloseContext(ctx context.Context) error { | |||
| 402 | if w == nil { | 434 | if w == nil { | ||
| 403 | return nil | 435 | return nil | ||
| 404 | } | 436 | } | ||
| 437 | + | if ctx == nil { | |||
| 438 | + | ctx = context.Background() | |||
| 439 | + | } | |||
| 405 | w.closeOnce.Do(func() { | 440 | w.closeOnce.Do(func() { | ||
| 406 | - | w.mu.Lock() | 441 | + | go func() { |
| 407 | - | w.closed = true | 442 | + | w.mu.Lock() |
| 408 | - | w.cond.Broadcast() | 443 | + | w.closed = true |
| 409 | - | w.mu.Unlock() | 444 | + | w.cond.Broadcast() |
| 410 | - | <-w.finished | 445 | + | w.mu.Unlock() |
| 411 | - | if !w.memory { | 446 | + | <-w.finished |
| 412 | - | if err := w.traceFile.Close(); err != nil { | 447 | + | if !w.memory { |
| 413 | - | w.setError(fmt.Errorf("trace file close: %w", err)) | 448 | + | if err := w.traceFile.Close(); err != nil { |
| 449 | + | w.setError(fmt.Errorf("trace file close: %w", err)) | |||
| 450 | + | } | |||
| 414 | } | 451 | } | ||
| 415 | - | } | 452 | + | close(w.closeDone) |
| 416 | - | close(w.closeDone) | 453 | + | }() |
| 417 | }) | 454 | }) | ||
| 418 | - | <-w.closeDone | 455 | + | select { |
| 419 | - | return w.recordingError() | 456 | + | case <-w.closeDone: |
| 457 | + | return w.recordingError() | |||
| 458 | + | case <-ctx.Done(): | |||
| 459 | + | return fmt.Errorf("close trace: %w", ctx.Err()) | |||
| 460 | + | } | |||
| 420 | } | 461 | } | ||
| 421 | 462 | ||||
| 422 | func (w *TraceWriter) recordingError() error { | 463 | func (w *TraceWriter) recordingError() error { | ||
| @@ -502,7 +543,7 @@ func redactTraceMap(data map[string]any) map[string]any { | |||||
| 502 | } | 543 | } | ||
| 503 | out := make(map[string]any, len(data)) | 544 | out := make(map[string]any, len(data)) | ||
| 504 | for key, value := range data { | 545 | for key, value := range data { | ||
| 505 | - | if _, sensitive := sensitiveTraceKeys[normalizeTraceKey(key)]; sensitive { | 546 | + | if sensitiveTraceKey(normalizeTraceKey(key)) { |
| 506 | out[key] = "[REDACTED]" | 547 | out[key] = "[REDACTED]" | ||
| 507 | continue | 548 | continue | ||
| 508 | } | 549 | } | ||
| @@ -511,7 +552,19 @@ func redactTraceMap(data map[string]any) map[string]any { | |||||
| 511 | return out | 552 | return out | ||
| 512 | } | 553 | } | ||
| 513 | 554 | ||||
| 555 | + | func sensitiveTraceKey(key string) bool { | |||
| 556 | + | for fragment := range sensitiveTraceKeys { | |||
| 557 | + | if strings.Contains(key, fragment) { | |||
| 558 | + | return true | |||
| 559 | + | } | |||
| 560 | + | } | |||
| 561 | + | return false | |||
| 562 | + | } | |||
| 563 | + | ||||
| 514 | func redactTraceValue(value any) any { | 564 | func redactTraceValue(value any) any { | ||
| 565 | + | if value == nil { | |||
| 566 | + | return nil | |||
| 567 | + | } | |||
| 515 | switch value := value.(type) { | 568 | switch value := value.(type) { | ||
| 516 | case map[string]any: | 569 | case map[string]any: | ||
| 517 | return redactTraceMap(value) | 570 | return redactTraceMap(value) | ||
| @@ -521,9 +574,31 @@ func redactTraceValue(value any) any { | |||||
| 521 | out[i] = redactTraceValue(value[i]) | 574 | out[i] = redactTraceValue(value[i]) | ||
| 522 | } | 575 | } | ||
| 523 | return out | 576 | return out | ||
| 524 | - | default: | |||
| 525 | - | return value | |||
| 526 | } | 577 | } | ||
| 578 | + | v := reflect.ValueOf(value) | |||
| 579 | + | if v.Kind() == reflect.Map && v.Type().Key().Kind() == reflect.String { | |||
| 580 | + | out := make(map[string]any, v.Len()) | |||
| 581 | + | iter := v.MapRange() | |||
| 582 | + | for iter.Next() { | |||
| 583 | + | key := iter.Key().String() | |||
| 584 | + | if sensitiveTraceKey(normalizeTraceKey(key)) { | |||
| 585 | + | out[key] = "[REDACTED]" | |||
| 586 | + | } else { | |||
| 587 | + | out[key] = redactTraceValue(iter.Value().Interface()) | |||
| 588 | + | } | |||
| 589 | + | } | |||
| 590 | + | return out | |||
| 591 | + | } | |||
| 592 | + | if v.Kind() == reflect.Struct { | |||
| 593 | + | encoded, err := json.Marshal(value) | |||
| 594 | + | if err == nil { | |||
| 595 | + | var data map[string]any | |||
| 596 | + | if json.Unmarshal(encoded, &data) == nil { | |||
| 597 | + | return redactTraceMap(data) | |||
| 598 | + | } | |||
| 599 | + | } | |||
| 600 | + | } | |||
| 601 | + | return value | |||
| 527 | } | 602 | } | ||
| 528 | 603 | ||||
| 529 | func cloneTraceMap(data map[string]any) map[string]any { | 604 | func cloneTraceMap(data map[string]any) map[string]any { | ||
M packages/agent/subagents/trace_read.go +59 -4
| @@ -2,32 +2,87 @@ package subagents | |||
| 2 | 2 | ||
| 3 | 3 | import ( | |
| 4 | 4 | "bufio" | |
| 5 | + | "context" | |
| 5 | 6 | "encoding/json" | |
| 7 | + | "errors" | |
| 6 | 8 | "fmt" | |
| 7 | 9 | "os" | |
| 8 | 10 | "path/filepath" | |
| 11 | + | "strings" | |
| 9 | 12 | ) | |
| 10 | 13 | ||
| 11 | - | // ReadTrace reads the stable JSONL event stream from a trace bundle. | |
| 14 | + | // ReadTrace reads a trace bundle using a background context. | |
| 12 | 15 | func ReadTrace(bundle string) ([]TraceEvent, error) { | |
| 13 | - | file, err := os.Open(filepath.Join(bundle, "trace.jsonl")) | |
| 16 | + | return ReadTraceContext(context.Background(), bundle) | |
| 17 | + | } | |
| 18 | + | ||
| 19 | + | // ReadTraceContext reads the manifest-selected JSONL trace stream. If a stream | |
| 20 | + | // ends with malformed data, events decoded before it are returned with the | |
| 21 | + | // error so callers can still diagnose the valid prefix. | |
| 22 | + | func ReadTraceContext(ctx context.Context, bundle string) ([]TraceEvent, error) { | |
| 23 | + | if ctx == nil { | |
| 24 | + | ctx = context.Background() | |
| 25 | + | } | |
| 26 | + | manifestPath := filepath.Join(bundle, "manifest.json") | |
| 27 | + | data, err := os.ReadFile(manifestPath) | |
| 28 | + | if err != nil { | |
| 29 | + | return nil, fmt.Errorf("read trace manifest: %w", err) | |
| 30 | + | } | |
| 31 | + | var manifest TraceManifest | |
| 32 | + | if err := json.Unmarshal(data, &manifest); err != nil { | |
| 33 | + | return nil, fmt.Errorf("decode trace manifest: %w", err) | |
| 34 | + | } | |
| 35 | + | if manifest.Version != 1 { | |
| 36 | + | return nil, fmt.Errorf("unsupported trace manifest version %d", manifest.Version) | |
| 37 | + | } | |
| 38 | + | tracePath, err := traceBundlePath(bundle, manifest.TraceFile) | |
| 39 | + | if err != nil { | |
| 40 | + | return nil, err | |
| 41 | + | } | |
| 42 | + | file, err := os.Open(tracePath) | |
| 14 | 43 | if err != nil { | |
| 15 | 44 | return nil, fmt.Errorf("open trace: %w", err) | |
| 16 | 45 | } | |
| 17 | 46 | defer file.Close() | |
| 47 | + | ||
| 18 | 48 | var events []TraceEvent | |
| 19 | 49 | scanner := bufio.NewScanner(file) | |
| 20 | 50 | buffer := make([]byte, 64*1024) | |
| 21 | 51 | scanner.Buffer(buffer, 4*1024*1024) | |
| 22 | 52 | for scanner.Scan() { | |
| 53 | + | if err := ctx.Err(); err != nil { | |
| 54 | + | return events, fmt.Errorf("read trace: %w", err) | |
| 55 | + | } | |
| 23 | 56 | var event TraceEvent | |
| 24 | 57 | if err := json.Unmarshal(scanner.Bytes(), &event); err != nil { | |
| 25 | - | return nil, fmt.Errorf("decode trace event: %w", err) | |
| 58 | + | return events, fmt.Errorf("decode trace event: %w", err) | |
| 26 | 59 | } | |
| 27 | 60 | events = append(events, event) | |
| 28 | 61 | } | |
| 29 | 62 | if err := scanner.Err(); err != nil { | |
| 30 | - | return nil, fmt.Errorf("read trace: %w", err) | |
| 63 | + | return events, fmt.Errorf("read trace: %w", err) | |
| 64 | + | } | |
| 65 | + | if err := ctx.Err(); err != nil { | |
| 66 | + | return events, fmt.Errorf("read trace: %w", err) | |
| 31 | 67 | } | |
| 32 | 68 | return events, nil | |
| 33 | 69 | } | |
| 70 | + | ||
| 71 | + | func traceBundlePath(bundle, name string) (string, error) { | |
| 72 | + | if strings.TrimSpace(name) == "" || filepath.IsAbs(name) { | |
| 73 | + | return "", errors.New("invalid trace manifest path") | |
| 74 | + | } | |
| 75 | + | root, err := filepath.Abs(bundle) | |
| 76 | + | if err != nil { | |
| 77 | + | return "", fmt.Errorf("resolve trace bundle: %w", err) | |
| 78 | + | } | |
| 79 | + | path, err := filepath.Abs(filepath.Join(root, name)) | |
| 80 | + | if err != nil { | |
| 81 | + | return "", fmt.Errorf("resolve trace path: %w", err) | |
| 82 | + | } | |
| 83 | + | rel, err := filepath.Rel(root, path) | |
| 84 | + | if err != nil || rel == ".." || strings.HasPrefix(rel, ".."+string(filepath.Separator)) { | |
| 85 | + | return "", errors.New("trace manifest path escapes bundle") | |
| 86 | + | } | |
| 87 | + | return path, nil | |
| 88 | + | } | |
| @@ -2,32 +2,87 @@ package subagents | |||||
| 2 | 2 | ||||
| 3 | import ( | 3 | import ( | ||
| 4 | "bufio" | 4 | "bufio" | ||
| 5 | + | "context" | |||
| 5 | "encoding/json" | 6 | "encoding/json" | ||
| 7 | + | "errors" | |||
| 6 | "fmt" | 8 | "fmt" | ||
| 7 | "os" | 9 | "os" | ||
| 8 | "path/filepath" | 10 | "path/filepath" | ||
| 11 | + | "strings" | |||
| 9 | ) | 12 | ) | ||
| 10 | 13 | ||||
| 11 | - | // ReadTrace reads the stable JSONL event stream from a trace bundle. | 14 | + | // ReadTrace reads a trace bundle using a background context. |
| 12 | func ReadTrace(bundle string) ([]TraceEvent, error) { | 15 | func ReadTrace(bundle string) ([]TraceEvent, error) { | ||
| 13 | - | file, err := os.Open(filepath.Join(bundle, "trace.jsonl")) | 16 | + | return ReadTraceContext(context.Background(), bundle) |
| 17 | + | } | |||
| 18 | + | ||||
| 19 | + | // ReadTraceContext reads the manifest-selected JSONL trace stream. If a stream | |||
| 20 | + | // ends with malformed data, events decoded before it are returned with the | |||
| 21 | + | // error so callers can still diagnose the valid prefix. | |||
| 22 | + | func ReadTraceContext(ctx context.Context, bundle string) ([]TraceEvent, error) { | |||
| 23 | + | if ctx == nil { | |||
| 24 | + | ctx = context.Background() | |||
| 25 | + | } | |||
| 26 | + | manifestPath := filepath.Join(bundle, "manifest.json") | |||
| 27 | + | data, err := os.ReadFile(manifestPath) | |||
| 28 | + | if err != nil { | |||
| 29 | + | return nil, fmt.Errorf("read trace manifest: %w", err) | |||
| 30 | + | } | |||
| 31 | + | var manifest TraceManifest | |||
| 32 | + | if err := json.Unmarshal(data, &manifest); err != nil { | |||
| 33 | + | return nil, fmt.Errorf("decode trace manifest: %w", err) | |||
| 34 | + | } | |||
| 35 | + | if manifest.Version != 1 { | |||
| 36 | + | return nil, fmt.Errorf("unsupported trace manifest version %d", manifest.Version) | |||
| 37 | + | } | |||
| 38 | + | tracePath, err := traceBundlePath(bundle, manifest.TraceFile) | |||
| 39 | + | if err != nil { | |||
| 40 | + | return nil, err | |||
| 41 | + | } | |||
| 42 | + | file, err := os.Open(tracePath) | |||
| 14 | if err != nil { | 43 | if err != nil { | ||
| 15 | return nil, fmt.Errorf("open trace: %w", err) | 44 | return nil, fmt.Errorf("open trace: %w", err) | ||
| 16 | } | 45 | } | ||
| 17 | defer file.Close() | 46 | defer file.Close() | ||
| 47 | + | ||||
| 18 | var events []TraceEvent | 48 | var events []TraceEvent | ||
| 19 | scanner := bufio.NewScanner(file) | 49 | scanner := bufio.NewScanner(file) | ||
| 20 | buffer := make([]byte, 64*1024) | 50 | buffer := make([]byte, 64*1024) | ||
| 21 | scanner.Buffer(buffer, 4*1024*1024) | 51 | scanner.Buffer(buffer, 4*1024*1024) | ||
| 22 | for scanner.Scan() { | 52 | for scanner.Scan() { | ||
| 53 | + | if err := ctx.Err(); err != nil { | |||
| 54 | + | return events, fmt.Errorf("read trace: %w", err) | |||
| 55 | + | } | |||
| 23 | var event TraceEvent | 56 | var event TraceEvent | ||
| 24 | if err := json.Unmarshal(scanner.Bytes(), &event); err != nil { | 57 | if err := json.Unmarshal(scanner.Bytes(), &event); err != nil { | ||
| 25 | - | return nil, fmt.Errorf("decode trace event: %w", err) | 58 | + | return events, fmt.Errorf("decode trace event: %w", err) |
| 26 | } | 59 | } | ||
| 27 | events = append(events, event) | 60 | events = append(events, event) | ||
| 28 | } | 61 | } | ||
| 29 | if err := scanner.Err(); err != nil { | 62 | if err := scanner.Err(); err != nil { | ||
| 30 | - | return nil, fmt.Errorf("read trace: %w", err) | 63 | + | return events, fmt.Errorf("read trace: %w", err) |
| 64 | + | } | |||
| 65 | + | if err := ctx.Err(); err != nil { | |||
| 66 | + | return events, fmt.Errorf("read trace: %w", err) | |||
| 31 | } | 67 | } | ||
| 32 | return events, nil | 68 | return events, nil | ||
| 33 | } | 69 | } | ||
| 70 | + | ||||
| 71 | + | func traceBundlePath(bundle, name string) (string, error) { | |||
| 72 | + | if strings.TrimSpace(name) == "" || filepath.IsAbs(name) { | |||
| 73 | + | return "", errors.New("invalid trace manifest path") | |||
| 74 | + | } | |||
| 75 | + | root, err := filepath.Abs(bundle) | |||
| 76 | + | if err != nil { | |||
| 77 | + | return "", fmt.Errorf("resolve trace bundle: %w", err) | |||
| 78 | + | } | |||
| 79 | + | path, err := filepath.Abs(filepath.Join(root, name)) | |||
| 80 | + | if err != nil { | |||
| 81 | + | return "", fmt.Errorf("resolve trace path: %w", err) | |||
| 82 | + | } | |||
| 83 | + | rel, err := filepath.Rel(root, path) | |||
| 84 | + | if err != nil || rel == ".." || strings.HasPrefix(rel, ".."+string(filepath.Separator)) { | |||
| 85 | + | return "", errors.New("trace manifest path escapes bundle") | |||
| 86 | + | } | |||
| 87 | + | return path, nil | |||
| 88 | + | } | |||
M packages/agent/subagents/trace_test.go +13 -6
| @@ -54,6 +54,9 @@ func TestTraceWriterCreatesPrivateBundleAndDetailedPayload(t *testing.T) { | |||
| 54 | 54 | if len(lines) != 1 { | |
| 55 | 55 | t.Fatalf("got %d trace lines, want 1", len(lines)) | |
| 56 | 56 | } | |
| 57 | + | if strings.Contains(lines[0], "sensitive result") { | |
| 58 | + | t.Fatalf("trace event leaked detailed payload: %s", lines[0]) | |
| 59 | + | } | |
| 57 | 60 | var event TraceEvent | |
| 58 | 61 | if err := json.Unmarshal([]byte(lines[0]), &event); err != nil { | |
| 59 | 62 | t.Fatal(err) | |
| @@ -71,8 +74,9 @@ func TestTraceWriterCreatesPrivateBundleAndDetailedPayload(t *testing.T) { | |||
| 71 | 74 | ||
| 72 | 75 | func TestMemoryTraceWriterRetainsOrderedEventsWithoutBundle(t *testing.T) { | |
| 73 | 76 | writer := NewMemoryTraceWriter() | |
| 74 | - | writer.Record(TraceEvent{Type: "agent.started"}) | |
| 75 | - | writer.Record(TraceEvent{Type: "turn.started", TurnID: "turn-1"}) | |
| 77 | + | for index := 0; index < memoryTraceEventLimit+2; index++ { | |
| 78 | + | writer.Record(TraceEvent{Type: "event", Data: map[string]any{"index": index}}) | |
| 79 | + | } | |
| 76 | 80 | if err := writer.Flush(); err != nil { | |
| 77 | 81 | t.Fatal(err) | |
| 78 | 82 | } | |
| @@ -80,8 +84,8 @@ func TestMemoryTraceWriterRetainsOrderedEventsWithoutBundle(t *testing.T) { | |||
| 80 | 84 | t.Fatalf("memory trace dir = %q, want empty", got) | |
| 81 | 85 | } | |
| 82 | 86 | events := writer.Events() | |
| 83 | - | if len(events) != 2 || events[0].Seq != 1 || events[1].Seq != 2 { | |
| 84 | - | t.Fatalf("memory events = %#v, want contiguous sequence", events) | |
| 87 | + | if len(events) != memoryTraceEventLimit || events[len(events)-1].Seq != memoryTraceEventLimit+2 { | |
| 88 | + | t.Fatalf("memory events retained %d with final seq %d", len(events), events[len(events)-1].Seq) | |
| 85 | 89 | } | |
| 86 | 90 | if err := writer.Close(); err != nil { | |
| 87 | 91 | t.Fatal(err) | |
| @@ -95,8 +99,11 @@ func TestTraceWriterNormalRedactsSensitiveValues(t *testing.T) { | |||
| 95 | 99 | } | |
| 96 | 100 | secret := "super-secret-value" | |
| 97 | 101 | writer.Record(TraceEvent{ | |
| 98 | - | Type: "request", | |
| 99 | - | Data: map[string]any{"name": "safe", "authorization": secret, "nested": map[string]any{"token": secret, "count": 2}}, | |
| 102 | + | Type: "request", | |
| 103 | + | Data: map[string]any{ | |
| 104 | + | "name": "safe", "authorization": secret, "openai_api_key": secret, | |
| 105 | + | "nested": map[string]any{"token": secret, "count": 2, "headers": map[string]string{"x_api_key": secret}}, | |
| 106 | + | }, | |
| 100 | 107 | Payload: map[string]any{"content": secret}, | |
| 101 | 108 | }) | |
| 102 | 109 | if err := writer.Close(); err != nil { | |
| @@ -54,6 +54,9 @@ func TestTraceWriterCreatesPrivateBundleAndDetailedPayload(t *testing.T) { | |||||
| 54 | if len(lines) != 1 { | 54 | if len(lines) != 1 { | ||
| 55 | t.Fatalf("got %d trace lines, want 1", len(lines)) | 55 | t.Fatalf("got %d trace lines, want 1", len(lines)) | ||
| 56 | } | 56 | } | ||
| 57 | + | if strings.Contains(lines[0], "sensitive result") { | |||
| 58 | + | t.Fatalf("trace event leaked detailed payload: %s", lines[0]) | |||
| 59 | + | } | |||
| 57 | var event TraceEvent | 60 | var event TraceEvent | ||
| 58 | if err := json.Unmarshal([]byte(lines[0]), &event); err != nil { | 61 | if err := json.Unmarshal([]byte(lines[0]), &event); err != nil { | ||
| 59 | t.Fatal(err) | 62 | t.Fatal(err) | ||
| @@ -71,8 +74,9 @@ func TestTraceWriterCreatesPrivateBundleAndDetailedPayload(t *testing.T) { | |||||
| 71 | 74 | ||||
| 72 | func TestMemoryTraceWriterRetainsOrderedEventsWithoutBundle(t *testing.T) { | 75 | func TestMemoryTraceWriterRetainsOrderedEventsWithoutBundle(t *testing.T) { | ||
| 73 | writer := NewMemoryTraceWriter() | 76 | writer := NewMemoryTraceWriter() | ||
| 74 | - | writer.Record(TraceEvent{Type: "agent.started"}) | 77 | + | for index := 0; index < memoryTraceEventLimit+2; index++ { |
| 75 | - | writer.Record(TraceEvent{Type: "turn.started", TurnID: "turn-1"}) | 78 | + | writer.Record(TraceEvent{Type: "event", Data: map[string]any{"index": index}}) |
| 79 | + | } | |||
| 76 | if err := writer.Flush(); err != nil { | 80 | if err := writer.Flush(); err != nil { | ||
| 77 | t.Fatal(err) | 81 | t.Fatal(err) | ||
| 78 | } | 82 | } | ||
| @@ -80,8 +84,8 @@ func TestMemoryTraceWriterRetainsOrderedEventsWithoutBundle(t *testing.T) { | |||||
| 80 | t.Fatalf("memory trace dir = %q, want empty", got) | 84 | t.Fatalf("memory trace dir = %q, want empty", got) | ||
| 81 | } | 85 | } | ||
| 82 | events := writer.Events() | 86 | events := writer.Events() | ||
| 83 | - | if len(events) != 2 || events[0].Seq != 1 || events[1].Seq != 2 { | 87 | + | if len(events) != memoryTraceEventLimit || events[len(events)-1].Seq != memoryTraceEventLimit+2 { |
| 84 | - | t.Fatalf("memory events = %#v, want contiguous sequence", events) | 88 | + | t.Fatalf("memory events retained %d with final seq %d", len(events), events[len(events)-1].Seq) |
| 85 | } | 89 | } | ||
| 86 | if err := writer.Close(); err != nil { | 90 | if err := writer.Close(); err != nil { | ||
| 87 | t.Fatal(err) | 91 | t.Fatal(err) | ||
| @@ -95,8 +99,11 @@ func TestTraceWriterNormalRedactsSensitiveValues(t *testing.T) { | |||||
| 95 | } | 99 | } | ||
| 96 | secret := "super-secret-value" | 100 | secret := "super-secret-value" | ||
| 97 | writer.Record(TraceEvent{ | 101 | writer.Record(TraceEvent{ | ||
| 98 | - | Type: "request", | 102 | + | Type: "request", |
| 99 | - | Data: map[string]any{"name": "safe", "authorization": secret, "nested": map[string]any{"token": secret, "count": 2}}, | 103 | + | Data: map[string]any{ |
| 104 | + | "name": "safe", "authorization": secret, "openai_api_key": secret, | |||
| 105 | + | "nested": map[string]any{"token": secret, "count": 2, "headers": map[string]string{"x_api_key": secret}}, | |||
| 106 | + | }, | |||
| 100 | Payload: map[string]any{"content": secret}, | 107 | Payload: map[string]any{"content": secret}, | ||
| 101 | }) | 108 | }) | ||
| 102 | if err := writer.Close(); err != nil { | 109 | if err := writer.Close(); err != nil { | ||
M packages/agent/tools/subagent_status_test.go +28 -10
| @@ -119,11 +119,7 @@ func TestSubagentStatusListsLiveWorkerWithoutPrivateOutput(t *testing.T) { | |||
| 119 | 119 | } | |
| 120 | 120 | ||
| 121 | 121 | var got struct { | |
| 122 | - | Agents []struct { | |
| 123 | - | ID string `json:"agent_id"` | |
| 124 | - | StartedAt string `json:"started_at"` | |
| 125 | - | TaskSummary string `json:"task_summary"` | |
| 126 | - | } `json:"agents"` | |
| 122 | + | Agents []json.RawMessage `json:"agents"` | |
| 127 | 123 | } | |
| 128 | 124 | text := textResult(res.Content) | |
| 129 | 125 | if err := json.Unmarshal([]byte(text), &got); err != nil { | |
| @@ -132,14 +128,36 @@ func TestSubagentStatusListsLiveWorkerWithoutPrivateOutput(t *testing.T) { | |||
| 132 | 128 | if len(got.Agents) != 1 { | |
| 133 | 129 | t.Fatalf("listed agents = %d, want 1", len(got.Agents)) | |
| 134 | 130 | } | |
| 135 | - | if got.Agents[0].ID != agent.ID { | |
| 136 | - | t.Fatalf("agent id = %q, want %q", got.Agents[0].ID, agent.ID) | |
| 131 | + | var entry map[string]json.RawMessage | |
| 132 | + | if err := json.Unmarshal(got.Agents[0], &entry); err != nil { | |
| 133 | + | t.Fatal(err) | |
| 134 | + | } | |
| 135 | + | for _, forbiddenField := range []string{"state", "process_state", "turn_state", "updated_at", "finished_at"} { | |
| 136 | + | if _, present := entry[forbiddenField]; present { | |
| 137 | + | t.Fatalf("status response exposes removed lifecycle field %q", forbiddenField) | |
| 138 | + | } | |
| 139 | + | } | |
| 140 | + | var agentID, startedAt, taskSummary string | |
| 141 | + | if err := json.Unmarshal(entry["agent_id"], &agentID); err != nil { | |
| 142 | + | t.Fatal(err) | |
| 143 | + | } | |
| 144 | + | if err := json.Unmarshal(entry["started_at"], &startedAt); err != nil { | |
| 145 | + | t.Fatal(err) | |
| 146 | + | } | |
| 147 | + | if err := json.Unmarshal(entry["task_summary"], &taskSummary); err != nil { | |
| 148 | + | t.Fatal(err) | |
| 149 | + | } | |
| 150 | + | if agentID != agent.ID { | |
| 151 | + | t.Fatalf("agent id = %q, want %q", agentID, agent.ID) | |
| 137 | 152 | } | |
| 138 | - | if got.Agents[0].StartedAt == "" { | |
| 153 | + | if startedAt == "" { | |
| 139 | 154 | t.Fatal("started_at is empty") | |
| 140 | 155 | } | |
| 141 | - | if got.Agents[0].TaskSummary != "review architecture" { | |
| 142 | - | t.Fatalf("task summary = %q, want first task line", got.Agents[0].TaskSummary) | |
| 156 | + | if taskSummary != "review architecture" { | |
| 157 | + | t.Fatalf("task summary = %q, want first task line", taskSummary) | |
| 158 | + | } | |
| 159 | + | if _, present := entry["last_event"]; !present { | |
| 160 | + | t.Fatal("status response omits trace-backed last_event") | |
| 143 | 161 | } | |
| 144 | 162 | for _, forbidden := range []string{ | |
| 145 | 163 | "PRIVATE_PROMPT_MARKER", | |
| @@ -119,11 +119,7 @@ func TestSubagentStatusListsLiveWorkerWithoutPrivateOutput(t *testing.T) { | |||||
| 119 | } | 119 | } | ||
| 120 | 120 | ||||
| 121 | var got struct { | 121 | var got struct { | ||
| 122 | - | Agents []struct { | 122 | + | Agents []json.RawMessage `json:"agents"` |
| 123 | - | ID string `json:"agent_id"` | |||
| 124 | - | StartedAt string `json:"started_at"` | |||
| 125 | - | TaskSummary string `json:"task_summary"` | |||
| 126 | - | } `json:"agents"` | |||
| 127 | } | 123 | } | ||
| 128 | text := textResult(res.Content) | 124 | text := textResult(res.Content) | ||
| 129 | if err := json.Unmarshal([]byte(text), &got); err != nil { | 125 | if err := json.Unmarshal([]byte(text), &got); err != nil { | ||
| @@ -132,14 +128,36 @@ func TestSubagentStatusListsLiveWorkerWithoutPrivateOutput(t *testing.T) { | |||||
| 132 | if len(got.Agents) != 1 { | 128 | if len(got.Agents) != 1 { | ||
| 133 | t.Fatalf("listed agents = %d, want 1", len(got.Agents)) | 129 | t.Fatalf("listed agents = %d, want 1", len(got.Agents)) | ||
| 134 | } | 130 | } | ||
| 135 | - | if got.Agents[0].ID != agent.ID { | 131 | + | var entry map[string]json.RawMessage |
| 136 | - | t.Fatalf("agent id = %q, want %q", got.Agents[0].ID, agent.ID) | 132 | + | if err := json.Unmarshal(got.Agents[0], &entry); err != nil { |
| 133 | + | t.Fatal(err) | |||
| 134 | + | } | |||
| 135 | + | for _, forbiddenField := range []string{"state", "process_state", "turn_state", "updated_at", "finished_at"} { | |||
| 136 | + | if _, present := entry[forbiddenField]; present { | |||
| 137 | + | t.Fatalf("status response exposes removed lifecycle field %q", forbiddenField) | |||
| 138 | + | } | |||
| 139 | + | } | |||
| 140 | + | var agentID, startedAt, taskSummary string | |||
| 141 | + | if err := json.Unmarshal(entry["agent_id"], &agentID); err != nil { | |||
| 142 | + | t.Fatal(err) | |||
| 143 | + | } | |||
| 144 | + | if err := json.Unmarshal(entry["started_at"], &startedAt); err != nil { | |||
| 145 | + | t.Fatal(err) | |||
| 146 | + | } | |||
| 147 | + | if err := json.Unmarshal(entry["task_summary"], &taskSummary); err != nil { | |||
| 148 | + | t.Fatal(err) | |||
| 149 | + | } | |||
| 150 | + | if agentID != agent.ID { | |||
| 151 | + | t.Fatalf("agent id = %q, want %q", agentID, agent.ID) | |||
| 137 | } | 152 | } | ||
| 138 | - | if got.Agents[0].StartedAt == "" { | 153 | + | if startedAt == "" { |
| 139 | t.Fatal("started_at is empty") | 154 | t.Fatal("started_at is empty") | ||
| 140 | } | 155 | } | ||
| 141 | - | if got.Agents[0].TaskSummary != "review architecture" { | 156 | + | if taskSummary != "review architecture" { |
| 142 | - | t.Fatalf("task summary = %q, want first task line", got.Agents[0].TaskSummary) | 157 | + | t.Fatalf("task summary = %q, want first task line", taskSummary) |
| 158 | + | } | |||
| 159 | + | if _, present := entry["last_event"]; !present { | |||
| 160 | + | t.Fatal("status response omits trace-backed last_event") | |||
| 143 | } | 161 | } | ||
| 144 | for _, forbidden := range []string{ | 162 | for _, forbidden := range []string{ | ||
| 145 | "PRIVATE_PROMPT_MARKER", | 163 | "PRIVATE_PROMPT_MARKER", | ||
M packages/agent/tracecmd.go +3 -3
| @@ -17,11 +17,11 @@ func runDebugCommand(rawArgs []string, stdout io.Writer) (bool, error) { | |||
| 17 | 17 | if len(rawArgs) != 3 || rawArgs[1] != "trace" { | |
| 18 | 18 | return true, fmt.Errorf("usage: zut debug trace <bundle>") | |
| 19 | 19 | } | |
| 20 | - | events, err := subagents.ReadTrace(rawArgs[2]) | |
| 21 | - | if err != nil { | |
| 20 | + | events, readErr := subagents.ReadTrace(rawArgs[2]) | |
| 21 | + | if err := renderTraceInspection(stdout, subagents.ProjectTrace(events), time.Now()); err != nil { | |
| 22 | 22 | return true, err | |
| 23 | 23 | } | |
| 24 | - | return true, renderTraceInspection(stdout, subagents.ProjectTrace(events), time.Now()) | |
| 24 | + | return true, readErr | |
| 25 | 25 | } | |
| 26 | 26 | ||
| 27 | 27 | func renderTraceInspection(w io.Writer, views map[string]subagents.AgentTraceView, now time.Time) error { | |
| @@ -17,11 +17,11 @@ func runDebugCommand(rawArgs []string, stdout io.Writer) (bool, error) { | |||||
| 17 | if len(rawArgs) != 3 || rawArgs[1] != "trace" { | 17 | if len(rawArgs) != 3 || rawArgs[1] != "trace" { | ||
| 18 | return true, fmt.Errorf("usage: zut debug trace <bundle>") | 18 | return true, fmt.Errorf("usage: zut debug trace <bundle>") | ||
| 19 | } | 19 | } | ||
| 20 | - | events, err := subagents.ReadTrace(rawArgs[2]) | 20 | + | events, readErr := subagents.ReadTrace(rawArgs[2]) |
| 21 | - | if err != nil { | 21 | + | if err := renderTraceInspection(stdout, subagents.ProjectTrace(events), time.Now()); err != nil { |
| 22 | return true, err | 22 | return true, err | ||
| 23 | } | 23 | } | ||
| 24 | - | return true, renderTraceInspection(stdout, subagents.ProjectTrace(events), time.Now()) | 24 | + | return true, readErr |
| 25 | } | 25 | } | ||
| 26 | 26 | ||||
| 27 | func renderTraceInspection(w io.Writer, views map[string]subagents.AgentTraceView, now time.Time) error { | 27 | func renderTraceInspection(w io.Writer, views map[string]subagents.AgentTraceView, now time.Time) error { | ||