From 9ba6efab5af1e6835130062e9b3122a8b349c805 Mon Sep 17 00:00:00 2001 From: Lex Date: Thu, 13 Aug 2026 12:43:29 +0800 Subject: [PATCH 1/8] fix(dag): enforce execution-location ownership on adoption, recovery, and wake (DAG-LOC-01) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The DAG runtime is per-directory InstanceState, but the durable store, the event bus, and the workflow rows are process-global. Guards keyed on the PROJECT ID let sibling worktrees of one project (same id, distinct directories) all adopt, recover-cancel, wake, and spawn for each other's workflows. This change installs a single execution-location authority and routes every adoption/recovery/wake guard through it. Authority — packages/opencode/src/dag/location.ts (single module): - ownsWorkflow(workflowID, directory): re-reads the durable workflow row on every check; project id is the fast-reject, the stamped DIRECTORY (realpath-normalized, raw-path fallback) is the deciding guard. R6 identity revalidation falls out of the re-read: a repainted project_id stops the stale in-memory entry from publishing transitions. - ownsSession(sessionID, directory): every durable workflow row of the session must match (vacuous-true for workflow-less sessions so goal-only sessions keep working). Key lives on the workflow row only — no session-table reads (R7's negative half). - Database is resolved lazily via Effect.serviceOption so the loops' static layer requirements stay unchanged (the optional-cross-dependency pattern). Join vs column: the round-1 analysis allowed either. R7 mandates the key on the workflow row itself and forbids session.directory reads in dag sources, so the column wins: WorkflowTable.directory, stamped at dag.create from the creating instance (WorkflowCreated.directory, optional for legacy decodes), plus migration 20260813040429_workflow_directory (generated by script/migration.ts) with a session-join backfill so in-flight workflows survive upgrades. A NULL stamp matches no instance (never adopted). Guard regions replaced in packages/opencode/src/dag/runtime/loop.ts: - recoverWorkflow (~L328): projectId guard -> ownsWorkflow(wf.id, ctx.directory) - recoverOrphanPending (~L439): same replacement - WorkflowStarted first-wave adoption (~L487): same replacement - startup wake sweep (~L1308): snapshot projectId check -> ownsSession - tryDeliverWake entry (~L1048, previously unguarded): ownsSession - checkCompletion (~L282): new revalidation gate (R6) - SessionV1.Event.Deleted teardown subscription (~L1233, R5): drops the session's runtime entries and interrupts their fibers/watchers, mirroring the workflow-terminal cleanup pattern GoalLoop idle trigger (packages/opencode/src/goal/loop.ts ~L120): routed through ownsSession — aligns the idle path with the directory-scoped goal scan. Probes (test/dag/dag-location-guards.test.ts): RED 7/7 before (/tmp/dag-loc-red-full.log), GREEN 7/7 after — R1 adoption, R2 startup recovery, R3 idle wake, R4 startup sweep, R5 deletion teardown, R6 identity migration, R7 static contract. R5/R6's negative-window assertions used pollWithTimeout (a positive-wait tool whose timeout errors the effect, so the "nothing must happen" outcome could never pass); their mechanics were fixed to settle-then-sleep-and-assert with identical intent. Pre-existing seeds gained the directory stamp (dag-wake-integration, dag-adoption-step-races, dag-orphan-pending-recovery, workflow-tool/summary-publisher fixtures). Mutations: bypassing the tryDeliverWake authority guard -> R3 red (6 pass); removing the Deleted teardown subscription -> R5 red (6 pass); both restored. Verification: packages/opencode test/dag + test/goal = 560 pass / 0 fail; core dag-projector-drift + dag-store-summaries pass; test:dag-core pass; test:httpapi 227 pass / 0 fail; bun typecheck clean (root); bun lint 4850 (ratchet tightened 4852 -> 4850: probe harness's `as never` fixture shims file-scoped suppressed like the dag-loop-guards template; two pre-existing `as never` casts replaced); check:generated clean for sdk/js and client. Co-Authored-By: Claude --- package.json | 4 +- packages/core/schema.json | 14 +- packages/core/src/dag/projector.ts | 4 + packages/core/src/dag/sql.ts | 6 + packages/core/src/dag/store.ts | 3 + packages/core/src/database/migration.gen.ts | 1 + .../20260813040429_workflow_directory.ts | 24 + packages/core/src/database/schema.gen.ts | 1 + packages/opencode/src/dag/dag.ts | 7 + packages/opencode/src/dag/location.ts | 105 +++ packages/opencode/src/dag/runtime/loop.ts | 87 ++- packages/opencode/src/goal/loop.ts | 18 +- .../test/dag/dag-adoption-step-races.test.ts | 3 +- .../test/dag/dag-location-guards.test.ts | 693 ++++++++++++++++++ .../dag/dag-orphan-pending-recovery.test.ts | 1 + .../dag-summary-publisher-behavior.test.ts | 1 + .../test/dag/dag-wake-integration.test.ts | 5 +- .../opencode/test/dag/workflow-tool.test.ts | 5 + packages/schema/src/dag-event.ts | 5 + 19 files changed, 967 insertions(+), 20 deletions(-) create mode 100644 packages/core/src/database/migration/20260813040429_workflow_directory.ts create mode 100644 packages/opencode/src/dag/location.ts create mode 100644 packages/opencode/test/dag/dag-location-guards.test.ts diff --git a/package.json b/package.json index 9a8ee27513..0e60f73af9 100644 --- a/package.json +++ b/package.json @@ -5,7 +5,7 @@ "private": true, "type": "module", "packageManager": "bun@1.3.14", - "_lint_ratchet_note": "Ratchet lowered to 4852 (the pre-batch-A CI baseline) after replacing the `as never` test-data idiom in the three dag timeout/escalation test files (dag-deadline-extended, dag-escalation-clear-flag, dag-timeout-escalation) with schema brand makers (Project.ID.make, Session.ID.make, DagEvent.NodeID.make, AbsolutePath.make) and fully-typed InstanceRef/Session mocks — their no-unsafe-type-assertion warnings are gone. CI lints ~3 extra install/platform-generated artifacts on an identical tree, adding ~10 same-category type-aware warnings (~4841 CI vs ~4831 local, 0 errors) — NOT new code warnings; 4852 keeps a small margin over the projected CI count. When you fix existing warnings locally, lower --max-warnings to match so the gate keeps tightening. See .oxlintrc.json header for the ratchet contract.", + "_lint_ratchet_note": "Ratchet lowered to 4850 after DAG-LOC-01: the new execution-location guards (dag/location.ts authority, DagLoop guard replacements, GoalLoop idle guard) and the dag-location-guards probe harness carry no net new warnings \u2014 the probe harness's `as never` fixture shims are file-scoped suppressed (mirrors dag-loop-guards.test.ts idiom), and two pre-existing `directory: process.cwd() as never` session-seed casts in dag-wake-integration/dag-adoption-step-races were replaced with plain strings (session.directory accepts them). CI lints ~3 extra install/platform-generated artifacts on an identical tree, adding ~10 same-category type-aware warnings (~4841 CI vs ~4831 local, 0 errors) \u2014 NOT new code warnings; 4850 keeps a small margin over the projected CI count. When you fix existing warnings locally, lower --max-warnings to match so the gate keeps tightening. See .oxlintrc.json header for the ratchet contract.", "scripts": { "dev": "bun run --cwd packages/opencode --conditions=browser src/index.ts", "dev:desktop": "bun --cwd packages/desktop dev", @@ -13,7 +13,7 @@ "dev:console": "ulimit -n 10240 2>/dev/null; bun run --cwd packages/console/app dev", "dev:stats": "bun sst shell --stage=production -- bun run --cwd packages/stats/app dev", "dev:storybook": "bun --cwd packages/storybook storybook", - "lint": "oxlint --max-warnings=4852", + "lint": "oxlint --max-warnings=4850", "typecheck": "bun turbo typecheck", "upgrade-opentui": "bun run script/upgrade-opentui.ts", "postinstall": "bun run --cwd packages/core fix-node-pty", diff --git a/packages/core/schema.json b/packages/core/schema.json index cd735b190a..a330c6136c 100644 --- a/packages/core/schema.json +++ b/packages/core/schema.json @@ -1,9 +1,9 @@ { "version": "7", "dialect": "sqlite", - "id": "7e8e00e9-7bbb-443e-996b-f646ec030c2b", + "id": "4142b961-0712-4834-b475-16ea4a74c43c", "prevIds": [ - "cce2163c-da01-4239-86fa-776d48a58d89" + "7e8e00e9-7bbb-443e-996b-f646ec030c2b" ], "ddl": [ { @@ -762,6 +762,16 @@ "entityType": "columns", "table": "workflow" }, + { + "type": "text", + "notNull": false, + "autoincrement": false, + "default": null, + "generated": null, + "name": "directory", + "entityType": "columns", + "table": "workflow" + }, { "type": "text", "notNull": true, diff --git a/packages/core/src/dag/projector.ts b/packages/core/src/dag/projector.ts index ba9d76097a..f255e43311 100644 --- a/packages/core/src/dag/projector.ts +++ b/packages/core/src/dag/projector.ts @@ -87,6 +87,10 @@ export const layer = Layer.effectDiscard( id: event.data.dagID, project_id: event.data.projectID, session_id: event.data.sessionID, + // DAG-LOC-01: the execution-location key, stamped at dag.create. + // A legacy event without the field projects to NULL — a row that + // matches no instance directory and is never adopted. + directory: event.data.directory ?? null, title: event.data.title, status: event.data.status, config: event.data.config, diff --git a/packages/core/src/dag/sql.ts b/packages/core/src/dag/sql.ts index 9e6b9d80d0..5268b82072 100644 --- a/packages/core/src/dag/sql.ts +++ b/packages/core/src/dag/sql.ts @@ -31,6 +31,12 @@ 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). + directory: text(), title: text().notNull(), status: text().notNull(), config: text().notNull(), // YAML string diff --git a/packages/core/src/dag/store.ts b/packages/core/src/dag/store.ts index 52cb17e338..07405305cb 100644 --- a/packages/core/src/dag/store.ts +++ b/packages/core/src/dag/store.ts @@ -17,6 +17,8 @@ export interface WorkflowRow { id: string projectId: string sessionId: string + /** Execution-location key (DAG-LOC-01): the creating instance's directory. */ + directory: string | null title: string status: string config: string @@ -83,6 +85,7 @@ const mapWorkflow = (r: typeof WorkflowTable.$inferSelect): WorkflowRow => ({ id: r.id, projectId: r.project_id, sessionId: r.session_id, + directory: r.directory, title: r.title, status: r.status, config: r.config, diff --git a/packages/core/src/database/migration.gen.ts b/packages/core/src/database/migration.gen.ts index ddffdb838b..8ffc6c9fdf 100644 --- a/packages/core/src/database/migration.gen.ts +++ b/packages/core/src/database/migration.gen.ts @@ -53,5 +53,6 @@ export const migrations = ( import("./migration/20260805094942_workflow_node_escalation_pending"), import("./migration/20260811060000_goal_outcome"), import("./migration/20260813020344_bored_skaar"), + import("./migration/20260813040429_workflow_directory"), ]) ).map((module) => module.default) satisfies DatabaseMigration.Migration[] diff --git a/packages/core/src/database/migration/20260813040429_workflow_directory.ts b/packages/core/src/database/migration/20260813040429_workflow_directory.ts new file mode 100644 index 0000000000..8f8d7e0d72 --- /dev/null +++ b/packages/core/src/database/migration/20260813040429_workflow_directory.ts @@ -0,0 +1,24 @@ +import { Effect } from "effect" +import type { DatabaseMigration } from "../migration" + +export default { + id: "20260813040429_workflow_directory", + up(tx) { + return Effect.gen(function* () { + yield* tx.run(`ALTER TABLE \`workflow\` ADD \`directory\` text;`) + // DAG-LOC-01 backfill: the ownership key is the workflow row's own + // directory, so existing installs must carry the owning session's + // directory forward or every in-flight workflow would turn foreign + // (never adopted / orphan-pending rows never terminalized). Rows whose + // session is already gone stay NULL — conservative: NULL matches no + // instance directory and is never adopted. + yield* tx.run(` + UPDATE \`workflow\` + SET \`directory\` = ( + SELECT \`directory\` FROM \`session\` WHERE \`session\`.\`id\` = \`workflow\`.\`session_id\` + ) + WHERE \`directory\` IS NULL; + `) + }) + }, +} satisfies DatabaseMigration.Migration diff --git a/packages/core/src/database/schema.gen.ts b/packages/core/src/database/schema.gen.ts index 8cc88289be..7504e0fbea 100644 --- a/packages/core/src/database/schema.gen.ts +++ b/packages/core/src/database/schema.gen.ts @@ -105,6 +105,7 @@ export default { \`id\` text PRIMARY KEY, \`project_id\` text NOT NULL, \`session_id\` text NOT NULL, + \`directory\` text, \`title\` text NOT NULL, \`status\` text NOT NULL, \`config\` text NOT NULL, diff --git a/packages/opencode/src/dag/dag.ts b/packages/opencode/src/dag/dag.ts index d38dced3fb..f818e83b05 100644 --- a/packages/opencode/src/dag/dag.ts +++ b/packages/opencode/src/dag/dag.ts @@ -32,6 +32,7 @@ import { } from "./admission" import { unresolvedReviewOutcomes } from "./review-lifecycle" import { DagValidation, StructuralValidationError } from "./validation" +import { DagLocation } from "./location" export { StructuralValidationError } from "./validation" @@ -400,6 +401,12 @@ export const layer = Layer.effect( config: JSON.stringify(durableConfig), status: "pending", timestamp: ts, + // DAG-LOC-01: stamp the execution-location key (the creating + // instance's canonical directory) on the workflow row at create. + // The DagLoop ownership guards decide on this stamp; sibling + // worktrees of the same project carry distinct directories and are + // mutually foreign. + directory: yield* DagLocation.stampDirectory(), }) for (const node of durableConfig.nodes) { yield* events.publish(DagEvent.NodeRegistered, { diff --git a/packages/opencode/src/dag/location.ts b/packages/opencode/src/dag/location.ts new file mode 100644 index 0000000000..0f037eb051 --- /dev/null +++ b/packages/opencode/src/dag/location.ts @@ -0,0 +1,105 @@ +export * as DagLocation from "./location" + +/** + * DAG-LOC-01 — the execution-location authority. + * + * The DAG runtime (DagLoop, GoalLoop) is per-directory InstanceState, but the + * durable store, the event bus, and the workflow rows are process-global. A + * multi-directory server (sibling worktrees of ONE project — same project id) + * would otherwise let every instance adopt, recover-cancel, wake, and spawn + * for every workflow. This module is the SINGLE authority that decides which + * instance may act: the location key is the DIRECTORY, not the project id. + * + * The key lives on the workflow row itself (WorkflowTable.directory), stamped + * at dag.create from the creating instance's directory. Ownership predicates + * re-read the durable row on every check, so a row whose durable identity was + * repainted (identity migration) or deleted stops matching and its in-memory + * runtime entry loses the right to publish transitions. + * + * Callers (the loops) pass their own instance directory and know nothing about + * the SQL or the realpath internals. The Database service is resolved lazily + * via serviceOption so the loops' static requirements stay unchanged (the + * optional-cross-dependency pattern); production graphs always carry it. + */ + +import { eq } from "drizzle-orm" +import { realpathSync } from "node:fs" +import { Effect } from "effect" +import { Database } from "@opencode-ai/core/database/database" +import { WorkflowTable } from "@opencode-ai/core/dag/sql" +import { InstanceRef } from "@/effect/instance-ref" + +/** + * Canonical execution-location key: the directory's realpath when resolvable, + * else the raw path (test directories like /wtA do not exist on disk; the + * fallback keeps the comparison a plain string equality in that case). Both + * stamping (dag.create) and checking go through this, so the two sides are + * always comparable under the same normalization. + */ +export const canonicalDirectory = (directory: string): string => { + try { + return realpathSync(directory) + } catch { + return directory + } +} + +/** The directory to stamp on a workflow created by the ambient instance. */ +export const stampDirectory = (): Effect.Effect => + Effect.map(InstanceRef, (instance) => (instance ? canonicalDirectory(instance.directory) : "")) + +/** + * Owns the workflow iff its DURABLE row (re-read on every check) still belongs + * to the ambient instance: the project id matches (fast-reject + R6 identity + * revalidation — a repainted project_id must not keep driving the old entry) + * and the stamped directory matches the caller's directory (the deciding + * guard: sibling worktrees share the project id). Fail-closed: a missing + * instance or a row without a stamp is never adopted. + */ +export const ownsWorkflow = (workflowID: string, directory: string): Effect.Effect => + Effect.gen(function* () { + const instance = yield* InstanceRef + if (!instance) return false + const db = yield* Effect.serviceOption(Database.Service) + if (db._tag === "None") return false + const row = yield* db.value.db + .select() + .from(WorkflowTable) + .where(eq(WorkflowTable.id, workflowID)) + .get() + .pipe(Effect.orDie) + if (!row) return false + if (row.project_id !== instance.project.id) return false + return row.directory !== null && canonicalDirectory(row.directory) === canonicalDirectory(directory) + }) + +/** + * Owns the session iff every durable workflow row of the session still belongs + * to the ambient instance (same project id + directory conjunct as + * ownsWorkflow). Vacuous-true when the session has no workflow rows: there is + * no wake data to deliver and goal-only sessions predate workflow stamping. + * Also vacuous-true when the Database service is absent from the runtime graph + * (synthetic goal tests; every production graph carries it) — ownership cannot + * be disproven there and the gate must not silently disable pre-existing + * loops. The workflow-row key keeps this module free of session-table reads: + * the execution-location key belongs on the workflow row itself (R7). + */ +export const ownsSession = (sessionID: string, directory: string): Effect.Effect => + Effect.gen(function* () { + const instance = yield* InstanceRef + if (!instance) return false + const db = yield* Effect.serviceOption(Database.Service) + if (db._tag === "None") return true + const rows = yield* db.value.db + .select() + .from(WorkflowTable) + .where(eq(WorkflowTable.session_id, sessionID)) + .all() + .pipe(Effect.orDie) + return rows.every( + (row) => + row.project_id === instance.project.id && + row.directory !== null && + canonicalDirectory(row.directory) === canonicalDirectory(directory), + ) + }) diff --git a/packages/opencode/src/dag/runtime/loop.ts b/packages/opencode/src/dag/runtime/loop.ts index 5da1c80553..a7509a0f85 100644 --- a/packages/opencode/src/dag/runtime/loop.ts +++ b/packages/opencode/src/dag/runtime/loop.ts @@ -5,11 +5,13 @@ export * as DagLoop from "./loop" import { Cause, Effect, Layer, Context, Stream, Semaphore, Fiber, Option, DateTime, Clock } from "effect" import { LayerNode } from "@opencode-ai/core/effect/layer-node" +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 { SessionStatusEvent } from "@opencode-ai/schema/session-status-event" import { DagStore } from "@opencode-ai/core/dag/store" +import { DagLocation } from "../location" import { WorkflowRuntime, toSchedulingNodes } from "@opencode-ai/core/dag/core/scheduling" import { isNodeTerminalStatus, isWorkflowTerminalStatus } from "@opencode-ai/core/dag/core/types" import { Dag, type WorkflowConfig, parseWorkflowConfig } from "../dag" @@ -293,6 +295,12 @@ const serviceLayer = Layer.effect( (node) => !isNodeTerminalStatus(node.status as never) && !entry.runtime.containsNode(node.id), ) if (hasUnseenActiveNode) return + // R6 identity revalidation: the ownership predicate re-reads the + // durable row on every check. A workflow whose durable identity was + // repainted (identity migration) or whose location moved away no + // longer belongs to this in-memory entry — the stale entry must not + // publish a workflow transition (complete/fail) for it. + if (!(yield* DagLocation.ownsWorkflow(dagID, ctx.directory))) return const wf = yield* store.getWorkflow(dagID).pipe(Effect.orDie) if (wf && isWorkflowTerminalStatus(wf.status as never)) return // A required-node failure is a workflow FAILURE, not a cancellation — @@ -333,9 +341,12 @@ const serviceLayer = Layer.effect( const recoverWorkflow = Effect.fn("DagLoop.recoverWorkflow")(function* (wf: DagStore.WorkflowRow) { // Cross-instance guard: DagLoop is per-directory InstanceState but the // event bus and store are process-global. Only the instance whose - // project owns the workflow may adopt it — otherwise a multi-directory - // server spawns children under a foreign directory context. - if (wf.projectId !== ctx.project.id) return + // DIRECTORY owns the workflow may adopt it — otherwise a + // multi-directory server spawns children under a foreign directory + // context. The execution-location authority (dag/location.ts) + // re-reads the durable row: the project id is a fast-reject, the + // stamped directory is the deciding guard. + if (!(yield* DagLocation.ownsWorkflow(wf.id, ctx.directory))) return const dagID = wf.id // Idempotency guard: the startup scan and the WorkflowReplanned // handler's re-adoption path can both reach here for the same @@ -442,9 +453,10 @@ const serviceLayer = Layer.effect( // durable WorkflowFailed event; cancelled is reserved for explicit // user/agent cancels (see the checkCompletion attribution comment). const recoverOrphanPending = Effect.fn("DagLoop.recoverOrphanPending")(function* (wf: DagStore.WorkflowRow) { - // Same cross-instance guard as recoverWorkflow: only the owning - // project's instance may dispose of the orphan. - if (wf.projectId !== ctx.project.id) return + // Same cross-instance guard as recoverWorkflow, through the same + // execution-location authority: only the instance whose DIRECTORY + // owns the workflow may dispose of the orphan. + if (!(yield* DagLocation.ownsWorkflow(wf.id, ctx.directory))) return const dagID = wf.id if (runtimes.has(dagID) || recovering.has(dagID)) return // Reserve the adoption slot for the whole terminalization sequence: @@ -489,10 +501,11 @@ const serviceLayer = Layer.effect( // already failed — adopting it would rebuild a runtime and start // scheduling nodes on a dead workflow. Accept running rows only. if (wf.status !== "running") return - // Cross-instance guard: only the owning project's instance adopts + // Cross-instance guard via the execution-location authority: + // only the instance whose DIRECTORY owns the workflow adopts // (see recoverWorkflow). First-wave spawns must not race across // directory contexts. - if (wf.projectId !== ctx.project.id) return + if (!(yield* DagLocation.ownsWorkflow(dagID, ctx.directory))) return const config = parseWorkflowConfig(wf.config) const nodes = yield* store.getNodes(dagID) const maxConcurrency = Math.max(1, config?.max_concurrency ?? Dag.DEFAULT_WORKFLOW_CONFIG.maxConcurrency) @@ -1079,6 +1092,14 @@ const serviceLayer = Layer.effect( let tryDeliverWake: (sessionID: string) => Effect.Effect = () => Effect.void tryDeliverWake = Effect.fn("DagLoop.tryDeliverWake")(function* (sessionID: string) { + // Cross-instance guard via the execution-location authority: wake + // delivery is store-global (idle Status events, node-terminal + // handlers, the startup sweep). Only the instance whose DIRECTORY + // owns the session's workflows may deliver its wakes — sibling + // worktrees of the same project must ignore each other's idle + // sessions. The guard also covers the workflow-terminal stimulus + // after a session deletion (the durable rows are gone). + if (!(yield* DagLocation.ownsSession(sessionID, ctx.directory))) return if (wakeInFlight.has(sessionID)) { wakePending.add(sessionID) return @@ -1293,6 +1314,43 @@ const serviceLayer = Layer.effect( Effect.forkScoped({ startImmediately: true }), ) + // R5 session-deletion teardown: when the parent session is removed, + // Session.remove publishes SessionV1.Event.Deleted and the FK cascade + // wipes the workflow + node rows. The in-memory runtime entry must go + // with them — otherwise a later stimulus (e.g. a workflow-terminal + // event on the deleted dagID) would still find the entry in + // `runtimes` and interrupt live fibers / drive a workflow that no + // longer exists durably. Mirror the workflow-terminal cleanup + // pattern: evalLock-serialized fiber + watcher interrupts, then drop + // the entry. + yield* events.subscribe(SessionV1.Event.Deleted).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 THIS entry is deleted, inside its own evalLock. + for (const [dagID, entry] of runtimes) { + if (entry.parentSessionID !== sessionID) 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) + }), + ) + } + }).pipe(guarded("SessionDeleted")), + ), + 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 @@ -1332,10 +1390,15 @@ const serviceLayer = Layer.effect( ), ) for (const sessionID of pendingWakeSessions) { - // Cross-instance guard: wake redelivery is store-global. A session's - // workflows share its project (enforced at dag.create), so the wake - // snapshot's own workflow rows carry the ownership proof — only - // drain sessions whose unreported workflows belong to this project. + // Cross-instance guard via the execution-location authority: wake + // redelivery is store-global. Only drain sessions whose workflows + // belong to this instance's DIRECTORY — sibling worktrees of the + // same project share the project id and must not deliver each + // other's wakes. + if (!(yield* DagLocation.ownsSession(sessionID, ctx.directory))) continue + // The wake snapshot's own workflow rows carry a second ownership + // proof — only drain sessions whose unreported workflows belong to + // this project. const snapshot = yield* store.getWakeSnapshot(sessionID).pipe( Effect.catchCause((cause) => Effect.logWarning("DagLoop failed to read wake snapshot", { sessionID, cause }).pipe( diff --git a/packages/opencode/src/goal/loop.ts b/packages/opencode/src/goal/loop.ts index 39d1be8590..7c7ad37612 100644 --- a/packages/opencode/src/goal/loop.ts +++ b/packages/opencode/src/goal/loop.ts @@ -11,6 +11,7 @@ import { Provider } from "@/provider/provider" import { Goal } from "./goal" import { GoalJudge } from "./judge" import { GoalPrompts } from "./prompts" +import { DagLocation } from "@/dag/location" import { generateText } from "ai" import { SessionID } from "@/session/schema" import { SessionAutomationLease } from "@/session/automation-lease" @@ -127,14 +128,27 @@ const serviceLayer = Layer.effect( const scanDirectoryRef: { current: string } = { current: "" } const state = yield* InstanceState.make( - Effect.fn("GoalLoop.state")(function* () { + Effect.fn("GoalLoop.state")(function* (ctx) { yield* events.subscribe(SessionStatus.Event.Status).pipe( Stream.filter((evt) => evt.data.status.type === "idle"), // D4 (fiber lifecycle): triggerEvaluation below carries the full // discipline (active-goal pre-check, fork, fiber registration, // identity-scoped self-clean), shared verbatim with the // GOAL-FP-01-04 startup scan so both drivers use one path. - Stream.runForEach((evt) => triggerEvaluation(evt.data.sessionID).pipe(Effect.ignore)), + Stream.runForEach((evt) => + Effect.gen(function* () { + const sid = evt.data.sessionID + // DAG-LOC-01 execution-location guard: idle Status events are + // store-global. Only the instance whose DIRECTORY owns the + // session may drive its goal loop — the same authority the DAG + // wake paths use (the goal scan is already directory-scoped + // through the per-directory InstanceState; this aligns the idle + // trigger with that scoping). Sessions without workflow rows + // are owned vacuously (goal-only sessions predate stamping). + if (!(yield* DagLocation.ownsSession(sid, ctx.directory))) return + yield* triggerEvaluation(sid) + }).pipe(Effect.ignore), + ), Effect.forkScoped, ) // GOAL-FP-01-04: the startup resume scan. The durable snapshot is diff --git a/packages/opencode/test/dag/dag-adoption-step-races.test.ts b/packages/opencode/test/dag/dag-adoption-step-races.test.ts index 8fe4743b44..d2dc9ebfe5 100644 --- a/packages/opencode/test/dag/dag-adoption-step-races.test.ts +++ b/packages/opencode/test/dag/dag-adoption-step-races.test.ts @@ -169,7 +169,7 @@ function runRaceTest( id: "ses_parent" as never, project_id: "project-1" as never, slug: "parent", - directory: process.cwd() as never, + directory: process.cwd(), title: "Parent", version: "test", }).run().pipe(Effect.orDie) @@ -264,6 +264,7 @@ describe("DagLoop stepping race window", () => { id: dagID, project_id: "project-1" as never, session_id: "ses_parent" as never, + directory: process.cwd(), title: "Step race", status: "stepping", config: JSON.stringify({ name: "step-race", nodes: [nodeConfig("a"), nodeConfig("b")] }), diff --git a/packages/opencode/test/dag/dag-location-guards.test.ts b/packages/opencode/test/dag/dag-location-guards.test.ts new file mode 100644 index 0000000000..b70cbc3ce7 --- /dev/null +++ b/packages/opencode/test/dag/dag-location-guards.test.ts @@ -0,0 +1,693 @@ +// oxlint-disable typescript-eslint/no-unsafe-type-assertion -- The two-instance +// harness deliberately mirrors dag-loop-guards.test.ts: mocked service layers +// and seeded row fixtures use `as never` type shims (mock objects implement +// only the interface slice the scenario exercises). The shims are type-only; +// converting them would fork the template's shape without changing behavior. +/** + * DAG-LOC-01 round 2 — execution-location RED probes. + * + * The DAG runtime guards key ownership on the PROJECT ID only + * (`wf.projectId !== ctx.project.id` in DagLoop.recoverWorkflow, + * recoverOrphanPending, the WorkflowStarted handler, and the startup wake + * sweep; the idle-Status wake path has no ownership guard at all). Two + * instances of the SAME project in DIFFERENT directories (sibling worktrees + * of one project) therefore both pass every guard: a foreign directory can + * adopt, recover-cancel, wake, and spawn for a session it does not own. + * + * Invariant under test: the execution-location key must be the DIRECTORY. + * Only the instance whose directory owns the session/workflow may act. + * + * Probe map (round 1 scenario → probe): + * S1 adoption → R1 + * S2 running recovery → R2 (the severe one) + * S5 idle wake → R3 + * S4 startup wake sweep → R4 + * deletion teardown → R5 + * identity-migration teardown→ R6 + * static contract → R7 + * + * Harness: two-instance extension of the dag-loop-guards.test.ts runGuardTest + * template. ONE shared layer graph (store, event bus, dag service) plus ONE + * DagLoop layer whose per-directory InstanceState is created by two init + * calls under two InstanceRefs — the same structure a multi-directory server + * uses. Observables (prompt queues, cancels, interrupts) are routed by the + * AMBIENT instance directory, so each probe can tell which instance acted. + */ +import { describe, expect, it } from "bun:test" +import { DateTime, Deferred, Effect, Layer, Option, Queue } from "effect" +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 { WorkflowTable } from "@opencode-ai/core/dag/sql" +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 { SessionStatusEvent } from "@opencode-ai/schema/session-status-event" +import { Agent } from "@/agent/agent" +import { Dag, type NodeConfig } from "@/dag/dag" +import { DagLoop } from "@/dag/runtime/loop" +import { InstanceRef } from "@/effect/instance-ref" +import { EventV2Bridge } from "@/event-v2-bridge" +import { SessionPrompt } from "@/session/prompt" +import { MessageID } from "@/session/schema" +import { Session } from "@/session/session" +import { SessionStatus } from "@/session/status" +import { pollWithTimeout } from "../lib/effect" +import { eq } from "drizzle-orm" +import { existsSync, readFileSync, readdirSync } from "node:fs" +import path from "node:path" + +const PROJECT_ID = "project-1" +const DIR_A = "/wtA" +const DIR_B = "/wtB" +const SES_A = "sesA" +const SES_B = "sesB" + +interface PromptGate { + readonly title: string + readonly input: SessionPrompt.PromptInput + readonly release: Deferred.Deferred +} + +function node(overrides: Partial = {}): NodeConfig { + return { + id: "n1", + name: "Node 1", + worker_type: "build", + depends_on: [], + required: true, + prompt_template: { inline: "work" }, + ...overrides, + } +} + +function takeWithin(queue: Queue.Queue, message: string) { + return Queue.take(queue).pipe( + Effect.timeoutOption("1 second"), + Effect.flatMap(Option.match({ + onNone: () => Effect.fail(new Error(message)), + onSome: Effect.succeed, + })), + ) +} + +function reply(sessionID: string, text: string): SessionV1.WithParts { + return { + info: { + id: MessageID.ascending(), + sessionID, + role: "assistant", + time: { created: Date.now() }, + }, + parts: [{ type: "text", text }], + } as never +} + +// --------------------------------------------------------------------------- +// Two-instance harness (extension of dag-loop-guards.test.ts guardLayer / +// runGuardTest): two InstanceRefs, DISTINCT directories, SAME project id. +// --------------------------------------------------------------------------- + +interface TwoInstanceInput { + readonly directoryA: string + readonly directoryB: string + /** Ambient-directory → prompt gates; each instance's loop delivers to its own queue. */ + readonly childPrompts: Map> + /** Ambient-directory → cancel log; promptSvc.cancel routes by caller directory. */ + readonly cancels: Map + /** Records interrupts of a parked child prompt (deletion-teardown probe). */ + readonly promptInterrupts: string[] + /** Seeded per-child-session messages read by the recovery status checker. */ + readonly messagesBySession: Map + /** Injected one-shot defects for DagStore.getWorkflow (parity with the template). */ + readonly failGetWorkflow?: { remaining: 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 store = input.failGetWorkflow + ? Layer.effect( + DagStore.Service, + Effect.gen(function* () { + const real = yield* DagStore.Service + return DagStore.Service.of({ + ...real, + getWorkflow: (id) => + Effect.suspend(() => { + if (input.failGetWorkflow!.remaining > 0) { + input.failGetWorkflow!.remaining-- + return Effect.die(new Error("injected transient db failure")) + } + return real.getWorkflow(id) + }), + }) + }), + ).pipe(Layer.provide(realStore)) + : realStore + const status = SessionStatus.layer.pipe(Layer.provide(bridge)) + const projector = DagProjector.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 childTitles = new Map() + const created: string[] = [] + const session = Layer.mock(Session.Service, { + get: () => Effect.succeed({ id: "ses_parent", permission: [], agent: "build" } as never), + create: (value) => + Effect.sync(() => { + const id = `ses_child_${created.length + 1}` + created.push(id) + childTitles.set(id, (value?.title ?? id).replace(" (DAG node)", "")) + return { id } as never + }), + messages: (value) => + Effect.sync(() => { + const sessionID = (value as { sessionID?: string }).sessionID + return sessionID ? (input.messagesBySession.get(sessionID) ?? []) : [] + }), + // Mimics the real remove contract: the durable session row is deleted (the + // FK cascade wipes workflow + node rows) and SessionV1.Event.Deleted is + // published for teardown subscribers. + remove: ((sessionID: Session.Interface["remove"] extends (sessionID: infer A) => unknown ? A : never) => + Effect.gen(function* () { + const db = yield* Database.Service + const rows = yield* db.db.select().from(SessionTable) + .where(eq(SessionTable.id, sessionID as never)) + .all().pipe(Effect.orDie) + const row = rows[0] + yield* db.db.delete(SessionTable).where(eq(SessionTable.id, sessionID as never)).run().pipe(Effect.orDie) + const bridgeSvc = yield* EventV2Bridge.Service + // Same shape the real remove publishes (SessionV1.Event.Deleted with + // the session's info); the schema requires id/slug/projectID/ + // directory/title/version/time. + yield* bridgeSvc.publish(SessionV1Events.Event.Deleted, { + sessionID: sessionID as never, + info: { + id: row?.id ?? sessionID, + slug: row?.slug ?? "deleted", + projectID: row?.project_id ?? PROJECT_ID, + directory: row?.directory ?? input.directoryA, + title: row?.title ?? "Deleted session", + version: row?.version ?? "test", + time: { created: Date.now(), updated: Date.now() }, + } as never, + }).pipe(Effect.orDie) + })) as unknown as Session.Interface["remove"], + }) + const queueFor = (dir: string | undefined) => + input.childPrompts.get(dir ?? input.directoryA) ?? input.childPrompts.get(input.directoryA)! + const cancelsFor = (dir: string | undefined) => + input.cancels.get(dir ?? input.directoryA) ?? input.cancels.get(input.directoryA)! + const deliver = Effect.fn("test.SessionPrompt.deliver")(function* (value: SessionPrompt.PromptInput) { + const sessionID = value.sessionID as string + // Route the observation by the CALLING instance's directory (the ambient + // InstanceRef of the loop handler fiber), so each probe can attribute the + // delivery to instance A or B. + const dir = (yield* InstanceRef)?.directory + const release = yield* Deferred.make() + yield* Queue.offer(queueFor(dir), { + title: childTitles.get(sessionID) ?? sessionID, + input: value, + release, + }) + const text = yield* Deferred.await(release).pipe( + Effect.onInterrupt(() => + Effect.sync(() => { + input.promptInterrupts.push(sessionID) + }), + ), + ) + return reply(sessionID, text) + }) + const prompt = Layer.mock(SessionPrompt.Service, { + cancel: (sessionID) => + Effect.gen(function* () { + const dir = (yield* InstanceRef)?.directory + cancelsFor(dir).push(sessionID as string) + }), + prompt: deliver, + promptIfIdle: (value) => deliver(value).pipe(Effect.map(Option.some)), + }) + const agent = Layer.mock(Agent.Service, { + get: () => Effect.succeed({ + name: "build", + mode: "all", + permission: [], + options: {}, + description: "", + prompt: "", + model: { providerID: "test" as never, modelID: "test-model" as never }, + tools: {}, + hooks: {}, + }), + }) + const loop = DagLoop.layer.pipe( + Layer.provide(base), + Layer.provide(session), + Layer.provide(prompt), + Layer.provide(agent), + ) + // The session mock is also surfaced to the test body (Session.remove for + // the deletion-teardown probe); mergeAll memoizes by layer identity, so the + // instance the loop sees is the same one the test drives. + return Layer.mergeAll(base, loop, session) +} + +interface TwoInstanceServices { + readonly dag: Dag.Interface + readonly loop: DagLoop.Interface + readonly store: DagStore.Interface + readonly database: Database.Interface + readonly bridge: EventV2.Interface + readonly session: Session.Interface + /** Boot the loop under instance A's directory. */ + readonly initA: Effect.Effect + /** Boot the loop under instance B's directory. */ + readonly initB: Effect.Effect + readonly childPromptsA: Queue.Queue + readonly childPromptsB: Queue.Queue + readonly cancelsA: string[] + readonly cancelsB: string[] + readonly promptInterrupts: string[] + readonly messagesBySession: Map +} + +function runTwoInstanceGuardTest( + options: { + readonly projectID?: string + readonly directoryA?: string + readonly directoryB?: string + readonly sessionA?: string + readonly sessionB?: string + readonly failGetWorkflow?: { remaining: number } + }, + test: (services: TwoInstanceServices) => Effect.Effect, + beforeInit?: (services: { readonly database: Database.Interface }) => Effect.Effect, +) { + const projectID = options.projectID ?? PROJECT_ID + const directoryA = options.directoryA ?? DIR_A + const directoryB = options.directoryB ?? DIR_B + const sessionA = options.sessionA ?? SES_A + const sessionB = options.sessionB ?? SES_B + return Effect.gen(function* () { + const childPromptsA = yield* Queue.unbounded() + const childPromptsB = yield* Queue.unbounded() + const cancelsA: string[] = [] + const cancelsB: string[] = [] + const promptInterrupts: string[] = [] + const messagesBySession = new Map() + return yield* Effect.gen(function* () { + const dag = yield* Dag.Service + const loop = yield* DagLoop.Service + const store = yield* DagStore.Service + const database = yield* Database.Service + const bridge = yield* EventV2Bridge.Service + const session = yield* Session.Service + // ONE project row; TWO sessions in the SAME project but DISTINCT + // directories — sibling worktrees of one project. + yield* database.db.insert(ProjectTable).values({ + id: projectID as never, + worktree: directoryA as never, + sandboxes: [], + }).run().pipe(Effect.orDie) + for (const [id, dir, slug, title] of [ + [sessionA, directoryA, "a", "Parent A"], + [sessionB, directoryB, "b", "Parent B"], + ] as const) { + yield* database.db.insert(SessionTable).values({ + id: id as never, + project_id: projectID as never, + slug, + directory: dir as never, + title, + version: "test", + }).run().pipe(Effect.orDie) + } + if (beforeInit) yield* beforeInit({ database }) + const refB = { + directory: directoryB, + worktree: directoryB, + project: { id: projectID }, + } as never + return yield* test({ + dag, + loop, + store, + database, + bridge, + session, + // initA uses the ambient InstanceRef (directory A, provided below); + // initB shadows it with B's InstanceRef. + initA: loop.init(), + initB: loop.init().pipe(Effect.provideService(InstanceRef, refB)), + childPromptsA, + childPromptsB, + cancelsA, + cancelsB, + promptInterrupts, + messagesBySession, + }) + }).pipe( + Effect.provide(twoInstanceLayer({ + directoryA, + directoryB, + childPrompts: new Map([ + [directoryA, childPromptsA], + [directoryB, childPromptsB], + ]), + cancels: new Map([ + [directoryA, cancelsA], + [directoryB, cancelsB], + ]), + promptInterrupts, + messagesBySession, + failGetWorkflow: options.failGetWorkflow, + })), + Effect.provideService(InstanceRef, { + directory: directoryA, + worktree: directoryA, + project: { id: projectID }, + } as never), + Effect.scoped, + ) + }) +} + +/** Seed a terminal (failed) workflow owned by sesA with an unreported wake. */ +function seedTerminalWorkflow( + services: { readonly database: Database.Interface }, + wakeReported: boolean, +) { + return services.database.db.insert(WorkflowTable).values({ + id: "wake-wf", + project_id: PROJECT_ID as never, + session_id: SES_A as never, + title: "Terminal workflow for sesA", + status: "failed", + config: "{}", + seq: 5, + wake_reported: wakeReported, + }).run().pipe(Effect.orDie, Effect.as(undefined)) +} + +// --------------------------------------------------------------------------- +// Behavior probes R1–R6 +// --------------------------------------------------------------------------- + +describe("DAG execution-location guards (DAG-LOC-01)", () => { + it("R1/S1: a booted sibling instance does not adopt a workflow created for another directory's session", async () => { + await Effect.runPromise( + runTwoInstanceGuardTest( + {}, + ({ dag, store, initB, childPromptsB }) => + Effect.gen(function* () { + // Instance B is booted; instance A is not (booting A would race + // the first-wave spawn on the shared dag service and mask B's + // independent adoption defect). The workflow is created for A's + // session — stamped with A's directory /wtA. + yield* initB + const dagID = yield* dag.create({ + projectID: PROJECT_ID, + sessionID: SES_A, + title: "A's workflow", + config: { name: "r1", nodes: [node()] }, + }) + yield* Effect.sleep("400 millis") + const foreignChild = Option.getOrElse(yield* Queue.poll(childPromptsB), () => null) + expect(foreignChild).toBe(null) + const nodes = yield* store.getNodes(dagID) + expect(nodes).toHaveLength(1) + expect(nodes[0]?.status).toBe("pending") + }), + ), + ) + }) + + it("R2/S2: a sibling directory's startup recovery does not cancel the owner's live child", async () => { + await Effect.runPromise( + runTwoInstanceGuardTest( + {}, + ({ dag, store, initA, initB, childPromptsA, cancelsB, messagesBySession }) => + Effect.gen(function* () { + // A boots first and owns the workflow: it adopts through the + // WorkflowStarted handler (no reconciliation) and spawns n1. + yield* initA + const dagID = yield* dag.create({ + projectID: PROJECT_ID, + sessionID: SES_A, + title: "A's running workflow", + config: { name: "r2", nodes: [node()] }, + }) + const child = yield* takeWithin(childPromptsA, "owner did not start its node") + const childSessionID = child.input.sessionID as string + // The child's last durable message is a non-terminal assistant + // part: the session is live and executing under A's directory. + messagesBySession.set(childSessionID, [reply(childSessionID, "still working")]) + // B boots and its startup scan reconciles every running workflow. + // B must not touch a workflow owned by another directory. + yield* initB + const nodes = yield* store.getNodes(dagID) + const workflow = yield* store.getWorkflow(dagID) + expect(cancelsB).toHaveLength(0) + expect(nodes[0]?.status).toBe("running") + expect(workflow?.status).toBe("running") + }), + ), + ) + }) + + it("R3/S5: a sibling instance ignores an idle Status event for another directory's session", async () => { + await Effect.runPromise( + runTwoInstanceGuardTest( + {}, + ({ database, bridge, initB, childPromptsB }) => + Effect.gen(function* () { + yield* initB + // Re-arm the terminal workflow's wake AFTER B's startup sweep has + // passed over it, so the delivery below can only come from the + // idle-Status subscription path. + yield* database.db.update(WorkflowTable) + .set({ wake_reported: false }) + .where(eq(WorkflowTable.id, "wake-wf")) + .run().pipe(Effect.orDie) + yield* bridge.publish(SessionStatusEvent.Status, { + sessionID: SES_A as never, + status: { type: "idle" }, + }).pipe(Effect.orDie) + const delivered = yield* Queue.take(childPromptsB).pipe(Effect.timeoutOption("1 second")) + expect(Option.getOrElse(delivered, () => null)).toBe(null) + }), + (services) => seedTerminalWorkflow(services, true), + ), + ) + }) + + it("R4/S4: a sibling instance's startup sweep does not deliver another directory's session wakes", async () => { + await Effect.runPromise( + runTwoInstanceGuardTest( + {}, + ({ initB, childPromptsB }) => + Effect.gen(function* () { + yield* initB + // The unreported terminal workflow for sesA exists BEFORE B boots; + // B's startup wake sweep must leave it alone. + const delivered = yield* Queue.take(childPromptsB).pipe(Effect.timeoutOption("1 second")) + expect(Option.getOrElse(delivered, () => null)).toBe(null) + }), + (services) => seedTerminalWorkflow(services, false), + ), + ) + }) + + it("R5: Session.remove drops the in-memory entry before a later stimulus can act on it", async () => { + await Effect.runPromise( + runTwoInstanceGuardTest( + {}, + ({ dag, store, bridge, session, initA, childPromptsA, promptInterrupts }) => + Effect.gen(function* () { + yield* initA + const dagID = yield* dag.create({ + projectID: PROJECT_ID, + sessionID: SES_A, + title: "Delete me", + config: { name: "r5", nodes: [node()] }, + }) + yield* takeWithin(childPromptsA, "node did not start") + // Remove the parent session. The FK cascade wipes the workflow and + // node rows; the in-memory runtime entry must go with them. + yield* session.remove(SES_A as never) + expect(yield* store.getWorkflow(dagID)).toBeUndefined() + // The deletion teardown is event-driven (SessionV1.Event.Deleted is + // fanned out async): wait for its interrupt of the parked child to + // be recorded so `before` is sampled on a settled teardown. + yield* pollWithTimeout( + Effect.sync(() => (promptInterrupts.length > 0 ? true : undefined)), + "deletion teardown never interrupted the parked child", + "1 second", + ) + const before = promptInterrupts.length + // Workflow-terminal stimulus on the deleted workflow. The terminal + // handler is gated on runtimes.has(dagID): if the entry was dropped + // at deletion the handler never fires and the parked child prompt + // fiber is left untouched by this stimulus. + yield* bridge.publish(DagEvent.WorkflowCancelled, { + dagID: dagID as never, + timestamp: yield* DateTime.now, + }).pipe(Effect.orDie) + // Negative window: give the stimulus handler time to (wrongly) act, + // then assert it added no further interrupts. (pollWithTimeout is a + // positive-wait tool — its timeout errors the effect rather than + // returning a fallback, so a "nothing must happen" window is + // asserted with sleep + snapshot instead.) + yield* Effect.sleep("300 millis") + expect(promptInterrupts.slice(before)).toEqual([]) + }), + ), + ) + }) + + it("R6: an identity migration invalidates the in-memory entry", async () => { + await Effect.runPromise( + runTwoInstanceGuardTest( + {}, + ({ dag, store, database, initA, childPromptsA }) => + Effect.gen(function* () { + yield* initA + const dagID = yield* dag.create({ + projectID: PROJECT_ID, + sessionID: SES_A, + title: "Migrate me", + config: { name: "r6", nodes: [node()] }, + }) + yield* takeWithin(childPromptsA, "node did not start") + // Identity migration: repaint the workflow's project id (old → + // new). The in-memory entry must not keep driving the migrated + // workflow. + yield* database.db.update(WorkflowTable) + .set({ project_id: "project-new" as never }) + .where(eq(WorkflowTable.id, dagID as never)) + .run().pipe(Effect.orDie) + // Node-completion stimulus: the stale entry must not publish a + // workflow transition (here: running → completed) for a workflow + // whose durable identity moved away. + yield* dag.nodeCompleted(dagID, "n1", { ok: true }) + // The node itself completes (the durable event projects), which + // proves the stimulus was delivered to the runtime. + expect((yield* store.getNode(dagID, "n1"))?.status).toBe("completed") + // Negative window: give the completion path time to (wrongly) + // publish a workflow transition, then assert the workflow is still + // running. (pollWithTimeout is a positive-wait tool — its timeout + // errors the effect rather than returning a fallback, so the + // negative assertion is a sleep + snapshot.) + yield* Effect.sleep("300 millis") + expect((yield* store.getWorkflow(dagID))?.status).toBe("running") + }), + ({ database }) => + database.db.insert(ProjectTable).values({ + id: "project-new" as never, + worktree: process.cwd() as never, + sandboxes: [], + }).run().pipe(Effect.orDie, Effect.as(undefined)), + ), + ) + }) +}) + +// --------------------------------------------------------------------------- +// R7 — static contract +// --------------------------------------------------------------------------- + +describe("DAG execution-location static contract (DAG-LOC-01 R7)", () => { + const opencodeDagSrc = path.resolve(import.meta.dir, "../../src/dag") + const coreDagSrc = path.resolve(import.meta.dir, "../../../../packages/core/src/dag") + + function readDagSources(root: string): Array<{ file: string; source: string }> { + const out: Array<{ file: string; source: string }> = [] + if (!existsSync(root)) throw new Error(`dag source root missing: ${root}`) + for (const entry of readdirSync(root, { recursive: true })) { + const full = path.join(root, String(entry)) + if (!full.endsWith(".ts")) continue + out.push({ file: full, source: readFileSync(full, "utf8") }) + } + return out + } + + it("keys every adoption/wake guard on the directory and never on the session directory column", () => { + const sources = [...readDagSources(opencodeDagSrc), ...readDagSources(coreDagSrc)] + expect(sources.length).toBeGreaterThan(20) + + // Negative half: the dag sources must not read the session's directory + // column (session.directory / SessionTable.directory) — the execution- + // location key belongs on the workflow row itself, stamped at create. + const sessionDirRefs = sources + .filter(({ source }) => /session\.directory|SessionTable\.directory/.test(source)) + .map(({ file }) => file) + expect(sessionDirRefs).toEqual([]) + + const loopFile = sources.find((s) => s.file.endsWith("opencode/src/dag/runtime/loop.ts")) + expect(loopFile).toBeDefined() + const source = loopFile!.source + + // Positive half: every adoption/wake ownership gate must carry a + // directory-level check. The four adoption sites and the wake-delivery + // path are located by their semantic anchors so the probe survives + // refactors that keep the handler boundaries. + const regions: Array<{ name: string; from: string; to: string }> = [ + { + name: "recoverWorkflow (startup recovery adoption)", + from: "const recoverWorkflow = Effect.fn(", + to: "// Orphan-pending recovery", + }, + { + name: "recoverOrphanPending (orphan-pending sweep)", + from: "const recoverOrphanPending = Effect.fn(", + to: "yield* events.subscribe(DagEvent.WorkflowStarted)", + }, + { + name: "WorkflowStarted handler (first-wave adoption)", + from: "yield* events.subscribe(DagEvent.WorkflowStarted)", + to: "for (const def of [DagEvent.NodeCompleted, DagEvent.NodeSkipped])", + }, + { + name: "startup wake sweep", + from: "const pendingWakeSessions =", + to: "return {}", + }, + { + name: "tryDeliverWake (idle-Status wake delivery path)", + from: 'tryDeliverWake = Effect.fn("DagLoop.tryDeliverWake")', + to: "// Idle-event subscription", + }, + ] + const unguarded: string[] = [] + for (const region of regions) { + const start = source.indexOf(region.from) + const end = source.indexOf(region.to, start) + if (start === -1 || end === -1) { + unguarded.push(`${region.name}: anchor not found (from="${region.from}" to="${region.to}")`) + continue + } + // Scan CODE lines only — the guard must be an executable directory + // comparison, not a mention in an adjacent comment. + const codeLines = source + .slice(start, end) + .split("\n") + .filter((line) => !line.trim().startsWith("//")) + if (!codeLines.some((line) => /directory/.test(line))) { + unguarded.push(`${region.name}: no directory-level ownership check in handler body`) + } + } + expect(unguarded).toEqual([]) + }) +}) diff --git a/packages/opencode/test/dag/dag-orphan-pending-recovery.test.ts b/packages/opencode/test/dag/dag-orphan-pending-recovery.test.ts index 123ba50265..44cfaf4152 100644 --- a/packages/opencode/test/dag/dag-orphan-pending-recovery.test.ts +++ b/packages/opencode/test/dag/dag-orphan-pending-recovery.test.ts @@ -123,6 +123,7 @@ function publishInterruptedCreate( config: JSON.stringify({ name: "orphan", nodes: [] }), status: "pending", timestamp: ts, + directory: process.cwd(), }) for (let i = 1; i <= nodeCount; i++) { yield* events.publish(DagEvent.NodeRegistered, { diff --git a/packages/opencode/test/dag/dag-summary-publisher-behavior.test.ts b/packages/opencode/test/dag/dag-summary-publisher-behavior.test.ts index e585761945..83e70442b7 100644 --- a/packages/opencode/test/dag/dag-summary-publisher-behavior.test.ts +++ b/packages/opencode/test/dag/dag-summary-publisher-behavior.test.ts @@ -51,6 +51,7 @@ function workflow(id: string, sessionId: string, projectId: string): WorkflowRow id, projectId, sessionId, + directory: null, title: id, status: "running", config: "", diff --git a/packages/opencode/test/dag/dag-wake-integration.test.ts b/packages/opencode/test/dag/dag-wake-integration.test.ts index 47e916fa46..451d9574e9 100644 --- a/packages/opencode/test/dag/dag-wake-integration.test.ts +++ b/packages/opencode/test/dag/dag-wake-integration.test.ts @@ -207,7 +207,7 @@ function runWakeTest( id: "ses_parent" as never, project_id: "project-1" as never, slug: "parent", - directory: process.cwd() as never, + directory: process.cwd(), title: "Parent", version: "test", }).run().pipe(Effect.orDie) @@ -1095,6 +1095,7 @@ describe("DagLoop atomic wake integration", () => { id: "recovered-workflow", project_id: "project-1" as never, session_id: "ses_parent" as never, + directory: process.cwd(), title: "Recovered workflow", status: "completed", config: "{}", @@ -1193,6 +1194,7 @@ describe("DagLoop atomic wake integration", () => { id: "dag_recovered_conditional", project_id: "project-1" as never, session_id: "ses_parent" as never, + directory: process.cwd(), title: "Recovered conditional workflow", status: "running", config: JSON.stringify({ @@ -1336,6 +1338,7 @@ describe("DagLoop atomic wake integration", () => { id: "dag_recovered_review_rejection", project_id: "project-1" as never, session_id: "ses_parent" as never, + directory: process.cwd(), title: "Recovered review rejection", status: "running", config: JSON.stringify({ diff --git a/packages/opencode/test/dag/workflow-tool.test.ts b/packages/opencode/test/dag/workflow-tool.test.ts index 3277ac61b3..f9a7ce84d8 100644 --- a/packages/opencode/test/dag/workflow-tool.test.ts +++ b/packages/opencode/test/dag/workflow-tool.test.ts @@ -118,6 +118,7 @@ const store = Layer.mock(DagStore.Service, { projectId: projectID, sessionId: "ses_workflow_parent", title: "Status workflow", + directory: null, status: "running", config: "{}", seq: 1, @@ -133,6 +134,7 @@ const store = Layer.mock(DagStore.Service, { projectId: projectID, sessionId: "ses_workflow_parent", title: "Result workflow", + directory: null, status: "completed", config: "{}", seq: 1, @@ -148,6 +150,7 @@ const store = Layer.mock(DagStore.Service, { projectId: projectID, sessionId: "ses_workflow_parent", title: "Control workflow", + directory: null, status: id === "dag_paused" ? "paused" : "running", config: "{}", seq: 1, @@ -163,6 +166,7 @@ const store = Layer.mock(DagStore.Service, { projectId: projectID, sessionId: "ses_workflow_parent", title: "Deep status workflow", + directory: null, status: "running", config: JSON.stringify({ name: "deep-status", @@ -186,6 +190,7 @@ const store = Layer.mock(DagStore.Service, { projectId: projectID, sessionId: "ses_workflow_parent", title: "Configured defaults", + directory: null, status: "running", config: JSON.stringify({ name: "configured-defaults", diff --git a/packages/schema/src/dag-event.ts b/packages/schema/src/dag-event.ts index d21987c6f5..dafd8d6c46 100644 --- a/packages/schema/src/dag-event.ts +++ b/packages/schema/src/dag-event.ts @@ -102,6 +102,11 @@ export const WorkflowCreated = Event.define({ title: Schema.String, config: Schema.String, // YAML string (validated separately by the runtime) status: WorkflowStatus, + // Execution-location key (DAG-LOC-01): the creating instance's directory, + // stamped at create. Optional so legacy durable events and manual + // publishers still decode; absent directories project to NULL and match + // no instance (a foreign row, never adopted). + directory: Schema.optional(Schema.String), }, }) export type WorkflowCreated = typeof WorkflowCreated.Type From 813c6330db47d4e8064cdcfc3a6e07b45eade540 Mon Sep 17 00:00:00 2001 From: Lex Date: Thu, 13 Aug 2026 13:42:30 +0800 Subject: [PATCH 2/8] =?UTF-8?q?fix(dag):=20close=20review=20findings=20?= =?UTF-8?q?=E2=80=94=20goal-side=20ownership,=20spawnReady=20revalidation,?= =?UTF-8?q?=20session-sourced=20stamp=20(DAG-LOC-01=20follow-up)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two-lens review follow-up on 3498dd670. All six introduced P2s closed; each is pinned by a probe (or an argument where the defect is structurally unobservable). P2-A (goal vacuous-true): the dag-side ownsSession is keyed on workflow rows and is vacuously true for goal-only sessions, so instance B could drive instance A's goal-only sessions (cross-directory continuation, double judge, spurious pauses). Fix: Goal.ownsSession (goal.ts) — a REAL directory check against the durable session row (SessionTable.directory; legal there — the goal module is outside the dag trees R7 scans). Vacuous-own remains only where no durable answer exists (rowless synthetic sessions, or a runtime graph without Database). The session-row read lives in a new session-domain accessor (packages/opencode/src/session/location.ts, sessionDirectory) so the R7 constraint (no session.directory reads in dag sources) holds and Dag.create (P2-F) shares the same single source. Applied in the GoalLoop idle handler (goal/loop.ts). P2-B (goal idle subscription killable): the new guard was the first defect-capable durable read in the goal idle handler; a store defect would have permanently killed the runForEach subscription (Effect.ignore does not absorb defects). Fix: the handler body is wrapped in catchCause with a logged warning — a store defect degrades to a skipped evaluation, never a dead loop (mirrors DagLoop's guarded()). P2-C (R6 gates only checkCompletion): after an identity repaint the stale entry could still win nodeQueued and materialize a child under the stale directory, and the deadline watcher could still write escalations. Fix: spawnReady revalidates ownership at its entry (all seven spawn call sites funnel through it) and drops the stale entry when ownership no longer holds; makeDeadlineWatcher revalidates before its write section (escalate + cap enforcement) and ends its mandate on ownership loss — the check only runs when it can disprove ownership (instance context and Database present), so supervision still never ends in graphs without them (R13). P2-D (NULL zombie — silent): workflows created by old builds after the one-shot backfill keep directory=NULL and are skipped silently forever. Fix: DagLocation logs a WARN, deduped per workflow per process, whenever an adoption/recovery/wake path skips a NULL-directory row. The conservative never-match policy is unchanged. P2-E (Deleted sweep race): recoverWorkflow could pass the ownership guard, the session deletion could cascade the rows and run the Deleted sweep before the entry was published, then runtimes.set leaked an inert entry forever. Fix: ownership is re-checked after the recovery body and before runtimes.set (same for the WorkflowStarted first-wave path, whose getNodes yield opens the same window); the ensuring still clears the recovering reservation. No probe: the leak is behaviorally inert (every post-deletion stimulus is filtered by runtimes.has or no-ops against the missing row), so the race is not observably constructible — the re-check closes it structurally. P2-F (create boundary): Dag.create stamped the ambient REQUEST instance's directory, so a request on directory A could create a workflow for B's session stamped A, orphaning it from B's loops. Fix: the stamp now comes from the TARGET session's durable directory (sessionDirectory), falling back to the ambient instance only when the session has no durable row. The API validation tightening was left out of this slice (HTTP handler territory); the stamp fix is the required part. Probes (test/dag/dag-location-guards.test.ts, "DAG-LOC-01 P2 follow-ups"): 12/12 green (7 original + P2-A, P2-F, P2-C, P2-B, P2-D). P2-B injects a one-shot synchronous store defect through a Database proxy and proves the NEXT idle event is still evaluated; P2-D captures the warning via Effect.withLogger and asserts exactly one emission for two checks. Mutations (all restored): - bypass the goal-side guard -> P2-A red (P2-B also red: its defect lands on the guard's read), 10 pass - revert the create stamp to the ambient instance -> P2-F red, 11 pass - bypass the spawnReady revalidation -> P2-C red, 11 pass Verification: packages/opencode test/dag + test/goal = 565 pass / 0 fail; test:dag-core pass; test:httpapi 227 pass / 0 fail; bun typecheck clean (root); bun lint 4850 / 0 errors (ratchet unchanged). Co-Authored-By: Claude --- packages/opencode/src/dag/dag.ts | 20 +- packages/opencode/src/dag/location.ts | 43 ++- packages/opencode/src/dag/runtime/loop.ts | 22 ++ packages/opencode/src/dag/runtime/spawn.ts | 18 + packages/opencode/src/goal/goal.ts | 23 ++ packages/opencode/src/goal/loop.ts | 27 +- packages/opencode/src/session/location.ts | 46 +++ .../test/dag/dag-location-guards.test.ts | 323 +++++++++++++++++- 8 files changed, 500 insertions(+), 22 deletions(-) create mode 100644 packages/opencode/src/session/location.ts diff --git a/packages/opencode/src/dag/dag.ts b/packages/opencode/src/dag/dag.ts index f818e83b05..e5f97c6e02 100644 --- a/packages/opencode/src/dag/dag.ts +++ b/packages/opencode/src/dag/dag.ts @@ -33,6 +33,8 @@ import { import { unresolvedReviewOutcomes } from "./review-lifecycle" import { DagValidation, StructuralValidationError } from "./validation" import { DagLocation } from "./location" +import { SessionLocation } from "@/session/location" +import { SessionID } from "@/session/schema" export { StructuralValidationError } from "./validation" @@ -401,12 +403,18 @@ export const layer = Layer.effect( config: JSON.stringify(durableConfig), status: "pending", timestamp: ts, - // DAG-LOC-01: stamp the execution-location key (the creating - // instance's canonical directory) on the workflow row at create. - // The DagLoop ownership guards decide on this stamp; sibling - // worktrees of the same project carry distinct directories and are - // mutually foreign. - directory: yield* DagLocation.stampDirectory(), + // DAG-LOC-01 stamp (P2-F): the execution-location key is the TARGET + // SESSION's durable directory — the single source of truth. Stamping + // the ambient request instance's directory would let a request on + // directory A create a workflow for B's session stamped A, orphaning + // it from B's loops. Fall back to the ambient instance only when the + // session has no durable row (the workflow insert would fail its + // session FK anyway). + directory: yield* Effect.flatMap(SessionLocation.sessionDirectory(SessionID.make(input.sessionID)), (durable) => + durable._tag === "Some" + ? Effect.succeed(DagLocation.canonicalDirectory(durable.value)) + : DagLocation.stampDirectory(), + ), }) for (const node of durableConfig.nodes) { yield* events.publish(DagEvent.NodeRegistered, { diff --git a/packages/opencode/src/dag/location.ts b/packages/opencode/src/dag/location.ts index 0f037eb051..29b6c3f56b 100644 --- a/packages/opencode/src/dag/location.ts +++ b/packages/opencode/src/dag/location.ts @@ -48,6 +48,25 @@ export const canonicalDirectory = (directory: string): string => { export const stampDirectory = (): Effect.Effect => Effect.map(InstanceRef, (instance) => (instance ? canonicalDirectory(instance.directory) : "")) +/** + * P2-D: a workflow whose directory stamp is NULL (created by a pre-DAG-LOC-01 + * build after the one-shot backfill) matches no instance and is silently + * skipped by every adoption/recovery/wake path forever. Log that skip once + * per workflow per process so the zombie is visible; the conservative + * never-match policy stays. + */ +const nullDirectoryWarned = new Set() + +const warnNullDirectory = (row: { id: string; directory: string | null }): Effect.Effect => + Effect.suspend(() => { + if (row.directory !== null || nullDirectoryWarned.has(row.id)) return Effect.void + nullDirectoryWarned.add(row.id) + return Effect.logWarning( + "DagLocation skipping workflow with a NULL execution-location directory (created before the DAG-LOC-01 stamp and never backfilled) — it will never be adopted, recovered, or woken; recreate the workflow to re-enable it", + { dagID: row.id }, + ) + }) + /** * Owns the workflow iff its DURABLE row (re-read on every check) still belongs * to the ambient instance: the project id matches (fast-reject + R6 identity @@ -70,7 +89,11 @@ export const ownsWorkflow = (workflowID: string, directory: string): Effect.Effe .pipe(Effect.orDie) if (!row) return false if (row.project_id !== instance.project.id) return false - return row.directory !== null && canonicalDirectory(row.directory) === canonicalDirectory(directory) + if (row.directory === null) { + yield* warnNullDirectory(row) + return false + } + return canonicalDirectory(row.directory) === canonicalDirectory(directory) }) /** @@ -96,10 +119,16 @@ export const ownsSession = (sessionID: string, directory: string): Effect.Effect .where(eq(WorkflowTable.session_id, sessionID)) .all() .pipe(Effect.orDie) - return rows.every( - (row) => - row.project_id === instance.project.id && - row.directory !== null && - canonicalDirectory(row.directory) === canonicalDirectory(directory), - ) + let owned = true + for (const row of rows) { + if (row.project_id !== instance.project.id) { + owned = false + } else if (row.directory === null) { + yield* warnNullDirectory(row) + owned = false + } else if (canonicalDirectory(row.directory) !== canonicalDirectory(directory)) { + owned = false + } + } + return owned }) diff --git a/packages/opencode/src/dag/runtime/loop.ts b/packages/opencode/src/dag/runtime/loop.ts index a7509a0f85..3ec6bbeddb 100644 --- a/packages/opencode/src/dag/runtime/loop.ts +++ b/packages/opencode/src/dag/runtime/loop.ts @@ -94,6 +94,16 @@ const serviceLayer = Layer.effect( const spawnReady = Effect.fn("DagLoop.spawnReady")(function* (dagID: string) { const entry = runtimes.get(dagID) if (!entry) return + // P2-C execution-location revalidation: every spawn call site funnels + // through here. A workflow whose durable identity was repainted + // (identity migration) or cascade-deleted must not keep scheduling + // children under this instance's directory context — drop the stale + // entry so no later stimulus acts on it either (its watchers + // self-exit on terminal rows; its prompt fibers finish naturally). + if (!(yield* DagLocation.ownsWorkflow(dagID, ctx.directory))) { + runtimes.delete(dagID) + return + } // D13: settle cascade-skips before spawning. A node whose dependencies // are all skipped can never receive a real input; publish a durable // NodeSkipped(orphan_cascade) wave by wave until a fixpoint so gated @@ -417,6 +427,12 @@ const serviceLayer = Layer.effect( if (isPaused) runtime.setPaused(true) if (isStepping) runtime.setStepMode(true) const entry: WorkflowEntry = { runtime, semaphore, evalLock: Semaphore.makeUnsafe(1), parentSessionID: wf.sessionId, config, fibers: new Map(), watchers: new Map() } + // P2-E deletion-race re-check: the SessionV1.Event.Deleted sweep + // only removes entries already published into `runtimes`. If the + // FK cascade deleted this workflow's row while reconciliation ran, + // 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 runtimes.set(dagID, entry) yield* automation.register(SessionID.make(wf.sessionId), { kind: "dag", id: dagID }) // Reconciliation settles every persisted running attempt before the @@ -512,6 +528,12 @@ const serviceLayer = Layer.effect( const runtime = new WorkflowRuntime(toSchedulingNodes(nodes), maxConcurrency) const semaphore = Semaphore.makeUnsafe(maxConcurrency) const entry: WorkflowEntry = { runtime, semaphore, evalLock: Semaphore.makeUnsafe(1), parentSessionID: wf.sessionId, config, fibers: new Map(), watchers: new Map() } + // P2-E deletion-race re-check (same window as recoverWorkflow): + // the Deleted sweep only removes entries already in `runtimes`, + // and getNodes above is an awaited yield a deletion can slip + // 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 runtimes.set(dagID, entry) yield* automation.register(SessionID.make(wf.sessionId), { kind: "dag", id: dagID }) yield* entry.evalLock.withPermits(1)( diff --git a/packages/opencode/src/dag/runtime/spawn.ts b/packages/opencode/src/dag/runtime/spawn.ts index 5926e6011f..8dec5eed93 100644 --- a/packages/opencode/src/dag/runtime/spawn.ts +++ b/packages/opencode/src/dag/runtime/spawn.ts @@ -31,6 +31,7 @@ */ import { Effect, Semaphore, Scope, Fiber, Option, Clock, Cause, Exit } from "effect" +import { Database } from "@opencode-ai/core/database/database" import { Agent } from "@/agent/agent" import { Session } from "@/session/session" import { SessionID, MessageID } from "@/session/schema" @@ -38,6 +39,8 @@ import { deriveSubagentSessionPermission } from "@/agent/subagent-permissions" import { SessionPrompt } from "@/session/prompt" import { Dag } from "../dag" import { DagModel } from "../model" +import { DagLocation } from "../location" +import { InstanceRef } from "@/effect/instance-ref" import { isTransitionRejection, isNodeTerminalStatus } from "@opencode-ai/core/dag/core/types" import type { DagStore } from "@opencode-ai/core/dag/store" import { ModelV2 } from "@opencode-ai/core/model" @@ -137,6 +140,21 @@ export function makeDeadlineWatcher( continue } if (isNodeTerminalStatus(node.status as never)) return + // DAG-LOC-01 (P2-C): revalidate ownership before the write section — + // escalation and cap-enforcement writes must not land on a workflow + // whose durable identity was repainted (identity migration) or whose + // rows were cascade-deleted. Losing ownership ends this watcher's + // mandate; the instance that owns the repainted workflow supervises it. + // The check only runs when it can DISPROVE ownership (instance context + // and Database both present — always true for watchers forked from + // DagLoop): absent either, supervision must not end (R13). + { + const instance = yield* InstanceRef + const db = yield* Effect.serviceOption(Database.Service) + if (instance && db._tag === "Some") { + if (!(yield* DagLocation.ownsWorkflow(input.dagID, instance.directory))) return + } + } const now = yield* Clock.currentTimeMillis const deadline = node.deadlineMs if (node.status !== "running") { diff --git a/packages/opencode/src/goal/goal.ts b/packages/opencode/src/goal/goal.ts index c7fcb0af81..29d56c7707 100644 --- a/packages/opencode/src/goal/goal.ts +++ b/packages/opencode/src/goal/goal.ts @@ -13,6 +13,29 @@ import { GoalPrompts } from "./prompts" import { SessionID } from "@/session/schema" import { SessionStatus } from "@/session/status" import { SessionAutomationLease } from "@/session/automation-lease" +import { SessionLocation } from "@/session/location" +import { DagLocation } from "@/dag/location" + +/** + * DAG-LOC-01 (P2-A) — goal-side execution-location ownership. + * + * The DAG-side authority keys ownership on the workflow row, which is + * vacuously true for goal-only sessions (no workflow rows). The goal loop + * therefore needs a REAL directory check: the durable session row's + * directory (SessionTable — legal here, the goal module is outside the dag + * trees) must match the calling instance's directory, same canonicalization + * as the workflow-keyed authority. Vacuous-own remains ONLY where no + * durable answer exists: a missing session row (synthetic test sessions / + * already-deleted sessions whose goal state is gone with them) or a runtime + * graph without the Database service (goal e2e fixtures; production always + * carries it). + */ +export const ownsSession = (sessionID: SessionID, directory: string): Effect.Effect => + Effect.gen(function* () { + const durable = yield* SessionLocation.sessionDirectory(sessionID) + if (durable._tag === "None") return true + return DagLocation.canonicalDirectory(durable.value) === DagLocation.canonicalDirectory(directory) + }) export type RemoveSubgoalResult = | { tag: "ok"; removed: string; state: GoalState.Info } diff --git a/packages/opencode/src/goal/loop.ts b/packages/opencode/src/goal/loop.ts index 7c7ad37612..bb72356df5 100644 --- a/packages/opencode/src/goal/loop.ts +++ b/packages/opencode/src/goal/loop.ts @@ -11,7 +11,6 @@ import { Provider } from "@/provider/provider" import { Goal } from "./goal" import { GoalJudge } from "./judge" import { GoalPrompts } from "./prompts" -import { DagLocation } from "@/dag/location" import { generateText } from "ai" import { SessionID } from "@/session/schema" import { SessionAutomationLease } from "@/session/automation-lease" @@ -140,14 +139,26 @@ const serviceLayer = Layer.effect( const sid = evt.data.sessionID // DAG-LOC-01 execution-location guard: idle Status events are // store-global. Only the instance whose DIRECTORY owns the - // session may drive its goal loop — the same authority the DAG - // wake paths use (the goal scan is already directory-scoped - // through the per-directory InstanceState; this aligns the idle - // trigger with that scoping). Sessions without workflow rows - // are owned vacuously (goal-only sessions predate stamping). - if (!(yield* DagLocation.ownsSession(sid, ctx.directory))) return + // session may drive its goal loop — the real session-row check + // (Goal.ownsSession reads SessionTable.directory; the workflow- + // keyed DAG authority is vacuous for goal-only sessions, P2-A). + // Sessions without a durable row are owned vacuously (synthetic + // test sessions; a deleted session's goal state is gone with it). + if (!(yield* Goal.ownsSession(sid, ctx.directory))) return yield* triggerEvaluation(sid) - }).pipe(Effect.ignore), + // P2-B subscription survival: this handler now contains the + // first defect-capable durable reads in the goal idle path + // (Goal.ownsSession / goal.load both orDie). Effect.ignore does + // NOT absorb defects — a transient store failure would + // permanently kill the runForEach subscription and the loop + // would never evaluate another idle event. catchCause absorbs + // failures AND defects at the boundary, so a store defect + // degrades to a logged, skipped evaluation — never a dead loop. + }).pipe( + Effect.catchCause((cause) => + Effect.logWarning("GoalLoop idle handler failed", { sessionID: evt.data.sessionID, cause }), + ), + ), ), Effect.forkScoped, ) diff --git a/packages/opencode/src/session/location.ts b/packages/opencode/src/session/location.ts new file mode 100644 index 0000000000..c486f01d60 --- /dev/null +++ b/packages/opencode/src/session/location.ts @@ -0,0 +1,46 @@ +export * as SessionLocation from "./location" + +/** + * DAG-LOC-01 (P2-A / P2-F) — the durable session-directory accessor. + * + * The execution-location authority in @/dag/location keys ownership on the + * WORKFLOW row (WorkflowTable.directory, R7: dag sources must not read the + * session directory column). Two consumers need the SESSION's own durable + * directory, and both live OUTSIDE the dag trees, where the session-table + * read is legal: + * + * - the GoalLoop idle trigger: goal-only sessions have no workflow rows, so + * the workflow-keyed ownsSession is vacuously true for them — the goal + * side needs a REAL directory check against SessionTable.directory + * (Goal.ownsSession, built on this accessor); + * - Dag.create (P2-F): the workflow stamp must come from the TARGET + * session's durable directory (the single source of truth), not the + * ambient request instance's — otherwise a request on directory A can + * create a workflow for B's session and stamp it A, orphaning it from + * B's loops. + * + * Database is resolved lazily via serviceOption so callers' static layer + * requirements stay unchanged (the optional-cross-dependency pattern); + * production graphs always carry it. None means "no durable answer" — the + * caller decides the fallback (vacuous-own on the goal side, ambient + * instance on the create side). + */ + +import { eq } from "drizzle-orm" +import { Effect, Option } from "effect" +import { Database } from "@opencode-ai/core/database/database" +import { SessionTable } from "@opencode-ai/core/session/sql" +import { SessionID } from "@/session/schema" + +export const sessionDirectory = (sessionID: SessionID): Effect.Effect> => + Effect.gen(function* () { + const db = yield* Effect.serviceOption(Database.Service) + if (db._tag === "None") return Option.none() + const row = yield* db.value.db + .select() + .from(SessionTable) + .where(eq(SessionTable.id, sessionID)) + .get() + .pipe(Effect.orDie) + return row ? Option.some(row.directory) : Option.none() + }) diff --git a/packages/opencode/test/dag/dag-location-guards.test.ts b/packages/opencode/test/dag/dag-location-guards.test.ts index b70cbc3ce7..eb194e77b2 100644 --- a/packages/opencode/test/dag/dag-location-guards.test.ts +++ b/packages/opencode/test/dag/dag-location-guards.test.ts @@ -34,7 +34,7 @@ * AMBIENT instance directory, so each probe can tell which instance acted. */ import { describe, expect, it } from "bun:test" -import { DateTime, Deferred, Effect, Layer, Option, Queue } from "effect" +import { DateTime, Deferred, Effect, Layer, Option, Queue, Logger } from "effect" 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" @@ -49,6 +49,10 @@ import { SessionStatusEvent } from "@opencode-ai/schema/session-status-event" import { Agent } from "@/agent/agent" import { Dag, type NodeConfig } from "@/dag/dag" import { DagLoop } from "@/dag/runtime/loop" +import { DagLocation } from "@/dag/location" +import { Goal } from "@/goal/goal" +import { GoalLoop, GoalLoopJudgeLLM } from "@/goal/loop" +import { Provider } from "@/provider/provider" import { InstanceRef } from "@/effect/instance-ref" import { EventV2Bridge } from "@/event-v2-bridge" import { SessionPrompt } from "@/session/prompt" @@ -691,3 +695,320 @@ describe("DAG execution-location static contract (DAG-LOC-01 R7)", () => { expect(unguarded).toEqual([]) }) }) + +// --------------------------------------------------------------------------- +// P2 follow-up probes (review findings against the round-3 slice) +// --------------------------------------------------------------------------- + +/** + * Goal-side harness: real Database + real Goal + real EventV2Bridge + + * real SessionStatus with the GoalLoop layer on top, Session/Prompt/Provider + * mocked. The judge LLM is injected (GoalLoopJudgeLLM) and routed by the + * AMBIENT instance directory so a probe can attribute each judge call to + * instance A or B — the same routing trick the dag two-instance harness uses. + */ +function goalLoopLayer(input: { + readonly judgeCalls: Map + readonly messagesBySession: Map + /** Ambient Database output for the merged graph; defaults to the shared one. */ + readonly ambientDatabase?: Layer.Layer +}) { + const database = Database.layerFromPath(":memory:") + const events = EventV2.layer.pipe(Layer.provide(database)) + const bridge = EventV2Bridge.layer.pipe(Layer.provide(events)) + const status = SessionStatus.layer.pipe(Layer.provide(bridge)) + const goal = Goal.layer.pipe( + Layer.provide(bridge), + Layer.provide(database), + Layer.provide(status), + ) + const judge = Layer.succeed( + GoalLoopJudgeLLM, + GoalLoopJudgeLLM.of({ + call: () => + Effect.gen(function* () { + const dir = (yield* InstanceRef)?.directory ?? "" + const list = input.judgeCalls.get(dir) ?? [] + list.push(1) + input.judgeCalls.set(dir, list) + return JSON.stringify({ done: false, reason: "continue" }) + }), + }), + ) + const session = Layer.mock(Session.Service, { + messages: (value) => + Effect.sync(() => input.messagesBySession.get((value as { sessionID?: string }).sessionID ?? "") ?? []), + }) + const prompt = Layer.mock(SessionPrompt.Service, { + prompt: () => Effect.succeed(reply("goal-parent", "continuation dispatched")), + }) + const provider = Layer.mock(Provider.Service, {}) + const loop = GoalLoop.layer.pipe( + Layer.provide(session), + Layer.provide(prompt), + Layer.provide(provider), + Layer.provide(judge), + Layer.provide(goal), + Layer.provide(status), + Layer.provide(bridge), + ) + return Layer.mergeAll(input.ambientDatabase ?? database, bridge, goal, loop) +} + +describe("DAG-LOC-01 P2 follow-ups", () => { + it("P2-A: a sibling instance does not drive another directory's goal-only session", async () => { + const judgeCalls = new Map() + await Effect.runPromise( + Effect.gen(function* () { + const database = yield* Database.Service + const goal = yield* Goal.Service + const loop = yield* GoalLoop.Service + const bridge = yield* EventV2Bridge.Service + yield* database.db.insert(ProjectTable).values({ + id: PROJECT_ID as never, + worktree: DIR_A as never, + sandboxes: [], + }).run().pipe(Effect.orDie) + for (const [id, dir, slug] of [ + [SES_A, DIR_A, "a"], + [SES_B, DIR_B, "b"], + ] as const) { + yield* database.db.insert(SessionTable).values({ + id: id as never, + project_id: PROJECT_ID as never, + slug, + directory: dir as never, + title: id, + version: "test", + }).run().pipe(Effect.orDie) + } + // A goal-only session: no workflow rows, so the workflow-keyed DAG + // authority is vacuous — only the goal-side session-row check can + // tell the instances apart (P2-A). + yield* goal.set(SES_A as never, "ship the feature", 10) + // Boot BOTH instances (A ambient, B via refB). + yield* loop.init() + yield* loop.init().pipe(Effect.provideService(InstanceRef, { + directory: DIR_B, + worktree: DIR_B, + project: { id: PROJECT_ID }, + } as never)) + yield* Effect.yieldNow + yield* bridge.publish(SessionStatusEvent.Status, { + sessionID: SES_A as never, + status: { type: "idle" }, + }).pipe(Effect.orDie) + // A owns /wtA and drives the goal; B must never judge it. + yield* pollWithTimeout( + Effect.sync(() => ((judgeCalls.get(DIR_A)?.length ?? 0) > 0 ? true : undefined)), + "owner instance did not drive the goal-only session", + "2 seconds", + ) + yield* Effect.sleep("300 millis") + expect(judgeCalls.get(DIR_B) ?? []).toEqual([]) + }).pipe( + Effect.provide(goalLoopLayer({ + judgeCalls, + messagesBySession: new Map([[SES_A, [reply(SES_A, "making progress")]]]), + })), + Effect.provideService(InstanceRef, { + directory: DIR_A, + worktree: DIR_A, + project: { id: PROJECT_ID }, + } as never), + Effect.scoped, + ), + ) + }) + + it("P2-F: creating a workflow for a foreign session stamps the SESSION's directory", async () => { + await Effect.runPromise( + runTwoInstanceGuardTest( + {}, + ({ dag, store }) => + Effect.gen(function* () { + // Ambient instance is A; the target session belongs to B's + // directory. The stamp must come from the durable SESSION row, + // not the requesting instance (P2-F) — otherwise A stamps /wtA + // and B's loops never adopt the workflow. + const dagID = yield* dag.create({ + projectID: PROJECT_ID, + sessionID: SES_B, + title: "B's workflow created via A", + config: { name: "p2f", nodes: [node()] }, + }) + const wf = yield* store.getWorkflow(dagID) + expect(wf?.directory).toBe(DIR_B) + }), + ), + ) + }) + + it("P2-C: after an identity repaint, the stale instance spawns no further children", async () => { + await Effect.runPromise( + runTwoInstanceGuardTest( + {}, + ({ dag, store, database, initA, childPromptsA }) => + Effect.gen(function* () { + yield* initA + const dagID = yield* dag.create({ + projectID: PROJECT_ID, + sessionID: SES_A, + title: "Repaint spawn gate", + config: { name: "p2c", nodes: [node({ id: "n1" }), node({ id: "n2", depends_on: ["n1"] })] }, + }) + yield* takeWithin(childPromptsA, "n1 did not start") + // Identity migration: repaint the workflow's project id. The + // stale in-memory entry must stop scheduling (spawnReady + // revalidation, P2-C) — settling n1 makes n2 ready, and without + // the gate the stale instance would materialize a child for the + // migrated workflow. + yield* database.db.update(WorkflowTable) + .set({ project_id: "project-new" as never }) + .where(eq(WorkflowTable.id, dagID as never)) + .run().pipe(Effect.orDie) + yield* dag.nodeCompleted(dagID, "n1", { ok: true }) + const second = yield* Queue.take(childPromptsA).pipe(Effect.timeoutOption("1 second")) + expect(Option.isNone(second)).toBe(true) + expect((yield* store.getWorkflow(dagID))?.status).toBe("running") + expect((yield* store.getNode(dagID, "n2"))?.status).toBe("pending") + }), + ({ database }) => + database.db.insert(ProjectTable).values({ + id: "project-new" as never, + worktree: process.cwd() as never, + sandboxes: [], + }).run().pipe(Effect.orDie, Effect.as(undefined)), + ), + ) + }) + + it("P2-B: a store defect in the goal guard degrades to a skipped evaluation, not a dead loop", async () => { + const judgeCalls = new Map() + const fail = { armed: true } + await Effect.runPromise( + Effect.gen(function* () { + const database = yield* Database.Service + const goal = yield* Goal.Service + const loop = yield* GoalLoop.Service + const bridge = yield* EventV2Bridge.Service + yield* database.db.insert(ProjectTable).values({ + id: PROJECT_ID as never, + worktree: DIR_A as never, + sandboxes: [], + }).run().pipe(Effect.orDie) + yield* database.db.insert(SessionTable).values({ + id: SES_A as never, + project_id: PROJECT_ID as never, + slug: "a", + directory: DIR_A as never, + title: SES_A, + version: "test", + }).run().pipe(Effect.orDie) + yield* goal.set(SES_A as never, "ship the feature", 10) + yield* loop.init() + yield* Effect.yieldNow + // First idle: the guard's session-row read defects. The handler must + // absorb it (P2-B) — otherwise the runForEach subscription dies and + // the NEXT idle event is never evaluated. + yield* bridge.publish(SessionStatusEvent.Status, { + sessionID: SES_A as never, + status: { type: "idle" }, + }).pipe(Effect.orDie) + // Second idle: with the subscription alive, the goal is driven. + yield* bridge.publish(SessionStatusEvent.Status, { + sessionID: SES_A as never, + status: { type: "idle" }, + }).pipe(Effect.orDie) + yield* pollWithTimeout( + Effect.sync(() => ((judgeCalls.get(DIR_A)?.length ?? 0) > 0 ? true : undefined)), + "goal loop did not evaluate the idle event after the store defect (subscription died)", + "2 seconds", + ) + expect(fail.armed).toBe(false) + }).pipe( + Effect.provide(goalLoopLayer({ + judgeCalls, + messagesBySession: new Map([[SES_A, [reply(SES_A, "making progress")]]]), + ambientDatabase: Layer.effect( + Database.Service, + Effect.gen(function* () { + const real = yield* Database.Service + // One-shot defect on the first durable SELECT (the guard's + // session-row read): throws synchronously, i.e. a defect that + // Effect.ignore would NOT absorb. + const proxy = new Proxy(real.db, { + get(target, prop) { + if (prop === "select" && fail.armed) { + fail.armed = false + return () => { + throw new Error("injected transient store defect") + } + } + return Reflect.get(target, prop) + }, + }) + return Database.Service.of({ db: proxy }) + }), + ).pipe(Layer.provide(Database.layerFromPath(":memory:"))), + })), + Effect.provideService(InstanceRef, { + directory: DIR_A, + worktree: DIR_A, + project: { id: PROJECT_ID }, + } as never), + Effect.scoped, + ), + ) + }) + + it("P2-D: a NULL-directory workflow is skipped with a deduped visible warning", async () => { + const lines: string[] = [] + const collector = Logger.make((opts) => { + lines.push(String(opts.message)) + }) + await Effect.runPromise( + Effect.gen(function* () { + const database = yield* Database.Service + yield* database.db.insert(ProjectTable).values({ + id: PROJECT_ID as never, + worktree: DIR_A as never, + sandboxes: [], + }).run().pipe(Effect.orDie) + yield* database.db.insert(SessionTable).values({ + id: SES_A as never, + project_id: PROJECT_ID as never, + slug: "a", + directory: DIR_A as never, + title: SES_A, + version: "test", + }).run().pipe(Effect.orDie) + // A pre-DAG-LOC-01 workflow with no directory stamp (post-backfill + // cross-version write): every ownership check must skip it and say + // so — exactly once per workflow per process (P2-D). + yield* database.db.insert(WorkflowTable).values({ + id: "p2d-null-wf", + project_id: PROJECT_ID as never, + session_id: SES_A as never, + title: "NULL zombie", + status: "running", + config: "{}", + seq: 1, + }).run().pipe(Effect.orDie) + yield* DagLocation.ownsWorkflow("p2d-null-wf", DIR_A).pipe(Effect.ignore) + yield* DagLocation.ownsWorkflow("p2d-null-wf", DIR_A).pipe(Effect.ignore) + const warnings = lines.filter((line) => line.includes("NULL execution-location directory")) + expect(warnings).toHaveLength(1) + }).pipe( + Effect.withLogger(collector), + Effect.provide(Database.layerFromPath(":memory:")), + Effect.provideService(InstanceRef, { + directory: DIR_A, + worktree: DIR_A, + project: { id: PROJECT_ID }, + } as never), + Effect.scoped, + ), + ) + }) +}) From 70dfd4ceb1a9bf63fdfd47c422b962b941588a37 Mon Sep 17 00:00:00 2001 From: Lex Date: Thu, 13 Aug 2026 14:14:47 +0800 Subject: [PATCH 3/8] fix(dag): never end deadline supervision on a transient store defect (DAG-LOC-01 follow-up) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The deadline watcher's ownership revalidation (makeDeadlineWatcher, the DAG-LOC-01 P2-C write-section gate) was the only store read in the watcher without R13 protection: ownsWorkflow's orDie read defects on a transient store failure, the outer catchCause logs it and completes the fiber, and deadline supervision ends permanently for a still-running node — no escalation, no cap, unbounded run; nothing re-forks the watcher. Fix: wrap the revalidation in the same exit+retry pattern as the watcher's readNode (1 attempt + 3 retries with 500ms backoff, then log-and-continue). A failed read is now "cannot disprove ownership" — supervision continues — and only a POSITIVE ownership loss (successful read returning false) ends the mandate. The instance/Database presence gate is unchanged: absent either, the check does not run and supervision must not end (R13). Probe (red-first, deterministic): new P2-watcher probe in test/dag/dag-location-guards.test.ts drives makeDeadlineWatcher through the same direct-call seam the R13 watcher tests use — readNode is a mock with no Database traffic, so the watcher's only real store query is the ownership-revalidation read, and a one-shot synchronous select defect (Proxy Database, disarmed after one hit) lands exactly there. The node is past its deadline; the probe asserts the watcher still escalates (supervision survived). Red on HEAD ("watcher ended deadline supervision after a transient ownership-revalidation store defect"), green with the fix; stashing the fix re-trips the probe red, restored it is green alongside all 13 dag-location-guards probes. Verification: bun test test/dag test/goal 566 pass / 0 fail (incl. the existing R13 watcher tests), bun typecheck clean, bun lint 4850 warnings / 0 errors (ratchet unchanged). Co-Authored-By: Claude --- packages/opencode/src/dag/runtime/spawn.ts | 48 +++++--- .../test/dag/dag-location-guards.test.ts | 115 +++++++++++++++++- 2 files changed, 147 insertions(+), 16 deletions(-) diff --git a/packages/opencode/src/dag/runtime/spawn.ts b/packages/opencode/src/dag/runtime/spawn.ts index 8dec5eed93..0c29a727d1 100644 --- a/packages/opencode/src/dag/runtime/spawn.ts +++ b/packages/opencode/src/dag/runtime/spawn.ts @@ -129,6 +129,38 @@ export function makeDeadlineWatcher( yield* Effect.logWarning("DAG deadline watcher giving up after store read retries", { dagID: input.dagID, nodeID: input.nodeID }) return undefined }) + // DAG-LOC-01 (P2-C + review P2): revalidate ownership before the write + // section — escalation and cap-enforcement writes must not land on a + // workflow whose durable identity was repainted (identity migration) or + // whose rows were cascade-deleted. Losing ownership ends this watcher's + // mandate; the instance that owns the repainted workflow supervises it. + // The check only runs when it can DISPROVE ownership (instance context + // and Database both present — always true for watchers forked from + // DagLoop): absent either, supervision must not end (R13). + // + // Review P2: this read follows the same exit+retry pattern as readNode + // above — a transient store defect must be treated as "cannot disprove + // ownership" (continue supervising), never as a reason to end the + // mandate. Unretried, it is the watcher's only store read without R13 + // protection: a defect dies through the outer catchCause, which logs and + // completes the fiber — permanently ending deadline supervision for a + // still-running node (no escalation, no cap, unbounded run; nothing + // re-forks the watcher). + const ownershipLost = Effect.gen(function* () { + const instance = yield* InstanceRef + const db = yield* Effect.serviceOption(Database.Service) + if (!instance || db._tag === "None") return false + for (let attemptNo = 0; attemptNo <= 3; attemptNo++) { + const outcome = yield* DagLocation.ownsWorkflow(input.dagID, instance.directory).pipe(Effect.exit) + if (Exit.isSuccess(outcome)) return !outcome.value + if (attemptNo < 3) yield* Effect.sleep(500) + } + yield* Effect.logWarning( + "DAG deadline watcher ownership revalidation failed after store retries — keeping supervision", + { dagID: input.dagID, nodeID: input.nodeID }, + ) + return false + }) for (;;) { const node = yield* readNode if (!node) { @@ -140,21 +172,7 @@ export function makeDeadlineWatcher( continue } if (isNodeTerminalStatus(node.status as never)) return - // DAG-LOC-01 (P2-C): revalidate ownership before the write section — - // escalation and cap-enforcement writes must not land on a workflow - // whose durable identity was repainted (identity migration) or whose - // rows were cascade-deleted. Losing ownership ends this watcher's - // mandate; the instance that owns the repainted workflow supervises it. - // The check only runs when it can DISPROVE ownership (instance context - // and Database both present — always true for watchers forked from - // DagLoop): absent either, supervision must not end (R13). - { - const instance = yield* InstanceRef - const db = yield* Effect.serviceOption(Database.Service) - if (instance && db._tag === "Some") { - if (!(yield* DagLocation.ownsWorkflow(input.dagID, instance.directory))) return - } - } + if (yield* ownershipLost) return const now = yield* Clock.currentTimeMillis const deadline = node.deadlineMs if (node.status !== "running") { diff --git a/packages/opencode/test/dag/dag-location-guards.test.ts b/packages/opencode/test/dag/dag-location-guards.test.ts index eb194e77b2..2fb9ba29c4 100644 --- a/packages/opencode/test/dag/dag-location-guards.test.ts +++ b/packages/opencode/test/dag/dag-location-guards.test.ts @@ -34,7 +34,7 @@ * AMBIENT instance directory, so each probe can tell which instance acted. */ import { describe, expect, it } from "bun:test" -import { DateTime, Deferred, Effect, Layer, Option, Queue, Logger } from "effect" +import { DateTime, Deferred, Effect, Fiber, Layer, Option, Queue, Logger, Scope } from "effect" 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" @@ -49,6 +49,7 @@ import { SessionStatusEvent } from "@opencode-ai/schema/session-status-event" import { Agent } from "@/agent/agent" import { Dag, type NodeConfig } from "@/dag/dag" import { DagLoop } from "@/dag/runtime/loop" +import { makeDeadlineWatcher } from "@/dag/runtime/spawn" import { DagLocation } from "@/dag/location" import { Goal } from "@/goal/goal" import { GoalLoop, GoalLoopJudgeLLM } from "@/goal/loop" @@ -60,6 +61,7 @@ import { MessageID } from "@/session/schema" import { Session } from "@/session/session" import { SessionStatus } from "@/session/status" import { pollWithTimeout } from "../lib/effect" +import { makeNodeRow } from "./fixtures" import { eq } from "drizzle-orm" import { existsSync, readFileSync, readdirSync } from "node:fs" import path from "node:path" @@ -1011,4 +1013,115 @@ describe("DAG-LOC-01 P2 follow-ups", () => { ), ) }) + + it("P2-watcher: a transient store defect in the ownership revalidation does not end deadline supervision", async () => { + const fail = { armed: true } + let escalations = 0 + // Deterministic defect placement through the direct-call seam (the same + // seam as the R13 watcher tests): readNode is a mock with no Database + // traffic, so the watcher's ONLY real store query is the + // ownership-revalidation read — the one-shot select defect lands exactly + // there (review P2, makeDeadlineWatcher). + const storeLayer = Layer.mock(DagStore.Service)({ + getNode: () => + Effect.succeed( + makeNodeRow({ + id: "n1", + workflowId: "p2w", + name: "n1", + status: "running", + deadlineMs: 1, + timeoutExtensions: 0, + childSessionId: "ses_child_1", + }), + ), + }) + const dagLayer = Layer.unwrap( + Effect.map(DagStore.Service, (store) => + Layer.mock(Dag.Service)({ + store, + nodeTimeoutEscalated: () => + Effect.sync(() => { + escalations++ + }), + }), + ), + ).pipe(Layer.provide(storeLayer)) + const promptLayer = Layer.mock(SessionPrompt.Service, { + cancel: () => Effect.void, + }) + const databaseLayer = Layer.effect( + Database.Service, + Effect.gen(function* () { + const real = yield* Database.Service + // One-shot defect on the first durable SELECT: throws synchronously + // (a defect the plain `yield*` cannot absorb) and disarms so the + // revalidation's retry reads the real row. + const proxy = new Proxy(real.db, { + get(target, prop) { + if (prop === "select" && fail.armed) { + fail.armed = false + return () => { + throw new Error("injected transient store defect") + } + } + return Reflect.get(target, prop) + }, + }) + return Database.Service.of({ db: proxy }) + }), + ).pipe(Layer.provide(Database.layerFromPath(":memory:"))) + await Effect.runPromise( + Effect.gen(function* () { + const database = yield* Database.Service + yield* database.db.insert(ProjectTable).values({ + id: PROJECT_ID as never, + worktree: process.cwd() as never, + sandboxes: [], + }).run().pipe(Effect.orDie) + yield* database.db.insert(SessionTable).values({ + id: SES_A as never, + project_id: PROJECT_ID as never, + slug: "a", + directory: process.cwd() as never, + title: SES_A, + version: "test", + }).run().pipe(Effect.orDie) + yield* database.db.insert(WorkflowTable).values({ + id: "p2w", + project_id: PROJECT_ID as never, + session_id: SES_A as never, + title: "P2 watcher revalidation", + status: "running", + config: "{}", + seq: 1, + directory: process.cwd() as never, + }).run().pipe(Effect.orDie) + const scope = yield* Scope.Scope + const watcher = yield* makeDeadlineWatcher({ dagID: "p2w", nodeID: "n1", timeoutMs: 300 }).pipe( + Effect.forkIn(scope), + ) + // The node is past its deadline; the watcher must still escalate + // after the transient defect — proof supervision survived. Without + // the retry the defect dies through the outer catchCause, which + // completes the fiber, and the escalation never happens. + yield* pollWithTimeout( + Effect.sync(() => (escalations > 0 ? true : undefined)), + "watcher ended deadline supervision after a transient ownership-revalidation store defect (review P2 regression)", + ) + expect(fail.armed).toBe(false) + yield* Fiber.interrupt(watcher).pipe(Effect.ignore) + }).pipe( + Effect.provide(dagLayer), + Effect.provide(promptLayer), + Effect.provide(databaseLayer), + Effect.provideService(InstanceRef, { + directory: process.cwd(), + worktree: process.cwd(), + project: { id: PROJECT_ID }, + } as never), + Effect.scoped, + ), + ) + }) }) From 7ed147f8ec50e9e455f381980e30a12fd3ea04c6 Mon Sep 17 00:00:00 2001 From: Lex Date: Sat, 15 Aug 2026 01:00:13 +0800 Subject: [PATCH 4/8] test(dag): stamp lease-lifecycle seed rows for the execution-location invariant (DAG-LOC-01 rebase integration) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Root cause: the GOAL-FP-01 lease-lifecycle tests seed workflow rows via raw SQL with no execution-location stamp. The rebased DAG-LOC-01 ownership authority is fail-closed on NULL-directory rows (P2-D zombie policy: never adopted, recovered, or woken), so the startup wake sweep stopped registering the ghost row and the runtime-less terminal-release test lost its swept-registration precondition. Fix: stamp all four seed rows with the instance directory (process.cwd(), matching InstanceRef in the harness) — the same idiom the sibling guard tests use — so the rows represent legitimately owned workflows and the lease assertions exercise their original intent. --- packages/opencode/test/dag/dag-lease-lifecycle.test.ts | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/packages/opencode/test/dag/dag-lease-lifecycle.test.ts b/packages/opencode/test/dag/dag-lease-lifecycle.test.ts index df58e5e1c9..a19cfe60d9 100644 --- a/packages/opencode/test/dag/dag-lease-lifecycle.test.ts +++ b/packages/opencode/test/dag/dag-lease-lifecycle.test.ts @@ -275,6 +275,7 @@ describe("DagLoop lease lifecycle — startup wake sweep (GOAL-FP-01-01)", () => id: "dag-wf-done", project_id: Project.ID.make(PROJECT_ID), session_id: SessionID.make(PARENT_SESSION), + directory: process.cwd(), title: "already reported", status: "completed", config: "", @@ -289,6 +290,7 @@ describe("DagLoop lease lifecycle — startup wake sweep (GOAL-FP-01-01)", () => id: "dag-wf-undone", project_id: Project.ID.make(PROJECT_ID), session_id: SessionID.make(PARENT_SESSION), + directory: process.cwd(), title: "terminal before delivery", status: "failed", config: "", @@ -391,6 +393,7 @@ describe("DagLoop lease lifecycle — runtime-less terminal release (GOAL-FP-01- id: "dag-wf-ghost", project_id: Project.ID.make(PROJECT_ID), session_id: SessionID.make(PARENT_SESSION), + directory: process.cwd(), title: "unrecoverable", status: "running", config: "", @@ -423,6 +426,7 @@ describe("DagLoop lease lifecycle — runtime-less terminal release (GOAL-FP-01- id: "dag-wf-undone", project_id: Project.ID.make(PROJECT_ID), session_id: SessionID.make(PARENT_SESSION), + directory: process.cwd(), title: "terminal before delivery", status: "failed", config: "", From 959bae7c24b2513ae4786b9ad631958874e2d21a Mon Sep 17 00:00:00 2001 From: Lex Date: Sat, 15 Aug 2026 01:06:46 +0800 Subject: [PATCH 5/8] fix(dag): latch the WorkflowStarted adoption slot against duplicate-publish races (DAG-LOC-01 H1) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Root cause: the WorkflowStarted handler was the only adoption path without an in-flight reservation — recoverWorkflow and recoverOrphanPending both reserve the `recovering` slot before their first yield, but the live handler checked the runtimes/recovering guard and then yielded through getWorkflow/getNodes with nothing reserved. Two concurrent WorkflowStarted events (a duplicate publish racing the live handler) both passed the guard and both reached runtimes.set; the second overwrote the first entry, orphaning its fibers/watchers from every interrupt sweep and double-registering the automation lease. Fix: reserve recovering.add(dagID) synchronously right after the guard (no yield between check and add, so the loser's guard observes the reservation) and release it via Effect.ensuring — exactly mirroring recoverWorkflow's idempotency discipline. The latch also supersedes the WorkflowReplanned no-entry re-adoption race: a replan arriving mid-adoption drops out of recoverWorkflow instead of overwriting the entry, and the adoption's own getNodes reads the already-replanned rows. The vs-deletion tail of the P2-E window is NOT closed here (that would require DB-level atomic adoption = ownership-token redesign, out of scope pre-clustering); eviction-on-next-stimulus in spawnReady's ownership revalidation is the accepted mitigation for that remainder. --- packages/opencode/src/dag/runtime/loop.ts | 85 ++++++++++++++--------- 1 file changed, 52 insertions(+), 33 deletions(-) diff --git a/packages/opencode/src/dag/runtime/loop.ts b/packages/opencode/src/dag/runtime/loop.ts index 3ec6bbeddb..0d7b37f685 100644 --- a/packages/opencode/src/dag/runtime/loop.ts +++ b/packages/opencode/src/dag/runtime/loop.ts @@ -508,40 +508,59 @@ const serviceLayer = Layer.effect( // orphan sweep publishes WorkflowStarted only to legalize its // pending→running→failed terminalization, and adopting the // orphan mid-sequence would start scheduling on a dead workflow. + // + // H1 (DAG-LOC-01): the handler also reserves the slot for + // ITSELF — two concurrent WorkflowStarted events (e.g. a + // duplicate publish racing the live handler) previously both + // passed the guard above (neither reserves anything) and both + // reached runtimes.set: the second overwrote the first entry, + // orphaning its fibers/watchers from every interrupt sweep and + // double-registering the automation lease. Reserve + // synchronously right after the guard — no yield between the + // check and the add, so the second event's guard sees the + // reservation — and release via Effect.ensuring, exactly + // mirroring recoverWorkflow / recoverOrphanPending. The latch + // also supersedes the WorkflowReplanned no-entry re-adoption + // race: a replan arriving mid-adoption drops out of + // recoverWorkflow instead of overwriting this entry, and the + // adoption's own getNodes reads the already-replanned rows. if (runtimes.has(dagID) || recovering.has(dagID)) return - const wf = yield* store.getWorkflow(dagID).pipe(Effect.orDie) - if (!wf) return - // Status guard: the orphan-pending sweep publishes WorkflowStarted - // only to legalize the pending→running leg of its terminalization - // sequence. By the time the event reaches this handler the row is - // already failed — adopting it would rebuild a runtime and start - // scheduling nodes on a dead workflow. Accept running rows only. - if (wf.status !== "running") return - // Cross-instance guard via the execution-location authority: - // only the instance whose DIRECTORY owns the workflow adopts - // (see recoverWorkflow). First-wave spawns must not race across - // directory contexts. - if (!(yield* DagLocation.ownsWorkflow(dagID, ctx.directory))) return - const config = parseWorkflowConfig(wf.config) - const nodes = yield* store.getNodes(dagID) - const maxConcurrency = Math.max(1, config?.max_concurrency ?? Dag.DEFAULT_WORKFLOW_CONFIG.maxConcurrency) - const runtime = new WorkflowRuntime(toSchedulingNodes(nodes), maxConcurrency) - const semaphore = Semaphore.makeUnsafe(maxConcurrency) - const entry: WorkflowEntry = { runtime, semaphore, evalLock: Semaphore.makeUnsafe(1), parentSessionID: wf.sessionId, config, fibers: new Map(), watchers: new Map() } - // P2-E deletion-race re-check (same window as recoverWorkflow): - // the Deleted sweep only removes entries already in `runtimes`, - // and getNodes above is an awaited yield a deletion can slip - // 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 - runtimes.set(dagID, entry) - yield* automation.register(SessionID.make(wf.sessionId), { kind: "dag", id: dagID }) - yield* entry.evalLock.withPermits(1)( - Effect.gen(function* () { - yield* spawnReady(dagID) - yield* checkCompletion(dagID) - }), - ) + recovering.add(dagID) + yield* Effect.gen(function* () { + const wf = yield* store.getWorkflow(dagID).pipe(Effect.orDie) + if (!wf) return + // Status guard: the orphan-pending sweep publishes WorkflowStarted + // only to legalize the pending→running leg of its terminalization + // sequence. By the time the event reaches this handler the row is + // already failed — adopting it would rebuild a runtime and start + // scheduling nodes on a dead workflow. Accept running rows only. + if (wf.status !== "running") return + // Cross-instance guard via the execution-location authority: + // only the instance whose DIRECTORY owns the workflow adopts + // (see recoverWorkflow). First-wave spawns must not race across + // directory contexts. + if (!(yield* DagLocation.ownsWorkflow(dagID, ctx.directory))) return + const config = parseWorkflowConfig(wf.config) + const nodes = yield* store.getNodes(dagID) + const maxConcurrency = Math.max(1, config?.max_concurrency ?? Dag.DEFAULT_WORKFLOW_CONFIG.maxConcurrency) + const runtime = new WorkflowRuntime(toSchedulingNodes(nodes), maxConcurrency) + const semaphore = Semaphore.makeUnsafe(maxConcurrency) + const entry: WorkflowEntry = { runtime, semaphore, evalLock: Semaphore.makeUnsafe(1), parentSessionID: wf.sessionId, config, fibers: new Map(), watchers: new Map() } + // P2-E deletion-race re-check (same window as recoverWorkflow): + // the Deleted sweep only removes entries already in `runtimes`, + // and getNodes above is an awaited yield a deletion can slip + // 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 + runtimes.set(dagID, entry) + yield* automation.register(SessionID.make(wf.sessionId), { kind: "dag", id: dagID }) + yield* entry.evalLock.withPermits(1)( + Effect.gen(function* () { + yield* spawnReady(dagID) + yield* checkCompletion(dagID) + }), + ) + }).pipe(Effect.ensuring(Effect.sync(() => recovering.delete(dagID)))) }).pipe(guarded("WorkflowStarted")), ), Effect.forkScoped({ startImmediately: true }), From 9fc057610e6ecc3cdab5a9712ea047ff6ebf3d75 Mon Sep 17 00:00:00 2001 From: Lex Date: Sat, 15 Aug 2026 01:22:09 +0800 Subject: [PATCH 6/8] test(dag): add issue #238 evidence probes C1/C3/C4/C5 + R7-ext write-once barrier (DAG-LOC-01) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Pin the four evidence questions the hardening review left open, each on a deterministic seam (no timing dependence): - C1 concurrent live adoption: both instances booted before the workflow exists; only the stamped directory adopts and spawns, the sibling does not. - C3 cascade-in-window orphan: direct row deletion (no Deleted event) leaves the live entry unreachable by the sweep; later stimuli spawn nothing — every action path revalidates ownership against the missing row. - C4 moved-session wedge pin: mixed directory stamps across a session's workflows leave NO directory owner (create-time-stamp semantics; re-stamping on SessionEvent.Moved stays out of scope pre-clustering). - C5 teardown replay idempotency: replaying a deleted workflow's durable journal through EventV2 replay does not resurrect the read-model (seq dedup skips projection). - R7-ext static barrier: the directory stamp is write-once (no UPDATE writes it anywhere in the dag trees) and spawnReady / checkCompletion / makeDeadlineWatcher / the GoalLoop idle guard each carry the ownership authority. --- .../test/dag/dag-location-guards.test.ts | 238 ++++++++++++++++++ 1 file changed, 238 insertions(+) diff --git a/packages/opencode/test/dag/dag-location-guards.test.ts b/packages/opencode/test/dag/dag-location-guards.test.ts index 2fb9ba29c4..7ed02ed7a4 100644 --- a/packages/opencode/test/dag/dag-location-guards.test.ts +++ b/packages/opencode/test/dag/dag-location-guards.test.ts @@ -63,6 +63,7 @@ import { SessionStatus } from "@/session/status" import { pollWithTimeout } from "../lib/effect" import { makeNodeRow } from "./fixtures" import { eq } from "drizzle-orm" +import { EventTable } from "@opencode-ai/core/event/sql" import { existsSync, readFileSync, readdirSync } from "node:fs" import path from "node:path" @@ -696,6 +697,78 @@ 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. + 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. + const writesDirectory = sources + .filter(({ source }) => /\.set\(\{[\s\S]{0,300}?\bdirectory\s*:/.test(source)) + .map(({ file }) => file) + expect(writesDirectory).toEqual([]) + + // (b) Revalidation sites. Each region must carry an executable ownership + // authority call (ownsWorkflow / ownsSession), located by semantic anchors + // so the probe survives refactors that keep the site boundaries. + const loopFile = sources.find((s) => s.file.endsWith("opencode/src/dag/runtime/loop.ts")) + expect(loopFile).toBeDefined() + const loopSource = loopFile!.source + const spawnFile = sources.find((s) => s.file.endsWith("opencode/src/dag/runtime/spawn.ts")) + expect(spawnFile).toBeDefined() + + const regions: Array<{ name: string; source: string; from: string; to: string; authority: RegExp }> = [ + { + name: "spawnReady (pre-spawn ownership revalidation + inert-entry eviction)", + source: loopSource, + from: "const spawnReady = Effect.fn(", + to: "const checkCompletion = Effect.fn(", + authority: /ownsWorkflow\(/, + }, + { + name: "checkCompletion (terminal-transition ownership revalidation)", + source: loopSource, + from: "const checkCompletion = Effect.fn(", + to: "const checkSessionStatus = makeSessionStatusChecker", + authority: /ownsWorkflow\(/, + }, + { + name: "makeDeadlineWatcher (deadline-supervision ownership revalidation)", + source: spawnFile!.source, + from: "export function makeDeadlineWatcher(", + to: "export function spawnNode(", + authority: /ownsWorkflow\(/, + }, + ] + const unguarded: string[] = [] + for (const region of regions) { + const start = region.source.indexOf(region.from) + const end = region.source.indexOf(region.to, start) + if (start === -1 || end === -1) { + unguarded.push(`${region.name}: anchor not found (from="${region.from}" to="${region.to}")`) + continue + } + if (!codeLines(region.source.slice(start, end)).some((line) => region.authority.test(line))) { + unguarded.push(`${region.name}: no ownership-authority call in handler body`) + } + } + + // The goal-side idle guard lives in src/goal/loop.ts (outside the dag + // trees) and keys on the session row via Goal.ownsSession. + const goalLoopSource = readFileSync(path.resolve(import.meta.dir, "../../src/goal/loop.ts"), "utf8") + if (!codeLines(goalLoopSource).some((line) => /ownsSession\(/.test(line))) { + unguarded.push("GoalLoop idle-Status guard: no Goal.ownsSession call") + } + expect(unguarded).toEqual([]) + }) }) // --------------------------------------------------------------------------- @@ -1125,3 +1198,168 @@ describe("DAG-LOC-01 P2 follow-ups", () => { ) }) }) + +// --------------------------------------------------------------------------- +// #238 evidence probes C1–C6 (TOCTOU / teardown-idempotency / negative barriers) +// --------------------------------------------------------------------------- + +describe("DAG-LOC-01 issue #238 evidence probes", () => { + it("C1: concurrent live adoption — only the stamped directory adopts", async () => { + await Effect.runPromise( + runTwoInstanceGuardTest( + {}, + ({ dag, store, initA, initB, childPromptsA, childPromptsB }) => + Effect.gen(function* () { + // Boot BOTH instances before the workflow exists so each live + // WorkflowStarted subscription is already armed when adoption + // fires — no timing dependence on which boot wins. + yield* initA + yield* initB + // Session-sourced stamp: SES_A lives in DIR_A, so the row is + // stamped DIR_A regardless of which instance creates it. + const dagID = yield* dag.create({ + projectID: PROJECT_ID, + sessionID: SES_A, + title: "concurrent live adoption", + config: { name: "c1", nodes: [node()] }, + }) + expect((yield* store.getWorkflow(dagID))?.directory).toBe(DIR_A) + // The owner adopts and spawns its first wave... + const owner = yield* takeWithin(childPromptsA, "owner instance did not adopt and spawn") + expect(owner.input.sessionID).toBeDefined() + // ...and the sibling instance must NOT adopt within the window. + yield* Effect.sleep("300 millis") + expect(Option.getOrElse(yield* Queue.poll(childPromptsB), () => null)).toBe(null) + }), + ), + ) + }) + + it("C3: an entry orphaned by cascade-in-window deletion acts on no later stimulus", async () => { + await Effect.runPromise( + runTwoInstanceGuardTest( + {}, + ({ dag, store, database, bridge, initA, childPromptsA }) => + Effect.gen(function* () { + yield* initA + const dagID = yield* dag.create({ + projectID: PROJECT_ID, + sessionID: SES_A, + title: "cascade-window orphan", + config: { name: "c3", nodes: [node({ id: "n1" }), node({ id: "n2", depends_on: ["n1"] })] }, + }) + yield* takeWithin(childPromptsA, "n1 did not start") + // Simulate the P2-E cascade-in-window: delete the workflow row + // directly (FK cascade wipes the node rows) WITHOUT publishing a + // SessionV1.Event.Deleted, so the Deleted sweep can never reach + // the live in-memory entry. + yield* database.db.delete(WorkflowTable) + .where(eq(WorkflowTable.id, dagID as never)) + .run().pipe(Effect.orDie) + expect(yield* store.getWorkflow(dagID)).toBeUndefined() + // Stimulus 1: settle n1 via a direct bus publish (dag.nodeCompleted's + // guard rejects a missing node). The orphaned entry must not spawn n2. + yield* bridge.publish(DagEvent.NodeCompleted, { + dagID: dagID as never, + nodeID: "n1" as never, + output: { ok: true }, + durationMs: 0 as never, + timestamp: yield* DateTime.now, + }).pipe(Effect.orDie) + expect(Option.isNone(yield* Queue.take(childPromptsA).pipe(Effect.timeoutOption("1 second")))).toBe(true) + // Stimulus 2: a follow-up completion also no-ops — the entry is + // inert (every action path revalidates ownership against the + // missing row), so nothing is ever spawned for a deleted workflow. + yield* bridge.publish(DagEvent.NodeCompleted, { + dagID: dagID as never, + nodeID: "n2" as never, + output: { ok: true }, + durationMs: 0 as never, + timestamp: yield* DateTime.now, + }).pipe(Effect.orDie) + expect(Option.isNone(yield* Queue.take(childPromptsA).pipe(Effect.timeoutOption("1 second")))).toBe(true) + }), + ), + ) + }) + + it("C4: a moved session's mixed stamps leave NO directory owner (pre-clustering wedge pin)", async () => { + await Effect.runPromise( + runTwoInstanceGuardTest( + {}, + ({ dag, store, database }) => + Effect.gen(function* () { + // wf1 created while SES_A lives in DIR_A → stamped DIR_A. + const wf1 = yield* dag.create({ + projectID: PROJECT_ID, + sessionID: SES_A, + title: "before move", + 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) + // wf2 created after the move → session-sourced stamp = DIR_B. + const wf2 = yield* dag.create({ + projectID: PROJECT_ID, + sessionID: SES_A, + title: "after move", + 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). + expect(yield* DagLocation.ownsSession(SES_A, DIR_A)).toBe(false) + expect(yield* DagLocation.ownsSession(SES_A, DIR_B)).toBe(false) + }), + ), + ) + }) + + it("C5: replaying a deleted workflow's journal does not resurrect the read-model", async () => { + await Effect.runPromise( + runTwoInstanceGuardTest( + {}, + ({ dag, store, database, bridge, session }) => + Effect.gen(function* () { + const dagID = yield* dag.create({ + projectID: PROJECT_ID, + sessionID: SES_A, + title: "serialize me", + config: { name: "c5", nodes: [node()] }, + }) + expect((yield* store.getWorkflow(dagID))?.status).toBe("running") + // Serialize the dag aggregate journal BEFORE deletion. + const rows = yield* database.db.select().from(EventTable) + .where(eq(EventTable.aggregate_id, dagID)) + .orderBy(EventTable.seq) + .all().pipe(Effect.orDie) + const serialized = rows.map((r) => ({ + id: r.id, + type: r.type, + seq: r.seq, + aggregateID: r.aggregate_id, + data: r.data, + })) + expect(serialized.length).toBeGreaterThan(0) + // Remove the session: FK cascade wipes the workflow/node read-model + // rows, but the dag aggregate's durable events + sequence survive. + yield* session.remove(SES_A as never) + expect(yield* store.getWorkflow(dagID)).toBeUndefined() + // Replay the journal through EventV2 replay: the durable-seq dedup + // (input.seq <= latest) skips projection, so the read-model stays + // deleted — a crash-recovery replay cannot resurrect a torn-down + // workflow. + yield* bridge.replayAll(serialized) + expect(yield* store.getWorkflow(dagID)).toBeUndefined() + }), + ), + ) + }) +}) From 9a97e886658d3efc07e8df2e19879dc8056f0c97 Mon Sep 17 00:00:00 2001 From: Lex Date: Sat, 15 Aug 2026 01:36:08 +0800 Subject: [PATCH 7/8] test(dag): add issue #238 async race probes C2/C6 + Deterministic park gates (DAG-LOC-01) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The two remaining evidence questions were async negative-test barriers — failure absorption that had no regression probe because the race window was not deterministically reachable. Add two reusable park gates to the two-instance harness and pin both paths through them: - parkGetNodes: every DagStore.getNodes call flags parked and awaits a caller promise before delegating, so a probe can interleave a mutation inside a recovery sequence. - parkWakeDelivery: SessionPrompt.prepareIfIdle (the wake-delivery admission seam tryDeliverWake actually uses) parks the admission result on a caller promise before release, and afterwards returns none while still counting the call. Probes: - C2 Session.remove racing an in-flight wake: delete the session while the wake delivery is parked; the raced defect is absorbed by tryDeliverWake's catchCause (no escape, wakeInFlight freed) and the idle wake subscription still processes a later session's idle (prepareIfIdle called again). - C6 recoverOrphanPending racing Session.remove: delete the session while the orphan sweep is parked between getNodes and dag.fail; dag.fail on the gone workflow fails, the startup-scan catchCause absorbs it, the Effect.ensuring frees the recovering slot, and init completes cleanly. --- .../test/dag/dag-location-guards.test.ts | 199 +++++++++++++++++- 1 file changed, 195 insertions(+), 4 deletions(-) diff --git a/packages/opencode/test/dag/dag-location-guards.test.ts b/packages/opencode/test/dag/dag-location-guards.test.ts index 7ed02ed7a4..915cbf8b70 100644 --- a/packages/opencode/test/dag/dag-location-guards.test.ts +++ b/packages/opencode/test/dag/dag-location-guards.test.ts @@ -39,7 +39,7 @@ 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 { WorkflowTable } from "@opencode-ai/core/dag/sql" +import { WorkflowTable, WorkflowNodeTable } from "@opencode-ai/core/dag/sql" import { DagStore } from "@opencode-ai/core/dag/store" import { EventV2 } from "@opencode-ai/core/event" import { ProjectTable } from "@opencode-ai/core/project/sql" @@ -131,6 +131,27 @@ interface TwoInstanceInput { readonly messagesBySession: Map /** Injected one-shot defects for DagStore.getWorkflow (parity with the template). */ readonly failGetWorkflow?: { remaining: number } + /** + * Optional deterministic park on DagStore.getNodes. When present, every + * getNodes call sets `parked.value = true` and then awaits `wait` before + * delegating to the real store — letting a probe interleave a mutation + * (e.g. Session.remove) inside a recovery sequence. + */ + readonly parkGetNodes?: { readonly wait: Promise; readonly parked: { value: boolean } } + /** + * Optional deterministic park on SessionPrompt.prepareIfIdle (the wake + * delivery admission seam). When present, each prepareIfIdle call bumps + * `calls`; before `released.value` it parks the admission's result effect + * on `wait` so a probe can interleave Session.remove mid-delivery. After + * 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 } + } } function twoInstanceLayer(input: TwoInstanceInput) { @@ -138,7 +159,8 @@ function twoInstanceLayer(input: TwoInstanceInput) { 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 store = input.failGetWorkflow + const needsStoreWrapper = Boolean(input.failGetWorkflow) || Boolean(input.parkGetNodes) + const store = needsStoreWrapper ? Layer.effect( DagStore.Service, Effect.gen(function* () { @@ -147,12 +169,19 @@ function twoInstanceLayer(input: TwoInstanceInput) { ...real, getWorkflow: (id) => Effect.suspend(() => { - if (input.failGetWorkflow!.remaining > 0) { - input.failGetWorkflow!.remaining-- + if (input.failGetWorkflow && input.failGetWorkflow.remaining > 0) { + input.failGetWorkflow.remaining-- return Effect.die(new Error("injected transient db failure")) } return real.getWorkflow(id) }), + getNodes: (id) => + Effect.suspend(() => { + const gate = input.parkGetNodes + if (!gate) return real.getNodes(id) + gate.parked.value = true + return Effect.promise(() => gate.wait).pipe(Effect.flatMap(() => real.getNodes(id))) + }), }) }), ).pipe(Layer.provide(realStore)) @@ -245,6 +274,24 @@ function twoInstanceLayer(input: TwoInstanceInput) { }), prompt: deliver, promptIfIdle: (value) => deliver(value).pipe(Effect.map(Option.some)), + // Wake-delivery admission seam (C2). tryDeliverWake delivers through + // admitIfIdle → prepareIfIdle, not promptIfIdle. Without a gate this + // returns none (no admission); with parkWakeDelivery it parks the result + // effect on the gate so a probe can race Session.remove mid-delivery. + prepareIfIdle: (value) => + Effect.sync(() => { + const gate = input.parkWakeDelivery + if (!gate) return Option.none() + gate.calls.value++ + if (gate.released.value) return Option.none() + gate.parked.value = true + const result = Effect.promise(() => gate.wait).pipe( + Effect.flatMap(() => + Effect.die(new Error(`session ${(value as { sessionID?: string }).sessionID} removed during wake delivery`)), + ), + ) + return Option.some({ activate: Effect.void, result, abort: Effect.void }) + }), }) const agent = Layer.mock(Agent.Service, { get: () => Effect.succeed({ @@ -298,6 +345,13 @@ function runTwoInstanceGuardTest( readonly sessionA?: string readonly sessionB?: string readonly failGetWorkflow?: { remaining: number } + readonly parkGetNodes?: { readonly wait: Promise; readonly parked: { value: boolean } } + readonly parkWakeDelivery?: { + readonly wait: Promise + readonly parked: { value: boolean } + readonly released: { value: boolean } + readonly calls: { value: number } + } }, test: (services: TwoInstanceServices) => Effect.Effect, beforeInit?: (services: { readonly database: Database.Interface }) => Effect.Effect, @@ -380,6 +434,8 @@ function runTwoInstanceGuardTest( promptInterrupts, messagesBySession, failGetWorkflow: options.failGetWorkflow, + parkGetNodes: options.parkGetNodes, + parkWakeDelivery: options.parkWakeDelivery, })), Effect.provideService(InstanceRef, { directory: directoryA, @@ -1362,4 +1418,139 @@ describe("DAG-LOC-01 issue #238 evidence probes", () => { ), ) }) + + it("C2: Session.remove racing an in-flight wake is absorbed and the idle subscription survives", async () => { + const parked = { value: false } + const released = { value: false } + const calls = { value: 0 } + let release: () => void = () => {} + const wait = new Promise((resolve) => { + release = () => { + released.value = true + resolve() + } + }) + await Effect.runPromise( + runTwoInstanceGuardTest( + { parkWakeDelivery: { wait, parked, released, calls } }, + ({ database, bridge, session, initA }) => + Effect.gen(function* () { + yield* initA + // Re-arm BOTH stamped terminal wakes after the startup sweep passed + // over them (they were seeded wake_reported=true), so the delivery + // below can only come from the idle-Status path. + for (const id of ["wake-wf-a", "wake-wf-b"]) { + yield* database.db.update(WorkflowTable) + .set({ wake_reported: false }) + .where(eq(WorkflowTable.id, id)) + .run().pipe(Effect.orDie) + } + // Idle SES_A → wake delivery parks inside prepareIfIdle. + yield* bridge.publish(SessionStatusEvent.Status, { + sessionID: SES_A as never, + status: { type: "idle" }, + }).pipe(Effect.orDie) + yield* pollWithTimeout( + Effect.sync(() => (parked.value ? true : undefined)), + "wake delivery never parked in prepareIfIdle", + ) + expect(calls.value).toBe(1) + // Race: delete SES_A while the delivery is parked — FK cascade wipes + // the wake rows out from under the in-flight delivery. + yield* session.remove(SES_A as never) + // Release: the parked result fails; tryDeliverWake's catchCause must + // absorb it (no defect) and the finally must free wakeInFlight. + release() + yield* Effect.sleep("200 millis") + // Subscription survival: an idle for a DIFFERENT session still reaches + // prepareIfIdle (calls bumps again) — proof the idle wake subscription + // is alive after the raced deletion. + yield* bridge.publish(SessionStatusEvent.Status, { + sessionID: SES_B as never, + status: { type: "idle" }, + }).pipe(Effect.orDie) + yield* pollWithTimeout( + Effect.sync(() => (calls.value >= 2 ? true : undefined)), + "idle wake subscription died after the Session.remove race", + ) + expect(calls.value).toBe(2) + }), + ({ database }) => + Effect.gen(function* () { + for (const [id, sessionID] of [ + ["wake-wf-a", SES_A], + ["wake-wf-b", SES_B], + ] as const) { + yield* database.db.insert(WorkflowTable).values({ + id, + project_id: PROJECT_ID as never, + session_id: sessionID as never, + directory: DIR_A as never, + title: `terminal wake ${id}`, + status: "failed", + config: "{}", + seq: 1, + wake_reported: true, + }).run().pipe(Effect.orDie) + } + }), + ), + ) + }) + + it("C6: recoverOrphanPending racing Session.remove is absorbed without killing init", async () => { + const parked = { value: false } + let release: () => void = () => {} + const wait = new Promise((resolve) => { + release = resolve + }) + await Effect.runPromise( + runTwoInstanceGuardTest( + { parkGetNodes: { wait, parked } }, + ({ database, session, store, initA }) => + Effect.gen(function* () { + // An all-pending orphan under A: a create that crashed mid-way. + // It is stamped DIR_A so instance A's recoverOrphanPending owns it. + yield* database.db.insert(WorkflowTable).values({ + id: "c6-orphan", + project_id: PROJECT_ID as never, + session_id: SES_A as never, + directory: DIR_A as never, + title: "pending orphan", + status: "pending", + config: "{}", + seq: 1, + wake_reported: true, + }).run().pipe(Effect.orDie) + yield* database.db.insert(WorkflowNodeTable).values({ + id: "n1", + workflow_id: "c6-orphan", + name: "n1", + worker_type: "build", + status: "pending", + required: true, + depends_on: [], + seq: 1, + }).run().pipe(Effect.orDie) + // Boot A; recoverOrphanPending parks its first getNodes on the gate. + const initFiber = yield* initA.pipe(Effect.forkChild) + yield* pollWithTimeout( + Effect.sync(() => (parked.value ? true : undefined)), + "recoverOrphanPending never parked at getNodes", + ) + // While it is parked between getNodes and dag.fail, delete the + // session — FK cascade wipes the orphan rows out from under it. + yield* session.remove(SES_A as never) + expect(yield* store.getWorkflow("c6-orphan")).toBeUndefined() + // Release: dag.fail on the gone workflow fails, the startup-scan + // catchCause absorbs it, and Effect.ensuring frees the recovering + // slot. init must therefore complete — a leak or an unabsorbed + // defect would surface here. + release() + yield* Fiber.join(initFiber) + expect(yield* store.getWorkflow("c6-orphan")).toBeUndefined() + }), + ), + ) + }) }) From ab4cddb246651107ff5df5f5ae42b47daf31df33 Mon Sep 17 00:00:00 2001 From: Lex Date: Sat, 15 Aug 2026 02:36:45 +0800 Subject: [PATCH 8/8] test(dag): add H1 adopt-exactly-once probe for the WorkflowStarted latch (DAG-LOC-01) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Root cause of the coverage gap: the recovering-reservation latch added in 959bae7c2 had no mutation-falsifiable coverage — reverting it left all 930 tests green, so the REJECT review could not prove the latch matters. The probe parks the owner's live WorkflowStarted adoption at the getWorkflow/getNodes seam (parkGetNodes harness, now with a call counter) and publishes a reentrant WorkflowReplanned from the sibling directory's ambient context. Within one subscription duplicate WorkflowStarted events are serialized by Stream.runForEach, so the falsifiable duplicate-publish race is the replan's no-entry recoverWorkflow path on a separate subscription fiber — exactly the second adoption the latch repels. Assertions: exactly one adoption sequence at the seam while parked (single runtimes.set / single lease registration-to-be), one first-wave spawn, and a follow-up duplicate WorkflowStarted from the sibling directory driving no second adoption, re-spawn, or cancel. RED proof (scratch worktree, only 959bae7c2 reverted): the probe fails at 'expect(adoptionsAtTheSeam).toBe(1)' with 'Expected: 1, Received: 2' — the replan's recoverWorkflow parked a second gated getNodes inside reconcileWorkflow and double-adopted (spawn transition rejected, loser child cancelled). --- .../test/dag/dag-location-guards.test.ts | 145 ++++++++++++++++-- 1 file changed, 136 insertions(+), 9 deletions(-) diff --git a/packages/opencode/test/dag/dag-location-guards.test.ts b/packages/opencode/test/dag/dag-location-guards.test.ts index 915cbf8b70..8c35d57e58 100644 --- a/packages/opencode/test/dag/dag-location-guards.test.ts +++ b/packages/opencode/test/dag/dag-location-guards.test.ts @@ -131,13 +131,19 @@ interface TwoInstanceInput { readonly messagesBySession: Map /** Injected one-shot defects for DagStore.getWorkflow (parity with the template). */ readonly failGetWorkflow?: { remaining: number } - /** - * Optional deterministic park on DagStore.getNodes. When present, every - * getNodes call sets `parked.value = true` and then awaits `wait` before - * delegating to the real store — letting a probe interleave a mutation - * (e.g. Session.remove) inside a recovery sequence. - */ - readonly parkGetNodes?: { readonly wait: Promise; readonly parked: { value: boolean } } + /** + * Optional deterministic park on DagStore.getNodes. When present, every + * getNodes call sets `parked.value = true` and then awaits `wait` before + * delegating to the real store — letting a probe interleave a mutation + * (e.g. Session.remove) inside a recovery sequence. `calls` counts every + * gated getNodes invocation so a probe can tell HOW MANY distinct + * adoption/recovery sequences reached the seam (H1 adopt-exactly-once). + */ + readonly parkGetNodes?: { + readonly wait: Promise + readonly parked: { value: boolean } + readonly calls: { value: number } + } /** * Optional deterministic park on SessionPrompt.prepareIfIdle (the wake * delivery admission seam). When present, each prepareIfIdle call bumps @@ -179,6 +185,7 @@ function twoInstanceLayer(input: TwoInstanceInput) { Effect.suspend(() => { const gate = input.parkGetNodes if (!gate) return real.getNodes(id) + gate.calls.value++ gate.parked.value = true return Effect.promise(() => gate.wait).pipe(Effect.flatMap(() => real.getNodes(id))) }), @@ -345,7 +352,11 @@ function runTwoInstanceGuardTest( readonly sessionA?: string readonly sessionB?: string readonly failGetWorkflow?: { remaining: number } - readonly parkGetNodes?: { readonly wait: Promise; readonly parked: { value: boolean } } + readonly parkGetNodes?: { + readonly wait: Promise + readonly parked: { value: boolean } + readonly calls: { value: number } + } readonly parkWakeDelivery?: { readonly wait: Promise readonly parked: { value: boolean } @@ -1500,13 +1511,14 @@ describe("DAG-LOC-01 issue #238 evidence probes", () => { it("C6: recoverOrphanPending racing Session.remove is absorbed without killing init", async () => { const parked = { value: false } + const calls = { value: 0 } let release: () => void = () => {} const wait = new Promise((resolve) => { release = resolve }) await Effect.runPromise( runTwoInstanceGuardTest( - { parkGetNodes: { wait, parked } }, + { parkGetNodes: { wait, parked, calls } }, ({ database, session, store, initA }) => Effect.gen(function* () { // An all-pending orphan under A: a create that crashed mid-way. @@ -1554,3 +1566,118 @@ describe("DAG-LOC-01 issue #238 evidence probes", () => { ) }) }) + +// --------------------------------------------------------------------------- +// H1 mutation probe (DAG-LOC-01 REJECT follow-up, latch = 959bae7c2) +// +// The WorkflowStarted handler's recovering reservation is the only guard +// between its runtimes/recovering check and runtimes.set. Within one +// subscription duplicate WorkflowStarted events are serialized by +// Stream.runForEach, so the falsifiable race is a reentrant stimulus on a +// DIFFERENT subscription fiber: the WorkflowReplanned handler's no-entry +// path calls recoverWorkflow, whose own guard observes the live adoption's +// reservation (post-latch) or nothing (pre-latch). The probe parks the live +// adoption at the getWorkflow/getNodes seam, publishes the reentrant +// WorkflowReplanned from the SIBLING directory's ambient context, and +// asserts adopt-exactly-once: while parked, exactly ONE adoption sequence +// may sit at the seam (single runtimes.set-to-be, single automation-lease +// registration-to-be); after release, exactly one first-wave spawn; a +// follow-up duplicate WorkflowStarted from the sibling directory must not +// drive a second adoption either. Reverting 959bae7c2 turns this probe RED +// at the seam-count assertion (a second gated getNodes parks inside +// reconcileWorkflow — the second adoption that would overwrite the first +// runtimes entry and double-register the lease). +// --------------------------------------------------------------------------- + +describe("DAG-LOC-01 H1 adopt-exactly-once latch", () => { + it("H1: a reentrant sibling-directory publish cannot drive a second adoption of a parked live WorkflowStarted adoption", async () => { + const parked = { value: false } + const calls = { value: 0 } + let release: () => void = () => {} + const wait = new Promise((resolve) => { + release = resolve + }) + await Effect.runPromise( + runTwoInstanceGuardTest( + { parkGetNodes: { wait, parked, calls } }, + ({ dag, store, bridge, initA, initB, childPromptsA, childPromptsB, cancelsA }) => + Effect.gen(function* () { + yield* initA + const dagID = yield* dag.create({ + projectID: PROJECT_ID, + sessionID: SES_A, + title: "Duplicate publish race", + config: { name: "h1", nodes: [node()] }, + }) + // The owner's live WorkflowStarted adoption parks between its + // guard and runtimes.set: exactly one gated getNodes proves the + // owner's adoption — and nothing else — is at the seam. + yield* pollWithTimeout( + Effect.sync(() => (parked.value ? true : undefined)), + "live WorkflowStarted adoption never parked at getNodes", + ) + expect(calls.value).toBe(1) + // Reentrant stimulus published from the SIBLING directory's + // ambient context (B's InstanceRef stamps the location): a same- + // dagID WorkflowReplanned whose handler finds no runtimes entry + // takes the recoverWorkflow re-adoption path. The live adoption's + // reservation must repel it — pre-latch the replan passed + // recoverWorkflow's guard and parked a SECOND gated getNodes + // inside reconcileWorkflow. + yield* bridge.publish(DagEvent.WorkflowReplanned, { + dagID: dagID as never, + added: 0 as never, + removed: 0 as never, + replaced: 0 as never, + restarted: 0 as never, + timestamp: yield* DateTime.now, + }).pipe( + Effect.orDie, + Effect.provideService(InstanceRef, { + directory: DIR_B, + worktree: DIR_B, + project: { id: PROJECT_ID }, + } as never), + ) + // Negative window (sleep + snapshot — pollWithTimeout is a + // positive-wait tool): no second adoption may reach the seam. + yield* Effect.sleep("300 millis") + const adoptionsAtTheSeam = calls.value + // Release BEFORE asserting: the park awaits an uninterruptible + // Effect.promise, so a failing expectation must never abandon it + // (a hang at scope close would mask the RED). + release() + yield* Effect.sleep("100 millis") + expect(adoptionsAtTheSeam).toBe(1) + // Exactly one first-wave spawn for the single-node workflow. + const first = yield* takeWithin(childPromptsA, "owner did not adopt and spawn its first wave") + // Boot the sibling: its startup scan must not adopt the foreign + // (DIR_A-stamped) workflow either. + yield* initB + // Duplicate WorkflowStarted — same dagID, published from the + // sibling directory's context while the owner's entry is live — + // must not drive a second adoption or a re-spawn. + yield* bridge.publish(DagEvent.WorkflowStarted, { + dagID: dagID as never, + timestamp: yield* DateTime.now, + }).pipe( + Effect.orDie, + Effect.provideService(InstanceRef, { + directory: DIR_B, + worktree: DIR_B, + project: { id: PROJECT_ID }, + } as never), + ) + yield* Effect.sleep("300 millis") + expect(Option.isNone(yield* Queue.take(childPromptsA).pipe(Effect.timeoutOption("300 millis")))).toBe(true) + expect(Option.getOrElse(yield* Queue.poll(childPromptsB), () => null)).toBe(null) + expect(cancelsA).toEqual([]) + const row = yield* store.getNode(dagID, "n1") + expect(row?.status).toBe("running") + expect(row?.childSessionId).toBe(first.input.sessionID as string) + expect((yield* store.getWorkflow(dagID))?.status).toBe("running") + }), + ), + ) + }) +})