Merge pull request #298 from pikasTech/fix/297-user-message-before-run-created

修复首轮用户消息与运行创建事件顺序

src/mgr/postgres-store.ts

@@ -6,7 +6,7 @@ import { redactJson } from "../common/redaction.js";
 import type { BackendProfile, BackendTurnResult, CancelRequestRecord, CancelStage, CancelTargetKind, CommandRecord, CommandState, CreateCommandInput, CreateQueueTaskInput, CreateRunInput, EventType, FailureKind, JsonRecord, JsonValue, KafkaEventOutboxRecord, ListGcExpiredSessionsInput, QueueCommanderSnapshot, QueueReadCursorRecord, 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 { AgentRunStore, DurableQueueClaim, EventOutboxConfig, ListQueueTasksInput, ListSessionsInput, RunnerRetentionFenceFacts, RunnerRetentionFenceInput, SaveRunnerJobInput, SessionEventPageInput, StoreHealth, UpdateQueueTaskAttemptInput } from "./store.js";
-import { assertSessionBoundary, buildQueueStats, buildQueueTaskSummary, buildSessionSummary, cancelStagePayload, clampQueueLimit, clampSessionLimit, commandStateFromTerminal, fenceLateEventForCancelledRun, isLeaseExpired, isSessionOutputEvent, isTerminalCommandState, isTerminalQueueTaskState, isTerminalRunStatus, lateWriteRejectedPayload, parseQueueCursor, parseSessionCursor, queueTaskMatchesCommander, queueTaskPayloadHash, queueTaskSort, sessionListFilters, sessionMatchesListState, sessionRefFromRecord, sessionSort, sessionTitleFromCommand, statusFromTerminal, summarizeResourceBundleRef, summarizeSessionRef, titleFromMetadata } from "./store.js";
+import { assertSessionBoundary, buildQueueStats, buildQueueTaskSummary, buildSessionSummary, cancelStagePayload, clampQueueLimit, clampSessionLimit, commandStateFromTerminal, fenceLateEventForCancelledRun, isLeaseExpired, isSessionOutputEvent, isTerminalCommandState, isTerminalQueueTaskState, isTerminalRunStatus, lateWriteRejectedPayload, parseQueueCursor, parseSessionCursor, 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";
@@ -516,7 +516,6 @@ CREATE TABLE IF NOT EXISTS agentrun_schema_migrations (
         [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 });
-      await this.appendEventWithLockedRun(client, run.id, "backend_status", { phase: "run-created", backendProfile: run.backendProfile, sessionRef: summarizeSessionRef(run.sessionRef ?? null), resourceBundleRef: summarizeResourceBundleRef(run.resourceBundleRef ?? null) });
       return run;
     });
   }
@@ -567,6 +566,16 @@ CREATE TABLE IF NOT EXISTS agentrun_schema_migrations (
           return attachRunnerDispatchIntent(command, intent);
         }
       }
+      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;
@@ -588,6 +597,7 @@ CREATE TABLE IF NOT EXISTS agentrun_schema_migrations (
       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);
     });

src/mgr/store.ts

@@ -216,7 +216,6 @@ export class MemoryAgentRunStore implements AgentRunStore {
     this.runs.set(run.id, run);
     this.eventsByRun.set(run.id, []);
     this.touchSessionForRun(run, { lastRunId: run.id, lastActivityAt: at }, { bumpVersion: false, at });
-    this.appendEvent(run.id, "backend_status", { phase: "run-created", backendProfile: run.backendProfile, sessionRef: summarizeSessionRef(run.sessionRef ?? null), resourceBundleRef: summarizeResourceBundleRef(run.resourceBundleRef ?? null) });
     return run;
   }
 
@@ -257,14 +256,16 @@ export class MemoryAgentRunStore implements AgentRunStore {
     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 };
     const userMessage = userMessagePayloadForCommand(run, command);
-    const eventSpecs: Array<{ type: RunEvent["type"]; payload: JsonRecord }> = [
-      ...(userMessage ? [{ type: "user_message" as const, payload: userMessage }] : []),
-      { type: "backend_status", payload: { phase: "command-created", commandId: command.id, commandType: command.type } },
+    const needsRunCreatedEvent = !(this.eventsByRun.get(runId) ?? []).some(isRunCreatedEvent);
+    const eventSpecs: Array<{ type: RunEvent["type"]; payload: JsonRecord; command: CommandRecord | null }> = [
+      ...(userMessage ? [{ type: "user_message" as const, payload: userMessage, command }] : []),
+      ...(needsRunCreatedEvent ? [{ type: "backend_status" as const, payload: runCreatedEventPayload(run), command: null }] : []),
+      { type: "backend_status", payload: { phase: "command-created", commandId: command.id, commandType: command.type }, command },
     ];
     const eventSeqBase = (this.eventsByRun.get(runId) ?? []).length;
     const outboxSeqBase = this.kafkaOutboxSeq;
     const preparedEvents = eventSpecs.map((spec, index) => this.prepareEvent(runId, spec.type, spec.payload, {
-      command,
+      command: spec.command,
       eventSeq: eventSeqBase + index + 1,
       outboxSeq: outboxSeqBase + index + 1,
     }));
@@ -1205,6 +1206,15 @@ export function summarizeSessionRef(sessionRef: SessionRef | null): JsonRecord |
   };
 }
 
+export function runCreatedEventPayload(run: RunRecord): JsonRecord {
+  return {
+    phase: "run-created",
+    backendProfile: run.backendProfile,
+    sessionRef: summarizeSessionRef(run.sessionRef ?? null),
+    resourceBundleRef: summarizeResourceBundleRef(run.resourceBundleRef ?? null),
+  };
+}
+
 export function summarizeResourceBundleRef(resourceBundleRef: RunRecord["resourceBundleRef"] | null | undefined): JsonRecord | null {
   if (!resourceBundleRef) return null;
   return {
@@ -1291,6 +1301,10 @@ export function sessionTitleFromCommand(command: CommandRecord): string | null {
   return value.trim().replace(/\s+/gu, " ").slice(0, 120) || null;
 }
 
+function isRunCreatedEvent(event: RunEvent): boolean {
+  return event.type === "backend_status" && event.payload.phase === "run-created";
+}
+
 function eventMatchesCommand(event: RunEvent, commandId: string): boolean {
   const payload = event.payload;
   return payload.commandId === commandId || payload.targetCommandId === commandId || (event.type === "terminal_status" && payload.commandId === undefined);

src/selftest/cases/22-runner-retention-stale-pending.ts

@@ -764,8 +764,8 @@ function pendingFact(index: number, createdAt: string, variant: "pending" | "cla
     acknowledgedAt: toolExecuting ? progressedAt : null,
   };
   const events: RunEvent[] = [
-    runEvent(runId, 1, createdAt, "backend_status", { phase: "run-created" }),
-    runEvent(runId, 2, createdAt, "user_message", { commandId, traceId: `trc-retention-${suffix}`, userMessageId: `msg-retention-${suffix}`, text: "retention self-test", source: "agentrun-turn-command-payload", valuesPrinted: false }),
+    runEvent(runId, 1, createdAt, "user_message", { commandId, traceId: `trc-retention-${suffix}`, userMessageId: `msg-retention-${suffix}`, text: "retention self-test", source: "agentrun-turn-command-payload", valuesPrinted: false }),
+    runEvent(runId, 2, createdAt, "backend_status", { phase: "run-created" }),
     runEvent(runId, 3, createdAt, "backend_status", { phase: "command-created", commandId }),
     runEvent(runId, 4, createdAt, "backend_status", { phase: "runner-job-created", commandId }),
     runEvent(runId, 5, createdAt, "backend_status", { phase: "runner-dispatch-completed", commandId }),

src/selftest/cases/35-kafka-durable-outbox.ts

@@ -110,8 +110,10 @@ const selfTest: SelfTestCase = async () => {
   assert.deepEqual(published.map((item) => item.sessionId), ["ses_kafka_durable", "ses_kafka_durable", "ses_kafka_durable"]);
   assert.equal(published[2]?.schema, "agentrun.event.v1");
   assert.equal(published[2]?.eventType, "agentrun.event.committed");
-  assert.equal((published[1]?.event as JsonRecord).type, "user_message");
-  assert.equal(((published[1]?.event as JsonRecord).payload as JsonRecord).userMessageId, `msg_${command.id}_user`);
+  assert.equal((published[0]?.event as JsonRecord).type, "user_message");
+  assert.equal(((published[0]?.event as JsonRecord).payload as JsonRecord).userMessageId, `msg_${command.id}_user`);
+  assert.equal(((published[1]?.event as JsonRecord).payload as JsonRecord).phase, "run-created");
+  assert.equal(published[1]?.commandId, null);
   assert.equal(((published[2]?.event as JsonRecord).payload as JsonRecord).phase, "command-created");
   const canonicalMetadata = kafkaMessageMetadata(published[2] ?? null);
   assert.equal(canonicalMetadata.eventId, published[2]?.eventId);
@@ -216,7 +218,8 @@ async function assertStaleClaimsCannotOverwrite(): Promise<void> {
   assert.equal(store.getRunnerDispatchIntent(command.id)?.state, "dispatched");
 
   const outboxStore = new MemoryAgentRunStore({ eventOutbox: { enabled: true, topic: config.agentrunEventTopic, source: config.clientId } });
-  outboxStore.createRun(runInput("ses_stale_outbox"));
+  const outboxRun = outboxStore.createRun(runInput("ses_stale_outbox"));
+  outboxStore.appendEvent(outboxRun.id, "backend_status", { phase: "stale-outbox-claim-fixture" });
   const outboxA = outboxStore.claimKafkaEventOutbox({ owner: "relay-a", leaseMs: 1, limit: 1 })[0];
   assert.ok(outboxA);
   await delay(5);

src/selftest/cases/37-user-message-event.ts

@@ -3,22 +3,26 @@ import { AgentRunError } from "../../common/errors.js";
 import type { CreateRunInput, JsonRecord, KafkaEventOutboxRecord, RunEvent } from "../../common/types.js";
 import { ManagerClient } from "../../mgr/client.js";
 import { startManagerServer } from "../../mgr/server.js";
-import { MemoryAgentRunStore } from "../../mgr/store.js";
+import { MemoryAgentRunStore, runCreatedEventPayload } from "../../mgr/store.js";
 import type { SelfTestCase } from "../harness.js";
 
 const outboxConfig = { enabled: true, topic: "agentrun.event.v1", source: "agentrun-user-message-selftest" };
 
 const selfTest: SelfTestCase = async () => {
   assertTurnAndSteerUserMessages();
+  assertFirstCommandWithoutPrompt();
+  assertLegacyRunCreatedIsNotDuplicated();
   assertCommandCreationIsAtomic();
   assertUserMessageIsManagerOwned();
   await assertHttpAppendRejectsUserMessage();
-  return { name: "user-message-event", tests: ["turn-user-message", "steer-user-message", "steer-target-trace-contract", "interrupt-no-user-message", "idempotent-user-message", "redacted-full-text", "atomic-command-event-batch", "manager-owned-append-authority", "http-append-authority"] };
+  return { name: "user-message-event", tests: ["first-turn-user-message-before-run-created", "first-command-without-prompt", "legacy-run-created-idempotence", "steer-user-message", "steer-target-trace-contract", "interrupt-no-user-message", "idempotent-user-message", "redacted-full-text", "atomic-command-event-batch-retry", "manager-owned-append-authority", "http-append-authority"] };
 };
 
 function assertTurnAndSteerUserMessages(): void {
   const store = new MemoryAgentRunStore({ eventOutbox: outboxConfig });
   const run = store.createRun(runInput("ses_user_message"));
+  assert.deepEqual(store.listEvents(run.id, 0, 100), []);
+  assert.equal(store.kafkaEventOutboxStatus().backlogCount, 0);
   const turnInput = {
     type: "turn" as const,
     idempotencyKey: "turn-user-message",
@@ -42,8 +46,9 @@ function assertTurnAndSteerUserMessages(): void {
 
   const events = store.listEvents(run.id, 0, 100);
   const turnIndex = events.findIndex((event) => event.type === "user_message" && event.payload.commandId === turn.id);
-  assert.ok(turnIndex > 0);
-  assert.equal(events[turnIndex + 1]?.payload.phase, "command-created");
+  assert.equal(turnIndex, 0);
+  assert.equal(events[turnIndex + 1]?.payload.phase, "run-created");
+  assert.equal(events[turnIndex + 2]?.payload.phase, "command-created");
   assert.deepEqual(events[turnIndex]?.payload, {
     commandId: turn.id,
     traceId: "trc_user_message_turn",
@@ -66,7 +71,31 @@ function assertTurnAndSteerUserMessages(): void {
   assert.deepEqual(userMessages.map((item) => item.value.traceId), ["trc_user_message_turn", "trc_user_message_steer"]);
   assert.equal((((userMessages[1]?.value.event as JsonRecord).payload as JsonRecord).targetTraceId), "trc_user_message_turn");
   const turnOutboxIndex = outbox.findIndex((item) => item.eventId === events[turnIndex]?.id);
-  assert.equal(((outbox[turnOutboxIndex + 1]?.value.event as JsonRecord).payload as JsonRecord).phase, "command-created");
+  assert.equal(((outbox[turnOutboxIndex + 1]?.value.event as JsonRecord).payload as JsonRecord).phase, "run-created");
+  assert.equal(((outbox[turnOutboxIndex + 2]?.value.event as JsonRecord).payload as JsonRecord).phase, "command-created");
+}
+
+function assertFirstCommandWithoutPrompt(): void {
+  const store = new MemoryAgentRunStore({ eventOutbox: outboxConfig });
+  const run = store.createRun(runInput("ses_first_command_without_prompt"));
+  const input = { type: "turn" as const, idempotencyKey: "first-command-without-prompt", payload: {} };
+  const command = store.createCommand(run.id, input);
+  assert.equal(store.createCommand(run.id, input).id, command.id);
+  assert.deepEqual(store.listEvents(run.id, 0, 100).map((event) => event.payload.phase ?? event.type), ["run-created", "command-created"]);
+  assert.deepEqual(drainOutbox(store).map((item) => ((item.value.event as JsonRecord).payload as JsonRecord).phase), ["run-created", "command-created"]);
+}
+
+function assertLegacyRunCreatedIsNotDuplicated(): void {
+  const store = new MemoryAgentRunStore({ eventOutbox: outboxConfig });
+  const run = store.createRun(runInput("ses_legacy_run_created"));
+  store.appendEvent(run.id, "backend_status", runCreatedEventPayload(run));
+  const input = { type: "turn" as const, idempotencyKey: "legacy-first-command", payload: { prompt: "legacy first prompt", traceId: "trc_legacy_first" } };
+  const command = store.createCommand(run.id, input);
+  assert.equal(store.createCommand(run.id, input).id, command.id);
+  const events = store.listEvents(run.id, 0, 100);
+  assert.deepEqual(events.map((event) => event.payload.phase ?? event.type), ["run-created", "user_message", "command-created"]);
+  assert.equal(events.filter((event) => event.payload.phase === "run-created").length, 1);
+  assert.equal(drainOutbox(store).filter((item) => ((item.value.event as JsonRecord).payload as JsonRecord).phase === "run-created").length, 1);
 }
 
 function assertUserMessageIsManagerOwned(): void {
@@ -100,13 +129,24 @@ function assertCommandCreationIsAtomic(): void {
   const run = store.createRun(runInput("ses_atomic_user_message"));
   assert.throws(() => store.createCommand(run.id, { type: "turn", payload: { prompt: "must roll back" }, idempotencyKey: "atomic-user-message" }), /reject command lifecycle/u);
   assert.deepEqual(store.listCommands(run.id, 0, 100), []);
-  assert.deepEqual(store.listEvents(run.id, 0, 100).map((event) => event.payload.phase ?? event.type), ["run-created"]);
-  assert.equal(drainOutbox(store).length, 1);
+  assert.deepEqual(store.listEvents(run.id, 0, 100), []);
+  assert.equal(drainOutbox(store).length, 0);
+
+  const input = { type: "turn" as const, payload: { prompt: "retry after rollback" }, idempotencyKey: "atomic-user-message" };
+  const command = store.createCommand(run.id, input);
+  assert.equal(store.createCommand(run.id, input).id, command.id);
+  assert.deepEqual(store.listEvents(run.id, 0, 100).map((event) => event.payload.phase ?? event.type), ["user_message", "run-created", "command-created"]);
+  assert.deepEqual(drainOutbox(store).map((item) => ((item.value.event as JsonRecord).payload as JsonRecord).phase ?? (item.value.event as JsonRecord).type), ["user_message", "run-created", "command-created"]);
 }
 
 class RejectCommandLifecycleStore extends MemoryAgentRunStore {
+  private rejectNextCommandLifecycle = true;
+
   protected override beforeEventCommit(event: RunEvent): void {
-    if (event.payload.phase === "command-created") throw new Error("reject command lifecycle");
+    if (this.rejectNextCommandLifecycle && event.payload.phase === "command-created") {
+      this.rejectNextCommandLifecycle = false;
+      throw new Error("reject command lifecycle");
+    }
   }
 }
 

src/selftest/integration/postgres-kafka-order.ts

@@ -3,6 +3,7 @@ import { Pool, type PoolClient } from "pg";
 import type { CreateRunInput, JsonRecord, KafkaEventOutboxRecord } from "../../common/types.js";
 import { validateCreateCommand } from "../../common/validation.js";
 import { createPostgresAgentRunStore } from "../../mgr/postgres-store.js";
+import { runCreatedEventPayload } from "../../mgr/store.js";
 
 const connectionString = process.env.AGENTRUN_SELFTEST_DATABASE_URL?.trim();
 if (!connectionString) throw new Error("AGENTRUN_SELFTEST_DATABASE_URL is required; this integration test never falls back to memory");
@@ -66,8 +67,8 @@ try {
   const userCommand = await store.createCommand(userRun.id, userCommandInput);
   assert.equal((await store.createCommand(userRun.id, userCommandInput)).id, userCommand.id);
   const userEvents = await store.listEvents(userRun.id, 0, 100);
-  assert.deepEqual(userEvents.map((event) => event.type), ["backend_status", "user_message", "backend_status"]);
-  assert.deepEqual(userEvents[1]?.payload, {
+  assert.deepEqual(userEvents.map((event) => event.type), ["user_message", "backend_status", "backend_status"]);
+  assert.deepEqual(userEvents[0]?.payload, {
     commandId: userCommand.id,
     traceId: "trc_pg_user_message",
     userMessageId: "msg_pg_user_message",
@@ -75,9 +76,10 @@ try {
     source: "agentrun-turn-command-payload",
     valuesPrinted: false,
   });
+  assert.equal(userEvents[1]?.payload.phase, "run-created");
   assert.equal(userEvents[2]?.payload.phase, "command-created");
   const userOutbox = await userMessageOutboxRows(controlPool, userRun.id);
-  assert.deepEqual(userOutbox.map((row) => row.eventType), ["backend_status", "user_message", "backend_status"]);
+  assert.deepEqual(userOutbox.map((row) => row.eventType), ["user_message", "backend_status", "backend_status"]);
   assert.deepEqual(userOutbox.map((row) => row.sourceSeq), [1, 2, 3]);
   assert.deepEqual(userOutbox.map((row) => row.partitionKey), ["ses_pg_user_message_order", "ses_pg_user_message_order", "ses_pg_user_message_order"]);
 
@@ -93,11 +95,43 @@ try {
   assert.equal(steerUserEvent?.payload.traceId, "trc_pg_user_message_steer");
   assert.equal(steerUserEvent?.payload.targetTraceId, "trc_pg_user_message");
   const steerOutbox = await userMessageOutboxRows(controlPool, userRun.id);
-  assert.deepEqual(steerOutbox.map((row) => row.eventType), ["backend_status", "user_message", "backend_status", "user_message", "backend_status"]);
+  assert.deepEqual(steerOutbox.map((row) => row.eventType), ["user_message", "backend_status", "backend_status", "user_message", "backend_status"]);
   assert.deepEqual(steerOutbox.map((row) => row.sourceSeq), [1, 2, 3, 4, 5]);
   assert.deepEqual(steerOutbox.map((row) => row.partitionKey), Array(5).fill("ses_pg_user_message_order"));
   assert.equal(steerOutbox[3]?.targetTraceId, "trc_pg_user_message");
 
+  const noPromptRun = await store.createRun(runInput("ses_pg_first_command_without_prompt"));
+  const noPromptInput = validateCreateCommand({ type: "turn", payload: {}, idempotencyKey: "pg-first-command-without-prompt" });
+  const noPromptCommand = await store.createCommand(noPromptRun.id, noPromptInput);
+  assert.equal((await store.createCommand(noPromptRun.id, noPromptInput)).id, noPromptCommand.id);
+  assert.deepEqual((await store.listEvents(noPromptRun.id, 0, 100)).map((event) => event.payload.phase ?? event.type), ["run-created", "command-created"]);
+  assert.deepEqual((await userMessageOutboxRows(controlPool, noPromptRun.id)).map((row) => row.sourceSeq), [1, 2]);
+
+  const legacyRun = await store.createRun(runInput("ses_pg_legacy_run_created"));
+  await store.appendEvent(legacyRun.id, "backend_status", runCreatedEventPayload(legacyRun));
+  const legacyInput = validateCreateCommand({ type: "turn", payload: { prompt: "legacy postgres prompt", traceId: "trc_pg_legacy" }, idempotencyKey: "pg-legacy-first-command" });
+  const legacyCommand = await store.createCommand(legacyRun.id, legacyInput);
+  assert.equal((await store.createCommand(legacyRun.id, legacyInput)).id, legacyCommand.id);
+  const legacyEvents = await store.listEvents(legacyRun.id, 0, 100);
+  assert.deepEqual(legacyEvents.map((event) => event.payload.phase ?? event.type), ["run-created", "user_message", "command-created"]);
+  assert.equal(legacyEvents.filter((event) => event.payload.phase === "run-created").length, 1);
+
+  const rollbackRun = await store.createRun(runInput("ses_pg_command_outbox_rollback"));
+  const rollbackInput = validateCreateCommand({ type: "turn", payload: { prompt: "postgres rollback prompt", traceId: "trc_pg_rollback" }, idempotencyKey: "pg-command-outbox-rollback" });
+  await installCommandOutboxFailureTrigger(controlPool, rollbackRun.id);
+  try {
+    await assert.rejects(() => store.createCommand(rollbackRun.id, rollbackInput), /agentrun selftest reject command outbox/u);
+  } finally {
+    await removeCommandOutboxFailureTrigger(controlPool);
+  }
+  assert.deepEqual(await store.listCommands(rollbackRun.id, 0, 100), []);
+  assert.deepEqual(await store.listEvents(rollbackRun.id, 0, 100), []);
+  assert.deepEqual(await userMessageOutboxRows(controlPool, rollbackRun.id), []);
+  const rollbackCommand = await store.createCommand(rollbackRun.id, rollbackInput);
+  assert.equal((await store.createCommand(rollbackRun.id, rollbackInput)).id, rollbackCommand.id);
+  assert.deepEqual((await store.listEvents(rollbackRun.id, 0, 100)).map((event) => event.payload.phase ?? event.type), ["user_message", "run-created", "command-created"]);
+  assert.deepEqual((await userMessageOutboxRows(controlPool, rollbackRun.id)).map((row) => row.sourceSeq), [1, 2, 3]);
+
   const dispatchRun = await store.createRun({ ...runInput("ses_pg_dispatch_parent_lock"), sessionRef: null });
   const dispatchCommand = await store.createCommand(dispatchRun.id, validateCreateCommand({ type: "turn", payload: {}, dispatch: { kind: "kubernetes-job", input: {} } }));
   const parentLock = await controlPool.connect();
@@ -118,10 +152,11 @@ try {
   await store.cancelRun(dispatchRun.id, "postgres parent-lock selftest");
   assert.equal((await store.getRunnerDispatchIntent(dispatchCommand.id))?.state, "cancelled");
 
-  console.log(JSON.stringify({ ok: true, test: "postgres-kafka-same-session-commit-order", outboxSeq: rows.map((row) => row.outboxSeq), publishOrder: [marker(first), marker(second)], partitionKey: sessionId, userMessageOrder: steerOutbox.map((row) => row.eventType), userMessageSourceSeq: steerOutbox.map((row) => row.sourceSeq), steerTargetTraceId: steerOutbox[3]?.targetTraceId ?? null, dispatchParentLockFenced: true, valuesPrinted: false }));
+  console.log(JSON.stringify({ ok: true, test: "postgres-kafka-same-session-commit-order", outboxSeq: rows.map((row) => row.outboxSeq), publishOrder: [marker(first), marker(second)], partitionKey: sessionId, userMessageOrder: steerOutbox.map((row) => row.eventType), userMessageSourceSeq: steerOutbox.map((row) => row.sourceSeq), steerTargetTraceId: steerOutbox[3]?.targetTraceId ?? null, firstCommandWithoutPrompt: true, legacyRunCreatedDeduplicated: true, commandOutboxRollbackRetried: true, dispatchParentLockFenced: true, valuesPrinted: false }));
 } finally {
   if (gateClient && gateHeld) await gateClient.query("SELECT pg_advisory_unlock($1::bigint)", [gate]).catch(() => undefined);
   gateClient?.release();
+  await removeCommandOutboxFailureTrigger(controlPool).catch(() => undefined);
   await removeGateTrigger(controlPool).catch(() => undefined);
   await store.close();
   await controlPool.end();
@@ -168,6 +203,28 @@ async function removeGateTrigger(pool: Pool): Promise<void> {
   await pool.query("DROP TRIGGER IF EXISTS agentrun_selftest_gate_outbox_a ON agentrun_kafka_event_outbox; DROP FUNCTION IF EXISTS agentrun_selftest_gate_outbox_a()");
 }
 
+async function installCommandOutboxFailureTrigger(pool: Pool, runId: string): Promise<void> {
+  assert.match(runId, /^run_[a-f0-9]{32}$/u);
+  await removeCommandOutboxFailureTrigger(pool);
+  await pool.query(`
+    CREATE FUNCTION agentrun_selftest_reject_command_outbox() RETURNS trigger LANGUAGE plpgsql AS $$
+    BEGIN
+      IF NEW.run_id = '${runId}' AND NEW.value #>> '{event,payload,phase}' = 'command-created' THEN
+        RAISE EXCEPTION 'agentrun selftest reject command outbox';
+      END IF;
+      RETURN NEW;
+    END;
+    $$;
+    CREATE TRIGGER agentrun_selftest_reject_command_outbox
+      BEFORE INSERT ON agentrun_kafka_event_outbox
+      FOR EACH ROW EXECUTE FUNCTION agentrun_selftest_reject_command_outbox();
+  `);
+}
+
+async function removeCommandOutboxFailureTrigger(pool: Pool): Promise<void> {
+  await pool.query("DROP TRIGGER IF EXISTS agentrun_selftest_reject_command_outbox ON agentrun_kafka_event_outbox; DROP FUNCTION IF EXISTS agentrun_selftest_reject_command_outbox()");
+}
+
 async function waitForBlockedOutboxInsert(pool: Pool): Promise<void> {
   const deadline = Date.now() + 5_000;
   while (Date.now() < deadline) {