From 88f39e70f690ef35257a25a96ba737592c9a857a Mon Sep 17 00:00:00 2001 From: root Date: Sun, 12 Jul 2026 03:01:18 +0200 Subject: [PATCH] =?UTF-8?q?test:=20=E5=9B=BA=E5=8C=96=E7=BB=88=E6=80=81=20?= =?UTF-8?q?command=20=E5=B9=82=E7=AD=89=E9=87=8D=E6=94=BE=E5=90=88?= =?UTF-8?q?=E5=90=8C?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/selftest/cases/35-kafka-durable-outbox.ts | 29 +++++++++++++++++-- 1 file changed, 27 insertions(+), 2 deletions(-) diff --git a/src/selftest/cases/35-kafka-durable-outbox.ts b/src/selftest/cases/35-kafka-durable-outbox.ts index e798077..b69b9fe 100644 --- a/src/selftest/cases/35-kafka-durable-outbox.ts +++ b/src/selftest/cases/35-kafka-durable-outbox.ts @@ -142,11 +142,36 @@ const selfTest: SelfTestCase = async () => { assert.equal(store.getRun(run.id).status, "failed"); assert.equal(store.getSessionSummary("ses_kafka_durable").executionState, "terminal"); assert.deepEqual(store.listEvents(run.id, 0, 100).slice(-3).map((event) => event.payload.phase ?? event.type), ["runner-dispatch-failed", "command-terminal", "terminal_status"]); - assert.throws(() => store.createCommand(run.id, commandInput), (error) => error instanceof AgentRunError && error.httpStatus !== 200); + const terminalEventCount = store.listEvents(run.id, 0, 100).length; + const terminalOutboxCount = Number(store.kafkaEventOutboxStatus().totalCount); + const terminalReplay = store.createCommand(run.id, commandInput); + const terminalIntent = store.getRunnerDispatchIntent(command.id); + assert.equal(terminalReplay.id, command.id); + assert.equal(terminalReplay.dispatchIntent?.id, command.dispatchIntent?.id); + assert.equal(terminalReplay.dispatchIntent?.state, terminalIntent?.state); + assert.equal(terminalReplay.dispatchIntent?.runnerJobId, terminalIntent?.runnerJobId); + assert.equal(terminalReplay.dispatchIntent?.attemptCount, terminalIntent?.attemptCount); + assert.equal(store.listEvents(run.id, 0, 100).length, terminalEventCount); + assert.equal(Number(store.kafkaEventOutboxStatus().totalCount), terminalOutboxCount); + const changedPayloadInput = validateCreateCommand({ + ...commandInput, + payload: { prompt: "changed payload must conflict", traceId: "trc_durable" }, + }); + assert.throws( + () => store.createCommand(run.id, changedPayloadInput), + (error) => error instanceof AgentRunError && error.httpStatus === 409 && /different payload/u.test(error.message), + ); + const differentKeyInput = validateCreateCommand({ ...commandInput, idempotencyKey: "command-durable-terminal-new-key" }); + assert.throws( + () => store.createCommand(run.id, differentKeyInput), + (error) => error instanceof AgentRunError && error.httpStatus === 409 && /already terminal/u.test(error.message), + ); + assert.equal(store.listEvents(run.id, 0, 100).length, terminalEventCount); + assert.equal(Number(store.kafkaEventOutboxStatus().totalCount), terminalOutboxCount); await assertStaleClaimsCannotOverwrite(); await assertProducerRecreatedAfterDisconnected(); - return { name: "kafka-durable-outbox", tests: ["explicit-enable", "canonical-event", "warm-run-command-trace-authority", "run-level-trace-fallback", "stdio-command-trace-context", "kafka-tail-correlation-summary", "durable-head-of-line", "atomic-dispatch-terminal", "stale-claim-fencing", "producer-reconnect"] }; + return { name: "kafka-durable-outbox", tests: ["explicit-enable", "canonical-event", "warm-run-command-trace-authority", "run-level-trace-fallback", "stdio-command-trace-context", "kafka-tail-correlation-summary", "durable-head-of-line", "atomic-dispatch-terminal", "terminal-command-idempotent-replay", "terminal-command-replay-no-new-events-or-outbox", "terminal-command-same-key-payload-conflict", "terminal-command-different-key-fails-closed", "stale-claim-fencing", "producer-reconnect"] }; }; function assertWarmRunCommandTraceAuthority(): void {