diff --git a/.tmp-651d.mjs b/.tmp-651d.mjs new file mode 100644 index 00000000..c861724e --- /dev/null +++ b/.tmp-651d.mjs @@ -0,0 +1,37 @@ +import fs from "fs"; +const f = "extensions/taskplane/hold-state.ts"; let s = fs.readFileSync(f, "utf8"); +const rep = (a, b) => { if (s.split(a).length !== 2) { console.error("ANCHOR", a.slice(0, 80), s.split(a).length); process.exit(1); } s = s.replace(a, b); }; +rep(` if (open.length > 0) { + if (existing.status === "held") return noop; + existing.status = "held"; + existing.exitReason = \`Held — awaiting ruling on \${open.map((h) => h.escalationId).join(", ")}\`; + existing.endTime = null; + return { changed: true, restore }; + } + if (releasedUnacked.length > 0) { + if (existing.status === "running" && !/^Held|^Hold timeout/.test(existing.exitReason)) return noop; + existing.status = "running"; + existing.exitReason = \`Ruling \${releasedUnacked.map((h) => h.ruling?.id ?? "?").join(", ")} accepted — worker relaunched\`; + existing.endTime = null; + return { changed: true, restore }; + } + return noop;`, +` let want: { status: LaneTaskOutcome["status"]; exitReason: string } | null = null; + if (open.length > 0) { + want = { + status: "held", + exitReason: \`Held — awaiting ruling on \${open.map((h) => h.escalationId).join(", ")}\`, + }; + } else if (releasedUnacked.length > 0) { + want = { + status: "running", + exitReason: \`Ruling \${releasedUnacked.map((h) => h.ruling?.id ?? "?").join(", ")} accepted — worker relaunched\`, + }; + } + if (!want) return noop; + if (existing.status === want.status && existing.exitReason === want.exitReason) return noop; + existing.status = want.status; + existing.exitReason = want.exitReason; + existing.endTime = null; + return { changed: true, restore };`); +fs.writeFileSync(f, s); console.log("ok"); diff --git a/CHANGELOG.md b/CHANGELOG.md index 53b0b56d..1c5231e9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- **A released hold left the task badged `held` and let its lane successor show `running` with the wrong telemetry** (#651). Three writers: (1) the strict hold-store write that persists a release (or an open) now projects the bound task's record in the same write — open → `held`, released → `running` with the stale hold-timeout reason cleared — instead of waiting for the next unrelated task-transition persist; (2) resume reconciliation decides "session alive" per **task** from the registry manifests' task ids, so a not-yet-started successor sharing the lane's session name stays `pending` (legacy registries without task ids keep the old session semantics); (3) the dashboard attaches a lane's live worker telemetry to the task named by the lane snapshot — including a row still badged `held` — never to a sibling that merely has `status: running`. - **Stale `⏸️ Lane held` alert after a fast ruling.** The escalation reaches the supervisor live, so a ruling is often queued before the worker exits; the hold loop published \"Lane held\" and then consumed that ruling on its first poll. diff --git a/dashboard/public/app.js b/dashboard/public/app.js index 87c6c902..ed5df08d 100644 --- a/dashboard/public/app.js +++ b/dashboard/public/app.js @@ -990,7 +990,15 @@ function renderLanesTasks(batch, sessions) { // so allow a task-status fallback while still avoiding duplicate rows. const reviewerActive = isReviewerActiveForTask(ls, task); const telemBadges = task.status !== "pending" ? telemetryBadgesHtml(tel, reviewerActive) : ""; - if (ls && ls.workerStatus === "running" && task.status === "running") { + // #651: the lane snapshot names the task its worker is running. Attach the + // lane's live numbers to THAT task (even if its batch-state badge is stale, + // e.g. still `held` right after a ruling), never to a sibling whose badge + // happens to say running. + const snapshotOwnsTask = !ls || !ls.taskId || ls.taskId === task.taskId; + const workerActiveHere = + !!ls && ls.workerStatus === "running" && snapshotOwnsTask && + (task.status === "running" || (task.status === "held" && ls.taskId === task.taskId)); + if (workerActiveHere) { const elapsed = ls.workerElapsed ? `${Math.round(ls.workerElapsed / 1000)}s` : ""; const tools = ls.workerToolCount || 0; const ctx = ls.workerContextPct ? `${Math.round(ls.workerContextPct)}%` : ""; @@ -1005,6 +1013,7 @@ function renderLanesTasks(batch, sessions) { workerHtml += telemBadges; workerHtml += ``; } else if (!ls && tel && task.status === "running") { + // (no snapshot to attribute by — early startup only) // Running task with telemetry but no lane-state yet (early startup) const lastTool = tel.lastTool || ""; workerHtml = `
`; diff --git a/extensions/taskplane/engine.ts b/extensions/taskplane/engine.ts index de019afc..5c26f6c4 100644 --- a/extensions/taskplane/engine.ts +++ b/extensions/taskplane/engine.ts @@ -2691,16 +2691,19 @@ export async function executeOrchBatch( // batch state STRICTLY (throw on failure) before the lane-runner acts on // them — never the best-effort persistRuntimeState path. if (!batchState.holds) batchState.holds = []; - const holdStore = createHoldStore(batchState, (reason) => - persistRuntimeStateStrict( - reason, - batchState, - wavePlan, - latestAllocatedLanes, - allTaskOutcomes, - discoveryRef, - stateRoot, - ), + const holdStore = createHoldStore( + batchState, + (reason) => + persistRuntimeStateStrict( + reason, + batchState, + wavePlan, + latestAllocatedLanes, + allTaskOutcomes, + discoveryRef, + stateRoot, + ), + { outcomes: () => allTaskOutcomes }, // #651: status projection in the same write ); // TP-029: Track all repo roots encountered during execution. // Maps repoRoot → repoId (undefined for primary/repo-mode). diff --git a/extensions/taskplane/hold-state.ts b/extensions/taskplane/hold-state.ts index 1d6de96d..3919ad60 100644 --- a/extensions/taskplane/hold-state.ts +++ b/extensions/taskplane/hold-state.ts @@ -19,7 +19,7 @@ import { existsSync, readdirSync, readFileSync } from "fs"; import { join } from "path"; -import type { MailboxMessage } from "./types.ts"; +import type { LaneTaskOutcome, MailboxMessage } from "./types.ts"; import { isValidMailboxMessage } from "./mailbox.ts"; // ── Types ───────────────────────────────────────────────────────────── @@ -634,16 +634,30 @@ export class HoldPersistenceError extends Error { */ export function createHoldStore( owner: { holds?: HoldRecord[] }, + /** Durable whole-state write; throws on failure. */ persist: (reason: string) => void, + opts: { + /** + * #651: the live outcome array `serializeBatchState` reads task statuses + * from. When supplied, every commit projects the bound task's status from + * the post-transition hold table in the SAME write, and rolls the + * projection back with the hold table if the write fails. + */ + outcomes?: () => LaneTaskOutcome[]; + } = {}, ): HoldStore { - const commit = (next: HoldRecord[], reason: string, escalationId: string): void => { + const commit = (next: HoldRecord[], reason: string, record: HoldRecord): void => { const prev = owner.holds; owner.holds = next; + const projection = opts.outcomes + ? projectHoldStateOntoOutcomes(opts.outcomes(), next, record.taskId) + : { changed: false, restore: () => {} }; try { persist(reason); } catch (err) { owner.holds = prev; - throw new HoldPersistenceError(escalationId, err); + projection.restore(); + throw new HoldPersistenceError(record.escalationId, err); } }; return { @@ -651,7 +665,7 @@ export function createHoldStore( open: (record) => { const current = owner.holds ?? []; if (current.some((h) => h.escalationId === record.escalationId)) return; // idempotent - commit([...current, { ...record }], `hold-open:${record.escalationId}`, record.escalationId); + commit([...current, { ...record }], `hold-open:${record.escalationId}`, record); }, update: (record) => { const current = owner.holds ?? []; @@ -661,12 +675,75 @@ export function createHoldStore( commit( upsertHold(current, { ...record }), `hold-${record.phase}:${record.escalationId}`, - record.escalationId, + record, ); }, }; } +/** + * #651: project the unit's hold state onto the task outcome that + * `serializeBatchState` reads statuses from. Evaluated against the POST-transition + * hold table (Sage: a unit with two open holds is still held when only one is + * ruled), never against a single record: + * - any OPEN hold binding the task → `held` + * - none open, some released & unacknowledged → `running` (worker relaunched to deliver) + * - otherwise → untouched + * Never downgrades a terminal outcome. Returns the previous field values so a + * failed persist can restore them (the store does this — projection and hold + * write are one transaction). + */ +export function projectHoldStateOntoOutcomes( + outcomes: LaneTaskOutcome[], + holds: readonly HoldRecord[], + taskId: string, +): { changed: boolean; restore: () => void } { + const noop = { changed: false, restore: () => {} }; + const existing = outcomes.find((o) => o.taskId === taskId); + if (!existing) return noop; + if ( + existing.status === "succeeded" || + existing.status === "failed" || + existing.status === "stalled" || + existing.status === "skipped" + ) { + return noop; + } + const mine = holdsForTask(holds, taskId); + const open = mine.filter((h) => h.phase === "open"); + const releasedUnacked = mine.filter( + (h) => h.phase === "released" && h.deliveryState !== "acknowledged", + ); + const prev = { + status: existing.status, + exitReason: existing.exitReason, + endTime: existing.endTime, + }; + const restore = () => { + existing.status = prev.status; + existing.exitReason = prev.exitReason; + existing.endTime = prev.endTime; + }; + let want: { status: LaneTaskOutcome["status"]; exitReason: string } | null = null; + if (open.length > 0) { + want = { + status: "held", + exitReason: `Held — awaiting ruling on ${open.map((h) => h.escalationId).join(", ")}`, + }; + } else if (releasedUnacked.length > 0) { + want = { + status: "running", + exitReason: `Ruling ${releasedUnacked.map((h) => h.ruling?.id ?? "?").join(", ")} accepted — worker relaunched`, + }; + } + if (!want) return noop; + if (existing.status === want.status && existing.exitReason === want.exitReason) return noop; + existing.status = want.status; + existing.exitReason = want.exitReason; + existing.endTime = null; + return { changed: true, restore }; +} + /** Volatile store for tests and legacy callers that run without an engine. */ export function createInMemoryHoldStore(initial: HoldRecord[] = []): HoldStore { return createHoldStore({ holds: [...initial] }, () => {}); diff --git a/extensions/taskplane/resume.ts b/extensions/taskplane/resume.ts index 6080a37e..a58a1e39 100644 --- a/extensions/taskplane/resume.ts +++ b/extensions/taskplane/resume.ts @@ -1010,9 +1010,17 @@ export function reconcileTaskStates( * ruling arrives. */ holdBlockedTaskIds: ReadonlySet = new Set(), + /** + * #651: when the registry names the task each alive worker is running, + * "alive" is decided per TASK, not per lane session. Undefined = legacy + * registry without task ids → fall back to session identity. + */ + aliveTaskIds?: ReadonlySet, ): ReconciledTaskState[] { return persistedState.tasks.map((task) => { - const sessionAlive = aliveSessions.has(task.sessionName); + const sessionAlive = aliveTaskIds + ? aliveTaskIds.has(task.taskId) + : aliveSessions.has(task.sessionName); const doneFileFound = doneTaskIds.has(task.taskId); const worktreeExists = existingWorktrees.has(task.taskId); @@ -1820,10 +1828,17 @@ export async function resumeOrchBatch( // TP-112/119: Runtime V2 session liveness check only. // Alive sessions are discovered from the process registry. const aliveSessions = new Set(); + // #651: tasks whose OWN worker is alive. Two tasks on one serial lane share + // the lane's sessionName, so "lane session alive" must not promote a + // not-yet-started successor to running/reconnect. + const aliveTaskIds = new Set(); + let manifestsCarryTaskIds = false; const registry = readRegistrySnapshot(stateRoot, persistedState.batchId); if (registry) { for (const manifest of Object.values(registry.agents)) { + if (typeof manifest.taskId === "string" && manifest.taskId) manifestsCarryTaskIds = true; if (!isTerminalStatus(manifest.status) && isProcessAlive(manifest.pid)) { + if (typeof manifest.taskId === "string" && manifest.taskId) aliveTaskIds.add(manifest.taskId); aliveSessions.add(manifest.agentId); // Also add lane session name (without role suffix) so reconciliation // matches persisted task.sessionName. @@ -1874,6 +1889,7 @@ export async function resumeOrchBatch( doneTaskIds, existingWorktreeTaskIds, holdBlockedTaskIds, + manifestsCarryTaskIds ? aliveTaskIds : undefined, ); // ── 4b. Clear stale session allocation for tasks reconciled as pending ── @@ -2118,16 +2134,20 @@ export async function resumeOrchBatch( outcomes: () => preWaveOutcomes, discovery: () => preWaveDiscovery, }; - const holdStore = createHoldStore(batchState, (reason) => - persistRuntimeStateStrict( - reason, - batchState, - holdPersistCtx.wavePlan(), - holdPersistCtx.lanes(), - holdPersistCtx.outcomes(), - holdPersistCtx.discovery(), - stateRoot, - ), + const holdStore = createHoldStore( + batchState, + (reason) => + persistRuntimeStateStrict( + reason, + batchState, + holdPersistCtx.wavePlan(), + holdPersistCtx.lanes(), + holdPersistCtx.outcomes(), + holdPersistCtx.discovery(), + stateRoot, + ), + // #651: project status onto whichever outcome set this checkpoint serializes. + { outcomes: () => holdPersistCtx.outcomes() }, ); // Carry forward unknown fields for roundtrip preservation if (persistedState._extraFields) { diff --git a/extensions/tests/held-state-recovery.test.ts b/extensions/tests/held-state-recovery.test.ts index b2f1f7c9..034a1384 100644 --- a/extensions/tests/held-state-recovery.test.ts +++ b/extensions/tests/held-state-recovery.test.ts @@ -19,7 +19,11 @@ import { reconcileTaskStates, replayUnrecordedEscalations, } from "../taskplane/resume.ts"; -import { reconstructHoldsFromMailbox } from "../taskplane/hold-state.ts"; +import { + applyRuling as applyRulingForProjection, + projectHoldStateOntoOutcomes, + reconstructHoldsFromMailbox, +} from "../taskplane/hold-state.ts"; import { quarantineUnauthorizedDoneMarkers } from "../taskplane/resume.ts"; import { readOutboxStrict } from "../taskplane/mailbox.ts"; import { loadBatchState, persistRuntimeStateStrict } from "../taskplane/persistence.ts"; @@ -147,7 +151,7 @@ describe("#627 — wiring: engine, resume, extension", () => { it("engine: hold store is strict, threaded into every executeWave call, and unresolved holds preserve worktrees", () => { expect(engine).toContain( - "const holdStore = createHoldStore(batchState, (reason) => persistRuntimeStateStrict(", + "const holdStore = createHoldStore( batchState, (reason) => persistRuntimeStateStrict(", ); const calls = engine.split("await executeWave(").length - 1; const passes = engine.split("onLaneRespawned ?? undefined, holdStore, );").length - 1; @@ -193,7 +197,9 @@ describe("#627 — wiring: engine, resume, extension", () => { expect(pinIdx).toBeLessThan(reconcileIdx); expect(resume).toContain('throw new ResumeError("RESUME_INVALID_STATE", replay.error);'); expect(resume).toContain("saveBatchState(JSON.stringify(persistedState, null, 2), stateRoot);"); - expect(resume).toContain("existingWorktreeTaskIds, holdBlockedTaskIds, );"); + expect(resume).toContain( + "existingWorktreeTaskIds, holdBlockedTaskIds, manifestsCarryTaskIds ? aliveTaskIds : undefined, );", + ); expect(resume).toContain( "await Promise.all( [...reExecByLane.values()].map(async (laneTasks) => { for (const task of laneTasks) { await reExecuteOne(task); } }), );", ); @@ -873,3 +879,106 @@ describe("#627 — Sage round 3: strict checkpoint keeps every task's folder and expect(src).toContain("discovery: () => preWaveDiscovery,"); }); }); + +describe("#651 — hold state projects task status (transactional); alive-by-task reconciliation; dashboard attribution", () => { + const rul = (id: string, replyTo: string) => + ({ id, replyTo, content: "go", actor: { role: "supervisor", id: "s" } }) as const; + const outcome = (taskId: string, status: string, exitReason = "Task in progress"): any => ({ + taskId, + status, + exitReason, + startTime: 1, + endTime: null, + sessionName: "s", + doneFileFound: false, + }); + + it("projection from the POST-transition table: open → held; one of two holds ruled → still held; last ruled → running with the stale hold reason cleared; ack → untouched; terminal → untouched", () => { + const a = hold({ escalationId: "A" }); + const b = hold({ escalationId: "B" }); + const outcomes = [outcome("TP-1", "running"), outcome("TP-D", "succeeded", "done")]; + expect(projectHoldStateOntoOutcomes(outcomes, [a, b], "TP-1").changed).toBe(true); + expect(outcomes[0].status).toBe("held"); + expect(outcomes[0].exitReason).toBe("Held — awaiting ruling on A, B"); + // Sage blocker: only A ruled — B still binds the unit + const aRel = applyRulingForProjection(a, rul("rA", "A"), 5); + outcomes[0].exitReason = "Hold timeout: … the hold remains open"; + const r1 = projectHoldStateOntoOutcomes(outcomes, [aRel, b], "TP-1"); + expect(outcomes[0].status).toBe("held"); + expect(r1.changed).toBe(true); // reason re-projected to the remaining open hold + expect(outcomes[0].exitReason).toBe("Held — awaiting ruling on B"); + // B ruled too → running + const bRel = applyRulingForProjection(b, rul("rB", "B"), 6); + expect(projectHoldStateOntoOutcomes(outcomes, [aRel, bRel], "TP-1").changed).toBe(true); + expect(outcomes[0].status).toBe("running"); + expect(outcomes[0].exitReason).toBe("Ruling rA, rB accepted — worker relaunched"); + expect(outcomes[0].endTime).toBe(null); + expect(projectHoldStateOntoOutcomes(outcomes, [aRel, bRel], "TP-1").changed).toBe(false); // idempotent + // terminal never downgraded + expect(projectHoldStateOntoOutcomes(outcomes, [hold({ taskId: "TP-D" })], "TP-D").changed).toBe( + false, + ); + expect(outcomes[1].status).toBe("succeeded"); + }); + + it("store: projection and hold write are ONE transaction — a failed release persist restores the hold AND the outcome; a later unrelated checkpoint sees the pre-failure status", () => { + const outcomes = [outcome("TP-1", "held", "Hold timeout: … the hold remains open")]; + let fail = false; + const writes: string[] = []; + const owner = { holds: [hold()] }; + const store = createHoldStore( + owner, + (reason) => { + if (fail) throw new Error("disk full"); + writes.push(`${reason}:${outcomes[0].status}`); + }, + { outcomes: () => outcomes }, + ); + fail = true; + expect(() => store.update(applyRulingForProjection(hold(), rul("r1", "esc-1"), 5))).toThrow( + /disk full/, + ); + expect(owner.holds[0].phase).toBe("open"); + expect(outcomes[0].status).toBe("held"); // Sage blocker: outcome rolled back with the table + expect(outcomes[0].exitReason).toBe("Hold timeout: … the hold remains open"); + fail = false; + store.update(applyRulingForProjection(hold(), rul("r1", "esc-1"), 5)); + expect(writes).toEqual(["hold-released:esc-1:running"]); + expect(outcomes[0].exitReason).toBe("Ruling r1 accepted — worker relaunched"); + // engine + resume pass their live outcome sets + const eng = readSrc("engine.ts").replace(/\s+/g, " "); + expect(eng).toContain("{ outcomes: () => allTaskOutcomes }, // #651"); + const res = readSrc("resume.ts").replace(/\s+/g, " "); + expect(res).toContain("{ outcomes: () => holdPersistCtx.outcomes() },"); + }); + + it("reconcile: two tasks sharing a lane session — only the task the alive worker manifest names is 'reconnect'; the successor stays pending (legacy registries without task ids keep session semantics)", () => { + const st = state([{ taskId: "TP-A" }, { taskId: "TP-B", status: "pending" }]); + const alive = new Set(["orch-op-lane-1", "orch-op-lane-1-worker"]); + const r = reconcileTaskStates( + st, + alive, + new Set(), + new Set(["TP-A", "TP-B"]), + new Set(), + new Set(["TP-A"]), + ); + expect(r[0].action).toBe("reconnect"); + expect(r[1].action).not.toBe("reconnect"); + expect(r[1].liveStatus).toBe("pending"); + const legacy = reconcileTaskStates(st, alive, new Set(), new Set(["TP-A", "TP-B"]), new Set()); + expect(legacy[1].action).toBe("reconnect"); + expect(readSrc("resume.ts").replace(/\s+/g, " ")).toContain( + "manifestsCarryTaskIds ? aliveTaskIds : undefined,", + ); + }); + + it("dashboard: worker telemetry attaches to the lane snapshot's taskId (held row included), never to a sibling badged running", () => { + const app = readSrc("../../dashboard/public/app.js").replace(/\s+/g, " "); + expect(app).toContain("const snapshotOwnsTask = !ls || !ls.taskId || ls.taskId === task.taskId;"); + expect(app).toContain( + '(task.status === "running" || (task.status === "held" && ls.taskId === task.taskId))', + ); + expect(app).toContain("if (workerActiveHere) {"); + }); +});