From 4ee396a7859ee3fa427408ec3a3a78ffceeda947 Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Thu, 20 Aug 2026 07:00:48 -0700 Subject: [PATCH 1/3] Add tests for turn event ordering and terminal-flip classification MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The event-collector registry test dispatches two full turns fire-and-forget, the way the session orchestrator's agent.event listener does, and expects every part persisted in order with both turns finalized — today the concurrent onEvent processing drops parts ("no active turn") and leaves the second turn running forever. The decideTerminalRunFlip tests pin the classification a workflow-run pack's terminal events need: a section occurrence's repo-local child run (turn__) has no workflow_run row by design and must not be reported as a foreign-deployment violation. --- .../src/event-collector-registry.test.ts | 114 ++++++++++++++++++ .../src/hub-session-lookups.test.ts | 40 ++++++ 2 files changed, 154 insertions(+) create mode 100644 vendor/intx/hub-sessions/src/event-collector-registry.test.ts diff --git a/vendor/intx/hub-sessions/src/event-collector-registry.test.ts b/vendor/intx/hub-sessions/src/event-collector-registry.test.ts new file mode 100644 index 000000000..348ddf249 --- /dev/null +++ b/vendor/intx/hub-sessions/src/event-collector-registry.test.ts @@ -0,0 +1,114 @@ +import { describe, expect, test } from "bun:test"; + +import type { InferenceEvent } from "@intx/types/runtime"; +import type { DB } from "@intx/db"; + +import { createEventCollectorRegistry } from "./event-collector-registry"; + +// Captures every insert/update the collector issues, resolving each on a +// later tick so concurrent onEvent calls would interleave exactly the way +// they do against a real database. Rows are told apart by shape: an +// inference_turn insert carries `model`, a turn_part insert carries +// `ordinal`. +function createRecordingDb(): { + db: DB["db"]; + turnInserts: Record[]; + partInserts: Record[]; + turnUpdates: Record[]; +} { + const turnInserts: Record[] = []; + const partInserts: Record[] = []; + const turnUpdates: Record[] = []; + const later = () => new Promise((resolve) => setTimeout(resolve, 1)); + const db = { + insert: () => ({ + values: async (row: Record) => { + await later(); + if ("model" in row) turnInserts.push(row); + else partInserts.push(row); + }, + }), + update: () => ({ + set: (row: Record) => ({ + where: async () => { + await later(); + turnUpdates.push(row); + }, + }), + }), + } as unknown as DB["db"]; + return { db, turnInserts, partInserts, turnUpdates }; +} + +function turnEvents(seqBase: number, text: string): InferenceEvent[] { + return [ + { + type: "inference.start", + seq: seqBase, + data: { model: "test-model" }, + }, + { + type: "inference.done", + seq: seqBase + 1, + data: { turn: { content: [{ type: "text", text }] } }, + }, + { + type: "connector.reply", + seq: seqBase + 2, + data: { content: text }, + }, + ] as InferenceEvent[]; +} + +async function until(check: () => boolean, ms: number): Promise { + const deadline = Date.now() + ms; + while (!check() && Date.now() < deadline) { + await new Promise((resolve) => setTimeout(resolve, 5)); + } +} + +describe("event collector registry dispatch ordering (CL-6379)", () => { + test("back-to-back turn events dispatched without awaiting persist every part, in order, and finalize every turn", async () => { + const { db, turnInserts, partInserts, turnUpdates } = createRecordingDb(); + const registry = createEventCollectorRegistry({ db }); + registry.create("agent@test", "tenant-1", "ses_1", "run-1"); + + // Fire-and-forget, exactly as the session orchestrator's `agent.event` + // listener does: two full turns arrive faster than any DB roundtrip. + for (const event of [...turnEvents(1, "first"), ...turnEvents(4, "second")]) { + registry.dispatch("agent@test", event); + } + + await until(() => turnUpdates.length >= 2, 2000); + + // Two turns opened, two finalized — the second turn must not be left + // "running" because the first turn's finalize interleaved with it. + expect(turnInserts).toHaveLength(2); + expect(turnUpdates).toHaveLength(2); + expect(turnUpdates.map((u) => u["status"])).toEqual([ + "completed", + "completed", + ]); + + // Each turn persists step-start, text, step-finish — nothing dropped + // by "no active turn", and ordinals reflect event order. + const byTurn = new Map[]>(); + for (const part of partInserts) { + const parts = byTurn.get(part["turnId"]) ?? []; + parts.push(part); + byTurn.set(part["turnId"], parts); + } + expect(byTurn.size).toBe(2); + for (const parts of byTurn.values()) { + expect(parts.map((p) => p["type"])).toEqual([ + "step-start", + "text", + "step-finish", + ]); + expect(parts.map((p) => p["ordinal"])).toEqual([0, 1, 2]); + } + + // After both turns settle the collector is idle: no current turn. + expect(registry.getCurrentTurnId("agent@test")).toBeNull(); + }); +}); diff --git a/vendor/intx/hub-sessions/src/hub-session-lookups.test.ts b/vendor/intx/hub-sessions/src/hub-session-lookups.test.ts index 9b7ac9f7d..93e60a2f2 100644 --- a/vendor/intx/hub-sessions/src/hub-session-lookups.test.ts +++ b/vendor/intx/hub-sessions/src/hub-session-lookups.test.ts @@ -2,6 +2,7 @@ import { describe, expect, test } from "bun:test"; import { anchorAddressForPackSource, + decideTerminalRunFlip, ownsWorkflowRunRepo, } from "./hub-session-lookups"; @@ -71,3 +72,42 @@ describe("anchorAddressForPackSource", () => { expect(anchorAddressForPackSource(`run_deadbeef@${domain}`)).toBe(null); }); }); + +describe("decideTerminalRunFlip (CL-6379)", () => { + const anchorId = "run_cd77dfa3f597bb7d9c9ad83b9419857b"; + const mintedRunId = "run_5554796211fb7e08f1748bd9db41f71f"; + + test("flips a row anchored on the source deployment", () => { + expect(decideTerminalRunFlip(mintedRunId, anchorId, anchorId)).toEqual({ + kind: "flip", + }); + }); + + test("skips a section occurrence's repo-local child run (turn__): no row is expected and none is a defect", () => { + expect(decideTerminalRunFlip("turn__1", undefined, anchorId)).toEqual({ + kind: "skip_repo_local", + }); + }); + + test("reports a minted run id whose row is missing: the run terminated before its anchor committed", () => { + expect(decideTerminalRunFlip(mintedRunId, undefined, anchorId)).toEqual({ + kind: "missing_row", + }); + }); + + test("rejects a row anchored on a different deployment", () => { + expect( + decideTerminalRunFlip( + mintedRunId, + "run_38f239787346a29593f3bcc73efa8062", + anchorId, + ), + ).toEqual({ kind: "foreign_anchor" }); + }); + + test("a lazily-anchored internal run row with a null anchor is not the source deployment's to flip", () => { + expect(decideTerminalRunFlip(mintedRunId, null, anchorId)).toEqual({ + kind: "foreign_anchor", + }); + }); +}); From b6bab28ecaab068e9faaf01481da372ce8c90a12 Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Thu, 20 Aug 2026 07:08:23 -0700 Subject: [PATCH 2/3] Turn event pipeline: serialize collector events, classify terminal flips MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The event-collector registry dispatches agent events fire-and-forget so it never blocks the websocket loop, but the collector processed each event concurrently across its DB awaits. Under an onTrigger section's rapid end-of-turn burst (inference.done, connector.reply back-to-back) the reply's finalize nulled the current turn while inference.done was still inserting parts — dropped as "no active turn" — and a finalize landing during the next turn's begin-insert marked the new turn finalized at birth, leaving its inference_turn row running forever (stuck busy status, no live output). The collector now chains onEvent/abandon through an internal tail promise so events settle in wire order while dispatch stays fire-and-forget. receiveWorkflowRunPack's terminal-event flip treated every run id without a workflow_run row anchored on the source deployment as an ownership violation. A section occurrence's child run (turn__) is repo-local: it never mints a row, so every completed chat turn logged a spurious ERR and the benign/defect cases were indistinguishable. The flip now classifies through the pure decideTerminalRunFlip: repo-local occurrence ids skip quietly, a minted id with no row still reports loudly, and a row anchored elsewhere is still refused. --- .../intx/hub-sessions/src/event-collector.ts | 38 ++++++++-- .../hub-sessions/src/hub-session-lookups.ts | 69 +++++++++++++++---- 2 files changed, 88 insertions(+), 19 deletions(-) diff --git a/vendor/intx/hub-sessions/src/event-collector.ts b/vendor/intx/hub-sessions/src/event-collector.ts index 4137f440b..cbde9c90d 100644 --- a/vendor/intx/hub-sessions/src/event-collector.ts +++ b/vendor/intx/hub-sessions/src/event-collector.ts @@ -112,7 +112,30 @@ export function createEventCollector( // Tool results that reported isError, accumulated for TurnFinalized. let accumulatedToolErrors: { name: string; content: string }[] = []; - async function onEvent(event: InferenceEvent): Promise { + // Serializes event processing. Callers fire onEvent without awaiting + // (the registry's dispatch is deliberately fire-and-forget so it never + // blocks the websocket message loop), so without this chain two events + // interleave across their DB awaits: a connector.reply's finalize can + // null currentTurnId while inference.done is still inserting parts + // (dropping them as "no active turn"), and a finalize processed while + // the next inference.start's beginTurn is mid-insert marks the NEW + // turn finalized, leaving its row "running" forever (CL-6379). Each + // event fully settles before the next begins, restoring wire order. + let eventTail: Promise = Promise.resolve(); + + function enqueue(work: () => Promise): Promise { + const run = eventTail.then(work); + // A rejected event must not wedge every later event; the caller + // still observes the rejection through the returned promise. + eventTail = run.catch(() => undefined); + return run; + } + + function onEvent(event: InferenceEvent): Promise { + return enqueue(() => processEvent(event)); + } + + async function processEvent(event: InferenceEvent): Promise { switch (event.type) { case "inference.start": await beginTurn(event.data.model); @@ -406,12 +429,17 @@ export function createEventCollector( currentTurnId = null; } - async function abandon(): Promise { - if (currentTurnId === null || finalized) return; + function abandon(): Promise { + // Chained behind any in-flight events so an abandon issued while a + // turn's events are still persisting closes the turn they produce, + // not a half-processed intermediate state. + return enqueue(async () => { + if (currentTurnId === null || finalized) return; - log.warn`Abandoning running turn ${currentTurnId} for session ${sessionId}`; + log.warn`Abandoning running turn ${currentTurnId} for session ${sessionId}`; - await finalizeTurn("failed", false, false); + await finalizeTurn("failed", false, false); + }); } async function insertPart( diff --git a/vendor/intx/hub-sessions/src/hub-session-lookups.ts b/vendor/intx/hub-sessions/src/hub-session-lookups.ts index cf5e35499..50b2879a0 100644 --- a/vendor/intx/hub-sessions/src/hub-session-lookups.ts +++ b/vendor/intx/hub-sessions/src/hub-session-lookups.ts @@ -100,6 +100,43 @@ export function anchorAddressForPackSource( return formatRunAddress(match[0], parsed.domain); } +/** + * What an accepted workflow-run pack's newly-terminal run means for the + * `workflow_run` table. `ownedAnchorRunId` is the run's row as the flip + * transaction read it: `undefined` when no row exists, otherwise the row's + * own `anchorRunId` (possibly null). + * + * Only a minted run id (`run_` + 32 hex, the shape `generateId("workflowRun") + * produces) ever owns a row. A section-mode occurrence runs as a repo-local + * child run (`turn__` — see `@corbits/agent-runtime`'s + * `agentRuntimeTurnRunId`) whose whole identity lives in the deployment's + * own event repo: it never crosses a route that mints a `workflow_run` row, + * so its terminal event has no DB flip to perform and its absence is not a + * defect. A minted id with no row IS one — the run terminated before its + * anchor committed — and a row anchored elsewhere (or lazily anchored with + * a null anchor) is not the source deployment's to flip. + */ +export type TerminalRunFlipDecision = + | { kind: "flip" } + | { kind: "skip_repo_local" } + | { kind: "missing_row" } + | { kind: "foreign_anchor" }; + +export function decideTerminalRunFlip( + runId: string, + ownedAnchorRunId: string | null | undefined, + sourceAnchorId: string, +): TerminalRunFlipDecision { + if (ownedAnchorRunId === undefined) { + return RUN_ID_PATTERN.test(runId) + ? { kind: "missing_row" } + : { kind: "skip_repo_local" }; + } + return ownedAnchorRunId === sourceAnchorId + ? { kind: "flip" } + : { kind: "foreign_anchor" }; +} + export type HubSessionLookupsDeps = { db: DB["db"]; agentRepoStore: AgentRepoStore; @@ -523,7 +560,21 @@ export function createHubSessionLookups( .from(workflowRun) .where(eq(workflowRun.id, runId)) .limit(1); - if (ownedRun?.anchorRunId !== anchor.id) { + const decision = decideTerminalRunFlip( + runId, + ownedRun === undefined ? undefined : ownedRun.anchorRunId, + anchor.id, + ); + if (decision.kind === "skip_repo_local") { + // A section occurrence's child run (`turn__`) settles in + // the deployment's own event repo; there is no row to flip. + return; + } + if (decision.kind === "missing_row") { + logger.error`Terminal event for run ${runId} (deployment ${anchor.id}, target status ${status}) has no workflow_run row; the run terminated before its anchor committed`; + return; + } + if (decision.kind === "foreign_anchor") { logger.error`Ignoring terminal event for run ${runId}: it does not belong to source deployment ${anchor.id}`; return; } @@ -534,19 +585,9 @@ export function createHubSessionLookups( tx, ); if (won === null) { - // No running row matched. Either the run is already terminal (a - // benign replay against an already-settled row) or no row exists - // at all -- the run reached a terminal event before its anchor - // committed, so its terminal state has nowhere to land. Only the - // second case is a defect; distinguish them and log the missing - // anchor loudly rather than silently treating both as done. - const [existing] = await tx - .select({ id: workflowRun.id }) - .from(workflowRun) - .where(eq(workflowRun.id, runId)); - if (existing === undefined) { - logger.error`Terminal event for run ${runId} (deployment ${anchor.id}, target status ${status}) has no workflow_run row; the run terminated before its anchor committed`; - } + // The row exists and is anchored here but no "running" row + // matched: the run is already terminal — a benign replay + // against an already-settled row. return; } // Deactivate the run's own principal, if it has one. Externally- From e09270abe27c52b8a7060faf9341ce1ad2e4bde0 Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Thu, 20 Aug 2026 07:08:30 -0700 Subject: [PATCH 3/3] Update docs: record CL-6379 hub-sessions local modifications in VENDORED.md --- VENDORED.md | 12 ++++++++++++ 1 file changed, 12 insertions(+) diff --git a/VENDORED.md b/VENDORED.md index d70edd2c0..6f5125728 100644 --- a/VENDORED.md +++ b/VENDORED.md @@ -165,6 +165,18 @@ folded run's system prompt, tool pins, model, credential bindings) had nowhere to get it. Keyed to the approved wire hash and stored beside it, this is one store per concept, not a second copy: the projection and the hash that addresses it are written and read together. +`vendor/intx/hub-sessions` (CL-6379) serializes the event +collector's `onEvent`/`abandon` through an internal promise chain: the +registry's dispatch is deliberately fire-and-forget, and without the chain +two events interleave across their DB awaits — a `connector.reply` finalize +nulls the current turn while `inference.done` is still inserting parts +(dropped as "no active turn"), and a finalize processed during the next +`inference.start`'s begin-insert marks the NEW turn finalized, leaving its +row "running" forever. The same change classifies an accepted workflow-run +pack's newly-terminal runs through the new pure `decideTerminalRunFlip` +before the DB flip: a section occurrence's repo-local child run +(`turn__`) has no `workflow_run` row by design and is skipped quietly +instead of being logged as a foreign-deployment violation on every turn. `vendor/intx/inference-catalog`'s own local modification also repoints the `./models` subpath's exports, not just the root export.