fix(dag): cross-instance adoption guard, subscription defect survival, cancel-skip race

.husky/pre-commit

@@ -1,3 +1,3 @@
 #!/bin/sh
 set -e
-bun typecheck
+sh .husky/run-typecheck

.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

.husky/run-typecheck

@@ -0,0 +1,42 @@
+#!/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"
+  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

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:

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

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, []],

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.

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<Info> {
+export function load(projectDir: string, options: { autoSeed?: boolean } = {}): Effect.Effect<Info> {
   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<NodeJS.ErrnoException | undefined> {
   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) {