fix(orch): prevent lost updates from concurrent store operations

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();
+  });
+});

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<unknown> = Promise.resolve();
+  let closing: Promise<void> | null = null;
+  const enqueue = <Args extends unknown[], Result>(
+    operation: (...args: Args) => Promise<Result>
+  ): ((...args: Args) => Promise<Result>) => (...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;
+    },
+  };
 }