fix: continue interrupted sandbox conversations after verified termination
Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
Harold Kim
parent
74ec4f685b
commit
973fdd2cb0
|
|
@ -1266,6 +1266,12 @@ before provisioning. A live lease protects the run during overlapping service
|
|||
deployments. An expired controller loses dispatch authority; a recovery worker
|
||||
must establish that the previous execution stopped before starting a successor.
|
||||
|
||||
Lost legacy sandbox conversations continue automatically after exact provider
|
||||
termination proof. Startup and periodic sweeps retain the original failure and
|
||||
unknown side effects, preserve accepted approvals and current task history, and
|
||||
use the existing bounded retry scheduler. Cleanup continues across restarts and
|
||||
provider outages without transferring recovery responsibility to the user.
|
||||
|
||||
## 11.7 Durable agent session goals
|
||||
|
||||
Runner Protocol v2 negotiates a required `sessionGoals` capability and typed
|
||||
|
|
|
|||
|
|
@ -846,7 +846,7 @@ Local recovery records a server-authored stop receipt before it clears a verifie
|
|||
|
||||
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.
|
||||
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. For remote process-loss and shutdown incidents, missing historical adapter evidence can be replaced by exact sandbox termination proof before starting a fresh conversational turn. A known process or webhook adapter never gains that eligibility from a later settings change. A terminal row with a live predecessor process or unreleased environment lease still blocks actual admission and Resume. Automatic remote retries wait for provider termination receipts before scheduling. Recovery folds their obsolete no-replay bookkeeping without changing task ownership or status; eligible interrupted sandbox work continues through the bounded retry scheduler. 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.
|
||||
|
||||
|
|
@ -972,3 +972,22 @@ operator attention is needed at that threshold, while automatic cleanup continue
|
|||
Provider outages never convert a live sandbox into an abandoned manual task.
|
||||
|
||||
A live cleanup attempt renews its durable claim every 30 seconds. Another sweep in the same controller cannot overlap it, even if the deadline passes. Completion writes require the current attempt identity. After controller loss, cleanup can repeat destruction of the exact quarantined provider resource; providers must make that operation idempotent. A timeout or claim expiry does not prove termination.
|
||||
|
||||
### Automatic continuation after sandbox loss
|
||||
|
||||
For legacy conversational runs lost during startup, shutdown, or loss of execution
|
||||
context, the server confirms termination of every recorded remote lease and then
|
||||
schedules one successor through normal admission. Historical leases with only a
|
||||
file-cleanup success are queued for provider cleanup; that success is not a stop
|
||||
receipt. Startup and periodic sweeps complete this sequence after another restart.
|
||||
|
||||
The new turn retains task history, accepted confirmations, and the latest user
|
||||
direction. It can reuse a compatible session and workspace through the existing
|
||||
continuation path. Unknown external effects remain unknown; no recorded tool call
|
||||
is replayed. Competing sweeps share the predecessor claim and the existing maximum
|
||||
of two automatic failure retries. Exhaustion stops inference retries, not sandbox
|
||||
cleanup, and does not restore an action-reconciliation hold. Pauses, budget stops,
|
||||
pending approvals, task completion, reassignment, and later executions still gate
|
||||
admission. Native runners keep their supported reattachment protocol.
|
||||
|
||||
Historical released reusable leases without termination receipts are not reclaimed by an old run. A concurrent resume can own the provider resource before its new lease row exists. Those runs remain gated until termination is confirmed. New cleanup attempts retain their durable pending-cleanup ownership.
|
||||
|
|
|
|||
|
|
@ -1150,6 +1150,7 @@ async function startServerWithDatabaseTeardown(
|
|||
["reconciliation_delivery", () => heartbeat ? deliverReconciledExecutions(db, heartbeat.wakeup) : undefined],
|
||||
["status_delivery", () => deliverExecutionStatuses(db)],
|
||||
["automatic_disposition", () => settleUnrecoverableExecutions(db)],
|
||||
["sandbox_continuation", () => heartbeat?.resumeInterruptedSandboxRuns()],
|
||||
] as const;
|
||||
const sweepExecutionControl = () => {
|
||||
if (heartbeatSchedulerStopped) return;
|
||||
|
|
|
|||
|
|
@ -0,0 +1,203 @@
|
|||
import { randomUUID } from "node:crypto";
|
||||
import { and, eq } from "drizzle-orm";
|
||||
import { afterAll, beforeAll, describe, expect, it } from "vitest";
|
||||
import { agents, authUsers, companies, createDb, environmentLeases, environments, heartbeatRuns,
|
||||
issueComments, issueRecoveryActions, issues, issueThreadInteractions } from "@paperclipai/db";
|
||||
import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../__tests__/helpers/embedded-postgres.js";
|
||||
import { prepareAutomaticSandboxContinuation } from "./automatic-sandbox-continuation.js";
|
||||
import { heartbeatService, type HeartbeatEnvironmentRuntime } from "./heartbeat.js";
|
||||
import { remoteTerminationReceipt } from "./remote-execution-termination.js";
|
||||
import { getExecutionBlocker } from "./execution-blocker.js";
|
||||
import { buildExecutionContinuation } from "./execution-continuation.js";
|
||||
|
||||
const support = await getEmbeddedPostgresTestSupport();
|
||||
(support.supported ? describe : describe.skip)("automatic sandbox conversation recovery", () => {
|
||||
let database: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
|
||||
let db: ReturnType<typeof createDb>;
|
||||
beforeAll(async () => {
|
||||
database = await startEmbeddedPostgresTestDatabase("automatic-sandbox-");
|
||||
db = createDb(database.connectionString);
|
||||
await db.insert(authUsers).values({ id: "recovery-owner", name: "Owner", email: "recovery@example.test", emailVerified: true, createdAt: new Date(), updatedAt: new Date() });
|
||||
}, 30000);
|
||||
afterAll(async () => { await database?.cleanup(); });
|
||||
async function seed(confirmed = true) {
|
||||
const companyId = randomUUID(), agentId = randomUUID(), issueId = randomUUID(), runId = randomUUID();
|
||||
const prefix = `R${companyId.slice(0, 7)}`;
|
||||
await db.insert(companies).values({ id: companyId, name: "Recovery", issuePrefix: prefix,
|
||||
requireBoardApprovalForNewAgents: false, defaultResponsibleUserId: "recovery-owner" });
|
||||
await db.insert(agents).values({ id: agentId, companyId, name: "Agent", role: "engineer", status: "idle",
|
||||
adapterType: "claude_local", runtimeConfig: { heartbeat: { wakeOnDemand: true, maxConcurrentRuns: 1 } } });
|
||||
await db.insert(issues).values({ id: issueId, companyId, title: "Continue approved work", status: "todo",
|
||||
assigneeAgentId: agentId, responsibleUserId: "recovery-owner", issueNumber: 1, identifier: `${prefix}-1` });
|
||||
const [comment] = await db.insert(issueComments).values({ companyId, issueId, authorType: "user",
|
||||
authorUserId: "recovery-owner", body: "Proceed with the approved work." }).returning();
|
||||
const [run] = await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId,
|
||||
status: "failed", errorCode: "process_lost", error: "Process lost", responsibleUserId: "recovery-owner",
|
||||
createdAt: new Date(Date.now() - 60_000), startedAt: new Date(Date.now() - 60_000), finishedAt: new Date(),
|
||||
contextSnapshot: { issueId, wakeReason: "issue_assigned", commentId: comment.id },
|
||||
// Deliberately no invocation event, session, or adapter evidence: pre-upgrade startup.
|
||||
}).returning();
|
||||
await db.insert(issueRecoveryActions).values({ companyId, sourceIssueId: issueId,
|
||||
kind: "active_run_watchdog", cause: "legacy_execution_requires_reconciliation", fingerprint: runId,
|
||||
status: "resolved", outcome: "blocked", nextAction: "Execution needs reconciliation",
|
||||
evidence: { runId, automaticRecovery: { replay: "blocked", actionOutcome: "unknown" } } });
|
||||
const [environment] = await db.insert(environments).values({ name: `Sandbox ${runId}`, driver: "sandbox" }).returning();
|
||||
const identity = { id: randomUUID(), companyId, heartbeatRunId: runId, provider: "daytona", providerLeaseId: `sandbox-${runId}` };
|
||||
await db.insert(environmentLeases).values({ ...identity, environmentId: environment.id, status: "released",
|
||||
leasePolicy: "ephemeral", releasedAt: new Date(), cleanupStatus: "success",
|
||||
metadata: { driver: "sandbox", ...(confirmed ? { remoteExecutionTermination: remoteTerminationReceipt(identity,
|
||||
{ providerLeaseId: identity.providerLeaseId, state: "destroyed" }) } : {}) } });
|
||||
return { companyId, agentId, issueId, run, identity, environmentId: environment.id };
|
||||
}
|
||||
async function successors(runId: string) {
|
||||
return db.select().from(heartbeatRuns).where(eq(heartbeatRuns.retryOfRunId, runId));
|
||||
}
|
||||
it("automatically resumes a historical startup failure after exact provider termination", async () => {
|
||||
const f = await seed();
|
||||
await heartbeatService(db).resumeInterruptedSandboxRuns();
|
||||
const [next] = await successors(f.run.id);
|
||||
expect(next).toMatchObject({ status: "scheduled_retry", scheduledRetryAttempt: 1 });
|
||||
expect(next.contextSnapshot).toMatchObject({ issueId: f.issueId, retryOfRunId: f.run.id, commentId: f.run.contextSnapshot!.commentId });
|
||||
expect(await getExecutionBlocker(db, f.companyId, f.issueId)).toBeNull();
|
||||
const [old] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.run.id));
|
||||
expect(old).toMatchObject({ status: "failed", errorCode: "process_lost" });
|
||||
expect(old.resultJson?.automaticSandboxRecovery).toMatchObject({ actionOutcomes: "unknown" });
|
||||
});
|
||||
it("queues missing termination proof for cleanup, then resumes after a restart", async () => {
|
||||
const f = await seed(false);
|
||||
expect(await prepareAutomaticSandboxContinuation(db, f.run)).toBeNull();
|
||||
const [pending] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.identity.id));
|
||||
expect(pending.status).toBe("pending_cleanup");
|
||||
await heartbeatService(db).resumeInterruptedSandboxRuns();
|
||||
expect(await successors(f.run.id)).toHaveLength(0);
|
||||
const restarted = heartbeatService(db, { environmentRuntime: {
|
||||
retryPendingSandboxTeardown: async ({ lease }: { lease: { providerLeaseId: string } }) => ({ providerLeaseId: lease.providerLeaseId, state: "destroyed" }),
|
||||
} as unknown as HeartbeatEnvironmentRuntime });
|
||||
await restarted.sweepPendingCleanupLeases();
|
||||
await restarted.resumeInterruptedSandboxRuns();
|
||||
expect(await successors(f.run.id)).toHaveLength(1);
|
||||
});
|
||||
it("does not reclaim a historical reusable resource that can be resuming elsewhere", async () => {
|
||||
const f = await seed(false);
|
||||
await db.update(environmentLeases).set({ leasePolicy: "reuse_by_environment" }).where(eq(environmentLeases.id, f.identity.id));
|
||||
expect(await prepareAutomaticSandboxContinuation(db, f.run)).toBeNull();
|
||||
const [lease] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.identity.id));
|
||||
expect(lease.status).toBe("released");
|
||||
expect(await successors(f.run.id)).toHaveLength(0);
|
||||
});
|
||||
it("does not terminate a resource with a later lease", async () => {
|
||||
const f = await seed(false);
|
||||
await db.insert(environmentLeases).values({ companyId: f.companyId, environmentId: f.environmentId,
|
||||
provider: f.identity.provider, providerLeaseId: f.identity.providerLeaseId, status: "active",
|
||||
acquiredAt: new Date(Date.now() + 1000), leasePolicy: "ephemeral" });
|
||||
expect(await prepareAutomaticSandboxContinuation(db, f.run)).toBeNull();
|
||||
const [lease] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.identity.id));
|
||||
expect(lease.status).toBe("released");
|
||||
});
|
||||
it("creates one successor under concurrent sweeps and repeated restarts", async () => {
|
||||
const f = await seed();
|
||||
await Promise.all([heartbeatService(db).resumeInterruptedSandboxRuns(), heartbeatService(db).resumeInterruptedSandboxRuns()]);
|
||||
await heartbeatService(db).resumeInterruptedSandboxRuns();
|
||||
expect(await successors(f.run.id)).toHaveLength(1);
|
||||
});
|
||||
it("finishes delivery after a crash between retiring the hold and scheduling", async () => {
|
||||
const f = await seed();
|
||||
expect(await prepareAutomaticSandboxContinuation(db, f.run)).not.toBeNull();
|
||||
expect(await successors(f.run.id)).toHaveLength(0);
|
||||
await heartbeatService(db).resumeInterruptedSandboxRuns();
|
||||
expect(await successors(f.run.id)).toHaveLength(1);
|
||||
});
|
||||
it("uses provider proof rather than probing a remote PID in the host namespace", async () => {
|
||||
const f = await seed();
|
||||
await db.update(heartbeatRuns).set({ processPid: process.pid, processGroupId: process.pid }).where(eq(heartbeatRuns.id, f.run.id));
|
||||
await heartbeatService(db).resumeInterruptedSandboxRuns();
|
||||
expect(await successors(f.run.id)).toHaveLength(1);
|
||||
});
|
||||
it.each(["paused", "terminated"])("keeps a %s agent from restarting", async status => {
|
||||
const f = await seed();
|
||||
await db.update(agents).set({ status }).where(eq(agents.id, f.agentId));
|
||||
await heartbeatService(db).resumeInterruptedSandboxRuns();
|
||||
expect(await successors(f.run.id)).toHaveLength(0);
|
||||
});
|
||||
it.each(["done", "cancelled"])("does not restart a %s task", async status => {
|
||||
const f = await seed();
|
||||
await db.update(issues).set({ status }).where(eq(issues.id, f.issueId));
|
||||
await heartbeatService(db).resumeInterruptedSandboxRuns();
|
||||
expect(await successors(f.run.id)).toHaveLength(0);
|
||||
});
|
||||
it("honors the budget hard stop", async () => {
|
||||
const f = await seed();
|
||||
await db.update(companies).set({ status: "paused", pauseReason: "budget" }).where(eq(companies.id, f.companyId));
|
||||
await heartbeatService(db).resumeInterruptedSandboxRuns();
|
||||
expect(await successors(f.run.id)).toHaveLength(0);
|
||||
});
|
||||
it("keeps an unanswered confirmation pending", async () => {
|
||||
const f = await seed();
|
||||
await db.insert(issueThreadInteractions).values({ companyId: f.companyId, issueId: f.issueId,
|
||||
kind: "request_confirmation", status: "pending", payload: { version: 1, prompt: "Approve deployment?" },
|
||||
createdByAgentId: f.agentId });
|
||||
await heartbeatService(db).resumeInterruptedSandboxRuns();
|
||||
expect(await successors(f.run.id)).toHaveLength(0);
|
||||
});
|
||||
it("continues an accepted confirmation without asking for approval again", async () => {
|
||||
const f = await seed();
|
||||
const [interaction] = await db.insert(issueThreadInteractions).values({ companyId: f.companyId, issueId: f.issueId,
|
||||
kind: "request_confirmation", status: "accepted", continuationPolicy: "wake_assignee_on_accept",
|
||||
createdByAgentId: f.agentId, resolvedByUserId: "recovery-owner", resolvedAt: new Date(),
|
||||
payload: { version: 1, prompt: "Approve the work?" }, result: { version: 1, outcome: "accepted" } }).returning();
|
||||
await db.update(heartbeatRuns).set({ contextSnapshot: { ...f.run.contextSnapshot,
|
||||
wakeReason: "issue_interaction_resolved", interactionId: interaction.id, interactionKind: "request_confirmation",
|
||||
interactionStatus: "accepted", continuationPolicy: "wake_assignee_on_accept" } }).where(eq(heartbeatRuns.id, f.run.id));
|
||||
await heartbeatService(db).resumeInterruptedSandboxRuns();
|
||||
const [next] = await successors(f.run.id);
|
||||
expect(next.contextSnapshot).toMatchObject({ interactionId: interaction.id, interactionStatus: "accepted" });
|
||||
const [saved] = await db.select().from(issueThreadInteractions).where(eq(issueThreadInteractions.id, interaction.id));
|
||||
expect(saved.result).toEqual({ version: 1, outcome: "accepted" });
|
||||
});
|
||||
it("does not restart work assigned to someone else", async () => {
|
||||
const f = await seed();
|
||||
await db.update(issues).set({ assigneeAgentId: null, assigneeUserId: "recovery-owner" }).where(eq(issues.id, f.issueId));
|
||||
await heartbeatService(db).resumeInterruptedSandboxRuns();
|
||||
expect(await successors(f.run.id)).toHaveLength(0);
|
||||
});
|
||||
it("does not replay a failure after a later run already completed the work", async () => {
|
||||
const f = await seed();
|
||||
await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId,
|
||||
status: "succeeded", contextSnapshot: { issueId: f.issueId }, finishedAt: new Date() });
|
||||
await heartbeatService(db).resumeInterruptedSandboxRuns();
|
||||
expect(await successors(f.run.id)).toHaveLength(0);
|
||||
});
|
||||
|
||||
it("does not reset an exhausted incident counter after restart", async () => {
|
||||
const f = await seed();
|
||||
await db.update(heartbeatRuns).set({ scheduledRetryAttempt: 2 }).where(eq(heartbeatRuns.id, f.run.id));
|
||||
await heartbeatService(db).resumeInterruptedSandboxRuns();
|
||||
await heartbeatService(db).resumeInterruptedSandboxRuns();
|
||||
expect(await successors(f.run.id)).toHaveLength(0);
|
||||
const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.run.id));
|
||||
expect(run.resultJson?.automaticSandboxRecovery).toMatchObject({ state: "attempts_exhausted" });
|
||||
expect(await getExecutionBlocker(db, f.companyId, f.issueId)).toBeNull();
|
||||
});
|
||||
it("does not turn a known process adapter into an automatic conversation retry", async () => {
|
||||
const f = await seed();
|
||||
await db.update(heartbeatRuns).set({ runnerProfileJson: { adapterDispatch: { adapterType: "process" } } }).where(eq(heartbeatRuns.id, f.run.id));
|
||||
await heartbeatService(db).resumeInterruptedSandboxRuns();
|
||||
expect(await successors(f.run.id)).toHaveLength(0);
|
||||
});
|
||||
it("preserves the latest user direction in the resumed conversation", async () => {
|
||||
const f = await seed();
|
||||
await db.insert(issueComments).values({ companyId: f.companyId, issueId: f.issueId,
|
||||
authorType: "user", authorUserId: "recovery-owner", body: "Use the blue design." });
|
||||
await heartbeatService(db).resumeInterruptedSandboxRuns();
|
||||
const [next] = await successors(f.run.id);
|
||||
const history = await buildExecutionContinuation({ db, companyId: f.companyId, issueId: f.issueId,
|
||||
agentId: f.agentId, context: next.contextSnapshot!, summary: null, exposeLowTrustRaw: false });
|
||||
expect(JSON.stringify(history)).toContain("Use the blue design.");
|
||||
expect(JSON.stringify(history)).toContain("Proceed with the approved work.");
|
||||
});
|
||||
it("rejects a foreign company source", async () => {
|
||||
const f = await seed();
|
||||
expect(await prepareAutomaticSandboxContinuation(db, { ...f.run, companyId: randomUUID() })).toBeNull();
|
||||
expect(await successors(f.run.id)).toHaveLength(0);
|
||||
});
|
||||
});
|
||||
|
|
@ -0,0 +1,103 @@
|
|||
import { and, eq, gt, ne, sql } from "drizzle-orm";
|
||||
import { agents, environmentLeases, heartbeatRuns, issueRecoveryActions, issues, type Db } from "@paperclipai/db";
|
||||
import { CONVERSATION_CONTINUATION_POLICY, isConversationAdapter, recordedRunAdapter } from "./conversation-continuation.js";
|
||||
import { hasRemoteTerminationReceipt } from "./remote-execution-termination.js";
|
||||
import { persistActivity } from "./activity-log.js";
|
||||
|
||||
export const SANDBOX_INFRASTRUCTURE_ERRORS = [
|
||||
"process_lost", "server_shutdown_interrupted", "execution_context_lost",
|
||||
];
|
||||
|
||||
export async function runHasUnconfirmedRemoteExecution(db: Db, companyId: string, runId: string) {
|
||||
const leases = await db.select().from(environmentLeases).where(and(
|
||||
eq(environmentLeases.companyId, companyId), eq(environmentLeases.heartbeatRunId, runId),
|
||||
));
|
||||
return leases.some(lease => lease.provider && lease.provider !== "local" && !hasRemoteTerminationReceipt(lease));
|
||||
}
|
||||
|
||||
/** Repair execution ownership independently of whether task admission is open.
|
||||
* A fresh conversation can inspect unknown prior effects; a stopped sandbox is
|
||||
* the prerequisite, not a user accepting responsibility for those effects. */
|
||||
export async function prepareAutomaticSandboxContinuation(db: Db, source: typeof heartbeatRuns.$inferSelect) {
|
||||
const issueId = source.nativeIssueId ?? source.contextSnapshot?.issueId;
|
||||
if (source.runtimeMode !== "legacy" || typeof issueId !== "string" ||
|
||||
!["failed", "interrupted"].includes(source.status) ||
|
||||
!SANDBOX_INFRASTRUCTURE_ERRORS.includes(source.errorCode ?? "")) return null;
|
||||
return db.transaction(async tx => {
|
||||
const [issue] = await tx.select().from(issues).where(and(
|
||||
eq(issues.companyId, source.companyId), sql`${issues.id}::text = ${issueId}`,
|
||||
)).for("update");
|
||||
const [run] = await tx.select().from(heartbeatRuns).where(and(
|
||||
eq(heartbeatRuns.id, source.id), eq(heartbeatRuns.companyId, source.companyId),
|
||||
)).for("update");
|
||||
if (!issue || !run || run.runtimeMode !== "legacy" || run.status !== source.status || run.errorCode !== source.errorCode ||
|
||||
(run.nativeIssueId ?? run.contextSnapshot?.issueId) !== issue.id) return null;
|
||||
const [agent] = await tx.select().from(agents).where(and(
|
||||
eq(agents.companyId, run.companyId), eq(agents.id, run.agentId),
|
||||
));
|
||||
if (!agent || !isConversationAdapter(agent.adapterType)) return null;
|
||||
const recorded = await recordedRunAdapter(tx as unknown as Db, run);
|
||||
if (recorded && !isConversationAdapter(recorded)) return null;
|
||||
// A later execution owns current task work. Never revive an older request.
|
||||
const [successor] = await tx.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and(
|
||||
eq(heartbeatRuns.companyId, run.companyId),
|
||||
sql`(${heartbeatRuns.retryOfRunId} = ${run.id} or
|
||||
(coalesce(${heartbeatRuns.nativeIssueId}::text, ${heartbeatRuns.contextSnapshot}->>'issueId') = ${issue.id}::text
|
||||
and ${heartbeatRuns.createdAt} > ${run.createdAt.toISOString()}
|
||||
and (${heartbeatRuns.startedAt} is not null or ${heartbeatRuns.status} in ('queued', 'running', 'scheduled_retry', 'succeeded'))))`,
|
||||
)).limit(1);
|
||||
if (successor) return null;
|
||||
const leases = await tx.select().from(environmentLeases).where(and(
|
||||
eq(environmentLeases.companyId, run.companyId), eq(environmentLeases.heartbeatRunId, run.id),
|
||||
));
|
||||
if (!leases.length || leases.some(lease => !lease.provider || lease.provider === "local" || !lease.providerLeaseId)) return null;
|
||||
if (!leases.every(hasRemoteTerminationReceipt)) {
|
||||
for (const lease of leases) {
|
||||
if (hasRemoteTerminationReceipt(lease) || lease.status === "pending_cleanup") continue;
|
||||
// Released reusable resources can already be in a concurrent resume,
|
||||
// before its new lease is persisted. Only their normal release path
|
||||
// can authorize teardown; an old run cannot reclaim them retroactively.
|
||||
if (lease.leasePolicy === "reuse_by_environment" && lease.status !== "active") continue;
|
||||
// A historic reusable sandbox may since have been assigned elsewhere.
|
||||
// Its old run grants no authority over a later lease of that resource.
|
||||
const [newerLease] = await tx.select({ id: environmentLeases.id }).from(environmentLeases).where(and(
|
||||
eq(environmentLeases.companyId, run.companyId), eq(environmentLeases.provider, lease.provider!),
|
||||
eq(environmentLeases.providerLeaseId, lease.providerLeaseId!), ne(environmentLeases.id, lease.id),
|
||||
gt(environmentLeases.acquiredAt, lease.acquiredAt),
|
||||
)).limit(1);
|
||||
if (newerLease) continue;
|
||||
await tx.update(environmentLeases).set({ status: "pending_cleanup", cleanupStatus: "failed",
|
||||
failureReason: "controller_lost_termination_unconfirmed", updatedAt: new Date(),
|
||||
}).where(and(eq(environmentLeases.id, lease.id), eq(environmentLeases.companyId, run.companyId),
|
||||
eq(environmentLeases.status, lease.status)));
|
||||
}
|
||||
return null;
|
||||
}
|
||||
// Remote process numbers belong to the remote namespace, not this host.
|
||||
const [ready] = await tx.update(heartbeatRuns).set({
|
||||
processPid: null, processGroupId: null, processStartedAt: null,
|
||||
resultJson: sql`coalesce(${heartbeatRuns.resultJson}, '{}'::jsonb) || ${JSON.stringify({
|
||||
conversationContinuation: CONVERSATION_CONTINUATION_POLICY,
|
||||
automaticSandboxRecovery: { state: "provider_terminated", actionOutcomes: "unknown" },
|
||||
})}::jsonb`,
|
||||
}).where(and(eq(heartbeatRuns.id, run.id), eq(heartbeatRuns.companyId, run.companyId))).returning();
|
||||
const retired = await tx.update(issueRecoveryActions).set({
|
||||
status: "resolved", outcome: "cancelled", resolvedAt: new Date(), updatedAt: new Date(),
|
||||
nextAction: "Sandbox termination confirmed; automatic conversation continuation can proceed.",
|
||||
resolutionNote: "Provider termination permits a new turn; prior external action outcomes remain unknown.",
|
||||
wakePolicy: null, monitorPolicy: null,
|
||||
evidence: sql`${issueRecoveryActions.evidence} || jsonb_build_object('automaticRecovery',
|
||||
coalesce(${issueRecoveryActions.evidence}->'automaticRecovery', '{}'::jsonb) || '{"replay":"conversation_continuation"}'::jsonb)`,
|
||||
}).where(and(eq(issueRecoveryActions.companyId, run.companyId), eq(issueRecoveryActions.sourceIssueId, issue.id),
|
||||
eq(issueRecoveryActions.cause, "legacy_execution_requires_reconciliation"),
|
||||
sql`${issueRecoveryActions.evidence}->>'runId' = ${run.id}`,
|
||||
sql`coalesce(${issueRecoveryActions.evidence}->'automaticRecovery'->>'replay', '') != 'conversation_continuation'`,
|
||||
)).returning({ id: issueRecoveryActions.id });
|
||||
if (retired.length) await persistActivity(tx as unknown as Db, {
|
||||
companyId: run.companyId, actorType: "system", actorId: "execution-recovery",
|
||||
action: "issue.execution_recovery_settled", entityType: "issue", entityId: issue.id,
|
||||
details: { runId: run.id, recoveryActionIds: retired.map(action => action.id), proof: "provider_termination_receipt" },
|
||||
});
|
||||
return { run: ready, agent };
|
||||
});
|
||||
}
|
||||
|
|
@ -1,3 +1,4 @@
|
|||
import { hasRemoteTerminationReceipt } from "./remote-execution-termination.js";
|
||||
import { and, desc, eq, inArray, isNotNull, or, sql } from "drizzle-orm";
|
||||
import { environmentLeases, heartbeatRunEvents, heartbeatRuns, issueRecoveryActions, type Db } from "@paperclipai/db";
|
||||
import { readProcessStartedAt } from "./hot-restart.js";
|
||||
|
|
@ -19,8 +20,20 @@ export function hasConversationContinuationPolicy(result: Record<string, unknown
|
|||
return result?.conversationContinuation === CONVERSATION_CONTINUATION_POLICY;
|
||||
}
|
||||
|
||||
/** Read only server-authored dispatch evidence, never a wake payload. */
|
||||
export async function recordedRunAdapter(db: Db, run: typeof heartbeatRuns.$inferSelect) {
|
||||
const dispatch = run.runnerProfileJson?.adapterDispatch as Record<string, unknown> | undefined;
|
||||
if (typeof dispatch?.adapterType === "string") return dispatch.adapterType;
|
||||
const [event] = await db.select({ payload: heartbeatRunEvents.payload }).from(heartbeatRunEvents).where(and(
|
||||
eq(heartbeatRunEvents.companyId, run.companyId), eq(heartbeatRunEvents.runId, run.id),
|
||||
eq(heartbeatRunEvents.eventType, "adapter.invoke"),
|
||||
)).orderBy(desc(heartbeatRunEvents.seq)).limit(1);
|
||||
return typeof event?.payload?.adapterType === "string" ? event.payload.adapterType : null;
|
||||
}
|
||||
|
||||
function conversationRunPredicate() {
|
||||
return or(
|
||||
inArray(sql`${heartbeatRuns.runnerProfileJson}->'adapterDispatch'->>'adapterType'`, [...CONVERSATION_ADAPTER_TYPES]),
|
||||
sql`${heartbeatRuns.resultJson}->>'conversationContinuation' = ${CONVERSATION_CONTINUATION_POLICY}`,
|
||||
sql`exists (
|
||||
select 1 from ${heartbeatRunEvents}
|
||||
|
|
@ -35,12 +48,8 @@ function conversationRunPredicate() {
|
|||
/** Recovery must not infer the old adapter from the agent's mutable settings. */
|
||||
export async function runUsedConversationAdapter(db: Db, run: typeof heartbeatRuns.$inferSelect): Promise<boolean> {
|
||||
if (hasConversationContinuationPolicy(run.resultJson)) return true;
|
||||
const [invocation] = await db.select({ payload: heartbeatRunEvents.payload }).from(heartbeatRunEvents)
|
||||
.where(and(eq(heartbeatRunEvents.companyId, run.companyId), eq(heartbeatRunEvents.runId, run.id),
|
||||
eq(heartbeatRunEvents.eventType, "adapter.invoke")))
|
||||
.orderBy(desc(heartbeatRunEvents.seq)).limit(1);
|
||||
const adapterType = invocation?.payload?.adapterType;
|
||||
return typeof adapterType === "string" && isConversationAdapter(adapterType);
|
||||
const adapterType = await recordedRunAdapter(db, run);
|
||||
return adapterType !== null && isConversationAdapter(adapterType);
|
||||
}
|
||||
|
||||
/** Only immutable run evidence can retire a historical conversation hold.
|
||||
|
|
@ -92,9 +101,19 @@ export async function getConversationOwnershipBlocker(db: Db, companyId: string,
|
|||
conversationRunPredicate(),
|
||||
sql`coalesce(${heartbeatRuns.nativeIssueId}::text, ${heartbeatRuns.contextSnapshot}->>'issueId') = ${issueId}`,
|
||||
inArray(heartbeatRuns.status, ["failed", "timed_out", "interrupted", "cancelled"]),
|
||||
or(isNotNull(heartbeatRuns.processPid), isNotNull(heartbeatRuns.processGroupId), activeLease),
|
||||
or(isNotNull(heartbeatRuns.processPid), isNotNull(heartbeatRuns.processGroupId), activeLease,
|
||||
sql`exists (select 1 from ${environmentLeases} where ${environmentLeases.companyId} = "heartbeat_runs"."company_id"
|
||||
and ${environmentLeases.heartbeatRunId} = "heartbeat_runs"."id" and ${environmentLeases.provider} != 'local')`),
|
||||
)).orderBy(desc(heartbeatRuns.createdAt), desc(heartbeatRuns.id));
|
||||
for (const { run, activeLease: leaseHeld } of candidates) {
|
||||
const leases = await db.select().from(environmentLeases).where(and(
|
||||
eq(environmentLeases.companyId, companyId), eq(environmentLeases.heartbeatRunId, run.id),
|
||||
));
|
||||
if (leases.some(lease => lease.provider && lease.provider !== "local")) {
|
||||
if (leases.every(hasRemoteTerminationReceipt)) continue;
|
||||
return { runId: run.id, agentId: run.agentId, cause: "execution_owner_active",
|
||||
nextAction: "Waiting for the provider to confirm sandbox termination. Cleanup will retry automatically." };
|
||||
}
|
||||
let pidAlive = run.processPid !== null && processMayBeAlive(run.processPid);
|
||||
if (pidAlive && run.processStartedAt) {
|
||||
// A recycled PID cannot keep an old task blocked. An unreadable identity
|
||||
|
|
|
|||
|
|
@ -1,4 +1,5 @@
|
|||
import { PROCESS_IDENTITY_RECORDED, recordNativeLocalProcessStop } from "./native-local-process-stop.js";
|
||||
import { prepareAutomaticSandboxContinuation, runHasUnconfirmedRemoteExecution, SANDBOX_INFRASTRUCTURE_ERRORS } from "./automatic-sandbox-continuation.js";
|
||||
import { legacyControllerBootId, legacyControllerClaim, hasLiveLegacyController, revokeExpiredLegacyController, watchLegacyControllerLease } from "./legacy-controller-lease.js";
|
||||
import { completeTerminatedRemoteNativeSessionCleanup } from "../vendor/paperclip-runner/index.js";
|
||||
import { remoteExecutionHasStopped, remoteTerminationReceipt, stoppedRemoteCleanupScopes } from "./remote-execution-termination.js";
|
||||
|
|
@ -14838,6 +14839,10 @@ export function heartbeatService(
|
|||
};
|
||||
}
|
||||
|
||||
if (run.runtimeMode === "legacy" && await runHasUnconfirmedRemoteExecution(db, run.companyId, run.id)) {
|
||||
return { outcome: "not_scheduled" as const, reason: "Waiting for confirmed sandbox termination",
|
||||
errorCode: "remote_execution_cleanup_pending" as const, issueId: readNonEmptyString(run.contextSnapshot?.issueId) };
|
||||
}
|
||||
if (legacyExecutionNeedsReconciliation(run)) {
|
||||
return {
|
||||
outcome: "not_scheduled" as const,
|
||||
|
|
@ -16926,6 +16931,7 @@ export function heartbeatService(
|
|||
.set({
|
||||
status: "running",
|
||||
...legacyControllerClaim(run.runtimeMode),
|
||||
runnerProfileJson: sql`(case when jsonb_typeof(${heartbeatRuns.runnerProfileJson}) = 'object' then ${heartbeatRuns.runnerProfileJson} else '{}'::jsonb end) || ${JSON.stringify({ adapterDispatch: { adapterType: agent.adapterType } })}::jsonb`,
|
||||
responsibleUserId,
|
||||
startedAt: lockedRun.startedAt ?? claimedAt,
|
||||
updatedAt: claimedAt,
|
||||
|
|
@ -17023,6 +17029,7 @@ export function heartbeatService(
|
|||
.set({
|
||||
status: "running",
|
||||
...legacyControllerClaim(run.runtimeMode),
|
||||
runnerProfileJson: sql`(case when jsonb_typeof(${heartbeatRuns.runnerProfileJson}) = 'object' then ${heartbeatRuns.runnerProfileJson} else '{}'::jsonb end) || ${JSON.stringify({ adapterDispatch: { adapterType: agent.adapterType } })}::jsonb`,
|
||||
responsibleUserId,
|
||||
startedAt: lockedRun.startedAt ?? claimedAt,
|
||||
contextSnapshot: withQueuedCommentIdsInRunContext(
|
||||
|
|
@ -17090,6 +17097,7 @@ export function heartbeatService(
|
|||
.set({
|
||||
status: "running",
|
||||
...legacyControllerClaim(run.runtimeMode),
|
||||
runnerProfileJson: sql`(case when jsonb_typeof(${heartbeatRuns.runnerProfileJson}) = 'object' then ${heartbeatRuns.runnerProfileJson} else '{}'::jsonb end) || ${JSON.stringify({ adapterDispatch: { adapterType: agent.adapterType } })}::jsonb`,
|
||||
responsibleUserId,
|
||||
startedAt: run.startedAt ?? claimedAt,
|
||||
updatedAt: claimedAt,
|
||||
|
|
@ -18414,8 +18422,11 @@ export function heartbeatService(
|
|||
// inherit legacy retry or termination authority. Use their PID/group only
|
||||
// for a read-only liveness check so a lost in-memory handle cannot cause
|
||||
// overlapping provider/tool execution while that child is still alive.
|
||||
const [remoteLease] = run.runtimeMode === "legacy" ? await db.select({ id: environmentLeases.id })
|
||||
.from(environmentLeases).where(and(eq(environmentLeases.companyId, run.companyId),
|
||||
eq(environmentLeases.heartbeatRunId, run.id), sql`${environmentLeases.provider} is not null and ${environmentLeases.provider} != 'local'`)).limit(1) : [];
|
||||
const checksPersistedChildLiveness =
|
||||
currentAdapterTracksLocalChild || run.runtimeMode === "native";
|
||||
!remoteLease && (currentAdapterTracksLocalChild || run.runtimeMode === "native");
|
||||
const processPidAlive =
|
||||
checksPersistedChildLiveness &&
|
||||
run.processPid &&
|
||||
|
|
@ -18602,6 +18613,37 @@ export function heartbeatService(
|
|||
return { reaped: reaped.length, runIds: reaped };
|
||||
}
|
||||
|
||||
async function resumeInterruptedSandboxRuns() {
|
||||
const candidates = await db.select().from(heartbeatRuns).where(and(
|
||||
eq(heartbeatRuns.runtimeMode, "legacy"), inArray(heartbeatRuns.status, ["failed", "interrupted"]),
|
||||
inArray(heartbeatRuns.errorCode, SANDBOX_INFRASTRUCTURE_ERRORS),
|
||||
sql`exists (select 1 from ${environmentLeases} where ${environmentLeases.companyId} = "heartbeat_runs"."company_id"
|
||||
and ${environmentLeases.heartbeatRunId} = "heartbeat_runs"."id")`,
|
||||
sql`not exists (select 1 from heartbeat_runs successor where successor.company_id = "heartbeat_runs"."company_id"
|
||||
and successor.retry_of_run_id = "heartbeat_runs"."id")`,
|
||||
sql`coalesce(${heartbeatRuns.resultJson}->'automaticSandboxRecovery'->>'state', '') != 'attempts_exhausted'`,
|
||||
)).orderBy(asc(heartbeatRuns.updatedAt)).limit(20);
|
||||
let scheduled = 0;
|
||||
for (const candidate of candidates) {
|
||||
if (activeRunExecutions.has(candidate.id) || adapterExecutionControls.has(candidate.id)) continue;
|
||||
try {
|
||||
const prepared = await prepareAutomaticSandboxContinuation(db, candidate);
|
||||
if (!prepared) continue;
|
||||
const result = await scheduleBoundedRetryForRun(prepared.run, prepared.agent);
|
||||
if (result.outcome === "scheduled") scheduled += 1;
|
||||
if (result.outcome === "retry_exhausted") await db.update(heartbeatRuns).set({
|
||||
resultJson: sql`jsonb_set(${heartbeatRuns.resultJson}, '{automaticSandboxRecovery,state}', '"attempts_exhausted"'::jsonb)`,
|
||||
}).where(and(eq(heartbeatRuns.id, candidate.id), eq(heartbeatRuns.companyId, candidate.companyId)));
|
||||
} catch (err) {
|
||||
logger.warn({ err, runId: candidate.id }, "automatic sandbox continuation will retry");
|
||||
} finally {
|
||||
// Rotate blocked work through the bounded page without resetting attempts.
|
||||
await db.update(heartbeatRuns).set({ updatedAt: new Date() }).where(eq(heartbeatRuns.id, candidate.id));
|
||||
}
|
||||
}
|
||||
return { scheduled };
|
||||
}
|
||||
|
||||
async function resumeQueuedRuns() {
|
||||
if ((await getSchedulingSuppression()).suppressed) return;
|
||||
await resumeExecutionWaitComments();
|
||||
|
|
@ -19310,7 +19352,7 @@ export function heartbeatService(
|
|||
let providerTraceFinalized = false;
|
||||
|
||||
try {
|
||||
const agent = await getAgent(run.agentId);
|
||||
let agent = await getAgent(run.agentId);
|
||||
if (!agent) {
|
||||
await setRunStatus(runId, "failed", {
|
||||
error: "Agent not found",
|
||||
|
|
@ -19326,6 +19368,10 @@ export function heartbeatService(
|
|||
return;
|
||||
}
|
||||
|
||||
const selectedAdapter = (run.runnerProfileJson?.adapterDispatch as Record<string, unknown> | undefined)?.adapterType;
|
||||
if (typeof selectedAdapter === "string" && selectedAdapter !== agent.adapterType) {
|
||||
agent = { ...agent, adapterType: selectedAdapter };
|
||||
}
|
||||
const runtime = await ensureRuntimeState(agent);
|
||||
const context = parseObject(run.contextSnapshot);
|
||||
const authorizeFailedChatRetryExecution = () =>
|
||||
|
|
@ -22667,12 +22713,13 @@ export function heartbeatService(
|
|||
nativeRuntimeResolution.resolverVersion,
|
||||
runtimeModeReason: nativeRuntimeResolution.reason,
|
||||
runtimeModeResolvedAt: run.runtimeModeResolvedAt ?? new Date(),
|
||||
// Preserve only this row's server-owned admission field at the
|
||||
// atomic write, never an input or previous runner's profile.
|
||||
runnerProfileJson: sql`case when ${heartbeatRuns.runnerProfileJson} ? ${CHAT_CONTROL_RECOVERY_ADMISSION_KEY}
|
||||
then ${JSON.stringify(providerTraceRequested ? { providerTrace: { mode: "raw", traceId: providerTraceCapture?.metadata.id ?? null, maxBytes: PROVIDER_TRACE_MAX_BYTES } } : {})}::jsonb
|
||||
|| jsonb_build_object(${CHAT_CONTROL_RECOVERY_ADMISSION_KEY}::text, ${heartbeatRuns.runnerProfileJson} -> ${CHAT_CONTROL_RECOVERY_ADMISSION_KEY})
|
||||
else ${JSON.stringify(providerTraceRequested ? { providerTrace: { mode: "raw", traceId: providerTraceCapture?.metadata.id ?? null, maxBytes: PROVIDER_TRACE_MAX_BYTES } } : null)}::jsonb end`,
|
||||
// Retain only this run's server-authored dispatch and admission evidence.
|
||||
runnerProfileJson: sql`(case when ${heartbeatRuns.runnerProfileJson} ? ${CHAT_CONTROL_RECOVERY_ADMISSION_KEY}
|
||||
then jsonb_build_object(${CHAT_CONTROL_RECOVERY_ADMISSION_KEY}::text, ${heartbeatRuns.runnerProfileJson}->${CHAT_CONTROL_RECOVERY_ADMISSION_KEY})
|
||||
else '{}'::jsonb end)
|
||||
|| (case when ${heartbeatRuns.runnerProfileJson} ? 'adapterDispatch'
|
||||
then jsonb_build_object('adapterDispatch', ${heartbeatRuns.runnerProfileJson}->'adapterDispatch') else '{}'::jsonb end)
|
||||
|| ${JSON.stringify(providerTraceRequested ? { providerTrace: { mode: "raw", traceId: providerTraceCapture?.metadata.id ?? null, maxBytes: PROVIDER_TRACE_MAX_BYTES } } : {})}::jsonb`,
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(eq(heartbeatRuns.id, run.id));
|
||||
|
|
@ -28136,6 +28183,7 @@ export function heartbeatService(
|
|||
reconcileHotRestartAdoption,
|
||||
recoverNativeRunsAfterRestart,
|
||||
reapOrphanedRuns,
|
||||
resumeInterruptedSandboxRuns,
|
||||
sweepPendingCleanupLeases,
|
||||
// Override-aware scheduling-suppression check (honors the worktree
|
||||
// run-execution experimental setting). Callers outside the service that
|
||||
|
|
|
|||
Loading…
Reference in New Issue