diff --git a/apps/api/src/modules/runtime/application/session-runs/session-run-view-events.service.ts b/apps/api/src/modules/runtime/application/session-runs/session-run-view-events.service.ts index 9095c67b..8c7f5ca1 100644 --- a/apps/api/src/modules/runtime/application/session-runs/session-run-view-events.service.ts +++ b/apps/api/src/modules/runtime/application/session-runs/session-run-view-events.service.ts @@ -51,6 +51,7 @@ export function createSessionRunUpdatedEvent( run: SessionRunSummary, sessionId: SessionId, lifecycle = toSessionLifecycleStatusForRunView(run.status), + sourceEventId?: string, ): RuntimeEventEnvelope { return createSessionRuntimeEvent({ kind: toRuntimeEventKindForRunStatus(run.status), @@ -60,6 +61,7 @@ export function createSessionRunUpdatedEvent( }, runId: run.id, sessionId, + ...(sourceEventId === undefined ? {} : { sourceEventId }), traceId: run.traceId, }); } diff --git a/apps/api/src/modules/runtime/application/session-runs/terminal-run-reconciliation.service.ts b/apps/api/src/modules/runtime/application/session-runs/terminal-run-reconciliation.service.ts new file mode 100644 index 00000000..264d4311 --- /dev/null +++ b/apps/api/src/modules/runtime/application/session-runs/terminal-run-reconciliation.service.ts @@ -0,0 +1,255 @@ +import type { SessionStatus } from "@mosoo/contracts/session"; +import type { SessionRunStatus, SessionRunSummary } from "@mosoo/contracts/session-run"; +import { + driverInstancesTable, + sessionEventsTable, + sessionRunsTable, + sessionsTable, +} from "@mosoo/db"; +import type { SessionId, SessionRunId } from "@mosoo/id"; +import { and, asc, eq, inArray, isNull, notExists, or, sql } from "drizzle-orm"; + +import type { ApiBindings } from "../../../../platform/cloudflare/worker-types"; +import { getAppDatabase } from "../../../../platform/db/drizzle"; +import { appendSessionRuntimeEvents } from "../../../sessions/application/session-event-write.service"; +import { createSessionRunTerminalSourceId } from "../../domain/session-run-terminal-event-id"; +import { + getSessionRunSummariesByIds, + setSessionRunStatus, +} from "../../infrastructure/session-runs/session-run-store.repository"; +import type { SessionRunTransitionOutcome } from "../../infrastructure/session-runs/session-run-store.repository"; +import { + createFailedSessionRunRuntimeEvent, + createSessionRunUpdatedEvent, +} from "./session-run-view-events.service"; + +const TERMINAL_RUN_STATUSES = ["cancelled", "completed", "expired", "failed"] as const; +const TERMINAL_DRIVER_STATUSES = ["failed", "stopped"] as const; + +interface TerminalRunCandidate { + readonly runId: SessionRunId; + readonly sessionId: SessionId; + readonly sessionLastRunId: SessionRunId | null; + readonly sessionStatus: SessionStatus; +} + +export interface TerminalRunReconciliationResult { + readonly reconciledRunIds: readonly SessionRunId[]; + readonly reconciledSessionIds: readonly SessionId[]; +} + +function assertTerminalRunProjection(outcome: SessionRunTransitionOutcome): void { + switch (outcome.kind) { + case "applied": + case "duplicate": { + return; + } + case "repair_needed": { + throw new Error("Terminal run reconciliation left the session lifecycle projection stale."); + } + case "rejected": + case "stale": { + throw new Error("Terminal run reconciliation lost a concurrent run transition."); + } + } +} + +function terminalEventKind( + status: SessionRunStatus, +): "run.cancelled" | "run.completed" | "run.failed" { + switch (status) { + case "completed": { + return "run.completed"; + } + case "failed": { + return "run.failed"; + } + case "cancelled": + case "expired": { + return "run.cancelled"; + } + case "queued": + case "booting": + case "running": + case "waiting_input": { + throw new Error(`Expected terminal Session Run status, received ${status}.`); + } + } +} + +function createTerminalRunRecoveryEvent(input: { + readonly kind: "run.cancelled" | "run.completed" | "run.failed"; + readonly run: SessionRunSummary; + readonly sessionId: SessionId; + readonly sourceEventId: string; +}) { + if (input.kind !== "run.failed") { + return createSessionRunUpdatedEvent(input.run, input.sessionId, "IDLE", input.sourceEventId); + } + + return createFailedSessionRunRuntimeEvent({ + run: input.run, + runError: input.run.error ?? { + code: "runtime.terminal_error_missing", + details: {}, + message: "The run failed without a persisted error.", + retryable: false, + }, + sessionId: input.sessionId, + sourceEventId: input.sourceEventId, + }); +} + +async function findTerminalRunCandidates( + bindings: ApiBindings, + limit: number, +): Promise { + const database = getAppDatabase(bindings.DB); + const expectedEventType = sql` + CASE ${sessionRunsTable.status} + WHEN 'completed' THEN 'run.completed' + WHEN 'failed' THEN 'run.failed' + ELSE 'run.cancelled' + END + `; + const missingTerminalEvent = notExists( + database + .select({ id: sessionEventsTable.id }) + .from(sessionEventsTable) + .where( + and( + eq(sessionEventsTable.sessionId, sessionRunsTable.sessionId), + eq(sessionEventsTable.runId, sessionRunsTable.id), + sql`${sessionEventsTable.eventType} = ${expectedEventType}`, + ), + ), + ); + const staleSessionProjection = and( + eq(sessionsTable.lastRunId, sessionRunsTable.id), + eq(sessionsTable.status, "RUNNING"), + ); + + return database + .select({ + runId: sessionRunsTable.id, + sessionId: sessionRunsTable.sessionId, + sessionLastRunId: sessionsTable.lastRunId, + sessionStatus: sessionsTable.status, + }) + .from(sessionRunsTable) + .innerJoin(sessionsTable, eq(sessionsTable.id, sessionRunsTable.sessionId)) + .leftJoin(driverInstancesTable, eq(driverInstancesTable.id, sessionRunsTable.driverInstanceId)) + .where( + and( + inArray(sessionRunsTable.status, TERMINAL_RUN_STATUSES), + isNull(sessionsTable.archivedAt), + inArray(sessionsTable.status, ["IDLE", "RESCHEDULING", "RUNNING"]), + or( + isNull(sessionRunsTable.driverInstanceId), + isNull(driverInstancesTable.id), + inArray(driverInstancesTable.status, TERMINAL_DRIVER_STATUSES), + ), + or(staleSessionProjection, missingTerminalEvent), + ), + ) + .orderBy(asc(sessionRunsTable.updatedAt), asc(sessionRunsTable.id)) + .limit(limit) + .all(); +} + +async function readPersistedTerminalEventKeys( + bindings: ApiBindings, + runIds: readonly SessionRunId[], +): Promise> { + if (runIds.length === 0) { + return new Set(); + } + + const rows = await getAppDatabase(bindings.DB) + .select({ + eventType: sessionEventsTable.eventType, + runId: sessionEventsTable.runId, + }) + .from(sessionEventsTable) + .where( + and( + inArray(sessionEventsTable.runId, [...runIds]), + inArray(sessionEventsTable.eventType, ["run.cancelled", "run.completed", "run.failed"]), + ), + ) + .all(); + + return new Set( + rows.flatMap((row) => (row.runId === null ? [] : [`${row.runId}:${row.eventType}`])), + ); +} + +/** + * Repairs terminal Run projections after the Driver can no longer replay its + * final event. The terminal Run row itself is the durable, idempotent repair + * obligation: a missing matching terminal session_event is reconstructed with + * a stable source id, while a duplicate status transition repairs the owning + * Session lifecycle projection. + */ +export async function reconcileTerminalSessionRuns( + bindings: ApiBindings, + input: { + readonly limit: number; + }, +): Promise { + const candidates = await findTerminalRunCandidates(bindings, input.limit); + const runIds = candidates.map((candidate) => candidate.runId); + const [runsById, persistedTerminalEventKeys] = await Promise.all([ + getSessionRunSummariesByIds(bindings.DB, runIds), + readPersistedTerminalEventKeys(bindings, runIds), + ]); + const reconciledRunIds: SessionRunId[] = []; + const reconciledSessionIds = new Set(); + + for (const candidate of candidates) { + const run = runsById.get(candidate.runId); + + if (run === undefined) { + continue; + } + + if (candidate.sessionLastRunId === run.id && candidate.sessionStatus === "RUNNING") { + const projection = await setSessionRunStatus(bindings.DB, { + runId: run.id, + source: "maintenance", + status: run.status, + }); + assertTerminalRunProjection(projection); + } + + const kind = terminalEventKind(run.status); + const eventKey = `${run.id}:${kind}`; + + if (!persistedTerminalEventKeys.has(eventKey)) { + const sourceEventId = createSessionRunTerminalSourceId(run.id, kind); + const persisted = await appendSessionRuntimeEvents({ + bindings, + events: [ + createTerminalRunRecoveryEvent({ + kind, + run, + sessionId: candidate.sessionId, + sourceEventId, + }), + ], + sessionId: candidate.sessionId, + }); + + if (persisted.persistedCount > 0) { + reconciledRunIds.push(run.id); + } + } + + reconciledSessionIds.add(candidate.sessionId); + } + + return { + reconciledRunIds, + reconciledSessionIds: [...reconciledSessionIds], + }; +} diff --git a/apps/api/src/modules/runtime/domain/session-run-terminal-event-id.ts b/apps/api/src/modules/runtime/domain/session-run-terminal-event-id.ts index 330098f0..2faef709 100644 --- a/apps/api/src/modules/runtime/domain/session-run-terminal-event-id.ts +++ b/apps/api/src/modules/runtime/domain/session-run-terminal-event-id.ts @@ -1,5 +1,18 @@ import type { SessionRunId } from "@mosoo/id"; +import type { RuntimeEventKind } from "@mosoo/runtime-events"; + +type TerminalSessionRunEventKind = Extract< + RuntimeEventKind, + "run.cancelled" | "run.completed" | "run.failed" +>; + +export function createSessionRunTerminalSourceId( + runId: SessionRunId, + kind: TerminalSessionRunEventKind, +): string { + return `session-run-terminal:${runId}:${kind}`; +} export function createSessionRunTerminalFailureSourceId(runId: SessionRunId): string { - return `session-run-terminal:${runId}:run.failed`; + return createSessionRunTerminalSourceId(runId, "run.failed"); } diff --git a/apps/api/src/modules/runtime/infrastructure/runtime-subject-lifecycle/runtime-subject-maintenance.service.ts b/apps/api/src/modules/runtime/infrastructure/runtime-subject-lifecycle/runtime-subject-maintenance.service.ts index e4bdc3b9..ff71e809 100644 --- a/apps/api/src/modules/runtime/infrastructure/runtime-subject-lifecycle/runtime-subject-maintenance.service.ts +++ b/apps/api/src/modules/runtime/infrastructure/runtime-subject-lifecycle/runtime-subject-maintenance.service.ts @@ -13,6 +13,7 @@ import { syncSessionViewerState } from "../../../sessions/application/session-vi import { RESCHEDULING_RECONNECT_WINDOW_MS } from "../../../sessions/domain/session-lifecycle"; import { createSessionLifecycleTerminatedEvent } from "../../application/session-runs/session-run-view-events.service"; import { reconcileStaleActiveSessionRuns } from "../../application/session-runs/stale-run-reconciliation.service"; +import { reconcileTerminalSessionRuns } from "../../application/session-runs/terminal-run-reconciliation.service"; import { getRuntimeKindPolicy } from "../../domain/runtime-kind-policy"; import { cleanupDriverInstances } from "../driver-instance/maintenance"; import { repairRuntimeCommandRecords } from "../session-runs/runtime-command-store.repository"; @@ -292,8 +293,16 @@ export async function runSandboxMaintenance(bindings: ApiBindings): Promise syncSessionViewerState(bindings, sessionId), ); diff --git a/apps/api/src/modules/runtime/infrastructure/session-runs/session-run-write.repository.ts b/apps/api/src/modules/runtime/infrastructure/session-runs/session-run-write.repository.ts index 45008870..1002cb39 100644 --- a/apps/api/src/modules/runtime/infrastructure/session-runs/session-run-write.repository.ts +++ b/apps/api/src/modules/runtime/infrastructure/session-runs/session-run-write.repository.ts @@ -16,10 +16,14 @@ import type { SessionRunId, } from "@mosoo/id"; import { generateTraceId } from "@mosoo/observability"; -import { and, eq, inArray, notInArray, sql } from "drizzle-orm"; +import { and, eq, exists, inArray, notInArray, sql } from "drizzle-orm"; import type { SQL } from "drizzle-orm"; -import { getAppDatabase, getD1ChangeCount } from "../../../../platform/db/drizzle"; +import { + getAppDatabase, + getD1ChangeCount, + runAppDatabaseBatch, +} from "../../../../platform/db/drizzle"; import { currentTimestampMs, toIsoString } from "../../../../time"; import { toSessionLifecycleStatusForRunStatus } from "../../../sessions/domain/session-lifecycle"; import type { BoundCapabilityRunProvenance } from "../../domain/bound-capability-run-provenance"; @@ -698,33 +702,31 @@ async function transitionSessionRunStatus( } } - const runUpdateValues = createSessionRunStatusUpdate(input, timestampMs); - - const runUpdateResult = await getAppDatabase(database) - .update(sessionRunsTable) - .set(runUpdateValues) - .where( - and( - eq(sessionRunsTable.id, input.runId), - eq(sessionRunsTable.status, current.status), - eq(sessionRunsTable.statusSeq, current.status_seq), - ), - ) - .run(); - - if (getD1ChangeCount(runUpdateResult) === 0) { - return { - currentStatus: current.status, - kind: "stale", - reason: "concurrent_transition", - targetStatus: input.status, - }; - } - const run = toUpdatedSessionRunSummary(current, input, timestampMs); const statusSeq = current.status_seq + 1; if (input.preserveSessionLifecycle === true) { + const runUpdateResult = await getAppDatabase(database) + .update(sessionRunsTable) + .set(createSessionRunStatusUpdate(input, timestampMs)) + .where( + and( + eq(sessionRunsTable.id, input.runId), + eq(sessionRunsTable.status, current.status), + eq(sessionRunsTable.statusSeq, current.status_seq), + ), + ) + .run(); + + if (getD1ChangeCount(runUpdateResult) === 0) { + return { + currentStatus: current.status, + kind: "stale", + reason: "concurrent_transition", + targetStatus: input.status, + }; + } + return { kind: "applied", previousStatus: current.status, @@ -735,6 +737,27 @@ async function transitionSessionRunStatus( } if (current.session_last_run_id !== input.runId) { + const runUpdateResult = await getAppDatabase(database) + .update(sessionRunsTable) + .set(createSessionRunStatusUpdate(input, timestampMs)) + .where( + and( + eq(sessionRunsTable.id, input.runId), + eq(sessionRunsTable.status, current.status), + eq(sessionRunsTable.statusSeq, current.status_seq), + ), + ) + .run(); + + if (getD1ChangeCount(runUpdateResult) === 0) { + return { + currentStatus: current.status, + kind: "stale", + reason: "concurrent_transition", + targetStatus: input.status, + }; + } + return { kind: "applied", previousStatus: current.status, @@ -744,22 +767,54 @@ async function transitionSessionRunStatus( }; } - const sessionUpdateResult = await getAppDatabase(database) - .update(sessionsTable) - .set( - createSessionStatusTransitionPatch({ - status: toSessionLifecycleStatusForRunStatus(input.status), - timestampMs, - }), - ) - .where( - and( - eq(sessionsTable.id, current.session_id), - eq(sessionsTable.lastRunId, input.runId), - notInArray(sessionsTable.status, ["TERMINATED"]), + const [runUpdateResult, sessionUpdateResult] = await runAppDatabaseBatch(database, (db) => [ + db + .update(sessionRunsTable) + .set(createSessionRunStatusUpdate(input, timestampMs)) + .where( + and( + eq(sessionRunsTable.id, input.runId), + eq(sessionRunsTable.status, current.status), + eq(sessionRunsTable.statusSeq, current.status_seq), + ), ), - ) - .run(); + db + .update(sessionsTable) + .set( + createSessionStatusTransitionPatch({ + status: toSessionLifecycleStatusForRunStatus(input.status), + timestampMs, + }), + ) + .where( + and( + eq(sessionsTable.id, current.session_id), + eq(sessionsTable.lastRunId, input.runId), + notInArray(sessionsTable.status, ["TERMINATED"]), + exists( + db + .select({ id: sessionRunsTable.id }) + .from(sessionRunsTable) + .where( + and( + eq(sessionRunsTable.id, input.runId), + eq(sessionRunsTable.status, input.status), + eq(sessionRunsTable.statusSeq, statusSeq), + ), + ), + ), + ), + ), + ]); + + if (getD1ChangeCount(runUpdateResult) === 0) { + return { + currentStatus: current.status, + kind: "stale", + reason: "concurrent_transition", + targetStatus: input.status, + }; + } if (getD1ChangeCount(sessionUpdateResult) === 0 && current.session_status !== "TERMINATED") { return { diff --git a/apps/api/tests/helpers/sqlite-d1.ts b/apps/api/tests/helpers/sqlite-d1.ts index 42e1a005..be3340bb 100644 --- a/apps/api/tests/helpers/sqlite-d1.ts +++ b/apps/api/tests/helpers/sqlite-d1.ts @@ -10,6 +10,7 @@ const statementQueries = new WeakMap(); export class SqliteD1Database implements D1Database { readonly #database = new Database(":memory:"); readonly #maxBoundParams: number | undefined; + #batchTail: Promise = Promise.resolve(); constructor(input: { foreignKeys?: boolean; maxBoundParams?: number } = {}) { this.#maxBoundParams = input.maxBoundParams; @@ -29,10 +30,18 @@ export class SqliteD1Database implements D1Database { } async batch(statements: D1PreparedStatement[]): Promise[]> { + let releaseBatch: (() => void) | null = null; + const previousBatch = this.#batchTail; + this.#batchTail = new Promise((resolve) => { + releaseBatch = resolve; + }); + await previousBatch; + const results: D1Result[] = []; - this.#database.run("BEGIN"); try { + this.#database.run("BEGIN"); + for (const statement of statements) { const query = statementQueries.get(statement) ?? ""; const returnsRows = /^\s*(select|with)\b/i.test(query) || /\breturning\b/i.test(query); @@ -45,6 +54,8 @@ export class SqliteD1Database implements D1Database { } catch (error) { this.#database.run("ROLLBACK"); throw error; + } finally { + releaseBatch?.(); } } diff --git a/apps/api/tests/session-run-lifecycle.test.ts b/apps/api/tests/session-run-lifecycle.test.ts index 3f31c61a..051f3ce1 100644 --- a/apps/api/tests/session-run-lifecycle.test.ts +++ b/apps/api/tests/session-run-lifecycle.test.ts @@ -64,6 +64,71 @@ async function insertSessionRun( .run(); } +function failSessionProjectionStatementInBatch(database: D1Database): D1Database { + return new Proxy(database, { + get(target, property, receiver) { + if (property === "batch") { + return async (statements: D1PreparedStatement[]) => { + const firstStatement = statements[0]; + + if (firstStatement === undefined) { + throw new Error("Expected a Run status statement in the D1 batch."); + } + + const failingStatement = new Proxy(target.prepare("SELECT 1"), { + get(statement, statementProperty, statementReceiver) { + if (statementProperty === "run") { + return async () => { + throw new Error("injected Session lifecycle projection failure"); + }; + } + + const value = Reflect.get(statement, statementProperty, statementReceiver); + return typeof value === "function" ? value.bind(statement) : value; + }, + }); + + return target.batch([firstStatement, failingStatement]); + }; + } + + const value = Reflect.get(target, property, receiver); + return typeof value === "function" ? value.bind(target) : value; + }, + }); +} + +function advanceRunBeforeBatch( + database: D1Database, + input: { + readonly runId: string; + }, +): D1Database { + let advanced = false; + + return new Proxy(database, { + get(target, property, receiver) { + if (property === "batch") { + return async (statements: D1PreparedStatement[]) => { + if (!advanced) { + advanced = true; + await setSessionRunStatus(target, { + runId: input.runId, + source: "driver", + status: "running", + }); + } + + return target.batch(statements); + }; + } + + const value = Reflect.get(target, property, receiver); + return typeof value === "function" ? value.bind(target) : value; + }, + }); +} + describe("session run lifecycle", () => { test("does not let a stale terminal event revive or overwrite a completed run", async () => { const database = await createPublicHttpContractDatabase(); @@ -223,6 +288,106 @@ describe("session run lifecycle", () => { }); }); + test("rolls back a terminal Run transition when its Session projection fails", async () => { + const database = await createPublicHttpContractDatabase(); + await insertNonOwnerSession(database); + await insertSessionRun(database, { + runId: "run-atomic-terminal-projection", + status: "running", + }); + + await expect( + setSessionRunStatus(failSessionProjectionStatementInBatch(database), { + runId: "run-atomic-terminal-projection", + source: "driver", + status: "completed", + }), + ).rejects.toThrow("injected Session lifecycle projection failure"); + + const interrupted = await database + .prepare( + ` + SELECT session.status AS session_status, session_run.status AS run_status + FROM session + INNER JOIN session_run ON session_run.id = session.last_run_id + WHERE session.id = ? + `, + ) + .bind("01J0000000000000000000000B") + .first<{ run_status: string; session_status: string }>(); + + expect(interrupted).toEqual({ + run_status: "running", + session_status: "RUNNING", + }); + + await setSessionRunStatus(database, { + runId: "run-atomic-terminal-projection", + source: "driver", + status: "completed", + }); + + const admitted = await createSessionRunRecordIfSessionIdle(database, { + agentId: "01J00000000000000000000009", + createdBy: "01J00000000000000000000002", + model: "gpt-5.4", + provider: "openai", + runtimeId: "openai-runtime", + sessionId: "01J0000000000000000000000B", + status: "queued", + trigger: "user_prompt", + }); + + expect(admitted.createdRun).not.toBeNull(); + }); + + test("does not project a stale terminal transition onto a newer Run state", async () => { + const database = await createPublicHttpContractDatabase(); + await insertNonOwnerSession(database); + await insertSessionRun(database, { + runId: "run-stale-terminal-projection", + status: "booting", + }); + + const outcome = await setSessionRunStatus( + advanceRunBeforeBatch(database, { + runId: "run-stale-terminal-projection", + }), + { + error: { + code: "runtime.stale_terminal", + details: {}, + message: "The stale terminal transition must not update the Session.", + retryable: false, + }, + runId: "run-stale-terminal-projection", + source: "driver", + status: "failed", + }, + ); + + expect(outcome).toMatchObject({ + kind: "stale", + reason: "concurrent_transition", + }); + const current = await database + .prepare( + ` + SELECT session.status AS session_status, session_run.status AS run_status + FROM session + INNER JOIN session_run ON session_run.id = session.last_run_id + WHERE session.id = ? + `, + ) + .bind("01J0000000000000000000000B") + .first<{ run_status: string; session_status: string }>(); + + expect(current).toEqual({ + run_status: "running", + session_status: "RUNNING", + }); + }); + test("does not revive terminated sessions from stale run projections", async () => { const database = await createPublicHttpContractDatabase(); await insertNonOwnerSession(database); diff --git a/apps/api/tests/session-run-terminal-failure.test.ts b/apps/api/tests/session-run-terminal-failure.test.ts index c4aa6f36..54b023a0 100644 --- a/apps/api/tests/session-run-terminal-failure.test.ts +++ b/apps/api/tests/session-run-terminal-failure.test.ts @@ -3,6 +3,7 @@ import { describe, expect, test } from "bun:test"; import type { DriverInstanceId, SessionRunId } from "@mosoo/id"; import { recordCanonicalSessionRunFailure } from "../src/modules/runtime/application/session-runs/session-run-terminal-failure.service"; +import { reconcileTerminalSessionRuns } from "../src/modules/runtime/application/session-runs/terminal-run-reconciliation.service"; import { getRuntimeSessionLink } from "../src/modules/runtime/infrastructure/driver-instance/session-link.repository"; import { recordDriverInstanceFailure } from "../src/modules/runtime/infrastructure/driver-instance/terminal-driver-events"; import { setSessionRunStatus } from "../src/modules/runtime/infrastructure/session-runs/session-run-store.repository"; @@ -302,6 +303,152 @@ describe("canonical session run terminal failure", () => { ]); }); + test("repairs a lost terminal event after the Driver has stopped", async () => { + const database = await createPublicHttpContractDatabase(); + await insertLinkedRunFixture(database); + const bindings = createPublicHttpTestBindings(database) as ApiBindings; + + await setSessionRunStatus(database, { + error: PROVISION_ERROR, + runId: RUN_ID, + source: "api", + status: "failed", + }); + + await database + .prepare("UPDATE driver_instance SET status = ? WHERE id = ?") + .bind("stopped", DRIVER_ID) + .run(); + + expect(await readFailureEvents(database)).toEqual([]); + + const firstRepair = await reconcileTerminalSessionRuns(bindings, { limit: 10 }); + const secondRepair = await reconcileTerminalSessionRuns(bindings, { limit: 10 }); + + expect(firstRepair.reconciledRunIds).toEqual([RUN_ID]); + expect(secondRepair.reconciledRunIds).toEqual([]); + expect(await readFailureEvents(database)).toEqual([ + { + content_text: PROVISION_ERROR.message, + event_type: "run.failed", + source_event_id: CANONICAL_FAILURE_SOURCE_ID, + }, + ]); + }); + + test("repairs an inherited terminal Run whose Session and completion event are both stale", async () => { + const database = await createPublicHttpContractDatabase(); + await insertLinkedRunFixture(database, "completed"); + const bindings = createPublicHttpTestBindings(database) as ApiBindings; + + await database + .prepare("UPDATE session SET status = ? WHERE id = ?") + .bind("RUNNING", PUBLIC_API_TEST_IDS.ownerSession) + .run(); + await database + .prepare("UPDATE driver_instance SET status = ? WHERE id = ?") + .bind("stopped", DRIVER_ID) + .run(); + + const repaired = await reconcileTerminalSessionRuns(bindings, { limit: 10 }); + const session = await database + .prepare("SELECT status FROM session WHERE id = ?") + .bind(PUBLIC_API_TEST_IDS.ownerSession) + .first<{ status: string }>(); + const completedEvents = await database + .prepare( + ` + SELECT event_type, source_event_id + FROM session_event + WHERE run_id = ? AND event_type = 'run.completed' + `, + ) + .bind(RUN_ID) + .all<{ event_type: string; source_event_id: string }>(); + + expect(repaired.reconciledRunIds).toEqual([RUN_ID]); + expect(session).toEqual({ status: "IDLE" }); + expect(completedEvents.results).toEqual([ + { + event_type: "run.completed", + source_event_id: `session-run-terminal:${RUN_ID}:run.completed`, + }, + ]); + }); + + test("repairs an old terminal receipt without changing a newer Run", async () => { + const database = await createPublicHttpContractDatabase(); + await insertLinkedRunFixture(database, "completed"); + const bindings = createPublicHttpTestBindings(database) as ApiBindings; + + await database + .prepare( + ` + INSERT INTO session_run ( + id, + session_id, + agent_id, + created_by_account_id, + trigger, + status, + provider, + model, + runtime_id, + trace_id, + created_at, + updated_at + ) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + `, + ) + .bind( + PUBLIC_API_TEST_IDS.runAlt, + PUBLIC_API_TEST_IDS.ownerSession, + PUBLIC_API_TEST_IDS.agent, + PUBLIC_API_TEST_IDS.ownerAccount, + "user_prompt", + "running", + "openai", + "gpt-5.4", + "openai-runtime", + "trace-newer-run", + 2, + 2, + ) + .run(); + await database + .prepare("UPDATE session SET last_run_id = ?, status = ? WHERE id = ?") + .bind(PUBLIC_API_TEST_IDS.runAlt, "RUNNING", PUBLIC_API_TEST_IDS.ownerSession) + .run(); + await database + .prepare("UPDATE driver_instance SET status = ? WHERE id = ?") + .bind("stopped", DRIVER_ID) + .run(); + + await reconcileTerminalSessionRuns(bindings, { limit: 10 }); + + const session = await database + .prepare("SELECT last_run_id, status FROM session WHERE id = ?") + .bind(PUBLIC_API_TEST_IDS.ownerSession) + .first<{ last_run_id: string; status: string }>(); + + expect(session).toEqual({ + last_run_id: PUBLIC_API_TEST_IDS.runAlt, + status: "RUNNING", + }); + const completedEvent = await database + .prepare( + "SELECT source_event_id FROM session_event WHERE run_id = ? AND event_type = 'run.completed'", + ) + .bind(RUN_ID) + .first<{ source_event_id: string }>(); + + expect(completedEvent).toEqual({ + source_event_id: `session-run-terminal:${RUN_ID}:run.completed`, + }); + expect(await readFailureEvents(database)).toEqual([]); + }); + test("does not append a failure after a non-failed terminal outcome", async () => { for (const status of ["completed", "cancelled"] as const) { const database = await createPublicHttpContractDatabase();