fix(subagents): synchronize startup timeout readiness

commit: 5e23cb11058e1d1be0458177a4df413ee22afdc5
author: brice · 2mo ago · 2026-08-10 09:06 UTC
parent: 2bd8257
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

3 files · +109 -12

M packages/agent/subagents/agent.go +24 -0


@@ -361,6 +361,30 @@ func (a *Agent) setProcessState(state ProcessState) {
361	361		a.lifecycleMu.Unlock()
362	362		}
363	363
	364	+	// markStartupTimeout atomically claims a worker that is still waiting for
	365	+	// agent_ready. Process-state transitions use lifecycleMu too, so a ready
	366	+	// worker cannot be timed out based on a stale ProcessStarting observation.
	367	+	func (a *Agent) markStartupTimeout(cause error) bool {
	368	+	if cause == nil {
	369	+	return false
	370	+	}
	371	+	a.lifecycleMu.Lock()
	372	+	defer a.lifecycleMu.Unlock()
	373	+	if a.processState != ProcessStarting {
	374	+	return false
	375	+	}
	376	+	a.mu.Lock()
	377	+	defer a.mu.Unlock()
	378	+	if a.status != StatusRunning {
	379	+	return false
	380	+	}
	381	+	a.startupErr = cause
	382	+	a.lastErr = cause
	383	+	a.activity = "startup timeout: worker did not become ready"
	384	+	a.updatedAt = time.Now()
	385	+	return true
	386	+	}
	387	+
364	388		func (a *Agent) incrementAttempt() int {
365	389		a.lifecycleMu.Lock()
366	390		a.Attempt++


@@ -361,6 +361,30 @@ func (a *Agent) setProcessState(state ProcessState) {
361	a.lifecycleMu.Unlock()	361		a.lifecycleMu.Unlock()
362	}	362		}
363		363
		364	+	// markStartupTimeout atomically claims a worker that is still waiting for
		365	+	// agent_ready. Process-state transitions use lifecycleMu too, so a ready
		366	+	// worker cannot be timed out based on a stale ProcessStarting observation.
		367	+	func (a *Agent) markStartupTimeout(cause error) bool {
		368	+	if cause == nil {
		369	+	return false
		370	+	}
		371	+	a.lifecycleMu.Lock()
		372	+	defer a.lifecycleMu.Unlock()
		373	+	if a.processState != ProcessStarting {
		374	+	return false
		375	+	}
		376	+	a.mu.Lock()
		377	+	defer a.mu.Unlock()
		378	+	if a.status != StatusRunning {
		379	+	return false
		380	+	}
		381	+	a.startupErr = cause
		382	+	a.lastErr = cause
		383	+	a.activity = "startup timeout: worker did not become ready"
		384	+	a.updatedAt = time.Now()
		385	+	return true
		386	+	}
		387	+
364	func (a *Agent) incrementAttempt() int {	388		func (a *Agent) incrementAttempt() int {
365	a.lifecycleMu.Lock()	389		a.lifecycleMu.Lock()
366	a.Attempt++	390		a.Attempt++

M packages/agent/subagents/scheduler.go +2 -8


@@ -337,19 +337,13 @@ func (f Supervisor) armStartupTimeout(a Agent) {
337	337		}
338	338
339	339		func (f Supervisor) expireStarting(a Agent, d time.Duration) {
340		-	if a == nil \|\| a.ProcessState() != ProcessStarting {
	340	+	if a == nil {
341	341		return
342	342		}
343	343		cause := fmt.Errorf("%w after %s waiting for agent_ready; inspect the worker output and retry", errStartupTimeout, d)
344		-	a.mu.Lock()
345		-	if a.status != StatusRunning {
346		-	a.mu.Unlock()
	344	+	if !a.markStartupTimeout(cause) {
347	345		return
348	346		}
349		-	a.startupErr = cause
350		-	a.lastErr = cause
351		-	a.activity = "startup timeout: worker did not become ready"
352		-	a.mu.Unlock()
353	347		f.persistAgent(a)
354	348		if a.cancel != nil {
355	349		a.cancel()

 }
 func (f *Supervisor) expireStarting(a *Agent, d time.Duration) {
-	if a == nil || a.ProcessState() != ProcessStarting {
+	if a == nil {
 		return
 	}
 	cause := fmt.Errorf("%w after %s waiting for agent_ready; inspect the worker output and retry", errStartupTimeout, d)
-	a.mu.Lock()
+	if !a.markStartupTimeout(cause) {
-	if a.status != StatusRunning {
-		a.mu.Unlock()
 		return
 	}
-	a.startupErr = cause
-	a.lastErr = cause
-	a.activity = "startup timeout: worker did not become ready"
-	a.mu.Unlock()
 	f.persistAgent(a)
 	if a.cancel != nil {
 		a.cancel()

M packages/agent/subagents/supervisor_test.go +83 -4


@@ -114,13 +114,16 @@ func TestSpawnRunsAndCompletes(t *testing.T) {
114	114		func TestStartupTimeoutFailsWorkerThatNeverBecomesReady(t *testing.T) {
115	115		root := t.TempDir()
116	116		started := make(chan struct{})
	117	+	cancelled := make(chan struct{})
	118	+	startupTimeout := time.Hour
117	119		f := New(Config{
118	120		Root: root, RepoRoot: root,
119		-	Policy: SubagentPolicy{StartupTimeout: 20 * time.Millisecond},
	121	+	Policy: SubagentPolicy{StartupTimeout: startupTimeout},
120	122		NewRunner: func(*Agent) Runner {
121	123		return RunnerFunc(func(ctx context.Context, _ Sink) error {
122	124		close(started)
123	125		<-ctx.Done()
	126	+	close(cancelled)
124	127		return ctx.Err()
125	128		})
126	129		},

@@ -136,6 +139,12 @@ func TestStartupTimeoutFailsWorkerThatNeverBecomesReady(t *testing.T) {
136	139		case <-time.After(time.Second):
137	140		t.Fatal("runner did not start")
138	141		}
	142	+	f.expireStarting(a, startupTimeout)
	143	+	select {
	144	+	case <-cancelled:
	145	+	case <-time.After(time.Second):
	146	+	t.Fatal("startup timeout did not cancel runner")
	147	+	}
139	148		waitCtx, cancel := context.WithTimeout(context.Background(), time.Second)
140	149		defer cancel()
141	150		if err := a.WaitContext(waitCtx); err != nil {

@@ -150,18 +159,78 @@ func TestStartupTimeoutFailsWorkerThatNeverBecomesReady(t *testing.T) {
150	159		if got := a.Activity(); !strings.Contains(got, "startup timeout") {
151	160		t.Fatalf("activity = %q, want startup timeout explanation", got)
152	161		}
	162	+	result, err := f.ReadResult(a.ID)
	163	+	if err != nil {
	164	+	t.Fatal(err)
	165	+	}
	166	+	if result.Status != ResultFailed \|\| result.Error == nil {
	167	+	t.Fatalf("startup timeout result = %#v, want failed result with error", result)
	168	+	}
	169	+	if !strings.Contains(result.Error.Message, errStartupTimeout.Error()) \|\| !strings.Contains(result.Error.Message, "inspect the worker output and retry") {
	170	+	t.Fatalf("startup timeout result error = %#v, want timeout and retry guidance", result.Error)
	171	+	}
	172	+	}
	173	+
	174	+	func TestStartupTimeoutDoesNotCancelReadyWorker(t *testing.T) {
	175	+	root := t.TempDir()
	176	+	started := make(chan struct{})
	177	+	cancelled := make(chan struct{})
	178	+	f := New(Config{
	179	+	Root: root, RepoRoot: root,
	180	+	Policy: SubagentPolicy{StartupTimeout: time.Hour},
	181	+	NewRunner: func(*Agent) Runner {
	182	+	return RunnerFunc(func(ctx context.Context, _ Sink) error {
	183	+	close(started)
	184	+	<-ctx.Done()
	185	+	close(cancelled)
	186	+	return ctx.Err()
	187	+	})
	188	+	},
	189	+	})
	190	+	t.Cleanup(f.StopAll)
	191	+
	192	+	a, err := f.Spawn(context.Background(), "ready startup")
	193	+	if err != nil {
	194	+	t.Fatal(err)
	195	+	}
	196	+	select {
	197	+	case <-started:
	198	+	case <-time.After(time.Second):
	199	+	t.Fatal("runner did not start")
	200	+	}
	201	+	if err := updateAgentFromEvent(a, NewEvent(EventAgentReady, nil)); err != nil {
	202	+	t.Fatal(err)
	203	+	}
	204	+	f.expireStarting(a, time.Hour)
	205	+
	206	+	if got := a.ProcessState(); got != ProcessAlive {
	207	+	t.Fatalf("process state = %s, want %s", got, ProcessAlive)
	208	+	}
	209	+	if err := a.Err(); err != nil {
	210	+	t.Fatalf("ready worker error = %v, want nil", err)
	211	+	}
	212	+	select {
	213	+	case <-cancelled:
	214	+	t.Fatal("startup timeout canceled ready worker")
	215	+	default:
	216	+	}
	217	+	if err := f.Stop(a.ID); err != nil {
	218	+	t.Fatal(err)
	219	+	}
	220	+	a.Wait()
153	221		}
154	222
155	223		func TestQueueTimeoutFailsWorkerWaitingForExecutionSlot(t *testing.T) {
156	224		root := t.TempDir()
157	225		started := make(chan struct{})
	226	+	queueTimeout := time.Hour
158	227		var startOnce sync.Once
159	228		f := New(Config{
160	229		Root: root, RepoRoot: root,
161	230		Policy: SubagentPolicy{
162	231		MaxConcurrent: 1,
163		-	QueueTimeout: 20 * time.Millisecond,
164		-	StartupTimeout: time.Second,
	232	+	QueueTimeout: queueTimeout,
	233	+	StartupTimeout: time.Hour,
165	234		},
166	235		NewRunner: func(*Agent) Runner {
167	236		return RunnerFunc(func(ctx context.Context, _ Sink) error {

@@ -186,6 +255,7 @@ func TestQueueTimeoutFailsWorkerWaitingForExecutionSlot(t *testing.T) {
186	255		if err != nil {
187	256		t.Fatal(err)
188	257		}
	258	+	f.expireQueued(queued.ID)
189	259		waitCtx, cancel := context.WithTimeout(context.Background(), time.Second)
190	260		defer cancel()
191	261		if err := queued.WaitContext(waitCtx); err != nil {

@@ -200,7 +270,16 @@ func TestQueueTimeoutFailsWorkerWaitingForExecutionSlot(t *testing.T) {
200	270		if got := queued.Err(); got == nil \|\| !strings.Contains(got.Error(), "waiting for an execution slot") {
201	271		t.Fatalf("queued worker error = %v, want queue-pressure explanation", got)
202	272		}
203		-	f.Stop(first.ID)
	273	+	result, err := f.ReadResult(queued.ID)
	274	+	if err != nil {
	275	+	t.Fatal(err)
	276	+	}
	277	+	if result.Status != ResultFailed \|\| result.Error == nil \|\| !strings.Contains(result.Error.Message, "waiting for an execution slot") {
	278	+	t.Fatalf("queued worker result = %#v, want failed result with queue-pressure error", result)
	279	+	}
	280	+	if err := f.Stop(first.ID); err != nil {
	281	+	t.Fatal(err)
	282	+	}
204	283		first.Wait()
205	284		}
206	285


@@ -114,13 +114,16 @@ func TestSpawnRunsAndCompletes(t *testing.T) {
114		func TestStartupTimeoutFailsWorkerThatNeverBecomesReady(t *testing.T) {	114		func TestStartupTimeoutFailsWorkerThatNeverBecomesReady(t *testing.T) {
115		root := t.TempDir()	115		root := t.TempDir()
116		started := make(chan struct{})	116		started := make(chan struct{})
			117	+	cancelled := make(chan struct{})
			118	+	startupTimeout := time.Hour
117		f := New(Config{	119		f := New(Config{
118		Root: root, RepoRoot: root,	120		Root: root, RepoRoot: root,
119	-	Policy: SubagentPolicy{StartupTimeout: 20 * time.Millisecond},	121	+	Policy: SubagentPolicy{StartupTimeout: startupTimeout},
120		NewRunner: func(*Agent) Runner {	122		NewRunner: func(*Agent) Runner {
121		return RunnerFunc(func(ctx context.Context, _ Sink) error {	123		return RunnerFunc(func(ctx context.Context, _ Sink) error {
122		close(started)	124		close(started)
123		<-ctx.Done()	125		<-ctx.Done()
			126	+	close(cancelled)
124		return ctx.Err()	127		return ctx.Err()
125		})	128		})
126		},	129		},

@@ -136,6 +139,12 @@ func TestStartupTimeoutFailsWorkerThatNeverBecomesReady(t *testing.T) {
136		case <-time.After(time.Second):	139		case <-time.After(time.Second):
137		t.Fatal("runner did not start")	140		t.Fatal("runner did not start")
138		}	141		}
			142	+	f.expireStarting(a, startupTimeout)
			143	+	select {
			144	+	case <-cancelled:
			145	+	case <-time.After(time.Second):
			146	+	t.Fatal("startup timeout did not cancel runner")
			147	+	}
139		waitCtx, cancel := context.WithTimeout(context.Background(), time.Second)	148		waitCtx, cancel := context.WithTimeout(context.Background(), time.Second)
140		defer cancel()	149		defer cancel()
141		if err := a.WaitContext(waitCtx); err != nil {	150		if err := a.WaitContext(waitCtx); err != nil {

@@ -150,18 +159,78 @@ func TestStartupTimeoutFailsWorkerThatNeverBecomesReady(t *testing.T) {
150		if got := a.Activity(); !strings.Contains(got, "startup timeout") {	159		if got := a.Activity(); !strings.Contains(got, "startup timeout") {
151		t.Fatalf("activity = %q, want startup timeout explanation", got)	160		t.Fatalf("activity = %q, want startup timeout explanation", got)
152		}	161		}
			162	+	result, err := f.ReadResult(a.ID)
			163	+	if err != nil {
			164	+	t.Fatal(err)
			165	+	}
			166	+	if result.Status != ResultFailed \|\| result.Error == nil {
			167	+	t.Fatalf("startup timeout result = %#v, want failed result with error", result)
			168	+	}
			169	+	if !strings.Contains(result.Error.Message, errStartupTimeout.Error()) \|\| !strings.Contains(result.Error.Message, "inspect the worker output and retry") {
			170	+	t.Fatalf("startup timeout result error = %#v, want timeout and retry guidance", result.Error)
			171	+	}
			172	+	}
			173	+
			174	+	func TestStartupTimeoutDoesNotCancelReadyWorker(t *testing.T) {
			175	+	root := t.TempDir()
			176	+	started := make(chan struct{})
			177	+	cancelled := make(chan struct{})
			178	+	f := New(Config{
			179	+	Root: root, RepoRoot: root,
			180	+	Policy: SubagentPolicy{StartupTimeout: time.Hour},
			181	+	NewRunner: func(*Agent) Runner {
			182	+	return RunnerFunc(func(ctx context.Context, _ Sink) error {
			183	+	close(started)
			184	+	<-ctx.Done()
			185	+	close(cancelled)
			186	+	return ctx.Err()
			187	+	})
			188	+	},
			189	+	})
			190	+	t.Cleanup(f.StopAll)
			191	+
			192	+	a, err := f.Spawn(context.Background(), "ready startup")
			193	+	if err != nil {
			194	+	t.Fatal(err)
			195	+	}
			196	+	select {
			197	+	case <-started:
			198	+	case <-time.After(time.Second):
			199	+	t.Fatal("runner did not start")
			200	+	}
			201	+	if err := updateAgentFromEvent(a, NewEvent(EventAgentReady, nil)); err != nil {
			202	+	t.Fatal(err)
			203	+	}
			204	+	f.expireStarting(a, time.Hour)
			205	+
			206	+	if got := a.ProcessState(); got != ProcessAlive {
			207	+	t.Fatalf("process state = %s, want %s", got, ProcessAlive)
			208	+	}
			209	+	if err := a.Err(); err != nil {
			210	+	t.Fatalf("ready worker error = %v, want nil", err)
			211	+	}
			212	+	select {
			213	+	case <-cancelled:
			214	+	t.Fatal("startup timeout canceled ready worker")
			215	+	default:
			216	+	}
			217	+	if err := f.Stop(a.ID); err != nil {
			218	+	t.Fatal(err)
			219	+	}
			220	+	a.Wait()
153		}	221		}
154			222
155		func TestQueueTimeoutFailsWorkerWaitingForExecutionSlot(t *testing.T) {	223		func TestQueueTimeoutFailsWorkerWaitingForExecutionSlot(t *testing.T) {
156		root := t.TempDir()	224		root := t.TempDir()
157		started := make(chan struct{})	225		started := make(chan struct{})
			226	+	queueTimeout := time.Hour
158		var startOnce sync.Once	227		var startOnce sync.Once
159		f := New(Config{	228		f := New(Config{
160		Root: root, RepoRoot: root,	229		Root: root, RepoRoot: root,
161		Policy: SubagentPolicy{	230		Policy: SubagentPolicy{
162		MaxConcurrent: 1,	231		MaxConcurrent: 1,
163	-	QueueTimeout: 20 * time.Millisecond,	232	+	QueueTimeout: queueTimeout,
164	-	StartupTimeout: time.Second,	233	+	StartupTimeout: time.Hour,
165		},	234		},
166		NewRunner: func(*Agent) Runner {	235		NewRunner: func(*Agent) Runner {
167		return RunnerFunc(func(ctx context.Context, _ Sink) error {	236		return RunnerFunc(func(ctx context.Context, _ Sink) error {

@@ -186,6 +255,7 @@ func TestQueueTimeoutFailsWorkerWaitingForExecutionSlot(t *testing.T) {
186		if err != nil {	255		if err != nil {
187		t.Fatal(err)	256		t.Fatal(err)
188		}	257		}
			258	+	f.expireQueued(queued.ID)
189		waitCtx, cancel := context.WithTimeout(context.Background(), time.Second)	259		waitCtx, cancel := context.WithTimeout(context.Background(), time.Second)
190		defer cancel()	260		defer cancel()
191		if err := queued.WaitContext(waitCtx); err != nil {	261		if err := queued.WaitContext(waitCtx); err != nil {

@@ -200,7 +270,16 @@ func TestQueueTimeoutFailsWorkerWaitingForExecutionSlot(t *testing.T) {
200		if got := queued.Err(); got == nil \|\| !strings.Contains(got.Error(), "waiting for an execution slot") {	270		if got := queued.Err(); got == nil \|\| !strings.Contains(got.Error(), "waiting for an execution slot") {
201		t.Fatalf("queued worker error = %v, want queue-pressure explanation", got)	271		t.Fatalf("queued worker error = %v, want queue-pressure explanation", got)
202		}	272		}
203	-	f.Stop(first.ID)	273	+	result, err := f.ReadResult(queued.ID)
			274	+	if err != nil {
			275	+	t.Fatal(err)
			276	+	}
			277	+	if result.Status != ResultFailed \|\| result.Error == nil \|\| !strings.Contains(result.Error.Message, "waiting for an execution slot") {
			278	+	t.Fatalf("queued worker result = %#v, want failed result with queue-pressure error", result)
			279	+	}
			280	+	if err := f.Stop(first.ID); err != nil {
			281	+	t.Fatal(err)
			282	+	}
204		first.Wait()	283		first.Wait()
205		}	284		}
206			285