detached ce35aab

Merge pull request #82 from bnema/fix/issue-81-session-scoped-shutdown

fix(subagents): scope interactive shutdown by session
commit
ce35aab947e076e2bd4d8dc06e02d520837cc09d
author
bnema · · 2026-08-08 10:41 UTC
committer
GitHub · · 2026-08-08 10:41 UTC
merge of
0c5f8d5 + bbbf486 · diff against the first parent
signature
GPG UnverifiedB5690EEEBB952194 The key is unknown
9 files · +288 -40
Unified
UnifiedSplit
M packages/agent/cli.go +9 -3
@@ -2281,10 +2281,16 @@ func runInteractive(ctx context.Context, args Args, version string) (runErr erro
22812281 if sess != nil && subagentsMgr != nil {
22822282 subagentsMgr.SetActiveSession(sess.ID)
22832283 }
2284- // Best-effort shutdown on interactive exit: stop all running
2285- // agents so they don't outlive their parent zut.
2284+ // Best-effort shutdown on normal interactive exit: stop only workers
2285+ // owned by the active host session. Other session scopes can share this
2286+ // supervisor and must remain alive.
22862287 if subagentsMgr != nil {
2287- defer subagentsMgr.StopAll()
2288+ defer func() {
2289+ rootSessionID := subagentsMgr.ActiveSession()
2290+ if rootSessionID != "" {
2291+ _ = subagentsMgr.StopByRootSession(rootSessionID)
2292+ }
2293+ }()
22882294 }
22892295
22902296 var startupSkills []*skills.Skill
@@ -2281,10 +2281,16 @@ func runInteractive(ctx context.Context, args Args, version string) (runErr erro
2281 if sess != nil && subagentsMgr != nil {2281 if sess != nil && subagentsMgr != nil {
2282 subagentsMgr.SetActiveSession(sess.ID)2282 subagentsMgr.SetActiveSession(sess.ID)
2283 }2283 }
2284- // Best-effort shutdown on interactive exit: stop all running2284+ // Best-effort shutdown on normal interactive exit: stop only workers
2285- // agents so they don't outlive their parent zut.2285+ // owned by the active host session. Other session scopes can share this
2286+ // supervisor and must remain alive.
2286 if subagentsMgr != nil {2287 if subagentsMgr != nil {
2287- defer subagentsMgr.StopAll()2288+ defer func() {
2289+ rootSessionID := subagentsMgr.ActiveSession()
2290+ if rootSessionID != "" {
2291+ _ = subagentsMgr.StopByRootSession(rootSessionID)
2292+ }
2293+ }()
2288 }2294 }
2289 2295
2290 var startupSkills []*skills.Skill2296 var startupSkills []*skills.Skill
M packages/agent/subagent_worker.go +27 -10
@@ -233,12 +233,13 @@ func runSubagentWorkerMode(ctx context.Context, args Args, version string) (runE
233233 lifetime: args.SubagentLifetimeTurns,
234234 current: args.SubagentRunTurns,
235235 }
236- turnWG sync.WaitGroup
237- closing bool
238- turnPending bool
239- seenCommands = make(map[string]struct{})
240- shutdown = make(chan struct{})
241- shutdownOnce sync.Once
236+ turnWG sync.WaitGroup
237+ closing bool
238+ turnPending bool
239+ seenCommands = make(map[string]struct{})
240+ shutdown = make(chan struct{})
241+ shutdownOnce sync.Once
242+ shutdownOrigin subagents.ShutdownOrigin
242243 )
243244 maxOutputBytes, maxOutputLines := workerOutputLimits()
244245
@@ -404,11 +405,14 @@ func runSubagentWorkerMode(ctx context.Context, args Args, version string) (runE
404405 turnWG.Wait()
405406 }
406407
407- requestShutdown := func() {
408+ requestShutdown := func(origin subagents.ShutdownOrigin) {
408409 mu.Lock()
409410 closing = true
410411 mu.Unlock()
411- shutdownOnce.Do(func() { close(shutdown) })
412+ shutdownOnce.Do(func() {
413+ shutdownOrigin = origin.Sanitized()
414+ close(shutdown)
415+ })
412416 }
413417
414418 // Initial task: run before processing the inbox so the agent
@@ -434,7 +438,7 @@ func runSubagentWorkerMode(ctx context.Context, args Args, version string) (runE
434438 // alive indefinitely while the host waits for its socket to
435439 // disappear.
436440 waitForActiveTurn(true)
437- em.emit("agent.exited", map[string]any{"reason": "shutdown"})
441+ em.emit("agent.exited", shutdownExitPayload(shutdownOrigin))
438442 return nil
439443 case msg, ok := <-ln.Lines():
440444 if !ok {
@@ -449,7 +453,12 @@ func runSubagentWorkerMode(ctx context.Context, args Args, version string) (runE
449453 }
450454 switch command.Type {
451455 case subagents.CommandAgentShutdown:
452- requestShutdown()
456+ var payload subagents.AgentShutdownPayload
457+ if err := command.DecodePayload(&payload); err != nil {
458+ em.emit("error", map[string]any{"message": "invalid agent.shutdown payload"})
459+ continue
460+ }
461+ requestShutdown(payload.Origin)
453462 case subagents.CommandTurnCancel:
454463 mu.Lock()
455464 if cancelFn != nil {
@@ -656,6 +665,14 @@ func boundedResultSummary(text string) string {
656665 return boundedResultOutput(firstLine(text), 4*1024, 1)
657666 }
658667
668+func shutdownExitPayload(origin subagents.ShutdownOrigin) map[string]any {
669+ payload := map[string]any{"reason": "shutdown"}
670+ if origin = origin.Sanitized(); origin != "" {
671+ payload["origin"] = origin
672+ }
673+ return payload
674+}
675+
659676 func workerOutputLimits() (maxBytes, maxLines int) {
660677 maxBytes, maxLines = 500_000, 5_000
661678 if value, err := strconv.Atoi(strings.TrimSpace(os.Getenv("ZUT_SUBAGENT_MAX_OUTPUT_BYTES"))); err == nil && value > 0 {
@@ -233,12 +233,13 @@ func runSubagentWorkerMode(ctx context.Context, args Args, version string) (runE
233 lifetime: args.SubagentLifetimeTurns,233 lifetime: args.SubagentLifetimeTurns,
234 current: args.SubagentRunTurns,234 current: args.SubagentRunTurns,
235 }235 }
236- turnWG sync.WaitGroup236+ turnWG sync.WaitGroup
237- closing bool237+ closing bool
238- turnPending bool238+ turnPending bool
239- seenCommands = make(map[string]struct{})239+ seenCommands = make(map[string]struct{})
240- shutdown = make(chan struct{})240+ shutdown = make(chan struct{})
241- shutdownOnce sync.Once241+ shutdownOnce sync.Once
242+ shutdownOrigin subagents.ShutdownOrigin
242 )243 )
243 maxOutputBytes, maxOutputLines := workerOutputLimits()244 maxOutputBytes, maxOutputLines := workerOutputLimits()
244 245
@@ -404,11 +405,14 @@ func runSubagentWorkerMode(ctx context.Context, args Args, version string) (runE
404 turnWG.Wait()405 turnWG.Wait()
405 }406 }
406 407
407- requestShutdown := func() {408+ requestShutdown := func(origin subagents.ShutdownOrigin) {
408 mu.Lock()409 mu.Lock()
409 closing = true410 closing = true
410 mu.Unlock()411 mu.Unlock()
411- shutdownOnce.Do(func() { close(shutdown) })412+ shutdownOnce.Do(func() {
413+ shutdownOrigin = origin.Sanitized()
414+ close(shutdown)
415+ })
412 }416 }
413 417
414 // Initial task: run before processing the inbox so the agent418 // Initial task: run before processing the inbox so the agent
@@ -434,7 +438,7 @@ func runSubagentWorkerMode(ctx context.Context, args Args, version string) (runE
434 // alive indefinitely while the host waits for its socket to438 // alive indefinitely while the host waits for its socket to
435 // disappear.439 // disappear.
436 waitForActiveTurn(true)440 waitForActiveTurn(true)
437- em.emit("agent.exited", map[string]any{"reason": "shutdown"})441+ em.emit("agent.exited", shutdownExitPayload(shutdownOrigin))
438 return nil442 return nil
439 case msg, ok := <-ln.Lines():443 case msg, ok := <-ln.Lines():
440 if !ok {444 if !ok {
@@ -449,7 +453,12 @@ func runSubagentWorkerMode(ctx context.Context, args Args, version string) (runE
449 }453 }
450 switch command.Type {454 switch command.Type {
451 case subagents.CommandAgentShutdown:455 case subagents.CommandAgentShutdown:
452- requestShutdown()456+ var payload subagents.AgentShutdownPayload
457+ if err := command.DecodePayload(&payload); err != nil {
458+ em.emit("error", map[string]any{"message": "invalid agent.shutdown payload"})
459+ continue
460+ }
461+ requestShutdown(payload.Origin)
453 case subagents.CommandTurnCancel:462 case subagents.CommandTurnCancel:
454 mu.Lock()463 mu.Lock()
455 if cancelFn != nil {464 if cancelFn != nil {
@@ -656,6 +665,14 @@ func boundedResultSummary(text string) string {
656 return boundedResultOutput(firstLine(text), 4*1024, 1)665 return boundedResultOutput(firstLine(text), 4*1024, 1)
657 }666 }
658 667
668+func shutdownExitPayload(origin subagents.ShutdownOrigin) map[string]any {
669+ payload := map[string]any{"reason": "shutdown"}
670+ if origin = origin.Sanitized(); origin != "" {
671+ payload["origin"] = origin
672+ }
673+ return payload
674+}
675+
659 func workerOutputLimits() (maxBytes, maxLines int) {676 func workerOutputLimits() (maxBytes, maxLines int) {
660 maxBytes, maxLines = 500_000, 5_000677 maxBytes, maxLines = 500_000, 5_000
661 if value, err := strconv.Atoi(strings.TrimSpace(os.Getenv("ZUT_SUBAGENT_MAX_OUTPUT_BYTES"))); err == nil && value > 0 {678 if value, err := strconv.Atoi(strings.TrimSpace(os.Getenv("ZUT_SUBAGENT_MAX_OUTPUT_BYTES"))); err == nil && value > 0 {
M packages/agent/subagent_worker_test.go +11 -0
@@ -19,6 +19,17 @@ import (
1919 "github.com/bnema/zut/packages/provider"
2020 )
2121
22+func TestShutdownExitPayloadIncludesOnlyKnownOrigin(t *testing.T) {
23+ payload := shutdownExitPayload(subagents.ShutdownOriginSession)
24+ if payload["reason"] != "shutdown" || payload["origin"] != subagents.ShutdownOriginSession {
25+ t.Fatalf("known shutdown payload = %#v", payload)
26+ }
27+ payload = shutdownExitPayload(subagents.ShutdownOrigin("private detail"))
28+ if _, ok := payload["origin"]; ok {
29+ t.Fatalf("unknown shutdown origin was persisted: %#v", payload)
30+ }
31+}
32+
2233 // TestWorkerOutputLimitsUseSupervisorPolicy verifies that the child reads
2334 // the effective output caps propagated by the supervisor and retains safe
2435 // defaults for standalone workers.
@@ -19,6 +19,17 @@ import (
19 "github.com/bnema/zut/packages/provider"19 "github.com/bnema/zut/packages/provider"
20 )20 )
21 21
22+func TestShutdownExitPayloadIncludesOnlyKnownOrigin(t *testing.T) {
23+ payload := shutdownExitPayload(subagents.ShutdownOriginSession)
24+ if payload["reason"] != "shutdown" || payload["origin"] != subagents.ShutdownOriginSession {
25+ t.Fatalf("known shutdown payload = %#v", payload)
26+ }
27+ payload = shutdownExitPayload(subagents.ShutdownOrigin("private detail"))
28+ if _, ok := payload["origin"]; ok {
29+ t.Fatalf("unknown shutdown origin was persisted: %#v", payload)
30+ }
31+}
32+
22 // TestWorkerOutputLimitsUseSupervisorPolicy verifies that the child reads33 // TestWorkerOutputLimitsUseSupervisorPolicy verifies that the child reads
23 // the effective output caps propagated by the supervisor and retains safe34 // the effective output caps propagated by the supervisor and retains safe
24 // defaults for standalone workers.35 // defaults for standalone workers.
M packages/agent/subagents/protocol.go +25 -3
@@ -564,7 +564,28 @@ func cloneRaw(value json.RawMessage) json.RawMessage {
564564 // protocol without forcing a receiver to use them. Receivers can always use
565565 // Envelope.Payload to retain fields added in later versions.
566566 type AgentPingPayload struct{}
567-type AgentShutdownPayload struct{}
567+
568+type ShutdownOrigin string
569+
570+const (
571+ ShutdownOriginTargeted ShutdownOrigin = "targeted"
572+ ShutdownOriginSession ShutdownOrigin = "session"
573+ ShutdownOriginDeadline ShutdownOrigin = "deadline"
574+ ShutdownOriginProcess ShutdownOrigin = "process"
575+)
576+
577+func (o ShutdownOrigin) Sanitized() ShutdownOrigin {
578+ switch o {
579+ case ShutdownOriginTargeted, ShutdownOriginSession, ShutdownOriginDeadline, ShutdownOriginProcess:
580+ return o
581+ default:
582+ return ""
583+ }
584+}
585+
586+type AgentShutdownPayload struct {
587+ Origin ShutdownOrigin `json:"origin,omitempty"`
588+}
568589 type AgentReadyPayload struct {
569590 // Version is the worker/application version. WorkerVersion is the
570591 // unambiguous spelling for new producers.
@@ -628,6 +649,7 @@ type TurnFailedPayload struct {
628649 Error *ProtocolError `json:"error,omitempty"`
629650 }
630651 type AgentExitedPayload struct {
631- Code int `json:"code,omitempty"`
632- Reason string `json:"reason,omitempty"`
652+ Code int `json:"code,omitempty"`
653+ Reason string `json:"reason,omitempty"`
654+ Origin ShutdownOrigin `json:"origin,omitempty"`
633655 }
@@ -564,7 +564,28 @@ func cloneRaw(value json.RawMessage) json.RawMessage {
564 // protocol without forcing a receiver to use them. Receivers can always use564 // protocol without forcing a receiver to use them. Receivers can always use
565 // Envelope.Payload to retain fields added in later versions.565 // Envelope.Payload to retain fields added in later versions.
566 type AgentPingPayload struct{}566 type AgentPingPayload struct{}
567-type AgentShutdownPayload struct{}567+
568+type ShutdownOrigin string
569+
570+const (
571+ ShutdownOriginTargeted ShutdownOrigin = "targeted"
572+ ShutdownOriginSession ShutdownOrigin = "session"
573+ ShutdownOriginDeadline ShutdownOrigin = "deadline"
574+ ShutdownOriginProcess ShutdownOrigin = "process"
575+)
576+
577+func (o ShutdownOrigin) Sanitized() ShutdownOrigin {
578+ switch o {
579+ case ShutdownOriginTargeted, ShutdownOriginSession, ShutdownOriginDeadline, ShutdownOriginProcess:
580+ return o
581+ default:
582+ return ""
583+ }
584+}
585+
586+type AgentShutdownPayload struct {
587+ Origin ShutdownOrigin `json:"origin,omitempty"`
588+}
568 type AgentReadyPayload struct {589 type AgentReadyPayload struct {
569 // Version is the worker/application version. WorkerVersion is the590 // Version is the worker/application version. WorkerVersion is the
570 // unambiguous spelling for new producers.591 // unambiguous spelling for new producers.
@@ -628,6 +649,7 @@ type TurnFailedPayload struct {
628 Error *ProtocolError `json:"error,omitempty"`649 Error *ProtocolError `json:"error,omitempty"`
629 }650 }
630 type AgentExitedPayload struct {651 type AgentExitedPayload struct {
631- Code int `json:"code,omitempty"`652+ Code int `json:"code,omitempty"`
632- Reason string `json:"reason,omitempty"`653+ Reason string `json:"reason,omitempty"`
654+ Origin ShutdownOrigin `json:"origin,omitempty"`
633 }655 }
M packages/agent/subagents/protocol_test.go +18 -0
@@ -233,6 +233,24 @@ func TestProtocolEmptyAndMalformedInputs(t *testing.T) {
233233 }
234234 }
235235
236+func TestAgentShutdownPayloadRoundTripsKnownOrigin(t *testing.T) {
237+ for _, origin := range []ShutdownOrigin{
238+ ShutdownOriginTargeted,
239+ ShutdownOriginSession,
240+ ShutdownOriginDeadline,
241+ ShutdownOriginProcess,
242+ } {
243+ e := NewCommand(CommandAgentShutdown, "agent-1", "turn-1", AgentShutdownPayload{Origin: origin})
244+ var decoded AgentShutdownPayload
245+ if err := e.DecodePayload(&decoded); err != nil {
246+ t.Fatalf("decode %q shutdown payload: %v", origin, err)
247+ }
248+ if decoded.Origin != origin {
249+ t.Fatalf("shutdown origin = %q, want %q", decoded.Origin, origin)
250+ }
251+ }
252+}
253+
236254 func TestProtocolResultShape(t *testing.T) {
237255 result := TurnResultPayload{
238256 Status: "succeeded",
@@ -233,6 +233,24 @@ func TestProtocolEmptyAndMalformedInputs(t *testing.T) {
233 }233 }
234 }234 }
235 235
236+func TestAgentShutdownPayloadRoundTripsKnownOrigin(t *testing.T) {
237+ for _, origin := range []ShutdownOrigin{
238+ ShutdownOriginTargeted,
239+ ShutdownOriginSession,
240+ ShutdownOriginDeadline,
241+ ShutdownOriginProcess,
242+ } {
243+ e := NewCommand(CommandAgentShutdown, "agent-1", "turn-1", AgentShutdownPayload{Origin: origin})
244+ var decoded AgentShutdownPayload
245+ if err := e.DecodePayload(&decoded); err != nil {
246+ t.Fatalf("decode %q shutdown payload: %v", origin, err)
247+ }
248+ if decoded.Origin != origin {
249+ t.Fatalf("shutdown origin = %q, want %q", decoded.Origin, origin)
250+ }
251+ }
252+}
253+
236 func TestProtocolResultShape(t *testing.T) {254 func TestProtocolResultShape(t *testing.T) {
237 result := TurnResultPayload{255 result := TurnResultPayload{
238 Status: "succeeded",256 Status: "succeeded",
M packages/agent/subagents/runner.go +1 -1
@@ -544,7 +544,7 @@ func (r *execRunner) stopOnContextDone(ctx context.Context, cmd *exec.Cmd, runne
544544 defer cancel()
545545 if r.agent != nil && r.agent.inbox != nil {
546546 _ = r.agent.inbox.SendCommandContext(shutdownCtx, NewCommand(
547- CommandAgentShutdown, r.agent.ID, r.agent.CurrentTurnID(), AgentShutdownPayload{},
547+ CommandAgentShutdown, r.agent.ID, r.agent.CurrentTurnID(), AgentShutdownPayload{Origin: ShutdownOriginDeadline},
548548 ))
549549 }
550550
@@ -544,7 +544,7 @@ func (r *execRunner) stopOnContextDone(ctx context.Context, cmd *exec.Cmd, runne
544 defer cancel()544 defer cancel()
545 if r.agent != nil && r.agent.inbox != nil {545 if r.agent != nil && r.agent.inbox != nil {
546 _ = r.agent.inbox.SendCommandContext(shutdownCtx, NewCommand(546 _ = r.agent.inbox.SendCommandContext(shutdownCtx, NewCommand(
547- CommandAgentShutdown, r.agent.ID, r.agent.CurrentTurnID(), AgentShutdownPayload{},547+ CommandAgentShutdown, r.agent.ID, r.agent.CurrentTurnID(), AgentShutdownPayload{Origin: ShutdownOriginDeadline},
548 ))548 ))
549 }549 }
550 550
M packages/agent/subagents/runner_test.go +47 -0
@@ -5,12 +5,59 @@ import (
55 "encoding/json"
66 "fmt"
77 "os"
8+ "os/exec"
89 "path/filepath"
10+ "runtime"
911 "strings"
1012 "testing"
1113 "time"
1214 )
1315
16+func TestDeadlineShutdownCommandIdentifiesOrigin(t *testing.T) {
17+ if runtime.GOOS == "windows" {
18+ t.Skip("subagent inbox transport uses Unix-domain sockets")
19+ }
20+ path := filepath.Join(shortSocketDir(t), "deadline.sock")
21+ listener, err := Listen(path)
22+ if err != nil {
23+ t.Fatal(err)
24+ }
25+ defer listener.Close()
26+
27+ a := &Agent{ID: "deadline-agent", InboxPath: path, inbox: NewInbox(path)}
28+ defer a.inbox.Close()
29+ r := &execRunner{agent: a, GracePeriod: time.Second}
30+ ctx, cancel := context.WithDeadline(context.Background(), time.Now().Add(-time.Second))
31+ defer cancel()
32+ runnerDone := make(chan struct{})
33+ watcherDone := make(chan struct{})
34+ go func() {
35+ r.stopOnContextDone(ctx, &exec.Cmd{}, runnerDone)
36+ close(watcherDone)
37+ }()
38+
39+ var command Envelope
40+ select {
41+ case line := <-listener.Lines():
42+ command, err = ParseCommand(line)
43+ if err != nil {
44+ t.Fatal(err)
45+ }
46+ case <-time.After(time.Second):
47+ t.Fatal("deadline shutdown command was not received")
48+ }
49+ close(runnerDone)
50+ <-watcherDone
51+
52+ var payload AgentShutdownPayload
53+ if err := command.DecodePayload(&payload); err != nil {
54+ t.Fatal(err)
55+ }
56+ if payload.Origin != ShutdownOriginDeadline {
57+ t.Fatalf("shutdown origin = %q, want %q", payload.Origin, ShutdownOriginDeadline)
58+ }
59+}
60+
1461 // TestSubagentWorkerArgs locks in the exact flag set the subprocess
1562 // runner uses to start a subagent agent in daemon mode. Past
1663 // regressions in this area:
@@ -5,12 +5,59 @@ import (
5 "encoding/json"5 "encoding/json"
6 "fmt"6 "fmt"
7 "os"7 "os"
8+ "os/exec"
8 "path/filepath"9 "path/filepath"
10+ "runtime"
9 "strings"11 "strings"
10 "testing"12 "testing"
11 "time"13 "time"
12 )14 )
13 15
16+func TestDeadlineShutdownCommandIdentifiesOrigin(t *testing.T) {
17+ if runtime.GOOS == "windows" {
18+ t.Skip("subagent inbox transport uses Unix-domain sockets")
19+ }
20+ path := filepath.Join(shortSocketDir(t), "deadline.sock")
21+ listener, err := Listen(path)
22+ if err != nil {
23+ t.Fatal(err)
24+ }
25+ defer listener.Close()
26+
27+ a := &Agent{ID: "deadline-agent", InboxPath: path, inbox: NewInbox(path)}
28+ defer a.inbox.Close()
29+ r := &execRunner{agent: a, GracePeriod: time.Second}
30+ ctx, cancel := context.WithDeadline(context.Background(), time.Now().Add(-time.Second))
31+ defer cancel()
32+ runnerDone := make(chan struct{})
33+ watcherDone := make(chan struct{})
34+ go func() {
35+ r.stopOnContextDone(ctx, &exec.Cmd{}, runnerDone)
36+ close(watcherDone)
37+ }()
38+
39+ var command Envelope
40+ select {
41+ case line := <-listener.Lines():
42+ command, err = ParseCommand(line)
43+ if err != nil {
44+ t.Fatal(err)
45+ }
46+ case <-time.After(time.Second):
47+ t.Fatal("deadline shutdown command was not received")
48+ }
49+ close(runnerDone)
50+ <-watcherDone
51+
52+ var payload AgentShutdownPayload
53+ if err := command.DecodePayload(&payload); err != nil {
54+ t.Fatal(err)
55+ }
56+ if payload.Origin != ShutdownOriginDeadline {
57+ t.Fatalf("shutdown origin = %q, want %q", payload.Origin, ShutdownOriginDeadline)
58+ }
59+}
60+
14 // TestSubagentWorkerArgs locks in the exact flag set the subprocess61 // TestSubagentWorkerArgs locks in the exact flag set the subprocess
15 // runner uses to start a subagent agent in daemon mode. Past62 // runner uses to start a subagent agent in daemon mode. Past
16 // regressions in this area:63 // regressions in this area:
M packages/agent/subagents/supervisor.go +51 -23
@@ -776,17 +776,17 @@ func waitForDetachedWorker(ctx context.Context, path string, timeout time.Durati
776776 // shutdown command through the durable inbox instead of pretending no
777777 // process exists.
778778 func (f *Supervisor) Stop(id string) error {
779- return f.stop(context.Background(), id)
779+ return f.stop(context.Background(), id, ShutdownOriginTargeted)
780780 }
781781
782782 // StopContext is Stop with a context that can interrupt detached-worker
783783 // shutdown and its graceful and force-stop waits. The context is not held
784784 // under the supervisor operation lock while those waits run.
785785 func (f *Supervisor) StopContext(ctx context.Context, id string) error {
786- return f.stop(ctx, id)
786+ return f.stop(ctx, id, ShutdownOriginTargeted)
787787 }
788788
789-func (f *Supervisor) stop(ctx context.Context, id string) error {
789+func (f *Supervisor) stop(ctx context.Context, id string, origin ShutdownOrigin) error {
790790 if ctx == nil {
791791 ctx = context.Background()
792792 }
@@ -820,7 +820,7 @@ func (f *Supervisor) stop(ctx context.Context, id string) error {
820820 // The detached worker may take the full grace period to exit. Do not
821821 // serialize unrelated supervisor operations for that entire wait.
822822 f.operationMu.Unlock()
823- return f.stopDetached(ctx, a, path, inbox, pid)
823+ return f.stopDetached(ctx, a, path, inbox, pid, origin)
824824 }
825825 a.status = StatusKilled
826826 a.activity = "canceling"
@@ -870,7 +870,7 @@ func (f *Supervisor) stop(ctx context.Context, id string) error {
870870 }
871871 graceful := false
872872 if a.inbox != nil {
873- err := a.inbox.SendCommandContext(ctx, NewCommand(CommandAgentShutdown, a.ID, a.CurrentTurnID(), AgentShutdownPayload{}))
873+ err := a.inbox.SendCommandContext(ctx, NewCommand(CommandAgentShutdown, a.ID, a.CurrentTurnID(), AgentShutdownPayload{Origin: origin.Sanitized()}))
874874 if err == nil {
875875 graceful = true
876876 }
@@ -910,7 +910,7 @@ func (f *Supervisor) stop(ctx context.Context, id string) error {
910910 return nil
911911 }
912912
913-func (f *Supervisor) stopDetached(ctx context.Context, a *Agent, path string, inbox *Inbox, pid int) error {
913+func (f *Supervisor) stopDetached(ctx context.Context, a *Agent, path string, inbox *Inbox, pid int, origin ShutdownOrigin) error {
914914 if inbox == nil {
915915 return nil
916916 }
@@ -923,7 +923,7 @@ func (f *Supervisor) stopDetached(ctx context.Context, a *Agent, path string, in
923923 if err := ctx.Err(); err != nil {
924924 return err
925925 }
926- if err := inbox.SendCommandContext(ctx, NewCommand(CommandAgentShutdown, a.ID, a.CurrentTurnID(), AgentShutdownPayload{})); err != nil {
926+ if err := inbox.SendCommandContext(ctx, NewCommand(CommandAgentShutdown, a.ID, a.CurrentTurnID(), AgentShutdownPayload{Origin: origin.Sanitized()})); err != nil {
927927 if ctxErr := ctx.Err(); ctxErr != nil {
928928 return ctxErr
929929 }
@@ -976,33 +976,38 @@ func (f *Supervisor) markDetachedKilled(a *Agent) {
976976 a.closeDone()
977977 }
978978
979-// StopAll cancels every running agent. Used on shutdown.
980-func (f *Supervisor) StopAll() {
981- _ = f.stopAll(context.Background())
979+// StopByRootSession stops and waits for every agent owned by rootSessionID.
980+// It leaves the supervisor lifetime and agents from other sessions running.
981+func (f *Supervisor) StopByRootSession(rootSessionID string) error {
982+ return f.StopByRootSessionContext(context.Background(), rootSessionID)
982983 }
983984
984-// StopAllContext is the context-aware shutdown variant. It returns the
985-// context error when cancellation interrupts a detached-worker wait.
986-func (f *Supervisor) StopAllContext(ctx context.Context) error {
987- return f.stopAll(ctx)
988-}
989-
990-func (f *Supervisor) stopAll(ctx context.Context) error {
985+// StopByRootSessionContext is the context-aware form of StopByRootSession.
986+func (f *Supervisor) StopByRootSessionContext(ctx context.Context, rootSessionID string) error {
991987 if ctx == nil {
992988 ctx = context.Background()
993989 }
990+ rootSessionID = strings.TrimSpace(rootSessionID)
991+ if rootSessionID == "" {
992+ return errors.New("subagents: root session id is required")
993+ }
994994 agents := f.List()
995+ owned := make([]*Agent, 0, len(agents))
996+ for _, a := range agents {
997+ if a.RootSessionID == rootSessionID {
998+ owned = append(owned, a)
999+ }
1000+ }
1001+ return f.stopAgents(ctx, owned, ShutdownOriginSession)
1002+}
1003+
1004+func (f *Supervisor) stopAgents(ctx context.Context, agents []*Agent, origin ShutdownOrigin) error {
9951005 var firstErr error
9961006 for _, a := range agents {
997- if err := f.stop(ctx, a.ID); err != nil && firstErr == nil {
1007+ if err := f.stop(ctx, a.ID, origin); err != nil && firstErr == nil {
9981008 firstErr = err
9991009 }
10001010 }
1001- defer func() {
1002- if f.lifetimeCancel != nil {
1003- f.lifetimeCancel()
1004- }
1005- }()
10061011 for _, a := range agents {
10071012 select {
10081013 case <-a.done:
@@ -1016,6 +1021,29 @@ func (f *Supervisor) stopAll(ctx context.Context) error {
10161021 return firstErr
10171022 }
10181023
1024+// StopAll cancels every running agent. Used on shutdown.
1025+func (f *Supervisor) StopAll() {
1026+ _ = f.stopAll(context.Background())
1027+}
1028+
1029+// StopAllContext is the context-aware shutdown variant. It returns the
1030+// context error when cancellation interrupts a detached-worker wait.
1031+func (f *Supervisor) StopAllContext(ctx context.Context) error {
1032+ return f.stopAll(ctx)
1033+}
1034+
1035+func (f *Supervisor) stopAll(ctx context.Context) error {
1036+ if ctx == nil {
1037+ ctx = context.Background()
1038+ }
1039+ defer func() {
1040+ if f.lifetimeCancel != nil {
1041+ f.lifetimeCancel()
1042+ }
1043+ }()
1044+ return f.stopAgents(ctx, f.List(), ShutdownOriginProcess)
1045+}
1046+
10191047 // Remove tears down the per-agent state for a terminated agent. It
10201048 // is an error to remove an agent that's still running; call Stop
10211049 // first and wait for the status to settle. Detached agents
@@ -776,17 +776,17 @@ func waitForDetachedWorker(ctx context.Context, path string, timeout time.Durati
776 // shutdown command through the durable inbox instead of pretending no776 // shutdown command through the durable inbox instead of pretending no
777 // process exists.777 // process exists.
778 func (f *Supervisor) Stop(id string) error {778 func (f *Supervisor) Stop(id string) error {
779- return f.stop(context.Background(), id)779+ return f.stop(context.Background(), id, ShutdownOriginTargeted)
780 }780 }
781 781
782 // StopContext is Stop with a context that can interrupt detached-worker782 // StopContext is Stop with a context that can interrupt detached-worker
783 // shutdown and its graceful and force-stop waits. The context is not held783 // shutdown and its graceful and force-stop waits. The context is not held
784 // under the supervisor operation lock while those waits run.784 // under the supervisor operation lock while those waits run.
785 func (f *Supervisor) StopContext(ctx context.Context, id string) error {785 func (f *Supervisor) StopContext(ctx context.Context, id string) error {
786- return f.stop(ctx, id)786+ return f.stop(ctx, id, ShutdownOriginTargeted)
787 }787 }
788 788
789-func (f *Supervisor) stop(ctx context.Context, id string) error {789+func (f *Supervisor) stop(ctx context.Context, id string, origin ShutdownOrigin) error {
790 if ctx == nil {790 if ctx == nil {
791 ctx = context.Background()791 ctx = context.Background()
792 }792 }
@@ -820,7 +820,7 @@ func (f *Supervisor) stop(ctx context.Context, id string) error {
820 // The detached worker may take the full grace period to exit. Do not820 // The detached worker may take the full grace period to exit. Do not
821 // serialize unrelated supervisor operations for that entire wait.821 // serialize unrelated supervisor operations for that entire wait.
822 f.operationMu.Unlock()822 f.operationMu.Unlock()
823- return f.stopDetached(ctx, a, path, inbox, pid)823+ return f.stopDetached(ctx, a, path, inbox, pid, origin)
824 }824 }
825 a.status = StatusKilled825 a.status = StatusKilled
826 a.activity = "canceling"826 a.activity = "canceling"
@@ -870,7 +870,7 @@ func (f *Supervisor) stop(ctx context.Context, id string) error {
870 }870 }
871 graceful := false871 graceful := false
872 if a.inbox != nil {872 if a.inbox != nil {
873- err := a.inbox.SendCommandContext(ctx, NewCommand(CommandAgentShutdown, a.ID, a.CurrentTurnID(), AgentShutdownPayload{}))873+ err := a.inbox.SendCommandContext(ctx, NewCommand(CommandAgentShutdown, a.ID, a.CurrentTurnID(), AgentShutdownPayload{Origin: origin.Sanitized()}))
874 if err == nil {874 if err == nil {
875 graceful = true875 graceful = true
876 }876 }
@@ -910,7 +910,7 @@ func (f *Supervisor) stop(ctx context.Context, id string) error {
910 return nil910 return nil
911 }911 }
912 912
913-func (f *Supervisor) stopDetached(ctx context.Context, a *Agent, path string, inbox *Inbox, pid int) error {913+func (f *Supervisor) stopDetached(ctx context.Context, a *Agent, path string, inbox *Inbox, pid int, origin ShutdownOrigin) error {
914 if inbox == nil {914 if inbox == nil {
915 return nil915 return nil
916 }916 }
@@ -923,7 +923,7 @@ func (f *Supervisor) stopDetached(ctx context.Context, a *Agent, path string, in
923 if err := ctx.Err(); err != nil {923 if err := ctx.Err(); err != nil {
924 return err924 return err
925 }925 }
926- if err := inbox.SendCommandContext(ctx, NewCommand(CommandAgentShutdown, a.ID, a.CurrentTurnID(), AgentShutdownPayload{})); err != nil {926+ if err := inbox.SendCommandContext(ctx, NewCommand(CommandAgentShutdown, a.ID, a.CurrentTurnID(), AgentShutdownPayload{Origin: origin.Sanitized()})); err != nil {
927 if ctxErr := ctx.Err(); ctxErr != nil {927 if ctxErr := ctx.Err(); ctxErr != nil {
928 return ctxErr928 return ctxErr
929 }929 }
@@ -976,33 +976,38 @@ func (f *Supervisor) markDetachedKilled(a *Agent) {
976 a.closeDone()976 a.closeDone()
977 }977 }
978 978
979-// StopAll cancels every running agent. Used on shutdown.979+// StopByRootSession stops and waits for every agent owned by rootSessionID.
980-func (f *Supervisor) StopAll() {980+// It leaves the supervisor lifetime and agents from other sessions running.
981- _ = f.stopAll(context.Background())981+func (f *Supervisor) StopByRootSession(rootSessionID string) error {
982+ return f.StopByRootSessionContext(context.Background(), rootSessionID)
982 }983 }
983 984
984-// StopAllContext is the context-aware shutdown variant. It returns the985+// StopByRootSessionContext is the context-aware form of StopByRootSession.
985-// context error when cancellation interrupts a detached-worker wait.986+func (f *Supervisor) StopByRootSessionContext(ctx context.Context, rootSessionID string) error {
986-func (f *Supervisor) StopAllContext(ctx context.Context) error {
987- return f.stopAll(ctx)
988-}
989-
990-func (f *Supervisor) stopAll(ctx context.Context) error {
991 if ctx == nil {987 if ctx == nil {
992 ctx = context.Background()988 ctx = context.Background()
993 }989 }
990+ rootSessionID = strings.TrimSpace(rootSessionID)
991+ if rootSessionID == "" {
992+ return errors.New("subagents: root session id is required")
993+ }
994 agents := f.List()994 agents := f.List()
995+ owned := make([]*Agent, 0, len(agents))
996+ for _, a := range agents {
997+ if a.RootSessionID == rootSessionID {
998+ owned = append(owned, a)
999+ }
1000+ }
1001+ return f.stopAgents(ctx, owned, ShutdownOriginSession)
1002+}
1003+
1004+func (f *Supervisor) stopAgents(ctx context.Context, agents []*Agent, origin ShutdownOrigin) error {
995 var firstErr error1005 var firstErr error
996 for _, a := range agents {1006 for _, a := range agents {
997- if err := f.stop(ctx, a.ID); err != nil && firstErr == nil {1007+ if err := f.stop(ctx, a.ID, origin); err != nil && firstErr == nil {
998 firstErr = err1008 firstErr = err
999 }1009 }
1000 }1010 }
1001- defer func() {
1002- if f.lifetimeCancel != nil {
1003- f.lifetimeCancel()
1004- }
1005- }()
1006 for _, a := range agents {1011 for _, a := range agents {
1007 select {1012 select {
1008 case <-a.done:1013 case <-a.done:
@@ -1016,6 +1021,29 @@ func (f *Supervisor) stopAll(ctx context.Context) error {
1016 return firstErr1021 return firstErr
1017 }1022 }
1018 1023
1024+// StopAll cancels every running agent. Used on shutdown.
1025+func (f *Supervisor) StopAll() {
1026+ _ = f.stopAll(context.Background())
1027+}
1028+
1029+// StopAllContext is the context-aware shutdown variant. It returns the
1030+// context error when cancellation interrupts a detached-worker wait.
1031+func (f *Supervisor) StopAllContext(ctx context.Context) error {
1032+ return f.stopAll(ctx)
1033+}
1034+
1035+func (f *Supervisor) stopAll(ctx context.Context) error {
1036+ if ctx == nil {
1037+ ctx = context.Background()
1038+ }
1039+ defer func() {
1040+ if f.lifetimeCancel != nil {
1041+ f.lifetimeCancel()
1042+ }
1043+ }()
1044+ return f.stopAgents(ctx, f.List(), ShutdownOriginProcess)
1045+}
1046+
1019 // Remove tears down the per-agent state for a terminated agent. It1047 // Remove tears down the per-agent state for a terminated agent. It
1020 // is an error to remove an agent that's still running; call Stop1048 // is an error to remove an agent that's still running; call Stop
1021 // first and wait for the status to settle. Detached agents1049 // first and wait for the status to settle. Detached agents
M packages/agent/subagents/supervisor_test.go +99 -0
@@ -163,6 +163,105 @@ func TestStopCancelsRunningAgent(t *testing.T) {
163163 }
164164 }
165165
166+func TestStopByRootSessionLeavesOtherSessionsRunning(t *testing.T) {
167+ started := make(chan string, 2)
168+ stopped := make(chan string, 2)
169+ f := newTestSupervisor(t, func(a *Agent) Runner {
170+ return RunnerFunc(func(ctx context.Context, _ Sink) error {
171+ started <- a.ID
172+ <-ctx.Done()
173+ stopped <- a.ID
174+ return ctx.Err()
175+ })
176+ })
177+ t.Cleanup(f.StopAll)
178+
179+ f.SetActiveSession("session-a")
180+ a, err := f.Spawn(context.Background(), "session A task")
181+ if err != nil {
182+ t.Fatal(err)
183+ }
184+ f.SetActiveSession("session-b")
185+ b, err := f.Spawn(context.Background(), "session B task")
186+ if err != nil {
187+ t.Fatal(err)
188+ }
189+
190+ for range 2 {
191+ <-started
192+ }
193+ if err := f.StopByRootSession("session-a"); err != nil {
194+ t.Fatal(err)
195+ }
196+ if got := <-stopped; got != a.ID {
197+ t.Fatalf("stopped agent = %q, want session A agent %q", got, a.ID)
198+ }
199+ select {
200+ case got := <-stopped:
201+ t.Fatalf("session cleanup also stopped %q", got)
202+ default:
203+ }
204+ if got := b.Status(); got != StatusRunning {
205+ t.Fatalf("session B status = %s, want running", got)
206+ }
207+}
208+
209+func TestSupervisorShutdownCommandsIdentifyOrigin(t *testing.T) {
210+ for _, tc := range []struct {
211+ name string
212+ origin ShutdownOrigin
213+ stop func(*Supervisor, *Agent) error
214+ }{
215+ {name: "targeted", origin: ShutdownOriginTargeted, stop: func(f *Supervisor, a *Agent) error { return f.Stop(a.ID) }},
216+ {name: "session", origin: ShutdownOriginSession, stop: func(f *Supervisor, _ *Agent) error { return f.StopByRootSession("session-a") }},
217+ {name: "process", origin: ShutdownOriginProcess, stop: func(f *Supervisor, _ *Agent) error { return f.StopAllContext(context.Background()) }},
218+ } {
219+ t.Run(tc.name, func(t *testing.T) {
220+ started := make(chan struct{})
221+ commands := make(chan Envelope, 1)
222+ f := newTestSupervisor(t, func(a *Agent) Runner {
223+ return RunnerFunc(func(ctx context.Context, _ Sink) error {
224+ listener, err := Listen(a.InboxPath)
225+ if err != nil {
226+ return err
227+ }
228+ defer listener.Close()
229+ close(started)
230+ select {
231+ case line := <-listener.Lines():
232+ command, err := ParseCommand(line)
233+ if err != nil {
234+ return err
235+ }
236+ commands <- command
237+ return nil
238+ case <-ctx.Done():
239+ return ctx.Err()
240+ }
241+ })
242+ })
243+ f.SetActiveSession("session-a")
244+ a, err := f.Spawn(context.Background(), "origin test")
245+ if err != nil {
246+ t.Fatal(err)
247+ }
248+ <-started
249+ if err := tc.stop(f, a); err != nil {
250+ t.Fatal(err)
251+ }
252+ command := <-commands
253+ var payload AgentShutdownPayload
254+ if err := command.DecodePayload(&payload); err != nil {
255+ t.Fatal(err)
256+ }
257+ if payload.Origin != tc.origin {
258+ t.Fatalf("shutdown origin = %q, want %q", payload.Origin, tc.origin)
259+ }
260+ a.Wait()
261+ })
262+ }
263+}
264+
166265 func TestStopContextCallerCancellationDoesNotEndGracePeriod(t *testing.T) {
167266 if runtime.GOOS == "windows" {
168267 t.Skip("subagent inbox transport uses Unix-domain sockets")
@@ -163,6 +163,105 @@ func TestStopCancelsRunningAgent(t *testing.T) {
163 }163 }
164 }164 }
165 165
166+func TestStopByRootSessionLeavesOtherSessionsRunning(t *testing.T) {
167+ started := make(chan string, 2)
168+ stopped := make(chan string, 2)
169+ f := newTestSupervisor(t, func(a *Agent) Runner {
170+ return RunnerFunc(func(ctx context.Context, _ Sink) error {
171+ started <- a.ID
172+ <-ctx.Done()
173+ stopped <- a.ID
174+ return ctx.Err()
175+ })
176+ })
177+ t.Cleanup(f.StopAll)
178+
179+ f.SetActiveSession("session-a")
180+ a, err := f.Spawn(context.Background(), "session A task")
181+ if err != nil {
182+ t.Fatal(err)
183+ }
184+ f.SetActiveSession("session-b")
185+ b, err := f.Spawn(context.Background(), "session B task")
186+ if err != nil {
187+ t.Fatal(err)
188+ }
189+
190+ for range 2 {
191+ <-started
192+ }
193+ if err := f.StopByRootSession("session-a"); err != nil {
194+ t.Fatal(err)
195+ }
196+ if got := <-stopped; got != a.ID {
197+ t.Fatalf("stopped agent = %q, want session A agent %q", got, a.ID)
198+ }
199+ select {
200+ case got := <-stopped:
201+ t.Fatalf("session cleanup also stopped %q", got)
202+ default:
203+ }
204+ if got := b.Status(); got != StatusRunning {
205+ t.Fatalf("session B status = %s, want running", got)
206+ }
207+}
208+
209+func TestSupervisorShutdownCommandsIdentifyOrigin(t *testing.T) {
210+ for _, tc := range []struct {
211+ name string
212+ origin ShutdownOrigin
213+ stop func(*Supervisor, *Agent) error
214+ }{
215+ {name: "targeted", origin: ShutdownOriginTargeted, stop: func(f *Supervisor, a *Agent) error { return f.Stop(a.ID) }},
216+ {name: "session", origin: ShutdownOriginSession, stop: func(f *Supervisor, _ *Agent) error { return f.StopByRootSession("session-a") }},
217+ {name: "process", origin: ShutdownOriginProcess, stop: func(f *Supervisor, _ *Agent) error { return f.StopAllContext(context.Background()) }},
218+ } {
219+ t.Run(tc.name, func(t *testing.T) {
220+ started := make(chan struct{})
221+ commands := make(chan Envelope, 1)
222+ f := newTestSupervisor(t, func(a *Agent) Runner {
223+ return RunnerFunc(func(ctx context.Context, _ Sink) error {
224+ listener, err := Listen(a.InboxPath)
225+ if err != nil {
226+ return err
227+ }
228+ defer listener.Close()
229+ close(started)
230+ select {
231+ case line := <-listener.Lines():
232+ command, err := ParseCommand(line)
233+ if err != nil {
234+ return err
235+ }
236+ commands <- command
237+ return nil
238+ case <-ctx.Done():
239+ return ctx.Err()
240+ }
241+ })
242+ })
243+ f.SetActiveSession("session-a")
244+ a, err := f.Spawn(context.Background(), "origin test")
245+ if err != nil {
246+ t.Fatal(err)
247+ }
248+ <-started
249+ if err := tc.stop(f, a); err != nil {
250+ t.Fatal(err)
251+ }
252+ command := <-commands
253+ var payload AgentShutdownPayload
254+ if err := command.DecodePayload(&payload); err != nil {
255+ t.Fatal(err)
256+ }
257+ if payload.Origin != tc.origin {
258+ t.Fatalf("shutdown origin = %q, want %q", payload.Origin, tc.origin)
259+ }
260+ a.Wait()
261+ })
262+ }
263+}
264+
166 func TestStopContextCallerCancellationDoesNotEndGracePeriod(t *testing.T) {265 func TestStopContextCallerCancellationDoesNotEndGracePeriod(t *testing.T) {
167 if runtime.GOOS == "windows" {266 if runtime.GOOS == "windows" {
168 t.Skip("subagent inbox transport uses Unix-domain sockets")267 t.Skip("subagent inbox transport uses Unix-domain sockets")

Keyboard shortcuts

Single-key shortcuts are on. ctrl k and esc work whatever this says.

ctrl k :
go to a section, branch, tag, file or repository
?
this list
j k
move the row cursor down / up
enter
open the row under the cursor; on Files, a directory opens or closes in place
↑ ↓ ← → home end
in the file tree (focus on a row): previous / next row, close / open a directory or go to its parent / first entry, first / last row. They never open a file
l h
on Files: open a directory or a file / close a directory or go up to its parent, then to the parent page. Elsewhere: into the side panel (on a merge request, its files) / back to the content; there j k move between its controls
g g G
first / last row
/
filter the list; on Files, find a file anywhere in the repository (esc clears, enter opens the best match)
b
switch branch or tag
g s g f g c g b g m g i g r g a
tabs: summary, files, commits, branches & tags, merge requests, issues, releases, actions (when available)
g h
home
w
wide mode: use the whole width of the screen, or not (remembered)
y
copy a permalink pinned to the commit
d u x
diff: split / unified / collapse or expand all
e E
diff: the whole file in view / every file, with the lines the hunks hide
n
merge request: next note in the diff
esc
close a dialog, leave a field

Branches 1

Tags 476

↑↓ moveenter openesc close