root
245 lines
14 KiB
TypeScript
245 lines
14 KiB
TypeScript
import assert from "node:assert/strict";
|
|
import { test } from "bun:test";
|
|
|
|
import { buildWorkbenchProjectionEventFacts } from "../cloud/workbench-projection-writer.ts";
|
|
import { projectionOutboxRealtimeEvents } from "../cloud/server-workbench-realtime-http.ts";
|
|
import {
|
|
claimHwlabKafkaOutbox,
|
|
commitAgentRunKafkaProjection,
|
|
completeHwlabKafkaOutbox,
|
|
projectionOutboxEventId,
|
|
recordFailedAgentRunKafkaMessage,
|
|
recordIgnoredAgentRunKafkaMessage,
|
|
retryHwlabKafkaOutbox
|
|
} from "./runtime-store-postgres-kafka.ts";
|
|
|
|
test("malformed duplicate source event at a new offset is durably DLQed without mutating projected inbox", async () => {
|
|
const queries = [];
|
|
const client = {
|
|
async query(sql, params) {
|
|
queries.push({ sql, params });
|
|
if (sql.startsWith("SELECT pg_advisory_xact_lock")) return { rows: [] };
|
|
if (sql.startsWith("SELECT source_topic")) return { rows: [{ source_topic: "agentrun.event.v1", source_partition: 0, source_offset: "10", input_sha256: "prior-hash", status: "projected", projected_seq: 7 }] };
|
|
if (sql.startsWith("INSERT INTO workbench_kafka_dlq")) return { rows: [] };
|
|
throw new Error(`unexpected SQL: ${sql}`);
|
|
}
|
|
};
|
|
const store = transactionStore(client);
|
|
const result = await recordFailedAgentRunKafkaMessage(store, {
|
|
transport: { sourceTopic: "agentrun.event.v1", sourcePartition: 0, sourceOffset: "11", sourceKey: "run-1", inputSha256: "new-hash" },
|
|
sourceEventId: "evt-duplicate",
|
|
envelope: { eventId: "evt-duplicate", valuesPrinted: false },
|
|
error: { code: "workbench_kafka_schema_invalid", message: "malformed" }
|
|
});
|
|
|
|
assert.equal(result.recorded, true);
|
|
assert.equal(result.conflict, true);
|
|
assert.equal(result.priorStatus, "projected");
|
|
assert.equal(queries[0].params[0], "workbench-kafka-inbox:agentrun.event.v1:transport:0:11");
|
|
assert.equal(queries[1].params[0], "workbench-kafka-inbox:agentrun.event.v1:event:evt-duplicate");
|
|
assert.equal(queries.some(({ sql }) => sql.startsWith("UPDATE workbench_kafka_inbox")), false);
|
|
assert.equal(queries.some(({ sql }) => sql.startsWith("INSERT INTO workbench_kafka_dlq")), true);
|
|
});
|
|
|
|
test("ignored Kafka messages acquire transport then event identity locks", async () => {
|
|
const queries = [];
|
|
const client = {
|
|
async query(sql, params) {
|
|
queries.push({ sql, params });
|
|
if (sql.startsWith("SELECT pg_advisory_xact_lock")) return { rows: [] };
|
|
if (sql.startsWith("SELECT input_sha256")) return { rows: [] };
|
|
if (sql.startsWith("INSERT INTO workbench_kafka_inbox")) return { rows: [] };
|
|
throw new Error(`unexpected SQL: ${sql}`);
|
|
}
|
|
};
|
|
const result = await recordIgnoredAgentRunKafkaMessage(transactionStore(client), {
|
|
transport: { sourceTopic: "agentrun.event.v1", sourcePartition: 2, sourceOffset: "31", sourceKey: "run-ignored", inputSha256: "hash-ignored" },
|
|
envelope: { eventId: "evt-ignored", sourceSeq: 1, runId: "run-ignored" },
|
|
reason: "hwlab-correlation-absent"
|
|
});
|
|
|
|
assert.equal(result.ignored, true);
|
|
assert.equal(result.duplicate, false);
|
|
assert.equal(queries[0].params[0], "workbench-kafka-inbox:agentrun.event.v1:transport:2:31");
|
|
assert.equal(queries[1].params[0], "workbench-kafka-inbox:agentrun.event.v1:event:evt-ignored");
|
|
});
|
|
|
|
test("projection outbox derives a stable non-null identity when a fact has no source event", () => {
|
|
const input = {
|
|
fact: { traceId: "trc-derived", projectedSeq: 3, sourceSeq: 2 },
|
|
event: { aggregateId: "trc-derived", aggregateSeq: 3, projectionRevision: 3 },
|
|
family: "traceEvents",
|
|
entityId: "wte-derived",
|
|
commitType: "event"
|
|
};
|
|
const first = projectionOutboxEventId(input);
|
|
assert.match(first, /^projection:[a-f0-9]{64}$/u);
|
|
assert.equal(projectionOutboxEventId(input), first);
|
|
assert.notEqual(projectionOutboxEventId({ ...input, event: { ...input.event, projectionRevision: 4 } }), first);
|
|
});
|
|
|
|
test("Kafka projection preserves admission owner and scope fields", async () => {
|
|
let persistedSession = null;
|
|
const client = {
|
|
async query(sql) {
|
|
if (sql.startsWith("SELECT pg_advisory_xact_lock")) return { rows: [] };
|
|
if (sql.startsWith("SELECT source_topic")) return { rows: [] };
|
|
if (sql.startsWith("INSERT INTO workbench_kafka_inbox")) return { rows: [] };
|
|
if (sql.startsWith("SELECT checkpoint_json")) return { rows: [] };
|
|
if (sql.startsWith("SELECT owner_user_id")) return { rows: [{ owner_user_id: "usr-owner", project_id: "prj-owner", conversation_id: "cnv-owner", thread_id: "thread-owner", session_json: { sessionId: "ses-owner", ownerUserId: "usr-owner", ownerRole: "user", launchContext: { source: "workbench" } } }] };
|
|
if (sql.startsWith("SELECT GREATEST")) return { rows: [{ max_projected_seq: 0 }] };
|
|
if (sql.startsWith("INSERT INTO hwlab_kafka_outbox")) return { rows: [] };
|
|
if (sql.startsWith("UPDATE workbench_kafka_inbox")) return { rows: [] };
|
|
throw new Error(`unexpected SQL: ${sql}`);
|
|
}
|
|
};
|
|
const store = {
|
|
...transactionStore(client),
|
|
memory: { writeWorkbenchFacts() {} },
|
|
async persistWorkbenchAggregateEvent(event) { return { ...event, eventSeq: 1, aggregateSeq: 1, projectionRevision: 1 }; },
|
|
async persistWorkbenchSessionInputFact() {},
|
|
async persistWorkbenchSessionFact(fact) { persistedSession = fact; },
|
|
async persistWorkbenchMessageFact() {},
|
|
async persistWorkbenchPartFact() {},
|
|
async persistWorkbenchTurnFact() {},
|
|
async persistWorkbenchTraceEventFact() {},
|
|
async persistWorkbenchProjectionCheckpoint() {},
|
|
async persistWorkbenchProjectionOutbox() {}
|
|
};
|
|
await commitAgentRunKafkaProjection(store, {
|
|
transport: { sourceTopic: "agentrun.event.v1", sourcePartition: 0, sourceOffset: "12", sourceKey: "run-owner", inputSha256: "hash-owner" },
|
|
canonicalEvent: { eventId: "evt-owner", sourceSeq: 1, traceId: "trc-owner", hwlabSessionId: "ses-owner", runId: "run-owner", commandId: "cmd-owner" },
|
|
projectedEvent: { traceId: "trc-owner", sessionId: "ses-owner" },
|
|
factsFactory: ({ projectedSeq, projectedAt }) => ({
|
|
written: true,
|
|
facts: {
|
|
sessions: [{ sessionId: "ses-owner", status: "running", lastTraceId: "trc-owner", projectedSeq, sourceSeq: 1, sourceEventId: "evt-owner", terminal: false, sealed: false, updatedAt: projectedAt }],
|
|
messages: [], parts: [], turns: [], traceEvents: [], checkpoints: []
|
|
}
|
|
}),
|
|
hwlabEvent: { eventId: "hwlab:evt-owner" },
|
|
hwlabTopic: "hwlab.event.v1",
|
|
partitionKey: "trc-owner",
|
|
headers: {}
|
|
});
|
|
|
|
assert.equal(persistedSession.ownerUserId, "usr-owner");
|
|
assert.equal(persistedSession.ownerRole, "user");
|
|
assert.equal(persistedSession.projectId, "prj-owner");
|
|
assert.equal(persistedSession.conversationId, "cnv-owner");
|
|
assert.equal(persistedSession.threadId, "thread-owner");
|
|
assert.deepEqual(persistedSession.launchContext, { source: "workbench" });
|
|
});
|
|
|
|
test("terminal Kafka transaction persists sealed turn and immutable SSE snapshot", async () => {
|
|
let persistedTurn = null;
|
|
const outbox = [];
|
|
const client = {
|
|
async query(sql) {
|
|
if (sql.startsWith("SELECT pg_advisory_xact_lock")) return { rows: [] };
|
|
if (sql.startsWith("SELECT source_topic")) return { rows: [] };
|
|
if (sql.startsWith("INSERT INTO workbench_kafka_inbox")) return { rows: [] };
|
|
if (sql.startsWith("SELECT checkpoint_json")) return { rows: [{ checkpoint_json: { traceId: "trc_terminal", sessionId: "ses_terminal", assistantText: "final body", finalResponse: { text: "final body", status: "running" }, projectedSeq: 2, sourceSeq: 2, terminal: false, sealed: false } }] };
|
|
if (sql.startsWith("SELECT owner_user_id")) return { rows: [{ owner_user_id: "usr-owner", session_json: { sessionId: "ses_terminal", ownerUserId: "usr-owner" } }] };
|
|
if (sql.startsWith("SELECT GREATEST")) return { rows: [{ max_projected_seq: 2 }] };
|
|
if (sql.startsWith("INSERT INTO hwlab_kafka_outbox") || sql.startsWith("UPDATE workbench_kafka_inbox")) return { rows: [] };
|
|
throw new Error(`unexpected SQL: ${sql}`);
|
|
}
|
|
};
|
|
const store = {
|
|
...transactionStore(client),
|
|
memory: { writeWorkbenchFacts() {} },
|
|
async persistWorkbenchAggregateEvent(event) { return { ...event, eventSeq: 3, aggregateSeq: 3, projectionRevision: 3 }; },
|
|
async persistWorkbenchSessionInputFact() {}, async persistWorkbenchSessionFact() {}, async persistWorkbenchMessageFact() {}, async persistWorkbenchPartFact() {},
|
|
async persistWorkbenchTurnFact(fact) { persistedTurn = fact; }, async persistWorkbenchTraceEventFact() {}, async persistWorkbenchProjectionCheckpoint() {},
|
|
async persistWorkbenchProjectionOutbox(record) { outbox.push(record); }
|
|
};
|
|
await commitAgentRunKafkaProjection(store, {
|
|
transport: { sourceTopic: "agentrun.event.v1", sourcePartition: 0, sourceOffset: "20", inputSha256: "hash-terminal" },
|
|
canonicalEvent: { eventId: "evt-terminal", sourceSeq: 3, traceId: "trc_terminal", hwlabSessionId: "ses_terminal", runId: "run-terminal", commandId: "cmd-terminal" },
|
|
projectedEvent: { traceId: "trc_terminal", sessionId: "ses_terminal", sourceEventId: "evt-terminal", sourceSeq: 3, type: "result", eventType: "terminal", status: "completed", terminal: true, createdAt: "2026-07-10T10:00:03.000Z" },
|
|
factsFactory: ({ projectedSeq, previousCheckpoint, projectedAt }) => buildWorkbenchProjectionEventFacts({ projectedSeq, previousCheckpoint, projectedAt, event: { traceId: "trc_terminal", sessionId: "ses_terminal", sourceEventId: "evt-terminal", sourceSeq: 3, type: "result", eventType: "terminal", status: "completed", terminal: true, createdAt: "2026-07-10T10:00:03.000Z" } }),
|
|
hwlabEvent: { eventId: "hwlab:evt-terminal" }, hwlabTopic: "hwlab.event.v1", partitionKey: "trc_terminal", headers: {}
|
|
});
|
|
|
|
assert.equal(persistedTurn.terminal, true);
|
|
assert.equal(persistedTurn.sealed, true);
|
|
assert.equal(persistedTurn.finalResponse.text, "final body");
|
|
const turnRow = outbox.find((row) => row.entityFamily === "turns");
|
|
assert.equal(turnRow.payload.fact.assistantText, "final body");
|
|
const [sse] = projectionOutboxRealtimeEvents({ events: [{ ...turnRow, outboxSeq: 9 }], facts: {} });
|
|
assert.equal(sse.name, "workbench.turn.snapshot");
|
|
assert.equal(sse.payload.turn.finalResponse.text, "final body");
|
|
});
|
|
|
|
test("late second terminal is acknowledged without facts or either projection outbox", async () => {
|
|
const queries = [];
|
|
const client = {
|
|
async query(sql, params) {
|
|
queries.push({ sql, params });
|
|
if (sql.startsWith("SELECT pg_advisory_xact_lock")) return { rows: [] };
|
|
if (sql.startsWith("SELECT source_topic")) return { rows: [] };
|
|
if (sql.startsWith("INSERT INTO workbench_kafka_inbox")) return { rows: [] };
|
|
if (sql.startsWith("SELECT checkpoint_json")) return { rows: [{ checkpoint_json: { traceId: "trc_sealed", projectedSeq: 9, terminal: true, sealed: true } }] };
|
|
if (sql.startsWith("SELECT owner_user_id")) return { rows: [] };
|
|
if (sql.startsWith("SELECT GREATEST")) return { rows: [{ max_projected_seq: 9 }] };
|
|
if (sql.startsWith("INSERT INTO hwlab_kafka_outbox") || sql.startsWith("UPDATE workbench_kafka_inbox")) return { rows: [] };
|
|
throw new Error(`unexpected SQL: ${sql}`);
|
|
}
|
|
};
|
|
const store = { ...transactionStore(client), memory: { writeWorkbenchFacts() { throw new Error("suppressed facts must not reach memory"); } } };
|
|
const result = await commitAgentRunKafkaProjection(store, {
|
|
transport: { sourceTopic: "agentrun.event.v1", sourcePartition: 0, sourceOffset: "21", inputSha256: "hash-late" },
|
|
canonicalEvent: { eventId: "evt-late", sourceSeq: 10, traceId: "trc_sealed", hwlabSessionId: "ses_sealed" },
|
|
projectedEvent: { traceId: "trc_sealed", sessionId: "ses_sealed" },
|
|
factsFactory: ({ projectedSeq, previousCheckpoint, projectedAt }) => buildWorkbenchProjectionEventFacts({
|
|
projectedSeq,
|
|
previousCheckpoint,
|
|
projectedAt,
|
|
event: { traceId: "trc_sealed", sessionId: "ses_sealed", sourceEventId: "evt-late", sourceSeq: 10, type: "result", eventType: "terminal", status: "failed", terminal: true }
|
|
}),
|
|
hwlabEvent: { eventId: "hwlab:evt-late" },
|
|
hwlabTopic: "hwlab.event.v1",
|
|
partitionKey: "trc_sealed",
|
|
headers: {}
|
|
});
|
|
|
|
assert.equal(result.suppressedAfterSeal, true);
|
|
assert.equal(result.projectionWritten, false);
|
|
assert.equal(result.projectedSeq, 9);
|
|
assert.equal(queries.some(({ sql }) => sql.startsWith("INSERT INTO hwlab_kafka_outbox")), false);
|
|
const inboxUpdate = queries.find(({ sql }) => sql.startsWith("UPDATE workbench_kafka_inbox"));
|
|
assert.equal(inboxUpdate.params[0], 9);
|
|
});
|
|
|
|
test("HWLAB outbox claim and completion use attempt and lease fencing", async () => {
|
|
const calls = [];
|
|
const row = { outbox_seq: 3, event_id: "evt-3", topic: "hwlab.event.v1", partition_key: "trc-3", payload_json: {}, headers_json: {}, attempt_count: 4, lease_owner: "relay-a", lease_expires_at: "2026-07-10T10:05:00.000Z", next_attempt_at: "2026-07-10T10:00:00.000Z", created_at: "2026-07-10T10:00:00.000Z" };
|
|
const store = {
|
|
now: () => "2026-07-10T10:04:00.000Z",
|
|
async withDurableTransaction(_label, operation) { return operation({ query: async (sql, params) => { calls.push({ sql, params }); return { rows: [row] }; } }); },
|
|
async query(sql, params) { calls.push({ sql, params }); return { rows: [{ outbox_seq: 3 }] }; }
|
|
};
|
|
const [item] = await claimHwlabKafkaOutbox(store, { owner: "relay-a", leaseMs: 30000, limit: 1 });
|
|
await completeHwlabKafkaOutbox(store, item);
|
|
await retryHwlabKafkaOutbox(store, item, new Error("retry"), "2026-07-10T10:04:30.000Z");
|
|
|
|
assert.match(calls[0].sql, /attempt_count = o\.attempt_count \+ 1/u);
|
|
assert.match(calls[0].sql, /NOT EXISTS/u);
|
|
assert.match(calls[1].sql, /lease_owner = \$3 AND attempt_count = \$4 AND lease_expires_at = \$5 AND lease_expires_at > \$1/u);
|
|
assert.deepEqual(calls[1].params.slice(2), ["relay-a", 4, "2026-07-10T10:05:00.000Z"]);
|
|
assert.match(calls[2].sql, /lease_owner = \$5 AND attempt_count = \$6 AND lease_expires_at = \$7 AND lease_expires_at > \$3/u);
|
|
});
|
|
|
|
test("stale HWLAB outbox fencing token cannot complete a claim", async () => {
|
|
const store = { now: () => "2026-07-10T10:06:00.000Z", async query() { return { rows: [] }; } };
|
|
await assert.rejects(completeHwlabKafkaOutbox(store, { outboxSeq: 3, leaseOwner: "relay-old", attemptCount: 1, leaseExpiresAt: "2026-07-10T10:05:00.000Z" }), (error) => error?.code === "hwlab_kafka_outbox_lease_conflict");
|
|
});
|
|
|
|
function transactionStore(client) {
|
|
return {
|
|
now: () => "2026-07-10T10:00:00.000Z",
|
|
async withDurableTransaction(_label, operation) { return operation(client); }
|
|
};
|
|
}
|