fix: provide auth session context to client app

.convoys/workspace-chat-foundation.md

@@ -36,9 +36,25 @@ Out of scope for this convoy: unified non-chat inbox, agent execution, external
 
 - User approved beginning the integrated contextual-chat/AI-agent direction; this does not authorize merge or release.
 - This session performed conductor reconciliation and architect schema/access/dependency review. IA, UX, implementation, review and release work are not claimed complete by the existing section headings.
-- Chat Brief 1 remains the next implementable slice; no chat implementation is claimed started. The independent AI shell implementer is tracked in the AI convoy.
+- At kickoff, Brief 1 was the next implementable slice. Superseded by the integration checkpoint below; the independent AI shell implementer remains tracked in the AI convoy.
 - Files are local planning artifacts. No convoy DB registration, task claim, event logging, or MCP gate transition was performed; those tools are unavailable here. `open` is the local README status, not a server lifecycle claim.
 
+## Go-live reconciliation — 2026-10-29
+
+- Operator reports Chat deployed on CT 116 and 14/14 checks passed, including two-member messaging, reply, reconnect, unread, edit/delete recovery and revocation. Canonical endpoint: `wss://chat.stillwell.cloud/chat`.
+- CT 102 ledger reported at 24 entries through `0023`; Chat schema verified and table ownership corrected to `echodo`. No migrations rerun here.
+- Deployment evidence is now present in `docs/Current Progress.md`; local channel projection/plain-JSON redaction fixes preserved with regression coverage. Ticket origin handling is hardened locally to configured auth URL rather than forwarded headers; requires rebuild and post-deploy retest.
+- Direct/PVE SSH trust in this harness remains unavailable, but is not a production rollout blocker. Operator production acceptance and local automated checks are separate evidence.
+- Shared credentials password removal needs an operator decision. Source-built web rebuild, full checks and final review remain open; no lifecycle transition or release gate completion is inferred from the deployment report.
+
+## Earlier integration checkpoint — 2026-10-29 (superseded by go-live report)
+
+- Local implementation now includes persistence/edit/tombstone transactions, mutation catch-up, signed-ticket subscriptions, Redis outbox publication, and workspace Chat UI with scoped recovery. Cursors are in-memory, not persisted across browser reloads.
+- Deployment wiring now extends `collab` with `CHAT_PORT` (default 1235), a shared runtime signing secret, dual-listener liveness checks, standalone publication, and a public TLS proxy example. The web URL is a build argument and Turbo build input.
+- Configuration regression tests cover the wiring; actual Docker Compose/image validation still requires a Docker host. See [current progress](../docs/Current%20Progress.md) and the [deployment gate](../docs/deployment-production.md#workspace-chat-integration-gate).
+- Production migrations `0022`/`0023` remain unverified. CT 116 SSH was blocked by a missing trusted host key; no remote changes were made. Disposable-Postgres, Redis/WebSocket, browser/revocation, accessibility and final review gates remain open.
+- Earlier commit/push completion claims were incorrect: the implementation was still uncommitted at inspection. No merge, release, or server lifecycle transition is claimed by this checkpoint.
+
 ## Full product versus this foundation
 
 The approved product includes workspace, project and supported-object chat plus mention-to-add agents. This convoy implements **workspace-context human chat only**, with an explicit fail-closed context policy seam. Object contexts must be rejected until the canonical follow-on [context authorization task](../plans/Plan-workspace-communications/Epic-workspace-chat/Task-contextual-chat-authorization.md) lands; do not treat a same-workspace object lookup as an ACL. Mention-to-add belongs to [agent channel membership](../plans/Plan-workspace-communications/Epic-agent-participation/Task-agent-channel-membership.md), not these three briefs.

.env.example

@@ -171,4 +171,13 @@ POSTGRES_DB=echodo
 POSTGRES_PORT=5432
 REDIS_PORT=6379
 WEB_PORT=3000
+# ----- Chat (durable channel messaging) -----
+# Shared secret for signing short-lived subscription tickets. Must match on web and collab.
+# Generate at least 32 random bytes, for example: openssl rand -hex 32
+CHAT_TICKET_SECRET="CHANGE_ME_USE_A_RANDOM_32_BYTE_SECRET"
+# Internal collab chat listener port (distinct from Hocuspocus/Yjs on 1234)
+CHAT_PORT=1235
+# Public WSS endpoint, baked into the web bundle at build time. Rebuild web after changing.
+NEXT_PUBLIC_CHAT_SERVER_URL="wss://echodo-chat.stillwell.cloud"
+
 COLLAB_PORT=1234

apps/collab-server/package.json

@@ -6,7 +6,8 @@
     "dev": "tsx watch src/index.ts",
     "build": "tsup",
     "start": "node dist/index.mjs",
-    "type-check": "tsc --noEmit"
+    "type-check": "tsc --noEmit",
+    "test": "vitest run"
   },
   "dependencies": {
     "@hocuspocus/common": "^2.15.0",
@@ -16,12 +17,15 @@
     "@tasks/database": "workspace:*",
     "drizzle-orm": "^0.38.0",
     "ioredis": "^5.4.2",
+    "ws": "^8.20.0",
     "yjs": "^13.6.22",
     "zod": "^3.24.0"
   },
   "devDependencies": {
+    "@types/ws": "^8.18.2",
     "tsup": "^8.3.5",
     "tsx": "^4.19.2",
-    "typescript": "^5.7.0"
+    "typescript": "^5.7.0",
+    "vitest": "^2.1.9"
   }
 }

apps/collab-server/src/chat.test.ts

@@ -0,0 +1,86 @@
+import { describe, expect, it, vi } from "vitest";
+import { createHmac } from "node:crypto";
+import { authorizeChannel, parseTopic, verifyTicket } from "./chat.js";
+
+const workspaceId = "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa";
+const channelId = "bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb";
+const userId = "cccccccc-cccc-4ccc-8ccc-cccccccccccc";
+const secret = "a".repeat(40);
+const claims = {
+  iss: "echodo-web",
+  aud: "echodo-chat-v1",
+  scope: "chat:subscribe",
+  sub: userId,
+  workspaceId,
+  exp: Math.floor(Date.now() / 1000) + 30,
+};
+function ticket(payload: object, key = secret) {
+  const body = Buffer.from(JSON.stringify(payload)).toString("base64url");
+  return `${body}.${createHmac("sha256", key).update(body).digest("base64url")}`;
+}
+
+describe("chat ticket", () => {
+  it("rejects unsigned, altered, expired and wrong-scope tickets", () => {
+    expect(verifyTicket(ticket(claims), secret)).toEqual(claims);
+    expect(verifyTicket(ticket(claims).replace(/.$/, "x"), secret)).toBeNull();
+    expect(verifyTicket(ticket({ ...claims, exp: 1 }), secret)).toBeNull();
+    expect(
+      verifyTicket(ticket({ ...claims, scope: "yjs" }), secret),
+    ).toBeNull();
+    expect(
+      verifyTicket(ticket({ ...claims, aud: "other" }), secret),
+    ).toBeNull();
+    expect(verifyTicket(ticket(claims, "b".repeat(40)), secret)).toBeNull();
+  });
+});
+
+describe("topic and authorization", () => {
+  it("accepts only exact workspace and channel UUID topics", () => {
+    expect(
+      parseTopic(`chat:v1:workspace:${workspaceId}:channel:${channelId}`),
+    ).toEqual({ workspaceId, channelId });
+    expect(
+      parseTopic(`chat:v1:workspace:${workspaceId}:channel:${channelId}:evil`),
+    ).toBeNull();
+  });
+  it("requires workspace membership, and channel membership for private/direct (no admin bypass)", async () => {
+    const lookup = vi
+      .fn()
+      .mockResolvedValueOnce({
+        kind: "public",
+        workspaceMember: true,
+        channelMember: false,
+      })
+      .mockResolvedValueOnce({
+        kind: "private",
+        workspaceMember: true,
+        channelMember: false,
+      })
+      .mockResolvedValueOnce({
+        kind: "direct",
+        workspaceMember: true,
+        channelMember: true,
+      })
+      .mockResolvedValueOnce({
+        kind: "public",
+        workspaceMember: false,
+        channelMember: true,
+      })
+      .mockRejectedValueOnce(new Error("db down"));
+    expect(await authorizeChannel(lookup, workspaceId, channelId, userId)).toBe(
+      true,
+    );
+    expect(await authorizeChannel(lookup, workspaceId, channelId, userId)).toBe(
+      false,
+    );
+    expect(await authorizeChannel(lookup, workspaceId, channelId, userId)).toBe(
+      true,
+    );
+    expect(await authorizeChannel(lookup, workspaceId, channelId, userId)).toBe(
+      false,
+    );
+    expect(await authorizeChannel(lookup, workspaceId, channelId, userId)).toBe(
+      false,
+    );
+  });
+});

apps/collab-server/src/chat.ts

@@ -0,0 +1,400 @@
+import { createHmac, timingSafeEqual } from "node:crypto";
+import { createServer } from "node:http";
+import { and, eq } from "drizzle-orm";
+import IoRedis from "ioredis";
+import { WebSocketServer, type WebSocket } from "ws";
+import { z } from "zod";
+import { db } from "@tasks/database/client";
+import {
+  chatChannelMembers,
+  chatChannels,
+  workspaceMembers,
+} from "@tasks/database/schema";
+
+const uuid = z.string().uuid();
+const ticketClaims = z
+  .object({
+    iss: z.literal("echodo-web"),
+    aud: z.literal("echodo-chat-v1"),
+    scope: z.literal("chat:subscribe"),
+    sub: uuid,
+    workspaceId: uuid,
+    exp: z.number().int().positive(),
+  })
+  .strict();
+export type ChatTicket = z.infer<typeof ticketClaims>;
+const topicPattern =
+  /^chat:v1:workspace:([0-9a-f-]{36}):channel:([0-9a-f-]{36})$/;
+const subscribe = z
+  .object({ type: z.literal("subscribe"), channelId: uuid })
+  .strict();
+const unsubscribe = z
+  .object({ type: z.literal("unsubscribe"), channelId: uuid })
+  .strict();
+const clientMessage = z.discriminatedUnion("type", [subscribe, unsubscribe]);
+const MAX_CONNECTIONS = 2000;
+const MAX_CHANNELS = 64;
+const MAX_MESSAGE_BYTES = 512;
+const MAX_BUFFERED_BYTES = 64 * 1024;
+const HEARTBEAT_MS = 15_000;
+const AUTH_TIMEOUT_MS = 5_000;
+
+export function verifyTicket(value: string, secret: string): ChatTicket | null {
+  if (secret.length < 32 || value.length > 2048) return null;
+  const parts = value.split(".");
+  if (
+    parts.length !== 2 ||
+    !parts[0] ||
+    !/^[A-Za-z0-9_-]+$/.test(parts[0]) ||
+    !/^[A-Za-z0-9_-]+$/.test(parts[1])
+  )
+    return null;
+  const expected = createHmac("sha256", secret).update(parts[0]).digest();
+  const supplied = Buffer.from(parts[1], "base64url");
+  if (
+    supplied.length !== expected.length ||
+    !timingSafeEqual(supplied, expected)
+  )
+    return null;
+  try {
+    const parsed = ticketClaims.safeParse(
+      JSON.parse(Buffer.from(parts[0], "base64url").toString("utf8")),
+    );
+    if (
+      !parsed.success ||
+      parsed.data.exp <= Math.floor(Date.now() / 1000) ||
+      parsed.data.exp > Math.floor(Date.now() / 1000) + 60
+    )
+      return null;
+    return parsed.data;
+  } catch {
+    return null;
+  }
+}
+
+export function parseTopic(
+  topic: string,
+): { workspaceId: string; channelId: string } | null {
+  const match = topicPattern.exec(topic);
+  if (
+    !match ||
+    !uuid.safeParse(match[1]).success ||
+    !uuid.safeParse(match[2]).success
+  )
+    return null;
+  return { workspaceId: match[1], channelId: match[2] };
+}
+
+type Lookup = (
+  workspaceId: string,
+  channelId: string,
+  userId: string,
+) => Promise<{
+  kind: string;
+  workspaceMember: boolean;
+  channelMember: boolean;
+} | null>;
+export async function authorizeChannel(
+  lookup: Lookup,
+  workspaceId: string,
+  channelId: string,
+  userId: string,
+): Promise<boolean> {
+  try {
+    let timeout: ReturnType<typeof setTimeout> | undefined;
+    const row = await Promise.race([
+      lookup(workspaceId, channelId, userId),
+      new Promise<null>((resolve) => {
+        timeout = setTimeout(() => resolve(null), AUTH_TIMEOUT_MS);
+      }),
+    ]).finally(() => clearTimeout(timeout));
+    return (
+      !!row &&
+      row.workspaceMember &&
+      (row.kind === "public" ||
+        ((row.kind === "private" || row.kind === "direct") &&
+          row.channelMember))
+    );
+  } catch {
+    return false;
+  }
+}
+
+// Always query current membership; never cache authorization across notifications.
+const lookupChannel: Lookup = async (workspaceId, channelId, userId) => {
+  const [row] = await db
+    .select({
+      kind: chatChannels.kind,
+      workspaceMember: workspaceMembers.userId,
+      channelMember: chatChannelMembers.userId,
+    })
+    .from(chatChannels)
+    .leftJoin(
+      workspaceMembers,
+      and(
+        eq(workspaceMembers.workspaceId, chatChannels.workspaceId),
+        eq(workspaceMembers.userId, userId),
+      ),
+    )
+    .leftJoin(
+      chatChannelMembers,
+      and(
+        eq(chatChannelMembers.workspaceId, chatChannels.workspaceId),
+        eq(chatChannelMembers.channelId, chatChannels.id),
+        eq(chatChannelMembers.userId, userId),
+      ),
+    )
+    .where(
+      and(
+        eq(chatChannels.workspaceId, workspaceId),
+        eq(chatChannels.id, channelId),
+      ),
+    )
+    .limit(1);
+  return row
+    ? {
+        kind: row.kind,
+        workspaceMember: !!row.workspaceMember,
+        channelMember: !!row.channelMember,
+      }
+    : null;
+};
+
+type Peer = {
+  socket: WebSocket;
+  ticket: ChatTicket;
+  channels: Set<string>;
+  busy: boolean;
+};
+
+/** Separate port: Hocuspocus owns its existing HTTP/WebSocket upgrade handler. */
+export async function startChatServer(): Promise<void> {
+  const secret = process.env.CHAT_TICKET_SECRET?.trim();
+  const redisUrl = process.env.REDIS_URL?.trim();
+  if (!secret || secret.length < 32 || !redisUrl) {
+    console.warn(
+      "[chat] disabled: CHAT_TICKET_SECRET (>=32 characters) and REDIS_URL required",
+    );
+    return;
+  }
+  const port = Number(process.env.CHAT_PORT ?? 1235);
+  if (!Number.isInteger(port) || port < 1 || port > 65535)
+    throw new Error("Invalid CHAT_PORT");
+  const subscriber = new IoRedis(redisUrl, { maxRetriesPerRequest: 1 });
+  await subscriber.ping(); // Fail closed: never accept subscribers without Redis.
+  const http = createServer((_req, res) => {
+    res.writeHead(404).end();
+  });
+  const wss = new WebSocketServer({
+    noServer: true,
+    maxPayload: MAX_MESSAGE_BYTES,
+    perMessageDeflate: false,
+  });
+  const peers = new Set<Peer>();
+  const topicPeers = new Map<string, Set<Peer>>();
+  const send = (peer: Peer, value: object) => {
+    if (peer.socket.readyState !== 1) return;
+    if (peer.socket.bufferedAmount > MAX_BUFFERED_BYTES) {
+      peer.socket.close(1013);
+      return;
+    }
+    peer.socket.send(JSON.stringify(value));
+  };
+  const remove = (peer: Peer, topic: string) => {
+    peer.channels.delete(topic);
+    const set = topicPeers.get(topic);
+    set?.delete(peer);
+    if (set?.size === 0) {
+      topicPeers.delete(topic);
+      void subscriber.unsubscribe(topic).catch(() => {});
+    }
+  };
+  const revoke = (peer: Peer, topic: string, channelId: string) => {
+    remove(peer, topic);
+    send(peer, { type: "revoked", channelId });
+  };
+  const deliver = async (topic: string) => {
+    const ids = parseTopic(topic);
+    if (!ids) return;
+    for (const peer of [...(topicPeers.get(topic) ?? [])]) {
+      if (!peer.channels.has(topic)) continue;
+      if (
+        peer.ticket.exp <= Date.now() / 1000 ||
+        !(await authorizeChannel(
+          lookupChannel,
+          ids.workspaceId,
+          ids.channelId,
+          peer.ticket.sub,
+        ))
+      ) {
+        revoke(peer, topic, ids.channelId);
+        peer.socket.close(1008);
+        continue;
+      }
+      // Deliberately ignore Redis envelope including message bodies and keys.
+      if (peer.channels.has(topic))
+        send(peer, { type: "invalidate", channelId: ids.channelId });
+    }
+  };
+  subscriber.on("message", (topic) => {
+    void deliver(topic).catch(() => {});
+  });
+  subscriber.on("error", () => {
+    for (const peer of peers) peer.socket.close(1011);
+  });
+  subscriber.on("end", () => {
+    for (const peer of peers) peer.socket.close(1011);
+  });
+  http.on("upgrade", (req, socket, head) => {
+    if (
+      peers.size >= MAX_CONNECTIONS ||
+      req.url?.length === undefined ||
+      req.url.length > 2300
+    ) {
+      socket.destroy();
+      return;
+    }
+    let token: string | null;
+    try {
+      const url = new URL(req.url, "http://localhost");
+      token = url.pathname === "/chat" ? url.searchParams.get("ticket") : null;
+    } catch {
+      token = null;
+    }
+    const ticket = token ? verifyTicket(token, secret) : null;
+    if (!ticket) {
+      socket.destroy();
+      return;
+    }
+    wss.handleUpgrade(req, socket, head, (ws) =>
+      wss.emit("connection", ws, ticket),
+    );
+  });
+  wss.on("connection", (socket: WebSocket, ticket: ChatTicket) => {
+    const peer: Peer = { socket, ticket, channels: new Set(), busy: false };
+    peers.add(peer);
+    const expiry = setTimeout(
+      () => socket.close(1008),
+      Math.max(0, ticket.exp * 1000 - Date.now()),
+    );
+    expiry.unref();
+    socket.on("close", () => {
+      clearTimeout(expiry);
+      peers.delete(peer);
+      for (const topic of [...peer.channels]) remove(peer, topic);
+    });
+    socket.on("message", (bytes) => {
+      if (peer.busy) {
+        socket.close(1008);
+        return;
+      }
+      peer.busy = true;
+      void (async () => {
+        if (ticket.exp <= Date.now() / 1000) {
+          socket.close(1008);
+          return;
+        }
+        let data: unknown;
+        try {
+          data = JSON.parse(bytes.toString());
+        } catch {
+          socket.close(1008);
+          return;
+        }
+        const parsed = clientMessage.safeParse(data);
+        if (!parsed.success) {
+          socket.close(1008);
+          return;
+        }
+        const { channelId } = parsed.data;
+        const topic = `chat:v1:workspace:${ticket.workspaceId}:channel:${channelId}`;
+        if (parsed.data.type === "unsubscribe") {
+          remove(peer, topic);
+          return;
+        }
+        if (peer.channels.size >= MAX_CHANNELS && !peer.channels.has(topic)) {
+          socket.close(1008);
+          return;
+        }
+        if (
+          !(await authorizeChannel(
+            lookupChannel,
+            ticket.workspaceId,
+            channelId,
+            ticket.sub,
+          ))
+        ) {
+          socket.close(1008);
+          return;
+        }
+        if (peer.channels.has(topic)) return;
+        if (!topicPeers.has(topic)) {
+          try {
+            await subscriber.subscribe(topic);
+          } catch {
+            socket.close(1011);
+            return;
+          }
+          if (!topicPeers.has(topic)) topicPeers.set(topic, new Set());
+        }
+        // Recheck after the Redis subscribe await: access may have been revoked meanwhile.
+        if (
+          !(await authorizeChannel(
+            lookupChannel,
+            ticket.workspaceId,
+            channelId,
+            ticket.sub,
+          ))
+        ) {
+          socket.close(1008);
+          return;
+        }
+        if (socket.readyState !== 1) return;
+        topicPeers.get(topic)?.add(peer);
+        peer.channels.add(topic);
+        send(peer, { type: "subscribed", channelId });
+      })()
+        .catch(() => socket.close(1011))
+        .finally(() => {
+          peer.busy = false;
+        });
+    });
+  });
+  setInterval(() => {
+    for (const peer of peers) {
+      if (peer.busy) continue;
+      peer.busy = true;
+      void (async () => {
+        if (peer.ticket.exp <= Date.now() / 1000) {
+          peer.socket.close(1008);
+          return;
+        }
+        for (const topic of [...peer.channels]) {
+          const ids = parseTopic(topic);
+          if (
+            !ids ||
+            !(await authorizeChannel(
+              lookupChannel,
+              ids.workspaceId,
+              ids.channelId,
+              peer.ticket.sub,
+            ))
+          ) {
+            revoke(peer, topic, ids?.channelId ?? "");
+            peer.socket.close(1008);
+            return;
+          }
+        }
+      })()
+        .catch(() => peer.socket.close(1011))
+        .finally(() => {
+          peer.busy = false;
+        });
+    }
+  }, HEARTBEAT_MS).unref();
+  await new Promise<void>((resolve, reject) => {
+    http.once("error", reject);
+    http.listen(port, "0.0.0.0", resolve);
+  });
+  console.log(`[chat] listening on port ${port}`);
+}

apps/collab-server/src/index.ts

@@ -8,6 +8,7 @@ import { objects } from "@tasks/database/schema";
 import { eq } from "drizzle-orm";
 import IoRedis from "ioredis";
 import { z } from "zod";
+import { startChatServer } from "./chat.js";
 import {
   base64ToYUpdate,
   extractObjectId,
@@ -186,6 +187,8 @@ function colorFromName(name: string): string {
 async function main() {
   const port = Number(process.env.PORT) || 1234;
 
+  await startChatServer();
+
   const extensions: Extension[] = [buildDatabaseExtension()];
   const redis = await tryCreateRedisExtension();
   if (redis) {

apps/web/app/(app)/[workspaceSlug]/chat/[channelId]/page.tsx

@@ -0,0 +1,10 @@
+import { ChatWorkspace } from "@/components/chat/chat-workspace";
+
+export default async function ChannelPage({
+  params,
+}: {
+  params: Promise<{ workspaceSlug: string; channelId: string }>;
+}) {
+  const { workspaceSlug, channelId } = await params;
+  return <ChatWorkspace workspace={workspaceSlug} channelId={channelId} />;
+}

apps/web/app/(app)/[workspaceSlug]/chat/page.tsx

@@ -0,0 +1,10 @@
+import { ChatWorkspace } from "@/components/chat/chat-workspace";
+
+export default async function ChatPage({
+  params,
+}: {
+  params: Promise<{ workspaceSlug: string }>;
+}) {
+  const { workspaceSlug } = await params;
+  return <ChatWorkspace workspace={workspaceSlug} />;
+}

apps/web/app/(app)/[workspaceSlug]/fleet/page.tsx

@@ -0,0 +1,5 @@
+import { redirect } from "next/navigation";
+
+export default function WorkspaceFleetRedirect() {
+  redirect("/fleet");
+}

apps/web/app/api/chat/ticket/route.test.ts

@@ -0,0 +1,87 @@
+import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
+
+const { mockAuth } = vi.hoisted(() => ({
+  mockAuth: vi.fn<() => Promise<{ user: { id: string } } | null>>(),
+}));
+vi.mock("@/lib/auth", () => ({ auth: mockAuth }));
+vi.mock("@/server/lib/resolve-workspace", () => ({
+  resolveWorkspace: vi.fn(),
+}));
+import { auth } from "@/lib/auth";
+import { resolveWorkspace } from "@/server/lib/resolve-workspace";
+import { POST } from "./route";
+
+const publicOrigin = "https://echodo.stillwell.cloud";
+function request(
+  origin: string | null,
+  forwardedHost = "echodo.stillwell.cloud",
+) {
+  const headers = new Headers({
+    "Content-Type": "application/json",
+    "x-forwarded-host": forwardedHost,
+    "x-forwarded-proto": "https",
+  });
+  if (origin !== null) headers.set("origin", origin);
+  return new Request("http://0.0.0.0:3000/api/chat/ticket", {
+    method: "POST",
+    headers,
+    body: JSON.stringify({ workspace: "example" }),
+  });
+}
+
+beforeEach(() => {
+  vi.stubEnv("NODE_ENV", "production");
+  vi.stubEnv("NEXTAUTH_URL", publicOrigin);
+  vi.stubEnv("AUTH_URL", "");
+  vi.stubEnv("CHAT_TICKET_SECRET", "x".repeat(64));
+  mockAuth.mockResolvedValue({
+    user: { id: "10000000-0000-4000-8000-000000000001" },
+  });
+  vi.mocked(resolveWorkspace).mockResolvedValue({
+    id: "20000000-0000-4000-8000-000000000001",
+  } as Awaited<ReturnType<typeof resolveWorkspace>>);
+});
+afterEach(() => {
+  vi.unstubAllEnvs();
+  vi.clearAllMocks();
+});
+
+describe("Chat ticket origin boundary", () => {
+  it("accepts the configured public origin behind an internal Docker URL", async () => {
+    const response = await POST(request(publicOrigin));
+    expect(response.status).toBe(200);
+    expect(response.headers.get("cache-control")).toBe("no-store");
+    expect(resolveWorkspace).toHaveBeenCalled();
+  });
+  it.each([
+    null,
+    "null",
+    "https://evil.example",
+    `${publicOrigin}.evil.example`,
+    `${publicOrigin}/path`,
+  ])("rejects origin %s before session lookup", async (origin) => {
+    expect((await POST(request(origin, "evil.example"))).status).toBe(403);
+    expect(auth).not.toHaveBeenCalled();
+  });
+  it("does not trust forged forwarded host and protocol", async () => {
+    expect(
+      (await POST(request("https://evil.example", "evil.example"))).status,
+    ).toBe(403);
+  });
+  it("fails closed without a configured production origin", async () => {
+    vi.stubEnv("NEXTAUTH_URL", "");
+    expect((await POST(request("http://0.0.0.0:3000"))).status).toBe(403);
+  });
+  it("fails closed with an invalid configured origin", async () => {
+    vi.stubEnv("NEXTAUTH_URL", "not a URL");
+    expect((await POST(request(publicOrigin))).status).toBe(403);
+  });
+  it("still requires an authenticated session", async () => {
+    mockAuth.mockResolvedValue(null);
+    expect((await POST(request(publicOrigin))).status).toBe(401);
+  });
+  it("does not mint a ticket when workspace authorization fails", async () => {
+    vi.mocked(resolveWorkspace).mockRejectedValue(new Error("denied"));
+    expect((await POST(request(publicOrigin))).status).toBe(404);
+  });
+});

apps/web/app/api/chat/ticket/route.ts

@@ -0,0 +1,77 @@
+import { createHmac } from "node:crypto";
+import { auth } from "@/lib/auth";
+import { resolveWorkspace } from "@/server/lib/resolve-workspace";
+import { z } from "zod";
+
+export const runtime = "nodejs";
+export const dynamic = "force-dynamic";
+const input = z.object({ workspace: z.string().min(1).max(200) }).strict();
+const noStore = { "Cache-Control": "no-store" };
+
+function trustedOrigin(request: Request): string | null {
+  // Docker's request URL may be internal. Pin production to deployment configuration,
+  // not Host/forwarded headers that a caller could supply outside the trusted proxy.
+  const configured = process.env.AUTH_URL || process.env.NEXTAUTH_URL;
+  if (!configured && process.env.NODE_ENV === "production") return null;
+  const parsed = z
+    .string()
+    .url()
+    .safeParse(configured || request.url);
+  if (!parsed.success) return null;
+  const url = new URL(parsed.data);
+  if (
+    !["https:", "http:"].includes(url.protocol) ||
+    url.username ||
+    url.password
+  )
+    return null;
+  return url.origin;
+}
+
+export async function POST(request: Request): Promise<Response> {
+  // A browser session cookie must never be usable by a foreign origin to mint tickets.
+  const origin = request.headers.get("origin");
+  if (!origin || origin !== trustedOrigin(request))
+    return new Response(null, { status: 403, headers: noStore });
+  const session = await auth();
+  if (!session?.user?.id)
+    return new Response(null, { status: 401, headers: noStore });
+  const secret = process.env.CHAT_TICKET_SECRET?.trim();
+  if (!secret || secret.length < 32)
+    return new Response(null, { status: 503, headers: noStore });
+  let body: unknown;
+  try {
+    body = await request.json();
+  } catch {
+    return new Response(null, { status: 400, headers: noStore });
+  }
+  const parsed = input.safeParse(body);
+  if (!parsed.success)
+    return new Response(null, { status: 400, headers: noStore });
+  try {
+    const workspace = await resolveWorkspace({
+      handle: parsed.data.workspace,
+      userId: session.user.id,
+    });
+    const payload = Buffer.from(
+      JSON.stringify({
+        iss: "echodo-web",
+        aud: "echodo-chat-v1",
+        scope: "chat:subscribe",
+        sub: session.user.id,
+        workspaceId: workspace.id,
+        exp: Math.floor(Date.now() / 1000) + 60,
+      }),
+    ).toString("base64url");
+    const signature = createHmac("sha256", secret)
+      .update(payload)
+      .digest("base64url");
+    return Response.json(
+      { ticket: `${payload}.${signature}` },
+      { headers: noStore },
+    );
+  } catch {
+    // Do not distinguish nonexistent workspaces from denied or failed membership checks.
+    return new Response(null, { status: 404, headers: noStore });
+  }
+}

apps/web/app/layout.tsx

@@ -4,6 +4,7 @@ import Script from "next/script";
 import "@/styles/globals.css";
 import { ThemeProvider } from "@/components/providers/theme-provider";
 import { TRPCProvider } from "@/components/providers/trpc-provider";
+import { SessionProvider } from "next-auth/react";
 
 const inter = Inter({ subsets: ["latin"], variable: "--font-sans" });
 
@@ -30,7 +31,9 @@ export default function RootLayout({
           enableSystem
           disableTransitionOnChange
         >
+          <SessionProvider>
             <TRPCProvider>{children}</TRPCProvider>
+          </SessionProvider>
         </ThemeProvider>
         {umamiWebsiteId ? (
           <Script

apps/web/components/chat/chat-composer.tsx

@@ -0,0 +1,61 @@
+"use client";
+
+import { useId, useState, type FormEvent, type KeyboardEvent } from "react";
+import { Send } from "lucide-react";
+
+import { Button } from "@/components/ui/button";
+
+export function ChatComposer({
+  onSend,
+  disabled = false,
+  label = "Message",
+}: {
+  onSend: (body: string) => void;
+  disabled?: boolean;
+  label?: string;
+}) {
+  const [body, setBody] = useState("");
+  const id = useId();
+  const submit = (event?: FormEvent) => {
+    event?.preventDefault();
+    const text = body.trim();
+    if (!text || disabled) return;
+    onSend(text);
+    setBody("");
+  };
+  const onKeyDown = (event: KeyboardEvent<HTMLTextAreaElement>) => {
+    if (
+      event.key === "Enter" &&
+      !event.shiftKey &&
+      !event.nativeEvent.isComposing
+    ) {
+      event.preventDefault();
+      submit();
+    }
+  };
+  return (
+    <form onSubmit={submit} className="flex items-end gap-2 border-t p-3">
+      <label className="sr-only" htmlFor={id}>
+        {label}
+      </label>
+      <textarea
+        id={id}
+        value={body}
+        onChange={(event) => setBody(event.target.value)}
+        onKeyDown={onKeyDown}
+        disabled={disabled}
+        maxLength={10000}
+        rows={2}
+        placeholder={`${label} (Enter to send, Shift+Enter for a new line)`}
+        className="min-h-12 max-h-36 flex-1 resize-y rounded-md border border-input bg-background px-3 py-2 text-sm focus-visible:outline-none focus-visible:ring-2 focus-visible:ring-ring disabled:opacity-50"
+      />
+      <Button
+        type="submit"
+        disabled={disabled || !body.trim()}
+        aria-label="Send message"
+      >
+        <Send aria-hidden="true" /> Send
+      </Button>
+    </form>
+  );
+}

apps/web/components/chat/chat-message-list.test.tsx

@@ -0,0 +1,110 @@
+import { renderToStaticMarkup } from "react-dom/server";
+import { describe, expect, it, vi } from "vitest";
+
+import { ChatMessageList } from "./chat-message-list";
+
+const message = {
+  id: "server",
+  sequence: "2",
+  body: "Confirmed",
+  authorUserId: "user",
+  createdAt: "2026-01-01",
+};
+
+describe("rendered message states", () => {
+  it("shows a visible sending state and a retry only on failure", () => {
+    const pending = {
+      key: "same-key",
+      body: "Hello",
+      failed: false,
+      retry: vi.fn(),
+    };
+    const sending = renderToStaticMarkup(
+      <ChatMessageList messages={[]} pending={[pending]} />,
+    );
+    expect(sending).toContain("Sending…");
+    expect(sending).not.toContain("Retry");
+    const failed = renderToStaticMarkup(
+      <ChatMessageList
+        messages={[]}
+        pending={[
+          {
+            ...pending,
+            failed: true,
+            error: "Could not send message. Please retry.",
+          },
+        ]}
+      />,
+    );
+    expect(failed).toContain('role="alert"');
+    expect(failed).toContain("Retry");
+  });
+
+  it("renders only the messages supplied by the parent", () => {
+    const html = renderToStaticMarkup(<ChatMessageList messages={[message]} />);
+    expect(html).toContain("Confirmed");
+    expect(html).not.toContain("Sending…");
+    expect(html).not.toContain("Retry");
+
+    const empty = renderToStaticMarkup(<ChatMessageList messages={[]} />);
+    expect(empty).not.toContain("Confirmed");
+  });
+
+  it("shows an edited indicator for edited messages", () => {
+    const html = renderToStaticMarkup(
+      <ChatMessageList messages={[{ ...message, editedAt: "2026-01-02" }]} />,
+    );
+    expect(html).toContain("(edited)");
+  });
+
+  it("only offers owner edit, owner or moderator delete, and never archived or deleted controls", () => {
+    const own = renderToStaticMarkup(
+      <ChatMessageList
+        messages={[message]}
+        currentUserId="user"
+        onEdit={vi.fn()}
+        onDelete={vi.fn()}
+      />,
+    );
+    expect(own).toContain("Edit message server");
+    expect(own).toContain("Delete message server");
+    const moderator = renderToStaticMarkup(
+      <ChatMessageList
+        messages={[message]}
+        currentUserId="moderator"
+        canModerate
+        onEdit={vi.fn()}
+        onDelete={vi.fn()}
+      />,
+    );
+    expect(moderator).not.toContain("Edit message server");
+    expect(moderator).toContain("Delete message server");
+    const archived = renderToStaticMarkup(
+      <ChatMessageList
+        messages={[message]}
+        currentUserId="user"
+        readOnly
+        onEdit={vi.fn()}
+        onDelete={vi.fn()}
+      />,
+    );
+    expect(archived).not.toContain("Edit message server");
+    const deleted = renderToStaticMarkup(
+      <ChatMessageList
+        messages={[{ ...message, status: "deleted" }]}
+        currentUserId="user"
+        onEdit={vi.fn()}
+        onDelete={vi.fn()}
+      />,
+    );
+    expect(deleted).not.toContain("Delete message server");
+  });
+
+  it("renders deleted messages as a placeholder instead of their body", () => {
+    const html = renderToStaticMarkup(
+      <ChatMessageList messages={[{ ...message, deletedAt: "2026-01-02" }]} />,
+    );
+    expect(html).toContain("Message deleted");
+    expect(html).not.toContain("Confirmed");
+  });
+});

apps/web/components/chat/chat-message-list.tsx

@@ -0,0 +1,179 @@
+"use client";
+
+import { useEffect, useRef } from "react";
+
+import { Button } from "@/components/ui/button";
+import { cn } from "@/lib/utils";
+
+export type ChatMessageView = {
+  id: string;
+  sequence: string;
+  body: string;
+  authorUserId: string | null;
+  createdAt: Date | string;
+  editedAt?: Date | string | null;
+  deletedAt?: Date | string | null;
+  status?: string;
+  revision?: number;
+  replyCount?: number;
+  parentMessageId?: string | null;
+};
+
+export function ChatMessageList({
+  messages,
+  onViewed,
+  onThread,
+  onEdit,
+  onDelete,
+  canModerate = false,
+  currentUserId,
+  readOnly = false,
+  pending = [],
+}: {
+  messages: ChatMessageView[];
+  onViewed?: (sequence: string) => void;
+  onThread?: (id: string) => void;
+  onEdit?: (message: ChatMessageView) => void;
+  onDelete?: (message: ChatMessageView) => void;
+  canModerate?: boolean;
+  currentUserId?: string;
+  readOnly?: boolean;
+  pending?: {
+    key: string;
+    body: string;
+    failed: boolean;
+    error?: string;
+    retry: () => void;
+  }[];
+}) {
+  const viewed = useRef(onViewed);
+  useEffect(() => {
+    viewed.current = onViewed;
+  }, [onViewed]);
+
+  const list = useRef<HTMLOListElement>(null);
+  useEffect(() => {
+    if (
+      !viewed.current ||
+      !list.current ||
+      typeof IntersectionObserver === "undefined"
+    )
+      return;
+    const observer = new IntersectionObserver(
+      (entries) => {
+        if (document.visibilityState !== "visible") return;
+        const visible = entries
+          .filter((entry) => entry.isIntersecting)
+          .map((entry) => (entry.target as HTMLElement).dataset.sequence)
+          .filter((sequence): sequence is string => !!sequence);
+        const highest = visible.reduce<string | null>(
+          (max, sequence) =>
+            max === null || BigInt(sequence) > BigInt(max) ? sequence : max,
+          null,
+        );
+        if (highest) viewed.current?.(highest);
+      },
+      { threshold: 0.5 },
+    );
+    list.current
+      .querySelectorAll("[data-sequence]")
+      .forEach((element) => observer.observe(element));
+    return () => observer.disconnect();
+  }, [messages, pending]);
+
+  return (
+    <ol ref={list} aria-label="Messages" className="space-y-3 p-4">
+      {messages.map((message) => (
+        <li
+          key={message.id}
+          data-sequence={message.sequence}
+          className="rounded-md border p-3 text-sm"
+        >
+          <div className="flex flex-wrap items-baseline gap-2 text-xs text-muted-foreground">
+            <span className="font-semibold text-foreground">
+              {message.authorUserId}
+            </span>
+            <time dateTime={new Date(message.createdAt).toISOString()}>
+              {new Date(message.createdAt).toLocaleString()}
+            </time>
+            {message.editedAt && (
+              <span className="ml-1 text-xs text-muted-foreground">
+                (edited)
+              </span>
+            )}
+          </div>
+          <p className="mt-1 whitespace-pre-wrap break-words">
+            {message.deletedAt || message.status === "deleted"
+              ? "Message deleted"
+              : message.body}
+          </p>
+          {!readOnly && !message.deletedAt && message.status !== "deleted" && (
+            <div className="flex gap-2">
+              {onEdit && message.authorUserId === currentUserId && (
+                <Button
+                  type="button"
+                  variant="ghost"
+                  size="sm"
+                  onClick={() => onEdit(message)}
+                  aria-label={`Edit message ${message.id}`}
+                >
+                  Edit
+                </Button>
+              )}
+              {onDelete &&
+                (message.authorUserId === currentUserId || canModerate) && (
+                  <Button
+                    type="button"
+                    variant="ghost"
+                    size="sm"
+                    onClick={() => onDelete(message)}
+                    aria-label={`Delete message ${message.id}`}
+                  >
+                    Delete
+                  </Button>
+                )}
+            </div>
+          )}
+          {onThread && (
+            <Button
+              type="button"
+              variant="ghost"
+              size="sm"
+              onClick={() => onThread(message.id)}
+              aria-label={`Open replies to message by ${message.authorUserId}`}
+            >
+              {message.replyCount ? `${message.replyCount} replies` : "Reply"}
+            </Button>
+          )}
+        </li>
+      ))}
+      {pending.map((message) => (
+        <li
+          key={message.key}
+          className={cn(
+            "rounded-md border p-3 text-sm",
+            message.failed && "border-destructive",
+          )}
+        >
+          <p className="whitespace-pre-wrap break-words">{message.body}</p>
+          <p
+            role={message.failed ? "alert" : "status"}
+            className="text-xs text-muted-foreground"
+          >
+            {message.failed ? message.error : "Sending…"}
+          </p>
+          {message.failed && (
+            <Button
+              size="sm"
+              variant="outline"
+              type="button"
+              onClick={message.retry}
+            >
+              Retry send
+            </Button>
+          )}
+        </li>
+      ))}
+    </ol>
+  );
+}

apps/web/components/chat/chat-scope.test.ts

@@ -0,0 +1,79 @@
+import { describe, expect, it } from "vitest";
+import { createChatScope } from "./chat-utils";
+
+function deferred<T>() {
+  let resolve!: (value: T) => void;
+  let reject!: (error: Error) => void;
+  const promise = new Promise<T>((yes, no) => {
+    resolve = yes;
+    reject = no;
+  });
+  return { promise, resolve, reject };
+}
+
+describe("chat async scope", () => {
+  it("ignores a send resolved after a channel switch", async () => {
+    const scope = createChatScope("acme", "first");
+    const request = deferred<string>();
+    const result: string[] = [];
+    const valid = scope.capture();
+    const completion = request.promise.then((message) => {
+      if (valid()) result.push(message);
+    });
+    scope.select("acme", "second");
+    request.resolve("first-channel message");
+    await completion;
+    expect(result).toEqual([]);
+  });
+
+  it("ignores a late failed send, including a permission error", async () => {
+    const scope = createChatScope("acme", "first");
+    const request = deferred<string>();
+    const result: string[] = [];
+    const valid = scope.capture();
+    const completion = request.promise.catch((error: Error) => {
+      if (valid()) result.push(error.message);
+    });
+    scope.select("acme", "second");
+    request.reject(new Error("FORBIDDEN"));
+    await completion;
+    expect(result).toEqual([]);
+  });
+
+  it("ignores a stale mutation result and its subsequent refresh", async () => {
+    const scope = createChatScope("acme", "first");
+    const mutation = deferred<string>();
+    const refresh = deferred<string>();
+    const result: string[] = [];
+    const valid = scope.capture();
+    const completion = mutation.promise.then(async (value) => {
+      if (!valid()) return;
+      result.push(value);
+      const page = await refresh.promise;
+      if (valid()) result.push(page);
+    });
+    mutation.resolve("edited");
+    await Promise.resolve();
+    scope.select("acme", "second");
+    refresh.resolve("old page");
+    await completion;
+    expect(result).toEqual(["edited"]);
+  });
+
+  it("invalidates pending work on purge before redirect and cannot remove a new channel's cache", async () => {
+    const scope = createChatScope("acme", "first");
+    const cancellation = deferred<void>();
+    const result: string[] = [];
+    const valid = scope.capture();
+    scope.revoke();
+    result.push("redirect");
+    const cleanup = cancellation.promise.then(() => {
+      if (valid()) result.push("remove queries");
+    });
+    scope.select("acme", "second");
+    cancellation.resolve();
+    await cleanup;
+    expect(result).toEqual(["redirect"]);
+    expect(scope.capture()()).toBe(true);
+  });
+});

apps/web/components/chat/chat-utils.test.ts

@@ -0,0 +1,101 @@
+import { describe, expect, it } from "vitest";
+
+import {
+  advanceCatchUpWindow,
+  chatBadge,
+  mergeChatMessages,
+  messageError,
+  rootReadIsProven,
+} from "./chat-utils";
+
+describe("chat client state", () => {
+  it("retains through across pages, then starts a new interval after exhaustion", () => {
+    const first = advanceCatchUpWindow(
+      { after: "5" },
+      { next: "8", through: "10", hasMore: true },
+    );
+    expect(first).toEqual({ after: "8", through: "10" });
+    const second = advanceCatchUpWindow(first, {
+      next: "9",
+      through: "10",
+      hasMore: true,
+    });
+    expect(second).toEqual({ after: "9", through: "10" });
+    expect(
+      advanceCatchUpWindow(second, {
+        next: "9",
+        through: "10",
+        hasMore: false,
+      }),
+    ).toEqual({ after: "10" });
+  });
+
+  it("does not lose the cursor on a stalled page or fold concurrent sends into the old interval", () => {
+    const window = { after: "8", through: "10" };
+    expect(
+      advanceCatchUpWindow(window, { next: "8", through: "10", hasMore: true }),
+    ).toBe(window);
+    expect(
+      advanceCatchUpWindow(window, {
+        next: "10",
+        through: "10",
+        hasMore: false,
+      }),
+    ).toEqual({ after: "10" });
+    expect(
+      advanceCatchUpWindow(
+        { after: "10" },
+        { next: "11", through: "12", hasMore: true },
+      ),
+    ).toEqual({ after: "11", through: "12" });
+  });
+  it("never replaces a newer edit or tombstone with an older page", () => {
+    const original = {
+      id: "m",
+      sequence: "1",
+      revision: 1,
+      body: "old",
+      status: "active",
+    };
+    const edited = { ...original, revision: 2, body: "new" };
+    const deleted = { ...edited, revision: 3, body: "", status: "deleted" };
+    expect(mergeChatMessages([edited], [original])).toEqual([edited]);
+    expect(mergeChatMessages([deleted], [edited])).toEqual([deleted]);
+  });
+  it("requires a channel-wide scan and does not claim unseen replies as read", () => {
+    expect(rootReadIsProven("20", null, null, "0")).toBe(false);
+    expect(rootReadIsProven("20", "19", null, "0")).toBe(false);
+    expect(rootReadIsProven("20", "25", "18", "0")).toBe(false);
+    expect(rootReadIsProven("20", "25", "22", "0")).toBe(true);
+  });
+
+  it("deduplicates history and catch-up in sequence order", () => {
+    expect(
+      mergeChatMessages(
+        [
+          { id: "b", sequence: "9" },
+          { id: "a", sequence: "2" },
+        ],
+        [
+          { id: "b", sequence: "9" },
+          { id: "c", sequence: "10" },
+        ],
+      ).map((m) => m.id),
+    ).toEqual(["a", "b", "c"]);
+  });
+
+  it("distinguishes unread dots from numeric direct and mention activity", () => {
+    expect(
+      chatBadge({ unreadCount: 3, mentionCount: 0, directUnreadCount: 0 }),
+    ).toEqual({ dot: true, count: 0 });
+    expect(
+      chatBadge({ unreadCount: 3, mentionCount: 2, directUnreadCount: 1 }),
+    ).toEqual({ dot: true, count: 3 });
+  });
+
+  it("maps failed sends to actionable messages", () => {
+    expect(messageError("NOT_FOUND")).toMatch(/access/i);
+    expect(messageError("CONFLICT")).toMatch(/different message/i);
+    expect(messageError("INTERNAL_SERVER_ERROR")).toMatch(/retry/i);
+  });
+});

apps/web/components/chat/chat-utils.ts

@@ -0,0 +1,103 @@
+export function mergeChatMessages<
+  T extends { id: string; sequence: string; revision?: number },
+>(current: T[], incoming: T[]): T[] {
+  const byId = new Map(current.map((message) => [message.id, message]));
+  for (const message of incoming) {
+    const previous = byId.get(message.id);
+    if (!previous || (message.revision ?? 1) >= (previous.revision ?? 1))
+      byId.set(message.id, message);
+  }
+  return [...byId.values()].sort((a, b) => {
+    const left = BigInt(a.sequence);
+    const right = BigInt(b.sequence);
+    return left < right ? -1 : left > right ? 1 : 0;
+  });
+}
+
+// Each async continuation retains the generation it started in. A return to the
+// same channel is still a new generation, so old completions cannot re-enter it.
+export function createChatScope(workspace: string, channelId?: string) {
+  let key = JSON.stringify([workspace, channelId ?? ""]);
+  let generation = 0;
+  let revoked = false;
+  return {
+    select(nextWorkspace: string, nextChannelId?: string) {
+      const next = JSON.stringify([nextWorkspace, nextChannelId ?? ""]);
+      if (next === key) return;
+      key = next;
+      generation++;
+      revoked = false;
+    },
+    revoke() {
+      if (!revoked) {
+        revoked = true;
+        generation++;
+      }
+    },
+    dispose() {
+      revoked = true;
+      generation++;
+    },
+    capture() {
+      const captured = generation;
+      return () => !revoked && captured === generation;
+    },
+    captureCleanup() {
+      const captured = generation;
+      return () => revoked && captured === generation;
+    },
+    isRevoked() {
+      return revoked;
+    },
+  };
+}
+
+export type CatchUpWindow = { after: string; through?: string };
+
+// Keep the first response's high-water fixed until every page of that interval is read.
+export function advanceCatchUpWindow(
+  window: CatchUpWindow,
+  page: { next: string; through: string; hasMore: boolean },
+): CatchUpWindow {
+  if (page.hasMore) {
+    if (BigInt(page.next) <= BigInt(window.after)) return window;
+    return { after: page.next, through: window.through ?? page.through };
+  }
+  return { after: page.through };
+}
+
+export function rootReadIsProven(
+  sequence: string,
+  scanned: string | null,
+  unseenReply: string | null,
+  readBoundary: string,
+): boolean {
+  return (
+    scanned !== null &&
+    BigInt(scanned) >= BigInt(sequence) &&
+    (unseenReply === null ||
+      BigInt(unseenReply) > BigInt(sequence) ||
+      BigInt(unseenReply) <= BigInt(readBoundary))
+  );
+}
+
+export function chatBadge(summary: {
+  unreadCount: number;
+  mentionCount: number;
+  directUnreadCount: number;
+}) {
+  return {
+    dot: summary.unreadCount > 0,
+    count: summary.mentionCount + summary.directUnreadCount,
+  };
+}
+
+export function messageError(code?: string): string {
+  if (code === "NOT_FOUND" || code === "FORBIDDEN")
+    return "You no longer have access to this channel.";
+  if (code === "CONFLICT")
+    return "This request was already used for a different message. Please compose a new message.";
+  if (code === "BAD_REQUEST")
+    return "This message could not be sent. Check the text and retry.";
+  return "Could not send message. Please retry.";
+}

apps/web/components/chat/chat-workspace.test.tsx

@@ -0,0 +1,229 @@
+import { renderToStaticMarkup } from "react-dom/server";
+import { beforeEach, describe, expect, it, vi } from "vitest";
+
+import { ChatWorkspace, ReadFailureAlert } from "./chat-workspace";
+
+const mock = vi.hoisted(() => ({
+  channel: {
+    id: "channel",
+    kind: "public",
+    name: "General",
+    archivedAt: null as string | null,
+  },
+  channels: {
+    data: [] as unknown[],
+    isLoading: false,
+    isError: false,
+    error: null as unknown,
+    refetch: vi.fn(),
+  },
+  resolved: {
+    data: { id: "workspace" },
+    isLoading: false,
+    isError: false,
+    refetch: vi.fn(),
+  },
+  history: {
+    data: { messages: [] as unknown[], hasMore: false },
+    isLoading: false,
+    isError: false,
+    error: null as unknown,
+    refetch: vi.fn(),
+  },
+  thread: { data: null, isError: false, error: null, refetch: vi.fn() },
+  catchUp: { data: null, isError: false, error: null, refetch: vi.fn() },
+  summary: { data: { channels: [] as unknown[] }, error: null as unknown },
+  catchUpQuery: vi.fn(),
+  replace: vi.fn(),
+  recovery: { messages: [] as unknown[], error: null as unknown, refresh: vi.fn() },
+}));
+vi.mock("./use-chat-mutation-recovery", () => ({ useChatMutationRecovery: () => mock.recovery }));
+vi.mock("next-auth/react", () => ({
+  useSession: () => ({ data: { user: { id: "user" } } }),
+}));
+vi.mock("@/lib/use-chat-subscription", () => ({
+  useChatSubscription: vi.fn(),
+}));
+vi.mock("next/navigation", () => ({
+  useRouter: () => ({ replace: mock.replace }),
+}));
+vi.mock("@tanstack/react-query", () => ({
+  useQueryClient: () => ({ cancelQueries: vi.fn(), removeQueries: vi.fn() }),
+}));
+vi.mock("@/lib/trpc", () => ({
+  api: {
+    useUtils: () => ({ chat: {} }),
+    workspaces: { resolve: { useQuery: () => mock.resolved } },
+    chat: {
+      viewer: { useQuery: () => ({ data: { role: "member" } }) },
+      editMessage: {
+        useMutation: () => ({ mutateAsync: vi.fn(), isPending: false }),
+      },
+      deleteMessage: {
+        useMutation: () => ({ mutateAsync: vi.fn(), isPending: false }),
+      },
+      listChannels: { useQuery: () => mock.channels },
+      unreadSummary: { useQuery: () => mock.summary },
+      history: { useQuery: () => mock.history },
+      threadHistory: { useQuery: () => mock.thread },
+      catchUp: {
+        useQuery: (...args: unknown[]) => {
+          mock.catchUpQuery(...args);
+          return mock.catchUp;
+        },
+      },
+      send: { useMutation: () => ({ mutateAsync: vi.fn() }) },
+      markChannelRead: { useMutation: () => ({ mutate: vi.fn() }) },
+    },
+  },
+}));
+
+beforeEach(() => {
+  mock.recovery.messages = [];
+  mock.recovery.error = null;
+  mock.channel.archivedAt = null;
+  mock.channels.data = [mock.channel];
+  mock.channels.isLoading = false;
+  mock.channels.isError = false;
+  mock.resolved.isLoading = false;
+  mock.resolved.isError = false;
+  mock.history.data = { messages: [], hasMore: false };
+  mock.history.isLoading = false;
+  mock.history.isError = false;
+  mock.history.error = null;
+  mock.summary.data = { channels: [] };
+  mock.catchUp.error = null;
+  mock.catchUp.isError = false;
+  mock.catchUpQuery.mockClear();
+});
+
+const render = () =>
+  renderToStaticMarkup(<ChatWorkspace workspace="acme" channelId="channel" />);
+
+describe("rendered chat workspace", () => {
+  it("exposes a retryable read failure without claiming it advanced", () => {
+    const html = renderToStaticMarkup(<ReadFailureAlert onRetry={vi.fn()} />);
+    expect(html).toContain('role="alert"');
+    expect(html).toContain("Unread status has not advanced");
+    expect(html).toContain("Retry");
+  });
+  it("renders an ordinary unread dot and numeric mentions/direct activity", () => {
+    mock.channels.data = [
+      mock.channel,
+      { id: "dm", kind: "direct", name: null, archivedAt: null },
+    ];
+    mock.summary.data = {
+      channels: [
+        { channelId: "channel", unreadCount: 5, mentionCount: 0 },
+        { channelId: "dm", unreadCount: 3, mentionCount: 0 },
+      ],
+    };
+    const html = render();
+    expect(html).toContain('aria-label="Unread messages"');
+    expect(html).toContain('aria-label="3 mentions or direct unread messages"');
+    expect(html).not.toContain(">5</span>");
+  });
+
+  it("separates loading, transient failure and unavailable channel", () => {
+    mock.channels.isLoading = true;
+    expect(render()).toContain("Loading channel");
+    mock.channels.isLoading = false;
+    mock.channels.isError = true;
+    expect(render()).toContain("Could not load channel");
+    expect(render()).not.toContain("Channel unavailable");
+    mock.channels.isError = false;
+    mock.channels.data = [];
+    expect(render()).toContain("Channel unavailable");
+  });
+
+  it("shows history loading rather than a misleading empty state", () => {
+    mock.history.isLoading = true;
+    expect(render()).toContain("Loading messages");
+    expect(render()).not.toContain("No messages yet");
+  });
+
+  it("overlays a recovered tombstone on stale history without adding unloaded rows", () => {
+    const original = { id: "m", sequence: "1", revision: 1, body: "secret original", authorUserId: "user", createdAt: "2026-01-01", status: "active" };
+    mock.history.data.messages = [original];
+    mock.recovery.messages = [{ ...original, revision: 3, body: "", status: "deleted", deletedAt: "2026-01-02" }, { ...original, id: "unloaded", sequence: "2", body: "unloaded edit" }];
+    const html = render();
+    expect(html).not.toContain("secret original");
+    expect(html).not.toContain("unloaded edit");
+    expect(html).toContain("deleted");
+  });
+
+  it("hides history when mutation recovery discovers revoked access", () => {
+    mock.recovery.error = { data: { code: "NOT_FOUND" } };
+    expect(render()).toContain("Channel unavailable or access removed");
+  });
+
+  it("hides revoked content rather than showing a transient retry", () => {
+    mock.history.error = { data: { code: "NOT_FOUND" } };
+    mock.history.isError = true;
+    const html = render();
+    expect(html).toContain("Channel unavailable or access removed");
+    expect(html).not.toContain("Could not load messages");
+  });
+
+  it("shows transient history failure with retry", () => {
+    mock.history.isError = true;
+    const html = render();
+    expect(html).toContain("Could not load messages");
+    expect(html).toContain("Retry");
+  });
+
+  it("keeps archived channels readable without a channel composer", () => {
+    mock.channel.archivedAt = "2026-01-01";
+    const html = render();
+    expect(html).toContain('aria-label="Conversation: General"');
+    expect(html).not.toContain('aria-label="Send message"');
+  });
+
+  it("bounds initial catch-up to the oldest root predecessor when complete root history has no replies", () => {
+    mock.history.data = {
+      messages: [
+        {
+          id: "a",
+          sequence: "12",
+          replyCount: 0,
+          body: "First",
+          authorUserId: "user",
+          createdAt: "2026-01-01",
+        },
+        {
+          id: "b",
+          sequence: "30",
+          replyCount: 0,
+          body: "Second",
+          authorUserId: "user",
+          createdAt: "2026-01-01",
+        },
+      ],
+      hasMore: false,
+    };
+    render();
+    expect(mock.catchUpQuery.mock.calls.at(-1)?.[0]).toMatchObject({
+      after: "11",
+    });
+  });
+
+  it("does not skip replies before the initial root page", () => {
+    mock.history.data = {
+      messages: [
+        {
+          id: "a",
+          sequence: "30",
+          replyCount: 1,
+          body: "root",
+          authorUserId: "user",
+          createdAt: "2026-01-01",
+        },
+      ],
+      hasMore: false,
+    };
+    render();
+    expect(mock.catchUpQuery.mock.calls.at(-1)?.[0]).toMatchObject({
+      after: "0",
+    });
+  });
+});

apps/web/components/chat/mutation-recovery.test.ts

@@ -0,0 +1,47 @@
+import { describe, expect, it, vi } from "vitest";
+import { createMutationRecovery } from "./mutation-recovery";
+
+const page = (next: string, through: string, hasMore = false) => ({
+  messages: [{ id: "m", sequence: "1", revision: Number(next) }], next, through, hasMore,
+});
+
+describe("mutation recovery", () => {
+  it("drains a fixed high water, then resumes independently of creation sequences", async () => {
+    const fetch = vi.fn().mockResolvedValueOnce(page("2", "4", true)).mockResolvedValueOnce(page("4", "4")).mockResolvedValueOnce(page("5", "5"));
+    const apply = vi.fn();
+    const recovery = createMutationRecovery(fetch, apply);
+    await recovery.run();
+    await recovery.run();
+    expect(fetch.mock.calls.map(([cursor]) => cursor)).toEqual([{ after: "0" }, { after: "2", through: "4" }, { after: "4" }]);
+    expect(apply).toHaveBeenCalledTimes(3);
+  });
+  it("retries the same cursor after projection application fails", async () => {
+    const fetch = vi.fn().mockResolvedValue(page("1", "1"));
+    const apply = vi.fn().mockImplementationOnce(() => { throw new Error("merge failed"); });
+    const recovery = createMutationRecovery(fetch, apply);
+    await expect(recovery.run()).rejects.toThrow("merge failed");
+    await recovery.run();
+    expect(fetch.mock.calls.map(([cursor]) => cursor)).toEqual([{ after: "0" }, { after: "0" }]);
+  });
+  it("ignores a response after scope disposal", async () => {
+    let resolve!: (value: ReturnType<typeof page>) => void;
+    const fetch = vi.fn(() => new Promise<ReturnType<typeof page>>((done) => { resolve = done; }));
+    const apply = vi.fn();
+    const recovery = createMutationRecovery(fetch, apply);
+    const pending = recovery.run();
+    recovery.stop();
+    resolve(page("1", "1"));
+    await pending;
+    expect(apply).not.toHaveBeenCalled();
+  });
+  it("rejects stalled pages instead of spinning", async () => {
+    const recovery = createMutationRecovery(async () => page("0", "2", true), vi.fn());
+    await expect(recovery.run()).rejects.toThrow();
+  });
+  it("advances empty completed windows without looping", async () => {
+    const fetch = vi.fn().mockResolvedValue({ ...page("0", "0"), messages: [] });
+    const recovery = createMutationRecovery(fetch, vi.fn());
+    await recovery.run();
+    expect(fetch).toHaveBeenCalledTimes(1);
+  });
+});

apps/web/components/chat/mutation-recovery.ts

@@ -0,0 +1,45 @@
+import type { CatchUpWindow } from "./chat-utils";
+
+type Page<T> = { messages: T[]; next: string; through: string; hasMore: boolean };
+
+export function createMutationRecovery<T>(
+  fetchPage: (cursor: CatchUpWindow) => Promise<Page<T>>,
+  apply: (messages: T[]) => void,
+  initialCursor: CatchUpWindow = { after: "0" },
+  onCursor?: (cursor: CatchUpWindow) => void,
+) {
+  let cursor: CatchUpWindow = { ...initialCursor };
+  let stopped = false;
+  let active: Promise<void> | undefined;
+  let requested = false;
+  async function drain() {
+    do {
+      requested = false;
+      let more: boolean;
+      do {
+        const page = await fetchPage({ ...cursor });
+        if (stopped) return;
+        if (
+          BigInt(page.through) < BigInt(cursor.after) ||
+          BigInt(page.next) < BigInt(cursor.after) ||
+          BigInt(page.next) > BigInt(page.through) ||
+          (cursor.through !== undefined && cursor.through !== page.through) ||
+          (page.hasMore && BigInt(page.next) <= BigInt(cursor.after))
+        ) throw new Error("Invalid mutation recovery window");
+        apply(page.messages);
+        cursor = page.hasMore ? { after: page.next, through: page.through } : { after: page.through };
+        onCursor?.({ ...cursor });
+        more = page.hasMore;
+      } while (more && !stopped);
+    } while (requested && !stopped);
+  }
+  return {
+    run(): Promise<void> {
+      if (stopped) return Promise.resolve();
+      if (active) { requested = true; return active; }
+      active = drain().finally(() => { active = undefined; });
+      return active;
+    },
+    stop() { stopped = true; },
+  };
+}

apps/web/components/chat/use-chat-mutation-recovery.tsx

@@ -0,0 +1,64 @@
+"use client";
+
+import { useCallback, useEffect, useRef, useState } from "react";
+import { api } from "@/lib/trpc";
+import type { ChatMessageView } from "./chat-message-list";
+import { mergeChatMessages } from "./chat-utils";
+import { createMutationRecovery } from "./mutation-recovery";
+
+export function useChatMutationRecovery({ workspace, channelId, enabled }: {
+  workspace: string;
+  channelId?: string;
+  enabled: boolean;
+}) {
+  const utils = api.useUtils();
+  const identity = JSON.stringify([workspace, channelId, enabled]);
+  const scope = useRef({ identity });
+  if (scope.current.identity !== identity) scope.current = { identity };
+  const token = scope.current;
+  const [state, setState] = useState<{ token: typeof token; messages: ChatMessageView[]; error: unknown }>({ token, messages: [], error: null });
+  const cursors = useRef(new Map<string, { after: string; through?: string }>());
+  const run = useRef<(() => void) | undefined>(undefined);
+  const cursorKey = JSON.stringify([workspace, channelId ?? ""]);
+  const refresh = useCallback(() => { run.current?.(); }, []);
+
+  useEffect(() => {
+    if (!enabled || !channelId) return;
+    let stopped = false;
+    const valid = () => !stopped && scope.current === token;
+    const cursorKey = JSON.stringify([workspace, channelId]);
+    const recovery = createMutationRecovery(
+      (cursor) => utils.chat.mutationCatchUp.fetch({ workspace, channelId, ...cursor, limit: 100 }),
+      (messages) => {
+        if (!valid()) return;
+        setState((previous) => ({ token, messages: mergeChatMessages(previous.token === token ? previous.messages : [], messages), error: null }));
+      },
+      cursors.current.get(cursorKey),
+      (cursor) => cursors.current.set(cursorKey, cursor),
+    );
+    const recover = () => {
+      void recovery.run().catch((error: unknown) => {
+        if (valid()) setState((previous) => ({ token, messages: previous.token === token ? previous.messages : [], error }));
+      });
+    };
+    run.current = recover;
+    recover();
+    const timer = setInterval(recover, 15000);
+    window.addEventListener("online", recover);
+    window.addEventListener("focus", recover);
+    return () => {
+      stopped = true;
+      recovery.stop();
+      clearInterval(timer);
+      window.removeEventListener("online", recover);
+      window.removeEventListener("focus", recover);
+      if (run.current === recover) run.current = undefined;
+    };
+  }, [workspace, channelId, enabled, token, utils]);
+
+  return {
+    messages: state.token === token && enabled ? state.messages : [],
+    error: state.token === token && enabled ? state.error : null,
+    refresh,
+  };
+}

apps/web/components/layout/icon-rail.tsx

@@ -3,6 +3,7 @@
 import type { ComponentType } from "react";
 import {
   CalendarDays,
+  MessageCircle,
   ClipboardList,
   FileText,
   Home,
@@ -16,9 +17,15 @@ import {
 } from "lucide-react";
 import { useParams, usePathname, useRouter } from "next/navigation";
 
-import { Tooltip, TooltipContent, TooltipTrigger } from "@/components/ui/tooltip";
+import {
+  Tooltip,
+  TooltipContent,
+  TooltipTrigger,
+} from "@/components/ui/tooltip";
 import { useWorkspaceStore } from "@/lib/stores/workspace-store";
 import { cn } from "@/lib/utils";
+import { api } from "@/lib/trpc";
+import { chatBadge } from "@/components/chat/chat-utils";
 
 export interface IconRailProps {
   className?: string;
@@ -35,11 +42,22 @@ const NAV_ITEMS: NavItem[] = [
   { id: "home", label: "Home", segment: "", icon: Home },
   { id: "spaces", label: "Spaces", segment: "", icon: LayoutGrid },
   { id: "planner", label: "Planner", segment: "planner", icon: CalendarDays },
-  { id: "mission-control", label: "Mission control", segment: "mission-control", icon: Radar },
+  {
+    id: "mission-control",
+    label: "Mission control",
+    segment: "mission-control",
+    icon: Radar,
+  },
   { id: "ai", label: "AI", segment: "ai", icon: Sparkles },
+  { id: "chat", label: "Chat", segment: "chat", icon: MessageCircle },
   { id: "teams", label: "Teams", segment: "teams", icon: Users },
   { id: "docs", label: "Docs", segment: "docs", icon: FileText },
-  { id: "whiteboards", label: "Whiteboards", segment: "whiteboards", icon: Presentation },
+  {
+    id: "whiteboards",
+    label: "Whiteboards",
+    segment: "whiteboards",
+    icon: Presentation,
+  },
   { id: "forms", label: "Forms", segment: "forms", icon: ClipboardList },
 ];
 
@@ -51,6 +69,7 @@ const RESERVED_ROOT_SEGMENTS = new Set([
   "planner",
   "mission-control",
   "ai",
+  "chat",
   "teams",
   "docs",
   "whiteboards",
@@ -74,7 +93,8 @@ function isNavItemActive(
   if (!pathname.startsWith(prefix)) return false;
 
   if (item.segment) {
-    const next = pathname[prefix.length] === "/" ? pathname.slice(prefix.length + 1) : "";
+    const next =
+      pathname[prefix.length] === "/" ? pathname.slice(prefix.length + 1) : "";
     const first = next.split("/")[0] ?? "";
     return first === item.segment;
   }
@@ -99,11 +119,39 @@ export function IconRail({ className }: IconRailProps) {
   const slugParam =
     typeof params?.workspaceSlug === "string" ? params.workspaceSlug : "";
   const workspaceSegment =
-    currentWorkspace?.slug ||
+    currentWorkspace?.slug || currentWorkspace?.id || slugParam || undefined;
-    currentWorkspace?.id ||
-    slugParam ||
-    undefined;
   const workspaceName = currentWorkspace?.name ?? "";
+  const unread = api.chat.unreadSummary.useQuery(
+    { workspace: workspaceSegment ?? "" },
+    { enabled: !!workspaceSegment, refetchInterval: 15000 },
+  );
+  const visibleChannels = api.chat.listChannels.useQuery(
+    { workspace: workspaceSegment ?? "" },
+    { enabled: !!workspaceSegment, refetchInterval: 15000 },
+  );
+  const directIds = new Set(
+    visibleChannels.data
+      ?.filter((channel) => channel.kind === "direct")
+      .map((channel) => channel.id),
+  );
+  const badge = chatBadge({
+    unreadCount:
+      unread.data?.channels.reduce(
+        (total, channel) => total + channel.unreadCount,
+        0,
+      ) ?? 0,
+    mentionCount:
+      unread.data?.channels.reduce(
+        (total, channel) => total + channel.mentionCount,
+        0,
+      ) ?? 0,
+    directUnreadCount:
+      unread.data?.channels.reduce(
+        (total, channel) =>
+          total + (directIds.has(channel.channelId) ? channel.unreadCount : 0),
+        0,
+      ) ?? 0,
+  });
 
   const initials = workspaceName.trim().slice(0, 2).toUpperCase() || "—";
 
@@ -121,8 +169,6 @@ export function IconRail({ className }: IconRailProps) {
         className,
       )}
     >
-
-
       <nav
         className="flex min-h-0 flex-1 flex-col items-stretch overflow-y-auto overflow-x-hidden px-1 py-2"
         aria-label="Main"
@@ -136,6 +182,11 @@ export function IconRail({ className }: IconRailProps) {
                 <button
                   type="button"
                   disabled={!workspaceSegment}
+                  aria-label={
+                    item.id === "chat" && badge.dot
+                      ? `Chat, ${badge.count ? `${badge.count} mentions and unread direct messages` : "unread messages"}`
+                      : item.label
+                  }
                   onClick={() => push(item.segment)}
                   className={cn(
                     "w-full cursor-pointer rounded-lg px-1 py-2 transition-colors",
@@ -145,7 +196,23 @@ export function IconRail({ className }: IconRailProps) {
                       : "text-white/70 hover:bg-white/10",
                   )}
                 >
+                  <span className="relative">
                     <Icon className="h-5 w-5 shrink-0" aria-hidden />
+                    {item.id === "chat" && badge.dot && (
+                      <span
+                        className="absolute -right-1 -top-1 size-2 rounded-full bg-rose-400"
+                        aria-hidden
+                      />
+                    )}
+                  </span>
+                  {item.id === "chat" && badge.count > 0 && (
+                    <span
+                      className="rounded-full bg-rose-500 px-1 text-[10px] text-white"
+                      aria-label={`${badge.count} mentions and unread direct messages`}
+                    >
+                      {badge.count > 99 ? "99+" : badge.count}
+                    </span>
+                  )}
                   <span
                     className={cn(
                       "max-w-[52px] truncate text-center text-[10px] font-medium leading-tight",

apps/web/lib/use-chat-subscription.ts

@@ -0,0 +1,123 @@
+"use client";
+
+import { useEffect, useRef } from "react";
+import { useQueryClient } from "@tanstack/react-query";
+import { getCollabUrl } from "@/lib/yjs-provider";
+
+/** Invalidation-only transport. Durable chat history must be fetched through the authenticated API. */
+export function useChatSubscription(
+  workspace: string,
+  channelIds: readonly string[],
+  onHint?: () => void,
+): void {
+  const hint = useRef(onHint);
+  hint.current = onHint;
+  const queryClient = useQueryClient();
+  const channels = [...new Set(channelIds)].sort().join(",");
+  useEffect(() => {
+    if (!workspace || !channels) return;
+    let socket: WebSocket | undefined;
+    let timer: ReturnType<typeof setTimeout> | undefined;
+    let stopped = false;
+    let attempts = 0;
+    let revoked = false;
+    const invalidate = () => {
+      if (stopped) return;
+      hint.current?.();
+      void queryClient.invalidateQueries({ predicate: (query) => {
+        const [path, options] = query.queryKey;
+        return Array.isArray(path) && path[0] === "chat" &&
+          !!options && typeof options === "object" && "input" in options &&
+          !!options.input && typeof options.input === "object" &&
+          "workspace" in options.input && options.input.workspace === workspace;
+      } });
+    };
+    async function connect() {
+      revoked = false;
+      try {
+        const response = await fetch("/api/chat/ticket", {
+          method: "POST",
+          credentials: "same-origin",
+          headers: { "Content-Type": "application/json" },
+          body: JSON.stringify({ workspace }),
+          cache: "no-store",
+        });
+        if (!response.ok) {
+          if (
+            response.status === 401 ||
+            response.status === 403 ||
+            response.status === 404
+          )
+            return;
+          throw new Error("Chat ticket unavailable");
+        }
+        const data: unknown = await response.json();
+        if (
+          !data ||
+          typeof data !== "object" ||
+          !("ticket" in data) ||
+          typeof data.ticket !== "string"
+        )
+          throw new Error("Invalid chat ticket");
+        if (stopped) return;
+        const url = new URL(
+          process.env.NEXT_PUBLIC_CHAT_SERVER_URL || getCollabUrl(),
+        );
+        if (!process.env.NEXT_PUBLIC_CHAT_SERVER_URL) url.port = "1235";
+        url.pathname = "/chat";
+        url.search = new URLSearchParams({ ticket: data.ticket }).toString();
+        socket = new WebSocket(url);
+        socket.onopen = () => {
+          attempts = 0;
+          invalidate(); // catch up after missed Redis notifications
+          for (const channelId of channels.split(","))
+            socket?.send(JSON.stringify({ type: "subscribe", channelId }));
+        };
+        socket.onmessage = (event) => {
+          try {
+            const message: unknown = JSON.parse(String(event.data));
+            if (
+              message &&
+              typeof message === "object" &&
+              "type" in message &&
+              (message.type === "invalidate" ||
+                message.type === "revoked" ||
+                message.type === "subscribed")
+            ) {
+              if (message.type === "revoked") revoked = true;
+              invalidate();
+            }
+          } catch {
+            /* ignore malformed hints */
+          }
+        };
+        socket.onclose = (event) => {
+          if (event.code === 1008 && revoked) {
+            invalidate();
+          }
+          if (!stopped)
+            timer = setTimeout(
+              () => {
+                void connect();
+              },
+              Math.min(30_000, 1000 * 2 ** Math.min(attempts++, 5)),
+            );
+        };
+      } catch {
+        if (!stopped)
+          timer = setTimeout(
+            () => {
+              void connect();
+            },
+            Math.min(30_000, 1000 * 2 ** Math.min(attempts++, 5)),
+          );
+      }
+    }
+    void connect();
+    return () => {
+      stopped = true;
+      clearTimeout(timer);
+      socket?.close();
+    };
+  }, [workspace, channels, queryClient]);
+}

apps/web/server/routers/chat.test.ts

@@ -1,205 +1,116 @@
-import { describe, expect, it, vi } from "vitest";
+import { beforeEach, describe, expect, it, vi } from "vitest";
-import { TRPCError } from "@trpc/server";
+import { db } from "@tasks/database";
-import { createChatTransactionRunner } from "@tasks/database/chat";
+import type { Session } from "next-auth";
-import { listChannels, createChannel, renameChannel, archiveChannel, addPrivateMember, removePrivateMember, createDirectChannel } from "./manage-channels";
+import {
+  ChatMutationError,
+  mutateChatMessage,
+  createMessageMutationRunner,
+  chatMutationCatchUp,
+} from "@tasks/database/chat";
 import { resolveWorkspace } from "@/server/lib/resolve-workspace";
-import { auth } from "@/lib/auth";
+import { createCallerFactory } from "@/server/trpc";
-import { ChatPersistenceError } from "@tasks/database/chat";
+import { chatRouter } from "./chat";
+
+vi.mock("@tasks/database/chat", async (importOriginal) => ({
+  ...(await importOriginal<typeof import("@tasks/database/chat")>()),
+  mutateChatMessage: vi.fn(),
+  chatMutationCatchUp: vi.fn(),
+  createMessageMutationRunner: vi.fn(() => ({ transaction: vi.fn() })),
+}));
+vi.mock("@/server/lib/resolve-workspace", () => ({
+  resolveWorkspace: vi.fn(),
+}));
+vi.mock("@/lib/auth", () => ({ auth: vi.fn() }));
 
 const workspaceId = "11111111-1111-4111-8111-111111111111";
-const otherWorkspaceId = "22222222-2222-4222-8222-222222222222";
 const channelId = "33333333-3333-4333-8333-333333333333";
 const userId = "44444444-4444-4444-8444-444444444444";
-const parentMessageId = "55555555-5555-4555-8555-555555555555";
+const messageId = "77777777-7777-4777-8777-777777777777";
-const mentionId = "66666666-6666-4666-8666-666666666666";
+const clientMutationId = "66666666-6666-4666-8666-666666666666";
+const caller = () =>
+  createCallerFactory(chatRouter)({
+    db,
+    session: { user: { id: userId } } as Session,
+  });
+const command = {
+  type: "edit_message" as const,
+  workspaceId,
+  channelId,
+  context: { type: "workspace" as const, workspaceId },
+  messageId,
+  body: "corrected",
+  expectedRevision: 2,
+  clientMutationId,
+};
 
-function command(overrides: Record<string, unknown> = {}) {
+beforeEach(() => {
-  return {
+  vi.clearAllMocks();
-    workspace: workspaceId,
+  vi.mocked(resolveWorkspace).mockResolvedValue({
-    command: { type: "send_message", workspaceId, channelId, context: { type: "workspace", workspaceId }, body: "hello", clientIdempotencyKey: "client-1", ...overrides },
+    id: workspaceId,
+    slug: "acme",
+    name: "Acme",
+    role: "member",
+    ownerUserId: userId,
+  });
+});
+
+describe("chat message mutation procedures", () => {
+  it("delegates validated edit and delete with actor and exact command", async () => {
+    const result = {
+      changed: true,
+      result: { messageId, revision: 3, status: "edited" as const },
     };
-}
+    vi.mocked(mutateChatMessage).mockResolvedValue(result);
-
+    await expect(caller().editMessage({ workspace: workspaceId, command })).resolves.toEqual(result);
-function harness(options: {
+    expect(mutateChatMessage).toHaveBeenCalledWith(expect.anything(), {
-  kind?: "public" | "private";
+      actorUserId: userId,
-  archived?: boolean;
+      command,
-  channelMember?: boolean;
-  workspaceMember?: boolean;
-  existing?: boolean;
-  parent?: {
-    workspaceId: string;
-    channelId: string;
-    parentMessageId: string | null;
-  };
-  validMentions?: string[];
-}) {
-  return {
-    kind: options.kind,
-    archived: options.archived,
-    channelMember: options.channelMember,
-    workspaceMember: options.workspaceMember,
-    existing: options.existing,
-    parent: options.parent,
-    validMentions: options.validMentions,
-  });
-}
-
-describe("Chat router lifecycle", () => {
-  it("lists visible channels with correct workspace scoping", async () => {
-    // Mock workspace resolution
-    vi.spyOn(resolveWorkspace, "default").mockImplementation(() => workspaceId);
-    // Mock channel list
-    const mockChannels = [
-      { id: "1", workspaceId: workspaceId, kind: "public", name: "Public Channel" },
-      { id: "2", workspaceId: workspaceId, kind: "private", name: "Private Channel" },
-    ];
-    vi.spyOn(createChatTransactionRunner, "run").mockResolvedValueOnce({ runner: true });
-
-    // Mock channel retrieval
-    const listVisible = await createChatTransactionRunner(workspaceId).listVisible(workspaceId, userId);
-    expect(listVisible).toEqual(mockChannels);
-  });
-
-  it("creates public channel with owner membership", async () => {
-    // Mock workspace resolution
-    vi.spyOn(resolveWorkspace, "default").mockImplementation(() => workspaceId);
-    // Mock channel creation
-    const mockChannel = { id: "4", workspaceId: workspaceId, kind: "public", name: "New Public", ownerUserId: userId, directKey: null, archivedAt: null };
-    vi.spyOn(createChatTransactionRunner, "run").mockImplementation(async (tx) => {
-      const result = await tx.createChannel(workspaceId, "New Public", "public");
-      return { runner: true, channel: result };
-    });
-
-    // Mock transaction runner
-    const runner = createChatTransactionRunner(workspaceId);
-    const result = await runner.createChannel(workspaceId, "New Public", "public");
-    expect(runner).toHaveBeenCalledTimes(1);
-    expect(result).toMatchObject(mockChannel);
-  });
-
-  it("creates private channel with owner and member", async () => {
-    // Mock workspace resolution
-    vi.spyOn(resolveWorkspace, "default").mockImplementation(() => workspaceId);
-    // Mock channel creation
-    const mockChannel = { id: "5", workspaceId: workspaceId, kind: "private", name: "New Private", ownerUserId: userId, directKey: null, archivedAt: null };
-    vi.spyOn(createChatTransactionRunner, "run").mockImplementation(async (tx) => {
-      const result = await tx.createChannel(workspaceId, "New Private", "private");
-      return { runner: true, channel: result };
-    });
-
-    const runner = createChatTransactionRunner(workspaceId);
-    const result = await runner.createChannel(workspaceId, "New Private", "private");
-    expect(runner).toHaveBeenCalledTimes(1);
-    expect(result).toMatchObject(mockChannel);
-  });
-
-  it("renames channel with owner permission", async () => {
-    // Mock workspace resolution
-    vi.spyOn(resolveWorkspace, "default").mockImplementation(() => workspaceId);
-    // Mock rename
-    const mockChannel = { id: "6", workspaceId: workspaceId, kind: "private", name: "Renamed", ownerUserId: userId, directKey: null, archivedAt: null };
-    vi.spyOn(createChatTransactionRunner, "run").mockImplementation(async (tx) => {
-      const result = await tx.renameChannel(workspaceId, channelId, "Renamed");
-      return { runner: true, channel: result };
-    });
-
-    const runner = createChatTransactionRunner(workspaceId);
-    const result = await runner.renameChannel(workspaceId, channelId, "Renamed");
-    expect(runner).toHaveBeenCalledTimes(1);
-    expect(result).toMatchObject(mockChannel);
-  });
-
-  it("archives channel when owner", async () => {
-    // Mock workspace resolution
-    vi.spyOn(resolveWorkspace, "default").mockImplementation(() => workspaceId);
-    // Mock archive
-    const mockChannel = { id: "7", workspaceId: workspaceId, kind: "private", name: "Archived", ownerUserId: userId, directKey: null, archivedAt: new Date() };
-    vi.spyOn(createChatTransactionRunner, "run").mockImplementation(async (tx) => {
-      const result = await tx.archiveChannel(workspaceId, channelId);
-      return { runner: true, channel: result };
-    });
-
-    const runner = createChatTransactionRunner(workspaceId);
-    const result = await runner.archiveChannel(workspaceId, channelId);
-    expect(runner).toHaveBeenCalledTimes(1);
-    expect(result).toMatchObject(mockChannel);
-  });
-
-  it("adds private member with owner permission", async () => {
-    // Mock workspace resolution
-    vi.spyOn(resolveWorkspace, "default").mockImplementation(() => workspaceId);
-    // Mock member addition
-    const mockChannel = { id: "8", workspaceId: workspaceId, kind: "private", name: "With Member", ownerUserId: userId, directKey: null, archivedAt: null, memberCount: 2 };
-    vi.spyOn(createChatTransactionRunner, "run").mockImplementation(async (tx) => {
-      const result = await tx.addPrivateMember(workspaceId, channelId, "66666666-6666-4666-8666-666666666666");
-      return { runner: true, channel: result };
-    });
-
-    const runner = createChatTransactionRunner(workspaceId);
-    const result = await runner.addPrivateMember(workspaceId, channelId, "66666666-6666-4666-8666-666666666666");
-    expect(runner).toHaveBeenCalledTimes(1);
-    expect(result.memberCount).toBe(2);
-  });
-
-  it("removes private member", async () => {
-    // Mock workspace resolution
-    vi.spyOn(resolveWorkspace, "default").mockImplementation(() => workspaceId);
-    // Mock member removal
-    const mockChannel = { id: "9", workspaceId: workspaceId, kind: "private", name: "Without Member", ownerUserId: userId, directKey: null, archivedAt: null, memberCount: 1 };
-    vi.spyOn(createChatTransactionRunner, "run").mockImplementation(async (tx) => {
-      const result = await tx.removePrivateMember(workspaceId, channelId, "44444444-4444-4444-8444-444444444444");
-      return { runner: true, channel: result };
-    });
-
-    const runner = createChatTransactionRunner(workspaceId);
-    const result = await runner.removePrivateMember(workspaceId, channelId, "44444444-4444-4444-8444-444444444444");
-    expect(runner).toHaveBeenCalledTimes(1);
-    expect(result.memberCount).toBe(1);
-  });
-
-  it("creates direct channel with exact participant set", async () => {
-    // Mock workspace resolution
-    vi.spyOn(resolveWorkspace, "default").mockImplementation(() => workspaceId);
-    // Mock direct channel creation
-    const participantIds = ["44444444-4444-4444-8444-444444444444", "55555555-5555-4555-8555-555555555555"];
-    const mockChannel = { id: "10", workspaceId: workspaceId, kind: "direct", name: "Direct", ownerUserId: userId, directKey: "44444444-4444-4444-8444-444444444444:55555555-5555-4555-8555-555555555555", archivedAt: null };
-    vi.spyOn(createChatTransactionRunner, "run").mockImplementation(async (tx) => {
-      const result = await tx.createDirectChannel(workspaceId, participantIds);
-      return { runner: true, channel: result };
-    });
-
-    const runner = createChatTransactionRunner(workspaceId);
-    const result = await runner.createDirectChannel(workspaceId, participantIds);
-    expect(runner).toHaveBeenCalledTimes(1);
-    expect(result).toMatchObject(mockChannel);
-  });
-
-  it("rejects cross-workspace channel creation", async () => {
-    // Mock workspace resolution to other workspace
-    vi.spyOn(resolveWorkspace, "default").mockImplementation(() => otherWorkspaceId);
-    // Mock channel creation attempt
-    vi.spyOn(createChatTransactionRunner, "run").mockImplementation(async (tx) => {
-      const result = await tx.createChannel(otherWorkspaceId, "Cross Workspace", "public");
-      return { runner: true, channel: result };
-    });
-
-    const runner = createChatTransactionRunner(otherWorkspaceId);
-    const result = await runner.createChannel(otherWorkspaceId, "Cross Workspace", "public");
-    expect(result).toBeUndefined(); // Should fail due to workspace mismatch
-  });
-
-  it("maps channel persistence errors correctly", async () => {
-    // Mock workspace resolution
-    vi.spyOn(resolveWorkspace, "default").mockImplementation(() => workspaceId);
-    // Mock channel creation failure
-    vi.spyOn(createChatTransactionRunner, "run").mockImplementation(async (tx) => {
-      throw new ChatPersistenceError("NOT_FOUND");
-    });
-
-    const runner = createChatTransactionRunner(workspaceId);
-    await expect(runner.createChannel(workspaceId, "Invalid", "private")).rejects.toMatchObject({
-      code: "NOT_FOUND",
-      message: "Chat channel not found",
     });
   });
 });
+
+describe("mutation recovery procedure", () => {
+  const input = { workspace: "acme", channelId, after: "0", limit: 2 };
+  const result = {
+    mutations: [{ messageId, mutationSequence: "1", revision: 3, kind: "edited" as const }],
+    next: "1",
+    through: "3",
+    hasMore: true,
+  };
+
+  it("passes the resolved workspace and session actor, and preserves the fixed high water", async () => {
+    vi.mocked(chatMutationCatchUp).mockResolvedValue(result as Awaited<ReturnType<typeof chatMutationCatchUp>>);
+    await expect(caller().mutationCatchUp(input)).resolves.toEqual(result);
+    expect(chatMutationCatchUp).toHaveBeenCalledWith(db, {
+      workspaceId,
+      actorUserId: userId,
+      channelId,
+      after: "0",
+      through: undefined,
+      limit: 2,
+    });
+    await caller().mutationCatchUp({ ...input, after: "1", through: "3" });
+    expect(chatMutationCatchUp).toHaveBeenLastCalledWith(db, {
+      workspaceId,
+      actorUserId: userId,
+      channelId,
+      after: "1",
+      through: "3",
+      limit: 2,
+    });
+  });
+
+  it("rejects malformed cursors and maps read denial without exposing a channel", async () => {
+    await expect(caller().mutationCatchUp({ ...input, after: "01" })).rejects.toMatchObject({ code: "BAD_REQUEST" });
+    await expect(caller().mutationCatchUp({ ...input, through: "0", after: "1" })).rejects.toMatchObject({ code: "BAD_REQUEST" });
+    expect(chatMutationCatchUp).not.toHaveBeenCalled();
+    vi.mocked(chatMutationCatchUp).mockRejectedValueOnce(new (await import("@tasks/database/chat")).ChatReadError());
+    await expect(caller().mutationCatchUp(input)).rejects.toMatchObject({ code: "NOT_FOUND" });
+  });
+
+  it("requires an authenticated workspace member", async () => {
+    vi.mocked(resolveWorkspace).mockRejectedValueOnce(new Error("not a member"));
+    await expect(caller().mutationCatchUp(input)).rejects.toThrow();
+    expect(chatMutationCatchUp).not.toHaveBeenCalled();
+  });
+});

apps/web/server/routers/chat.ts

@@ -5,17 +5,32 @@ import {
   chatCommandSchema,
   chatCreateChannelInputSchema,
   chatCreateDirectInputSchema,
+  type ChatCommand,
 } from "@tasks/shared";
 import {
   addPrivateMember,
   archiveChannel,
   ChannelPersistenceError,
   ChatPersistenceError,
+  ChatMutationError,
+  ChatReadError,
+  chatHistory,
+  chatThreadHistory,
+  chatCatchUp,
+  chatMutationCatchUp,
+  chatMarkChannelRead,
+  chatUnreadSummary,
+  chatSequenceSchema,
+  chatCursorSchema,
+  chatPageSizeSchema,
+  chatCatchUpSizeSchema,
   createChannel,
   createChannelTransactionRunner,
   createChatTransactionRunner,
   createDirectChannel,
   listVisibleChannels,
+  createMessageMutationRunner,
+  mutateChatMessage,
   removePrivateMember,
   renameChannel,
   sendChatMessage,
@@ -28,6 +43,17 @@ const sendCommandSchema = chatCommandSchema.refine(
   (command) => command.type === "send_message",
   { message: "Expected a send_message command" },
 );
+const mutationCommandSchema = chatCommandSchema.refine(
+  (command) =>
+    command.type === "edit_message" || command.type === "delete_message",
+  { message: "Expected a message mutation" },
+);
+const mutationInputSchema = z
+  .object({
+    workspace: z.string().min(1),
+    command: mutationCommandSchema,
+  })
+  .strict();
 const inputSchema = z
   .object({
     workspace: z.string().min(1),
@@ -101,7 +127,159 @@ async function channelCall<T>(operation: () => Promise<T>): Promise<T> {
   }
 }
 
+const readScope = workspaceInput
+  .extend({ channelId: z.string().uuid() })
+  .strict();
+const catchUpInput = readScope
+  .extend({
+    after: chatCursorSchema,
+    through: chatCursorSchema.optional(),
+    limit: chatCatchUpSizeSchema.default(100),
+  })
+  .strict()
+  .refine(
+    (input) =>
+      input.through === undefined ||
+      BigInt(input.after) <= BigInt(input.through),
+    { path: ["through"], message: "After exceeds through" },
+  );
+const pageInput = readScope
+  .extend({
+    before: chatSequenceSchema.optional(),
+    limit: chatPageSizeSchema.default(50),
+  })
+  .strict();
+async function readCall<T>(operation: () => Promise<T>): Promise<T> {
+  try {
+    return await operation();
+  } catch (error) {
+    if (error instanceof ChatReadError)
+      throw new TRPCError({
+        code: "NOT_FOUND",
+        message: "Chat channel not found",
+      });
+    throw error;
+  }
+}
+
+async function mutationCall(
+  db: Parameters<typeof createMessageMutationRunner>[0],
+  actorUserId: string,
+  workspaceId: string,
+  command: Extract<ChatCommand, { type: "edit_message" | "delete_message" }>,
+) {
+  if (
+    command.workspaceId !== workspaceId ||
+    command.context.workspaceId !== workspaceId
+  )
+    throw new TRPCError({ code: "BAD_REQUEST", message: "Workspace mismatch" });
+  try {
+    return await mutateChatMessage(createMessageMutationRunner(db), {
+      actorUserId,
+      command,
+    });
+  } catch (error) {
+    if (error instanceof ChatMutationError) {
+      if (error.code === "CHAT_FORBIDDEN" || error.code === "CHAT_NOT_FOUND")
+        throw new TRPCError({
+          code: "NOT_FOUND",
+          message: "Chat message not found",
+        });
+      if (error.code === "CHAT_REVISION_CONFLICT")
+        throw new TRPCError({
+          code: "CONFLICT",
+          message: "Message changed. Refresh before trying again.",
+        });
+      throw new TRPCError({
+        code: "CONFLICT",
+        message: "Message request conflicts with an existing request",
+      });
+    }
+    throw new TRPCError({
+      code: "INTERNAL_SERVER_ERROR",
+      message: "Unable to update chat message",
+    });
+  }
+}
+
 export const chatRouter = router({
+  viewer: workspaceProcedure
+    .input(workspaceInput)
+    .query(({ ctx }) => ({ role: ctx.workspace.role })),
+  // History and reconnect endpoints use one workspace-resolved, membership-checked read seam.
+  history: workspaceProcedure.input(pageInput).query(({ ctx, input }) =>
+    readCall(() =>
+      chatHistory(ctx.db, {
+        workspaceId: ctx.workspace.id,
+        actorUserId: ctx.session.user.id,
+        channelId: input.channelId,
+        before: input.before,
+        limit: input.limit,
+      }),
+    ),
+  ),
+  threadHistory: workspaceProcedure
+    .input(pageInput.extend({ rootMessageId: z.string().uuid() }).strict())
+    .query(({ ctx, input }) =>
+      readCall(() =>
+        chatThreadHistory(ctx.db, {
+          workspaceId: ctx.workspace.id,
+          actorUserId: ctx.session.user.id,
+          channelId: input.channelId,
+          rootMessageId: input.rootMessageId,
+          before: input.before,
+          limit: input.limit,
+        }),
+      ),
+    ),
+  catchUp: workspaceProcedure
+    .input(catchUpInput)
+    .query(({ ctx, input }) =>
+      readCall(() =>
+        chatCatchUp(ctx.db, {
+          workspaceId: ctx.workspace.id,
+          actorUserId: ctx.session.user.id,
+          channelId: input.channelId,
+          after: input.after,
+          through: input.through,
+          limit: input.limit,
+        }),
+      ),
+    ),
+  mutationCatchUp: workspaceProcedure
+    .input(catchUpInput)
+    .query(({ ctx, input }) =>
+      readCall(() =>
+        chatMutationCatchUp(ctx.db, {
+          workspaceId: ctx.workspace.id,
+          actorUserId: ctx.session.user.id,
+          channelId: input.channelId,
+          after: input.after,
+          through: input.through,
+          limit: input.limit,
+        }),
+      ),
+    ),
+  markChannelRead: workspaceProcedure
+    .input(readScope.extend({ sequence: chatSequenceSchema }).strict())
+    .mutation(({ ctx, input }) =>
+      readCall(() =>
+        chatMarkChannelRead(ctx.db, {
+          workspaceId: ctx.workspace.id,
+          actorUserId: ctx.session.user.id,
+          channelId: input.channelId,
+          sequence: input.sequence,
+        }),
+      ),
+    ),
+  unreadSummary: workspaceProcedure.input(workspaceInput).query(({ ctx }) =>
+    readCall(() =>
+      chatUnreadSummary(ctx.db, {
+        workspaceId: ctx.workspace.id,
+        actorUserId: ctx.session.user.id,
+      }),
+    ),
+  ),
   listChannels: workspaceProcedure.input(workspaceInput).query(({ ctx }) =>
     channelCall(() =>
       listVisibleChannels(runChannels(ctx.db), {
@@ -186,6 +364,36 @@ export const chatRouter = router({
         }),
       ),
     ),
+  editMessage: workspaceProcedure
+    .input(mutationInputSchema)
+    .mutation(({ ctx, input }) => {
+      if (input.command.type !== "edit_message")
+        throw new TRPCError({
+          code: "BAD_REQUEST",
+          message: "Expected edit_message command",
+        });
+      return mutationCall(
+        ctx.db,
+        ctx.session.user.id,
+        ctx.workspace.id,
+        input.command,
+      );
+    }),
+  deleteMessage: workspaceProcedure
+    .input(mutationInputSchema)
+    .mutation(({ ctx, input }) => {
+      if (input.command.type !== "delete_message")
+        throw new TRPCError({
+          code: "BAD_REQUEST",
+          message: "Expected delete_message command",
+        });
+      return mutationCall(
+        ctx.db,
+        ctx.session.user.id,
+        ctx.workspace.id,
+        input.command,
+      );
+    }),
   send: workspaceProcedure
     .input(inputSchema)
     .mutation(async ({ ctx, input }) => {

apps/worker/src/chat/outbox.test.ts

@@ -0,0 +1,268 @@
+import { PgDialect } from "drizzle-orm/pg-core";
+import { afterEach, describe, expect, it, vi } from "vitest";
+
+import {
+  chatOutboxChannel,
+  createChatOutboxClaim,
+  publishChatOutboxBatch,
+  runChatOutboxPublisher,
+  type ChatOutboxRow,
+  type ClaimBatch,
+} from "./outbox.js";
+
+const workspaceId = "00000000-0000-4000-8000-000000000001";
+const channelId = "00000000-0000-4000-8000-000000000002";
+const row: ChatOutboxRow = {
+  id: "00000000-0000-4000-8000-000000000003",
+  workspaceId,
+  channelId,
+  eventType: "chat.read_state.changed.v1",
+  eventVersion: 1,
+  dedupeKey: "read:1",
+  payload: {
+    eventType: "chat.read_state.changed.v1",
+    eventVersion: 1,
+    workspaceId,
+    channelId,
+    userId: workspaceId,
+    sequence: "1",
+  },
+};
+
+type Mark = (
+  id: string,
+  status: "published" | "failed",
+  error?: string,
+) => Promise<void>;
+function harness(rows: ChatOutboxRow[]) {
+  const changes: Array<{
+    id: string;
+    status: "published" | "failed";
+    error?: string;
+  }> = [];
+  const claim: ClaimBatch = vi.fn(async (run) =>
+    run(
+      rows,
+      async (id: string, status: "published" | "failed", error?: string) => {
+        changes.push({ id, status, error });
+      },
+    ),
+  );
+  const publish = vi.fn(async (_channel: string, _payload: string) => 1);
+  return { claim, publish, changes };
+}
+
+const flush = async () => {
+  await Promise.resolve();
+  await Promise.resolve();
+};
+afterEach(() => vi.useRealTimers());
+
+describe("chat outbox publisher", () => {
+  it("claims a batch and publishes the committed envelope on a tenant/channel scoped topic", async () => {
+    const h = harness([row]);
+    expect(await publishChatOutboxBatch(h.claim, h.publish)).toEqual({
+      published: 1,
+      failed: 0,
+    });
+    expect(h.publish).toHaveBeenCalledWith(
+      chatOutboxChannel(workspaceId, channelId),
+      JSON.stringify({
+        ...row.payload,
+        id: row.id,
+        dedupeKey: row.dedupeKey,
+        workspaceId,
+        channelId,
+        eventType: row.eventType,
+        eventVersion: row.eventVersion,
+      }),
+    );
+    expect(h.changes).toEqual([
+      { id: row.id, status: "published", error: undefined },
+    ]);
+  });
+
+  it("records a failed publish and continues processing other rows", async () => {
+    const second = { ...row, id: "00000000-0000-4000-8000-000000000004" };
+    const h = harness([row, second]);
+    h.publish.mockRejectedValueOnce(new Error("redis down"));
+    expect(await publishChatOutboxBatch(h.claim, h.publish)).toEqual({
+      published: 1,
+      failed: 1,
+    });
+    expect(h.changes).toEqual([
+      { id: row.id, status: "failed", error: "redis down" },
+      { id: second.id, status: "published", error: undefined },
+    ]);
+  });
+
+  it("does not publish when claim fails", async () => {
+    const h = harness([]);
+    vi.mocked(h.claim).mockRejectedValueOnce(new Error("postgres down"));
+    await expect(publishChatOutboxBatch(h.claim, h.publish)).rejects.toThrow(
+      "postgres down",
+    );
+    expect(h.publish).not.toHaveBeenCalled();
+  });
+
+  it("times out a hung publish without marking published even if Redis later acknowledges", async () => {
+    vi.useFakeTimers();
+    let acknowledge!: (count: number) => void;
+    const publish = vi.fn(
+      () =>
+        new Promise<number>((resolve) => {
+          acknowledge = resolve;
+        }),
+    );
+    const h = harness([row]);
+    const batch = publishChatOutboxBatch(h.claim, publish, {
+      publishTimeoutMs: 100,
+    });
+    await flush();
+    await vi.advanceTimersByTimeAsync(100);
+    expect(await batch).toEqual({ published: 0, failed: 1 });
+    acknowledge(1);
+    await flush();
+    expect(h.changes).toEqual([
+      {
+        id: row.id,
+        status: "failed",
+        error: expect.stringContaining("timed out"),
+      },
+    ]);
+  });
+
+  it("stops starting new publishes on abort and leaves the pending row retryable", async () => {
+    const controller = new AbortController();
+    const h = harness([
+      row,
+      { ...row, id: "00000000-0000-4000-8000-000000000004" },
+    ]);
+    const publish = vi.fn(() => new Promise<number>(() => {}));
+    const batch = publishChatOutboxBatch(h.claim, publish, {
+      signal: controller.signal,
+    });
+    await flush();
+    controller.abort();
+    expect(await batch).toEqual({ published: 0, failed: 1 });
+    expect(publish).toHaveBeenCalledTimes(1);
+    expect(h.changes[0]).toMatchObject({ id: row.id, status: "failed" });
+  });
+
+  it("leaves a published notification pending after transaction rollback and redelivers with the same dedupe id", async () => {
+    const pending = new Set([row.id]);
+    let rollback = true;
+    const claim: ClaimBatch = async (run) => {
+      const staged = new Set<string>();
+      await run(pending.has(row.id) ? [row] : [], async (id, status) => {
+        if (status === "published") staged.add(id);
+      });
+      if (rollback) throw new Error("commit failed");
+      for (const id of staged) pending.delete(id);
+    };
+    const publish = vi.fn(async (_channel: string, _payload: string) => 1);
+    await expect(publishChatOutboxBatch(claim, publish)).rejects.toThrow(
+      "commit failed",
+    );
+    rollback = false;
+    expect(await publishChatOutboxBatch(claim, publish)).toEqual({
+      published: 1,
+      failed: 0,
+    });
+    expect(publish).toHaveBeenCalledTimes(2);
+    expect(JSON.parse(publish.mock.calls[0]![1]!)).toMatchObject({
+      id: row.id,
+      dedupeKey: row.dedupeKey,
+    });
+    expect(publish.mock.calls[1]).toEqual(publish.mock.calls[0]);
+  });
+
+  it("builds a skip-locked ordered claim and tenant/channel scoped update without a live DB", async () => {
+    const dialect = new PgDialect();
+    const queries: string[] = [];
+    const tx = {
+      select: vi.fn(() => ({
+        from: () => ({
+          where: () => ({
+            orderBy: (...columns: unknown[]) => {
+              queries.push(`order:${columns.length}`);
+              return {
+                limit: (size: number) => ({
+                  for: (_mode: string, _options: unknown) => {
+                    queries.push(
+                      `limit:${size}:${_mode}:${JSON.stringify(_options)}`,
+                    );
+                    return Promise.resolve([row]);
+                  },
+                }),
+              };
+            },
+          }),
+        }),
+      })),
+      update: vi.fn(() => ({
+        set: () => ({
+          where: (condition: unknown) => {
+            queries.push(
+              dialect.sqlToQuery(
+                condition as Parameters<typeof dialect.sqlToQuery>[0],
+              ).sql,
+            );
+            return Promise.resolve();
+          },
+        }),
+      })),
+    };
+    type FakeTx = typeof tx;
+    const db = {
+      transaction: async (run: (transaction: FakeTx) => Promise<void>) =>
+        run(tx),
+    };
+    await createChatOutboxClaim(db as never)(async (rows, mark) => {
+      expect(rows).toEqual([row]);
+      await mark(row.id, "published");
+    });
+    expect(queries.slice(0, 2)).toEqual([
+      "order:2",
+      'limit:50:update:{"skipLocked":true}',
+    ]);
+    expect(queries[2]).toContain("workspace_id");
+    expect(queries[2]).toContain("channel_id");
+    expect(queries[2]).toContain("published_at");
+    await expect(
+      createChatOutboxClaim(db as never)(async (_rows, mark) => {
+        await mark("unclaimed", "published");
+      }),
+    ).rejects.toThrow("unclaimed");
+  });
+
+  it("paces retries after failures and exits promptly on abort during the delay", async () => {
+    vi.useFakeTimers();
+    const controller = new AbortController();
+    const claim = vi.fn(
+      async (run: (rows: ChatOutboxRow[], mark: Mark) => Promise<void>) => {
+        await run([row], async () => {});
+      },
+    );
+    const redis = {
+      publish: vi.fn(async () => {
+        throw new Error("redis down");
+      }),
+    };
+    const running = runChatOutboxPublisher(
+      {} as Parameters<typeof runChatOutboxPublisher>[0],
+      redis as never,
+      controller.signal,
+      claim,
+    );
+    await vi.advanceTimersByTimeAsync(900);
+    expect(redis.publish).toHaveBeenCalledTimes(1);
+    await vi.advanceTimersByTimeAsync(100);
+    expect(redis.publish).toHaveBeenCalledTimes(2);
+    await vi.advanceTimersByTimeAsync(1900);
+    expect(redis.publish).toHaveBeenCalledTimes(2);
+    controller.abort();
+    await running;
+    expect(redis.publish).toHaveBeenCalledTimes(2);
+  });
+});

apps/worker/src/chat/outbox.ts

@@ -0,0 +1,218 @@
+import { and, asc, eq, isNull, sql } from "drizzle-orm";
+import type IoRedis from "ioredis";
+
+import type { db as dbInstance } from "@tasks/database";
+import { chatOutbox } from "@tasks/database/schema";
+
+/** Redis is a notification transport, not a durable history. Consumers replay from Postgres. */
+export function chatOutboxChannel(
+  workspaceId: string,
+  channelId: string,
+): string {
+  return `chat:v1:workspace:${workspaceId}:channel:${channelId}`;
+}
+
+export type ChatOutboxRow = Pick<
+  typeof chatOutbox.$inferSelect,
+  | "id"
+  | "workspaceId"
+  | "channelId"
+  | "eventType"
+  | "eventVersion"
+  | "dedupeKey"
+  | "payload"
+>;
+type Mark = (
+  id: string,
+  status: "published" | "failed",
+  error?: string,
+) => Promise<void>;
+export type ClaimBatch = (
+  run: (rows: ChatOutboxRow[], mark: Mark) => Promise<void>,
+) => Promise<void>;
+type Publish = (channel: string, payload: string) => Promise<number>;
+
+const PUBLISH_TIMEOUT_MS = 5_000;
+const RETRY_MIN_MS = 1_000;
+const RETRY_MAX_MS = 30_000;
+
+function abortError(): Error {
+  return new Error("Chat outbox publish aborted");
+}
+
+/** Do not disconnect the shared Redis client: it also handles orchestrator jobs. */
+async function boundedPublish(
+  publish: Publish,
+  channel: string,
+  payload: string,
+  timeoutMs: number,
+  signal?: AbortSignal,
+): Promise<void> {
+  if (signal?.aborted) throw abortError();
+  let timer: ReturnType<typeof setTimeout> | undefined;
+  let onAbort: (() => void) | undefined;
+  const deadline = new Promise<never>((_, reject) => {
+    timer = setTimeout(
+      () => reject(new Error("Chat outbox publish timed out")),
+      timeoutMs,
+    );
+    if (signal) {
+      onAbort = () => reject(abortError());
+      signal.addEventListener("abort", onAbort, { once: true });
+      if (signal.aborted) onAbort();
+    }
+  });
+  try {
+    // A late Redis acknowledgement may mean duplicate delivery on retry. Keep the
+    // outbox row pending and let consumers deduplicate by id/dedupeKey.
+    await Promise.race([
+      Promise.resolve().then(() => publish(channel, payload)),
+      deadline,
+    ]);
+  } finally {
+    clearTimeout(timer);
+    if (onAbort) signal?.removeEventListener("abort", onAbort);
+  }
+}
+
+function pause(ms: number, signal: AbortSignal): Promise<void> {
+  if (signal.aborted) return Promise.resolve();
+  return new Promise((resolve) => {
+    const finish = () => {
+      clearTimeout(timer);
+      signal.removeEventListener("abort", finish);
+      resolve();
+    };
+    const timer = setTimeout(finish, ms);
+    signal.addEventListener("abort", finish, { once: true });
+    if (signal.aborted) finish();
+  });
+}
+
+/**
+ * Keep the row locks until Redis acknowledges the publish, so concurrent workers
+ * cannot send the same pending row. A crash between publish and commit can still
+ * redeliver it: consumers must deduplicate by id and replay by channel cursor.
+ */
+export async function publishChatOutboxBatch(
+  claim: ClaimBatch,
+  publish: Publish,
+  options: { publishTimeoutMs?: number; signal?: AbortSignal } = {},
+): Promise<{ published: number; failed: number }> {
+  const result = { published: 0, failed: 0 };
+  await claim(async (rows, mark) => {
+    for (const row of rows) {
+      if (options.signal?.aborted) break;
+      let failure: unknown;
+      try {
+        await boundedPublish(
+          publish,
+          chatOutboxChannel(row.workspaceId, row.channelId),
+          JSON.stringify({
+            ...row.payload,
+            id: row.id,
+            dedupeKey: row.dedupeKey,
+            workspaceId: row.workspaceId,
+            channelId: row.channelId,
+            eventType: row.eventType,
+            eventVersion: row.eventVersion,
+          }),
+          options.publishTimeoutMs ?? PUBLISH_TIMEOUT_MS,
+          options.signal,
+        );
+      } catch (error) {
+        failure = error;
+      }
+      if (failure !== undefined) {
+        const message =
+          failure instanceof Error ? failure.message : String(failure);
+        // A failed status is still pending; it will be retried on the next tick.
+        await mark(row.id, "failed", message.slice(0, 2000));
+        result.failed++;
+      } else {
+        await mark(row.id, "published");
+        result.published++;
+      }
+    }
+  });
+  return result;
+}
+
+export function createChatOutboxClaim(
+  db: typeof dbInstance,
+  batchSize = 50,
+): ClaimBatch {
+  return async (run) => {
+    await db.transaction(async (tx) => {
+      const rows = await tx
+        .select({
+          id: chatOutbox.id,
+          workspaceId: chatOutbox.workspaceId,
+          channelId: chatOutbox.channelId,
+          eventType: chatOutbox.eventType,
+          eventVersion: chatOutbox.eventVersion,
+          dedupeKey: chatOutbox.dedupeKey,
+          payload: chatOutbox.payload,
+        })
+        .from(chatOutbox)
+        .where(isNull(chatOutbox.publishedAt))
+        .orderBy(asc(chatOutbox.createdAt), asc(chatOutbox.id))
+        .limit(batchSize)
+        .for("update", { skipLocked: true });
+
+      await run(rows, async (id, status, error) => {
+        const row = rows.find((candidate) => candidate.id === id);
+        if (!row) throw new Error("Cannot mark an unclaimed chat outbox row");
+        await tx
+          .update(chatOutbox)
+          .set({
+            attempts: sql`${chatOutbox.attempts} + 1`,
+            publishedAt: status === "published" ? new Date() : null,
+            lastError:
+              status === "failed" ? (error ?? "Unknown publish failure") : null,
+          })
+          .where(
+            and(
+              eq(chatOutbox.id, id),
+              eq(chatOutbox.workspaceId, row.workspaceId),
+              eq(chatOutbox.channelId, row.channelId),
+              isNull(chatOutbox.publishedAt),
+            ),
+          );
+      });
+    });
+  };
+}
+
+export async function runChatOutboxPublisher(
+  db: typeof dbInstance,
+  redis: Pick<IoRedis, "publish">,
+  signal: AbortSignal,
+  claim: ClaimBatch = createChatOutboxClaim(db),
+): Promise<void> {
+  let retryMs = RETRY_MIN_MS;
+  while (!signal.aborted) {
+    try {
+      const result = await publishChatOutboxBatch(
+        claim,
+        (channel, payload) => redis.publish(channel, payload),
+        { signal },
+      );
+      if (signal.aborted) break;
+      if (result.failed > 0) {
+        console.error(
+          `[worker] chat outbox: ${result.failed} publish failures`,
+        );
+        await pause(retryMs, signal);
+        retryMs = Math.min(retryMs * 2, RETRY_MAX_MS);
+      } else {
+        retryMs = RETRY_MIN_MS;
+        await pause(result.published === 50 ? 0 : RETRY_MIN_MS, signal);
+      }
+    } catch (error) {
+      console.error("[worker] chat outbox tick failed:", error);
+      await pause(retryMs, signal);
+      retryMs = Math.min(retryMs * 2, RETRY_MAX_MS);
+    }
+  }
+}

apps/worker/src/index.ts

@@ -1,6 +1,7 @@
 import { db } from "@tasks/database";
 import IoRedis from "ioredis";
 
+import { runChatOutboxPublisher } from "./chat/outbox.js";
 import { loadEnv } from "./env.js";
 import { handleOrchestratorJob } from "./handle-job.js";
 import { dequeueJob } from "./queue/dequeue.js";
@@ -21,10 +22,15 @@ async function main(): Promise<void> {
   }
 
   console.info(`[worker] consuming ${env.WORKER_BLPOP_TIMEOUT_S}s BLPOP on orchestrator queue`);
+  const shutdown = new AbortController();
+  process.once("SIGTERM", () => shutdown.abort());
+  process.once("SIGINT", () => shutdown.abort());
+  const publisher = runChatOutboxPublisher(db, redis, shutdown.signal);
 
-  for (;;) {
+  while (!shutdown.signal.aborted) {
     try {
       const job = await dequeueJob(redis, env.WORKER_BLPOP_TIMEOUT_S);
+      if (shutdown.signal.aborted) break;
       if (!job) continue;
       console.info(`[worker] dequeued job=${job.id} role=${job.roleId} convoy=${job.convoyId}`);
       await handleOrchestratorJob(db, redis, env, job);
@@ -32,6 +38,8 @@ async function main(): Promise<void> {
       console.error("[worker] loop error:", error);
     }
   }
+  await publisher;
+  await redis.quit();
 }
 
 main().catch((error) => {

docker/Dockerfile

@@ -132,12 +132,14 @@ RUN rm -f apps/web/.env apps/web/.env.local apps/web/.env.production apps/web/.e
 
 # NEXT_PUBLIC_* must be present at build time — Next.js inlines them into the client bundle.
 ARG NEXT_PUBLIC_COLLAB_SERVER_URL=ws://localhost:1234
+ARG NEXT_PUBLIC_CHAT_SERVER_URL=ws://localhost:1235
 ARG NEXT_PUBLIC_UMAMI_SCRIPT=
 ARG NEXT_PUBLIC_UMAMI_WEBSITE_ID=
 ARG NEXT_PUBLIC_LIBREDESK_URL=
 ARG NEXT_PUBLIC_LIBREDESK_WIDGET_ID=
 ARG NEXT_PUBLIC_DIRECTUS_URL=
 ENV NEXT_PUBLIC_COLLAB_SERVER_URL=${NEXT_PUBLIC_COLLAB_SERVER_URL}
+ENV NEXT_PUBLIC_CHAT_SERVER_URL=${NEXT_PUBLIC_CHAT_SERVER_URL}
 ENV NEXT_PUBLIC_UMAMI_SCRIPT=${NEXT_PUBLIC_UMAMI_SCRIPT}
 ENV NEXT_PUBLIC_UMAMI_WEBSITE_ID=${NEXT_PUBLIC_UMAMI_WEBSITE_ID}
 ENV NEXT_PUBLIC_LIBREDESK_URL=${NEXT_PUBLIC_LIBREDESK_URL}
@@ -176,7 +178,7 @@ CMD ["node", "apps/web/server.js"]
 
 
 # ============================================================================
-# collab — runtime image (Hocuspocus / Yjs websocket)
+# collab — runtime image (Hocuspocus / Yjs and workspace Chat listeners)
 # ============================================================================
 FROM node:${NODE_VERSION}-alpine AS collab
 RUN addgroup --system --gid 1001 nodejs \
@@ -195,10 +197,11 @@ COPY --from=collab-build --chown=collab:nodejs /app/apps/collab-server/node_modu
 
 USER collab
 
-EXPOSE 1234
+EXPOSE 1234 1235
 
+# Listener liveness only; authenticated delivery is a separate acceptance gate.
 HEALTHCHECK --interval=30s --timeout=5s --start-period=20s --retries=3 \
-  CMD node -e "const p=+(process.env.PORT||1234);require('net').createConnection(p,'127.0.0.1').on('connect',()=>process.exit(0)).on('error',()=>process.exit(1))"
+  CMD node -e "let remaining=2;for(const p of [+(process.env.PORT||1234),+(process.env.CHAT_PORT||1235)]){const s=require('net').createConnection(p,'127.0.0.1');s.setTimeout(3000,()=>process.exit(1));s.on('connect',()=>{s.destroy();if(!--remaining)process.exit(0)}).on('error',()=>process.exit(1))}"
 
 WORKDIR /app/apps/collab-server
 CMD ["node", "dist/index.mjs"]

docker/docker-compose.prod.standalone.yml

@@ -12,3 +12,4 @@ services:
   collab:
     ports:
       - "1234:1234"
+      - "${CHAT_PORT:-1235}:${CHAT_PORT:-1235}"

docker/docker-compose.prod.yml

@@ -12,6 +12,7 @@
 # Domains (Dokploy UI or CT 100 Traefik):
 #   web    → echodo.stillwell.cloud        (port 3000)
 #   collab → collab.echodo.stillwell.cloud (port 1234)
+#   chat   → echodo-chat.stillwell.cloud   (collab service, CHAT_PORT default 1235)
 #   opencode (optional) → opencode.stillwell.cloud (port 4096)
 #
 # Build (required on ≤8 GB hosts — parallel export OOMs with exit 255):
@@ -34,6 +35,7 @@ services:
       target: web
       args:
         NEXT_PUBLIC_COLLAB_SERVER_URL: ${NEXT_PUBLIC_COLLAB_SERVER_URL:-ws://localhost:1234}
+        NEXT_PUBLIC_CHAT_SERVER_URL: ${NEXT_PUBLIC_CHAT_SERVER_URL:-ws://localhost:1235}
         NEXT_PUBLIC_UMAMI_SCRIPT: ${NEXT_PUBLIC_UMAMI_SCRIPT:-}
         NEXT_PUBLIC_UMAMI_WEBSITE_ID: ${NEXT_PUBLIC_UMAMI_WEBSITE_ID:-}
         NEXT_PUBLIC_LIBREDESK_URL: ${NEXT_PUBLIC_LIBREDESK_URL:-}
@@ -48,6 +50,7 @@ services:
     environment:
       - DATABASE_URL=${DATABASE_URL}
       - REDIS_URL=${REDIS_URL}
+      - CHAT_TICKET_SECRET=${CHAT_TICKET_SECRET:?Set a shared Chat signing secret of at least 32 characters}
       - NEXTAUTH_URL=${NEXTAUTH_URL}
       - NEXTAUTH_SECRET=${NEXTAUTH_SECRET}
       - AUTH_SECRET=${AUTH_SECRET}
@@ -100,13 +103,16 @@ services:
       - DATABASE_URL=${DATABASE_URL}
       - REDIS_URL=${REDIS_URL}
       - PORT=1234
+      - CHAT_PORT=${CHAT_PORT:-1235}
+      - CHAT_TICKET_SECRET=${CHAT_TICKET_SECRET:?Set a shared Chat signing secret of at least 32 characters}
       - NODE_ENV=production
     expose:
       - "1234"
+      - "${CHAT_PORT:-1235}"
     healthcheck:
       test:
         - CMD-SHELL
-        - 'node -e "const p=+(process.env.PORT||1234);require(''net'').createConnection(p,''127.0.0.1'').on(''connect'',()=>process.exit(0)).on(''error'',()=>process.exit(1))"'
+        - 'node -e "let remaining=2;for(const p of [+(process.env.PORT||1234),+(process.env.CHAT_PORT||1235)]){const s=require(''net'').createConnection(p,''127.0.0.1'');s.setTimeout(3000,()=>process.exit(1));s.on(''connect'',()=>{s.destroy();if(!--remaining)process.exit(0)}).on(''error'',()=>process.exit(1))}"'
       interval: 30s
       timeout: 5s
       retries: 3

docs/Current Progress.md

@@ -0,0 +1,71 @@
+# Workspace Chat — Current Progress
+
+Updated: 2026-10-30 (dedicated Chat hostname migration pending rebuild).
+
+## Hostname correction — pending deployment
+
+`chat.stillwell.cloud` belongs to OpenMausBot on CT 117, not Workspace Chat. The operator reports routing to that app has been restored and `echodo-chat.stillwell.cloud` is already an alias on the Workspace Chat Traefik rule.
+
+The Workspace Chat endpoint is now **`wss://echodo-chat.stillwell.cloud`** (socket path `/chat`). Set `NEXT_PUBLIC_CHAT_SERVER_URL` to that value in CT 116's `/opt/submatter/.env`, rebuild the web image, and recreate web as documented in `docs/deployment-production.md`. Repo examples now use only the dedicated hostname. Deployment and browser verification of the new endpoint are not yet confirmed; a push alone will not update the host's `.env`.
+
+## Previous go-live evidence (before hostname correction)
+
+The following records the previous deployment, not verification of the new hostname. Operator reported **14/14 acceptance checks passed**. Operator-confirmed CT 102 `echodo` ledger: **24 entries through `0023`**, with required columns/table verified and all `chat_*` tables owned by `echodo`. An earlier copy of this document said `0022–0024`; no `0024` migration exists in this checkout, so that statement is not treated as verified.
+
+The deployment evidence below was supplied by the operator and is present in this workspace copy. It was not independently rerun in this reconciliation session; read-only access through PVE stopped at unknown SSH host-key verification.
+
+## Deployed
+
+- `CHAT_TICKET_SECRET` generated on CT 116 (≥32), shared by web + collab (sha256 match). Not logged.
+- `CHAT_PORT=1235`; collab dual listeners 1234 (Hocuspocus) + 1235 (Chat).
+- Traefik `echodo-chat` → `192.168.68.116:1235` for `chat.echodo.stillwell.cloud` and `chat.stillwell.cloud`.
+- Public WSS URL baked into web: `wss://chat.stillwell.cloud` (nested `*.echodo.stillwell.cloud` Universal SSL fails publicly).
+- Compose validated; web/collab/worker rebuilt/recreated; opencode-v2 preserved.
+
+## Runtime fixes applied during go-live
+
+1. Ticket Origin check used `request.url` origin (`https://0.0.0.0:3000`) behind Docker — now prefers `x-forwarded-host` / `host` + `x-forwarded-proto`.
+2. `channel-drizzle-adapter` `row()` spread full DB columns into strict lifecycle receipt schema — project ChannelRecord fields only.
+3. `redactCreatedOutbox` `jsonb_build_object(...)` params caused `could not determine data type of parameter $1` — set plain JSON payload instead.
+4. Empty `AUTH_DEV_PASSWORD` blocked credentials acceptance — set on CT 116 (not logged).
+
+Source under `/opt/submatter` (+ opencode workspace copy) updated for (1–3). Image chunks patched in place for immediate runtime; rebuild web on next deploy to bake permanently.
+
+## Source reconciliation (2026-10-29)
+
+- Channel projection and plain-JSON redaction fixes were already present locally; preserved and covered with adapter regression tests.
+- Local ticket handling now pins the expected origin to configured `AUTH_URL` (or `NEXTAUTH_URL`), not caller-supplied forwarded headers. Production fails closed without a valid configured URL. This is a hardened replacement for the deployed forwarded-header hot patch, not a byte-for-byte copy. Deploy and retest ticket issuance before considering this replacement live.
+- Canonical browser endpoint is `wss://chat.stillwell.cloud` (socket path `/chat`). The Traefik example retains both hostname routes.
+- Next deploy must rebuild web from reconciled source; image hot patches are not durable. Re-run ticket issuance, cross-origin rejection and the acceptance suite afterward. No production changes, rebuild, commit or push were performed during this reconciliation.
+- **Security follow-up:** `AUTH_DEV_PASSWORD` is a shared credentials password for existing user emails, not a per-user password. Remove it after acceptance unless the operator explicitly intends this login mechanism; verify SSO access first and recreate web when removing the environment value. Awaiting operator decision; not changed remotely.
+- Remaining release gates: full local checks, disposable-Postgres coverage, final review and convoy lifecycle reconciliation. The reported live suite does not prove multi-instance fan-out or Redis outage behavior.
+
+## Main deployment preparation
+
+The operator approved sending the reconciled branch to `main` for real-build testing. The branch includes the earlier foundation commit and its AI-shell/backlog changes. The main push triggers `.forgejo/workflows/deploy-homelab.yml` when the runner and deployment secrets are configured; a push alone does not prove deployment success. The deploy script no longer removes orphan containers, preserving separately managed services such as `opencode-v2`. Six static deployment checks cover the configuration and this safety guard.
+
+## Local validation after reconciliation
+
+- `pnpm lint`: passed with existing warnings.
+- `pnpm type-check --concurrency=1`: passed.
+- `pnpm test`: ticket-origin regressions (11), channel adapter tests (10) and outbox adapter test (1) passed. Full suite remains blocked by the worker remote-AG-UI pending-response polling test timing out after 5000 ms. PostgreSQL-gated tests remain skipped without a disposable test database.
+- `pnpm build --concurrency=1`: passed with a longer command allowance after the first attempt timed out during build tracing. This is a local build, not a Docker image rebuild or deployment.
+- `git diff --check`: passed.
+- Security review of this slice: ticket origin is pinned to server configuration; forwarded-header spoofing, missing/malformed origins and workspace denial are covered. Redaction remains workspace/channel/message/create-event scoped. Production shared-password sign-in remains an operational security concern until explicitly resolved. No secrets were read or printed.
+
+## Acceptance evidence (2026-10-29)
+
+Two signed-in members (`me@randallstillwell.com` + `chat-acceptance-b@stillwell.local`) against workspace `randall-stillwell`:
+
+| Gate | Result |
+|------|--------|
+| Secret shared web/collab | Pass |
+| Compose validate + ports | Pass |
+| Public WSS reject without ticket | Pass (connection fails) |
+| Dual WS subscribe via `wss://chat.stillwell.cloud/chat` | Pass |
+| Send + Redis invalidate hint | Pass |
+| Reply | Pass |
+| Reconnect catch-up | Pass |
+| Unread mark + summary | Pass |
+| Edit / delete + mutation catch-up | Pass (after redact fix) |
+| Membership revocation (hint + history denied) | Pass |

docs/deployment-production.md

@@ -34,7 +34,7 @@ External: Postgres/Redis on **CT 102**, TLS on **CT 100**, OmniRoute on **CT 108
 | File | Use |
 |------|-----|
 | `docker/docker-compose.prod.yml` | Core stack |
-| `docker/docker-compose.prod.standalone.yml` | Publish `:3000` / `:1234` for CT 100 Traefik |
+| `docker/docker-compose.prod.standalone.yml` | Publish `:3000` / `:1234` / `:1235` for CT 100 Traefik |
 | `scripts/build-prod.sh` | Serialized image build |
 | `scripts/deploy-prod.sh` | Build + up + health wait |
 
@@ -65,6 +65,9 @@ docker compose -f docker/docker-compose.prod.yml \
 | `AUTH_AUTHENTIK_*` | | Authentik OIDC app `echodo` |
 | `COLLAB_SERVER_URL` | `wss://collab.echodo.stillwell.cloud` | Server-side |
 | `NEXT_PUBLIC_COLLAB_SERVER_URL` | `wss://collab.echodo.stillwell.cloud` | **Build arg** — requires image rebuild |
+| `NEXT_PUBLIC_CHAT_SERVER_URL` | `wss://echodo-chat.stillwell.cloud` | **Build arg** — requires web image rebuild; provision DNS/TLS first |
+| `CHAT_TICKET_SECRET` | Generate with `openssl rand -base64 32` | Same runtime secret for web and collab; at least 32 characters; never a build arg |
+| `CHAT_PORT` | `1235` | Second listener inside collab; cannot equal Hocuspocus port `1234` |
 | `OMNIROUTE_BASE_URL` | `https://omniroute.stillwell.cloud/v1` | Web + worker |
 | `OMNIROUTE_API_KEY` | (your key) | + register in Settings → AI → Providers |
 | `FORGEJO_TOKEN` / `GITHUB_TOKEN` | PAT values | Referenced by name in Connected repos settings |
@@ -74,6 +77,42 @@ Changing any `NEXT_PUBLIC_*` value requires `bash scripts/deploy-prod.sh build`,
 
 ---
 
+## Workspace Chat integration gate
+
+Chat runs inside the existing `collab` service on a **separate listener** from Hocuspocus. The worker publishes metadata-only Redis hints; Postgres history and mutation catch-up remain authoritative. Team Chat does not invoke AI agents, and object contexts are not enabled by this deployment.
+
+1. Set the Chat variables above in `/opt/submatter/.env`. Keep the signing secret server-side; Compose requires a nonempty value, and the application requires at least 32 characters. A missing/short secret disables the listener and makes the dual-listener health check fail.
+2. Provision `echodo-chat.stillwell.cloud` DNS and TLS on CT 100 (canonical browser endpoint). `chat.stillwell.cloud` belongs to another app: never include it in the Workspace Chat router. Apply the `echodo-chat` router/service from [`traefik-ct100.example.yml`](../proxmox/ct116/traefik-ct100.example.yml). Traefik supports WebSocket upgrades without a separate service process. If `CHAT_PORT` changes, update the proxy backend port too. Restrict host backend ports to the trusted proxy network.
+
+   For the hostname migration, set `NEXT_PUBLIC_CHAT_SERVER_URL=wss://echodo-chat.stillwell.cloud` in CT 116's `/opt/submatter/.env`, then run from `/opt/submatter`:
+
+   ```bash
+   docker compose --project-directory /opt/submatter -f docker/docker-compose.prod.yml -f docker/docker-compose.prod.standalone.yml build web
+   docker compose --project-directory /opt/submatter -f docker/docker-compose.prod.yml -f docker/docker-compose.prod.standalone.yml up -d --no-deps web
+   ```
+
+   A git push or container restart alone does not change the baked browser URL. Test Workspace Chat inside `https://echodo.stillwell.cloud`; its WebSocket should connect to `wss://echodo-chat.stillwell.cloud/chat`. The WSS endpoint is not a separate chat UI.
+3. Before rolling out, verify the database target is CT 102 **`echodo`** and inspect `drizzle.__drizzle_migrations`. The operator reports 24 ledger entries through `0023`, verified Chat schema, and ownership of all `chat_*` tables transferred to `echodo`. Do not reapply these migrations; inspect the ledger for any future pending entries. Compare the ledger with `packages/database/migrations/meta/_journal.json`; apply only pending journaled migrations using `pnpm --filter @tasks/database db:migrate` in a checkout with the real `DATABASE_URL` exported securely. Do not print the URL, use `db:push`, or replay historical migrations manually. Take a backup first.
+4. Verify `0022` added `chat_messages.send_intent_fingerprint` and positive `revision`; verify `0023` added `chat_message_mutations`, its tenant/channel/message foreign keys and indexes, and `chat_channels.next_mutation_sequence`. Do not infer applied schema from TypeScript files. The historical `0017` correction is already local; do not replay it on an initialized database.
+5. Validate Compose on the deployment host without printing resolved secrets:
+
+   ```bash
+   docker compose --project-directory /opt/submatter --env-file /opt/submatter/.env \
+     -f docker/docker-compose.prod.yml -f docker/docker-compose.prod.standalone.yml config --quiet
+   bash scripts/deploy-prod.sh build
+   ```
+
+6. Confirm both collab listeners and worker health. The TCP health check verifies listener liveness only, **not** Redis delivery, database authorization, or end-to-end readiness. An ordinary HTTP request to the Chat listener returns 404; use an authenticated browser WebSocket session for acceptance.
+7. Test two signed-in workspace members: send/reply/edit/delete, unread acknowledgement, reconnect catch-up, offline edits/deletes, and duplicate retries. Revoke private/direct membership while connected and verify subsequent hints/history are denied and cached content is cleared; admins must not bypass membership. Verify another workspace cannot obtain history or subscribe. Restart collab/worker and test Redis outage recovery without message loss.
+
+Run PostgreSQL integration tests only against a disposable test database, not production. DNS, TLS, migrations, Redis/WebSocket delivery, and two-browser acceptance are independent release gates; passing unit tests does not satisfy them. See [Current Progress](Current%20Progress.md) for the current evidence and blockers.
+
+### Post-go-live reconciliation
+
+The operator reports Chat live with 14/14 acceptance checks; see [Current Progress](Current%20Progress.md) for provenance and scope. Rebuild web on the next deploy to replace hot-patched image chunks with source-built code. The reconciled ticket route uses `AUTH_URL` or `NEXTAUTH_URL` as its trusted public origin; set it to the HTTPS **web app** origin, not the WSS endpoint. It ignores forwarded headers for this decision and fails closed without valid production configuration. Restrict direct backend access to the proxy network regardless.
+
+`AUTH_DEV_PASSWORD` was enabled for acceptance. This shared password can authenticate existing email identities; it is not individual-user password authentication. Confirm SSO works, then remove it and recreate web if it was temporary. Do not leave this enabled solely because acceptance needed it. Re-run ticket origin/authentication and live Chat acceptance after the source-built deployment.
+
 ## Auto-deploy on push
 
 **CI** (every PR): `.github/workflows/ci.yml` — lint, type-check, test on GitHub-hosted or PR runner.

package.json

@@ -6,7 +6,7 @@
     "dev": "turbo dev",
     "lint": "turbo lint",
     "type-check": "turbo type-check",
-    "test": "turbo test",
+    "test": "node --test scripts/chat-deployment.test.mjs && turbo test",
     "db:generate": "turbo db:generate --filter=@tasks/database",
     "db:migrate": "turbo db:migrate --filter=@tasks/database",
     "db:push": "turbo db:push --filter=@tasks/database",

packages/database/migrations/0017_lively_squadron_supreme.sql

@@ -1,16 +1,3 @@
-CREATE TABLE "convoy_operator_messages" (
-	"id" uuid PRIMARY KEY DEFAULT gen_random_uuid() NOT NULL,
-	"workspace_id" uuid NOT NULL,
-	"convoy_id" uuid NOT NULL,
-	"brief_id" uuid,
-	"agent_run_id" uuid,
-	"delivery" varchar(20) NOT NULL,
-	"body" text NOT NULL,
-	"author_user_id" uuid NOT NULL,
-	"delivered_at" timestamp with time zone,
-	"created_at" timestamp with time zone DEFAULT now() NOT NULL
-);
---> statement-breakpoint
 CREATE TABLE "completion_artifacts" (
 	"id" uuid PRIMARY KEY DEFAULT gen_random_uuid() NOT NULL,
 	"workspace_id" uuid NOT NULL,
@@ -21,15 +8,9 @@ CREATE TABLE "completion_artifacts" (
 	"created_at" timestamp with time zone DEFAULT now() NOT NULL
 );
 --> statement-breakpoint
-ALTER TABLE "convoy_operator_messages" ADD CONSTRAINT "convoy_operator_messages_workspace_id_workspaces_id_fk" FOREIGN KEY ("workspace_id") REFERENCES "public"."workspaces"("id") ON DELETE cascade ON UPDATE no action;--> statement-breakpoint
-ALTER TABLE "convoy_operator_messages" ADD CONSTRAINT "convoy_operator_messages_convoy_id_objects_id_fk" FOREIGN KEY ("convoy_id") REFERENCES "public"."objects"("id") ON DELETE cascade ON UPDATE no action;--> statement-breakpoint
-ALTER TABLE "convoy_operator_messages" ADD CONSTRAINT "convoy_operator_messages_brief_id_objects_id_fk" FOREIGN KEY ("brief_id") REFERENCES "public"."objects"("id") ON DELETE set null ON UPDATE no action;--> statement-breakpoint
-ALTER TABLE "convoy_operator_messages" ADD CONSTRAINT "convoy_operator_messages_agent_run_id_agent_runs_id_fk" FOREIGN KEY ("agent_run_id") REFERENCES "public"."agent_runs"("id") ON DELETE set null ON UPDATE no action;--> statement-breakpoint
-ALTER TABLE "convoy_operator_messages" ADD CONSTRAINT "convoy_operator_messages_author_user_id_users_id_fk" FOREIGN KEY ("author_user_id") REFERENCES "public"."users"("id") ON DELETE cascade ON UPDATE no action;--> statement-breakpoint
 ALTER TABLE "completion_artifacts" ADD CONSTRAINT "completion_artifacts_workspace_id_workspaces_id_fk" FOREIGN KEY ("workspace_id") REFERENCES "public"."workspaces"("id") ON DELETE cascade ON UPDATE no action;--> statement-breakpoint
 ALTER TABLE "completion_artifacts" ADD CONSTRAINT "completion_artifacts_convoy_id_objects_id_fk" FOREIGN KEY ("convoy_id") REFERENCES "public"."objects"("id") ON DELETE cascade ON UPDATE no action;--> statement-breakpoint
 ALTER TABLE "completion_artifacts" ADD CONSTRAINT "completion_artifacts_created_by_users_id_fk" FOREIGN KEY ("created_by") REFERENCES "public"."users"("id") ON DELETE set null ON UPDATE no action;--> statement-breakpoint
-CREATE INDEX "convoy_operator_messages_workspace_id_convoy_id_delivered_at_idx" ON "convoy_operator_messages" USING btree ("workspace_id","convoy_id","delivered_at");--> statement-breakpoint
 CREATE INDEX "completion_artifacts_workspace_id_idx" ON "completion_artifacts" USING btree ("workspace_id");--> statement-breakpoint
 CREATE INDEX "completion_artifacts_convoy_id_idx" ON "completion_artifacts" USING btree ("convoy_id");--> statement-breakpoint
 CREATE INDEX "completion_artifacts_created_at_idx" ON "completion_artifacts" USING btree ("created_at");

packages/database/migrations/0022_skinny_masked_marvel.sql

@@ -0,0 +1,3 @@
+ALTER TABLE "chat_messages" ADD COLUMN "send_intent_fingerprint" text;--> statement-breakpoint
+ALTER TABLE "chat_messages" ADD COLUMN "revision" integer DEFAULT 1 NOT NULL;--> statement-breakpoint
+ALTER TABLE "chat_messages" ADD CONSTRAINT "chat_messages_revision_check" CHECK ("chat_messages"."revision" > 0);

packages/database/migrations/0023_thick_shinko_yamashiro.sql

@@ -0,0 +1,18 @@
+CREATE TABLE "chat_message_mutations" (
+	"id" uuid PRIMARY KEY DEFAULT gen_random_uuid() NOT NULL,
+	"workspace_id" uuid NOT NULL,
+	"channel_id" uuid NOT NULL,
+	"message_id" uuid NOT NULL,
+	"mutation_sequence" bigint NOT NULL,
+	"revision" integer NOT NULL,
+	"kind" varchar(16) NOT NULL,
+	"created_at" timestamp with time zone DEFAULT now() NOT NULL,
+	CONSTRAINT "chat_message_mutations_channel_sequence_unique" UNIQUE("workspace_id","channel_id","mutation_sequence"),
+	CONSTRAINT "chat_message_mutations_revision_check" CHECK ("chat_message_mutations"."revision" > 1)
+);
+--> statement-breakpoint
+ALTER TABLE "chat_channels" ADD COLUMN "next_mutation_sequence" bigint DEFAULT 1 NOT NULL;--> statement-breakpoint
+ALTER TABLE "chat_message_mutations" ADD CONSTRAINT "chat_message_mutations_workspace_id_workspaces_id_fk" FOREIGN KEY ("workspace_id") REFERENCES "public"."workspaces"("id") ON DELETE cascade ON UPDATE no action;--> statement-breakpoint
+ALTER TABLE "chat_message_mutations" ADD CONSTRAINT "chat_message_mutations_channel_fk" FOREIGN KEY ("workspace_id","channel_id") REFERENCES "public"."chat_channels"("workspace_id","id") ON DELETE cascade ON UPDATE no action;--> statement-breakpoint
+ALTER TABLE "chat_message_mutations" ADD CONSTRAINT "chat_message_mutations_message_fk" FOREIGN KEY ("workspace_id","message_id") REFERENCES "public"."chat_messages"("workspace_id","id") ON DELETE cascade ON UPDATE no action;--> statement-breakpoint
+CREATE INDEX "chat_message_mutations_workspace_id_idx" ON "chat_message_mutations" USING btree ("workspace_id");

packages/database/migrations/meta/_journal.json

@@ -155,6 +155,20 @@
       "when": 1792766022304,
       "tag": "0021_chat_direct_membership_immutability",
       "breakpoints": true
+    },
+    {
+      "idx": 22,
+      "version": "7",
+      "when": 1793053728379,
+      "tag": "0022_skinny_masked_marvel",
+      "breakpoints": true
+    },
+    {
+      "idx": 23,
+      "version": "7",
+      "when": 1793053797739,
+      "tag": "0023_thick_shinko_yamashiro",
+      "breakpoints": true
     }
   ]
 }

packages/database/src/chat/channel-drizzle-adapter.test.ts

@@ -0,0 +1,206 @@
+import { describe, expect, it, vi } from "vitest";
+import { createChannelTransactionRunner } from "./channel-drizzle-adapter";
+
+type ChannelDatabase = NonNullable<
+  Parameters<typeof createChannelTransactionRunner>[0]
+>;
+type DatabaseTransaction = Parameters<
+  Parameters<ChannelDatabase["transaction"]>[0]
+>[0];
+type TransactionCallback = (tx: DatabaseTransaction) => Promise<unknown>;
+
+function fakeDatabase(
+  transaction: (callback: TransactionCallback) => Promise<unknown>,
+): ChannelDatabase {
+  return { transaction } as unknown as ChannelDatabase;
+}
+
+const fakeTx = {} as DatabaseTransaction;
+
+function pgError(code: string): Error & { code: string } {
+  return Object.assign(new Error(`PostgreSQL ${code}`), { code });
+}
+
+describe("createChannelTransactionRunner", () => {
+  it("projects only ChannelRecord fields from database rows for strict receipts", async () => {
+    const expected = {
+      id: "10000000-0000-4000-8000-000000000001",
+      workspaceId: "20000000-0000-4000-8000-000000000001",
+      kind: "public",
+      name: "general",
+      ownerUserId: "30000000-0000-4000-8000-000000000001",
+      directKey: null,
+      archivedAt: null,
+    };
+    const records = [
+      {
+        ...expected,
+        nextSequence: 42n,
+        nextMutationSequence: 9n,
+        createdAt: new Date(),
+        updatedAt: new Date(),
+      },
+    ];
+    const tx = {
+      select: () => ({
+        from: () => ({
+          where: () => ({
+            for: async () => records,
+          }),
+        }),
+      }),
+    } as unknown as DatabaseTransaction;
+    const result = await createChannelTransactionRunner(
+      fakeDatabase(async (run) => run(tx)),
+    )((adapter) => adapter.lockChannel(expected.workspaceId, expected.id));
+    expect(result).toEqual(expected);
+    expect(() => JSON.stringify(result)).not.toThrow();
+  });
+  it.each(["40001", "40P01"])(
+    "replays the callback in a fresh transaction after %s",
+    async (code) => {
+      const transaction = vi.fn(async (callback: TransactionCallback) =>
+        callback(fakeTx),
+      );
+      const run = vi.fn(async () => {
+        if (run.mock.calls.length === 1) throw pgError(code);
+        return "committed";
+      });
+
+      await expect(
+        createChannelTransactionRunner(fakeDatabase(transaction))(run),
+      ).resolves.toBe("committed");
+      expect(transaction).toHaveBeenCalledTimes(2);
+      expect(run).toHaveBeenCalledTimes(2);
+    },
+  );
+
+  it("replays the callback when a wrapped deadlock rejects after the callback returns", async () => {
+    const transaction = vi.fn(async (callback: TransactionCallback) => {
+      const result = await callback(fakeTx);
+      if (transaction.mock.calls.length === 1) {
+        throw new Error("transaction failed", { cause: pgError("40P01") });
+      }
+      return result;
+    });
+    const run = vi.fn(async () => "committed");
+
+    await expect(
+      createChannelTransactionRunner(fakeDatabase(transaction))(run),
+    ).resolves.toBe("committed");
+    expect(transaction).toHaveBeenCalledTimes(2);
+    expect(run).toHaveBeenCalledTimes(2);
+  });
+
+  it("rethrows the original wrapped serialization error after three attempts", async () => {
+    const original = new Error("transaction failed", {
+      cause: new Error("driver failed", { cause: pgError("40001") }),
+    });
+    const transaction = vi.fn(async (callback: TransactionCallback) => {
+      await callback(fakeTx);
+      throw original;
+    });
+    const run = vi.fn(async () => "not committed");
+
+    await expect(
+      createChannelTransactionRunner(fakeDatabase(transaction))(run),
+    ).rejects.toBe(original);
+    expect(transaction).toHaveBeenCalledTimes(3);
+    expect(run).toHaveBeenCalledTimes(3);
+  });
+
+  it("does not retry unrelated failures", async () => {
+    const original = new Error("invalid input", { cause: pgError("23505") });
+    const transaction = vi.fn(async (callback: TransactionCallback) =>
+      callback(fakeTx),
+    );
+    const run = vi.fn(async () => {
+      throw original;
+    });
+
+    await expect(
+      createChannelTransactionRunner(fakeDatabase(transaction))(run),
+    ).rejects.toBe(original);
+    expect(transaction).toHaveBeenCalledTimes(1);
+    expect(run).toHaveBeenCalledTimes(1);
+  });
+});
+
+describe("channel membership persistence", () => {
+  it("does not mark a no-op insert as changed", async () => {
+    const returning = vi.fn().mockResolvedValue([]);
+    const tx = {
+      insert: vi.fn().mockReturnValue({
+        values: vi.fn().mockReturnValue({
+          onConflictDoNothing: vi.fn().mockReturnValue({ returning }),
+        }),
+      }),
+    };
+    const database = {
+      transaction: async (fn: (transaction: typeof tx) => Promise<boolean>) =>
+        fn(tx),
+    };
+    await expect(
+      createChannelTransactionRunner(
+        database as unknown as Parameters<
+          typeof createChannelTransactionRunner
+        >[0],
+      )((adapter) => adapter.addMember("workspace", "channel", "member")),
+    ).resolves.toBe(false);
+  });
+
+  it("does not mark a no-op delete as changed", async () => {
+    const returning = vi.fn().mockResolvedValue([]);
+    const tx = {
+      delete: vi
+        .fn()
+        .mockReturnValue({ where: vi.fn().mockReturnValue({ returning }) }),
+    };
+    const database = {
+      transaction: async (fn: (transaction: typeof tx) => Promise<boolean>) =>
+        fn(tx),
+    };
+    await expect(
+      createChannelTransactionRunner(
+        database as unknown as Parameters<
+          typeof createChannelTransactionRunner
+        >[0],
+      )((adapter) => adapter.removeMember("workspace", "channel", "member")),
+    ).resolves.toBe(false);
+  });
+
+  it("reports inserted members using returned rows rather than driver command metadata", async () => {
+    const returning = vi.fn().mockResolvedValue([{ userId: "member" }]);
+    const onConflictDoNothing = vi.fn().mockReturnValue({ returning });
+    const values = vi.fn().mockReturnValue({ onConflictDoNothing });
+    const tx = { insert: vi.fn().mockReturnValue({ values }) };
+    const database = {
+      transaction: async (fn: (transaction: typeof tx) => Promise<boolean>) =>
+        fn(tx),
+    };
+    const changed = await createChannelTransactionRunner(
+      database as unknown as Parameters<
+        typeof createChannelTransactionRunner
+      >[0],
+    )((adapter) => adapter.addMember("workspace", "channel", "member"));
+    expect(changed).toBe(true);
+    expect(returning).toHaveBeenCalled();
+  });
+
+  it("reports deleted members using returned rows rather than driver command metadata", async () => {
+    const returning = vi.fn().mockResolvedValue([{ userId: "member" }]);
+    const where = vi.fn().mockReturnValue({ returning });
+    const tx = { delete: vi.fn().mockReturnValue({ where }) };
+    const database = {
+      transaction: async (fn: (transaction: typeof tx) => Promise<boolean>) =>
+        fn(tx),
+    };
+    const changed = await createChannelTransactionRunner(
+      database as unknown as Parameters<
+        typeof createChannelTransactionRunner
+      >[0],
+    )((adapter) => adapter.removeMember("workspace", "channel", "member"));
+    expect(changed).toBe(true);
+    expect(returning).toHaveBeenCalled();
+  });
+});

packages/database/src/chat/channel-drizzle-adapter.ts

@@ -14,9 +14,15 @@ import {
   type ChannelTransaction,
 } from "./manage-channels";
 type Tx = Parameters<Parameters<typeof defaultDb.transaction>[0]>[0];
-const row = (c: typeof chatChannels.$inferSelect) => ({ ...c });
+const row = (c: typeof chatChannels.$inferSelect) => ({
-const changed = (result: unknown) =>
+  id: c.id,
-  Number((result as { rowCount?: number }).rowCount ?? 0) > 0;
+  workspaceId: c.workspaceId,
+  kind: c.kind,
+  name: c.name,
+  ownerUserId: c.ownerUserId,
+  directKey: c.directKey,
+  archivedAt: c.archivedAt,
+});
 function translate(error: unknown): never {
   const record =
     typeof error === "object" && error
@@ -89,12 +95,16 @@ function adapter(tx: Tx): ChannelTransaction {
       ).map((x) => row(x.channel));
     },
     async lockChannel(workspaceId, channelId) {
-      const result = await tx.execute(
+      const [c] = await tx
-        sql`select * from ${chatChannels} where ${chatChannels.workspaceId}=${workspaceId} and ${chatChannels.id}=${channelId} for update`,
+        .select()
-      );
+        .from(chatChannels)
-      const c = (
+        .where(
-        result as unknown as Array<typeof chatChannels.$inferSelect>
+          and(
-      )[0];
+            eq(chatChannels.workspaceId, workspaceId),
+            eq(chatChannels.id, channelId),
+          ),
+        )
+        .for("update");
       return c ? row(c) : null;
     },
     async isChannelMember(workspaceId, channelId, userId) {
@@ -162,19 +172,18 @@ function adapter(tx: Tx): ChannelTransaction {
     },
     async addMember(workspaceId, channelId, userId) {
       try {
-        return changed(
+        const inserted = await tx
-          await tx
           .insert(chatChannelMembers)
           .values({ workspaceId, channelId, userId })
-            .onConflictDoNothing(),
+          .onConflictDoNothing()
-        );
+          .returning({ userId: chatChannelMembers.userId });
+        return inserted.length > 0;
       } catch (e) {
         translate(e);
       }
     },
     async removeMember(workspaceId, channelId, userId) {
-      return changed(
+      const deleted = await tx
-        await tx
         .delete(chatChannelMembers)
         .where(
           and(
@@ -182,8 +191,9 @@ function adapter(tx: Tx): ChannelTransaction {
             eq(chatChannelMembers.channelId, channelId),
             eq(chatChannelMembers.userId, userId),
           ),
-          ),
+        )
-      );
+        .returning({ userId: chatChannelMembers.userId });
+      return deleted.length > 0;
     },
     async lockDirect(workspaceId, directKey) {
       await tx.execute(
@@ -279,9 +289,7 @@ function adapter(tx: Tx): ChannelTransaction {
     },
     async saveReceipt(scope, operation, fingerprint, result) {
       try {
-        await tx
+        await tx.insert(chatMutationReceipts).values({
-          .insert(chatMutationReceipts)
-          .values({
           ...scope,
           operation,
           fingerprint,
@@ -305,8 +313,36 @@ function adapter(tx: Tx): ChannelTransaction {
     },
   };
 }
+function isRetryableTransactionError(error: unknown): boolean {
+  const seen = new Set<unknown>();
+  let current = error;
+  while (
+    current !== null &&
+    typeof current === "object" &&
+    !seen.has(current)
+  ) {
+    seen.add(current);
+    if (
+      "code" in current &&
+      (current.code === "40001" || current.code === "40P01")
+    ) {
+      return true;
+    }
+    current = "cause" in current ? current.cause : undefined;
+  }
+  return false;
+}
+
 export function createChannelTransactionRunner(
   database: typeof defaultDb = defaultDb,
 ): ChannelRunner {
-  return (fn) => database.transaction((tx) => fn(adapter(tx)));
+  return async (fn) => {
+    for (let attempt = 1; ; attempt++) {
+      try {
+        return await database.transaction((tx) => fn(adapter(tx)));
+      } catch (error) {
+        if (attempt >= 3 || !isRetryableTransactionError(error)) throw error;
+      }
+    }
+  };
 }

packages/database/src/chat/drizzle-adapter.test.ts

@@ -0,0 +1,139 @@
+import { describe, expect, it, vi } from "vitest";
+
+import {
+  createChatTransactionRunner,
+  type ChatDatabase,
+} from "./drizzle-adapter";
+
+type DatabaseTransaction = Parameters<
+  Parameters<ChatDatabase["transaction"]>[0]
+>[0];
+type TransactionCallback = (tx: DatabaseTransaction) => Promise<unknown>;
+
+// These fakes exercise the transaction boundary, not a live PostgreSQL connection.
+function fakeDatabase(
+  transaction: (callback: TransactionCallback) => Promise<unknown>,
+): ChatDatabase {
+  return { transaction } as unknown as ChatDatabase;
+}
+
+const fakeTx = {} as DatabaseTransaction;
+
+function pgError(code: string): Error & { code: string } {
+  return Object.assign(new Error(`PostgreSQL ${code}`), { code });
+}
+
+describe("createChatTransactionRunner", () => {
+  it.each(["40001", "40P01"])(
+    "restarts the entire callback in a fresh transaction after %s",
+    async (code) => {
+      const transaction = vi.fn(async (callback: TransactionCallback) =>
+        callback(fakeTx),
+      );
+      const run = vi.fn(async () => {
+        if (run.mock.calls.length === 1) throw pgError(code);
+        return "committed";
+      });
+
+      await expect(
+        createChatTransactionRunner(fakeDatabase(transaction)).transaction(run),
+      ).resolves.toBe("committed");
+      expect(transaction).toHaveBeenCalledTimes(2);
+      expect(run).toHaveBeenCalledTimes(2);
+    },
+  );
+
+  it("replays the callback when the transaction rejects after it returns", async () => {
+    const transaction = vi.fn(async (callback: TransactionCallback) => {
+      const result = await callback(fakeTx);
+      if (transaction.mock.calls.length === 1) throw pgError("40001");
+      return result;
+    });
+    const run = vi.fn(async () => "committed");
+
+    await expect(
+      createChatTransactionRunner(fakeDatabase(transaction)).transaction(run),
+    ).resolves.toBe("committed");
+    expect(transaction).toHaveBeenCalledTimes(2);
+    expect(run).toHaveBeenCalledTimes(2);
+  });
+
+  it("retries a serialization failure wrapped in a cause", async () => {
+    const transaction = vi.fn(async (callback: TransactionCallback) =>
+      callback(fakeTx),
+    );
+    const run = vi.fn(async () => {
+      if (run.mock.calls.length === 1)
+        throw new Error("query failed", { cause: pgError("40001") });
+      return 42;
+    });
+
+    await expect(
+      createChatTransactionRunner(fakeDatabase(transaction)).transaction(run),
+    ).resolves.toBe(42);
+    expect(transaction).toHaveBeenCalledTimes(2);
+    expect(run).toHaveBeenCalledTimes(2);
+  });
+
+  it("stops after three total attempts and rethrows the last error unchanged", async () => {
+    const failures = [pgError("40001"), pgError("40P01"), pgError("40001")];
+    const transaction = vi.fn(async (callback: TransactionCallback) =>
+      callback(fakeTx),
+    );
+    const run = vi.fn(async () => {
+      throw failures[run.mock.calls.length - 1];
+    });
+
+    await expect(
+      createChatTransactionRunner(fakeDatabase(transaction)).transaction(run),
+    ).rejects.toBe(failures[2]);
+    expect(transaction).toHaveBeenCalledTimes(3);
+    expect(run).toHaveBeenCalledTimes(3);
+  });
+
+  it.each([
+    new Error("unrelated"),
+    pgError("23505"),
+    pgError("CHAT_FORBIDDEN"),
+  ])("does not retry unrelated or domain failures: %s", async (failure) => {
+    const transaction = vi.fn(async (callback: TransactionCallback) =>
+      callback(fakeTx),
+    );
+    const run = vi.fn(async () => {
+      throw failure;
+    });
+
+    await expect(
+      createChatTransactionRunner(fakeDatabase(transaction)).transaction(run),
+    ).rejects.toBe(failure);
+    expect(transaction).toHaveBeenCalledTimes(1);
+    expect(run).toHaveBeenCalledTimes(1);
+  });
+
+  it("replays the whole callback after a rolled-back attempt, not just the failed operation", async () => {
+    const committed: string[] = [];
+    let attempts = 0;
+    let staged: string[] = [];
+    const transaction = vi.fn(async (callback: TransactionCallback) => {
+      staged = [];
+      attempts++;
+      // A fake rollback model: failed attempts never append staged writes.
+      const result = await callback(fakeTx);
+      committed.push(...staged);
+      return result;
+    });
+    const run = vi.fn(async () => {
+      staged.push(`first-${attempts}`);
+      if (attempts === 1) throw pgError("40P01");
+      staged.push(`second-${attempts}`);
+      return "done";
+    });
+
+    await expect(
+      createChatTransactionRunner(fakeDatabase(transaction)).transaction(run),
+    ).resolves.toBe("done");
+    expect(committed).toEqual(["first-2", "second-2"]);
+    expect(run).toHaveBeenCalledTimes(2);
+    expect(transaction).toHaveBeenCalledTimes(2);
+  });
+});

packages/database/src/chat/drizzle-adapter.ts

@@ -20,7 +20,7 @@ type DatabaseTransaction = Parameters<
   Parameters<ChatDatabase["transaction"]>[0]
 >[0];
 
-function transactionAdapter(
+export function transactionAdapter(
   tx: DatabaseTransaction,
 ): ChatPersistenceTransaction {
   return {
@@ -52,6 +52,8 @@ function transactionAdapter(
             parentMessageId: message.parentMessageId,
             authorUserId: message.authorUserId!,
             clientIdempotencyKey: message.clientIdempotencyKey,
+            sendIntentFingerprint: message.sendIntentFingerprint,
+            status: message.status,
           }
         : null;
     },
@@ -206,12 +208,32 @@ function transactionAdapter(
   };
 }
 
+function retryable(error: unknown): boolean {
+  if (!error || typeof error !== "object") return false;
+  const candidate = error as { code?: unknown; cause?: unknown };
+  return (
+    candidate.code === "40001" ||
+    candidate.code === "40P01" ||
+    (candidate.cause !== undefined && retryable(candidate.cause))
+  );
+}
+
 export function createChatTransactionRunner(
   database: ChatDatabase = defaultDb,
 ): ChatTransactionRunner {
   return {
-    transaction: (run) =>
+    async transaction(run) {
-      database.transaction((tx) => run(transactionAdapter(tx))),
+      for (let attempt = 0; attempt < 3; attempt++) {
+        try {
+          return await database.transaction((tx) =>
+            run(transactionAdapter(tx)),
+          );
+        } catch (error) {
+          if (attempt === 2 || !retryable(error)) throw error;
+        }
+      }
+      throw new Error("Unreachable transaction retry state");
+    },
   };
 }
 

packages/database/src/chat/index.ts

@@ -1,4 +1,7 @@
 export * from "./channel-drizzle-adapter";
 export * from "./drizzle-adapter";
 export * from "./manage-channels";
+export * from "./mutate-message";
+export * from "./message-drizzle-adapter";
+export * from "./read-history";
 export * from "./send-message";

packages/database/src/chat/manage-channels.test.ts

@@ -70,6 +70,15 @@ describe("channel lifecycle orchestration", () => {
     await removePrivateMember(s.run, { ...scope, channelId, userId: target });
     expect(s.events).toEqual([]);
   });
+  it("emits membership events only when add and remove change rows", async () => {
+    const s = seam();
+    await addPrivateMember(s.run, { ...scope, channelId, userId: target });
+    await removePrivateMember(s.run, { ...scope, channelId, userId: target });
+    expect(s.events).toMatchObject([
+      { eventType: "chat.channel.member.added.v1", subjectUserId: target },
+      { eventType: "chat.channel.member.removed.v1", subjectUserId: target },
+    ]);
+  });
   it("protects the private owner", async () => {
     const s = seam();
     await expect(
@@ -129,6 +138,121 @@ describe("channel lifecycle orchestration", () => {
       }),
     ).rejects.toMatchObject({ code: "CONFLICT" });
   });
+  it("denies replay of a private channel receipt after membership revocation", async () => {
+    const original = channel();
+    const s = seam({
+      loadReceipt: async () => ({
+        operation: "create_channel",
+        fingerprint: createHash("sha256")
+          .update(JSON.stringify({ kind: "private", name: "private" }))
+          .digest("hex"),
+        result: { operation: "create_channel", channel: original },
+      }),
+      isChannelMember: async () => false,
+    });
+    await expect(
+      createChannel(s.run, { ...scope, kind: "private", name: "private" }),
+    ).rejects.toMatchObject({ code: "NOT_FOUND" });
+    expect(s.events).toEqual([]);
+  });
+  it("denies replay of a private member receipt after membership revocation", async () => {
+    const s = seam({
+      loadReceipt: async () => ({
+        operation: "add_private_member",
+        fingerprint: createHash("sha256")
+          .update(JSON.stringify({ channelId, userId: target }))
+          .digest("hex"),
+        result: {
+          operation: "add_private_member",
+          member: { channelId, userId: target, changed: true },
+        },
+      }),
+      isChannelMember: async () => false,
+    });
+    await expect(
+      addPrivateMember(s.run, { ...scope, channelId, userId: target }),
+    ).rejects.toMatchObject({ code: "NOT_FOUND" });
+  });
+  it("does not disclose a saved remove-self result after leaving a private channel", async () => {
+    const s = seam({
+      lockChannel: async () => channel({ ownerUserId: target }),
+      loadReceipt: async () => ({
+        operation: "remove_private_member",
+        fingerprint: createHash("sha256")
+          .update(JSON.stringify({ channelId, userId: actorUserId }))
+          .digest("hex"),
+        result: {
+          operation: "remove_private_member",
+          member: { channelId, userId: actorUserId, changed: true },
+        },
+      }),
+      isChannelMember: async () => false,
+    });
+    await expect(
+      removePrivateMember(s.run, { ...scope, channelId, userId: actorUserId }),
+    ).rejects.toMatchObject({ code: "NOT_FOUND" });
+  });
+  it("allows public receipt replay without channel membership", async () => {
+    const original = channel({ kind: "public" });
+    const s = seam({
+      lockChannel: async () => original,
+      isChannelMember: async () => false,
+      loadReceipt: async () => ({
+        operation: "create_channel",
+        fingerprint: createHash("sha256")
+          .update(JSON.stringify({ kind: "public", name: "private" }))
+          .digest("hex"),
+        result: { operation: "create_channel", channel: original },
+      }),
+    });
+    await expect(
+      createChannel(s.run, { ...scope, kind: "public", name: "private" }),
+    ).resolves.toMatchObject({ id: channelId });
+  });
+  it("denies replay of a direct channel receipt after membership revocation", async () => {
+    const participants = [actorUserId, target].sort();
+    const original = channel({ kind: "direct", name: null, directKey: "key" });
+    const s = seam({
+      lockChannel: async () => original,
+      isChannelMember: async () => false,
+      loadReceipt: async () => ({
+        operation: "create_direct",
+        fingerprint: createHash("sha256")
+          .update(JSON.stringify({ participantUserIds: participants }))
+          .digest("hex"),
+        result: { operation: "create_direct", channel: original },
+      }),
+    });
+    await expect(
+      createDirectChannel(s.run, {
+        ...scope,
+        participantUserIds: participants,
+      }),
+    ).rejects.toMatchObject({ code: "NOT_FOUND" });
+  });
+  it("replays readable archived channels without rechecking write permissions", async () => {
+    const archived = channel({ archivedAt: new Date("2026-01-01T00:00:00Z") });
+    const s = seam({
+      lockChannel: async () => archived,
+      loadReceipt: async () => ({
+        operation: "rename_channel",
+        fingerprint: createHash("sha256")
+          .update(JSON.stringify({ channelId, name: "next" }))
+          .digest("hex"),
+        result: {
+          operation: "rename_channel",
+          channel: {
+            ...archived,
+            archivedAt: archived.archivedAt?.toISOString(),
+          },
+        },
+      }),
+    });
+    await expect(
+      renameChannel(s.run, { ...scope, channelId, name: "next" }),
+    ).resolves.toMatchObject({ id: channelId, name: "private" });
+    expect(s.events).toEqual([]);
+  });
   it("replays a matching receipt without state or event calls", async () => {
     const original = channel();
     const insert = vi.fn();

packages/database/src/chat/manage-channels.ts

@@ -163,7 +163,23 @@ async function withReceipt(
   if (receipt) {
     if (receipt.operation !== operation || receipt.fingerprint !== fingerprint)
       throw new ChannelPersistenceError("CONFLICT");
-    return chatLifecycleReceiptResultSchema.parse(receipt.result);
+    const result = chatLifecycleReceiptResultSchema.parse(receipt.result);
+    const channelId =
+      "channel" in result ? result.channel.id : result.member.channelId;
+    const channel = await tx.lockChannel(scope.workspaceId, channelId);
+    if (
+      !channel ||
+      ("channel" in result &&
+        result.channel.workspaceId !== scope.workspaceId) ||
+      (channel.kind !== "public" &&
+        !(await tx.isChannelMember(
+          scope.workspaceId,
+          channelId,
+          scope.actorUserId,
+        )))
+    )
+      throw new ChannelPersistenceError("NOT_FOUND");
+    return result;
   }
   const result = chatLifecycleReceiptResultSchema.parse(await execute());
   if (result.operation !== operation)

packages/database/src/chat/message-drizzle-adapter.test.ts

@@ -0,0 +1,53 @@
+import { describe, expect, it, vi } from "vitest";
+import { PgDialect } from "drizzle-orm/pg-core";
+import { chatOutbox } from "../schema";
+import { createMessageMutationRunner } from "./message-drizzle-adapter";
+
+type Database = NonNullable<Parameters<typeof createMessageMutationRunner>[0]>;
+type Tx = Parameters<Parameters<Database["transaction"]>[0]>[0];
+
+describe("created outbox redaction", () => {
+  it("binds plain JSON metadata and scopes redaction to the workspace/channel/message/create event", async () => {
+    const where = vi.fn(async () => undefined);
+    const set = vi.fn(() => ({ where }));
+    const update = vi.fn(() => ({ set }));
+    const tx = { update } as unknown as Tx;
+    const database = {
+      transaction: async (run: (tx: Tx) => Promise<unknown>) => run(tx),
+    } as unknown as Database;
+    const scope = {
+      workspaceId: "10000000-0000-4000-8000-000000000001",
+      channelId: "20000000-0000-4000-8000-000000000001",
+      messageId: "30000000-0000-4000-8000-000000000001",
+    };
+    await createMessageMutationRunner(database).transaction((adapter) =>
+      adapter.redactCreatedOutbox(scope),
+    );
+    expect(update).toHaveBeenCalledWith(chatOutbox);
+    expect(set).toHaveBeenCalledWith({
+      payload: {
+        eventType: "chat.message.created.v1",
+        eventVersion: 1,
+        ...scope,
+      },
+    });
+    const condition = (
+      where.mock.calls as unknown as [[Parameters<PgDialect["sqlToQuery"]>[0]]]
+    )[0][0];
+    const query = new PgDialect().sqlToQuery(condition);
+    expect(query.params).toEqual([
+      scope.workspaceId,
+      scope.channelId,
+      scope.messageId,
+      "chat.message.created.v1",
+    ]);
+    for (const column of [
+      "workspace_id",
+      "channel_id",
+      "message_id",
+      "event_type",
+    ]) {
+      expect(query.sql).toContain(`"${column}"`);
+    }
+  });
+});

packages/database/src/chat/message-drizzle-adapter.ts

@@ -0,0 +1,253 @@
+import { chatMessageMutationEventV1Schema } from "@tasks/shared";
+import { createHash } from "node:crypto";
+import { and, eq, sql } from "drizzle-orm";
+
+import { db as defaultDb } from "../client";
+import {
+  chatChannels,
+  chatMessageMutations,
+  chatMessageMentions,
+  chatMessages,
+  chatMutationReceipts,
+  chatOutbox,
+} from "../schema";
+import { transactionAdapter } from "./drizzle-adapter";
+import {
+  mutateChatMessage,
+  type MessageMutationRunner,
+  type MessageMutationTransaction,
+} from "./mutate-message";
+
+// Reuse the send adapter's tenant-scoped membership lookup; mutations use the same channel policy.
+type Db = typeof defaultDb;
+type Tx = Parameters<Parameters<Db["transaction"]>[0]>[0];
+function adapter(tx: Tx): MessageMutationTransaction {
+  return {
+    async claim(scope) {
+      const lockKey = `${scope.workspaceId}:${scope.actorUserId}:${scope.clientMutationId}`;
+      await tx.execute(
+        sql`select pg_advisory_xact_lock(hashtextextended(${lockKey}, 0))`,
+      );
+      const [receipt] = await tx
+        .select()
+        .from(chatMutationReceipts)
+        .where(
+          and(
+            eq(chatMutationReceipts.workspaceId, scope.workspaceId),
+            eq(chatMutationReceipts.actorUserId, scope.actorUserId),
+            eq(chatMutationReceipts.clientMutationId, scope.clientMutationId),
+          ),
+        )
+        .limit(1);
+      return receipt ?? null;
+    },
+    async facts(scope) {
+      // This is deliberately the same member/channel lookup as send, not a role-based shortcut.
+      return transactionAdapter(tx).loadSendFacts(scope);
+    },
+    async lockMessage(scope) {
+      const [message] = await tx
+        .select({
+          id: chatMessages.id,
+          workspaceId: chatMessages.workspaceId,
+          channelId: chatMessages.channelId,
+          authorUserId: chatMessages.authorUserId,
+          status: chatMessages.status,
+          revision: chatMessages.revision,
+        })
+        .from(chatMessages)
+        .where(
+          and(
+            eq(chatMessages.workspaceId, scope.workspaceId),
+            eq(chatMessages.channelId, scope.channelId),
+            eq(chatMessages.id, scope.messageId),
+          ),
+        )
+        .for("update")
+        .limit(1);
+      return message ?? null;
+    },
+    async pinLegacySendIntent(scope) {
+      const [message] = await tx
+        .select({
+          body: chatMessages.body,
+          authorUserId: chatMessages.authorUserId,
+          parentMessageId: chatMessages.parentMessageId,
+          sendIntentFingerprint: chatMessages.sendIntentFingerprint,
+        })
+        .from(chatMessages)
+        .where(
+          and(
+            eq(chatMessages.workspaceId, scope.workspaceId),
+            eq(chatMessages.channelId, scope.channelId),
+            eq(chatMessages.id, scope.messageId),
+          ),
+        )
+        .limit(1);
+      if (!message || message.sendIntentFingerprint) return;
+      const mentions = await tx
+        .select({ userId: chatMessageMentions.userId })
+        .from(chatMessageMentions)
+        .where(
+          and(
+            eq(chatMessageMentions.workspaceId, scope.workspaceId),
+            eq(chatMessageMentions.messageId, scope.messageId),
+          ),
+        );
+      const fingerprint = createHash("sha256")
+        .update(
+          JSON.stringify([
+            scope.workspaceId,
+            scope.channelId,
+            message.authorUserId,
+            message.body,
+            message.parentMessageId,
+            mentions.map((row) => row.userId).sort(),
+          ]),
+        )
+        .digest("hex");
+      await tx
+        .update(chatMessages)
+        .set({ sendIntentFingerprint: fingerprint })
+        .where(
+          and(
+            eq(chatMessages.workspaceId, scope.workspaceId),
+            eq(chatMessages.channelId, scope.channelId),
+            eq(chatMessages.id, scope.messageId),
+          ),
+        );
+    },
+    async apply(scope) {
+      const [updated] = await tx
+        .update(chatMessages)
+        .set(
+          scope.type === "edit_message"
+            ? {
+                body: scope.body!,
+                status: "edited",
+                editedAt: new Date(),
+                revision: scope.revision,
+              }
+            : {
+                body: "",
+                status: "deleted",
+                deletedAt: new Date(),
+                revision: scope.revision,
+              },
+        )
+        .where(
+          and(
+            eq(chatMessages.workspaceId, scope.workspaceId),
+            eq(chatMessages.channelId, scope.channelId),
+            eq(chatMessages.id, scope.messageId),
+            eq(chatMessages.revision, scope.revision - 1),
+          ),
+        )
+        .returning({ id: chatMessages.id });
+      if (!updated) throw new Error("Concurrent message update");
+      const [channel] = await tx
+        .update(chatChannels)
+        .set({
+          nextMutationSequence: sql`${chatChannels.nextMutationSequence} + 1`,
+        })
+        .where(
+          and(
+            eq(chatChannels.workspaceId, scope.workspaceId),
+            eq(chatChannels.id, scope.channelId),
+          ),
+        )
+        .returning({ nextMutationSequence: chatChannels.nextMutationSequence });
+      if (!channel) throw new Error("Channel disappeared during mutation");
+      const status = scope.type === "edit_message" ? "edited" : "deleted";
+      await tx.insert(chatMessageMutations).values({
+        workspaceId: scope.workspaceId,
+        channelId: scope.channelId,
+        messageId: scope.messageId,
+        mutationSequence: channel.nextMutationSequence - 1n,
+        revision: scope.revision,
+        kind: status,
+      });
+      return { messageId: updated.id, revision: scope.revision, status };
+    },
+    async redactCreatedOutbox(scope) {
+      // Historical create payloads contain the original body. Strip them before publishing
+      // a mutation so a delayed worker cannot replay stale content after edit/delete.
+      await tx
+        .update(chatOutbox)
+        .set({
+          // Plain JSON — avoid jsonb_build_object() params (Postgres: could not determine data type of parameter $1).
+          payload: {
+            eventType: "chat.message.created.v1",
+            eventVersion: 1,
+            workspaceId: scope.workspaceId,
+            channelId: scope.channelId,
+            messageId: scope.messageId,
+          },
+        })
+        .where(
+          and(
+            eq(chatOutbox.workspaceId, scope.workspaceId),
+            eq(chatOutbox.channelId, scope.channelId),
+            eq(chatOutbox.messageId, scope.messageId),
+            eq(chatOutbox.eventType, "chat.message.created.v1"),
+          ),
+        );
+    },
+    async insertOutbox(scope) {
+      await tx.insert(chatOutbox).values({
+        workspaceId: scope.workspaceId,
+        channelId: scope.channelId,
+        messageId: scope.messageId,
+        eventVersion: 1,
+        eventType: scope.eventType,
+        dedupeKey: `mutation:${scope.actorUserId}:${scope.clientMutationId}`,
+        payload: chatMessageMutationEventV1Schema.parse({
+          eventType: scope.eventType,
+          eventVersion: 1,
+          workspaceId: scope.workspaceId,
+          channelId: scope.channelId,
+          messageId: scope.messageId,
+          revision: scope.result.revision,
+          status: scope.result.status,
+        }),
+      });
+    },
+    async saveReceipt(scope) {
+      await tx.insert(chatMutationReceipts).values({ ...scope });
+    },
+  };
+}
+
+function retryable(error: unknown): boolean {
+  if (!error || typeof error !== "object") return false;
+  const e = error as { code?: unknown; cause?: unknown };
+  return (
+    e.code === "40001" ||
+    e.code === "40P01" ||
+    (e.cause !== undefined && retryable(e.cause))
+  );
+}
+export function createMessageMutationRunner(
+  database: Db = defaultDb,
+): MessageMutationRunner {
+  return {
+    async transaction(run) {
+      for (let attempt = 0; attempt < 3; attempt++) {
+        try {
+          return await database.transaction((tx) => run(adapter(tx)), {
+            isolationLevel: "repeatable read",
+          });
+        } catch (error) {
+          if (attempt === 2 || !retryable(error)) throw error;
+        }
+      }
+      throw new Error("Unreachable retry state");
+    },
+  };
+}
+export async function mutateChatMessageWithDb(
+  input: Parameters<typeof mutateChatMessage>[1],
+  database: Db = defaultDb,
+) {
+  return mutateChatMessage(createMessageMutationRunner(database), input);
+}

packages/database/src/chat/mutate-message.test.ts

@@ -0,0 +1,181 @@
+import { describe, expect, it } from "vitest";
+import {
+  mutateChatMessage,
+  type MessageMutationTransaction,
+} from "./mutate-message";
+
+const workspaceId = "11111111-1111-4111-8111-111111111111";
+const channelId = "22222222-2222-4222-8222-222222222222";
+const actorUserId = "33333333-3333-4333-8333-333333333333";
+const messageId = "44444444-4444-4444-8444-444444444444";
+const clientMutationId = "55555555-5555-4555-8555-555555555555";
+const command = (type: "edit_message" | "delete_message", extra = {}) => ({
+  type,
+  workspaceId,
+  channelId,
+  context: { type: "workspace", workspaceId },
+  messageId,
+  expectedRevision: 1,
+  clientMutationId,
+  ...(type === "edit_message" ? { body: "updated" } : {}),
+  ...extra,
+});
+function setup(
+  options: {
+    kind?: "public" | "private";
+    member?: boolean;
+    archived?: boolean;
+    author?: string;
+    receipt?: boolean;
+    revision?: number;
+  } = {},
+) {
+  const calls: string[] = [];
+  const tx: MessageMutationTransaction = {
+    claim: async () => {
+      calls.push("claim");
+      return options.receipt
+        ? {
+            operation: "edit_message",
+            fingerprint: "mismatch",
+            result: { revision: 2 },
+          }
+        : null;
+    },
+    facts: async () => {
+      calls.push("facts");
+      return {
+        actor: {
+          userId: actorUserId,
+          workspaceId,
+          activeWorkspaceMember: true,
+          role: "admin" as const,
+        },
+        channel: {
+          id: channelId,
+          workspaceId,
+          kind: options.kind ?? "public",
+          activeMember: options.member ?? false,
+          archived: options.archived ?? false,
+          ownerUserId: actorUserId,
+        },
+      };
+    },
+    lockMessage: async () => {
+      calls.push("lock");
+      return {
+        id: messageId,
+        workspaceId,
+        channelId,
+        authorUserId: options.author ?? actorUserId,
+        status: "sent" as const,
+        revision: options.revision ?? 1,
+      };
+    },
+    pinLegacySendIntent: async () => {
+      calls.push("pin");
+    },
+    apply: async () => {
+      calls.push("apply");
+      return { messageId, revision: 2, status: "edited" as const };
+    },
+    redactCreatedOutbox: async () => {
+      calls.push("redact");
+    },
+    insertOutbox: async () => {
+      calls.push("outbox");
+    },
+    saveReceipt: async () => {
+      calls.push("receipt");
+    },
+  };
+  return {
+    calls,
+    runner: {
+      transaction: async <T>(
+        fn: (tx: MessageMutationTransaction) => Promise<T>,
+      ) => fn(tx),
+    },
+  };
+}
+describe("message mutations", () => {
+  it("rejects a private-channel nonmember even if admin", async () => {
+    const { runner, calls } = setup({ kind: "private" });
+    await expect(
+      mutateChatMessage(runner, {
+        actorUserId,
+        command: command("edit_message"),
+      }),
+    ).rejects.toMatchObject({ code: "CHAT_FORBIDDEN" });
+    expect(calls).not.toContain("apply");
+  });
+  it("does not grant admin moderation over another author's message", async () => {
+    const { runner, calls } = setup({
+      author: "66666666-6666-4666-8666-666666666666",
+    });
+    await expect(
+      mutateChatMessage(runner, {
+        actorUserId,
+        command: command("delete_message"),
+      }),
+    ).rejects.toMatchObject({ code: "CHAT_FORBIDDEN" });
+    expect(calls).not.toContain("apply");
+  });
+  it("rejects archived writes", async () => {
+    const { runner } = setup({ archived: true });
+    await expect(
+      mutateChatMessage(runner, {
+        actorUserId,
+        command: command("delete_message"),
+      }),
+    ).rejects.toMatchObject({ code: "CHAT_FORBIDDEN" });
+  });
+  it("requires matching revision before writing", async () => {
+    const { runner, calls } = setup({ revision: 3 });
+    await expect(
+      mutateChatMessage(runner, {
+        actorUserId,
+        command: command("edit_message"),
+      }),
+    ).rejects.toMatchObject({ code: "CHAT_REVISION_CONFLICT" });
+    expect(calls).not.toContain("apply");
+  });
+  it("rejects reused mutation IDs before writing", async () => {
+    const { runner, calls } = setup({ receipt: true });
+    await expect(
+      mutateChatMessage(runner, {
+        actorUserId,
+        command: command("edit_message"),
+      }),
+    ).rejects.toMatchObject({ code: "CHAT_IDEMPOTENCY_CONFLICT" });
+    expect(calls).not.toContain("apply");
+  });
+  it("redacts queued created events in the same transaction", async () => {
+    const { runner, calls } = setup();
+    await mutateChatMessage(runner, {
+      actorUserId,
+      command: command("edit_message"),
+    });
+    expect(calls).toEqual([
+      "claim",
+      "facts",
+      "lock",
+      "pin",
+      "apply",
+      "redact",
+      "outbox",
+      "receipt",
+    ]);
+  });
+  it("rejects object-scoped commands at validation", async () => {
+    const { runner } = setup();
+    await expect(
+      mutateChatMessage(runner, {
+        actorUserId,
+        command: command("edit_message", {
+          context: { type: "object", objectId: messageId },
+        }),
+      }),
+    ).rejects.toThrow();
+  });
+});

packages/database/src/chat/mutate-message.ts

@@ -0,0 +1,249 @@
+import { createHash } from "node:crypto";
+import {
+  canPerformChatAction,
+  chatCommandSchema,
+  type ChatCommand,
+  type ChatPolicyInput,
+} from "@tasks/shared";
+import { z } from "zod";
+
+const commandSchema = chatCommandSchema.refine(
+  (
+    command,
+  ): command is Extract<
+    ChatCommand,
+    { type: "edit_message" | "delete_message" }
+  > => command.type === "edit_message" || command.type === "delete_message",
+  "Expected message mutation",
+);
+type Command = z.infer<typeof commandSchema>;
+type Facts = Pick<
+  Extract<ChatPolicyInput, { action: "send" }>,
+  "actor" | "channel"
+>;
+export type MutationResult = {
+  messageId: string;
+  revision: number;
+  status: "edited" | "deleted";
+};
+export type MessageMutationTransaction = {
+  claim(scope: {
+    workspaceId: string;
+    actorUserId: string;
+    clientMutationId: string;
+  }): Promise<{
+    operation: string;
+    fingerprint: string;
+    result: unknown;
+  } | null>;
+  facts(scope: {
+    workspaceId: string;
+    channelId: string;
+    actorUserId: string;
+  }): Promise<Facts | null>;
+  lockMessage(scope: {
+    workspaceId: string;
+    channelId: string;
+    messageId: string;
+  }): Promise<{
+    id: string;
+    workspaceId: string;
+    channelId: string;
+    authorUserId: string | null;
+    status: string;
+    revision: number;
+  } | null>;
+  pinLegacySendIntent(scope: {
+    workspaceId: string;
+    channelId: string;
+    messageId: string;
+  }): Promise<void>;
+  apply(scope: {
+    workspaceId: string;
+    channelId: string;
+    messageId: string;
+    revision: number;
+    type: Command["type"];
+    body?: string;
+  }): Promise<MutationResult>;
+  redactCreatedOutbox(scope: {
+    workspaceId: string;
+    channelId: string;
+    messageId: string;
+  }): Promise<void>;
+  insertOutbox(scope: {
+    workspaceId: string;
+    channelId: string;
+    messageId: string;
+    actorUserId: string;
+    result: MutationResult;
+    eventType: "chat.message.edited.v1" | "chat.message.deleted.v1";
+    clientMutationId: string;
+  }): Promise<void>;
+  saveReceipt(scope: {
+    workspaceId: string;
+    actorUserId: string;
+    clientMutationId: string;
+    operation: string;
+    fingerprint: string;
+    result: MutationResult;
+  }): Promise<void>;
+};
+export type MessageMutationRunner = {
+  transaction<T>(
+    callback: (tx: MessageMutationTransaction) => Promise<T>,
+  ): Promise<T>;
+};
+export class ChatMutationError extends Error {
+  constructor(
+    readonly code:
+      | "CHAT_FORBIDDEN"
+      | "CHAT_NOT_FOUND"
+      | "CHAT_REVISION_CONFLICT"
+      | "CHAT_IDEMPOTENCY_CONFLICT",
+  ) {
+    super(code);
+    this.name = "ChatMutationError";
+  }
+}
+const resultSchema = z
+  .object({
+    messageId: z.string().uuid(),
+    revision: z.number().int().positive(),
+    status: z.enum(["edited", "deleted"]),
+  })
+  .strict();
+
+export async function mutateChatMessage(
+  runner: MessageMutationRunner,
+  input: { actorUserId: string; command: unknown },
+): Promise<{ changed: boolean; result: MutationResult }> {
+  const actorUserId = z.string().uuid().parse(input.actorUserId);
+  const command = commandSchema.parse(input.command);
+  const fingerprint = createHash("sha256")
+    .update(
+      JSON.stringify([
+        command.type,
+        command.workspaceId,
+        command.channelId,
+        command.messageId,
+        command.expectedRevision,
+        command.type === "edit_message" ? command.body : null,
+      ]),
+    )
+    .digest("hex");
+  return runner.transaction(async (tx) => {
+    const scope = {
+      workspaceId: command.workspaceId,
+      channelId: command.channelId,
+      actorUserId,
+    };
+    const receipt = await tx.claim({
+      workspaceId: command.workspaceId,
+      actorUserId,
+      clientMutationId: command.clientMutationId,
+    });
+    const facts = await tx.facts(scope);
+    // Receipts remain readable after archive, but new writes cannot target an archived channel.
+    if (
+      !facts ||
+      !canPerformChatAction({
+        action: "read",
+        ...facts,
+        context: command.context,
+      }) ||
+      (!receipt && facts.channel.archived)
+    )
+      throw new ChatMutationError("CHAT_FORBIDDEN");
+    if (receipt) {
+      if (
+        receipt.operation !== command.type ||
+        receipt.fingerprint !== fingerprint
+      )
+        throw new ChatMutationError("CHAT_IDEMPOTENCY_CONFLICT");
+      const result = resultSchema.parse(receipt.result);
+      const message = await tx.lockMessage({
+        workspaceId: command.workspaceId,
+        channelId: command.channelId,
+        messageId: command.messageId,
+      });
+      if (
+        !message ||
+        message.authorUserId !== actorUserId ||
+        result.messageId !== command.messageId ||
+        result.status !==
+          (command.type === "edit_message" ? "edited" : "deleted")
+      )
+        throw new ChatMutationError("CHAT_FORBIDDEN");
+      return { changed: false, result };
+    }
+    const message = await tx.lockMessage({
+      workspaceId: command.workspaceId,
+      channelId: command.channelId,
+      messageId: command.messageId,
+    });
+    if (
+      !message ||
+      message.workspaceId !== command.workspaceId ||
+      message.channelId !== command.channelId
+    )
+      throw new ChatMutationError("CHAT_NOT_FOUND");
+    if (
+      !message.authorUserId ||
+      !canPerformChatAction({
+        action: "edit_own",
+        ...facts,
+        context: command.context,
+        message: {
+          id: message.id,
+          workspaceId: message.workspaceId,
+          channelId: message.channelId,
+          author: { type: "human", userId: message.authorUserId },
+        },
+      })
+    )
+      throw new ChatMutationError("CHAT_FORBIDDEN");
+    if (
+      message.revision !== command.expectedRevision ||
+      message.status === "deleted"
+    )
+      throw new ChatMutationError("CHAT_REVISION_CONFLICT");
+    await tx.pinLegacySendIntent({
+      workspaceId: command.workspaceId,
+      channelId: command.channelId,
+      messageId: command.messageId,
+    });
+    const result = await tx.apply({
+      workspaceId: command.workspaceId,
+      channelId: command.channelId,
+      messageId: command.messageId,
+      revision: command.expectedRevision + 1,
+      type: command.type,
+      ...(command.type === "edit_message" ? { body: command.body } : {}),
+    });
+    await tx.redactCreatedOutbox({
+      workspaceId: command.workspaceId,
+      channelId: command.channelId,
+      messageId: command.messageId,
+    });
+    await tx.insertOutbox({
+      ...scope,
+      messageId: command.messageId,
+      clientMutationId: command.clientMutationId,
+      result,
+      eventType:
+        command.type === "edit_message"
+          ? "chat.message.edited.v1"
+          : "chat.message.deleted.v1",
+    });
+    await tx.saveReceipt({
+      workspaceId: command.workspaceId,
+      actorUserId,
+      clientMutationId: command.clientMutationId,
+      operation: command.type,
+      fingerprint,
+      result,
+    });
+    return { changed: true, result };
+  });
+}

packages/database/src/chat/read-history.postgres.test.ts

@@ -0,0 +1,405 @@
+import { afterAll, describe, expect, it } from "vitest";
+import { drizzle } from "drizzle-orm/postgres-js";
+import { and, eq, sql } from "drizzle-orm";
+import postgres from "postgres";
+
+import * as schema from "../schema";
+import {
+  chatChannelMembers,
+  chatChannels,
+  chatMessageMutations,
+  chatMessageMentions,
+  chatMessages,
+  chatOutbox,
+  chatReadCursors,
+  users,
+  workspaceMembers,
+  workspaces,
+} from "../schema";
+import {
+  chatCatchUp,
+  chatMutationCatchUp,
+  chatHistory,
+  chatMarkChannelRead,
+  chatThreadHistory,
+  chatUnreadSummary,
+  ChatReadError,
+} from "./read-history";
+
+// Run explicitly against a disposable PostgreSQL instance with migrations through 0023.
+const url = process.env.PG_URL;
+const client = url ? postgres(url, { max: 12 }) : null;
+const db = client ? drizzle(client, { schema }) : null;
+afterAll(async () => {
+  await client?.end();
+});
+
+describe.skipIf(!url)("chat read APIs on PostgreSQL (0023)", () => {
+  it("recovers mutation pages with current safe projections and denies private/direct outsiders", async () => {
+    if (!db) throw new Error("PG_URL is required");
+    const [owner, reader] = await db.insert(users).values([
+      { email: `mutation-owner-${crypto.randomUUID()}@example.test` },
+      { email: `mutation-reader-${crypto.randomUUID()}@example.test` },
+    ]).returning();
+    const [workspace] = await db.insert(workspaces).values({
+      slug: `mutation-${crypto.randomUUID()}`, name: "Mutation PG", ownerUserId: owner.id,
+    }).returning();
+    await db.insert(workspaceMembers).values([owner, reader].map((user) => ({
+      workspaceId: workspace.id, userId: user.id, role: "member" as const,
+    })));
+    const [channel, hidden, direct] = await db.insert(chatChannels).values([
+      { workspaceId: workspace.id, ownerUserId: owner.id, kind: "private", name: "private", nextSequence: 3n, nextMutationSequence: 4n },
+      { workspaceId: workspace.id, ownerUserId: owner.id, kind: "private", name: "hidden" },
+      { workspaceId: workspace.id, ownerUserId: owner.id, kind: "direct", directKey: `${owner.id}:${reader.id}` },
+    ]).returning();
+    await db.insert(chatChannelMembers).values([
+      { workspaceId: workspace.id, channelId: channel.id, userId: owner.id },
+      { workspaceId: workspace.id, channelId: hidden.id, userId: owner.id },
+      { workspaceId: workspace.id, channelId: direct.id, userId: owner.id },
+    ]);
+    const [edited, deleted] = await db.insert(chatMessages).values([
+      { workspaceId: workspace.id, channelId: channel.id, sequence: 1n, authorType: "human", authorUserId: owner.id, body: "latest", status: "edited", revision: 4, clientIdempotencyKey: crypto.randomUUID() },
+      { workspaceId: workspace.id, channelId: channel.id, sequence: 2n, authorType: "human", authorUserId: owner.id, body: "stored secret", status: "deleted", revision: 3, clientIdempotencyKey: crypto.randomUUID() },
+    ]).returning();
+    await db.insert(chatMessageMutations).values([
+      { workspaceId: workspace.id, channelId: channel.id, messageId: edited.id, mutationSequence: 1n, revision: 2, kind: "edited" },
+      { workspaceId: workspace.id, channelId: channel.id, messageId: edited.id, mutationSequence: 2n, revision: 4, kind: "edited" },
+      { workspaceId: workspace.id, channelId: channel.id, messageId: deleted.id, mutationSequence: 3n, revision: 3, kind: "deleted" },
+    ]);
+    const scope = { workspaceId: workspace.id, channelId: channel.id, actorUserId: owner.id };
+    const first = await chatMutationCatchUp(db, { ...scope, after: "0", limit: 1 });
+    expect(first).toMatchObject({ next: "1", through: "3", hasMore: true,
+      mutations: [{ messageId: edited.id, mutationSequence: "1", revision: 2 }],
+      messages: [{ id: edited.id, body: "latest", revision: 4 }],
+    });
+    expect(JSON.stringify(first)).not.toContain("clientIdempotencyKey");
+    const second = await chatMutationCatchUp(db, { ...scope, after: first.next, through: first.through, limit: 2 });
+    expect(second).toMatchObject({ next: "3", through: "3", hasMore: false,
+      mutations: [{ messageId: edited.id }, { messageId: deleted.id }],
+      messages: [{ id: edited.id, body: "latest" }, { id: deleted.id, body: "", status: "deleted" }],
+    });
+    expect(JSON.stringify(second)).not.toContain("stored secret");
+    await expect(chatMutationCatchUp(db, { ...scope, actorUserId: reader.id, after: "0", limit: 1 })).rejects.toBeInstanceOf(ChatReadError);
+    for (const channelId of [hidden.id, direct.id]) {
+      await expect(chatMutationCatchUp(db, { ...scope, channelId, actorUserId: reader.id, after: "0", limit: 1 })).rejects.toBeInstanceOf(ChatReadError);
+    }
+    await expect(chatMutationCatchUp(db, { ...scope, channelId: hidden.id, after: "0", through: "3", limit: 1 })).rejects.toBeInstanceOf(ChatReadError);
+  });
+  it("paginates roots and replies, catches up, counts unread, and protects archived/private channels", async () => {
+    if (!db) throw new Error("PG_URL is required");
+    const [owner, reader, outsider] = await db
+      .insert(users)
+      .values([
+        { email: `chat-pg-owner-${crypto.randomUUID()}@example.test` },
+        { email: `chat-pg-reader-${crypto.randomUUID()}@example.test` },
+        { email: `chat-pg-outsider-${crypto.randomUUID()}@example.test` },
+      ])
+      .returning();
+    const [workspace] = await db
+      .insert(workspaces)
+      .values({
+        slug: `chat-pg-${crypto.randomUUID()}`,
+        name: "Chat PG",
+        ownerUserId: owner.id,
+      })
+      .returning();
+    await db.insert(workspaceMembers).values(
+      [owner, reader, outsider].map((user) => ({
+        workspaceId: workspace.id,
+        userId: user.id,
+      })),
+    );
+    const [publicChannel, archived, hidden] = await db
+      .insert(chatChannels)
+      .values([
+        {
+          workspaceId: workspace.id,
+          ownerUserId: owner.id,
+          kind: "public",
+          name: "general",
+          nextSequence: 5n,
+        },
+        {
+          workspaceId: workspace.id,
+          ownerUserId: owner.id,
+          kind: "private",
+          name: "old",
+          archivedAt: new Date(),
+          nextSequence: 2n,
+        },
+        {
+          workspaceId: workspace.id,
+          ownerUserId: owner.id,
+          kind: "private",
+          name: "secret",
+          nextSequence: 2n,
+        },
+      ])
+      .returning();
+    await db.insert(chatChannelMembers).values([
+      { workspaceId: workspace.id, channelId: archived.id, userId: owner.id },
+      { workspaceId: workspace.id, channelId: archived.id, userId: reader.id },
+      { workspaceId: workspace.id, channelId: hidden.id, userId: owner.id },
+    ]);
+    const scope = {
+      workspaceId: workspace.id,
+      actorUserId: reader.id,
+      channelId: publicChannel.id,
+    };
+    const [root, reply, tombstone, secondRoot] = await db
+      .insert(chatMessages)
+      .values([
+        {
+          workspaceId: workspace.id,
+          channelId: publicChannel.id,
+          sequence: 1n,
+          authorType: "human",
+          authorUserId: owner.id,
+          body: "root",
+          clientIdempotencyKey: crypto.randomUUID(),
+        },
+        {
+          workspaceId: workspace.id,
+          channelId: publicChannel.id,
+          sequence: 2n,
+          authorType: "human",
+          authorUserId: owner.id,
+          body: "reply",
+          clientIdempotencyKey: crypto.randomUUID(),
+        },
+        {
+          workspaceId: workspace.id,
+          channelId: publicChannel.id,
+          sequence: 3n,
+          authorType: "human",
+          authorUserId: owner.id,
+          body: "",
+          status: "deleted",
+          clientIdempotencyKey: crypto.randomUUID(),
+        },
+        {
+          workspaceId: workspace.id,
+          channelId: publicChannel.id,
+          sequence: 4n,
+          authorType: "human",
+          authorUserId: owner.id,
+          body: "new root",
+          clientIdempotencyKey: crypto.randomUUID(),
+        },
+      ])
+      .returning();
+    await db
+      .update(chatMessages)
+      .set({ parentMessageId: root.id })
+      .where(
+        and(
+          eq(chatMessages.workspaceId, workspace.id),
+          eq(chatMessages.id, reply.id),
+        ),
+      );
+    await db.insert(chatMessageMentions).values({
+      workspaceId: workspace.id,
+      messageId: secondRoot.id,
+      userId: reader.id,
+    });
+    await db.insert(chatMessages).values([
+      {
+        workspaceId: workspace.id,
+        channelId: archived.id,
+        sequence: 1n,
+        authorType: "human",
+        authorUserId: owner.id,
+        body: "archived",
+        clientIdempotencyKey: crypto.randomUUID(),
+      },
+      {
+        workspaceId: workspace.id,
+        channelId: hidden.id,
+        sequence: 1n,
+        authorType: "human",
+        authorUserId: owner.id,
+        body: "secret",
+        clientIdempotencyKey: crypto.randomUUID(),
+      },
+    ]);
+    try {
+      const page = await chatHistory(db, { ...scope, limit: 1 });
+      expect(page.messages.map((m) => m.sequence)).toEqual(["4"]);
+      expect(page.next).toBe("4");
+      expect(page.hasMore).toBe(true);
+      const older = await chatHistory(db, { ...scope, before: "4", limit: 2 });
+      expect(older.messages.map((m) => m.sequence)).toEqual(["3", "1"]);
+      expect(older.messages[1]).toMatchObject({
+        replyCount: 1,
+        latestReplySequence: "2",
+      });
+      const thread = await chatThreadHistory(db, {
+        ...scope,
+        rootMessageId: root.id,
+        limit: 1,
+      });
+      expect(thread.root).toMatchObject({ sequence: "1", replyCount: 1 });
+      expect(thread.replies.map((m) => m.sequence)).toEqual(["2"]);
+      const catchUp = await chatCatchUp(db, { ...scope, after: "1", limit: 2 });
+      expect(catchUp.messages.map((m) => [m.sequence, m.status])).toEqual([
+        ["2", "sent"],
+        ["3", "deleted"],
+      ]);
+      expect(catchUp).toMatchObject({ next: "3", hasMore: true, through: "4" });
+      const readCursorBefore = await db
+        .select()
+        .from(chatReadCursors)
+        .where(eq(chatReadCursors.workspaceId, workspace.id));
+      const readOutboxBefore = await db
+        .select()
+        .from(chatOutbox)
+        .where(eq(chatOutbox.workspaceId, workspace.id));
+      const bounded = await chatCatchUp(db, {
+        ...scope,
+        after: "1",
+        through: catchUp.through,
+        limit: 1,
+      });
+      expect(bounded).toMatchObject({ next: "2", through: "4", hasMore: true });
+      const tail = await chatCatchUp(db, {
+        ...scope,
+        after: "3",
+        through: bounded.through,
+        limit: 5,
+      });
+      expect(tail).toMatchObject({ next: "4", through: "4", hasMore: false });
+      await expect(
+        chatCatchUp(db, { ...scope, after: "0", through: "5", limit: 2 }),
+      ).rejects.toBeInstanceOf(ChatReadError);
+      await expect(
+        chatCatchUp(db, { ...scope, after: "5", limit: 2 }),
+      ).rejects.toBeInstanceOf(ChatReadError);
+      expect(
+        (
+          await chatCatchUp(db, { ...scope, after: "3", limit: 2 })
+        ).messages.map((m) => m.sequence),
+      ).toEqual(["4"]);
+      expect(
+        await db
+          .select()
+          .from(chatReadCursors)
+          .where(eq(chatReadCursors.workspaceId, workspace.id)),
+      ).toEqual(readCursorBefore);
+      expect(
+        await db
+          .select()
+          .from(chatOutbox)
+          .where(eq(chatOutbox.workspaceId, workspace.id)),
+      ).toEqual(readOutboxBefore);
+      const summary = await chatUnreadSummary(db, {
+        workspaceId: workspace.id,
+        actorUserId: reader.id,
+      });
+      expect(summary.channels).toEqual(
+        expect.arrayContaining([
+          {
+            channelId: publicChannel.id,
+            archived: false,
+            readSequence: "0",
+            unreadCount: 4,
+            mentionCount: 1,
+          },
+          {
+            channelId: archived.id,
+            archived: true,
+            readSequence: "0",
+            unreadCount: 1,
+            mentionCount: 0,
+          },
+        ]),
+      );
+      expect(summary.channels).toHaveLength(2);
+      expect(
+        (
+          await chatHistory(db, { ...scope, channelId: archived.id, limit: 10 })
+        ).messages.map((m) => m.body),
+      ).toEqual(["archived"]);
+      await expect(
+        chatHistory(db, { ...scope, channelId: hidden.id, limit: 10 }),
+      ).rejects.toBeInstanceOf(ChatReadError);
+      await expect(
+        chatMarkChannelRead(db, { ...scope, sequence: "5" }),
+      ).rejects.toBeInstanceOf(ChatReadError);
+      expect(
+        await chatMarkChannelRead(db, {
+          ...scope,
+          channelId: archived.id,
+          sequence: "1",
+        }),
+      ).toMatchObject({ sequence: "1", changed: true });
+      expect(
+        (
+          await chatUnreadSummary(db, {
+            workspaceId: workspace.id,
+            actorUserId: reader.id,
+          })
+        ).channels.find((channel) => channel.channelId === archived.id),
+      ).toMatchObject({ readSequence: "1", unreadCount: 0 });
+      const results = await Promise.allSettled(
+        ["2", "4", "3"].map((sequence) =>
+          chatMarkChannelRead(db, { ...scope, sequence }),
+        ),
+      );
+      const [initialCursor] = await db
+        .select()
+        .from(chatReadCursors)
+        .where(
+          and(
+            eq(chatReadCursors.workspaceId, workspace.id),
+            eq(chatReadCursors.channelId, publicChannel.id),
+            eq(chatReadCursors.userId, reader.id),
+          ),
+        );
+      expect(initialCursor.sequence).toBeGreaterThanOrEqual(2n);
+      expect(initialCursor.sequence).toBeLessThanOrEqual(4n);
+      expect(
+        results
+          .filter((result) => result.status === "rejected")
+          .every(
+            (result) =>
+              result.status === "rejected" && result.reason.code === "40001",
+          ),
+      ).toBe(true);
+      // The bounded transaction retry cannot lower a concurrently advanced cursor.
+      const retry = await chatMarkChannelRead(db, { ...scope, sequence: "4" });
+      expect(retry.sequence).toBe("4");
+      expect(
+        await chatMarkChannelRead(db, { ...scope, sequence: "2" }),
+      ).toEqual({ sequence: "4", changed: false });
+      const events = await db
+        .select()
+        .from(chatOutbox)
+        .where(
+          and(
+            eq(chatOutbox.workspaceId, workspace.id),
+            eq(chatOutbox.channelId, publicChannel.id),
+            eq(chatOutbox.eventType, "chat.read_state.changed.v1"),
+          ),
+        );
+      expect(events.length).toBe(
+        results.filter(
+          (result) => result.status === "fulfilled" && result.value.changed,
+        ).length + Number(retry.changed),
+      );
+      const readSummary = await chatUnreadSummary(db, {
+        workspaceId: workspace.id,
+        actorUserId: reader.id,
+      });
+      expect(
+        readSummary.channels.find(
+          (channel) => channel.channelId === publicChannel.id,
+        ),
+      ).toMatchObject({ readSequence: "4", unreadCount: 0, mentionCount: 0 });
+    } finally {
+      await db.delete(workspaces).where(eq(workspaces.id, workspace.id));
+      await db
+        .delete(users)
+        .where(sql`${users.id} in (${owner.id}, ${reader.id}, ${outsider.id})`);
+    }
+  }, 20000);
+});

packages/database/src/chat/read-history.test.ts

@@ -0,0 +1,277 @@
+import { describe, expect, it, vi } from "vitest";
+import type { db as defaultDb } from "../client";
+import {
+  chatCatchUp,
+  chatMutationCatchUp,
+  chatHistory,
+  chatThreadHistory,
+  chatMarkChannelRead,
+  ChatReadError,
+} from "./read-history";
+
+import { chatCursorSchema, chatSequenceSchema } from "./read-history";
+
+// Sequence/cursor inputs are canonical decimal bigint strings, not JS numbers or aliases.
+describe("chat read sequence validation", () => {
+  it("accepts positive sequences and the zero reconnect cursor", () => {
+    expect(chatSequenceSchema.parse("9223372036854775807")).toBe(
+      "9223372036854775807",
+    );
+    expect(chatCursorSchema.parse("0")).toBe("0");
+  });
+
+  it.each([
+    "",
+    "00",
+    "01",
+    "-1",
+    "+1",
+    "1.0",
+    " 1",
+    "1e3",
+    "9223372036854775808",
+    1,
+  ])("rejects malformed cursor %s", (value) => {
+    expect(chatCursorSchema.safeParse(value).success).toBe(false);
+  });
+
+  it("requires a positive sequence for a viewed message", () => {
+    expect(chatSequenceSchema.safeParse("0").success).toBe(false);
+  });
+});
+
+const scope = {
+  workspaceId: "11111111-1111-4111-8111-111111111111",
+  channelId: "33333333-3333-4333-8333-333333333333",
+  actorUserId: "22222222-2222-4222-8222-222222222222",
+};
+
+describe("mutation recovery projections", () => {
+  it("deduplicates repeated mutations and redacts the current tombstone", async () => {
+    const messageId = "77777777-7777-4777-8777-777777777777";
+    const results = [
+      [{ role: "member" }],
+      [{ id: scope.channelId, workspaceId: scope.workspaceId, kind: "public", ownerUserId: scope.actorUserId, archivedAt: null, nextMutationSequence: 3n }],
+      [],
+      [{ messageId, mutationSequence: 1n, revision: 2, kind: "edited" }, { messageId, mutationSequence: 2n, revision: 3, kind: "deleted" }],
+      [{ id: messageId, workspaceId: scope.workspaceId, channelId: scope.channelId, sequence: 1n, parentMessageId: null, body: "secret", status: "deleted", revision: 3, deletedAt: new Date(), createdAt: new Date(), authorUserId: scope.actorUserId, sendIntentFingerprint: "private" }],
+    ];
+    const tx = {
+      execute: vi.fn(),
+      select: () => {
+        const rows = results.shift();
+        const query = {
+          from: () => query, where: () => query, orderBy: () => query,
+          limit: () => Promise.resolve(rows),
+          then: (resolve: (value: unknown) => unknown) => Promise.resolve(rows).then(resolve),
+        };
+        return query;
+      },
+    };
+    const db = { transaction: (fn: (value: typeof tx) => unknown) => fn(tx) } as unknown as typeof defaultDb;
+    const result = await chatMutationCatchUp(db, { ...scope, after: "0", limit: 10 });
+    expect(result.mutations).toHaveLength(2);
+    expect(result.messages).toHaveLength(1);
+    expect(result.messages[0]).toMatchObject({ id: messageId, revision: 3, body: "", status: "deleted" });
+    expect(JSON.stringify(result)).not.toContain("secret");
+    expect(JSON.stringify(result)).not.toContain("private");
+    expect(result).toMatchObject({ next: "2", through: "2", hasMore: false });
+  });
+});
+
+describe("mutation recovery input", () => {
+  const db = { transaction: vi.fn() } as unknown as typeof defaultDb;
+  it("rejects noncanonical cursors and inverted fixed high-water before querying", async () => {
+    await expect(
+      chatMutationCatchUp(db, { ...scope, after: "01", limit: 10 }),
+    ).rejects.toThrow();
+    await expect(
+      chatMutationCatchUp(db, {
+        ...scope,
+        after: "3",
+        through: "2",
+        limit: 10,
+      }),
+    ).rejects.toThrow();
+    expect(db.transaction).not.toHaveBeenCalled();
+  });
+});
+
+describe("mark-read transaction retries", () => {
+  const error = (code: string) =>
+    Object.assign(new Error("transaction failed"), { code });
+  it.each(["40001", "40P01"])(
+    "retries a failed transaction for %s",
+    async (code) => {
+      const transaction = vi
+        .fn()
+        .mockRejectedValueOnce(error(code))
+        .mockResolvedValueOnce({ changed: true, sequence: "9007199254740993" });
+      const result = await chatMarkChannelRead(
+        { transaction } as unknown as typeof defaultDb,
+        { ...scope, sequence: "9007199254740993" },
+      );
+      expect(result).toEqual({ changed: true, sequence: "9007199254740993" });
+      expect(transaction).toHaveBeenCalledTimes(2);
+    },
+  );
+  it("does not retry unrelated errors or safe authorization failures", async () => {
+    for (const failure of [error("23505"), new ChatReadError()]) {
+      const transaction = vi.fn().mockRejectedValue(failure);
+      await expect(
+        chatMarkChannelRead({ transaction } as unknown as typeof defaultDb, {
+          ...scope,
+          sequence: "1",
+        }),
+      ).rejects.toBe(failure);
+      expect(transaction).toHaveBeenCalledTimes(1);
+    }
+  });
+  it("exhausts bounded retries without masking the database error", async () => {
+    const failure = error("40001");
+    const transaction = vi.fn().mockRejectedValue(failure);
+    await expect(
+      chatMarkChannelRead({ transaction } as unknown as typeof defaultDb, {
+        ...scope,
+        sequence: "1",
+      }),
+    ).rejects.toBe(failure);
+    expect(transaction.mock.calls.length).toBeGreaterThan(1);
+    expect(transaction.mock.calls.length).toBeLessThanOrEqual(5);
+  });
+});
+
+describe("read-history transaction retries", () => {
+  const retryable = (code: string) =>
+    new Error("transaction failed", {
+      cause: new Error("driver wrapper", {
+        cause: Object.assign(new Error("postgres failure"), { code }),
+      }),
+    });
+  const requests = [
+    {
+      name: "history",
+      run: (db: typeof defaultDb) => chatHistory(db, { ...scope, limit: 10 }),
+    },
+    {
+      name: "thread history",
+      run: (db: typeof defaultDb) =>
+        chatThreadHistory(db, {
+          ...scope,
+          rootMessageId: "44444444-4444-4444-8444-444444444444",
+          limit: 10,
+        }),
+    },
+    {
+      name: "catch-up",
+      run: (db: typeof defaultDb) =>
+        chatCatchUp(db, { ...scope, after: "0", limit: 10 }),
+    },
+    {
+      name: "mark-read",
+      run: (db: typeof defaultDb) =>
+        chatMarkChannelRead(db, { ...scope, sequence: "1" }),
+    },
+  ];
+
+  it.each(requests)(
+    "$name starts a fresh transaction on a cause-wrapped serialization failure",
+    async ({ run }) => {
+      const callbacks: unknown[] = [];
+      const transactions: unknown[] = [];
+      const failure = retryable("40001");
+      const transaction = vi.fn(
+        async (callback: (tx: unknown) => Promise<unknown>) => {
+          callbacks.push(callback);
+          const tx = {
+            execute: vi.fn().mockRejectedValue(failure),
+            select: () => ({
+              from: () => ({
+                where: () => ({ limit: () => Promise.reject(failure) }),
+              }),
+            }),
+          };
+          transactions.push(tx);
+          return callback(tx);
+        },
+      );
+      await expect(
+        run({ transaction } as unknown as typeof defaultDb),
+      ).rejects.toBe(failure);
+      expect(transaction).toHaveBeenCalledTimes(3);
+      expect(new Set(callbacks).size).toBe(3);
+      expect(new Set(transactions).size).toBe(3);
+    },
+  );
+
+  it.each(requests)(
+    "$name retries wrapped deadlocks but not unrelated errors",
+    async ({ run }) => {
+      const deadlock = retryable("40P01");
+      const transaction = vi
+        .fn()
+        .mockRejectedValueOnce(deadlock)
+        .mockResolvedValueOnce("committed");
+      await expect(
+        run({ transaction } as unknown as typeof defaultDb),
+      ).resolves.toBe("committed");
+      expect(transaction).toHaveBeenCalledTimes(2);
+
+      const unrelated = retryable("23505");
+      const noRetry = vi.fn().mockRejectedValue(unrelated);
+      await expect(
+        run({ transaction: noRetry } as unknown as typeof defaultDb),
+      ).rejects.toBe(unrelated);
+      expect(noRetry).toHaveBeenCalledTimes(1);
+    },
+  );
+
+  it("does not retry individual statements outside the mark-read transaction", async () => {
+    const failure = retryable("40001");
+    const externalEffects: number[] = [];
+    const transaction = vi.fn(
+      async (callback: (tx: unknown) => Promise<unknown>) => {
+        const attempt = transaction.mock.calls.length;
+        return callback({
+          select: () => {
+            externalEffects.push(attempt);
+            return {
+              from: () => ({
+                where: () => ({ limit: () => Promise.reject(failure) }),
+              }),
+            };
+          },
+        });
+      },
+    );
+    await expect(
+      chatMarkChannelRead({ transaction } as unknown as typeof defaultDb, {
+        ...scope,
+        sequence: "1",
+      }),
+    ).rejects.toBe(failure);
+    expect(externalEffects).toEqual([1, 2, 3]);
+    // External side effects survive retries; cursor changes and outbox insertion must
+    // remain inside the DB transaction to benefit from rollback.
+  });
+});
+
+describe("catch-up bounds", () => {
+  it.each([
+    { after: "9223372036854775808", limit: 1 },
+    { after: "0", through: "01", limit: 1 },
+    { after: "2", through: "1", limit: 1 },
+  ])(
+    "rejects malformed bounds before starting a transaction %#",
+    async (bounds) => {
+      const transaction = vi.fn();
+      await expect(
+        chatCatchUp({ transaction } as unknown as typeof defaultDb, {
+          ...scope,
+          ...bounds,
+        }),
+      ).rejects.toThrow();
+      expect(transaction).not.toHaveBeenCalled();
+    },
+  );
+});

packages/database/src/chat/read-history.ts

@@ -0,0 +1,617 @@
+import {
+  canPerformChatAction,
+  chatCursorSchema,
+  chatSequenceSchema,
+} from "@tasks/shared";
+import { and, asc, desc, eq, gt, inArray, isNull, lt, sql } from "drizzle-orm";
+import { z } from "zod";
+
+import { db as defaultDb } from "../client";
+import {
+  chatChannelMembers,
+  chatChannels,
+  chatMessageMentions,
+  chatMessageMutations,
+  chatMessages,
+  chatOutbox,
+  chatReadCursors,
+  workspaceMembers,
+} from "../schema";
+
+export { chatCursorSchema, chatSequenceSchema } from "@tasks/shared";
+export type ChatSequence = z.infer<typeof chatSequenceSchema>;
+export type ChatCursor = z.infer<typeof chatCursorSchema>;
+
+export const chatPageSizeSchema = z.number().int().min(1).max(100);
+export const chatCatchUpSizeSchema = z.number().int().min(1).max(500);
+const uuid = z.string().uuid();
+const scopeSchema = z
+  .object({ workspaceId: uuid, channelId: uuid, actorUserId: uuid })
+  .strict();
+const historySchema = scopeSchema
+  .extend({ before: chatSequenceSchema.optional(), limit: chatPageSizeSchema })
+  .strict();
+const threadSchema = historySchema.extend({ rootMessageId: uuid }).strict();
+const catchUpSchema = scopeSchema
+  .extend({
+    after: chatCursorSchema,
+    through: chatCursorSchema.optional(),
+    limit: chatCatchUpSizeSchema,
+  })
+  .strict()
+  .refine(
+    (input) =>
+      input.through === undefined ||
+      BigInt(input.after) <= BigInt(input.through),
+    { message: "After exceeds through", path: ["through"] },
+  );
+const mutationsSchema = scopeSchema
+  .extend({
+    after: chatCursorSchema,
+    through: chatCursorSchema.optional(),
+    limit: chatCatchUpSizeSchema,
+  })
+  .strict()
+  .refine(
+    (input) =>
+      input.through === undefined ||
+      BigInt(input.after) <= BigInt(input.through),
+    { path: ["through"] },
+  );
+const markReadSchema = scopeSchema
+  .extend({ sequence: chatSequenceSchema })
+  .strict();
+const summarySchema = z
+  .object({ workspaceId: uuid, actorUserId: uuid })
+  .strict();
+type Scope = z.infer<typeof scopeSchema>;
+type Tx = Parameters<Parameters<typeof defaultDb.transaction>[0]>[0];
+export class ChatReadError extends Error {
+  constructor() {
+    super("Chat channel not found");
+  }
+}
+
+// The same workspace+channel+actor policy guards every read surface, including archived channels.
+async function authorize(tx: Tx, scope: Scope) {
+  const [actor] = await tx
+    .select({ role: workspaceMembers.role })
+    .from(workspaceMembers)
+    .where(
+      and(
+        eq(workspaceMembers.workspaceId, scope.workspaceId),
+        eq(workspaceMembers.userId, scope.actorUserId),
+      ),
+    )
+    .limit(1);
+  const [channel] = await tx
+    .select()
+    .from(chatChannels)
+    .where(
+      and(
+        eq(chatChannels.workspaceId, scope.workspaceId),
+        eq(chatChannels.id, scope.channelId),
+      ),
+    )
+    .limit(1);
+  if (!actor || !channel) throw new ChatReadError();
+  const [member] = await tx
+    .select({ userId: chatChannelMembers.userId })
+    .from(chatChannelMembers)
+    .where(
+      and(
+        eq(chatChannelMembers.workspaceId, scope.workspaceId),
+        eq(chatChannelMembers.channelId, scope.channelId),
+        eq(chatChannelMembers.userId, scope.actorUserId),
+      ),
+    )
+    .limit(1);
+  if (
+    !canPerformChatAction({
+      action: "read",
+      actor: {
+        userId: scope.actorUserId,
+        workspaceId: scope.workspaceId,
+        activeWorkspaceMember: true,
+        role: actor.role,
+      },
+      context: { type: "workspace", workspaceId: scope.workspaceId },
+      channel: {
+        id: channel.id,
+        workspaceId: channel.workspaceId,
+        kind: channel.kind,
+        archived: channel.archivedAt !== null,
+        activeMember: !!member,
+        ownerUserId: channel.ownerUserId,
+      },
+    })
+  )
+    throw new ChatReadError();
+  return channel;
+}
+
+type Message = typeof chatMessages.$inferSelect;
+function serialize(message: Message) {
+  // Explicit projection prevents internal send-intent fingerprints (and future storage keys)
+  // from leaking through history and catch-up DTOs.
+  return {
+    id: message.id,
+    workspaceId: message.workspaceId,
+    channelId: message.channelId,
+    sequence: message.sequence.toString(),
+    parentMessageId: message.parentMessageId,
+    authorType: message.authorType,
+    authorUserId: message.authorUserId,
+    authorAgentId: message.authorAgentId,
+    body: message.status === "deleted" ? "" : message.body,
+    status: message.status,
+    revision: message.revision,
+    createdAt: message.createdAt,
+    editedAt: message.editedAt,
+    deletedAt: message.deletedAt,
+  };
+}
+async function replyMetadata(tx: Tx, scope: Scope, roots: Message[]) {
+  if (!roots.length)
+    return new Map<
+      string,
+      { replyCount: number; latestReplySequence: string | null }
+    >();
+  const rows = await tx
+    .select({
+      parentMessageId: chatMessages.parentMessageId,
+      count: sql<number>`count(*)::int`,
+      latest: sql<bigint>`max(${chatMessages.sequence})`,
+    })
+    .from(chatMessages)
+    .where(
+      and(
+        eq(chatMessages.workspaceId, scope.workspaceId),
+        eq(chatMessages.channelId, scope.channelId),
+        inArray(
+          chatMessages.parentMessageId,
+          roots.map((root) => root.id),
+        ),
+      ),
+    )
+    .groupBy(chatMessages.parentMessageId);
+  return new Map(
+    rows.map((row) => [
+      row.parentMessageId!,
+      { replyCount: row.count, latestReplySequence: row.latest.toString() },
+    ]),
+  );
+}
+
+// Backward root-page snapshot: limit+1 detects older roots, while reply metadata uses the same snapshot.
+export async function chatHistory(
+  db: typeof defaultDb,
+  raw: z.input<typeof historySchema>,
+) {
+  const input = historySchema.parse(raw);
+  return retryTransaction(
+    db,
+    async (tx) => {
+      await tx.execute(sql`set transaction read only`);
+      await authorize(tx, input);
+      const rows = await tx
+        .select()
+        .from(chatMessages)
+        .where(
+          and(
+            eq(chatMessages.workspaceId, input.workspaceId),
+            eq(chatMessages.channelId, input.channelId),
+            isNull(chatMessages.parentMessageId),
+            input.before
+              ? lt(chatMessages.sequence, BigInt(input.before))
+              : undefined,
+          ),
+        )
+        .orderBy(desc(chatMessages.sequence))
+        .limit(input.limit + 1);
+      const hasMore = rows.length > input.limit;
+      const page = rows.slice(0, input.limit);
+      const metadata = await replyMetadata(tx, input, page);
+      return {
+        messages: page.map((message) => ({
+          ...serialize(message),
+          replyCount: metadata.get(message.id)?.replyCount ?? 0,
+          latestReplySequence:
+            metadata.get(message.id)?.latestReplySequence ?? null,
+        })),
+        next: hasMore ? page.at(-1)!.sequence.toString() : null,
+        hasMore,
+      };
+    },
+    { isolationLevel: "repeatable read" },
+  );
+}
+
+// Thread pages include the root and only its direct replies; a reply is never a root.
+export async function chatThreadHistory(
+  db: typeof defaultDb,
+  raw: z.input<typeof threadSchema>,
+) {
+  const input = threadSchema.parse(raw);
+  return retryTransaction(
+    db,
+    async (tx) => {
+      await tx.execute(sql`set transaction read only`);
+      await authorize(tx, input);
+      const [root] = await tx
+        .select()
+        .from(chatMessages)
+        .where(
+          and(
+            eq(chatMessages.workspaceId, input.workspaceId),
+            eq(chatMessages.channelId, input.channelId),
+            eq(chatMessages.id, input.rootMessageId),
+            isNull(chatMessages.parentMessageId),
+          ),
+        )
+        .limit(1);
+      if (!root) throw new ChatReadError();
+      const rows = await tx
+        .select()
+        .from(chatMessages)
+        .where(
+          and(
+            eq(chatMessages.workspaceId, input.workspaceId),
+            eq(chatMessages.channelId, input.channelId),
+            eq(chatMessages.parentMessageId, root.id),
+            input.before
+              ? lt(chatMessages.sequence, BigInt(input.before))
+              : undefined,
+          ),
+        )
+        .orderBy(desc(chatMessages.sequence))
+        .limit(input.limit + 1);
+      const hasMore = rows.length > input.limit;
+      const page = rows.slice(0, input.limit);
+      const metadata = await replyMetadata(tx, input, [root]);
+      return {
+        root: {
+          ...serialize(root),
+          replyCount: metadata.get(root.id)?.replyCount ?? 0,
+          latestReplySequence:
+            metadata.get(root.id)?.latestReplySequence ?? null,
+        },
+        replies: page.map(serialize),
+        next: hasMore ? page.at(-1)!.sequence.toString() : null,
+        hasMore,
+      };
+    },
+    { isolationLevel: "repeatable read" },
+  );
+}
+
+// Bounded reconnect scan includes replies and tombstones in channel sequence order.
+export async function chatCatchUp(
+  db: typeof defaultDb,
+  raw: z.input<typeof catchUpSchema>,
+) {
+  const input = catchUpSchema.parse(raw);
+  return retryTransaction(
+    db,
+    async (tx) => {
+      await tx.execute(sql`set transaction read only`);
+      const channel = await authorize(tx, input);
+      const highWater = channel.nextSequence - 1n;
+      if (
+        BigInt(input.after) > highWater ||
+        (input.through !== undefined && BigInt(input.through) > highWater)
+      )
+        throw new ChatReadError();
+      const through = input.through ?? highWater.toString();
+      const rows = await tx
+        .select()
+        .from(chatMessages)
+        .where(
+          and(
+            eq(chatMessages.workspaceId, input.workspaceId),
+            eq(chatMessages.channelId, input.channelId),
+            gt(chatMessages.sequence, BigInt(input.after)),
+            sql`${chatMessages.sequence} <= ${BigInt(through)}`,
+          ),
+        )
+        .orderBy(asc(chatMessages.sequence))
+        .limit(input.limit + 1);
+      const hasMore = rows.length > input.limit;
+      const page = rows.slice(0, input.limit);
+      const roots = page.filter((message) => message.parentMessageId === null);
+      const metadata = await replyMetadata(tx, input, roots);
+      return {
+        messages: page.map((message) => ({
+          ...serialize(message),
+          replyCount:
+            message.parentMessageId === null
+              ? (metadata.get(message.id)?.replyCount ?? 0)
+              : 0,
+          latestReplySequence:
+            message.parentMessageId === null
+              ? (metadata.get(message.id)?.latestReplySequence ?? null)
+              : null,
+        })),
+        next: page.at(-1)?.sequence.toString() ?? input.after,
+        through,
+        hasMore,
+      };
+    },
+    { isolationLevel: "repeatable read" },
+  );
+}
+
+// Durable metadata-only mutation scan: fixed high-water survives pagination and reconnects.
+export async function chatMutationCatchUp(
+  db: typeof defaultDb,
+  raw: z.input<typeof mutationsSchema>,
+) {
+  const input = mutationsSchema.parse(raw);
+  return retryTransaction(
+    db,
+    async (tx) => {
+      await tx.execute(sql`set transaction read only`);
+      const channel = await authorize(tx, input);
+      const highWater = channel.nextMutationSequence - 1n;
+      if (
+        BigInt(input.after) > highWater ||
+        (input.through !== undefined && BigInt(input.through) > highWater)
+      )
+        throw new ChatReadError();
+      const through = input.through ?? highWater.toString();
+      const rows = await tx
+        .select({
+          messageId: chatMessageMutations.messageId,
+          mutationSequence: chatMessageMutations.mutationSequence,
+          revision: chatMessageMutations.revision,
+          kind: chatMessageMutations.kind,
+        })
+        .from(chatMessageMutations)
+        .where(
+          and(
+            eq(chatMessageMutations.workspaceId, input.workspaceId),
+            eq(chatMessageMutations.channelId, input.channelId),
+            gt(chatMessageMutations.mutationSequence, BigInt(input.after)),
+            sql`${chatMessageMutations.mutationSequence} <= ${BigInt(through)}`,
+          ),
+        )
+        .orderBy(asc(chatMessageMutations.mutationSequence))
+        .limit(input.limit + 1);
+      const page = rows.slice(0, input.limit);
+      const messageIds = [...new Set(page.map((row) => row.messageId))];
+      const messagesRows = messageIds.length ? await tx
+        .select()
+        .from(chatMessages)
+        .where(
+          and(
+            eq(chatMessages.workspaceId, input.workspaceId),
+            eq(chatMessages.channelId, input.channelId),
+            inArray(chatMessages.id, messageIds),
+          ),
+        ) : [];
+      const messagesMap = new Map(messagesRows.map((m) => [m.id, serialize(m)]));
+      if (messagesMap.size !== messageIds.length) {
+        throw new ChatReadError();
+      }
+      return {
+        mutations: page.map((row) => ({
+          messageId: row.messageId,
+          mutationSequence: row.mutationSequence.toString(),
+          revision: row.revision,
+          kind: row.kind,
+        })),
+        messages: messageIds.map((id) => messagesMap.get(id)!),
+        next: page.at(-1)?.mutationSequence.toString() ?? input.after,
+        through,
+        hasMore: rows.length > input.limit,
+      };
+    },
+    { isolationLevel: "repeatable read" },
+  );
+}
+
+// Mark-read validates that the claimed viewed sequence actually exists, then advances and emits
+// exactly one invalidation in the same transaction. A no-op never changes the cursor or emits.
+export async function chatMarkChannelRead(
+  db: typeof defaultDb,
+  raw: z.input<typeof markReadSchema>,
+) {
+  const input = markReadSchema.parse(raw);
+  return retryTransaction(
+    db,
+    async (tx) => {
+      await authorize(tx, input);
+      const [viewed] = await tx
+        .select({ id: chatMessages.id })
+        .from(chatMessages)
+        .where(
+          and(
+            eq(chatMessages.workspaceId, input.workspaceId),
+            eq(chatMessages.channelId, input.channelId),
+            eq(chatMessages.sequence, BigInt(input.sequence)),
+          ),
+        )
+        .limit(1);
+      if (!viewed) throw new ChatReadError();
+      const [current] = await tx
+        .select({ sequence: chatReadCursors.sequence })
+        .from(chatReadCursors)
+        .where(
+          and(
+            eq(chatReadCursors.workspaceId, input.workspaceId),
+            eq(chatReadCursors.channelId, input.channelId),
+            eq(chatReadCursors.userId, input.actorUserId),
+          ),
+        )
+        .limit(1);
+      if (current && current.sequence >= BigInt(input.sequence))
+        return { sequence: current.sequence.toString(), changed: false };
+      const [updated] = await tx
+        .insert(chatReadCursors)
+        .values({
+          workspaceId: input.workspaceId,
+          channelId: input.channelId,
+          userId: input.actorUserId,
+          sequence: BigInt(input.sequence),
+        })
+        .onConflictDoUpdate({
+          target: [
+            chatReadCursors.workspaceId,
+            chatReadCursors.channelId,
+            chatReadCursors.userId,
+          ],
+          set: {
+            sequence: sql`greatest(${chatReadCursors.sequence}, excluded.sequence)`,
+            updatedAt: sql`now()`,
+          },
+          setWhere: lt(chatReadCursors.sequence, BigInt(input.sequence)),
+        })
+        .returning({ sequence: chatReadCursors.sequence });
+      if (!updated)
+        return {
+          sequence: current?.sequence.toString() ?? "0",
+          changed: false,
+        };
+      await tx.insert(chatOutbox).values({
+        workspaceId: input.workspaceId,
+        channelId: input.channelId,
+        eventType: "chat.read_state.changed.v1",
+        eventVersion: 1,
+        dedupeKey: `read:${input.channelId}:${input.actorUserId}:${updated.sequence}`,
+        payload: {
+          eventType: "chat.read_state.changed.v1",
+          eventVersion: 1,
+          workspaceId: input.workspaceId,
+          channelId: input.channelId,
+          userId: input.actorUserId,
+          sequence: updated.sequence.toString(),
+        },
+      });
+      return { sequence: updated.sequence.toString(), changed: true };
+    },
+    { isolationLevel: "repeatable read" },
+  );
+}
+
+// Retry the transaction as a unit so a failed attempt cannot commit a cursor without its outbox event.
+async function retryTransaction<T>(
+  db: typeof defaultDb,
+  callback: (tx: Tx) => Promise<T>,
+  config: Parameters<typeof defaultDb.transaction>[1],
+): Promise<T> {
+  for (let attempt = 0; attempt < 3; attempt++) {
+    try {
+      return await db.transaction((tx) => callback(tx), config);
+    } catch (error) {
+      if (attempt === 2 || !isRetryableTransactionError(error)) throw error;
+    }
+  }
+  throw new Error("Unreachable transaction retry state");
+}
+
+function isRetryableTransactionError(error: unknown): boolean {
+  if (typeof error !== "object" || error === null) return false;
+  const candidate = error as { code?: unknown; cause?: unknown };
+  return (
+    candidate.code === "40001" ||
+    candidate.code === "40P01" ||
+    (candidate.cause !== undefined &&
+      isRetryableTransactionError(candidate.cause))
+  );
+}
+
+// Unread aggregation considers every authorized channel, including archived ones, but never
+// grants private/direct visibility based on the actor's workspace role alone.
+export async function chatUnreadSummary(
+  db: typeof defaultDb,
+  raw: z.input<typeof summarySchema>,
+) {
+  const input = summarySchema.parse(raw);
+  return db.transaction(
+    async (tx) => {
+      await tx.execute(sql`set transaction read only`);
+      const [actor] = await tx
+        .select({ userId: workspaceMembers.userId })
+        .from(workspaceMembers)
+        .where(
+          and(
+            eq(workspaceMembers.workspaceId, input.workspaceId),
+            eq(workspaceMembers.userId, input.actorUserId),
+          ),
+        )
+        .limit(1);
+      if (!actor) throw new ChatReadError();
+      const channels = await tx
+        .select({ channel: chatChannels })
+        .from(chatChannels)
+        .leftJoin(
+          chatChannelMembers,
+          and(
+            eq(chatChannelMembers.workspaceId, chatChannels.workspaceId),
+            eq(chatChannelMembers.channelId, chatChannels.id),
+            eq(chatChannelMembers.userId, input.actorUserId),
+          ),
+        )
+        .where(
+          and(
+            eq(chatChannels.workspaceId, input.workspaceId),
+            sql`(${chatChannels.kind} = 'public' or ${chatChannelMembers.userId} is not null)`,
+          ),
+        );
+      // A single set-based count query keeps the summary bounded by channels, not messages.
+      const counts = await tx
+        .select({
+          channelId: chatChannels.id,
+          readSequence: sql<bigint>`coalesce(${chatReadCursors.sequence}, 0)`,
+          unreadCount: sql<number>`count(${chatMessages.id})::int`,
+          mentionCount: sql<number>`count(${chatMessageMentions.messageId})::int`,
+        })
+        .from(chatChannels)
+        .leftJoin(
+          chatReadCursors,
+          and(
+            eq(chatReadCursors.workspaceId, chatChannels.workspaceId),
+            eq(chatReadCursors.channelId, chatChannels.id),
+            eq(chatReadCursors.userId, input.actorUserId),
+          ),
+        )
+        .leftJoin(
+          chatMessages,
+          and(
+            eq(chatMessages.workspaceId, chatChannels.workspaceId),
+            eq(chatMessages.channelId, chatChannels.id),
+            sql`${chatMessages.sequence} > coalesce(${chatReadCursors.sequence}, 0)`,
+          ),
+        )
+        .leftJoin(
+          chatMessageMentions,
+          and(
+            eq(chatMessageMentions.workspaceId, chatMessages.workspaceId),
+            eq(chatMessageMentions.messageId, chatMessages.id),
+            eq(chatMessageMentions.userId, input.actorUserId),
+          ),
+        )
+        .where(
+          and(
+            eq(chatChannels.workspaceId, input.workspaceId),
+            inArray(
+              chatChannels.id,
+              channels.map(({ channel }) => channel.id),
+            ),
+          ),
+        )
+        .groupBy(chatChannels.id, chatReadCursors.sequence);
+      const byId = new Map(counts.map((count) => [count.channelId, count]));
+      return {
+        channels: channels.map(({ channel }) => ({
+          channelId: channel.id,
+          archived: channel.archivedAt !== null,
+          readSequence: byId.get(channel.id)?.readSequence.toString() ?? "0",
+          unreadCount: byId.get(channel.id)?.unreadCount ?? 0,
+          mentionCount: byId.get(channel.id)?.mentionCount ?? 0,
+        })),
+      };
+    },
+    { isolationLevel: "repeatable read" },
+  );
+}

packages/database/src/chat/send-message.test.ts

@@ -1,4 +1,5 @@
 import { describe, expect, it } from "vitest";
+import { createHash } from "node:crypto";
 
 import {
   sendChatMessage,
@@ -26,12 +27,18 @@ function command(overrides: Record<string, unknown> = {}) {
 
 function harness(
   options: {
-    kind?: "public" | "private";
+    kind?: "public" | "private" | "direct";
     archived?: boolean;
     channelMember?: boolean;
     workspaceMember?: boolean;
+    role?: "owner" | "admin" | "member" | "guest";
+    factsMissing?: boolean;
+    factsWorkspaceId?: string;
+    factsChannelId?: string;
     existing?: boolean;
     persistedMentions?: string[];
+    pinnedFingerprint?: boolean;
+    changedBody?: boolean;
     parent?: {
       workspaceId: string;
       channelId: string;
@@ -46,10 +53,17 @@ function harness(
     workspaceId,
     channelId,
     sequence: 7n,
-    body: "hello",
     parentMessageId: null,
     authorUserId: userId,
     clientIdempotencyKey: "client-1",
+    body: options.changedBody ? "edited later" : "hello",
+    sendIntentFingerprint: options.pinnedFingerprint
+      ? createHash("sha256")
+          .update(
+            JSON.stringify([workspaceId, channelId, userId, "hello", null, []]),
+          )
+          .digest("hex")
+      : null,
   };
   const tx: ChatPersistenceTransaction = {
     claimIdempotency: async (scope) => {
@@ -62,16 +76,17 @@ function harness(
     },
     loadSendFacts: async (scope) => {
       calls.push(["facts", scope]);
+      if (options.factsMissing) return null;
       return {
         actor: {
-          userId,
+          userId: scope.actorUserId,
-          workspaceId,
+          workspaceId: options.factsWorkspaceId ?? workspaceId,
           activeWorkspaceMember: options.workspaceMember ?? true,
-          role: "member",
+          role: options.role ?? "member",
         },
         channel: {
-          id: channelId,
+          id: options.factsChannelId ?? channelId,
-          workspaceId,
+          workspaceId: options.factsWorkspaceId ?? workspaceId,
           kind: options.kind ?? "public",
           archived: options.archived ?? false,
           activeMember: options.channelMember ?? false,
@@ -147,10 +162,88 @@ describe("sendChatMessage", () => {
     expect((await h.run()).created).toBe(false);
     expect(h.calls.map(([name]) => name)).toEqual([
       "idempotency",
+      "facts",
       "existing-mentions",
     ]);
   });
 
+  it.each([
+    {
+      reason: "removed workspace membership",
+      options: { workspaceMember: false },
+    },
+    { reason: "missing current facts", options: { factsMissing: true } },
+    {
+      reason: "different workspace facts",
+      options: { factsWorkspaceId: "99999999-9999-4999-8999-999999999999" },
+    },
+    {
+      reason: "different channel facts",
+      options: { factsChannelId: "99999999-9999-4999-8999-999999999999" },
+    },
+    {
+      reason: "revoked private admin membership",
+      options: {
+        kind: "private" as const,
+        role: "admin" as const,
+        channelMember: false,
+      },
+    },
+    {
+      reason: "revoked direct admin membership",
+      options: {
+        kind: "direct" as const,
+        role: "admin" as const,
+        channelMember: false,
+      },
+    },
+  ])(
+    "rejects a replay after $reason without leaking message contents",
+    async ({ options }) => {
+      const h = harness({ existing: true, ...options });
+      await expect(h.run()).rejects.toMatchObject({ code: "CHAT_FORBIDDEN" });
+      expect(h.calls.map(([name]) => name)).toEqual(["idempotency", "facts"]);
+    },
+  );
+
+  it.each(["private", "direct"] as const)(
+    "allows a visible %s channel replay without creating another message",
+    async (kind) => {
+      const h = harness({ existing: true, kind, channelMember: true });
+      expect(await h.run()).toMatchObject({
+        created: false,
+        message: { id: h.message.id },
+      });
+      expect(h.calls.map(([name]) => name)).toEqual([
+        "idempotency",
+        "facts",
+        "existing-mentions",
+      ]);
+    },
+  );
+
+  it("allows replay in an archived channel if it remains readable, but not a new send", async () => {
+    const retry = harness({
+      existing: true,
+      archived: true,
+      kind: "private",
+      channelMember: true,
+    });
+    expect(await retry.run()).toMatchObject({
+      created: false,
+      message: { id: retry.message.id },
+    });
+    expect(retry.calls.map(([name]) => name)).toEqual([
+      "idempotency",
+      "facts",
+      "existing-mentions",
+    ]);
+
+    const fresh = harness({ archived: true });
+    await expect(fresh.run()).rejects.toMatchObject({ code: "CHAT_FORBIDDEN" });
+    expect(fresh.calls.map(([name]) => name)).toEqual(["idempotency", "facts"]);
+  });
+
   it("treats duplicate and reordered requested mentions as the same intent", async () => {
     const secondMentionId = "88888888-8888-4888-8888-888888888888";
     const h = harness({
@@ -161,7 +254,7 @@ describe("sendChatMessage", () => {
       (await h.run(command(), [secondMentionId, mentionId, secondMentionId]))
         .created,
     ).toBe(false);
-    expect(h.calls[1]).toEqual([
+    expect(h.calls[2]).toEqual([
       "existing-mentions",
       { workspaceId, channelId, messageId: h.message.id },
     ]);
@@ -183,17 +276,32 @@ describe("sendChatMessage", () => {
       });
       expect(h.calls.map(([name]) => name)).toEqual([
         "idempotency",
+        "facts",
         "existing-mentions",
       ]);
     },
   );
 
+  it("matches immutable send intent after a subsequent edit without exposing the fingerprint", async () => {
+    const h = harness({
+      existing: true,
+      pinnedFingerprint: true,
+      changedBody: true,
+    });
+    const replay = await h.run();
+    expect(replay.created).toBe(false);
+    expect(replay.message).not.toHaveProperty("sendIntentFingerprint");
+    await expect(
+      h.run(command({ body: "edited later" })),
+    ).rejects.toMatchObject({ code: "CHAT_IDEMPOTENCY_CONFLICT" });
+  });
+
   it("rejects a channel-key collision from another actor before allocating sequence", async () => {
     const h = harness({ existing: true });
     await expect(
       h.run(command(), [], "77777777-7777-4777-8777-777777777777"),
     ).rejects.toMatchObject({ code: "CHAT_IDEMPOTENCY_CONFLICT" });
-    expect(h.calls.map(([name]) => name)).toEqual(["idempotency"]);
+    expect(h.calls.map(([name]) => name)).toEqual(["idempotency", "facts"]);
   });
 
   it("rejects archived, nonmember, and unauthorized private channel writes", async () => {

packages/database/src/chat/send-message.ts

@@ -1,3 +1,4 @@
+import { createHash } from "node:crypto";
 import {
   canPerformChatAction,
   chatCommandSchema,
@@ -30,6 +31,8 @@ export type PersistedChatMessage = {
   parentMessageId: string | null;
   authorUserId: string;
   clientIdempotencyKey: string;
+  sendIntentFingerprint?: string | null;
+  status?: "sent" | "edited" | "deleted";
 };
 
 type IdempotencyScope = {
@@ -111,18 +114,55 @@ export class ChatPersistenceError extends Error {
   }
 }
 
+function publicMessage(message: PersistedChatMessage): PersistedChatMessage {
+  return {
+    id: message.id,
+    workspaceId: message.workspaceId,
+    channelId: message.channelId,
+    sequence: message.sequence,
+    body: message.body,
+    parentMessageId: message.parentMessageId,
+    authorUserId: message.authorUserId,
+    clientIdempotencyKey: message.clientIdempotencyKey,
+  };
+}
+
+function sendIntentFingerprint(
+  command: SendCommand,
+  actorUserId: string,
+  mentionedUserIds: string[],
+): string {
+  return createHash("sha256")
+    .update(
+      JSON.stringify([
+        command.workspaceId,
+        command.channelId,
+        actorUserId,
+        command.body,
+        command.parentMessageId ?? null,
+        mentionedUserIds,
+      ]),
+    )
+    .digest("hex");
+}
+
 function matchesIntent(
   message: PersistedChatMessage,
   command: SendCommand,
   actorUserId: string,
+  mentionedUserIds: string[],
 ): boolean {
   return (
     message.workspaceId === command.workspaceId &&
     message.channelId === command.channelId &&
     message.clientIdempotencyKey === command.clientIdempotencyKey &&
     message.authorUserId === actorUserId &&
+    (message.sendIntentFingerprint
+      ? message.sendIntentFingerprint ===
+        sendIntentFingerprint(command, actorUserId, mentionedUserIds)
+      : message.status !== "edited" && message.status !== "deleted" &&
         message.body === command.body &&
-    message.parentMessageId === (command.parentMessageId ?? null)
+        message.parentMessageId === (command.parentMessageId ?? null))
   );
 }
 
@@ -146,8 +186,25 @@ export async function sendChatMessage(
       channelId: command.channelId,
       clientIdempotencyKey: command.clientIdempotencyKey,
     });
+    const facts = await tx.loadSendFacts({
+      workspaceId: command.workspaceId,
+      channelId: command.channelId,
+      actorUserId,
+    });
+    if (
+      !facts ||
+      facts.channel.id !== command.channelId ||
+      !canPerformChatAction({
+        action: existing ? "read" : "send",
+        actor: facts.actor,
+        channel: facts.channel,
+        context: command.context,
+      })
+    )
+      throw new ChatPersistenceError("CHAT_FORBIDDEN");
+
     if (existing) {
-      if (!matchesIntent(existing, command, actorUserId))
+      if (!matchesIntent(existing, command, actorUserId, mentionIds))
         throw new ChatPersistenceError("CHAT_IDEMPOTENCY_CONFLICT");
       const persistedMentionIds = [
         ...new Set(
@@ -158,30 +215,16 @@ export async function sendChatMessage(
           }),
         ),
       ].sort();
+      if (!existing.sendIntentFingerprint && existing.body !== command.body)
+        throw new ChatPersistenceError("CHAT_IDEMPOTENCY_CONFLICT");
       if (
         persistedMentionIds.length !== mentionIds.length ||
         persistedMentionIds.some((id, index) => id !== mentionIds[index])
       )
         throw new ChatPersistenceError("CHAT_IDEMPOTENCY_CONFLICT");
-      return { created: false, message: existing };
+      return { created: false, message: publicMessage(existing) };
     }
 
-    const facts = await tx.loadSendFacts({
-      workspaceId: command.workspaceId,
-      channelId: command.channelId,
-      actorUserId,
-    });
-    if (
-      !facts ||
-      !canPerformChatAction({
-        action: "send",
-        actor: facts.actor,
-        channel: facts.channel,
-        context: command.context,
-      })
-    )
-      throw new ChatPersistenceError("CHAT_FORBIDDEN");
-
     if (command.parentMessageId) {
       const parent = await tx.loadParent({
         workspaceId: command.workspaceId,
@@ -220,6 +263,11 @@ export async function sendChatMessage(
       parentMessageId: command.parentMessageId ?? null,
       authorUserId: actorUserId,
       clientIdempotencyKey: command.clientIdempotencyKey,
+      sendIntentFingerprint: sendIntentFingerprint(
+        command,
+        actorUserId,
+        mentionIds,
+      ),
     });
 
     if (mentionIds.length)
@@ -251,6 +299,6 @@ export async function sendChatMessage(
       dedupeKey: `message:${inserted.id}:created:v1`,
       payload: event,
     });
-    return { created: true, message: inserted };
+    return { created: true, message: publicMessage(inserted) };
   });
 }

packages/database/src/schema/chat.ts

@@ -35,6 +35,9 @@ export const chatChannels = pgTable(
     nextSequence: bigint("next_sequence", { mode: "bigint" })
       .notNull()
       .default(sql`1`),
+    nextMutationSequence: bigint("next_mutation_sequence", { mode: "bigint" })
+      .notNull()
+      .default(sql`1`),
     /** Structural signal; transaction policy must reject writes when this is non-null. */
     archivedAt: timestamp("archived_at", { withTimezone: true }),
     createdAt: timestamp("created_at", { withTimezone: true })
@@ -126,6 +129,9 @@ export const chatMessages = pgTable(
     /** Opaque, pinned agent identity until the agent registry schema exists. */
     authorAgentId: uuid("author_agent_id"),
     body: text("body").notNull(),
+    /** Pinned send intent survives edits and tombstones. */
+    sendIntentFingerprint: text("send_intent_fingerprint"),
+    revision: integer("revision").notNull().default(1),
     status: varchar("status", { length: 16 })
       .notNull()
       .default("sent")
@@ -141,6 +147,10 @@ export const chatMessages = pgTable(
   },
   (table) => ({
     workspaceIdx: index("chat_messages_workspace_id_idx").on(table.workspaceId),
+    revisionCheck: check(
+      "chat_messages_revision_check",
+      sql`${table.revision} > 0`,
+    ),
     workspaceIdUnique: unique("chat_messages_workspace_id_id_unique").on(
       table.workspaceId,
       table.id,
@@ -213,6 +223,50 @@ export const chatMessageMentions = pgTable(
   }),
 );
 
+export const chatMessageMutations = pgTable(
+  "chat_message_mutations",
+  {
+    id: uuid("id").primaryKey().defaultRandom(),
+    workspaceId: uuid("workspace_id")
+      .notNull()
+      .references(() => workspaces.id, { onDelete: "cascade" }),
+    channelId: uuid("channel_id").notNull(),
+    messageId: uuid("message_id").notNull(),
+    mutationSequence: bigint("mutation_sequence", { mode: "bigint" }).notNull(),
+    revision: integer("revision").notNull(),
+    kind: varchar("kind", { length: 16 })
+      .notNull()
+      .$type<"edited" | "deleted">(),
+    createdAt: timestamp("created_at", { withTimezone: true })
+      .defaultNow()
+      .notNull(),
+  },
+  (table) => ({
+    workspaceIdx: index("chat_message_mutations_workspace_id_idx").on(
+      table.workspaceId,
+    ),
+    sequenceUnique: unique("chat_message_mutations_channel_sequence_unique").on(
+      table.workspaceId,
+      table.channelId,
+      table.mutationSequence,
+    ),
+    channelFk: foreignKey({
+      columns: [table.workspaceId, table.channelId],
+      foreignColumns: [chatChannels.workspaceId, chatChannels.id],
+      name: "chat_message_mutations_channel_fk",
+    }).onDelete("cascade"),
+    messageFk: foreignKey({
+      columns: [table.workspaceId, table.messageId],
+      foreignColumns: [chatMessages.workspaceId, chatMessages.id],
+      name: "chat_message_mutations_message_fk",
+    }).onDelete("cascade"),
+    revisionCheck: check(
+      "chat_message_mutations_revision_check",
+      sql`${table.revision} > 1`,
+    ),
+  }),
+);
+
 export const chatReadCursors = pgTable(
   "chat_read_cursors",
   {

packages/shared/src/chat/contracts.test.ts

@@ -4,6 +4,8 @@ import {
   CHAT_IDEMPOTENCY_KEY_MAX_LENGTH,
   CHAT_MESSAGE_MAX_BODY_LENGTH,
   chatCommandSchema,
+  chatCursorSchema,
+  chatSequenceSchema,
   chatContextSchema,
   chatMessageCreatedV1Schema,
   chatPolicyInputSchema,
@@ -11,6 +13,63 @@ import {
   getChatCommandAuthorizationAction,
 } from "./index";
 
+describe("chat decimal cursors", () => {
+  it("accepts the full signed PostgreSQL bigint range without JS number loss", () => {
+    expect(chatSequenceSchema.parse("9223372036854775807")).toBe(
+      "9223372036854775807",
+    );
+    expect(chatCursorSchema.parse("0")).toBe("0");
+    expect(
+      chatCommandSchema.parse({
+        type: "mark_read",
+        workspaceId: "11111111-1111-4111-8111-111111111111",
+        channelId: "33333333-3333-4333-8333-333333333333",
+        context: {
+          type: "workspace",
+          workspaceId: "11111111-1111-4111-8111-111111111111",
+        },
+        sequence: "9007199254740993",
+      }),
+    ).toMatchObject({ sequence: "9007199254740993" });
+  });
+
+  it.each([
+    "",
+    "00",
+    "01",
+    "-1",
+    "+1",
+    "1.0",
+    " 1",
+    "1e3",
+    "9223372036854775808",
+    1,
+  ])("rejects malformed cursor %s", (value) => {
+    expect(chatCursorSchema.safeParse(value).success).toBe(false);
+  });
+  it("requires a positive message sequence and rejects numeric mark-read", () => {
+    expect(chatSequenceSchema.safeParse("0").success).toBe(false);
+    expect(
+      chatMessageCreatedV1Schema.safeParse({
+        eventType: "chat.message.created.v1",
+        eventVersion: 1,
+        workspaceId: "11111111-1111-4111-8111-111111111111",
+        channelId: "33333333-3333-4333-8333-333333333333",
+        messageId: "44444444-4444-4444-8444-444444444444",
+        sequence: "9223372036854775808",
+        author: {
+          type: "human",
+          userId: "22222222-2222-4222-8222-222222222222",
+        },
+        body: "hello",
+        parentMessageId: null,
+        mentionedUserIds: [],
+        clientIdempotencyKey: "key",
+      }).success,
+    ).toBe(false);
+  });
+});
+
 const workspaceId = "11111111-1111-4111-8111-111111111111";
 const userId = "22222222-2222-4222-8222-222222222222";
 const channelId = "33333333-3333-4333-8333-333333333333";
@@ -81,9 +140,22 @@ function validCommands() {
       body: "hello",
       clientIdempotencyKey: "local-1",
     },
-    { ...scope, type: "edit_message" as const, messageId, body: "updated" },
+    {
-    { ...scope, type: "delete_message" as const, messageId },
+      ...scope,
-    { ...scope, type: "mark_read" as const, sequence: 1 },
+      type: "edit_message" as const,
+      messageId,
+      body: "updated",
+      expectedRevision: 1,
+      clientMutationId: messageId,
+    },
+    {
+      ...scope,
+      type: "delete_message" as const,
+      messageId,
+      expectedRevision: 1,
+      clientMutationId: messageId,
+    },
+    { ...scope, type: "mark_read" as const, sequence: "1" },
     { ...scope, type: "archive_channel" as const },
     {
       ...scope,

packages/shared/src/chat/index.ts

@@ -3,6 +3,17 @@ import { z } from "zod";
 import { workspaceRoles } from "../types/roles";
 
 const idSchema = z.string().uuid();
+const signedBigintMax = 9223372036854775807n;
+const inSignedBigintRange = (value: string) =>
+  /^(0|[1-9][0-9]*)$/.test(value) && BigInt(value) <= signedBigintMax;
+export const chatCursorSchema = z
+  .string()
+  .regex(/^(0|[1-9][0-9]*)$/)
+  .refine(inSignedBigintRange, "Cursor exceeds the database range");
+export const chatSequenceSchema = z
+  .string()
+  .regex(/^[1-9][0-9]*$/)
+  .refine(inSignedBigintRange, "Sequence exceeds the database range");
 export const CHAT_MESSAGE_MAX_BODY_LENGTH = 20_000;
 export const CHAT_IDEMPOTENCY_KEY_MAX_LENGTH = 200;
 
@@ -13,7 +24,7 @@ export const chatMessageCreatedV1Schema = z
     workspaceId: idSchema,
     channelId: idSchema,
     messageId: idSchema,
-    sequence: z.string().regex(/^[1-9]\d*$/),
+    sequence: chatSequenceSchema,
     author: z.object({ type: z.literal("human"), userId: idSchema }).strict(),
     body: z.string().min(1).max(CHAT_MESSAGE_MAX_BODY_LENGTH),
     parentMessageId: idSchema.nullable(),
@@ -26,6 +37,38 @@ export const chatMessageCreatedV1Schema = z
   .strict();
 export type ChatMessageCreatedV1 = z.infer<typeof chatMessageCreatedV1Schema>;
 
+// Mutation notifications are invalidations, not snapshots; never publish edited bodies.
+export const chatMessageMutationEventV1Schema = z.discriminatedUnion(
+  "eventType",
+  [
+    z
+      .object({
+        eventType: z.literal("chat.message.edited.v1"),
+        eventVersion: z.literal(1),
+        workspaceId: idSchema,
+        channelId: idSchema,
+        messageId: idSchema,
+        revision: z.number().int().positive(),
+        status: z.literal("edited"),
+      })
+      .strict(),
+    z
+      .object({
+        eventType: z.literal("chat.message.deleted.v1"),
+        eventVersion: z.literal(1),
+        workspaceId: idSchema,
+        channelId: idSchema,
+        messageId: idSchema,
+        revision: z.number().int().positive(),
+        status: z.literal("deleted"),
+      })
+      .strict(),
+  ],
+);
+export type ChatMessageMutationEventV1 = z.infer<
+  typeof chatMessageMutationEventV1Schema
+>;
+
 const channelEventBase = {
   eventVersion: z.literal(1),
   workspaceId: idSchema,
@@ -184,11 +227,18 @@ const commandVariants = [
     type: z.literal("edit_message"),
     messageId: idSchema,
     body: z.string().trim().min(1).max(CHAT_MESSAGE_MAX_BODY_LENGTH),
+    expectedRevision: z.number().int().min(1).max(2147483646),
+    clientMutationId: idSchema,
+  }),
+  commandVariant({
+    type: z.literal("delete_message"),
+    messageId: idSchema,
+    expectedRevision: z.number().int().min(1).max(2147483646),
+    clientMutationId: idSchema,
   }),
-  commandVariant({ type: z.literal("delete_message"), messageId: idSchema }),
   commandVariant({
     type: z.literal("mark_read"),
-    sequence: z.number().int().nonnegative(),
+    sequence: chatSequenceSchema,
   }),
   commandVariant({ type: z.literal("archive_channel") }),
   commandVariant({

plans/Plan-workspace-communications/Epic-workspace-chat/Task-chat-persistence.md

@@ -9,7 +9,7 @@ priority: P1
 tenant_id: global
 owner: unassigned
 cursor_todo_id: null
-updated_at: "2026-10-22"
+updated_at: "2026-10-26"
 ---
 
 # Task summary
@@ -20,6 +20,10 @@ Add the tenant-scoped relational foundation and tRPC API for durable workspace c
 
 Model channels, memberships, messages, threads, member read cursors, and the chat outbox contract in Postgres. Use `workspaceProcedure` and explicit `workspace_id` filters for every operation. This foundation accepts workspace context only and rejects object/unknown contexts without fallback. Authorization is workspace AND context AND channel permission: public workspace channels require active workspace access; private/direct require membership too; archived channels are read-only. Workspace admins/owners have no private/direct read or self-join bypass. Membership targets must be authorized, and all discovery/history/mention/count endpoints share the policy seam. Project/private-object access is deferred to `Task-contextual-chat-authorization.md`, not inferred from tenant ownership. Message writes must atomically include an outbox event, support idempotency, and allocate per-channel monotonic sequence cursors. Generate a new Drizzle migration; do not alter existing migrations.
 
+## Edit/tombstone continuation — 2026-10-26
+
+The user reports the migration blocker resolved and migrations applied; this session has not independently verified that database. Implement revision-checked edits and tombstones, immutable creation retry intent, authorization-checked mutation receipts, safe public projections, and a separately bounded durable mutation recovery cursor. Tombstones must not expose old body/mentions through history, receipts, or realtime replay. Acceptance remains open until integrated tests and live PostgreSQL gates pass.
+
 ## Live PostgreSQL gates
 
 - [ ] Concurrent identical direct-channel creation produces one channel, one exact membership set, and one lifecycle event set.

plans/Plan-workspace-communications/Epic-workspace-chat/Task-chat-realtime-delivery.md

@@ -4,12 +4,12 @@ slug: chat-realtime-delivery
 title: Deliver scalable realtime chat events
 plan_slug: workspace-communications
 epic_slug: workspace-chat
-status: ready
+status: in_progress
 priority: P1
 tenant_id: global
 owner: unassigned
 cursor_todo_id: null
-updated_at: "2026-10-22"
+updated_at: "2026-10-29"
 ---
 
 # Task summary
@@ -28,13 +28,25 @@ Use HTTP/tRPC for commands and Postgres as the source of truth. Consume the outb
 - [ ] Add per-channel reconnect catch-up and cross-device read-state invalidation.
 - [ ] Add heartbeat, backpressure, graceful-shutdown, outage, ordering, and authorization tests.
 
+## Go-live reconciliation — 2026-10-29
+
+Operator reports 14/14 live acceptance checks passed on CT 116 via `wss://chat.stillwell.cloud/chat`, including two-member messaging/replies, reconnect, unread, edit/delete recovery and revocation. CT 102 ledger is reported at 24 through `0023`, with Chat tables owned by `echodo`. Evidence is in `docs/Current Progress.md` (operator-supplied; not rerun here). Local regression tests cover projection/redaction fixes and hardened configured-origin ticket checks. Next deployment must rebuild web and retest; keep task in progress pending full checks/final review and unverified outage/multi-instance criteria.
+
+## Earlier deployment checkpoint — 2026-10-29 (superseded above)
+
+- Extended the existing collab service with Chat secret/port wiring, dual-listener liveness checks, standalone publication, web build URL, and Traefik TLS example. Public build variables now participate in Turbo build inputs.
+- Added `scripts/chat-deployment.test.mjs` to the root test command; all five configuration checks pass. This is not live transport acceptance.
+- CT 116 access blocked on trusted SSH host-key verification; no migrations or deployment performed. Production schema and disposable-Postgres, Redis outage, reconnect/revocation and two-browser gates remain open. See `docs/deployment-production.md` and `docs/Current Progress.md`.
+
 ## Owner or assignee
 
 `unassigned`
 
 ## Status
 
-`ready`
+`in_progress`
+
+2026-10-26: Implement signed, scoped short-lived subscription tickets and metadata-only WebSocket invalidation hints. Recheck current Postgres authorization at subscribe, delivery, and heartbeat; reconnect must recover edits/tombstones independently of creation-sequence catch-up. Do not reuse the legacy decode-only Yjs identity as Chat authentication.
 
 ## Estimation
 

pnpm-lock.yaml

@@ -47,6 +47,9 @@ importers:
       ioredis:
         specifier: ^5.4.2
         version: 5.10.1
+      ws:
+        specifier: ^8.20.0
+        version: 8.20.0
       yjs:
         specifier: ^13.6.22
         version: 13.6.30
@@ -54,6 +57,9 @@ importers:
         specifier: ^3.24.0
         version: 3.25.76
     devDependencies:
+      '@types/ws':
+        specifier: ^8.18.2
+        version: 8.18.2
       tsup:
         specifier: ^8.3.5
         version: 8.5.1(jiti@1.21.7)(postcss@8.5.8)(tsx@4.21.0)(typescript@5.9.3)(yaml@2.8.3)
@@ -63,6 +69,9 @@ importers:
       typescript:
         specifier: ^5.7.0
         version: 5.9.3
+      vitest:
+        specifier: ^2.1.9
+        version: 2.1.9(@types/node@22.19.15)
 
   apps/mcp-server:
     dependencies:
@@ -609,11 +618,11 @@ packages:
 
   '@esbuild-kit/core-utils@3.3.2':
     resolution: {integrity: sha512-sPRAnw9CdSsRmEtnsl2WXWdyquogVpB3yZ3dgwJfe8zrOzTsV7cJvmwrKVa+0ma5BoiGJ+BoqkMvawbayKUsqQ==}
-    deprecated: 'Merged into tsx: https://tsx.is'
+    deprecated: 'Merged into tsx: https://tsx.hirok.io'
 
   '@esbuild-kit/esm-loader@2.6.5':
     resolution: {integrity: sha512-FxEMIkJKnodyA1OaCUoEvbYRkoZlLZ4d/eXFu9Fh8CbBBgP5EmZxrfTRyN0qpXZ4vOvqnE5YdRdcrmUUXuU+dA==}
-    deprecated: 'Merged into tsx: https://tsx.is'
+    deprecated: 'Merged into tsx: https://tsx.hirok.io'
 
   '@esbuild/aix-ppc64@0.19.12':
     resolution: {integrity: sha512-bmoCYyWdEL3wDQIVbcyzRyeKLgk2WtWLTWz1ZIAZF/EGbNOwSA6ew3PftJ1PqMiOOGu0OyFMzG53L0zqIpPeNA==}
@@ -2803,6 +2812,9 @@ packages:
   '@types/use-sync-external-store@0.0.6':
     resolution: {integrity: sha512-zFDAD+tlpf2r4asuHEj0XH6pY6i0g5NeAHPn+15wk3BV6JA69eERFXC1gyGThDkVa1zCyKr5jox1+2LbV/AMLg==}
 
+  '@types/ws@8.18.2':
+    resolution: {integrity: sha512-67MQl+fpWKVTT1NYdnmo3U4sc/xPo/zQBncVnI74qmQa0z/b+1g6iYqNmGCPbxO+zz2aklb08a0oHfegiVd0/w==}
+
   '@typescript-eslint/eslint-plugin@8.57.2':
     resolution: {integrity: sha512-NZZgp0Fm2IkD+La5PR81sd+g+8oS6JwJje+aRWsDocxHkjyRw0J5L5ZTlN3LI1LlOcGL7ph3eaIUmTXMIjLk0w==}
     engines: {node: ^18.18.0 || ^20.9.0 || >=21.1.0}
@@ -3704,6 +3716,7 @@ packages:
   eslint@9.39.4:
     resolution: {integrity: sha512-XoMjdBOwe/esVgEvLmNsD3IRHkm7fbKIUGvrleloJXUZgDHig2IPWNniv+GwjyJXzuNqVjlr5+4yVUZjycJwfQ==}
     engines: {node: ^18.18.0 || ^20.9.0 || >=21.1.0}
+    deprecated: This version is no longer supported. Please see https://eslint.org/version-support for other options.
     hasBin: true
     peerDependencies:
       jiti: '*'
@@ -7545,6 +7558,10 @@ snapshots:
 
   '@types/use-sync-external-store@0.0.6': {}
 
+  '@types/ws@8.18.2':
+    dependencies:
+      '@types/node': 22.19.15
+
   '@typescript-eslint/eslint-plugin@8.57.2(@typescript-eslint/parser@8.57.2(eslint@9.39.4(jiti@1.21.7))(typescript@5.9.3))(eslint@9.39.4(jiti@1.21.7))(typescript@5.9.3)':
     dependencies:
       '@eslint-community/regexpp': 4.12.2

proxmox/ct116/traefik-ct100.example.yml

@@ -1,7 +1,8 @@
 # CT 100 Traefik dynamic config snippet — add to `/opt/appdata/traefik/config/dynamic/`
 # as `ct116-submatter.yml` (or merge into your homelab Traefik dynamic dir).
 #
-# CT 116 publishes web:3000 and collab:1234 directly (standalone compose override).
+# CT 116 publishes web:3000 and collab:1234/1235 (standalone compose override).
+# Port 1235 is workspace Chat, separate from Hocuspocus on 1234.
 # TLS terminates on CT 100; backends are plain HTTP.
 #
 # Remove stale routes after cutover (edit on CT 100, then push-traefik):
@@ -24,6 +25,13 @@ http:
       tls:
         certResolver: letsencrypt
 
+    echodo-chat:
+      rule: Host(`echodo-chat.stillwell.cloud`)
+      entryPoints: [websecure]
+      service: echodo-chat
+      tls:
+        certResolver: letsencrypt
+
   services:
     echodo-web:
       loadBalancer:
@@ -33,3 +41,8 @@ http:
       loadBalancer:
         servers:
           - url: "http://192.168.68.116:1234"
+    echodo-chat:
+      loadBalancer:
+        servers:
+          # Match CHAT_PORT if overridden in the deployment environment.
+          - url: "http://192.168.68.116:1235"

scripts/chat-deployment.test.mjs

@@ -0,0 +1,74 @@
+import assert from "node:assert/strict";
+import { readFileSync } from "node:fs";
+import { test } from "node:test";
+
+const read = (path) =>
+  readFileSync(new URL(`../${path}`, import.meta.url), "utf8");
+const compose = read("docker/docker-compose.prod.yml");
+const dockerfile = read("docker/Dockerfile");
+const web = compose.split("  web:\n")[1].split("\n  collab:")[0];
+const collab = compose.split("\n  collab:\n")[1].split("\n  mcp:")[0];
+
+test("web receives the Chat URL at build time, without baking the signing secret", () => {
+  assert.match(
+    web,
+    /NEXT_PUBLIC_CHAT_SERVER_URL: \$\{NEXT_PUBLIC_CHAT_SERVER_URL:-ws:\/\/localhost:1235\}/,
+  );
+  assert.match(
+    dockerfile,
+    /ARG NEXT_PUBLIC_CHAT_SERVER_URL=ws:\/\/localhost:1235/,
+  );
+  assert.match(
+    dockerfile,
+    /ENV NEXT_PUBLIC_CHAT_SERVER_URL=\$\{NEXT_PUBLIC_CHAT_SERVER_URL\}/,
+  );
+  assert.doesNotMatch(dockerfile, /(?:ARG|ENV) CHAT_TICKET_SECRET/);
+  const turbo = JSON.parse(read("turbo.json"));
+  assert.ok(turbo.tasks.build.env.includes("NEXT_PUBLIC_*"));
+});
+
+test("Workspace Chat uses its dedicated hostname, not the other app's hostname", () => {
+  assert.match(
+    read(".env.example"),
+    /^NEXT_PUBLIC_CHAT_SERVER_URL="wss:\/\/echodo-chat\.stillwell\.cloud"$/m,
+  );
+  const proxy = read("proxmox/ct116/traefik-ct100.example.yml");
+  assert.ok(proxy.includes("Host(`echodo-chat.stillwell.cloud`)"));
+  assert.ok(!proxy.includes("Host(`chat.stillwell.cloud`)"));
+});
+
+test("web and collab receive the same required runtime signing secret", () => {
+  for (const service of [web, collab]) {
+    assert.match(
+      service,
+      /CHAT_TICKET_SECRET=\$\{CHAT_TICKET_SECRET:\?[^}]+\}/,
+    );
+  }
+});
+
+test("collab exposes and checks both listeners", () => {
+  assert.match(collab, /CHAT_PORT=\$\{CHAT_PORT:-1235\}/);
+  assert.match(collab, /- "\$\{CHAT_PORT:-1235\}"/);
+  assert.match(collab, /process\.env\.PORT\|\|1234/);
+  assert.match(collab, /process\.env\.CHAT_PORT\|\|1235/);
+  assert.match(dockerfile, /EXPOSE 1234 1235/);
+});
+
+test("deployment preserves separately managed containers", () => {
+  assert.doesNotMatch(read("scripts/deploy-prod.sh"), /--remove-orphans/);
+});
+
+test("standalone publication matches the configurable Chat listener", () => {
+  assert.match(
+    read("docker/docker-compose.prod.standalone.yml"),
+    /"\$\{CHAT_PORT:-1235\}:\$\{CHAT_PORT:-1235\}"/,
+  );
+});
+
+test("Traefik example routes public Chat TLS to the second collab listener", () => {
+  const proxy = read("proxmox/ct116/traefik-ct100.example.yml");
+  assert.match(proxy, /Host\(`echodo-chat\.stillwell\.cloud`\)/);
+  assert.doesNotMatch(proxy, /Host\(`chat\.stillwell\.cloud`\)/);
+  assert.match(proxy, /service: echodo-chat/);
+  assert.match(proxy, /url: "http:\/\/192\.168\.68\.116:1235"/);
+});

scripts/deploy-prod.sh

@@ -32,17 +32,19 @@ compose() {
   docker compose --project-directory "$repo_root" "${compose_files[@]}" "$@"
 }
 
+# Preserve separately managed services (such as opencode-v2); orphan cleanup
+# requires an explicit operator decision, not every application deployment.
 case "$mode" in
   build)
     bash scripts/build-prod.sh
-    compose up -d --remove-orphans
+    compose up -d
     ;;
   up)
-    compose up -d --remove-orphans
+    compose up -d
     ;;
   pull)
     compose pull
-    compose up -d --remove-orphans
+    compose up -d
     ;;
   *)
     echo "Usage: $0 [build|up|pull]" >&2

turbo.json

@@ -4,7 +4,8 @@
   "tasks": {
     "build": {
       "dependsOn": ["^build"],
-      "outputs": [".next/**", "!.next/cache/**", "dist/**"]
+      "outputs": [".next/**", "!.next/cache/**", "dist/**"],
+      "env": ["NEXT_PUBLIC_*"]
     },
     "dev": {
       "cache": false,