diff --git a/.specgit.yaml b/.specgit.yaml index 300153e43b..1007468662 100644 --- a/.specgit.yaml +++ b/.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 diff --git a/packages/core/src/plugin/command.ts b/packages/core/src/plugin/command.ts index 2636a82076..653ed3e895 100644 --- a/packages/core/src/plugin/command.ts +++ b/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 diff --git a/packages/opencode/src/command/index.ts b/packages/opencode/src/command/index.ts index 446e7a5ec3..fea6d00978 100644 --- a/packages/opencode/src/command/index.ts +++ b/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, diff --git a/packages/opencode/src/dag/runtime/spawn.ts b/packages/opencode/src/dag/runtime/spawn.ts index 8ccad71b82..cf1e4b1575 100644 --- a/packages/opencode/src/dag/runtime/spawn.ts +++ b/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 diff --git a/packages/opencode/src/dag/templates/resolve.ts b/packages/opencode/src/dag/templates/resolve.ts index a68356f486..6aa281e762 100644 --- a/packages/opencode/src/dag/templates/resolve.ts +++ b/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 (`/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 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 () => { diff --git a/packages/opencode/src/tool/workflow.ts b/packages/opencode/src/tool/workflow.ts index 7266b1a2e2..0a66a9552e 100644 --- a/packages/opencode/src/tool/workflow.ts +++ b/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: `\nAdded: ${r.add.join(", ")}\n`, + output: `\nAdded: ${r.add.join(", ")}${pauseNote}\n`, metadata: { workflowId: params.workflow_id, added: r.add } as Metadata, } } diff --git a/packages/opencode/test/command/command.test.ts b/packages/opencode/test/command/command.test.ts index 1fdc536faf..a1a7c182b4 100644 --- a/packages/opencode/test/command/command.test.ts +++ b/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 diff --git a/packages/opencode/test/dag/dag-spawn-pause-fence.test.ts b/packages/opencode/test/dag/dag-spawn-pause-fence.test.ts new file mode 100644 index 0000000000..1eafb1e6f9 --- /dev/null +++ b/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 = { + 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, + ) + }) + + 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, + ) + }) +}) diff --git a/packages/opencode/test/dag/dag-templates.test.ts b/packages/opencode/test/dag/dag-templates.test.ts index 6c259a8fe2..9184943162 100644 --- a/packages/opencode/test/dag/dag-templates.test.ts +++ b/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") diff --git a/packages/opencode/test/dag/fixtures.ts b/packages/opencode/test/dag/fixtures.ts index 7fc681d304..872a71591d 100644 --- a/packages/opencode/test/dag/fixtures.ts +++ b/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 { + 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 { return { id: "node-1", diff --git a/packages/opencode/test/dag/workflow-tool.test.ts b/packages/opencode/test/dag/workflow-tool.test.ts index 97fbca7200..50b74f0ac3 100644 --- a/packages/opencode/test/dag/workflow-tool.test.ts +++ b/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