diff --git a/apps/web/src/hooks/chat/use-chat-agent.ts b/apps/web/src/hooks/chat/use-chat-agent.ts index 93ee8cc..b0845fb 100644 --- a/apps/web/src/hooks/chat/use-chat-agent.ts +++ b/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 => { + // 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, }; }; diff --git a/apps/web/src/lib/chat/constants.ts b/apps/web/src/lib/chat/constants.ts index 0fa7a00..38e1bb9 100644 --- a/apps/web/src/lib/chat/constants.ts +++ b/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; diff --git a/apps/web/src/root.tsx b/apps/web/src/root.tsx index a5593cc..3040c5b 100644 --- a/apps/web/src/root.tsx +++ b/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 => { + if (cachedAccessToken) { + return cachedAccessToken; + } + const { data } = await authClient.convex.token({ + fetchOptions: { throw: false }, + }); + cachedAccessToken = data?.token ?? null; + return cachedAccessToken; +}; + +const resolveFlueHeaders = async (): Promise> => { + const headers: Record = {}; + 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 }) => ( diff --git a/packages/agents/src/agents/zopu.ts b/packages/agents/src/agents/zopu.ts index 756f628..dbf795d 100644 --- a/packages/agents/src/agents/zopu.ts +++ b/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); diff --git a/packages/agents/src/auth.ts b/packages/agents/src/auth.ts new file mode 100644 index 0000000..455de88 --- /dev/null +++ b/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, + 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+(?\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 } +): Promise => { + 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 }; + } +): Promise => { + 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, + client: ConvexHttpClient, + getResponse: () => Response | undefined, + evidence: { + readonly organizationId: string; + readonly clientRequestId: string; + } +): Promise => { + 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, + }); +}; diff --git a/packages/auth/src/web/hooks.ts b/packages/auth/src/web/hooks.ts index afa9a46..2ee230b 100644 --- a/packages/auth/src/web/hooks.ts +++ b/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) => { + const cachedTokenRef = useRef(null); + const pendingRef = useRef | null>(null); + const [, forceRender] = useState(0); + + return useCallback(async (): Promise => { + if (cachedTokenRef.current) { + return cachedTokenRef.current; + } + if (pendingRef.current) { + return pendingRef.current; + } + pendingRef.current = (async (): Promise => { + 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; + }, []); +}; diff --git a/packages/auth/src/web/index.ts b/packages/auth/src/web/index.ts index 1c5e31c..165cedf 100644 --- a/packages/auth/src/web/index.ts +++ b/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"; diff --git a/packages/backend/convex/conversationIngress.smoke.test.ts b/packages/backend/convex/conversationIngress.smoke.test.ts new file mode 100644 index 0000000..f659c70 --- /dev/null +++ b/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 Promise>; + } +} + +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); + }); +}); diff --git a/packages/backend/convex/conversationMessages.test.ts b/packages/backend/convex/conversationMessages.test.ts new file mode 100644 index 0000000..e6894f9 --- /dev/null +++ b/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 Promise>; + } +} + +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, + 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"); + }); +}); diff --git a/packages/backend/convex/conversationMessages.ts b/packages/backend/convex/conversationMessages.ts new file mode 100644 index 0000000..c997e4c --- /dev/null +++ b/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 => { + 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 => { + 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 => { + 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 => { + 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 => { + 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); + }, +}); diff --git a/packages/backend/convex/schema.ts b/packages/backend/convex/schema.ts index 5d25127..112d570 100644 --- a/packages/backend/convex/schema.ts +++ b/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). // -----------------------------------------------------------------