diff --git a/cmd/codeaf/factory_person_stop.go b/cmd/codeaf/factory_person_stop.go index d644600f0e..1196ace138 100644 --- a/cmd/codeaf/factory_person_stop.go +++ b/cmd/codeaf/factory_person_stop.go @@ -17,3 +17,22 @@ func (d seamRunDoor) AutomaticRefusal(conversation string) string { } return "" } + +// WatchedRefusal is AutomaticRefusal for the watcher that looks ten times a +// second for as long as a job's chat is open. It costs one look at the floor's +// folder, not a read of every job ([store.Store.TalkStop]). What grants work +// still asks AutomaticRefusal. +func (d seamRunDoor) WatchedRefusal(conversation string) string { + want := sameFileKey(conversation) + if want == "" || d.st == nil { + return "" + } + manager, stopped, err := d.st.TalkStop(want) + if err != nil { + return "job could not be read; automatic work stopped" + } + if manager && stopped { + return factory.PersonStopReason + } + return "" +} diff --git a/cmd/codeaf/factory_person_stop_watch_test.go b/cmd/codeaf/factory_person_stop_watch_test.go new file mode 100644 index 0000000000..06b160cb3e --- /dev/null +++ b/cmd/codeaf/factory_person_stop_watch_test.go @@ -0,0 +1,422 @@ +package main + +import ( + "fmt" + "os" + "path/filepath" + "strings" + "sync" + "testing" + "time" + + "github.com/Agent-Field/codeaf/internal/factory" + factoryrun "github.com/Agent-Field/codeaf/internal/factory/run" + "github.com/Agent-Field/codeaf/internal/factory/store" +) + +// watchFloor is a floor of jobs, the first managers of them with a chat each. +func watchFloor(t *testing.T, jobs, managers int) (*store.Store, seamRunDoor, []string, []int) { + t.Helper() + st, err := store.Open(filepath.Join(t.TempDir(), "factory")) + if err != nil { + t.Fatal(err) + } + talks := t.TempDir() + var chats []string + var ids []int + for i := range jobs { + it := factory.Item{Repo: "acme/api", Title: fmt.Sprintf("job %d", i), State: factory.StateNew, Cap: 1} + if i < managers { + it.Talk = filepath.Join(talks, fmt.Sprintf("chat-%d", i), "transcript.jsonl") + chats = append(chats, it.Talk) + } + made, err := st.Create(it) + if err != nil { + t.Fatal(err) + } + ids = append(ids, made.ID) + } + return st, seamRunDoor{st: st}, chats, ids +} + +// settleFloor moves the floor's folder time into the past, as a floor nobody has +// written to for a while, so a look may trust the folder. +func settleFloor(t *testing.T, st *store.Store) { + t.Helper() + old := time.Now().Add(-time.Hour) + if err := os.Chtimes(st.Root(), old, old); err != nil { + t.Fatal(err) + } +} + +func setStopped(t *testing.T, st *store.Store, id int, stopped bool) { + t.Helper() + if err := st.Annotate(id, func(it *factory.Item) error { + if stopped { + factory.MarkPersonStopped(it) + } else { + factory.ClearPersonStop(it) + } + return nil + }); err != nil { + t.Fatal(err) + } +} + +// THE WATCHER RUNS TEN TIMES A SECOND FOR EVERY OPEN JOB CHAT. On a floor where +// nothing changes, it must read no job document at all, however many jobs and +// chats there are, and a chat that manages nothing is no dearer. +func TestWatchingManyJobChatsReadsNoDocumentWhileNothingChanges(t *testing.T) { + st, door, chats, _ := watchFloor(t, 300, 20) + stranger := filepath.Join(t.TempDir(), "transcript.jsonl") + look := func() { + t.Helper() + for _, chat := range chats { + if why := door.WatchedRefusal(chat); why != "" { + t.Fatalf("an unstopped job refused its chat: %q", why) + } + } + if why := door.WatchedRefusal(stranger); why != "" { + t.Fatalf("a chat that manages nothing was refused: %q", why) + } + } + look() + before := st.DocumentReads() + const ticks = 50 + for range ticks { + look() + } + if reads := st.DocumentReads() - before; reads != 0 { + t.Fatalf("%d looks over %d chats and 300 jobs read %d job documents, want none", ticks, len(chats)+1, reads) + } + + settleFloor(t, st) + look() + before = st.DocumentReads() + for range ticks { + look() + } + if reads := st.DocumentReads() - before; reads != 0 { + t.Fatalf("on a settled floor, %d looks read %d job documents, want none", ticks, reads) + } +} + +// One job changing costs that job's document once, not the floor's. +func TestOneJobChangingReadsOneDocument(t *testing.T) { + st, door, chats, ids := watchFloor(t, 300, 5) + door.WatchedRefusal(chats[0]) + before := st.DocumentReads() + setStopped(t, st, ids[3], true) + writes := st.DocumentReads() - before // what the write itself reads + for _, chat := range chats { + door.WatchedRefusal(chat) + } + if reads := st.DocumentReads() - before - writes; reads != 1 { + t.Fatalf("one changed job cost %d document reads, want 1", reads) + } +} + +// NO STALE PASS. The very next look after any write gives the answer a full +// read of the floor gives, whether the floor was busy or long quiet, whoever +// wrote, and for every way the answer can change. +func TestTheNextLookSeesEveryChange(t *testing.T) { + for _, quiet := range []bool{false, true} { + t.Run(fmt.Sprintf("quiet=%v", quiet), func(t *testing.T) { + st, door, chats, ids := watchFloor(t, 40, 3) + // The other writer: another process's store on the same folder. + other, err := store.Open(st.Root()) + if err != nil { + t.Fatal(err) + } + prepare := func() { + door.WatchedRefusal(chats[0]) + if quiet { + settleFloor(t, st) + door.WatchedRefusal(chats[0]) + } + } + agrees := func(label, chat, want string) { + t.Helper() + if got := door.WatchedRefusal(chat); got != want { + t.Fatalf("%s: the next look said %q, want %q", label, got, want) + } + if fresh := door.AutomaticRefusal(chat); fresh != want { + t.Fatalf("%s: a full read said %q, want %q", label, fresh, want) + } + } + for round := range 25 { + prepare() + setStopped(t, st, ids[0], true) + agrees(fmt.Sprintf("round %d stop", round), chats[0], factory.PersonStopReason) + prepare() + setStopped(t, other, ids[0], false) + agrees(fmt.Sprintf("round %d start", round), chats[0], "") + } + + // A chat is handed to a job: the same chat now watches that job. + fresh := filepath.Join(t.TempDir(), "transcript.jsonl") + agrees("before the handover", fresh, "") + setStopped(t, st, ids[20], true) + prepare() + if err := st.Annotate(ids[20], func(it *factory.Item) error { it.Talk = fresh; return nil }); err != nil { + t.Fatal(err) + } + agrees("the job gained this chat", fresh, factory.PersonStopReason) + prepare() + if err := st.Annotate(ids[20], func(it *factory.Item) error { it.Talk = ""; return nil }); err != nil { + t.Fatal(err) + } + agrees("the job dropped this chat", fresh, "") + + // A new job is created already stopped, and a job goes away. + prepare() + talk := filepath.Join(t.TempDir(), "transcript.jsonl") + stopped := true + made, err := st.Create(factory.Item{Repo: "acme/api", Title: "new", State: factory.StateNew, Cap: 1, Talk: talk, PersonStopped: &stopped}) + if err != nil { + t.Fatal(err) + } + agrees("a new stopped job", talk, factory.PersonStopReason) + prepare() + if err := os.Remove(st.ItemPath(made.ID)); err != nil { + t.Fatal(err) + } + agrees("the job is gone", talk, "") + }) + } +} + +// The budget's own route keeps its answer: a budget changed mid-watch is the +// next look's refusal, and raising it is the next look's pass, alongside the Stop. +func TestTheNextLookSeesABudgetChange(t *testing.T) { + st, door, chats, ids := watchFloor(t, 40, 1) + chat := jobChatDoorOf(st).Chat(chats[0]) + if chat == nil { + t.Fatal("the job's chat was not found") + } + settleFloor(t, st) + if chat.Refusal() != "" || door.WatchedRefusal(chats[0]) != "" { + t.Fatal("an unspent, unstopped job was refused") + } + if err := st.Annotate(ids[0], func(it *factory.Item) error { it.ChatSpent = 1; return nil }); err != nil { + t.Fatal(err) + } + if want := fmt.Sprintf(factoryrun.ChatBudgetReached, "$1", "$1"); chat.Refusal() != want { + t.Fatalf("after the spend the budget said %q, want %q", chat.Refusal(), want) + } + settleFloor(t, st) + if err := st.Annotate(ids[0], func(it *factory.Item) error { it.Cap = 5; return nil }); err != nil { + t.Fatal(err) + } + if why := chat.Refusal(); why != "" { + t.Fatalf("after the raise the budget still said %q", why) + } +} + +// A document that can't be read is skipped, as a full read skips it, and is +// read again once it is whole, even with the folder's time unchanged. +func TestADocumentReadableAgainIsSeen(t *testing.T) { + st, door, chats, ids := watchFloor(t, 5, 1) + setStopped(t, st, ids[0], true) + path := st.ItemPath(ids[0]) + kept, err := os.ReadFile(path) + if err != nil { + t.Fatal(err) + } + if err := os.WriteFile(path, []byte("{torn"), 0o600); err != nil { + t.Fatal(err) + } + settleFloor(t, st) + if got, want := door.WatchedRefusal(chats[0]), door.AutomaticRefusal(chats[0]); got != want { + t.Fatalf("over a torn document the watcher said %q, a full read %q", got, want) + } + if err := os.WriteFile(path, kept, 0o600); err != nil { + t.Fatal(err) + } + if got := door.WatchedRefusal(chats[0]); got != factory.PersonStopReason { + t.Fatalf("the whole document again said %q", got) + } +} + +// A write that keeps the document's size and time is still a new file, and the +// next look sees it through both routes: the Stop, and the budget. +func TestAWriteKeepingTheFilesTimeIsSeen(t *testing.T) { + st, door, chats, ids := watchFloor(t, 20, 1) + chat := jobChatDoorOf(st).Chat(chats[0]) + path := st.ItemPath(ids[0]) + before, err := os.Stat(path) + if err != nil { + t.Fatal(err) + } + // Same length, same time, new file: what a same-instant rename leaves. + swap := func(old, next string) { + t.Helper() + data, err := os.ReadFile(path) + if err != nil { + t.Fatal(err) + } + changed := strings.Replace(string(data), old, next, 1) + if changed == string(data) || len(changed) != len(data) { + t.Fatalf("no same-length change from %q to %q in %s", old, next, data) + } + temp := path + ".swap" + if err := os.WriteFile(temp, []byte(changed), 0o600); err != nil { + t.Fatal(err) + } + if err := os.Chtimes(temp, before.ModTime(), before.ModTime()); err != nil { + t.Fatal(err) + } + if err := os.Rename(temp, path); err != nil { + t.Fatal(err) + } + } + setStopped(t, st, ids[0], false) + if before, err = os.Stat(path); err != nil { + t.Fatal(err) + } + settleFloor(t, st) + if door.WatchedRefusal(chats[0]) != "" || chat.Refusal() != "" { + t.Fatal("a job not stopped was refused") + } + swap(`"PersonStopped": false`, `"PersonStopped": true `) + if got := door.WatchedRefusal(chats[0]); got != factory.PersonStopReason { + t.Fatalf("a Stop written with the file's time kept: the next look said %q", got) + } + swap(`"PersonStopped": true `, `"PersonStopped": false`) + if got := door.WatchedRefusal(chats[0]); got != "" { + t.Fatalf("a Start written with the file's time kept: the next look said %q", got) + } + if err := st.Annotate(ids[0], func(it *factory.Item) error { it.ChatSpent = 2; it.Cap = 9; return nil }); err != nil { + t.Fatal(err) + } + if before, err = os.Stat(path); err != nil { + t.Fatal(err) + } + settleFloor(t, st) + if chat.Refusal() != "" { + t.Fatal("a job within its budget was refused") + } + swap(`"Cap": 9`, `"Cap": 1`) + if chat.Refusal() == "" { + t.Fatal("a budget lowered with the file's time kept: the next look still passed") + } +} + +// The floor going away is an answer, not a crash, and coming back is seen. +func TestAFloorThatGoesAwayIsNoManagerAndNotACrash(t *testing.T) { + st, door, chats, ids := watchFloor(t, 3, 1) + setStopped(t, st, ids[0], true) + if got := door.WatchedRefusal(chats[0]); got != factory.PersonStopReason { + t.Fatalf("before: %q", got) + } + aside := st.Root() + ".aside" + if err := os.Rename(st.Root(), aside); err != nil { + t.Fatal(err) + } + if got := door.WatchedRefusal(chats[0]); got != "" { + t.Fatalf("with no floor, the watcher said %q", got) + } + if err := os.Rename(aside, st.Root()); err != nil { + t.Fatal(err) + } + if got := door.WatchedRefusal(chats[0]); got != factory.PersonStopReason { + t.Fatalf("with the floor back, the watcher said %q", got) + } +} + +// A write through this store is seen on the next look even if the folder's time +// comes out identical to the one the last sweep trusted. +func TestAWriteHereIsSeenWhateverTheFolderTimeSays(t *testing.T) { + st, door, chats, ids := watchFloor(t, 3, 1) + setStopped(t, st, ids[0], true) + settleFloor(t, st) + if got := door.WatchedRefusal(chats[0]); got != factory.PersonStopReason { + t.Fatalf("before: %q", got) + } + settled, err := os.Stat(st.Root()) + if err != nil { + t.Fatal(err) + } + setStopped(t, st, ids[0], false) + if err := os.Chtimes(st.Root(), settled.ModTime(), settled.ModTime()); err != nil { + t.Fatal(err) + } + if got := door.WatchedRefusal(chats[0]); got != "" { + t.Fatalf("after a person's Start here, the watcher still said %q", got) + } +} + +// A link moved under a job's chat path changes no document, and a sweep sees it. +func TestALinkRetargetedUnderAChatIsSeenWithinASweep(t *testing.T) { + st, door, _, _ := watchFloor(t, 3, 0) + base := t.TempDir() + here, there := filepath.Join(base, "here"), filepath.Join(base, "there") + for _, dir := range []string{here, there} { + if err := os.MkdirAll(dir, 0o700); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(filepath.Join(dir, "transcript.jsonl"), nil, 0o600); err != nil { + t.Fatal(err) + } + } + link := filepath.Join(base, "link") + if err := os.Symlink(here, link); err != nil { + t.Fatal(err) + } + stopped := true + if _, err := st.Create(factory.Item{Repo: "acme/api", Title: "linked", State: factory.StateNew, Cap: 1, Talk: filepath.Join(link, "transcript.jsonl"), PersonStopped: &stopped}); err != nil { + t.Fatal(err) + } + settleFloor(t, st) + moved := filepath.Join(there, "transcript.jsonl") + if got := door.WatchedRefusal(moved); got != "" { + t.Fatalf("before the link moved, the other chat was refused: %q", got) + } + if err := os.Remove(link); err != nil { + t.Fatal(err) + } + if err := os.Symlink(there, link); err != nil { + t.Fatal(err) + } + time.Sleep(1100 * time.Millisecond) + if got := door.WatchedRefusal(moved); got != factory.PersonStopReason { + t.Fatalf("after the link moved and a sweep, the watcher said %q", got) + } +} + +// Many chats look at once while a job changes; every look after the write ends +// sees it. The race detector is the other half of this test. +func TestLooksDuringAWriteEndOnTheWrite(t *testing.T) { + st, door, chats, ids := watchFloor(t, 60, 8) + var lookers sync.WaitGroup + done := make(chan struct{}) + for _, chat := range chats { + lookers.Add(1) + go func() { + defer lookers.Done() + for { + select { + case <-done: + return + default: + door.WatchedRefusal(chat) + } + } + }() + } + for round := range 40 { + stopped := round%2 == 0 + setStopped(t, st, ids[0], stopped) + want := "" + if stopped { + want = factory.PersonStopReason + } + if got := door.WatchedRefusal(chats[0]); got != want { + close(done) + lookers.Wait() + t.Fatalf("round %d: the next look said %q, want %q", round, got, want) + } + } + close(done) + lookers.Wait() +} diff --git a/cmd/codeaf/factory_shape.go b/cmd/codeaf/factory_shape.go index bd498d8a79..466faeccc0 100644 --- a/cmd/codeaf/factory_shape.go +++ b/cmd/codeaf/factory_shape.go @@ -4,7 +4,6 @@ import ( "context" "errors" "fmt" - "path/filepath" "strings" "time" @@ -393,16 +392,7 @@ func (d seamRunDoor) managed(conversation string) (factory.Item, error) { } // sameFileKey is a path as one spelling, links resolved. -func sameFileKey(path string) string { - path = strings.TrimSpace(path) - if path == "" { - return "" - } - if real, err := filepath.EvalSymlinks(path); err == nil { - path = real - } - return filepath.Clean(path) -} +func sameFileKey(path string) string { return store.FileKey(path) } // ManagerStatus reads the durable job and lease without opening another agent // or making a provider request. The saved transcript is not execution state. diff --git a/docs/changes/unreleased/2358-memory-off-idle.md b/docs/changes/unreleased/2358-memory-off-idle.md new file mode 100644 index 0000000000..d5eec4259e --- /dev/null +++ b/docs/changes/unreleased/2358-memory-off-idle.md @@ -0,0 +1,8 @@ +--- +kind: fixed +title: With Memory off, nothing reads saved chats until the Memory page opens +pr: 2358 +surface: [engine] +invalidates: + - "With Memory off, a pass every two seconds listed saved chats and re-read them whenever one changed, to keep a count. Now the Knowledge offer asks for the count when it opens, and the last count stays. The offer itself appears as before." +--- diff --git a/docs/changes/unreleased/2358-stop-watch-cpu.md b/docs/changes/unreleased/2358-stop-watch-cpu.md new file mode 100644 index 0000000000..51f7782c77 --- /dev/null +++ b/docs/changes/unreleased/2358-stop-watch-cpu.md @@ -0,0 +1,8 @@ +--- +kind: fixed +title: An open job chat no longer reads every job ten times a second +pr: 2358 +surface: [engine] +invalidates: + - "Each open job chat re-read every job document ten times a second to learn whether a person had stopped its job, so CPU grew with chats times jobs while idle. It now looks at the folder and reads only documents that changed; a Stop is still seen on the next check." +--- diff --git a/docs/changes/unreleased/2361-floor-reads.md b/docs/changes/unreleased/2361-floor-reads.md new file mode 100644 index 0000000000..aef96fb5a7 --- /dev/null +++ b/docs/changes/unreleased/2361-floor-reads.md @@ -0,0 +1,18 @@ +--- +kind: fixed +title: Idle factory polls read only job documents that changed +pr: 2361 +surface: [engine] +invalidates: + - "Idle lease pulls, settlement and runner capacity checks decoded every job document on each look. They now reuse unchanged documents after checking every file's identity, size and modification time." +--- + +Each List result owns its mutable data. Repository migration and output retention +still run on each look. Writes from another store and replacements with unchanged +size and modification time are visible on the next look. Every document is read +again at least once a minute, so an in-place edit that keeps its size and time is +seen within that bound. + +Lease grants and execution gates still read the job under its lock. A Stop that +lands during selection permits only a thread-only lease for a pending reply, +without restarting job steps. diff --git a/internal/factory/lease/floor_reads_test.go b/internal/factory/lease/floor_reads_test.go new file mode 100644 index 0000000000..f3bc8cf42f --- /dev/null +++ b/internal/factory/lease/floor_reads_test.go @@ -0,0 +1,244 @@ +package lease_test + +import ( + "errors" + "os" + "strings" + "sync" + "testing" + + "github.com/Agent-Field/codeaf/internal/factory" + "github.com/Agent-Field/codeaf/internal/factory/lease" + "github.com/Agent-Field/codeaf/internal/factory/store" +) + +func TestIdleFloorReadsOnlyChangedJobs(t *testing.T) { + for _, route := range []string{"Pull", "OverLimit"} { + t.Run(route, func(t *testing.T) { + f := setup(t) + for range 300 { + f.add(t, factory.Item{State: factory.StateNew}) + } + look := func() { + t.Helper() + if route == "Pull" { + if _, err := f.desk.Pull(worker("idle", "sam")); !errors.Is(err, lease.ErrNoJob) { + t.Fatalf("Pull: %v", err) + } + } else if n, err := f.desk.OverLimit(worker("idle", "sam")); err != nil || n != 0 { + t.Fatalf("OverLimit: %d, %v", n, err) + } + } + look() + before := f.store.DocumentReads() + for range 3 { + look() + } + if n := f.store.DocumentReads() - before; n != 0 { + t.Fatalf("unchanged 300-job floor: read %d documents, want 0", n) + } + other, err := store.Open(f.store.Root()) + if err != nil { + t.Fatal(err) + } + if err := other.Update(150, func(it *factory.Item) error { it.Title = "changed elsewhere"; return nil }); err != nil { + t.Fatal(err) + } + before = f.store.DocumentReads() + look() + if n := f.store.DocumentReads() - before; n != 1 { + t.Fatalf("one changed job: read %d documents, want 1", n) + } + }) + } +} + +func TestPullAndCapacitySeeExternalFloorChanges(t *testing.T) { + for _, change := range []string{"stop", "start", "budget", "create", "delete", "take", "release", "rename"} { + t.Run(change, func(t *testing.T) { + f := setup(t) + in := factory.Item{State: factory.StateQueued, Taker: "sam", OverLimit: true, Cap: 10, + Autonomy: factory.AutonomyAuto, Stream: &factory.Stream{Spent: 2}, Recovery: &factory.StepRecovery{State: "queued"}} + if change == "start" || change == "create" { + in.State = factory.StateNew + } + if change == "release" { + in.Taker = "other" + } + it := f.add(t, in) + if _, err := f.desk.OverLimit(worker("idle", "sam")); err != nil { + t.Fatal(err) + } + other, err := store.Open(f.store.Root()) + if err != nil { + t.Fatal(err) + } + want := false + switch change { + case "create": + in.State = factory.StateQueued + it, err = other.Create(in) + want = true + case "delete": + err = os.Remove(other.ItemPath(it.ID)) + case "rename": + path := other.ItemPath(it.ID) + info, e := os.Stat(path) + if e != nil { + t.Fatal(e) + } + data, e := os.ReadFile(path) + if e != nil { + t.Fatal(e) + } + if strings.Count(string(data), `"sam"`) != 1 { + t.Fatal("expected one owner to replace") + } + tmp := path + ".replacement" + if e = os.WriteFile(tmp, []byte(strings.Replace(string(data), `"sam"`, `"pat"`, 1)), 0o600); e != nil { + t.Fatal(e) + } + if e = os.Chtimes(tmp, info.ModTime(), info.ModTime()); e != nil { + t.Fatal(e) + } + err = os.Rename(tmp, path) + default: + err = other.Update(it.ID, func(cur *factory.Item) error { + switch change { + case "stop": + factory.MarkPersonStopped(cur) + case "start": + factory.ClearPersonStop(cur) + cur.State = factory.StateQueued + want = true + case "budget": + cur.Cap = 1 + case "take": + cur.Taker = "other" + case "release": + cur.Taker = "" + want = true + } + return nil + }) + } + if err != nil { + t.Fatal(err) + } + w := worker("idle", "sam") + w.TakesUnassigned = map[string]bool{"acme/api": true} + n, err := f.desk.OverLimit(w) + if err != nil || (n == 1) != want || n > 1 { + t.Fatalf("capacity after %s: %d, %v; eligible=%v", change, n, err, want) + } + held, err := f.desk.Pull(w) + if want { + if err != nil || held.Job != it.ID { + t.Fatalf("next Pull missed %s: %+v, %v", change, held, err) + } + } else if !errors.Is(err, lease.ErrNoJob) { + t.Fatalf("next Pull admitted after %s: %+v, %v", change, held, err) + } + }) + } +} + +// The live output-policy seam runs after List has copied its document, before +// returning the candidate. Commit a second store's change there to force the +// race between candidate selection and the locked reservation without sleeps. +func TestPullRechecksLockedJobAfterListing(t *testing.T) { + for _, change := range []string{"stop", "stop with reply", "budget", "take"} { + t.Run(change, func(t *testing.T) { + f := setup(t) + in := recoveryJob() + if change == "budget" { + in.Recovery = &factory.StepRecovery{State: "queued"} + } + if change == "stop with reply" { + in.Talk = "chat" + in.Thread = []factory.Message{{State: "pending", Text: "explain"}} + } + it := f.add(t, in) + if _, err := f.store.List(); err != nil { + t.Fatal(err) + } + other, err := store.Open(f.store.Root()) + if err != nil { + t.Fatal(err) + } + var once sync.Once + f.store.SetOutputPolicy(func() (int, int) { + once.Do(func() { + if err := other.Update(it.ID, func(cur *factory.Item) error { + switch change { + case "stop", "stop with reply": + factory.MarkPersonStopped(cur) + case "budget": + cur.Cap = 1 + case "take": + cur.Assignee = "other" + } + return nil + }); err != nil { + t.Fatal(err) + } + }) + return 0, 0 + }) + held, err := f.desk.Pull(worker("idle", "sam")) + if change == "stop with reply" { + if err != nil || !held.ThreadOnly { + t.Fatalf("Stop during selection granted job steps: %+v, %v", held, err) + } + f.state(t, it.ID, factory.StateQueued) + } else if !errors.Is(err, lease.ErrNoJob) { + t.Fatalf("locked reservation ignored %s: %+v, %v", change, held, err) + } + }) + } +} + +func TestFloorLooksRaceExternalWrites(t *testing.T) { + f := setup(t) + it := f.add(t, factory.Item{State: factory.StateQueued, Assignee: "sam", OverLimit: true}) + other, err := store.Open(f.store.Root()) + if err != nil { + t.Fatal(err) + } + start := make(chan struct{}) + var wg sync.WaitGroup + for range 3 { + wg.Add(1) + go func() { + defer wg.Done() + <-start + for range 30 { + if _, err := f.desk.Pull(worker("stranger", "other")); !errors.Is(err, lease.ErrNoJob) { + t.Error(err) + } + if _, err := f.desk.OverLimit(worker("idle", "sam")); err != nil { + t.Error(err) + } + } + }() + } + close(start) + for i := range 30 { + stopped := i%2 == 0 + if err := other.Update(it.ID, func(cur *factory.Item) error { + if stopped { + factory.MarkPersonStopped(cur) + } else { + factory.ClearPersonStop(cur) + } + return nil + }); err != nil { + t.Fatal(err) + } + n, err := f.desk.OverLimit(worker("idle", "sam")) + if err != nil || (n == 0) != stopped { + t.Errorf("next look after write %d: capacity=%d, stopped=%v, err=%v", i, n, stopped, err) + } + } + wg.Wait() +} diff --git a/internal/factory/serve/floor_reads_test.go b/internal/factory/serve/floor_reads_test.go new file mode 100644 index 0000000000..1675ce96fb --- /dev/null +++ b/internal/factory/serve/floor_reads_test.go @@ -0,0 +1,141 @@ +package serve_test + +import ( + "net/http/httptest" + "sync" + "testing" + + "github.com/Agent-Field/codeaf/internal/factory" + "github.com/Agent-Field/codeaf/internal/factory/store" +) + +// runner/self crosses ServeHTTP's settle scan and the capacity scan on the +// same route the idle runner calls, without asking the UI to load the floor. +func TestIdleRequestsReadOnlyChangedJobs(t *testing.T) { + f := workerSetup(t) + t.Cleanup(f.handler.Close) + for range 300 { + if _, err := f.st.Create(factory.Item{Repo: "owner/repo", State: factory.StateNew}); err != nil { + t.Fatal(err) + } + } + runner := f.attach(t, "sam") + look := func() { + t.Helper() + r := httptest.NewRequest("GET", "/v1/runner/self", nil) + r.Header.Set("Authorization", "Bearer "+runner.Credential) + w := httptest.NewRecorder() + f.handler.ServeHTTP(w, r) + if w.Code != 200 { + t.Fatalf("runner/self: %d %s", w.Code, w.Body.String()) + } + } + look() + before := f.st.DocumentReads() + for range 3 { + look() + } + if n := f.st.DocumentReads() - before; n != 0 { + t.Fatalf("settle and capacity on unchanged 300-job floor: read %d documents, want 0", n) + } + other, err := store.Open(f.st.Root()) + if err != nil { + t.Fatal(err) + } + if err := other.Update(150, func(it *factory.Item) error { it.Title = "changed elsewhere"; return nil }); err != nil { + t.Fatal(err) + } + before = f.st.DocumentReads() + look() + if n := f.st.DocumentReads() - before; n != 1 { + t.Fatalf("settle and capacity after one change: read %d documents, want 1", n) + } +} + +func TestSettleSeesExternalStopStartAndBudget(t *testing.T) { + for _, change := range []string{"stop", "start", "budget", "create"} { + t.Run(change, func(t *testing.T) { + f := newV2Jobs(t) + in := autoProof() + in.Autonomy = factory.AutonomyManual + if change == "start" { + in.Autonomy = factory.AutonomyAuto + factory.MarkPersonStopped(&in) + } + if change == "budget" { + in.Autonomy, in.Cap = factory.AutonomyAuto, 1 + } + it := f.create(t, in) + f.read(t, it) // Warm List while publication is blocked. + other, err := store.Open(f.st.Root()) + if err != nil { + t.Fatal(err) + } + if change == "create" { + in = autoProof() + in.Repo = "acme/api" + it, err = other.Create(in) + } else { + err = other.Update(it.ID, func(cur *factory.Item) error { + cur.Autonomy = factory.AutonomyAuto + switch change { + case "stop": + factory.MarkPersonStopped(cur) + case "start": + factory.ClearPersonStop(cur) + case "budget": + cur.Cap = 10 + } + return nil + }) + } + if err != nil { + t.Fatal(err) + } + f.read(t, it) // The next request must settle the new facts. + got, err := f.st.Get(it.ID) + if err != nil || got.Publishing != (change != "stop") { + t.Fatalf("next settlement after %s: publishing=%v, err=%v", change, got.Publishing, err) + } + }) + } +} + +func TestSettleRechecksStopAndBudgetAfterListing(t *testing.T) { + for _, change := range []string{"stop", "budget"} { + t.Run(change, func(t *testing.T) { + f := newV2Jobs(t) + it := f.create(t, autoProof()) + if _, err := f.st.List(); err != nil { + t.Fatal(err) + } + other, err := store.Open(f.st.Root()) + if err != nil { + t.Fatal(err) + } + // Retention is the public seam after a candidate is copied. Change + // the disk before settlement can act on that candidate. + var once sync.Once + f.st.SetOutputPolicy(func() (int, int) { + once.Do(func() { + if err := other.Update(it.ID, func(cur *factory.Item) error { + if change == "stop" { + factory.MarkPersonStopped(cur) + } else { + cur.Cap = 1 + } + return nil + }); err != nil { + t.Fatal(err) + } + }) + return 0, 0 + }) + f.read(t, it) + got, err := f.st.Get(it.ID) + if err != nil || got.Publishing { + t.Fatalf("settlement used stale %s authority: publishing=%v, err=%v", change, got.Publishing, err) + } + }) + } +} diff --git a/internal/factory/store/list.go b/internal/factory/store/list.go new file mode 100644 index 0000000000..e525d088a7 --- /dev/null +++ b/internal/factory/store/list.go @@ -0,0 +1,130 @@ +package store + +import ( + "os" + "reflect" + "sync" + "time" +) + +// listRereadEvery bounds how long a document may be trusted by file identity. An +// in-place rewrite keeping its size and time looks unchanged, and a Stop or a +// budget must not rest on file times with no limit. +const listRereadEvery = time.Minute + +// listIndex owns the decoded documents. Nothing returned by List aliases them. +// mu guards only these fields and is never held across a read or a write. +type listIndex struct { + mu sync.Mutex + docs map[string]listedDoc + // fullAt is when the running epoch began. Each epoch starts empty, so its + // first look reads every document. + fullAt time.Time + // epoch stops a scan that began before a full read from publishing over it. + epoch int +} + +// listSnapshot is what one scan reuses. A published map is never changed, so the +// scan may read it unlocked: migration during a scan takes item locks. +func (st *Store) listSnapshot() (map[string]listedDoc, int) { + st.list.mu.Lock() + defer st.list.mu.Unlock() + if now := st.now(); now.Sub(st.list.fullAt) >= listRereadEvery { + st.list.docs, st.list.fullAt = nil, now + st.list.epoch++ + } + return st.list.docs, st.list.epoch +} + +// listPublish keeps a scan's documents unless a newer epoch began meanwhile. A +// scan publishing over a newer one of its epoch costs only reads, because each +// file's identity is checked on every look. +func (st *Store) listPublish(docs map[string]listedDoc, epoch int) { + st.list.mu.Lock() + defer st.list.mu.Unlock() + if epoch == st.list.epoch { + st.list.docs = docs + } +} + +type listedDoc struct { + file os.FileInfo + doc doc +} + +// listed reads only changed documents. Stat precedes read: a rename during +// read may cause one extra read next time, never cache old bytes as a new file. +// EVERY LOOK STATS EVERY FILE. A folder timestamp is not sufficient evidence +// for a grant candidate: another process's rename must be seen on the next look, +// including a replacement whose size and modification time stayed identical. +func (st *Store) listed(path string, previous map[string]listedDoc, next map[string]listedDoc) (doc, error) { + info, err := os.Stat(path) + if err != nil { + return doc{}, err + } + cached, ok := previous[path] + if !ok || !sameDocument(cached.file, info) { + d, err := st.read(path) + if err != nil { + return doc{}, err + } + cached = listedDoc{file: info, doc: d} + } + next[path] = cached + // Copy before migration or retention as well as before the caller sees it. + // A retained view must not discard bytes needed by a later retention policy. + return copyListed(reflect.ValueOf(cached.doc)).Interface().(doc), nil +} + +// copyListed copies the mutable JSON graph without encoding or decoding it +// again. Struct values retain their private fields (notably time.Time); all +// exported pointers, slices and maps are copied recursively, including fields +// added to Item later. Documents come from JSON and therefore contain no cycles. +func copyListed(v reflect.Value) reflect.Value { + switch v.Kind() { + case reflect.Pointer: + if v.IsNil() { + return reflect.Zero(v.Type()) + } + out := reflect.New(v.Type().Elem()) + out.Elem().Set(copyListed(v.Elem())) + return out + case reflect.Interface: + if v.IsNil() { + return reflect.Zero(v.Type()) + } + out := reflect.New(v.Type()).Elem() + out.Set(copyListed(v.Elem())) + return out + case reflect.Slice: + if v.IsNil() { + return reflect.Zero(v.Type()) + } + out := reflect.MakeSlice(v.Type(), v.Len(), v.Len()) + for i := 0; i < v.Len(); i++ { + out.Index(i).Set(copyListed(v.Index(i))) + } + return out + case reflect.Map: + if v.IsNil() { + return reflect.Zero(v.Type()) + } + out := reflect.MakeMapWithSize(v.Type(), v.Len()) + iter := v.MapRange() + for iter.Next() { + out.SetMapIndex(iter.Key(), copyListed(iter.Value())) + } + return out + case reflect.Struct: + out := reflect.New(v.Type()).Elem() + out.Set(v) + for i := 0; i < v.NumField(); i++ { + if v.Type().Field(i).IsExported() { + out.Field(i).Set(copyListed(v.Field(i))) + } + } + return out + default: + return v + } +} diff --git a/internal/factory/store/list_reread_test.go b/internal/factory/store/list_reread_test.go new file mode 100644 index 0000000000..def4ea5c89 --- /dev/null +++ b/internal/factory/store/list_reread_test.go @@ -0,0 +1,71 @@ +package store + +import ( + "os" + "strings" + "testing" + "time" + + "github.com/Agent-Field/codeaf/internal/factory" +) + +// File identity cannot see an in-place rewrite that keeps the size and time, so +// no decision may rest on it for longer than the bound. +func TestListSeesInPlaceSameTimeEditWithinItsBound(t *testing.T) { + st, err := Open(t.TempDir()) + if err != nil { + t.Fatal(err) + } + now := time.Now() + st.clock = func() time.Time { return now } + made, err := st.Create(factory.Item{Title: "before"}) + if err != nil { + t.Fatal(err) + } + title := func() string { + t.Helper() + items, err := st.List() + if err != nil || len(items) != 1 { + t.Fatalf("list: %v, %v", items, err) + } + return items[0].Title + } + if got := title(); got != "before" { + t.Fatalf("title = %q", got) + } + path := st.ItemPath(made.ID) + info, err := os.Stat(path) + if err != nil { + t.Fatal(err) + } + data, err := os.ReadFile(path) + if err != nil { + t.Fatal(err) + } + if err := os.WriteFile(path, []byte(strings.Replace(string(data), `"before"`, `"after!"`, 1)), info.Mode()); err != nil { + t.Fatal(err) + } + if err := os.Chtimes(path, info.ModTime(), info.ModTime()); err != nil { + t.Fatal(err) + } + if next, err := os.Stat(path); err != nil || !os.SameFile(info, next) || next.Size() != info.Size() || !next.ModTime().Equal(info.ModTime()) { + t.Fatalf("the edit did not keep the file's identity: %v", err) + } + + now = now.Add(listRereadEvery - time.Second) + before := st.DocumentReads() + title() + if got := st.DocumentReads() - before; got != 0 { + t.Fatalf("inside the bound a look read %d documents, want 0", got) + } + + now = now.Add(time.Second) + if got := title(); got != "after!" { + t.Fatalf("at the bound title = %q, want the edit", got) + } + before = st.DocumentReads() + title() + if got := st.DocumentReads() - before; got != 0 { + t.Fatalf("after the re-read a look read %d documents, want 0", got) + } +} diff --git a/internal/factory/store/list_test.go b/internal/factory/store/list_test.go new file mode 100644 index 0000000000..da1b1df0aa --- /dev/null +++ b/internal/factory/store/list_test.go @@ -0,0 +1,239 @@ +package store_test + +import ( + "os" + "path/filepath" + "reflect" + "strings" + "sync" + "testing" + "time" + + "github.com/Agent-Field/codeaf/internal/factory" + "github.com/Agent-Field/codeaf/internal/factory/store" + "sync/atomic" +) + +func listStore(t *testing.T) *store.Store { + t.Helper() + st, err := store.Open(t.TempDir()) + if err != nil { + t.Fatal(err) + } + return st +} + +func listOne(t *testing.T, st *store.Store) factory.Item { + t.Helper() + items, err := st.List() + if err != nil || len(items) != 1 { + t.Fatalf("List: %d items, %v", len(items), err) + } + return items[0] +} + +func TestListOwnsMutableDocuments(t *testing.T) { + st := listStore(t) + stopped := false + made, err := st.Create(factory.Item{ + Repo: "acme/api", State: factory.StateNew, PersonStopped: &stopped, + SourceAssignees: []string{}, Labels: []string{"original"}, + BriefSeen: map[string]int{"sam": 2}, Stages: []factory.Stage{{Name: "test", Proof: []string{"original"}}}, + Stream: &factory.Stream{Phases: []factory.Phase{{RoundCosts: map[int]float64{1: 2}, + Result: &factory.StageResult{Files: []string{"original"}, Tests: &factory.TestCounts{Passed: 2}}}}, + Log: []factory.LogLine{{Text: "original"}}}, + AnsweredQuestions: []factory.Question{{Options: []string{"original"}, Answer: &factory.Answer{Text: "original"}}}, + }) + if err != nil { + t.Fatal(err) + } + want := listOne(t, st) + before := st.DocumentReads() + mutate := func(it factory.Item) { + *it.PersonStopped = true + it.Labels[0] = "mutated" + it.BriefSeen["sam"] = 99 + it.Stages[0].Proof[0] = "mutated" + it.Stream.Phases[0].RoundCosts[1] = 99 + it.Stream.Phases[0].Result.Files[0] = "mutated" + it.Stream.Phases[0].Result.Tests.Passed = 99 + it.Stream.Log[0].Text = "mutated" + it.AnsweredQuestions[0].Options[0] = "mutated" + it.AnsweredQuestions[0].Answer.Text = "mutated" + } + var wg sync.WaitGroup + for range 8 { + wg.Add(1) + go func() { + defer wg.Done() + for range 10 { + mutate(listOne(t, st)) + } + }() + } + wg.Wait() + if got := listOne(t, st); !reflect.DeepEqual(got, want) { + t.Fatal("mutating a List result changed a later result") + } + if n := st.DocumentReads() - before; n != 0 { + t.Fatalf("copies read %d documents, want 0", n) + } + if got, err := st.Get(made.ID); err != nil || !reflect.DeepEqual(got, want) { + t.Fatal("mutating a List result changed the stored document", err) + } +} + +func TestListReappliesOutputRetention(t *testing.T) { + st := listStore(t) + _, err := st.Create(factory.Item{Stream: &factory.Stream{Log: []factory.LogLine{ + {At: time.Now().Add(-48 * time.Hour), Text: "older"}, {At: time.Now(), Text: "newer"}, + }}}) + if err != nil { + t.Fatal(err) + } + bytes := 100 + days := 3 + st.SetOutputPolicy(func() (int, int) { return days, bytes }) + if got := listOne(t, st); len(got.Stream.Log) != 2 { + t.Fatal("initial output missing") + } + before := st.DocumentReads() + bytes = 5 + if got := listOne(t, st); len(got.Stream.Log) != 1 || got.Stream.Log[0].Text != "newer" { + t.Fatal("cached output ignored stricter retention") + } + bytes = 100 + if got := listOne(t, st); len(got.Stream.Log) != 2 || got.Stream.Log[0].Text != "older" { + t.Fatal("a retained view destroyed the cached output") + } + days = 1 + if got := listOne(t, st); len(got.Stream.Log) != 1 || got.Stream.Log[0].Text != "newer" { + t.Fatal("cached output ignored age retention") + } + if n := st.DocumentReads() - before; n != 0 { + t.Fatalf("retention read %d unchanged documents, want 0", n) + } +} + +func TestListReappliesRepositoryMigration(t *testing.T) { + st := listStore(t) + made, err := st.Create(factory.Item{Repo: "ledger", Places: []string{"ledger"}}) + if err != nil { + t.Fatal(err) + } + if got := listOne(t, st); got.Repo != "ledger" { + t.Fatal("unresolved repository changed") + } + if err := st.SetRepos([]string{"acme/ledger"}); err != nil { + t.Fatal(err) + } + if got := listOne(t, st); got.Repo != "acme/ledger" || got.Places[0] != "acme/ledger" || !got.Changed.Equal(made.Changed) { + t.Fatalf("cached legacy document did not migrate: %+v", got) + } + if got, err := st.Get(made.ID); err != nil || got.Repo != "acme/ledger" { + t.Fatal("migration was not persisted", err) + } + listOne(t, st) // The migration itself replaced the file. + before := st.DocumentReads() + listOne(t, st) + if n := st.DocumentReads() - before; n != 0 { + t.Fatalf("migration kept rereading: %d", n) + } +} + +func TestListSeesRenameWithSameSizeAndTime(t *testing.T) { + st := listStore(t) + made, err := st.Create(factory.Item{Title: "before"}) + if err != nil { + t.Fatal(err) + } + listOne(t, st) + path := st.ItemPath(made.ID) + info, err := os.Stat(path) + if err != nil { + t.Fatal(err) + } + data, err := os.ReadFile(path) + if err != nil { + t.Fatal(err) + } + // Replace one equal-length string without changing the document's encoding. + if strings.Count(string(data), `"before"`) != 1 { + t.Fatal("expected exactly one title to replace") + } + data = []byte(strings.Replace(string(data), `"before"`, `"after!"`, 1)) + tmp := filepath.Join(st.Root(), ".replacement") + if err := os.WriteFile(tmp, data, 0o600); err != nil { + t.Fatal(err) + } + if err := os.Chtimes(tmp, info.ModTime(), info.ModTime()); err != nil { + t.Fatal(err) + } + if err := os.Rename(tmp, path); err != nil { + t.Fatal(err) + } + if next, err := os.Stat(path); err != nil || next.Size() != info.Size() || !next.ModTime().Equal(info.ModTime()) || os.SameFile(info, next) { + t.Fatal("replacement did not preserve size/time with a new identity", err) + } + before := st.DocumentReads() + if got := listOne(t, st); got.Title != "after!" { + t.Fatal("same-size/time replacement was hidden") + } + if n := st.DocumentReads() - before; n != 1 { + t.Fatalf("replacement read %d documents, want 1", n) + } +} + +// An Update callback holding an item's lock may list the floor while another +// List is migrating that item. List must not wait on the item lock while holding +// anything the callback's List needs. +func TestListDoesNotHoldItsIndexAcrossMigration(t *testing.T) { + st := listStore(t) + if _, err := st.Create(factory.Item{Repo: "ledger", Stream: &factory.Stream{}}); err != nil { + t.Fatal(err) + } + legacy, err := st.Create(factory.Item{Repo: "ledger"}) + if err != nil { + t.Fatal(err) + } + if err := st.SetRepos([]string{"acme/ledger"}); err != nil { + t.Fatal(err) + } + locked, listing, proceed := make(chan struct{}), make(chan struct{}), make(chan struct{}) + finished := make(chan error, 2) + var once atomic.Bool + st.SetOutputPolicy(func() (int, int) { + if once.CompareAndSwap(false, true) { + close(listing) + <-proceed + } + return 0, 0 + }) + go func() { + finished <- st.Update(legacy.ID, func(*factory.Item) error { + close(locked) + <-listing + if err := st.SetRepos(nil); err != nil { + return err + } + close(proceed) + _, err := st.List() + return err + }) + }() + <-locked + go func() { + _, err := st.List() + finished <- err + }() + for range 2 { + select { + case err := <-finished: + if err != nil { + t.Fatal(err) + } + case <-time.After(5 * time.Second): + t.Fatal("deadlock: List held its index while waiting for an item lock") + } + } +} diff --git a/internal/factory/store/repository.go b/internal/factory/store/repository.go index be328ad2be..b39c83968b 100644 --- a/internal/factory/store/repository.go +++ b/internal/factory/store/repository.go @@ -183,7 +183,7 @@ func (st *Store) migrateRepository(d doc, ids repositoryIDs) (doc, error) { if err != nil { return err } - return writeAtomic(st.ItemPath(d.Item.ID), data) + return st.writeItem(st.ItemPath(d.Item.ID), data) }) return d, err } diff --git a/internal/factory/store/store.go b/internal/factory/store/store.go index 23e97df65c..95693b4434 100644 --- a/internal/factory/store/store.go +++ b/internal/factory/store/store.go @@ -61,8 +61,17 @@ type Store struct { // clock is the store's now. A test may hold it still; it is nil in every // real build, which reads as time.Now. clock func() time.Time + // reads counts the job documents this store has decoded. + reads atomic.Int64 + // talks is what [Store.TalkStop] last read. + talks talkIndex + // list owns the documents last decoded by List. + list listIndex } +// DocumentReads is how many job documents this store has decoded so far. +func (st *Store) DocumentReads() int64 { return st.reads.Load() } + // Open answers the store at root, creating the directory. It holds no handles. func Open(root string) (*Store, error) { if strings.TrimSpace(root) == "" { @@ -129,7 +138,7 @@ func (st *Store) Create(it factory.Item) (factory.Item, error) { } return st.underItemLock(it.ID, func() error { return live.Commit(st.root, live.Changes(factory.Item{}, it, st.now()), func() error { - return writeAtomic(st.ItemPath(it.ID), data) + return st.writeItem(st.ItemPath(it.ID), data) }) }) }) @@ -220,10 +229,12 @@ func (st *Store) Get(id int) (factory.Item, error) { return d.Item, nil } -// List reads every item, newest first by Created. An unreadable or +// List lists every item, newest first by Created. Unchanged files reuse their +// decoded documents; every returned item owns its mutable data. An unreadable or // newer-schema document is skipped rather than fatal: ONE DOCUMENT FROM THE // FUTURE, OR ONE TORN BY A FULL DISK, MUST NOT EMPTY THE WHOLE FLOOR. func (st *Store) List() ([]factory.Item, error) { + previous, epoch := st.listSnapshot() entries, err := os.ReadDir(st.root) if err != nil { if os.IsNotExist(err) { @@ -231,6 +242,7 @@ func (st *Store) List() ([]factory.Item, error) { } return nil, err } + next := make(map[string]listedDoc, len(entries)) items := make([]factory.Item, 0, len(entries)) var ids repositoryIDs loadedIDs := false @@ -245,7 +257,7 @@ func (st *Store) List() ([]factory.Item, error) { if _, err := strconv.Atoi(strings.TrimSuffix(entry.Name(), ".json")); err != nil { continue } - d, err := st.read(filepath.Join(st.root, entry.Name())) + d, err := st.listed(filepath.Join(st.root, entry.Name()), previous, next) if err != nil { continue } @@ -263,6 +275,7 @@ func (st *Store) List() ([]factory.Item, error) { st.retainOutput(&d.Item) items = append(items, d.Item) } + st.listPublish(next, epoch) sort.SliceStable(items, func(a, b int) bool { if items[a].Created.Equal(items[b].Created) { return items[a].ID > items[b].ID @@ -315,7 +328,7 @@ func (st *Store) Save(it factory.Item) error { return err } err = live.Commit(st.root, live.Changes(current.Item, it, st.now()), func() error { - return writeAtomic(st.ItemPath(it.ID), data) + return st.writeItem(st.ItemPath(it.ID), data) }) was, wrote = current.Item, err == nil return err @@ -520,7 +533,7 @@ func (st *Store) update(id int, mode writeMode, expect *int, change func(*factor return err } err = live.Commit(st.root, live.Changes(before, it, now), func() error { - return writeAtomic(st.ItemPath(id), data) + return st.writeItem(st.ItemPath(id), data) }) was, after, wrote = before, it, err == nil return err @@ -564,6 +577,7 @@ func (st *Store) now() time.Time { } func (st *Store) read(path string) (doc, error) { + st.reads.Add(1) data, err := os.ReadFile(path) if err != nil { if os.IsNotExist(err) { @@ -674,6 +688,13 @@ func (st *Store) underLock(path string, write func() error) error { return write() } +// writeItem is every write of a job's document. +func (st *Store) writeItem(path string, data []byte) error { + err := writeAtomic(path, data) + st.talks.stale.Store(true) + return err +} + func writeAtomic(path string, data []byte) error { temporary, err := os.CreateTemp(filepath.Dir(path), "."+filepath.Base(path)+".*") if err != nil { diff --git a/internal/factory/store/talks.go b/internal/factory/store/talks.go new file mode 100644 index 0000000000..e5cc0cef94 --- /dev/null +++ b/internal/factory/store/talks.go @@ -0,0 +1,203 @@ +package store + +import ( + "errors" + "os" + "path/filepath" + "sort" + "strconv" + "strings" + "sync" + "sync/atomic" + "time" + + "github.com/Agent-Field/codeaf/internal/factory" +) + +// ── WHICH JOB IS THIS CHAT'S, AND IS IT STOPPED ───────────────── +// +// Every open job chat asks this ten times a second for as long as it is open +// (session.watchJobBudget), so reading the floor to answer it kept an idle +// engine busy in proportion to chats times jobs. [Store.TalkStop] answers from +// what it last read and proves nothing changed from the files themselves. +// +// THE PROOF. Every store write renames a fresh file into this folder. That +// changes the folder's time, and gives the document a new file identity, in +// this process or any other. So: +// - The folder unchanged, and last changed long enough before the last sweep +// that a coarse clock cannot hide a later write, means no document changed. +// - Otherwise, a sweep compares each document's file with the one last read, +// and reads only those that differ. +// - A rewrite that bypasses the store and keeps the file's identity proves +// nothing either way, so every document is read again each talkRereadEvery. +// +// Both are looked at BEFORE reading, so a write landing during a sweep leaves +// the older identity beside the newer content and the next look reads it again. +// +// THIS ANSWERS A WATCHER, NOT A GATE. [factory.CheckPersonStop] on the locked +// job still decides any grant of work; a watcher only notices a Stop sooner. + +const ( + // folderTimeSlack is how recently the folder may have changed and still be + // swept: wider than the coarsest file time on a supported disk. + folderTimeSlack = 2 * time.Second + // talkSweepEvery bounds how long a quiet folder goes unswept. A sweep resolves + // each job's chat path again: a link retargeted under it changes no document. + talkSweepEvery = time.Second + // talkRereadEvery bounds how long a document is trusted by file identity. An + // in-place rewrite keeping its size and time looks unchanged, and a Stop or a + // budget must not rest on file times with no limit. + talkRereadEvery = time.Minute +) + +type talkDoc struct { + file os.FileInfo + id int + created time.Time + talk string // the job's Talk as written + key string // talk as [FileKey], resolved again at every sweep + stopped bool +} + +type talkIndex struct { + mu sync.Mutex + // stale says a write through this store came after the last sweep. Other + // writers show in the folder's time, unless a clock hides them: see quiet. + stale atomic.Bool + dir os.FileInfo + swept time.Time + reread time.Time // when every document was last read + retry bool // a document could not be read, and is owed another try + docs map[string]talkDoc + owners []talkDoc // the docs with a chat, newest first as [Store.List] orders +} + +// TalkStop answers for the conversation with this [FileKey]: whether a job names +// it as its chat, and whether a person has stopped that job. It agrees with +// finding the first match in [Store.List], reading no document that has not +// changed since the last call on this store. +func (st *Store) TalkStop(key string) (manager, stopped bool, err error) { + ix := &st.talks + ix.mu.Lock() + defer ix.mu.Unlock() + if err := st.refreshTalks(ix); err != nil { + return false, false, err + } + for _, doc := range ix.owners { + if doc.key == key { + return true, doc.stopped, nil + } + } + return false, false, nil +} + +func (st *Store) refreshTalks(ix *talkIndex) error { + now := st.now() + dir, err := os.Stat(st.root) + if err != nil { + if os.IsNotExist(err) { + ix.forget() + return nil + } + return err + } + if ix.quiet(dir, now) { + return nil + } + // Cleared before the sweep, so a write during it is owed another. + ix.stale.Store(false) + entries, err := os.ReadDir(st.root) + if err != nil { + if os.IsNotExist(err) { + ix.forget() + return nil + } + ix.stale.Store(true) // still owed the sweep this failed to do + return err + } + known, reread := ix.docs, ix.reread + if now.Sub(reread) >= talkRereadEvery { + known, reread = nil, now + } + docs := make(map[string]talkDoc, len(entries)) + retry := false + for _, entry := range entries { + name := entry.Name() + if entry.IsDir() || !strings.HasSuffix(name, ".json") { + continue + } + id, err := strconv.Atoi(strings.TrimSuffix(name, ".json")) + if err != nil { + continue + } + path := filepath.Join(st.root, name) + info, err := os.Stat(path) + if err != nil { + retry = retry || !os.IsNotExist(err) + continue + } + if last, ok := known[name]; ok && sameDocument(last.file, info) { + last.key = FileKey(last.talk) + docs[name] = last + continue + } + d, err := st.read(path) + switch { + case err == nil: + docs[name] = talkDoc{file: info, id: id, created: d.Item.Created, talk: d.Item.Talk, key: FileKey(d.Item.Talk), stopped: factory.StoppedByPerson(d.Item)} + case errors.Is(err, ErrNewer): + // Skipped until the file changes: another read could not answer differently. + docs[name] = talkDoc{file: info, id: id} + case !errors.Is(err, ErrNotFound): + retry = true + } + } + owners := make([]talkDoc, 0, 8) + for _, doc := range docs { + if doc.key != "" { + owners = append(owners, doc) + } + } + sort.Slice(owners, func(a, b int) bool { + if owners[a].created.Equal(owners[b].created) { + return owners[a].id > owners[b].id + } + return owners[a].created.After(owners[b].created) + }) + ix.dir, ix.swept, ix.reread, ix.retry, ix.docs, ix.owners = dir, now, reread, retry, docs, owners + return nil +} + +// forget drops what was read, as for a floor that isn't there. Its lock is held. +func (ix *talkIndex) forget() { + ix.dir, ix.swept, ix.reread, ix.retry, ix.docs, ix.owners = nil, time.Time{}, time.Time{}, false, nil, nil +} + +// quiet is whether the last sweep still stands, with dir the folder as it is now. +// THE ONE LIMIT: another process's write whose folder time comes out identical, +// which takes a clock moved back past the slack, waits for the next sweep, at +// most talkSweepEvery. A write through this store never waits. +func (ix *talkIndex) quiet(dir os.FileInfo, now time.Time) bool { + return ix.docs != nil && !ix.retry && !ix.stale.Load() && + now.Sub(ix.swept) < talkSweepEvery && + now.Sub(ix.reread) < talkRereadEvery && + os.SameFile(ix.dir, dir) && ix.dir.ModTime().Equal(dir.ModTime()) && + ix.swept.Sub(dir.ModTime()) >= folderTimeSlack +} + +// sameDocument is whether info is still the file a document was read from. +func sameDocument(read, info os.FileInfo) bool { + return os.SameFile(read, info) && read.Size() == info.Size() && read.ModTime().Equal(info.ModTime()) +} + +// FileKey is a path as one spelling, links resolved; empty for no path. +func FileKey(path string) string { + path = strings.TrimSpace(path) + if path == "" { + return "" + } + if real, err := filepath.EvalSymlinks(path); err == nil { + path = real + } + return filepath.Clean(path) +} diff --git a/internal/factory/store/talks_reread_test.go b/internal/factory/store/talks_reread_test.go new file mode 100644 index 0000000000..d3c1997bf1 --- /dev/null +++ b/internal/factory/store/talks_reread_test.go @@ -0,0 +1,77 @@ +package store + +import ( + "os" + "strings" + "testing" + "time" + + "github.com/Agent-Field/codeaf/internal/factory" +) + +// File identity cannot see an in-place rewrite that keeps the size and time, so +// a watcher may not rest on it for longer than the bound. +func TestTalkStopSeesInPlaceSameTimeEditWithinItsBound(t *testing.T) { + st, err := Open(t.TempDir()) + if err != nil { + t.Fatal(err) + } + now := time.Now() + st.clock = func() time.Time { return now } + running := false + const talk = "/nowhere/chat.jsonl" + made, err := st.Create(factory.Item{Talk: talk, PersonStopped: &running}) + if err != nil { + t.Fatal(err) + } + stopped := func() bool { + t.Helper() + manager, stopped, err := st.TalkStop(FileKey(talk)) + if err != nil || !manager { + t.Fatalf("TalkStop: manager %v, %v", manager, err) + } + return stopped + } + if stopped() { + t.Fatal("stopped before any Stop") + } + path := st.ItemPath(made.ID) + info, err := os.Stat(path) + if err != nil { + t.Fatal(err) + } + data, err := os.ReadFile(path) + if err != nil { + t.Fatal(err) + } + const was, is = `"PersonStopped": false`, `"PersonStopped": true ` + if strings.Count(string(data), was) != 1 { + t.Fatalf("expected one %s in %s", was, data) + } + if err := os.WriteFile(path, []byte(strings.Replace(string(data), was, is, 1)), info.Mode()); err != nil { + t.Fatal(err) + } + if err := os.Chtimes(path, info.ModTime(), info.ModTime()); err != nil { + t.Fatal(err) + } + if next, err := os.Stat(path); err != nil || !os.SameFile(info, next) || next.Size() != info.Size() || !next.ModTime().Equal(info.ModTime()) { + t.Fatalf("the edit did not keep the file's identity: %v", err) + } + + now = now.Add(talkRereadEvery - time.Second) + before := st.DocumentReads() + stopped() + if got := st.DocumentReads() - before; got != 0 { + t.Fatalf("inside the bound a look read %d documents, want 0", got) + } + + now = now.Add(time.Second) + if !stopped() { + t.Fatal("at the bound the Stop was still not seen") + } + before = st.DocumentReads() + stopped() + if got := st.DocumentReads() - before; got != 0 { + t.Fatalf("after the re-read a look read %d documents, want 0", got) + } +} diff --git a/internal/identity/l4fix2_startup_test.go b/internal/identity/l4fix2_startup_test.go index 37a2bf8f9f..cdfa2efa5c 100644 --- a/internal/identity/l4fix2_startup_test.go +++ b/internal/identity/l4fix2_startup_test.go @@ -67,6 +67,11 @@ func firstStart(t *testing.T) string { // worker is: a loaded machine must not turn setup into a failure. ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) defer cancel() + // Memory off reads no history until the Memory page asks, so the fixture asks. + if err := vault.AskDiscovery(); err != nil { + cancel() + t.Fatal(err) + } done := make(chan error, 1) guard.Go("identity/startup-fixture", func() { done <- vault.Run(ctx, session.KnowledgeHistory(filepath.Join(home, "v3/projects")), func() string { return "" }) @@ -168,6 +173,20 @@ func TestL4Fix2FirstStartCanJoin(t *testing.T) { // L4-H2: Narrow startup rows must not block the base memory merge, and an // unknown graph event must still refuse a move even when its projections are empty. +// Opening the Memory page with nothing else done is still a first start: the ask +// it leaves for the knowledge worker is not the person's work. +func TestL4Fix2AnOpenedMemoryPageCanJoin(t *testing.T) { + home := firstStart(t) + startupFile(t, home, "knowledge/.discovery-asked", "") + joinable, err := identity.Joinable(home) + if err != nil { + t.Fatal(err) + } + if !joinable { + t.Fatal("the Memory page's ask for a chat count made this home unable to join") + } +} + func TestL4Fix2GraphMemoryRemainsJoinable(t *testing.T) { home := firstStart(t) graph, err := store.Open(filepath.Join(home, "graph.db")) diff --git a/internal/identity/pristine.go b/internal/identity/pristine.go index 6bd685e107..eb43292983 100644 --- a/internal/identity/pristine.go +++ b/internal/identity/pristine.go @@ -39,6 +39,7 @@ var startupRules = rules{ "model-catalog-*.json": catalogCache, // Provider catalog caches are fetched again on demand. "knowledge": within(rules{ ".worker.lock": never, + ".discovery-asked": never, // the Memory page asking the worker for a count "knowledge.sqlite": knowledgeStartup, "knowledge.sqlite-wal": never, "knowledge.sqlite-shm": never, }), diff --git a/internal/knowledge/build.go b/internal/knowledge/build.go index 2bfdd48516..93dd2c6fa9 100644 --- a/internal/knowledge/build.go +++ b/internal/knowledge/build.go @@ -198,6 +198,11 @@ func markJobSessions(db *sql.DB, journals []Journal) error { } func (v *Vault) runPass(ctx context.Context, history History, key func() string, lastInventory *string) error { + // Memory off costs one look at the setting and one at the ask: no saved chat + // is read, and no database opened, until the Memory page asks for a count. + if !v.Enabled() && !v.takeDiscoveryAsk() { + return nil + } journals, err := history.Inventory(ctx) if err != nil { return err @@ -212,7 +217,8 @@ func (v *Vault) runPass(ctx context.Context, history History, key func() string, return err } if !v.Enabled() { - // Discovery is local. No inventory, including an empty one, opts in. + // Discovery is local, and asked for. No inventory, including an empty + // one, opts in. fingerprint := identity(journals) if textValue(job["inventory_hash"]) == fingerprint { return nil @@ -428,3 +434,22 @@ func (v *Vault) runPass(ctx context.Context, history History, key func() string, } return saveState(db, "job", job) } + +// discoveryAsk is the file that says the Memory page asked for a count. +const discoveryAsk = ".discovery-asked" + +// AskDiscovery asks the worker, in whichever process runs it, to count the +// saved chats while Memory is off. The Memory page asks when it opens; nothing +// reads saved chats with Memory off until then. +func (v *Vault) AskDiscovery() error { + if err := os.MkdirAll(v.root, 0700); err != nil { + return err + } + return os.WriteFile(filepath.Join(v.root, discoveryAsk), nil, 0600) +} + +// takeDiscoveryAsk says whether a count was asked for, and clears the ask first, +// so one that arrives during the count is answered by another count. +func (v *Vault) takeDiscoveryAsk() bool { + return os.Remove(filepath.Join(v.root, discoveryAsk)) == nil +} diff --git a/internal/knowledge/discovery_test.go b/internal/knowledge/discovery_test.go new file mode 100644 index 0000000000..b740030368 --- /dev/null +++ b/internal/knowledge/discovery_test.go @@ -0,0 +1,82 @@ +package knowledge + +import ( + "context" + "sync/atomic" + "testing" + "time" +) + +// countingHistory is saved chats that count how often they were looked at. +type countingHistory struct { + inventories, reads atomic.Int64 + chats atomic.Int64 +} + +func (h *countingHistory) Inventory(context.Context) ([]Journal, error) { + h.inventories.Add(1) + out := []Journal{} + for i := range h.chats.Load() { + out = append(out, Journal{ID: string(rune('a' + i)), Size: 1}) + } + return out, nil +} + +func (h *countingHistory) Read(_ context.Context, j Journal, _ JournalCursor) (JournalBatch, error) { + h.reads.Add(1) + return JournalBatch{Sources: []Source{{ID: j.ID}}, Cursor: JournalCursor{Offset: j.Size}}, nil +} + +func waitChats(t *testing.T, v *Vault, want int) { + t.Helper() + deadline := time.Now().Add(6 * time.Second) + for time.Now().Before(deadline) { + s, err := v.Status() + if err != nil { + t.Fatal(err) + } + if s.Build.Chats == want && s.Build.Phase == BuildUnbuilt { + return + } + time.Sleep(20 * time.Millisecond) + } + s, _ := v.Status() + t.Fatalf("the count never reached %d: %+v", want, s.Build) +} + +// With Memory off nothing reads saved chats in the background. The count is +// taken when the Memory page asks, once, and it stays for the next visit. +func TestMemoryOffReadsNoChatsUntilThePageAsks(t *testing.T) { + v := Open(t.TempDir()) + if v.Enabled() { + t.Fatal("a new profile should start with Memory off") + } + history := &countingHistory{} + history.chats.Store(2) + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + go func() { defer close(done); _ = v.Run(ctx, history, func() string { return "" }) }() + t.Cleanup(func() { cancel(); <-done }) + + // Two passes run two seconds apart. + time.Sleep(2300 * time.Millisecond) + if n, r := history.inventories.Load(), history.reads.Load(); n != 0 || r != 0 { + t.Fatalf("with Memory off and nobody asking, the worker took %d inventories and %d reads", n, r) + } + + if err := v.AskDiscovery(); err != nil { + t.Fatal(err) + } + waitChats(t, v, 2) + time.Sleep(2300 * time.Millisecond) + if n := history.inventories.Load(); n != 1 { + t.Fatalf("one ask took %d inventories, want 1", n) + } + waitChats(t, v, 2) // the last count stays + + history.chats.Store(3) + if err := v.AskDiscovery(); err != nil { + t.Fatal(err) + } + waitChats(t, v, 3) +} diff --git a/internal/knowledge/runtime_test.go b/internal/knowledge/runtime_test.go index fc686e287a..caf8985a97 100644 --- a/internal/knowledge/runtime_test.go +++ b/internal/knowledge/runtime_test.go @@ -320,8 +320,9 @@ func TestKnowledgeMemoryOffDoesNotBuild(t *testing.T) { if err := v.Flush(context.Background(), "fixture", true); !errors.Is(err, knowledge.ErrDisabled) { t.Fatalf("explicit compilation while off: %v", err) } + // No phase at all: with Memory off and nothing asked for, no pass looked. state, err := v.Snapshot() - if err != nil || state.Enabled || state.Pending != 0 || len(state.Pages) != 0 || state.Build.Phase != knowledge.BuildUnbuilt || calls.Load() != 0 { + if err != nil || state.Enabled || state.Pending != 0 || len(state.Pages) != 0 || state.Build.Phase != "" || calls.Load() != 0 { t.Fatalf("memory off built or opted in: state=%+v calls=%d err=%v", state, calls.Load(), err) } }) diff --git a/internal/session/factory_person_stop.go b/internal/session/factory_person_stop.go index 8d026054f0..1a066e6837 100644 --- a/internal/session/factory_person_stop.go +++ b/internal/session/factory_person_stop.go @@ -19,6 +19,17 @@ func (a *Agent) automaticFactoryRefusal() string { return "" } +// watchedFactoryRefusal is the same question for the ten-a-second budget +// watcher, through the door's cheaper answer where it has one. +func (a *Agent) watchedFactoryRefusal() string { + if a.config.FactoryWorkRefusal == nil { + if door, ok := a.config.FactoryRun.(interface{ WatchedRefusal(string) string }); ok { + return door.WatchedRefusal(a.config.SessionFile) + } + } + return a.automaticFactoryRefusal() +} + // factoryTurnRefusal excludes every turn owed to the person, including steering. func (a *Agent) factoryTurnRefusal(user userMessage) string { if user.factoryWork || user.wake && !user.answersPerson && !user.answer { diff --git a/internal/session/job_budget.go b/internal/session/job_budget.go index a20702e677..3cbde11dc6 100644 --- a/internal/session/job_budget.go +++ b/internal/session/job_budget.go @@ -50,7 +50,7 @@ func (a *Agent) watchJobBudget() { stopOwned := a.factoryOwnedStop() if err := a.jobBudgetBlock(); err != nil && err.Error() != JobBudgetUnreadable { _ = a.stopWorkEpoch(epoch) - } else if a.automaticFactoryRefusal() == factory.PersonStopReason { + } else if a.watchedFactoryRefusal() == factory.PersonStopReason { stopOwned() } } diff --git a/internal/session/knowledge_history_test.go b/internal/session/knowledge_history_test.go index d50f5d0067..8acbf11e6f 100644 --- a/internal/session/knowledge_history_test.go +++ b/internal/session/knowledge_history_test.go @@ -276,6 +276,9 @@ func TestKnowledgeInitialBuildImportsAllProjectsAndAutopopulates(t *testing.T) { started := make(chan struct{}, 1) release := make(chan struct{}) calls := knowledgeProviderFixture(t, started, release) + if err := v.AskDiscovery(); err != nil { + t.Fatal(err) + } runKnowledgeWorker(t, v, root) state := waitKnowledgeState(t, v, func(s knowledge.Snapshot) bool { return s.Build.Phase == knowledge.BuildUnbuilt }) if state.Enabled || state.Build.Chats != 2 || calls.Load() != 0 { diff --git a/internal/session/stop_owned_regression_test.go b/internal/session/stop_owned_regression_test.go index d56d22ed15..227f467ba1 100644 --- a/internal/session/stop_owned_regression_test.go +++ b/internal/session/stop_owned_regression_test.go @@ -5,6 +5,7 @@ import ( "fmt" "github.com/Agent-Field/agentfield/sdk/go/ai" "os" + "path/filepath" "strings" "sync" "sync/atomic" @@ -359,3 +360,50 @@ func TestBudgetReadErrorHoldsAdmissionWithoutCancellingWork(t *testing.T) { t.Fatal("unknown budget admitted a paid call") } } + +// watchedStopDoor answers the watcher's question and the gate's separately. +type watchedStopDoor struct { + RunDoor + watched, gate *atomic.Int64 + stopped *atomic.Bool +} + +func (d watchedStopDoor) answer() string { + if d.stopped.Load() { + return factory.PersonStopReason + } + return "" +} +func (d watchedStopDoor) WatchedRefusal(string) string { d.watched.Add(1); return d.answer() } +func (d watchedStopDoor) AutomaticRefusal(string) string { + d.gate.Add(1) + return d.answer() +} + +// The watcher takes the cheap answer and still stops owned work within three +// polls; a turn deciding whether to spend asks the gate, never the cheap answer. +func TestTheBudgetWatcherStopsOwnedWorkThroughTheWatchedAnswer(t *testing.T) { + var watched, gate atomic.Int64 + var stopped atomic.Bool + a, _ := newTestAgent(t, &scriptedCompleter{}, func(c *Config) { + c.JobBudget = &liveBudgetStub{cap: 10} + c.SessionFile = filepath.Join(t.TempDir(), "transcript.jsonl") + c.FactoryRun = watchedStopDoor{watched: &watched, gate: &gate, stopped: &stopped} + }) + id := startJob(t, a, "sleep 30") + time.Sleep(250 * time.Millisecond) + if watched.Load() == 0 { + t.Fatal("the watcher did not ask the watched answer") + } + if gate.Load() != 0 { + t.Fatalf("the watcher asked the gate %d times", gate.Load()) + } + stopped.Store(true) + time.Sleep(350 * time.Millisecond) + if a.jobs.find(id).running() { + t.Fatal("a person's Stop reached the watcher, but owned work still runs") + } + if a.automaticFactoryRefusal() != factory.PersonStopReason || gate.Load() == 0 { + t.Fatal("the turn gate did not ask the full answer") + } +} diff --git a/internal/tui3/knowledge.go b/internal/tui3/knowledge.go index f7cda9915a..081e6091f0 100644 --- a/internal/tui3/knowledge.go +++ b/internal/tui3/knowledge.go @@ -2,6 +2,7 @@ package tui3 import ( "fmt" + "log" "path/filepath" "sort" "strings" @@ -155,6 +156,13 @@ func (a *app) knowledgeRead(create bool) tea.Cmd { } } r, err := v.Status() + if err == nil && !r.Enabled && cached.Revision == "" { + // The offer is on screen: count the saved chats now, as nothing + // does with Memory off until a person is looking. + if problem := v.AskDiscovery(); problem != nil { + log.Printf("knowledge: asking for a count: %v", problem) + } + } if err == nil { if r.Revision == cached.Revision { r.Pages = cached.Pages diff --git a/internal/tui3/knowledge_test.go b/internal/tui3/knowledge_test.go index 875a0e4e0d..c1cf35128f 100644 --- a/internal/tui3/knowledge_test.go +++ b/internal/tui3/knowledge_test.go @@ -2,6 +2,7 @@ package tui3 import ( tea "charm.land/bubbletea/v2" + "context" "encoding/json" "fmt" "github.com/Agent-Field/codeaf/internal/knowledge" @@ -12,6 +13,7 @@ import ( "path/filepath" "strings" "testing" + "time" ) func TestMemoryOffersKnowledgeWithoutStartingCapture(t *testing.T) { @@ -42,6 +44,51 @@ type knowledgeTestAgent struct { func (a *knowledgeTestAgent) Knowledge() *knowledge.Vault { return a.vault } func (a *knowledgeTestAgent) KnowledgeCaptureError() string { return a.captureErr } +// The offer is where it was, and opening it is what asks for the chat count: a +// worker that had read nothing answers with the right count, Memory still off. +func TestOpeningTheOfferAsksForTheChatCount(t *testing.T) { + a := placeApp(t) + v := knowledge.Open(t.TempDir()) + a.agent = &knowledgeTestAgent{fakeAgent: &fakeAgent{model: "m"}, vault: v} + history := offerHistory{} + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + go func() { defer close(done); _ = v.Run(ctx, history, func() string { return "" }) }() + t.Cleanup(func() { cancel(); <-done }) + if s, err := v.Status(); err != nil || s.Build.Chats != 0 { + t.Fatalf("before the page opened: %+v %v", s.Build, err) + } + + drive(t, a, key(placeChord(pageMemory)), key("alt+u")) + if got := placeFrameText(a); !strings.Contains(got, "Build knowledge base") { + t.Fatalf("the offer did not appear at once: %s", got) + } + deadline := time.Now().Add(6 * time.Second) + for { + s, err := v.Status() + if err != nil { + t.Fatal(err) + } + if s.Build.Chats == 2 && s.Build.Phase == knowledge.BuildUnbuilt && !s.Enabled { + return + } + if time.Now().After(deadline) { + t.Fatalf("no count after the offer opened: %+v", s.Build) + } + time.Sleep(20 * time.Millisecond) + } +} + +type offerHistory struct{} + +func (offerHistory) Inventory(context.Context) ([]knowledge.Journal, error) { + return []knowledge.Journal{{ID: "a", Size: 1}, {ID: "b", Size: 1}}, nil +} + +func (offerHistory) Read(_ context.Context, j knowledge.Journal, _ knowledge.JournalCursor) (knowledge.JournalBatch, error) { + return knowledge.JournalBatch{Sources: []knowledge.Source{{ID: j.ID}}, Cursor: knowledge.JournalCursor{Offset: j.Size}}, nil +} + func TestMemoryCreatesProjectKnowledgeOnlyOnEnter(t *testing.T) { a := placeApp(t) v := knowledge.Open(t.TempDir())