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

Slice 4 control-plane repairs, lint/catalog cleanup, Flue adapter fixes

Browse Source 
Convex control plane (Slices 1-4 prerequisites for Slice 5):
- Versioned Definition/Design persistence: revisions preserve history
  instead of deleting prior slices/approvals
- Fenced conversation turn queue: attempt-numbered leases prevent stale
  worker overwrites; expired turns reconciled by cron
- Fenced Run/Attempt claiming: only queued attempts from running Runs can
  be claimed; lease expiry checks on every finish/checkpoint
- Multi-slice progression: successful slice marks next slice ready instead
  of completing the whole Work; completed Runs blocked from retry
- Typed AttemptClassification in schema and resolver decisions table
- Durable resolver decisions for normal outcomes, cancellation, and
  lease-expiry recovery
- Persistent artifacts and delivery metadata with provenance, source
  revision, verification status, and controlled delivery transitions
- High-impact open questions block definition approval
- Question submission gated to defining/awaiting-approval states only
- Delivery status updates validate JSON metadata before persisting
- Planner failures recorded as durable blocked events
- Approval identity uses canonical tokenIdentifier

Primitives:
- decodeDefinition separates decode from validate
- Design validation enforces 1-4 slices with unique IDs and valid deps
- Artifact and delivery draft schemas with verified-revision binding
- Delivery status transition table

Package management:
- Consolidated shared deps into single catalog (hono, valibot, streamdown,
  @types/react, @types/node, @tailwindcss/*, react-native)
- Removed cross-version Hono boundary: flue() exported as Fetchable
- Deleted bun.lock and regenerated from clean catalog resolution

Lint/format:
- Removed .eslintignore; oxc uses native ignorePatterns in oxlint.config.ts
- Deleted redundant packages/backend/.oxlintrc.json
- Added repos/** and scripts/** to ignore patterns
- Convex-specific rule overrides for ES2022 target constraints
- Fixed all oxc errors in fluePersistence.ts and agents/src/db.ts
- Fixed all formatting across changed files
This commit is contained in:
-Puter
2026-07-28 02:53:05 +05:30
parent 0e32c35515
commit a8d946b6a9
43 changed files with 3066 additions and 1424 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
packages/backend/.oxlintrc.json
View File

@@ -1,10 +0,0 @@
{
"overrides": [
{
"files": ["convex/**/*.ts"],
"rules": {
"unicorn/filename-case": "off"
}
}
]
}

11
packages/backend/convex/README.md
View File

@@ -1,8 +1,6 @@
# Convex Backend
Convex is the only application API used by web, desktop, and mobile clients.
The active Slice 1 backend is intentionally limited to tenancy, projects,
conversation turns, Signals, Work, and Flue's required persistence adapter.
Convex is the only application API used by web, desktop, and mobile clients. The active backend covers the durable control plane through Slice 4: tenancy, projects, conversation turns, Signals, Work planning, approved Design slices, simulated Runs and Attempts, artifacts/delivery metadata, and Flue persistence.
## Domain relations
@@ -11,6 +9,9 @@ conversation turns, Signals, Work, and Flue's required persistence adapter.
- `conversations`, `conversationTurns`, `conversationMessages`, and `conversationAttachments` own chat admission and reactive history.
- `signals`, `signalConstraints`, and `signalSources` preserve structured intent and exact evidence.
- `works`, `signalWorkAttachments`, and `workEvents` own proposed Work and provenance.
- `workDefinitions`, `workQuestions`, and `workApprovals` own versioned outcome contracts and approvals.
- `designPackets` and `workSlices` preserve versioned implementation intent and executable slice order.
- `workRuns`, `workAttempts`, `workAttemptEvents`, and `resolverDecisions` own bounded simulated execution and recovery.
- `workArtifacts` and `workDeliveries` preserve evidence and delivery metadata without performing Git or sandbox work.
The `flue*` tables are infrastructure tables required by Flue's persistence
contract. They are deliberately separate from the product relations.
The `flue*` tables are infrastructure tables required by Flue's persistence contract. They are deliberately separate from the product relations.

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

@@ -1,5 +1,5 @@
import { convexTest } from "convex-test";
import { anyApi } from "convex/server";
import { anyApi, makeFunctionReference } from "convex/server";
import { describe, expect, test } from "vitest";
import schema from "./schema";
@@ -15,6 +15,33 @@ const api = anyApi;
const identityA = { tokenIdentifier: "https://convex.test|user-a" };
const identityB = { tokenIdentifier: "https://convex.test|user-b" };
const newTest = () => convexTest({ modules, schema });
const markProcessingRef = makeFunctionReference<
"mutation",
{ 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",
{
turnId: string;
attempt: number;
leaseOwner: string;
error: string;
retry: boolean;
},
boolean
>("conversationMessages:failTurn");
const ensureOrg = async (
t: ReturnType<typeof newTest>,
@@ -109,4 +136,66 @@ describe("conversationMessages", () => {
})
).rejects.toThrow(/Organization membership required/u);
});
test("fences stale turn attempts from overwriting a retry", async () => {
const t = newTest();
const organization = await ensureOrg(t, identityA);
const sent = await t
.withIdentity(identityA)
.mutation(api.conversationMessages.send, {
clientRequestId: "request-fenced",
images: [],
organizationId: organization._id,
rawText: "Keep only the current response",
});
expect(
await t.mutation(markProcessingRef, {
attempt: 1,
leaseOwner: "worker-1",
turnId: sent.turnId,
})
).toBe(true);
expect(
await t.mutation(failTurnRef, {
attempt: 1,
error: "retry",
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");
});
});

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

@@ -31,18 +31,34 @@ const getTurnRef = makeFunctionReference<
>("conversationMessages:getTurn");
const markProcessingRef = makeFunctionReference<
"mutation",
{ turnId: Id<"conversationTurns"> },
{
turnId: Id<"conversationTurns">;
attempt: number;
leaseOwner: string;
},
boolean
>("conversationMessages:markProcessing");
const completeTurnRef = makeFunctionReference<
"mutation",
{ turnId: Id<"conversationTurns">; submissionId: string; text: string },
null
{
turnId: Id<"conversationTurns">;
attempt: number;
leaseOwner: string;
submissionId: string;
text: string;
},
boolean
>("conversationMessages:completeTurn");
const failTurnRef = makeFunctionReference<
"mutation",
{ turnId: Id<"conversationTurns">; error: string; retry: boolean },
null
{
turnId: Id<"conversationTurns">;
attempt: number;
leaseOwner: string;
error: string;
retry: boolean;
},
boolean
>("conversationMessages:failTurn");
export const generateUploadUrl = mutation({
@@ -57,16 +73,16 @@ export const generateUploadUrl = mutation({
export const send = mutation({
args: {
organizationId: v.id("organizations"),
clientRequestId: v.string(),
rawText: v.string(),
images: v.array(
v.object({
storageId: v.id("_storage"),
filename: v.optional(v.string()),
mimeType: v.string(),
storageId: v.id("_storage"),
})
),
organizationId: v.id("organizations"),
rawText: v.string(),
},
handler: async (ctx, args) => {
await requireOrganizationMember(ctx, args.organizationId);
@@ -115,6 +131,7 @@ export const send = mutation({
.first();
const ordinal = (lastMessage?.ordinal ?? -1) + 1;
const turnId = await ctx.db.insert("conversationTurns", {
attemptNumber: 1,
clientRequestId: args.clientRequestId,
conversationId: conversation._id,
createdAt,
@@ -233,24 +250,49 @@ export const getTurn = internalQuery({
});
export const markProcessing = internalMutation({
args: { turnId: v.id("conversationTurns") },
args: {
attempt: v.number(),
leaseOwner: v.string(),
turnId: v.id("conversationTurns"),
},
handler: async (ctx, args): Promise<boolean> => {
const turn = await ctx.db.get(args.turnId);
if (!turn || turn.status === "completed") {
if (
!turn ||
turn.status !== "queued" ||
(turn.attemptNumber ?? 1) !== args.attempt
) {
return false;
}
await ctx.db.patch(turn._id, { error: undefined, status: "processing" });
await ctx.db.patch(turn._id, {
error: undefined,
leaseExpiresAt: Date.now() + 60_000,
leaseOwner: args.leaseOwner,
status: "processing",
});
return true;
},
});
export const completeTurn = internalMutation({
args: {
turnId: v.id("conversationTurns"),
attempt: v.number(),
leaseOwner: v.string(),
submissionId: v.string(),
text: v.string(),
turnId: v.id("conversationTurns"),
},
handler: async (ctx, args): Promise<null> => {
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) =>
@@ -263,29 +305,43 @@ export const completeTurn = internalMutation({
await ctx.db.patch(args.turnId, {
completedAt: Date.now(),
error: undefined,
leaseExpiresAt: undefined,
leaseOwner: undefined,
status: "completed",
submissionId: args.submissionId,
});
return null;
return true;
},
});
export const failTurn = internalMutation({
args: {
turnId: v.id("conversationTurns"),
attempt: v.number(),
error: v.string(),
leaseOwner: v.string(),
retry: v.boolean(),
turnId: v.id("conversationTurns"),
},
handler: async (ctx, args): Promise<null> => {
handler: async (ctx, args): Promise<boolean> => {
const turn = await ctx.db.get(args.turnId);
if (turn && turn.status !== "completed") {
await ctx.db.patch(turn._id, {
completedAt: args.retry ? undefined : Date.now(),
error: args.error,
status: args.retry ? "queued" : "failed",
});
if (
!turn ||
turn.status !== "processing" ||
(turn.attemptNumber ?? 1) !== args.attempt ||
turn.leaseOwner !== args.leaseOwner ||
(turn.leaseExpiresAt !== undefined && turn.leaseExpiresAt < Date.now())
) {
return false;
}
return null;
await ctx.db.patch(turn._id, {
attemptNumber: args.retry ? args.attempt + 1 : args.attempt,
completedAt: args.retry ? undefined : Date.now(),
error: args.error,
leaseExpiresAt: undefined,
leaseOwner: undefined,
status: args.retry ? "queued" : "failed",
});
return true;
},
});
@@ -298,12 +354,17 @@ const toBase64 = (buffer: ArrayBuffer): string => {
};
export const runTurn = internalAction({
args: { turnId: v.id("conversationTurns"), attempt: v.number() },
args: { attempt: v.number(), turnId: v.id("conversationTurns") },
handler: async (ctx, args): Promise<null> => {
const leaseOwner = `conversation:${args.turnId}:${args.attempt}`;
const turn = await ctx.runQuery(getTurnRef, { turnId: args.turnId });
if (
!turn ||
!(await ctx.runMutation(markProcessingRef, { turnId: args.turnId }))
!(await ctx.runMutation(markProcessingRef, {
attempt: args.attempt,
leaseOwner,
turnId: args.turnId,
}))
) {
return null;
}
@@ -358,6 +419,8 @@ export const runTurn = internalAction({
throw new Error(`Flue turn failed (${response.status})`);
}
await ctx.runMutation(completeTurnRef, {
attempt: args.attempt,
leaseOwner,
submissionId: payload.submissionId,
text: payload.result.text,
turnId: args.turnId,
@@ -365,7 +428,9 @@ export const runTurn = internalAction({
} catch (error) {
const retry = args.attempt < MAX_ATTEMPTS;
await ctx.runMutation(failTurnRef, {
attempt: args.attempt,
error: error instanceof Error ? error.message : String(error),
leaseOwner,
retry,
turnId: args.turnId,
});
@@ -379,3 +444,34 @@ export const runTurn = internalAction({
return null;
},
});
export const reconcileExpiredTurns = internalMutation({
args: {},
handler: async (ctx): Promise<{ reconciled: number }> => {
const expired = await ctx.db
.query("conversationTurns")
.withIndex("by_status_and_leaseExpiresAt", (q) =>
q.eq("status", "processing").lt("leaseExpiresAt", Date.now())
)
.collect();
for (const turn of expired) {
const attempt = turn.attemptNumber ?? 1;
const retry = attempt < MAX_ATTEMPTS;
await ctx.db.patch(turn._id, {
attemptNumber: retry ? attempt + 1 : attempt,
completedAt: retry ? undefined : Date.now(),
error: "Conversation worker lease expired",
leaseExpiresAt: undefined,
leaseOwner: undefined,
status: retry ? "queued" : "failed",
});
if (retry) {
await ctx.scheduler.runAfter(0, runTurnRef, {
attempt: attempt + 1,
turnId: turn._id,
});
}
}
return { reconciled: expired.length };
},
});

6
packages/backend/convex/convex.config.ts
View File

@@ -4,12 +4,12 @@ import { v } from "convex/values";
const app = defineApp({
env: {
NATIVE_APP_URL: v.optional(v.string()),
SITE_URL: v.string(),
FLUE_DB_TOKEN: v.string(),
FLUE_URL: v.optional(v.string()),
GITEA_URL: v.optional(v.string()),
GITEA_TOKEN: v.optional(v.string()),
GITEA_URL: v.optional(v.string()),
NATIVE_APP_URL: v.optional(v.string()),
SITE_URL: v.string(),
},
});
app.use(betterAuth);

10
packages/backend/convex/crons.ts
View File

@@ -7,6 +7,11 @@ const reconcileRef = makeFunctionReference<
Record<string, never>,
{ reconciled: number } | null
>("workExecution:reconcileExpiredAttempts");
const reconcileConversationTurnsRef = makeFunctionReference<
"mutation",
Record<string, never>,
{ reconciled: number }
>("conversationMessages:reconcileExpiredTurns");
const crons = cronJobs();
@@ -15,5 +20,10 @@ crons.interval(
{ seconds: 30 },
reconcileRef
);
crons.interval(
"reconcile expired conversation turns",
{ seconds: 30 },
reconcileConversationTurnsRef
);
export default crons;

142
packages/backend/convex/fluePersistence.test.ts Normal file
View File

@@ -0,0 +1,142 @@
import { env } from "@code/env/convex";
import { convexTest } from "convex-test";
import { anyApi } from "convex/server";
import { describe, expect, test } from "vitest";
import schema from "./schema";
declare global {
interface ImportMeta {
readonly glob: (pattern: string) => Record<string, () => Promise<unknown>>;
}
}
const modules = import.meta.glob("./**/*.ts");
const api = anyApi;
const token = env.FLUE_DB_TOKEN;
describe("Flue Convex persistence", () => {
test("admits idempotently and claims only the session head", async () => {
const t = convexTest({ modules, schema });
const firstInput = {
acceptedAt: 1,
inputJson: '{"message":"first"}',
kind: "dispatch" as const,
sessionKey: "agent/instance/default",
submissionId: "dispatch-1",
};
const first = await t.mutation(api.fluePersistence.admitSubmission, {
input: firstInput,
token,
});
const replay = await t.mutation(api.fluePersistence.admitSubmission, {
input: firstInput,
token,
});
expect(first.kind).toBe("submission");
expect(replay.kind).toBe("retained_receipt");
await t.mutation(api.fluePersistence.admitSubmission, {
input: {
...firstInput,
acceptedAt: 2,
inputJson: '{"message":"second"}',
submissionId: "dispatch-2",
},
token,
});
await t.mutation(api.fluePersistence.markSubmissionCanonicalReady, {
submissionId: "dispatch-1",
token,
});
await t.mutation(api.fluePersistence.markSubmissionCanonicalReady, {
submissionId: "dispatch-2",
token,
});
const secondClaim = await t.mutation(api.fluePersistence.claimSubmission, {
attemptId: "attempt-2",
leaseExpiresAt: 100,
ownerId: "owner",
submissionId: "dispatch-2",
token,
});
expect(secondClaim).toBeNull();
const firstClaim = await t.mutation(api.fluePersistence.claimSubmission, {
attemptId: "attempt-1",
leaseExpiresAt: 100,
ownerId: "owner",
submissionId: "dispatch-1",
token,
});
expect(firstClaim).toMatchObject({
attemptId: "attempt-1",
status: "running",
});
});
test("fences stale conversation producers and conflicting attachments", async () => {
const t = convexTest({ modules, schema });
await t.mutation(api.fluePersistence.createConversationStream, {
identity: { agentName: "work-planner", instanceId: "org-1" },
path: "agents/work-planner/org-1",
token,
});
const first = await t.mutation(
api.fluePersistence.acquireConversationProducer,
{
path: "agents/work-planner/org-1",
producerId: "producer-1",
token,
}
);
await t.mutation(api.fluePersistence.acquireConversationProducer, {
path: "agents/work-planner/org-1",
producerId: "producer-2",
token,
});
await expect(
t.mutation(api.fluePersistence.appendConversationBatch, {
incarnation: first.incarnation,
path: "agents/work-planner/org-1",
producerEpoch: first.producerEpoch,
producerId: "producer-1",
producerSequence: 0,
recordsJson: '[{"id":"record-1","type":"message"}]',
token,
})
).rejects.toThrow(/producer ownership is stale/u);
const attachment = {
digest: "digest",
id: "attachment-1",
mimeType: "text/plain",
size: 3,
};
const inserted = await t.mutation(api.fluePersistence.putAttachment, {
attachment,
bytes: new TextEncoder().encode("one").buffer,
conversationId: "conversation-1",
streamPath: "agents/work-planner/org-1",
token,
});
const replay = await t.mutation(api.fluePersistence.putAttachment, {
attachment,
bytes: new TextEncoder().encode("one").buffer,
conversationId: "conversation-1",
streamPath: "agents/work-planner/org-1",
token,
});
const conflict = await t.mutation(api.fluePersistence.putAttachment, {
attachment,
bytes: new TextEncoder().encode("two").buffer,
conversationId: "conversation-1",
streamPath: "agents/work-planner/org-1",
token,
});
expect([inserted, replay, conflict]).toEqual([
"inserted",
"existing",
"conflict",
]);
});
});

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

File diff suppressed because it is too large Load Diff

6
packages/backend/convex/healthCheck.ts
View File

@@ -1,7 +1,7 @@
import { query } from "./_generated/server";
export const get = query({
handler: async () => {
return "OK";
},
handler: () =>
"OK"
,
});

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

@@ -23,7 +23,7 @@ const ID_B = "https://convex.test|user-b";
const identityA = { tokenIdentifier: ID_A };
const identityB = { tokenIdentifier: ID_B };
const newTest = () => convexTest({ schema, modules });
const newTest = () => convexTest({ modules, schema });
describe("organizations", () => {
test("first ensure creates one organization and owner membership", async () => {

28
packages/backend/convex/projects.ts
View File

@@ -1,8 +1,10 @@
import {
decodePublicGitImportResult,
preparePublicGitSource,
type ProjectImportOutcome,
type ProjectView,
} from "@code/primitives/project";
import type {
ProjectImportOutcome,
ProjectView,
} from "@code/primitives/project";
import { ConvexError, v } from "convex/values";
import { Effect } from "effect";
@@ -54,18 +56,11 @@ const toProjectView = async (
export const persistPublicGitImport = internalMutation({
args: {
userId: v.string(),
source: v.object({
host: v.string(),
projectName: v.string(),
repositoryPath: v.string(),
normalizedUrl: v.string(),
url: v.string(),
}),
remote: v.object({
defaultBranch: v.optional(v.string()),
documents: v.array(
v.object({
content: v.string(),
kind: v.union(
v.literal("readme"),
v.literal("agents"),
@@ -75,11 +70,18 @@ export const persistPublicGitImport = internalMutation({
v.literal("tech")
),
path: v.string(),
content: v.string(),
})
),
warnings: v.array(v.object({ path: v.string(), message: v.string() })),
warnings: v.array(v.object({ message: v.string(), path: v.string() })),
}),
source: v.object({
host: v.string(),
normalizedUrl: v.string(),
projectName: v.string(),
repositoryPath: v.string(),
url: v.string(),
}),
userId: v.string(),
},
handler: async (ctx, args): Promise<ProjectImportOutcome> => {
const organization = await ctx.db
@@ -106,9 +108,9 @@ export const persistPublicGitImport = internalMutation({
await ctx.db.patch(projectId, {
defaultBranch: args.remote.defaultBranch,
name: args.source.projectName,
repositoryPath: args.source.repositoryPath,
sourceHost: args.source.host,
sourceUrl: args.source.url,
repositoryPath: args.source.repositoryPath,
updatedAt: timestamp,
});
const oldDocuments = await ctx.db

10
packages/backend/convex/publicGit.ts
View File

@@ -1,11 +1,5 @@
import {
CONTEXT_KINDS,
contextPathForKind,
MAX_CONTEXT_DOCUMENT_CHARACTERS,
type PreparedPublicGitSource,
type PublicGitImportResult,
type RepositoryContextDocument,
} from "@code/primitives/project";
import { CONTEXT_KINDS, contextPathForKind, MAX_CONTEXT_DOCUMENT_CHARACTERS } from '@code/primitives/project';
import type { PreparedPublicGitSource, PublicGitImportResult, RepositoryContextDocument } from '@code/primitives/project';
const GITHUB_HOST = "github.com";
const GITEA_HOST = "git.openputer.com";

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

@@ -1,32 +1,59 @@
import { defineSchema, defineTable } from "convex/server";
import { v } from "convex/values";
/* eslint-disable sort-keys */
const attemptClassification = v.union(
v.literal("Succeeded"),
v.literal("RetryableFailure"),
v.literal("NeedsInput"),
v.literal("Blocked"),
v.literal("VerificationFailed"),
v.literal("BudgetExhausted"),
v.literal("Cancelled"),
v.literal("PermanentFailure")
);
const workStatus = v.union(
v.literal("proposed"),
v.literal("defining"),
v.literal("awaiting-definition-approval"),
v.literal("designing"),
v.literal("awaiting-design-approval"),
v.literal("ready"),
v.literal("executing"),
v.literal("needs-input"),
v.literal("blocked"),
v.literal("completed"),
v.literal("failed"),
v.literal("cancelled")
);
export default defineSchema({
organizations: defineTable({
name: v.string(),
kind: v.union(v.literal("personal"), v.literal("team")),
createdBy: v.string(),
createdAt: v.number(),
createdBy: v.string(),
kind: v.union(v.literal("personal"), v.literal("team")),
name: v.string(),
}).index("by_createdBy_and_kind", ["createdBy", "kind"]),
organizationMembers: defineTable({
organizationId: v.id("organizations"),
userId: v.string(),
role: v.union(v.literal("owner"), v.literal("member")),
createdAt: v.number(),
organizationId: v.id("organizations"),
role: v.union(v.literal("owner"), v.literal("member")),
userId: v.string(),
})
.index("by_userId", ["userId"])
.index("by_organizationId", ["organizationId"])
.index("by_organizationId_and_userId", ["organizationId", "userId"]),
projects: defineTable({
organizationId: v.id("organizations"),
name: v.string(),
description: v.optional(v.string()),
sourceUrl: v.string(),
normalizedSourceUrl: v.string(),
sourceHost: v.string(),
repositoryPath: v.string(),
defaultBranch: v.optional(v.string()),
createdAt: v.number(),
defaultBranch: v.optional(v.string()),
description: v.optional(v.string()),
name: v.string(),
normalizedSourceUrl: v.string(),
organizationId: v.id("organizations"),
repositoryPath: v.string(),
sourceHost: v.string(),
sourceUrl: v.string(),
updatedAt: v.number(),
})
.index("by_organizationId_and_createdAt", ["organizationId", "createdAt"])
@@ -35,7 +62,8 @@ export default defineSchema({
"normalizedSourceUrl",
]),
projectContextDocuments: defineTable({
projectId: v.id("projects"),
content: v.string(),
createdAt: v.number(),
kind: v.union(
v.literal("readme"),
v.literal("agents"),
@@ -44,20 +72,25 @@ export default defineSchema({
v.literal("design"),
v.literal("tech")
),
path: v.string(),
content: v.string(),
origin: v.literal("repository"),
path: v.string(),
projectId: v.id("projects"),
sourceUrl: v.string(),
createdAt: v.number(),
updatedAt: v.number(),
}).index("by_projectId_and_path", ["projectId", "path"]),
conversations: defineTable({
organizationId: v.id("organizations"),
createdAt: v.number(),
organizationId: v.id("organizations"),
}).index("by_organizationId", ["organizationId"]),
conversationTurns: defineTable({
conversationId: v.id("conversations"),
attemptNumber: v.optional(v.number()),
clientRequestId: v.string(),
completedAt: v.optional(v.number()),
conversationId: v.id("conversations"),
createdAt: v.number(),
error: v.optional(v.string()),
leaseExpiresAt: v.optional(v.number()),
leaseOwner: v.optional(v.string()),
status: v.union(
v.literal("queued"),
v.literal("processing"),
@@ -65,100 +98,86 @@ export default defineSchema({
v.literal("failed")
),
submissionId: v.optional(v.string()),
error: v.optional(v.string()),
createdAt: v.number(),
completedAt: v.optional(v.number()),
}).index("by_conversationId_and_clientRequestId", [
"conversationId",
"clientRequestId",
]),
})
.index("by_conversationId_and_clientRequestId", [
"conversationId",
"clientRequestId",
])
.index("by_status_and_leaseExpiresAt", ["status", "leaseExpiresAt"]),
conversationMessages: defineTable({
conversationId: v.id("conversations"),
turnId: v.id("conversationTurns"),
role: v.union(v.literal("user"), v.literal("assistant")),
content: v.string(),
ordinal: v.number(),
conversationId: v.id("conversations"),
createdAt: v.number(),
ordinal: v.number(),
role: v.union(v.literal("user"), v.literal("assistant")),
turnId: v.id("conversationTurns"),
})
.index("by_conversationId_and_ordinal", ["conversationId", "ordinal"])
.index("by_turnId_and_role", ["turnId", "role"]),
conversationAttachments: defineTable({
messageId: v.id("conversationMessages"),
storageId: v.id("_storage"),
filename: v.optional(v.string()),
mimeType: v.string(),
createdAt: v.number(),
filename: v.optional(v.string()),
messageId: v.id("conversationMessages"),
mimeType: v.string(),
storageId: v.id("_storage"),
}).index("by_messageId", ["messageId"]),
signals: defineTable({
organizationId: v.id("organizations"),
projectId: v.id("projects"),
conversationId: v.id("conversations"),
sourceKey: v.string(),
title: v.string(),
summary: v.string(),
desiredOutcome: v.string(),
processedByAgentName: v.string(),
processedByAgentInstanceId: v.string(),
createdAt: v.number(),
desiredOutcome: v.string(),
organizationId: v.id("organizations"),
processedByAgentInstanceId: v.string(),
processedByAgentName: v.string(),
projectId: v.id("projects"),
sourceKey: v.string(),
summary: v.string(),
title: v.string(),
})
.index("by_organization_and_createdAt", ["organizationId", "createdAt"])
.index("by_organization_and_sourceKey", ["organizationId", "sourceKey"])
.index("by_project_and_createdAt", ["projectId", "createdAt"]),
signalConstraints: defineTable({
signalId: v.id("signals"),
ordinal: v.number(),
signalId: v.id("signals"),
value: v.string(),
}).index("by_signalId_and_ordinal", ["signalId", "ordinal"]),
signalSources: defineTable({
signalId: v.id("signals"),
messageId: v.id("conversationMessages"),
ordinal: v.number(),
rawTextSnapshot: v.string(),
signalId: v.id("signals"),
sourceCreatedAt: v.number(),
})
.index("by_signalId_and_ordinal", ["signalId", "ordinal"])
.index("by_messageId", ["messageId"]),
works: defineTable({
createdAt: v.number(),
definitionApprovalVersion: v.optional(v.number()),
definitionVersion: v.optional(v.number()),
designApprovalVersion: v.optional(v.number()),
designVersion: v.optional(v.number()),
objective: v.string(),
organizationId: v.id("organizations"),
projectId: v.id("projects"),
status: workStatus,
title: v.string(),
objective: v.string(),
status: v.union(
v.literal("proposed"),
v.literal("defining"),
v.literal("awaiting-definition-approval"),
v.literal("designing"),
v.literal("awaiting-design-approval"),
v.literal("ready"),
v.literal("executing"),
v.literal("needs-input"),
v.literal("blocked"),
v.literal("completed"),
v.literal("failed"),
v.literal("cancelled")
),
definitionVersion: v.optional(v.number()),
definitionApprovalVersion: v.optional(v.number()),
designVersion: v.optional(v.number()),
designApprovalVersion: v.optional(v.number()),
createdAt: v.number(),
updatedAt: v.number(),
})
.index("by_project_and_createdAt", ["projectId", "createdAt"])
.index("by_organization_and_createdAt", ["organizationId", "createdAt"]),
signalWorkAttachments: defineTable({
createdAt: v.number(),
signalId: v.id("signals"),
workId: v.id("works"),
createdAt: v.number(),
})
.index("by_signal", ["signalId"])
.index("by_work", ["workId"])
.index("by_signal_and_work", ["signalId", "workId"]),
workEvents: defineTable({
workId: v.id("works"),
signalId: v.optional(v.id("signals")),
createdAt: v.number(),
idempotencyKey: v.string(),
kind: v.union(
v.literal("work.proposed"),
v.literal("signal.attached"),
@@ -167,6 +186,7 @@ export default defineSchema({
v.literal("definition.revised"),
v.literal("definition.approved"),
v.literal("definition.invalidated"),
v.literal("question.created"),
v.literal("question.answered"),
v.literal("question.withdrawn"),
v.literal("design.requested"),
@@ -174,24 +194,33 @@ export default defineSchema({
v.literal("design.revised"),
v.literal("design.approved"),
v.literal("design.invalidated"),
v.literal("planner.failed"),
v.literal("slice.started"),
v.literal("slice.completed"),
v.literal("slice.ready"),
v.literal("run.started"),
v.literal("run.completed"),
v.literal("run.cancelled"),
v.literal("attempt.claimed"),
v.literal("attempt.event"),
v.literal("attempt.completed"),
v.literal("attempt.reconciled")
v.literal("attempt.reconciled"),
v.literal("resolver.decided"),
v.literal("artifact.recorded"),
v.literal("delivery.recorded"),
v.literal("delivery.updated")
),
referenceId: v.optional(v.string()),
payloadJson: v.optional(v.string()),
idempotencyKey: v.string(),
createdAt: v.number(),
referenceId: v.optional(v.string()),
signalId: v.optional(v.id("signals")),
workId: v.id("works"),
})
.index("by_work_and_createdAt", ["workId", "createdAt"])
.index("by_work_and_idempotencyKey", ["workId", "idempotencyKey"]),
workDefinitions: defineTable({
workId: v.id("works"),
version: v.number(),
createdAt: v.number(),
createdBy: v.string(),
payloadJson: v.string(),
risk: v.union(v.literal("low"), v.literal("medium"), v.literal("high")),
status: v.union(
@@ -199,42 +228,55 @@ export default defineSchema({
v.literal("current"),
v.literal("superseded")
),
createdBy: v.string(),
createdAt: v.number(),
version: v.number(),
workId: v.id("works"),
})
.index("by_work_and_version", ["workId", "version"])
.index("by_work_and_status", ["workId", "status"]),
workQuestions: defineTable({
workId: v.id("works"),
definitionVersion: v.number(),
questionId: v.string(),
prompt: v.string(),
impact: v.union(v.literal("low"), v.literal("medium"), v.literal("high")),
recommendation: v.optional(v.string()),
alternativesJson: v.string(),
answer: v.optional(v.string()),
createdAt: v.number(),
definitionVersion: v.number(),
impact: v.union(v.literal("low"), v.literal("medium"), v.literal("high")),
prompt: v.string(),
questionId: v.string(),
recommendation: v.optional(v.string()),
status: v.union(
v.literal("open"),
v.literal("answered"),
v.literal("withdrawn")
),
answer: v.optional(v.string()),
createdAt: v.number(),
}).index("by_work_and_definitionVersion", ["workId", "definitionVersion"]),
workId: v.id("works"),
})
.index("by_workId_and_definitionVersion", ["workId", "definitionVersion"])
.index("by_workId_and_definitionVersion_and_questionId", [
"workId",
"definitionVersion",
"questionId",
]),
workApprovals: defineTable({
workId: v.id("works"),
kind: v.union(v.literal("definition"), v.literal("design")),
approvedAt: v.number(),
approvedBy: v.string(),
definitionVersion: v.number(),
designVersion: v.optional(v.number()),
approvedBy: v.string(),
approvedAt: v.number(),
kind: v.union(v.literal("definition"), v.literal("design")),
status: v.union(v.literal("active"), v.literal("invalidated")),
}).index("by_work_and_kind", ["workId", "kind"]),
workId: v.id("works"),
})
.index("by_workId_and_kind", ["workId", "kind"])
.index("by_workId_and_kind_and_definitionVersion_and_designVersion", [
"workId",
"kind",
"definitionVersion",
"designVersion",
]),
designPackets: defineTable({
workId: v.id("works"),
version: v.number(),
createdAt: v.number(),
createdBy: v.string(),
definitionVersion: v.number(),
payloadJson: v.string(),
status: v.union(
@@ -242,19 +284,18 @@ export default defineSchema({
v.literal("current"),
v.literal("superseded")
),
createdBy: v.string(),
createdAt: v.number(),
version: v.number(),
workId: v.id("works"),
}).index("by_work_and_version", ["workId", "version"]),
workSlices: defineTable({
workId: v.id("works"),
createdAt: v.optional(v.number()),
designVersion: v.number(),
sliceId: v.string(),
ordinal: v.number(),
title: v.string(),
objective: v.string(),
observableBehavior: v.string(),
ordinal: v.number(),
payloadJson: v.string(),
sliceId: v.string(),
status: v.union(
v.literal("planned"),
v.literal("ready"),
@@ -262,17 +303,22 @@ export default defineSchema({
v.literal("completed"),
v.literal("blocked")
),
}).index("by_work_and_designVersion", ["workId", "designVersion"]),
title: v.string(),
workId: v.id("works"),
})
.index("by_workId_and_designVersion", ["workId", "designVersion"])
.index("by_workId_and_designVersion_and_sliceId", [
"workId",
"designVersion",
"sliceId",
]),
workRuns: defineTable({
workId: v.id("works"),
sliceId: v.optional(v.string()),
status: v.union(
v.literal("ready"),
v.literal("running"),
v.literal("terminal"),
v.literal("cancelled")
),
createdAt: v.number(),
designVersion: v.optional(v.number()),
endedAt: v.optional(v.number()),
kitId: v.string(),
kitVersion: v.string(),
scenario: v.union(
v.literal("success"),
v.literal("transient-failure-then-success"),
@@ -280,42 +326,140 @@ export default defineSchema({
v.literal("permanent-failure"),
v.literal("cancelled")
),
kitId: v.string(),
kitVersion: v.string(),
createdAt: v.number(),
sliceId: v.optional(v.string()),
sliceRowId: v.optional(v.id("workSlices")),
startedAt: v.optional(v.number()),
endedAt: v.optional(v.number()),
terminalClassification: v.optional(v.string()),
status: v.union(
v.literal("ready"),
v.literal("running"),
v.literal("terminal"),
v.literal("cancelled")
),
terminalClassification: v.optional(attemptClassification),
terminalSummary: v.optional(v.string()),
workId: v.id("works"),
}).index("by_work_and_createdAt", ["workId", "createdAt"]),
workAttempts: defineTable({
runId: v.id("workRuns"),
workId: v.id("works"),
classification: v.optional(attemptClassification),
endedAt: v.optional(v.number()),
leaseExpiresAt: v.optional(v.number()),
leaseOwner: v.optional(v.string()),
number: v.number(),
runId: v.id("workRuns"),
startedAt: v.optional(v.number()),
status: v.union(
v.literal("queued"),
v.literal("claimed"),
v.literal("running"),
v.literal("terminal")
),
leaseOwner: v.optional(v.string()),
leaseExpiresAt: v.optional(v.number()),
startedAt: v.optional(v.number()),
endedAt: v.optional(v.number()),
classification: v.optional(v.string()),
summary: v.optional(v.string()),
}).index("by_run_and_number", ["runId", "number"]),
workId: v.id("works"),
})
.index("by_runId_and_number", ["runId", "number"])
.index("by_status_and_leaseExpiresAt", ["status", "leaseExpiresAt"]),
workAttemptEvents: defineTable({
attemptId: v.id("workAttempts"),
sequence: v.number(),
kind: v.string(),
message: v.string(),
metadataJson: v.string(),
occurredAt: v.number(),
sequence: v.number(),
}).index("by_attempt_and_sequence", ["attemptId", "sequence"]),
resolverDecisions: defineTable({
attemptId: v.id("workAttempts"),
attemptNumber: v.number(),
classification: attemptClassification,
createdAt: v.number(),
decision: v.union(v.literal("retry"), v.literal("terminal")),
resultingWorkStatus: v.optional(workStatus),
runId: v.id("workRuns"),
summary: v.string(),
workId: v.id("works"),
})
.index("by_runId_and_attemptNumber", ["runId", "attemptNumber"])
.index("by_attemptId", ["attemptId"]),
workArtifacts: defineTable({
attemptId: v.optional(v.id("workAttempts")),
contentHash: v.optional(v.string()),
createdAt: v.number(),
designVersion: v.optional(v.number()),
environmentId: v.optional(v.string()),
idempotencyKey: v.string(),
kind: v.union(
v.literal("definition"),
v.literal("design"),
v.literal("diff"),
v.literal("test-report"),
v.literal("verification-report"),
v.literal("runtime-log"),
v.literal("screenshot"),
v.literal("video"),
v.literal("commit"),
v.literal("branch"),
v.literal("pull-request"),
v.literal("preview"),
v.literal("deployment"),
v.literal("other")
),
metadataJson: v.string(),
organizationId: v.id("organizations"),
producer: v.string(),
projectId: v.id("projects"),
provenanceJson: v.string(),
runId: v.optional(v.id("workRuns")),
sliceId: v.optional(v.string()),
sourceRevision: v.optional(v.string()),
title: v.string(),
uri: v.optional(v.string()),
verificationStatus: v.union(
v.literal("unverified"),
v.literal("verified"),
v.literal("rejected")
),
workId: v.id("works"),
})
.index("by_workId_and_createdAt", ["workId", "createdAt"])
.index("by_workId_and_idempotencyKey", ["workId", "idempotencyKey"])
.index("by_runId_and_createdAt", ["runId", "createdAt"]),
workDeliveries: defineTable({
artifactId: v.optional(v.id("workArtifacts")),
createdAt: v.number(),
externalId: v.string(),
idempotencyKey: v.string(),
kind: v.union(
v.literal("branch"),
v.literal("commit"),
v.literal("pull-request"),
v.literal("preview"),
v.literal("deployment")
),
metadataJson: v.string(),
organizationId: v.id("organizations"),
projectId: v.id("projects"),
provider: v.string(),
sourceRevision: v.string(),
status: v.union(
v.literal("recorded"),
v.literal("ready"),
v.literal("approved"),
v.literal("delivered"),
v.literal("failed"),
v.literal("cancelled")
),
target: v.string(),
updatedAt: v.number(),
url: v.optional(v.string()),
workId: v.id("works"),
})
.index("by_workId_and_createdAt", ["workId", "createdAt"])
.index("by_workId_and_idempotencyKey", ["workId", "idempotencyKey"]),
// -----------------------------------------------------------------
// Flue persistence stores (schema/format version 4).
// -----------------------------------------------------------------
@@ -382,38 +526,38 @@ export default defineSchema({
// Durable evidence that a submission attempt started and has not yet
// settled. Append-once by (submissionId, attemptId).
flueAttemptMarkers: defineTable({
submissionId: v.string(),
attemptId: v.string(),
createdAt: v.number(),
submissionId: v.string(),
})
.index("by_submissionId_and_attemptId", ["submissionId", "attemptId"])
.index("by_submissionId", ["submissionId"]),
// Conversation-stream metadata: one row per stream path.
flueConversationStreams: defineTable({
path: v.string(),
identityJson: v.string(),
incarnation: v.string(),
producerId: v.optional(v.string()),
producerEpoch: v.number(),
nextProducerSequence: v.number(),
nextOffset: v.number(),
closed: v.boolean(),
createdAt: v.number(),
identityJson: v.string(),
incarnation: v.string(),
nextOffset: v.number(),
nextProducerSequence: v.number(),
path: v.string(),
producerEpoch: v.number(),
producerId: v.optional(v.string()),
}).index("by_path", ["path"]),
// Conversation-stream batches: one row per appended batch. Offset is
// 0-based; `seq` is the row's position in the stream.
flueConversationBatches: defineTable({
appendedAt: v.number(),
attemptId: v.optional(v.string()),
path: v.string(),
seq: v.number(),
producerId: v.string(),
producerEpoch: v.number(),
producerId: v.string(),
producerSequence: v.number(),
recordsJson: v.string(),
seq: v.number(),
submissionId: v.optional(v.string()),
attemptId: v.optional(v.string()),
appendedAt: v.number(),
})
.index("by_path_and_seq", ["path", "seq"])
.index("by_path_producer_epoch_producerSequence", [
@@ -425,41 +569,41 @@ export default defineSchema({
// Event-stream metadata: one row per stream path.
flueEventStreams: defineTable({
path: v.string(),
nextSeq: v.number(),
closed: v.boolean(),
createdAt: v.number(),
nextSeq: v.number(),
path: v.string(),
}).index("by_path", ["path"]),
// Event-stream entries: one row per appended event. `onceKey` carries the
// idempotency key for `appendEventOnce` and is null for plain appends.
flueEventEntries: defineTable({
appendedAt: v.number(),
dataJson: v.string(),
onceKey: v.optional(v.string()),
path: v.string(),
seq: v.number(),
onceKey: v.optional(v.string()),
dataJson: v.string(),
appendedAt: v.number(),
})
.index("by_path_and_seq", ["path", "seq"])
.index("by_path_and_onceKey", ["path", "onceKey"]),
// Workflow run records.
flueRuns: defineTable({
durationMs: v.optional(v.number()),
endedAt: v.optional(v.string()),
errorJson: v.optional(v.string()),
inputJson: v.optional(v.string()),
isError: v.optional(v.boolean()),
resultJson: v.optional(v.string()),
runId: v.string(),
workflowName: v.string(),
startedAt: v.string(),
status: v.union(
v.literal("active"),
v.literal("completed"),
v.literal("errored")
),
startedAt: v.string(),
inputJson: v.optional(v.string()),
traceCarrierJson: v.optional(v.string()),
endedAt: v.optional(v.string()),
isError: v.optional(v.boolean()),
durationMs: v.optional(v.number()),
resultJson: v.optional(v.string()),
errorJson: v.optional(v.string()),
workflowName: v.string(),
})
.index("by_runId", ["runId"])
.index("by_startedAt_and_runId", ["startedAt", "runId"])
@@ -479,15 +623,15 @@ export default defineSchema({
// Immutable attachment bytes. Identity is (streamPath, attachmentId); reads
// are additionally scoped by conversationId.
flueAttachments: defineTable({
streamPath: v.string(),
conversationId: v.string(),
attachmentId: v.string(),
mimeType: v.string(),
size: v.number(),
bytes: v.bytes(),
conversationId: v.string(),
createdAt: v.number(),
digest: v.string(),
filename: v.optional(v.string()),
bytes: v.bytes(),
createdAt: v.number(),
mimeType: v.string(),
size: v.number(),
streamPath: v.string(),
})
.index("by_streamPath", ["streamPath"])
.index("by_streamPath_and_attachmentId", ["streamPath", "attachmentId"])

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

@@ -82,16 +82,16 @@ export const listEvidence = query({
export const createSignal = mutation({
args: {
organizationId: v.id("organizations"),
projectId: v.id("projects"),
messageIds: v.array(v.string()),
organizationId: v.id("organizations"),
problemStatement: v.object({
title: v.string(),
summary: v.string(),
desiredOutcome: v.string(),
constraints: v.array(v.string()),
desiredOutcome: v.string(),
summary: v.string(),
title: v.string(),
}),
processedByAgentInstanceId: v.string(),
projectId: v.id("projects"),
token: v.string(),
},
handler: async (ctx, args) => {

108
packages/backend/convex/workArtifacts.test.ts Normal file
View File

@@ -0,0 +1,108 @@
import { env } from "@code/env/convex";
import { convexTest } from "convex-test";
import { anyApi } from "convex/server";
import { describe, expect, test } from "vitest";
import schema from "./schema";
declare global {
interface ImportMeta {
readonly glob: (pattern: string) => Record<string, () => Promise<unknown>>;
}
}
const modules = import.meta.glob("./**/*.ts");
const api = anyApi;
describe("Work artifacts and delivery", () => {
test("records exact idempotent artifact and delivery metadata", async () => {
const t = convexTest({ modules, schema });
const workId = await t.run(async (ctx) => {
const organizationId = await ctx.db.insert("organizations", {
createdAt: 1,
createdBy: "user",
kind: "personal",
name: "Personal",
});
const projectId = await ctx.db.insert("projects", {
createdAt: 1,
name: "Zopu",
normalizedSourceUrl: "https://example.com/zopu",
organizationId,
repositoryPath: "puter/zopu",
sourceHost: "example.com",
sourceUrl: "https://example.com/zopu",
updatedAt: 1,
});
return await ctx.db.insert("works", {
createdAt: 1,
objective: "Persist evidence",
organizationId,
projectId,
status: "executing",
title: "Artifact proof",
updatedAt: 1,
});
});
const draft = {
idempotencyKey: "attempt-1:test-report",
kind: "test-report" as const,
metadataJson: "{}",
producer: "fake-harness",
provenanceJson: "{}",
sourceRevision: "abc123",
title: "Focused tests",
verificationStatus: "verified" as const,
};
const first = await t.mutation(api.workArtifacts.recordArtifact, {
draft,
token: env.FLUE_DB_TOKEN,
workId,
});
const replay = await t.mutation(api.workArtifacts.recordArtifact, {
draft,
token: env.FLUE_DB_TOKEN,
workId,
});
expect(first.created).toBe(true);
expect(replay).toEqual({ artifactId: first.artifactId, created: false });
const delivery = await t.mutation(api.workArtifacts.recordDelivery, {
artifactId: first.artifactId,
draft: {
externalId: "42",
idempotencyKey: "pr:42",
kind: "pull-request",
metadataJson: "{}",
provider: "gitea",
sourceRevision: "abc123",
status: "ready",
target: "main",
url: "https://git.example/pulls/42",
},
token: env.FLUE_DB_TOKEN,
workId,
});
expect(delivery.created).toBe(true);
const updated = await t.mutation(api.workArtifacts.updateDeliveryStatus, {
deliveryId: delivery.deliveryId,
metadataJson: '{"approvedBy":"user"}',
status: "approved",
token: env.FLUE_DB_TOKEN,
});
expect(updated).toEqual({ changed: true, status: "approved" });
const stored = await t.run(async (ctx) => ({
artifacts: await ctx.db.query("workArtifacts").collect(),
deliveries: await ctx.db.query("workDeliveries").collect(),
events: await ctx.db.query("workEvents").collect(),
}));
expect(stored.artifacts).toHaveLength(1);
expect(stored.deliveries).toHaveLength(1);
expect(stored.events.map((event) => event.kind)).toEqual([
"artifact.recorded",
"delivery.recorded",
"delivery.updated",
]);
});
});

370
packages/backend/convex/workArtifacts.ts Normal file
View File

@@ -0,0 +1,370 @@
import { env } from "@code/env/convex";
import {
canTransitionWorkDelivery,
decodeWorkArtifactDraft,
decodeWorkDeliveryDraft,
} from "@code/primitives/work-artifact";
import { ConvexError, v } from "convex/values";
import { Effect } from "effect";
import type { Id } from "./_generated/dataModel";
import { mutation, query } from "./_generated/server";
import type { MutationCtx } from "./_generated/server";
import { requireProjectMember } from "./authz";
const artifactKind = v.union(
v.literal("definition"),
v.literal("design"),
v.literal("diff"),
v.literal("test-report"),
v.literal("verification-report"),
v.literal("runtime-log"),
v.literal("screenshot"),
v.literal("video"),
v.literal("commit"),
v.literal("branch"),
v.literal("pull-request"),
v.literal("preview"),
v.literal("deployment"),
v.literal("other")
);
const verificationStatus = v.union(
v.literal("unverified"),
v.literal("verified"),
v.literal("rejected")
);
const deliveryKind = v.union(
v.literal("branch"),
v.literal("commit"),
v.literal("pull-request"),
v.literal("preview"),
v.literal("deployment")
);
const deliveryStatus = v.union(
v.literal("recorded"),
v.literal("ready"),
v.literal("approved"),
v.literal("delivered"),
v.literal("failed"),
v.literal("cancelled")
);
const requireAgent = (token: string): void => {
if (token !== env.FLUE_DB_TOKEN) {
throw new ConvexError("Invalid agent control token");
}
};
const appendEvent = async (
ctx: MutationCtx,
workId: Id<"works">,
kind: "artifact.recorded" | "delivery.recorded" | "delivery.updated",
idempotencyKey: string,
referenceId: string,
payloadJson: string
): Promise<void> => {
const existing = await ctx.db
.query("workEvents")
.withIndex("by_work_and_idempotencyKey", (q) =>
q.eq("workId", workId).eq("idempotencyKey", idempotencyKey)
)
.unique();
if (existing) {
return;
}
await ctx.db.insert("workEvents", {
createdAt: Date.now(),
idempotencyKey,
kind,
payloadJson,
referenceId,
workId,
});
};
export const recordArtifact = mutation({
args: {
attemptId: v.optional(v.id("workAttempts")),
designVersion: v.optional(v.number()),
draft: v.object({
contentHash: v.optional(v.string()),
environmentId: v.optional(v.string()),
idempotencyKey: v.string(),
kind: artifactKind,
metadataJson: v.string(),
producer: v.string(),
provenanceJson: v.string(),
sourceRevision: v.optional(v.string()),
title: v.string(),
uri: v.optional(v.string()),
verificationStatus,
}),
runId: v.optional(v.id("workRuns")),
sliceId: v.optional(v.string()),
token: v.string(),
workId: v.id("works"),
},
handler: async (ctx, args) => {
requireAgent(args.token);
const work = await ctx.db.get(args.workId);
if (!work) {
throw new ConvexError("Work not found");
}
const draft = await Effect.runPromise(
decodeWorkArtifactDraft(args.draft)
).catch((error: unknown) => {
throw new ConvexError(
error instanceof Error ? error.message : "Invalid Work artifact"
);
});
const attempt = args.attemptId ? await ctx.db.get(args.attemptId) : null;
if (args.attemptId && !attempt) {
throw new ConvexError("Attempt not found");
}
const referencedRunId = args.runId ?? attempt?.runId;
const run = referencedRunId ? await ctx.db.get(referencedRunId) : null;
if (referencedRunId && !run) {
throw new ConvexError("Run not found");
}
if (run && run.workId !== work._id) {
throw new ConvexError("Run does not belong to Work");
}
if (
attempt &&
(attempt.workId !== work._id ||
(run !== null && attempt.runId !== run._id))
) {
throw new ConvexError("Attempt does not belong to Work Run");
}
if (
run &&
((args.designVersion !== undefined &&
args.designVersion !== run.designVersion) ||
(args.sliceId !== undefined && args.sliceId !== run.sliceId))
) {
throw new ConvexError("Artifact provenance conflicts with Work Run");
}
const designVersion = args.designVersion ?? run?.designVersion;
const sliceId = args.sliceId ?? run?.sliceId;
if (designVersion !== undefined && sliceId !== undefined) {
const slice = await ctx.db
.query("workSlices")
.withIndex("by_workId_and_designVersion_and_sliceId", (q) =>
q
.eq("workId", work._id)
.eq("designVersion", designVersion)
.eq("sliceId", sliceId)
)
.unique();
if (!slice) {
throw new ConvexError("Artifact slice not found");
}
} else if (designVersion !== undefined) {
const design = await ctx.db
.query("designPackets")
.withIndex("by_work_and_version", (q) =>
q.eq("workId", work._id).eq("version", designVersion)
)
.unique();
if (!design) {
throw new ConvexError("Artifact Design not found");
}
}
const existing = await ctx.db
.query("workArtifacts")
.withIndex("by_workId_and_idempotencyKey", (q) =>
q.eq("workId", work._id).eq("idempotencyKey", draft.idempotencyKey)
)
.unique();
if (existing) {
const exactReplay =
existing.kind === draft.kind &&
existing.title === draft.title &&
existing.uri === draft.uri &&
existing.contentHash === draft.contentHash &&
existing.sourceRevision === draft.sourceRevision &&
existing.environmentId === draft.environmentId &&
existing.producer === draft.producer &&
existing.provenanceJson === draft.provenanceJson &&
existing.metadataJson === draft.metadataJson &&
existing.verificationStatus === draft.verificationStatus &&
existing.designVersion === designVersion &&
existing.sliceId === sliceId &&
existing.runId === run?._id &&
existing.attemptId === attempt?._id;
if (!exactReplay) {
throw new ConvexError("Artifact idempotency key has conflicting data");
}
return { artifactId: existing._id, created: false };
}
const createdAt = Date.now();
const artifactId = await ctx.db.insert("workArtifacts", {
...draft,
attemptId: attempt?._id,
createdAt,
designVersion,
organizationId: work.organizationId,
projectId: work.projectId,
runId: run?._id,
sliceId,
workId: work._id,
});
await appendEvent(
ctx,
work._id,
"artifact.recorded",
`artifact:${draft.idempotencyKey}`,
String(artifactId),
JSON.stringify({ kind: draft.kind, title: draft.title })
);
return { artifactId, created: true };
},
});
export const recordDelivery = mutation({
args: {
artifactId: v.optional(v.id("workArtifacts")),
draft: v.object({
externalId: v.string(),
idempotencyKey: v.string(),
kind: deliveryKind,
metadataJson: v.string(),
provider: v.string(),
sourceRevision: v.string(),
status: deliveryStatus,
target: v.string(),
url: v.optional(v.string()),
}),
token: v.string(),
workId: v.id("works"),
},
handler: async (ctx, args) => {
requireAgent(args.token);
const work = await ctx.db.get(args.workId);
if (!work) {
throw new ConvexError("Work not found");
}
const draft = await Effect.runPromise(
decodeWorkDeliveryDraft(args.draft)
).catch((error: unknown) => {
throw new ConvexError(
error instanceof Error ? error.message : "Invalid Work delivery"
);
});
const artifact = args.artifactId ? await ctx.db.get(args.artifactId) : null;
if (args.artifactId && !artifact) {
throw new ConvexError("Artifact not found");
}
if (artifact && artifact.workId !== work._id) {
throw new ConvexError("Artifact does not belong to Work");
}
const existing = await ctx.db
.query("workDeliveries")
.withIndex("by_workId_and_idempotencyKey", (q) =>
q.eq("workId", work._id).eq("idempotencyKey", draft.idempotencyKey)
)
.unique();
if (existing) {
const exactReplay =
existing.kind === draft.kind &&
existing.provider === draft.provider &&
existing.externalId === draft.externalId &&
existing.url === draft.url &&
existing.sourceRevision === draft.sourceRevision &&
existing.target === draft.target &&
existing.status === draft.status &&
existing.metadataJson === draft.metadataJson &&
existing.artifactId === artifact?._id;
if (!exactReplay) {
throw new ConvexError("Delivery idempotency key has conflicting data");
}
return { created: false, deliveryId: existing._id };
}
const timestamp = Date.now();
const deliveryId = await ctx.db.insert("workDeliveries", {
...draft,
artifactId: artifact?._id,
createdAt: timestamp,
organizationId: work.organizationId,
projectId: work.projectId,
updatedAt: timestamp,
workId: work._id,
});
await appendEvent(
ctx,
work._id,
"delivery.recorded",
`delivery:${draft.idempotencyKey}`,
String(deliveryId),
JSON.stringify({ kind: draft.kind, sourceRevision: draft.sourceRevision })
);
return { created: true, deliveryId };
},
});
export const updateDeliveryStatus = mutation({
args: {
deliveryId: v.id("workDeliveries"),
metadataJson: v.string(),
status: deliveryStatus,
token: v.string(),
},
handler: async (ctx, args) => {
requireAgent(args.token);
// Validate JSON before persisting, matching recordDelivery's decode path.
const metadata = JSON.parse(args.metadataJson) as unknown;
if (typeof metadata !== "object" || metadata === null) {
throw new ConvexError("Delivery metadata must be a JSON object");
}
const delivery = await ctx.db.get(args.deliveryId);
if (!delivery) {
throw new ConvexError("Delivery not found");
}
if (delivery.status === args.status) {
return { changed: false, status: delivery.status };
}
if (!canTransitionWorkDelivery(delivery.status, args.status)) {
throw new ConvexError(
`Invalid delivery transition: ${delivery.status} -> ${args.status}`
);
}
await ctx.db.patch(delivery._id, {
metadataJson: args.metadataJson,
status: args.status,
updatedAt: Date.now(),
});
await appendEvent(
ctx,
delivery.workId,
"delivery.updated",
`delivery-updated:${delivery._id}:${args.status}`,
String(delivery._id),
JSON.stringify({ from: delivery.status, to: args.status })
);
return { changed: true, status: args.status };
},
});
export const listForWork = query({
args: { workId: v.id("works") },
handler: async (ctx, args) => {
const work = await ctx.db.get(args.workId);
if (!work) {
return null;
}
await requireProjectMember(ctx, work.projectId);
const [artifacts, deliveries] = await Promise.all([
ctx.db
.query("workArtifacts")
.withIndex("by_workId_and_createdAt", (q) => q.eq("workId", work._id))
.order("desc")
.take(100),
ctx.db
.query("workDeliveries")
.withIndex("by_workId_and_createdAt", (q) => q.eq("workId", work._id))
.order("desc")
.take(50),
]);
return { artifacts, deliveries };
},
});

111
packages/backend/convex/workExecution.test.ts
View File

@@ -1,5 +1,5 @@
import { convexTest } from "convex-test";
import { anyApi } from "convex/server";
import { anyApi, makeFunctionReference } from "convex/server";
import { describe, expect, test } from "vitest";
import type { Id } from "./_generated/dataModel";
@@ -15,6 +15,11 @@ const modules = {
};
const identity = { tokenIdentifier: "https://convex.test|exec-user" };
const api = anyApi;
const claimAttemptRef = makeFunctionReference<
"mutation",
{ attemptId: Id<"workAttempts">; leaseMs: number; owner: string },
{ number: number } | null
>("workExecution:claimAttempt");
const makeTest = () => convexTest({ modules, schema });
type TestT = ReturnType<typeof makeTest>;
@@ -26,7 +31,7 @@ type Scenario =
| "permanent-failure"
| "cancelled";
const seedReadyWork = async () => {
const seedReadyWork = async (options: { secondSlice?: boolean } = {}) => {
const t = convexTest({ modules, schema });
const { workId } = await t.withIdentity(identity).run(async (ctx) => {
const organizationId = await ctx.db.insert("organizations", {
@@ -74,6 +79,7 @@ const seedReadyWork = async () => {
workId: id,
});
await ctx.db.insert("workSlices", {
createdAt: 1,
designVersion: 1,
objective: "execute the slice",
observableBehavior: "terminal run",
@@ -84,6 +90,20 @@ const seedReadyWork = async () => {
title: "first",
workId: id,
});
if (options.secondSlice) {
await ctx.db.insert("workSlices", {
createdAt: 1,
designVersion: 1,
objective: "execute the second slice",
observableBehavior: "second terminal run",
ordinal: 1,
payloadJson: "{}",
sliceId: "slice-2",
status: "planned",
title: "second",
workId: id,
});
}
return { workId: id };
});
return { t, workId };
@@ -103,12 +123,14 @@ const attemptNumber = async (
const attempt = await t.run(async (ctx) =>
ctx.db
.query("workAttempts")
.withIndex("by_run_and_number", (q) =>
.withIndex("by_runId_and_number", (q) =>
q.eq("runId", runId).eq("number", number)
)
.unique()
);
if (!attempt) throw new Error(`attempt ${number} not found`);
if (!attempt) {
throw new Error(`attempt ${number} not found`);
}
return attempt._id;
};
@@ -116,15 +138,21 @@ const snapshot = async (t: TestT, workId: Id<"works">, runId: Id<"workRuns">) =>
t.run(async (ctx) => ({
attempts: await ctx.db
.query("workAttempts")
.withIndex("by_run_and_number", (q) => q.eq("runId", runId))
.withIndex("by_runId_and_number", (q) => q.eq("runId", runId))
.collect(),
run: await ctx.db.get(runId),
slice: await ctx.db
.query("workSlices")
.withIndex("by_work_and_designVersion", (q) =>
.withIndex("by_workId_and_designVersion", (q) =>
q.eq("workId", workId).eq("designVersion", 1)
)
.first(),
slices: await ctx.db
.query("workSlices")
.withIndex("by_workId_and_designVersion", (q) =>
q.eq("workId", workId).eq("designVersion", 1)
)
.collect(),
work: await ctx.db.get(workId),
}));
@@ -144,6 +172,77 @@ describe("simulated execution resolver", () => {
expect(state.run?.terminalClassification).toBe("Succeeded");
expect(state.attempts).toHaveLength(1);
expect(state.slice?.status).toBe("completed");
const decisions = await t.run(async (ctx) =>
ctx.db.query("resolverDecisions").collect()
);
expect(decisions).toHaveLength(1);
expect(decisions[0]).toMatchObject({
classification: "Succeeded",
decision: "terminal",
resultingWorkStatus: "completed",
});
});
test("completes Work only after every approved slice succeeds", async () => {
const { t, workId } = await seedReadyWork({ secondSlice: true });
const first = await t
.withIdentity(identity)
.mutation(api.workExecution.startSimulatedExecution, {
scenario: "success",
workId,
});
await runAttempt(t, first.attemptId, "success");
const afterFirst = await snapshot(t, workId, first.runId);
expect(afterFirst.work?.status).toBe("ready");
expect(
afterFirst.slices.find((slice) => slice.sliceId === "slice-1")?.status
).toBe("completed");
expect(
afterFirst.slices.find((slice) => slice.sliceId === "slice-2")?.status
).toBe("ready");
await expect(
t
.withIdentity(identity)
.mutation(api.workExecution.retrySimulatedExecution, {
runId: first.runId,
})
).rejects.toThrow(/retryable state/u);
const second = await t
.withIdentity(identity)
.mutation(api.workExecution.startSimulatedExecution, {
scenario: "success",
sliceId: "slice-2",
workId,
});
await runAttempt(t, second.attemptId, "success");
const completed = await snapshot(t, workId, second.runId);
expect(completed.work?.status).toBe("completed");
expect(
completed.slices.every((slice) => slice.status === "completed")
).toBe(true);
});
test("allows only one owner to claim a queued attempt", async () => {
const { t, workId } = await seedReadyWork();
const started = await t
.withIdentity(identity)
.mutation(api.workExecution.startSimulatedExecution, {
scenario: "success",
workId,
});
const first = await t.mutation(claimAttemptRef, {
attemptId: started.attemptId,
leaseMs: 60_000,
owner: "worker-a",
});
const second = await t.mutation(claimAttemptRef, {
attemptId: started.attemptId,
leaseMs: 60_000,
owner: "worker-b",
});
expect(first?.number).toBe(1);
expect(second).toBeNull();
});
test("transient failure retries within the kit budget and then succeeds", async () => {

455
packages/backend/convex/workExecution.ts
View File

@@ -1,12 +1,7 @@
import {
FakeHarnessLive,
type FakeScenario,
} from "@code/primitives/harness-runtime";
import {
type AttemptClassification,
defaultCodingKitV0,
resolveOutcome,
} from "@code/primitives/resolver";
import { FakeHarnessLive } from "@code/primitives/harness-runtime";
import type { FakeScenario } from "@code/primitives/harness-runtime";
import { defaultCodingKitV0, resolveOutcome } from "@code/primitives/resolver";
import type { WorkEventKind } from "@code/primitives/work";
import { makeFunctionReference } from "convex/server";
import { ConvexError, v } from "convex/values";
import { Effect } from "effect";
@@ -18,6 +13,7 @@ import {
mutation,
query,
} from "./_generated/server";
import type { MutationCtx } from "./_generated/server";
import { requireProjectMember } from "./authz";
const executeRef = makeFunctionReference<
@@ -53,34 +49,51 @@ const finishRef = makeFunctionReference<
unknown
>("workExecution:finishAttempt");
const now = () => Date.now();
const attemptClassification = v.union(
v.literal("Succeeded"),
v.literal("RetryableFailure"),
v.literal("NeedsInput"),
v.literal("Blocked"),
v.literal("VerificationFailed"),
v.literal("BudgetExhausted"),
v.literal("Cancelled"),
v.literal("PermanentFailure")
);
const requireWorkForMember = async (ctx: any, workId: Id<"works">) => {
const requireWorkForMember = async (
ctx: MutationCtx,
workId: Id<"works">
): Promise<Doc<"works">> => {
const work = await ctx.db.get(workId);
if (!work) throw new ConvexError("Work not found");
if (!work) {
throw new ConvexError("Work not found");
}
await requireProjectMember(ctx, work.projectId);
return work;
};
const appendWorkEvent = async (
ctx: any,
ctx: MutationCtx,
workId: Id<"works">,
kind: any,
kind: WorkEventKind,
idempotencyKey: string,
referenceId?: string,
payloadJson?: string
) => {
const existing = await ctx.db
.query("workEvents")
.withIndex("by_work_and_idempotencyKey", (q: any) =>
.withIndex("by_work_and_idempotencyKey", (q) =>
q.eq("workId", workId).eq("idempotencyKey", idempotencyKey)
)
.unique();
if (existing) return;
if (existing) {
return;
}
await ctx.db.insert("workEvents", {
workId,
kind,
idempotencyKey,
createdAt: now(),
idempotencyKey,
kind,
workId,
...(referenceId ? { referenceId } : {}),
...(payloadJson ? { payloadJson } : {}),
});
@@ -89,54 +102,64 @@ const appendWorkEvent = async (
// Slice the Run targets for the currently approved Design. `sliceId` is
// optional only to let the resolver default to the first ready slice.
const resolveRunSlice = async (
ctx: any,
ctx: MutationCtx,
work: Doc<"works">,
sliceId?: string
) => {
): Promise<Doc<"workSlices">> => {
const currentDesignVersion = work.designVersion;
if (currentDesignVersion === undefined)
if (currentDesignVersion === undefined) {
throw new ConvexError("Work has no approved Design to execute");
}
const slices = await ctx.db
.query("workSlices")
.withIndex("by_work_and_designVersion", (q: any) =>
.withIndex("by_workId_and_designVersion", (q) =>
q.eq("workId", work._id).eq("designVersion", currentDesignVersion)
)
.collect();
if (slices.length === 0)
if (slices.length === 0) {
throw new ConvexError("No slices found for the current approved Design");
}
const slice = sliceId
? slices.find((row: any) => row.sliceId === sliceId)
: slices.sort((a: any, b: any) => a.ordinal - b.ordinal)[0];
if (!slice)
? slices.find((row) => row.sliceId === sliceId)
: slices
.sort((a, b) => a.ordinal - b.ordinal)
.find((row) => row.status === "ready");
if (!slice) {
throw new ConvexError(
"Requested slice does not belong to the current approved Design"
);
return slice as Doc<"workSlices">;
}
if (slice.status !== "ready") {
throw new ConvexError("Only the next ready slice can be executed");
}
return slice;
};
const setSliceStatus = async (
ctx: any,
workId: Id<"works">,
sliceId: string | undefined,
status: any
) => {
if (!sliceId) return;
const slice = (
await ctx.db
.query("workSlices")
.withIndex("by_work_and_designVersion", (q: any) =>
q.eq("workId", workId)
)
.collect()
).find((row: any) => row.sliceId === sliceId);
if (slice) await ctx.db.patch(slice._id, { status });
const getRunSlice = async (
ctx: MutationCtx,
run: Doc<"workRuns">
): Promise<Doc<"workSlices"> | null> => {
if (run.sliceRowId) {
return await ctx.db.get(run.sliceRowId);
}
if (run.designVersion === undefined || run.sliceId === undefined) {
return null;
}
return await ctx.db
.query("workSlices")
.withIndex("by_workId_and_designVersion_and_sliceId", (q) =>
q
.eq("workId", run.workId)
.eq("designVersion", run.designVersion!)
.eq("sliceId", run.sliceId!)
)
.unique();
};
const RETRYABLE_WORK_STATUSES = ["failed", "needs-input", "blocked", "ready"];
const RETRYABLE_WORK_STATUSES = new Set(["failed", "needs-input", "blocked"]);
export const startSimulatedExecution = mutation({
args: {
workId: v.id("works"),
scenario: v.union(
v.literal("success"),
v.literal("transient-failure-then-success"),
@@ -145,41 +168,46 @@ export const startSimulatedExecution = mutation({
v.literal("cancelled")
),
sliceId: v.optional(v.string()),
workId: v.id("works"),
},
handler: async (ctx, args) => {
const work = await requireWorkForMember(ctx, args.workId);
if (work.status !== "ready")
if (work.status !== "ready") {
throw new ConvexError("Work must be Ready before simulated execution");
}
if (
work.definitionApprovalVersion !== work.definitionVersion ||
work.designApprovalVersion !== work.designVersion
)
) {
throw new ConvexError(
"Execution requires exact approved Definition and Design versions"
);
}
const slice = await resolveRunSlice(ctx, work, args.sliceId);
const createdAt = now();
const runId = await ctx.db.insert("workRuns", {
workId: work._id,
sliceId: slice.sliceId,
status: "ready",
scenario: args.scenario,
createdAt,
designVersion: slice.designVersion,
kitId: defaultCodingKitV0.id,
kitVersion: defaultCodingKitV0.version,
createdAt,
scenario: args.scenario,
sliceId: slice.sliceId,
sliceRowId: slice._id,
status: "ready",
workId: work._id,
});
const attemptId = await ctx.db.insert("workAttempts", {
runId,
workId: work._id,
number: 1,
runId,
status: "queued",
workId: work._id,
});
await ctx.db.patch(work._id, {
status: "executing",
updatedAt: createdAt,
});
await ctx.db.patch(slice._id, { status: "running" });
await ctx.db.patch(runId, { status: "running", startedAt: createdAt });
await ctx.db.patch(runId, { startedAt: createdAt, status: "running" });
await appendWorkEvent(
ctx,
work._id,
@@ -187,11 +215,18 @@ export const startSimulatedExecution = mutation({
`run-started:${runId}`,
String(runId)
);
await appendWorkEvent(
ctx,
work._id,
"slice.started",
`slice-started:${runId}:${slice.sliceId}`,
String(slice._id)
);
await ctx.scheduler.runAfter(0, executeRef, {
attemptId,
scenario: args.scenario,
});
return { runId, attemptId };
return { attemptId, runId };
},
});
@@ -199,29 +234,56 @@ export const cancelSimulatedExecution = mutation({
args: { runId: v.id("workRuns") },
handler: async (ctx, args) => {
const run = await ctx.db.get(args.runId);
if (!run) throw new ConvexError("Run not found");
if (!run) {
throw new ConvexError("Run not found");
}
const work = await requireWorkForMember(ctx, run.workId);
if (run.status === "terminal" || run.status === "cancelled")
if (run.status === "terminal" || run.status === "cancelled") {
return { cancelled: false };
}
await ctx.db.patch(run._id, {
status: "cancelled",
endedAt: now(),
status: "cancelled",
terminalClassification: "Cancelled",
terminalSummary: "Simulation cancelled",
});
const attempts = await ctx.db
.query("workAttempts")
.withIndex("by_run_and_number", (q: any) => q.eq("runId", run._id))
.withIndex("by_runId_and_number", (q) => q.eq("runId", run._id))
.collect();
for (const attempt of attempts)
if (attempt.status !== "terminal")
for (const attempt of attempts) {
if (attempt.status !== "terminal") {
await ctx.db.patch(attempt._id, {
status: "terminal",
endedAt: now(),
classification: "Cancelled",
endedAt: now(),
status: "terminal",
summary: "Simulation cancelled",
});
await setSliceStatus(ctx, work._id, run.sliceId, "ready");
await ctx.db.insert("resolverDecisions", {
attemptId: attempt._id,
attemptNumber: attempt.number,
classification: "Cancelled",
createdAt: now(),
decision: "terminal",
resultingWorkStatus: "ready",
runId: run._id,
summary: "Simulation cancelled",
workId: work._id,
});
await appendWorkEvent(
ctx,
work._id,
"resolver.decided",
`resolver-decided:${attempt._id}`,
String(attempt._id),
JSON.stringify({ classification: "Cancelled", decision: "terminal" })
);
}
}
const slice = await getRunSlice(ctx, run);
if (slice) {
await ctx.db.patch(slice._id, { status: "ready" });
}
await ctx.db.patch(work._id, { status: "ready", updatedAt: now() });
await appendWorkEvent(
ctx,
@@ -240,32 +302,40 @@ export const retrySimulatedExecution = mutation({
args: { runId: v.id("workRuns") },
handler: async (ctx, args) => {
const run = await ctx.db.get(args.runId);
if (!run) throw new ConvexError("Run not found");
if (!run) {
throw new ConvexError("Run not found");
}
const work = await requireWorkForMember(ctx, run.workId);
if (run.status !== "terminal")
if (run.status !== "terminal") {
throw new ConvexError("Only terminal Runs can be retried");
if (!RETRYABLE_WORK_STATUSES.includes(work.status))
}
if (!RETRYABLE_WORK_STATUSES.has(work.status)) {
throw new ConvexError("Work is not in a retryable state");
}
const attempts = await ctx.db
.query("workAttempts")
.withIndex("by_run_and_number", (q: any) => q.eq("runId", run._id))
.withIndex("by_runId_and_number", (q) => q.eq("runId", run._id))
.collect();
const number = attempts.length + 1;
const attemptId = await ctx.db.insert("workAttempts", {
runId: run._id,
workId: work._id,
number,
runId: run._id,
status: "queued",
workId: work._id,
});
await ctx.db.patch(run._id, {
status: "running",
startedAt: now(),
endedAt: undefined,
startedAt: now(),
status: "running",
terminalClassification: undefined,
terminalSummary: undefined,
});
await ctx.db.patch(work._id, { status: "executing", updatedAt: now() });
await setSliceStatus(ctx, work._id, run.sliceId, "running");
const slice = await getRunSlice(ctx, run);
if (!slice) {
throw new ConvexError("Run slice not found");
}
await ctx.db.patch(slice._id, { status: "running" });
await ctx.scheduler.runAfter(0, executeRef, {
attemptId,
scenario: run.scenario,
@@ -277,62 +347,71 @@ export const retrySimulatedExecution = mutation({
export const claimAttempt = internalMutation({
args: {
attemptId: v.id("workAttempts"),
owner: v.string(),
leaseMs: v.number(),
owner: v.string(),
},
handler: async (ctx, args) => {
const attempt = await ctx.db.get(args.attemptId);
if (!attempt || attempt.status === "terminal") return null;
if (!attempt || attempt.status !== "queued") {
return null;
}
const run = await ctx.db.get(attempt.runId);
if (!run || run.status !== "running") {
return null;
}
const claimedAt = now();
await ctx.db.patch(attempt._id, {
status: "claimed",
leaseOwner: args.owner,
leaseExpiresAt: claimedAt + args.leaseMs,
leaseOwner: args.owner,
startedAt: attempt.startedAt ?? claimedAt,
status: "claimed",
});
await ctx.db.patch(attempt.runId, {
status: "running",
startedAt: claimedAt,
status: "running",
});
return { ...attempt, status: "claimed" as const, leaseOwner: args.owner };
return { ...attempt, leaseOwner: args.owner, status: "claimed" as const };
},
});
export const checkpointAttempt = internalMutation({
args: {
attemptId: v.id("workAttempts"),
owner: v.string(),
sequence: v.number(),
kind: v.string(),
message: v.string(),
metadataJson: v.string(),
owner: v.string(),
sequence: v.number(),
},
handler: async (ctx, args) => {
const attempt = await ctx.db.get(args.attemptId);
if (
!attempt ||
attempt.leaseOwner !== args.owner ||
attempt.status === "terminal"
)
attempt.status === "terminal" ||
(attempt.leaseExpiresAt !== undefined && attempt.leaseExpiresAt < now())
) {
return false;
}
const existing = await ctx.db
.query("workAttemptEvents")
.withIndex("by_attempt_and_sequence", (q: any) =>
q.eq("attemptId", attempt._id).eq("sequence", args.sequence)
)
.unique();
if (!existing)
if (!existing) {
await ctx.db.insert("workAttemptEvents", {
attemptId: attempt._id,
sequence: args.sequence,
kind: args.kind,
message: args.message,
metadataJson: args.metadataJson,
occurredAt: now(),
sequence: args.sequence,
});
}
await ctx.db.patch(attempt._id, {
status: "running",
leaseExpiresAt: now() + 60_000,
status: "running",
});
return true;
},
@@ -345,8 +424,8 @@ export const checkpointAttempt = internalMutation({
export const finishAttempt = internalMutation({
args: {
attemptId: v.id("workAttempts"),
classification: attemptClassification,
owner: v.string(),
classification: v.string(),
retryable: v.boolean(),
summary: v.string(),
},
@@ -355,19 +434,21 @@ export const finishAttempt = internalMutation({
if (
!attempt ||
attempt.leaseOwner !== args.owner ||
attempt.status === "terminal"
)
attempt.status === "terminal" ||
(attempt.leaseExpiresAt !== undefined && attempt.leaseExpiresAt < now())
) {
return null;
}
const endedAt = now();
await ctx.db.patch(attempt._id, {
status: "terminal",
endedAt,
classification: args.classification,
summary: args.summary,
endedAt,
leaseExpiresAt: undefined,
status: "terminal",
summary: args.summary,
});
const outcome = {
classification: args.classification as AttemptClassification,
classification: args.classification,
retryable: args.retryable,
summary: args.summary,
};
@@ -377,7 +458,9 @@ export const finishAttempt = internalMutation({
defaultCodingKitV0.retryPolicy
);
const run = await ctx.db.get(attempt.runId);
if (!run) return null;
if (!run) {
return null;
}
await appendWorkEvent(
ctx,
attempt.workId,
@@ -390,13 +473,34 @@ export const finishAttempt = internalMutation({
summary: args.summary,
})
);
await appendWorkEvent(
ctx,
attempt.workId,
"resolver.decided",
`resolver-decided:${attempt._id}`,
String(attempt._id),
JSON.stringify({
classification: args.classification,
decision: resolution.kind,
})
);
if (resolution.kind === "retry") {
await ctx.db.insert("resolverDecisions", {
attemptId: attempt._id,
attemptNumber: attempt.number,
classification: args.classification,
createdAt: endedAt,
decision: "retry",
runId: run._id,
summary: args.summary,
workId: attempt.workId,
});
// Keep Run running and Work executing; spin the next attempt.
const nextAttemptId = await ctx.db.insert("workAttempts", {
runId: run._id,
workId: attempt.workId,
number: attempt.number + 1,
runId: run._id,
status: "queued",
workId: attempt.workId,
});
await ctx.scheduler.runAfter(0, executeRef, {
attemptId: nextAttemptId,
@@ -405,22 +509,78 @@ export const finishAttempt = internalMutation({
return { nextAttemptId, retried: true };
}
await ctx.db.patch(run._id, {
status: "terminal",
endedAt,
status: "terminal",
terminalClassification: args.classification,
terminalSummary: args.summary,
});
const work = await ctx.db.get(run.workId);
if (work)
const slice = await getRunSlice(ctx, run);
let resultingWorkStatus = resolution.workStatus;
if (args.classification === "Succeeded" && slice) {
await ctx.db.patch(slice._id, { status: "completed" });
await appendWorkEvent(
ctx,
attempt.workId,
"slice.completed",
`slice-completed:${run._id}:${slice.sliceId}`,
String(slice._id)
);
const slices = await ctx.db
.query("workSlices")
.withIndex("by_workId_and_designVersion", (q) =>
q
.eq("workId", attempt.workId)
.eq("designVersion", slice.designVersion)
)
.collect();
const nextSlice = slices
.sort((left, right) => left.ordinal - right.ordinal)
.find((candidate) => candidate.status === "planned");
if (nextSlice) {
await ctx.db.patch(nextSlice._id, { status: "ready" });
resultingWorkStatus = "ready";
await appendWorkEvent(
ctx,
attempt.workId,
"slice.ready",
`slice-ready:${run._id}:${nextSlice.sliceId}`,
String(nextSlice._id)
);
}
} else if (slice) {
const blocked = ["NeedsInput", "Blocked", "VerificationFailed"].includes(
args.classification
);
await ctx.db.patch(slice._id, { status: blocked ? "blocked" : "ready" });
}
if (work) {
await ctx.db.patch(work._id, {
status: resolution.workStatus,
status: resultingWorkStatus,
updatedAt: endedAt,
});
await setSliceStatus(
}
await ctx.db.insert("resolverDecisions", {
attemptId: attempt._id,
attemptNumber: attempt.number,
classification: args.classification,
createdAt: endedAt,
decision: "terminal",
resultingWorkStatus,
runId: run._id,
summary: args.summary,
workId: attempt.workId,
});
await appendWorkEvent(
ctx,
attempt.workId,
run.sliceId,
args.classification === "Succeeded" ? "completed" : "ready"
"run.completed",
`run-completed:${run._id}`,
String(run._id),
JSON.stringify({
classification: args.classification,
workStatus: resultingWorkStatus,
})
);
return { retried: false };
},
@@ -441,17 +601,19 @@ export const executeFakeAttempt = internalAction({
const owner = `fake-worker:${args.attemptId}`;
const claimed = await ctx.runMutation(claimRef, {
attemptId: args.attemptId,
owner,
leaseMs: 60_000,
owner,
});
if (!claimed) return null;
if (!claimed) {
return null;
}
const [events, outcome] = await Effect.runPromise(
FakeHarnessLive().run({
attemptNumber: claimed.number,
scenario: args.scenario,
})
);
for (const item of events)
for (const item of events) {
await ctx.runMutation(checkpointRef, {
attemptId: args.attemptId,
kind: item.kind,
@@ -460,6 +622,7 @@ export const executeFakeAttempt = internalAction({
owner,
sequence: item.sequence,
});
}
await ctx.runMutation(finishRef, {
attemptId: args.attemptId,
classification: outcome.classification,
@@ -477,7 +640,22 @@ export const executeFakeAttempt = internalAction({
export const reconcileExpiredAttempts = internalMutation({
args: {},
handler: async (ctx) => {
const active = await ctx.db.query("workAttempts").collect();
const timestamp = now();
const [claimed, running] = await Promise.all([
ctx.db
.query("workAttempts")
.withIndex("by_status_and_leaseExpiresAt", (q) =>
q.eq("status", "claimed").lt("leaseExpiresAt", timestamp)
)
.collect(),
ctx.db
.query("workAttempts")
.withIndex("by_status_and_leaseExpiresAt", (q) =>
q.eq("status", "running").lt("leaseExpiresAt", timestamp)
)
.collect(),
]);
const active = [...claimed, ...running];
let reconciled = 0;
for (const attempt of active) {
if (
@@ -486,12 +664,12 @@ export const reconcileExpiredAttempts = internalMutation({
attempt.leaseExpiresAt < now()
) {
await ctx.db.patch(attempt._id, {
status: "terminal",
endedAt: now(),
classification: "Blocked",
summary: "Attempt lease expired and was reconciled",
leaseOwner: undefined,
endedAt: now(),
leaseExpiresAt: undefined,
leaseOwner: undefined,
status: "terminal",
summary: "Attempt lease expired and was reconciled",
});
const run = await ctx.db.get(attempt.runId);
await appendWorkEvent(
@@ -503,19 +681,44 @@ export const reconcileExpiredAttempts = internalMutation({
JSON.stringify({ attemptNumber: attempt.number })
);
reconciled += 1;
if (!run) continue;
if (!run) {
continue;
}
const attempts = await ctx.db
.query("workAttempts")
.withIndex("by_run_and_number", (q: any) => q.eq("runId", run._id))
.withIndex("by_runId_and_number", (q) => q.eq("runId", run._id))
.collect();
const withinBudget =
attempts.length < defaultCodingKitV0.retryPolicy.maxAttempts;
if (run.status === "running" && withinBudget) {
const retry = run.status === "running" && withinBudget;
await ctx.db.insert("resolverDecisions", {
attemptId: attempt._id,
attemptNumber: attempt.number,
classification: "Blocked",
createdAt: now(),
decision: retry ? "retry" : "terminal",
resultingWorkStatus: retry ? undefined : "blocked",
runId: run._id,
summary: "Attempt lease expired and was reconciled",
workId: attempt.workId,
});
await appendWorkEvent(
ctx,
attempt.workId,
"resolver.decided",
`resolver-decided:${attempt._id}`,
String(attempt._id),
JSON.stringify({
classification: "Blocked",
decision: retry ? "retry" : "terminal",
})
);
if (retry) {
const nextAttemptId = await ctx.db.insert("workAttempts", {
runId: run._id,
workId: attempt.workId,
number: attempts.length + 1,
runId: run._id,
status: "queued",
workId: attempt.workId,
});
await ctx.scheduler.runAfter(0, executeRef, {
attemptId: nextAttemptId,
@@ -523,18 +726,22 @@ export const reconcileExpiredAttempts = internalMutation({
});
} else {
await ctx.db.patch(run._id, {
status: "terminal",
endedAt: now(),
status: "terminal",
terminalClassification: "Blocked",
terminalSummary: "Run reconciled after expired lease",
});
const work = await ctx.db.get(run.workId);
if (work)
if (work) {
await ctx.db.patch(work._id, {
status: "blocked",
updatedAt: now(),
});
await setSliceStatus(ctx, attempt.workId, run.sliceId, "ready");
}
const slice = await getRunSlice(ctx, run);
if (slice) {
await ctx.db.patch(slice._id, { status: "blocked" });
}
}
}
}
@@ -546,13 +753,17 @@ export const listRunEvents = query({
args: { attemptId: v.id("workAttempts") },
handler: async (ctx, args) => {
const attempt = await ctx.db.get(args.attemptId);
if (!attempt) return [];
if (!attempt) {
return [];
}
const work = await ctx.db.get(attempt.workId);
if (!work) return [];
if (!work) {
return [];
}
await requireProjectMember(ctx, work.projectId);
return await ctx.db
.query("workAttemptEvents")
.withIndex("by_attempt_and_sequence", (q: any) =>
.withIndex("by_attempt_and_sequence", (q) =>
q.eq("attemptId", args.attemptId)
)
.order("asc")

203
packages/backend/convex/workPlanning.test.ts
View File

@@ -30,10 +30,10 @@ describe("Work planning and execution commands", () => {
name: "Personal",
});
await ctx.db.insert("organizationMembers", {
organizationId,
userId: identity.tokenIdentifier,
role: "owner",
createdAt: 1,
organizationId,
role: "owner",
userId: identity.tokenIdentifier,
});
const projectId = await ctx.db.insert("projects", {
createdAt: 1,
@@ -46,12 +46,12 @@ describe("Work planning and execution commands", () => {
updatedAt: 1,
});
const workId = await ctx.db.insert("works", {
createdAt: 1,
objective: "Prove the flow",
organizationId,
projectId,
title: "Executable work",
objective: "Prove the flow",
status: "proposed",
createdAt: 1,
title: "Executable work",
updatedAt: 1,
});
return { projectId, workId };
@@ -63,67 +63,67 @@ describe("Work planning and execution commands", () => {
const definition = await t
.withIdentity(identity)
.mutation(api.workPlanning.saveDefinitionProposal, {
workId: fixture.workId,
payloadJson: JSON.stringify({
problem: "Need a proof",
desiredOutcome: "A terminal simulation",
acceptanceCriteria: ["terminal"],
affectedUsers: ["user"],
assumptions: [],
constraints: [],
desiredOutcome: "A terminal simulation",
inScope: ["fake"],
outOfScope: [],
acceptanceCriteria: ["terminal"],
constraints: [],
assumptions: [],
problem: "Need a proof",
questions: [],
risk: "low",
requiredArtifacts: ["events"],
risk: "low",
}),
workId: fixture.workId,
});
await t
.withIdentity(identity)
.mutation(api.workPlanning.approveDefinition, {
workId: fixture.workId,
version: definition.version,
workId: fixture.workId,
});
const design = await t
.withIdentity(identity)
.mutation(api.workPlanning.saveDesignProposal, {
workId: fixture.workId,
payloadJson: JSON.stringify({
impactMap: { files: [], modules: [], risks: [], summary: "small" },
architectureSummary: "fake",
fileTreeDelta: [],
callFlowDelta: [],
keyInterfaces: [],
invariants: ["terminal"],
concerns: [],
evidenceRequirements: ["event"],
fileTreeDelta: [],
impactMap: { files: [], modules: [], risks: [], summary: "small" },
invariants: ["terminal"],
keyInterfaces: [],
slices: [
{
codeBoundaries: ["runtime"],
dependsOn: [],
evidenceRequirements: ["event"],
id: "slice-1",
title: "fake",
objective: "fake",
observableBehavior: "event",
codeBoundaries: ["runtime"],
evidenceRequirements: ["event"],
verification: ["assert"],
reviewRequired: false,
dependsOn: [],
title: "fake",
verification: ["assert"],
},
],
evidenceRequirements: ["event"],
tradeoffs: [],
}),
workId: fixture.workId,
});
await t.withIdentity(identity).mutation(api.workPlanning.approveDesign, {
workId: fixture.workId,
definitionVersion: definition.version,
designVersion: design.version,
workId: fixture.workId,
});
const started = await t
.withIdentity(identity)
.mutation(api.workExecution.startSimulatedExecution, {
workId: fixture.workId,
scenario: "success",
sliceId: "slice-1",
workId: fixture.workId,
});
expect(started.runId).toBeTruthy();
const state = await t.run(async (ctx) => ctx.db.get(fixture.workId));
@@ -137,7 +137,7 @@ describe("Work planning and execution commands", () => {
const completed = await t.run(async (ctx) => ({
attempt: await ctx.db
.query("workAttempts")
.withIndex("by_run_and_number", (q) => q.eq("runId", started.runId))
.withIndex("by_runId_and_number", (q) => q.eq("runId", started.runId))
.unique(),
events: await ctx.db.query("workAttemptEvents").collect(),
run: await ctx.db.get(started.runId as Id<"workRuns">),
@@ -164,7 +164,154 @@ const validDefinitionPayload = () => ({
risk: "low",
});
const validDesignPayload = (sliceId = "slice-1") => ({
architectureSummary: "fake",
callFlowDelta: [],
concerns: [],
evidenceRequirements: ["event"],
fileTreeDelta: [],
impactMap: { files: [], modules: [], risks: [], summary: "small" },
invariants: ["terminal"],
keyInterfaces: [],
slices: [
{
codeBoundaries: ["runtime"],
dependsOn: [],
evidenceRequirements: ["event"],
id: sliceId,
objective: "fake",
observableBehavior: "event",
reviewRequired: false,
title: "fake",
verification: ["assert"],
},
],
tradeoffs: [],
});
describe("Work planning guards", () => {
test("persists high-impact questions but blocks approval", async () => {
const t = convexTest({ modules, schema });
const { workId } = await t.withIdentity(identity).run(async (ctx) => {
const organizationId = await ctx.db.insert("organizations", {
createdAt: 1,
createdBy: identity.tokenIdentifier,
kind: "personal",
name: "Personal",
});
await ctx.db.insert("organizationMembers", {
createdAt: 1,
organizationId,
role: "owner",
userId: identity.tokenIdentifier,
});
const projectId = await ctx.db.insert("projects", {
createdAt: 1,
name: "Zopu",
normalizedSourceUrl: "https://example.com/zopu",
organizationId,
repositoryPath: "puter/zopu",
sourceHost: "example.com",
sourceUrl: "https://example.com/zopu",
updatedAt: 1,
});
const id = await ctx.db.insert("works", {
createdAt: 1,
objective: "Resolve a critical question",
organizationId,
projectId,
status: "defining",
title: "Question gate",
updatedAt: 1,
});
return { workId: id };
});
const saved = await t
.withIdentity(identity)
.mutation(api.workPlanning.saveDefinitionProposal, {
payloadJson: JSON.stringify({
...validDefinitionPayload(),
questions: [
{
alternatives: ["public", "private"],
id: "visibility",
impact: "high",
prompt: "Should the endpoint be public?",
status: "open",
},
],
}),
workId,
});
await expect(
t.withIdentity(identity).mutation(api.workPlanning.approveDefinition, {
version: saved.version,
workId,
})
).rejects.toThrow(/High-impact open questions/u);
const stored = await t.run(async (ctx) =>
ctx.db.query("workQuestions").collect()
);
expect(stored).toHaveLength(1);
expect(stored[0]?.status).toBe("open");
});
test("keeps prior Design slices as versioned history", async () => {
const t = convexTest({ modules, schema });
const { workId } = await t.withIdentity(identity).run(async (ctx) => {
const organizationId = await ctx.db.insert("organizations", {
createdAt: 1,
createdBy: identity.tokenIdentifier,
kind: "personal",
name: "Personal",
});
await ctx.db.insert("organizationMembers", {
createdAt: 1,
organizationId,
role: "owner",
userId: identity.tokenIdentifier,
});
const projectId = await ctx.db.insert("projects", {
createdAt: 1,
name: "Zopu",
normalizedSourceUrl: "https://example.com/zopu",
organizationId,
repositoryPath: "puter/zopu",
sourceHost: "example.com",
sourceUrl: "https://example.com/zopu",
updatedAt: 1,
});
const id = await ctx.db.insert("works", {
createdAt: 1,
definitionApprovalVersion: 1,
definitionVersion: 1,
objective: "Preserve Design history",
organizationId,
projectId,
status: "designing",
title: "Versioned Design",
updatedAt: 1,
});
return { workId: id };
});
await t
.withIdentity(identity)
.mutation(api.workPlanning.saveDesignProposal, {
payloadJson: JSON.stringify(validDesignPayload("slice-v1")),
workId,
});
await t
.withIdentity(identity)
.mutation(api.workPlanning.saveDesignProposal, {
payloadJson: JSON.stringify(validDesignPayload("slice-v2")),
workId,
});
const slices = await t.run(async (ctx) =>
ctx.db.query("workSlices").collect()
);
expect(slices.map((slice) => slice.designVersion).sort()).toEqual([1, 2]);
});
test("a Definition cannot be revised while a Run is executing", async () => {
const t = convexTest({ modules, schema });
const { workId } = await t.withIdentity(identity).run(async (ctx) => {

469
packages/backend/convex/workPlanning.ts
View File

@@ -1,8 +1,10 @@
import type { WorkEventKind } from "@code/primitives/work";
import {
type WorkDefinition,
decodeDefinition,
validateDefinition,
WorkQuestion,
} from "@code/primitives/work-definition";
import type { WorkDefinition } from "@code/primitives/work-definition";
import { validateDesignPacket } from "@code/primitives/work-design";
import { makeFunctionReference } from "convex/server";
import { ConvexError, v } from "convex/values";
@@ -12,10 +14,12 @@ import type { Doc, Id } from "./_generated/dataModel";
import {
env,
internalAction,
internalMutation,
internalQuery,
mutation,
query,
} from "./_generated/server";
import type { MutationCtx, QueryCtx } from "./_generated/server";
import { requireProjectMember } from "./authz";
const plannerRef = makeFunctionReference<
@@ -27,6 +31,11 @@ const plannerContextRef = makeFunctionReference<
{ workId: Id<"works"> },
{ objective: string; status: string; title: string } | null
>("workPlanning:getPlannerContext");
const plannerFailureRef = makeFunctionReference<
"mutation",
{ workId: Id<"works">; error: string },
null
>("workPlanning:recordPlannerFailure");
const parsePayload = (payloadJson: string): unknown => {
try {
@@ -38,67 +47,72 @@ const parsePayload = (payloadJson: string): unknown => {
const objectPayload = (payloadJson: string): Record<string, unknown> => {
const value = parsePayload(payloadJson);
if (typeof value !== "object" || value === null || Array.isArray(value))
if (typeof value !== "object" || value === null || Array.isArray(value)) {
throw new ConvexError("Proposal payload must be an object");
}
return value as Record<string, unknown>;
};
const requireWork = async (ctx: { db: any }, workId: Id<"works">) => {
const work = (await ctx.db.get(workId)) as Doc<"works"> | null;
if (!work) throw new ConvexError("Work not found");
const requireWork = async (
ctx: QueryCtx | MutationCtx,
workId: Id<"works">
): Promise<Doc<"works">> => {
const work = await ctx.db.get(workId);
if (!work) {
throw new ConvexError("Work not found");
}
return work;
};
const appendEvent = async (
ctx: { db: any },
ctx: MutationCtx,
workId: Id<"works">,
kind: any,
kind: WorkEventKind,
idempotencyKey: string,
referenceId?: string,
payloadJson?: string
) => {
const existing = await ctx.db
.query("workEvents")
.withIndex("by_work_and_idempotencyKey", (q: any) =>
.withIndex("by_work_and_idempotencyKey", (q) =>
q.eq("workId", workId).eq("idempotencyKey", idempotencyKey)
)
.unique();
if (existing) return existing._id;
if (existing) {
return existing._id;
}
return await ctx.db.insert("workEvents", {
workId,
kind,
idempotencyKey,
createdAt: Date.now(),
idempotencyKey,
kind,
workId,
...(referenceId ? { referenceId } : {}),
...(payloadJson ? { payloadJson } : {}),
});
};
const invalidateApprovalsAndDesign = async (
ctx: { db: any },
ctx: MutationCtx,
workId: Id<"works">
) => {
const approvals = await ctx.db
.query("workApprovals")
.withIndex("by_work_and_kind", (q: any) => q.eq("workId", workId))
.withIndex("by_workId_and_kind", (q) => q.eq("workId", workId))
.collect();
for (const approval of approvals) {
if (approval.status === "active")
if (approval.status === "active") {
await ctx.db.patch(approval._id, { status: "invalidated" });
}
}
const designs = await ctx.db
.query("designPackets")
.withIndex("by_work_and_version", (q: any) => q.eq("workId", workId))
.withIndex("by_work_and_version", (q) => q.eq("workId", workId))
.collect();
for (const design of designs) {
if (design.status === "current")
if (design.status === "current") {
await ctx.db.patch(design._id, { status: "superseded" });
}
}
const slices = await ctx.db
.query("workSlices")
.withIndex("by_work_and_designVersion", (q: any) => q.eq("workId", workId))
.collect();
for (const slice of slices) await ctx.db.delete(slice._id);
};
export const requestDefinition = mutation({
@@ -106,8 +120,16 @@ export const requestDefinition = mutation({
handler: async (ctx, args) => {
const work = await requireWork(ctx, args.workId);
await requireProjectMember(ctx, work.projectId);
if (work.status !== "proposed" && work.status !== "defining")
if (
work.status !== "proposed" &&
work.status !== "defining" &&
!(
work.status === "blocked" &&
work.definitionApprovalVersion !== work.definitionVersion
)
) {
throw new ConvexError("Work is not available for definition");
}
const now = Date.now();
await ctx.db.patch(work._id, { status: "defining", updatedAt: now });
await appendEvent(
@@ -125,15 +147,16 @@ export const requestDefinition = mutation({
});
const saveDefinition = async (
ctx: { db: any },
ctx: MutationCtx,
work: Doc<"works">,
payloadJson: string,
createdBy: string
) => {
if (work.status === "executing")
if (work.status === "executing") {
throw new ConvexError("Cannot revise Work while a Run is executing");
}
const decoded = await Effect.runPromise(
validateDefinition({
decodeDefinition({
...objectPayload(payloadJson),
version: (work.definitionVersion ?? 0) + 1,
})
@@ -142,52 +165,45 @@ const saveDefinition = async (
error instanceof Error ? error.message : "Invalid Definition"
);
});
const version = decoded.version;
const { version } = decoded;
await invalidateApprovalsAndDesign(ctx, work._id);
const previous = await ctx.db
.query("workDefinitions")
.withIndex("by_work_and_version", (q: any) => q.eq("workId", work._id))
.withIndex("by_work_and_version", (q) => q.eq("workId", work._id))
.collect();
for (const row of previous)
if (row.status === "current")
for (const row of previous) {
if (row.status === "current") {
await ctx.db.patch(row._id, { status: "superseded" });
}
}
const definitionId = await ctx.db.insert("workDefinitions", {
workId: work._id,
version,
createdAt: Date.now(),
createdBy,
payloadJson: JSON.stringify(decoded),
risk: decoded.risk,
status: "current",
createdBy,
createdAt: Date.now(),
version,
workId: work._id,
});
const questions = await ctx.db
.query("workQuestions")
.withIndex("by_work_and_definitionVersion", (q: any) =>
q.eq("workId", work._id)
)
.collect();
for (const question of questions)
if (question.definitionVersion !== version)
await ctx.db.delete(question._id);
for (const question of decoded.questions) {
await ctx.db.insert("workQuestions", {
workId: work._id,
definitionVersion: version,
questionId: question.id,
prompt: question.prompt,
impact: question.impact,
recommendation: question.recommendation,
alternativesJson: JSON.stringify(question.alternatives),
status: question.status,
answer: question.answer,
createdAt: Date.now(),
definitionVersion: version,
impact: question.impact,
prompt: question.prompt,
questionId: question.id,
recommendation: question.recommendation,
status: question.status,
workId: work._id,
});
}
await ctx.db.patch(work._id, {
definitionVersion: version,
definitionApprovalVersion: undefined,
designVersion: undefined,
definitionVersion: version,
designApprovalVersion: undefined,
designVersion: undefined,
status: "awaiting-definition-approval",
updatedAt: Date.now(),
});
@@ -203,7 +219,7 @@ const saveDefinition = async (
};
export const saveDefinitionProposal = mutation({
args: { workId: v.id("works"), payloadJson: v.string() },
args: { payloadJson: v.string(), workId: v.id("works") },
handler: async (ctx, args) => {
const work = await requireWork(ctx, args.workId);
await requireProjectMember(ctx, work.projectId);
@@ -214,22 +230,25 @@ export const saveDefinitionProposal = mutation({
export const reviseDefinition = saveDefinitionProposal;
export const approveDefinition = mutation({
args: { workId: v.id("works"), version: v.number() },
args: { version: v.number(), workId: v.id("works") },
handler: async (ctx, args) => {
const work = await requireWork(ctx, args.workId);
await requireProjectMember(ctx, work.projectId);
if (
work.definitionVersion !== args.version ||
work.status !== "awaiting-definition-approval"
)
) {
throw new ConvexError("Definition version is not current or approvable");
}
const row = await ctx.db
.query("workDefinitions")
.withIndex("by_work_and_version", (q: any) =>
.withIndex("by_work_and_version", (q) =>
q.eq("workId", work._id).eq("version", args.version)
)
.unique();
if (!row) throw new ConvexError("Definition not found");
if (!row) {
throw new ConvexError("Definition not found");
}
const definition = JSON.parse(row.payloadJson) as WorkDefinition;
const valid = await Effect.runPromise(validateDefinition(definition)).catch(
(error: unknown) => {
@@ -240,19 +259,36 @@ export const approveDefinition = mutation({
);
}
);
const questions = await ctx.db
.query("workQuestions")
.withIndex("by_workId_and_definitionVersion", (q) =>
q.eq("workId", work._id).eq("definitionVersion", args.version)
)
.collect();
if (
questions.some(
(question) => question.status === "open" && question.impact === "high"
)
) {
throw new ConvexError(
"High-impact open questions must be resolved before approval"
);
}
const identity = await ctx.auth.getUserIdentity();
if (!identity) throw new ConvexError("Authentication required");
if (!identity) {
throw new ConvexError("Authentication required");
}
await ctx.db.insert("workApprovals", {
workId: work._id,
kind: "definition",
definitionVersion: args.version,
approvedBy: identity.subject,
approvedAt: Date.now(),
approvedBy: identity.tokenIdentifier,
definitionVersion: args.version,
kind: "definition",
status: "active",
workId: work._id,
});
await ctx.db.patch(work._id, {
status: "designing",
definitionApprovalVersion: valid.version,
status: "designing",
updatedAt: Date.now(),
});
await appendEvent(
@@ -270,36 +306,57 @@ export const approveDefinition = mutation({
});
const reviseQuestion = async (
ctx: { db: any },
ctx: MutationCtx,
work: Doc<"works">,
questionId: string,
update: { status: "answered" | "withdrawn"; answer?: string }
) => {
const current = await ctx.db
.query("workDefinitions")
.withIndex("by_work_and_version", (q: any) =>
.withIndex("by_work_and_version", (q) =>
q.eq("workId", work._id).eq("version", work.definitionVersion ?? 0)
)
.unique();
if (!current) throw new ConvexError("Current Definition not found");
if (!current) {
throw new ConvexError("Current Definition not found");
}
const definition = JSON.parse(current.payloadJson) as WorkDefinition;
const persistedQuestions = await ctx.db
.query("workQuestions")
.withIndex("by_workId_and_definitionVersion", (q) =>
q.eq("workId", work._id).eq("definitionVersion", current.version)
)
.collect();
if (
!persistedQuestions.some((question) => question.questionId === questionId)
) {
throw new ConvexError("Question not found");
}
const next = {
...definition,
questions: definition.questions.map((question) =>
question.id === questionId ? { ...question, ...update } : question
),
questions: persistedQuestions.map((question) => ({
alternatives: JSON.parse(question.alternativesJson) as string[],
answer:
question.questionId === questionId ? update.answer : question.answer,
id: question.questionId,
impact: question.impact,
prompt: question.prompt,
recommendation: question.recommendation,
status:
question.questionId === questionId ? update.status : question.status,
})),
};
return await saveDefinition(ctx, work, JSON.stringify(next), "user");
};
export const answerQuestion = mutation({
args: { workId: v.id("works"), questionId: v.string(), answer: v.string() },
args: { answer: v.string(), questionId: v.string(), workId: v.id("works") },
handler: async (ctx, args) => {
const work = await requireWork(ctx, args.workId);
await requireProjectMember(ctx, work.projectId);
const result = await reviseQuestion(ctx, work, args.questionId, {
status: "answered",
answer: args.answer,
status: "answered",
});
await appendEvent(
ctx,
@@ -312,7 +369,7 @@ export const answerQuestion = mutation({
});
export const withdrawQuestion = mutation({
args: { workId: v.id("works"), questionId: v.string() },
args: { questionId: v.string(), workId: v.id("works") },
handler: async (ctx, args) => {
const work = await requireWork(ctx, args.workId);
await requireProjectMember(ctx, work.projectId);
@@ -335,35 +392,48 @@ export const requestDesign = mutation({
const work = await requireWork(ctx, args.workId);
await requireProjectMember(ctx, work.projectId);
if (
work.status !== "designing" ||
(work.status !== "designing" && work.status !== "blocked") ||
work.definitionApprovalVersion !== work.definitionVersion
)
) {
throw new ConvexError("Definition must be approved before Design");
}
if (work.status === "blocked") {
await ctx.db.patch(work._id, {
status: "designing",
updatedAt: Date.now(),
});
}
await appendEvent(
ctx,
work._id,
"design.requested",
`design-requested:${work.definitionVersion}`
);
return { status: work.status };
await ctx.scheduler.runAfter(0, plannerRef, {
organizationId: work.organizationId,
workId: work._id,
});
return { status: "designing" as const };
},
});
const saveDesign = async (
ctx: { db: any },
ctx: MutationCtx,
work: Doc<"works">,
payloadJson: string,
createdBy: string
) => {
if (work.status === "executing")
if (work.status === "executing") {
throw new ConvexError("Cannot revise Work while a Run is executing");
if (work.definitionApprovalVersion !== work.definitionVersion)
}
if (work.definitionApprovalVersion !== work.definitionVersion) {
throw new ConvexError("Design must bind an approved Definition version");
}
const decoded = await Effect.runPromise(
validateDesignPacket({
...objectPayload(payloadJson),
version: (work.designVersion ?? 0) + 1,
definitionVersion: work.definitionVersion,
version: (work.designVersion ?? 0) + 1,
})
).catch((error: unknown) => {
throw new ConvexError(
@@ -372,51 +442,50 @@ const saveDesign = async (
});
const prior = await ctx.db
.query("designPackets")
.withIndex("by_work_and_version", (q: any) => q.eq("workId", work._id))
.withIndex("by_work_and_version", (q) => q.eq("workId", work._id))
.collect();
for (const row of prior)
if (row.status === "current")
for (const row of prior) {
if (row.status === "current") {
await ctx.db.patch(row._id, { status: "superseded" });
}
}
const designApprovals = await ctx.db
.query("workApprovals")
.withIndex("by_work_and_kind", (q: any) =>
.withIndex("by_workId_and_kind", (q) =>
q.eq("workId", work._id).eq("kind", "design")
)
.collect();
for (const approval of designApprovals)
if (approval.status === "active")
for (const approval of designApprovals) {
if (approval.status === "active") {
await ctx.db.patch(approval._id, { status: "invalidated" });
}
}
const designId = await ctx.db.insert("designPackets", {
workId: work._id,
version: decoded.version,
createdAt: Date.now(),
createdBy,
definitionVersion: decoded.definitionVersion,
payloadJson: JSON.stringify(decoded),
status: "current",
createdBy,
createdAt: Date.now(),
version: decoded.version,
workId: work._id,
});
const priorSlices = await ctx.db
.query("workSlices")
.withIndex("by_work_and_designVersion", (q: any) =>
q.eq("workId", work._id)
)
.collect();
for (const slice of priorSlices) await ctx.db.delete(slice._id);
for (const [ordinal, slice] of decoded.slices.entries())
for (const [ordinal, slice] of decoded.slices.entries()) {
await ctx.db.insert("workSlices", {
workId: work._id,
createdAt: Date.now(),
designVersion: decoded.version,
sliceId: slice.id,
ordinal,
title: slice.title,
objective: slice.objective,
observableBehavior: slice.observableBehavior,
ordinal,
payloadJson: JSON.stringify(slice),
sliceId: slice.id,
status: ordinal === 0 ? "ready" : "planned",
title: slice.title,
workId: work._id,
});
}
await ctx.db.patch(work._id, {
designVersion: decoded.version,
designApprovalVersion: undefined,
designVersion: decoded.version,
status: "awaiting-design-approval",
updatedAt: Date.now(),
});
@@ -432,7 +501,7 @@ const saveDesign = async (
};
export const saveDesignProposal = mutation({
args: { workId: v.id("works"), payloadJson: v.string() },
args: { payloadJson: v.string(), workId: v.id("works") },
handler: async (ctx, args) => {
const work = await requireWork(ctx, args.workId);
await requireProjectMember(ctx, work.projectId);
@@ -444,9 +513,9 @@ export const reviseDesign = saveDesignProposal;
export const approveDesign = mutation({
args: {
workId: v.id("works"),
definitionVersion: v.number(),
designVersion: v.number(),
workId: v.id("works"),
},
handler: async (ctx, args) => {
const work = await requireWork(ctx, args.workId);
@@ -455,24 +524,27 @@ export const approveDesign = mutation({
work.definitionApprovalVersion !== args.definitionVersion ||
work.designVersion !== args.designVersion ||
work.status !== "awaiting-design-approval"
)
) {
throw new ConvexError(
"Design approval must bind current Definition and Design versions"
);
}
const identity = await ctx.auth.getUserIdentity();
if (!identity) throw new ConvexError("Authentication required");
if (!identity) {
throw new ConvexError("Authentication required");
}
await ctx.db.insert("workApprovals", {
workId: work._id,
kind: "design",
approvedAt: Date.now(),
approvedBy: identity.tokenIdentifier,
definitionVersion: args.definitionVersion,
designVersion: args.designVersion,
approvedBy: identity.subject,
approvedAt: Date.now(),
kind: "design",
status: "active",
workId: work._id,
});
await ctx.db.patch(work._id, {
status: "ready",
designApprovalVersion: args.designVersion,
status: "ready",
updatedAt: Date.now(),
});
await appendEvent(
@@ -486,83 +558,168 @@ export const approveDesign = mutation({
});
export const submitDefinitionProposal = mutation({
args: { workId: v.id("works"), payloadJson: v.string(), token: v.string() },
args: { payloadJson: v.string(), token: v.string(), workId: v.id("works") },
handler: async (ctx, args) => {
if (args.token !== env.FLUE_DB_TOKEN)
if (args.token !== env.FLUE_DB_TOKEN) {
throw new ConvexError("Invalid agent control token");
}
const work = await requireWork(ctx, args.workId);
return await saveDefinition(ctx, work, args.payloadJson, "work-planner");
},
});
export const submitDesignProposal = mutation({
args: { workId: v.id("works"), payloadJson: v.string(), token: v.string() },
args: { payloadJson: v.string(), token: v.string(), workId: v.id("works") },
handler: async (ctx, args) => {
if (args.token !== env.FLUE_DB_TOKEN)
if (args.token !== env.FLUE_DB_TOKEN) {
throw new ConvexError("Invalid agent control token");
}
const work = await requireWork(ctx, args.workId);
return await saveDesign(ctx, work, args.payloadJson, "work-planner");
},
});
export const submitQuestion = mutation({
args: { workId: v.id("works"), questionJson: v.string(), token: v.string() },
args: { questionJson: v.string(), token: v.string(), workId: v.id("works") },
handler: async (ctx, args) => {
if (args.token !== env.FLUE_DB_TOKEN)
if (args.token !== env.FLUE_DB_TOKEN) {
throw new ConvexError("Invalid agent control token");
}
const work = await requireWork(ctx, args.workId);
if (work.definitionVersion === undefined)
if (work.definitionVersion === undefined) {
throw new ConvexError("Work has no Definition to attach a question to");
}
if (
work.status !== "defining" &&
work.status !== "awaiting-definition-approval"
) {
throw new ConvexError(
"Questions can only be submitted while the Work is being defined"
);
}
const question = await Effect.runPromise(
Schema.decodeUnknownEffect(WorkQuestion)(JSON.parse(args.questionJson))
Schema.decodeUnknownEffect(WorkQuestion)(parsePayload(args.questionJson))
).catch((error: unknown) => {
throw new ConvexError(
error instanceof Error ? error.message : "Invalid question"
);
});
await ctx.db.insert("workQuestions", {
workId: work._id,
definitionVersion: work.definitionVersion,
questionId: question.id,
prompt: question.prompt,
impact: question.impact,
recommendation: question.recommendation,
const existing = await ctx.db
.query("workQuestions")
.withIndex("by_workId_and_definitionVersion_and_questionId", (q) =>
q
.eq("workId", work._id)
.eq("definitionVersion", work.definitionVersion!)
.eq("questionId", question.id)
)
.unique();
const questionRow = {
alternativesJson: JSON.stringify(question.alternatives),
status: question.status,
answer: question.answer,
createdAt: Date.now(),
});
definitionVersion: work.definitionVersion,
impact: question.impact,
prompt: question.prompt,
questionId: question.id,
recommendation: question.recommendation,
status: question.status,
workId: work._id,
};
if (existing) {
const same =
existing.prompt === questionRow.prompt &&
existing.impact === questionRow.impact &&
existing.recommendation === questionRow.recommendation &&
existing.alternativesJson === questionRow.alternativesJson &&
existing.status === questionRow.status &&
existing.answer === questionRow.answer;
if (!same) {
throw new ConvexError("Question ID already has different content");
}
return { accepted: false };
}
await ctx.db.insert("workQuestions", questionRow);
await appendEvent(
ctx,
work._id,
"question.created",
`question-created:${work.definitionVersion}:${question.id}`,
question.id,
JSON.stringify(question)
);
return { accepted: true };
},
});
export const recordPlannerFailure = internalMutation({
args: { error: v.string(), workId: v.id("works") },
handler: async (ctx, args): Promise<null> => {
const work = await ctx.db.get(args.workId);
if (!work || (work.status !== "defining" && work.status !== "designing")) {
return null;
}
await ctx.db.patch(work._id, { status: "blocked", updatedAt: Date.now() });
await appendEvent(
ctx,
work._id,
"planner.failed",
`planner-failed:${work._id}:${work.status}`,
undefined,
JSON.stringify({ error: args.error, phase: work.status })
);
return null;
},
});
export const runPlanner = internalAction({
args: { organizationId: v.id("organizations"), workId: v.id("works") },
handler: async (ctx, args) => {
const flueUrl = (env as typeof env & { readonly FLUE_URL?: string })
.FLUE_URL;
if (!flueUrl) return null;
if (!flueUrl) {
await ctx.runMutation(plannerFailureRef, {
error: "FLUE_URL is not configured in Convex",
workId: args.workId,
});
return null;
}
const context = await ctx.runQuery(plannerContextRef, {
workId: args.workId,
});
if (!context) return null;
if (!context) {
return null;
}
const endpoint = new URL(
`agents/work-planner/${encodeURIComponent(String(args.organizationId))}`,
`${flueUrl.replace(/\/+$/u, "")}/`
);
endpoint.searchParams.set("wait", "result");
await fetch(endpoint, {
method: "POST",
headers: {
authorization: `Bearer ${env.FLUE_DB_TOKEN}`,
"content-type": "application/json",
"x-zopu-organization-id": String(args.organizationId),
"x-zopu-request-id": `work-planner:${args.workId}`,
},
body: JSON.stringify({
message: `Plan Work ${String(args.workId)} titled "${context.title}" with objective "${context.objective}". Current state is ${context.status}. If the Work is defining, submit only a Definition proposal and questions. If the Work is designing, submit only a Design Packet proposal bound to the approved Definition. Never approve or execute.`,
}),
});
try {
const response = await fetch(endpoint, {
body: JSON.stringify({
message: `Plan Work ${String(args.workId)} titled "${context.title}" with objective "${context.objective}". Current state is ${context.status}. If the Work is defining, submit only a Definition proposal and questions. If the Work is designing, submit only a Design Packet proposal bound to the approved Definition. Never approve or execute.`,
}),
headers: {
authorization: `Bearer ${env.FLUE_DB_TOKEN}`,
"content-type": "application/json",
"x-zopu-organization-id": String(args.organizationId),
"x-zopu-request-id": `work-planner:${args.workId}`,
},
method: "POST",
});
if (response.ok) {
return null;
}
await ctx.runMutation(plannerFailureRef, {
error: `Work Planner request failed (${response.status})`,
workId: args.workId,
});
} catch (error) {
await ctx.runMutation(plannerFailureRef, {
error: error instanceof Error ? error.message : String(error),
workId: args.workId,
});
}
// The private worker submits typed proposals through Convex mutations; it
// never approves or advances Work itself.
return null;
@@ -608,8 +765,10 @@ export const listForProject = query({
.take(10);
const slices = await ctx.db
.query("workSlices")
.withIndex("by_work_and_designVersion", (q: any) =>
q.eq("workId", work._id)
.withIndex("by_workId_and_designVersion", (q) =>
q
.eq("workId", work._id)
.eq("designVersion", work.designVersion ?? 0)
)
.collect();
const runs = await ctx.db
@@ -633,18 +792,18 @@ export const listForProject = query({
const signals = await Promise.all(
attachments.map(async (attachment) => {
const signal = await ctx.db.get(attachment.signalId);
if (!signal) return null;
if (!signal) {
return null;
}
const sources = await ctx.db
.query("signalSources")
.withIndex("by_signalId_and_ordinal", (q: any) =>
.withIndex("by_signalId_and_ordinal", (q) =>
q.eq("signalId", signal._id)
)
.collect();
return {
createdAt: signal.createdAt,
signalId: signal._id,
summary: signal.summary,
title: signal.title,
sources: sources
.sort((a, b) => a.ordinal - b.ordinal)
.map((source) => ({
@@ -653,6 +812,8 @@ export const listForProject = query({
rawText: source.rawTextSnapshot,
submissionId: null,
})),
summary: signal.summary,
title: signal.title,
};
})
);
@@ -664,16 +825,16 @@ export const listForProject = query({
);
return {
...work,
definitions,
designs,
slices: slices.sort((a, b) => a.ordinal - b.ordinal),
runs,
events,
signals: signals.filter((signal) => signal !== null),
definition: currentDefinition
? JSON.parse(currentDefinition.payloadJson)
: null,
definitions,
design: currentDesign ? JSON.parse(currentDesign.payloadJson) : null,
designs,
events,
runs,
signals: signals.filter((signal) => signal !== null),
slices: slices.sort((a, b) => a.ordinal - b.ordinal),
};
})
);

5
packages/backend/convex/works.ts
View File

@@ -8,7 +8,8 @@ import { ConvexError, v } from "convex/values";
import { Effect } from "effect";
import type { Doc, Id } from "./_generated/dataModel";
import { mutation, type MutationCtx, query } from "./_generated/server";
import { mutation, query } from "./_generated/server";
import type { MutationCtx } from "./_generated/server";
import { requireProjectMember } from "./authz";
const requireAgent = (token: string) => {
@@ -241,7 +242,6 @@ export const listForProject = query({
return {
createdAt: signal.createdAt,
signalId: signal._id,
summary: signal.summary,
sources: sources
.sort((a, b) => a.ordinal - b.ordinal)
.map((source) => ({
@@ -250,6 +250,7 @@ export const listForProject = query({
rawText: source.rawTextSnapshot,
submissionId: null,
})),
summary: signal.summary,
title: signal.title,
};
})

6
packages/backend/package.json
View File

@@ -8,8 +8,8 @@
"dev": "convex dev --env-file ../../.env",
"dev:setup": "convex dev --env-file ../../.env --configure --until-success",
"check-types": "tsc --noEmit -p convex/tsconfig.json",
"test": "vitest run",
"test:watch": "vitest"
"test": "CONVEX_SITE_URL=https://convex.test FLUE_DB_TOKEN=test-token SITE_URL=https://site.test vitest run",
"test:watch": "CONVEX_SITE_URL=https://convex.test FLUE_DB_TOKEN=test-token SITE_URL=https://site.test vitest"
},
"dependencies": {
"@better-auth/expo": "catalog:",
@@ -23,7 +23,7 @@
},
"devDependencies": {
"@code/config": "workspace:*",
"@types/node": "^24.3.0",
"@types/node": "catalog:",
"convex-test": "catalog:",
"typescript": "catalog:",
"vitest": "catalog:"