test(orchestrator): record what the SIGKILL resume test observes

internal/orchestrator/crash_test.go

@@ -1,10 +1,15 @@
 package orchestrator
 
 import (
+	"bytes"
 	"context"
+	"encoding/json"
+	"flag"
+	"fmt"
 	"os"
 	"os/exec"
 	"path/filepath"
+	"regexp"
 	"strings"
 	"syscall"
 	"testing"
@@ -17,6 +22,28 @@ import (
 	"github.com/akynte/boundedcode/internal/telemetry"
 )
 
+var updateReplay = flag.Bool("update-replay", false, "rewrite testdata/crash-replay.json from this run of TestSIGKILLResume")
+
+// crashReplay is what TestSIGKILLResume observes, in order. The test compares
+// it with testdata/crash-replay.json and fails on any difference, so a passing
+// run proves the file is what a real kill and resume produced. Readers (such as
+// a site that replays it) can rely on that; -update-replay rewrites the file
+// when the behaviour changes on purpose.
+type crashReplay struct {
+	Test  string       `json:"test"`
+	About string       `json:"about"`
+	Steps []replayStep `json:"steps"`
+}
+
+// rfc3339 matches the wall-clock times in observed messages, which differ on every run.
+var rfc3339 = regexp.MustCompile(`\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(\.\d+)?Z`)
+
+type replayStep struct {
+	Actor    string `json:"actor"`
+	Action   string `json:"action"`
+	Observed string `json:"observed"`
+}
+
 // TestKilledRunnerHelper is the child process of TestSIGKILLResume: it runs
 // the task with an agent that edits the repository and then hangs, until the
 // parent SIGKILLs it. Not a test on its own.
@@ -55,6 +82,16 @@ func TestSIGKILLResume(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
+	l := task.Ledger{DB: s.DB}
+	rec := crashReplay{Test: "TestSIGKILLResume", About: "A runner process is killed with SIGKILL in the middle of an agent turn; " +
+		"a second runner is refused, a third takes over the stale lease and finishes the task. Each step is what the test observed."}
+	see := func(actor, action, observed string) {
+		observed = rfc3339.ReplaceAllString(strings.ReplaceAll(observed, tk.ID, "<task>"), "<time>")
+		rec.Steps = append(rec.Steps, replayStep{actor, action, observed})
+	}
+	if t0, err := l.Get(ctx, tk.ID); err == nil {
+		see("test", "creates the task", fmt.Sprintf("status %s, phase %s", t0.Status, t0.Phase))
+	}
 	ready := filepath.Join(t.TempDir(), "ready")
 	cmd := exec.Command(os.Args[0], "-test.run=^TestKilledRunnerHelper$", "-test.count=1")
 	cmd.Env = append(os.Environ(), "BC_KILL_CHILD_DB="+filepath.Join(root, "state.db"), "BC_KILL_CHILD_ROOT="+root,
@@ -73,33 +110,52 @@ func TestSIGKILLResume(t *testing.T) {
 		}
 		time.Sleep(50 * time.Millisecond)
 	}
+	partial := "ledger-service/internal/consumer/balance_test.go"
+	_, statErr := os.Stat(filepath.Join(r.WorkDir(tk.ID), partial))
+	see("runner A", "runs as a separate process (lease owner \"child\"); its agent writes a file and stays mid-turn",
+		fmt.Sprintf("%s in the worktree: %t", partial, statErr == nil))
 	if err := cmd.Process.Signal(syscall.SIGKILL); err != nil {
 		t.Fatal(err)
 	}
-	_ = cmd.Wait()
+	waitErr := cmd.Wait()
+	see("test", "kills runner A with SIGKILL: no signal handler, no cleanup", fmt.Sprint(waitErr))
 
 	// Dead runner state: lease held, attempt active, phase implementing.
-	l := task.Ledger{DB: s.DB}
-	if ls, _ := l.LeaseOf(ctx, tk.ID); ls.Owner != "child" {
+	ls, _ := l.LeaseOf(ctx, tk.ID)
+	if ls.Owner != "child" {
 		t.Fatalf("lease after kill: %+v", ls)
 	}
+	if t1, err := l.Get(ctx, tk.ID); err == nil {
+		st, _ := l.Strategies(ctx, tk.ID)
+		attempt := "no attempt"
+		if len(st) > 0 {
+			attempt = fmt.Sprintf("attempt %d %s", st[0].Attempt, st[0].Outcome)
+		}
+		see("ledger", "what the dead runner left behind", fmt.Sprintf("lease held by %q; task %s in phase %s; %s", ls.Owner, t1.Status, t1.Phase, attempt))
+	}
 	// A second runner is refused while the lease is fresh...
 	r2 := newRunner(s, root, &scripted.Runtime{}, nil)
-	if _, err := r2.Run(ctx, tk.ID, RunOptions{}); err == nil || !strings.Contains(err.Error(), "another process") {
-		t.Fatalf("concurrent run allowed: %v", err)
+	_, err2 := r2.Run(ctx, tk.ID, RunOptions{})
+	if err2 == nil || !strings.Contains(err2.Error(), "another process") {
+		t.Fatalf("concurrent run allowed: %v", err2)
 	}
+	see("runner B", "tries to run the task while the lease is fresh", "refused: "+err2.Error())
 	// ...and takes over once it is stale.
 	old := leaseStale
 	leaseStale = 50 * time.Millisecond
 	defer func() { leaseStale = old }()
 	time.Sleep(100 * time.Millisecond)
+	see("test", "waits for the lease to go stale",
+		fmt.Sprintf("no heartbeat for longer than %s (%s in production, shortened for the test)", leaseStale, old))
 
 	var resumeMsg string
+	survived := false
 	fix := func(ws, msg string) (string, error) {
 		resumeMsg = msg
-		if _, err := os.Stat(filepath.Join(ws, "ledger-service/internal/consumer/balance_test.go")); err != nil {
+		if _, err := os.Stat(filepath.Join(ws, partial)); err != nil {
 			return "", err // the killed attempt's edits must survive
 		}
+		survived = true
 		return "negated the settlement amount", replaceIn(filepath.Join(ws, consumerFile),
 			"AmountCents: ev.AmountCents, Currency: ev.Currency},\n\t)", "AmountCents: -ev.AmountCents, Currency: ev.Currency},\n\t)")
 	}
@@ -115,26 +171,67 @@ func TestSIGKILLResume(t *testing.T) {
 	if !strings.Contains(resumeMsg, "resuming a task") {
 		t.Fatalf("resume pack not used:\n%s", resumeMsg)
 	}
+	for _, line := range strings.Split(resumeMsg, "\n") {
+		if strings.Contains(line, "resuming a task") {
+			see("runner C", "takes over the stale lease; its agent is told", strings.TrimSpace(line))
+			break
+		}
+	}
+	see("runner C", "its agent looks for the killed attempt's file", fmt.Sprintf("%s still in the worktree: %t", partial, survived))
 	strats, _ := l.Strategies(ctx, tk.ID)
 	if strats[0].Outcome != "rejected" || !strings.Contains(strats[0].Reason, "previous run stopped") {
 		t.Fatalf("orphaned attempt not reconciled: %+v", strats[0])
 	}
+	see("ledger", "reconciles the orphaned attempt", fmt.Sprintf("attempt %d %s: %s", strats[0].Attempt, strats[0].Outcome, strats[0].Reason))
 	evs, _ := telemetry.Events(ctx, s.DB, tk.ID, 0, 1000)
 	recovered := false
+	kinds := make([]string, 0, len(evs))
 	for _, e := range evs {
 		recovered = recovered || e.Kind == "task.recovered"
+		kinds = append(kinds, e.Kind)
 	}
 	if !recovered {
 		t.Fatal("no task.recovered event")
 	}
-	if ls, _ := l.LeaseOf(ctx, tk.ID); ls.Owner != "" {
+	see("ledger", "the task's event log, in order", strings.Join(kinds, " → "))
+	ls, _ = l.LeaseOf(ctx, tk.ID)
+	if ls.Owner != "" {
 		t.Fatalf("lease not released after completion: %+v", ls)
 	}
 	wts, _ := l.Worktrees(ctx, tk.ID)
 	if out, _ := gitops.Run(ctx, wts[0].Path, "status", "--porcelain"); out != "" {
 		t.Fatalf("uncommitted work left: %s", out)
 	}
+	see("runner C", "finishes the task", fmt.Sprintf("status %s; lease released; worktree clean", got.Status))
 	_ = s.Close()
+	checkReplay(t, rec)
+}
+
+// checkReplay compares what the run observed with testdata/crash-replay.json.
+func checkReplay(t *testing.T, rec crashReplay) {
+	t.Helper()
+	b, err := json.MarshalIndent(rec, "", "  ")
+	if err != nil {
+		t.Fatal(err)
+	}
+	b = append(b, '\n')
+	golden := filepath.Join("testdata", "crash-replay.json")
+	if *updateReplay {
+		if err := os.MkdirAll("testdata", 0o755); err != nil {
+			t.Fatal(err)
+		}
+		if err := os.WriteFile(golden, b, 0o644); err != nil {
+			t.Fatal(err)
+		}
+		return
+	}
+	want, err := os.ReadFile(golden)
+	if err != nil {
+		t.Fatalf("%v (create it with -update-replay)", err)
+	}
+	if !bytes.Equal(want, b) {
+		t.Fatalf("this run observed something other than %s (rerun with -update-replay if the change is intended):\n%s", golden, b)
+	}
 }
 
 // TestModelServerOutageDoesNotBurnAttempts: the model server dies during an

internal/orchestrator/testdata/crash-replay.json

@@ -0,0 +1,61 @@
+{
+  "test": "TestSIGKILLResume",
+  "about": "A runner process is killed with SIGKILL in the middle of an agent turn; a second runner is refused, a third takes over the stale lease and finishes the task. Each step is what the test observed.",
+  "steps": [
+    {
+      "actor": "test",
+      "action": "creates the task",
+      "observed": "status active, phase setup"
+    },
+    {
+      "actor": "runner A",
+      "action": "runs as a separate process (lease owner \"child\"); its agent writes a file and stays mid-turn",
+      "observed": "ledger-service/internal/consumer/balance_test.go in the worktree: true"
+    },
+    {
+      "actor": "test",
+      "action": "kills runner A with SIGKILL: no signal handler, no cleanup",
+      "observed": "signal: killed"
+    },
+    {
+      "actor": "ledger",
+      "action": "what the dead runner left behind",
+      "observed": "lease held by \"child\"; task active in phase implementing; attempt 1 active"
+    },
+    {
+      "actor": "runner B",
+      "action": "tries to run the task while the lease is fresh",
+      "observed": "refused: task is being run by another process (child, last heartbeat \u003ctime\u003e)"
+    },
+    {
+      "actor": "test",
+      "action": "waits for the lease to go stale",
+      "observed": "no heartbeat for longer than 50ms (2m0s in production, shortened for the test)"
+    },
+    {
+      "actor": "runner C",
+      "action": "takes over the stale lease; its agent is told",
+      "observed": "You are resuming a task after a context reset. The state below was rebuilt from the task ledger, git and verification results; trust it over memory."
+    },
+    {
+      "actor": "runner C",
+      "action": "its agent looks for the killed attempt's file",
+      "observed": "ledger-service/internal/consumer/balance_test.go still in the worktree: true"
+    },
+    {
+      "actor": "ledger",
+      "action": "reconciles the orphaned attempt",
+      "observed": "attempt 1 rejected: interrupted: the previous run stopped without finishing this attempt"
+    },
+    {
+      "actor": "ledger",
+      "action": "the task's event log, in order",
+      "observed": "task.created → agent.session → task.contract_failed → attempt.started → context.pack → task.recovered → agent.session → task.contract_failed → attempt.started → context.pack → agent.turn → verify.result → verify.result → frontier.triggered → verify.evidence → verify.evidence_requested → attempt.started → context.pack → agent.turn → verify.result → verify.result → verify.evidence → task.completed"
+    },
+    {
+      "actor": "runner C",
+      "action": "finishes the task",
+      "observed": "status completed; lease released; worktree clean"
+    }
+  ]
+}