feat(chat): scope agent to current organization

apps/web/src/hooks/chat/use-chat-agent.ts

@@ -1,16 +1,36 @@
+import { api } from "@code/backend/convex/_generated/api";
 import { useFlueAgent } from "@flue/react";
+import { useQuery } from "convex/react";
 
 import { CHAT_AGENT } from "@/lib/chat/constants";
 import type { ChatAgentState } from "@/lib/chat/types";
+import { setSendRequestId } from "@/root";
 
 export const useChatAgent = (): ChatAgentState => {
-  const agent = useFlueAgent(CHAT_AGENT);
+  // The global Zopu agent instance id is the user's current personal
+  // organization id. This is the hard tenancy boundary: the server rejects an
+  // instance id that is not the caller's own organization.
+  const organization = useQuery(api.organizations.getCurrent);
+  const agent = useFlueAgent({
+    ...CHAT_AGENT,
+    id: organization?._id,
+  });
+
+  const sendMessage = async (message: string): Promise<void> => {
+    // Generate a stable idempotency request id once per send, immediately
+    // before the message is submitted. It is reused across fetch/stream
+    // retries for this send and reset only on the next send, so the server
+    // can capture normalized evidence keyed by (organization, request id)
+    // without creating duplicates.
+    setSendRequestId(crypto.randomUUID());
+    await agent.sendMessage(message);
+  };
 
   return {
     error: agent.error,
     historyReady: agent.historyReady,
     messages: agent.messages,
-    sendMessage: agent.sendMessage,
+    sendMessage,
     status: agent.status,
   };
 };

apps/web/src/lib/chat/constants.ts

@@ -1,7 +1,11 @@
 import type { AgentStatus } from "@flue/react";
 
+/**
+ * The global Zopu agent. The instance `id` is the current organization id
+ * (resolved at runtime by the chat hook), not the legacy `main` shared
+ * instance. Organization scoping is the hard tenancy boundary.
+ */
 export const CHAT_AGENT = {
-  id: "main",
   live: "sse",
   name: "zopu",
 } as const;

apps/web/src/root.tsx

@@ -1,4 +1,4 @@
-import { WebAuthProvider } from "@code/auth/web";
+import { authClient, WebAuthProvider } from "@code/auth/web";
 import { env } from "@code/env/web";
 import { Toaster } from "@code/ui/components/sonner";
 import { FlueProvider } from "@flue/react";
@@ -31,6 +31,49 @@ export const links: Route.LinksFunction = () => [
 ];
 
 const flueBaseUrl = new URL(env.VITE_FLUE_URL);
+
+// ---------------------------------------------------------------------------
+// Authenticated Flue request headers.
+//
+// The access token is the Better Auth/Convex JWT, resolved once per session
+// through the installed `convex.token` plugin endpoint and cached. The stable
+// per-send request id is generated by the chat hook immediately before a user
+// sends a message and is reused across fetch/stream retries for that send, so
+// it is a reliable idempotency key for server-side evidence capture. It is
+// reset only when a new message is sent.
+// ---------------------------------------------------------------------------
+
+let cachedAccessToken: string | null = null;
+let currentSendRequestId: string | null = null;
+
+/** Set the idempotency request id for the next send. Called by the chat hook. */
+export const setSendRequestId = (id: string): void => {
+  currentSendRequestId = id;
+};
+
+const resolveAccessToken = async (): Promise<string | null> => {
+  if (cachedAccessToken) {
+    return cachedAccessToken;
+  }
+  const { data } = await authClient.convex.token({
+    fetchOptions: { throw: false },
+  });
+  cachedAccessToken = data?.token ?? null;
+  return cachedAccessToken;
+};
+
+const resolveFlueHeaders = async (): Promise<Record<string, string>> => {
+  const headers: Record<string, string> = {};
+  const accessToken = await resolveAccessToken();
+  if (accessToken) {
+    headers.authorization = `Bearer ${accessToken}`;
+  }
+  if (currentSendRequestId) {
+    headers["x-zopu-request-id"] = currentSendRequestId;
+  }
+  return headers;
+};
+
 const flueFetch: typeof fetch = async (input, init) => {
   const response = await fetch(input, init);
   if (
@@ -67,6 +110,7 @@ const flueFetch: typeof fetch = async (input, init) => {
 export const flueClient = createFlueClient({
   baseUrl: flueBaseUrl.toString(),
   fetch: flueFetch,
+  headers: resolveFlueHeaders,
 });
 export const Layout = ({ children }: { children: React.ReactNode }) => (
   <html lang="en">

packages/agents/src/agents/zopu.ts

@@ -2,7 +2,6 @@ import path from "node:path";
 
 import { parseAgentEnv } from "@code/env/agent";
 import { defineAgent } from "@flue/runtime";
-import type { AgentRouteHandler } from "@flue/runtime";
 // import { agentos } from "../sandboxes/agentos";
 import { local } from "@flue/runtime/node";
 
@@ -11,7 +10,7 @@ import { paseoCli } from "../tools/paseo";
 
 const repositoryRoot = path.resolve(process.cwd(), "../..");
 
-export const route: AgentRouteHandler = (_context, next) => next();
+export { authenticatedAgentRoute as route } from "../auth";
 
 export default defineAgent(({ env }) => {
   const { AGENT_MODEL_NAME, AGENT_MODEL_PROVIDER } = parseAgentEnv(env);

packages/agents/src/auth.ts

@@ -0,0 +1,318 @@
+/**
+ * Authenticated Flue route middleware for the organization-scoped global agent.
+ *
+ * Every request to the Zopu agent routes through here. The middleware:
+ *
+ * 1. Requires a Bearer access token (the Better Auth/Convex JWT).
+ * 2. Creates a fresh, request-scoped {@link ConvexHttpClient} and sets the
+ *    token on it — never on a shared/global client. Auth is per-request.
+ * 3. Resolves (ensuring) the caller's current personal organization.
+ * 4. Requires the agent instance id (`:id`) to equal that organization id, so
+ *    one organization cannot use or observe another's agent instance.
+ * 5. Authorizes GET/SSE observation routes as strictly as POST.
+ * 6. For POST only: captures exact user-message evidence *before* Flue
+ *    admission (`beginUserMessage`, status `admitting`), then patches it to
+ *    `admitted` with the Flue receipt submission id, or `failed` on error.
+ *    Observation requests (GET/HEAD) never create evidence.
+ *
+ * The organization id, conversation id, message id, role, and timestamp are
+ * never trusted from the client or an agent tool — they are resolved
+ * server-side. Only the raw message text travels from the user, and it is
+ * stored verbatim (never trimmed or rewritten).
+ */
+import { parseAgentEnv } from "@code/env/agent";
+import type { AgentRouteHandler } from "@flue/runtime";
+import { ConvexHttpClient } from "convex/browser";
+import { makeFunctionReference } from "convex/server";
+
+// ---------------------------------------------------------------------------
+// Typed function references (no generated backend bindings imported).
+// ---------------------------------------------------------------------------
+
+interface OrganizationView {
+  readonly _id: string;
+  readonly _creationTime: number;
+  readonly name: string;
+  readonly kind: "personal" | "team";
+  readonly createdBy: string;
+  readonly createdAt: number;
+}
+
+interface ConversationMessageView {
+  readonly _id: string;
+  readonly _creationTime: number;
+  readonly organizationId: string;
+  readonly conversationId: string;
+  readonly messageId: string;
+  readonly submissionId: string | null;
+  readonly clientRequestId: string;
+  readonly role: "user";
+  readonly rawText: string;
+  readonly status: "admitting" | "admitted" | "failed";
+  readonly createdAt: number;
+}
+
+const ensurePersonalOrganization = makeFunctionReference<
+  "mutation",
+  Record<string, never>,
+  OrganizationView
+>("organizations:ensurePersonalOrganization");
+
+const beginUserMessage = makeFunctionReference<
+  "mutation",
+  {
+    readonly organizationId: string;
+    readonly clientRequestId: string;
+    readonly rawText: string;
+  },
+  ConversationMessageView
+>("conversationMessages:beginUserMessage");
+
+const markAdmitted = makeFunctionReference<
+  "mutation",
+  {
+    readonly organizationId: string;
+    readonly clientRequestId: string;
+    readonly submissionId: string;
+  },
+  ConversationMessageView
+>("conversationMessages:markAdmitted");
+
+const markFailed = makeFunctionReference<
+  "mutation",
+  {
+    readonly organizationId: string;
+    readonly clientRequestId: string;
+  },
+  ConversationMessageView
+>("conversationMessages:markFailed");
+
+// ---------------------------------------------------------------------------
+// Request-scoped authenticated Convex client.
+// ---------------------------------------------------------------------------
+
+const { CONVEX_URL } = parseAgentEnv(process.env);
+
+/**
+ * A short-lived authenticated Convex client bound to one HTTP request. The
+ * caller's JWT is set on this instance only and discarded when the request
+ * ends. We never mutate auth on a shared/global Convex client.
+ */
+const createAuthenticatedClient = (accessToken: string): ConvexHttpClient => {
+  const client = new ConvexHttpClient(CONVEX_URL);
+  client.setAuth(accessToken);
+  return client;
+};
+
+/**
+ * Minimal structural shape the header helpers read from an incoming request.
+ * Using this avoids a structural clash between the global (Bun) `Request`/
+ * `Headers` and the undici types that Hono's context exposes, while staying
+ * type-safe.
+ */
+interface HeaderSource {
+  readonly headers: {
+    readonly get: (name: string) => string | null;
+  };
+}
+
+/**
+ * Extract and validate the Bearer access token from the request. Returns null
+ * when absent or malformed so the caller can produce a clean 401.
+ */
+const extractBearerToken = (request: HeaderSource): string | null => {
+  const header = request.headers.get("authorization");
+  if (!header) {
+    return null;
+  }
+  const match = /^Bearer\s+(?<token>\S+)$/u.exec(header);
+  return match?.groups?.token ?? null;
+};
+
+/**
+ * Read the stable per-send request id supplied by the web client. This is
+ * generated once when the user sends a message and reused across fetch/stream
+ * retries, so it is a reliable idempotency key for evidence capture.
+ */
+const extractClientRequestId = (request: HeaderSource): string | null => {
+  const header = request.headers.get("x-zopu-request-id");
+  if (!header || header.length === 0) {
+    return null;
+  }
+  return header;
+};
+
+/**
+ * Safely parse the direct agent payload to extract the exact user message
+ * text. Mirrors Flue's `DirectAgentPayload` shape `{ message, images? }`.
+ * Returns null for non-JSON or missing `message` so the request is rejected
+ * with 400 rather than crashing the middleware.
+ */
+const extractMessageText = async (
+  request: HeaderSource & { readonly json: () => Promise<unknown> }
+): Promise<string | null> => {
+  const contentType = request.headers.get("content-type") ?? "";
+  if (!contentType.includes("application/json")) {
+    return null;
+  }
+  try {
+    const body = (await request.json()) as unknown;
+    if (
+      typeof body !== "object" ||
+      body === null ||
+      !("message" in body) ||
+      typeof (body as { message?: unknown }).message !== "string"
+    ) {
+      return null;
+    }
+    return (body as { message: string }).message;
+  } catch {
+    return null;
+  }
+};
+
+/**
+ * Read the Flue admission receipt submission id from the response Flue
+ * produced after `next()`. The response body is `202 { streamUrl, offset,
+ * submissionId }`. Returns null when the body has no submission id (e.g. an
+ * error response).
+ */
+const extractSubmissionId = async (
+  response: HeaderSource & {
+    readonly clone: () => { readonly json: () => Promise<unknown> };
+  }
+): Promise<string | null> => {
+  const contentType = response.headers.get("content-type") ?? "";
+  if (!contentType.includes("application/json")) {
+    return null;
+  }
+  try {
+    const body = (await response.clone().json()) as unknown;
+    if (
+      typeof body === "object" &&
+      body !== null &&
+      "submissionId" in body &&
+      typeof (body as { submissionId?: unknown }).submissionId === "string"
+    ) {
+      return (body as { submissionId: string }).submissionId;
+    }
+  } catch {
+    // Non-JSON or unreadable body: no submission id to record.
+  }
+  return null;
+};
+
+/**
+ * Admit a direct submission through Flue and record the admission outcome on
+ * the normalized evidence row. On admission failure, the evidence is marked
+ * failed and the original error is rethrown — never swallowed. On success,
+ * the Flue receipt submission id (if present) patches the evidence to
+ * admitted.
+ */
+const admitAndRecordEvidence = async (
+  continueAdmission: () => Promise<void>,
+  client: ConvexHttpClient,
+  getResponse: () => Response | undefined,
+  evidence: {
+    readonly organizationId: string;
+    readonly clientRequestId: string;
+  }
+): Promise<void> => {
+  try {
+    await continueAdmission();
+  } catch (error) {
+    try {
+      await client.mutation(markFailed, evidence);
+    } catch {
+      // Best-effort: the original error is the primary concern.
+    }
+    throw error;
+  }
+
+  // Patch the evidence with the Flue receipt submission id. If the response
+  // carries no submission id (unexpected), leave the evidence in its begun
+  // state rather than guessing.
+  const response = getResponse();
+  if (!response) {
+    return;
+  }
+  const submissionId = await extractSubmissionId(response);
+  if (!submissionId) {
+    return;
+  }
+  try {
+    await client.mutation(markAdmitted, {
+      ...evidence,
+      submissionId,
+    });
+  } catch {
+    // Best-effort: the admission already succeeded and the response is
+    // finalized. A failure to patch metadata must not break the request.
+  }
+};
+
+/**
+ * The authenticated Flue route middleware. Authorizes every method, and for
+ * POST captures normalized user-message evidence around Flue admission.
+ */
+export const authenticatedAgentRoute: AgentRouteHandler = async (c, next) => {
+  const accessToken = extractBearerToken(c.req.raw);
+  if (!accessToken) {
+    return c.json({ error: "Unauthorized" }, 401);
+  }
+
+  const client = createAuthenticatedClient(accessToken);
+
+  // Resolve (ensuring) the caller's current personal organization.
+  let organization: OrganizationView;
+  try {
+    organization = await client.mutation(ensurePersonalOrganization, {});
+  } catch {
+    return c.json({ error: "Unauthorized" }, 401);
+  }
+
+  // The agent instance id must be the caller's organization id. This is the
+  // hard tenancy boundary: one organization cannot use or observe another's
+  // agent instance.
+  const instanceId = c.req.param("id") ?? "";
+  if (instanceId !== organization._id) {
+    return c.json({ error: "Forbidden" }, 403);
+  }
+
+  // Observation routes (GET/HEAD) are authorized but never create evidence.
+  if (c.req.method !== "POST") {
+    return next();
+  }
+
+  // --- Evidence capture (POST only) -------------------------------------
+
+  const clientRequestId = extractClientRequestId(c.req.raw);
+  const messageText = await extractMessageText(c.req.raw.clone());
+
+  if (!clientRequestId) {
+    return c.json({ error: "Missing x-zopu-request-id" }, 400);
+  }
+  if (messageText === null) {
+    return c.json({ error: "Invalid message payload" }, 400);
+  }
+
+  // Begin evidence before Flue admission so the message is durable even if
+  // admission fails or the process is interrupted.
+  try {
+    await client.mutation(beginUserMessage, {
+      clientRequestId,
+      organizationId: organization._id,
+      rawText: messageText,
+    });
+  } catch {
+    return c.json({ error: "Evidence capture failed" }, 500);
+  }
+
+  // Admit through Flue. After next(), c.res holds the admission response.
+  // Evidence is patched from the admission receipt, or marked failed without
+  // swallowing the original error.
+  await admitAndRecordEvidence(next, client, () => c.res, {
+    clientRequestId,
+    organizationId: organization._id,
+  });
+};

packages/auth/src/web/hooks.ts

@@ -1,12 +1,13 @@
 import { useForm } from "@tanstack/react-form";
+import { useCallback, useRef, useState } from "react";
 import { toast } from "sonner";
 
 import { signInSchema, signUpSchema } from "../shared/forms";
 import { authClient } from "./auth-client";
 
-type AuthFormOptions = {
+interface AuthFormOptions {
   readonly onSuccess?: () => void;
-};
+}
 
 export const useSignInForm = ({ onSuccess }: AuthFormOptions = {}) =>
   useForm({
@@ -23,7 +24,7 @@ export const useSignInForm = ({ onSuccess }: AuthFormOptions = {}) =>
             toast.success("Signed in successfully");
             onSuccess?.();
           },
-        },
+        }
       );
     },
     validators: { onSubmit: signInSchema },
@@ -34,7 +35,11 @@ export const useSignUpForm = ({ onSuccess }: AuthFormOptions = {}) =>
     defaultValues: { confirmPassword: "", email: "", name: "", password: "" },
     onSubmit: async ({ formApi, value }) => {
       await authClient.signUp.email(
-        { email: value.email.trim(), name: value.name.trim(), password: value.password },
+        {
+          email: value.email.trim(),
+          name: value.name.trim(),
+          password: value.password,
+        },
         {
           onError: ({ error }) => {
             toast.error(error.message || error.statusText);
@@ -44,8 +49,48 @@ export const useSignUpForm = ({ onSuccess }: AuthFormOptions = {}) =>
             toast.success("Account created successfully");
             onSuccess?.();
           },
-        },
+        }
       );
     },
     validators: { onSubmit: signUpSchema },
   });
+
+/**
+ * Resolve the current Better Auth/Convex JWT access token for the signed-in
+ * session. Uses the installed `convex.token` plugin endpoint (the supported
+ * API) and caches the result so repeated fetches within a session reuse it.
+ * Returns null when the user is not authenticated.
+ */
+export const useConvexAccessToken = (): (() => Promise<string | null>) => {
+  const cachedTokenRef = useRef<string | null>(null);
+  const pendingRef = useRef<Promise<string | null> | null>(null);
+  const [, forceRender] = useState(0);
+
+  return useCallback(async (): Promise<string | null> => {
+    if (cachedTokenRef.current) {
+      return cachedTokenRef.current;
+    }
+    if (pendingRef.current) {
+      return pendingRef.current;
+    }
+    pendingRef.current = (async (): Promise<string | null> => {
+      try {
+        const { data } = await authClient.convex.token({
+          fetchOptions: { throw: false },
+        });
+        const token = data?.token ?? null;
+        cachedTokenRef.current = token;
+        if (!token) {
+          forceRender((n) => n + 1);
+        }
+        return token;
+      } catch {
+        cachedTokenRef.current = null;
+        return null;
+      } finally {
+        pendingRef.current = null;
+      }
+    })();
+    return await pendingRef.current;
+  }, []);
+};

packages/auth/src/web/index.ts

@@ -1,4 +1,5 @@
 export { authClient } from "./auth-client";
 export { WebAuthProvider } from "./auth-provider";
+export { useConvexAccessToken } from "./hooks";
 export { LoginForm } from "./login-form";
 export { SignupForm } from "./signup-form";

packages/backend/convex/conversationIngress.smoke.test.ts

@@ -0,0 +1,90 @@
+/**
+ * Targeted runtime smoke for the authenticated conversation ingress.
+ *
+ * Simulates the exact sequence the Flue route middleware drives, end to end,
+ * against the real Convex schema and modules via convexTest:
+ *
+ *   1. ensurePersonalOrganization  (bearer auth → org resolution)
+ *   2. beginUserMessage            (evidence before Flue admission)
+ *   3. markAdmitted                (Flue receipt submission id)
+ *
+ * Then verifies the stored normalized evidence has exact raw text, organization
+ * scope, conversation/message identity, request id, role user, and the admitted
+ * submission id.
+ *
+ * This is a smoke, not a unit test: delete after the live browser smoke is
+ * confirmed.
+ */
+import { convexTest } from "convex-test";
+import { anyApi } from "convex/server";
+import { describe, expect, test } from "vitest";
+
+import schema from "./schema";
+
+declare global {
+  interface ImportMeta {
+    readonly glob: (pattern: string) => Record<string, () => Promise<unknown>>;
+  }
+}
+
+const modules = import.meta.glob("./**/*.ts");
+const api = anyApi;
+
+const ID_A = "https://convex.test|smoke-user";
+const identityA = { tokenIdentifier: ID_A };
+
+const newTest = () => convexTest({ schema, modules });
+
+describe("conversation ingress smoke", () => {
+  test("one user message produces normalized admitted evidence end to end", async () => {
+    const t = newTest();
+
+    // 1. Auth + org resolution (the middleware resolves this from the bearer
+    //    token via a request-scoped ConvexHttpClient).
+    const org = await t
+      .withIdentity(identityA)
+      .mutation(api.organizations.ensurePersonalOrganization, {});
+
+    const rawText = "  Fix the login bug  \n\t  please  ";
+    const clientRequestId = "smoke-req-001";
+
+    // 2. Begin evidence before Flue admission.
+    const begun = await t
+      .withIdentity(identityA)
+      .mutation(api.conversationMessages.beginUserMessage, {
+        clientRequestId,
+        organizationId: org._id,
+        rawText,
+      });
+    expect(begun.status).toBe("admitting");
+    expect(begun.rawText).toBe(rawText);
+
+    // 3. Mark admitted with the Flue receipt submission id.
+    const submissionId = "flue-sub-abc123";
+    await t
+      .withIdentity(identityA)
+      .mutation(api.conversationMessages.markAdmitted, {
+        clientRequestId,
+        organizationId: org._id,
+        submissionId,
+      });
+
+    // 4. Verify normalized evidence via authorized observation.
+    const observed = await t
+      .withIdentity(identityA)
+      .query(api.conversationMessages.getByRequest, {
+        clientRequestId,
+        organizationId: org._id,
+      });
+
+    expect(observed).not.toBeNull();
+    expect(observed?.rawText).toBe(rawText);
+    expect(observed?.organizationId).toBe(org._id);
+    expect(observed?.conversationId).toBe(org._id);
+    expect(typeof observed?.messageId).toBe("string");
+    expect(observed?.clientRequestId).toBe(clientRequestId);
+    expect(observed?.role).toBe("user");
+    expect(observed?.status).toBe("admitted");
+    expect(observed?.submissionId).toBe(submissionId);
+  });
+});

packages/backend/convex/conversationMessages.test.ts

@@ -0,0 +1,300 @@
+import { convexTest } from "convex-test";
+import { anyApi } from "convex/server";
+import { describe, expect, test } from "vitest";
+
+import { requireOrganizationMember } from "./authz";
+import schema from "./schema";
+
+declare global {
+  interface ImportMeta {
+    readonly glob: (pattern: string) => Record<string, () => Promise<unknown>>;
+  }
+}
+
+const modules = import.meta.glob("./**/*.ts");
+const api = anyApi;
+
+const ID_A = "https://convex.test|user-a";
+const ID_B = "https://convex.test|user-b";
+
+const identityA = { tokenIdentifier: ID_A };
+const identityB = { tokenIdentifier: ID_B };
+
+const newTest = () => convexTest({ schema, modules });
+
+const ensureOrg = async (
+  t: ReturnType<typeof newTest>,
+  identity: { readonly tokenIdentifier: string }
+) => {
+  const org = await t
+    .withIdentity(identity)
+    .mutation(api.organizations.ensurePersonalOrganization, {});
+  return org._id as string;
+};
+
+describe("conversationMessages", () => {
+  test("unauthenticated access is rejected", async () => {
+    const t = newTest();
+    const orgId = await ensureOrg(t, identityA);
+
+    await expect(
+      t.mutation(api.conversationMessages.beginUserMessage, {
+        clientRequestId: "req-1",
+        organizationId: orgId,
+        rawText: "hello",
+      })
+    ).rejects.toThrow(/Authentication required/u);
+
+    await expect(
+      t.mutation(api.conversationMessages.markAdmitted, {
+        clientRequestId: "req-1",
+        organizationId: orgId,
+        submissionId: "sub-1",
+      })
+    ).rejects.toThrow(/Authentication required/u);
+
+    await expect(
+      t.query(api.conversationMessages.getByRequest, {
+        clientRequestId: "req-1",
+        organizationId: orgId,
+      })
+    ).rejects.toThrow(/Authentication required/u);
+  });
+
+  test("cross-organization access is denied for begin and observation", async () => {
+    const t = newTest();
+    const orgA = await ensureOrg(t, identityA);
+    const _orgB = await ensureOrg(t, identityB);
+
+    // User A captures a message in its own organization.
+    const created = await t
+      .withIdentity(identityA)
+      .mutation(api.conversationMessages.beginUserMessage, {
+        clientRequestId: "req-cross",
+        organizationId: orgA,
+        rawText: "private to org A",
+      });
+
+    // User B cannot begin a message under organization A.
+    await expect(
+      t
+        .withIdentity(identityB)
+        .mutation(api.conversationMessages.beginUserMessage, {
+          clientRequestId: "req-cross",
+          organizationId: orgA,
+          rawText: "intruder",
+        })
+    ).rejects.toThrow(/Organization membership required/u);
+
+    // User B cannot observe organization A's message by request id.
+    await expect(
+      t.withIdentity(identityB).query(api.conversationMessages.getByRequest, {
+        clientRequestId: "req-cross",
+        organizationId: orgA,
+      })
+    ).rejects.toThrow(/Organization membership required/u);
+
+    // User B cannot list organization A's conversation.
+    await expect(
+      t
+        .withIdentity(identityB)
+        .query(api.conversationMessages.listForCurrentOrganization, {
+          organizationId: orgA,
+        })
+    ).rejects.toThrow(/Organization membership required/u);
+
+    // The membership boundary independently denies cross-org access.
+    await expect(
+      t
+        .withIdentity(identityB)
+        .query((ctx) => requireOrganizationMember(ctx, orgA as never))
+    ).rejects.toThrow(/Organization membership required/u);
+
+    expect(created.status).toBe("admitting");
+  });
+
+  test("duplicate request id is idempotent and preserves exact whitespace and casing", async () => {
+    const t = newTest();
+    const orgId = await ensureOrg(t, identityA);
+    const rawText = "   Hello   WORLD  \n\t with  spaces  ";
+
+    const first = await t
+      .withIdentity(identityA)
+      .mutation(api.conversationMessages.beginUserMessage, {
+        clientRequestId: "req-dup",
+        organizationId: orgId,
+        rawText,
+      });
+
+    // A second begin with the same request id returns the same row — no
+    // duplicate is created.
+    const second = await t
+      .withIdentity(identityA)
+      .mutation(api.conversationMessages.beginUserMessage, {
+        clientRequestId: "req-dup",
+        organizationId: orgId,
+        rawText: "DIFFERENT TEXT MUST BE IGNORED",
+      });
+
+    expect(second._id).toBe(first._id);
+    expect(second.messageId).toBe(first.messageId);
+    expect(second.rawText).toBe(rawText);
+    expect(second.clientRequestId).toBe("req-dup");
+    expect(second.role).toBe("user");
+    expect(second.status).toBe("admitting");
+
+    // The conversation id equals the organization id (global agent scope).
+    expect(first.conversationId).toBe(orgId);
+  });
+
+  test("admitted transition records the submission id", async () => {
+    const t = newTest();
+    const orgId = await ensureOrg(t, identityA);
+
+    const begun = await t
+      .withIdentity(identityA)
+      .mutation(api.conversationMessages.beginUserMessage, {
+        clientRequestId: "req-admit",
+        organizationId: orgId,
+        rawText: "please help",
+      });
+    expect(begun.status).toBe("admitting");
+    expect(begun.submissionId).toBeNull();
+
+    const admitted = await t
+      .withIdentity(identityA)
+      .mutation(api.conversationMessages.markAdmitted, {
+        clientRequestId: "req-admit",
+        organizationId: orgId,
+        submissionId: "sub-receipt-123",
+      });
+
+    expect(admitted.status).toBe("admitted");
+    expect(admitted.submissionId).toBe("sub-receipt-123");
+
+    // Marking admitted again is idempotent and keeps the same submission id.
+    const again = await t
+      .withIdentity(identityA)
+      .mutation(api.conversationMessages.markAdmitted, {
+        clientRequestId: "req-admit",
+        organizationId: orgId,
+        submissionId: "sub-receipt-123",
+      });
+    expect(again.status).toBe("admitted");
+    expect(again.submissionId).toBe("sub-receipt-123");
+  });
+
+  test("failed transition records the failed status and preserves evidence", async () => {
+    const t = newTest();
+    const orgId = await ensureOrg(t, identityA);
+
+    await t
+      .withIdentity(identityA)
+      .mutation(api.conversationMessages.beginUserMessage, {
+        clientRequestId: "req-fail",
+        organizationId: orgId,
+        rawText: "doomed message",
+      });
+
+    const failed = await t
+      .withIdentity(identityA)
+      .mutation(api.conversationMessages.markFailed, {
+        clientRequestId: "req-fail",
+        organizationId: orgId,
+      });
+
+    expect(failed.status).toBe("failed");
+    expect(failed.submissionId).toBeNull();
+    expect(failed.rawText).toBe("doomed message");
+
+    // A failed message is still observable as evidence.
+    const observed = await t
+      .withIdentity(identityA)
+      .query(api.conversationMessages.getByRequest, {
+        clientRequestId: "req-fail",
+        organizationId: orgId,
+      });
+    expect(observed?.status).toBe("failed");
+    expect(observed?.rawText).toBe("doomed message");
+  });
+
+  test("admission wins over a concurrent failure and cannot be overwritten", async () => {
+    const t = newTest();
+    const orgId = await ensureOrg(t, identityA);
+
+    const begun = await t
+      .withIdentity(identityA)
+      .mutation(api.conversationMessages.beginUserMessage, {
+        clientRequestId: "req-race",
+        organizationId: orgId,
+        rawText: "race",
+      });
+
+    // Admission completes first.
+    await t
+      .withIdentity(identityA)
+      .mutation(api.conversationMessages.markAdmitted, {
+        clientRequestId: "req-race",
+        organizationId: orgId,
+        submissionId: "sub-won",
+      });
+
+    // A late failure is a no-op: an admitted message keeps its submission id.
+    const afterFail = await t
+      .withIdentity(identityA)
+      .mutation(api.conversationMessages.markFailed, {
+        clientRequestId: "req-race",
+        organizationId: orgId,
+      });
+    expect(afterFail.status).toBe("admitted");
+    expect(afterFail.submissionId).toBe("sub-won");
+
+    expect(begun.status).toBe("admitting");
+  });
+
+  test("list returns only the authenticated organization's messages", async () => {
+    const t = newTest();
+    const orgA = await ensureOrg(t, identityA);
+    const orgB = await ensureOrg(t, identityB);
+
+    await t
+      .withIdentity(identityA)
+      .mutation(api.conversationMessages.beginUserMessage, {
+        clientRequestId: "req-a1",
+        organizationId: orgA,
+        rawText: "a-one",
+      });
+    await t
+      .withIdentity(identityA)
+      .mutation(api.conversationMessages.beginUserMessage, {
+        clientRequestId: "req-a2",
+        organizationId: orgA,
+        rawText: "a-two",
+      });
+    await t
+      .withIdentity(identityB)
+      .mutation(api.conversationMessages.beginUserMessage, {
+        clientRequestId: "req-b1",
+        organizationId: orgB,
+        rawText: "b-one",
+      });
+
+    const forA = await t
+      .withIdentity(identityA)
+      .query(api.conversationMessages.listForCurrentOrganization, {
+        organizationId: orgA,
+      });
+    const forB = await t
+      .withIdentity(identityB)
+      .query(api.conversationMessages.listForCurrentOrganization, {
+        organizationId: orgB,
+      });
+
+    expect(forA).toHaveLength(2);
+    expect(
+      forA.map((m: { clientRequestId: string }) => m.clientRequestId)
+    ).toEqual(expect.arrayContaining(["req-a1", "req-a2"]));
+    expect(forB).toHaveLength(1);
+    expect(forB[0]?.clientRequestId).toBe("req-b1");
+  });
+});

packages/backend/convex/conversationMessages.ts

@@ -0,0 +1,220 @@
+import { ConvexError, v } from "convex/values";
+
+import type { Doc, Id } from "./_generated/dataModel";
+import { mutation, query } from "./_generated/server";
+import { requireOrganizationMember } from "./authz";
+
+export type ConversationMessageStatus = "admitting" | "admitted" | "failed";
+
+interface ConversationMessageView {
+  readonly _id: Id<"conversationMessages">;
+  readonly _creationTime: number;
+  readonly organizationId: Id<"organizations">;
+  readonly conversationId: string;
+  readonly messageId: string;
+  readonly submissionId: string | null;
+  readonly clientRequestId: string;
+  readonly role: "user";
+  readonly rawText: string;
+  readonly status: ConversationMessageStatus;
+  readonly createdAt: number;
+}
+
+const toView = (doc: Doc<"conversationMessages">): ConversationMessageView => ({
+  _creationTime: doc._creationTime,
+  _id: doc._id,
+  clientRequestId: doc.clientRequestId,
+  conversationId: doc.conversationId,
+  createdAt: doc.createdAt,
+  messageId: doc.messageId,
+  organizationId: doc.organizationId,
+  rawText: doc.rawText,
+  role: doc.role,
+  status: doc.status,
+  submissionId: doc.submissionId ?? null,
+});
+
+/**
+ * The global Zopu conversation uses the organization ID as its conversation
+ * identity (and agent instance ID). This resolves the conversation ID for the
+ * authenticated user's current personal organization, so callers never supply
+ * tenancy themselves.
+ */
+const resolveConversationId = (organizationId: Id<"organizations">): string =>
+  organizationId;
+
+/**
+ * Idempotently begin a user message at the conversation ingress. If a row with
+ * the same (organizationId, clientRequestId) already exists, return it instead
+ * of creating a duplicate — Flue retries that replay the same request id must
+ * not produce duplicate evidence. The raw text is stored verbatim; it is never
+ * trimmed, collapsed, or rewritten.
+ *
+ * Authorization: the authenticated identity must be a member of the target
+ * organization. The conversation ID, message ID, and timestamp are all
+ * generated server-side; none are trusted from the caller.
+ */
+export const beginUserMessage = mutation({
+  args: {
+    organizationId: v.id("organizations"),
+    clientRequestId: v.string(),
+    rawText: v.string(),
+  },
+  handler: async (ctx, args): Promise<ConversationMessageView> => {
+    await requireOrganizationMember(ctx, args.organizationId);
+    const existing = await ctx.db
+      .query("conversationMessages")
+      .withIndex("by_organization_and_request", (q) =>
+        q
+          .eq("organizationId", args.organizationId)
+          .eq("clientRequestId", args.clientRequestId)
+      )
+      .unique();
+    if (existing) {
+      return toView(existing);
+    }
+    const createdAt = Date.now();
+    const messageId = crypto.randomUUID();
+    const conversationId = resolveConversationId(args.organizationId);
+    const rowId = await ctx.db.insert("conversationMessages", {
+      clientRequestId: args.clientRequestId,
+      conversationId,
+      createdAt,
+      messageId,
+      organizationId: args.organizationId,
+      rawText: args.rawText,
+      role: "user",
+      status: "admitting",
+    });
+    const created = await ctx.db.get(rowId);
+    if (!created) {
+      throw new Error("Failed to read created conversation message");
+    }
+    return toView(created);
+  },
+});
+
+/**
+ * Mark a begun message as admitted after Flue accepts the submission. The
+ * submission id links the normalized evidence to the Flue transcript. Only an
+ * `admitting` message may transition to `admitted`; an already-admitted or
+ * failed message is returned unchanged so a retry is idempotent.
+ */
+export const markAdmitted = mutation({
+  args: {
+    organizationId: v.id("organizations"),
+    clientRequestId: v.string(),
+    submissionId: v.string(),
+  },
+  handler: async (ctx, args): Promise<ConversationMessageView> => {
+    await requireOrganizationMember(ctx, args.organizationId);
+    const existing = await ctx.db
+      .query("conversationMessages")
+      .withIndex("by_organization_and_request", (q) =>
+        q
+          .eq("organizationId", args.organizationId)
+          .eq("clientRequestId", args.clientRequestId)
+      )
+      .unique();
+    if (!existing) {
+      throw new ConvexError("Conversation message not found");
+    }
+    if (existing.status === "admitting") {
+      await ctx.db.patch(existing._id, {
+        status: "admitted",
+        submissionId: args.submissionId,
+      });
+    }
+    const updated = await ctx.db.get(existing._id);
+    if (!updated) {
+      throw new Error("Failed to read updated conversation message");
+    }
+    return toView(updated);
+  },
+});
+
+/**
+ * Mark a begun message as failed when Flue admission rejects it. Failed
+ * messages remain as evidence with an explicit status; they are never silently
+ * deleted. Only an `admitting` message may transition to `failed`; an
+ * already-admitted message is never overwritten (admission won), and an
+ * already-failed message is returned unchanged.
+ */
+export const markFailed = mutation({
+  args: {
+    organizationId: v.id("organizations"),
+    clientRequestId: v.string(),
+  },
+  handler: async (ctx, args): Promise<ConversationMessageView> => {
+    await requireOrganizationMember(ctx, args.organizationId);
+    const existing = await ctx.db
+      .query("conversationMessages")
+      .withIndex("by_organization_and_request", (q) =>
+        q
+          .eq("organizationId", args.organizationId)
+          .eq("clientRequestId", args.clientRequestId)
+      )
+      .unique();
+    if (!existing) {
+      throw new ConvexError("Conversation message not found");
+    }
+    if (existing.status === "admitting") {
+      await ctx.db.patch(existing._id, { status: "failed" });
+    }
+    const updated = await ctx.db.get(existing._id);
+    if (!updated) {
+      throw new Error("Failed to read updated conversation message");
+    }
+    return toView(updated);
+  },
+});
+
+/**
+ * Read a single message by its client request id. Authorization requires
+ * membership in the message's organization; cross-organization observation is
+ * denied. Returns null when no message matches so observers cannot probe for
+ * another organization's request ids.
+ */
+export const getByRequest = query({
+  args: {
+    organizationId: v.id("organizations"),
+    clientRequestId: v.string(),
+  },
+  handler: async (ctx, args): Promise<ConversationMessageView | null> => {
+    await requireOrganizationMember(ctx, args.organizationId);
+    const existing = await ctx.db
+      .query("conversationMessages")
+      .withIndex("by_organization_and_request", (q) =>
+        q
+          .eq("organizationId", args.organizationId)
+          .eq("clientRequestId", args.clientRequestId)
+      )
+      .unique();
+    return existing ? toView(existing) : null;
+  },
+});
+
+/**
+ * List the user messages for the authenticated user's current organization
+ * conversation, newest first. This is the authorized observation path used by
+ * GET/SSE routes and product clients. Cross-organization access is denied.
+ */
+export const listForCurrentOrganization = query({
+  args: {
+    organizationId: v.id("organizations"),
+  },
+  handler: async (ctx, args): Promise<ConversationMessageView[]> => {
+    await requireOrganizationMember(ctx, args.organizationId);
+    const conversationId = resolveConversationId(args.organizationId);
+    const rows = await ctx.db
+      .query("conversationMessages")
+      .withIndex("by_organization_and_message", (q) =>
+        q
+          .eq("organizationId", args.organizationId)
+          .eq("conversationId", conversationId)
+      )
+      .order("desc")
+      .take(100);
+    return rows.map(toView);
+  },
+});

packages/backend/convex/schema.ts

@@ -167,6 +167,38 @@ export default defineSchema({
     .index("by_userId", ["userId"])
     .index("by_organizationId_and_userId", ["organizationId", "userId"]),
 
+  // -----------------------------------------------------------------
+  // Normalized conversation message evidence. Each row is one user
+  // message captured at the conversation ingress, before/with Flue
+  // admission. The raw text is stored exactly as the user typed it
+  // (never trimmed or rewritten). Organization scope is authoritative;
+  // an agent instance may only observe its own organization's messages.
+  // -----------------------------------------------------------------
+  conversationMessages: defineTable({
+    organizationId: v.id("organizations"),
+    conversationId: v.string(),
+    messageId: v.string(),
+    submissionId: v.optional(v.string()),
+    clientRequestId: v.string(),
+    role: v.literal("user"),
+    rawText: v.string(),
+    status: v.union(
+      v.literal("admitting"),
+      v.literal("admitted"),
+      v.literal("failed")
+    ),
+    createdAt: v.number(),
+  })
+    .index("by_organization_and_message", [
+      "organizationId",
+      "conversationId",
+      "messageId",
+    ])
+    .index("by_organization_and_request", [
+      "organizationId",
+      "clientRequestId",
+    ]),
+
   // -----------------------------------------------------------------
   // Flue persistence stores (schema/format version 4).
   // -----------------------------------------------------------------