diff --git a/package.json b/package.json index 17e65e9..9d2bcb5 100644 --- a/package.json +++ b/package.json @@ -9,6 +9,7 @@ "self-test:postgres-kafka-order": "bun run src/selftest/integration/postgres-kafka-order.ts", "self-test:postgres-queue-retry": "bun run src/selftest/integration/postgres-queue-retry.ts", "self-test:postgres-runner-retention-fence": "bun run src/selftest/integration/postgres-runner-retention-fence.ts", + "self-test:postgres-session-turn-admission": "bun run src/selftest/integration/postgres-session-turn-admission.ts", "test": "bun run src/selftest/run.ts", "cli": "bun scripts/agentrun-cli.ts" }, diff --git a/src/common/validation.ts b/src/common/validation.ts index 1cc0312..04fb178 100644 --- a/src/common/validation.ts +++ b/src/common/validation.ts @@ -2,6 +2,7 @@ import { createHash, randomUUID } from "node:crypto"; import type { BackendProfile, CreateCommandInput, CreateQueueTaskInput, CreateRunInput, ExecutionPolicy, JsonRecord, JsonValue, QueueTaskState, ResourceBundleRef, SecretRef, SessionListState, SessionRef, WorkspaceRef } from "./types.js"; import { AgentRunError } from "./errors.js"; import { backendProfileIdPattern, backendProfileSpec, isBackendProfile } from "./backend-profiles.js"; +import { validateAipodImageRef } from "./aipod-image-ref-validation.js"; const backendProfilePatternText = String(backendProfileIdPattern); @@ -395,6 +396,31 @@ function validateDurableDispatchInput(input: JsonRecord): JsonRecord { return normalized; } +export function validateSessionRunnerJobInput(value: unknown): JsonRecord { + const input = asRecord(value ?? {}, "sessionSend.runnerJob"); + assertNoSecretNamedFields(input, "sessionSend.runnerJob"); + const allowedKeys = new Set([ + "managerUrl", "image", "namespace", "attemptId", "runnerId", "sourceCommit", "serviceAccountName", + "runnerIdleTimeoutMs", "backendRetryMaxAttempts", "backendRetryInitialBackoffMs", "backendRetryMaxBackoffMs", + "idempotencyKey", "imageRef", "transientEnv", + ]); + const rejectedKeys = Object.keys(input).filter((key) => !allowedKeys.has(key)); + if (rejectedKeys.length > 0) throw new AgentRunError("schema-invalid", "session runner admission contains unsupported fields", { httpStatus: 400, details: { rejectedKeys: rejectedKeys.sort(), valuesPrinted: false } }); + const normalized: JsonRecord = {}; + for (const key of ["managerUrl", "image", "namespace", "attemptId", "runnerId", "sourceCommit", "serviceAccountName", "idempotencyKey"] as const) { + if (input[key] !== undefined) normalized[key] = credentialFreeConfigValue(boundedDispatchString(input[key], `sessionSend.runnerJob.${key}`, key === "managerUrl" || key === "image" ? 2048 : 256), `sessionSend.runnerJob.${key}`); + } + for (const key of ["runnerIdleTimeoutMs", "backendRetryMaxAttempts", "backendRetryInitialBackoffMs", "backendRetryMaxBackoffMs"] as const) { + if (input[key] === undefined) continue; + const number = Number(input[key]); + if (!Number.isSafeInteger(number) || number < 1) throw new AgentRunError("schema-invalid", `sessionSend.runnerJob.${key} must be a positive integer`, { httpStatus: 400 }); + normalized[key] = number; + } + if (input.imageRef !== undefined) normalized.imageRef = validateAipodImageRef(input.imageRef, "sessionSend.runnerJob.imageRef"); + if (input.transientEnv !== undefined) normalized.transientEnv = validateDurableTransientEnv(input.transientEnv); + return normalized; +} + function validateDurableTransientEnv(value: unknown): JsonRecord[] { if (!Array.isArray(value)) throw new AgentRunError("schema-invalid", "command.dispatch.input.transientEnv must be an array", { httpStatus: 400 }); if (value.length > 64) throw new AgentRunError("schema-invalid", "command.dispatch.input.transientEnv must contain at most 64 entries", { httpStatus: 400 }); diff --git a/src/mgr/postgres-store.ts b/src/mgr/postgres-store.ts index 1d4a3e5..6d58678 100644 --- a/src/mgr/postgres-store.ts +++ b/src/mgr/postgres-store.ts @@ -5,8 +5,8 @@ import { AgentRunError } from "../common/errors.js"; import { redactJson } from "../common/redaction.js"; import type { BackendProfile, BackendTurnResult, CancelRequestRecord, CancelStage, CancelTargetKind, CommandRecord, CommandState, CreateCommandInput, CreateQueueTaskInput, CreateRunInput, EventType, FailureKind, JsonRecord, JsonValue, KafkaEventOutboxRecord, ListGcExpiredSessionsInput, QueueAttemptListResult, QueueAttemptRecord, QueueAttemptRef, QueueCommanderSnapshot, QueueReadCursorRecord, QueueRetryActivation, QueueRetryReservation, QueueStats, QueueTaskListResult, QueueTaskRecord, QueueTaskState, RunEvent, RunnerDispatchCompletion, RunnerDispatchIntentRecord, RunnerJobRecord, RunnerRecord, RunRecord, RunStatus, SessionEventPage, SessionListResult, SessionReadCursorRecord, SessionRecord, SessionRef, SessionStoragePatch, SessionSummary, TerminalStatus, UpsertSessionInput } from "../common/types.js"; import { newId, nowIso, stableHash } from "../common/validation.js"; -import type { ActivateQueueRetryAttemptInput, AgentRunStore, CreateRunIdentity, DurableQueueClaim, EventOutboxConfig, ListQueueAttemptsInput, ListQueueTasksInput, ListSessionsInput, RecordQueueRetryAttemptFailureInput, ReserveQueueRetryAttemptInput, RunnerRetentionFenceFacts, RunnerRetentionFenceInput, SaveRunnerJobInput, SessionEventPageInput, StoreHealth, UpdateQueueTaskAttemptInput } from "./store.js"; -import { assertQueueAttemptActivationReplay, assertQueueRetryable, assertQueueTaskPayloadHash, assertRunCreateReplay, assertSessionBoundary, buildQueueStats, buildQueueTaskSummary, buildSessionSummary, cancelStagePayload, clampQueueLimit, clampSessionLimit, commandStateFromTerminal, fenceLateEventForCancelledRun, isLeaseExpired, isSessionOutputEvent, isTerminalCommandState, isTerminalQueueAttemptState, isTerminalQueueTaskState, isTerminalRunStatus, latestQueueAttempt, lateWriteRejectedPayload, nextQueueRetryIndex, parseQueueCursor, parseSessionCursor, queueAttemptRef, queueRetryReservation, queueRetryStateError, queueTaskMatchesCommander, queueTaskPayloadHash, queueTaskSort, runCreatedEventPayload, sessionListFilters, sessionMatchesListState, sessionRefFromRecord, sessionSort, sessionTitleFromCommand, statusFromTerminal, summarizeSessionRef, titleFromMetadata } from "./store.js"; +import type { ActivateQueueRetryAttemptInput, AgentRunStore, CreateRunIdentity, DurableQueueClaim, EventOutboxConfig, ListQueueAttemptsInput, ListQueueTasksInput, ListSessionsInput, RecordQueueRetryAttemptFailureInput, ReserveQueueRetryAttemptInput, RunnerRetentionFenceFacts, RunnerRetentionFenceInput, SaveRunnerJobInput, SessionEventPageInput, SessionTurnAdmission, SessionTurnAdmissionInput, StoreHealth, UpdateQueueTaskAttemptInput } from "./store.js"; +import { activeSessionAdmissionConflict, assertQueueAttemptActivationReplay, assertQueueRetryable, assertQueueTaskPayloadHash, assertRunCreateReplay, assertSessionBoundary, assertSessionCommandReplay, assertSessionTurnAdmissionContract, buildQueueStats, buildQueueTaskSummary, buildSessionSummary, cancelStagePayload, clampQueueLimit, clampSessionLimit, commandStateFromTerminal, fenceLateEventForCancelledRun, invalidSessionAdmissionProjection, isLeaseExpired, isSessionOutputEvent, isTerminalCommandState, isTerminalQueueAttemptState, isTerminalQueueTaskState, isTerminalRunStatus, latestQueueAttempt, lateWriteRejectedPayload, nextQueueRetryIndex, parseQueueCursor, parseSessionCursor, queueAttemptRef, queueRetryReservation, queueRetryStateError, queueTaskMatchesCommander, queueTaskPayloadHash, queueTaskSort, runCreatedEventPayload, sessionListFilters, sessionMatchesListState, sessionRefFromRecord, sessionSort, sessionTitleFromCommand, statusFromTerminal, summarizeSessionRef, titleFromMetadata } from "./store.js"; import { backendCapabilitiesSqlValues, mergeBackendCapability } from "../common/backend-profiles.js"; import { normalizeRunEventPayload, requireEventType, requireExternallyAppendableEventType, userMessagePayloadForCommand } from "../common/events.js"; import { buildKafkaEventOutboxRecord, canonicalAgentRunEventPartitionKey } from "./event-outbox.js"; @@ -574,28 +574,7 @@ CREATE TABLE IF NOT EXISTS agentrun_schema_migrations ( } async createRun(input: CreateRunInput, identity?: CreateRunIdentity): Promise { - const at = nowIso(); - return this.withTransaction(async (client) => { - if (identity) { - if (!identity.id.trim()) throw new AgentRunError("schema-invalid", "reserved run identity must be non-empty", { httpStatus: 400 }); - await client.query("SELECT pg_advisory_xact_lock(hashtext($1), hashtext($2))", ["agentrun-create-run", identity.id]); - const existing = await client.query("SELECT * FROM agentrun_runs WHERE id = $1", [identity.id]); - if (existing.rows[0]) { - const run = runFromRow(existing.rows[0]); - assertRunCreateReplay(run, input); - return run; - } - } - const sessionRef = await this.resolveSessionForRun(client, input, at); - const run: RunRecord = { ...input, sessionRef, resourceBundleRef: input.resourceBundleRef ?? null, id: identity?.id ?? newId("run"), status: "pending", terminalStatus: null, failureKind: null, failureMessage: null, cancelEpoch: 0, cancelRequestId: null, cancelRequestedAt: null, cancelReason: null, createdAt: at, updatedAt: at, claimedBy: null, leaseExpiresAt: null }; - await client.query( - `INSERT INTO agentrun_runs (id, tenant_id, project_id, workspace_ref, session_ref, resource_bundle_ref, provider_id, backend_profile, execution_policy, trace_sink, status, terminal_status, failure_kind, failure_message, created_at, updated_at, claimed_by, lease_expires_at) - VALUES ($1, $2, $3, $4::jsonb, $5::jsonb, $6::jsonb, $7, $8, $9::jsonb, $10::jsonb, $11, $12, $13, $14, $15, $16, $17, $18)`, - [run.id, run.tenantId, run.projectId, JSON.stringify(run.workspaceRef), JSON.stringify(run.sessionRef), JSON.stringify(run.resourceBundleRef), run.providerId, run.backendProfile, JSON.stringify(run.executionPolicy), JSON.stringify(run.traceSink), run.status, run.terminalStatus, run.failureKind, run.failureMessage, run.createdAt, run.updatedAt, run.claimedBy, run.leaseExpiresAt], - ); - await this.touchSessionForRun(client, run, { lastRunId: run.id, lastActivityAt: at }, { bumpVersion: false, at }); - return run; - }); + return this.withTransaction(async (client) => await this.createRunWithClient(client, input, identity)); } async getRun(runId: string): Promise { @@ -629,55 +608,53 @@ CREATE TABLE IF NOT EXISTS agentrun_schema_migrations ( } async createCommand(runId: string, input: CreateCommandInput): Promise { - const payloadHash = stableHash(input.payload); + return this.withTransaction(async (client) => await this.createCommandWithClient(client, runId, input)); + } + + async admitSessionTurn(input: SessionTurnAdmissionInput): Promise { + assertSessionTurnAdmissionContract(input); return this.withTransaction(async (client) => { - const run = await this.requireRunForUpdate(client, runId); - if (input.idempotencyKey) { - const existing = await client.query("SELECT * FROM agentrun_commands WHERE run_id = $1 AND idempotency_key = $2", [runId, input.idempotencyKey]); - if (existing.rows[0]) { - const command = commandFromRow(existing.rows[0]); - if (command.payloadHash !== payloadHash) throw new AgentRunError("schema-invalid", "idempotency key reused with different payload", { httpStatus: 409 }); - const intentResult = await client.query("SELECT * FROM agentrun_runner_dispatch_intents WHERE command_id = $1", [command.id]); - const intent = intentResult.rows[0] ? runnerDispatchIntentFromRow(intentResult.rows[0]) : null; - assertRunnerDispatchReplay(intent, command, input.dispatch); - return attachRunnerDispatchIntent(command, intent); + await client.query("SELECT pg_advisory_xact_lock(hashtext($1), hashtext($2))", ["agentrun-session-turn-admission", input.sessionId]); + const sessionResult = await client.query("SELECT * FROM agentrun_sessions WHERE session_id = $1 FOR UPDATE", [input.sessionId]); + const session = sessionResult.rows[0] ? sessionFromRow(sessionResult.rows[0]) : null; + if (session) assertSessionBoundary(session, input.run); + + if (input.command.idempotencyKey) { + const replayResult = await client.query( + `SELECT c.* FROM agentrun_commands c + JOIN agentrun_runs r ON r.id = c.run_id + WHERE r.session_ref->>'sessionId' = $1 AND c.idempotency_key = $2 + ORDER BY c.created_at DESC LIMIT 1 FOR UPDATE OF r, c`, + [input.sessionId, input.command.idempotencyKey], + ); + if (replayResult.rows[0]) { + const command = commandFromRow(replayResult.rows[0]); + const run = await this.requireRunForUpdate(client, command.runId); + return await this.replayOrRecoverSessionTurnWithClient(client, run, command, input, "replayed"); } } - if (isTerminalRunStatus(run.status)) throw new AgentRunError(run.failureKind ?? (run.status === "cancelled" ? "cancelled" : "schema-invalid"), `run ${runId} is already terminal: ${run.status}`, { httpStatus: 409 }); - const runCreatedResult = await client.query<{ exists: boolean }>( - `SELECT EXISTS ( - SELECT 1 FROM agentrun_events - WHERE run_id = $1 - AND type = 'backend_status' - AND payload->>'phase' = 'run-created' - ) AS exists`, - [runId], - ); - const needsRunCreatedEvent = runCreatedResult.rows[0]?.exists !== true; - const seq = await this.nextSeq(client, "agentrun_commands", runId); - const at = nowIso(); - const { dispatch: _dispatch, ...commandInput } = input; - const command: CommandRecord = { ...commandInput, id: newId("cmd"), runId, seq, state: "pending", payloadHash, cancelEpoch: 0, cancelRequestId: null, cancelRequestedAt: null, cancelReason: null, createdAt: at, updatedAt: at, acknowledgedAt: null }; - await client.query( - `INSERT INTO agentrun_commands (id, run_id, seq, type, payload, payload_hash, idempotency_key, state, created_at, updated_at, acknowledged_at) - VALUES ($1, $2, $3, $4, $5::jsonb, $6, $7, $8, $9, $10, $11)`, - [command.id, command.runId, command.seq, command.type, JSON.stringify(command.payload), command.payloadHash, command.idempotencyKey ?? null, command.state, command.createdAt, command.updatedAt, command.acknowledgedAt], - ); - const intent = input.dispatch ? newRunnerDispatchIntent(command, input.dispatch, at) : null; - if (intent) { - await client.query( - `INSERT INTO agentrun_runner_dispatch_intents (id, run_id, command_id, kind, input, input_hash, state, attempt_count, runner_job_id, lease_owner, lease_expires_at, next_attempt_at, last_error, dispatched_at, created_at, updated_at) - VALUES ($1, $2, $3, $4, $5::jsonb, $6, $7, $8, $9, $10, $11, $12, $13::jsonb, $14, $15, $16)`, - [intent.id, intent.runId, intent.commandId, intent.kind, JSON.stringify(intent.input), intent.inputHash, intent.state, intent.attemptCount, intent.runnerJobId, intent.leaseOwner, intent.leaseExpiresAt, intent.nextAttemptAt, JSON.stringify(intent.lastError), intent.dispatchedAt, intent.createdAt, intent.updatedAt], - ); + + if (session?.activeRunId || session?.activeCommandId) { + if (!session.activeRunId || !session.activeCommandId) throw invalidSessionAdmissionProjection(input.sessionId); + const runResult = await client.query("SELECT * FROM agentrun_runs WHERE id = $1 FOR UPDATE", [session.activeRunId]); + const commandResult = await client.query("SELECT * FROM agentrun_commands WHERE id = $1 FOR UPDATE", [session.activeCommandId]); + if (!runResult.rows[0] || !commandResult.rows[0]) throw invalidSessionAdmissionProjection(input.sessionId); + const run = runFromRow(runResult.rows[0]); + const command = commandFromRow(commandResult.rows[0]); + if (command.runId !== run.id || run.sessionRef?.sessionId !== input.sessionId) throw invalidSessionAdmissionProjection(input.sessionId); + if (!isTerminalRunStatus(run.status) && !isTerminalCommandState(command.state)) { + try { + return await this.replayOrRecoverSessionTurnWithClient(client, run, command, input, "replayed"); + } catch (error) { + if (error instanceof AgentRunError && error.failureKind === "schema-invalid") throw activeSessionAdmissionConflict(input.sessionId, run, command); + throw error; + } + } } - if (command.type === "turn") await this.touchSessionForRun(client, run, { executionState: "running", lastRunId: run.id, lastCommandId: command.id, activeRunId: run.id, activeCommandId: command.id, terminalStatus: null, failureKind: null, title: sessionTitleFromCommand(command), lastActivityAt: at }, { bumpVersion: true, at }); - else if (command.type === "steer") await this.touchSessionForRun(client, run, { executionState: "running", lastRunId: run.id, lastCommandId: command.id, activeRunId: run.id, lastActivityAt: at }, { bumpVersion: true, at }); - const userMessage = userMessagePayloadForCommand(run, command); - if (userMessage) await this.appendEventWithLockedRun(client, runId, "user_message", userMessage); - if (needsRunCreatedEvent) await this.appendEventWithLockedRun(client, runId, "backend_status", runCreatedEventPayload(run)); - await this.appendEventWithLockedRun(client, runId, "backend_status", { phase: "command-created", commandId: command.id, commandType: command.type }); - return attachRunnerDispatchIntent(command, intent); + + const run = await this.createRunWithClient(client, input.run); + const command = await this.createCommandWithClient(client, run.id, input.command); + return { run, command, disposition: "created", recoveredPriorPartialWrite: false, partialWrite: false }; }); } @@ -767,6 +744,7 @@ CREATE TABLE IF NOT EXISTS agentrun_schema_migrations ( if (locked.fenced) return { intent, command, run, dispatchFenced: true, valuesPrinted: false }; const failure = redactJson(error); const message = typeof failure.message === "string" ? failure.message : "runner dispatch failed"; + const reason = typeof failure.reason === "string" ? failure.reason : "runner-dispatch-attempt-failed"; const at = nowIso(); const updatedIntentResult = await client.query( `UPDATE agentrun_runner_dispatch_intents @@ -789,7 +767,7 @@ CREATE TABLE IF NOT EXISTS agentrun_schema_migrations ( ); updatedRun = runFromRow(updated.rows[0]); await fenceActiveRunnerDispatchIntents(client, { runId: run.id }, "run terminalized by runner dispatch failure", at); - await this.appendEventWithLockedRun(client, run.id, "backend_status", { phase: "runner-dispatch-failed", commandId: command.id, dispatchIntentId: intent.id, attemptCount: intent.attemptCount, failureKind: "infra-failed", message, valuesPrinted: false }); + await this.appendEventWithLockedRun(client, run.id, "backend_status", { phase: "runner-dispatch-failed", reason, commandId: command.id, dispatchIntentId: intent.id, attemptCount: intent.attemptCount, failureKind: "infra-failed", message, valuesPrinted: false }); await this.appendEventWithLockedRun(client, run.id, "backend_status", { phase: "command-terminal", commandId: command.id, state: "failed", terminalStatus: "failed", failureKind: "infra-failed", message, threadId: null, turnId: null }); await this.appendEventWithLockedRun(client, run.id, "terminal_status", { terminalStatus: "failed", failureKind: "infra-failed", message }); await this.touchSessionForRun(client, updatedRun, { executionState: "terminal", activeRunId: null, activeCommandId: null, lastRunId: run.id, lastCommandId: command.id, terminalStatus: "failed", failureKind: "infra-failed", lastActivityAt: at }, { bumpVersion: true, at }); @@ -1821,6 +1799,109 @@ CREATE TABLE IF NOT EXISTS agentrun_schema_migrations ( return runFromRow(row); } + private async createRunWithClient(client: PoolClient, input: CreateRunInput, identity?: CreateRunIdentity): Promise { + const at = nowIso(); + if (identity) { + if (!identity.id.trim()) throw new AgentRunError("schema-invalid", "reserved run identity must be non-empty", { httpStatus: 400 }); + await client.query("SELECT pg_advisory_xact_lock(hashtext($1), hashtext($2))", ["agentrun-create-run", identity.id]); + const existing = await client.query("SELECT * FROM agentrun_runs WHERE id = $1", [identity.id]); + if (existing.rows[0]) { + const run = runFromRow(existing.rows[0]); + assertRunCreateReplay(run, input); + return run; + } + } + const sessionRef = await this.resolveSessionForRun(client, input, at); + const run: RunRecord = { ...input, sessionRef, resourceBundleRef: input.resourceBundleRef ?? null, id: identity?.id ?? newId("run"), status: "pending", terminalStatus: null, failureKind: null, failureMessage: null, cancelEpoch: 0, cancelRequestId: null, cancelRequestedAt: null, cancelReason: null, createdAt: at, updatedAt: at, claimedBy: null, leaseExpiresAt: null }; + await client.query( + `INSERT INTO agentrun_runs (id, tenant_id, project_id, workspace_ref, session_ref, resource_bundle_ref, provider_id, backend_profile, execution_policy, trace_sink, status, terminal_status, failure_kind, failure_message, created_at, updated_at, claimed_by, lease_expires_at) + VALUES ($1, $2, $3, $4::jsonb, $5::jsonb, $6::jsonb, $7, $8, $9::jsonb, $10::jsonb, $11, $12, $13, $14, $15, $16, $17, $18)`, + [run.id, run.tenantId, run.projectId, JSON.stringify(run.workspaceRef), JSON.stringify(run.sessionRef), JSON.stringify(run.resourceBundleRef), run.providerId, run.backendProfile, JSON.stringify(run.executionPolicy), JSON.stringify(run.traceSink), run.status, run.terminalStatus, run.failureKind, run.failureMessage, run.createdAt, run.updatedAt, run.claimedBy, run.leaseExpiresAt], + ); + await this.touchSessionForRun(client, run, { lastRunId: run.id, lastActivityAt: at }, { bumpVersion: false, at }); + return run; + } + + private async createCommandWithClient(client: PoolClient, runId: string, input: CreateCommandInput): Promise { + const payloadHash = stableHash(input.payload); + const run = await this.requireRunForUpdate(client, runId); + if (input.idempotencyKey) { + const existing = await client.query("SELECT * FROM agentrun_commands WHERE run_id = $1 AND idempotency_key = $2", [runId, input.idempotencyKey]); + if (existing.rows[0]) { + const command = commandFromRow(existing.rows[0]); + assertSessionCommandReplay(command, input); + const intent = await this.runnerDispatchIntentForCommandWithClient(client, command.id); + assertRunnerDispatchReplay(intent, command, input.dispatch); + return attachRunnerDispatchIntent(command, intent); + } + } + if (isTerminalRunStatus(run.status)) throw new AgentRunError(run.failureKind ?? (run.status === "cancelled" ? "cancelled" : "schema-invalid"), `run ${runId} is already terminal: ${run.status}`, { httpStatus: 409 }); + const runCreatedResult = await client.query<{ exists: boolean }>( + `SELECT EXISTS ( + SELECT 1 FROM agentrun_events + WHERE run_id = $1 + AND type = 'backend_status' + AND payload->>'phase' = 'run-created' + ) AS exists`, + [runId], + ); + const needsRunCreatedEvent = runCreatedResult.rows[0]?.exists !== true; + const seq = await this.nextSeq(client, "agentrun_commands", runId); + const at = nowIso(); + const { dispatch: _dispatch, ...commandInput } = input; + const command: CommandRecord = { ...commandInput, id: newId("cmd"), runId, seq, state: "pending", payloadHash, cancelEpoch: 0, cancelRequestId: null, cancelRequestedAt: null, cancelReason: null, createdAt: at, updatedAt: at, acknowledgedAt: null }; + await client.query( + `INSERT INTO agentrun_commands (id, run_id, seq, type, payload, payload_hash, idempotency_key, state, created_at, updated_at, acknowledged_at) + VALUES ($1, $2, $3, $4, $5::jsonb, $6, $7, $8, $9, $10, $11)`, + [command.id, command.runId, command.seq, command.type, JSON.stringify(command.payload), command.payloadHash, command.idempotencyKey ?? null, command.state, command.createdAt, command.updatedAt, command.acknowledgedAt], + ); + const intent = input.dispatch ? newRunnerDispatchIntent(command, input.dispatch, at) : null; + if (intent) await this.insertRunnerDispatchIntentWithClient(client, intent); + if (command.type === "turn") await this.touchSessionForRun(client, run, { executionState: "running", lastRunId: run.id, lastCommandId: command.id, activeRunId: run.id, activeCommandId: command.id, terminalStatus: null, failureKind: null, title: sessionTitleFromCommand(command), lastActivityAt: at }, { bumpVersion: true, at }); + else if (command.type === "steer") await this.touchSessionForRun(client, run, { executionState: "running", lastRunId: run.id, lastCommandId: command.id, activeRunId: run.id, lastActivityAt: at }, { bumpVersion: true, at }); + const userMessage = userMessagePayloadForCommand(run, command); + if (userMessage) await this.appendEventWithLockedRun(client, runId, "user_message", userMessage); + if (needsRunCreatedEvent) await this.appendEventWithLockedRun(client, runId, "backend_status", runCreatedEventPayload(run)); + await this.appendEventWithLockedRun(client, runId, "backend_status", { phase: "command-created", commandId: command.id, commandType: command.type }); + return attachRunnerDispatchIntent(command, intent); + } + + private async replayOrRecoverSessionTurnWithClient(client: PoolClient, run: RunRecord, command: CommandRecord, input: SessionTurnAdmissionInput, disposition: "replayed" | "recovered"): Promise { + assertRunCreateReplay(run, input.run); + assertSessionCommandReplay(command, input.command); + const intent = await this.runnerDispatchIntentForCommandWithClient(client, command.id); + if (intent) { + assertRunnerDispatchReplay(intent, command, input.command.dispatch); + return { run, command: attachRunnerDispatchIntent(command, intent), disposition: "replayed", recoveredPriorPartialWrite: false, partialWrite: false }; + } + if (!input.command.dispatch) return { run, command: attachRunnerDispatchIntent(command, null), disposition, recoveredPriorPartialWrite: false, partialWrite: false }; + if (run.status !== "pending" || run.claimedBy || command.state !== "pending") throw activeSessionAdmissionConflict(input.sessionId, run, command); + const recovered = newRunnerDispatchIntent(command, input.command.dispatch, nowIso()); + await this.insertRunnerDispatchIntentWithClient(client, recovered); + await this.appendEventWithLockedRun(client, run.id, "backend_status", { + phase: "runner-dispatch-recovered", + reason: "session-runner-admission-partial-write-recovered", + commandId: command.id, + dispatchIntentId: recovered.id, + plannedRunnerJobId: recovered.runnerJobId, + valuesPrinted: false, + }); + return { run, command: attachRunnerDispatchIntent(command, recovered), disposition: "recovered", recoveredPriorPartialWrite: true, partialWrite: false }; + } + + private async runnerDispatchIntentForCommandWithClient(client: PoolClient, commandId: string): Promise { + const result = await client.query("SELECT * FROM agentrun_runner_dispatch_intents WHERE command_id = $1", [commandId]); + return result.rows[0] ? runnerDispatchIntentFromRow(result.rows[0]) : null; + } + + private async insertRunnerDispatchIntentWithClient(client: PoolClient, intent: RunnerDispatchIntentRecord): Promise { + await client.query( + `INSERT INTO agentrun_runner_dispatch_intents (id, run_id, command_id, kind, input, input_hash, state, attempt_count, runner_job_id, lease_owner, lease_expires_at, next_attempt_at, last_error, dispatched_at, created_at, updated_at) + VALUES ($1, $2, $3, $4, $5::jsonb, $6, $7, $8, $9, $10, $11, $12, $13::jsonb, $14, $15, $16)`, + [intent.id, intent.runId, intent.commandId, intent.kind, JSON.stringify(intent.input), intent.inputHash, intent.state, intent.attemptCount, intent.runnerJobId, intent.leaseOwner, intent.leaseExpiresAt, intent.nextAttemptAt, JSON.stringify(intent.lastError), intent.dispatchedAt, intent.createdAt, intent.updatedAt], + ); + } + private async resolveSessionForRun(client: PoolClient, input: CreateRunInput, at: string): Promise { if (!input.sessionRef) return null; const existing = await client.query("SELECT * FROM agentrun_sessions WHERE session_id = $1 FOR UPDATE", [input.sessionRef.sessionId]); diff --git a/src/mgr/runner-dispatcher.ts b/src/mgr/runner-dispatcher.ts index 47e4fac..675a94a 100644 --- a/src/mgr/runner-dispatcher.ts +++ b/src/mgr/runner-dispatcher.ts @@ -1,6 +1,6 @@ import type { JsonRecord, RunnerDispatchCompletion, RunnerDispatchIntentRecord } from "../common/types.js"; import { AgentRunError } from "../common/errors.js"; -import { redactText } from "../common/redaction.js"; +import { redactJson, redactText } from "../common/redaction.js"; import { nowIso } from "../common/validation.js"; import type { AgentRunStore } from "./store.js"; import { createKubernetesRunnerJob, type RunnerJobDefaults } from "./kubernetes-runner-job.js"; @@ -57,7 +57,7 @@ export async function dispatchRunnerIntentsOnce(input: { await input.store.terminalizeRunnerDispatchIntent(intent, failure); failedCount++; } else { - await input.store.retryRunnerDispatchIntent(intent, failure, nextAttemptAt, { phase: "runner-dispatch-retry", commandId: intent.commandId, dispatchIntentId: intent.id, attemptCount: intent.attemptCount, nextAttemptAt, failureKind: "infra-failed", message: failure.message, valuesPrinted: false }); + await input.store.retryRunnerDispatchIntent(intent, failure, nextAttemptAt, { phase: "runner-dispatch-retry", reason: failure.reason, commandId: intent.commandId, dispatchIntentId: intent.id, attemptCount: intent.attemptCount, nextAttemptAt, failureKind: "infra-failed", message: failure.message, valuesPrinted: false }); retryCount++; } } catch (settleError) { @@ -99,9 +99,13 @@ export function startRunnerDispatcher(input: { } function dispatchFailure(error: unknown, intent: RunnerDispatchIntentRecord): JsonRecord { + const message = redactText(error instanceof Error ? error.message : String(error)).slice(0, 500); + const details = error instanceof AgentRunError ? redactJson(error.details ?? {}) : {}; return { - failureKind: "infra-failed", - message: redactText(error instanceof Error ? error.message : String(error)).slice(0, 500), + failureKind: error instanceof AgentRunError ? error.failureKind : "infra-failed", + reason: typeof details.reason === "string" ? details.reason : dispatchFailureReason(message), + message, + kubernetesObservation: details, dispatchIntentId: intent.id, runId: intent.runId, commandId: intent.commandId, @@ -111,6 +115,13 @@ function dispatchFailure(error: unknown, intent: RunnerDispatchIntentRecord): Js }; } +function dispatchFailureReason(message: string): string { + if (/kubectl get|retention.*observe|kubernetes.*observation/iu.test(message)) return "runner-kubernetes-observation-failed"; + if (/kubectl create|create runner job|runner job.*create/iu.test(message)) return "runner-kubernetes-job-create-failed"; + if (/aborted|attempt timeout|exceeded .* timeout/iu.test(message)) return "runner-dispatch-attempt-aborted"; + return "runner-dispatch-attempt-failed"; +} + function itemSummary(intent: RunnerDispatchIntentRecord, state: string, error: JsonRecord | null, completion?: RunnerDispatchCompletion): JsonRecord { return { dispatchIntentId: intent.id, runId: intent.runId, commandId: intent.commandId, runnerJobId: intent.runnerJobId, actualRunnerJobId: completion?.actualRunnerJobId ?? intent.actualRunnerJobId, activeRunnerId: completion?.activeRunnerId ?? intent.activeRunnerId, dispatchOutcome: completion?.dispatchOutcome ?? intent.dispatchOutcome, attemptCount: intent.attemptCount, state, failureKind: error?.failureKind ?? null, valuesPrinted: false }; } diff --git a/src/mgr/server.ts b/src/mgr/server.ts index 58b1a64..1ec7c5d 100644 --- a/src/mgr/server.ts +++ b/src/mgr/server.ts @@ -2,12 +2,12 @@ import type { Server } from "node:http"; import { createServer } from "node:http"; import type { AddressInfo } from "node:net"; import type { AgentRunStore, ListQueueTasksInput, ListSessionsInput, SessionEventPageInput } from "./store.js"; -import { openAgentRunStoreFromEnv } from "./store.js"; +import { assertSessionBoundary, openAgentRunStoreFromEnv } from "./store.js"; import { agentRunKafkaConfig, type AgentRunKafkaConfig } from "../common/kafka-events.js"; import { AgentRunError, errorToJson } from "../common/errors.js"; -import { asRecord, validateBackendProfile, validateCreateCommand, validateCreateQueueTask, validateCreateRun, validateQueueTaskState, validateSessionListState } from "../common/validation.js"; +import { asRecord, stableHash, validateBackendProfile, validateCreateCommand, validateCreateQueueTask, validateCreateRun, validateQueueTaskState, validateSessionListState, validateSessionRunnerJobInput } from "../common/validation.js"; import { isBackendProfile } from "../common/backend-profiles.js"; -import type { ApiErrorBody, ApiOkBody, CommandRecord, JsonRecord, JsonValue, QueueTaskRecord, RunEvent, RunRecord, SessionRecord } from "../common/types.js"; +import type { ApiErrorBody, ApiOkBody, CommandRecord, CreateRunInput, JsonRecord, JsonValue, QueueTaskRecord, RunEvent, RunRecord, RunnerDispatchIntentRecord, SessionRecord } from "../common/types.js"; import { createKubernetesRunnerJob, type RunnerJobDefaults } from "./kubernetes-runner-job.js"; import type { RunnerRetentionOptions } from "./runner-retention.js"; import { dispatchQueueTask, refreshQueueTaskFromCore } from "./queue-dispatch.js"; @@ -761,7 +761,7 @@ async function route({ method, url, body, store, sourceCommit, authSummary, diag } const sessionSendMatch = path.match(/^\/api\/v1\/sessions\/([^/]+)\/send$/u); if (method === "POST" && sessionSendMatch) { - return await sendToSession({ store, sessionId: sessionSendMatch[1] ?? "", body, sourceCommit, runnerJobDefaults }) as JsonValue; + return await sendToSession({ store, sessionId: sessionSendMatch[1] ?? "", body, runnerDispatcherOptions }) as JsonValue; } const sessionReadMatch = path.match(/^\/api\/v1\/sessions\/([^/]+)\/read$/u); if (method === "POST" && sessionReadMatch) { @@ -1127,7 +1127,7 @@ async function applyNamedAipodSpec(name: string, body: unknown, dir?: string): P return await applyAipodSpec(spec, dir) as JsonValue; } -async function sendToSession(input: { store: AgentRunStore; sessionId: string; body: unknown; sourceCommit: string; runnerJobDefaults?: ManagerServerOptions["runnerJobDefaults"] }): Promise { +async function sendToSession(input: { store: AgentRunStore; sessionId: string; body: unknown; runnerDispatcherOptions: RunnerDispatcherOptions }): Promise { const startedAt = Date.now(); const record = asRecord(input.body ?? {}, "sessionSend"); const dryRun = record.dryRun === true; @@ -1136,6 +1136,12 @@ async function sendToSession(input: { store: AgentRunStore; sessionId: string; b const payload = sessionSendPayload(record); const commandIdempotencyKey = optionalString(record.commandIdempotencyKey) ?? optionalString(record.idempotencyKey); if (active) { + if (commandIdempotencyKey && active.command.idempotencyKey === commandIdempotencyKey && active.command.payloadHash === stableHash(payload)) { + const intent = await input.store.getRunnerDispatchIntent(active.command.id); + const command = intent ? { ...active.command, dispatchIntent: { id: intent.id, state: intent.state, runnerJobId: intent.runnerJobId, dispatchOutcome: intent.dispatchOutcome, actualRunnerJobId: intent.actualRunnerJobId, activeRunnerId: intent.activeRunnerId, attemptCount: intent.attemptCount, durable: true } } as CommandRecord : active.command; + const admission = intent ? runnerAdmissionSummary(input.sessionId, active.run, command, "replayed", false) : runnerAdmissionNotRequested(input.sessionId, active.run, command, "replayed"); + return sessionSendResponse({ sessionId: input.sessionId, decision: "turn", run: active.run, command, runnerJob: intent ? plannedRunnerJob(admission) : null, runnerAdmission: admission, activeBefore: active, dryRun: false }); + } const commandBody: JsonRecord = { type: "steer", payload }; if (commandIdempotencyKey) commandBody.idempotencyKey = commandIdempotencyKey; const request = { method: "POST", path: `/api/v1/runs/${active.run.id}/commands`, commandType: "steer", payloadBytes: jsonByteLength(payload), valuesPrinted: false }; @@ -1143,9 +1149,11 @@ async function sendToSession(input: { store: AgentRunStore; sessionId: string; b const command = await input.store.createCommand(active.run.id, validateCreateCommand(commandBody)); void emitAgentRunOtelSpan("command_created", active.run, process.env, { command, startTimeMs: startedAt, kind: 2, attributes: { "http.method": "POST", "http.route": "/api/v1/sessions/:sessionId/send", "http.status_code": 200, decision: "steer", commandType: command.type, commandState: command.state } }); void emitAgentRunOtelSpan("session_send", active.run, process.env, { command, startTimeMs: startedAt, kind: 2, attributes: { "http.method": "POST", "http.route": "/api/v1/sessions/:sessionId/send", "http.status_code": 200, decision: "steer", reusedActiveRun: true } }); - return sessionSendResponse({ sessionId: input.sessionId, decision: "steer", run: active.run, command, runnerJob: null, activeBefore: active, dryRun: false }); + return sessionSendResponse({ sessionId: input.sessionId, decision: "steer", run: active.run, command, runnerJob: null, runnerAdmission: null, activeBefore: active, dryRun: false }); } + const pendingAdmission = existing ? await matchingPendingSessionTurn(input.store, existing, payload, commandIdempotencyKey) : null; + const idleRun = existing ? await idleReceivableRun(input.store, existing) : null; if (idleRun) { const commandBody: JsonRecord = { type: "turn", payload }; @@ -1156,14 +1164,14 @@ async function sendToSession(input: { store: AgentRunStore; sessionId: string; b const run = await input.store.getRun(idleRun.run.id); void emitAgentRunOtelSpan("command_created", run, process.env, { command, startTimeMs: startedAt, kind: 2, attributes: { "http.method": "POST", "http.route": "/api/v1/sessions/:sessionId/send", "http.status_code": 200, decision: "turn", reusedIdleRun: true, commandType: command.type, commandState: command.state } }); void emitAgentRunOtelSpan("session_send", run, process.env, { command, startTimeMs: startedAt, kind: 2, attributes: { "http.method": "POST", "http.route": "/api/v1/sessions/:sessionId/send", "http.status_code": 200, decision: "turn", createRunnerJob: false, reusedIdleRun: true } }); - return sessionSendResponse({ sessionId: input.sessionId, decision: "turn", run, command, runnerJob: null, activeBefore: active, reusedIdleRun: idleRunSummary(idleRun), dryRun: false }); + return sessionSendResponse({ sessionId: input.sessionId, decision: "turn", run, command, runnerJob: null, runnerAdmission: null, activeBefore: active, reusedIdleRun: idleRunSummary(idleRun), dryRun: false }); } const runRecord = asRecord(record.run ?? record.runBase ?? null, "sessionSend.run"); const runBody = sessionSendRunBody(input.sessionId, runRecord); const commandBody: JsonRecord = { type: "turn", payload }; if (commandIdempotencyKey) commandBody.idempotencyKey = commandIdempotencyKey; - const runnerJobBody = record.runnerJob === undefined || record.runnerJob === null ? {} : asRecord(record.runnerJob, "sessionSend.runnerJob"); + const runnerJobBody = validateSessionRunnerJobInput(record.runnerJob === undefined || record.runnerJob === null ? {} : record.runnerJob); const createRunnerJob = record.createRunnerJob !== false; const request = { method: "POST", @@ -1176,23 +1184,47 @@ async function sendToSession(input: { store: AgentRunStore; sessionId: string; b valuesPrinted: false, }; if (dryRun) return sessionSendPlan(input.sessionId, "turn", active, request, runBody, null); - const run = await input.store.createRun(validateCreateRun(runBody)); - void emitAgentRunOtelSpan("run_created", run, process.env, { startTimeMs: startedAt, kind: 2, attributes: { "http.method": "POST", "http.route": "/api/v1/sessions/:sessionId/send", "http.status_code": 200, decision: "turn", backendProfile: run.backendProfile, providerId: run.providerId } }); - const command = await input.store.createCommand(run.id, validateCreateCommand(commandBody)); - void emitAgentRunOtelSpan("command_created", run, process.env, { command, startTimeMs: startedAt, kind: 2, attributes: { "http.method": "POST", "http.route": "/api/v1/sessions/:sessionId/send", "http.status_code": 200, decision: "turn", commandType: command.type, commandState: command.state } }); - let runnerJob: JsonValue = null; - if (createRunnerJob) { - runnerJob = await createKubernetesRunnerJob({ - store: input.store, - runId: run.id, - input: { ...runnerJobBody, commandId: command.id } as never, - defaults: runnerJobDefaultsForRequest(input.runnerJobDefaults, input.sourceCommit), - }) as unknown as JsonValue; - const runnerJobRecord = asJsonRecord(runnerJob); - void emitAgentRunOtelSpan("runner_job_created", run, process.env, { command, startTimeMs: startedAt, kind: 2, attributes: { "http.method": "POST", "http.route": "/api/v1/sessions/:sessionId/send", "http.status_code": 200, decision: "turn", jobName: stringJsonValue(runnerJobRecord?.jobName), namespace: stringJsonValue(runnerJobRecord?.namespace) } }); + if (createRunnerJob && !input.runnerDispatcherOptions.enabled) { + throw new AgentRunError("infra-failed", "durable runner dispatch is not enabled for session turn admission", { + httpStatus: 503, + details: { reason: "session-runner-dispatcher-disabled", env: "AGENTRUN_RUNNER_DISPATCHER_ENABLED", mutation: false, partialWrite: false, sessionId: input.sessionId, valuesPrinted: false }, + }); } + const requestedRun = validateCreateRun(runBody); + if (existing) assertSessionBoundary(existing, requestedRun); + const commandInput = validateCreateCommand(commandBody); + if (createRunnerJob) commandInput.dispatch = { kind: "kubernetes-job", input: pendingAdmission?.intent?.input ?? runnerJobBody }; + const admitted = await input.store.admitSessionTurn({ sessionId: input.sessionId, run: pendingAdmission ? frozenRunContract(pendingAdmission.run) : requestedRun, command: commandInput }); + const { run, command } = admitted; + if (admitted.disposition === "created") void emitAgentRunOtelSpan("run_created", run, process.env, { startTimeMs: startedAt, kind: 2, attributes: { "http.method": "POST", "http.route": "/api/v1/sessions/:sessionId/send", "http.status_code": 200, decision: "turn", backendProfile: run.backendProfile, providerId: run.providerId } }); + void emitAgentRunOtelSpan("command_created", run, process.env, { command, startTimeMs: startedAt, kind: 2, attributes: { "http.method": "POST", "http.route": "/api/v1/sessions/:sessionId/send", "http.status_code": 200, decision: "turn", commandType: command.type, commandState: command.state } }); + const runnerAdmission = createRunnerJob ? runnerAdmissionSummary(input.sessionId, run, command, admitted.disposition, admitted.recoveredPriorPartialWrite) : runnerAdmissionNotRequested(input.sessionId, run, command, admitted.disposition); + const runnerJob = createRunnerJob ? plannedRunnerJob(runnerAdmission) : null; void emitAgentRunOtelSpan("session_send", run, process.env, { command, startTimeMs: startedAt, kind: 2, attributes: { "http.method": "POST", "http.route": "/api/v1/sessions/:sessionId/send", "http.status_code": 200, decision: "turn", createRunnerJob } }); - return sessionSendResponse({ sessionId: input.sessionId, decision: "turn", run, command, runnerJob, activeBefore: active, reusedIdleRun: null, dryRun: false }); + return sessionSendResponse({ sessionId: input.sessionId, decision: "turn", run, command, runnerJob, runnerAdmission, activeBefore: active, reusedIdleRun: null, dryRun: false }); +} + +async function matchingPendingSessionTurn(store: AgentRunStore, session: SessionRecord, payload: JsonRecord, idempotencyKey: string | null): Promise<{ run: RunRecord; command: CommandRecord; intent: RunnerDispatchIntentRecord | null } | null> { + if (!session.activeRunId || !session.activeCommandId) return null; + const [run, command] = await Promise.all([store.getRun(session.activeRunId), store.getCommand(session.activeCommandId)]); + if (run.sessionRef?.sessionId !== session.sessionId || command.runId !== run.id || command.type !== "turn") return null; + if (runIsTerminal(run) || commandIsTerminal(command)) return null; + if (command.payloadHash !== stableHash(payload) || (command.idempotencyKey ?? null) !== idempotencyKey) return null; + return { run, command, intent: await store.getRunnerDispatchIntent(command.id) }; +} + +function frozenRunContract(run: RunRecord): CreateRunInput { + return { + tenantId: run.tenantId, + projectId: run.projectId, + workspaceRef: run.workspaceRef, + sessionRef: run.sessionRef, + resourceBundleRef: run.resourceBundleRef, + providerId: run.providerId, + backendProfile: run.backendProfile, + executionPolicy: run.executionPolicy, + traceSink: run.traceSink, + }; } async function activeReceivableCommand(store: AgentRunStore, session: SessionRecord): Promise<{ run: RunRecord; command: CommandRecord; reason: string; leaseExpired: boolean } | null> { @@ -1254,17 +1286,59 @@ function sessionSendPlan(sessionId: string, decision: "steer" | "turn", active: }; } -function sessionSendResponse(input: { sessionId: string; decision: "steer" | "turn"; run: RunRecord; command: CommandRecord; runnerJob: JsonValue; activeBefore: Awaited>; reusedIdleRun?: JsonRecord | null; dryRun: false }): JsonRecord { +function runnerAdmissionSummary(sessionId: string, run: RunRecord, command: CommandRecord, disposition: "created" | "replayed" | "recovered", recoveredPriorPartialWrite: boolean): JsonRecord { + const intent = asJsonRecord(command.dispatchIntent); + if (!intent || typeof intent.id !== "string" || typeof intent.runnerJobId !== "string") { + throw new AgentRunError("infra-failed", "session turn admission did not return a durable runner dispatch intent", { + httpStatus: 503, + details: { reason: "session-runner-dispatch-intent-missing", sessionId, runId: run.id, commandId: command.id, mutation: false, partialWrite: false, valuesPrinted: false }, + }); + } + const mutation = disposition !== "replayed"; + return { + mode: "durable-dispatch-intent", + state: intent.state, + reason: disposition === "recovered" ? "session-runner-admission-recovered" : disposition === "replayed" ? "session-runner-admission-replayed" : "session-runner-admission-created", + failureKind: null, + sessionId, + runId: run.id, + commandId: command.id, + dispatchIntentId: intent.id, + plannedRunnerJobId: intent.runnerJobId, + attemptCount: intent.attemptCount ?? 0, + durable: true, + mutation, + partialWrite: false, + recoveredPriorPartialWrite, + recoveryActions: [ + managerActionDescriptor({ action: "inspect-session", operation: "describe", resourceKind: "session", resourceName: sessionId, sessionId, runId: run.id, commandId: command.id }), + managerActionDescriptor({ action: "poll-events", operation: "events", resourceKind: "run", resourceName: run.id, sessionId, runId: run.id, commandId: command.id, afterSeq: 0, limit: 100 }), + ], + valuesPrinted: false, + }; +} + +function runnerAdmissionNotRequested(sessionId: string, run: RunRecord, command: CommandRecord, disposition: "created" | "replayed" | "recovered"): JsonRecord { + return { mode: "none", state: "not-requested", reason: "session-runner-dispatch-not-requested", failureKind: null, sessionId, runId: run.id, commandId: command.id, dispatchIntentId: null, plannedRunnerJobId: null, durable: false, mutation: disposition !== "replayed", partialWrite: false, recoveredPriorPartialWrite: false, recoveryActions: [], valuesPrinted: false }; +} + +function plannedRunnerJob(admission: JsonRecord): JsonRecord { + return { action: "runner-dispatch-admitted", state: admission.state, runId: admission.runId, commandId: admission.commandId, runnerJobId: admission.plannedRunnerJobId, dispatchIntentId: admission.dispatchIntentId, mutation: false, durable: true, valuesPrinted: false }; +} + +function sessionSendResponse(input: { sessionId: string; decision: "steer" | "turn"; run: RunRecord; command: CommandRecord; runnerJob: JsonValue; runnerAdmission: JsonRecord | null; activeBefore: Awaited>; reusedIdleRun?: JsonRecord | null; dryRun: false }): JsonRecord { return { action: "session-send", dryRun: input.dryRun, - mutation: true, + mutation: input.runnerAdmission?.mutation === false && input.decision === "turn" ? false : true, + partialWrite: false, sessionId: input.sessionId, decision: input.decision, internalCommandType: input.command.type, run: input.run as unknown as JsonRecord, command: input.command as unknown as JsonRecord, runnerJob: input.runnerJob, + runnerAdmission: input.runnerAdmission, activeBefore: input.activeBefore ? activeBeforeSummary(input.activeBefore) : null, reusedIdleRun: input.reusedIdleRun ?? null, pollActions: [ diff --git a/src/mgr/store.ts b/src/mgr/store.ts index 915b572..84dee46 100644 --- a/src/mgr/store.ts +++ b/src/mgr/store.ts @@ -47,6 +47,20 @@ export interface RunnerRetentionFenceInput { reason: string; } +export interface SessionTurnAdmissionInput { + sessionId: string; + run: CreateRunInput; + command: CreateCommandInput; +} + +export interface SessionTurnAdmission extends JsonRecord { + run: RunRecord; + command: CommandRecord; + disposition: "created" | "replayed" | "recovered"; + recoveredPriorPartialWrite: boolean; + partialWrite: false; +} + export interface AgentRunStore { health(): MaybePromise; createRun(input: CreateRunInput, identity?: CreateRunIdentity): MaybePromise; @@ -54,6 +68,7 @@ export interface AgentRunStore { listEvents(runId: string, afterSeq: number, limit: number): MaybePromise; listEventsForCommand(runId: string, commandId: string, limit: number): MaybePromise; createCommand(runId: string, input: CreateCommandInput): MaybePromise; + admitSessionTurn(input: SessionTurnAdmissionInput): MaybePromise; getCommand(commandId: string): MaybePromise; listCommands(runId: string, afterSeq: number, limit: number): MaybePromise; getRunnerDispatchIntent(commandId: string): MaybePromise; @@ -320,6 +335,83 @@ export class MemoryAgentRunStore implements AgentRunStore { return attachRunnerDispatchIntent(command, intent); } + admitSessionTurn(input: SessionTurnAdmissionInput): SessionTurnAdmission { + assertSessionTurnAdmissionContract(input); + const session = this.sessions.get(input.sessionId) ?? null; + const replay = input.command.idempotencyKey + ? Array.from(this.commands.values()) + .filter((command) => command.idempotencyKey === input.command.idempotencyKey) + .map((command) => ({ command, run: this.runs.get(command.runId) ?? null })) + .find((item) => item.run?.sessionRef?.sessionId === input.sessionId) ?? null + : null; + if (replay?.run) return this.replayOrRecoverSessionTurn(replay.run, replay.command, input, "replayed"); + + if (session?.activeRunId || session?.activeCommandId) { + if (!session.activeRunId || !session.activeCommandId) throw invalidSessionAdmissionProjection(input.sessionId); + const run = this.runs.get(session.activeRunId); + const command = this.commands.get(session.activeCommandId); + if (!run || !command || command.runId !== run.id || run.sessionRef?.sessionId !== input.sessionId) throw invalidSessionAdmissionProjection(input.sessionId); + if (!isTerminalRunStatus(run.status) && !isTerminalCommandState(command.state)) { + try { + return this.replayOrRecoverSessionTurn(run, command, input, "replayed"); + } catch (error) { + if (error instanceof AgentRunError && error.failureKind === "schema-invalid") throw activeSessionAdmissionConflict(input.sessionId, run, command); + throw error; + } + } + } + + const sessionBefore = session ? { ...session } : null; + const sessionVersionBefore = this.sessionVersion; + const kafkaOutboxSeqBefore = this.kafkaOutboxSeq; + let run: RunRecord | null = null; + try { + run = this.createRun(input.run); + const command = this.createCommand(run.id, input.command); + return { run, command, disposition: "created", recoveredPriorPartialWrite: false, partialWrite: false }; + } catch (error) { + if (run) { + this.runs.delete(run.id); + this.eventsByRun.delete(run.id); + for (const [commandId, command] of this.commands) if (command.runId === run.id) this.commands.delete(commandId); + for (const [intentId, intent] of this.runnerDispatchIntents) if (intent.runId === run.id) this.runnerDispatchIntents.delete(intentId); + for (const [outboxId, item] of this.kafkaEventOutbox) if (item.runId === run.id) this.kafkaEventOutbox.delete(outboxId); + } + if (sessionBefore) this.sessions.set(input.sessionId, sessionBefore); + else this.sessions.delete(input.sessionId); + this.sessionVersion = sessionVersionBefore; + this.kafkaOutboxSeq = kafkaOutboxSeqBefore; + throw error; + } + } + + private replayOrRecoverSessionTurn(run: RunRecord, command: CommandRecord, input: SessionTurnAdmissionInput, disposition: "replayed" | "recovered"): SessionTurnAdmission { + assertRunCreateReplay(run, input.run); + assertSessionCommandReplay(command, input.command); + const intent = this.getRunnerDispatchIntent(command.id); + if (intent) { + assertRunnerDispatchReplay(intent, command, input.command.dispatch); + return { run, command: attachRunnerDispatchIntent(command, intent), disposition: "replayed", recoveredPriorPartialWrite: false, partialWrite: false }; + } + if (!input.command.dispatch) { + return { run, command: attachRunnerDispatchIntent(command, null), disposition, recoveredPriorPartialWrite: false, partialWrite: false }; + } + if (run.status !== "pending" || run.claimedBy || command.state !== "pending") throw activeSessionAdmissionConflict(input.sessionId, run, command); + const at = nowIso(); + const recovered = newRunnerDispatchIntent(command, input.command.dispatch, at); + const event = this.prepareEvent(run.id, "backend_status", { + phase: "runner-dispatch-recovered", + reason: "session-runner-admission-partial-write-recovered", + commandId: command.id, + dispatchIntentId: recovered.id, + plannedRunnerJobId: recovered.runnerJobId, + valuesPrinted: false, + }); + this.runnerDispatchIntents.set(recovered.id, recovered); + this.commitPreparedEvent(event); + return { run, command: attachRunnerDispatchIntent(command, recovered), disposition: "recovered", recoveredPriorPartialWrite: true, partialWrite: false }; + } + getRunnerDispatchIntent(commandId: string): RunnerDispatchIntentRecord | null { return Array.from(this.runnerDispatchIntents.values()).find((intent) => intent.commandId === commandId) ?? null; } @@ -370,6 +462,7 @@ export class MemoryAgentRunStore implements AgentRunStore { const at = nowIso(); const failure = redactJson(error); const message = typeof failure.message === "string" ? failure.message : "runner dispatch failed"; + const reason = typeof failure.reason === "string" ? failure.reason : "runner-dispatch-attempt-failed"; const updatedIntent: RunnerDispatchIntentRecord = { ...intent, state: "failed", leaseOwner: null, leaseExpiresAt: null, nextAttemptAt: at, lastError: failure, updatedAt: at }; this.runnerDispatchIntents.set(intent.id, updatedIntent); const command = this.getCommand(intent.commandId); @@ -379,7 +472,7 @@ export class MemoryAgentRunStore implements AgentRunStore { const updatedRun = isTerminalRunStatus(run.status) ? run : this.updateRun(run.id, { status: "failed", terminalStatus: "failed", failureKind: "infra-failed", failureMessage: message }); if (!isTerminalRunStatus(run.status)) { this.cancelRunnerDispatchIntentsForRun(run.id, "run terminalized by runner dispatch failure", at); - this.appendEvent(run.id, "backend_status", { phase: "runner-dispatch-failed", commandId: command.id, dispatchIntentId: intent.id, attemptCount: intent.attemptCount, failureKind: "infra-failed", message, valuesPrinted: false }); + this.appendEvent(run.id, "backend_status", { phase: "runner-dispatch-failed", reason, commandId: command.id, dispatchIntentId: intent.id, attemptCount: intent.attemptCount, failureKind: "infra-failed", message, valuesPrinted: false }); this.appendEvent(run.id, "backend_status", { phase: "command-terminal", commandId: command.id, state: "failed", terminalStatus: "failed", failureKind: "infra-failed", message, threadId: null, turnId: null }); this.appendEvent(run.id, "terminal_status", { terminalStatus: "failed", failureKind: "infra-failed", message }); this.touchSessionForRun(this.getRun(run.id), { executionState: "terminal", activeRunId: null, activeCommandId: null, lastRunId: run.id, lastCommandId: command.id, terminalStatus: "failed", failureKind: "infra-failed", lastActivityAt: at }, { bumpVersion: true, at }); @@ -1484,6 +1577,48 @@ export function assertRunCreateReplay(existing: RunRecord, input: CreateRunInput } } +export function assertSessionTurnAdmissionContract(input: SessionTurnAdmissionInput): void { + if (!input.sessionId.trim()) throw new AgentRunError("schema-invalid", "session turn admission requires a non-empty sessionId", { httpStatus: 400 }); + if (input.run.sessionRef?.sessionId !== input.sessionId) { + throw new AgentRunError("schema-invalid", "session turn admission run must reference the target session", { + httpStatus: 400, + details: { sessionId: input.sessionId, runSessionId: input.run.sessionRef?.sessionId ?? null, valuesPrinted: false }, + }); + } + if (input.command.type !== "turn") throw new AgentRunError("schema-invalid", "session turn admission only accepts turn commands", { httpStatus: 400 }); +} + +export function assertSessionCommandReplay(existing: CommandRecord, input: CreateCommandInput): void { + if (existing.type !== input.type || existing.payloadHash !== stableHash(input.payload) || (existing.idempotencyKey ?? null) !== (input.idempotencyKey ?? null)) { + throw new AgentRunError("schema-invalid", "session turn admission replay does not match the existing command", { + httpStatus: 409, + details: { runId: existing.runId, commandId: existing.id, reason: "session-turn-admission-replay-conflict", valuesPrinted: false }, + }); + } +} + +export function activeSessionAdmissionConflict(sessionId: string, run: RunRecord, command: CommandRecord): AgentRunError { + return new AgentRunError("runner-lease-conflict", `session ${sessionId} already has an active turn admission`, { + httpStatus: 409, + details: { + reason: "session-turn-admission-active", + sessionId, + runId: run.id, + commandId: command.id, + runStatus: run.status, + commandState: command.state, + valuesPrinted: false, + }, + }); +} + +export function invalidSessionAdmissionProjection(sessionId: string): AgentRunError { + return new AgentRunError("infra-failed", `session ${sessionId} active admission projection is inconsistent`, { + httpStatus: 503, + details: { reason: "session-turn-admission-projection-inconsistent", sessionId, valuesPrinted: false }, + }); +} + export function assertQueueTaskPayloadHash(task: QueueTaskRecord): void { if (queueTaskPayloadHash(task) !== task.payloadHash) throw new AgentRunError("schema-invalid", `queue task ${task.id} payloadHash does not match its immutable contract`, { httpStatus: 409 }); } diff --git a/src/selftest/cases/66-session-turn-admission.ts b/src/selftest/cases/66-session-turn-admission.ts new file mode 100644 index 0000000..b55131a --- /dev/null +++ b/src/selftest/cases/66-session-turn-admission.ts @@ -0,0 +1,379 @@ +import assert from "node:assert/strict"; +import { chmod } from "node:fs/promises"; +import path from "node:path"; +import { AgentRunError } from "../../common/errors.js"; +import type { CommandRecord, CreateCommandInput, CreateRunInput, JsonRecord, RunRecord } from "../../common/types.js"; +import { validateCreateCommand, validateCreateRun } from "../../common/validation.js"; +import { ManagerClient } from "../../mgr/client.js"; +import type { RunnerJobDefaults } from "../../mgr/kubernetes-runner-job.js"; +import { dispatchRunnerIntentsOnce, type RunnerDispatcherOptions } from "../../mgr/runner-dispatcher.js"; +import { startManagerServer } from "../../mgr/server.js"; +import { MemoryAgentRunStore, type SessionTurnAdmissionInput } from "../../mgr/store.js"; +import type { SelfTestCase } from "../harness.js"; + +const image = "127.0.0.1:5000/agentrun/agentrun-mgr@sha256:1111111111111111111111111111111111111111111111111111111111111111"; + +const selfTest: SelfTestCase = async (context) => { + const fakeKubectl = path.join(context.root, "src/selftest/fixtures/fake-kubectl-session-admission.mjs"); + await chmod(fakeKubectl, 0o755); + const previousMode = process.env.AGENTRUN_SELFTEST_KUBECTL_MODE; + try { + await assertHttpAdmissionDoesNotCallKubectl(fakeKubectl); + await assertDisabledDispatcherFailsBeforeFormalFacts(fakeKubectl); + assertMemoryAdmissionRollback(); + assertLegacyHalfCommitRecovery(); + await assertRestartVersionForwardRecovery(fakeKubectl); + assertMemoryReplayAndConflict(); + await assertKubernetesObservationFailureTerminalizes(fakeKubectl); + await assertKubernetesCreateFailureTerminalizes(fakeKubectl); + await assertSuccessfulSameSessionContinuity(fakeKubectl); + } finally { + restoreEnv("AGENTRUN_SELFTEST_KUBECTL_MODE", previousMode); + } + return { + name: "session-turn-admission", + tests: [ + "session-send-durable-admission-no-sync-kubectl", + "session-send-dispatcher-disabled-prewrite-failure", + "memory-session-turn-admission-rollback", + "legacy-half-commit-recovered-in-place", + "restart-version-forward-recovery-keeps-frozen-run-contract", + "session-turn-idempotent-replay-and-active-conflict", + "kubectl-observation-failure-terminalizes-and-releases-session", + "kubectl-create-failure-terminalizes-and-releases-session", + "successful-same-session-continuity", + ], + }; +}; + +async function assertHttpAdmissionDoesNotCallKubectl(kubectlCommand: string): Promise { + process.env.AGENTRUN_SELFTEST_KUBECTL_MODE = "create-fail"; + const store = new MemoryAgentRunStore(); + const server = await startManagerServer({ + store, + sourceCommit: "selftest", + runnerJobDefaults: defaults(kubectlCommand), + runnerDispatcherOptions: { ...dispatcherOptions(3), intervalMs: 60_000 }, + kafkaOutboxRelayOptions: { enabled: false }, + }); + try { + await delay(20); + const client = new ManagerClient(server.baseUrl); + const sessionId = "ses_session_admission_http"; + const response = await client.post(`/api/v1/sessions/${sessionId}/send`, sessionSendBody(sessionId, "atomic HTTP admission", "session-admission-http")) as JsonRecord; + const run = response.run as RunRecord; + const command = response.command as CommandRecord; + const admission = response.runnerAdmission as JsonRecord; + assert.equal(response.decision, "turn"); + assert.equal(response.partialWrite, false); + assert.equal(admission.mode, "durable-dispatch-intent"); + assert.equal(admission.state, "pending"); + assert.equal(admission.durable, true); + assert.equal(admission.runId, run.id); + assert.equal(admission.commandId, command.id); + assert.equal(command.dispatchIntent?.state, "pending"); + assert.equal(store.getRunnerDispatchIntent(command.id)?.state, "pending"); + assert.deepEqual(store.listEvents(run.id, 0, 100).map((event) => event.type === "backend_status" ? event.payload.phase : event.type), ["user_message", "run-created", "command-created"]); + } finally { + await closeServer(server.server); + } +} + +async function assertDisabledDispatcherFailsBeforeFormalFacts(kubectlCommand: string): Promise { + const store = new MemoryAgentRunStore(); + const server = await startManagerServer({ store, sourceCommit: "selftest", runnerJobDefaults: defaults(kubectlCommand), runnerDispatcherOptions: { enabled: false }, kafkaOutboxRelayOptions: { enabled: false } }); + try { + const client = new ManagerClient(server.baseUrl); + const sessionId = "ses_session_admission_disabled"; + await assert.rejects( + () => client.post(`/api/v1/sessions/${sessionId}/send`, sessionSendBody(sessionId, "must fail before write", "session-admission-disabled")), + (error) => error instanceof AgentRunError + && error.httpStatus === 503 + && JSON.stringify(error.details).includes("session-runner-dispatcher-disabled") + && JSON.stringify(error.details).includes("partialWrite"), + ); + assert.equal(store.getSession(sessionId), null); + assert.equal(store.runnerDispatchStatus().backlogCount, 0); + } finally { + await closeServer(server.server); + } +} + +function assertMemoryAdmissionRollback(): void { + const store = new FailingSessionCommandStore(); + const sessionId = "ses_session_admission_rollback"; + assert.throws(() => store.admitSessionTurn(admissionInput(sessionId, "rollback all formal facts", "session-admission-rollback")), /simulated command stage failure/u); + assert.equal(store.getSession(sessionId), null); + assert.equal(store.runnerDispatchStatus().backlogCount, 0); + assert.equal(store.kafkaEventOutboxStatus().backlogCount, 0); + assert.match(store.createdRunId ?? "", /^run_[a-f0-9]{32}$/u); + assert.throws(() => store.getRun(store.createdRunId ?? ""), /was not found/u); +} + +function assertLegacyHalfCommitRecovery(): void { + const store = new MemoryAgentRunStore(); + const sessionId = "ses_session_admission_legacy"; + const input = admissionInput(sessionId, "recover legacy half commit", "session-admission-legacy"); + const legacyRun = store.createRun(input.run); + const legacyCommand = store.createCommand(legacyRun.id, withoutDispatch(input.command)); + const beforeEvents = store.listEvents(legacyRun.id, 0, 100); + assert.equal(store.getRunnerDispatchIntent(legacyCommand.id), null); + + const recovered = store.admitSessionTurn(input); + assert.equal(recovered.disposition, "recovered"); + assert.equal(recovered.recoveredPriorPartialWrite, true); + assert.equal(recovered.partialWrite, false); + assert.equal(recovered.run.id, legacyRun.id); + assert.equal(recovered.command.id, legacyCommand.id); + assert.equal(recovered.command.dispatchIntent?.state, "pending"); + const events = store.listEvents(legacyRun.id, 0, 100); + assert.equal(events.length, beforeEvents.length + 1); + assert.equal(events.filter((event) => event.type === "user_message").length, 1); + assert.equal(events.filter((event) => event.payload.phase === "run-created").length, 1); + assert.equal(events.filter((event) => event.payload.phase === "command-created").length, 1); + assert.equal(events.filter((event) => event.payload.phase === "runner-dispatch-recovered").length, 1); + + const replayed = store.admitSessionTurn(input); + assert.equal(replayed.disposition, "replayed"); + assert.equal(replayed.run.id, legacyRun.id); + assert.equal(replayed.command.id, legacyCommand.id); + assert.equal(store.listEvents(legacyRun.id, 0, 100).length, events.length); +} + +async function assertRestartVersionForwardRecovery(kubectlCommand: string): Promise { + process.env.AGENTRUN_SELFTEST_KUBECTL_MODE = "create-fail"; + const store = new MemoryAgentRunStore(); + const sessionId = "ses_session_admission_version_forward"; + const prompt = "recover after manager and source version advance"; + const idempotencyKey = "session-admission-version-forward"; + const frozenInput = admissionInput(sessionId, prompt, idempotencyKey, "1111111111111111111111111111111111111111"); + const legacyRun = store.createRun(frozenInput.run); + const legacyCommand = store.createCommand(legacyRun.id, withoutDispatch(frozenInput.command)); + const advancedRun = validateCreateRun(runInput(sessionId, "2222222222222222222222222222222222222222")); + assert.throws( + () => store.admitSessionTurn({ ...frozenInput, run: advancedRun }), + /reused with a different creation contract/u, + ); + + const server = await startManagerServer({ + store, + sourceCommit: "manager-version-after-half-commit", + runnerJobDefaults: defaults(kubectlCommand), + runnerDispatcherOptions: { ...dispatcherOptions(3), intervalMs: 60_000 }, + kafkaOutboxRelayOptions: { enabled: false }, + }); + try { + await delay(20); + const client = new ManagerClient(server.baseUrl); + const response = await client.post( + `/api/v1/sessions/${sessionId}/send`, + sessionSendBody(sessionId, prompt, idempotencyKey, advancedRun), + ) as JsonRecord; + const recoveredRun = response.run as RunRecord; + const recoveredCommand = response.command as CommandRecord; + const admission = response.runnerAdmission as JsonRecord; + assert.equal(recoveredRun.id, legacyRun.id); + assert.equal(recoveredCommand.id, legacyCommand.id); + assert.equal((recoveredRun.resourceBundleRef as JsonRecord | null)?.commitId, "1111111111111111111111111111111111111111"); + assert.equal((store.getRun(legacyRun.id).resourceBundleRef as JsonRecord | null)?.commitId, "1111111111111111111111111111111111111111"); + assert.equal(admission.reason, "session-runner-admission-recovered"); + assert.equal(admission.recoveredPriorPartialWrite, true); + const events = store.listEvents(legacyRun.id, 0, 100); + assert.equal(events.filter((event) => event.type === "user_message").length, 1); + assert.equal(events.filter((event) => event.payload.phase === "run-created").length, 1); + assert.equal(events.filter((event) => event.payload.phase === "command-created").length, 1); + assert.equal(events.filter((event) => event.payload.phase === "runner-dispatch-recovered").length, 1); + } finally { + await closeServer(server.server); + } +} + +function assertMemoryReplayAndConflict(): void { + const store = new MemoryAgentRunStore(); + const sessionId = "ses_session_admission_replay"; + const input = admissionInput(sessionId, "same key is one admission", "session-admission-replay"); + const created = store.admitSessionTurn(input); + const replayed = store.admitSessionTurn(input); + assert.equal(created.disposition, "created"); + assert.equal(replayed.disposition, "replayed"); + assert.equal(replayed.run.id, created.run.id); + assert.equal(replayed.command.id, created.command.id); + assert.equal(store.listCommands(created.run.id, 0, 100).length, 1); + assert.equal(store.listEvents(created.run.id, 0, 100).length, 3); + assert.throws( + () => store.admitSessionTurn(admissionInput(sessionId, "different active admission", "session-admission-conflict")), + (error) => error instanceof AgentRunError + && error.failureKind === "runner-lease-conflict" + && error.details?.reason === "session-turn-admission-active", + ); +} + +async function assertKubernetesObservationFailureTerminalizes(kubectlCommand: string): Promise { + process.env.AGENTRUN_SELFTEST_KUBECTL_MODE = "list-fail"; + const store = new MemoryAgentRunStore(); + const sessionId = "ses_session_admission_list_failure"; + const admitted = store.admitSessionTurn(admissionInput(sessionId, "terminalize list failure", "session-admission-list-failure")); + const result = await dispatchRunnerIntentsOnce({ store, defaults: defaults(kubectlCommand, true), options: dispatcherOptions(1) }); + assert.equal(result.failedCount, 1, JSON.stringify({ result, intent: store.getRunnerDispatchIntent(admitted.command.id) })); + assertTerminalizedAdmission(store, admitted.run.id, admitted.command.id, sessionId, "runner-kubernetes-observation-failed"); +} + +async function assertKubernetesCreateFailureTerminalizes(kubectlCommand: string): Promise { + process.env.AGENTRUN_SELFTEST_KUBECTL_MODE = "create-fail"; + const store = new MemoryAgentRunStore(); + const sessionId = "ses_session_admission_create_failure"; + const admitted = store.admitSessionTurn(admissionInput(sessionId, "terminalize create failure", "session-admission-create-failure")); + const result = await dispatchRunnerIntentsOnce({ store, defaults: defaults(kubectlCommand), options: dispatcherOptions(1) }); + assert.equal(result.failedCount, 1, JSON.stringify({ result, intent: store.getRunnerDispatchIntent(admitted.command.id) })); + assertTerminalizedAdmission(store, admitted.run.id, admitted.command.id, sessionId, "runner-kubernetes-job-create-failed"); +} + +async function assertSuccessfulSameSessionContinuity(kubectlCommand: string): Promise { + process.env.AGENTRUN_SELFTEST_KUBECTL_MODE = "success"; + const store = new MemoryAgentRunStore(); + const sessionId = "ses_session_admission_continuity"; + const first = store.admitSessionTurn(admissionInput(sessionId, "first successful admission", "session-admission-continuity-1")); + const dispatched = await dispatchRunnerIntentsOnce({ store, defaults: defaults(kubectlCommand), options: dispatcherOptions(1) }); + assert.equal(dispatched.dispatchedCount, 1, JSON.stringify(dispatched)); + assert.equal(store.getRunnerDispatchIntent(first.command.id)?.state, "dispatched"); + store.finishCommand(first.command.id, { terminalStatus: "completed", failureKind: null, failureMessage: null }); + store.finishRun(first.run.id, { terminalStatus: "completed", failureKind: null, failureMessage: null }); + + const second = store.admitSessionTurn(admissionInput(sessionId, "second successful admission", "session-admission-continuity-2")); + assert.equal(second.disposition, "created"); + assert.notEqual(second.run.id, first.run.id); + assert.notEqual(second.command.id, first.command.id); + const session = store.getSession(sessionId); + assert.equal(session?.activeRunId, second.run.id); + assert.equal(session?.activeCommandId, second.command.id); + assert.equal(session?.lastRunId, second.run.id); + assert.equal(session?.lastCommandId, second.command.id); +} + +function assertTerminalizedAdmission(store: MemoryAgentRunStore, runId: string, commandId: string, sessionId: string, reason: string): void { + const intent = store.getRunnerDispatchIntent(commandId); + assert.equal(intent?.state, "failed"); + assert.equal(intent?.lastError?.reason, reason); + assert.equal(intent?.lastError?.valuesPrinted, false); + assert.equal(store.getCommand(commandId).state, "failed"); + assert.equal(store.getRun(runId).status, "failed"); + const session = store.getSession(sessionId); + assert.equal(session?.activeRunId, null); + assert.equal(session?.activeCommandId, null); + assert.equal(session?.executionState, "terminal"); + assert.equal(store.listEvents(runId, 0, 100).find((event) => event.payload.phase === "runner-dispatch-failed")?.payload.reason, reason); + assert.equal(store.listEvents(runId, 0, 100).filter((event) => event.type === "terminal_status").length, 1); +} + +function admissionInput(sessionId: string, prompt: string, idempotencyKey: string, resourceCommit?: string): SessionTurnAdmissionInput { + return { + sessionId, + run: validateCreateRun(runInput(sessionId, resourceCommit)), + command: validateCreateCommand({ + type: "turn", + payload: { prompt }, + idempotencyKey, + dispatch: { kind: "kubernetes-job", input: {} }, + }), + }; +} + +function withoutDispatch(input: CreateCommandInput): CreateCommandInput { + const { dispatch: _dispatch, ...command } = input; + return command; +} + +function sessionSendBody(sessionId: string, prompt: string, idempotencyKey: string, run: CreateRunInput = runInput(sessionId)): JsonRecord { + return { prompt, commandIdempotencyKey: idempotencyKey, createRunnerJob: true, run }; +} + +function runInput(sessionId: string, resourceCommit?: string): CreateRunInput { + return { + tenantId: "unidesk", + projectId: "pikasTech/agentrun", + workspaceRef: resourceCommit ? { kind: "opaque", path: "." } : { kind: "host-path", path: "/tmp/agentrun-session-admission" }, + sessionRef: { sessionId, metadata: { title: "session admission selftest" } }, + resourceBundleRef: resourceCommit ? { + kind: "gitbundle", + repoUrl: "git@github.com:pikasTech/unidesk.git", + commitId: resourceCommit, + bundles: [{ name: "unidesk-skills", subpath: ".agents/skills", targetPath: ".agents/skills" }], + submodules: false, + lfs: false, + } : null, + providerId: "NC01", + backendProfile: "codex", + executionPolicy: { + sandbox: "workspace-write", + approval: "never", + timeoutMs: 60_000, + network: "default", + secretScope: { + allowCredentialEcho: false, + providerCredentials: [{ profile: "codex", secretRef: { name: "agentrun-v02-provider-codex", keys: ["auth.json", "config.toml"] } }], + }, + }, + traceSink: null, + }; +} + +function defaults(kubectlCommand: string, withRetention = false): RunnerJobDefaults { + return { + namespace: "agentrun-v02", + managerUrl: "http://agentrun-mgr.agentrun-v02.svc.cluster.local:8080", + runnerApiKeySecretRef: { name: "agentrun-v02-api-key", key: "HWLAB_API_KEY" }, + image, + sourceCommit: "selftest", + serviceAccountName: "agentrun-v02-runner", + jobNamePrefix: "agentrun-v02-runner", + lane: "v0.2", + kubectlCommand, + ...(withRetention ? { + retention: { + namespace: "agentrun-v02", + maxRunners: 20, + cleanupOrder: "oldest-inactive-last-active-first" as const, + activeHeartbeatMaxAgeMs: 900_000, + stalePendingMaxAgeMs: 900_000, + matchLabels: { "app.kubernetes.io/part-of": "agentrun", "app.kubernetes.io/name": "agentrun-runner", "app.kubernetes.io/component": "runner" }, + jobNamePrefixes: ["agentrun-v02-runner"], + ageBasedCleanup: { enabled: false, maxAgeHours: null }, + kubectlCommand, + }, + } : {}), + }; +} + +function dispatcherOptions(maxAttempts: number): RunnerDispatcherOptions { + return { enabled: true, intervalMs: 1_000, batchSize: 10, leaseMs: 60_000, maxAttempts, retryBackoffMs: 1_000, attemptTimeoutMs: 30_000, owner: "session-admission-selftest" }; +} + +class FailingSessionCommandStore extends MemoryAgentRunStore { + createdRunId: string | null = null; + + override createRun(input: CreateRunInput): RunRecord { + const run = super.createRun(input); + this.createdRunId = run.id; + return run; + } + + override createCommand(runId: string, input: CreateCommandInput): CommandRecord { + super.createCommand(runId, input); + throw new Error("simulated command stage failure"); + } +} + +async function closeServer(server: import("node:http").Server): Promise { + await new Promise((resolve) => server.close(() => resolve())); +} + +async function delay(ms: number): Promise { + await new Promise((resolve) => setTimeout(resolve, ms)); +} + +function restoreEnv(key: string, value: string | undefined): void { + if (value === undefined) delete process.env[key]; + else process.env[key] = value; +} + +export default selfTest; diff --git a/src/selftest/fixtures/fake-kubectl-session-admission.mjs b/src/selftest/fixtures/fake-kubectl-session-admission.mjs new file mode 100755 index 0000000..deae570 --- /dev/null +++ b/src/selftest/fixtures/fake-kubectl-session-admission.mjs @@ -0,0 +1,50 @@ +#!/usr/bin/env bun + +const args = process.argv.slice(2); +const mode = process.env.AGENTRUN_SELFTEST_KUBECTL_MODE ?? "success"; + +if (args[0] === "get") { + const resource = args[1] ?? ""; + if (mode === "list-fail" && (resource === "jobs" || resource === "pods" || resource === "events")) { + console.error(`simulated kubectl get ${resource} outage`); + process.exit(1); + } + if (resource === "pvc") { + const name = args[2] ?? "session-pvc"; + process.stdout.write(JSON.stringify({ + apiVersion: "v1", + kind: "PersistentVolumeClaim", + metadata: { name, namespace: namespaceFromArgs(args), uid: "selftest-pvc-uid", resourceVersion: "1" }, + status: { phase: "Bound", capacity: { storage: "1Gi" } }, + })); + process.exit(0); + } + if (resource === "jobs" || resource === "pods" || resource === "events") { + process.stdout.write(JSON.stringify({ apiVersion: "v1", items: [] })); + process.exit(0); + } + console.error(`unsupported fake kubectl get resource: ${resource}`); + process.exit(2); +} + +if (args[0] === "create") { + const chunks = []; + for await (const chunk of process.stdin) chunks.push(Buffer.from(chunk)); + const object = JSON.parse(Buffer.concat(chunks).toString("utf8")); + if (mode === "create-fail" && object.kind === "Job") { + console.error("simulated kubectl create Job outage"); + process.exit(1); + } + object.metadata = { ...object.metadata, uid: `selftest-${String(object.kind).toLowerCase()}-uid`, resourceVersion: "1" }; + if (object.kind === "PersistentVolumeClaim") object.status = { phase: "Bound", capacity: { storage: "1Gi" } }; + process.stdout.write(JSON.stringify(object)); + process.exit(0); +} + +console.error(`unsupported fake kubectl args: ${args.join(" ")}`); +process.exit(2); + +function namespaceFromArgs(values) { + const index = values.findIndex((value) => value === "-n" || value === "--namespace"); + return index >= 0 ? values[index + 1] ?? "agentrun-v02" : "agentrun-v02"; +} diff --git a/src/selftest/integration/postgres-session-turn-admission.ts b/src/selftest/integration/postgres-session-turn-admission.ts new file mode 100644 index 0000000..51c7065 --- /dev/null +++ b/src/selftest/integration/postgres-session-turn-admission.ts @@ -0,0 +1,198 @@ +import assert from "node:assert/strict"; +import { Pool } from "pg"; +import { AgentRunError } from "../../common/errors.js"; +import type { CreateRunInput } from "../../common/types.js"; +import { validateCreateCommand, validateCreateRun } from "../../common/validation.js"; +import { createPostgresAgentRunStore } from "../../mgr/postgres-store.js"; +import type { SessionTurnAdmissionInput } from "../../mgr/store.js"; + +const connectionString = requiredDatabaseUrl(); + +const suffix = `${Date.now()}_${process.pid}`; +const concurrentSessionId = `ses_pg_admission_${suffix}`; +const legacySessionId = `ses_pg_admission_legacy_${suffix}`; +const rollbackSessionId = `ses_pg_admission_rollback_${suffix}`; +const eventOutbox = { enabled: true, topic: "agentrun.event.v1", source: "session-admission-selftest" } as const; +const primary = await createPostgresAgentRunStore({ connectionString, poolMax: 4, eventOutbox }); +const concurrent = await createPostgresAgentRunStore({ connectionString, poolMax: 4, eventOutbox }); +const control = new Pool({ connectionString, application_name: "agentrun-session-admission-selftest", max: 2 }); +const reopenedStores: Array>> = []; + +try { + await assertConcurrentAdmissionAndRestartRecovery(); + await assertLegacyHalfCommitRecoveryAfterRestart(); + await assertAdmissionTransactionRollback(); + console.log(JSON.stringify({ + ok: true, + tests: [ + "postgres-session-turn-concurrent-idempotency", + "postgres-session-turn-restart-durable-dispatch-intent", + "postgres-session-turn-active-conflict", + "postgres-session-turn-same-session-continuity", + "postgres-legacy-half-commit-recovery-after-restart", + "postgres-session-turn-transaction-rollback", + ], + valuesPrinted: false, + })); +} finally { + await dropRollbackTrigger().catch(() => undefined); + await cleanupSession(concurrentSessionId).catch(() => undefined); + await cleanupSession(legacySessionId).catch(() => undefined); + await cleanupSession(rollbackSessionId).catch(() => undefined); + await Promise.all(reopenedStores.map(async (store) => await store.close())); + await Promise.all([primary.close(), concurrent.close(), control.end()]); +} + +async function assertConcurrentAdmissionAndRestartRecovery(): Promise { + const input = admissionInput(concurrentSessionId, "postgres concurrent session admission", `pg-admission-${suffix}`); + const [first, second] = await Promise.all([primary.admitSessionTurn(input), concurrent.admitSessionTurn(input)]); + assert.equal(first.run.id, second.run.id); + assert.equal(first.command.id, second.command.id); + assert.deepEqual(new Set([first.disposition, second.disposition]), new Set(["created", "replayed"])); + assert.equal((await primary.listCommands(first.run.id, 0, 100)).length, 1); + assert.deepEqual((await primary.listEvents(first.run.id, 0, 100)).map((event) => event.type === "backend_status" ? event.payload.phase : event.type), ["user_message", "run-created", "command-created"]); + const counts = await control.query( + `SELECT + (SELECT count(*)::int FROM agentrun_runs WHERE session_ref->>'sessionId' = $1) AS run_count, + (SELECT count(*)::int FROM agentrun_commands c JOIN agentrun_runs r ON r.id = c.run_id WHERE r.session_ref->>'sessionId' = $1) AS command_count, + (SELECT count(*)::int FROM agentrun_runner_dispatch_intents i JOIN agentrun_runs r ON r.id = i.run_id WHERE r.session_ref->>'sessionId' = $1) AS intent_count, + (SELECT count(*)::int FROM agentrun_kafka_event_outbox o JOIN agentrun_runs r ON r.id = o.run_id WHERE r.session_ref->>'sessionId' = $1) AS outbox_count`, + [concurrentSessionId], + ); + assert.deepEqual(counts.rows[0], { run_count: 1, command_count: 1, intent_count: 1, outbox_count: 3 }); + + await assert.rejects( + () => concurrent.admitSessionTurn(admissionInput(concurrentSessionId, "different active postgres admission", `pg-admission-conflict-${suffix}`)), + (error) => error instanceof AgentRunError + && error.failureKind === "runner-lease-conflict" + && error.details?.reason === "session-turn-admission-active", + ); + + const restarted = await createPostgresAgentRunStore({ connectionString, poolMax: 2, eventOutbox }); + reopenedStores.push(restarted); + const durableIntent = await restarted.getRunnerDispatchIntent(first.command.id); + assert.equal(durableIntent?.id, first.command.dispatchIntent?.id); + assert.equal(durableIntent?.state, "pending"); + const claimed = await restarted.claimRunnerDispatchIntents({ owner: `pg-restarted-${suffix}`, leaseMs: 60_000, limit: 1 }); + assert.equal(claimed.length, 1); + assert.equal(claimed[0]?.id, durableIntent?.id); + const replayed = await restarted.admitSessionTurn(input); + assert.equal(replayed.run.id, first.run.id); + assert.equal(replayed.command.id, first.command.id); + assert.equal(replayed.disposition, "replayed"); + + await restarted.finishCommand(first.command.id, { terminalStatus: "completed", failureKind: null, failureMessage: null }); + await restarted.finishRun(first.run.id, { terminalStatus: "completed", failureKind: null, failureMessage: null }); + const next = await concurrent.admitSessionTurn(admissionInput(concurrentSessionId, "postgres same session next turn", `pg-admission-next-${suffix}`)); + assert.equal(next.disposition, "created"); + assert.notEqual(next.run.id, first.run.id); + const session = await primary.getSession(concurrentSessionId); + assert.equal(session?.activeRunId, next.run.id); + assert.equal(session?.activeCommandId, next.command.id); +} + +async function assertLegacyHalfCommitRecoveryAfterRestart(): Promise { + const input = admissionInput(legacySessionId, "postgres recover legacy half commit", `pg-admission-legacy-${suffix}`); + const run = await primary.createRun(input.run); + const { dispatch: _dispatch, ...legacyCommandInput } = input.command; + const command = await primary.createCommand(run.id, legacyCommandInput); + assert.equal(await primary.getRunnerDispatchIntent(command.id), null); + + const restarted = await createPostgresAgentRunStore({ connectionString, poolMax: 2, eventOutbox }); + reopenedStores.push(restarted); + const recovered = await restarted.admitSessionTurn(input); + assert.equal(recovered.run.id, run.id); + assert.equal(recovered.command.id, command.id); + assert.equal(recovered.disposition, "recovered"); + assert.equal(recovered.recoveredPriorPartialWrite, true); + assert.equal(recovered.command.dispatchIntent?.state, "pending"); + const events = await restarted.listEvents(run.id, 0, 100); + assert.equal(events.filter((event) => event.type === "user_message").length, 1); + assert.equal(events.filter((event) => event.payload.phase === "run-created").length, 1); + assert.equal(events.filter((event) => event.payload.phase === "command-created").length, 1); + assert.equal(events.filter((event) => event.payload.phase === "runner-dispatch-recovered").length, 1); + const replayed = await concurrent.admitSessionTurn(input); + assert.equal(replayed.disposition, "replayed"); + assert.equal((await concurrent.listEvents(run.id, 0, 100)).length, events.length); +} + +async function assertAdmissionTransactionRollback(): Promise { + await installRollbackTrigger(); + const input = admissionInput(rollbackSessionId, `postgres admission rollback ${suffix}`, `pg-admission-rollback-${suffix}`); + await assert.rejects(() => primary.admitSessionTurn(input), /agentrun selftest reject session admission/u); + await dropRollbackTrigger(); + const counts = await control.query( + `SELECT + (SELECT count(*)::int FROM agentrun_sessions WHERE session_id = $1) AS session_count, + (SELECT count(*)::int FROM agentrun_runs WHERE session_ref->>'sessionId' = $1) AS run_count, + (SELECT count(*)::int FROM agentrun_commands c JOIN agentrun_runs r ON r.id = c.run_id WHERE r.session_ref->>'sessionId' = $1) AS command_count, + (SELECT count(*)::int FROM agentrun_events e JOIN agentrun_runs r ON r.id = e.run_id WHERE r.session_ref->>'sessionId' = $1) AS event_count, + (SELECT count(*)::int FROM agentrun_runner_dispatch_intents i JOIN agentrun_runs r ON r.id = i.run_id WHERE r.session_ref->>'sessionId' = $1) AS intent_count, + (SELECT count(*)::int FROM agentrun_kafka_event_outbox o JOIN agentrun_runs r ON r.id = o.run_id WHERE r.session_ref->>'sessionId' = $1) AS outbox_count`, + [rollbackSessionId], + ); + assert.deepEqual(counts.rows[0], { session_count: 0, run_count: 0, command_count: 0, event_count: 0, intent_count: 0, outbox_count: 0 }); +} + +function admissionInput(sessionId: string, prompt: string, idempotencyKey: string): SessionTurnAdmissionInput { + return { + sessionId, + run: validateCreateRun(runInput(sessionId)), + command: validateCreateCommand({ type: "turn", payload: { prompt }, idempotencyKey, dispatch: { kind: "kubernetes-job", input: {} } }), + }; +} + +function runInput(sessionId: string): CreateRunInput { + return { + tenantId: "unidesk", + projectId: "pikasTech/agentrun", + workspaceRef: { kind: "host-path", path: "/tmp/agentrun-postgres-session-admission" }, + sessionRef: { sessionId, metadata: { title: "postgres session admission selftest" } }, + resourceBundleRef: null, + providerId: "NC01", + backendProfile: "codex", + executionPolicy: { + sandbox: "workspace-write", + approval: "never", + timeoutMs: 60_000, + network: "default", + secretScope: { + allowCredentialEcho: false, + providerCredentials: [{ profile: "codex", secretRef: { name: "agentrun-v02-provider-codex", keys: ["auth.json", "config.toml"] } }], + }, + }, + traceSink: null, + }; +} + +async function installRollbackTrigger(): Promise { + await dropRollbackTrigger(); + await control.query(` + CREATE FUNCTION agentrun_selftest_reject_session_admission() RETURNS trigger LANGUAGE plpgsql AS $$ + BEGIN + IF NEW.payload->>'prompt' = 'postgres admission rollback ${suffix}' THEN + RAISE EXCEPTION 'agentrun selftest reject session admission'; + END IF; + RETURN NEW; + END; + $$; + CREATE TRIGGER agentrun_selftest_reject_session_admission + BEFORE INSERT ON agentrun_commands + FOR EACH ROW EXECUTE FUNCTION agentrun_selftest_reject_session_admission(); + `); +} + +async function dropRollbackTrigger(): Promise { + await control.query("DROP TRIGGER IF EXISTS agentrun_selftest_reject_session_admission ON agentrun_commands; DROP FUNCTION IF EXISTS agentrun_selftest_reject_session_admission()"); +} + +async function cleanupSession(sessionId: string): Promise { + await control.query("DELETE FROM agentrun_runs WHERE session_ref->>'sessionId' = $1", [sessionId]); + await control.query("DELETE FROM agentrun_sessions WHERE session_id = $1", [sessionId]); +} + +function requiredDatabaseUrl(): string { + const value = process.env.AGENTRUN_SELFTEST_DATABASE_URL?.trim(); + if (!value) throw new Error("AGENTRUN_SELFTEST_DATABASE_URL is required; this integration test never falls back to memory"); + return value; +}