From f6aea98512c29c858def4c7e4c16cc97217a36aa Mon Sep 17 00:00:00 2001 From: Rolando Santamaria Maso Date: Tue, 28 Jul 2026 22:04:28 +0200 Subject: [PATCH 1/2] =?UTF-8?q?feat(agent):=20long-horizon=20readiness=20?= =?UTF-8?q?=E2=80=94=20tool=20heartbeat,=20shell=20timeout,=20bounded=20pr?= =?UTF-8?q?obes,=20telegram=20per-turn=20persist?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Four gaps found in a long-horizon task review (per-turn session persistence for CLI/REPL/serve landed separately in #122): 1. Tool-running heartbeat (internal/loop, internal/render, odek.go, cmd/odek/telegram.go): a single long tool call (e.g. shell running a test suite for minutes) showed a static spinner with zero feedback inside the shell tool's 30-minute buffered window. The loop now wraps each tool call with a watchdog that emits a 'tool_running' SignalEvent every 60s (toolHeartbeatInterval, test-overridable) until the call returns — closed on normal, error, and panic paths and on engine-context cancel, so no goroutine leaks. CLI renders '⏳ still running ()'; Telegram surfaces it even when skills.verbose is off (a silent multi-minute call looks like a hang); serve forwards it generically. 2. shell timeout_seconds (cmd/odek/shell.go): the LLM can now bound long commands explicitly (schema + Description advise it for builds/test suites). Clamped to [1, 1800]s via clampShellTimeoutSeconds, mirroring parallel_shell's cap; zero/ negative treated as absent. Output stays fully buffered — the heartbeat is the progress signal. 3. Bounded Docker availability probes: dockerAvailable() in cmd/odek/main_test.go and internal/sandbox/sandbox_test.go, plus the 'docker image inspect' probe in internal/sandbox, now use exec.CommandContext with a 5s timeout — a stopped/starting daemon can no longer hang the test suite or sandbox setup. 4. Telegram per-turn persistence (internal/telegram/session.go, cmd/odek/telegram.go): SessionManager.SaveNoIndex mirrors Save via Store.SaveNoIndex (no per-iteration embedding call) but does NOT advance the user-visible TurnCount — only a completed turn's final Save may. handleChatMessage wires SetMessagesPersistCallback with dropDanglingToolCalls (extracted from persistPartialMessages) and a skip-on-trim length check against the latest persisted state. /stop, restarts, and crashes now resume from the last completed step; SIGKILL may still lose the in-flight step (documented in docs/CLI.md). Docs: docs/CLI.md continue-row resume semantics; AGENTS.md testing guidance (scope test runs, -race on changed packages only, buffered shell + timeout_seconds) and shell.go layout line. Tests: TestToolHeartbeat_LongRunningToolEmitsSignals (fires during, stops after), shell timeout/clamp/non-positive tests, TestSessionManager_SaveNoIndex (no vector indexing, no TurnCount inflation, final Save indexes). Verified with clean env: go build ./... + go vet ./..., affected package tests, -race on internal/loop + internal/telegram, full ./cmd/odek suite (24.4s), golangci-lint 0 issues. --- AGENTS.md | 13 ++-- cmd/odek/main.go | 19 ++++-- cmd/odek/main_test.go | 6 +- cmd/odek/shell.go | 37 +++++++++-- cmd/odek/shell_test.go | 77 ++++++++++++++++++++++ cmd/odek/telegram.go | 39 +++++++++-- docs/CLI.md | 2 +- internal/loop/loop.go | 42 ++++++++++++ internal/loop/signal.go | 5 ++ internal/loop/signal_test.go | 105 +++++++++++++++++++++++++++++- internal/render/render.go | 11 ++++ internal/render/render_test.go | 5 ++ internal/sandbox/sandbox.go | 9 ++- internal/sandbox/sandbox_test.go | 9 ++- internal/telegram/session.go | 46 +++++++++++++ internal/telegram/session_test.go | 69 ++++++++++++++++++++ odek.go | 2 + 17 files changed, 469 insertions(+), 27 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index f28a21b4..2f567f26 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -20,7 +20,7 @@ odek.go Public API (Config, New, Run, Close, ModelProfile, cmd/odek/ main.go CLI entry point, flag parsing, commands, sandbox setup, system prompt dispatch.go CLI subcommand dispatch - shell.go Built-in shell tool (local or docker exec; danger-gated) + shell.go Built-in shell tool (local or docker exec; danger-gated; optional timeout_seconds) serve.go Web UI server (HTTP + WebSocket; @-resource completion) repl.go Interactive REPL with multi-turn session support repl_editor.go Terminal raw-mode input editor @@ -55,7 +55,7 @@ cmd/odek/ *_test.go 250+ unit + E2E tests covering all tools internal/ llm/ OpenAI-compatible HTTP client with reasoning_content support - loop/ ReAct engine: observe → think → parallel-act → repeat. signal.go — SignalEvent observability (context_trimmed, tool_recovery). + loop/ ReAct engine: observe → think → parallel-act → repeat. signal.go — SignalEvent observability (context_trimmed, tool_recovery, tool_running heartbeat). tool/ Thread-safe tool registry, clarify.go, send_message.go danger/ Command/URL classification + bypass-resistant tokenizer. TTYApprover with friction mode. auth/ Interactive approval system @@ -92,7 +92,7 @@ ReAct cycle: observe → think → act → repeat. - **Interaction modes** — engaging (narrated), enhance (persistent), verbose (raw), off. - Max 300 iterations by default. - **Post-response async processing** — skill learning, episode extraction, and per-turn extended-memory extraction run in background goroutines tracked by `MemoryManager.RunBackground`; `Agent.Close` drains them via `WaitForBackground` (bounded, ~15s) so the work survives CLI exit without hanging `odek run`. -- **Per-turn session persistence** — the loop fires an optional `SetMessagesPersistCallback` after each completed step (after a tool batch's result messages, and after the final assistant message) with a freshly-copied message snapshot. `run --session`, `continue`, the REPL, and `odek serve` wire it to `Store.SaveNoIndex` — an atomic save that skips the vector-index update (embedding can be a remote HTTP call and must not fire every iteration) — so an interrupted run (Ctrl-C, SIGTERM, crash) can be resumed via `odek continue` from the last completed step. Error/interrupt paths additionally persist the partial history minus any trailing assistant message with unanswered tool calls (`persistPartialMessages`). The final successful save still goes through `Save`/`Append`, updating the semantic index once per completed turn. If the loop trimmed history in place, the callback skips the save rather than overwriting richer persisted state. `odek serve` filters dynamically-injected system messages (skills/memory/episodes) out of persisted snapshots, preserving the session's original leading system message. +- **Per-turn session persistence** — the loop fires an optional `SetMessagesPersistCallback` after each completed step (after a tool batch's result messages, and after the final assistant message) with a freshly-copied message snapshot. `run --session`, `continue`, the REPL, `odek serve`, and the Telegram bot wire it to `Store.SaveNoIndex` (Telegram via `SessionManager.SaveNoIndex`) — an atomic save that skips the vector-index update (embedding can be a remote HTTP call and must not fire every iteration) — so an interrupted run (Ctrl-C, SIGTERM, crash) can be resumed via `odek continue` from the last completed step. Error/interrupt paths additionally persist the partial history minus any trailing assistant message with unanswered tool calls (`persistPartialMessages` / `dropDanglingToolCalls`). The final successful save still goes through `Save`/`Append`, updating the semantic index once per completed turn. If the loop trimmed history in place, the callback skips the save rather than overwriting richer persisted state. `odek serve` filters dynamically-injected system messages (skills/memory/episodes) out of persisted snapshots, preserving the session's original leading system message. - **Storage maintenance janitor** — `maintenance.Start` (internal/maintenance) runs a periodic sweep of `~/.odek` (expired sessions/audit/plans, oversized-log rotation, media sweep, skill skip-list GC) inside `odek telegram`, `odek serve`, and `odek schedule daemon` when `maintenance.enabled` is set; `odek cleanup [--dry-run]` runs the same sweep on demand. Session files are also trimmed at write time (oldest groups first, system message kept, `[Session storage limit: ...]` marker persisted into the transcript) when they would exceed `MaxSessionFileBytes`, so a session can never grow past the load cap. Operator-only config — project `./odek.json` cannot set it. See docs/MAINTENANCE.md. - **Artifact-aware file search** — `search_files` and `multi_grep` skip build/artifact directories (`node_modules`, `vendor`, `.git`, `__pycache__`, `.venv`, etc.) automatically, reducing noise and speeding scans. - **Semantic session search** — the `session_search` tool uses go-vector RandomProjections + k-NN for semantic similarity search through session content, with a two-tier pipeline: vector index (fast, ~1ms) → deepSearch fallback (exhaustive, slower). @@ -257,4 +257,9 @@ go test -fuzz=FuzzParseSkillContent -fuzztime=30s ./internal/skills/ go test -fuzz=FuzzSessionLoad -fuzztime=30s ./internal/session/ ``` -Note: MCP client E2E tests build the fakeserver from `internal/mcpclient/testdata/main.go` at test time (no pre-compiled binary). macOS temp dirs are classified as `LocalWrite` (not `SystemWrite`), and the Docker availability check verifies daemon reachability before running sandbox tests. +Note: MCP client E2E tests build the fakeserver from `internal/mcpclient/testdata/main.go` at test time (no pre-compiled binary). macOS temp dirs are classified as `LocalWrite` (not `SystemWrite`), and the Docker availability check verifies daemon reachability (5s timeout) before running sandbox tests. + +Agent workflow notes: + +- **Scope test runs.** Prefer `go test -count=1 ` over `./...`; run `-race` only on changed packages. A full `go test -race ./...` takes several minutes (race-instrumented build + 5–10× runtime slowdown). +- **The `shell` tool is fully buffered.** Nothing is shown or returned until the command exits, and the default timeout is 30 minutes — a long command looks "stuck" even though it is running (a `tool_running` heartbeat signal fires every 60s). Set `timeout_seconds` explicitly for known long-running commands (builds, test suites) so a genuinely stuck command fails fast, and pass `go test -timeout` for test runs. diff --git a/cmd/odek/main.go b/cmd/odek/main.go index 951d6074..b9652478 100644 --- a/cmd/odek/main.go +++ b/cmd/odek/main.go @@ -2447,6 +2447,19 @@ func expandHome(path string) string { // ── Continue (Multi-Turn) ───────────────────────────────────────────── +// dropDanglingToolCalls returns messages with any trailing assistant messages +// that carry unanswered tool calls removed. Their tool results never +// completed, and resuming with dangling tool calls is an invalid request for +// OpenAI-compatible APIs. +func dropDanglingToolCalls(messages []llm.Message) []llm.Message { + for len(messages) > 0 && + messages[len(messages)-1].Role == "assistant" && + len(messages[len(messages)-1].ToolCalls) > 0 { + messages = messages[:len(messages)-1] + } + return messages +} + // persistPartialMessages saves the in-flight message history of an // interrupted/failed run so the session keeps progress up to the last // completed step. A trailing assistant message with unanswered tool calls @@ -2456,11 +2469,7 @@ func persistPartialMessages(store *session.Store, sess *session.Session, message if store == nil || sess == nil || len(messages) == 0 { return } - for len(messages) > 0 && - messages[len(messages)-1].Role == "assistant" && - len(messages[len(messages)-1].ToolCalls) > 0 { - messages = messages[:len(messages)-1] - } + messages = dropDanglingToolCalls(messages) if len(messages) < len(sess.Messages) { // The loop trimmed history in place — keep the richer state already // persisted by the per-turn callback instead of overwriting it. diff --git a/cmd/odek/main_test.go b/cmd/odek/main_test.go index 4b1faeda..b54c7c01 100644 --- a/cmd/odek/main_test.go +++ b/cmd/odek/main_test.go @@ -2,6 +2,7 @@ package main import ( "bytes" + "context" "encoding/json" "fmt" "io" @@ -765,7 +766,10 @@ func dockerAvailable() bool { return false } // Verify the daemon is actually reachable (not just the CLI installed). - cmd := exec.Command("docker", "info") + // Bounded so a stopped/starting daemon cannot hang the test suite. + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + cmd := exec.CommandContext(ctx, "docker", "info") cmd.Stdout = nil cmd.Stderr = nil return cmd.Run() == nil diff --git a/cmd/odek/shell.go b/cmd/odek/shell.go index bacf12c4..2ab09f87 100644 --- a/cmd/odek/shell.go +++ b/cmd/odek/shell.go @@ -24,6 +24,19 @@ import ( // immediately regardless of this backstop. const defaultShellTimeout = 30 * time.Minute +// clampShellTimeoutSeconds bounds an LLM-provided timeout_seconds to +// [1, 1800] seconds, mirroring parallel_shell's per-command cap so the +// agent can tighten the backstop but never exceed it. +func clampShellTimeoutSeconds(sec int) int { + if sec < 1 { + return 1 + } + if max := int(defaultShellTimeout / time.Second); sec > max { + return max + } + return sec +} + // maxShellOutputBytes caps the stdout + stderr captured from a single shell // command to prevent memory DoS from commands that dump huge files. const maxShellOutputBytes = 1 << 20 // 1 MiB @@ -121,7 +134,10 @@ In host mode (default), commands run with the same permissions as the odek proce Risk classes: safe, local_write, system_write, destructive, network_egress, code_execution, install, unknown, blocked High-risk operations may prompt for approval (configurable via dangerous section in odek.json). -The gate fails closed: an unrecognised command classifies as "unknown" and is denied by default.` +The gate fails closed: an unrecognised command classifies as "unknown" and is denied by default. + +Output is fully buffered: nothing is returned until the command finishes. For known long-running +commands (builds, test suites), set timeout_seconds explicitly so a stuck command fails fast.` } func (t *shellTool) Schema() any { @@ -136,6 +152,10 @@ func (t *shellTool) Schema() any { "type": "string", "description": "Optional: explain what this command does and why. Shown in the approval prompt for high-risk operations.", }, + "timeout_seconds": map[string]any{ + "type": "integer", + "description": "Optional: per-command timeout in seconds. Set it explicitly for known long-running commands (builds, test suites); values are clamped to 1800 max. Default: 1800.", + }, }, "required": []string{"command"}, } @@ -146,8 +166,9 @@ func (t *shellTool) Schema() any { // Both stdout and stderr are captured and merged into the return string. func (t *shellTool) Call(args string) (string, error) { var input struct { - Command string `json:"command"` - Description string `json:"description,omitempty"` + Command string `json:"command"` + Description string `json:"description,omitempty"` + TimeoutSeconds int `json:"timeout_seconds,omitempty"` } if err := json.Unmarshal([]byte(args), &input); err != nil { return "", fmt.Errorf("shell: parse args: %w", err) @@ -162,8 +183,9 @@ func (t *shellTool) Call(args string) (string, error) { } // Bound execution: cancel with the agent context (Ctrl-C / turn timeout) - // and a generous backstop timeout so a stuck command can never wedge the - // agent forever. In sandbox mode this kills the host-side `docker exec` + // and a timeout — an LLM-provided timeout_seconds when set, otherwise a + // generous backstop — so a stuck command can never wedge the agent + // forever. In sandbox mode this kills the host-side `docker exec` // client, which unblocks the agent — but Docker does not propagate the // signal to the in-container process, so buildCmd also returns a // follow-up that kills the in-container process group explicitly. @@ -172,6 +194,11 @@ func (t *shellTool) Call(args string) (string, error) { if timeout <= 0 { timeout = defaultShellTimeout } + // An explicit timeout_seconds from the LLM overrides the tool default. + // Zero or negative is treated as absent (same as parallel_shell). + if input.TimeoutSeconds > 0 { + timeout = time.Duration(clampShellTimeoutSeconds(input.TimeoutSeconds)) * time.Second + } ctx, cancel := context.WithTimeout(base, timeout) defer cancel() diff --git a/cmd/odek/shell_test.go b/cmd/odek/shell_test.go index 6fea156b..e3ed899c 100644 --- a/cmd/odek/shell_test.go +++ b/cmd/odek/shell_test.go @@ -411,6 +411,83 @@ func stringSlicesEqual(a, b []string) bool { return true } +// TestShellTool_TimeoutSeconds verifies an explicit timeout_seconds from the +// LLM overrides the generous default: a 1s timeout kills `sleep 5` promptly +// and reports a timeout error. +func TestShellTool_TimeoutSeconds(t *testing.T) { + st := &shellTool{} + done := make(chan struct{}) + var err error + go func() { + _, err = st.Call(`{"command":"sleep 5","timeout_seconds":1}`) + close(done) + }() + select { + case <-done: + case <-time.After(10 * time.Second): + t.Fatal("Call did not return after timeout_seconds=1 — timeout_seconds was ignored") + } + if err == nil { + t.Fatal("expected a timeout error") + } + if !strings.Contains(err.Error(), "timed out") { + t.Errorf("error should mention the timeout, got: %v", err) + } +} + +// TestShellTool_TimeoutSeconds_FastCommand verifies a fast command succeeds +// with an explicit timeout_seconds well above its runtime. +func TestShellTool_TimeoutSeconds_FastCommand(t *testing.T) { + st := &shellTool{} + result, err := st.Call(`{"command":"echo hello","timeout_seconds":10}`) + if err != nil { + t.Fatalf("Call() error: %v", err) + } + if !strings.Contains(result, "hello") { + t.Errorf("result = %q, want it to contain 'hello'", result) + } +} + +// TestClampShellTimeoutSeconds covers the clamp boundaries: values below 1 +// floor to 1, values above the 1800s max clamp down, in-range values pass +// through unchanged. +func TestClampShellTimeoutSeconds(t *testing.T) { + cases := []struct { + in, want int + }{ + {1, 1}, + {10, 10}, + {1800, 1800}, + {1801, 1800}, + {3600, 1800}, + {1 << 30, 1800}, + } + for _, c := range cases { + if got := clampShellTimeoutSeconds(c.in); got != c.want { + t.Errorf("clampShellTimeoutSeconds(%d) = %d, want %d", c.in, got, c.want) + } + } +} + +// TestShellTool_TimeoutSeconds_NonPositive verifies zero and negative +// timeout_seconds values are treated as absent (same as parallel_shell): +// a fast command succeeds without being clamped to an immediate timeout. +func TestShellTool_TimeoutSeconds_NonPositive(t *testing.T) { + st := &shellTool{} + for _, args := range []string{ + `{"command":"echo ok","timeout_seconds":0}`, + `{"command":"echo ok","timeout_seconds":-5}`, + } { + result, err := st.Call(args) + if err != nil { + t.Fatalf("Call(%s) error: %v", args, err) + } + if !strings.Contains(result, "ok") { + t.Errorf("Call(%s) result = %q, want it to contain 'ok'", args, result) + } + } +} + func TestShellTool_PromptUser_ReusesTTYApprover(t *testing.T) { dir := t.TempDir() ttyPath := filepath.Join(dir, "tty") diff --git a/cmd/odek/telegram.go b/cmd/odek/telegram.go index bdda6e8a..de0dfd76 100644 --- a/cmd/odek/telegram.go +++ b/cmd/odek/telegram.go @@ -1882,15 +1882,21 @@ func handleChatMessage( sendAsync(bot, chatID, msg, &telegram.SendOpts{ReplyToMessageID: messageID}) }, AgentSignalHandler: func(event loop.SignalEvent) { - if skillsCfg == nil || !skillsCfg.Verbose { - return - } var msg string switch event.Type { - case "context_trimmed": - msg = fmt.Sprintf("✂️ Context trimmed (%s): %d group(s) dropped", event.Detail, event.Count) - case "tool_recovery": - msg = "🔁 Tool recovery: " + event.Tool + case "tool_running": + // Long-running tool heartbeat — surface even when not verbose: + // a silent multi-minute tool call looks like a hang. + msg = fmt.Sprintf("⏳ %s still running (%s)", event.Tool, event.Detail) + case "context_trimmed", "tool_recovery": + if skillsCfg == nil || !skillsCfg.Verbose { + return + } + if event.Type == "context_trimmed" { + msg = fmt.Sprintf("✂️ Context trimmed (%s): %d group(s) dropped", event.Detail, event.Count) + } else { + msg = "🔁 Tool recovery: " + event.Tool + } default: return } @@ -1928,6 +1934,25 @@ func handleChatMessage( chatRunInfos.LoadAndDelete(chatID) }() + // Persist per-turn progress so an interrupted run (/stop, restart, crash) + // can be resumed from the last completed step instead of losing the whole + // in-progress turn. Uses SaveNoIndex: embedding can be a remote HTTP call + // and must not fire every loop iteration — the final Save below still + // updates the vector index once per completed turn. + persistedLen := len(cs.Messages) + agent.SetMessagesPersistCallback(func(snapshot []llm.Message) { + trimmed := dropDanglingToolCalls(snapshot) + if len(trimmed) < persistedLen { + // The loop trimmed history in place — keep the richer state + // already persisted instead of overwriting it. + return + } + if err := sessionManager.SaveNoIndex(chatID, trimmed); err != nil { + log.Error("per-turn session persist", "chat_id", chatID, "error", err) + } + persistedLen = len(trimmed) + }) + // Run the agent with the full message history (multi-turn). response, updatedMessages, err := agent.RunWithMessages(agentCtx, cs.Messages) if err != nil { diff --git a/docs/CLI.md b/docs/CLI.md index d37fbbd3..393084d3 100644 --- a/docs/CLI.md +++ b/docs/CLI.md @@ -7,7 +7,7 @@ | `odek run [flags] ` | Execute a task with the agent loop (single-shot by default) | | `odek run --session [flags] ` | Execute and save conversation as a multi-turn session | | `odek run [--no-learn] [flags] ` | Execute with skill learning (on by default, use --no-learn to disable) | -| `odek continue [--id ] ` | Continue the most recent session (or by `--id`) | +| `odek continue [--id ] ` | Continue the most recent session (or by `--id`). Sessions persist per completed step: Ctrl-C/SIGTERM resumes from the last step; SIGKILL may lose the in-flight step | | `odek repl [flags]` | Interactive REPL mode (persistent multi-turn session). Accepts `--model`, `--thinking`, `--sandbox`, `--sandbox-*`, `--tool`, and `--no-tool` flags. | | `odek session list` | List sessions | | `odek session show [id]` | Show session details (default: latest) | diff --git a/internal/loop/loop.go b/internal/loop/loop.go index 06fd3e3f..f669713f 100644 --- a/internal/loop/loop.go +++ b/internal/loop/loop.go @@ -19,6 +19,41 @@ import ( "github.com/BackendStack21/odek/internal/tool" ) +// toolHeartbeatInterval is how often a "tool_running" signal fires while a +// single tool call is still executing. Package-level var so tests can +// override it. +var toolHeartbeatInterval = time.Minute + +// startToolHeartbeat launches a watchdog goroutine that emits a +// "tool_running" SignalEvent every toolHeartbeatInterval until the returned +// channel is closed or ctx is cancelled. The SignalHandler contract is +// non-blocking, so the heartbeat never delays tool execution or the loop. +// Callers must close the returned channel when the tool call ends (including +// panic paths) so the watchdog goroutine cannot leak. +func (e *Engine) startToolHeartbeat(ctx context.Context, toolName string) chan<- struct{} { + done := make(chan struct{}) + go func() { + ticker := time.NewTicker(toolHeartbeatInterval) + defer ticker.Stop() + start := time.Now() + for { + select { + case <-done: + return + case <-ctx.Done(): + return + case <-ticker.C: + e.emitSignal(SignalEvent{ + Type: "tool_running", + Tool: toolName, + Detail: fmt.Sprintf("running for %s", time.Since(start).Round(time.Second)), + }) + } + } + }() + return done +} + // ingestRecorderKey is the context key used to carry the per-run audit // ingest recorder through the agent loop to tool implementations. type ingestRecorderKey struct{} @@ -1451,11 +1486,18 @@ func (e *Engine) runLoop(ctx context.Context, messages []llm.Message) (string, [ if ctxTool, ok := t.(interface{ SetContext(context.Context) }); ok { ctxTool.SetContext(ctx) } + // Heartbeat watchdog: emit "tool_running" signals while + // this call is still executing so long-running tools + // (e.g. a shell test suite) don't look like a hang. + // Closed when the call returns, on panic paths too, so + // the watchdog goroutine always terminates. + stopHeartbeat := e.startToolHeartbeat(ctx, tcRef.Function.Name) // Capture any panic from the tool so it does not kill the agent. // The recovered message falls through to results[idx] like any // other tool error, so the LLM sees it and the consecutive-error // tracking counts it. func() { + defer close(stopHeartbeat) defer func() { if r := recover(); r != nil { output = fmt.Sprintf("error: tool %q panicked: %v", tcRef.Function.Name, r) diff --git a/internal/loop/signal.go b/internal/loop/signal.go index 3aab54ca..79f830ca 100644 --- a/internal/loop/signal.go +++ b/internal/loop/signal.go @@ -19,6 +19,11 @@ type SignalEvent struct { // "tool_recovery" — a tool failed repeatedly and the engine injected a // corrective hint so the model changes approach // (Tool = failing tool, Detail = the correction) + // "tool_running" — a single tool call is still executing after the + // heartbeat interval (Tool = tool name, Detail = + // human-readable elapsed, e.g. "running for 2m0s"). + // Fires every interval until the call returns, so + // long-running tools no longer look like a hang. Type string Detail string // human-readable detail (mode, correction text, etc.) Tool string // tool name for tool_recovery diff --git a/internal/loop/signal_test.go b/internal/loop/signal_test.go index 30c62d88..e96ad4d3 100644 --- a/internal/loop/signal_test.go +++ b/internal/loop/signal_test.go @@ -1,6 +1,37 @@ package loop -import "testing" +import ( + "context" + "fmt" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" + "time" + + "github.com/BackendStack21/odek/internal/llm" + "github.com/BackendStack21/odek/internal/tool" +) + +// blockingTool simulates a long-running tool call. +type blockingTool struct { + name string + delay time.Duration +} + +func (b *blockingTool) Name() string { return b.name } +func (b *blockingTool) Description() string { return "blocks for a fixed delay" } +func (b *blockingTool) Schema() any { + return map[string]any{ + "type": "object", + "properties": map[string]any{}, + } +} +func (b *blockingTool) Call(args string) (string, error) { + time.Sleep(b.delay) + return "ok", nil +} func TestEmitSignal_NilHandlerIsSafe(t *testing.T) { e := &Engine{} @@ -40,3 +71,75 @@ func TestSetSignalHandler_NilDisables(t *testing.T) { t.Error("handler should not fire after being set to nil") } } + +func TestToolHeartbeat_LongRunningToolEmitsSignals(t *testing.T) { + old := toolHeartbeatInterval + toolHeartbeatInterval = 50 * time.Millisecond + defer func() { toolHeartbeatInterval = old }() + + callCount := 0 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + callCount++ + if callCount == 1 { + fmt.Fprint(w, `{"choices":[{"message":{"content":"","tool_calls":[{"id":"call_1","function":{"name":"slow","arguments":"{}"}}]}}]}`) + } else { + fmt.Fprint(w, `{"choices":[{"message":{"content":"done"}}]}`) + } + })) + defer server.Close() + + slowTool := &blockingTool{name: "slow", delay: 300 * time.Millisecond} + registry := tool.NewRegistry([]tool.Tool{slowTool}) + client := llm.New(server.URL, "sk-test", "test-model", "", 0, 0) + engine := New(client, registry, 10, "", nil, 0) + + var mu sync.Mutex + var signals []SignalEvent + engine.SetSignalHandler(func(ev SignalEvent) { + mu.Lock() + signals = append(signals, ev) + mu.Unlock() + }) + + if _, err := engine.Run(context.Background(), "run the slow tool"); err != nil { + t.Fatalf("Run() error: %v", err) + } + + countBeats := func() int { + mu.Lock() + defer mu.Unlock() + n := 0 + for _, ev := range signals { + if ev.Type == "tool_running" { + n++ + } + } + return n + } + + mu.Lock() + for _, ev := range signals { + if ev.Type != "tool_running" { + continue + } + if ev.Tool != "slow" { + t.Errorf("tool_running signal tool = %q, want %q", ev.Tool, "slow") + } + if !strings.HasPrefix(ev.Detail, "running for ") { + t.Errorf("tool_running signal detail = %q, want \"running for ...\"", ev.Detail) + } + } + mu.Unlock() + + beats := countBeats() + if beats == 0 { + t.Fatal("expected at least one tool_running signal for a 300ms tool call at a 50ms interval") + } + + // After the call returned the watchdog must have stopped: the count + // stays stable across two more intervals. + time.Sleep(2 * toolHeartbeatInterval) + if after := countBeats(); after != beats { + t.Errorf("tool_running signals kept firing after the call returned: %d -> %d", beats, after) + } +} diff --git a/internal/render/render.go b/internal/render/render.go index a6d330cc..f432d04e 100644 --- a/internal/render/render.go +++ b/internal/render/render.go @@ -559,6 +559,17 @@ func (r *Renderer) ToolRecovery(tool, detail string) { fmt.Fprintln(r.w, r.style(yellow, fmt.Sprintf("🔁 tool recovery [%s]: %s", tool, r.truncate(detail, 80)))) } +// ToolRunning reports that a single tool call is still executing after the +// heartbeat interval, so a long-running tool does not look like a hang. +// Unlike the other signal renderers it is NOT gated behind memoryVerbose — +// the heartbeat is the only feedback during a multi-minute tool call. +func (r *Renderer) ToolRunning(tool, elapsed string) { + if r.disable() { + return + } + fmt.Fprintln(r.w, r.style(yellow, fmt.Sprintf("⏳ %s still running (%s)", tool, elapsed))) +} + // ── Helpers ─────────────────────────────────────────────────────────── // style wraps text in ANSI codes. Returns plain text when color is off. diff --git a/internal/render/render_test.go b/internal/render/render_test.go index aa7dc5c0..183716e9 100644 --- a/internal/render/render_test.go +++ b/internal/render/render_test.go @@ -744,6 +744,7 @@ func TestRenderer_SignalEvents(t *testing.T) { r.ContextTrimmed("survival", 4) r.ToolRecovery("shell", "try a different approach to the failing command") + r.ToolRunning("shell", "running for 2m0s") out := buf.String() if !strings.Contains(out, "context trimmed") || !strings.Contains(out, "survival") { t.Errorf("missing context trim signal: %q", out) @@ -751,6 +752,9 @@ func TestRenderer_SignalEvents(t *testing.T) { if !strings.Contains(out, "tool recovery") || !strings.Contains(out, "shell") { t.Errorf("missing tool recovery signal: %q", out) } + if !strings.Contains(out, "still running") || !strings.Contains(out, "2m0s") { + t.Errorf("missing tool running signal: %q", out) + } } func TestRenderer_MemoryEvents_NilSafe(t *testing.T) { @@ -760,6 +764,7 @@ func TestRenderer_MemoryEvents_NilSafe(t *testing.T) { r.MemoryEpisode("stored", "x") r.ContextTrimmed("proactive", 1) r.ToolRecovery("shell", "x") + r.ToolRunning("shell", "x") } func TestRenderer_NarratorMessage(t *testing.T) { diff --git a/internal/sandbox/sandbox.go b/internal/sandbox/sandbox.go index c839634b..cbe4d752 100644 --- a/internal/sandbox/sandbox.go +++ b/internal/sandbox/sandbox.go @@ -12,6 +12,7 @@ package sandbox import ( + "context" "crypto/sha256" "encoding/hex" "fmt" @@ -19,6 +20,7 @@ import ( "os/exec" "path/filepath" "strings" + "time" "github.com/BackendStack21/odek/internal/pathutil" ) @@ -87,7 +89,12 @@ func buildFromDockerfile() (string, error) { hash := sha256.Sum256(data) tag := "odek-sandbox:" + hex.EncodeToString(hash[:12]) - if _, err := exec.Command("docker", "image", "inspect", tag).CombinedOutput(); err != nil { + // Image-existence probe, bounded so a stopped/starting daemon cannot + // hang sandbox setup. + inspectCtx, inspectCancel := context.WithTimeout(context.Background(), 5*time.Second) + _, inspectErr := exec.CommandContext(inspectCtx, "docker", "image", "inspect", tag).CombinedOutput() + inspectCancel() + if inspectErr != nil { fmt.Fprintf(os.Stderr, "odek: building sandbox image from %s...\n", DockerfileName) build := exec.Command("docker", dockerBuildArgs(tag)...) build.Stderr = os.Stderr diff --git a/internal/sandbox/sandbox_test.go b/internal/sandbox/sandbox_test.go index de8653e4..c41c1525 100644 --- a/internal/sandbox/sandbox_test.go +++ b/internal/sandbox/sandbox_test.go @@ -1,22 +1,27 @@ package sandbox import ( + "context" "os" "os/exec" "path/filepath" "runtime" "strings" "testing" + "time" ) // dockerAvailable returns true if the docker CLI is installed and the // daemon is reachable. Tests that exercise real docker invocations skip -// otherwise. +// otherwise. The probe is bounded so a stopped/starting daemon cannot +// hang the test suite. func dockerAvailable() bool { if _, err := exec.LookPath("docker"); err != nil { return false } - cmd := exec.Command("docker", "info") + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + cmd := exec.CommandContext(ctx, "docker", "info") cmd.Stdout = nil cmd.Stderr = nil return cmd.Run() == nil diff --git a/internal/telegram/session.go b/internal/telegram/session.go index 7512c9b0..0589f382 100644 --- a/internal/telegram/session.go +++ b/internal/telegram/session.go @@ -143,6 +143,52 @@ func (sm *SessionManager) Save(chatID int64, messages []llm.Message) error { return sm.Store.Save(sess) } +// SaveNoIndex mirrors Save but persists through Store.SaveNoIndex, skipping +// the vector-index update. Embedding can be a remote HTTP call and must not +// fire every loop iteration — this is used by the per-turn persist callback +// for crash/interrupt-safe resume. The semantic index is still updated once +// per completed turn by the final Save. Unlike Save it does NOT increment +// TurnCount: it checkpoints mid-turn progress, and TurnCount is +// user-visible in /sessions — only a completed turn may advance it. +func (sm *SessionManager) SaveNoIndex(chatID int64, messages []llm.Message) error { + sm.Mu.Lock() + cs, ok := sm.Cache[chatID] + if ok { + // Copy-on-write: create a new ChatSession so existing pointers + // held by Load() callers are not mutated, avoiding data races. + updated := *cs + updated.Messages = messages + updated.LastActive = time.Now() + cs = &updated + sm.Cache[chatID] = cs + } else { + cs = &ChatSession{ + ChatID: chatID, + SessionID: fmt.Sprintf("tg-%d", chatID), + Messages: messages, + LastActive: time.Now(), + } + sm.Cache[chatID] = cs + } + // Snapshot fields needed after unlock to avoid data race: + sessionID := cs.SessionID + createdAt := cs.CreatedAt + turnCount := cs.TurnCount + sm.Mu.Unlock() + + sess := &session.Session{ + ID: sessionID, + CreatedAt: createdAt, + UpdatedAt: time.Now(), + Model: "", + Turns: turnCount, + Task: fmt.Sprintf("tg-%d", chatID), + Messages: messages, + } + + return sm.Store.SaveNoIndex(sess) +} + // Load retrieves a ChatSession from the cache first, then from the // backing store. If the session exists in the store but not in cache, // it is loaded from disk, converted to a ChatSession, and cached. diff --git a/internal/telegram/session_test.go b/internal/telegram/session_test.go index d55785d0..68f41aaa 100644 --- a/internal/telegram/session_test.go +++ b/internal/telegram/session_test.go @@ -911,3 +911,72 @@ func TestPrunePlans_NoDir(t *testing.T) { t.Errorf("PrunePlans on missing dir = %d, want 0", removed) } } + +// --------------------------------------------------------------------------- +// TestSessionManager_SaveNoIndex — per-turn persistence path: messages are +// persisted and load back, but the vector index is NOT updated (embedding +// can be a remote HTTP call and must not fire every loop iteration). A +// final Save still indexes the completed session. +// --------------------------------------------------------------------------- + +func TestSessionManager_SaveNoIndex(t *testing.T) { + sm, st := setupTestSessionManager(t) + if err := st.InitVectorIndex(nil); err != nil { + t.Fatalf("InitVectorIndex(nil): %v", err) + } + + const chatID int64 = 4242 + msgs := []llm.Message{ + {Role: "user", Content: "quixotic telegram per-turn persistence marker"}, + } + if err := sm.SaveNoIndex(chatID, msgs); err != nil { + t.Fatalf("SaveNoIndex() error: %v", err) + } + + // Persisted to disk and loadable through the manager and the store. + loaded, err := sm.Load(chatID) + if err != nil { + t.Fatalf("Load() after SaveNoIndex error: %v", err) + } + if loaded == nil || len(loaded.Messages) != 1 || + loaded.Messages[0].Content != "quixotic telegram per-turn persistence marker" { + t.Fatalf("loaded messages = %+v, want the saved user message", loaded) + } + + // NOT in the vector index. + if results, err := st.Vec.Search("quixotic", 5); err == nil { + for _, r := range results { + if r.SessionID == "tg-4242" { + t.Errorf("session tg-4242 must not appear in vector index after SaveNoIndex") + } + } + } + + // SaveNoIndex checkpoints mid-turn progress — it must NOT advance the + // user-visible TurnCount (only a completed turn's final Save may). + if loaded.TurnCount != 0 { + t.Errorf("TurnCount = %d after SaveNoIndex, want 0 (per-step saves must not inflate it)", loaded.TurnCount) + } + + // A final Save still indexes the completed session. + if err := sm.Save(chatID, msgs); err != nil { + t.Fatalf("Save() error: %v", err) + } + if cs, _ := sm.Load(chatID); cs.TurnCount != 1 { + t.Errorf("TurnCount = %d after final Save, want 1", cs.TurnCount) + } + results, err := st.Vec.Search("quixotic", 5) + if err != nil { + t.Fatalf("Search error: %v", err) + } + found := false + for _, r := range results { + if r.SessionID == "tg-4242" { + found = true + break + } + } + if !found { + t.Errorf("session tg-4242 not found in vector search results after Save: %+v", results) + } +} diff --git a/odek.go b/odek.go index b74d0a07..3ecab8aa 100644 --- a/odek.go +++ b/odek.go @@ -657,6 +657,8 @@ func New(cfg Config) (*Agent, error) { renderer.ContextTrimmed(ev.Detail, ev.Count) case "tool_recovery": renderer.ToolRecovery(ev.Tool, ev.Detail) + case "tool_running": + renderer.ToolRunning(ev.Tool, ev.Detail) } } }) From 70f5bd0f512d4a80cbd97aec7b934f64f2c95d92 Mon Sep 17 00:00:00 2001 From: Rolando Santamaria Maso Date: Tue, 28 Jul 2026 22:46:15 +0200 Subject: [PATCH 2/2] =?UTF-8?q?feat(agent):=203h+=20session=20readiness=20?= =?UTF-8?q?=E2=80=94=20LLM=20retry=20resilience,=20graceful=20iteration-bu?= =?UTF-8?q?dget=20exit,=20stall=20detection,=20compaction=20default-on?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A. LLM client resilience (internal/llm) — the retry budget was sized for second-scale blips and killed long unattended runs on the first sustained incident: - Retryable statuses now include 408, 500, and 529 (Anthropic overloaded) alongside 429/502/503/504. - maxRetries 3 -> 7 (8 attempts), exponential backoff 1s..30s with +/-25% jitter (crypto/rand), worst-case ~114s; Retry-After honored verbatim, cap raised 60s -> 120s. - Malformed JSON bodies and zero-choice 200 responses (transient gateway artifacts) now consume retry attempts instead of aborting immediately; validated inside the retry loop via validateCompletionBody with error strings unchanged. - retrySleep package var makes backoff injectable for fast tests. B. Iteration ceiling + stall guard (internal/loop): - max_iterations exhaustion no longer returns a bare error: the engine makes one final tool-less LLM call (30s bound, compaction-style side-call) for a partial-progress summary and returns it as the final answer marked '[Iteration budget reached - partial summary]'. Summarizer failure (or a non-compliant tool-call response) falls back to the original error unchanged. - Stall detection: 3 consecutive identical SUCCESSFUL tool calls (name+args fingerprint) inject a corrective hint and emit a tool_recovery signal, closing the polling-loop blind spot where only failing tools were detected. Hint, not enforcement - never aborts. C. Compaction ON by default + serve digest persistence: - ResolvedConfig seeds compaction=true; explicit false from ~/.odek/config.json, ./odek.json, ODEK_COMPACTION=false, or the new --no-compaction flag still disables (plumbing was already *bool end-to-end). odek.Config/loop.Engine stay explicit. - Config templates: global pins "compaction": true; local omits the key so projects inherit (an explicit false would actively disable). - odek serve's persist filter now preserves compaction digest system messages ([Compacted earlier context: prefix) instead of stripping them - resumed web sessions keep their compacted history. Other injected system messages (skills/memory/episodes) still dropped. - AGENTS.md: corrected stale '300 iterations' claim (actual: 90), documented the partial-summary exit, stall detection, compaction default, and serve digest preservation. docs/CLI.md + docs/CONFIG.md updated for --no-compaction and the new default. Tests: llm retry tests for 529/500/parse/zero-choice + jitter bounds; loop tests for partial-summary success/fallback/tool-call-rejection/ persistence and stall streak + reset; config tests for default-on and explicit-false-wins; serve filter tests for digest preservation. Verified with clean env: go build ./... + go vet ./..., go test -race ./internal/llm ./internal/loop ./internal/config, root package, full ./cmd/odek suite (multiple consecutive -count=1 passes). Note: go test -count=2 ./cmd/odek fails on path-confinement tests identically on clean main (pre-existing cross-test interference, not introduced here). --- AGENTS.md | 8 +- cmd/odek/main.go | 18 ++- cmd/odek/main_test.go | 67 +++++++++ cmd/odek/serve.go | 46 ++++-- cmd/odek/serve_api_test.go | 65 +++++++++ docs/CLI.md | 3 +- docs/CONFIG.md | 6 +- internal/config/loader.go | 16 ++- internal/config/loader_test.go | 55 +++++++ internal/llm/client.go | 99 +++++++++++-- internal/llm/client_test.go | 8 +- internal/llm/retry_test.go | 187 +++++++++++++++++++++++- internal/loop/loop.go | 172 +++++++++++++++++++++- internal/loop/loop_test.go | 254 ++++++++++++++++++++++++++++++++- internal/loop/signal.go | 9 +- odek.go | 7 +- 16 files changed, 958 insertions(+), 62 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index 2f567f26..af5e3074 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -87,12 +87,12 @@ ReAct cycle: observe → think → act → repeat. - LLM returns tool calls or a final answer. - **Parallel tool execution** — multiple independent tool calls run concurrently (`max_tool_parallel`, default: 4). - **Batch approval gate** (`internal/loop/loop.go`) — multiple risky tools shown at once in a single prompt. `classifyToolCall` now classifies every command inside `parallel_shell`, every path inside `batch_patch`, and the modern `browser` tool, shows full (untruncated) commands, and withholds blanket `SetTrustAll` when unclassifiable tools remain in the iteration. -- **Tool-failure recovery** — systematic recovery from tool call failures: retry transient errors, skip permanently failed tools, and continue the loop without crashing. -- **Context-limit protection** — graduated trimming when approaching the model's context window: old, large tool results are first replaced with a marker (the 4 most recent are always kept intact), then the oldest turn groups are dropped atomically (tool messages stay grouped with their parent assistant message). The protected head (system prompt, memory block, compaction digest, original task) is never dropped. A trim warning is injected before the most recent user message and updated in place with cumulative totals (including which tools lost earlier results). The token estimator counts tool-def parameter schemas and reasoning content, and the safety margin self-tightens (0.75 → 0.65) when provider-reported input tokens exceed the estimate by >15% (`margin_calibrated` signal). `trimToSurvival` (on provider context-length errors) keeps the system prompt, digest, original task, last 2 turn groups, and latest user message. Optional **rolling compaction** (`compaction` config / `--compaction` / `ODEK_COMPACTION`) summarizes dropped groups into a rolling digest system message (untrusted-wrapped) instead of losing them — one extra LLM call per trim. +- **Tool-failure recovery** — systematic recovery from tool call failures: retry transient errors, skip permanently failed tools, and continue the loop without crashing. Stall detection covers the successful-call case: 3 consecutive identical tool calls (same name + args) inject a corrective hint and fire a `tool_recovery` signal — a hint, not enforcement; the run is never aborted. +- **Context-limit protection** — graduated trimming when approaching the model's context window: old, large tool results are first replaced with a marker (the 4 most recent are always kept intact), then the oldest turn groups are dropped atomically (tool messages stay grouped with their parent assistant message). The protected head (system prompt, memory block, compaction digest, original task) is never dropped. A trim warning is injected before the most recent user message and updated in place with cumulative totals (including which tools lost earlier results). The token estimator counts tool-def parameter schemas and reasoning content, and the safety margin self-tightens (0.75 → 0.65) when provider-reported input tokens exceed the estimate by >15% (`margin_calibrated` signal). `trimToSurvival` (on provider context-length errors) keeps the system prompt, digest, original task, last 2 turn groups, and latest user message. **Rolling compaction** (on by default; disable via `compaction: false`, `ODEK_COMPACTION=false`, or `--no-compaction`) summarizes dropped groups into a rolling digest system message (untrusted-wrapped) instead of losing them — one extra LLM call per trim. - **Interaction modes** — engaging (narrated), enhance (persistent), verbose (raw), off. -- Max 300 iterations by default. +- Max 90 iterations by default. On budget exhaustion the engine makes one final tool-less LLM call for a partial-progress summary (30s bound) and returns it as the final answer marked `[Iteration budget reached — partial summary]`; if that call fails, the original `reached max iterations` error is returned. - **Post-response async processing** — skill learning, episode extraction, and per-turn extended-memory extraction run in background goroutines tracked by `MemoryManager.RunBackground`; `Agent.Close` drains them via `WaitForBackground` (bounded, ~15s) so the work survives CLI exit without hanging `odek run`. -- **Per-turn session persistence** — the loop fires an optional `SetMessagesPersistCallback` after each completed step (after a tool batch's result messages, and after the final assistant message) with a freshly-copied message snapshot. `run --session`, `continue`, the REPL, `odek serve`, and the Telegram bot wire it to `Store.SaveNoIndex` (Telegram via `SessionManager.SaveNoIndex`) — an atomic save that skips the vector-index update (embedding can be a remote HTTP call and must not fire every iteration) — so an interrupted run (Ctrl-C, SIGTERM, crash) can be resumed via `odek continue` from the last completed step. Error/interrupt paths additionally persist the partial history minus any trailing assistant message with unanswered tool calls (`persistPartialMessages` / `dropDanglingToolCalls`). The final successful save still goes through `Save`/`Append`, updating the semantic index once per completed turn. If the loop trimmed history in place, the callback skips the save rather than overwriting richer persisted state. `odek serve` filters dynamically-injected system messages (skills/memory/episodes) out of persisted snapshots, preserving the session's original leading system message. +- **Per-turn session persistence** — the loop fires an optional `SetMessagesPersistCallback` after each completed step (after a tool batch's result messages, and after the final assistant message) with a freshly-copied message snapshot. `run --session`, `continue`, the REPL, `odek serve`, and the Telegram bot wire it to `Store.SaveNoIndex` (Telegram via `SessionManager.SaveNoIndex`) — an atomic save that skips the vector-index update (embedding can be a remote HTTP call and must not fire every iteration) — so an interrupted run (Ctrl-C, SIGTERM, crash) can be resumed via `odek continue` from the last completed step. Error/interrupt paths additionally persist the partial history minus any trailing assistant message with unanswered tool calls (`persistPartialMessages` / `dropDanglingToolCalls`). The final successful save still goes through `Save`/`Append`, updating the semantic index once per completed turn. If the loop trimmed history in place, the callback skips the save rather than overwriting richer persisted state. `odek serve` filters dynamically-injected system messages (skills/memory/episodes) out of persisted snapshots, preserving the session's original leading system message **and** compaction digest messages (`[Compacted earlier context:` prefix) so resumed web sessions keep their compacted history. - **Storage maintenance janitor** — `maintenance.Start` (internal/maintenance) runs a periodic sweep of `~/.odek` (expired sessions/audit/plans, oversized-log rotation, media sweep, skill skip-list GC) inside `odek telegram`, `odek serve`, and `odek schedule daemon` when `maintenance.enabled` is set; `odek cleanup [--dry-run]` runs the same sweep on demand. Session files are also trimmed at write time (oldest groups first, system message kept, `[Session storage limit: ...]` marker persisted into the transcript) when they would exceed `MaxSessionFileBytes`, so a session can never grow past the load cap. Operator-only config — project `./odek.json` cannot set it. See docs/MAINTENANCE.md. - **Artifact-aware file search** — `search_files` and `multi_grep` skip build/artifact directories (`node_modules`, `vendor`, `.git`, `__pycache__`, `.venv`, etc.) automatically, reducing noise and speeding scans. - **Semantic session search** — the `session_search` tool uses go-vector RandomProjections + k-NN for semantic similarity search through session content, with a two-tier pipeline: vector index (fast, ~1ms) → deepSearch fallback (exhaustive, slower). diff --git a/cmd/odek/main.go b/cmd/odek/main.go index b9652478..5f01c8f8 100644 --- a/cmd/odek/main.go +++ b/cmd/odek/main.go @@ -416,6 +416,9 @@ func parseRunFlags(args []string) (runFlags, error) { case "--compaction": f.Compaction = boolPtr(true) i++ + case "--no-compaction": + f.Compaction = boolPtr(false) + i++ case "--session": f.Session = boolPtr(true) i++ @@ -663,6 +666,10 @@ done: f.Compaction = boolPtr(true) taskArgs = append(taskArgs[:j], taskArgs[j+1:]...) j-- + case "--no-compaction": + f.Compaction = boolPtr(false) + taskArgs = append(taskArgs[:j], taskArgs[j+1:]...) + j-- case "--sandbox-readonly": f.SandboxReadonly = boolPtr(true) taskArgs = append(taskArgs[:j], taskArgs[j+1:]...) @@ -821,6 +828,9 @@ func parseReplFlags(args []string) (replFlags, error) { case "--compaction": f.Compaction = boolPtr(true) i++ + case "--no-compaction": + f.Compaction = boolPtr(false) + i++ case "--interaction-mode": f.InteractionMode = args[i+1] i += 2 @@ -906,7 +916,8 @@ Run flags: --no-color Disable colored terminal output --no-agents Skip loading AGENTS.md from working directory --prompt-caching Enable prompt caching markers (Anthropic/DeepSeek/OpenAI) - --compaction Enable LLM-based rolling compaction of trimmed context + --compaction Enable LLM-based rolling compaction of trimmed context (default: on) + --no-compaction Disable rolling compaction (overrides config/default) --session Save conversation as a multi-turn session --learn Enable skill learning mode — on by default, no flag needed --no-learn Disable skill learning mode (overrides config/default) @@ -1008,7 +1019,7 @@ const globalConfigTemplate = `{ "max_iterations": 90, "max_tool_parallel": 4, "prompt_caching": false, - "compaction": false, + "compaction": true, "interaction_mode": "engaging", "no_color": false, "no_agents": false, @@ -1131,7 +1142,6 @@ const localConfigTemplate = `{ "max_iterations": 0, "max_tool_parallel": 0, "prompt_caching": false, - "compaction": false, "interaction_mode": "", "no_color": false, "no_agents": false, @@ -1239,7 +1249,7 @@ func initConfig(args []string) error { fmt.Println(" thinking Reasoning depth (enabled/disabled/low/medium/high)") fmt.Println(" max_iterations Max think→act cycles (default: 90)") fmt.Println(" prompt_caching Provider prompt caching (true/false)") - fmt.Println(" compaction Rolling LLM context compaction (true/false)") + fmt.Println(" compaction Rolling LLM context compaction (default: true)") fmt.Println(" interaction_mode engaging | enhance | verbose | off") fmt.Println(" sandbox Run in Docker sandbox (true/false)") fmt.Println(" system System prompt override") diff --git a/cmd/odek/main_test.go b/cmd/odek/main_test.go index b54c7c01..84d5c101 100644 --- a/cmd/odek/main_test.go +++ b/cmd/odek/main_test.go @@ -129,6 +129,56 @@ func TestParseRunFlags_AllFlags(t *testing.T) { } } +func TestParseRunFlags_CompactionFlags(t *testing.T) { + // --compaction and --no-compaction are explicit overrides; absent means + // "not set" (nil), so config/default resolution decides. + f, err := parseRunFlags([]string{"--compaction", "do the thing"}) + if err != nil { + t.Fatalf("parseRunFlags error: %v", err) + } + if f.Compaction == nil || !*f.Compaction { + t.Error("--compaction should set Compaction to true") + } + + f, err = parseRunFlags([]string{"--no-compaction", "do the thing"}) + if err != nil { + t.Fatalf("parseRunFlags error: %v", err) + } + if f.Compaction == nil || *f.Compaction { + t.Error("--no-compaction should set Compaction to false") + } + + // Trailing form after the task phrase. + f, err = parseRunFlags([]string{"do the thing", "--no-compaction"}) + if err != nil { + t.Fatalf("parseRunFlags error: %v", err) + } + if f.Compaction == nil || *f.Compaction { + t.Error("trailing --no-compaction should set Compaction to false") + } + if f.Task != "do the thing" { + t.Errorf("Task = %q, want %q", f.Task, "do the thing") + } + + f, err = parseRunFlags([]string{"do the thing"}) + if err != nil { + t.Fatalf("parseRunFlags error: %v", err) + } + if f.Compaction != nil { + t.Error("Compaction should be nil (not set) without a compaction flag") + } +} + +func TestParseReplFlags_NoCompaction(t *testing.T) { + f, err := parseReplFlags([]string{"--no-compaction", "x"}) + if err != nil { + t.Fatalf("parseReplFlags error: %v", err) + } + if f.Compaction == nil || *f.Compaction { + t.Error("--no-compaction should set Compaction to false") + } +} + func TestParseRunFlags_NoTask(t *testing.T) { _, err := parseRunFlags([]string{}) if err == nil { @@ -818,6 +868,12 @@ func TestInitConfig_Local(t *testing.T) { t.Errorf("local config must not contain %q (project configs may only enable it), got: %s", field, content) } } + // compaction defaults to ON; an explicit "compaction": false in a fresh + // project config would silently disable it, so the key must be omitted + // (inherit) rather than pinned. + if strings.Contains(content, "compaction") { + t.Errorf("local config must not pin compaction (default-on; omit to inherit), got: %s", content) + } // Must be valid JSON. var parsed map[string]any if err := json.Unmarshal(data, &parsed); err != nil { @@ -851,6 +907,11 @@ func TestInitConfig_Global(t *testing.T) { t.Errorf("global config should contain %q, got: %s", field, content) } } + // Compaction is default-on; the global template must pin it to true so a + // fresh operator config matches the documented default explicitly. + if !strings.Contains(content, `"compaction": true`) { + t.Errorf("global config should set compaction to true, got: %s", content) + } // Must be valid JSON. var parsed map[string]any if err := json.Unmarshal(data, &parsed); err != nil { @@ -981,6 +1042,9 @@ func TestInitConfig_LocalTemplateLoadsClean(t *testing.T) { if fc.Sandbox != nil || fc.SandboxReadonly != nil { t.Error("localConfigTemplate must not pin sandbox/sandbox_readonly (project configs may only enable, never disable)") } + if fc.Compaction != nil { + t.Error("localConfigTemplate must not pin compaction (default-on; omit the key to inherit)") + } if fc.Skills != nil && len(fc.Skills.Dirs) > 0 { t.Error("localConfigTemplate must not set skills.dirs (rejected from project configs)") } @@ -1008,6 +1072,9 @@ func TestInitConfig_GlobalTemplateLoadsClean(t *testing.T) { if fc.Skills == nil || fc.Skills.AutoSave == nil { t.Error("global template should include skills.auto_save") } + if fc.Compaction == nil || !*fc.Compaction { + t.Error("global template should pin compaction to true (the documented default)") + } } // TestInitConfig_RestrictivePermissions verifies that config files diff --git a/cmd/odek/serve.go b/cmd/odek/serve.go index ed17b160..435b7606 100644 --- a/cmd/odek/serve.go +++ b/cmd/odek/serve.go @@ -49,6 +49,35 @@ const maxPromptBytes = 1 * 1024 * 1024 // 1 MiB // maxModelIDBytes caps the length of a model ID supplied by the Web UI. const maxModelIDBytes = 128 +// compactionDigestPrefix mirrors the unexported digestMsgPrefix in +// internal/loop/loop.go (kept as a literal here to avoid widening the +// loop package's API surface). Rolling-compaction digest system messages +// must survive the per-turn persist filter — dropping them would make a +// resumed serve session lose its compacted history. +const compactionDigestPrefix = "[Compacted earlier context:" + +// filterPersistSnapshot selects which messages of a per-turn snapshot are +// persisted for a serve session. The session's own leading system message +// (head) is preserved; dynamically-injected system messages (skills, +// memory, episodes, trim warnings) are dropped so persisted snapshots +// don't accumulate internal injections or corrupt future origLen +// calculations; compaction digest system messages are kept so a resumed +// session retains its compacted history. +func filterPersistSnapshot(head, snapshot []llm.Message) []llm.Message { + filtered := make([]llm.Message, 0, len(snapshot)) + filtered = append(filtered, head...) + for i, m := range snapshot { + if i == 0 && len(head) > 0 { + continue // replaced by the session's original head + } + if m.Role == "system" && !strings.HasPrefix(m.Content, compactionDigestPrefix) { + continue + } + filtered = append(filtered, m) + } + return filtered +} + // modelIDPattern restricts model IDs to printable ASCII characters commonly // used by model providers (alphanumeric, punctuation, and path separators). var modelIDPattern = regexp.MustCompile(`^[A-Za-z0-9_.:/@-]+$`) @@ -1088,7 +1117,9 @@ func handlePrompt( // Mirror the store path below: dynamically-injected system messages // (skills, memory, episodes) are filtered out so persisted snapshots // don't accumulate internal injections or corrupt future origLen - // calculations. The session's own leading system message is preserved. + // calculations. The session's own leading system message and rolling- + // compaction digest system messages are preserved (see + // filterPersistSnapshot). if sess != nil { var head []llm.Message if len(sess.Messages) > 0 && sess.Messages[0].Role == "system" { @@ -1100,18 +1131,7 @@ func handlePrompt( // already persisted instead of overwriting it. return } - filtered := make([]llm.Message, 0, len(snapshot)) - filtered = append(filtered, head...) - for i, m := range snapshot { - if i == 0 && len(head) > 0 { - continue // replaced by the session's original head - } - if m.Role == "system" { - continue - } - filtered = append(filtered, m) - } - sess.Messages = filtered + sess.Messages = filterPersistSnapshot(head, snapshot) _ = store.SaveNoIndex(sess) }) } diff --git a/cmd/odek/serve_api_test.go b/cmd/odek/serve_api_test.go index 07b954be..a638d370 100644 --- a/cmd/odek/serve_api_test.go +++ b/cmd/odek/serve_api_test.go @@ -850,3 +850,68 @@ func TestHandleResourceSearch_LimitCapped(t *testing.T) { t.Errorf("expected results capped to 100, got %d", len(results)) } } + +// TestFilterPersistSnapshot verifies the per-turn persist filter: the +// session's leading system message and rolling-compaction digest system +// messages survive, while dynamically-injected system messages (skills, +// memory, episodes, trim warnings) are dropped. +func TestFilterPersistSnapshot(t *testing.T) { + head := []llm.Message{{Role: "system", Content: "You are odek."}} + snapshot := []llm.Message{ + {Role: "system", Content: "You are odek."}, + {Role: "system", Content: "## Skill: deploy\nDo the deploy dance."}, + {Role: "system", Content: "[Compacted earlier context: turns 1-8 summarized. User asked about the migration.]"}, + {Role: "user", Content: "continue the migration"}, + {Role: "system", Content: "[Context trimmed: 2 turn groups dropped]"}, + {Role: "assistant", Content: "On it."}, + } + + got := filterPersistSnapshot(head, snapshot) + + var systemMsgs []string + for _, m := range got { + if m.Role == "system" { + systemMsgs = append(systemMsgs, m.Content) + } + } + + // Exactly two system messages survive: the session head and the + // compaction digest. Skill injection and trim warning are dropped. + if len(systemMsgs) != 2 { + t.Fatalf("expected 2 system messages, got %d: %q", len(systemMsgs), systemMsgs) + } + if systemMsgs[0] != "You are odek." { + t.Errorf("system[0] = %q, want session head", systemMsgs[0]) + } + if !strings.HasPrefix(systemMsgs[1], compactionDigestPrefix) { + t.Errorf("system[1] = %q, want compaction digest (prefix %q)", systemMsgs[1], compactionDigestPrefix) + } + for _, s := range systemMsgs { + if strings.Contains(s, "Skill: deploy") || strings.HasPrefix(s, "[Context trimmed:") { + t.Errorf("dynamically-injected system message survived: %q", s) + } + } + + // Non-system messages pass through untouched. + var roles []string + for _, m := range got { + roles = append(roles, m.Role) + } + want := []string{"system", "system", "user", "assistant"} + if strings.Join(roles, ",") != strings.Join(want, ",") { + t.Errorf("roles = %v, want %v", roles, want) + } +} + +// TestFilterPersistSnapshot_NoHead verifies the filter also keeps digests +// when the session has no leading system message. +func TestFilterPersistSnapshot_NoHead(t *testing.T) { + snapshot := []llm.Message{ + {Role: "system", Content: "[Compacted earlier context: digest]"}, + {Role: "user", Content: "hi"}, + } + got := filterPersistSnapshot(nil, snapshot) + if len(got) != 2 || got[0].Role != "system" { + t.Fatalf("digest should survive with empty head, got %+v", got) + } +} diff --git a/docs/CLI.md b/docs/CLI.md index 393084d3..6743671c 100644 --- a/docs/CLI.md +++ b/docs/CLI.md @@ -49,7 +49,8 @@ | `--interaction-mode ` | string | `engaging` | Tool-call rendering: `engaging` (emoji narration) or `verbose` (raw tool output) | | `--no-color` | bool | false | Disable colored terminal output | | `--prompt-caching` | bool | false | Enable Anthropic/OpenAI/DeepSeek prompt caching markers | -| `--compaction` | bool | false | Enable LLM-based rolling compaction of trimmed context | +| `--compaction` | bool | `true` | Enable LLM-based rolling compaction of trimmed context. On by default | +| `--no-compaction` | bool | `false` | Disable rolling compaction (overrides config/default) | | `--no-agents` | bool | false | Skip loading AGENTS.md | | `--session` | bool | false | Save conversation as a multi-turn session | | `--learn` | bool | `true` | Enable skill learning mode (detects patterns, saves skills). On by default | diff --git a/docs/CONFIG.md b/docs/CONFIG.md index 254f0cb7..bc954922 100644 --- a/docs/CONFIG.md +++ b/docs/CONFIG.md @@ -247,11 +247,11 @@ I/O-bound tools (read_file, search_files, shell) benefit most — latency drops ## Rolling compaction (`compaction`) -When context trimming drops old conversation turns to stay within the model's context window, those turns are normally lost. With `compaction` enabled, the dropped turns are instead summarized by the model into a rolling digest message, preserving a compressed history of the session. +When context trimming drops old conversation turns to stay within the model's context window, those turns are normally lost. With `compaction` enabled (the default), the dropped turns are instead summarized by the model into a rolling digest message, preserving a compressed history of the session. | Field | Default | Env var | CLI flag | Description | |-------|---------|---------|----------|-------------| -| `compaction` | `false` | `ODEK_COMPACTION` | `--compaction` | Enable LLM-based rolling compaction of trimmed context. Each compaction costs one extra LLM call per trim. | +| `compaction` | `true` | `ODEK_COMPACTION` | `--compaction` / `--no-compaction` | Enable LLM-based rolling compaction of trimmed context. Each compaction costs one extra LLM call per trim. Set to `false` (or pass `--no-compaction`) to disable. | ## Concurrency and reverse-proxy trust @@ -900,7 +900,7 @@ odek init --force The **global template** covers the full schema: connection (`model`, `base_url`, `api_key`), execution (`max_iterations`, `max_tool_parallel`, `prompt_caching`, `compaction`, `interaction_mode`), sandboxing, `dangerous`, `tools`, `skills` (incl. `auto_save` and `curation`), `memory`, `subagent`, `mcp_servers`, `web_search`, `schedules`, `maintenance`, and `telegram`. -The **local template** contains only fields a project may legitimately set (`model`, `thinking`, iteration/parallelism limits, `prompt_caching`, `compaction`, `interaction_mode`, sandbox resource knobs, `tools.disabled`, `skills` without `dirs`, `subagent`, `mcp_servers`, `schedules`). Operator-only fields (`api_key`, `base_url`, `system`, `dangerous`, `memory`, `sessions`, `embedding`, `guard`, `maintenance`, `telegram`, `web_search`, `trusted_proxies`, `tools.enabled`, `skills.dirs`) belong in `~/.odek/config.json`. Note that project configs may only *enable* the sandbox — `"sandbox": false` is rejected, so neither template pins it locally. +The **local template** contains only fields a project may legitimately set (`model`, `thinking`, iteration/parallelism limits, `prompt_caching`, `interaction_mode`, sandbox resource knobs, `tools.disabled`, `skills` without `dirs`, `subagent`, `mcp_servers`, `schedules`). Operator-only fields (`api_key`, `base_url`, `system`, `dangerous`, `memory`, `sessions`, `embedding`, `guard`, `maintenance`, `telegram`, `web_search`, `trusted_proxies`, `tools.enabled`, `skills.dirs`) belong in `~/.odek/config.json`. Note that project configs may only *enable* the sandbox — `"sandbox": false` is rejected, so neither template pins it locally. `compaction` is likewise omitted from the local template: it defaults to on, and pinning `"compaction": false` in a fresh project config would silently disable it (add the key explicitly if you want it off). ## Quick examples diff --git a/internal/config/loader.go b/internal/config/loader.go index 383dbfa5..f020e4d3 100644 --- a/internal/config/loader.go +++ b/internal/config/loader.go @@ -72,8 +72,9 @@ type CLIFlags struct { // Config: prompt_caching, ODEK_PROMPT_CACHING, --prompt-caching. PromptCaching *bool // nil = not set - // Compaction enables LLM-based rolling compaction of trimmed context. - // Config: compaction, ODEK_COMPACTION, --compaction. + // Compaction enables LLM-based rolling compaction of trimmed context + // (default: on). Config: compaction, ODEK_COMPACTION, + // --compaction / --no-compaction. Compaction *bool // nil = not set // Sandbox-specific @@ -244,7 +245,8 @@ type FileConfig struct { // PromptCaching enables prompt caching markers for supported providers. PromptCaching *bool `json:"prompt_caching,omitempty"` - // Compaction enables LLM-based rolling compaction of trimmed context. + // Compaction enables LLM-based rolling compaction of trimmed context + // (default: on; set false to explicitly disable). Compaction *bool `json:"compaction,omitempty"` System string `json:"system,omitempty"` @@ -1703,7 +1705,8 @@ func LoadConfig(cli CLIFlags) ResolvedConfig { // MaxToolParallel: 0 = use loop engine default (4) resolved.MaxToolParallel = cfg.MaxToolParallel - // Booleans: default to false if not set + // Booleans: default to false if not set (Compaction below is the + // exception — it defaults to true). if cfg.Sandbox != nil { resolved.Sandbox = *cfg.Sandbox } @@ -1716,6 +1719,11 @@ func LoadConfig(cli CLIFlags) ResolvedConfig { if cfg.PromptCaching != nil { resolved.PromptCaching = *cfg.PromptCaching } + // Compaction defaults to ON: for long sessions, turn groups dropped by + // context trimming are summarized into a rolling digest instead of + // vanishing. An explicit false from any layer (config file, + // ODEK_COMPACTION=false, --no-compaction) disables it. + resolved.Compaction = true if cfg.Compaction != nil { resolved.Compaction = *cfg.Compaction } diff --git a/internal/config/loader_test.go b/internal/config/loader_test.go index d8d80e6d..7c67e544 100644 --- a/internal/config/loader_test.go +++ b/internal/config/loader_test.go @@ -1728,3 +1728,58 @@ func TestCLIFlags_Compaction(t *testing.T) { t.Error("Compaction should be true when CLIFlags.Compaction is set") } } + +// TestCompaction_DefaultOn verifies compaction resolves to true when no +// layer (config file, env, CLI) sets it. +func TestCompaction_DefaultOn(t *testing.T) { + t.Setenv("HOME", t.TempDir()) + t.Chdir(t.TempDir()) + + cfg := LoadConfig(CLIFlags{}) + if !cfg.Compaction { + t.Error("Compaction should default to true when no layer sets it") + } +} + +// TestCompaction_ExplicitFalseWins verifies an explicit false from any +// layer (env, CLI, config file) disables compaction despite the default-on. +func TestCompaction_ExplicitFalseWins(t *testing.T) { + t.Run("env", func(t *testing.T) { + t.Setenv("HOME", t.TempDir()) + t.Chdir(t.TempDir()) + t.Setenv("ODEK_COMPACTION", "false") + + cfg := LoadConfig(CLIFlags{}) + if cfg.Compaction { + t.Error("Compaction should be false when ODEK_COMPACTION=false") + } + }) + t.Run("cli", func(t *testing.T) { + t.Setenv("HOME", t.TempDir()) + t.Chdir(t.TempDir()) + + disabled := false + cfg := LoadConfig(CLIFlags{Compaction: &disabled}) + if cfg.Compaction { + t.Error("Compaction should be false when CLIFlags.Compaction is explicitly false") + } + }) + t.Run("file", func(t *testing.T) { + home := t.TempDir() + t.Setenv("HOME", home) + t.Chdir(t.TempDir()) + + odekDir := filepath.Join(home, ".odek") + if err := os.MkdirAll(odekDir, 0o755); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(filepath.Join(odekDir, "config.json"), []byte(`{"compaction": false}`), 0o600); err != nil { + t.Fatal(err) + } + + cfg := LoadConfig(CLIFlags{}) + if cfg.Compaction { + t.Error("Compaction should be false when config.json sets compaction=false") + } + }) +} diff --git a/internal/llm/client.go b/internal/llm/client.go index d461b264..c4b93cf3 100644 --- a/internal/llm/client.go +++ b/internal/llm/client.go @@ -4,6 +4,8 @@ package llm import ( "bytes" "context" + crand "crypto/rand" + "encoding/binary" "encoding/json" "fmt" "io" @@ -397,29 +399,63 @@ func reasoningEffortRejected(err error) bool { return strings.Contains(msg, "400") && strings.Contains(msg, `"reasoning_effort"`) } +// retrySleep waits d before the next retry attempt, returning early with +// ctx.Err() on cancellation. A package var so tests can stub out the wait. +var retrySleep = func(ctx context.Context, d time.Duration) error { + select { + case <-ctx.Done(): + return ctx.Err() + case <-time.After(d): + return nil + } +} + +// maxRetryBackoff caps the exponential backoff between attempts. +const maxRetryBackoff = 30 * time.Second + +// jitterBackoff scales d by a random factor in [0.75, 1.25) to decorrelate +// retries across concurrent clients (crypto/rand, like the rest of the +// codebase). Falls back to d if the entropy source fails. +func jitterBackoff(d time.Duration) time.Duration { + var b [8]byte + if _, err := crand.Read(b[:]); err != nil { + return d + } + f := float64(binary.LittleEndian.Uint64(b[:])) / (1 << 64) // [0, 1) + return time.Duration(float64(d) * (0.75 + 0.5*f)) +} + // postChatWithRetry POSTs reqBytes to /chat/completions and returns the raw 200 -// response body, retrying transient network errors and retryable HTTP statuses -// (429, 502, 503, 504) with exponential backoff. Shared by every chat call so +// response body, retrying transient network errors, retryable HTTP statuses +// (408, 429, 500, 502, 503, 504, 529), and malformed completions (unparseable +// JSON or zero choices — often a transient gateway/proxy artifact during an +// incident) with jittered exponential backoff. Shared by every chat call so // the main loop and the lightweight secondary calls (SimpleCall) get identical // resilience. Respects ctx cancellation during the backoff sleep. func (c *Client) postChatWithRetry(ctx context.Context, reqBytes []byte) ([]byte, error) { url := c.BaseURL + "/chat/completions" - const maxRetries = 3 + // 8 attempts total; worst-case backoff sum ≈ 91s before jitter + // (1+2+4+8+16+30+30), ~114s with worst-case +25% jitter — covers + // minute-scale provider incidents without wedging a turn for too long. + const maxRetries = 7 var lastErr error var wait time.Duration // how long to sleep before the next attempt for attempt := 0; attempt <= maxRetries; attempt++ { if attempt > 0 { - select { - case <-ctx.Done(): - return nil, ctx.Err() - case <-time.After(wait): + if err := retrySleep(ctx, wait); err != nil { + return nil, err } } - // Default backoff for the next attempt if this one fails: 1s, 2s, 4s. - // A Retry-After header on a 429/503 overrides it below. + // Default backoff for the next attempt if this one fails: + // 1s, 2s, 4s, 8s, 16s, 30s, 30s (capped), each ±25% jitter. + // A Retry-After header on a retryable status overrides it below. wait = time.Duration(1< maxRetryBackoff { + wait = maxRetryBackoff + } + wait = jitterBackoff(wait) req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewReader(reqBytes)) if err != nil { @@ -459,8 +495,8 @@ func (c *Client) postChatWithRetry(ctx context.Context, reqBytes []byte) ([]byte if isRetryableHTTPStatus(resp.StatusCode) { // Honor the server's Retry-After (seconds or HTTP-date) when it // asks us to wait longer than our default backoff — otherwise a - // rate-limited turn burns all three retries in ~7s and fails - // even though the server told us exactly when to come back. + // rate-limited turn burns its retries in seconds and fails even + // though the server told us exactly when to come back. if ra := parseRetryAfter(retryAfter); ra > 0 { wait = ra } @@ -469,6 +505,14 @@ func (c *Client) postChatWithRetry(ctx context.Context, reqBytes []byte) ([]byte return nil, lastErr } + // A 200 with an unparseable body or zero choices is often a transient + // gateway/proxy artifact during an incident — retry it through the + // same budget instead of aborting the turn on the first bad body. + if err := validateCompletionBody(respBytes); err != nil { + lastErr = err + continue + } + return respBytes, nil } @@ -478,7 +522,7 @@ func (c *Client) postChatWithRetry(ctx context.Context, reqBytes []byte) ([]byte // maxRetryAfter caps how long we'll honor a server's Retry-After. A pathological // or hostile value (e.g. "Retry-After: 86400") must not wedge a turn for hours; // ctx cancellation can still break the wait sooner. -const maxRetryAfter = 60 * time.Second +const maxRetryAfter = 120 * time.Second // parseRetryAfter interprets an HTTP Retry-After header, which is either an // integer number of seconds or an HTTP-date. Returns 0 when absent or @@ -510,12 +554,37 @@ func parseRetryAfter(v string) time.Duration { } // isRetryableHTTPStatus returns true for HTTP status codes that indicate -// a transient error safe to retry after a backoff. +// a transient error safe to retry after a backoff: 408 (request timeout), +// 429 (rate limited), 500 (internal server error), 502/503/504 (gateway +// errors), and 529 (Anthropic "overloaded" — their most common incident +// response during capacity events). func isRetryableHTTPStatus(code int) bool { - return code == http.StatusTooManyRequests || + return code == http.StatusRequestTimeout || + code == http.StatusTooManyRequests || + code == http.StatusInternalServerError || code == http.StatusBadGateway || code == http.StatusServiceUnavailable || - code == http.StatusGatewayTimeout + code == http.StatusGatewayTimeout || + code == 529 +} + +// validateCompletionBody checks that a 200 response body looks like a chat +// completion before the retry loop accepts it. Gateways and proxies +// occasionally answer 200 with an HTML error page, truncated JSON, or a +// zero-choices body during incidents; these are retried like any other +// transient failure. Error strings match parseResponse so callers see the +// same messages after exhaustion. +func validateCompletionBody(data []byte) error { + var raw struct { + Choices []json.RawMessage `json:"choices"` + } + if err := json.Unmarshal(data, &raw); err != nil { + return fmt.Errorf("llm: parse response: %w", err) + } + if len(raw.Choices) == 0 { + return fmt.Errorf("llm: no choices in response") + } + return nil } // isRetryableNetworkError returns true for network errors that are likely diff --git a/internal/llm/client_test.go b/internal/llm/client_test.go index a59fdece..77d3f705 100644 --- a/internal/llm/client_test.go +++ b/internal/llm/client_test.go @@ -395,6 +395,7 @@ func TestClient_Call_Success(t *testing.T) { } func TestClient_Call_HTTPError(t *testing.T) { + stubRetrySleep(t) // 500 is retryable — full exhaustion without the stub takes ~90s server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusInternalServerError) w.Write([]byte(`{"error":"internal"}`)) @@ -441,6 +442,7 @@ func TestClient_Call_WithReasoningEffort(t *testing.T) { } func TestClient_Call_InvalidEndpoint(t *testing.T) { + stubRetrySleep(t) // connection refused is retryable — stub the backoff c := New("http://127.0.0.1:1", "sk-test", "model", "", 0, 0) _, err := c.Call(context.Background(), []Message{{Role: "user", Content: "hi"}}, nil, nil) if err == nil { @@ -491,8 +493,11 @@ func TestClient_Call_Unauthorized(t *testing.T) { } } -// Test Call() with invalid JSON in the response body. +// Test Call() with invalid JSON in the response body. Malformed 200 bodies +// are retried (transient gateway artifact), so the error surfaces only after +// the full retry budget is spent. func TestClient_Call_InvalidJSONResponse(t *testing.T) { + stubRetrySleep(t) server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", "application/json") w.Write([]byte(`not json`)) @@ -625,6 +630,7 @@ func TestClient_SimpleCall_HTTPError(t *testing.T) { } func TestClient_SimpleCall_EmptyResponse(t *testing.T) { + stubRetrySleep(t) // zero-choices 200 is retried — stub the backoff server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", "application/json") w.Write([]byte(`{"choices":[]}`)) diff --git a/internal/llm/retry_test.go b/internal/llm/retry_test.go index 933cb100..d3d17600 100644 --- a/internal/llm/retry_test.go +++ b/internal/llm/retry_test.go @@ -4,11 +4,23 @@ import ( "context" "net/http" "net/http/httptest" + "strings" "sync/atomic" "testing" "time" ) +// stubRetrySleep replaces the real backoff sleep with a no-op for the +// duration of the test so exhaustion paths (8 attempts, ~91s of real +// backoff) stay fast. Tests in this package don't run in parallel, so the +// package var swap is race-safe. +func stubRetrySleep(t *testing.T) { + t.Helper() + orig := retrySleep + retrySleep = func(context.Context, time.Duration) error { return nil } + t.Cleanup(func() { retrySleep = orig }) +} + func TestParseRetryAfter(t *testing.T) { if d := parseRetryAfter("2"); d != 2*time.Second { t.Errorf("parseRetryAfter(\"2\") = %v, want 2s", d) @@ -176,6 +188,7 @@ func TestClient_Call_NoRetryOn400(t *testing.T) { } func TestClient_Call_RetryExhausted(t *testing.T) { + stubRetrySleep(t) var callCount atomic.Int32 ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { callCount.Add(1) @@ -193,13 +206,13 @@ func TestClient_Call_RetryExhausted(t *testing.T) { if err == nil { t.Fatal("expected error after exhausting retries, got nil") } - errStr := err.Error() - if errStr == "" || errStr == "expected error after exhausting retries, got nil" { - t.Error("expected non-empty error message") + // The exhaustion error must name the attempt count. + if !strings.Contains(err.Error(), "retry exhausted (8 attempts)") { + t.Errorf("error = %q, want it to mention %q", err.Error(), "retry exhausted (8 attempts)") } - // Should have tried: initial + 3 retries = 4 total - if callCount.Load() != 4 { - t.Errorf("call count = %d, want 4 (1 initial + 3 retries)", callCount.Load()) + // Should have tried: initial + 7 retries = 8 total + if callCount.Load() != 8 { + t.Errorf("call count = %d, want 8 (1 initial + 7 retries)", callCount.Load()) } } @@ -230,3 +243,165 @@ func TestClient_Call_RetryOnNetworkError(t *testing.T) { t.Errorf("content = %q, want %q", result.Content, "recovered") } } + +// TestClient_Call_RetryOn529ThenSuccess verifies Anthropic's 529 Overloaded +// response — the most common signal during capacity incidents — is retried +// and the call ultimately succeeds. +func TestClient_Call_RetryOn529ThenSuccess(t *testing.T) { + stubRetrySleep(t) + var callCount atomic.Int32 + ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if int(callCount.Add(1)) <= 2 { + w.WriteHeader(529) + w.Write([]byte(`{"type":"error","error":{"type":"overloaded_error","message":"Overloaded"}}`)) + return + } + w.Header().Set("Content-Type", "application/json") + w.Write([]byte(`{"choices":[{"message":{"content":"ok"}}]}`)) + })) + defer ts.Close() + + c := New(ts.URL, "key", "model", "", 0, 10*time.Second) + result, err := c.Call(context.Background(), []Message{{Role: "user", Content: "hi"}}, nil, nil) + if err != nil { + t.Fatalf("unexpected error after 529 retries: %v", err) + } + if result.Content != "ok" { + t.Errorf("content = %q, want ok", result.Content) + } + if callCount.Load() != 3 { + t.Errorf("call count = %d, want 3 (529 should be retried)", callCount.Load()) + } +} + +// TestClient_Call_RetryOn500 verifies a 500 Internal Server Error — common +// from gateways and providers mid-incident — is retried rather than aborting +// the turn. +func TestClient_Call_RetryOn500(t *testing.T) { + stubRetrySleep(t) + var callCount atomic.Int32 + ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if int(callCount.Add(1)) == 1 { + w.WriteHeader(http.StatusInternalServerError) + w.Write([]byte(`{"error":"internal"}`)) + return + } + w.Header().Set("Content-Type", "application/json") + w.Write([]byte(`{"choices":[{"message":{"content":"ok"}}]}`)) + })) + defer ts.Close() + + c := New(ts.URL, "key", "model", "", 0, 10*time.Second) + result, err := c.Call(context.Background(), []Message{{Role: "user", Content: "hi"}}, nil, nil) + if err != nil { + t.Fatalf("unexpected error after 500 retry: %v", err) + } + if result.Content != "ok" { + t.Errorf("content = %q, want ok", result.Content) + } + if callCount.Load() != 2 { + t.Errorf("call count = %d, want 2 (500 should be retried)", callCount.Load()) + } +} + +// TestClient_Call_RetryOnParseError verifies a 200 with an unparseable body +// (a transient gateway/proxy artifact, e.g. an HTML error page) is retried +// through the same budget instead of aborting the turn. +func TestClient_Call_RetryOnParseError(t *testing.T) { + stubRetrySleep(t) + var callCount atomic.Int32 + ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if int(callCount.Add(1)) == 1 { + w.Header().Set("Content-Type", "text/html") + w.Write([]byte(`502 Bad Gateway`)) + return + } + w.Header().Set("Content-Type", "application/json") + w.Write([]byte(`{"choices":[{"message":{"content":"ok"}}]}`)) + })) + defer ts.Close() + + c := New(ts.URL, "key", "model", "", 0, 10*time.Second) + result, err := c.Call(context.Background(), []Message{{Role: "user", Content: "hi"}}, nil, nil) + if err != nil { + t.Fatalf("unexpected error after parse-error retry: %v", err) + } + if result.Content != "ok" { + t.Errorf("content = %q, want ok", result.Content) + } + if callCount.Load() != 2 { + t.Errorf("call count = %d, want 2 (malformed 200 should be retried)", callCount.Load()) + } +} + +// TestClient_Call_RetryOnZeroChoices verifies a 200 with a valid JSON body +// but zero choices (another transient gateway artifact) is retried. +func TestClient_Call_RetryOnZeroChoices(t *testing.T) { + stubRetrySleep(t) + var callCount atomic.Int32 + ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + if int(callCount.Add(1)) == 1 { + w.Write([]byte(`{"choices":[]}`)) + return + } + w.Write([]byte(`{"choices":[{"message":{"content":"ok"}}]}`)) + })) + defer ts.Close() + + c := New(ts.URL, "key", "model", "", 0, 10*time.Second) + result, err := c.Call(context.Background(), []Message{{Role: "user", Content: "hi"}}, nil, nil) + if err != nil { + t.Fatalf("unexpected error after zero-choices retry: %v", err) + } + if result.Content != "ok" { + t.Errorf("content = %q, want ok", result.Content) + } + if callCount.Load() != 2 { + t.Errorf("call count = %d, want 2 (zero-choices 200 should be retried)", callCount.Load()) + } +} + +// TestClient_Call_ParseErrorExhausted verifies a persistently malformed body +// surfaces the parse error (wrapped in the attempt-count message) only after +// the full retry budget is spent. +func TestClient_Call_ParseErrorExhausted(t *testing.T) { + stubRetrySleep(t) + var callCount atomic.Int32 + ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + callCount.Add(1) + w.Header().Set("Content-Type", "application/json") + w.Write([]byte(`not json`)) + })) + defer ts.Close() + + c := New(ts.URL, "key", "model", "", 0, 10*time.Second) + _, err := c.Call(context.Background(), []Message{{Role: "user", Content: "hi"}}, nil, nil) + if err == nil { + t.Fatal("expected error after exhausting retries on malformed body, got nil") + } + if !strings.Contains(err.Error(), "retry exhausted (8 attempts)") { + t.Errorf("error = %q, want it to mention %q", err.Error(), "retry exhausted (8 attempts)") + } + if !strings.Contains(err.Error(), "parse response") { + t.Errorf("error = %q, want it to wrap the parse error", err.Error()) + } + if callCount.Load() != 8 { + t.Errorf("call count = %d, want 8", callCount.Load()) + } +} + +// TestJitterBackoffBounds verifies jitter stays within ±25% of the base and +// never returns a negative/zero duration for the smallest base. +func TestJitterBackoffBounds(t *testing.T) { + base := 16 * time.Second + for i := 0; i < 200; i++ { + d := jitterBackoff(base) + if d < base*3/4 || d >= base*5/4 { + t.Fatalf("jitterBackoff(%v) = %v, outside ±25%% bounds", base, d) + } + } + if d := jitterBackoff(time.Second); d < 750*time.Millisecond { + t.Fatalf("jitterBackoff(1s) = %v, below lower bound", d) + } +} diff --git a/internal/loop/loop.go b/internal/loop/loop.go index f669713f..45e42525 100644 --- a/internal/loop/loop.go +++ b/internal/loop/loop.go @@ -221,6 +221,17 @@ type Engine struct { // LLM keep retrying the same failing tool. maxConsecutiveToolErrors map[string]int + // lastToolFingerprint + toolRepeatStreak track consecutive identical + // successful tool calls (fingerprint = tool name + "\x00" + args). A + // model stuck polling the same call with the same arguments burns + // iterations undetected — only failures got a corrective hint — so after + // a few repeats the loop injects a stall warning (same machinery as the + // error-recovery correction) and resets the streak. Any different or + // failed call resets it. This is a hint, not enforcement: legitimate + // polling is allowed to continue. + lastToolFingerprint string + toolRepeatStreak int + // approver gates dangerous operations. When set and the LLM returns // multiple tool calls in one iteration, a single batch approval prompt // is shown before any tool executes, but ONLY for tools whose risk @@ -859,6 +870,29 @@ const compactionMaxSourceBytes = 32 * 1024 // summarizer input. const compactionSnippetBytes = 1000 +// ── Iteration-budget summary ─────────────────────────────────────────── + +// budgetSummaryMarker prefixes the final answer when the iteration budget +// was exhausted and the engine summarized partial progress instead of +// reaching a real completion. +const budgetSummaryMarker = "[Iteration budget reached — partial summary]" + +// budgetSummarySystemPrompt instructs the model to summarize partial +// progress when the iteration budget is exhausted. Kept short — this is a +// single bounded side-call, not a new turn of the loop. +const budgetSummarySystemPrompt = "You are a progress summarizer. The agent ran out of its " + + "iteration budget before completing the task. Based on the conversation below, summarize " + + "in a few short paragraphs: what was accomplished, the current state, and what remains " + + "to be done. Output only the summary." + +// budgetSummaryMaxMessages caps how many recent messages are rendered into +// the budget-summarizer input so the side-call stays cheap. +const budgetSummaryMaxMessages = 30 + +// budgetSummarySnippetBytes caps each message excerpt in the +// budget-summarizer input. +const budgetSummarySnippetBytes = 2000 + // refreshDigest summarizes newly dropped turn groups and inserts (or updates) // the rolling compaction digest system message. The digest is derived from // potentially untrusted tool output, so its body is wrapped with the @@ -946,6 +980,60 @@ func (e *Engine) summarizeDropped(ctx context.Context, dropped []llm.Message) st return strings.TrimSpace(res.Content) } +// summarizeProgress renders the tail of the conversation into a bounded +// summarizer input and makes one final tool-less LLM call (30s bound, same +// pattern as the compaction side-call) asking for a progress summary. +// Returns an empty string on any failure — including a non-compliant +// response that still requests tool calls — so the caller can fall back to +// the plain budget-exhaustion error. +func (e *Engine) summarizeProgress(ctx context.Context, messages []llm.Message) string { + if e.client == nil { + return "" + } + tail := messages + if len(tail) > budgetSummaryMaxMessages { + tail = tail[len(tail)-budgetSummaryMaxMessages:] + } + var b strings.Builder + for _, m := range tail { + content := m.Content + if m.Role == "assistant" && len(m.ToolCalls) > 0 { + names := make([]string, 0, len(m.ToolCalls)) + for _, tc := range m.ToolCalls { + names = append(names, tc.Function.Name) + } + content = strings.TrimSpace(content + " [called tools: " + strings.Join(names, ", ") + "]") + } + if len(content) > budgetSummarySnippetBytes { + content = content[:budgetSummarySnippetBytes] + "…" + } + if content == "" { + continue + } + fmt.Fprintf(&b, "%s: %s\n", m.Role, content) + } + if b.Len() == 0 { + return "" + } + + callCtx, cancel := context.WithTimeout(ctx, 30*time.Second) + defer cancel() + res, err := e.client.Call(callCtx, []llm.Message{ + {Role: "system", Content: budgetSummarySystemPrompt}, + {Role: "user", Content: b.String()}, + }, nil, nil) + if err != nil || res == nil { + return "" + } + // The summary call passes no tools; a response that still requests tool + // calls is not a summary (its content is pre-tool chatter), so treat it + // as a failure and keep the original error path. + if len(res.ToolCalls) > 0 { + return "" + } + return strings.TrimSpace(res.Content) +} + // ── Loop ────────────────────────────────────────────────────────────── // Run executes the loop for a given task and returns the final response. @@ -1006,6 +1094,9 @@ func (e *Engine) runLoop(ctx context.Context, messages []llm.Message) (string, [ startTime := time.Now() // Reset per-session tool error tracking e.maxConsecutiveToolErrors = make(map[string]int) + // Reset per-session repeated-call (stall) tracking + e.lastToolFingerprint = "" + e.toolRepeatStreak = 0 // Backstop: clear any batch trustAll grant when this run returns, even on // early exit or panic, so it never leaks into a later prompt that reuses @@ -1577,8 +1668,9 @@ func (e *Engine) runLoop(ctx context.Context, messages []llm.Message) (string, [ // system message so the LLM picks a different approach instead // of retrying the same failing tool. const ( - errThreshold = 3 // consecutive errors before intervention - errPrefixRead = "\"error\":" // JSON error indicator + errThreshold = 3 // consecutive errors before intervention + errPrefixRead = "\"error\":" // JSON error indicator + stallThreshold = 3 // consecutive identical successful calls before intervention ) var corrections []string for idx, tc := range result.ToolCalls { @@ -1589,8 +1681,41 @@ func (e *Engine) runLoop(ctx context.Context, messages []llm.Message) (string, [ if isErr { e.maxConsecutiveToolErrors[toolName]++ + // A failed call is not an identical successful call — the + // stall streak only counts successes. + e.lastToolFingerprint = "" + e.toolRepeatStreak = 0 } else { e.maxConsecutiveToolErrors[toolName] = 0 + + // ── Stall detection: repeated identical successful calls ── + // A model polling the same tool with the same arguments + // burns iterations without failing, so the error tracker + // above never fires. Track the fingerprint (name + args) of + // the most recent successful call; on enough consecutive + // repeats, inject a corrective hint and reset — a hint, not + // enforcement, since legitimate polling exists. + fp := toolName + "\x00" + tc.Function.Arguments + if fp == e.lastToolFingerprint { + e.toolRepeatStreak++ + } else { + e.lastToolFingerprint = fp + e.toolRepeatStreak = 1 + } + if e.toolRepeatStreak >= stallThreshold { + correction := fmt.Sprintf( + "⚠️ You called %q with identical arguments %d times in a row with no new information. Change approach: vary the arguments, switch to a different tool, or move on to the next step — repeating the same call will not produce a different result.", + toolName, e.toolRepeatStreak) + corrections = append(corrections, correction) + e.emitSignal(SignalEvent{ + Type: "tool_recovery", + Tool: toolName, + Detail: fmt.Sprintf("repeated identical call (%dx in a row)", e.toolRepeatStreak), + }) + // Reset streak after injecting suggestion (same + // semantics as the error-recovery counter). + e.toolRepeatStreak = 0 + } } if e.maxConsecutiveToolErrors[toolName] >= errThreshold { @@ -1658,6 +1783,49 @@ func (e *Engine) runLoop(ctx context.Context, messages []llm.Message) (string, [ } } + // Iteration budget exhausted. Rather than discarding all progress with a + // bare error, make one final tool-less LLM call asking the model to + // summarize what was accomplished, the current state, and what remains — + // then return that summary as the final answer (mirroring the normal + // completion path: render, iteration callback, assistant message + // appended, persist callback) so callers persist and display it like a + // normal completion. On summarizer failure, fall back to the error. + if summary := e.summarizeProgress(ctx, messages); summary != "" { + final := budgetSummaryMarker + "\n\n" + summary + + if e.renderer != nil && e.interactionMode != "off" { + e.renderer.FinalAnswer(final) + e.renderer.Summary( + e.TotalInputTokens, + e.TotalOutputTokens, + e.TotalCacheCreationTokens, + e.TotalCacheReadTokens, + e.TotalCachedTokens, + ) + } + if e.iterationCallback != nil { + e.iterationCallback(IterationInfo{ + Turn: e.maxIter, + MaxTurns: e.maxIter, + ToolNames: nil, + InputTokens: e.TotalInputTokens, + OutputTokens: e.TotalOutputTokens, + CacheCreationTokens: e.TotalCacheCreationTokens, + CacheReadTokens: e.TotalCacheReadTokens, + CachedTokens: e.TotalCachedTokens, + CacheReported: e.TotalCacheReported, + TotalLatency: time.Since(startTime), + HasFinalAnswer: true, + }) + } + messages = append(messages, llm.Message{ + Role: "assistant", + Content: final, + }) + e.emitMessagesPersist(messages) + return final, messages, nil + } + return "", messages, fmt.Errorf("reached max iterations (%d) without final answer", e.maxIter) } diff --git a/internal/loop/loop_test.go b/internal/loop/loop_test.go index 5d38bdc5..a1cc6759 100644 --- a/internal/loop/loop_test.go +++ b/internal/loop/loop_test.go @@ -104,12 +104,106 @@ func TestEngine_Run_ToolCallLoop(t *testing.T) { } func TestEngine_Run_MaxIterations(t *testing.T) { - // Server that always requests a tool call, never gives a final answer. + // Server that always requests a tool call during loop iterations, then + // answers the final budget-summary side-call with plain text. + callCount := 0 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + callCount++ + if callCount <= 3 { + fmt.Fprint(w, `{ + "choices":[{ + "message":{ + "content":"", + "tool_calls":[{ + "id":"call_1", + "function":{ + "name":"echo", + "arguments":"{}" + } + }] + } + }] + }`) + } else { + // Budget-summary call (no tools passed): return a text summary. + fmt.Fprint(w, `{"choices":[{"message":{"content":"Did some work. Still TODO: finish."}}]}`) + } + })) + defer server.Close() + + echoTool := &fakeTool{name: "echo", description: "echo", output: "ok"} + registry := tool.NewRegistry([]tool.Tool{echoTool}) + client := llm.New(server.URL, "sk-test", "test-model", "", 0, 0) + engine := New(client, registry, 3, "", nil, 0) + + // Budget exhaustion no longer errors: the engine summarizes partial + // progress in one final tool-less call and returns it as the answer. + result, err := engine.Run(context.Background(), "Loop forever") + if err != nil { + t.Fatalf("expected graceful budget-exhaustion summary, got error: %v", err) + } + if !strings.HasPrefix(result, "[Iteration budget reached") { + t.Errorf("result missing partial-summary marker: %q", result) + } + if !strings.Contains(result, "Did some work") { + t.Errorf("result missing summary body: %q", result) + } + if callCount != 4 { + t.Errorf("expected 3 loop calls + 1 summary call, got %d", callCount) + } +} + +func TestEngine_Run_MaxIterationsSummaryFallback(t *testing.T) { + // The budget-summary side-call fails (HTTP 500) — the engine must fall + // back to the original max-iterations error. + callCount := 0 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + callCount++ + if callCount == 1 { + fmt.Fprint(w, `{ + "choices":[{ + "message":{ + "content":"", + "tool_calls":[{ + "id":"call_1", + "function":{ + "name":"echo", + "arguments":"{}" + } + }] + } + }] + }`) + } else { + // 400 is non-retryable, so the summary call fails fast (a 500 + // would burn the full 30s summary-call budget in retries). + w.WriteHeader(http.StatusBadRequest) + } + })) + defer server.Close() + + echoTool := &fakeTool{name: "echo", description: "echo", output: "ok"} + registry := tool.NewRegistry([]tool.Tool{echoTool}) + client := llm.New(server.URL, "sk-test", "test-model", "", 0, 0) + engine := New(client, registry, 1, "", nil, 0) + + _, err := engine.Run(context.Background(), "Loop forever") + if err == nil { + t.Fatal("expected max iterations error when the summary call fails") + } + if !strings.Contains(err.Error(), "reached max iterations") { + t.Errorf("error = %v, want max-iterations error", err) + } +} + +func TestEngine_Run_MaxIterationsSummaryIgnoresToolCalls(t *testing.T) { + // A budget-summary response that still requests tool calls is not a + // summary — the engine must fall back to the original error. server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { fmt.Fprint(w, `{ "choices":[{ "message":{ - "content":"", + "content":"Running.", "tool_calls":[{ "id":"call_1", "function":{ @@ -126,11 +220,71 @@ func TestEngine_Run_MaxIterations(t *testing.T) { echoTool := &fakeTool{name: "echo", description: "echo", output: "ok"} registry := tool.NewRegistry([]tool.Tool{echoTool}) client := llm.New(server.URL, "sk-test", "test-model", "", 0, 0) - engine := New(client, registry, 3, "", nil, 0) + engine := New(client, registry, 1, "", nil, 0) _, err := engine.Run(context.Background(), "Loop forever") if err == nil { - t.Fatal("expected max iterations error") + t.Fatal("expected max iterations error when the summary call returns tool calls") + } +} + +func TestEngine_Run_MaxIterationsSummaryAppended(t *testing.T) { + // The partial summary must be appended to the message history like a + // normal final assistant message, so callers can persist it. + callCount := 0 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + callCount++ + if callCount == 1 { + fmt.Fprint(w, `{ + "choices":[{ + "message":{ + "content":"", + "tool_calls":[{ + "id":"call_1", + "function":{ + "name":"echo", + "arguments":"{}" + } + }] + } + }] + }`) + } else { + fmt.Fprint(w, `{"choices":[{"message":{"content":"Progress so far."}}]}`) + } + })) + defer server.Close() + + echoTool := &fakeTool{name: "echo", description: "echo", output: "ok"} + registry := tool.NewRegistry([]tool.Tool{echoTool}) + client := llm.New(server.URL, "sk-test", "test-model", "", 0, 0) + engine := New(client, registry, 1, "", nil, 0) + + var snapshots [][]llm.Message + engine.SetMessagesPersistCallback(func(msgs []llm.Message) { + snapshots = append(snapshots, msgs) + }) + + result, messages, err := engine.RunWithMessages(context.Background(), []llm.Message{ + {Role: "user", Content: "Loop forever"}, + }) + if err != nil { + t.Fatalf("expected graceful budget-exhaustion summary, got error: %v", err) + } + if !strings.HasPrefix(result, "[Iteration budget reached") { + t.Errorf("result missing partial-summary marker: %q", result) + } + last := messages[len(messages)-1] + if last.Role != "assistant" || last.Content != result { + t.Errorf("last message = %+v, want assistant message with the summary", last) + } + // Persist callback: tool batch + summary answer. + if len(snapshots) != 2 { + t.Fatalf("expected 2 persist callbacks, got %d", len(snapshots)) + } + if got := snapshots[1][len(snapshots[1])-1]; got.Role != "assistant" || + !strings.Contains(got.Content, "Progress so far.") { + t.Errorf("final snapshot missing summary assistant message: %+v", got) } } @@ -538,6 +692,98 @@ func (e *errorTool) Description() string { return e.description } func (e *errorTool) Schema() any { return map[string]any{"type": "object"} } func (e *errorTool) Call(args string) (string, error) { return "", fmt.Errorf("tool error") } +// TestEngine_Run_StallDetection verifies that repeating the SAME successful +// tool call with identical arguments triggers a corrective system message +// and a tool_recovery signal after 3 consecutive repeats, and that a +// different call resets the streak. +func TestEngine_Run_StallDetection(t *testing.T) { + // Iteration plan: + // 1-3: echo {"text":"a"} → streak hits 3 → correction injected, reset + // 4-5: echo {"text":"b"} → different args reset the streak (max 2) + // 6: final answer + callCount := 0 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + callCount++ + switch { + case callCount <= 3: + fmt.Fprint(w, `{ + "choices":[{ + "message":{ + "content":"", + "tool_calls":[{ + "id":"call_a", + "function":{"name":"echo","arguments":"{\"text\":\"a\"}"} + }] + } + }] + }`) + case callCount <= 5: + fmt.Fprint(w, `{ + "choices":[{ + "message":{ + "content":"", + "tool_calls":[{ + "id":"call_b", + "function":{"name":"echo","arguments":"{\"text\":\"b\"}"} + }] + } + }] + }`) + default: + fmt.Fprint(w, `{"choices":[{"message":{"content":"done"}}]}`) + } + })) + defer server.Close() + + echoTool := &fakeTool{name: "echo", description: "echo", output: "ok"} + registry := tool.NewRegistry([]tool.Tool{echoTool}) + client := llm.New(server.URL, "sk-test", "test-model", "", 0, 0) + engine := New(client, registry, 10, "", nil, 0) + + var signals []SignalEvent + engine.SetSignalHandler(func(ev SignalEvent) { signals = append(signals, ev) }) + + result, messages, err := engine.RunWithMessages(context.Background(), []llm.Message{ + {Role: "user", Content: "poll away"}, + }) + if err != nil { + t.Fatalf("RunWithMessages() error: %v", err) + } + if result != "done" { + t.Errorf("result = %q, want %q", result, "done") + } + + // Exactly one corrective system message: injected when the identical + // "a" call hit 3 repeats. The "b" calls (2 consecutive) must not + // trigger another one — a different call resets the streak. + stallCorrections := 0 + for _, m := range messages { + if m.Role == "system" && strings.Contains(m.Content, "identical arguments") { + stallCorrections++ + if !strings.Contains(m.Content, `"echo"`) { + t.Errorf("correction should name the stalled tool: %q", m.Content) + } + } + } + if stallCorrections != 1 { + t.Errorf("expected exactly 1 stall correction, got %d", stallCorrections) + } + + // A tool_recovery signal fired with Detail mentioning the repetition. + recoverySignals := 0 + for _, ev := range signals { + if ev.Type == "tool_recovery" && strings.Contains(ev.Detail, "repeated identical call") { + recoverySignals++ + if ev.Tool != "echo" { + t.Errorf("signal Tool = %q, want %q", ev.Tool, "echo") + } + } + } + if recoverySignals != 1 { + t.Errorf("expected exactly 1 stall tool_recovery signal, got %d (all: %+v)", recoverySignals, signals) + } +} + // ═════════════════════════════════════════════════════════════════════ // Context Trimming Tests // ═════════════════════════════════════════════════════════════════════ diff --git a/internal/loop/signal.go b/internal/loop/signal.go index 79f830ca..fd0c81d4 100644 --- a/internal/loop/signal.go +++ b/internal/loop/signal.go @@ -16,9 +16,12 @@ type SignalEvent struct { // post-error nuclear trim, or "margin_calibrated" // when the safety margin tightened after the provider // reported more input tokens than estimated) - // "tool_recovery" — a tool failed repeatedly and the engine injected a - // corrective hint so the model changes approach - // (Tool = failing tool, Detail = the correction) + // "tool_recovery" — a tool failed repeatedly, or the same successful + // call was repeated with identical arguments, and + // the engine injected a corrective hint so the model + // changes approach + // (Tool = failing/stalled tool, Detail = the + // correction or "repeated identical call (Nx)") // "tool_running" — a single tool call is still executing after the // heartbeat interval (Tool = tool name, Detail = // human-readable elapsed, e.g. "running for 2m0s"). diff --git a/odek.go b/odek.go index 3ecab8aa..2386ce6b 100644 --- a/odek.go +++ b/odek.go @@ -214,11 +214,14 @@ type Config struct { // production surfaces). UntrustedWrapper func(source, content string) string - // Compaction enables LLM-based rolling compaction (default: false). When - // enabled, conversation turn groups dropped by context trimming are + // Compaction enables LLM-based rolling compaction. When enabled, + // conversation turn groups dropped by context trimming are // summarized by the model into a rolling digest system message instead of // vanishing entirely, so long sessions retain a compressed memory of // earlier work. Each compaction costs one extra LLM call per trim. + // The CLI resolves it to ON by default (an explicit compaction=false, + // ODEK_COMPACTION=false, or --no-compaction disables it); library + // users of New must opt in explicitly here. Compaction bool }