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..e9e1a53ac --- /dev/null +++ b/docs/changes/unreleased/1911-completed-plan-recovery.md @@ -0,0 +1,17 @@ +--- +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. + +When the last checkpoint predates the final spending receipt, recovery restores +the run's price from its existing ledger without charging the conversation again. 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..f24e58881 --- /dev/null +++ b/internal/e2e/completed_plan_recovery_wire_e2e_test.go @@ -0,0 +1,538 @@ +//go:build e2e + +package e2e + +import ( + "context" + "encoding/json" + "fmt" + "math" + "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 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") + } + 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) + statesWorkBranch(t, ws) + var mu sync.Mutex + 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++ + 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'", 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 + } + 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_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) + 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) + t.Cleanup(func() { + if completedEnginePaused(pid) && completedEngineIdentity(pid, home, ws) { + _ = syscall.Kill(pid, syscall.SIGCONT) + } + }) + beforeHead := strings.TrimSpace(completedGit(t, ws, "rev-parse", "HEAD")) + 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 completed run never reached its guarded landing pause: %s", rig.capture()) + } + time.Sleep(20 * time.Millisecond) + } + // 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") + 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") + } + 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") + } + 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") + } + if after := completedTaskLedger(t, home, env, plan); after != beforeLedger { + t.Fatalf("task ledger changed across restart: %g -> %g", beforeLedger, after) + } + 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) + // 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(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() + } + 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()) + } + time.Sleep(100 * time.Millisecond) + } + + finalHead := completedGit(t, ws, "rev-parse", "HEAD") + if finalHead == beforeHead { + t.Fatal("explicit acceptance did not land a commit") + } + checkCompletedProof(t, ws) + 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) + 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) + } + 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 { + 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 "" +} + +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 { + 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 { + 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, "done") && 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) + } +} 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..dd5c2dca5 --- /dev/null +++ b/internal/session/completed_run_recovery_test.go @@ -0,0 +1,530 @@ +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) { + return completedRecoveryFixtureWithSpend(t, 0.42, 0) +} + +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") + 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 ledgerCost > 0 { + if err := store.AddSpend(store.RootID(), "mock/model", "work", ledgerCost, 100, 25); err != nil { + 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) + } + 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: checkpointCost, 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) +} + +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) +} + +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) +} + +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) +} 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..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 } @@ -85,7 +89,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) + } +}