From 5cc093b7f3f92c156cf57384a68c3c9c0a4867eb Mon Sep 17 00:00:00 2001 From: vitry Date: Tue, 1 Sep 2026 00:43:34 +0800 Subject: [PATCH 1/9] docs: specify permission turn lifecycle fix --- ...e-0165-permission-turn-lifecycle-design.md | 57 +++++++++++++++++++ 1 file changed, 57 insertions(+) create mode 100644 docs/superpowers/specs/2026-09-01-zcode-0165-permission-turn-lifecycle-design.md diff --git a/docs/superpowers/specs/2026-09-01-zcode-0165-permission-turn-lifecycle-design.md b/docs/superpowers/specs/2026-09-01-zcode-0165-permission-turn-lifecycle-design.md new file mode 100644 index 0000000..c594cf7 --- /dev/null +++ b/docs/superpowers/specs/2026-09-01-zcode-0165-permission-turn-lifecycle-design.md @@ -0,0 +1,57 @@ +# ZCode 0.16.5 Permission Turn Lifecycle Design + +## Context + +The executor already treats ZCode 0.16.5 `state.updated` notifications with reason `prompt_completed` as legacy liveness signals. True completion is established by the v4 conversation observer or a coherent current-turn snapshot. However, `executeJob` obtains the liveness signal through `client.waitForCompletion()`, whose public contract consumes completion and deletes the protocol client's active-turn state. + +ZCode 0.16.5 can emit that legacy notification immediately after admission, then request tool permission while the real runtime turn is still active. Once the legacy waiter deletes the turn, the permission request fails the exact-active-turn check and is returned as JSON-RPC `-32000` (`ZCODE_PERMISSION_SESSION_INVALID`). This is independent of the configured Codex permission mode: the incident job already carried `bypassPermissions`. + +## Goals + +- Keep the exact turn armed after an admission-time legacy completion wake so later permission requests can be evaluated normally. +- Preserve the current destructive behavior of `waitForCompletion()` for every existing caller. +- Clear local turn state when `executeJob` reaches its real terminal or cleanup boundary. +- Cover the captured 0.16.5 event ordering with deterministic regression tests. + +## Non-goals + +- No change to `decidePermission`, permission snapshots, risk mapping, or offered-response validation. +- No new appserver flag, permission field, broker authorization rule, or persisted schema. +- No reinterpretation of legacy completion as authoritative success or failure. +- No broad refactor of the protocol or executor lifecycle. + +## Design + +### Non-destructive legacy observation + +Add a narrowly named protocol/client operation for observing the next validated completion notification without consuming the active turn. It must apply the same session, timeout, active-turn, duplicate-waiter, and `isCompletionFor` validation used by `waitForCompletion()`, but resolution must not call `abortTurn()` and must not consume turn ownership. + +The existing `waitForCompletion()` remains unchanged in observable behavior: queued or live completion resolution consumes the turn, while timeout and cancellation retain their present destructive cleanup semantics. + +Only `executeJob` switches its `legacyWake` construction to the non-destructive observer. The coordinator continues to use the wake solely to trigger authoritative v4/snapshot reconciliation. + +### Explicit executor cleanup + +Because the wake no longer consumes local state, `executeJob` must explicitly release the protocol turn after the authoritative lifecycle has finished. Cleanup belongs in the executor's existing unconditional teardown, after any terminal/cancellation reconciliation that may still need permission and turn identity, and before client close completes. + +The cleanup operation is local and idempotent. It must not send `session/stop`, alter durable job state, or replace the existing cancellation paths. Successful terminal, provider failure, remote interruption, local abort, timeout, and error cleanup all converge on the same local turn release when a session was created or resumed. + +### Safety invariants + +- A validated early legacy completion leaves `turnState(sessionId) === 'armed'`. +- A later permission request for that session reaches the configured handler and returns one offered response. +- Authoritative executor teardown leaves `turnState(sessionId) === null`. +- Ordinary `waitForCompletion()` still leaves `turnState(sessionId) === null` immediately after resolution. +- Permission policy and durable job outcomes are unchanged. + +## Testing + +1. Add a protocol/client regression proving non-destructive observation preserves the armed turn and permits a subsequent permission request. +2. Retain or strengthen the destructive waiter assertion so its compatibility contract is explicit. +3. Add an executor regression with captured 0.16.5 ordering: admission, early legacy wake, later permission request, authoritative terminal, successful result, and final local turn cleanup. +4. Exercise cleanup on a non-success path so an observer cannot leave an armed turn behind. +5. Run focused tests, then `npm run check` before review and PR creation. + +## Rollout and compatibility + +This is an internal additive API and a one-call-site migration. No user configuration or data migration is required. Older ZCode versions continue to produce the same wake signal, and all callers outside `executeJob` retain existing semantics. From 9dd62372b54829f53e0b18526187f109b906675e Mon Sep 17 00:00:00 2001 From: vitry Date: Tue, 1 Sep 2026 00:47:04 +0800 Subject: [PATCH 2/9] docs: plan permission turn lifecycle implementation --- ...01-zcode-0165-permission-turn-lifecycle.md | 218 ++++++++++++++++++ ...e-0165-permission-turn-lifecycle-design.md | 2 + 2 files changed, 220 insertions(+) create mode 100644 docs/superpowers/plans/2026-09-01-zcode-0165-permission-turn-lifecycle.md diff --git a/docs/superpowers/plans/2026-09-01-zcode-0165-permission-turn-lifecycle.md b/docs/superpowers/plans/2026-09-01-zcode-0165-permission-turn-lifecycle.md new file mode 100644 index 0000000..bbb108e --- /dev/null +++ b/docs/superpowers/plans/2026-09-01-zcode-0165-permission-turn-lifecycle.md @@ -0,0 +1,218 @@ +# ZCode 0.16.5 Permission Turn Lifecycle Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Preserve the active protocol turn across an admission-time legacy completion wake so later ZCode permission requests succeed, then release the local turn at the executor's authoritative cleanup boundary. + +**Architecture:** Add a non-destructive completion observer alongside the existing destructive waiter in the protocol/client layer. Migrate only `executeJob` to that observer and explicitly release its local turn during unconditional teardown; retain compatibility fallbacks for injected test clients that expose only the historical interface. + +**Tech Stack:** Node.js 22.13, ECMAScript modules, `node:test`, the existing JSON-RPC protocol client and job executor. + +--- + +## File map + +- Modify `scripts/lib/zcode-protocol.mjs`: share completion validation/wait registration while distinguishing destructive consumption from observation; add local turn release behavior. +- Modify `scripts/lib/zcode-client.mjs`: expose `observeCompletion()` and `releaseTurn()` without changing `waitForCompletion()`. +- Modify `scripts/lib/review.mjs`: use non-destructive observation for `legacyWake` and release the local turn during teardown. +- Modify `tests/process-zcode.test.mjs`: cover low-level observer, permission, timeout, release, and destructive-wait invariants. +- Modify `tests/job-control.test.mjs`: cover captured 0.16.5 executor ordering and success/error cleanup. + +### Task 1: Add non-destructive completion observation + +**Files:** +- Modify: `scripts/lib/zcode-protocol.mjs:95-145` +- Modify: `scripts/lib/zcode-client.mjs:125-135` +- Test: `tests/process-zcode.test.mjs` + +- [ ] **Step 1: Write failing protocol tests** + +Add tests that construct `ZCodeProtocolClient` with `PassThrough` streams, arm a turn, start `observeCompletion(sessionId)`, emit a matching `prompt_completed`, and assert: + +```js +const completion = protocol.observeCompletion(sessionId); +protocol.handleLine(JSON.stringify({ method: 'state.updated', params: matchingCompletion })); +assert.equal((await completion).reason, 'prompt_completed'); +assert.equal(protocol.turnState(sessionId), 'armed'); +``` + +Then emit `interaction/requestPermission` for the same session, return an offered allow response from the handler, and verify the protocol writes the allow result instead of `ZCODE_PERMISSION_SESSION_INVALID`. Add companion assertions that `waitForCompletion()` still clears the turn, observer timeout leaves it armed, and `releaseTurn()` clears it and rejects any still-pending observer. + +- [ ] **Step 2: Run the focused tests and verify RED** + +Run: + +```bash +node --test --test-name-pattern='non-destructive completion|completion observer|destructive completion' tests/process-zcode.test.mjs +``` + +Expected: FAIL because `observeCompletion` and `releaseTurn` do not exist. + +- [ ] **Step 3: Implement the minimal protocol behavior** + +Refactor the current waiter registration into one internal path with an explicit consumption mode. Preserve the public destructive path exactly, and add: + +```js +observeCompletion(sessionId, timeoutMs) { + return this.waitForCompletionMode(sessionId, timeoutMs, false); +} + +releaseTurn(sessionId) { + if (!nonEmpty(sessionId)) throw protocolInputError(); + this.cancelTurn(sessionId, localTurnReleasedError(sessionId)); +} +``` + +For observation mode, peek at an already queued completion rather than shifting it, do not call `abortTurn()` on resolution or timeout, and always unregister the observer. For destructive mode, keep the current shift, timeout cleanup, and `abortTurn()` behavior. Continue enforcing one waiter/observer per session with `waiterSessions`. + +Expose the two operations from `ZCodeClient` with normal session-ID validation/documentation: + +```js +observeCompletion(sessionId, timeoutMs) { + requireSessionId(sessionId); + return this.protocol.observeCompletion(sessionId, timeoutMs); +} + +releaseTurn(sessionId) { + requireSessionId(sessionId); + this.protocol.releaseTurn(sessionId); +} +``` + +- [ ] **Step 4: Run focused and adjacent tests and verify GREEN** + +Run: + +```bash +node --test tests/process-zcode.test.mjs tests/zcode-client.test.mjs +``` + +Expected: PASS, including all existing destructive completion and permission tests. + +- [ ] **Step 5: Self-review and commit** + +Check that no existing call site changed and that observer cleanup cannot retain a timer, subscriber, or waiter-session entry. Then commit: + +```bash +git add scripts/lib/zcode-protocol.mjs scripts/lib/zcode-client.mjs tests/process-zcode.test.mjs +git commit -m "fix: observe legacy completion without ending turn" +``` + +### Task 2: Migrate executor wake and own local cleanup + +**Files:** +- Modify: `scripts/lib/review.mjs:220-360` +- Test: `tests/job-control.test.mjs` + +- [ ] **Step 1: Write the captured-ordering regression test** + +Extend the existing `0.16.5 foreground execution treats legacy completion as admission` fixture client with `observeCompletion`, `releaseTurn`, and a permission handler capture. Make `observeCompletion` publish the legacy wake first, then invoke the captured permission handler with a medium-risk Write request offering allow/deny. Assert the handler returns allow while the executor remains running, then publish the v4 authoritative terminal and assert success plus one local release: + +```js +assert.deepEqual(permissionDecision, { decision: 'allow' }); +assert.equal(releaseTurnCalls, 1); +assert.equal(releasedSessionId, sessionId); +``` + +Add an error-path test where observation or authoritative reconciliation fails after admission; assert `releaseTurn(sessionId)` still runs once before `close()`. + +- [ ] **Step 2: Run the executor regressions and verify RED** + +Run: + +```bash +node --test --test-name-pattern='0.16.5 foreground execution|releases local turn' tests/job-control.test.mjs +``` + +Expected: FAIL because `executeJob` still calls destructive `waitForCompletion()` and never releases the local turn explicitly. + +- [ ] **Step 3: Implement the executor migration** + +Construct `legacyWake` from `client.observeCompletion(activeSessionId)` when available. Keep a fallback to `client.waitForCompletion(activeSessionId)` only for existing injected test doubles that predate the internal interface: + +```js +const observeLegacyCompletion = typeof client.observeCompletion === 'function' + ? client.observeCompletion.bind(client) + : client.waitForCompletion.bind(client); +const legacyWake = waitForCompletionOrAbort(observeLegacyCompletion(activeSessionId), input.signal); +``` + +In unconditional teardown, after all terminal/cancellation/error reconciliation and progress cleanup but before `client.close()`, release the known local session exactly once when supported: + +```js +try { + if (sessionId && typeof client.releaseTurn === 'function') client.releaseTurn(sessionId); +} catch (cleanupError) { + if (!primaryError) primaryError = cleanupError; +} +``` + +Preserve the primary-error and cleanup-error conventions already used by the executor. Do not send an upstream stop from this release path and do not modify `decidePermission`. + +- [ ] **Step 4: Run focused and executor-adjacent tests and verify GREEN** + +Run: + +```bash +node --test tests/job-control.test.mjs tests/integration/companion.test.mjs +``` + +Expected: PASS with the captured ordering, cleanup tests, and existing cancellation behavior unchanged. + +- [ ] **Step 5: Self-review and commit** + +Inspect the diff for exactly one production caller migration, one unconditional local cleanup, and no permission-policy changes. Then commit: + +```bash +git add scripts/lib/review.mjs tests/job-control.test.mjs +git commit -m "fix: retain active turn through legacy wake" +``` + +### Task 3: Verify contracts and release readiness + +**Files:** +- Modify if required by generated parity checks: checked-in marketplace mirrors only through the repository's existing builder +- Test: repository-wide verification + +- [ ] **Step 1: Run permission-policy and protocol contract tests** + +Run: + +```bash +node --test --test-name-pattern='permission|completion' tests/process-zcode.test.mjs tests/zcode-client.test.mjs tests/job-control.test.mjs +``` + +Expected: PASS; Rescue permission decisions remain unchanged and destructive completion callers retain their contract. + +- [ ] **Step 2: Run the full repository check** + +Run: + +```bash +npm run check +``` + +Expected: PASS for line endings, lint, typecheck, all tests, qualification tests, and marketplace parity/build checks. + +- [ ] **Step 3: Inspect final scope** + +Run: + +```bash +git diff --check origin/main...HEAD +git diff --stat origin/main...HEAD +git diff origin/main...HEAD -- scripts/lib/review.mjs scripts/lib/zcode-client.mjs scripts/lib/zcode-protocol.mjs +``` + +Expected: no whitespace errors; changes remain limited to the completion lifecycle, tests, and approved docs. + +- [ ] **Step 4: Commit any verification-only generated parity update** + +If and only if the repository's official check regenerates tracked marketplace parity files, review and commit those exact generated changes: + +```bash +git add marketplace +git commit -m "build: refresh marketplace snapshot" +``` + +If there are no generated tracked changes, skip this commit. diff --git a/docs/superpowers/specs/2026-09-01-zcode-0165-permission-turn-lifecycle-design.md b/docs/superpowers/specs/2026-09-01-zcode-0165-permission-turn-lifecycle-design.md index c594cf7..b9e6387 100644 --- a/docs/superpowers/specs/2026-09-01-zcode-0165-permission-turn-lifecycle-design.md +++ b/docs/superpowers/specs/2026-09-01-zcode-0165-permission-turn-lifecycle-design.md @@ -26,6 +26,8 @@ ZCode 0.16.5 can emit that legacy notification immediately after admission, then Add a narrowly named protocol/client operation for observing the next validated completion notification without consuming the active turn. It must apply the same session, timeout, active-turn, duplicate-waiter, and `isCompletionFor` validation used by `waitForCompletion()`, but resolution must not call `abortTurn()` and must not consume turn ownership. +An observer timeout removes only that observer; it does not silently acquire authority to end the turn. Executor teardown remains responsible for local release. Add an idempotent client-level local release operation that cancels outstanding local completion observation and clears the protocol turn without sending an upstream stop request. + The existing `waitForCompletion()` remains unchanged in observable behavior: queued or live completion resolution consumes the turn, while timeout and cancellation retain their present destructive cleanup semantics. Only `executeJob` switches its `legacyWake` construction to the non-destructive observer. The coordinator continues to use the wake solely to trigger authoritative v4/snapshot reconciliation. From 0deaa7424915bf6e99263e07b471130db8a84f8d Mon Sep 17 00:00:00 2001 From: vitry Date: Tue, 1 Sep 2026 00:56:26 +0800 Subject: [PATCH 3/9] fix: observe legacy completion without ending turn --- scripts/lib/zcode-client.mjs | 2 + scripts/lib/zcode-protocol.mjs | 44 ++++++++++++++++-- tests/process-zcode.test.mjs | 85 ++++++++++++++++++++++++++++++++++ 3 files changed, 127 insertions(+), 4 deletions(-) diff --git a/scripts/lib/zcode-client.mjs b/scripts/lib/zcode-client.mjs index f22d538..389740f 100644 --- a/scripts/lib/zcode-client.mjs +++ b/scripts/lib/zcode-client.mjs @@ -129,6 +129,8 @@ export class ZCodeClient { } /** Wait for a validated terminal notification; no deadline applies unless configured on the client or supplied here. @param {string} sessionId @param {number} [timeoutMs] */ waitForCompletion(sessionId, timeoutMs) { return this.protocol.waitForCompletion(sessionId, timeoutMs); } + /** Observe a validated terminal notification without consuming the active turn. @param {string} sessionId @param {number} [timeoutMs] */ observeCompletion(sessionId, timeoutMs) { requireSessionId(sessionId); return this.protocol.observeCompletion(sessionId, timeoutMs); } + /** Locally release an active turn without sending an upstream request. @param {string} sessionId */ releaseTurn(sessionId) { requireSessionId(sessionId); this.protocol.releaseTurn(sessionId); } /** Exact local protocol invariant used to prove whether this client owns an active turn. @param {string} sessionId */ turnState(sessionId) { requireSessionId(sessionId); return this.protocol.turnState(sessionId); } /** @param {string} sessionId @param {{connectionId:string,clientMode:'desktop-continuous'|'web-remote-replayable'}} options */ async subscribeConversation(sessionId, options) { diff --git a/scripts/lib/zcode-protocol.mjs b/scripts/lib/zcode-protocol.mjs index e2455e7..c902878 100644 --- a/scripts/lib/zcode-protocol.mjs +++ b/scripts/lib/zcode-protocol.mjs @@ -37,6 +37,7 @@ export class ZCodeProtocolClient { /** @type {Set>} */ this.serverTasks = new Set(); /** @type {Set} */ this.serverTaskControllers = new Set(); /** @type {Map} */ this.serverTaskSessions = new Map(); + /** @type {Map>} */ this.serverTasksByController = new Map(); this.nextId = 1; this.buffer = ''; this.closed = false; @@ -102,9 +103,12 @@ export class ZCodeProtocolClient { /** @param {string} sessionId @param {PluginError} [error] */ cancelTurn(sessionId, error = new PluginError('ZCODE_SESSION_STOPPED', `ZCode session ${sessionId} was stopped.`, { category: 'state', remedy: 'Start a new turn before waiting for completion.', details: { sessionId } })) { for (const waiter of this.completionWaiters) if (waiter.sessionId === sessionId) { if (waiter.timer) clearTimeout(waiter.timer); waiter.unsubscribe(); this.completionWaiters.delete(waiter); this.waiterSessions.delete(sessionId); waiter.reject(error); } - for (const [controller, taskSessionId] of this.serverTaskSessions) if (taskSessionId === sessionId) controller.abort(); + for (const [controller, taskSessionId] of this.serverTaskSessions) if (taskSessionId === sessionId) { controller.abort(); const task = this.serverTasksByController.get(controller); if (task) this.serverTasks.delete(task); this.serverTaskControllers.delete(controller); this.serverTaskSessions.delete(controller); this.serverTasksByController.delete(controller); } + for (const key of this.permissionRequestIds.keys()) { try { if (JSON.parse(key)?.[0] === sessionId) this.permissionRequestIds.delete(key); } catch { /* internal replay keys are always JSON */ } } this.abortTurn(sessionId); } + /** Locally release a turn without sending an upstream request. @param {string} sessionId */ + releaseTurn(sessionId) { if (!nonEmpty(sessionId)) throw protocolInputError(); this.cancelTurn(sessionId); } /** @param {(message:any)=>void} handler */ subscribe(handler) { if (typeof handler !== 'function' || this.subscribers.size >= 256) throw protocolInputError(); this.subscribers.add(handler); return () => this.subscribers.delete(handler); } @@ -141,6 +145,37 @@ export class ZCodeProtocolClient { }); } + /** Observe a validated terminal notification without consuming it or ending the turn. @param {string} sessionId @param {number} [timeoutMs] */ + observeCompletion(sessionId, timeoutMs) { + const effectiveTimeoutMs = timeoutMs === undefined ? this.completionTimeoutMs : timeoutMs; + if (!nonEmpty(sessionId) || effectiveTimeoutMs !== undefined && (!Number.isSafeInteger(effectiveTimeoutMs) || effectiveTimeoutMs < 1 || effectiveTimeoutMs > 86_400_000) || this.turns.get(sessionId)?.status !== 'armed' || this.waiterSessions.has(sessionId)) return Promise.reject(protocolInputError()); + const queued = this.completed.get(sessionId)?.[0]; + if (queued) return Promise.resolve(queued); + return new Promise((resolve, reject) => { + let unsubscribe = () => {}; + this.waiterSessions.add(sessionId); const waiter = { reject, timer: /** @type {NodeJS.Timeout|null} */ (null), unsubscribe, sessionId }; + const timer = effectiveTimeoutMs === undefined ? null : setTimeout(() => { + this.completionWaiters.delete(waiter); this.waiterSessions.delete(sessionId); + unsubscribe(); + reject(new PluginError('ZCODE_COMPLETION_TIMEOUT', `ZCode session ${sessionId} did not complete in time.`, { category: 'timeout', remedy: 'Read or resume the session before retrying.', details: { sessionId, timeoutMs: effectiveTimeoutMs } })); + }, effectiveTimeoutMs); + waiter.timer = timer; + timer?.unref?.(); + try { + unsubscribe = this.subscribe((message) => { + if (!isCompletionFor(message, sessionId, this.turns.get(sessionId))) return; + this.completionWaiters.delete(waiter); this.waiterSessions.delete(sessionId); if (timer) clearTimeout(timer); unsubscribe(); resolve(message.params); + }); + } catch (error) { + if (timer) clearTimeout(timer); + this.completionWaiters.delete(waiter); this.waiterSessions.delete(sessionId); waiter.unsubscribe(); + reject(error); return; + } + waiter.unsubscribe = unsubscribe; + this.completionWaiters.add(waiter); + }); + } + close() { this.closePromise ??= this.closeOnce(); return this.closePromise; @@ -159,7 +194,7 @@ export class ZCodeProtocolClient { } const tasks = [...this.serverTasks]; if (tasks.length) await Promise.race([Promise.allSettled(tasks), boundedDelay(25)]); - this.serverTasks.clear(); this.serverTaskControllers.clear(); this.serverTaskSessions.clear(); + this.serverTasks.clear(); this.serverTaskControllers.clear(); this.serverTaskSessions.clear(); this.serverTasksByController.clear(); this.terminationPromise ??= terminateProcess(this.child); await this.terminationPromise; } @@ -241,6 +276,7 @@ export class ZCodeProtocolClient { if (this.permissionRequestIds.size >= 1024) throw new PluginError('ZCODE_PERMISSION_OVERFLOW', 'Too many permission requests were rejected.', { category: 'authorization', remedy: 'Restart the affected ZCode turn.' }); this.permissionRequestIds.set(replayKey, Date.now()); const result = this.permissionHandler ? await this.permissionHandler(message.params, signal) : message.params.options.find((/** @type {any} */ option) => option.response.decision === 'deny')?.response; + if (signal.aborted) return; validatePermissionResult(result); if (!message.params.options.some((/** @type {any} */ option) => JSON.stringify(option.response) === JSON.stringify(result))) throw new PluginError('ZCODE_PERMISSION_OPTION_INVALID', 'Permission response was not one of the offered options.', { category: 'authorization', remedy: 'Return an exact response offered by ZCode.' }); if (!this.closed) this.sendFrame({ id: message.id, result }); @@ -254,7 +290,7 @@ export class ZCodeProtocolClient { } /** @param {Promise} task @param {AbortController} controller @param {unknown} sessionId */ - trackServerTask(task, controller, sessionId) { this.serverTasks.add(task); this.serverTaskControllers.add(controller); if (nonEmpty(sessionId)) this.serverTaskSessions.set(controller, /** @type {string} */ (sessionId)); const cleanup = () => { this.serverTasks.delete(task); this.serverTaskControllers.delete(controller); this.serverTaskSessions.delete(controller); }; void task.then(cleanup, (error) => { cleanup(); this.fail(asDisconnected(error, this.stderrTail.value())); }); } + trackServerTask(task, controller, sessionId) { this.serverTasks.add(task); this.serverTaskControllers.add(controller); this.serverTasksByController.set(controller, task); if (nonEmpty(sessionId)) this.serverTaskSessions.set(controller, /** @type {string} */ (sessionId)); const cleanup = () => { this.serverTasks.delete(task); this.serverTaskControllers.delete(controller); this.serverTaskSessions.delete(controller); this.serverTasksByController.delete(controller); }; void task.then(cleanup, (error) => { cleanup(); this.fail(asDisconnected(error, this.stderrTail.value())); }); } /** @param {Record} frame */ sendFrame(frame) { @@ -268,7 +304,7 @@ export class ZCodeProtocolClient { if (this.closed) return; this.closed = true; for (const controller of this.serverTaskControllers) controller.abort(); - this.serverTasks.clear(); this.serverTaskControllers.clear(); this.serverTaskSessions.clear(); this.permissionRequestIds.clear(); + this.serverTasks.clear(); this.serverTaskControllers.clear(); this.serverTaskSessions.clear(); this.serverTasksByController.clear(); this.permissionRequestIds.clear(); this.stderrTail.close(); const diagnosticError = withStderr(error, this.stderrTail.value()); this.writer.close(); this.rejectPending(diagnosticError); this.rejectCompletionWaiters(diagnosticError); for (const timer of this.completionExpiry.values()) clearTimeout(timer); this.completionExpiry.clear(); this.completed.clear(); this.earlyCompletions.clear(); this.turns.clear(); diff --git a/tests/process-zcode.test.mjs b/tests/process-zcode.test.mjs index 0ed1408..3e00dd7 100644 --- a/tests/process-zcode.test.mjs +++ b/tests/process-zcode.test.mjs @@ -10,6 +10,7 @@ import { PassThrough } from 'node:stream'; import test from 'node:test'; import { drainExitedProcessStreams, runProcess, spawnProcess, terminateProcess } from '../scripts/lib/process.mjs'; +import { ZCodeClient } from '../scripts/lib/zcode-client.mjs'; import { BoundedWriter, RedactedTail, ZCodeProtocolClient } from '../scripts/lib/zcode-protocol.mjs'; const fakeFixture = fileURLToPath(new URL('./fixtures/fake-zcode-cli.mjs', import.meta.url)); @@ -235,6 +236,90 @@ test('subscriber failures are isolated and permission work cannot write after cl assert.equal(child.stdin.readableLength, beforeClose); }); +test('observed completion leaves the turn armed and a later permission request can be allowed', async () => { + const child = new EventEmitter(); child.stdin = new PassThrough(); child.stdout = new PassThrough(); child.stderr = new PassThrough(); child.exitCode = 0; child.signalCode = null; + const protocol = new ZCodeProtocolClient(child); + protocol.beginTurn('session-1'); protocol.armTurn('session-1', 1, 'input-1'); + const waiting = protocol.observeCompletion('session-1', 1_000); + protocol.handleLine(JSON.stringify({ method: 'state.updated', params: { scope: 'session', sessionId: 'session-1', revision: 2, reason: 'prompt_completed' } })); + assert.equal((await waiting).reason, 'prompt_completed'); + assert.equal(protocol.turnState('session-1'), 'armed'); + assert.equal(protocol.completed.get('session-1')?.length, 1, 'observation must not consume the queued completion'); + + let handled = 0; + protocol.setPermissionHandler(() => { handled += 1; return { decision: 'allow' }; }); + protocol.handleLine(JSON.stringify({ id: 99, method: 'interaction/requestPermission', params: { requestId: 'r', sessionId: 'session-1', toolCallId: 't', toolName: 'write', reason: 'test', riskLevel: 'low', input: {}, options: [{ optionId: 'allow', kind: 'allow', name: 'Allow', response: { decision: 'allow' } }, { optionId: 'deny', kind: 'deny', name: 'Deny', response: { decision: 'deny' } }] } })); + await new Promise((resolve) => setImmediate(resolve)); + const response = JSON.parse(child.stdin.read().toString()); + assert.equal(handled, 1); + assert.deepEqual(response, { id: 99, result: { decision: 'allow' } }); + protocol.releaseTurn('session-1'); +}); + +test('completion observer timeout unregisters without ending the active turn', async () => { + const child = new EventEmitter(); child.stdin = new PassThrough(); child.stdout = new PassThrough(); child.stderr = new PassThrough(); child.exitCode = 0; child.signalCode = null; + const protocol = new ZCodeProtocolClient(child); + protocol.beginTurn('session-1'); protocol.armTurn('session-1', 1, 'input-1'); + await assert.rejects(protocol.observeCompletion('session-1', 10), { code: 'ZCODE_COMPLETION_TIMEOUT' }); + assert.equal(protocol.turnState('session-1'), 'armed'); + assert.equal(protocol.completionWaiters.size, 0); + assert.equal(protocol.waiterSessions.size, 0); + assert.equal(protocol.subscribers.size, 0); + protocol.releaseTurn('session-1'); +}); + +test('releaseTurn is local and idempotent and rejects a pending observer', async () => { + const child = new EventEmitter(); child.stdin = new PassThrough(); child.stdout = new PassThrough(); child.stderr = new PassThrough(); child.exitCode = 0; child.signalCode = null; + const protocol = new ZCodeProtocolClient(child); const client = new ZCodeClient(protocol, '/repo'); + protocol.beginTurn('session-1'); protocol.armTurn('session-1', 1, 'input-1'); + const waiting = client.observeCompletion('session-1'); + assert.equal(child.stdin.readableLength, 0); + client.releaseTurn('session-1'); client.releaseTurn('session-1'); + await assert.rejects(waiting, { code: 'ZCODE_SESSION_STOPPED' }); + assert.equal(client.turnState('session-1'), null); + assert.equal(protocol.completionWaiters.size, 0); + assert.equal(protocol.waiterSessions.size, 0); + assert.equal(protocol.subscribers.size, 0); + assert.equal(child.stdin.readableLength, 0, 'local release must not send an upstream RPC'); + for (const invalid of ['', 'bad\nsession', null]) { + assert.throws(() => client.observeCompletion(invalid), { code: 'ZCODE_INPUT_INVALID' }); + assert.throws(() => client.releaseTurn(invalid), { code: 'ZCODE_INPUT_INVALID' }); + } +}); + +test('releaseTurn aborts and clears permission task state without writing a stale response', async () => { + const child = new EventEmitter(); child.stdin = new PassThrough(); child.stdout = new PassThrough(); child.stderr = new PassThrough(); child.exitCode = 0; child.signalCode = null; + const protocol = new ZCodeProtocolClient(child); + protocol.beginTurn('session-1'); protocol.armTurn('session-1', 1, 'input-1'); + let handlerSignal; + protocol.setPermissionHandler((_request, signal) => { + handlerSignal = signal; + return new Promise(() => {}); + }); + protocol.handleLine(JSON.stringify({ id: 99, method: 'interaction/requestPermission', params: { requestId: 'r', sessionId: 'session-1', toolCallId: 't', toolName: 'write', reason: 'test', riskLevel: 'low', input: {}, options: [{ optionId: 'deny', kind: 'deny', name: 'Deny', response: { decision: 'deny' } }] } })); + await new Promise((resolve) => setImmediate(resolve)); + assert.equal(protocol.permissionRequestIds.size, 1); + assert.equal(protocol.serverTaskSessions.size, 1); + protocol.releaseTurn('session-1'); + assert.equal(handlerSignal.aborted, true); + assert.equal(protocol.permissionRequestIds.size, 0); + assert.equal(protocol.serverTaskSessions.size, 0); + assert.equal(protocol.serverTaskControllers.size, 0); + assert.equal(protocol.serverTasks.size, 0); + assert.equal(child.stdin.readableLength, 0); +}); + +test('ordinary completion waiting remains destructive', async () => { + const child = new EventEmitter(); child.stdin = new PassThrough(); child.stdout = new PassThrough(); child.stderr = new PassThrough(); child.exitCode = 0; child.signalCode = null; + const protocol = new ZCodeProtocolClient(child); + protocol.beginTurn('session-1'); protocol.armTurn('session-1', 1, 'input-1'); + const waiting = protocol.waitForCompletion('session-1', 1_000); + protocol.handleLine(JSON.stringify({ method: 'state.updated', params: { scope: 'session', sessionId: 'session-1', revision: 2, reason: 'prompt_completed' } })); + assert.equal((await waiting).reason, 'prompt_completed'); + assert.equal(protocol.turnState('session-1'), null); + assert.equal(protocol.completed.has('session-1'), false); +}); + test('close aborts and detaches a never-settling permission task under strict rejections', async () => { const protocolUrl = new URL('../scripts/lib/zcode-protocol.mjs', import.meta.url).href; const source = ` From 4c219508198b19fe10805294f97ee03adbad6404 Mon Sep 17 00:00:00 2001 From: vitry Date: Tue, 1 Sep 2026 01:08:41 +0800 Subject: [PATCH 4/9] fix: cancel permission work when completion expires --- scripts/lib/zcode-client.mjs | 2 +- scripts/lib/zcode-protocol.mjs | 2 +- tests/process-zcode.test.mjs | 38 +++++++++++++++++++++++++++++++++- 3 files changed, 39 insertions(+), 3 deletions(-) diff --git a/scripts/lib/zcode-client.mjs b/scripts/lib/zcode-client.mjs index 389740f..3f9c717 100644 --- a/scripts/lib/zcode-client.mjs +++ b/scripts/lib/zcode-client.mjs @@ -129,7 +129,7 @@ export class ZCodeClient { } /** Wait for a validated terminal notification; no deadline applies unless configured on the client or supplied here. @param {string} sessionId @param {number} [timeoutMs] */ waitForCompletion(sessionId, timeoutMs) { return this.protocol.waitForCompletion(sessionId, timeoutMs); } - /** Observe a validated terminal notification without consuming the active turn. @param {string} sessionId @param {number} [timeoutMs] */ observeCompletion(sessionId, timeoutMs) { requireSessionId(sessionId); return this.protocol.observeCompletion(sessionId, timeoutMs); } + /** Observe a validated terminal notification without consuming the active turn. @param {string} sessionId @param {number} [timeoutMs] */ observeCompletion(sessionId, timeoutMs) { return this.protocol.observeCompletion(sessionId, timeoutMs); } /** Locally release an active turn without sending an upstream request. @param {string} sessionId */ releaseTurn(sessionId) { requireSessionId(sessionId); this.protocol.releaseTurn(sessionId); } /** Exact local protocol invariant used to prove whether this client owns an active turn. @param {string} sessionId */ turnState(sessionId) { requireSessionId(sessionId); return this.protocol.turnState(sessionId); } /** @param {string} sessionId @param {{connectionId:string,clientMode:'desktop-continuous'|'web-remote-replayable'}} options */ diff --git a/scripts/lib/zcode-protocol.mjs b/scripts/lib/zcode-protocol.mjs index c902878..186ec62 100644 --- a/scripts/lib/zcode-protocol.mjs +++ b/scripts/lib/zcode-protocol.mjs @@ -320,7 +320,7 @@ export class ZCodeProtocolClient { rejectCompletionWaiters(error) { for (const waiter of this.completionWaiters) { if (waiter.timer) clearTimeout(waiter.timer); waiter.unsubscribe(); this.waiterSessions.delete(waiter.sessionId); waiter.reject(error); } this.completionWaiters.clear(); } /** @param {string} sessionId @param {any} params */ - queueCompletion(sessionId, params) { if (this.consumeTerminal) { const turn = this.turns.get(sessionId); this.abortTurn(sessionId); if (turn?.status === 'armed' && typeof turn.baseline === 'number' && typeof turn.inputId === 'string') this.terminalHandler?.(params, { status: 'armed', baseline: turn.baseline, inputId: turn.inputId }); return; } if (!this.completed.has(sessionId) && this.completed.size >= 1024) { this.fail(new PluginError('ZCODE_COMPLETION_OVERFLOW', 'Too many unconsumed completions were received.', { category: 'protocol', remedy: 'Restart the connection and consume completions promptly.' })); return; } const queue = this.completed.get(sessionId) ?? []; queue.splice(0, queue.length, params); this.completed.set(sessionId, queue); clearTimeout(this.completionExpiry.get(sessionId)); const expiry = setTimeout(() => this.abortTurn(sessionId), 10 * 60_000); expiry.unref?.(); this.completionExpiry.set(sessionId, expiry); } + queueCompletion(sessionId, params) { if (this.consumeTerminal) { const turn = this.turns.get(sessionId); this.abortTurn(sessionId); if (turn?.status === 'armed' && typeof turn.baseline === 'number' && typeof turn.inputId === 'string') this.terminalHandler?.(params, { status: 'armed', baseline: turn.baseline, inputId: turn.inputId }); return; } if (!this.completed.has(sessionId) && this.completed.size >= 1024) { this.fail(new PluginError('ZCODE_COMPLETION_OVERFLOW', 'Too many unconsumed completions were received.', { category: 'protocol', remedy: 'Restart the connection and consume completions promptly.' })); return; } const queue = this.completed.get(sessionId) ?? []; queue.splice(0, queue.length, params); this.completed.set(sessionId, queue); clearTimeout(this.completionExpiry.get(sessionId)); const expiry = setTimeout(() => this.cancelTurn(sessionId), 10 * 60_000); expiry.unref?.(); this.completionExpiry.set(sessionId, expiry); } } export class BoundedWriter { diff --git a/tests/process-zcode.test.mjs b/tests/process-zcode.test.mjs index 3e00dd7..fcadd45 100644 --- a/tests/process-zcode.test.mjs +++ b/tests/process-zcode.test.mjs @@ -282,7 +282,7 @@ test('releaseTurn is local and idempotent and rejects a pending observer', async assert.equal(protocol.subscribers.size, 0); assert.equal(child.stdin.readableLength, 0, 'local release must not send an upstream RPC'); for (const invalid of ['', 'bad\nsession', null]) { - assert.throws(() => client.observeCompletion(invalid), { code: 'ZCODE_INPUT_INVALID' }); + await assert.rejects(client.observeCompletion(invalid), { code: 'ZCODE_PROTOCOL_INPUT_INVALID' }); assert.throws(() => client.releaseTurn(invalid), { code: 'ZCODE_INPUT_INVALID' }); } }); @@ -309,6 +309,42 @@ test('releaseTurn aborts and clears permission task state without writing a stal assert.equal(child.stdin.readableLength, 0); }); +test('completion expiry cancels pending permission tasks before late resolution or rejection', async () => { + for (const outcome of ['resolve', 'reject']) { + const child = new EventEmitter(); child.stdin = new PassThrough(); child.stdout = new PassThrough(); child.stderr = new PassThrough(); child.exitCode = 0; child.signalCode = null; + const protocol = new ZCodeProtocolClient(child); const failures = []; + protocol.setCloseHandler((error) => failures.push(error)); + protocol.beginTurn('session-1'); protocol.armTurn('session-1', 1, 'input-1'); + const observed = protocol.observeCompletion('session-1'); + const originalSetTimeout = globalThis.setTimeout; let expire; + globalThis.setTimeout = (callback, timeoutMs, ...args) => { + if (timeoutMs === 10 * 60_000) { expire = () => callback(...args); return { unref() {} }; } + return originalSetTimeout(callback, timeoutMs, ...args); + }; + try { protocol.handleLine(JSON.stringify({ method: 'state.updated', params: { scope: 'session', sessionId: 'session-1', revision: 2, reason: 'prompt_completed' } })); } + finally { globalThis.setTimeout = originalSetTimeout; } + await observed; + assert.equal(typeof expire, 'function'); + + let handlerSignal; let settle; + protocol.setPermissionHandler((_request, signal) => { + handlerSignal = signal; + return new Promise((resolve, reject) => { settle = outcome === 'resolve' ? () => resolve({ decision: 'deny' }) : () => reject(new Error('late rejection')); }); + }); + protocol.handleLine(JSON.stringify({ id: 99, method: 'interaction/requestPermission', params: { requestId: 'r', sessionId: 'session-1', toolCallId: 't', toolName: 'write', reason: 'test', riskLevel: 'low', input: {}, options: [{ optionId: 'deny', kind: 'deny', name: 'Deny', response: { decision: 'deny' } }] } })); + await new Promise((resolve) => setImmediate(resolve)); + expire(); + assert.equal(handlerSignal.aborted, true, outcome); + assert.equal(protocol.turnState('session-1'), null, outcome); + for (const collection of [protocol.serverTasks, protocol.serverTaskControllers, protocol.serverTaskSessions, protocol.serverTasksByController, protocol.permissionRequestIds]) assert.equal(collection.size, 0, outcome); + settle(); + await new Promise((resolve) => setImmediate(resolve)); + assert.equal(child.stdin.readableLength, 0, outcome); + assert.equal(protocol.closed, false, outcome); + assert.deepEqual(failures, [], outcome); + } +}); + test('ordinary completion waiting remains destructive', async () => { const child = new EventEmitter(); child.stdin = new PassThrough(); child.stdout = new PassThrough(); child.stderr = new PassThrough(); child.exitCode = 0; child.signalCode = null; const protocol = new ZCodeProtocolClient(child); From 5436060d2e7e0eda0ecfa6576c5ee16785ebb06d Mon Sep 17 00:00:00 2001 From: vitry Date: Tue, 1 Sep 2026 01:18:46 +0800 Subject: [PATCH 5/9] fix: retain active turn through legacy wake --- scripts/lib/review.mjs | 10 +++++++- tests/job-control.test.mjs | 48 ++++++++++++++++++++++++++++++++++---- 2 files changed, 53 insertions(+), 5 deletions(-) diff --git a/scripts/lib/review.mjs b/scripts/lib/review.mjs index 7d7ee95..e485a0a 100644 --- a/scripts/lib/review.mjs +++ b/scripts/lib/review.mjs @@ -225,7 +225,10 @@ export async function executeJob(input) { reporter.activateAcceptedBoundary({ readSnapshot: () => client.readSession(activeSessionId), describer: sessionDescriber }); } catch { reporter.activateAcceptedBoundary({}); } reporter.activate({ method: 'state.updated', params: { scope: 'session', sessionId: activeSessionId, reason: 'prompt_started' } }); - const legacyWake = waitForCompletionOrAbort(client.waitForCompletion(activeSessionId), input.signal); + const observeLegacyCompletion = typeof client.observeCompletion === 'function' + ? client.observeCompletion.bind(client) + : client.waitForCompletion.bind(client); + const legacyWake = waitForCompletionOrAbort(observeLegacyCompletion(activeSessionId), input.signal); const terminal = await awaitCurrentTurnTerminal({ legacyWake, conversationObserver, readSnapshot: () => client.readSession(activeSessionId), turnBoundary, signal: input.signal, }); @@ -336,6 +339,11 @@ export async function executeJob(input) { } // Cleanup order is part of the progress lifecycle contract. await cleanupProgress(); + try { + if (sessionId && typeof client.releaseTurn === 'function') client.releaseTurn(sessionId); + } catch (cleanupError) { + if (!primaryError) primaryError = cleanupError; + } await client.close().catch(() => {}); if (!primaryError && appliedFinalization && output?.job?.status === 'succeeded' && typeof output.result === 'string') { await jobLog?.appendBlock('Final output', output.result, Date.now() + OPTIONAL_PROGRESS_FENCE_MS); diff --git a/tests/job-control.test.mjs b/tests/job-control.test.mjs index fa4b60d..7318ab0 100644 --- a/tests/job-control.test.mjs +++ b/tests/job-control.test.mjs @@ -1512,20 +1512,35 @@ test('0.16.5 foreground execution treats legacy completion as admission and wait const { root, workspace, store } = await setup(); const job = await store.reserveJob({ workspace, ...reservation }); const sessionId = 'zs-0165-true-terminal'; const subscriptionId = 'subscription-0165'; /** @type {null|((message:any)=>void)} */ let handler = null; + /** @type {null|((request:any)=>any)} */ let permissionHandler = null; const emit = (/** @type {any} */ message) => { if (!handler) throw new Error('0.16.5 progress handler missing'); handler(message); }; let trueTerminal = false; let reads = 0; let boundaryDurable = false; let signalBoundary = () => {}; + /** @type {any} */ let permissionDecision; let signalPermission = () => {}; + const permissionDecided = new Promise((resolve) => { signalPermission = () => resolve(undefined); }); + /** @type {string[]} */ const cleanupCalls = []; const boundaryReached = new Promise((resolve) => { signalBoundary = () => resolve(undefined); }); const legacy = { method: 'state.updated', params: { type: 'state.updated', scope: 'session', sessionId, revision: 1, reason: 'prompt_completed', patch: {}, futureNotificationField: true } }; const client = { createSession: async () => ({ session: { sessionId }, settings: { model: { current: { providerId: 'p', modelId: 'm' }, available: [] } }, messages: [] }), - setPermissionHandler: () => {}, + setPermissionHandler: (/** @type {(request:any)=>any} */ nextHandler) => { permissionHandler = nextHandler; }, subscribe: (/** @type {(message:any)=>void} */ subscriber) => { handler = subscriber; return () => { handler = null; }; }, subscribeConversation: async () => { emit(conversationFrame(/** @type {any} */ ({ sessionId, subscriptionId, deliveryKind: 'initial', ordinal: 1, fromSeq: 0, toSeq: 484, snapshot: boundedSnapshotFixture({ sessionId, seq: 484 }) }))); return { subscriptionId, unsubscribe: async () => {} }; }, send: async () => ({ inputId: 'input-0165', stateRevision: 7 }), - waitForCompletion: async () => { emit(legacy); }, + waitForCompletion: async () => { throw new Error('executor must use non-destructive completion observation'); }, + observeCompletion: async () => { + emit(legacy); + setImmediate(() => { + if (!permissionHandler) throw new Error('0.16.5 permission handler missing'); + permissionDecision = permissionHandler({ + sessionId, toolName: 'Write', riskLevel: 'medium', + options: [{ response: { decision: 'allow' } }, { response: { decision: 'deny' } }], + }); + signalPermission(); + }); + }, readSession: async () => { assert.equal(boundaryDurable, true, 'legacy admission must not permit a read before the accepted boundary is durable'); reads += 1; @@ -1535,15 +1550,18 @@ test('0.16.5 foreground execution treats legacy completion as admission and wait { info: { role: 'assistant', messageId: 'assistant-0165', parentMessageId: 'input-0165', time: { created: 2, completed: 3 }, finish: 'stop', semantics: { origin: 'agent_runtime', kind: 'assistant_response', uiVisibility: 'visible' } }, parts: [{ type: 'text', text: 'real 0.16.5 result' }] }, ] }; }, - stopSession: async () => {}, close: async () => {}, + stopSession: async () => {}, + releaseTurn: (/** @type {string} */ releasedSessionId) => { cleanupCalls.push(`release:${releasedSessionId}`); }, + close: async () => { cleanupCalls.push('close'); }, }; const execution = executeJob({ job, workspace, dataRoot: join(root, 'data'), store, client, task: 'task', onBoundaryPersisted: async () => { boundaryDurable = true; signalBoundary(); }, }); execution.catch(() => {}); - await boundaryReached; await new Promise((resolve) => setTimeout(resolve, 40)); + await boundaryReached; await Promise.race([permissionDecided, execution]); await new Promise((resolve) => setTimeout(resolve, 40)); assert.equal((await store.readJob(workspace, job.id)).status, 'running'); + assert.deepEqual(permissionDecision, { decision: 'allow' }); assert.ok(reads >= 1, 'legacy admission should wake transitional snapshot reconciliation'); emit(conversationFrame(/** @type {any} */ ({ sessionId, subscriptionId, ordinal: 2, fromSeq: 484, toSeq: 485, deltas: [captured0165TurnRow({ rowId: 101, turnId: 'turn-0165', state: 'running' })] }))); trueTerminal = true; @@ -1551,6 +1569,28 @@ test('0.16.5 foreground execution treats legacy completion as admission and wait const output = await execution; assert.equal(output.result, 'real 0.16.5 result'); assert.equal(output.job.status, 'succeeded'); + assert.deepEqual(cleanupCalls, [`release:${sessionId}`, 'close']); +}); + +test('executor releases the local turn before close on failure and preserves the primary error', async () => { + const { root, workspace, store } = await setup(); const job = await store.reserveJob({ workspace, ...reservation }); + const sessionId = 'zs-release-failure-cleanup'; + const primary = new PluginError('PRIMARY_RELEASE_TEST', 'primary execution failure', { category: 'protocol', remedy: 'retain primary' }); + const releaseError = new Error('local release failed'); + /** @type {string[]} */ const cleanupCalls = []; + const client = { + createSession: async () => ({ session: { sessionId }, settings: { model: { current: { providerId: 'p', modelId: 'm' }, available: [] } }, messages: [] }), + setPermissionHandler: () => {}, subscribe: silentSubscribe, + send: async () => ({ inputId: 'input-release-failure-cleanup', stateRevision: 1 }), + observeCompletion: async () => { throw primary; }, + waitForCompletion: async () => { throw new Error('legacy waiter must not be used'); }, + stopSession: async () => {}, + releaseTurn: (/** @type {string} */ releasedSessionId) => { cleanupCalls.push(`release:${releasedSessionId}`); throw releaseError; }, + close: async () => { cleanupCalls.push('close'); }, + }; + const caught = await executeJob({ job, workspace, dataRoot: join(root, 'data'), store, client, task: 'task' }).catch((error) => error); + assert.equal(caught, primary); + assert.deepEqual(cleanupCalls, [`release:${sessionId}`, 'close']); }); test('execution does not wait indefinitely for a late initial baseline and uses coherent snapshot fallback', { timeout: 10_000 }, async () => { From 7a0271ce322ddf32bf88c20e246ac7be54347daa Mon Sep 17 00:00:00 2001 From: vitry Date: Tue, 1 Sep 2026 01:32:47 +0800 Subject: [PATCH 6/9] fix: preserve durable success on turn release failure --- scripts/lib/review.mjs | 2 +- tests/job-control.test.mjs | 29 +++++++++++++++++++++++++++++ 2 files changed, 30 insertions(+), 1 deletion(-) diff --git a/scripts/lib/review.mjs b/scripts/lib/review.mjs index e485a0a..dbd496b 100644 --- a/scripts/lib/review.mjs +++ b/scripts/lib/review.mjs @@ -342,7 +342,7 @@ export async function executeJob(input) { try { if (sessionId && typeof client.releaseTurn === 'function') client.releaseTurn(sessionId); } catch (cleanupError) { - if (!primaryError) primaryError = cleanupError; + if (!primaryError && output?.job?.status !== 'succeeded') primaryError = cleanupError; } await client.close().catch(() => {}); if (!primaryError && appliedFinalization && output?.job?.status === 'succeeded' && typeof output.result === 'string') { diff --git a/tests/job-control.test.mjs b/tests/job-control.test.mjs index 7318ab0..d37efd7 100644 --- a/tests/job-control.test.mjs +++ b/tests/job-control.test.mjs @@ -1562,6 +1562,7 @@ test('0.16.5 foreground execution treats legacy completion as admission and wait await boundaryReached; await Promise.race([permissionDecided, execution]); await new Promise((resolve) => setTimeout(resolve, 40)); assert.equal((await store.readJob(workspace, job.id)).status, 'running'); assert.deepEqual(permissionDecision, { decision: 'allow' }); + assert.deepEqual(cleanupCalls, [], 'the active turn must remain retained until authoritative terminal reconciliation'); assert.ok(reads >= 1, 'legacy admission should wake transitional snapshot reconciliation'); emit(conversationFrame(/** @type {any} */ ({ sessionId, subscriptionId, ordinal: 2, fromSeq: 484, toSeq: 485, deltas: [captured0165TurnRow({ rowId: 101, turnId: 'turn-0165', state: 'running' })] }))); trueTerminal = true; @@ -1572,6 +1573,34 @@ test('0.16.5 foreground execution treats legacy completion as admission and wait assert.deepEqual(cleanupCalls, [`release:${sessionId}`, 'close']); }); +test('durable success remains authoritative when local turn release fails', async () => { + const { root, workspace, store } = await setup(); const job = await store.reserveJob({ workspace, ...reservation }); + const dataRoot = join(root, 'data'); const sessionId = 'zs-release-after-success'; + const releaseError = new Error('local release failed after durable success'); + /** @type {string[]} */ const cleanupCalls = []; + const client = { + createSession: async () => ({ session: { sessionId }, settings: { model: { current: { providerId: 'p', modelId: 'm' }, available: [] } }, messages: [] }), + setPermissionHandler: () => {}, subscribe: silentSubscribe, + send: async () => ({ inputId: 'input-release-after-success', stateRevision: 1 }), + observeCompletion: async () => {}, + waitForCompletion: async () => { throw new Error('legacy waiter must not be used'); }, + readSession: async () => ({ projection: { status: 'completed' }, runtime: { stateRevision: 2 }, messages: [ + completedUser('input-release-after-success'), + { info: { role: 'assistant', messageId: 'assistant-release-after-success', parentMessageId: 'input-release-after-success', finish: 'stop' }, parts: [{ type: 'text', text: 'durable release result' }] }, + ] }), + stopSession: async () => {}, + releaseTurn: (/** @type {string} */ releasedSessionId) => { cleanupCalls.push(`release:${releasedSessionId}`); throw releaseError; }, + close: async () => { cleanupCalls.push('close'); }, + }; + const output = await executeJob({ job, workspace, dataRoot, store, client, task: 'task' }); + assert.equal(output.job.status, 'succeeded'); + assert.equal(output.result, 'durable release result'); + assert.deepEqual(cleanupCalls, [`release:${sessionId}`, 'close']); + const persisted = await store.readJob(workspace, job.id); + assert.equal(persisted.status, 'succeeded'); + assert.match(await readFile(persisted.logFile, 'utf8'), /Final output\ndurable release result\n/); +}); + test('executor releases the local turn before close on failure and preserves the primary error', async () => { const { root, workspace, store } = await setup(); const job = await store.reserveJob({ workspace, ...reservation }); const sessionId = 'zs-release-failure-cleanup'; From 92ee17cd36efa7cbc1ed6c026bf9c8bf341a542c Mon Sep 17 00:00:00 2001 From: vitry Date: Tue, 1 Sep 2026 01:47:07 +0800 Subject: [PATCH 7/9] build: refresh marketplace snapshot --- marketplace/.agents/plugins/provenance.json | 18 ++++---- .../plugins/zcode/scripts/lib/review.mjs | 10 +++- .../zcode/scripts/lib/zcode-client.mjs | 2 + .../zcode/scripts/lib/zcode-protocol.mjs | 46 +++++++++++++++++-- 4 files changed, 61 insertions(+), 15 deletions(-) diff --git a/marketplace/.agents/plugins/provenance.json b/marketplace/.agents/plugins/provenance.json index 17ed417..fedba55 100644 --- a/marketplace/.agents/plugins/provenance.json +++ b/marketplace/.agents/plugins/provenance.json @@ -1,15 +1,15 @@ { "packageVersion": "0.1.0", "pluginVersion": "0.1.0", - "sourceRef": "e7c986bcc8e743f6ba2ae35004cb51ad00a995ac", - "sourceSha": "e7c986bcc8e743f6ba2ae35004cb51ad00a995ac", + "sourceRef": "7a0271ce322ddf32bf88c20e246ac7be54347daa", + "sourceSha": "7a0271ce322ddf32bf88c20e246ac7be54347daa", "dependencyLock": { "file": "npm-shrinkwrap.json", "sha256": "fa927194e6ca0b25c1d3f428859b2ab4798b8eabb4d31bc87188d73c630f9938" }, "content": { "algorithm": "sha256", - "sha256": "02b5640cd5cf5ad4c1dbae7dd9e5028710d8ac3f12077ea08ce2f670c4dd2543", + "sha256": "16df3422c1c64706109a44164afc2f21cd1237a88dfd4616f8e9eb472c75cf03", "files": [ { "path": ".agents/plugins/marketplace.json", @@ -658,8 +658,8 @@ }, { "path": "plugins/zcode/scripts/lib/review.mjs", - "size": 43083, - "sha256": "5495fe99558e3d9595f9c016bdef9ed9aca4940b7f89438e544c5fd280c0f25f" + "size": 43482, + "sha256": "d0f2ba538daf82eb6616651ad8ac91da6e9229a6dfadfa2843930736555b7d60" }, { "path": "plugins/zcode/scripts/lib/session-progress.mjs", @@ -703,8 +703,8 @@ }, { "path": "plugins/zcode/scripts/lib/zcode-client.mjs", - "size": 52052, - "sha256": "2f38c2c6f0b1abbe094fbc7e67f359e6256ebfc0f05a602138f3c3a1e3559d77" + "size": 52495, + "sha256": "0d4c854d4045260cc86d14d30e647db22efb08bd4f9ad30a2f795c3b5e37f901" }, { "path": "plugins/zcode/scripts/lib/zcode-discovery.mjs", @@ -713,8 +713,8 @@ }, { "path": "plugins/zcode/scripts/lib/zcode-protocol.mjs", - "size": 38990, - "sha256": "a6531d8d112b7c605a26d8eee8c0af5f3e01f8e14ef34bfe3226df60a31f9db7" + "size": 42018, + "sha256": "68e98ef8be8b670648e3add7966fb1619b3836be0ce6b2a29e678899f68f9426" }, { "path": "plugins/zcode/scripts/lib/zcode-runtime-config.mjs", diff --git a/marketplace/plugins/zcode/scripts/lib/review.mjs b/marketplace/plugins/zcode/scripts/lib/review.mjs index 7d7ee95..dbd496b 100644 --- a/marketplace/plugins/zcode/scripts/lib/review.mjs +++ b/marketplace/plugins/zcode/scripts/lib/review.mjs @@ -225,7 +225,10 @@ export async function executeJob(input) { reporter.activateAcceptedBoundary({ readSnapshot: () => client.readSession(activeSessionId), describer: sessionDescriber }); } catch { reporter.activateAcceptedBoundary({}); } reporter.activate({ method: 'state.updated', params: { scope: 'session', sessionId: activeSessionId, reason: 'prompt_started' } }); - const legacyWake = waitForCompletionOrAbort(client.waitForCompletion(activeSessionId), input.signal); + const observeLegacyCompletion = typeof client.observeCompletion === 'function' + ? client.observeCompletion.bind(client) + : client.waitForCompletion.bind(client); + const legacyWake = waitForCompletionOrAbort(observeLegacyCompletion(activeSessionId), input.signal); const terminal = await awaitCurrentTurnTerminal({ legacyWake, conversationObserver, readSnapshot: () => client.readSession(activeSessionId), turnBoundary, signal: input.signal, }); @@ -336,6 +339,11 @@ export async function executeJob(input) { } // Cleanup order is part of the progress lifecycle contract. await cleanupProgress(); + try { + if (sessionId && typeof client.releaseTurn === 'function') client.releaseTurn(sessionId); + } catch (cleanupError) { + if (!primaryError && output?.job?.status !== 'succeeded') primaryError = cleanupError; + } await client.close().catch(() => {}); if (!primaryError && appliedFinalization && output?.job?.status === 'succeeded' && typeof output.result === 'string') { await jobLog?.appendBlock('Final output', output.result, Date.now() + OPTIONAL_PROGRESS_FENCE_MS); diff --git a/marketplace/plugins/zcode/scripts/lib/zcode-client.mjs b/marketplace/plugins/zcode/scripts/lib/zcode-client.mjs index f22d538..3f9c717 100644 --- a/marketplace/plugins/zcode/scripts/lib/zcode-client.mjs +++ b/marketplace/plugins/zcode/scripts/lib/zcode-client.mjs @@ -129,6 +129,8 @@ export class ZCodeClient { } /** Wait for a validated terminal notification; no deadline applies unless configured on the client or supplied here. @param {string} sessionId @param {number} [timeoutMs] */ waitForCompletion(sessionId, timeoutMs) { return this.protocol.waitForCompletion(sessionId, timeoutMs); } + /** Observe a validated terminal notification without consuming the active turn. @param {string} sessionId @param {number} [timeoutMs] */ observeCompletion(sessionId, timeoutMs) { return this.protocol.observeCompletion(sessionId, timeoutMs); } + /** Locally release an active turn without sending an upstream request. @param {string} sessionId */ releaseTurn(sessionId) { requireSessionId(sessionId); this.protocol.releaseTurn(sessionId); } /** Exact local protocol invariant used to prove whether this client owns an active turn. @param {string} sessionId */ turnState(sessionId) { requireSessionId(sessionId); return this.protocol.turnState(sessionId); } /** @param {string} sessionId @param {{connectionId:string,clientMode:'desktop-continuous'|'web-remote-replayable'}} options */ async subscribeConversation(sessionId, options) { diff --git a/marketplace/plugins/zcode/scripts/lib/zcode-protocol.mjs b/marketplace/plugins/zcode/scripts/lib/zcode-protocol.mjs index e2455e7..186ec62 100644 --- a/marketplace/plugins/zcode/scripts/lib/zcode-protocol.mjs +++ b/marketplace/plugins/zcode/scripts/lib/zcode-protocol.mjs @@ -37,6 +37,7 @@ export class ZCodeProtocolClient { /** @type {Set>} */ this.serverTasks = new Set(); /** @type {Set} */ this.serverTaskControllers = new Set(); /** @type {Map} */ this.serverTaskSessions = new Map(); + /** @type {Map>} */ this.serverTasksByController = new Map(); this.nextId = 1; this.buffer = ''; this.closed = false; @@ -102,9 +103,12 @@ export class ZCodeProtocolClient { /** @param {string} sessionId @param {PluginError} [error] */ cancelTurn(sessionId, error = new PluginError('ZCODE_SESSION_STOPPED', `ZCode session ${sessionId} was stopped.`, { category: 'state', remedy: 'Start a new turn before waiting for completion.', details: { sessionId } })) { for (const waiter of this.completionWaiters) if (waiter.sessionId === sessionId) { if (waiter.timer) clearTimeout(waiter.timer); waiter.unsubscribe(); this.completionWaiters.delete(waiter); this.waiterSessions.delete(sessionId); waiter.reject(error); } - for (const [controller, taskSessionId] of this.serverTaskSessions) if (taskSessionId === sessionId) controller.abort(); + for (const [controller, taskSessionId] of this.serverTaskSessions) if (taskSessionId === sessionId) { controller.abort(); const task = this.serverTasksByController.get(controller); if (task) this.serverTasks.delete(task); this.serverTaskControllers.delete(controller); this.serverTaskSessions.delete(controller); this.serverTasksByController.delete(controller); } + for (const key of this.permissionRequestIds.keys()) { try { if (JSON.parse(key)?.[0] === sessionId) this.permissionRequestIds.delete(key); } catch { /* internal replay keys are always JSON */ } } this.abortTurn(sessionId); } + /** Locally release a turn without sending an upstream request. @param {string} sessionId */ + releaseTurn(sessionId) { if (!nonEmpty(sessionId)) throw protocolInputError(); this.cancelTurn(sessionId); } /** @param {(message:any)=>void} handler */ subscribe(handler) { if (typeof handler !== 'function' || this.subscribers.size >= 256) throw protocolInputError(); this.subscribers.add(handler); return () => this.subscribers.delete(handler); } @@ -141,6 +145,37 @@ export class ZCodeProtocolClient { }); } + /** Observe a validated terminal notification without consuming it or ending the turn. @param {string} sessionId @param {number} [timeoutMs] */ + observeCompletion(sessionId, timeoutMs) { + const effectiveTimeoutMs = timeoutMs === undefined ? this.completionTimeoutMs : timeoutMs; + if (!nonEmpty(sessionId) || effectiveTimeoutMs !== undefined && (!Number.isSafeInteger(effectiveTimeoutMs) || effectiveTimeoutMs < 1 || effectiveTimeoutMs > 86_400_000) || this.turns.get(sessionId)?.status !== 'armed' || this.waiterSessions.has(sessionId)) return Promise.reject(protocolInputError()); + const queued = this.completed.get(sessionId)?.[0]; + if (queued) return Promise.resolve(queued); + return new Promise((resolve, reject) => { + let unsubscribe = () => {}; + this.waiterSessions.add(sessionId); const waiter = { reject, timer: /** @type {NodeJS.Timeout|null} */ (null), unsubscribe, sessionId }; + const timer = effectiveTimeoutMs === undefined ? null : setTimeout(() => { + this.completionWaiters.delete(waiter); this.waiterSessions.delete(sessionId); + unsubscribe(); + reject(new PluginError('ZCODE_COMPLETION_TIMEOUT', `ZCode session ${sessionId} did not complete in time.`, { category: 'timeout', remedy: 'Read or resume the session before retrying.', details: { sessionId, timeoutMs: effectiveTimeoutMs } })); + }, effectiveTimeoutMs); + waiter.timer = timer; + timer?.unref?.(); + try { + unsubscribe = this.subscribe((message) => { + if (!isCompletionFor(message, sessionId, this.turns.get(sessionId))) return; + this.completionWaiters.delete(waiter); this.waiterSessions.delete(sessionId); if (timer) clearTimeout(timer); unsubscribe(); resolve(message.params); + }); + } catch (error) { + if (timer) clearTimeout(timer); + this.completionWaiters.delete(waiter); this.waiterSessions.delete(sessionId); waiter.unsubscribe(); + reject(error); return; + } + waiter.unsubscribe = unsubscribe; + this.completionWaiters.add(waiter); + }); + } + close() { this.closePromise ??= this.closeOnce(); return this.closePromise; @@ -159,7 +194,7 @@ export class ZCodeProtocolClient { } const tasks = [...this.serverTasks]; if (tasks.length) await Promise.race([Promise.allSettled(tasks), boundedDelay(25)]); - this.serverTasks.clear(); this.serverTaskControllers.clear(); this.serverTaskSessions.clear(); + this.serverTasks.clear(); this.serverTaskControllers.clear(); this.serverTaskSessions.clear(); this.serverTasksByController.clear(); this.terminationPromise ??= terminateProcess(this.child); await this.terminationPromise; } @@ -241,6 +276,7 @@ export class ZCodeProtocolClient { if (this.permissionRequestIds.size >= 1024) throw new PluginError('ZCODE_PERMISSION_OVERFLOW', 'Too many permission requests were rejected.', { category: 'authorization', remedy: 'Restart the affected ZCode turn.' }); this.permissionRequestIds.set(replayKey, Date.now()); const result = this.permissionHandler ? await this.permissionHandler(message.params, signal) : message.params.options.find((/** @type {any} */ option) => option.response.decision === 'deny')?.response; + if (signal.aborted) return; validatePermissionResult(result); if (!message.params.options.some((/** @type {any} */ option) => JSON.stringify(option.response) === JSON.stringify(result))) throw new PluginError('ZCODE_PERMISSION_OPTION_INVALID', 'Permission response was not one of the offered options.', { category: 'authorization', remedy: 'Return an exact response offered by ZCode.' }); if (!this.closed) this.sendFrame({ id: message.id, result }); @@ -254,7 +290,7 @@ export class ZCodeProtocolClient { } /** @param {Promise} task @param {AbortController} controller @param {unknown} sessionId */ - trackServerTask(task, controller, sessionId) { this.serverTasks.add(task); this.serverTaskControllers.add(controller); if (nonEmpty(sessionId)) this.serverTaskSessions.set(controller, /** @type {string} */ (sessionId)); const cleanup = () => { this.serverTasks.delete(task); this.serverTaskControllers.delete(controller); this.serverTaskSessions.delete(controller); }; void task.then(cleanup, (error) => { cleanup(); this.fail(asDisconnected(error, this.stderrTail.value())); }); } + trackServerTask(task, controller, sessionId) { this.serverTasks.add(task); this.serverTaskControllers.add(controller); this.serverTasksByController.set(controller, task); if (nonEmpty(sessionId)) this.serverTaskSessions.set(controller, /** @type {string} */ (sessionId)); const cleanup = () => { this.serverTasks.delete(task); this.serverTaskControllers.delete(controller); this.serverTaskSessions.delete(controller); this.serverTasksByController.delete(controller); }; void task.then(cleanup, (error) => { cleanup(); this.fail(asDisconnected(error, this.stderrTail.value())); }); } /** @param {Record} frame */ sendFrame(frame) { @@ -268,7 +304,7 @@ export class ZCodeProtocolClient { if (this.closed) return; this.closed = true; for (const controller of this.serverTaskControllers) controller.abort(); - this.serverTasks.clear(); this.serverTaskControllers.clear(); this.serverTaskSessions.clear(); this.permissionRequestIds.clear(); + this.serverTasks.clear(); this.serverTaskControllers.clear(); this.serverTaskSessions.clear(); this.serverTasksByController.clear(); this.permissionRequestIds.clear(); this.stderrTail.close(); const diagnosticError = withStderr(error, this.stderrTail.value()); this.writer.close(); this.rejectPending(diagnosticError); this.rejectCompletionWaiters(diagnosticError); for (const timer of this.completionExpiry.values()) clearTimeout(timer); this.completionExpiry.clear(); this.completed.clear(); this.earlyCompletions.clear(); this.turns.clear(); @@ -284,7 +320,7 @@ export class ZCodeProtocolClient { rejectCompletionWaiters(error) { for (const waiter of this.completionWaiters) { if (waiter.timer) clearTimeout(waiter.timer); waiter.unsubscribe(); this.waiterSessions.delete(waiter.sessionId); waiter.reject(error); } this.completionWaiters.clear(); } /** @param {string} sessionId @param {any} params */ - queueCompletion(sessionId, params) { if (this.consumeTerminal) { const turn = this.turns.get(sessionId); this.abortTurn(sessionId); if (turn?.status === 'armed' && typeof turn.baseline === 'number' && typeof turn.inputId === 'string') this.terminalHandler?.(params, { status: 'armed', baseline: turn.baseline, inputId: turn.inputId }); return; } if (!this.completed.has(sessionId) && this.completed.size >= 1024) { this.fail(new PluginError('ZCODE_COMPLETION_OVERFLOW', 'Too many unconsumed completions were received.', { category: 'protocol', remedy: 'Restart the connection and consume completions promptly.' })); return; } const queue = this.completed.get(sessionId) ?? []; queue.splice(0, queue.length, params); this.completed.set(sessionId, queue); clearTimeout(this.completionExpiry.get(sessionId)); const expiry = setTimeout(() => this.abortTurn(sessionId), 10 * 60_000); expiry.unref?.(); this.completionExpiry.set(sessionId, expiry); } + queueCompletion(sessionId, params) { if (this.consumeTerminal) { const turn = this.turns.get(sessionId); this.abortTurn(sessionId); if (turn?.status === 'armed' && typeof turn.baseline === 'number' && typeof turn.inputId === 'string') this.terminalHandler?.(params, { status: 'armed', baseline: turn.baseline, inputId: turn.inputId }); return; } if (!this.completed.has(sessionId) && this.completed.size >= 1024) { this.fail(new PluginError('ZCODE_COMPLETION_OVERFLOW', 'Too many unconsumed completions were received.', { category: 'protocol', remedy: 'Restart the connection and consume completions promptly.' })); return; } const queue = this.completed.get(sessionId) ?? []; queue.splice(0, queue.length, params); this.completed.set(sessionId, queue); clearTimeout(this.completionExpiry.get(sessionId)); const expiry = setTimeout(() => this.cancelTurn(sessionId), 10 * 60_000); expiry.unref?.(); this.completionExpiry.set(sessionId, expiry); } } export class BoundedWriter { From ee1fa5c9778cf6dd02e094cf2bfe7f2835c02dab Mon Sep 17 00:00:00 2001 From: vitry Date: Tue, 1 Sep 2026 02:12:51 +0800 Subject: [PATCH 8/9] fix: always release turn after reconciliation failure --- scripts/lib/review.mjs | 5 ++++- tests/job-control.test.mjs | 27 +++++++++++++++++++++++++++ 2 files changed, 31 insertions(+), 1 deletion(-) diff --git a/scripts/lib/review.mjs b/scripts/lib/review.mjs index dbd496b..6683b18 100644 --- a/scripts/lib/review.mjs +++ b/scripts/lib/review.mjs @@ -255,7 +255,7 @@ export async function executeJob(input) { } } catch (error) { primaryError = error instanceof SuccessfulResultFinalizationError ? error.cause : error; - let current = initialBoundStopGuardComplete ? await input.store.readJob(workspace, job.id).catch(() => running) : null; + try { let current = initialBoundStopGuardComplete ? await input.store.readJob(workspace, job.id).catch(() => running) : null; let resumeFailureSettlementRejected = false; if (input.resumeSessionId && current?.status === 'queued' && input.onResumeFailure) { try { @@ -336,6 +336,9 @@ export async function executeJob(input) { } if (canFail) try { await input.store.finishJob(workspace, job.id, [current.status], 'failed', { error: safeError(error), exitCode: 1 }); } catch (finalizeError) { primaryError = finalizeError; } } + } catch (reconciliationError) { + primaryError = reconciliationError; + } } // Cleanup order is part of the progress lifecycle contract. await cleanupProgress(); diff --git a/tests/job-control.test.mjs b/tests/job-control.test.mjs index d37efd7..5240678 100644 --- a/tests/job-control.test.mjs +++ b/tests/job-control.test.mjs @@ -1382,6 +1382,33 @@ test('session creation completion observes abort before configuration and stops assert.equal((await store.readJob(workspace, job.id)).status, 'cancelled'); }); +test('queued interruption releases the local turn when stop revalidation reconciliation fails', async () => { + const { root, workspace, store } = await setup(); + const executor = { parentSessionId: 'session-a', parentTurnId: 'parent-turn', agentId: 'reconciliation-failure-child', agentType: 'zcode-rescue', agentPath: '/root/reconciliation-failure-child', workspace, parentPermissionMode: 'workspace-write' }; + const active = await store.reserveFreshRescueJob({ workspace, reservation: { workspace, ...reservation }, executor }); + const workerLeaseId = 'f'.repeat(64); const job = await store.claimJobWorkerForExecution(workspace, active.job.id, { childPid: 999_999, workerLeaseId }); + const controller = new AbortController(); const interruption = new PluginError('JOB_INTERRUPTED', 'model interrupted'); + const reconciliationError = new Error('queued stop revalidation failed'); + const sessionId = 'zs-queued-reconciliation-failure'; let revalidations = 0; + /** @type {string[]} */ const cleanupCalls = []; + const wrapped = { ...store, revalidateBoundRescueStop: async (/** @type {any} */ input) => { + revalidations += 1; + if (revalidations === 2) throw reconciliationError; + return store.revalidateBoundRescueStop(input); + } }; + const client = { + createSession: async () => ({ session: { sessionId }, settings: { model: { current: { providerId: 'p', modelId: 'old' }, available: [] } }, messages: [] }), + setModel: async () => { controller.abort(interruption); throw new Error('model transport closed'); }, + stopSession: async () => { throw new Error('stop must not run after failed revalidation'); }, + releaseTurn: (/** @type {string} */ releasedSessionId) => { cleanupCalls.push(`release:${releasedSessionId}`); }, + close: async () => { cleanupCalls.push('close'); }, + }; + const caught = await executeJobProduction({ job, workspace, dataRoot: join(root, 'data'), store: wrapped, client, task: 'task', model: { providerId: 'p', modelId: 'new' }, childPid: 999_999, workerLeaseId, signal: controller.signal }).catch((error) => error); + assert.equal(caught, reconciliationError); + assert.equal(revalidations, 2); + assert.deepEqual(cleanupCalls, [`release:${sessionId}`, 'close']); +}); + test('resume transport rejection after abort preserves the interruption and stops the known session once', async () => { const { root, workspace, store } = await setup(); const job = await store.reserveJob({ workspace, ...reservation }); const controller = new AbortController(); let stops = 0; From 2e6abb0748e37a89c056d46f4f961897715380f0 Mon Sep 17 00:00:00 2001 From: vitry Date: Tue, 1 Sep 2026 02:14:28 +0800 Subject: [PATCH 9/9] build: refresh marketplace snapshot after cleanup fix --- marketplace/.agents/plugins/provenance.json | 10 +++++----- marketplace/plugins/zcode/scripts/lib/review.mjs | 5 ++++- 2 files changed, 9 insertions(+), 6 deletions(-) diff --git a/marketplace/.agents/plugins/provenance.json b/marketplace/.agents/plugins/provenance.json index fedba55..c075c01 100644 --- a/marketplace/.agents/plugins/provenance.json +++ b/marketplace/.agents/plugins/provenance.json @@ -1,15 +1,15 @@ { "packageVersion": "0.1.0", "pluginVersion": "0.1.0", - "sourceRef": "7a0271ce322ddf32bf88c20e246ac7be54347daa", - "sourceSha": "7a0271ce322ddf32bf88c20e246ac7be54347daa", + "sourceRef": "ee1fa5c9778cf6dd02e094cf2bfe7f2835c02dab", + "sourceSha": "ee1fa5c9778cf6dd02e094cf2bfe7f2835c02dab", "dependencyLock": { "file": "npm-shrinkwrap.json", "sha256": "fa927194e6ca0b25c1d3f428859b2ab4798b8eabb4d31bc87188d73c630f9938" }, "content": { "algorithm": "sha256", - "sha256": "16df3422c1c64706109a44164afc2f21cd1237a88dfd4616f8e9eb472c75cf03", + "sha256": "40d1a65be6976f5f070b2784a35635f5aba4597c4846ec798b86d3f30dd77af7", "files": [ { "path": ".agents/plugins/marketplace.json", @@ -658,8 +658,8 @@ }, { "path": "plugins/zcode/scripts/lib/review.mjs", - "size": 43482, - "sha256": "d0f2ba538daf82eb6616651ad8ac91da6e9229a6dfadfa2843930736555b7d60" + "size": 43572, + "sha256": "0d945d223ce618ee28e2e81831ce855df1b18cc975f45dcc5538b8cc6e934d11" }, { "path": "plugins/zcode/scripts/lib/session-progress.mjs", diff --git a/marketplace/plugins/zcode/scripts/lib/review.mjs b/marketplace/plugins/zcode/scripts/lib/review.mjs index dbd496b..6683b18 100644 --- a/marketplace/plugins/zcode/scripts/lib/review.mjs +++ b/marketplace/plugins/zcode/scripts/lib/review.mjs @@ -255,7 +255,7 @@ export async function executeJob(input) { } } catch (error) { primaryError = error instanceof SuccessfulResultFinalizationError ? error.cause : error; - let current = initialBoundStopGuardComplete ? await input.store.readJob(workspace, job.id).catch(() => running) : null; + try { let current = initialBoundStopGuardComplete ? await input.store.readJob(workspace, job.id).catch(() => running) : null; let resumeFailureSettlementRejected = false; if (input.resumeSessionId && current?.status === 'queued' && input.onResumeFailure) { try { @@ -336,6 +336,9 @@ export async function executeJob(input) { } if (canFail) try { await input.store.finishJob(workspace, job.id, [current.status], 'failed', { error: safeError(error), exitCode: 1 }); } catch (finalizeError) { primaryError = finalizeError; } } + } catch (reconciliationError) { + primaryError = reconciliationError; + } } // Cleanup order is part of the progress lifecycle contract. await cleanupProgress();