fix: fence legacy repair accounting and preserve source identity

server/src/__tests__/heartbeat-accepted-plan-workspace-refresh.test.ts

@@ -112,7 +112,6 @@ describeEmbeddedPostgres("accepted plan workspace refresh", () => {
   }, 20_000);
 
   afterEach(async () => {
-    adapterExecute.mockClear();
     // Await every in-flight background heartbeat run to quiescence before the
     // deletes below. A wakeup claims a run and dispatches its execution
     // fire-and-forget, and that run can dispatch a follow-up wakeup, so a run or
@@ -121,6 +120,7 @@ describeEmbeddedPostgres("accepted plan workspace refresh", () => {
     // wakeup that is still before run registration, which a plain run table
     // status poll cannot see.
     await drainHeartbeatRunsToQuiescence(db, heartbeatService(db));
+    adapterExecute.mockClear();
     while (tempRoots.length > 0) {
       const root = tempRoots.pop();
       if (root) await rm(root, { recursive: true, force: true }).catch(() => undefined);
@@ -535,7 +535,14 @@ describeEmbeddedPostgres("accepted plan workspace refresh", () => {
     });
 
     const heartbeat = heartbeatService(db);
-    adapterExecute.mockImplementationOnce(async () => ({
+    adapterExecute.mockImplementationOnce(async () => {
+      // Planning is awaiting a real confirmation, not a prose-only promise.
+      await db.insert(issueThreadInteractions).values({
+        companyId, issueId: sourceIssueId, kind: "request_confirmation", status: "pending",
+        requestedResolverPolicy: "anyone", effectiveResolverPolicy: "anyone",
+        payload: { version: 1, prompt: "Review the source plan" },
+      });
+      return {
       exitCode: 0,
       signal: null,
       timedOut: false,
@@ -544,7 +551,7 @@ describeEmbeddedPostgres("accepted plan workspace refresh", () => {
       summary: "Realized the planning source workspace.",
       provider: "test",
       model: "test-model",
-    }));
+    }; });
 
     const sourceRun = await heartbeat.wakeup(agentId, {
       source: "automation",
@@ -582,6 +589,8 @@ describeEmbeddedPostgres("accepted plan workspace refresh", () => {
     await runGit(repoRoot, ["push", "origin", "HEAD:master"]);
     await runGit(repoRoot, ["fetch", "origin", "master"]);
 
+    await drainHeartbeatRunsToQuiescence(db, heartbeat);
+    await db.delete(issueThreadInteractions).where(eq(issueThreadInteractions.issueId, sourceIssueId));
     const acceptedPlanRevisionId = await seedAcceptedPlanAcceptance({
       companyId,
       issueId: sourceIssueId,

server/src/__tests__/heartbeat-comment-wake-batching.test.ts

@@ -68,7 +68,7 @@ async function closeDbClient(db: ReturnType<typeof createDb> | undefined) {
   await db?.$client?.end?.({ timeout: 0 });
 }
 
-async function createControlledGatewayServer() {
+async function createControlledGatewayServer(beforeComplete?: (turn: number) => Promise<void>) {
   const server = createServer();
   const wss = new WebSocketServer({ server });
   const agentPayloads: Array<Record<string, unknown>> = [];
@@ -151,6 +151,7 @@ async function createControlledGatewayServer() {
         if (waitCount === 1) {
           await firstWaitGate;
         }
+        await beforeComplete?.(waitCount);
         socket.send(
           JSON.stringify({
             type: "res",
@@ -697,7 +698,9 @@ describeEmbeddedPostgres("heartbeat comment wake batching", () => {
   }, 120_000);
 
   it("cancels an empty deferred comment wake instead of promoting deleted input", async () => {
-    const gateway = await createControlledGatewayServer();
+    const gateway = await createControlledGatewayServer(async turn => {
+      if (turn === 1) await db.update(issues).set({ status: "done" }).where(eq(issues.id, issueId));
+    });
     const companyId = randomUUID();
     const agentId = randomUUID();
     const issueId = randomUUID();
@@ -1644,7 +1647,9 @@ describeEmbeddedPostgres("heartbeat comment wake batching", () => {
   }, 120_000);
 
   it("promotes an interaction continuation with its full authoritative source comment after removing a coalesced self-comment", async () => {
-    const gateway = await createControlledGatewayServer();
+    const gateway = await createControlledGatewayServer(async turn => {
+      if (turn === 2) await db.update(issues).set({ status: "done" }).where(eq(issues.id, issueId));
+    });
     const companyId = randomUUID();
     const agentId = randomUUID();
     const issueId = randomUUID();
@@ -2929,7 +2934,9 @@ describeEmbeddedPostgres("heartbeat comment wake batching", () => {
   }, 120_000);
 
   it("promotes an interaction continuation after removing a coalesced self-authored comment", async () => {
-    const gateway = await createControlledGatewayServer();
+    const gateway = await createControlledGatewayServer(async turn => {
+      if (turn === 2) await db.update(issues).set({ status: "done" }).where(eq(issues.id, issueId));
+    });
     const companyId = randomUUID();
     const agentId = randomUUID();
     const issueId = randomUUID();
@@ -3616,7 +3623,9 @@ describeEmbeddedPostgres("heartbeat comment wake batching", () => {
     }
   }, 120_000);
   it("treats the automatic run summary as fallback-only when the run already posted a comment", async () => {
-    const gateway = await createControlledGatewayServer();
+    const gateway = await createControlledGatewayServer(async turn => {
+      if (turn === 2) await db.update(issues).set({ status: "done" }).where(eq(issues.id, issueId));
+    });
     const companyId = randomUUID();
     const agentId = randomUUID();
     const issueId = randomUUID();
@@ -3714,30 +3723,10 @@ describeEmbeddedPostgres("heartbeat comment wake batching", () => {
       expect(sourceRun?.issueCommentStatus).toBe("satisfied");
       expect(sourceRun?.issueCommentSatisfiedByCommentId).not.toBeNull();
 
-      await waitFor(async () => {
-        const comments = await db
-          .select()
-          .from(issueComments)
-          .where(eq(issueComments.issueId, issueId));
-        const wakeups = await db
-          .select()
-          .from(agentWakeupRequests)
-          .where(
-            and(
-              eq(agentWakeupRequests.companyId, companyId),
-              eq(agentWakeupRequests.agentId, agentId),
-            ),
-          );
-
-        const hasHandoffComment = comments.some(
-          (comment) =>
-            comment.body === SUCCESSFUL_RUN_HANDOFF_REQUIRED_NOTICE_BODY,
-        );
-        const hasHandoffWake = wakeups.some(
-          (wakeup) => wakeup.reason === "finish_successful_run_handoff",
-        );
-        return hasHandoffComment && hasHandoffWake;
-      });
+      await waitFor(async () => (await db.select().from(agentWakeupRequests)
+        .where(eq(agentWakeupRequests.companyId, companyId)))
+        .some(wake => wake.reason === "issue_disposition_repair"));
+      await heartbeat.drainActiveRunExecutions();
 
       const comments = await db
         .select()
@@ -3751,12 +3740,7 @@ describeEmbeddedPostgres("heartbeat comment wake batching", () => {
             comment.body === "Manual completion comment from the run.",
         ),
       ).toBe(true);
-      expect(
-        comments.some(
-          (comment) =>
-            comment.body === SUCCESSFUL_RUN_HANDOFF_REQUIRED_NOTICE_BODY,
-        ),
-      ).toBe(true);
+      expect(comments.some(comment => comment.body === SUCCESSFUL_RUN_HANDOFF_REQUIRED_NOTICE_BODY)).toBe(false);
       expect(
         comments.every((comment) => !comment.body.startsWith("## Run summary")),
       ).toBe(true);
@@ -3776,7 +3760,7 @@ describeEmbeddedPostgres("heartbeat comment wake batching", () => {
       ).toBe(false);
       expect(
         wakeups.some(
-          (wakeup) => wakeup.reason === "finish_successful_run_handoff",
+          (wakeup) => wakeup.reason === "issue_disposition_repair",
         ),
       ).toBe(true);
     } finally {

server/src/__tests__/heartbeat-responsible-user-invariant.test.ts

@@ -87,6 +87,14 @@ describeEmbeddedPostgres("heartbeat responsible-user invariant", () => {
     tempDb = await startEmbeddedPostgresTestDatabase("paperclip-heartbeat-responsible-user-");
     db = createDb(tempDb.connectionString);
     heartbeat = heartbeatService(db);
+    const baseExecute = mockAdapterExecute.getMockImplementation()!;
+    mockAdapterExecute.mockImplementation(async (...args: unknown[]) => {
+      const context = (args[0] as { context?: Record<string, unknown> } | undefined)?.context;
+      if (context?.wakeReason === "issue_disposition_repair" && typeof context.issueId === "string") {
+        await db.update(issues).set({ status: "done" }).where(eq(issues.id, context.issueId));
+      }
+      return baseExecute();
+    });
   }, 20_000);
 
   afterEach(async () => {
@@ -301,6 +309,7 @@ describeEmbeddedPostgres("heartbeat responsible-user invariant", () => {
 
     const sourceRunIds: string[] = [];
     for (let attempt = 0; attempt < 3; attempt += 1) {
+      await db.update(issues).set({ status: "todo" }).where(eq(issues.id, issueId));
       const wakeReason = "issue_blockers_resolved";
       const run = await heartbeat.wakeup(agentId, {
         source: "automation",
@@ -321,16 +330,16 @@ describeEmbeddedPostgres("heartbeat responsible-user invariant", () => {
       await drainHeartbeatRunsToQuiescence(db, heartbeat);
     }
     // The deliberately disposition-free adapter response schedules one bounded
-    // handoff per source run. Those automatic continuations retain its identity.
+    // repair per source run; the fixture records done on repair. Each retains its identity.
     const runs = await db.select().from(heartbeatRuns);
     const handoffs = runs.filter((run) => !sourceRunIds.includes(run.id));
     expect(handoffs).toHaveLength(3);
     expect(
-      handoffs.map((run) => run.contextSnapshot?.parentRunId).sort(),
+      handoffs.map((run) => run.contextSnapshot?.retryOfRunId).sort(),
     ).toEqual(sourceRunIds.sort());
     for (const handoff of handoffs) {
       expect(handoff.contextSnapshot?.wakeReason).toBe(
-        "finish_successful_run_handoff",
+        "issue_disposition_repair",
       );
       expect(handoff.responsibleUserId).toBe(issueResponsibleUserId);
       expect(handoff.status).toBe("succeeded");
@@ -379,7 +388,10 @@ describeEmbeddedPostgres("heartbeat responsible-user invariant", () => {
       expect(completed?.responsibleUserId).toBe(commenterUserId);
       const [issue] = await db.select().from(issues).where(eq(issues.id, issueId));
       expect(issue?.responsibleUserId).toBe(issueResponsibleUserId);
-      expect(mockAdapterExecute).toHaveBeenCalledTimes(1);
+      await drainHeartbeatRunsToQuiescence(db, heartbeat);
+      const runs = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, companyId));
+      expect(runs.every(row => row.responsibleUserId === commenterUserId)).toBe(true);
+      expect(mockAdapterExecute).toHaveBeenCalledTimes(runs.length);
     },
   );
 

server/src/__tests__/legacy-continuation-authority.test.ts

@@ -16,13 +16,13 @@ describe("legacy continuation persisted authority", () => {
     await db.insert(agents).values({ id: agentId, companyId, name: "Worker", role: "engineer", status: "idle", adapterType: "codex_local", runtimeConfig: { heartbeat: { wakeOnDemand: true } } });
     await db.insert(issues).values({ id: issueId, companyId, title: "Implement export", status: "in_progress", assigneeAgentId: agentId, responsibleUserId: "fixture-owner" });
     await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, invocationSource: "on_demand", status: "succeeded", runtimeMode: "legacy", continuationAttempt, contextSnapshot: { issueId, ...context }, livenessState: "blocked", resultJson: { summary: "All done. Need approval. I will continue." } });
-    const createRecovery = () => recoveryService(db, {
+    const createRecovery = (afterEnqueue?: (run: typeof heartbeatRuns.$inferSelect) => Promise<void>) => recoveryService(db, {
       enqueueWakeup: async (targetAgentId, opts) => db.transaction(async tx => {
         const [wake] = await tx.insert(agentWakeupRequests).values({ companyId, agentId: targetAgentId, source: "automation", reason: opts?.reason, payload: opts?.payload, idempotencyKey: opts?.idempotencyKey, status: "queued" }).returning();
         const [run] = await tx.insert(heartbeatRuns).values({ companyId, agentId: targetAgentId, invocationSource: "automation", status: "queued", runtimeMode: "legacy", wakeupRequestId: wake.id, contextSnapshot: opts?.contextSnapshot }).returning();
         await tx.update(agentWakeupRequests).set({ runId: run.id }).where(eq(agentWakeupRequests.id, wake.id));
         return run;
-      }),
+      }).then(async run => { await afterEnqueue?.(run); return run; }),
     });
     const runs = () => db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, companyId));
     const actions = () => db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.companyId, companyId));
@@ -43,6 +43,17 @@ describe("legacy continuation persisted authority", () => {
     const repair = (await f.runs()).find(r => r.id !== f.runId)!;
     expect(repair.contextSnapshot).toMatchObject({ legacyDispositionEpisode: { id: f.runId, attempt: 1, maxAttempts: 2 } });
   });
+  it("does not roll back the ledger when a fast repair finishes before enqueue returns", async () => {
+    const f = await fixture();
+    await f.createRecovery(async first => {
+      await f.finish(first);
+      await f.createRecovery().reconcileLegacyContinuation(first.id);
+      expect((await f.actions())[0].attemptCount).toBe(2);
+    }).reconcileLegacyContinuation(f.runId);
+    const second = (await f.runs()).find(r => r.status === "scheduled_retry")!;
+    expect((await f.actions())[0]).toMatchObject({ attemptCount: 2, wakePolicy: { scheduledRunId: second.id, attempt: 2 } });
+    expect(await f.runs()).toHaveLength(3);
+  });
   it("keeps the same bounded episode across restart and commentary, then escalates only after agent attempts", async () => {
     const f = await fixture();
     await f.createRecovery().reconcileLegacyContinuation(f.runId);
@@ -59,6 +70,21 @@ describe("legacy continuation persisted authority", () => {
     await f.createRecovery().reconcileLegacyContinuation(second.id);
     expect(await f.runs()).toHaveLength(3);
   });
+  it("persists the source identity while the second repair waits to dispatch", async () => {
+    const f = await fixture();
+    await f.createRecovery().reconcileLegacyContinuation(f.runId);
+    const first = (await f.runs()).find(r => r.id !== f.runId)!;
+    await db.update(heartbeatRuns).set({ responsibleUserId: "initiating-operator" }).where(eq(heartbeatRuns.id, first.id));
+    await f.finish(first);
+    await f.createRecovery().reconcileLegacyContinuation(first.id);
+    expect((await f.runs()).find(r => r.status === "scheduled_retry")?.responsibleUserId).toBe("initiating-operator");
+  });
+  it.each(["retry_queued", "retry_exhausted"])("does not stack a new repair budget on the %s comment policy", async issueCommentStatus => {
+    const f = await fixture();
+    await db.update(heartbeatRuns).set({ issueCommentStatus }).where(eq(heartbeatRuns.id, f.runId));
+    expect(await f.createRecovery().reconcileLegacyContinuation(f.runId)).toBe("skipped");
+    expect(await f.runs()).toHaveLength(1);
+  });
   it("does not grant a new budget to exhausted pre-upgrade continuation", async () => {
     const f = await fixture({}, 2);
     expect(await f.createRecovery().reconcileLegacyContinuation(f.runId)).toBe("escalated");

server/src/__tests__/legacy-finalization-regression.test.ts

@@ -104,6 +104,11 @@ describe("P6-32 legacy finalization regression", () => {
   });
 
   it("executes a flag-off heartbeat through the legacy adapter with byte-stable reads and zero native rows", async () => {
+    const execute = adapterExecute.getMockImplementation()!;
+    adapterExecute.mockImplementationOnce(async () => {
+      await db.update(issues).set({ status: "done" }).where(eq(issues.id, issueId));
+      return execute();
+    });
     const heartbeat = heartbeatService(db);
     const queued = await heartbeat.wakeup(agentId, {
       source: "automation",

server/src/__tests__/native-session-resumption.test.ts

@@ -1392,6 +1392,11 @@ describe.each(["unchanged", "newer_active", "stale_idle"] as const)(
         assigneeAgentId: agentId,
         workMode: "standard",
       });
+      const directExecute = legacyAdapterExecute.getMockImplementation()!;
+      legacyAdapterExecute.mockImplementationOnce(async () => {
+        await db.update(issues).set({ status: "done" }).where(eq(issues.id, freshIssueId));
+        return directExecute();
+      });
       const fresh = await heartbeat.wakeup(agentId, {
         source: "automation",
         triggerDetail: "system",

server/src/services/recovery/service.ts

@@ -3093,6 +3093,11 @@ export function recoveryService(
   }) {
     const agentId = input.issue.assigneeAgentId;
     if (!agentId) return null;
+    // Preserve the initiating identity on the durable row, including while a
+    // delayed repair is waiting to dispatch. Never substitute the issue owner.
+    const sourceRun = input.latestRun?.id ? await db.select({ responsibleUserId: heartbeatRuns.responsibleUserId })
+      .from(heartbeatRuns).where(and(eq(heartbeatRuns.id, input.latestRun.id), eq(heartbeatRuns.companyId, input.issue.companyId)))
+      .limit(1).then(rows => rows[0]) : null;
     const timing = dispositionRepairDelayMs(input.attempt, input.fingerprint);
     const now = new Date();
     const retryAt = new Date(now.getTime() + timing.delayMs);
@@ -3208,6 +3213,7 @@ export function recoveryService(
                 scheduledRetryAt: retryAt,
                 scheduledRetryAttempt: input.attempt,
                 scheduledRetryReason: ISSUE_DISPOSITION_REPAIR_RETRY_REASON,
+                responsibleUserId: sourceRun?.responsibleUserId ?? null,
                 contextSnapshot: context,
                 updatedAt: now,
               })
@@ -3254,6 +3260,12 @@ export function recoveryService(
         and(
           eq(issueRecoveryActions.id, input.action.id),
           eq(issueRecoveryActions.companyId, input.issue.companyId),
+          // A fast successor can reserve the next slot before enqueue returns.
+          // An older scheduler must not rewind its ledger or resurrect a wait.
+          eq(issueRecoveryActions.status, "active"),
+          eq(issueRecoveryActions.ownerType, "agent"),
+          eq(issueRecoveryActions.fingerprint, input.fingerprint),
+          sql`${issueRecoveryActions.attemptCount} <= ${input.attempt}`,
         ),
       );
 
@@ -3658,7 +3670,8 @@ export function recoveryService(
         paused: pause, budgetBlocked: budget,
         pendingWait: state.hasDurableWaitingPath || durableWait || parseIssueExecutionState(issue.executionState)?.status === "pending" || Boolean(issue.monitorNextCheckAt),
         activeExecution: state.hasActiveExecutionPath || latest?.id !== (dispatchRunId ?? run.id),
-        ownedLifecycle: Boolean(readNonEmptyString(context.goalControlRequestId)) || context.resumeSessionGoalHeartbeat === true ||
+        ownedLifecycle: run.issueCommentStatus === "retry_queued" || run.issueCommentStatus === "retry_exhausted" ||
+          Boolean(readNonEmptyString(context.goalControlRequestId)) || context.resumeSessionGoalHeartbeat === true ||
           isPluginManagedIssueLifecycle(issue) || routine.length > 0 || workspaceChildren.length > 0 ||
           Boolean(goal[0]?.status && goal[0].status !== "complete") || Boolean(active && !ownsRepair),
         conversation: Boolean(issue.conversationAgentId) || isWaitingConversation(issue),

tests/runner-e2e/lifecycle-live-cases.ts

@@ -60,7 +60,7 @@ export const lifecycleLiveCases = lifecycleNarrativePairs.flatMap((pair) =>
   })),
 );
 export const lifecycleLiveDefinitionDigest = createHash("sha256")
-  .update(JSON.stringify({ version: 4, cases: lifecycleLiveCases }))
+  .update(JSON.stringify({ version: 5, cases: lifecycleLiveCases }))
   .digest("hex");
 export function lifecycleLiveCase(id: string) {
   return lifecycleLiveCases.find((c) => c.id === id);
@@ -107,8 +107,9 @@ export const lifecycleLiveTasks: readonly RunnerTaskFixture[] =
         buildVisibleMarker: nonce => `REPAIRED_${nonce}`,
         buildPrompt: nonce => [
           "This is a bounded two-turn disposition repair exercise. Use the public Paperclip API with normal authorization and run attribution headers.",
-          `On the initial assignment turn only: post exactly one task comment whose complete body is ${JSON.stringify(probe.narrative)}, then end the provider turn with the same text. Leave task status in_progress. Do not record completion, a blocker, an approval, a question, a monitor, children, or a continuation. The quoted words do not describe actual authorization or missing prerequisites.`,
-          `When Paperclip wakes you for task disposition repair: PATCH the current task to status done, with the exact comment REPAIRED_${nonce}. End the provider turn. Do not repeat the initial comment or create further work.`,
+          "Before any mutation, read PAPERCLIP_WAKE_REASON from the process environment. Branch only on that exact value: issue_assigned is INITIAL; issue_disposition_repair is REPAIR. Never infer the phase from the task title, these instructions, or quoted words. Execute only the selected branch, then end this turn.",
+          `INITIAL branch (PAPERCLIP_WAKE_REASON is issue_assigned): post exactly one task comment whose complete body is ${JSON.stringify(probe.narrative)}, then end the provider turn with the same text. Leave task status in_progress. Do not record completion, a blocker, an approval, a question, a monitor, children, or a continuation. The quoted words do not describe actual authorization or missing prerequisites.`,
+          `REPAIR branch (PAPERCLIP_WAKE_REASON is issue_disposition_repair): PATCH the current task to status done, with the exact comment REPAIRED_${nonce}. End the provider turn. Do not repeat the initial comment or create further work.`,
         ].join("\n"),
         buildMatchers: (nonce, execution) => [
           { kind: "message_exact", expected: `REPAIRED_${nonce}` },