diff --git a/apps/web/src/routers/cli-sessions-v2-router.test.ts b/apps/web/src/routers/cli-sessions-v2-router.test.ts index 4b9b9bd5c9..50e77912b9 100644 --- a/apps/web/src/routers/cli-sessions-v2-router.test.ts +++ b/apps/web/src/routers/cli-sessions-v2-router.test.ts @@ -1713,4 +1713,138 @@ describe('cli-sessions-v2-router', () => { ]); }); }); + + describe('list / search hide never-ingested placeholders', () => { + // Bare POST /api/session placeholders: title/status/cost NULL and platform + // still at the column default 'unknown'. Content may still exist in DO/R2; + // the four-column conjunction is only a list/search visibility predicate. + const placeholderId = 'ses_hide_placeholder_bare_0001'; + const titledUnknownId = 'ses_hide_placeholder_titled_0001'; + const statusUnknownId = 'ses_hide_placeholder_status_0001'; + const costOnlyZeroId = 'ses_hide_placeholder_cost0_0001'; + const normalCliId = 'ses_hide_placeholder_cli_0001'; + const allSessionIds = [ + placeholderId, + titledUnknownId, + statusUnknownId, + costOnlyZeroId, + normalCliId, + ]; + + beforeEach(async () => { + const baseTime = Date.parse('2026-06-01T12:00:00.000Z'); + await db.insert(cli_sessions_v2).values([ + { + session_id: placeholderId, + kilo_user_id: regularUser.id, + // defaults: created_on_platform 'unknown', title/status/cost NULL + created_at: new Date(baseTime).toISOString(), + updated_at: new Date(baseTime).toISOString(), + }, + { + session_id: titledUnknownId, + kilo_user_id: regularUser.id, + created_on_platform: 'unknown', + title: 'titled but still unknown platform', + created_at: new Date(baseTime + 1000).toISOString(), + updated_at: new Date(baseTime + 1000).toISOString(), + }, + { + session_id: statusUnknownId, + kilo_user_id: regularUser.id, + created_on_platform: 'unknown', + status: 'running', + created_at: new Date(baseTime + 2000).toISOString(), + updated_at: new Date(baseTime + 2000).toISOString(), + }, + { + session_id: costOnlyZeroId, + kilo_user_id: regularUser.id, + created_on_platform: 'unknown', + // Metrics emission can persist 0 (writer clamps with Math.max(0, …)) + // while no metadata projection ever succeeded. + total_cost_microdollars: 0, + created_at: new Date(baseTime + 3000).toISOString(), + updated_at: new Date(baseTime + 3000).toISOString(), + }, + { + session_id: normalCliId, + kilo_user_id: regularUser.id, + created_on_platform: 'cli', + title: 'normal cli session', + status: 'completed', + created_at: new Date(baseTime + 4000).toISOString(), + updated_at: new Date(baseTime + 4000).toISOString(), + }, + ]); + }); + + afterEach(async () => { + await db.delete(cli_sessions_v2).where(inArray(cli_sessions_v2.session_id, allSessionIds)); + }); + + it('list omits bare placeholder rows', async () => { + const caller = await createCallerForUser(regularUser.id); + const result = await caller.cliSessionsV2.list({}); + const ids = result.cliSessions.map(session => session.session_id); + + expect(ids).not.toContain(placeholderId); + }); + + it('list returns a row with a title even when platform is still unknown', async () => { + const caller = await createCallerForUser(regularUser.id); + const result = await caller.cliSessionsV2.list({}); + const ids = result.cliSessions.map(session => session.session_id); + + expect(ids).toContain(titledUnknownId); + }); + + it('list returns a row with a status even when title is null and platform is unknown', async () => { + const caller = await createCallerForUser(regularUser.id); + const result = await caller.cliSessionsV2.list({}); + const ids = result.cliSessions.map(session => session.session_id); + + expect(ids).toContain(statusUnknownId); + }); + + it('list returns a row with only total_cost_microdollars set (including zero)', async () => { + const caller = await createCallerForUser(regularUser.id); + const result = await caller.cliSessionsV2.list({}); + const ids = result.cliSessions.map(session => session.session_id); + + expect(ids).toContain(costOnlyZeroId); + const costOnly = result.cliSessions.find(session => session.session_id === costOnlyZeroId); + expect(costOnly?.total_cost_microdollars).toBe(0); + }); + + it('list returns a normal cli session and keeps pagination stable with placeholders interleaved', async () => { + const caller = await createCallerForUser(regularUser.id); + // Fixtures are ordered by created_at; placeholders sit between visible + // rows. limit=2 over created_at should page only visible rows. + const page1 = await caller.cliSessionsV2.list({ limit: 2, orderBy: 'created_at' }); + const page1Ids = page1.cliSessions.map(session => session.session_id); + + expect(page1Ids).toEqual([normalCliId, costOnlyZeroId]); + expect(page1Ids).not.toContain(placeholderId); + expect(page1.nextCursor).not.toBeNull(); + + const page2 = await caller.cliSessionsV2.list({ + limit: 2, + orderBy: 'created_at', + cursor: page1.nextCursor!, + }); + const page2Ids = page2.cliSessions.map(session => session.session_id); + + expect(page2Ids).toEqual([statusUnknownId, titledUnknownId]); + expect(page2Ids).not.toContain(placeholderId); + }); + + it('search by exact session_id does not return a bare placeholder', async () => { + const caller = await createCallerForUser(regularUser.id); + const result = await caller.cliSessionsV2.search({ search_string: placeholderId }); + + expect(result.results.map(session => session.session_id)).not.toContain(placeholderId); + expect(result.total).toBe(0); + }); + }); }); diff --git a/apps/web/src/routers/cli-sessions-v2-router.ts b/apps/web/src/routers/cli-sessions-v2-router.ts index 09a0dad8f6..5a0f0f3447 100644 --- a/apps/web/src/routers/cli-sessions-v2-router.ts +++ b/apps/web/src/routers/cli-sessions-v2-router.ts @@ -470,6 +470,41 @@ async function addOrganizationCondition( whereConditions.push(eq(cli_sessions_v2.organization_id, organizationId)); } +/** + * Hide never-ingested placeholder rows from list/search. + * + * POST /api/session creates bare placeholders (title/status/cost NULL, + * created_on_platform default 'unknown') before any user turn. Metadata and + * cost arrive later via ingest; if the client dies in that window the row + * stays permanently unwritten. + * + * All four columns unwritten proves only that no metadata projection ever + * succeeded and no metrics emission ever persisted a cost. It does not prove + * the row has no content — content lives in the DO and R2 and commits + * independently of the metadata projection. + * + * total_cost_microdollars is written by the alarm-driven metrics emission in + * SessionIngestDO.emitSessionMetrics (best-effort: stays NULL when the metric + * is non-finite or the UPDATE throws and is swallowed), not per flush — so + * presence proves the session reached a metrics emission and must be shown; + * absence proves nothing. + * + * Invariant: any row whose four list columns are all unwritten is hidden, + * regardless of whether the DO holds content. The predicate is the definition + * of what gets hidden; no Postgres-visible signal can do better on a + * paginated list query. + */ +function addHideUningestedPlaceholderCondition(whereConditions: SQL[]): void { + whereConditions.push( + sql`( + ${isNotNull(cli_sessions_v2.title)} + OR ${isNotNull(cli_sessions_v2.status)} + OR ${cli_sessions_v2.created_on_platform} != 'unknown' + OR ${isNotNull(cli_sessions_v2.total_cost_microdollars)} + )` + ); +} + function joinWithAnd(fragments: SQL[]): SQL { return sql.join(fragments, sql` AND `); } @@ -506,6 +541,7 @@ export const cliSessionsV2Router = createTRPCRouter({ await addOrganizationCondition(whereConditions, ctx, organizationId); addCreatedOnPlatformConditions(whereConditions, createdOnPlatform); addGitUrlConditions(whereConditions, gitUrl); + addHideUningestedPlaceholderCondition(whereConditions); if (cursor) { whereConditions.push(lt(orderColumn, cursor)); @@ -586,6 +622,7 @@ export const cliSessionsV2Router = createTRPCRouter({ await addOrganizationCondition(whereConditions, ctx, organizationId); addCreatedOnPlatformConditions(whereConditions, createdOnPlatform); addGitUrlConditions(whereConditions, gitUrl); + addHideUningestedPlaceholderCondition(whereConditions); if (!includeChildren) { whereConditions.push(isNull(cli_sessions_v2.parent_session_id)); diff --git a/packages/worker-utils/src/index.ts b/packages/worker-utils/src/index.ts index c719fb17ea..4e79e1a60b 100644 --- a/packages/worker-utils/src/index.ts +++ b/packages/worker-utils/src/index.ts @@ -196,3 +196,5 @@ export type { SecurityFindingAuditSnapshotSource, SecurityFindingAuditWriterDb, } from './security-finding-audit.js'; + +export { hasOrganizationAccess } from './organization-membership.js'; diff --git a/packages/worker-utils/src/organization-membership.test.ts b/packages/worker-utils/src/organization-membership.test.ts new file mode 100644 index 0000000000..700674111d --- /dev/null +++ b/packages/worker-utils/src/organization-membership.test.ts @@ -0,0 +1,59 @@ +import { describe, expect, it, vi } from 'vitest'; + +import { hasOrganizationAccess } from './organization-membership.js'; + +type MembershipFixture = { kind: 'member' } | { kind: 'non-member' } | { kind: 'soft-deleted' }; + +function createMembershipDb(fixture: MembershipFixture) { + const rows = + fixture.kind === 'member' + ? [{ id: 'mem_1' }] + : // non-member and soft-deleted org both yield no row: the join requires + // membership + organizations.deleted_at IS NULL. + []; + + const limit = vi.fn(async () => rows); + const where = vi.fn(() => ({ limit })); + const innerJoin = vi.fn(() => ({ where })); + const from = vi.fn(() => ({ innerJoin })); + const select = vi.fn(() => ({ from })); + + return { select, from, innerJoin, where, limit, rows }; +} + +describe('hasOrganizationAccess', () => { + it('returns true when the user has a direct membership in a live org', async () => { + const db = createMembershipDb({ kind: 'member' }); + + await expect( + hasOrganizationAccess(db as never, { kiloUserId: 'usr_1', organizationId: 'org_1' }) + ).resolves.toBe(true); + + expect(db.select).toHaveBeenCalledOnce(); + expect(db.from).toHaveBeenCalledOnce(); + expect(db.innerJoin).toHaveBeenCalledOnce(); + expect(db.where).toHaveBeenCalledOnce(); + expect(db.limit).toHaveBeenCalledWith(1); + }); + + it('returns false when the user has no membership row', async () => { + const db = createMembershipDb({ kind: 'non-member' }); + + await expect( + hasOrganizationAccess(db as never, { kiloUserId: 'usr_1', organizationId: 'org_1' }) + ).resolves.toBe(false); + }); + + it('returns false when the user is a member of a soft-deleted org', async () => { + // Soft-deleted orgs are filtered by isNull(organizations.deleted_at) on the + // join, so the query returns no row — same as non-membership to the caller. + const db = createMembershipDb({ kind: 'soft-deleted' }); + + await expect( + hasOrganizationAccess(db as never, { + kiloUserId: 'usr_1', + organizationId: 'org_deleted', + }) + ).resolves.toBe(false); + }); +}); diff --git a/packages/worker-utils/src/organization-membership.ts b/packages/worker-utils/src/organization-membership.ts new file mode 100644 index 0000000000..4828ec3329 --- /dev/null +++ b/packages/worker-utils/src/organization-membership.ts @@ -0,0 +1,46 @@ +import type { WorkerDb } from '@kilocode/db/client'; +import { organization_memberships, organizations } from '@kilocode/db/schema'; +import { and, eq, isNull } from 'drizzle-orm'; + +type OrganizationMembershipDb = Pick; + +/** + * True when the user has a direct membership row for the organization and the + * organization is not soft-deleted. Mirrors worker session-access predicates in + * `cloud-agent-session-access.ts` as a standalone query (no session join). + * + * Deliberately excluded: + * - Parent-organization inherited roles — honoured only by the tRPC path + * (`apps/web/src/routers/organizations/utils.ts`), restricted to + * owner/billing_manager. No worker-side check considers them; adding + * inheritance here would make this the single most permissive worker check in + * the repo, in a security fix. The pre-existing gap is uniform across every + * worker path and is not this PR's to change. + * - `kilocode_users.is_admin` — no worker session-access path consults it, and + * session-ingest JWT auth deliberately discards every JWT claim except + * `kiloUserId`. + */ +export async function hasOrganizationAccess( + db: OrganizationMembershipDb, + params: { kiloUserId: string; organizationId: string } +): Promise { + const rows = await db + .select({ id: organization_memberships.id }) + .from(organization_memberships) + .innerJoin( + organizations, + and( + eq(organizations.id, organization_memberships.organization_id), + isNull(organizations.deleted_at) + ) + ) + .where( + and( + eq(organization_memberships.organization_id, params.organizationId), + eq(organization_memberships.kilo_user_id, params.kiloUserId) + ) + ) + .limit(1); + + return rows[0] !== undefined; +} diff --git a/services/session-ingest/src/ingest/metadata.test.ts b/services/session-ingest/src/ingest/metadata.test.ts index e8cfc88602..0dfbb15b07 100644 --- a/services/session-ingest/src/ingest/metadata.test.ts +++ b/services/session-ingest/src/ingest/metadata.test.ts @@ -31,8 +31,10 @@ vi.mock('../session-events', () => ({ })); import { getWorkerDb } from '@kilocode/db/client'; +import { getSessionAccessCacheDO } from '../dos/SessionAccessCacheDO'; import { notifyUserSessionEvent } from '../session-events'; import { + applyMetadataChanges, CLI_DISCONNECT_ATTENTION_RESET_STATUS, resetAttentionStatusOnCliDisconnect, } from './metadata'; @@ -105,6 +107,116 @@ function createTransactionDb(options: { return { transaction, select, applyUpdate, updateSet, updateWhere }; } +type ApplyMetadataDbOptions = { + /** Membership join row count (0 = unauthorized / missing / soft-deleted). */ + membershipRows?: number; + /** When set, the next non-lock session select is treated as a parent lookup. */ + parentExists?: boolean; + initialStatus?: string | null; + rowMissing?: boolean; +}; + +/** + * Fluent drizzle double for applyMetadataChanges. + * + * Distinguishes query kinds by chain shape: + * - membership (hasOrganizationAccess): select → from → innerJoin → where → limit + * - status lock: select → from → where → limit → for('update') + * - parent / read-back: select → from → where → limit (awaited without for) + */ +function createApplyMetadataDb(options: ApplyMetadataDbOptions = {}) { + const updateSets: unknown[] = []; + const updateWhere = vi.fn(async () => undefined); + const updateSet = vi.fn((values: unknown) => { + updateSets.push(values); + return { where: updateWhere }; + }); + // Named without the substring "update" so oxlint drizzle rules do not flag test spies. + const applyUpdate = vi.fn(() => ({ set: updateSet })); + + const queryLog: Array<'session-lock' | 'membership' | 'parent' | 'read-back'> = []; + let parentLookupDone = false; + + function persistedSessionRow() { + return { + session_id: 'ses_1', + created_at: '2026-01-01T00:00:00.000Z', + updated_at: '2026-01-01T00:00:01.000Z', + title: 'T', + created_on_platform: 'cli', + organization_id: null, + git_url: null, + git_branch: null, + parent_session_id: null, + status: options.initialStatus ?? 'idle', + status_updated_at: '2026-07-25T00:00:00.000Z', + }; + } + + function sessionLimitResult() { + // Dual-mode: `.for('update')` ⇒ status lock; bare await ⇒ parent lookup or read-back. + let settled: Promise | undefined; + + const resolveWithoutFor = () => { + if (options.parentExists !== undefined && !parentLookupDone) { + parentLookupDone = true; + queryLog.push('parent'); + return options.parentExists ? [{ session_id: 'ses_parent' }] : []; + } + queryLog.push('read-back'); + return options.rowMissing ? [] : [persistedSessionRow()]; + }; + + const thenable = { + for: vi.fn(() => { + queryLog.push('session-lock'); + const rows = options.rowMissing + ? [] + : ([{ status: options.initialStatus ?? 'idle' }] satisfies StatusRow[]); + settled = Promise.resolve(rows); + return settled; + }), + then(onFulfilled: (value: unknown[]) => unknown, onRejected?: (reason: unknown) => unknown) { + settled ??= Promise.resolve(resolveWithoutFor()); + return settled.then(onFulfilled, onRejected); + }, + }; + return thenable; + } + + const select = vi.fn(() => ({ + from: vi.fn(() => ({ + innerJoin: vi.fn(() => ({ + where: vi.fn(() => ({ + limit: vi.fn(async () => { + queryLog.push('membership'); + const count = options.membershipRows ?? 0; + return count > 0 ? [{ id: 'mem_1' }] : []; + }), + })), + })), + where: vi.fn(() => ({ + limit: vi.fn(() => sessionLimitResult()), + })), + })), + })); + + const transaction = vi.fn(async (fn: (tx: unknown) => Promise) => + fn({ select, update: applyUpdate }) + ); + + return { + transaction, + select, + applyUpdate, + updateSet, + updateWhere, + updateSets, + queryLog, + membershipQueryCount: () => queryLog.filter(k => k === 'membership').length, + }; +} + describe('resetAttentionStatusOnCliDisconnect', () => { beforeEach(() => { vi.mocked(getWorkerDb).mockReset(); @@ -208,3 +320,216 @@ describe('resetAttentionStatusOnCliDisconnect', () => { expect(notifyUserSessionEvent).not.toHaveBeenCalled(); }); }); + +describe('applyMetadataChanges', () => { + const env = { HYPERDRIVE: { connectionString: 'postgres://unused' } } as never; + const cacheRemove = vi.fn(async () => undefined); + + beforeEach(() => { + vi.mocked(getWorkerDb).mockReset(); + vi.mocked(notifyUserSessionEvent).mockReset(); + vi.mocked(getSessionAccessCacheDO).mockReset(); + cacheRemove.mockReset(); + vi.mocked(getSessionAccessCacheDO).mockReturnValue({ + remove: cacheRemove, + } as never); + vi.spyOn(console, 'warn').mockImplementation(() => undefined); + }); + + it('persists organization_id and invalidates access cache when the user is a member', async () => { + const db = createApplyMetadataDb({ membershipRows: 1 }); + vi.mocked(getWorkerDb).mockReturnValue(db as never); + + await applyMetadataChanges( + env, + 'usr_1', + 'ses_1', + new Map([ + ['orgId', 'org_live'], + ['title', 'Hello'], + ]) + ); + + expect(db.membershipQueryCount()).toBe(1); + expect(db.updateSets).toEqual([ + expect.objectContaining({ + organization_id: 'org_live', + title: 'Hello', + }), + ]); + expect(getSessionAccessCacheDO).toHaveBeenCalledWith(env, { kiloUserId: 'usr_1' }); + expect(cacheRemove).toHaveBeenCalledWith('ses_1'); + expect(notifyUserSessionEvent).toHaveBeenCalledWith( + env, + 'usr_1', + expect.objectContaining({ type: 'session.updated' }), + undefined + ); + }); + + it('refuses unauthorized organization_id while persisting the rest of the batch', async () => { + const db = createApplyMetadataDb({ membershipRows: 0 }); + vi.mocked(getWorkerDb).mockReturnValue(db as never); + const warnSpy = vi.mocked(console.warn); + + await applyMetadataChanges( + env, + 'usr_1', + 'ses_1', + new Map([ + ['orgId', 'org_foreign'], + ['title', 'Kept title'], + ['gitUrl', 'https://github.com/acme/repo.git'], + ['status', 'busy'], + ]) + ); + + expect(db.membershipQueryCount()).toBe(1); + expect(db.updateSets).toHaveLength(1); + const written = db.updateSets[0] as Record; + expect(written).not.toHaveProperty('organization_id'); + expect(written.title).toBe('Kept title'); + expect(written.git_url).toBe('https://github.com/acme/repo'); + expect(written.status).toBe('busy'); + expect(written.status_updated_at).toEqual(expect.any(String)); + expect(warnSpy).toHaveBeenCalledWith( + 'Refusing unauthorized organization_id metadata write', + expect.objectContaining({ + kiloUserId: 'usr_1', + sessionId: 'ses_1', + organizationId: 'org_foreign', + }) + ); + }); + + it('does not treat a refused orgId-only batch as a scope change or session.updated', async () => { + const db = createApplyMetadataDb({ membershipRows: 0 }); + vi.mocked(getWorkerDb).mockReturnValue(db as never); + + await applyMetadataChanges(env, 'usr_1', 'ses_1', new Map([['orgId', 'org_foreign']])); + + // Refused field is stripped; empty updates object skips the UPDATE entirely. + expect(db.applyUpdate).not.toHaveBeenCalled(); + expect(db.updateSets).toEqual([]); + expect(getSessionAccessCacheDO).not.toHaveBeenCalled(); + expect(cacheRemove).not.toHaveBeenCalled(); + expect(notifyUserSessionEvent).not.toHaveBeenCalled(); + }); + + it('still emits session.updated when a refused orgId is paired with parentId', async () => { + const db = createApplyMetadataDb({ membershipRows: 0, parentExists: true }); + vi.mocked(getWorkerDb).mockReturnValue(db as never); + + await applyMetadataChanges( + env, + 'usr_1', + 'ses_1', + new Map([ + ['orgId', 'org_foreign'], + ['parentId', 'ses_parent'], + ]) + ); + + expect(db.updateSets).toEqual([expect.objectContaining({ parent_session_id: 'ses_parent' })]); + expect(getSessionAccessCacheDO).not.toHaveBeenCalled(); + expect(notifyUserSessionEvent).toHaveBeenCalledWith( + env, + 'usr_1', + expect.objectContaining({ type: 'session.updated' }), + undefined + ); + }); + + it('refuses organization_id for a soft-deleted org while persisting the rest', async () => { + // Soft-deleted orgs yield no membership join row (deleted_at IS NULL filter). + const db = createApplyMetadataDb({ membershipRows: 0 }); + vi.mocked(getWorkerDb).mockReturnValue(db as never); + const warnSpy = vi.mocked(console.warn); + + await applyMetadataChanges( + env, + 'usr_1', + 'ses_1', + new Map([ + ['orgId', 'org_deleted'], + ['title', 'Still written'], + ['status', 'idle'], + ]) + ); + + const written = db.updateSets[0] as Record; + expect(written).not.toHaveProperty('organization_id'); + expect(written.title).toBe('Still written'); + expect(written.status).toBe('idle'); + expect(warnSpy).toHaveBeenCalledWith( + 'Refusing unauthorized organization_id metadata write', + expect.objectContaining({ organizationId: 'org_deleted' }) + ); + expect(getSessionAccessCacheDO).not.toHaveBeenCalled(); + }); + + it('performs zero membership queries when orgId is absent', async () => { + const db = createApplyMetadataDb(); + vi.mocked(getWorkerDb).mockReturnValue(db as never); + + await applyMetadataChanges( + env, + 'usr_1', + 'ses_1', + new Map([ + ['title', 'No org'], + ['platform', 'cli'], + ['status', 'busy'], + ]) + ); + + expect(db.membershipQueryCount()).toBe(0); + expect(db.updateSets).toEqual([ + expect.objectContaining({ + title: 'No org', + created_on_platform: 'cli', + status: 'busy', + }), + ]); + const written = db.updateSets[0] as Record; + expect(written).not.toHaveProperty('organization_id'); + expect(getSessionAccessCacheDO).not.toHaveBeenCalled(); + }); + + it('clears organization_id on explicit null without a membership query', async () => { + const db = createApplyMetadataDb(); + vi.mocked(getWorkerDb).mockReturnValue(db as never); + + await applyMetadataChanges(env, 'usr_1', 'ses_1', new Map([['orgId', null]])); + + expect(db.membershipQueryCount()).toBe(0); + expect(db.updateSets).toEqual([expect.objectContaining({ organization_id: null })]); + expect(getSessionAccessCacheDO).toHaveBeenCalledWith(env, { kiloUserId: 'usr_1' }); + expect(cacheRemove).toHaveBeenCalledWith('ses_1'); + }); + + it('refuses a nonexistent org claim without aborting the rest of the batch', async () => { + // Nonexistent org looks like no membership row to the check; never reaches FK. + const db = createApplyMetadataDb({ membershipRows: 0 }); + vi.mocked(getWorkerDb).mockReturnValue(db as never); + + await applyMetadataChanges( + env, + 'usr_1', + 'ses_1', + new Map([ + ['orgId', '00000000-0000-4000-8000-000000000099'], + ['title', 'Survives'], + ['platform', 'cli'], + ['status', 'busy'], + ]) + ); + + const written = db.updateSets[0] as Record; + expect(written).not.toHaveProperty('organization_id'); + expect(written.title).toBe('Survives'); + expect(written.created_on_platform).toBe('cli'); + expect(written.status).toBe('busy'); + expect(db.applyUpdate).toHaveBeenCalled(); + }); +}); diff --git a/services/session-ingest/src/ingest/metadata.ts b/services/session-ingest/src/ingest/metadata.ts index b33a95c86a..9beef2536c 100644 --- a/services/session-ingest/src/ingest/metadata.ts +++ b/services/session-ingest/src/ingest/metadata.ts @@ -1,7 +1,7 @@ import { and, eq, inArray, sql } from 'drizzle-orm'; import { getWorkerDb } from '@kilocode/db/client'; import { cli_sessions_v2 } from '@kilocode/db/schema'; -import { normalizeGitUrl, withDORetry } from '@kilocode/worker-utils'; +import { hasOrganizationAccess, normalizeGitUrl, withDORetry } from '@kilocode/worker-utils'; import type { Env } from '../env'; import { getSessionAccessCacheDO } from '../dos/SessionAccessCacheDO'; @@ -65,13 +65,8 @@ export async function applyMetadataChanges( const parentSessionId = mergedChanges.has('parentId') ? (mergedChanges.get('parentId') ?? null) : undefined; - const changedNonStatus = - mergedChanges.has('title') || - mergedChanges.has('platform') || - mergedChanges.has('orgId') || - mergedChanges.has('gitUrl') || - mergedChanges.has('gitBranch') || - parentSessionId !== undefined; + /** True only when an organization_id write was actually applied (authorized claim or explicit null clear). */ + let organizationIdWriteApplied = false; const notification = await db.transaction(async tx => { const statusChange = @@ -96,6 +91,46 @@ export async function applyMetadataChanges( if (!statusChange) return null; + // Membership check only for non-null org claims; run on the same tx as the UPDATE. + // Residual: SessionIngestDO.writeIngestMetaIfChanged records the claimed orgId in DO + // SQLite and emits a change only when the value differs; after a refused write the DO + // believes the org is set while Postgres does not, so re-sending the same orgId later + // will not re-emit it. Desirable in the attack case; in the benign case (user genuinely + // joins the org afterwards) the session stays personal until the CLI sends a different + // value. Follow-up tracked in the PR body. + if (mergedChanges.has('orgId')) { + const organizationId = mergedChanges.get('orgId') ?? null; + if (organizationId !== null) { + const authorized = await hasOrganizationAccess(tx, { + kiloUserId, + organizationId, + }); + if (!authorized) { + console.warn('Refusing unauthorized organization_id metadata write', { + kiloUserId, + sessionId, + organizationId, + }); + delete updates.organization_id; + } else { + organizationIdWriteApplied = true; + } + } else { + organizationIdWriteApplied = true; + } + } + + // Gate only the orgId contribution: a refused-only orgId must not count as a + // non-status change (no phantom session.updated). Keep parentSessionId and every + // other non-org key exactly as before — do not derive this from `updates` alone. + const changedNonStatus = + mergedChanges.has('title') || + mergedChanges.has('platform') || + organizationIdWriteApplied || + mergedChanges.has('gitUrl') || + mergedChanges.has('gitBranch') || + parentSessionId !== undefined; + if (Object.keys(updates).length > 0) { await tx .update(cli_sessions_v2) @@ -179,7 +214,7 @@ export async function applyMetadataChanges( }; }); - if (mergedChanges.has('orgId')) { + if (organizationIdWriteApplied) { try { await withDORetry( () => getSessionAccessCacheDO(env, { kiloUserId }), diff --git a/services/session-ingest/src/queue-consumer.test.ts b/services/session-ingest/src/queue-consumer.test.ts index cab27e7fa5..7c26642d1d 100644 --- a/services/session-ingest/src/queue-consumer.test.ts +++ b/services/session-ingest/src/queue-consumer.test.ts @@ -945,10 +945,17 @@ describe('queue organization changes', () => { status: null, status_updated_at: null, }; - const selectResults: unknown[][] = [[{ session_id: sessionId }], [persistedSession]]; + // loadSession → membership join (authorized) → read-back after org write. + const selectResults: unknown[][] = [ + [{ session_id: sessionId }], + [{ id: 'mem_1' }], + [persistedSession], + ]; const selectResult = vi.fn(async () => selectResults.shift() ?? []); const select = { from: vi.fn(() => select), + // hasOrganizationAccess joins memberships → organizations (deleted_at IS NULL). + innerJoin: vi.fn(() => select), where: vi.fn(() => select), limit: vi.fn(() => select), for: vi.fn(() => select),