diff --git a/bun.lock b/bun.lock index a86571ba07..59bd1574e3 100644 --- a/bun.lock +++ b/bun.lock @@ -33,6 +33,7 @@ "dependencies": { "@kobalte/core": "catalog:", "@opencode-ai/core": "workspace:*", + "@opencode-ai/schema": "workspace:*", "@opencode-ai/sdk": "workspace:*", "@opencode-ai/session-ui": "workspace:*", "@opencode-ai/ui": "workspace:*", diff --git a/packages/app/package.json b/packages/app/package.json index a4773e4da2..98f37c764f 100644 --- a/packages/app/package.json +++ b/packages/app/package.json @@ -46,6 +46,7 @@ "dependencies": { "@kobalte/core": "catalog:", "@opencode-ai/core": "workspace:*", + "@opencode-ai/schema": "workspace:*", "@opencode-ai/sdk": "workspace:*", "@opencode-ai/session-ui": "workspace:*", "@opencode-ai/ui": "workspace:*", diff --git a/packages/app/src/utils/id-wrap-boundary.test.ts b/packages/app/src/utils/id-wrap-boundary.test.ts new file mode 100644 index 0000000000..5e26a1ab08 --- /dev/null +++ b/packages/app/src/utils/id-wrap-boundary.test.ts @@ -0,0 +1,58 @@ +import { describe, expect, test } from "bun:test" +import { create } from "@opencode-ai/schema/identifier" + +const WRAP_BOUNDARY = 1786706395136 +const chars = "0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz" + +describe("app identifier seam: 48-bit wrap", () => { + test("ascending ids stay lexicographically ascending across the wrap boundary", () => { + const preWrap = create(false, WRAP_BOUNDARY - 1) + const postWrap = create(false, WRAP_BOUNDARY + 1) + expect(preWrap < postWrap).toBe(true) + }) + + test("descending ids stay lexicographically descending across the wrap boundary", () => { + const preWrap = create(true, WRAP_BOUNDARY - 1) + const postWrap = create(true, WRAP_BOUNDARY + 1) + expect(postWrap < preWrap).toBe(true) + }) + + test("same-millisecond ids are strictly ascending and unique", () => { + const ids = Array.from({ length: 10 }, (_, index) => create(false, WRAP_BOUNDARY + 1000 + index)) + for (let i = 1; i < ids.length; i++) { + expect(ids[i - 1] < ids[i]).toBe(true) + } + expect(new Set(ids).size).toBe(ids.length) + }) + + test("same-millisecond descending ids are strictly descending and unique", () => { + const ids = Array.from({ length: 10 }, (_, index) => create(true, WRAP_BOUNDARY + 2000 + index)) + for (let i = 1; i < ids.length; i++) { + expect(ids[i - 1] > ids[i]).toBe(true) + } + expect(new Set(ids).size).toBe(ids.length) + }) + + test("id format is 12 hex chars plus 14 base62 chars", () => { + const id = create(false, WRAP_BOUNDARY + 1) + expect(id).toHaveLength(26) + expect(id.slice(0, 12)).toMatch(/^[0-9a-f]{12}$/) + for (const char of id.slice(12)) { + expect(chars).toContain(char) + } + }) + + test("ids stay ascending when the clock regresses (latch absorbs regression)", () => { + const first = create(false, WRAP_BOUNDARY + 3000) + const regressed = create(false, WRAP_BOUNDARY + 3000 - 50) + const later = create(false, WRAP_BOUNDARY + 3000 + 50) + expect(first < regressed).toBe(true) + expect(regressed < later).toBe(true) + }) + + test("new-scheme ids sort below historical pre-wrap ids (comparisons must be time-based)", () => { + const now = create(false, Date.now()) + const historical = create(false, WRAP_BOUNDARY - 1) + expect(now < historical).toBe(true) + }) +}) diff --git a/packages/app/src/utils/id.test.ts b/packages/app/src/utils/id.test.ts new file mode 100644 index 0000000000..00d7b8af93 --- /dev/null +++ b/packages/app/src/utils/id.test.ts @@ -0,0 +1,75 @@ +import { describe, expect, test } from "bun:test" +import { create } from "@opencode-ai/core/id/id" +import { Identifier } from "./id" + +const prefixes = { + session: "ses", + message: "msg", + permission: "per", + user: "usr", + part: "prt", + pty: "pty", +} as const + +function decodeTime(id: string): number { + const start = id.indexOf("_") + 1 + return Number(BigInt("0x" + id.slice(start, start + 12))) +} + +describe("Identifier", () => { + test("every prefix keeps the prefix_underscore_26-char shape", () => { + for (const prefix of ["session", "message", "permission", "user", "part", "pty"] as const) { + const ascending = Identifier.ascending(prefix) + const descending = Identifier.descending(prefix) + expect(ascending.startsWith(`${prefixes[prefix]}_`)).toBe(true) + expect(descending.startsWith(`${prefixes[prefix]}_`)).toBe(true) + for (const id of [ascending, descending]) { + expect(id).toHaveLength(prefixes[prefix].length + 1 + 26) + expect(id.slice(prefixes[prefix].length + 1, prefixes[prefix].length + 13)).toMatch(/^[0-9a-f]{12}$/) + } + } + }) + + test("ascending ids decode their time prefix back to raw wall-clock ms", () => { + // The legacy encoding shifted the millisecond value 12 bits and wrapped at + // 2026-08-14 (issue 271), so under it a live id no longer decodes near + // Date.now(). + const before = Date.now() + const id = Identifier.ascending("message") + const after = Date.now() + const decoded = decodeTime(id) + expect(decoded).toBeGreaterThanOrEqual(before - 5000) + expect(decoded).toBeLessThanOrEqual(after + 5000) + }) + + test("app ids sort consistently with core ids for the same wall-clock", () => { + const appID = Identifier.ascending("message") + const coreID = create("msg", "ascending") + expect(Math.abs(decodeTime(appID) - decodeTime(coreID))).toBeLessThanOrEqual(5000) + }) + + test("same-millisecond bursts stay strictly ascending and unique", () => { + const ids = Array.from({ length: 20 }, () => Identifier.ascending("message")) + for (let index = 1; index < ids.length; index++) { + expect(ids[index - 1] < ids[index]).toBe(true) + } + expect(new Set(ids).size).toBe(ids.length) + }) + + test("descending ids stay strictly descending and unique", () => { + const ids = Array.from({ length: 20 }, () => Identifier.descending("session")) + for (let index = 1; index < ids.length; index++) { + expect(ids[index - 1] > ids[index]).toBe(true) + } + expect(new Set(ids).size).toBe(ids.length) + }) + + test("given id passes through when the prefix matches", () => { + expect(Identifier.ascending("session", "ses_abc123")).toBe("ses_abc123") + expect(Identifier.descending("message", "msg_abc123")).toBe("msg_abc123") + }) + + test("given id with a wrong prefix throws", () => { + expect(() => Identifier.ascending("message", "ses_abc123")).toThrow("ID ses_abc123 does not start with msg") + }) +}) diff --git a/packages/app/src/utils/id.ts b/packages/app/src/utils/id.ts index dba7a8d951..0f4e363568 100644 --- a/packages/app/src/utils/id.ts +++ b/packages/app/src/utils/id.ts @@ -1,3 +1,5 @@ +import { create } from "@opencode-ai/schema/identifier" + const prefixes = { session: "ses", message: "msg", @@ -7,10 +9,6 @@ const prefixes = { pty: "pty", } as const -const LENGTH = 26 -let lastTimestamp = 0 -let counter = 0 - type Prefix = keyof typeof prefixes export namespace Identifier { export function ascending(prefix: Prefix, given?: string) { @@ -24,7 +22,7 @@ export namespace Identifier { function generateID(prefix: Prefix, descending: boolean, given?: string): string { if (!given) { - return create(prefix, descending) + return prefixes[prefix] + "_" + create(descending) } if (!given.startsWith(prefixes[prefix])) { @@ -33,61 +31,3 @@ function generateID(prefix: Prefix, descending: boolean, given?: string): string return given } - -function create(prefix: Prefix, descending: boolean, timestamp?: number): string { - const currentTimestamp = timestamp ?? Date.now() - - if (currentTimestamp !== lastTimestamp) { - lastTimestamp = currentTimestamp - counter = 0 - } - - counter += 1 - - let now = BigInt(currentTimestamp) * BigInt(0x1000) + BigInt(counter) - - if (descending) { - now = ~now - } - - const timeBytes = new Uint8Array(6) - for (let i = 0; i < 6; i += 1) { - timeBytes[i] = Number((now >> BigInt(40 - 8 * i)) & BigInt(0xff)) - } - - return prefixes[prefix] + "_" + bytesToHex(timeBytes) + randomBase62(LENGTH - 12) -} - -function bytesToHex(bytes: Uint8Array): string { - let hex = "" - for (let i = 0; i < bytes.length; i += 1) { - hex += bytes[i].toString(16).padStart(2, "0") - } - return hex -} - -function randomBase62(length: number): string { - const chars = "0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz" - const bytes = getRandomBytes(length) - let result = "" - for (let i = 0; i < length; i += 1) { - result += chars[bytes[i] % 62] - } - return result -} - -function getRandomBytes(length: number): Uint8Array { - const bytes = new Uint8Array(length) - const cryptoObj = typeof globalThis !== "undefined" ? globalThis.crypto : undefined - - if (cryptoObj && typeof cryptoObj.getRandomValues === "function") { - cryptoObj.getRandomValues(bytes) - return bytes - } - - for (let i = 0; i < length; i += 1) { - bytes[i] = Math.floor(Math.random() * 256) - } - - return bytes -} diff --git a/packages/core/src/dag/sql.ts b/packages/core/src/dag/sql.ts index 5268b82072..cc946745b4 100644 --- a/packages/core/src/dag/sql.ts +++ b/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(), diff --git a/packages/core/src/dag/store.ts b/packages/core/src/dag/store.ts index 07405305cb..40bc65d45b 100644 --- a/packages/core/src/dag/store.ts +++ b/packages/core/src/dag/store.ts @@ -148,6 +148,7 @@ const wakeDeliverableNodePredicate = or( export interface Interface { readonly getWorkflow: (id: string) => Effect.Effect + readonly tryClaimAdoption: (id: string) => Effect.Effect readonly listWorkflows: () => Effect.Effect readonly listBySession: (sessionId: string) => Effect.Effect readonly listByProject: (projectId: string) => Effect.Effect @@ -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) diff --git a/packages/core/src/database/migration.gen.ts b/packages/core/src/database/migration.gen.ts index 8ffc6c9fdf..a69db114e0 100644 --- a/packages/core/src/database/migration.gen.ts +++ b/packages/core/src/database/migration.gen.ts @@ -54,5 +54,6 @@ export const migrations = ( import("./migration/20260811060000_goal_outcome"), import("./migration/20260813020344_bored_skaar"), import("./migration/20260813040429_workflow_directory"), + import("./migration/20260815083000_workflow_directory_convergence"), ]) ).map((module) => module.default) satisfies DatabaseMigration.Migration[] diff --git a/packages/core/src/database/migration/20260815083000_workflow_directory_convergence.ts b/packages/core/src/database/migration/20260815083000_workflow_directory_convergence.ts new file mode 100644 index 0000000000..80a98282cc --- /dev/null +++ b/packages/core/src/database/migration/20260815083000_workflow_directory_convergence.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 diff --git a/packages/core/src/session/projector.ts b/packages/core/src/session/projector.ts index 6605139903..0064fcb1d6 100644 --- a/packages/core/src/session/projector.ts +++ b/packages/core/src/session/projector.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) }), ) diff --git a/packages/opencode/src/dag/runtime/loop.ts b/packages/opencode/src/dag/runtime/loop.ts index 0d7b37f685..24a2ef8f50 100644 --- a/packages/opencode/src/dag/runtime/loop.ts +++ b/packages/opencode/src/dag/runtime/loop.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 diff --git a/packages/opencode/src/dag/runtime/spawn.ts b/packages/opencode/src/dag/runtime/spawn.ts index 0c29a727d1..d705c2890b 100644 --- a/packages/opencode/src/dag/runtime/spawn.ts +++ b/packages/opencode/src/dag/runtime/spawn.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 diff --git a/packages/opencode/test/dag/dag-location-guards.test.ts b/packages/opencode/test/dag/dag-location-guards.test.ts index 8c35d57e58..ba712ecd7d 100644 --- a/packages/opencode/test/dag/dag-location-guards.test.ts +++ b/packages/opencode/test/dag/dag-location-guards.test.ts @@ -39,12 +39,16 @@ import type { SessionV1 } from "@opencode-ai/core/v1/session" import { SessionV1 as SessionV1Events } from "@opencode-ai/core/v1/session" import { Database } from "@opencode-ai/core/database/database" import { DagProjector } from "@opencode-ai/core/dag/projector" +import { SessionProjector } from "@opencode-ai/core/session/projector" import { WorkflowTable, WorkflowNodeTable } from "@opencode-ai/core/dag/sql" +import workflowDirectoryConvergence from "@opencode-ai/core/database/migration/20260815083000_workflow_directory_convergence" import { DagStore } from "@opencode-ai/core/dag/store" import { EventV2 } from "@opencode-ai/core/event" import { ProjectTable } from "@opencode-ai/core/project/sql" import { SessionTable } from "@opencode-ai/core/session/sql" import { DagEvent } from "@opencode-ai/schema/dag-event" +import { Location } from "@opencode-ai/schema/location" +import { SessionEvent } from "@opencode-ai/schema/session-event" import { SessionStatusEvent } from "@opencode-ai/schema/session-status-event" import { Agent } from "@/agent/agent" import { Dag, type NodeConfig } from "@/dag/dag" @@ -113,6 +117,24 @@ function reply(sessionID: string, text: string): SessionV1.WithParts { } as never } +/** + * Publish the durable SessionEvent.Moved that the control-plane's move-session + * publishes. The session projector's Moved projection re-stamps every workflow + * row of the session in the SAME transaction (payload-sourced, no SessionTable + * read) — the #269 atomic-adoption transition under test. + */ +function moveSession(bridge: EventV2.Interface, sessionID: string, directory: string): Effect.Effect { + return Effect.gen(function* () { + yield* bridge + .publish(SessionEvent.Moved, { + sessionID: sessionID as never, + location: Location.Ref.make({ directory: directory as never }), + timestamp: yield* DateTime.now, + }) + .pipe(Effect.orDie) + }) +} + // --------------------------------------------------------------------------- // Two-instance harness (extension of dag-loop-guards.test.ts guardLayer / // runGuardTest): two InstanceRefs, DISTINCT directories, SAME project id. @@ -152,20 +174,34 @@ interface TwoInstanceInput { * release, prepareIfIdle returns none (the call itself still proves the * idle subscription survived). */ - readonly parkWakeDelivery?: { - readonly wait: Promise - readonly parked: { value: boolean } - readonly released: { value: boolean } - readonly calls: { value: number } - } -} + readonly parkWakeDelivery?: { + readonly wait: Promise + readonly parked: { value: boolean } + readonly released: { value: boolean } + readonly calls: { value: number } + } + /** + * Optional deterministic park on DagStore.tryClaimAdoption (the #270 atomic + * admission fence). Each claim bumps `calls`; the first `skip` claims pass + * through unparked (e.g. the adoption claim), every later claim sets + * `parked.value = true` and parks on `wait` before delegating — letting a + * probe interleave Session.remove between the passed admission and the + * in-flight spawn's child-session creation (the deletion-race probe C7). + */ + readonly parkAdoptionClaim?: { + readonly wait: Promise + readonly parked: { value: boolean } + readonly calls: { value: number } + readonly skip: number + } + } function twoInstanceLayer(input: TwoInstanceInput) { const database = Database.layerFromPath(":memory:") const events = EventV2.layer.pipe(Layer.provide(database)) const bridge = EventV2Bridge.layer.pipe(Layer.provide(events)) const realStore = DagStore.layer.pipe(Layer.provide(database)) - const needsStoreWrapper = Boolean(input.failGetWorkflow) || Boolean(input.parkGetNodes) + const needsStoreWrapper = Boolean(input.failGetWorkflow) || Boolean(input.parkGetNodes) || Boolean(input.parkAdoptionClaim) const store = needsStoreWrapper ? Layer.effect( DagStore.Service, @@ -189,6 +225,15 @@ function twoInstanceLayer(input: TwoInstanceInput) { gate.parked.value = true return Effect.promise(() => gate.wait).pipe(Effect.flatMap(() => real.getNodes(id))) }), + tryClaimAdoption: (id) => + Effect.suspend(() => { + const gate = input.parkAdoptionClaim + if (!gate) return real.tryClaimAdoption(id) + gate.calls.value++ + if (gate.calls.value <= gate.skip) return real.tryClaimAdoption(id) + gate.parked.value = true + return Effect.promise(() => gate.wait).pipe(Effect.flatMap(() => real.tryClaimAdoption(id))) + }), }) }), ).pipe(Layer.provide(realStore)) @@ -198,11 +243,19 @@ function twoInstanceLayer(input: TwoInstanceInput) { Layer.provide(events), Layer.provide(database), ) + // The session projector runs the SessionEvent.Moved projection (session row + // + workflow directory re-stamp — the #269 atomic-adoption transition), so + // publishing Moved through the bridge converges the workflow stamps in the + // same durable transaction, exactly as production. + const sessionProjector = SessionProjector.layer.pipe( + Layer.provide(events), + Layer.provide(database), + ) const dag = Dag.layer.pipe( Layer.provide(bridge), Layer.provide(store), ) - const base = Layer.mergeAll(database, events, bridge, store, projector, dag, status) + const base = Layer.mergeAll(database, events, bridge, store, projector, sessionProjector, dag, status) const childTitles = new Map() const created: string[] = [] const session = Layer.mock(Session.Service, { @@ -363,6 +416,12 @@ function runTwoInstanceGuardTest( readonly released: { value: boolean } readonly calls: { value: number } } + readonly parkAdoptionClaim?: { + readonly wait: Promise + readonly parked: { value: boolean } + readonly calls: { value: number } + readonly skip: number + } }, test: (services: TwoInstanceServices) => Effect.Effect, beforeInit?: (services: { readonly database: Database.Interface }) => Effect.Effect, @@ -447,6 +506,7 @@ function runTwoInstanceGuardTest( failGetWorkflow: options.failGetWorkflow, parkGetNodes: options.parkGetNodes, parkWakeDelivery: options.parkWakeDelivery, + parkAdoptionClaim: options.parkAdoptionClaim, })), Effect.provideService(InstanceRef, { directory: directoryA, @@ -765,23 +825,44 @@ describe("DAG execution-location static contract (DAG-LOC-01 R7)", () => { expect(unguarded).toEqual([]) }) - // #238 probe ⑤ (R7-ext): the directory stamp is WRITE-ONCE and every - // revalidation site that acts on a possibly-stale in-memory entry carries - // the ownership authority. + // #238 probe ⑤ (R7-ext): the directory stamp is WRITE-ONCE EXCEPT the single + // whitelisted SessionEvent.Moved re-stamp (#269 — an invariant UPDATE mandated + // by #269's own wording, not a weakening), and every revalidation site that + // acts on a possibly-stale in-memory entry carries the ownership authority. it("R7-ext: the stamp is write-once and every revalidation site carries the authority", () => { const sources = [...readDagSources(opencodeDagSrc), ...readDagSources(coreDagSrc)] expect(sources.length).toBeGreaterThan(20) const codeLines = (source: string) => source.split("\n").filter((line) => !line.trim().startsWith("//")) - // (a) Write-once: no UPDATE writes the directory column anywhere in the - // dag trees. The stamp lands via the projector's INSERT (workflow create - // → onConflictDoNothing); a `.set({ directory })` would re-stamp a live - // row and violate the create-time-pins-ownership invariant. + // (a) TWO-WRITER WHITELIST (amended for #269 — an invariant UPDATE, not a + // weakening): the directory stamp is WRITE-ONCE except for exactly ONE + // sanctioned re-stamp. The ONLY directory writers are + // 1. the dag projector's WorkflowCreated INSERT (create-time stamp, + // onConflictDoNothing — a replay can never rewrite an existing stamp), + // 2. the session projector's SessionEvent.Moved re-stamp (#269 — the stamp + // moves WITH the session in one transaction, payload-sourced, no + // SessionTable read). + // Negative half (unchanged): no `.set({ directory })` anywhere in the dag + // trees — a dag-side re-stamp would violate create-time-pins-ownership. const writesDirectory = sources .filter(({ source }) => /\.set\(\{[\s\S]{0,300}?\bdirectory\s*:/.test(source)) .map(({ file }) => file) expect(writesDirectory).toEqual([]) + // Positive whitelist half: the Moved re-stamp MUST exist in the session + // projector (outside the dag trees) and MUST be the payload-sourced + // WorkflowTable re-stamp (event.data.location.directory). Mandated by #269's + // "the directory stamp must move WITH the session in one transaction" — so + // this probe pins it; removing the re-stamp is a regression here. + const sessionProjectorSource = readFileSync( + path.resolve(import.meta.dir, "../../../../packages/core/src/session/projector.ts"), + "utf8", + ) + expect( + /\.update\(WorkflowTable\)[\s\S]{0,400}?\.set\(\{\s*directory:\s*event\.data\.location\.directory/.test( + sessionProjectorSource, + ), + ).toBe(true) // (b) Revalidation sites. Each region must carry an executable ownership // authority call (ownsWorkflow / ownsSession), located by semantic anchors @@ -1350,11 +1431,11 @@ describe("DAG-LOC-01 issue #238 evidence probes", () => { ) }) - it("C4: a moved session's mixed stamps leave NO directory owner (pre-clustering wedge pin)", async () => { + it("C4: a moved session's workflow stamps move WITH the session — exactly one live owner, the new durable directory (#269 resolution)", async () => { await Effect.runPromise( runTwoInstanceGuardTest( {}, - ({ dag, store, database }) => + ({ dag, store, bridge }) => Effect.gen(function* () { // wf1 created while SES_A lives in DIR_A → stamped DIR_A. const wf1 = yield* dag.create({ @@ -1364,11 +1445,14 @@ describe("DAG-LOC-01 issue #238 evidence probes", () => { config: { name: "c4a", nodes: [node()] }, }) expect((yield* store.getWorkflow(wf1))?.directory).toBe(DIR_A) - // Move the session to DIR_B via the durable session row. - yield* database.db.update(SessionTable) - .set({ directory: DIR_B as never }) - .where(eq(SessionTable.id, SES_A as never)) - .run().pipe(Effect.orDie) + // Move the session to DIR_B via the durable SessionEvent.Moved. The + // session projector re-stamps every workflow row of the session in + // the SAME transaction (payload-sourced, no SessionTable read), so + // the stamp moves WITH the session — no mixed-stamp window. + yield* moveSession(bridge, SES_A, DIR_B) + // The pre-move workflow's stamp MOVED WITH the session (write-once + // except the whitelisted Moved re-stamp). + expect((yield* store.getWorkflow(wf1))?.directory).toBe(DIR_B) // wf2 created after the move → session-sourced stamp = DIR_B. const wf2 = yield* dag.create({ projectID: PROJECT_ID, @@ -1377,13 +1461,169 @@ describe("DAG-LOC-01 issue #238 evidence probes", () => { config: { name: "c4b", nodes: [node()] }, }) expect((yield* store.getWorkflow(wf2))?.directory).toBe(DIR_B) - // Mixed stamps: the session's workflow rows no longer agree on a - // single directory, so NO instance owns the session's wakes. This - // pins the pre-clustering create-time-stamp semantics (the wedge - // is pinned, not fixed — re-stamping on SessionEvent.Moved is out - // of scope for the single-authority design). + // Resolution (acceptance #269): the session's rows agree on ONE + // directory, so exactly one instance owns it — the session's NEW + // durable directory. The old directory loses ownership (the wedge is + // resolved, not pinned), and the two directories never own + // simultaneously (no cross-directory double-adoption). + expect(yield* DagLocation.ownsSession(SES_A, DIR_B)).toBe(true) expect(yield* DagLocation.ownsSession(SES_A, DIR_A)).toBe(false) - expect(yield* DagLocation.ownsSession(SES_A, DIR_B)).toBe(false) + // Per-workflow ownership converges the same way: every workflow is + // owned by DIR_B only. + expect(yield* DagLocation.ownsWorkflow(wf1, DIR_B)).toBe(true) + expect(yield* DagLocation.ownsWorkflow(wf1, DIR_A)).toBe(false) + expect(yield* DagLocation.ownsWorkflow(wf2, DIR_B)).toBe(true) + expect(yield* DagLocation.ownsWorkflow(wf2, DIR_A)).toBe(false) + }), + ), + ) + }) + + it("C8: a moved session's wedged wake is delivered by the NEW owner immediately — no idle event, no restart (#269 bounded-time resolution)", async () => { + const parked = { value: false } + const released = { value: false } // never released: the parked admission stays parked; scope cleanup interrupts it + const calls = { value: 0 } + const wait = new Promise(() => {}) + await Effect.runPromise( + runTwoInstanceGuardTest( + { parkWakeDelivery: { wait, parked, released, calls } }, + ({ bridge, database, initA, initB }) => + Effect.gen(function* () { + yield* initA + yield* initB + // Re-arm the seeded terminal wake AFTER both startup sweeps passed + // over it (seeded wake_reported=true), so the ONLY later trigger is + // the Moved subscription's re-fork (no idle event, no restart). + yield* database.db.update(WorkflowTable) + .set({ wake_reported: false }) + .where(eq(WorkflowTable.id, "c8-wake")) + .run().pipe(Effect.orDie) + // Move SES_A → DIR_B. The Moved projection re-stamps the wake row to + // DIR_B (exactly one owner), and the NEW owner's Moved subscription + // re-forks tryDeliverWake — the wake delivers immediately (bounded + // time), no wake stays wedged behind the old mixed stamps. + yield* moveSession(bridge, SES_A, DIR_B) + yield* pollWithTimeout( + Effect.sync(() => (parked.value && calls.value === 1 ? true : undefined)), + "moved wake was not delivered to the new owner within bounded time (expected exactly one delivery attempt from the new directory)", + ) + // Exactly one delivery: after the re-stamp the old directory's + // ownsSession is false, so only the NEW owner re-forks the drain — + // no cross-directory double delivery. + expect(calls.value).toBe(1) + }), + ({ database }) => + Effect.gen(function* () { + yield* database.db.insert(WorkflowTable).values({ + id: "c8-wake", + project_id: PROJECT_ID as never, + session_id: SES_A as never, + directory: DIR_A as never, + title: "terminal wake c8", + status: "failed", + config: "{}", + seq: 1, + wake_reported: true, + }).run().pipe(Effect.orDie) + }), + ), + ) + }) + + it("C9: the convergence migration moves pre-fix divergent stamps WITH the session and preserves fail-closed NULL (#269 backfill)", async () => { + await Effect.runPromise( + runTwoInstanceGuardTest( + {}, + ({ database }) => + Effect.gen(function* () { + // Simulate a pre-fix moved session: the durable session row already + // points at DIR_B, but one workflow still carries the pre-move stamp + // DIR_A (the v1.0.13 mixed-stamp wedge input) and one is a NULL-stamp + // zombie (legacy, must stay fail-closed). + yield* database.db.update(SessionTable) + .set({ directory: DIR_B as never }) + .where(eq(SessionTable.id, SES_A as never)) + .run().pipe(Effect.orDie) + for (const [id, dir] of [ + ["c9-wf-stale", DIR_A], + ["c9-wf-zombie", null], + ] as const) { + yield* database.db.insert(WorkflowTable).values({ + id, + project_id: PROJECT_ID as never, + session_id: SES_A as never, + directory: dir as never, + title: "convergence probe", + status: "failed", + config: "{}", + seq: 1, + wake_reported: true, + }).run().pipe(Effect.orDie) + } + // Apply the convergence migration against the current db (twice — the + // second application must be a no-op: idempotent convergence). + for (const _ of [0, 1]) { + yield* database.db + .transaction((tx) => workflowDirectoryConvergence.up(tx as never)) + .pipe(Effect.orDie) + } + const stamps = yield* database.db.select({ id: WorkflowTable.id, directory: WorkflowTable.directory }) + .from(WorkflowTable) + .all().pipe(Effect.orDie) + const byId = new Map(stamps.map((row) => [row.id, row.directory])) + // The divergent stamp converged to the session's CURRENT directory... + expect(byId.get("c9-wf-stale")).toBe(DIR_B) + // ...and the NULL zombie stayed NULL (fail-closed is preserved). + expect(byId.get("c9-wf-zombie")).toBeNull() + }), + ), + ) + }) + + it("C7: a deletion committed after a passed admission check fences the in-flight spawn — no post-deletion child survives (#270 deletion race)", async () => { + const parked = { value: false } + const calls = { value: 0 } + let release: () => void = () => {} + const wait = new Promise((resolve) => { + release = resolve + }) + await Effect.runPromise( + runTwoInstanceGuardTest( + { parkAdoptionClaim: { wait, parked, calls, skip: 1 } }, + ({ dag, database, initA, cancelsA, childPromptsA }) => + Effect.gen(function* () { + yield* initA + // A ready-node workflow: the adoption claim (skipped by the gate) runs + // first, then spawnReady admits the node and the SECOND claim — the + // window-2 spawn-admission fence — parks just before the child session + // would materialize. + const dagID = yield* dag.create({ + projectID: PROJECT_ID, + sessionID: SES_A, + title: "deletion race", + config: { name: "c7", nodes: [node()] }, + }) + yield* pollWithTimeout( + Effect.sync(() => (parked.value ? true : undefined)), + "the spawn admission claim never parked", + ) + // Deletion committed AFTER the ownership/admission check passed: drop + // the workflow row directly (FK cascade wipes the nodes) WITHOUT + // publishing SessionV1.Event.Deleted, so no in-memory sweep interrupts + // the parked admission — the fence must hold purely because the atomic + // claim re-reads the row. + yield* database.db.delete(WorkflowTable) + .where(eq(WorkflowTable.id, dagID as never)) + .run().pipe(Effect.orDie) + // Release: the claim conditional UPDATE matches no row (cascade already + // removed it) → the spawn aborts BEFORE sessions.create — nothing a + // deleted workflow may run survives the admission window. + release() + yield* Effect.sleep("400 millis") + // Fenced: no post-deletion spawn survived — no child prompt ran and no + // child session was created-then-cancelled. + expect(Option.isNone(yield* Queue.take(childPromptsA).pipe(Effect.timeoutOption("500 millis")))).toBe(true) + expect(cancelsA).toEqual([]) }), ), ) diff --git a/packages/opencode/test/dag/dag-structured-output.test.ts b/packages/opencode/test/dag/dag-structured-output.test.ts index 5b1e1b8c5f..2bc0175f4c 100644 --- a/packages/opencode/test/dag/dag-structured-output.test.ts +++ b/packages/opencode/test/dag/dag-structured-output.test.ts @@ -20,6 +20,7 @@ function makeEventTracker() { const events: TrackedEvent[] = [] capturedStore = new Map() const storeStub: Partial = { + tryClaimAdoption: () => Effect.succeed(true), getNode: Effect.fn("s")((_workflowID: string, nodeID: string) => Effect.sync(() => ({ ...makeNodeRow({ id: nodeID }), capturedOutput: capturedStore.get(nodeID) }))), setCapturedOutput: Effect.fn("s")((_childSessionID: string, payload: unknown) => diff --git a/packages/opencode/test/dag/spawn-completion.test.ts b/packages/opencode/test/dag/spawn-completion.test.ts index 585bc8c45f..799379e39d 100644 --- a/packages/opencode/test/dag/spawn-completion.test.ts +++ b/packages/opencode/test/dag/spawn-completion.test.ts @@ -16,7 +16,7 @@ type TrackedEvent = { type: string; dagID: string; nodeID: string; output?: unkn function makeEventTracker() { const events: TrackedEvent[] = [] const dagLayer = Layer.mock(Dag.Service, { - store: {} as DagStore.Interface, + store: { tryClaimAdoption: () => Effect.succeed(true) } as unknown as DagStore.Interface, nodeQueued: Effect.fn("stub.nodeQueued")((dagID: string, nodeID: string) => Effect.sync(() => events.push({ type: "nodeQueued", dagID, nodeID })), ), @@ -285,7 +285,7 @@ describe("spawnNode terminalization during spawn window", () => { const events: TrackedEvent[] = [] let cancelCalled = false const dagLayer = Layer.mock(Dag.Service, { - store: {} as DagStore.Interface, + store: { tryClaimAdoption: () => Effect.succeed(true) } as unknown as DagStore.Interface, nodeQueued: () => Effect.void, nodeStarted: () => Effect.fail(new TerminalViolationError("node-1", "failed", "running")), nodeCompleted: Effect.fn("stub.nodeCompleted")((dagID: string, nodeID: string) => @@ -322,7 +322,7 @@ describe("spawnNode terminalization during spawn window", () => { let cancelCalled = false let promptCalled = false const dagLayer = Layer.mock(Dag.Service, { - store: {} as DagStore.Interface, + store: { tryClaimAdoption: () => Effect.succeed(true) } as unknown as DagStore.Interface, nodeQueued: () => Effect.void, nodeStarted: () => Effect.fail(new Error("nodeStarted write failed")), nodeCompleted: Effect.fn("stub.nodeCompleted")((dagID: string, nodeID: string) =>