feat(db): persist caller session tag into call_logs for per-session cost attribution (#8249) (#8334)

changelog.d/features/8249-call-logs-session-tag.md

@@ -0,0 +1 @@
+- feat(db): persist caller session tag into call_logs for per-session cost attribution (#8249)

docs/reference/API_REFERENCE.md

@@ -68,6 +68,7 @@ Content-Type: application/json
 | `X-OmniRoute-Progress`   | Request   | Set to `true` for progress events                                                                                                                                                 |
 | `X-Session-Id`           | Request   | Sticky session key for external session affinity                                                                                                                                  |
 | `x_session_id`           | Request   | Underscore variant also accepted (direct HTTP)                                                                                                                                    |
+| `X-OmniRoute-Session-Id` | Request   | Caller-supplied session/conversation tag (also feeds memory). When present, persisted verbatim to `call_logs.session_tag` for per-session cost attribution (#8249) — never synthesized when absent |
 | `Idempotency-Key`        | Request   | Dedup key (5s window)                                                                                                                                                             |
 | `X-Request-Id`           | Request   | Alternative dedup key                                                                                                                                                             |
 | `X-OmniRoute-Cache`      | Response  | `HIT` or `MISS` (non-streaming)                                                                                                                                                   |

open-sse/handlers/chatCore.ts

@@ -857,13 +857,17 @@ export async function handleChatCore({
       pendingWrite: compressionAnalyticsWritePromise,
       skillRequestId,
     });
-  const pipelineSessionId =
+  // #8249: raw header value, kept separate from `pipelineSessionId`'s skillRequestId fallback
+  // below so call_logs.session_tag is only ever set when the caller explicitly supplied the
+  // header — never synthesized from the internal per-request skillRequestId.
+  const explicitSessionIdHeader =
     (clientRawRequest?.headers && typeof clientRawRequest.headers.get === "function"
       ? clientRawRequest.headers.get("x-omniroute-session-id")
       : getHeaderValueCaseInsensitive(
           clientRawRequest?.headers ?? null,
           "x-omniroute-session-id"
-        )) || skillRequestId;
+        )) || null;
+  const pipelineSessionId = explicitSessionIdHeader || skillRequestId;
   // persistAttemptLogs extracted to chatCore/attemptLogging.ts (#3501); bind the per-request context
   // once so the 16 call sites keep passing only the per-attempt args (byte-identical).
   const persistAttemptLogs = (args: PersistAttemptLogsArgs) =>
@@ -891,6 +895,7 @@ export async function handleChatCore({
       noLogEnabled,
       correlationId,
       modelPinned,
+      sessionTag: explicitSessionIdHeader,
     });
 
   // Primary path: merge client model id + alias target so config on either key applies; resolved

open-sse/handlers/chatCore/attemptLogging.ts

@@ -57,6 +57,10 @@ export type PersistAttemptLogsContext = {
   noLogEnabled: unknown;
   correlationId?: string | null;
   modelPinned?: boolean;
+  /** #8249: caller-supplied X-OmniRoute-Session-Id header, only set when the header was
+   * explicitly present (never synthesized from skillRequestId) — persisted as call_logs.session_tag
+   * for per-session cost attribution. */
+  sessionTag?: string | null;
 };
 
 function toConnectionId(value: unknown): string | null {
@@ -171,6 +175,7 @@ export function persistAttemptLogs(args: PersistAttemptLogsArgs, ctx: PersistAtt
     noLogEnabled,
     correlationId,
     modelPinned,
+    sessionTag,
   } = ctx;
   const initialConnectionId = toConnectionId(connectionId);
   const finalConnectionId = toConnectionId(credentials?.connectionId) || initialConnectionId;
@@ -270,6 +275,7 @@ export function persistAttemptLogs(args: PersistAttemptLogsArgs, ctx: PersistAtt
     pipelinePayloads,
     correlationId,
     modelPinned: modelPinned || false,
+    sessionTag: sessionTag || null,
   }).catch(() => {});
 
   // Emit the terminal request-lifecycle event to the live dashboard bus. `request.started`

src/lib/db/migrations/133_call_logs_session_tag.sql

@@ -0,0 +1,2 @@
+ALTER TABLE call_logs ADD COLUMN session_tag TEXT DEFAULT NULL;
+CREATE INDEX IF NOT EXISTS idx_cl_session_tag ON call_logs(session_tag);

src/lib/db/schemaColumns.ts

@@ -253,6 +253,10 @@ export function ensureCallLogsColumns(db: SqliteDatabase) {
       db.exec("ALTER TABLE call_logs ADD COLUMN model_pinned INTEGER DEFAULT 0");
       console.log("[DB] Added call_logs.model_pinned column");
     }
+    if (!columnNames.has("session_tag")) {
+      db.exec("ALTER TABLE call_logs ADD COLUMN session_tag TEXT DEFAULT NULL");
+      console.log("[DB] Added call_logs.session_tag column");
+    }
 
     db.exec(
       "CREATE INDEX IF NOT EXISTS idx_call_logs_requested_model ON call_logs(requested_model)"
@@ -262,6 +266,7 @@ export function ensureCallLogsColumns(db: SqliteDatabase) {
       "CREATE INDEX IF NOT EXISTS idx_cl_combo_target ON call_logs(combo_name, combo_execution_key, timestamp)"
     );
     db.exec("CREATE INDEX IF NOT EXISTS idx_cl_correlation_id ON call_logs(correlation_id)");
+    db.exec("CREATE INDEX IF NOT EXISTS idx_cl_session_tag ON call_logs(session_tag)");
   } catch (error: unknown) {
     const message = error instanceof Error ? error.message : String(error);
     console.warn("[DB] Failed to verify call_logs schema:", message);

src/lib/usage/callLogs.ts

@@ -93,6 +93,7 @@ type CallLogSummaryRow = {
   resolved_account?: string | null;
   correlation_id?: string | null;
   model_pinned?: number | null;
+  session_tag?: string | null;
 };
 
 const RESOLVED_ACCOUNT_SQL = "COALESCE(NULLIF(pc.name, ''), NULLIF(pc.email, ''), cl.account)";
@@ -535,6 +536,7 @@ function mapSummaryRow(row: CallLogSummaryRow) {
     hasPipelineDetails: toNumber(row.has_pipeline_details) === 1,
     correlationId: row.correlation_id || null,
     modelPinned: toNumber(row.model_pinned) === 1,
+    sessionTag: row.session_tag || null,
   };
 }
 
@@ -624,6 +626,7 @@ export async function saveCallLog(entry: any) {
         toStringOrNull(entry.comboExecutionKey) || toStringOrNull(entry.comboStepId),
       correlationId: entry.correlationId || null,
       modelPinned: entry.modelPinned ? 1 : 0,
+      sessionTag: entry.sessionTag || null,
     };
 
     const requestSummary = noLogEnabled
@@ -672,7 +675,7 @@ export async function saveCallLog(entry: any) {
         combo_name, combo_step_id, combo_execution_key, error_summary, detail_state,
         artifact_relpath, artifact_size_bytes, artifact_sha256,
         has_request_body, has_response_body, has_pipeline_details, request_summary,
-        correlation_id, model_pinned
+        correlation_id, model_pinned, session_tag
       )
       VALUES (
         @id, @timestamp, @method, @path, @status, @model, @requestedModel, @provider,
@@ -683,7 +686,7 @@ export async function saveCallLog(entry: any) {
         @comboName, @comboStepId, @comboExecutionKey, @errorSummary, @detailState,
         @artifactRelPath, @artifactSizeBytes, @artifactSha256,
         @hasRequestBody, @hasResponseBody, @hasPipelineDetails, @requestSummary,
-        @correlationId, @modelPinned
+        @correlationId, @modelPinned, @sessionTag
       )
     `
     ).run({
@@ -757,6 +760,22 @@ if (shouldPersistToDisk && process.env.NODE_ENV !== "test") {
   scheduleCallLogRotation();
 }
 
+/**
+ * Pushes a `column LIKE %value%` condition (mirrors the correlationId/sessionTag substring-match
+ * precedent). Extracted so getCallLogs stays under the max-lines-per-function ratchet — #8249.
+ */
+function pushLikeFilter(
+  conditions: string[],
+  params: Record<string, unknown>,
+  column: string,
+  paramKey: string,
+  value: unknown
+) {
+  if (!value) return;
+  conditions.push(`cl.${column} LIKE @${paramKey}`);
+  params[paramKey] = `%${value}%`;
+}
+
 export async function getCallLogs(filter: any = {}) {
   const db = getDbInstance();
   let sql = `
@@ -800,10 +819,8 @@ export async function getCallLogs(filter: any = {}) {
     conditions.push("(cl.api_key_name LIKE @apiKeyQ OR cl.api_key_id LIKE @apiKeyQ)");
     params.apiKeyQ = `%${filter.apiKey}%`;
   }
-  if (filter.correlationId) {
-    conditions.push("cl.correlation_id LIKE @correlationId");
-    params.correlationId = `%${filter.correlationId}%`;
-  }
+  pushLikeFilter(conditions, params, "correlation_id", "correlationId", filter.correlationId);
+  pushLikeFilter(conditions, params, "session_tag", "sessionTag", filter.sessionTag);
   if (filter.combo) {
     conditions.push("cl.combo_name IS NOT NULL");
   }

tests/unit/call-logs-session-tag.test.ts

@@ -0,0 +1,132 @@
+import test from "node:test";
+import assert from "node:assert/strict";
+import fs from "node:fs";
+import os from "node:os";
+import path from "node:path";
+
+// #8249: caller session tag (X-OmniRoute-Session-Id header) propagated into call_logs
+// so operators can attribute cost per caller session. Isolated DATA_DIR per PII learnings §3
+// (resetDbInstance() + handle cleanup in test.after so the node:test runner doesn't hang).
+const TEST_DATA_DIR = fs.mkdtempSync(path.join(os.tmpdir(), "omniroute-session-tag-"));
+process.env.DATA_DIR = TEST_DATA_DIR;
+
+const core = await import("../../src/lib/db/core.ts");
+const callLogs = await import("../../src/lib/usage/callLogs.ts");
+
+test.after(() => {
+  core.resetDbInstance();
+  fs.rmSync(TEST_DATA_DIR, { recursive: true, force: true });
+});
+
+test("saveCallLog persists sessionTag when explicitly supplied", async () => {
+  const testId = `test-sessiontag-${Date.now()}`;
+
+  await callLogs.saveCallLog({
+    id: testId,
+    method: "POST",
+    path: "/v1/chat/completions",
+    status: 200,
+    model: "test-model",
+    provider: "test-provider",
+    duration: 100,
+    tokens: { in: 10, out: 5 },
+    sessionTag: "sess-abc",
+  });
+
+  const db = core.getDbInstance();
+  const row = db
+    .prepare("SELECT id, session_tag FROM call_logs WHERE id = ?")
+    .get(testId) as Record<string, unknown>;
+  assert.ok(row, "row should exist in call_logs");
+  assert.equal(row.session_tag, "sess-abc");
+});
+
+test("saveCallLog stores NULL session_tag when absent (never synthesized)", async () => {
+  const testId = `test-nosessiontag-${Date.now()}`;
+
+  await callLogs.saveCallLog({
+    id: testId,
+    method: "POST",
+    path: "/v1/chat/completions",
+    status: 200,
+    model: "test-model",
+    provider: "test-provider",
+    duration: 100,
+    tokens: { in: 10, out: 5 },
+  });
+
+  const db = core.getDbInstance();
+  const row = db
+    .prepare("SELECT id, session_tag FROM call_logs WHERE id = ?")
+    .get(testId) as Record<string, unknown>;
+  assert.ok(row, "row should exist in call_logs");
+  assert.equal(row.session_tag, null, "session_tag must be null when no header was supplied");
+});
+
+test("getCallLogs returns sessionTag on the mapped row", async () => {
+  const testId = `test-getsessiontag-${Date.now()}`;
+
+  await callLogs.saveCallLog({
+    id: testId,
+    method: "POST",
+    path: "/v1/chat/completions",
+    status: 200,
+    model: "test-model",
+    provider: "test-provider",
+    duration: 100,
+    tokens: { in: 10, out: 5 },
+    sessionTag: "sess-roundtrip",
+  });
+
+  const logs = await callLogs.getCallLogs({ limit: 200 });
+  const found = logs.find((l: { id: string }) => l.id === testId);
+  assert.ok(found, "log entry should be found via getCallLogs");
+  assert.equal(found.sessionTag, "sess-roundtrip");
+});
+
+test("getCallLogs filters by sessionTag (substring match, mirroring correlationId)", async () => {
+  const idMatch = `test-filter-match-${Date.now()}`;
+  const idOther = `test-filter-other-${Date.now()}`;
+
+  await callLogs.saveCallLog({
+    id: idMatch,
+    method: "POST",
+    path: "/v1/chat/completions",
+    status: 200,
+    model: "test-model",
+    provider: "test-provider",
+    duration: 100,
+    tokens: { in: 10, out: 5 },
+    sessionTag: "customer-42-session",
+  });
+  await callLogs.saveCallLog({
+    id: idOther,
+    method: "POST",
+    path: "/v1/chat/completions",
+    status: 200,
+    model: "test-model",
+    provider: "test-provider",
+    duration: 100,
+    tokens: { in: 10, out: 5 },
+    sessionTag: "unrelated-session",
+  });
+
+  const results = await callLogs.getCallLogs({ sessionTag: "customer-42" });
+  const ids = results.map((r: { id: string }) => r.id);
+  assert.ok(ids.includes(idMatch), "matching sessionTag row must be returned");
+  assert.ok(!ids.includes(idOther), "non-matching sessionTag row must be excluded");
+});
+
+test("schemaColumns self-heal ALTER for session_tag is idempotent", async () => {
+  const db = core.getDbInstance();
+  // Re-run the exact idempotent guard twice; must not throw either time.
+  const { ensureCallLogsColumns } = await import("../../src/lib/db/schemaColumns.ts");
+  assert.doesNotThrow(() => ensureCallLogsColumns(db));
+  assert.doesNotThrow(() => ensureCallLogsColumns(db));
+
+  const columns = db.prepare("PRAGMA table_info(call_logs)").all() as Array<{ name: string }>;
+  assert.ok(
+    columns.some((c) => c.name === "session_tag"),
+    "call_logs.session_tag column must exist"
+  );
+});