d422b10 fix(subagents): harden queued resume delivery · zut

M packages/agent/subagent_worker.go +7 -0


@@ -252,6 +252,13 @@ func runSubagentWorkerMode(ctx context.Context, args Args, version string) error
252	252		em.emit("turn.result", map[string]any{"status": "failed", "turn_id": turnID, "error": errPayload})
253	253		em.emit("turn.failed", map[string]any{"turn_id": turnID, "error": errPayload})
254	254		em.emit("turn_end", map[string]any{"step": step, "turn_id": turnID, "error": errPayload["message"]})
	255	+	// Max-turn rejection ends the current run, but the worker remains
	256	+	// alive and can accept a queued follow-up as a fresh run.
	257	+	em.emit("agent.idle", map[string]any{
	258	+	"turn_id": turnID,
	259	+	"lifetime_turns": lifetime,
	260	+	"current_run_turns": currentRun,
	261	+	})
255	262		return
256	263		}
257	264		turnID := fmt.Sprintf("turn-%d", step)


@@ -252,6 +252,13 @@ func runSubagentWorkerMode(ctx context.Context, args Args, version string) error
252	em.emit("turn.result", map[string]any{"status": "failed", "turn_id": turnID, "error": errPayload})	252		em.emit("turn.result", map[string]any{"status": "failed", "turn_id": turnID, "error": errPayload})
253	em.emit("turn.failed", map[string]any{"turn_id": turnID, "error": errPayload})	253		em.emit("turn.failed", map[string]any{"turn_id": turnID, "error": errPayload})
254	em.emit("turn_end", map[string]any{"step": step, "turn_id": turnID, "error": errPayload["message"]})	254		em.emit("turn_end", map[string]any{"step": step, "turn_id": turnID, "error": errPayload["message"]})
		255	+	// Max-turn rejection ends the current run, but the worker remains
		256	+	// alive and can accept a queued follow-up as a fresh run.
		257	+	em.emit("agent.idle", map[string]any{
		258	+	"turn_id": turnID,
		259	+	"lifetime_turns": lifetime,
		260	+	"current_run_turns": currentRun,
		261	+	})
255	return	262		return
256	}	263		}
257	turnID := fmt.Sprintf("turn-%d", step)	264		turnID := fmt.Sprintf("turn-%d", step)

M packages/agent/subagents/ack_test.go +25 -0


@@ -38,3 +38,28 @@ func TestResumePromptAcknowledgementIgnoresNestedTurnStart(t *testing.T) {
38	38		t.Fatal("nested turn.started acknowledged a pending delegated prompt")
39	39		}
40	40		}
	41	+
	42	+	func TestResumePromptAcknowledgementIgnoresStaleMatchingUserMessage(t *testing.T) {
	43	+	acceptedAt := time.Now()
	44	+	prompt := "follow-up prompt"
	45	+	user := NewEvent("user_message", map[string]any{
	46	+	"content": []any{map[string]any{"type": "text", "text": prompt}},
	47	+	})
	48	+	user.Time = acceptedAt.Add(-time.Millisecond)
	49	+	started := NewEvent(EventTurnStarted, map[string]any{"lifetime_turns": 2, "current_run_turns": 1})
	50	+	if resumePromptAcknowledged([]Event{user, started}, prompt, acceptedAt) {
	51	+	t.Fatal("stale matching user message acknowledged a pending prompt")
	52	+	}
	53	+	}
	54	+
	55	+	func TestResumePromptAcknowledgementIgnoresAmbiguousUserMessage(t *testing.T) {
	56	+	acceptedAt := time.Now().Add(-time.Second)
	57	+	prompt := "follow-up prompt"
	58	+	user := NewEvent("user_message", map[string]any{
	59	+	"content": []any{map[string]any{"type": "text", "text": "older prompt"}},
	60	+	})
	61	+	started := NewEvent(EventTurnStarted, map[string]any{"lifetime_turns": 2, "current_run_turns": 1})
	62	+	if resumePromptAcknowledged([]Event{user, started}, prompt, acceptedAt) {
	63	+	t.Fatal("ambiguous user message acknowledged a pending prompt")
	64	+	}
	65	+	}


@@ -38,3 +38,28 @@ func TestResumePromptAcknowledgementIgnoresNestedTurnStart(t *testing.T) {
38	t.Fatal("nested turn.started acknowledged a pending delegated prompt")	38		t.Fatal("nested turn.started acknowledged a pending delegated prompt")
39	}	39		}
40	}	40		}
		41	+
		42	+	func TestResumePromptAcknowledgementIgnoresStaleMatchingUserMessage(t *testing.T) {
		43	+	acceptedAt := time.Now()
		44	+	prompt := "follow-up prompt"
		45	+	user := NewEvent("user_message", map[string]any{
		46	+	"content": []any{map[string]any{"type": "text", "text": prompt}},
		47	+	})
		48	+	user.Time = acceptedAt.Add(-time.Millisecond)
		49	+	started := NewEvent(EventTurnStarted, map[string]any{"lifetime_turns": 2, "current_run_turns": 1})
		50	+	if resumePromptAcknowledged([]Event{user, started}, prompt, acceptedAt) {
		51	+	t.Fatal("stale matching user message acknowledged a pending prompt")
		52	+	}
		53	+	}
		54	+
		55	+	func TestResumePromptAcknowledgementIgnoresAmbiguousUserMessage(t *testing.T) {
		56	+	acceptedAt := time.Now().Add(-time.Second)
		57	+	prompt := "follow-up prompt"
		58	+	user := NewEvent("user_message", map[string]any{
		59	+	"content": []any{map[string]any{"type": "text", "text": "older prompt"}},
		60	+	})
		61	+	started := NewEvent(EventTurnStarted, map[string]any{"lifetime_turns": 2, "current_run_turns": 1})
		62	+	if resumePromptAcknowledged([]Event{user, started}, prompt, acceptedAt) {
		63	+	t.Fatal("ambiguous user message acknowledged a pending prompt")
		64	+	}
		65	+	}

M packages/agent/subagents/agent.go +46 -9


@@ -25,6 +25,12 @@ type queuedResumePrompt struct {
25	25		CommandID string `json:"command_id,omitempty"`
26	26		}
27	27
	28	+	type resumePromptSnapshot struct {
	29	+	Prompt string
	30	+	AcceptedAt time.Time
	31	+	CommandID string
	32	+	}
	33	+
28	34		type Agent struct {
29	35		ID string
30	36		Task string

@@ -481,14 +487,39 @@ func (a *Agent) ResumePromptInfo() (string, time.Time) {
481	487		return a.resumePromptText, a.resumePromptAt
482	488		}
483	489
484		-	func (a *Agent) setResumePrompt(prompt string, acceptedAt time.Time) (string, time.Time) {
	490	+	func (a *Agent) setResumePrompt(prompt string, acceptedAt time.Time) resumePromptSnapshot {
485	491		a.mu.Lock()
486	492		defer a.mu.Unlock()
487		-	previousPrompt, previousAcceptedAt := a.resumePromptText, a.resumePromptAt
	493	+	previous := resumePromptSnapshot{
	494	+	Prompt: a.resumePromptText,
	495	+	AcceptedAt: a.resumePromptAt,
	496	+	CommandID: a.resumePromptID,
	497	+	}
488	498		a.resumePromptText = prompt
489	499		a.resumePromptAt = acceptedAt
490	500		a.resumePromptID = uuid.NewString()
491		-	return previousPrompt, previousAcceptedAt
	501	+	return previous
	502	+	}
	503	+
	504	+	func (a *Agent) restoreResumePrompt(previous resumePromptSnapshot) {
	505	+	a.mu.Lock()
	506	+	defer a.mu.Unlock()
	507	+	a.resumePromptText = previous.Prompt
	508	+	a.resumePromptAt = previous.AcceptedAt
	509	+	a.resumePromptID = previous.CommandID
	510	+	if a.resumePromptText != "" && a.resumePromptID == "" {
	511	+	a.resumePromptID = uuid.NewString()
	512	+	}
	513	+	}
	514	+
	515	+	func (a *Agent) resumePromptState() resumePromptSnapshot {
	516	+	a.mu.Lock()
	517	+	defer a.mu.Unlock()
	518	+	return resumePromptSnapshot{
	519	+	Prompt: a.resumePromptText,
	520	+	AcceptedAt: a.resumePromptAt,
	521	+	CommandID: a.resumePromptID,
	522	+	}
492	523		}
493	524
494	525		func (a *Agent) resumePromptCommandID() string {

@@ -509,18 +540,22 @@ func (a *Agent) clearResumePrompt() bool {
509	540		return true
510	541		}
511	542
512		-	func (a *Agent) queueResumePrompt(prompt string, acceptedAt time.Time) {
	543	+	func (a *Agent) queueResumePrompt(prompt string, acceptedAt time.Time) string {
	544	+	commandID := uuid.NewString()
513	545		a.mu.Lock()
514		-	a.resumePromptQueue = append(a.resumePromptQueue, queuedResumePrompt{Prompt: prompt, AcceptedAt: acceptedAt, CommandID: uuid.NewString()})
	546	+	a.resumePromptQueue = append(a.resumePromptQueue, queuedResumePrompt{Prompt: prompt, AcceptedAt: acceptedAt, CommandID: commandID})
515	547		a.mu.Unlock()
	548	+	return commandID
516	549		}
517	550
518		-	func (a *Agent) removeQueuedResumePrompt(prompt string, acceptedAt time.Time) bool {
	551	+	func (a *Agent) removeQueuedResumePrompt(commandID string) bool {
	552	+	if commandID == "" {
	553	+	return false
	554	+	}
519	555		a.mu.Lock()
520	556		defer a.mu.Unlock()
521	557		for i := len(a.resumePromptQueue) - 1; i >= 0; i-- {
522		-	queued := a.resumePromptQueue[i]
523		-	if queued.Prompt == prompt && queued.AcceptedAt.Equal(acceptedAt) {
	558	+	if a.resumePromptQueue[i].CommandID == commandID {
524	559		a.resumePromptQueue = append(a.resumePromptQueue[:i], a.resumePromptQueue[i+1:]...)
525	560		return true
526	561		}

@@ -623,7 +658,9 @@ func (a *Agent) recordPersistenceError(err error) {
623	658		return
624	659		}
625	660		a.mu.Lock()
626		-	a.lastErr = fmt.Errorf("subagent metadata persistence: %w", err)
	661	+	if a.lastErr == nil {
	662	+	a.lastErr = fmt.Errorf("subagent metadata persistence: %w", err)
	663	+	}
627	664		if a.status != StatusDone && a.status != StatusFailed && a.status != StatusKilled {
628	665		a.activity = "metadata error: " + truncate(err.Error(), 120)
629	666		}

 	CommandID  string    `json:"command_id,omitempty"`
 }
+type resumePromptSnapshot struct {
+	Prompt     string
+	AcceptedAt time.Time
+	CommandID  string
+}
 type Agent struct {
 	ID      string
 	Task    string
 	return a.resumePromptText, a.resumePromptAt
 }
-func (a *Agent) setResumePrompt(prompt string, acceptedAt time.Time) (string, time.Time) {
+func (a *Agent) setResumePrompt(prompt string, acceptedAt time.Time) resumePromptSnapshot {
 	a.mu.Lock()
 	defer a.mu.Unlock()
-	previousPrompt, previousAcceptedAt := a.resumePromptText, a.resumePromptAt
+	previous := resumePromptSnapshot{
+		Prompt:     a.resumePromptText,
+		AcceptedAt: a.resumePromptAt,
+		CommandID:  a.resumePromptID,
+	}
 	a.resumePromptText = prompt
 	a.resumePromptAt = acceptedAt
 	a.resumePromptID = uuid.NewString()
-	return previousPrompt, previousAcceptedAt
+	return previous
+}
+func (a *Agent) restoreResumePrompt(previous resumePromptSnapshot) {
+	a.mu.Lock()
+	defer a.mu.Unlock()
+	a.resumePromptText = previous.Prompt
+	a.resumePromptAt = previous.AcceptedAt
+	a.resumePromptID = previous.CommandID
+	if a.resumePromptText != "" && a.resumePromptID == "" {
+		a.resumePromptID = uuid.NewString()
+	}
+}
+func (a *Agent) resumePromptState() resumePromptSnapshot {
+	a.mu.Lock()
+	defer a.mu.Unlock()
+	return resumePromptSnapshot{
+		Prompt:     a.resumePromptText,
+		AcceptedAt: a.resumePromptAt,
+		CommandID:  a.resumePromptID,
+	}
 }
 func (a *Agent) resumePromptCommandID() string {
 	return true
 }
-func (a *Agent) queueResumePrompt(prompt string, acceptedAt time.Time) {
+func (a *Agent) queueResumePrompt(prompt string, acceptedAt time.Time) string {
+	commandID := uuid.NewString()
 	a.mu.Lock()
-	a.resumePromptQueue = append(a.resumePromptQueue, queuedResumePrompt{Prompt: prompt, AcceptedAt: acceptedAt, CommandID: uuid.NewString()})
+	a.resumePromptQueue = append(a.resumePromptQueue, queuedResumePrompt{Prompt: prompt, AcceptedAt: acceptedAt, CommandID: commandID})
 	a.mu.Unlock()
+	return commandID
 }
-func (a *Agent) removeQueuedResumePrompt(prompt string, acceptedAt time.Time) bool {
+func (a *Agent) removeQueuedResumePrompt(commandID string) bool {
+	if commandID == "" {
+		return false
+	}
 	a.mu.Lock()
 	defer a.mu.Unlock()
 	for i := len(a.resumePromptQueue) - 1; i >= 0; i-- {
-		queued := a.resumePromptQueue[i]
+		if a.resumePromptQueue[i].CommandID == commandID {
-		if queued.Prompt == prompt && queued.AcceptedAt.Equal(acceptedAt) {
 			a.resumePromptQueue = append(a.resumePromptQueue[:i], a.resumePromptQueue[i+1:]...)
 			return true
 		}
 		return
 	}
 	a.mu.Lock()
-	a.lastErr = fmt.Errorf("subagent metadata persistence: %w", err)
+	if a.lastErr == nil {
+		a.lastErr = fmt.Errorf("subagent metadata persistence: %w", err)
+	}
 	if a.status != StatusDone && a.status != StatusFailed && a.status != StatusKilled {
 		a.activity = "metadata error: " + truncate(err.Error(), 120)
 	}

M packages/agent/subagents/inbox.go +11 -12


@@ -119,12 +119,7 @@ func (b *Inbox) sendBytesContext(ctx context.Context, data []byte) error {
119	119		// caller's context error would invite a duplicate command with a new
120	120		// identity.
121	121		stopCancel()
122		-	b.mu.Lock()
123		-	if b.conn == conn {
124		-	_ = b.conn.Close()
125		-	b.conn = nil
126		-	}
127		-	b.mu.Unlock()
	122	+	b.dropConnection(conn)
128	123		if n > 0 {
129	124		return fmt.Errorf("%w: wrote %d of %d bytes: %v", ErrDeliveryUnknown, n, len(data), writeErr)
130	125		}

@@ -135,12 +130,7 @@ func (b *Inbox) sendBytesContext(ctx context.Context, data []byte) error {
135	130		}
136	131		if n != len(data) {
137	132		stopCancel()
138		-	b.mu.Lock()
139		-	if b.conn == conn {
140		-	_ = b.conn.Close()
141		-	b.conn = nil
142		-	}
143		-	b.mu.Unlock()
	133	+	b.dropConnection(conn)
144	134		return fmt.Errorf("%w: wrote %d of %d bytes", ErrDeliveryUnknown, n, len(data))
145	135		}
146	136		if !stopCancel() {

@@ -156,6 +146,15 @@ func (b *Inbox) sendBytesContext(ctx context.Context, data []byte) error {
156	146		return nil
157	147		}
158	148
	149	+	func (b *Inbox) dropConnection(conn net.Conn) {
	150	+	b.mu.Lock()
	151	+	defer b.mu.Unlock()
	152	+	if b.conn == conn {
	153	+	_ = b.conn.Close()
	154	+	b.conn = nil
	155	+	}
	156	+	}
	157	+
159	158		func (b *Inbox) acquireSend(ctx context.Context) (chan struct{}, error) {
160	159		b.mu.Lock()
161	160		if b.send == nil {

 		// caller's context error would invite a duplicate command with a new
 		// identity.
 		stopCancel()
-		b.mu.Lock()
+		b.dropConnection(conn)
-		if b.conn == conn {
-			_ = b.conn.Close()
-			b.conn = nil
-		}
-		b.mu.Unlock()
 		if n > 0 {
 			return fmt.Errorf("%w: wrote %d of %d bytes: %v", ErrDeliveryUnknown, n, len(data), writeErr)
 		}
 	}
 	if n != len(data) {
 		stopCancel()
-		b.mu.Lock()
+		b.dropConnection(conn)
-		if b.conn == conn {
-			_ = b.conn.Close()
-			b.conn = nil
-		}
-		b.mu.Unlock()
 		return fmt.Errorf("%w: wrote %d of %d bytes", ErrDeliveryUnknown, n, len(data))
 	}
 	if !stopCancel() {
 	return nil
 }
+func (b *Inbox) dropConnection(conn net.Conn) {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	if b.conn == conn {
+		_ = b.conn.Close()
+		b.conn = nil
+	}
+}
 func (b *Inbox) acquireSend(ctx context.Context) (chan struct{}, error) {
 	b.mu.Lock()
 	if b.send == nil {

M packages/agent/subagents/inbox_test.go +37 -0


@@ -137,6 +137,22 @@ func (c *cancelAfterWriteConn) Write(data []byte) (int, error) {
137	137		return len(data), nil
138	138		}
139	139
	140	+	type partialWriteConn struct {
	141	+	net.Conn
	142	+	withError bool
	143	+	}
	144	+
	145	+	func (c *partialWriteConn) Write(data []byte) (int, error) {
	146	+	written := len(data) / 2
	147	+	if written == 0 {
	148	+	written = 1
	149	+	}
	150	+	if c.withError {
	151	+	return written, errors.New("partial write")
	152	+	}
	153	+	return written, nil
	154	+	}
	155	+
140	156		type notifyingContext struct {
141	157		context.Context
142	158		doneCalled chan struct{}

@@ -161,6 +177,27 @@ func TestInboxSendContextTreatsCompleteWriteBeforeCancellationAsDelivered(t *tes
161	177		}
162	178		}
163	179
	180	+	func TestInboxSendContextTreatsPartialWritesAsUnknownDelivery(t *testing.T) {
	181	+	for _, withError := range []bool{false, true} {
	182	+	name := "short write"
	183	+	if withError {
	184	+	name = "partial write error"
	185	+	}
	186	+	t.Run(name, func(t *testing.T) {
	187	+	local, remote := net.Pipe()
	188	+	defer remote.Close()
	189	+	inbox := NewInbox("unused")
	190	+	inbox.conn = &partialWriteConn{Conn: local, withError: withError}
	191	+	defer inbox.Close()
	192	+
	193	+	err := inbox.sendBytesContext(context.Background(), []byte("follow-up\n"))
	194	+	if !errors.Is(err, ErrDeliveryUnknown) {
	195	+	t.Fatalf("partial write error = %v, want ErrDeliveryUnknown", err)
	196	+	}
	197	+	})
	198	+	}
	199	+	}
	200	+
164	201		func TestInboxSendContextCancelsBlockedWrite(t *testing.T) {
165	202		local, remote := net.Pipe()
166	203		defer remote.Close()


@@ -137,6 +137,22 @@ func (c *cancelAfterWriteConn) Write(data []byte) (int, error) {
137	return len(data), nil	137		return len(data), nil
138	}	138		}
139		139
		140	+	type partialWriteConn struct {
		141	+	net.Conn
		142	+	withError bool
		143	+	}
		144	+
		145	+	func (c *partialWriteConn) Write(data []byte) (int, error) {
		146	+	written := len(data) / 2
		147	+	if written == 0 {
		148	+	written = 1
		149	+	}
		150	+	if c.withError {
		151	+	return written, errors.New("partial write")
		152	+	}
		153	+	return written, nil
		154	+	}
		155	+
140	type notifyingContext struct {	156		type notifyingContext struct {
141	context.Context	157		context.Context
142	doneCalled chan struct{}	158		doneCalled chan struct{}

@@ -161,6 +177,27 @@ func TestInboxSendContextTreatsCompleteWriteBeforeCancellationAsDelivered(t *tes
161	}	177		}
162	}	178		}
163		179
		180	+	func TestInboxSendContextTreatsPartialWritesAsUnknownDelivery(t *testing.T) {
		181	+	for _, withError := range []bool{false, true} {
		182	+	name := "short write"
		183	+	if withError {
		184	+	name = "partial write error"
		185	+	}
		186	+	t.Run(name, func(t *testing.T) {
		187	+	local, remote := net.Pipe()
		188	+	defer remote.Close()
		189	+	inbox := NewInbox("unused")
		190	+	inbox.conn = &partialWriteConn{Conn: local, withError: withError}
		191	+	defer inbox.Close()
		192	+
		193	+	err := inbox.sendBytesContext(context.Background(), []byte("follow-up\n"))
		194	+	if !errors.Is(err, ErrDeliveryUnknown) {
		195	+	t.Fatalf("partial write error = %v, want ErrDeliveryUnknown", err)
		196	+	}
		197	+	})
		198	+	}
		199	+	}
		200	+
164	func TestInboxSendContextCancelsBlockedWrite(t *testing.T) {	201		func TestInboxSendContextCancelsBlockedWrite(t *testing.T) {
165	local, remote := net.Pipe()	202		local, remote := net.Pipe()
166	defer remote.Close()	203		defer remote.Close()

M packages/agent/subagents/persist.go +47 -15


@@ -22,6 +22,8 @@ import (
22	22		"sort"
23	23		"strings"
24	24		"time"
	25	+
	26	+	"github.com/google/uuid"
25	27		)
26	28
27	29		// agentMeta is the durable identity record for one agent. Only fields

@@ -844,9 +846,9 @@ func (f Supervisor) ResumeWithPrompt(ctx context.Context, id, prompt string) (
844	846		// Keep the prompt durable and let the idle transition deliver it in
845	847		// order after the current message turn finishes.
846	848		acceptedAt := f.cfg.Now()
847		-	existing.queueResumePrompt(prompt, acceptedAt)
	849	+	commandID := existing.queueResumePrompt(prompt, acceptedAt)
848	850		if err := f.persistAgent(existing); err != nil {
849		-	existing.removeQueuedResumePrompt(prompt, acceptedAt)
	851	+	existing.removeQueuedResumePrompt(commandID)
850	852		f.operationMu.Unlock()
851	853		return nil, fmt.Errorf("subagent: persist queued follow-up for %s: %w", existing.ID, err)
852	854		}

@@ -863,7 +865,7 @@ func (f Supervisor) ResumeWithPrompt(ctx context.Context, id, prompt string) (
863	865		// must not enqueue another turn in the small interval before the worker
864	866		// reports its turn.start event. Persist the prompt first so a host exit
865	867		// before that event can resume and redeliver it.
866		-	previousPrompt, previousPromptAt := existing.setResumePrompt(prompt, f.cfg.Now())
	868	+	previousPrompt := existing.setResumePrompt(prompt, f.cfg.Now())
867	869		existing.setTurnState(TurnQueued, turnID)
868	870		inbox := existing.inbox
869	871		command := NewCommand(CommandTurnStart, existing.ID, turnID, TurnStartPayload{Prompt: prompt, NewRun: true})

@@ -871,7 +873,7 @@ func (f Supervisor) ResumeWithPrompt(ctx context.Context, id, prompt string) (
871	873		if err := f.persistAgent(existing); err != nil {
872	874		existing.setTurnState(TurnIdle, turnID)
873	875		existing.setCurrentRunTurns(previousRunTurns)
874		-	existing.setResumePrompt(previousPrompt, previousPromptAt)
	876	+	existing.restoreResumePrompt(previousPrompt)
875	877		f.operationMu.Unlock()
876	878		return nil, fmt.Errorf("subagent: persist follow-up for %s: %w", existing.ID, err)
877	879		}

@@ -887,7 +889,7 @@ func (f Supervisor) ResumeWithPrompt(ctx context.Context, id, prompt string) (
887	889		existing.rejectActiveResumePrompt()
888	890		retryUnknown = true
889	891		} else {
890		-	existing.setResumePrompt(previousPrompt, previousPromptAt)
	892	+	existing.restoreResumePrompt(previousPrompt)
891	893		}
892	894		if persistErr := f.persistAgent(existing); persistErr != nil {
893	895		err = fmt.Errorf("%w; persist recovery: %v", err, persistErr)

@@ -897,7 +899,7 @@ func (f Supervisor) ResumeWithPrompt(ctx context.Context, id, prompt string) (
897	899		if retryUnknown {
898	900		// The command keeps its durable identity. A retry is safe whether
899	901		// the worker parsed the first write or only received a prefix.
900		-	f.dispatchQueuedResume(existing)
	902	+	f.dispatchQueuedResume(ctx, existing)
901	903		}
902	904		return nil, fmt.Errorf("subagent: send follow-up to %s: %w", existing.ID, err)
903	905		}

@@ -908,11 +910,26 @@ func (f Supervisor) ResumeWithPrompt(ctx context.Context, id, prompt string) (
908	910		return f.resume(ctx, id, true, prompt)
909	911		}
910	912
	913	+	// dispatchQueuedResumeWithTimeout keeps the idle callback bounded. Unlike a
	914	+	// manager request, an idle transition has no caller context to propagate.
	915	+	func (f Supervisor) dispatchQueuedResumeWithTimeout(a Agent) {
	916	+	timeout := f.cfg.Policy.ReconnectTimeout
	917	+	if timeout <= 0 {
	918	+	timeout = 5 * time.Second
	919	+	}
	920	+	ctx, cancel := context.WithTimeout(context.Background(), timeout)
	921	+	defer cancel()
	922	+	f.dispatchQueuedResume(ctx, a)
	923	+	}
	924	+
911	925		// dispatchQueuedResume delivers the oldest follow-up accepted during an
912	926		// active turn after the worker reports that its inbox can accept a new turn.
913	927		// The operation lock serializes this transition with manager resume calls and
914	928		// the agent's durable prompt state prevents a host exit from losing it.
915		-	func (f Supervisor) dispatchQueuedResume(a Agent) {
	929	+	func (f Supervisor) dispatchQueuedResume(ctx context.Context, a Agent) {
	930	+	if ctx == nil {
	931	+	ctx = context.Background()
	932	+	}
916	933		if a == nil {
917	934		return
918	935		}

@@ -941,7 +958,7 @@ func (f Supervisor) dispatchQueuedResume(a Agent) {
941	958		}
942	959		f.operationMu.Unlock()
943	960
944		-	if err := inbox.SendCommand(command); err == nil {
	961	+	if err := inbox.SendCommandContext(ctx, command); err == nil {
945	962		return
946	963		}
947	964

@@ -953,7 +970,12 @@ func (f Supervisor) dispatchQueuedResume(a Agent) {
953	970		if a.restoreResumePromptToQueue(prompt, acceptedAt) {
954	971		a.setCurrentRunTurns(previousRunTurns)
955	972		a.setTurnState(TurnIdle, turnID)
956		-	f.persistAgent(a)
	973	+	if persistErr := f.persistAgent(a); persistErr != nil {
	974	+	// persistAgent records the failure on the agent. Keep the explicit
	975	+	// check here because this recovery write is the only durable copy
	976	+	// of the restored queue head.
	977	+	a.recordPersistenceError(persistErr)
	978	+	}
957	979		}
958	980		}
959	981		f.operationMu.Unlock()

@@ -1026,23 +1048,33 @@ func (f *Supervisor) resume(ctx context.Context, id string, resuming bool, resum
1026	1048		fastMode := existing.FastMode
1027	1049		lifetimeTurns := existing.LifetimeTurnsValue()
1028	1050		currentRunTurns := existing.CurrentRunTurnsValue()
1029		-	pendingResumePrompt, pendingResumePromptAt := existing.ResumePromptInfo()
	1051	+	pendingResume := existing.resumePromptState()
	1052	+	pendingResumePrompt, pendingResumePromptAt, pendingResumePromptID := pendingResume.Prompt, pendingResume.AcceptedAt, pendingResume.CommandID
1030	1053		pendingResumeQueue := existing.resumePromptQueueSnapshot()
	1054	+	if pendingResumePrompt != "" && pendingResumePromptID == "" {
	1055	+	pendingResumePromptID = uuid.NewString()
	1056	+	}
1031	1057		if resumePrompt != "" {
1032	1058		currentRunTurns = 0
1033	1059		if pendingResumePrompt == "" {
1034	1060		pendingResumePrompt = resumePrompt
1035	1061		pendingResumePromptAt = now
	1062	+	pendingResumePromptID = uuid.NewString()
1036	1063		} else {
1037		-	pendingResumeQueue = append(pendingResumeQueue, queuedResumePrompt{Prompt: resumePrompt, AcceptedAt: now})
	1064	+	pendingResumeQueue = append(pendingResumeQueue, queuedResumePrompt{Prompt: resumePrompt, AcceptedAt: now, CommandID: uuid.NewString()})
1038	1065		}
1039	1066		}
1040	1067		if resuming && resumePrompt == "" && pendingResumePrompt == "" && len(pendingResumeQueue) > 0 {
1041	1068		// A host can exit after accepting an active-worker follow-up but before
1042	1069		// dispatching it. Make the first durable queued prompt the resumed
1043	1070		// worker's initial turn rather than leaving the queue stranded.
1044		-	pendingResumePrompt = pendingResumeQueue[0].Prompt
1045		-	pendingResumePromptAt = pendingResumeQueue[0].AcceptedAt
	1071	+	queued := pendingResumeQueue[0]
	1072	+	pendingResumePrompt = queued.Prompt
	1073	+	pendingResumePromptAt = queued.AcceptedAt
	1074	+	pendingResumePromptID = queued.CommandID
	1075	+	if pendingResumePromptID == "" {
	1076	+	pendingResumePromptID = uuid.NewString()
	1077	+	}
1046	1078		pendingResumeQueue = pendingResumeQueue[1:]
1047	1079		}
1048	1080		if resuming && pendingResumePrompt != "" {

@@ -1073,7 +1105,7 @@ func (f *Supervisor) resume(ctx context.Context, id string, resuming bool, resum
1073	1105		SessionPath: existing.SessionPath,
1074	1106		ResumePrompt: pendingResumePrompt,
1075	1107		ResumePromptAt: pendingResumePromptAt,
1076		-	ResumePromptID: existing.resumePromptCommandID(),
	1108	+	ResumePromptID: pendingResumePromptID,
1077	1109		ResumeQueue: pendingResumeQueue,
1078	1110		}
1079	1111

@@ -1192,7 +1224,7 @@ func (f *Supervisor) resume(ctx context.Context, id string, resuming bool, resum
1192	1224		}
1193	1225		a.ctx, a.cancel = runCtx, cancel
1194	1226		a.persistFn = f.persistAgent
1195		-	a.setOnTurnIdle(func() { f.dispatchQueuedResume(a) })
	1227	+	a.setOnTurnIdle(func() { f.dispatchQueuedResumeWithTimeout(a) })
1196	1228		a.workspaceCleanup = func() error { return workspace.Cleanup(context.Background()) }
1197	1229		a.workspaceCapture = func() (WorkspaceCapture, error) { return workspace.Capture(context.Background()) }
1198	1230		a.runner = f.cfg.NewRunner(a)

 	"sort"
 	"strings"
 	"time"
+	"github.com/google/uuid"
 )
 // agentMeta is the durable identity record for one agent. Only fields
 			// Keep the prompt durable and let the idle transition deliver it in
 			// order after the current message turn finishes.
 			acceptedAt := f.cfg.Now()
-			existing.queueResumePrompt(prompt, acceptedAt)
+			commandID := existing.queueResumePrompt(prompt, acceptedAt)
 			if err := f.persistAgent(existing); err != nil {
-				existing.removeQueuedResumePrompt(prompt, acceptedAt)
+				existing.removeQueuedResumePrompt(commandID)
 				f.operationMu.Unlock()
 				return nil, fmt.Errorf("subagent: persist queued follow-up for %s: %w", existing.ID, err)
 			}
 		// must not enqueue another turn in the small interval before the worker
 		// reports its turn.start event. Persist the prompt first so a host exit
 		// before that event can resume and redeliver it.
-		previousPrompt, previousPromptAt := existing.setResumePrompt(prompt, f.cfg.Now())
+		previousPrompt := existing.setResumePrompt(prompt, f.cfg.Now())
 		existing.setTurnState(TurnQueued, turnID)
 		inbox := existing.inbox
 		command := NewCommand(CommandTurnStart, existing.ID, turnID, TurnStartPayload{Prompt: prompt, NewRun: true})
 		if err := f.persistAgent(existing); err != nil {
 			existing.setTurnState(TurnIdle, turnID)
 			existing.setCurrentRunTurns(previousRunTurns)
-			existing.setResumePrompt(previousPrompt, previousPromptAt)
+			existing.restoreResumePrompt(previousPrompt)
 			f.operationMu.Unlock()
 			return nil, fmt.Errorf("subagent: persist follow-up for %s: %w", existing.ID, err)
 		}
 					existing.rejectActiveResumePrompt()
 					retryUnknown = true
 				} else {
-					existing.setResumePrompt(previousPrompt, previousPromptAt)
+					existing.restoreResumePrompt(previousPrompt)
 				}
 				if persistErr := f.persistAgent(existing); persistErr != nil {
 					err = fmt.Errorf("%w; persist recovery: %v", err, persistErr)
 			if retryUnknown {
 				// The command keeps its durable identity. A retry is safe whether
 				// the worker parsed the first write or only received a prefix.
-				f.dispatchQueuedResume(existing)
+				f.dispatchQueuedResume(ctx, existing)
 			}
 			return nil, fmt.Errorf("subagent: send follow-up to %s: %w", existing.ID, err)
 		}
 	return f.resume(ctx, id, true, prompt)
 }
+// dispatchQueuedResumeWithTimeout keeps the idle callback bounded. Unlike a
+// manager request, an idle transition has no caller context to propagate.
+func (f *Supervisor) dispatchQueuedResumeWithTimeout(a *Agent) {
+	timeout := f.cfg.Policy.ReconnectTimeout
+	if timeout <= 0 {
+		timeout = 5 * time.Second
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), timeout)
+	defer cancel()
+	f.dispatchQueuedResume(ctx, a)
+}
 // dispatchQueuedResume delivers the oldest follow-up accepted during an
 // active turn after the worker reports that its inbox can accept a new turn.
 // The operation lock serializes this transition with manager resume calls and
 // the agent's durable prompt state prevents a host exit from losing it.
-func (f *Supervisor) dispatchQueuedResume(a *Agent) {
+func (f *Supervisor) dispatchQueuedResume(ctx context.Context, a *Agent) {
+	if ctx == nil {
+		ctx = context.Background()
+	}
 	if a == nil {
 		return
 	}
 	}
 	f.operationMu.Unlock()
-	if err := inbox.SendCommand(command); err == nil {
+	if err := inbox.SendCommandContext(ctx, command); err == nil {
 		return
 	}
 		if a.restoreResumePromptToQueue(prompt, acceptedAt) {
 			a.setCurrentRunTurns(previousRunTurns)
 			a.setTurnState(TurnIdle, turnID)
-			f.persistAgent(a)
+			if persistErr := f.persistAgent(a); persistErr != nil {
+				// persistAgent records the failure on the agent. Keep the explicit
+				// check here because this recovery write is the only durable copy
+				// of the restored queue head.
+				a.recordPersistenceError(persistErr)
+			}
 		}
 	}
 	f.operationMu.Unlock()
 	fastMode := existing.FastMode
 	lifetimeTurns := existing.LifetimeTurnsValue()
 	currentRunTurns := existing.CurrentRunTurnsValue()
-	pendingResumePrompt, pendingResumePromptAt := existing.ResumePromptInfo()
+	pendingResume := existing.resumePromptState()
+	pendingResumePrompt, pendingResumePromptAt, pendingResumePromptID := pendingResume.Prompt, pendingResume.AcceptedAt, pendingResume.CommandID
 	pendingResumeQueue := existing.resumePromptQueueSnapshot()
+	if pendingResumePrompt != "" && pendingResumePromptID == "" {
+		pendingResumePromptID = uuid.NewString()
+	}
 	if resumePrompt != "" {
 		currentRunTurns = 0
 		if pendingResumePrompt == "" {
 			pendingResumePrompt = resumePrompt
 			pendingResumePromptAt = now
+			pendingResumePromptID = uuid.NewString()
 		} else {
-			pendingResumeQueue = append(pendingResumeQueue, queuedResumePrompt{Prompt: resumePrompt, AcceptedAt: now})
+			pendingResumeQueue = append(pendingResumeQueue, queuedResumePrompt{Prompt: resumePrompt, AcceptedAt: now, CommandID: uuid.NewString()})
 		}
 	}
 	if resuming && resumePrompt == "" && pendingResumePrompt == "" && len(pendingResumeQueue) > 0 {
 		// A host can exit after accepting an active-worker follow-up but before
 		// dispatching it. Make the first durable queued prompt the resumed
 		// worker's initial turn rather than leaving the queue stranded.
-		pendingResumePrompt = pendingResumeQueue[0].Prompt
+		queued := pendingResumeQueue[0]
-		pendingResumePromptAt = pendingResumeQueue[0].AcceptedAt
+		pendingResumePrompt = queued.Prompt
+		pendingResumePromptAt = queued.AcceptedAt
+		pendingResumePromptID = queued.CommandID
+		if pendingResumePromptID == "" {
+			pendingResumePromptID = uuid.NewString()
+		}
 		pendingResumeQueue = pendingResumeQueue[1:]
 	}
 	if resuming && pendingResumePrompt != "" {
 		SessionPath:    existing.SessionPath,
 		ResumePrompt:   pendingResumePrompt,
 		ResumePromptAt: pendingResumePromptAt,
-		ResumePromptID: existing.resumePromptCommandID(),
+		ResumePromptID: pendingResumePromptID,
 		ResumeQueue:    pendingResumeQueue,
 	}
 	}
 	a.ctx, a.cancel = runCtx, cancel
 	a.persistFn = f.persistAgent
-	a.setOnTurnIdle(func() { f.dispatchQueuedResume(a) })
+	a.setOnTurnIdle(func() { f.dispatchQueuedResumeWithTimeout(a) })
 	a.workspaceCleanup = func() error { return workspace.Cleanup(context.Background()) }
 	a.workspaceCapture = func() (WorkspaceCapture, error) { return workspace.Capture(context.Background()) }
 	a.runner = f.cfg.NewRunner(a)

M packages/agent/subagents/resume_prompt_test.go +16 -4


@@ -90,8 +90,13 @@ func TestResumeWithPromptQueuesFollowUpsForActiveWorker(t *testing.T) {
90	90		if persisted.ResumePrompt != "" \|\| len(persisted.ResumeQueue) != 2 {
91	91		t.Fatalf("active follow-ups persisted as prompt %q queue %d, want empty prompt and queue 2", persisted.ResumePrompt, len(persisted.ResumeQueue))
92	92		}
	93	+	if firstID, secondID := persisted.ResumeQueue[0].CommandID, persisted.ResumeQueue[1].CommandID; firstID == "" \|\| secondID == "" \|\| firstID == secondID {
	94	+	t.Fatalf("queued follow-up command IDs = %q, %q; want distinct non-empty IDs", firstID, secondID)
	95	+	}
93	96
94		-	updateAgentFromEvent(a, NewEvent(EventAgentIdle, map[string]any{"turn_id": "turn-1"}))
	97	+	if err := updateAgentFromEvent(a, NewEvent(EventAgentIdle, map[string]any{"turn_id": "turn-1"})); err != nil {
	98	+	t.Fatalf("apply first idle event: %v", err)
	99	+	}
95	100		select {
96	101		case command := <-commands:
97	102		var payload TurnStartPayload

@@ -108,10 +113,14 @@ func TestResumeWithPromptQueuesFollowUpsForActiveWorker(t *testing.T) {
108	113		t.Fatalf("turn state after first queued delivery = %s, want %s", got, TurnQueued)
109	114		}
110	115
111		-	updateAgentFromEvent(a, NewEvent(EventTurnStarted, map[string]any{
	116	+	if err := updateAgentFromEvent(a, NewEvent(EventTurnStarted, map[string]any{
112	117		"turn_id": "turn-2", "step": float64(2), "lifetime_turns": 2, "current_run_turns": 1,
113		-	}))
114		-	updateAgentFromEvent(a, NewEvent(EventAgentIdle, map[string]any{"turn_id": "turn-2"}))
	118	+	})); err != nil {
	119	+	t.Fatalf("apply second turn started event: %v", err)
	120	+	}
	121	+	if err := updateAgentFromEvent(a, NewEvent(EventAgentIdle, map[string]any{"turn_id": "turn-2"})); err != nil {
	122	+	t.Fatalf("apply second idle event: %v", err)
	123	+	}
115	124		select {
116	125		case command := <-commands:
117	126		var payload TurnStartPayload

@@ -129,6 +138,9 @@ func TestResumeWithPromptQueuesFollowUpsForActiveWorker(t *testing.T) {
129	138		if persisted.ResumePrompt != secondFollowUp \|\| len(persisted.ResumeQueue) != 0 {
130	139		t.Fatalf("second follow-up persistence = prompt %q queue %d, want prompt %q and empty queue", persisted.ResumePrompt, len(persisted.ResumeQueue), secondFollowUp)
131	140		}
	141	+	if persisted.ResumePromptID == "" {
	142	+	t.Fatal("promoted follow-up lost its command ID")
	143	+	}
132	144		if err := f.Stop(a.ID); err != nil {
133	145		t.Fatal(err)
134	146		}

 	if persisted.ResumePrompt != "" || len(persisted.ResumeQueue) != 2 {
 		t.Fatalf("active follow-ups persisted as prompt %q queue %d, want empty prompt and queue 2", persisted.ResumePrompt, len(persisted.ResumeQueue))
 	}
+	if firstID, secondID := persisted.ResumeQueue[0].CommandID, persisted.ResumeQueue[1].CommandID; firstID == "" || secondID == "" || firstID == secondID {
+		t.Fatalf("queued follow-up command IDs = %q, %q; want distinct non-empty IDs", firstID, secondID)
+	}
-	updateAgentFromEvent(a, NewEvent(EventAgentIdle, map[string]any{"turn_id": "turn-1"}))
+	if err := updateAgentFromEvent(a, NewEvent(EventAgentIdle, map[string]any{"turn_id": "turn-1"})); err != nil {
+		t.Fatalf("apply first idle event: %v", err)
+	}
 	select {
 	case command := <-commands:
 		var payload TurnStartPayload
 		t.Fatalf("turn state after first queued delivery = %s, want %s", got, TurnQueued)
 	}
-	updateAgentFromEvent(a, NewEvent(EventTurnStarted, map[string]any{
+	if err := updateAgentFromEvent(a, NewEvent(EventTurnStarted, map[string]any{
 		"turn_id": "turn-2", "step": float64(2), "lifetime_turns": 2, "current_run_turns": 1,
-	}))
+	})); err != nil {
-	updateAgentFromEvent(a, NewEvent(EventAgentIdle, map[string]any{"turn_id": "turn-2"}))
+		t.Fatalf("apply second turn started event: %v", err)
+	}
+	if err := updateAgentFromEvent(a, NewEvent(EventAgentIdle, map[string]any{"turn_id": "turn-2"})); err != nil {
+		t.Fatalf("apply second idle event: %v", err)
+	}
 	select {
 	case command := <-commands:
 		var payload TurnStartPayload
 	if persisted.ResumePrompt != secondFollowUp || len(persisted.ResumeQueue) != 0 {
 		t.Fatalf("second follow-up persistence = prompt %q queue %d, want prompt %q and empty queue", persisted.ResumePrompt, len(persisted.ResumeQueue), secondFollowUp)
 	}
+	if persisted.ResumePromptID == "" {
+		t.Fatal("promoted follow-up lost its command ID")
+	}
 	if err := f.Stop(a.ID); err != nil {
 		t.Fatal(err)
 	}

M packages/agent/subagents/runner.go +34 -3


@@ -6,6 +6,7 @@ import (
6	6		"context"
7	7		"encoding/json"
8	8		"fmt"
	9	+	"math"
9	10		"os"
10	11		"os/exec"
11	12		"path/filepath"

@@ -614,7 +615,11 @@ func eventMatchesPendingResume(a *Agent, ev Event) bool {
614	615		}
615	616		commandID, _ := ev.Data["command_id"].(string)
616	617		if commandID == "" {
617		-	return true // protocol-v1 events predate command identities
	618	+	// Protocol-v1 turn.started events predate command identities, but a
	619	+	// commandless rejection cannot be safely associated with the current
	620	+	// pending prompt and must not consume it.
	621	+	code, _ := ev.Data["code"].(string)
	622	+	return code != "turn_rejected"
618	623		}
619	624		expected := a.resumePromptCommandID()
620	625		return expected == "" \|\| expected == commandID

@@ -625,16 +630,42 @@ func eventCounter(data map[string]any, key string) (int, bool) {
625	630		if !ok {
626	631		return 0, false
627	632		}
	633	+	maxInt := int(^uint(0) >> 1)
	634	+	valid := func(value int64) (int, bool) {
	635	+	if value < 0 \|\| value > int64(maxInt) {
	636	+	return 0, false
	637	+	}
	638	+	return int(value), true
	639	+	}
628	640		switch value := value.(type) {
629	641		case int:
	642	+	if value < 0 {
	643	+	return 0, false
	644	+	}
630	645		return value, true
631	646		case int64:
	647	+	return valid(value)
	648	+	case uint:
	649	+	if uint64(value) > uint64(maxInt) {
	650	+	return 0, false
	651	+	}
	652	+	return int(value), true
	653	+	case uint64:
	654	+	if value > uint64(maxInt) {
	655	+	return 0, false
	656	+	}
632	657		return int(value), true
633	658		case float64:
634		-	return int(value), value == float64(int(value))
	659	+	if value < 0 \|\| math.Trunc(value) != value \|\| value >= float64(maxInt)+1 {
	660	+	return 0, false
	661	+	}
	662	+	return int(value), true
635	663		case json.Number:
636	664		parsed, err := value.Int64()
637		-	return int(parsed), err == nil
	665	+	if err != nil {
	666	+	return 0, false
	667	+	}
	668	+	return valid(parsed)
638	669		default:
639	670		return 0, false
640	671		}


@@ -6,6 +6,7 @@ import (
6		"context"	6		"context"
7		"encoding/json"	7		"encoding/json"
8		"fmt"	8		"fmt"
			9	+	"math"
9		"os"	10		"os"
10		"os/exec"	11		"os/exec"
11		"path/filepath"	12		"path/filepath"

@@ -614,7 +615,11 @@ func eventMatchesPendingResume(a *Agent, ev Event) bool {
614		}	615		}
615		commandID, _ := ev.Data["command_id"].(string)	616		commandID, _ := ev.Data["command_id"].(string)
616		if commandID == "" {	617		if commandID == "" {
617	-	return true // protocol-v1 events predate command identities	618	+	// Protocol-v1 turn.started events predate command identities, but a
			619	+	// commandless rejection cannot be safely associated with the current
			620	+	// pending prompt and must not consume it.
			621	+	code, _ := ev.Data["code"].(string)
			622	+	return code != "turn_rejected"
618		}	623		}
619		expected := a.resumePromptCommandID()	624		expected := a.resumePromptCommandID()
620		return expected == "" \|\| expected == commandID	625		return expected == "" \|\| expected == commandID

@@ -625,16 +630,42 @@ func eventCounter(data map[string]any, key string) (int, bool) {
625		if !ok {	630		if !ok {
626		return 0, false	631		return 0, false
627		}	632		}
			633	+	maxInt := int(^uint(0) >> 1)
			634	+	valid := func(value int64) (int, bool) {
			635	+	if value < 0 \|\| value > int64(maxInt) {
			636	+	return 0, false
			637	+	}
			638	+	return int(value), true
			639	+	}
628		switch value := value.(type) {	640		switch value := value.(type) {
629		case int:	641		case int:
			642	+	if value < 0 {
			643	+	return 0, false
			644	+	}
630		return value, true	645		return value, true
631		case int64:	646		case int64:
			647	+	return valid(value)
			648	+	case uint:
			649	+	if uint64(value) > uint64(maxInt) {
			650	+	return 0, false
			651	+	}
			652	+	return int(value), true
			653	+	case uint64:
			654	+	if value > uint64(maxInt) {
			655	+	return 0, false
			656	+	}
632		return int(value), true	657		return int(value), true
633		case float64:	658		case float64:
634	-	return int(value), value == float64(int(value))	659	+	if value < 0 \|\| math.Trunc(value) != value \|\| value >= float64(maxInt)+1 {
			660	+	return 0, false
			661	+	}
			662	+	return int(value), true
635		case json.Number:	663		case json.Number:
636		parsed, err := value.Int64()	664		parsed, err := value.Int64()
637	-	return int(parsed), err == nil	665	+	if err != nil {
			666	+	return 0, false
			667	+	}
			668	+	return valid(parsed)
638		default:	669		default:
639		return 0, false	670		return 0, false
640		}	671		}

M packages/agent/subagents/runner_test.go +59 -2


@@ -423,8 +423,9 @@ func TestUpdateAgentFromEventRecoversRejectedFollowUp(t *testing.T) {
423	423		a.setResumePrompt("retry me", time.Now())
424	424		a.setTurnState(TurnQueued, "turn-1")
425	425		if err := updateAgentFromEvent(a, NewEvent("error", map[string]any{
426		-	"code": "turn_rejected",
427		-	"reason": "busy",
	426	+	"code": "turn_rejected",
	427	+	"reason": "busy",
	428	+	"command_id": a.resumePromptCommandID(),
428	429		})); err != nil {
429	430		t.Fatal(err)
430	431		}

@@ -590,6 +591,62 @@ func TestUpdateAgentFromEventPersistsCanonicalTurnCounters(t *testing.T) {
590	591		}
591	592		}
592	593
	594	+	func TestRecordPersistenceErrorPreservesFirstFailure(t *testing.T) {
	595	+	a := &Agent{}
	596	+	a.recordPersistenceError(fmt.Errorf("first failure"))
	597	+	a.recordPersistenceError(fmt.Errorf("later failure"))
	598	+	if got := a.Snapshot().Err; !strings.Contains(got, "first failure") \|\| strings.Contains(got, "later failure") {
	599	+	t.Fatalf("recorded persistence error = %q, want only first failure", got)
	600	+	}
	601	+	}
	602	+
	603	+	func TestRestoreResumePromptPreservesCommandIdentity(t *testing.T) {
	604	+	a := &Agent{}
	605	+	a.setResumePrompt("old prompt", time.Now())
	606	+	oldID := a.resumePromptCommandID()
	607	+	previous := a.setResumePrompt("new prompt", time.Now())
	608	+	a.restoreResumePrompt(previous)
	609	+	if got := a.resumePromptCommandID(); got != oldID {
	610	+	t.Fatalf("restored command ID = %q, want %q", got, oldID)
	611	+	}
	612	+	}
	613	+
	614	+	func TestEventMatchesPendingResumeRejectsIdentityLessRejection(t *testing.T) {
	615	+	a := &Agent{resumePromptID: "command-1"}
	616	+	if eventMatchesPendingResume(a, Event{Type: "error", Data: map[string]any{"code": "turn_rejected"}}) {
	617	+	t.Fatal("identity-less turn rejection matched a pending prompt")
	618	+	}
	619	+	if !eventMatchesPendingResume(a, Event{Type: EventTurnStarted, Data: map[string]any{}}) {
	620	+	t.Fatal("identity-less legacy turn.started did not match a pending prompt")
	621	+	}
	622	+	if !eventMatchesPendingResume(a, Event{Type: "error", Data: map[string]any{"code": "turn_rejected", "command_id": "command-1"}}) {
	623	+	t.Fatal("matching turn rejection did not match a pending prompt")
	624	+	}
	625	+	}
	626	+
	627	+	func TestEventCounterRejectsNegativeFractionalAndOverflowValues(t *testing.T) {
	628	+	cases := []struct {
	629	+	name string
	630	+	value any
	631	+	ok bool
	632	+	}{
	633	+	{name: "valid integer", value: float64(3), ok: true},
	634	+	{name: "negative integer", value: int64(-1)},
	635	+	{name: "fractional float", value: 1.5},
	636	+	{name: "negative float", value: -1.0},
	637	+	{name: "fractional json number", value: json.Number("1.5")},
	638	+	{name: "overflow float", value: float64(^uint(0) >> 1)},
	639	+	}
	640	+	for _, tc := range cases {
	641	+	t.Run(tc.name, func(t *testing.T) {
	642	+	got, ok := eventCounter(map[string]any{"counter": tc.value}, "counter")
	643	+	if ok != tc.ok {
	644	+	t.Fatalf("eventCounter ok = %t, want %t (value %v, got %d)", ok, tc.ok, tc.value, got)
	645	+	}
	646	+	})
	647	+	}
	648	+	}
	649	+
593	650		func indexOf(xs []string, want string) int {
594	651		for i, x := range xs {
595	652		if x == want {


@@ -423,8 +423,9 @@ func TestUpdateAgentFromEventRecoversRejectedFollowUp(t *testing.T) {
423		a.setResumePrompt("retry me", time.Now())	423		a.setResumePrompt("retry me", time.Now())
424		a.setTurnState(TurnQueued, "turn-1")	424		a.setTurnState(TurnQueued, "turn-1")
425		if err := updateAgentFromEvent(a, NewEvent("error", map[string]any{	425		if err := updateAgentFromEvent(a, NewEvent("error", map[string]any{
426	-	"code": "turn_rejected",	426	+	"code": "turn_rejected",
427	-	"reason": "busy",	427	+	"reason": "busy",
			428	+	"command_id": a.resumePromptCommandID(),
428		})); err != nil {	429		})); err != nil {
429		t.Fatal(err)	430		t.Fatal(err)
430		}	431		}

@@ -590,6 +591,62 @@ func TestUpdateAgentFromEventPersistsCanonicalTurnCounters(t *testing.T) {
590		}	591		}
591		}	592		}
592			593
			594	+	func TestRecordPersistenceErrorPreservesFirstFailure(t *testing.T) {
			595	+	a := &Agent{}
			596	+	a.recordPersistenceError(fmt.Errorf("first failure"))
			597	+	a.recordPersistenceError(fmt.Errorf("later failure"))
			598	+	if got := a.Snapshot().Err; !strings.Contains(got, "first failure") \|\| strings.Contains(got, "later failure") {
			599	+	t.Fatalf("recorded persistence error = %q, want only first failure", got)
			600	+	}
			601	+	}
			602	+
			603	+	func TestRestoreResumePromptPreservesCommandIdentity(t *testing.T) {
			604	+	a := &Agent{}
			605	+	a.setResumePrompt("old prompt", time.Now())
			606	+	oldID := a.resumePromptCommandID()
			607	+	previous := a.setResumePrompt("new prompt", time.Now())
			608	+	a.restoreResumePrompt(previous)
			609	+	if got := a.resumePromptCommandID(); got != oldID {
			610	+	t.Fatalf("restored command ID = %q, want %q", got, oldID)
			611	+	}
			612	+	}
			613	+
			614	+	func TestEventMatchesPendingResumeRejectsIdentityLessRejection(t *testing.T) {
			615	+	a := &Agent{resumePromptID: "command-1"}
			616	+	if eventMatchesPendingResume(a, Event{Type: "error", Data: map[string]any{"code": "turn_rejected"}}) {
			617	+	t.Fatal("identity-less turn rejection matched a pending prompt")
			618	+	}
			619	+	if !eventMatchesPendingResume(a, Event{Type: EventTurnStarted, Data: map[string]any{}}) {
			620	+	t.Fatal("identity-less legacy turn.started did not match a pending prompt")
			621	+	}
			622	+	if !eventMatchesPendingResume(a, Event{Type: "error", Data: map[string]any{"code": "turn_rejected", "command_id": "command-1"}}) {
			623	+	t.Fatal("matching turn rejection did not match a pending prompt")
			624	+	}
			625	+	}
			626	+
			627	+	func TestEventCounterRejectsNegativeFractionalAndOverflowValues(t *testing.T) {
			628	+	cases := []struct {
			629	+	name string
			630	+	value any
			631	+	ok bool
			632	+	}{
			633	+	{name: "valid integer", value: float64(3), ok: true},
			634	+	{name: "negative integer", value: int64(-1)},
			635	+	{name: "fractional float", value: 1.5},
			636	+	{name: "negative float", value: -1.0},
			637	+	{name: "fractional json number", value: json.Number("1.5")},
			638	+	{name: "overflow float", value: float64(^uint(0) >> 1)},
			639	+	}
			640	+	for _, tc := range cases {
			641	+	t.Run(tc.name, func(t *testing.T) {
			642	+	got, ok := eventCounter(map[string]any{"counter": tc.value}, "counter")
			643	+	if ok != tc.ok {
			644	+	t.Fatalf("eventCounter ok = %t, want %t (value %v, got %d)", ok, tc.ok, tc.value, got)
			645	+	}
			646	+	})
			647	+	}
			648	+	}
			649	+
593		func indexOf(xs []string, want string) int {	650		func indexOf(xs []string, want string) int {
594		for i, x := range xs {	651		for i, x := range xs {
595		if x == want {	652		if x == want {

M packages/agent/subagents/supervisor.go +1 -1


@@ -543,7 +543,7 @@ func (f Supervisor) SpawnReq(ctx context.Context, req SpawnRequest) (Agent, er
543	543		}
544	544		a.ctx, a.cancel = runCtx, cancel
545	545		a.persistFn = f.persistAgent
546		-	a.setOnTurnIdle(func() { f.dispatchQueuedResume(a) })
	546	+	a.setOnTurnIdle(func() { f.dispatchQueuedResumeWithTimeout(a) })
547	547		a.workspaceCleanup = func() error { return workspace.Cleanup(context.Background()) }
548	548		a.workspaceCapture = func() (WorkspaceCapture, error) { return workspace.Capture(context.Background()) }
549	549		a.runner = f.cfg.NewRunner(a)


@@ -543,7 +543,7 @@ func (f Supervisor) SpawnReq(ctx context.Context, req SpawnRequest) (Agent, er
543		}	543		}
544		a.ctx, a.cancel = runCtx, cancel	544		a.ctx, a.cancel = runCtx, cancel
545		a.persistFn = f.persistAgent	545		a.persistFn = f.persistAgent
546	-	a.setOnTurnIdle(func() { f.dispatchQueuedResume(a) })	546	+	a.setOnTurnIdle(func() { f.dispatchQueuedResumeWithTimeout(a) })
547		a.workspaceCleanup = func() error { return workspace.Cleanup(context.Background()) }	547		a.workspaceCleanup = func() error { return workspace.Cleanup(context.Background()) }
548		a.workspaceCapture = func() (WorkspaceCapture, error) { return workspace.Capture(context.Background()) }	548		a.workspaceCapture = func() (WorkspaceCapture, error) { return workspace.Capture(context.Background()) }
549		a.runner = f.cfg.NewRunner(a)	549		a.runner = f.cfg.NewRunner(a)

M packages/agent/subagents/turn_classification.go +0 -5


@@ -25,11 +25,6 @@ func classifyTurnEvent(ev Event) turnEventClass {
25	25		case "turn_start":
26	26		return turnEventNested
27	27		case EventTurnStarted:
28		-	if _, lifetimeOK := eventCounter(ev.Data, "lifetime_turns"); lifetimeOK {
29		-	if _, currentOK := eventCounter(ev.Data, "current_run_turns"); currentOK {
30		-	return turnEventDelegated
31		-	}
32		-	}
33	28		// v1 turn.started without counters is the delegated boundary. A
34	29		// current worker marks nested canonical events explicitly above.
35	30		return turnEventDelegated


@@ -25,11 +25,6 @@ func classifyTurnEvent(ev Event) turnEventClass {
25		case "turn_start":	25	case "turn_start":
26		return turnEventNested	26	return turnEventNested
27		case EventTurnStarted:	27	case EventTurnStarted:
28	-	if _, lifetimeOK := eventCounter(ev.Data, "lifetime_turns"); lifetimeOK {
29	-	if _, currentOK := eventCounter(ev.Data, "current_run_turns"); currentOK {
30	-	return turnEventDelegated
31	-	}
32	-	}
33		// v1 turn.started without counters is the delegated boundary. A	28	// v1 turn.started without counters is the delegated boundary. A
34		// current worker marks nested canonical events explicitly above.	29	// current worker marks nested canonical events explicitly above.
35		return turnEventDelegated	30	return turnEventDelegated

M packages/agent/tools/subagent_resume.go +1 -1


@@ -50,7 +50,7 @@ const subagentResumeSchema = `{
50	50		func (t *SubagentResumeTool) Name() string { return "subagent_resume" }
51	51
52	52		func (t *SubagentResumeTool) Description() string {
53		-	return "Start a fresh-budget follow-up for a sub-agent, queue it if the worker is active, or restart a stopped one with its prior session context preserved."
	53	+	return "Continue a sub-agent with a fresh turn budget and its retained session; deliver immediately when idle, queue while active, or restart a stopped worker."
54	54		}
55	55
56	56		func (t *SubagentResumeTool) Schema() json.RawMessage {


@@ -50,7 +50,7 @@ const subagentResumeSchema = `{
50		func (t *SubagentResumeTool) Name() string { return "subagent_resume" }	50		func (t *SubagentResumeTool) Name() string { return "subagent_resume" }
51			51
52		func (t *SubagentResumeTool) Description() string {	52		func (t *SubagentResumeTool) Description() string {
53	-	return "Start a fresh-budget follow-up for a sub-agent, queue it if the worker is active, or restart a stopped one with its prior session context preserved."	53	+	return "Continue a sub-agent with a fresh turn budget and its retained session; deliver immediately when idle, queue while active, or restart a stopped worker."
54		}	54		}
55			55
56		func (t *SubagentResumeTool) Schema() json.RawMessage {	56		func (t *SubagentResumeTool) Schema() json.RawMessage {

M packages/core/session.go +3 -0


@@ -1173,6 +1173,9 @@ func (s *Session) Sync() error {
1173	1173		if s == nil \|\| s.writer == nil {
1174	1174		return nil
1175	1175		}
	1176	+	if err := s.buf.Flush(); err != nil {
	1177	+	return err
	1178	+	}
1176	1179		return s.writer.Sync()
1177	1180		}
1178	1181


@@ -1173,6 +1173,9 @@ func (s *Session) Sync() error {
1173	if s == nil \|\| s.writer == nil {	1173		if s == nil \|\| s.writer == nil {
1174	return nil	1174		return nil
1175	}	1175		}
		1176	+	if err := s.buf.Flush(); err != nil {
		1177	+	return err
		1178	+	}
1176	return s.writer.Sync()	1179		return s.writer.Sync()
1177	}	1180		}
1178		1181