Merge pull request #83 from bnema/fix/subagent-turn-timeout-lifecycle
fix(subagents): scope timeouts to active turns
10 files · +131 -15
M docs/subagents.md +1 -1
| @@ -127,7 +127,7 @@ subagent://<id>/patch | |||
| 127 | 127 | ||
| 128 | 128 | `events.jsonl` is appended as the worker emits events, including partial `message.delta` output. The dashboard and `subagent://<id>/history` replay those deltas, so output received before a worker finishes remains recoverable rather than waiting for a final assistant message. | |
| 129 | 129 | ||
| 130 | - | When a subagent deadline expires, the supervisor requests worker shutdown and allows the configured grace period for the worker to cancel its turn and write its session and result. The terminal result remains a failure with error code `deadline_exceeded`, preserves any output already received, and directs the caller to its history. If the worker does not stop during that grace period, the supervisor forcefully cancels it to preserve the timeout limit. | |
| 130 | + | A subagent timeout applies to each active delegated turn. When the deadline expires, the worker cancels that turn, writes a failed result with error code `deadline_exceeded`, preserves any output already received, and remains idle so an explicit follow-up can resume its session. Idle time does not consume the next turn's timeout, and each resumed turn receives a fresh deadline. Process shutdown still allows the configured grace period for the worker to write its session and result before forceful cancellation. | |
| 131 | 131 | ||
| 132 | 132 | Use `/subagents resume-session <id>` to continue the existing session without replaying its original task. Use `/subagents restart-task <id>` only when intentionally starting the stored task again. Cancellation requests graceful shutdown first, then forcefully cancels after the configured grace period. | |
| 133 | 133 | ||
| @@ -127,7 +127,7 @@ subagent://<id>/patch | |||||
| 127 | 127 | ||||
| 128 | `events.jsonl` is appended as the worker emits events, including partial `message.delta` output. The dashboard and `subagent://<id>/history` replay those deltas, so output received before a worker finishes remains recoverable rather than waiting for a final assistant message. | 128 | `events.jsonl` is appended as the worker emits events, including partial `message.delta` output. The dashboard and `subagent://<id>/history` replay those deltas, so output received before a worker finishes remains recoverable rather than waiting for a final assistant message. | ||
| 129 | 129 | ||||
| 130 | - | When a subagent deadline expires, the supervisor requests worker shutdown and allows the configured grace period for the worker to cancel its turn and write its session and result. The terminal result remains a failure with error code `deadline_exceeded`, preserves any output already received, and directs the caller to its history. If the worker does not stop during that grace period, the supervisor forcefully cancels it to preserve the timeout limit. | 130 | + | A subagent timeout applies to each active delegated turn. When the deadline expires, the worker cancels that turn, writes a failed result with error code `deadline_exceeded`, preserves any output already received, and remains idle so an explicit follow-up can resume its session. Idle time does not consume the next turn's timeout, and each resumed turn receives a fresh deadline. Process shutdown still allows the configured grace period for the worker to write its session and result before forceful cancellation. |
| 131 | 131 | ||||
| 132 | Use `/subagents resume-session <id>` to continue the existing session without replaying its original task. Use `/subagents restart-task <id>` only when intentionally starting the stored task again. Cancellation requests graceful shutdown first, then forcefully cancels after the configured grace period. | 132 | Use `/subagents resume-session <id>` to continue the existing session without replaying its original task. Use `/subagents restart-task <id>` only when intentionally starting the stored task again. Cancellation requests graceful shutdown first, then forcefully cancels after the configured grace period. | ||
| 133 | 133 | ||||
M packages/agent/args.go +14 -0
| @@ -5,6 +5,7 @@ import ( | |||
| 5 | 5 | "os" | |
| 6 | 6 | "strconv" | |
| 7 | 7 | "strings" | |
| 8 | + | "time" | |
| 8 | 9 | ||
| 9 | 10 | "github.com/bnema/zut/packages/agent/subagents" | |
| 10 | 11 | "github.com/bnema/zut/packages/agent/tools" | |
| @@ -143,6 +144,9 @@ type Args struct { | |||
| 143 | 144 | ||
| 144 | 145 | // SubagentMaxTurns limits message turns in one worker run. | |
| 145 | 146 | SubagentMaxTurns int | |
| 147 | + | // SubagentTurnTimeout limits each active delegated turn while leaving the | |
| 148 | + | // worker process alive between turns for explicit follow-ups. | |
| 149 | + | SubagentTurnTimeout time.Duration | |
| 146 | 150 | // SubagentLifetimeTurns and SubagentRunTurns are supervisor-provided | |
| 147 | 151 | // counters used to resume a worker without losing observability or its | |
| 148 | 152 | // current-run budget. They are internal worker flags, not user policy. | |
| @@ -384,6 +388,16 @@ func ParseArgs(in []string) (Args, error) { | |||
| 384 | 388 | return a, fmt.Errorf("--max-turns must be a positive integer") | |
| 385 | 389 | } | |
| 386 | 390 | a.SubagentMaxTurns = n | |
| 391 | + | case "--subagent-turn-timeout": | |
| 392 | + | v, err := want(&i, arg) | |
| 393 | + | if err != nil { | |
| 394 | + | return a, err | |
| 395 | + | } | |
| 396 | + | timeout, err := time.ParseDuration(v) | |
| 397 | + | if err != nil || timeout <= 0 { | |
| 398 | + | return a, fmt.Errorf("--subagent-turn-timeout must be a positive duration") | |
| 399 | + | } | |
| 400 | + | a.SubagentTurnTimeout = timeout | |
| 387 | 401 | case "--subagent-lifetime-turns": | |
| 388 | 402 | v, err := want(&i, arg) | |
| 389 | 403 | if err != nil { | |
| @@ -5,6 +5,7 @@ import ( | |||||
| 5 | "os" | 5 | "os" | ||
| 6 | "strconv" | 6 | "strconv" | ||
| 7 | "strings" | 7 | "strings" | ||
| 8 | + | "time" | |||
| 8 | 9 | ||||
| 9 | "github.com/bnema/zut/packages/agent/subagents" | 10 | "github.com/bnema/zut/packages/agent/subagents" | ||
| 10 | "github.com/bnema/zut/packages/agent/tools" | 11 | "github.com/bnema/zut/packages/agent/tools" | ||
| @@ -143,6 +144,9 @@ type Args struct { | |||||
| 143 | 144 | ||||
| 144 | // SubagentMaxTurns limits message turns in one worker run. | 145 | // SubagentMaxTurns limits message turns in one worker run. | ||
| 145 | SubagentMaxTurns int | 146 | SubagentMaxTurns int | ||
| 147 | + | // SubagentTurnTimeout limits each active delegated turn while leaving the | |||
| 148 | + | // worker process alive between turns for explicit follow-ups. | |||
| 149 | + | SubagentTurnTimeout time.Duration | |||
| 146 | // SubagentLifetimeTurns and SubagentRunTurns are supervisor-provided | 150 | // SubagentLifetimeTurns and SubagentRunTurns are supervisor-provided | ||
| 147 | // counters used to resume a worker without losing observability or its | 151 | // counters used to resume a worker without losing observability or its | ||
| 148 | // current-run budget. They are internal worker flags, not user policy. | 152 | // current-run budget. They are internal worker flags, not user policy. | ||
| @@ -384,6 +388,16 @@ func ParseArgs(in []string) (Args, error) { | |||||
| 384 | return a, fmt.Errorf("--max-turns must be a positive integer") | 388 | return a, fmt.Errorf("--max-turns must be a positive integer") | ||
| 385 | } | 389 | } | ||
| 386 | a.SubagentMaxTurns = n | 390 | a.SubagentMaxTurns = n | ||
| 391 | + | case "--subagent-turn-timeout": | |||
| 392 | + | v, err := want(&i, arg) | |||
| 393 | + | if err != nil { | |||
| 394 | + | return a, err | |||
| 395 | + | } | |||
| 396 | + | timeout, err := time.ParseDuration(v) | |||
| 397 | + | if err != nil || timeout <= 0 { | |||
| 398 | + | return a, fmt.Errorf("--subagent-turn-timeout must be a positive duration") | |||
| 399 | + | } | |||
| 400 | + | a.SubagentTurnTimeout = timeout | |||
| 387 | case "--subagent-lifetime-turns": | 401 | case "--subagent-lifetime-turns": | ||
| 388 | v, err := want(&i, arg) | 402 | v, err := want(&i, arg) | ||
| 389 | if err != nil { | 403 | if err != nil { | ||
M packages/agent/args_test.go +17 -0
| @@ -2,6 +2,7 @@ package agent | |||
| 2 | 2 | ||
| 3 | 3 | import ( | |
| 4 | 4 | "testing" | |
| 5 | + | "time" | |
| 5 | 6 | ||
| 6 | 7 | "github.com/bnema/zut/packages/agent/subagents" | |
| 7 | 8 | "github.com/google/uuid" | |
| @@ -17,6 +18,22 @@ func TestParseArgsSubagentAndReasoning(t *testing.T) { | |||
| 17 | 18 | } | |
| 18 | 19 | } | |
| 19 | 20 | ||
| 21 | + | func TestParseArgsSubagentTurnTimeout(t *testing.T) { | |
| 22 | + | args, err := ParseArgs([]string{"--subagent-worker", "/tmp/in.sock", "--subagent-turn-timeout", "15m"}) | |
| 23 | + | if err != nil { | |
| 24 | + | t.Fatal(err) | |
| 25 | + | } | |
| 26 | + | if args.SubagentTurnTimeout != 15*time.Minute { | |
| 27 | + | t.Fatalf("turn timeout = %s, want %s", args.SubagentTurnTimeout, 15*time.Minute) | |
| 28 | + | } | |
| 29 | + | ||
| 30 | + | for _, value := range []string{"0s", "-1s", "invalid"} { | |
| 31 | + | if _, err := ParseArgs([]string{"--subagent-worker", "/tmp/in.sock", "--subagent-turn-timeout", value}); err == nil { | |
| 32 | + | t.Errorf("timeout %q unexpectedly accepted", value) | |
| 33 | + | } | |
| 34 | + | } | |
| 35 | + | } | |
| 36 | + | ||
| 20 | 37 | func TestParseArgsNoLSP(t *testing.T) { | |
| 21 | 38 | args, err := ParseArgs([]string{"--no-lsp"}) | |
| 22 | 39 | if err != nil { | |
| @@ -2,6 +2,7 @@ package agent | |||||
| 2 | 2 | ||||
| 3 | import ( | 3 | import ( | ||
| 4 | "testing" | 4 | "testing" | ||
| 5 | + | "time" | |||
| 5 | 6 | ||||
| 6 | "github.com/bnema/zut/packages/agent/subagents" | 7 | "github.com/bnema/zut/packages/agent/subagents" | ||
| 7 | "github.com/google/uuid" | 8 | "github.com/google/uuid" | ||
| @@ -17,6 +18,22 @@ func TestParseArgsSubagentAndReasoning(t *testing.T) { | |||||
| 17 | } | 18 | } | ||
| 18 | } | 19 | } | ||
| 19 | 20 | ||||
| 21 | + | func TestParseArgsSubagentTurnTimeout(t *testing.T) { | |||
| 22 | + | args, err := ParseArgs([]string{"--subagent-worker", "/tmp/in.sock", "--subagent-turn-timeout", "15m"}) | |||
| 23 | + | if err != nil { | |||
| 24 | + | t.Fatal(err) | |||
| 25 | + | } | |||
| 26 | + | if args.SubagentTurnTimeout != 15*time.Minute { | |||
| 27 | + | t.Fatalf("turn timeout = %s, want %s", args.SubagentTurnTimeout, 15*time.Minute) | |||
| 28 | + | } | |||
| 29 | + | ||||
| 30 | + | for _, value := range []string{"0s", "-1s", "invalid"} { | |||
| 31 | + | if _, err := ParseArgs([]string{"--subagent-worker", "/tmp/in.sock", "--subagent-turn-timeout", value}); err == nil { | |||
| 32 | + | t.Errorf("timeout %q unexpectedly accepted", value) | |||
| 33 | + | } | |||
| 34 | + | } | |||
| 35 | + | } | |||
| 36 | + | ||||
| 20 | func TestParseArgsNoLSP(t *testing.T) { | 37 | func TestParseArgsNoLSP(t *testing.T) { | ||
| 21 | args, err := ParseArgs([]string{"--no-lsp"}) | 38 | args, err := ParseArgs([]string{"--no-lsp"}) | ||
| 22 | if err != nil { | 39 | if err != nil { | ||
M packages/agent/subagent_worker.go +11 -1
| @@ -86,6 +86,13 @@ func (b *workerTurnBudget) start(maxTurns int, newRun bool) (step, lifetime, cur | |||
| 86 | 86 | return b.sequence, b.lifetime, b.current, true | |
| 87 | 87 | } | |
| 88 | 88 | ||
| 89 | + | func subagentTurnContext(parent context.Context, timeout time.Duration) (context.Context, context.CancelFunc) { | |
| 90 | + | if timeout > 0 { | |
| 91 | + | return context.WithTimeout(parent, timeout) | |
| 92 | + | } | |
| 93 | + | return context.WithCancel(parent) | |
| 94 | + | } | |
| 95 | + | ||
| 89 | 96 | func runSubagentWorkerMode(ctx context.Context, args Args, version string) (runErr error) { | |
| 90 | 97 | if args.SubagentWorker == "" { | |
| 91 | 98 | return fmt.Errorf("--subagent-worker requires a socket path") | |
| @@ -277,7 +284,7 @@ func runSubagentWorkerMode(ctx context.Context, args Args, version string) (runE | |||
| 277 | 284 | return | |
| 278 | 285 | } | |
| 279 | 286 | turnID := fmt.Sprintf("turn-%d", step) | |
| 280 | - | c, cancel := context.WithCancel(ctx) | |
| 287 | + | c, cancel := subagentTurnContext(ctx, args.SubagentTurnTimeout) | |
| 281 | 288 | cancelFn = cancel | |
| 282 | 289 | mu.Unlock() | |
| 283 | 290 | ||
| @@ -725,6 +732,9 @@ func resultErrorPayload(err error) map[string]any { | |||
| 725 | 732 | if provider.IsContextOverflowError(err) { | |
| 726 | 733 | return map[string]any{"code": "context_limit", "message": subagentContextLimitMessage} | |
| 727 | 734 | } | |
| 735 | + | if errors.Is(err, context.DeadlineExceeded) { | |
| 736 | + | return map[string]any{"code": "deadline_exceeded", "message": "subagent turn deadline exceeded; partial output is preserved in the result and history"} | |
| 737 | + | } | |
| 728 | 738 | return map[string]any{"code": "turn_failed", "message": truncateForLog(err.Error(), 500)} | |
| 729 | 739 | } | |
| 730 | 740 | ||
| @@ -86,6 +86,13 @@ func (b *workerTurnBudget) start(maxTurns int, newRun bool) (step, lifetime, cur | |||||
| 86 | return b.sequence, b.lifetime, b.current, true | 86 | return b.sequence, b.lifetime, b.current, true | ||
| 87 | } | 87 | } | ||
| 88 | 88 | ||||
| 89 | + | func subagentTurnContext(parent context.Context, timeout time.Duration) (context.Context, context.CancelFunc) { | |||
| 90 | + | if timeout > 0 { | |||
| 91 | + | return context.WithTimeout(parent, timeout) | |||
| 92 | + | } | |||
| 93 | + | return context.WithCancel(parent) | |||
| 94 | + | } | |||
| 95 | + | ||||
| 89 | func runSubagentWorkerMode(ctx context.Context, args Args, version string) (runErr error) { | 96 | func runSubagentWorkerMode(ctx context.Context, args Args, version string) (runErr error) { | ||
| 90 | if args.SubagentWorker == "" { | 97 | if args.SubagentWorker == "" { | ||
| 91 | return fmt.Errorf("--subagent-worker requires a socket path") | 98 | return fmt.Errorf("--subagent-worker requires a socket path") | ||
| @@ -277,7 +284,7 @@ func runSubagentWorkerMode(ctx context.Context, args Args, version string) (runE | |||||
| 277 | return | 284 | return | ||
| 278 | } | 285 | } | ||
| 279 | turnID := fmt.Sprintf("turn-%d", step) | 286 | turnID := fmt.Sprintf("turn-%d", step) | ||
| 280 | - | c, cancel := context.WithCancel(ctx) | 287 | + | c, cancel := subagentTurnContext(ctx, args.SubagentTurnTimeout) |
| 281 | cancelFn = cancel | 288 | cancelFn = cancel | ||
| 282 | mu.Unlock() | 289 | mu.Unlock() | ||
| 283 | 290 | ||||
| @@ -725,6 +732,9 @@ func resultErrorPayload(err error) map[string]any { | |||||
| 725 | if provider.IsContextOverflowError(err) { | 732 | if provider.IsContextOverflowError(err) { | ||
| 726 | return map[string]any{"code": "context_limit", "message": subagentContextLimitMessage} | 733 | return map[string]any{"code": "context_limit", "message": subagentContextLimitMessage} | ||
| 727 | } | 734 | } | ||
| 735 | + | if errors.Is(err, context.DeadlineExceeded) { | |||
| 736 | + | return map[string]any{"code": "deadline_exceeded", "message": "subagent turn deadline exceeded; partial output is preserved in the result and history"} | |||
| 737 | + | } | |||
| 728 | return map[string]any{"code": "turn_failed", "message": truncateForLog(err.Error(), 500)} | 738 | return map[string]any{"code": "turn_failed", "message": truncateForLog(err.Error(), 500)} | ||
| 729 | } | 739 | } | ||
| 730 | 740 | ||||
M packages/agent/subagent_worker_test.go +40 -0
| @@ -12,6 +12,7 @@ import ( | |||
| 12 | 12 | "strings" | |
| 13 | 13 | "sync" | |
| 14 | 14 | "testing" | |
| 15 | + | "time" | |
| 15 | 16 | ||
| 16 | 17 | "github.com/bnema/zut/packages/agent/modes" | |
| 17 | 18 | "github.com/bnema/zut/packages/agent/subagents" | |
| @@ -19,6 +20,45 @@ import ( | |||
| 19 | 20 | "github.com/bnema/zut/packages/provider" | |
| 20 | 21 | ) | |
| 21 | 22 | ||
| 23 | + | func TestSubagentTurnContextResetsDeadlineAndHonorsShutdown(t *testing.T) { | |
| 24 | + | expired, stopExpired := subagentTurnContext(context.Background(), time.Nanosecond) | |
| 25 | + | defer stopExpired() | |
| 26 | + | <-expired.Done() | |
| 27 | + | if !errors.Is(expired.Err(), context.DeadlineExceeded) { | |
| 28 | + | t.Fatalf("expired turn error = %v, want context deadline exceeded", expired.Err()) | |
| 29 | + | } | |
| 30 | + | ||
| 31 | + | parent, stopParent := context.WithCancel(context.Background()) | |
| 32 | + | first, stopFirst := subagentTurnContext(parent, time.Hour) | |
| 33 | + | if _, ok := first.Deadline(); !ok { | |
| 34 | + | t.Fatal("active turn context has no deadline") | |
| 35 | + | } | |
| 36 | + | stopFirst() | |
| 37 | + | if !errors.Is(first.Err(), context.Canceled) { | |
| 38 | + | t.Fatalf("first turn error = %v, want context canceled", first.Err()) | |
| 39 | + | } | |
| 40 | + | ||
| 41 | + | second, stopSecond := subagentTurnContext(parent, time.Hour) | |
| 42 | + | defer stopSecond() | |
| 43 | + | select { | |
| 44 | + | case <-second.Done(): | |
| 45 | + | t.Fatalf("new turn inherited previous turn cancellation: %v", second.Err()) | |
| 46 | + | default: | |
| 47 | + | } | |
| 48 | + | stopParent() | |
| 49 | + | <-second.Done() | |
| 50 | + | if !errors.Is(second.Err(), context.Canceled) { | |
| 51 | + | t.Fatalf("shutdown error = %v, want context canceled", second.Err()) | |
| 52 | + | } | |
| 53 | + | } | |
| 54 | + | ||
| 55 | + | func TestResultErrorPayloadClassifiesTurnDeadline(t *testing.T) { | |
| 56 | + | payload := resultErrorPayload(context.DeadlineExceeded) | |
| 57 | + | if payload["code"] != "deadline_exceeded" { | |
| 58 | + | t.Fatalf("deadline error code = %v, want deadline_exceeded", payload["code"]) | |
| 59 | + | } | |
| 60 | + | } | |
| 61 | + | ||
| 22 | 62 | func TestShutdownExitPayloadIncludesOnlyKnownOrigin(t *testing.T) { | |
| 23 | 63 | payload := shutdownExitPayload(subagents.ShutdownOriginSession) | |
| 24 | 64 | if payload["reason"] != "shutdown" || payload["origin"] != subagents.ShutdownOriginSession { | |
| @@ -12,6 +12,7 @@ import ( | |||||
| 12 | "strings" | 12 | "strings" | ||
| 13 | "sync" | 13 | "sync" | ||
| 14 | "testing" | 14 | "testing" | ||
| 15 | + | "time" | |||
| 15 | 16 | ||||
| 16 | "github.com/bnema/zut/packages/agent/modes" | 17 | "github.com/bnema/zut/packages/agent/modes" | ||
| 17 | "github.com/bnema/zut/packages/agent/subagents" | 18 | "github.com/bnema/zut/packages/agent/subagents" | ||
| @@ -19,6 +20,45 @@ import ( | |||||
| 19 | "github.com/bnema/zut/packages/provider" | 20 | "github.com/bnema/zut/packages/provider" | ||
| 20 | ) | 21 | ) | ||
| 21 | 22 | ||||
| 23 | + | func TestSubagentTurnContextResetsDeadlineAndHonorsShutdown(t *testing.T) { | |||
| 24 | + | expired, stopExpired := subagentTurnContext(context.Background(), time.Nanosecond) | |||
| 25 | + | defer stopExpired() | |||
| 26 | + | <-expired.Done() | |||
| 27 | + | if !errors.Is(expired.Err(), context.DeadlineExceeded) { | |||
| 28 | + | t.Fatalf("expired turn error = %v, want context deadline exceeded", expired.Err()) | |||
| 29 | + | } | |||
| 30 | + | ||||
| 31 | + | parent, stopParent := context.WithCancel(context.Background()) | |||
| 32 | + | first, stopFirst := subagentTurnContext(parent, time.Hour) | |||
| 33 | + | if _, ok := first.Deadline(); !ok { | |||
| 34 | + | t.Fatal("active turn context has no deadline") | |||
| 35 | + | } | |||
| 36 | + | stopFirst() | |||
| 37 | + | if !errors.Is(first.Err(), context.Canceled) { | |||
| 38 | + | t.Fatalf("first turn error = %v, want context canceled", first.Err()) | |||
| 39 | + | } | |||
| 40 | + | ||||
| 41 | + | second, stopSecond := subagentTurnContext(parent, time.Hour) | |||
| 42 | + | defer stopSecond() | |||
| 43 | + | select { | |||
| 44 | + | case <-second.Done(): | |||
| 45 | + | t.Fatalf("new turn inherited previous turn cancellation: %v", second.Err()) | |||
| 46 | + | default: | |||
| 47 | + | } | |||
| 48 | + | stopParent() | |||
| 49 | + | <-second.Done() | |||
| 50 | + | if !errors.Is(second.Err(), context.Canceled) { | |||
| 51 | + | t.Fatalf("shutdown error = %v, want context canceled", second.Err()) | |||
| 52 | + | } | |||
| 53 | + | } | |||
| 54 | + | ||||
| 55 | + | func TestResultErrorPayloadClassifiesTurnDeadline(t *testing.T) { | |||
| 56 | + | payload := resultErrorPayload(context.DeadlineExceeded) | |||
| 57 | + | if payload["code"] != "deadline_exceeded" { | |||
| 58 | + | t.Fatalf("deadline error code = %v, want deadline_exceeded", payload["code"]) | |||
| 59 | + | } | |||
| 60 | + | } | |||
| 61 | + | ||||
| 22 | func TestShutdownExitPayloadIncludesOnlyKnownOrigin(t *testing.T) { | 62 | func TestShutdownExitPayloadIncludesOnlyKnownOrigin(t *testing.T) { | ||
| 23 | payload := shutdownExitPayload(subagents.ShutdownOriginSession) | 63 | payload := shutdownExitPayload(subagents.ShutdownOriginSession) | ||
| 24 | if payload["reason"] != "shutdown" || payload["origin"] != subagents.ShutdownOriginSession { | 64 | if payload["reason"] != "shutdown" || payload["origin"] != subagents.ShutdownOriginSession { | ||
M packages/agent/subagents/persist.go +1 -1
| @@ -1159,7 +1159,7 @@ func (f *Supervisor) resume(ctx context.Context, id string, resuming bool, resum | |||
| 1159 | 1159 | if timeout <= 0 { | |
| 1160 | 1160 | timeout = f.cfg.Policy.DefaultTimeout | |
| 1161 | 1161 | } | |
| 1162 | - | runCtx, cancel := f.workerContext(timeout) | |
| 1162 | + | runCtx, cancel := f.workerContext() | |
| 1163 | 1163 | a := &Agent{ | |
| 1164 | 1164 | ID: m.ID, | |
| 1165 | 1165 | Task: m.Task, | |
| @@ -1159,7 +1159,7 @@ func (f *Supervisor) resume(ctx context.Context, id string, resuming bool, resum | |||||
| 1159 | if timeout <= 0 { | 1159 | if timeout <= 0 { | ||
| 1160 | timeout = f.cfg.Policy.DefaultTimeout | 1160 | timeout = f.cfg.Policy.DefaultTimeout | ||
| 1161 | } | 1161 | } | ||
| 1162 | - | runCtx, cancel := f.workerContext(timeout) | 1162 | + | runCtx, cancel := f.workerContext() |
| 1163 | a := &Agent{ | 1163 | a := &Agent{ | ||
| 1164 | ID: m.ID, | 1164 | ID: m.ID, | ||
| 1165 | Task: m.Task, | 1165 | Task: m.Task, | ||
M packages/agent/subagents/runner.go +5 -0
| @@ -80,6 +80,7 @@ type subagentWorkerArgsOpts struct { | |||
| 80 | 80 | FastModeSet bool | |
| 81 | 81 | Subagent string | |
| 82 | 82 | MaxTurns int | |
| 83 | + | TurnTimeout time.Duration | |
| 83 | 84 | LifetimeTurns int | |
| 84 | 85 | RunTurns int | |
| 85 | 86 | CountersSet bool | |
| @@ -118,6 +119,7 @@ func defaultChildArgs(exe string, a *Agent, sessionPath, inboxPath string) []str | |||
| 118 | 119 | FastModeSet: true, | |
| 119 | 120 | Subagent: a.Subagent, | |
| 120 | 121 | MaxTurns: a.MaxTurns, | |
| 122 | + | TurnTimeout: a.Timeout, | |
| 121 | 123 | LifetimeTurns: a.LifetimeTurnsValue(), | |
| 122 | 124 | RunTurns: a.CurrentRunTurnsValue(), | |
| 123 | 125 | CountersSet: true, | |
| @@ -164,6 +166,9 @@ func subagentWorkerArgs(opts subagentWorkerArgsOpts) []string { | |||
| 164 | 166 | if opts.MaxTurns > 0 { | |
| 165 | 167 | args = append(args, "--max-turns", fmt.Sprint(opts.MaxTurns)) | |
| 166 | 168 | } | |
| 169 | + | if opts.TurnTimeout > 0 { | |
| 170 | + | args = append(args, "--subagent-turn-timeout", opts.TurnTimeout.String()) | |
| 171 | + | } | |
| 167 | 172 | if opts.CountersSet { | |
| 168 | 173 | args = append(args, | |
| 169 | 174 | "--subagent-lifetime-turns", fmt.Sprint(opts.LifetimeTurns), | |
| @@ -80,6 +80,7 @@ type subagentWorkerArgsOpts struct { | |||||
| 80 | FastModeSet bool | 80 | FastModeSet bool | ||
| 81 | Subagent string | 81 | Subagent string | ||
| 82 | MaxTurns int | 82 | MaxTurns int | ||
| 83 | + | TurnTimeout time.Duration | |||
| 83 | LifetimeTurns int | 84 | LifetimeTurns int | ||
| 84 | RunTurns int | 85 | RunTurns int | ||
| 85 | CountersSet bool | 86 | CountersSet bool | ||
| @@ -118,6 +119,7 @@ func defaultChildArgs(exe string, a *Agent, sessionPath, inboxPath string) []str | |||||
| 118 | FastModeSet: true, | 119 | FastModeSet: true, | ||
| 119 | Subagent: a.Subagent, | 120 | Subagent: a.Subagent, | ||
| 120 | MaxTurns: a.MaxTurns, | 121 | MaxTurns: a.MaxTurns, | ||
| 122 | + | TurnTimeout: a.Timeout, | |||
| 121 | LifetimeTurns: a.LifetimeTurnsValue(), | 123 | LifetimeTurns: a.LifetimeTurnsValue(), | ||
| 122 | RunTurns: a.CurrentRunTurnsValue(), | 124 | RunTurns: a.CurrentRunTurnsValue(), | ||
| 123 | CountersSet: true, | 125 | CountersSet: true, | ||
| @@ -164,6 +166,9 @@ func subagentWorkerArgs(opts subagentWorkerArgsOpts) []string { | |||||
| 164 | if opts.MaxTurns > 0 { | 166 | if opts.MaxTurns > 0 { | ||
| 165 | args = append(args, "--max-turns", fmt.Sprint(opts.MaxTurns)) | 167 | args = append(args, "--max-turns", fmt.Sprint(opts.MaxTurns)) | ||
| 166 | } | 168 | } | ||
| 169 | + | if opts.TurnTimeout > 0 { | |||
| 170 | + | args = append(args, "--subagent-turn-timeout", opts.TurnTimeout.String()) | |||
| 171 | + | } | |||
| 167 | if opts.CountersSet { | 172 | if opts.CountersSet { | ||
| 168 | args = append(args, | 173 | args = append(args, | ||
| 169 | "--subagent-lifetime-turns", fmt.Sprint(opts.LifetimeTurns), | 174 | "--subagent-lifetime-turns", fmt.Sprint(opts.LifetimeTurns), | ||
M packages/agent/subagents/runner_test.go +11 -5
| @@ -301,6 +301,7 @@ func TestSubagentWorkerArgs(t *testing.T) { | |||
| 301 | 301 | Task: "do the thing", | |
| 302 | 302 | Reasoning: "high", | |
| 303 | 303 | Subagent: "reviewer", | |
| 304 | + | TurnTimeout: 15 * time.Minute, | |
| 304 | 305 | }) | |
| 305 | 306 | if len(args) < 7 { | |
| 306 | 307 | t.Fatalf("argv unexpectedly short: %v", args) | |
| @@ -315,11 +316,12 @@ func TestSubagentWorkerArgs(t *testing.T) { | |||
| 315 | 316 | } | |
| 316 | 317 | ||
| 317 | 318 | mustHave := map[string]string{ | |
| 318 | - | "--subagent-worker": "/tmp/state/in.sock", | |
| 319 | - | "--session": "/tmp/state/session.json", | |
| 320 | - | "--cwd": "/tmp/worktree", | |
| 321 | - | "--reasoning": "high", | |
| 322 | - | "--subagent": "reviewer", | |
| 319 | + | "--subagent-worker": "/tmp/state/in.sock", | |
| 320 | + | "--session": "/tmp/state/session.json", | |
| 321 | + | "--cwd": "/tmp/worktree", | |
| 322 | + | "--reasoning": "high", | |
| 323 | + | "--subagent": "reviewer", | |
| 324 | + | "--subagent-turn-timeout": "15m0s", | |
| 323 | 325 | } | |
| 324 | 326 | for flag, value := range mustHave { | |
| 325 | 327 | i := indexOf(args, flag) | |
| @@ -444,6 +446,7 @@ func TestDefaultChildArgsPropagatesProviderConnectionSettings(t *testing.T) { | |||
| 444 | 446 | Dir: "/work", | |
| 445 | 447 | BaseURL: "https://gateway.example.test/v1", | |
| 446 | 448 | InsecureTLS: true, | |
| 449 | + | Timeout: 15 * time.Minute, | |
| 447 | 450 | } | |
| 448 | 451 | args := defaultChildArgs("/zut", a, "/state/session.json", "/state/inbox.sock") | |
| 449 | 452 | if i := indexOf(args, "--base-url"); i < 0 || safeAt(args, i+1) != a.BaseURL { | |
| @@ -452,6 +455,9 @@ func TestDefaultChildArgsPropagatesProviderConnectionSettings(t *testing.T) { | |||
| 452 | 455 | if !containsArg(args, "--insecure") { | |
| 453 | 456 | t.Fatalf("argv = %v, want Agent.InsecureTLS propagated", args) | |
| 454 | 457 | } | |
| 458 | + | if i := indexOf(args, "--subagent-turn-timeout"); i < 0 || safeAt(args, i+1) != a.Timeout.String() { | |
| 459 | + | t.Fatalf("argv = %v, want Agent.Timeout propagated per turn", args) | |
| 460 | + | } | |
| 455 | 461 | } | |
| 456 | 462 | ||
| 457 | 463 | // TestSubagentWorkerArgsEmptyTaskOmitsPositional makes sure that when the | |
| @@ -301,6 +301,7 @@ func TestSubagentWorkerArgs(t *testing.T) { | |||||
| 301 | Task: "do the thing", | 301 | Task: "do the thing", | ||
| 302 | Reasoning: "high", | 302 | Reasoning: "high", | ||
| 303 | Subagent: "reviewer", | 303 | Subagent: "reviewer", | ||
| 304 | + | TurnTimeout: 15 * time.Minute, | |||
| 304 | }) | 305 | }) | ||
| 305 | if len(args) < 7 { | 306 | if len(args) < 7 { | ||
| 306 | t.Fatalf("argv unexpectedly short: %v", args) | 307 | t.Fatalf("argv unexpectedly short: %v", args) | ||
| @@ -315,11 +316,12 @@ func TestSubagentWorkerArgs(t *testing.T) { | |||||
| 315 | } | 316 | } | ||
| 316 | 317 | ||||
| 317 | mustHave := map[string]string{ | 318 | mustHave := map[string]string{ | ||
| 318 | - | "--subagent-worker": "/tmp/state/in.sock", | 319 | + | "--subagent-worker": "/tmp/state/in.sock", |
| 319 | - | "--session": "/tmp/state/session.json", | 320 | + | "--session": "/tmp/state/session.json", |
| 320 | - | "--cwd": "/tmp/worktree", | 321 | + | "--cwd": "/tmp/worktree", |
| 321 | - | "--reasoning": "high", | 322 | + | "--reasoning": "high", |
| 322 | - | "--subagent": "reviewer", | 323 | + | "--subagent": "reviewer", |
| 324 | + | "--subagent-turn-timeout": "15m0s", | |||
| 323 | } | 325 | } | ||
| 324 | for flag, value := range mustHave { | 326 | for flag, value := range mustHave { | ||
| 325 | i := indexOf(args, flag) | 327 | i := indexOf(args, flag) | ||
| @@ -444,6 +446,7 @@ func TestDefaultChildArgsPropagatesProviderConnectionSettings(t *testing.T) { | |||||
| 444 | Dir: "/work", | 446 | Dir: "/work", | ||
| 445 | BaseURL: "https://gateway.example.test/v1", | 447 | BaseURL: "https://gateway.example.test/v1", | ||
| 446 | InsecureTLS: true, | 448 | InsecureTLS: true, | ||
| 449 | + | Timeout: 15 * time.Minute, | |||
| 447 | } | 450 | } | ||
| 448 | args := defaultChildArgs("/zut", a, "/state/session.json", "/state/inbox.sock") | 451 | args := defaultChildArgs("/zut", a, "/state/session.json", "/state/inbox.sock") | ||
| 449 | if i := indexOf(args, "--base-url"); i < 0 || safeAt(args, i+1) != a.BaseURL { | 452 | if i := indexOf(args, "--base-url"); i < 0 || safeAt(args, i+1) != a.BaseURL { | ||
| @@ -452,6 +455,9 @@ func TestDefaultChildArgsPropagatesProviderConnectionSettings(t *testing.T) { | |||||
| 452 | if !containsArg(args, "--insecure") { | 455 | if !containsArg(args, "--insecure") { | ||
| 453 | t.Fatalf("argv = %v, want Agent.InsecureTLS propagated", args) | 456 | t.Fatalf("argv = %v, want Agent.InsecureTLS propagated", args) | ||
| 454 | } | 457 | } | ||
| 458 | + | if i := indexOf(args, "--subagent-turn-timeout"); i < 0 || safeAt(args, i+1) != a.Timeout.String() { | |||
| 459 | + | t.Fatalf("argv = %v, want Agent.Timeout propagated per turn", args) | |||
| 460 | + | } | |||
| 455 | } | 461 | } | ||
| 456 | 462 | ||||
| 457 | // TestSubagentWorkerArgsEmptyTaskOmitsPositional makes sure that when the | 463 | // TestSubagentWorkerArgsEmptyTaskOmitsPositional makes sure that when the | ||
M packages/agent/subagents/supervisor.go +4 -7
| @@ -180,16 +180,13 @@ func New(cfg Config) *Supervisor { | |||
| 180 | 180 | } | |
| 181 | 181 | ||
| 182 | 182 | // workerContext derives a child worker context from the supervisor lifetime. | |
| 183 | - | // A positive timeout limits the worker lifetime; otherwise it is cancellable | |
| 184 | - | // only when the supervisor shuts down or the agent is stopped. | |
| 185 | - | func (f *Supervisor) workerContext(timeout time.Duration) (context.Context, context.CancelFunc) { | |
| 183 | + | // Turn timeouts are enforced by the worker while a delegated turn is active; | |
| 184 | + | // the process remains cancellable while idle so it can accept follow-ups. | |
| 185 | + | func (f *Supervisor) workerContext() (context.Context, context.CancelFunc) { | |
| 186 | 186 | base := f.lifetimeCtx | |
| 187 | 187 | if base == nil { | |
| 188 | 188 | base = context.Background() | |
| 189 | 189 | } | |
| 190 | - | if timeout > 0 { | |
| 191 | - | return context.WithTimeout(base, timeout) | |
| 192 | - | } | |
| 193 | 190 | return context.WithCancel(base) | |
| 194 | 191 | } | |
| 195 | 192 | ||
| @@ -489,7 +486,7 @@ func (f *Supervisor) SpawnReq(ctx context.Context, req SpawnRequest) (*Agent, er | |||
| 489 | 486 | if timeout <= 0 { | |
| 490 | 487 | timeout = f.cfg.Policy.DefaultTimeout | |
| 491 | 488 | } | |
| 492 | - | runCtx, cancel := f.workerContext(timeout) | |
| 489 | + | runCtx, cancel := f.workerContext() | |
| 493 | 490 | tools := append([]string(nil), req.Tools...) | |
| 494 | 491 | if len(tools) == 0 && len(f.cfg.Policy.AllowedTools) > 0 { | |
| 495 | 492 | tools = append([]string(nil), f.cfg.Policy.AllowedTools...) | |
| @@ -180,16 +180,13 @@ func New(cfg Config) *Supervisor { | |||||
| 180 | } | 180 | } | ||
| 181 | 181 | ||||
| 182 | // workerContext derives a child worker context from the supervisor lifetime. | 182 | // workerContext derives a child worker context from the supervisor lifetime. | ||
| 183 | - | // A positive timeout limits the worker lifetime; otherwise it is cancellable | 183 | + | // Turn timeouts are enforced by the worker while a delegated turn is active; |
| 184 | - | // only when the supervisor shuts down or the agent is stopped. | 184 | + | // the process remains cancellable while idle so it can accept follow-ups. |
| 185 | - | func (f *Supervisor) workerContext(timeout time.Duration) (context.Context, context.CancelFunc) { | 185 | + | func (f *Supervisor) workerContext() (context.Context, context.CancelFunc) { |
| 186 | base := f.lifetimeCtx | 186 | base := f.lifetimeCtx | ||
| 187 | if base == nil { | 187 | if base == nil { | ||
| 188 | base = context.Background() | 188 | base = context.Background() | ||
| 189 | } | 189 | } | ||
| 190 | - | if timeout > 0 { | |||
| 191 | - | return context.WithTimeout(base, timeout) | |||
| 192 | - | } | |||
| 193 | return context.WithCancel(base) | 190 | return context.WithCancel(base) | ||
| 194 | } | 191 | } | ||
| 195 | 192 | ||||
| @@ -489,7 +486,7 @@ func (f *Supervisor) SpawnReq(ctx context.Context, req SpawnRequest) (*Agent, er | |||||
| 489 | if timeout <= 0 { | 486 | if timeout <= 0 { | ||
| 490 | timeout = f.cfg.Policy.DefaultTimeout | 487 | timeout = f.cfg.Policy.DefaultTimeout | ||
| 491 | } | 488 | } | ||
| 492 | - | runCtx, cancel := f.workerContext(timeout) | 489 | + | runCtx, cancel := f.workerContext() |
| 493 | tools := append([]string(nil), req.Tools...) | 490 | tools := append([]string(nil), req.Tools...) | ||
| 494 | if len(tools) == 0 && len(f.cfg.Policy.AllowedTools) > 0 { | 491 | if len(tools) == 0 && len(f.cfg.Policy.AllowedTools) > 0 { | ||
| 495 | tools = append([]string(nil), f.cfg.Policy.AllowedTools...) | 492 | tools = append([]string(nil), f.cfg.Policy.AllowedTools...) | ||
M packages/agent/subagents/supervisor_test.go +27 -0
| @@ -45,6 +45,33 @@ func TestUnqualifiedModelID(t *testing.T) { | |||
| 45 | 45 | } | |
| 46 | 46 | } | |
| 47 | 47 | ||
| 48 | + | func TestSpawnTimeoutDoesNotLimitResumableWorkerLifetime(t *testing.T) { | |
| 49 | + | observedDeadline := make(chan bool, 1) | |
| 50 | + | f := newTestSupervisor(t, func(a *Agent) Runner { | |
| 51 | + | return RunnerFunc(func(ctx context.Context, _ Sink) error { | |
| 52 | + | _, hasDeadline := ctx.Deadline() | |
| 53 | + | observedDeadline <- hasDeadline | |
| 54 | + | <-ctx.Done() | |
| 55 | + | return ctx.Err() | |
| 56 | + | }) | |
| 57 | + | }) | |
| 58 | + | a, err := f.SpawnReq(context.Background(), SpawnRequest{ | |
| 59 | + | Task: "remain resumable", | |
| 60 | + | Timeout: time.Minute, | |
| 61 | + | }) | |
| 62 | + | if err != nil { | |
| 63 | + | t.Fatal(err) | |
| 64 | + | } | |
| 65 | + | if hasDeadline := <-observedDeadline; hasDeadline { | |
| 66 | + | t.Fatal("worker lifetime context has a deadline; turn timeout must not expire an idle resumable worker") | |
| 67 | + | } | |
| 68 | + | if a.Timeout != time.Minute { | |
| 69 | + | t.Fatalf("agent timeout = %s, want %s", a.Timeout, time.Minute) | |
| 70 | + | } | |
| 71 | + | f.StopAll() | |
| 72 | + | a.Wait() | |
| 73 | + | } | |
| 74 | + | ||
| 48 | 75 | func TestSpawnRunsAndCompletes(t *testing.T) { | |
| 49 | 76 | ran := make(chan string, 1) | |
| 50 | 77 | f := newTestSupervisor(t, func(a *Agent) Runner { | |
| @@ -45,6 +45,33 @@ func TestUnqualifiedModelID(t *testing.T) { | |||||
| 45 | } | 45 | } | ||
| 46 | } | 46 | } | ||
| 47 | 47 | ||||
| 48 | + | func TestSpawnTimeoutDoesNotLimitResumableWorkerLifetime(t *testing.T) { | |||
| 49 | + | observedDeadline := make(chan bool, 1) | |||
| 50 | + | f := newTestSupervisor(t, func(a *Agent) Runner { | |||
| 51 | + | return RunnerFunc(func(ctx context.Context, _ Sink) error { | |||
| 52 | + | _, hasDeadline := ctx.Deadline() | |||
| 53 | + | observedDeadline <- hasDeadline | |||
| 54 | + | <-ctx.Done() | |||
| 55 | + | return ctx.Err() | |||
| 56 | + | }) | |||
| 57 | + | }) | |||
| 58 | + | a, err := f.SpawnReq(context.Background(), SpawnRequest{ | |||
| 59 | + | Task: "remain resumable", | |||
| 60 | + | Timeout: time.Minute, | |||
| 61 | + | }) | |||
| 62 | + | if err != nil { | |||
| 63 | + | t.Fatal(err) | |||
| 64 | + | } | |||
| 65 | + | if hasDeadline := <-observedDeadline; hasDeadline { | |||
| 66 | + | t.Fatal("worker lifetime context has a deadline; turn timeout must not expire an idle resumable worker") | |||
| 67 | + | } | |||
| 68 | + | if a.Timeout != time.Minute { | |||
| 69 | + | t.Fatalf("agent timeout = %s, want %s", a.Timeout, time.Minute) | |||
| 70 | + | } | |||
| 71 | + | f.StopAll() | |||
| 72 | + | a.Wait() | |||
| 73 | + | } | |||
| 74 | + | ||||
| 48 | func TestSpawnRunsAndCompletes(t *testing.T) { | 75 | func TestSpawnRunsAndCompletes(t *testing.T) { | ||
| 49 | ran := make(chan string, 1) | 76 | ran := make(chan string, 1) | ||
| 50 | f := newTestSupervisor(t, func(a *Agent) Runner { | 77 | f := newTestSupervisor(t, func(a *Agent) Runner { | ||