feat(subagents): expose live activity and watch option in status (#225)
- commit
- efd47b39906fa6f1b2a2ef22d1a4b4180116ce98
- author
- bnema · · 2026-09-24 10:16 UTC
- committer
- GitHub · · 2026-09-24 10:16 UTC
- parent
- 21dd32a
- signature
- GPG Unverified
B5690EEEBB952194The key is unknown - refs
- tag: v0.44.0
5 files · +255 -6
M docs/subagents.md +7 -0
| @@ -238,6 +238,13 @@ all of them. | |||
| 238 | 238 | metadata-only; foreign-owned results cannot be read. Result-read errors | |
| 239 | 239 | distinguish foreign ownership, a missing saved result, and permission | |
| 240 | 240 | denial without exposing filesystem paths or saved content. | |
| 241 | + | With `agent_id`, a running child also includes `activity`: a `phase` | |
| 242 | + | (`waiting_for_model`, `writing`, `tools`, or `working`), the tail of its | |
| 243 | + | visible text, and its active tools with truncated arguments. Hidden | |
| 244 | + | reasoning is never exposed. `watch` (1–60 seconds, requires `agent_id`) | |
| 245 | + | observes that activity and returns a `timeline` of phase and tool changes. | |
| 246 | + | It returns early when the child stops running. Use it to inspect progress, | |
| 247 | + | not to wait for completion. | |
| 241 | 248 | - `stop` accepts `agent_id` and stops one live child. | |
| 242 | 249 | - `resume` accepts `agent_id`, `prompt`, and optional `wait`. `prompt` is an | |
| 243 | 250 | explicit follow-up prompt for an existing child. After a terminal failure, | |
| @@ -238,6 +238,13 @@ all of them. | |||||
| 238 | metadata-only; foreign-owned results cannot be read. Result-read errors | 238 | metadata-only; foreign-owned results cannot be read. Result-read errors | ||
| 239 | distinguish foreign ownership, a missing saved result, and permission | 239 | distinguish foreign ownership, a missing saved result, and permission | ||
| 240 | denial without exposing filesystem paths or saved content. | 240 | denial without exposing filesystem paths or saved content. | ||
| 241 | + | With `agent_id`, a running child also includes `activity`: a `phase` | |||
| 242 | + | (`waiting_for_model`, `writing`, `tools`, or `working`), the tail of its | |||
| 243 | + | visible text, and its active tools with truncated arguments. Hidden | |||
| 244 | + | reasoning is never exposed. `watch` (1–60 seconds, requires `agent_id`) | |||
| 245 | + | observes that activity and returns a `timeline` of phase and tool changes. | |||
| 246 | + | It returns early when the child stops running. Use it to inspect progress, | |||
| 247 | + | not to wait for completion. | |||
| 241 | - `stop` accepts `agent_id` and stops one live child. | 248 | - `stop` accepts `agent_id` and stops one live child. | ||
| 242 | - `resume` accepts `agent_id`, `prompt`, and optional `wait`. `prompt` is an | 249 | - `resume` accepts `agent_id`, `prompt`, and optional `wait`. `prompt` is an | ||
| 243 | explicit follow-up prompt for an existing child. After a terminal failure, | 250 | explicit follow-up prompt for an existing child. After a terminal failure, | ||
M packages/agent/tools/subagent.go +8 -2
| @@ -43,7 +43,7 @@ const subagentSchema = `{ | |||
| 43 | 43 | "action": { | |
| 44 | 44 | "type": "string", | |
| 45 | 45 | "enum": ["spawn", "status", "stop", "resume"], | |
| 46 | - | "description": "One action per call. spawn delegates a new resident sub-agent. status reads bounded state for one child or lists the current set. stop requests termination of a stuck child. resume continues an existing child with a new prompt, keeping its session context, and can wait for that follow-up turn." | |
| 46 | + | "description": "One action per call. spawn delegates a new resident sub-agent. status reads bounded state for one child, including its current live activity, or lists the current set. stop requests termination of a stuck child. resume continues an existing child with a new prompt, keeping its session context, and can wait for that follow-up turn." | |
| 47 | 47 | }, | |
| 48 | 48 | "task": { | |
| 49 | 49 | "type": "string", | |
| @@ -93,6 +93,12 @@ const subagentSchema = `{ | |||
| 93 | 93 | "type": "boolean", | |
| 94 | 94 | "description": "With status and agent_id, retrieve the saved terminal result without running the child. Only valid when action is status." | |
| 95 | 95 | }, | |
| 96 | + | "watch": { | |
| 97 | + | "type": "integer", | |
| 98 | + | "minimum": 1, | |
| 99 | + | "maximum": 60, | |
| 100 | + | "description": "With status and agent_id, observe the child's live activity (waiting for the model, writing text, or running tools) for this many seconds and return a timeline of changes. Returns early when the child stops running. Use it to inspect progress, not to wait for completion. Only valid when action is status." | |
| 101 | + | }, | |
| 96 | 102 | "prompt": { | |
| 97 | 103 | "type": "string", | |
| 98 | 104 | "description": "New manager follow-up for the sub-agent. Its earlier task and conversation remain available in the retained session. Required when action is resume. After a terminal failure, inspect its saved result first; resume continues the retained session without discarding progress or satisfying required work. Combine with wait to block on that follow-up turn." | |
| @@ -106,7 +112,7 @@ const subagentSchema = `{ | |||
| 106 | 112 | // action is rejected before dispatch. | |
| 107 | 113 | var subagentActionFields = map[string][]string{ | |
| 108 | 114 | SubagentActionSpawn: {"task", "agent", "model", "provider", "reasoning", "fast_mode", "required", "wait", "isolation"}, | |
| 109 | - | SubagentActionStatus: {"agent_id", "include_result"}, | |
| 115 | + | SubagentActionStatus: {"agent_id", "include_result", "watch"}, | |
| 110 | 116 | SubagentActionStop: {"agent_id"}, | |
| 111 | 117 | SubagentActionResume: {"agent_id", "prompt", "wait"}, | |
| 112 | 118 | } | |
| @@ -43,7 +43,7 @@ const subagentSchema = `{ | |||||
| 43 | "action": { | 43 | "action": { | ||
| 44 | "type": "string", | 44 | "type": "string", | ||
| 45 | "enum": ["spawn", "status", "stop", "resume"], | 45 | "enum": ["spawn", "status", "stop", "resume"], | ||
| 46 | - | "description": "One action per call. spawn delegates a new resident sub-agent. status reads bounded state for one child or lists the current set. stop requests termination of a stuck child. resume continues an existing child with a new prompt, keeping its session context, and can wait for that follow-up turn." | 46 | + | "description": "One action per call. spawn delegates a new resident sub-agent. status reads bounded state for one child, including its current live activity, or lists the current set. stop requests termination of a stuck child. resume continues an existing child with a new prompt, keeping its session context, and can wait for that follow-up turn." |
| 47 | }, | 47 | }, | ||
| 48 | "task": { | 48 | "task": { | ||
| 49 | "type": "string", | 49 | "type": "string", | ||
| @@ -93,6 +93,12 @@ const subagentSchema = `{ | |||||
| 93 | "type": "boolean", | 93 | "type": "boolean", | ||
| 94 | "description": "With status and agent_id, retrieve the saved terminal result without running the child. Only valid when action is status." | 94 | "description": "With status and agent_id, retrieve the saved terminal result without running the child. Only valid when action is status." | ||
| 95 | }, | 95 | }, | ||
| 96 | + | "watch": { | |||
| 97 | + | "type": "integer", | |||
| 98 | + | "minimum": 1, | |||
| 99 | + | "maximum": 60, | |||
| 100 | + | "description": "With status and agent_id, observe the child's live activity (waiting for the model, writing text, or running tools) for this many seconds and return a timeline of changes. Returns early when the child stops running. Use it to inspect progress, not to wait for completion. Only valid when action is status." | |||
| 101 | + | }, | |||
| 96 | "prompt": { | 102 | "prompt": { | ||
| 97 | "type": "string", | 103 | "type": "string", | ||
| 98 | "description": "New manager follow-up for the sub-agent. Its earlier task and conversation remain available in the retained session. Required when action is resume. After a terminal failure, inspect its saved result first; resume continues the retained session without discarding progress or satisfying required work. Combine with wait to block on that follow-up turn." | 104 | "description": "New manager follow-up for the sub-agent. Its earlier task and conversation remain available in the retained session. Required when action is resume. After a terminal failure, inspect its saved result first; resume continues the retained session without discarding progress or satisfying required work. Combine with wait to block on that follow-up turn." | ||
| @@ -106,7 +112,7 @@ const subagentSchema = `{ | |||||
| 106 | // action is rejected before dispatch. | 112 | // action is rejected before dispatch. | ||
| 107 | var subagentActionFields = map[string][]string{ | 113 | var subagentActionFields = map[string][]string{ | ||
| 108 | SubagentActionSpawn: {"task", "agent", "model", "provider", "reasoning", "fast_mode", "required", "wait", "isolation"}, | 114 | SubagentActionSpawn: {"task", "agent", "model", "provider", "reasoning", "fast_mode", "required", "wait", "isolation"}, | ||
| 109 | - | SubagentActionStatus: {"agent_id", "include_result"}, | 115 | + | SubagentActionStatus: {"agent_id", "include_result", "watch"}, |
| 110 | SubagentActionStop: {"agent_id"}, | 116 | SubagentActionStop: {"agent_id"}, | ||
| 111 | SubagentActionResume: {"agent_id", "prompt", "wait"}, | 117 | SubagentActionResume: {"agent_id", "prompt", "wait"}, | ||
| 112 | } | 118 | } | ||
M packages/agent/tools/subagent_status.go +173 -3
| @@ -6,6 +6,8 @@ import ( | |||
| 6 | 6 | "fmt" | |
| 7 | 7 | "os" | |
| 8 | 8 | "strings" | |
| 9 | + | "time" | |
| 10 | + | "unicode/utf8" | |
| 9 | 11 | ||
| 10 | 12 | "github.com/bnema/zut/packages/agent/subagents" | |
| 11 | 13 | "github.com/bnema/zut/packages/core" | |
| @@ -25,12 +27,51 @@ type SubagentStatusTool struct { | |||
| 25 | 27 | type subagentStatusArgs struct { | |
| 26 | 28 | AgentID string `json:"agent_id,omitempty"` | |
| 27 | 29 | IncludeResult bool `json:"include_result,omitempty"` | |
| 30 | + | Watch int `json:"watch,omitempty"` | |
| 28 | 31 | } | |
| 29 | 32 | ||
| 33 | + | const ( | |
| 34 | + | subagentWatchMaxSeconds = 60 | |
| 35 | + | subagentWatchInterval = 250 * time.Millisecond | |
| 36 | + | subagentWatchMaxEntries = 40 | |
| 37 | + | subagentActivityText = 300 | |
| 38 | + | subagentActivityArgs = 200 | |
| 39 | + | ) | |
| 40 | + | ||
| 30 | 41 | type subagentStatusResponse struct { | |
| 31 | - | Agent *subagentStatusEntry `json:"agent,omitempty"` | |
| 32 | - | Agents []subagentStatusEntry `json:"agents"` | |
| 33 | - | Result *subagents.ResidentResult `json:"result,omitempty"` | |
| 42 | + | Agent *subagentStatusEntry `json:"agent,omitempty"` | |
| 43 | + | Agents []subagentStatusEntry `json:"agents"` | |
| 44 | + | Activity *subagentActivity `json:"activity,omitempty"` | |
| 45 | + | Watch *subagentWatchReport `json:"watch,omitempty"` | |
| 46 | + | Result *subagents.ResidentResult `json:"result,omitempty"` | |
| 47 | + | } | |
| 48 | + | ||
| 49 | + | // subagentActivity is a bounded view of the child's unfinished turn: what it | |
| 50 | + | // is doing now, the tail of visible text it is writing, and active tools. | |
| 51 | + | // Hidden reasoning is never exposed. | |
| 52 | + | type subagentActivity struct { | |
| 53 | + | Phase string `json:"phase"` | |
| 54 | + | Text string `json:"text,omitempty"` | |
| 55 | + | Tools []subagentActivityTool `json:"tools,omitempty"` | |
| 56 | + | } | |
| 57 | + | ||
| 58 | + | type subagentActivityTool struct { | |
| 59 | + | Name string `json:"name"` | |
| 60 | + | State string `json:"state"` | |
| 61 | + | Args string `json:"args,omitempty"` | |
| 62 | + | } | |
| 63 | + | ||
| 64 | + | type subagentWatchReport struct { | |
| 65 | + | Seconds float64 `json:"observed_seconds"` | |
| 66 | + | EndState subagents.ResidentState `json:"end_state"` | |
| 67 | + | Timeline []subagentWatchEntry `json:"timeline"` | |
| 68 | + | Truncated bool `json:"truncated,omitempty"` | |
| 69 | + | } | |
| 70 | + | ||
| 71 | + | type subagentWatchEntry struct { | |
| 72 | + | At string `json:"at"` | |
| 73 | + | State subagents.ResidentState `json:"state"` | |
| 74 | + | Activity *subagentActivity `json:"activity,omitempty"` | |
| 34 | 75 | } | |
| 35 | 76 | ||
| 36 | 77 | type subagentStatusEntry struct { | |
| @@ -74,6 +115,9 @@ func (t *SubagentStatusTool) Execute(ctx context.Context, raw json.RawMessage, _ | |||
| 74 | 115 | if args.IncludeResult { | |
| 75 | 116 | return protocolToolError(prefix + ": include_result requires agent_id") | |
| 76 | 117 | } | |
| 118 | + | if args.Watch != 0 { | |
| 119 | + | return protocolToolError(prefix + ": watch requires agent_id") | |
| 120 | + | } | |
| 77 | 121 | entries := make([]subagentStatusEntry, 0, len(snapshots)) | |
| 78 | 122 | for _, snapshot := range snapshots { | |
| 79 | 123 | entries = append(entries, publicResidentStatus(snapshot)) | |
| @@ -84,8 +128,28 @@ func (t *SubagentStatusTool) Execute(ctx context.Context, raw json.RawMessage, _ | |||
| 84 | 128 | if !ok { | |
| 85 | 129 | return protocolToolError(fmt.Sprintf("%s: no such agent %q", prefix, id)) | |
| 86 | 130 | } | |
| 131 | + | if args.Watch < 0 || args.Watch > subagentWatchMaxSeconds { | |
| 132 | + | return protocolToolError(fmt.Sprintf("%s: watch must be between 1 and %d seconds", prefix, subagentWatchMaxSeconds)) | |
| 133 | + | } | |
| 134 | + | if args.Watch > 0 && snapshot.OwnedElsewhere { | |
| 135 | + | return protocolToolError(prefix + ": watch is unavailable: child is owned by another zut process") | |
| 136 | + | } | |
| 87 | 137 | entry := publicResidentStatus(snapshot) | |
| 88 | 138 | response := subagentStatusResponse{Agent: &entry} | |
| 139 | + | if args.Watch > 0 { | |
| 140 | + | report, err := t.watch(ctx, snapshot.ID, time.Duration(args.Watch)*time.Second) | |
| 141 | + | if err != nil { | |
| 142 | + | return core.ToolResult{}, err | |
| 143 | + | } | |
| 144 | + | if report.EndState == "" { | |
| 145 | + | report.EndState = entry.State | |
| 146 | + | } | |
| 147 | + | response.Watch = &report | |
| 148 | + | entry.State = report.EndState | |
| 149 | + | } | |
| 150 | + | if live, ok := t.ResidentManager.Live(snapshot.ID); ok { | |
| 151 | + | response.Activity = residentActivity(entry.State, live) | |
| 152 | + | } | |
| 89 | 153 | if args.IncludeResult { | |
| 90 | 154 | result, err := t.ResidentManager.Result(snapshot.ID) | |
| 91 | 155 | if err != nil { | |
| @@ -107,6 +171,112 @@ func (t *SubagentStatusTool) Execute(ctx context.Context, raw json.RawMessage, _ | |||
| 107 | 171 | return renderSubagentStatus(response) | |
| 108 | 172 | } | |
| 109 | 173 | ||
| 174 | + | // watch samples the child's live projection until the duration elapses, the | |
| 175 | + | // child leaves the running/queued states, or ctx is canceled. It records an | |
| 176 | + | // entry only when the observable activity changes. | |
| 177 | + | func (t *SubagentStatusTool) watch(ctx context.Context, id string, duration time.Duration) (subagentWatchReport, error) { | |
| 178 | + | start := time.Now() | |
| 179 | + | deadline := time.NewTimer(duration) | |
| 180 | + | defer deadline.Stop() | |
| 181 | + | ticker := time.NewTicker(subagentWatchInterval) | |
| 182 | + | defer ticker.Stop() | |
| 183 | + | report := subagentWatchReport{Timeline: []subagentWatchEntry{}} | |
| 184 | + | last := "" | |
| 185 | + | if ctx == nil { | |
| 186 | + | ctx = context.Background() | |
| 187 | + | } | |
| 188 | + | for { | |
| 189 | + | state, ok := t.ResidentManager.State(id) | |
| 190 | + | if !ok { | |
| 191 | + | break | |
| 192 | + | } | |
| 193 | + | report.EndState = state | |
| 194 | + | var activity *subagentActivity | |
| 195 | + | if live, ok := t.ResidentManager.Live(id); ok { | |
| 196 | + | activity = residentActivity(state, live) | |
| 197 | + | } | |
| 198 | + | entry := subagentWatchEntry{State: state, Activity: activity} | |
| 199 | + | if key := watchChangeKey(entry); key != last { | |
| 200 | + | last = key | |
| 201 | + | if len(report.Timeline) < subagentWatchMaxEntries { | |
| 202 | + | entry.At = fmt.Sprintf("+%.1fs", time.Since(start).Seconds()) | |
| 203 | + | report.Timeline = append(report.Timeline, entry) | |
| 204 | + | } else { | |
| 205 | + | report.Truncated = true | |
| 206 | + | } | |
| 207 | + | } | |
| 208 | + | if state != subagents.ResidentRunning && state != subagents.ResidentQueued { | |
| 209 | + | break | |
| 210 | + | } | |
| 211 | + | select { | |
| 212 | + | case <-ctx.Done(): | |
| 213 | + | return subagentWatchReport{}, ctx.Err() | |
| 214 | + | case <-deadline.C: | |
| 215 | + | report.Seconds = roundSeconds(time.Since(start)) | |
| 216 | + | return report, nil | |
| 217 | + | case <-ticker.C: | |
| 218 | + | } | |
| 219 | + | } | |
| 220 | + | report.Seconds = roundSeconds(time.Since(start)) | |
| 221 | + | return report, nil | |
| 222 | + | } | |
| 223 | + | ||
| 224 | + | // watchChangeKey ignores streamed text growth so the timeline records phase | |
| 225 | + | // and tool transitions rather than every delta. | |
| 226 | + | func watchChangeKey(entry subagentWatchEntry) string { | |
| 227 | + | var key strings.Builder | |
| 228 | + | key.WriteString(string(entry.State)) | |
| 229 | + | if entry.Activity != nil { | |
| 230 | + | key.WriteString("|" + entry.Activity.Phase) | |
| 231 | + | for _, tool := range entry.Activity.Tools { | |
| 232 | + | key.WriteString("|" + tool.Name + ":" + tool.State) | |
| 233 | + | } | |
| 234 | + | } | |
| 235 | + | return key.String() | |
| 236 | + | } | |
| 237 | + | ||
| 238 | + | func roundSeconds(d time.Duration) float64 { | |
| 239 | + | return float64(d.Round(100*time.Millisecond)) / float64(time.Second) | |
| 240 | + | } | |
| 241 | + | ||
| 242 | + | // residentActivity summarizes a live snapshot. It returns nil when the child | |
| 243 | + | // is not running a turn. | |
| 244 | + | func residentActivity(state subagents.ResidentState, live subagents.ResidentLiveSnapshot) *subagentActivity { | |
| 245 | + | if state != subagents.ResidentRunning { | |
| 246 | + | return nil | |
| 247 | + | } | |
| 248 | + | activity := &subagentActivity{Text: tailRunes(live.AssistantText, subagentActivityText)} | |
| 249 | + | for _, tool := range live.Tools { | |
| 250 | + | activity.Tools = append(activity.Tools, subagentActivityTool{Name: tool.Name, State: string(tool.State), Args: headRunes(string(tool.Args), subagentActivityArgs)}) | |
| 251 | + | } | |
| 252 | + | switch { | |
| 253 | + | case len(activity.Tools) > 0: | |
| 254 | + | activity.Phase = "tools" | |
| 255 | + | case live.WaitingForModel: | |
| 256 | + | activity.Phase = "waiting_for_model" | |
| 257 | + | case activity.Text != "": | |
| 258 | + | activity.Phase = "writing" | |
| 259 | + | default: | |
| 260 | + | activity.Phase = "working" | |
| 261 | + | } | |
| 262 | + | return activity | |
| 263 | + | } | |
| 264 | + | ||
| 265 | + | func headRunes(s string, limit int) string { | |
| 266 | + | if utf8.RuneCountInString(s) <= limit { | |
| 267 | + | return s | |
| 268 | + | } | |
| 269 | + | return string([]rune(s)[:limit]) + "…" | |
| 270 | + | } | |
| 271 | + | ||
| 272 | + | func tailRunes(s string, limit int) string { | |
| 273 | + | runes := []rune(s) | |
| 274 | + | if len(runes) <= limit { | |
| 275 | + | return s | |
| 276 | + | } | |
| 277 | + | return "…" + string(runes[len(runes)-limit:]) | |
| 278 | + | } | |
| 279 | + | ||
| 110 | 280 | func findResidentStatusSnapshot(snapshots []subagents.ResidentSnapshot, id string) (subagents.ResidentSnapshot, bool) { | |
| 111 | 281 | for _, snapshot := range snapshots { | |
| 112 | 282 | if snapshot.ID == id { | |
| @@ -6,6 +6,8 @@ import ( | |||||
| 6 | "fmt" | 6 | "fmt" | ||
| 7 | "os" | 7 | "os" | ||
| 8 | "strings" | 8 | "strings" | ||
| 9 | + | "time" | |||
| 10 | + | "unicode/utf8" | |||
| 9 | 11 | ||||
| 10 | "github.com/bnema/zut/packages/agent/subagents" | 12 | "github.com/bnema/zut/packages/agent/subagents" | ||
| 11 | "github.com/bnema/zut/packages/core" | 13 | "github.com/bnema/zut/packages/core" | ||
| @@ -25,12 +27,51 @@ type SubagentStatusTool struct { | |||||
| 25 | type subagentStatusArgs struct { | 27 | type subagentStatusArgs struct { | ||
| 26 | AgentID string `json:"agent_id,omitempty"` | 28 | AgentID string `json:"agent_id,omitempty"` | ||
| 27 | IncludeResult bool `json:"include_result,omitempty"` | 29 | IncludeResult bool `json:"include_result,omitempty"` | ||
| 30 | + | Watch int `json:"watch,omitempty"` | |||
| 28 | } | 31 | } | ||
| 29 | 32 | ||||
| 33 | + | const ( | |||
| 34 | + | subagentWatchMaxSeconds = 60 | |||
| 35 | + | subagentWatchInterval = 250 * time.Millisecond | |||
| 36 | + | subagentWatchMaxEntries = 40 | |||
| 37 | + | subagentActivityText = 300 | |||
| 38 | + | subagentActivityArgs = 200 | |||
| 39 | + | ) | |||
| 40 | + | ||||
| 30 | type subagentStatusResponse struct { | 41 | type subagentStatusResponse struct { | ||
| 31 | - | Agent *subagentStatusEntry `json:"agent,omitempty"` | 42 | + | Agent *subagentStatusEntry `json:"agent,omitempty"` |
| 32 | - | Agents []subagentStatusEntry `json:"agents"` | 43 | + | Agents []subagentStatusEntry `json:"agents"` |
| 33 | - | Result *subagents.ResidentResult `json:"result,omitempty"` | 44 | + | Activity *subagentActivity `json:"activity,omitempty"` |
| 45 | + | Watch *subagentWatchReport `json:"watch,omitempty"` | |||
| 46 | + | Result *subagents.ResidentResult `json:"result,omitempty"` | |||
| 47 | + | } | |||
| 48 | + | ||||
| 49 | + | // subagentActivity is a bounded view of the child's unfinished turn: what it | |||
| 50 | + | // is doing now, the tail of visible text it is writing, and active tools. | |||
| 51 | + | // Hidden reasoning is never exposed. | |||
| 52 | + | type subagentActivity struct { | |||
| 53 | + | Phase string `json:"phase"` | |||
| 54 | + | Text string `json:"text,omitempty"` | |||
| 55 | + | Tools []subagentActivityTool `json:"tools,omitempty"` | |||
| 56 | + | } | |||
| 57 | + | ||||
| 58 | + | type subagentActivityTool struct { | |||
| 59 | + | Name string `json:"name"` | |||
| 60 | + | State string `json:"state"` | |||
| 61 | + | Args string `json:"args,omitempty"` | |||
| 62 | + | } | |||
| 63 | + | ||||
| 64 | + | type subagentWatchReport struct { | |||
| 65 | + | Seconds float64 `json:"observed_seconds"` | |||
| 66 | + | EndState subagents.ResidentState `json:"end_state"` | |||
| 67 | + | Timeline []subagentWatchEntry `json:"timeline"` | |||
| 68 | + | Truncated bool `json:"truncated,omitempty"` | |||
| 69 | + | } | |||
| 70 | + | ||||
| 71 | + | type subagentWatchEntry struct { | |||
| 72 | + | At string `json:"at"` | |||
| 73 | + | State subagents.ResidentState `json:"state"` | |||
| 74 | + | Activity *subagentActivity `json:"activity,omitempty"` | |||
| 34 | } | 75 | } | ||
| 35 | 76 | ||||
| 36 | type subagentStatusEntry struct { | 77 | type subagentStatusEntry struct { | ||
| @@ -74,6 +115,9 @@ func (t *SubagentStatusTool) Execute(ctx context.Context, raw json.RawMessage, _ | |||||
| 74 | if args.IncludeResult { | 115 | if args.IncludeResult { | ||
| 75 | return protocolToolError(prefix + ": include_result requires agent_id") | 116 | return protocolToolError(prefix + ": include_result requires agent_id") | ||
| 76 | } | 117 | } | ||
| 118 | + | if args.Watch != 0 { | |||
| 119 | + | return protocolToolError(prefix + ": watch requires agent_id") | |||
| 120 | + | } | |||
| 77 | entries := make([]subagentStatusEntry, 0, len(snapshots)) | 121 | entries := make([]subagentStatusEntry, 0, len(snapshots)) | ||
| 78 | for _, snapshot := range snapshots { | 122 | for _, snapshot := range snapshots { | ||
| 79 | entries = append(entries, publicResidentStatus(snapshot)) | 123 | entries = append(entries, publicResidentStatus(snapshot)) | ||
| @@ -84,8 +128,28 @@ func (t *SubagentStatusTool) Execute(ctx context.Context, raw json.RawMessage, _ | |||||
| 84 | if !ok { | 128 | if !ok { | ||
| 85 | return protocolToolError(fmt.Sprintf("%s: no such agent %q", prefix, id)) | 129 | return protocolToolError(fmt.Sprintf("%s: no such agent %q", prefix, id)) | ||
| 86 | } | 130 | } | ||
| 131 | + | if args.Watch < 0 || args.Watch > subagentWatchMaxSeconds { | |||
| 132 | + | return protocolToolError(fmt.Sprintf("%s: watch must be between 1 and %d seconds", prefix, subagentWatchMaxSeconds)) | |||
| 133 | + | } | |||
| 134 | + | if args.Watch > 0 && snapshot.OwnedElsewhere { | |||
| 135 | + | return protocolToolError(prefix + ": watch is unavailable: child is owned by another zut process") | |||
| 136 | + | } | |||
| 87 | entry := publicResidentStatus(snapshot) | 137 | entry := publicResidentStatus(snapshot) | ||
| 88 | response := subagentStatusResponse{Agent: &entry} | 138 | response := subagentStatusResponse{Agent: &entry} | ||
| 139 | + | if args.Watch > 0 { | |||
| 140 | + | report, err := t.watch(ctx, snapshot.ID, time.Duration(args.Watch)*time.Second) | |||
| 141 | + | if err != nil { | |||
| 142 | + | return core.ToolResult{}, err | |||
| 143 | + | } | |||
| 144 | + | if report.EndState == "" { | |||
| 145 | + | report.EndState = entry.State | |||
| 146 | + | } | |||
| 147 | + | response.Watch = &report | |||
| 148 | + | entry.State = report.EndState | |||
| 149 | + | } | |||
| 150 | + | if live, ok := t.ResidentManager.Live(snapshot.ID); ok { | |||
| 151 | + | response.Activity = residentActivity(entry.State, live) | |||
| 152 | + | } | |||
| 89 | if args.IncludeResult { | 153 | if args.IncludeResult { | ||
| 90 | result, err := t.ResidentManager.Result(snapshot.ID) | 154 | result, err := t.ResidentManager.Result(snapshot.ID) | ||
| 91 | if err != nil { | 155 | if err != nil { | ||
| @@ -107,6 +171,112 @@ func (t *SubagentStatusTool) Execute(ctx context.Context, raw json.RawMessage, _ | |||||
| 107 | return renderSubagentStatus(response) | 171 | return renderSubagentStatus(response) | ||
| 108 | } | 172 | } | ||
| 109 | 173 | ||||
| 174 | + | // watch samples the child's live projection until the duration elapses, the | |||
| 175 | + | // child leaves the running/queued states, or ctx is canceled. It records an | |||
| 176 | + | // entry only when the observable activity changes. | |||
| 177 | + | func (t *SubagentStatusTool) watch(ctx context.Context, id string, duration time.Duration) (subagentWatchReport, error) { | |||
| 178 | + | start := time.Now() | |||
| 179 | + | deadline := time.NewTimer(duration) | |||
| 180 | + | defer deadline.Stop() | |||
| 181 | + | ticker := time.NewTicker(subagentWatchInterval) | |||
| 182 | + | defer ticker.Stop() | |||
| 183 | + | report := subagentWatchReport{Timeline: []subagentWatchEntry{}} | |||
| 184 | + | last := "" | |||
| 185 | + | if ctx == nil { | |||
| 186 | + | ctx = context.Background() | |||
| 187 | + | } | |||
| 188 | + | for { | |||
| 189 | + | state, ok := t.ResidentManager.State(id) | |||
| 190 | + | if !ok { | |||
| 191 | + | break | |||
| 192 | + | } | |||
| 193 | + | report.EndState = state | |||
| 194 | + | var activity *subagentActivity | |||
| 195 | + | if live, ok := t.ResidentManager.Live(id); ok { | |||
| 196 | + | activity = residentActivity(state, live) | |||
| 197 | + | } | |||
| 198 | + | entry := subagentWatchEntry{State: state, Activity: activity} | |||
| 199 | + | if key := watchChangeKey(entry); key != last { | |||
| 200 | + | last = key | |||
| 201 | + | if len(report.Timeline) < subagentWatchMaxEntries { | |||
| 202 | + | entry.At = fmt.Sprintf("+%.1fs", time.Since(start).Seconds()) | |||
| 203 | + | report.Timeline = append(report.Timeline, entry) | |||
| 204 | + | } else { | |||
| 205 | + | report.Truncated = true | |||
| 206 | + | } | |||
| 207 | + | } | |||
| 208 | + | if state != subagents.ResidentRunning && state != subagents.ResidentQueued { | |||
| 209 | + | break | |||
| 210 | + | } | |||
| 211 | + | select { | |||
| 212 | + | case <-ctx.Done(): | |||
| 213 | + | return subagentWatchReport{}, ctx.Err() | |||
| 214 | + | case <-deadline.C: | |||
| 215 | + | report.Seconds = roundSeconds(time.Since(start)) | |||
| 216 | + | return report, nil | |||
| 217 | + | case <-ticker.C: | |||
| 218 | + | } | |||
| 219 | + | } | |||
| 220 | + | report.Seconds = roundSeconds(time.Since(start)) | |||
| 221 | + | return report, nil | |||
| 222 | + | } | |||
| 223 | + | ||||
| 224 | + | // watchChangeKey ignores streamed text growth so the timeline records phase | |||
| 225 | + | // and tool transitions rather than every delta. | |||
| 226 | + | func watchChangeKey(entry subagentWatchEntry) string { | |||
| 227 | + | var key strings.Builder | |||
| 228 | + | key.WriteString(string(entry.State)) | |||
| 229 | + | if entry.Activity != nil { | |||
| 230 | + | key.WriteString("|" + entry.Activity.Phase) | |||
| 231 | + | for _, tool := range entry.Activity.Tools { | |||
| 232 | + | key.WriteString("|" + tool.Name + ":" + tool.State) | |||
| 233 | + | } | |||
| 234 | + | } | |||
| 235 | + | return key.String() | |||
| 236 | + | } | |||
| 237 | + | ||||
| 238 | + | func roundSeconds(d time.Duration) float64 { | |||
| 239 | + | return float64(d.Round(100*time.Millisecond)) / float64(time.Second) | |||
| 240 | + | } | |||
| 241 | + | ||||
| 242 | + | // residentActivity summarizes a live snapshot. It returns nil when the child | |||
| 243 | + | // is not running a turn. | |||
| 244 | + | func residentActivity(state subagents.ResidentState, live subagents.ResidentLiveSnapshot) *subagentActivity { | |||
| 245 | + | if state != subagents.ResidentRunning { | |||
| 246 | + | return nil | |||
| 247 | + | } | |||
| 248 | + | activity := &subagentActivity{Text: tailRunes(live.AssistantText, subagentActivityText)} | |||
| 249 | + | for _, tool := range live.Tools { | |||
| 250 | + | activity.Tools = append(activity.Tools, subagentActivityTool{Name: tool.Name, State: string(tool.State), Args: headRunes(string(tool.Args), subagentActivityArgs)}) | |||
| 251 | + | } | |||
| 252 | + | switch { | |||
| 253 | + | case len(activity.Tools) > 0: | |||
| 254 | + | activity.Phase = "tools" | |||
| 255 | + | case live.WaitingForModel: | |||
| 256 | + | activity.Phase = "waiting_for_model" | |||
| 257 | + | case activity.Text != "": | |||
| 258 | + | activity.Phase = "writing" | |||
| 259 | + | default: | |||
| 260 | + | activity.Phase = "working" | |||
| 261 | + | } | |||
| 262 | + | return activity | |||
| 263 | + | } | |||
| 264 | + | ||||
| 265 | + | func headRunes(s string, limit int) string { | |||
| 266 | + | if utf8.RuneCountInString(s) <= limit { | |||
| 267 | + | return s | |||
| 268 | + | } | |||
| 269 | + | return string([]rune(s)[:limit]) + "…" | |||
| 270 | + | } | |||
| 271 | + | ||||
| 272 | + | func tailRunes(s string, limit int) string { | |||
| 273 | + | runes := []rune(s) | |||
| 274 | + | if len(runes) <= limit { | |||
| 275 | + | return s | |||
| 276 | + | } | |||
| 277 | + | return "…" + string(runes[len(runes)-limit:]) | |||
| 278 | + | } | |||
| 279 | + | ||||
| 110 | func findResidentStatusSnapshot(snapshots []subagents.ResidentSnapshot, id string) (subagents.ResidentSnapshot, bool) { | 280 | func findResidentStatusSnapshot(snapshots []subagents.ResidentSnapshot, id string) (subagents.ResidentSnapshot, bool) { | ||
| 111 | for _, snapshot := range snapshots { | 281 | for _, snapshot := range snapshots { | ||
| 112 | if snapshot.ID == id { | 282 | if snapshot.ID == id { | ||
M packages/agent/tools/subagent_status_test.go +65 -0
| @@ -9,8 +9,73 @@ import ( | |||
| 9 | 9 | "testing" | |
| 10 | 10 | ||
| 11 | 11 | "github.com/bnema/zut/packages/agent/subagents" | |
| 12 | + | "github.com/bnema/zut/packages/core" | |
| 12 | 13 | ) | |
| 13 | 14 | ||
| 15 | + | func TestResidentStatusWatchReportsLiveActivity(t *testing.T) { | |
| 16 | + | ready := make(chan struct{}) | |
| 17 | + | release := make(chan struct{}) | |
| 18 | + | manager := subagents.NewResidentManager(t.TempDir(), func(_ subagents.ResidentChildSpec, journal *subagents.ResidentJournal) (subagents.ResidentTurnRunner, error) { | |
| 19 | + | return func(ctx context.Context, _ string) error { | |
| 20 | + | _ = journal.RecordAgentEvent(core.EvTextDelta{Delta: "Reading the parser"}) | |
| 21 | + | _ = journal.RecordAgentEvent(core.EvToolExecutionStarted{ID: "call-1", Name: "read"}) | |
| 22 | + | close(ready) | |
| 23 | + | select { | |
| 24 | + | case <-release: | |
| 25 | + | return nil | |
| 26 | + | case <-ctx.Done(): | |
| 27 | + | return ctx.Err() | |
| 28 | + | } | |
| 29 | + | }, nil | |
| 30 | + | }) | |
| 31 | + | t.Cleanup(func() { _ = manager.Close(context.Background()) }) | |
| 32 | + | spec := subagents.ResidentChildSpec{ID: "watched", InitialTurnID: "initial", SessionID: "session", Provider: "openai", Model: "test"} | |
| 33 | + | if _, err := manager.Spawn(t.Context(), spec, "investigate"); err != nil { | |
| 34 | + | t.Fatal(err) | |
| 35 | + | } | |
| 36 | + | <-ready | |
| 37 | + | tool := &SubagentStatusTool{ResidentManager: manager, Enabled: func() bool { return true }} | |
| 38 | + | ||
| 39 | + | result, err := tool.Execute(t.Context(), json.RawMessage(`{"agent_id":"watched","watch":1}`), nil) | |
| 40 | + | if err != nil || result.IsError { | |
| 41 | + | t.Fatalf("result = %#v, error = %v", result, err) | |
| 42 | + | } | |
| 43 | + | var response subagentStatusResponse | |
| 44 | + | if err := json.Unmarshal([]byte(toolResultText(t, result)), &response); err != nil { | |
| 45 | + | t.Fatal(err) | |
| 46 | + | } | |
| 47 | + | if response.Watch == nil || len(response.Watch.Timeline) == 0 || response.Watch.EndState != subagents.ResidentRunning { | |
| 48 | + | t.Fatalf("watch = %#v", response.Watch) | |
| 49 | + | } | |
| 50 | + | activity := response.Activity | |
| 51 | + | if activity == nil || activity.Phase != "tools" || activity.Text != "Reading the parser" || len(activity.Tools) != 1 || activity.Tools[0].Name != "read" { | |
| 52 | + | t.Fatalf("activity = %#v", activity) | |
| 53 | + | } | |
| 54 | + | ||
| 55 | + | close(release) | |
| 56 | + | result, err = tool.Execute(t.Context(), json.RawMessage(`{"agent_id":"watched","watch":60}`), nil) | |
| 57 | + | if err != nil || result.IsError { | |
| 58 | + | t.Fatalf("result = %#v, error = %v", result, err) | |
| 59 | + | } | |
| 60 | + | response = subagentStatusResponse{} | |
| 61 | + | if err := json.Unmarshal([]byte(toolResultText(t, result)), &response); err != nil { | |
| 62 | + | t.Fatal(err) | |
| 63 | + | } | |
| 64 | + | if response.Watch == nil || response.Watch.EndState == subagents.ResidentRunning || response.Watch.Seconds >= 30 || response.Activity != nil { | |
| 65 | + | t.Fatalf("watch did not end with the turn: %#v", response) | |
| 66 | + | } | |
| 67 | + | } | |
| 68 | + | ||
| 69 | + | func TestResidentStatusWatchValidation(t *testing.T) { | |
| 70 | + | manager := subagents.NewResidentManager(t.TempDir(), nil) | |
| 71 | + | t.Cleanup(func() { _ = manager.Close(context.Background()) }) | |
| 72 | + | tool := &SubagentStatusTool{ResidentManager: manager, Enabled: func() bool { return true }} | |
| 73 | + | result, err := tool.Execute(t.Context(), json.RawMessage(`{"watch":5}`), nil) | |
| 74 | + | if err != nil || !result.IsError || !strings.Contains(toolResultText(t, result), "watch requires agent_id") { | |
| 75 | + | t.Fatalf("result = %#v, error = %v", result, err) | |
| 76 | + | } | |
| 77 | + | } | |
| 78 | + | ||
| 14 | 79 | func TestResidentStatusResultReadErrors(t *testing.T) { | |
| 15 | 80 | for _, scenario := range []struct { | |
| 16 | 81 | name string | |
| @@ -9,8 +9,73 @@ import ( | |||||
| 9 | "testing" | 9 | "testing" | ||
| 10 | 10 | ||||
| 11 | "github.com/bnema/zut/packages/agent/subagents" | 11 | "github.com/bnema/zut/packages/agent/subagents" | ||
| 12 | + | "github.com/bnema/zut/packages/core" | |||
| 12 | ) | 13 | ) | ||
| 13 | 14 | ||||
| 15 | + | func TestResidentStatusWatchReportsLiveActivity(t *testing.T) { | |||
| 16 | + | ready := make(chan struct{}) | |||
| 17 | + | release := make(chan struct{}) | |||
| 18 | + | manager := subagents.NewResidentManager(t.TempDir(), func(_ subagents.ResidentChildSpec, journal *subagents.ResidentJournal) (subagents.ResidentTurnRunner, error) { | |||
| 19 | + | return func(ctx context.Context, _ string) error { | |||
| 20 | + | _ = journal.RecordAgentEvent(core.EvTextDelta{Delta: "Reading the parser"}) | |||
| 21 | + | _ = journal.RecordAgentEvent(core.EvToolExecutionStarted{ID: "call-1", Name: "read"}) | |||
| 22 | + | close(ready) | |||
| 23 | + | select { | |||
| 24 | + | case <-release: | |||
| 25 | + | return nil | |||
| 26 | + | case <-ctx.Done(): | |||
| 27 | + | return ctx.Err() | |||
| 28 | + | } | |||
| 29 | + | }, nil | |||
| 30 | + | }) | |||
| 31 | + | t.Cleanup(func() { _ = manager.Close(context.Background()) }) | |||
| 32 | + | spec := subagents.ResidentChildSpec{ID: "watched", InitialTurnID: "initial", SessionID: "session", Provider: "openai", Model: "test"} | |||
| 33 | + | if _, err := manager.Spawn(t.Context(), spec, "investigate"); err != nil { | |||
| 34 | + | t.Fatal(err) | |||
| 35 | + | } | |||
| 36 | + | <-ready | |||
| 37 | + | tool := &SubagentStatusTool{ResidentManager: manager, Enabled: func() bool { return true }} | |||
| 38 | + | ||||
| 39 | + | result, err := tool.Execute(t.Context(), json.RawMessage(`{"agent_id":"watched","watch":1}`), nil) | |||
| 40 | + | if err != nil || result.IsError { | |||
| 41 | + | t.Fatalf("result = %#v, error = %v", result, err) | |||
| 42 | + | } | |||
| 43 | + | var response subagentStatusResponse | |||
| 44 | + | if err := json.Unmarshal([]byte(toolResultText(t, result)), &response); err != nil { | |||
| 45 | + | t.Fatal(err) | |||
| 46 | + | } | |||
| 47 | + | if response.Watch == nil || len(response.Watch.Timeline) == 0 || response.Watch.EndState != subagents.ResidentRunning { | |||
| 48 | + | t.Fatalf("watch = %#v", response.Watch) | |||
| 49 | + | } | |||
| 50 | + | activity := response.Activity | |||
| 51 | + | if activity == nil || activity.Phase != "tools" || activity.Text != "Reading the parser" || len(activity.Tools) != 1 || activity.Tools[0].Name != "read" { | |||
| 52 | + | t.Fatalf("activity = %#v", activity) | |||
| 53 | + | } | |||
| 54 | + | ||||
| 55 | + | close(release) | |||
| 56 | + | result, err = tool.Execute(t.Context(), json.RawMessage(`{"agent_id":"watched","watch":60}`), nil) | |||
| 57 | + | if err != nil || result.IsError { | |||
| 58 | + | t.Fatalf("result = %#v, error = %v", result, err) | |||
| 59 | + | } | |||
| 60 | + | response = subagentStatusResponse{} | |||
| 61 | + | if err := json.Unmarshal([]byte(toolResultText(t, result)), &response); err != nil { | |||
| 62 | + | t.Fatal(err) | |||
| 63 | + | } | |||
| 64 | + | if response.Watch == nil || response.Watch.EndState == subagents.ResidentRunning || response.Watch.Seconds >= 30 || response.Activity != nil { | |||
| 65 | + | t.Fatalf("watch did not end with the turn: %#v", response) | |||
| 66 | + | } | |||
| 67 | + | } | |||
| 68 | + | ||||
| 69 | + | func TestResidentStatusWatchValidation(t *testing.T) { | |||
| 70 | + | manager := subagents.NewResidentManager(t.TempDir(), nil) | |||
| 71 | + | t.Cleanup(func() { _ = manager.Close(context.Background()) }) | |||
| 72 | + | tool := &SubagentStatusTool{ResidentManager: manager, Enabled: func() bool { return true }} | |||
| 73 | + | result, err := tool.Execute(t.Context(), json.RawMessage(`{"watch":5}`), nil) | |||
| 74 | + | if err != nil || !result.IsError || !strings.Contains(toolResultText(t, result), "watch requires agent_id") { | |||
| 75 | + | t.Fatalf("result = %#v, error = %v", result, err) | |||
| 76 | + | } | |||
| 77 | + | } | |||
| 78 | + | ||||
| 14 | func TestResidentStatusResultReadErrors(t *testing.T) { | 79 | func TestResidentStatusResultReadErrors(t *testing.T) { | ||
| 15 | for _, scenario := range []struct { | 80 | for _, scenario := range []struct { | ||
| 16 | name string | 81 | name string | ||
M packages/agent/tools/subagent_test.go +2 -1
| @@ -63,7 +63,7 @@ func TestSubagentFacadeNameAndSchema(t *testing.T) { | |||
| 63 | 63 | } | |
| 64 | 64 | for _, field := range []string{ | |
| 65 | 65 | "action", "task", "agent", "model", "provider", "reasoning", "fast_mode", | |
| 66 | - | "required", "wait", "isolation", "agent_id", "include_result", "prompt", | |
| 66 | + | "required", "wait", "isolation", "agent_id", "include_result", "watch", "prompt", | |
| 67 | 67 | } { | |
| 68 | 68 | if _, ok := schema.Properties[field]; !ok { | |
| 69 | 69 | t.Fatalf("schema is missing property %q", field) | |
| @@ -123,6 +123,7 @@ var subagentActionAnnotation = map[string]string{ | |||
| 123 | 123 | "isolation": "Only valid when action is spawn.", | |
| 124 | 124 | "agent_id": "Required when action is stop or resume.", | |
| 125 | 125 | "include_result": "Only valid when action is status.", | |
| 126 | + | "watch": "Only valid when action is status.", | |
| 126 | 127 | "prompt": "Required when action is resume.", | |
| 127 | 128 | } | |
| 128 | 129 | ||
| @@ -63,7 +63,7 @@ func TestSubagentFacadeNameAndSchema(t *testing.T) { | |||||
| 63 | } | 63 | } | ||
| 64 | for _, field := range []string{ | 64 | for _, field := range []string{ | ||
| 65 | "action", "task", "agent", "model", "provider", "reasoning", "fast_mode", | 65 | "action", "task", "agent", "model", "provider", "reasoning", "fast_mode", | ||
| 66 | - | "required", "wait", "isolation", "agent_id", "include_result", "prompt", | 66 | + | "required", "wait", "isolation", "agent_id", "include_result", "watch", "prompt", |
| 67 | } { | 67 | } { | ||
| 68 | if _, ok := schema.Properties[field]; !ok { | 68 | if _, ok := schema.Properties[field]; !ok { | ||
| 69 | t.Fatalf("schema is missing property %q", field) | 69 | t.Fatalf("schema is missing property %q", field) | ||
| @@ -123,6 +123,7 @@ var subagentActionAnnotation = map[string]string{ | |||||
| 123 | "isolation": "Only valid when action is spawn.", | 123 | "isolation": "Only valid when action is spawn.", | ||
| 124 | "agent_id": "Required when action is stop or resume.", | 124 | "agent_id": "Required when action is stop or resume.", | ||
| 125 | "include_result": "Only valid when action is status.", | 125 | "include_result": "Only valid when action is status.", | ||
| 126 | + | "watch": "Only valid when action is status.", | |||
| 126 | "prompt": "Required when action is resume.", | 127 | "prompt": "Required when action is resume.", | ||
| 127 | } | 128 | } | ||
| 128 | 129 | ||||