diff --git a/apps/web/src/hooks/chat/use-chat-agent.ts b/apps/web/src/hooks/chat/use-chat-agent.ts index 49e558f..9264152 100644 --- a/apps/web/src/hooks/chat/use-chat-agent.ts +++ b/apps/web/src/hooks/chat/use-chat-agent.ts @@ -55,7 +55,12 @@ export const useOrganizationChatAgent = ( const projected = projectConversation(rows ?? []); - let status: AgentStatus = projected.pending ? "submitted" : "idle"; + let status: AgentStatus = "idle"; + if (projected.streaming) { + status = "streaming"; + } else if (projected.pending) { + status = "submitted"; + } if (!organizationId || rows === undefined) { status = organization.error ? "error" : "connecting"; } else if (projected.failedError || sendError) { diff --git a/apps/web/src/lib/chat/conversation.test.ts b/apps/web/src/lib/chat/conversation.test.ts index e47cc27..48212e1 100644 --- a/apps/web/src/lib/chat/conversation.test.ts +++ b/apps/web/src/lib/chat/conversation.test.ts @@ -20,7 +20,7 @@ describe("projectConversation", () => { messageId: "user-1", rawText: "Build it", role: "user", - status: "processing", + status: "dispatching", }), row({ messageId: "assistant-1", role: "assistant", status: "queued" }), ]); @@ -28,6 +28,21 @@ describe("projectConversation", () => { expect(state.messages).toHaveLength(1); }); + test("projects partial assistant text as streaming", () => { + const state = projectConversation([ + row({ + messageId: "assistant-streaming", + rawText: "Working", + role: "assistant", + status: "running", + }), + ]); + expect(state.streaming).toBe(true); + expect(state.messages[0]?.parts).toEqual([ + { state: "streaming", text: "Working", type: "text" }, + ]); + }); + test("projects completed Convex rows into renderable messages", () => { const state = projectConversation([ row({ diff --git a/apps/web/src/lib/chat/conversation.ts b/apps/web/src/lib/chat/conversation.ts index 761f5de..1529494 100644 --- a/apps/web/src/lib/chat/conversation.ts +++ b/apps/web/src/lib/chat/conversation.ts @@ -11,40 +11,67 @@ export interface ConversationRow { readonly messageId: string; readonly rawText: string; readonly role: "assistant" | "user"; - readonly status: "completed" | "failed" | "processing" | "queued"; + readonly status: + | "aborted" + | "completed" + | "dispatching" + | "failed" + | "queued" + | "running"; } -export const projectConversation = (rows: readonly ConversationRow[]) => ({ - failedError: rows.findLast( - (row) => row.role === "assistant" && row.status === "failed" - )?.error, - messages: rows - .filter((row) => row.role === "user" || row.status === "completed") - .map((row) => ({ - id: row.messageId, - parts: [ - ...row.attachments.map((attachment) => ({ - filename: attachment.filename ?? undefined, - id: attachment.id, - mediaType: attachment.mediaType, - type: "file" as const, - url: attachment.url ?? undefined, - })), - ...(row.rawText - ? [ - { - state: "done" as const, - text: row.rawText, - type: "text" as const, - }, - ] - : []), - ], - role: row.role, - })), - pending: rows.some( +export const projectConversation = (rows: readonly ConversationRow[]) => { + const streaming = rows.some( (row) => row.role === "assistant" && - (row.status === "queued" || row.status === "processing") - ), -}); + row.status === "running" && + row.rawText.length > 0 + ); + return { + failedError: rows.findLast( + (row) => row.role === "assistant" && row.status === "failed" + )?.error, + messages: rows + .filter( + (row) => + row.role === "user" || + row.status === "completed" || + (row.role === "assistant" && + row.status === "running" && + row.rawText.length > 0) + ) + .map((row) => ({ + id: row.messageId, + parts: [ + ...row.attachments.map((attachment) => ({ + filename: attachment.filename ?? undefined, + id: attachment.id, + mediaType: attachment.mediaType, + type: "file" as const, + url: attachment.url ?? undefined, + })), + ...(row.rawText + ? [ + { + state: + row.status === "running" + ? ("streaming" as const) + : ("done" as const), + text: row.rawText, + type: "text" as const, + }, + ] + : []), + ], + role: row.role, + })), + pending: rows.some( + (row) => + row.role === "assistant" && + (row.status === "queued" || + row.status === "dispatching" || + row.status === "running") + ), + streaming, + }; +}; diff --git a/apps/web/src/routes.ts b/apps/web/src/routes.ts index 606d643..49082cb 100644 --- a/apps/web/src/routes.ts +++ b/apps/web/src/routes.ts @@ -6,7 +6,5 @@ export default [ route("login", "./routes/auth/login/page.tsx"), route("signup", "./routes/auth/signup/page.tsx"), ]), - layout("./routes/app/layout.tsx", [ - index("./routes/app/workspace/page.tsx"), - ]), + layout("./routes/app/layout.tsx", [index("./routes/app/workspace/page.tsx")]), ] satisfies RouteConfig; diff --git a/deploy/dokploy/agent.Dockerfile b/deploy/dokploy/agent.Dockerfile new file mode 100644 index 0000000..f669a6d --- /dev/null +++ b/deploy/dokploy/agent.Dockerfile @@ -0,0 +1,20 @@ +FROM oven/bun:1.3.14 AS build + +WORKDIR /app + +COPY . . +RUN bun install --frozen-lockfile +RUN bun run --filter @code/agents build + +FROM node:24-bookworm-slim + +ENV NODE_ENV=production +ENV PORT=3000 + +WORKDIR /app + +COPY --from=build /app /app + +EXPOSE 3000 + +CMD ["node", "packages/agents/dist/server.mjs"] diff --git a/deploy/dokploy/runner.Dockerfile b/deploy/dokploy/runner.Dockerfile new file mode 100644 index 0000000..9e96f73 --- /dev/null +++ b/deploy/dokploy/runner.Dockerfile @@ -0,0 +1,23 @@ +FROM oven/bun:1.3.14 AS bun + +FROM node:24-bookworm-slim + +COPY --from=bun /usr/local/bin/bun /usr/local/bin/bun + +RUN apt-get update \ + && apt-get install -y --no-install-recommends ca-certificates g++ git make python3 \ + && rm -rf /var/lib/apt/lists/* + +WORKDIR /opt/zopu-source + +COPY . . +RUN bun install --frozen-lockfile + +ENV AGENT_WORKSPACE_ROOT=/var/lib/zopu/workspaces +ENV BUN_EXECUTABLE=/usr/local/bin/bun +ENV DAEMON_ID=zopu-agentos-runner +ENV ZOPU_SOURCE_REPOSITORY=/opt/zopu-source + +VOLUME ["/var/lib/zopu/workspaces"] + +CMD ["bun", "packages/agents/src/runner.ts"] diff --git a/deploy/dokploy/runner.compose.yml b/deploy/dokploy/runner.compose.yml new file mode 100644 index 0000000..fe000c9 --- /dev/null +++ b/deploy/dokploy/runner.compose.yml @@ -0,0 +1,26 @@ +services: + runner: + build: + context: . + dockerfile: deploy/dokploy/runner.Dockerfile + environment: + AGENT_MODEL_API: ${AGENT_MODEL_API} + AGENT_MODEL_API_KEY: ${AGENT_MODEL_API_KEY} + AGENT_MODEL_BASE_URL: ${AGENT_MODEL_BASE_URL} + AGENT_MODEL_CONTEXT_WINDOW: ${AGENT_MODEL_CONTEXT_WINDOW} + AGENT_MODEL_MAX_TOKENS: ${AGENT_MODEL_MAX_TOKENS} + AGENT_MODEL_NAME: ${AGENT_MODEL_NAME} + AGENT_MODEL_PROVIDER: ${AGENT_MODEL_PROVIDER} + CONVEX_URL: ${CONVEX_URL} + DAEMON_ID: zopu-agentos-runner + FLUE_DB_TOKEN: ${FLUE_DB_TOKEN} + RIVET_ENDPOINT: ${RIVET_ENDPOINT} + RIVET_PUBLIC_ENDPOINT: ${RIVET_PUBLIC_ENDPOINT} + RIVET_WORKSPACE_TOKEN: ${RIVET_WORKSPACE_TOKEN} + ZOPU_SOURCE_REPOSITORY: /opt/zopu-source + volumes: + - runner-workspaces:/var/lib/zopu/workspaces + restart: unless-stopped + +volumes: + runner-workspaces: diff --git a/deploy/dokploy/web.Dockerfile b/deploy/dokploy/web.Dockerfile new file mode 100644 index 0000000..f8eac6a --- /dev/null +++ b/deploy/dokploy/web.Dockerfile @@ -0,0 +1,21 @@ +FROM oven/bun:1.3.14 AS build + +WORKDIR /app + +COPY . . +RUN bun install --frozen-lockfile +RUN bun run --filter web build + +FROM node:24-bookworm-slim + +ENV HOST=0.0.0.0 +ENV NODE_ENV=production +ENV PORT=3000 + +WORKDIR /app + +COPY --from=build /app /app + +EXPOSE 3000 + +CMD ["node", "apps/web/node_modules/.bin/react-router-serve", "apps/web/build/server/index.js"] diff --git a/deploy/zopu-runtime/.env.template b/deploy/zopu-runtime/.env.template index 9c3a510..10f67d8 100644 --- a/deploy/zopu-runtime/.env.template +++ b/deploy/zopu-runtime/.env.template @@ -43,7 +43,7 @@ AGENT_MODEL_MAX_TOKENS=131072 # --------------------------------------------------------------------------- RIVET_ENDPOINT=https://default:@rivet.example.com RIVET_PUBLIC_ENDPOINT=https://default@rivet.example.com -RIVET_ENVOY_VERSION=1 +RIVET_RUNNER_VERSION=1 RIVET_WORKSPACE_TOKEN=replace-with-a-long-random-workspace-token ZOPU_SOURCE_REPOSITORY=/opt/zopu-source # --------------------------------------------------------------------------- diff --git a/packages/agents/src/admission-context.ts b/packages/agents/src/admission-context.ts new file mode 100644 index 0000000..60686d5 --- /dev/null +++ b/packages/agents/src/admission-context.ts @@ -0,0 +1,16 @@ +import { AsyncLocalStorage } from "node:async_hooks"; + +export interface TurnAdmissionContext { + readonly clientRequestId: string; + readonly turnId: string; +} + +const turnAdmissionContext = new AsyncLocalStorage(); + +export const currentTurnAdmission = (): TurnAdmissionContext | undefined => + turnAdmissionContext.getStore(); + +export const withTurnAdmission = ( + context: TurnAdmissionContext, + run: () => T +): T => turnAdmissionContext.run(context, run); diff --git a/packages/agents/src/app.ts b/packages/agents/src/app.ts index ad2fdcd..491aeeb 100644 --- a/packages/agents/src/app.ts +++ b/packages/agents/src/app.ts @@ -80,6 +80,7 @@ app.post("/internal/work-attempts/:workspaceKey/cancel", async (context) => { }); app.all("/api/rivet/*", (context) => runtimeRegistry.handler(context.req.raw)); + app.route("/", flue()); export default app satisfies Fetchable; diff --git a/packages/agents/src/auth.ts b/packages/agents/src/auth.ts index 3899bc3..e2af8e3 100644 --- a/packages/agents/src/auth.ts +++ b/packages/agents/src/auth.ts @@ -1,6 +1,8 @@ import { parseAgentEnv } from "@code/env/agent"; import type { AgentRouteHandler } from "@flue/runtime"; +import { withTurnAdmission } from "./admission-context"; + /** Only Convex may invoke or observe the organization-scoped Flue agent. */ export const convexAgentRoute: AgentRouteHandler = async (context, next) => { const env = parseAgentEnv(process.env); @@ -14,5 +16,15 @@ export const convexAgentRoute: AgentRouteHandler = async (context, next) => { return context.json({ error: "Forbidden" }, 403); } - return await next().then(() => context.res); + if (context.req.method !== "POST") { + return await next(); + } + + const clientRequestId = context.req.header("x-zopu-request-id"); + const turnId = context.req.header("x-zopu-turn-id"); + if (!clientRequestId || !turnId) { + return context.json({ error: "Missing turn correlation headers" }, 400); + } + + return await withTurnAdmission({ clientRequestId, turnId }, () => next()); }; diff --git a/packages/agents/src/db.ts b/packages/agents/src/db.ts index 472edcd..d3f1d0a 100644 --- a/packages/agents/src/db.ts +++ b/packages/agents/src/db.ts @@ -26,11 +26,74 @@ */ import { env } from "@code/env/server"; -import { AttachmentConflictError, DEFAULT_LIST_LIMIT, DEFAULT_READ_LIMIT, MAX_LIST_LIMIT, MAX_READ_LIMIT, StreamListenerRegistry, assertSupportedFlueSchemaVersion, clampLimit, copyAttachmentBytes, createSessionStorageKey, decodeRunCursor, encodeRunCursor, formatOffset, hydratePersistedDirectSubmission, parseOffset, prepareDirectSubmission, verifyAttachmentBytes } from '@flue/runtime/adapter'; -import type { AgentAttemptMarker, AgentDispatchAdmission, AgentDispatchReceipt, AgentExecutionStore, AgentSubmission, AgentSubmissionStore, AttachmentRef, AttachmentStore, ConversationProducerClaim, ConversationRecord, ConversationStreamBatch, ConversationStreamIdentity, ConversationStreamMeta, ConversationStreamReadResult, ConversationStreamStore, CreateRunInput, DirectAgentSubmissionInput, DispatchAgentSubmissionInput, DispatchInput, EndRunInput, EventStreamMeta, EventStreamReadResult, EventStreamStore, GetAttachmentInput, PersistedChunkRow, PersistenceAdapter, PersistenceStores, PutAttachmentInput, RunPointer, RunRecord, RunStore, RunStatus, StoredAttachment, SubmissionAttemptRef, SubmissionClaimRef, SubmissionDurability, SubmissionSettlementObligation, SubmissionSettledRecord } from '@flue/runtime/adapter'; +import { + AttachmentConflictError, + DEFAULT_LIST_LIMIT, + DEFAULT_READ_LIMIT, + MAX_LIST_LIMIT, + MAX_READ_LIMIT, + StreamListenerRegistry, + assertSupportedFlueSchemaVersion, + clampLimit, + copyAttachmentBytes, + createSessionStorageKey, + decodeRunCursor, + encodeRunCursor, + formatOffset, + hydratePersistedDirectSubmission, + parseOffset, + prepareDirectSubmission, + verifyAttachmentBytes, +} from "@flue/runtime/adapter"; +import type { + AgentAttemptMarker, + AgentDispatchAdmission, + AgentDispatchReceipt, + AgentExecutionStore, + AgentSubmission, + AgentSubmissionStore, + AttachmentRef, + AttachmentStore, + ConversationProducerClaim, + ConversationRecord, + ConversationStreamBatch, + ConversationStreamIdentity, + ConversationStreamMeta, + ConversationStreamReadResult, + ConversationStreamStore, + CreateRunInput, + DirectAgentSubmissionInput, + DispatchAgentSubmissionInput, + DispatchInput, + EndRunInput, + EventStreamMeta, + EventStreamReadResult, + EventStreamStore, + GetAttachmentInput, + PersistedChunkRow, + PersistenceAdapter, + PersistenceStores, + PutAttachmentInput, + RunPointer, + RunRecord, + RunStore, + RunStatus, + StoredAttachment, + SubmissionAttemptRef, + SubmissionClaimRef, + SubmissionDurability, + SubmissionSettlementObligation, + SubmissionSettledRecord, +} from "@flue/runtime/adapter"; import { ConvexHttpClient } from "convex/browser"; -import { makeFunctionReference } from 'convex/server'; -import type { FunctionArgs, FunctionReference, FunctionReturnType } from 'convex/server'; +import { makeFunctionReference } from "convex/server"; +import type { + FunctionArgs, + FunctionReference, + FunctionReturnType, +} from "convex/server"; + +import { currentTurnAdmission } from "./admission-context"; // --------------------------------------------------------------------------- // Token + arg types @@ -42,7 +105,10 @@ import type { FunctionArgs, FunctionReference, FunctionReturnType } from 'convex // --------------------------------------------------------------------------- /** Auth token present on every Convex function call. */ -interface TokenArgs { readonly token: string; readonly [key: string]: unknown } +interface TokenArgs { + readonly token: string; + readonly [key: string]: unknown; +} const TOKEN = (): string => env.FLUE_DB_TOKEN; @@ -152,7 +218,7 @@ interface AttachmentRefWire { // --------------------------------------------------------------------------- const checkSchemaVersion = makeFunctionReference<"query", TokenArgs, string>( - "fluePersistence:checkSchemaVersion", + "fluePersistence:checkSchemaVersion" ); const getSubmission = makeFunctionReference< @@ -221,7 +287,11 @@ interface AdmitSubmissionResponse { } const admitSubmission = makeFunctionReference< "mutation", - TokenArgs & { readonly input: AdmitSubmissionEnvelope }, + TokenArgs & { + readonly clientRequestId?: string; + readonly input: AdmitSubmissionEnvelope; + readonly turnId?: string; + }, AdmitSubmissionResponse >("fluePersistence:admitSubmission"); @@ -290,11 +360,9 @@ type SettleArgs = TokenArgs & readonly outcome: "completed" | "failed"; readonly errorJson?: string; }; -const settleSubmission = makeFunctionReference< - "mutation", - SettleArgs, - boolean ->("fluePersistence:settleSubmission"); +const settleSubmission = makeFunctionReference<"mutation", SettleArgs, boolean>( + "fluePersistence:settleSubmission" +); const insertAttemptMarker = makeFunctionReference< "mutation", @@ -349,10 +417,16 @@ type AppendBatchArgs = TokenArgs & { readonly producerEpoch: number; readonly incarnation: string; readonly producerSequence: number; - readonly submission?: { readonly submissionId: string; readonly attemptId: string }; + readonly submission?: { + readonly submissionId: string; + readonly attemptId: string; + }; readonly recordsJson: string; }; -interface AppendResult { readonly offset: number; readonly appended: boolean } +interface AppendResult { + readonly offset: number; + readonly appended: boolean; +} const appendConversationBatch = makeFunctionReference< "mutation", AppendBatchArgs, @@ -438,7 +512,10 @@ const closeEventStream = makeFunctionReference< void >("fluePersistence:closeEventStream"); -interface EventStreamMetaRow { readonly nextOffset: number; readonly closed: boolean } +interface EventStreamMetaRow { + readonly nextOffset: number; + readonly closed: boolean; +} const getEventStreamMeta = makeFunctionReference< "query", TokenArgs & { readonly path: string }, @@ -454,7 +531,7 @@ type CreateRunArgs = TokenArgs & { readonly traceCarrierJson?: string; }; const createRun = makeFunctionReference<"mutation", CreateRunArgs, void>( - "fluePersistence:createRun", + "fluePersistence:createRun" ); type EndRunArgs = TokenArgs & { @@ -466,7 +543,7 @@ type EndRunArgs = TokenArgs & { readonly errorJson?: string; }; const endRun = makeFunctionReference<"mutation", EndRunArgs, void>( - "fluePersistence:endRun", + "fluePersistence:endRun" ); const getRun = makeFunctionReference< @@ -492,7 +569,7 @@ interface ListRunsResult { readonly hasMore: boolean; } const listRuns = makeFunctionReference<"query", ListRunsArgs, ListRunsResult>( - "fluePersistence:listRuns", + "fluePersistence:listRuns" ); // Attachments @@ -534,7 +611,9 @@ const deleteAttachmentsForInstance = makeFunctionReference< // --------------------------------------------------------------------------- const safeJsonParse = (text: string | null | undefined): T | undefined => { - if (text === null || text === undefined || text === "") {return undefined;} + if (text === null || text === undefined || text === "") { + return undefined; + } return JSON.parse(text) as T; }; @@ -544,7 +623,7 @@ const parseAcceptedAt = (value: string, label: string): number => { const ms = Date.parse(value); if (!Number.isFinite(ms)) { throw new TypeError( - `[flue] ${label} produced non-finite acceptedAt: ${value}`, + `[flue] ${label} produced non-finite acceptedAt: ${value}` ); } return ms; @@ -610,7 +689,7 @@ const hydrateSubmission = (row: SubmissionRow): AgentSubmission => { const input = safeJsonParse(row.inputJson); if (input === undefined) { throw new Error( - `[flue] persisted dispatch submission ${row.submissionId} has empty inputJson`, + `[flue] persisted dispatch submission ${row.submissionId} has empty inputJson` ); } const inputWithCarrier = @@ -621,7 +700,7 @@ const hydrateSubmission = (row: SubmissionRow): AgentSubmission => { const stripped = safeJsonParse(row.inputJson); if (stripped === undefined) { throw new Error( - `[flue] persisted direct submission ${row.submissionId} has empty inputJson`, + `[flue] persisted direct submission ${row.submissionId} has empty inputJson` ); } const chunks = safeJsonParse(row.chunksJson) ?? []; @@ -632,12 +711,12 @@ const hydrateSubmission = (row: SubmissionRow): AgentSubmission => { }; const hydrateObligation = ( - row: SettlementObligationRow, + row: SettlementObligationRow ): SubmissionSettlementObligation => { const record = safeJsonParse(row.recordJson); if (record === undefined) { throw new Error( - `[flue] persisted settlement obligation ${row.submissionId} has empty recordJson`, + `[flue] persisted settlement obligation ${row.submissionId} has empty recordJson` ); } return { @@ -692,14 +771,14 @@ class ConvexClient { query>( ref: F, - args: FunctionArgs, + args: FunctionArgs ): Promise> { return this.client.query(ref, args); } mutation>( ref: F, - args: FunctionArgs, + args: FunctionArgs ): Promise> { return this.client.mutation(ref, args); } @@ -757,7 +836,7 @@ class ConvexAgentSubmissionStore implements AgentSubmissionStore { async replaceSubmissionAttempt( attempt: SubmissionAttemptRef, nextAttemptId: string, - lease?: { ownerId: string; leaseExpiresAt: number }, + lease?: { ownerId: string; leaseExpiresAt: number } ): Promise { const row = await this.convex.mutation(replaceSubmissionAttempt, { attemptId: attempt.attemptId, @@ -808,10 +887,11 @@ class ConvexAgentSubmissionStore implements AgentSubmissionStore { } async admitDirect( - input: DirectAgentSubmissionInput, + input: DirectAgentSubmissionInput ): Promise { const sessionKey = createSessionStorageKey(input.id, "default", "default"); const extracted = prepareDirectSubmission(input); + const admission = currentTurnAdmission(); const res = await this.convex.mutation(admitSubmission, { input: { acceptedAt: parseAcceptedAt(input.acceptedAt, "admitDirect"), @@ -825,6 +905,12 @@ class ConvexAgentSubmissionStore implements AgentSubmissionStore { : { traceCarrierJson: jsonStringify(input.traceCarrier) }), }, token: TOKEN(), + ...(admission === undefined + ? {} + : { + clientRequestId: admission.clientRequestId, + turnId: admission.turnId, + }), }); if (res.kind === "submission" && res.submission !== undefined) { return hydrateSubmission(res.submission); @@ -833,12 +919,12 @@ class ConvexAgentSubmissionStore implements AgentSubmissionStore { // submission; if the backend signals a non-submission result we cannot // satisfy the `AgentSubmission` return type. throw new Error( - `[flue] admitDirect for ${input.submissionId} did not return a submission`, + `[flue] admitDirect for ${input.submissionId} did not return a submission` ); } async markSubmissionCanonicalReady( - submissionId: string, + submissionId: string ): Promise { const row = await this.convex.mutation(markSubmissionCanonicalReady, { submissionId, @@ -848,7 +934,7 @@ class ConvexAgentSubmissionStore implements AgentSubmissionStore { } async claimSubmission( - claim: SubmissionClaimRef, + claim: SubmissionClaimRef ): Promise { const row = await this.convex.mutation(claimSubmission, { attemptId: claim.attemptId, @@ -862,7 +948,7 @@ class ConvexAgentSubmissionStore implements AgentSubmissionStore { markSubmissionInputApplied( attempt: SubmissionAttemptRef, - durability?: SubmissionDurability, + durability?: SubmissionDurability ): Promise { return this.convex.mutation(markSubmissionInputApplied, { attemptId: attempt.attemptId, @@ -872,9 +958,7 @@ class ConvexAgentSubmissionStore implements AgentSubmissionStore { }); } - requestSubmissionRecovery( - attempt: SubmissionAttemptRef, - ): Promise { + requestSubmissionRecovery(attempt: SubmissionAttemptRef): Promise { return this.convex.mutation(requestSubmissionRecovery, { attemptId: attempt.attemptId, submissionId: attempt.submissionId, @@ -890,7 +974,7 @@ class ConvexAgentSubmissionStore implements AgentSubmissionStore { } requeueSubmissionBeforeInputApplied( - attempt: SubmissionAttemptRef, + attempt: SubmissionAttemptRef ): Promise { return this.convex.mutation(requeueSubmissionBeforeInputApplied, { attemptId: attempt.attemptId, @@ -901,7 +985,7 @@ class ConvexAgentSubmissionStore implements AgentSubmissionStore { async reserveSubmissionSettlement( attempt: SubmissionAttemptRef, - settlement: { recordId: string; record: SubmissionSettledRecord }, + settlement: { recordId: string; record: SubmissionSettledRecord } ): Promise { const row = await this.convex.mutation(reserveSubmissionSettlement, { attemptId: attempt.attemptId, @@ -915,7 +999,7 @@ class ConvexAgentSubmissionStore implements AgentSubmissionStore { finalizeSubmissionSettlement( attempt: SubmissionAttemptRef, - recordId: string, + recordId: string ): Promise { return this.convex.mutation(finalizeSubmissionSettlement, { attemptId: attempt.attemptId, @@ -936,7 +1020,7 @@ class ConvexAgentSubmissionStore implements AgentSubmissionStore { failSubmission( attempt: SubmissionAttemptRef, - error: unknown, + error: unknown ): Promise { return this.convex.mutation(settleSubmission, { attemptId: attempt.attemptId, @@ -1001,7 +1085,7 @@ class ConvexConversationStreamStore implements ConversationStreamStore { async createStream( path: string, - identity: ConversationStreamIdentity, + identity: ConversationStreamIdentity ): Promise { await this.convex.mutation(createConversationStream, { identity, @@ -1012,7 +1096,7 @@ class ConvexConversationStreamStore implements ConversationStreamStore { acquireProducer( path: string, - producerId: string, + producerId: string ): Promise { return this.convex.mutation(acquireConversationProducer, { path, @@ -1038,7 +1122,9 @@ class ConvexConversationStreamStore implements ConversationStreamStore { producerSequence: input.producerSequence, recordsJson: jsonStringify(input.records), token: TOKEN(), - ...(input.submission === undefined ? {} : { submission: input.submission }), + ...(input.submission === undefined + ? {} + : { submission: input.submission }), }); if (result.appended) { this.listeners.notify(input.path); @@ -1048,12 +1134,12 @@ class ConvexConversationStreamStore implements ConversationStreamStore { async read( path: string, - options?: { offset?: string; limit?: number }, + options?: { offset?: string; limit?: number } ): Promise { const limit = clampLimit( options?.limit, DEFAULT_READ_LIMIT, - MAX_READ_LIMIT, + MAX_READ_LIMIT ); const result = await this.convex.query(readConversationBatches, { afterOffset: afterOffset(options?.offset), @@ -1077,7 +1163,9 @@ class ConvexConversationStreamStore implements ConversationStreamStore { path, token: TOKEN(), }); - if (row === null) {return null;} + if (row === null) { + return null; + } return { identity: row.identity, incarnation: row.incarnation, @@ -1131,7 +1219,7 @@ class ConvexEventStreamStore implements EventStreamStore { async appendEventOnce( path: string, key: string, - event: unknown, + event: unknown ): Promise { const result = await this.convex.mutation(appendEventOnce, { dataJson: jsonStringify(event), @@ -1147,7 +1235,7 @@ class ConvexEventStreamStore implements EventStreamStore { async readEvents( path: string, - opts?: { offset?: string; limit?: number }, + opts?: { offset?: string; limit?: number } ): Promise { const limit = clampLimit(opts?.limit, DEFAULT_READ_LIMIT, MAX_READ_LIMIT); const result = await this.convex.query(readEventsFn, { @@ -1179,7 +1267,9 @@ class ConvexEventStreamStore implements EventStreamStore { path, token: TOKEN(), }); - if (row === null) {return null;} + if (row === null) { + return null; + } return { closed: row.closed, nextOffset: formatOffset(row.nextOffset), @@ -1233,7 +1323,7 @@ class ConvexRunStore implements RunStore { } lookupRun( - runId: string, + runId: string ): Promise<{ runId: string; workflowName: string } | null> { return this.convex.query(lookupRun, { runId, token: TOKEN() }); } @@ -1281,7 +1371,7 @@ class ConvexAttachmentStore implements AttachmentStore { attachment: attachmentRefToWire(input.attachment), bytes: bytes.buffer.slice( bytes.byteOffset, - bytes.byteOffset + bytes.byteLength, + bytes.byteOffset + bytes.byteLength ) as ArrayBuffer, conversationId: input.conversationId, streamPath: input.streamPath, @@ -1302,7 +1392,9 @@ class ConvexAttachmentStore implements AttachmentStore { streamPath: input.streamPath, token: TOKEN(), }); - if (row === null) {return null;} + if (row === null) { + return null; + } const attachment = wireRefToAttachmentRef(row.attachment); const bytes = new Uint8Array(row.bytes); // Verify integrity before handing bytes back to the runtime; throws @@ -1345,7 +1437,7 @@ class ConvexPersistenceAdapterImpl implements PersistenceAdapter { connect(): PersistenceStores { if (!this.migrated) { throw new Error( - "[flue] ConvexPersistenceAdapter.connect() called before migrate() completed", + "[flue] ConvexPersistenceAdapter.connect() called before migrate() completed" ); } const submissions = new ConvexAgentSubmissionStore(this.convex); diff --git a/packages/backend/convex/_generated/api.d.ts b/packages/backend/convex/_generated/api.d.ts index 9c9398c..cc8f90c 100644 --- a/packages/backend/convex/_generated/api.d.ts +++ b/packages/backend/convex/_generated/api.d.ts @@ -11,6 +11,7 @@ import type * as auth from "../auth.js"; import type * as authz from "../authz.js"; import type * as conversationMessages from "../conversationMessages.js"; +import type * as conversationProjections from "../conversationProjections.js"; import type * as crons from "../crons.js"; import type * as fluePersistence from "../fluePersistence.js"; import type * as gitConnectionData from "../gitConnectionData.js"; @@ -39,6 +40,7 @@ declare const fullApi: ApiFromModules<{ auth: typeof auth; authz: typeof authz; conversationMessages: typeof conversationMessages; + conversationProjections: typeof conversationProjections; crons: typeof crons; fluePersistence: typeof fluePersistence; gitConnectionData: typeof gitConnectionData; diff --git a/packages/backend/convex/conversationMessages.test.ts b/packages/backend/convex/conversationMessages.test.ts index 1389b3c..181aed0 100644 --- a/packages/backend/convex/conversationMessages.test.ts +++ b/packages/backend/convex/conversationMessages.test.ts @@ -20,17 +20,6 @@ const markProcessingRef = makeFunctionReference< { turnId: string; attempt: number; leaseOwner: string }, boolean >("conversationMessages:markProcessing"); -const completeTurnRef = makeFunctionReference< - "mutation", - { - turnId: string; - attempt: number; - leaseOwner: string; - submissionId: string; - text: string; - }, - boolean ->("conversationMessages:completeTurn"); const failTurnRef = makeFunctionReference< "mutation", { @@ -137,7 +126,7 @@ describe("conversationMessages", () => { ).rejects.toThrow(/Organization membership required/u); }); - test("fences stale turn attempts from overwriting a retry", async () => { + test("fences stale dispatch failures after admission", async () => { const t = newTest(); const organization = await ensureOrg(t, identityA); const sent = await t @@ -146,7 +135,7 @@ describe("conversationMessages", () => { clientRequestId: "request-fenced", images: [], organizationId: organization._id, - rawText: "Keep only the current response", + rawText: "Keep only the admitted submission", }); expect( @@ -156,46 +145,27 @@ describe("conversationMessages", () => { turnId: sent.turnId, }) ).toBe(true); + await t.run(async (ctx) => { + await ctx.db.patch(sent.turnId, { + leaseExpiresAt: undefined, + leaseOwner: undefined, + status: "running", + submissionId: "submission-1", + }); + }); + expect( await t.mutation(failTurnRef, { attempt: 1, - error: "retry", + error: "lost 202", leaseOwner: "worker-1", retry: true, turnId: sent.turnId, }) - ).toBe(true); - expect( - await t.mutation(completeTurnRef, { - attempt: 1, - leaseOwner: "worker-1", - submissionId: "stale", - text: "stale response", - turnId: sent.turnId, - }) ).toBe(false); - expect( - await t.mutation(markProcessingRef, { - attempt: 2, - leaseOwner: "worker-2", - turnId: sent.turnId, - }) - ).toBe(true); - expect( - await t.mutation(completeTurnRef, { - attempt: 2, - leaseOwner: "worker-2", - submissionId: "current", - text: "current response", - turnId: sent.turnId, - }) - ).toBe(true); - - const messages = await t - .withIdentity(identityA) - .query(api.conversationMessages.listForCurrentOrganization, { - organizationId: organization._id, - }); - expect(messages[1]?.rawText).toBe("current response"); + expect(await t.run((ctx) => ctx.db.get(sent.turnId))).toMatchObject({ + status: "running", + submissionId: "submission-1", + }); }); }); diff --git a/packages/backend/convex/conversationMessages.ts b/packages/backend/convex/conversationMessages.ts index f1903c1..6376c48 100644 --- a/packages/backend/convex/conversationMessages.ts +++ b/packages/backend/convex/conversationMessages.ts @@ -38,17 +38,6 @@ const markProcessingRef = makeFunctionReference< }, boolean >("conversationMessages:markProcessing"); -const completeTurnRef = makeFunctionReference< - "mutation", - { - turnId: Id<"conversationTurns">; - attempt: number; - leaseOwner: string; - submissionId: string; - text: string; - }, - boolean ->("conversationMessages:completeTurn"); const failTurnRef = makeFunctionReference< "mutation", { @@ -268,47 +257,7 @@ export const markProcessing = internalMutation({ error: undefined, leaseExpiresAt: Date.now() + 60_000, leaseOwner: args.leaseOwner, - status: "processing", - }); - return true; - }, -}); - -export const completeTurn = internalMutation({ - args: { - attempt: v.number(), - leaseOwner: v.string(), - submissionId: v.string(), - text: v.string(), - turnId: v.id("conversationTurns"), - }, - handler: async (ctx, args): Promise => { - const turn = await ctx.db.get(args.turnId); - if ( - !turn || - turn.status !== "processing" || - (turn.attemptNumber ?? 1) !== args.attempt || - turn.leaseOwner !== args.leaseOwner || - (turn.leaseExpiresAt !== undefined && turn.leaseExpiresAt < Date.now()) - ) { - return false; - } - const assistant = await ctx.db - .query("conversationMessages") - .withIndex("by_turnId_and_role", (q) => - q.eq("turnId", args.turnId).eq("role", "assistant") - ) - .unique(); - if (assistant) { - await ctx.db.patch(assistant._id, { content: args.text }); - } - await ctx.db.patch(args.turnId, { - completedAt: Date.now(), - error: undefined, - leaseExpiresAt: undefined, - leaseOwner: undefined, - status: "completed", - submissionId: args.submissionId, + status: "dispatching", }); return true; }, @@ -326,10 +275,9 @@ export const failTurn = internalMutation({ const turn = await ctx.db.get(args.turnId); if ( !turn || - turn.status !== "processing" || - (turn.attemptNumber ?? 1) !== args.attempt || + turn.status !== "dispatching" || turn.leaseOwner !== args.leaseOwner || - (turn.leaseExpiresAt !== undefined && turn.leaseExpiresAt < Date.now()) + (turn.attemptNumber ?? 1) !== args.attempt ) { return false; } @@ -392,7 +340,6 @@ export const runTurn = internalAction({ `agents/zopu/${encodeURIComponent(String(turn.organizationId))}`, `${flueUrl.replace(/\/+$/u, "")}/` ); - endpoint.searchParams.set("wait", "result"); const response = await fetch(endpoint, { body: JSON.stringify({ images, message: turn.user.content }), headers: { @@ -400,41 +347,29 @@ export const runTurn = internalAction({ "content-type": "application/json", "x-zopu-organization-id": String(turn.organizationId), "x-zopu-request-id": turn.turn.clientRequestId, + "x-zopu-turn-id": String(args.turnId), }, method: "POST", }); - const payload: unknown = await response.json().catch(() => null); - if ( - !response.ok || - typeof payload !== "object" || - payload === null || - !("submissionId" in payload) || - typeof payload.submissionId !== "string" || - !("result" in payload) || - typeof payload.result !== "object" || - payload.result === null || - !("text" in payload.result) || - typeof payload.result.text !== "string" - ) { - throw new Error(`Flue turn failed (${response.status})`); + if (!response.ok) { + throw new Error( + `Flue admission failed (${response.status}): ${await response.text().catch(() => "")}` + ); + } + const admitted = await ctx.runQuery(getTurnRef, { turnId: args.turnId }); + if (admitted?.turn.submissionId === undefined) { + throw new Error("Flue admission did not bind the product turn"); } - await ctx.runMutation(completeTurnRef, { - attempt: args.attempt, - leaseOwner, - submissionId: payload.submissionId, - text: payload.result.text, - turnId: args.turnId, - }); } catch (error) { const retry = args.attempt < MAX_ATTEMPTS; - await ctx.runMutation(failTurnRef, { + const failed = await ctx.runMutation(failTurnRef, { attempt: args.attempt, error: error instanceof Error ? error.message : String(error), leaseOwner, retry, turnId: args.turnId, }); - if (retry) { + if (failed && retry) { await ctx.scheduler.runAfter(args.attempt * 1000, runTurnRef, { attempt: args.attempt + 1, turnId: args.turnId, @@ -451,10 +386,18 @@ export const reconcileExpiredTurns = internalMutation({ const expired = await ctx.db .query("conversationTurns") .withIndex("by_status_and_leaseExpiresAt", (q) => - q.eq("status", "processing").lt("leaseExpiresAt", Date.now()) + q.eq("status", "dispatching").lt("leaseExpiresAt", Date.now()) ) .collect(); for (const turn of expired) { + if (turn.submissionId !== undefined) { + await ctx.db.patch(turn._id, { + leaseExpiresAt: undefined, + leaseOwner: undefined, + status: "running", + }); + continue; + } const attempt = turn.attemptNumber ?? 1; const retry = attempt < MAX_ATTEMPTS; await ctx.db.patch(turn._id, { diff --git a/packages/backend/convex/conversationProjections.ts b/packages/backend/convex/conversationProjections.ts new file mode 100644 index 0000000..b86a762 --- /dev/null +++ b/packages/backend/convex/conversationProjections.ts @@ -0,0 +1,145 @@ +import type { Id } from "./_generated/dataModel"; +import type { MutationCtx } from "./_generated/server"; + +interface ProjectionContext { + readonly assistantMessageId: Id<"conversationMessages"> | null; + readonly turnId: Id<"conversationTurns"> | null; +} + +const parseStringField = (record: unknown, key: string): string | undefined => { + if (typeof record !== "object" || record === null) { + return undefined; + } + const value = (record as Record)[key]; + return typeof value === "string" ? value : undefined; +}; + +const resolveContext = async ( + ctx: MutationCtx, + submissionId: string | undefined +): Promise => { + if (submissionId === undefined) { + return { assistantMessageId: null, turnId: null }; + } + const turn = await ctx.db + .query("conversationTurns") + .withIndex("by_submissionId", (q) => q.eq("submissionId", submissionId)) + .unique(); + if (turn === null) { + return { assistantMessageId: null, turnId: null }; + } + const assistant = await ctx.db + .query("conversationMessages") + .withIndex("by_turnId_and_role", (q) => + q.eq("turnId", turn._id).eq("role", "assistant") + ) + .unique(); + return { + assistantMessageId: assistant?._id ?? null, + turnId: turn._id, + }; +}; + +const terminalError = (record: unknown): string => { + if (typeof record !== "object" || record === null) { + return "Flue submission failed"; + } + const value = (record as Record).error; + if (typeof value === "string") { + return value.slice(0, 2000); + } + if (typeof value === "object" && value !== null) { + const { message } = value as Record; + if (typeof message === "string") { + return message.slice(0, 2000); + } + } + return "Flue submission failed"; +}; + +/** + * Project the small product view from Flue's canonical conversation stream. + * The raw stream remains authoritative; this function only updates the existing + * assistant message and turn row that the web already reads. + */ +export const projectConversationRecords = async ( + ctx: MutationCtx, + recordsJson: string, + submissionId: string | undefined +): Promise => { + let records: unknown[]; + try { + const parsed = JSON.parse(recordsJson) as unknown; + if (!Array.isArray(parsed)) { + return; + } + records = parsed; + } catch { + return; + } + + const projection = await resolveContext(ctx, submissionId); + if (projection.turnId === null) { + return; + } + + for (const record of records) { + const type = parseStringField(record, "type"); + if (type === "assistant_text_delta" && projection.assistantMessageId) { + const delta = parseStringField(record, "delta"); + if (delta !== undefined) { + const assistant = await ctx.db.get(projection.assistantMessageId); + if (assistant !== null) { + await ctx.db.patch(assistant._id, { + content: `${assistant.content}${delta}`, + }); + } + } + continue; + } + + if (type !== "submission_settled") { + continue; + } + + const turn = await ctx.db.get(projection.turnId); + if ( + turn === null || + turn.status === "completed" || + turn.status === "failed" || + turn.status === "aborted" + ) { + continue; + } + + const outcome = parseStringField(record, "outcome"); + if (outcome === "completed" && projection.assistantMessageId) { + const result = + typeof record === "object" && record !== null + ? (record as Record).result + : undefined; + const text = + typeof result === "object" && result !== null + ? (result as Record).text + : undefined; + if (typeof text === "string") { + await ctx.db.patch(projection.assistantMessageId, { content: text }); + } + } + + let status: "aborted" | "completed" | "failed" = "failed"; + if (outcome === "completed") { + status = "completed"; + } else if (outcome === "aborted") { + status = "aborted"; + } + + await ctx.db.patch(turn._id, { + completedAt: Date.now(), + error: outcome === "failed" ? terminalError(record) : undefined, + leaseExpiresAt: undefined, + leaseOwner: undefined, + status, + }); + } +}; diff --git a/packages/backend/convex/fluePersistence.test.ts b/packages/backend/convex/fluePersistence.test.ts index bebd4d4..4e1c7ca 100644 --- a/packages/backend/convex/fluePersistence.test.ts +++ b/packages/backend/convex/fluePersistence.test.ts @@ -74,6 +74,53 @@ describe("Flue Convex persistence", () => { }); }); + test("binds a direct admission to the product turn atomically", async () => { + const t = convexTest({ modules, schema }); + const organizationId = await t.run(async (ctx) => + ctx.db.insert("organizations", { + createdAt: 1, + createdBy: "user-1", + kind: "personal", + name: "Test", + }) + ); + const conversationId = await t.run(async (ctx) => + ctx.db.insert("conversations", { createdAt: 1, organizationId }) + ); + const turnId = await t.run(async (ctx) => + ctx.db.insert("conversationTurns", { + attemptNumber: 1, + clientRequestId: "request-1", + conversationId, + createdAt: 1, + leaseExpiresAt: 100, + leaseOwner: "worker", + status: "dispatching", + }) + ); + + await t.mutation(api.fluePersistence.admitSubmission, { + clientRequestId: "request-1", + input: { + acceptedAt: 1, + chunksJson: "[]", + inputJson: '{"kind":"direct"}', + kind: "direct", + sessionKey: "agent/instance/default", + submissionId: "submission-1", + }, + token, + turnId, + }); + const turn = await t.run((ctx) => ctx.db.get(turnId)); + expect(turn).toMatchObject({ + status: "running", + submissionId: "submission-1", + }); + expect(turn).not.toHaveProperty("leaseExpiresAt"); + expect(turn).not.toHaveProperty("leaseOwner"); + }); + test("fences stale conversation producers and conflicting attachments", async () => { const t = convexTest({ modules, schema }); await t.mutation(api.fluePersistence.createConversationStream, { diff --git a/packages/backend/convex/fluePersistence.ts b/packages/backend/convex/fluePersistence.ts index 4d6c1b4..a61f4b9 100644 --- a/packages/backend/convex/fluePersistence.ts +++ b/packages/backend/convex/fluePersistence.ts @@ -1,9 +1,11 @@ /* eslint-disable unicorn/no-array-sort, no-await-in-loop, unicorn/no-await-expression-member, unicorn/filename-case, unicorn/prefer-at, unicorn/no-array-reduce, @typescript-eslint/no-non-null-assertion */ import { env } from "@code/env/convex"; +import { v } from "convex/values"; + import type { Doc } from "./_generated/dataModel"; import { mutation, query } from "./_generated/server"; import type { MutationCtx, QueryCtx } from "./_generated/server"; -import { v } from "convex/values"; +import { projectConversationRecords } from "./conversationProjections"; const FLUE_SCHEMA_VERSION = "4"; const DURABILITY_DEFAULT_MAX_ATTEMPTS = 10; @@ -129,16 +131,24 @@ interface AttachmentWire { readonly bytes: ArrayBuffer; } - interface AdmitSubmissionResponse { readonly kind: "submission" | "retained_receipt" | "conflict"; readonly submission?: SubmissionRow; - readonly receipt?: { readonly submissionId: string; readonly acceptedAt: number }; + readonly receipt?: { + readonly submissionId: string; + readonly acceptedAt: number; + }; } -interface AppendResult { readonly offset: number; readonly appended: boolean } +interface AppendResult { + readonly offset: number; + readonly appended: boolean; +} -interface ListRunsCursor { readonly startedAt: string; readonly runId: string } +interface ListRunsCursor { + readonly startedAt: string; + readonly runId: string; +} interface OwnedConversationRecord { readonly id?: string; @@ -151,7 +161,7 @@ const submissionKind = v.union(v.literal("dispatch"), v.literal("direct")); const runStatus = v.union( v.literal("active"), v.literal("completed"), - v.literal("errored"), + v.literal("errored") ); const attachmentRef = v.object({ digest: v.string(), @@ -199,7 +209,7 @@ const toSubmissionRow = (doc: SubmissionDoc): SubmissionRow => ({ }); const toSettlementObligationRow = ( - doc: SubmissionDoc, + doc: SubmissionDoc ): SettlementObligationRow | null => { if ( doc.attemptId === undefined || @@ -224,7 +234,7 @@ const toAttemptMarkerRow = (doc: AttemptMarkerDoc): AttemptMarkerRow => ({ }); const toConversationStreamRow = ( - doc: ConversationStreamDoc, + doc: ConversationStreamDoc ): ConversationStreamRow => ({ identity: safeJsonParse(doc.identityJson), incarnation: doc.incarnation, @@ -234,7 +244,9 @@ const toConversationStreamRow = ( producerId: doc.producerId ?? null, }); -const toConversationBatchRow = (doc: ConversationBatchDoc): ConversationBatchRow => ({ +const toConversationBatchRow = ( + doc: ConversationBatchDoc +): ConversationBatchRow => ({ offset: doc.seq, recordsJson: doc.recordsJson, }); @@ -299,7 +311,7 @@ const sameAttachment = ( readonly conversationId: string; readonly attachment: AttachmentRefWire; readonly bytes: ArrayBuffer; - }, + } ): boolean => existing.conversationId === input.conversationId && existing.attachmentId === input.attachment.id && @@ -309,7 +321,10 @@ const sameAttachment = ( existing.filename === input.attachment.filename && compareBuffers(existing.bytes, input.bytes); -const compareRunPointerDesc = (left: RunPointerRow, right: RunPointerRow): number => { +const compareRunPointerDesc = ( + left: RunPointerRow, + right: RunPointerRow +): number => { if (left.startedAt !== right.startedAt) { return left.startedAt < right.startedAt ? 1 : -1; } @@ -326,11 +341,10 @@ const parseSessionInstance = (sessionKey: string): string | undefined => { return undefined; } try { - const parsed = JSON.parse(sessionKey.slice("agent-session:".length)) as unknown; - if ( - Array.isArray(parsed) && - typeof parsed[0] === "string" - ) { + const parsed = JSON.parse( + sessionKey.slice("agent-session:".length) + ) as unknown; + if (Array.isArray(parsed) && typeof parsed[0] === "string") { return parsed[0]; } } catch { @@ -350,11 +364,15 @@ const parseOwnedRecord = (record: unknown): OwnedConversationRecord | null => { ...(typeof value.submissionId === "string" ? { submissionId: value.submissionId } : {}), - ...(typeof value.attemptId === "string" ? { attemptId: value.attemptId } : {}), + ...(typeof value.attemptId === "string" + ? { attemptId: value.attemptId } + : {}), }; }; -const parseSettledOutcome = (recordJson: string | undefined): SettledOutcome | undefined => { +const parseSettledOutcome = ( + recordJson: string | undefined +): SettledOutcome | undefined => { if (recordJson === undefined) { return undefined; } @@ -376,10 +394,9 @@ const parseSettledOutcome = (recordJson: string | undefined): SettledOutcome | u type ReadCtx = QueryCtx | MutationCtx; - const getSubmissionDoc = ( ctx: ReadCtx, - submissionId: string, + submissionId: string ): Promise => ctx.db .query("flueSubmissions") @@ -388,14 +405,13 @@ const getSubmissionDoc = ( const getConversationStreamDoc = ( ctx: ReadCtx, - path: string, + path: string ): Promise => ctx.db .query("flueConversationStreams") .withIndex("by_path", (q) => q.eq("path", path)) .unique(); - const assertSubmissionAuthorization = async ( ctx: ReadCtx, path: string, @@ -405,7 +421,7 @@ const assertSubmissionAuthorization = async ( readonly attemptId: string; } | undefined, - recordsJson: string, + recordsJson: string ): Promise => { const records = safeJsonParse(recordsJson); const owned = records @@ -413,13 +429,13 @@ const assertSubmissionAuthorization = async ( .filter( (record): record is OwnedConversationRecord => record !== null && - (record.submissionId !== undefined || record.attemptId !== undefined), + (record.submissionId !== undefined || record.attemptId !== undefined) ); if (submission === undefined) { if (owned.length > 0) { throw new Error( - `[flue] Conversation stream "${path}" received submission-owned records without authorization.`, + `[flue] Conversation stream "${path}" received submission-owned records without authorization.` ); } return; @@ -429,11 +445,11 @@ const assertSubmissionAuthorization = async ( owned.some( (record) => record.submissionId !== submission.submissionId || - record.attemptId !== submission.attemptId, + record.attemptId !== submission.attemptId ) ) { throw new Error( - `[flue] Conversation stream "${path}" record ownership does not match the authorized submission attempt.`, + `[flue] Conversation stream "${path}" record ownership does not match the authorized submission attempt.` ); } @@ -441,11 +457,13 @@ const assertSubmissionAuthorization = async ( const stored = await getSubmissionDoc(ctx, submission.submissionId); if (stream === null || stored === null) { throw new Error( - `[flue] Submission attempt no longer owns work for agent instance "${path}".`, + `[flue] Submission attempt no longer owns work for agent instance "${path}".` ); } - const streamIdentity = safeJsonParse(stream.identityJson); + const streamIdentity = safeJsonParse( + stream.identityJson + ); const terminalizingSettlement = stored.status === "terminalizing" && stored.attemptId === submission.attemptId && @@ -466,7 +484,7 @@ const assertSubmissionAuthorization = async ( ) ) { throw new Error( - `[flue] Submission attempt no longer owns work for agent instance "${path}".`, + `[flue] Submission attempt no longer owns work for agent instance "${path}".` ); } }; @@ -489,7 +507,9 @@ export const getSubmission = query({ assertToken(args.token); const submission = await ctx.db .query("flueSubmissions") - .withIndex("by_submissionId", (q) => q.eq("submissionId", args.submissionId)) + .withIndex("by_submissionId", (q) => + q.eq("submissionId", args.submissionId) + ) .unique(); return submission === null ? null : toSubmissionRow(submission); }, @@ -509,7 +529,7 @@ export const listRunnableSubmissions = query({ handler: async (ctx, args) => { assertToken(args.token); const rows = (await ctx.db.query("flueSubmissions").collect()).sort( - (left, right) => left.sequence - right.sequence, + (left, right) => left.sequence - right.sequence ); return rows .filter( @@ -520,8 +540,8 @@ export const listRunnableSubmissions = query({ (candidate) => candidate.sessionKey === row.sessionKey && candidate.sequence < row.sequence && - isUnsettled(candidate.status), - ), + isUnsettled(candidate.status) + ) ) .map(toSubmissionRow); }, @@ -533,7 +553,7 @@ export const listUnreadySubmissions = query({ assertToken(args.token); return (await ctx.db.query("flueSubmissions").collect()) .filter( - (row) => row.status === "queued" && row.canonicalReadyAt === undefined, + (row) => row.status === "queued" && row.canonicalReadyAt === undefined ) .sort((left, right) => left.sequence - right.sequence) .map(toSubmissionRow); @@ -576,7 +596,9 @@ export const replaceSubmissionAttempt = mutation({ assertToken(args.token); const submission = await ctx.db .query("flueSubmissions") - .withIndex("by_submissionId", (q) => q.eq("submissionId", args.submissionId)) + .withIndex("by_submissionId", (q) => + q.eq("submissionId", args.submissionId) + ) .unique(); if ( submission === null || @@ -591,7 +613,9 @@ export const replaceSubmissionAttempt = mutation({ attemptId: args.nextAttemptId, recoveryRequestedAt: undefined, startedAt: now, - ...(args.ownerId === undefined ? { ownerId: undefined } : { ownerId: args.ownerId }), + ...(args.ownerId === undefined + ? { ownerId: undefined } + : { ownerId: args.ownerId }), ...(args.leaseExpiresAt === undefined ? { leaseExpiresAt: submission.leaseExpiresAt } : { leaseExpiresAt: args.leaseExpiresAt }), @@ -605,6 +629,7 @@ export const replaceSubmissionAttempt = mutation({ export const admitSubmission = mutation({ args: { ...tokenArgs, + clientRequestId: v.optional(v.string()), input: v.object({ acceptedAt: v.number(), chunksJson: v.optional(v.string()), @@ -614,12 +639,25 @@ export const admitSubmission = mutation({ submissionId: v.string(), traceCarrierJson: v.optional(v.string()), }), + turnId: v.optional(v.id("conversationTurns")), }, handler: async (ctx, args): Promise => { assertToken(args.token); + const correlatedTurn = + args.clientRequestId === undefined || args.turnId === undefined + ? null + : await ctx.db.get(args.turnId); + if ( + correlatedTurn !== null && + correlatedTurn.clientRequestId !== args.clientRequestId + ) { + throw new Error("[flue] Turn admission correlation does not match."); + } const existing = await ctx.db .query("flueSubmissions") - .withIndex("by_submissionId", (q) => q.eq("submissionId", args.input.submissionId)) + .withIndex("by_submissionId", (q) => + q.eq("submissionId", args.input.submissionId) + ) .unique(); if (existing !== null) { @@ -629,7 +667,8 @@ export const admitSubmission = mutation({ existing.acceptedAt === args.input.acceptedAt && existing.inputJson === args.input.inputJson && existing.chunksJson === (args.input.chunksJson ?? "[]") && - (existing.traceCarrierJson ?? undefined) === args.input.traceCarrierJson; + (existing.traceCarrierJson ?? undefined) === + args.input.traceCarrierJson; if (!exactMatch) { return { kind: "conflict" }; } @@ -642,14 +681,25 @@ export const admitSubmission = mutation({ }, }; } + if ( + correlatedTurn !== null && + correlatedTurn.submissionId !== existing.submissionId + ) { + await ctx.db.patch(correlatedTurn._id, { + error: undefined, + leaseExpiresAt: undefined, + leaseOwner: undefined, + status: "running", + submissionId: existing.submissionId, + }); + } return { kind: "submission", submission: toSubmissionRow(existing) }; } const all = await ctx.db.query("flueSubmissions").collect(); - const nextSequence = all.reduce( - (max, row) => (row.sequence > max ? row.sequence : max), - -1, - ) + 1; + const nextSequence = + all.reduce((max, row) => (row.sequence > max ? row.sequence : max), -1) + + 1; const now = Date.now(); const rowId = await ctx.db.insert("flueSubmissions", { acceptedAt: args.input.acceptedAt, @@ -671,6 +721,15 @@ export const admitSubmission = mutation({ if (created === null) { throw new Error("[flue] Failed to create submission row."); } + if (correlatedTurn !== null) { + await ctx.db.patch(correlatedTurn._id, { + error: undefined, + leaseExpiresAt: undefined, + leaseOwner: undefined, + status: "running", + submissionId: args.input.submissionId, + }); + } return { kind: "submission", submission: toSubmissionRow(created) }; }, }); @@ -681,7 +740,9 @@ export const markSubmissionCanonicalReady = mutation({ assertToken(args.token); const submission = await ctx.db .query("flueSubmissions") - .withIndex("by_submissionId", (q) => q.eq("submissionId", args.submissionId)) + .withIndex("by_submissionId", (q) => + q.eq("submissionId", args.submissionId) + ) .unique(); if (submission === null || submission.status !== "queued") { return null; @@ -708,9 +769,10 @@ export const claimSubmission = mutation({ handler: async (ctx, args) => { assertToken(args.token); const rows = (await ctx.db.query("flueSubmissions").collect()).sort( - (left, right) => left.sequence - right.sequence, + (left, right) => left.sequence - right.sequence ); - const submission = rows.find((row) => row.submissionId === args.submissionId) ?? null; + const submission = + rows.find((row) => row.submissionId === args.submissionId) ?? null; if ( submission === null || submission.status !== "queued" || @@ -722,7 +784,7 @@ export const claimSubmission = mutation({ (row) => row.sessionKey === submission.sessionKey && row.sequence < submission.sequence && - isUnsettled(row.status), + isUnsettled(row.status) ); if (hasEarlierUnsettled) { return null; @@ -762,7 +824,9 @@ export const markSubmissionInputApplied = mutation({ assertToken(args.token); const submission = await ctx.db .query("flueSubmissions") - .withIndex("by_submissionId", (q) => q.eq("submissionId", args.submissionId)) + .withIndex("by_submissionId", (q) => + q.eq("submissionId", args.submissionId) + ) .unique(); if ( submission === null || @@ -793,7 +857,9 @@ export const requestSubmissionRecovery = mutation({ assertToken(args.token); const submission = await ctx.db .query("flueSubmissions") - .withIndex("by_submissionId", (q) => q.eq("submissionId", args.submissionId)) + .withIndex("by_submissionId", (q) => + q.eq("submissionId", args.submissionId) + ) .unique(); if ( submission === null || @@ -817,7 +883,7 @@ export const requestSessionAbort = mutation({ const rows = (await ctx.db.query("flueSubmissions").collect()).filter( (row) => row.sessionKey === args.sessionKey && - (row.status === "queued" || row.status === "running"), + (row.status === "queued" || row.status === "running") ); const now = Date.now(); for (const row of rows) { @@ -835,7 +901,9 @@ export const requeueSubmissionBeforeInputApplied = mutation({ assertToken(args.token); const submission = await ctx.db .query("flueSubmissions") - .withIndex("by_submissionId", (q) => q.eq("submissionId", args.submissionId)) + .withIndex("by_submissionId", (q) => + q.eq("submissionId", args.submissionId) + ) .unique(); if ( submission === null || @@ -870,7 +938,9 @@ export const reserveSubmissionSettlement = mutation({ assertToken(args.token); const submission = await ctx.db .query("flueSubmissions") - .withIndex("by_submissionId", (q) => q.eq("submissionId", args.submissionId)) + .withIndex("by_submissionId", (q) => + q.eq("submissionId", args.submissionId) + ) .unique(); if (submission === null) { return null; @@ -901,12 +971,19 @@ export const reserveSubmissionSettlement = mutation({ }); export const finalizeSubmissionSettlement = mutation({ - args: { ...tokenArgs, attemptId: v.string(), recordId: v.string(), submissionId: v.string() }, + args: { + ...tokenArgs, + attemptId: v.string(), + recordId: v.string(), + submissionId: v.string(), + }, handler: async (ctx, args) => { assertToken(args.token); const submission = await ctx.db .query("flueSubmissions") - .withIndex("by_submissionId", (q) => q.eq("submissionId", args.submissionId)) + .withIndex("by_submissionId", (q) => + q.eq("submissionId", args.submissionId) + ) .unique(); if ( submission === null || @@ -940,7 +1017,9 @@ export const settleSubmission = mutation({ assertToken(args.token); const submission = await ctx.db .query("flueSubmissions") - .withIndex("by_submissionId", (q) => q.eq("submissionId", args.submissionId)) + .withIndex("by_submissionId", (q) => + q.eq("submissionId", args.submissionId) + ) .unique(); if ( submission === null || @@ -968,7 +1047,7 @@ export const insertAttemptMarker = mutation({ const existing = await ctx.db .query("flueAttemptMarkers") .withIndex("by_submissionId_and_attemptId", (q) => - q.eq("submissionId", args.submissionId).eq("attemptId", args.attemptId), + q.eq("submissionId", args.submissionId).eq("attemptId", args.attemptId) ) .unique(); if (existing !== null) { @@ -994,7 +1073,7 @@ export const deleteAttemptMarker = mutation({ const existing = await ctx.db .query("flueAttemptMarkers") .withIndex("by_submissionId_and_attemptId", (q) => - q.eq("submissionId", args.submissionId).eq("attemptId", args.attemptId), + q.eq("submissionId", args.submissionId).eq("attemptId", args.attemptId) ) .collect(); for (const row of existing) { @@ -1014,7 +1093,11 @@ export const listAttemptMarkers = query({ }); export const renewLeases = mutation({ - args: { ...tokenArgs, ownerId: v.string(), submissionIds: v.array(v.string()) }, + args: { + ...tokenArgs, + ownerId: v.string(), + submissionIds: v.array(v.string()), + }, handler: async (ctx, args) => { assertToken(args.token); const wanted = new Set(args.submissionIds); @@ -1046,7 +1129,7 @@ export const listExpiredSubmissions = query({ (row) => row.status === "running" && row.leaseExpiresAt > 0 && - row.leaseExpiresAt < now, + row.leaseExpiresAt < now ) .sort((left, right) => left.sequence - right.sequence) .map(toSubmissionRow); @@ -1065,7 +1148,7 @@ export const createConversationStream = mutation({ if (existing !== null) { if (existing.identityJson !== identityJson) { throw new Error( - `[flue] Conversation stream "${args.path}" identity conflicts with the existing stream.`, + `[flue] Conversation stream "${args.path}" identity conflicts with the existing stream.` ); } return; @@ -1092,7 +1175,9 @@ export const acquireConversationProducer = mutation({ .withIndex("by_path", (q) => q.eq("path", args.path)) .unique(); if (stream === null) { - throw new Error(`[flue] Conversation stream "${args.path}" does not exist.`); + throw new Error( + `[flue] Conversation stream "${args.path}" does not exist.` + ); } const producerEpoch = stream.producerEpoch + 1; await ctx.db.patch(stream._id, { @@ -1120,7 +1205,7 @@ export const appendConversationBatch = mutation({ producerSequence: v.number(), recordsJson: v.string(), submission: v.optional( - v.object({ attemptId: v.string(), submissionId: v.string() }), + v.object({ attemptId: v.string(), submissionId: v.string() }) ), }, handler: async (ctx, args): Promise => { @@ -1128,7 +1213,7 @@ export const appendConversationBatch = mutation({ const records = safeJsonParse(args.recordsJson); if (records.length === 0) { throw new Error( - `[flue] Conversation stream "${args.path}" cannot append an empty canonical batch.`, + `[flue] Conversation stream "${args.path}" cannot append an empty canonical batch.` ); } const stream = await ctx.db @@ -1136,7 +1221,9 @@ export const appendConversationBatch = mutation({ .withIndex("by_path", (q) => q.eq("path", args.path)) .unique(); if (stream === null) { - throw new Error(`[flue] Conversation stream "${args.path}" does not exist.`); + throw new Error( + `[flue] Conversation stream "${args.path}" does not exist.` + ); } if ( stream.producerId !== args.producerId || @@ -1144,7 +1231,7 @@ export const appendConversationBatch = mutation({ stream.incarnation !== args.incarnation ) { throw new Error( - `[flue] Conversation stream "${args.path}" producer ownership is stale.`, + `[flue] Conversation stream "${args.path}" producer ownership is stale.` ); } const existing = await ctx.db @@ -1154,7 +1241,7 @@ export const appendConversationBatch = mutation({ .eq("path", args.path) .eq("producerId", args.producerId) .eq("producerEpoch", args.producerEpoch) - .eq("producerSequence", args.producerSequence), + .eq("producerSequence", args.producerSequence) ) .unique(); if (existing !== null) { @@ -1163,17 +1250,22 @@ export const appendConversationBatch = mutation({ existing.attemptId === args.submission?.attemptId; if (!sameSubmission || existing.recordsJson !== args.recordsJson) { throw new Error( - `[flue] Conversation stream "${args.path}" producer sequence has conflicting content.`, + `[flue] Conversation stream "${args.path}" producer sequence has conflicting content.` ); } return { appended: false, offset: existing.seq }; } if (stream.nextProducerSequence !== args.producerSequence) { throw new Error( - `[flue] Conversation stream "${args.path}" producer sequence is not the next expected value.`, + `[flue] Conversation stream "${args.path}" producer sequence is not the next expected value.` ); } - await assertSubmissionAuthorization(ctx, args.path, args.submission, args.recordsJson); + await assertSubmissionAuthorization( + ctx, + args.path, + args.submission, + args.recordsJson + ); const seq = stream.nextOffset; await ctx.db.insert("flueConversationBatches", { appendedAt: Date.now(), @@ -1190,12 +1282,28 @@ export const appendConversationBatch = mutation({ nextOffset: seq + 1, nextProducerSequence: stream.nextProducerSequence + 1, }); + // Projection is a disposable product view. Canonical persistence must win + // even if a future projector record shape is malformed. + try { + await projectConversationRecords( + ctx, + args.recordsJson, + args.submission?.submissionId + ); + } catch { + // The raw canonical batch above remains durable and replayable. + } return { appended: true, offset: seq }; }, }); export const readConversationBatches = query({ - args: { ...tokenArgs, afterOffset: v.number(), limit: v.number(), path: v.string() }, + args: { + ...tokenArgs, + afterOffset: v.number(), + limit: v.number(), + path: v.string(), + }, handler: async (ctx, args) => { assertToken(args.token); const stream = await ctx.db @@ -1209,10 +1317,12 @@ export const readConversationBatches = query({ upToDate: true, }; } - const rows = (await ctx.db - .query("flueConversationBatches") - .withIndex("by_path_and_seq", (q) => q.eq("path", args.path)) - .collect()) + const rows = ( + await ctx.db + .query("flueConversationBatches") + .withIndex("by_path_and_seq", (q) => q.eq("path", args.path)) + .collect() + ) .filter((row) => row.seq > args.afterOffset) .sort((left, right) => left.seq - right.seq); const page = rows.slice(0, args.limit); @@ -1305,7 +1415,12 @@ export const appendEvent = mutation({ }); export const appendEventOnce = mutation({ - args: { ...tokenArgs, dataJson: v.string(), key: v.string(), path: v.string() }, + args: { + ...tokenArgs, + dataJson: v.string(), + key: v.string(), + path: v.string(), + }, handler: async (ctx, args): Promise => { assertToken(args.token); const stream = await ctx.db @@ -1321,13 +1436,13 @@ export const appendEventOnce = mutation({ const existing = await ctx.db .query("flueEventEntries") .withIndex("by_path_and_onceKey", (q) => - q.eq("path", args.path).eq("onceKey", args.key), + q.eq("path", args.path).eq("onceKey", args.key) ) .unique(); if (existing !== null) { if (existing.dataJson !== args.dataJson) { throw new Error( - `[flue] Event key "${args.key}" already has a conflicting payload.`, + `[flue] Event key "${args.key}" already has a conflicting payload.` ); } return { appended: false, offset: existing.seq }; @@ -1346,7 +1461,12 @@ export const appendEventOnce = mutation({ }); export const readEvents = query({ - args: { ...tokenArgs, afterOffset: v.number(), limit: v.number(), path: v.string() }, + args: { + ...tokenArgs, + afterOffset: v.number(), + limit: v.number(), + path: v.string(), + }, handler: async (ctx, args) => { assertToken(args.token); const stream = await ctx.db @@ -1361,10 +1481,12 @@ export const readEvents = query({ upToDate: true, }; } - const rows = (await ctx.db - .query("flueEventEntries") - .withIndex("by_path_and_seq", (q) => q.eq("path", args.path)) - .collect()) + const rows = ( + await ctx.db + .query("flueEventEntries") + .withIndex("by_path_and_seq", (q) => q.eq("path", args.path)) + .collect() + ) .filter((row) => row.seq > args.afterOffset) .sort((left, right) => left.seq - right.seq); const page = rows.slice(0, args.limit); @@ -1507,7 +1629,8 @@ export const listRuns = query({ .filter( (row) => (args.status === undefined || row.status === args.status) && - (args.workflowName === undefined || row.workflowName === args.workflowName), + (args.workflowName === undefined || + row.workflowName === args.workflowName) ) .sort(compareRunPointerDesc) .filter((row) => { @@ -1540,7 +1663,9 @@ export const putAttachment = mutation({ const existing = await ctx.db .query("flueAttachments") .withIndex("by_streamPath_and_attachmentId", (q) => - q.eq("streamPath", args.streamPath).eq("attachmentId", args.attachment.id), + q + .eq("streamPath", args.streamPath) + .eq("attachmentId", args.attachment.id) ) .unique(); if (existing !== null) { @@ -1576,7 +1701,7 @@ export const getAttachment = query({ q .eq("streamPath", args.streamPath) .eq("conversationId", args.conversationId) - .eq("attachmentId", args.attachmentId), + .eq("attachmentId", args.attachmentId) ) .unique(); return attachment === null ? null : toAttachmentWire(attachment); diff --git a/packages/backend/convex/schema.ts b/packages/backend/convex/schema.ts index 0148125..625a27e 100644 --- a/packages/backend/convex/schema.ts +++ b/packages/backend/convex/schema.ts @@ -111,9 +111,11 @@ export default defineSchema({ leaseOwner: v.optional(v.string()), status: v.union( v.literal("queued"), - v.literal("processing"), + v.literal("dispatching"), + v.literal("running"), v.literal("completed"), - v.literal("failed") + v.literal("failed"), + v.literal("aborted") ), submissionId: v.optional(v.string()), }) @@ -121,7 +123,8 @@ export default defineSchema({ "conversationId", "clientRequestId", ]) - .index("by_status_and_leaseExpiresAt", ["status", "leaseExpiresAt"]), + .index("by_status_and_leaseExpiresAt", ["status", "leaseExpiresAt"]) + .index("by_submissionId", ["submissionId"]), conversationMessages: defineTable({ content: v.string(), conversationId: v.id("conversations"), diff --git a/packages/backend/convex/signalRouting.test.ts b/packages/backend/convex/signalRouting.test.ts index af42907..1b50729 100644 --- a/packages/backend/convex/signalRouting.test.ts +++ b/packages/backend/convex/signalRouting.test.ts @@ -42,7 +42,7 @@ describe("signalRouting", () => { clientRequestId: "request-1", conversationId, createdAt: 1, - status: "processing", + status: "running", }); const messageId = await ctx.db.insert("conversationMessages", { content: "Fix the deploy", diff --git a/packages/backend/convex/signalRouting.ts b/packages/backend/convex/signalRouting.ts index 5fed257..b543832 100644 --- a/packages/backend/convex/signalRouting.ts +++ b/packages/backend/convex/signalRouting.ts @@ -60,7 +60,7 @@ export const listEvidence = query({ continue; } const turn = await ctx.db.get(message.turnId); - if (!turn || !["processing", "completed"].includes(turn.status)) { + if (!turn || !["running", "completed"].includes(turn.status)) { continue; } const consumed = await ctx.db @@ -131,7 +131,7 @@ export const createSignal = mutation({ throw new ConvexError(`Source message not found: ${messageId}`); } const turn = await ctx.db.get(message.turnId); - if (!turn || !["processing", "completed"].includes(turn.status)) { + if (!turn || !["running", "completed"].includes(turn.status)) { throw new ConvexError(`Source message is not admitted: ${messageId}`); } messages.push(message);