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..ff644fab5034 --- /dev/null +++ b/apps/server/src/persistence/Migrations/054_ProviderSessionHistory.test.ts @@ -0,0 +1,207 @@ +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 providerInstanceKey: string; + readonly nativeSessionId: string; + readonly parentNativeSessionId: string | null; + readonly origin: string; + readonly firstSeenAt: string; + readonly lastSeenAt: string; +} + +layer("054_ProviderSessionHistory", (it) => { + it.effect("backfills the current cursor with runtime nativeSessionIdOf semantics", () => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + + yield* runMigrations({ toMigrationInclusive: 53 }); + + const insertRuntime = ( + threadId: string, + providerName: string, + cursor: string | null, + providerInstanceId: string | null = providerName, + ) => { + const instanceJson = providerInstanceId === null ? null : providerInstanceId; + return 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}, + ${instanceJson}, + ${providerName}, + 'full-access', + 'running', + '2026-09-23T09:00:00.000Z', + ${cursor}, + NULL + ) + `; + }; + + // Valid text resume. + yield* insertRuntime("thread-resume", "claudeAgent", '{"resume":"resume-id"}'); + // Empty/whitespace resume falls through to a valid threadId. + yield* insertRuntime( + "thread-empty-resume", + "claudeAgent", + '{"resume":"","threadId":"thread-id-fallback"}', + ); + // Non-string resume falls through to a valid threadId. + yield* insertRuntime( + "thread-nonstring-resume", + "codex", + '{"resume":123,"threadId":"thread-id-2"}', + ); + // Blank resume and threadId fall through to a valid sessionId. + yield* insertRuntime( + "thread-session-fallback", + "codex", + '{"resume":"","threadId":" ","sessionId":"session-id-3"}', + ); + // Whitespace-only text fields are ignored entirely. + yield* insertRuntime("thread-whitespace", "codex", '{"resume":" ","threadId":" "}'); + // Non-string candidates never become an id. + yield* insertRuntime("thread-number-only", "codex", '{"resume":123}'); + yield* insertRuntime("thread-bool-only", "codex", '{"threadId":true}'); + yield* insertRuntime("thread-object-only", "codex", '{"sessionId":{"nested":1}}'); + // No recognised id field. + yield* insertRuntime("thread-unknown", "codex", '{"opaque":true}'); + // No cursor at all. + yield* insertRuntime("thread-absent", "codex", null); + // A null provider instance uses the deterministic empty-string key. + yield* insertRuntime("thread-null-instance", "codex", '{"resume":"null-instance-id"}', null); + // A whitespace-padded instance id is trimmed in the key. + yield* insertRuntime( + "thread-trimmed-instance", + "codex", + '{"resume":"trimmed-id"}', + " codex-x ", + ); + + yield* runMigrations({ toMigrationInclusive: 55 }); + + const rows = yield* sql` + SELECT + thread_id AS "threadId", + provider_name AS "providerName", + provider_instance_key AS "providerInstanceKey", + 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-null-instance", + providerName: "codex", + providerInstanceKey: "", + nativeSessionId: "null-instance-id", + parentNativeSessionId: null, + origin: "runtimeCursor", + firstSeenAt: "2026-09-23T09:00:00.000Z", + lastSeenAt: "2026-09-23T09:00:00.000Z", + }, + { + threadId: "thread-resume", + providerName: "claudeAgent", + providerInstanceKey: "claudeAgent", + nativeSessionId: "resume-id", + parentNativeSessionId: null, + origin: "runtimeCursor", + firstSeenAt: "2026-09-23T09:00:00.000Z", + lastSeenAt: "2026-09-23T09:00:00.000Z", + }, + { + threadId: "thread-session-fallback", + providerName: "codex", + providerInstanceKey: "codex", + nativeSessionId: "session-id-3", + parentNativeSessionId: null, + origin: "runtimeCursor", + firstSeenAt: "2026-09-23T09:00:00.000Z", + lastSeenAt: "2026-09-23T09:00:00.000Z", + }, + { + threadId: "thread-nonstring-resume", + providerName: "codex", + providerInstanceKey: "codex", + nativeSessionId: "thread-id-2", + parentNativeSessionId: null, + origin: "runtimeCursor", + firstSeenAt: "2026-09-23T09:00:00.000Z", + lastSeenAt: "2026-09-23T09:00:00.000Z", + }, + { + threadId: "thread-empty-resume", + providerName: "claudeAgent", + providerInstanceKey: "claudeAgent", + nativeSessionId: "thread-id-fallback", + parentNativeSessionId: null, + origin: "runtimeCursor", + firstSeenAt: "2026-09-23T09:00:00.000Z", + lastSeenAt: "2026-09-23T09:00:00.000Z", + }, + { + threadId: "thread-trimmed-instance", + providerName: "codex", + providerInstanceKey: "codex-x", + nativeSessionId: "trimmed-id", + 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 (including the provider + // instance key), so a repeat cannot duplicate. + const indexes = yield* sql<{ readonly name: string }>` + PRAGMA index_list(provider_session_history) + `; + assert.ok(indexes.length > 0); + + // Migration rerun is idempotent: no new rows, no duplicate backfill. + yield* runMigrations({ toMigrationInclusive: 55 }); + const afterRerun = yield* sql<{ readonly count: number }>` + SELECT COUNT(*) AS count FROM provider_session_history + `; + assert.equal(afterRerun[0]!.count, rows.length); + }), + ); + + 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..90cc75244af4 --- /dev/null +++ b/apps/server/src/persistence/Migrations/054_ProviderSessionHistory.ts @@ -0,0 +1,118 @@ +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 + * durable identity so a thread can answer which native sessions it used even + * after the cursor moved on. Repeated observations of the same identity only + * advance `last_seen_at`; they never replace a different session's row. + * + * Identity includes the configured provider instance, not just the provider + * name. Two instances of one driver can expose the same native session id on + * one thread (for example native OpenCode Go and a CLIProxyAPI loopback), and + * collapsing them would erase which instance produced the session. SQLite + * treats `NULL` as distinct in `UNIQUE`, so the instance is stored as a + * normalized non-null `provider_instance_key`: a trimmed instance id, or `""` + * for an unknown/null instance. `""` is a deterministic bucket, so two + * unknown-instance observations collapse to one row while a real instance + * stays distinct. The runtime writer computes the same key in JS via + * `normalizeProviderInstanceKey`. + * + * The backfill seeds the table from whatever cursor already exists so an + * upgraded database does not start empty. Its id selection mirrors the runtime + * `nativeSessionIdOf` semantics exactly: only a JSON text value that is + * non-empty after trimming is accepted, and the precedence `resume` → + * `threadId` → `sessionId` falls through to the next candidate on any absent, + * null, non-string, or blank value. A numeric/boolean/object cursor never + * becomes an id. + */ +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, + provider_instance_key TEXT NOT NULL, + 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, provider_instance_key, 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, + provider_instance_key, + 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, + COALESCE(NULLIF(TRIM(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, + CASE + WHEN json_type(runtime.cursor, '$.resume') = 'text' + AND TRIM(json_extract(runtime.cursor, '$.resume')) <> '' + THEN TRIM(json_extract(runtime.cursor, '$.resume')) + WHEN json_type(runtime.cursor, '$.threadId') = 'text' + AND TRIM(json_extract(runtime.cursor, '$.threadId')) <> '' + THEN TRIM(json_extract(runtime.cursor, '$.threadId')) + WHEN json_type(runtime.cursor, '$.sessionId') = 'text' + AND TRIM(json_extract(runtime.cursor, '$.sessionId')) <> '' + THEN TRIM(json_extract(runtime.cursor, '$.sessionId')) + ELSE NULL + END 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 + `; +}); 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..baf727eca383 --- /dev/null +++ b/apps/server/src/persistence/Migrations/055_ThreadRouteEvents.ts @@ -0,0 +1,49 @@ +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. + * + * `escalation_reason` is a bounded code/slug, not free text; the writer rejects + * anything that is not a low-cardinality token. `selection_conflict` marks an + * automatic request row whose stored selection disagreed with a later + * observation, so a conflict is surfaced rather than silently overwritten. + */ +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, + selection_conflict INTEGER NOT NULL DEFAULT 0, + 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..1b0f0dfcc360 --- /dev/null +++ b/apps/server/src/persistence/ProviderSessionRuntime.history.test.ts @@ -0,0 +1,614 @@ +// @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 Exit from "effect/Exit"; +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; + providerName?: string; + providerInstanceId?: ProviderInstanceId | null; +}) { + return { + threadId: overrides.threadId, + providerName: overrides.providerName ?? "codex", + providerInstanceId: + overrides.providerInstanceId === undefined + ? ProviderInstanceId.make("codex") + : overrides.providerInstanceId, + 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 + `; +} + +interface HistoryIdentityRow { + readonly nativeSessionId: string; + readonly providerInstanceId: string | null; + readonly providerInstanceKey: string; + readonly lastSeenAt: string; +} + +function historyIdentityRows(sql: SqlClient.SqlClient, threadId: ThreadId) { + return sql` + SELECT + native_session_id AS "nativeSessionId", + provider_instance_id AS "providerInstanceId", + provider_instance_key AS "providerInstanceKey", + last_seen_at AS "lastSeenAt" + FROM provider_session_history + WHERE thread_id = ${threadId} + ORDER BY provider_instance_key ASC, native_session_id ASC + `; +} + +interface RequestEventRow { + readonly eventId: string; + readonly requestedProvider: string | null; + readonly requestedModel: string | null; + readonly requestedEffort: string | null; + readonly selectionConflict: number; +} + +function requestEventRows(sql: SqlClient.SqlClient, threadId: ThreadId) { + return sql` + SELECT + event_id AS "eventId", + requested_provider AS "requestedProvider", + requested_model AS "requestedModel", + requested_effort AS "requestedEffort", + selection_conflict AS "selectionConflict" + FROM thread_route_events + WHERE thread_id = ${threadId} AND route_event_kind IS NULL + ORDER BY event_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: "canary_measured", + 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 === `${threadId}::declared::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, "canary_measured"); + 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); + }), + ); + + it.effect("rolls back the whole upsert when the history append fails, then retries cleanly", () => + Effect.gen(function* () { + const repository = yield* ProviderSessionRuntime.ProviderSessionRuntimeRepository; + const sql = yield* SqlClient.SqlClient; + const threadId = ThreadId.make("thread-atomic-history"); + + yield* sql` + CREATE TRIGGER fail_history_write BEFORE INSERT ON provider_session_history + BEGIN SELECT RAISE(ABORT, 'injected history failure'); END + `; + + const failed = yield* Effect.exit( + repository.upsert( + runtimeRow({ + threadId, + lastSeenAt: "2026-09-23T10:00:00.000Z", + resumeCursor: { threadId: "atomic" }, + }), + ), + ); + assert.equal(Exit.isFailure(failed), true); + // The runtime change must NOT remain: one logical upsert, all-or-nothing. + assert.equal(Option.isNone(yield* repository.getByThreadId({ threadId })), true); + assert.equal((yield* historyRows(sql, threadId)).length, 0); + + yield* sql`DROP TRIGGER fail_history_write`; + // Retry after a rolled-back transaction is safe and idempotent. + yield* repository.upsert( + runtimeRow({ + threadId, + lastSeenAt: "2026-09-23T10:00:00.000Z", + resumeCursor: { threadId: "atomic" }, + }), + ); + const runtime = Option.getOrThrow(yield* repository.getByThreadId({ threadId })); + assert.deepEqual(runtime.resumeCursor, { threadId: "atomic" }); + const rows = yield* historyRows(sql, threadId); + assert.deepEqual( + rows.map((row) => row.nativeSessionId), + ["atomic"], + ); + }), + ); + + it.effect("rolls back the runtime and history when a route event write fails", () => + Effect.gen(function* () { + const repository = yield* ProviderSessionRuntime.ProviderSessionRuntimeRepository; + const sql = yield* SqlClient.SqlClient; + const threadId = ThreadId.make("thread-atomic-route"); + + yield* sql` + CREATE TRIGGER fail_route_event_write BEFORE INSERT ON thread_route_events + BEGIN SELECT RAISE(ABORT, 'injected route event failure'); END + `; + + const failed = yield* Effect.exit( + repository.upsert( + runtimeRow({ + threadId, + lastSeenAt: "2026-09-23T10:10:00.000Z", + resumeCursor: { sessionId: "route-atomic" }, + }), + { + attribution: { + requestedRoute: { provider: "codex", model: "gpt-6-luna", effort: "high" }, + }, + }, + ), + ); + assert.equal(Exit.isFailure(failed), true); + assert.equal(Option.isNone(yield* repository.getByThreadId({ threadId })), true); + assert.equal((yield* historyRows(sql, threadId)).length, 0); + assert.equal((yield* requestEventRows(sql, threadId)).length, 0); + + yield* sql`DROP TRIGGER fail_route_event_write`; + yield* repository.upsert( + runtimeRow({ + threadId, + lastSeenAt: "2026-09-23T10:10:00.000Z", + resumeCursor: { sessionId: "route-atomic" }, + }), + { + attribution: { + requestedRoute: { provider: "codex", model: "gpt-6-luna", effort: "high" }, + }, + }, + ); + const events = yield* requestEventRows(sql, threadId); + assert.equal(events.length, 1); + assert.equal(events[0]!.requestedModel, "gpt-6-luna"); + }), + ); + + it.effect("keeps distinct provider-instance identities for one native session", () => + Effect.gen(function* () { + const repository = yield* ProviderSessionRuntime.ProviderSessionRuntimeRepository; + const sql = yield* SqlClient.SqlClient; + const threadId = ThreadId.make("thread-instance-identity"); + + yield* repository.upsert( + runtimeRow({ + threadId, + lastSeenAt: "2026-09-23T11:00:00.000Z", + resumeCursor: { sessionId: "ses_shared" }, + providerInstanceId: ProviderInstanceId.make("opencode-go"), + }), + ); + yield* repository.upsert( + runtimeRow({ + threadId, + lastSeenAt: "2026-09-23T11:05:00.000Z", + resumeCursor: { sessionId: "ses_shared" }, + providerInstanceId: ProviderInstanceId.make("cliproxy-loopback"), + }), + ); + + const rows = yield* historyIdentityRows(sql, threadId); + assert.equal(rows.length, 2); + assert.deepEqual( + rows.map((row) => row.providerInstanceId), + ["cliproxy-loopback", "opencode-go"], + ); + // Both keep the same native session id: instance identity was the only + // distinguishing axis, and it was not erased. + assert.deepEqual( + rows.map((row) => row.nativeSessionId), + ["ses_shared", "ses_shared"], + ); + }), + ); + + it.effect("does not let a later provider instance overwrite an earlier instance's row", () => + Effect.gen(function* () { + const repository = yield* ProviderSessionRuntime.ProviderSessionRuntimeRepository; + const sql = yield* SqlClient.SqlClient; + const threadId = ThreadId.make("thread-instance-no-erase"); + + yield* repository.upsert( + runtimeRow({ + threadId, + lastSeenAt: "2026-09-23T11:10:00.000Z", + resumeCursor: { sessionId: "ses_noerase" }, + providerInstanceId: ProviderInstanceId.make("instance-a"), + }), + ); + yield* repository.upsert( + runtimeRow({ + threadId, + lastSeenAt: "2026-09-23T11:15:00.000Z", + resumeCursor: { sessionId: "ses_noerase" }, + providerInstanceId: ProviderInstanceId.make("instance-b"), + }), + ); + + const rows = yield* historyIdentityRows(sql, threadId); + assert.equal(rows.length, 2); + const a = rows.find((row) => row.providerInstanceKey === "instance-a"); + assert.equal(a?.providerInstanceId, "instance-a"); + // The single runtime cursor reflects the latest writer; history retains both. + const runtime = Option.getOrThrow(yield* repository.getByThreadId({ threadId })); + assert.equal(runtime.providerInstanceId, "instance-b"); + }), + ); + + it.effect("collapses a null/unknown provider instance deterministically", () => + Effect.gen(function* () { + const repository = yield* ProviderSessionRuntime.ProviderSessionRuntimeRepository; + const sql = yield* SqlClient.SqlClient; + const threadId = ThreadId.make("thread-instance-null"); + + yield* repository.upsert( + runtimeRow({ + threadId, + lastSeenAt: "2026-09-23T11:20:00.000Z", + resumeCursor: { sessionId: "ses_unknown" }, + providerInstanceId: null, + }), + ); + yield* repository.upsert( + runtimeRow({ + threadId, + lastSeenAt: "2026-09-23T11:25:00.000Z", + resumeCursor: { sessionId: "ses_unknown" }, + providerInstanceId: null, + }), + ); + + const rows = yield* historyIdentityRows(sql, threadId); + assert.equal(rows.length, 1); + assert.equal(rows[0]!.providerInstanceKey, ""); + assert.equal(rows[0]!.providerInstanceId, null); + }), + ); + + it.effect("monotonically enriches an automatic request record", () => + Effect.gen(function* () { + const repository = yield* ProviderSessionRuntime.ProviderSessionRuntimeRepository; + const sql = yield* SqlClient.SqlClient; + const threadId = ThreadId.make("thread-request-enrich"); + + const write = (model: string | null, effort: string | null, lastSeenAt: string) => + repository.upsert( + runtimeRow({ + threadId, + lastSeenAt, + resumeCursor: { threadId: "ses_request" }, + }), + { attribution: { requestedRoute: { provider: "codex", model, effort } } }, + ); + + // First observation has no model; a later one fills it (null -> non-null). + yield* write(null, null, "2026-09-23T12:00:00.000Z"); + yield* write("gpt-6-luna", "high", "2026-09-23T12:01:00.000Z"); + let rows = yield* requestEventRows(sql, threadId); + assert.equal(rows.length, 1); + assert.equal(rows[0]!.requestedModel, "gpt-6-luna"); + assert.equal(rows[0]!.requestedEffort, "high"); + assert.equal(rows[0]!.selectionConflict, 0); + + // Same value is idempotent. + yield* write("gpt-6-luna", "high", "2026-09-23T12:02:00.000Z"); + rows = yield* requestEventRows(sql, threadId); + assert.equal(rows[0]!.requestedModel, "gpt-6-luna"); + assert.equal(rows[0]!.selectionConflict, 0); + + // A model-less recovery/stop write must not erase the known value. + yield* write(null, null, "2026-09-23T12:03:00.000Z"); + rows = yield* requestEventRows(sql, threadId); + assert.equal(rows[0]!.requestedModel, "gpt-6-luna"); + assert.equal(rows[0]!.requestedEffort, "high"); + assert.equal(rows[0]!.selectionConflict, 0); + + // A conflicting non-null value is retained (first wins) and surfaced. + yield* write("gpt-7", "low", "2026-09-23T12:04:00.000Z"); + rows = yield* requestEventRows(sql, threadId); + assert.equal(rows[0]!.requestedModel, "gpt-6-luna"); + assert.equal(rows[0]!.requestedEffort, "high"); + assert.equal(rows[0]!.selectionConflict, 1); + }), + ); + + it.effect("scopes a globally reused declared event id per thread", () => + Effect.gen(function* () { + const repository = yield* ProviderSessionRuntime.ProviderSessionRuntimeRepository; + const sql = yield* SqlClient.SqlClient; + const threadA = ThreadId.make("thread-declared-a"); + const threadB = ThreadId.make("thread-declared-b"); + const declared = { + eventId: "shared-declared-id", + kind: "canary" as const, + taskStratum: "implementation" as const, + }; + + yield* repository.upsert( + runtimeRow({ + threadId: threadA, + lastSeenAt: "2026-09-23T12:10:00.000Z", + resumeCursor: { threadId: "ses-declared-a" }, + }), + { attribution: { routeEvent: declared } }, + ); + yield* repository.upsert( + runtimeRow({ + threadId: threadB, + lastSeenAt: "2026-09-23T12:10:00.000Z", + resumeCursor: { threadId: "ses-declared-b" }, + }), + { attribution: { routeEvent: declared } }, + ); + + const rows = yield* sql<{ readonly eventId: string; readonly threadId: string }>` + SELECT event_id AS "eventId", thread_id AS "threadId" + FROM thread_route_events + WHERE route_event_kind = 'canary' + AND thread_id IN (${threadA}, ${threadB}) + ORDER BY thread_id ASC + `; + // A shared declared id must not suppress the second thread's event. + assert.equal(rows.length, 2); + assert.deepEqual( + rows.map((row) => row.threadId), + ["thread-declared-a", "thread-declared-b"], + ); + assert.equal(new Set(rows.map((row) => row.eventId)).size, 2); + }), + ); + + it.effect("is idempotent under reordered repeated upserts", () => + Effect.gen(function* () { + const repository = yield* ProviderSessionRuntime.ProviderSessionRuntimeRepository; + const sql = yield* SqlClient.SqlClient; + const threadId = ThreadId.make("thread-reorder"); + + yield* repository.upsert( + runtimeRow({ + threadId, + lastSeenAt: "2026-09-23T13:00:00.000Z", + resumeCursor: { threadId: "order-b" }, + }), + ); + yield* repository.upsert( + runtimeRow({ + threadId, + lastSeenAt: "2026-09-23T13:05:00.000Z", + resumeCursor: { threadId: "order-a" }, + }), + ); + yield* repository.upsert( + runtimeRow({ + threadId, + lastSeenAt: "2026-09-23T13:10:00.000Z", + resumeCursor: { threadId: "order-b" }, + }), + ); + + const rows = yield* historyRows(sql, threadId); + assert.deepEqual(rows.map((row) => row.nativeSessionId).toSorted(), ["order-a", "order-b"]); + assert.equal(rows.length, 2); + }), + ); +}); diff --git a/apps/server/src/persistence/ProviderSessionRuntime.ts b/apps/server/src/persistence/ProviderSessionRuntime.ts index 2673512edf10..19b15592988a 100644 --- a/apps/server/src/persistence/ProviderSessionRuntime.ts +++ b/apps/server/src/persistence/ProviderSessionRuntime.ts @@ -18,6 +18,14 @@ import { ThreadId, } from "@t3tools/contracts"; +import { + normalizeEscalationReason, + normalizeProviderInstanceKey, + normalizeTaskStratum, + type RouteEventInput, + type RouteSelectionMetadata, +} from "../usage/routeMetadata.ts"; + import { PersistenceDecodeError, type PersistenceErrorCorrelation, @@ -67,6 +75,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 +291,316 @@ 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), + providerInstanceKey: 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, + provider_instance_key, + adapter_key, + native_session_id, + parent_native_session_id, + origin, + first_seen_at, + last_seen_at + ) + VALUES ( + ${entry.threadId}, + ${entry.providerName}, + ${entry.providerInstanceId}, + ${entry.providerInstanceKey}, + ${entry.adapterKey}, + ${entry.nativeSessionId}, + ${entry.parentNativeSessionId}, + 'runtimeCursor', + ${entry.seenAt}, + ${entry.seenAt} + ) + ON CONFLICT (thread_id, provider_name, provider_instance_key, 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 RouteEventRowRequest = 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, + }); + + /** + * Automatic request record. Unlike a declared event, it is not authoritative + * and is written on every status transition (start, recovery, stop, + * rollback), so the first write can carry less than a later one. It uses an + * explicit enrichment policy instead of `INSERT OR IGNORE`: + * + * - null → non-null: enrich (a later observation fills an unknown field); + * - equal → equal: idempotent, no change; + * - non-null → null: retain the known value (never erased by a model-less + * recovery/stop/rollback write); + * - conflicting non-null → different non-null: retain the first value and + * raise `selection_conflict` so the disagreement is surfaced, never + * silently last-write-wins. + */ + const enrichRequestRouteEventRow = SqlSchema.void({ + Request: RouteEventRowRequest, + execute: (event) => + sql` + INSERT 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, + selection_conflict, + 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}, + 0, + ${event.recordedAt} + ) + ON CONFLICT (event_id) + DO UPDATE SET + requested_provider = CASE + WHEN thread_route_events.requested_provider IS NULL THEN excluded.requested_provider + WHEN excluded.requested_provider IS NULL THEN thread_route_events.requested_provider + ELSE thread_route_events.requested_provider + END, + requested_model = CASE + WHEN thread_route_events.requested_model IS NULL THEN excluded.requested_model + WHEN excluded.requested_model IS NULL THEN thread_route_events.requested_model + ELSE thread_route_events.requested_model + END, + requested_effort = CASE + WHEN thread_route_events.requested_effort IS NULL THEN excluded.requested_effort + WHEN excluded.requested_effort IS NULL THEN thread_route_events.requested_effort + ELSE thread_route_events.requested_effort + END, + selection_conflict = CASE + WHEN ( + (thread_route_events.requested_provider IS NOT NULL + AND excluded.requested_provider IS NOT NULL + AND thread_route_events.requested_provider <> excluded.requested_provider) + OR (thread_route_events.requested_model IS NOT NULL + AND excluded.requested_model IS NOT NULL + AND thread_route_events.requested_model <> excluded.requested_model) + OR (thread_route_events.requested_effort IS NOT NULL + AND excluded.requested_effort IS NOT NULL + AND thread_route_events.requested_effort <> excluded.requested_effort) + ) THEN 1 + ELSE thread_route_events.selection_conflict + END + `, + }); + + /** + * Declared experiment/fallback/escalation event. Authority-supplied and + * deterministic, so a repeat is ignored rather than overwritten. The caller + * scopes the persisted id by thread, so a globally reused declared id cannot + * silently suppress another thread's event. + */ + const insertDeclaredRouteEventRow = SqlSchema.void({ + Request: RouteEventRowRequest, + 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, + selection_conflict, + 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}, + 0, + ${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 mode: "request" | "declared"; + 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({ + mode: "request" as const, + 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; + const declaredId = declared.eventId?.trim(); + events.push({ + mode: "declared" as const, + // Scope the persisted primary key by thread: a declared id reused by + // another thread must not collide and suppress that event. + eventId: + declaredId !== undefined && declaredId.length > 0 + ? `${runtime.threadId}::declared::${declaredId}` + : `${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: normalizeEscalationReason(declared.reason), + recordedAt: runtime.lastSeenAt, + }); + } + return events; + }; + const recordImportedTranscriptRow = SqlSchema.void({ Request: RecordImportedTranscriptRequestSchema, execute: ({ threadId, source }) => @@ -364,8 +697,45 @@ export const make = Effect.gen(function* () { `, }); - const upsert: ProviderSessionRuntimeRepository["Service"]["upsert"] = (runtime, options) => - (options?.onConflict === "ignore" ? insertRuntimeRow(runtime) : upsertRuntimeRow(runtime)).pipe( + const upsert: ProviderSessionRuntimeRepository["Service"]["upsert"] = (runtime, options) => { + const writeEffects = 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. This is a single statement, so it is + // already atomic. + yield* insertRuntimeRow(runtime); + return; + } + // One logical upsert is one atomic durable effect: the runtime cursor, the + // history identity it implies, and the route events. A failure in any of + // them rolls the whole thing back, so a caller that sees an error never + // leaves a runtime cursor without its history (or vice versa). + yield* sql.withTransaction( + Effect.gen(function* () { + 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, + providerInstanceKey: normalizeProviderInstanceKey(runtime.providerInstanceId), + adapterKey: runtime.adapterKey, + nativeSessionId, + parentNativeSessionId: attribution?.parentNativeSessionId ?? null, + seenAt: runtime.lastSeenAt, + }); + } + for (const event of routeEventsFor(runtime, attribution, nativeSessionId)) { + yield* event.mode === "request" + ? enrichRequestRouteEventRow(event) + : insertDeclaredRouteEventRow(event); + } + }), + ); + }); + return writeEffects.pipe( Effect.mapError( toPersistenceSqlOrDecodeError( "ProviderSessionRuntimeRepository.upsert:query", @@ -374,6 +744,7 @@ export const make = Effect.gen(function* () { ), ), ); + }; const recordImportedTranscript: ProviderSessionRuntimeRepository["Service"]["recordImportedTranscript"] = (input) => 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..410742ef8739 --- /dev/null +++ b/apps/server/src/usage/routeMetadata.ts @@ -0,0 +1,218 @@ +/** + * 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; +} + +/** + * Normalized provider-instance identity key for durable session-history + * identity. + * + * SQLite `UNIQUE` treats `NULL` as distinct, so a nullable instance column + * cannot express "these are the same instance" — every null would be its own + * row. The key is therefore a non-null string: a trimmed instance id, or `""` + * for an unknown/null instance. `""` is a first-class bucket, not a wildcard, + * so two unknown-instance observations of one `(thread, provider, native + * session)` collapse to one row while a real instance stays distinct. + * + * The migration backfill computes the same value in SQL as + * `COALESCE(NULLIF(TRIM(provider_instance_id), ''), '')`. + */ +export function normalizeProviderInstanceKey(value: unknown): string { + const trimmed = readOptionalString(value); + return trimmed ?? ""; +} + +/** + * `escalation_reason` is bounded metadata, not a content channel. Only a + * low-cardinality code/slug is accepted: lowercase letters, digits, and + * `_`, `-`, `.`, `:` — no whitespace, no multiline payload. Anything else + * normalizes to `null` rather than being stored, so arbitrary prose cannot + * leak into the metadata carrier. + */ +export const MAX_ESCALATION_REASON_LENGTH = 64; +const ESCALATION_REASON_PATTERN = /^[a-z0-9][a-z0-9_.:-]*$/; + +export function normalizeEscalationReason(value: unknown): string | null { + const trimmed = readOptionalString(value); + if (trimmed === null) return null; + if (trimmed.length > MAX_ESCALATION_REASON_LENGTH) return null; + return ESCALATION_REASON_PATTERN.test(trimmed) ? trimmed : null; +} + +/** + * 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; + /** + * `true` when a later automatic request observation disagreed with an + * already-stored non-null requested field. The stored value is retained + * (first non-null wins) and the conflict is surfaced rather than silently + * overwritten. + */ + readonly selectionConflict: boolean; + 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; + /** 0/1; `1` when a later request observation conflicted with the stored selection. */ + readonly selectionConflict: number; + 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..bb20a1041f96 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,244 @@ 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); + }); + + it("de-duplicates repeated rows by canonical identity", () => { + const row = historyRow(); + const { history, bindings, diagnostics } = extractAttributionHistory([row, { ...row }]); + + expect(history).toHaveLength(1); + expect(bindings).toHaveLength(1); + expect(diagnostics.duplicatesDropped).toBe(1); + expect(diagnostics.conflictingDuplicates).toBe(0); + }); + + it("keeps the same native session under two provider instances distinct", () => { + const { history, diagnostics } = extractAttributionHistory([ + historyRow({ + providerName: "opencode", + adapterKey: "opencode", + providerInstanceId: "opencode-go", + }), + historyRow({ + providerName: "opencode", + adapterKey: "opencode", + providerInstanceId: "cliproxy-loopback", + }), + ]); + + expect(history).toHaveLength(2); + expect(history.map((entry) => entry.providerInstanceId)).toEqual([ + "opencode-go", + "cliproxy-loopback", + ]); + expect(diagnostics.duplicatesDropped).toBe(0); + }); + + it("collapses a null provider instance deterministically", () => { + const { history, diagnostics } = extractAttributionHistory([ + historyRow({ providerInstanceId: null }), + historyRow({ providerInstanceId: null }), + ]); + + expect(history).toHaveLength(1); + expect(diagnostics.duplicatesDropped).toBe(1); + }); + + it("merges duplicate rows monotonically and surfaces a conflicting field", () => { + const { history, diagnostics } = extractAttributionHistory([ + historyRow({ lastSeenAt: "2026-09-23T09:10:00.000Z" }), + historyRow({ + adapterKey: "codex-alt", + lastSeenAt: "2026-09-23T09:20:00.000Z", + parentNativeSessionId: "parent-x", + }), + ]); + + expect(history).toHaveLength(1); + expect(history[0]!.lastSeenAt).toBe("2026-09-23T09:20:00.000Z"); + expect(history[0]!.parentNativeSessionId).toBe("parent-x"); + expect(diagnostics.conflictingDuplicates).toBe(1); + }); +}); + +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, + selectionConflict: 0, + 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, + selectionConflict: false, + 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); + }); + + it("rejects an unbounded escalation reason and keeps a bounded code", () => { + const { events } = extractAttributionRouteEvents([ + routeEventRow({ escalationReason: "this is human prose with spaces and CAPS!" }), + routeEventRow({ eventId: "evt-2", escalationReason: "reviewer_blocker" }), + routeEventRow({ eventId: "evt-3", escalationReason: "x".repeat(200) }), + ]); + + expect(events.map((event) => event.reason)).toEqual([null, "reviewer_blocker", null]); + }); + + it("surfaces a persisted selection conflict on a request record", () => { + const { events, diagnostics } = extractAttributionRouteEvents([ + routeEventRow({ routeEventKind: null, selectionConflict: 1 }), + ]); + + expect(events[0]!.selectionConflict).toBe(true); + expect(diagnostics.conflictingSelections).toBe(1); + }); + + it("normalizes a malformed requested route to null", () => { + const { events } = extractAttributionRouteEvents([ + routeEventRow({ + requestedProvider: 123 as unknown as string, + requestedModel: "", + requestedEffort: " ", + }), + ]); + + expect(events[0]!.requested).toBeNull(); + }); +}); + describe("extractAttributionLinks", () => { it("reads allowlisted fields from a persisted projection row", () => { const { links, diagnostics } = extractAttributionLinks([linkRow()]); @@ -303,6 +547,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 +685,8 @@ describe("extractAttributionSnapshot", () => { url: "https://github.com/acme/repo/pull/12", }, ], + history: [], + routeEvents: [], diagnostics: { bindings: { runtimeRows: 3, @@ -449,8 +700,72 @@ describe("extractAttributionSnapshot", () => { overwrittenThreads: 1, ambiguousSessionIds: 0, }, + history: { + rows: 0, + sessions: 0, + bindings: 0, + withParent: 0, + duplicatesDropped: 0, + conflictingDuplicates: 0, + unsupportedProviderBindings: 0, + malformed: 0, + }, + routeEvents: { + rows: 0, + events: 0, + declared: 0, + requests: 0, + conflictingSelections: 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, + selectionConflict: 0, + 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..4aa4d0044d6f 100644 --- a/apps/server/src/usage/usageAttributionSources.ts +++ b/apps/server/src/usage/usageAttributionSources.ts @@ -21,6 +21,17 @@ import type { ThreadPullRequestLinkSource, UsageProviderKind } from "@t3tools/contracts"; import type { AttributionPullRequestLink, AttributionThreadBinding } from "./usageAttribution.ts"; +import { + isRouteEventKind, + normalizeEscalationReason, + normalizeProviderInstanceKey, + 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 +122,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 +355,231 @@ 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; + /** Rows whose canonical identity repeated an earlier row and were merged. */ + readonly duplicatesDropped: number; + /** Merged duplicate rows that disagreed on a non-null field (kept-first). */ + readonly conflictingDuplicates: 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, + duplicatesDropped: 0, + conflictingDuplicates: 0, + unsupportedProviderBindings: 0, + malformed: 0, + }; + // Canonical identity is exactly what the table's unique key is: thread, + // provider, normalized provider instance, native session. A join that + // repeats a row must not become a second report, and two instances that + // share a native session id must stay distinct. + const byIdentity = new Map(); + + 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 identity = `${threadId}\u0000${providerName}\u0000${normalizeProviderInstanceKey( + row.providerInstanceId, + )}\u0000${nativeSessionId}`; + const existingIndex = byIdentity.get(identity); + if (existingIndex !== undefined) { + const existing = history[existingIndex]!; + diagnostics.duplicatesDropped += 1; + // Merge only monotonically: never let a duplicate erase a known value. + if (adapterKey !== existing.adapterKey) diagnostics.conflictingDuplicates += 1; + const parentNativeSessionId = readOptionalString(row.parentNativeSessionId); + if (existing.parentNativeSessionId === null && parentNativeSessionId !== null) { + diagnostics.withParent += 1; + history[existingIndex] = { ...existing, parentNativeSessionId }; + } + const firstSeenAt = readOptionalString(row.firstSeenAt) ?? row.lastSeenAt; + history[existingIndex] = { + ...history[existingIndex]!, + firstSeenAt: + firstSeenAt < history[existingIndex]!.firstSeenAt + ? firstSeenAt + : history[existingIndex]!.firstSeenAt, + lastSeenAt: + row.lastSeenAt > history[existingIndex]!.lastSeenAt + ? row.lastSeenAt + : history[existingIndex]!.lastSeenAt, + }; + continue; + } + + const usageProvider = usageProviderOf(providerName, adapterKey); + const parentNativeSessionId = readOptionalString(row.parentNativeSessionId); + diagnostics.sessions += 1; + if (parentNativeSessionId !== null) diagnostics.withParent += 1; + byIdentity.set(identity, history.length); + 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; + /** Request records whose stored selection disagreed with a later observation. */ + readonly conflictingSelections: 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, + conflictingSelections: 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; + const selectionConflict = row.selectionConflict === 1; + diagnostics.events += 1; + if (kind === null) diagnostics.requests += 1; + else diagnostics.declared += 1; + if (selectionConflict) diagnostics.conflictingSelections += 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: normalizeEscalationReason(row.escalationReason), + selectionConflict, + recordedAt, + }); + } + + return { events, diagnostics }; +} + /** Reads thread → PR links from persisted projection rows, dropping payloads. */ export function extractAttributionLinks( rows: readonly PersistedThreadPullRequestRow[], @@ -379,15 +624,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..c2754e5c2244 --- /dev/null +++ b/apps/server/src/usage/usageRouteAttribution.test.ts @@ -0,0 +1,812 @@ +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 { decodeScanCache } from "./usageScanCache.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, + selectionConflict: false, + 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(); + }); +}); + +describe("durable history preserves #4 zero and invalid quality", () => { + const CLAUDE_SESSION = "5a128faa-8253-489e-b935-6c08e8e670c0"; + const ZERO_TOTALS: UsageTokenTotals = { + uncachedInputTokens: 0, + cachedInputTokens: 0, + cacheCreationTokens: 0, + outputTokens: 0, + reasoningTokens: 0, + }; + + function claudeBinding(): AttributionThreadBinding { + return binding({ provider: "claude", nativeSessionId: CLAUDE_SESSION }); + } + + function claudeHistory(): ExtractedSessionHistory { + return history({ + providerName: "claudeAgent", + adapterKey: "claudeAgent", + nativeSessionId: CLAUDE_SESSION, + usageProvider: "claude", + }); + } + + // Regressions A-D from the R3-T3 restack brief. They prove the #6 route view + // is a faithful mirror of the #4 (M3D) measurement/identity quality axes and + // never promotes an unknown or invalid measurement to an exact one. + + it("keeps an explicit measured zero as a measured zero, not missing (A)", () => { + const report = build({ + records: [ + record({ + totals: ZERO_TOTALS, + costUsd: 0, + measurement: "observed", + measurementCompleteness: "complete", + }), + ], + bindings: [binding()], + history: [history()], + }); + + const session = sessionOf(report, DEEPSEEK_SESSION); + // A measured zero is present and complete; it is not the absence of a + // measurement, and it is not dropped. + expect(session.usage).not.toBeNull(); + expect(session.usage!.totalTokens).toBe(0); + expect(session.quality?.measurement).toBe("measured"); + expect(session.quality?.measurement).not.toBe("missing"); + }); + + it("keeps an all-invalid eligible observation invalid, not missing or zero (B)", () => { + const report = build({ + records: [ + record({ + totals: ZERO_TOTALS, + costUsd: 0, + measurement: "invalid", + }), + ], + bindings: [binding()], + history: [history()], + }); + + const session = sessionOf(report, DEEPSEEK_SESSION); + expect(session.quality?.measurement).toBe("invalid"); + expect(session.quality?.measurement).not.toBe("missing"); + expect(session.quality?.measurement).not.toBe("measured"); + // The route view mirrors the base projection rather than recomputing it. + const baseSession = report.base.sessions.find((entry) => entry.sessionId === DEEPSEEK_SESSION)!; + expect(baseSession.measurementQuality).toBe("invalid"); + }); + + it("does not promote predecessor-quality partial cache evidence to complete (C)", () => { + // A predecessor v4 shape: 15-field rows written before completeness metadata + // existed. The M3D decoder reads the missing metadata as `partial`, never + // `complete`; the route view must not undo that. + const predecessorDocument = { + version: 4, + models: ["claude-fable-5"], + sessions: [CLAUDE_SESSION], + files: { + "/pred/live.jsonl": { + s: 120, + m: 500, + p: "claude", + r: [ + [ + 1_785_578_400_000, + 0, + 0, + 10, + 0, + 0, + 0, + 0, + "msg_partial:req_msg_partial", + null, + "req_msg_partial", + "msg_partial", + null, + 0, + 0, + ], + [ + 1_785_578_400_000, + 0, + 0, + 4, + 0, + 0, + 6, + 0, + "msg_valid:req_msg_valid", + null, + "req_msg_valid", + "msg_valid", + null, + 0, + 0, + ], + ], + t: [], + o: 120, + gl: 64, + gh: 12_345, + cs: null, + }, + }, + }; + + const cache = decodeScanCache(JSON.parse(JSON.stringify(predecessorDocument))); + const entry = cache.get("/pred/live.jsonl")!; + expect(entry.qualityMetadata).toBe("predecessor"); + expect(entry.records.map((row) => row.measurementCompleteness)).toEqual(["partial", "partial"]); + + const records = entry.records.map((row): AttributionUsageRecord => ({ + provider: row.provider, + sessionId: row.sessionId, + model: row.model, + timestampMs: row.timestampMs, + totals: row.totals, + costUsd: row.reportedCostUsd ?? 0, + dedupeKey: row.dedupeKey, + sourceFingerprint: "/pred/live.jsonl", + ...(row.measurement === undefined ? {} : { measurement: row.measurement }), + ...(row.measurementCompleteness === undefined + ? {} + : { measurementCompleteness: row.measurementCompleteness }), + ...(row.invalidTokenFields === undefined + ? {} + : { invalidTokenFields: row.invalidTokenFields }), + ...(row.dedupeKeyScope === undefined ? {} : { dedupeKeyScope: row.dedupeKeyScope }), + })); + + const report = build({ + records, + bindings: [claudeBinding()], + history: [claudeHistory()], + }); + + const session = sessionOf(report, CLAUDE_SESSION); + // Totals are retained, but the measurement is a lower bound, never complete. + expect(session.usage).not.toBeNull(); + expect(session.quality?.measurement).toBe("partial"); + expect(session.quality?.measurement).not.toBe("measured"); + expect( + report.base.sessions.find((entry) => entry.sessionId === CLAUDE_SESSION)?.measurementQuality, + ).toBe("partial"); + }); + + it("keeps a history-only measured-provider identity at usage null / quality null (D)", () => { + const missingSession = "019f0000-0000-7000-8000-0000000000bb"; + const report = build({ + records: [], + bindings: [], + history: [history({ nativeSessionId: missingSession })], + }); + + const session = sessionOf(report, missingSession); + // The provider is known from durable history, but no usage was measured: + // unknown, never a measured zero and never a fabricated quality. + expect(session.provider).toBe("codex"); + expect(session.usage).toBeNull(); + expect(session.quality).toBeNull(); + }); +}); + +describe("route-view allocation and provider-instance identity", () => { + const THREAD_1 = "thread-amb-1"; + const THREAD_2 = "thread-amb-2"; + + it("does not double count one measured session bound to two threads", () => { + const report = build({ + records: [record({ sessionId: DEEPSEEK_SESSION })], + bindings: [ + binding({ threadId: THREAD_1 }), + binding({ threadId: THREAD_2, providerInstanceId: "codex-secondary" }), + ], + history: [history({ threadId: THREAD_1 }), history({ threadId: THREAD_2 })], + }); + + const session = sessionOf(report, DEEPSEEK_SESSION); + expect(session.allocation).toBe("ambiguous"); + expect(session.threadId).toBeNull(); + expect(session.boundThreadIds).toEqual([THREAD_1, THREAD_2]); + expect(report.identity.ambiguousSessions).toBe(1); + + // Neither candidate thread receives the ambiguous session's usage, but both + // remain visible as (empty) threads because the bindings name them. + for (const threadId of [THREAD_1, THREAD_2]) { + const thread = report.threads.find((entry) => entry.threadId === threadId); + expect(thread).toBeDefined(); + expect(thread!.usage.totalTokens).toBe(0); + expect(thread!.sessionLabels).toEqual([]); + } + + // Global reconciliation is additive exactly once: per-thread totals plus + // the unallocated bucket equal the single measured session usage. + const sessionTotal = session.usage!.totalTokens; + const threadSum = report.threads.reduce((sum, thread) => sum + thread.usage.totalTokens, 0); + expect(threadSum + report.unallocated.totalTokens).toBe(sessionTotal); + expect(report.unallocated.totalTokens).toBe(sessionTotal); + expect(report.limitations.some((line) => line.includes("unallocated"))).toBe(true); + }); + + it("counts a uniquely bound session once in per-thread totals", () => { + const report = build({ + records: [record({ sessionId: DEEPSEEK_SESSION })], + bindings: [binding()], + history: [history()], + }); + + const session = sessionOf(report, DEEPSEEK_SESSION); + expect(session.allocation).toBe("unallocated"); + const thread = report.threads.find((entry) => entry.threadId === THREAD)!; + expect(thread.usage.totalTokens).toBe(session.usage!.totalTokens); + expect(report.unallocated.totalTokens).toBe(0); + }); + + it("exposes all provider instances that produced one session", () => { + const report = build({ + records: [record({ sessionId: DEEPSEEK_SESSION })], + bindings: [ + binding({ providerInstanceId: "codex-default" }), + binding({ providerInstanceId: "codex-secondary" }), + ], + history: [ + history({ providerInstanceId: "codex-default" }), + history({ providerInstanceId: "codex-secondary" }), + ], + }); + + const session = sessionOf(report, DEEPSEEK_SESSION); + expect(session.providerInstanceIds).toEqual(["codex-default", "codex-secondary"]); + }); + + it("keeps two provider instances of one unmeasured session distinct", () => { + const report = build({ + history: [ + history({ + usageProvider: null, + providerName: "opencode", + adapterKey: "opencode", + providerInstanceId: "opencode-go", + }), + history({ + usageProvider: null, + providerName: "opencode", + adapterKey: "opencode", + providerInstanceId: "cliproxy-loopback", + }), + ], + }); + + const sessions = report.sessions.filter((entry) => entry.sessionId === DEEPSEEK_SESSION); + expect(sessions).toHaveLength(2); + expect( + sessions + .flatMap((entry) => entry.providerInstanceIds) + .toSorted((left, right) => left.localeCompare(right)), + ).toEqual(["cliproxy-loopback", "opencode-go"]); + expect(sessions.every((entry) => entry.usage === null && entry.provider === null)).toBe(true); + }); + + it("surfaces a selection conflict carried by the selected route event", () => { + const report = build({ + records: [record({ sessionId: DEEPSEEK_SESSION })], + bindings: [binding()], + history: [history()], + routeEvents: [routeEvent({ kind: null, selectionConflict: true, eventId: "req-conflict" })], + }); + + const session = sessionOf(report, DEEPSEEK_SESSION); + expect(session.routeSelectionConflict).toBe(true); + expect(report.identity.conflictingSelections).toBe(1); + }); +}); diff --git a/apps/server/src/usage/usageRouteAttribution.ts b/apps/server/src/usage/usageRouteAttribution.ts new file mode 100644 index 000000000000..a435de73a273 --- /dev/null +++ b/apps/server/src/usage/usageRouteAttribution.ts @@ -0,0 +1,574 @@ +/** + * 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 AttributionAllocation, + 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, + normalizeProviderInstanceKey, + 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; + /** + * Base allocation evidence, carried so the route view cannot drop it. In + * particular `ambiguous` marks a session bound to more than one thread; such + * a session's usage is never duplicated onto each candidate thread. + */ + readonly allocation: AttributionAllocation; + /** Threads this identity is bound to, from the base binding evidence. */ + readonly boundThreadIds: readonly string[]; + /** + * Configured provider-instance identities that produced this session. + * Provider-instance identity is never erased from the report. + */ + readonly providerInstanceIds: readonly string[]; + 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; + /** `true` when the selected request record carried a selection conflict. */ + readonly routeSelectionConflict: boolean; +} + +export interface ThreadRouteReport { + readonly threadId: string; + readonly sessionLabels: readonly string[]; + readonly models: readonly string[]; + /** + * Additive over sessions uniquely bound to this thread. A session bound to + * more than one thread is `ambiguous` and is not summed here; its usage is + * pooled once in `UsageRouteAttribution.unallocated`. + */ + 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; + /** Sessions bound to more than one thread; their usage is not per-thread. */ + readonly ambiguousSessions: number; + /** Request records whose stored selection disagreed with a later observation. */ + readonly conflictingSelections: 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[]; + /** + * Usage held in no single thread: sessions that are `ambiguous` (bound to + * more than one thread) or `unallocated` (no thread). It is additive once + * with the per-thread totals, never duplicated into each candidate thread. + */ + readonly unallocated: AttributionTotals; + 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); + } + + 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 unionSorted = (left: readonly string[], right: Iterable): string[] => + sortedUnique([...left, ...right]); + + const routeEventFields = (selected: RouteEventMetadata | null) => + ({ + requested: selected?.requested ?? null, + requestedQuality: (selected?.requested ?? null) === null ? "unknown" : "declared", + routeEventKind: selected?.kind ?? null, + taskStratum: selected?.taskStratum ?? "unknown", + experimentId: selected?.experimentId ?? null, + managerId: selected?.managerId ?? null, + agentId: selected?.agentId ?? null, + escalationReason: selected?.reason ?? null, + routeSelectionConflict: selected?.selectionConflict ?? false, + }) as const; + + const reportForBase = ( + session: AttributionSessionReport, + threadId: string | null, + parentSessionId: string | null, + boundThreadIds: readonly string[], + providerInstanceIds: readonly string[], + ): SessionRouteReport => { + const selected = selectRouteEvent( + session.sessionId, + threadId, + declaredByThread, + requestsByThread, + ); + 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, + }, + allocation: session.allocation, + boundThreadIds, + providerInstanceIds, + parentSessionId, + ...routeEventFields(selected), + actualProvider: session.provider, + actualModel, + actualEffort: null, + actualEffortQuality: "unsupported", + }; + }; + + /** + * A durable identity for a provider T3 cannot measure (OpenCode, a + * CLIProxyAPI loopback, ...). Its usage is `null` — unknown, never zero — and + * its provider-instance identity is preserved rather than erased. + */ + const reportForUnmeasured = ( + entry: ExtractedSessionHistory, + threadId: string | null, + boundThreadIds: readonly string[], + providerInstanceIds: readonly string[], + ): SessionRouteReport => { + const selected = selectRouteEvent( + entry.nativeSessionId, + threadId, + declaredByThread, + requestsByThread, + ); + return { + provider: entry.usageProvider, + sessionId: entry.nativeSessionId, + threadId, + models: [], + usage: null, + quality: null, + allocation: "unallocated", + boundThreadIds, + providerInstanceIds, + parentSessionId: entry.parentNativeSessionId, + ...routeEventFields(selected), + actualProvider: null, + actualModel: null, + actualEffort: null, + actualEffortQuality: "unsupported", + }; + }; + + // Group measured history by base session identity (provider + native id) so a + // session observed on two threads yields one report. Provider-instance ids + // are merged, never dropped, and the thread set decides ambiguity. + interface MeasuredHistoryGroup { + readonly first: ExtractedSessionHistory; + readonly threadIds: Set; + readonly instanceIds: Set; + parent: string | null; + } + const measuredHistory = new Map(); + const unmeasuredHistory = new Map(); + for (const entry of input.history) { + if (entry.usageProvider !== null) { + const key = sessionKey(entry.usageProvider, entry.nativeSessionId); + const group = measuredHistory.get(key) ?? { + first: entry, + threadIds: new Set(), + instanceIds: new Set(), + parent: null, + }; + group.threadIds.add(entry.threadId); + if (entry.providerInstanceId !== null) group.instanceIds.add(entry.providerInstanceId); + if (group.parent === null && entry.parentNativeSessionId !== null) { + group.parent = entry.parentNativeSessionId; + } + measuredHistory.set(key, group); + } else { + const key = `${entry.threadId}\u0000${entry.providerName}\u0000${normalizeProviderInstanceKey( + entry.providerInstanceId, + )}\u0000${entry.nativeSessionId}`; + if (!unmeasuredHistory.has(key)) unmeasuredHistory.set(key, entry); + } + } + + // 1a. Measured durable identities, including providers T3 cannot measure. An + // unmeasured identity reports a null total, never a zero. + for (const [baseKey, group] of measuredHistory) { + consumedBaseKeys.add(baseKey); + const baseSession = baseByKey.get(baseKey); + if (baseSession !== undefined) { + const boundThreadIds = unionSorted(baseSession.boundThreadIds, group.threadIds); + const threadId = boundThreadIds.length === 1 ? boundThreadIds[0]! : null; + sessionReports.push( + reportForBase( + baseSession, + threadId, + group.parent, + boundThreadIds, + unionSorted(baseSession.providerInstanceIds, group.instanceIds), + ), + ); + continue; + } + // A measured-provider history row with no matching base slot: usage unknown. + const boundThreadIds = [...group.threadIds].toSorted(); + const threadId = boundThreadIds.length === 1 ? boundThreadIds[0]! : null; + sessionReports.push( + reportForUnmeasured(group.first, threadId, boundThreadIds, [...group.instanceIds].toSorted()), + ); + } + + // 1b. Unmeasured durable identities: each row is its own history + // contribution and its provider instance is preserved. + for (const entry of unmeasuredHistory.values()) { + sessionReports.push( + reportForUnmeasured( + entry, + entry.threadId, + [entry.threadId], + entry.providerInstanceId === null ? [] : [entry.providerInstanceId], + ), + ); + } + + // 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 threadId = session.boundThreadIds.length === 1 ? session.boundThreadIds[0]! : null; + sessionReports.push( + reportForBase(session, threadId, null, session.boundThreadIds, session.providerInstanceIds), + ); + } + + 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); + // Every thread that appears in the binding evidence, even if all of its + // sessions are ambiguous, so a candidate thread is visible with zero usage + // rather than silently disappearing. + for (const binding of input.bindings) threadIds.add(binding.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)), + }; + }); + + // 4. Usage held in no single thread is pooled exactly once. An ambiguous + // session (bound to two threads) is never summed into either candidate. + let unallocated = ZERO; + for (const session of sessionReports) { + if (session.threadId === null && session.usage !== null) { + unallocated = addTotalsOf(unallocated, session.usage); + } + } + + 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, + ambiguousSessions: sessionReports.filter((session) => session.allocation === "ambiguous") + .length, + conflictingSelections: input.routeEvents.filter((event) => event.selectionConflict).length, + }; + + return { + contractVersion: USAGE_ROUTE_ATTRIBUTION_VERSION, + generatedAtMs: input.cutoffMs, + base, + sessions: sessionReports, + threads, + unallocated, + 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 (sessions.some((session) => session.allocation === "ambiguous")) { + limitations.push( + "Some sessions are bound to more than one thread. Their usage is reported once in `unallocated`, never split or duplicated across candidate threads.", + ); + } + 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 6585e291043e..0f48145461ab 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 @@ -172,13 +173,77 @@ fallback rows, and a deleted file is never re-parsed at all, so its history surv with conservative `partial` quality. Native-id availability alone does **not** establish numeric metadata freshness; the two are separate axes. +## 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, provider_instance_key, native_session_id)`. `provider_instance_key` +is a normalized non-null string (a trimmed instance id, or `""` for an unknown/null +instance), because SQLite treats `NULL` as distinct in a `UNIQUE` key; two instances of one +driver that expose the same native session id on one thread therefore stay two rows instead +of collapsing and erasing one instance. The runtime writer computes the key the same way, +and the backfill mirrors the runtime `nativeSessionIdOf` precedence exactly — `resume`, +then `threadId`, then `sessionId`, accepting only a non-empty JSON text value and falling +through on absent, null, non-string, or blank candidates. The whole `ProviderSessionRuntime.upsert` +is one SQLite transaction: the runtime cursor, the history row it implies, and the route +events commit together or not at all. + +A repeated observation of one identity only advances `last_seen_at`; it never replaces a +different session's row, and a conflicting `onConflict: "ignore"` write appends nothing +because that cursor was never applied. `extractAttributionHistory` de-duplicates its input +by the same canonical identity, so a join that repeats a row cannot become a second report. + +`parent_native_session_id` carries a true sub-agent parent when a caller supplies one. T3 +does not yet persist OpenCode child session ids (they live only in the adapter's in-memory +`relatedSessionIds` and its native event log), and no production caller sets this field yet, +so a child identity is structurally supported and 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. + +The automatic request record is monotonically enrichable rather than first-write-wins: a +later observation may fill a previously unknown/null requested field, a model-less +recovery/stop/rollback write never erases a known value, and a conflicting non-null value is +retained (first non-null wins) while `selection_conflict` is set to surface the +disagreement. `escalation_reason` is a bounded lowercase code/slug, not free text; anything +outside the allowlist length/charset is rejected rather than stored. Declared event ids are +scoped by thread so a globally reused id cannot suppress another thread's event. + +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-session allocation and provider-instance identity, per-thread rollups, and +identity diagnostics. A session bound to more than one thread is `ambiguous`; its usage is +not duplicated onto each candidate thread but pooled once in the route view's `unallocated` +bucket, so per-thread totals plus `unallocated` reconcile additively to each session's +usage exactly once. 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 +storage or transport for the result and registers no endpoint. The two new tables have a +tested write seam and extraction seam but no production read consumer yet. Only the +automatic request record is written by a production caller; every declared kind and +`parent_native_session_id` require a future agent-config/harness producer, so they are +structurally supported and remain `null`/absent in practice today. Provider-instance +identity is carried through the route view but is still not recoverable from a transcript scan; correlate that when the scan starts tagging files with the instance that produced them.