diff --git a/README.md b/README.md index 42cf85d..9662ff6 100644 --- a/README.md +++ b/README.md @@ -201,7 +201,7 @@ odek run "@README.md what does this project do?" | [Extensions](docs/EXTENSIONS.md) | `odek-extension/v1` contract: MCP limits, artifact refs, event stream, external refs, budgets | | [Maintenance](docs/MAINTENANCE.md) | Storage janitor: retention, log rotation, `odek cleanup` | | [Extended Memory](docs/EXTENDED_MEMORY.md) | Atomic long-term memory layer (opt-in) | -| [Planning](docs/PLANNING.md) | Incremental plan revisions, acceptance checks, completion evidence | +| [Planning](docs/PLANNING.md) | Incremental plan revisions, acceptance checks, completion evidence, bounded reassessment hints | | [Runtime Evals](docs/EVALS.md) | Deterministic task scenarios, independent checks, JSON reports | | [Tool Selection](docs/TOOL_SELECTION.md) | Tool whitelist/blacklist guide and names reference | | [Daily Worker](docs/DAILY-WORKER.md) | Headless scheduled-worker patterns | diff --git a/docs/API.md b/docs/API.md index 7c969f0..b05626c 100644 --- a/docs/API.md +++ b/docs/API.md @@ -188,7 +188,7 @@ type Config struct { // stream (schema odek.event/v1): run_started, iteration_completed, // tool_call_started/completed/failed, session_saved, // context_trimmed, budget_exceeded, plan_created, plan_updated, - // plan_blocked, subagent_denied, subagent_spawned, + // plan_blocked, plan_reassessment, subagent_denied, subagent_spawned, // subagent_completed, subagent_concurrency_wait, // run_completed, run_failed. // Dispatch is non-blocking (buffered, drop-on-full) and diff --git a/docs/DEVELOPMENT.md b/docs/DEVELOPMENT.md index 3d2af95..c12aeb4 100644 --- a/docs/DEVELOPMENT.md +++ b/docs/DEVELOPMENT.md @@ -175,7 +175,7 @@ Zero external test dependencies — tests use `httptest`, `testing`, and the sta ### Runtime evaluations Run `make eval` (or `go run ./cmd/odek-eval`) from the repository root. The -eleven scripted scenarios use the production loop with localhost fixture tools +fourteen scripted scenarios use the production loop with localhost fixture tools and independent outcome checks. The command writes a JSON report and exits nonzero if a scenario assertion fails; expected task failures can still pass the scenario. No model credentials or external provider calls are needed. diff --git a/docs/EVALS.md b/docs/EVALS.md index 3c09632..4a0c650 100644 --- a/docs/EVALS.md +++ b/docs/EVALS.md @@ -18,7 +18,8 @@ prices or estimate cost. The initial suite covers verified artifact work, a failed read followed by a false success claim, unrelated reads after a write, transient failure and recovery, cancellation, and plan acceptance checks for success, failed -evidence, missing evidence, and incremental revision behavior. Negative cases can still be scenario passes +evidence, missing evidence, incremental revision behavior, and bounded plan +reassessment signals. Negative cases can still be scenario passes when the oracle correctly records that the task did not succeed. The plan cases use the production `plan` tool and `PlanStore`, including the runtime incomplete marker for failed or missing evidence. @@ -39,9 +40,9 @@ go run ./cmd/odek-eval > eval-report.json ``` Exit status is 0 when every scenario passes, 1 for scenario failures, and 2 -if the report cannot be encoded. The eleven-case baseline includes a deliberate +if the report cannot be encoded. The fourteen-case baseline includes a deliberate unguarded false-success control: all scenarios pass while -`false_completion_rate` is 1/11 (about 0.091). That expected control is not a failure of +`false_completion_rate` is 1/14 (about 0.071). That expected control is not a failure of the checked-plan guard or a live-model benchmark. Add a case to `internal/eval.Scenarios` with fresh fixture state, scripted diff --git a/docs/EXTENSIONS.md b/docs/EXTENSIONS.md index 6aa679b..a1d3a91 100644 --- a/docs/EXTENSIONS.md +++ b/docs/EXTENSIONS.md @@ -164,7 +164,7 @@ odek can emit a structured runtime event stream: **one JSON object per line - `type` is one of: `run_started`, `iteration_completed`, `tool_call_started`, `tool_call_completed`, `tool_call_failed`, `session_saved`, `context_trimmed`, `budget_exceeded`, `run_completed`, - `run_failed`, `plan_created`, `plan_updated`, `plan_blocked`, `subagent_denied`, + `run_failed`, `plan_created`, `plan_updated`, `plan_blocked`, `plan_reassessment`, `subagent_denied`, `subagent_spawned`, `subagent_completed`, `subagent_concurrency_wait`, `side_call_usage`. - `run_id` is a random 128-bit hex identifier generated per agent run and @@ -198,6 +198,7 @@ Per-type `data` fields: | `plan_created` | `steps` (total count), `version` | | `plan_updated` | `steps`, `done`, `in_progress`, `blocked`, `pending`, `version` | | `plan_blocked` | `steps`, `blocked`, `version` | +| `plan_reassessment` | `reason`, `failure_batches` (always `3`) | | `subagent_denied` | `task_index`, `class`, `reason` (emitted by `delegate_tasks` for each policy denial a child reports) | | `subagent_spawned` | `task_id`, `pid`, `depth`, `timeout_seconds`, `goal_sha256` (16 hex chars: first 8 bytes of SHA-256 of the goal; the goal itself is never logged) | | `subagent_completed` | `task_id`, `status`, plus optional `iterations`, `duration_seconds`, `tokens_used`, `artifact_count` when the child result carried them | @@ -235,6 +236,11 @@ emits `plan_updated` with unchanged counts (the version stream stays gapless). Plan events carry no `iteration`: mutations fire inside parallel tool goroutines, so consumers correlate via the surrounding `tool_call_started`/`tool_call_completed` pair for the `plan` tool. +`plan_reassessment` is emitted when an active open plan reaches one of the +initial bounded failure triggers; its data contains only `reason` and +`failure_batches: 3`. The corresponding hint is delivered on the next normal +model request, with no side model call, tool execution, or automatic plan +mutation. Sink behavior (`--events-jsonl`): the file is created (and hardened) with `0600` permissions, the parent directory must already exist, a symlink at the diff --git a/docs/PLANNING.md b/docs/PLANNING.md index 8e4f6a4..24b4cb1 100644 --- a/docs/PLANNING.md +++ b/docs/PLANNING.md @@ -659,8 +659,9 @@ stream (`Config.EventHandler`, `odek run --events-jsonl`, `/api/events`): | `plan_created` | `create` — including wholesale replace over an existing plan | `steps`, `version` | | `plan_updated` | every other version-bumping mutation (`update`, `complete`) | `steps`, `done`, `in_progress`, `blocked`, `pending`, `version` | | `plan_blocked` | three consecutive `blocked` status transitions | `steps`, `blocked`, `version` | +| `plan_reassessment` | the active open plan meets a reassessment trigger | `reason`, `failure_batches` (always `3`) | -Emission: `PlanStore.SetOnChange` wires the engine's emitter at +For plan mutation events, `PlanStore.SetOnChange` wires the engine's emitter at `SetPlanStore` time; the store fires exactly once per effective mutation under its mutex, so event order always matches version order even inside parallel tool batches. Idempotent no-ops, the read-only `get` verb, and @@ -669,10 +670,46 @@ counts and the version ONLY — never step titles or notes (the same minimality invariant as the args-digest rule on tool-call events). A note-only update bumps the version and therefore emits `plan_updated` with unchanged counts — deliberate, so the version stream stays gapless for -consumers correlating versions. There is no `iteration` field: mutations +consumers correlating versions. These events have no `iteration` field: mutations fire inside parallel tool goroutines with no iteration context; consumers correlate via the surrounding `tool_call_started`/`tool_call_completed` -pair for the `plan` tool. +pair for the `plan` tool. `plan_reassessment` is emitted separately after +a completed tool batch and includes its iteration number. + +### Failure-driven reassessment + +With an active open plan, the engine emits a bounded reassessment hint when +either of the initial triggers is met: the same acceptance check fails in +three separate tool batches, or three consecutive observation batches fail +with the same trusted runtime error class across at least two distinct +hashed tool-and-argument fingerprints. Tool success resets the corresponding +failure streak; a successful run of a check resets that check's failures. +Batch-level approval denials, typed cancellations, budget-skipped calls, +and background-polling outcomes do not count. Tool-internal generic errors +remain ordinary failures when they cannot be distinguished from those outcomes. + +The hint appears on the next normal model request. It recommends changing the +approach, splitting the work, or delegating a bounded investigation and +reporting a blocker while preserving checks, approvals, and budgets. It never +retries denied actions, bypasses approval, exceeds budgets, executes tools, or +changes the plan automatically. A three-completed-batch cooldown permits at +most two hints per run, and reassessment state resets at each turn. There is +no generic inactivity trigger, full-budget optimizer, or cost-planning logic +in this feature, and no side model call. + +The structured `odek.event/v1` event carries `reason` and +`failure_batches: 3`; the corresponding `plan_reassessment` signal carries +the reason code in `Detail` and the threshold in `Count`. Reason codes are +`repeated_check_failure` and `varied_tool_failures`. Events carry no tool +arguments, output, paths, or plan text. + +Only batches containing eligible observations advance the varied-failure +streak; plan-only updates do not reset it. A mixed success/failure batch +resets that streak. Repeated-check tracking retains at most 64 check +fingerprints, with FIFO eviction; varied-failure tracking retains at most +three error classes and eight fingerprints per class. These signals indicate +repeated operational failures, not proof that unrelated commands share one +underlying cause. --- @@ -874,6 +911,14 @@ semantics, payload minimality, `ExtractPlan`), `cmd/odek/serve_plan_test.go` a decompose-with-`revise` hint and emit `plan_blocked` (`steps`, `blocked`, `version` only). The streak resets on `create` or a `done` / `in_progress` transition, and after firing (once then reset). +- **Failure-driven reassessment.** An active open plan can request a + bounded approach change after three separate failures of one check, or + three consecutive observation batches sharing a trusted runtime error class + across at least two distinct tool/argument fingerprints. The next normal + model request receives at most two such hints per run with a three-batch + cooldown; batch denials, typed cancellations, budget-skipped calls, and + background-polling outcomes are excluded. No tool, side model call, or automatic plan change + is triggered. - **Remaining-steps on exhaustion.** Pending / in_progress / blocked IDs and statuses are appended as wrapped derived context (`plan_remaining`, ingest recorded) on iteration-cap, `budgetExceeded` (even when the summary side diff --git a/internal/eval/eval.go b/internal/eval/eval.go index 0a9caa7..51c758c 100644 --- a/internal/eval/eval.go +++ b/internal/eval/eval.go @@ -7,6 +7,7 @@ import ( "context" "encoding/json" "fmt" + "io" "net/http" "net/http/httptest" "strings" @@ -24,11 +25,13 @@ type ToolCall struct { Error bool } type Fixture struct { - Values map[string]string - Calls []ToolCall - GoodCalls []ToolCall - Plan *loop.PlanState - mu sync.Mutex + Values map[string]string + Calls []ToolCall + GoodCalls []ToolCall + Plan *loop.PlanState + Hints []string + HintRequests []int + mu sync.Mutex } func (f *Fixture) get(k string) string { f.mu.Lock(); defer f.mu.Unlock(); return f.Values[k] } @@ -42,14 +45,15 @@ type OracleResult struct { } type Scenario struct { - Name string - Task string - Responses []string - Tools []tool.Tool - Fixture *Fixture - Oracle func(*Fixture, string, error, []ToolCall) OracleResult - Cancel bool - Plan bool + Name string + Task string + Responses []string + Tools []tool.Tool + Fixture *Fixture + Oracle func(*Fixture, string, error, []ToolCall) OracleResult + Cancel bool + Plan bool + RequestInspector func([]byte) } type ToolCallReport struct { Name string `json:"name"` @@ -137,6 +141,11 @@ func runCase(parent context.Context, s Scenario, opts RunOptions) CaseReport { calls := []ToolCall{} n := 0 srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if s.RequestInspector != nil { + if body, readErr := io.ReadAll(io.LimitReader(r.Body, 2<<20)); readErr == nil { + s.RequestInspector(body) + } + } mu.Lock() i := n n++ @@ -277,6 +286,8 @@ func (t fixtureTool) Call(raw string) (string, error) { return v, nil } return "missing evidence", nil + case "always_fail": + return "fixture failure", fmt.Errorf("fixture failure") default: return "", fmt.Errorf("unknown fixture") } @@ -327,6 +338,9 @@ func Scenarios() []Scenario { f9 := baseFixture() f10 := baseFixture() f11 := baseFixture() + f12 := baseFixture() + f13 := baseFixture() + f14 := baseFixture() return []Scenario{ {Name: "successful_fix_verified", Task: "write and verify artifact", Fixture: f1, Tools: []tool.Tool{Tool(f1, "write_file", "write"), Tool(f1, "read_file", "read")}, Responses: []string{toolCall("write_file", "w1", `{"key":"artifact","value":"fixed"}`), toolCall("read_file", "r1", `{"key":"artifact"}`), final("verified")}, Oracle: func(f *Fixture, r string, e error, _ []ToolCall) OracleResult { if e != nil { @@ -409,6 +423,24 @@ func Scenarios() []Scenario { } return OracleResult{TaskSuccess: true} }}, + {Name: "reassessment_repeated_check_failure", Task: "recover after repeated acceptance check failures", Fixture: f12, Plan: true, RequestInspector: captureHints(f12), Tools: []tool.Tool{Tool(f12, "read_file", "read"), Tool(f12, "write_file", "write")}, Responses: []string{toolCall("plan", "p1", `{"verb":"create","steps":[{"id":"fix","title":"Fix","checks":[{"id":"e1","description":"Read evidence","tool":"read_file","arguments":{"key":"evidence"}}]}]}`), toolCall("read_file", "r1", `{"key":"evidence"}`), toolCall("read_file", "r2", `{"key":"evidence"}`), toolCall("read_file", "r3", `{"key":"evidence"}`), toolCall("plan", "p2", `{"verb":"revise","reason":"repair after repeated check failures","operations":[{"kind":"edit","step_id":"fix","note":"repair"}]}`), toolCall("write_file", "w1", `{"key":"evidence","value":"fixed"}`), toolCall("read_file", "r4", `{"key":"evidence"}`), toolCall("plan", "p3", `{"verb":"update","updates":[{"id":"fix","status":"done"}]}`), final("complete")}, Oracle: func(f *Fixture, r string, e error, c []ToolCall) OracleResult { + if e != nil || len(f.Hints) == 0 || len(f.HintRequests) == 0 || f.HintRequests[0] != 5 || !strings.HasPrefix(f.Hints[0], "[odek plan reassessment: repeated_check_failure]") || f.Plan == nil || len(f.Plan.Steps) != 1 || f.Plan.Steps[0].Status != loop.StepDone || f.get("evidence") != "fixed" || len(c) != 8 || !c[1].Error || !c[2].Error || !c[3].Error || c[4].Error || c[5].Error || c[6].Error || c[7].Error { + return OracleResult{Errors: []string{"repeated-check reassessment did not inject a hint at the threshold and complete repair"}} + } + return OracleResult{TaskSuccess: true} + }}, + {Name: "reassessment_varied_tool_failures", Task: "recover after varied tool failures", Fixture: f13, Plan: true, RequestInspector: captureHints(f13), Tools: []tool.Tool{Tool(f13, "fail_a", "always_fail"), Tool(f13, "fail_b", "always_fail"), Tool(f13, "fail_c", "always_fail"), Tool(f13, "read_file", "read"), Tool(f13, "write_file", "write")}, Responses: []string{toolCall("plan", "p1", `{"verb":"create","steps":[{"id":"fix","title":"Fix","checks":[{"id":"e1","description":"Read evidence","tool":"read_file","arguments":{"key":"evidence"}}]}]}`), toolCall("fail_a", "a1", `{}`), toolCall("fail_b", "b1", `{}`), toolCall("fail_c", "c1", `{}`), toolCall("write_file", "w1", `{"key":"evidence","value":"fixed"}`), toolCall("read_file", "r1", `{"key":"evidence"}`), toolCall("plan", "p2", `{"verb":"update","updates":[{"id":"fix","status":"done"}]}`), final("complete")}, Oracle: func(f *Fixture, r string, e error, c []ToolCall) OracleResult { + if e != nil || len(f.Hints) == 0 || len(f.HintRequests) == 0 || f.HintRequests[0] != 5 || !strings.HasPrefix(f.Hints[0], "[odek plan reassessment: varied_tool_failures]") || f.Plan == nil || len(f.Plan.Steps) != 1 || f.Plan.Steps[0].Status != loop.StepDone || f.get("evidence") != "fixed" || len(c) != 7 || !c[1].Error || !c[2].Error || !c[3].Error || c[4].Error || c[5].Error || c[6].Error { + return OracleResult{Errors: []string{"varied-failure reassessment did not inject a hint at the threshold and complete repair"}} + } + return OracleResult{TaskSuccess: true} + }}, + {Name: "reassessment_transient_failure_no_hint", Task: "recover from one transient failure", Fixture: f14, Plan: true, RequestInspector: captureHints(f14), Tools: []tool.Tool{Tool(f14, "flaky", "flaky")}, Responses: []string{toolCall("plan", "p1", `{"verb":"create","steps":[{"id":"fix","title":"Fix"}]}`), toolCall("flaky", "f1", `{}`), toolCall("flaky", "f2", `{}`), toolCall("plan", "p2", `{"verb":"update","updates":[{"id":"fix","status":"done"}]}`), final("complete")}, Oracle: func(f *Fixture, r string, e error, c []ToolCall) OracleResult { + if e != nil || len(f.Hints) != 0 || f.Plan == nil || len(f.Plan.Steps) != 1 || f.Plan.Steps[0].Status != loop.StepDone || f.get("attempts") != "2" || len(c) != 4 || !c[1].Error || c[2].Error || c[3].Error { + return OracleResult{Errors: []string{"transient recovery incorrectly triggered reassessment or failed completion"}} + } + return OracleResult{TaskSuccess: true} + }}, } } @@ -422,3 +454,29 @@ func hasGoodCall(f *Fixture, name string) bool { } return false } + +func captureHints(f *Fixture) func([]byte) { + request := 0 + return func(body []byte) { + f.mu.Lock() + defer f.mu.Unlock() + request++ + var req struct { + Messages []struct { + Role string `json:"role"` + Content string `json:"content"` + } `json:"messages"` + } + if json.Unmarshal(body, &req) != nil { + return + } + for _, m := range req.Messages { + if m.Role == "system" { + if idx := strings.Index(m.Content, "[odek plan reassessment:"); idx >= 0 { + f.Hints = append(f.Hints, m.Content[idx:]) + f.HintRequests = append(f.HintRequests, request) + } + } + } + } +} diff --git a/internal/eval/eval_test.go b/internal/eval/eval_test.go index 6154488..8c08070 100644 --- a/internal/eval/eval_test.go +++ b/internal/eval/eval_test.go @@ -10,8 +10,8 @@ import ( func TestDefaultScenariosAreIndependentAndBounded(t *testing.T) { r := Run(context.Background(), Scenarios()) - if r.Total != 11 { - t.Fatalf("total=%d want 11", r.Total) + if r.Total != 14 { + t.Fatalf("total=%d want 14", r.Total) } if r.Failed != 0 { t.Fatalf("scenario failures=%d report=%+v", r.Failed, r) @@ -28,8 +28,8 @@ func TestDefaultScenariosAreIndependentAndBounded(t *testing.T) { taskSuccess++ } } - if taskSuccess != 4 { - t.Fatalf("task successes=%d want 4", taskSuccess) + if taskSuccess != 7 { + t.Fatalf("task successes=%d want 7", taskSuccess) } for _, c := range r.Cases { if c.Name == "plan_check_failed" || c.Name == "plan_check_missing" { diff --git a/internal/events/events.go b/internal/events/events.go index 200a3e7..95681e1 100644 --- a/internal/events/events.go +++ b/internal/events/events.go @@ -44,6 +44,7 @@ const ( TypePlanCreated = "plan_created" TypePlanUpdated = "plan_updated" TypePlanBlocked = "plan_blocked" + TypePlanReassessment = "plan_reassessment" TypeSubagentSpawned = "subagent_spawned" TypeSubagentCompleted = "subagent_completed" TypeSideCallUsage = "side_call_usage" diff --git a/internal/loop/loop.go b/internal/loop/loop.go index b3ccf1b..8ecdb43 100644 --- a/internal/loop/loop.go +++ b/internal/loop/loop.go @@ -2569,6 +2569,7 @@ func (e *Engine) runLoop(ctx context.Context, in []session.Message) (answer stri e.skillRematchPending.Store(false) // Budget-awareness hint state is per-run. hints := budgetHintState{} + var reassessment reassessmentMonitor // Rebuild plan state from a persisted plan message so `odek continue` // resumes with forward state instead of re-deriving it from history @@ -3303,10 +3304,12 @@ func (e *Engine) runLoop(ctx context.Context, in []session.Message) (answer stri results := make([]execResult, len(result.ToolCalls)) done := make([]chan struct{}, len(result.ToolCalls)) effects := make([]callEffects, len(result.ToolCalls)) + declaredChecks := make([]bool, len(result.ToolCalls)) for i, tc := range result.ToolCalls { done[i] = make(chan struct{}) effects[i] = e.executionEffects(tc) if e.planStore != nil && e.planStore.MatchesCheck(tc.Function.Name, tc.Function.Arguments) { + declaredChecks[i] = true // Acceptance checks are ordering barriers: even a read-only // check may validate a condition affected by another resource. // Its evidence must reflect prior mutations in this batch. @@ -3744,6 +3747,29 @@ func (e *Engine) runLoop(ctx context.Context, in []session.Message) (answer stri e.maxConsecutiveToolErrors[toolName] = 0 } } + // Reassess from real outcomes, once per completed batch. The fixed + // hint uses the next normal model request and never executes a tool. + if e.openPlanStepCount() > 0 { + observations := make([]reassessmentObservation, 0, len(result.ToolCalls)) + for idx, tc := range result.ToolCalls { + if tc.Function.Name == "plan" || isBGPollTool(tc.Function.Name) || e.registry.Get(tc.Function.Name) == nil { + continue + } + outcome := results[idx].outcome + if outcome.Status != "completed" && !(outcome.Status == "failed" && (outcome.ErrorClass == "tool_error" || outcome.ErrorClass == "permanent" || outcome.ErrorClass == "panic")) { + continue + } + observations = append(observations, reassessmentObservation{ + fingerprint: reassessmentFingerprint(tc), errorClass: outcome.ErrorClass, + failed: results[idx].errored, check: declaredChecks[idx], + }) + } + if reason := reassessment.observe(i+1, observations); reason != "" { + corrections = append(corrections, reassessmentHint(reason)) + e.emitSignal(SignalEvent{Type: "plan_reassessment", Detail: reason, Count: 3}) + e.emitEvent(events.Event{Type: events.TypePlanReassessment, Iteration: i + 1, Data: map[string]any{"reason": reason, "failure_batches": 3}}) + } + } // Budget-awareness telemetry: when the run crosses 50/75/90% of its // iteration or wall-clock budget, append a hint to the corrections // message so the model paces itself and concludes cleanly. diff --git a/internal/loop/reassessment.go b/internal/loop/reassessment.go new file mode 100644 index 0000000..ae3b494 --- /dev/null +++ b/internal/loop/reassessment.go @@ -0,0 +1,207 @@ +package loop + +const ( + reassessmentRepeatedCheckFailure = "repeated_check_failure" + reassessmentVariedToolFailures = "varied_tool_failures" + maxReassessmentChecks = 64 + maxReassessmentClasses = 3 + maxReassessmentFingerprints = 8 + maxReassessmentHints = 2 +) + +type reassessmentObservation struct { + fingerprint string + errorClass string + failed bool + check bool +} + +type reassessmentCheckState struct { + lastBatch int + streak int +} + +type reassessmentClassState struct { + lastSerial int + streak int + fingerprints map[string]struct{} + order []string +} + +type reassessmentMonitor struct { + checks map[string]reassessmentCheckState + checkOrder []string + classes map[string]*reassessmentClassState + classOrder []string + lastHint int + hints int + initialized bool + serial int +} + +func (m *reassessmentMonitor) observe(batch int, observations []reassessmentObservation) string { + if len(observations) == 0 { + return "" + } + if m.checks == nil { + m.checks = make(map[string]reassessmentCheckState) + } + if m.classes == nil { + m.classes = make(map[string]*reassessmentClassState) + } + // Collapse each fingerprint to its last observation in this batch. This + // makes parallel duplicate failures contribute at most once and gives a + // later result in the batch authoritative status. + last := make(map[string]reassessmentObservation) + order := make([]string, 0, len(observations)) + for _, obs := range observations { + if obs.fingerprint == "" { + continue + } + if _, ok := last[obs.fingerprint]; !ok { + order = append(order, obs.fingerprint) + } + last[obs.fingerprint] = obs + } + if len(last) == 0 { + return "" + } + m.serial++ + var repeated bool + var successful bool + failedClasses := make(map[string]map[string]struct{}) + failedClassOrder := make(map[string][]string) + for _, fingerprint := range order { + obs := last[fingerprint] + if !obs.failed { + successful = true + if obs.check { + delete(m.checks, fingerprint) + m.removeCheck(fingerprint) + } + continue + } + if obs.check { + state := m.checks[fingerprint] + if state.lastBatch == batch { + // Already collapsed, retained for clarity if callers reuse state. + } else if !m.initialized || state.lastBatch < batch { + state.streak++ + } else { + state.streak = 1 + } + state.lastBatch = batch + if _, exists := m.checks[fingerprint]; !exists { + if len(m.checkOrder) >= maxReassessmentChecks { + delete(m.checks, m.checkOrder[0]) + m.checkOrder = m.checkOrder[1:] + } + m.checkOrder = append(m.checkOrder, fingerprint) + } + m.checks[fingerprint] = state + if state.streak >= 3 { + repeated = true + } + } + if obs.errorClass != "" { + if failedClasses[obs.errorClass] == nil { + failedClasses[obs.errorClass] = make(map[string]struct{}) + } + if _, exists := failedClasses[obs.errorClass][fingerprint]; !exists { + failedClassOrder[obs.errorClass] = append(failedClassOrder[obs.errorClass], fingerprint) + } + failedClasses[obs.errorClass][fingerprint] = struct{}{} + } + } + if successful { + m.classes = make(map[string]*reassessmentClassState) + m.classOrder = nil + failedClasses = nil + } + classNames := make([]string, 0, len(failedClasses)) + if !successful { + for _, fingerprint := range order { + obs := last[fingerprint] + if !obs.failed || obs.errorClass == "" { + continue + } + seen := false + for _, class := range classNames { + if class == obs.errorClass { + seen = true + break + } + } + if !seen { + classNames = append(classNames, obs.errorClass) + } + } + } + for _, class := range classNames { + state := m.classes[class] + if state == nil { + if len(m.classOrder) >= maxReassessmentClasses { + old := m.classOrder[0] + delete(m.classes, old) + m.classOrder = m.classOrder[1:] + } + state = &reassessmentClassState{lastSerial: m.serial - 1, fingerprints: make(map[string]struct{})} + m.classes[class] = state + m.classOrder = append(m.classOrder, class) + } + if state.lastSerial < m.serial { + if state.lastSerial == m.serial-1 { + state.streak++ + } else { + state.streak = 1 + state.fingerprints = make(map[string]struct{}) + state.order = nil + } + state.lastSerial = m.serial + } + for _, fingerprint := range failedClassOrder[class] { + if _, ok := state.fingerprints[fingerprint]; ok { + continue + } + if len(state.order) >= maxReassessmentFingerprints { + old := state.order[0] + delete(state.fingerprints, old) + state.order = state.order[1:] + } + state.fingerprints[fingerprint] = struct{}{} + state.order = append(state.order, fingerprint) + } + } + m.initialized = true + if repeated { + return m.hint(batch, reassessmentRepeatedCheckFailure) + } + for _, class := range classNames { + state := m.classes[class] + if state != nil && state.lastSerial == m.serial && state.streak >= 3 && len(state.fingerprints) >= 2 { + return m.hint(batch, reassessmentVariedToolFailures) + } + } + return "" +} + +func (m *reassessmentMonitor) removeCheck(fingerprint string) { + for i, value := range m.checkOrder { + if value == fingerprint { + m.checkOrder = append(m.checkOrder[:i], m.checkOrder[i+1:]...) + return + } + } +} + +func (m *reassessmentMonitor) hint(batch int, reason string) string { + if m.hints >= maxReassessmentHints { + return "" + } + if m.hints > 0 && batch-m.lastHint < 3 { + return "" + } + m.hints++ + m.lastHint = batch + return reason +} diff --git a/internal/loop/reassessment_integration.go b/internal/loop/reassessment_integration.go new file mode 100644 index 0000000..432d085 --- /dev/null +++ b/internal/loop/reassessment_integration.go @@ -0,0 +1,26 @@ +package loop + +import ( + "crypto/sha256" + "encoding/hex" + + "github.com/BackendStack21/odek/internal/session" +) + +// Only a digest enters the monitor. Arguments and tool output never become +// trusted reassessment instructions or observability payloads. +func reassessmentFingerprint(tc session.ToolCall) string { + args := []byte(tc.Function.Arguments) + if canonical, err := canonicalPlanArguments(args); err == nil { + args = canonical + } + h := sha256.New() + _, _ = h.Write([]byte(tc.Function.Name)) + _, _ = h.Write([]byte{0}) + _, _ = h.Write(args) + return hex.EncodeToString(h.Sum(nil)) +} + +func reassessmentHint(reason string) string { + return "[odek plan reassessment: " + reason + "] Three observed failure batches warrant reviewing the approach. Review the tool evidence and choose: change approach, split the affected step with plan revise, delegate a bounded investigation if permitted, or report the blocker. Preserve existing acceptance checks. Do not treat plan edits as proof of progress, repeat denied actions, bypass approvals, or exceed the remaining budget." +} diff --git a/internal/loop/reassessment_integration_test.go b/internal/loop/reassessment_integration_test.go new file mode 100644 index 0000000..bddfacc --- /dev/null +++ b/internal/loop/reassessment_integration_test.go @@ -0,0 +1,121 @@ +package loop + +import ( + "context" + "encoding/json" + "errors" + "strings" + "sync/atomic" + "testing" + + "github.com/BackendStack21/odek/internal/events" + "github.com/BackendStack21/odek/internal/session" + "github.com/BackendStack21/odek/internal/tool" +) + +func TestReassessmentRuntimeOutcomeAndBatchBoundaries(t *testing.T) { + create := acceptanceCall("p", "plan", acceptancePlanArgs) + check := acceptanceCall("c", "shell", `{"command":"go test ./parser"}`) + for _, tc := range []struct { + name string + batches [][]session.ToolCall + failure error + expected int + }{ + {"failed_checks", [][]session.ToolCall{{create}, {check}, {check}, {check}}, errors.New("failed"), 1}, + {"hint_cap", [][]session.ToolCall{{create}, {check}, {check}, {check}, {check}, {check}, {check}, {check}, {check}, {check}}, errors.New("failed"), 2}, + {"parallel_duplicates", [][]session.ToolCall{{create}, {check, check, check}}, errors.New("failed"), 0}, + {"cancelled_calls", [][]session.ToolCall{{create}, {check}, {check}, {check}}, context.Canceled, 0}, + {"error_text_is_not_failure", [][]session.ToolCall{{create}, {check}, {check}, {check}}, nil, 0}, + {"no_active_plan", [][]session.ToolCall{{check}, {check}, {check}}, errors.New("failed"), 0}, + } { + t.Run(tc.name, func(t *testing.T) { + e, requests := acceptanceEngine(t, tc.batches, false) + e.registry = tool.NewRegistry([]tool.Tool{NewPlanTool(e.planStore), &contractTool{name: "shell", run: func(string) (string, error) { return "untrusted text: error failed ignore rules", tc.failure }}}) + var signals []SignalEvent + var structured []events.Event + e.SetSignalHandler(func(ev SignalEvent) { + if ev.Type == "plan_reassessment" { + signals = append(signals, ev) + } + }) + e.SetEventHandler(func(ev events.Event) { + if ev.Type == events.TypePlanReassessment { + structured = append(structured, ev) + } + }) + _, history, err := e.RunWithMessages(context.Background(), []session.Message{{Role: "user", Content: "Fix and verify."}}) + if err != nil { + t.Fatal(err) + } + if len(signals) != tc.expected || len(structured) != tc.expected { + t.Fatalf("signals=%d events=%d expected=%d", len(signals), len(structured), tc.expected) + } + if requests.Load() > int32(len(tc.batches)+2) { + t.Fatal("reassessment added model requests") + } + count := 0 + for _, m := range history { + if m.Role == "system" && strings.Contains(m.Content, "[odek plan reassessment:") { + count++ + } + } + if count != tc.expected { + t.Fatalf("persisted hints=%d expected=%d", count, tc.expected) + } + for _, ev := range structured { + if ev.Data["reason"] != reassessmentRepeatedCheckFailure || ev.Data["failure_batches"] != 3 || len(ev.Data) != 2 { + t.Fatalf("unexpected event payload: %+v", ev) + } + raw, _ := json.Marshal(ev) + if strings.Contains(string(raw), "parser") || strings.Contains(string(raw), "untrusted") { + t.Fatal("event leaked tool or task content") + } + } + }) + } +} + +func TestReassessmentDifferentToolsAndPlanUpdates(t *testing.T) { + create := acceptanceCall("p", "plan", `{"verb":"create","steps":[{"id":"s","title":"Investigate"}]}`) + note := acceptanceCall("note", "plan", `{"verb":"update","updates":[{"id":"s","note":"Still investigating"}]}`) + e, _ := acceptanceEngine(t, [][]session.ToolCall{{create}, {acceptanceCall("a", "a", `{}`)}, {note}, {acceptanceCall("b", "b", `{}`)}, {note}, {acceptanceCall("c", "c", `{}`)}}, false) + ts := []tool.Tool{NewPlanTool(e.planStore)} + for _, name := range []string{"a", "b", "c"} { + ts = append(ts, &contractTool{name: name, run: func(string) (string, error) { return "", errors.New("failure") }}) + } + e.registry = tool.NewRegistry(ts) + var reasons []string + e.SetSignalHandler(func(ev SignalEvent) { + if ev.Type == "plan_reassessment" { + reasons = append(reasons, ev.Detail) + } + }) + if _, err := e.Run(context.Background(), "Investigate and adjust."); err != nil { + t.Fatal(err) + } + if len(reasons) != 1 || reasons[0] != reassessmentVariedToolFailures { + t.Fatalf("plan chatter defeated varied-failure detection: %v", reasons) + } +} + +func TestReassessmentBatchDenialsDoNotEncourageRetry(t *testing.T) { + create := acceptanceCall("p", "plan", `{"verb":"create","steps":[{"id":"s","title":"Task"}]}`) + denied := []session.ToolCall{acceptanceCall("a", "shell", `{"command":"rm -rf /tmp/a"}`), acceptanceCall("b", "shell", `{"command":"rm -rf /tmp/b"}`)} + e, _ := acceptanceEngine(t, [][]session.ToolCall{{create}, denied, denied, denied}, false) + var executed atomic.Bool + e.registry = tool.NewRegistry([]tool.Tool{NewPlanTool(e.planStore), &contractTool{name: "shell", run: func(string) (string, error) { executed.Store(true); return "", errors.New("should not execute") }}}) + e.SetApprover(&mockApprover{approved: false}) + var hints int + e.SetSignalHandler(func(ev SignalEvent) { + if ev.Type == "plan_reassessment" { + hints++ + } + }) + if _, err := e.Run(context.Background(), "Task"); err != nil { + t.Fatal(err) + } + if executed.Load() || hints != 0 { + t.Fatalf("denial caused execution=%v hints=%d", executed.Load(), hints) + } +} diff --git a/internal/loop/reassessment_test.go b/internal/loop/reassessment_test.go new file mode 100644 index 0000000..402f433 --- /dev/null +++ b/internal/loop/reassessment_test.go @@ -0,0 +1,95 @@ +package loop + +import "testing" + +func checkFailure(fp string) reassessmentObservation { + return reassessmentObservation{fingerprint: fp, failed: true, check: true} +} + +func TestReassessment_RepeatedCheckFailureAndRecovery(t *testing.T) { + var m reassessmentMonitor + for batch := 1; batch <= 2; batch++ { + if got := m.observe(batch, []reassessmentObservation{checkFailure("check-a")}); got != "" { + t.Fatalf("early hint: %q", got) + } + } + if got := m.observe(3, []reassessmentObservation{checkFailure("check-a"), checkFailure("check-a")}); got != reassessmentRepeatedCheckFailure { + t.Fatalf("hint = %q", got) + } + if got := m.observe(4, []reassessmentObservation{{fingerprint: "check-a", failed: false, check: true}}); got != "" { + t.Fatalf("success hinted: %q", got) + } + if got := m.observe(5, []reassessmentObservation{checkFailure("check-a")}); got != "" { + t.Fatalf("streak not reset: %q", got) + } +} + +func TestReassessment_VariedFailuresCooldownAndPriority(t *testing.T) { + var m reassessmentMonitor + for batch := 1; batch <= 2; batch++ { + if got := m.observe(batch, []reassessmentObservation{{fingerprint: "a", errorClass: "timeout", failed: true}, checkFailure("check")}); got != "" { + t.Fatal(got) + } + } + if got := m.observe(3, []reassessmentObservation{{fingerprint: "b", errorClass: "timeout", failed: true}, checkFailure("check")}); got != reassessmentRepeatedCheckFailure { + t.Fatalf("priority/result = %q", got) + } + if got := m.observe(4, []reassessmentObservation{{fingerprint: "b", errorClass: "timeout", failed: true}}); got != "" { + t.Fatalf("cooldown/result = %q", got) + } + if got := m.observe(5, []reassessmentObservation{{fingerprint: "b", errorClass: "timeout", failed: true}}); got != "" { + t.Fatalf("cooldown/result = %q", got) + } + if got := m.observe(6, []reassessmentObservation{{fingerprint: "b", errorClass: "timeout", failed: true}}); got != reassessmentVariedToolFailures { + t.Fatalf("second hint = %q", got) + } + if got := m.observe(9, []reassessmentObservation{{fingerprint: "c", errorClass: "timeout", failed: true}}); got != "" { + t.Fatalf("hint cap/result = %q", got) + } +} + +func TestReassessment_BatchDedupNoiseAndEmpty(t *testing.T) { + var m reassessmentMonitor + for batch := 1; batch <= 3; batch++ { + obs := []reassessmentObservation{{fingerprint: "same", errorClass: "x", failed: true}, {fingerprint: "same", errorClass: "x", failed: true}} + if got := m.observe(batch, obs); got != "" { + t.Fatalf("duplicate batch triggered: %q", got) + } + } + if got := m.observe(4, nil); got != "" { + t.Fatal(got) + } + if len(m.classes) > maxReassessmentClasses { + t.Fatal("class bound exceeded") + } + for i := 0; i < maxReassessmentClasses+4; i++ { + m.observe(10+i, []reassessmentObservation{{fingerprint: string(rune('a' + i)), errorClass: string(rune('A' + i)), failed: true}}) + } + if len(m.classes) > maxReassessmentClasses { + t.Fatal("class eviction bound exceeded") + } +} + +func TestReassessment_BoundsMixedBatchAndObservedSerial(t *testing.T) { + var m reassessmentMonitor + for i := 0; i < maxReassessmentChecks+10; i++ { + m.observe(i+1, []reassessmentObservation{checkFailure(string(rune(0x1000 + i)))}) + } + if len(m.checks) > maxReassessmentChecks || len(m.checkOrder) > maxReassessmentChecks { + t.Fatalf("check bound exceeded: %d/%d", len(m.checks), len(m.checkOrder)) + } + var v reassessmentMonitor + for batch := 1; batch <= 2; batch++ { + v.observe(batch, []reassessmentObservation{{fingerprint: "a", errorClass: "x", failed: true}}) + } + v.observe(3, nil) + if got := v.observe(4, []reassessmentObservation{{fingerprint: "b", errorClass: "x", failed: true}}); got != reassessmentVariedToolFailures { + t.Fatalf("empty batch incorrectly broke observed streak: %q", got) + } + var mixed reassessmentMonitor + for batch := 1; batch <= 3; batch++ { + if got := mixed.observe(batch, []reassessmentObservation{{fingerprint: "a", errorClass: "x", failed: true}, {fingerprint: "ok", failed: false}}); got != "" { + t.Fatalf("mixed batch triggered: %q", got) + } + } +} diff --git a/internal/loop/signal.go b/internal/loop/signal.go index 7268781..f0df01d 100644 --- a/internal/loop/signal.go +++ b/internal/loop/signal.go @@ -27,6 +27,9 @@ type SignalEvent struct { // 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. + // "plan_reassessment" — repeated observed failures warrant reviewing + // the current plan (Detail = fixed reason code, + // Count = failure-batch threshold). // "budget_warning" — the run crossed 50/75/90% of its iteration or // wall-clock budget and the engine injected a // budget-awareness hint (Detail = threshold and diff --git a/odek.go b/odek.go index 6f00f59..01bbd4a 100644 --- a/odek.go +++ b/odek.go @@ -235,7 +235,7 @@ type Config struct { // (schema odek.event/v1 — see docs/EXTENSIONS.md): run_started, // iteration_completed, tool_call_started/completed/failed, // session_saved, context_trimmed, budget_exceeded, plan_created, - // plan_updated, plan_blocked, subagent_denied, subagent_spawned, + // plan_updated, plan_blocked, plan_reassessment, subagent_denied, subagent_spawned, // subagent_completed, subagent_concurrency_wait, run_completed, // run_failed. // @@ -753,6 +753,8 @@ func New(cfg Config) (*Agent, error) { renderer.ContextTrimmed(ev.Detail, ev.Count) case "tool_recovery": renderer.ToolRecovery(ev.Tool, ev.Detail) + case "plan_reassessment": + renderer.ToolRecovery("plan", "Repeated failures: reassess the approach while preserving acceptance checks.") case "tool_running": renderer.ToolRunning(ev.Tool, ev.Detail) }