corago/codex-keep/reconcile_test.go

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.human(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")
	}
}