Validate claimed demo actors at socket admission

apps/server/src/chat/actor.test.ts

@@ -0,0 +1,110 @@
+import { describe, expect, it } from "bun:test";
+
+import * as Chat from "./service";
+
+import type { Server } from "bun";
+import type { Chat as Wire, Request } from "@chopin/protocol";
+import type { Config } from "../config";
+import type { Plan } from "../plan/service";
+import type { Socket } from "../wire";
+import type { SocketData } from "../wire";
+
+type Sent = {
+	kind: string;
+	id?: string;
+	message?: string;
+	queued?: boolean;
+	rid?: string;
+	[key: string]: unknown;
+};
+
+function room(options: { agent?: boolean; busy?: boolean } = {}) {
+	let sent: Sent[] = [];
+	let chat = Chat.create();
+	chat.busy = options.busy ?? false;
+
+	let server = {
+		publish(_topic: string, data: string) {
+			sent.push(JSON.parse(data) as Sent);
+		},
+	} as unknown as Server<SocketData>;
+
+	let context: Chat.Room = {
+		chat,
+		config: { agent: options.agent ?? true } as Config,
+		plan: { chat } as Plan,
+		room: "test",
+		server,
+		auth: {} as Chat.Room["auth"],
+		claimantSessionId: "session",
+		repository: { id: "repo", owner: "owner", name: "repo", defaultBranch: "main" },
+		persist: async () => {},
+	};
+
+	return { chat, context, sent };
+}
+
+function sender(handle = "ana", userId = `U_${handle}`, frames: Sent[] = []): Socket {
+	return {
+		data: {
+			handle,
+			principalId: userId,
+			...(handle.startsWith("demo:") ? { actor: { id: handle } } : {}),
+		},
+		send(value: string) {
+			frames.push(JSON.parse(value) as Sent);
+		},
+	} as unknown as Socket;
+}
+
+function message(
+	text: string,
+	to: Wire.Destination,
+	references?: Wire.ReferenceRequest[],
+	rid = "request",
+	requestId: string = crypto.randomUUID(),
+): Request<Wire.Send> {
+	return {
+		kind: "chat:send",
+		rid,
+		requestId,
+		text,
+		to,
+		ts: 0,
+		...(references ? { references } : {}),
+	};
+}
+
+describe("demo actors sharing one operator", () => {
+	it("keeps retries and withdrawal apart per actor", async () => {
+		let { chat, context } = room({ busy: true });
+		let lily = sender("demo:lily", "U_op");
+		let felix = sender("demo:felix", "U_op");
+		let requestId = crypto.randomUUID();
+
+		await Chat.send(context, lily, message("draft it", "planner", undefined, "a", requestId));
+		await Chat.send(context, lily, message("draft it", "planner", undefined, "b", requestId));
+		expect(chat.waiting).toHaveLength(1);
+
+		let replies: Sent[] = [];
+		let rival = sender("demo:felix", "U_op", replies);
+		await Chat.send(context, rival, message("draft it", "planner", undefined, "c", requestId));
+		expect(replies[0]).toMatchObject({
+			kind: "session:error",
+			message: "chat request id was reused with different content",
+		});
+		expect(chat.waiting).toHaveLength(1);
+
+		Chat.unqueue(context, felix, { kind: "chat:unqueue", rid: "u", id: requestId, ts: 0 });
+		expect(chat.waiting).toHaveLength(1);
+		Chat.unqueue(context, lily, { kind: "chat:unqueue", rid: "u", id: requestId, ts: 0 });
+		expect(chat.waiting).toHaveLength(0);
+	});
+
+	it("authors room messages as the admitted actor", async () => {
+		let { chat, context } = room();
+		let msg = { ...message("hello", "room"), author: { kind: "member", handle: "mallory" } };
+		await Chat.send(context, sender("demo:lily", "U_op"), msg);
+		expect(chat.entries[0]).toMatchObject({ author: { kind: "member", handle: "demo:lily" } });
+	});
+});

apps/server/src/chat/service.ts

@@ -373,7 +373,11 @@ function digest(value: string): string {
 	return `sha256:${createHash("sha256").update(value).digest("hex")}`;
 }
 
-function requestFingerprint(msg: Request<Wire.Send>, principalId: string): string {
+function requestFingerprint(
+	msg: Request<Wire.Send>,
+	principalId: string,
+	actor?: string,
+): string {
 	let references = Array.isArray(msg.references)
 		? msg.references.map(value => {
 			if (!value || typeof value !== "object" || Array.isArray(value)) return value;
@@ -389,6 +393,7 @@ function requestFingerprint(msg: Request<Wire.Send>, principalId: string): strin
 		msg.text,
 		references,
 		...(focus !== undefined ? [focus] : []),
+		...(actor ? [{ actor }] : []),
 	]));
 }
 
@@ -769,7 +774,7 @@ async function processSend(context: Room, ws: Socket, msg: Request<Wire.Send>):
 	if (Buffer.byteLength(msg.text) > MAX_MESSAGE_BYTES) {
 		return fail(ws, msg.rid, "chat message exceeds the 64 KiB limit");
 	}
-	let request = requestFingerprint(msg, ws.data.principalId);
+	let request = requestFingerprint(msg, ws.data.principalId, ws.data.actor?.id);
 	if (replay(chat, ws, msg, requestId, request)) return;
 	if (chat.closed) return fail(ws, msg.rid, "chat is closed");
 	let focus: Focus | undefined;

apps/server/src/main.ts

@@ -1377,7 +1377,13 @@ function openRoomSocket(ws: Socket): void {
 		...(ws.data.channelDescription
 			? { description: ws.data.channelDescription }
 			: {}),
-		you: { handle: ws.data.handle, client: ws.data.client },
+		you: {
+			handle: ws.data.handle,
+			client: ws.data.client,
+			...(ws.data.actor
+				? { actor: { name: ws.data.actor.name, avatar: ws.data.actor.avatar } }
+				: {}),
+		},
 		members: Rooms.members(room),
 		canEdit: ws.data.canEdit,
 		canManage: ws.data.canManage,

apps/server/src/research/service.ts

@@ -183,7 +183,7 @@ const MAX_EVIDENCE_SOURCES = 64;
 const MAX_TARGET_SCAN = 1_000;
 const REQUEST_ID = /^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i;
 const SAFE_ID = /^[A-Za-z0-9][A-Za-z0-9._:-]*$/;
-const HANDLE = /^[A-Za-z0-9](?:[A-Za-z0-9-]{0,38})$/;
+const HANDLE = /^(?:[A-Za-z0-9](?:[A-Za-z0-9-]{0,38})|demo:[a-z0-9-]{1,32})$/;
 const ACTIVE_JOB_STATES = new Set(["pending", "paused", "running"]);
 
 function safeId(value: unknown, field: string, maximum = MAX_ID): string {

apps/server/src/rooms.ts

@@ -9,7 +9,8 @@
  * and the agent session attach to the same record as they arrive.
  */
 
-import type { Identity, Socket } from "./wire";
+import type { Session } from "@chopin/protocol";
+import type { Socket } from "./wire";
 // Type-only: the plan attaches to a room, but a room knows nothing about how
 // one is built, and a value import here would close a cycle.
 import type { Plan } from "./plan/service";
@@ -96,8 +97,14 @@ export function forget(room: Room): void {
 	if (rooms.get(room.id) === room) rooms.delete(room.id);
 }
 
-export function members(room: Room): Identity[] {
+export function members(room: Room): Session.Member[] {
 	return [...room.members.values()]
-		.map(ws => ({ handle: ws.data.handle, client: ws.data.client }))
+		.map(ws => ({
+			handle: ws.data.handle,
+			client: ws.data.client,
+			...(ws.data.actor
+				? { actor: { name: ws.data.actor.name, avatar: ws.data.actor.avatar } }
+				: {}),
+		}))
 		.sort((a, b) => a.handle.localeCompare(b.handle) || a.client.localeCompare(b.client));
 }

apps/server/src/socket/admission-actor.test.ts

@@ -0,0 +1,176 @@
+import { Machines } from "../auth/machines";
+import { describe, expect, it } from "bun:test";
+
+import { Sessions } from "../auth/session";
+import { Admission } from "../auth/admission";
+import { GitHubError } from "../github/client";
+import { MemoryStorage } from "../storage/memory/adapter";
+import { admit } from "./admission";
+
+import type { HostedAuth } from "../auth/routes";
+import type {
+	GitHub,
+	GitHubTokenGrant,
+	GitHubUser,
+	InstallationPage,
+	Repository,
+	RepositoryPage,
+} from "../github/client";
+
+class FakeGitHub implements GitHub {
+	repositoryId = "R_score";
+	push = false;
+	failure: number | undefined;
+
+	authorize(): string {
+		return "";
+	}
+
+	async exchange(): Promise<GitHubTokenGrant> {
+		return grant("ghu_user");
+	}
+
+	async refresh(): Promise<GitHubTokenGrant> {
+		return grant("ghu_refreshed");
+	}
+
+	async user(): Promise<GitHubUser> {
+		return { id: "U_octocat", login: "octocat", avatarUrl: "avatar" };
+	}
+
+	async organizationMembership() {
+		return undefined;
+	}
+
+	async installations(): Promise<InstallationPage> {
+		return { installations: [], nextPage: undefined };
+	}
+
+	async installationRepositories(): Promise<RepositoryPage> {
+		return { repositories: [], nextPage: undefined };
+	}
+
+	async repository(_token: string, owner: string, name: string): Promise<Repository> {
+		return {
+			id: this.repositoryId,
+			owner,
+			name,
+			fullName: `${owner}/${name}`,
+			private: true,
+			url: "",
+			defaultBranch: "main",
+			permissions: { pull: true, push: this.push, admin: false },
+		};
+	}
+
+	async repositoryAccess(token: string, owner: string, name: string): Promise<Repository> {
+		if (this.failure) throw new GitHubError("unavailable", this.failure);
+		return this.repository(token, owner, name);
+	}
+
+	invalidate(): void {}
+}
+
+function grant(accessToken: string): GitHubTokenGrant {
+	return {
+		accessToken,
+		accessExpiresIn: 28_800,
+		refreshToken: "ghr_user",
+		refreshExpiresIn: 15_897_600,
+	};
+}
+
+function pair(cookie: string): string {
+	return cookie.split(";", 1)[0]!;
+}
+
+async function setup(demoPersonas: boolean) {
+	let now = new Date("2026-08-13T12:00:00.000Z");
+	let storage = new MemoryStorage();
+	await storage.users.put({ id: "U_octocat", login: "octocat", avatarUrl: "avatar", now });
+	let sessions = new Sessions(storage, true, () => now);
+	let issued = await sessions.issue("U_octocat", grant("ghu_user"));
+	let channel = await storage.channels.create({
+		id: crypto.randomUUID(),
+		repositoryId: "R_score",
+		repositoryOwner: "octo-org",
+		repositoryName: "score",
+		title: "Plan",
+		createdBy: "U_octocat",
+		now,
+	});
+	let github = new FakeGitHub();
+	let config = {
+		origin: "https://chopin.test",
+		appSlug: "chopin-test",
+		clientId: "client",
+		clientSecret: "secret",
+		encryptionKey: new Uint8Array(32).fill(2),
+		...(demoPersonas ? { demoPersonas: true as const } : {}),
+	};
+	let auth: HostedAuth = {
+		config,
+		storage,
+		github,
+		admission: new Admission(config, github, () => now.getTime()),
+		sessions,
+		clock: () => now,
+		machines: new Machines(storage, () => new Date(), {
+			origin: "https://chopin.test",
+			clientId: "test",
+		}),
+		pairings: {},
+	};
+	let connect = (actor?: string[]) => {
+		let url = new URL(`https://chopin.test/ws?channel=${channel.id}`);
+		for (let value of actor ?? []) url.searchParams.append("actor", value);
+		let request = new Request(url, {
+			headers: { cookie: pair(issued.cookie), origin: "https://chopin.test" },
+		});
+		return admit(request, url, auth);
+	};
+	return { github, connect };
+}
+
+describe("socket admission of demo actors", () => {
+	it("binds a rostered actor while the operator keeps principal and access", async () => {
+		let { github, connect } = await setup(true);
+		let plain = await connect();
+		expect("data" in plain && plain.data.actor).toBeUndefined();
+		let first = await connect(["demo:lily"]);
+		expect("data" in first && first.data).toMatchObject({
+			handle: "demo:lily",
+			actor: { id: "demo:lily", name: "Lily" },
+			principalId: "U_octocat",
+			canEdit: false,
+		});
+		github.push = true;
+		let reconnect = await connect(["demo:lily"]);
+		expect("data" in reconnect && reconnect.data).toMatchObject({
+			handle: "demo:lily",
+			principalId: "U_octocat",
+			canEdit: true,
+			canManage: true,
+		});
+		github.repositoryId = "R_other";
+		expect(await connect(["demo:lily"])).toEqual({ status: 404, reason: "channel not found" });
+	});
+
+	it("rejects unknown and repeated actors on first connection and reconnect", async () => {
+		let { connect } = await setup(true);
+		let refusal = { status: 403, reason: "actor is not allowed" };
+		expect(await connect(["demo:nobody"])).toEqual(refusal);
+		expect(await connect(["octocat"])).toEqual(refusal);
+		expect(await connect([""])).toEqual(refusal);
+		expect(await connect(["demo:lily", "demo:felix"])).toEqual(refusal);
+		expect("data" in await connect(["demo:felix"])).toBe(true);
+		expect(await connect(["demo:nobody"])).toEqual(refusal);
+	});
+
+	it("rejects any actor while the flag is off", async () => {
+		let { connect } = await setup(false);
+		expect(await connect(["demo:lily"])).toEqual({ status: 403, reason: "actor is not allowed" });
+		let plain = await connect();
+		expect("data" in plain && plain.data.handle).toBe("octocat");
+	});
+});

apps/server/src/socket/admission.ts

@@ -2,6 +2,7 @@ import { isChannelId } from "../channels/id";
 import { GitHubError } from "../github/client";
 import { uid } from "../ids";
 import { StorageError } from "../storage/errors";
+import { demoRoster } from "../demo/roster";
 import { decisionWatch } from "./decision-watch";
 
 import type { HostedAuth } from "../auth/routes";
@@ -58,6 +59,12 @@ export async function admit(
 		}
 		let session = await auth.sessions.authenticate(request);
 		if (!session) return { status: 401, reason: "authentication required" };
+		let claimed = url.searchParams.getAll("actor");
+		let actor;
+		if (claimed.length) {
+			actor = claimed.length === 1 ? demoRoster(auth.config).get(claimed[0]!) : undefined;
+			if (!actor) return { status: 403, reason: "actor is not allowed" };
+		}
 		let id = (url.searchParams.get("channel") || "").toLowerCase();
 		if (!isChannelId(id)) return { status: 400, reason: "bad channel" };
 		if (options.deleting?.(id)) return { status: 404, reason: "channel not found" };
@@ -93,7 +100,8 @@ export async function admit(
 				...(channel.archivedAt
 					? { channelArchivedAt: channel.archivedAt.toISOString() }
 					: {}),
-				handle: session.user.login,
+				handle: actor ? actor.id : session.user.login,
+				...(actor ? { actor } : {}),
 				client: uid(),
 				canEdit: canManage && !channel.archivedAt && !options.readOnly?.(channel.id),
 				canManage,