refactor(hub,web): remove legacy agent_session and inference_turn shim (CL-10991)

apps/hub/src/mailbox-persist.ts

@@ -1,56 +1,12 @@
 // Wires @corbits/mailbox's createMailboxPersist onto the hub's own
-// persistMail lookup so every outbound frame also lands a durable
-// principal_mail row, not just session_mail.
+// persistMail lookup so every outbound frame lands a durable
+// principal_mail row.
 
 import { eq } from "drizzle-orm";
 import type { DB } from "@intx/db";
 import { tenant as tenantTable } from "@intx/db/schema";
-import type { AuthorizeMailboxSender, MailboxPersistArgs } from "@corbits/mailbox";
+import type { AuthorizeMailboxSender } from "@corbits/mailbox";
 import { resolveRoutableAddress } from "@intx/hub-sessions";
-import { ensureRunSession, type EventCollectorPort } from "./lazy-run-collector";
-import { reportError } from "@corbits/error-sink";
-import { getLogger } from "@intx/log";
-
-const logger = getLogger(["hub", "mailbox-persist"]);
-
-/**
- * Ensures a run's agent_session/event collector exist before the first
- * mail-triggered write reaches it, since workflow_run.principal_id only
- * reconciles onto the trigger that just fired. Best-effort: a failure
- * here must not block the mail upstream is about to persist regardless.
- */
-export function createHubPersistMailWithSessionEnsure<R extends readonly unknown[]>(
-  db: DB["db"],
-  eventCollectorsRef: { current?: Pick<EventCollectorPort, "create" | "has"> },
-  upstream: (args: MailboxPersistArgs) => Promise<R>,
-): (args: MailboxPersistArgs) => Promise<R> {
-  return async (args) => {
-    let sender = await resolveRoutableAddress(db, args.senderAddress);
-    try {
-      const eventCollectors = eventCollectorsRef.current;
-      if (sender !== undefined && eventCollectors !== undefined) {
-        await ensureRunSession({ db, eventCollectors, runId: sender.id });
-        sender = await resolveRoutableAddress(db, args.senderAddress);
-      }
-    } catch (err) {
-      reportError(err, {
-        operation: "hub.mailboxPersist.ensureRunSession",
-        extra: { senderAddress: args.senderAddress },
-      });
-    }
-    // The vendored session_mail write keys on the sender's session, which
-    // may still not exist; skip it rather than let it throw, since
-    // @corbits/mailbox's own write already persists this frame.
-    if (sender !== undefined && sender.sessionId === null) {
-      logger.debug("skipping vendored persistMail for run {runId}: no session yet", {
-        runId: sender.id,
-        senderAddress: args.senderAddress,
-      });
-      return [] as unknown as R;
-    }
-    return upstream(args);
-  };
-}
 
 export function createHubMailboxAuthorizeSender(db: DB["db"]): AuthorizeMailboxSender {
   return async (senderAddress) => {

apps/hub/src/server.ts

@@ -64,7 +64,6 @@ import {
   type SidecarProvisionerChooser,
   type WsHandle,
 } from "@intx/hub-sessions";
-import { withLazyRunCollector } from "./lazy-run-collector";
 import { generateKeyPair } from "@intx/crypto";
 import { timeWindowEvaluator } from "@intx/authz";
 import type { ConditionRegistry } from "@intx/types/authz";
@@ -109,10 +108,7 @@ import {
 } from "@corbits/xai-provider";
 import { createAgentTokenVerifier, mountAgentTokens } from "@corbits/agent-token";
 import { createCronTicker, createCronRoutes, createRunTriggerCronDeliver } from "@corbits/cron";
-import {
-  createHubMailboxAuthorizeSender,
-  createHubPersistMailWithSessionEnsure,
-} from "./mailbox-persist";
+import { createHubMailboxAuthorizeSender } from "./mailbox-persist";
 import { captureMailboxRequest, createMailboxDeliver } from "./mailbox-send";
 import { reportError } from "@corbits/error-sink";
 import {
@@ -345,27 +341,24 @@ export async function createHubServer({
     void pushCredentialReconcile(db, sidecarRouter, agentAddress, credentialCipher);
   };
 
-  const eventCollectors = withLazyRunCollector(
-    createEventCollectorRegistry({
-      db,
-      onTurnFinalized(agentAddress, turn) {
-        sidecarRouter.dispatchAgentEvent(agentAddress, {
-          type: "turn.committed",
-          data: {
-            turnId: turn.turnId,
-            status: turn.status,
-            text: turn.text,
-            hadReply: turn.hadReply,
-            hadError: turn.hadError,
-            errors: turn.errors,
-            toolCalls: turn.toolCalls,
-            toolErrors: turn.toolErrors,
-          },
-        });
-      },
-    }),
+  const eventCollectors = createEventCollectorRegistry({
     db,
-  );
+    onTurnFinalized(agentAddress, turn) {
+      sidecarRouter.dispatchAgentEvent(agentAddress, {
+        type: "turn.committed",
+        data: {
+          turnId: turn.turnId,
+          status: turn.status,
+          text: turn.text,
+          hadReply: turn.hadReply,
+          hadError: turn.hadError,
+          errors: turn.errors,
+          toolCalls: turn.toolCalls,
+          toolErrors: turn.toolErrors,
+        },
+      });
+    },
+  });
 
   createHubSessionOrchestrator({
     events: sidecarRouter.events,
@@ -623,26 +616,11 @@ export async function createHubServer({
   const corbitsDatabaseUrl = `postgres://${encodeURIComponent(process.env["DB_USER"] ?? "postgres")}:${encodeURIComponent(process.env["DB_PASSWORD"] ?? "postgres")}@${process.env["DB_HOST"] ?? "localhost"}:${String(Number(process.env["DB_PORT"] ?? 5432))}/${process.env["DB_NAME"] ?? "interchange"}`;
   const { db: mailboxDb } = createMailboxDb(corbitsDatabaseUrl);
   const mailboxBus = createInMemoryMailboxEventBus();
-  const eventCollectorsRef: { current?: typeof eventCollectors } = { current: eventCollectors };
-  const mailboxLookups = {
-    ...lookups,
-    persistMail: createMailboxPersist(mailboxDb, {
-      upstream: createHubPersistMailWithSessionEnsure(
-        db,
-        eventCollectorsRef,
-        createHubSessionLookups({
-          db,
-          agentRepoStore,
-          historyReceives: workflowHistoryReceives,
-        }).persistMail,
-      ),
-      authorizeSender: createHubMailboxAuthorizeSender(db),
-      bus: mailboxBus,
-    }),
-  };
-  // The sidecar router captured `lookups` before this wrapper existed; an
-  // agent's outbound mail must also land in principal inboxes.
-  lookups.persistMail = mailboxLookups.persistMail;
+  lookups.persistMail = createMailboxPersist(mailboxDb, {
+    upstream: async () => [],
+    authorizeSender: createHubMailboxAuthorizeSender(db),
+    bus: mailboxBus,
+  });
 
   const artifactContentStore = InlineContentStore;
 
@@ -760,7 +738,7 @@ export async function createHubServer({
         // before replying, so the canonical stamped form is lowercase.
         return `${principalRow.refId}@${tenantRow.domain}`.toLowerCase();
       },
-      deliver: createMailboxDeliver({ app, persistMail: mailboxLookups.persistMail }),
+      deliver: createMailboxDeliver({ app, persistMail: lookups.persistMail }),
     });
     app.route(`${TENANT_PREFIX}/mailbox`, mailboxApp);
   }

apps/web/src/insights-api.ts

@@ -5,19 +5,12 @@
 import { type } from "arktype";
 import { RunApprovalsResponse, WorkflowRunResponse, paginatedSchema } from "@intx/types";
 
-export const ListingTurnSchema = type({
-  status: "string",
-  "endedAt?": "string | null",
-});
-
-// `routineId`/`routineName` and `turns`/`hasInFlightTurn` are this app's
-// own additions, absent on the native feed — see docs/insights-native-runs.md.
+// `routineId`/`routineName` are this app's own additions, absent on the
+// native feed — see docs/insights-native-runs.md.
 export const InsightsRunSchema = WorkflowRunResponse.and(
   type({
     "routineId?": "string | null",
     "routineName?": "string | null",
-    "hasInFlightTurn?": "boolean",
-    "turns?": ListingTurnSchema.array(),
   }),
 );
 export type InsightsRun = typeof InsightsRunSchema.infer;

apps/web/src/insights-stats.ts

@@ -2,7 +2,7 @@
 // routines). No new analytics backend — I1 is an honest live surface on
 // existing endpoints.
 
-import { runOutcomeStatus, withListingAbandoned } from "./run-outcome";
+import { runOutcomeStatus } from "./run-outcome";
 
 import { isAgentDeploySourceAssetName } from "./agent-deploy";
 import type { Approval } from "./api";
@@ -129,8 +129,8 @@ function runDurationMs(run: InsightsRun): number | null {
 }
 
 type Bucket = "ok" | "fail" | "other";
-function bucketOf(run: InsightsRun, now: number): Bucket {
-  const outcome = runOutcomeStatus(withListingAbandoned(run, now), now) ?? run.status;
+function bucketOf(run: InsightsRun): Bucket {
+  const outcome = runOutcomeStatus(run);
   if (outcome === "completed") return "ok";
   if (outcome === "failed" || outcome === "error") return "fail";
   return "other";
@@ -162,7 +162,7 @@ export function computeBenchInsights(
   let fail = 0;
   const groups = new Map<string, InsightsRun[]>();
   for (const run of inRange) {
-    const bucket = bucketOf(run, now);
+    const bucket = bucketOf(run);
     const midnight = new Date(run.createdAt);
     midnight.setHours(0, 0, 0, 0);
     // round, not floor: DST days are 23h or 25h long.
@@ -193,8 +193,8 @@ export function computeBenchInsights(
         name: newest[0] !== undefined ? runDisplayName(newest[0]) : key,
         definitionName: newest[0]?.definitionName ?? key,
         runs: rs.length,
-        ok: rs.filter((r) => bucketOf(r, now) === "ok").length,
-        fail: rs.filter((r) => bucketOf(r, now) === "fail").length,
+        ok: rs.filter((r) => bucketOf(r) === "ok").length,
+        fail: rs.filter((r) => bucketOf(r) === "fail").length,
         medianMs: medianMs(durations(rs)),
         lastRun: newest[0]?.createdAt ?? "",
       };
@@ -209,7 +209,7 @@ export function computeBenchInsights(
     days,
     workflows,
     failures: inRange
-      .filter((r) => bucketOf(r, now) === "fail")
+      .filter((r) => bucketOf(r) === "fail")
       .sort((a, b) => b.createdAt.localeCompare(a.createdAt))
       .slice(0, 5),
   };