feat: 发布用户输入 Kafka 正式事件

scripts/src/cli.ts

@@ -528,7 +528,7 @@ export function summarizeSessionEventPage(page: JsonValue, options: SessionEvent
     items,
     drillDownCommands: sessionEventDrillDownCommands(options.kind, options.sessionId, { afterSeq: options.afterSeq, limit: options.limit, runId }),
     progressiveDisclosure: {
-      default: "assistant messages plus tool summaries; command_output/backend_status chunks are counted, not displayed",
+      default: "user and assistant messages plus tool summaries; command_output/backend_status chunks are counted, not displayed",
       includeOutputFlag: "--include-output",
       detailFlags: "--seq <seq> or --item-id <itemId> --full",
       fullFlag: "--full",
@@ -576,8 +576,9 @@ export function summarizeQueueDispatchResult(result: JsonValue, taskId: string):
 function summarizeRunEvent(event: JsonRecord, summaryChars: number): JsonRecord {
   const payload = jsonRecordValue(event.payload) ?? {};
   const type = stringValue(event.type) ?? "unknown";
-  const command = type === "assistant_message" ? null : boundedSummaryString(stringValue(payload.command), Math.min(summaryChars, 240));
-  const text = type === "assistant_message" ? boundedSummaryString(stringValue(payload.text), summaryChars) : null;
+  const messageEvent = type === "user_message" || type === "assistant_message";
+  const command = messageEvent ? null : boundedSummaryString(stringValue(payload.command), Math.min(summaryChars, 240));
+  const text = messageEvent ? boundedSummaryString(stringValue(payload.text), summaryChars) : null;
   const outputSummary = boundedSummaryString(stringValue(payload.outputSummary) ?? nestedSummaryText(payload), summaryChars);
   const message = boundedSummaryString(stringValue(payload.failureMessage) ?? stringValue(payload.message), summaryChars);
   const summary = text ?? outputSummary ?? command ?? message ?? "";
@@ -610,7 +611,7 @@ function summarizeRunEvent(event: JsonRecord, summaryChars: number): JsonRecord
 
 function isDefaultVisibleSessionEvent(item: JsonRecord): boolean {
   const type = stringValue(item.type);
-  return type === "assistant_message" || type === "tool_call" || type === "error";
+  return type === "user_message" || type === "assistant_message" || type === "tool_call" || type === "error";
 }
 
 function summarizeSuppressedSessionEvents(all: JsonRecord[], visible: JsonRecord[]): JsonRecord {

src/common/events.ts

@@ -1,12 +1,15 @@
 import { AgentRunError } from "./errors.js";
-import type { EventType, JsonRecord, RunEvent, TerminalStatus } from "./types.js";
+import type { CommandRecord, EventType, JsonRecord, RunEvent, RunRecord, TerminalStatus } from "./types.js";
 import { boundedTextSummary, commandOutputPayload } from "./output.js";
+import { agentRunBusinessTraceId } from "./otel-trace.js";
 import { redactJson, redactText } from "./redaction.js";
 
-export const eventTypes = ["backend_status", "assistant_message", "tool_call", "command_output", "diff", "error", "terminal_status"] as const satisfies readonly EventType[];
+export const eventTypes = ["backend_status", "user_message", "assistant_message", "tool_call", "command_output", "diff", "error", "terminal_status"] as const satisfies readonly EventType[];
+export const externallyAppendableEventTypes = ["backend_status", "assistant_message", "tool_call", "command_output", "diff", "error", "terminal_status"] as const satisfies readonly EventType[];
 export const terminalStatuses = ["completed", "failed", "blocked", "cancelled"] as const satisfies readonly TerminalStatus[];
 
 const eventTypeSet = new Set<string>(eventTypes);
+const externallyAppendableEventTypeSet = new Set<string>(externallyAppendableEventTypes);
 const terminalStatusSet = new Set<string>(terminalStatuses);
 
 export function isEventType(value: unknown): value is EventType {
@@ -22,14 +25,37 @@ export function requireEventType(value: unknown): EventType {
   throw new AgentRunError("schema-invalid", `event.type ${String(value)} is not supported`, { httpStatus: 400, details: { allowedEventTypes: [...eventTypes] } });
 }
 
+export function requireExternallyAppendableEventType(value: unknown): EventType {
+  const type = requireEventType(value);
+  if (externallyAppendableEventTypeSet.has(type)) return type;
+  throw new AgentRunError("tenant-policy-denied", `${type} events are manager-owned and may only be created with the command transaction`, { httpStatus: 403 });
+}
+
 export function normalizeRunEventPayload(type: EventType, payload: JsonRecord): JsonRecord {
   if (type === "terminal_status") return normalizeTerminalStatusPayload(payload);
   if (type === "command_output") return normalizeCommandOutputPayload(payload);
+  if (type === "user_message") return normalizeUserMessagePayload(payload);
   if (type === "assistant_message") return normalizeTextPayload(payload);
   if (type === "tool_call") return normalizeToolCallPayload(payload);
   return payload;
 }
 
+export function userMessagePayloadForCommand(run: RunRecord, command: CommandRecord): JsonRecord | null {
+  if (command.type !== "turn" && command.type !== "steer") return null;
+  const text = promptBearingText(command.payload);
+  if (text === null) return null;
+  const targetTraceId = command.type === "steer" ? nonEmptyText(command.payload.targetTraceId) : null;
+  return {
+    commandId: command.id,
+    traceId: agentRunBusinessTraceId(run, command),
+    userMessageId: nonEmptyText(command.payload.userMessageId) ?? `msg_${command.id}_user`,
+    text,
+    source: command.type === "steer" ? "agentrun-steer-command-payload" : "agentrun-turn-command-payload",
+    ...(targetTraceId ? { targetTraceId } : {}),
+    valuesPrinted: false,
+  };
+}
+
 export function eventContractSummary(events: RunEvent[]): JsonRecord {
   const issues: JsonRecord[] = [];
   let terminalStatusCount = 0;
@@ -69,6 +95,31 @@ function normalizeCommandOutputPayload(payload: JsonRecord): JsonRecord {
   return { ...rest, ...commandOutputPayload(stream, value) };
 }
 
+function normalizeUserMessagePayload(payload: JsonRecord): JsonRecord {
+  const { targetTraceId: _targetTraceId, ...rest } = payload;
+  const commandId = nonEmptyText(payload.commandId);
+  const userMessageId = nonEmptyText(payload.userMessageId);
+  const text = typeof payload.text === "string" && payload.text.trim().length > 0 ? redactText(payload.text) : null;
+  const source = nonEmptyText(payload.source);
+  const targetTraceId = nonEmptyText(payload.targetTraceId);
+  if (!commandId || !userMessageId || text === null || !source) {
+    throw new AgentRunError("schema-invalid", "user_message event requires commandId, userMessageId, non-empty text, and source", { httpStatus: 400 });
+  }
+  return { ...rest, commandId, userMessageId, text, source, ...(targetTraceId ? { targetTraceId } : {}), valuesPrinted: false };
+}
+
+function promptBearingText(payload: JsonRecord): string | null {
+  for (const key of ["prompt", "message", "text"]) {
+    const value = payload[key];
+    if (typeof value === "string" && value.trim().length > 0) return value;
+  }
+  return null;
+}
+
+function nonEmptyText(value: unknown): string | null {
+  return typeof value === "string" && value.trim().length > 0 ? value.trim() : null;
+}
+
 function normalizeTextPayload(payload: JsonRecord): JsonRecord {
   const { text: _text, delta: _delta, content: _content, summary: _summary, textBytes: _textBytes, textTruncated: _textTruncated, outputBytes: _outputBytes, outputTruncated: _outputTruncated, ...rest } = payload;
   const value = typeof payload.text === "string" ? payload.text : typeof payload.delta === "string" ? payload.delta : typeof payload.content === "string" ? payload.content : "";

src/common/types.ts

@@ -335,7 +335,7 @@ export interface CancelRequestRecord extends JsonRecord {
   updatedAt: string;
 }
 
-export type EventType = "backend_status" | "assistant_message" | "tool_call" | "command_output" | "diff" | "error" | "terminal_status";
+export type EventType = "backend_status" | "user_message" | "assistant_message" | "tool_call" | "command_output" | "diff" | "error" | "terminal_status";
 
 export interface RunEvent extends JsonRecord {
   id: string;

src/mgr/postgres-store.ts

@@ -8,7 +8,7 @@ import { newId, nowIso, stableHash } from "../common/validation.js";
 import type { AgentRunStore, DurableQueueClaim, EventOutboxConfig, ListQueueTasksInput, ListSessionsInput, 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 { backendCapabilitiesSqlValues, mergeBackendCapability } from "../common/backend-profiles.js";
-import { normalizeRunEventPayload, requireEventType } from "../common/events.js";
+import { normalizeRunEventPayload, requireEventType, requireExternallyAppendableEventType, userMessagePayloadForCommand } from "../common/events.js";
 import { buildKafkaEventOutboxRecord, canonicalAgentRunEventPartitionKey } from "./event-outbox.js";
 import { assertRunnerDispatchReplay, attachRunnerDispatchIntent, newRunnerDispatchIntent } from "./runner-dispatch-intent.js";
 import { durableDispatchAndKafkaOutboxMigrationSql, durableQueueStatusFromRow, kafkaEventOutboxFromRow, runnerDispatchIntentFromRow, runnerDispatchOutcomeMigrationSql } from "./postgres-durable-queues.js";
@@ -584,6 +584,8 @@ CREATE TABLE IF NOT EXISTS agentrun_schema_migrations (
       }
       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);
       await this.appendEventWithLockedRun(client, runId, "backend_status", { phase: "command-created", commandId: command.id, commandType: command.type });
       return attachRunnerDispatchIntent(command, intent);
     });
@@ -955,6 +957,7 @@ CREATE TABLE IF NOT EXISTS agentrun_schema_migrations (
   }
 
   async appendEvent(runId: string, type: EventType, payload: JsonRecord): Promise<RunEvent> {
+    type = requireExternallyAppendableEventType(type);
     return this.withTransaction(async (client) => {
       await this.requireRunForUpdate(client, runId);
       return this.appendEventWithLockedRun(client, runId, type, payload);

src/mgr/runner-reconciler.ts

@@ -1,5 +1,6 @@
 import { spawn } from "node:child_process";
 import { AgentRunError } from "../common/errors.js";
+import { externallyAppendableEventTypes } from "../common/events.js";
 import { redactJson, redactText } from "../common/redaction.js";
 import type { BackendEvent, CommandRecord, EventType, FailureKind, JsonRecord, JsonValue, RunnerJobRecord, TerminalStatus } from "../common/types.js";
 import { emitAgentRunOtelSpan } from "../common/otel-trace.js";
@@ -556,7 +557,7 @@ function recordValue(value: unknown): JsonRecord | null {
   return typeof value === "object" && value !== null && !Array.isArray(value) ? value as JsonRecord : null;
 }
 
-const eventTypes = new Set<string>(["backend_status", "assistant_message", "tool_call", "command_output", "diff", "error", "terminal_status"]);
+const eventTypes = new Set<string>(externallyAppendableEventTypes);
 const terminalStatuses = new Set<string>(["completed", "failed", "blocked", "cancelled"]);
 const failureKinds = new Set<string>([
   "auth-missing",

src/mgr/store.ts

@@ -3,7 +3,7 @@ import { AgentRunError } from "../common/errors.js";
 import { newId, nowIso, stableHash } from "../common/validation.js";
 import { redactJson } from "../common/redaction.js";
 import { backendCapabilities } from "../common/backend-profiles.js";
-import { normalizeRunEventPayload, requireEventType } from "../common/events.js";
+import { normalizeRunEventPayload, requireEventType, requireExternallyAppendableEventType, userMessagePayloadForCommand } from "../common/events.js";
 import { buildKafkaEventOutboxRecord } from "./event-outbox.js";
 import { assertRunnerDispatchReplay, attachRunnerDispatchIntent, newRunnerDispatchIntent } from "./runner-dispatch-intent.js";
 
@@ -238,12 +238,24 @@ export class MemoryAgentRunStore implements AgentRunStore {
     const seq = Array.from(this.commands.values()).filter((command) => command.runId === runId).length + 1;
     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 eventSeqBase = (this.eventsByRun.get(runId) ?? []).length;
+    const outboxSeqBase = this.kafkaOutboxSeq;
+    const preparedEvents = eventSpecs.map((spec, index) => this.prepareEvent(runId, spec.type, spec.payload, {
+      command,
+      eventSeq: eventSeqBase + index + 1,
+      outboxSeq: outboxSeqBase + index + 1,
+    }));
     this.commands.set(command.id, command);
     const intent = input.dispatch ? newRunnerDispatchIntent(command, input.dispatch, at) : null;
     if (intent) this.runnerDispatchIntents.set(intent.id, intent);
     if (command.type === "turn") this.touchSessionForRun(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") this.touchSessionForRun(run, { executionState: "running", lastRunId: run.id, lastCommandId: command.id, activeRunId: run.id, lastActivityAt: at }, { bumpVersion: true, at });
-    this.appendEvent(runId, "backend_status", { phase: "command-created", commandId: command.id, commandType: command.type });
+    for (const prepared of preparedEvents) this.commitPreparedEvent(prepared);
     return attachRunnerDispatchIntent(command, intent);
   }
 
@@ -460,24 +472,24 @@ export class MemoryAgentRunStore implements AgentRunStore {
   }
 
   appendEvent(runId: string, type: RunEvent["type"], payload: JsonRecord): RunEvent {
-    const prepared = this.prepareEvent(runId, type, payload);
+    const prepared = this.prepareEvent(runId, requireExternallyAppendableEventType(type), payload);
     this.commitPreparedEvent(prepared);
     return prepared.event;
   }
 
   protected beforeEventCommit(_event: RunEvent): void {}
 
-  private prepareEvent(runId: string, type: RunEvent["type"], payload: JsonRecord): { run: RunRecord; event: RunEvent; outbox: KafkaEventOutboxRecord | null } {
+  private prepareEvent(runId: string, type: RunEvent["type"], payload: JsonRecord, options: { command?: CommandRecord | null; eventSeq?: number; outboxSeq?: number } = {}): { run: RunRecord; event: RunEvent; outbox: KafkaEventOutboxRecord | null } {
     const run = this.getRun(runId);
     const eventType = requireEventType(type);
     const fenced = fenceLateEventForCancelledRun(run, eventType, payload);
     const eventPayload = normalizeRunEventPayload(fenced.type, fenced.payload);
     const events = this.eventsByRun.get(runId) ?? [];
-    const event: RunEvent = { id: newId("evt"), runId, seq: events.length + 1, type: fenced.type, payload: redactJson(eventPayload), createdAt: nowIso() };
+    const event: RunEvent = { id: newId("evt"), runId, seq: options.eventSeq ?? events.length + 1, type: fenced.type, payload: redactJson(eventPayload), createdAt: nowIso() };
     let outbox: KafkaEventOutboxRecord | null = null;
     if (this.eventOutboxConfig.enabled) {
-      const command = commandForEvent(this.commands, event);
-      outbox = buildKafkaEventOutboxRecord({ source: this.eventOutboxConfig.source as string, topic: this.eventOutboxConfig.topic as string, outboxSeq: this.kafkaOutboxSeq + 1, run, command, event });
+      const command = options.command === undefined ? commandForEvent(this.commands, event) : options.command;
+      outbox = buildKafkaEventOutboxRecord({ source: this.eventOutboxConfig.source as string, topic: this.eventOutboxConfig.topic as string, outboxSeq: options.outboxSeq ?? this.kafkaOutboxSeq + 1, run, command, event });
     }
     this.beforeEventCommit(event);
     return { run, event, outbox };

src/selftest/cases/15-cli-events-summary.ts

@@ -30,6 +30,14 @@ export default function selfTest(_context: SelfTestContext): SelfTestResult {
   assert.equal(tsv.includes("\n12\ttool_call\tshell\tcompleted\t0\t321\ttrue\t4096"), true);
   assert.equal(tsv.includes("test-token-material"), false);
 
+  const userMessageSummary = summarizeRunEventPage({ items: [
+    { id: "evt_user", seq: 14, type: "user_message", payload: { commandId: "cmd_user", userMessageId: "msg_user", text: "replay this user input", source: "agentrun-turn-command-payload", valuesPrinted: false } },
+  ] }, { runId: "run_selftest", afterSeq: 13, limit: 1, tail: null, summaryChars: 80 });
+  const userMessageItem = (userMessageSummary.items as JsonRecord[])[0];
+  assert.equal(userMessageItem?.type, "user_message");
+  assert.equal(userMessageItem?.summary, "replay this user input");
+  assert.equal(userMessageItem?.text, "replay this user input");
+
   const hugeRunnerTrace = "runner trace line with test-token-material\n".repeat(2_000);
   const sessionSummary = summarizeSessionEventPage({
     sessionId: "sess_noise",
@@ -149,5 +157,5 @@ export default function selfTest(_context: SelfTestContext): SelfTestResult {
   assert.equal(String(((showSummary.pollCommands as JsonRecord).logs)).includes("sess_noisy"), true);
   assertNoSecretLeak(showSummary);
 
-  return { name: "15-cli-events-summary", tests: ["runs-events-summary-tail", "runs-events-summary-tsv-redaction", "sessions-events-low-noise-summary", "queue-dispatch-low-noise-summary", "queue-list-show-commander-low-noise-summary"] };
+  return { name: "15-cli-events-summary", tests: ["runs-events-summary-tail", "runs-events-user-message-summary", "runs-events-summary-tsv-redaction", "sessions-events-low-noise-summary", "queue-dispatch-low-noise-summary", "queue-list-show-commander-low-noise-summary"] };
 }

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

@@ -102,17 +102,21 @@ const selfTest: SelfTestCase = async () => {
   const publish = async (_config: AgentRunKafkaConfig, message: { value: JsonRecord }): Promise<void> => { published.push(message.value); };
   const firstCycle = await relayKafkaEventOutboxOnce({ store, options: relayOptions, publish });
   const secondCycle = await relayKafkaEventOutboxOnce({ store, options: relayOptions, publish });
+  const thirdCycle = await relayKafkaEventOutboxOnce({ store, options: relayOptions, publish });
   assert.equal(firstCycle.deliveredCount, 1);
   assert.equal(secondCycle.deliveredCount, 1);
-  assert.deepEqual(published.map((item) => item.outboxSeq), [1, 2]);
-  assert.deepEqual(published.map((item) => item.sessionId), ["ses_kafka_durable", "ses_kafka_durable"]);
-  assert.equal(published[1]?.schema, "agentrun.event.v1");
-  assert.equal(published[1]?.eventType, "agentrun.event.committed");
-  assert.equal(((published[1]?.event as JsonRecord).payload as JsonRecord).phase, "command-created");
-  const canonicalMetadata = kafkaMessageMetadata(published[1] ?? null);
-  assert.equal(canonicalMetadata.eventId, published[1]?.eventId);
-  assert.equal(canonicalMetadata.outboxSeq, published[1]?.outboxSeq);
-  assert.equal(canonicalMetadata.sourceSeq, published[1]?.sourceSeq);
+  assert.equal(thirdCycle.deliveredCount, 1);
+  assert.deepEqual(published.map((item) => item.outboxSeq), [1, 2, 3]);
+  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[2]?.event as JsonRecord).payload as JsonRecord).phase, "command-created");
+  const canonicalMetadata = kafkaMessageMetadata(published[2] ?? null);
+  assert.equal(canonicalMetadata.eventId, published[2]?.eventId);
+  assert.equal(canonicalMetadata.outboxSeq, published[2]?.outboxSeq);
+  assert.equal(canonicalMetadata.sourceSeq, published[2]?.sourceSeq);
   assert.equal(canonicalMetadata.frameSeq, null);
   const stdioMetadata = kafkaMessageMetadata(rawFrameValue({
     schema: "codex-stdio.raw.v1",

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

@@ -0,0 +1,137 @@
+import assert from "node:assert/strict";
+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 type { SelfTestCase } from "../harness.js";
+
+const outboxConfig = { enabled: true, topic: "agentrun.event.v1", source: "agentrun-user-message-selftest" };
+
+const selfTest: SelfTestCase = async () => {
+  assertTurnAndSteerUserMessages();
+  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"] };
+};
+
+function assertTurnAndSteerUserMessages(): void {
+  const store = new MemoryAgentRunStore({ eventOutbox: outboxConfig });
+  const run = store.createRun(runInput("ses_user_message"));
+  const turnInput = {
+    type: "turn" as const,
+    idempotencyKey: "turn-user-message",
+    payload: {
+      prompt: "  inspect sk-12345678 now  ",
+      traceId: "trc_user_message_turn",
+      userMessageId: "msg_hwlab_user_turn",
+    },
+  };
+  const turn = store.createCommand(run.id, turnInput);
+  assert.equal(store.createCommand(run.id, turnInput).id, turn.id);
+  const steerInput = {
+    type: "steer" as const,
+    idempotencyKey: "steer-user-message",
+    payload: { message: "steer next", traceId: "trc_user_message_steer", targetTraceId: "trc_user_message_turn" },
+  };
+  const steer = store.createCommand(run.id, steerInput);
+  assert.equal(store.createCommand(run.id, steerInput).id, steer.id);
+  store.createCommand(run.id, { type: "interrupt", idempotencyKey: "interrupt-no-user-message", payload: { message: "not a prompt event" } });
+  store.createCommand(run.id, { type: "turn", idempotencyKey: "empty-turn-no-user-message", payload: {} });
+
+  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.deepEqual(events[turnIndex]?.payload, {
+    commandId: turn.id,
+    traceId: "trc_user_message_turn",
+    userMessageId: "msg_hwlab_user_turn",
+    text: "  inspect REDACTED now  ",
+    source: "agentrun-turn-command-payload",
+    valuesPrinted: false,
+  });
+  const steerEvent = events.find((event) => event.type === "user_message" && event.payload.commandId === steer.id);
+  assert.equal(steerEvent?.payload.userMessageId, `msg_${steer.id}_user`);
+  assert.equal(steerEvent?.payload.source, "agentrun-steer-command-payload");
+  assert.equal(steerEvent?.payload.traceId, "trc_user_message_steer");
+  assert.equal(steerEvent?.payload.targetTraceId, "trc_user_message_turn");
+  assert.equal(events.filter((event) => event.type === "user_message").length, 2);
+
+  const outbox = drainOutbox(store);
+  const userMessages = outbox.filter((item) => ((item.value.event as JsonRecord).type === "user_message"));
+  assert.equal(userMessages.length, 2);
+  assert.deepEqual(userMessages.map((item) => item.partitionKey), ["ses_user_message", "ses_user_message"]);
+  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");
+}
+
+function assertUserMessageIsManagerOwned(): void {
+  const store = new MemoryAgentRunStore({ eventOutbox: outboxConfig });
+  const run = store.createRun(runInput("ses_user_message_authority"));
+  assert.throws(
+    () => store.appendEvent(run.id, "user_message", { commandId: "cmd_forged", userMessageId: "msg_forged", text: "forged", source: "runner" }),
+    (error) => error instanceof AgentRunError && error.failureKind === "tenant-policy-denied" && error.httpStatus === 403,
+  );
+  assert.equal(store.listEvents(run.id, 0, 100).filter((event) => event.type === "user_message").length, 0);
+}
+
+async function assertHttpAppendRejectsUserMessage(): Promise<void> {
+  const store = new MemoryAgentRunStore({ eventOutbox: outboxConfig });
+  const run = store.createRun(runInput("ses_user_message_http_authority"));
+  const server = await startManagerServer({ port: 0, host: "127.0.0.1", sourceCommit: "self-test", store });
+  try {
+    const client = new ManagerClient(server.baseUrl);
+    await assert.rejects(
+      () => client.post(`/api/v1/runs/${run.id}/events`, { type: "user_message", payload: { commandId: "cmd_forged", userMessageId: "msg_forged", text: "forged", source: "runner" } }),
+      (error) => error instanceof AgentRunError && error.failureKind === "tenant-policy-denied" && error.httpStatus === 403,
+    );
+    assert.equal(store.listEvents(run.id, 0, 100).filter((event) => event.type === "user_message").length, 0);
+  } finally {
+    await new Promise<void>((resolve) => server.server.close(() => resolve()));
+  }
+}
+
+function assertCommandCreationIsAtomic(): void {
+  const store = new RejectCommandLifecycleStore({ eventOutbox: outboxConfig });
+  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);
+}
+
+class RejectCommandLifecycleStore extends MemoryAgentRunStore {
+  protected override beforeEventCommit(event: RunEvent): void {
+    if (event.payload.phase === "command-created") throw new Error("reject command lifecycle");
+  }
+}
+
+function drainOutbox(store: MemoryAgentRunStore): KafkaEventOutboxRecord[] {
+  const items: KafkaEventOutboxRecord[] = [];
+  while (true) {
+    const item = store.claimKafkaEventOutbox({ owner: "user-message-selftest", leaseMs: 60_000, limit: 100 })[0];
+    if (!item) return items;
+    items.push(item);
+    store.completeKafkaEventOutbox(item);
+  }
+}
+
+function runInput(sessionId: string): CreateRunInput {
+  return {
+    tenantId: "unidesk",
+    projectId: "pikasTech/HWLAB",
+    workspaceRef: { kind: "host-path", path: "/tmp/agentrun-user-message-selftest" },
+    sessionRef: { sessionId, metadata: { hwlabSessionId: sessionId, hwlabTraceId: `trc_${sessionId}` } },
+    resourceBundleRef: null,
+    providerId: "NC01",
+    backendProfile: "codex",
+    executionPolicy: { sandbox: "workspace-write", approval: "never", timeoutMs: 60_000, network: "default", secretScope: { allowCredentialEcho: false, providerCredentials: [] } },
+    traceSink: null,
+  };
+}
+
+export default selfTest;

src/selftest/cases/50-hwlab-manual-dispatch.ts

@@ -268,10 +268,10 @@ process.exit(1);
     assert.equal(secondEnvelope.terminalStatus, "completed");
     assert.equal(secondEnvelope.reply, "fake codex stdio reply");
 
-    const steerRun = await createHwlabRun(client, context, bundle, "hwlab-session-steer", "start a turn that waits for steer", "hwlab-command-steer-turn", 10_000);
+    const steerRun = await createHwlabRun(client, context, bundle, "hwlab-session-steer", "start a turn that waits for steer", "trc_hwlab_command_steer_turn", 10_000);
     const steerRunner = runOnce({ managerUrl: server.baseUrl, runId: steerRun.runId, commandId: steerRun.commandId, codexCommand: context.fakeCodexCommand, codexArgs: context.fakeCodexArgs, codexHome: context.codexHome, env: { CODEX_HOME: context.codexHome, AGENTRUN_FAKE_CODEX_MODE: "steer-waits", AGENTRUN_WORKSPACE_ROOT: path.join(context.tmp, "workspaces-steer") }, oneShot: true, pollIntervalMs: 50 });
     await waitForCommandState(client, steerRun.runId, steerRun.commandId, "acknowledged");
-    const steerCommand = await client.post(`/api/v1/runs/${steerRun.runId}/commands`, { type: "steer", payload: { prompt: "STEER_MARK_SELFTEST", traceId: "hwlab-command-steer" }, idempotencyKey: "hwlab-command-steer" }) as { id: string };
+    const steerCommand = await client.post(`/api/v1/runs/${steerRun.runId}/commands`, { type: "steer", payload: { prompt: "STEER_MARK_SELFTEST", traceId: "trc_hwlab_command_steer", targetTraceId: "trc_hwlab_command_steer_turn" }, idempotencyKey: "trc_hwlab_command_steer" }) as { id: string };
     await waitForCommandState(client, steerRun.runId, steerCommand.id, "completed");
     const steerRunnerResult = await steerRunner as JsonRecord;
     assert.equal(steerRunnerResult.terminalStatus, "completed");
@@ -287,6 +287,7 @@ process.exit(1);
     assert.match(String(steerDelivery.semantics), /target command liveness/u);
     const steerEventsResponse = await client.get(`/api/v1/runs/${steerRun.runId}/events?afterSeq=0&limit=200`) as { items?: Array<{ type?: string; payload?: JsonRecord }> };
     const steerEvents = steerEventsResponse.items ?? [];
+    assert.ok(steerEvents.some((event) => event.type === "user_message" && event.payload?.commandId === steerCommand.id && event.payload?.traceId === "trc_hwlab_command_steer" && event.payload?.targetTraceId === "trc_hwlab_command_steer_turn"));
     assert.ok(steerEvents.some((event) => event.type === "backend_status" && event.payload?.phase === "steer-command-acknowledged" && event.payload?.commandId === steerCommand.id && event.payload?.targetCommandId === steerRun.commandId));
     assert.ok(steerEvents.some((event) => event.type === "backend_status" && event.payload?.phase === "turn/steer:completed" && event.payload?.commandId === steerCommand.id && event.payload?.targetCommandId === steerRun.commandId && event.payload.deliveryState === "forwarded-to-backend" && event.payload.targetEffect === "not-guaranteed"));
 
@@ -323,7 +324,7 @@ process.exit(1);
     const runningResult = await running;
     assert.equal(runningResult.terminalStatus, "cancelled");
 
-    return { name: "hwlab-manual-dispatch", tests: ["runner-job-idempotency", "pending-cancel", "result-envelope", "session-ref-resume", "resource-gitbundle-materialization", "gitbundle-ref-resolution", "gitbundle-tools-path", "gitbundle-skill-dir-assembly", "resource-prompt-required-blocker", "resource-required-skill-blocker", "same-run-runner-multiturn", "running-steer", "no-event-watchdog-after-tool", "tool-output-hard-timeout", "running-cancel"] };
+    return { name: "hwlab-manual-dispatch", tests: ["runner-job-idempotency", "pending-cancel", "result-envelope", "session-ref-resume", "resource-gitbundle-materialization", "gitbundle-ref-resolution", "gitbundle-tools-path", "gitbundle-skill-dir-assembly", "resource-prompt-required-blocker", "resource-required-skill-blocker", "same-run-runner-multiturn", "running-steer", "running-steer-user-message-target-trace", "no-event-watchdog-after-tool", "tool-output-hard-timeout", "running-cancel"] };
   } finally {
     await new Promise<void>((resolve) => server.server.close(() => resolve()));
   }

src/selftest/cases/55-timeout-liveness.ts

@@ -19,7 +19,7 @@ const selfTest: SelfTestCase = async (context: SelfTestContext) => {
     const toolLive = toolResult.liveness as JsonRecord;
     assert.equal(toolLive.phase, "waiting-tool");
     assert.equal(((toolLive.lastActivity as JsonRecord).activityKind), "tool-in-flight");
-    assert.equal(((toolLive.lastActivity as JsonRecord).sourceSeq), 4);
+    assert.equal(((toolLive.lastActivity as JsonRecord).sourceSeq), 5);
     assert.equal(((toolLive.timeoutBudget as JsonRecord).state), "within-budget");
     assert.ok(Array.isArray(toolLive.recoveryActions));
 
@@ -88,7 +88,7 @@ const selfTest: SelfTestCase = async (context: SelfTestContext) => {
     assert.equal(((noSessionDiagnosis.session as JsonRecord).sessionRefNull), true);
     assert.ok((noSessionDiagnosis.recoveryActions as JsonRecord[]).some((action) => action.action === "session-unavailable"));
     assert.match(String(noSessionClassification.providerInterruptionReason), /cannot distinguish provider outage/u);
-    assert.equal((noSessionLive.transportDisconnect as JsonRecord).sourceSeq, 4);
+    assert.equal((noSessionLive.transportDisconnect as JsonRecord).sourceSeq, 5);
     assert.equal((noSessionLive.recoveryActions as JsonRecord[]).some((action) => action.action === "continue-session"), false, "sessionId=null must not suggest session-only continuation");
     assert.equal((noSessionLive.recoveryActions as JsonRecord[]).some((action) => action.action === "poll-output"), false, "sessionId=null must not suggest session output path");
     assert.ok((noSessionLive.recoveryActions as JsonRecord[]).some((action) => action.action === "poll-trace" && action.operation === "events" && action.resourceKind === "run" && action.resourceName === noSession.runId));

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

@@ -57,6 +57,47 @@ try {
   assert.equal(marker(second), "B");
   await store.completeKafkaEventOutbox(second);
 
+  const userRun = await store.createRun(runInput("ses_pg_user_message_order"));
+  const userCommandInput = validateCreateCommand({
+    type: "turn",
+    payload: { prompt: "postgres prompt sk-12345678", traceId: "trc_pg_user_message", userMessageId: "msg_pg_user_message" },
+    idempotencyKey: "pg-user-message-order",
+  });
+  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, {
+    commandId: userCommand.id,
+    traceId: "trc_pg_user_message",
+    userMessageId: "msg_pg_user_message",
+    text: "postgres prompt REDACTED",
+    source: "agentrun-turn-command-payload",
+    valuesPrinted: false,
+  });
+  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.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"]);
+
+  const steerCommandInput = validateCreateCommand({
+    type: "steer",
+    payload: { message: "postgres steer", traceId: "trc_pg_user_message_steer", targetTraceId: "trc_pg_user_message", userMessageId: "msg_pg_user_message_steer" },
+    idempotencyKey: "pg-user-message-steer-order",
+  });
+  const steerCommand = await store.createCommand(userRun.id, steerCommandInput);
+  assert.equal((await store.createCommand(userRun.id, steerCommandInput)).id, steerCommand.id);
+  const steerEvents = await store.listEvents(userRun.id, 0, 100);
+  const steerUserEvent = steerEvents.find((event) => event.type === "user_message" && event.payload.commandId === steerCommand.id);
+  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.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 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();
@@ -77,7 +118,7 @@ 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, 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, dispatchParentLockFenced: true, valuesPrinted: false }));
 } finally {
   if (gateClient && gateHeld) await gateClient.query("SELECT pg_advisory_unlock($1::bigint)", [gate]).catch(() => undefined);
   gateClient?.release();
@@ -154,6 +195,17 @@ async function markerRows(pool: Pool): Promise<Array<{ outboxSeq: number; marker
   return result.rows.map((row) => ({ outboxSeq: Number(row.outbox_seq), marker: row.marker, partitionKey: row.partition_key }));
 }
 
+async function userMessageOutboxRows(pool: Pool, runId: string): Promise<Array<{ sourceSeq: number; eventType: string; partitionKey: string; targetTraceId: string | null }>> {
+  const result = await pool.query<{ source_seq: string | number; event_type: string; partition_key: string; target_trace_id: string | null }>(`
+    SELECT source_seq, value #>> '{event,type}' AS event_type, partition_key,
+           value #>> '{event,payload,targetTraceId}' AS target_trace_id
+    FROM agentrun_kafka_event_outbox
+    WHERE run_id = $1
+    ORDER BY source_seq ASC
+  `, [runId]);
+  return result.rows.map((row) => ({ sourceSeq: Number(row.source_seq), eventType: row.event_type, partitionKey: row.partition_key, targetTraceId: row.target_trace_id }));
+}
+
 function singleClaim(items: KafkaEventOutboxRecord[]): KafkaEventOutboxRecord {
   assert.equal(items.length, 1);
   return items[0] as KafkaEventOutboxRecord;

src/selftest/run.ts

@@ -63,6 +63,11 @@ try {
 
   console.log(JSON.stringify({ ok: true, selectedCases: selectedCaseFiles.map(caseName), cases: results, tests: results.flatMap((result) => result.tests) }));
 } catch (error) {
-  console.error(JSON.stringify({ ok: false, error: error instanceof Error ? error.message : String(error) }));
+  console.error(JSON.stringify({
+    ok: false,
+    error: error instanceof Error ? error.message : String(error),
+    errorName: error instanceof Error ? error.name : "UnknownError",
+    stack: error instanceof Error ? error.stack ?? null : null,
+  }));
   process.exit(1);
 }