[ROUTE6-1:T3] Durable model-canary attribution across resumes and model switches

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);

apps/server/src/persistence/Migrations/054_ProviderSessionHistory.test.ts

@@ -0,0 +1,115 @@
+import { assert, it } from "@effect/vitest";
+import * as Effect from "effect/Effect";
+import * as Layer from "effect/Layer";
+import * as SqlClient from "effect/unstable/sql/SqlClient";
+import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient";
+
+import { migrationManifest, runMigrations } from "../Migrations.ts";
+
+const layer = it.layer(Layer.mergeAll(NodeSqliteClient.layer({ filename: ":memory:" })));
+
+interface HistoryRow {
+  readonly threadId: string;
+  readonly providerName: string;
+  readonly nativeSessionId: string;
+  readonly parentNativeSessionId: string | null;
+  readonly origin: string;
+  readonly firstSeenAt: string;
+  readonly lastSeenAt: string;
+}
+
+layer("054_ProviderSessionHistory", (it) => {
+  it.effect("creates the history tables and backfills the current cursor", () =>
+    Effect.gen(function* () {
+      const sql = yield* SqlClient.SqlClient;
+
+      yield* runMigrations({ toMigrationInclusive: 53 });
+
+      const insertRuntime = (threadId: string, providerName: string, cursor: string | null) =>
+        sql`
+          INSERT INTO provider_session_runtime (
+            thread_id,
+            provider_name,
+            provider_instance_id,
+            adapter_key,
+            runtime_mode,
+            status,
+            last_seen_at,
+            resume_cursor_json,
+            runtime_payload_json
+          )
+          VALUES (
+            ${threadId},
+            ${providerName},
+            ${providerName},
+            ${providerName},
+            'full-access',
+            'running',
+            '2026-09-23T09:00:00.000Z',
+            ${cursor},
+            NULL
+          )
+        `;
+
+      yield* insertRuntime("thread-codex", "codex", '{"threadId":"session-a"}');
+      yield* insertRuntime("thread-claude", "claudeAgent", '{"resume":"session-b"}');
+      // No recognised id field: nothing to backfill.
+      yield* insertRuntime("thread-unknown", "codex", '{"opaque":true}');
+      // No cursor at all.
+      yield* insertRuntime("thread-absent", "codex", null);
+
+      yield* runMigrations({ toMigrationInclusive: 55 });
+
+      const rows = yield* sql<HistoryRow>`
+        SELECT
+          thread_id AS "threadId",
+          provider_name AS "providerName",
+          native_session_id AS "nativeSessionId",
+          parent_native_session_id AS "parentNativeSessionId",
+          origin,
+          first_seen_at AS "firstSeenAt",
+          last_seen_at AS "lastSeenAt"
+        FROM provider_session_history
+        ORDER BY native_session_id ASC
+      `;
+
+      assert.deepStrictEqual(rows, [
+        {
+          threadId: "thread-codex",
+          providerName: "codex",
+          nativeSessionId: "session-a",
+          parentNativeSessionId: null,
+          origin: "runtimeCursor",
+          firstSeenAt: "2026-09-23T09:00:00.000Z",
+          lastSeenAt: "2026-09-23T09:00:00.000Z",
+        },
+        {
+          threadId: "thread-claude",
+          providerName: "claudeAgent",
+          nativeSessionId: "session-b",
+          parentNativeSessionId: null,
+          origin: "runtimeCursor",
+          firstSeenAt: "2026-09-23T09:00:00.000Z",
+          lastSeenAt: "2026-09-23T09:00:00.000Z",
+        },
+      ]);
+
+      const routeEventCount = yield* sql<{ readonly count: number }>`
+        SELECT COUNT(*) AS count FROM thread_route_events
+      `;
+      assert.equal(routeEventCount[0]!.count, 0);
+
+      // The unique key is on the durable identity, so a repeat cannot duplicate.
+      const indexes = yield* sql<{ readonly name: string }>`
+        PRAGMA index_list(provider_session_history)
+      `;
+      assert.ok(indexes.length > 0);
+    }),
+  );
+
+  it("registers both migrations in the manifest", () => {
+    const entries = new Set(migrationManifest.map(([id, name]) => `${id}_${name}`));
+    assert.ok(entries.has("54_ProviderSessionHistory"));
+    assert.ok(entries.has("55_ThreadRouteEvents"));
+  });
+});

apps/server/src/persistence/Migrations/054_ProviderSessionHistory.ts

@@ -0,0 +1,94 @@
+import * as SqlClient from "effect/unstable/sql/SqlClient";
+import * as Effect from "effect/Effect";
+
+/**
+ * Append-only native-session identity history per T3 thread.
+ *
+ * `provider_session_runtime` keeps a single current `resume_cursor_json`. A
+ * resume, fork, or model switch overwrites it, so earlier native sessions that
+ * contributed to a thread become unrecoverable. This table records one row per
+ * `(thread_id, provider_name, native_session_id)` so a thread can answer which
+ * native sessions it used even after the cursor moved on. Repeated observations
+ * of the same session only advance `last_seen_at`; they never replace a
+ * different session's row.
+ *
+ * The backfill seeds the table from whatever cursor already exists so an
+ * upgraded database does not start empty.
+ */
+export default Effect.gen(function* () {
+  const sql = yield* SqlClient.SqlClient;
+
+  yield* sql`
+    CREATE TABLE IF NOT EXISTS provider_session_history (
+      history_id INTEGER PRIMARY KEY AUTOINCREMENT,
+      thread_id TEXT NOT NULL,
+      provider_name TEXT NOT NULL,
+      provider_instance_id TEXT,
+      adapter_key TEXT NOT NULL,
+      native_session_id TEXT NOT NULL,
+      parent_native_session_id TEXT,
+      origin TEXT NOT NULL,
+      first_seen_at TEXT NOT NULL,
+      last_seen_at TEXT NOT NULL,
+      UNIQUE (thread_id, provider_name, native_session_id)
+    )
+  `;
+
+  yield* sql`
+    CREATE INDEX IF NOT EXISTS idx_provider_session_history_thread
+    ON provider_session_history(thread_id, first_seen_at)
+  `;
+
+  yield* sql`
+    INSERT OR IGNORE INTO provider_session_history (
+      thread_id,
+      provider_name,
+      provider_instance_id,
+      adapter_key,
+      native_session_id,
+      parent_native_session_id,
+      origin,
+      first_seen_at,
+      last_seen_at
+    )
+    SELECT
+      current.thread_id,
+      current.provider_name,
+      current.provider_instance_id,
+      current.adapter_key,
+      current.native_session_id,
+      NULL,
+      'runtimeCursor',
+      current.last_seen_at,
+      current.last_seen_at
+    FROM (
+      SELECT
+        runtime.thread_id,
+        runtime.provider_name,
+        runtime.provider_instance_id,
+        runtime.adapter_key,
+        runtime.last_seen_at,
+        COALESCE(
+          json_extract(runtime.cursor, '$.resume'),
+          json_extract(runtime.cursor, '$.threadId'),
+          json_extract(runtime.cursor, '$.sessionId')
+        ) AS native_session_id
+      FROM (
+        SELECT
+          thread_id,
+          provider_name,
+          provider_instance_id,
+          adapter_key,
+          last_seen_at,
+          CASE
+            WHEN resume_cursor_json IS NOT NULL AND json_valid(resume_cursor_json)
+            THEN resume_cursor_json
+            ELSE NULL
+          END AS cursor
+        FROM provider_session_runtime
+      ) AS runtime
+    ) AS current
+    WHERE current.native_session_id IS NOT NULL
+      AND current.native_session_id <> ''
+  `;
+});

apps/server/src/persistence/Migrations/055_ThreadRouteEvents.ts

@@ -0,0 +1,43 @@
+import * as SqlClient from "effect/unstable/sql/SqlClient";
+import * as Effect from "effect/Effect";
+
+/**
+ * Append-only pre-execution route and experiment metadata.
+ *
+ * This is low-cardinality metadata about a routing decision, not conversation
+ * content: which provider/model/effort was requested, the pre-execution task
+ * stratum, the experiment/cohort, the readable manager/agent identifiers, the
+ * route event kind, and any escalation reason. T3 only carries the metadata;
+ * the canonical policy that chooses a route lives in agent-config.
+ *
+ * `route_event_kind` is nullable so an automatic "this is what was requested"
+ * event is distinguishable from a declared canary/fallback/escalation event.
+ * Observed (actual) values are deliberately NOT stored here: they must come
+ * from measured usage, never be copied from the request.
+ */
+export default Effect.gen(function* () {
+  const sql = yield* SqlClient.SqlClient;
+
+  yield* sql`
+    CREATE TABLE IF NOT EXISTS thread_route_events (
+      event_id TEXT PRIMARY KEY,
+      thread_id TEXT NOT NULL,
+      native_session_id TEXT,
+      route_event_kind TEXT,
+      task_stratum TEXT NOT NULL,
+      experiment_id TEXT,
+      manager_id TEXT,
+      agent_id TEXT,
+      requested_provider TEXT,
+      requested_model TEXT,
+      requested_effort TEXT,
+      escalation_reason TEXT,
+      recorded_at TEXT NOT NULL
+    )
+  `;
+
+  yield* sql`
+    CREATE INDEX IF NOT EXISTS idx_thread_route_events_thread
+    ON thread_route_events(thread_id, recorded_at)
+  `;
+});