From c74aaed8487e6309150f3125333318654c612915 Mon Sep 17 00:00:00 2001 From: root Date: Sat, 11 Jul 2026 04:26:07 +0200 Subject: [PATCH] =?UTF-8?q?feat:=20=E5=8F=91=E5=B8=83=E7=94=A8=E6=88=B7?= =?UTF-8?q?=E8=BE=93=E5=85=A5=20Kafka=20=E6=AD=A3=E5=BC=8F=E4=BA=8B?= =?UTF-8?q?=E4=BB=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- scripts/src/cli.ts | 9 +- src/common/events.ts | 55 ++++++- src/common/types.ts | 2 +- src/mgr/postgres-store.ts | 5 +- src/mgr/runner-reconciler.ts | 3 +- src/mgr/store.ts | 26 +++- src/selftest/cases/15-cli-events-summary.ts | 10 +- src/selftest/cases/35-kafka-durable-outbox.ts | 22 +-- src/selftest/cases/37-user-message-event.ts | 137 ++++++++++++++++++ .../cases/50-hwlab-manual-dispatch.ts | 7 +- src/selftest/cases/55-timeout-liveness.ts | 4 +- .../integration/postgres-kafka-order.ts | 54 ++++++- src/selftest/run.ts | 7 +- 13 files changed, 308 insertions(+), 33 deletions(-) create mode 100644 src/selftest/cases/37-user-message-event.ts diff --git a/scripts/src/cli.ts b/scripts/src/cli.ts index f928153..ebdf2a5 100644 --- a/scripts/src/cli.ts +++ b/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 or --item-id --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 { diff --git a/src/common/events.ts b/src/common/events.ts index 53833a7..6ed780e 100644 --- a/src/common/events.ts +++ b/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(eventTypes); +const externallyAppendableEventTypeSet = new Set(externallyAppendableEventTypes); const terminalStatusSet = new Set(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 : ""; diff --git a/src/common/types.ts b/src/common/types.ts index fe1ad40..5ad5386 100644 --- a/src/common/types.ts +++ b/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; diff --git a/src/mgr/postgres-store.ts b/src/mgr/postgres-store.ts index d04625b..a222087 100644 --- a/src/mgr/postgres-store.ts +++ b/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 { + type = requireExternallyAppendableEventType(type); return this.withTransaction(async (client) => { await this.requireRunForUpdate(client, runId); return this.appendEventWithLockedRun(client, runId, type, payload); diff --git a/src/mgr/runner-reconciler.ts b/src/mgr/runner-reconciler.ts index 1d6721e..3994d21 100644 --- a/src/mgr/runner-reconciler.ts +++ b/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(["backend_status", "assistant_message", "tool_call", "command_output", "diff", "error", "terminal_status"]); +const eventTypes = new Set(externallyAppendableEventTypes); const terminalStatuses = new Set(["completed", "failed", "blocked", "cancelled"]); const failureKinds = new Set([ "auth-missing", diff --git a/src/mgr/store.ts b/src/mgr/store.ts index 9186d71..64ffbc9 100644 --- a/src/mgr/store.ts +++ b/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 }; diff --git a/src/selftest/cases/15-cli-events-summary.ts b/src/selftest/cases/15-cli-events-summary.ts index 7ccff5e..b0edbbe 100644 --- a/src/selftest/cases/15-cli-events-summary.ts +++ b/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"] }; } diff --git a/src/selftest/cases/35-kafka-durable-outbox.ts b/src/selftest/cases/35-kafka-durable-outbox.ts index cc85449..e439c34 100644 --- a/src/selftest/cases/35-kafka-durable-outbox.ts +++ b/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 => { 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", diff --git a/src/selftest/cases/37-user-message-event.ts b/src/selftest/cases/37-user-message-event.ts new file mode 100644 index 0000000..eebf572 --- /dev/null +++ b/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 { + 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((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; diff --git a/src/selftest/cases/50-hwlab-manual-dispatch.ts b/src/selftest/cases/50-hwlab-manual-dispatch.ts index f4c2d2c..94b7b3e 100644 --- a/src/selftest/cases/50-hwlab-manual-dispatch.ts +++ b/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((resolve) => server.server.close(() => resolve())); } diff --git a/src/selftest/cases/55-timeout-liveness.ts b/src/selftest/cases/55-timeout-liveness.ts index af585f5..88d3a33 100644 --- a/src/selftest/cases/55-timeout-liveness.ts +++ b/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)); diff --git a/src/selftest/integration/postgres-kafka-order.ts b/src/selftest/integration/postgres-kafka-order.ts index a06452a..2326d7d 100644 --- a/src/selftest/integration/postgres-kafka-order.ts +++ b/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 ({ outboxSeq: Number(row.outbox_seq), marker: row.marker, partitionKey: row.partition_key })); } +async function userMessageOutboxRows(pool: Pool, runId: string): Promise> { + 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; diff --git a/src/selftest/run.ts b/src/selftest/run.ts index 0496163..fe16f0c 100644 --- a/src/selftest/run.ts +++ b/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); }