mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-07 16:35:48 +01:00
fix(recovery): release holds for unstarted dispatch cancellations (#18838)
Release false recovery holds for legacy dispatch cancellations whose retained evidence proves execution never started. Keep audit history, current admission gates, review-wait Retry, and retained workspace obligations. Share the execution-absence proof across cancellation policies and make the dispatcher error-code policy exhaustive. Add PostgreSQL coverage and stabilize the browser history fixture by waiting for pause cleanup before seeding it. Fixes #15483 Refs #18085 Co-authored-by: commitperclip[bot] <284812584+commitperclip[bot]@users.noreply.github.com> Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
8 files changed
+209
-31
No files matched your search
@@ -151,6 +151,16 @@ const support = await getEmbeddedPostgresTestSupport();
|
||||
});
|
||||
});
|
||||
|
||||
it.each(["nativeExecutionInput", "nativeWorkspaceSync", "remoteWorkspace", "nativeWorkspaceFinalizationOwner", "sessionCheckpoint"])(
|
||||
"retains an admission-wait hold with a retained %s profile", async key => {
|
||||
const f = await seed(true);
|
||||
const [run] = await db.update(heartbeatRuns).set({ runnerProfileJson: { [key]: { retained: true } } })
|
||||
.where(eq(heartbeatRuns.id, f.runId)).returning();
|
||||
expect(await legacyExecutionNeedsReconciliationWithEvidence(db, run)).toBe(true);
|
||||
expect(await getExecutionBlocker(db, f.companyId, f.issueId)).toMatchObject({ recoveryActionId: f.actionId });
|
||||
},
|
||||
);
|
||||
|
||||
it.each(["started", "runtime_selected", "process", "session", "controller", "stage", "receipt", "receipt_issue", "receipt_action", "lease", "coordinator", "launch", "provider", "workspace"])(
|
||||
"retains a hold with conflicting or incomplete %s evidence", async kind => {
|
||||
const f = await seed(true);
|
||||
|
||||
@@ -1,10 +1,13 @@
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { eq } from "drizzle-orm";
|
||||
import { afterAll, beforeAll, describe, expect, it } from "vitest";
|
||||
import { agents, agentWakeupRequests, companies, createDb, heartbeatRuns, issueComments, issueRecoveryActions, issueThreadInteractions, issues } from "@paperclipai/db";
|
||||
import { agents, agentWakeupRequests, companies, createDb, environmentLeases, heartbeatRunEvents, heartbeatRuns, issueComments, issueRecoveryActions, issueThreadInteractions, issues } from "@paperclipai/db";
|
||||
import { heartbeatService } from "../services/heartbeat.js";
|
||||
import { recoveryService } from "../services/recovery/service.js";
|
||||
import { startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
|
||||
import { createPostgresRunDispatchAdapter } from "../modules/run-dispatch/adapters/postgres.js";
|
||||
import { legacyExecutionNeedsReconciliationWithEvidence, terminalizeLegacyExecution } from "../services/legacy-execution-recovery.js";
|
||||
import { getExecutionBlocker } from "../services/execution-blocker.js";
|
||||
|
||||
describe("legacy continuation persisted authority", () => {
|
||||
let temporary: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
|
||||
@@ -33,6 +36,103 @@ describe("legacy continuation persisted authority", () => {
|
||||
}
|
||||
return { companyId, agentId, issueId, runId, createRecovery, runs, actions, finish };
|
||||
}
|
||||
it("does not freeze a never-started dispatch cancellation or retain its historical hold", async () => {
|
||||
const f = await fixture();
|
||||
await db.update(heartbeatRuns).set({ status: "queued", resultJson: null }).where(eq(heartbeatRuns.id, f.runId));
|
||||
await db.update(issues).set({ assigneeAgentId: null }).where(eq(issues.id, f.issueId));
|
||||
expect(await createPostgresRunDispatchAdapter(db).cancelStaleQueuedRun({ companyId: f.companyId,
|
||||
runId: f.runId, expectedStatus: "queued", now: new Date() })).toMatchObject({ outcome: "cancelled", errorCode: "issue_assignee_changed" });
|
||||
const run = (await f.runs())[0];
|
||||
expect(await legacyExecutionNeedsReconciliationWithEvidence(db, run)).toBe(false);
|
||||
await terminalizeLegacyExecution({ db, run, status: "cancelled", fromStatuses: ["cancelled"], reconcileIfNeeded: true });
|
||||
expect(await f.actions()).toHaveLength(0);
|
||||
await db.insert(issueRecoveryActions).values({ companyId: f.companyId, sourceIssueId: f.issueId,
|
||||
kind: "active_run_watchdog", cause: "legacy_execution_requires_reconciliation", ownerType: "board",
|
||||
fingerprint: `legacy-execution:${run.id}`, status: "resolved", outcome: "blocked", nextAction: "Reconcile",
|
||||
evidence: { runId: run.id, automaticRecovery: { replay: "blocked" } } });
|
||||
expect(await getExecutionBlocker(db, f.companyId, f.issueId)).toBeNull();
|
||||
for (const errorCode of ["issue_not_found", "issue_not_in_progress", "issue_waiting_for_response",
|
||||
"issue_dependencies_blocked", "issue_blocked",
|
||||
"issue_execution_lock_changed", "issue_terminal_status", "issue_review_participant_changed"]) {
|
||||
const changed = { ...run, errorCode };
|
||||
await db.update(heartbeatRuns).set(changed).where(eq(heartbeatRuns.id, run.id));
|
||||
expect(await legacyExecutionNeedsReconciliationWithEvidence(db, changed)).toBe(false);
|
||||
expect(await getExecutionBlocker(db, f.companyId, f.issueId)).toBeNull();
|
||||
}
|
||||
const reviewWait = { ...run, errorCode: "issue_continuation_waiting_on_review", resultJson: {
|
||||
stopReason: "issue_continuation_waiting_on_review", timeoutSource: "stale_queued_run_gate",
|
||||
} };
|
||||
await db.update(issues).set({ assigneeAgentId: f.agentId }).where(eq(issues.id, f.issueId));
|
||||
await db.update(heartbeatRuns).set(reviewWait).where(eq(heartbeatRuns.id, run.id));
|
||||
expect(await legacyExecutionNeedsReconciliationWithEvidence(db, reviewWait)).toBe(false);
|
||||
expect(await getExecutionBlocker(db, f.companyId, f.issueId)).toMatchObject({ canRetry: true });
|
||||
for (const patch of [{ finishedAt: null }, { startedAt: new Date() }, { runtimeModeResolvedAt: new Date() },
|
||||
{ processPid: 123456789 }, { processGroupId: 123456789 }, { processStartedAt: new Date() },
|
||||
{ sessionIdAfter: "provider-session" }, { controllerBootId: randomUUID() },
|
||||
{ controllerLeaseExpiresAt: new Date() }, { executionStage: "adapter_running" }, { errorCode: "cancelled" },
|
||||
{ resultJson: { workspaceRestoreFailure: "restore_unsafe_archive" } },
|
||||
{ resultJson: { workspaceRestoreRecovery: { schema: "paperclip.workspace-restore-recovery.v1" } } },
|
||||
...["nativeExecutionInput", "nativeWorkspaceSync", "remoteWorkspace", "nativeWorkspaceFinalizationOwner", "sessionCheckpoint"]
|
||||
.map(key => ({ runnerProfileJson: { [key]: { retained: true } } }))]) {
|
||||
const changed = { ...run, ...patch };
|
||||
await db.update(heartbeatRuns).set(changed).where(eq(heartbeatRuns.id, run.id));
|
||||
expect(await legacyExecutionNeedsReconciliationWithEvidence(db, changed)).toBe(true);
|
||||
expect(await getExecutionBlocker(db, f.companyId, f.issueId)).not.toBeNull();
|
||||
}
|
||||
await db.update(heartbeatRuns).set(run).where(eq(heartbeatRuns.id, run.id));
|
||||
await db.insert(heartbeatRunEvents).values({ companyId: f.companyId, runId: run.id, agentId: f.agentId,
|
||||
seq: 2, eventType: "adapter.invoke" });
|
||||
await db.update(heartbeatRuns).set({ nextEventSeq: 3 }).where(eq(heartbeatRuns.id, run.id));
|
||||
expect(await legacyExecutionNeedsReconciliationWithEvidence(db, run)).toBe(true);
|
||||
expect(await getExecutionBlocker(db, f.companyId, f.issueId)).not.toBeNull();
|
||||
});
|
||||
it("keeps historical holds for missing, foreign, or conflicting sources and workspace recovery", async () => {
|
||||
const f = await fixture(), foreign = await fixture();
|
||||
await db.update(issues).set({ status: "done" }).where(eq(issues.id, foreign.issueId));
|
||||
await db.update(heartbeatRuns).set({ status: "cancelled", finishedAt: new Date(),
|
||||
errorCode: "issue_dependencies_blocked", resultJson: null }).where(eq(heartbeatRuns.id, f.runId));
|
||||
const evidence = { runId: f.runId, automaticRecovery: { replay: "blocked" } };
|
||||
const [hold] = await db.insert(issueRecoveryActions).values({ companyId: f.companyId, sourceIssueId: f.issueId,
|
||||
kind: "active_run_watchdog", cause: "legacy_execution_requires_reconciliation", ownerType: "board",
|
||||
fingerprint: `legacy-execution:${f.runId}`, status: "resolved", outcome: "blocked", nextAction: "Reconcile", evidence }).returning();
|
||||
expect(await getExecutionBlocker(db, f.companyId, f.issueId)).toBeNull();
|
||||
for (const changed of [{ ...evidence, runId: "malformed" }, { ...evidence, runId: randomUUID() },
|
||||
{ ...evidence, runId: foreign.runId }, { automaticRecovery: evidence.automaticRecovery },
|
||||
{ ...evidence, workspaceRestoreFailure: "restore_unsafe_archive" },
|
||||
{ ...evidence, workspaceRestoreRecovery: { schema: "paperclip.workspace-restore-recovery.v1" } }]) {
|
||||
await db.update(issueRecoveryActions).set({ evidence: changed }).where(eq(issueRecoveryActions.id, hold.id));
|
||||
expect(await getExecutionBlocker(db, f.companyId, f.issueId)).not.toBeNull();
|
||||
}
|
||||
await db.update(issueRecoveryActions).set({ evidence: { sourceRunId: f.runId,
|
||||
automaticRecovery: evidence.automaticRecovery } }).where(eq(issueRecoveryActions.id, hold.id));
|
||||
expect(await getExecutionBlocker(db, f.companyId, f.issueId)).toBeNull();
|
||||
const [other] = await db.insert(issues).values({ companyId: f.companyId, title: "Another task" }).returning();
|
||||
await db.update(heartbeatRuns).set({ contextSnapshot: { issueId: other.id } }).where(eq(heartbeatRuns.id, f.runId));
|
||||
expect(await getExecutionBlocker(db, f.companyId, f.issueId)).not.toBeNull();
|
||||
await db.update(heartbeatRuns).set({ issueId: f.issueId,
|
||||
contextSnapshot: { issueId: f.issueId } }).where(eq(heartbeatRuns.id, f.runId));
|
||||
expect(await getExecutionBlocker(db, f.companyId, f.issueId)).toBeNull();
|
||||
expect(await f.actions()).toHaveLength(1);
|
||||
});
|
||||
it.each(["released environment", "provider event"])("retains a hold with a recorded %s", async retained => {
|
||||
const f = await fixture();
|
||||
const [run] = await db.update(heartbeatRuns).set({ status: "cancelled", finishedAt: new Date(),
|
||||
errorCode: "issue_dependencies_blocked", resultJson: null }).where(eq(heartbeatRuns.id, f.runId)).returning();
|
||||
await db.insert(issueRecoveryActions).values({ companyId: f.companyId, sourceIssueId: f.issueId,
|
||||
kind: "active_run_watchdog", cause: "legacy_execution_requires_reconciliation", ownerType: "board",
|
||||
fingerprint: `legacy-execution:${f.runId}`, status: "active", nextAction: "Reconcile", evidence: { runId: f.runId } });
|
||||
expect(await legacyExecutionNeedsReconciliationWithEvidence(db, run)).toBe(false);
|
||||
expect(await getExecutionBlocker(db, f.companyId, f.issueId)).toBeNull();
|
||||
if (retained === "released environment") await db.insert(environmentLeases).values({ companyId: f.companyId,
|
||||
heartbeatRunId: f.runId, issueId: f.issueId, provider: "local", status: "released", releasedAt: new Date(), cleanupStatus: "success" });
|
||||
else {
|
||||
await db.insert(heartbeatRunEvents).values({ companyId: f.companyId, runId: f.runId, agentId: f.agentId,
|
||||
seq: 1, eventType: "provider.diagnostic", sourceEventId: "provider-event-1" });
|
||||
await db.update(heartbeatRuns).set({ nextEventSeq: 2 }).where(eq(heartbeatRuns.id, f.runId));
|
||||
}
|
||||
expect(await legacyExecutionNeedsReconciliationWithEvidence(db, run)).toBe(true);
|
||||
expect(await getExecutionBlocker(db, f.companyId, f.issueId)).not.toBeNull();
|
||||
});
|
||||
it.each([1, 2, 3, 4, 5])("deduplicates concurrent replay with stale prose-derived liveness (%s)", async () => {
|
||||
const f = await fixture();
|
||||
const recovery = f.createRecovery();
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
import { isPreDispatchReviewWait } from "./pre-dispatch-review-wait.js";
|
||||
import { isPreDispatchReviewWait, neverStartedDispatchCancellationPredicate } from "./pre-dispatch-review-wait.js";
|
||||
import { canonicalExecutionUuid } from "./execution-identity.js";
|
||||
import { preDispatchExecutionWaitRecoveryActionPredicate } from "./pre-dispatch-execution-wait.js";
|
||||
import { and, desc, eq, gt, inArray, not, or, sql } from "drizzle-orm";
|
||||
import { conversationRecoveryActionPredicate, getConversationOwnershipBlocker } from "./conversation-continuation.js";
|
||||
@@ -97,6 +98,17 @@ export function executionBlockerPredicate() {
|
||||
return and(
|
||||
not(conversationRecoveryActionPredicate()!),
|
||||
not(preDispatchExecutionWaitRecoveryActionPredicate()),
|
||||
sql`not (${issueRecoveryActions.cause} = 'legacy_execution_requires_reconciliation' and exists (
|
||||
select 1 from ${heartbeatRuns}
|
||||
where ${heartbeatRuns.companyId} = ${issueRecoveryActions.companyId}
|
||||
and ${heartbeatRuns.id} = ${canonicalExecutionUuid(sql`coalesce(${issueRecoveryActions.evidence}->>'runId', ${issueRecoveryActions.evidence}->>'sourceRunId')`)}
|
||||
and ${issueRecoveryActions.sourceIssueId} = ${canonicalExecutionUuid(sql`${heartbeatRuns.contextSnapshot}->>'issueId'`)}
|
||||
and (${heartbeatRuns.issueId} is null or ${heartbeatRuns.issueId} = ${issueRecoveryActions.sourceIssueId})
|
||||
and ${neverStartedDispatchCancellationPredicate()}
|
||||
and ${heartbeatRuns.errorCode} <> 'issue_continuation_waiting_on_review'
|
||||
and coalesce(${issueRecoveryActions.evidence}->>'workspaceRestoreFailure', '') <> 'restore_unsafe_archive'
|
||||
and coalesce(${issueRecoveryActions.evidence}->'workspaceRestoreRecovery'->>'schema', '') <> ${LEGACY_WORKSPACE_RECOVERY_SCHEMA}
|
||||
))`,
|
||||
or(unreconciledNativeExportPredicate(), and(
|
||||
inArray(issueRecoveryActions.cause, [...EXECUTION_RECONCILIATION_CAUSES]),
|
||||
or(inArray(issueRecoveryActions.status, ["active", "escalated"]),
|
||||
|
||||
@@ -0,0 +1,32 @@
|
||||
import { and, eq, inArray, isNotNull, isNull, or, sql } from "drizzle-orm";
|
||||
import { environmentLeases, heartbeatRunEvents, heartbeatRuns, nativeRunFinalizations } from "@paperclipai/db";
|
||||
import { PROCESS_IDENTITY_RECORDED, PROCESS_START_REQUESTED } from "./native-local-process-stop.js";
|
||||
import { LEGACY_WORKSPACE_RECOVERY_SCHEMA } from "./workspace-restore-recovery-state.js";
|
||||
|
||||
export function executionStartedEventPredicate() {
|
||||
return or(isNotNull(heartbeatRunEvents.sourceEventId), inArray(heartbeatRunEvents.eventType,
|
||||
["adapter.invoke", PROCESS_START_REQUESTED, PROCESS_IDENTITY_RECORDED, "harness.ready", "session.started",
|
||||
"session.resumed", "session.updated", "turn.started", "provider.event", "provider.rpc_result", "tool.execution.started"]));
|
||||
}
|
||||
|
||||
/** Both dispatch cancellation policies require this same retained-evidence
|
||||
* proof. Each caller still owns its terminal status and admission receipt. */
|
||||
export function noRetainedExecutionPredicate() {
|
||||
return and(
|
||||
isNull(heartbeatRuns.startedAt), isNull(heartbeatRuns.runtimeModeResolvedAt),
|
||||
isNull(heartbeatRuns.processPid), isNull(heartbeatRuns.processGroupId), isNull(heartbeatRuns.processStartedAt),
|
||||
isNull(heartbeatRuns.nativeIssueId), isNull(heartbeatRuns.nativeSessionId), isNull(heartbeatRuns.sessionIdAfter),
|
||||
isNull(heartbeatRuns.controllerBootId), isNull(heartbeatRuns.controllerLeaseExpiresAt), isNull(heartbeatRuns.executionStage),
|
||||
sql`coalesce(${heartbeatRuns.resultJson}->>'workspaceRestoreFailure', '') <> 'restore_unsafe_archive'`,
|
||||
sql`coalesce(${heartbeatRuns.resultJson}->'workspaceRestoreRecovery'->>'schema', '') <> ${LEGACY_WORKSPACE_RECOVERY_SCHEMA}`,
|
||||
sql`not (coalesce(${heartbeatRuns.runnerProfileJson}, '{}'::jsonb) ?| array[
|
||||
'nativeExecutionInput', 'nativeWorkspaceSync', 'remoteWorkspace', 'nativeWorkspaceFinalizationOwner', 'sessionCheckpoint'])`,
|
||||
sql`not exists (select 1 from ${nativeRunFinalizations}
|
||||
where ${nativeRunFinalizations.companyId} = ${heartbeatRuns.companyId} and ${nativeRunFinalizations.runId} = ${heartbeatRuns.id})`,
|
||||
sql`not exists (select 1 from ${environmentLeases}
|
||||
where ${environmentLeases.companyId} = ${heartbeatRuns.companyId} and ${environmentLeases.heartbeatRunId} = ${heartbeatRuns.id})`,
|
||||
sql`not exists (select 1 from ${heartbeatRunEvents}
|
||||
where ${and(eq(heartbeatRunEvents.companyId, heartbeatRuns.companyId),
|
||||
eq(heartbeatRunEvents.runId, heartbeatRuns.id), executionStartedEventPredicate())})`,
|
||||
);
|
||||
}
|
||||
@@ -1,7 +1,7 @@
|
||||
import { and, eq, sql } from "drizzle-orm";
|
||||
import { environmentLeases, heartbeatRunEvents, heartbeatRuns, issueRecoveryActions, nativeRunFinalizations, type Db } from "@paperclipai/db";
|
||||
import { heartbeatRuns, issueRecoveryActions, type Db } from "@paperclipai/db";
|
||||
import { canonicalExecutionUuid } from "./execution-identity.js";
|
||||
import { PROCESS_IDENTITY_RECORDED, PROCESS_START_REQUESTED } from "./native-local-process-stop.js";
|
||||
import { noRetainedExecutionPredicate } from "./pre-dispatch-execution-evidence.js";
|
||||
import { LEGACY_WORKSPACE_RECOVERY_SCHEMA } from "./workspace-restore-recovery-state.js";
|
||||
|
||||
/** Admission denied by an existing hold is not a provider attempt. Require the
|
||||
@@ -18,27 +18,7 @@ function preDispatchExecutionWaitPredicate() {
|
||||
AND (${heartbeatRuns.issueId} IS NULL OR ${heartbeatRuns.issueId}
|
||||
= ${canonicalExecutionUuid(sql`${heartbeatRuns.contextSnapshot}->>'issueId'`)})
|
||||
AND ${canonicalExecutionUuid(sql`${heartbeatRuns.resultJson}->'executionWait'->>'recoveryActionId'`)} IS NOT NULL
|
||||
AND ${heartbeatRuns.startedAt} IS NULL AND ${heartbeatRuns.runtimeModeResolvedAt} IS NULL
|
||||
AND ${heartbeatRuns.processPid} IS NULL AND ${heartbeatRuns.processGroupId} IS NULL
|
||||
AND ${heartbeatRuns.processStartedAt} IS NULL AND ${heartbeatRuns.nativeIssueId} IS NULL
|
||||
AND ${heartbeatRuns.nativeSessionId} IS NULL AND ${heartbeatRuns.sessionIdAfter} IS NULL
|
||||
AND ${heartbeatRuns.controllerBootId} IS NULL AND ${heartbeatRuns.controllerLeaseExpiresAt} IS NULL
|
||||
AND ${heartbeatRuns.executionStage} IS NULL
|
||||
AND coalesce(${heartbeatRuns.resultJson}->>'workspaceRestoreFailure', '') <> 'restore_unsafe_archive'
|
||||
AND coalesce(${heartbeatRuns.resultJson}->'workspaceRestoreRecovery'->>'schema', '') <> ${LEGACY_WORKSPACE_RECOVERY_SCHEMA}
|
||||
AND NOT EXISTS (SELECT 1 FROM ${environmentLeases}
|
||||
WHERE ${environmentLeases.companyId} = ${heartbeatRuns.companyId}
|
||||
AND ${environmentLeases.heartbeatRunId} = ${heartbeatRuns.id})
|
||||
AND NOT EXISTS (SELECT 1 FROM ${nativeRunFinalizations}
|
||||
WHERE ${nativeRunFinalizations.companyId} = ${heartbeatRuns.companyId}
|
||||
AND ${nativeRunFinalizations.runId} = ${heartbeatRuns.id})
|
||||
AND NOT EXISTS (SELECT 1 FROM ${heartbeatRunEvents}
|
||||
WHERE ${heartbeatRunEvents.companyId} = ${heartbeatRuns.companyId}
|
||||
AND ${heartbeatRunEvents.runId} = ${heartbeatRuns.id}
|
||||
AND (${heartbeatRunEvents.sourceEventId} IS NOT NULL OR ${heartbeatRunEvents.eventType} IN
|
||||
('adapter.invoke', ${PROCESS_START_REQUESTED}, ${PROCESS_IDENTITY_RECORDED}, 'harness.ready',
|
||||
'session.started', 'session.resumed', 'session.updated', 'turn.started',
|
||||
'provider.event', 'provider.rpc_result', 'tool.execution.started'))), false)`;
|
||||
AND ${noRetainedExecutionPredicate()}, false)`;
|
||||
}
|
||||
|
||||
export async function isPreDispatchExecutionWaitVerified(db: Db, run: typeof heartbeatRuns.$inferSelect): Promise<boolean> {
|
||||
|
||||
@@ -1,13 +1,43 @@
|
||||
import { and, eq, inArray, isNotNull, or } from "drizzle-orm";
|
||||
import { and, eq, inArray, isNotNull } from "drizzle-orm";
|
||||
import { environmentLeases, heartbeatRunEvents, heartbeatRuns, nativeRunFinalizations, type Db } from "@paperclipai/db";
|
||||
import { PROCESS_IDENTITY_RECORDED, PROCESS_START_REQUESTED } from "./native-local-process-stop.js";
|
||||
import { hasRemoteTerminationReceipt } from "./remote-execution-termination.js";
|
||||
import { hasRequiredWorkspaceRecovery } from "./workspace-restore-recovery-state.js";
|
||||
import { executionStartedEventPredicate, noRetainedExecutionPredicate } from "./pre-dispatch-execution-evidence.js";
|
||||
import type { QueuedRunStalenessErrorCode } from "../modules/run-dispatch/index.js";
|
||||
|
||||
type Run = typeof heartbeatRuns.$inferSelect;
|
||||
type DispatchFields = "startedAt" | "runtimeModeResolvedAt" | "processPid" | "processGroupId" |
|
||||
"processStartedAt" | "nativeIssueId" | "nativeSessionId" | "sessionIdAfter" |
|
||||
"controllerBootId" | "controllerLeaseExpiresAt" | "executionStage" | "contextSnapshot";
|
||||
|
||||
const staleDispatchReconciliationExemptions = {
|
||||
issue_not_found: true,
|
||||
issue_not_in_progress: true,
|
||||
issue_assignee_changed: true,
|
||||
issue_waiting_for_response: true,
|
||||
issue_continuation_waiting_on_review: true,
|
||||
issue_dependencies_blocked: true,
|
||||
issue_blocked: true,
|
||||
issue_execution_lock_changed: true,
|
||||
issue_terminal_status: true,
|
||||
issue_review_participant_changed: true,
|
||||
// This code requires its own executionWait receipt in pre-dispatch-execution-wait.ts.
|
||||
execution_reconciliation_required: false,
|
||||
} satisfies Record<QueuedRunStalenessErrorCode, boolean>;
|
||||
const staleDispatchErrorCodes = Object.entries(staleDispatchReconciliationExemptions)
|
||||
.filter(([, exempt]) => exempt).map(([errorCode]) => errorCode);
|
||||
|
||||
/** A dispatch cancellation is not an interrupted provider attempt. Historical
|
||||
* rows need no new receipt, but every retained execution reference must agree. */
|
||||
export function neverStartedDispatchCancellationPredicate() {
|
||||
return and(
|
||||
eq(heartbeatRuns.runtimeMode, "legacy"), eq(heartbeatRuns.status, "cancelled"),
|
||||
inArray(heartbeatRuns.errorCode, staleDispatchErrorCodes),
|
||||
isNotNull(heartbeatRuns.finishedAt),
|
||||
noRetainedExecutionPredicate(),
|
||||
);
|
||||
}
|
||||
|
||||
/** The queued-run gate owns this receipt before execution authority is claimed.
|
||||
* A deliberate review or connection response wait has no provider actions to reconcile. The error code
|
||||
* alone is not proof; missing or conflicting dispatch evidence retains the hold. */
|
||||
@@ -39,6 +69,17 @@ export function isPreDispatchReviewWait(
|
||||
* retained execution or cleanup evidence. Historical false holds must be
|
||||
* checked against the same evidence that gates explicit user admission. */
|
||||
export async function isPreDispatchReviewWaitVerified(db: Db, run: Run): Promise<boolean> {
|
||||
if (run.runtimeMode === "legacy" && run.status === "cancelled" && run.finishedAt !== null &&
|
||||
staleDispatchErrorCodes.some(errorCode => errorCode === run.errorCode) &&
|
||||
[run.startedAt, run.runtimeModeResolvedAt, run.processPid, run.processGroupId, run.processStartedAt,
|
||||
run.nativeIssueId, run.nativeSessionId, run.sessionIdAfter, run.controllerBootId,
|
||||
run.controllerLeaseExpiresAt, run.executionStage].every(value => value === null) &&
|
||||
run.resultJson?.workspaceRestoreFailure !== "restore_unsafe_archive" && !hasRequiredWorkspaceRecovery(run.resultJson)) {
|
||||
const [unstarted] = await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and(
|
||||
eq(heartbeatRuns.companyId, run.companyId), eq(heartbeatRuns.id, run.id), neverStartedDispatchCancellationPredicate(),
|
||||
)).limit(1);
|
||||
if (unstarted) return true;
|
||||
}
|
||||
if (!isPreDispatchReviewWait(run)) return false;
|
||||
const [coordinator] = await db.select({ id: nativeRunFinalizations.runId }).from(nativeRunFinalizations).where(and(
|
||||
eq(nativeRunFinalizations.companyId, run.companyId), eq(nativeRunFinalizations.runId, run.id),
|
||||
@@ -52,10 +93,7 @@ export async function isPreDispatchReviewWaitVerified(db: Db, run: Run): Promise
|
||||
: !hasRemoteTerminationReceipt(lease))) return false;
|
||||
const [execution] = await db.select({ id: heartbeatRunEvents.id }).from(heartbeatRunEvents).where(and(
|
||||
eq(heartbeatRunEvents.companyId, run.companyId), eq(heartbeatRunEvents.runId, run.id),
|
||||
or(isNotNull(heartbeatRunEvents.sourceEventId),
|
||||
inArray(heartbeatRunEvents.eventType, ["adapter.invoke", PROCESS_START_REQUESTED, PROCESS_IDENTITY_RECORDED,
|
||||
"harness.ready", "session.started", "session.resumed", "session.updated", "turn.started",
|
||||
"provider.event", "provider.rpc_result", "tool.execution.started"])),
|
||||
executionStartedEventPredicate(),
|
||||
)).limit(1);
|
||||
return !execution;
|
||||
}
|
||||
Reference in new issue
Block a user