From 199bd9d16ce5fd1f9570fb6548a2fc695088b7ad Mon Sep 17 00:00:00 2001 From: santoshkumarradha Date: Wed, 23 Sep 2026 14:37:19 -0400 Subject: [PATCH 1/8] Tests first: one run per batch, no silent fallback, an honest orphan row The tests, and only the seams they need (a slow or failing copy cut, a waiting hook, the run's root handed to a belt worker, the name of the run binding). No behaviour changes in this commit; every new test is expected to fail on it. --- internal/plandb/cli.go | 6 + internal/plandb/cli_run_test.go | 39 +++ internal/run/bashworker.go | 2 +- internal/run/plandb_own_run_test.go | 66 +++++ internal/session/bashbelt_worker.go | 10 +- internal/session/bashbelt_worker_test.go | 2 +- internal/session/plandb_plan.go | 7 +- internal/session/task_batch_run_test.go | 359 +++++++++++++++++++++++ internal/session/task_run_belt.go | 14 +- 9 files changed, 499 insertions(+), 6 deletions(-) create mode 100644 internal/plandb/cli_run_test.go create mode 100644 internal/run/plandb_own_run_test.go create mode 100644 internal/session/task_batch_run_test.go diff --git a/internal/plandb/cli.go b/internal/plandb/cli.go index dc7e9fd3ad..b94568b612 100644 --- a/internal/plandb/cli.go +++ b/internal/plandb/cli.go @@ -228,6 +228,12 @@ func cliRefusal(p *cliParsed) (string, bool) { return "", false } +// RunEnv names the run a worker belongs to, by its root task's id. The door +// that seats a run worker exports it beside PLANDB_DB, and a store found at +// that path whose root is ANOTHER run's is refused rather than written: a path +// says where a run's store was, and only the root says which run it is. +const RunEnv = "PLANDB_RUN" + // cliStore opens the run's store without being told where it is: --db, then // PLANDB_DB, then the first ancestor holding plandb.db or // .codeaf/plandb.db. One store per file; --project is accepted and checked diff --git a/internal/plandb/cli_run_test.go b/internal/plandb/cli_run_test.go new file mode 100644 index 0000000000..64e509fba8 --- /dev/null +++ b/internal/plandb/cli_run_test.go @@ -0,0 +1,39 @@ +package plandb + +import ( + "path/filepath" + "testing" +) + +// A WORKER'S `plandb` WRITES ONLY ITS OWN RUN'S STORE. The worker is bound to +// its store by path (PLANDB_DB) and to its run by the run's root (RunEnv). A +// store at that path whose root is another run's is the store a later request +// left there, and the CLI refuses it and writes nothing: a worker once filed +// four children and ten `done`s into the run beside its own. +func TestPlandbCliRefusesAnotherRunsStore(t *testing.T) { + db := filepath.Join(t.TempDir(), "plandb.db") + other, err := Open(db, "the other run", "8", "the other run", "") + if err != nil { + t.Fatal(err) + } + before := len(other.Tasks()) + _ = other.Close() + + h := cliNewHarness(t) + t.Setenv("PLANDB_DB", db) + t.Setenv(RunEnv, "1") + code := h.run("add", "not this run's work", "--as", "hijack") + cliWantError(t, h, code, "another run") + reread, err := Open(db, "", "", "", "") + if err != nil { + t.Fatal(err) + } + if got := len(reread.Tasks()); got != before { + t.Fatalf("the other run's store went from %d tasks to %d", before, got) + } + _ = reread.Close() + + // AND THE RUN'S OWN STORE IS WRITTEN AS EVER. + t.Setenv(RunEnv, "8") + cliWantCode(t, h.run("add", "this run's work", "--as", "own"), 0) +} diff --git a/internal/run/bashworker.go b/internal/run/bashworker.go index 6dca0bbc83..90d3aa1fae 100644 --- a/internal/run/bashworker.go +++ b/internal/run/bashworker.go @@ -100,7 +100,7 @@ func (w *BashWorker) Run(ctx context.Context, task plandb.Task) (Report, error) agent, err := session.NewBeltWorker(session.Config{ Workspace: w.workspace, Model: w.model, - }, w.completer, &task, w.store.Path()) + }, w.completer, &task, w.store.Path(), w.store.RootID()) if err != nil { return Report{}, err } diff --git a/internal/run/plandb_own_run_test.go b/internal/run/plandb_own_run_test.go new file mode 100644 index 0000000000..6988fe9eea --- /dev/null +++ b/internal/run/plandb_own_run_test.go @@ -0,0 +1,66 @@ +package run_test + +import ( + "context" + "os" + "path/filepath" + "testing" + + "github.com/Agent-Field/agentfield/sdk/go/ai" + "github.com/Agent-Field/codeaf/internal/plandb" + "github.com/Agent-Field/codeaf/internal/run" +) + +// A WORKER CANNOT WRITE INTO ANOTHER RUN'S STORE. The worker's run opened its +// store at one path; while it works, the store at that path is set aside and a +// different run's store is made in its place, which is what a second hand-off +// racing the first used to do. The worker's `plandb add` must not land in the +// store that is not its run's: its binding names the run, not only the path. +func TestBashWorkerCannotWriteIntoAnotherRunsStore(t *testing.T) { + t.Setenv("CODEAF_TASK_BELT", "bash") + t.Setenv("CODEAF_PLANDB_BIN", realPlandbDoor(t)) + store := runOpenStore(t) + path := store.Path() + if _, err := store.AddMany([]plandb.TaskSpec{leafDone("mine")}); err != nil { + t.Fatalf("add the leaf: %v", err) + } + if _, err := store.Claim("mine", "mine", "test-owner"); err != nil { + t.Fatalf("claim the leaf: %v", err) + } + const title = "Work filed into the wrong run" + seat := &seat{script: []step{ + func(context.Context, []ai.Message) (*ai.Response, error) { + // ANOTHER RUN TAKES THE PATH. The worker's own store is moved aside + // whole (its handle keeps working on the moved file) and a store with + // a different root is created where it was. + for _, suffix := range []string{"", "-wal", "-shm"} { + if err := os.Rename(path+suffix, path+".1"+suffix); err != nil && !os.IsNotExist(err) { + t.Errorf("set the worker's store aside: %v", err) + } + } + other, err := plandb.Open(path, "the other run", "other", "the other run", "") + if err != nil { + t.Errorf("open the other run's store: %v", err) + } else { + _ = other.Close() + } + return toolReply(bashArguments(t, `plandb add '`+title+`' --description 'not this run'`)), nil + }, + }} + worker := run.NewBashWorker(store, t.TempDir(), "test/model", seat) + _, _ = worker.Run(run.WithStepsPerTask(runContext(t), 3), *store.Task("mine")) + + other, err := plandb.Open(path, "", "", "", "") + if err != nil { + t.Fatalf("re-open the store at the path: %v", err) + } + defer other.Close() + if other.RootID() != "other" { + t.Fatalf("the store at the path is run %q, want the other run", other.RootID()) + } + for _, task := range other.Tasks() { + if task.Title == title { + t.Fatalf("the worker filed %q into the other run's store at %s", title, filepath.Base(path)) + } + } +} diff --git a/internal/session/bashbelt_worker.go b/internal/session/bashbelt_worker.go index d3cfd9b15b..b119e52f60 100644 --- a/internal/session/bashbelt_worker.go +++ b/internal/session/bashbelt_worker.go @@ -48,7 +48,13 @@ import ( // conversation's account-aware view ([Agent.beltRunCompleter]) and a test hands // a scripted one; nil is the road where nobody handed one and [New] builds the // real client itself. -func NewBeltWorker(config Config, completer Completer, task *plandb.Task, storePath string) (*Agent, error) { +// +// rootID IS THE RUN THE WORKER BELONGS TO, read off the run's own open handle +// and never off the file at storePath. The path is where the run's store WAS +// when the run opened it; the root is which run it is, and a worker's +// `plandb` refuses a store at that path whose root is another run's +// ([plandb.RunEnv]). Empty binds the path alone. +func NewBeltWorker(config Config, completer Completer, task *plandb.Task, storePath, rootID string) (*Agent, error) { if !bashBeltAsked() { return nil, errors.New("the bash belt is off: CODEAF_TASK_BELT names the node belt") } @@ -105,7 +111,7 @@ func NewBeltWorker(config Config, completer Completer, task *plandb.Task, storeP // back — and a shim that never landed is a seat that cannot run, because // every `plandb` its worker runs would resolve to whatever shares the // machine's PATH and write a plan this run would never read. - plan := &planState{path: storePath} + plan := &planState{path: storePath, root: rootID} if err := plan.armShim(); err != nil { _ = agent.Close() return nil, fmt.Errorf("arm the plandb shim: %w", err) diff --git a/internal/session/bashbelt_worker_test.go b/internal/session/bashbelt_worker_test.go index c963d0e374..8d8fee1f69 100644 --- a/internal/session/bashbelt_worker_test.go +++ b/internal/session/bashbelt_worker_test.go @@ -72,7 +72,7 @@ func newRunBeltWorker(t *testing.T, taskRung effort.Rung) *Agent { if taskRung.Valid() { config.Effort = taskRung } - agent, err := NewBeltWorker(config, &scriptedCompleter{}, task, store.Path()) + agent, err := NewBeltWorker(config, &scriptedCompleter{}, task, store.Path(), store.RootID()) if err != nil { t.Fatalf("NewBeltWorker: %v", err) } diff --git a/internal/session/plandb_plan.go b/internal/session/plandb_plan.go index fa6ecd4374..cd0ce0941d 100644 --- a/internal/session/plandb_plan.go +++ b/internal/session/plandb_plan.go @@ -42,7 +42,12 @@ type planState struct { // chat is the conversation's tag: the session folder's own name, stamped on // every row the seed makes so the plan can be read back as this chat's // (PlanTasks). It is settled with the path at the seed and never moves. - chat string + chat string + // root is the run a worker's plan belongs to, set only on a run worker's + // own plan ([NewBeltWorker]) from the run's open handle. It is what the + // worker's `plandb` checks the store at path against ([plandb.RunEnv]), so + // a later store at the same path cannot take the worker's writes. + root string shimmed bool // archives holds read handles for ended stores. Ended stores are immutable, // so each is opened at most once for the life of this conversation. diff --git a/internal/session/task_batch_run_test.go b/internal/session/task_batch_run_test.go new file mode 100644 index 0000000000..104bb84d98 --- /dev/null +++ b/internal/session/task_batch_run_test.go @@ -0,0 +1,359 @@ +package session + +// ONE RUN PER BATCH. A message that proposes several tasks, all approved at +// once, commits every hand-off at the same moment. They used to race to open +// the conversation's one store: each one found no live run, each one opened +// (or set aside) the same plandb.db, two of them started runs of their own and +// the rest fell back without a word to the older engine's tree. These tests +// hold the three laws that replaced it: the batch is one run with every +// hand-off in it, a run road that fails says so rather than becoming a node +// of the older tree, and a run row nothing drives still answers and can be +// stopped. + +import ( + "context" + "errors" + "fmt" + "os" + "path/filepath" + "strconv" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/Agent-Field/codeaf/internal/plandb" +) + +// batchRunDouble is a run engine that can be started more than once, because +// the defect under test is exactly that it WAS: every Start is counted, and +// each holds until the test lets the runs go. +type batchRunDouble struct { + mu sync.Mutex + starts int + release chan struct{} +} + +func newBatchRunDouble() *batchRunDouble { + return &batchRunDouble{release: make(chan struct{})} +} + +func (d *batchRunDouble) Start(ctx context.Context, spec RunSpec) RunSummary { + d.mu.Lock() + d.starts++ + d.mu.Unlock() + select { + case <-d.release: + case <-ctx.Done(): + } + if spec.Store != nil { + _ = spec.Store.CompleteRoot("done") + } + return RunSummary{Outcome: beltRunOutcomeDone, Result: "done"} +} + +func (d *batchRunDouble) Land(context.Context, *plandb.Store, string, string) (RunLanding, error) { + return RunLanding{}, nil +} + +func (d *batchRunDouble) started() int { + d.mu.Lock() + defer d.mu.Unlock() + return d.starts +} + +// batchAgent is a conversation on the bash belt with its session folder at +// dir, a fixed clock, and an older-engine runner that runs nothing, so a node +// admitted by a fallback is a node the test can count and never a worker. +func batchAgent(t *testing.T, dir string) *Agent { + t.Helper() + agent, _ := newTestAgent(t, beltRunCompleter{text: "done"}, func(config *Config) { + config.Workspace = newTestRepo(t) + config.Place = Place{Dir: dir} + config.AskConsent = false + config.TaskAutoApproveSeconds = 0 + }) + clock := &fakeClock{at: time.Date(2026, time.September, 23, 9, 0, 0, 0, time.UTC)} + agent.taskNow = clock.now + agent.graph().run = func(*TaskNode) {} + return agent +} + +// stageApproved puts one proposal on the card and answers it yes, without +// committing it, so a test can commit a whole batch at the same moment. +func stageApproved(t *testing.T, agent *Agent, title string) *stagedProposal { + t.Helper() + staged := agent.stageTask(context.Background(), beltProposalArgs(title, "the part is done")) + proposal, ok := staged.(*stagedProposal) + if !ok { + answer, _, _ := staged.Commit(context.Background()) + t.Fatalf("the proposal %q was not staged: %q", title, answer) + } + agent.ResolveTask(proposal.id, TaskAnswer{Approved: true}) + return proposal +} + +// endBatchRun lets every run the double holds go, and waits for the +// conversation's run to be over. +func endBatchRun(t *testing.T, agent *Agent, double *batchRunDouble) { + t.Helper() + close(double.release) + beltRunWaitFor(t, "the run to end", func() bool { + agent.beltMu.Lock() + defer agent.beltMu.Unlock() + return agent.beltRun == nil + }) +} + +// EIGHT HAND-OFFS APPROVED AT ONCE ARE ONE RUN. The first to arrive opens the +// run, and its working copy is slow to cut: that cut is the window the batch +// used to race through. Every other hand-off waits for the run to exist and +// joins it as a child, exactly as a hand-off made a minute later would, so the +// store holds all eight, one engine is started, nothing is set aside and no +// hand-off becomes a node of the older tree. +func TestEightHandoffsApprovedAtOnceAreOneRun(t *testing.T) { + t.Setenv("CODEAF_TASK_BELT", "bash") + double := newBatchRunDouble() + registerBeltRunEngine(t, double) + dir := t.TempDir() + agent := batchAgent(t, dir) + + slow := make(chan struct{}) + var cuts atomic.Int32 + previous := beltRunPrepare + beltRunPrepare = func(ctx context.Context, place Place, workspace, session string, id uint64, title string, stand taskStand) (taskTree, error) { + if cuts.Add(1) == 1 { + <-slow + } + return previous(ctx, place, workspace, session, id, title, stand) + } + t.Cleanup(func() { beltRunPrepare = previous }) + var waiting atomic.Int32 + beltStartWaits = func() { waiting.Add(1) } + t.Cleanup(func() { beltStartWaits = nil }) + + const batch = 8 + proposals := make([]*stagedProposal, batch) + for i := range proposals { + proposals[i] = stageApproved(t, agent, fmt.Sprintf("Part %d of the batch", i+1)) + } + type answer struct { + text string + failed bool + err error + } + answers := make([]answer, batch) + var returned atomic.Int32 + var wg sync.WaitGroup + for i, proposal := range proposals { + wg.Add(1) + go func() { + defer wg.Done() + text, failed, err := proposal.Commit(context.Background()) + answers[i] = answer{text, failed, err} + returned.Add(1) + }() + } + // The first cut holds until every other hand-off has either come to wait + // for the run it is starting, or gone on without it. + beltRunWaitFor(t, "the rest of the batch to wait on the first run's start", func() bool { + return waiting.Load() == batch-1 || returned.Load() == batch-1 + }) + close(slow) + wg.Wait() + + for i, got := range answers { + if got.err != nil || got.failed { + t.Errorf("hand-off %d answered failed=%v err=%v: %q", proposals[i].id, got.failed, got.err, got.text) + } + } + for _, proposal := range proposals { + if agent.graph().node(proposal.id) != nil { + t.Errorf("hand-off %d became a node of the older tree", proposal.id) + } + } + store := beltRunStoreAt(t, dir) + defer store.Close() + root := store.RootID() + joined := 0 + for _, proposal := range proposals { + id := strconv.FormatUint(proposal.id, 10) + task := store.Task(id) + switch { + case task == nil: + t.Errorf("the run's store holds no task %s", id) + case id == root: + case task.ParentID != root: + t.Errorf("task %s hangs under %q, want the run's root %s", id, task.ParentID, root) + default: + joined++ + } + } + if joined != batch-1 { + t.Errorf("%d hand-offs joined the run, want %d", joined, batch-1) + } + if archives := planArchivePaths(filepath.Join(dir, planStoreFilename)); len(archives) != 0 { + t.Errorf("the batch set %d stores aside, want none: %v", len(archives), archives) + } + beltRunWaitFor(t, "the run's engine to start", func() bool { return double.started() >= 1 }) + if got := double.started(); got != 1 { + t.Errorf("the batch started %d runs, want one", got) + } + endBatchRun(t, agent, double) +} + +// A HAND-OFF WHOSE RUN ROAD FAILS IS NOT A SILENT NODE. The copy would not cut +// (a disk that would not answer, here), and the hand-off used to fall through +// to the older engine's tree and answer `task N started` as if nothing had +// happened. It now says it did not start and why, on both doors, and reads as +// a failure rather than as the success it is not. +func TestAHandoffWhoseRunRoadFailsIsNotASilentNode(t *testing.T) { + t.Setenv("CODEAF_TASK_BELT", "bash") + double := newBatchRunDouble() + registerBeltRunEngine(t, double) + dir := t.TempDir() + agent := batchAgent(t, dir) + previous := beltRunPrepare + beltRunPrepare = func(context.Context, Place, string, string, uint64, string, taskStand) (taskTree, error) { + return taskTree{}, errors.New("disk I/O error") + } + t.Cleanup(func() { beltRunPrepare = previous }) + + proposal := stageApproved(t, agent, "Break the run road") + answer, failed, err := proposal.Commit(context.Background()) + if err != nil { + t.Fatalf("Commit: %v", err) + } + if !failed { + t.Errorf("the failed hand-off reads as a success: %q", answer) + } + if strings.Contains(answer, "started") || !strings.Contains(answer, "did not start") || !strings.Contains(answer, "disk I/O error") { + t.Errorf("the failed hand-off's receipt = %q, want it to say it did not start and why", answer) + } + if agent.graph().node(proposal.id) != nil { + t.Errorf("the failed hand-off became a node of the older tree") + } + + id, _, _, err := agent.StartTask(context.Background(), "break the typed road", false) + if err == nil || !strings.Contains(err.Error(), "did not start") || !strings.Contains(err.Error(), "disk I/O error") { + t.Errorf("the typed /task answered id=%d err=%v, want it to say it did not start and why", id, err) + } + if id != 0 && agent.graph().node(id) != nil { + t.Errorf("the typed /task became a node of the older tree") + } + if double.started() != 0 { + t.Errorf("a run road that failed started the engine") + } +} + +// A RUN ROW NOTHING DRIVES STILL ANSWERS, AND STOPS. A row the rail shows +// running, whose run is not the one this conversation is driving, used to open +// a page whose box answered `no task 1 in this session` and whose stop said +// nothing. It now says what the row is and that it can be stopped, and a stop +// settles it. +func TestARunRowNothingDrivesAnswersAMessageAndStops(t *testing.T) { + t.Setenv("CODEAF_TASK_BELT", "bash") + registerBeltRunEngine(t, newBatchRunDouble()) + agent := batchAgent(t, t.TempDir()) + g := agent.graph() + id := g.reserve() + agent.publishRunRow(g, TaskNotice{ + ID: id, Title: "orphaned run", State: TaskRunning, StartedAt: agent.taskClockNow(), + PlanTask: planStoreID(strconv.FormatUint(id, 10)), + }) + + _, err := agent.SteerTask(id, "can you hear me") + if err == nil { + t.Fatal("a message to a run row nothing drives was taken as delivered") + } + if strings.Contains(err.Error(), "in this session") || !strings.Contains(err.Error(), "stop") { + t.Errorf("the message was answered %q, want what the row is and that it can be stopped", err) + } + line, err := agent.Cancel(CancelTask + ":" + strconv.FormatUint(id, 10)) + if err != nil || !strings.Contains(line, "stopped") { + t.Fatalf("the stop answered %q, %v; want it stopped", line, err) + } + rows := g.runRows(id) + if len(rows) != 1 || rows[0].State == TaskRunning || !rows[0].Stopped || rows[0].EndedAt.IsZero() { + t.Fatalf("the stopped row = %+v, want it settled as stopped", rows) + } +} + +// A MESSAGE TO A LIVE RUN'S ROW REACHES ITS TASK, and a joined row whose store +// task is missing answers and stops like the orphan above. The run's rows are +// not nodes of the older tree, so the room's box used to answer every one of +// them `no task N in this session`. +func TestALiveRunsRowsTakeAMessageAndAMissingOneStops(t *testing.T) { + t.Setenv("CODEAF_TASK_BELT", "bash") + double := newBatchRunDouble() + registerBeltRunEngine(t, double) + dir := t.TempDir() + agent := batchAgent(t, dir) + stand := taskStand{dir: agent.config.Workspace, mode: TaskModeWorktree} + if err := agent.startKnownTaskRun(context.Background(), 61, "the run", "brief", nil, stand, ""); err != nil { + t.Fatalf("start the run: %v", err) + } + if err := agent.startKnownTaskRun(context.Background(), 62, "a joined part", "brief", nil, stand, ""); err != nil { + t.Fatalf("join the run: %v", err) + } + + receipt, err := agent.SteerTask(62, "use the new schema") + if err != nil || receipt.Landing == "" { + t.Fatalf("a message to a joined row answered %+v, %v; want it delivered", receipt, err) + } + store := beltRunStoreAt(t, dir) + said := false + for _, note := range store.Notes("62", 0) { + said = said || strings.Contains(note.Body, "use the new schema") + } + _ = store.Close() + if !said { + t.Fatal("the message is not on the joined task's page") + } + + // A row the run holds whose store task is gone. + g := agent.graph() + agent.beltMu.Lock() + agent.beltRun.joined = append(agent.beltRun.joined, 63) + agent.beltMu.Unlock() + agent.publishRunRow(g, TaskNotice{ID: 63, Title: "lost part", State: TaskRunning, Parent: 61, StartedAt: agent.taskClockNow()}) + if _, err := agent.SteerTask(63, "hello"); err == nil || strings.Contains(err.Error(), "in this session") || !strings.Contains(err.Error(), "stop") { + t.Errorf("a message to a joined row with no store task answered %v, want what it is and that it can be stopped", err) + } + line, err := agent.Cancel(CancelTask + ":63") + if err != nil || !strings.Contains(line, "stopped") { + t.Fatalf("the stop answered %q, %v; want it stopped", line, err) + } + if rows := g.runRows(63); len(rows) != 1 || rows[0].State == TaskRunning || !rows[0].Stopped { + t.Fatalf("the stopped row = %+v, want it settled as stopped", rows) + } + endBatchRun(t, agent, double) +} + +// A RUN WORKER IS BOUND TO ITS OWN RUN, NOT ONLY TO A PATH. Its commands carry +// the run's root beside the store's path, so a later store at the same path +// cannot take its writes ([plandb.RunEnv]). +func TestABeltWorkerIsBoundToItsOwnRun(t *testing.T) { + t.Setenv("HOME", t.TempDir()) + t.Setenv("CODEAF_TASK_BELT", "bash") + stub := filepath.Join(t.TempDir(), "stub-codeaf") + if err := os.WriteFile(stub, []byte("#!/bin/sh\nexit 0\n"), 0o755); err != nil { + t.Fatal(err) + } + t.Setenv(planCLIBinEnv, stub) + store, err := plandb.Open(filepath.Join(t.TempDir(), planStoreFilename), "the work", "7", "the run", "") + if err != nil { + t.Fatal(err) + } + defer store.Close() + agent, err := NewBeltWorker(Config{Workspace: t.TempDir(), Model: "test/model"}, &scriptedCompleter{}, store.Task("7"), store.Path(), store.RootID()) + if err != nil { + t.Fatalf("NewBeltWorker: %v", err) + } + defer agent.Close() + got := agent.planCommand("plandb status") + if want := plandb.RunEnv + "=" + quoteShWord("7"); !strings.Contains(got, want) { + t.Fatalf("the worker's command %q does not carry its run %q", got, want) + } +} diff --git a/internal/session/task_run_belt.go b/internal/session/task_run_belt.go index 4440116182..783b605030 100644 --- a/internal/session/task_run_belt.go +++ b/internal/session/task_run_belt.go @@ -339,7 +339,7 @@ func (a *Agent) startKnownTaskRun(ctx context.Context, id uint64, title, brief s return err } } - tree, err := prepareTaskTreeOn(ctx, a.config.Place, a.config.Workspace, a.journalID(), id, title, stand) + tree, err := beltRunPrepare(ctx, a.config.Place, a.config.Workspace, a.journalID(), id, title, stand) if err != nil { _ = store.Close() return err @@ -378,6 +378,18 @@ func (a *Agent) startKnownTaskRun(ctx context.Context, id uint64, title, brief s return nil } +// beltRunPrepare cuts a run's working copy. It is [prepareTaskTreeOn] in the +// product, and a variable only so a test can make the cut slow or make it fail: +// the slow cut is the window a batch of simultaneous hand-offs used to race +// through, and a failed cut is a run road that did not open. +var beltRunPrepare = prepareTaskTreeOn + +// beltStartWaits is a test's observation point: it is called when a hand-off +// finds another hand-off in the middle of starting the conversation's run and +// is about to wait for it. Nil outside tests, and nothing in the product reads +// it. +var beltStartWaits func() + // beltJoinWaits is a test's observation point: it is called when a hand-off has // met a run on its way out and is about to wait for it to be over. Nil outside // tests, and nothing in the product reads it. From 7807751ae793de0fc16cac6a6716894cd4b2f555 Mon Sep 17 00:00:00 2001 From: santoshkumarradha Date: Wed, 23 Sep 2026 14:42:12 -0400 Subject: [PATCH 2/8] One run per batch: approved hand-offs join one run, never race to one store Starting a run is now one critical section, from the look for a live run until the run is registered, and every other hand-off waits at its door and then joins the run as a child. A batch approved at one moment is one run with every hand-off in it. A run road that fails says "task N did not start: " on both the proposal door and the typed /task, reads as a failure, and starts nothing on the older engine. Only a missing run engine or plan place still takes the older road. A worker is bound to its run's root as well as its store path (PLANDB_RUN), and the plandb CLI refuses a store at that path whose root is another run's. A run row's room takes a message as a note on its task's page. A run row nothing drives answers that it can be stopped instead of "no task N in this session", and a stop settles it. --- internal/manual/chat/worker-harness.md | 44 ++++++++- internal/plandb/cli.go | 8 ++ internal/plandb/testmain_test.go | 1 + internal/run/testmain_test.go | 1 + internal/session/agent.go | 6 ++ internal/session/plandb_plan.go | 13 ++- internal/session/session.go | 6 ++ internal/session/stoprun.go | 105 +++++++++++++++++++- internal/session/task.go | 24 ++++- internal/session/task_room.go | 7 ++ internal/session/task_run_belt.go | 130 +++++++++++++++++++++---- internal/session/task_run_continue.go | 12 ++- 12 files changed, 318 insertions(+), 39 deletions(-) diff --git a/internal/manual/chat/worker-harness.md b/internal/manual/chat/worker-harness.md index a59cfa7b85..82c2212dda 100644 --- a/internal/manual/chat/worker-harness.md +++ b/internal/manual/chat/worker-harness.md @@ -148,6 +148,41 @@ and the store is kept beside the new one, readable with the earlier runs. Its ro read `interrupted`, never `running`. Carrying an interrupted run on is not possible from any surface today. +## I approved several tasks at once — are they one run? Why a task says it did not start + +**Yes: tasks approved together are one run.** When the chat proposes several tasks +in one message and they are all approved at the same moment, the first one to start +opens the run and every other waits the moment that takes, then joins it as a child +of the run's own task, exactly as a task handed off a minute later would. A batch +never opens a second run beside the first and never sets the first run's plan +aside, and none of it starts on the older engine instead. + +**A task whose run could not start says so, and nothing else starts.** If the run's +plan could not be opened or its copy could not be cut, the answer is +`task N did not start: . Nothing is running for it and nothing was +started in its place; propose it again, or tell the person what stopped it.` A typed +`/task` answers the same sentence. It reads as a failure, never as `task N started`, +and the task is not quietly put on the older engine's tree. Only a build with no run +engine at all, or a conversation with nowhere to keep a plan, uses the older engine, +because there the run road was never there to take. + +**A worker writes only its own run's plan.** A run's worker is bound to its run, not +only to where its plan was. If another run's plan is ever found in that place, the +worker's `plandb` refuses it: `the plan store at is another run's (t-), not this worker's run (t-), so nothing was read or written`. + +## A run's row says running but nothing answers — a message to a run's task, stopping a row nothing drives + +A message typed in the room of a run's row is left as a note on that task's page, +and the room says `left on the task's page — its worker reads it between steps`. + +A run row that nothing drives any more, because its run is not the one this +conversation is driving or its plan holds no such task, answers a message with +`nothing is driving this task any more, so no worker can read a message; stop it to +clear the row`. It never answers `no task N in this session` while the side list +draws it running. Stopping it settles the row as `stopped`, and the stop answers +`stopped task N () — nothing was driving it any more`. + ## Why is this task indented under that one? The pane draws the run's **plan as a tree, not a flat list**. A task sits under the @@ -751,7 +786,8 @@ admitted, and where `codeaf do` chooses its road — and **with one of those thr words set, not one byte of any prompt, belt or landing moves from the older road**. -Everything behind the switch is a seam. A build with no run engine linked answers -the older road, and every refusal on the run road falls back to it rather than -inventing a sentence of its own — so a conversation the run road cannot serve gets -exactly the door it always had. +Everything behind the switch is a seam. A build with no run engine linked, or a +conversation with nowhere to keep a plan, answers the older road — so a conversation +the run road cannot serve at all gets exactly the door it always had. A run road +that was there and failed does NOT fall back: the task answers `task N did not +start: <the reason>` and nothing is started on the older engine in its place. diff --git a/internal/plandb/cli.go b/internal/plandb/cli.go index b94568b612..422b24aca2 100644 --- a/internal/plandb/cli.go +++ b/internal/plandb/cli.go @@ -259,6 +259,14 @@ func cliStore(p *cliParsed) (*Store, error) { if want := p.vals["project"]; want != "" && st.Project() != want { return nil, fmt.Errorf("the plan store at %s belongs to project %q, not %q", path, st.Project(), want) } + // A STORE THAT IS ANOTHER RUN'S IS REFUSED WHOLE, reads and writes alike: a + // worker reading another run's plan would plan against work that is not its + // own, and one writing it filed its children under the other run's root. + if want := os.Getenv(RunEnv); want != "" && st.RootID() != want { + root := st.RootID() + _ = st.Close() + return nil, fmt.Errorf("the plan store at %s is another run's (t-%s), not this worker's run (t-%s), so nothing was read or written; this worker's run is over or was set aside", path, root, want) + } return st, nil } diff --git a/internal/plandb/testmain_test.go b/internal/plandb/testmain_test.go index a5ef3cfbe0..829e14579c 100644 --- a/internal/plandb/testmain_test.go +++ b/internal/plandb/testmain_test.go @@ -10,5 +10,6 @@ import ( // behavior they mean to exercise; the ambient run store is never a fixture. func TestMain(m *testing.M) { _ = os.Unsetenv("PLANDB_DB") + _ = os.Unsetenv("PLANDB_RUN") os.Exit(m.Run()) } diff --git a/internal/run/testmain_test.go b/internal/run/testmain_test.go index 97a6bbc601..b402cd1519 100644 --- a/internal/run/testmain_test.go +++ b/internal/run/testmain_test.go @@ -9,5 +9,6 @@ import ( // launched go test. Tests that exercise the bound door set PLANDB_DB themselves. func TestMain(m *testing.M) { _ = os.Unsetenv("PLANDB_DB") + _ = os.Unsetenv("PLANDB_RUN") os.Exit(m.Run()) } diff --git a/internal/session/agent.go b/internal/session/agent.go index d858765eb3..febbd1cd0c 100644 --- a/internal/session/agent.go +++ b/internal/session/agent.go @@ -1600,6 +1600,12 @@ const ( // sent, and the sentence says that rather than reporting a second delivery // that did not happen. steerAgainWord = "already on the task's record from the same message — nothing was sent a second time" + // steerRunNoteWord is a line said to a run's own row. A run's task has no + // worker to splice into; its worker reads the notes on its task's page + // between its steps, so the line is left there, and the sentence says when + // it is read rather than claiming it arrived now (stoprun.go's + // [Agent.sayToRunRow]). + steerRunNoteWord = "left on the task's page — its worker reads it between steps" ) // steerRecord is what the JOURNAL keeps about this line when it is a correction diff --git a/internal/session/plandb_plan.go b/internal/session/plandb_plan.go index cd0ce0941d..421b486f9a 100644 --- a/internal/session/plandb_plan.go +++ b/internal/session/plandb_plan.go @@ -738,7 +738,18 @@ func (g *TaskGraph) planBashPrefix() string { if bin == "" { return "" } - return "export PATH=" + quoteShWord(bin) + ":$PATH PLANDB_DB=" + quoteShWord(plan.path) + "; " + prefix := "export PATH=" + quoteShWord(bin) + ":$PATH PLANDB_DB=" + quoteShWord(plan.path) + // AND THE RUN IS BOUND, NOT ONLY THE PATH. A path says where the run's store + // was when the run opened it; a later request can set that store aside and + // seed another at the same path, and a worker bound by the path alone then + // wrote its children and its `done`s into a run that was not its own + // (measured on the owner's session: four children and ten `done`s). The + // root names which run this is, and the CLI refuses a store at the path + // whose root is another's ([plandb.RunEnv]). + if plan.root != "" { + prefix += " " + plandb.RunEnv + "=" + quoteShWord(plan.root) + } + return prefix + "; " } // planCLIBinEnv is the resolver's one override: it names a binary that diff --git a/internal/session/session.go b/internal/session/session.go index b39b96f329..15e5922c26 100644 --- a/internal/session/session.go +++ b/internal/session/session.go @@ -3240,6 +3240,12 @@ type Agent struct { // held. beltMu sync.Mutex beltRun *beltRun + // beltStartMu is the start lock: it is held from a hand-off's look for a + // live run until the run it opens is registered on beltRun, so a batch of + // hand-offs committed at one moment is one run and never several racing to + // one store ([Agent.lockBeltStart]). It is never taken while beltMu is + // held; beltMu is taken inside it. + beltStartMu sync.Mutex // taskAnswers is the proposals a person owes an answer to, keyed by the id // the EventTaskProposal carried. It is consent's pending-id machinery for a // question whose CLOCK can be held: the wait ends on an answer, on an active diff --git a/internal/session/stoprun.go b/internal/session/stoprun.go index 8f1e4b2a98..ffad50f8ea 100644 --- a/internal/session/stoprun.go +++ b/internal/session/stoprun.go @@ -40,8 +40,13 @@ package session // run that is stopping says so, and a press on a run that is over says that. import ( + "errors" + "fmt" + "slices" "strconv" "strings" + + "github.com/Agent-Field/codeaf/internal/plandb" ) // beltStoppedWhere is how a stopped run says where its work is and what a @@ -67,7 +72,7 @@ func (a *Agent) stopBeltRow(id uint64, why string) (string, bool, error) { run := a.beltRun if run == nil { a.beltMu.Unlock() - return a.endedBeltRow(id) + return a.endedBeltRow(id, why) } if run.row == id { name := taskStopName(id, run.title) @@ -98,7 +103,7 @@ func (a *Agent) stopBeltRow(id uint64, why string) (string, bool, error) { } a.beltMu.Unlock() if !joined { - return a.endedBeltRow(id) + return a.endedBeltRow(id, why) } return a.stopJoinedRow(run, id, why) } @@ -159,7 +164,13 @@ func (a *Agent) stopJoinedRow(run *beltRun, id uint64, why string) (string, bool title = task.Title } name := taskStopName(id, title) - if task == nil || terminalStoreStatus(task.Status) { + if task == nil { + // A JOINED ROW WHOSE STORE TASK IS GONE is a row nothing can move, and + // answering "already finished" over a row the rail draws running is the + // same row saying two things. It is settled as stopped instead. + return a.stopUndrivenRow(id, run.row, why) + } + if terminalStoreStatus(task.Status) { return name + " has already finished; there is nothing to stop", true, nil } if _, err := run.store.Cancel(key, stopBecause(taskStoppedWord, why)); err != nil { @@ -183,19 +194,105 @@ func (a *Agent) stopJoinedRow(run *beltRun, id uint64, why string) (string, bool // endedBeltRow answers a stop on a run's row whose run is already over. The // row is still on the person's screen, so the press is a real one and is owed // the sentence every settled task answers, not the graph's "there is no task". -func (a *Agent) endedBeltRow(id uint64) (string, bool, error) { +func (a *Agent) endedBeltRow(id uint64, why string) (string, bool, error) { g := a.graph() if g == nil { return "", false, nil } for _, kept := range g.runRows(id) { if kept.ID == id && kept.Run == "" { + // A ROW STILL SAYING RUNNING WITH NO RUN BEHIND IT is not finished, + // and the stop is what clears it ([Agent.stopUndrivenRow]). + if kept.State == TaskRunning { + return a.stopUndrivenRow(id, kept.Parent, why) + } return taskStopName(id, kept.Title) + " has already finished; there is nothing to stop", true, nil } } return "", false, nil } +// runRowUndrivenWord is the one sentence a run row with nothing behind it +// answers a message with: the run that published it is not the one this +// conversation is driving, or the plan it names holds no such task, so no +// worker can read the words. It says the one thing a person can do about it. +const runRowUndrivenWord = "nothing is driving this task any more, so no worker can read a message; stop it to clear the row" + +// stopUndrivenRow settles a run row nothing drives: a row the rail draws +// running whose run is gone, or whose task the run's plan does not hold. THERE +// IS NO WORK TO CUT, so the stop is the row's ending and nothing else, written +// where every run row's ending is written ([Agent.publishRunRow]) so the rail +// and the conversation read back tomorrow agree that it stopped. +func (a *Agent) stopUndrivenRow(id, parent uint64, why string) (string, bool, error) { + g := a.graph() + notice := TaskNotice{ID: id, Parent: parent} + for _, kept := range g.runRows(id) { + if kept.ID == id { + notice = kept + } + } + notice.State, notice.Stopped = TaskFailed, true + notice.Report = stopBecause(taskStoppedWord, why) + " · nothing was driving it any more" + notice.EndedAt = a.taskClockNow() + if g != nil { + a.publishRunRow(g, notice) + } else { + a.emitTaskUpdate(notice) + } + return "stopped " + stopBecause(taskStopName(id, notice.Title), why) + " — nothing was driving it any more", true, nil +} + +// sayToRunRow is a message to a row a run owns, and it reports whether the +// number is a run's at all: false sends the caller on to the answer that there +// is no such task. +// +// A RUN'S ROWS ARE NOT NODES, so the node door ([Agent.sayToTask]) had nothing +// to deliver to and answered every one of them `no task N in this session`, +// over a row the rail was drawing running. The live run's own rows take the +// words as a note on their task's page, which is where a run's worker reads +// what it is told between its steps (the page's own box writes the same note, +// [Agent.PlanNote]). A row nothing drives says so, and says it can be stopped. +func (a *Agent) sayToRunRow(id uint64, text string, origin messageOrigin) (SteerReceipt, bool, error) { + a.beltMu.Lock() + run := a.beltRun + owned := run != nil && (run.row == id || slices.Contains(run.joined, id)) + a.beltMu.Unlock() + if owned { + key := strconv.FormatUint(id, 10) + task := run.store.Task(key) + if task == nil { + return SteerReceipt{}, true, errors.New(taskStopName(id, "") + ": " + runRowUndrivenWord) + } + if terminalStoreStatus(task.Status) { + return SteerReceipt{}, true, fmt.Errorf("%s has finished, not running", taskStopName(id, task.Title)) + } + var err error + if origin == fromPerson { + _, err = run.store.AddPersonNote(key, text) + } else { + _, err = run.store.AddNote(key, plandb.NoteAgentChat, text) + } + if err != nil { + return SteerReceipt{}, true, err + } + return SteerReceipt{Landing: steerRunNoteWord}, true, nil + } + g := a.graph() + if g == nil { + return SteerReceipt{}, false, nil + } + for _, kept := range g.runRows(id) { + if kept.ID != id || kept.Run != "" { + continue + } + if kept.State == TaskRunning { + return SteerReceipt{}, true, errors.New(taskStopName(id, kept.Title) + ": " + runRowUndrivenWord) + } + return SteerReceipt{}, true, fmt.Errorf("%s is %s, not running", taskStopName(id, kept.Title), kept.State) + } + return SteerReceipt{}, false, nil +} + // beltRunStopped reports whether a person stopped this run, and the words they // gave for it. func (a *Agent) beltRunStopped(run *beltRun) (bool, string) { diff --git a/internal/session/task.go b/internal/session/task.go index 457392cc04..0e0292aef1 100644 --- a/internal/session/task.go +++ b/internal/session/task.go @@ -811,9 +811,16 @@ func (p *stagedProposal) Commit(ctx context.Context) (string, bool, error) { // and its depends_on as the store's own dependencies, and takes the person's // ask with it when this turn owes one (CHAT-ROLE.md, "A landing speaks only // when an answer is owed"). A task about ANOTHER FOLDER than the work - // already underway is refused here ([standsElsewhereError]); any other - // failure of the run road falls through to the shipped engine, exactly as a - // typed /task does, and that engine cuts its own copy from the same stand. + // already underway is refused here ([standsElsewhereError]). + // + // AND A RUN ROAD THAT FAILS IS SAID, NOT HIDDEN. Any other failure of the run + // road used to fall through to the older engine's tree, which is the one + // thing this comment's first line says an approved hand-off never becomes: + // a batch of eight approved at once raced to one store and six of them + // quietly became old-tree nodes. The receipt now says the task did not start + // and why ([runDidNotStart]), and reads as a failure. Only a run road that + // is not there at all (no engine linked, no place for a store) leaves this + // door for the older one. if bashBeltAsked() && chatRunEngine != nil && !a.config.InTask { a.mu.Lock() question := questionAtTaskHandoff(a.owedAsks) @@ -830,8 +837,12 @@ func (p *stagedProposal) Commit(ctx context.Context) (string, bool, error) { // turn's cancellation here without either of those is what left a run's // life belonging to the PROCESS, and a run whose room had closed went on // spending with nobody able to read it or stop it. - joined := a.beltRunStandsOn(p.stand) - err := a.startKnownTaskRun(context.WithoutCancel(ctx), p.id, spec.title, description, spec.dependsOn, p.stand, question) + // + // WHETHER IT JOINED is the start door's answer and not a look taken + // before it: in a batch committed at one moment none of the hand-offs + // could see a live run beforehand, and the one that opened the run is + // decided under the start lock ([Agent.startOrJoinTaskRun]). + joined, err := a.startOrJoinTaskRun(context.WithoutCancel(ctx), p.id, spec.title, description, spec.dependsOn, p.stand, question) if refusal := (standsElsewhereError{}); errors.As(err, &refusal) { return refusal.Error(), true, nil } @@ -842,6 +853,9 @@ func (p *stagedProposal) Commit(ctx context.Context) (string, bool, error) { } return receipt, false, nil } + if !errors.Is(err, errRunRoadUnavailable) { + return withElsewhere(runDidNotStart(p.id, err), elsewhere), true, nil + } } state := graph.admit(p.id, spec) admitted = true diff --git a/internal/session/task_room.go b/internal/session/task_room.go index 8275480ab6..672b4ee928 100644 --- a/internal/session/task_room.go +++ b/internal/session/task_room.go @@ -362,6 +362,13 @@ func (a *Agent) sayToTask(id uint64, text string, origin messageOrigin, source s } node := a.taskNode(id) if node == nil { + // A RUN'S ROWS WEAR TASK NUMBERS AND ARE NOT NODES, so the run is asked + // before the answer that there is no such task (stoprun.go's + // [Agent.sayToRunRow]). A row the rail draws running must never be + // answered as though it did not exist. + if receipt, owned, err := a.sayToRunRow(id, text, origin); owned { + return receipt, err + } return SteerReceipt{}, fmt.Errorf("no task %d in this session", id) } // WHAT THIS TASK ALREADY HOLDS IS ASKED BEFORE WHETHER IT IS STILL RUNNING, diff --git a/internal/session/task_run_belt.go b/internal/session/task_run_belt.go index 783b605030..8a0bc9b0ef 100644 --- a/internal/session/task_run_belt.go +++ b/internal/session/task_run_belt.go @@ -245,9 +245,10 @@ type beltRun struct { // startTaskRun is StartTask's second road, taken whenever the bash belt is asked // for and a run engine is linked. It seeds or reuses the conversation's store, // adds this brief's work to it, publishes the row a surface draws, and starts -// the engine in a goroutine the moment the run is new. Every refusal falls back -// to the legacy road rather than inventing a sentence of its own, so a -// conversation the run road cannot serve gets exactly the door it always had. +// the engine in a goroutine the moment the run is new. A conversation the run +// road cannot serve at all (no engine linked, no place for a store) gets +// exactly the door it always had; a run road that was there and failed says so +// ([runDidNotStart]) and starts nothing on another engine. func (a *Agent) startTaskRun(ctx context.Context, brief string, solo bool, question string) (uint64, string, string, error) { engine := chatRunEngine g := a.graph() @@ -263,11 +264,43 @@ func (a *Agent) startTaskRun(ctx context.Context, brief string, solo bool, quest title := taskPersonTitle(brief) stand := taskStand{dir: a.config.Workspace, mode: TaskModeWorktree} if err := a.startKnownTaskRun(ctx, id, title, brief, nil, stand, question); err != nil { - return a.startTaskLegacy(ctx, brief, solo) + if errors.Is(err, errRunRoadUnavailable) { + return a.startTaskLegacy(ctx, brief, solo) + } + // A RUN ROAD THAT OPENED AND THEN FAILED IS SAID, NEVER HIDDEN. It used + // to fall through to the older engine's tree here, so a store that would + // not open or a copy that would not cut turned the person's task into a + // node of a different engine with nothing on the screen saying so + // ([runDidNotStart] is the same sentence the proposal door answers). + if refusal := (standsElsewhereError{}); errors.As(err, &refusal) { + return 0, "", "", refusal + } + return 0, "", "", errors.New(runDidNotStart(id, err)) } return id, title, "", nil } +// errRunRoadUnavailable is the one failure of the run road that sends a +// hand-off to the older engine: there is no run engine linked or no place for a +// store, so the run road was never there to take ([Agent.startTaskRun]). Every +// other failure happened ON the run road and is said to the person +// ([runDidNotStart]), because falling through to a different engine without a +// word is how a batch of approved hand-offs became old-tree nodes nobody asked +// for. +var errRunRoadUnavailable = errors.New("the run road is unavailable") + +// runDidNotStart is what a hand-off whose run did not start answers, on both +// doors: that it did not start, the reason in the store's or the disk's own +// words, and that nothing else was started in its place. +// +// IT MUST NOT READ LIKE SUCCESS. A receipt that said `task N started` over work +// that never started is one output with two meanings, and the person reading it +// cannot tell them apart; so this one opens on the one fact that differs. +func runDidNotStart(id uint64, err error) string { + reason := strings.TrimSuffix(strings.TrimSpace(err.Error()), ".") + return fmt.Sprintf("task %d did not start: %s. Nothing is running for it and nothing was started in its place; propose it again, or tell the person what stopped it.", id, reason) +} + // standsElsewhereError is the one refusal that STAYS AT THE RUN'S DOOR: a task // handed off while other work is underway shares that work's copy, and a copy is // of one folder. It says both folders and what to do, because the conversation @@ -284,10 +317,19 @@ func (e standsElsewhereError) Error() string { // An approved hand-off under the bash belt belongs to the run store and never to the session tree. func (a *Agent) startKnownTaskRun(ctx context.Context, id uint64, title, brief string, dependsOn []uint64, stand taskStand, question string) error { + _, err := a.startOrJoinTaskRun(ctx, id, title, brief, dependsOn, stand, question) + return err +} + +// startOrJoinTaskRun is [Agent.startKnownTaskRun] answering, too, whether the +// hand-off JOINED a run already underway rather than starting one, which only +// this door can know: a batch of hand-offs is one run, and which of them opened +// it is decided here, under the start lock, and nowhere before. +func (a *Agent) startOrJoinTaskRun(ctx context.Context, id uint64, title, brief string, dependsOn []uint64, stand taskStand, question string) (bool, error) { engine := chatRunEngine g := a.graph() if engine == nil || g == nil || g.planPath() == "" { - return errors.New("the run road is unavailable") + return false, errRunRoadUnavailable } path := g.planPath() storeID := strconv.FormatUint(id, 10) @@ -307,9 +349,25 @@ func (a *Agent) startKnownTaskRun(ctx context.Context, id uint64, title, brief s // with no report and no ending. So a hand-off that meets a run on its way // out waits for the run to be over and then starts a fresh one of its own // ([joinOrWait] is the whole of the decision). + // + // ── ONE RUN PER BATCH ── + // + // STARTING A RUN IS ONE CRITICAL SECTION, from "is there a live run" to the + // run being registered on the Agent, and every other hand-off waits at its + // door ([Agent.lockBeltStart]). A message that proposes eight tasks, all + // approved at once, commits eight hand-offs at the same moment, and without + // this each of them found no live run — the run is registered only after its + // store is open and its copy is cut, which takes seconds — and each opened or + // set aside the same store. Measured on the owner's own session: two started + // runs over one path, one run's worker filed its children into the other's + // store, and the other six fell through to the older engine. Held here, the + // first hand-off opens the run and the other seven find it live and join it + // as children, exactly as a hand-off made a minute later would. + a.lockBeltStart() + defer a.beltStartMu.Unlock() live, err := a.joinOrWait(ctx, stand, id, title, brief, dependencies) if err != nil { - return err + return false, err } if live != nil { a.publishRunRow(g, TaskNotice{ @@ -326,23 +384,23 @@ func (a *Agent) startKnownTaskRun(ctx context.Context, id uint64, title, brief s // under. The bare stored id is answered under by nothing. PlanTask: planStoreID(storeID), }) - return nil + return true, nil } plan, store, err := a.openBeltRunStore(g, path, storeID, title, brief, false) if err != nil { - return err + return false, err } if question = strings.TrimSpace(question); question != "" { if _, err := store.Revise(store.RootID(), plandb.TaskPatch{Question: &question}); err != nil { - _ = store.Close() - return err + discardUnstartedRunStore(store) + return false, err } } tree, err := beltRunPrepare(ctx, a.config.Place, a.config.Workspace, a.journalID(), id, title, stand) if err != nil { - _ = store.Close() - return err + discardUnstartedRunStore(store) + return false, err } // THE COPY IS A SHELL WORKER'S, so its landing stages the tree's own status: // a run's workers edit through bash and fill no write ledger. @@ -375,7 +433,35 @@ func (a *Agent) startKnownTaskRun(ctx context.Context, id uint64, title, brief s }) go a.driveBeltRun(runCtx, engine, run, a.beltRunSpec(run, brief)) - return nil + return false, nil +} + +// lockBeltStart takes the conversation's start lock, the one door every road +// that may open a run's store passes ([Agent.startOrJoinTaskRun] and +// [Agent.ContinueRun]). It is held from the look for a live run until the run +// is registered, so two hand-offs can never both decide there is no run and +// both open one. +func (a *Agent) lockBeltStart() { + if a.beltStartMu.TryLock() { + return + } + if beltStartWaits != nil { + beltStartWaits() + } + a.beltStartMu.Lock() +} + +// discardUnstartedRunStore takes back a store this door seeded for a run that +// then did not start. NOTHING WAS EVER RUN ON IT, so it is removed rather than +// left for the plan to read: a root nobody drives drew as work in flight, and +// the hand-off it was seeded for has already said it did not start +// ([runDidNotStart]). The path is free again for the next request. +func discardUnstartedRunStore(store *plandb.Store) { + path := store.Path() + _ = store.Close() + for _, suffix := range []string{"", "-wal", "-shm"} { + _ = os.Remove(path + suffix) + } } // beltRunPrepare cuts a run's working copy. It is [prepareTaskTreeOn] in the @@ -512,6 +598,17 @@ func (a *Agent) beltRunSpec(run *beltRun, brief string) RunSpec { // run's brief under the new request's number and dropped the new words, and // nobody was asked. func (a *Agent) openBeltRunStore(g *TaskGraph, path, rootID, title, brief string, carryOn bool) (*planState, *plandb.Store, error) { + // TWO RUNS NEVER SHARE A STORE PATH. Every caller holds the start lock and + // has seen no live run ([Agent.lockBeltStart]); this is the same fact asked + // once more where it would do the damage, because setting aside the store + // of a run that is still driving it is what split one batch into two runs + // writing through one path. + a.beltMu.Lock() + live := a.beltRun != nil + a.beltMu.Unlock() + if live { + return nil, nil, errors.New("a run is already live on this conversation's plan, so a second one may not open it") + } plan := &planState{path: path, chat: g.planChat()} if _, err := os.Stat(path); os.IsNotExist(err) { // NO STORE AT ALL IS NOBODY ELSE'S RUN, on either road: the run is seeded @@ -1053,13 +1150,6 @@ func (a *Agent) missingRunDependencies(ids []uint64) []uint64 { return missing } -// beltRunStandsOn reports whether a hand-off may share the live run's copy. -func (a *Agent) beltRunStandsOn(stand taskStand) bool { - a.beltMu.Lock() - defer a.beltMu.Unlock() - return a.beltRun != nil && a.beltRun.ground == canonicalPath(stand.dir) -} - // planArchivePaths names the ended run stores beside path in oldest-run-first // order. The run door uses the same naming read pages use, so archive creation // and discovery cannot drift apart. diff --git a/internal/session/task_run_continue.go b/internal/session/task_run_continue.go index be3441c66d..07980032b6 100644 --- a/internal/session/task_run_continue.go +++ b/internal/session/task_run_continue.go @@ -19,7 +19,6 @@ package session import ( "context" - "errors" "fmt" "strconv" ) @@ -45,12 +44,15 @@ func (a *Agent) ContinueRun(ctx context.Context, row uint64) (string, error) { engine := chatRunEngine g := a.graph() if engine == nil || g == nil || g.planPath() == "" { - return "", errors.New("the run road is unavailable") + return "", errRunRoadUnavailable } - // A RUN ALREADY GOING IS NOT CARRIED ON. Read under the belt's own lock, - // because a run installed between this look and the work below would be a - // second run on one conversation's store. + // A RUN ALREADY GOING IS NOT CARRIED ON. Read under the start lock, held + // until the carried-on run is registered, because a run installed between + // this look and the work below would be a second run on one conversation's + // store ([Agent.lockBeltStart]). + a.lockBeltStart() + defer a.beltStartMu.Unlock() a.beltMu.Lock() live := a.beltRun a.beltMu.Unlock() From f993c857f1d46da9373d9b5fd8a4cf6e2609f7e8 Mon Sep 17 00:00:00 2001 From: santoshkumarradha <instrument.santosh@gmail.com> Date: Wed, 23 Sep 2026 14:44:33 -0400 Subject: [PATCH 3/8] The failed-road test reads the success receipt's own words, not the word started --- internal/session/task_batch_run_test.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/internal/session/task_batch_run_test.go b/internal/session/task_batch_run_test.go index 104bb84d98..f38ab88d2a 100644 --- a/internal/session/task_batch_run_test.go +++ b/internal/session/task_batch_run_test.go @@ -228,7 +228,7 @@ func TestAHandoffWhoseRunRoadFailsIsNotASilentNode(t *testing.T) { if !failed { t.Errorf("the failed hand-off reads as a success: %q", answer) } - if strings.Contains(answer, "started") || !strings.Contains(answer, "did not start") || !strings.Contains(answer, "disk I/O error") { + if strings.Contains(answer, fmt.Sprintf("task %d started", proposal.id)) || !strings.Contains(answer, "did not start") || !strings.Contains(answer, "disk I/O error") { t.Errorf("the failed hand-off's receipt = %q, want it to say it did not start and why", answer) } if agent.graph().node(proposal.id) != nil { From 9ce0412e5e68bc699b372550540f940ce02c6e5d Mon Sep 17 00:00:00 2001 From: santoshkumarradha <instrument.santosh@gmail.com> Date: Wed, 23 Sep 2026 14:44:53 -0400 Subject: [PATCH 4/8] Change entry for one run per batch --- docs/changes/unreleased/1417-one-run-per-batch.md | 11 +++++++++++ 1 file changed, 11 insertions(+) create mode 100644 docs/changes/unreleased/1417-one-run-per-batch.md diff --git a/docs/changes/unreleased/1417-one-run-per-batch.md b/docs/changes/unreleased/1417-one-run-per-batch.md new file mode 100644 index 0000000000..da94e9d1d1 --- /dev/null +++ b/docs/changes/unreleased/1417-one-run-per-batch.md @@ -0,0 +1,11 @@ +--- +kind: fixed +title: tasks approved together are one run, and a run that could not start says so +pr: 1417 +surface: [chat, engine] +invalidates: + - "Several hand-offs approved at the same moment each opened their own run. They raced to the conversation's one plan: two started runs over one path, one run's worker wrote its children and its `done`s into the other's plan, and the rest became tasks of the older engine. Now the first opens the run and every other joins it as a child." + - "A run road that failed fell back to the older engine and answered `task N started`. It now answers `task N did not start: <reason>`, reads as a failure, and starts nothing in its place. Only a build with no run engine, or a conversation with nowhere to keep a plan, uses the older engine." + - "A run worker was bound to its plan by path alone. It is now bound to its run's root too (`PLANDB_RUN`), and `plandb` refuses a plan at that path whose root is another run's." + - "A message typed in a run row's room answered `no task N in this session`. It is now left as a note on the task's page; a row nothing drives says so and can be stopped." +--- From b54497ced4816ae98f42063a582c6a9a7b0015ce Mon Sep 17 00:00:00 2001 From: santoshkumarradha <instrument.santosh@gmail.com> Date: Wed, 23 Sep 2026 14:53:45 -0400 Subject: [PATCH 5/8] The stop law names the start door by its new name; the row-message heading stops competing with how to stop a task --- internal/manual/chat/worker-harness.md | 2 +- internal/session/stoplaw_test.go | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/internal/manual/chat/worker-harness.md b/internal/manual/chat/worker-harness.md index 82c2212dda..a2bf069738 100644 --- a/internal/manual/chat/worker-harness.md +++ b/internal/manual/chat/worker-harness.md @@ -171,7 +171,7 @@ only to where its plan was. If another run's plan is ever found in that place, t worker's `plandb` refuses it: `the plan store at <path> is another run's (t-<its task>), not this worker's run (t-<its own>), so nothing was read or written`. -## A run's row says running but nothing answers — a message to a run's task, stopping a row nothing drives +## Typing into a run's row, and a row nothing drives any more A message typed in the room of a run's row is left as a note on that task's page, and the room says `left on the task's page — its worker reads it between steps`. diff --git a/internal/session/stoplaw_test.go b/internal/session/stoplaw_test.go index 9f0749a735..9bb6424d14 100644 --- a/internal/session/stoplaw_test.go +++ b/internal/session/stoplaw_test.go @@ -31,7 +31,7 @@ import ( // here because they are not built from a literal state: their notices copy the // node's state, and the graph is the owner `task:N` has always reached. var stoppableRowPublishers = map[string]struct{ kind, proof string }{ - "startKnownTaskRun": {CancelTask, "TestAStopOnARunsOwnRowEndsTheRun"}, + "startOrJoinTaskRun": {CancelTask, "TestAStopOnARunsOwnRowEndsTheRun"}, "ContinueRun": {CancelTask, "TestAStopReachesARunThatWasCarriedOn"}, "newOrchestrateFamily": {CancelRun, "TestCancelStopsAnAdaptiveRun"}, "sayForming": {CancelRun, "TestCancelStopsAnAdaptiveRun"}, From 2a274692d62b6f71c8476f7dabbce4638ff88a95 Mon Sep 17 00:00:00 2001 From: santoshkumarradha <instrument.santosh@gmail.com> Date: Wed, 23 Sep 2026 14:55:06 -0400 Subject: [PATCH 6/8] The stop of a row nothing drives is said where stopping a run is --- internal/manual/chat/worker-harness.md | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/internal/manual/chat/worker-harness.md b/internal/manual/chat/worker-harness.md index a2bf069738..8237c8c08e 100644 --- a/internal/manual/chat/worker-harness.md +++ b/internal/manual/chat/worker-harness.md @@ -180,8 +180,7 @@ A run row that nothing drives any more, because its run is not the one this conversation is driving or its plan holds no such task, answers a message with `nothing is driving this task any more, so no worker can read a message; stop it to clear the row`. It never answers `no task N in this session` while the side list -draws it running. Stopping it settles the row as `stopped`, and the stop answers -`stopped task N (<title>) — nothing was driving it any more`. +still draws it. *How do I stop a run?* says what clearing it does. ## Why is this task indented under that one? @@ -500,7 +499,9 @@ off, and no further model call is made for it. The row reads `stopped`. A second on a run that is already stopping answers that it is already stopping. `x` on one PART of a run ends that part only, at once and without a card, and the -rest of the run carries on. A run cannot be paused as a whole, so under the run's own +rest of the run carries on. A row nothing drives any more is cleared the same way: +the stop settles it as `stopped` and answers `stopped task N (<title>) — nothing was +driving it any more`. A run cannot be paused as a whole, so under the run's own task no `p pause` is named. Closing the window, `ctrl+c` and `/quit` do NOT stop a run: it carries on without the From f7f24c528d3da3789c81c923e2b720136be3447e Mon Sep 17 00:00:00 2001 From: santoshkumarradha <instrument.santosh@gmail.com> Date: Wed, 23 Sep 2026 15:20:57 -0400 Subject: [PATCH 7/8] The #1418 change entry carries its pull request, so the entries are well formed again --- ...ground-lint-remote-path.md => 1418-ground-lint-remote-path.md} | 0 1 file changed, 0 insertions(+), 0 deletions(-) rename docs/changes/unreleased/{ground-lint-remote-path.md => 1418-ground-lint-remote-path.md} (100%) diff --git a/docs/changes/unreleased/ground-lint-remote-path.md b/docs/changes/unreleased/1418-ground-lint-remote-path.md similarity index 100% rename from docs/changes/unreleased/ground-lint-remote-path.md rename to docs/changes/unreleased/1418-ground-lint-remote-path.md From 78637f7de730473d972b4648bc97356b97faf3a5 Mon Sep 17 00:00:00 2001 From: santoshkumarradha <instrument.santosh@gmail.com> Date: Wed, 23 Sep 2026 15:38:55 -0400 Subject: [PATCH 8/8] The #1418 entry's pr line, and the remote-path test names no real person's folder The pr line was left unstaged by the rename. The test spelled a real checkout under /home, which exists on the machine it was measured on, so it failed there. --- docs/changes/unreleased/1418-ground-lint-remote-path.md | 1 + internal/session/taskstands_test.go | 6 +++++- 2 files changed, 6 insertions(+), 1 deletion(-) diff --git a/docs/changes/unreleased/1418-ground-lint-remote-path.md b/docs/changes/unreleased/1418-ground-lint-remote-path.md index d99eff099a..b297656730 100644 --- a/docs/changes/unreleased/1418-ground-lint-remote-path.md +++ b/docs/changes/unreleased/1418-ground-lint-remote-path.md @@ -1,6 +1,7 @@ --- kind: fixed title: a deliverable path on another machine no longer refuses the task +pr: 1418 surface: [engine] invalidates: - "`groundLint` refused a task whose deliverable named any absolute path outside its ground, whether or not that path was a directory on this machine. A path is a place only when the directory it names exists here and is not the filesystem root. A directory merely somewhere along the path does not count: on macOS `/home` is a symlink to a directory that is there, so a path from another host that begins `/home` was still read as a place and still refused." diff --git a/internal/session/taskstands_test.go b/internal/session/taskstands_test.go index d521234d68..94d964e979 100644 --- a/internal/session/taskstands_test.go +++ b/internal/session/taskstands_test.go @@ -423,7 +423,11 @@ func TestAPathOnAnotherMachineDoesNotRefuseTheTask(t *testing.T) { // THE PATH THAT FAILED. It begins /home, and on macOS /home is a symlink to a // directory that is really there, so a reading that walks up the path finds a // place and refuses the task. The directory the path itself names is not here. - remote := "/home/santosh/src/af-dev2-probe/bin/codeaf" + // + // IT NAMES NOBODY'S REAL FOLDER. It spelled a real person's checkout once, and + // on the machine that holds that checkout the directory is there, so the test + // failed on the one box it was written about. + remote := "/home/remote-builder/src/codeaf-probe/bin/codeaf" if _, ok := placeOnThisMachine(remote); ok { t.Fatalf("%s resolves to a directory on this machine, so it cannot stand in for a remote path", remote) }