One run per batch: simultaneous hand-offs join one run, a failed run road says so

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: a typed `/task` answered as if it had started, and an approved hand-off's receipt said `It runs on the older task engine, because the run engine could not start it:` (#1416). Neither falls back now. Both answer `task N did not start: <reason>`, the hand-off's receipt reads as a failure, and nothing starts 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."
+---

internal/manual/chat/worker-harness.md

@@ -162,6 +162,40 @@ 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: <the reason>. 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 <path> is another run's (t-<its
+task>), not this worker's run (t-<its own>), so nothing was read or written`.
+
+## 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`.
+
+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
+still draws it. *How do I stop a run?* says what clearing it does.
+
 ## 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
@@ -482,7 +516,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
@@ -815,9 +851,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 a task the run road cannot start falls back to it — so a
-conversation the run road cannot serve gets exactly the door it always had. When a
-task you approved falls back that way, its receipt says so: `It runs on the older
-task engine, because the run engine could not start it:` and the run road's own
-reason.
+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.

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
@@ -253,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
 }
 

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)
+}

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())
 }

internal/run/bashworker.go

@@ -105,7 +105,7 @@ func (w *BashWorker) Run(ctx context.Context, task plandb.Task) (rep Report, run
 	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
 	}

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))
+		}
+	}
+}

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())
 }

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

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)

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)
 	}

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.
@@ -733,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

internal/session/session.go

@@ -3236,6 +3236,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

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"},