Merge pull request #274 from closedloop-ai/feat/fea-1548-identity-columns

FEA-1548: Add session multi-tenant identity columns

Author: mikeangstadt

apps/desktop/src/main/agent-session-sync-contract.ts

@@ -79,6 +79,13 @@ export type SyncedAgentSession = {
   awaitingInputSince?: string | null;
   metadata?: SyncJsonObject | null;
   attribution?: SyncedAgentSessionAttribution;
+  /**
+   * FEA-1548: multi-tenant identity columns. Nullable — sessions created before
+   * account signup have no user/org context. Optional + additive; the relay
+   * already ignores unknown fields, so this is backward-compatible.
+   */
+  userId?: string | null;
+  organizationId?: string | null;
   agents: SyncedAgentSessionAgent[];
   events: SyncedAgentSessionEvent[];
   tokenUsageByModel: SyncedAgentSessionTokenUsage[];

apps/desktop/src/main/agent-session-sync-service.ts

@@ -73,6 +73,8 @@ type SessionRow = {
   metadata: string | null;
   harness: string | null;
   billing_mode: string | null;
+  user_id: string | null;
+  organization_id: string | null;
 };
 
 type AgentRow = {
@@ -921,6 +923,12 @@ export function loadSyncedSessions(
     return [];
   }
 
+  const hasIdentityCols =
+    columnExists(db, "sessions", "user_id")
+    && columnExists(db, "sessions", "organization_id");
+  const identityColsSql = hasIdentityCols
+    ? "user_id, organization_id"
+    : "NULL AS user_id, NULL AS organization_id";
   const sessionRows = selectRowsByIds<SessionRow>(
     db,
     `
@@ -936,7 +944,8 @@ export function loadSyncedSessions(
         awaiting_input_since,
         metadata,
         harness,
-        billing_mode
+        billing_mode,
+        ${identityColsSql}
       FROM sessions
       WHERE id IN (__IDS__)
     `,
@@ -1044,6 +1053,8 @@ export function loadSyncedSessions(
         awaitingInputSince: row.awaiting_input_since,
         metadata: parseJsonObjectText(row.metadata),
         ...(attribution ? { attribution } : {}),
+        ...(row.user_id != null ? { userId: row.user_id } : {}),
+        ...(row.organization_id != null ? { organizationId: row.organization_id } : {}),
         agents: (agentsBySessionId.get(id) ?? []).map((agentRow) => ({
           externalAgentId: agentRow.id,
           name: agentRow.name,
@@ -1109,6 +1120,17 @@ export function resolveBillingModeForRow(row: SessionRow): BillingMode {
   });
 }
 
+function columnExists(
+  db: DatabaseSync,
+  table: string,
+  column: string,
+): boolean {
+  const rows = db
+    .prepare(`PRAGMA table_info(${table})`)
+    .all() as Array<{ name: string }>;
+  return rows.some((row) => row.name === column);
+}
+
 function selectRowsByIds<T>(
   db: DatabaseSync,
   sql: string,

apps/desktop/src/main/collectors/import-session.ts

@@ -37,6 +37,11 @@ export interface ImporterDeps {
   now?: () => string;
   /** Key-free diagnostic sink. */
   log?: (message: string) => void;
+  /**
+   * FEA-1548: resolve the current authenticated user's identity for stamping
+   * on new sessions. Returns null when no user is signed in.
+   */
+  getUserIdentity?: () => { userId: string; organizationId: string | null } | null;
 }
 
 export interface ImportResult {
@@ -64,8 +69,8 @@ export function createImporter(db: DatabaseSync, deps: ImporterDeps): Importer {
     "SELECT id, status, ended_at FROM sessions WHERE id = ?",
   );
   const insertSessionStmt = db.prepare(`
-    INSERT INTO sessions (id, name, status, cwd, model, started_at, updated_at, ended_at, harness, billing_mode, metadata)
-    VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+    INSERT INTO sessions (id, name, status, cwd, model, started_at, updated_at, ended_at, harness, billing_mode, metadata, user_id, organization_id)
+    VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
   `);
   // Fill only missing fields on an existing row; never clobber a live status.
   // Always refresh metadata so new fields (diffStats, artifacts, etc.) are populated.
@@ -193,6 +198,7 @@ export function createImporter(db: DatabaseSync, deps: ImporterDeps): Importer {
       if (!existing) {
         const status = recentlyActive ? "active" : "completed";
         const billingMode = safe(() => deps.detectBillingMode(harness)) ?? "unknown";
+        const identity = safe(() => deps.getUserIdentity?.()) ?? null;
         insertSessionStmt.run(
           session.sessionId,
           session.name ?? null,
@@ -205,6 +211,8 @@ export function createImporter(db: DatabaseSync, deps: ImporterDeps): Importer {
           harness,
           billingMode,
           buildMetadata(session, harness),
+          identity?.userId ?? null,
+          identity?.organizationId ?? null,
         );
         insertAgentStmt.run(
           mainId,

apps/desktop/src/main/database/lifecycle.ts

@@ -52,6 +52,11 @@ export interface LifecycleDeps {
   log?: (message: string) => void;
   /** Minutes of inactivity after which a still-active session is abandoned. */
   staleMinutes?: number;
+  /**
+   * FEA-1548: resolve the current authenticated user's identity for stamping
+   * on new sessions. Returns null when no user is signed in.
+   */
+  getUserIdentity?: () => { userId: string; organizationId: string | null } | null;
 }
 
 const COMPACTION_RE = /compact|compress|context.*(reduc|truncat|summar)/i;
@@ -85,8 +90,8 @@ export function createLifecycle(db: DatabaseSync, deps: LifecycleDeps) {
     "SELECT id, status, harness, billing_mode, model FROM sessions WHERE id = ?",
   );
   const insertSessionStmt = db.prepare(`
-    INSERT INTO sessions (id, name, status, cwd, model, started_at, updated_at, harness, billing_mode)
-    VALUES (?, ?, 'active', ?, ?, ?, ?, ?, ?)
+    INSERT INTO sessions (id, name, status, cwd, model, started_at, updated_at, harness, billing_mode, user_id, organization_id)
+    VALUES (?, ?, 'active', ?, ?, ?, ?, ?, ?, ?, ?)
   `);
   const setSessionStatusStmt = db.prepare(
     "UPDATE sessions SET status = ?, updated_at = ?, ended_at = ? WHERE id = ?",
@@ -169,6 +174,7 @@ export function createLifecycle(db: DatabaseSync, deps: LifecycleDeps) {
       return existing;
     }
     const billingMode = safe(() => deps.detectBillingMode(harness)) ?? "unknown";
+    const identity = safe(() => deps.getUserIdentity?.()) ?? null;
     insertSessionStmt.run(
       sessionId,
       data.session_name ?? null,
@@ -178,6 +184,8 @@ export function createLifecycle(db: DatabaseSync, deps: LifecycleDeps) {
       now,
       harness,
       billingMode,
+      identity?.userId ?? null,
+      identity?.organizationId ?? null,
     );
     // Every session has a synthetic main agent.
     insertAgentStmt.run(

apps/desktop/src/main/database/schema.ts

@@ -1,4 +1,4 @@
-export const CURRENT_SCHEMA_VERSION = 5;
+export const CURRENT_SCHEMA_VERSION = 6;
 
 /**
  * Each migration runs against the DB when user_version < CURRENT_SCHEMA_VERSION.
@@ -121,5 +121,16 @@ ALTER TABLE sessions ADD COLUMN billing_mode TEXT;
   `
 CREATE INDEX IF NOT EXISTS idx_sessions_started_at ON sessions(started_at DESC);
 CREATE INDEX IF NOT EXISTS idx_sessions_status_started_at ON sessions(status, started_at DESC);
+`,
+
+  // Version 5 -> 6 (FEA-1548 Phase 1): multi-tenant identity columns.
+  // Nullable because sessions created before account signup have no user/org
+  // context. Backfilled when the user creates an account and joins an org.
+  `
+ALTER TABLE sessions ADD COLUMN user_id TEXT;
+ALTER TABLE sessions ADD COLUMN organization_id TEXT;
+
+CREATE INDEX IF NOT EXISTS idx_sessions_user_id ON sessions(user_id) WHERE user_id IS NOT NULL;
+CREATE INDEX IF NOT EXISTS idx_sessions_organization_id ON sessions(organization_id) WHERE organization_id IS NOT NULL;
 `,
 ];

apps/desktop/src/main/database/sessions.ts

@@ -78,6 +78,8 @@ export function createSessionStore(db: DatabaseSync) {
       metadata: (raw.metadata as string) ?? null,
       harness: (raw.harness as string) ?? null,
       billingMode: (raw.billing_mode as string) ?? null,
+      userId: (raw.user_id as string) ?? null,
+      organizationId: (raw.organization_id as string) ?? null,
     };
   }
 

apps/desktop/src/shared/agent-db-contract.ts

@@ -25,6 +25,8 @@ export interface SessionRow {
   metadata: string | null;
   harness: string | null;
   billingMode: string | null;
+  userId: string | null;
+  organizationId: string | null;
 }
 
 export interface AgentRow {

apps/desktop/test/agent-session-sync-service.test.ts

@@ -73,7 +73,9 @@ test("agent-session sync loads normalized session payloads with attribution and
       awaiting_input_since TEXT,
       metadata TEXT,
       harness TEXT NOT NULL,
-      billing_mode TEXT NOT NULL DEFAULT 'unknown'
+      billing_mode TEXT NOT NULL DEFAULT 'unknown',
+      user_id TEXT,
+      organization_id TEXT
     );
     CREATE TABLE agents (
       id TEXT PRIMARY KEY,

apps/desktop/test/helpers/agent-session-sync-test-utils.ts

@@ -43,7 +43,9 @@ export function createAgentMonitorTestDatabase(rootDir: string): DatabaseSync {
       awaiting_input_since TEXT,
       metadata TEXT,
       harness TEXT NOT NULL,
-      billing_mode TEXT NOT NULL DEFAULT 'unknown'
+      billing_mode TEXT NOT NULL DEFAULT 'unknown',
+      user_id TEXT,
+      organization_id TEXT
     );
     CREATE TABLE agents (
       id TEXT PRIMARY KEY,