Merge recorded master baseline into Pi 1 production candidate

Integrate 8ec4b84e1c before runtime qualification, preserving the new workspace restore lock, continuation and Docker packaging fixes. Pi profile 11 source and distribution inputs are unchanged by this merge.

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

server/src/services/dashboard-run-selection.ts

@@ -0,0 +1,49 @@
+import type { Db } from "@paperclipai/db";
+import { heartbeatRuns } from "@paperclipai/db";
+import { and, desc, eq, inArray, sql } from "drizzle-orm";
+
+const MAX_CANDIDATES_PER_POOL = 1_000;
+const ACTIVE_STATUSES = ["queued", "running"];
+
+/** Select recent dashboard cards without ranking the company's full run history. */
+export async function selectDashboardRunIds(db: Db, companyId: string, limit: number): Promise<string[]> {
+  if (limit <= 0) return [];
+
+  const columns = {
+    id: heartbeatRuns.id,
+    status: heartbeatRuns.status,
+    createdAt: heartbeatRuns.createdAt,
+    issueId: sql<string | null>`${heartbeatRuns.contextSnapshot} ->> 'issueId'`,
+  };
+  const [activeRuns, recentRuns] = await Promise.all([
+    db.select(columns)
+      .from(heartbeatRuns)
+      .where(and(eq(heartbeatRuns.companyId, companyId), inArray(heartbeatRuns.status, ACTIVE_STATUSES)))
+      .orderBy(desc(heartbeatRuns.createdAt), desc(heartbeatRuns.id))
+      .limit(MAX_CANDIDATES_PER_POOL),
+    db.select(columns)
+      .from(heartbeatRuns)
+      .where(eq(heartbeatRuns.companyId, companyId))
+      .orderBy(desc(heartbeatRuns.createdAt), desc(heartbeatRuns.id))
+      .limit(MAX_CANDIDATES_PER_POOL),
+  ]);
+
+  const candidates = [...new Map([...activeRuns, ...recentRuns].map((run) => [run.id, run])).values()];
+  candidates.sort((a, b) => {
+    const activeOrder = Number(ACTIVE_STATUSES.includes(b.status)) - Number(ACTIVE_STATUSES.includes(a.status));
+    if (activeOrder !== 0) return activeOrder;
+    const createdOrder = b.createdAt.getTime() - a.createdAt.getTime();
+    return createdOrder || (a.id < b.id ? 1 : a.id > b.id ? -1 : 0);
+  });
+
+  const selectedIds: string[] = [];
+  const seenCards = new Set<string>();
+  for (const run of candidates) {
+    const cardKey = run.issueId ? `issue:${run.issueId}` : `run:${run.id}`;
+    if (seenCards.has(cardKey)) continue;
+    seenCards.add(cardKey);
+    selectedIds.push(run.id);
+    if (selectedIds.length === limit) break;
+  }
+  return selectedIds;
+}

server/src/services/explicit-native-continuation.test.ts

@@ -1,3 +1,6 @@
+import { instanceSettingsService } from "./instance-settings.js";
+import { createPostgresWakeQueueAdapter } from "../modules/wake-queue/adapters/postgres.js";
+import { createReleaseIssueExecution } from "../modules/wake-queue/application/use-cases.js";
 import * as nativeExecutor from "./native-runtime/native-session-executor.js";
 import { appendHeartbeatRunEvent } from "./heartbeat-run-events.js";
 import { recordNativeLocalProcessStop, hasNativeLocalProcessStop, PROCESS_START_REQUESTED } from "./native-local-process-stop.js";
@@ -19,6 +22,7 @@ import { buildExecutionContinuation } from "./execution-continuation.js";
 import { heartbeatService, persistHeartbeatRunProcessMetadata, type HeartbeatEnvironmentRuntime } from "./heartbeat.js";
 import { getExecutionBlocker } from "./execution-blocker.js";
 import { createDurableChatWakeupRequest } from "./durable-chat-wakeup.js";
+import { recordNativeFinalizationFailure } from "./native-runtime/native-run-finalizer.js";
 const support = await getEmbeddedPostgresTestSupport();
 (support.supported ? describe : describe.skip)("explicit native conversation continuation", () => {
   let database: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
@@ -1029,6 +1033,188 @@ const support = await getEmbeddedPostgresTestSupport();
         expect(await getExecutionBlocker(db, f.companyId, f.issueId)).not.toBeNull();
       }
     });
+  async function seedResultBeforeTerminalFailure(f: Fixture) {
+    const contractId = randomUUID(), resultId = randomUUID();
+    await db.insert(completionContracts).values({ id: contractId, companyId: f.companyId, issueId: f.issueId,
+      revision: 1, schemaVersion: "paperclip.completion-contract.v1", policyVersion: "test",
+      risk: "standard", completionAuthority: "server_arbiter", incompleteCriteriaPolicy: "preserve_non_terminal",
+      contractJson: {}, canonicalSha256: contractId, createdByActorType: "system", createdByActorId: "test" });
+    await db.update(heartbeatRuns).set({ completionContractId: contractId, errorCode: "provider_transport_failed" })
+      .where(eq(heartbeatRuns.id, f.sourceRunId));
+    // Dickens returned a result before checkpoint/suspension failed. Accepting
+    // that result did not commit its disposition or make the failed run live.
+    await db.insert(nativeRunResults).values({ id: resultId, companyId: f.companyId, issueId: f.issueId,
+      runId: f.sourceRunId, completionContractId: contractId, serverFingerprint: resultId,
+      schemaStatus: "accepted", canonicalSha256: resultId,
+      resultJson: { terminal: { runTerminalState: "succeeded", reportedWorkDisposition: "blocked" } } });
+    await db.update(nativeRunFinalizations).set({ resultId, failureCode: "native_session_retry_exhausted" })
+      .where(eq(nativeRunFinalizations.runId, f.sourceRunId));
+    await db.update(issueRecoveryActions).set({ status: "active", cause: "native_session_retry_exhausted",
+      evidence: { runId: f.sourceRunId, coordinatorAttempt: 3, sourceFailureCode: "provider_transport_failed" },
+    }).where(eq(issueRecoveryActions.sourceIssueId, f.issueId));
+    return resultId;
+  }
+  it.each(["stopped", "controller", "successor", "still_committing", "live_process", "identity_missing"])(
+    "continues after terminal failure with a retained result only when execution stopped (%s)", async scenario => {
+      const f = await seed(), resultId = await seedResultBeforeTerminalFailure(f);
+      await db.update(nativeRunFinalizations).set({
+        phase: scenario === "still_committing" ? "result_accepted" : "terminal_failure",
+        leaseOwner: scenario === "controller" ? "active-controller" : null,
+        failureDetail: scenario === "successor" ? { successorRunId: randomUUID() } : null,
+      }).where(eq(nativeRunFinalizations.runId, f.sourceRunId));
+      if (scenario === "live_process" || scenario === "identity_missing") {
+        await db.update(heartbeatRuns).set({ processPid: scenario === "live_process" ? process.pid : null })
+          .where(eq(heartbeatRuns.id, f.sourceRunId));
+      }
+      const result = await admit(f);
+      expect(result).toEqual(scenario === "stopped" ? { previousRunId: f.sourceRunId, commentId: f.commentId } : null);
+      expect(await getExecutionBlocker(db, f.companyId, f.issueId)).toEqual(scenario === "stopped" ? null : expect.anything());
+      const [coordinator] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.sourceRunId));
+      expect(coordinator).toMatchObject({ resultId, attempt: 3,
+        phase: scenario === "still_committing" ? "result_accepted" : "terminal_failure" });
+      const [source] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.sourceRunId));
+      expect(source).toMatchObject({ status: "failed", errorCode: "provider_transport_failed" });
+      const [savedResult] = await db.select().from(nativeRunResults).where(eq(nativeRunResults.id, resultId));
+      expect(savedResult.resultJson).toEqual({ terminal: { runTerminalState: "succeeded", reportedWorkDisposition: "blocked" } });
+    });
+  it.each(["task", "chat", "chat_with_next_message"])("delivers a queued %s message exactly once after a retained-result failure and delayed remote stop", async kind => {
+    const f = await seed(), resultId = await seedResultBeforeTerminalFailure(f);
+    const settings = instanceSettingsService(db);
+    const experimental = await settings.getExperimental();
+    await settings.updateExperimental({ enableAgentChat: true });
+    try {
+      if (kind !== "task") await db.update(issues).set({ conversationAgentId: f.agentId,
+        conversationUserId: "board", conversationState: "active",
+      }).where(eq(issues.id, f.issueId));
+      // Occupy the agent slot so admission is real without launching a provider.
+      await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, status: "running" });
+      const lease = { id: randomUUID(), companyId: f.companyId, heartbeatRunId: f.sourceRunId,
+        provider: "daytona", providerLeaseId: "dickens-stopped-sandbox" };
+      await db.insert(environmentLeases).values({ ...lease, status: "active", leasePolicy: "ephemeral" });
+      const heartbeat = heartbeatService(db);
+      const wake = await heartbeat.wakeup(f.agentId, {
+        source: "on_demand", triggerDetail: "manual", reason: "issue_commented",
+        requestedByActorType: "user", requestedByActorId: "board",
+        payload: { issueId: f.issueId, commentId: f.commentId },
+        contextSnapshot: { source: "issue.comment", issueId: f.issueId, wakeCommentId: f.commentId },
+      });
+      expect(wake).toBeNull();
+      const [waiting] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, f.companyId));
+      expect(waiting).toMatchObject({ status: "deferred_issue_execution", runId: null,
+        payload: { executionWait: { reason: "remote_cleanup" } } });
+      let nextCommentId: string | null = null;
+      let nextWakeId: string | null = null;
+      if (kind === "chat_with_next_message") {
+        nextCommentId = randomUUID();
+        nextWakeId = randomUUID();
+        await db.insert(issueComments).values({ id: nextCommentId, companyId: f.companyId, issueId: f.issueId,
+          authorType: "user", authorUserId: "board", body: "A separate follow-up." });
+        await db.insert(agentWakeupRequests).values({ id: nextWakeId, companyId: f.companyId, agentId: f.agentId,
+          source: "on_demand", triggerDetail: "manual", reason: "issue_commented", status: "deferred_issue_execution",
+          requestedByActorType: "user", requestedByActorId: "board", payload: {
+            issueId: f.issueId, commentId: nextCommentId,
+            _paperclipWakeContext: { source: "issue.comment", issueId: f.issueId, wakeCommentId: nextCommentId, wakeReason: "issue_commented" },
+          },
+        });
+      }
+      const [source] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.sourceRunId));
+      await heartbeat.resumeRemoteStopComments(source);
+      expect(await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, f.companyId), eq(heartbeatRuns.status, "queued")))).toHaveLength(0);
+      await db.update(environmentLeases).set({ status: "released", releasedAt: new Date(), cleanupStatus: "success",
+        metadata: { remoteExecutionTermination: remoteTerminationReceipt(lease, { providerLeaseId: lease.providerLeaseId, state: "stopped" }) },
+      }).where(eq(environmentLeases.id, lease.id));
+      // Cleanup and periodic recovery can race to dispatch the same receipt.
+      await Promise.all([heartbeat.resumeRemoteStopComments(source), heartbeat.resumeRemoteStopComments(source)]);
+      const successors = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, f.companyId), eq(heartbeatRuns.status, "queued")));
+      expect(successors).toHaveLength(1);
+      expect(successors[0].contextSnapshot).toMatchObject({ forceFreshSession: true, previousRunId: f.sourceRunId, wakeCommentId: f.commentId });
+      if (nextWakeId) {
+        expect(successors[0].contextSnapshot?.wakeCommentIds).not.toContain(nextCommentId);
+        const [nextWake] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, nextWakeId));
+        expect(nextWake.status).toBe("deferred_issue_execution");
+      }
+      // Complete the admitted turn and run the same transactional queue drain
+      // used by finalization. A stale receipt must not create another turn.
+      await db.update(heartbeatRuns).set({ status: "succeeded", startedAt: new Date(), finishedAt: new Date() })
+        .where(eq(heartbeatRuns.id, successors[0].id));
+      await db.update(issues).set({ executionRunId: successors[0].id, status: "in_progress" })
+        .where(eq(issues.id, f.issueId));
+      const release = createReleaseIssueExecution({
+        issueLock: createPostgresWakeQueueAdapter(db, {
+          resolveResponsibleUserId: async () => "board",
+          getRoutineEnv: async () => ({ routineId: null, env: null, responsibleUserId: null }),
+          resolveSessionBeforeForWakeup: async () => null,
+        }),
+        recovery: { escalateStrandedAssignedIssue: async () => {}, escalateStrandedRecoveryIssueInPlace: async () => {} },
+      });
+      const drained = await release({ companyId: f.companyId, runId: successors[0].id, now: new Date(), suppressImmediateRecovery: true });
+      expect(drained.outcome.kind).toBe(nextWakeId ? "promoted" : "released");
+      await heartbeat.resumeRemoteStopComments(source);
+      await release({ companyId: f.companyId, runId: successors[0].id, now: new Date(), suppressImmediateRecovery: true });
+      const afterDrain = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, f.companyId), eq(heartbeatRuns.status, "queued")));
+      expect(afterDrain).toHaveLength(nextWakeId ? 1 : 0);
+      if (nextWakeId) {
+        expect(afterDrain[0]).toMatchObject({ wakeupRequestId: nextWakeId, contextSnapshot: { wakeCommentId: nextCommentId } });
+        expect(afterDrain[0].contextSnapshot?.wakeCommentIds ?? []).not.toContain(f.commentId);
+      }
+      const [delivered] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, waiting.id));
+      expect(delivered).toMatchObject({ status: "coalesced", runId: successors[0].id });
+      const [coordinator] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.sourceRunId));
+      expect(coordinator).toMatchObject({ phase: "terminal_failure", resultId, attempt: 3 });
+      expect(await getExecutionBlocker(db, f.companyId, f.issueId)).toBeNull();
+    } finally {
+      await settings.updateExperimental({ enableAgentChat: experimental.enableAgentChat });
+    }
+  });
+  it("keeps exhausted workspace export on its own repair path after the sandbox stops", async () => {
+    const f = await seed(), resultId = await seedResultBeforeTerminalFailure(f);
+    await db.delete(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, f.issueId));
+    await db.update(nativeRunFinalizations).set({ phase: "result_accepted", attempt: 1, failureCode: null, failureDetail: null })
+      .where(eq(nativeRunFinalizations.runId, f.sourceRunId));
+    await db.update(heartbeatRuns).set({ status: "running", nativePhase: "result_accepted", finishedAt: null })
+      .where(eq(heartbeatRuns.id, f.sourceRunId));
+    const lease = { id: randomUUID(), companyId: f.companyId, heartbeatRunId: f.sourceRunId,
+      provider: "daytona", providerLeaseId: "pending-workspace-export" };
+    await db.insert(environmentLeases).values({ ...lease, status: "active", leasePolicy: "ephemeral" });
+    const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.sourceRunId));
+    await db.update(heartbeatRuns).set({ runnerProfileJson: { ...run.runnerProfileJson,
+      nativeWorkspaceSync: { schema: "paperclip.native-workspace-sync/v1", state: "prepared",
+        descriptorSha256: "a".repeat(64), baselineSha256: "b".repeat(64), finalHostSha256: null,
+        workspaceId: randomUUID(), leaseId: lease.id, providerLeaseId: lease.providerLeaseId,
+        remoteCwd: "/work", resourceDisposition: "destroy" },
+    } }).where(eq(heartbeatRuns.id, f.sourceRunId));
+    for (let attempt = 0; attempt < 3; attempt++) {
+      await recordNativeFinalizationFailure({ db, runId: f.sourceRunId,
+        error: new Error("native_workspace_sync_out_failed"), failureScope: "workspace", projectRunStatus: true });
+    }
+    const [retainedLease] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, lease.id));
+    expect(retainedLease).toMatchObject({ status: "pending_cleanup", metadata: {
+      nativeWorkspaceExportResume: { runId: f.sourceRunId, leaseId: lease.id, resultId },
+    } });
+    await db.update(environmentLeases).set({ status: "released", releasedAt: new Date(), cleanupStatus: "success",
+      metadata: { ...retainedLease.metadata,
+        remoteExecutionTermination: remoteTerminationReceipt(lease, { providerLeaseId: lease.providerLeaseId, state: "stopped" }) },
+    }).where(eq(environmentLeases.id, lease.id));
+    const exportState = () => Promise.all([
+      db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.sourceRunId)),
+      db.select().from(nativeRunResults).where(eq(nativeRunResults.id, resultId)),
+      db.select().from(environmentLeases).where(eq(environmentLeases.id, lease.id)),
+      db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, f.issueId)),
+    ]);
+    const before = await exportState();
+    // Export repair is intentionally excluded before conversation stop checks.
+    // A new message must leave the accepted result, retention, and repair intact.
+    await db.update(issueComments).set({ createdAt: new Date(Date.now() + 1000) }).where(eq(issueComments.id, f.commentId));
+    expect(await admit(f)).toBeNull();
+    expect(await exportState()).toEqual(before);
+    const [coordinator] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.sourceRunId));
+    expect(coordinator).toMatchObject({ phase: "terminal_failure", resultId, attempt: 1,
+      failureCode: "native_workspace_sync_out_retry_exhausted", failureDetail: { workspaceFinalizeAttempt: 3 } });
+    const [action] = await db.select().from(issueRecoveryActions).where(and(
+      eq(issueRecoveryActions.sourceIssueId, f.issueId), eq(issueRecoveryActions.cause, "native_workspace_sync_out_retry_exhausted"),
+    ));
+    expect(action).toMatchObject({ status: "active", cause: "native_workspace_sync_out_retry_exhausted" });
+  });
   it.each(["linked", "revoked", "unlinked", "wrong_author"])(
     "admits a fresh Slack message after failure only with current linked-user authority (%s)", async scenario => {
       const f = await seed();

server/src/services/explicit-native-continuation.ts

@@ -212,9 +212,13 @@ export async function admitExplicitNativeContinuation(input: {
           sql`${nativeRunResults.resultJson}->'terminal'->>'runTerminalState' = 'failed'`,
         )).limit(1)
       : [];
+    // A terminal failure may retain a result accepted before checkpoint or
+    // cleanup failed. That immutable result is history, not an active commit.
+    // Keep it intact and require the same controller/process/lease stop proofs
+    // before admitting new user input; never apply or replay the old result.
     if (!cancelledStartup && coordinator && (
         (coordinator.phase !== "terminal_failure" && !committedFailure) || coordinator.leaseOwner ||
-        (coordinator.resultId && !committedFailure) || coordinator.failureDetail?.successorRunId)) return blocked("controller_settling",
+        coordinator.failureDetail?.successorRunId)) return blocked("controller_settling",
           run.status === "cancelled" && !coordinator.leaseOwner
             ? "The cancelled run still needs verified cleanup. Your message is saved. Inspect the run and its environment for details."
             : "Waiting for the previous run to finish recovery. Your message will start automatically.");

server/src/services/heartbeat.ts

@@ -28524,8 +28524,12 @@ export function heartbeatService(
                   or(isNull(heartbeatRuns.nativeIssueId), eq(heartbeatRuns.nativeIssueId, issue.id)),
                 )).then(rows => rows[0] ?? null)
               : null;
+          const canCoalesceComments = !isConversation(issue) && opts.allowRunCoalescing !== false;
+          // A resumed receipt already passed admission under this issue lock.
+          // Consume it with its successor even when chat keeps other messages
+          // in separate turns, or finalization will deliver it a second time.
           const pendingComments =
-            !isConversation(issue) && opts.allowRunCoalescing !== false &&
+            (executionWaitRequestId || canCoalesceComments) &&
             !(await getExecutionBlocker(tx as unknown as Db, issue.companyId, issue.id))
               ? await tx
                   .select()
@@ -28536,11 +28540,13 @@ export function heartbeatService(
                       inArray(agentWakeupRequests.agentId, handoffSource ? [agentId, handoffSource.agentId] : [agentId]),
                       eq(agentWakeupRequests.status, "deferred_issue_execution"),
                       sql`${agentWakeupRequests.payload}->>'issueId' = ${issue.id}`,
+                      canCoalesceComments ? undefined : eq(agentWakeupRequests.id, executionWaitRequestId!),
                     ),
                   )
                   .orderBy(asc(agentWakeupRequests.requestedAt))
               : [];
           const adoptedComments = pendingComments.filter((wake) => {
+            if (wake.id === executionWaitRequestId) return true;
             if (wake.id === opts.queuedCommentInterruptId || wake.id === opts.queuedCommentRequestId) return true;
             const deferredPayload = parseObject(wake.payload);
             const deferredContext = parseObject(