From 0fb2edfed15a0b29d75005dc7ff53e920615b549 Mon Sep 17 00:00:00 2001 From: Rudy Celekli Date: Fri, 16 Oct 2026 17:38:48 -0400 Subject: [PATCH] fix(pstack): serialize operations sharing an orch store handle --- .../poteto-mode/scripts/orch/orch.test.ts | 89 +++++++++++++++++++ .../skills/poteto-mode/scripts/orch/store.ts | 57 +++++++++++- .../poteto-mode/scripts/orch/orch.test.ts | 89 +++++++++++++++++++ .../skills/poteto-mode/scripts/orch/store.ts | 57 +++++++++++- 4 files changed, 290 insertions(+), 2 deletions(-) diff --git a/harness/pstack/claude-code/skills/poteto-mode/scripts/orch/orch.test.ts b/harness/pstack/claude-code/skills/poteto-mode/scripts/orch/orch.test.ts index 2ef435d7d..cd1c6f3f4 100644 --- a/harness/pstack/claude-code/skills/poteto-mode/scripts/orch/orch.test.ts +++ b/harness/pstack/claude-code/skills/poteto-mode/scripts/orch/orch.test.ts @@ -759,3 +759,92 @@ describe("orch CLI", () => { expect(missingLedger.stderr).toBe(""); }); }); + +describe("Store same-handle ordering", () => { + it("retains concurrent unit writes before a following read", async () => { + const { store } = await initializedStore(); + const writes = Array.from({ length: 12 }, (_, index) => + store.units.add({ id: `parallel-${index}`, track: "build" }) + ); + const read = store.units.list(); + await Promise.all(writes); + expect((await read).map((unit) => unit.id)).toEqual( + Array.from({ length: 12 }, (_, index) => `parallel-${index}`) + ); + }); + + it("retains concurrent ledger, gate and standing-order updates", async () => { + const { store } = await initializedStore(); + await Promise.all(Array.from({ length: 4 }, (_, index) => + store.ledger.record({ + pr: index + 1, + sha: `sha-${index}`, + verdict: "unit-test-verified", + evidence: `evidence-${index}`, + }) + )); + await Promise.all(Array.from({ length: 4 }, (_, index) => + store.gates.park({ + id: `gate-${index}`, + question: "Choose an option", + options: "one or two", + defaultAnswer: "one", + }) + )); + await Promise.all(Array.from({ length: 4 }, (_, index) => + store.standing.add({ line: `Preserve order ${index}` }) + )); + expect(await store.ledger.summary()).toEqual({ "unit-test-verified": 4 }); + expect(await store.gates.list()).toHaveLength(4); + expect((await store.standing.show()).map((row) => row.number)).toEqual([1, 2, 3, 4]); + }); + + it("orders inbox pushes and drains without losing accepted messages", async () => { + const { store } = await initializedStore(); + const first = store.inbox.push({ agent: "owner", unit: "one", status: "done" }); + const second = store.inbox.push({ agent: "owner", unit: "two", status: "done" }); + const drained = store.inbox.drain(); + const remaining = store.inbox.peek(); + const outcomes = await Promise.allSettled([first, second, drained, remaining]); + expect(outcomes.map((result) => result.status)).toEqual([ + "fulfilled", "fulfilled", "fulfilled", "fulfilled", + ]); + expect((await drained).map((row) => row.unit).sort()).toEqual(["one", "two"]); + expect(await remaining).toEqual([]); + }); + + it("a rejected write does not poison later queued operations", async () => { + const { store } = await initializedStore(); + await store.units.add({ id: "existing", track: "build" }); + const outcomes = await Promise.allSettled([ + store.units.add({ id: "existing", track: "build" }), + store.units.add({ id: "following", track: "build" }), + ]); + expect(outcomes.map((result) => result.status)).toEqual(["rejected", "fulfilled"]); + expect((await store.units.list()).map((unit) => unit.id)).toEqual(["existing", "following"]); + }); + + it("close waits for accepted initialization and writes before releasing the lock", async () => { + const directory = await makeDirectory(); + const store = useStore(directory); + const initialized = store.init(); + const written = store.units.add({ id: "accepted", track: "build" }); + const closed = store.close(); + const rejected = store.units.add({ id: "too-late", track: "build" }); + const outcomes = await Promise.allSettled([initialized, written, closed, rejected]); + expect(outcomes.map((result) => result.status)).toEqual([ + "fulfilled", "fulfilled", "fulfilled", "rejected", + ]); + const late = outcomes[3]; + if (late?.status === "rejected") { + expect(String(late.reason)).toContain("store is closed"); + } + expect(await readFile(join(directory, "units.tsv"), "utf8")).toContain("accepted"); + expect(await readdir(directory)).not.toContain(".orch.lock"); + const reopened = useStore(directory); + await reopened.units.add({ id: "new-handle", track: "build" }); + expect((await reopened.units.list()).map((unit) => unit.id)).toEqual(["accepted", "new-handle"]); + await expect(store.units.list()).rejects.toThrow("store is closed"); + await store.close(); + }); +}); diff --git a/harness/pstack/claude-code/skills/poteto-mode/scripts/orch/store.ts b/harness/pstack/claude-code/skills/poteto-mode/scripts/orch/store.ts index 417368918..8f7ea4352 100644 --- a/harness/pstack/claude-code/skills/poteto-mode/scripts/orch/store.ts +++ b/harness/pstack/claude-code/skills/poteto-mode/scripts/orch/store.ts @@ -1309,7 +1309,7 @@ export function openStore( await ensureLock(); }; - return { + const api: Store = { units: { add: async (params) => { await beginWrite(); @@ -1674,4 +1674,59 @@ export function openStore( } }, }; + + // The PID lock excludes other handles, but operations on this handle can + // still interleave their read/modify/write steps. Queue complete operations. + let pending: Promise = Promise.resolve(); + let closing: Promise | null = null; + const enqueue = ( + operation: (...args: Args) => Promise + ): ((...args: Args) => Promise) => (...args) => { + if (closing !== null) { + return Promise.reject(new UserError("store is closed")); + } + const result = pending.then(() => operation(...args)); + pending = result.then(() => undefined, () => undefined); + return result; + }; + + return { + units: { + add: enqueue(api.units.add), + set: enqueue(api.units.set), + get: enqueue(api.units.get), + list: enqueue(api.units.list), + counts: enqueue(api.units.counts), + }, + ledger: { + record: enqueue(api.ledger.record), + check: enqueue(api.ledger.check), + summary: enqueue(api.ledger.summary), + }, + inbox: { + push: enqueue(api.inbox.push), + drain: enqueue(api.inbox.drain), + peek: enqueue(api.inbox.peek), + count: enqueue(api.inbox.count), + }, + gates: { + park: enqueue(api.gates.park), + list: enqueue(api.gates.list), + resolve: enqueue(api.gates.resolve), + }, + frontier: { + set: enqueue(api.frontier.set), + show: enqueue(api.frontier.show), + }, + standing: { + show: enqueue(api.standing.show), + add: enqueue(api.standing.add), + }, + status: { render: enqueue(api.status.render) }, + init: enqueue(api.init), + close: () => { + closing ??= pending.then(() => api.close()); + return closing; + }, + }; } diff --git a/pstack/skills/poteto-mode/scripts/orch/orch.test.ts b/pstack/skills/poteto-mode/scripts/orch/orch.test.ts index 2ef435d7d..cd1c6f3f4 100644 --- a/pstack/skills/poteto-mode/scripts/orch/orch.test.ts +++ b/pstack/skills/poteto-mode/scripts/orch/orch.test.ts @@ -759,3 +759,92 @@ describe("orch CLI", () => { expect(missingLedger.stderr).toBe(""); }); }); + +describe("Store same-handle ordering", () => { + it("retains concurrent unit writes before a following read", async () => { + const { store } = await initializedStore(); + const writes = Array.from({ length: 12 }, (_, index) => + store.units.add({ id: `parallel-${index}`, track: "build" }) + ); + const read = store.units.list(); + await Promise.all(writes); + expect((await read).map((unit) => unit.id)).toEqual( + Array.from({ length: 12 }, (_, index) => `parallel-${index}`) + ); + }); + + it("retains concurrent ledger, gate and standing-order updates", async () => { + const { store } = await initializedStore(); + await Promise.all(Array.from({ length: 4 }, (_, index) => + store.ledger.record({ + pr: index + 1, + sha: `sha-${index}`, + verdict: "unit-test-verified", + evidence: `evidence-${index}`, + }) + )); + await Promise.all(Array.from({ length: 4 }, (_, index) => + store.gates.park({ + id: `gate-${index}`, + question: "Choose an option", + options: "one or two", + defaultAnswer: "one", + }) + )); + await Promise.all(Array.from({ length: 4 }, (_, index) => + store.standing.add({ line: `Preserve order ${index}` }) + )); + expect(await store.ledger.summary()).toEqual({ "unit-test-verified": 4 }); + expect(await store.gates.list()).toHaveLength(4); + expect((await store.standing.show()).map((row) => row.number)).toEqual([1, 2, 3, 4]); + }); + + it("orders inbox pushes and drains without losing accepted messages", async () => { + const { store } = await initializedStore(); + const first = store.inbox.push({ agent: "owner", unit: "one", status: "done" }); + const second = store.inbox.push({ agent: "owner", unit: "two", status: "done" }); + const drained = store.inbox.drain(); + const remaining = store.inbox.peek(); + const outcomes = await Promise.allSettled([first, second, drained, remaining]); + expect(outcomes.map((result) => result.status)).toEqual([ + "fulfilled", "fulfilled", "fulfilled", "fulfilled", + ]); + expect((await drained).map((row) => row.unit).sort()).toEqual(["one", "two"]); + expect(await remaining).toEqual([]); + }); + + it("a rejected write does not poison later queued operations", async () => { + const { store } = await initializedStore(); + await store.units.add({ id: "existing", track: "build" }); + const outcomes = await Promise.allSettled([ + store.units.add({ id: "existing", track: "build" }), + store.units.add({ id: "following", track: "build" }), + ]); + expect(outcomes.map((result) => result.status)).toEqual(["rejected", "fulfilled"]); + expect((await store.units.list()).map((unit) => unit.id)).toEqual(["existing", "following"]); + }); + + it("close waits for accepted initialization and writes before releasing the lock", async () => { + const directory = await makeDirectory(); + const store = useStore(directory); + const initialized = store.init(); + const written = store.units.add({ id: "accepted", track: "build" }); + const closed = store.close(); + const rejected = store.units.add({ id: "too-late", track: "build" }); + const outcomes = await Promise.allSettled([initialized, written, closed, rejected]); + expect(outcomes.map((result) => result.status)).toEqual([ + "fulfilled", "fulfilled", "fulfilled", "rejected", + ]); + const late = outcomes[3]; + if (late?.status === "rejected") { + expect(String(late.reason)).toContain("store is closed"); + } + expect(await readFile(join(directory, "units.tsv"), "utf8")).toContain("accepted"); + expect(await readdir(directory)).not.toContain(".orch.lock"); + const reopened = useStore(directory); + await reopened.units.add({ id: "new-handle", track: "build" }); + expect((await reopened.units.list()).map((unit) => unit.id)).toEqual(["accepted", "new-handle"]); + await expect(store.units.list()).rejects.toThrow("store is closed"); + await store.close(); + }); +}); diff --git a/pstack/skills/poteto-mode/scripts/orch/store.ts b/pstack/skills/poteto-mode/scripts/orch/store.ts index 417368918..8f7ea4352 100644 --- a/pstack/skills/poteto-mode/scripts/orch/store.ts +++ b/pstack/skills/poteto-mode/scripts/orch/store.ts @@ -1309,7 +1309,7 @@ export function openStore( await ensureLock(); }; - return { + const api: Store = { units: { add: async (params) => { await beginWrite(); @@ -1674,4 +1674,59 @@ export function openStore( } }, }; + + // The PID lock excludes other handles, but operations on this handle can + // still interleave their read/modify/write steps. Queue complete operations. + let pending: Promise = Promise.resolve(); + let closing: Promise | null = null; + const enqueue = ( + operation: (...args: Args) => Promise + ): ((...args: Args) => Promise) => (...args) => { + if (closing !== null) { + return Promise.reject(new UserError("store is closed")); + } + const result = pending.then(() => operation(...args)); + pending = result.then(() => undefined, () => undefined); + return result; + }; + + return { + units: { + add: enqueue(api.units.add), + set: enqueue(api.units.set), + get: enqueue(api.units.get), + list: enqueue(api.units.list), + counts: enqueue(api.units.counts), + }, + ledger: { + record: enqueue(api.ledger.record), + check: enqueue(api.ledger.check), + summary: enqueue(api.ledger.summary), + }, + inbox: { + push: enqueue(api.inbox.push), + drain: enqueue(api.inbox.drain), + peek: enqueue(api.inbox.peek), + count: enqueue(api.inbox.count), + }, + gates: { + park: enqueue(api.gates.park), + list: enqueue(api.gates.list), + resolve: enqueue(api.gates.resolve), + }, + frontier: { + set: enqueue(api.frontier.set), + show: enqueue(api.frontier.show), + }, + standing: { + show: enqueue(api.standing.show), + add: enqueue(api.standing.add), + }, + status: { render: enqueue(api.status.render) }, + init: enqueue(api.init), + close: () => { + closing ??= pending.then(() => api.close()); + return closing; + }, + }; }