diff --git a/server/README.md b/server/README.md
index d833e8b..23ed2aa 100644
--- a/server/README.md
+++ b/server/README.md
@@ -31,6 +31,10 @@ without breaking the typecheck, which is the wrong order to find out.
| `GET /api/v1/markers` | — | `MarkersBody` | **private; 401 unless signed in** — public empty body when `TERA_MARKERS_SOURCE=none` |
| `GET /api/v1/offices/:id` | — | `OfficeDoc` | public offices only |
| `GET /api/v1/offices/:id/presence` | — | `PresenceBody` | **private; 401 unless signed in** — never publicly cached |
+| `POST /api/v1/realtime/join` | — | strict `JoinRequest` or `ResumeRequest` | **private; never cached** |
+| `POST /api/v1/realtime/events` | — | fetch-streamed SSE, opened with a strict `ResumeRequest` | **private; never cached** |
+| `POST /api/v1/realtime/pose` | — | one owned `EntityPoseSnapshot` | **private; never cached** |
+| `POST /api/v1/realtime/leave` | — | session id + opaque resume token | **private; never cached** |
Every body is declared once, in `src/server/wire.ts` in the **root** package —
type-only, so it compiles to nothing and both the browser build and this service
@@ -48,6 +52,29 @@ reply before any route runs, and a route opts in explicitly. A request that
arrived with an `Authorization` header or a cookie never gets a public policy,
whatever the route asked for.
+### Realtime sessions
+
+Hosted presence is an optional, in-memory adapter. It stores no pose history on
+disk and disappears cleanly on restart; the static city and office continue to
+work when it is absent. Joining requires the deployment's existing auth, and a
+session owns exactly one opaque actor plus a bounded list of interest cells.
+The server validates entity ownership, monotonic sequence/time, cell membership,
+speed, climb and turn rate before broadcasting a pose only to the same cell.
+
+The event stream is deliberately a **POST consumed with `fetch()`**, not an
+`EventSource` URL. Session/resume credentials stay in the request body and
+therefore out of browser history, referrers, Caddy access logs and copied URLs.
+Tokens are random, stored only as SHA-256 hashes, rotated on resume, and bounded
+by absolute TTL and disconnect grace. Reconnect, room capacity and pose rate
+are capped. Authentication subjects remain server-only; peers see page-scoped
+opaque entity ids and no profile name, face texture, email, or media locator.
+
+California, city, office, floor and room cells are distinct coordinate and
+authorization boundaries. Disk-backed office packs are checked before a floor
+or room may be joined. The three public demo offices bundled into this repo may
+use their office envelope without a duplicate server pack; that exception does
+not invent floor/room access and does not apply to arbitrary tenant ids.
+
## Regions
**A caller's coordinate is never forwarded upstream. It only selects among the
diff --git a/server/src/app.ts b/server/src/app.ts
index d1b1d47..9b54ae6 100644
--- a/server/src/app.ts
+++ b/server/src/app.ts
@@ -20,6 +20,7 @@ import { registerHealth } from "./routes/health.ts";
import { registerMarkers } from "./routes/markers.ts";
import { registerOffices } from "./routes/offices.ts";
import { registerPresence } from "./routes/presence.ts";
+import { registerRealtime } from "./routes/realtime.ts";
import { registerSatellites } from "./routes/satellites.ts";
import { registerSession } from "./routes/session.ts";
import { registerWeather } from "./routes/weather.ts";
@@ -51,6 +52,7 @@ export function buildApp(config: Config = loadConfig()): FastifyInstance {
registerMarkers(app, services);
registerOffices(app, services);
registerPresence(app, services);
+ registerRealtime(app, services);
registerSession(app, services);
app.setNotFoundHandler(async (_req, reply) => {
@@ -86,7 +88,7 @@ function registerCors(app: FastifyInstance, allowed: string[]): void {
reply.header("access-control-allow-origin", origin);
reply.header("access-control-allow-headers", "authorization, content-type");
- reply.header("access-control-allow-methods", "GET, OPTIONS");
+ reply.header("access-control-allow-methods", "GET, POST, OPTIONS");
// Whatever the response ends up being, it depended on the Origin header.
reply.header("vary", "Origin");
diff --git a/server/src/realtime/index.ts b/server/src/realtime/index.ts
new file mode 100644
index 0000000..176c6c4
--- /dev/null
+++ b/server/src/realtime/index.ts
@@ -0,0 +1,13 @@
+export {
+ createRealtimeService,
+ type RealtimeConnection,
+ type RealtimeFailureCode,
+ type RealtimeJoinInput,
+ type RealtimeJoinResult,
+ type RealtimeListener,
+ type RealtimeResult,
+ type RealtimeResumeConnection,
+ type RealtimeService,
+ type RealtimeServiceOptions,
+ type RealtimeSessionView,
+} from "./service.ts";
diff --git a/server/src/realtime/service.ts b/server/src/realtime/service.ts
new file mode 100644
index 0000000..9fc149a
--- /dev/null
+++ b/server/src/realtime/service.ts
@@ -0,0 +1,664 @@
+/**
+ * Dependency-free, in-memory authoritative realtime sessions.
+ *
+ * The service owns credentials, interest filtering, sequence allocation and
+ * movement validation. HTTP/SSE routes are adapters only. No poses survive a
+ * process restart and usable bearer tokens are never retained in memory.
+ */
+
+import { createHash, randomBytes, timingSafeEqual } from "node:crypto";
+import {
+ isEntityPoseSnapshot,
+ isInterestCell,
+ validateMotion,
+} from "../../../src/realtime/protocol.ts";
+import type {
+ EntityPoseSnapshot,
+ InterestCell,
+ JoinGrant,
+ MotionLimits,
+ PoseDelta,
+ RealtimeMemberRole,
+ ResumeGrant,
+ ResumeRequest,
+ ServerRealtimeMessage,
+} from "../../../src/realtime/types.ts";
+
+const UINT32_MAX = 4_294_967_295;
+
+export interface RealtimeServiceOptions {
+ now?: () => number;
+ sessionTtlMs?: number;
+ disconnectGraceMs?: number;
+ maximumSessions?: number;
+ maximumSessionsPerCell?: number;
+ maximumInterestsPerSession?: number;
+ maximumPoseUpdatesPerWindow?: number;
+ rateWindowMs?: number;
+ actorMotionLimits?: MotionLimits;
+ vehicleMotionLimits?: MotionLimits;
+}
+
+export interface RealtimeJoinInput {
+ requestId: string;
+ actorId: string;
+ subject: string;
+ role: RealtimeMemberRole;
+ interests: readonly InterestCell[];
+}
+
+export interface RealtimeJoinResult {
+ sessionId: string;
+ token: string;
+ grant: JoinGrant;
+}
+
+export type RealtimeFailureCode =
+ | "invalid"
+ | "unauthorized"
+ | "expired"
+ | "capacity"
+ | "conflict"
+ | "rate-limited"
+ | "sequence"
+ | "interest"
+ | "ownership"
+ | "motion";
+
+export type RealtimeResult
=
+ | { ok: true; value: T }
+ | { ok: false; code: RealtimeFailureCode; message: string };
+
+export type RealtimeListener = (message: ServerRealtimeMessage) => void;
+
+interface Session {
+ id: string;
+ tokenHash: Buffer;
+ previousTokenHash: Buffer | null;
+ previousTokenExpiresAt: number;
+ subject: string;
+ actorId: string;
+ role: RealtimeMemberRole;
+ interests: readonly InterestCell[];
+ interestKeys: ReadonlySet;
+ createdAt: number;
+ expiresAt: number;
+ disconnectedAt: number | null;
+ lastClientSequence: number;
+ rateWindowStartedAt: number;
+ rateCount: number;
+ proposedPoses: Map;
+ poses: Map;
+ listeners: Set;
+}
+
+interface ResolvedOptions {
+ now: () => number;
+ sessionTtlMs: number;
+ disconnectGraceMs: number;
+ maximumSessions: number;
+ maximumSessionsPerCell: number;
+ maximumInterestsPerSession: number;
+ maximumPoseUpdatesPerWindow: number;
+ rateWindowMs: number;
+ actorMotionLimits: MotionLimits;
+ vehicleMotionLimits: MotionLimits;
+}
+
+export interface RealtimeService {
+ join(input: RealtimeJoinInput): RealtimeResult;
+ authenticate(sessionId: string, token: string): RealtimeResult;
+ connect(sessionId: string, token: string, listener: RealtimeListener): RealtimeResult;
+ resume(request: ResumeRequest, subject: string): RealtimeResult;
+ connectResume(
+ request: ResumeRequest,
+ subject: string,
+ listener: RealtimeListener,
+ ): RealtimeResult;
+ submitPose(
+ sessionId: string,
+ token: string,
+ snapshot: unknown,
+ ): RealtimeResult;
+ leave(sessionId: string, token: string): RealtimeResult;
+ revokeSubject(subject: string): number;
+ cleanup(): number;
+ sessionCount(): number;
+}
+
+export interface RealtimeSessionView {
+ sessionId: string;
+ actorId: string;
+ role: RealtimeMemberRole;
+ interests: readonly InterestCell[];
+ expiresAt: number;
+ snapshot: readonly EntityPoseSnapshot[];
+}
+
+export interface RealtimeConnection {
+ view: RealtimeSessionView;
+ disconnect(): void;
+}
+
+export interface RealtimeResumeConnection extends RealtimeConnection {
+ grant: ResumeGrant;
+}
+
+function finitePositive(value: number | undefined, fallback: number): number {
+ return typeof value === "number" && Number.isFinite(value) && value > 0 ? value : fallback;
+}
+
+function finiteCount(value: number | undefined, fallback: number): number {
+ return Math.max(1, Math.floor(finitePositive(value, fallback)));
+}
+
+function resolveOptions(value: RealtimeServiceOptions): ResolvedOptions {
+ return {
+ now: value.now ?? Date.now,
+ sessionTtlMs: finitePositive(value.sessionTtlMs, 5 * 60_000),
+ disconnectGraceMs: finitePositive(value.disconnectGraceMs, 20_000),
+ maximumSessions: finiteCount(value.maximumSessions, 1_024),
+ maximumSessionsPerCell: finiteCount(value.maximumSessionsPerCell, 96),
+ maximumInterestsPerSession: finiteCount(value.maximumInterestsPerSession, 16),
+ maximumPoseUpdatesPerWindow: finiteCount(value.maximumPoseUpdatesPerWindow, 120),
+ rateWindowMs: finitePositive(value.rateWindowMs, 10_000),
+ actorMotionLimits: value.actorMotionLimits ?? {
+ maximumHorizontalSpeedMps: 24,
+ maximumVerticalSpeedMps: 18,
+ maximumTurnRateDegPerSec: 540,
+ positionSlackM: 1.5,
+ maximumDeltaSeconds: 2,
+ },
+ vehicleMotionLimits: value.vehicleMotionLimits ?? {
+ maximumHorizontalSpeedMps: 105,
+ maximumVerticalSpeedMps: 45,
+ maximumTurnRateDegPerSec: 240,
+ positionSlackM: 4,
+ maximumDeltaSeconds: 2,
+ },
+ };
+}
+
+function hashToken(token: string): Buffer {
+ return createHash("sha256").update(token).digest();
+}
+
+function tokenMatches(expected: Buffer, token: string): boolean {
+ if (typeof token !== "string" || token.length < 16 || token.length > 512) return false;
+ const found = hashToken(token);
+ return found.length === expected.length && timingSafeEqual(found, expected);
+}
+
+function interestKey(cell: InterestCell): string {
+ switch (cell.kind) {
+ case "california-tile": return `tile:${cell.level}:${cell.x}:${cell.y}`;
+ case "city": return `city:${cell.cityId}`;
+ case "office": return `office:${cell.officeId}`;
+ case "floor": return `floor:${cell.officeId}:${cell.floorId}`;
+ case "room": return `room:${cell.officeId}:${cell.floorId}:${cell.roomId}`;
+ }
+}
+
+function cloneCell(cell: InterestCell): InterestCell {
+ return { ...cell };
+}
+
+function clonePose(pose: EntityPoseSnapshot): EntityPoseSnapshot {
+ return JSON.parse(JSON.stringify(pose)) as EntityPoseSnapshot;
+}
+
+function entityKey(pose: EntityPoseSnapshot): string {
+ return pose.entity === "actor" ? `actor:${pose.actorId}` : `vehicle:${pose.vehicleId}`;
+}
+
+function poseInterestKey(pose: EntityPoseSnapshot): string | null {
+ if (pose.pose.space === "local") return interestKey(pose.pose.cell);
+ // Geographic coordinates are interpreted through the session's authored
+ // outdoor interest: either one exact California tile or one exact city.
+ return null;
+}
+
+function validIdentity(value: string): boolean {
+ return typeof value === "string" && value.length > 0 && value.length <= 256;
+}
+
+export function createRealtimeService(options: RealtimeServiceOptions = {}): RealtimeService {
+ const settings = resolveOptions(options);
+ const sessions = new Map();
+ const epoch = randomBytes(16).toString("base64url");
+ let serverSequence = 0;
+
+ function nextSequence(): number {
+ serverSequence = serverSequence >= UINT32_MAX ? 0 : serverSequence + 1;
+ return serverSequence;
+ }
+
+ function geographicInterestKeys(session: Session): string[] {
+ return [...session.interestKeys].filter(
+ (key) => key.startsWith("tile:") || key.startsWith("city:"),
+ );
+ }
+
+ function poseBroadcastKeys(session: Session, pose: EntityPoseSnapshot): string[] {
+ const local = poseInterestKey(pose);
+ return local === null ? geographicInterestKeys(session) : [local];
+ }
+
+ function broadcast(delta: PoseDelta, keys: readonly string[], excluding?: Session): void {
+ if (keys.length === 0) return;
+ for (const peer of sessions.values()) {
+ if (peer === excluding || !keys.some((key) => peer.interestKeys.has(key))) continue;
+ for (const listener of peer.listeners) listener(delta);
+ }
+ }
+
+ function removalDelta(session: Session): PoseDelta | null {
+ const removedEntityIds = [...session.poses.keys()];
+ if (removedEntityIds.length === 0) return null;
+ return {
+ type: "pose-delta",
+ protocolVersion: 1,
+ serverEpoch: epoch,
+ sequence: nextSequence(),
+ timestampMs: settings.now(),
+ updates: [],
+ removedEntityIds,
+ };
+ }
+
+ function broadcastRemoval(session: Session, keys = [...session.interestKeys]): void {
+ const delta = removalDelta(session);
+ if (delta) broadcast(delta, keys, session);
+ }
+
+ function remove(session: Session, announce = true): void {
+ if (announce) broadcastRemoval(session);
+ sessions.delete(session.id);
+ session.listeners.clear();
+ session.proposedPoses.clear();
+ session.poses.clear();
+ session.tokenHash.fill(0);
+ session.previousTokenHash?.fill(0);
+ }
+
+ function cleanup(): number {
+ const now = settings.now();
+ let removed = 0;
+ for (const session of sessions.values()) {
+ const graceExpired =
+ session.disconnectedAt !== null && now - session.disconnectedAt > settings.disconnectGraceMs;
+ if (now >= session.expiresAt || graceExpired) {
+ remove(session);
+ removed += 1;
+ }
+ }
+ return removed;
+ }
+
+ function authenticate(
+ sessionId: string,
+ token: string,
+ allowPrevious = false,
+ ): RealtimeResult {
+ cleanup();
+ if (!validIdentity(sessionId)) {
+ return { ok: false, code: "unauthorized", message: "Realtime session is invalid." };
+ }
+ const session = sessions.get(sessionId);
+ const matchesCurrent = session ? tokenMatches(session.tokenHash, token) : false;
+ const matchesPrevious = allowPrevious && session?.previousTokenHash !== null &&
+ settings.now() <= (session?.previousTokenExpiresAt ?? 0) &&
+ tokenMatches(session?.previousTokenHash ?? Buffer.alloc(0), token);
+ if (!session || (!matchesCurrent && !matchesPrevious)) {
+ return { ok: false, code: "unauthorized", message: "Realtime session is invalid." };
+ }
+ if (settings.now() >= session.expiresAt) {
+ remove(session);
+ return { ok: false, code: "expired", message: "Realtime session expired." };
+ }
+ return { ok: true, value: session };
+ }
+
+ function view(session: Session): RealtimeSessionView {
+ const snapshot: EntityPoseSnapshot[] = [];
+ for (const peer of sessions.values()) {
+ for (const pose of peer.poses.values()) {
+ const visible = poseBroadcastKeys(peer, pose).some((key) => session.interestKeys.has(key));
+ if (visible) snapshot.push(clonePose(pose));
+ }
+ }
+ return {
+ sessionId: session.id,
+ actorId: session.actorId,
+ role: session.role,
+ interests: session.interests.map(cloneCell),
+ expiresAt: session.expiresAt,
+ snapshot,
+ };
+ }
+
+ function join(input: RealtimeJoinInput): RealtimeResult {
+ cleanup();
+ if (
+ !validIdentity(input.requestId) || !validIdentity(input.actorId) ||
+ !validIdentity(input.subject) ||
+ (input.role !== "member" && input.role !== "admin") ||
+ !Array.isArray(input.interests) || input.interests.length === 0 ||
+ input.interests.length > settings.maximumInterestsPerSession ||
+ !input.interests.every(isInterestCell)
+ ) return { ok: false, code: "invalid", message: "Realtime join request is invalid." };
+ if (sessions.size >= settings.maximumSessions) {
+ return { ok: false, code: "capacity", message: "Realtime service is at capacity." };
+ }
+ if ([...sessions.values()].some((session) => session.actorId === input.actorId)) {
+ return { ok: false, code: "conflict", message: "Realtime actor id is already active." };
+ }
+ const unique = new Map(input.interests.map((cell) => [interestKey(cell), cloneCell(cell)]));
+ for (const key of unique.keys()) {
+ let count = 0;
+ for (const session of sessions.values()) if (session.interestKeys.has(key)) count += 1;
+ if (count >= settings.maximumSessionsPerCell) {
+ return { ok: false, code: "capacity", message: "Realtime interest cell is at capacity." };
+ }
+ }
+ const now = settings.now();
+ const id = randomBytes(18).toString("base64url");
+ const token = randomBytes(32).toString("base64url");
+ const interests = [...unique.values()];
+ const session: Session = {
+ id,
+ tokenHash: hashToken(token),
+ previousTokenHash: null,
+ previousTokenExpiresAt: 0,
+ subject: input.subject,
+ actorId: input.actorId,
+ role: input.role,
+ interests,
+ interestKeys: new Set(unique.keys()),
+ createdAt: now,
+ expiresAt: now + settings.sessionTtlMs,
+ disconnectedAt: null,
+ lastClientSequence: -1,
+ rateWindowStartedAt: now,
+ rateCount: 0,
+ proposedPoses: new Map(),
+ poses: new Map(),
+ listeners: new Set(),
+ };
+ sessions.set(id, session);
+ const grant: JoinGrant = {
+ type: "join-grant",
+ protocolVersion: 1,
+ requestId: input.requestId,
+ sessionId: id,
+ actorId: input.actorId,
+ role: input.role,
+ serverEpoch: epoch,
+ serverTimeMs: now,
+ nextSequence: serverSequence >= UINT32_MAX ? 0 : serverSequence + 1,
+ resumeToken: token,
+ interests: interests.map(cloneCell),
+ initial: view(session).snapshot,
+ };
+ return { ok: true, value: { sessionId: id, token, grant } };
+ }
+
+ function connect(
+ sessionId: string,
+ token: string,
+ listener: RealtimeListener,
+ ): RealtimeResult {
+ if (typeof listener !== "function") {
+ return { ok: false, code: "invalid", message: "Realtime listener is invalid." };
+ }
+ const found = authenticate(sessionId, token);
+ if (!found.ok) return found;
+ const session = found.value;
+ session.disconnectedAt = null;
+ session.listeners.add(listener);
+ let connected = true;
+ return {
+ ok: true,
+ value: {
+ view: view(session),
+ disconnect() {
+ if (!connected) return;
+ connected = false;
+ session.listeners.delete(listener);
+ if (session.listeners.size === 0) session.disconnectedAt = settings.now();
+ },
+ },
+ };
+ }
+
+ function updateInterests(
+ session: Session,
+ requested: readonly InterestCell[],
+ ): RealtimeResult {
+ if (
+ !Array.isArray(requested) || requested.length === 0 ||
+ requested.length > settings.maximumInterestsPerSession || !requested.every(isInterestCell)
+ ) return { ok: false, code: "invalid", message: "Realtime interests are invalid." };
+ const unique = new Map(requested.map((cell) => [interestKey(cell), cloneCell(cell)]));
+ for (const key of unique.keys()) {
+ let count = 0;
+ for (const peer of sessions.values()) {
+ if (peer.id !== session.id && peer.interestKeys.has(key)) count += 1;
+ }
+ if (count >= settings.maximumSessionsPerCell) {
+ return { ok: false, code: "capacity", message: "Realtime interest cell is at capacity." };
+ }
+ }
+ const changed = unique.size !== session.interestKeys.size ||
+ [...unique.keys()].some((key) => !session.interestKeys.has(key));
+ const previousKeys = [...session.interestKeys];
+ if (changed) broadcastRemoval(session, previousKeys);
+ session.interests = [...unique.values()];
+ session.interestKeys = new Set(unique.keys());
+ if (changed) {
+ // A pose from the old cell cannot become the first authoritative point in
+ // a new cell's speed check or leak through its next snapshot.
+ session.poses.clear();
+ session.proposedPoses.clear();
+ }
+ return { ok: true, value: session.interests };
+ }
+
+ function rotateToken(session: Session): string {
+ const token = randomBytes(32).toString("base64url");
+ session.previousTokenHash?.fill(0);
+ session.previousTokenHash = session.tokenHash;
+ session.previousTokenExpiresAt = settings.now() + 10_000;
+ session.tokenHash = hashToken(token);
+ return token;
+ }
+
+ function resume(
+ request: ResumeRequest,
+ subject: string,
+ ): RealtimeResult {
+ // The immediately previous token is accepted only for resume. This small
+ // overlap lets a client retry when the rotated grant was lost in transit,
+ // without letting an old token continue publishing or leaving a session.
+ const found = authenticate(request.sessionId, request.resumeToken, true);
+ if (!found.ok) return found;
+ const session = found.value;
+ if (session.subject !== subject) {
+ return { ok: false, code: "unauthorized", message: "Realtime session is invalid." };
+ }
+ if (request.serverEpoch !== epoch || request.lastReceivedSequence > serverSequence) {
+ return { ok: false, code: "sequence", message: "Realtime resume point is invalid." };
+ }
+ const updated = updateInterests(session, request.interests);
+ if (!updated.ok) return updated;
+ const continuous = request.lastReceivedSequence === serverSequence;
+ const token = rotateToken(session);
+ return {
+ ok: true,
+ value: {
+ type: "resume-grant",
+ protocolVersion: 1,
+ requestId: request.requestId,
+ sessionId: session.id,
+ serverEpoch: epoch,
+ serverTimeMs: settings.now(),
+ nextSequence: serverSequence >= UINT32_MAX ? 0 : serverSequence + 1,
+ resumeToken: token,
+ continuous,
+ snapshot: view(session).snapshot,
+ },
+ };
+ }
+
+ function connectResume(
+ request: ResumeRequest,
+ subject: string,
+ listener: RealtimeListener,
+ ): RealtimeResult {
+ if (typeof listener !== "function") {
+ return { ok: false, code: "invalid", message: "Realtime listener is invalid." };
+ }
+ const resumed = resume(request, subject);
+ if (!resumed.ok) return resumed;
+ const session = sessions.get(request.sessionId);
+ if (!session) return { ok: false, code: "expired", message: "Realtime session expired." };
+ session.disconnectedAt = null;
+ session.listeners.add(listener);
+ let connected = true;
+ return {
+ ok: true,
+ value: {
+ grant: resumed.value,
+ view: view(session),
+ disconnect() {
+ if (!connected) return;
+ connected = false;
+ session.listeners.delete(listener);
+ if (session.listeners.size === 0) session.disconnectedAt = settings.now();
+ },
+ },
+ };
+ }
+
+ function submitPose(
+ sessionId: string,
+ token: string,
+ input: unknown,
+ ): RealtimeResult {
+ const found = authenticate(sessionId, token);
+ if (!found.ok) return found;
+ const session = found.value;
+ const now = settings.now();
+ if (now - session.rateWindowStartedAt >= settings.rateWindowMs) {
+ session.rateWindowStartedAt = now;
+ session.rateCount = 0;
+ }
+ session.rateCount += 1;
+ if (session.rateCount > settings.maximumPoseUpdatesPerWindow) {
+ return { ok: false, code: "rate-limited", message: "Realtime pose rate exceeded." };
+ }
+ if (!isEntityPoseSnapshot(input)) {
+ return { ok: false, code: "invalid", message: "Realtime pose is invalid." };
+ }
+ const snapshot = input;
+ if (snapshot.sequence <= session.lastClientSequence) {
+ return { ok: false, code: "sequence", message: "Realtime pose sequence is stale." };
+ }
+ if (snapshot.timestampMs > now + 5_000 || snapshot.timestampMs < now - 10_000) {
+ return { ok: false, code: "sequence", message: "Realtime pose timestamp is stale." };
+ }
+ if (
+ (snapshot.entity === "actor" && snapshot.actorId !== session.actorId) ||
+ (snapshot.entity === "vehicle" && snapshot.driverActorId !== session.actorId)
+ ) return { ok: false, code: "ownership", message: "Realtime entity is not owned by this session." };
+ const localKey = poseInterestKey(snapshot);
+ if (localKey !== null && !session.interestKeys.has(localKey)) {
+ return { ok: false, code: "interest", message: "Realtime pose is outside the joined interest." };
+ }
+ if (localKey === null && geographicInterestKeys(session).length !== 1) {
+ return {
+ ok: false,
+ code: "interest",
+ message: "Geographic pose requires exactly one outdoor interest.",
+ };
+ }
+ const key = entityKey(snapshot);
+ const previous = session.proposedPoses.get(key);
+ if (previous) {
+ const motion = validateMotion(
+ previous,
+ snapshot,
+ snapshot.entity === "vehicle" ? settings.vehicleMotionLimits : settings.actorMotionLimits,
+ );
+ if (!motion.ok) {
+ return { ok: false, code: "motion", message: `Realtime motion rejected: ${motion.reason}.` };
+ }
+ }
+ session.lastClientSequence = snapshot.sequence;
+ const authoritative = clonePose({ ...snapshot, sequence: nextSequence(), timestampMs: now });
+ const removedEntityIds = [...session.poses.keys()].filter((existing) => existing !== key);
+ for (const removed of removedEntityIds) {
+ session.poses.delete(removed);
+ session.proposedPoses.delete(removed);
+ }
+ session.proposedPoses.set(key, clonePose(snapshot));
+ session.poses.set(key, authoritative);
+ const delta: PoseDelta = {
+ type: "pose-delta",
+ protocolVersion: 1,
+ serverEpoch: epoch,
+ sequence: authoritative.sequence,
+ timestampMs: now,
+ updates: [clonePose(authoritative)],
+ removedEntityIds,
+ };
+ broadcast(delta, poseBroadcastKeys(session, snapshot));
+ return { ok: true, value: delta };
+ }
+
+ function leave(sessionId: string, token: string): RealtimeResult {
+ const found = authenticate(sessionId, token);
+ if (!found.ok) return found;
+ remove(found.value);
+ return { ok: true, value: null };
+ }
+
+ function revokeSubject(subject: string): number {
+ let count = 0;
+ for (const session of sessions.values()) {
+ if (session.subject !== subject) continue;
+ const message: ServerRealtimeMessage = {
+ type: "membership-revoked",
+ protocolVersion: 1,
+ serverEpoch: epoch,
+ sequence: nextSequence(),
+ timestampMs: settings.now(),
+ sessionId: session.id,
+ actorId: session.actorId,
+ reason: "membership-revoked",
+ reconnectAllowed: false,
+ };
+ for (const listener of session.listeners) listener(message);
+ broadcastRemoval(session);
+ remove(session, false);
+ count += 1;
+ }
+ return count;
+ }
+
+ return {
+ join,
+ authenticate(sessionId, token) {
+ const found = authenticate(sessionId, token);
+ return found.ok ? { ok: true, value: view(found.value) } : found;
+ },
+ connect,
+ resume,
+ connectResume,
+ submitPose,
+ leave,
+ revokeSubject,
+ cleanup,
+ sessionCount: () => sessions.size,
+ };
+}
diff --git a/server/src/routes/realtime.ts b/server/src/routes/realtime.ts
new file mode 100644
index 0000000..f7095a1
--- /dev/null
+++ b/server/src/routes/realtime.ts
@@ -0,0 +1,186 @@
+/** Authenticated HTTP + SSE adapter for the in-memory realtime service. */
+
+import type { FastifyInstance, FastifyReply } from "fastify";
+import type { InterestCell } from "../../../src/realtime/types.ts";
+import type { ResumeRequest } from "../../../src/realtime/types.ts";
+import { validateClientRealtimeMessage } from "../../../src/realtime/protocol.ts";
+import type { RealtimeFailureCode } from "../realtime/index.ts";
+import type { Services } from "../services.ts";
+
+const BODY_LIMIT = 32 * 1024;
+const UNAUTHORIZED = { error: "unauthorized", message: "Realtime requires a signed-in member." };
+
+/** Public, statically bundled demos that exist even when TERA_OFFICES_DIR is empty. */
+export const BUNDLED_REALTIME_OFFICE_IDS: ReadonlySet = new Set([
+ "lumbridge-hq",
+ "frontier-valley",
+ "mateo-court",
+]);
+
+interface CredentialBody {
+ sessionId?: unknown;
+ token?: unknown;
+}
+
+interface PoseBody extends CredentialBody {
+ snapshot?: unknown;
+}
+
+function text(value: unknown): value is string {
+ return typeof value === "string" && value.length > 0 && value.length <= 512;
+}
+
+function failureStatus(code: RealtimeFailureCode): number {
+ switch (code) {
+ case "unauthorized":
+ case "expired": return 401;
+ case "capacity":
+ case "rate-limited": return 429;
+ case "conflict":
+ case "sequence":
+ case "interest":
+ case "ownership":
+ case "motion": return 409;
+ case "invalid": return 400;
+ }
+}
+
+function sendFailure(reply: FastifyReply, result: { code: RealtimeFailureCode; message: string }) {
+ const status = failureStatus(result.code);
+ if (status === 401) reply.header("www-authenticate", "Bearer");
+ return reply.code(status).send({ error: result.code, message: result.message });
+}
+
+async function interestsExist(interests: readonly InterestCell[], services: Services): Promise {
+ for (const cell of interests) {
+ if (cell.kind === "california-tile" || cell.kind === "city") continue;
+ const doc = await services.offices.get(cell.officeId);
+ if (doc === null) {
+ // Only the office envelope is public in a client bundle. Without a
+ // server pack we have no authoritative floor or room catalogue, so those
+ // finer cells remain unavailable rather than guessed.
+ if (cell.kind === "office" && BUNDLED_REALTIME_OFFICE_IDS.has(cell.officeId)) continue;
+ return false;
+ }
+ if (cell.kind === "office") continue;
+ const level = doc.floor.levels.find((candidate) => candidate.id === cell.floorId);
+ if (!level) return false;
+ if (
+ cell.kind === "room" &&
+ !level.floorplan.rooms.some((room) => room.id === cell.roomId)
+ ) return false;
+ }
+ return true;
+}
+
+export function registerRealtime(app: FastifyInstance, services: Services): void {
+ app.post<{ Body: unknown }>(
+ "/api/v1/realtime/join",
+ { bodyLimit: BODY_LIMIT },
+ async (req, reply) => {
+ // Authentication precedes interest parsing so anonymous callers cannot
+ // use office cell errors to learn private identifiers.
+ const viewer = await services.auth.resolve(req);
+ if (!viewer.authenticated || viewer.subject === null) {
+ return reply.code(401).header("www-authenticate", "Bearer").send(UNAUTHORIZED);
+ }
+ const parsed = validateClientRealtimeMessage(req.body);
+ if (!parsed.ok) {
+ return reply.code(400).send({ error: "invalid", message: "Realtime join request is invalid." });
+ }
+ const request = parsed.value;
+ const interests = request.interests;
+ if (!(await interestsExist(interests, services))) {
+ return reply.code(400).send({ error: "invalid", message: "Realtime interest is unavailable." });
+ }
+ if (request.type === "resume-request") {
+ const resumed = services.realtime.resume(request, viewer.subject);
+ if (!resumed.ok) return sendFailure(reply, resumed);
+ return reply.send(resumed.value);
+ }
+ const joined = services.realtime.join({
+ requestId: request.requestId,
+ actorId: request.actorId,
+ subject: viewer.subject,
+ role: viewer.admin ? "admin" : "member",
+ interests,
+ });
+ if (!joined.ok) return sendFailure(reply, joined);
+ return reply.code(201).send(joined.value.grant);
+ },
+ );
+
+ // POST is deliberate: usable session tokens must never enter URLs, reverse-
+ // proxy logs, referrers, or browser history. Clients consume this SSE body
+ // through fetch streaming rather than the URL-only EventSource constructor.
+ app.post<{ Body: unknown }>(
+ "/api/v1/realtime/events",
+ { bodyLimit: 4 * 1024 },
+ async (req, reply) => {
+ const viewer = await services.auth.resolve(req);
+ if (!viewer.authenticated || viewer.subject === null) {
+ return reply.code(401).header("www-authenticate", "Bearer").send(UNAUTHORIZED);
+ }
+ const parsed = validateClientRealtimeMessage(req.body);
+ if (!parsed.ok || parsed.value.type !== "resume-request") {
+ return reply.code(400).send({ error: "invalid", message: "Realtime resume request is invalid." });
+ }
+ const request: ResumeRequest = parsed.value;
+ if (!(await interestsExist(request.interests, services))) {
+ return reply.code(400).send({ error: "invalid", message: "Realtime interest is unavailable." });
+ }
+ const connection = services.realtime.connectResume(request, viewer.subject, (message) => {
+ if (!reply.raw.destroyed) reply.raw.write(`event: ${message.type}\ndata: ${JSON.stringify(message)}\n\n`);
+ });
+ if (!connection.ok) return sendFailure(reply, connection);
+
+ reply.hijack();
+ reply.raw.statusCode = 200;
+ reply.raw.setHeader("content-type", "text/event-stream; charset=utf-8");
+ reply.raw.setHeader("cache-control", "private, no-store");
+ reply.raw.setHeader("connection", "keep-alive");
+ reply.raw.setHeader("x-accel-buffering", "no");
+ reply.raw.flushHeaders();
+ reply.raw.write(`event: resume-grant\ndata: ${JSON.stringify(connection.value.grant)}\n\n`);
+ const keepalive = setInterval(() => {
+ if (!reply.raw.destroyed) reply.raw.write(`: keepalive ${Date.now()}\n\n`);
+ }, 15_000);
+ keepalive.unref();
+ // The POST request body finishes immediately; listening on IncomingMessage
+ // `close` would therefore tear the stream down as soon as credentials
+ // were parsed. ServerResponse stays open for the lifetime of the SSE feed.
+ reply.raw.once("close", () => {
+ clearInterval(keepalive);
+ connection.value.disconnect();
+ });
+ },
+ );
+
+ app.post<{ Body: PoseBody }>(
+ "/api/v1/realtime/pose",
+ { bodyLimit: BODY_LIMIT },
+ async (req, reply) => {
+ const body = req.body;
+ if (!body || !text(body.sessionId) || !text(body.token)) {
+ return reply.code(400).send({ error: "invalid", message: "Realtime credentials are required." });
+ }
+ const submitted = services.realtime.submitPose(body.sessionId, body.token, body.snapshot);
+ if (!submitted.ok) return sendFailure(reply, submitted);
+ return reply.code(202).send(submitted.value);
+ },
+ );
+
+ app.post<{ Body: CredentialBody }>(
+ "/api/v1/realtime/leave",
+ { bodyLimit: 4 * 1024 },
+ async (req, reply) => {
+ const body = req.body;
+ if (!body || !text(body.sessionId) || !text(body.token)) {
+ return reply.code(400).send({ error: "invalid", message: "Realtime credentials are required." });
+ }
+ const left = services.realtime.leave(body.sessionId, body.token);
+ if (!left.ok) return sendFailure(reply, left);
+ return reply.code(204).send();
+ },
+ );
+}
diff --git a/server/src/services.ts b/server/src/services.ts
index 9866c37..e7ec385 100644
--- a/server/src/services.ts
+++ b/server/src/services.ts
@@ -12,6 +12,7 @@ import { createFlightsService, type FlightsService } from "./flights/index.ts";
import { createMarkerStore, type MarkerStore } from "./markers/store.ts";
import { createOfficeStore, type OfficeStore } from "./offices/store.ts";
import { createPresenceStore, type PresenceStore } from "./presence/store.ts";
+import { createRealtimeService, type RealtimeService } from "./realtime/index.ts";
import { createSatellitesService, type SatellitesService } from "./satellites/index.ts";
import { createWeatherService, type WeatherService } from "./weather/index.ts";
import type { Config } from "./config.ts";
@@ -24,6 +25,7 @@ export interface Services {
markers: MarkerStore;
offices: OfficeStore;
presence: PresenceStore;
+ realtime: RealtimeService;
auth: AuthService;
/** Epoch milliseconds, for `uptimeSeconds` on the health body. */
startedAt: number;
@@ -43,6 +45,7 @@ export function createServices(config: Config, log: ServiceLog): Services {
markers: createMarkerStore(config, log),
offices: createOfficeStore(config.offices.dir),
presence: createPresenceStore(config.presence.dir),
+ realtime: createRealtimeService(),
auth: createAuth(config.auth),
startedAt: Date.now(),
};
diff --git a/server/src/test/realtime.test.ts b/server/src/test/realtime.test.ts
new file mode 100644
index 0000000..fca3da1
--- /dev/null
+++ b/server/src/test/realtime.test.ts
@@ -0,0 +1,663 @@
+import assert from "node:assert/strict";
+import { createHmac } from "node:crypto";
+import { after, describe, it } from "node:test";
+import type {
+ ActorPoseSnapshot,
+ FloorInterest,
+ InterestCell,
+ PoseDelta,
+ VehiclePoseSnapshot,
+} from "../../../src/realtime/types.ts";
+import { buildApp } from "../app.ts";
+import { loadConfig } from "../config.ts";
+import { createRealtimeService } from "../realtime/index.ts";
+import { BUNDLED_REALTIME_OFFICE_IDS } from "../routes/realtime.ts";
+
+const FLOOR: FloorInterest = {
+ kind: "floor",
+ officeId: "lumbridge-hq",
+ floorId: "level-1",
+};
+const OTHER_FLOOR: FloorInterest = {
+ kind: "floor",
+ officeId: "mateo-court",
+ floorId: "level-1",
+};
+
+function actorPose(
+ actorId: string,
+ sequence: number,
+ timestampMs: number,
+ xM: number,
+ cell: InterestCell = FLOOR,
+): ActorPoseSnapshot {
+ if (cell.kind === "california-tile") throw new Error("fixture needs local interest");
+ return {
+ entity: "actor",
+ actorId,
+ kind: "humanoid",
+ sequence,
+ timestampMs,
+ pose: {
+ space: "local",
+ cell,
+ xM,
+ yM: 0,
+ zM: 0,
+ headingDeg: 0,
+ pitchDeg: 0,
+ },
+ velocity: { xMps: 0, yMps: 0, zMps: 0, yawDegPerSec: 0 },
+ };
+}
+
+function geographicActor(
+ actorId: string,
+ sequence: number,
+ timestampMs: number,
+ lat = 37.7,
+): ActorPoseSnapshot {
+ return {
+ entity: "actor",
+ actorId,
+ kind: "humanoid",
+ sequence,
+ timestampMs,
+ pose: {
+ space: "geographic",
+ lat,
+ lng: -122.4,
+ altitudeM: 10,
+ headingDeg: 0,
+ pitchDeg: 0,
+ },
+ velocity: { xMps: 0, yMps: 0, zMps: 0, yawDegPerSec: 0 },
+ };
+}
+
+function vehiclePose(
+ driverActorId: string,
+ sequence: number,
+ timestampMs: number,
+): VehiclePoseSnapshot {
+ return {
+ entity: "vehicle",
+ vehicleId: `vehicle-${driverActorId}`,
+ kind: "model-x",
+ driverActorId,
+ sequence,
+ timestampMs,
+ pose: {
+ space: "local",
+ cell: FLOOR,
+ xM: 0,
+ yM: 0,
+ zM: 0,
+ headingDeg: 0,
+ pitchDeg: 0,
+ },
+ velocity: { xMps: 0, yMps: 0, zMps: 0, yawDegPerSec: 0 },
+ steering: 0,
+ wheelRadians: 0,
+ };
+}
+
+function joined(
+ service: ReturnType,
+ actorId: string,
+ interests: readonly InterestCell[] = [FLOOR],
+ subject = `auth-subject-for-${actorId}`,
+) {
+ const result = service.join({
+ requestId: `request-${actorId}`,
+ actorId,
+ subject,
+ role: "member",
+ interests,
+ });
+ assert.equal(result.ok, true);
+ if (!result.ok) throw new Error("realtime fixture could not join");
+ return result.value;
+}
+
+describe("authoritative realtime service", () => {
+ it("stores only hashed credentials and authenticates opaque sessions", () => {
+ const service = createRealtimeService();
+ const session = joined(service, "member-a");
+ assert.notEqual(session.sessionId, session.token);
+ assert.equal(service.authenticate(session.sessionId, session.token).ok, true);
+ assert.equal(service.authenticate(session.sessionId, `${session.token}x`).ok, false);
+ assert.equal(session.grant.resumeToken, session.token);
+ assert.deepEqual(session.grant.initial, []);
+ });
+
+ it("keeps auth subjects private while reserving each opaque actor id", () => {
+ const service = createRealtimeService();
+ const first = service.join({
+ requestId: "request",
+ actorId: "page-f4f9445d",
+ subject: "private-auth-subject",
+ role: "member",
+ interests: [FLOOR],
+ });
+ assert.equal(first.ok, true);
+ if (!first.ok) return;
+ assert.equal(JSON.stringify(first.value.grant).includes("private-auth-subject"), false);
+ const view = service.authenticate(first.value.sessionId, first.value.token);
+ assert.equal(view.ok, true);
+ if (view.ok) assert.equal(JSON.stringify(view.value).includes("private-auth-subject"), false);
+
+ const collision = service.join({
+ requestId: "request-2",
+ actorId: "page-f4f9445d",
+ subject: "another-private-subject",
+ role: "member",
+ interests: [FLOOR],
+ });
+ assert.deepEqual(collision, {
+ ok: false,
+ code: "conflict",
+ message: "Realtime actor id is already active.",
+ });
+ });
+
+ it("broadcasts poses only to listeners sharing the exact interest cell", () => {
+ let now = 10_000;
+ const service = createRealtimeService({ now: () => now });
+ const a = joined(service, "member-a");
+ const b = joined(service, "member-b");
+ const c = joined(service, "member-c", [OTHER_FLOOR]);
+ const receivedA: unknown[] = [];
+ const receivedB: unknown[] = [];
+ const receivedC: unknown[] = [];
+ service.connect(a.sessionId, a.token, (message) => receivedA.push(message));
+ service.connect(b.sessionId, b.token, (message) => receivedB.push(message));
+ service.connect(c.sessionId, c.token, (message) => receivedC.push(message));
+ const submitted = service.submitPose(
+ a.sessionId,
+ a.token,
+ actorPose("member-a", 1, now, 0),
+ );
+ assert.equal(submitted.ok, true);
+ assert.equal(receivedA.length, 1);
+ assert.equal(receivedB.length, 1);
+ assert.equal(receivedC.length, 0);
+ now += 1_000;
+ const second = service.submitPose(a.sessionId, a.token, actorPose("member-a", 2, now, 5));
+ assert.equal(second.ok, true);
+ const late = joined(service, "member-late");
+ assert.equal(late.grant.initial.length, 1, "a late join receives the current cell snapshot");
+ });
+
+ it("allows geographic city poses and scopes them to the exact detailed city", () => {
+ const now = 15_000;
+ const service = createRealtimeService({ now: () => now });
+ const bay = { kind: "city" as const, cityId: "bay-area" as const };
+ const socal = { kind: "city" as const, cityId: "socal" as const };
+ const source = joined(service, "page-bay-source", [bay]);
+ const bayPeer = joined(service, "page-bay-peer", [bay]);
+ const socalPeer = joined(service, "page-socal-peer", [socal]);
+ const bayMessages: unknown[] = [];
+ const socalMessages: unknown[] = [];
+ service.connect(bayPeer.sessionId, bayPeer.token, (message) => bayMessages.push(message));
+ service.connect(socalPeer.sessionId, socalPeer.token, (message) => socalMessages.push(message));
+ const sent = service.submitPose(
+ source.sessionId,
+ source.token,
+ geographicActor("page-bay-source", 1, now),
+ );
+ assert.equal(sent.ok, true);
+ assert.equal(bayMessages.length, 1);
+ assert.equal(socalMessages.length, 0);
+ });
+
+ it("removes the previous entity when one session switches actor to vehicle", () => {
+ let now = 17_000;
+ const service = createRealtimeService({ now: () => now });
+ const source = joined(service, "page-switch");
+ const peer = joined(service, "page-peer");
+ const received: PoseDelta[] = [];
+ service.connect(peer.sessionId, peer.token, (message) => {
+ if (message.type === "pose-delta") received.push(message);
+ });
+ service.submitPose(source.sessionId, source.token, actorPose("page-switch", 1, now, 0));
+ now += 100;
+ const switched = service.submitPose(
+ source.sessionId,
+ source.token,
+ vehiclePose("page-switch", 2, now),
+ );
+ assert.equal(switched.ok, true);
+ if (!switched.ok) return;
+ assert.deepEqual(switched.value.removedEntityIds, ["actor:page-switch"]);
+ assert.equal(switched.value.updates[0]?.entity, "vehicle");
+ assert.ok(switched.value.sequence > (received[0]?.sequence ?? -1));
+ assert.deepEqual(received.at(-1)?.removedEntityIds, ["actor:page-switch"]);
+ });
+
+ it("broadcasts removals to old-cell peers on interest move and leave", () => {
+ let now = 18_000;
+ const service = createRealtimeService({ now: () => now });
+ const source = joined(service, "page-moving");
+ const oldPeer = joined(service, "page-old-peer");
+ const received: PoseDelta[] = [];
+ service.connect(oldPeer.sessionId, oldPeer.token, (message) => {
+ if (message.type === "pose-delta") received.push(message);
+ });
+ service.submitPose(source.sessionId, source.token, actorPose("page-moving", 1, now, 0));
+ const moved = service.resume({
+ type: "resume-request",
+ protocolVersion: 1,
+ requestId: "move-request",
+ sessionId: source.sessionId,
+ resumeToken: source.token,
+ serverEpoch: source.grant.serverEpoch,
+ lastReceivedSequence: received.at(-1)?.sequence ?? 0,
+ interests: [OTHER_FLOOR],
+ }, "auth-subject-for-page-moving");
+ assert.equal(moved.ok, true);
+ assert.deepEqual(received.at(-1)?.removedEntityIds, ["actor:page-moving"]);
+
+ const leaving = joined(service, "page-leaving");
+ now += 100;
+ service.submitPose(leaving.sessionId, leaving.token, actorPose("page-leaving", 1, now, 0));
+ const beforeLeave = received.at(-1)?.sequence ?? -1;
+ assert.equal(service.leave(leaving.sessionId, leaving.token).ok, true);
+ assert.deepEqual(received.at(-1)?.removedEntityIds, ["actor:page-leaving"]);
+ assert.ok((received.at(-1)?.sequence ?? -1) > beforeLeave);
+ });
+
+ it("rejects stale, unowned, off-interest and impossible motion", () => {
+ let now = 20_000;
+ const service = createRealtimeService({ now: () => now });
+ const session = joined(service, "member-a");
+ assert.equal(service.submitPose(
+ session.sessionId,
+ session.token,
+ actorPose("member-b", 1, now, 0),
+ ).ok, false);
+ assert.equal(service.submitPose(
+ session.sessionId,
+ session.token,
+ actorPose("member-a", 1, now, 0, OTHER_FLOOR),
+ ).ok, false);
+ assert.equal(service.submitPose(
+ session.sessionId,
+ session.token,
+ actorPose("member-a", 1, now, 0),
+ ).ok, true);
+ assert.equal(service.submitPose(
+ session.sessionId,
+ session.token,
+ actorPose("member-a", 1, now, 1),
+ ).ok, false);
+ now += 100;
+ const hacked = service.submitPose(
+ session.sessionId,
+ session.token,
+ actorPose("member-a", 2, now, 1_000),
+ );
+ assert.equal(hacked.ok, false);
+ if (!hacked.ok) assert.equal(hacked.code, "motion");
+ });
+
+ it("caps session/cell occupancy and pose rate", () => {
+ let now = 30_000;
+ const service = createRealtimeService({
+ now: () => now,
+ maximumSessionsPerCell: 1,
+ maximumPoseUpdatesPerWindow: 1,
+ });
+ const session = joined(service, "member-a");
+ const refused = service.join({
+ requestId: "request-b",
+ actorId: "member-b",
+ subject: "member-b",
+ role: "member",
+ interests: [FLOOR],
+ });
+ assert.equal(refused.ok, false);
+ assert.equal(service.submitPose(
+ session.sessionId,
+ session.token,
+ actorPose("member-a", 1, now, 0),
+ ).ok, true);
+ now += 50;
+ const limited = service.submitPose(
+ session.sessionId,
+ session.token,
+ actorPose("member-a", 2, now, 0),
+ );
+ assert.equal(limited.ok, false);
+ if (!limited.ok) assert.equal(limited.code, "rate-limited");
+ });
+
+ it("honours disconnect grace, absolute TTL, leave and membership revocation", () => {
+ let now = 40_000;
+ const service = createRealtimeService({
+ now: () => now,
+ disconnectGraceMs: 500,
+ sessionTtlMs: 2_000,
+ });
+ const session = joined(service, "member-a");
+ const messages: unknown[] = [];
+ const connected = service.connect(session.sessionId, session.token, (message) => messages.push(message));
+ assert.equal(connected.ok, true);
+ if (!connected.ok) return;
+ connected.value.disconnect();
+ now += 400;
+ assert.equal(service.cleanup(), 0);
+ assert.equal(service.authenticate(session.sessionId, session.token).ok, true);
+ now += 600;
+ assert.equal(service.cleanup(), 1);
+
+ const revoked = joined(service, "member-b");
+ service.connect(revoked.sessionId, revoked.token, (message) => messages.push(message));
+ assert.equal(service.revokeSubject("auth-subject-for-member-b"), 1);
+ assert.equal(messages.length, 1);
+ assert.equal(service.authenticate(revoked.sessionId, revoked.token).ok, false);
+
+ const left = joined(service, "member-c");
+ assert.equal(service.leave(left.sessionId, left.token).ok, true);
+ assert.equal(service.sessionCount(), 0);
+ });
+
+ it("strictly resumes, changes interest, and rotates the usable token", () => {
+ const service = createRealtimeService({ now: () => 50_000 });
+ const session = joined(service, "member-a");
+ const resumed = service.resume({
+ type: "resume-request",
+ protocolVersion: 1,
+ requestId: "resume-1",
+ sessionId: session.sessionId,
+ resumeToken: session.token,
+ serverEpoch: session.grant.serverEpoch,
+ lastReceivedSequence: session.grant.nextSequence === 0 ? 0 : session.grant.nextSequence - 1,
+ interests: [OTHER_FLOOR],
+ }, "auth-subject-for-member-a");
+ assert.equal(resumed.ok, true);
+ if (!resumed.ok) return;
+ assert.equal(resumed.value.type, "resume-grant");
+ assert.equal(resumed.value.requestId, "resume-1");
+ assert.notEqual(resumed.value.resumeToken, session.token);
+ assert.equal(service.authenticate(session.sessionId, session.token).ok, false);
+ const current = service.authenticate(session.sessionId, resumed.value.resumeToken);
+ assert.equal(current.ok, true);
+ if (current.ok) assert.deepEqual(current.value.interests, [OTHER_FLOOR]);
+ });
+});
+
+const SECRET = "realtime-test-secret-that-is-long-enough";
+
+function jwt(subject: string): string {
+ const encode = (value: unknown): string => Buffer.from(JSON.stringify(value)).toString("base64url");
+ const signed = `${encode({ alg: "HS256", typ: "JWT" })}.${encode({
+ sub: subject,
+ exp: Math.floor(Date.now() / 1_000) + 3_600,
+ })}`;
+ return `${signed}.${createHmac("sha256", SECRET).update(signed).digest("base64url")}`;
+}
+
+function app() {
+ const config = loadConfig({ TERA_AUTH_MODE: "jwt", TERA_AUTH_JWT_SECRET: SECRET });
+ config.logLevel = "silent";
+ return buildApp(config);
+}
+
+function joinBody(actorId: string, interests: readonly InterestCell[]) {
+ return {
+ type: "join-request",
+ protocolVersion: 1,
+ requestId: `request-${actorId}`,
+ actorId,
+ resumeToken: null,
+ lastReceivedSequence: null,
+ interests,
+ };
+}
+
+describe("realtime HTTP routes", () => {
+ it("declares exactly the three public bundled office fallbacks", () => {
+ assert.deepEqual([...BUNDLED_REALTIME_OFFICE_IDS].sort(), [
+ "frontier-valley",
+ "lumbridge-hq",
+ "mateo-court",
+ ]);
+ });
+
+ it("never accepts stream credentials in a URL", async () => {
+ const server = app();
+ after(() => server.close());
+ const response = await server.inject({
+ method: "GET",
+ url: "/api/v1/realtime/events?sessionId=leak&token=must-not-be-logged",
+ });
+ assert.equal(response.statusCode, 404);
+ });
+
+ it("authenticates before joining and rejects opaque actor-id collisions", async () => {
+ const server = app();
+ after(() => server.close());
+ const anonymous = await server.inject({
+ method: "POST",
+ url: "/api/v1/realtime/join",
+ payload: joinBody("member-a", [{ kind: "city", cityId: "bay-area" }]),
+ });
+ assert.equal(anonymous.statusCode, 401);
+ const opaque = await server.inject({
+ method: "POST",
+ url: "/api/v1/realtime/join",
+ headers: { authorization: `Bearer ${jwt("member-a")}` },
+ payload: joinBody("page-random-opaque", [{ kind: "city", cityId: "bay-area" }]),
+ });
+ assert.equal(opaque.statusCode, 201);
+ assert.equal(JSON.stringify(opaque.json()).includes("member-a"), false);
+ const collision = await server.inject({
+ method: "POST",
+ url: "/api/v1/realtime/join",
+ headers: { authorization: `Bearer ${jwt("member-b")}` },
+ payload: joinBody("page-random-opaque", [{ kind: "city", cityId: "bay-area" }]),
+ });
+ assert.equal(collision.statusCode, 409);
+ const absentOffice = await server.inject({
+ method: "POST",
+ url: "/api/v1/realtime/join",
+ headers: { authorization: `Bearer ${jwt("member-a")}` },
+ payload: joinBody("member-a", [{ kind: "office", officeId: "invented-private-office" }]),
+ });
+ assert.equal(absentOffice.statusCode, 400);
+ assert.deepEqual(absentOffice.json(), {
+ error: "invalid",
+ message: "Realtime interest is unavailable.",
+ });
+ });
+
+ it("allows bundled demo office envelopes without disk packs but not invented floors", async () => {
+ const server = app();
+ after(() => server.close());
+ const authorization = `Bearer ${jwt("member-office")}`;
+ const office = await server.inject({
+ method: "POST",
+ url: "/api/v1/realtime/join",
+ headers: { authorization },
+ payload: joinBody("member-office", [{ kind: "office", officeId: "lumbridge-hq" }]),
+ });
+ assert.equal(office.statusCode, 201);
+ const floor = await server.inject({
+ method: "POST",
+ url: "/api/v1/realtime/join",
+ headers: { authorization },
+ payload: joinBody("member-office", [{
+ kind: "floor",
+ officeId: "lumbridge-hq",
+ floorId: "invented-floor",
+ }]),
+ });
+ assert.equal(floor.statusCode, 400);
+ });
+
+ it("joins, accepts an owned pose, and leaves without persistence", async () => {
+ const server = app();
+ after(() => server.close());
+ const joinedResponse = await server.inject({
+ method: "POST",
+ url: "/api/v1/realtime/join",
+ headers: { authorization: `Bearer ${jwt("member-a")}` },
+ payload: joinBody("member-a", [{ kind: "california-tile", level: 6, x: 1, y: 2 }]),
+ });
+ assert.equal(joinedResponse.statusCode, 201);
+ const grant = joinedResponse.json<{ sessionId: string; resumeToken: string }>();
+ const timestampMs = Date.now();
+ const pose = await server.inject({
+ method: "POST",
+ url: "/api/v1/realtime/pose",
+ payload: {
+ sessionId: grant.sessionId,
+ token: grant.resumeToken,
+ snapshot: {
+ entity: "actor",
+ actorId: "member-a",
+ kind: "humanoid",
+ sequence: 1,
+ timestampMs,
+ pose: {
+ space: "geographic",
+ lat: 37.7,
+ lng: -122.4,
+ altitudeM: 10,
+ headingDeg: 0,
+ pitchDeg: 0,
+ },
+ velocity: { xMps: 0, yMps: 0, zMps: 0, yawDegPerSec: 0 },
+ },
+ },
+ });
+ assert.equal(pose.statusCode, 202);
+ assert.equal(pose.json<{ type: string }>().type, "pose-delta");
+ const left = await server.inject({
+ method: "POST",
+ url: "/api/v1/realtime/leave",
+ payload: { sessionId: grant.sessionId, token: grant.resumeToken },
+ });
+ assert.equal(left.statusCode, 204);
+ const gone = await server.inject({
+ method: "POST",
+ url: "/api/v1/realtime/pose",
+ payload: {
+ sessionId: grant.sessionId,
+ token: grant.resumeToken,
+ snapshot: {
+ entity: "actor",
+ actorId: "member-a",
+ kind: "humanoid",
+ sequence: 2,
+ timestampMs: timestampMs + 100,
+ pose: {
+ space: "geographic",
+ lat: 37.7,
+ lng: -122.4,
+ altitudeM: 10,
+ headingDeg: 0,
+ pitchDeg: 0,
+ },
+ velocity: { xMps: 0, yMps: 0, zMps: 0, yawDegPerSec: 0 },
+ },
+ },
+ });
+ assert.equal(gone.statusCode, 401);
+ });
+
+ it("streams the immediate snapshot from a credential-body POST", async () => {
+ const server = app();
+ const address = await server.listen({ host: "127.0.0.1", port: 0 });
+ const controller = new AbortController();
+ try {
+ const joinResponse = await fetch(`${address}/api/v1/realtime/join`, {
+ method: "POST",
+ headers: {
+ authorization: `Bearer ${jwt("member-stream")}`,
+ "content-type": "application/json",
+ },
+ body: JSON.stringify(joinBody("member-stream", [{ kind: "city", cityId: "bay-area" }])),
+ });
+ assert.equal(joinResponse.status, 201);
+ const grant = await joinResponse.json() as {
+ sessionId: string;
+ resumeToken: string;
+ serverEpoch: string;
+ nextSequence: number;
+ };
+ const events = await fetch(`${address}/api/v1/realtime/events`, {
+ method: "POST",
+ headers: {
+ authorization: `Bearer ${jwt("member-stream")}`,
+ "content-type": "application/json",
+ },
+ body: JSON.stringify({
+ type: "resume-request",
+ protocolVersion: 1,
+ requestId: "stream-resume",
+ sessionId: grant.sessionId,
+ resumeToken: grant.resumeToken,
+ serverEpoch: grant.serverEpoch,
+ lastReceivedSequence: grant.nextSequence === 0 ? 0 : grant.nextSequence - 1,
+ interests: [{ kind: "city", cityId: "bay-area" }],
+ }),
+ signal: controller.signal,
+ });
+ assert.equal(events.status, 200);
+ assert.match(events.headers.get("content-type") ?? "", /^text\/event-stream/);
+ const reader = events.body?.getReader();
+ assert.ok(reader);
+ const first = await reader.read();
+ assert.equal(first.done, false);
+ const event = new TextDecoder().decode(first.value);
+ assert.match(event, /event: resume-grant/);
+ assert.match(event, /"type":"resume-grant"/);
+ controller.abort();
+ await reader.cancel().catch(() => undefined);
+ } finally {
+ controller.abort();
+ await server.close();
+ }
+ });
+
+ it("accepts the root client's exact move-interest resume shape on join", async () => {
+ const server = app();
+ after(() => server.close());
+ const authorization = `Bearer ${jwt("member-client")}`;
+ const first = await server.inject({
+ method: "POST",
+ url: "/api/v1/realtime/join",
+ headers: { authorization },
+ payload: joinBody("member-client", [{ kind: "city", cityId: "bay-area" }]),
+ });
+ assert.equal(first.statusCode, 201);
+ const grant = first.json<{
+ sessionId: string;
+ resumeToken: string;
+ serverEpoch: string;
+ nextSequence: number;
+ }>();
+ const moved = await server.inject({
+ method: "POST",
+ url: "/api/v1/realtime/join",
+ headers: { authorization },
+ payload: {
+ type: "resume-request",
+ protocolVersion: 1,
+ requestId: "tera-client-2",
+ sessionId: grant.sessionId,
+ resumeToken: grant.resumeToken,
+ serverEpoch: grant.serverEpoch,
+ lastReceivedSequence: grant.nextSequence === 0 ? 0 : grant.nextSequence - 1,
+ interests: [{ kind: "city", cityId: "socal" }],
+ },
+ });
+ assert.equal(moved.statusCode, 200);
+ const resume = moved.json<{ type: string; requestId: string; resumeToken: string }>();
+ assert.equal(resume.type, "resume-grant");
+ assert.equal(resume.requestId, "tera-client-2");
+ assert.notEqual(resume.resumeToken, grant.resumeToken);
+ });
+});
diff --git a/src/engine/scene.ts b/src/engine/scene.ts
index b2bef60..1ff2d45 100644
--- a/src/engine/scene.ts
+++ b/src/engine/scene.ts
@@ -78,6 +78,10 @@ import type {
AircraftActionSnapshot,
AircraftControllerSnapshot,
} from "../aircraft/controller.ts";
+import type { ScenePeers, ScenePeersOptions } from "../realtime/scenePeers.ts";
+import type { EntityPoseSnapshot } from "../realtime/types.ts";
+
+export type CityRealtimePeersOptions = Omit;
export interface SceneOptions {
city: City;
@@ -99,6 +103,8 @@ export interface SceneOptions {
actorAnchor?: { lat: number; lng: number };
/** Optional possessed fixed-wing aircraft; the city supplies its projection and terrain. */
aircraft?: Omit;
+ /** Optional remote authoritative entities. Nothing is imported or allocated when absent. */
+ realtimePeers?: CityRealtimePeersOptions;
flights?: FlightSource;
/**
* Element sets to propagate, if this deployment has any.
@@ -205,6 +211,11 @@ export interface SceneHandle {
aircraftState(): Readonly | null;
setAircraftActive(active: boolean): void;
aircraftActive(): boolean;
+ /** Canonical remote snapshots only; a no-op when realtime peers were not configured. */
+ upsertRemoteSnapshot(snapshot: EntityPoseSnapshot): boolean;
+ removeRemoteEntity(id: string): boolean;
+ clearRemoteEntities(): void;
+ remoteEntityCount(): number;
setMarkers(markers: Marker[]): void;
/** Take this city off the stage and release everything it built. */
dispose(): void;
@@ -401,6 +412,23 @@ export async function createScene(
})
: null;
if (sceneAircraft) scene.add(sceneAircraft.root);
+ // Dynamic on purpose: the offline/default entry must not download four peer
+ // prototypes merely because `createScene` can optionally host them.
+ let realtimePeers: ScenePeers | null = null;
+ if (options.realtimePeers) {
+ const { createScenePeers } = await import("../realtime/scenePeers.ts");
+ realtimePeers = createScenePeers({
+ ...options.realtimePeers,
+ project: (lat, lng) => world.project(lat, lng),
+ groundAt: (lat, lng) => world.groundAt(lat, lng),
+ geographicSceneUnitsPerMetre:
+ options.realtimePeers.geographicSceneUnitsPerMetre ?? 1 / world.metresPerUnit,
+ geographicVisualScale:
+ options.realtimePeers.geographicVisualScale ??
+ (city.id === "california" ? 0.025 : 1 / world.metresPerUnit),
+ });
+ scene.add(realtimePeers.root);
+ }
let flightLayer: FlightLayer | null = null;
let flightTimer = 0;
@@ -521,6 +549,7 @@ export async function createScene(
if (actorPlaying && sceneActor) kit.setPose(sceneActor.followPose());
sceneAircraft?.tick(dt);
if (aircraftPlaying && sceneAircraft) kit.setPose(sceneAircraft.followPose());
+ realtimePeers?.tick(Date.now());
roadTraffic?.tick(dt);
clouds.tick(dt);
if (options.flights && flightLayer) {
@@ -572,6 +601,7 @@ export async function createScene(
markerLayer.dispose();
roadTraffic?.dispose();
sceneAircraft?.dispose();
+ realtimePeers?.dispose();
kit.dispose();
scene.traverse((obj) => {
const mesh = obj as THREE.Mesh;
@@ -644,12 +674,26 @@ export async function createScene(
kit.camera.updateProjectionMatrix();
},
aircraftActive: () => sceneAircraft?.active() ?? false,
+ upsertRemoteSnapshot: (snapshot) => {
+ if (snapshot.pose.space === "local") {
+ // Local office coordinates have no meaning on a geographic board.
+ // Local city coordinates are accepted only for this detailed board;
+ // California actors use geographic poses.
+ const expected = city.id === "sf" ? "bay-area" : city.id === "socal" ? "socal" : null;
+ if (snapshot.pose.cell.kind !== "city" || snapshot.pose.cell.cityId !== expected) return false;
+ }
+ return realtimePeers?.upsert(snapshot) ?? false;
+ },
+ removeRemoteEntity: (id) => realtimePeers?.remove(id) ?? false,
+ clearRemoteEntities: () => { realtimePeers?.clear(); },
+ remoteEntityCount: () => realtimePeers?.count() ?? 0,
setMarkers(markers) {
markerLayer.setMarkers(markers);
},
dispose() {
sceneActor?.dispose();
sceneAircraft?.dispose();
+ realtimePeers?.dispose();
/**
* Off the stage, then released — and the stage itself is left running.
*
diff --git a/src/interiors/officeScene.ts b/src/interiors/officeScene.ts
index ebcbbeb..22d783b 100644
--- a/src/interiors/officeScene.ts
+++ b/src/interiors/officeScene.ts
@@ -99,6 +99,13 @@ import {
type RobotSpec,
type RobotView,
} from "./robots.ts";
+import {
+ createScenePeers,
+ type ScenePeersOptions,
+} from "../realtime/scenePeers.ts";
+import type { EntityPoseSnapshot } from "../realtime/types.ts";
+
+export type OfficeRealtimePeersOptions = Omit;
/** Shared empty, so a pack with no robots does not allocate one per call. */
const NO_ROBOTS: readonly RobotView[] = [];
@@ -222,6 +229,8 @@ export interface OfficeSceneOptions {
robots?: readonly RobotSpec[];
/** Optional local walk actor. Constructed inactive unless `walker.active` says otherwise. */
walker?: OfficeWalkerOptions;
+ /** Full-depth-only authoritative remote actors/vehicles in local metre coordinates. */
+ realtimePeers?: OfficeRealtimePeersOptions;
/** Defaults to false — the lid comes off, because that is the whole view. */
showCeilings?: boolean;
/** Fade the walls you are looking through. Defaults to true. */
@@ -306,6 +315,10 @@ export interface OfficeScene extends StageScene {
* The plan panel and the ceiling lights both consume it that way.
*/
robots(): readonly RobotView[];
+ upsertRemoteSnapshot(snapshot: EntityPoseSnapshot): boolean;
+ removeRemoteEntity(id: string): boolean;
+ clearRemoteEntities(): void;
+ remoteEntityCount(): number;
}
export function createOfficeScene(office: Office, options: OfficeSceneOptions): OfficeScene {
@@ -509,6 +522,18 @@ export function createOfficeScene(office: Office, options: OfficeSceneOptions):
: null;
const officeWalker = options.walker ? createOfficeWalker(plan, options.walker) : null;
if (officeWalker) scene.add(officeWalker.root);
+ // Live occupancy is excluded from the public-depth build rather than hidden.
+ // Geographic callbacks are required by the generic adapter but harmless here:
+ // an office subscription sends validated local poses in metre coordinates.
+ const realtimePeers = depth === "full" && options.realtimePeers
+ ? createScenePeers({
+ ...options.realtimePeers,
+ project: () => [0, 0],
+ groundAt: () => 0,
+ localSceneUnitsPerMetre: options.realtimePeers.localSceneUnitsPerMetre ?? 1,
+ })
+ : null;
+ if (realtimePeers) scene.add(realtimePeers.root);
if (robots) {
scene.add(robots.group);
/**
@@ -830,6 +855,18 @@ export function createOfficeScene(office: Office, options: OfficeSceneOptions):
setWalkers(walkers) {
luminaires.setWalkers(walkers);
},
+ upsertRemoteSnapshot: (snapshot) => {
+ // The generic peer adapter can also project geographic poses, but an
+ // office must never render a stale city stream or another office's local
+ // coordinates at its origin during an interest handoff.
+ if (snapshot.pose.space !== "local") return false;
+ const cell = snapshot.pose.cell;
+ if (!("officeId" in cell) || cell.officeId !== office.id) return false;
+ return realtimePeers?.upsert(snapshot) ?? false;
+ },
+ removeRemoteEntity: (id) => realtimePeers?.remove(id) ?? false,
+ clearRemoteEntities: () => { realtimePeers?.clear(); },
+ remoteEntityCount: () => realtimePeers?.count() ?? 0,
// Stepping back out to the city should retire the hover with it, or the
// detail card for whoever the pointer was over survives the journey.
onExit: () => kit.resetPick(),
@@ -840,6 +877,7 @@ export function createOfficeScene(office: Office, options: OfficeSceneOptions):
kit.tick(dt);
officeWalker?.tick(dt);
if (walking && officeWalker) kit.setPose(officeWalker.followPose());
+ realtimePeers?.tick(Date.now());
updateOcclusion();
// Robots first: the lights above them should respond to where they are
// *now*, not to where they were last frame.
@@ -849,6 +887,7 @@ export function createOfficeScene(office: Office, options: OfficeSceneOptions):
dispose() {
mediaSurfaces.dispose();
officeWalker?.dispose();
+ realtimePeers?.dispose();
robots?.dispose();
luminaires.dispose();
if (horizonPlane) {
diff --git a/src/main.ts b/src/main.ts
index 957e5dc..0b0f28a 100644
--- a/src/main.ts
+++ b/src/main.ts
@@ -114,6 +114,13 @@ import type { MaterialRegistry } from "./assets/materials.ts";
import type { OfficeMinimap } from "./engine/officeMinimap.ts";
import type { Godmode, GodmodeHouseLights, GodmodePlace } from "./tools/index.ts";
import type { PoseEditor } from "./tools/poseEditor.ts";
+import type {
+ EntityPoseSnapshot,
+ InterestCell,
+ RealtimeClient,
+ ServerRealtimeMessage,
+} from "./realtime/index.ts";
+import type { PresenceIndicator } from "./realtime/presenceIndicator.ts";
const CITIES: { id: string; label: string; city: City }[] = [
{ id: "california", label: "California", city: CALIFORNIA },
@@ -247,6 +254,15 @@ let wantedCity = "california";
let office: OfficeScene | null = null;
let inside = false;
let markers: Marker[] = SAMPLE_MARKERS;
+let realtimeClient: RealtimeClient | null = null;
+let presenceIndicator: PresenceIndicator | null = null;
+let stopRealtimeSubscription: (() => void) | null = null;
+let realtimeOperation = 0;
+let lastRealtimePublishAt = 0;
+let realtimePageActive = true;
+/** Page-scoped wire identities; auth subject never enters the spatial protocol. */
+const realtimeActorId = crypto.randomUUID();
+const realtimeVehicleId = crypto.randomUUID();
/**
* The buildings you can walk into, as procedural glyphs on the city.
@@ -921,6 +937,9 @@ async function mountCity(id: string) {
onProgress: (p) => {
if (!mount.signal.aborted) bootProgress(entry.label, p.fraction, p.onMainThread);
},
+ ...(access.subject !== null
+ ? { realtimePeers: { localSceneUnitsPerMetre: entry.city.latScale / 111_320 } }
+ : {}),
});
if (!handle) {
/**
@@ -934,6 +953,7 @@ async function mountCity(id: string) {
return;
}
city = handle;
+ moveRealtimePresence();
// A new board builds a new layer, and a new layer starts visible. Reapply
// whatever the panel last said, or the setting silently undoes itself on the
// first city switch.
@@ -1086,6 +1106,7 @@ requestAnimationFrame(function pumpMinimap() {
// everyone who is not god, so this is one property read per frame on a
// public page.
poseEditor?.tick();
+ publishRealtimePresence(performance.now());
pollLiveness();
});
@@ -1231,12 +1252,14 @@ async function enterOffice() {
...(depth === "full"
? { onPresencePick: (p) => showDetail(p ? p.label : null) }
: { onPlacePick: (place) => showDetail(place ? place.label : null) }),
+ ...(access.subject !== null ? { realtimePeers: {} } : {}),
});
office.onViewChange(() => renderLegend());
officePlan = buildOfficePlan(createOfficeMinimap, office);
}
city.stage.setScene(office);
inside = true;
+ moveRealtimePresence();
const desiredCity: JourneyCity = officeId === "mateo-court" ? "socal" : "bay-area";
journeyToCity(desiredCity);
dispatchJourney({ type: "enter-office", officeId: officeId as "lumbridge-hq" | "frontier-valley" | "mateo-court" });
@@ -1362,6 +1385,7 @@ function leaveOffice() {
dispatchJourney({ type: "leave-office" });
city.stage.setScene(city.stageScene);
inside = false;
+ moveRealtimePresence();
showPlan();
showDetail(null);
refreshGodmodePlace();
@@ -1512,6 +1536,7 @@ const source = document.querySelector("#source");
const minimapFrame = document.querySelector("#minimap .minimap-frame");
const minimapReadout = document.querySelector("#minimap-readout");
const tierBadge = document.querySelector("#tier");
+const presenceHost = document.querySelector("#presence-host");
const officeBadge = document.querySelector("#office-badge");
const panelToggle = document.querySelector("#panel-toggle");
const panelToggleLabel = document.querySelector("#panel-toggle-label");
@@ -1927,6 +1952,256 @@ function renderTierBadge() {
tierBadge.hidden = false;
}
+// ---- Hosted realtime presence --------------------------------------------
+
+function realtimeInterest(): InterestCell {
+ if (inside) return { kind: "office", officeId };
+ if (cityId === "california") return { kind: "california-tile", x: 0, y: 0, level: 0 };
+ return { kind: "city", cityId: cityId === "socal" ? "socal" : "bay-area" };
+}
+
+function degrees(radians: number): number {
+ return ((radians * 180 / Math.PI + 540) % 360) - 180;
+}
+
+function realtimePose(): EntityPoseSnapshot | null {
+ if (access.subject === null) return null;
+ const timestampMs = Date.now();
+
+ if (inside) {
+ const walker = office?.walker;
+ if (!walker) return null;
+ const state = walker.state();
+ const headingDeg = -degrees(Math.atan2(-state.facing.x, -state.facing.z));
+ return {
+ entity: "actor",
+ actorId: realtimeActorId,
+ kind: state.actor === "anonymous-dog" ? "dog" : "humanoid",
+ sequence: 0,
+ timestampMs,
+ pose: {
+ space: "local",
+ cell: { kind: "office", officeId },
+ xM: state.position.x,
+ yM: office?.plan.level(state.levelId)?.floorY ?? 0,
+ zM: state.position.z,
+ headingDeg,
+ pitchDeg: 0,
+ },
+ velocity: {
+ xMps: state.action.x * 1.6,
+ yMps: 0,
+ zMps: state.action.z * 1.6,
+ yawDegPerSec: 0,
+ },
+ };
+ }
+
+ const vehicle = journey.vehicle !== null ? city?.vehicleState() : null;
+ if (vehicle) {
+ const heading = vehicle.headingDeg * Math.PI / 180;
+ return {
+ entity: "vehicle",
+ vehicleId: realtimeVehicleId,
+ kind: "model-x",
+ driverActorId: realtimeActorId,
+ sequence: 0,
+ timestampMs,
+ pose: {
+ space: "geographic",
+ lat: vehicle.lat,
+ lng: vehicle.lng,
+ altitudeM: 0,
+ headingDeg: vehicle.headingDeg,
+ pitchDeg: 0,
+ },
+ velocity: {
+ xMps: Math.sin(heading) * vehicle.speedMps,
+ yMps: 0,
+ zMps: Math.cos(heading) * vehicle.speedMps,
+ yawDegPerSec: 0,
+ },
+ steering: vehicle.steering,
+ wheelRadians: vehicle.wheelRadians,
+ };
+ }
+
+ const actor = city?.actorState();
+ if (!actor || !city) return null;
+ const headingDeg = -degrees(actor.yaw);
+ const velocity = {
+ xMps: -Math.sin(actor.yaw) * actor.speedMps,
+ yMps: actor.verticalSpeedMps,
+ // Geographic velocity is east/up/north; scene +Z is south.
+ zMps: Math.cos(actor.yaw) * actor.speedMps,
+ yawDegPerSec: 0,
+ };
+ const origin = cityId === "california" ? { lat: 35.5, lng: -119.5 } : city.world.city.center;
+ const [originX, originZ] = city.world.project(origin.lat, origin.lng);
+ const [lat, lng] = city.world.unproject(
+ originX + actor.x / city.world.metresPerUnit,
+ originZ + actor.z / city.world.metresPerUnit,
+ );
+ return {
+ entity: "actor",
+ actorId: realtimeActorId,
+ kind: actor.kind,
+ sequence: 0,
+ timestampMs,
+ pose: {
+ space: "geographic",
+ lat,
+ lng,
+ altitudeM: actor.y,
+ headingDeg,
+ pitchDeg: degrees(actor.pitch),
+ },
+ velocity,
+ };
+}
+
+function ownRealtimeEntity(snapshot: EntityPoseSnapshot): boolean {
+ if (snapshot.entity === "actor") return snapshot.actorId === realtimeClient?.state().actorId;
+ return snapshot.driverActorId === realtimeClient?.state().actorId || snapshot.vehicleId === realtimeVehicleId;
+}
+
+function clearRealtimePeers(): void {
+ city?.clearRemoteEntities();
+ office?.clearRemoteEntities();
+ updatePresenceIndicator();
+}
+
+function applyRealtimeMessage(message: ServerRealtimeMessage): void {
+ if (message.type === "membership-revoked") {
+ clearRealtimePeers();
+ return;
+ }
+ const updates = message.type === "pose-delta"
+ ? message.updates
+ : message.type === "join-grant"
+ ? message.initial
+ : message.type === "resume-grant"
+ ? message.snapshot
+ : [];
+ const target = inside ? office : city;
+ for (const snapshot of updates) {
+ if (!ownRealtimeEntity(snapshot)) target?.upsertRemoteSnapshot(snapshot);
+ }
+ if (message.type === "pose-delta") {
+ for (const id of message.removedEntityIds) {
+ if (id !== `actor:${realtimeClient?.state().actorId}` && id !== `vehicle:${realtimeVehicleId}`) {
+ target?.removeRemoteEntity(id);
+ }
+ }
+ }
+ updatePresenceIndicator();
+}
+
+function updatePresenceIndicator(state = realtimeClient?.state()): void {
+ if (!presenceIndicator || access.subject === null) return;
+ const connection: "connecting" | "live" | "reconnecting" | "offline" =
+ state?.status === "streaming" ? "live"
+ : state?.status === "reconnecting" ? "reconnecting"
+ : state?.status === "left" || state?.status === "disposed" ? "offline"
+ : "connecting";
+ const nearbyPeerCount = connection === "live"
+ ? (inside ? office?.remoteEntityCount() : city?.remoteEntityCount()) ?? 0
+ : 0;
+ presenceIndicator.update({ signedIn: true, connection, nearbyPeerCount });
+}
+
+async function joinRealtimePresence(operation: number): Promise {
+ const client = realtimeClient;
+ const pose = realtimePose();
+ if (!client || !pose) return;
+ try {
+ const grant = await client.join(realtimeInterest(), pose);
+ if (operation !== realtimeOperation) return;
+ applyRealtimeMessage(grant);
+ } catch {
+ if (operation === realtimeOperation) {
+ presenceIndicator?.update({ signedIn: true, connection: "offline", nearbyPeerCount: 0 });
+ }
+ }
+}
+
+function moveRealtimePresence(): void {
+ clearRealtimePeers();
+ const client = realtimeClient;
+ if (!client || access.subject === null) return;
+ const operation = ++realtimeOperation;
+ if (client.state().sessionId === null) {
+ void joinRealtimePresence(operation);
+ return;
+ }
+ updatePresenceIndicator({ ...client.state(), status: "reconnecting" });
+ void client.moveInterest(realtimeInterest())
+ .then((grant) => {
+ if (operation === realtimeOperation) applyRealtimeMessage(grant);
+ })
+ .catch(() => {
+ if (operation === realtimeOperation) {
+ presenceIndicator?.update({ signedIn: true, connection: "offline", nearbyPeerCount: 0 });
+ }
+ });
+}
+
+function publishRealtimePresence(now: number): void {
+ if (!realtimeClient || now - lastRealtimePublishAt < 100) return;
+ const pose = realtimePose();
+ if (!pose || realtimeClient.state().sessionId === null) return;
+ lastRealtimePublishAt = now;
+ try {
+ realtimeClient.publishPose(pose);
+ } catch {
+ presenceIndicator?.update({ signedIn: true, connection: "offline", nearbyPeerCount: 0 });
+ }
+}
+
+async function initializeRealtimePresence(): Promise {
+ if (!realtimePageActive || access.subject === null || !presenceHost || realtimeClient) return;
+ const [{ createRealtimeClient }, { createPresenceIndicator }] = await Promise.all([
+ import("./realtime/client.ts"),
+ import("./realtime/presenceIndicator.ts"),
+ ]);
+ // `pagehide` can overtake the lazy chunks. Never construct a fresh client
+ // after the page has already run its terminal cleanup.
+ if (!realtimePageActive || access.subject === null || realtimeClient) return;
+ presenceHost.hidden = false;
+ document.body.classList.add("presence-on");
+ presenceIndicator = createPresenceIndicator({
+ container: presenceHost,
+ onRetry: () => moveRealtimePresence(),
+ });
+ presenceIndicator.update({ signedIn: true, connection: "connecting", nearbyPeerCount: 0 });
+ realtimeClient = createRealtimeClient({
+ actorId: realtimeActorId,
+ authenticatedFetch: authFetch,
+ sendIntervalMs: 100,
+ });
+ stopRealtimeSubscription = realtimeClient.subscribe({
+ onMessage: applyRealtimeMessage,
+ onStateChange: updatePresenceIndicator,
+ onError: () => presenceIndicator?.update({ signedIn: true, connection: "offline", nearbyPeerCount: 0 }),
+ });
+ moveRealtimePresence();
+}
+
+window.addEventListener("pagehide", () => {
+ realtimePageActive = false;
+ realtimeOperation += 1;
+ stopRealtimeSubscription?.();
+ stopRealtimeSubscription = null;
+ clearRealtimePeers();
+ // Page teardown cannot await the leave round trip. Contain a rotated-token
+ // or offline failure so navigation never produces an unhandled rejection;
+ // the server's disconnect grace still removes the ephemeral session.
+ void realtimeClient?.dispose().catch(() => undefined);
+ realtimeClient = null;
+ presenceIndicator?.dispose();
+ presenceIndicator = null;
+});
+
function applyProfilePreview(profile: LocalProfile): void {
if (access.subject === null) return;
if (inside) {
@@ -3165,6 +3440,12 @@ async function boot() {
mountCity(first?.id ?? "california"),
);
+ // There must be a mounted scene before remote snapshots have anywhere to go.
+ // Anonymous and zero-server builds retain their existing local-only path.
+ // Hosted presence is an enhancement: a failed lazy chunk or unavailable API
+ // must not stop profile/tools initialization after the world is already live.
+ await initializeRealtimePresence().catch(() => undefined);
+
// The instruments, after the first board, because the panel reads a live
// stage and there is not one before this line.
await mountGodmode();
diff --git a/src/realtime/client.ts b/src/realtime/client.ts
new file mode 100644
index 0000000..7930f4e
--- /dev/null
+++ b/src/realtime/client.ts
@@ -0,0 +1,567 @@
+import {
+ isEntityPoseSnapshot,
+ isInterestCell,
+ validateClientRealtimeMessage,
+ validateServerRealtimeMessage,
+} from "./protocol.ts";
+import type {
+ EntityPoseSnapshot,
+ InterestCell,
+ JoinGrant,
+ JoinRequest,
+ ResumeGrant,
+ ResumeRequest,
+ ServerRealtimeMessage,
+} from "./types.ts";
+
+const MIN_SEND_INTERVAL_MS = 1_000 / 15;
+const MAX_SEND_INTERVAL_MS = 1_000 / 10;
+const UINT32_MAX = 4_294_967_295;
+
+type Fetch = typeof globalThis.fetch;
+type TimerHandle = ReturnType;
+
+export interface RealtimeClientTimers {
+ setTimeout(callback: () => void, delayMs: number): TimerHandle;
+ clearTimeout(handle: TimerHandle): void;
+}
+
+export interface RealtimeClientEndpoints {
+ join: string;
+ pose: string;
+ events: string;
+ leave: string;
+}
+
+export type RealtimeClientStatus =
+ | "idle"
+ | "joined"
+ | "streaming"
+ | "reconnecting"
+ | "left"
+ | "disposed";
+
+export interface RealtimeClientState {
+ status: RealtimeClientStatus;
+ sessionId: string | null;
+ actorId: string;
+ interests: readonly InterestCell[];
+ serverEpoch: string | null;
+ lastReceivedSequence: number | null;
+ nextPoseSequence: number;
+ posePending: boolean;
+ subscribed: boolean;
+ reconnectAttempt: number;
+}
+
+export interface RealtimeSubscriptionCallbacks {
+ onMessage?(message: ServerRealtimeMessage): void;
+ onStateChange?(state: RealtimeClientState): void;
+ /** Receives protocol/parser errors without the untrusted payload or credentials. */
+ onMalformedEvent?(error: Error): void;
+ onError?(error: Error): void;
+}
+
+export interface RealtimeClientOptions {
+ actorId: string;
+ /**
+ * Optional SSO bearer, kept only in this closure and never placed in a URL.
+ * Omit it when the deployment uses an HttpOnly same-origin session cookie.
+ */
+ accessToken?: string;
+ /**
+ * Preferred request adapter for hosted builds. Passing `authFetch` keeps SSO
+ * token lookup at request time and also supports HttpOnly cookie sessions.
+ */
+ authenticatedFetch?: Fetch;
+ endpoints?: Partial;
+ fetch?: Fetch;
+ timers?: RealtimeClientTimers;
+ sendIntervalMs?: number;
+ reconnectBaseMs?: number;
+ reconnectMaximumMs?: number;
+ reconnectJitter?: number;
+ random?: () => number;
+}
+
+export interface RealtimeClient {
+ join(
+ interest: InterestCell | readonly InterestCell[],
+ entity: EntityPoseSnapshot,
+ ): Promise;
+ publishPose(entity: EntityPoseSnapshot): void;
+ subscribe(callbacks: RealtimeSubscriptionCallbacks): () => void;
+ moveInterest(interest: InterestCell | readonly InterestCell[]): Promise;
+ rejoin(): Promise;
+ state(): RealtimeClientState;
+ leave(): Promise;
+ dispose(): Promise;
+}
+
+const DEFAULT_ENDPOINTS: RealtimeClientEndpoints = {
+ join: "/api/v1/realtime/join",
+ pose: "/api/v1/realtime/pose",
+ events: "/api/v1/realtime/events",
+ leave: "/api/v1/realtime/leave",
+};
+
+const DEFAULT_TIMERS: RealtimeClientTimers = {
+ setTimeout: (callback, delayMs) => globalThis.setTimeout(callback, delayMs),
+ clearTimeout: (handle) => globalThis.clearTimeout(handle),
+};
+
+function requireNonEmpty(value: string, name: string): void {
+ if (value.length === 0 || value.length > 256) throw new TypeError(`${name} must be 1-256 characters`);
+}
+
+function normalizeInterests(value: InterestCell | readonly InterestCell[]): readonly InterestCell[] {
+ const values = Array.isArray(value) ? value : [value];
+ if (values.length === 0 || values.length > 128 || !values.every(isInterestCell)) {
+ throw new TypeError("realtime interests are invalid");
+ }
+ return values.map((interest) => ({ ...interest }));
+}
+
+function errorFrom(reason: unknown, fallback: string): Error {
+ return reason instanceof Error ? reason : new Error(fallback);
+}
+
+function isTerminalStatus(status: RealtimeClientStatus): boolean {
+ return status === "disposed" || status === "left";
+}
+
+/**
+ * Browser transport for the server-authoritative realtime protocol.
+ *
+ * Streaming uses fetch + ReadableStream instead of EventSource. This permits a
+ * normal Authorization header and POST body, keeping both the access token and
+ * resume token out of URLs, referrers, history, and intermediary URL logs.
+ */
+export function createRealtimeClient(options: RealtimeClientOptions): RealtimeClient {
+ requireNonEmpty(options.actorId, "actorId");
+ if (options.accessToken !== undefined) requireNonEmpty(options.accessToken, "accessToken");
+ if (options.fetch && options.authenticatedFetch) {
+ throw new TypeError("provide either fetch or authenticatedFetch, not both");
+ }
+ const fetcher = options.authenticatedFetch ?? options.fetch ?? globalThis.fetch;
+ if (typeof fetcher !== "function") throw new TypeError("fetch is required");
+ const timers = options.timers ?? DEFAULT_TIMERS;
+ const endpoints = { ...DEFAULT_ENDPOINTS, ...options.endpoints };
+ for (const [name, endpoint] of Object.entries(endpoints)) requireNonEmpty(endpoint, `${name} endpoint`);
+
+ const sendIntervalMs = options.sendIntervalMs ?? MAX_SEND_INTERVAL_MS;
+ if (!Number.isFinite(sendIntervalMs) || sendIntervalMs < MIN_SEND_INTERVAL_MS || sendIntervalMs > MAX_SEND_INTERVAL_MS) {
+ throw new RangeError("sendIntervalMs must be between 66.67ms (15Hz) and 100ms (10Hz)");
+ }
+ const reconnectBaseMs = options.reconnectBaseMs ?? 500;
+ const reconnectMaximumMs = options.reconnectMaximumMs ?? 30_000;
+ const reconnectJitter = options.reconnectJitter ?? 0.2;
+ if (!Number.isFinite(reconnectBaseMs) || reconnectBaseMs <= 0) throw new RangeError("reconnectBaseMs must be positive");
+ if (!Number.isFinite(reconnectMaximumMs) || reconnectMaximumMs < reconnectBaseMs) {
+ throw new RangeError("reconnectMaximumMs must be at least reconnectBaseMs");
+ }
+ if (!Number.isFinite(reconnectJitter) || reconnectJitter < 0 || reconnectJitter > 1) {
+ throw new RangeError("reconnectJitter must be between zero and one");
+ }
+ const random = options.random ?? Math.random;
+
+ let status: RealtimeClientStatus = "idle";
+ let interests: readonly InterestCell[] = [];
+ let sessionId: string | null = null;
+ let serverEpoch: string | null = null;
+ let resumeToken: string | null = null;
+ let lastReceivedSequence: number | null = null;
+ let nextPoseSequence = 0;
+ let pendingPose: EntityPoseSnapshot | null = null;
+ let poseTimer: TimerHandle | null = null;
+ let reconnectTimer: TimerHandle | null = null;
+ let reconnectAttempt = 0;
+ let subscription: RealtimeSubscriptionCallbacks | null = null;
+ let streamAbort: AbortController | null = null;
+ // Resume rotates the credential. Do not race a pose against the first grant
+ // and send the token the server has just retired.
+ let streamReady = false;
+ let streamGeneration = 0;
+ // Scene changes can overtake join/resume HTTP calls. Only the newest
+ // transition may commit credentials or interest cells.
+ let transitionGeneration = 0;
+ let requestCounter = 0;
+
+ function snapshot(): RealtimeClientState {
+ return {
+ status,
+ sessionId,
+ actorId: options.actorId,
+ interests: interests.map((interest) => ({ ...interest })),
+ serverEpoch,
+ lastReceivedSequence,
+ nextPoseSequence,
+ posePending: pendingPose !== null,
+ subscribed: subscription !== null,
+ reconnectAttempt,
+ };
+ }
+
+ function changed(): void {
+ subscription?.onStateChange?.(snapshot());
+ }
+
+ function requestId(): string {
+ requestCounter += 1;
+ return `tera-client-${requestCounter}`;
+ }
+
+ function authHeaders(accept = "application/json"): HeadersInit {
+ const headers: Record = {
+ Accept: accept,
+ "Content-Type": "application/json",
+ };
+ if (options.accessToken !== undefined) headers.Authorization = `Bearer ${options.accessToken}`;
+ return headers;
+ }
+
+ async function post(endpoint: string, body: unknown, accept?: string, signal?: AbortSignal): Promise {
+ return fetcher(endpoint, {
+ method: "POST",
+ credentials: "same-origin",
+ cache: "no-store",
+ redirect: "error",
+ headers: authHeaders(accept),
+ body: JSON.stringify(body),
+ signal,
+ });
+ }
+
+ async function json(response: Response): Promise {
+ if (!response.ok) throw new Error(`realtime request failed (${response.status})`);
+ return response.json() as Promise;
+ }
+
+ function cancelPose(): void {
+ if (poseTimer !== null) timers.clearTimeout(poseTimer);
+ poseTimer = null;
+ pendingPose = null;
+ }
+
+ function cancelStream(): void {
+ streamGeneration += 1;
+ streamAbort?.abort();
+ streamAbort = null;
+ streamReady = false;
+ if (reconnectTimer !== null) timers.clearTimeout(reconnectTimer);
+ reconnectTimer = null;
+ }
+
+ function acceptMessage(message: ServerRealtimeMessage): boolean {
+ if (serverEpoch !== null && "serverEpoch" in message && message.serverEpoch !== serverEpoch) {
+ subscription?.onMalformedEvent?.(new Error("realtime event belongs to a different server epoch"));
+ return false;
+ }
+ if (message.type === "pose-delta" || message.type === "membership-revoked") {
+ if (lastReceivedSequence !== null && message.sequence <= lastReceivedSequence) {
+ subscription?.onMalformedEvent?.(new Error("realtime event sequence is not monotonic"));
+ return false;
+ }
+ lastReceivedSequence = message.sequence;
+ } else if (message.type === "resume-grant") {
+ resumeToken = message.resumeToken;
+ serverEpoch = message.serverEpoch;
+ lastReceivedSequence = message.nextSequence === 0 ? 0 : message.nextSequence - 1;
+ streamReady = true;
+ status = "streaming";
+ }
+ reconnectAttempt = 0;
+ subscription?.onMessage?.(message);
+ if (message.type === "membership-revoked" && !message.reconnectAllowed) {
+ cancelStream();
+ status = "left";
+ resumeToken = null;
+ }
+ changed();
+ return true;
+ }
+
+ function parseEvent(data: string): void {
+ if (data.length === 0) return;
+ let raw: unknown;
+ try {
+ raw = JSON.parse(data);
+ } catch {
+ subscription?.onMalformedEvent?.(new Error("realtime event is not valid JSON"));
+ return;
+ }
+ const result = validateServerRealtimeMessage(raw);
+ if (!result.ok) {
+ subscription?.onMalformedEvent?.(new Error(result.error));
+ return;
+ }
+ acceptMessage(result.value);
+ }
+
+ async function readSse(response: Response, generation: number): Promise {
+ if (!response.ok) throw new Error(`realtime stream failed (${response.status})`);
+ if (!response.body) throw new Error("realtime stream response has no body");
+ const reader = response.body.getReader();
+ const decoder = new TextDecoder();
+ let buffer = "";
+ let dataLines: string[] = [];
+ const line = (value: string): void => {
+ if (value === "") {
+ if (dataLines.length > 0) parseEvent(dataLines.join("\n"));
+ dataLines = [];
+ } else if (value.startsWith("data:")) {
+ dataLines.push(value.slice(5).replace(/^ /, ""));
+ }
+ };
+ try {
+ while (generation === streamGeneration) {
+ const chunk = await reader.read();
+ // An abort and an already-queued chunk may settle together. Never
+ // parse the old cell after another transition has won that race.
+ if (generation !== streamGeneration) break;
+ if (chunk.done) break;
+ buffer += decoder.decode(chunk.value, { stream: true }).replace(/\r\n/g, "\n").replace(/\r/g, "\n");
+ let newline = buffer.indexOf("\n");
+ while (newline >= 0) {
+ line(buffer.slice(0, newline));
+ buffer = buffer.slice(newline + 1);
+ newline = buffer.indexOf("\n");
+ }
+ }
+ if (generation !== streamGeneration) return;
+ buffer += decoder.decode();
+ if (buffer.length > 0) line(buffer);
+ if (dataLines.length > 0) parseEvent(dataLines.join("\n"));
+ } finally {
+ reader.releaseLock();
+ }
+ }
+
+ function reconnectDelay(): number {
+ const exponential = Math.min(reconnectMaximumMs, reconnectBaseMs * 2 ** reconnectAttempt);
+ const factor = 1 + (random() * 2 - 1) * reconnectJitter;
+ return Math.max(0, exponential * factor);
+ }
+
+ function scheduleReconnect(): void {
+ if (!subscription || status === "left" || status === "disposed" || reconnectTimer !== null) return;
+ status = "reconnecting";
+ const delay = reconnectDelay();
+ reconnectAttempt += 1;
+ changed();
+ reconnectTimer = timers.setTimeout(() => {
+ reconnectTimer = null;
+ void startStream();
+ }, delay);
+ }
+
+ async function startStream(): Promise {
+ if (!subscription || !sessionId || !resumeToken || !serverEpoch || lastReceivedSequence === null) return;
+ cancelStream();
+ const generation = streamGeneration;
+ const abort = new AbortController();
+ streamAbort = abort;
+ const resume: ResumeRequest = {
+ type: "resume-request",
+ protocolVersion: 1,
+ requestId: requestId(),
+ sessionId,
+ resumeToken,
+ serverEpoch,
+ lastReceivedSequence,
+ interests,
+ };
+ if (!validateClientRealtimeMessage(resume).ok) throw new Error("realtime resume state is invalid");
+ try {
+ const response = await post(endpoints.events, resume, "text/event-stream", abort.signal);
+ if (generation !== streamGeneration) return;
+ await readSse(response, generation);
+ if (generation === streamGeneration) scheduleReconnect();
+ } catch (reason) {
+ if (generation !== streamGeneration || abort.signal.aborted) return;
+ subscription?.onError?.(errorFrom(reason, "realtime stream failed"));
+ scheduleReconnect();
+ }
+ }
+
+ async function sendPose(): Promise {
+ poseTimer = null;
+ const pose = pendingPose;
+ pendingPose = null;
+ if (!pose || !sessionId || status === "left" || status === "disposed") return;
+ if (subscription && !streamReady) {
+ pendingPose = pose;
+ poseTimer = timers.setTimeout(() => void sendPose(), sendIntervalMs);
+ return;
+ }
+ try {
+ const response = await post(endpoints.pose, { sessionId, token: resumeToken, snapshot: pose });
+ if (!response.ok) throw new Error(`realtime pose request failed (${response.status})`);
+ } catch (reason) {
+ subscription?.onError?.(errorFrom(reason, "realtime pose request failed"));
+ }
+ changed();
+ }
+
+ async function join(interest: InterestCell | readonly InterestCell[], entity: EntityPoseSnapshot): Promise {
+ if (status === "disposed") throw new Error("realtime client is disposed");
+ const generation = ++transitionGeneration;
+ const nextInterests = normalizeInterests(interest);
+ if (!isEntityPoseSnapshot(entity)) throw new TypeError("realtime entity pose is invalid");
+ const request: JoinRequest = {
+ type: "join-request",
+ protocolVersion: 1,
+ requestId: requestId(),
+ actorId: options.actorId,
+ resumeToken,
+ lastReceivedSequence,
+ interests: nextInterests,
+ };
+ if (!validateClientRealtimeMessage(request).ok) throw new Error("realtime join request is invalid");
+ const parsed = validateServerRealtimeMessage(await json(await post(endpoints.join, { ...request, entity })));
+ if (!parsed.ok || parsed.value.type !== "join-grant") throw new Error("realtime join response is invalid");
+ const grant = parsed.value;
+ if (grant.actorId !== options.actorId || grant.requestId !== request.requestId) {
+ throw new Error("realtime join response identity does not match request");
+ }
+ if (generation !== transitionGeneration || isTerminalStatus(status)) {
+ // A superseded request can still have created a server session before
+ // the browser ignored its answer. Retire it without adopting its token.
+ void post(endpoints.leave, { sessionId: grant.sessionId, token: grant.resumeToken })
+ .catch(() => { /* normal idle expiry is the fallback */ });
+ throw new Error("realtime join was superseded");
+ }
+ interests = grant.interests.map((entry) => ({ ...entry }));
+ sessionId = grant.sessionId;
+ serverEpoch = grant.serverEpoch;
+ resumeToken = grant.resumeToken;
+ lastReceivedSequence = grant.nextSequence === 0 ? 0 : grant.nextSequence - 1;
+ nextPoseSequence = Math.max(entity.sequence + 1, grant.nextSequence);
+ status = "joined";
+ streamReady = subscription === null;
+ reconnectAttempt = 0;
+ changed();
+ if (subscription && lastReceivedSequence !== null) void startStream();
+ return grant;
+ }
+
+ function publishPose(entity: EntityPoseSnapshot): void {
+ if (!sessionId || status === "idle" || status === "left" || status === "disposed") {
+ throw new Error("realtime client must be joined before publishing a pose");
+ }
+ if (!isEntityPoseSnapshot(entity)) throw new TypeError("realtime entity pose is invalid");
+ if (nextPoseSequence > UINT32_MAX) throw new RangeError("realtime pose sequence exhausted");
+ const sequenced = { ...entity, sequence: nextPoseSequence } as EntityPoseSnapshot;
+ if (!isEntityPoseSnapshot(sequenced)) throw new TypeError("realtime entity pose is invalid");
+ nextPoseSequence += 1;
+ pendingPose = sequenced;
+ if (poseTimer === null) poseTimer = timers.setTimeout(() => void sendPose(), sendIntervalMs);
+ changed();
+ }
+
+ function subscribe(callbacks: RealtimeSubscriptionCallbacks): () => void {
+ if (status === "disposed") throw new Error("realtime client is disposed");
+ subscription = callbacks;
+ changed();
+ if (sessionId && lastReceivedSequence !== null) void startStream();
+ let active = true;
+ return () => {
+ if (!active || subscription !== callbacks) return;
+ active = false;
+ subscription = null;
+ cancelStream();
+ if (status === "streaming" || status === "reconnecting") status = sessionId ? "joined" : "idle";
+ };
+ }
+
+ async function resume(nextInterests: readonly InterestCell[]): Promise {
+ if (status === "disposed") throw new Error("realtime client is disposed");
+ if (!sessionId || !resumeToken || !serverEpoch || lastReceivedSequence === null) {
+ throw new Error("realtime client has no resumable session");
+ }
+ const generation = ++transitionGeneration;
+ const expectedSessionId = sessionId;
+ const request: ResumeRequest = {
+ type: "resume-request",
+ protocolVersion: 1,
+ requestId: requestId(),
+ sessionId,
+ resumeToken,
+ serverEpoch,
+ lastReceivedSequence,
+ interests: nextInterests,
+ };
+ if (!validateClientRealtimeMessage(request).ok) throw new Error("realtime resume request is invalid");
+ // The old stream belongs to a different privacy and coordinate cell. Close
+ // it before the handoff request, not after its network round trip.
+ cancelStream();
+ status = "joined";
+ changed();
+ try {
+ const parsed = validateServerRealtimeMessage(await json(await post(endpoints.join, request)));
+ if (!parsed.ok || parsed.value.type !== "resume-grant") throw new Error("realtime resume response is invalid");
+ const grant = parsed.value;
+ if (grant.sessionId !== expectedSessionId || grant.requestId !== request.requestId) {
+ throw new Error("realtime resume response identity does not match request");
+ }
+ if (generation !== transitionGeneration || isTerminalStatus(status)) {
+ throw new Error("realtime resume was superseded");
+ }
+ interests = nextInterests.map((entry) => ({ ...entry }));
+ resumeToken = grant.resumeToken;
+ serverEpoch = grant.serverEpoch;
+ lastReceivedSequence = grant.nextSequence === 0 ? 0 : grant.nextSequence - 1;
+ streamReady = subscription === null;
+ reconnectAttempt = 0;
+ changed();
+ if (subscription && lastReceivedSequence !== null) void startStream();
+ return grant;
+ } catch (error) {
+ if (generation === transitionGeneration && !isTerminalStatus(status)) {
+ status = "joined";
+ changed();
+ }
+ throw error;
+ }
+ }
+
+ async function rejoin(): Promise {
+ return resume(interests);
+ }
+
+ async function moveInterest(interest: InterestCell | readonly InterestCell[]): Promise {
+ return resume(normalizeInterests(interest));
+ }
+
+ async function leave(): Promise {
+ if (status === "disposed") return;
+ transitionGeneration += 1;
+ cancelPose();
+ cancelStream();
+ const leavingSession = sessionId;
+ const leavingToken = resumeToken;
+ status = "left";
+ sessionId = null;
+ serverEpoch = null;
+ resumeToken = null;
+ lastReceivedSequence = null;
+ reconnectAttempt = 0;
+ changed();
+ if (leavingSession && leavingToken) {
+ const response = await post(endpoints.leave, { sessionId: leavingSession, token: leavingToken });
+ if (!response.ok) throw new Error(`realtime leave request failed (${response.status})`);
+ }
+ }
+
+ async function dispose(): Promise {
+ if (status === "disposed") return;
+ try {
+ await leave();
+ } finally {
+ subscription = null;
+ status = "disposed";
+ }
+ }
+
+ return { join, publishPose, subscribe, moveInterest, rejoin, state: snapshot, leave, dispose };
+}
diff --git a/src/realtime/index.ts b/src/realtime/index.ts
index 7dd6bda..f4c7f41 100644
--- a/src/realtime/index.ts
+++ b/src/realtime/index.ts
@@ -8,4 +8,27 @@ export {
validateServerRealtimeMessage,
} from "./protocol.ts";
export { PoseInterpolationBuffer } from "./interpolation.ts";
+export {
+ createRealtimeClient,
+ type RealtimeClient,
+ type RealtimeClientEndpoints,
+ type RealtimeClientOptions,
+ type RealtimeClientState,
+ type RealtimeClientStatus,
+ type RealtimeClientTimers,
+ type RealtimeSubscriptionCallbacks,
+} from "./client.ts";
+export {
+ createScenePeers,
+ scenePeerId,
+ type ScenePeers,
+ type ScenePeersOptions,
+} from "./scenePeers.ts";
+export {
+ createPresenceIndicator,
+ type PresenceConnectionState,
+ type PresenceIndicator,
+ type PresenceIndicatorOptions,
+ type PresenceIndicatorState,
+} from "./presenceIndicator.ts";
export type * from "./types.ts";
diff --git a/src/realtime/presenceIndicator.ts b/src/realtime/presenceIndicator.ts
new file mode 100644
index 0000000..c12f7a3
--- /dev/null
+++ b/src/realtime/presenceIndicator.ts
@@ -0,0 +1,121 @@
+/** Small, identity-free hosted-presence status for a caller-owned container. */
+
+export type PresenceConnectionState = "off" | "connecting" | "live" | "reconnecting" | "offline";
+
+export interface PresenceIndicatorState {
+ signedIn: boolean;
+ connection: PresenceConnectionState;
+ /** Aggregate only. Individual peer identity never enters this adapter. */
+ nearbyPeerCount: number;
+}
+
+export interface PresenceIndicatorOptions {
+ container: HTMLElement;
+ onRetry?: () => void;
+}
+
+export interface PresenceIndicator {
+ root: HTMLElement;
+ update(state: PresenceIndicatorState): PresenceIndicatorState;
+ state(): PresenceIndicatorState;
+ dispose(): void;
+}
+
+const CONNECTIONS = new Set([
+ "off", "connecting", "live", "reconnecting", "offline",
+]);
+
+function copy(state: PresenceIndicatorState): PresenceIndicatorState {
+ return { ...state };
+}
+
+function normalize(input: PresenceIndicatorState): PresenceIndicatorState {
+ if (!input || typeof input !== "object") throw new TypeError("presence indicator: invalid state");
+ if (typeof input.signedIn !== "boolean" || !CONNECTIONS.has(input.connection)) {
+ throw new TypeError("presence indicator: invalid connection state");
+ }
+ if (!Number.isSafeInteger(input.nearbyPeerCount) || input.nearbyPeerCount < 0 || input.nearbyPeerCount > 10_000) {
+ throw new RangeError("presence indicator: nearbyPeerCount must be an integer from 0 to 10000");
+ }
+ if (!input.signedIn) return { signedIn: false, connection: "off", nearbyPeerCount: 0 };
+ return {
+ signedIn: true,
+ connection: input.connection,
+ nearbyPeerCount: input.connection === "live" ? input.nearbyPeerCount : 0,
+ };
+}
+
+function label(state: PresenceIndicatorState): string {
+ if (!state.signedIn || state.connection === "off") return "Hosted presence is off.";
+ switch (state.connection) {
+ case "connecting": return "Hosted presence is connecting.";
+ case "reconnecting": return "Hosted presence is reconnecting.";
+ case "offline": return "Hosted presence is offline.";
+ case "live": {
+ const nearby = state.nearbyPeerCount === 1 ? "1 nearby peer" : `${state.nearbyPeerCount} nearby peers`;
+ return `Hosted presence is live · ${nearby}.`;
+ }
+ }
+}
+
+export function createPresenceIndicator(options: PresenceIndicatorOptions): PresenceIndicator {
+ if (!options.container || typeof options.container.append !== "function") {
+ throw new TypeError("presence indicator: container must be an HTMLElement");
+ }
+ if (options.onRetry !== undefined && typeof options.onRetry !== "function") {
+ throw new TypeError("presence indicator: onRetry must be a function");
+ }
+
+ const doc = options.container.ownerDocument;
+ const root = doc.createElement("div");
+ root.className = "tera-presence";
+ root.setAttribute("data-presence", "off");
+
+ const statusLine = doc.createElement("span");
+ statusLine.className = "tera-presence__status";
+ statusLine.setAttribute("role", "status");
+ statusLine.setAttribute("aria-live", "polite");
+ statusLine.setAttribute("aria-atomic", "true");
+
+ const retry = doc.createElement("button");
+ retry.type = "button";
+ retry.className = "tera-presence__retry";
+ retry.textContent = "Retry presence";
+ retry.hidden = true;
+ root.append(statusLine, retry);
+ options.container.append(root);
+
+ let current: PresenceIndicatorState = { signedIn: false, connection: "off", nearbyPeerCount: 0 };
+ let disposed = false;
+ let onRetry: (() => void) | null = options.onRetry ?? null;
+
+ function render(): void {
+ statusLine.textContent = label(current);
+ root.setAttribute("data-presence", current.connection);
+ retry.hidden = onRetry === null || !current.signedIn ||
+ (current.connection !== "offline" && current.connection !== "reconnecting");
+ }
+
+ retry.addEventListener("click", () => {
+ if (!disposed && !retry.hidden) onRetry?.();
+ });
+ render();
+
+ return {
+ root,
+ update(next) {
+ if (disposed) return copy(current);
+ current = normalize(next);
+ render();
+ return copy(current);
+ },
+ state: () => copy(current),
+ dispose() {
+ if (disposed) return;
+ disposed = true;
+ onRetry = null;
+ current = { signedIn: false, connection: "off", nearbyPeerCount: 0 };
+ root.remove();
+ },
+ };
+}
diff --git a/src/realtime/scenePeers.ts b/src/realtime/scenePeers.ts
new file mode 100644
index 0000000..49e9bfd
--- /dev/null
+++ b/src/realtime/scenePeers.ts
@@ -0,0 +1,359 @@
+/**
+ * Three.js presentation for already-validated remote entity snapshots.
+ *
+ * Networking, trust, auth, and display names stay outside this module. It keeps
+ * only canonical protocol ids in a private Map, buffers authoritative motion,
+ * and gives each peer one stable, identity-free scene root. Procedural clones
+ * share four prototype resource sets for the lifetime of the adapter.
+ */
+import * as THREE from "three";
+import {
+ buildCrow,
+ buildDog,
+ buildHumanoid,
+ cloneCrow,
+ cloneDog,
+ cloneHumanoid,
+ disposeCrow,
+ disposeDog,
+ disposeHumanoid,
+ poseCrowFlight,
+ poseDogAttention,
+ poseDogWalk,
+ poseHumanoid,
+ type CrowRig,
+ type DogRig,
+ type HumanoidRig,
+} from "../assets/actors/index.ts";
+import {
+ buildModelX,
+ cloneModelX,
+ disposeModelX,
+ setModelXSteering,
+ setModelXWheelRotation,
+ type ModelXRig,
+} from "../assets/vehicles/index.ts";
+import { PoseInterpolationBuffer } from "./interpolation.ts";
+import { isEntityPoseSnapshot } from "./protocol.ts";
+import type {
+ BufferedPoseSample,
+ EntityPoseSnapshot,
+ InterpolationOptions,
+ SpatialPose,
+} from "./types.ts";
+
+export interface ScenePeersOptions {
+ /** Geographic map projection. Returned X/Z are already scene units. */
+ project(lat: number, lng: number): readonly [number, number];
+ /** Terrain height in scene units at a geographic position. */
+ groundAt(lat: number, lng: number): number;
+ /** Converts geographic altitude metres and, by default, asset metres to scene units. */
+ geographicSceneUnitsPerMetre?: number;
+ /** Optional presentation exaggeration independent of geographic motion scale. */
+ geographicVisualScale?: number;
+ /** Local office/city metre scale. Defaults to direct 1 metre = 1 scene unit. */
+ localSceneUnitsPerMetre?: number;
+ /** Optional presentation scale independent of local coordinate scale. */
+ localVisualScale?: number;
+ interpolation?: InterpolationOptions;
+ /** Bounded nearby peer count. Defaults to 128, maximum 512. */
+ maximumPeers?: number;
+}
+
+export interface ScenePeers {
+ root: THREE.Group;
+ /** Add a new authoritative sample. Invalid/stale/over-capacity input returns false. */
+ upsert(snapshot: EntityPoseSnapshot): boolean;
+ /** Remove by canonical id from `scenePeerId`/`ids`. */
+ remove(id: string): boolean;
+ /** Apply interpolated state at authoritative server time; returns roots updated. */
+ tick(serverTimeMs: number): number;
+ ids(): readonly string[];
+ count(): number;
+ /** Interest reset: drops peers but retains shared prototype GPU resources. */
+ clear(): void;
+ dispose(): void;
+}
+
+type PeerVisual =
+ | { kind: "humanoid"; rig: HumanoidRig }
+ | { kind: "dog"; rig: DogRig }
+ | { kind: "crow"; rig: CrowRig }
+ | { kind: "model-x"; rig: ModelXRig };
+
+interface Prototypes {
+ humanoid: HumanoidRig;
+ dog: DogRig;
+ crow: CrowRig;
+ vehicle: ModelXRig;
+}
+
+interface Peer {
+ readonly id: string;
+ readonly root: THREE.Group;
+ readonly buffer: PoseInterpolationBuffer;
+ visual: PeerVisual;
+ kind: EntityPoseSnapshot["kind"];
+ spaceKey: string;
+ latest: EntityPoseSnapshot;
+ lastSequence: number;
+ lastTimestampMs: number;
+}
+
+export function scenePeerId(snapshot: EntityPoseSnapshot): string {
+ return snapshot.entity === "actor" ? `actor:${snapshot.actorId}` : `vehicle:${snapshot.vehicleId}`;
+}
+
+function poseSpaceKey(pose: SpatialPose): string {
+ return pose.space === "geographic" ? "geographic" : `local:${JSON.stringify(pose.cell)}`;
+}
+
+function positive(value: number, name: string): number {
+ if (!(value > 0) || !Number.isFinite(value)) throw new RangeError(`${name} must be finite and positive`);
+ return value;
+}
+
+function resolveMaximumPeers(value: number | undefined): number {
+ const resolved = value ?? 128;
+ if (!Number.isInteger(resolved) || resolved < 1 || resolved > 512) {
+ throw new RangeError("maximumPeers must be an integer from 1 to 512");
+ }
+ return resolved;
+}
+
+function makePrototypes(): Prototypes {
+ const vehicle = buildModelX({ detail: "corridor", paint: 0x050607 });
+ vehicle.root.name = "generic-black-ev.prototype";
+ vehicle.root.userData.vehicleModel = "generic-black-ev";
+ return {
+ humanoid: buildHumanoid(),
+ dog: buildDog(),
+ crow: buildCrow(),
+ vehicle,
+ };
+}
+
+function cloneVisual(snapshot: EntityPoseSnapshot, prototypes: Prototypes): PeerVisual {
+ if (snapshot.entity === "vehicle") {
+ const rig = cloneModelX(prototypes.vehicle);
+ rig.root.name = "generic-black-ev";
+ rig.root.userData.vehicleModel = "generic-black-ev";
+ return { kind: "model-x", rig };
+ }
+ if (snapshot.kind === "dog") return { kind: "dog", rig: cloneDog(prototypes.dog) };
+ if (snapshot.kind === "crow") return { kind: "crow", rig: cloneCrow(prototypes.crow) };
+ return { kind: "humanoid", rig: cloneHumanoid(prototypes.humanoid) };
+}
+
+function visualRoot(visual: PeerVisual): THREE.Group {
+ return visual.rig.root;
+}
+
+function replaceVisual(peer: Peer, snapshot: EntityPoseSnapshot, prototypes: Prototypes): void {
+ visualRoot(peer.visual).removeFromParent();
+ peer.visual = cloneVisual(snapshot, prototypes);
+ peer.root.add(visualRoot(peer.visual));
+ peer.kind = snapshot.kind;
+}
+
+function animateActor(
+ visual: Exclude,
+ sample: BufferedPoseSample,
+): void {
+ const speed = Math.hypot(sample.velocity.xMps, sample.velocity.zMps);
+ const seconds = sample.timestampMs / 1_000;
+ if (visual.kind === "humanoid") {
+ poseHumanoid(visual.rig, {
+ walkPhase: seconds * Math.max(1.2, speed * 5.8),
+ stride: THREE.MathUtils.clamp(speed / 3.2, 0, 0.68),
+ });
+ } else if (visual.kind === "dog") {
+ poseDogWalk(
+ visual.rig,
+ seconds * Math.max(1.4, speed * 7.2),
+ THREE.MathUtils.clamp(speed / 4, 0, 0.68),
+ );
+ poseDogAttention(visual.rig, 0, seconds * 5);
+ } else {
+ poseCrowFlight(
+ visual.rig,
+ seconds * Math.max(3, speed * 1.5),
+ THREE.MathUtils.clamp(0.35 + speed / 14, 0.35, 1),
+ );
+ }
+}
+
+export function createScenePeers(options: ScenePeersOptions): ScenePeers {
+ if (typeof options.project !== "function" || typeof options.groundAt !== "function") {
+ throw new RangeError("scene peers require project and groundAt functions");
+ }
+ const geographicScale = positive(
+ options.geographicSceneUnitsPerMetre ?? 1,
+ "geographicSceneUnitsPerMetre",
+ );
+ const geographicVisualScale = positive(
+ options.geographicVisualScale ?? geographicScale,
+ "geographicVisualScale",
+ );
+ const localScale = positive(options.localSceneUnitsPerMetre ?? 1, "localSceneUnitsPerMetre");
+ const localVisualScale = positive(options.localVisualScale ?? localScale, "localVisualScale");
+ const limit = resolveMaximumPeers(options.maximumPeers);
+ const interpolation = { ...(options.interpolation ?? {}) };
+ const prototypes = makePrototypes();
+ const peers = new Map();
+ const root = new THREE.Group();
+ root.name = "realtime-scene-peers";
+ root.userData.kind = "remote-presence";
+ let disposed = false;
+
+ function add(snapshot: EntityPoseSnapshot): Peer {
+ const visual = cloneVisual(snapshot, prototypes);
+ const peerRoot = new THREE.Group();
+ // IDs deliberately stay in the private Map, not names/userData inspected by render/debug tooling.
+ peerRoot.name = "remote-peer";
+ peerRoot.userData.entity = snapshot.entity;
+ peerRoot.userData.kind = snapshot.kind;
+ peerRoot.userData.forwardAxis = "-Z";
+ peerRoot.add(visualRoot(visual));
+ root.add(peerRoot);
+ const peer: Peer = {
+ id: scenePeerId(snapshot),
+ root: peerRoot,
+ buffer: new PoseInterpolationBuffer(interpolation),
+ visual,
+ kind: snapshot.kind,
+ spaceKey: poseSpaceKey(snapshot.pose),
+ latest: snapshot,
+ lastSequence: -1,
+ lastTimestampMs: -1,
+ };
+ peers.set(peer.id, peer);
+ return peer;
+ }
+
+ function apply(peer: Peer, sample: BufferedPoseSample): boolean {
+ const pose = sample.pose;
+ let visualScale: number;
+ if (pose.space === "geographic") {
+ const projected = options.project(pose.lat, pose.lng);
+ const ground = options.groundAt(pose.lat, pose.lng);
+ if (
+ projected.length < 2 ||
+ !Number.isFinite(projected[0]) ||
+ !Number.isFinite(projected[1]) ||
+ !Number.isFinite(ground)
+ ) {
+ peer.root.visible = false;
+ return false;
+ }
+ peer.root.position.set(
+ projected[0],
+ ground + pose.altitudeM * geographicScale,
+ projected[1],
+ );
+ visualScale = geographicVisualScale;
+ } else {
+ peer.root.position.set(pose.xM * localScale, pose.yM * localScale, pose.zM * localScale);
+ visualScale = localVisualScale;
+ }
+ peer.root.visible = true;
+ peer.root.scale.setScalar(visualScale);
+ peer.root.rotation.order = "YXZ";
+ peer.root.rotation.set(
+ THREE.MathUtils.degToRad(pose.pitchDeg),
+ -THREE.MathUtils.degToRad(pose.headingDeg),
+ 0,
+ );
+ if (peer.visual.kind === "model-x") {
+ const latest = peer.latest.entity === "vehicle" ? peer.latest : null;
+ if (latest) {
+ setModelXSteering(peer.visual.rig, latest.steering * 0.62);
+ // Simulation publishes positive travelled radians; the -Z-forward
+ // wheel geometry rolls forward with negative local-X rotation.
+ setModelXWheelRotation(peer.visual.rig, -latest.wheelRadians);
+ }
+ } else {
+ animateActor(peer.visual, sample);
+ }
+ return true;
+ }
+
+ return {
+ root,
+ upsert(snapshot) {
+ if (disposed || !isEntityPoseSnapshot(snapshot)) return false;
+ const id = scenePeerId(snapshot);
+ let peer = peers.get(id);
+ if (!peer) {
+ if (peers.size >= limit) return false;
+ peer = add(snapshot);
+ } else if (
+ snapshot.sequence <= peer.lastSequence ||
+ snapshot.timestampMs <= peer.lastTimestampMs
+ ) {
+ return false;
+ }
+
+ if (peer.kind !== snapshot.kind) replaceVisual(peer, snapshot, prototypes);
+ const nextSpaceKey = poseSpaceKey(snapshot.pose);
+ if (nextSpaceKey !== peer.spaceKey) {
+ // Interest boundaries are not continuous coordinate frames. Resetting
+ // prevents a floor/city switch from interpolating through empty space.
+ peer.buffer.clear();
+ peer.spaceKey = nextSpaceKey;
+ }
+ if (!peer.buffer.push(snapshot)) {
+ // A failed first push must not consume bounded capacity.
+ if (peer.lastSequence < 0) {
+ peer.root.removeFromParent();
+ peers.delete(id);
+ }
+ return false;
+ }
+ peer.latest = snapshot;
+ peer.lastSequence = snapshot.sequence;
+ peer.lastTimestampMs = snapshot.timestampMs;
+ peer.root.userData.entity = snapshot.entity;
+ peer.root.userData.kind = snapshot.kind;
+ return true;
+ },
+ remove(id) {
+ if (disposed) return false;
+ const peer = peers.get(id);
+ if (!peer) return false;
+ peer.root.removeFromParent();
+ peer.buffer.clear();
+ peers.delete(id);
+ return true;
+ },
+ tick(serverTimeMs) {
+ if (disposed || !Number.isFinite(serverTimeMs)) return 0;
+ let updated = 0;
+ for (const peer of peers.values()) {
+ const sample = peer.buffer.sample(serverTimeMs);
+ if (sample && apply(peer, sample)) updated += 1;
+ }
+ return updated;
+ },
+ ids: () => [...peers.keys()],
+ count: () => peers.size,
+ clear() {
+ for (const peer of peers.values()) {
+ peer.root.removeFromParent();
+ peer.buffer.clear();
+ }
+ peers.clear();
+ },
+ dispose() {
+ if (disposed) return;
+ for (const peer of peers.values()) peer.root.removeFromParent();
+ peers.clear();
+ root.removeFromParent();
+ disposeHumanoid(prototypes.humanoid);
+ disposeDog(prototypes.dog);
+ disposeCrow(prototypes.crow);
+ disposeModelX(prototypes.vehicle);
+ disposed = true;
+ },
+ };
+}
diff --git a/src/test/presenceIndicator.test.ts b/src/test/presenceIndicator.test.ts
new file mode 100644
index 0000000..07042ba
--- /dev/null
+++ b/src/test/presenceIndicator.test.ts
@@ -0,0 +1,114 @@
+import assert from "node:assert/strict";
+import { describe, it } from "node:test";
+import { createPresenceIndicator } from "../realtime/index.ts";
+
+type Listener = () => void;
+
+class FakeElement {
+ readonly children: FakeElement[] = [];
+ readonly attributes = new Map();
+ readonly listeners = new Map();
+ parentElement: FakeElement | null = null;
+ className = "";
+ textContent = "";
+ hidden = false;
+ type = "";
+ readonly ownerDocument: FakeDocument;
+ constructor(ownerDocument: FakeDocument) { this.ownerDocument = ownerDocument; }
+ append(...nodes: FakeElement[]): void {
+ for (const node of nodes) { node.parentElement = this; this.children.push(node); }
+ }
+ setAttribute(name: string, value: string): void { this.attributes.set(name, value); }
+ getAttribute(name: string): string | null { return this.attributes.get(name) ?? null; }
+ addEventListener(type: string, listener: Listener): void {
+ const listeners = this.listeners.get(type);
+ if (listeners) listeners.push(listener); else this.listeners.set(type, [listener]);
+ }
+ click(): void { for (const listener of this.listeners.get("click") ?? []) listener(); }
+ remove(): void {
+ if (!this.parentElement) return;
+ const index = this.parentElement.children.indexOf(this);
+ if (index >= 0) this.parentElement.children.splice(index, 1);
+ this.parentElement = null;
+ }
+ text(): string { return this.textContent + this.children.map((child) => child.text()).join(""); }
+}
+
+class FakeDocument {
+ createElement(): FakeElement { return new FakeElement(this); }
+}
+
+function setup(onRetry?: () => void) {
+ const document = new FakeDocument();
+ const container = document.createElement();
+ const indicator = createPresenceIndicator({
+ container: container as unknown as HTMLElement,
+ onRetry,
+ });
+ return { container, indicator, root: indicator.root as unknown as FakeElement };
+}
+
+describe("hosted presence indicator", () => {
+ it("is private and off by default with a polite atomic status", () => {
+ const { indicator, root } = setup();
+ assert.deepEqual(indicator.state(), { signedIn: false, connection: "off", nearbyPeerCount: 0 });
+ assert.equal(root.getAttribute("data-presence"), "off");
+ const status = root.children[0]!;
+ assert.equal(status.getAttribute("role"), "status");
+ assert.equal(status.getAttribute("aria-live"), "polite");
+ assert.equal(status.getAttribute("aria-atomic"), "true");
+ assert.equal(root.text(), "Hosted presence is off.Retry presence");
+ assert.equal(root.children[1]?.hidden, true);
+ });
+
+ it("renders only signed-in connection state and aggregate nearby count", () => {
+ const { indicator, root } = setup();
+ const source = { signedIn: true, connection: "live" as const, nearbyPeerCount: 7 };
+ assert.deepEqual(indicator.update(source), source);
+ source.nearbyPeerCount = 99;
+ assert.deepEqual(indicator.state(), { signedIn: true, connection: "live", nearbyPeerCount: 7 });
+ assert.match(root.text(), /live · 7 nearby peers/);
+ assert.doesNotMatch(root.text(), /actor|token|display name/i);
+
+ indicator.update({ signedIn: false, connection: "live", nearbyPeerCount: 400 });
+ assert.deepEqual(indicator.state(), { signedIn: false, connection: "off", nearbyPeerCount: 0 });
+ assert.equal(root.text(), "Hosted presence is off.Retry presence");
+ });
+
+ it("shows retry only while signed-in offline/reconnecting and emits intent only", () => {
+ let retries = 0;
+ const { indicator, root } = setup(() => { retries += 1; });
+ const retry = root.children[1]!;
+ retry.click();
+ assert.equal(retries, 0);
+ indicator.update({ signedIn: true, connection: "connecting", nearbyPeerCount: 0 });
+ assert.equal(retry.hidden, true);
+ indicator.update({ signedIn: true, connection: "reconnecting", nearbyPeerCount: 12 });
+ assert.equal(retry.hidden, false);
+ assert.match(root.text(), /reconnecting/);
+ assert.doesNotMatch(root.text(), /12/);
+ retry.click();
+ indicator.update({ signedIn: true, connection: "offline", nearbyPeerCount: 0 });
+ retry.click();
+ assert.equal(retries, 2);
+ assert.equal("fetch" in indicator, false);
+ });
+
+ it("rejects invalid aggregate state and disposes idempotently", () => {
+ const { container, indicator, root } = setup(() => assert.fail("disposed retry fired"));
+ assert.throws(() => indicator.update({ signedIn: true, connection: "live", nearbyPeerCount: -1 }), /integer/);
+ assert.throws(() => indicator.update({
+ signedIn: true, connection: "secret" as "live", nearbyPeerCount: 0,
+ }), /connection/);
+ indicator.update({ signedIn: true, connection: "offline", nearbyPeerCount: 0 });
+ indicator.dispose();
+ indicator.dispose();
+ root.children[1]!.click();
+ assert.equal(container.children.length, 0);
+ assert.deepEqual(indicator.state(), { signedIn: false, connection: "off", nearbyPeerCount: 0 });
+ assert.deepEqual(
+ indicator.update({ signedIn: true, connection: "live", nearbyPeerCount: 2 }),
+ { signedIn: false, connection: "off", nearbyPeerCount: 0 },
+ );
+ });
+});
diff --git a/src/test/realtimeClient.test.ts b/src/test/realtimeClient.test.ts
new file mode 100644
index 0000000..9f38f73
--- /dev/null
+++ b/src/test/realtimeClient.test.ts
@@ -0,0 +1,412 @@
+import assert from "node:assert/strict";
+import { describe, it } from "node:test";
+import {
+ createRealtimeClient,
+ type ActorPoseSnapshot,
+ type RealtimeClientTimers,
+} from "../realtime/index.ts";
+
+const OFFICE = { kind: "office" as const, officeId: "hq" };
+
+function actor(sequence = 0, xM = 0): ActorPoseSnapshot {
+ return {
+ entity: "actor",
+ actorId: "actor-1",
+ kind: "humanoid",
+ sequence,
+ timestampMs: 1_765_000_000_000 + sequence,
+ pose: {
+ space: "local",
+ cell: OFFICE,
+ xM,
+ yM: 0,
+ zM: 0,
+ headingDeg: 0,
+ pitchDeg: 0,
+ },
+ velocity: { xMps: 0, yMps: 0, zMps: 0, yawDegPerSec: 0 },
+ };
+}
+
+function joinGrant(requestId: string) {
+ return {
+ type: "join-grant",
+ protocolVersion: 1,
+ requestId,
+ sessionId: "session-1",
+ actorId: "actor-1",
+ role: "member",
+ serverEpoch: "epoch-1",
+ serverTimeMs: 1_765_000_000_000,
+ nextSequence: 4,
+ resumeToken: "resume-secret",
+ interests: [OFFICE],
+ initial: [],
+ };
+}
+
+function resumeGrant(requestId: string, interests = [OFFICE]) {
+ return {
+ type: "resume-grant",
+ protocolVersion: 1,
+ requestId,
+ sessionId: "session-1",
+ serverEpoch: "epoch-1",
+ serverTimeMs: 1_765_000_000_001,
+ nextSequence: 5,
+ resumeToken: "rotated-resume-secret",
+ continuous: true,
+ snapshot: [],
+ interests,
+ };
+}
+
+function jsonResponse(value: unknown, status = 200): Response {
+ return new Response(JSON.stringify(value), {
+ status,
+ headers: { "Content-Type": "application/json" },
+ });
+}
+
+class FakeTimers implements RealtimeClientTimers {
+ readonly pending = new Map void; delay: number }>();
+ private next = 1;
+
+ setTimeout(callback: () => void, delay: number): ReturnType {
+ const id = this.next++;
+ this.pending.set(id, { callback, delay });
+ return id as unknown as ReturnType;
+ }
+
+ clearTimeout(handle: ReturnType): void {
+ this.pending.delete(handle as unknown as number);
+ }
+
+ runFirst(): number {
+ const entry = this.pending.entries().next().value as [number, { callback: () => void; delay: number }] | undefined;
+ if (!entry) throw new Error("no timer pending");
+ this.pending.delete(entry[0]);
+ entry[1].callback();
+ return entry[1].delay;
+ }
+}
+
+async function settle(): Promise {
+ await Promise.resolve();
+ await Promise.resolve();
+ await new Promise((resolve) => globalThis.setTimeout(resolve, 0));
+}
+
+describe("realtime browser client", () => {
+ it("uses memory-only bearer auth, same-origin credentials, and validates join identity", async () => {
+ const calls: Array<{ url: string; init: RequestInit; body: Record }> = [];
+ const fetcher: typeof fetch = async (input, init = {}) => {
+ const body = JSON.parse(String(init.body)) as Record;
+ calls.push({ url: String(input), init, body });
+ return jsonResponse(joinGrant(String(body.requestId)));
+ };
+ const client = createRealtimeClient({ actorId: "actor-1", accessToken: "access-secret", fetch: fetcher });
+ await client.join(OFFICE, actor());
+
+ assert.equal(calls[0]?.url, "/api/v1/realtime/join");
+ assert.equal(calls[0]?.init.credentials, "same-origin");
+ assert.equal(calls[0]?.init.cache, "no-store");
+ assert.equal(new Headers(calls[0]?.init.headers).get("Authorization"), "Bearer access-secret");
+ assert.equal(calls[0]?.url.includes("access-secret"), false);
+ assert.equal(JSON.stringify(client.state()).includes("secret"), false);
+ assert.throws(() => client.publishPose({
+ ...actor(), velocity: { ...actor().velocity, xMps: Number.NaN },
+ }), /invalid/);
+ });
+
+ it("uses an HttpOnly same-origin session when no browser-readable bearer exists", async () => {
+ const calls: RequestInit[] = [];
+ const fetcher: typeof fetch = async (_input, init = {}) => {
+ const body = JSON.parse(String(init.body)) as Record;
+ calls.push(init);
+ return jsonResponse(joinGrant(String(body.requestId)), 201);
+ };
+ const client = createRealtimeClient({ actorId: "actor-1", fetch: fetcher });
+ await client.join(OFFICE, actor());
+
+ assert.equal(new Headers(calls[0]?.headers).has("authorization"), false);
+ assert.equal(calls[0]?.credentials, "same-origin");
+ });
+
+ it("accepts one authenticated request adapter without retaining its credential", async () => {
+ let used = false;
+ const authenticatedFetch: typeof fetch = async (_input, init = {}) => {
+ used = true;
+ const body = JSON.parse(String(init.body)) as Record;
+ return jsonResponse(joinGrant(String(body.requestId)), 201);
+ };
+ const client = createRealtimeClient({ actorId: "actor-1", authenticatedFetch });
+ await client.join(OFFICE, actor());
+ assert.equal(used, true);
+ assert.throws(
+ () => createRealtimeClient({ actorId: "actor-1", authenticatedFetch, fetch }),
+ /either fetch or authenticatedFetch/,
+ );
+ });
+
+ it("coalesces poses at 10-15Hz and assigns monotonic client sequences", async () => {
+ const timers = new FakeTimers();
+ const poses: ActorPoseSnapshot[] = [];
+ const fetcher: typeof fetch = async (input, init = {}) => {
+ const body = JSON.parse(String(init.body)) as Record;
+ if (String(input).endsWith("/join")) return jsonResponse(joinGrant(String(body.requestId)));
+ if (String(input).endsWith("/pose")) {
+ poses.push(body.snapshot as ActorPoseSnapshot);
+ return new Response(null, { status: 204 });
+ }
+ return new Response(null, { status: 204 });
+ };
+ const client = createRealtimeClient({
+ actorId: "actor-1", accessToken: "token", fetch: fetcher, timers, sendIntervalMs: 80,
+ });
+ await client.join(OFFICE, actor());
+ client.publishPose(actor(0, 1));
+ client.publishPose(actor(0, 2));
+ assert.equal(timers.pending.size, 1);
+ assert.equal(timers.runFirst(), 80);
+ await settle();
+ assert.equal(poses.length, 1);
+ assert.equal(poses[0]?.pose.space === "local" ? poses[0].pose.xM : -1, 2);
+ assert.equal(poses[0]?.sequence, 5, "the superseding pose keeps the later monotonic sequence");
+
+ client.publishPose(actor(0, 3));
+ timers.runFirst();
+ await settle();
+ assert.equal(poses[1]?.sequence, 6);
+ assert.throws(
+ () => createRealtimeClient({ actorId: "actor-1", accessToken: "token", fetch: fetcher, sendIntervalMs: 50 }),
+ /66\.67ms/,
+ );
+ });
+
+ it("holds a pose until a stream resume delivers the rotated credential", async () => {
+ const timers = new FakeTimers();
+ const poseTokens: unknown[] = [];
+ const stream: { current: ReadableStreamDefaultController | null } = { current: null };
+ const fetcher: typeof fetch = async (input, init = {}) => {
+ const body = JSON.parse(String(init.body)) as Record;
+ if (String(input).endsWith("/join")) return jsonResponse(joinGrant(String(body.requestId)));
+ if (String(input).endsWith("/events")) {
+ return new Response(new ReadableStream({ start(controller) { stream.current = controller; } }), {
+ headers: { "Content-Type": "text/event-stream" },
+ });
+ }
+ if (String(input).endsWith("/pose")) {
+ poseTokens.push(body.token);
+ return new Response(null, { status: 202 });
+ }
+ return new Response(null, { status: 204 });
+ };
+ const client = createRealtimeClient({
+ actorId: "actor-1", accessToken: "token", fetch: fetcher, timers,
+ });
+ client.subscribe({});
+ await client.join(OFFICE, actor());
+ await settle();
+ client.publishPose(actor(0, 1));
+ timers.runFirst();
+ await settle();
+ assert.equal(poseTokens.length, 0);
+
+ stream.current?.enqueue(new TextEncoder().encode(
+ `data: ${JSON.stringify(resumeGrant("stream-resume"))}\n\n`,
+ ));
+ await settle();
+ timers.runFirst();
+ await settle();
+ assert.deepEqual(poseTokens, ["rotated-resume-secret"]);
+ await client.dispose();
+ });
+
+ it("parses validated POST SSE events, rejects malformed input, and reconnects with backoff", async () => {
+ const timers = new FakeTimers();
+ const malformed: string[] = [];
+ const messages: string[] = [];
+ let streamCalls = 0;
+ const fetcher: typeof fetch = async (input, init = {}) => {
+ const body = JSON.parse(String(init.body)) as Record;
+ if (String(input).endsWith("/join")) return jsonResponse(joinGrant(String(body.requestId)));
+ if (String(input).endsWith("/events")) {
+ streamCalls += 1;
+ const delta = {
+ type: "pose-delta", protocolVersion: 1, serverEpoch: "epoch-1", sequence: 4,
+ timestampMs: 1_765_000_000_004, updates: [], removedEntityIds: [],
+ };
+ return new Response(`data: not-json\n\ndata: ${JSON.stringify(delta)}\n\n`, {
+ headers: { "Content-Type": "text/event-stream" },
+ });
+ }
+ return new Response(null, { status: 204 });
+ };
+ const client = createRealtimeClient({
+ actorId: "actor-1",
+ accessToken: "token",
+ fetch: fetcher,
+ timers,
+ reconnectBaseMs: 200,
+ reconnectMaximumMs: 2_000,
+ reconnectJitter: 0,
+ random: () => 0.5,
+ });
+ client.subscribe({
+ onMalformedEvent: (error) => malformed.push(error.message),
+ onMessage: (message) => messages.push(message.type),
+ });
+ await client.join(OFFICE, actor());
+ await settle();
+ assert.deepEqual(messages, ["pose-delta"]);
+ assert.equal(malformed.length, 1);
+ assert.equal(client.state().status, "reconnecting");
+ assert.equal(timers.runFirst(), 200);
+ await settle();
+ assert.equal(streamCalls, 2);
+ });
+
+ it("moves interest through resume and aborts streaming before leave/dispose", async () => {
+ const requests: Array<{ url: string; body: Record; signal: AbortSignal | null }> = [];
+ const fetcher: typeof fetch = async (input, init = {}) => {
+ const body = JSON.parse(String(init.body)) as Record;
+ requests.push({ url: String(input), body, signal: init.signal ?? null });
+ if (String(input).endsWith("/join")) {
+ if (body.type === "resume-request") return jsonResponse(resumeGrant(String(body.requestId)));
+ return jsonResponse(joinGrant(String(body.requestId)));
+ }
+ if (String(input).endsWith("/events")) {
+ return new Response(new ReadableStream({ start() { /* held open until aborted */ } }), {
+ headers: { "Content-Type": "text/event-stream" },
+ });
+ }
+ return new Response(null, { status: 204 });
+ };
+ const client = createRealtimeClient({ actorId: "actor-1", accessToken: "token", fetch: fetcher });
+ client.subscribe({});
+ await client.join(OFFICE, actor());
+ await settle();
+ const firstStream = requests.find((request) => request.url.endsWith("/events"));
+ assert.equal(firstStream?.signal?.aborted, false);
+
+ const moved = { kind: "floor" as const, officeId: "hq", floorId: "two" };
+ await client.moveInterest(moved);
+ assert.deepEqual(client.state().interests, [moved]);
+ assert.equal(firstStream?.signal?.aborted, true);
+ await client.dispose();
+ assert.equal(client.state().status, "disposed");
+ assert.equal(requests.filter((request) => request.url.endsWith("/leave")).length, 1);
+ assert.equal(JSON.stringify(client.state()).includes("resume-secret"), false);
+ });
+
+ it("closes the old-cell stream when an interest handoff fails", async () => {
+ const requests: Array<{ url: string; signal: AbortSignal | null }> = [];
+ let resumes = 0;
+ const fetcher: typeof fetch = async (input, init = {}) => {
+ const url = String(input);
+ const body = JSON.parse(String(init.body)) as Record;
+ requests.push({ url, signal: init.signal ?? null });
+ if (url.endsWith("/join") && body.type === "resume-request") {
+ resumes += 1;
+ return jsonResponse({ error: "interest" }, 409);
+ }
+ if (url.endsWith("/join")) return jsonResponse(joinGrant(String(body.requestId)));
+ if (url.endsWith("/events")) {
+ return new Response(new ReadableStream({ start() { /* held until abort */ } }), {
+ headers: { "Content-Type": "text/event-stream" },
+ });
+ }
+ return new Response(null, { status: 204 });
+ };
+ const client = createRealtimeClient({ actorId: "actor-1", accessToken: "token", fetch: fetcher });
+ client.subscribe({});
+ await client.join(OFFICE, actor());
+ await settle();
+ const oldStream = requests.find((request) => request.url.endsWith("/events"));
+ await assert.rejects(
+ client.moveInterest({ kind: "city", cityId: "bay-area" }),
+ /request failed/,
+ );
+ assert.equal(resumes, 1);
+ assert.equal(oldStream?.signal?.aborted, true);
+ assert.equal(client.state().status, "joined");
+ assert.deepEqual(client.state().interests, [OFFICE]);
+ await client.dispose();
+ });
+
+ it("closes the old-cell stream before a slow interest handoff settles", async () => {
+ const requests: Array<{ url: string; signal: AbortSignal | null }> = [];
+ let finishResume: ((response: Response) => void) | null = null;
+ let resumeRequestId = "";
+ const fetcher: typeof fetch = async (input, init = {}) => {
+ const url = String(input);
+ const body = JSON.parse(String(init.body)) as Record;
+ requests.push({ url, signal: init.signal ?? null });
+ if (url.endsWith("/join") && body.type === "resume-request") {
+ resumeRequestId = String(body.requestId);
+ return new Promise((resolve) => { finishResume = resolve; });
+ }
+ if (url.endsWith("/join")) return jsonResponse(joinGrant(String(body.requestId)));
+ if (url.endsWith("/events")) {
+ return new Response(new ReadableStream({ start() { /* held until abort */ } }), {
+ headers: { "Content-Type": "text/event-stream" },
+ });
+ }
+ return new Response(null, { status: 204 });
+ };
+ const client = createRealtimeClient({ actorId: "actor-1", fetch: fetcher });
+ client.subscribe({});
+ await client.join(OFFICE, actor());
+ await settle();
+ const oldStream = requests.find((request) => request.url.endsWith("/events"));
+ const moved = { kind: "city" as const, cityId: "bay-area" as const };
+ const handoff = client.moveInterest(moved);
+ assert.equal(oldStream?.signal?.aborted, true, "privacy-cell stream closes before HTTP completes");
+ assert.ok(finishResume);
+ (finishResume as (response: Response) => void)(jsonResponse(resumeGrant(resumeRequestId)));
+ await handoff;
+ assert.deepEqual(client.state().interests, [moved]);
+ await client.dispose();
+ });
+
+ it("lets only the newest concurrent join commit client state", async () => {
+ const pending: Array<{
+ body: Record;
+ resolve: (response: Response) => void;
+ }> = [];
+ const leaves: Record[] = [];
+ const fetcher: typeof fetch = async (input, init = {}) => {
+ const body = JSON.parse(String(init.body)) as Record;
+ if (String(input).endsWith("/leave")) {
+ leaves.push(body);
+ return new Response(null, { status: 204 });
+ }
+ return new Promise((resolve) => pending.push({ body, resolve }));
+ };
+ const client = createRealtimeClient({ actorId: "actor-1", fetch: fetcher });
+ const first = client.join(OFFICE, actor());
+ const latestInterest = { kind: "city" as const, cityId: "socal" as const };
+ const latest = client.join(latestInterest, actor());
+ const secondRequest = pending[1];
+ assert.ok(secondRequest);
+ secondRequest.resolve(jsonResponse({
+ ...joinGrant(String(secondRequest.body.requestId)),
+ sessionId: "session-latest",
+ interests: [latestInterest],
+ }));
+ await latest;
+ const firstRequest = pending[0];
+ assert.ok(firstRequest);
+ firstRequest.resolve(jsonResponse({
+ ...joinGrant(String(firstRequest.body.requestId)),
+ sessionId: "session-stale",
+ }));
+ await assert.rejects(first, /superseded/);
+ await settle();
+ assert.equal(client.state().sessionId, "session-latest");
+ assert.deepEqual(client.state().interests, [latestInterest]);
+ assert.deepEqual(leaves, [{ sessionId: "session-stale", token: "resume-secret" }]);
+ await client.dispose();
+ });
+});
diff --git a/src/test/scenePeers.test.ts b/src/test/scenePeers.test.ts
new file mode 100644
index 0000000..1a20abc
--- /dev/null
+++ b/src/test/scenePeers.test.ts
@@ -0,0 +1,173 @@
+import assert from "node:assert/strict";
+import { describe, it } from "node:test";
+import * as THREE from "three";
+import {
+ createScenePeers,
+ scenePeerId,
+ type ActorPoseSnapshot,
+ type EntityPoseSnapshot,
+ type InterestCell,
+ type ScenePeersOptions,
+ type VehiclePoseSnapshot,
+} from "../realtime/index.ts";
+
+const VELOCITY = { xMps: 0, yMps: 0, zMps: 0, yawDegPerSec: 0 };
+
+function actor(
+ id: string,
+ kind: "humanoid" | "dog" | "crow",
+ sequence: number,
+ timestampMs: number,
+ xM: number,
+ cell: InterestCell = { kind: "floor", officeId: "hq", floorId: "one" },
+): ActorPoseSnapshot {
+ if (cell.kind === "california-tile") throw new Error("local poses cannot use tile cells");
+ return {
+ entity: "actor",
+ actorId: id,
+ kind,
+ sequence,
+ timestampMs,
+ pose: { space: "local", cell, xM, yM: 2, zM: 3, headingDeg: 0, pitchDeg: 0 },
+ velocity: { ...VELOCITY, xMps: 2 },
+ };
+}
+
+function vehicle(sequence: number, timestampMs: number, lat: number): VehiclePoseSnapshot {
+ return {
+ entity: "vehicle",
+ vehicleId: "ev-1",
+ kind: "model-x",
+ driverActorId: null,
+ sequence,
+ timestampMs,
+ pose: { space: "geographic", lat, lng: -120, altitudeM: 10, headingDeg: 90, pitchDeg: 2 },
+ velocity: { ...VELOCITY, xMps: 20 },
+ steering: 0.4,
+ wheelRadians: 3,
+ };
+}
+
+function fixture(over: Partial = {}) {
+ return createScenePeers({
+ project: (lat, lng) => [lng * 2, -lat * 3],
+ groundAt: () => 5,
+ geographicSceneUnitsPerMetre: 0.1,
+ localSceneUnitsPerMetre: 1,
+ interpolation: { interpolationDelayMs: 0, maximumExtrapolationMs: 0 },
+ ...over,
+ });
+}
+
+function mesh(root: THREE.Object3D, name: string): THREE.Mesh {
+ const found = root.getObjectByName(name);
+ if (!(found instanceof THREE.Mesh)) throw new Error(`missing mesh ${name}`);
+ return found;
+}
+
+describe("remote scene peers", () => {
+ it("maps local poses directly and keeps one identity-free stable root per peer", () => {
+ const peers = fixture();
+ const first = actor("private-member-id", "humanoid", 1, 1_000, 1);
+ assert.equal(peers.upsert(first), true);
+ assert.deepEqual(peers.ids(), ["actor:private-member-id"]);
+ assert.equal(peers.count(), 1);
+ const root = peers.root.children[0] as THREE.Group;
+ assert.equal(root.name, "remote-peer");
+ assert.equal(JSON.stringify(root.userData).includes("private-member-id"), false);
+ assert.equal(root.getObjectByName("humanoid.face") instanceof THREE.Mesh, true);
+ assert.equal((mesh(root, "humanoid.face").material as THREE.MeshBasicMaterial).map, null);
+ peers.tick(1_000);
+ assert.deepEqual(root.position.toArray(), [1, 2, 3]);
+
+ assert.equal(peers.upsert(actor("private-member-id", "humanoid", 2, 2_000, 5)), true);
+ peers.tick(1_500);
+ assert.equal(peers.root.children[0], root);
+ assert.equal(root.position.x, 3);
+ assert.equal(peers.upsert(actor("private-member-id", "humanoid", 2, 2_100, 8)), false);
+ peers.dispose();
+ });
+
+ it("resets interpolation at local interest-cell boundaries without replacing the root", () => {
+ const peers = fixture();
+ peers.upsert(actor("a", "dog", 1, 1_000, 0));
+ peers.upsert(actor("a", "dog", 2, 2_000, 10));
+ peers.tick(1_500);
+ const root = peers.root.children[0] as THREE.Group;
+ assert.equal(root.position.x, 5);
+ const nextCell = { kind: "floor", officeId: "hq", floorId: "two" } as const;
+ assert.equal(peers.upsert(actor("a", "dog", 3, 3_000, 100, nextCell)), true);
+ peers.tick(2_500);
+ assert.equal(peers.root.children[0], root);
+ assert.equal(root.position.x, 100, "new coordinate frame is held, never mixed with old floor");
+ peers.dispose();
+ });
+
+ it("projects geographic vehicles onto caller terrain and renders a generic black EV", () => {
+ const peers = fixture();
+ const snapshot = vehicle(1, 1_000, 34);
+ assert.equal(scenePeerId(snapshot), "vehicle:ev-1");
+ peers.upsert(snapshot);
+ assert.equal(peers.tick(1_000), 1);
+ const root = peers.root.children[0] as THREE.Group;
+ assert.deepEqual(root.position.toArray(), [-240, 6, -102]);
+ assert.equal(root.scale.x, 0.1);
+ assert.ok(Math.abs(root.rotation.y + Math.PI / 2) < 1e-12);
+ assert.equal(root.getObjectByName("generic-black-ev")?.userData.vehicleModel, "generic-black-ev");
+ assert.equal(root.getObjectByName("frontLeft.spin")?.rotation.x, -snapshot.wheelRadians);
+ const paint = root.getObjectByName("model-x.body:model-x.paint") as THREE.Mesh;
+ assert.equal((paint.material as THREE.MeshStandardMaterial).color.getHex(), 0x050607);
+ peers.dispose();
+ });
+
+ it("shares prototype resources while kind changes retain the entity root", () => {
+ const peers = fixture();
+ peers.upsert(actor("one", "humanoid", 1, 1_000, 0));
+ peers.upsert(actor("two", "humanoid", 1, 1_000, 2));
+ const one = peers.root.children[0] as THREE.Group;
+ const two = peers.root.children[1] as THREE.Group;
+ assert.equal(mesh(one, "humanoid.chest").geometry, mesh(two, "humanoid.chest").geometry);
+ assert.equal(mesh(one, "humanoid.chest").material, mesh(two, "humanoid.chest").material);
+ const oldRig = one.children[0];
+ peers.upsert(actor("one", "crow", 2, 2_000, 1));
+ peers.tick(2_000);
+ assert.equal(peers.root.children[0], one);
+ assert.equal(oldRig?.parent, null);
+ assert.ok(one.getObjectByName("crow"));
+ peers.dispose();
+ });
+
+ it("bounds nearby peers and removes/clears without prematurely disposing shared assets", () => {
+ const peers = fixture({ maximumPeers: 2 });
+ peers.upsert(actor("one", "humanoid", 1, 1_000, 0));
+ peers.upsert(actor("two", "dog", 1, 1_000, 1));
+ assert.equal(peers.upsert(actor("three", "crow", 1, 1_000, 2)), false);
+ const firstPeer = peers.root.children[0];
+ if (!firstPeer) throw new Error("missing first peer");
+ const geometry = mesh(firstPeer, "humanoid.chest").geometry;
+ let disposals = 0;
+ geometry.addEventListener("dispose", () => disposals++);
+ assert.equal(peers.remove("actor:one"), true);
+ assert.equal(peers.remove("actor:one"), false);
+ assert.equal(disposals, 0, "removing one clone cannot release shared prototype resources");
+ peers.clear();
+ assert.equal(peers.count(), 0);
+ assert.deepEqual(peers.ids(), []);
+ assert.equal(disposals, 0);
+ peers.dispose();
+ peers.dispose();
+ assert.equal(disposals, 1);
+ assert.equal(peers.upsert(actor("late", "humanoid", 1, 1_000, 0)), false);
+ });
+
+ it("contains invalid snapshots and projection failures", () => {
+ const peers = fixture({ project: () => [Number.NaN, 0] });
+ assert.equal(peers.upsert({} as EntityPoseSnapshot), false);
+ peers.upsert(vehicle(1, 1_000, 34));
+ assert.equal(peers.tick(1_000), 0);
+ assert.equal(peers.root.children[0]?.visible, false);
+ assert.equal(peers.tick(Number.NaN), 0);
+ peers.dispose();
+ assert.throws(() => fixture({ maximumPeers: 0 }), RangeError);
+ });
+});