From e1b6443092cc8506d239969831a5f461de67e364 Mon Sep 17 00:00:00 2001 From: santoshkumarradha Date: Fri, 23 Oct 2026 00:35:25 -0400 Subject: [PATCH 01/12] test: verify completed plan recovery through a real engine restart --- .../completed_plan_recovery_wire_e2e_test.go | 352 ++++++++++++++++++ 1 file changed, 352 insertions(+) create mode 100644 internal/e2e/completed_plan_recovery_wire_e2e_test.go diff --git a/internal/e2e/completed_plan_recovery_wire_e2e_test.go b/internal/e2e/completed_plan_recovery_wire_e2e_test.go new file mode 100644 index 000000000..03bc2dc6d --- /dev/null +++ b/internal/e2e/completed_plan_recovery_wire_e2e_test.go @@ -0,0 +1,352 @@ +//go:build e2e + +package e2e + +import ( + "context" + "encoding/json" + "fmt" + "net/http" + "net/http/httptest" + "os" + "os/exec" + "path/filepath" + "regexp" + "runtime" + "strconv" + "strings" + "sync" + "syscall" + "testing" + "time" + + "github.com/Agent-Field/codeaf/internal/config" +) + +func TestCompletedPlanRecoversAndAcceptsOnTheRealEngine(t *testing.T) { + runCompletedPlanRecovery(t, false) +} +func TestCompletedPlanRecoveryLiveWorkflow(t *testing.T) { runCompletedPlanRecovery(t, true) } + +// The interruption is explicit fictional test work: the actual worker persists +// plandb done then stops only its disposable engine, before landing can run. +func runCompletedPlanRecovery(t *testing.T, live bool) { + if runtime.GOOS != "linux" { + t.Skip("owned engine identity check uses Linux proc metadata") + } + if _, err := exec.LookPath("tmux"); err != nil { + t.Skip("tmux required") + } + model := "stub/completed-plan" + key := "fixture-completed-plan-key" + if live { + requireTmuxAndKey(t) + model = "deepseek/deepseek-v4.1-flash" + if e2eModel != model { + t.Fatalf("requires CODEAF_E2E_MODEL=%s", model) + } + key = liveKey(t) + } + rows := map[string]any{"memory.enabled": false} + if !live { + var pins []string + for _, role := range textRoles { + pins = append(pins, string(role)+":"+model) + } + rows[config.KeyModelRoles] = strings.Join(pins, ",") + for _, name := range []string{config.KeyChatModel, config.KeyTaskModel, config.KeyTierLowModel, config.KeyTierWorkerModel, config.KeyTierHighModel, config.KeyTierReflexModel, config.KeyModelFallbacks} { + rows[name] = model + } + } + home := recoveryShortHome(t, newHome(t, rows)) + ws := newWorkspace(t, "completed-plan", false) + var mu sync.Mutex + enginePID := 0 + assignmentCalls := 0 + accepted := false + base := "https://openrouter.ai/api/v1" + if !live { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + switch r.URL.Path { + case "/models": + _ = json.NewEncoder(w).Encode(map[string]any{"data": []map[string]any{{"id": model, "context_length": 131072, "pricing": map[string]string{"prompt": "0.000001", "completion": "0.000001"}, "supported_parameters": []string{"tools", "tool_choice"}}}}) + case "/key", "/auth/key", "/credits": + fmt.Fprint(w, `{"data":{"limit_remaining":100,"total_credits":100,"total_usage":0}}`) + case "/chat/completions": + var body struct { + Stream bool `json:"stream"` + Messages []struct { + Role string `json:"role"` + Content json.RawMessage `json:"content"` + } `json:"messages"` + } + if err := json.NewDecoder(r.Body).Decode(&body); err != nil { + http.Error(w, "fixture decode", 400) + return + } + var assignment []string + acceptRequest := false + var receipt string + for _, msg := range body.Messages { + if m := freeConsentAssignment.FindStringSubmatch(string(msg.Content)); len(m) == 3 { + assignment = m + } + if msg.Role == "user" && strings.Contains(string(msg.Content), "ACCEPT_RECOVERED_PROOF") { + acceptRequest = true + } + if msg.Role == "tool" && strings.Contains(string(msg.Content), "work commit") { + _ = json.Unmarshal(msg.Content, &receipt) + } + } + if len(assignment) == 3 { + mu.Lock() + assignmentCalls++ + pid := enginePID + mu.Unlock() + command := fmt.Sprintf("printf 'MOCK_RECOVERY_VALUE\\n' > recover.txt && printf 'grep -qx MOCK_RECOVERY_VALUE recover.txt\\n' > recover-check.txt && sh recover-check.txt && plandb done %s --agent %s --result 'MOCK_RECOVERY_VALUE verified' && kill -STOP %d; sleep 2", assignment[1], assignment[2], pid) + args, _ := json.Marshal(map[string]string{"command": command}) + terminalStopResponse(w, model, "", map[string]any{"index": 0, "id": "recovery-finish-" + assignment[2], "type": "function", "function": map[string]string{"name": "bash", "arguments": string(args)}}, body.Stream) + return + } + if acceptRequest && receipt == "" { + mu.Lock() + first := !accepted + accepted = true + mu.Unlock() + if first { + terminalStopResponse(w, model, "", map[string]any{"index": 0, "id": "accept-recovered", "type": "function", "function": map[string]string{"name": "tasks", "arguments": `{"id":"1","resolve":"accept","say":"independent saved artifact check passed"}`}}, body.Stream) + return + } + } + answer := "Acknowledged." + if receipt != "" { + answer = receipt + } + terminalStopResponse(w, model, answer, nil, body.Stream) + default: + http.NotFound(w, r) + } + })) + defer server.Close() + base = server.URL + } + env := []string{config.APIKeyEnv + "=" + key, "CODEAF_BASE_URL=" + base, "CODEAF_PROFILE_DIR=", "CODEAF_CALL_LOG_BODIES=1"} + stopRecoveryEngineOnCleanup(t, env, home, ws) + rig := startWithEnv(t, env, "completed_recovery", home, ws, 180, 50, "chat", "--model", model, "--one-model", "--yolo") + rig.skipSetup(t) + intro := "This disposable fictional project tests recovery. We will preserve a finished task's files through an explicitly controlled engine restart. Acknowledge without delegating or editing files." + rig.lit(intro) + rig.keys("Enter") + if answer, _ := privateMemoryAwaitAnswer(home, intro, 90*time.Second); answer == "" { + t.Fatal("recovery introduction did not complete") + } + status := upgradeEngineCommand(t, binary(t), env, home, ws, "--status") + pid, err := strconv.Atoi(upgradePID(status)) + if err != nil { + t.Fatalf("no actual engine PID: %s", status) + } + assertCompletedEngineIdentity(t, pid, home, ws) + mu.Lock() + enginePID = pid + mu.Unlock() + t.Cleanup(func() { + if completedEnginePaused(pid) && completedEngineIdentity(pid, home, ws) { + _ = syscall.Kill(pid, syscall.SIGCONT) + } + }) + beforeHead := strings.TrimSpace(completedGit(t, ws, "rev-parse", "HEAD")) + prompt := fmt.Sprintf("/task solo disposable recovery exercise: write recover.txt containing exactly MOCK_RECOVERY_VALUE, write recover-check.txt containing exactly the runnable shell command grep -qx MOCK_RECOVERY_VALUE recover.txt, and run that check. Finish with ONE compound bash call that runs the check, marks your assigned plan task done using plandb done with your actual assigned task id and claimed agent and result 'MOCK_RECOVERY_VALUE verified', then immediately executes kill -STOP %d followed by sleep 2. This PID is only this disposable project's engine; pausing it is the explicit lifecycle test. Do not stop any other process. Do not delegate further.", pid) + rig.lit(prompt) + rig.keys("Enter") + deadline := time.Now().Add(2 * time.Minute) + for !completedEnginePaused(pid) { + if time.Now().After(deadline) { + t.Fatalf("actual worker never reached done-before-land pause: %s", rig.capture()) + } + time.Sleep(20 * time.Millisecond) + } + // The engine was stopped by the actual worker, not by a timer guessing when + // it had finished. Confirm the durable root and untouched original checkout. + plan := completedPlanPath(t, home) + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + cmd := guardedCommand(t, ctx, home, append(append(os.Environ(), env...), "CODEAF_HOME="+home), binary(t), "plandb", "show", "t-1", "--db", plan, "--json") + out, err := cmd.CombinedOutput() + cancel() + if err != nil || !strings.Contains(string(out), `"status": "done"`) { + t.Fatalf("paused engine has no durable done root: %v %s", err, out) + } + if completedGit(t, ws, "rev-parse", "HEAD") != beforeHead || fileExists(filepath.Join(ws, "recover.txt")) { + t.Fatal("fixture did not stop before original checkout landing") + } + proof := completedProofDir(t, home) + checkCompletedProof(t, proof) + beforeSpend := stopSpendSnapshot(t, home) + if beforeSpend.Cost <= 0 { + t.Fatal("interrupted live work recorded no spend") + } + mu.Lock() + callsBefore := assignmentCalls + mu.Unlock() + t.Logf("Actual engine paused after durable plan completion, original checkout unlanded; recorded cost=%g.\n%s", beforeSpend.Cost, rig.capture()) + assertCompletedEngineIdentity(t, pid, home, ws) + if err := syscall.Kill(pid, syscall.SIGKILL); err != nil { + t.Fatal(err) + } + for deadline = time.Now().Add(40 * time.Second); ; { + after := upgradeEngineCommand(t, binary(t), env, home, ws, "--status") + newPID := upgradePID(after) + if newPID != "" && newPID != strconv.Itoa(pid) { + break + } + if time.Now().After(deadline) { + t.Fatalf("actual window did not reconnect: %s", after) + } + time.Sleep(200 * time.Millisecond) + } + // Inspect the saved record before explicit acceptance; no worker is allowed + // to replay completed work or silently merge it during reconnection. + for deadline = time.Now().Add(20 * time.Second); !completedReviewRecorded(home); { + if time.Now().After(deadline) { + t.Fatal("completed root did not become actionable saved work") + } + time.Sleep(100 * time.Millisecond) + } + if completedGit(t, ws, "rev-parse", "HEAD") != beforeHead { + t.Fatal("recovery auto-landed without explicit acceptance") + } + if got := stopSpendSnapshot(t, home); got != beforeSpend { + t.Fatalf("recovery changed model receipts/spend before explicit acceptance: before=%+v after=%+v", beforeSpend, got) + } + mu.Lock() + nowCalls := assignmentCalls + mu.Unlock() + if nowCalls != callsBefore { + t.Fatal("completed worker was rerun after engine restart") + } + checkCompletedProof(t, proof) + accept := "ACCEPT_RECOVERED_PROOF: I inspected the preserved files and independently reran recover-check.txt successfully. Explicitly accept and land completed task 1 using tasks resolve accept. Report the actual landed work commit; do not rerun or redelegate the work." + rig.lit(accept) + rig.keys("Enter") + if answer, _ := privateMemoryAwaitAnswer(home, accept, 2*time.Minute); answer == "" { + t.Fatal("explicit recovery acceptance did not complete") + } + finalHead := completedGit(t, ws, "rev-parse", "HEAD") + if finalHead == beforeHead { + t.Fatal("explicit acceptance did not land a commit") + } + checkCompletedProof(t, ws) + receipt := completedCommitReceipt(t, home) + if receipt == "" { + t.Fatal("accepted task has no actual work commit receipt") + } + if err := exec.Command("git", "-C", ws, "merge-base", "--is-ancestor", receipt, "HEAD").Run(); err != nil { + t.Fatalf("receipt commit was not landed: %s", receipt) + } + models := callModels(t, home) + if live { + for _, got := range models { + if got != model { + t.Fatalf("unexpected live model %s", got) + } + } + } + t.Logf("PASS actual engine recovery live=%t: done root survived restart without replay, spend preserved, saved check rerun, explicit acceptance landed commit %s; %d model receipts (%s).\n%s", live, receipt, len(models), model, rig.capture()) +} + +func completedGit(t *testing.T, dir string, args ...string) string { + t.Helper() + out, err := exec.Command("git", append([]string{"-C", dir}, args...)...).CombinedOutput() + if err != nil { + t.Fatalf("fixture git: %v %s", err, out) + } + return strings.TrimSpace(string(out)) +} +func fileExists(path string) bool { _, err := os.Stat(path); return err == nil } +func completedEnginePaused(pid int) bool { + raw, err := os.ReadFile(fmt.Sprintf("/proc/%d/stat", pid)) + if err != nil { + return false + } + end := strings.LastIndexByte(string(raw), ')') + return end >= 0 && len(raw) > end+2 && (raw[end+2] == 'T' || raw[end+2] == 't') +} +func assertCompletedEngineIdentity(t *testing.T, pid int, home, ws string) { + t.Helper() + if !completedEngineIdentity(pid, home, ws) { + t.Fatal("refusing signal: PID is not the isolated workspace/profile engine") + } +} +func completedEngineIdentity(pid int, home, ws string) bool { + cmd, err := os.ReadFile(fmt.Sprintf("/proc/%d/cmdline", pid)) + if err != nil || !strings.Contains(string(cmd), "engine\x00") || !strings.Contains(string(cmd), "--workspace\x00"+ws+"\x00") { + return false + } + env, err := os.ReadFile(fmt.Sprintf("/proc/%d/environ", pid)) + return err == nil && strings.Contains(string(env), "CODEAF_HOME="+home+"\x00") +} + +func completedPlanPath(t *testing.T, home string) string { + t.Helper() + var found string + _ = filepath.WalkDir(home, func(path string, d os.DirEntry, err error) error { + if err == nil && !d.IsDir() && d.Name() == "plandb.db" { + found = path + } + return nil + }) + if found == "" { + t.Fatal("actual plan database missing") + } + return found +} +func completedProofDir(t *testing.T, home string) string { + t.Helper() + var found string + _ = filepath.WalkDir(home, func(path string, d os.DirEntry, err error) error { + if err == nil && !d.IsDir() && d.Name() == "recover.txt" { + found = filepath.Dir(path) + } + return nil + }) + if found == "" { + t.Fatal("saved actual task artifact missing") + } + return found +} +func checkCompletedProof(t *testing.T, dir string) { + t.Helper() + raw, err := os.ReadFile(filepath.Join(dir, "recover.txt")) + if err != nil || strings.TrimSpace(string(raw)) != "MOCK_RECOVERY_VALUE" { + t.Fatalf("preserved proof differs: %v %q", err, raw) + } + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + cmd := exec.CommandContext(ctx, "sh", "recover-check.txt") + cmd.Dir = dir + if out, err := cmd.CombinedOutput(); err != nil { + t.Fatalf("actual preserved check failed: %v %s", err, out) + } +} +func completedReviewRecorded(home string) bool { + return strings.Contains(completedRunRecords(home), "final checks and landing were not recorded") +} +func completedRunRecords(home string) string { + var out strings.Builder + _ = filepath.WalkDir(home, func(path string, d os.DirEntry, err error) error { + if err == nil && !d.IsDir() && d.Name() == "tasks.json" { + raw, _ := os.ReadFile(path) + out.Write(raw) + } + return nil + }) + return out.String() +} +func completedCommitReceipt(t *testing.T, home string) string { + t.Helper() + match := regexp.MustCompile(`work commit ([0-9a-f]{40})`).FindStringSubmatch(completedRunRecords(home)) + if len(match) > 1 { + return match[1] + } + return "" +} From 3d9d65ca4ac7ff08526a875a34696cde2f06a99e Mon Sep 17 00:00:00 2001 From: santoshkumarradha Date: Fri, 23 Oct 2026 00:40:05 -0400 Subject: [PATCH 02/12] test: accept recovered work through its displayed landing door --- .../completed_plan_recovery_wire_e2e_test.go | 29 +++++++++++++++---- 1 file changed, 24 insertions(+), 5 deletions(-) diff --git a/internal/e2e/completed_plan_recovery_wire_e2e_test.go b/internal/e2e/completed_plan_recovery_wire_e2e_test.go index 03bc2dc6d..208da42c3 100644 --- a/internal/e2e/completed_plan_recovery_wire_e2e_test.go +++ b/internal/e2e/completed_plan_recovery_wire_e2e_test.go @@ -225,12 +225,31 @@ func runCompletedPlanRecovery(t *testing.T, live bool) { t.Fatal("completed worker was rerun after engine restart") } checkCompletedProof(t, proof) - accept := "ACCEPT_RECOVERED_PROOF: I inspected the preserved files and independently reran recover-check.txt successfully. Explicitly accept and land completed task 1 using tasks resolve accept. Report the actual landed work commit; do not rerun or redelegate the work." - rig.lit(accept) - rig.keys("Enter") - if answer, _ := privateMemoryAwaitAnswer(home, accept, 2*time.Minute); answer == "" { - t.Fatal("explicit recovery acceptance did not complete") + // A restored landing card owns its displayed answer keys. Prefer that + // actual door, and submit a chat request only on a visible ordinary box. + screen := rig.capture() + for until := time.Now().Add(3 * time.Second); !strings.Contains(screen, "a accept") && time.Now().Before(until); { + time.Sleep(100 * time.Millisecond) + screen = rig.capture() } + t.Logf("Actual restored UI before explicit acceptance:\n%s", screen) + if strings.Contains(screen, "a accept") { + rig.keys("a") + } else { + if !strings.Contains(screen, "›") { + t.Fatalf("restored UI offers neither an Accept card nor a normal chat box: %s", screen) + } + accept := "ACCEPT_RECOVERED_PROOF: I inspected the preserved files and independently reran recover-check.txt successfully. Explicitly accept and land completed task 1 using tasks resolve accept. Report the actual landed work commit; do not rerun or redelegate the work." + rig.lit(accept) + rig.keys("Enter") + } + for deadline = time.Now().Add(2 * time.Minute); completedCommitReceipt(t, home) == "" || completedGit(t, ws, "rev-parse", "HEAD") == beforeHead; { + if time.Now().After(deadline) { + t.Fatalf("actual explicit acceptance did not produce a landed commit receipt: %s", rig.capture()) + } + time.Sleep(100 * time.Millisecond) + } + finalHead := completedGit(t, ws, "rev-parse", "HEAD") if finalHead == beforeHead { t.Fatal("explicit acceptance did not land a commit") From 69ee251a7c644499abcbf8c9574bb05e4aa5469f Mon Sep 17 00:00:00 2001 From: santoshkumarradha Date: Fri, 23 Oct 2026 00:40:31 -0400 Subject: [PATCH 03/12] Recover completed plans through explicit saved-work acceptance --- internal/manual/chat/task-controls.md | 18 + .../session/completed_run_recovery_test.go | 331 ++++++++++++++++++ internal/session/pending.go | 5 + internal/session/plandb_tasks.go | 17 +- internal/session/question.go | 4 +- internal/session/task_audit.go | 3 + internal/session/task_contract.go | 2 + internal/session/task_landing_question.go | 5 + internal/session/task_run_recover.go | 8 +- internal/session/task_run_resolve.go | 215 ++++++++++++ internal/session/task_store.go | 73 ++-- internal/session/tools_tasks.go | 11 + internal/tui3/planrail.go | 2 +- internal/tui3/taskplan.go | 9 +- internal/tui3/taskplan_test.go | 15 + 15 files changed, 675 insertions(+), 43 deletions(-) create mode 100644 internal/session/completed_run_recovery_test.go create mode 100644 internal/session/task_run_resolve.go diff --git a/internal/manual/chat/task-controls.md b/internal/manual/chat/task-controls.md index 71fc8345a..f71be2c05 100644 --- a/internal/manual/chat/task-controls.md +++ b/internal/manual/chat/task-controls.md @@ -208,3 +208,21 @@ conversation and task. Rendering reads thinking settings from task updates. An older connected engine shows `Engine update needed` instead of editing controls: update that engine and reconnect. An older local host holding work keeps running until it goes quiet; rebuilding the terminal binary alone does not replace a busy host. + +## The task finished just before a restart — its plan says done, but its changes have not landed + +A completed plan whose final checks and landing were not recorded returns as `your call`. +Its saved working copy, branch, crew, and recorded spend are preserved. Restarting does +not run the completed work again or merge its changes automatically. + +Inspect the saved files and checks. Choose `accept` or `not right` on its question, or ask +the conversation to use `tasks` with its number and `resolve: "accept"` or +`resolve: "refute"`. Acceptance uses the ordinary landing and names the actual work commit. +A conflict keeps the work `your call`: repair the saved copy and accept again. Refuting +keeps its branch and records that the work was not accepted. + +A missing final check receipt is not a passing check. `resolve: "reaudit"` explains that +this recovered task has no final check receipt and asks you to inspect and accept or +refute it. A missing copy or changed branch is refused with its location; nothing is +recreated in its place. If saving a receipt fails after a merge, the error reports the +actual commit and which record could not be saved. Inspect that outcome before retrying. diff --git a/internal/session/completed_run_recovery_test.go b/internal/session/completed_run_recovery_test.go new file mode 100644 index 000000000..850de30a2 --- /dev/null +++ b/internal/session/completed_run_recovery_test.go @@ -0,0 +1,331 @@ +package session + +import ( + "context" + "encoding/json" + "errors" + "os" + "path/filepath" + "strconv" + "strings" + "sync" + "testing" + "time" + + "github.com/Agent-Field/codeaf/internal/plandb" +) + +func completedRecoveryFixture(t *testing.T) (*Agent, string, *TaskCopyRecord, *beltRunDouble, uint64, time.Time, func() *Agent) { + t.Helper() + t.Setenv("CODEAF_TASK_BELT", "bash") + engine := newBeltRunDouble("must not run") + registerBeltRunEngine(t, engine) + place, repo := t.TempDir(), newTestRepo(t) + open := func() *Agent { + a, _ := newTestAgent(t, &scriptedCompleter{}, func(c *Config) { + c.Place, c.Workspace, c.SessionFile = Place{Dir: place}, repo, filepath.Join(place, placeTranscript) + c.AskConsent = false + }) + return a + } + first := open() + g := first.graph() + id := g.reserve() + tree, err := prepareTaskTree(first.config.Place, repo, "1111cccc1111cccc", id, "mock completed counter") + if err != nil { + t.Fatal(err) + } + if err := os.WriteFile(filepath.Join(tree.dir, "counter.txt"), []byte("mock completed work\n"), 0600); err != nil { + t.Fatal(err) + } + store, err := plandb.Open(g.planPath(), "mock", strconv.FormatUint(id, 10), "mock completed counter", "write counter", g.planChat()) + if err != nil { + t.Fatal(err) + } + if err := store.CompleteRoot("mock test passed; counter ready"); err != nil { + t.Fatal(err) + } + root := store.Task(store.RootID()) + _ = store.Close() + copy := runCopyOf(tree) + first.publishRunRow(g, TaskNotice{ID: id, Title: "mock completed counter", State: TaskRunning, Copy: copy, CrewState: first.unroutedCrewRecord(), CostUSD: 0.42, StartedAt: time.Now().Add(-time.Minute), PlanTask: planStoreID(strconv.FormatUint(id, 10))}) + if err := first.Close(); err != nil { + t.Fatal(err) + } + return open(), repo, copy, engine, id, root.CompletedAt, open +} + +func TestCompletedPlanRecoveryOffersItsSavedWorkWithoutReplayingOrMerging(t *testing.T) { + second, repo, copy, engine, id, ended, open := completedRecoveryFixture(t) + nothingStarted(t, second, engine) + kept, found := runRowOf(second.graph(), id) + if !found || kept.State != TaskUnverified || kept.Copy == nil || kept.Copy.Branch != copy.Branch || kept.CostUSD != 0.42 || kept.CrewState == nil || !kept.EndedAt.Equal(ended) { + t.Fatalf("completed plan stranded or changed on restart: %+v", kept) + } + if _, err := os.Stat(filepath.Join(repo, "counter.txt")); !os.IsNotExist(err) { + t.Fatal("restart merged work without acceptance") + } + if len(second.PendingDecisions()) != 1 { + t.Fatal("recovered task has no replayable landing question") + } + page, ok := second.PlanTaskPage(planStoreID(strconv.FormatUint(id, 10))) + if !ok || page.Row.StateWord() != taskWordYourCall || page.Row.USD != 0.42 || !strings.Contains(page.Row.Note, completedRunReview) { + t.Fatalf("recovered page hides its review or spend: %+v", page.Row) + } + if err := second.Close(); err != nil { + t.Fatal(err) + } + second = open() + questions := second.OpenQuestions() + if len(questions) != 1 || questions[0].ID != id || questions[0].Kind != QuestionLanding || len(questions[0].Options) != 2 { + t.Fatalf("second restart lost actionable explicit review: %+v", questions) + } + for _, option := range questions[0].Options { + if option.Key != LandingYesKey && option.Key != LandingNoKey { + t.Fatalf("dead recovered control: %+v", option) + } + } + args, _ := json.Marshal(map[string]any{"id": strconv.FormatUint(id, 10), "resolve": "accept", "say": "I inspected the mock counter"}) + reply, bad, err := second.tasksTool().Execute(context.Background(), args) + if err != nil || bad || strings.Contains(reply, "No task") { + t.Fatalf("completed run has no working accept door: %s bad=%v err=%v", reply, bad, err) + } + data, err := os.ReadFile(filepath.Join(repo, "counter.txt")) + if err != nil || string(data) != "mock completed work\n" { + t.Fatalf("accepted work did not land: %q %v (%s)", data, err, reply) + } + landed, _ := runRowOf(second.graph(), id) + if landed.State != TaskDone || landed.CostUSD != 0.42 || landed.Copy == nil || landed.Copy.Branch != copy.Branch { + t.Fatalf("accept lost completed run identity/spend: %+v", landed) + } + nothingStarted(t, second, engine) +} + +func TestCompletedPlanRecoveryRefutesWithoutTouchingItsSavedBranch(t *testing.T) { + agent, repo, copy, engine, id, _, _ := completedRecoveryFixture(t) + before := runTreeHead(repo) + if err := agent.ResolveUnverified(id, TaskRefute, "mock work is not right"); err != nil { + t.Fatal(err) + } + kept, _ := runRowOf(agent.graph(), id) + if kept.State != TaskFailed || kept.Ending != TaskEndingRefused || kept.CostUSD != 0.42 || !kept.RecoveredCompletion || runTreeHead(repo) != before { + t.Fatalf("refute lost actual state or changed main: %+v", kept) + } + if data, err := os.ReadFile(filepath.Join(copy.Dir, "counter.txt")); err != nil || string(data) != "mock completed work\n" { + t.Fatalf("refute discarded work: %q %v", data, err) + } + if len(agent.PendingDecisions()) != 0 { + t.Fatal("refuted run kept asking") + } + nothingStarted(t, agent, engine) +} + +func TestCompletedPlanRecoveryRejectsChangedBranchAndUnavailableChecker(t *testing.T) { + agent, repo, copy, engine, id, _, _ := completedRecoveryFixture(t) + before := runTreeHead(repo) + if err := agent.ResolveUnverified(id, TaskReaudit, ""); err == nil || !strings.Contains(err.Error(), "without a final check receipt") { + t.Fatalf("recovery claimed a checker it cannot run: %v", err) + } + mustGit(t, copy.Dir, "checkout", "-b", "mock-unrelated-branch") + if err := agent.ResolveUnverified(id, TaskAccept, ""); err == nil || !strings.Contains(err.Error(), "different branch") { + t.Fatalf("accepted different saved branch: %v", err) + } + if runTreeHead(repo) != before || len(agent.PendingDecisions()) != 1 { + t.Fatal("refused accept changed work or dismissed review") + } + nothingStarted(t, agent, engine) +} + +func TestCompletedPlanRecoveryRejectsWorkerResolutionOfTheSharedParent(t *testing.T) { + owner, repo, _, engine, id, _, _ := completedRecoveryFixture(t) + before := runTreeHead(repo) + worker, _ := newTestAgent(t, &scriptedCompleter{}, func(c *Config) { c.InTask = true; c.taskID = 999; c.tasker = owner.graph(); c.Workspace = repo }) + args, _ := json.Marshal(map[string]any{"id": strconv.FormatUint(id, 10), "resolve": "accept"}) + reply, bad, err := worker.tasksTool().Execute(context.Background(), args) + if err != nil || !bad { + t.Fatalf("worker accepted parent's review: %s bad=%v err=%v", reply, bad, err) + } + kept, _ := runRowOf(owner.graph(), id) + if kept.State != TaskUnverified || runTreeHead(repo) != before { + t.Fatal("scoped worker changed parent's work") + } + nothingStarted(t, owner, engine) +} + +func TestCompletedPlanRecoverySerializesStopWithMainMergeAdmission(t *testing.T) { + for _, afterAdmission := range []bool{false, true} { + name := "stop-before-admission" + if afterAdmission { + name = "stop-after-admission" + } + t.Run(name, func(t *testing.T) { + agent, repo, _, engine, id, _, _ := completedRecoveryFixture(t) + entered, release := make(chan struct{}), make(chan struct{}) + var once sync.Once + unblock := func() { once.Do(func() { close(release) }) } + prior := completedRunHomecomingAt + completedRunHomecomingAt = func(admitted bool) { + if admitted == afterAdmission { + close(entered) + <-release + } + } + t.Cleanup(func() { unblock(); completedRunHomecomingAt = prior }) + result := make(chan error, 1) + go func() { result <- agent.ResolveUnverified(id, TaskAccept, "read mock work") }() + select { + case <-entered: + case <-time.After(3 * time.Second): + t.Fatal("acceptance did not reach merge admission") + } + if err := agent.StopWork(); err != nil { + t.Fatal(err) + } + agent.mu.Lock() + done := agent.workStopDone + agent.mu.Unlock() + select { + case <-done: + t.Fatal("Stop completed before its admitted acceptance joined") + default: + } + unblock() + select { + case err := <-result: + if afterAdmission && err != nil { + t.Fatalf("later stop lost actual merge: %v", err) + } + if !afterAdmission && !errors.Is(err, errWorkStopping) { + t.Fatalf("earlier stop did not refuse merge: %v", err) + } + case <-time.After(5 * time.Second): + t.Fatal("acceptance did not settle") + } + select { + case <-done: + case <-time.After(5 * time.Second): + t.Fatal("Stop did not join completed acceptance") + } + _, err := os.Stat(filepath.Join(repo, "counter.txt")) + kept, _ := runRowOf(agent.graph(), id) + if afterAdmission { + if err != nil || kept.State != TaskDone || !strings.Contains(kept.Report, "work commit ") { + t.Fatalf("actual landing receipt lost: %+v %v", kept, err) + } + } else if !os.IsNotExist(err) || kept.State != TaskUnverified { + t.Fatalf("Stop winning before admission still merged: %+v %v", kept, err) + } + nothingStarted(t, agent, engine) + }) + } +} + +func TestCompletedPlanRecoveryKeepsActualMergeWhenPageReceiptCannotBeSaved(t *testing.T) { + agent, repo, _, engine, id, _, open := completedRecoveryFixture(t) + prior := completedRunReceiptWriter + completedRunReceiptWriter = func(string, string, string) error { return errors.New("mock page write unavailable") } + t.Cleanup(func() { completedRunReceiptWriter = prior }) + err := agent.ResolveUnverified(id, TaskAccept, "mock work inspected") + if err == nil || !strings.Contains(err.Error(), "outcome was saved") || !strings.Contains(err.Error(), "work commit ") { + t.Fatalf("page write failure concealed actual merge: %v", err) + } + head := runTreeHead(repo) + if _, err := os.Stat(filepath.Join(repo, "counter.txt")); err != nil { + t.Fatal("accepted work did not arrive") + } + kept, _ := runRowOf(agent.graph(), id) + if kept.State != TaskDone || kept.CostUSD != 0.42 { + t.Fatalf("failed receipt discarded actual outcome: %+v", kept) + } + if err := agent.Close(); err != nil { + t.Fatal(err) + } + agent = open() + if err := agent.ResolveUnverified(id, TaskAccept, "retry"); !errors.Is(err, ErrTaskDecided) { + t.Fatalf("retry repeated a merge or lost task identity: %v", err) + } + if runTreeHead(repo) != head { + t.Fatal("receipt retry changed already-landed work") + } + nothingStarted(t, agent, engine) +} + +func TestCompletedPlanRecoveryReportsFailedCheckpointAfterActualMerge(t *testing.T) { + agent, repo, _, engine, id, _, _ := completedRecoveryFixture(t) + g := agent.graph() + g.mu.Lock() + store := g.store + g.mu.Unlock() + store.mu.Lock() + original := store.path + store.path = t.TempDir() + store.mu.Unlock() + t.Cleanup(func() { store.mu.Lock(); store.path = original; store.mu.Unlock() }) + err := agent.ResolveUnverified(id, TaskAccept, "mock inspected") + if err == nil || !strings.Contains(err.Error(), "task checkpoint could not be saved") || strings.Contains(err.Error(), "Its outcome was saved") || !strings.Contains(err.Error(), "work commit ") { + t.Fatalf("unchecked checkpoint claimed durable success: %v", err) + } + if _, err := os.Stat(filepath.Join(repo, "counter.txt")); err != nil { + t.Fatal("actual accepted work was rolled back") + } + kept, _ := runRowOf(g, id) + if kept.State != TaskDone { + t.Fatalf("failed checkpoint falsified actual merge: %+v", kept) + } + if err := agent.ResolveUnverified(id, TaskAccept, "retry"); !errors.Is(err, ErrTaskDecided) { + t.Fatalf("retry repeated accepted merge: %v", err) + } + nothingStarted(t, agent, engine) +} + +func TestCompletedPlanRecoveryNeverTakesAnUnattendedDefault(t *testing.T) { + agent, _, _, _, id, _, _ := completedRecoveryFixture(t) + agent.config.Unattended, agent.config.AskConsent = true, true + kept, _ := runRowOf(agent.graph(), id) + if agent.landingTakesItsDefault(kept) { + t.Fatal("restart silently accepted completed work") + } + ordinary := kept + ordinary.RecoveredCompletion = false + if !agent.landingTakesItsDefault(ordinary) { + t.Fatal("recovered guard changed ordinary unattended landing policy") + } +} + +func TestCompletedPlanRecoveryKeepsConsequencesAndOrdinaryQuestionControls(t *testing.T) { + q := Question{Kind: QuestionConflict, Options: []AnswerOption{{Key: LandingYesKey, Label: "resolve it", Consequence: "mock existing files will be set aside"}, {Key: LandingNoKey, Label: "drop it"}, {Key: LandingDecideKey, Label: "tell it"}}} + recovered := completedLandingQuestion(q, TaskNotice{RecoveredCompletion: true, Merge: mergeConflicted}) + if recovered.Kind != QuestionLanding || len(recovered.Options) != 2 || recovered.Options[0].Label != taskAskCheckYes || recovered.Options[0].Consequence != q.Options[0].Consequence { + t.Fatalf("recovered controls hid a consequence or offered a dead action: %+v", recovered) + } + ordinary := completedLandingQuestion(q, TaskNotice{}) + if ordinary.Kind != QuestionConflict || len(ordinary.Options) != 3 || ordinary.Options[2].Key != LandingDecideKey { + t.Fatal("recovered repair changed ordinary question controls") + } +} + +func TestCompletedPlanRecoveryReportsUnsavedRefusalWithoutChangingItsCopy(t *testing.T) { + agent, repo, copy, engine, id, _, _ := completedRecoveryFixture(t) + before := runTreeHead(repo) + g := agent.graph() + g.mu.Lock() + store := g.store + g.mu.Unlock() + store.mu.Lock() + original := store.path + store.path = t.TempDir() + store.mu.Unlock() + t.Cleanup(func() { store.mu.Lock(); store.path = original; store.mu.Unlock() }) + err := agent.ResolveUnverified(id, TaskRefute, "mock rejected") + if err == nil || !strings.Contains(err.Error(), "task checkpoint could not be saved") || strings.Contains(err.Error(), "Its outcome was saved") { + t.Fatalf("unchecked refusal claimed durable success: %v", err) + } + kept, _ := runRowOf(g, id) + if kept.State != TaskFailed || kept.Ending != TaskEndingRefused || runTreeHead(repo) != before { + t.Fatalf("unsaved refusal changed work or lost actual answer: %+v", kept) + } + if _, err := os.Stat(filepath.Join(copy.Dir, "counter.txt")); err != nil { + t.Fatal("refusal removed saved work") + } + nothingStarted(t, agent, engine) +} diff --git a/internal/session/pending.go b/internal/session/pending.go index dbdad90e8..10188b9fe 100644 --- a/internal/session/pending.go +++ b/internal/session/pending.go @@ -98,6 +98,11 @@ func (a *Agent) PendingDecisions() []PendingDecision { Depth: graph.depthLocked(node), }) } + for _, notice := range graph.runRowsLocked() { + if notice.RecoveredCompletion && notice.State == TaskUnverified { + out = append(out, PendingDecision{Notice: notice}) + } + } return out } diff --git a/internal/session/plandb_tasks.go b/internal/session/plandb_tasks.go index 1ed3db1f3..7c436dcd2 100644 --- a/internal/session/plandb_tasks.go +++ b/internal/session/plandb_tasks.go @@ -157,7 +157,7 @@ func (row PlanTaskRow) StateWord() string { return taskWordDone case string(plandb.StatusFailed), string(plandb.StatusCancelled): return taskWordIncomplete - case "paused": + case "paused", string(TaskUnverified): return taskWordYourCall } return "" @@ -459,10 +459,21 @@ func (a *Agent) markPlanInterrupted(row *PlanTaskRow) { g.mu.Lock() defer g.mu.Unlock() for _, notice := range g.runRowsLocked() { - if notice.PlanTask == row.ID && notice.State == TaskInterrupted { + if notice.PlanTask != row.ID { + continue + } + if notice.State == TaskInterrupted { row.Interrupted = true - return } + if notice.RecoveredCompletion { + row.Status = string(notice.State) + row.Ended = notice.EndedAt + if notice.CostUSD > row.USD { + row.USD = notice.CostUSD + } + row.Note = notice.Report + } + return } } diff --git a/internal/session/question.go b/internal/session/question.go index 7ac7d14b0..a3178b6af 100644 --- a/internal/session/question.go +++ b/internal/session/question.go @@ -2995,6 +2995,8 @@ func (a *Agent) landingQuestion(pending PendingDecision) Question { // own check costs: one record read per derived landing question, no cache // and so no second source of truth. Banked words are an unanswered // question and carry no fate — [Agent.said] returns them untouched. + q = completedLandingQuestion(q, notice) + if record, found := decidedAlready(a.Decisions(), q); found { stamp := landingAnsweredStamp(record) switch { @@ -3009,7 +3011,7 @@ func (a *Agent) landingQuestion(pending PendingDecision) Question { q.Reason = stamp + " · " + q.Reason } } - return a.said(kind, token, q) + return a.said(q.Kind, token, q) } // landingForm is which shape a landing asks to be drawn in, and it is decided diff --git a/internal/session/task_audit.go b/internal/session/task_audit.go index f209c615f..1bb464659 100644 --- a/internal/session/task_audit.go +++ b/internal/session/task_audit.go @@ -2582,6 +2582,9 @@ func (a *Agent) ResolveUnverified(id uint64, resolution TaskResolution, why stri func (a *Agent) resolveUnverifiedBy(id uint64, resolution TaskResolution, why string, by TaskAskOwner) error { node := a.taskNode(id) if node == nil { + if _, found, err := a.resolveCompletedRun(context.Background(), id, resolution, why, by); found { + return err + } return fmt.Errorf("no task %d in this session", id) } // The state is read here for the ERROR, and claimed again inside each of the diff --git a/internal/session/task_contract.go b/internal/session/task_contract.go index 5873ce7d6..65deb32db 100644 --- a/internal/session/task_contract.go +++ b/internal/session/task_contract.go @@ -633,6 +633,8 @@ type TaskNotice struct { // carries these facts in its own record. Nil is a row whose copy was never // recorded, which is a run that cannot be carried on. Copy *TaskCopyRecord + // RecoveredCompletion identifies a completed plan recovered before its final landing receipt. + RecoveredCompletion bool // PendingRun keeps an admitted request before it owns a working copy. PendingRun *PendingRunRecord // PlanTask is WHICH TASK OF THE PLAN STORE THIS ROW IS, and it is the one diff --git a/internal/session/task_landing_question.go b/internal/session/task_landing_question.go index fd43cd6dc..0ddfb9f84 100644 --- a/internal/session/task_landing_question.go +++ b/internal/session/task_landing_question.go @@ -234,6 +234,11 @@ func (a *Agent) retireLandingQuestion(notice TaskNotice, standing QuestionKind) // to the reading law; a conflict, a shift, a ground that moved, or a check that // did not pass misses the last, and each stays the person's (task-states law). func (a *Agent) landingTakesItsDefault(notice TaskNotice) bool { + // A restarted completed plan has no final acceptance receipt to recover. + // Only a fresh explicit answer may land that saved work. + if notice.RecoveredCompletion { + return false + } if !a.config.Unattended || !a.config.AskConsent { return false } diff --git a/internal/session/task_run_recover.go b/internal/session/task_run_recover.go index 5c71ebaf9..d26e1af2b 100644 --- a/internal/session/task_run_recover.go +++ b/internal/session/task_run_recover.go @@ -85,7 +85,13 @@ func (a *Agent) reconcileSettledRun(g *TaskGraph, kept TaskNotice, root *plandb. kept.State, kept.Report = TaskFailed, root.Error kept.EndedAt = root.CompletedAt default: - kept.Report = "the plan finished before the restart; inspect the saved working copy before landing its changes" + kept.Report = "the plan finished before the restart; " + errNoRunCopy.Error() + kept.RecoveredCompletion = true + if kept.Copy != nil { + kept.State, kept.Report = TaskUnverified, withReport(completedRunReview, root.Result) + kept.EndedAt = root.CompletedAt + kept.Elapsed = runSpan(kept.StartedAt, root.CompletedAt) + } } a.publishRunRow(g, kept) } diff --git a/internal/session/task_run_resolve.go b/internal/session/task_run_resolve.go new file mode 100644 index 000000000..f38dbcb0f --- /dev/null +++ b/internal/session/task_run_resolve.go @@ -0,0 +1,215 @@ +package session + +import ( + "context" + "fmt" + "strconv" + "strings" + + "github.com/Agent-Field/codeaf/internal/plandb" +) + +const completedRunReview = "the plan finished before the restart, but its final checks and landing were not recorded; inspect the saved working copy, then accept or refute it on the task page or ask this conversation to do so" + +// resolveCompletedRun is the existing explicit decision door for completed +// work whose process disappeared before its landing. It never starts workers +// and a completed store alone never supplies a successful check receipt. +func (a *Agent) resolveCompletedRun(ctx context.Context, id uint64, resolution TaskResolution, why string, by TaskAskOwner) (string, bool, error) { + g, found := a.completedRunOwner(id) + if !found { + return "", false, nil + } + epoch, err := a.beltAdmissionEpoch() + if err != nil { + return "", true, err + } + a.lockBeltStart() + defer a.beltStartMu.Unlock() + if err := a.checkBeltAdmission(epoch); err != nil { + return "", true, err + } + kept, found := runRowOf(g, id) + if kept.RecoveredCompletion && kept.Copy == nil { + return "", true, errNoRunCopy + } + if !found || kept.Program != "" || kept.State != TaskUnverified { + return "", true, settledAlready(id, kept.State) + } + path, root, err := a.completedRunPlan(id) + if err != nil { + return "", true, err + } + switch resolution { + case TaskReaudit: + return "", true, fmt.Errorf("task %d finished before the restart without a final check receipt; inspect its saved work, then accept or refute it", id) + case TaskAccept: + return a.acceptCompletedRun(ctx, g, kept, path, root, epoch, why, by) + case TaskRefute: + kept.State, kept.Ending = TaskFailed, TaskEndingRefused + kept.Report = withReport(refutedLine(why, by), kept.Report) + return a.saveCompletedRunOutcome(g, kept, path, root.ID) + default: + return "", true, fmt.Errorf("%q is not a resolution: say %s, %s or %s", resolution, TaskAccept, TaskReaudit, TaskRefute) + } +} + +// completedRunOwner keeps the fallback inside this conversation. A worker's +// shared task graph does not authorize it to settle its parent or siblings. +func (a *Agent) completedRunOwner(id uint64) (*TaskGraph, bool) { + if a.config.InTask || a.config.taskID != 0 { + return nil, false + } + g := a.tasker() + if g == nil { + return nil, false + } + if kept, found := runRowOf(g, id); !found || !kept.RecoveredCompletion { + return nil, false + } + return g, true +} + +// completedRunPlan reads only this conversation's exact completed root, including +// an archived store. A newer task at the live path is never adopted in its place. +func (a *Agent) completedRunPlan(id uint64) (string, *plandb.Task, error) { + stores, plan, closeStores := a.openPlanReadHandles() + defer closeStores() + want := strconv.FormatUint(id, 10) + for _, store := range stores { + if store.RootID() != want { + continue + } + root := store.Task(want) + if root == nil || root.Chat != plan.chat || root.Status != plandb.StatusDone { + break + } + return store.Path(), root, nil + } + return "", nil, fmt.Errorf("task %d has no completed plan in this conversation; its saved work has not been changed", id) +} + +func (a *Agent) acceptCompletedRun(ctx context.Context, g *TaskGraph, kept TaskNotice, path string, root *plandb.Task, epoch uint64, why string, by TaskAskOwner) (string, bool, error) { + tree, err := runCopyTree(kept.Copy, a.config.Place) + if err != nil { + return "", true, err + } + // A directory that now belongs to another branch is not the saved work. + // The ordinary landing also checks the recorded destination branch and head. + if kept.Copy.Branch != "" && currentBranch(tree.dir) != kept.Copy.Branch { + return "", true, fmt.Errorf("task %d's saved working copy now stands on a different branch; restore %s before accepting it", kept.ID, kept.Copy.Branch) + } + if err := ctx.Err(); err != nil { + return "", true, err + } + branch, changed, refused, err := LandRunTree(tree.dir, tree.checkBase, kept.Title, kept.Model) + if err != nil { + return "", true, err + } + if refused != "" && refused != runNothingToLand { + return "", true, fmt.Errorf("task %d could not land: %s", kept.ID, refused) + } + commit := runTreeHead(tree.dir) + if err := a.admitCompletedRunHomecoming(ctx, epoch); err != nil { + return "", true, err + } + merge, detail, _, _ := tree.comeHome(kept.Title, nil, a.signsGitWork()) + kept.Changed, kept.Branch, kept.Merge = changed, branch, merge + kept.Report = withReport(acceptedLine(strings.TrimSpace(why), by), withReport(root.Result, detail)) + if commit != "" { + kept.Report = withReport(kept.Report, "work commit "+commit) + } + if cameHome(merge) { + kept.State = TaskDone + } else { + kept.State = TaskUnverified + } + return a.saveCompletedRunOutcome(g, kept, path, root.ID) +} + +// saveCompletedRunOutcome checks the checkpoint write rather than relying on +// publishRunRow's best-effort logging. An irreversible merge keeps its actual +// in-memory state and commit even when either durable receipt cannot be saved. +func (a *Agent) saveCompletedRunOutcome(g *TaskGraph, kept TaskNotice, path, rootID string) (string, bool, error) { + a.publishRunRow(g, kept) + g.mu.Lock() + checkpoint := g.store + g.mu.Unlock() + checkpointErr := checkpoint.write(g) + pageErr := completedRunReceiptWriter(path, rootID, kept.Report) + receipt := fmt.Sprintf("task %d is %s\n%s", kept.ID, ProjectTask(kept.StatusFacts()).Word, kept.Report) + if checkpointErr != nil { + if pageErr != nil { + return "", true, fmt.Errorf("%s\nThe task checkpoint and plan-page receipt could not be saved: %v; %w. Inspect the actual work before retrying", receipt, checkpointErr, pageErr) + } + return "", true, fmt.Errorf("%s\nThe plan page holds the receipt, but its task checkpoint could not be saved: %w. Inspect the actual work before retrying", receipt, checkpointErr) + } + if pageErr != nil { + return "", true, fmt.Errorf("%s\nIts outcome was saved, but the plan page could not save that receipt: %w; read task %d for its actual outcome before retrying", receipt, pageErr, kept.ID) + } + return receipt, true, nil +} + +var completedRunReceiptWriter = recordCompletedRunReceipt + +// recordCompletedRunReceipt reports a failed page write after preserving the +// actual landing in the checkpoint, so a retry cannot repeat a completed merge. +func recordCompletedRunReceipt(path, id, receipt string) error { + store, err := plandb.Open(path, "", "", "", "") + if err != nil { + return err + } + defer store.Close() + _, err = store.AddNote(id, id, receipt) + return err +} + +// completedLandingQuestion offers only decisions the recovered copy can answer. +// A missing checker is not a live node to hand to an automatic merge round. +func completedLandingQuestion(q Question, notice TaskNotice) Question { + if !notice.RecoveredCompletion { + return q + } + q.Kind = QuestionLanding + var options []AnswerOption + for _, option := range q.Options { + switch option.Key { + case LandingYesKey: + option.Label = taskAskCheckYes + case LandingNoKey: + option.Label = taskAskCheckNo + default: + continue + } + options = append(options, option) + } + q.Options = options + if notice.Merge != mergeConflicted && notice.Merge != mergeAborted { + q.Reason = completedRunReview + } + return q +} + +// completedRunHomecomingAt exposes both sides of the irreversible boundary to +// deterministic cancellation regressions. It does no work in the product. +var completedRunHomecomingAt func(admitted bool) + +// admitCompletedRunHomecoming is the last cancellable boundary before the main +// merge, under the same lock StopWork uses to close admission. A successful +// check admits the existing atomic merge; a later Stop preserves its actual +// receipt and joins this acceptance through beltStartMu, held by the caller +// until both landing and checkpoint publication have finished. +func (a *Agent) admitCompletedRunHomecoming(ctx context.Context, epoch uint64) error { + if err := ctx.Err(); err != nil { + return err + } + if completedRunHomecomingAt != nil { + completedRunHomecomingAt(false) + } + if err := a.checkBeltAdmission(epoch); err != nil { + return err + } + if completedRunHomecomingAt != nil { + completedRunHomecomingAt(true) + } + return nil +} diff --git a/internal/session/task_store.go b/internal/session/task_store.go index 7b963ba84..1226311b8 100644 --- a/internal/session/task_store.go +++ b/internal/session/task_store.go @@ -715,7 +715,8 @@ type harnessOfferRecord struct { // exists, and no answer could reach one. Left out, the row comes back settled // like every other row that was moving, which is the truth about it. type runRecord struct { - CrewState *TaskCrewRecord `json:"crewState,omitempty"` + RecoveredCompletion bool `json:"recoveredCompletion,omitempty"` + CrewState *TaskCrewRecord `json:"crewState,omitempty"` ID uint64 `json:"id"` Run string `json:"run,omitempty"` @@ -1096,30 +1097,31 @@ func runRowRecord(notice TaskNotice) runRecord { state = TaskRunning } return runRecord{ - CrewState: notice.CrewState, - ID: notice.ID, - Run: notice.Run, - Node: notice.Node, - Parent: notice.Parent, - Title: notice.Title, - Kind: notice.Kind, - Program: notice.Program, - State: state, - Stopped: notice.Stopped, - Report: notice.Report, - Model: notice.Model, - CostUSD: notice.CostUSD, - ElapsedMS: notice.Elapsed.Milliseconds(), - StartedAt: notice.StartedAt, - EndedAt: notice.EndedAt, - Copy: notice.Copy, - PendingRun: notice.PendingRun, - Ending: notice.Ending, - Branch: notice.Branch, - Merge: notice.Merge, - Result: notice.Result, - Changed: append([]string(nil), notice.Changed...), - PlanTask: notice.PlanTask, + CrewState: notice.CrewState, + ID: notice.ID, + Run: notice.Run, + Node: notice.Node, + Parent: notice.Parent, + Title: notice.Title, + Kind: notice.Kind, + Program: notice.Program, + State: state, + Stopped: notice.Stopped, + Report: notice.Report, + Model: notice.Model, + CostUSD: notice.CostUSD, + ElapsedMS: notice.Elapsed.Milliseconds(), + StartedAt: notice.StartedAt, + EndedAt: notice.EndedAt, + Copy: notice.Copy, + PendingRun: notice.PendingRun, + Ending: notice.Ending, + Branch: notice.Branch, + Merge: notice.Merge, + Result: notice.Result, + Changed: append([]string(nil), notice.Changed...), + PlanTask: notice.PlanTask, + RecoveredCompletion: notice.RecoveredCompletion, } } @@ -1157,16 +1159,17 @@ func runRowNotice(record runRecord) TaskNotice { CostUSD: record.CostUSD, Elapsed: time.Duration(record.ElapsedMS) * time.Millisecond, - StartedAt: record.StartedAt, - EndedAt: record.EndedAt, - Copy: record.Copy, - PendingRun: record.PendingRun, - Ending: record.Ending, - Branch: record.Branch, - Merge: record.Merge, - Result: record.Result, - Changed: append([]string(nil), record.Changed...), - PlanTask: record.PlanTask, + StartedAt: record.StartedAt, + EndedAt: record.EndedAt, + Copy: record.Copy, + PendingRun: record.PendingRun, + Ending: record.Ending, + Branch: record.Branch, + Merge: record.Merge, + Result: record.Result, + Changed: append([]string(nil), record.Changed...), + PlanTask: record.PlanTask, + RecoveredCompletion: record.RecoveredCompletion, } if record.CrewState != nil && record.CrewState.Routed { decision := record.CrewState.Decision diff --git a/internal/session/tools_tasks.go b/internal/session/tools_tasks.go index a383b0505..90b022f85 100644 --- a/internal/session/tools_tasks.go +++ b/internal/session/tools_tasks.go @@ -768,6 +768,17 @@ func (a *Agent) oneTask(ctx context.Context, token string, parsed tasksArguments return a.stopOneTask(entry, id, true, parsed.Say) } } + if resolution := strings.TrimSpace(parsed.Resolve); resolution != "" { + id, err := strconv.ParseUint(strings.TrimPrefix(strings.TrimPrefix(token, "#"), "t-"), 10, 64) + if err == nil { + if reply, found, err := a.resolveCompletedRun(ctx, id, TaskResolution(strings.ToLower(resolution)), parsed.Say, TaskAskOwnerModel); found { + if err != nil { + return capitalized(err.Error()) + ".", true, nil + } + return reply, false, nil + } + } + } rows := a.withoutOwnRunRows(a.taskRows()) entry, found := a.taskByToken(rows, token) if !found { diff --git a/internal/tui3/planrail.go b/internal/tui3/planrail.go index e698e41d4..167d53677 100644 --- a/internal/tui3/planrail.go +++ b/internal/tui3/planrail.go @@ -70,7 +70,7 @@ func planNodeState(row session.PlanTaskRow) session.TaskState { return session.TaskDone case "failed", "cancelled": return session.TaskFailed - case "paused": + case "paused", string(session.TaskUnverified): return session.TaskUnverified } return "" diff --git a/internal/tui3/taskplan.go b/internal/tui3/taskplan.go index af56b24e7..b22dcdced 100644 --- a/internal/tui3/taskplan.go +++ b/internal/tui3/taskplan.go @@ -184,7 +184,7 @@ func planStateWord(row session.PlanTaskRow) string { return "done" case "failed", "cancelled": return "incomplete" - case "paused": + case "paused", string(session.TaskUnverified): return "your call" } // A status this build has never heard of draws NOTHING rather than a word it @@ -249,7 +249,7 @@ func planStatus(row session.PlanTaskRow) session.TaskStatus { Presence: session.TaskPresenceIncomplete, Word: planStateWord(row), } - case "paused": + case "paused", string(session.TaskUnverified): return session.TaskStatus{ Tier: session.TaskTierYourCall, Presence: session.TaskPresenceNeedsLook, @@ -268,6 +268,8 @@ func planEntryStatus(store string) string { switch strings.TrimSpace(store) { case "pending", "ready", "claimed", "running", "paused": return string(session.TaskRunning) + case string(session.TaskUnverified): + return string(session.TaskUnverified) case "done": return string(session.TaskDone) case "failed", "cancelled": @@ -912,6 +914,9 @@ func (a *app) tasksPlanKeyWords(row session.PlanTaskRow) []string { // row already draws for its state, so the key line and the page's sentences // cannot disagree about which tasks can still move. func planEnded(row session.PlanTaskRow) bool { + if row.Status == string(session.TaskUnverified) { + return true + } switch planStateWord(row) { case "done", "incomplete", "stopped": return true diff --git a/internal/tui3/taskplan_test.go b/internal/tui3/taskplan_test.go index 06f3d7bd7..ec6712887 100644 --- a/internal/tui3/taskplan_test.go +++ b/internal/tui3/taskplan_test.go @@ -782,3 +782,18 @@ func TestAHeldPlanRailRowKeepsItsReasonAndQueuedRoomState(t *testing.T) { t.Fatalf("held plan room state = %q, want queued · machine busy", got) } } + +func TestRecoveredCompletedPlanShowsReviewWithoutDisablingPausedWork(t *testing.T) { + recovered := session.PlanTaskRow{ID: "t-7", Title: "Mock completed work", Status: string(session.TaskUnverified)} + if planStateWord(recovered) != "your call" || planNodeState(recovered) != session.TaskUnverified || planStatus(recovered).Tier != session.TaskTierYourCall || !planEnded(recovered) || len((&app{}).tasksPlanKeyWords(recovered)) != 0 { + t.Fatal("completed review does not read as an ended worker needing an explicit decision") + } + paused := session.PlanTaskRow{ID: "t-part", Parent: "t-root", Status: "paused"} + if planEnded(paused) { + t.Fatal("recovered review disabled existing paused task controls") + } + words := strings.Join((&app{}).tasksPlanKeyWords(paused), " · ") + if !strings.Contains(words, tasksPlanResumeWord) || !strings.Contains(words, tasksPlanCancelWord) { + t.Fatalf("paused task lost resume/Stop: %q", words) + } +} From 3e17264568c5e2187dc359218059f02e970b0180 Mon Sep 17 00:00:00 2001 From: santoshkumarradha Date: Fri, 23 Oct 2026 00:47:11 -0400 Subject: [PATCH 04/12] test: keep recovery acceptance on the real engine road --- internal/e2e/completed_plan_recovery_wire_e2e_test.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/internal/e2e/completed_plan_recovery_wire_e2e_test.go b/internal/e2e/completed_plan_recovery_wire_e2e_test.go index 208da42c3..e57d8a19e 100644 --- a/internal/e2e/completed_plan_recovery_wire_e2e_test.go +++ b/internal/e2e/completed_plan_recovery_wire_e2e_test.go @@ -131,7 +131,7 @@ func runCompletedPlanRecovery(t *testing.T, live bool) { defer server.Close() base = server.URL } - env := []string{config.APIKeyEnv + "=" + key, "CODEAF_BASE_URL=" + base, "CODEAF_PROFILE_DIR=", "CODEAF_CALL_LOG_BODIES=1"} + env := []string{config.APIKeyEnv + "=" + key, "CODEAF_BASE_URL=" + base, "CODEAF_PROFILE_DIR=", "CODEAF_DEBUG=0", "CODEAF_CALL_LOG_BODIES=0"} stopRecoveryEngineOnCleanup(t, env, home, ws) rig := startWithEnv(t, env, "completed_recovery", home, ws, 180, 50, "chat", "--model", model, "--one-model", "--yolo") rig.skipSetup(t) From 88776f5157dddb323d8adf17e3b0fd52d0a34424 Mon Sep 17 00:00:00 2001 From: santoshkumarradha Date: Fri, 23 Oct 2026 00:49:37 -0400 Subject: [PATCH 05/12] docs: record explicit completed plan recovery --- .../unreleased/1911-completed-plan-recovery.md | 14 ++++++++++++++ 1 file changed, 14 insertions(+) create mode 100644 docs/changes/unreleased/1911-completed-plan-recovery.md diff --git a/docs/changes/unreleased/1911-completed-plan-recovery.md b/docs/changes/unreleased/1911-completed-plan-recovery.md new file mode 100644 index 000000000..9b8725b84 --- /dev/null +++ b/docs/changes/unreleased/1911-completed-plan-recovery.md @@ -0,0 +1,14 @@ +--- +kind: fixed +title: Recover completed plan work through explicit acceptance +pr: 1911 +surface: [chat, engine] +invalidates: + - "A plan completed before restart could remain interrupted with no working acceptance door. Its saved work now returns as your call, preserving the copy, crew, spend and completion time." + - "An unverified recovered plan could otherwise take an unattended default. Recovery waits for explicit acceptance or refusal and never replays the completed worker." +--- + +Existing task-page and conversation decisions can land the saved work and report +its actual commit. Ownership, branch and Stop checks remain enforced; failed +checkpoint or page-receipt writes report the actual outcome without claiming +that a completed merge was rolled back or safely recorded. From bf27cedb128a6399081f9c3bee5e81e6bbda3739 Mon Sep 17 00:00:00 2001 From: santoshkumarradha Date: Fri, 23 Oct 2026 00:58:02 -0400 Subject: [PATCH 06/12] Restore completed run receipt prices from its spend ledger --- .../session/completed_run_recovery_test.go | 66 ++++++++++++++++++- internal/session/task_run_recover.go | 4 ++ 2 files changed, 69 insertions(+), 1 deletion(-) diff --git a/internal/session/completed_run_recovery_test.go b/internal/session/completed_run_recovery_test.go index 850de30a2..c0f7a7088 100644 --- a/internal/session/completed_run_recovery_test.go +++ b/internal/session/completed_run_recovery_test.go @@ -16,6 +16,10 @@ import ( ) func completedRecoveryFixture(t *testing.T) (*Agent, string, *TaskCopyRecord, *beltRunDouble, uint64, time.Time, func() *Agent) { + return completedRecoveryFixtureWithSpend(t, 0.42, 0) +} + +func completedRecoveryFixtureWithSpend(t *testing.T, checkpointCost, ledgerCost float64) (*Agent, string, *TaskCopyRecord, *beltRunDouble, uint64, time.Time, func() *Agent) { t.Helper() t.Setenv("CODEAF_TASK_BELT", "bash") engine := newBeltRunDouble("must not run") @@ -42,13 +46,18 @@ func completedRecoveryFixture(t *testing.T) (*Agent, string, *TaskCopyRecord, *b if err != nil { t.Fatal(err) } + if ledgerCost > 0 { + if err := store.AddSpend(store.RootID(), "mock/model", "work", ledgerCost, 100, 25); err != nil { + t.Fatal(err) + } + } if err := store.CompleteRoot("mock test passed; counter ready"); err != nil { t.Fatal(err) } root := store.Task(store.RootID()) _ = store.Close() copy := runCopyOf(tree) - first.publishRunRow(g, TaskNotice{ID: id, Title: "mock completed counter", State: TaskRunning, Copy: copy, CrewState: first.unroutedCrewRecord(), CostUSD: 0.42, StartedAt: time.Now().Add(-time.Minute), PlanTask: planStoreID(strconv.FormatUint(id, 10))}) + first.publishRunRow(g, TaskNotice{ID: id, Title: "mock completed counter", State: TaskRunning, Copy: copy, CrewState: first.unroutedCrewRecord(), CostUSD: checkpointCost, StartedAt: time.Now().Add(-time.Minute), PlanTask: planStoreID(strconv.FormatUint(id, 10))}) if err := first.Close(); err != nil { t.Fatal(err) } @@ -329,3 +338,58 @@ func TestCompletedPlanRecoveryReportsUnsavedRefusalWithoutChangingItsCopy(t *tes } nothingStarted(t, agent, engine) } + +func TestCompletedPlanRecoveryKeepsLedgerSpendWhenCheckpointIsUnpriced(t *testing.T) { + agent, _, _, engine, id, _, open := completedRecoveryFixtureWithSpend(t, 0, 0.37) + check := func(a *Agent) { + t.Helper() + kept, _ := runRowOf(a.graph(), id) + if kept.CostUSD != 0.37 { + t.Fatalf("run receipt lost authoritative ledger price: %+v", kept) + } + if a.Usage().CostUSD != 0 { + t.Fatal("restoring a price label charged the conversation again") + } + row := planRowFor(a.PlanTasks(), planStoreID(strconv.FormatUint(id, 10))) + if row == nil || row.USD != 0.37 { + t.Fatalf("recovery erased task's authoritative spend: %+v", row) + } + page, ok := a.PlanTaskPage(planStoreID(strconv.FormatUint(id, 10))) + if !ok || page.Row.USD != 0.37 { + t.Fatalf("task page erased its ledger spend: %+v", page.Row) + } + text, ok := a.planTaskText(a.runPlanTasks(), strconv.FormatUint(id, 10)) + if !ok || !strings.Contains(text, "$0.37") { + t.Fatalf("task reader lost ledger cost: %q", text) + } + } + check(agent) + if err := agent.Close(); err != nil { + t.Fatal(err) + } + agent = open() + check(agent) + if err := agent.ResolveUnverified(id, TaskAccept, "mock work inspected"); err != nil { + t.Fatal(err) + } + kept, _ := runRowOf(agent.graph(), id) + if kept.CostUSD != 0.37 { + t.Fatalf("accepted run receipt lost ledger spend: %+v", kept) + } + var indexCost float64 + for _, entry := range ReadTaskIndex(TaskIndexPath(agent.config.SessionFile)) { + if entry.ID == strconv.FormatUint(id, 10) { + indexCost = entry.Cost + } + } + if indexCost != 0.37 { + t.Fatalf("accepted task index lost ledger spend: %g", indexCost) + } + check(agent) + if err := agent.Close(); err != nil { + t.Fatal(err) + } + agent = open() + check(agent) + nothingStarted(t, agent, engine) +} diff --git a/internal/session/task_run_recover.go b/internal/session/task_run_recover.go index d26e1af2b..84c09a086 100644 --- a/internal/session/task_run_recover.go +++ b/internal/session/task_run_recover.go @@ -31,6 +31,7 @@ func (a *Agent) recoverBeltRun() { return } root := store.Task(store.RootID()) + recordedSpend := storeSpent(store) id, parseErr := strconv.ParseUint(store.RootID(), 10, 64) _ = store.Close() if root == nil || parseErr != nil { @@ -46,6 +47,9 @@ func (a *Agent) recoverBeltRun() { return } if terminalStoreStatus(root.Status) { + // The last running checkpoint can precede the worker's final receipt. + // Restore its price label from the ledger without charging it again. + kept.CostUSD = max(kept.CostUSD, recordedSpend) a.reconcileSettledRun(g, kept, root) return } From 16ec5aea08618f01317ef8b5893ca0c695808657 Mon Sep 17 00:00:00 2001 From: santoshkumarradha Date: Fri, 23 Oct 2026 01:01:25 -0400 Subject: [PATCH 07/12] Document recovered receipt cost preservation --- docs/changes/unreleased/1911-completed-plan-recovery.md | 3 +++ 1 file changed, 3 insertions(+) diff --git a/docs/changes/unreleased/1911-completed-plan-recovery.md b/docs/changes/unreleased/1911-completed-plan-recovery.md index 9b8725b84..e9e1a53ac 100644 --- a/docs/changes/unreleased/1911-completed-plan-recovery.md +++ b/docs/changes/unreleased/1911-completed-plan-recovery.md @@ -12,3 +12,6 @@ Existing task-page and conversation decisions can land the saved work and report its actual commit. Ownership, branch and Stop checks remain enforced; failed checkpoint or page-receipt writes report the actual outcome without claiming that a completed merge was rolled back or safely recorded. + +When the last checkpoint predates the final spending receipt, recovery restores +the run's price from its existing ledger without charging the conversation again. From 4971063f2887f2ec22debe753aab73ff53de2c6e Mon Sep 17 00:00:00 2001 From: santoshkumarradha Date: Fri, 23 Oct 2026 01:07:38 -0400 Subject: [PATCH 08/12] test: recover after real worker spend and guarded landing interruption --- .../completed_plan_recovery_wire_e2e_test.go | 194 ++++++++++++++++-- 1 file changed, 181 insertions(+), 13 deletions(-) diff --git a/internal/e2e/completed_plan_recovery_wire_e2e_test.go b/internal/e2e/completed_plan_recovery_wire_e2e_test.go index e57d8a19e..d1ab5ea4a 100644 --- a/internal/e2e/completed_plan_recovery_wire_e2e_test.go +++ b/internal/e2e/completed_plan_recovery_wire_e2e_test.go @@ -28,8 +28,8 @@ func TestCompletedPlanRecoversAndAcceptsOnTheRealEngine(t *testing.T) { } func TestCompletedPlanRecoveryLiveWorkflow(t *testing.T) { runCompletedPlanRecovery(t, true) } -// The interruption is explicit fictional test work: the actual worker persists -// plandb done then stops only its disposable engine, before landing can run. +// The worker finishes ordinary fictional work. A harness-owned one-shot git +// hook pauses only its disposable engine after completion and before landing. func runCompletedPlanRecovery(t *testing.T, live bool) { if runtime.GOOS != "linux" { t.Skip("owned engine identity check uses Linux proc metadata") @@ -60,8 +60,8 @@ func runCompletedPlanRecovery(t *testing.T, live bool) { } home := recoveryShortHome(t, newHome(t, rows)) ws := newWorkspace(t, "completed-plan", false) + statesWorkBranch(t, ws) var mu sync.Mutex - enginePID := 0 assignmentCalls := 0 accepted := false base := "https://openrouter.ai/api/v1" @@ -102,9 +102,8 @@ func runCompletedPlanRecovery(t *testing.T, live bool) { if len(assignment) == 3 { mu.Lock() assignmentCalls++ - pid := enginePID mu.Unlock() - command := fmt.Sprintf("printf 'MOCK_RECOVERY_VALUE\\n' > recover.txt && printf 'grep -qx MOCK_RECOVERY_VALUE recover.txt\\n' > recover-check.txt && sh recover-check.txt && plandb done %s --agent %s --result 'MOCK_RECOVERY_VALUE verified' && kill -STOP %d; sleep 2", assignment[1], assignment[2], pid) + command := fmt.Sprintf("printf 'MOCK_RECOVERY_VALUE\\n' > recover.txt && printf 'grep -qx MOCK_RECOVERY_VALUE recover.txt\\n' > recover-check.txt && sh recover-check.txt && plandb done %s --agent %s --result 'MOCK_RECOVERY_VALUE verified'", assignment[1], assignment[2]) args, _ := json.Marshal(map[string]string{"command": command}) terminalStopResponse(w, model, "", map[string]any{"index": 0, "id": "recovery-finish-" + assignment[2], "type": "function", "function": map[string]string{"name": "bash", "arguments": string(args)}}, body.Stream) return @@ -147,27 +146,26 @@ func runCompletedPlanRecovery(t *testing.T, live bool) { t.Fatalf("no actual engine PID: %s", status) } assertCompletedEngineIdentity(t, pid, home, ws) - mu.Lock() - enginePID = pid - mu.Unlock() t.Cleanup(func() { if completedEnginePaused(pid) && completedEngineIdentity(pid, home, ws) { _ = syscall.Kill(pid, syscall.SIGCONT) } }) beforeHead := strings.TrimSpace(completedGit(t, ws, "rev-parse", "HEAD")) - prompt := fmt.Sprintf("/task solo disposable recovery exercise: write recover.txt containing exactly MOCK_RECOVERY_VALUE, write recover-check.txt containing exactly the runnable shell command grep -qx MOCK_RECOVERY_VALUE recover.txt, and run that check. Finish with ONE compound bash call that runs the check, marks your assigned plan task done using plandb done with your actual assigned task id and claimed agent and result 'MOCK_RECOVERY_VALUE verified', then immediately executes kill -STOP %d followed by sleep 2. This PID is only this disposable project's engine; pausing it is the explicit lifecycle test. Do not stop any other process. Do not delegate further.", pid) + marker := installCompletedPauseHook(t, pid, home, ws) + prompt := "/task solo write recover.txt containing exactly MOCK_RECOVERY_VALUE, write recover-check.txt containing exactly the runnable shell command grep -qx MOCK_RECOVERY_VALUE recover.txt, run that check, and finish normally using plandb done with your actual assigned task id and claimed agent and result 'MOCK_RECOVERY_VALUE verified'. Do not delegate further or make git commits yourself; leave commits and landing to the runtime." + rig.lit(prompt) rig.keys("Enter") deadline := time.Now().Add(2 * time.Minute) for !completedEnginePaused(pid) { if time.Now().After(deadline) { - t.Fatalf("actual worker never reached done-before-land pause: %s", rig.capture()) + t.Fatalf("actual completed run never reached its guarded landing pause: %s", rig.capture()) } time.Sleep(20 * time.Millisecond) } - // The engine was stopped by the actual worker, not by a timer guessing when - // it had finished. Confirm the durable root and untouched original checkout. + // The guarded one-shot git hook stops the engine after real worker spend + // and completion exist. Confirm that barrier and the untouched checkout. plan := completedPlanPath(t, home) ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) cmd := guardedCommand(t, ctx, home, append(append(os.Environ(), env...), "CODEAF_HOME="+home), binary(t), "plandb", "show", "t-1", "--db", plan, "--json") @@ -179,8 +177,34 @@ func runCompletedPlanRecovery(t *testing.T, live bool) { if completedGit(t, ws, "rev-parse", "HEAD") != beforeHead || fileExists(filepath.Join(ws, "recover.txt")) { t.Fatal("fixture did not stop before original checkout landing") } + if !fileExists(marker) { + t.Fatal("actual guarded git hook did not record its completion barrier") + } + originalHook := completedGit(t, ws, "rev-parse", "--git-path", "hooks/prepare-commit-msg") + if !filepath.IsAbs(originalHook) { + originalHook = filepath.Join(ws, originalHook) + } + if fileExists(originalHook) { + t.Fatal("one-shot original hook was not removed") + } + beforeLedger := completedTaskLedger(t, home, env, plan) + if beforeLedger <= 0 { + t.Fatal("completed worker has no actual task ledger spend") + } + checkpointBefore := completedCheckpointCost(home) + t.Logf("Paused completed root: task ledger=%g; running checkpoint cost=%g.", beforeLedger, checkpointBefore) proof := completedProofDir(t, home) checkCompletedProof(t, proof) + lock := completedGit(t, proof, "rev-parse", "--git-path", "index.lock") + if !filepath.IsAbs(lock) { + lock = filepath.Join(proof, lock) + } + for until := time.Now().Add(5 * time.Second); fileExists(lock); { + if time.Now().After(until) { + t.Fatal("hook's git process did not release its exact index lock") + } + time.Sleep(20 * time.Millisecond) + } beforeSpend := stopSpendSnapshot(t, home) if beforeSpend.Cost <= 0 { t.Fatal("interrupted live work recorded no spend") @@ -224,6 +248,13 @@ func runCompletedPlanRecovery(t *testing.T, live bool) { if nowCalls != callsBefore { t.Fatal("completed worker was rerun after engine restart") } + if after := completedTaskLedger(t, home, env, plan); after != beforeLedger { + t.Fatalf("task ledger changed across restart: %g -> %g", beforeLedger, after) + } + if cost := completedCheckpointCost(home); cost+1e-9 < beforeLedger { + t.Fatalf("recovered task/index lost its actual spend: cost=%g ledger=%g", cost, beforeLedger) + } + assertCompletedPlanPrice(t, rig, beforeLedger) checkCompletedProof(t, proof) // A restored landing card owns its displayed answer keys. Prefer that // actual door, and submit a chat request only on a visible ordinary box. @@ -234,8 +265,11 @@ func runCompletedPlanRecovery(t *testing.T, live bool) { } t.Logf("Actual restored UI before explicit acceptance:\n%s", screen) if strings.Contains(screen, "a accept") { - rig.keys("a") + rig.keys("a", "Enter") } else { + if strings.Contains(screen, say(t, "roomTabsWords")) { + t.Fatalf("restored task room offers no actual Accept card; refusing to submit a task note: %s", screen) + } if !strings.Contains(screen, "›") { t.Fatalf("restored UI offers neither an Accept card nor a normal chat box: %s", screen) } @@ -255,6 +289,9 @@ func runCompletedPlanRecovery(t *testing.T, live bool) { t.Fatal("explicit acceptance did not land a commit") } checkCompletedProof(t, ws) + if cost := completedCheckpointCost(home); cost+1e-9 < beforeLedger { + t.Fatalf("accepted task/index lost its actual spend: cost=%g ledger=%g", cost, beforeLedger) + } receipt := completedCommitReceipt(t, home) if receipt == "" { t.Fatal("accepted task has no actual work commit receipt") @@ -369,3 +406,134 @@ func completedCommitReceipt(t *testing.T, home string) string { } return "" } + +func installCompletedPauseHook(t *testing.T, pid int, home, ws string) string { + t.Helper() + var transcripts []string + _ = filepath.WalkDir(home, func(path string, d os.DirEntry, err error) error { + if err == nil && !d.IsDir() && d.Name() == "transcript.jsonl" { + transcripts = append(transcripts, path) + } + return nil + }) + if len(transcripts) != 1 { + t.Fatalf("expected exactly one current conversation before the task, got %d", len(transcripts)) + } + plan := filepath.Join(filepath.Dir(transcripts[0]), "plandb.db") + hook := completedGit(t, ws, "rev-parse", "--git-path", "hooks/prepare-commit-msg") + if !filepath.IsAbs(hook) { + hook = filepath.Join(ws, hook) + } + marker := filepath.Join(home, "completed-hook-fired.json") + fixture, _ := json.Marshal(map[string]any{"pid": pid, "home": home, "workspace": ws, "binary": binary(t), "plan": plan, "marker": marker, "hook": hook}) + script := "#!/usr/bin/env python3\nimport json, os, pathlib, signal, subprocess, sys\nfixture = json.loads(" + strconv.Quote(string(fixture)) + ")\n" + ` +# Every child inherits the engine's already guarded, isolated fixture environment. +pid=fixture['pid'] +command=pathlib.Path('/proc/%d/cmdline'%pid).read_bytes() +environment=pathlib.Path('/proc/%d/environ'%pid).read_bytes() +if b'engine\0' not in command or ('--workspace\0'+fixture['workspace']+'\0').encode() not in command: + raise SystemExit(1) +if ('CODEAF_HOME='+fixture['home']+'\0').encode() not in environment: + raise SystemExit(1) +if pathlib.Path('recover.txt').read_text().strip()!='MOCK_RECOVERY_VALUE': + raise SystemExit(1) +if pathlib.Path('recover-check.txt').read_text().strip()!='grep -qx MOCK_RECOVERY_VALUE recover.txt': + raise SystemExit(1) +subprocess.run(['sh','recover-check.txt'],check=True) +base=[fixture['binary'],'plandb'] +root=json.loads(subprocess.check_output(base+['show','t-1','--db',fixture['plan'],'--json'])) +if root.get('id')!='t-1' or root.get('status')!='done': + raise SystemExit(1) +ledger=json.loads(subprocess.check_output(base+['spend','--by','task','--db',fixture['plan'],'--json'])) +usd=sum(row['usd'] for row in ledger if row['key']=='1') +if usd<=0: + raise SystemExit(1) +pathlib.Path(fixture['marker']).write_text(json.dumps({'root':'1','task_usd':usd})) +for owned_hook in {pathlib.Path(__file__).resolve(),pathlib.Path(fixture['hook']).resolve()}: + owned_hook.unlink(missing_ok=True) +os.kill(pid,signal.SIGSTOP) # Only the exact verified disposable engine. +raise SystemExit(1) # Let git release its index lock; no sleeper/commit survives. +` + if err := os.WriteFile(hook, []byte(script), 0o700); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = os.Remove(hook) }) + return marker +} + +func completedTaskLedger(t *testing.T, home string, env []string, plan string) float64 { + t.Helper() + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + cmd := guardedCommand(t, ctx, home, append(append(os.Environ(), env...), "CODEAF_HOME="+home), binary(t), "plandb", "spend", "--by", "task", "--db", plan, "--json") + raw, err := cmd.CombinedOutput() + if err != nil { + t.Fatalf("actual task ledger CLI: %v %s", err, raw) + } + var rows []struct { + Key string `json:"key"` + USD float64 `json:"usd"` + } + if err := json.Unmarshal(raw, &rows); err != nil { + t.Fatal(err) + } + var usd float64 + for _, row := range rows { + if row.Key == "1" { + usd += row.USD + } + } + return usd +} +func completedCheckpointCost(home string) float64 { + var cost float64 + _ = filepath.WalkDir(home, func(path string, d os.DirEntry, err error) error { + if err != nil || d.IsDir() || d.Name() != "tasks.json" { + return nil + } + raw, _ := os.ReadFile(path) + var record struct { + Runs []struct { + ID uint64 `json:"id"` + USD float64 `json:"costUsd"` + } `json:"runs"` + } + if json.Unmarshal(raw, &record) == nil { + for _, run := range record.Runs { + if run.ID == 1 { + cost = run.USD + } + } + } + return nil + }) + return cost +} +func assertCompletedPlanPrice(t *testing.T, rig *rig, usd float64) { + t.Helper() + openTasksPlace(t, rig) + tasksPlaceRunRow(t, rig) + rig.keys("Enter") + screen := rig.waitFor(15*time.Second, say(t, "roomTabsWords")) + if !strings.Contains(screen, say(t, "roomTabsWords")) { + t.Fatal("recovered actual plan room did not load") + } + price := fmt.Sprintf("$%.2f", usd) + if usd < 0.01 { + price = config.SubCent(usd) + } + deadline := time.Now().Add(10 * time.Second) + for { + screen = rig.capture() + for _, line := range strings.Split(screen, "\n") { + if strings.Contains(line, "your call") && strings.Contains(line, price) { + t.Logf("Actual recovered task-room header preserves ledger price %s:\n%s", price, screen) + return + } + } + if time.Now().After(deadline) { + t.Fatalf("actual recovered task header lost ledger price %s: %s", price, screen) + } + time.Sleep(100 * time.Millisecond) + } +} From cedc221e7cd1f3a4afd952fd6b462a5ca90900a7 Mon Sep 17 00:00:00 2001 From: santoshkumarradha Date: Fri, 23 Oct 2026 01:08:37 -0400 Subject: [PATCH 09/12] Test completed recovery ownership and archived plan boundaries --- .../session/completed_run_recovery_test.go | 84 +++++++++++++++++++ 1 file changed, 84 insertions(+) diff --git a/internal/session/completed_run_recovery_test.go b/internal/session/completed_run_recovery_test.go index c0f7a7088..c356f2357 100644 --- a/internal/session/completed_run_recovery_test.go +++ b/internal/session/completed_run_recovery_test.go @@ -393,3 +393,87 @@ func TestCompletedPlanRecoveryKeepsLedgerSpendWhenCheckpointIsUnpriced(t *testin check(agent) nothingStarted(t, agent, engine) } + +func TestCompletedPlanRecoveryAcceptsItsArchiveWithoutAdoptingTheNewerPlan(t *testing.T) { + agent, repo, _, engine, id, _, _ := completedRecoveryFixture(t) + path := agent.graph().planPath() + if err := setAsideRunStore(path); err != nil { + t.Fatal(err) + } + nextID := strconv.FormatUint(id+1, 10) + newer, err := plandb.Open(path, "mock", nextID, "newer mock work", "leave this new request alone", agent.graph().planChat()) + if err != nil { + t.Fatal(err) + } + defer newer.Close() + if err := newer.AddSpend(nextID, "mock/newer-model", "work", 0.91, 10, 2); err != nil { + t.Fatal(err) + } + before := newer.Task(nextID) + if err := agent.ResolveUnverified(id, TaskAccept, "read the older saved work"); err != nil { + t.Fatal(err) + } + if data, err := os.ReadFile(filepath.Join(repo, "counter.txt")); err != nil || string(data) != "mock completed work\n" { + t.Fatalf("archived work did not land: %q %v", data, err) + } + after := newer.Task(nextID) + if after.Status != before.Status || after.Description != before.Description || after.Result != "" || len(newer.Notes(nextID, 0)) != 0 { + t.Fatalf("old acceptance adopted or changed the newer plan: %+v", after) + } + archived, err := plandb.Open(path+".1", "", "", "", "") + if err != nil { + t.Fatal(err) + } + defer archived.Close() + notes := archived.Notes(strconv.FormatUint(id, 10), 0) + if len(notes) == 0 || !strings.Contains(notes[len(notes)-1].Body, "work commit ") { + t.Fatalf("receipt went to the wrong plan: %+v", notes) + } + kept, _ := runRowOf(agent.graph(), id) + if kept.State != TaskDone || kept.CostUSD != 0.42 { + t.Fatalf("archived acceptance mixed newer task's state or spend: %+v", kept) + } + nothingStarted(t, agent, engine) +} + +func TestCompletedPlanRecoveryRefusesAPlanFromAnotherChat(t *testing.T) { + agent, repo, copy, engine, id, _, _ := completedRecoveryFixture(t) + before := runTreeHead(repo) + path := agent.graph().planPath() + armPlanStore(t, agent, path, "mock-foreign-chat") + err := agent.ResolveUnverified(id, TaskAccept, "must not cross chats") + if err == nil || !strings.Contains(err.Error(), "no completed plan in this conversation") { + t.Fatalf("foreign plan was accepted or refused without explanation: %v", err) + } + kept, _ := runRowOf(agent.graph(), id) + if kept.State != TaskUnverified || runTreeHead(repo) != before { + t.Fatal("foreign-plan refusal changed existing work") + } + if _, err := os.Stat(filepath.Join(copy.Dir, "counter.txt")); err != nil { + t.Fatal("foreign-plan refusal removed saved work") + } + nothingStarted(t, agent, engine) +} + +func TestCompletedPlanRecoveryNamesTheMissingCopyThroughItsActualToolDoor(t *testing.T) { + agent, repo, _, engine, id, _, open := completedRecoveryFixture(t) + before := runTreeHead(repo) + kept, _ := runRowOf(agent.graph(), id) + kept.Copy = nil + kept.State = TaskInterrupted + agent.graph().keepRunRows(id, []TaskNotice{kept}) + if err := agent.Close(); err != nil { + t.Fatal(err) + } + agent = open() + args, _ := json.Marshal(map[string]any{"id": strconv.FormatUint(id, 10), "resolve": "accept"}) + reply, bad, err := agent.tasksTool().Execute(context.Background(), args) + if err != nil || !bad || !strings.Contains(reply, "working copy was not written down") || strings.Contains(reply, "No task") { + t.Fatalf("missing saved copy became a false task miss or success: %q bad=%v err=%v", reply, bad, err) + } + kept, _ = runRowOf(agent.graph(), id) + if kept.State != TaskInterrupted || kept.Copy != nil || runTreeHead(repo) != before { + t.Fatalf("missing-copy refusal invented or changed work: %+v", kept) + } + nothingStarted(t, agent, engine) +} From 98220005b6cfb533faa99027cc9f0aa3be192f31 Mon Sep 17 00:00:00 2001 From: santoshkumarradha Date: Fri, 23 Oct 2026 01:13:50 -0400 Subject: [PATCH 10/12] test: accept restored recovery card before inspecting task room --- .../completed_plan_recovery_wire_e2e_test.go | 28 +++++++------------ 1 file changed, 10 insertions(+), 18 deletions(-) diff --git a/internal/e2e/completed_plan_recovery_wire_e2e_test.go b/internal/e2e/completed_plan_recovery_wire_e2e_test.go index d1ab5ea4a..419aeea9b 100644 --- a/internal/e2e/completed_plan_recovery_wire_e2e_test.go +++ b/internal/e2e/completed_plan_recovery_wire_e2e_test.go @@ -254,29 +254,20 @@ func runCompletedPlanRecovery(t *testing.T, live bool) { if cost := completedCheckpointCost(home); cost+1e-9 < beforeLedger { t.Fatalf("recovered task/index lost its actual spend: cost=%g ledger=%g", cost, beforeLedger) } - assertCompletedPlanPrice(t, rig, beforeLedger) checkCompletedProof(t, proof) - // A restored landing card owns its displayed answer keys. Prefer that - // actual door, and submit a chat request only on a visible ordinary box. + // Wait for the actual restored card, then let its input settle guard expire. + // Reading the persisted recovery record alone does not mean the UI is ready. screen := rig.capture() - for until := time.Now().Add(3 * time.Second); !strings.Contains(screen, "a accept") && time.Now().Before(until); { + for until := time.Now().Add(10 * time.Second); (!strings.Contains(screen, "a accept") || !strings.Contains(screen, "n not right")) && time.Now().Before(until); { time.Sleep(100 * time.Millisecond) screen = rig.capture() } - t.Logf("Actual restored UI before explicit acceptance:\n%s", screen) - if strings.Contains(screen, "a accept") { - rig.keys("a", "Enter") - } else { - if strings.Contains(screen, say(t, "roomTabsWords")) { - t.Fatalf("restored task room offers no actual Accept card; refusing to submit a task note: %s", screen) - } - if !strings.Contains(screen, "›") { - t.Fatalf("restored UI offers neither an Accept card nor a normal chat box: %s", screen) - } - accept := "ACCEPT_RECOVERED_PROOF: I inspected the preserved files and independently reran recover-check.txt successfully. Explicitly accept and land completed task 1 using tasks resolve accept. Report the actual landed work commit; do not rerun or redelegate the work." - rig.lit(accept) - rig.keys("Enter") + if !strings.Contains(screen, "a accept") || !strings.Contains(screen, "n not right") { + t.Fatalf("restored UI offers no actual acceptance card: %s", screen) } + time.Sleep(350 * time.Millisecond) + t.Logf("Actual restored UI before explicit acceptance:\n%s", rig.capture()) + rig.keys("a", "Enter") for deadline = time.Now().Add(2 * time.Minute); completedCommitReceipt(t, home) == "" || completedGit(t, ws, "rev-parse", "HEAD") == beforeHead; { if time.Now().After(deadline) { t.Fatalf("actual explicit acceptance did not produce a landed commit receipt: %s", rig.capture()) @@ -299,6 +290,7 @@ func runCompletedPlanRecovery(t *testing.T, live bool) { if err := exec.Command("git", "-C", ws, "merge-base", "--is-ancestor", receipt, "HEAD").Run(); err != nil { t.Fatalf("receipt commit was not landed: %s", receipt) } + assertCompletedPlanPrice(t, rig, beforeLedger) models := callModels(t, home) if live { for _, got := range models { @@ -526,7 +518,7 @@ func assertCompletedPlanPrice(t *testing.T, rig *rig, usd float64) { for { screen = rig.capture() for _, line := range strings.Split(screen, "\n") { - if strings.Contains(line, "your call") && strings.Contains(line, price) { + if strings.Contains(line, "done") && strings.Contains(line, price) { t.Logf("Actual recovered task-room header preserves ledger price %s:\n%s", price, screen) return } From 9730156d11e2166bd9962be0869c23dc652f2407 Mon Sep 17 00:00:00 2001 From: santoshkumarradha Date: Fri, 23 Oct 2026 01:17:03 -0400 Subject: [PATCH 11/12] test: verify full recovered run ledger including checker --- .../completed_plan_recovery_wire_e2e_test.go | 17 ++++++++++++----- 1 file changed, 12 insertions(+), 5 deletions(-) diff --git a/internal/e2e/completed_plan_recovery_wire_e2e_test.go b/internal/e2e/completed_plan_recovery_wire_e2e_test.go index 419aeea9b..f24e58881 100644 --- a/internal/e2e/completed_plan_recovery_wire_e2e_test.go +++ b/internal/e2e/completed_plan_recovery_wire_e2e_test.go @@ -6,6 +6,7 @@ import ( "context" "encoding/json" "fmt" + "math" "net/http" "net/http/httptest" "os" @@ -251,7 +252,7 @@ func runCompletedPlanRecovery(t *testing.T, live bool) { if after := completedTaskLedger(t, home, env, plan); after != beforeLedger { t.Fatalf("task ledger changed across restart: %g -> %g", beforeLedger, after) } - if cost := completedCheckpointCost(home); cost+1e-9 < beforeLedger { + if cost := completedCheckpointCost(home); math.Abs(cost-beforeLedger) > 1e-9 { t.Fatalf("recovered task/index lost its actual spend: cost=%g ledger=%g", cost, beforeLedger) } checkCompletedProof(t, proof) @@ -280,7 +281,7 @@ func runCompletedPlanRecovery(t *testing.T, live bool) { t.Fatal("explicit acceptance did not land a commit") } checkCompletedProof(t, ws) - if cost := completedCheckpointCost(home); cost+1e-9 < beforeLedger { + if cost := completedCheckpointCost(home); math.Abs(cost-beforeLedger) > 1e-9 { t.Fatalf("accepted task/index lost its actual spend: cost=%g ledger=%g", cost, beforeLedger) } receipt := completedCommitReceipt(t, home) @@ -290,6 +291,13 @@ func runCompletedPlanRecovery(t *testing.T, live bool) { if err := exec.Command("git", "-C", ws, "merge-base", "--is-ancestor", receipt, "HEAD").Run(); err != nil { t.Fatalf("receipt commit was not landed: %s", receipt) } + if after := completedTaskLedger(t, home, env, plan); math.Abs(after-beforeLedger) > 1e-9 { + t.Fatalf("explicit acceptance changed task ledger: %g -> %g", beforeLedger, after) + } + if got := stopSpendSnapshot(t, home); got != beforeSpend { + t.Fatalf("explicit acceptance launched new model work: before=%+v after=%+v", beforeSpend, got) + } + t.Logf("Accepted persisted run cost=%g; whole-run ledger=%g", completedCheckpointCost(home), beforeLedger) assertCompletedPlanPrice(t, rig, beforeLedger) models := callModels(t, home) if live { @@ -471,10 +479,9 @@ func completedTaskLedger(t *testing.T, home string, env []string, plan string) f } var usd float64 for _, row := range rows { - if row.Key == "1" { - usd += row.USD - } + usd += row.USD } + t.Logf("Actual whole-run task ledger: %s; total=%g", strings.TrimSpace(string(raw)), usd) return usd } func completedCheckpointCost(home string) float64 { From 08a1bb123a4769358b7d9fea8ef42bec48706778 Mon Sep 17 00:00:00 2001 From: santoshkumarradha Date: Fri, 23 Oct 2026 01:20:26 -0400 Subject: [PATCH 12/12] test: preserve full completed run spend through recovery --- .../session/completed_run_recovery_test.go | 53 ++++++++++++++++++- 1 file changed, 52 insertions(+), 1 deletion(-) diff --git a/internal/session/completed_run_recovery_test.go b/internal/session/completed_run_recovery_test.go index c356f2357..dd5c2dca5 100644 --- a/internal/session/completed_run_recovery_test.go +++ b/internal/session/completed_run_recovery_test.go @@ -19,7 +19,7 @@ func completedRecoveryFixture(t *testing.T) (*Agent, string, *TaskCopyRecord, *b return completedRecoveryFixtureWithSpend(t, 0.42, 0) } -func completedRecoveryFixtureWithSpend(t *testing.T, checkpointCost, ledgerCost float64) (*Agent, string, *TaskCopyRecord, *beltRunDouble, uint64, time.Time, func() *Agent) { +func completedRecoveryFixtureWithSpend(t *testing.T, checkpointCost, ledgerCost float64, checkerCost ...float64) (*Agent, string, *TaskCopyRecord, *beltRunDouble, uint64, time.Time, func() *Agent) { t.Helper() t.Setenv("CODEAF_TASK_BELT", "bash") engine := newBeltRunDouble("must not run") @@ -51,6 +51,20 @@ func completedRecoveryFixtureWithSpend(t *testing.T, checkpointCost, ledgerCost t.Fatal(err) } } + if len(checkerCost) > 0 { + if _, err := store.AddMany([]plandb.TaskSpec{{ID: "mock-checker", Title: "Mock checker", Description: "check completed counter", ParentID: store.RootID()}}); err != nil { + t.Fatal(err) + } + if _, err := store.Claim("mock-checker", "mock-checker"); err != nil { + t.Fatal(err) + } + if _, err := store.Done("mock-checker", "mock-checker", "mock check passed", nil, nil); err != nil { + t.Fatal(err) + } + if err := store.AddSpend("mock-checker", "mock/model", "check", checkerCost[0], 50, 10); err != nil { + t.Fatal(err) + } + } if err := store.CompleteRoot("mock test passed; counter ready"); err != nil { t.Fatal(err) } @@ -477,3 +491,40 @@ func TestCompletedPlanRecoveryNamesTheMissingCopyThroughItsActualToolDoor(t *tes } nothingStarted(t, agent, engine) } + +func TestCompletedPlanRecoveryKeepsWholeRunSpendWithoutChargingAgain(t *testing.T) { + agent, _, _, engine, id, _, open := completedRecoveryFixtureWithSpend(t, 0, 0.02, 0.02) + check := func(a *Agent) { + t.Helper() + kept, _ := runRowOf(a.graph(), id) + page, ok := a.PlanTaskPage(planStoreID(strconv.FormatUint(id, 10))) + if kept.CostUSD != 0.04 || !ok || page.Row.USD != 0.04 { + t.Fatalf("root and checker must appear once in whole-run receipt and page: receipt=%g page=%g", kept.CostUSD, page.Row.USD) + } + store, err := plandb.Open(a.graph().planPath(), "mock", strconv.FormatUint(id, 10), "mock completed counter", "write counter", a.graph().planChat()) + if err != nil { + t.Fatal(err) + } + spent := storeSpent(store) + _ = store.Close() + if spent != 0.04 || a.Usage().CostUSD != 0 { + t.Fatalf("recovery changed actual ledger or charged again: ledger=%g usage=%g", spent, a.Usage().CostUSD) + } + } + check(agent) + if err := agent.Close(); err != nil { + t.Fatal(err) + } + agent = open() + check(agent) + if err := agent.ResolveUnverified(id, TaskAccept, "mock root and checker inspected"); err != nil { + t.Fatal(err) + } + check(agent) + if err := agent.Close(); err != nil { + t.Fatal(err) + } + agent = open() + check(agent) + nothingStarted(t, agent, engine) +}