Add multiplayer chat, durable streaming, and site observability

apps/site/AGENT_STREAMING.md

@@ -0,0 +1,93 @@
+# Agent streaming
+
+How live assistant output travels from the model to every viewer of a shared
+session, and why the persistence is shaped the way it is.
+
+## Pipeline
+
+```
+AI SDK fullStream
+  → executor text segments        (packages/aisdk-executor/src/executor.ts)
+  → onStreamUpdate callbacks      (out-of-band; cumulative text + streamSeq)
+  → streamWriter                  (convex/agent.ts — throttle, diff to deltas)
+  → appendStreamDelta mutation    (convex/messages.ts — append-only rows)
+  → streamText query              (per-message subscription in the client)
+```
+
+**Segments.** The executor consumes the AI SDK `fullStream` and opens one
+segment per `text-start` part. Each segment gets its own message id, so a turn
+that interleaves prose and tool calls (progress update → tool → answer)
+produces distinct assistant messages in the order the user watched them
+stream. `text-end` enqueues the completed segment as its own frame
+immediately — before a following tool call can finish — so frame replay
+preserves the observed order.
+
+**streamWriter.** Stream callbacks are fire-and-forget from the executor's
+point of view; the writer serializes them onto a single chained tail promise
+(`tail = tail.then(drainAll)`) — one mechanism, no flags or polling. It keeps
+only the newest snapshot per segment, caps writes at one per 250ms (4 fps),
+and diffs each snapshot against the last persisted text so every write is a
+small immutable delta, not a cumulative rewrite. `flush()` runs in the
+action's `finally` and just awaits the tail. `drainAll` never rejects (both
+mutation calls are individually guarded), so the chain cannot wedge.
+
+**Deltas.** `appendStreamDelta` upserts the durable message row (created
+`streaming` with empty content) and appends a `messageStreamDeltas` row. The
+serialized writer guarantees arrival order; `streamSeq` is stored only as the
+deterministic sort key for reads. On completion the frame-persistence pass
+writes the full text onto the message row and deletes the deltas; the delta
+table only ever holds in-flight tails.
+
+## Read isolation
+
+A delta lands every 250ms during a stream, so what that write invalidates is
+the whole cost model. The queries are split so it invalidates almost nothing:
+
+- `messages.list` — the transcript. No stream text, no viewer identity. Only
+  re-runs when a row is created or settled.
+- `messages.viewerActors` — who the caller is (GitHub actor + owned anonymous
+  actor). Depends on the session doc and auth state only.
+- `messages.streamText` — live text for one streaming message: durable
+  content plus the concatenated delta tail. Each streaming `Message`
+  component subscribes to its own; a delta write invalidates exactly this one
+  small query per viewer.
+
+## Failure handling
+
+- **Model or tool throws:** the executor emits `error` stream updates for
+  unenqueued segments; the writer's terminal path calls `markStreamFailed`,
+  which folds the delta tail into `content` and settles the row.
+- **Barge-in cancel:** same shape with `cancelled`; the partial text goes to
+  the frame log marked interrupted, and hosts skip message persistence.
+- **Hard death (action timeout, crash — no code runs):** when
+  `appendStreamDelta` creates a streaming row it schedules
+  `markStreamFailed` 15 minutes out — past the 10-minute action ceiling. A
+  normally settled stream makes the reap a no-op; an orphaned row gets its
+  text recovered, its state set to `error`, and its deltas deleted.
+
+## Relation to @convex-dev/agent
+
+The server side here is deliberately the same design as the agent component's
+`DeltaStreamer`: append-only delta rows written by a throttled single-flight
+writer, durable message settled at completion. We keep our own implementation
+because the component's thread/message persistence would fight the frame log,
+which is the framework's actual source of truth.
+
+The one thing the component does that we deliberately don't: **client-side
+cursor sync**. Its `syncStreams` query takes the last position a client has
+seen and returns only newer deltas; clients accumulate text locally, so each
+delta write costs O(1) reads per subscriber. Our `streamText` re-reads the
+message's full delta tail on each invalidation — O(deltas-so-far) per tick.
+
+That trade is intentional. Replies are capped at 4096 output tokens, so a
+typical stream is under a minute (~120 deltas ≈ a few thousand tiny document
+reads per viewer over the whole stream — noise). The cursor upgrade stays
+cheap precisely because live text is already isolated in `streamText`: give
+it an `afterSeq` arg and let `Message` accumulate in state. Nothing about the
+schema, the writer, or the transcript query would change.
+
+Revisit if any of these become true:
+
+- streams routinely run past a couple of minutes;
+- sessions regularly have many simultaneous viewers;
+- Convex insights flag `messages.streamText` read bandwidth.

apps/site/AUTH.md

@@ -39,6 +39,20 @@ These controls deter casual resets; they do not identify a person reliably.
    npx convex env set ANONYMOUS_IP_SALT <random-secret>
    ```
 
+5. Grant developer-panel access with a comma- or whitespace-separated list of
+   stable numeric GitHub account IDs. You can find an account's ID from the
+   GitHub API (`https://api.github.com/users/<handle>` → `id`):
+
+   ```sh
+   npx convex env set ADMIN_GITHUB_IDS "583231,9919"
+   ```
+
+   Convex Auth stores this value as the GitHub account's `providerAccountId`,
+   so authorization survives handle changes. The navigation capability probe
+   reveals only whether the signed-in user is an admin. Every query that
+   returns developer-panel data independently enforces the same server-side
+   admin check.
+
 Production needs its own OAuth App/callback and its own Convex Auth keys and
 environment variables; dev values do not carry over. Run the corresponding
 commands with `--prod`.

apps/site/README.md

@@ -28,3 +28,6 @@ bun run dev
 
 GitHub sign-in and the anonymous inference boundary require one-time deployment
 configuration. See [AUTH.md](./AUTH.md).
+
+How live assistant output is streamed, persisted, and read by multiple viewers
+is documented in [AGENT_STREAMING.md](./AGENT_STREAMING.md).

apps/site/convex/_generated/api.d.ts

@@ -9,13 +9,21 @@
  */
 
 import type * as access from "../access.js";
+import type * as actors from "../actors.js";
 import type * as agent from "../agent.js";
 import type * as artifacts from "../artifacts.js";
 import type * as auth from "../auth.js";
 import type * as convexJson from "../convexJson.js";
+import type * as dev_access from "../dev/access.js";
+import type * as dev_artifacts from "../dev/artifacts.js";
+import type * as dev_prompts from "../dev/prompts.js";
+import type * as dev_sessions from "../dev/sessions.js";
+import type * as executorConfig from "../executorConfig.js";
 import type * as frameHistory from "../frameHistory.js";
 import type * as http from "../http.js";
+import type * as messageActor from "../messageActor.js";
 import type * as messages from "../messages.js";
+import type * as sessionParticipants from "../sessionParticipants.js";
 import type * as sessions from "../sessions.js";
 import type * as topics from "../topics.js";
 
@@ -27,13 +35,21 @@ import type {
 
 declare const fullApi: ApiFromModules<{
   access: typeof access;
+  actors: typeof actors;
   agent: typeof agent;
   artifacts: typeof artifacts;
   auth: typeof auth;
   convexJson: typeof convexJson;
+  "dev/access": typeof dev_access;
+  "dev/artifacts": typeof dev_artifacts;
+  "dev/prompts": typeof dev_prompts;
+  "dev/sessions": typeof dev_sessions;
+  executorConfig: typeof executorConfig;
   frameHistory: typeof frameHistory;
   http: typeof http;
+  messageActor: typeof messageActor;
   messages: typeof messages;
+  sessionParticipants: typeof sessionParticipants;
   sessions: typeof sessions;
   topics: typeof topics;
 }>;

apps/site/convex/_generated/server.d.ts

@@ -21,6 +21,14 @@ import {
 } from "convex/server";
 import type { DataModel } from "./dataModel.js";
 
+/**
+ * Typesafe environment variables declared in `convex.config.ts`.
+ */
+type Env = {
+  readonly ADMIN_GITHUB_IDS: string | undefined;
+  readonly OPENAI_MODEL: string | undefined;
+};
+
 /**
  * Define a query in this Convex app's public API.
  *
@@ -95,6 +103,11 @@ export declare const internalAction: ActionBuilder<DataModel, "internal">;
  */
 export declare const httpAction: HttpActionBuilder;
 
+/**
+ * Typesafe environment variables declared in `convex.config.ts`.
+ */
+export declare const env: Env;
+
 /**
  * A set of services for use within Convex query functions.
  *

apps/site/convex/_generated/server.js

@@ -91,3 +91,8 @@ export const internalAction = internalActionGeneric;
  * @returns The wrapped function. Import this function from `convex/http.js` and route it to hook it up.
  */
 export const httpAction = httpActionGeneric;
+
+/**
+ * Typesafe environment variables declared in `convex.config.ts`.
+ */
+export const env = process.env;

apps/site/convex/access.ts

@@ -1,9 +1,10 @@
 import { DAY, RateLimiter } from "@convex-dev/rate-limiter";
-import { getAuthUserId } from "@convex-dev/auth/server";
 import { ConvexError } from "convex/values";
 import { components } from "./_generated/api";
 import type { Doc } from "./_generated/dataModel";
 import type { MutationCtx } from "./_generated/server";
+import { anonymousActor, authenticatedUser } from "./actors";
+import type { MessageActor } from "./messageActor";
 
 const rateLimiter = new RateLimiter(components.rateLimiter, {
   // The cookie is the primary one-free-turn boundary. The IP limit is a
@@ -32,38 +33,63 @@ export async function hashGuestSecret(secret: string): Promise<string> {
   return Array.from(new Uint8Array(digest), (byte) => byte.toString(16).padStart(2, "0")).join("");
 }
 
+// The one definition of "this caller holds the session's guest secret" —
+// shared by write authorization and viewer attribution so the two can never
+// disagree.
+export async function guestSecretMatches(
+  session: Doc<"sessions">,
+  guestSecret: string | undefined,
+): Promise<boolean> {
+  return (
+    guestSecret !== undefined &&
+    isValidGuestSecret(guestSecret) &&
+    session.guestSecretHash !== undefined &&
+    (await hashGuestSecret(guestSecret)) === session.guestSecretHash
+  );
+}
+
 export async function authorizeSessionWrite(
   ctx: MutationCtx,
   session: Doc<"sessions">,
   guestSecret?: string,
-): Promise<"authenticated" | "guest"> {
-  const userId = await getAuthUserId(ctx);
+): Promise<MessageActor> {
+  const user = await authenticatedUser(ctx);
 
-  if (session.ownerUserId !== undefined) {
-    if (userId !== session.ownerUserId) {
-      throw new ConvexError(ACCESS_ERROR.forbidden);
+  // Public demo sessions are collaborative: authentication is sufficient to
+  // write, regardless of who created the session. Preserve ownerUserId only
+  // as creator metadata for a guest who later signs in.
+  if (user) {
+    if (
+      session.ownerUserId === undefined &&
+      (await guestSecretMatches(session, guestSecret))
+    ) {
+      await ctx.db.patch(session._id, {
+        ownerUserId: user.userId,
+        guestSecretHash: undefined,
+      });
     }
-    return "authenticated";
+    return user.actor;
   }
 
-  const guestMatches =
-    guestSecret !== undefined &&
-    isValidGuestSecret(guestSecret) &&
-    session.guestSecretHash !== undefined &&
-    (await hashGuestSecret(guestSecret)) === session.guestSecretHash;
-  if (!guestMatches) {
-    throw new ConvexError(userId ? ACCESS_ERROR.forbidden : ACCESS_ERROR.authRequired);
+  if (session.ownerUserId !== undefined) {
+    throw new ConvexError(ACCESS_ERROR.authRequired);
   }
 
-  if (userId) {
-    await ctx.db.patch(session._id, {
-      ownerUserId: userId,
-      guestSecretHash: undefined,
-    });
-    return "authenticated";
+  if (!(await guestSecretMatches(session, guestSecret))) {
+    throw new ConvexError(ACCESS_ERROR.authRequired);
   }
 
-  return "guest";
+  return anonymousActor(session._id);
+}
+
+export async function consumeAnonymousTurn(
+  ctx: MutationCtx,
+  session: Doc<"sessions">,
+): Promise<void> {
+  if (session.ownerUserId !== undefined || session.anonymousTurnUsedAt !== undefined) {
+    throw new ConvexError(ACCESS_ERROR.authRequired);
+  }
+  await ctx.db.patch(session._id, { anonymousTurnUsedAt: Date.now() });
 }
 
 export async function reserveAnonymousTurn(
@@ -87,5 +113,5 @@ export async function reserveAnonymousTurn(
   const ipLimit = await rateLimiter.limit(ctx, "anonymousByIp", { key: ipHash });
   if (!ipLimit.ok) throw new ConvexError(ACCESS_ERROR.ipRateLimited);
 
-  await ctx.db.patch(session._id, { anonymousTurnUsedAt: Date.now() });
+  await consumeAnonymousTurn(ctx, session);
 }

apps/site/convex/actors.ts

@@ -0,0 +1,33 @@
+import { getAuthUserId } from "@convex-dev/auth/server";
+import type { Doc, Id } from "./_generated/dataModel";
+import type { MutationCtx, QueryCtx } from "./_generated/server";
+import type { MessageActor } from "./messageActor";
+
+export function anonymousActor(sessionId: Id<"sessions">): MessageActor {
+  return {
+    id: `anonymous:${sessionId}`,
+    kind: "anonymous",
+    label: "ANON",
+  };
+}
+
+export function githubActor(user: Doc<"users">): MessageActor {
+  // auth.ts deliberately stores GitHub's login in `name`; never use the
+  // provider's human-facing display name for shared-room attribution.
+  const handle = (user.name?.trim() || "user").toLowerCase();
+  return {
+    id: `github:${user._id}`,
+    kind: "github",
+    label: `@github/${handle}`,
+    profileUrl: `https://github.com/${encodeURIComponent(handle)}`,
+  };
+}
+
+export async function authenticatedUser(
+  ctx: MutationCtx | QueryCtx,
+): Promise<{ userId: Id<"users">; actor: MessageActor } | null> {
+  const userId = await getAuthUserId(ctx);
+  if (!userId) return null;
+  const user = await ctx.db.get(userId);
+  return user ? { userId, actor: githubActor(user) } : null;
+}