feat(core): stitch continued route attempts into one turn

packages/core/schema.json

@@ -1,8 +1,8 @@
 {
   "version": "7",
   "dialect": "sqlite",
-  "id": "3b8437ee-1a59-400d-9a69-ad74f9d1e8f5",
-  "prevIds": ["c9735d25-ef18-4dc1-ae6d-00504c68d035"],
+  "id": "2d44615a-d376-47d0-bd31-b05ed305aeb1",
+  "prevIds": ["3b8437ee-1a59-400d-9a69-ad74f9d1e8f5"],
   "ddl": [
     {
       "name": "account_state",
@@ -1384,6 +1384,26 @@
       "entityType": "columns",
       "table": "route_attempt"
     },
+    {
+      "type": "integer",
+      "notNull": false,
+      "autoincrement": false,
+      "default": null,
+      "generated": null,
+      "name": "continuation_of",
+      "entityType": "columns",
+      "table": "route_attempt"
+    },
+    {
+      "type": "integer",
+      "notNull": false,
+      "autoincrement": false,
+      "default": "0",
+      "generated": null,
+      "name": "continuation_index",
+      "entityType": "columns",
+      "table": "route_attempt"
+    },
     {
       "type": "text",
       "notNull": false,
@@ -3389,6 +3409,20 @@
       "entityType": "indexes",
       "table": "route_attempt"
     },
+    {
+      "columns": [
+        {
+          "value": "continuation_of",
+          "isExpression": false
+        }
+      ],
+      "isUnique": false,
+      "where": null,
+      "origin": "manual",
+      "name": "route_attempt_continuation_idx",
+      "entityType": "indexes",
+      "table": "route_attempt"
+    },
     {
       "columns": [
         {

packages/core/src/database/migration/20261028203055_route_attempt_continuation.ts

@@ -0,0 +1,15 @@
+import { Effect } from "effect"
+import type { DatabaseMigration } from "../migration.js"
+
+const migration: DatabaseMigration.Migration = {
+  id: "20261028203055_route_attempt_continuation",
+  up(tx) {
+    return Effect.gen(function* () {
+      yield* tx.run(`ALTER TABLE \`route_attempt\` ADD \`continuation_of\` integer;`)
+      yield* tx.run(`ALTER TABLE \`route_attempt\` ADD \`continuation_index\` integer DEFAULT 0;`)
+      yield* tx.run(`CREATE INDEX \`route_attempt_continuation_idx\` ON \`route_attempt\` (\`continuation_of\`);`)
+    })
+  },
+}
+
+export default migration

packages/core/src/database/schema.gen.ts

@@ -200,6 +200,8 @@ const schema: Omit<DatabaseMigration.Migration, "id"> = {
           \`tokens_cache_read\` integer,
           \`tokens_cache_write\` integer,
           \`cost_usd\` real,
+          \`continuation_of\` integer,
+          \`continuation_index\` integer DEFAULT 0,
           \`quota\` text
         );
       `)
@@ -445,6 +447,7 @@ const schema: Omit<DatabaseMigration.Migration, "id"> = {
       yield* tx.run(
         `CREATE INDEX \`route_attempt_session_idx\` ON \`route_attempt\` (\`session_id\`,\`time_started\`);`,
       )
+      yield* tx.run(`CREATE INDEX \`route_attempt_continuation_idx\` ON \`route_attempt\` (\`continuation_of\`);`)
       yield* tx.run(`CREATE INDEX \`route_decision_route_time_idx\` ON \`route_decision\` (\`route_id\`,\`time\`);`)
       yield* tx.run(`CREATE INDEX \`route_decision_session_idx\` ON \`route_decision\` (\`session_id\`,\`time\`);`)
       yield* tx.run(

packages/core/src/model-route-log.ts

@@ -16,7 +16,13 @@ export type Note = typeof RouteNoteTable.$inferInsert
 
 export type Event =
   | { readonly kind: "decision"; readonly row: Decision }
-  | { readonly kind: "attempt"; readonly row: Attempt }
+  | {
+      readonly kind: "attempt"
+      readonly row: Attempt
+      /** The assistant message of the attempt this request continues. The writer resolves it to that attempt's row id
+       * when it persists the row, because the id only exists after the root was inserted. */
+      readonly continuationOfMessageID?: string
+    }
   | { readonly kind: "health"; readonly row: Health }
   | { readonly kind: "note"; readonly row: Note }
 
@@ -272,15 +278,37 @@ export const failureFields = (reason: unknown, options?: { readonly captureBody?
 const persist = (db: Database.Interface["db"], events: readonly Event[]) =>
   Effect.gen(function* () {
     const decisions = events.flatMap((event) => (event.kind === "decision" ? [event.row] : []))
-    const attempts = events.flatMap((event) => (event.kind === "attempt" ? [event.row] : []))
+    const attempts = events.flatMap((event) => (event.kind === "attempt" ? [event] : []))
     const health = events.flatMap((event) => (event.kind === "health" ? [event.row] : []))
     if (decisions.length) yield* db.insert(RouteDecisionTable).values(decisions).run()
-    if (attempts.length) yield* db.insert(RouteAttemptTable).values(attempts).run()
+    // One statement per attempt: a continuation resolves its root from the row inserted just before it.
+    for (const attempt of attempts) yield* persistAttempt(db, attempt)
     if (health.length) yield* db.insert(RouteHealthTable).values(health).run()
     const notes = events.flatMap((event) => (event.kind === "note" ? [event.row] : []))
     if (notes.length) yield* db.insert(RouteNoteTable).values(notes).run()
   })
 
+/** Inserts one attempt and, for a continuation, points `continuation_of` at the attempt that started the response.
+ * The root is read back by its assistant message, so the link survives the root having been persisted in an earlier
+ * batch and stays correct across a restart. Retries and hedge losers share that message, so the newest standalone row
+ * is the attempt that actually answered — the one whose response hit the output token limit. */
+const persistAttempt = (db: Database.Interface["db"], event: Extract<Event, { kind: "attempt" }>) =>
+  Effect.gen(function* () {
+    const inserted = yield* db
+      .insert(RouteAttemptTable)
+      .values(event.row)
+      .returning({ id: RouteAttemptTable.id })
+      .get()
+    if (event.continuationOfMessageID === undefined) return
+    yield* db
+      .update(RouteAttemptTable)
+      .set({
+        continuation_of: sql`(select root.id from route_attempt root where root.session_id = ${event.row.session_id} and root.assistant_message_id = ${event.continuationOfMessageID} and root.continuation_index = 0 order by root.id desc limit 1)`,
+      })
+      .where(eq(RouteAttemptTable.id, inserted.id))
+      .run()
+  })
+
 /** Receives the still-active adjustments stored by earlier runs. Set by the routing module, which
  * this one cannot import without a cycle. */
 let restore: ((notes: readonly Record<string, unknown>[]) => void) | undefined
@@ -627,6 +655,39 @@ export const attemptRows = (db: Database.Interface["db"], since: number, routeID
     .limit(ROW_LIMIT)
     .all()
 
+/** Attempt rows in the window with their continuation link, so a response split by output-token-limit
+ * continuations can be rolled up as one turn. Newest first; `ModelUsage.turns` does the arithmetic. */
+export const turnAttempts = (db: Database.Interface["db"], since: number, sessionID?: string) =>
+  db
+    .select({
+      id: RouteAttemptTable.id,
+      continuationOf: RouteAttemptTable.continuation_of,
+      continuationIndex: RouteAttemptTable.continuation_index,
+      assistantMessageID: RouteAttemptTable.assistant_message_id,
+      sessionID: RouteAttemptTable.session_id,
+      providerID: RouteAttemptTable.provider_id,
+      modelID: RouteAttemptTable.model_id,
+      timeStarted: RouteAttemptTable.time_started,
+      timeEnded: RouteAttemptTable.time_ended,
+      outcome: RouteAttemptTable.outcome,
+      input: RouteAttemptTable.tokens_input,
+      output: RouteAttemptTable.tokens_output,
+      reasoning: RouteAttemptTable.tokens_reasoning,
+      cacheRead: RouteAttemptTable.tokens_cache_read,
+      cacheWrite: RouteAttemptTable.tokens_cache_write,
+      costUsd: RouteAttemptTable.cost_usd,
+    })
+    .from(RouteAttemptTable)
+    .where(
+      and(
+        gte(RouteAttemptTable.time_started, since),
+        sessionID === undefined ? undefined : eq(RouteAttemptTable.session_id, sessionID),
+      ),
+    )
+    .orderBy(desc(RouteAttemptTable.time_started))
+    .limit(ROW_LIMIT)
+    .all()
+
 /** Selections in the window, newest first: who was eligible, the weights and exclusions behind each, and who was chosen. */
 export const decisionRows = (db: Database.Interface["db"], since: number, routeID?: string) =>
   db

packages/core/src/model-route-log/sql.ts

@@ -67,13 +67,20 @@ export const RouteAttemptTable = sqliteTable(
     /** What the request cost in USD, from the model's catalog prices and the tokens it used. Null when the request
      * did not finish with usage. Stored with the attempt so the cost outlives the session it belongs to. */
     cost_usd: real(),
+    /** The attempt that started the response this request continued, when a response hit the output token limit and
+     * the session runner asked for the rest. Null for a request that stands on its own. A turn is a root attempt plus
+     * every attempt that names it here, so its tokens and cost can be read as one figure. */
+    continuation_of: integer(),
+    /** 0 for a standalone request, then 1, 2, … for each chunk that continues the root attempt. */
+    continuation_index: integer().default(0),
     /** Provider rate-limit and quota headers or fields, when the error carried them. */
     quota: text({ mode: "json" }).$type<Record<string, unknown>>(),
   },
   (table) => [
     index("route_attempt_target_time_idx").on(table.provider_id, table.model_id, table.time_started),
     index("route_attempt_route_time_idx").on(table.route_id, table.time_started),
     index("route_attempt_session_idx").on(table.session_id, table.time_started),
+    index("route_attempt_continuation_idx").on(table.continuation_of),
   ],
 )
 

packages/core/src/model-usage.ts

@@ -239,6 +239,94 @@ export interface DayReport {
   >
 }
 
+/** One request as far as turn stitching needs it: the link to the attempt it continues, plus what it cost. */
+export interface TurnAttemptRow {
+  readonly id: number
+  readonly continuationOf: number | null
+  readonly continuationIndex: number | null
+  readonly assistantMessageID: string | null
+  readonly sessionID: string | null
+  readonly providerID: string
+  readonly modelID: string
+  readonly timeStarted: number
+  readonly timeEnded: number
+  readonly outcome: string
+  readonly input: number | null
+  readonly output: number | null
+  readonly reasoning: number | null
+  readonly cacheRead: number | null
+  readonly cacheWrite: number | null
+  readonly costUsd: number | null
+}
+
+export interface TurnRow extends Totals {
+  /** Id of the attempt that started the turn; for a partial turn whose root fell outside the window, the id it names. */
+  readonly rootID: number
+  readonly sessionID: string | null
+  /** The target the response started on; a failover mid-turn does not move the turn to another model. */
+  readonly providerID: string
+  readonly modelID: string
+  /** Response segments: the root attempt plus each output-token-limit continuation, however many requests each took. */
+  readonly chunks: number
+  readonly first: number
+  readonly last: number
+  readonly durationMs: number
+}
+
+/** Rolls attempts up into turns: a root attempt plus every attempt whose `continuation_of` names it. Retries and hedge
+ * legs share the assistant message of the chunk they belong to, so the newest standalone attempt of a message is that
+ * chunk's root and the others join it; `requests` therefore counts every billed attempt, while `chunks` counts response
+ * segments, so a response split by output-token-limit continuations reads as one cost instead of several rows. */
+export const turns = (rows: readonly TurnAttemptRow[]): TurnRow[] => {
+  // Each chunk's root is the newest standalone attempt of its assistant message; retries and hedge legs of the same
+  // chunk share the message and join it instead of forming turns of their own.
+  const roots = new Map<string, number>()
+  for (const row of rows) {
+    if ((row.continuationIndex ?? 0) !== 0 || row.assistantMessageID === null) continue
+    roots.set(row.assistantMessageID, Math.max(roots.get(row.assistantMessageID) ?? 0, row.id))
+  }
+  const rootOf = (row: TurnAttemptRow) =>
+    row.continuationOf ?? (row.assistantMessageID === null ? row.id : (roots.get(row.assistantMessageID) ?? row.id))
+  const grouped = new Map<number, { readonly lead: TurnAttemptRow; readonly attempts: TurnAttemptRow[] }>()
+  for (const row of rows) {
+    const rootID = rootOf(row)
+    const group = grouped.get(rootID)
+    if (group) group.attempts.push(row)
+    else grouped.set(rootID, { lead: row, attempts: [row] })
+  }
+  return [...grouped]
+    .map(([rootID, { lead, attempts }]) => {
+      // The lead row stands in when the root fell outside the window, so the turn keeps a target to report.
+      const root = attempts.find((attempt) => attempt.id === rootID) ?? lead
+      const totals = sum(
+        attempts.map((attempt) => ({
+          requests: 1,
+          failures: attempt.outcome === "failure" || attempt.outcome === "timeout" ? 1 : 0,
+          input: attempt.input ?? 0,
+          output: attempt.output ?? 0,
+          reasoning: attempt.reasoning ?? 0,
+          cacheRead: attempt.cacheRead ?? 0,
+          cacheWrite: attempt.cacheWrite ?? 0,
+          costUsd: attempt.costUsd ?? 0,
+        })),
+      )
+      const first = Math.min(...attempts.map((attempt) => attempt.timeStarted))
+      const last = Math.max(...attempts.map((attempt) => attempt.timeEnded))
+      return {
+        ...totals,
+        rootID,
+        sessionID: root.sessionID,
+        providerID: root.providerID,
+        modelID: root.modelID,
+        chunks: 1 + Math.max(...attempts.map((attempt) => attempt.continuationIndex ?? 0)),
+        first,
+        last,
+        durationMs: last - first,
+      }
+    })
+    .sort((a, b) => b.last - a.last || b.rootID - a.rootID)
+}
+
 export const dayReports = (rows: readonly DayRow[]): DayReport[] => {
   const days = new Map<string, DayRow[]>()
   for (const row of rows) days.set(row.day, [...(days.get(row.day) ?? []), row])

packages/core/src/session/runner/index.ts

@@ -7,6 +7,7 @@ import { SessionSchema } from "../schema.js"
 import type { Promotable } from "../inbox.js"
 import type { AgentNotFoundError, MessageDecodeError, StepFailedError } from "../error.js"
 import { SessionRunnerModel } from "./model.js"
+import type { SessionStep } from "./step.js"
 import type { Instructions } from "../../instructions/index.js"
 
 export type RunError =
@@ -18,7 +19,11 @@ export type RunError =
   | Instructions.InitializationBlocked
   | FileSystem.DirectoryNotFoundError
 
-export type Continuation = { readonly step: number }
+export type Continuation = {
+  readonly step: number
+  /** The response link this step's continuation carries, restored when the drain resumes at a new Location. */
+  readonly link?: SessionStep.ContinuationLink
+}
 
 export type DrainResult = Data.TaggedEnum<{
   Complete: {}