From a3bdd03ecfeea9f604f0adf3538ad760007345cd Mon Sep 17 00:00:00 2001 From: Taras Mankovski Date: Tue, 3 Nov 2026 16:06:08 -0500 Subject: [PATCH] =?UTF-8?q?=F0=9F=90=9B=20Release=20passive=20supervision?= =?UTF-8?q?=20barriers=20before=20joining=20fixture=20cleanup?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- packages/cli/tests/repl-sidekick-acp.test.ts | 366 ++++++++++++------- 1 file changed, 235 insertions(+), 131 deletions(-) diff --git a/packages/cli/tests/repl-sidekick-acp.test.ts b/packages/cli/tests/repl-sidekick-acp.test.ts index 704c560b5..ad663757f 100644 --- a/packages/cli/tests/repl-sidekick-acp.test.ts +++ b/packages/cli/tests/repl-sidekick-acp.test.ts @@ -3068,146 +3068,250 @@ describe("direct submission attachment rights", () => { }); }); +function* controlledSupervision(outcome: string) { + const entered = withResolvers(); + const releaseBody = withResolvers(); + const cleaning = withResolvers(); + const releaseCleanup = withResolvers(); + const following = withResolvers(); + const methods: string[] = []; + const resultDirectory = yield* useTemporary("xmd-supervision-result-"); + const resultFile = join(resultDirectory, "result.json"); + let cleaned = false; + const one = yield* world( + { + entryInstallation: { + *install() { + yield* Execution.around({ + *document([request], next) { + yield* ensure(function* () { + cleaning.resolve(); + yield* releaseCleanup.operation; + if (outcome === "completed") { + yield* writeTextFile(resultFile, JSON.stringify({ status: "hold", calls: 1 })); + } + cleaned = true; + }); + entered.resolve(); + yield* releaseBody.operation; + yield* next(request); + }, + }); + }, + }, + }, + () => countWaitRequests(methods, () => following.resolve()), + ); + return { + one, + entered, + releaseBody, + cleaning, + releaseCleanup, + following, + methods, + resultFile, + get cleaned() { + return cleaned; + }, + }; +} + +type ControlledSupervision = + ReturnType extends Operation ? T : never; + +function withControlledSupervision( + outcome: string, + body: (controlled: ControlledSupervision) => Operation, +) { + return scoped(function* () { + const controlled = yield* controlledSupervision(outcome); + try { + yield* body(controlled); + } finally { + // These barriers must open before scope teardown joins the producer and entry. + controlled.releaseBody.resolve(); + controlled.releaseCleanup.resolve(); + } + }); +} + describe("Passive workflow supervision", () => { for (const outcome of ["completed", "failed"]) { it(`waits without polling for a ${outcome} entry and joined cleanup`, function* () { - const entered = withResolvers(); - const releaseBody = withResolvers(); - const cleaning = withResolvers(); - const releaseCleanup = withResolvers(); - const following = withResolvers(); - const progressed = withResolvers(); - const progress: string[] = []; - const methods: string[] = []; - const resultDirectory = yield* useTemporary("xmd-supervision-result-"); - const resultFile = join(resultDirectory, "result.json"); - let cleaned = false; - yield* ensure(() => { + yield* withControlledSupervision(outcome, function* (controlled) { + const { + one, + entered, + releaseBody, + cleaning, + releaseCleanup, + following, + methods, + resultFile, + } = controlled; + const progressed = withResolvers(); + const progress: string[] = []; + const attached = attachResult.parse( + yield* requestAttachment(one.path, { method: "attach", cwd: one.descriptor.cwd }), + ); + const source = + outcome === "completed" + ? "The controlled step result." + : ''; + one.held.fake.script({ reply: fills(source) + "\n" }); + one.held.fake.script({ reply: "The exchange settled." }); + const producer = yield* spawn(() => + requestAttachment(one.path, { + method: "prompt", + attachment: attached.attachment, + text: "Execute one controlled step.", + }), + ); + yield* race([ + entered.operation, + producer, + deadline("The original controlled entry did not start"), + ]); + const reference = one.dispatch.current?.snapshot.message; + if (reference === undefined) { + throw new Error("The step has no acknowledged message"); + } + const mark = methods.length; + let returned = false; + const waiting = yield* spawn(function* () { + const result = yield* waitDocument(one, 1, reference, { + resultFile, + resultSchema: { + type: "object", + properties: { status: { const: "hold" }, calls: { const: 1 } }, + required: ["status", "calls"], + additionalProperties: false, + }, + progress(text) { + progress.push(text); + if (progress.join("").includes("[entry 1] evaluating")) { + progressed.resolve(); + } + }, + }); + returned = true; + return result; + }); + yield* race([ + following.operation, + waiting, + deadline("The wait document did not subscribe"), + ]); + yield* race([ + progressed.operation, + waiting, + deadline("Progress did not arrive before completion"), + ]); + expect(returned).toBe(false); releaseBody.resolve(); + yield* cleaning.operation; + expect(returned).toBe(false); + expect(controlled.cleaned).toBe(false); releaseCleanup.resolve(); + const result = yield* race([waiting, deadline("The document did not receive completion")]); + yield* producer; + expect(controlled.cleaned).toBe(true); + expect(result.status).toBe(outcome); + expect(result.joinedExchange).toBe(true); + expect(result.exchangeOutcome).toBe("completed"); + expect(progress.join("")).toContain("exchange completed"); + expect(progress.join("")).not.toContain("The exchange settled."); + expect(progress.join("").length).toBeLessThan(1500); + if (outcome === "completed") { + expect(result.read).toEqual({ entry: 1, view: "journal" }); + expect(result.resultStatus).toBe("readable"); + expect(result.result).toEqual({ status: "hold", calls: 1 }); + } + if (outcome === "failed") { + expect(result.diagnostic).toContain("controlled step failure"); + } + expect(methods.slice(mark)).toEqual(["hello", "message", "hello", "entry"]); + expect(one.held.fake.prompts).toHaveLength(one.baselineTurns + 2); + const lateMark = methods.length; + const late = yield* waitDocument(one, 1, reference); + expect(late.status).toBe(outcome); + expect(methods.slice(lateMark)).toEqual(["hello", "message", "hello", "entry"]); + expect(one.held.fake.prompts).toHaveLength(one.baselineTurns + 2); + console.log( + JSON.stringify({ + kind: "wait-document-result", + outcome, + cleaned: controlled.cleaned, + requests: methods.slice(mark), + statusPolls: 0, + observerRoleCalls: 0, + alreadyCompletedRead: late.status, + }), + ); }); - const one = yield* world( - { - entryInstallation: { - *install() { - yield* Execution.around({ - *document([request], next) { - yield* ensure(function* () { - cleaning.resolve(); - yield* releaseCleanup.operation; - if (outcome === "completed") { - yield* writeTextFile( - resultFile, - JSON.stringify({ status: "hold", calls: 1 }), - ); - } - cleaned = true; - }); - entered.resolve(); - yield* releaseBody.operation; - yield* next(request); - }, - }); - }, - }, - }, - () => countWaitRequests(methods, () => following.resolve()), - ); - const attached = attachResult.parse( - yield* requestAttachment(one.path, { method: "attach", cwd: one.descriptor.cwd }), - ); - const source = - outcome === "completed" - ? "The controlled step result." - : ''; - one.held.fake.script({ reply: fills(source) + "\n" }); - one.held.fake.script({ reply: "The exchange settled." }); - const producer = yield* spawn(() => - requestAttachment(one.path, { - method: "prompt", - attachment: attached.attachment, - text: "Execute one controlled step.", - }), - ); - yield* race([ - entered.operation, - producer, - deadline("The original controlled entry did not start"), - ]); - const reference = one.dispatch.current?.snapshot.message; - if (reference === undefined) { - throw new Error("The step has no acknowledged message"); - } - const mark = methods.length; - let returned = false; - const waiting = yield* spawn(function* () { - const result = yield* waitDocument(one, 1, reference, { - resultFile, - resultSchema: { - type: "object", - properties: { status: { const: "hold" }, calls: { const: 1 } }, - required: ["status", "calls"], - additionalProperties: false, - }, - progress(text) { - progress.push(text); - if (progress.join("").includes("[entry 1] evaluating")) { - progressed.resolve(); - } - }, + }); + } + + it("exposes an early supervisor failure after joining held entry cleanup", function* () { + const ready = withResolvers(); + const joined = withResolvers(); + let initiating: unknown; + const owner = yield* spawn(function* () { + try { + yield* withControlledSupervision("completed", function* (controlled) { + ready.resolve(controlled); + const { one } = controlled; + const attached = attachResult.parse( + yield* requestAttachment(one.path, { method: "attach", cwd: one.descriptor.cwd }), + ); + one.held.fake.script({ + reply: fills("The held entry.") + "\n", + }); + yield* spawn(() => + requestAttachment(one.path, { + method: "prompt", + attachment: attached.attachment, + text: "Hold the entry through supervisor failure.", + }), + ); + yield* controlled.entered.operation; + try { + yield* waitDocument(one, 1, undefined, { + document: join(one.home, "missing-supervisor.md"), + }); + } catch (error) { + initiating = error; + throw error; + } }); - returned = true; - return result; - }); - yield* race([following.operation, waiting, deadline("The wait document did not subscribe")]); - yield* race([ - progressed.operation, - waiting, - deadline("Progress did not arrive before completion"), - ]); - expect(returned).toBe(false); - releaseBody.resolve(); - yield* cleaning.operation; - expect(returned).toBe(false); - expect(cleaned).toBe(false); - releaseCleanup.resolve(); - const result = yield* race([waiting, deadline("The document did not receive completion")]); - yield* producer; - expect(cleaned).toBe(true); - expect(result.status).toBe(outcome); - expect(result.joinedExchange).toBe(true); - expect(result.exchangeOutcome).toBe("completed"); - expect(progress.join("")).toContain("exchange completed"); - expect(progress.join("")).not.toContain("The exchange settled."); - expect(progress.join("").length).toBeLessThan(1500); - if (outcome === "completed") { - expect(result.read).toEqual({ entry: 1, view: "journal" }); - expect(result.resultStatus).toBe("readable"); - expect(result.result).toEqual({ status: "hold", calls: 1 }); - } - if (outcome === "failed") { - expect(result.diagnostic).toContain("controlled step failure"); + } catch (error) { + joined.resolve(error); } - expect(methods.slice(mark)).toEqual(["hello", "message", "hello", "entry"]); - expect(one.held.fake.prompts).toHaveLength(one.baselineTurns + 2); - const lateMark = methods.length; - const late = yield* waitDocument(one, 1, reference); - expect(late.status).toBe(outcome); - expect(methods.slice(lateMark)).toEqual(["hello", "message", "hello", "entry"]); - expect(one.held.fake.prompts).toHaveLength(one.baselineTurns + 2); - console.log( - JSON.stringify({ - kind: "wait-document-result", - outcome, - cleaned, - requests: methods.slice(mark), - statusPolls: 0, - observerRoleCalls: 0, - alreadyCompletedRead: late.status, - }), - ); }); - } + const controlled = yield* ready.operation; + const settled = yield* race([ + (function* () { + yield* joined.operation; + return true; + })(), + (function* () { + yield* sleep(5000); + return false; + })(), + ]); + // Release from outside the owner so a broken ordering reports an assertion instead of hanging. + controlled.releaseBody.resolve(); + controlled.releaseCleanup.resolve(); + yield* owner; + expect(settled).toBe(true); + expect(yield* joined.operation).toBe(initiating); + expect(initiating).toBeInstanceOf(Error); + if (initiating instanceof Error) { + expect(initiating.message).toContain("missing-supervisor.md"); + } + expect(controlled.cleaned).toBe(true); + }); it("detaching a passive follow leaves the held producer running", function* () { const following = withResolvers();