bbbf486 fix(subagents): scope interactive shutdown by session · zut

M packages/agent/cli.go +9 -3


@@ -2281,10 +2281,16 @@ func runInteractive(ctx context.Context, args Args, version string) (runErr erro
2281	2281		if sess != nil && subagentsMgr != nil {
2282	2282		subagentsMgr.SetActiveSession(sess.ID)
2283	2283		}
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.
2286	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	2296		var startupSkills []*skills.Skill

 	if sess != nil && subagentsMgr != nil {
 		subagentsMgr.SetActiveSession(sess.ID)
 	}
-	// Best-effort shutdown on interactive exit: stop all running
+	// Best-effort shutdown on normal interactive exit: stop only workers
-	// agents so they don't outlive their parent zut.
+	// owned by the active host session. Other session scopes can share this
+	// supervisor and must remain alive.
 	if subagentsMgr != nil {
-		defer subagentsMgr.StopAll()
+		defer func() {
+			rootSessionID := subagentsMgr.ActiveSession()
+			if rootSessionID != "" {
+				_ = subagentsMgr.StopByRootSession(rootSessionID)
+			}
+		}()
 	}
 	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
233	233		lifetime: args.SubagentLifetimeTurns,
234	234		current: args.SubagentRunTurns,
235	235		}
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
242	243		)
243	244		maxOutputBytes, maxOutputLines := workerOutputLimits()
244	245

@@ -404,11 +405,14 @@ func runSubagentWorkerMode(ctx context.Context, args Args, version string) (runE
404	405		turnWG.Wait()
405	406		}
406	407
407		-	requestShutdown := func() {
	408	+	requestShutdown := func(origin subagents.ShutdownOrigin) {
408	409		mu.Lock()
409	410		closing = true
410	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	418		// 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	438		// alive indefinitely while the host waits for its socket to
435	439		// disappear.
436	440		waitForActiveTurn(true)
437		-	em.emit("agent.exited", map[string]any{"reason": "shutdown"})
	441	+	em.emit("agent.exited", shutdownExitPayload(shutdownOrigin))
438	442		return nil
439	443		case msg, ok := <-ln.Lines():
440	444		if !ok {

@@ -449,7 +453,12 @@ func runSubagentWorkerMode(ctx context.Context, args Args, version string) (runE
449	453		}
450	454		switch command.Type {
451	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	462		case subagents.CommandTurnCancel:
454	463		mu.Lock()
455	464		if cancelFn != nil {

@@ -656,6 +665,14 @@ func boundedResultSummary(text string) string {
656	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	676		func workerOutputLimits() (maxBytes, maxLines int) {
660	677		maxBytes, maxLines = 500_000, 5_000
661	678		if value, err := strconv.Atoi(strings.TrimSpace(os.Getenv("ZUT_SUBAGENT_MAX_OUTPUT_BYTES"))); err == nil && value > 0 {

 			lifetime: args.SubagentLifetimeTurns,
 			current:  args.SubagentRunTurns,
 		}
-		turnWG       sync.WaitGroup
+		turnWG         sync.WaitGroup
-		closing      bool
+		closing        bool
-		turnPending  bool
+		turnPending    bool
-		seenCommands = make(map[string]struct{})
+		seenCommands   = make(map[string]struct{})
-		shutdown     = make(chan struct{})
+		shutdown       = make(chan struct{})
-		shutdownOnce sync.Once
+		shutdownOnce   sync.Once
+		shutdownOrigin subagents.ShutdownOrigin
 	)
 	maxOutputBytes, maxOutputLines := workerOutputLimits()
 		turnWG.Wait()
 	}
-	requestShutdown := func() {
+	requestShutdown := func(origin subagents.ShutdownOrigin) {
 		mu.Lock()
 		closing = true
 		mu.Unlock()
-		shutdownOnce.Do(func() { close(shutdown) })
+		shutdownOnce.Do(func() {
+			shutdownOrigin = origin.Sanitized()
+			close(shutdown)
+		})
 	}
 	// Initial task: run before processing the inbox so the agent
 			// alive indefinitely while the host waits for its socket to
 			// disappear.
 			waitForActiveTurn(true)
-			em.emit("agent.exited", map[string]any{"reason": "shutdown"})
+			em.emit("agent.exited", shutdownExitPayload(shutdownOrigin))
 			return nil
 		case msg, ok := <-ln.Lines():
 			if !ok {
 			}
 			switch command.Type {
 			case subagents.CommandAgentShutdown:
-				requestShutdown()
+				var payload subagents.AgentShutdownPayload
+				if err := command.DecodePayload(&payload); err != nil {
+					em.emit("error", map[string]any{"message": "invalid agent.shutdown payload"})
+					continue
+				}
+				requestShutdown(payload.Origin)
 			case subagents.CommandTurnCancel:
 				mu.Lock()
 				if cancelFn != nil {
 	return boundedResultOutput(firstLine(text), 4*1024, 1)
 }
+func shutdownExitPayload(origin subagents.ShutdownOrigin) map[string]any {
+	payload := map[string]any{"reason": "shutdown"}
+	if origin = origin.Sanitized(); origin != "" {
+		payload["origin"] = origin
+	}
+	return payload
+}
 func workerOutputLimits() (maxBytes, maxLines int) {
 	maxBytes, maxLines = 500_000, 5_000
 	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 (
19	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	33		// TestWorkerOutputLimitsUseSupervisorPolicy verifies that the child reads
23	34		// the effective output caps propagated by the supervisor and retains safe
24	35		// 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 reads	33		// TestWorkerOutputLimitsUseSupervisorPolicy verifies that the child reads
23	// the effective output caps propagated by the supervisor and retains safe	34		// 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 {
564	564		// protocol without forcing a receiver to use them. Receivers can always use
565	565		// Envelope.Payload to retain fields added in later versions.
566	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	589		type AgentReadyPayload struct {
569	590		// Version is the worker/application version. WorkerVersion is the
570	591		// unambiguous spelling for new producers.

@@ -628,6 +649,7 @@ type TurnFailedPayload struct {
628	649		Error *ProtocolError `json:"error,omitempty"`
629	650		}
630	651		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"`
633	655		}


@@ -564,7 +564,28 @@ func cloneRaw(value json.RawMessage) json.RawMessage {
564		// protocol without forcing a receiver to use them. Receivers can always use	564		// 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 the	590		// 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) {
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	254		func TestProtocolResultShape(t *testing.T) {
237	255		result := TurnResultPayload{
238	256		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
544	544		defer cancel()
545	545		if r.agent != nil && r.agent.inbox != nil {
546	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


@@ -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 (
5	5		"encoding/json"
6	6		"fmt"
7	7		"os"
	8	+	"os/exec"
8	9		"path/filepath"
	10	+	"runtime"
9	11		"strings"
10	12		"testing"
11	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	61		// TestSubagentWorkerArgs locks in the exact flag set the subprocess
15	62		// runner uses to start a subagent agent in daemon mode. Past
16	63		// 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 subprocess	61		// TestSubagentWorkerArgs locks in the exact flag set the subprocess
15	// runner uses to start a subagent agent in daemon mode. Past	62		// 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
776	776		// shutdown command through the durable inbox instead of pretending no
777	777		// process exists.
778	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	782		// StopContext is Stop with a context that can interrupt detached-worker
783	783		// shutdown and its graceful and force-stop waits. The context is not held
784	784		// under the supervisor operation lock while those waits run.
785	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	790		if ctx == nil {
791	791		ctx = context.Background()
792	792		}

@@ -820,7 +820,7 @@ func (f *Supervisor) stop(ctx context.Context, id string) error {
820	820		// The detached worker may take the full grace period to exit. Do not
821	821		// serialize unrelated supervisor operations for that entire wait.
822	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	825		a.status = StatusKilled
826	826		a.activity = "canceling"

@@ -870,7 +870,7 @@ func (f *Supervisor) stop(ctx context.Context, id string) error {
870	870		}
871	871		graceful := false
872	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	874		if err == nil {
875	875		graceful = true
876	876		}

@@ -910,7 +910,7 @@ func (f *Supervisor) stop(ctx context.Context, id string) error {
910	910		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	914		if inbox == nil {
915	915		return nil
916	916		}

@@ -923,7 +923,7 @@ func (f Supervisor) stopDetached(ctx context.Context, a Agent, path string, in
923	923		if err := ctx.Err(); err != nil {
924	924		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	927		if ctxErr := ctx.Err(); ctxErr != nil {
928	928		return ctxErr
929	929		}

@@ -976,33 +976,38 @@ func (f Supervisor) markDetachedKilled(a Agent) {
976	976		a.closeDone()
977	977		}
978	978
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)
982	983		}
983	984
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 {
991	987		if ctx == nil {
992	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	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	1005		var firstErr error
996	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	1008		firstErr = err
999	1009		}
1000	1010		}
1001		-	defer func() {
1002		-	if f.lifetimeCancel != nil {
1003		-	f.lifetimeCancel()
1004		-	}
1005		-	}()
1006	1011		for _, a := range agents {
1007	1012		select {
1008	1013		case <-a.done:

@@ -1016,6 +1021,29 @@ func (f *Supervisor) stopAll(ctx context.Context) error {
1016	1021		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	1047		// Remove tears down the per-agent state for a terminated agent. It
1020	1048		// is an error to remove an agent that's still running; call Stop
1021	1049		// first and wait for the status to settle. Detached agents

 // shutdown command through the durable inbox instead of pretending no
 // process exists.
 func (f *Supervisor) Stop(id string) error {
-	return f.stop(context.Background(), id)
+	return f.stop(context.Background(), id, ShutdownOriginTargeted)
 }
 // StopContext is Stop with a context that can interrupt detached-worker
 // shutdown and its graceful and force-stop waits. The context is not held
 // under the supervisor operation lock while those waits run.
 func (f *Supervisor) StopContext(ctx context.Context, id string) error {
-	return f.stop(ctx, id)
+	return f.stop(ctx, id, ShutdownOriginTargeted)
 }
-func (f *Supervisor) stop(ctx context.Context, id string) error {
+func (f *Supervisor) stop(ctx context.Context, id string, origin ShutdownOrigin) error {
 	if ctx == nil {
 		ctx = context.Background()
 	}
 		// The detached worker may take the full grace period to exit. Do not
 		// serialize unrelated supervisor operations for that entire wait.
 		f.operationMu.Unlock()
-		return f.stopDetached(ctx, a, path, inbox, pid)
+		return f.stopDetached(ctx, a, path, inbox, pid, origin)
 	}
 	a.status = StatusKilled
 	a.activity = "canceling"
 	}
 	graceful := false
 	if a.inbox != nil {
-		err := a.inbox.SendCommandContext(ctx, NewCommand(CommandAgentShutdown, a.ID, a.CurrentTurnID(), AgentShutdownPayload{}))
+		err := a.inbox.SendCommandContext(ctx, NewCommand(CommandAgentShutdown, a.ID, a.CurrentTurnID(), AgentShutdownPayload{Origin: origin.Sanitized()}))
 		if err == nil {
 			graceful = true
 		}
 	return nil
 }
-func (f *Supervisor) stopDetached(ctx context.Context, a *Agent, path string, inbox *Inbox, pid int) error {
+func (f *Supervisor) stopDetached(ctx context.Context, a *Agent, path string, inbox *Inbox, pid int, origin ShutdownOrigin) error {
 	if inbox == nil {
 		return nil
 	}
 	if err := ctx.Err(); err != nil {
 		return err
 	}
-	if err := inbox.SendCommandContext(ctx, NewCommand(CommandAgentShutdown, a.ID, a.CurrentTurnID(), AgentShutdownPayload{})); err != nil {
+	if err := inbox.SendCommandContext(ctx, NewCommand(CommandAgentShutdown, a.ID, a.CurrentTurnID(), AgentShutdownPayload{Origin: origin.Sanitized()})); err != nil {
 		if ctxErr := ctx.Err(); ctxErr != nil {
 			return ctxErr
 		}
 	a.closeDone()
 }
-// StopAll cancels every running agent. Used on shutdown.
+// StopByRootSession stops and waits for every agent owned by rootSessionID.
-func (f *Supervisor) StopAll() {
+// It leaves the supervisor lifetime and agents from other sessions running.
-	_ = f.stopAll(context.Background())
+func (f *Supervisor) StopByRootSession(rootSessionID string) error {
+	return f.StopByRootSessionContext(context.Background(), rootSessionID)
 }
-// StopAllContext is the context-aware shutdown variant. It returns the
+// StopByRootSessionContext is the context-aware form of StopByRootSession.
-// context error when cancellation interrupts a detached-worker wait.
+func (f *Supervisor) StopByRootSessionContext(ctx context.Context, rootSessionID string) error {
-func (f *Supervisor) StopAllContext(ctx context.Context) error {
-	return f.stopAll(ctx)
-}
-func (f *Supervisor) stopAll(ctx context.Context) error {
 	if ctx == nil {
 		ctx = context.Background()
 	}
+	rootSessionID = strings.TrimSpace(rootSessionID)
+	if rootSessionID == "" {
+		return errors.New("subagents: root session id is required")
+	}
 	agents := f.List()
+	owned := make([]*Agent, 0, len(agents))
+	for _, a := range agents {
+		if a.RootSessionID == rootSessionID {
+			owned = append(owned, a)
+		}
+	}
+	return f.stopAgents(ctx, owned, ShutdownOriginSession)
+}
+func (f *Supervisor) stopAgents(ctx context.Context, agents []*Agent, origin ShutdownOrigin) error {
 	var firstErr error
 	for _, a := range agents {
-		if err := f.stop(ctx, a.ID); err != nil && firstErr == nil {
+		if err := f.stop(ctx, a.ID, origin); err != nil && firstErr == nil {
 			firstErr = err
 		}
 	}
-	defer func() {
-		if f.lifetimeCancel != nil {
-			f.lifetimeCancel()
-		}
-	}()
 	for _, a := range agents {
 		select {
 		case <-a.done:
 	return firstErr
 }
+// StopAll cancels every running agent. Used on shutdown.
+func (f *Supervisor) StopAll() {
+	_ = f.stopAll(context.Background())
+}
+// StopAllContext is the context-aware shutdown variant. It returns the
+// context error when cancellation interrupts a detached-worker wait.
+func (f *Supervisor) StopAllContext(ctx context.Context) error {
+	return f.stopAll(ctx)
+}
+func (f *Supervisor) stopAll(ctx context.Context) error {
+	if ctx == nil {
+		ctx = context.Background()
+	}
+	defer func() {
+		if f.lifetimeCancel != nil {
+			f.lifetimeCancel()
+		}
+	}()
+	return f.stopAgents(ctx, f.List(), ShutdownOriginProcess)
+}
 // Remove tears down the per-agent state for a terminated agent. It
 // is an error to remove an agent that's still running; call Stop
 // 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) {
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	265		func TestStopContextCallerCancellationDoesNotEndGracePeriod(t *testing.T) {
167	266		if runtime.GOOS == "windows" {
168	267		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")