fix(mgr): 原子准入会话新轮次

2026-07-12 10:45:12 +02:00
parent ae0fb3357c
commit 895423481a
9 changed files with 1055 additions and 100 deletions
@@ -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"
  },
@@ -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 });
@@ -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<RunRecord> {
-    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<RunRecord> {
@@ -629,55 +608,53 @@ CREATE TABLE IF NOT EXISTS agentrun_schema_migrations (
  }

  async createCommand(runId: string, input: CreateCommandInput): Promise<CommandRecord> {
-    const payloadHash = stableHash(input.payload);
+    return this.withTransaction(async (client) => await this.createCommandWithClient(client, runId, input));
+  }
+
+  async admitSessionTurn(input: SessionTurnAdmissionInput): Promise<SessionTurnAdmission> {
+    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<RunRecord> {
+    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<CommandRecord> {
+    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<SessionTurnAdmission> {
+    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<RunnerDispatchIntentRecord | null> {
+    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<void> {
+    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<SessionRef | null> {
    if (!input.sessionRef) return null;
    const existing = await client.query("SELECT * FROM agentrun_sessions WHERE session_id = $1 FOR UPDATE", [input.sessionRef.sessionId]);
@@ -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 };
 }
@@ -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<JsonRecord> {
+async function sendToSession(input: { store: AgentRunStore; sessionId: string; body: unknown; runnerDispatcherOptions: RunnerDispatcherOptions }): Promise<JsonRecord> {
  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<ReturnType<typeof activeReceivableCommand>>; 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<ReturnType<typeof activeReceivableCommand>>; 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: [
@@ -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<StoreHealth>;
  createRun(input: CreateRunInput, identity?: CreateRunIdentity): MaybePromise<RunRecord>;
@@ -54,6 +68,7 @@ export interface AgentRunStore {
  listEvents(runId: string, afterSeq: number, limit: number): MaybePromise<RunEvent[]>;
  listEventsForCommand(runId: string, commandId: string, limit: number): MaybePromise<RunEvent[]>;
  createCommand(runId: string, input: CreateCommandInput): MaybePromise<CommandRecord>;
+  admitSessionTurn(input: SessionTurnAdmissionInput): MaybePromise<SessionTurnAdmission>;
  getCommand(commandId: string): MaybePromise<CommandRecord>;
  listCommands(runId: string, afterSeq: number, limit: number): MaybePromise<CommandRecord[]>;
  getRunnerDispatchIntent(commandId: string): MaybePromise<RunnerDispatchIntentRecord | null>;
@@ -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 });
 }
@@ -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<void> {
+  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<void> {
+  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<void> {
+  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<void> {
+  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<void> {
+  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<void> {
+  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<void> {
+  await new Promise<void>((resolve) => server.close(() => resolve()));
+}
+
+async function delay(ms: number): Promise<void> {
+  await new Promise<void>((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;
@@ -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";
+}
@@ -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<Awaited<ReturnType<typeof createPostgresAgentRunStore>>> = [];
+
+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<void> {
+  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<void> {
+  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<void> {
+  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<void> {
+  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<void> {
+  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<void> {
+  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;
+}