Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 7 additions & 4 deletions .specgit.yaml
Original file line number Diff line number Diff line change
@@ -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
1 change: 1 addition & 0 deletions packages/core/src/plugin/command.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down
8 changes: 8 additions & 0 deletions packages/opencode/src/command/index.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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",
Expand Down Expand Up @@ -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,
Expand Down
25 changes: 25 additions & 0 deletions packages/opencode/src/dag/runtime/spawn.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
10 changes: 7 additions & 3 deletions packages/opencode/src/dag/templates/resolve.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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()`.
Expand All @@ -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 {
Expand Down Expand Up @@ -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 () => {
Expand Down
26 changes: 25 additions & 1 deletion packages/opencode/src/tool/workflow.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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,
}
}
Expand Down
14 changes: 14 additions & 0 deletions packages/opencode/test/command/command.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
129 changes: 129 additions & 0 deletions packages/opencode/test/dag/dag-spawn-pause-fence.test.ts
Original file line number Diff line number Diff line change
@@ -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>,
)
})
})
24 changes: 24 additions & 0 deletions packages/opencode/test/dag/dag-templates.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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")
Expand Down
20 changes: 20 additions & 0 deletions packages/opencode/test/dag/fixtures.ts
Original file line number Diff line number Diff line change
@@ -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",
Expand Down
31 changes: 31 additions & 0 deletions packages/opencode/test/dag/workflow-tool.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
Loading