fix(server): serialize direct routine status hooks

Co-Authored-By: Paperclip <noreply@paperclip.ing>

doc/plans/2026-10-30-routine-status-sleep.md

@@ -18,7 +18,9 @@ and belong to a routine in that company. This is the existing routine status
 owner contract. Origin labels can change when a plugin declaration changes or
 is removed; retained reciprocal links preserve historical execution custody. Repair
 locks the routine and task and rechecks the linked invocation before writing.
-It preserves skipped/coalesced history and unrelated dispatch failures. A task
+Direct route hooks also lock the execution task and linked invocation in their
+own transaction, nested under caller transactions when present. An older hook
+cannot write after a newer task status has committed. It preserves skipped/coalesced history and unrelated dispatch failures. A task
 that resumes clears an execution-task transient failure through the same owner
 used by native completion and issue routes.
 

server/src/__tests__/routine-execution-status-work.test.ts

@@ -174,6 +174,40 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu
     await db.transaction(async tx => { await notifyRoutineExecutionStatus(tx, otherRun.linkedIssueId!); });
     await projected(otherRun.id, "completed"); await quiet(); expect(errors).toEqual([]);
   });
+  it("serializes a direct status hook with later task changes before projecting the routine", async () => {
+    const f = await fixture(), run = await dispatch(f); await start(); await quiet();
+    await db.update(issues).set({ status: "done" }).where(eq(issues.id, run.linkedIssueId!));
+    let entered!: () => void, release!: () => void;
+    const enteredGate = new Promise<void>(resolve => { entered = resolve; });
+    const releaseGate = new Promise<void>(resolve => { release = resolve; });
+    const holding = db.transaction(async tx => {
+      await tx.select().from(routineRuns).where(eq(routineRuns.id, run.id)).for("update");
+      entered(); await releaseGate;
+    });
+    let direct: Promise<unknown> | undefined, changing: Promise<unknown> | undefined;
+    let directPid: number | undefined;
+    try {
+      await enteredGate; direct = f.service.syncRunStatusForIssue(run.linkedIssueId!);
+      await vi.waitFor(async () => {
+        const rows = await db.execute(sql`select pid from pg_stat_activity
+          where datname = current_database() and wait_event_type = 'Lock'
+            and (query like 'update "routine_runs"%' or query like 'select %"routine_runs"%')`);
+        expect(rows).toHaveLength(1); directPid = Number(rows[0]!.pid);
+      }, { timeout: 5000 });
+      await expect(db.transaction(tx => tx.select({ id: issues.id }).from(issues)
+        .where(eq(issues.id, run.linkedIssueId!)).for("update", { noWait: true })))
+        .rejects.toMatchObject({ cause: expect.objectContaining({ code: "55P03" }) });
+      changing = issueService(db).update(run.linkedIssueId!, { status: "blocked" });
+      await vi.waitFor(async () => {
+        const rows = await db.execute(sql`select count(*)::int as count from pg_stat_activity
+          where datname = current_database() and wait_event_type = 'Lock'
+            and ${directPid}::int = any(pg_blocking_pids(pid))`);
+        expect(Number(rows[0]?.count)).toBeGreaterThan(0);
+      }, { timeout: 5000 });
+      release(); await holding; await direct; await changing;
+      await projected(run.id, "failed"); await quiet(); expect(errors).toEqual([]);
+    } finally { release(); await holding; await direct; await changing; }
+  });
   it("waits for the real outer transaction and forgets a rolled-back nested status write", async () => {
     const f = await fixture(), run = await dispatch(f); await start(); await quiet();
     let release!: () => void, entered!: () => void;

server/src/services/routine-run-status.ts

@@ -51,6 +51,10 @@ export async function finalizeRoutineRun(executor: Db, runId: string, patch: Par
 }
 
 export async function syncRoutineRunStatusForIssue(executor: Db, issueId: string) {
+  return executor.transaction(tx => syncLockedRoutineRunStatusForIssue(tx as unknown as Db, issueId));
+}
+
+async function syncLockedRoutineRunStatusForIssue(executor: Db, issueId: string) {
   const issue = await executor
     .select({
       id: issues.id,
@@ -60,7 +64,8 @@ export async function syncRoutineRunStatusForIssue(executor: Db, issueId: string
       originRunId: issues.originRunId,
     })
     .from(issues)
-    .where(eq(issues.id, issueId))
+    .where(and(eq(issues.id, issueId), eq(issues.originKind, "routine_execution")))
+    .for("update")
     .then((rows) => rows[0] ?? null);
   if (!issue || issue.originKind !== "routine_execution" || !issue.originRunId) return null;
   const run = await executor
@@ -75,6 +80,7 @@ export async function syncRoutineRunStatusForIssue(executor: Db, issueId: string
     .innerJoin(routines, and(eq(routines.id, routineRuns.routineId), eq(routines.companyId, issue.companyId)))
     .where(and(eq(routineRuns.id, issue.originRunId), eq(routineRuns.companyId, issue.companyId),
       eq(routineRuns.linkedIssueId, issue.id)))
+    .for("update", { of: routineRuns })
     .then((rows) => rows[0] ?? null);
   if (!run || ["received", "skipped", "coalesced"].includes(run.status)) return null;
   if (issue.status === "done") {