From 055df49f0800e9c8f248059e618cfe52ca0f4535 Mon Sep 17 00:00:00 2001 From: Taras Mankovski Date: Tue, 10 Nov 2026 21:33:23 -0500 Subject: [PATCH 1/3] Capture stalled Sidekick recovery test phases --- .../cli/tests/fixtures/repl/issue-883/host.ts | 87 ++++++++++++++++++- .../fixtures/repl/issue-883/recovery-phase.ts | 20 ++++- 2 files changed, 103 insertions(+), 4 deletions(-) diff --git a/packages/cli/tests/fixtures/repl/issue-883/host.ts b/packages/cli/tests/fixtures/repl/issue-883/host.ts index 91cc222da..7e0cdeb9f 100644 --- a/packages/cli/tests/fixtures/repl/issue-883/host.ts +++ b/packages/cli/tests/fixtures/repl/issue-883/host.ts @@ -12,11 +12,21 @@ * case brings of its own is the interruption. */ -import { appendFile, mkdir, mkdtemp, open, rm } from "node:fs/promises"; +import { + appendFile, + mkdir, + mkdtemp, + open, + readdir, + readFile, + readlink, + rm, + writeFile, +} from "node:fs/promises"; import { randomBytes } from "node:crypto"; import { tmpdir } from "node:os"; import { join } from "node:path"; -import { ensure, run, spawn, until } from "effection"; +import { ensure, run, sleep, spawn, until } from "effection"; import type { Stream } from "effection"; import { exec } from "@effectionx/process"; import process from "node:process"; @@ -26,6 +36,11 @@ import type { Operation } from "effection"; import { installReplHost } from "../../../../src/repl-assembly.ts"; import { REPL_DIRECTORY } from "../../../../src/repl/storage.ts"; +import { + ENTRIES_FILE, + PAIRED_DIRECTORY, + SIDEKICK_FILE, +} from "../../../../src/repl/sidekick/history.ts"; /** How a gate can hold or fail one real append, by the path it is reaching. */ export interface AppendGate { @@ -177,9 +192,19 @@ export function* runPhase( execution, ...(extra === undefined ? [] : [extra]), ]); + const env = phaseEnv(); + const artifacts = process.env.XMD_TEST_ARTIFACTS; + const diagnose = + phase === "chat-say" && + artifacts !== undefined && + extra !== undefined && + ["delivered", "laterTurn"].includes(JSON.parse(extra).stop); + if (diagnose) { + env.XMD_RECOVERY_PHASE_TRACE = "1"; + } const child = yield* exec(launch.command, { arguments: launch.arguments, - env: phaseEnv(), + env, }); // Collected *while* the child runs, not after it. A subscription taken once // the process has already settled never settles itself, so reading the @@ -187,7 +212,13 @@ export function* runPhase( const captured = { stdout: "", stderr: "" }; yield* spawn(() => accumulate(child.stdout, captured, "stdout")); yield* spawn(() => accumulate(child.stderr, captured, "stderr")); + const observation = diagnose + ? yield* spawn(() => observeSlowPhase(child.pid, root, execution, artifacts, captured)) + : undefined; const status = yield* child.join(); + if (observation !== undefined) { + yield* observation.halt(); + } if (status.code !== 0) { throw new Error( `the ${phase} phase exited ${status.code ?? "on a signal"}: ${captured.stderr}`, @@ -200,6 +231,56 @@ export function* runPhase( return JSON.parse(line) as Record; } +function* observeSlowPhase( + pid: number, + root: RepositoryRoot, + execution: string, + artifacts: string, + captured: { stdout: string; stderr: string }, +): Operation { + yield* sleep(12_000); + const directory = join(root.repository, PAIRED_DIRECTORY, execution); + const proc = `/proc/${pid}`; + const sessions = join(root.data, "acpx", "sessions"); + const sessionNames = yield* until(readdir(sessions).catch(() => [])); + const sessionFiles: Record = {}; + for (const name of sessionNames.slice(0, 8)) { + sessionFiles[name] = yield* snapshot(join(sessions, name)); + } + const report = { + pid, + process: yield* until(readFile(join(proc, "status"), "utf8").catch(String)), + waiting: yield* until(readFile(join(proc, "wchan"), "utf8").catch(String)), + stdout: yield* until(readlink(join(proc, "fd", "1")).catch(String)), + stderr: yield* until(readlink(join(proc, "fd", "2")).catch(String)), + stdoutBytes: captured.stdout.length, + trace: captured.stderr.slice(-4096), + sidekick: yield* snapshot(join(directory, SIDEKICK_FILE)), + entries: yield* snapshot(join(directory, ENTRIES_FILE)), + sessions: sessionFiles, + }; + yield* until(mkdir(artifacts, { recursive: true })); + const target = join(artifacts, `recovery-phase-${pid}.json`); + yield* until(writeFile(target, JSON.stringify(report, null, 2))); + console.error(`Slow recovery phase ${pid} captured in ${target}\n${report.trace}`); +} + +function* snapshot(path: string): Operation { + try { + const file = yield* until(open(path, "r")); + try { + const { size } = yield* until(file.stat()); + const bytes = new Uint8Array(Math.min(size, 65_536)); + const { bytesRead } = yield* until(file.read(bytes, 0, bytes.length, size - bytes.length)); + return new TextDecoder().decode(bytes.subarray(0, bytesRead)); + } finally { + yield* until(file.close()); + } + } catch (error) { + return String(error); + } +} + /** How this runtime launches a TypeScript entrypoint of its own. */ function phaseCommand( script: string, diff --git a/packages/cli/tests/fixtures/repl/issue-883/recovery-phase.ts b/packages/cli/tests/fixtures/repl/issue-883/recovery-phase.ts index b41822c16..90b4b4b88 100644 --- a/packages/cli/tests/fixtures/repl/issue-883/recovery-phase.ts +++ b/packages/cli/tests/fixtures/repl/issue-883/recovery-phase.ts @@ -164,6 +164,13 @@ let holdWhen: ((record: string) => boolean) | undefined; /** The exact record the hold is sitting on, so a case can identify it. */ let holding: string | undefined; const reached = withResolvers(); +const tracing = process.env.XMD_RECOVERY_PHASE_TRACE === "1"; + +function trace(message: string): void { + if (tracing) { + process.stderr.write(`recovery phase: ${message}\n`); + } +} function* useHost(dataRoot: string): Operation { yield* installReplHost({ @@ -173,6 +180,7 @@ function* useHost(dataRoot: string): Operation { appendRecord: async (path, record) => { if (holdWhen?.(record) === true) { holding = record; + trace("target journal append reached"); reached.resolve(); await new Promise(() => {}); } @@ -1101,7 +1109,9 @@ function* phases(): Operation { if (script.stop !== undefined && holdWhen === undefined) { throw new Error(`there is no boundary "${script.stop}"`); } + trace(`chat-say entered with stop ${script.stop}`); const held = yield* useConversing(dataRoot, execution, script); + trace("owner assembled"); const said = script.say ?? "do something"; if (script.stopAtQuestion === true) { // The process ends while a question is pending. Nothing is held in the @@ -1120,11 +1130,16 @@ function* phases(): Operation { } if (holdWhen !== undefined) { yield* spawn(function* (): Operation { - yield* held.owner.message(said); + trace("owner message started"); + const result = yield* held.owner.message(said); + trace( + result.ok ? "owner message returned" : `owner message refused: ${result.error.message}`, + ); }); // The boundary, observed. Returning unwinds the owner, the conversation // and the turn under the write that never happened. yield* reached.operation; + trace("journal boundary observed"); // Observed while the acknowledgement is still held: the provider took the // turn, and its own store already retains the native conversation it // created for it. Both are facts about the world this process is about to @@ -1135,10 +1150,12 @@ function* phases(): Operation { // The scripted fake issues saves without waiting; observe its asserting // save before reading the disk store. yield* held.fake.assertedSessions(1); + trace("native identity save acknowledged"); const asserted: (string | undefined)[] = []; for (const key of conversationKeys) { asserted.push((yield* held.retained(key))?.agentSessionId); } + trace("native identity read back"); return { stopped: script.stop, held: describeHeld(), @@ -1208,5 +1225,6 @@ function* phases(): Operation { await main(function* () { const answer = yield* phases(); + trace("phase returning result before scope teardown"); process.stdout.write(`${JSON.stringify(answer)}\n`); }); From 6a9669436a1ca4963a3f5063e368e1a1776a4760 Mon Sep 17 00:00:00 2001 From: Taras Mankovski Date: Tue, 10 Nov 2026 21:42:16 -0500 Subject: [PATCH 2/3] Run bounded Linux trials for the R2q timeout --- .github/workflows/ci.yml | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 0494adb91..50405aeae 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -742,6 +742,16 @@ jobs: - name: Test run: bun run test:bun ${{ matrix.shard }}/10 + - name: Observe repeated Sidekick recovery cuts + if: ${{ !cancelled() && matrix.shard == 10 }} + timeout-minutes: 8 + run: | + set -euo pipefail + for trial in {1..16}; do + echo "R2q diagnostic trial $trial of 16" + bun test --timeout=300000 -t R2q packages/cli/tests/repl-sidekick-recovery.test.ts + done + # test:bun never loads packages/cli/src/bun.ts, so it cannot tell whether # the Bun entrypoint's API.Env providers work. This document drives a # scenario, which makes the parent relaunch bun.ts as From 2893a6165b3311c095e8ae9329ccc751083357e3 Mon Sep 17 00:00:00 2001 From: Taras Mankovski Date: Tue, 10 Nov 2026 22:08:45 -0500 Subject: [PATCH 3/3] Order recovery fixture reads after queued provider saves The R2 conversation fixture serialized provider saves but let loads read the disk store immediately. The fake ACP does not join its saves, so a later session resolution could miss the record an earlier turn had already queued and refuse the continuation; the cut then waited on a journal write that could no longer happen. A load now waits for the saves queued when it is asked. A cut phase also races its boundary against the message itself, so a message that settles first fails the phase with its answer instead of hanging. The owner stays suspended under the held write when the boundary wins. Removes the hosted R2q trial step and phase instrumentation used to diagnose #1231. Co-Authored-By: Claude Opus 5.5 --- .github/workflows/ci.yml | 10 --- .../cli/tests/fixtures/repl/issue-883/host.ts | 87 +------------------ .../fixtures/repl/issue-883/recovery-phase.ts | 55 ++++++------ .../cli/tests/repl-sidekick-recovery.test.ts | 22 +++++ 4 files changed, 54 insertions(+), 120 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 50405aeae..0494adb91 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -742,16 +742,6 @@ jobs: - name: Test run: bun run test:bun ${{ matrix.shard }}/10 - - name: Observe repeated Sidekick recovery cuts - if: ${{ !cancelled() && matrix.shard == 10 }} - timeout-minutes: 8 - run: | - set -euo pipefail - for trial in {1..16}; do - echo "R2q diagnostic trial $trial of 16" - bun test --timeout=300000 -t R2q packages/cli/tests/repl-sidekick-recovery.test.ts - done - # test:bun never loads packages/cli/src/bun.ts, so it cannot tell whether # the Bun entrypoint's API.Env providers work. This document drives a # scenario, which makes the parent relaunch bun.ts as diff --git a/packages/cli/tests/fixtures/repl/issue-883/host.ts b/packages/cli/tests/fixtures/repl/issue-883/host.ts index 7e0cdeb9f..91cc222da 100644 --- a/packages/cli/tests/fixtures/repl/issue-883/host.ts +++ b/packages/cli/tests/fixtures/repl/issue-883/host.ts @@ -12,21 +12,11 @@ * case brings of its own is the interruption. */ -import { - appendFile, - mkdir, - mkdtemp, - open, - readdir, - readFile, - readlink, - rm, - writeFile, -} from "node:fs/promises"; +import { appendFile, mkdir, mkdtemp, open, rm } from "node:fs/promises"; import { randomBytes } from "node:crypto"; import { tmpdir } from "node:os"; import { join } from "node:path"; -import { ensure, run, sleep, spawn, until } from "effection"; +import { ensure, run, spawn, until } from "effection"; import type { Stream } from "effection"; import { exec } from "@effectionx/process"; import process from "node:process"; @@ -36,11 +26,6 @@ import type { Operation } from "effection"; import { installReplHost } from "../../../../src/repl-assembly.ts"; import { REPL_DIRECTORY } from "../../../../src/repl/storage.ts"; -import { - ENTRIES_FILE, - PAIRED_DIRECTORY, - SIDEKICK_FILE, -} from "../../../../src/repl/sidekick/history.ts"; /** How a gate can hold or fail one real append, by the path it is reaching. */ export interface AppendGate { @@ -192,19 +177,9 @@ export function* runPhase( execution, ...(extra === undefined ? [] : [extra]), ]); - const env = phaseEnv(); - const artifacts = process.env.XMD_TEST_ARTIFACTS; - const diagnose = - phase === "chat-say" && - artifacts !== undefined && - extra !== undefined && - ["delivered", "laterTurn"].includes(JSON.parse(extra).stop); - if (diagnose) { - env.XMD_RECOVERY_PHASE_TRACE = "1"; - } const child = yield* exec(launch.command, { arguments: launch.arguments, - env, + env: phaseEnv(), }); // Collected *while* the child runs, not after it. A subscription taken once // the process has already settled never settles itself, so reading the @@ -212,13 +187,7 @@ export function* runPhase( const captured = { stdout: "", stderr: "" }; yield* spawn(() => accumulate(child.stdout, captured, "stdout")); yield* spawn(() => accumulate(child.stderr, captured, "stderr")); - const observation = diagnose - ? yield* spawn(() => observeSlowPhase(child.pid, root, execution, artifacts, captured)) - : undefined; const status = yield* child.join(); - if (observation !== undefined) { - yield* observation.halt(); - } if (status.code !== 0) { throw new Error( `the ${phase} phase exited ${status.code ?? "on a signal"}: ${captured.stderr}`, @@ -231,56 +200,6 @@ export function* runPhase( return JSON.parse(line) as Record; } -function* observeSlowPhase( - pid: number, - root: RepositoryRoot, - execution: string, - artifacts: string, - captured: { stdout: string; stderr: string }, -): Operation { - yield* sleep(12_000); - const directory = join(root.repository, PAIRED_DIRECTORY, execution); - const proc = `/proc/${pid}`; - const sessions = join(root.data, "acpx", "sessions"); - const sessionNames = yield* until(readdir(sessions).catch(() => [])); - const sessionFiles: Record = {}; - for (const name of sessionNames.slice(0, 8)) { - sessionFiles[name] = yield* snapshot(join(sessions, name)); - } - const report = { - pid, - process: yield* until(readFile(join(proc, "status"), "utf8").catch(String)), - waiting: yield* until(readFile(join(proc, "wchan"), "utf8").catch(String)), - stdout: yield* until(readlink(join(proc, "fd", "1")).catch(String)), - stderr: yield* until(readlink(join(proc, "fd", "2")).catch(String)), - stdoutBytes: captured.stdout.length, - trace: captured.stderr.slice(-4096), - sidekick: yield* snapshot(join(directory, SIDEKICK_FILE)), - entries: yield* snapshot(join(directory, ENTRIES_FILE)), - sessions: sessionFiles, - }; - yield* until(mkdir(artifacts, { recursive: true })); - const target = join(artifacts, `recovery-phase-${pid}.json`); - yield* until(writeFile(target, JSON.stringify(report, null, 2))); - console.error(`Slow recovery phase ${pid} captured in ${target}\n${report.trace}`); -} - -function* snapshot(path: string): Operation { - try { - const file = yield* until(open(path, "r")); - try { - const { size } = yield* until(file.stat()); - const bytes = new Uint8Array(Math.min(size, 65_536)); - const { bytesRead } = yield* until(file.read(bytes, 0, bytes.length, size - bytes.length)); - return new TextDecoder().decode(bytes.subarray(0, bytesRead)); - } finally { - yield* until(file.close()); - } - } catch (error) { - return String(error); - } -} - /** How this runtime launches a TypeScript entrypoint of its own. */ function phaseCommand( script: string, diff --git a/packages/cli/tests/fixtures/repl/issue-883/recovery-phase.ts b/packages/cli/tests/fixtures/repl/issue-883/recovery-phase.ts index 90b4b4b88..165e886b8 100644 --- a/packages/cli/tests/fixtures/repl/issue-883/recovery-phase.ts +++ b/packages/cli/tests/fixtures/repl/issue-883/recovery-phase.ts @@ -24,7 +24,7 @@ import { appendFile, mkdir, open, readFile, rm, writeFile } from "node:fs/promis import { randomBytes } from "node:crypto"; import { join } from "node:path"; import process from "node:process"; -import { ensure, main, Ok, spawn, suspend, until, withResolvers } from "effection"; +import { ensure, main, Ok, race, spawn, suspend, until, withResolvers } from "effection"; import type { Operation, Result } from "effection"; import { parseDurableEvent, serializeDurableEvent } from "@executablemd/durable-streams"; import type { DurableEvent } from "@executablemd/durable-streams"; @@ -64,7 +64,11 @@ import { assembleReplProfile } from "../../../../src/repl-profile.ts"; import { NO_PLUGINS } from "../../../../src/plugin-host.ts"; import { rootReads } from "../../../../src/repl/sidekick/requests.ts"; import { useSidekickOwner } from "../../../../src/repl/sidekick/owner.ts"; -import type { SidekickAssembly, SidekickOwner } from "../../../../src/repl/sidekick/owner.ts"; +import type { + SidekickAssembly, + SidekickOwner, + SidekickTurn, +} from "../../../../src/repl/sidekick/owner.ts"; import type { SubmitReceipt } from "../../../../src/repl/sidekick/controls.ts"; import type { PairedChatStorage } from "../../../../src/repl/sidekick/repository.ts"; import { createFakeAcp, makeRegistry } from "../../../support/fake-acp.ts"; @@ -164,13 +168,6 @@ let holdWhen: ((record: string) => boolean) | undefined; /** The exact record the hold is sitting on, so a case can identify it. */ let holding: string | undefined; const reached = withResolvers(); -const tracing = process.env.XMD_RECOVERY_PHASE_TRACE === "1"; - -function trace(message: string): void { - if (tracing) { - process.stderr.write(`recovery phase: ${message}\n`); - } -} function* useHost(dataRoot: string): Operation { yield* installReplHost({ @@ -180,7 +177,6 @@ function* useHost(dataRoot: string): Operation { appendRecord: async (path, record) => { if (holdWhen?.(record) === true) { holding = record; - trace("target journal append reached"); reached.resolve(); await new Promise(() => {}); } @@ -335,8 +331,10 @@ function* useConversing( yield* ensure(() => until(writes)); // The scripted fake does not await saves; snapshots keep its pending write // from adopting an identity before the queued asserting write publishes it. + // A read waits for the saves already queued when it was asked, so a later + // session resolution sees the record an earlier turn has already saved. const store: typeof disk = { - load: (key) => disk.load(key), + load: (key) => writes.then(() => disk.load(key)), save: (record) => { const snapshot = structuredClone(record); const saving = writes.then(() => disk.save(snapshot)); @@ -1109,9 +1107,7 @@ function* phases(): Operation { if (script.stop !== undefined && holdWhen === undefined) { throw new Error(`there is no boundary "${script.stop}"`); } - trace(`chat-say entered with stop ${script.stop}`); const held = yield* useConversing(dataRoot, execution, script); - trace("owner assembled"); const said = script.say ?? "do something"; if (script.stopAtQuestion === true) { // The process ends while a question is pending. Nothing is held in the @@ -1129,17 +1125,27 @@ function* phases(): Operation { }; } if (holdWhen !== undefined) { - yield* spawn(function* (): Operation { - trace("owner message started"); - const result = yield* held.owner.message(said); - trace( - result.ok ? "owner message returned" : `owner message refused: ${result.error.message}`, - ); - }); + const sending = yield* spawn(() => held.owner.message(said)); // The boundary, observed. Returning unwinds the owner, the conversation - // and the turn under the write that never happened. - yield* reached.operation; - trace("journal boundary observed"); + // and the turn under the write that never happened. A message that + // settles first can no longer reach the boundary, so its answer is the + // phase's failure. Losing the race halts only the observer, not the + // owner still suspended under the held write. + const early = yield* race([ + (function* (): Operation { + yield* reached.operation; + return undefined; + })(), + (function* (): Operation> { + return yield* sending; + })(), + ]); + if (early !== undefined) { + const answer = early.ok + ? `completed with ${early.value.failure ?? "no failure"}` + : `refused: ${early.error.message}`; + throw new Error(`the message settled before the ${script.stop} boundary: ${answer}`); + } // Observed while the acknowledgement is still held: the provider took the // turn, and its own store already retains the native conversation it // created for it. Both are facts about the world this process is about to @@ -1150,12 +1156,10 @@ function* phases(): Operation { // The scripted fake issues saves without waiting; observe its asserting // save before reading the disk store. yield* held.fake.assertedSessions(1); - trace("native identity save acknowledged"); const asserted: (string | undefined)[] = []; for (const key of conversationKeys) { asserted.push((yield* held.retained(key))?.agentSessionId); } - trace("native identity read back"); return { stopped: script.stop, held: describeHeld(), @@ -1225,6 +1229,5 @@ function* phases(): Operation { await main(function* () { const answer = yield* phases(); - trace("phase returning result before scope teardown"); process.stdout.write(`${JSON.stringify(answer)}\n`); }); diff --git a/packages/cli/tests/repl-sidekick-recovery.test.ts b/packages/cli/tests/repl-sidekick-recovery.test.ts index e69a8d07a..7f878baf9 100644 --- a/packages/cli/tests/repl-sidekick-recovery.test.ts +++ b/packages/cli/tests/repl-sidekick-recovery.test.ts @@ -1362,6 +1362,28 @@ describe("Tier R2 — cases 2, 5 and 6: a conversation across real process bound yield* root.discard(); }); + + it("R2q fixture: a message that settles before its cut fails the phase rather than waiting", function* () { + const root = yield* createRepositoryRoot(); + yield* runPhase("chat-open", root, "chat"); + + // One reply that reads nothing, so the exchange has one canonical Prompt + // and the second one this cut holds is never written. + let observed: unknown; + try { + yield* runPhase( + "chat-say", + root, + "chat", + script({ replies: ["Done.\n"], stop: "laterTurn" }), + ); + } catch (error) { + observed = error; + } + expect(String(observed)).toContain("the message settled before the laterTurn boundary"); + + yield* root.discard(); + }); }); describe("Tier R2 — a reservation binds what the writer has acknowledged", () => {