fix(subagents,llm): enforce process-wide child concurrency + configurable stream idle watchdog

AGENTS.md

@@ -121,7 +121,7 @@ ReAct cycle: observe → think → act → repeat.
 ### Extension capabilities (odek-extension/v1, v1.24.0)
 - **MCP per-server limits** — `timeout_seconds` (30s/3600s cap), `max_response_bytes` (10 MiB/64 MiB ceiling), `max_result_chars` (200k/1M cap, structured truncation notice), `artifact_roots` (empty ⇒ refs rejected). Resolved per client; approval keys hash all four fields.
 - **Artifact references** — MCP tools return `odek.tool-result/v1` envelopes with `file://` refs instead of bulk content; validated fail-closed in `internal/artifact`; model sees metadata only.
-- **Runtime events** — `odek.event/v1` via `Config.EventHandler` (non-blocking, panic-isolated) and `run --events-jsonl`. Types: run_started, iteration_completed, tool_call_*, session_saved, context_trimmed, budget_exceeded, run_completed/run_failed, plan_created, plan_updated, subagent_denied, subagent_spawned, subagent_completed.
+- **Runtime events** — `odek.event/v1` via `Config.EventHandler` (non-blocking, panic-isolated) and `run --events-jsonl`. Types: run_started, iteration_completed, tool_call_*, session_saved, context_trimmed, budget_exceeded, run_completed/run_failed, plan_created, plan_updated, subagent_denied, subagent_spawned, subagent_completed, subagent_concurrency_wait.
 - **External session refs** — `Session.ExternalRefs` + `--external-ref` on run/continue; validated, deduped, never dereferenced.
 - **Execution budgets** — `limits` config section + `--max-runtime/--max-tool-calls/--max-input-tokens/--max-output-tokens/--max-cost-usd` on `run`; typed `budget.Error` → CLI exit code 4; session persisted before return. Per-model prices via `limits.model_prices` with flat-pair fallback; cost enforcement only when cap + prices configured. `odek init --global` scaffolds the section (zeros = off). `GET /api/limits` on serve exposes limits + effective prices for cost rendering.
 

cmd/odek/main.go

@@ -2304,6 +2304,7 @@ func builtinTools(dc danger.DangerousConfig, sm *skills.SkillManager, approver d
 		},
 		&delegateTasksTool{
 			maxConcurrency: subConcurrency,
+			sharedSem:      sharedChildSem(subConcurrency),
 			odekPath:       os.Args[0],
 			apiKey:         apiKey,
 			timeout:        time.Duration(subTimeout) * time.Second,

cmd/odek/subagent_tool.go

@@ -40,9 +40,18 @@ type delegateTasksTool struct {
 	ctxTool
 
 	maxConcurrency int
-	odekPath       string // path to the odek binary
-	apiKey         string // re-injected into sub-agent environment
-	timeout        time.Duration
+
+	// sharedSem is the process-wide child limiter wired by builtinTools
+	// (sharedChildSem): provider plans are account-wide, so every
+	// delegate_tasks instance in the process — sibling tool calls in one
+	// parallel batch, and concurrent serve sessions — shares a single
+	// subagent.max_concurrency bound. Nil (bare-struct tests) falls back
+	// to a private per-call semaphore.
+	sharedSem chan struct{}
+	runTaskFn func(taskIdx int, taskID, goal, taskContext, guidance, trustLevel, maxRisk, profile, artifactDir string) string
+	odekPath  string // path to the odek binary
+	apiKey    string // re-injected into sub-agent environment
+	timeout   time.Duration
 
 	// sessionID is the parent agent's session id (optional — SetSessionID).
 	// Artifact outputs file under <artifactsRoot>/<sessionID>/<taskID>/;
@@ -239,14 +248,23 @@ func (t *delegateTasksTool) Call(args string) (string, error) {
 		return fmt.Sprintf(`{"error":"delegation depth limit reached (depth %d, max %d); do this work yourself instead of delegating"}`, depth, t.maxDepth), nil
 	}
 
-	// Run sub-agents in parallel with concurrency limit
+	// Run sub-agents in parallel with a concurrency limit. sem is the
+	// process-wide shared limiter when wired (provider plans are
+	// account-wide — sibling delegate_tasks calls in one batch and
+	// concurrent serve sessions share it); bare-struct tests fall back to
+	// a private per-call semaphore. Capacity < 1 is normalized to 1: an
+	// unbuffered channel would deadlock the acquire-before-spawn loop.
 	results := make([]string, len(input.Tasks))
 	dirs := make([]string, len(input.Tasks)) // per-task artifact dirs (parent-created)
-	sem := make(chan struct{}, t.maxConcurrency)
+	sem := t.concurrencySem()
 	var mu sync.Mutex
+	var wg sync.WaitGroup
+	t.eventMu.Lock()
+	emitFn := t.emitEventFn
+	t.eventMu.Unlock()
 
 	for i, task := range input.Tasks {
-		sem <- struct{}{}
+		t.acquireSem(sem, emitFn, i)
 		// Task id is minted HERE (not inside runTask) so the per-task
 		// artifact dir can be created serially and correlated with the id
 		// the child echoes on every telemetry record.
@@ -259,19 +277,25 @@ func (t *delegateTasksTool) Call(args string) (string, error) {
 				dirs[i] = d
 			}
 		}
+		run := t.runTaskFn
+		if run == nil {
+			run = t.runTask
+		}
+		wg.Add(1)
 		go func(i int, taskID, goal, ctx, guidance, trust, maxRisk, profile, artifactDir string) {
+			defer wg.Done()
 			defer func() { <-sem }()
-			r := t.runTask(i, taskID, goal, ctx, guidance, trust, maxRisk, profile, artifactDir)
+			r := run(i, taskID, goal, ctx, guidance, trust, maxRisk, profile, artifactDir)
 			mu.Lock()
 			results[i] = r
 			mu.Unlock()
 		}(i, taskID, task.Goal, task.Context, task.Guidance, task.TrustLevel, task.MaxRisk, task.Profile, dirs[i])
 	}
 
-	// Drain semaphore = wait for all goroutines
-	for i := 0; i < cap(sem); i++ {
-		sem <- struct{}{}
-	}
+	// Wait for every goroutine. Never refill a shared limiter's slots to
+	// "drain" it — the tokens would be permanently consumed and deadlock
+	// later calls sharing the same limiter.
+	wg.Wait()
 
 	// P1: surface child denials on the runtime event stream (best effort —
 	// unparseable results simply carry no denials).
@@ -824,6 +848,75 @@ func newTaskEnvelope(taskID, goal, context, guidance, trustLevel, maxRisk, profi
 
 // subagentDeniedEvent is emitted on the runtime event stream for every
 // policy denial observed by a child sub-agent (P1).
+// subagentWaitEventThreshold is how long a task may queue on the shared
+// child limiter before odek emits subagent_concurrency_wait. Var so tests
+// can shrink it.
+var subagentWaitEventThreshold = 5 * time.Second
+
+const subagentConcurrencyWaitEvent = "subagent_concurrency_wait"
+
+// acquireSem blocks until a limiter slot is free, emitting a
+// subagent_concurrency_wait runtime event when a task visibly queues.
+func (t *delegateTasksTool) acquireSem(sem chan struct{}, emit func(events.Event), taskIdx int) {
+	if emit == nil {
+		sem <- struct{}{}
+		return
+	}
+	start := time.Now()
+	timer := time.NewTimer(subagentWaitEventThreshold)
+	defer timer.Stop()
+	warned := false
+	for {
+		select {
+		case sem <- struct{}{}:
+			return
+		case <-timer.C:
+			if !warned && time.Since(start) >= subagentWaitEventThreshold {
+				warned = true
+				emit(events.Event{
+					Type: subagentConcurrencyWaitEvent,
+					Data: map[string]any{"task_index": taskIdx, "waited_ms": time.Since(start).Milliseconds()},
+				})
+			}
+			timer.Reset(subagentWaitEventThreshold)
+		}
+	}
+}
+
+// sharedChildSem returns the process-wide child limiter, creating it on
+// first use sized to the resolved subagent.max_concurrency. Idempotent:
+// the first caller's capacity wins (resolved config is stable per process),
+// so every builtinTools call in the process — all serve sessions — shares
+// one bound.
+func sharedChildSem(capacity int) chan struct{} {
+	if capacity < 1 {
+		capacity = 1
+	}
+	sharedSemOnce.Do(func() {
+		processChildSem = make(chan struct{}, capacity)
+	})
+	return processChildSem
+}
+
+var (
+	sharedSemOnce   sync.Once
+	processChildSem chan struct{}
+)
+
+// concurrencySem resolves the limiter for this call: the explicit shared
+// semaphore (builtinTools wiring), else a private per-call semaphore
+// (bare-struct tests).
+func (t *delegateTasksTool) concurrencySem() chan struct{} {
+	if t.sharedSem != nil {
+		return t.sharedSem
+	}
+	capacity := t.maxConcurrency
+	if capacity < 1 {
+		capacity = 1
+	}
+	return make(chan struct{}, capacity)
+}
+
 const subagentDeniedEvent = "subagent_denied"
 
 // emitSubagentEvent forwards a sub-agent lifecycle event to the runtime

cmd/odek/subagent_tool_concurrency_test.go

@@ -0,0 +1,120 @@
+package main
+
+import (
+	"strings"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/BackendStack21/odek/internal/events"
+)
+
+func stubRun(d time.Duration) func(int, string, string, string, string, string, string, string, string) string {
+	return func(int, string, string, string, string, string, string, string, string) string {
+		time.Sleep(d)
+		return `{"status":"success","summary":"ok"}`
+	}
+}
+
+// peakTrackingRun instruments a stub run to record the peak number of
+// concurrently executing tasks.
+func peakTrackingRun(peak, cur *atomic.Int64, d time.Duration) func(int, string, string, string, string, string, string, string, string) string {
+	return func(int, string, string, string, string, string, string, string, string) string {
+		n := cur.Add(1)
+		for {
+			p := peak.Load()
+			if n <= p || peak.CompareAndSwap(p, n) {
+				break
+			}
+		}
+		time.Sleep(d)
+		cur.Add(-1)
+		return `{"status":"success","summary":"ok"}`
+	}
+}
+
+func taskJSON(n int) string {
+	return `{"tasks":[` + strings.TrimSuffix(strings.Repeat(`{"goal":"g"},`, n), ",") + `]}`
+}
+
+// TestDelegateTasks_SharedLimiterBoundsPeakAcrossInstances pins M-concurrency:
+// the child limiter is process-wide. Two delegate_tasks instances sharing one
+// limiter (as sibling tool calls in one parallel batch, or two serve sessions,
+// do) must never exceed the cap in total — per-instance semaphores would allow
+// cap×instances concurrent child streams against an account-wide provider
+// plan.
+func TestDelegateTasks_SharedLimiterBoundsPeakAcrossInstances(t *testing.T) {
+	shared := make(chan struct{}, 2)
+	var peak, cur atomic.Int64
+	mk := func() *delegateTasksTool {
+		return &delegateTasksTool{
+			maxConcurrency: 3, // per-instance cap would allow 3 each
+			sharedSem:      shared,
+			runTaskFn:      peakTrackingRun(&peak, &cur, 150*time.Millisecond),
+			odekPath:       "unused",
+		}
+	}
+	a, b := mk(), mk()
+	var wg sync.WaitGroup
+	wg.Add(2)
+	go func() { defer wg.Done(); _, _ = a.Call(taskJSON(4)) }()
+	go func() { defer wg.Done(); _, _ = b.Call(taskJSON(4)) }()
+	wg.Wait()
+	if p := peak.Load(); p > 2 {
+		t.Fatalf("peak concurrent children = %d, want <= 2 (shared limiter)", p)
+	}
+}
+
+// TestDelegateTasks_ZeroCapacityNormalized pins the defensive floor:
+// maxConcurrency < 1 must normalize to 1 (an unbuffered semaphore channel
+// would deadlock the acquire-before-spawn loop).
+func TestDelegateTasks_ZeroCapacityNormalized(t *testing.T) {
+	tool := &delegateTasksTool{
+		maxConcurrency: 0,
+		runTaskFn:      stubRun(5 * time.Millisecond),
+		odekPath:       "unused",
+	}
+	out, err := tool.Call(taskJSON(3))
+	if err != nil {
+		t.Fatalf("Call with maxConcurrency=0: %v", err)
+	}
+	if n := strings.Count(out, "status: success"); n != 3 {
+		t.Fatalf("expected 3 successful task summaries, got %d in: %s", n, out)
+	}
+}
+
+// TestDelegateTasks_ConcurrencyWaitEvent pins the queueing telemetry: a task
+// waiting longer than subagentWaitEventThreshold on the shared limiter emits
+// a subagent_concurrency_wait runtime event.
+func TestDelegateTasks_ConcurrencyWaitEvent(t *testing.T) {
+	old := subagentWaitEventThreshold
+	subagentWaitEventThreshold = 50 * time.Millisecond
+	t.Cleanup(func() { subagentWaitEventThreshold = old })
+
+	shared := make(chan struct{}, 1)
+	var mu sync.Mutex
+	var evs []events.Event
+	tool := &delegateTasksTool{
+		maxConcurrency: 1,
+		sharedSem:      shared,
+		runTaskFn:      stubRun(200 * time.Millisecond),
+		odekPath:       "unused",
+	}
+	tool.SetEventEmitter(func(e events.Event) {
+		mu.Lock()
+		evs = append(evs, e)
+		mu.Unlock()
+	})
+	if _, err := tool.Call(taskJSON(2)); err != nil {
+		t.Fatalf("Call: %v", err)
+	}
+	mu.Lock()
+	defer mu.Unlock()
+	for _, e := range evs {
+		if e.Type == subagentConcurrencyWaitEvent {
+			return
+		}
+	}
+	t.Fatalf("expected a subagent_concurrency_wait event, got %d events", len(evs))
+}

docs/CONFIG.md

@@ -146,6 +146,7 @@ Most config knobs have a `ODEK_*` counterpart:
 | `ODEK_PROMPT_CACHING` | `prompt_caching` | bool |
 | `ODEK_STREAM` | `stream` | bool |
 | `ODEK_COMPACTION` | `compaction` | bool |
+| `ODEK_STREAM_IDLE_TIMEOUT_SECONDS` | `llm.stream_idle_timeout_seconds` | int |
 | `ODEK_SANDBOX_IMAGE` | `--sandbox-image` | string |
 | `ODEK_SANDBOX_NETWORK` | `--sandbox-network` | string |
 | `ODEK_SANDBOX_READONLY` | `--sandbox-readonly` | bool |
@@ -288,6 +289,22 @@ Top-level execution knobs. Every one also exists as a CLI flag and an `ODEK_*` e
 | `no_agents` | `false` | Skip loading project `AGENTS.md` |
 | `system` | built-in | Override the system-prompt identity layer — name/mission/persona (operator-only; rejected from project configs). The invariant security pillar is always composed on top and cannot be overridden. |
 
+## LLM client (`llm`)
+
+Tunes the shared LLM client (streaming and buffered calls share one retry policy):
+
+```json
+{
+  "llm": {
+    "stream_idle_timeout_seconds": 120
+  }
+}
+```
+
+| Field | Default | Description |
+|-------|---------|-------------|
+| `stream_idle_timeout_seconds` | `120` | Time between SSE events (keepalives count) before the stream is dropped and retried. Thinking models can spend minutes before their first event — raise it if long-thinking models hit `stream idle` errors. Floor 5s; `0` keeps the default. Eight retry attempts with jittered exponential backoff (and `Retry-After` honor) are shared with the buffered client; billing/quota errors fail fast. |
+
 ## Dangerous-operations policy (`dangerous`)
 
 The `dangerous` section is the operator's safety policy for tool calls. Every shell command, file write, and network operation is classified into a risk class, and the class maps to an action. Project-level `./odek.json` cannot set this section.

docs/EXTENSIONS.md

@@ -163,7 +163,7 @@ odek can emit a structured runtime event stream: **one JSON object per line
   `tool_call_started`, `tool_call_completed`, `tool_call_failed`,
   `session_saved`, `context_trimmed`, `budget_exceeded`, `run_completed`,
   `run_failed`, `plan_created`, `plan_updated`, `subagent_denied`,
-  `subagent_spawned`, `subagent_completed`.
+  `subagent_spawned`, `subagent_completed`, `subagent_concurrency_wait`.
 - `run_id` is a random 128-bit hex identifier generated per agent run and
   stamped on every event of that run. `session_id` appears once the session
   is known; earlier events omit it. `iteration` is the 1-based loop

docs/STREAMING.md

@@ -97,3 +97,7 @@ The reasoning block is dimmed with a single 🧠 cue, the answer follows after a
 - Streaming requests use a pooled HTTP client without a client-level timeout (`transport.NewPooledClientNoDeadline`) — a whole-request `http.Client.Timeout` would kill long body reads — sharing the connection pool with the buffered client. Deadlines are enforced per request via context.
 - The engine wires streaming through `loop.Engine.SetStream` / `SetDeltaHandler`, following the existing optional-callback pattern (`SetSignalHandler`, `SetToolEventHandler`).
 - Offline test coverage lives in `internal/llm/stream_test.go` (the provider-variance and failure-mode matrix) and `internal/loop/loop_test.go` (engine dispatch and the buffered default).
+
+## Idle watchdog
+
+A stream that produces no SSE events — keepalive comment lines count — for `llm.stream_idle_timeout_seconds` (default **120s**, env `ODEK_STREAM_IDLE_TIMEOUT_SECONDS`) is dropped and retried like any transient failure, as long as nothing was emitted yet. Once deltas have been delivered, an idle abort is never retried (that would duplicate text); the partial result is surfaced with the error. Eight attempts with jittered exponential backoff are shared with the buffered client.

docs/SUBAGENTS.md

@@ -141,7 +141,7 @@ On failure:
 2. **Validates**: rejects empty, >8 tasks, or malformed JSON
 3. **Writes** each task to a temp file (`odek-task-*.json`) — avoids CLI argument length limits (useful for 100KB+ context)
 4. **Spawns** `odek subagent --task <file> --quiet` for each task
-5. **Limits concurrency** via a buffered channel semaphore (default: 3, max: configurable)
+5. **Limits concurrency** via a **process-wide** buffered-channel semaphore (default: 3, max: configurable) — sibling delegate_tasks calls in one batch and concurrent `odek serve` sessions share the same bound, since provider plans are account-wide
 6. **Collects** JSON result from each subprocess stdout
 7. **Returns** a formatted summary with all sub-agent results tagged by task number
 

internal/config/llm.go

@@ -0,0 +1,35 @@
+package config
+
+import "time"
+
+// LLMConfig tunes the shared LLM client (internal/llm). Nil section = the
+// built-in defaults.
+type LLMConfig struct {
+	// StreamIdleTimeoutSeconds caps the time between SSE events (keepalive
+	// comment lines count) before the stream is dropped and retried.
+	// Thinking models can legitimately spend minutes before their first
+	// event, so the built-in default is generous (120s). 0 keeps the
+	// default. Config: llm.stream_idle_timeout_seconds,
+	// ODEK_STREAM_IDLE_TIMEOUT_SECONDS.
+	StreamIdleTimeoutSeconds int `json:"stream_idle_timeout_seconds,omitempty"`
+}
+
+// llmStreamIdleTimeoutFrom merges the file value with the env override (env
+// wins, per the priority chain) and clamps to a sane floor. Returns 0 when
+// unset so the caller keeps the llm package default.
+func llmStreamIdleTimeoutFrom(llm *LLMConfig, envSeconds *int) time.Duration {
+	v := 0
+	if llm != nil {
+		v = llm.StreamIdleTimeoutSeconds
+	}
+	if envSeconds != nil && *envSeconds > 0 {
+		v = *envSeconds
+	}
+	if v <= 0 {
+		return 0
+	}
+	if v < 5 {
+		v = 5
+	}
+	return time.Duration(v) * time.Second
+}