CL-6379: Turn event pipeline — serialize collector events, classify terminal flips

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__<n>`) 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.

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<string, unknown>[];
+  partInserts: Record<string, unknown>[];
+  turnUpdates: Record<string, unknown>[];
+} {
+  const turnInserts: Record<string, unknown>[] = [];
+  const partInserts: Record<string, unknown>[] = [];
+  const turnUpdates: Record<string, unknown>[] = [];
+  const later = () => new Promise<void>((resolve) => setTimeout(resolve, 1));
+  const db = {
+    insert: () => ({
+      values: async (row: Record<string, unknown>) => {
+        await later();
+        if ("model" in row) turnInserts.push(row);
+        else partInserts.push(row);
+      },
+    }),
+    update: () => ({
+      set: (row: Record<string, unknown>) => ({
+        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<void> {
+  const deadline = Date.now() + ms;
+  while (!check() && Date.now() < deadline) {
+    await new Promise<void>((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<unknown, Record<string, unknown>[]>();
+    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();
+  });
+});

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<void> {
+  // 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<void> = Promise.resolve();
+
+  function enqueue(work: () => Promise<void>): Promise<void> {
+    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<void> {
+    return enqueue(() => processEvent(event));
+  }
+
+  async function processEvent(event: InferenceEvent): Promise<void> {
     switch (event.type) {
       case "inference.start":
         await beginTurn(event.data.model);
@@ -406,12 +429,17 @@ export function createEventCollector(
     currentTurnId = null;
   }
 
-  async function abandon(): Promise<void> {
-    if (currentTurnId === null || finalized) return;
+  function abandon(): Promise<void> {
+    // 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(

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__<n>): 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",
+    });
+  });
+});

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__<n>` — 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__<n>`) 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-