puter/zopu-code
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: project durable agent conversations

Browse Source
This commit is contained in:
-Puter
2026-07-29 08:19:17 +05:30
parent 18eb150d7d
commit 3ffa1cfc7c
22 changed files with 800 additions and 309 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
packages/backend/convex/_generated/api.d.ts vendored
View File

@@ -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;

62
packages/backend/convex/conversationMessages.test.ts
View File

@@ -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",
});
});
});

103
packages/backend/convex/conversationMessages.ts
View File

@@ -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<boolean> => {
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, {

145
packages/backend/convex/conversationProjections.ts Normal file
View File

@@ -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<string, unknown>)[key];
return typeof value === "string" ? value : undefined;
};
const resolveContext = async (
ctx: MutationCtx,
submissionId: string | undefined
): Promise<ProjectionContext> => {
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<string, unknown>).error;
if (typeof value === "string") {
return value.slice(0, 2000);
}
if (typeof value === "object" && value !== null) {
const { message } = value as Record<string, unknown>;
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<void> => {
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<string, unknown>).result
: undefined;
const text =
typeof result === "object" && result !== null
? (result as Record<string, unknown>).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,
});
}
};

47
packages/backend/convex/fluePersistence.test.ts
View File

@@ -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, {

295
packages/backend/convex/fluePersistence.ts
View File

@@ -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<ConversationStreamIdentity>(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<SubmissionDoc | null> =>
ctx.db
.query("flueSubmissions")
@@ -388,14 +405,13 @@ const getSubmissionDoc = (
const getConversationStreamDoc = (
ctx: ReadCtx,
path: string,
path: string
): Promise<ConversationStreamDoc | null> =>
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<void> => {
const records = safeJsonParse<unknown[]>(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<ConversationStreamIdentity>(stream.identityJson);
const streamIdentity = safeJsonParse<ConversationStreamIdentity>(
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<AdmitSubmissionResponse> => {
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<AppendResult> => {
@@ -1128,7 +1213,7 @@ export const appendConversationBatch = mutation({
const records = safeJsonParse<unknown[]>(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<AppendResult> => {
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);

9
packages/backend/convex/schema.ts
View File

@@ -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"),

2
packages/backend/convex/signalRouting.test.ts
View File

@@ -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",

4
packages/backend/convex/signalRouting.ts
View File

@@ -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);