fix(dag): product audit batch — #378 #379 #380 #381

.specgit.yaml

@@ -1,8 +1,11 @@
 version: 1
-delivery: issue374
+delivery: issue378
 context:
   kind: branch
-  branch: feat/375-issue375
+  branch: feat/378-issue378
 issues:
-  - 374
-pr: 375
+  - 378
+  - 379
+  - 380
+  - 381
+pr: 382

packages/core/src/plugin/command.ts

@@ -29,6 +29,7 @@ export const OrchestrationPolicyContent = orchestrationPolicy
 export const OrchestrationDomainsContent = orchestrationDomains
 export const WorkflowContent = workflowRouting
 export const DagFlowContent = DAG_FLOW_PROMPT
+export const DagTemplateUpdateContent = DAG_TEMPLATE_UPDATE_PROMPT
 export const DagInitContent = DAG_INIT_PROMPT
 export const DagAutoContent = DAG_AUTO_PROMPT
 

packages/opencode/src/command/index.ts

@@ -52,6 +52,7 @@ export const Default = {
   SUBGOAL: "subgoal",
   MEMORY: "memory",
   DAG_FLOW: "dag-flow",
+  DAG_TEMPLATE_UPDATE: "dag-template-update",
   DAG_INIT: "dag-init",
   DAG_AUTO: "dag-auto",
   IMPORT_HOOKS: "import-claude-hooks",
@@ -124,6 +125,13 @@ export const layer = Layer.effect(
         template: CommandPlugin.DagFlowContent,
         hints: hints(CommandPlugin.DagFlowContent),
       }
+      commands[Default.DAG_TEMPLATE_UPDATE] = {
+        name: Default.DAG_TEMPLATE_UPDATE,
+        description: CommandPlugin.DagTemplateUpdateDescription,
+        source: "command",
+        template: CommandPlugin.DagTemplateUpdateContent,
+        hints: hints(CommandPlugin.DagTemplateUpdateContent),
+      }
       commands[Default.DAG_INIT] = {
         name: Default.DAG_INIT,
         description: CommandPlugin.DagInitDescription,

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

@@ -407,6 +407,31 @@ export function spawnNode(
           return
         }
         try {
+          // #379 pause fence: scheduler admission is the only pause gate, so a
+          // node queued before control(pause) still holds a spawn fiber that
+          // would materialize its child session the moment a permit frees —
+          // contradicting the documented pause contract ("prevents new nodes
+          // from spawning"). Hold the fiber here while the durable workflow
+          // row reads `paused`: the node stays queued (never terminalized) and
+          // the fiber proceeds on control(resume). A workflow that was deleted
+          // or terminalized during the wait falls through to the adoption
+          // fence / nodeStarted guard below, which already no-op dead targets.
+          // The status read fails open (unreadable ⇒ treat as not paused):
+          // the adoption fence below is the authoritative revalidation, so a
+          // transient store blip must not kill the spawn fiber here.
+          // Effect.suspend defers the call so even a store facade lacking the
+          // method surfaces as a captured defect instead of a synchronous throw.
+          const readPauseStatus = Effect.exit(Effect.suspend(() => dag.store.getWorkflow(input.dagID))).pipe(
+            Effect.map((outcome) => (Exit.isSuccess(outcome) ? outcome.value?.status : undefined)),
+          )
+          if ((yield* readPauseStatus) === "paused") {
+            yield* Effect.logWarning(
+              `Workflow ${input.dagID} paused while node ${input.nodeID} was queued — holding spawn until resume`,
+            )
+            while ((yield* readPauseStatus) === "paused") {
+              yield* Effect.sleep(250)
+            }
+          }
           // #270 window-2 spawn-admission fence (C4): the node was durably
           // admitted (nodeQueued above) but the child session is about to
           // materialize — a deletion cascade (Session.remove → FK) committed in

packages/opencode/src/dag/templates/resolve.ts

@@ -6,7 +6,7 @@
  *
  * Resolves a node's `prompt_template` declaration into a final prompt string:
  * - `id` reference → reads the `.md` file from project (`.opencode/dag-prompts/`)
- *   or global (`~/.config/opencode/dag-prompts/`) directory
+ *   or global (`<config dir>/dag-prompts/`) directory
  * - `inline` → used directly as the template source (no filesystem round-trip)
  *
  * Both paths go through `{{var}}` interpolation and `sanitize()`.
@@ -16,9 +16,10 @@
  */
 
 import { Effect } from "effect"
-import * as os from "node:os"
 import * as path from "node:path"
 import * as fs from "node:fs/promises"
+import { Global } from "@opencode-ai/core/global"
+import { Flag } from "@opencode-ai/core/flag/flag"
 import { sanitizeInput } from "./sanitize"
 
 export interface TemplateRef {
@@ -92,7 +93,10 @@ function readById(id: string, projectDir: string): Effect.Effect<string, Error>
       return yield* Effect.fail(new Error(`Invalid template id: ${id}`))
     }
     const projectPath = path.join(projectDir, ".opencode", "dag-prompts", `${id}.md`)
-    const globalPath = path.join(os.homedir(), ".config", "opencode", "dag-prompts", `${id}.md`)
+    // Same OPENCODE_CONFIG_DIR redirect the Global service applies (siblings:
+    // dag/workflows.ts, dag/config.ts) so redirected setups resolve their
+    // globally installed prompts.
+    const globalPath = path.join(Flag.OPENCODE_CONFIG_DIR ?? Global.Path.config, "dag-prompts", `${id}.md`)
 
     // Try project first (overrides global), then global
     const result = yield* Effect.promise(async () => {

packages/opencode/src/tool/workflow.ts

@@ -663,9 +663,33 @@ export const WorkflowTool = Tool.define<
                 dag.extend(params.workflow_id, result.prepared.nodes),
                 "Terminal workflows are immutable except for the additive-extend reopen, which requires the workflow to have completed naturally at a wake-eligible reporting checkpoint (fragment adds new node ids; no early control(complete); no executed node beyond the checkpoint — condition-skipped dependents are fine). When the reopen does not apply, recover by starting a NEW workflow spec that reuses this workflow's completed outputs as static input.",
               ).pipe(Effect.orDie)
+              // #381: extend shares replan's resume contract — a paused
+              // workflow must resume for the added nodes to ever run (pause
+              // admits nothing and no wake path prompts a resume), and the
+              // extend intent is "the graph grew, proceed". Resume races with
+              // concurrent control ops are tolerated: the extend already
+              // landed, so never die on them.
+              const wfAfterExtend = yield* dag.store.getWorkflow(params.workflow_id).pipe(Effect.orDie)
+              const resumedFromPause = wfAfterExtend?.status === "paused"
+              const resumedOk = resumedFromPause
+                ? yield* dag.resume(params.workflow_id).pipe(
+                    Effect.map(() => true),
+                    Effect.catch((error) =>
+                      Effect.gen(function* () {
+                        yield* Effect.logWarning("Workflow resume after extend failed", { wfId: params.workflow_id, error })
+                        return false
+                      }),
+                    ),
+                  )
+                : false
+              const pauseNote = !resumedFromPause
+                ? ""
+                : resumedOk
+                  ? "\nWorkflow was paused and has been resumed — added nodes are now schedulable."
+                  : "\nWorkflow was paused; automatic resume raced with another control op — check status and issue control(resume) if still paused."
               return {
                 title: `Workflow extended: ${r.add.length} nodes added`,
-                output: `<workflow id="${params.workflow_id}" action="extend">\nAdded: ${r.add.join(", ")}\n</workflow>`,
+                output: `<workflow id="${params.workflow_id}" action="extend">\nAdded: ${r.add.join(", ")}${pauseNote}\n</workflow>`,
                 metadata: { workflowId: params.workflow_id, added: r.add } as Metadata,
               }
             }

packages/opencode/test/command/command.test.ts

@@ -72,6 +72,20 @@ describe("legacy command registry", () => {
     }),
   )
 
+  it.instance("registers the canonical dag-template-update command", () =>
+    Effect.gen(function* () {
+      const commands = yield* Command.Service
+
+      expect(yield* commands.get("dag-template-update")).toMatchObject({
+        name: "dag-template-update",
+        description: CommandPlugin.DagTemplateUpdateDescription,
+        source: "command",
+        template: CommandPlugin.DagTemplateUpdateContent,
+        hints: ["$ARGUMENTS"],
+      })
+    }),
+  )
+
   it.instance("registers the canonical dag-init and dag-auto commands", () =>
     Effect.gen(function* () {
       const commands = yield* Command.Service

packages/opencode/test/dag/dag-spawn-pause-fence.test.ts

@@ -0,0 +1,129 @@
+// oxlint-disable typescript-eslint/no-unsafe-type-assertion -- Mirrors the
+// dag-structured-output.test.ts harness idiom (and dag-location-guards.test.ts
+// suppression precedent): mocked Agent/Session/prompt layers and the typed
+// reply fixture use type-only shims for branded IDs — converting them would
+// fork the shared harness shape without changing behavior.
+import { describe, expect, it } from "bun:test"
+import { Effect, Layer, Semaphore, Fiber } from "effect"
+import type { SessionV1 } from "@opencode-ai/core/v1/session"
+import { SessionPrompt } from "@/session/prompt"
+import { Dag } from "@/dag/dag"
+import { Agent } from "@/agent/agent"
+import { Session } from "@/session/session"
+import { spawnNode, type NodeSpawnInput } from "@/dag/runtime/spawn"
+import { makeNodeRow, makeWorkflowRow } from "./fixtures"
+import type { DagStore } from "@opencode-ai/core/dag/store"
+
+type TrackedEvent = { type: string; nodeID: string }
+
+// #379 regression: control(pause) must hold a queued node's spawn fiber at the
+// pause fence — no child session, no terminal event — and control(resume)
+// (workflow status back to "running") must let the fiber proceed and complete.
+
+function makeHarness(workflowStatus: () => string, claimAdoption: () => boolean) {
+  const events: TrackedEvent[] = []
+  let createCalls = 0
+  const storeStub: Partial<DagStore.Interface> = {
+    tryClaimAdoption: () => Effect.sync(() => claimAdoption()),
+    getWorkflow: () => Effect.sync(() => makeWorkflowRow({ status: workflowStatus() })),
+    getNode: Effect.fn("s")((_workflowID: string, nodeID: string) =>
+      Effect.sync(() => makeNodeRow({ id: nodeID, status: "queued" }))),
+  }
+  const dagLayer = Layer.mock(Dag.Service, {
+    store: storeStub as DagStore.Interface,
+    nodeQueued: Effect.fn("s")(() => Effect.void),
+    nodeStarted: Effect.fn("s")(() => Effect.void),
+    nodeCompleted: Effect.fn("s")((_dagID: string, nodeID: string) =>
+      Effect.sync(() => events.push({ type: "nodeCompleted", nodeID }))),
+    nodeFailed: Effect.fn("s")((_dagID: string, nodeID: string) =>
+      Effect.sync(() => events.push({ type: "nodeFailed", nodeID }))),
+    nodeSkipped: Effect.fn("s")((_dagID: string, nodeID: string) =>
+      Effect.sync(() => events.push({ type: "nodeSkipped", nodeID }))),
+  })
+  const agentLayer = 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: {},
+    }),
+    list: () => Effect.succeed([]),
+    defaultAgent: () => Effect.succeed("build"),
+  })
+  const sessionLayer = Layer.mock(Session.Service, {
+    get: () => Effect.succeed({ id: "ses_parent" as never, permission: [], agent: "build" } as never),
+    create: () => Effect.sync(() => {
+      createCalls++
+      return { id: "ses_child" as never } as never
+    }),
+    list: () => Effect.succeed([]),
+    messages: () => Effect.succeed([]),
+  })
+  const promptLayer = Layer.mock(SessionPrompt.Service, {
+    prompt: () => Effect.succeed(reply()),
+  })
+  return { events, createCalls: () => createCalls, fullLayer: Layer.mergeAll(dagLayer, agentLayer, sessionLayer, promptLayer) }
+}
+
+function reply(): SessionV1.WithParts {
+  return {
+    info: {
+      id: "msg_reply", role: "assistant", parentID: "msg_parent", sessionID: "ses_child",
+      mode: "build", agent: "build", cost: 0, path: { cwd: "/tmp", root: "/tmp" },
+      tokens: { input: 0, output: 0, reasoning: 0, cache: { read: 0, write: 0 } },
+      modelID: "test-model", providerID: "test",
+      time: { created: 0 }, finish: "stop",
+    },
+    parts: [{ type: "text", text: "Task completed" }],
+  } as SessionV1.WithParts
+}
+
+function makeSpawnInput(): NodeSpawnInput {
+  return {
+    dagID: "wf-1",
+    nodeID: "node-1",
+    node: makeNodeRow(),
+    parentSessionID: "ses_parent",
+    promptParts: [{ type: "text", text: "do the thing" }],
+  }
+}
+
+describe("spawnNode pause fence (#379)", () => {
+  it("holds a queued node at the fence while the workflow is paused, then spawns after resume", async () => {
+    let paused = true
+    const harness = makeHarness(() => (paused ? "paused" : "running"), () => true)
+    await Effect.runPromise(
+      Effect.scoped(
+        Effect.gen(function* () {
+          const result = yield* spawnNode(Semaphore.makeUnsafe(1), makeSpawnInput())
+          // Two+ poll cycles past the initial read: the fiber must still be
+          // held — no child session, no terminal transition.
+          yield* Effect.sleep(600)
+          expect(harness.createCalls()).toBe(0)
+          expect(harness.events).toHaveLength(0)
+          paused = false
+          yield* Fiber.await(result.fiber)
+          expect(harness.createCalls()).toBe(1)
+          expect(harness.events.find((e) => e.type === "nodeCompleted")).toBeDefined()
+        }),
+      ).pipe(Effect.provide(harness.fullLayer)) as Effect.Effect<never>,
+    )
+  })
+
+  it("exits without creating a session when the workflow terminalizes during the pause hold", async () => {
+    let status = "paused"
+    const harness = makeHarness(() => status, () => false)
+    await Effect.runPromise(
+      Effect.scoped(
+        Effect.gen(function* () {
+          const result = yield* spawnNode(Semaphore.makeUnsafe(1), makeSpawnInput())
+          yield* Effect.sleep(300)
+          expect(harness.createCalls()).toBe(0)
+          status = "cancelled"
+          yield* Fiber.await(result.fiber)
+          expect(harness.createCalls()).toBe(0)
+          expect(harness.events).toHaveLength(0)
+        }),
+      ).pipe(Effect.provide(harness.fullLayer)) as Effect.Effect<never>,
+    )
+  })
+})

packages/opencode/test/dag/dag-templates.test.ts

@@ -199,6 +199,30 @@ describe("resolveTemplate", () => {
     await fs.rm(tmpDir, { recursive: true })
   })
 
+  it("resolves a global template from a redirected OPENCODE_CONFIG_DIR", async () => {
+    // #380: the global dag-prompts lookup must honor the same
+    // OPENCODE_CONFIG_DIR redirect the Global service applies, not a
+    // hardcoded ~/.config/opencode path.
+    const globalDir = await fs.mkdtemp(path.join(os.tmpdir(), "dag-global-"))
+    const promptsDir = path.join(globalDir, "dag-prompts")
+    await fs.mkdir(promptsDir, { recursive: true })
+    await fs.writeFile(path.join(promptsDir, "redirected-tmpl.md"), "Global {{scope}} template!", "utf-8")
+    const previous = process.env.OPENCODE_CONFIG_DIR
+    process.env.OPENCODE_CONFIG_DIR = globalDir
+    try {
+      const projectDir = await fs.mkdtemp(path.join(os.tmpdir(), "dag-proj-"))
+      const result = await Effect.runPromise(
+        resolveTemplate({ id: "redirected-tmpl", input: { scope: "redirected" } }, projectDir),
+      )
+      expect(result).toBe("Global redirected template!")
+      await fs.rm(projectDir, { recursive: true })
+    } finally {
+      if (previous === undefined) delete process.env.OPENCODE_CONFIG_DIR
+      else process.env.OPENCODE_CONFIG_DIR = previous
+      await fs.rm(globalDir, { recursive: true })
+    }
+  })
+
   it("fails for non-existent template id", async () => {
     const program = resolveTemplate({ id: "non-existent-template" }, "/tmp")
     await expect(Effect.runPromise(program)).rejects.toThrow("not found")

packages/opencode/test/dag/fixtures.ts

@@ -1,5 +1,25 @@
 import type { DagStore } from "@opencode-ai/core/dag/store"
 
+export function makeWorkflowRow(overrides: Partial<DagStore.WorkflowRow> = {}): DagStore.WorkflowRow {
+  return {
+    id: "wf-1",
+    projectId: "proj-1",
+    sessionId: "ses_parent",
+    directory: null,
+    title: "Test Workflow",
+    status: "running",
+    config: "{}",
+    seq: 1,
+    wakeReported: false,
+    graphRev: 1,
+    startedAt: null,
+    completedAt: null,
+    timeCreated: 1,
+    timeUpdated: 1,
+    ...overrides,
+  }
+}
+
 export function makeNodeRow(overrides: Partial<DagStore.NodeRow> = {}): DagStore.NodeRow {
   return {
     id: "node-1",

packages/opencode/test/dag/workflow-tool.test.ts

@@ -1258,6 +1258,37 @@ describe("workflow tool execution", () => {
     }),
   )
 
+  runtime.effect("extending a paused workflow resumes it so added nodes can run (#381)", () =>
+    Effect.gen(function* () {
+      published.length = 0
+      const info = yield* WorkflowTool
+      const workflow = yield* info.init()
+      const spec_path = yield* writeWorkflowSpec("paused-extend", {
+        nodes: [
+          {
+            id: "added-while-paused",
+            name: "Added while paused",
+            worker_type: "general",
+            depends_on: [],
+            prompt_template: { inline: "work" },
+          },
+        ],
+      })
+      const result = yield* workflow.execute(
+        Schema.decodeUnknownSync(Parameters)({ params: {
+          action: "extend",
+          workflow_id: "dag_paused",
+          spec_path,
+        }}),
+        toolContext(),
+      )
+
+      expect(result.title).toContain("Workflow extended: 1 nodes added")
+      expect(result.output).toContain("has been resumed")
+      expect(published.some((event) => event.type === DagEvent.WorkflowResumed.type)).toBe(true)
+    }),
+  )
+
   runtime.effect("rejects inline or missing spec sources before side effects", () =>
     Effect.gen(function* () {
       const info = yield* WorkflowTool