Merge pull request #272 from LeXwDeX/fix/adoption-state-machine

fix(dag): session-following location stamps + adoption-vs-deletion fencing (#269, #270)

Author: LeXwDeX

packages/core/src/dag/sql.ts

@@ -31,11 +31,16 @@ export const WorkflowTable = sqliteTable(
     session_id: text()
       .notNull()
       .references(() => SessionTable.id, { onDelete: "cascade" }),
-    // Execution-location key (DAG-LOC-01): the creating instance's directory,
-    // stamped at dag.create. Only the instance whose directory matches may
-    // adopt, recover, wake, or spawn for this workflow. Nullable: legacy rows
-    // predating the column match no instance (conservative — never adopted
-    // until recreated).
+    // Execution-location key (DAG-LOC-01): the directory that owns this workflow.
+    // Only the instance whose directory matches may adopt, recover, wake, or spawn
+    // for it. TWO-WRITER WHITELIST (#269): the stamp is written at dag.create
+    // (WorkflowCreated projection INSERT, onConflictDoNothing — a replay can never
+    // rewrite an existing stamp) and re-stamped ONLY by the session projector's
+    // SessionEvent.Moved projection (the stamp moves WITH the session in one
+    // transaction, payload-sourced from the Moved event — no SessionTable read).
+    // No other writer may set it; R7-ext pins the whitelist. Nullable: legacy rows
+    // predating the column match no instance (fail-closed — never adopted until
+    // recreated).
     directory: text(),
     title: text().notNull(),
     status: text().notNull(),

packages/core/src/dag/store.ts

@@ -148,6 +148,7 @@ const wakeDeliverableNodePredicate = or(
 
 export interface Interface {
   readonly getWorkflow: (id: string) => Effect.Effect<WorkflowRow | undefined>
+  readonly tryClaimAdoption: (id: string) => Effect.Effect<boolean>
   readonly listWorkflows: () => Effect.Effect<WorkflowRow[]>
   readonly listBySession: (sessionId: string) => Effect.Effect<WorkflowRow[]>
   readonly listByProject: (projectId: string) => Effect.Effect<WorkflowRow[]>
@@ -182,6 +183,31 @@ export const layer = Layer.effect(
         return row ? mapWorkflow(row) : undefined
       }),
 
+      // #270 atomic-adoption fence (C2). The adoption sites previously re-read the
+      // row (ownsWorkflow) and then published their entry into the in-memory map —
+      // a check-then-act pair a deletion cascade could commit between. The claim
+      // collapses the admission into ONE conditional UPDATE: it matches the row
+      // only while the row STILL EXISTS and is in an adoptable (non-terminal)
+      // status, and returns whether it claimed. A Session.remove (FK cascade) or a
+      // terminal transition that commits before the claim therefore makes the claim
+      // match zero rows and the adoption aborts atomically — no post-deletion
+      // admission survives. Directory ownership is NOT re-asserted here: the caller
+      // has already passed DagLocation.ownsWorkflow, which canonicalizes both sides;
+      // duplicating a directory comparison in SQL would diverge from that
+      // canonicalization (create stamps are realpathed, Moved re-stamps are not),
+      // so status conditionality is the fence and the read authority keeps the
+      // directory key. No lease column — the status conditionality IS the claim.
+      tryClaimAdoption: Effect.fn("DagStore.tryClaimAdoption")(function* (id) {
+        const claimed = yield* db
+          .update(WorkflowTable)
+          .set({ time_updated: Date.now() })
+          .where(and(eq(WorkflowTable.id, id), inArray(WorkflowTable.status, ["pending", "running", "paused", "stepping"])))
+          .returning({ id: WorkflowTable.id })
+          .get()
+          .pipe(Effect.orDie)
+        return claimed !== undefined
+      }),
+
       listWorkflows: Effect.fn("DagStore.listWorkflows")(function* () {
         const rows = yield* db.select().from(WorkflowTable).orderBy(desc(WorkflowTable.time_created)).all().pipe(Effect.orDie)
         return rows.map(mapWorkflow)

packages/core/src/database/migration.gen.ts

@@ -0,0 +1,35 @@
+import { Effect } from "effect"
+import type { DatabaseMigration } from "../migration"
+
+export default {
+  id: "20260815083000_workflow_directory_convergence",
+  up(tx) {
+    return Effect.gen(function* () {
+      // #269 atomic-adoption convergence (C6). The execution-location stamp now
+      // moves WITH the session at SessionEvent.Moved time (session projector),
+      // but installs that moved a session BEFORE that transition shipped carry
+      // divergent stamps: the session row points at the destination directory
+      // while its pre-move workflow rows stay pinned at the old one. With the
+      // fail-closed ownership conjunct, those mixed stamps leave the session's
+      // wakes with NO owner (the wedge from v1.0.13). Converge every NON-NULL
+      // workflow stamp to its session's CURRENT directory — the same direction
+      // the live Moved re-stamp moves it. No ALTER: the directory column already
+      // exists. Preserves the fail-closed invariants: a NULL stamp (legacy row
+      // never backfilled) stays NULL, and a workflow whose session has no
+      // directory is left untouched. Idempotent — re-running converges to the
+      // same state.
+      yield* tx.run(`
+        UPDATE \`workflow\`
+        SET \`directory\` = (
+          SELECT \`directory\` FROM \`session\` WHERE \`session\`.\`id\` = \`workflow\`.\`session_id\`
+        )
+        WHERE \`workflow\`.\`directory\` IS NOT NULL
+          AND EXISTS (
+            SELECT 1 FROM \`session\`
+            WHERE \`session\`.\`id\` = \`workflow\`.\`session_id\`
+              AND \`session\`.\`directory\` IS NOT NULL
+          );
+      `)
+    })
+  },
+} satisfies DatabaseMigration.Migration

packages/core/src/database/migration/20260815083000_workflow_directory_convergence.ts

@@ -13,6 +13,7 @@ import { SessionMessageUpdater } from "./message-updater"
 import { SessionInput } from "./input"
 import { WorkspaceV2 } from "../workspace"
 import { SessionContextEpoch } from "./context-epoch"
+import { WorkflowTable } from "../dag/sql"
 import { MessageTable, PartTable, SessionInputTable, SessionMessageTable, SessionTable } from "./sql"
 import type { DeepMutable } from "../schema"
 import { SessionMessageID } from "./message-id"
@@ -254,6 +255,21 @@ export const layer = Layer.effectDiscard(
           .where(eq(SessionTable.id, event.data.sessionID))
           .run()
           .pipe(Effect.orDie)
+        // #269 atomic-adoption resolution: the execution-location stamp must move
+        // WITH the session in one transaction. Re-stamp every durable workflow
+        // row of the session to the payload-sourced destination directory (NO
+        // SessionTable read — the Moved payload carries it). This runs inside the
+        // same durable publish transaction as the SessionTable update, so there is
+        // never a window where the session's rows carry mixed stamps (fail-closed
+        // ownership would otherwise wedge every wake for the session). This is the
+        // second whitelisted directory writer (see WorkflowTable.directory): the
+        // create-time INSERT is the first; only SessionEvent.Moved may re-stamp.
+        yield* db
+          .update(WorkflowTable)
+          .set({ directory: event.data.location.directory })
+          .where(eq(WorkflowTable.session_id, event.data.sessionID))
+          .run()
+          .pipe(Effect.orDie)
         yield* SessionContextEpoch.reset(db, event.data.sessionID)
       }),
     )

packages/core/src/session/projector.ts

@@ -9,6 +9,7 @@ import { SessionV1 } from "@opencode-ai/core/v1/session"
 import { InstanceState } from "@/effect/instance-state"
 import { EventV2Bridge } from "@/event-v2-bridge"
 import { DagEvent } from "@opencode-ai/schema/dag-event"
+import { SessionEvent } from "@opencode-ai/schema/session-event"
 import { SessionStatusEvent } from "@opencode-ai/schema/session-status-event"
 import { DagStore } from "@opencode-ai/core/dag/store"
 import { DagLocation } from "../location"
@@ -433,6 +434,13 @@ const serviceLayer = Layer.effect(
             // publishing now would leak an inert entry the sweep can no longer
             // reach. The ensuring below still clears the recovering reservation.
             if (!(yield* DagLocation.ownsWorkflow(dagID, ctx.directory))) return
+            // #270 atomic-admission fence (C3): the durable read above and the
+            // runtimes publish below are still two statements — a deletion can
+            // commit between them. Collapse the admission into ONE conditional
+            // UPDATE that matches only while the row exists and is non-terminal.
+            // A cascade committed in that final window matches zero rows and the
+            // adoption aborts here, before it ever publishes an entry.
+            if (!(yield* store.tryClaimAdoption(dagID))) return
             runtimes.set(dagID, entry)
             yield* automation.register(SessionID.make(wf.sessionId), { kind: "dag", id: dagID })
             // Reconciliation settles every persisted running attempt before the
@@ -552,6 +560,12 @@ const serviceLayer = Layer.effect(
                 // through. A row cascade-deleted after the first guard must not
                 // be adopted into an inert entry.
                 if (!(yield* DagLocation.ownsWorkflow(dagID, ctx.directory))) return
+                // #270 atomic-admission fence (C3, same as recoverWorkflow): the
+                // durable read and the runtimes publish are two statements a
+                // deletion can slip between; collapse the admission into one
+                // conditional UPDATE (exists + non-terminal). A cascade committed
+                // in the final window matches zero rows and the adoption aborts.
+                if (!(yield* store.tryClaimAdoption(dagID))) return
                 runtimes.set(dagID, entry)
                 yield* automation.register(SessionID.make(wf.sessionId), { kind: "dag", id: dagID })
                 yield* entry.evalLock.withPermits(1)(
@@ -1392,6 +1406,52 @@ const serviceLayer = Layer.effect(
           Effect.forkScoped({ startImmediately: true }),
         )
 
+        // #269 SessionMoved ownership convergence: the Moved projection
+        // (core session projector) re-stamps the session's workflow rows to the
+        // destination directory in the SAME durable transaction, so by the time
+        // this handler runs the durable rows already agree on ONE directory.
+        // Converge the in-memory side: (a) the instance that no LONGER owns the
+        // moved session's workflows evicts its stale runtime entries (fail-closed
+        // — its directory must not keep acting on them), and (b) the NEW owner
+        // re-forks the serialized wake drain so a terminal wake that was wedged
+        // behind the old mixed stamps delivers immediately (bounded time) instead
+        // of waiting for a fresh idle event or a restart.
+        yield* events.subscribe(SessionEvent.Moved).pipe(
+          Stream.runForEach((evt) =>
+            Effect.gen(function* () {
+              const sessionID = evt.data.sessionID as string
+              // Map iteration is mutation-safe for deletions of visited entries —
+              // only entries of THIS session are deleted, each inside its own
+              // evalLock. Evict only entries the re-stamp moved AWAY from this
+              // instance (ownsWorkflow re-reads the durable row).
+              for (const [dagID, entry] of runtimes) {
+                if (entry.parentSessionID !== sessionID) continue
+                if (yield* DagLocation.ownsWorkflow(dagID, ctx.directory)) continue
+                yield* entry.evalLock.withPermits(1)(
+                  Effect.gen(function* () {
+                    for (const [nodeID, fiber] of entry.fibers) {
+                      const node = yield* store.getNode(dagID, nodeID)
+                      yield* abortChild(nodeID, node?.childSessionId ?? null).pipe(Effect.ignore)
+                      yield* Fiber.interrupt(fiber).pipe(Effect.ignore)
+                      const watcher = entry.watchers.get(nodeID)
+                      if (watcher) yield* Fiber.interrupt(watcher).pipe(Effect.ignore)
+                    }
+                    entry.fibers.clear()
+                    entry.watchers.clear()
+                    runtimes.delete(dagID)
+                  }),
+                )
+              }
+              // New owner: the re-stamp moved ownership HERE, so wake rows that
+              // were wedged (mixed stamps → no owner) are now deliverable.
+              if (yield* DagLocation.ownsSession(sessionID, ctx.directory)) {
+                yield* tryDeliverWake(sessionID).pipe(Effect.ignore, Effect.forkScoped)
+              }
+            }).pipe(guarded("SessionMoved")),
+          ),
+          Effect.forkScoped({ startImmediately: true }),
+        )
+
         // Install all live event handlers before spawning recovery watchers so
         // a child that settles immediately cannot leave the runtime stale.
         // Orphan-pending sweep first: the WorkflowStarted it publishes for the

packages/opencode/src/dag/runtime/loop.ts

@@ -404,6 +404,17 @@ export function spawnNode(
           return
         }
         try {
+          // #270 window-2 spawn-admission fence (C4): the node was durably
+          // admitted (nodeQueued above) but the child session is about to
+          // materialize — a deletion cascade (Session.remove → FK) committed in
+          // that window must fence the spawn instead of letting it create an
+          // orphan child for a dead workflow. Re-admit ATOMICALLY right before
+          // sessions.create: the claim matches only while the workflow row exists
+          // and is non-terminal, so a committed deletion matches zero rows and the
+          // spawn aborts before any child session exists (no post-deletion spawn
+          // survives). This is the revalidation that closes the spawn window the
+          // nodeQueued guard alone leaves open between its read and its publish.
+          if (!(yield* dag.store.tryClaimAdoption(input.dagID))) return
           // Permit acquired — only NOW materialize the child session and mark
           // the node running (P0-2). Before this point the node is durably
           // "queued" with no session: a 100-node fan-out holds at most