diff --git a/AGENTS.md b/AGENTS.md index 2b2a0e3..719e001 100644 --- a/AGENTS.md +++ b/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. diff --git a/cmd/odek/main.go b/cmd/odek/main.go index 3400fcd..3cd7db0 100644 --- a/cmd/odek/main.go +++ b/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, diff --git a/cmd/odek/subagent_tool.go b/cmd/odek/subagent_tool.go index 7ec47c8..424a44c 100644 --- a/cmd/odek/subagent_tool.go +++ b/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 ///; @@ -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 diff --git a/cmd/odek/subagent_tool_concurrency_test.go b/cmd/odek/subagent_tool_concurrency_test.go new file mode 100644 index 0000000..bfdcf45 --- /dev/null +++ b/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)) +} diff --git a/docs/CONFIG.md b/docs/CONFIG.md index e5a876b..228e4a6 100644 --- a/docs/CONFIG.md +++ b/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. diff --git a/docs/EXTENSIONS.md b/docs/EXTENSIONS.md index 476bbe6..d90bc55 100644 --- a/docs/EXTENSIONS.md +++ b/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 diff --git a/docs/STREAMING.md b/docs/STREAMING.md index e9c0785..efc2d59 100644 --- a/docs/STREAMING.md +++ b/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. diff --git a/docs/SUBAGENTS.md b/docs/SUBAGENTS.md index 4ecea4d..b6360c7 100644 --- a/docs/SUBAGENTS.md +++ b/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 --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 diff --git a/internal/config/llm.go b/internal/config/llm.go new file mode 100644 index 0000000..7394de2 --- /dev/null +++ b/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 +} diff --git a/internal/config/llm_test.go b/internal/config/llm_test.go new file mode 100644 index 0000000..481cf99 --- /dev/null +++ b/internal/config/llm_test.go @@ -0,0 +1,29 @@ +package config + +import ( + "testing" + "time" +) + +func TestLLMStreamIdleTimeoutFrom(t *testing.T) { + cases := []struct { + name string + llm *LLMConfig + env *int + want time.Duration + }{ + {"nil section, no env", nil, nil, 0}, + {"file value", &LLMConfig{StreamIdleTimeoutSeconds: 90}, nil, 90 * time.Second}, + {"env wins over file", &LLMConfig{StreamIdleTimeoutSeconds: 90}, testIntPtr(30), 30 * time.Second}, + {"env only", nil, testIntPtr(45), 45 * time.Second}, + {"env non-positive ignored", &LLMConfig{StreamIdleTimeoutSeconds: 90}, testIntPtr(0), 90 * time.Second}, + {"floor clamps to 5s", &LLMConfig{StreamIdleTimeoutSeconds: 1}, nil, 5 * time.Second}, + } + for _, tc := range cases { + if got := llmStreamIdleTimeoutFrom(tc.llm, tc.env); got != tc.want { + t.Errorf("%s: got %v, want %v", tc.name, got, tc.want) + } + } +} + +func testIntPtr(v int) *int { return &v } diff --git a/internal/config/loader.go b/internal/config/loader.go index 5ecaf07..c3bd77c 100644 --- a/internal/config/loader.go +++ b/internal/config/loader.go @@ -28,6 +28,7 @@ import ( "github.com/BackendStack21/odek/internal/danger" "github.com/BackendStack21/odek/internal/embedding" "github.com/BackendStack21/odek/internal/guard" + "github.com/BackendStack21/odek/internal/llm" "github.com/BackendStack21/odek/internal/maintenance" "github.com/BackendStack21/odek/internal/mcpclient" "github.com/BackendStack21/odek/internal/memory" @@ -391,6 +392,13 @@ type FileConfig struct { // Only used by `odek serve`. TrustedProxies []string `json:"trusted_proxies,omitempty"` + // LLM tunes the shared LLM client (internal/llm). Currently: the SSE + // stream idle watchdog — time between events (keepalives count) before + // the stream is dropped and retried. Thinking models can spend minutes + // before their first event; 0 keeps the built-in default (120s). + // Config: llm.stream_idle_timeout_seconds, ODEK_STREAM_IDLE_TIMEOUT_SECONDS. + LLM *LLMConfig `json:"llm,omitempty"` + // Telegram configures the Telegram bot integration. Telegram *telegram.TelegramConfig `json:"telegram,omitempty"` @@ -2215,6 +2223,13 @@ func LoadConfig(cli CLIFlags) ResolvedConfig { } } + // LLM client tuning (llm section + ODEK_STREAM_IDLE_TIMEOUT_SECONDS): + // env wins over the file value, per the priority chain. 0/unset keeps + // the llm package default (120s). + if d := llmStreamIdleTimeoutFrom(cfg.LLM, envIntPtr("ODEK_STREAM_IDLE_TIMEOUT_SECONDS")); d > 0 { + llm.SetStreamIdleTimeout(d) + } + // API key fallback chain: resolved → DEEPSEEK_API_KEY → OPENAI_API_KEY if resolved.APIKey == "" { resolved.APIKey = os.Getenv("DEEPSEEK_API_KEY") diff --git a/internal/llm/set_idle_test.go b/internal/llm/set_idle_test.go new file mode 100644 index 0000000..098361a --- /dev/null +++ b/internal/llm/set_idle_test.go @@ -0,0 +1,27 @@ +package llm + +import ( + "testing" + "time" +) + +// TestSetStreamIdleTimeout pins the setter contract: positive values apply, +// non-positive values are ignored (the built-in default stands). +func TestSetStreamIdleTimeout(t *testing.T) { + orig := streamIdleTimeout + t.Cleanup(func() { streamIdleTimeout = orig }) + + SetStreamIdleTimeout(5 * time.Second) + if streamIdleTimeout != 5*time.Second { + t.Fatalf("streamIdleTimeout = %v, want 5s", streamIdleTimeout) + } + if StreamIdleTimeout() != 5*time.Second { + t.Fatalf("StreamIdleTimeout() = %v, want 5s", StreamIdleTimeout()) + } + + SetStreamIdleTimeout(0) + SetStreamIdleTimeout(-1 * time.Second) + if streamIdleTimeout != 5*time.Second { + t.Fatalf("non-positive override applied; streamIdleTimeout = %v, want 5s", streamIdleTimeout) + } +} diff --git a/internal/llm/stream.go b/internal/llm/stream.go index 06ac0a0..1da28d6 100644 --- a/internal/llm/stream.go +++ b/internal/llm/stream.go @@ -66,7 +66,26 @@ func (e *StreamAbortedError) Unwrap() error { return e.Reason } // streamIdleTimeout bounds the silence between SSE events (keepalive // comments reset it). Package var so tests can shorten it. -var streamIdleTimeout = 60 * time.Second +// streamIdleTimeout is the SSE idle watchdog: the time between events +// (keepalive comment lines count) before the stream is dropped and retried. +// Thinking models can legitimately spend minutes before their first event, +// so the default is generous (120s) and operator-configurable via +// llm.stream_idle_timeout_seconds / ODEK_STREAM_IDLE_TIMEOUT_SECONDS +// (see SetStreamIdleTimeout). +var streamIdleTimeout = 120 * time.Second + +// SetStreamIdleTimeout overrides the SSE idle watchdog. Call at startup, +// before the first request; non-positive values are ignored. +func SetStreamIdleTimeout(d time.Duration) { + if d > 0 { + streamIdleTimeout = d + } +} + +// StreamIdleTimeout reports the active idle watchdog (introspection/tests). +func StreamIdleTimeout() time.Duration { + return streamIdleTimeout +} // CallStream sends a chat completion request with stream:true and delivers // fragments to cb as they arrive, returning the fully assembled result —