2bd8257 fix(subagents): bound worker startup and surface SSE failures · zut

M docs/subagents.md +1 -1


@@ -164,4 +164,4 @@ Use `subagent_stop` with an `agent_id` to request termination of a stuck worker
164	164
165	165		## Resource policy
166	166
167		-	The persisted `subagents` config object supports `max_concurrent`, `max_concurrent_per_parent`, `queue_timeout`, `default_timeout`, `max_turns`, output caps, allowed tools/roots, heartbeat and idle timeouts. `max_turns` bounds one active worker run; lifetime and current-run turn counters remain persisted and visible in worker status. An explicit `subagent_resume` starts a new run with a fresh budget without replaying the original task or losing session context. Previously saved `max_total_spawned` values are accepted but ignored. A missing or non-positive configured `max_turns` uses the default ceiling of 3. The `max_turns` field in `subagent_spawn` is optional: omit it to use the policy ceiling, or provide a value from 1 through the configured maximum. Concurrency limits apply to slash commands, `subagent_spawn`, and batch operations. A child cannot create descendants in v1. Per-agent timeouts are retained across reload/resume. Packaged `zut run` agents keep their declared capability ceiling by disabling subagent delegation; profiles are not a substitute for that permission boundary. A primary whose launch-time policy disables delegation still receives the strict orchestrator contract, but does not receive actionable profile metadata or a replacement implementation path.
	167	+	The persisted `subagents` config object supports `max_concurrent`, `max_concurrent_per_parent`, `queue_timeout`, `startup_timeout`, `default_timeout`, `max_turns`, output caps, allowed tools/roots, heartbeat and idle timeouts. A worker fails with an actionable queue-timeout error when no execution slot becomes available before `queue_timeout` (5 minutes by default). Once admitted, it must report `agent_ready` before `startup_timeout` (1 minute by default) or fails with guidance to inspect worker output and retry. `max_turns` bounds one active worker run; lifetime and current-run turn counters remain persisted and visible in worker status. An explicit `subagent_resume` starts a new run with a fresh budget without replaying the original task or losing session context. Previously saved `max_total_spawned` values are accepted but ignored. A missing or non-positive configured `max_turns` uses the default ceiling of 3. The `max_turns` field in `subagent_spawn` is optional: omit it to use the policy ceiling, or provide a value from 1 through the configured maximum. Concurrency limits apply to slash commands, `subagent_spawn`, and batch operations. A child cannot create descendants in v1. Per-agent timeouts are retained across reload/resume. Packaged `zut run` agents keep their declared capability ceiling by disabling subagent delegation; profiles are not a substitute for that permission boundary. A primary whose launch-time policy disables delegation still receives the strict orchestrator contract, but does not receive actionable profile metadata or a replacement implementation path.


@@ -164,4 +164,4 @@ Use `subagent_stop` with an `agent_id` to request termination of a stuck worker
164			164
165		## Resource policy	165		## Resource policy
166			166
167	-	The persisted `subagents` config object supports `max_concurrent`, `max_concurrent_per_parent`, `queue_timeout`, `default_timeout`, `max_turns`, output caps, allowed tools/roots, heartbeat and idle timeouts. `max_turns` bounds one active worker run; lifetime and current-run turn counters remain persisted and visible in worker status. An explicit `subagent_resume` starts a new run with a fresh budget without replaying the original task or losing session context. Previously saved `max_total_spawned` values are accepted but ignored. A missing or non-positive configured `max_turns` uses the default ceiling of 3. The `max_turns` field in `subagent_spawn` is optional: omit it to use the policy ceiling, or provide a value from 1 through the configured maximum. Concurrency limits apply to slash commands, `subagent_spawn`, and batch operations. A child cannot create descendants in v1. Per-agent timeouts are retained across reload/resume. Packaged `zut run` agents keep their declared capability ceiling by disabling subagent delegation; profiles are not a substitute for that permission boundary. A primary whose launch-time policy disables delegation still receives the strict orchestrator contract, but does not receive actionable profile metadata or a replacement implementation path.	167	+	The persisted `subagents` config object supports `max_concurrent`, `max_concurrent_per_parent`, `queue_timeout`, `startup_timeout`, `default_timeout`, `max_turns`, output caps, allowed tools/roots, heartbeat and idle timeouts. A worker fails with an actionable queue-timeout error when no execution slot becomes available before `queue_timeout` (5 minutes by default). Once admitted, it must report `agent_ready` before `startup_timeout` (1 minute by default) or fails with guidance to inspect worker output and retry. `max_turns` bounds one active worker run; lifetime and current-run turn counters remain persisted and visible in worker status. An explicit `subagent_resume` starts a new run with a fresh budget without replaying the original task or losing session context. Previously saved `max_total_spawned` values are accepted but ignored. A missing or non-positive configured `max_turns` uses the default ceiling of 3. The `max_turns` field in `subagent_spawn` is optional: omit it to use the policy ceiling, or provide a value from 1 through the configured maximum. Concurrency limits apply to slash commands, `subagent_spawn`, and batch operations. A child cannot create descendants in v1. Per-agent timeouts are retained across reload/resume. Packaged `zut run` agents keep their declared capability ceiling by disabling subagent delegation; profiles are not a substitute for that permission boundary. A primary whose launch-time policy disables delegation still receives the strict orchestrator contract, but does not receive actionable profile metadata or a replacement implementation path.

M packages/agent/config.go +1 -0


@@ -28,6 +28,7 @@ type SubagentsConfig struct {
28	28		MaxConcurrent int `json:"max_concurrent,omitempty"`
29	29		MaxConcurrentPerParent int `json:"max_concurrent_per_parent,omitempty"`
30	30		QueueTimeout string `json:"queue_timeout,omitempty"`
	31	+	StartupTimeout string `json:"startup_timeout,omitempty"`
31	32		DefaultTimeout string `json:"default_timeout,omitempty"`
32	33		MaxTurns int `json:"max_turns,omitempty"`
33	34		MaxOutputBytes int `json:"max_output_bytes,omitempty"`


@@ -28,6 +28,7 @@ type SubagentsConfig struct {
28	MaxConcurrent int `json:"max_concurrent,omitempty"`	28		MaxConcurrent int `json:"max_concurrent,omitempty"`
29	MaxConcurrentPerParent int `json:"max_concurrent_per_parent,omitempty"`	29		MaxConcurrentPerParent int `json:"max_concurrent_per_parent,omitempty"`
30	QueueTimeout string `json:"queue_timeout,omitempty"`	30		QueueTimeout string `json:"queue_timeout,omitempty"`
		31	+	StartupTimeout string `json:"startup_timeout,omitempty"`
31	DefaultTimeout string `json:"default_timeout,omitempty"`	32		DefaultTimeout string `json:"default_timeout,omitempty"`
32	MaxTurns int `json:"max_turns,omitempty"`	33		MaxTurns int `json:"max_turns,omitempty"`
33	MaxOutputBytes int `json:"max_output_bytes,omitempty"`	34		MaxOutputBytes int `json:"max_output_bytes,omitempty"`

M packages/agent/subagent_policy_test.go +8 -0


@@ -4,10 +4,18 @@ import (
4	4		"context"
5	5		"os"
6	6		"testing"
	7	+	"time"
7	8
8	9		"github.com/bnema/zut/packages/agent/subagents"
9	10		)
10	11
	12	+	func TestSubagentPolicyParsesStartupTimeout(t *testing.T) {
	13	+	policy := subagentPolicyFromConfig(SubagentsConfig{StartupTimeout: "42s"})
	14	+	if got := policy.StartupTimeout; got != 42*time.Second {
	15	+	t.Fatalf("StartupTimeout = %s, want 42s", got)
	16	+	}
	17	+	}
	18	+
11	19		func TestSubagentPolicyIgnoresLegacyMaxTotalSpawned(t *testing.T) {
12	20		t.Setenv("ZUT_HOME", t.TempDir())
13	21		if err := os.WriteFile(ConfigPath(), []byte(`{"subagents":{"max_concurrent":1,"max_total_spawned":1}}`), 0o600); err != nil {


@@ -4,10 +4,18 @@ import (
4	"context"	4		"context"
5	"os"	5		"os"
6	"testing"	6		"testing"
		7	+	"time"
7		8
8	"github.com/bnema/zut/packages/agent/subagents"	9		"github.com/bnema/zut/packages/agent/subagents"
9	)	10		)
10		11
		12	+	func TestSubagentPolicyParsesStartupTimeout(t *testing.T) {
		13	+	policy := subagentPolicyFromConfig(SubagentsConfig{StartupTimeout: "42s"})
		14	+	if got := policy.StartupTimeout; got != 42*time.Second {
		15	+	t.Fatalf("StartupTimeout = %s, want 42s", got)
		16	+	}
		17	+	}
		18	+
11	func TestSubagentPolicyIgnoresLegacyMaxTotalSpawned(t *testing.T) {	19		func TestSubagentPolicyIgnoresLegacyMaxTotalSpawned(t *testing.T) {
12	t.Setenv("ZUT_HOME", t.TempDir())	20		t.Setenv("ZUT_HOME", t.TempDir())
13	if err := os.WriteFile(ConfigPath(), []byte(`{"subagents":{"max_concurrent":1,"max_total_spawned":1}}`), 0o600); err != nil {	21		if err := os.WriteFile(ConfigPath(), []byte(`{"subagents":{"max_concurrent":1,"max_total_spawned":1}}`), 0o600); err != nil {

M packages/agent/subagent_runtime.go +1 -0


@@ -472,6 +472,7 @@ func subagentPolicyFromConfig(cfg SubagentsConfig) subagents.SubagentPolicy {
472	472		MaxConcurrent: cfg.MaxConcurrent,
473	473		MaxConcurrentPerParent: cfg.MaxConcurrentPerParent,
474	474		QueueTimeout: parseDuration(cfg.QueueTimeout),
	475	+	StartupTimeout: parseDuration(cfg.StartupTimeout),
475	476		DefaultTimeout: parseDuration(cfg.DefaultTimeout),
476	477		MaxTurns: cfg.MaxTurns,
477	478		MaxOutputBytes: cfg.MaxOutputBytes,


@@ -472,6 +472,7 @@ func subagentPolicyFromConfig(cfg SubagentsConfig) subagents.SubagentPolicy {
472	MaxConcurrent: cfg.MaxConcurrent,	472		MaxConcurrent: cfg.MaxConcurrent,
473	MaxConcurrentPerParent: cfg.MaxConcurrentPerParent,	473		MaxConcurrentPerParent: cfg.MaxConcurrentPerParent,
474	QueueTimeout: parseDuration(cfg.QueueTimeout),	474		QueueTimeout: parseDuration(cfg.QueueTimeout),
		475	+	StartupTimeout: parseDuration(cfg.StartupTimeout),
475	DefaultTimeout: parseDuration(cfg.DefaultTimeout),	476		DefaultTimeout: parseDuration(cfg.DefaultTimeout),
476	MaxTurns: cfg.MaxTurns,	477		MaxTurns: cfg.MaxTurns,
477	MaxOutputBytes: cfg.MaxOutputBytes,	478		MaxOutputBytes: cfg.MaxOutputBytes,

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


@@ -152,6 +152,7 @@ type Agent struct {
152	152		streamingAssistantLines int
153	153		finished time.Time
154	154		lastErr error
	155	+	startupErr error
155	156		shutdownOrigin ShutdownOrigin
156	157		transcriptLoaded bool
157	158		legacyEventState bool


@@ -152,6 +152,7 @@ type Agent struct {
152	streamingAssistantLines int	152		streamingAssistantLines int
153	finished time.Time	153		finished time.Time
154	lastErr error	154		lastErr error
		155	+	startupErr error
155	shutdownOrigin ShutdownOrigin	156		shutdownOrigin ShutdownOrigin
156	transcriptLoaded bool	157		transcriptLoaded bool
157	legacyEventState bool	158		legacyEventState bool

M packages/agent/subagents/model.go +8 -4


@@ -106,13 +106,14 @@ const (
106	106		)
107	107
108	108		// SubagentPolicy contains manager-owned safety and resource limits. normalize
109		-	// replaces zero MaxOutputBytes, MaxOutputLines, QueueTimeout, and
110		-	// DefaultTimeout values with positive manager defaults. Other zero values retain
111		-	// their documented semantics.
	109	+	// replaces zero MaxOutputBytes, MaxOutputLines, QueueTimeout, StartupTimeout,
	110	+	// and DefaultTimeout values with positive manager defaults. Other zero values
	111	+	// retain their documented semantics.
112	112		type SubagentPolicy struct {
113	113		MaxConcurrent int
114	114		MaxConcurrentPerParent int
115	115		QueueTimeout time.Duration
	116	+	StartupTimeout time.Duration
116	117		DefaultTimeout time.Duration
117	118		MaxTurns int
118	119		MaxOutputBytes int

@@ -133,7 +134,10 @@ func (p *SubagentPolicy) normalize() {
133	134		p.MaxConcurrentPerParent = 4
134	135		}
135	136		if p.QueueTimeout <= 0 {
136		-	p.QueueTimeout = 30 * time.Minute
	137	+	p.QueueTimeout = 5 * time.Minute
	138	+	}
	139	+	if p.StartupTimeout <= 0 {
	140	+	p.StartupTimeout = time.Minute
137	141		}
138	142		if p.DefaultTimeout <= 0 {
139	143		p.DefaultTimeout = 20 * time.Minute

 )
 // SubagentPolicy contains manager-owned safety and resource limits. normalize
-// replaces zero MaxOutputBytes, MaxOutputLines, QueueTimeout, and
+// replaces zero MaxOutputBytes, MaxOutputLines, QueueTimeout, StartupTimeout,
-// DefaultTimeout values with positive manager defaults. Other zero values retain
+// and DefaultTimeout values with positive manager defaults. Other zero values
-// their documented semantics.
+// retain their documented semantics.
 type SubagentPolicy struct {
 	MaxConcurrent          int
 	MaxConcurrentPerParent int
 	QueueTimeout           time.Duration
+	StartupTimeout         time.Duration
 	DefaultTimeout         time.Duration
 	MaxTurns               int
 	MaxOutputBytes         int
 		p.MaxConcurrentPerParent = 4
 	}
 	if p.QueueTimeout <= 0 {
-		p.QueueTimeout = 30 * time.Minute
+		p.QueueTimeout = 5 * time.Minute
+	}
+	if p.StartupTimeout <= 0 {
+		p.StartupTimeout = time.Minute
 	}
 	if p.DefaultTimeout <= 0 {
 		p.DefaultTimeout = 20 * time.Minute

M packages/agent/subagents/scheduler.go +48 -3


@@ -11,6 +11,8 @@ import (
11	11		"time"
12	12		)
13	13
	14	+	var errStartupTimeout = errors.New("subagents: worker startup timeout")
	15	+
14	16		func (f *Supervisor) resolveWebSearchPolicy(req SpawnRequest) WebSearchPolicy {
15	17		// Inherit is valid only as the request's unresolved value. Unknown values
16	18		// are not another way to inherit the parent's capability.

@@ -207,6 +209,7 @@ func (f *Supervisor) schedule() {
207	209		for _, a := range starts {
208	210		f.persistAgent(a)
209	211		f.startLifecycleMonitor(a)
	212	+	f.armStartupTimeout(a)
210	213		go f.run(a)
211	214		}
212	215		}

@@ -309,7 +312,49 @@ func (f Supervisor) armQueueTimeout(a Agent) {
309	312		}()
310	313		}
311	314
312		-	func (f *Supervisor) expireQueued(id string) { f.cancelQueued(id, context.DeadlineExceeded) }
	315	+	func (f *Supervisor) expireQueued(id string) {
	316	+	f.cancelQueued(id, fmt.Errorf("subagents: queue timeout after %s waiting for an execution slot", f.cfg.Policy.QueueTimeout))
	317	+	}
	318	+
	319	+	func (f Supervisor) armStartupTimeout(a Agent) {
	320	+	if a == nil {
	321	+	return
	322	+	}
	323	+	d := f.cfg.Policy.StartupTimeout
	324	+	if d <= 0 {
	325	+	return
	326	+	}
	327	+	go func() {
	328	+	timer := time.NewTimer(d)
	329	+	defer timer.Stop()
	330	+	select {
	331	+	case <-a.done:
	332	+	return
	333	+	case <-timer.C:
	334	+	f.expireStarting(a, d)
	335	+	}
	336	+	}()
	337	+	}
	338	+
	339	+	func (f Supervisor) expireStarting(a Agent, d time.Duration) {
	340	+	if a == nil \|\| a.ProcessState() != ProcessStarting {
	341	+	return
	342	+	}
	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()
	347	+	return
	348	+	}
	349	+	a.startupErr = cause
	350	+	a.lastErr = cause
	351	+	a.activity = "startup timeout: worker did not become ready"
	352	+	a.mu.Unlock()
	353	+	f.persistAgent(a)
	354	+	if a.cancel != nil {
	355	+	a.cancel()
	356	+	}
	357	+	}
313	358
314	359		func (f *Supervisor) cancelQueued(id string, cause error) {
315	360		a := f.Get(id)

@@ -330,10 +375,10 @@ func (f *Supervisor) cancelQueued(id string, cause error) {
330	375		}
331	376		a.status = StatusFailed
332	377		atomic.CompareAndSwapInt32(&a.launchState, 0, 2)
333		-	if cause == context.Canceled {
	378	+	if errors.Is(cause, context.Canceled) {
334	379		a.activity = "cancelled while queued"
335	380		} else {
336		-	a.activity = "queue timeout"
	381	+	a.activity = "queue timeout: no execution slot became available"
337	382		}
338	383		a.lastErr = cause
339	384		a.finished = f.cfg.Now()


@@ -11,6 +11,8 @@ import (
11		"time"	11		"time"
12		)	12		)
13			13
			14	+	var errStartupTimeout = errors.New("subagents: worker startup timeout")
			15	+
14		func (f *Supervisor) resolveWebSearchPolicy(req SpawnRequest) WebSearchPolicy {	16		func (f *Supervisor) resolveWebSearchPolicy(req SpawnRequest) WebSearchPolicy {
15		// Inherit is valid only as the request's unresolved value. Unknown values	17		// Inherit is valid only as the request's unresolved value. Unknown values
16		// are not another way to inherit the parent's capability.	18		// are not another way to inherit the parent's capability.

@@ -207,6 +209,7 @@ func (f *Supervisor) schedule() {
207		for _, a := range starts {	209		for _, a := range starts {
208		f.persistAgent(a)	210		f.persistAgent(a)
209		f.startLifecycleMonitor(a)	211		f.startLifecycleMonitor(a)
			212	+	f.armStartupTimeout(a)
210		go f.run(a)	213		go f.run(a)
211		}	214		}
212		}	215		}

@@ -309,7 +312,49 @@ func (f Supervisor) armQueueTimeout(a Agent) {
309		}()	312		}()
310		}	313		}
311			314
312	-	func (f *Supervisor) expireQueued(id string) { f.cancelQueued(id, context.DeadlineExceeded) }	315	+	func (f *Supervisor) expireQueued(id string) {
			316	+	f.cancelQueued(id, fmt.Errorf("subagents: queue timeout after %s waiting for an execution slot", f.cfg.Policy.QueueTimeout))
			317	+	}
			318	+
			319	+	func (f Supervisor) armStartupTimeout(a Agent) {
			320	+	if a == nil {
			321	+	return
			322	+	}
			323	+	d := f.cfg.Policy.StartupTimeout
			324	+	if d <= 0 {
			325	+	return
			326	+	}
			327	+	go func() {
			328	+	timer := time.NewTimer(d)
			329	+	defer timer.Stop()
			330	+	select {
			331	+	case <-a.done:
			332	+	return
			333	+	case <-timer.C:
			334	+	f.expireStarting(a, d)
			335	+	}
			336	+	}()
			337	+	}
			338	+
			339	+	func (f Supervisor) expireStarting(a Agent, d time.Duration) {
			340	+	if a == nil \|\| a.ProcessState() != ProcessStarting {
			341	+	return
			342	+	}
			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()
			347	+	return
			348	+	}
			349	+	a.startupErr = cause
			350	+	a.lastErr = cause
			351	+	a.activity = "startup timeout: worker did not become ready"
			352	+	a.mu.Unlock()
			353	+	f.persistAgent(a)
			354	+	if a.cancel != nil {
			355	+	a.cancel()
			356	+	}
			357	+	}
313			358
314		func (f *Supervisor) cancelQueued(id string, cause error) {	359		func (f *Supervisor) cancelQueued(id string, cause error) {
315		a := f.Get(id)	360		a := f.Get(id)

@@ -330,10 +375,10 @@ func (f *Supervisor) cancelQueued(id string, cause error) {
330		}	375		}
331		a.status = StatusFailed	376		a.status = StatusFailed
332		atomic.CompareAndSwapInt32(&a.launchState, 0, 2)	377		atomic.CompareAndSwapInt32(&a.launchState, 0, 2)
333	-	if cause == context.Canceled {	378	+	if errors.Is(cause, context.Canceled) {
334		a.activity = "cancelled while queued"	379		a.activity = "cancelled while queued"
335		} else {	380		} else {
336	-	a.activity = "queue timeout"	381	+	a.activity = "queue timeout: no execution slot became available"
337		}	382		}
338		a.lastErr = cause	383		a.lastErr = cause
339		a.finished = f.cfg.Now()	384		a.finished = f.cfg.Now()

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


@@ -631,9 +631,16 @@ func (f Supervisor) run(a Agent) {
631	631
632	632		a.mu.Lock()
633	633		a.finished = f.cfg.Now()
	634	+	startupErr := a.startupErr
	635	+	terminalErr := err
634	636		switch {
635	637		case a.status == StatusKilled:
636	638		// Stop requested a terminal state; preserve it.
	639	+	case startupErr != nil:
	640	+	a.status = StatusFailed
	641	+	a.activity = "startup timeout: worker did not become ready"
	642	+	a.lastErr = startupErr
	643	+	terminalErr = startupErr
637	644		case errors.Is(err, context.Canceled):
638	645		a.status = StatusKilled
639	646		a.activity = "cancelled"

@@ -676,7 +683,7 @@ func (f Supervisor) run(a Agent) {
676	683		// temporary workspace is removed. This ordering is the recovery
677	684		// invariant: a user can always inspect a durable result after failure.
678	685		f.captureWorkspace(a)
679		-	f.ensureResult(a, finalStatus, err)
	686	+	f.ensureResult(a, finalStatus, terminalErr)
680	687		if a.cancel != nil {
681	688		a.cancel()
682	689		}

 	a.mu.Lock()
 	a.finished = f.cfg.Now()
+	startupErr := a.startupErr
+	terminalErr := err
 	switch {
 	case a.status == StatusKilled:
 		// Stop requested a terminal state; preserve it.
+	case startupErr != nil:
+		a.status = StatusFailed
+		a.activity = "startup timeout: worker did not become ready"
+		a.lastErr = startupErr
+		terminalErr = startupErr
 	case errors.Is(err, context.Canceled):
 		a.status = StatusKilled
 		a.activity = "cancelled"
 	// temporary workspace is removed. This ordering is the recovery
 	// invariant: a user can always inspect a durable result after failure.
 	f.captureWorkspace(a)
-	f.ensureResult(a, finalStatus, err)
+	f.ensureResult(a, finalStatus, terminalErr)
 	if a.cancel != nil {
 		a.cancel()
 	}

M packages/agent/subagents/supervisor_test.go +93 -0


@@ -111,6 +111,99 @@ func TestSpawnRunsAndCompletes(t *testing.T) {
111	111		}
112	112		}
113	113
	114	+	func TestStartupTimeoutFailsWorkerThatNeverBecomesReady(t *testing.T) {
	115	+	root := t.TempDir()
	116	+	started := make(chan struct{})
	117	+	f := New(Config{
	118	+	Root: root, RepoRoot: root,
	119	+	Policy: SubagentPolicy{StartupTimeout: 20 * time.Millisecond},
	120	+	NewRunner: func(*Agent) Runner {
	121	+	return RunnerFunc(func(ctx context.Context, _ Sink) error {
	122	+	close(started)
	123	+	<-ctx.Done()
	124	+	return ctx.Err()
	125	+	})
	126	+	},
	127	+	})
	128	+	t.Cleanup(f.StopAll)
	129	+
	130	+	a, err := f.Spawn(context.Background(), "stuck startup")
	131	+	if err != nil {
	132	+	t.Fatal(err)
	133	+	}
	134	+	select {
	135	+	case <-started:
	136	+	case <-time.After(time.Second):
	137	+	t.Fatal("runner did not start")
	138	+	}
	139	+	waitCtx, cancel := context.WithTimeout(context.Background(), time.Second)
	140	+	defer cancel()
	141	+	if err := a.WaitContext(waitCtx); err != nil {
	142	+	t.Fatalf("worker did not fail after startup timeout: %v", err)
	143	+	}
	144	+	if got := a.Status(); got != StatusFailed {
	145	+	t.Fatalf("status = %s, want %s", got, StatusFailed)
	146	+	}
	147	+	if !errors.Is(a.Err(), errStartupTimeout) {
	148	+	t.Fatalf("worker error = %v, want startup timeout", a.Err())
	149	+	}
	150	+	if got := a.Activity(); !strings.Contains(got, "startup timeout") {
	151	+	t.Fatalf("activity = %q, want startup timeout explanation", got)
	152	+	}
	153	+	}
	154	+
	155	+	func TestQueueTimeoutFailsWorkerWaitingForExecutionSlot(t *testing.T) {
	156	+	root := t.TempDir()
	157	+	started := make(chan struct{})
	158	+	var startOnce sync.Once
	159	+	f := New(Config{
	160	+	Root: root, RepoRoot: root,
	161	+	Policy: SubagentPolicy{
	162	+	MaxConcurrent: 1,
	163	+	QueueTimeout: 20 * time.Millisecond,
	164	+	StartupTimeout: time.Second,
	165	+	},
	166	+	NewRunner: func(*Agent) Runner {
	167	+	return RunnerFunc(func(ctx context.Context, _ Sink) error {
	168	+	startOnce.Do(func() { close(started) })
	169	+	<-ctx.Done()
	170	+	return ctx.Err()
	171	+	})
	172	+	},
	173	+	})
	174	+	t.Cleanup(f.StopAll)
	175	+
	176	+	first, err := f.Spawn(context.Background(), "occupy only slot")
	177	+	if err != nil {
	178	+	t.Fatal(err)
	179	+	}
	180	+	select {
	181	+	case <-started:
	182	+	case <-time.After(time.Second):
	183	+	t.Fatal("first runner did not start")
	184	+	}
	185	+	queued, err := f.Spawn(context.Background(), "wait for slot")
	186	+	if err != nil {
	187	+	t.Fatal(err)
	188	+	}
	189	+	waitCtx, cancel := context.WithTimeout(context.Background(), time.Second)
	190	+	defer cancel()
	191	+	if err := queued.WaitContext(waitCtx); err != nil {
	192	+	t.Fatalf("queued worker did not time out: %v", err)
	193	+	}
	194	+	if got := queued.Status(); got != StatusFailed {
	195	+	t.Fatalf("queued worker status = %s, want %s", got, StatusFailed)
	196	+	}
	197	+	if got := queued.ProcessState(); got != ProcessExited {
	198	+	t.Fatalf("queued worker process state = %s, want %s", got, ProcessExited)
	199	+	}
	200	+	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)
	202	+	}
	203	+	f.Stop(first.ID)
	204	+	first.Wait()
	205	+	}
	206	+
114	207		// TestSpawnAgentSharesRepoRoot verifies the only-mode-we-support:
115	208		// every spawned agent points its cwd at the parent zut's RepoRoot.
116	209		func TestSpawnAgentSharesRepoRoot(t *testing.T) {


@@ -111,6 +111,99 @@ func TestSpawnRunsAndCompletes(t *testing.T) {
111	}	111		}
112	}	112		}
113		113
		114	+	func TestStartupTimeoutFailsWorkerThatNeverBecomesReady(t *testing.T) {
		115	+	root := t.TempDir()
		116	+	started := make(chan struct{})
		117	+	f := New(Config{
		118	+	Root: root, RepoRoot: root,
		119	+	Policy: SubagentPolicy{StartupTimeout: 20 * time.Millisecond},
		120	+	NewRunner: func(*Agent) Runner {
		121	+	return RunnerFunc(func(ctx context.Context, _ Sink) error {
		122	+	close(started)
		123	+	<-ctx.Done()
		124	+	return ctx.Err()
		125	+	})
		126	+	},
		127	+	})
		128	+	t.Cleanup(f.StopAll)
		129	+
		130	+	a, err := f.Spawn(context.Background(), "stuck startup")
		131	+	if err != nil {
		132	+	t.Fatal(err)
		133	+	}
		134	+	select {
		135	+	case <-started:
		136	+	case <-time.After(time.Second):
		137	+	t.Fatal("runner did not start")
		138	+	}
		139	+	waitCtx, cancel := context.WithTimeout(context.Background(), time.Second)
		140	+	defer cancel()
		141	+	if err := a.WaitContext(waitCtx); err != nil {
		142	+	t.Fatalf("worker did not fail after startup timeout: %v", err)
		143	+	}
		144	+	if got := a.Status(); got != StatusFailed {
		145	+	t.Fatalf("status = %s, want %s", got, StatusFailed)
		146	+	}
		147	+	if !errors.Is(a.Err(), errStartupTimeout) {
		148	+	t.Fatalf("worker error = %v, want startup timeout", a.Err())
		149	+	}
		150	+	if got := a.Activity(); !strings.Contains(got, "startup timeout") {
		151	+	t.Fatalf("activity = %q, want startup timeout explanation", got)
		152	+	}
		153	+	}
		154	+
		155	+	func TestQueueTimeoutFailsWorkerWaitingForExecutionSlot(t *testing.T) {
		156	+	root := t.TempDir()
		157	+	started := make(chan struct{})
		158	+	var startOnce sync.Once
		159	+	f := New(Config{
		160	+	Root: root, RepoRoot: root,
		161	+	Policy: SubagentPolicy{
		162	+	MaxConcurrent: 1,
		163	+	QueueTimeout: 20 * time.Millisecond,
		164	+	StartupTimeout: time.Second,
		165	+	},
		166	+	NewRunner: func(*Agent) Runner {
		167	+	return RunnerFunc(func(ctx context.Context, _ Sink) error {
		168	+	startOnce.Do(func() { close(started) })
		169	+	<-ctx.Done()
		170	+	return ctx.Err()
		171	+	})
		172	+	},
		173	+	})
		174	+	t.Cleanup(f.StopAll)
		175	+
		176	+	first, err := f.Spawn(context.Background(), "occupy only slot")
		177	+	if err != nil {
		178	+	t.Fatal(err)
		179	+	}
		180	+	select {
		181	+	case <-started:
		182	+	case <-time.After(time.Second):
		183	+	t.Fatal("first runner did not start")
		184	+	}
		185	+	queued, err := f.Spawn(context.Background(), "wait for slot")
		186	+	if err != nil {
		187	+	t.Fatal(err)
		188	+	}
		189	+	waitCtx, cancel := context.WithTimeout(context.Background(), time.Second)
		190	+	defer cancel()
		191	+	if err := queued.WaitContext(waitCtx); err != nil {
		192	+	t.Fatalf("queued worker did not time out: %v", err)
		193	+	}
		194	+	if got := queued.Status(); got != StatusFailed {
		195	+	t.Fatalf("queued worker status = %s, want %s", got, StatusFailed)
		196	+	}
		197	+	if got := queued.ProcessState(); got != ProcessExited {
		198	+	t.Fatalf("queued worker process state = %s, want %s", got, ProcessExited)
		199	+	}
		200	+	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)
		202	+	}
		203	+	f.Stop(first.ID)
		204	+	first.Wait()
		205	+	}
		206	+
114	// TestSpawnAgentSharesRepoRoot verifies the only-mode-we-support:	207		// TestSpawnAgentSharesRepoRoot verifies the only-mode-we-support:
115	// every spawned agent points its cwd at the parent zut's RepoRoot.	208		// every spawned agent points its cwd at the parent zut's RepoRoot.
116	func TestSpawnAgentSharesRepoRoot(t *testing.T) {	209		func TestSpawnAgentSharesRepoRoot(t *testing.T) {

M packages/core/agent_retry_test.go +21 -2


@@ -4,6 +4,7 @@ import (
4	4		"context"
5	5		"errors"
6	6		"fmt"
	7	+	"io"
7	8		"strings"
8	9		"sync/atomic"
9	10		"testing"

@@ -14,7 +15,8 @@ import (
14	15		)
15	16
16	17		type retryFakeClient struct {
17		-	calls int32
	18	+	calls int32
	19	+	firstErr error
18	20		}
19	21
20	22		func (c *retryFakeClient) Name() string { return "retry-fake" }

@@ -29,7 +31,11 @@ func (c *retryFakeClient) Stream(ctx context.Context, req provider.Request) (<-c
29	31		defer close(out)
30	32		out <- provider.EventStart{Provider: "retry-fake", Model: req.Model}
31	33		if call == 1 {
32		-	out <- provider.EventDone{Stop: provider.StopError, Err: fmt.Errorf("anthropic overloaded_error: Overloaded")}
	34	+	err := c.firstErr
	35	+	if err == nil {
	36	+	err = fmt.Errorf("anthropic overloaded_error: Overloaded")
	37	+	}
	38	+	out <- provider.EventDone{Stop: provider.StopError, Err: err}
33	39		return
34	40		}
35	41		out <- provider.EventTextDelta{Delta: "ok"}

@@ -70,6 +76,19 @@ func TestAgentRetriesOverloadedStreamError(t *testing.T) {
70	76		}
71	77		}
72	78
	79	+	func TestAgentRetriesUnexpectedEOFStreamError(t *testing.T) {
	80	+	client := &retryFakeClient{firstErr: fmt.Errorf("read SSE: %w", io.ErrUnexpectedEOF)}
	81	+	a := NewAgent(client, "fake-model", "system", Registry{})
	82	+	a.RetryBaseDelay = time.Millisecond
	83	+
	84	+	if err := a.Prompt(context.Background(), "hello", nil, nil); err != nil {
	85	+	t.Fatalf("Prompt returned %v", err)
	86	+	}
	87	+	if got := atomic.LoadInt32(&client.calls); got != 2 {
	88	+	t.Fatalf("Stream calls = %d; want 2", got)
	89	+	}
	90	+	}
	91	+
73	92		func TestAgentEmitsRetryLifecycleEvents(t *testing.T) {
74	93		client := &retryFakeClient{}
75	94		a := NewAgent(client, "fake-model", "system", Registry{})

 	"context"
 	"errors"
 	"fmt"
+	"io"
 	"strings"
 	"sync/atomic"
 	"testing"
 )
 type retryFakeClient struct {
-	calls int32
+	calls    int32
+	firstErr error
 }
 func (c *retryFakeClient) Name() string { return "retry-fake" }
 		defer close(out)
 		out <- provider.EventStart{Provider: "retry-fake", Model: req.Model}
 		if call == 1 {
-			out <- provider.EventDone{Stop: provider.StopError, Err: fmt.Errorf("anthropic overloaded_error: Overloaded")}
+			err := c.firstErr
+			if err == nil {
+				err = fmt.Errorf("anthropic overloaded_error: Overloaded")
+			}
+			out <- provider.EventDone{Stop: provider.StopError, Err: err}
 			return
 		}
 		out <- provider.EventTextDelta{Delta: "ok"}
 	}
 }
+func TestAgentRetriesUnexpectedEOFStreamError(t *testing.T) {
+	client := &retryFakeClient{firstErr: fmt.Errorf("read SSE: %w", io.ErrUnexpectedEOF)}
+	a := NewAgent(client, "fake-model", "system", Registry{})
+	a.RetryBaseDelay = time.Millisecond
+	if err := a.Prompt(context.Background(), "hello", nil, nil); err != nil {
+		t.Fatalf("Prompt returned %v", err)
+	}
+	if got := atomic.LoadInt32(&client.calls); got != 2 {
+		t.Fatalf("Stream calls = %d; want 2", got)
+	}
+}
 func TestAgentEmitsRetryLifecycleEvents(t *testing.T) {
 	client := &retryFakeClient{}
 	a := NewAgent(client, "fake-model", "system", Registry{})

M packages/provider/anthropic.go +6 -0


@@ -680,6 +680,12 @@ func (c anthropicClient) runStream(ctx context.Context, resp http.Response, re
680	680		sendDone()
681	681		return
682	682		}
	683	+	if ev.Err != nil {
	684	+	stop = StopError
	685	+	finalErr = fmt.Errorf("read SSE: %w", ev.Err)
	686	+	sendDone()
	687	+	return
	688	+	}
683	689		// Parse event payload based on event: type.
684	690		var payload map[string]json.RawMessage
685	691		if err := json.Unmarshal([]byte(ev.Data), &payload); err != nil {


@@ -680,6 +680,12 @@ func (c anthropicClient) runStream(ctx context.Context, resp http.Response, re
680	sendDone()	680		sendDone()
681	return	681		return
682	}	682		}
		683	+	if ev.Err != nil {
		684	+	stop = StopError
		685	+	finalErr = fmt.Errorf("read SSE: %w", ev.Err)
		686	+	sendDone()
		687	+	return
		688	+	}
683	// Parse event payload based on event: type.	689		// Parse event payload based on event: type.
684	var payload map[string]json.RawMessage	690		var payload map[string]json.RawMessage
685	if err := json.Unmarshal([]byte(ev.Data), &payload); err != nil {	691		if err := json.Unmarshal([]byte(ev.Data), &payload); err != nil {

M packages/provider/gemini.go +6 -0


@@ -661,6 +661,12 @@ func (c geminiClient) runStream(ctx context.Context, resp http.Response, req R
661	661		sendDone()
662	662		return
663	663		}
	664	+	if ev.Err != nil {
	665	+	stop = StopError
	666	+	finalErr = fmt.Errorf("read SSE: %w", ev.Err)
	667	+	sendDone()
	668	+	return
	669	+	}
664	670		if strings.TrimSpace(ev.Data) == "" {
665	671		continue
666	672		}


@@ -661,6 +661,12 @@ func (c geminiClient) runStream(ctx context.Context, resp http.Response, req R
661	sendDone()	661		sendDone()
662	return	662		return
663	}	663		}
		664	+	if ev.Err != nil {
		665	+	stop = StopError
		666	+	finalErr = fmt.Errorf("read SSE: %w", ev.Err)
		667	+	sendDone()
		668	+	return
		669	+	}
664	if strings.TrimSpace(ev.Data) == "" {	670		if strings.TrimSpace(ev.Data) == "" {
665	continue	671		continue
666	}	672		}

M packages/provider/openai.go +6 -0


@@ -622,6 +622,12 @@ func (c openaiClient) runStream(ctx context.Context, resp http.Response, req R
622	622		sendDone()
623	623		return
624	624		}
	625	+	if ev.Err != nil {
	626	+	stop = StopError
	627	+	finalErr = fmt.Errorf("read SSE: %w", ev.Err)
	628	+	sendDone()
	629	+	return
	630	+	}
625	631		if ev.Data == "[DONE]" {
626	632		sendDone()
627	633		return


@@ -622,6 +622,12 @@ func (c openaiClient) runStream(ctx context.Context, resp http.Response, req R
622	sendDone()	622		sendDone()
623	return	623		return
624	}	624		}
		625	+	if ev.Err != nil {
		626	+	stop = StopError
		627	+	finalErr = fmt.Errorf("read SSE: %w", ev.Err)
		628	+	sendDone()
		629	+	return
		630	+	}
625	if ev.Data == "[DONE]" {	631		if ev.Data == "[DONE]" {
626	sendDone()	632		sendDone()
627	return	633		return

M packages/provider/openai_codex.go +6 -0


@@ -518,6 +518,12 @@ func (c codexClient) runStream(ctx context.Context, resp http.Response, req Re
518	518		sendDone()
519	519		return
520	520		}
	521	+	if ev.Err != nil {
	522	+	stop = StopError
	523	+	finalErr = fmt.Errorf("read SSE: %w", ev.Err)
	524	+	sendDone()
	525	+	return
	526	+	}
521	527		var head struct {
522	528		Type string `json:"type"`
523	529		}


@@ -518,6 +518,12 @@ func (c codexClient) runStream(ctx context.Context, resp http.Response, req Re
518	sendDone()	518		sendDone()
519	return	519		return
520	}	520		}
		521	+	if ev.Err != nil {
		522	+	stop = StopError
		523	+	finalErr = fmt.Errorf("read SSE: %w", ev.Err)
		524	+	sendDone()
		525	+	return
		526	+	}
521	var head struct {	527		var head struct {
522	Type string `json:"type"`	528		Type string `json:"type"`
523	}	529		}

M packages/provider/provider_test.go +100 -0


@@ -2,6 +2,8 @@ package provider
2	2
3	3		import (
4	4		"context"
	5	+	"errors"
	6	+	"io"
5	7		"math"
6	8		"net/http"
7	9		"net/http/httptest"

@@ -27,6 +29,104 @@ func TestSSEParse(t *testing.T) {
27	29		}
28	30		}
29	31
	32	+	type errorAfterReader struct {
	33	+	data *strings.Reader
	34	+	err error
	35	+	}
	36	+
	37	+	func (r *errorAfterReader) Read(p []byte) (int, error) {
	38	+	if r.data.Len() > 0 {
	39	+	return r.data.Read(p)
	40	+	}
	41	+	return 0, r.err
	42	+	}
	43	+
	44	+	func TestSSEReadErrorIsDeliveredBeforeChannelClose(t *testing.T) {
	45	+	want := io.ErrUnexpectedEOF
	46	+	ch := make(chan sseEvent, 2)
	47	+	go readSSE(&errorAfterReader{
	48	+	data: strings.NewReader("data: hello\n\n"),
	49	+	err: want,
	50	+	}, ch)
	51	+
	52	+	if event := <-ch; event.Data != "hello" {
	53	+	t.Fatalf("event = %+v, want payload before read error", event)
	54	+	}
	55	+	terminal, ok := <-ch
	56	+	if !ok {
	57	+	t.Fatal("channel closed without scanner error")
	58	+	}
	59	+	if !errors.Is(terminal.Err, want) {
	60	+	t.Fatalf("scanner error = %v, want %v", terminal.Err, want)
	61	+	}
	62	+	if _, ok := <-ch; ok {
	63	+	t.Fatal("channel remained open after scanner error")
	64	+	}
	65	+	}
	66	+
	67	+	func TestStreamsReportSSEReadFailures(t *testing.T) {
	68	+	want := io.ErrUnexpectedEOF
	69	+	response := func() *http.Response {
	70	+	return &http.Response{Body: io.NopCloser(&errorAfterReader{
	71	+	data: strings.NewReader(""),
	72	+	err: want,
	73	+	})}
	74	+	}
	75	+	cases := []struct {
	76	+	name string
	77	+	run func(chan<- Event)
	78	+	}{
	79	+	{
	80	+	name: "anthropic",
	81	+	run: func(out chan<- Event) {
	82	+	(&anthropicClient{}).runStream(context.Background(), response(), Request{Model: "test"}, out)
	83	+	},
	84	+	},
	85	+	{
	86	+	name: "openai",
	87	+	run: func(out chan<- Event) {
	88	+	(&openaiClient{}).runStream(context.Background(), response(), Request{Model: "test"}, out)
	89	+	},
	90	+	},
	91	+	{
	92	+	name: "gemini",
	93	+	run: func(out chan<- Event) {
	94	+	(&geminiClient{}).runStream(context.Background(), response(), Request{Model: "test"}, out)
	95	+	},
	96	+	},
	97	+	{
	98	+	name: "openai codex",
	99	+	run: func(out chan<- Event) {
	100	+	(&codexClient{}).runStream(context.Background(), response(), Request{Model: "test"}, out)
	101	+	},
	102	+	},
	103	+	}
	104	+	for _, tc := range cases {
	105	+	t.Run(tc.name, func(t *testing.T) {
	106	+	out := make(chan Event, 4)
	107	+	go tc.run(out)
	108	+
	109	+	var done EventDone
	110	+	foundDone := false
	111	+	for event := range out {
	112	+	if e, ok := event.(EventDone); ok {
	113	+	done = e
	114	+	foundDone = true
	115	+	}
	116	+	}
	117	+	if !foundDone {
	118	+	t.Fatal("stream ended without EventDone")
	119	+	}
	120	+	if done.Stop != StopError {
	121	+	t.Fatalf("stop = %s, want %s", done.Stop, StopError)
	122	+	}
	123	+	if !errors.Is(done.Err, want) {
	124	+	t.Fatalf("stream error = %v, want %v", done.Err, want)
	125	+	}
	126	+	})
	127	+	}
	128	+	}
	129	+
30	130		func TestModelCatalog(t *testing.T) {
31	131		if len(Catalog) == 0 {
32	132		t.Fatal("empty catalog")


@@ -2,6 +2,8 @@ package provider
2		2
3	import (	3		import (
4	"context"	4		"context"
		5	+	"errors"
		6	+	"io"
5	"math"	7		"math"
6	"net/http"	8		"net/http"
7	"net/http/httptest"	9		"net/http/httptest"

@@ -27,6 +29,104 @@ func TestSSEParse(t *testing.T) {
27	}	29		}
28	}	30		}
29		31
		32	+	type errorAfterReader struct {
		33	+	data *strings.Reader
		34	+	err error
		35	+	}
		36	+
		37	+	func (r *errorAfterReader) Read(p []byte) (int, error) {
		38	+	if r.data.Len() > 0 {
		39	+	return r.data.Read(p)
		40	+	}
		41	+	return 0, r.err
		42	+	}
		43	+
		44	+	func TestSSEReadErrorIsDeliveredBeforeChannelClose(t *testing.T) {
		45	+	want := io.ErrUnexpectedEOF
		46	+	ch := make(chan sseEvent, 2)
		47	+	go readSSE(&errorAfterReader{
		48	+	data: strings.NewReader("data: hello\n\n"),
		49	+	err: want,
		50	+	}, ch)
		51	+
		52	+	if event := <-ch; event.Data != "hello" {
		53	+	t.Fatalf("event = %+v, want payload before read error", event)
		54	+	}
		55	+	terminal, ok := <-ch
		56	+	if !ok {
		57	+	t.Fatal("channel closed without scanner error")
		58	+	}
		59	+	if !errors.Is(terminal.Err, want) {
		60	+	t.Fatalf("scanner error = %v, want %v", terminal.Err, want)
		61	+	}
		62	+	if _, ok := <-ch; ok {
		63	+	t.Fatal("channel remained open after scanner error")
		64	+	}
		65	+	}
		66	+
		67	+	func TestStreamsReportSSEReadFailures(t *testing.T) {
		68	+	want := io.ErrUnexpectedEOF
		69	+	response := func() *http.Response {
		70	+	return &http.Response{Body: io.NopCloser(&errorAfterReader{
		71	+	data: strings.NewReader(""),
		72	+	err: want,
		73	+	})}
		74	+	}
		75	+	cases := []struct {
		76	+	name string
		77	+	run func(chan<- Event)
		78	+	}{
		79	+	{
		80	+	name: "anthropic",
		81	+	run: func(out chan<- Event) {
		82	+	(&anthropicClient{}).runStream(context.Background(), response(), Request{Model: "test"}, out)
		83	+	},
		84	+	},
		85	+	{
		86	+	name: "openai",
		87	+	run: func(out chan<- Event) {
		88	+	(&openaiClient{}).runStream(context.Background(), response(), Request{Model: "test"}, out)
		89	+	},
		90	+	},
		91	+	{
		92	+	name: "gemini",
		93	+	run: func(out chan<- Event) {
		94	+	(&geminiClient{}).runStream(context.Background(), response(), Request{Model: "test"}, out)
		95	+	},
		96	+	},
		97	+	{
		98	+	name: "openai codex",
		99	+	run: func(out chan<- Event) {
		100	+	(&codexClient{}).runStream(context.Background(), response(), Request{Model: "test"}, out)
		101	+	},
		102	+	},
		103	+	}
		104	+	for _, tc := range cases {
		105	+	t.Run(tc.name, func(t *testing.T) {
		106	+	out := make(chan Event, 4)
		107	+	go tc.run(out)
		108	+
		109	+	var done EventDone
		110	+	foundDone := false
		111	+	for event := range out {
		112	+	if e, ok := event.(EventDone); ok {
		113	+	done = e
		114	+	foundDone = true
		115	+	}
		116	+	}
		117	+	if !foundDone {
		118	+	t.Fatal("stream ended without EventDone")
		119	+	}
		120	+	if done.Stop != StopError {
		121	+	t.Fatalf("stop = %s, want %s", done.Stop, StopError)
		122	+	}
		123	+	if !errors.Is(done.Err, want) {
		124	+	t.Fatalf("stream error = %v, want %v", done.Err, want)
		125	+	}
		126	+	})
		127	+	}
		128	+	}
		129	+
30	func TestModelCatalog(t *testing.T) {	130		func TestModelCatalog(t *testing.T) {
31	if len(Catalog) == 0 {	131		if len(Catalog) == 0 {
32	t.Fatal("empty catalog")	132		t.Fatal("empty catalog")

M packages/provider/sse.go +6 -2


@@ -10,10 +10,11 @@ import (
10	10		type sseEvent struct {
11	11		Event string // value of "event:" field (may be empty)
12	12		Data string // concatenated "data:" lines
	13	+	Err error // terminal scanner failure, if any
13	14		}
14	15
15		-	// readSSE reads events from r and sends them on out. It closes out when r
16		-	// is exhausted or a read error occurs.
	16	+	// readSSE reads events from r and sends them on out. A scanner failure is
	17	+	// delivered as a final event before out closes.
17	18		func readSSE(r io.Reader, out chan<- sseEvent) {
18	19		defer close(out)
19	20		sc := bufio.NewScanner(r)

@@ -56,4 +57,7 @@ func readSSE(r io.Reader, out chan<- sseEvent) {
56	57		}
57	58		}
58	59		flush()
	60	+	if err := sc.Err(); err != nil {
	61	+	out <- sseEvent{Err: err}
	62	+	}
59	63		}

 type sseEvent struct {
 	Event string // value of "event:" field (may be empty)
 	Data  string // concatenated "data:" lines
+	Err   error  // terminal scanner failure, if any
 }
-// readSSE reads events from r and sends them on out. It closes out when r
+// readSSE reads events from r and sends them on out. A scanner failure is
-// is exhausted or a read error occurs.
+// delivered as a final event before out closes.
 func readSSE(r io.Reader, out chan<- sseEvent) {
 	defer close(out)
 	sc := bufio.NewScanner(r)
 		}
 	}
 	flush()
+	if err := sc.Err(); err != nil {
+		out <- sseEvent{Err: err}
+	}
 }