Reconcile goal recovery with human intent and model state transitions

codex-keep/reconcile_test.go

@@ -0,0 +1,477 @@
+package main
+
+import (
+	"encoding/json"
+	"io"
+	"testing"
+	"time"
+)
+
+func TestMissedStart(t *testing.T) {
+	w := activeWatch()
+	w.observe(testTurn("completed"))
+	w.update(testGoal(blocked))
+	failure := testTurn("failed")
+	failure.ID = "turn-2"
+	w.observe(failure)
+	if !w.eligible() {
+		t.Fatal("API failure after a missed turn start was not recovered")
+	}
+}
+
+func TestStopInFlight(t *testing.T) {
+	for _, status := range []goalStatus{paused, "complete", "budgetLimited", "usageLimited"} {
+		t.Run(string(status), func(t *testing.T) {
+			f := newFake(t, nil)
+			f.fail()
+			w := failedWatch()
+			w.Due = time.Now().Add(-time.Second)
+			f.op.store.Watches["thread-1"] = w
+			f.before = func(m message) {
+				if m.Method != "thread/goal/set" {
+					return
+				}
+				f.before = nil
+				f.goal = testGoal(status)
+				f.notify("thread/goal/updated", map[string]any{"threadId": "thread-1", "goal": f.goal})
+			}
+			for range 3 {
+				if err := f.reconcile(); err != nil {
+					t.Fatal(err)
+				}
+			}
+			if f.goal.Status != status || w.Armed || w.eligible() || !w.Due.IsZero() {
+				t.Fatalf("human stop did not win reconciliation: target=%+v watch=%+v", f.goal, w)
+			}
+		})
+	}
+}
+
+func TestStopLostReply(t *testing.T) {
+	f := newFake(t, nil)
+	f.fail()
+	f.op.attached["thread-1"] = true
+	f.lostReply = true
+	w := failedWatch()
+	w.Due = time.Now().Add(-time.Second)
+	f.op.store.Watches["thread-1"] = w
+	f.before = func(m message) {
+		if m.Method != "thread/goal/set" {
+			return
+		}
+		f.before = nil
+		f.goal = testGoal(paused)
+		f.notify("thread/goal/updated", map[string]any{"threadId": "thread-1", "goal": f.goal})
+	}
+	if err := f.reconcile(); err != io.EOF {
+		t.Fatalf("error = %v", err)
+	}
+	path := f.op.store.path
+	closeStore(t, f.op.store)
+	s, err := openStore(path)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer closeStore(t, s)
+	next := newFake(t, s)
+	next.goal, next.turn = f.goal, f.turn
+	for range 2 {
+		if err := next.reconcile(); err != nil {
+			t.Fatal(err)
+		}
+	}
+	if next.goal.Status != paused || s.watch("thread-1").Armed {
+		t.Fatalf("lost human cancellation across restart: target=%+v watch=%+v", next.goal, s.watch("thread-1"))
+	}
+}
+
+func TestBlockStamp(t *testing.T) {
+	w := failedWatch()
+	w.Due = time.Now().Add(time.Second)
+	due := w.Due
+	g := testGoal(blocked)
+	g.UpdatedAt++
+	w.update(g)
+	if !w.eligible() || w.Due != due {
+		t.Fatal("accounting timestamp cancelled recovery")
+	}
+}
+
+func TestStaleActiveEvent(t *testing.T) {
+	f := newFake(t, nil)
+	f.goal = testGoal(paused)
+	w := activeWatch()
+	w.update(f.goal)
+	f.op.store.Watches["thread-1"] = w
+	raw, err := json.Marshal(params{ThreadID: "thread-1", Goal: testGoal(active)})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := f.op.event(message{Method: "thread/goal/updated", Params: raw}); err != nil {
+		t.Fatal(err)
+	}
+	if w.Armed || w.Goal.Status != paused {
+		t.Fatal("stale active echo revoked cancellation")
+	}
+	if err := f.reconcile(); err != nil {
+		t.Fatal(err)
+	}
+	if f.goal.Status != paused || w.Armed {
+		t.Fatal("stale active echo changed desired state")
+	}
+}
+
+func TestResumeMissed(t *testing.T) {
+	w := activeWatch()
+	w.observe(testTurn("interrupted"))
+	w.update(testGoal(blocked))
+	failure := testTurn("failed")
+	failure.ID = "turn-2"
+	w.observe(failure)
+	if !w.eligible() {
+		t.Fatal("new API-failed turn did not restore known goal authorization")
+	}
+}
+
+func TestRepairIntent(t *testing.T) {
+	for _, final := range []goalStatus{"complete", active} {
+		t.Run(string(final), func(t *testing.T) {
+			f := newFake(t, nil)
+			f.fail()
+			w := failedWatch()
+			w.Due = time.Now().Add(-time.Second)
+			f.op.store.Watches["thread-1"] = w
+			phase := 0
+			f.before = func(m message) {
+				if m.Method != "thread/goal/set" {
+					return
+				}
+				var p setParams
+				if err := json.Unmarshal(m.Params, &p); err != nil {
+					t.Fatal(err)
+				}
+				switch phase {
+				case 0:
+					if p.Status != active {
+						t.Fatal("missing initial resume")
+					}
+					phase++
+					f.goal = testGoal(paused)
+				case 1:
+					if p.Status != paused {
+						return
+					}
+					phase++
+					f.goal = testGoal(final)
+				default:
+					return
+				}
+				if phase == 2 && final == active {
+					f.human(f.goal)
+				} else {
+					f.notify("thread/goal/updated", map[string]any{"threadId": "thread-1", "goal": f.goal})
+				}
+			}
+			for range 6 {
+				if err := f.reconcile(); err != nil {
+					t.Fatal(err)
+				}
+			}
+			if phase != 2 || f.goal.Status != final {
+				t.Fatalf("latest observed intent lost: phase=%d target=%+v watch=%+v", phase, f.goal, w)
+			}
+		})
+	}
+}
+
+func TestClearInFlight(t *testing.T) {
+	f := newFake(t, nil)
+	f.fail()
+	w := failedWatch()
+	w.Due = time.Now().Add(-time.Second)
+	f.op.store.Watches["thread-1"] = w
+	f.before = func(m message) {
+		if m.Method != "thread/goal/set" {
+			return
+		}
+		f.before = nil
+		f.goal = nil
+		f.notify("thread/goal/cleared", map[string]any{"threadId": "thread-1"})
+	}
+	for range 3 {
+		if err := f.reconcile(); err != nil {
+			t.Fatal(err)
+		}
+	}
+	if f.goal != nil || w.Goal != nil || w.eligible() {
+		t.Fatal("cleared goal was recreated")
+	}
+}
+
+func TestSwapInFlight(t *testing.T) {
+	f := newFake(t, nil)
+	f.fail()
+	w := failedWatch()
+	w.Due = time.Now().Add(-time.Second)
+	f.op.store.Watches["thread-1"] = w
+	f.before = func(m message) {
+		if m.Method != "thread/goal/set" {
+			return
+		}
+		f.before = nil
+		f.goal = &goal{Objective: "Replacement", CreatedAt: 2, UpdatedAt: 2, Status: blocked}
+		f.notify("thread/goal/updated", map[string]any{"threadId": "thread-1", "goal": f.goal})
+	}
+	for range 4 {
+		if err := f.reconcile(); err != nil {
+			t.Fatal(err)
+		}
+	}
+	if f.goal.Objective != "Replacement" || f.goal.Status != blocked || w.eligible() {
+		t.Fatalf("replacement was changed: target=%+v watch=%+v", f.goal, w)
+	}
+}
+
+func TestBusyRecovery(t *testing.T) {
+	f := newFake(t, nil)
+	f.fail()
+	f.status = "active"
+	w := failedWatch()
+	w.Due = time.Now().Add(-time.Second)
+	f.op.store.Watches["thread-1"] = w
+	if err := f.reconcile(); err != nil {
+		t.Fatal(err)
+	}
+	f.status = "systemError"
+	f.turn.ID = "turn-2"
+	if err := f.reconcile(); err != nil {
+		t.Fatal(err)
+	}
+	if !w.eligible() || w.Due.IsZero() {
+		t.Fatal("busy observation lost goal authorization")
+	}
+}
+
+func TestLateBlock(t *testing.T) {
+	f := newFake(t, nil)
+	f.fail()
+	w := failedWatch()
+	w.Due = time.Now().Add(-time.Second)
+	f.op.store.Watches["thread-1"] = w
+	f.before = func(m message) {
+		if m.Method != "thread/goal/set" {
+			return
+		}
+		f.before = nil
+		f.goal = testGoal(paused)
+		f.notify("thread/goal/updated", map[string]any{"threadId": "thread-1", "goal": f.goal})
+		f.notify("thread/goal/updated", map[string]any{"threadId": "thread-1", "goal": testGoal(blocked)})
+	}
+	for range 4 {
+		if err := f.reconcile(); err != nil {
+			t.Fatal(err)
+		}
+	}
+	if f.goal.Status != paused || w.Armed {
+		t.Fatalf("server block replaced human stop: %+v", w)
+	}
+}
+
+func TestNewGoalNotice(t *testing.T) {
+	f := newFake(t, nil)
+	f.goal = &goal{Objective: "Replacement", CreatedAt: 2, UpdatedAt: 2, Status: active}
+	raw, err := json.Marshal(params{ThreadID: "thread-1", Goal: f.goal})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := f.op.event(message{Method: "thread/goal/updated", Params: raw}); err != nil {
+		t.Fatal(err)
+	}
+	f.goal.Status = blocked
+	f.turn = testTurn("failed")
+	if err := f.reconcile(); err != nil {
+		t.Fatal(err)
+	}
+	if !f.op.store.watch("thread-1").eligible() {
+		t.Fatal("observed active replacement lost authorization before polling")
+	}
+}
+
+func TestSameTurnResume(t *testing.T) {
+	for _, failedFirst := range []bool{false, true} {
+		w := activeWatch()
+		f := newFake(t, nil)
+		f.op.store.Watches["thread-1"] = w
+		if err := f.reconcile(); err != nil {
+			t.Fatal(err)
+		}
+		w.update(testGoal("budgetLimited"))
+		f.human(testGoal(active))
+		if err := f.reconcile(); err != nil {
+			t.Fatal(err)
+		}
+		if failedFirst {
+			w.observe(testTurn("failed"))
+		}
+		w.update(testGoal(blocked))
+		if !failedFirst {
+			w.observe(testTurn("failed"))
+		}
+		if !w.eligible() {
+			t.Fatalf("resumed turn lost recovery, failedFirst=%v", failedFirst)
+		}
+	}
+}
+
+func TestReconnectEcho(t *testing.T) {
+	f := newFake(t, nil)
+	f.fail()
+	w := failedWatch()
+	w.Due = time.Now().Add(-time.Second)
+	f.op.store.Watches["thread-1"] = w
+	if err := f.reconcile(); err != nil {
+		t.Fatal(err)
+	}
+	next := newFake(t, f.op.store)
+	next.fail()
+	next.turn.ID = "turn-2"
+	if err := next.reconcile(); err != nil {
+		t.Fatal(err)
+	}
+	w.Due = time.Now().Add(-time.Second)
+	if err := next.reconcile(); err != nil {
+		t.Fatal(err)
+	}
+	if next.count("thread/goal/set") != 1 {
+		t.Fatal("old connection echo blocked recovery after reconnection")
+	}
+}
+
+func TestAccountStop(t *testing.T) {
+	f := newFake(t, nil)
+	f.fail()
+	w := failedWatch()
+	w.Due = time.Now().Add(-time.Second)
+	f.op.store.Watches["thread-1"] = w
+	f.before = func(m message) {
+		if m.Method != "thread/goal/set" {
+			return
+		}
+		f.before = nil
+		f.goal = testGoal(paused)
+		f.notify("thread/goal/updated", map[string]any{"threadId": "thread-1", "goal": f.goal})
+	}
+	if err := f.reconcile(); err != nil {
+		t.Fatal(err)
+	}
+	f.before = func(m message) {
+		if m.Method != "thread/loaded/list" {
+			return
+		}
+		f.before = nil
+		g := *f.goal
+		g.UpdatedAt++
+		f.goal = &g
+		f.notify("thread/goal/updated", map[string]any{"threadId": "thread-1", "goal": f.goal})
+	}
+	for range 4 {
+		if err := f.reconcile(); err != nil {
+			t.Fatal(err)
+		}
+	}
+	if f.goal.Status != paused || w.eligible() {
+		t.Fatal("accounting update revoked human cancellation")
+	}
+}
+
+func TestDelayedWrite(t *testing.T) {
+	f := newFake(t, nil)
+	w := failedWatch()
+	w.Pending = true
+	f.op.store.Watches["thread-1"] = w
+	var err error
+	w.Source, err = openIntent(f.path)
+	if err != nil {
+		t.Fatal(err)
+	}
+	f.goal = testGoal(paused)
+	f.human(f.goal)
+	for range 2 {
+		if err := f.reconcile(); err != nil {
+			t.Fatal(err)
+		}
+	}
+	f.goal = testGoal(active)
+	f.notify("thread/goal/updated", map[string]any{"threadId": "thread-1", "goal": f.goal})
+	for range 3 {
+		if err := f.reconcile(); err != nil {
+			t.Fatal(err)
+		}
+	}
+	if f.goal.Status != paused || w.Armed {
+		t.Fatal("delayed activation revoked settled cancellation")
+	}
+}
+
+func TestQueuedResume(t *testing.T) {
+	for _, bound := range []bool{false, true} {
+		t.Run(map[bool]string{false: "startup", true: "bound"}[bound], func(t *testing.T) {
+			f := newFake(t, nil)
+			f.op.store.Watches["thread-1"] = activeWatch()
+			if bound {
+				if err := f.reconcile(); err != nil {
+					t.Fatal(err)
+				}
+			}
+			f.human(testGoal(paused))
+			f.human(testGoal(active))
+			for range 3 {
+				if err := f.reconcile(); err != nil {
+					t.Fatal(err)
+				}
+			}
+			if f.goal.Status != active {
+				t.Fatal("queued resumption was discarded by an older pause notification")
+			}
+		})
+	}
+}
+
+func TestBarrierChange(t *testing.T) {
+	f := newFake(t, nil)
+	f.op.store.Watches["thread-1"] = activeWatch()
+	if err := f.reconcile(); err != nil {
+		t.Fatal(err)
+	}
+	f.before = func(m message) {
+		if m.Method != "thread/resume" {
+			return
+		}
+		f.before = nil
+		f.goal = testGoal("complete")
+		f.human(f.goal)
+	}
+	if err := f.reconcile(); err != nil {
+		t.Fatal(err)
+	}
+	for range 2 {
+		if err := f.reconcile(); err != nil {
+			t.Fatal(err)
+		}
+	}
+	if f.count("thread/goal/set") != 0 {
+		t.Fatal("barrier notification left an invalid goal snapshot in use")
+	}
+}
+
+func TestNoGoalPath(t *testing.T) {
+	f := newFake(t, nil)
+	f.goal = nil
+	f.path = ""
+	if err := f.reconcile(); err != nil {
+		t.Fatal(err)
+	}
+	if f.count("thread/goal/set") != 0 {
+		t.Fatal("mutated a thread without a goal")
+	}
+}