fix(session-ingest): enforce current organization access

Scope session-ingest operations to the user's current organization and
revoke access when org membership changes. Adds resolveAccessibleKiloSession
and a putValidated cache entry carrying organizationId, used across the
session delete/ingest/share/unshare routes. Moves the org member removal
rowCount check ahead of the audit log so failures bail early.

Addresses review findings on access revocation.

Author: eshurakov

apps/web/src/lib/cloud-agent/session-ownership.test.ts

@@ -8,7 +8,10 @@ import {
   type User,
 } from '@kilocode/db/schema';
 import { and, eq, inArray } from 'drizzle-orm';
-import { queryAccessibleCloudAgentSession } from '@kilocode/worker-utils/cloud-agent-session-access';
+import {
+  queryAccessibleCloudAgentSession,
+  queryAccessibleKiloSession,
+} from '@kilocode/worker-utils/cloud-agent-session-access';
 import {
   verifyOrgOwnsSessionV2ByCloudAgentId,
   verifyUserOwnsSessionV2ByCloudAgentId,
@@ -71,6 +74,10 @@ describe('Cloud Agent session ownership', () => {
   });
 
   beforeEach(async () => {
+    await db
+      .update(organizations)
+      .set({ deleted_at: null })
+      .where(eq(organizations.id, organization.id));
     await db.insert(cli_sessions_v2).values([
       {
         session_id: PERSONAL_SESSION_ID,
@@ -178,6 +185,12 @@ describe('Cloud Agent session ownership', () => {
         cloudAgentSessionId: ORGANIZATION_CLOUD_AGENT_SESSION_ID,
       })
     ).resolves.toBeNull();
+    await expect(
+      queryAccessibleKiloSession(db, {
+        kiloUserId: owner.id,
+        kiloSessionId: ORGANIZATION_SESSION_ID,
+      })
+    ).resolves.toBeNull();
 
     await db.insert(organization_memberships).values({
       organization_id: organization.id,
@@ -202,5 +215,34 @@ describe('Cloud Agent session ownership', () => {
       kiloSessionId: ORGANIZATION_SESSION_ID,
       organizationId: organization.id,
     });
+    await expect(
+      queryAccessibleKiloSession(db, {
+        kiloUserId: owner.id,
+        kiloSessionId: ORGANIZATION_SESSION_ID,
+      })
+    ).resolves.toEqual({
+      kiloSessionId: ORGANIZATION_SESSION_ID,
+      organizationId: organization.id,
+    });
+  });
+
+  it('denies access after the organization is soft-deleted', async () => {
+    await db
+      .update(organizations)
+      .set({ deleted_at: new Date().toISOString() })
+      .where(eq(organizations.id, organization.id));
+
+    await expect(
+      queryAccessibleCloudAgentSession(db, {
+        kiloUserId: owner.id,
+        cloudAgentSessionId: ORGANIZATION_CLOUD_AGENT_SESSION_ID,
+      })
+    ).resolves.toBeNull();
+    await expect(
+      queryAccessibleKiloSession(db, {
+        kiloUserId: owner.id,
+        kiloSessionId: ORGANIZATION_SESSION_ID,
+      })
+    ).resolves.toBeNull();
   });
 });

apps/web/src/lib/organizations/organizations.test.ts

@@ -1,4 +1,4 @@
-import { describe, test, expect, afterEach } from '@jest/globals';
+import { describe, test, expect, afterEach, beforeEach } from '@jest/globals';
 import { db, sql } from '@/lib/drizzle';
 import {
   organizations,
@@ -25,6 +25,11 @@ import {
 } from './organizations';
 import { fromMicrodollars } from '@/lib/utils';
 import { DEFAULT_MEMBER_DAILY_LIMIT_USD } from '@/lib/organizations/constants';
+import { invalidateOrganizationSessionAccess } from '@/lib/session-ingest-client';
+
+jest.mock('@/lib/session-ingest-client', () => ({
+  invalidateOrganizationSessionAccess: jest.fn().mockResolvedValue(undefined),
+}));
 
 describe('Organizations', () => {
   afterEach(async () => {
@@ -400,6 +405,10 @@ describe('Organizations', () => {
   });
 
   describe('removeUserFromOrganization', () => {
+    beforeEach(() => {
+      jest.mocked(invalidateOrganizationSessionAccess).mockClear();
+    });
+
     test('should remove user from organization', async () => {
       const owner = await insertTestUser();
       const member = await insertTestUser();
@@ -418,6 +427,7 @@ describe('Organizations', () => {
       // Verify user was removed
       memberOrgs = await getUserOrganizationsWithSeats(member.id);
       expect(memberOrgs).toHaveLength(0);
+      expect(invalidateOrganizationSessionAccess).toHaveBeenCalledWith(member.id, organization.id);
     });
 
     test('should handle removing non-existent membership gracefully', async () => {
@@ -430,6 +440,23 @@ describe('Organizations', () => {
 
       expect(result).toBeDefined();
       // Should not throw error, just return empty result
+      expect(invalidateOrganizationSessionAccess).not.toHaveBeenCalled();
+    });
+
+    test('should complete removal when session access invalidation fails', async () => {
+      const owner = await insertTestUser();
+      const member = await insertTestUser();
+      const organization = await createOrganization('Test Org', owner.id);
+      await addUserToOrganization(organization.id, member.id, 'member');
+      jest
+        .mocked(invalidateOrganizationSessionAccess)
+        .mockRejectedValueOnce(new Error('invalidation unavailable'));
+
+      await expect(removeUserFromOrganization(organization.id, member.id)).resolves.toBeDefined();
+
+      const memberOrgs = await getUserOrganizationsWithSeats(member.id);
+      expect(memberOrgs).toHaveLength(0);
+      expect(invalidateOrganizationSessionAccess).toHaveBeenCalledWith(member.id, organization.id);
     });
 
     test('should remove specific user without affecting others', async () => {

apps/web/src/lib/organizations/organizations.ts

@@ -30,7 +30,8 @@ import { randomUUID } from 'crypto';
 import { fromMicrodollars, getLowerDomainFromEmail, normalizeEmail } from '@/lib/utils';
 import { resolveEffectiveOrganizationSsoPolicy } from './organization-sso-policy';
 import { classifyOrganizationEntitlement } from './trial-utils';
-import { logExceptInTest } from '@/lib/utils.server';
+import { errorExceptInTest, logExceptInTest } from '@/lib/utils.server';
+import { invalidateOrganizationSessionAccess } from '@/lib/session-ingest-client';
 import { APP_URL } from '@/lib/constants';
 import { createAuditLog } from '@/lib/organizations/organization-audit-logs';
 import { failureResult, successResult } from '@/lib/maybe-result';
@@ -362,7 +363,7 @@ export async function removeUserFromOrganization(
   userId: User['id'],
   removedBy?: User['id']
 ): Promise<{ rowCount: number | null }> {
-  return await db.transaction(async tx => {
+  const result = await db.transaction(async tx => {
     await lockOrganizationMembershipMutation(tx, organizationId, userId);
     // Look up the user's current role before deleting
     const [membership] = await tx
@@ -422,6 +423,34 @@ export async function removeUserFromOrganization(
 
     return result;
   });
+
+  if ((result.rowCount ?? 0) > 0) {
+    await invalidateRemovedMemberSessionAccess(organizationId, userId);
+  }
+
+  return result;
+}
+
+/**
+ * Best-effort: a removed member loses cached Session Ingest access within the
+ * cache TTL even when this call fails, so removal never fails on it.
+ */
+async function invalidateRemovedMemberSessionAccess(
+  organizationId: Organization['id'],
+  userId: User['id']
+): Promise<void> {
+  try {
+    await invalidateOrganizationSessionAccess(userId, organizationId);
+  } catch (error) {
+    errorExceptInTest(
+      'Failed to invalidate cached session access for removed organization member',
+      {
+        organizationId,
+        userId,
+        error: error instanceof Error ? error.message : String(error),
+      }
+    );
+  }
 }
 
 export async function updateUserRoleInOrganization(

apps/web/src/lib/session-ingest-client.test.ts

@@ -6,6 +6,7 @@ import {
   fetchSessionMessages,
   deleteSession,
   shareSession,
+  invalidateOrganizationSessionAccess,
 } from './session-ingest-client';
 
 // ---------------------------------------------------------------------------
@@ -18,6 +19,7 @@ jest.mock('@sentry/nextjs', () => ({
 
 jest.mock('@/lib/config.server', () => ({
   SESSION_INGEST_WORKER_URL: 'https://ingest.test.example.com',
+  INTERNAL_API_SECRET: 'internal-secret',
 }));
 
 jest.mock('@/lib/tokens', () => ({
@@ -410,6 +412,82 @@ describe('shareSession', () => {
 // fetchSessionMessages (thin wrapper)
 // ---------------------------------------------------------------------------
 
+describe('invalidateOrganizationSessionAccess', () => {
+  beforeEach(() => {
+    mockFetch.mockReset();
+    mockCaptureException.mockReset();
+  });
+
+  it('reports and throws invalidation failures', async () => {
+    mockFetch.mockResolvedValue({
+      ok: false,
+      status: 503,
+      statusText: 'Service Unavailable',
+      text: () => Promise.resolve('cache unavailable'),
+    });
+
+    await expect(
+      invalidateOrganizationSessionAccess('usr_removed', '11111111-1111-4111-8111-111111111111')
+    ).rejects.toThrow(
+      'Session access invalidation failed: 503 Service Unavailable - cache unavailable'
+    );
+    expect(mockCaptureException).toHaveBeenCalledWith(
+      expect.any(Error),
+      expect.objectContaining({
+        tags: { source: 'session-ingest-client', endpoint: 'invalidate-session-access' },
+        extra: {
+          kiloUserId: 'usr_removed',
+          organizationId: '11111111-1111-4111-8111-111111111111',
+          status: 503,
+        },
+      })
+    );
+  });
+
+  it('calls the secret-protected organization invalidation endpoint', async () => {
+    mockFetch.mockResolvedValue({ ok: true, status: 204 });
+
+    await invalidateOrganizationSessionAccess(
+      'usr_removed',
+      '11111111-1111-4111-8111-111111111111'
+    );
+
+    expect(mockFetch).toHaveBeenCalledWith(
+      'https://ingest.test.example.com/internal/session-access/invalidate',
+      {
+        method: 'POST',
+        headers: {
+          'content-type': 'application/json',
+          'X-Internal-Secret': 'internal-secret',
+        },
+        body: JSON.stringify({
+          kiloUserId: 'usr_removed',
+          organizationId: '11111111-1111-4111-8111-111111111111',
+        }),
+        signal: expect.any(AbortSignal),
+      }
+    );
+  });
+
+  it('sets a 30-second invalidation deadline', async () => {
+    const signal = new AbortController().signal;
+    const timeout = jest.spyOn(AbortSignal, 'timeout').mockReturnValue(signal);
+    mockFetch.mockResolvedValue({ ok: true, status: 204 });
+
+    await invalidateOrganizationSessionAccess(
+      'usr_removed',
+      '11111111-1111-4111-8111-111111111111'
+    );
+
+    expect(timeout).toHaveBeenCalledWith(30_000);
+    expect(mockFetch).toHaveBeenCalledWith(
+      'https://ingest.test.example.com/internal/session-access/invalidate',
+      expect.objectContaining({ signal })
+    );
+    timeout.mockRestore();
+  });
+});
+
 describe('fetchSessionMessages', () => {
   beforeEach(() => {
     mockFetch.mockReset();

apps/web/src/lib/session-ingest-client.ts

@@ -2,7 +2,7 @@ import 'server-only';
 
 import { captureException } from '@sentry/nextjs';
 import { z } from 'zod';
-import { SESSION_INGEST_WORKER_URL } from '@/lib/config.server';
+import { INTERNAL_API_SECRET, SESSION_INGEST_WORKER_URL } from '@/lib/config.server';
 import { generateInternalServiceToken } from '@/lib/tokens';
 import type { User } from '@kilocode/db/schema';
 
@@ -162,6 +162,44 @@ export async function shareSession(
   return { public_id: body.public_id };
 }
 
+// ---------------------------------------------------------------------------
+// Authorization cache invalidation
+// ---------------------------------------------------------------------------
+
+export async function invalidateOrganizationSessionAccess(
+  kiloUserId: string,
+  organizationId: string
+): Promise<void> {
+  if (!SESSION_INGEST_WORKER_URL) {
+    throw new Error('SESSION_INGEST_WORKER_URL is not configured');
+  }
+  if (!INTERNAL_API_SECRET) {
+    throw new Error('INTERNAL_API_SECRET is not configured');
+  }
+
+  const response = await fetch(`${SESSION_INGEST_WORKER_URL}/internal/session-access/invalidate`, {
+    method: 'POST',
+    headers: {
+      'content-type': 'application/json',
+      'X-Internal-Secret': INTERNAL_API_SECRET,
+    },
+    body: JSON.stringify({ kiloUserId, organizationId }),
+    signal: AbortSignal.timeout(30_000),
+  });
+
+  if (!response.ok) {
+    const errorText = await response.text().catch(() => '');
+    const error = new Error(
+      `Session access invalidation failed: ${response.status} ${response.statusText}${errorText ? ` - ${errorText}` : ''}`
+    );
+    captureException(error, {
+      tags: { source: 'session-ingest-client', endpoint: 'invalidate-session-access' },
+      extra: { kiloUserId, organizationId, status: response.status },
+    });
+    throw error;
+  }
+}
+
 // ---------------------------------------------------------------------------
 // Delete
 // ---------------------------------------------------------------------------