From a5a01c82b4537800c75cc2f43475247d85dd3f98 Mon Sep 17 00:00:00 2001 From: lex Date: Tue, 28 Jul 2026 21:30:55 +0800 Subject: [PATCH 1/3] fix(dag): guard cross-instance adoption, handler defects, cancel-skip race - Adopt workflows only in the owning project's instance (recoverWorkflow, WorkflowStarted, wake redelivery) so multi-directory servers no longer spawn children under a foreign directory context - Replace handler-level Effect.ignore with a catchCause boundary so orDie defects from transient DB errors cannot kill event subscriptions for the process lifetime - Abort still-live fibers (abortChild + interrupt) when NodeSkipped lands on a running node, closing the cancel-race child session leak - Seed global dag.jsonc once at instance init via opt-in autoSeed; scheduling rounds stay pure reads and non-EEXIST seed failures are logged - Align the node PAUSED transitions with production paths and weld projector from-guards to the declared tables via exported projections + drift test --- packages/core/src/dag/core/types.ts | 9 +- packages/core/src/dag/projector.ts | 75 +++- packages/core/test/dag-core.test.ts | 2 +- .../core/test/dag-projector-drift.test.ts | 69 ++++ packages/opencode/src/dag/config.ts | 35 +- packages/opencode/src/dag/runtime/loop.ts | 74 +++- packages/opencode/test/dag/dag-config.test.ts | 18 +- .../opencode/test/dag/dag-loop-guards.test.ts | 371 ++++++++++++++++++ 8 files changed, 612 insertions(+), 41 deletions(-) create mode 100644 packages/core/test/dag-projector-drift.test.ts create mode 100644 packages/opencode/test/dag/dag-loop-guards.test.ts diff --git a/packages/core/src/dag/core/types.ts b/packages/core/src/dag/core/types.ts index bd5d9c758..9212ca014 100644 --- a/packages/core/src/dag/core/types.ts +++ b/packages/core/src/dag/core/types.ts @@ -149,9 +149,14 @@ export function isNodeTerminalStatus(status: NodeStatus): boolean { * callers MUST treat attempts to transition further as TerminalViolationError. * * The `restart` path (running -> paused -> pending -> running) is encoded here: - * PAUSED returns [RUNNING] and RUNNING returns [PENDING]. All terminal states + * PAUSED returns [RUNNING, ...] and RUNNING returns [PENDING]. All terminal states * (COMPLETED/FAILED/ABORTED/SKIPPED) return [] — restarts never originate from * a terminal node; the projector's NodeRestarted resets the row to PENDING. + * + * PAUSED additionally admits SKIPPED and FAILED: workflow-level cancel/fail + * terminalizes paused nodes via NodeSkipped, and a replan NodeCancelled + * projects them to failed — the table documents what the production paths + * (and the projector's from-guards) actually do. */ export function getValidNextNodeStatuses(currentStatus: NodeStatus): NodeStatus[] { switch (currentStatus) { @@ -165,7 +170,7 @@ export function getValidNextNodeStatuses(currentStatus: NodeStatus): NodeStatus[ case NodeStatus.RUNNING: return [NodeStatus.COMPLETED, NodeStatus.FAILED, NodeStatus.PAUSED, NodeStatus.PENDING, NodeStatus.SKIPPED] case NodeStatus.PAUSED: - return [NodeStatus.RUNNING] + return [NodeStatus.RUNNING, NodeStatus.SKIPPED, NodeStatus.FAILED] case NodeStatus.COMPLETED: case NodeStatus.FAILED: case NodeStatus.ABORTED: diff --git a/packages/core/src/dag/projector.ts b/packages/core/src/dag/projector.ts index 7762e6772..239475e71 100644 --- a/packages/core/src/dag/projector.ts +++ b/packages/core/src/dag/projector.ts @@ -15,6 +15,44 @@ const toMillis = (dt: DateTime.Utc) => DateTime.toEpochMillis(dt) /** Cast a status string to the WorkflowStatus literal type for Drizzle. */ const ws = (s: WorkflowStatus) => s as DagEvent.WorkflowStatus +/** + * Event → status projection guards — the single source the drift test checks + * against the declared transition tables in core/types.ts. Every `from` entry + * must be a legal predecessor of `to` (self-transitions are idempotent + * re-application and exempt). Editing a guard here without updating the + * declared table (or vice versa) fails the consistency test in + * test/dag-projector-drift.test.ts instead of drifting silently. + */ +export const NodeStatusProjection = { + queued: { to: "queued", from: ["pending", "queued"] }, + started: { to: "running", from: ["pending", "queued", "paused", "running"] }, + // Publisher-side guardNode admits NodeCompleted only from running; the + // former pending/queued/paused tolerance had no producing path and + // contradicted the declared table. + completed: { to: "completed", from: ["running"] }, + failed: { to: "failed", from: ["running", "pending", "queued"] }, + skipped: { to: "skipped", from: ["pending", "queued", "running", "paused"] }, + cancelled: { to: "failed", from: ["pending", "queued", "running", "paused"] }, + restarted: { to: "pending", from: ["running"] }, +} as const + +export const WorkflowStatusProjection = { + started: { to: "running", from: ["pending"] }, + paused: { to: "paused", from: ["running", "stepping"] }, + resumed: { to: "running", from: ["paused", "stepping"] }, + stepped: { to: "stepping", from: ["running", "stepping"] }, + completed: { to: "completed", from: ["running", "stepping"] }, + failed: { to: "failed", from: ["running", "stepping"] }, + cancelled: { to: "cancelled", from: ["running", "paused", "stepping"] }, + /** + * Additive-extend reopen — the ONLY sanctioned exception to terminal + * irreversibility: a naturally-completed workflow with a reporting leaf + * checkpoint may be reopened by _extend (see dag.ts reopenCompleted). + * The drift test exempts exactly this entry. + */ + replanReopen: { to: "running", from: ["completed"] }, +} as const + /** * DAG projector: EventV2 → read-model tables (CQRS). * @@ -67,14 +105,14 @@ export const layer = Layer.effectDiscard( started_at: toMillis(event.data.timestamp), time_updated: toMillis(event.data.timestamp), }) - .where(and(eq(WorkflowTable.id, event.data.dagID), eq(WorkflowTable.status, "pending"))) + .where(and(eq(WorkflowTable.id, event.data.dagID), inArray(WorkflowTable.status, [...WorkflowStatusProjection.started.from]))) .run() .pipe(Effect.orDie), ) - yield* events.project(DagEvent.WorkflowPaused, setWorkflowStatus(ws("paused"), [ws("running"), ws("stepping")])) - yield* events.project(DagEvent.WorkflowResumed, setWorkflowStatus(ws("running"), [ws("paused"), ws("stepping")])) - yield* events.project(DagEvent.WorkflowStepped, setWorkflowStatus(ws("stepping"), [ws("running"), ws("stepping")])) + yield* events.project(DagEvent.WorkflowPaused, setWorkflowStatus(ws("paused"), [...WorkflowStatusProjection.paused.from])) + yield* events.project(DagEvent.WorkflowResumed, setWorkflowStatus(ws("running"), [...WorkflowStatusProjection.resumed.from])) + yield* events.project(DagEvent.WorkflowStepped, setWorkflowStatus(ws("stepping"), [...WorkflowStatusProjection.stepped.from])) const setWorkflowTerminal = (status: WorkflowStatus, from: WorkflowStatus[]) => (event: { data: { dagID: DagEvent.DagID; timestamp: DateTime.Utc }; durable?: { seq: number } }) => db @@ -84,9 +122,9 @@ export const layer = Layer.effectDiscard( .run() .pipe(Effect.orDie) - yield* events.project(DagEvent.WorkflowCompleted, setWorkflowTerminal(ws("completed"), [ws("running"), ws("stepping")])) - yield* events.project(DagEvent.WorkflowFailed, setWorkflowTerminal(ws("failed"), [ws("running"), ws("stepping")])) - yield* events.project(DagEvent.WorkflowCancelled, setWorkflowTerminal(ws("cancelled"), [ws("running"), ws("paused"), ws("stepping")])) + yield* events.project(DagEvent.WorkflowCompleted, setWorkflowTerminal(ws("completed"), [...WorkflowStatusProjection.completed.from])) + yield* events.project(DagEvent.WorkflowFailed, setWorkflowTerminal(ws("failed"), [...WorkflowStatusProjection.failed.from])) + yield* events.project(DagEvent.WorkflowCancelled, setWorkflowTerminal(ws("cancelled"), [...WorkflowStatusProjection.cancelled.from])) yield* events.project(DagEvent.WorkflowReplanned, (event) => Effect.gen(function* () { @@ -104,7 +142,7 @@ export const layer = Layer.effectDiscard( }) .where(and( eq(WorkflowTable.id, event.data.dagID), - eq(WorkflowTable.status, "completed"), + inArray(WorkflowTable.status, [...WorkflowStatusProjection.replanReopen.from]), )) .run() .pipe(Effect.orDie) @@ -180,7 +218,7 @@ export const layer = Layer.effectDiscard( .where(and( eq(WorkflowNodeTable.workflow_id, event.data.dagID), eq(WorkflowNodeTable.id, event.data.nodeID), - inArray(WorkflowNodeTable.status, ["pending", "queued"]), + inArray(WorkflowNodeTable.status, [...NodeStatusProjection.queued.from]), )) .run() .pipe(Effect.orDie), @@ -209,7 +247,7 @@ export const layer = Layer.effectDiscard( .where(and( eq(WorkflowNodeTable.workflow_id, event.data.dagID), eq(WorkflowNodeTable.id, event.data.nodeID), - inArray(WorkflowNodeTable.status, ["pending", "queued", "paused", "running"]), + inArray(WorkflowNodeTable.status, [...NodeStatusProjection.started.from]), )) .run() .pipe(Effect.orDie), @@ -225,13 +263,14 @@ export const layer = Layer.effectDiscard( seq: event.durable!.seq, time_updated: toMillis(event.data.timestamp), }) - // Only complete nodes in non-terminal status. Prevents a stale - // NodeCompleted from flipping an already-terminal node (e.g. failed by - // a replan-ceiling check) back to completed. + // Only complete nodes the publisher could legally complete: guardNode + // admits NodeCompleted from running only, and a stale NodeCompleted + // must not flip an already-terminal node (e.g. failed by a + // replan-ceiling check) back to completed. .where(and( eq(WorkflowNodeTable.workflow_id, event.data.dagID), eq(WorkflowNodeTable.id, event.data.nodeID), - inArray(WorkflowNodeTable.status, ["pending", "queued", "running", "paused"]), + inArray(WorkflowNodeTable.status, [...NodeStatusProjection.completed.from]), )) .run() .pipe(Effect.orDie), @@ -253,7 +292,7 @@ export const layer = Layer.effectDiscard( .where(and( eq(WorkflowNodeTable.workflow_id, event.data.dagID), eq(WorkflowNodeTable.id, event.data.nodeID), - inArray(WorkflowNodeTable.status, ["running", "pending", "queued"]), + inArray(WorkflowNodeTable.status, [...NodeStatusProjection.failed.from]), )) .run() .pipe(Effect.orDie), @@ -271,7 +310,7 @@ export const layer = Layer.effectDiscard( .where(and( eq(WorkflowNodeTable.workflow_id, event.data.dagID), eq(WorkflowNodeTable.id, event.data.nodeID), - inArray(WorkflowNodeTable.status, ["pending", "queued", "running", "paused"]), + inArray(WorkflowNodeTable.status, [...NodeStatusProjection.skipped.from]), )) .run() .pipe(Effect.orDie), @@ -284,7 +323,7 @@ export const layer = Layer.effectDiscard( .where(and( eq(WorkflowNodeTable.workflow_id, event.data.dagID), eq(WorkflowNodeTable.id, event.data.nodeID), - inArray(WorkflowNodeTable.status, ["pending", "queued", "running", "paused"]), + inArray(WorkflowNodeTable.status, [...NodeStatusProjection.cancelled.from]), )) .run() .pipe(Effect.orDie), @@ -308,7 +347,7 @@ export const layer = Layer.effectDiscard( .where(and( eq(WorkflowNodeTable.workflow_id, event.data.dagID), eq(WorkflowNodeTable.id, event.data.nodeID), - eq(WorkflowNodeTable.status, "running"), + inArray(WorkflowNodeTable.status, [...NodeStatusProjection.restarted.from]), )) .run() .pipe(Effect.orDie), diff --git a/packages/core/test/dag-core.test.ts b/packages/core/test/dag-core.test.ts index 6cb1eda33..f8b11a55e 100644 --- a/packages/core/test/dag-core.test.ts +++ b/packages/core/test/dag-core.test.ts @@ -170,7 +170,7 @@ describe("iron laws (transition tables)", () => { // →FAILED queue-wait timeout. [NodeStatus.QUEUED, [NodeStatus.QUEUED, NodeStatus.RUNNING, NodeStatus.PENDING, NodeStatus.SKIPPED, NodeStatus.FAILED]], [NodeStatus.RUNNING, [NodeStatus.COMPLETED, NodeStatus.FAILED, NodeStatus.PAUSED, NodeStatus.PENDING, NodeStatus.SKIPPED]], - [NodeStatus.PAUSED, [NodeStatus.RUNNING]], + [NodeStatus.PAUSED, [NodeStatus.RUNNING, NodeStatus.SKIPPED, NodeStatus.FAILED]], [NodeStatus.COMPLETED, []], [NodeStatus.FAILED, []], [NodeStatus.ABORTED, []], diff --git a/packages/core/test/dag-projector-drift.test.ts b/packages/core/test/dag-projector-drift.test.ts new file mode 100644 index 000000000..5accf890c --- /dev/null +++ b/packages/core/test/dag-projector-drift.test.ts @@ -0,0 +1,69 @@ +/** + * Drift guard: the projector's event→status from-guards must stay a subset of + * the declared iron-law transition tables in core/types.ts. + * + * The tables (getValidNextNodeStatuses / getValidNextWorkflowStatuses) are the + * declared state machine enforced by publisher-side guards; the projector's + * from-lists are a second encoding used as SQL race protection. Historically + * the two drifted apart silently (e.g. the projector accepted + * pending→completed which the table forbids). This test welds them together: + * any edit to one side without the other fails here. + * + * Exemptions: + * - self-transitions (from === to): idempotent re-application of a replayed + * event, not a state change. + * - WorkflowStatusProjection.replanReopen (completed→running): the single + * sanctioned exception to terminal irreversibility — additive extend may + * reopen a naturally-completed workflow (see dag.ts reopenCompleted). + */ +import { describe, expect, it } from "bun:test" +import { NodeStatusProjection, WorkflowStatusProjection } from "../src/dag/projector" +import { + getValidNextNodeStatuses, + getValidNextWorkflowStatuses, + type NodeStatus, + type WorkflowStatus, +} from "../src/dag/core/types" + +describe("projector from-guards vs declared transition tables", () => { + it("every node projection guard is a declared transition (or a self-transition)", () => { + const violations: string[] = [] + for (const [event, projection] of Object.entries(NodeStatusProjection)) { + for (const from of projection.from) { + if (from === projection.to) continue + const allowed = getValidNextNodeStatuses(from as NodeStatus) + if (!allowed.includes(projection.to as NodeStatus)) { + violations.push(`${event}: ${from} -> ${projection.to} is not in the declared node table`) + } + } + } + expect(violations).toEqual([]) + }) + + it("every workflow projection guard is a declared transition (or the sanctioned reopen)", () => { + const violations: string[] = [] + for (const [event, projection] of Object.entries(WorkflowStatusProjection)) { + // The documented terminal-irreversibility exception: additive-extend + // reopen of a completed workflow. Exempt exactly this entry. + if (event === "replanReopen") continue + for (const from of projection.from) { + if (from === projection.to) continue + const allowed = getValidNextWorkflowStatuses(from as WorkflowStatus) + if (!allowed.includes(projection.to as WorkflowStatus)) { + violations.push(`${event}: ${from} -> ${projection.to} is not in the declared workflow table`) + } + } + } + expect(violations).toEqual([]) + }) + + it("keeps the reopen exemption scoped to completed→running only", () => { + expect(WorkflowStatusProjection.replanReopen.to).toBe("running") + expect([...WorkflowStatusProjection.replanReopen.from]).toEqual(["completed"]) + }) +}) + +// Note: core/dag/core/transitions.ts (transitionToNodeEvent & friends) is a +// third encoding of the same machine with zero production callers — a +// capability reservoir kept for the event-semantics mapping. It is exercised +// by dag-core.test.ts only and intentionally not welded here. diff --git a/packages/opencode/src/dag/config.ts b/packages/opencode/src/dag/config.ts index fb21a14a8..717f2605e 100644 --- a/packages/opencode/src/dag/config.ts +++ b/packages/opencode/src/dag/config.ts @@ -10,7 +10,9 @@ * * When neither exists, a commented default `dag.jsonc` is seeded into the * global opencode config directory (same generation pattern as the main - * config's loadGlobal). Files are read lazily at spawn-scheduling time — like + * config's loadGlobal) — but only when the caller opts in via `autoSeed`; + * the spawn-scheduling read path never writes. Files are read lazily at + * spawn-scheduling time — like * templates/resolve.ts, nothing is loaded at startup, and edits take effect on * the next scheduling round. */ @@ -55,11 +57,22 @@ const DEFAULT_CONTENT = `{ } ` -export function load(projectDir: string): Effect.Effect { +export function load(projectDir: string, options: { autoSeed?: boolean } = {}): Effect.Effect { return Effect.gen(function* () { const found = yield* Effect.promise(() => readFirst(candidates(projectDir))) if (!found) { - yield* Effect.promise(() => seedGlobalDefault()).pipe(Effect.ignore) + // Seeding writes to the user's global config dir — opt-in so read paths + // (spawn scheduling) stay side-effect free. EEXIST is the benign + // seed race; anything else (EACCES/EROFS/…) is logged, never thrown. + if (options.autoSeed) { + const seedError = yield* Effect.promise(() => seedGlobalDefault()) + if (seedError) { + yield* Effect.logWarning("failed to seed global dag.jsonc", { + dir: globalConfigDir(), + code: seedError.code ?? String(seedError), + }) + } + } return {} } const decoded = Schema.decodeUnknownOption(Info)(parseJsonc(found.text, [], { allowTrailingComma: true }) ?? {}) @@ -111,11 +124,19 @@ async function readFirst(paths: string[]) { return undefined } -async function seedGlobalDefault() { +async function seedGlobalDefault(): Promise { const file = path.join(globalConfigDir(), "dag.jsonc") - await fs.mkdir(path.dirname(file), { recursive: true }) - // wx: never clobber a file that appeared between the read and the seed. - await fs.writeFile(file, DEFAULT_CONTENT, { flag: "wx" }) + try { + await fs.mkdir(path.dirname(file), { recursive: true }) + // wx: never clobber a file that appeared between the read and the seed. + await fs.writeFile(file, DEFAULT_CONTENT, { flag: "wx" }) + return undefined + } catch (err) { + const error = err as NodeJS.ErrnoException + // EEXIST is the expected outcome of the read-then-seed race. + if (error.code === "EEXIST") return undefined + return error + } } function parseModelRef(ref: string | undefined) { diff --git a/packages/opencode/src/dag/runtime/loop.ts b/packages/opencode/src/dag/runtime/loop.ts index 9f66f6c84..6852073ed 100644 --- a/packages/opencode/src/dag/runtime/loop.ts +++ b/packages/opencode/src/dag/runtime/loop.ts @@ -61,6 +61,11 @@ export const layer = Layer.effect( const wakeInFlight = new Set() const wakePending = new Set() + // Seed the commented global dag.jsonc once per instance init — the + // per-round DagConfig.load below stays a pure read so the spawn + // scheduling hot path never writes to the user's config dir. + yield* DagConfig.load(ctx.directory, { autoSeed: true }).pipe(Effect.ignore) + const spawnReady = Effect.fn("DagLoop.spawnReady")(function* (dagID: string) { const entry = runtimes.get(dagID) if (!entry) return @@ -264,7 +269,20 @@ export const layer = Layer.effect( yield* promptSvc.cancel(childSessionId as never).pipe(Effect.ignore) }) + // Subscription handlers must never die. dag.ts guards and checkCompletion + // use orDie, whose defects punch straight through Effect.ignore (it only + // absorbs the error channel) and would kill the forked runForEach fiber — + // leaving that event type permanently unhandled for the rest of the + // process. catchCause absorbs failures AND defects at the boundary. + const guarded = (event: string) => (self: Effect.Effect) => + self.pipe(Effect.catchCause((cause) => Effect.logWarning("DagLoop handler failed", { event, cause }))) + 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 const dagID = wf.id const config = parseWorkflowConfig(wf.config) const recovery = yield* reconcileWorkflow( @@ -337,6 +355,10 @@ export const layer = Layer.effect( if (runtimes.has(dagID)) return const wf = yield* store.getWorkflow(dagID).pipe(Effect.orDie) if (!wf) return + // Cross-instance guard: only the owning project's instance adopts + // (see recoverWorkflow). First-wave spawns must not race across + // directory contexts. + if (wf.projectId !== ctx.project.id) 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) @@ -350,7 +372,7 @@ export const layer = Layer.effect( yield* checkCompletion(dagID) }), ) - }).pipe(Effect.ignore), + }).pipe(guarded("WorkflowStarted")), ), Effect.forkScoped({ startImmediately: true }), ) @@ -371,7 +393,24 @@ export const layer = Layer.effect( if (!entry) return yield* entry.evalLock.withPermits(1)( Effect.gen(function* () { - entry.fibers.delete(evt.data.nodeID as string) + const nodeID = evt.data.nodeID as string + // Cancel-skip race: workflow-level cancel publishes NodeSkipped + // for running nodes, and this handler may win the cross-stream + // race against WorkflowCancelled. Deleting the fiber here + // uninterrupted would orphan it from the WorkflowCancelled + // sweep and the child session would keep running until its + // prompt finishes or times out. Stop it now, mirroring the + // NodeCancelled handler. Completed nodes keep the plain + // delete — their fiber published the event and is finishing. + if (def === DagEvent.NodeSkipped) { + const fiber = entry.fibers.get(nodeID) + if (fiber) { + const node = yield* store.getNode(dagID, nodeID) + yield* abortChild(nodeID, node?.childSessionId ?? null).pipe(Effect.ignore) + yield* Fiber.interrupt(fiber).pipe(Effect.ignore) + } + } + entry.fibers.delete(nodeID) const workflow = yield* store.getWorkflow(dagID) entry.runtime.setPaused(workflow?.status === "paused") entry.runtime.setStepMode(workflow?.status === "stepping") @@ -392,7 +431,7 @@ export const layer = Layer.effect( // the parent session may already be idle (no new idle event // will fire), so we can't rely on the idle subscription alone. yield* tryDeliverWake(entry.parentSessionID).pipe(Effect.ignore, Effect.forkScoped) - }).pipe(Effect.ignore), + }).pipe(guarded(def === DagEvent.NodeSkipped ? "NodeSkipped" : "NodeCompleted")), ), Effect.forkScoped({ startImmediately: true }), ) @@ -419,7 +458,7 @@ export const layer = Layer.effect( yield* checkCompletion(dagID) }), ) - }).pipe(Effect.ignore), + }).pipe(guarded("NodeCancelled")), ), Effect.forkScoped({ startImmediately: true }), ) @@ -454,7 +493,7 @@ export const layer = Layer.effect( }), ) yield* tryDeliverWake(entry.parentSessionID).pipe(Effect.ignore, Effect.forkScoped) - }).pipe(Effect.ignore), + }).pipe(guarded("NodeFailed")), ), Effect.forkScoped({ startImmediately: true }), ) @@ -466,7 +505,7 @@ export const layer = Layer.effect( const entry = runtimes.get(evt.data.dagID as string) if (!entry) return yield* entry.evalLock.withPermits(1)(Effect.sync(() => entry.runtime.setPaused(true))) - }).pipe(Effect.ignore), + }).pipe(guarded("WorkflowPaused")), ), Effect.forkScoped({ startImmediately: true }), ) @@ -484,7 +523,7 @@ export const layer = Layer.effect( yield* spawnReady(dagID) }), ) - }).pipe(Effect.ignore), + }).pipe(guarded("WorkflowStepped")), ), Effect.forkScoped({ startImmediately: true }), ) @@ -508,7 +547,7 @@ export const layer = Layer.effect( yield* checkCompletion(dagID) }), ) - }).pipe(Effect.ignore), + }).pipe(guarded("WorkflowResumed")), ), Effect.forkScoped({ startImmediately: true }), ) @@ -533,7 +572,7 @@ export const layer = Layer.effect( yield* checkCompletion(dagID) }), ) - }).pipe(Effect.ignore), + }).pipe(guarded("WorkflowReplanned")), ), Effect.forkScoped({ startImmediately: true }), ) @@ -564,7 +603,7 @@ export const layer = Layer.effect( if (parentSessionID) { yield* tryDeliverWake(parentSessionID).pipe(Effect.ignore, Effect.forkScoped) } - }).pipe(Effect.ignore), + }).pipe(guarded("WorkflowTerminal")), ), Effect.forkScoped({ startImmediately: true }), ) @@ -744,11 +783,14 @@ export const layer = Layer.effect( } }) - // Idle-event subscription: the primary wake trigger + // Idle-event subscription: the primary wake trigger. The handler only + // forks, so the subscription itself cannot die — guarded here so a + // defect inside a delivery attempt is logged instead of silently + // killing its fiber. yield* events.subscribe(SessionStatusEvent.Status).pipe( Stream.filter((evt) => evt.data.status.type === "idle"), Stream.runForEach((evt) => - tryDeliverWake(evt.data.sessionID as string).pipe(Effect.ignore, Effect.forkScoped), + tryDeliverWake(evt.data.sessionID as string).pipe(guarded("WakeDelivery"), Effect.forkScoped), ), Effect.forkScoped({ startImmediately: true }), ) @@ -773,6 +815,14 @@ export const layer = Layer.effect( Effect.catch(() => Effect.succeed([] as string[])), ) 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. + const snapshot = yield* store.getWakeSnapshot(sessionID).pipe( + Effect.catch(() => Effect.succeed({ nodes: [], workflows: [] } satisfies DagStore.WakeSnapshot)), + ) + if (!snapshot.workflows.some((wf) => wf.projectId === ctx.project.id)) continue yield* tryDeliverWake(sessionID).pipe(Effect.forkScoped) } diff --git a/packages/opencode/test/dag/dag-config.test.ts b/packages/opencode/test/dag/dag-config.test.ts index de9b3fbba..fa7812a0e 100644 --- a/packages/opencode/test/dag/dag-config.test.ts +++ b/packages/opencode/test/dag/dag-config.test.ts @@ -27,9 +27,16 @@ afterEach(async () => { }) describe("DagConfig.load", () => { - it("seeds a commented default dag.jsonc into the global config dir when none exists", async () => { + it("does not write anything by default when no config exists (pure read path)", async () => { const info = await Effect.runPromise(DagConfig.load(projectDir)) expect(info).toEqual({}) + // The scheduling hot path must never seed — the global dir stays absent. + expect(fs.access(path.join(globalDir, "dag.jsonc"))).rejects.toThrow() + }) + + it("seeds a commented default dag.jsonc into the global config dir with autoSeed", async () => { + const info = await Effect.runPromise(DagConfig.load(projectDir, { autoSeed: true })) + expect(info).toEqual({}) const seeded = await fs.readFile(path.join(globalDir, "dag.jsonc"), "utf-8") expect(seeded).toContain("thinking_depth") expect(seeded).toContain("advanced") @@ -41,6 +48,15 @@ describe("DagConfig.load", () => { expect(DagConfig.tierModel(reloaded, { required: true, workerType: "build" })).toBeUndefined() }) + it("returns empty config when seeding fails for non-EEXIST reasons", async () => { + // Point the global dir AT A FILE so mkdir/writeFile fail with ENOTDIR — + // load must swallow it (warning only) and still return {}. + await fs.writeFile(path.join(dir, "not-a-dir"), "") + process.env.OPENCODE_CONFIG_DIR = path.join(dir, "not-a-dir") + const info = await Effect.runPromise(DagConfig.load(projectDir, { autoSeed: true })) + expect(info).toEqual({}) + }) + it("reads the global dag.jsonc with comments and trailing commas", async () => { await fs.mkdir(globalDir, { recursive: true }) await fs.writeFile( diff --git a/packages/opencode/test/dag/dag-loop-guards.test.ts b/packages/opencode/test/dag/dag-loop-guards.test.ts new file mode 100644 index 000000000..7b461f74a --- /dev/null +++ b/packages/opencode/test/dag/dag-loop-guards.test.ts @@ -0,0 +1,371 @@ +/** + * DagLoop guard regressions: + * + * 1. Cross-instance adoption (P0): DagLoop is per-directory InstanceState but + * the event bus and store are process-global. Only the instance whose + * project owns a workflow may adopt it — at WorkflowStarted AND at startup + * recovery. Regression: a foreign instance won the first-wave spawn race + * and children ran under the wrong directory context. + * + * 2. Subscription survival (P1): orDie defects punch through Effect.ignore + * (it only absorbs the error channel) and killed the forked runForEach + * fiber, leaving that event type permanently unhandled. Handlers are now + * guarded with catchCause. + * + * 3. Cancel-skip race (P1): workflow-level cancel publishes NodeSkipped for + * running nodes; when that handler won the cross-stream race against + * WorkflowCancelled it deleted the fiber uninterrupted and the child + * session kept running. The NodeSkipped handler now aborts live fibers. + */ +import { describe, expect, it } from "bun:test" +import { Deferred, Effect, Layer, Option, Queue } from "effect" +import type { SessionV1 } from "@opencode-ai/core/v1/session" +import { Database } from "@opencode-ai/core/database/database" +import { DagProjector } from "@opencode-ai/core/dag/projector" +import { WorkflowNodeTable, 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 { 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" + +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 +} + +function guardLayer(input: { + readonly childPrompts: Queue.Queue + readonly cancels: string[] + /** Injected one-shot defects for DagStore.getWorkflow (P1 survival test). */ + readonly failGetWorkflow?: { remaining: number } +}) { + 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: () => Effect.succeed([]), + }) + const deliver = Effect.fn("test.SessionPrompt.deliver")(function* (value: SessionPrompt.PromptInput) { + const sessionID = value.sessionID as string + const release = yield* Deferred.make() + yield* Queue.offer(input.childPrompts, { + title: childTitles.get(sessionID) ?? sessionID, + input: value, + release, + }) + return reply(sessionID, yield* Deferred.await(release)) + }) + const prompt = Layer.mock(SessionPrompt.Service, { + cancel: (sessionID) => + Effect.sync(() => { + input.cancels.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), + ) + return Layer.merge(base, loop) +} + +function runGuardTest( + options: { + /** Project the current instance belongs to. */ + readonly instanceProject: string + readonly failGetWorkflow?: { remaining: number } + }, + test: (services: { + readonly dag: Dag.Interface + readonly loop: DagLoop.Interface + readonly store: DagStore.Interface + readonly childPrompts: Queue.Queue + readonly cancels: string[] + }) => Effect.Effect, + beforeInit?: (services: { readonly database: Database.Interface }) => Effect.Effect, +) { + return Effect.gen(function* () { + const childPrompts = yield* Queue.unbounded() + const cancels: string[] = [] + 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 + // Two projects sharing the process-global store, one session in each. + for (const project of ["project-1", "project-2"]) { + yield* database.db.insert(ProjectTable).values({ + id: project as never, + worktree: process.cwd() as never, + sandboxes: [], + }).run().pipe(Effect.orDie) + yield* database.db.insert(SessionTable).values({ + id: `ses_${project}` as never, + project_id: project as never, + slug: project, + directory: process.cwd() as never, + title: `Parent of ${project}`, + version: "test", + }).run().pipe(Effect.orDie) + } + if (beforeInit) yield* beforeInit({ database }) + yield* loop.init() + return yield* test({ dag, loop, store, childPrompts, cancels }) + }).pipe( + Effect.provide(guardLayer({ childPrompts, cancels, failGetWorkflow: options.failGetWorkflow })), + Effect.provideService(InstanceRef, { + directory: process.cwd(), + worktree: process.cwd(), + project: { id: options.instanceProject }, + } as never), + Effect.scoped, + ) + }) +} + +describe("DagLoop cross-instance adoption guard", () => { + it("does not adopt a foreign project's workflow at WorkflowStarted", async () => { + await Effect.runPromise( + runGuardTest({ instanceProject: "project-2" }, ({ dag, store, childPrompts }) => + Effect.gen(function* () { + const dagID = yield* dag.create({ + projectID: "project-1", + sessionID: "ses_project-1", + title: "Foreign workflow", + config: { name: "foreign", nodes: [node({ id: "foreign-node", name: "foreign-node" })] }, + }) + // Give the (wrongly subscribed) handler time to act, then assert + // nothing spawned and the durable node is untouched. + yield* Effect.sleep("300 millis") + expect(Option.isNone(yield* Queue.poll(childPrompts))).toBe(true) + const nodes = yield* store.getNodes(dagID) + expect(nodes).toHaveLength(1) + expect(nodes[0].status).toBe("pending") + }), + ), + ) + }) + + it("adopts and spawns its own project's workflow (control)", async () => { + await Effect.runPromise( + runGuardTest({ instanceProject: "project-2" }, ({ dag, childPrompts }) => + Effect.gen(function* () { + yield* dag.create({ + projectID: "project-2", + sessionID: "ses_project-2", + title: "Own workflow", + config: { name: "own", nodes: [node({ id: "own-node", name: "own-node" })] }, + }) + const child = yield* takeWithin(childPrompts, "own-project node did not start") + expect(child.title).toBe("own-node") + yield* Deferred.succeed(child.release, "done") + }), + ), + ) + }) + + it("does not reconcile a foreign project's running workflow at startup recovery", async () => { + await Effect.runPromise( + runGuardTest( + { instanceProject: "project-2" }, + ({ store, cancels }) => + Effect.gen(function* () { + yield* Effect.sleep("300 millis") + // Without the guard, recovery invents an ownership-lost failure + // and cancels the child session. The foreign row must stay put. + const nodes = yield* store.getNodes("foreign-recovery") + expect(nodes[0].status).toBe("running") + expect(cancels).toHaveLength(0) + }), + ({ database }) => + database.db.transaction((tx) => + Effect.gen(function* () { + yield* tx.insert(WorkflowTable).values({ + id: "foreign-recovery", + project_id: "project-1" as never, + session_id: "ses_project-1" as never, + title: "Foreign running workflow", + status: "running", + config: "{}", + seq: 10, + }).run() + yield* tx.insert(WorkflowNodeTable).values({ + id: "orphan-node", + workflow_id: "foreign-recovery", + name: "orphan-node", + worker_type: "build", + status: "running", + required: true, + depends_on: [], + child_session_id: "ses_orphan", + seq: 9, + }).run() + }), + ).pipe(Effect.orDie), + ), + ) + }) +}) + +describe("DagLoop subscription survival", () => { + it("keeps processing WorkflowStarted after a handler defect", async () => { + await Effect.runPromise( + runGuardTest( + { instanceProject: "project-1", failGetWorkflow: { remaining: 1 } }, + ({ dag, childPrompts }) => + Effect.gen(function* () { + // First workflow: the handler's getWorkflow dies (injected defect). + // The guarded boundary must log-and-survive, not kill the stream. + yield* dag.create({ + projectID: "project-1", + sessionID: "ses_project-1", + title: "Poisoned workflow", + config: { name: "poisoned", nodes: [node({ id: "poisoned-node", name: "poisoned-node" })] }, + }) + yield* Effect.sleep("200 millis") + // Second workflow on the same subscription must still spawn. + yield* dag.create({ + projectID: "project-1", + sessionID: "ses_project-1", + title: "Healthy workflow", + config: { name: "healthy", nodes: [node({ id: "healthy-node", name: "healthy-node" })] }, + }) + const child = yield* takeWithin(childPrompts, "subscription died after defect — healthy workflow never spawned") + expect(child.title).toBe("healthy-node") + yield* Deferred.succeed(child.release, "done") + }), + ), + ) + }) +}) + +describe("DagLoop cancel-skip race", () => { + it("aborts the live child session when NodeSkipped lands on a running node", async () => { + await Effect.runPromise( + runGuardTest({ instanceProject: "project-1" }, ({ dag, store, childPrompts, cancels }) => + Effect.gen(function* () { + const dagID = yield* dag.create({ + projectID: "project-1", + sessionID: "ses_project-1", + title: "Skip race", + config: { name: "skip-race", nodes: [node({ id: "racy-node", name: "racy-node" })] }, + }) + // Node is running: its prompt is parked on the deferred gate. + const child = yield* takeWithin(childPrompts, "racy-node did not start") + const childSessionID = child.input.sessionID as string + // Simulate the workflow-cancel skip arriving before the + // WorkflowCancelled sweep: publish NodeSkipped directly. + yield* dag.nodeSkipped(dagID, "racy-node", "workflow_cancelled") + // The handler must abort the child instead of orphaning its fiber. + yield* pollWithTimeout( + Effect.sync(() => (cancels.includes(childSessionID) ? true as const : undefined)), + "NodeSkipped handler did not cancel the live child session", + ) + const updated = yield* store.getNode(dagID, "racy-node") + expect(updated?.status).toBe("skipped") + }), + ), + ) + }) +}) From 94c100eff4418e8cdd89fa1076a857d57bc72333 Mon Sep 17 00:00:00 2001 From: lex Date: Tue, 28 Jul 2026 21:31:27 +0800 Subject: [PATCH 2/3] chore(husky): fall back to per-package typecheck when bun workspace discovery fails bun's ancestor-directory workspace walk dies with CouldntReadCurrentDirectory in sandboxed checkouts, failing hooks on an environment defect. The shared runner keeps 'bun typecheck' as the primary path and falls back to running each package's own typecheck script only on that specific error; real type errors still fail on either path. --- .husky/pre-commit | 2 +- .husky/pre-push | 2 +- .husky/run-typecheck | 31 +++++++++++++++++++++++++++++++ 3 files changed, 33 insertions(+), 2 deletions(-) create mode 100644 .husky/run-typecheck diff --git a/.husky/pre-commit b/.husky/pre-commit index cdc74043c..ce93ce383 100644 --- a/.husky/pre-commit +++ b/.husky/pre-commit @@ -1,3 +1,3 @@ #!/bin/sh set -e -bun typecheck +sh .husky/run-typecheck diff --git a/.husky/pre-push b/.husky/pre-push index 5d3cc5341..fd8dfe238 100755 --- a/.husky/pre-push +++ b/.husky/pre-push @@ -17,4 +17,4 @@ if (process.versions.bun !== expectedBunVersion) { console.warn(`Warning: Bun version ${process.versions.bun} differs from expected ${expectedBunVersion}`); } ' -bun typecheck +sh .husky/run-typecheck diff --git a/.husky/run-typecheck b/.husky/run-typecheck new file mode 100644 index 000000000..e886963c3 --- /dev/null +++ b/.husky/run-typecheck @@ -0,0 +1,31 @@ +#!/bin/sh +# Shared typecheck runner for husky hooks. +# +# `bun typecheck` (turbo across the workspace) is the primary path. bun's +# workspace discovery walks ancestor directories and dies with +# CouldntReadCurrentDirectory in sandboxed checkouts whose parents are +# unreadable — an environment defect, not a code failure. In exactly that +# case, fall back to running every workspace package's own typecheck script +# directly (the turbo typecheck task declares no dependsOn, so per-package +# execution is equivalent). Real type errors still fail on either path. +set -e + +out=$(bun typecheck 2>&1) && { printf '%s\n' "$out"; exit 0; } +printf '%s\n' "$out" +case "$out" in + *CouldntReadCurrentDirectory*) ;; + *) exit 1 ;; +esac + +echo "bun workspace discovery unavailable — falling back to per-package typecheck" +root=$(pwd) +status=0 +# Mirrors the workspaces globs in package.json. +for pkg in packages/* packages/console/* packages/stats/* packages/sdk/js; do + [ -f "$pkg/package.json" ] || continue + script=$(sed -n 's/.*"typecheck":[[:space:]]*"\([^"]*\)".*/\1/p' "$pkg/package.json" | head -1) + [ -n "$script" ] || continue + echo "typecheck: $pkg" + (cd "$pkg" && PATH="$root/node_modules/.bin:$PATH" sh -c "$script") || status=1 +done +exit $status From 657707e17585bbd8a26249ea1f1d5e3beadda9cc Mon Sep 17 00:00:00 2001 From: lex Date: Tue, 28 Jul 2026 21:36:27 +0800 Subject: [PATCH 3/3] chore(husky): tolerate sandboxed buildinfo write denial in typecheck fallback tsgo -b packages (app/desktop) can fail with TS5033 'operation not permitted' writing .tsbuildinfo under sandboxed hook contexts. Tolerate a package failure only when TS5033 is the sole TS diagnostic; real type errors still fail. --- .husky/run-typecheck | 13 ++++++++++++- 1 file changed, 12 insertions(+), 1 deletion(-) diff --git a/.husky/run-typecheck b/.husky/run-typecheck index e886963c3..0723420bc 100644 --- a/.husky/run-typecheck +++ b/.husky/run-typecheck @@ -26,6 +26,17 @@ for pkg in packages/* packages/console/* packages/stats/* packages/sdk/js; do script=$(sed -n 's/.*"typecheck":[[:space:]]*"\([^"]*\)".*/\1/p' "$pkg/package.json" | head -1) [ -n "$script" ] || continue echo "typecheck: $pkg" - (cd "$pkg" && PATH="$root/node_modules/.bin:$PATH" sh -c "$script") || status=1 + pkgout=$( (cd "$pkg" && PATH="$root/node_modules/.bin:$PATH" sh -c "$script") 2>&1 ) && continue + printf '%s\n' "$pkgout" + # Same environment-defect class as above: sandboxed hook contexts can deny + # the .tsbuildinfo write of `tsgo -b` (TS5033 "operation not permitted"). + # Tolerate a failure ONLY when every reported TS error is that write denial + # — any real type error (or a crash with no TS output) still fails. + if printf '%s\n' "$pkgout" | grep -q "error TS5033.*operation not permitted" \ + && ! printf '%s\n' "$pkgout" | grep "error TS" | grep -qv "error TS5033"; then + echo "warn: $pkg buildinfo write denied by environment (TS5033) — no type errors, continuing" + continue + fi + status=1 done exit $status