feat(session-ingest): accept worktree-scoped Cloud Agent requests

packages/session-ingest-contracts/src/cloud-agent-worktree-scope.ts

@@ -0,0 +1,123 @@
+import { z } from 'zod';
+
+import { containedKiloSessionIdSchema } from './cloud-agent-session-scope';
+import { cloudAgentWorktreeIdSchema } from './rpc-contract';
+
+export const cloudAgentSessionScopeProtocolVersionV2 = '2' as const;
+
+export const cloudAgentSessionScopeV2Headers = {
+  worktreeId: 'X-Kilo-Cloud-Agent-Worktree',
+  expectedOrganization: 'X-Kilo-Expected-Organization',
+  memberRootKiloSessionIds: 'X-Kilo-Member-Root-Sessions',
+} as const;
+
+export const CLOUD_AGENT_SESSION_SCOPE_PERSONAL_ORGANIZATION = 'personal' as const;
+
+export const MAX_CLOUD_AGENT_SESSION_SCOPE_MEMBER_ROOTS = 32;
+export const MAX_CLOUD_AGENT_SESSION_SCOPE_HEADER_BYTES = 4096;
+
+export const cloudAgentSessionScopeV2AssertionSchema = z.object({
+  protocolVersion: z.literal(cloudAgentSessionScopeProtocolVersionV2),
+  worktreeId: cloudAgentWorktreeIdSchema,
+  expectedOrganization: z.union([
+    z.literal(CLOUD_AGENT_SESSION_SCOPE_PERSONAL_ORGANIZATION),
+    z.uuid(),
+  ]),
+  memberRootKiloSessionIds: z
+    .array(containedKiloSessionIdSchema)
+    .min(1)
+    .max(MAX_CLOUD_AGENT_SESSION_SCOPE_MEMBER_ROOTS),
+});
+
+export type CloudAgentSessionScopeV2Assertion = z.infer<
+  typeof cloudAgentSessionScopeV2AssertionSchema
+>;
+
+export type CloudAgentSessionScopeV2 = {
+  worktreeId: string;
+  organizationId: string | null;
+  memberRootKiloSessionIds: string[];
+};
+
+export type CloudAgentSessionScopeV2HeaderInput = {
+  worktreeId?: string;
+  expectedOrganization?: string;
+  memberRootKiloSessionIds?: string;
+};
+
+export type CloudAgentSessionScopeV2ParseResult =
+  | { kind: 'ok'; scope: CloudAgentSessionScopeV2 }
+  | { kind: 'invalid' };
+
+function headerValue(value: string | undefined): string {
+  return value ?? '';
+}
+
+export function measureCloudAgentSessionScopeV2HeaderBytes(
+  input: CloudAgentSessionScopeV2HeaderInput
+): number {
+  const encoder = new TextEncoder();
+  return (
+    encoder.encode(headerValue(input.worktreeId)).length +
+    encoder.encode(headerValue(input.expectedOrganization)).length +
+    encoder.encode(headerValue(input.memberRootKiloSessionIds)).length
+  );
+}
+
+export function encodeCloudAgentSessionScopeV2Headers(
+  scope: CloudAgentSessionScopeV2
+): Required<CloudAgentSessionScopeV2HeaderInput> {
+  return {
+    worktreeId: scope.worktreeId,
+    expectedOrganization: scope.organizationId ?? CLOUD_AGENT_SESSION_SCOPE_PERSONAL_ORGANIZATION,
+    memberRootKiloSessionIds: scope.memberRootKiloSessionIds.join(','),
+  };
+}
+
+export function parseCloudAgentSessionScopeV2Headers(
+  input: CloudAgentSessionScopeV2HeaderInput
+): CloudAgentSessionScopeV2ParseResult {
+  const { worktreeId, expectedOrganization, memberRootKiloSessionIds } = input;
+  if (
+    worktreeId === undefined ||
+    expectedOrganization === undefined ||
+    memberRootKiloSessionIds === undefined
+  ) {
+    return { kind: 'invalid' };
+  }
+  if (
+    measureCloudAgentSessionScopeV2HeaderBytes(input) > MAX_CLOUD_AGENT_SESSION_SCOPE_HEADER_BYTES
+  ) {
+    return { kind: 'invalid' };
+  }
+
+  const rawMemberRoots = memberRootKiloSessionIds.split(',');
+  if (rawMemberRoots.some(value => value.trim().length === 0)) return { kind: 'invalid' };
+  const normalizedMemberRoots = [...new Set(rawMemberRoots.map(value => value.trim()))];
+  if (
+    normalizedMemberRoots.length === 0 ||
+    normalizedMemberRoots.length > MAX_CLOUD_AGENT_SESSION_SCOPE_MEMBER_ROOTS
+  ) {
+    return { kind: 'invalid' };
+  }
+
+  const parsed = cloudAgentSessionScopeV2AssertionSchema.safeParse({
+    protocolVersion: cloudAgentSessionScopeProtocolVersionV2,
+    worktreeId,
+    expectedOrganization,
+    memberRootKiloSessionIds: normalizedMemberRoots,
+  });
+  if (!parsed.success) return { kind: 'invalid' };
+
+  return {
+    kind: 'ok',
+    scope: {
+      worktreeId: parsed.data.worktreeId,
+      organizationId:
+        parsed.data.expectedOrganization === CLOUD_AGENT_SESSION_SCOPE_PERSONAL_ORGANIZATION
+          ? null
+          : parsed.data.expectedOrganization,
+      memberRootKiloSessionIds: parsed.data.memberRootKiloSessionIds,
+    },
+  };
+}

packages/session-ingest-contracts/src/index.ts

@@ -1,6 +1,7 @@
 export * from './rpc-contract';
 export * from './session-live-summary';
 export * from './cloud-agent-session-scope';
+export * from './cloud-agent-worktree-scope';
 export * from './public-worktree-projection';
 
 export const DEFAULT_SESSION_TITLE_PATTERN =

services/session-ingest/package.json

@@ -7,10 +7,11 @@
     "cf-typegen": "wrangler types --include-runtime=false --env-interface CloudflareBindings",
     "test": "vitest run",
     "test:watch": "vitest",
+    "test:postgres": "vitest run --config vitest.postgres.config.ts",
     "test:integration": "vitest run --config vitest.workers.config.ts",
     "test:integration:watch": "vitest --config vitest.workers.config.ts",
     "typecheck": "tsgo --noEmit",
-    "lint": "pnpm -w exec oxlint --config .oxlintrc.json services/session-ingest/src"
+    "lint": "pnpm -w exec oxlint --config .oxlintrc.json services/session-ingest/src services/session-ingest/test/postgres"
   },
   "dependencies": {
     "@kilocode/db": "workspace:*",

services/session-ingest/src/routes/api.ts

@@ -23,8 +23,8 @@ import { createConnectionTicket } from '../dos/connection-ticket';
 import { getSessionExport } from '../services/session-export';
 import { mapSessionEventRow, notifyUserSessionEvent } from '../session-events';
 import { handleDirectIngestRequest } from '../ingest/direct-ingest';
-import { isDefaultSessionTitle } from '../ingest/default-session-title';
 import { resolveAccessibleKiloSession } from '../services/session-access';
+import { applyReportedSessionTitle, reportSessionTitleSchema } from '../services/session-title';
 import { signSessionShareToken } from '../services/session-share-token';
 import { canCreateCliSessionForUser } from '../services/user-session-admission';
 import { isWorktreeSessionDeleting } from '../services/worktree-deletion';
@@ -820,11 +820,6 @@ api.post('/session/:sessionId/unshare', async c => {
   return c.json({ success: true }, 200);
 });
 
-const reportSessionTitleSchema = z.object({
-  title: z.string().trim().min(1).max(200),
-  generated: z.boolean(),
-});
-
 api.post('/session/:sessionId/rename-notify', async c => {
   const rawSessionId = c.req.param('sessionId');
   const parsed = sessionIdSchema.safeParse(rawSessionId);
@@ -867,66 +862,26 @@ api.post('/session/:sessionId/title', async c => {
 
   const db = getWorkerDb(c.env.HYPERDRIVE.connectionString);
   const kiloUserId = c.get('user_id');
-  const { title, generated } = bodyParse.data;
-
-  const existingRows = await db
-    .select()
-    .from(cli_sessions_v2)
-    .where(
-      and(eq(cli_sessions_v2.session_id, parsed.data), eq(cli_sessions_v2.kilo_user_id, kiloUserId))
-    )
-    .limit(1);
 
-  const existing = existingRows[0];
-  if (!existing) {
+  const result = await applyReportedSessionTitle(db, {
+    sessionId: parsed.data,
+    kiloUserId,
+    input: bodyParse.data,
+  });
+  if (result.kind === 'not_found') {
     return c.json({ success: false, error: 'session_not_found' }, 404);
   }
-
-  // generated: true applies only over NULL or default-pattern titles (Decision 8).
-  // generated: false is last-write-wins.
-  if (generated && !isDefaultSessionTitle(existing.title)) {
+  if (result.kind === 'refused') {
     return c.json({ applied: false }, 200);
   }
 
-  // Ownership filter shared by both paths. For generated writes, also CAS on the
-  // title observed at read time so a concurrent rename (or other non-default write)
-  // between select and update cannot be clobbered — Decision 8 must hold at write time.
-  const ownership = and(
-    eq(cli_sessions_v2.session_id, parsed.data),
-    eq(cli_sessions_v2.kilo_user_id, kiloUserId)
-  );
-  const writeWhere = generated
-    ? and(
-        ownership,
-        existing.title == null
-          ? isNull(cli_sessions_v2.title)
-          : eq(cli_sessions_v2.title, existing.title)
-      )
-    : ownership;
-
-  // Preserve updated_at so title-only writes do not reorder activity lists
-  // (same as the web rename path).
-  const [updatedRow] = await db
-    .update(cli_sessions_v2)
-    .set({ title, updated_at: existing.updated_at })
-    .where(writeWhere)
-    .returning();
-
-  if (!updatedRow) {
-    // Generated CAS miss (intervening rename / non-default write) → refused, not 404.
-    if (generated) {
-      return c.json({ applied: false }, 200);
-    }
-    return c.json({ success: false, error: 'session_not_found' }, 404);
-  }
-
-  const session = mapSessionEventRow(updatedRow);
+  const session = mapSessionEventRow(result.row);
   notifyUserSessionEventFromContext(c, kiloUserId, {
     type: 'session.updated',
     data: { source: 'v2', session, changedAt: session.updatedAt },
   });
 
-  return c.json({ title: session.title ?? title, applied: true }, 200);
+  return c.json({ title: session.title ?? bodyParse.data.title, applied: true }, 200);
 });
 
 api.get('/sessions/active', async c => {