From 67b4c9c48e5def2f9bb97547ecb893d794d11d2a Mon Sep 17 00:00:00 2001 From: nullstack65 Date: Wed, 23 Sep 2026 17:31:40 -0400 Subject: [PATCH] feat(usage): retain durable model-canary attribution across resumes T3's single resume cursor lost earlier native sessions on a resume, fork, or model switch, so a thread could not say which models contributed. Add an append-only provider_session_history (migration 054, with a current-cursor backfill) written alongside the runtime upsert, and read it through usageAttributionSources as sessionHistory bindings. Add a content-free thread_route_events store (migration 055) for pre-execution route/experiment metadata: requested provider/model/effort captured at session start, task stratum, experiment/cohort, route event kind, manager/agent ids, and escalation reason. Actual values are derived from measured usage, never copied from the request. usageRouteAttribution composes the base projection with this view and keeps child sessions distinct. T3 carries the metadata only; agent-config remains the routing authority. No prompts, responses, code, or tool bodies are stored. --- apps/server/src/persistence/Migrations.ts | 4 + .../054_ProviderSessionHistory.test.ts | 115 ++++ .../Migrations/054_ProviderSessionHistory.ts | 94 ++++ .../Migrations/055_ThreadRouteEvents.ts | 43 ++ .../ProviderSessionRuntime.history.test.ts | 239 ++++++++ .../src/persistence/ProviderSessionRuntime.ts | 253 ++++++++- .../src/provider/Layers/ProviderService.ts | 25 +- .../Layers/ProviderSessionDirectory.ts | 9 +- .../Services/ProviderSessionDirectory.ts | 17 + apps/server/src/usage/routeMetadata.ts | 173 ++++++ apps/server/src/usage/usageAttribution.ts | 12 +- .../src/usage/usageAttributionSources.test.ts | 220 ++++++++ .../src/usage/usageAttributionSources.ts | 211 +++++++- .../src/usage/usageRouteAttribution.test.ts | 512 ++++++++++++++++++ .../server/src/usage/usageRouteAttribution.ts | 446 +++++++++++++++ docs/internals/usage-attribution.md | 56 +- 16 files changed, 2412 insertions(+), 17 deletions(-) create mode 100644 apps/server/src/persistence/Migrations/054_ProviderSessionHistory.test.ts create mode 100644 apps/server/src/persistence/Migrations/054_ProviderSessionHistory.ts create mode 100644 apps/server/src/persistence/Migrations/055_ThreadRouteEvents.ts create mode 100644 apps/server/src/persistence/ProviderSessionRuntime.history.test.ts create mode 100644 apps/server/src/usage/routeMetadata.ts create mode 100644 apps/server/src/usage/usageRouteAttribution.test.ts create mode 100644 apps/server/src/usage/usageRouteAttribution.ts diff --git a/apps/server/src/persistence/Migrations.ts b/apps/server/src/persistence/Migrations.ts index 18aae09febf8..62e9b5574712 100644 --- a/apps/server/src/persistence/Migrations.ts +++ b/apps/server/src/persistence/Migrations.ts @@ -65,6 +65,8 @@ import Migration0050 from "./Migrations/050_ProjectionThreadPullRequests.ts"; import Migration0051 from "./Migrations/051_ProjectionThreadMessageContext.ts"; import Migration0052 from "./Migrations/052_ProjectionThreadTitleState.ts"; import Migration0053 from "./Migrations/053_PullRequestFilesViewed.ts"; +import Migration0054 from "./Migrations/054_ProviderSessionHistory.ts"; +import Migration0055 from "./Migrations/055_ThreadRouteEvents.ts"; /** * Migration loader with all migrations defined inline. @@ -130,6 +132,8 @@ const migrationEntries = [ [51, "ProjectionThreadMessageContext", Migration0051], [52, "ProjectionThreadTitleState", Migration0052], [53, "PullRequestFilesViewed", Migration0053], + [54, "ProviderSessionHistory", Migration0054], + [55, "ThreadRouteEvents", Migration0055], ] as const; export const migrationManifest = migrationEntries.map(([id, name]) => [id, name] as const); diff --git a/apps/server/src/persistence/Migrations/054_ProviderSessionHistory.test.ts b/apps/server/src/persistence/Migrations/054_ProviderSessionHistory.test.ts new file mode 100644 index 000000000000..6faa40a03766 --- /dev/null +++ b/apps/server/src/persistence/Migrations/054_ProviderSessionHistory.test.ts @@ -0,0 +1,115 @@ +import { assert, it } from "@effect/vitest"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as SqlClient from "effect/unstable/sql/SqlClient"; +import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient"; + +import { migrationManifest, runMigrations } from "../Migrations.ts"; + +const layer = it.layer(Layer.mergeAll(NodeSqliteClient.layer({ filename: ":memory:" }))); + +interface HistoryRow { + readonly threadId: string; + readonly providerName: string; + readonly nativeSessionId: string; + readonly parentNativeSessionId: string | null; + readonly origin: string; + readonly firstSeenAt: string; + readonly lastSeenAt: string; +} + +layer("054_ProviderSessionHistory", (it) => { + it.effect("creates the history tables and backfills the current cursor", () => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + + yield* runMigrations({ toMigrationInclusive: 53 }); + + const insertRuntime = (threadId: string, providerName: string, cursor: string | null) => + sql` + INSERT INTO provider_session_runtime ( + thread_id, + provider_name, + provider_instance_id, + adapter_key, + runtime_mode, + status, + last_seen_at, + resume_cursor_json, + runtime_payload_json + ) + VALUES ( + ${threadId}, + ${providerName}, + ${providerName}, + ${providerName}, + 'full-access', + 'running', + '2026-09-23T09:00:00.000Z', + ${cursor}, + NULL + ) + `; + + yield* insertRuntime("thread-codex", "codex", '{"threadId":"session-a"}'); + yield* insertRuntime("thread-claude", "claudeAgent", '{"resume":"session-b"}'); + // No recognised id field: nothing to backfill. + yield* insertRuntime("thread-unknown", "codex", '{"opaque":true}'); + // No cursor at all. + yield* insertRuntime("thread-absent", "codex", null); + + yield* runMigrations({ toMigrationInclusive: 55 }); + + const rows = yield* sql` + SELECT + thread_id AS "threadId", + provider_name AS "providerName", + native_session_id AS "nativeSessionId", + parent_native_session_id AS "parentNativeSessionId", + origin, + first_seen_at AS "firstSeenAt", + last_seen_at AS "lastSeenAt" + FROM provider_session_history + ORDER BY native_session_id ASC + `; + + assert.deepStrictEqual(rows, [ + { + threadId: "thread-codex", + providerName: "codex", + nativeSessionId: "session-a", + parentNativeSessionId: null, + origin: "runtimeCursor", + firstSeenAt: "2026-09-23T09:00:00.000Z", + lastSeenAt: "2026-09-23T09:00:00.000Z", + }, + { + threadId: "thread-claude", + providerName: "claudeAgent", + nativeSessionId: "session-b", + parentNativeSessionId: null, + origin: "runtimeCursor", + firstSeenAt: "2026-09-23T09:00:00.000Z", + lastSeenAt: "2026-09-23T09:00:00.000Z", + }, + ]); + + const routeEventCount = yield* sql<{ readonly count: number }>` + SELECT COUNT(*) AS count FROM thread_route_events + `; + assert.equal(routeEventCount[0]!.count, 0); + + // The unique key is on the durable identity, so a repeat cannot duplicate. + const indexes = yield* sql<{ readonly name: string }>` + PRAGMA index_list(provider_session_history) + `; + assert.ok(indexes.length > 0); + }), + ); + + it("registers both migrations in the manifest", () => { + const entries = new Set(migrationManifest.map(([id, name]) => `${id}_${name}`)); + assert.ok(entries.has("54_ProviderSessionHistory")); + assert.ok(entries.has("55_ThreadRouteEvents")); + }); +}); diff --git a/apps/server/src/persistence/Migrations/054_ProviderSessionHistory.ts b/apps/server/src/persistence/Migrations/054_ProviderSessionHistory.ts new file mode 100644 index 000000000000..343142c329a0 --- /dev/null +++ b/apps/server/src/persistence/Migrations/054_ProviderSessionHistory.ts @@ -0,0 +1,94 @@ +import * as SqlClient from "effect/unstable/sql/SqlClient"; +import * as Effect from "effect/Effect"; + +/** + * Append-only native-session identity history per T3 thread. + * + * `provider_session_runtime` keeps a single current `resume_cursor_json`. A + * resume, fork, or model switch overwrites it, so earlier native sessions that + * contributed to a thread become unrecoverable. This table records one row per + * `(thread_id, provider_name, native_session_id)` so a thread can answer which + * native sessions it used even after the cursor moved on. Repeated observations + * of the same session only advance `last_seen_at`; they never replace a + * different session's row. + * + * The backfill seeds the table from whatever cursor already exists so an + * upgraded database does not start empty. + */ +export default Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + + yield* sql` + CREATE TABLE IF NOT EXISTS provider_session_history ( + history_id INTEGER PRIMARY KEY AUTOINCREMENT, + thread_id TEXT NOT NULL, + provider_name TEXT NOT NULL, + provider_instance_id TEXT, + adapter_key TEXT NOT NULL, + native_session_id TEXT NOT NULL, + parent_native_session_id TEXT, + origin TEXT NOT NULL, + first_seen_at TEXT NOT NULL, + last_seen_at TEXT NOT NULL, + UNIQUE (thread_id, provider_name, native_session_id) + ) + `; + + yield* sql` + CREATE INDEX IF NOT EXISTS idx_provider_session_history_thread + ON provider_session_history(thread_id, first_seen_at) + `; + + yield* sql` + INSERT OR IGNORE INTO provider_session_history ( + thread_id, + provider_name, + provider_instance_id, + adapter_key, + native_session_id, + parent_native_session_id, + origin, + first_seen_at, + last_seen_at + ) + SELECT + current.thread_id, + current.provider_name, + current.provider_instance_id, + current.adapter_key, + current.native_session_id, + NULL, + 'runtimeCursor', + current.last_seen_at, + current.last_seen_at + FROM ( + SELECT + runtime.thread_id, + runtime.provider_name, + runtime.provider_instance_id, + runtime.adapter_key, + runtime.last_seen_at, + COALESCE( + json_extract(runtime.cursor, '$.resume'), + json_extract(runtime.cursor, '$.threadId'), + json_extract(runtime.cursor, '$.sessionId') + ) AS native_session_id + FROM ( + SELECT + thread_id, + provider_name, + provider_instance_id, + adapter_key, + last_seen_at, + CASE + WHEN resume_cursor_json IS NOT NULL AND json_valid(resume_cursor_json) + THEN resume_cursor_json + ELSE NULL + END AS cursor + FROM provider_session_runtime + ) AS runtime + ) AS current + WHERE current.native_session_id IS NOT NULL + AND current.native_session_id <> '' + `; +}); diff --git a/apps/server/src/persistence/Migrations/055_ThreadRouteEvents.ts b/apps/server/src/persistence/Migrations/055_ThreadRouteEvents.ts new file mode 100644 index 000000000000..abb9b406fbe2 --- /dev/null +++ b/apps/server/src/persistence/Migrations/055_ThreadRouteEvents.ts @@ -0,0 +1,43 @@ +import * as SqlClient from "effect/unstable/sql/SqlClient"; +import * as Effect from "effect/Effect"; + +/** + * Append-only pre-execution route and experiment metadata. + * + * This is low-cardinality metadata about a routing decision, not conversation + * content: which provider/model/effort was requested, the pre-execution task + * stratum, the experiment/cohort, the readable manager/agent identifiers, the + * route event kind, and any escalation reason. T3 only carries the metadata; + * the canonical policy that chooses a route lives in agent-config. + * + * `route_event_kind` is nullable so an automatic "this is what was requested" + * event is distinguishable from a declared canary/fallback/escalation event. + * Observed (actual) values are deliberately NOT stored here: they must come + * from measured usage, never be copied from the request. + */ +export default Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + + yield* sql` + CREATE TABLE IF NOT EXISTS thread_route_events ( + event_id TEXT PRIMARY KEY, + thread_id TEXT NOT NULL, + native_session_id TEXT, + route_event_kind TEXT, + task_stratum TEXT NOT NULL, + experiment_id TEXT, + manager_id TEXT, + agent_id TEXT, + requested_provider TEXT, + requested_model TEXT, + requested_effort TEXT, + escalation_reason TEXT, + recorded_at TEXT NOT NULL + ) + `; + + yield* sql` + CREATE INDEX IF NOT EXISTS idx_thread_route_events_thread + ON thread_route_events(thread_id, recorded_at) + `; +}); diff --git a/apps/server/src/persistence/ProviderSessionRuntime.history.test.ts b/apps/server/src/persistence/ProviderSessionRuntime.history.test.ts new file mode 100644 index 000000000000..e494e5a3874c --- /dev/null +++ b/apps/server/src/persistence/ProviderSessionRuntime.history.test.ts @@ -0,0 +1,239 @@ +// @effect-diagnostics nodeBuiltinImport:off +import { ProviderInstanceId, ThreadId } from "@t3tools/contracts"; +import { assert, expect, it } from "@effect/vitest"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as SqlClient from "effect/unstable/sql/SqlClient"; + +import { SqlitePersistenceMemory } from "./Layers/Sqlite.ts"; +import * as ProviderSessionRuntime from "./ProviderSessionRuntime.ts"; + +function runtimeRow(overrides: { threadId: ThreadId; lastSeenAt: string; resumeCursor: unknown }) { + return { + threadId: overrides.threadId, + providerName: "codex", + providerInstanceId: ProviderInstanceId.make("codex"), + adapterKey: "codex", + runtimeMode: "full-access" as const, + status: "running" as const, + lastSeenAt: overrides.lastSeenAt, + resumeCursor: overrides.resumeCursor, + runtimePayload: null, + }; +} + +const layer = Layer.mergeAll( + SqlitePersistenceMemory, + ProviderSessionRuntime.layer.pipe(Layer.provide(SqlitePersistenceMemory)), +); + +interface HistoryRow { + readonly nativeSessionId: string; + readonly parentNativeSessionId: string | null; + readonly origin: string; + readonly firstSeenAt: string; + readonly lastSeenAt: string; +} + +function historyRows(sql: SqlClient.SqlClient, threadId: ThreadId) { + return sql` + SELECT + native_session_id AS "nativeSessionId", + parent_native_session_id AS "parentNativeSessionId", + origin, + first_seen_at AS "firstSeenAt", + last_seen_at AS "lastSeenAt" + FROM provider_session_history + WHERE thread_id = ${threadId} + ORDER BY first_seen_at ASC, native_session_id ASC + `; +} + +it.layer(layer)("ProviderSessionRuntime durable history", (it) => { + it.effect("retains every native session when the resume cursor changes", () => + Effect.gen(function* () { + const repository = yield* ProviderSessionRuntime.ProviderSessionRuntimeRepository; + const sql = yield* SqlClient.SqlClient; + const threadId = ThreadId.make("thread-history-resume"); + + yield* repository.upsert( + runtimeRow({ + threadId, + lastSeenAt: "2026-09-23T09:00:00.000Z", + resumeCursor: { threadId: "session-a" }, + }), + ); + yield* repository.upsert( + runtimeRow({ + threadId, + lastSeenAt: "2026-09-23T09:30:00.000Z", + resumeCursor: { threadId: "session-b" }, + }), + { attribution: { parentNativeSessionId: "session-a" } }, + ); + + const rows = yield* historyRows(sql, threadId); + assert.deepEqual( + rows.map((row) => row.nativeSessionId), + ["session-a", "session-b"], + ); + assert.equal(rows[0]!.parentNativeSessionId, null); + assert.equal(rows[1]!.parentNativeSessionId, "session-a"); + assert.equal(rows[0]!.origin, "runtimeCursor"); + // The cursor now points at session-b, but session-a is still durable. + const runtime = Option.getOrThrow(yield* repository.getByThreadId({ threadId })); + assert.deepEqual(runtime.resumeCursor, { threadId: "session-b" }); + expect(rows).toHaveLength(2); + }), + ); + + it.effect("advances last_seen_at without duplicating a repeated session", () => + Effect.gen(function* () { + const repository = yield* ProviderSessionRuntime.ProviderSessionRuntimeRepository; + const sql = yield* SqlClient.SqlClient; + const threadId = ThreadId.make("thread-history-repeat"); + + yield* repository.upsert( + runtimeRow({ + threadId, + lastSeenAt: "2026-09-23T09:00:00.000Z", + resumeCursor: { resume: "same-session" }, + }), + ); + yield* repository.upsert( + runtimeRow({ + threadId, + lastSeenAt: "2026-09-23T09:45:00.000Z", + resumeCursor: { resume: "same-session" }, + }), + ); + + const rows = yield* historyRows(sql, threadId); + assert.equal(rows.length, 1); + assert.equal(rows[0]!.firstSeenAt, "2026-09-23T09:00:00.000Z"); + assert.equal(rows[0]!.lastSeenAt, "2026-09-23T09:45:00.000Z"); + }), + ); + + it.effect("does not append history when a conflicting write is ignored", () => + Effect.gen(function* () { + const repository = yield* ProviderSessionRuntime.ProviderSessionRuntimeRepository; + const sql = yield* SqlClient.SqlClient; + const threadId = ThreadId.make("thread-history-ignore"); + + yield* repository.upsert( + runtimeRow({ + threadId, + lastSeenAt: "2026-09-23T09:00:00.000Z", + resumeCursor: { sessionId: "active" }, + }), + ); + yield* repository.upsert( + runtimeRow({ + threadId, + lastSeenAt: "2026-09-23T09:05:00.000Z", + resumeCursor: { sessionId: "stale" }, + }), + { onConflict: "ignore" }, + ); + + const rows = yield* historyRows(sql, threadId); + assert.deepEqual( + rows.map((row) => row.nativeSessionId), + ["active"], + ); + }), + ); + + it.effect("records requested route and declared experiment metadata", () => + Effect.gen(function* () { + const repository = yield* ProviderSessionRuntime.ProviderSessionRuntimeRepository; + const sql = yield* SqlClient.SqlClient; + const threadId = ThreadId.make("thread-route-events"); + + yield* repository.upsert( + runtimeRow({ + threadId, + lastSeenAt: "2026-09-23T09:00:00.000Z", + resumeCursor: { sessionId: "ses_opencode_1" }, + }), + { + attribution: { + requestedRoute: { provider: "opencode", model: "gpt-6-luna", effort: "high" }, + routeEvent: { + eventId: "evt-canary-1", + kind: "canary", + taskStratum: "implementation", + experimentId: "canary-2026-09", + managerId: "ROUTE6-1", + agentId: "T3", + reason: "measured canary cohort", + requested: { provider: "opencode", model: "gpt-6-luna", effort: "high" }, + }, + }, + }, + ); + + const events = yield* sql<{ + eventId: string; + routeEventKind: string | null; + taskStratum: string; + managerId: string | null; + agentId: string | null; + requestedModel: string | null; + escalationReason: string | null; + }>` + SELECT + event_id AS "eventId", + route_event_kind AS "routeEventKind", + task_stratum AS "taskStratum", + manager_id AS "managerId", + agent_id AS "agentId", + requested_model AS "requestedModel", + escalation_reason AS "escalationReason" + FROM thread_route_events + WHERE thread_id = ${threadId} + ORDER BY recorded_at ASC, event_id ASC + `; + + const declared = events.find((event) => event.eventId === "evt-canary-1"); + assert.ok(declared); + assert.equal(declared.routeEventKind, "canary"); + assert.equal(declared.taskStratum, "implementation"); + assert.equal(declared.managerId, "ROUTE6-1"); + assert.equal(declared.agentId, "T3"); + assert.equal(declared.escalationReason, "measured canary cohort"); + assert.equal(declared.requestedModel, "gpt-6-luna"); + // An automatic request record is written alongside it, unclassified. + const request = events.find((event) => event.eventId.includes("::request::")); + assert.ok(request); + assert.equal(request.routeEventKind, null); + assert.equal(request.requestedModel, "gpt-6-luna"); + // Repeated writes collapse instead of duplicating. + yield* repository.upsert( + runtimeRow({ + threadId, + lastSeenAt: "2026-09-23T09:10:00.000Z", + resumeCursor: { sessionId: "ses_opencode_1" }, + }), + { + attribution: { + requestedRoute: { provider: "opencode", model: "gpt-6-luna", effort: "high" }, + routeEvent: { + eventId: "evt-canary-1", + kind: "canary", + taskStratum: "implementation", + managerId: "ROUTE6-1", + agentId: "T3", + }, + }, + }, + ); + const after = yield* sql<{ count: number }>` + SELECT COUNT(*) AS count FROM thread_route_events WHERE thread_id = ${threadId} + `; + assert.equal(after[0]!.count, 2); + }), + ); +}); diff --git a/apps/server/src/persistence/ProviderSessionRuntime.ts b/apps/server/src/persistence/ProviderSessionRuntime.ts index 2673512edf10..6ae99e5233a5 100644 --- a/apps/server/src/persistence/ProviderSessionRuntime.ts +++ b/apps/server/src/persistence/ProviderSessionRuntime.ts @@ -18,6 +18,12 @@ import { ThreadId, } from "@t3tools/contracts"; +import { + normalizeTaskStratum, + type RouteEventInput, + type RouteSelectionMetadata, +} from "../usage/routeMetadata.ts"; + import { PersistenceDecodeError, type PersistenceErrorCorrelation, @@ -67,6 +73,21 @@ export type RecordImportedTranscriptInput = typeof RecordImportedTranscriptInput export interface ProviderSessionRuntimeUpsertOptions { readonly onConflict?: "update" | "ignore"; + /** + * Additive attribution metadata. Never changes the runtime row itself; it + * only appends to `provider_session_history` (identity) and + * `thread_route_events` (pre-execution route/experiment metadata). + */ + readonly attribution?: ProviderSessionRuntimeAttributionOptions; +} + +export interface ProviderSessionRuntimeAttributionOptions { + /** True sub-agent parent native session id, when the caller knows it. */ + readonly parentNativeSessionId?: string | null; + /** What this session was asked to run. Recorded, never treated as observed. */ + readonly requestedRoute?: RouteSelectionMetadata | null; + /** Full declared route/experiment event, when the route authority supplies it. */ + readonly routeEvent?: RouteEventInput | null; } /** @@ -268,6 +289,212 @@ export const make = Effect.gen(function* () { `, }); + /** + * The three cursor shapes adapters write: `{ resume }` (Claude), + * `{ threadId }` (Codex), `{ sessionId }` (Grok, OpenCode, Antigravity). + * The value arrives here JSON-encoded, so it is parsed defensively. + */ + const nativeSessionIdOf = (cursor: unknown): string | null => { + let parsed: unknown = cursor; + if (typeof cursor === "string") { + if (cursor.length === 0) return null; + try { + parsed = JSON.parse(cursor); + } catch { + return null; + } + } + if (parsed === null || typeof parsed !== "object" || Array.isArray(parsed)) return null; + const record = parsed as Record; + for (const field of ["resume", "threadId", "sessionId"] as const) { + const value = record[field]; + if (typeof value === "string" && value.trim().length > 0) return value.trim(); + } + return null; + }; + + const recordSessionHistoryRow = SqlSchema.void({ + Request: Schema.Struct({ + threadId: Schema.String, + providerName: Schema.String, + providerInstanceId: Schema.NullOr(Schema.String), + adapterKey: Schema.String, + nativeSessionId: Schema.String, + parentNativeSessionId: Schema.NullOr(Schema.String), + seenAt: Schema.String, + }), + execute: (entry) => + sql` + INSERT INTO provider_session_history ( + thread_id, + provider_name, + provider_instance_id, + adapter_key, + native_session_id, + parent_native_session_id, + origin, + first_seen_at, + last_seen_at + ) + VALUES ( + ${entry.threadId}, + ${entry.providerName}, + ${entry.providerInstanceId}, + ${entry.adapterKey}, + ${entry.nativeSessionId}, + ${entry.parentNativeSessionId}, + 'runtimeCursor', + ${entry.seenAt}, + ${entry.seenAt} + ) + ON CONFLICT (thread_id, provider_name, native_session_id) + DO UPDATE SET + adapter_key = excluded.adapter_key, + provider_instance_id = COALESCE( + excluded.provider_instance_id, + provider_session_history.provider_instance_id + ), + last_seen_at = CASE + WHEN excluded.last_seen_at > provider_session_history.last_seen_at + THEN excluded.last_seen_at + ELSE provider_session_history.last_seen_at + END, + parent_native_session_id = COALESCE( + provider_session_history.parent_native_session_id, + excluded.parent_native_session_id + ) + `, + }); + + const recordRouteEventRow = SqlSchema.void({ + Request: Schema.Struct({ + eventId: Schema.String, + threadId: Schema.String, + nativeSessionId: Schema.NullOr(Schema.String), + routeEventKind: Schema.NullOr(Schema.String), + taskStratum: Schema.String, + experimentId: Schema.NullOr(Schema.String), + managerId: Schema.NullOr(Schema.String), + agentId: Schema.NullOr(Schema.String), + requestedProvider: Schema.NullOr(Schema.String), + requestedModel: Schema.NullOr(Schema.String), + requestedEffort: Schema.NullOr(Schema.String), + escalationReason: Schema.NullOr(Schema.String), + recordedAt: Schema.String, + }), + execute: (event) => + sql` + INSERT OR IGNORE INTO thread_route_events ( + event_id, + thread_id, + native_session_id, + route_event_kind, + task_stratum, + experiment_id, + manager_id, + agent_id, + requested_provider, + requested_model, + requested_effort, + escalation_reason, + recorded_at + ) + VALUES ( + ${event.eventId}, + ${event.threadId}, + ${event.nativeSessionId}, + ${event.routeEventKind}, + ${event.taskStratum}, + ${event.experimentId}, + ${event.managerId}, + ${event.agentId}, + ${event.requestedProvider}, + ${event.requestedModel}, + ${event.requestedEffort}, + ${event.escalationReason}, + ${event.recordedAt} + ) + `, + }); + + /** + * Builds the route events a single upsert implies. At most two: an automatic + * request record (what T3 was asked to run) and, when supplied, one declared + * experiment/fallback/escalation event. Ids are deterministic so repeated + * status writes collapse instead of accumulating duplicates. + */ + const routeEventsFor = ( + runtime: { + readonly threadId: string; + readonly providerName: string; + readonly providerInstanceId: string | null; + readonly adapterKey: string; + readonly lastSeenAt: string; + readonly resumeCursor: unknown; + }, + attribution: ProviderSessionRuntimeAttributionOptions | undefined, + nativeSessionId: string | null, + ): ReadonlyArray<{ + readonly eventId: string; + readonly threadId: string; + readonly nativeSessionId: string | null; + readonly routeEventKind: string | null; + readonly taskStratum: string; + readonly experimentId: string | null; + readonly managerId: string | null; + readonly agentId: string | null; + readonly requestedProvider: string | null; + readonly requestedModel: string | null; + readonly requestedEffort: string | null; + readonly escalationReason: string | null; + readonly recordedAt: string; + }> => { + const events = []; + const requested = attribution?.requestedRoute ?? null; + if ( + requested !== null && + (requested.provider ?? requested.model ?? requested.effort) !== null + ) { + events.push({ + eventId: `${runtime.threadId}::request::${nativeSessionId ?? "unbound"}`, + threadId: runtime.threadId, + nativeSessionId, + routeEventKind: null, + taskStratum: "unknown", + experimentId: null, + managerId: null, + agentId: null, + requestedProvider: requested.provider, + requestedModel: requested.model, + requestedEffort: requested.effort, + escalationReason: null, + recordedAt: runtime.lastSeenAt, + }); + } + const declared = attribution?.routeEvent; + if (declared !== undefined && declared !== null) { + const kind = declared.kind ?? null; + events.push({ + eventId: + declared.eventId?.trim() || + `${runtime.threadId}::declared::${kind ?? "unclassified"}::${nativeSessionId ?? "thread"}`, + threadId: runtime.threadId, + nativeSessionId: declared.nativeSessionId ?? nativeSessionId, + routeEventKind: kind, + taskStratum: normalizeTaskStratum(declared.taskStratum), + experimentId: declared.experimentId ?? null, + managerId: declared.managerId ?? null, + agentId: declared.agentId ?? null, + requestedProvider: declared.requested?.provider ?? null, + requestedModel: declared.requested?.model ?? null, + requestedEffort: declared.requested?.effort ?? null, + escalationReason: declared.reason ?? null, + recordedAt: runtime.lastSeenAt, + }); + } + return events; + }; + const recordImportedTranscriptRow = SqlSchema.void({ Request: RecordImportedTranscriptRequestSchema, execute: ({ threadId, source }) => @@ -365,7 +592,31 @@ export const make = Effect.gen(function* () { }); const upsert: ProviderSessionRuntimeRepository["Service"]["upsert"] = (runtime, options) => - (options?.onConflict === "ignore" ? insertRuntimeRow(runtime) : upsertRuntimeRow(runtime)).pipe( + Effect.gen(function* () { + if (options?.onConflict === "ignore") { + // A conflicting write is a stale caller; it must not append history for + // a cursor that was never applied. + yield* insertRuntimeRow(runtime); + return; + } + yield* upsertRuntimeRow(runtime); + const attribution = options?.attribution; + const nativeSessionId = nativeSessionIdOf(runtime.resumeCursor); + if (nativeSessionId !== null) { + yield* recordSessionHistoryRow({ + threadId: runtime.threadId, + providerName: runtime.providerName, + providerInstanceId: runtime.providerInstanceId, + adapterKey: runtime.adapterKey, + nativeSessionId, + parentNativeSessionId: attribution?.parentNativeSessionId ?? null, + seenAt: runtime.lastSeenAt, + }); + } + for (const event of routeEventsFor(runtime, attribution, nativeSessionId)) { + yield* recordRouteEventRow(event); + } + }).pipe( Effect.mapError( toPersistenceSqlOrDecodeError( "ProviderSessionRuntimeRepository.upsert:query", diff --git a/apps/server/src/provider/Layers/ProviderService.ts b/apps/server/src/provider/Layers/ProviderService.ts index cdac979c4dfd..a88cfdef5ffe 100644 --- a/apps/server/src/provider/Layers/ProviderService.ts +++ b/apps/server/src/provider/Layers/ProviderService.ts @@ -55,6 +55,7 @@ import * as Stream from "effect/Stream"; import { appendUserInputAttachmentPaths } from "../userInputAttachments.ts"; import { resolveAttachmentPath } from "../../attachmentStore.ts"; +import type { RouteSelectionMetadata } from "../../usage/routeMetadata.ts"; import * as ServerConfig from "../../config.ts"; import * as DeviceService from "../../device/DeviceService.ts"; import { ensureAgentDeviceShim } from "../../device/AgentDeviceShim.ts"; @@ -327,6 +328,27 @@ function turnEffort(modelSelection: ProviderSendTurnInput["modelSelection"]): st ); } +/** + * The route T3 was actually asked to run at session start. This is the + * *requested* selection only: the observed model comes from measured usage and + * is never copied from here. Extra options (agent/variant) are deliberately + * excluded because they are not a portable effort value. + */ +function requestedRouteOf( + provider: ProviderDriverKind, + modelSelection: ModelSelection | null | undefined, +): RouteSelectionMetadata { + const model = + typeof modelSelection?.model === "string" && modelSelection.model.trim().length > 0 + ? modelSelection.model.trim() + : null; + const effort = + getModelSelectionStringOptionValue(modelSelection, "reasoningEffort")?.trim() || + getModelSelectionStringOptionValue(modelSelection, "effort")?.trim() || + null; + return { provider, model, effort }; +} + type ProviderServiceMethod = ProviderService.ProviderService["Service"][Name]; @@ -1054,7 +1076,7 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( session: ProviderSession, threadId: ThreadId, extra?: { - readonly modelSelection?: unknown; + readonly modelSelection?: ModelSelection | null | undefined; readonly continueAfterServerUpdate?: TurnId; readonly lastRuntimeEvent?: string; readonly lastRuntimeEventAt?: string; @@ -1073,6 +1095,7 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( status: toRuntimeStatus(session), ...(session.resumeCursor !== undefined ? { resumeCursor: session.resumeCursor } : {}), runtimePayload: toRuntimePayloadFromSession(session, extra), + requestedRoute: requestedRouteOf(session.provider, extra?.modelSelection), }); }); diff --git a/apps/server/src/provider/Layers/ProviderSessionDirectory.ts b/apps/server/src/provider/Layers/ProviderSessionDirectory.ts index 29ec8d2ed168..2cc09657d1bc 100644 --- a/apps/server/src/provider/Layers/ProviderSessionDirectory.ts +++ b/apps/server/src/provider/Layers/ProviderSessionDirectory.ts @@ -148,7 +148,14 @@ const makeProviderSessionDirectory = Effect.gen(function* () { binding.runtimePayload, ), }, - options, + { + ...options, + attribution: { + parentNativeSessionId: binding.parentNativeSessionId ?? null, + requestedRoute: binding.requestedRoute ?? null, + routeEvent: binding.routeEvent ?? null, + }, + }, ) .pipe(Effect.mapError(toPersistenceError("ProviderSessionDirectory.upsert:upsert"))); }); diff --git a/apps/server/src/provider/Services/ProviderSessionDirectory.ts b/apps/server/src/provider/Services/ProviderSessionDirectory.ts index 9dbafd3e804e..15525b65930a 100644 --- a/apps/server/src/provider/Services/ProviderSessionDirectory.ts +++ b/apps/server/src/provider/Services/ProviderSessionDirectory.ts @@ -14,6 +14,7 @@ import type { ProviderSessionDirectoryPersistenceError, ProviderValidationError, } from "../Errors.ts"; +import type { RouteEventInput, RouteSelectionMetadata } from "../../usage/routeMetadata.ts"; export interface ProviderRuntimeBinding { readonly threadId: ThreadId; @@ -29,6 +30,22 @@ export interface ProviderRuntimeBinding { readonly resumeCursor?: unknown | null; readonly runtimePayload?: unknown | null; readonly runtimeMode?: RuntimeMode; + /** + * True sub-agent parent native session id, when the caller knows it. Kept + * separate from the resume cursor so child sessions stay distinguishable + * instead of being flattened into the parent. + */ + readonly parentNativeSessionId?: string | null; + /** + * What this session was asked to run. Persisted as a request record; it is + * never treated as the observed model. + */ + readonly requestedRoute?: RouteSelectionMetadata | null; + /** + * A declared canary/fallback/review/escalation event. T3 only carries it; + * the route authority (agent-config policy / OMP Skill) decides it. + */ + readonly routeEvent?: RouteEventInput | null; } export interface ProviderRuntimeBindingWithMetadata extends ProviderRuntimeBinding { diff --git a/apps/server/src/usage/routeMetadata.ts b/apps/server/src/usage/routeMetadata.ts new file mode 100644 index 000000000000..69e677b75b07 --- /dev/null +++ b/apps/server/src/usage/routeMetadata.ts @@ -0,0 +1,173 @@ +/** + * Pure types and parsers for pre-execution route and experiment metadata. + * + * T3 carries this metadata so the coding-model canary can be measured across + * resumes, model switches, child sessions, and retries. T3 does not decide + * which route a task should take: `route_event_kind` records what the route + * authority (agent-config policy / an OMP Skill) declared, and the requested + * selection records what T3 was actually asked to run. Observed/actual values + * are derived from measured usage elsewhere and are never stored here, so a + * requested value can never be mistaken for an observed one. + * + * Everything is deliberately low-cardinality and content-free: no prompts, no + * responses, no code, no tool bodies. + * + * @module routeMetadata + */ + +/** Why a route event happened. `null` means "request only, not classified". */ +export const ROUTE_EVENT_KINDS = [ + "normal", + "availability_fallback", + "canary", + "independent_review", + "quality_escalation", +] as const; +export type RouteEventKind = (typeof ROUTE_EVENT_KINDS)[number]; + +/** + * A bounded, coarse task class known *before* execution. `unknown` is a first + * class value: the caller must be able to abstain rather than guess. + */ +export const TASK_STRATA = [ + "investigation", + "docs", + "tests", + "simple_edit", + "implementation", + "review", + "ci_repair", + "architecture", + "security", + "unknown", +] as const; +export type TaskStratum = (typeof TASK_STRATA)[number]; + +const ROUTE_EVENT_KIND_SET: ReadonlySet = new Set(ROUTE_EVENT_KINDS); +const TASK_STRATUM_SET: ReadonlySet = new Set(TASK_STRATA); + +export function isRouteEventKind(value: unknown): value is RouteEventKind { + return typeof value === "string" && ROUTE_EVENT_KIND_SET.has(value); +} + +export function isTaskStratum(value: unknown): value is TaskStratum { + return typeof value === "string" && TASK_STRATUM_SET.has(value); +} + +/** Coerces anything unrecognized (including `null`) to the `unknown` stratum. */ +export function normalizeTaskStratum(value: unknown): TaskStratum { + return isTaskStratum(value) ? value : "unknown"; +} + +/** A non-empty trimmed string, or `null`. Never invents a value. */ +export function readOptionalString(value: unknown): string | null { + if (typeof value !== "string") return null; + const trimmed = value.trim(); + return trimmed.length === 0 ? null : trimmed; +} + +/** + * The provider/model/effort a route decision asked for. Each field is + * independently optional: a caller that knows the model but not the effort + * must leave the effort `null` rather than default it. + */ +export interface RouteSelectionMetadata { + readonly provider: string | null; + readonly model: string | null; + readonly effort: string | null; +} + +/** + * How a selection field is grounded. + * - `declared` — a route event supplied it (requested side). + * - `observed` — a measured source established it (actual side). + * - `unsupported` — the source cannot expose the field. + * - `unknown` — nothing supplied it; never treated as a value. + */ +export type RouteSelectionQuality = "declared" | "observed" | "unsupported" | "unknown"; + +export function hasRouteSelectionValue(selection: RouteSelectionMetadata | null): boolean { + return ( + selection !== null && + (selection.provider !== null || selection.model !== null || selection.effort !== null) + ); +} + +/** Normalizes an arbitrary selection-shaped value; all-null collapses to `null`. */ +export function normalizeRouteSelection(value: unknown): RouteSelectionMetadata | null { + if (value === null || value === undefined || typeof value !== "object") return null; + const record = value as Record; + const selection: RouteSelectionMetadata = { + provider: readOptionalString(record["provider"]), + model: readOptionalString(record["model"]), + effort: readOptionalString(record["effort"]), + }; + return hasRouteSelectionValue(selection) ? selection : null; +} + +/** + * A normalized, persisted route event. + * + * `kind === null` marks an automatic request record (T3 wrote down what it was + * asked to run) rather than a classified experiment/fallback/escalation event. + */ +export interface RouteEventMetadata { + readonly eventId: string; + readonly threadId: string; + readonly nativeSessionId: string | null; + readonly kind: RouteEventKind | null; + readonly taskStratum: TaskStratum; + readonly experimentId: string | null; + readonly managerId: string | null; + readonly agentId: string | null; + readonly requested: RouteSelectionMetadata | null; + /** Availability-fallback or quality-escalation reason, when declared. */ + readonly reason: string | null; + readonly recordedAt: string; +} + +/** + * Input accepted at a write seam. The writer fills missing fields with + * truthful `null`/`unknown`; it never synthesizes a value. + */ +export interface RouteEventInput { + readonly eventId?: string; + readonly nativeSessionId?: string | null; + readonly kind?: RouteEventKind | null; + readonly taskStratum?: TaskStratum | null; + readonly experimentId?: string | null; + readonly managerId?: string | null; + readonly agentId?: string | null; + readonly requested?: RouteSelectionMetadata | null; + readonly reason?: string | null; +} + +/** The allowlisted `thread_route_events` row shape the projection consumes. */ +export interface PersistedRouteEventRow { + readonly eventId: string; + readonly threadId: string; + readonly nativeSessionId: string | null; + readonly routeEventKind: string | null; + readonly taskStratum: string; + readonly experimentId: string | null; + readonly managerId: string | null; + readonly agentId: string | null; + readonly requestedProvider: string | null; + readonly requestedModel: string | null; + readonly requestedEffort: string | null; + readonly escalationReason: string | null; + readonly recordedAt: string; +} + +/** The allowlisted `provider_session_history` row shape the projection reads. */ +export interface PersistedProviderSessionHistoryRow { + readonly threadId: string; + readonly providerName: string; + readonly providerInstanceId: string | null; + readonly adapterKey: string; + readonly nativeSessionId: string; + readonly parentNativeSessionId: string | null; + readonly origin: string; + readonly firstSeenAt: string; + readonly lastSeenAt: string; +} diff --git a/apps/server/src/usage/usageAttribution.ts b/apps/server/src/usage/usageAttribution.ts index 4efff2a1e740..850172204c63 100644 --- a/apps/server/src/usage/usageAttribution.ts +++ b/apps/server/src/usage/usageAttribution.ts @@ -211,9 +211,10 @@ export interface AttributionThreadBinding { /** * Where the binding came from. `runtimeCursor` is the single current cursor * on `provider_session_runtime`; `importedTranscript` is the accumulated - * imported-file metadata. Nothing else preserves a historical native id. + * imported-file metadata; `sessionHistory` is the append-only + * `provider_session_history` row that survives a cursor overwrite. */ - readonly origin: "runtimeCursor" | "importedTranscript"; + readonly origin: "runtimeCursor" | "importedTranscript" | "sessionHistory"; } /** An existing thread → pull-request link, already canonicalized by the caller. */ @@ -1199,7 +1200,12 @@ function limitationsFor( } if (input.bindings.some((binding) => binding.origin === "runtimeCursor")) { limitations.push( - "Only the newest native session id per thread is durable. Additive retention must land before historical re-attribution is possible.", + "Only the newest native session id per thread is durable on `provider_session_runtime`; earlier ids survive only when the caller also supplies append-only `provider_session_history` bindings.", + ); + } + if (!input.bindings.some((binding) => binding.origin === "sessionHistory")) { + limitations.push( + "No durable session-history bindings were supplied. A resume, fork, or restart that overwrote the cursor leaves earlier usage unattributed to the thread.", ); } return limitations; diff --git a/apps/server/src/usage/usageAttributionSources.test.ts b/apps/server/src/usage/usageAttributionSources.test.ts index 265f90cbfbbe..e2e18423e77e 100644 --- a/apps/server/src/usage/usageAttributionSources.test.ts +++ b/apps/server/src/usage/usageAttributionSources.test.ts @@ -2,11 +2,17 @@ import { describe, expect, it } from "@effect/vitest"; import { extractAttributionBindings, + extractAttributionHistory, extractAttributionLinks, + extractAttributionRouteEvents, extractAttributionSnapshot, type PersistedProviderSessionRuntimeRow, type PersistedThreadPullRequestRow, } from "./usageAttributionSources.ts"; +import type { + PersistedProviderSessionHistoryRow, + PersistedRouteEventRow, +} from "./routeMetadata.ts"; function runtimeRow( overrides: Partial = {}, @@ -201,6 +207,153 @@ describe("extractAttributionBindings", () => { }); }); +describe("extractAttributionHistory", () => { + function historyRow( + overrides: Partial = {}, + ): PersistedProviderSessionHistoryRow { + return { + threadId: "thread-1", + providerName: "codex", + providerInstanceId: "codex-default", + adapterKey: "codex", + nativeSessionId: "session-a", + parentNativeSessionId: null, + origin: "runtimeCursor", + firstSeenAt: "2026-09-23T09:00:00.000Z", + lastSeenAt: "2026-09-23T09:10:00.000Z", + ...overrides, + }; + } + + it("keeps every session a thread used, not just the current cursor", () => { + const { history, bindings, diagnostics } = extractAttributionHistory([ + historyRow({ nativeSessionId: "session-a" }), + historyRow({ + nativeSessionId: "session-b", + firstSeenAt: "2026-09-23T09:30:00.000Z", + lastSeenAt: "2026-09-23T09:40:00.000Z", + }), + ]); + + expect(history.map((entry) => entry.nativeSessionId)).toEqual(["session-a", "session-b"]); + expect(bindings).toEqual([ + { + threadId: "thread-1", + provider: "codex", + providerInstanceId: "codex-default", + nativeSessionId: "session-a", + origin: "sessionHistory", + }, + { + threadId: "thread-1", + provider: "codex", + providerInstanceId: "codex-default", + nativeSessionId: "session-b", + origin: "sessionHistory", + }, + ]); + expect(diagnostics).toMatchObject({ rows: 2, sessions: 2, bindings: 2, withParent: 0 }); + }); + + it("keeps an unmeasured provider identity without inventing a usage provider", () => { + const { history, bindings, diagnostics } = extractAttributionHistory([ + historyRow({ + providerName: "opencode", + adapterKey: "opencode", + nativeSessionId: "ses_child", + parentNativeSessionId: "ses_parent", + }), + ]); + + expect(bindings).toEqual([]); + expect(history[0]).toMatchObject({ + nativeSessionId: "ses_child", + parentNativeSessionId: "ses_parent", + usageProvider: null, + }); + expect(diagnostics.unsupportedProviderBindings).toBe(1); + expect(diagnostics.withParent).toBe(1); + }); + + it("counts malformed rows", () => { + const { diagnostics } = extractAttributionHistory([ + historyRow(), + historyRow({ threadId: "" }), + historyRow({ nativeSessionId: "" }), + ]); + + expect(diagnostics.sessions).toBe(1); + expect(diagnostics.malformed).toBe(2); + }); +}); + +describe("extractAttributionRouteEvents", () => { + function routeEventRow(overrides: Partial = {}): PersistedRouteEventRow { + return { + eventId: "evt-1", + threadId: "thread-1", + nativeSessionId: "session-a", + routeEventKind: "canary", + taskStratum: "implementation", + experimentId: "canary-2026-09", + managerId: "ROUTE6-1", + agentId: "T3", + requestedProvider: "openai", + requestedModel: "gpt-6-luna", + requestedEffort: "high", + escalationReason: null, + recordedAt: "2026-09-23T09:00:00.000Z", + ...overrides, + }; + } + + it("reads a declared event with its requested selection and stratum", () => { + const { events, diagnostics } = extractAttributionRouteEvents([routeEventRow()]); + + expect(events).toEqual([ + { + eventId: "evt-1", + threadId: "thread-1", + nativeSessionId: "session-a", + kind: "canary", + taskStratum: "implementation", + experimentId: "canary-2026-09", + managerId: "ROUTE6-1", + agentId: "T3", + requested: { provider: "openai", model: "gpt-6-luna", effort: "high" }, + reason: null, + recordedAt: "2026-09-23T09:00:00.000Z", + }, + ]); + expect(diagnostics).toMatchObject({ events: 1, declared: 1, requests: 0, malformed: 0 }); + }); + + it("separates an unclassified request record from a declared event", () => { + const { events, diagnostics } = extractAttributionRouteEvents([ + routeEventRow({ eventId: "req-1", routeEventKind: null, taskStratum: "", managerId: null }), + ]); + + expect(events[0]).toMatchObject({ kind: null, taskStratum: "unknown" }); + expect(events[0]!.requested).toEqual({ + provider: "openai", + model: "gpt-6-luna", + effort: "high", + }); + expect(diagnostics).toMatchObject({ declared: 0, requests: 1 }); + }); + + it("rejects an unknown kind and a blank event id", () => { + const { events, diagnostics } = extractAttributionRouteEvents([ + routeEventRow({ routeEventKind: "made-up-kind" }), + routeEventRow({ eventId: "" }), + routeEventRow({ eventId: "ok" }), + ]); + + expect(events).toHaveLength(1); + expect(diagnostics.malformed).toBe(2); + }); +}); + describe("extractAttributionLinks", () => { it("reads allowlisted fields from a persisted projection row", () => { const { links, diagnostics } = extractAttributionLinks([linkRow()]); @@ -303,6 +456,11 @@ describe("extractAttributionSnapshot", () => { * canonical `provider` (never the adapter key) and `origin`. * - `nativeSessions[]` — every native identity, including OpenCode with * `usageProvider: null`; a label, never a join key. + * - `history[]` — append-only identity from `provider_session_history` that + * survives a cursor overwrite. Empty here because this fixture supplies no + * history rows; the durable-history test below populates it. + * - `routeEvents[]` — content-free pre-execution route/experiment metadata. + * Empty here because this fixture supplies no route-event rows. * - `links[]` — canonical thread -> PR links; `stack-dismissed` tombstones are * preserved for the projection to filter. * - `diagnostics` — what could not be read, never dropped silently. @@ -436,6 +594,8 @@ describe("extractAttributionSnapshot", () => { url: "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/acme/repo/pull/12", }, ], + history: [], + routeEvents: [], diagnostics: { bindings: { runtimeRows: 3, @@ -449,8 +609,68 @@ describe("extractAttributionSnapshot", () => { overwrittenThreads: 1, ambiguousSessionIds: 0, }, + history: { + rows: 0, + sessions: 0, + bindings: 0, + withParent: 0, + unsupportedProviderBindings: 0, + malformed: 0, + }, + routeEvents: { + rows: 0, + events: 0, + declared: 0, + requests: 0, + malformed: 0, + }, links: { rows: 3, links: 3, dismissed: 0, malformed: 0 }, }, }); }); + + it("merges durable history and route events into one snapshot", () => { + const snapshot = extractAttributionSnapshot({ + cutoffMs: 1_786_100_000_000, + runtimeRows: [], + historyRows: [ + { + threadId: "thread-1", + providerName: "opencode", + providerInstanceId: "opencode-default", + adapterKey: "opencode", + nativeSessionId: "ses_child", + parentNativeSessionId: "ses_parent", + origin: "runtimeCursor", + firstSeenAt: "2026-09-23T09:00:00.000Z", + lastSeenAt: "2026-09-23T09:10:00.000Z", + }, + ], + routeEventRows: [ + { + eventId: "evt-1", + threadId: "thread-1", + nativeSessionId: "ses_child", + routeEventKind: "canary", + taskStratum: "implementation", + experimentId: "canary-2026-09", + managerId: "ROUTE6-1", + agentId: "T3", + requestedProvider: "openai", + requestedModel: "gpt-6-luna", + requestedEffort: "high", + escalationReason: null, + recordedAt: "2026-09-23T09:00:00.000Z", + }, + ], + linkRows: [], + }); + + // OpenCode is unmeasured, so it is history only, never a usage binding. + expect(snapshot.bindings).toEqual([]); + expect(snapshot.history).toHaveLength(1); + expect(snapshot.routeEvents).toHaveLength(1); + expect(snapshot.diagnostics.history.sessions).toBe(1); + expect(snapshot.diagnostics.routeEvents.declared).toBe(1); + }); }); diff --git a/apps/server/src/usage/usageAttributionSources.ts b/apps/server/src/usage/usageAttributionSources.ts index 9c1b0be09966..1deb0b5d9fc1 100644 --- a/apps/server/src/usage/usageAttributionSources.ts +++ b/apps/server/src/usage/usageAttributionSources.ts @@ -21,6 +21,15 @@ import type { ThreadPullRequestLinkSource, UsageProviderKind } from "@t3tools/contracts"; import type { AttributionPullRequestLink, AttributionThreadBinding } from "./usageAttribution.ts"; +import { + isRouteEventKind, + normalizeRouteSelection, + normalizeTaskStratum, + readOptionalString, + type PersistedProviderSessionHistoryRow, + type PersistedRouteEventRow, + type RouteEventMetadata, +} from "./routeMetadata.ts"; /** * Allowlisted `provider_session_runtime` row. This is the shape the repository @@ -111,11 +120,20 @@ export interface AttributionLinkExtraction { export interface AttributionSnapshotExtraction { /** Read cutoff; associations are as of this instant. */ readonly cutoffMs: number; + /** + * Every binding the projection should consider: the current cursor plus the + * durable history, so a session the cursor no longer points at still binds. + */ readonly bindings: readonly AttributionThreadBinding[]; readonly nativeSessions: readonly ExtractedNativeSession[]; + /** Append-only native-session identity history, including unmeasured providers. */ + readonly history: readonly ExtractedSessionHistory[]; readonly links: readonly AttributionPullRequestLink[]; + readonly routeEvents: readonly RouteEventMetadata[]; readonly diagnostics: { readonly bindings: AttributionBindingDiagnostics; + readonly history: AttributionHistoryDiagnostics; + readonly routeEvents: AttributionRouteEventDiagnostics; readonly links: AttributionLinkDiagnostics; }; } @@ -335,6 +353,184 @@ function addSessionThread( index.set(key, threads); } +const HISTORY_ORIGINS = new Set(["runtimeCursor", "importedTranscript"]); + +/** + * One durable native-session identity read from `provider_session_history`. + * Unlike the current cursor, every row survives a resume/model switch, so a + * thread's earlier sessions stay attributable. + */ +export interface ExtractedSessionHistory { + readonly threadId: string; + readonly providerName: string; + readonly adapterKey: string; + readonly providerInstanceId: string | null; + readonly nativeSessionId: string; + readonly parentNativeSessionId: string | null; + readonly source: "runtimeCursor" | "importedTranscript" | "unknown"; + readonly firstSeenAt: string; + readonly lastSeenAt: string; + readonly usageProvider: UsageProviderKind | null; +} + +export interface AttributionHistoryDiagnostics { + readonly rows: number; + readonly sessions: number; + readonly bindings: number; + readonly withParent: number; + /** Durable identities for a provider T3 does not scan for usage. */ + readonly unsupportedProviderBindings: number; + readonly malformed: number; +} + +export interface AttributionHistoryExtraction { + readonly history: readonly ExtractedSessionHistory[]; + /** Only identities for providers with a scanned usage source. */ + readonly bindings: readonly AttributionThreadBinding[]; + readonly diagnostics: AttributionHistoryDiagnostics; +} + +/** + * Reads the append-only session history. Every row is an identity that + * contributed to the thread, including providers T3 cannot measure yet + * (exposed as history with `usageProvider: null`, never as a fabricated zero). + */ +export function extractAttributionHistory( + rows: readonly PersistedProviderSessionHistoryRow[], +): AttributionHistoryExtraction { + const history: ExtractedSessionHistory[] = []; + const bindings: AttributionThreadBinding[] = []; + const diagnostics = { + rows: rows.length, + sessions: 0, + bindings: 0, + withParent: 0, + unsupportedProviderBindings: 0, + malformed: 0, + }; + + for (const row of rows) { + const threadId = readOptionalString(row.threadId); + const providerName = readOptionalString(row.providerName); + const adapterKey = readOptionalString(row.adapterKey); + const nativeSessionId = readOptionalString(row.nativeSessionId); + if ( + threadId === null || + providerName === null || + adapterKey === null || + nativeSessionId === null + ) { + diagnostics.malformed += 1; + continue; + } + const usageProvider = usageProviderOf(providerName, adapterKey); + const parentNativeSessionId = readOptionalString(row.parentNativeSessionId); + diagnostics.sessions += 1; + if (parentNativeSessionId !== null) diagnostics.withParent += 1; + history.push({ + threadId, + providerName, + adapterKey, + providerInstanceId: readOptionalString(row.providerInstanceId), + nativeSessionId, + parentNativeSessionId, + source: HISTORY_ORIGINS.has(row.origin) + ? (row.origin as ExtractedSessionHistory["source"]) + : "unknown", + firstSeenAt: readOptionalString(row.firstSeenAt) ?? row.lastSeenAt, + lastSeenAt: row.lastSeenAt, + usageProvider, + }); + if (usageProvider === null) { + diagnostics.unsupportedProviderBindings += 1; + continue; + } + diagnostics.bindings += 1; + bindings.push({ + threadId, + provider: usageProvider, + providerInstanceId: readOptionalString(row.providerInstanceId), + nativeSessionId, + origin: "sessionHistory", + }); + } + + return { history, bindings, diagnostics }; +} + +export interface AttributionRouteEventDiagnostics { + readonly rows: number; + readonly events: number; + /** Events with a classified kind (canary, fallback, review, escalation). */ + readonly declared: number; + /** Automatic "what was requested" records with no classified kind. */ + readonly requests: number; + readonly malformed: number; +} + +export interface AttributionRouteEventExtraction { + readonly events: readonly RouteEventMetadata[]; + readonly diagnostics: AttributionRouteEventDiagnostics; +} + +/** + * Reads the allowlisted fields of `thread_route_events`. Content never crosses + * this seam: only ids, the requested selection, stratum, cohort, and reason. + */ +export function extractAttributionRouteEvents( + rows: readonly PersistedRouteEventRow[], +): AttributionRouteEventExtraction { + const events: RouteEventMetadata[] = []; + const diagnostics = { + rows: rows.length, + events: 0, + declared: 0, + requests: 0, + malformed: 0, + }; + + for (const row of rows) { + const eventId = readOptionalString(row.eventId); + const threadId = readOptionalString(row.threadId); + const recordedAt = readOptionalString(row.recordedAt); + const rawKind = row.routeEventKind; + // A non-null kind that is not one of the known kinds is a malformed row, + // not an unknown-but-valid one. + if (rawKind !== null && rawKind !== undefined && !isRouteEventKind(rawKind)) { + diagnostics.malformed += 1; + continue; + } + if (eventId === null || threadId === null || recordedAt === null) { + diagnostics.malformed += 1; + continue; + } + const requested = normalizeRouteSelection({ + provider: row.requestedProvider, + model: row.requestedModel, + effort: row.requestedEffort, + }); + const kind = isRouteEventKind(rawKind) ? rawKind : null; + diagnostics.events += 1; + if (kind === null) diagnostics.requests += 1; + else diagnostics.declared += 1; + events.push({ + eventId, + threadId, + nativeSessionId: readOptionalString(row.nativeSessionId), + kind, + taskStratum: normalizeTaskStratum(row.taskStratum), + experimentId: readOptionalString(row.experimentId), + managerId: readOptionalString(row.managerId), + agentId: readOptionalString(row.agentId), + requested, + reason: readOptionalString(row.escalationReason), + recordedAt, + }); + } + + return { events, diagnostics }; +} + /** Reads thread → PR links from persisted projection rows, dropping payloads. */ export function extractAttributionLinks( rows: readonly PersistedThreadPullRequestRow[], @@ -379,15 +575,26 @@ export function extractAttributionLinks( export function extractAttributionSnapshot(input: { readonly cutoffMs: number; readonly runtimeRows: readonly PersistedProviderSessionRuntimeRow[]; + readonly historyRows?: readonly PersistedProviderSessionHistoryRow[]; readonly linkRows: readonly PersistedThreadPullRequestRow[]; + readonly routeEventRows?: readonly PersistedRouteEventRow[]; }): AttributionSnapshotExtraction { const bindings = extractAttributionBindings(input.runtimeRows); + const history = extractAttributionHistory(input.historyRows ?? []); + const routeEvents = extractAttributionRouteEvents(input.routeEventRows ?? []); const links = extractAttributionLinks(input.linkRows); return { cutoffMs: input.cutoffMs, - bindings: bindings.bindings, + bindings: [...bindings.bindings, ...history.bindings], nativeSessions: bindings.nativeSessions, + history: history.history, links: links.links, - diagnostics: { bindings: bindings.diagnostics, links: links.diagnostics }, + routeEvents: routeEvents.events, + diagnostics: { + bindings: bindings.diagnostics, + history: history.diagnostics, + routeEvents: routeEvents.diagnostics, + links: links.diagnostics, + }, }; } diff --git a/apps/server/src/usage/usageRouteAttribution.test.ts b/apps/server/src/usage/usageRouteAttribution.test.ts new file mode 100644 index 000000000000..d59d6439a263 --- /dev/null +++ b/apps/server/src/usage/usageRouteAttribution.test.ts @@ -0,0 +1,512 @@ +import { describe, expect, it } from "@effect/vitest"; + +import type { UsageTokenTotals } from "@t3tools/contracts"; + +import type { + AttributionPullRequestLink, + AttributionThreadBinding, + AttributionUsageRecord, +} from "./usageAttribution.ts"; +import type { ExtractedSessionHistory } from "./usageAttributionSources.ts"; +import type { RouteEventMetadata } from "./routeMetadata.ts"; +import { buildUsageRouteAttribution } from "./usageRouteAttribution.ts"; +import { totalTokens } from "./usageTranscripts.ts"; + +const DEEPSEEK_SESSION = "019f0000-0000-7000-8000-000000000001"; +const LUNA_SESSION = "019f0000-0000-7000-8000-000000000002"; +const OPENCODE_PARENT = "ses_parent_opencode"; +const OPENCODE_CHILD = "ses_child_opencode"; +const THREAD = "thread-route-1"; +const FINGERPRINT = "host\u0000codex\u0000/home/u/.codex\u00000:2"; + +function totals(overrides: Partial = {}): UsageTokenTotals { + return { + uncachedInputTokens: 100, + cachedInputTokens: 10, + cacheCreationTokens: 0, + outputTokens: 20, + reasoningTokens: 0, + ...overrides, + }; +} + +function record(overrides: Partial = {}): AttributionUsageRecord { + return { + provider: "codex", + sessionId: DEEPSEEK_SESSION, + model: "deepseek-v4.1-flash", + timestampMs: 1_786_000_000_000, + totals: totals(), + costUsd: 0.01, + dedupeKey: "occ:1", + sourceFingerprint: FINGERPRINT, + measurement: "observed", + ...overrides, + }; +} + +function binding(overrides: Partial = {}): AttributionThreadBinding { + return { + threadId: THREAD, + provider: "codex", + providerInstanceId: "codex-default", + nativeSessionId: DEEPSEEK_SESSION, + origin: "sessionHistory", + ...overrides, + }; +} + +function history(overrides: Partial = {}): ExtractedSessionHistory { + return { + threadId: THREAD, + providerName: "codex", + adapterKey: "codex", + providerInstanceId: "codex-default", + nativeSessionId: DEEPSEEK_SESSION, + parentNativeSessionId: null, + source: "runtimeCursor", + firstSeenAt: "2026-09-23T09:00:00.000Z", + lastSeenAt: "2026-09-23T09:10:00.000Z", + usageProvider: "codex", + ...overrides, + }; +} + +function routeEvent(overrides: Partial = {}): RouteEventMetadata { + return { + eventId: "evt-1", + threadId: THREAD, + nativeSessionId: DEEPSEEK_SESSION, + kind: "canary", + taskStratum: "implementation", + experimentId: "canary-2026-09", + managerId: "ROUTE6-1", + agentId: "T3", + requested: { provider: "openai", model: "gpt-6-luna", effort: "high" }, + reason: null, + recordedAt: "2026-09-23T09:00:00.000Z", + ...overrides, + }; +} + +function link(overrides: Partial = {}): AttributionPullRequestLink { + return { + threadId: THREAD, + host: "github.com", + repository: "acme/repo", + number: 12, + source: "manual", + linkedAt: "2026-09-23T09:20:00.000Z", + ...overrides, + }; +} + +function build(overrides: { + records?: readonly AttributionUsageRecord[]; + bindings?: readonly AttributionThreadBinding[]; + history?: readonly ExtractedSessionHistory[]; + routeEvents?: readonly RouteEventMetadata[]; + links?: readonly AttributionPullRequestLink[]; +}) { + return buildUsageRouteAttribution({ + cutoffMs: 1_786_100_000_000, + records: overrides.records ?? [], + bindings: overrides.bindings ?? [], + links: overrides.links ?? [], + sources: [], + history: overrides.history ?? [], + routeEvents: overrides.routeEvents ?? [], + }); +} + +function sessionOf(report: ReturnType, sessionId: string) { + const session = report.sessions.find((entry) => entry.sessionId === sessionId); + if (session === undefined) throw new Error(`session ${sessionId} not found`); + return session; +} + +describe("durable session history across model switches", () => { + it("retains both a DeepSeek and a Luna session for one thread", () => { + const report = build({ + records: [ + record({ sessionId: DEEPSEEK_SESSION, model: "deepseek-v4.1-flash", dedupeKey: "ds:1" }), + record({ sessionId: LUNA_SESSION, model: "gpt-6-luna", dedupeKey: "luna:1" }), + ], + bindings: [binding(), binding({ nativeSessionId: LUNA_SESSION })], + history: [ + history(), + history({ + nativeSessionId: LUNA_SESSION, + firstSeenAt: "2026-09-23T09:30:00.000Z", + lastSeenAt: "2026-09-23T09:40:00.000Z", + }), + ], + links: [link()], + }); + + expect(report.sessions).toHaveLength(2); + const deepseek = sessionOf(report, DEEPSEEK_SESSION); + const luna = sessionOf(report, LUNA_SESSION); + expect(deepseek.actualModel).toBe("deepseek-v4.1-flash"); + expect(luna.actualModel).toBe("gpt-6-luna"); + const thread = report.threads.find((entry) => entry.threadId === THREAD)!; + expect(thread.models).toEqual(["deepseek-v4.1-flash", "gpt-6-luna"]); + expect(thread.sessionLabels).toHaveLength(2); + // Both sessions are bound to the single thread, so both attribute to its PR. + const pr = report.base.pullRequests.find((entry) => entry.number === 12)!; + expect(pr.attributed.totalTokens).toBe(deepseek.usage!.totalTokens + luna.usage!.totalTokens); + }); + + it("keeps the earlier session after the resume cursor moves to a new session", () => { + const report = build({ + // Only the *current* cursor is available as a runtime binding. + bindings: [binding({ nativeSessionId: LUNA_SESSION, origin: "runtimeCursor" })], + records: [ + record({ sessionId: DEEPSEEK_SESSION, dedupeKey: "ds:1" }), + record({ sessionId: LUNA_SESSION, model: "gpt-6-luna", dedupeKey: "luna:1" }), + ], + history: [history(), history({ nativeSessionId: LUNA_SESSION })], + }); + + const deepseek = sessionOf(report, DEEPSEEK_SESSION); + expect(deepseek.usage).not.toBeNull(); + expect(deepseek.threadId).toBe(THREAD); + // Without history this session would be unbound and its usage lost. + expect(report.threads[0]!.sessionLabels).toEqual([ + `codex:${DEEPSEEK_SESSION}`, + `codex:${LUNA_SESSION}`, + ]); + }); + + it("does not flatten a child OpenCode session into its parent", () => { + const report = build({ + bindings: [], + history: [ + history({ + providerName: "opencode", + adapterKey: "opencode", + providerInstanceId: "opencode-default", + nativeSessionId: OPENCODE_PARENT, + usageProvider: null, + }), + history({ + providerName: "opencode", + adapterKey: "opencode", + providerInstanceId: "opencode-default", + nativeSessionId: OPENCODE_CHILD, + parentNativeSessionId: OPENCODE_PARENT, + usageProvider: null, + }), + ], + }); + + const child = sessionOf(report, OPENCODE_CHILD); + const parent = sessionOf(report, OPENCODE_PARENT); + expect(child.parentSessionId).toBe(OPENCODE_PARENT); + expect(parent.parentSessionId).toBeNull(); + expect(child.provider).toBeNull(); + // Unknown, never zero. + expect(child.usage).toBeNull(); + expect(report.identity.sessionsWithParent).toBe(1); + }); +}); + +describe("route and experiment metadata", () => { + it("keeps a readable manager/agent id distinct from the native session id", () => { + const report = build({ + records: [record()], + bindings: [binding()], + history: [history()], + routeEvents: [routeEvent()], + }); + + const session = sessionOf(report, DEEPSEEK_SESSION); + expect(session.managerId).toBe("ROUTE6-1"); + expect(session.agentId).toBe("T3"); + expect(session.managerId).not.toBe(session.sessionId); + expect(session.agentId).not.toBe(session.sessionId); + expect(report.identity.managerIds).toEqual(["ROUTE6-1"]); + expect(report.identity.agentIds).toEqual(["T3"]); + }); + + it("keeps requested and observed models separate and never copies one into the other", () => { + const report = build({ + records: [record({ sessionId: LUNA_SESSION, model: "gpt-6-luna-2026-09", dedupeKey: "l:1" })], + bindings: [binding({ nativeSessionId: LUNA_SESSION })], + history: [history({ nativeSessionId: LUNA_SESSION })], + routeEvents: [routeEvent({ nativeSessionId: LUNA_SESSION })], + }); + + const session = sessionOf(report, LUNA_SESSION); + expect(session.requested).toEqual({ + provider: "openai", + model: "gpt-6-luna", + effort: "high", + }); + expect(session.requestedQuality).toBe("declared"); + expect(session.actualModel).toBe("gpt-6-luna-2026-09"); + expect(session.actualModel).not.toBe(session.requested!.model); + }); + + it("leaves unsupported actual effort unknown rather than copying the request", () => { + const report = build({ + records: [record()], + bindings: [binding()], + history: [history()], + routeEvents: [routeEvent()], + }); + + const session = sessionOf(report, DEEPSEEK_SESSION); + expect(session.requested!.effort).toBe("high"); + expect(session.actualEffort).toBeNull(); + expect(session.actualEffortQuality).toBe("unsupported"); + }); + + it("distinguishes an availability fallback from an independent review", () => { + const report = build({ + records: [ + record({ sessionId: DEEPSEEK_SESSION, dedupeKey: "ds:1" }), + record({ sessionId: LUNA_SESSION, model: "gpt-6-luna", dedupeKey: "luna:1" }), + ], + bindings: [binding(), binding({ nativeSessionId: LUNA_SESSION })], + history: [history(), history({ nativeSessionId: LUNA_SESSION })], + routeEvents: [ + routeEvent({ + eventId: "fallback-1", + kind: "availability_fallback", + reason: "provider 503 during send", + }), + routeEvent({ + eventId: "review-1", + nativeSessionId: LUNA_SESSION, + kind: "independent_review", + reason: "independent review gate", + recordedAt: "2026-09-23T09:35:00.000Z", + }), + ], + }); + + expect(sessionOf(report, DEEPSEEK_SESSION).routeEventKind).toBe("availability_fallback"); + expect(sessionOf(report, DEEPSEEK_SESSION).escalationReason).toBe("provider 503 during send"); + expect(sessionOf(report, LUNA_SESSION).routeEventKind).toBe("independent_review"); + expect(sessionOf(report, LUNA_SESSION).escalationReason).toBe("independent review gate"); + const thread = report.threads.find((entry) => entry.threadId === THREAD)!; + expect(thread.routeEventKinds).toEqual(["availability_fallback", "independent_review"]); + }); + + it("preserves a quality escalation reason", () => { + const report = build({ + records: [record()], + bindings: [binding()], + history: [history()], + routeEvents: [ + routeEvent({ + kind: "quality_escalation", + reason: "two materially similar failed repair attempts", + }), + ], + }); + + const session = sessionOf(report, DEEPSEEK_SESSION); + expect(session.routeEventKind).toBe("quality_escalation"); + expect(session.escalationReason).toBe("two materially similar failed repair attempts"); + expect(session.taskStratum).toBe("implementation"); + expect(session.experimentId).toBe("canary-2026-09"); + }); + + it("reports an unknown stratum and unclassified kind when nothing was declared", () => { + const report = build({ + records: [record()], + bindings: [binding()], + history: [history()], + }); + + const session = sessionOf(report, DEEPSEEK_SESSION); + expect(session.routeEventKind).toBeNull(); + expect(session.taskStratum).toBe("unknown"); + expect(session.requested).toBeNull(); + expect(session.requestedQuality).toBe("unknown"); + }); +}); + +describe("pull-request association without duplicated usage", () => { + it("links one thread to multiple PRs without duplicating its usage", () => { + const only = record({ dedupeKey: "ds:1" }); + const report = build({ + records: [only], + bindings: [binding()], + history: [history()], + links: [link({ number: 12 }), link({ number: 13 })], + }); + + const thread = report.threads.find((entry) => entry.threadId === THREAD)!; + expect(thread.pullRequestKeys).toEqual(["github.com/acme/repo#12", "github.com/acme/repo#13"]); + expect(thread.usage.totalTokens).toBe(totalTokens(only.totals)); + // Association is not attribution: a session on two PRs sits in `shared`, + // and its total is not cloned onto each PR. + expect(report.base.shared.totalTokens).toBe(totalTokens(only.totals)); + for (const pr of report.base.pullRequests) { + expect(pr.attributed.totalTokens).toBe(0); + expect(pr.shared.totalTokens).toBe(totalTokens(only.totals)); + } + }); + + it("does not attribute usage through a dismissed PR link", () => { + const report = build({ + records: [record()], + bindings: [binding()], + history: [history()], + links: [link({ number: 14, source: "stack-dismissed" })], + }); + + expect(report.base.pullRequests).toHaveLength(0); + const thread = report.threads.find((entry) => entry.threadId === THREAD)!; + expect(thread.pullRequestKeys).toEqual([]); + // The usage is still measured; it is unallocated, not zero. + expect(report.base.measured.totalTokens).toBe(totalTokens(record().totals)); + expect(report.base.unallocated.totalTokens).toBe(totalTokens(record().totals)); + }); + + it("treats missing usage as unknown, never zero", () => { + const missingSession = "019f0000-0000-7000-8000-0000000000aa"; + const report = build({ + bindings: [binding({ nativeSessionId: missingSession })], + history: [history({ nativeSessionId: missingSession })], + }); + + const session = sessionOf(report, missingSession); + expect(session.usage).toBeNull(); + // The provider is known from the durable binding; the model is unknown + // because no usage was measured. Neither is a zero. + expect(session.actualProvider).toBe("codex"); + expect(session.actualModel).toBeNull(); + const baseSession = report.base.sessions.find((entry) => entry.sessionId === missingSession)!; + expect(baseSession.measurementQuality).toBe("missing"); + const thread = report.threads.find((entry) => entry.threadId === THREAD)!; + expect(thread.usage.totalTokens).toBe(0); + expect(thread.usage.records).toBe(0); + }); +}); + +describe("durable history keeps #4 measurement and identity quality", () => { + // Claude has a supported request level, so an erased native identity is + // observable as `unavailable` rather than collapsing to `missing`. + const CLAUDE_SESSION = "5a128faa-8253-489e-b935-6c08e8e670c0"; + + function claudeBinding() { + return binding({ provider: "claude", nativeSessionId: CLAUDE_SESSION }); + } + + function claudeHistory() { + return history({ + providerName: "claudeAgent", + adapterKey: "claudeAgent", + nativeSessionId: CLAUDE_SESSION, + usageProvider: "claude", + }); + } + + it("carries partial measurement completeness into the route view", () => { + const report = build({ + records: [ + record({ + provider: "claude", + sessionId: CLAUDE_SESSION, + model: "claude-fable-5", + dedupeKey: "m1:", + measurement: "observed", + measurementCompleteness: "partial", + invalidTokenFields: 1, + }), + ], + bindings: [claudeBinding()], + history: [claudeHistory()], + }); + + const session = sessionOf(report, CLAUDE_SESSION); + expect(session.quality?.measurement).toBe("partial"); + expect(session.quality?.identity).toBe("valid"); + // The base projection still carries the same axis, so the route view is a + // faithful mirror rather than a second, divergent computation. + const baseSession = report.base.sessions.find((entry) => entry.sessionId === CLAUDE_SESSION)!; + expect(baseSession.measurementQuality).toBe(session.quality?.measurement); + }); + + it("keeps a legacy identity-erased durable row unavailable, not missing", () => { + const report = build({ + records: [ + record({ + provider: "claude", + sessionId: CLAUDE_SESSION, + model: "claude-fable-5", + dedupeKey: "legacy:1", + measurement: "observed", + measurementCompleteness: "partial", + identityAvailable: false, + totals: totals({ outputTokens: 40 }), + }), + ], + bindings: [claudeBinding()], + history: [claudeHistory()], + }); + + const session = sessionOf(report, CLAUDE_SESSION); + // The erased native id leaves request identity `unavailable`, never + // `missing`, and the nonzero total never implies a complete measurement. + expect(session.quality?.request).toBe("unavailable"); + expect(session.quality?.measurement).toBe("partial"); + }); + + it("keeps an invalid measurement invalid rather than measured", () => { + const report = build({ + records: [ + record({ + provider: "claude", + sessionId: CLAUDE_SESSION, + model: "claude-fable-5", + dedupeKey: "m1:", + measurement: "invalid", + }), + ], + bindings: [claudeBinding()], + history: [claudeHistory()], + }); + + expect(sessionOf(report, CLAUDE_SESSION).quality?.measurement).toBe("invalid"); + }); + + it("surfaces a cost-only conflict instead of silently deduping", () => { + const report = build({ + records: [ + record({ dedupeKey: "same:1", costUsd: 0.01 }), + record({ dedupeKey: "same:1", costUsd: 0.02 }), + ], + bindings: [binding()], + history: [history()], + }); + + const session = sessionOf(report, DEEPSEEK_SESSION); + expect(session.quality?.conflict).toBe(true); + expect(session.quality?.recordIdentity).toBe("exact"); + }); + + it("reports no measurement quality for a history-only unmeasured identity", () => { + const report = build({ + bindings: [], + history: [ + history({ + providerName: "opencode", + adapterKey: "opencode", + nativeSessionId: OPENCODE_CHILD, + usageProvider: null, + }), + ], + }); + + const child = sessionOf(report, OPENCODE_CHILD); + // Unknown usage has no measurement to qualify: null, never a fabricated zero + // or a fabricated `measured`. + expect(child.usage).toBeNull(); + expect(child.quality).toBeNull(); + }); +}); diff --git a/apps/server/src/usage/usageRouteAttribution.ts b/apps/server/src/usage/usageRouteAttribution.ts new file mode 100644 index 000000000000..88d7f8c9b63d --- /dev/null +++ b/apps/server/src/usage/usageRouteAttribution.ts @@ -0,0 +1,446 @@ +/** + * Route, identity, and experiment view over the usage attribution projection. + * + * `buildUsageAttribution` answers "how much usage did each native session and + * pull request get". This companion module answers the canary's question: "which + * models, sessions, and assignments contributed to this T3 thread, and what was + * requested versus observed". It composes the base projection rather than + * duplicating it, and it adds only what the base levels cannot express: + * + * - readable manager/agent identity, kept explicitly distinct from the native + * session id (a label is never a substitute for a real session id); + * - requested provider/model/effort from route metadata versus the observed + * model from measured usage (the two are never copied into each other); + * - pre-execution task stratum, experiment/cohort, route event kind, and the + * fallback/escalation reason; + * - parent/child session lineage so a child session is never flattened into + * its parent's usage. + * + * It is pure. It never reads the clock, the filesystem, or the database, and it + * never sees a prompt, response, or tool body. + * + * @module usageRouteAttribution + */ +import type { UsageProviderKind } from "@t3tools/contracts"; + +import { + buildUsageAttribution, + type AttributionIdentityQuality, + type AttributionPullRequestLink, + type AttributionQuality, + type AttributionSessionReport, + type AttributionSource, + type AttributionThreadBinding, + type AttributionTotals, + type AttributionUsageRecord, + type UsageAttribution, +} from "./usageAttribution.ts"; +import type { ExtractedSessionHistory } from "./usageAttributionSources.ts"; +import { + hasRouteSelectionValue, + type RouteEventKind, + type RouteEventMetadata, + type RouteSelectionMetadata, + type RouteSelectionQuality, + type TaskStratum, +} from "./routeMetadata.ts"; +import { addTotals, EMPTY_TOTALS } from "./usageTranscripts.ts"; + +export const USAGE_ROUTE_ATTRIBUTION_VERSION = 1 as const; + +export interface UsageRouteAttributionInput { + /** Read cutoff; associations are as of this instant. */ + readonly cutoffMs: number; + readonly records: readonly AttributionUsageRecord[]; + /** Bindings must already include durable session-history bindings. */ + readonly bindings: readonly AttributionThreadBinding[]; + readonly links: readonly AttributionPullRequestLink[]; + readonly sources: readonly AttributionSource[]; + /** Append-only native-session identity history, including unmeasured providers. */ + readonly history: readonly ExtractedSessionHistory[]; + /** Pre-execution route and experiment metadata. */ + readonly routeEvents: readonly RouteEventMetadata[]; +} + +/** + * The base session report's measurement and identity axes, carried on the route + * session so the follow-on view cannot silently drop them and present a partial + * or invalid measurement — or a legacy identity-erased row — as exact. + */ +export interface SessionRouteQuality { + /** Numeric completeness: `measured | partial | invalid | missing | unavailable`. */ + readonly measurement: AttributionQuality; + /** Native-identity validity, independent of the measurement. */ + readonly identity: AttributionIdentityQuality; + /** Prompt-level identity quality. */ + readonly prompt: AttributionQuality; + /** Request-level identity quality; a legacy identity-erased row is `unavailable`. */ + readonly request: AttributionQuality; + /** `uncertain` when a contributing record had no scan/delivery identity. */ + readonly recordIdentity: "exact" | "uncertain"; + /** `true` when two versions of one identity disagreed (including cost-only). */ + readonly conflict: boolean; +} + +export interface SessionRouteReport { + /** `null` when T3 has no scanned usage source for the provider (OpenCode). */ + readonly provider: UsageProviderKind | null; + readonly sessionId: string; + readonly threadId: string | null; + readonly models: readonly string[]; + /** `null` means unknown, never zero. Absence is not a measured zero. */ + readonly usage: AttributionTotals | null; + /** + * Measurement and identity quality from the base projection, or `null` when no + * usage was measured for this identity (so there is no measurement to qualify). + */ + readonly quality: SessionRouteQuality | null; + readonly parentSessionId: string | null; + readonly requested: RouteSelectionMetadata | null; + readonly requestedQuality: RouteSelectionQuality; + readonly actualProvider: string | null; + readonly actualModel: string | null; + readonly actualEffort: string | null; + /** No scanned source exposes reasoning effort, so this is always `unsupported`. */ + readonly actualEffortQuality: RouteSelectionQuality; + readonly routeEventKind: RouteEventKind | null; + readonly taskStratum: TaskStratum; + readonly experimentId: string | null; + readonly managerId: string | null; + readonly agentId: string | null; + readonly escalationReason: string | null; +} + +export interface ThreadRouteReport { + readonly threadId: string; + readonly sessionLabels: readonly string[]; + readonly models: readonly string[]; + /** Additive over this thread's sessions; misses nothing, double counts nothing. */ + readonly usage: AttributionTotals; + readonly pullRequestKeys: readonly string[]; + readonly experimentIds: readonly string[]; + readonly managerIds: readonly string[]; + readonly agentIds: readonly string[]; + readonly routeEventKinds: readonly RouteEventKind[]; +} + +export interface RouteIdentityDiagnostics { + readonly managerIds: readonly string[]; + readonly agentIds: readonly string[]; + readonly sessionsWithRouteMetadata: number; + readonly sessionsWithParent: number; +} + +export interface UsageRouteAttribution { + readonly contractVersion: typeof USAGE_ROUTE_ATTRIBUTION_VERSION; + readonly generatedAtMs: number; + /** The unchanged base projection, so callers keep the proven levels. */ + readonly base: UsageAttribution; + readonly sessions: readonly SessionRouteReport[]; + readonly threads: readonly ThreadRouteReport[]; + readonly routeEvents: readonly RouteEventMetadata[]; + readonly identity: RouteIdentityDiagnostics; + readonly limitations: readonly string[]; +} + +const ZERO: AttributionTotals = { + tokens: EMPTY_TOTALS, + totalTokens: 0, + costUsd: 0, + records: 0, +}; + +function addTotalsOf(left: AttributionTotals, right: AttributionTotals): AttributionTotals { + return { + tokens: addTotals(left.tokens, right.tokens), + totalTokens: left.totalTokens + right.totalTokens, + costUsd: left.costUsd + right.costUsd, + records: left.records + right.records, + }; +} + +function sessionKey(provider: UsageProviderKind, sessionId: string): string { + return `${provider}\u0000${sessionId}`; +} + +function sessionLabel(provider: UsageProviderKind | null, sessionId: string): string { + return `${provider ?? "unmeasured"}:${sessionId}`; +} + +function sortedUnique(values: Iterable): string[] { + return [...new Set(values)].toSorted((left, right) => left.localeCompare(right)); +} + +function compareEvents(left: RouteEventMetadata, right: RouteEventMetadata): number { + if (left.recordedAt !== right.recordedAt) return left.recordedAt < right.recordedAt ? -1 : 1; + return left.eventId.localeCompare(right.eventId); +} + +/** + * Picks the route event that best describes one session: a declared event for + * the exact session first, then a declared event for its thread, then the + * automatic request record. A declared event always outranks a request record + * because it carries the classification the canary cares about. + */ +function selectRouteEvent( + sessionId: string, + threadId: string | null, + declaredByThread: ReadonlyMap, + requestsByThread: ReadonlyMap, +): RouteEventMetadata | null { + const consider = ( + events: readonly RouteEventMetadata[] | undefined, + ): RouteEventMetadata | null => { + if (events === undefined || events.length === 0) return null; + const exact = events.find((event) => event.nativeSessionId === sessionId); + return exact ?? events[0]!; + }; + if (threadId === null) return null; + return ( + consider(declaredByThread.get(threadId)) ?? consider(requestsByThread.get(threadId)) ?? null + ); +} + +export function buildUsageRouteAttribution( + input: UsageRouteAttributionInput, +): UsageRouteAttribution { + const base = buildUsageAttribution({ + generatedAtMs: input.cutoffMs, + records: input.records, + bindings: input.bindings, + links: input.links, + sources: input.sources, + }); + + const baseByKey = new Map(); + for (const session of base.sessions) { + baseByKey.set(sessionKey(session.provider, session.sessionId), session); + } + // Thread(s) a session is bound to, for sessions that have no history row. + const threadsByKey = new Map(); + for (const binding of input.bindings) { + const key = sessionKey(binding.provider, binding.nativeSessionId); + const threads = threadsByKey.get(key) ?? []; + if (!threads.includes(binding.threadId)) threads.push(binding.threadId); + threadsByKey.set(key, threads); + } + + const declaredByThread = new Map(); + const requestsByThread = new Map(); + for (const event of input.routeEvents) { + const target = event.kind === null ? requestsByThread : declaredByThread; + const list = target.get(event.threadId) ?? []; + list.push(event); + target.set(event.threadId, list); + } + for (const list of [...declaredByThread.values(), ...requestsByThread.values()]) { + list.sort(compareEvents); + } + + const sessionReports: SessionRouteReport[] = []; + const consumedBaseKeys = new Set(); + + const reportForBase = ( + session: AttributionSessionReport, + threadId: string | null, + parentSessionId: string | null, + ): SessionRouteReport => { + const selected = selectRouteEvent( + session.sessionId, + threadId, + declaredByThread, + requestsByThread, + ); + const requested = selected?.requested ?? null; + const actualModel = session.models.length === 1 ? (session.models[0] ?? null) : null; + return { + provider: session.provider, + sessionId: session.sessionId, + threadId, + models: session.models, + usage: session.totals, + quality: { + measurement: session.measurementQuality, + identity: session.identityQuality, + prompt: session.promptQuality, + request: session.requestQuality, + recordIdentity: session.recordIdentity, + conflict: session.conflict, + }, + parentSessionId, + requested, + requestedQuality: requested === null ? "unknown" : "declared", + actualProvider: session.provider, + actualModel, + actualEffort: null, + actualEffortQuality: "unsupported", + routeEventKind: selected?.kind ?? null, + taskStratum: selected?.taskStratum ?? "unknown", + experimentId: selected?.experimentId ?? null, + managerId: selected?.managerId ?? null, + agentId: selected?.agentId ?? null, + escalationReason: selected?.reason ?? null, + }; + }; + + // 1. Every durable identity, including providers T3 cannot measure. An + // unmeasured identity reports a null total, never a zero. + for (const entry of input.history) { + const key = + entry.usageProvider === null ? null : sessionKey(entry.usageProvider, entry.nativeSessionId); + const baseSession = key === null ? undefined : baseByKey.get(key); + if (key !== null) consumedBaseKeys.add(key); + if (baseSession !== undefined) { + sessionReports.push(reportForBase(baseSession, entry.threadId, entry.parentNativeSessionId)); + continue; + } + const selected = selectRouteEvent( + entry.nativeSessionId, + entry.threadId, + declaredByThread, + requestsByThread, + ); + const requested = selected?.requested ?? null; + sessionReports.push({ + provider: entry.usageProvider, + sessionId: entry.nativeSessionId, + threadId: entry.threadId, + models: [], + usage: null, + quality: null, + parentSessionId: entry.parentNativeSessionId, + requested, + requestedQuality: requested === null ? "unknown" : "declared", + actualProvider: null, + actualModel: null, + actualEffort: null, + actualEffortQuality: "unsupported", + routeEventKind: selected?.kind ?? null, + taskStratum: selected?.taskStratum ?? "unknown", + experimentId: selected?.experimentId ?? null, + managerId: selected?.managerId ?? null, + agentId: selected?.agentId ?? null, + escalationReason: selected?.reason ?? null, + }); + } + + // 2. Sessions the base projection knows but history did not name (for + // example a caller that supplied only cursor bindings). + for (const session of base.sessions) { + const key = sessionKey(session.provider, session.sessionId); + if (consumedBaseKeys.has(key)) continue; + const threads = threadsByKey.get(key) ?? []; + const threadId = threads.length === 1 ? (threads[0] ?? null) : null; + sessionReports.push(reportForBase(session, threadId, null)); + } + + sessionReports.sort((left, right) => + sessionLabel(left.provider, left.sessionId).localeCompare( + sessionLabel(right.provider, right.sessionId), + ), + ); + + // 3. Per-thread rollup. Usage is summed once per session, so association with + // multiple PRs can never duplicate a session's tokens. + const prKeysByThread = new Map>(); + for (const pr of base.pullRequests) { + for (const threadId of pr.threadIds) { + const set = prKeysByThread.get(threadId) ?? new Set(); + set.add(pr.key); + prKeysByThread.set(threadId, set); + } + } + const threadIds = new Set(); + for (const session of sessionReports) + if (session.threadId !== null) threadIds.add(session.threadId); + for (const threadId of declaredByThread.keys()) threadIds.add(threadId); + for (const threadId of requestsByThread.keys()) threadIds.add(threadId); + + const threads: ThreadRouteReport[] = [...threadIds] + .toSorted((left, right) => left.localeCompare(right)) + .map((threadId): ThreadRouteReport => { + const sessions = sessionReports.filter((session) => session.threadId === threadId); + let usage = ZERO; + for (const session of sessions) { + if (session.usage !== null) usage = addTotalsOf(usage, session.usage); + } + const events = [ + ...(declaredByThread.get(threadId) ?? []), + ...(requestsByThread.get(threadId) ?? []), + ]; + return { + threadId, + sessionLabels: sessions.map((session) => sessionLabel(session.provider, session.sessionId)), + models: sortedUnique(sessions.flatMap((session) => session.models)), + usage, + pullRequestKeys: [...(prKeysByThread.get(threadId) ?? [])].toSorted((left, right) => + left.localeCompare(right), + ), + experimentIds: sortedUnique( + events.map((event) => event.experimentId).filter((id): id is string => id !== null), + ), + managerIds: sortedUnique( + events.map((event) => event.managerId).filter((id): id is string => id !== null), + ), + agentIds: sortedUnique( + events.map((event) => event.agentId).filter((id): id is string => id !== null), + ), + routeEventKinds: [ + ...new Set( + events.map((event) => event.kind).filter((k): k is RouteEventKind => k !== null), + ), + ].toSorted((left, right) => left.localeCompare(right)), + }; + }); + + const identity: RouteIdentityDiagnostics = { + managerIds: sortedUnique( + input.routeEvents.map((event) => event.managerId).filter((id): id is string => id !== null), + ), + agentIds: sortedUnique( + input.routeEvents.map((event) => event.agentId).filter((id): id is string => id !== null), + ), + sessionsWithRouteMetadata: sessionReports.filter( + (session) => session.routeEventKind !== null || hasRouteSelectionValue(session.requested), + ).length, + sessionsWithParent: sessionReports.filter((session) => session.parentSessionId !== null).length, + }; + + return { + contractVersion: USAGE_ROUTE_ATTRIBUTION_VERSION, + generatedAtMs: input.cutoffMs, + base, + sessions: sessionReports, + threads, + routeEvents: input.routeEvents.slice().sort(compareEvents), + identity, + limitations: limitationsFor(input, sessionReports), + }; +} + +function limitationsFor( + input: UsageRouteAttributionInput, + sessions: readonly SessionRouteReport[], +): readonly string[] { + const limitations: string[] = [ + "Observed reasoning effort is not exposed by any scanned source, so `actualEffort` is always null with quality `unsupported`; it is never copied from the requested effort.", + "A readable manager/agent id is a label for a route decision. It never replaces the native provider session id, and it is not used to join usage.", + "Requested values are pre-execution declarations. An observed value is only reported from a measured usage record.", + ]; + if (sessions.some((session) => session.provider === null)) { + limitations.push( + "Some durable identities belong to a provider T3 does not scan for usage (for example OpenCode child sessions). Their usage is unknown, not zero, and is preserved as its own identity rather than flattened into the parent.", + ); + } + if (sessions.some((session) => session.usage === null)) { + limitations.push( + "Some sessions have no measured usage. A null total is unknown, not a zero-cost success.", + ); + } + if (input.routeEvents.length === 0) { + limitations.push( + "No route metadata was supplied, so requested provider/model/effort and the experiment/cohort are unknown for every session.", + ); + } + return limitations; +} diff --git a/docs/internals/usage-attribution.md b/docs/internals/usage-attribution.md index e3475349ec6c..387585bf697d 100644 --- a/docs/internals/usage-attribution.md +++ b/docs/internals/usage-attribution.md @@ -11,9 +11,10 @@ response, or tool payload. [`usageAttributionSources.ts`](../../apps/server/src/usage/usageAttributionSources.ts) is the read-only extraction seam that proves the pure function can be fed from what the server actually writes. It reads the allowlisted fields of -`provider_session_runtime` (`resume_cursor_json`, `runtime_payload_json.importedTranscripts`) -and `projection_thread_pull_requests`, and returns a binding/link snapshot plus -diagnostics for what it could not read. It never returns a runtime payload. +`provider_session_runtime` (`resume_cursor_json`, `runtime_payload_json.importedTranscripts`), +`provider_session_history`, `thread_route_events`, and `projection_thread_pull_requests`, +and returns a binding/link snapshot plus diagnostics for what it could not read. It never +returns a runtime payload. ## Granularity is a source property @@ -144,15 +145,52 @@ presence without double counting because the re-parse replaces the entry. A read during that re-parse keeps the retained fallback rows, and a deleted file is never re-parsed at all, so its history survives. +## Durable session history + +The current resume cursor is single-valued: a resume, fork, or model switch overwrites +it, which previously left every earlier native session unattributable. Migration `054` +adds `provider_session_history`, an append-only identity record keyed on +`(thread_id, provider_name, native_session_id)`. `ProviderSessionRuntime.upsert` appends +one row per distinct native session and only advances `last_seen_at` on a repeat, so a +thread can answer "which sessions and models contributed" after the cursor moved on. The +migration backfills the current cursor of an upgraded database, and a conflicting +`onConflict: "ignore"` write appends nothing because that cursor was never applied. + +`parent_native_session_id` carries a true sub-agent parent when a caller knows it. T3 +does not yet persist OpenCode child session ids (they live only in the adapter's +in-memory `relatedSessionIds` and its native event log), so a child identity stays +distinguishable but its usage is `null` — unknown, never folded into the parent. + +## Requested versus observed, and experiment metadata + +`thread_route_events` (migration `055`) is an append-only, content-free record of a +routing decision. T3 carries it; agent-config remains the policy authority that decides +which route to use. Each row has a nullable `route_event_kind` — `normal`, +`availability_fallback`, `canary`, `independent_review`, `quality_escalation` — so an +automatic "this is what was requested" record (`kind = null`) is distinguishable from a +declared experiment/fallback/escalation event. It also carries the pre-execution task +stratum, experiment/cohort id, readable manager/agent ids, and the escalation reason. + +Requested provider/model/effort are captured at session start from the model selection +T3 was actually given. The observed model is read from measured usage records. The two are +never copied into each other: a requested value is not evidence of what ran, and an +unobserved value stays `null`. No scanned source exposes reasoning effort, so +`actualEffort` is always `null` with quality `unsupported`. A readable manager/agent id is +a label for a route decision, never a substitute for the native session id and never a +join key. + +[`usageRouteAttribution.ts`](../../apps/server/src/usage/usageRouteAttribution.ts) +composes the base projection and adds this view: per-session requested/observed/experiment +metadata, per-thread rollups, and identity diagnostics. Its usage rollup sums each session +once, so associating a thread with several PRs cannot duplicate tokens. No prompts, +responses, code, or tool bodies are stored anywhere in this path. + ## What still needs architecture approval The projection proves the join and the levels with fixtures. It does not choose a -storage or transport for the result and registers no endpoint. Durable per-thread -session history (an additive cursor/identity record rather than the single current -cursor) is the one schema change that would widen coverage, and it is deliberately not -adopted here. Provider-instance identity is likewise not recoverable from a transcript -scan; correlate that when the scan starts tagging files with the instance that produced -them. +storage or transport for the result and registers no endpoint. Provider-instance identity +is still not recoverable from a transcript scan; correlate that when the scan starts +tagging files with the instance that produced them. `UsageProviderKind` is `claude | codex | grok`. OpenCode, Antigravity, and Cursor have a native cursor id but no transcript T3 scans, so they have no usage source here.