diff --git a/server/src/__tests__/closed-native-receipt-accounting.test.ts b/server/src/__tests__/closed-native-receipt-accounting.test.ts index c03b8aa17b5..0e2a0388150 100644 --- a/server/src/__tests__/closed-native-receipt-accounting.test.ts +++ b/server/src/__tests__/closed-native-receipt-accounting.test.ts @@ -288,6 +288,40 @@ const support = await getEmbeddedPostgresTestSupport(); expect((await f.block())?.reason).toMatch(/no reliable price|budget hard-stop/); }); + // Chaos wave s9802 t02: a follow-up run refused by a quarantined cleanup + // before any launch carried its predecessor's runner identity. Its debt + // waited ten minutes, then settled as unknown unpriced usage, and the + // blocking policy paused the company and held the board's Retry. + it.each([ + { label: "settles at once without a charge", inherited: true, ownStart: false }, + { label: "keeps the debt pending for its own recorded runner start", inherited: true, ownStart: true }, + { label: "keeps the debt pending for an identity it did not record as inherited", inherited: false, ownStart: false }, + ])("$label: a run that only carries its predecessor's runner identity", async ({ inherited, ownStart }) => { + const f = await fixture({ unpricedUsagePolicy: "block", processPid: 11356, failureCode: "native_session_cleanup_quarantined" }); + await db.delete(heartbeatRunEvents).where(eq(heartbeatRunEvents.runId, f.run.id)); + if (ownStart) await db.insert(heartbeatRunEvents).values({ companyId: f.companyId, runId: f.run.id, agentId: f.agentId, + seq: 1, eventType: "native.process_start_requested", stream: "system" }); + await db.update(heartbeatRuns).set({ errorCode: "native_session_cleanup_quarantined", + runnerProfileJson: inherited ? { inheritedProcessIdentityRunId: randomUUID() } : {}, + usageJson: { accountingReceiptSourceId: f.sourceId, accountingReceiptSequence: 0, accountingReceiptReady: false, + accountingReceiptUnavailable: true, accountingReceiptSourceFences: {}, + ledgerScope: { issueId: f.issue.id, projectId: null, billingCode: null } } }) + .where(eq(heartbeatRuns.id, f.run.id)); + const settled = await accountRunCost(db, f.run.id); + if (!inherited || ownStart) { + expect(settled).toBe(false); + expect((await f.readRun()).costAccountingPending).toBe(true); + return; + } + expect(settled).toBe(true); + await budgetService(db).evaluateAffectedScopes(f.companyId, f.agentId, null); + await flushBudgetEvaluationWork(db); + expect(await f.charges()).toEqual([]); + const [company] = await db.select().from(companies).where(eq(companies.id, f.companyId)); + expect(company).toMatchObject({ status: "active", pauseReason: null }); + expect(await f.block()).toBeNull(); + }); + it("keeps the debt pending while the provider process identity is still recorded", async () => { const f = await fixture({ processPid: 424242 }); expect(await reconcileRunCosts(db)).toMatchObject({ accounted: 0 }); diff --git a/server/src/services/heartbeat/runtime-selection.test.ts b/server/src/services/heartbeat/runtime-selection.test.ts index 51415004914..381c4c329ab 100644 --- a/server/src/services/heartbeat/runtime-selection.test.ts +++ b/server/src/services/heartbeat/runtime-selection.test.ts @@ -4,7 +4,8 @@ import { execFile as execFileCallback } from "node:child_process"; import { promisify } from "node:util"; import { tmpdir } from "node:os"; import { join } from "node:path"; -import { activityLog, agents, authUsers, companyMemberships, companySkillRepositorySnapshots, principalPermissionGrants, environments, environmentLeases, companies, completionContracts, createDb, applyPendingMigrations, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, executionWorkspaces, heartbeatRuns, issues, nativeRunFinalizations, workspaceOperations, type Db } from "@paperclipai/db"; +import { activityLog, agents, authUsers, companyMemberships, companySkillRepositorySnapshots, principalPermissionGrants, environments, environmentLeases, companies, completionContracts, costEvents, createDb, applyPendingMigrations, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, executionWorkspaces, heartbeatRuns, issues, nativeRunFinalizations, workspaceOperations, type Db } from "@paperclipai/db"; +import { accountRunCost } from "../run-cost-accounting.js"; import { eq, sql } from "drizzle-orm"; import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js"; @@ -330,6 +331,53 @@ describe.skipIf(!support.supported)("heartbeat runtime selection boundary", () = if (changed) expect(await readFile(join(canonical, "memory.txt"), "utf8")).toBe("Synthetic changed memory.\n"); }, 45_000); + it.each([true, false])("records which earlier run a follow-up's process identity came from (prior identity: %s)", async priorIdentity => { + const { input, run, agent, issue } = await fixture(); + const sessionId = randomUUID(); + const prior = await selectHeartbeatRuntime(db, input); + if (!prior.selected || !prior.nativeExecution) throw new Error("Expected prior native selection"); + const priorInput = { ...prior.nativeExecution, session: { ...prior.nativeExecution.session, normalizedSessionId: sessionId } }; + const priorProfile = (await read(run.id)).runnerProfileJson; + const processStartedAt = new Date(Date.now() - 60_000); + await db.update(heartbeatRuns).set({ status: "succeeded", nativeSessionId: sessionId, + ...(priorIdentity ? { processPid: 11_356, processGroupId: 11_356, processStartedAt } : {}), + runnerProfileJson: { ...priorProfile, nativeExecutionInput: priorInput, sessionCheckpoint: { + backendKind: "runner", driverKind: "codex_app_server", sessionId: "retained-provider-thread", + providerSessionId: "retained-provider-thread", identity: { ...priorInput.binding, sessionId }, + } }, + }).where(eq(heartbeatRuns.id, run.id)); + const [nextRun] = await db.insert(heartbeatRuns).values({ companyId: agent.companyId, agentId: agent.id, + issueId: issue.id, scopeKind: "issue", status: "running", nativeSessionId: sessionId, + createdAt: new Date(run.createdAt.getTime() + 1000), + }).returning(); + input.run = nextRun!; + const selected = await selectHeartbeatRuntime(db, input); + if (!selected.selected) throw new Error("Expected follow-up native selection"); + const saved = await read(nextRun!.id); + if (priorIdentity) { + expect(saved).toMatchObject({ processPid: 11_356, processGroupId: 11_356 }); + expect(saved.runnerProfileJson?.inheritedProcessIdentityRunId).toBe(run.id); + // Settled-process cleanup clears the predecessor before the follow-up's + // accounting runs; the follow-up was refused before it launched. + await db.update(heartbeatRuns).set({ processPid: null, processGroupId: null, processStartedAt: null }) + .where(eq(heartbeatRuns.id, run.id)); + const sourceId = randomUUID(); + await db.update(heartbeatRuns).set({ status: "failed", finishedAt: new Date(), costAccountingPending: true, + errorCode: "native_session_cleanup_quarantined", + usageJson: { accountingReceiptSourceId: sourceId, accountingReceiptSequence: 0, accountingReceiptReady: false, + accountingReceiptUnavailable: true, accountingReceiptSourceFences: {}, + ledgerScope: { issueId: issue.id, projectId: null, billingCode: null } } }) + .where(eq(heartbeatRuns.id, nextRun!.id)); + await db.update(nativeRunFinalizations).set({ phase: "terminal_failure", failureCode: "native_session_cleanup_quarantined" }) + .where(eq(nativeRunFinalizations.runId, nextRun!.id)); + expect(await accountRunCost(db, nextRun!.id)).toBe(true); + expect(await db.select().from(costEvents).where(eq(costEvents.heartbeatRunId, nextRun!.id))).toEqual([]); + } else { + expect(saved.processPid).toBeNull(); + expect(saved.runnerProfileJson?.inheritedProcessIdentityRunId).toBeUndefined(); + } + }); + it.each(["matching", "different-session", "absent"])("persists the migrated logical cwd only for the receipt's exact session: %s", async receipt => { const sameSession = receipt === "matching"; const { input, run, agent, issue } = await fixture(); diff --git a/server/src/services/heartbeat/runtime-selection.ts b/server/src/services/heartbeat/runtime-selection.ts index 9d0e54afe62..6af0b2490f6 100644 --- a/server/src/services/heartbeat/runtime-selection.ts +++ b/server/src/services/heartbeat/runtime-selection.ts @@ -1019,6 +1019,13 @@ export async function selectHeartbeatRuntime(db: Db, input: HeartbeatRuntimeSele : {}), nativeExecutionInput: lockedProfile.nativeExecutionInput ?? nativeExecution, + ...(typeof lockedProfile.inheritedProcessIdentityRunId === "string" + ? { inheritedProcessIdentityRunId: lockedProfile.inheritedProcessIdentityRunId } + : lockedRun.processPid == null && lockedRun.processGroupId == null && lockedRun.processStartedAt == null + && previousNativeRun?.nativeSessionId === nativeSessionId + && (previousNativeRun.processPid != null || previousNativeRun.processGroupId != null) + ? { inheritedProcessIdentityRunId: previousNativeRun.id } + : {}), nativeResourceDisposition: providerResourceDispositionForRun, nativeToolContractFingerprint: nativeToolContractFingerprintForTarget( diff --git a/server/src/services/run-cost-accounting.ts b/server/src/services/run-cost-accounting.ts index 81c7f524a33..3965daeb6a8 100644 --- a/server/src/services/run-cost-accounting.ts +++ b/server/src/services/run-cost-accounting.ts @@ -108,6 +108,12 @@ const closedNativeReceiptSource = sql`${heartbeatRuns.runtimeMode} = 'native' * failure, instead of holding company work. A usage recorder binds every run * with an empty fence map; only a replaced receipt source (a non-empty map) * shows that provider work began. */ +const inheritedProcessIdentityOnly = sql`${heartbeatRuns.runnerProfileJson}->>'inheritedProcessIdentityRunId' is not null + and not exists (select 1 from ${heartbeatRunEvents} own_process + where own_process.company_id = ${heartbeatRuns.companyId} and own_process.run_id = ${heartbeatRuns.id} + and own_process.source_event_id is null + and own_process.event_type in ('native.process_start_requested', 'native.process_identity_recorded'))`; + const nativeProviderWorkNeverStarted = sql`${heartbeatRuns.runtimeMode} = 'native' and ${heartbeatRuns.status} in ('failed', 'cancelled', 'timed_out', 'interrupted') and ${settledNativeCoordinator} @@ -115,7 +121,7 @@ const nativeProviderWorkNeverStarted = sql`${heartbeatRuns.runtimeMode} = 'nativ and coalesce(${heartbeatRuns.usageJson}->'accountingReceiptSourceFences', '{}'::jsonb) = '{}'::jsonb and not exists (select 1 from ${runUsageReceipts} journaled_receipt where journaled_receipt.company_id = ${heartbeatRuns.companyId} and journaled_receipt.run_id = ${heartbeatRuns.id}) - and ${heartbeatRuns.processPid} is null and ${heartbeatRuns.processGroupId} is null + and ((${heartbeatRuns.processPid} is null and ${heartbeatRuns.processGroupId} is null) or (${inheritedProcessIdentityOnly})) and coalesce((select process_event.event_type from ${heartbeatRunEvents} process_event where process_event.company_id = ${heartbeatRuns.companyId} and process_event.run_id = ${heartbeatRuns.id} and process_event.source_event_id is null