Skip to content

fix(dag): mark wake reported at admit time #326

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
LeXwDeX merged 2 commits into from
Aug 17, 2026
Merged

fix(dag): mark wake reported at admit time #326

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
938a197
fix(dag): mark wake reported at admit time (issue #321)
LeXwDeX Aug 17, 2026
311091b
docs(dag): correct stale unregister-ordering comment after admit-time…
LeXwDeX Aug 17, 2026
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
120 changes: 80 additions & 40 deletions packages/opencode/src/dag/runtime/loop.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -74,16 +74,15 @@ const serviceLayer = Layer.effect(
const recovering = new Set<string>()
const wakeInFlight = new Set<string>()
const wakePending = new Set<string>()
// GOAL-FP-01-14: per-session record of the last wake summary whose
// transcript part was written. The durable mark runs AFTER the write
// (at-least-once delivery: a mark failure keeps the batch unreported
// for a retry), so a retry of an already-written summary would
// re-inject the same digest into the transcript. The retry dedupes on
// this map and only re-marks. In-process only — a crash between write
// and mark still duplicates on the restart sweep (a durable
// delivering-marker would need a schema change; registered, see
// GOAL-FP-01-14). Capped: evicting entries degrades to the pre-fix
// duplicate visibility, never to a lost wake.
// Per-session record of the last wake summary whose transcript part was
// written. Admit success IS the delivery (issue #321): the durable
// wake_reported mark lands immediately after the write, not after the
// wake-driven turn completes, so the in-memory map only needs to dedupe
// the narrow mark-retry path — if the leased mark returns None the rows
// stay unreported and a later trigger re-marks WITHOUT re-prompting.
// In-process only; restart durability now comes from wake_reported being
// persisted at admit time, not from this map. Capped: evicting entries
// degrades to a redundant re-prompt, never to a lost wake.
const deliveredWakeSummaries = new Map<string, string>()

// Seed the commented global dag.jsonc once per instance init — the
Expand Down Expand Up @@ -1315,53 +1314,94 @@ const serviceLayer = Layer.effect(
)
if (!wakeLease) return

// Persist wake_reported AFTER successful delivery only.
// A failure stays durable for a later idle event or restart scan;
// it must not spin synchronously on the same row.
// The part is marked synthetic: model-visible (the orchestrator
// receives the node result and can act) but NOT rendered as a user
// message in the TUI chat — DAG data surfaces via the sidebar panel
// and Inspector, keeping the chat conversation clean.
//
// GOAL-FP-01-14: the transcript part is written BEFORE the
// durable mark. A mark failure (or a crash between the two)
// leaves the batch unreported and the retry would re-inject the
// SAME summary. When this session already had this exact summary
// written, skip the prompt and only re-mark — the write is
// idempotent in effect because an identical digest adds no
// information. A differing summary (new results committed
// between attempts) always prompts.
// ADMIT SUCCESS == DELIVERED (issue #321). The previous contract
// (GOAL-FP-01-14) persisted wake_reported only AFTER the whole
// wake-driven parent turn completed. A restart or mid-turn
// interruption therefore left wake_reported=false while the
// synthetic part was already durable in transcript, so the startup
// sweep re-injected a byte-identical wake (real incident: the same
// 4391-char wake injected twice, ~10 min apart, after a TUI
// restart). Redelivery adds duplicates, never information. The
// mark (and the terminal-workflow unregisters) now land right
// after admitIfIdle admits the part, BEFORE awaiting the turn.
//
// The in-memory dedup map still guards the retry path: if the
// leased mark below returns None (generation/owner changed), the
// rows stay unreported and a later trigger re-marks without
// re-prompting. A differing summary (new results committed between
// attempts) always prompts.
if (deliveredWakeSummaries.size > 1024) deliveredWakeSummaries.clear()
const didDeliver = yield* Effect.gen(function* () {
let wakeTurn: Effect.Effect<SessionV1.WithParts> | undefined
if (deliveredWakeSummaries.get(sessionID) !== summary) {
const delivered = yield* SessionPrompt.admitIfIdle(promptSvc, automation, wakeLease, {
sessionID: SessionID.make(sessionID),
parts: [{ type: "text", text: summary, synthetic: true }],
})
if (Option.isNone(delivered)) return false
deliveredWakeSummaries.set(sessionID, summary)
yield* delivered.value.pipe(
Effect.onError(() =>
Effect.sync(() => {
if (deliveredWakeSummaries.get(sessionID) === summary) {
deliveredWakeSummaries.delete(sessionID)
}
}),
),
)
wakeTurn = delivered.value
}

const markLease = yield* automation.claim(SessionID.make(sessionID), { kind: "dag" })
if (Option.isNone(markLease)) return false
const marked = yield* automation.use(markLease.value, store.markWakeBatchReported(batch))
if (Option.isNone(marked)) return false
plan.unresponsiveDagIDs.forEach((workflowID) => deliveredUnresponsiveDagIDs.add(workflowID))
yield* Effect.forEach(
batch.workflows.filter((workflow) => isWorkflowTerminalStatus(workflow.status as never)),
(workflow) => automation.unregister(SessionID.make(sessionID), { kind: "dag", id: workflow.id }),
{ discard: true },
// Admit success == delivered (issue #321): persist wake_reported
// at admit time. A leased mark returning None (generation/owner
// raced) is treated as retry-later — the rows stay unreported and
// a later trigger re-marks — but the admitted turn still runs
// below (its end-of-turn idle is what re-arms that retry).
const markLease = Option.getOrUndefined(
yield* automation.claim(SessionID.make(sessionID), { kind: "dag" }),
)
return true
// Any mark failure (lease lost, generation raced, or the store
// write dying) degrades to retry-later instead of propagating:
// the rows stay unreported and a later trigger re-marks, while
// the admitted turn below still runs (its end-of-turn idle is
// what re-arms that retry). Only interruption propagates.
const markSucceeded = markLease
? Option.isSome(
yield* automation.use(markLease, store.markWakeBatchReported(batch)).pipe(
Effect.catchCause((cause) =>
Cause.hasInterrupts(cause)
? Effect.failCause(cause)
: Effect.logWarning("DAG wake batch mark failed; rows stay unreported for retry", {
sessionID,
cause: Cause.pretty(cause),
}).pipe(Effect.as(Option.none())),
),
),
)
: false
if (markSucceeded) {
plan.unresponsiveDagIDs.forEach((workflowID) => deliveredUnresponsiveDagIDs.add(workflowID))
yield* Effect.forEach(
batch.workflows.filter((workflow) => isWorkflowTerminalStatus(workflow.status as never)),
(workflow) => automation.unregister(SessionID.make(sessionID), { kind: "dag", id: workflow.id }),
{ discard: true },
)
}

// Pacing — keep one wake turn at a time. The turn runs AFTER the
// durable mark, so its failure can no longer lose the report;
// swallow non-interrupt failures instead of surfacing them as a
// delivery failure. It also runs when the mark raced, so its
// end-of-turn idle re-arms the mark retry.
if (wakeTurn) {
yield* wakeTurn.pipe(
Effect.catchCause((cause) =>
Cause.hasInterrupts(cause)
? Effect.failCause(cause)
: Effect.logInfo("DAG wake turn did not complete; wake already reported", {
sessionID,
cause: Cause.pretty(cause),
}),
),
)
}
return markSucceeded
}).pipe(
Effect.catchCause((cause) =>
Effect.logWarning("DAG wake delivery failed", { sessionID, cause: Cause.pretty(cause) }).pipe(
Expand Down
11 changes: 7 additions & 4 deletions packages/opencode/src/session/automation-lease.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -101,10 +101,13 @@ export const layer = Layer.effect(
// GOAL-FP-01-02: when the dag ownership actually disappears (owner
// transitions dag → goal/none), re-trigger the goal evaluation through
// the EXISTING idle status event mechanism so a goal that yielded to the
// dag on the last idle event gets a fresh evaluation. The final dag
// unregister of a wake delivery (U2 in dag/runtime/loop.ts) lands AFTER
// the wake turn's idle event — without this re-trigger the goal silently
// stalls until the next external idle. This is also the GOAL-FP-01-11
// dag on the last idle event gets a fresh evaluation. Since issue #321
// the final dag unregister of a wake delivery lands MID-TURN at admit
// time (the mark moves with it), so the session is busy when the owner
// flips: the idle gate below skips the re-emit here, and the in-flight
// wake turn re-emits idle on completion, which re-drives the goal.
// Without this re-trigger path the goal silently stalls until the next
// external idle. This is also the GOAL-FP-01-11
// mitigation surface: a claim that lost the ownership race gets another
// chance once the owner actually transfers.
//
Expand Down
53 changes: 32 additions & 21 deletions packages/opencode/test/dag/dag-timeout-escalation.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -195,6 +195,7 @@ function runLoopTest<A>(
test: (services: {
readonly dag: Dag.Interface
readonly store: DagStore.Interface
readonly status: SessionStatus.Interface
readonly childPrompts: Queue.Queue<PromptGate>
readonly parentPrompts: Queue.Queue<ParentPromptGate>
readonly getCancelCount: () => number
Expand All @@ -212,6 +213,7 @@ function runLoopTest<A>(
const dag = yield* Dag.Service
const loop = yield* DagLoop.Service
const store = yield* DagStore.Service
const status = yield* SessionStatus.Service
const database = yield* Database.Service
yield* database.db.insert(ProjectTable).values({
id: Project.ID.make("project-1"),
Expand All @@ -230,6 +232,7 @@ function runLoopTest<A>(
return yield* test({
dag,
store,
status,
childPrompts,
parentPrompts,
getCancelCount: harness.getCancelCount,
Expand Down Expand Up @@ -1044,7 +1047,7 @@ describe("DagLoop timeout escalation", () => {
// P5 escalationPending ∧ wakeReported → proceed [recovery — test below + L880/L567]
it("blocks re-time while the escalation wake is undelivered (Q2: escalationPending ∧ ¬wakeReported ⇒ skip)", async () => {
await Effect.runPromise(
runLoopTest(({ dag, store, childPrompts, parentPrompts }) =>
runLoopTest(({ dag, store, status, childPrompts, parentPrompts }) =>
Effect.gen(function* () {
const dagID = yield* dag.create({
projectID: "project-1",
Expand All @@ -1054,10 +1057,16 @@ describe("DagLoop timeout escalation", () => {
})
yield* takeWithin(childPrompts, "a did not start")

// Acceptance #2: the deadline-driven INITIAL escalation (watchdog →
// nodeTimeoutEscalated → first wake) is NOT touched by the gate —
// the gate only governs the replan re-time path. It fires and its
// wake reaches the parent.
// issue #321: wake_reported now lands at ADMIT time, so an in-flight
// turn no longer holds wake_reported=false (the wake is reported as
// soon as it is admitted). The only window that keeps the escalation
// wake UNDELIVERED now is a BUSY parent session — the idle-gate never
// admits it. Hold the wake there.
yield* status.set(SessionID.make("ses_parent"), { type: "busy" })

// Initial escalation fires; while the parent is busy its wake is held
// UNDELIVERED (never admitted). The node sits at the public-path state
// [escalationPending ∧ ¬wakeReported ∧ deadline≤now].
const escalated = yield* pollWithTimeout(
store.getNode(dagID, "a").pipe(
Effect.map((current) => current?.timeoutExtensions === 1 ? current : undefined),
Expand All @@ -1066,17 +1075,10 @@ describe("DagLoop timeout escalation", () => {
)
expect(escalated.status).toBe("running")
expect(escalated.escalationPending).toBe(true)
expect(escalated.wakeReported).toBe(false)
const baselineDeadline = escalated.deadlineMs
const timeoutWake = yield* takeWithin(parentPrompts, "initial escalation wake did not reach the parent")
expect(timeoutWake.text).toContain("[DAG Node Timeout]")

// Hold the wake UNDELIVERED: the harness blocks delivery on the
// release Deferred, and the loop persists wake_reported=true only
// AFTER successful delivery (loop.ts:1125). The node sits at the
// public-path state [escalationPending ∧ ¬wakeReported ∧ deadline≤now].
const undelivered = yield* store.getNode(dagID, "a")
expect(undelivered?.escalationPending).toBe(true)
expect(undelivered?.wakeReported).toBe(false)
// The wake is held by the busy gate — nothing reached the parent yet.
expect(Option.isNone(yield* Queue.poll(parentPrompts))).toBe(true)

// Main agent replans with a NEW timeout. Q2 must SKIP the re-time:
// adjudication cannot land before the escalation wake was delivered.
Expand Down Expand Up @@ -1106,7 +1108,11 @@ describe("DagLoop timeout escalation", () => {
expect(reEscalated.wakeReported).toBe(false)
expect(reEscalated.deadlineMs).toBe(baselineDeadline)

// Release the held wake so the loop marks delivery before teardown.
// Release the held wake: the parent goes idle, the wake is admitted
// (wake_reported lands at admit, issue #321), and the turn settles.
yield* status.set(SessionID.make("ses_parent"), { type: "idle" })
const timeoutWake = yield* takeWithin(parentPrompts, "escalation wake did not reach the parent once idle")
expect(timeoutWake.text).toContain("[DAG Node Timeout]")
yield* Deferred.succeed(timeoutWake.release, "success")
}),
),
Expand Down Expand Up @@ -1179,7 +1185,7 @@ describe("DagLoop timeout escalation", () => {
// terminal-cleanup path.
it("C1: nodeExtendTimeout distinguishes Q2 rejection (-2) from terminal rejection (0) — three-valued contract", async () => {
await Effect.runPromise(
runLoopTest(({ dag, store, childPrompts, parentPrompts }) =>
runLoopTest(({ dag, store, status, childPrompts, parentPrompts }) =>
Effect.gen(function* () {
const dagID = yield* dag.create({
projectID: "project-1",
Expand All @@ -1189,12 +1195,15 @@ describe("DagLoop timeout escalation", () => {
})
const gate = yield* takeWithin(childPrompts, "a did not start")

// Escalation fires; the wake is held UNDELIVERED so the node sits at
// the Q2 state [escalationPending ∧ ¬wakeReported ∧ running].
// issue #321: wake_reported lands at ADMIT time, so the Q2 state
// [escalationPending ∧ ¬wakeReported] can no longer be held by an
// in-flight turn — hold the escalation wake UNDELIVERED via the busy
// idle-gate instead, which keeps it never admitted.
yield* status.set(SessionID.make("ses_parent"), { type: "busy" })
const escalated = yield* pollWithTimeout(
store.getNode(dagID, "a").pipe(
Effect.map((current) =>
current?.timeoutExtensions === 1 && current.escalationPending && !current.wakeReported
current && current.status === "running" && current.escalationPending && !current.wakeReported
? current
: undefined,
),
Expand All @@ -1213,7 +1222,9 @@ describe("DagLoop timeout escalation", () => {
expect(afterQ2?.status).toBe("running")
expect(afterQ2?.deadlineMs).toBe(frozenDeadline)

// Deliver the wake and let the child finish — the node terminalizes.
// Release the held wake (idle admits it) and let the child finish —
// the node terminalizes.
yield* status.set(SessionID.make("ses_parent"), { type: "idle" })
const wake = yield* takeWithin(parentPrompts, "escalation wake did not reach the parent")
yield* Deferred.succeed(wake.release, "success")
yield* Deferred.succeed(gate.release, "done")
Expand Down
Loading
Loading