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

Compare commits

merge into: puter:t3code/review-slices-sandbox-next
Branches Tags
puter:master puter:feat/slice5-complete puter:feat/slice5-agentos-runner puter:t3code/slice-5-implementation puter:t3code/review-slices-sandbox-next puter:feat/auth-and-project-requests puter:t3code/explore-primitives-package puter:sai/changes
...
pull from: puter:t3code/slice-5-implementation
Branches Tags
puter:master puter:feat/slice5-complete puter:feat/slice5-agentos-runner puter:t3code/slice-5-implementation puter:t3code/review-slices-sandbox-next puter:feat/auth-and-project-requests puter:t3code/explore-primitives-package puter:sai/changes

2 Commits
t3code/rev ... t3code/sli

Author SHA1 Message Date
-Puter
a8d946b6a9 Slice 4 control-plane repairs, lint/catalog cleanup, Flue adapter fixes 
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
 2026-07-28 02:53:05 +05:30
puter
0e32c35515 Slice 4 hardening: bounded resolver, terminal states, reconciliation (#21) 
Fixes review findings before Slice 5. Resolver: resolveOutcome + WORK_STATUS_FOR_OUTCOME + lifecycle fixes. Execution: bounded finishAttempt retry, terminal Work status preservation, slice validation, bounded reconciliation with 30s cron. Planning: revision guards, durable submitQuestion, design-approval invalidation. Primitives 69 (+4), backend 22 (+10) tests; root typecheck/build/Ultracite pass.
 2026-07-27 18:39:09 +00:00
44 changed files with 3709 additions and 1397 deletions
Expand all files Collapse all files

8
apps/web/package.json
View File

@@ -26,14 +26,14 @@
"react-dom": "catalog:",
"react-router": "^8.1.0",
"sonner": "catalog:",
"streamdown": "2.5.0"
"streamdown": "catalog:"
},
"devDependencies": {
"@code/config": "workspace:*",
"@react-router/dev": "^8.1.0",
"@tailwindcss/vite": "^4.3.2",
"@types/node": "^22.13.14",
"@types/react": "^19.2.17",
"@tailwindcss/vite": "catalog:",
"@types/node": "catalog:",
"@types/react": "catalog:",
"@types/react-dom": "catalog:",
"react-router-devtools": "^6.2.1",
"tailwindcss": "catalog:",

680
bun.lock
View File

File diff suppressed because it is too large Load Diff

25
oxlint.config.ts
View File

@@ -5,5 +5,28 @@ import remix from "ultracite/oxlint/remix";
export default defineConfig({
extends: [core, react, remix],
ignorePatterns: core.ignorePatterns,
ignorePatterns: [...core.ignorePatterns, "repos/**", "scripts/**"],
overrides: [
{
files: ["**/convex/**/*.ts", "**/convex/**/*.test.ts"],
rules: {
"unicorn/filename-case": "off",
// Convex targets ES2022: no toSorted/at; sequential awaits ensure atomicity
"unicorn/no-array-sort": "off",
"no-await-in-loop": "off",
// Convex query builders use `any` in index callbacks
"@typescript-eslint/no-explicit-any": "off",
"@typescript-eslint/no-non-null-assertion": "off",
"eslint/complexity": "off",
},
},
{
files: ["**/convex/**/*.test.ts"],
rules: {
"unicorn/no-await-expression-member": "off",
"eslint/require-await": "off",
"sort-keys": "off",
},
},
],
});

16
package.json
View File

@@ -37,7 +37,15 @@
"heroui-native": "^1.0.5",
"vite": "^7.3.6",
"vitest": "^4.1.10",
"convex-test": "^0.0.54"
"convex-test": "^0.0.54",
"react-native": "0.86.0",
"@types/react": "^19.2.17",
"@types/node": "^22.13.14",
"hono": "^4.8.3",
"valibot": "^1.4.2",
"streamdown": "2.5.0",
"@tailwindcss/postcss": "^4.3.2",
"@tailwindcss/vite": "^4.3.2"
}
},
"type": "module",
@@ -45,8 +53,8 @@
"dev": "vp run -r dev",
"build": "vp run -r build",
"check-types": "vp run -r check-types",
"check": "ultracite check package.json vite.config.ts apps/web/package.json apps/web/src/root.tsx apps/web/src/index.css apps/web/src/components/chat apps/web/src/components/slice-one apps/web/src/hooks/chat apps/web/src/hooks/slice-one apps/web/src/lib/chat apps/web/src/lib/slice-one apps/web/src/routes.ts apps/web/src/routes/app/dashboard/page.tsx apps/web/src/routes/app/mobile/page.tsx packages/agents/package.json packages/agents/src/agents/zopu.ts packages/agents/src/app.ts packages/agents/src/auth.ts packages/agents/src/tools/slice-one.ts packages/backend/package.json packages/backend/convex/conversationMessages.ts packages/backend/convex/conversationMessages.test.ts packages/backend/convex/projects.ts packages/backend/convex/schema.ts packages/backend/convex/signalRouting.ts packages/backend/convex/signalRouting.test.ts packages/backend/convex/works.ts packages/backend/convex/works.test.ts packages/primitives/src/work.ts packages/primitives/src/work.test.ts",
"lint": "vp lint",
"check": "ultracite check package.json vite.config.ts apps/web/package.json apps/web/src/root.tsx apps/web/src/index.css apps/web/src/components/chat apps/web/src/components/slice-one apps/web/src/hooks/chat apps/web/src/hooks/slice-one apps/web/src/lib/chat apps/web/src/lib/slice-one apps/web/src/routes.ts apps/web/src/routes/app/dashboard/page.tsx apps/web/src/routes/app/mobile/page.tsx packages/agents/package.json packages/agents/src/agents/zopu.ts packages/agents/src/app.ts packages/agents/src/auth.ts packages/agents/src/tools/slice-one.ts packages/backend/package.json packages/backend/convex/conversationMessages.ts packages/backend/convex/conversationMessages.test.ts packages/backend/convex/fluePersistence.test.ts packages/backend/convex/projects.ts packages/backend/convex/schema.ts packages/backend/convex/signalRouting.ts packages/backend/convex/signalRouting.test.ts packages/backend/convex/works.ts packages/backend/convex/works.test.ts packages/backend/convex/workArtifacts.ts packages/backend/convex/workArtifacts.test.ts packages/backend/convex/workExecution.ts packages/backend/convex/workExecution.test.ts packages/backend/convex/workPlanning.ts packages/backend/convex/workPlanning.test.ts packages/backend/convex/crons.ts packages/primitives/src/work.ts packages/primitives/src/work.test.ts packages/primitives/src/work-artifact.ts packages/primitives/src/work-artifact.test.ts packages/primitives/src/resolver.ts packages/primitives/src/work-lifecycle.ts packages/primitives/src/work-resolution.test.ts",
"lint": "oxlint --disable-nested-config",
"format": "vp fmt",
"staged": "vp staged",
"hooks:setup": "vp config",
@@ -72,7 +80,7 @@
"@code/primitives": "workspace:*",
"@flue/sdk": "1.0.0-beta.9",
"@types/bun": "catalog:",
"@types/node": "^22.13.14",
"@types/node": "catalog:",
"convex": "catalog:",
"convex-test": "catalog:",
"effect": "catalog:",

4
packages/agents/package.json
View File

@@ -17,8 +17,8 @@
"@code/env": "workspace:*",
"@flue/runtime": "latest",
"convex": "catalog:",
"hono": "4.12.31",
"valibot": "^1.4.2"
"hono": "catalog:",
"valibot": "catalog:"
},
"devDependencies": {
"@code/config": "workspace:*",

8
packages/agents/src/app.ts
View File

@@ -1,7 +1,7 @@
import { parseAgentEnv } from "@code/env/agent";
import { registerProvider } from "@flue/runtime";
import type { Fetchable } from "@flue/runtime/routing";
import { flue } from "@flue/runtime/routing";
import { Hono } from "hono";
const agentEnv = parseAgentEnv(process.env);
@@ -24,7 +24,5 @@ registerProvider(agentEnv.AGENT_MODEL_PROVIDER, {
},
});
const app = new Hono();
app.route("/", flue() as unknown as Hono);
export default app;
// flue() returns a complete Hono app; the Fetchable contract accepts it.
export default flue() satisfies Fetchable;

384
packages/agents/src/db.ts
View File

@@ -1,3 +1,4 @@
/* eslint-disable max-classes-per-file, @typescript-eslint/no-invalid-void-type, class-methods-use-this, @typescript-eslint/parameter-properties */
/**
* Convex-backed Flue {@link PersistenceAdapter} for the agents package.
*
@@ -25,70 +26,11 @@
*/
import { env } from "@code/env/server";
import {
type AgentAttemptMarker,
type AgentDispatchAdmission,
type AgentDispatchReceipt,
type AgentExecutionStore,
type AgentSubmission,
type AgentSubmissionStore,
AttachmentConflictError,
type AttachmentRef,
type AttachmentStore,
type ConversationProducerClaim,
type ConversationRecord,
type ConversationStreamBatch,
type ConversationStreamIdentity,
type ConversationStreamMeta,
type ConversationStreamReadResult,
type ConversationStreamStore,
type CreateRunInput,
DEFAULT_LIST_LIMIT,
DEFAULT_READ_LIMIT,
type DirectAgentSubmissionInput,
type DispatchAgentSubmissionInput,
type DispatchInput,
type EndRunInput,
type EventStreamMeta,
type EventStreamReadResult,
type EventStreamStore,
type GetAttachmentInput,
MAX_LIST_LIMIT,
MAX_READ_LIMIT,
type PersistedChunkRow,
type PersistenceAdapter,
type PersistenceStores,
type PutAttachmentInput,
type RunPointer,
type RunRecord,
type RunStore,
type RunStatus,
StreamListenerRegistry,
type StoredAttachment,
type SubmissionAttemptRef,
type SubmissionClaimRef,
type SubmissionDurability,
type SubmissionSettlementObligation,
type SubmissionSettledRecord,
assertSupportedFlueSchemaVersion,
clampLimit,
copyAttachmentBytes,
createSessionStorageKey,
decodeRunCursor,
encodeRunCursor,
formatOffset,
hydratePersistedDirectSubmission,
parseOffset,
prepareDirectSubmission,
verifyAttachmentBytes,
} from "@flue/runtime/adapter";
import { AttachmentConflictError, DEFAULT_LIST_LIMIT, DEFAULT_READ_LIMIT, MAX_LIST_LIMIT, MAX_READ_LIMIT, StreamListenerRegistry, assertSupportedFlueSchemaVersion, clampLimit, copyAttachmentBytes, createSessionStorageKey, decodeRunCursor, encodeRunCursor, formatOffset, hydratePersistedDirectSubmission, parseOffset, prepareDirectSubmission, verifyAttachmentBytes } from '@flue/runtime/adapter';
import type { AgentAttemptMarker, AgentDispatchAdmission, AgentDispatchReceipt, AgentExecutionStore, AgentSubmission, AgentSubmissionStore, AttachmentRef, AttachmentStore, ConversationProducerClaim, ConversationRecord, ConversationStreamBatch, ConversationStreamIdentity, ConversationStreamMeta, ConversationStreamReadResult, ConversationStreamStore, CreateRunInput, DirectAgentSubmissionInput, DispatchAgentSubmissionInput, DispatchInput, EndRunInput, EventStreamMeta, EventStreamReadResult, EventStreamStore, GetAttachmentInput, PersistedChunkRow, PersistenceAdapter, PersistenceStores, PutAttachmentInput, RunPointer, RunRecord, RunStore, RunStatus, StoredAttachment, SubmissionAttemptRef, SubmissionClaimRef, SubmissionDurability, SubmissionSettlementObligation, SubmissionSettledRecord } from '@flue/runtime/adapter';
import { ConvexHttpClient } from "convex/browser";
import {
type FunctionArgs,
type FunctionReference,
type FunctionReturnType,
makeFunctionReference,
} from "convex/server";
import { makeFunctionReference } from 'convex/server';
import type { FunctionArgs, FunctionReference, FunctionReturnType } from 'convex/server';
// ---------------------------------------------------------------------------
// Token + arg types
@@ -100,7 +42,7 @@ import {
// ---------------------------------------------------------------------------
/** Auth token present on every Convex function call. */
type TokenArgs = { readonly token: string; readonly [key: string]: unknown };
interface TokenArgs { readonly token: string; readonly [key: string]: unknown }
const TOKEN = (): string => env.FLUE_DB_TOKEN;
@@ -112,7 +54,7 @@ type TraceCarrier = NonNullable<DirectAgentSubmissionInput["traceCarrier"]>;
// ---------------------------------------------------------------------------
/** Common fields stored for every submission row. */
type SubmissionRow = {
interface SubmissionRow {
readonly sequence: number;
readonly submissionId: string;
readonly sessionKey: string;
@@ -134,42 +76,42 @@ type SubmissionRow = {
readonly ownerId: string | null;
readonly leaseExpiresAt: number;
readonly traceCarrierJson: string | null;
};
}
type SettlementObligationRow = {
interface SettlementObligationRow {
readonly submissionId: string;
readonly sessionKey: string;
readonly attemptId: string;
readonly recordId: string;
readonly recordJson: string;
};
}
type AttemptMarkerRow = {
interface AttemptMarkerRow {
readonly submissionId: string;
readonly attemptId: string;
readonly createdAt: number;
};
}
type ConversationStreamRow = {
interface ConversationStreamRow {
readonly identity: ConversationStreamIdentity;
readonly incarnation: string;
readonly nextOffset: number;
readonly producerId: string | null;
readonly producerEpoch: number;
readonly nextProducerSequence: number;
};
}
type ConversationBatchRow = {
interface ConversationBatchRow {
readonly offset: number;
readonly recordsJson: string;
};
}
type EventEntryRow = {
interface EventEntryRow {
readonly offset: number;
readonly dataJson: string;
};
}
type RunRow = {
interface RunRow {
readonly runId: string;
readonly workflowName: string;
readonly status: RunStatus;
@@ -181,9 +123,9 @@ type RunRow = {
readonly resultJson: string | null;
readonly errorJson: string | null;
readonly traceCarrierJson: string | null;
};
}
type RunPointerRow = {
interface RunPointerRow {
readonly runId: string;
readonly workflowName: string;
readonly status: RunStatus;
@@ -191,15 +133,15 @@ type RunPointerRow = {
readonly endedAt: string | null;
readonly durationMs: number | null;
readonly isError: boolean | null;
};
}
type AttachmentRefWire = {
interface AttachmentRefWire {
readonly id: string;
readonly mimeType: string;
readonly size: number;
readonly digest: string;
readonly filename?: string;
};
}
// ---------------------------------------------------------------------------
// Typed function references
@@ -263,7 +205,7 @@ const replaceSubmissionAttempt = makeFunctionReference<
>("fluePersistence:replaceSubmissionAttempt");
/** Unified admission envelope — `kind` discriminates dispatch vs direct. */
type AdmitSubmissionEnvelope = {
interface AdmitSubmissionEnvelope {
readonly kind: "dispatch" | "direct";
readonly submissionId: string;
readonly sessionKey: string;
@@ -271,12 +213,12 @@ type AdmitSubmissionEnvelope = {
readonly inputJson: string;
readonly chunksJson?: string;
readonly traceCarrierJson?: string;
};
type AdmitSubmissionResponse = {
}
interface AdmitSubmissionResponse {
readonly kind: "submission" | "retained_receipt" | "conflict";
readonly submission?: SubmissionRow;
readonly receipt?: AgentDispatchReceipt;
};
}
const admitSubmission = makeFunctionReference<
"mutation",
TokenArgs & { readonly input: AdmitSubmissionEnvelope },
@@ -410,7 +352,7 @@ type AppendBatchArgs = TokenArgs & {
readonly submission?: { readonly submissionId: string; readonly attemptId: string };
readonly recordsJson: string;
};
type AppendResult = { readonly offset: number; readonly appended: boolean };
interface AppendResult { readonly offset: number; readonly appended: boolean }
const appendConversationBatch = makeFunctionReference<
"mutation",
AppendBatchArgs,
@@ -422,11 +364,11 @@ type ReadBatchesArgs = TokenArgs & {
readonly afterOffset: number;
readonly limit: number;
};
type ReadBatchesResult = {
interface ReadBatchesResult {
readonly batches: ConversationBatchRow[];
readonly nextOffset: number;
readonly upToDate: boolean;
};
}
const readConversationBatches = makeFunctionReference<
"query",
ReadBatchesArgs,
@@ -478,12 +420,12 @@ type ReadEventsArgs = TokenArgs & {
readonly afterOffset: number;
readonly limit: number;
};
type ReadEventsResult = {
interface ReadEventsResult {
readonly events: EventEntryRow[];
readonly nextOffset: number;
readonly upToDate: boolean;
readonly closed: boolean;
};
}
const readEventsFn = makeFunctionReference<
"query",
ReadEventsArgs,
@@ -496,7 +438,7 @@ const closeEventStream = makeFunctionReference<
void
>("fluePersistence:closeEventStream");
type EventStreamMetaRow = { readonly nextOffset: number; readonly closed: boolean };
interface EventStreamMetaRow { readonly nextOffset: number; readonly closed: boolean }
const getEventStreamMeta = makeFunctionReference<
"query",
TokenArgs & { readonly path: string },
@@ -545,10 +487,10 @@ type ListRunsArgs = TokenArgs & {
readonly limit: number;
readonly cursor?: { readonly startedAt: string; readonly runId: string };
};
type ListRunsResult = {
interface ListRunsResult {
readonly rows: RunPointerRow[];
readonly hasMore: boolean;
};
}
const listRuns = makeFunctionReference<"query", ListRunsArgs, ListRunsResult>(
"fluePersistence:listRuns",
);
@@ -571,10 +513,10 @@ type GetAttachmentArgs = TokenArgs & {
readonly conversationId: string;
readonly attachmentId: string;
};
type AttachmentWire = {
interface AttachmentWire {
readonly attachment: AttachmentRefWire;
readonly bytes: ArrayBuffer;
};
}
const getAttachment = makeFunctionReference<
"query",
GetAttachmentArgs,
@@ -592,7 +534,7 @@ const deleteAttachmentsForInstance = makeFunctionReference<
// ---------------------------------------------------------------------------
const safeJsonParse = <T>(text: string | null | undefined): T | undefined => {
if (text === null || text === undefined || text === "") return undefined;
if (text === null || text === undefined || text === "") {return undefined;}
return JSON.parse(text) as T;
};
@@ -601,7 +543,7 @@ const jsonStringify = (value: unknown): string => JSON.stringify(value);
const parseAcceptedAt = (value: string, label: string): number => {
const ms = Date.parse(value);
if (!Number.isFinite(ms)) {
throw new Error(
throw new TypeError(
`[flue] ${label} produced non-finite acceptedAt: ${value}`,
);
}
@@ -613,18 +555,18 @@ const afterOffset = (offset: string | undefined): number =>
offset === undefined ? -1 : parseOffset(offset);
const wireRefToAttachmentRef = (wire: AttachmentRefWire): AttachmentRef => ({
digest: wire.digest,
id: wire.id,
mimeType: wire.mimeType,
size: wire.size,
digest: wire.digest,
...(wire.filename === undefined ? {} : { filename: wire.filename }),
});
const attachmentRefToWire = (ref: AttachmentRef): AttachmentRefWire => ({
digest: ref.digest,
id: ref.id,
mimeType: ref.mimeType,
size: ref.size,
digest: ref.digest,
...(ref.filename === undefined ? {} : { filename: ref.filename }),
});
@@ -636,17 +578,17 @@ const attachmentRefToWire = (ref: AttachmentRef): AttachmentRefWire => ({
*/
const hydrateSubmission = (row: SubmissionRow): AgentSubmission => {
const baseSubmission: Omit<AgentSubmission, "input"> = {
sequence: row.sequence,
submissionId: row.submissionId,
sessionKey: row.sessionKey,
kind: row.kind,
status: row.status,
acceptedAt: row.acceptedAt,
canonicalReadyAt: row.canonicalReadyAt,
attemptCount: row.attemptCount,
maxRetry: row.maxRetry,
timeoutAt: row.timeoutAt,
canonicalReadyAt: row.canonicalReadyAt,
kind: row.kind,
leaseExpiresAt: row.leaseExpiresAt,
maxRetry: row.maxRetry,
sequence: row.sequence,
sessionKey: row.sessionKey,
status: row.status,
submissionId: row.submissionId,
timeoutAt: row.timeoutAt,
...(row.attemptId === null ? {} : { attemptId: row.attemptId }),
...(row.inputAppliedAt === null
? {}
@@ -699,19 +641,19 @@ const hydrateObligation = (
);
}
return {
submissionId: row.submissionId,
sessionKey: row.sessionKey,
attemptId: row.attemptId,
recordId: row.recordId,
record,
recordId: row.recordId,
sessionKey: row.sessionKey,
submissionId: row.submissionId,
};
};
const toRunRecord = (row: RunRow): RunRecord => ({
runId: row.runId,
workflowName: row.workflowName,
status: row.status,
startedAt: row.startedAt,
status: row.status,
workflowName: row.workflowName,
...(row.endedAt === null ? {} : { endedAt: row.endedAt }),
...(row.isError === null ? {} : { isError: row.isError }),
...(row.durationMs === null ? {} : { durationMs: row.durationMs }),
@@ -725,9 +667,9 @@ const toRunRecord = (row: RunRow): RunRecord => ({
const toRunPointer = (row: RunPointerRow): RunPointer => ({
runId: row.runId,
workflowName: row.workflowName,
status: row.status,
startedAt: row.startedAt,
status: row.status,
workflowName: row.workflowName,
...(row.endedAt === null ? {} : { endedAt: row.endedAt }),
...(row.durationMs === null ? {} : { durationMs: row.durationMs }),
...(row.isError === null ? {} : { isError: row.isError }),
@@ -772,13 +714,13 @@ class ConvexAgentSubmissionStore implements AgentSubmissionStore {
async getSubmission(submissionId: string): Promise<AgentSubmission | null> {
const row = await this.convex.query(getSubmission, {
token: TOKEN(),
submissionId,
token: TOKEN(),
});
return row === null ? null : hydrateSubmission(row);
}
async hasUnsettledSubmissions(): Promise<boolean> {
hasUnsettledSubmissions(): Promise<boolean> {
return this.convex.query(hasUnsettledSubmissions, { token: TOKEN() });
}
@@ -818,10 +760,10 @@ class ConvexAgentSubmissionStore implements AgentSubmissionStore {
lease?: { ownerId: string; leaseExpiresAt: number },
): Promise<AgentSubmission | null> {
const row = await this.convex.mutation(replaceSubmissionAttempt, {
token: TOKEN(),
submissionId: attempt.submissionId,
attemptId: attempt.attemptId,
nextAttemptId,
submissionId: attempt.submissionId,
token: TOKEN(),
...(lease === undefined ? {} : lease),
});
return row === null ? null : hydrateSubmission(row);
@@ -833,25 +775,25 @@ class ConvexAgentSubmissionStore implements AgentSubmissionStore {
DispatchAgentSubmissionInput,
"kind" | "submissionId"
> = {
dispatchId: input.dispatchId,
acceptedAt: input.acceptedAt,
agent: input.agent,
dispatchId: input.dispatchId,
id: input.id,
input: input.input,
acceptedAt: input.acceptedAt,
};
const res = await this.convex.mutation(admitSubmission, {
token: TOKEN(),
input: {
kind: "dispatch",
submissionId: input.dispatchId,
sessionKey,
acceptedAt: parseAcceptedAt(input.acceptedAt, "admitDispatch"),
inputJson: jsonStringify({
kind: "dispatch",
submissionId: input.dispatchId,
...submissionInput,
}),
kind: "dispatch",
sessionKey,
submissionId: input.dispatchId,
},
token: TOKEN(),
});
if (res.kind === "submission" && res.submission !== undefined) {
return {
@@ -871,18 +813,18 @@ class ConvexAgentSubmissionStore implements AgentSubmissionStore {
const sessionKey = createSessionStorageKey(input.id, "default", "default");
const extracted = prepareDirectSubmission(input);
const res = await this.convex.mutation(admitSubmission, {
token: TOKEN(),
input: {
kind: "direct",
submissionId: input.submissionId,
sessionKey,
acceptedAt: parseAcceptedAt(input.acceptedAt, "admitDirect"),
inputJson: jsonStringify(extracted.value),
chunksJson: jsonStringify(extracted.chunks),
inputJson: jsonStringify(extracted.value),
kind: "direct",
sessionKey,
submissionId: input.submissionId,
...(input.traceCarrier === undefined
? {}
: { traceCarrierJson: jsonStringify(input.traceCarrier) }),
},
token: TOKEN(),
});
if (res.kind === "submission" && res.submission !== undefined) {
return hydrateSubmission(res.submission);
@@ -899,8 +841,8 @@ class ConvexAgentSubmissionStore implements AgentSubmissionStore {
submissionId: string,
): Promise<AgentSubmission | null> {
const row = await this.convex.mutation(markSubmissionCanonicalReady, {
token: TOKEN(),
submissionId,
token: TOKEN(),
});
return row === null ? null : hydrateSubmission(row);
}
@@ -909,51 +851,51 @@ class ConvexAgentSubmissionStore implements AgentSubmissionStore {
claim: SubmissionClaimRef,
): Promise<AgentSubmission | null> {
const row = await this.convex.mutation(claimSubmission, {
token: TOKEN(),
submissionId: claim.submissionId,
attemptId: claim.attemptId,
ownerId: claim.ownerId,
leaseExpiresAt: claim.leaseExpiresAt,
ownerId: claim.ownerId,
submissionId: claim.submissionId,
token: TOKEN(),
});
return row === null ? null : hydrateSubmission(row);
}
async markSubmissionInputApplied(
markSubmissionInputApplied(
attempt: SubmissionAttemptRef,
durability?: SubmissionDurability,
): Promise<boolean> {
return this.convex.mutation(markSubmissionInputApplied, {
token: TOKEN(),
submissionId: attempt.submissionId,
attemptId: attempt.attemptId,
submissionId: attempt.submissionId,
token: TOKEN(),
...(durability === undefined ? {} : durability),
});
}
async requestSubmissionRecovery(
requestSubmissionRecovery(
attempt: SubmissionAttemptRef,
): Promise<boolean> {
return this.convex.mutation(requestSubmissionRecovery, {
token: TOKEN(),
submissionId: attempt.submissionId,
attemptId: attempt.attemptId,
});
}
async requestSessionAbort(sessionKey: string): Promise<string[]> {
return this.convex.mutation(requestSessionAbort, {
submissionId: attempt.submissionId,
token: TOKEN(),
sessionKey,
});
}
async requeueSubmissionBeforeInputApplied(
requestSessionAbort(sessionKey: string): Promise<string[]> {
return this.convex.mutation(requestSessionAbort, {
sessionKey,
token: TOKEN(),
});
}
requeueSubmissionBeforeInputApplied(
attempt: SubmissionAttemptRef,
): Promise<boolean> {
return this.convex.mutation(requeueSubmissionBeforeInputApplied, {
token: TOKEN(),
submissionId: attempt.submissionId,
attemptId: attempt.attemptId,
submissionId: attempt.submissionId,
token: TOKEN(),
});
}
@@ -962,62 +904,62 @@ class ConvexAgentSubmissionStore implements AgentSubmissionStore {
settlement: { recordId: string; record: SubmissionSettledRecord },
): Promise<SubmissionSettlementObligation | null> {
const row = await this.convex.mutation(reserveSubmissionSettlement, {
token: TOKEN(),
submissionId: attempt.submissionId,
attemptId: attempt.attemptId,
recordId: settlement.recordId,
recordJson: jsonStringify(settlement.record),
submissionId: attempt.submissionId,
token: TOKEN(),
});
return row === null ? null : hydrateObligation(row);
}
async finalizeSubmissionSettlement(
finalizeSubmissionSettlement(
attempt: SubmissionAttemptRef,
recordId: string,
): Promise<boolean> {
return this.convex.mutation(finalizeSubmissionSettlement, {
token: TOKEN(),
submissionId: attempt.submissionId,
attemptId: attempt.attemptId,
recordId,
submissionId: attempt.submissionId,
token: TOKEN(),
});
}
async completeSubmission(attempt: SubmissionAttemptRef): Promise<boolean> {
completeSubmission(attempt: SubmissionAttemptRef): Promise<boolean> {
return this.convex.mutation(settleSubmission, {
token: TOKEN(),
submissionId: attempt.submissionId,
attemptId: attempt.attemptId,
outcome: "completed",
submissionId: attempt.submissionId,
token: TOKEN(),
});
}
async failSubmission(
failSubmission(
attempt: SubmissionAttemptRef,
error: unknown,
): Promise<boolean> {
return this.convex.mutation(settleSubmission, {
token: TOKEN(),
submissionId: attempt.submissionId,
attemptId: attempt.attemptId,
outcome: "failed",
errorJson: jsonStringify(error),
outcome: "failed",
submissionId: attempt.submissionId,
token: TOKEN(),
});
}
async insertAttemptMarker(attempt: SubmissionAttemptRef): Promise<void> {
await this.convex.mutation(insertAttemptMarker, {
token: TOKEN(),
submissionId: attempt.submissionId,
attemptId: attempt.attemptId,
submissionId: attempt.submissionId,
token: TOKEN(),
});
}
async deleteAttemptMarker(attempt: SubmissionAttemptRef): Promise<void> {
await this.convex.mutation(deleteAttemptMarker, {
token: TOKEN(),
submissionId: attempt.submissionId,
attemptId: attempt.attemptId,
submissionId: attempt.submissionId,
token: TOKEN(),
});
}
@@ -1026,17 +968,17 @@ class ConvexAgentSubmissionStore implements AgentSubmissionStore {
token: TOKEN(),
});
return rows.map((row) => ({
submissionId: row.submissionId,
attemptId: row.attemptId,
createdAt: row.createdAt,
submissionId: row.submissionId,
}));
}
async renewLeases(ownerId: string, submissionIds: string[]): Promise<void> {
await this.convex.mutation(renewLeases, {
token: TOKEN(),
ownerId,
submissionIds,
token: TOKEN(),
});
}
@@ -1062,20 +1004,20 @@ class ConvexConversationStreamStore implements ConversationStreamStore {
identity: ConversationStreamIdentity,
): Promise<void> {
await this.convex.mutation(createConversationStream, {
token: TOKEN(),
path,
identity,
path,
token: TOKEN(),
});
}
async acquireProducer(
acquireProducer(
path: string,
producerId: string,
): Promise<ConversationProducerClaim> {
return this.convex.mutation(acquireConversationProducer, {
token: TOKEN(),
path,
producerId,
token: TOKEN(),
});
}
@@ -1089,14 +1031,14 @@ class ConvexConversationStreamStore implements ConversationStreamStore {
records: readonly ConversationRecord[];
}): Promise<{ offset: string }> {
const result = await this.convex.mutation(appendConversationBatch, {
token: TOKEN(),
path: input.path,
producerId: input.producerId,
producerEpoch: input.producerEpoch,
incarnation: input.incarnation,
path: input.path,
producerEpoch: input.producerEpoch,
producerId: input.producerId,
producerSequence: input.producerSequence,
...(input.submission === undefined ? {} : { submission: input.submission }),
recordsJson: jsonStringify(input.records),
token: TOKEN(),
...(input.submission === undefined ? {} : { submission: input.submission }),
});
if (result.appended) {
this.listeners.notify(input.path);
@@ -1114,10 +1056,10 @@ class ConvexConversationStreamStore implements ConversationStreamStore {
MAX_READ_LIMIT,
);
const result = await this.convex.query(readConversationBatches, {
token: TOKEN(),
path,
afterOffset: afterOffset(options?.offset),
limit,
path,
token: TOKEN(),
});
const batches: ConversationStreamBatch[] = result.batches.map((row) => ({
offset: formatOffset(row.offset),
@@ -1132,24 +1074,24 @@ class ConvexConversationStreamStore implements ConversationStreamStore {
async getMeta(path: string): Promise<ConversationStreamMeta | null> {
const row = await this.convex.query(getConversationStreamMeta, {
token: TOKEN(),
path,
token: TOKEN(),
});
if (row === null) return null;
if (row === null) {return null;}
return {
identity: row.identity,
incarnation: row.incarnation,
nextOffset: formatOffset(row.nextOffset),
producerId: row.producerId,
producerEpoch: row.producerEpoch,
nextProducerSequence: row.nextProducerSequence,
producerEpoch: row.producerEpoch,
producerId: row.producerId,
};
}
async delete(path: string): Promise<void> {
await this.convex.mutation(deleteConversationStream, {
token: TOKEN(),
path,
token: TOKEN(),
});
}
@@ -1169,16 +1111,16 @@ class ConvexEventStreamStore implements EventStreamStore {
async createStream(path: string): Promise<void> {
await this.convex.mutation(createEventStream, {
token: TOKEN(),
path,
token: TOKEN(),
});
}
async appendEvent(path: string, event: unknown): Promise<string> {
const result = await this.convex.mutation(appendEventFn, {
token: TOKEN(),
path,
dataJson: jsonStringify(event),
path,
token: TOKEN(),
});
if (result.appended) {
this.listeners.notify(path);
@@ -1192,10 +1134,10 @@ class ConvexEventStreamStore implements EventStreamStore {
event: unknown,
): Promise<string> {
const result = await this.convex.mutation(appendEventOnce, {
token: TOKEN(),
path,
key,
dataJson: jsonStringify(event),
key,
path,
token: TOKEN(),
});
if (result.appended) {
this.listeners.notify(path);
@@ -1209,38 +1151,38 @@ class ConvexEventStreamStore implements EventStreamStore {
): Promise<EventStreamReadResult> {
const limit = clampLimit(opts?.limit, DEFAULT_READ_LIMIT, MAX_READ_LIMIT);
const result = await this.convex.query(readEventsFn, {
token: TOKEN(),
path,
afterOffset: afterOffset(opts?.offset),
limit,
path,
token: TOKEN(),
});
return {
closed: result.closed,
events: result.events.map((row) => ({
offset: formatOffset(row.offset),
data: safeJsonParse(row.dataJson),
offset: formatOffset(row.offset),
})),
nextOffset: formatOffset(result.nextOffset),
upToDate: result.upToDate,
closed: result.closed,
};
}
async closeStream(path: string): Promise<void> {
await this.convex.mutation(closeEventStream, {
token: TOKEN(),
path,
token: TOKEN(),
});
}
async getStreamMeta(path: string): Promise<EventStreamMeta | null> {
const row = await this.convex.query(getEventStreamMeta, {
token: TOKEN(),
path,
token: TOKEN(),
});
if (row === null) return null;
if (row === null) {return null;}
return {
nextOffset: formatOffset(row.nextOffset),
closed: row.closed,
nextOffset: formatOffset(row.nextOffset),
};
}
@@ -1258,11 +1200,11 @@ class ConvexRunStore implements RunStore {
async createRun(input: CreateRunInput): Promise<void> {
await this.convex.mutation(createRun, {
token: TOKEN(),
runId: input.runId,
workflowName: input.workflowName,
startedAt: input.startedAt,
inputJson: jsonStringify(input.input),
runId: input.runId,
startedAt: input.startedAt,
token: TOKEN(),
workflowName: input.workflowName,
...(input.traceCarrier === undefined
? {}
: { traceCarrierJson: jsonStringify(input.traceCarrier) }),
@@ -1271,11 +1213,11 @@ class ConvexRunStore implements RunStore {
async endRun(input: EndRunInput): Promise<void> {
await this.convex.mutation(endRun, {
token: TOKEN(),
runId: input.runId,
durationMs: input.durationMs,
endedAt: input.endedAt,
isError: input.isError,
durationMs: input.durationMs,
runId: input.runId,
token: TOKEN(),
...(input.result === undefined
? {}
: { resultJson: jsonStringify(input.result) }),
@@ -1286,14 +1228,14 @@ class ConvexRunStore implements RunStore {
}
async getRun(runId: string): Promise<RunRecord | null> {
const row = await this.convex.query(getRun, { token: TOKEN(), runId });
const row = await this.convex.query(getRun, { runId, token: TOKEN() });
return row === null ? null : toRunRecord(row);
}
async lookupRun(
lookupRun(
runId: string,
): Promise<{ runId: string; workflowName: string } | null> {
return this.convex.query(lookupRun, { token: TOKEN(), runId });
return this.convex.query(lookupRun, { runId, token: TOKEN() });
}
async listRuns(opts?: {
@@ -1314,10 +1256,10 @@ class ConvexRunStore implements RunStore {
...(decoded === undefined ? {} : { cursor: decoded }),
});
const runs = result.rows.map(toRunPointer);
const last = runs[runs.length - 1];
const last = runs.at(-1);
const nextCursor =
result.hasMore && last !== undefined
? encodeRunCursor({ startedAt: last.startedAt, runId: last.runId })
? encodeRunCursor({ runId: last.runId, startedAt: last.startedAt })
: undefined;
return { runs, ...(nextCursor === undefined ? {} : { nextCursor }) };
}
@@ -1336,31 +1278,31 @@ class ConvexAttachmentStore implements AttachmentStore {
// payload cannot be mutated after the call returns.
const bytes = copyAttachmentBytes(input.bytes);
const status = await this.convex.mutation(putAttachment, {
token: TOKEN(),
streamPath: input.streamPath,
conversationId: input.conversationId,
attachment: attachmentRefToWire(input.attachment),
bytes: bytes.buffer.slice(
bytes.byteOffset,
bytes.byteOffset + bytes.byteLength,
) as ArrayBuffer,
conversationId: input.conversationId,
streamPath: input.streamPath,
token: TOKEN(),
});
if (status === "conflict") {
throw new AttachmentConflictError({
path: input.streamPath,
attachmentId: input.attachment.id,
path: input.streamPath,
});
}
}
async get(input: GetAttachmentInput): Promise<StoredAttachment | null> {
const row = await this.convex.query(getAttachment, {
token: TOKEN(),
streamPath: input.streamPath,
conversationId: input.conversationId,
attachmentId: input.attachmentId,
conversationId: input.conversationId,
streamPath: input.streamPath,
token: TOKEN(),
});
if (row === null) return null;
if (row === null) {return null;}
const attachment = wireRefToAttachmentRef(row.attachment);
const bytes = new Uint8Array(row.bytes);
// Verify integrity before handing bytes back to the runtime; throws
@@ -1374,8 +1316,8 @@ class ConvexAttachmentStore implements AttachmentStore {
async deleteForInstance(streamPath: string): Promise<void> {
await this.convex.mutation(deleteAttachmentsForInstance, {
token: TOKEN(),
streamPath,
token: TOKEN(),
});
}
}
@@ -1409,11 +1351,11 @@ class ConvexPersistenceAdapterImpl implements PersistenceAdapter {
const submissions = new ConvexAgentSubmissionStore(this.convex);
const executionStore: AgentExecutionStore = { submissions };
return {
attachmentStore: new ConvexAttachmentStore(this.convex),
conversationStreamStore: new ConvexConversationStreamStore(this.convex),
eventStreamStore: new ConvexEventStreamStore(this.convex),
executionStore,
runStore: new ConvexRunStore(this.convex),
eventStreamStore: new ConvexEventStreamStore(this.convex),
conversationStreamStore: new ConvexConversationStreamStore(this.convex),
attachmentStore: new ConvexAttachmentStore(this.convex),
};
}

4
packages/auth/package.json
View File

@@ -23,13 +23,13 @@
"expo-secure-store": "~57.0.0",
"heroui-native": "catalog:",
"react": "catalog:",
"react-native": "0.86.0",
"react-native": "catalog:",
"sonner": "catalog:",
"zod": "catalog:"
},
"devDependencies": {
"@code/config": "workspace:*",
"@types/react": "~19.2.17",
"@types/react": "catalog:",
"typescript": "catalog:",
"vite": "catalog:",
"vitest": "catalog:"

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

138
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") {
if (
!turn ||
turn.status !== "processing" ||
(turn.attemptNumber ?? 1) !== args.attempt ||
turn.leaseOwner !== args.leaseOwner ||
(turn.leaseExpiresAt !== undefined && turn.leaseExpiresAt < Date.now())
) {
return false;
}
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 null;
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);

29
packages/backend/convex/crons.ts Normal file
View File

@@ -0,0 +1,29 @@
import { cronJobs, makeFunctionReference } from "convex/server";
// Reconciles attempts whose worker lease expired (crash/restart). The target
// is referenced by string so this file does not depend on regenerated codegen.
const reconcileRef = makeFunctionReference<
"mutation",
Record<string, never>,
{ reconciled: number } | null
>("workExecution:reconcileExpiredAttempts");
const reconcileConversationTurnsRef = makeFunctionReference<
"mutation",
Record<string, never>,
{ reconciled: number }
>("conversationMessages:reconcileExpiredTurns");
const crons = cronJobs();
crons.interval(
"reconcile expired work attempts",
{ 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";

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

@@ -1,129 +1,19 @@
import { defineSchema, defineTable } from "convex/server";
import { v } from "convex/values";
export default defineSchema({
organizations: defineTable({
name: v.string(),
kind: v.union(v.literal("personal"), v.literal("team")),
createdBy: v.string(),
createdAt: v.number(),
}).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(),
})
.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(),
updatedAt: v.number(),
})
.index("by_organizationId_and_createdAt", ["organizationId", "createdAt"])
.index("by_organizationId_and_normalizedSourceUrl", [
"organizationId",
"normalizedSourceUrl",
]),
projectContextDocuments: defineTable({
projectId: v.id("projects"),
kind: v.union(
v.literal("readme"),
v.literal("agents"),
v.literal("product"),
v.literal("business"),
v.literal("design"),
v.literal("tech")
),
path: v.string(),
content: v.string(),
origin: v.literal("repository"),
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(),
}).index("by_organizationId", ["organizationId"]),
conversationTurns: defineTable({
conversationId: v.id("conversations"),
clientRequestId: v.string(),
status: v.union(
v.literal("queued"),
v.literal("processing"),
v.literal("completed"),
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",
]),
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(),
createdAt: v.number(),
})
.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(),
}).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(),
})
.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(),
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(),
sourceCreatedAt: v.number(),
})
.index("by_signalId_and_ordinal", ["signalId", "ordinal"])
.index("by_messageId", ["messageId"]),
works: defineTable({
organizationId: v.id("organizations"),
projectId: v.id("projects"),
title: v.string(),
objective: v.string(),
status: v.union(
/* 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"),
@@ -136,29 +26,158 @@ export default defineSchema({
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()),
);
export default defineSchema({
organizations: defineTable({
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({
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({
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"])
.index("by_organizationId_and_normalizedSourceUrl", [
"organizationId",
"normalizedSourceUrl",
]),
projectContextDocuments: defineTable({
content: v.string(),
createdAt: v.number(),
kind: v.union(
v.literal("readme"),
v.literal("agents"),
v.literal("product"),
v.literal("business"),
v.literal("design"),
v.literal("tech")
),
origin: v.literal("repository"),
path: v.string(),
projectId: v.id("projects"),
sourceUrl: v.string(),
updatedAt: v.number(),
}).index("by_projectId_and_path", ["projectId", "path"]),
conversations: defineTable({
createdAt: v.number(),
organizationId: v.id("organizations"),
}).index("by_organizationId", ["organizationId"]),
conversationTurns: defineTable({
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"),
v.literal("completed"),
v.literal("failed")
),
submissionId: v.optional(v.string()),
})
.index("by_conversationId_and_clientRequestId", [
"conversationId",
"clientRequestId",
])
.index("by_status_and_leaseExpiresAt", ["status", "leaseExpiresAt"]),
conversationMessages: defineTable({
content: v.string(),
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({
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({
conversationId: v.id("conversations"),
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({
ordinal: v.number(),
signalId: v.id("signals"),
value: v.string(),
}).index("by_signalId_and_ordinal", ["signalId", "ordinal"]),
signalSources: defineTable({
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(),
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 };
},
});

373
packages/backend/convex/workExecution.test.ts Normal file
View File

@@ -0,0 +1,373 @@
import { convexTest } from "convex-test";
import { anyApi, makeFunctionReference } from "convex/server";
import { describe, expect, test } from "vitest";
import type { Id } from "./_generated/dataModel";
import schema from "./schema";
// Execution-only module set: avoids loading workPlanning (and its env read) so
// these tests run without Convex env vars.
const modules = {
"./_generated/api.ts": () => import("./_generated/api"),
"./_generated/server.ts": () => import("./_generated/server"),
"./authz.ts": () => import("./authz"),
"./workExecution.ts": () => import("./workExecution"),
};
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>;
type Scenario =
| "success"
| "transient-failure-then-success"
| "needs-input"
| "permanent-failure"
| "cancelled";
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", {
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,
designApprovalVersion: 1,
designVersion: 1,
objective: "Prove execution",
organizationId,
projectId,
status: "ready",
title: "Executable work",
updatedAt: 1,
});
await ctx.db.insert("designPackets", {
createdAt: 1,
createdBy: "test",
definitionVersion: 1,
payloadJson: "{}",
status: "current",
version: 1,
workId: id,
});
await ctx.db.insert("workSlices", {
createdAt: 1,
designVersion: 1,
objective: "execute the slice",
observableBehavior: "terminal run",
ordinal: 0,
payloadJson: "{}",
sliceId: "slice-1",
status: "ready",
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 };
};
const runAttempt = async (
t: TestT,
attemptId: Id<"workAttempts">,
scenario: Scenario
) => t.action(api.workExecution.executeFakeAttempt, { attemptId, scenario });
const attemptNumber = async (
t: TestT,
runId: Id<"workRuns">,
number: number
): Promise<Id<"workAttempts">> => {
const attempt = await t.run(async (ctx) =>
ctx.db
.query("workAttempts")
.withIndex("by_runId_and_number", (q) =>
q.eq("runId", runId).eq("number", number)
)
.unique()
);
if (!attempt) {
throw new Error(`attempt ${number} not found`);
}
return attempt._id;
};
const snapshot = async (t: TestT, workId: Id<"works">, runId: Id<"workRuns">) =>
t.run(async (ctx) => ({
attempts: await ctx.db
.query("workAttempts")
.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_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),
}));
describe("simulated execution resolver", () => {
test("success settles Work as completed and marks the slice completed", async () => {
const { t, workId } = await seedReadyWork();
const started = await t
.withIdentity(identity)
.mutation(api.workExecution.startSimulatedExecution, {
scenario: "success",
sliceId: "slice-1",
workId,
});
await runAttempt(t, started.attemptId, "success");
const state = await snapshot(t, workId, started.runId);
expect(state.work?.status).toBe("completed");
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 () => {
const { t, workId } = await seedReadyWork();
const started = await t
.withIdentity(identity)
.mutation(api.workExecution.startSimulatedExecution, {
scenario: "transient-failure-then-success",
workId,
});
await runAttempt(t, started.attemptId, "transient-failure-then-success");
// resolver auto-scheduled attempt 2; drive it manually in the test runtime
const second = await attemptNumber(t, started.runId, 2);
await runAttempt(t, second, "transient-failure-then-success");
const state = await snapshot(t, workId, started.runId);
expect(state.work?.status).toBe("completed");
expect(state.run?.terminalClassification).toBe("Succeeded");
expect(state.attempts).toHaveLength(2);
expect(state.attempts[0]?.classification).toBe("RetryableFailure");
});
test("permanent failure settles Work as failed, not silently runnable", async () => {
const { t, workId } = await seedReadyWork();
const started = await t
.withIdentity(identity)
.mutation(api.workExecution.startSimulatedExecution, {
scenario: "permanent-failure",
workId,
});
await runAttempt(t, started.attemptId, "permanent-failure");
const state = await snapshot(t, workId, started.runId);
expect(state.work?.status).toBe("failed");
expect(state.run?.terminalClassification).toBe("PermanentFailure");
});
test("needs-input surfaces a blocker instead of looping", async () => {
const { t, workId } = await seedReadyWork();
const started = await t
.withIdentity(identity)
.mutation(api.workExecution.startSimulatedExecution, {
scenario: "needs-input",
workId,
});
await runAttempt(t, started.attemptId, "needs-input");
const state = await snapshot(t, workId, started.runId);
expect(state.work?.status).toBe("needs-input");
expect(state.run?.terminalClassification).toBe("NeedsInput");
});
test("cancellation terminates the attempt and returns Work to ready", async () => {
const { t, workId } = await seedReadyWork();
const started = await t
.withIdentity(identity)
.mutation(api.workExecution.startSimulatedExecution, {
scenario: "success",
workId,
});
await t
.withIdentity(identity)
.mutation(api.workExecution.cancelSimulatedExecution, {
runId: started.runId,
});
const state = await snapshot(t, workId, started.runId);
expect(state.work?.status).toBe("ready");
expect(state.run?.status).toBe("cancelled");
expect(state.slice?.status).toBe("ready");
});
test("a slice that is not part of the current approved Design is rejected", async () => {
const { t, workId } = await seedReadyWork();
await expect(
t
.withIdentity(identity)
.mutation(api.workExecution.startSimulatedExecution, {
scenario: "success",
sliceId: "not-a-real-slice",
workId,
})
).rejects.toThrow(/current approved Design/u);
});
test("manual retry restarts a failed Run", async () => {
const { t, workId } = await seedReadyWork();
const started = await t
.withIdentity(identity)
.mutation(api.workExecution.startSimulatedExecution, {
scenario: "permanent-failure",
workId,
});
await runAttempt(t, started.attemptId, "permanent-failure");
expect((await snapshot(t, workId, started.runId)).work?.status).toBe(
"failed"
);
const retried = await t
.withIdentity(identity)
.mutation(api.workExecution.retrySimulatedExecution, {
runId: started.runId,
});
await runAttempt(t, retried.attemptId as Id<"workAttempts">, "success");
const state = await snapshot(t, workId, started.runId);
expect(state.work?.status).toBe("completed");
expect(retried.number).toBe(2);
});
test("reconciling an expired lease resumes within budget", async () => {
const { t, workId } = await seedReadyWork();
const started = await t
.withIdentity(identity)
.mutation(api.workExecution.startSimulatedExecution, {
scenario: "success",
workId,
});
// simulate a worker that claimed but died before finishing
await t.run(async (ctx) => {
await ctx.db.patch(started.attemptId, {
leaseExpiresAt: Date.now() - 1000,
leaseOwner: "dead-worker",
status: "claimed",
});
});
await t.mutation(api.workExecution.reconcileExpiredAttempts, {});
const after = await snapshot(t, workId, started.runId);
expect(after.attempts.find((a) => a.number === 1)?.status).toBe("terminal");
expect(
after.attempts.find((a) => a.number === 2 && a.status === "queued")
).toBeTruthy();
});
});

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

@@ -1,19 +1,19 @@
import {
FakeHarnessLive,
type FakeScenario,
} from "@code/primitives/harness-runtime";
import { defaultCodingKitV0 } 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";
import type { Id } from "./_generated/dataModel";
import type { Doc, Id } from "./_generated/dataModel";
import {
internalAction,
internalMutation,
mutation,
query,
} from "./_generated/server";
import type { MutationCtx } from "./_generated/server";
import { requireProjectMember } from "./authz";
const executeRef = makeFunctionReference<
@@ -37,53 +37,129 @@ const checkpointRef = makeFunctionReference<
},
unknown
>("workExecution:checkpointAttempt");
const completeRef = makeFunctionReference<
const finishRef = makeFunctionReference<
"mutation",
{
attemptId: Id<"workAttempts">;
classification: string;
owner: string;
retryable: boolean;
summary: string;
},
unknown
>("workExecution:completeAttempt");
>("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 } : {}),
});
};
// 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: MutationCtx,
work: Doc<"works">,
sliceId?: string
): Promise<Doc<"workSlices">> => {
const currentDesignVersion = work.designVersion;
if (currentDesignVersion === undefined) {
throw new ConvexError("Work has no approved Design to execute");
}
const slices = await ctx.db
.query("workSlices")
.withIndex("by_workId_and_designVersion", (q) =>
q.eq("workId", work._id).eq("designVersion", currentDesignVersion)
)
.collect();
if (slices.length === 0) {
throw new ConvexError("No slices found for the current approved Design");
}
const slice = sliceId
? 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"
);
}
if (slice.status !== "ready") {
throw new ConvexError("Only the next ready slice can be executed");
}
return slice;
};
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 = 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"),
@@ -92,36 +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: args.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(runId, { status: "running", startedAt: createdAt });
await ctx.db.patch(work._id, {
status: "executing",
updatedAt: createdAt,
});
await ctx.db.patch(slice._id, { status: "running" });
await ctx.db.patch(runId, { startedAt: createdAt, status: "running" });
await appendWorkEvent(
ctx,
work._id,
@@ -129,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 };
},
});
@@ -141,28 +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 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,
@@ -175,33 +296,46 @@ export const cancelSimulatedExecution = mutation({
},
});
// User-initiated restart of a terminal Run. Unlike the automatic resolver
// retry, this is an explicit decision and may exceed the kit budget.
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.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() });
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,
@@ -213,72 +347,86 @@ 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;
},
});
export const completeAttempt = internalMutation({
// Single resolver mutation: record the attempt outcome, then either schedule
// the next attempt within the kit retry policy or settle the Run/Work at the
// terminal classification. Replaces the old completeAttempt that always reset
// Work to "ready" and ignored the retry policy.
export const finishAttempt = internalMutation({
args: {
attemptId: v.id("workAttempts"),
classification: attemptClassification,
owner: v.string(),
classification: v.string(),
retryable: v.boolean(),
summary: v.string(),
},
handler: async (ctx, args) => {
@@ -286,44 +434,155 @@ export const completeAttempt = internalMutation({
if (
!attempt ||
attempt.leaseOwner !== args.owner ||
attempt.status === "terminal"
)
return false;
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,
leaseExpiresAt: undefined,
});
const run = await ctx.db.get(attempt.runId);
if (!run) return false;
await ctx.db.patch(run._id, {
status: "terminal",
endedAt,
leaseExpiresAt: undefined,
status: "terminal",
summary: args.summary,
});
const outcome = {
classification: args.classification,
retryable: args.retryable,
summary: args.summary,
};
const resolution = resolveOutcome(
outcome,
attempt.number,
defaultCodingKitV0.retryPolicy
);
const run = await ctx.db.get(attempt.runId);
if (!run) {
return null;
}
await appendWorkEvent(
ctx,
attempt.workId,
"attempt.completed",
`attempt-completed:${attempt._id}`,
String(attempt._id),
JSON.stringify({
classification: args.classification,
retried: resolution.kind === "retry",
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", {
number: attempt.number + 1,
runId: run._id,
status: "queued",
workId: attempt.workId,
});
await ctx.scheduler.runAfter(0, executeRef, {
attemptId: nextAttemptId,
scenario: run.scenario,
});
return { nextAttemptId, retried: true };
}
await ctx.db.patch(run._id, {
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: args.classification === "NeedsInput" ? "needs-input" : "ready",
status: resultingWorkStatus,
updatedAt: endedAt,
});
if (work)
await ctx.db.insert("workEvents", {
workId: work._id,
kind: "attempt.completed",
idempotencyKey: `attempt-completed:${attempt._id}`,
referenceId: String(attempt._id),
payloadJson: JSON.stringify({
}
await ctx.db.insert("resolverDecisions", {
attemptId: attempt._id,
attemptNumber: attempt.number,
classification: args.classification,
summary: args.summary,
}),
createdAt: endedAt,
decision: "terminal",
resultingWorkStatus,
runId: run._id,
summary: args.summary,
workId: attempt.workId,
});
return true;
await appendWorkEvent(
ctx,
attempt.workId,
"run.completed",
`run-completed:${run._id}`,
String(run._id),
JSON.stringify({
classification: args.classification,
workStatus: resultingWorkStatus,
})
);
return { retried: false };
},
});
@@ -342,39 +601,61 @@ 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,
owner,
sequence: item.sequence,
kind: item.kind,
message: item.message,
metadataJson: JSON.stringify(item.metadata),
});
await ctx.runMutation(completeRef, {
attemptId: args.attemptId,
owner,
sequence: item.sequence,
});
}
await ctx.runMutation(finishRef, {
attemptId: args.attemptId,
classification: outcome.classification,
owner,
retryable: outcome.retryable,
summary: outcome.summary,
});
return null;
},
});
// Bounded recovery: a claimed/running attempt whose lease expired means the
// worker died mid-flight. Terminate that attempt, then either resume within
// the kit budget or settle the Run/Work so nothing stays "running" forever.
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 (
@@ -383,17 +664,85 @@ export const reconcileExpiredAttempts = internalMutation({
attempt.leaseExpiresAt < now()
) {
await ctx.db.patch(attempt._id, {
status: "queued",
leaseOwner: undefined,
classification: "Blocked",
endedAt: now(),
leaseExpiresAt: undefined,
leaseOwner: undefined,
status: "terminal",
summary: "Attempt lease expired and was reconciled",
});
const run = await ctx.db.get(attempt.runId);
if (run)
await ctx.scheduler.runAfter(0, executeRef, {
await appendWorkEvent(
ctx,
attempt.workId,
"attempt.reconciled",
`attempt-reconciled:${attempt._id}`,
String(attempt._id),
JSON.stringify({ attemptNumber: attempt.number })
);
reconciled += 1;
if (!run) {
continue;
}
const attempts = await ctx.db
.query("workAttempts")
.withIndex("by_runId_and_number", (q) => q.eq("runId", run._id))
.collect();
const withinBudget =
attempts.length < defaultCodingKitV0.retryPolicy.maxAttempts;
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", {
number: attempts.length + 1,
runId: run._id,
status: "queued",
workId: attempt.workId,
});
await ctx.scheduler.runAfter(0, executeRef, {
attemptId: nextAttemptId,
scenario: run.scenario,
});
reconciled += 1;
} else {
await ctx.db.patch(run._id, {
endedAt: now(),
status: "terminal",
terminalClassification: "Blocked",
terminalSummary: "Run reconciled after expired lease",
});
const work = await ctx.db.get(run.workId);
if (work) {
await ctx.db.patch(work._id, {
status: "blocked",
updatedAt: now(),
});
}
const slice = await getRunSlice(ctx, run);
if (slice) {
await ctx.db.patch(slice._id, { status: "blocked" });
}
}
}
}
return { reconciled };
@@ -404,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")

321
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,15 +137,278 @@ 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">),
work: await ctx.db.get(fixture.workId),
}));
expect(completed.work?.status).toBe("ready");
expect(completed.work?.status).toBe("completed");
expect(completed.run?.terminalClassification).toBe("Succeeded");
expect(completed.attempt?.status).toBe("terminal");
expect(completed.events.length).toBeGreaterThan(0);
});
});
const validDefinitionPayload = () => ({
acceptanceCriteria: ["terminal outcome"],
affectedUsers: ["user"],
assumptions: [],
constraints: [],
desiredOutcome: "A terminal simulation",
inScope: ["fake"],
outOfScope: [],
problem: "Need a proof",
questions: [],
requiredArtifacts: ["events"],
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) => {
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,
designApprovalVersion: 1,
designVersion: 1,
objective: "Prove the guard",
organizationId,
projectId,
status: "executing",
title: "Guarded work",
updatedAt: 1,
});
return { workId: id };
});
await expect(
t
.withIdentity(identity)
.mutation(api.workPlanning.saveDefinitionProposal, {
payloadJson: JSON.stringify(validDefinitionPayload()),
workId,
})
).rejects.toThrow(/while a Run is executing/u);
});
test("planner questions are persisted as durable Work questions", 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",
});
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,
definitionVersion: 1,
objective: "Prove questions",
organizationId,
projectId,
status: "awaiting-definition-approval",
title: "Questionable work",
updatedAt: 1,
});
return { workId: id };
});
await t.mutation(api.workPlanning.submitQuestion, {
questionJson: JSON.stringify({
alternatives: [],
id: "q1",
impact: "medium",
prompt: "Which runtime?",
status: "open",
}),
token: "test-token",
workId,
});
const questions = await t.run(async (ctx) =>
ctx.db.query("workQuestions").collect()
);
expect(questions).toHaveLength(1);
expect(questions[0]?.questionId).toBe("q1");
expect(questions[0]?.definitionVersion).toBe(1);
});
});

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

@@ -1,20 +1,25 @@
import type { WorkEventKind } from "@code/primitives/work";
import {
decodeDefinition,
validateDefinition,
type WorkDefinition,
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";
import { Effect } from "effect";
import { Effect, Schema } from "effect";
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<
@@ -26,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 {
@@ -37,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({
@@ -105,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(
@@ -124,13 +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") {
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,
})
@@ -139,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(),
});
@@ -200,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);
@@ -211,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) => {
@@ -237,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(
@@ -267,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,
@@ -309,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);
@@ -332,33 +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.definitionApprovalVersion !== work.definitionVersion)
if (work.status === "executing") {
throw new ConvexError("Cannot revise Work while a Run is executing");
}
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(
@@ -367,42 +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_workId_and_kind", (q) =>
q.eq("workId", work._id).eq("kind", "design")
)
.collect();
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(),
});
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())
await ctx.db.insert("workSlices", {
version: decoded.version,
workId: work._id,
});
for (const [ordinal, slice] of decoded.slices.entries()) {
await ctx.db.insert("workSlices", {
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(),
});
@@ -418,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);
@@ -430,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);
@@ -441,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(
@@ -472,31 +558,116 @@ 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");
return { accepted: Boolean(JSON.parse(args.questionJson)) };
}
const work = await requireWork(ctx, args.workId);
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)(parsePayload(args.questionJson))
).catch((error: unknown) => {
throw new ConvexError(
error instanceof Error ? error.message : "Invalid question"
);
});
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),
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;
},
});
@@ -505,28 +676,50 @@ export const runPlanner = internalAction({
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",
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}`,
},
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.`,
}),
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;
@@ -572,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
@@ -597,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) => ({
@@ -617,6 +812,8 @@ export const listForProject = query({
rawText: source.rawTextSnapshot,
submissionId: null,
})),
summary: signal.summary,
title: signal.title,
};
})
);
@@ -628,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:"

2
packages/env/package.json vendored
View File

@@ -19,7 +19,7 @@
},
"devDependencies": {
"@code/config": "workspace:*",
"@types/node": "^22.13.14",
"@types/node": "catalog:",
"typescript": "catalog:"
}
}

3
packages/primitives/package.json
View File

@@ -16,6 +16,7 @@
"./signal": "./src/signal.ts",
"./smoke": "./src/smoke.ts",
"./work": "./src/work.ts",
"./work-artifact": "./src/work-artifact.ts",
"./work-definition": "./src/work-definition.ts",
"./work-design": "./src/work-design.ts",
"./work-lifecycle": "./src/work-lifecycle.ts",
@@ -35,7 +36,7 @@
},
"devDependencies": {
"@code/config": "workspace:*",
"@types/node": "^22.13.14",
"@types/node": "catalog:",
"typescript": "catalog:",
"vitest": "catalog:"
}

13
packages/primitives/package.json.new
View File

@@ -1,13 +0,0 @@
{
"name": "@code/primitives",
"exports": {
".": "./src/index.ts",
"./agent-os": "./src/agent-os.ts",
"./git": "./src/git.ts",
"./project": "./src/project.ts",
"./project-issue": "./src/project-issue.ts",
"./project-workspace": "./src/project-workspace.ts",
"./project-work": "./src/project-work.ts",
"./signal": "./src/signal.ts",
"./smoke": "./src/smoke.ts"
},

1
packages/primitives/src/index.ts
View File

@@ -10,6 +10,7 @@ export * from "./project-work";
export * from "./signal";
export * from "./smoke";
export * from "./work";
export * from "./work-artifact";
export * from "./work-definition";
export * from "./work-design";
export * from "./work-lifecycle";

45
packages/primitives/src/resolver.ts
View File

@@ -1,12 +1,19 @@
import { Schema } from "effect";
import type { WorkStatus } from "./work-lifecycle";
const Text = Schema.String.check(
Schema.makeFilter((value) => value.trim().length > 0, {
expected: "a non-empty string",
})
);
export const RunStatus = Schema.Literals(["ready", "running", "terminal"]);
export const RunStatus = Schema.Literals([
"ready",
"running",
"terminal",
"cancelled",
]);
export type RunStatus = typeof RunStatus.Type;
export const AttemptStatus = Schema.Literals([
"queued",
@@ -48,9 +55,6 @@ export const CodingKitV0 = Schema.Struct({
});
export type CodingKitV0 = typeof CodingKitV0.Type;
export const isTerminalClassification = (_: AttemptClassification): true =>
true;
export const shouldRetry = (
outcome: AttemptOutcome,
attemptNumber: number,
@@ -60,6 +64,39 @@ export const shouldRetry = (
attemptNumber < policy.maxAttempts &&
policy.retryableClassifications.includes(outcome.classification);
// Terminal Work status the resolver settles on once it will run no further
// attempt. RetryableFailure lands here only after the retry budget is gone.
export const WORK_STATUS_FOR_OUTCOME: Readonly<
Record<AttemptClassification, WorkStatus>
> = {
Blocked: "blocked",
BudgetExhausted: "failed",
Cancelled: "ready",
NeedsInput: "needs-input",
PermanentFailure: "failed",
RetryableFailure: "failed",
Succeeded: "completed",
VerificationFailed: "blocked",
};
export type Resolution =
| { readonly kind: "retry" }
| { readonly kind: "terminal"; readonly workStatus: WorkStatus };
// Resolver decision after an attempt finishes: retry within policy, or settle
// Work at the terminal status mapped from the attempt classification.
export const resolveOutcome = (
outcome: AttemptOutcome,
attemptNumber: number,
policy: RetryPolicy
): Resolution =>
shouldRetry(outcome, attemptNumber, policy)
? { kind: "retry" }
: {
kind: "terminal",
workStatus: WORK_STATUS_FOR_OUTCOME[outcome.classification],
};
export const defaultCodingKitV0: CodingKitV0 = {
harness: "fake",
id: "coding-kit-v0",

58
packages/primitives/src/work-artifact.test.ts Normal file
View File

@@ -0,0 +1,58 @@
import { Effect } from "effect";
import { describe, expect, test } from "vitest";
import {
decodeWorkArtifactDraft,
decodeWorkDeliveryDraft,
} from "./work-artifact";
describe("Work artifact contracts", () => {
test("accepts traceable artifact and delivery metadata", async () => {
await expect(
Effect.runPromise(
decodeWorkArtifactDraft({
idempotencyKey: "attempt:1:test-report",
kind: "test-report",
metadataJson: "{}",
producer: "fake-harness",
provenanceJson: "{}",
sourceRevision: "abc123",
title: "Focused test report",
verificationStatus: "verified",
})
)
).resolves.toMatchObject({ kind: "test-report" });
await expect(
Effect.runPromise(
decodeWorkDeliveryDraft({
externalId: "42",
idempotencyKey: "delivery:pr:42",
kind: "pull-request",
metadataJson: "{}",
provider: "gitea",
sourceRevision: "abc123",
status: "ready",
target: "main",
url: "https://git.example/pulls/42",
})
)
).resolves.toMatchObject({ kind: "pull-request" });
});
test("rejects unverifiable evidence metadata", async () => {
await expect(
Effect.runPromise(
decodeWorkArtifactDraft({
idempotencyKey: "attempt:1:test-report",
kind: "test-report",
metadataJson: "not-json",
producer: "fake-harness",
provenanceJson: "{}",
title: "Focused test report",
verificationStatus: "verified",
})
)
).rejects.toThrow();
});
});

141
packages/primitives/src/work-artifact.ts Normal file
View File

@@ -0,0 +1,141 @@
import { Effect, Schema } from "effect";
const Text = Schema.String.check(
Schema.makeFilter((value) => value.trim().length > 0, {
expected: "a non-empty string",
})
);
const JsonText = Schema.String.check(
Schema.makeFilter(
(value) => {
try {
JSON.parse(value);
return true;
} catch {
return false;
}
},
{ expected: "valid JSON text" }
)
);
export const WorkArtifactKind = Schema.Literals([
"definition",
"design",
"diff",
"test-report",
"verification-report",
"runtime-log",
"screenshot",
"video",
"commit",
"branch",
"pull-request",
"preview",
"deployment",
"other",
]);
export type WorkArtifactKind = typeof WorkArtifactKind.Type;
export const ArtifactVerificationStatus = Schema.Literals([
"unverified",
"verified",
"rejected",
]);
export type ArtifactVerificationStatus = typeof ArtifactVerificationStatus.Type;
export const WorkArtifactDraft = Schema.Struct({
contentHash: Schema.optional(Text),
environmentId: Schema.optional(Text),
idempotencyKey: Text,
kind: WorkArtifactKind,
metadataJson: JsonText,
producer: Text,
provenanceJson: JsonText,
sourceRevision: Schema.optional(Text),
title: Text,
uri: Schema.optional(Text),
verificationStatus: ArtifactVerificationStatus,
});
export type WorkArtifactDraft = typeof WorkArtifactDraft.Type;
export const WorkDeliveryKind = Schema.Literals([
"branch",
"commit",
"pull-request",
"preview",
"deployment",
]);
export type WorkDeliveryKind = typeof WorkDeliveryKind.Type;
export const WorkDeliveryStatus = Schema.Literals([
"recorded",
"ready",
"approved",
"delivered",
"failed",
"cancelled",
]);
export type WorkDeliveryStatus = typeof WorkDeliveryStatus.Type;
export const WorkDeliveryDraft = Schema.Struct({
externalId: Text,
idempotencyKey: Text,
kind: WorkDeliveryKind,
metadataJson: JsonText,
provider: Text,
sourceRevision: Text,
status: WorkDeliveryStatus,
target: Text,
url: Schema.optional(Text),
});
export type WorkDeliveryDraft = typeof WorkDeliveryDraft.Type;
export class WorkArtifactError extends Schema.TaggedErrorClass<WorkArtifactError>()(
"WorkArtifactError",
{ message: Schema.String }
) {}
const decode = <A, I, R>(schema: Schema.Codec<A, I, R>, input: unknown) =>
Schema.decodeUnknownEffect(schema)(input).pipe(
Effect.mapError(
(cause) => new WorkArtifactError({ message: cause.message })
)
);
export const decodeWorkArtifactDraft = Effect.fn("Work.decodeArtifactDraft")(
function* decodeArtifactDraft(input: unknown) {
const draft = yield* decode(WorkArtifactDraft, input);
if (
draft.verificationStatus === "verified" &&
draft.sourceRevision === undefined
) {
return yield* Effect.fail(
new WorkArtifactError({
message: "Verified artifacts must bind an exact source revision",
})
);
}
return draft;
}
);
export const decodeWorkDeliveryDraft = Effect.fn("Work.decodeDeliveryDraft")(
(input: unknown) => decode(WorkDeliveryDraft, input)
);
const DELIVERY_TRANSITIONS: Readonly<
Record<WorkDeliveryStatus, readonly WorkDeliveryStatus[]>
> = {
approved: ["delivered", "failed", "cancelled"],
cancelled: [],
delivered: [],
failed: [],
ready: ["approved", "delivered", "failed", "cancelled"],
recorded: ["ready", "failed", "cancelled"],
};
export const canTransitionWorkDelivery = (
from: WorkDeliveryStatus,
to: WorkDeliveryStatus
): boolean => DELIVERY_TRANSITIONS[from].includes(to);

23
packages/primitives/src/work-definition.ts
View File

@@ -59,8 +59,8 @@ export class DefinitionError extends Schema.TaggedErrorClass<DefinitionError>()(
}
) {}
export const validateDefinition = Effect.fn("Work.validateDefinition")(
function* validateDefinition(input: unknown) {
export const decodeDefinition = Effect.fn("Work.decodeDefinition")(
function* decodeDefinition(input: unknown) {
const definition = yield* Schema.decodeUnknownEffect(WorkDefinition)(
input
).pipe(
@@ -69,6 +69,25 @@ export const validateDefinition = Effect.fn("Work.validateDefinition")(
new DefinitionError({ message: cause.message, reason: "Invalid" })
)
);
const questionIds = new Set<string>();
for (const question of definition.questions) {
if (questionIds.has(question.id)) {
return yield* Effect.fail(
new DefinitionError({
message: `Question ID must be unique: ${question.id}`,
reason: "Invalid",
})
);
}
questionIds.add(question.id);
}
return definition;
}
);
export const validateDefinition = Effect.fn("Work.validateDefinition")(
function* validateDefinition(input: unknown) {
const definition = yield* decodeDefinition(input);
const blocked = definition.questions.some(
(question) => question.status === "open" && question.impact === "high"
);

21
packages/primitives/src/work-design.ts
View File

@@ -59,6 +59,7 @@ export class DesignError extends Schema.TaggedErrorClass<DesignError>()(
"Invalid",
"NoObservableSlices",
"DefinitionMismatch",
"InvalidSliceGraph",
]),
}
) {}
@@ -73,8 +74,10 @@ export const validateDesignPacket = Effect.fn("Work.validateDesignPacket")(
);
if (
design.slices.length === 0 ||
design.slices.length > 4 ||
design.slices.some(
(slice) =>
slice.codeBoundaries.length === 0 ||
slice.observableBehavior.trim().length === 0 ||
slice.evidenceRequirements.length === 0 ||
slice.verification.length === 0
@@ -88,6 +91,24 @@ export const validateDesignPacket = Effect.fn("Work.validateDesignPacket")(
})
);
}
const seen = new Set<string>();
for (const slice of design.slices) {
if (
seen.has(slice.id) ||
slice.dependsOn.some(
(dependency) => dependency === slice.id || !seen.has(dependency)
)
) {
return yield* Effect.fail(
new DesignError({
message:
"Slice IDs must be unique and dependencies must reference earlier slices",
reason: "InvalidSliceGraph",
})
);
}
seen.add(slice.id);
}
return design;
}
);

10
packages/primitives/src/work-lifecycle.ts
View File

@@ -20,14 +20,10 @@ export const WORK_TRANSITIONS: Readonly<
Record<WorkStatus, readonly WorkStatus[]>
> = {
"awaiting-definition-approval": ["designing", "defining"],
"awaiting-design-approval": [
"ready",
"designing",
"awaiting-definition-approval",
],
"awaiting-design-approval": ["ready", "designing"],
blocked: ["ready", "executing", "cancelled"],
cancelled: ["ready", "proposed"],
completed: ["proposed", "defining"],
cancelled: ["ready"],
completed: [],
defining: ["awaiting-definition-approval", "proposed"],
designing: ["awaiting-design-approval", "awaiting-definition-approval"],
executing: [

73
packages/primitives/src/work-resolution.test.ts
View File

@@ -2,8 +2,8 @@ import { Effect } from "effect";
import { describe, expect, test } from "vitest";
import { FakeHarnessLive } from "./harness-runtime";
import { defaultCodingKitV0, shouldRetry } from "./resolver";
import { validateDefinition } from "./work-definition";
import { defaultCodingKitV0, resolveOutcome, shouldRetry } from "./resolver";
import { decodeDefinition, validateDefinition } from "./work-definition";
import { validateDesignPacket } from "./work-design";
import { canTransitionWork } from "./work-lifecycle";
@@ -30,13 +30,32 @@ const definition = {
version: 1,
};
const probeOutcome = (
classification: Parameters<typeof resolveOutcome>[0]["classification"],
retryable = false
) => ({ classification, retryable, summary: "probe" });
describe("work resolution contracts", () => {
test("high-impact open questions block definition approval", async () => {
await expect(
Effect.runPromise(decodeDefinition(definition))
).resolves.toEqual(definition);
await expect(
Effect.runPromise(validateDefinition(definition))
).rejects.toThrow(/High-impact/u);
});
test("question identities are unique within a Definition version", async () => {
await expect(
Effect.runPromise(
decodeDefinition({
...definition,
questions: [definition.questions[0], definition.questions[0]],
})
)
).rejects.toThrow(/Question ID must be unique/u);
});
test("observable slices and terminal fake outcomes are enforced", async () => {
const approved = await Effect.runPromise(
validateDefinition({ ...definition, questions: [] })
@@ -103,3 +122,53 @@ describe("work resolution contracts", () => {
expect(canTransitionWork("ready", "executing")).toBe(true);
});
});
describe("resolver decisions", () => {
const policy = defaultCodingKitV0.retryPolicy;
test("retries a retryable failure while the budget remains", () => {
expect(
resolveOutcome(probeOutcome("RetryableFailure", true), 1, policy)
).toEqual({ kind: "retry" });
});
test("settles RetryableFailure as failed once the budget is exhausted", () => {
expect(
resolveOutcome(probeOutcome("RetryableFailure", true), 3, policy)
).toEqual({
kind: "terminal",
workStatus: "failed",
});
});
test("maps each terminal classification to its Work status", () => {
expect(resolveOutcome(probeOutcome("Succeeded"), 1, policy)).toEqual({
kind: "terminal",
workStatus: "completed",
});
expect(resolveOutcome(probeOutcome("PermanentFailure"), 1, policy)).toEqual(
{
kind: "terminal",
workStatus: "failed",
}
);
expect(resolveOutcome(probeOutcome("NeedsInput"), 1, policy)).toEqual({
kind: "terminal",
workStatus: "needs-input",
});
expect(resolveOutcome(probeOutcome("Blocked"), 1, policy)).toEqual({
kind: "terminal",
workStatus: "blocked",
});
expect(resolveOutcome(probeOutcome("Cancelled"), 1, policy)).toEqual({
kind: "terminal",
workStatus: "ready",
});
});
test("completed and cancelled are not silently reopened", () => {
expect(canTransitionWork("completed", "defining")).toBe(false);
expect(canTransitionWork("cancelled", "proposed")).toBe(false);
expect(canTransitionWork("ready", "executing")).toBe(true);
});
});

10
packages/primitives/src/work.ts
View File

@@ -96,6 +96,7 @@ export const WorkEventKind = Schema.Literals([
"definition.revised",
"definition.approved",
"definition.invalidated",
"question.created",
"question.answered",
"question.withdrawn",
"design.requested",
@@ -103,12 +104,21 @@ export const WorkEventKind = Schema.Literals([
"design.revised",
"design.approved",
"design.invalidated",
"planner.failed",
"slice.started",
"slice.completed",
"slice.ready",
"run.started",
"run.completed",
"run.cancelled",
"attempt.claimed",
"attempt.event",
"attempt.completed",
"attempt.reconciled",
"resolver.decided",
"artifact.recorded",
"delivery.recorded",
"delivery.updated",
]);
export type WorkEventKind = typeof WorkEventKind.Type;

6
packages/ui/package.json
View File

@@ -30,15 +30,15 @@
"shadcn": "^4.12.0",
"shiki": "^4.3.1",
"sonner": "catalog:",
"streamdown": "^2.5.0",
"streamdown": "catalog:",
"tailwind-merge": "catalog:",
"tw-animate-css": "^1.4.0",
"use-stick-to-bottom": "^1.1.6"
},
"devDependencies": {
"@code/config": "workspace:*",
"@tailwindcss/postcss": "^4.3.2",
"@types/react": "^19.2.17",
"@tailwindcss/postcss": "catalog:",
"@types/react": "catalog:",
"@types/react-dom": "catalog:",
"tailwindcss": "catalog:",
"typescript": "catalog:"

2
vite.config.ts
View File

@@ -7,6 +7,7 @@ export default defineConfig({
"**/node_modules/**",
".agents/skills/**",
"repos/**",
"scripts/**",
"apps/daemon/**",
"apps/desktop/**",
"apps/native/**",
@@ -28,6 +29,7 @@ export default defineConfig({
"**/node_modules/**",
".agents/skills/**",
"repos/**",
"scripts/**",
"apps/daemon/**",
"apps/desktop/**",
"apps/native/**",