orch: serialize stale-lock takeover, and resolve a reused branch to its open PR

CONTRIBUTING.md

@@ -9,7 +9,7 @@ Thanks for helping out. This repo is a **port**, not an original work: the skill
 Upstream owns the skill text. This port owns the rewrite of upstream's Cursor mechanics into their T3 form, and nothing else. The rewrite has two parts:
 
 - **Substitutions.** [`tools/substitutions.json`](tools/substitutions.json) rewrites each upstream sentence that names a Cursor mechanism: the `Task` tool, `subagent_type`, model slugs, the `pstack-models.mdc` rule, cloud agents, `/loop`, Cursor paths. Most T3 behavior the port adds, such as the dispatch ledger or the review loop, is an anchored insert: a rule whose pattern is an upstream sentence and whose replacement keeps that sentence and appends the T3 text. A substitution leaves upstream's later edits elsewhere in the file mechanical, so the port prefers it to a fork. Each rule's `rationale` names why it exists, and most cite the section of t3-pstack 1.0.14's `t3-tools.md` it replaces.
-- **Forks.** [`tools/forks.json`](tools/forks.json) declares each file the port keeps in its own form, with a `kind`, a one-sentence `why`, the `since` release, and an `upstream` status. Today that is setup-pstack, which writes the T3 model sheet, the hook points of the watch-pr settle overlay, the files that carry the rules and watch-pr fixes restored from Michael Denyer's pstack-claude, and the worktree audit and cleanup playbook, which fix [cursor/plugins#449](https://github.com/cursor/plugins/issues/449). Every upstream change to a forked file needs a person to merge it, so a fork is the last resort.
+- **Forks.** [`tools/forks.json`](tools/forks.json) declares each file the port keeps in its own form, with a `kind`, a one-sentence `why`, the `since` release, and an `upstream` status. Today that is setup-pstack, which writes the T3 model sheet, the hook points of the watch-pr settle overlay, the files that carry the rules and watch-pr fixes restored from Michael Denyer's pstack-claude, the worktree audit and cleanup playbook, which fix [cursor/plugins#449](https://github.com/cursor/plugins/issues/449), and the orchestrate store, whose lock takeover and reused-branch lookup carry fixes. Every upstream change to a forked file needs a person to merge it, so a fork is the last resort.
 
 The port also owns files upstream does not have: the plugin manifests, the session-start hooks and their routing text, `models.json`, the sheet checker `t3-sheet.mjs`, the watch-pr settle overlay (`settle.ts`), and the transcript lookups reflect uses. The sync lists them as port-only.
 

plugins/pstack/skills/poteto-mode/scripts/orch/orch.test.ts

@@ -8,7 +8,7 @@ import {
   rm,
   writeFile,
 } from "node:fs/promises";
-import { realpathSync } from "node:fs";
+import { realpathSync, unlinkSync, writeFileSync } from "node:fs";
 import { tmpdir } from "node:os";
 import { join } from "node:path";
 import {
@@ -436,6 +436,60 @@ describe("Store", () => {
     expect(await readdir(directory)).not.toContain(".orch.lock");
   });
 
+  it("keeps a lock another writer took over from the same dead holder", async () => {
+    const { directory } = await initializedStore();
+    const lock = join(directory, ".orch.lock");
+    const exited = Bun.spawn(["true"]);
+    await exited.exited;
+    await writeFile(lock, `${exited.pid}\n`);
+    const rival = Bun.spawn(["sleep", "30"]);
+
+    try {
+      const late = useStore(directory, {
+        onStaleLock: () => {
+          unlinkSync(lock);
+          writeFileSync(lock, `${rival.pid}\n`, { flag: "wx" });
+        },
+      });
+      await expect(
+        late.units.add({ id: "u1", track: "build" })
+      ).rejects.toThrow(`store lock held by pid ${rival.pid}`);
+      expect(await readFile(lock, "utf8")).toBe(`${rival.pid}\n`);
+    } finally {
+      rival.kill();
+      await rival.exited;
+    }
+  });
+
+  it("serializes stale-lock takeovers behind a claim a dead writer cannot hold", async () => {
+    const { directory } = await initializedStore();
+    const exited = Bun.spawn(["true"]);
+    await exited.exited;
+    await writeFile(join(directory, ".orch.lock"), `${exited.pid}\n`);
+    const claim = join(directory, ".orch.lock.takeover");
+    await mkdir(claim);
+    const claimant = Bun.spawn(["sleep", "30"]);
+    await writeFile(join(claim, String(claimant.pid)), "");
+
+    try {
+      await expect(
+        useStore(directory).units.add({ id: "u1", track: "build" })
+      ).rejects.toThrow(
+        `store lock held by pid ${exited.pid} is being replaced by another writer; retry`
+      );
+    } finally {
+      claimant.kill();
+      await claimant.exited;
+    }
+
+    const recovered = useStore(directory);
+    expect(
+      await recovered.units.add({ id: "u1", track: "build" })
+    ).toMatchObject({ id: "u1" });
+    expect(await readdir(directory)).not.toContain(".orch.lock.takeover");
+    await recovered.close();
+  });
+
   it("blocks a writer and steals the pid lock only with force", async () => {
     const { directory, store } = await initializedStore();
     await store.close();
@@ -573,6 +627,65 @@ describe("Store", () => {
       });
     });
 
+    it(`resolves a ${forge} branch reused after a merged and a closed PR to its open PR`, async () => {
+      const { directory, store } = await initializedStore();
+      const stack = await makeGitStack(directory);
+
+      await withFakeForge({
+        directory,
+        forge,
+        currentRows: forgeRows(forge, [
+          {
+            number: 11,
+            state: "open",
+            head: "stack/open",
+            base: "stack/closed",
+          },
+          { number: 9, state: "closed", head: "stack/open", base: "main" },
+          { number: 8, state: "merged", head: "stack/open", base: "main" },
+        ]),
+        operation: async () => {
+          expect(
+            (await store.frontier.set({ repo: stack.repo, forge })).prs.map(
+              (row) => row.pr
+            )
+          ).toEqual([10, 13, 11, 14]);
+        },
+      });
+    });
+
+    it(`rejects a ${forge} branch with more than one open PR`, async () => {
+      const { directory, store } = await initializedStore();
+      const stack = await makeGitStack(directory);
+
+      await withFakeForge({
+        directory,
+        forge,
+        currentRows: forgeRows(forge, [
+          {
+            number: 12,
+            state: "draft",
+            head: "stack/open",
+            base: "stack/closed",
+          },
+          {
+            number: 11,
+            state: "open",
+            head: "stack/open",
+            base: "stack/closed",
+          },
+          { number: 9, state: "closed", head: "stack/open", base: "main" },
+        ]),
+        operation: async () => {
+          await expect(
+            store.frontier.set({ repo: stack.repo, forge })
+          ).rejects.toThrow(
+            "branch stack/open has multiple pull requests (#12, #11); close the stale ones first"
+          );
+        },
+      });
+    });
+
     it(`rejects unparseable ${forge} output loudly`, async () => {
       const { directory, store } = await initializedStore();
       const stack = await makeGitStack(directory);

plugins/pstack/skills/poteto-mode/scripts/orch/store.ts

@@ -17,6 +17,7 @@ import { basename, dirname, join, resolve } from "node:path";
 const UNIT_HEADER = "id\ttrack\tstate\tbranch\tpr\tsha\tbrief";
 const LEDGER_HEADER = "pr\tsha\tverdict\tevidence\tverifier\tts";
 const LOCK_FILE = ".orch.lock";
+const TAKEOVER_DIR = ".orch.lock.takeover";
 
 export type Verdict =
   | "live-ui-verified"
@@ -384,24 +385,69 @@ async function acquireLock(
   options: OpenStoreOptions
 ): Promise<() => Promise<void>> {
   const path = join(store, LOCK_FILE);
+  const takeover = join(store, TAKEOVER_DIR);
   const pid = String(process.pid);
   const create = async (): Promise<void> => {
     const handle = await open(path, "wx");
     await handle.writeFile(`${pid}\n`);
     await handle.close();
   };
 
-  const takeOver = async (): Promise<void> => {
-    await unlink(path);
+  // POSIX cannot unlink a file only if it still names the holder judged
+  // replaceable, so the read and unlink run behind a claim: a directory that
+  // holds one file named for the claimant's pid. rename cannot replace a
+  // directory that holds a file, so one writer takes over at a time, and a
+  // dead claimant's file is removed by name, which cannot remove a live one's.
+  const takeOver = async (holder: string): Promise<void> => {
+    const staged = `${takeover}.${pid}.${randomUUID()}`;
+    await mkdir(staged);
     try {
-      await create();
-    } catch (retryError) {
-      if (errorCode(retryError) === "EEXIST") {
-        const retryHolder =
-          (await readFile(path, "utf8")).trim() || "unknown";
-        throw new UserError(`store lock held by pid ${retryHolder}`);
+      await writeFile(join(staged, pid), "");
+      for (const claimant of await readdir(takeover).catch(() => [])) {
+        if (holderIsDead(claimant)) {
+          await rm(join(takeover, claimant), { force: true });
+        }
+      }
+      try {
+        await rename(staged, takeover);
+      } catch (error) {
+        const code = errorCode(error);
+        if (code === "ENOTEMPTY" || code === "EEXIST") {
+          throw new UserError(
+            `store lock held by pid ${holder} is being replaced by another writer; retry`
+          );
+        }
+        throw error;
+      }
+      try {
+        const current = await readFile(path, "utf8").then(
+          (text) => text.trim() || "unknown",
+          (error: unknown) => {
+            if (errorCode(error) === "ENOENT") {
+              return null;
+            }
+            throw error;
+          }
+        );
+        if (current !== null) {
+          if (current !== holder) {
+            throw new UserError(`store lock held by pid ${current}`);
+          }
+          await unlink(path);
+        }
+        await create();
+      } catch (retryError) {
+        if (errorCode(retryError) === "EEXIST") {
+          const retryHolder =
+            (await readFile(path, "utf8")).trim() || "unknown";
+          throw new UserError(`store lock held by pid ${retryHolder}`);
+        }
+        throw retryError;
+      } finally {
+        await rename(takeover, staged);
       }
-      throw retryError;
+    } finally {
+      await rm(staged, { recursive: true, force: true });
     }
   };
 
@@ -419,10 +465,10 @@ async function acquireLock(
     }
     if (holderIsDead(holder)) {
       options.onStaleLock?.(holder);
-      await takeOver();
+      await takeOver(holder);
     } else if (options.force) {
       options.onLockStolen?.(holder);
-      await takeOver();
+      await takeOver(holder);
     } else {
       throw new UserError(`store lock held by pid ${holder}`);
     }
@@ -1119,12 +1165,16 @@ function branchPullRequest({
       `branch ${branch} has no pull request; open one with ${forge} pr create, or resolve the frontier from a branch whose chain is fully submitted`
     );
   }
-  if (rows.length > 1) {
+  // A reused branch keeps its merged and closed PRs, so an open PR is the
+  // branch's current one.
+  const open = rows.filter((item) => item.state === "OPEN");
+  const current = open.length > 0 ? open : rows;
+  if (current.length > 1) {
     throw new UserError(
-      `branch ${branch} has multiple pull requests (${rows.map((item) => `#${item.pr}`).join(", ")}); close the stale ones first`
+      `branch ${branch} has multiple pull requests (${current.map((item) => `#${item.pr}`).join(", ")}); close the stale ones first`
     );
   }
-  return row;
+  return current[0];
 }
 
 function openChildren({

tools/forks.json

@@ -30,6 +30,18 @@
       "since": "2.0.0-rc.7",
       "upstream": "https://github.com/cursor/plugins/issues/449"
     },
+    "plugins/pstack/skills/poteto-mode/scripts/orch/orch.test.ts": {
+      "kind": "policy",
+      "why": "Tests the serialized lock takeover and the reused-branch resolution that store.ts adds.",
+      "since": "2.0.0-rc.16",
+      "upstream": "not-proposed"
+    },
+    "plugins/pstack/skills/poteto-mode/scripts/orch/store.ts": {
+      "kind": "policy",
+      "why": "Runs each stale or forced lock takeover behind a claim directory named for the claimant's pid, so two writers that both read a dead holder cannot both acquire the store lock, and resolves a branch with one open pull request to that one when the branch also has merged or closed pull requests from earlier use.",
+      "since": "2.0.0-rc.16",
+      "upstream": "not-proposed"
+    },
     "plugins/pstack/skills/poteto-mode/scripts/watch-pr/cli.test.ts": {
       "kind": "port-feature",
       "why": "Expects the --settle default that cli.ts adds.",