From 778bbfdf3081d8dd38ce1cc3b45298678ff30c04 Mon Sep 17 00:00:00 2001 From: root Date: Sat, 11 Jul 2026 20:12:18 +0200 Subject: [PATCH] =?UTF-8?q?fix:=20=E4=BF=9D=E8=AF=81=E9=A6=96=E6=9D=A1?= =?UTF-8?q?=E7=94=A8=E6=88=B7=E4=BA=8B=E4=BB=B6=E5=85=88=E4=BA=8E=E8=BF=90?= =?UTF-8?q?=E8=A1=8C=E5=88=9B=E5=BB=BA=E4=BA=8B=E4=BB=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/mgr/postgres-store.ts | 14 +++- src/mgr/store.ts | 24 +++++-- .../22-runner-retention-stale-pending.ts | 4 +- src/selftest/cases/35-kafka-durable-outbox.ts | 9 ++- src/selftest/cases/37-user-message-event.ts | 56 +++++++++++++--- .../integration/postgres-kafka-order.ts | 67 +++++++++++++++++-- 6 files changed, 149 insertions(+), 25 deletions(-) diff --git a/src/mgr/postgres-store.ts b/src/mgr/postgres-store.ts index 072489c..5d05a86 100644 --- a/src/mgr/postgres-store.ts +++ b/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); }); diff --git a/src/mgr/store.ts b/src/mgr/store.ts index 99c9199..a594d98 100644 --- a/src/mgr/store.ts +++ b/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); diff --git a/src/selftest/cases/22-runner-retention-stale-pending.ts b/src/selftest/cases/22-runner-retention-stale-pending.ts index d7e90ef..1cd13ba 100644 --- a/src/selftest/cases/22-runner-retention-stale-pending.ts +++ b/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 }), diff --git a/src/selftest/cases/35-kafka-durable-outbox.ts b/src/selftest/cases/35-kafka-durable-outbox.ts index e439c34..e798077 100644 --- a/src/selftest/cases/35-kafka-durable-outbox.ts +++ b/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 { 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); diff --git a/src/selftest/cases/37-user-message-event.ts b/src/selftest/cases/37-user-message-event.ts index eebf572..f0b3151 100644 --- a/src/selftest/cases/37-user-message-event.ts +++ b/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"); + } } } diff --git a/src/selftest/integration/postgres-kafka-order.ts b/src/selftest/integration/postgres-kafka-order.ts index 2326d7d..a0db713 100644 --- a/src/selftest/integration/postgres-kafka-order.ts +++ b/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 { 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 { + 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 { + 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 { const deadline = Date.now() + 5_000; while (Date.now() < deadline) {