RomanNum3ral
/
paperclip
mirror of https://github.com/paperclipai/paperclip.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge commit '1c4bcff2b182c6a38d0128a25c6be217dd33f843' into codex/work-folders-base-refresh 

* commit '1c4bcff2b182c6a38d0128a25c6be217dd33f843':
  fix(test): await issue lock before retry race assertions (#13273)
  fix: resume saved user messages after execution recovery (#13270)
This commit is contained in:
Dotta 2026-09-11 17:35:12 -05:00
parent 3dd2b51596 1c4bcff2b1
commit bfa8d00c31
18 changed files with 477 additions and 99 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
doc/SPEC-implementation.md
View File

@ -41,7 +41,7 @@ These decisions close open questions from `SPEC.md` for V1.
| Communication | Tasks + comments only (no separate chat system) |
| Task ownership | Single assignee; atomic checkout required for `in_progress` transition |
| Task watchdogs | A task watchdog is an explicitly configured, issue-subtree-scoped verification and recovery capacity. It may restore live task paths inside the watched subtree; for issue-thread interaction resolution it is an ordinary agent subject to the same audience and containment checks, not board authority, active-run output monitoring, or general liveness recovery. |
| Recovery | Liveness/watchdog recovery preserves explicit ownership: continue interrupted local conversations with bounded fresh turns and preserved history, never replay tool calls automatically; retain native ownership and real execution gates; otherwise open visible source-scoped recovery actions by default, use issue-backed recovery only for independent repair work, or require human escalation (see `doc/execution-semantics.md`) |
| Recovery | Liveness/watchdog recovery preserves explicit ownership: continue interrupted local conversations with bounded fresh turns and preserved history, never replay tool calls automatically; retain native ownership and real execution gates; preserve verified stop evidence and reconsider saved post-stop user messages after cleanup; otherwise open visible source-scoped recovery actions by default, use issue-backed recovery only for independent repair work, or require human escalation (see `doc/execution-semantics.md`) |
| Agent adapters | Built-in `process`, `http`, local CLI/session adapters, and OpenClaw gateway support; external adapters can also be loaded through the adapter plugin flow |
| Plugin framework | Local/self-hosted early plugin runtime is in scope; cloud marketplace and packaged public distribution remain out of scope |
| Auth | Mode-dependent human auth (`local_trusted` implicit board in current code; authenticated mode uses sessions), API keys for agents |

6
doc/execution-semantics.md
View File

@ -840,6 +840,12 @@ Real gates still apply: company and task ownership, active provider ownership, b
An operator Stop waits for provider termination. Remote sandbox providers may return a stopped/deleted receipt after their control-plane operation completes. Paperclip binds that receipt to the company, run, and exact lease; successful file cleanup, a terminal run row, or an in-sandbox shutdown event is not sufficient. Legacy conversational runs receive their cancellation acknowledgement after all remote leases have confirmed termination. Stop alone never creates a continuation. A user message queued during remote cleanup is reconsidered when the provider confirms termination; it still passes normal admission and adopts pending comment IDs in order. Once stopped, the next explicit wake uses the same queue. A compatible saved ACP session can resume, and an unavailable or incompatible session can start fresh with the full task context. Run credentials and scratch paths remain scoped to the new run. A subtree pause requires Resume; a message does not bypass it.
For native conversations, an authenticated user message sent after the previous run finishes can retire its execution recovery holds and start a fresh turn. Hold retirement and the new run are atomic. The previous transcript, tool outcomes, and recovery history remain intact. This starts a new conversation; it does not replay tool calls with unknown outcomes.
Local recovery records a server-authored stop receipt before it clears a verified absent process identity. A new execution request invalidates that receipt before any process can spawn; recording a new process identity also invalidates it. Missing process IDs without a receipt still block admission. Remote execution continues to require termination receipts for every lease.
If cleanup or another execution gate is still pending, the message stays in its existing queue receipt. Startup and periodic scheduling reconsider up to 50 due receipts per pass, at most once per 30 seconds per receipt, without calling a model or resetting recovery attempts. Cleanup callbacks use the same admission path. The issue lock prevents concurrent workers from delivering an adopted or discarded receipt again. The queued-message area shows the current wait reason. Pauses, approvals, budgets, ownership, and external chat authorization remain enforced. A message sent before the run finished does not grant new post-stop authority.
Historical legacy interruption holds for conversational adapters no longer block new messages or Resume. Classification uses the run’s saved adapter invocation or continuation policy, never the agent’s current adapter settings. Missing historical adapter evidence retains the hold. A terminal row with a live predecessor process or unreleased environment lease still blocks actual admission and Resume. Retry scheduling can happen before cleanup, but grants no execution authority. Recovery folds their obsolete no-replay bookkeeping without changing task ownership, status, or automatically waking old work. The audit trail remains readable. Native integrity and ownership holds, and non-conversational adapter holds, remain enforced.
The server projection remains available for diagnostics. Normal working, finishing, and interaction waits add no badges or cards to task lists or feeds. Active transcript headers keep saying Working during automatic retry and execution confirmation; attempts, causes, and recovery decisions belong in the run log. Recovery uses the existing transcript and run log rather than adding a reconciliation form. A cancelled run that never started says “Couldn't start” instead of implying that the agent answered.

15
doc/run-log-events.md
View File

@ -42,6 +42,21 @@ proofs, encryption keys, environment variables, provider credentials, command
arguments, or an unsanitized stderr stream. Detailed failed-attempt diagnostics
remain in the bounded `native_run_finalizations.recovery_history` ledger.
## Native Local Process Stop Evidence
The server writes `native.local_process_stopped` in the same transaction that
clears a local run's process identity, after it verifies that its PID and process
group are absent. The payload contains only those process IDs. Remote process
IDs are never checked against the control-plane host.
The server writes `native.process_start_requested` before a backend can spawn,
and `native.process_identity_recorded` when it stores a new native process
identity. Either invalidates an earlier local stop receipt, including a crash
before the new PID callback. Continuation admission accepts only the latest
server-authored event among these three types; provider
source events cannot supply stop authority. These records stay in the local run
log and do not add Telemetry or OpenTelemetry data.
## Sandbox Startup Run-Log Event
Paperclip writes one `run.startup.step` event to the run log for each bring-up

2
packages/shared/src/types/issue.ts
View File

@ -989,6 +989,8 @@ export interface IssueQueuedCommentQueue {
protocol: IssueQueuedCommentProtocol;
steeringDisposition: IssueQueuedCommentSteeringDisposition;
entries: IssueQueuedCommentEntry[];
/** Current admission condition for a saved user continuation. */
executionWait?: { reason: string; message: string } | null;
}
interface IssueCommentMetadataRowBase {

15
server/src/__tests__/issue-queued-comments-routes.test.ts
View File

@ -213,6 +213,21 @@ describeEmbeddedPostgres("issue queued-comment routes", () => {
return queueRunId;
}
it("projects the recovery wait reason only while the message is deferred", async () => {
const seeded = await seedQueue();
const executionWait = { reason: "remote_cleanup", message: "Waiting for the previous environment to stop." };
await db.update(agentWakeupRequests).set({
payload: sql`coalesce(${agentWakeupRequests.payload}, '{}'::jsonb) || ${JSON.stringify({ executionWait })}::jsonb`,
}).where(eq(agentWakeupRequests.id, seeded.wakeId));
const waiting = await request(app(seeded.companyId)).get(`/api/issues/${seeded.issueId}/queued-comments`);
expect(waiting.status).toBe(200);
expect(waiting.body.executionWait).toEqual(executionWait);
await promoteQueue(seeded);
const admitted = await request(app(seeded.companyId)).get(`/api/issues/${seeded.issueId}/queued-comments`);
expect(admitted.status).toBe(200);
expect(admitted.body.executionWait).toBeUndefined();
});
it("returns the authoritative order, preserves full Markdown edits, and rejects stale revisions", async () => {
const seeded = await seedQueue();
const initial = await request(app(seeded.companyId))

10
server/src/modules/run-dispatch/adapters/postgres.test.ts
View File

@ -319,8 +319,10 @@ describeEmbeddedPostgres("run-dispatch postgres adapter", () => {
.set({ assigneeAgentId: newAssigneeAgentId })
.where(eq(issues.id, issueId));
});
await locked;
return transaction;
// Await lock acquisition before starting the competing operation. Keep
// completion separate so setup does not wait for that operation to finish.
await Promise.race([locked, transaction]);
return { done: transaction };
}
describe("evaluateScheduledRetryGate", () => {
@ -572,7 +574,7 @@ describeEmbeddedPostgres("run-dispatch postgres adapter", () => {
contextSnapshot: { issueId, wakeReason: "issue_assigned" },
});
const holderDone = reassignIssueAndLockRunOnceAConcurrentWaiterBlocks(
const { done: holderDone } = await reassignIssueAndLockRunOnceAConcurrentWaiterBlocks(
issueId,
runId,
replacementAgentId,
@ -733,7 +735,7 @@ describeEmbeddedPostgres("run-dispatch postgres adapter", () => {
// Acquire the issue row lock first and hold it until it observes a
// concurrent `for update` waiter — the promote call below — proving
// this is a real block, not a race the assertion got lucky on.
const holderDone = reassignIssueAndLockRunOnceAConcurrentWaiterBlocks(
const { done: holderDone } = await reassignIssueAndLockRunOnceAConcurrentWaiterBlocks(
issueId,
runId,
newAgentId,

3
server/src/routes/issues.ts
View File

@ -6889,8 +6889,11 @@ export function issueRoutes(
?? (await getNativeSessionSteeringState(steering.steeringRunId)
.then((state) => state.disposition)
.catch(() => "temporarily_unavailable" as const));
const wait = queueState?.state === "deferred" ? readObject(readObject(wake?.payload).executionWait) : {};
return buildQueuedCommentQueueSnapshot({
issueId: input.issue.id,
executionWait: typeof wait.reason === "string" && typeof wait.message === "string"
? { reason: wait.reason, message: wait.message } : null,
queueId: wake?.id ?? null,
state: queueState?.state ?? null,
activeRunId: input.activeRun?.id ?? null,

102
server/src/services/explicit-native-continuation.test.ts
View File

@ -1,3 +1,5 @@
import { appendHeartbeatRunEvent } from "./heartbeat-run-events.js";
import { recordNativeLocalProcessStop, hasNativeLocalProcessStop, PROCESS_START_REQUESTED } from "./native-local-process-stop.js";
import { remoteTerminationReceipt } from "./remote-execution-termination.js";
import { randomUUID } from "node:crypto";
import { and, eq } from "drizzle-orm";
@ -10,7 +12,7 @@ import {
import { startEmbeddedPostgresTestDatabase, getEmbeddedPostgresTestSupport } from "../__tests__/helpers/embedded-postgres.js";
import { admitExplicitNativeContinuation } from "./explicit-native-continuation.js";
import { buildExecutionContinuation } from "./execution-continuation.js";
import { heartbeatService, type HeartbeatEnvironmentRuntime } from "./heartbeat.js";
import { heartbeatService, persistHeartbeatRunProcessMetadata, type HeartbeatEnvironmentRuntime } from "./heartbeat.js";
import { getExecutionBlocker } from "./execution-blocker.js";
const support = await getEmbeddedPostgresTestSupport();
(support.supported ? describe : describe.skip)("explicit native conversation continuation", () => {
@ -47,6 +49,104 @@ const support = await getEmbeddedPostgresTestSupport();
agentId: f.agentId, status: "queued", contextSnapshot: { issueId: f.issueId, previousRunId: result.previousRunId, forceFreshSession: true } });
return result;
});
it("preserves local stop proof after process metadata is cleared and invalidates it on another launch", async () => {
const f = await seed();
const [source] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.sourceRunId));
await db.transaction(async tx => {
expect(await recordNativeLocalProcessStop(tx as unknown as typeof db, source)).toBe(true);
await tx.update(heartbeatRuns).set({ processPid: null }).where(eq(heartbeatRuns.id, source.id));
});
expect(await hasNativeLocalProcessStop(db, f.companyId, source.id)).toBe(true);
expect(await hasNativeLocalProcessStop(db, randomUUID(), source.id)).toBe(false);
expect(await admit(f, true)).toMatchObject({ previousRunId: source.id });
await persistHeartbeatRunProcessMetadata(db, source.id, { pid: 999999999, processGroupId: null, startedAt: new Date().toISOString() });
await db.update(heartbeatRuns).set({ processPid: null }).where(eq(heartbeatRuns.id, source.id));
expect(await admit(f, true)).toBeNull();
expect(await recordNativeLocalProcessStop(db, source)).toBe(true);
await appendHeartbeatRunEvent(db, { companyId: f.companyId, runId: source.id, agentId: f.agentId,
eventType: PROCESS_START_REQUESTED });
// No PID was stored for the new launch, as when the server dies after spawn.
expect(await admit(f, true)).toBeNull();
});
it.each(["live", "remote", "provider_event"])("does not accept invalid local stop proof: %s", async kind => {
const f = await seed();
if (kind === "remote") {
const [environment] = await db.insert(environments).values({ name: "Remote stop", driver: "sandbox" }).returning();
await db.insert(environmentLeases).values({ companyId: f.companyId, heartbeatRunId: f.sourceRunId,
environmentId: environment.id, provider: "daytona", status: "active", leasePolicy: "ephemeral" });
}
if (kind === "live") await db.update(heartbeatRuns).set({ processPid: process.pid }).where(eq(heartbeatRuns.id, f.sourceRunId));
const [source] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.sourceRunId));
if (kind === "provider_event") {
await db.insert(heartbeatRunEvents).values({ companyId: f.companyId, runId: source.id, agentId: f.agentId,
eventType: "native.local_process_stopped", seq: 1, sourceEventId: randomUUID() });
} else expect(await recordNativeLocalProcessStop(db, source)).toBe(false);
expect(await hasNativeLocalProcessStop(db, f.companyId, source.id)).toBe(false);
});
it("resumes saved local messages after restart exactly once and keeps the same wait receipt while blocked", async () => {
const f = await seed();
await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, status: "running" });
await db.update(heartbeatRuns).set({ processPid: process.pid }).where(eq(heartbeatRuns.id, f.sourceRunId));
// A prior cancelled admission is also held, but cannot select the native
// retry source. The newest blocker must win just as it does on Send.
const oldAdmissionId = randomUUID();
await db.insert(heartbeatRuns).values({ id: oldAdmissionId, companyId: f.companyId, agentId: f.agentId,
status: "cancelled", errorCode: "execution_reconciliation_required", contextSnapshot: { issueId: f.issueId },
finishedAt: new Date("2026-09-11T09:00:00Z") });
await db.update(issueRecoveryActions).set({ updatedAt: new Date(0), evidence: { runId: oldAdmissionId,
automaticRecovery: { replay: "blocked", actionOutcome: "unknown" } } }).where(eq(issueRecoveryActions.sourceIssueId, f.issueId));
await db.insert(issueRecoveryActions).values({ companyId: f.companyId, sourceIssueId: f.issueId,
kind: "active_run_watchdog", cause: "uncertain_external_action", fingerprint: randomUUID(), status: "active",
nextAction: "Waiting for the current run.", evidence: { runId: f.sourceRunId } });
expect(await getExecutionBlocker(db, f.companyId, f.issueId)).toMatchObject({ runId: f.sourceRunId });
await heartbeatService(db).wakeup(f.agentId, { source: "automation", triggerDetail: "system", reason: "issue_commented",
requestedByActorType: "user", requestedByActorId: "board", payload: { issueId: f.issueId, commentId: f.commentId },
contextSnapshot: { issueId: f.issueId, wakeCommentId: f.commentId } });
const [waiting] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, f.companyId));
expect(waiting.payload?.executionWait).toMatchObject({ reason: "process_running" });
const makeDue = () => db.update(agentWakeupRequests).set({ updatedAt: new Date(0) }).where(eq(agentWakeupRequests.id, waiting.id));
await makeDue();
await heartbeatService(db).resumeExecutionWaitComments();
expect(await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, f.companyId))).toHaveLength(1);
await db.update(heartbeatRuns).set({ processPid: 999999999 }).where(eq(heartbeatRuns.id, f.sourceRunId));
const [stopped] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.sourceRunId));
await db.transaction(async tx => {
await recordNativeLocalProcessStop(tx as unknown as typeof db, stopped);
await tx.update(heartbeatRuns).set({ processPid: null }).where(eq(heartbeatRuns.id, stopped.id));
});
const holdId = randomUUID();
await db.insert(issueTreeHolds).values({ id: holdId, companyId: f.companyId, rootIssueId: f.issueId, mode: "pause", status: "active" });
await db.insert(issueTreeHoldMembers).values({ companyId: f.companyId, holdId, issueId: f.issueId, depth: 0, issueTitle: "Deploy", issueStatus: "blocked" });
for (let attempt = 0; attempt < 2; attempt++) {
await makeDue();
await heartbeatService(db).resumeExecutionWaitComments();
}
expect(await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, f.companyId))).toHaveLength(1);
expect(await getExecutionBlocker(db, f.companyId, f.issueId)).not.toBeNull();
const [paused] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, waiting.id));
expect(paused.payload?.executionWait).toMatchObject({ reason: "issue_tree_hold_active" });
await db.update(issueTreeHolds).set({ status: "released" }).where(eq(issueTreeHolds.id, holdId));
await db.update(agents).set({ runtimeConfig: { heartbeat: { maxConcurrentRuns: 1, maxDailyRuns: 0 } } }).where(eq(agents.id, f.agentId));
for (let attempt = 0; attempt < 2; attempt++) {
await makeDue();
await heartbeatService(db).resumeExecutionWaitComments();
}
expect(await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, f.companyId))).toHaveLength(1);
expect(await getExecutionBlocker(db, f.companyId, f.issueId)).not.toBeNull();
await db.update(agents).set({ runtimeConfig: { heartbeat: { maxConcurrentRuns: 1 } } }).where(eq(agents.id, f.agentId));
await makeDue();
await Promise.all([heartbeatService(db).resumeExecutionWaitComments(), heartbeatService(db).resumeExecutionWaitComments()]);
const runs = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, f.companyId), eq(heartbeatRuns.status, "queued")));
expect(runs).toHaveLength(1);
expect(runs[0].contextSnapshot).toMatchObject({ forceFreshSession: true, previousRunId: f.sourceRunId,
explicitUserContinuation: { commentId: f.commentId } });
const [adopted] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, waiting.id));
expect(adopted).toMatchObject({ status: "coalesced", runId: runs[0].id });
expect(await getExecutionBlocker(db, f.companyId, f.issueId)).toBeNull();
});
it.each([true, false])("acknowledges a legacy remote Stop only after confirmed lease cleanup: %s", async confirmed => {
const f = await seed();
await db.update(agents).set({ adapterType: "claude_local" }).where(eq(agents.id, f.agentId));

27
server/src/services/explicit-native-continuation.ts
View File

@ -1,3 +1,4 @@
import { hasNativeLocalProcessStop } from "./native-local-process-stop.js";
import { completeTerminatedRemoteNativeSessionCleanup } from "../vendor/paperclip-runner/index.js";
import { hasRemoteTerminationReceipt, remoteLeaseCleanupScope } from "./remote-execution-termination.js";
import { z } from "zod";
@ -29,8 +30,10 @@ export async function admitExplicitNativeContinuation(input: {
actorType: string | null | undefined; actorId: string | null | undefined;
reason: string | null; commentId: string | null; successorRunId: string;
dryRun?: boolean;
onBlocked?: (reason: string, message: string) => void;
}): Promise<{ previousRunId: string; commentId: string } | null> {
const { db, companyId, issueId, agentId, actorId, commentId } = input;
const blocked = (reason: string, message: string) => { input.onBlocked?.(reason, message); return null; };
if (input.actorType !== "user" || !actorId || !commentId ||
!["issue_commented", "issue_reopened_via_comment"].includes(input.reason ?? "")) return null;
if (!z.string().guid().safeParse(commentId).success) return null;
@ -60,20 +63,21 @@ export async function admitExplicitNativeContinuation(input: {
eq(approvals.id, issueApprovals.approvalId), eq(approvals.companyId, companyId),
)).where(and(eq(issueApprovals.companyId, companyId), eq(issueApprovals.issueId, issueId),
inArray(approvals.status, ["pending", "revision_requested"]))).limit(1);
if (pendingInteraction || pendingApproval) return null;
if (pendingInteraction || pendingApproval) return blocked("decision_pending", "A pending approval or question must be resolved before this message can start.");
const sources: Run[] = [];
for (const action of actions) {
const runId = action.evidence.runId ?? action.evidence.sourceRunId;
if (typeof runId !== "string") return null;
if (typeof runId !== "string") return blocked("source_missing", "The stopped run could not be identified. Your message is saved.");
// Text comparison keeps malformed historical evidence a hold, not a UUID cast error.
const [run] = await db.select().from(heartbeatRuns).where(and(
eq(heartbeatRuns.companyId, companyId), sql`${heartbeatRuns.id}::text = ${runId}`,
));
if (!run || run.agentId !== agentId || !terminal.includes(run.status) ||
(run.nativeIssueId ?? run.contextSnapshot?.issueId) !== issueId ||
!run.finishedAt || comment.createdAt <= run.finishedAt) return null;
if (adapterExecutionControls.has(run.id)) return null;
!run.finishedAt) return blocked("source_unavailable", "The previous execution has not finished or its owner changed. Your message is saved.");
if (comment.createdAt <= run.finishedAt) return blocked("message_predates_stop", "This message arrived before the previous run stopped. Send a new message to continue.");
if (adapterExecutionControls.has(run.id)) return blocked("execution_settling", "Waiting for the previous run to stop. Your message will start automatically.");
const unusedAdmission = run.status === "cancelled" && !run.startedAt &&
run.errorCode === "execution_reconciliation_required" &&
!run.processPid && !run.processGroupId && !run.nativeSessionId;
@ -82,7 +86,7 @@ export async function admitExplicitNativeContinuation(input: {
eq(nativeRunFinalizations.companyId, companyId), eq(nativeRunFinalizations.runId, run.id),
)).for("update");
if (coordinator && (coordinator.phase !== "terminal_failure" || coordinator.leaseOwner ||
coordinator.resultId || coordinator.failureDetail?.successorRunId)) return null;
coordinator.resultId || coordinator.failureDetail?.successorRunId)) return blocked("controller_settling", "Waiting for the previous run to finish recovery. Your message will start automatically.");
const leases = await db.select()
.from(environmentLeases).where(and(
eq(environmentLeases.companyId, companyId), eq(environmentLeases.heartbeatRunId, run.id),
@ -90,17 +94,18 @@ export async function admitExplicitNativeContinuation(input: {
const remote = leases.some(lease => lease.provider !== "local");
if (remote) {
// Never interpret remote PIDs using the control-plane host's process table.
if (!leases.every(hasRemoteTerminationReceipt)) return null;
if (!leases.every(hasRemoteTerminationReceipt)) return blocked("remote_cleanup", "Waiting for the previous environment to stop. Your message will start automatically.");
if (!input.dryRun && !leases.every(lease => completeTerminatedRemoteNativeSessionCleanup({
companyId, runId: run.id, remoteCleanupScope: remoteLeaseCleanupScope(lease)!,
}))) return null;
} else {
if (leases.some(lease => !lease.releasedAt || lease.cleanupStatus === "failed")) return null;
if (leases.some(lease => !lease.releasedAt || lease.cleanupStatus === "failed")) return blocked("local_cleanup", "Waiting for the previous environment to finish cleanup. Your message will start automatically.");
if (!unusedAdmission) {
// A missing process identity is not evidence that a provider exited.
if (!run.processPid && !run.processGroupId) return null;
if (run.processPid && !processStopped(run.processPid)) return null;
if (run.processGroupId && !processStopped(-run.processGroupId)) return null;
if (!run.processPid && !run.processGroupId &&
!await hasNativeLocalProcessStop(db, companyId, run.id)) return blocked("process_identity_missing", "The previous run has no verified stop record. Paperclip cannot start this message yet.");
if (run.processPid && !processStopped(run.processPid)) return blocked("process_running", "Waiting for the previous process to stop. Your message will start automatically.");
if (run.processGroupId && !processStopped(-run.processGroupId)) return blocked("process_running", "Waiting for the previous process to stop. Your message will start automatically.");
}
}
sources.push(run);
@ -113,7 +118,7 @@ export async function admitExplicitNativeContinuation(input: {
inArray(heartbeatRuns.status, ["running", "queued", "scheduled_retry"]),
ne(heartbeatRuns.id, input.successorRunId),
)).limit(1);
if (active) return null;
if (active) return blocked("execution_active", "Waiting for the current run. Your message is saved.");
const previous = nativeSources.sort((a, b) => b.createdAt.getTime() - a.createdAt.getTime())[0]!;
// Prove required task history is available before retiring any hold.
await buildExecutionContinuation({ db, companyId, issueId, agentId,

273
server/src/services/heartbeat.ts
View File

@ -1,8 +1,9 @@
import { PROCESS_IDENTITY_RECORDED, recordNativeLocalProcessStop } from "./native-local-process-stop.js";
import { completeTerminatedRemoteNativeSessionCleanup } from "../vendor/paperclip-runner/index.js";
import { remoteExecutionHasStopped, remoteTerminationReceipt, stoppedRemoteCleanupScopes } from "./remote-execution-termination.js";
import { applyConnectorSkills, prepareConnectorSkillDelivery, resolveConnectorAssignments } from "./connector-runtime.js";
import { admitExplicitNativeContinuation } from "./explicit-native-continuation.js";
import { getExecutionBlocker } from "./execution-blocker.js";
import { executionBlockerPredicate, getExecutionBlocker } from "./execution-blocker.js";
import { CONVERSATION_CONTINUATION_POLICY, runUsedConversationAdapter, hasConversationContinuationPolicy, isConversationAdapter } from "./conversation-continuation.js";
import { recordExecutionWait } from "./execution-wait.js";
import {
@ -52,6 +53,7 @@ import {
asc,
desc,
eq,
exists,
getTableColumns,
gt,
gte,
@ -8602,19 +8604,27 @@ export async function persistHeartbeatRunProcessMetadata(
() => null,
);
const startedAt = new Date(observedStartedAt ?? meta.startedAt);
return db
.update(heartbeatRuns)
.set({
processPid: meta.pid,
processGroupId: meta.processGroupId,
processStartedAt: Number.isNaN(startedAt.getTime())
? new Date()
: startedAt,
updatedAt: new Date(),
})
.where(eq(heartbeatRuns.id, runId))
.returning()
.then((rows) => rows[0] ?? null);
return db.transaction(async tx => {
const run = await tx
.update(heartbeatRuns)
.set({
processPid: meta.pid,
processGroupId: meta.processGroupId,
processStartedAt: Number.isNaN(startedAt.getTime())
? new Date()
: startedAt,
updatedAt: new Date(),
})
.where(eq(heartbeatRuns.id, runId))
.returning()
.then((rows) => rows[0] ?? null);
if (run?.runtimeMode === "native") await appendHeartbeatRunEvent(tx as unknown as Db, {
companyId: run.companyId, runId, agentId: run.agentId,
eventType: PROCESS_IDENTITY_RECORDED, stream: "system", level: "info",
message: "Process identity recorded; prior stop evidence no longer applies.",
});
return run;
});
}
async function terminateHeartbeatRunProcess(input: {
@ -9952,9 +9962,9 @@ export function heartbeatService(
}
}
async function resumeRemoteStopComments(run: typeof heartbeatRuns.$inferSelect) {
if (!isHeartbeatRunTerminalStatus(run.status) || adapterExecutionControls.has(run.id) ||
!(await remoteExecutionHasStopped(db, run.companyId, run.id))) return;
async function resumeRemoteStopComments(run: typeof heartbeatRuns.$inferSelect, requestId?: string) {
if (!isHeartbeatRunTerminalStatus(run.status) || adapterExecutionControls.has(run.id)) return;
if (run.runtimeMode !== "native" && !(await remoteExecutionHasStopped(db, run.companyId, run.id))) return;
const issueId = run.nativeIssueId ?? (typeof run.contextSnapshot?.issueId === "string" ? run.contextSnapshot.issueId : null);
if (!issueId) return;
const legacyContinuation = run.runtimeMode === "legacy" && run.status === "cancelled" &&
@ -9965,9 +9975,11 @@ export function heartbeatService(
eq(agentWakeupRequests.companyId, run.companyId), eq(agentWakeupRequests.agentId, run.agentId),
eq(agentWakeupRequests.status, "deferred_issue_execution"),
eq(agentWakeupRequests.requestedByActorType, "user"),
requestId ? eq(agentWakeupRequests.id, requestId) : undefined,
sql`${agentWakeupRequests.payload}->>'issueId' = ${issueId}`,
)).orderBy(asc(agentWakeupRequests.requestedAt));
)).orderBy(asc(agentWakeupRequests.requestedAt)).limit(50);
for (const wake of pending) {
if (wake.idempotencyKey?.startsWith("chat-inbound:")) continue;
const payload = parseObject(wake.payload);
const context = parseObject(payload[DEFERRED_WAKE_CONTEXT_KEY]);
const commentId = deriveCommentId(context, payload);
@ -9981,19 +9993,76 @@ export function heartbeatService(
isNull(issueComments.createdByRunId), gt(issueComments.createdAt, run.finishedAt),
));
if (!comment?.body.trim()) continue;
} else if (!await admitExplicitNativeContinuation({ db, companyId: run.companyId, issueId,
agentId: run.agentId, actorType: wake.requestedByActorType, actorId: wake.requestedByActorId,
reason: wake.reason, commentId, successorRunId: randomUUID(), dryRun: true })) continue;
} else {
let wait = { reason: "execution_recovery", message: "Waiting for execution recovery. Your message is saved." };
const admitted = await admitExplicitNativeContinuation({ db, companyId: run.companyId, issueId,
agentId: run.agentId, actorType: wake.requestedByActorType, actorId: wake.requestedByActorId,
reason: wake.reason, commentId, successorRunId: randomUUID(), dryRun: true,
onBlocked: (reason, message) => { wait = { reason, message }; },
});
if (!admitted) {
await db.update(agentWakeupRequests).set({
payload: sql`jsonb_set(coalesce(${agentWakeupRequests.payload}, '{}'::jsonb), '{executionWait}',
coalesce(${agentWakeupRequests.payload}->'executionWait', '{}'::jsonb) || ${JSON.stringify(wait)}::jsonb)`,
updatedAt: new Date(),
}).where(and(eq(agentWakeupRequests.id, wake.id), eq(agentWakeupRequests.companyId, run.companyId),
eq(agentWakeupRequests.status, "deferred_issue_execution")));
continue;
}
}
// Re-enter ordinary admission with the original user's authority. It
// atomically adopts the deferred comments and still applies every gate.
await enqueueWakeup(run.agentId, { source: wake.source as WakeupOptions["source"], triggerDetail: (wake.triggerDetail ?? undefined) as WakeupOptions["triggerDetail"],
reason: wake.reason, payload, contextSnapshot: context,
requestedByActorType: "user", requestedByActorId: wake.requestedByActorId,
idempotencyKey: `remote-stop-comment:${run.id}:${wake.id}` });
idempotencyKey: `remote-stop-comment:${run.id}:${wake.id}` }, wake.id);
break;
}
}
async function resumeExecutionWaitComments() {
if ((await getSchedulingSuppression()).suppressed) return;
const waits = await db.select({ wake: agentWakeupRequests })
.from(agentWakeupRequests)
.innerJoin(issues, and(eq(issues.companyId, agentWakeupRequests.companyId),
sql`${issues.id}::text = ${agentWakeupRequests.payload}->>'issueId'`,
eq(issues.assigneeAgentId, agentWakeupRequests.agentId)))
.innerJoin(companies, and(eq(companies.id, issues.companyId), eq(companies.status, "active")))
.where(and(exists(db.select({ id: issueRecoveryActions.id }).from(issueRecoveryActions).where(and(
eq(issueRecoveryActions.companyId, issues.companyId), eq(issueRecoveryActions.sourceIssueId, issues.id),
executionBlockerPredicate(),
))), eq(agentWakeupRequests.status, "deferred_issue_execution"),
eq(agentWakeupRequests.requestedByActorType, "user"),
sql`${agentWakeupRequests.payload}->'executionWait' is not null`,
lte(agentWakeupRequests.updatedAt, new Date(Date.now() - 30_000)),
notInArray(issues.status, ["done", "cancelled"])))
.orderBy(asc(agentWakeupRequests.updatedAt)).limit(50);
const seen = new Set<string>();
for (const { wake } of waits) {
const issueId = String(wake.payload?.issueId);
if (seen.has(issueId)) continue;
seen.add(issueId);
// Advance the cursor even for invalid evidence so one damaged task cannot
// starve later requests in this bounded scan. Preserve concurrent edits.
const [claimed] = await db.update(agentWakeupRequests).set({ updatedAt: new Date() }).where(and(
eq(agentWakeupRequests.id, wake.id), eq(agentWakeupRequests.companyId, wake.companyId),
eq(agentWakeupRequests.status, "deferred_issue_execution"),
lte(agentWakeupRequests.updatedAt, new Date(Date.now() - 30_000)),
)).returning({ id: agentWakeupRequests.id });
if (!claimed) continue;
// Match normal admission's deterministic current blocker selection. An
// arbitrary historical action must not choose the retry's source run.
const blocker = await getExecutionBlocker(db, wake.companyId, issueId);
const sourceId = blocker?.runId;
if (!sourceId || !isUuidLike(sourceId)) continue;
const run = await getRun(sourceId);
if (!run || run.companyId !== wake.companyId || run.agentId !== wake.agentId) continue;
await resumeRemoteStopComments(run, wake.id).catch(err => {
logger.warn({ err, runId: run.id }, "failed to resume saved execution-wait message");
});
}
}
async function hasUnsafeTextProjectionDatabase() {
if (!unsafeTextProjectionPromise) {
unsafeTextProjectionPromise = db
@ -18019,31 +18088,35 @@ export function heartbeatService(
});
continue;
}
const cleared = await db
.update(heartbeatRuns)
.set({
processPid: null,
processGroupId: null,
processStartedAt: null,
updatedAt: now,
})
.where(
and(
eq(heartbeatRuns.id, run.id),
eq(heartbeatRuns.runtimeMode, "native"),
run.processPid === null
? isNull(heartbeatRuns.processPid)
: eq(heartbeatRuns.processPid, run.processPid),
run.processGroupId === null
? isNull(heartbeatRuns.processGroupId)
: eq(heartbeatRuns.processGroupId, run.processGroupId),
run.processStartedAt === null
? isNull(heartbeatRuns.processStartedAt)
: eq(heartbeatRuns.processStartedAt, run.processStartedAt),
),
)
.returning({ id: heartbeatRuns.id })
.then((rows) => rows[0] ?? null);
const cleared = await db.transaction(async tx => {
const cleared = await tx
.update(heartbeatRuns)
.set({
processPid: null,
processGroupId: null,
processStartedAt: null,
updatedAt: now,
})
.where(
and(
eq(heartbeatRuns.id, run.id),
eq(heartbeatRuns.runtimeMode, "native"),
run.processPid === null
? isNull(heartbeatRuns.processPid)
: eq(heartbeatRuns.processPid, run.processPid),
run.processGroupId === null
? isNull(heartbeatRuns.processGroupId)
: eq(heartbeatRuns.processGroupId, run.processGroupId),
run.processStartedAt === null
? isNull(heartbeatRuns.processStartedAt)
: eq(heartbeatRuns.processStartedAt, run.processStartedAt),
),
)
.returning({ id: heartbeatRuns.id })
.then((rows) => rows[0] ?? null);
if (cleared) await recordNativeLocalProcessStop(tx as unknown as Db, run);
return cleared;
});
if (cleared) claimableNativeRunIds.add(cleared.id);
}
@ -18477,6 +18550,7 @@ export function heartbeatService(
async function resumeQueuedRuns() {
if ((await getSchedulingSuppression()).suppressed) return;
await resumeExecutionWaitComments();
const cutoff = await getWorktreeExecutionCutoff();
const queuedRuns = await db
@ -19109,31 +19183,35 @@ export function heartbeatService(
}
runningProcesses.delete(run.id);
if (run.processPid || run.processGroupId || run.processStartedAt) {
const cleared = await db
.update(heartbeatRuns)
.set({
processPid: null,
processGroupId: null,
processStartedAt: null,
updatedAt: new Date(),
})
.where(
and(
eq(heartbeatRuns.id, run.id),
eq(heartbeatRuns.runtimeMode, "native"),
run.processPid === null
? isNull(heartbeatRuns.processPid)
: eq(heartbeatRuns.processPid, run.processPid),
run.processGroupId === null
? isNull(heartbeatRuns.processGroupId)
: eq(heartbeatRuns.processGroupId, run.processGroupId),
run.processStartedAt === null
? isNull(heartbeatRuns.processStartedAt)
: eq(heartbeatRuns.processStartedAt, run.processStartedAt),
),
)
.returning()
.then((rows) => rows[0] ?? null);
const cleared = await db.transaction(async tx => {
const cleared = await tx
.update(heartbeatRuns)
.set({
processPid: null,
processGroupId: null,
processStartedAt: null,
updatedAt: new Date(),
})
.where(
and(
eq(heartbeatRuns.id, run.id),
eq(heartbeatRuns.runtimeMode, "native"),
run.processPid === null
? isNull(heartbeatRuns.processPid)
: eq(heartbeatRuns.processPid, run.processPid),
run.processGroupId === null
? isNull(heartbeatRuns.processGroupId)
: eq(heartbeatRuns.processGroupId, run.processGroupId),
run.processStartedAt === null
? isNull(heartbeatRuns.processStartedAt)
: eq(heartbeatRuns.processStartedAt, run.processStartedAt),
),
)
.returning()
.then((rows) => rows[0] ?? null);
if (cleared) await recordNativeLocalProcessStop(tx as unknown as Db, run);
return cleared;
});
if (!cleared) {
const current = await getRun(run.id);
if (current) {
@ -24913,7 +24991,7 @@ export function heartbeatService(
}
}
async function enqueueWakeup(agentId: string, opts: WakeupOptions = {}) {
async function enqueueWakeup(agentId: string, opts: WakeupOptions = {}, executionWaitRequestId?: string) {
const source = opts.source ?? "on_demand";
const triggerDetail = opts.triggerDetail ?? null;
const contextSnapshot: Record<string, unknown> = {
@ -25041,6 +25119,20 @@ export function heartbeatService(
patch: Partial<typeof agentWakeupRequests.$inferInsert> = {},
waitCondition?: Record<string, unknown>,
) => {
if (executionWaitRequestId) {
await db.update(agentWakeupRequests).set({
payload: sql`jsonb_set(coalesce(${agentWakeupRequests.payload}, '{}'::jsonb), '{executionWait}',
coalesce(${agentWakeupRequests.payload}->'executionWait', '{}'::jsonb) || ${JSON.stringify({
reason: skipReason, message: patch.error ?? (skipReason === "issue_tree_hold_active"
? "This task is paused. Resume it to send your saved message."
: "Waiting for task execution to be enabled. Your message is saved."),
})}::jsonb)`,
updatedAt: new Date(),
}).where(and(eq(agentWakeupRequests.id, executionWaitRequestId),
eq(agentWakeupRequests.companyId, agent.companyId), eq(agentWakeupRequests.agentId, agentId),
eq(agentWakeupRequests.status, "deferred_issue_execution")));
return { created: false };
}
const request = {
...durableReceiptFields,
companyId: agent.companyId,
@ -25286,7 +25378,7 @@ export function heartbeatService(
const invokability = await getAgentInvokability(agent);
if (!invokability.invokable) {
if (opts.requestedByActorType !== "user") {
if (opts.requestedByActorType !== "user" || executionWaitRequestId) {
await writeSkippedRequest("agent.not_invokable", {
error: invokability.message,
}, { status: agent.status, reason: invokability.reason });
@ -25400,6 +25492,20 @@ export function heartbeatService(
sql`select id from issues where id = ${issueId} and company_id = ${agent.companyId} for update`,
);
if (executionWaitRequestId) {
const [pending] = await tx.select().from(agentWakeupRequests).where(and(
eq(agentWakeupRequests.id, executionWaitRequestId), eq(agentWakeupRequests.companyId, agent.companyId),
eq(agentWakeupRequests.agentId, agentId), eq(agentWakeupRequests.status, "deferred_issue_execution"),
eq(agentWakeupRequests.requestedByActorType, "user"),
eq(agentWakeupRequests.requestedByActorId, opts.requestedByActorId ?? ""),
sql`${agentWakeupRequests.payload}->>'issueId' = ${issueId}`,
));
// The issue lock serializes cleanup callbacks and periodic workers.
// An adopted, discarded, or edited receipt is no longer authority.
if (!pending || !wakeCommentId || !queuedCommentIdsFromWakePayload(pending.payload).includes(wakeCommentId)) {
return { kind: "deferred" as const };
}
}
let automaticParentRunId: string | null = null;
if (
source === "automation" &&
@ -25704,10 +25810,18 @@ export function heartbeatService(
reconciledSourceRunId = sourceRunId;
}
let continuationWait = { reason: "execution_recovery", message: "Waiting for execution recovery. Your message is saved." };
const deferBlockedExecution = async (
executionBlocker: NonNullable<Awaited<ReturnType<typeof getExecutionBlocker>>>,
) => {
const condition = { recoveryActionId: executionBlocker.recoveryActionId };
const condition = { recoveryActionId: executionBlocker.recoveryActionId, ...continuationWait };
if (executionWaitRequestId) {
await tx.update(agentWakeupRequests).set({
payload: sql`jsonb_set(coalesce(${agentWakeupRequests.payload}, '{}'::jsonb), '{executionWait}', ${JSON.stringify(condition)}::jsonb)`,
updatedAt: new Date(),
}).where(eq(agentWakeupRequests.id, executionWaitRequestId));
return { kind: "deferred" as const };
}
if (durableRequest || wakeCommentId || hasInteractionContinuationWakeContext(enrichedContextSnapshot)) {
await tx.insert(agentWakeupRequests).values({
...durableReceiptFields,
@ -25754,6 +25868,7 @@ export function heartbeatService(
agentId, actorType: opts.requestedByActorType, actorId: opts.requestedByActorId,
reason, commentId: wakeCommentId ?? null, successorRunId: explicitContinuationRunId,
dryRun: true,
onBlocked: (reason, message) => { continuationWait = { reason, message }; },
}))) return deferBlockedExecution(executionBlocker);
const issueStateGuard = opts.issueStateGuard;
@ -26470,6 +26585,11 @@ export function heartbeatService(
tx,
);
if (dailyCapBlock) {
if (executionWaitRequestId && executionBlocker) {
continuationWait = { reason: dailyCapBlock.reason,
message: "The agent has reached its daily limit. Your message is saved until work can resume." };
return deferBlockedExecution(executionBlocker);
}
const now = new Date();
await tx.insert(agentWakeupRequests).values({
...durableReceiptFields,
@ -26564,7 +26684,7 @@ export function heartbeatService(
!wake.idempotencyKey?.startsWith("chat-inbound:") &&
!isInteractionResolutionWakePayload(deferredPayload) &&
!hasInteractionContinuationWakeContext(deferredContext) &&
(deferredContext.wakeReason ?? wake.reason) === "issue_commented" &&
["issue_commented", "issue_reopened_via_comment"].includes(String(deferredContext.wakeReason ?? wake.reason)) &&
queuedCommentIdsFromWakePayload(wake.payload).length > 0
);
});
@ -28090,6 +28210,7 @@ export function heartbeatService(
releaseEnvironmentLeasesForRun,
resumeRemoteStopComments,
resumeExecutionWaitComments,
sweepStaleIssueLocks,

2
server/src/services/issue-queued-comment-queue.ts
View File

@ -177,6 +177,7 @@ export function buildQueuedCommentQueueSnapshot<TComment extends QueuedCommentQu
protocol: QueuedCommentQueueProtocol;
steeringDisposition: IssueQueuedCommentQueue["steeringDisposition"];
comments: TComment[];
executionWait?: IssueQueuedCommentQueue["executionWait"];
actorType: "agent" | "user";
actorId: string;
}): IssueQueuedCommentQueue {
@ -188,6 +189,7 @@ export function buildQueuedCommentQueueSnapshot<TComment extends QueuedCommentQu
revision: queuedCommentQueueRevision({ queueId: facts.queueId, comments: facts.comments }),
protocol: facts.protocol,
steeringDisposition: facts.steeringDisposition,
...(facts.executionWait ? { executionWait: facts.executionWait } : {}),
entries: facts.comments.map((comment, position) => ({
comment: comment as unknown as IssueComment,
position,

60
server/src/services/native-local-process-stop.ts Normal file
View File

@ -0,0 +1,60 @@
import { and, desc, eq, inArray, isNull } from "drizzle-orm";
import { environmentLeases, heartbeatRunEvents, heartbeatRuns, type Db } from "@paperclipai/db";
import { appendHeartbeatRunEvent } from "./heartbeat-run-events.js";
export const PROCESS_START_REQUESTED = "native.process_start_requested";
export const PROCESS_IDENTITY_RECORDED = "native.process_identity_recorded";
const LOCAL_PROCESS_STOPPED = "native.local_process_stopped";
function absent(pid: number): boolean {
try {
process.kill(pid, 0);
return false;
} catch (error) {
return (error as NodeJS.ErrnoException).code === "ESRCH";
}
}
/**
* Preserve the host's observation before recovery clears the process fields.
* Call in the same transaction as that compare-and-set. Never inspect remote
* process IDs in the control-plane host's process table.
*/
export async function recordNativeLocalProcessStop(db: Db, run: typeof heartbeatRuns.$inferSelect) {
if (run.runtimeMode !== "native" || (!run.processPid && !run.processGroupId)) return false;
const leases = await db.select({ provider: environmentLeases.provider })
.from(environmentLeases)
.where(and(
eq(environmentLeases.companyId, run.companyId),
eq(environmentLeases.heartbeatRunId, run.id),
));
if (leases.some(lease => lease.provider !== "local")) return false;
if ((run.processPid && !absent(run.processPid)) ||
(run.processGroupId && !absent(-run.processGroupId))) return false;
await appendHeartbeatRunEvent(db, {
companyId: run.companyId,
runId: run.id,
agentId: run.agentId,
eventType: LOCAL_PROCESS_STOPPED,
stream: "system",
level: "info",
message: "Recovery verified the local process stopped before clearing its identity.",
payload: { processPid: run.processPid, processGroupId: run.processGroupId },
});
return true;
}
/** Only server-authored evidence counts. A later launch invalidates the receipt. */
export async function hasNativeLocalProcessStop(db: Db, companyId: string, runId: string) {
const [event] = await db.select({ eventType: heartbeatRunEvents.eventType })
.from(heartbeatRunEvents)
.where(and(
eq(heartbeatRunEvents.companyId, companyId),
eq(heartbeatRunEvents.runId, runId),
isNull(heartbeatRunEvents.sourceEventId),
inArray(heartbeatRunEvents.eventType, [PROCESS_START_REQUESTED, PROCESS_IDENTITY_RECORDED, LOCAL_PROCESS_STOPPED]),
))
.orderBy(desc(heartbeatRunEvents.seq))
.limit(1);
return event?.eventType === LOCAL_PROCESS_STOPPED;
}

5
server/src/services/native-runtime/native-restart-recovery.ts
View File

@ -1,3 +1,4 @@
import { recordNativeLocalProcessStop } from "../native-local-process-stop.js";
import { randomUUID } from "node:crypto";
import { and, desc, eq, inArray, isNull, lte, or, sql } from "drizzle-orm";
import type { Db } from "@paperclipai/db";
@ -909,6 +910,10 @@ export async function claimNativeRestartRecoveries(input: {
} as const;
}
if (claimKind !== "reattach_existing_runner") {
await recordNativeLocalProcessStop(tx as unknown as Db, row.run);
}
await tx
.update(heartbeatRuns)
.set({

5
server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts
View File

@ -1,3 +1,4 @@
import { hasNativeLocalProcessStop } from "../native-local-process-stop.js";
import { randomUUID } from "node:crypto";
import { spawn } from "node:child_process";
import { existsSync } from "node:fs";
@ -580,6 +581,10 @@ describeEmbeddedPostgres("native runner restart recovery with real processes", (
if (!claim || claim.kind !== "resume_dead_runner") {
throw new Error("Expected dead-runner recovery claim");
}
expect(await hasNativeLocalProcessStop(fixture.db, companyId, fixture.runId)).toBe(true);
const [stoppedRun] = await fixture.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, fixture.runId));
expect(stoppedRun.processPid).toBeNull();
expect(stoppedRun.processGroupId).toBeNull();
restored = createRunnerdCodexTransport({
...options,

23
server/src/services/native-runtime/native-session-executor.test.ts
View File

@ -4120,7 +4120,7 @@ function leaseDb(
returning: () => Promise<Array<{ runId: string }>>;
};
result.returning = () =>
Promise.resolve([{ runId: coordinator.runId }]);
Promise.resolve([{ runId: coordinator.runId, nextEventSeq: 2 }]);
return result;
},
};
@ -4163,12 +4163,20 @@ function leaseDb(
return query;
},
});
const insert = (table: unknown) => ({
values: (values: Record<string, unknown>) => {
updates.push({ table, values });
return { returning: async () => [values] };
},
});
const tx = {
insert,
execute: async () => [],
select,
update,
};
return {
insert,
select,
transaction: async (operation: (transaction: Db) => Promise<unknown>) =>
operation(tx as unknown as Db),
@ -6464,13 +6472,16 @@ describe("native process ownership", () => {
highestContiguousSourceSeq: 1,
};
});
state.createBackend.mockImplementationOnce((_input, options) => ({
kind: "test",
onSpawn: options.onSpawn,
}));
const updates: Array<{ table: unknown; values: Record<string, unknown> }> = [];
state.createBackend.mockImplementationOnce((_input, options) => {
expect(updates).toContainEqual({ table: heartbeatRunEvents, values: expect.objectContaining({
eventType: "native.process_start_requested", runId: execution.binding.runId,
}) });
return { kind: "test", onSpawn: options.onSpawn };
});
await executePaperclipNativeSession({
db: leaseDb(),
db: leaseDb(execution, {}, {}, updates),
execution,
runnerInstanceId: "runner",
onSpawn,

12
server/src/services/native-runtime/native-session-executor.ts
View File

@ -1,3 +1,4 @@
import { PROCESS_START_REQUESTED } from "../native-local-process-stop.js";
import { remoteLeaseCleanupScope } from "../remote-execution-termination.js";
import { resolveConnectorAssignments, isConnectorSkill } from "../connector-runtime.js";
import {
@ -7600,6 +7601,17 @@ async function executePaperclipNativeSessionWithinScope(
input.db,
input.execution.binding,
);
// Invalidate prior stop evidence before a backend can spawn. A crash between
// spawn and the PID callback must not make an old receipt authorize a turn.
await appendHeartbeatRunEvent(input.db, {
companyId: input.execution.binding.companyId,
runId: input.execution.binding.runId,
agentId: input.execution.binding.agentId,
eventType: PROCESS_START_REQUESTED,
stream: "system",
level: "info",
message: "Native execution requested; prior local stop evidence no longer applies.",
});
const runnerdBackend =
input.useRunnerd && input.backend === undefined
? await createRunnerdBackend({

9
ui/src/components/task-chat/TaskChatQueuedMessages.test.tsx
View File

@ -78,6 +78,15 @@ describe("TaskChatQueuedMessages", () => {
return props;
}
it("shows the saved message's wait reason and removes it when admission succeeds", () => {
const message = "Waiting for the previous environment to stop. Your message will start automatically.";
render({ queue: { ...queue, executionWait: { reason: "remote_cleanup", message } } });
expect(container.querySelector('[role="status"]')?.textContent).toContain(message);
expect(container.textContent).toContain("First queued message");
render();
expect(container.textContent).not.toContain(message);
});
it("renders each queued message once as a compact one-line row", () => {
render();
const pane = container.querySelector(

5
ui/src/components/task-chat/TaskChatQueuedMessages.tsx
View File

@ -365,6 +365,11 @@ export function TaskChatQueuedMessages({
data-testid="task-chat-queued-messages"
aria-label="Queued messages"
>
{queue.executionWait && (
<div role="status" aria-live="polite" className="px-3 py-1.5 text-xs text-muted-foreground">
{queue.executionWait.message}
</div>
)}
<DndContext
sensors={sensors}
collisionDetection={closestCenter}