diff --git a/server/src/services/native-runtime/native-stop-accounting.integration.test.ts b/server/src/services/native-runtime/native-stop-accounting.integration.test.ts index 33ea4d80323..51c194e21df 100644 --- a/server/src/services/native-runtime/native-stop-accounting.integration.test.ts +++ b/server/src/services/native-runtime/native-stop-accounting.integration.test.ts @@ -29,7 +29,7 @@ import { executePaperclipNativeSession } from "./native-session-executor.js"; import { buildNativeCompletionContract } from "./completion-contracts.js"; import { prepareNativeHeartbeatRun } from "./prepare-native-run.js"; import { createRunUsageRecorder } from "../usage-receipts.js"; -import { createCostAccountingReconciler } from "../run-cost-accounting.js"; +import { ACCOUNTING_CLOSURE_TIMEOUT_MS, createCostAccountingReconciler, readCostAccountingWork } from "../run-cost-accounting.js"; import { registerCostAccountingWork } from "../budget-enforcement-worker.js"; import { createDeliveryWorkCoordinator } from "../delivery-work-coordinator.js"; @@ -118,9 +118,21 @@ describe("native Stop accounting after the last usage receipt", () => { .toMatchObject({ phase: "terminal_failure", failureCode: "native_retry_cancelled" }); await expect.poll(() => reconcile.mock.calls.length).toBeGreaterThan(beforeClosure); if (!complete) { - expect((await readRun()).costAccountingPending).toBe(true); + const waiting = await readRun(); + expect(waiting).toMatchObject({ costAccountingPending: true, usageJson: { accountingReceiptReady: false } }); + expect(waiting.finishedAt).not.toBeNull(); + const closureAt = waiting.finishedAt!.getTime() + ACCOUNTING_CLOSURE_TIMEOUT_MS + 1; + expect(await readCostAccountingWork(db)).toEqual({ pending: false, nextRunAt: closureAt }); + await expect.poll(() => coordinator.nextWakeAt()).toBe(closureAt); expect(await db.select().from(costEvents).where(eq(costEvents.heartbeatRunId, runId))).toHaveLength(0); - expect(coordinator.nextWakeAt()).toBeNull(); + expect(closureAt).toBeGreaterThan(Date.now()); + const debug = db.$client.options.debug, queries: string[] = []; + db.$client.options.debug = (_connection, query) => { queries.push(query); }; + try { + await new Promise(resolve => setTimeout(resolve, 200)); + expect(queries).toEqual([]); + expect(coordinator.nextWakeAt()).toBe(closureAt); + } finally { db.$client.options.debug = debug; } await usage.capture({ complete: true, provider: "openai", model: "test-model", billingType: "metered_api", usageBasis: "per_run", usage: { inputTokens: 120, outputTokens: 10 }, costUsd: 0.25 }); } @@ -132,7 +144,7 @@ describe("native Stop accounting after the last usage receipt", () => { expect(charges).toHaveLength(1); expect(charges[0].costCents).toBe(25); expect(provider.execute).toHaveBeenCalledOnce(); - expect(coordinator.nextWakeAt()).toBeNull(); + await expect.poll(() => coordinator.nextWakeAt(), { timeout: 1500 }).toBeNull(); } finally { worker.stop(); await coordinator.stop(); } }, 30_000); it.each([{ actorType: "user", entry: "dispatch" }, { actorType: "system", entry: "dispatch" }, { actorType: "user", entry: "heartbeat" }] as const)("recovers the combined Stop fence through existing-session cleanup without new paid work ($actorType/$entry)", async ({ actorType, entry }) => { diff --git a/server/src/services/online-workspace-source.ts b/server/src/services/online-workspace-source.ts index d7fbb38bbf5..9a8e8408786 100644 --- a/server/src/services/online-workspace-source.ts +++ b/server/src/services/online-workspace-source.ts @@ -23,6 +23,7 @@ export type OnlineWorkspaceSource = WorkspaceSourceScope & { holds: Array }>; }; const intersects = (values: string[], keys: Set) => values.some(value => keys.has(value)); +type CompanionRun = { id: string; issueId: string | null; state: string }; export async function requiresOnlineWorkspaceSource(workspace: Workspace): Promise { if (await requiresOnlineOrdinarySource(workspace)) return true; @@ -39,13 +40,7 @@ export async function requiresOnlineWorkspaceSource(workspace: Workspace): Promi return false; } -async function companion(db: ReadDb, workspace: Workspace) { - const runs = await db.select({ id: heartbeatRuns.id, issueId: heartbeatRuns.nativeIssueId, - state: sql`${heartbeatRuns.runnerProfileJson}#>>'{nativeWorkspaceSync,state}'` }).from(heartbeatRuns).where(and( - eq(heartbeatRuns.companyId, workspace.companyId), - sql`${heartbeatRuns.runnerProfileJson}#>>'{nativeWorkspaceSync,workspaceId}'=${workspace.id}`, - sql`${heartbeatRuns.runnerProfileJson}#>>'{nativeWorkspaceSync,state}' in ('prepared','finalized')`, - )).orderBy(heartbeatRuns.id); +async function companion(db: ReadDb, workspace: Workspace, runs: CompanionRun[]) { const sources: LegacyWorkspaceRecoverySource[] = []; for (const run of runs.filter(run => run.state === "prepared")) { const source = await inspectOnlineLegacyWorkspaceRecoverySource(db, { workspace, runId: run.id }); @@ -163,9 +158,22 @@ export async function inspectOnlineWorkspaceSource(db: ReadDb, workspace: Worksp const recipe = readGitCacheRestoration(workspace); if (recipe && (!recipe.restoredAt || recipe.contents && !recipe.contentsRestoredAt || recipe.restorePoint && !recipe.restorePoint.expiredAt && !recipe.restorePoint.appliedAt)) return null; + const runs = await db.select({ id: heartbeatRuns.id, issueId: heartbeatRuns.nativeIssueId, + workspaceId: sql`${heartbeatRuns.runnerProfileJson}#>>'{nativeWorkspaceSync,workspaceId}'`, + state: sql`${heartbeatRuns.runnerProfileJson}#>>'{nativeWorkspaceSync,state}'` }).from(heartbeatRuns).where(and( + eq(heartbeatRuns.companyId, workspace.companyId), + inArray(sql`${heartbeatRuns.runnerProfileJson}#>>'{nativeWorkspaceSync,workspaceId}'`, included.map(node => node.row.id)), + sql`${heartbeatRuns.runnerProfileJson}#>>'{nativeWorkspaceSync,state}' in ('prepared','finalized')`, + )).orderBy(heartbeatRuns.id); + const runsByWorkspace = new Map(); + for (const run of runs) { + const group = runsByWorkspace.get(run.workspaceId) ?? []; + group.push(run); + runsByWorkspace.set(run.workspaceId, group); + } const companions: OnlineWorkspaceSource["companions"] = [], settledPreparedWorkspaceIds: string[] = [], settledRecoveryIssueIds = new Set(); for (const node of included) { - const proof = await companion(db, node.row); + const proof = await companion(db, node.row, runsByWorkspace.get(node.row.id) ?? []); if (!proof) return null; companions.push({ workspaceId: proof.workspaceId, sources: proof.sources, holds: proof.holds }); if (proof.sources.length) settledPreparedWorkspaceIds.push(proof.workspaceId); diff --git a/server/src/services/remote-workspace-online-source.test.ts b/server/src/services/remote-workspace-online-source.test.ts index d32d690d06c..2a603dcbde1 100644 --- a/server/src/services/remote-workspace-online-source.test.ts +++ b/server/src/services/remote-workspace-online-source.test.ts @@ -21,6 +21,8 @@ import { withExecutionWorkspaceLifecycleLock } from "./execution-workspace-lifec import { IsolatedRepositorySourceBusy, withIsolatedRepositorySourceLock, withWorkspaceSourceLocks } from "./workspace-source-lock.js"; import { tryLockNativeWorkspaceFinalization } from "./native-runtime/native-workspace-finalization-ownership.js"; import { recordNativeFinalizationFailure } from "./native-runtime/native-run-finalizer.js"; +import { inspectOnlineWorkspaceSource } from "./online-workspace-source.js"; +import { inspectOnlineLegacyWorkspaceRecoveryHold, inspectOnlineLegacyWorkspaceRecoverySource } from "./online-legacy-workspace-recovery.js"; import { withWorkspaceMoveLocks } from "./isolated-repository-rehome.js"; import { ensureRuntimeServicesForRun, startRuntimeServicesForWorkspaceControl, stopRuntimeServicesForExecutionWorkspace } from "./workspace-runtime.js"; @@ -508,6 +510,52 @@ describe.skipIf(!support.supported)("online ordinary physical source fence", () status: "active", evidence: { runId } }); return { ...f, run, owner, lease, warning }; } + it("discovers all Git companions in one fresh company-scoped scan per inspection", async () => { + const recovery = await recoveryFixture(), f = await gitFixture(recovery); + const [finalizedIssue] = await db.insert(issues).values({ companyId: f.companyId, projectId: f.projectId, + parentId: f.aliases[1].issue.id, title: "Retained finalized companion" }).returning(); + await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: recovery.run.agentId, + nativeIssueId: finalizedIssue.id, status: "succeeded", runtimeMode: "legacy", runnerProfileJson: { + nativeWorkspaceSync: { workspaceId: f.aliases[1].workspace.id, state: "finalized" } } }); + const [finalizedRecovery] = await db.insert(issueRecoveryActions).values({ companyId: f.companyId, + sourceIssueId: finalizedIssue.id, kind: "stranded_assigned_issue", cause: "fixture", fingerprint: randomUUID(), + ownerType: "board", nextAction: "Retain the original recovery owner" }).returning(); + const foreignCompany = randomUUID(); + await db.insert(companies).values({ id: foreignCompany, name: "Foreign companion", issuePrefix: "FOREIGN" }); + const [foreignAgent] = await db.insert(agents).values({ companyId: foreignCompany, name: "Foreign native worker" }).returning(); + await db.insert(heartbeatRuns).values({ companyId: foreignCompany, agentId: foreignAgent.id, runtimeMode: "native", + status: "running", runnerProfileJson: { nativeWorkspaceSync: { workspaceId: f.workspaceId, state: "prepared" } } }); + const source = await inspectOnlineLegacyWorkspaceRecoverySource(db, { workspace: f.aliases[0].workspace, runId: recovery.run.id }); + const preparedHold = await inspectOnlineLegacyWorkspaceRecoveryHold(db, { workspace: f.aliases[0].workspace, recovery: recovery.warning }); + const finalizedHold = await inspectOnlineLegacyWorkspaceRecoveryHold(db, { workspace: f.aliases[1].workspace, recovery: finalizedRecovery }); + expect(source).not.toBeNull(); + expect(preparedHold).not.toBeNull(); + expect(finalizedHold).not.toBeNull(); + const selected = await stored(f.child.id), debug = db.$client.options.debug; + const scans: string[] = []; + db.$client.options.debug = (_connection, query) => { + if (query.includes('from "heartbeat_runs"') && query.includes("nativeWorkspaceSync,workspaceId")) scans.push(query); + }; + try { + const scope = await inspectOnlineWorkspaceSource(db, selected); + expect(scope).not.toBeNull(); + expect(scope!.companions).toEqual([...f.aliases.map(alias => alias.workspace.id), f.child.id].sort().map(workspaceId => ({ + workspaceId, + sources: workspaceId === f.workspaceId ? [source] : [], + holds: workspaceId === f.workspaceId ? [{ actionId: recovery.warning.id, ...preparedHold }] + : workspaceId === f.aliases[1].workspace.id ? [{ actionId: finalizedRecovery.id, ...finalizedHold }] : [], + }))); + expect(scope!.settledRecoveryIssueIds).toContain(finalizedIssue.id); + expect(scans).toHaveLength(1); + + await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: recovery.run.agentId, + nativeIssueId: f.aliases[2].issue.id, status: "running", runtimeMode: "native", + runnerProfileJson: { nativeWorkspaceSync: { workspaceId: f.aliases[2].workspace.id, state: "prepared" } } }); + scans.length = 0; + expect(await inspectOnlineWorkspaceSource(db, selected)).toBeNull(); + expect(scans).toHaveLength(1); + } finally { db.$client.options.debug = debug; } + }); it("imports a shared Git child while a terminal sibling recovery and unrelated remote execution remain intact", async () => { const recovery = await recoveryFixture(), f = await gitFixture(recovery); const unrelated = await remoteWriter(f.companyId, recovery.run.agentId);