diff --git a/cli/src/commands/worktree.ts b/cli/src/commands/worktree.ts index e50897a1eba..013f62c9110 100644 --- a/cli/src/commands/worktree.ts +++ b/cli/src/commands/worktree.ts @@ -18,6 +18,7 @@ import path from "node:path"; import { execFileSync } from "node:child_process"; import { randomUUID } from "node:crypto"; import { createServer } from "node:net"; +import { randomUUID } from "node:crypto"; import { Readable } from "node:stream"; import * as p from "@clack/prompts"; import pc from "picocolors"; @@ -1756,6 +1757,7 @@ export async function resetSeededWorktreeRunExecutionActivation(connectionString ...experimental, enableWorktreeRunExecution: false, worktreeRunExecutionInstanceNonce: null, + worktreeRunExecutionSeedEpoch: randomUUID(), worktreeRunExecutionActivatedAt: null, worktreeRunExecutionActivationInstanceId: null, }; diff --git a/packages/db/src/migrations/0228_run_execution_provenance.sql b/packages/db/src/migrations/0228_run_execution_provenance.sql new file mode 100644 index 00000000000..8a04679acac --- /dev/null +++ b/packages/db/src/migrations/0228_run_execution_provenance.sql @@ -0,0 +1,38 @@ +ALTER TABLE "heartbeat_runs" ADD COLUMN IF NOT EXISTS "instance_nonce" uuid; +ALTER TABLE "heartbeat_runs" ADD COLUMN IF NOT EXISTS "seed_epoch" uuid; +ALTER TABLE "agent_wakeup_requests" ADD COLUMN IF NOT EXISTS "instance_nonce" uuid; +ALTER TABLE "agent_wakeup_requests" ADD COLUMN IF NOT EXISTS "seed_epoch" uuid; + +CREATE OR REPLACE FUNCTION stamp_run_execution_provenance() +RETURNS trigger +LANGUAGE plpgsql +AS $$ +DECLARE + experimental jsonb; +BEGIN + SELECT settings.experimental INTO experimental + FROM instance_settings settings + WHERE settings.singleton_key = 'default' + LIMIT 1; + + NEW.instance_nonce := COALESCE( + NEW.instance_nonce, + NULLIF(experimental ->> 'worktreeRunExecutionInstanceNonce', '')::uuid + ); + NEW.seed_epoch := COALESCE( + NEW.seed_epoch, + NULLIF(experimental ->> 'worktreeRunExecutionSeedEpoch', '')::uuid + ); + RETURN NEW; +END; +$$; + +DROP TRIGGER IF EXISTS heartbeat_runs_stamp_execution_provenance ON heartbeat_runs; +CREATE TRIGGER heartbeat_runs_stamp_execution_provenance +BEFORE INSERT ON heartbeat_runs +FOR EACH ROW EXECUTE FUNCTION stamp_run_execution_provenance(); + +DROP TRIGGER IF EXISTS agent_wakeup_requests_stamp_execution_provenance ON agent_wakeup_requests; +CREATE TRIGGER agent_wakeup_requests_stamp_execution_provenance +BEFORE INSERT ON agent_wakeup_requests +FOR EACH ROW EXECUTE FUNCTION stamp_run_execution_provenance(); diff --git a/packages/db/src/migrations/meta/_journal.json b/packages/db/src/migrations/meta/_journal.json index 585f274aab1..23db6f3455a 100644 --- a/packages/db/src/migrations/meta/_journal.json +++ b/packages/db/src/migrations/meta/_journal.json @@ -1583,6 +1583,13 @@ "when": 1787668790923, "tag": "0227_modern_pandemic", "breakpoints": true + }, + { + "idx": 228, + "version": "7", + "when": 1787686000000, + "tag": "0228_run_execution_provenance", + "breakpoints": true } ] -} \ No newline at end of file +} diff --git a/packages/db/src/schema/agent_wakeup_requests.ts b/packages/db/src/schema/agent_wakeup_requests.ts index a0888d9700b..94630a32cbc 100644 --- a/packages/db/src/schema/agent_wakeup_requests.ts +++ b/packages/db/src/schema/agent_wakeup_requests.ts @@ -9,6 +9,8 @@ export const agentWakeupRequests = pgTable( id: uuid("id").primaryKey().defaultRandom(), companyId: uuid("company_id").notNull().references(() => companies.id), agentId: uuid("agent_id").notNull().references(() => agents.id), + instanceNonce: uuid("instance_nonce"), + seedEpoch: uuid("seed_epoch"), source: text("source").notNull(), triggerDetail: text("trigger_detail"), reason: text("reason"), diff --git a/packages/db/src/schema/heartbeat_runs.ts b/packages/db/src/schema/heartbeat_runs.ts index 1433605cfad..2629980abd5 100644 --- a/packages/db/src/schema/heartbeat_runs.ts +++ b/packages/db/src/schema/heartbeat_runs.ts @@ -22,6 +22,8 @@ export const heartbeatRuns = pgTable( id: uuid("id").primaryKey().defaultRandom(), companyId: uuid("company_id").notNull().references(() => companies.id), agentId: uuid("agent_id").notNull().references(() => agents.id), + instanceNonce: uuid("instance_nonce"), + seedEpoch: uuid("seed_epoch"), invocationSource: text("invocation_source").notNull().default("on_demand"), triggerDetail: text("trigger_detail"), status: text("status").notNull().default("queued"), diff --git a/packages/shared/src/types/instance.ts b/packages/shared/src/types/instance.ts index 3521866a9df..a118699915c 100644 --- a/packages/shared/src/types/instance.ts +++ b/packages/shared/src/types/instance.ts @@ -108,6 +108,8 @@ export interface InstanceExperimentalSettings { * clear it so the first boot stamps a fresh value. */ worktreeRunExecutionInstanceNonce: string | null; + /** Server-managed identity for the current seed/restore epoch. */ + worktreeRunExecutionSeedEpoch: string | null; /** * Server-managed cutoff recorded when worktree run execution is enabled in * this instance. Client PATCH payloads must not control this value. diff --git a/packages/shared/src/validators/instance.ts b/packages/shared/src/validators/instance.ts index 7832e5f1d82..a08a18e2c6c 100644 --- a/packages/shared/src/validators/instance.ts +++ b/packages/shared/src/validators/instance.ts @@ -76,6 +76,7 @@ export const instanceExperimentalSettingsSchema = z.object({ enableSandboxDuplexBridge: z.boolean().default(false), enableWorktreeRunExecution: z.boolean().default(false), worktreeRunExecutionInstanceNonce: z.string().uuid().nullable().default(null), + worktreeRunExecutionSeedEpoch: z.string().uuid().nullable().default(null), worktreeRunExecutionActivatedAt: z.string().datetime().nullable().default(null), worktreeRunExecutionActivationInstanceId: z.string().min(1).nullable().default(null), issueGraphLivenessAutoRecoveryLookbackHours: z @@ -92,6 +93,7 @@ export const patchInstanceExperimentalSettingsSchema = z instanceExperimentalSettingsSchema .omit({ worktreeRunExecutionInstanceNonce: true, + worktreeRunExecutionSeedEpoch: true, worktreeRunExecutionActivatedAt: true, worktreeRunExecutionActivationInstanceId: true, }) diff --git a/server/src/__tests__/heartbeat-worktree-suppression.test.ts b/server/src/__tests__/heartbeat-worktree-suppression.test.ts index e9f0e1c4b1b..dc2ed876e08 100644 --- a/server/src/__tests__/heartbeat-worktree-suppression.test.ts +++ b/server/src/__tests__/heartbeat-worktree-suppression.test.ts @@ -280,6 +280,41 @@ describeEmbeddedPostgres("heartbeat worktree suppression", () => { expect(runningCount).toBe(0); }); + it("terminalizes a foreign-epoch running row without adopting its live pid", async () => { + const { companyId, agentId, issueId } = await insertAgentAndIssue(); + await armWorktreeRunExecution(new Date(Date.now() - 1_000)); + const runId = randomUUID(); + await db.insert(heartbeatRuns).values({ + id: runId, + companyId, + agentId, + instanceNonce: randomUUID(), + seedEpoch: randomUUID(), + invocationSource: "assignment", + triggerDetail: "system", + status: "running", + processPid: process.pid, + responsibleUserId: "responsible-user", + contextSnapshot: { issueId, wakeReason: "issue_assigned" }, + startedAt: new Date(), + }); + + const heartbeat = heartbeatService(db, { + runtimeEnv: { PAPERCLIP_IN_WORKTREE: "true" }, + }); + const result = await heartbeat.reapOrphanedRuns(); + + expect(result).toEqual({ reaped: 1, runIds: [runId] }); + const runs = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.agentId, agentId)); + expect(runs).toHaveLength(1); + expect(runs[0]).toMatchObject({ + id: runId, + status: "failed", + errorCode: "foreign_execution_provenance", + processLossRetryCount: 0, + }); + }); + it("skips pre-cutoff system wakes but allows user wakes in an armed worktree", async () => { const { agentId, issueId } = await insertAgentAndIssue(); await armWorktreeRunExecution(new Date(Date.now() + 1_000)); diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 2e6fe3c6eb3..5063fd85818 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -6772,14 +6772,24 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) // Only worktree runtimes ever read the setting; a short TTL keeps the hot-path // suppression checks off the DB, and a read failure falls back to prior/default // (fail closed to suppression). - let cachedWorktreeRunExecutionOverride: { allowed: boolean; cutoff: Date | null; at: number } = { + let cachedWorktreeRunExecutionOverride: { + allowed: boolean; + cutoff: Date | null; + instanceNonce: string | null; + seedEpoch: string | null; + at: number; + } = { allowed: false, cutoff: null, + instanceNonce: null, + seedEpoch: null, at: 0, }; const WORKTREE_RUN_EXECUTION_OVERRIDE_TTL_MS = 3_000; const resolveWorktreeRunExecutionOverride = async () => { - if (!inWorktreeRuntime) return { allowed: false, cutoff: null }; + if (!inWorktreeRuntime) { + return { allowed: false, cutoff: null, instanceNonce: null, seedEpoch: null }; + } const now = Date.now(); if (now - cachedWorktreeRunExecutionOverride.at < WORKTREE_RUN_EXECUTION_OVERRIDE_TTL_MS) { return cachedWorktreeRunExecutionOverride; @@ -6792,6 +6802,8 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) cachedWorktreeRunExecutionOverride = { allowed: Boolean(activation.armed && cutoff && !Number.isNaN(cutoff.getTime())), cutoff: cutoff && !Number.isNaN(cutoff.getTime()) ? cutoff : null, + instanceNonce: activation.armed ? activation.instanceNonce : null, + seedEpoch: activation.armed ? activation.seedEpoch : null, at: now, }; } catch { @@ -6810,6 +6822,12 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) const override = await resolveWorktreeRunExecutionOverride(); return override.allowed ? override.cutoff : null; }; + const getWorktreeExecutionProvenance = async () => { + const override = await resolveWorktreeRunExecutionOverride(); + return override.allowed && override.instanceNonce && override.seedEpoch + ? { instanceNonce: override.instanceNonce, seedEpoch: override.seedEpoch } + : null; + }; const runLogStore = getRunLogStore(); const secretsSvc = secretService(db); @@ -12167,6 +12185,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) async function promoteDueScheduledRetries(now = new Date()) { const cutoff = await getWorktreeExecutionCutoff(); + const provenance = await getWorktreeExecutionProvenance(); const dueRuns = await db .select() .from(heartbeatRuns) @@ -12175,6 +12194,8 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) eq(heartbeatRuns.status, "scheduled_retry"), lte(heartbeatRuns.scheduledRetryAt, now), cutoff ? gte(heartbeatRuns.createdAt, cutoff) : undefined, + provenance ? eq(heartbeatRuns.instanceNonce, provenance.instanceNonce) : undefined, + provenance ? eq(heartbeatRuns.seedEpoch, provenance.seedEpoch) : undefined, ), ) .orderBy(asc(heartbeatRuns.scheduledRetryAt), asc(heartbeatRuns.createdAt), asc(heartbeatRuns.id)) @@ -13638,6 +13659,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) async function reapOrphanedRuns(opts?: { staleThresholdMs?: number }) { const staleThresholdMs = opts?.staleThresholdMs ?? 0; const now = new Date(); + const provenance = await getWorktreeExecutionProvenance(); // Find all runs stuck in "running" state (queued runs are legitimately waiting; resumeQueuedRuns handles them) const activeRuns = await db @@ -13678,6 +13700,25 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) for (const { run, adapterType, adapterConfig } of activeRuns) { if (runningProcesses.has(run.id) || activeRunExecutions.has(run.id)) continue; + if ( + provenance && + (run.instanceNonce !== provenance.instanceNonce || run.seedEpoch !== provenance.seedEpoch) + ) { + const message = "Run belongs to a foreign worktree instance or seed epoch"; + await setRunStatus(run.id, "failed", { + error: message, + errorCode: "foreign_execution_provenance", + finishedAt: now, + }); + await setWakeupStatus(run.wakeupRequestId, "failed", { + finishedAt: now, + error: message, + }); + runningProcesses.delete(run.id); + reaped.push(run.id); + continue; + } + // Apply staleness threshold to avoid false positives if (staleThresholdMs > 0) { const refTime = run.updatedAt ? new Date(run.updatedAt).getTime() : 0; @@ -13855,6 +13896,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) async function resumeQueuedRuns() { if ((await getSchedulingSuppression()).suppressed) return; const cutoff = await getWorktreeExecutionCutoff(); + const provenance = await getWorktreeExecutionProvenance(); const queuedRuns = await db .select({ agentId: heartbeatRuns.agentId }) @@ -13864,6 +13906,8 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) eq(heartbeatRuns.status, "queued"), eq(companies.status, "active"), cutoff ? gte(heartbeatRuns.createdAt, cutoff) : undefined, + provenance ? eq(heartbeatRuns.instanceNonce, provenance.instanceNonce) : undefined, + provenance ? eq(heartbeatRuns.seedEpoch, provenance.seedEpoch) : undefined, )); const agentIds = [...new Set(queuedRuns.map((r) => r.agentId))]; @@ -13873,7 +13917,10 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) } async function reconcileStrandedAssignedIssues() { - return recovery.reconcileStrandedAssignedIssues({ issueCreatedAtGte: await getWorktreeExecutionCutoff() }); + return recovery.reconcileStrandedAssignedIssues({ + issueCreatedAtGte: await getWorktreeExecutionCutoff(), + executionProvenance: await getWorktreeExecutionProvenance(), + }); } async function sweepStaleIssueLocks() { @@ -13996,6 +14043,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) async function startNextQueuedRunForAgent(agentId: string) { if ((await getSchedulingSuppression()).suppressed) return []; const cutoff = await getWorktreeExecutionCutoff(); + const provenance = await getWorktreeExecutionProvenance(); return withAgentStartLock(agentId, async () => { const agent = await getAgent(agentId); @@ -14019,6 +14067,8 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) eq(heartbeatRuns.agentId, agentId), eq(heartbeatRuns.status, "queued"), cutoff ? gte(heartbeatRuns.createdAt, cutoff) : undefined, + provenance ? eq(heartbeatRuns.instanceNonce, provenance.instanceNonce) : undefined, + provenance ? eq(heartbeatRuns.seedEpoch, provenance.seedEpoch) : undefined, )) .orderBy(asc(heartbeatRuns.createdAt)); if (queuedRuns.length === 0) return []; diff --git a/server/src/services/instance-settings.ts b/server/src/services/instance-settings.ts index 99be3e68f7c..4538ae6726c 100644 --- a/server/src/services/instance-settings.ts +++ b/server/src/services/instance-settings.ts @@ -41,6 +41,7 @@ interface InstanceSettingsServiceOptions { runtimeEnv?: Record; now?: () => Date; generateInstanceNonce?: () => string; + generateSeedEpoch?: () => string; } type WorktreeRunExecutionSuppressedReason = @@ -48,6 +49,7 @@ type WorktreeRunExecutionSuppressedReason = | "flag_disabled" | "missing_cutoff" | "missing_instance_id" + | "missing_seed_epoch" | "instance_id_mismatch" | "settings_read_error"; @@ -56,6 +58,8 @@ export type WorktreeRunExecutionActivationState = armed: true; cutoff: string; activationInstanceId: string; + instanceNonce: string; + seedEpoch: string; reason: null; } | { @@ -74,6 +78,7 @@ function stripServerManagedExperimentalPatchFields( ): PatchInstanceExperimentalSettings { const { worktreeRunExecutionInstanceNonce: _ignoredInstanceNonce, + worktreeRunExecutionSeedEpoch: _ignoredSeedEpoch, worktreeRunExecutionActivatedAt: _ignoredActivatedAt, worktreeRunExecutionActivationInstanceId: _ignoredActivationInstanceId, ...patchable @@ -158,6 +163,12 @@ export function resolveWorktreeRunExecutionActivation( experimental.worktreeRunExecutionActivationInstanceId, ); } + if (!experimental.worktreeRunExecutionSeedEpoch) { + return suppressWorktreeRunExecution( + "missing_seed_epoch", + experimental.worktreeRunExecutionActivationInstanceId, + ); + } if ( experimental.worktreeRunExecutionActivationInstanceId !== experimental.worktreeRunExecutionInstanceNonce @@ -171,6 +182,8 @@ export function resolveWorktreeRunExecutionActivation( armed: true, cutoff: experimental.worktreeRunExecutionActivatedAt, activationInstanceId: experimental.worktreeRunExecutionInstanceNonce, + instanceNonce: experimental.worktreeRunExecutionInstanceNonce, + seedEpoch: experimental.worktreeRunExecutionSeedEpoch, reason: null, }; } @@ -245,6 +258,7 @@ export function normalizeExperimentalSettings(raw: unknown): InstanceExperimenta enableSandboxDuplexBridge: parsed.data.enableSandboxDuplexBridge ?? false, enableWorktreeRunExecution: parsed.data.enableWorktreeRunExecution ?? false, worktreeRunExecutionInstanceNonce: parsed.data.worktreeRunExecutionInstanceNonce ?? null, + worktreeRunExecutionSeedEpoch: parsed.data.worktreeRunExecutionSeedEpoch ?? null, worktreeRunExecutionActivatedAt: parsed.data.worktreeRunExecutionActivatedAt ?? null, worktreeRunExecutionActivationInstanceId: parsed.data.worktreeRunExecutionActivationInstanceId ?? null, @@ -284,6 +298,7 @@ export function normalizeExperimentalSettings(raw: unknown): InstanceExperimenta enableSandboxDuplexBridge: false, enableWorktreeRunExecution: false, worktreeRunExecutionInstanceNonce: null, + worktreeRunExecutionSeedEpoch: null, worktreeRunExecutionActivatedAt: null, worktreeRunExecutionActivationInstanceId: null, issueGraphLivenessAutoRecoveryLookbackHours: @@ -351,7 +366,8 @@ export function instanceSettingsService(db: Db, options: InstanceSettingsService runner: InstanceSettingsWriteDb = db, ) { if (!isTruthyRuntimeEnvValue(runtimeEnv.PAPERCLIP_IN_WORKTREE)) return row; - if (normalizeExperimentalSettings(row.experimental).worktreeRunExecutionInstanceNonce) return row; + const normalized = normalizeExperimentalSettings(row.experimental); + if (normalized.worktreeRunExecutionInstanceNonce && normalized.worktreeRunExecutionSeedEpoch) return row; const now = options.now?.() ?? new Date(); const instanceNonce = (options.generateInstanceNonce ?? randomUUID)(); @@ -363,13 +379,17 @@ export function instanceSettingsService(db: Db, options: InstanceSettingsService .set({ experimental: { ...experimental, - worktreeRunExecutionInstanceNonce: instanceNonce, + worktreeRunExecutionInstanceNonce: + normalized.worktreeRunExecutionInstanceNonce ?? instanceNonce, + worktreeRunExecutionSeedEpoch: + normalized.worktreeRunExecutionSeedEpoch ?? (options.generateSeedEpoch ?? randomUUID)(), }, updatedAt: now, }) .where(and( eq(instanceSettings.id, row.id), - sql`${instanceSettings.experimental} ->> 'worktreeRunExecutionInstanceNonce' is null`, + sql`${instanceSettings.experimental} ->> 'worktreeRunExecutionInstanceNonce' is null + or ${instanceSettings.experimental} ->> 'worktreeRunExecutionSeedEpoch' is null`, )) .returning(); if (updated) return updated; diff --git a/server/src/services/recovery/service.ts b/server/src/services/recovery/service.ts index 4d3f08d9aba..8669d73bdcc 100644 --- a/server/src/services/recovery/service.ts +++ b/server/src/services/recovery/service.ts @@ -94,6 +94,7 @@ import { const EXECUTION_PATH_HEARTBEAT_RUN_STATUSES = ["queued", "running", "scheduled_retry"] as const; const UNSUCCESSFUL_HEARTBEAT_RUN_TERMINAL_STATUSES = ["interrupted", "failed", "cancelled", "timed_out"] as const; +type ExecutionProvenance = { instanceNonce: string; seedEpoch: string }; export const ACTIVE_RUN_OUTPUT_SUSPICION_THRESHOLD_MS = 60 * 60 * 1000; export const ACTIVE_RUN_OUTPUT_CRITICAL_THRESHOLD_MS = 4 * 60 * 60 * 1000; export const ACTIVE_RUN_OUTPUT_CONTINUE_REARM_MS = 30 * 60 * 1000; @@ -809,7 +810,11 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup }) return (await evaluateAgentInvokabilityFromDb(db, agent)).invokable; } - async function getLatestIssueRun(companyId: string, issueId: string): Promise { + async function getLatestIssueRun( + companyId: string, + issueId: string, + provenance?: ExecutionProvenance | null, + ): Promise { return db .select({ id: heartbeatRuns.id, @@ -828,6 +833,8 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup }) and( eq(heartbeatRuns.companyId, companyId), sql`${heartbeatRuns.contextSnapshot} ->> 'issueId' = ${issueId}`, + provenance ? eq(heartbeatRuns.instanceNonce, provenance.instanceNonce) : undefined, + provenance ? eq(heartbeatRuns.seedEpoch, provenance.seedEpoch) : undefined, ), ) .orderBy(desc(heartbeatRuns.createdAt), desc(heartbeatRuns.id)) @@ -839,6 +846,7 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup }) companyId: string, issueId: string, agentId: string, + provenance?: ExecutionProvenance | null, ): Promise { return db .select({ @@ -859,6 +867,8 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup }) eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.agentId, agentId), sql`${heartbeatRuns.contextSnapshot} ->> 'issueId' = ${issueId}`, + provenance ? eq(heartbeatRuns.instanceNonce, provenance.instanceNonce) : undefined, + provenance ? eq(heartbeatRuns.seedEpoch, provenance.seedEpoch) : undefined, ), ) .orderBy(desc(heartbeatRuns.createdAt), desc(heartbeatRuns.id)) @@ -872,6 +882,7 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup }) agentId: string, errorCodeToMatch: string | null, since: Date | null = null, + provenance?: ExecutionProvenance | null, ) { const rows = await db .select({ @@ -887,6 +898,8 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup }) eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.agentId, agentId), sql`${heartbeatRuns.contextSnapshot} ->> 'issueId' = ${issueId}`, + provenance ? eq(heartbeatRuns.instanceNonce, provenance.instanceNonce) : undefined, + provenance ? eq(heartbeatRuns.seedEpoch, provenance.seedEpoch) : undefined, ...(since ? [or(gte(heartbeatRuns.createdAt, since), gte(heartbeatRuns.finishedAt, since))] : []), ), ) @@ -918,7 +931,12 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup }) return { consecutive, latestFinishedAt }; } - async function hasActiveExecutionPath(companyId: string, issueId: string, agentId?: string | null) { + async function hasActiveExecutionPath( + companyId: string, + issueId: string, + agentId?: string | null, + provenance?: ExecutionProvenance | null, + ) { const [run, deferredWake] = await Promise.all([ db .select({ id: heartbeatRuns.id }) @@ -929,6 +947,8 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup }) inArray(heartbeatRuns.status, [...EXECUTION_PATH_HEARTBEAT_RUN_STATUSES]), sql`${heartbeatRuns.contextSnapshot} ->> 'issueId' = ${issueId}`, agentId ? eq(heartbeatRuns.agentId, agentId) : sql`true`, + provenance ? eq(heartbeatRuns.instanceNonce, provenance.instanceNonce) : undefined, + provenance ? eq(heartbeatRuns.seedEpoch, provenance.seedEpoch) : undefined, ), ) .limit(1) @@ -942,6 +962,8 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup }) eq(agentWakeupRequests.status, "deferred_issue_execution"), sql`${agentWakeupRequests.payload} ->> 'issueId' = ${issueId}`, agentId ? eq(agentWakeupRequests.agentId, agentId) : sql`true`, + provenance ? eq(agentWakeupRequests.instanceNonce, provenance.instanceNonce) : undefined, + provenance ? eq(agentWakeupRequests.seedEpoch, provenance.seedEpoch) : undefined, ), ) .limit(1) @@ -1011,7 +1033,12 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup }) .then((rows) => Boolean(rows[0])); } - async function hasQueuedIssueWake(companyId: string, issueId: string, agentId?: string | null) { + async function hasQueuedIssueWake( + companyId: string, + issueId: string, + agentId?: string | null, + provenance?: ExecutionProvenance | null, + ) { return db .select({ id: agentWakeupRequests.id }) .from(agentWakeupRequests) @@ -1021,6 +1048,8 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup }) eq(agentWakeupRequests.status, "queued"), sql`${agentWakeupRequests.payload} ->> 'issueId' = ${issueId}`, agentId ? eq(agentWakeupRequests.agentId, agentId) : sql`true`, + provenance ? eq(agentWakeupRequests.instanceNonce, provenance.instanceNonce) : undefined, + provenance ? eq(agentWakeupRequests.seedEpoch, provenance.seedEpoch) : undefined, ), ) .limit(1) @@ -1078,7 +1107,13 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup }) .then((rows) => Boolean(rows[0])); } - async function getLatestIssueRunSince(companyId: string, issueId: string, agentId: string, since: Date): Promise { + async function getLatestIssueRunSince( + companyId: string, + issueId: string, + agentId: string, + since: Date, + provenance?: ExecutionProvenance | null, + ): Promise { return db .select({ id: heartbeatRuns.id, @@ -1099,6 +1134,8 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup }) eq(heartbeatRuns.agentId, agentId), sql`${heartbeatRuns.contextSnapshot} ->> 'issueId' = ${issueId}`, or(gte(heartbeatRuns.createdAt, since), gte(heartbeatRuns.finishedAt, since)), + provenance ? eq(heartbeatRuns.instanceNonce, provenance.instanceNonce) : undefined, + provenance ? eq(heartbeatRuns.seedEpoch, provenance.seedEpoch) : undefined, ), ) .orderBy(desc(heartbeatRuns.createdAt), desc(heartbeatRuns.id)) @@ -4045,7 +4082,10 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup }) readNonEmptyString(monitor.externalRef) === latestRun.id; } - async function reconcileStrandedAssignedIssues(opts?: { issueCreatedAtGte?: Date | null }) { + async function reconcileStrandedAssignedIssues(opts?: { + issueCreatedAtGte?: Date | null; + executionProvenance?: ExecutionProvenance | null; + }) { const candidates = await db .select() .from(issues) @@ -4097,7 +4137,7 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup }) continue; } - let latestRun = await getLatestIssueRun(issue.companyId, issue.id); + let latestRun = await getLatestIssueRun(issue.companyId, issue.id, opts?.executionProvenance); const agent = await getAgent(agentId); const agentInvokable = agent && agent.companyId === issue.companyId @@ -4139,6 +4179,7 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup }) issue.companyId, issue.id, issue.status === "in_review" ? agentId : null, + opts?.executionProvenance, )) { result.skipped += 1; continue; @@ -4198,7 +4239,12 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup }) } const recoveryNow = new Date(); const participantLatestRunForRecovery = issue.status === "in_review" && participantAgentId - ? await getLatestIssueRunForAgent(issue.companyId, issue.id, participantAgentId) + ? await getLatestIssueRunForAgent( + issue.companyId, + issue.id, + participantAgentId, + opts?.executionProvenance, + ) : null; const providerQuotaMonitorRun = issue.status === "in_review" ? participantLatestRunForRecovery @@ -4278,6 +4324,7 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup }) agentId, CONTINUATION_WAITING_ON_REVIEW_ERROR_CODE, acceptedInteractionResolvedAt, + opts?.executionProvenance, ); const successfulRunSinceResolution = await hasSuccessfulIssueRunSince( issue.companyId, @@ -4293,7 +4340,7 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup }) continue; } - if (await hasQueuedIssueWake(issue.companyId, issue.id, agentId)) { + if (await hasQueuedIssueWake(issue.companyId, issue.id, agentId, opts?.executionProvenance)) { result.skipped += 1; continue; } @@ -4308,6 +4355,7 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup }) issue.id, agentId, acceptedInteractionResolvedAt, + opts?.executionProvenance, ); if ( classifyContinuationFailure(latestPostResolutionRun).kind === @@ -4494,7 +4542,12 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup }) continue; } - if (await hasQueuedIssueWake(issue.companyId, issue.id, participantAgentId)) { + if (await hasQueuedIssueWake( + issue.companyId, + issue.id, + participantAgentId, + opts?.executionProvenance, + )) { result.skipped += 1; continue; } @@ -4529,7 +4582,7 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup }) if (issue.status === "todo") { if (!latestRun) { - if (await hasQueuedIssueWake(issue.companyId, issue.id)) { + if (await hasQueuedIssueWake(issue.companyId, issue.id, null, opts?.executionProvenance)) { result.skipped += 1; continue; } @@ -4761,6 +4814,8 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup }) issue.id, agentId, classification.errorCode, + null, + opts?.executionProvenance, ); if (consecutive >= classification.maxAttempts) { const attemptCopy = consecutive <= 1 ? "" : ` (${consecutive}× attempts)`;