1 Commits

Author SHA1 Message Date
-Puter
85633a5124 Slice 4 hardening: bounded resolver, terminal states, reconciliation
Addresses review findings before Slice 5 sandbox work.

Resolver primitives:
- resolveOutcome maps an attempt outcome to retry-or-terminal; removed the
  no-op isTerminalClassification.
- WORK_STATUS_FOR_OUTCOME preserves real terminal Work status instead of
  collapsing everything back to ready.
- Lifecycle: completed is terminal, cancelled reopens only to ready, and the
  invalid awaiting-design-approval -> awaiting-definition-approval reopening
  is gone.

Execution (workExecution.ts):
- finishAttempt replaces completeAttempt: auto-retries within the kit budget
  and settles Work at the real classification (Succeeded->completed,
  PermanentFailure/BudgetExhausted->failed, NeedsInput->needs-input,
  Blocked->blocked). No Work stays silently runnable after a failure.
- startSimulatedExecution validates sliceId against the current approved
  Design (foreign ids rejected; defaults to the first slice) and advances
  slice status running -> completed/ready.
- retrySimulatedExecution is now an explicit user restart with a retryable-
  state guard, not the automatic resolver path.
- reconcileExpiredAttempts is bounded: a dead lease terminates as Blocked,
  resumes within budget or settles the Run/Work so nothing runs forever.
- crons.ts runs the reconciler every 30s, string-referenced so it does not
  depend on the stale codegen.

Planning (workPlanning.ts):
- saveDefinition/saveDesign reject revision while a Run is executing.
- submitQuestion validates and persists a durable workQuestions row instead
  of discarding planner questions.
- Design revision invalidates active design approvals.

Verification: primitives 69 tests (+4), backend 22 tests (+10) covering
retry/cancel/failure/restart, slice validation, reconciliation, the
revision guard, and question persistence. Root typecheck, web+agents build,
and the configured Ultracite gate (53 files) all pass.

package.json: the recurring check gate now covers workExecution, workPlanning,
crons, resolver, and work-lifecycle.
2026-07-28 00:07:35 +05:30
64 changed files with 1524 additions and 5041 deletions

View File

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

View File

@@ -17,7 +17,6 @@ import {
RotateCcw, RotateCcw,
Send, Send,
Sparkles, Sparkles,
Settings,
X, X,
} from "lucide-react"; } from "lucide-react";
import { useMemo, useRef, useState } from "react"; import { useMemo, useRef, useState } from "react";
@@ -259,61 +258,19 @@ const WorkCard = ({ onSourceSelect, work, slice }: WorkCardProps) => {
Build Build
</p> </p>
{latestRun ? ( {latestRun ? (
<div className="mt-1 space-y-2 text-[#626057]"> <p className="mt-1 leading-5 text-[#626057]">
<p className="leading-5"> Run {latestRun.status}:{" "}
{latestRun.executionKind === "real" {latestRun.terminalSummary ??
? "AgentOS" latestRun.terminalClassification ??
: "Simulation"}{" "} "activity is still arriving"}
run {latestRun.status}:{" "} </p>
{latestRun.terminalSummary ??
latestRun.terminalClassification ??
"activity is still arriving"}
</p>
{latestRun.attemptEvents?.slice(-5).map((item) => (
<p
className="border-l-2 border-[#b8c760] pl-2"
key={item._id}
>
{item.message}
</p>
))}
{latestRun.baseRevision ? (
<p className="font-mono text-[10px] text-[#747168]">
{latestRun.baseRevision.slice(0, 8)} {" "}
{latestRun.candidateRevision?.slice(0, 8) ?? "working"}
</p>
) : null}
{latestRun.artifacts?.map((artifact) => (
<a
className="block font-medium underline"
href={artifact.uri}
key={artifact._id}
rel="noreferrer"
target="_blank"
>
{artifact.title}
</a>
))}
</div>
) : ( ) : (
<p className="mt-1 text-[#747168]">No implementation Run yet.</p> <p className="mt-1 text-[#747168]">No simulation Run yet.</p>
)} )}
<div className="mt-2 flex flex-wrap gap-2"> <div className="mt-2 flex flex-wrap gap-2">
{work.status === "ready" ? ( {work.status === "ready" ? (
<Button <Button
size="sm" size="sm"
disabled={!slice.projectGitConnection}
onClick={() =>
void slice.startExecution(work._id, design?.slices?.[0]?.id)
}
>
<Play className="size-3.5" /> Run
</Button>
) : null}
{work.status === "ready" ? (
<Button
size="sm"
variant="outline"
onClick={() => onClick={() =>
void slice.startSimulation( void slice.startSimulation(
work._id, work._id,
@@ -329,11 +286,7 @@ const WorkCard = ({ onSourceSelect, work, slice }: WorkCardProps) => {
<Button <Button
size="sm" size="sm"
variant="outline" variant="outline"
onClick={() => onClick={() => void slice.cancelSimulation(latestRun._id)}
void (latestRun.executionKind === "real"
? slice.cancelExecution(latestRun._id)
: slice.cancelSimulation(latestRun._id))
}
> >
<X className="size-3.5" /> Cancel <X className="size-3.5" /> Cancel
</Button> </Button>
@@ -435,10 +388,6 @@ export const SliceOnePage = () => {
const attachments = useChatImages(); const attachments = useChatImages();
const imageInput = useRef<HTMLInputElement>(null); const imageInput = useRef<HTMLInputElement>(null);
const [drawerOpen, setDrawerOpen] = useState(false); const [drawerOpen, setDrawerOpen] = useState(false);
const [settingsOpen, setSettingsOpen] = useState(false);
const [giteaUrl, setGiteaUrl] = useState("https://git.openputer.com");
const [giteaUsername, setGiteaUsername] = useState("");
const [giteaToken, setGiteaToken] = useState("");
const [highlightedMessageId, setHighlightedMessageId] = useState<string>(); const [highlightedMessageId, setHighlightedMessageId] = useState<string>();
const highlightTimer = useRef<ReturnType<typeof setTimeout> | null>(null); const highlightTimer = useRef<ReturnType<typeof setTimeout> | null>(null);
const works = slice.works ?? EMPTY_WORKS; const works = slice.works ?? EMPTY_WORKS;
@@ -501,7 +450,7 @@ export const SliceOnePage = () => {
style={viewportStyle} style={viewportStyle}
> >
<section className="flex min-w-0 flex-1 flex-col"> <section className="flex min-w-0 flex-1 flex-col">
<header className="relative flex h-14 shrink-0 items-center border-b border-[#d7d3c7] bg-[#faf9f4] px-4"> <header className="flex h-14 shrink-0 items-center border-b border-[#d7d3c7] bg-[#faf9f4] px-4">
<div className="min-w-0 flex-1 pr-2"> <div className="min-w-0 flex-1 pr-2">
<select <select
aria-label="Current project" aria-label="Current project"
@@ -519,15 +468,6 @@ export const SliceOnePage = () => {
Conversation to proposed Work Conversation to proposed Work
</p> </p>
</div> </div>
<Button
aria-label="Project settings"
className="mr-2 size-9"
onClick={() => setSettingsOpen((open) => !open)}
size="icon"
variant="outline"
>
<Settings className="size-4" />
</Button>
<button <button
className="flex h-9 items-center gap-2 border border-[#c9c5b9] bg-white px-3 text-xs lg:hidden" className="flex h-9 items-center gap-2 border border-[#c9c5b9] bg-white px-3 text-xs lg:hidden"
onClick={() => setDrawerOpen(true)} onClick={() => setDrawerOpen(true)}
@@ -536,78 +476,6 @@ export const SliceOnePage = () => {
<Menu className="size-4" /> Work{" "} <Menu className="size-4" /> Work{" "}
{slice.works === undefined ? "…" : works.length} {slice.works === undefined ? "…" : works.length}
</button> </button>
{settingsOpen ? (
<section className="absolute right-4 top-12 z-40 w-[min(92vw,360px)] border border-[#c9c5b9] bg-[#fffefa] p-4 shadow-xl">
<div className="flex items-center justify-between">
<h2 className="text-sm font-semibold">Project Git</h2>
<button
aria-label="Close settings"
onClick={() => setSettingsOpen(false)}
type="button"
>
<X className="size-4" />
</button>
</div>
<p className="mt-1 text-xs text-[#747168]">
{slice.projectGitConnection
? `${slice.projectGitConnection.provider} · ${slice.projectGitConnection.serverUrl}`
: "No Git credentials attached"}
</p>
<div className="mt-4 flex gap-2">
<Button
size="sm"
variant="outline"
onClick={() => void slice.authorizeGithub()}
>
Authorize GitHub
</Button>
<Button
size="sm"
onClick={() => void slice.connectLinkedGithub()}
>
Use GitHub
</Button>
</div>
<div className="mt-4 space-y-2 border-t border-[#e7e3d9] pt-4">
<input
aria-label="Gitea server URL"
className="h-9 w-full border px-2 text-xs"
onChange={(event) => setGiteaUrl(event.target.value)}
value={giteaUrl}
/>
<input
aria-label="Gitea username"
className="h-9 w-full border px-2 text-xs"
onChange={(event) => setGiteaUsername(event.target.value)}
placeholder="Username (optional)"
value={giteaUsername}
/>
<input
aria-label="Gitea access token"
className="h-9 w-full border px-2 text-xs"
onChange={(event) => setGiteaToken(event.target.value)}
placeholder="Personal access token"
type="password"
value={giteaToken}
/>
<Button
className="w-full"
disabled={!giteaToken.trim()}
size="sm"
onClick={() => {
void slice.connectGitea({
serverUrl: giteaUrl,
token: giteaToken,
username: giteaUsername || undefined,
});
setGiteaToken("");
}}
>
Connect Gitea
</Button>
</div>
</section>
) : null}
</header> </header>
<Conversation className="min-h-0 flex-1"> <Conversation className="min-h-0 flex-1">
<ConversationContent className="mx-auto min-h-full w-full max-w-2xl gap-4 px-4 py-5 sm:px-6"> <ConversationContent className="mx-auto min-h-full w-full max-w-2xl gap-4 px-4 py-5 sm:px-6">

View File

@@ -1,4 +1,3 @@
import { authClient } from "@code/auth/web";
import { api } from "@code/backend/convex/_generated/api"; import { api } from "@code/backend/convex/_generated/api";
import type { Id } from "@code/backend/convex/_generated/dataModel"; import type { Id } from "@code/backend/convex/_generated/dataModel";
import { useAction, useMutation, useQuery } from "convex/react"; import { useAction, useMutation, useQuery } from "convex/react";
@@ -28,23 +27,6 @@ interface WorkRecord {
readonly designs: readonly unknown[]; readonly designs: readonly unknown[];
readonly slices: readonly unknown[]; readonly slices: readonly unknown[];
readonly runs: readonly { readonly runs: readonly {
readonly artifacts?: readonly {
_id: string;
kind: string;
metadataJson: string;
sourceRevision?: string;
title: string;
uri?: string;
}[];
readonly attemptEvents?: readonly {
_id: string;
kind: string;
message: string;
occurredAt: number;
}[];
readonly baseRevision?: string;
readonly candidateRevision?: string;
readonly executionKind?: string;
_id: Id<"workRuns">; _id: Id<"workRuns">;
status: string; status: string;
terminalClassification?: string; terminalClassification?: string;
@@ -114,57 +96,6 @@ const retrySimulationRef = makeFunctionReference<
{ runId: Id<"workRuns"> }, { runId: Id<"workRuns"> },
unknown unknown
>("workExecution:retrySimulatedExecution"); >("workExecution:retrySimulatedExecution");
const startExecutionRef = makeFunctionReference<
"mutation",
{ workId: Id<"works">; sliceId?: string },
unknown
>("workExecutionWorkflow:startExecution");
const cancelExecutionRef = makeFunctionReference<
"mutation",
{ runId: Id<"workRuns"> },
unknown
>("workExecutionWorkflow:cancelExecution");
const listGitConnectionsRef = makeFunctionReference<
"query",
Record<string, never>,
readonly {
id: string;
provider: "github" | "gitea";
serverUrl: string;
username?: string;
}[]
>("gitConnectionData:list");
const projectGitConnectionRef = makeFunctionReference<
"query",
{ projectId: Id<"projects"> },
{
id: string;
provider: "github" | "gitea";
serverUrl: string;
username?: string;
} | null
>("gitConnectionData:getForProject");
const connectGiteaRef = makeFunctionReference<
"action",
{ serverUrl: string; token: string; username?: string },
{ connectionId: Id<"gitConnections"> }
>("gitConnections:connectGitea");
const connectGithubRef = makeFunctionReference<
"action",
Record<string, never>,
{ connectionId: Id<"gitConnections"> }
>("gitConnections:connectGithub");
const attachGitConnectionRef = makeFunctionReference<
"mutation",
{ connectionId: Id<"gitConnections">; projectId: Id<"projects"> },
unknown
>("gitConnectionData:attachToProject");
const authorizeGithub = () =>
authClient.signIn.social({
callbackURL: window.location.href,
provider: "github",
});
export const useSliceOne = () => { export const useSliceOne = () => {
const organization = usePersonalOrganization(); const organization = usePersonalOrganization();
@@ -189,14 +120,6 @@ export const useSliceOne = () => {
workListRef, workListRef,
activeProjectId ? { projectId: activeProjectId } : "skip" activeProjectId ? { projectId: activeProjectId } : "skip"
); );
const gitConnections = useQuery(
listGitConnectionsRef,
organization.organizationId ? {} : "skip"
);
const projectGitConnection = useQuery(
projectGitConnectionRef,
activeProjectId ? { projectId: activeProjectId } : "skip"
);
const requestDefinitionMutation = useMutation(requestDefinitionRef); const requestDefinitionMutation = useMutation(requestDefinitionRef);
const saveDefinitionMutation = useMutation(saveDefinitionRef); const saveDefinitionMutation = useMutation(saveDefinitionRef);
const approveDefinitionMutation = useMutation(approveDefinitionRef); const approveDefinitionMutation = useMutation(approveDefinitionRef);
@@ -205,11 +128,6 @@ export const useSliceOne = () => {
const startSimulationMutation = useMutation(startSimulationRef); const startSimulationMutation = useMutation(startSimulationRef);
const cancelSimulationMutation = useMutation(cancelSimulationRef); const cancelSimulationMutation = useMutation(cancelSimulationRef);
const retrySimulationMutation = useMutation(retrySimulationRef); const retrySimulationMutation = useMutation(retrySimulationRef);
const startExecutionMutation = useMutation(startExecutionRef);
const cancelExecutionMutation = useMutation(cancelExecutionRef);
const attachGitConnectionMutation = useMutation(attachGitConnectionRef);
const connectGiteaAction = useAction(connectGiteaRef);
const connectGithubAction = useAction(connectGithubRef);
const selectedProject = useMemo( const selectedProject = useMemo(
() => () =>
projects?.find( projects?.find(
@@ -267,46 +185,15 @@ export const useSliceOne = () => {
cancelSimulationMutation({ runId }); cancelSimulationMutation({ runId });
const retrySimulation = (runId: Id<"workRuns">) => const retrySimulation = (runId: Id<"workRuns">) =>
retrySimulationMutation({ runId }); retrySimulationMutation({ runId });
const startExecution = (workId: Id<"works">, sliceId?: string) =>
startExecutionMutation({ sliceId, workId });
const cancelExecution = (runId: Id<"workRuns">) =>
cancelExecutionMutation({ runId });
const attachConnection = async (connectionId: Id<"gitConnections">) => {
if (!activeProjectId) {
throw new Error("Select a project first");
}
await attachGitConnectionMutation({
connectionId,
projectId: activeProjectId,
});
};
const connectGitea = async (input: {
serverUrl: string;
token: string;
username?: string;
}) => {
const result = await connectGiteaAction(input);
await attachConnection(result.connectionId);
};
const connectLinkedGithub = async () => {
const result = await connectGithubAction({});
await attachConnection(result.connectionId);
};
return { return {
agent, agent,
approveDefinition, approveDefinition,
approveDesign, approveDesign,
authorizeGithub,
cancelExecution,
cancelSimulation, cancelSimulation,
connectGitea,
connectLinkedGithub,
connectRepository, connectRepository,
error, error,
gitConnections,
pending, pending,
projectGitConnection,
projects, projects,
repository, repository,
requestDefinition, requestDefinition,
@@ -316,7 +203,6 @@ export const useSliceOne = () => {
selectProject, selectProject,
selectedProject, selectedProject,
setRepository, setRepository,
startExecution,
startSimulation, startSimulation,
works, works,
} as const; } as const;

View File

@@ -7,9 +7,6 @@ import { defineConfig } from "vite-plus";
export default defineConfig({ export default defineConfig({
envDir: path.resolve(import.meta.dirname, "../.."), envDir: path.resolve(import.meta.dirname, "../.."),
plugins: [tailwindcss(), reactRouter()], plugins: [tailwindcss(), reactRouter()],
ssr: {
noExternal: true,
},
resolve: { resolve: {
dedupe: ["convex", "react", "react-dom"], dedupe: ["convex", "react", "react-dom"],
tsconfigPaths: true, tsconfigPaths: true,

838
bun.lock

File diff suppressed because it is too large Load Diff

View File

@@ -1,16 +0,0 @@
FROM oven/bun:1.3.14 AS bun
FROM node:24-bookworm-slim
COPY --from=bun /usr/local/bin/bun /usr/local/bin/bun
RUN apt-get update \
&& apt-get install -y --no-install-recommends g++ make python3 \
&& rm -rf /var/lib/apt/lists/*
WORKDIR /app
COPY . .
RUN bun install --frozen-lockfile
CMD ["bun", "packages/agents/src/runner.ts"]

View File

@@ -1,11 +0,0 @@
services:
zopu-agentos-runner:
build:
context: ../../..
dockerfile: deploy/zopu-runtime/runner/Dockerfile
environment:
RIVET_ENDPOINT: ${RIVET_ENDPOINT}
RIVET_PUBLIC_ENDPOINT: ${RIVET_PUBLIC_ENDPOINT}
RIVET_ENVOY_VERSION: ${RIVET_ENVOY_VERSION}
RIVET_POOL: default
restart: unless-stopped

View File

@@ -18,28 +18,29 @@ Convex application backend
├── normalized product data ├── normalized product data
├── conversation turn queue ├── conversation turn queue
└── reactive client projections └── reactive client projections
durable Workflow steps + service-authenticated dispatch │ service-authenticated dispatch
Private agent backend FLUE orchestration service
├── FLUE product agents and typed tools ├── model calls
├── AgentOS execution environments ├── typed tools
── Codex implementation harness ── canonical Flue persistence in Convex
└── canonical events/results returned to Convex │ later execution commands
│ optional attached full sandbox
Cube/E2B-compatible runtime (later) Rivet Engine + AgentOS (post-Slice 1)
└── sandboxes, harnesses, and durable execution
``` ```
Clients MUST communicate only with Convex. FLUE and future Rivet/AgentOS services are private workers: Convex admits durable commands, invokes the worker, and stores the product-facing result before clients observe it. Clients MUST communicate only with Convex. FLUE and future Rivet/AgentOS
services are private workers: Convex admits durable commands, invokes the
worker, and stores the product-facing result before clients observe it.
### Ownership ### Ownership
| Layer | Owns | | Layer | Owns |
| --- | --- | |---|---|
| Convex | authentication, normalized product records, command admission, reactive reads | | Convex | authentication, normalized product records, command admission, reactive reads |
| Convex Workflow | durable sequencing, retries, cancellation, scheduling, and completion callbacks | | FLUE | private programmable orchestration and domain-specific agents |
| Agent backend | private programmable agents and execution adapters | | Rivet actors (later) | serialized execution ownership, recovery, timers, leases |
| Rivet Engine + runners | placement, routing, sleep/wake, and execution ownership for AgentOS workspaces |
| Harness | bounded coding/tool loop | | Harness | bounded coding/tool loop |
| Sandbox/runtime | filesystem, processes, network, isolation | | Sandbox/runtime | filesystem, processes, network, isolation |
| Git | source revision history | | Git | source revision history |
@@ -62,7 +63,11 @@ signals ──< signalWorkAttachments >── works
works ──< workEvents works ──< workEvents
``` ```
Convex mutations provide atomic transactions and optimistic serializability. Foreign-key integrity and uniqueness are enforced in the owning mutations; composite indexes back every identity and relation lookup. Flue's adapter tables remain isolated infrastructure persistence and are not product-domain relations. Convex mutations provide atomic transactions and optimistic serializability.
Foreign-key integrity and uniqueness are enforced in the owning mutations;
composite indexes back every identity and relation lookup. Flue's adapter
tables remain isolated infrastructure persistence and are not product-domain
relations.
## 2. Domain boundaries ## 2. Domain boundaries
@@ -101,93 +106,93 @@ Avoid package proliferation in v0; logical boundaries may begin as folders.
Minimal durable model: Minimal durable model:
```ts ```ts
type Id = string; type Id = string
interface Message { interface Message {
id: Id; id: Id
projectId: Id; projectId: Id
content: string; content: string
createdAt: number; createdAt: number
} }
interface Signal { interface Signal {
id: Id; id: Id
projectId: Id; projectId: Id
sourceType: string; sourceType: string
sourceId: string; sourceId: string
sourcePayloadRef?: string; sourcePayloadRef?: string
summary: string; summary: string
fingerprint: string; fingerprint: string
status: "candidate" | "accepted" | "dismissed"; status: "candidate" | "accepted" | "dismissed"
} }
interface Work { interface Work {
id: Id; id: Id
projectId: Id; projectId: Id
title: string; title: string
objective: string; objective: string
risk: "low" | "medium" | "high"; risk: "low" | "medium" | "high"
status: WorkStatus; status: WorkStatus
definitionVersion?: number; definitionVersion?: number
designVersion?: number; designVersion?: number
createdAt: number; createdAt: number
updatedAt: number; updatedAt: number
} }
interface Step { interface Step {
id: Id; id: Id
workId: Id; workId: Id
sliceId?: Id; sliceId?: Id
kind: "design" | "implement" | "verify" | "integrate" | "publish" | "observe"; kind: "design" | "implement" | "verify" | "integrate" | "publish" | "observe"
objective: string; objective: string
dependsOn: readonly Id[]; dependsOn: readonly Id[]
status: StepStatus; status: StepStatus
} }
interface Run { interface Run {
id: Id; id: Id
workId: Id; workId: Id
stepId: Id; stepId: Id
kitVersion: string; kitVersion: string
status: RunStatus; status: RunStatus
} }
interface Attempt { interface Attempt {
id: Id; id: Id
runId: Id; runId: Id
number: number; number: number
harness: string; harness: string
runtime: string; runtime: string
sourceRevision: string; sourceRevision: string
status: AttemptStatus; status: AttemptStatus
startedAt?: number; startedAt?: number
endedAt?: number; endedAt?: number
} }
interface Artifact { interface Artifact {
id: Id; id: Id
workId: Id; workId: Id
stepId?: Id; stepId?: Id
runId?: Id; runId?: Id
attemptId?: Id; attemptId?: Id
type: string; type: string
uri?: string; uri?: string
contentHash?: string; contentHash?: string
sourceRevision?: string; sourceRevision?: string
environmentId?: string; environmentId?: string
metadata: unknown; metadata: unknown
} }
interface Question { interface Question {
id: Id; id: Id
workId: Id; workId: Id
stepId?: Id; stepId?: Id
attemptId?: Id; attemptId?: Id
prompt: string; prompt: string
recommendation?: string; recommendation?: string
alternatives: readonly string[]; alternatives: readonly string[]
status: "open" | "answered" | "withdrawn"; status: "open" | "answered" | "withdrawn"
answer?: string; answer?: string
} }
``` ```
@@ -332,49 +337,35 @@ Keep domain/application code provider-neutral.
```ts ```ts
interface HarnessRuntime { interface HarnessRuntime {
open(input: OpenHarnessInput): Effect.Effect<HarnessSession, HarnessError>; open(input: OpenHarnessInput): Effect.Effect<HarnessSession, HarnessError>
prompt(id: string, content: string): Effect.Effect<void, HarnessError>; prompt(id: string, content: string): Effect.Effect<void, HarnessError>
events(id: string): Stream.Stream<HarnessEvent, HarnessError>; events(id: string): Stream.Stream<HarnessEvent, HarnessError>
approve(input: PermissionDecision): Effect.Effect<void, HarnessError>; approve(input: PermissionDecision): Effect.Effect<void, HarnessError>
abort(id: string): Effect.Effect<void, HarnessError>; abort(id: string): Effect.Effect<void, HarnessError>
close(id: string): Effect.Effect<void, HarnessError>; close(id: string): Effect.Effect<void, HarnessError>
} }
interface SandboxRuntime { interface SandboxRuntime {
create(spec: SandboxSpec): Effect.Effect<SandboxLease, SandboxError>; create(spec: SandboxSpec): Effect.Effect<SandboxLease, SandboxError>
exec( exec(lease: SandboxLease, cmd: Command): Effect.Effect<CommandResult, SandboxError>
lease: SandboxLease, readFile(lease: SandboxLease, path: string): Effect.Effect<Uint8Array, SandboxError>
cmd: Command writeFile(lease: SandboxLease, path: string, body: Uint8Array): Effect.Effect<void, SandboxError>
): Effect.Effect<CommandResult, SandboxError>; pause(lease: SandboxLease): Effect.Effect<void, SandboxError>
readFile( resume(id: string): Effect.Effect<SandboxLease, SandboxError>
lease: SandboxLease, terminate(lease: SandboxLease): Effect.Effect<void, SandboxError>
path: string
): Effect.Effect<Uint8Array, SandboxError>;
writeFile(
lease: SandboxLease,
path: string,
body: Uint8Array
): Effect.Effect<void, SandboxError>;
pause(lease: SandboxLease): Effect.Effect<void, SandboxError>;
resume(id: string): Effect.Effect<SandboxLease, SandboxError>;
terminate(lease: SandboxLease): Effect.Effect<void, SandboxError>;
} }
interface SourceControl { interface SourceControl {
prepareWorktree(input: WorktreeInput): Effect.Effect<Worktree, GitError>; prepareWorktree(input: WorktreeInput): Effect.Effect<Worktree, GitError>
diff(worktree: Worktree): Effect.Effect<DiffArtifact, GitError>; diff(worktree: Worktree): Effect.Effect<DiffArtifact, GitError>
commit(input: CommitInput): Effect.Effect<CommitArtifact, GitError>; commit(input: CommitInput): Effect.Effect<CommitArtifact, GitError>
push(input: PushInput): Effect.Effect<BranchArtifact, GitError>; push(input: PushInput): Effect.Effect<BranchArtifact, GitError>
createPullRequest( createPullRequest(input: PullRequestInput): Effect.Effect<PullRequestArtifact, GitError>
input: PullRequestInput
): Effect.Effect<PullRequestArtifact, GitError>;
} }
interface VerificationRuntime { interface VerificationRuntime {
execute( execute(plan: VerificationPlan, env: EnvironmentRef):
plan: VerificationPlan, Effect.Effect<VerificationResult, VerificationError>
env: EnvironmentRef
): Effect.Effect<VerificationResult, VerificationError>;
} }
``` ```
@@ -429,10 +420,6 @@ Zopu owns Work, branches, worktrees, budgets, approvals, artifacts, and completi
## 10. Runtime strategy ## 10. Runtime strategy
### Convex orchestration
Slice 5 uses the Convex Workflow component as the durable product orchestration layer. Workflow steps call private agent-backend actions, while all user-visible state, events, artifacts, idempotency keys, and cancellation remain canonical in Convex.
### CubeSandbox ### CubeSandbox
Best for full Linux execution: Best for full Linux execution:
@@ -456,7 +443,7 @@ Best for:
- actor-adjacent orchestration; - actor-adjacent orchestration;
- context/files/networking that fit runtime limits. - context/files/networking that fit runtime limits.
Slice 5 starts here with one Codex-backed AgentOS actor and one authenticated repository checkout per project. Convex Workflow owns product orchestration; Rivet Engine coordinates placement while a normal runner executes the actor. Use an attached full sandbox through the E2B-compatible boundary when native/heavy tooling is needed. Use an attached full sandbox when native/heavy tooling is needed.
### Persistent project machine ### Persistent project machine

View File

@@ -211,18 +211,16 @@ Zopu implements one approved slice in an isolated repository environment and str
Initial adapter choice: Initial adapter choice:
```text ```text
SandboxRuntime = AgentOsSandboxLive SandboxRuntime = CubeSandboxLive
HarnessRuntime = CodexHarnessLive HarnessRuntime = OmpHarnessLive (or one chosen harness)
Durable orchestration = Convex Workflow
``` ```
Flow: Flow:
```text ```text
load the project's authenticated Git connection prepare worktree
start durable Convex workflow create sandbox
→ create AgentOS execution environment → clone/mount repo
→ clone the single configured repo
→ inject context → inject context
→ run one slice → run one slice
→ normalize events → normalize events
@@ -232,15 +230,14 @@ load the project's authenticated Git connection
Security: Security:
- GitHub OAuth or self-hosted Gitea PAT, scoped to one project; - scoped Git/model tokens;
- scoped Git/model tokens passed only to private execution;
- isolated HOME/worktree; - isolated HOME/worktree;
- one mutating attempt per worktree; - one mutating attempt per worktree;
- timeout/cancel cleanup. - timeout/cancel cleanup.
### Frontend ### Frontend
Project selector/settings, Git connection status, current activity, changed files, artifact links, expandable raw logs, cancel/retry, and manual Git delivery controls. Current activity, changed files, artifact links, expandable raw logs.
### Acceptance ### Acceptance
@@ -250,8 +247,6 @@ Project selector/settings, Git connection status, current activity, changed file
- provider failure becomes classified attempt outcome; - provider failure becomes classified attempt outcome;
- exact base/candidate revision recorded. - exact base/candidate revision recorded.
Rivet Engine coordinates AgentOS actor placement through a normal runner, but does not own product orchestration; Convex Workflow remains canonical for the Run lifecycle. Cube/Kubernetes sandbox support remains behind `SandboxRuntime` and can be mounted through AgentOS incrementally.
--- ---
## Slice 6 — Independent verification and repair ## Slice 6 — Independent verification and repair
@@ -495,4 +490,4 @@ browser-tester integration-coordinator
1 → 2 → 3 → 4 → 5 → 6 → 7 → 8 → 9 → 10 → 11 → 12 1 → 2 → 3 → 4 → 5 → 6 → 7 → 8 → 9 → 10 → 11 → 12
``` ```
Parallel engineering is allowed _within_ a slice after contracts land, but product release order remains sequential. Parallel engineering is allowed *within* a slice after contracts land, but product release order remains sequential.

View File

@@ -5,28 +5,5 @@ import remix from "ultracite/oxlint/remix";
export default defineConfig({ export default defineConfig({
extends: [core, react, remix], extends: [core, react, remix],
ignorePatterns: [...core.ignorePatterns, "repos/**", "scripts/**"], ignorePatterns: core.ignorePatterns,
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",
},
},
],
}); });

View File

@@ -37,15 +37,7 @@
"heroui-native": "^1.0.5", "heroui-native": "^1.0.5",
"vite": "^7.3.6", "vite": "^7.3.6",
"vitest": "^4.1.10", "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", "type": "module",
@@ -53,8 +45,8 @@
"dev": "vp run -r dev", "dev": "vp run -r dev",
"build": "vp run -r build", "build": "vp run -r build",
"check-types": "vp run -r check-types", "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/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", "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/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/resolver.ts packages/primitives/src/work-lifecycle.ts packages/primitives/src/work-resolution.test.ts",
"lint": "oxlint --disable-nested-config", "lint": "vp lint",
"format": "vp fmt", "format": "vp fmt",
"staged": "vp staged", "staged": "vp staged",
"hooks:setup": "vp config", "hooks:setup": "vp config",
@@ -80,7 +72,7 @@
"@code/primitives": "workspace:*", "@code/primitives": "workspace:*",
"@flue/sdk": "1.0.0-beta.9", "@flue/sdk": "1.0.0-beta.9",
"@types/bun": "catalog:", "@types/bun": "catalog:",
"@types/node": "catalog:", "@types/node": "^22.13.14",
"convex": "catalog:", "convex": "catalog:",
"convex-test": "catalog:", "convex-test": "catalog:",
"effect": "catalog:", "effect": "catalog:",
@@ -93,8 +85,6 @@
"vitest": "catalog:" "vitest": "catalog:"
}, },
"overrides": { "overrides": {
"react": "19.2.8",
"react-dom": "19.2.8",
"vite": "npm:@voidzero-dev/vite-plus-core@0.2.2" "vite": "npm:@voidzero-dev/vite-plus-core@0.2.2"
}, },
"packageManager": "bun@1.3.14" "packageManager": "bun@1.3.14"

View File

@@ -9,22 +9,16 @@
"dev": "bun --env-file=../../.env flue dev", "dev": "bun --env-file=../../.env flue dev",
"dev:tailscale": "bun --env-file=../../.env flue dev", "dev:tailscale": "bun --env-file=../../.env flue dev",
"run": "bun --env-file=../../.env flue run", "run": "bun --env-file=../../.env flue run",
"runner": "bun --env-file=../../.env src/runner.ts",
"run:zopu": "bun --env-file=../../.env flue run zopu", "run:zopu": "bun --env-file=../../.env flue run zopu",
"run:work-planner": "bun --env-file=../../.env flue run work-planner" "run:work-planner": "bun --env-file=../../.env flue run work-planner"
}, },
"dependencies": { "dependencies": {
"@agentos-software/codex-cli": "0.3.4",
"@agentos-software/git": "0.3.3",
"@code/backend": "workspace:*", "@code/backend": "workspace:*",
"@code/env": "workspace:*", "@code/env": "workspace:*",
"@code/primitives": "workspace:*",
"@flue/runtime": "latest", "@flue/runtime": "latest",
"@rivet-dev/agentos": "0.2.10",
"convex": "catalog:", "convex": "catalog:",
"hono": "catalog:", "hono": "4.12.31",
"rivetkit": "2.3.9", "valibot": "^1.4.2"
"valibot": "catalog:"
}, },
"devDependencies": { "devDependencies": {
"@code/config": "workspace:*", "@code/config": "workspace:*",

View File

@@ -1,15 +1,8 @@
import { parseAgentEnv } from "@code/env/agent"; import { parseAgentEnv } from "@code/env/agent";
import { registerProvider } from "@flue/runtime"; import { registerProvider } from "@flue/runtime";
import type { Fetchable } from "@flue/runtime/routing";
import { flue } from "@flue/runtime/routing"; import { flue } from "@flue/runtime/routing";
import { Hono } from "hono"; import { Hono } from "hono";
import {
cancelAgentOsAttempt,
executeAgentOsAttempt,
runtimeRegistry,
} from "./runtime/agent-os";
const agentEnv = parseAgentEnv(process.env); const agentEnv = parseAgentEnv(process.env);
registerProvider(agentEnv.AGENT_MODEL_PROVIDER, { registerProvider(agentEnv.AGENT_MODEL_PROVIDER, {
@@ -32,34 +25,6 @@ registerProvider(agentEnv.AGENT_MODEL_PROVIDER, {
}); });
const app = new Hono(); const app = new Hono();
app.route("/", flue() as unknown as Hono);
app.post("/internal/work-attempts/execute", async (context) => { export default app;
if (
context.req.header("authorization") !== `Bearer ${agentEnv.FLUE_DB_TOKEN}`
) {
return context.json({ error: "Unauthorized" }, 401);
}
try {
return context.json(await executeAgentOsAttempt(await context.req.json()));
} catch (error) {
return context.json(
{ error: error instanceof Error ? error.message : "Execution failed" },
500
);
}
});
app.post("/internal/work-attempts/:workspaceKey/cancel", async (context) => {
if (
context.req.header("authorization") !== `Bearer ${agentEnv.FLUE_DB_TOKEN}`
) {
return context.json({ error: "Unauthorized" }, 401);
}
await cancelAgentOsAttempt(context.req.param("workspaceKey"));
return context.json({ cancelled: true });
});
app.all("/api/rivet/*", (context) => runtimeRegistry.handler(context.req.raw));
app.route("/", flue());
export default app satisfies Fetchable;

View File

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

View File

@@ -1,3 +0,0 @@
import { runtimeRegistry } from "./runtime/agent-os";
await runtimeRegistry.startAndWait();

View File

@@ -1,213 +0,0 @@
import { parseAgentEnv } from "@code/env/agent";
import { agentOS, setup } from "@rivet-dev/agentos";
import { createClient } from "@rivet-dev/agentos/client";
import { Effect } from "effect";
import {
codexSessionEnv,
makeCodexAgentOsConfig,
} from "../../../primitives/src/agent-os";
import { decodeWorkAttemptExecutionInput } from "../../../primitives/src/execution-runtime";
import type {
ExecutionEvent,
WorkAttemptExecutionResult,
} from "../../../primitives/src/execution-runtime";
const workspace = agentOS(makeCodexAgentOsConfig() as never);
export const runtimeRegistry = setup({ use: { workspace } });
const shellQuote = (value: string): string =>
`'${value.replaceAll("'", `'"'"'`)}'`;
const event = (
sequence: number,
kind: ExecutionEvent["kind"],
message: string,
metadata: Record<string, string> = {}
): ExecutionEvent => ({
kind,
message,
metadata,
occurredAt: Date.now(),
sequence,
});
const requireSuccess = (
result: { exitCode: number; stderr: string; stdout: string },
operation: string
) => {
if (result.exitCode !== 0) {
throw new Error(`${operation} failed: ${result.stderr || result.stdout}`);
}
return result.stdout.trim();
};
const gitAuthEnv = (username: string, credential: string) => ({
GIT_CONFIG_COUNT: "1",
GIT_CONFIG_KEY_0: "http.extraHeader",
GIT_CONFIG_VALUE_0: `Authorization: Basic ${Buffer.from(`${username}:${credential}`).toString("base64")}`,
});
export const executeAgentOsAttempt = async (
rawInput: unknown
): Promise<WorkAttemptExecutionResult> => {
const input = await Effect.runPromise(
decodeWorkAttemptExecutionInput(rawInput)
);
const env = parseAgentEnv(process.env);
const client = createClient<typeof runtimeRegistry>({
endpoint: env.RIVET_PUBLIC_ENDPOINT ?? env.RIVET_ENDPOINT,
});
const vm = client.workspace.getOrCreate([input.workspaceKey]);
const events: ExecutionEvent[] = [
event(0, "runtime.preparing", "AgentOS workspace selected", {
workspaceKey: input.workspaceKey,
}),
];
const authEnv = gitAuthEnv(
input.auth.username ??
(input.auth.provider === "github" ? "x-access-token" : "git"),
input.auth.credential
);
await vm.mkdir("/workspace", { recursive: true });
if (!(await vm.exists("/workspace/repository/.git"))) {
events.push(event(1, "repository.cloning", "Cloning project repository"));
const clone = await vm.execArgv(
"git",
["clone", input.repositoryUrl, "/workspace/repository"],
{
cwd: "/workspace",
env: authEnv,
}
);
requireSuccess(clone, "Repository clone");
}
const checkout = await vm.exec(
`git fetch origin ${shellQuote(input.baseBranch)} && git checkout -B ${shellQuote(`zopu/${input.runId}`)} ${shellQuote(`origin/${input.baseBranch}`)}`,
{ cwd: "/workspace/repository", env: authEnv }
);
requireSuccess(checkout, "Repository checkout");
const baseRevision = requireSuccess(
await vm.execArgv("git", ["rev-parse", "HEAD"], {
cwd: "/workspace/repository",
}),
"Base revision lookup"
);
events.push(
event(2, "repository.ready", "Repository checkout is ready", {
baseRevision,
})
);
const sessionId = `codex-${input.attemptId}`;
await vm.openSession({
additionalInstructions:
"Work only inside /workspace/repository. Do not reveal credentials. Make the requested change and run focused verification. Do not push or open a pull request.",
agent: "codex",
cwd: "/workspace/repository",
env: codexSessionEnv({
apiKey: env.AGENT_MODEL_API_KEY,
baseUrl: env.AGENT_MODEL_BASE_URL,
model: env.AGENT_MODEL_NAME,
}),
permissionPolicy: "allow_all",
sessionId,
});
events.push(
event(3, "harness.started", "Codex implementation session started")
);
const promptResult = await vm.prompt({
content: [{ text: input.prompt, type: "text" }],
idempotencyKey: input.attemptId,
sessionId,
});
events.push(
event(4, "harness.progress", "Codex implementation turn completed", {
stopReason: String(promptResult.stopReason),
})
);
const status = requireSuccess(
await vm.execArgv("git", ["status", "--porcelain"], {
cwd: "/workspace/repository",
}),
"Changed file lookup"
);
const diff = requireSuccess(
await vm.execArgv("git", ["diff", "--binary", "HEAD"], {
cwd: "/workspace/repository",
}),
"Diff collection"
);
const changedFiles = status
.split("\n")
.filter(Boolean)
.map((line) => line.slice(3).trim());
let candidateRevision = baseRevision;
if (changedFiles.length > 0) {
requireSuccess(
await vm.execArgv("git", ["add", "-A"], { cwd: "/workspace/repository" }),
"Candidate staging"
);
const tree = requireSuccess(
await vm.execArgv("git", ["write-tree"], {
cwd: "/workspace/repository",
}),
"Candidate tree creation"
);
candidateRevision = requireSuccess(
await vm.exec(
`printf %s ${shellQuote(`Zopu candidate for ${input.attemptId}`)} | git commit-tree ${shellQuote(tree)} -p ${shellQuote(baseRevision)}`,
{
cwd: "/workspace/repository",
env: {
GIT_AUTHOR_EMAIL: "agent@zopu.dev",
GIT_AUTHOR_NAME: "Zopu Agent",
GIT_COMMITTER_EMAIL: "agent@zopu.dev",
GIT_COMMITTER_NAME: "Zopu Agent",
},
}
),
"Candidate revision creation"
);
requireSuccess(
await vm.execArgv("git", ["reset"], { cwd: "/workspace/repository" }),
"Candidate index reset"
);
}
events.push(
event(
5,
"repository.changed",
`${changedFiles.length} changed file(s) collected`,
{
candidateRevision,
}
),
event(6, "runtime.completed", "AgentOS execution completed")
);
return {
baseRevision,
candidateRevision,
changedFiles,
diff,
environmentId: input.workspaceKey,
events,
summary:
changedFiles.length > 0
? `Codex changed ${changedFiles.length} file(s)`
: "Codex completed without repository changes",
};
};
export const cancelAgentOsAttempt = async (workspaceKey: string) => {
const env = parseAgentEnv(process.env);
const client = createClient<typeof runtimeRegistry>({
endpoint: env.RIVET_PUBLIC_ENDPOINT ?? env.RIVET_ENDPOINT,
});
await client.workspace.getOrCreate([workspaceKey]).cancelPrompt({});
};

View File

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

View File

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

View File

@@ -1,6 +1,8 @@
# Convex Backend # Convex Backend
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. 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.
## Domain relations ## Domain relations
@@ -9,9 +11,6 @@ Convex is the only application API used by web, desktop, and mobile clients. The
- `conversations`, `conversationTurns`, `conversationMessages`, and `conversationAttachments` own chat admission and reactive history. - `conversations`, `conversationTurns`, `conversationMessages`, and `conversationAttachments` own chat admission and reactive history.
- `signals`, `signalConstraints`, and `signalSources` preserve structured intent and exact evidence. - `signals`, `signalConstraints`, and `signalSources` preserve structured intent and exact evidence.
- `works`, `signalWorkAttachments`, and `workEvents` own proposed Work and provenance. - `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.

View File

@@ -11,10 +11,7 @@
import type * as auth from "../auth.js"; import type * as auth from "../auth.js";
import type * as authz from "../authz.js"; import type * as authz from "../authz.js";
import type * as conversationMessages from "../conversationMessages.js"; import type * as conversationMessages from "../conversationMessages.js";
import type * as crons from "../crons.js";
import type * as fluePersistence from "../fluePersistence.js"; import type * as fluePersistence from "../fluePersistence.js";
import type * as gitConnectionData from "../gitConnectionData.js";
import type * as gitConnections from "../gitConnections.js";
import type * as healthCheck from "../healthCheck.js"; import type * as healthCheck from "../healthCheck.js";
import type * as http from "../http.js"; import type * as http from "../http.js";
import type * as organizations from "../organizations.js"; import type * as organizations from "../organizations.js";
@@ -22,11 +19,6 @@ import type * as privateData from "../privateData.js";
import type * as projects from "../projects.js"; import type * as projects from "../projects.js";
import type * as publicGit from "../publicGit.js"; import type * as publicGit from "../publicGit.js";
import type * as signalRouting from "../signalRouting.js"; import type * as signalRouting from "../signalRouting.js";
import type * as workArtifacts from "../workArtifacts.js";
import type * as workExecution from "../workExecution.js";
import type * as workExecutionAgent from "../workExecutionAgent.js";
import type * as workExecutionWorkflow from "../workExecutionWorkflow.js";
import type * as workPlanning from "../workPlanning.js";
import type * as works from "../works.js"; import type * as works from "../works.js";
import type { import type {
@@ -39,10 +31,7 @@ declare const fullApi: ApiFromModules<{
auth: typeof auth; auth: typeof auth;
authz: typeof authz; authz: typeof authz;
conversationMessages: typeof conversationMessages; conversationMessages: typeof conversationMessages;
crons: typeof crons;
fluePersistence: typeof fluePersistence; fluePersistence: typeof fluePersistence;
gitConnectionData: typeof gitConnectionData;
gitConnections: typeof gitConnections;
healthCheck: typeof healthCheck; healthCheck: typeof healthCheck;
http: typeof http; http: typeof http;
organizations: typeof organizations; organizations: typeof organizations;
@@ -50,11 +39,6 @@ declare const fullApi: ApiFromModules<{
projects: typeof projects; projects: typeof projects;
publicGit: typeof publicGit; publicGit: typeof publicGit;
signalRouting: typeof signalRouting; signalRouting: typeof signalRouting;
workArtifacts: typeof workArtifacts;
workExecution: typeof workExecution;
workExecutionAgent: typeof workExecutionAgent;
workExecutionWorkflow: typeof workExecutionWorkflow;
workPlanning: typeof workPlanning;
works: typeof works; works: typeof works;
}>; }>;
@@ -86,5 +70,4 @@ export declare const internal: FilterApi<
export declare const components: { export declare const components: {
betterAuth: import("@convex-dev/better-auth/_generated/component.js").ComponentApi<"betterAuth">; betterAuth: import("@convex-dev/better-auth/_generated/component.js").ComponentApi<"betterAuth">;
workflow: import("@convex-dev/workflow/_generated/component.js").ComponentApi<"workflow">;
}; };

View File

@@ -25,14 +25,10 @@ import type { DataModel } from "./dataModel.js";
* Typesafe environment variables declared in `convex.config.ts`. * Typesafe environment variables declared in `convex.config.ts`.
*/ */
type Env = { type Env = {
readonly AGENT_BACKEND_URL: string | undefined;
readonly FLUE_DB_TOKEN: string; readonly FLUE_DB_TOKEN: string;
readonly FLUE_URL: string | undefined; readonly FLUE_URL: string | undefined;
readonly GITEA_TOKEN: string | undefined; readonly GITEA_TOKEN: string | undefined;
readonly GITEA_URL: string | undefined; readonly GITEA_URL: string | undefined;
readonly GITHUB_CLIENT_ID: string | undefined;
readonly GITHUB_CLIENT_SECRET: string | undefined;
readonly GIT_CREDENTIAL_ENCRYPTION_KEY: string | undefined;
readonly NATIVE_APP_URL: string | undefined; readonly NATIVE_APP_URL: string | undefined;
readonly SITE_URL: string; readonly SITE_URL: string;
}; };

View File

@@ -30,15 +30,6 @@ const createAuth = (ctx: GenericCtx<DataModel>) =>
jwksRotateOnTokenGenerationError: true, jwksRotateOnTokenGenerationError: true,
}), }),
], ],
socialProviders:
env.GITHUB_CLIENT_ID && env.GITHUB_CLIENT_SECRET
? {
github: {
clientId: env.GITHUB_CLIENT_ID,
clientSecret: env.GITHUB_CLIENT_SECRET,
},
}
: {},
trustedOrigins: [siteUrl, nativeAppUrl, "exp://"], trustedOrigins: [siteUrl, nativeAppUrl, "exp://"],
}); });

View File

@@ -1,5 +1,5 @@
import { convexTest } from "convex-test"; import { convexTest } from "convex-test";
import { anyApi, makeFunctionReference } from "convex/server"; import { anyApi } from "convex/server";
import { describe, expect, test } from "vitest"; import { describe, expect, test } from "vitest";
import schema from "./schema"; import schema from "./schema";
@@ -15,33 +15,6 @@ const api = anyApi;
const identityA = { tokenIdentifier: "https://convex.test|user-a" }; const identityA = { tokenIdentifier: "https://convex.test|user-a" };
const identityB = { tokenIdentifier: "https://convex.test|user-b" }; const identityB = { tokenIdentifier: "https://convex.test|user-b" };
const newTest = () => convexTest({ modules, schema }); 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 ( const ensureOrg = async (
t: ReturnType<typeof newTest>, t: ReturnType<typeof newTest>,
@@ -136,66 +109,4 @@ describe("conversationMessages", () => {
}) })
).rejects.toThrow(/Organization membership required/u); ).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");
});
}); });

View File

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

View File

@@ -1,23 +1,17 @@
import betterAuth from "@convex-dev/better-auth/convex.config"; import betterAuth from "@convex-dev/better-auth/convex.config";
import workflow from "@convex-dev/workflow/convex.config.js";
import { defineApp } from "convex/server"; import { defineApp } from "convex/server";
import { v } from "convex/values"; import { v } from "convex/values";
const app = defineApp({ const app = defineApp({
env: { env: {
AGENT_BACKEND_URL: v.optional(v.string()),
FLUE_DB_TOKEN: v.string(),
FLUE_URL: v.optional(v.string()),
GITEA_TOKEN: v.optional(v.string()),
GITEA_URL: v.optional(v.string()),
GITHUB_CLIENT_ID: v.optional(v.string()),
GITHUB_CLIENT_SECRET: v.optional(v.string()),
GIT_CREDENTIAL_ENCRYPTION_KEY: v.optional(v.string()),
NATIVE_APP_URL: v.optional(v.string()), NATIVE_APP_URL: v.optional(v.string()),
SITE_URL: 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()),
}, },
}); });
app.use(betterAuth); app.use(betterAuth);
app.use(workflow);
export default app; export default app;

View File

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

View File

@@ -1,142 +0,0 @@
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",
]);
});
});

File diff suppressed because it is too large Load Diff

View File

@@ -1,119 +0,0 @@
import { ConvexError, v } from "convex/values";
import { internalMutation, mutation, query } from "./_generated/server";
import { requireCurrentOrganization, requireProjectMember } from "./authz";
export const persist = internalMutation({
args: {
credentialCiphertext: v.string(),
credentialIv: v.string(),
credentialKind: v.union(v.literal("oauth"), v.literal("token")),
provider: v.union(v.literal("github"), v.literal("gitea")),
serverUrl: v.string(),
userId: v.string(),
username: v.optional(v.string()),
},
handler: async (ctx, args) => {
const organization = await ctx.db
.query("organizations")
.withIndex("by_createdBy_and_kind", (q) =>
q.eq("createdBy", args.userId).eq("kind", "personal")
)
.unique();
if (!organization) {
throw new ConvexError("Organization not found");
}
const existing = await ctx.db
.query("gitConnections")
.withIndex("by_organizationId_and_provider_and_serverUrl", (q) =>
q
.eq("organizationId", organization._id)
.eq("provider", args.provider)
.eq("serverUrl", args.serverUrl)
)
.unique();
const timestamp = Date.now();
if (existing) {
await ctx.db.patch(existing._id, {
credentialCiphertext: args.credentialCiphertext,
credentialIv: args.credentialIv,
credentialKind: args.credentialKind,
updatedAt: timestamp,
username: args.username,
});
return existing._id;
}
return await ctx.db.insert("gitConnections", {
connectedAt: timestamp,
credentialCiphertext: args.credentialCiphertext,
credentialIv: args.credentialIv,
credentialKind: args.credentialKind,
organizationId: organization._id,
provider: args.provider,
serverUrl: args.serverUrl,
updatedAt: timestamp,
username: args.username,
});
},
});
export const list = query({
args: {},
handler: async (ctx) => {
const { organizationId } = await requireCurrentOrganization(ctx);
const connections = await ctx.db
.query("gitConnections")
.withIndex("by_organizationId", (q) =>
q.eq("organizationId", organizationId)
)
.collect();
return connections.map((connection) => ({
connectedAt: connection.connectedAt,
credentialKind: connection.credentialKind,
id: String(connection._id),
provider: connection.provider,
serverUrl: connection.serverUrl,
username: connection.username,
}));
},
});
export const getForProject = query({
args: { projectId: v.id("projects") },
handler: async (ctx, args) => {
await requireProjectMember(ctx, args.projectId);
const project = await ctx.db.get(args.projectId);
const connection = project?.gitConnectionId
? await ctx.db.get(project.gitConnectionId)
: null;
return connection
? {
connectedAt: connection.connectedAt,
credentialKind: connection.credentialKind,
id: String(connection._id),
provider: connection.provider,
serverUrl: connection.serverUrl,
username: connection.username,
}
: null;
},
});
export const attachToProject = mutation({
args: {
connectionId: v.id("gitConnections"),
projectId: v.id("projects"),
},
handler: async (ctx, args) => {
const { organizationId } = await requireProjectMember(ctx, args.projectId);
const connection = await ctx.db.get(args.connectionId);
if (!connection || connection.organizationId !== organizationId) {
throw new ConvexError("Git connection not found");
}
await ctx.db.patch(args.projectId, {
gitConnectionId: connection._id,
updatedAt: Date.now(),
});
return { attached: true };
},
});

View File

@@ -1,124 +0,0 @@
"use node";
import { env } from "@code/env/convex";
import { decodeGitConnectionInput } from "@code/primitives/execution-runtime";
import { ConvexError, v } from "convex/values";
import { Effect } from "effect";
import { internal } from "./_generated/api";
import type { Id } from "./_generated/dataModel";
import { action } from "./_generated/server";
import { authComponent, createAuth } from "./auth";
const encryptionKey = async (): Promise<CryptoKey> => {
if (!env.GIT_CREDENTIAL_ENCRYPTION_KEY) {
throw new ConvexError("Git credential encryption is not configured");
}
const bytes = Buffer.from(env.GIT_CREDENTIAL_ENCRYPTION_KEY, "base64url");
if (bytes.byteLength !== 32) {
throw new ConvexError("Git credential encryption key must be 32 bytes");
}
return await crypto.subtle.importKey("raw", bytes, "AES-GCM", false, [
"encrypt",
"decrypt",
]);
};
const encryptCredential = async (credential: string) => {
const iv = crypto.getRandomValues(new Uint8Array(12));
const encrypted = await crypto.subtle.encrypt(
{ iv, name: "AES-GCM" },
await encryptionKey(),
new TextEncoder().encode(credential)
);
return {
credentialCiphertext: Buffer.from(encrypted).toString("base64url"),
credentialIv: Buffer.from(iv).toString("base64url"),
};
};
export const decryptCredential = async (
credentialCiphertext: string,
credentialIv: string
): Promise<string> => {
const decrypted = await crypto.subtle.decrypt(
{
iv: Buffer.from(credentialIv, "base64url"),
name: "AES-GCM",
},
await encryptionKey(),
Buffer.from(credentialCiphertext, "base64url")
);
return new TextDecoder().decode(decrypted);
};
export const connectGitea = action({
args: {
serverUrl: v.string(),
token: v.string(),
username: v.optional(v.string()),
},
handler: async (
ctx,
args
): Promise<{ connectionId: Id<"gitConnections"> }> => {
const userId = await ctx.auth.getUserIdentity().then((identity) => {
if (!identity) {
throw new ConvexError("Authentication required");
}
return identity.tokenIdentifier;
});
const connection = await Effect.runPromise(
decodeGitConnectionInput({
credential: args.token,
credentialKind: "token",
provider: "gitea",
serverUrl: args.serverUrl,
username: args.username,
})
);
const encrypted = await encryptCredential(connection.credential);
const connectionId = await ctx.runMutation(
internal.gitConnectionData.persist,
{
...encrypted,
credentialKind: connection.credentialKind,
provider: connection.provider,
serverUrl: connection.serverUrl,
userId,
username: connection.username,
}
);
return { connectionId };
},
});
export const connectGithub = action({
args: {},
handler: async (ctx): Promise<{ connectionId: Id<"gitConnections"> }> => {
const identity = await ctx.auth.getUserIdentity();
if (!identity) {
throw new ConvexError("Authentication required");
}
const { auth, headers } = await authComponent.getAuth(createAuth, ctx);
const token = await auth.api.getAccessToken({
body: { providerId: "github" },
headers,
});
if (!token.accessToken) {
throw new ConvexError("GitHub account is not connected");
}
const encrypted = await encryptCredential(token.accessToken);
const connectionId = await ctx.runMutation(
internal.gitConnectionData.persist,
{
...encrypted,
credentialKind: "oauth",
provider: "github",
serverUrl: "https://github.com",
userId: identity.tokenIdentifier,
}
);
return { connectionId };
},
});

View File

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

View File

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

View File

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

View File

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

View File

@@ -1,77 +1,32 @@
import { defineSchema, defineTable } from "convex/server"; import { defineSchema, defineTable } from "convex/server";
import { v } from "convex/values"; import { v } from "convex/values";
/* eslint-disable sort-keys */
const attemptClassification = v.union(
v.literal("Succeeded"),
v.literal("RetryableFailure"),
v.literal("NeedsInput"),
v.literal("Blocked"),
v.literal("VerificationFailed"),
v.literal("BudgetExhausted"),
v.literal("Cancelled"),
v.literal("PermanentFailure")
);
const workStatus = v.union(
v.literal("proposed"),
v.literal("defining"),
v.literal("awaiting-definition-approval"),
v.literal("designing"),
v.literal("awaiting-design-approval"),
v.literal("ready"),
v.literal("executing"),
v.literal("needs-input"),
v.literal("blocked"),
v.literal("completed"),
v.literal("failed"),
v.literal("cancelled")
);
export default defineSchema({ export default defineSchema({
organizations: defineTable({ organizations: defineTable({
createdAt: v.number(),
createdBy: v.string(),
kind: v.union(v.literal("personal"), v.literal("team")),
name: v.string(), 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"]), }).index("by_createdBy_and_kind", ["createdBy", "kind"]),
organizationMembers: defineTable({ organizationMembers: defineTable({
createdAt: v.number(),
organizationId: v.id("organizations"), organizationId: v.id("organizations"),
role: v.union(v.literal("owner"), v.literal("member")),
userId: v.string(), userId: v.string(),
role: v.union(v.literal("owner"), v.literal("member")),
createdAt: v.number(),
}) })
.index("by_userId", ["userId"]) .index("by_userId", ["userId"])
.index("by_organizationId", ["organizationId"]) .index("by_organizationId", ["organizationId"])
.index("by_organizationId_and_userId", ["organizationId", "userId"]), .index("by_organizationId_and_userId", ["organizationId", "userId"]),
gitConnections: defineTable({
connectedAt: v.number(),
credentialCiphertext: v.string(),
credentialIv: v.string(),
credentialKind: v.union(v.literal("oauth"), v.literal("token")),
organizationId: v.id("organizations"),
provider: v.union(v.literal("github"), v.literal("gitea")),
serverUrl: v.string(),
updatedAt: v.number(),
username: v.optional(v.string()),
})
.index("by_organizationId", ["organizationId"])
.index("by_organizationId_and_provider_and_serverUrl", [
"organizationId",
"provider",
"serverUrl",
]),
projects: defineTable({ projects: defineTable({
createdAt: v.number(),
defaultBranch: v.optional(v.string()),
description: v.optional(v.string()),
gitConnectionId: v.optional(v.id("gitConnections")),
name: v.string(),
normalizedSourceUrl: v.string(),
organizationId: v.id("organizations"), organizationId: v.id("organizations"),
repositoryPath: v.string(), name: v.string(),
sourceHost: v.string(), description: v.optional(v.string()),
sourceUrl: 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(), updatedAt: v.number(),
}) })
.index("by_organizationId_and_createdAt", ["organizationId", "createdAt"]) .index("by_organizationId_and_createdAt", ["organizationId", "createdAt"])
@@ -80,8 +35,7 @@ export default defineSchema({
"normalizedSourceUrl", "normalizedSourceUrl",
]), ]),
projectContextDocuments: defineTable({ projectContextDocuments: defineTable({
content: v.string(), projectId: v.id("projects"),
createdAt: v.number(),
kind: v.union( kind: v.union(
v.literal("readme"), v.literal("readme"),
v.literal("agents"), v.literal("agents"),
@@ -90,25 +44,20 @@ export default defineSchema({
v.literal("design"), v.literal("design"),
v.literal("tech") v.literal("tech")
), ),
origin: v.literal("repository"),
path: v.string(), path: v.string(),
projectId: v.id("projects"), content: v.string(),
origin: v.literal("repository"),
sourceUrl: v.string(), sourceUrl: v.string(),
createdAt: v.number(),
updatedAt: v.number(), updatedAt: v.number(),
}).index("by_projectId_and_path", ["projectId", "path"]), }).index("by_projectId_and_path", ["projectId", "path"]),
conversations: defineTable({ conversations: defineTable({
createdAt: v.number(),
organizationId: v.id("organizations"), organizationId: v.id("organizations"),
createdAt: v.number(),
}).index("by_organizationId", ["organizationId"]), }).index("by_organizationId", ["organizationId"]),
conversationTurns: defineTable({ conversationTurns: defineTable({
attemptNumber: v.optional(v.number()),
clientRequestId: v.string(),
completedAt: v.optional(v.number()),
conversationId: v.id("conversations"), conversationId: v.id("conversations"),
createdAt: v.number(), clientRequestId: v.string(),
error: v.optional(v.string()),
leaseExpiresAt: v.optional(v.number()),
leaseOwner: v.optional(v.string()),
status: v.union( status: v.union(
v.literal("queued"), v.literal("queued"),
v.literal("processing"), v.literal("processing"),
@@ -116,86 +65,100 @@ export default defineSchema({
v.literal("failed") v.literal("failed")
), ),
submissionId: v.optional(v.string()), submissionId: v.optional(v.string()),
}) error: 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(), createdAt: v.number(),
ordinal: v.number(), completedAt: v.optional(v.number()),
role: v.union(v.literal("user"), v.literal("assistant")), }).index("by_conversationId_and_clientRequestId", [
"conversationId",
"clientRequestId",
]),
conversationMessages: defineTable({
conversationId: v.id("conversations"),
turnId: v.id("conversationTurns"), 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_conversationId_and_ordinal", ["conversationId", "ordinal"])
.index("by_turnId_and_role", ["turnId", "role"]), .index("by_turnId_and_role", ["turnId", "role"]),
conversationAttachments: defineTable({ conversationAttachments: defineTable({
createdAt: v.number(),
filename: v.optional(v.string()),
messageId: v.id("conversationMessages"), messageId: v.id("conversationMessages"),
mimeType: v.string(),
storageId: v.id("_storage"), storageId: v.id("_storage"),
filename: v.optional(v.string()),
mimeType: v.string(),
createdAt: v.number(),
}).index("by_messageId", ["messageId"]), }).index("by_messageId", ["messageId"]),
signals: defineTable({ signals: defineTable({
conversationId: v.id("conversations"),
createdAt: v.number(),
desiredOutcome: v.string(),
organizationId: v.id("organizations"), organizationId: v.id("organizations"),
processedByAgentInstanceId: v.string(),
processedByAgentName: v.string(),
projectId: v.id("projects"), projectId: v.id("projects"),
conversationId: v.id("conversations"),
sourceKey: v.string(), sourceKey: v.string(),
summary: v.string(),
title: 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_createdAt", ["organizationId", "createdAt"])
.index("by_organization_and_sourceKey", ["organizationId", "sourceKey"]) .index("by_organization_and_sourceKey", ["organizationId", "sourceKey"])
.index("by_project_and_createdAt", ["projectId", "createdAt"]), .index("by_project_and_createdAt", ["projectId", "createdAt"]),
signalConstraints: defineTable({ signalConstraints: defineTable({
ordinal: v.number(),
signalId: v.id("signals"), signalId: v.id("signals"),
ordinal: v.number(),
value: v.string(), value: v.string(),
}).index("by_signalId_and_ordinal", ["signalId", "ordinal"]), }).index("by_signalId_and_ordinal", ["signalId", "ordinal"]),
signalSources: defineTable({ signalSources: defineTable({
signalId: v.id("signals"),
messageId: v.id("conversationMessages"), messageId: v.id("conversationMessages"),
ordinal: v.number(), ordinal: v.number(),
rawTextSnapshot: v.string(), rawTextSnapshot: v.string(),
signalId: v.id("signals"),
sourceCreatedAt: v.number(), sourceCreatedAt: v.number(),
}) })
.index("by_signalId_and_ordinal", ["signalId", "ordinal"]) .index("by_signalId_and_ordinal", ["signalId", "ordinal"])
.index("by_messageId", ["messageId"]), .index("by_messageId", ["messageId"]),
works: defineTable({ 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"), organizationId: v.id("organizations"),
projectId: v.id("projects"), projectId: v.id("projects"),
status: workStatus,
title: v.string(), title: v.string(),
objective: v.string(),
status: v.union(
v.literal("proposed"),
v.literal("defining"),
v.literal("awaiting-definition-approval"),
v.literal("designing"),
v.literal("awaiting-design-approval"),
v.literal("ready"),
v.literal("executing"),
v.literal("needs-input"),
v.literal("blocked"),
v.literal("completed"),
v.literal("failed"),
v.literal("cancelled")
),
definitionVersion: v.optional(v.number()),
definitionApprovalVersion: v.optional(v.number()),
designVersion: v.optional(v.number()),
designApprovalVersion: v.optional(v.number()),
createdAt: v.number(),
updatedAt: v.number(), updatedAt: v.number(),
}) })
.index("by_project_and_createdAt", ["projectId", "createdAt"]) .index("by_project_and_createdAt", ["projectId", "createdAt"])
.index("by_organization_and_createdAt", ["organizationId", "createdAt"]), .index("by_organization_and_createdAt", ["organizationId", "createdAt"]),
signalWorkAttachments: defineTable({ signalWorkAttachments: defineTable({
createdAt: v.number(),
signalId: v.id("signals"), signalId: v.id("signals"),
workId: v.id("works"), workId: v.id("works"),
createdAt: v.number(),
}) })
.index("by_signal", ["signalId"]) .index("by_signal", ["signalId"])
.index("by_work", ["workId"]) .index("by_work", ["workId"])
.index("by_signal_and_work", ["signalId", "workId"]), .index("by_signal_and_work", ["signalId", "workId"]),
workEvents: defineTable({ workEvents: defineTable({
createdAt: v.number(), workId: v.id("works"),
idempotencyKey: v.string(), signalId: v.optional(v.id("signals")),
kind: v.union( kind: v.union(
v.literal("work.proposed"), v.literal("work.proposed"),
v.literal("signal.attached"), v.literal("signal.attached"),
@@ -204,7 +167,6 @@ export default defineSchema({
v.literal("definition.revised"), v.literal("definition.revised"),
v.literal("definition.approved"), v.literal("definition.approved"),
v.literal("definition.invalidated"), v.literal("definition.invalidated"),
v.literal("question.created"),
v.literal("question.answered"), v.literal("question.answered"),
v.literal("question.withdrawn"), v.literal("question.withdrawn"),
v.literal("design.requested"), v.literal("design.requested"),
@@ -212,33 +174,24 @@ export default defineSchema({
v.literal("design.revised"), v.literal("design.revised"),
v.literal("design.approved"), v.literal("design.approved"),
v.literal("design.invalidated"), 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.started"),
v.literal("run.completed"),
v.literal("run.cancelled"), v.literal("run.cancelled"),
v.literal("attempt.claimed"), v.literal("attempt.claimed"),
v.literal("attempt.event"), v.literal("attempt.event"),
v.literal("attempt.completed"), 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")
), ),
payloadJson: v.optional(v.string()),
referenceId: v.optional(v.string()), referenceId: v.optional(v.string()),
signalId: v.optional(v.id("signals")), payloadJson: v.optional(v.string()),
workId: v.id("works"), idempotencyKey: v.string(),
createdAt: v.number(),
}) })
.index("by_work_and_createdAt", ["workId", "createdAt"]) .index("by_work_and_createdAt", ["workId", "createdAt"])
.index("by_work_and_idempotencyKey", ["workId", "idempotencyKey"]), .index("by_work_and_idempotencyKey", ["workId", "idempotencyKey"]),
workDefinitions: defineTable({ workDefinitions: defineTable({
createdAt: v.number(), workId: v.id("works"),
createdBy: v.string(), version: v.number(),
payloadJson: v.string(), payloadJson: v.string(),
risk: v.union(v.literal("low"), v.literal("medium"), v.literal("high")), risk: v.union(v.literal("low"), v.literal("medium"), v.literal("high")),
status: v.union( status: v.union(
@@ -246,55 +199,42 @@ export default defineSchema({
v.literal("current"), v.literal("current"),
v.literal("superseded") v.literal("superseded")
), ),
version: v.number(), createdBy: v.string(),
workId: v.id("works"), createdAt: v.number(),
}) })
.index("by_work_and_version", ["workId", "version"]) .index("by_work_and_version", ["workId", "version"])
.index("by_work_and_status", ["workId", "status"]), .index("by_work_and_status", ["workId", "status"]),
workQuestions: defineTable({ workQuestions: defineTable({
alternativesJson: v.string(), workId: v.id("works"),
answer: v.optional(v.string()),
createdAt: v.number(),
definitionVersion: v.number(), definitionVersion: v.number(),
impact: v.union(v.literal("low"), v.literal("medium"), v.literal("high")),
prompt: v.string(),
questionId: v.string(), questionId: v.string(),
prompt: v.string(),
impact: v.union(v.literal("low"), v.literal("medium"), v.literal("high")),
recommendation: v.optional(v.string()), recommendation: v.optional(v.string()),
alternativesJson: v.string(),
status: v.union( status: v.union(
v.literal("open"), v.literal("open"),
v.literal("answered"), v.literal("answered"),
v.literal("withdrawn") v.literal("withdrawn")
), ),
workId: v.id("works"), answer: v.optional(v.string()),
}) createdAt: v.number(),
.index("by_workId_and_definitionVersion", ["workId", "definitionVersion"]) }).index("by_work_and_definitionVersion", ["workId", "definitionVersion"]),
.index("by_workId_and_definitionVersion_and_questionId", [
"workId",
"definitionVersion",
"questionId",
]),
workApprovals: defineTable({ workApprovals: defineTable({
approvedAt: v.number(), workId: v.id("works"),
approvedBy: v.string(), kind: v.union(v.literal("definition"), v.literal("design")),
definitionVersion: v.number(), definitionVersion: v.number(),
designVersion: v.optional(v.number()), designVersion: v.optional(v.number()),
kind: v.union(v.literal("definition"), v.literal("design")), approvedBy: v.string(),
approvedAt: v.number(),
status: v.union(v.literal("active"), v.literal("invalidated")), status: v.union(v.literal("active"), v.literal("invalidated")),
workId: v.id("works"), }).index("by_work_and_kind", ["workId", "kind"]),
})
.index("by_workId_and_kind", ["workId", "kind"])
.index("by_workId_and_kind_and_definitionVersion_and_designVersion", [
"workId",
"kind",
"definitionVersion",
"designVersion",
]),
designPackets: defineTable({ designPackets: defineTable({
createdAt: v.number(), workId: v.id("works"),
createdBy: v.string(), version: v.number(),
definitionVersion: v.number(), definitionVersion: v.number(),
payloadJson: v.string(), payloadJson: v.string(),
status: v.union( status: v.union(
@@ -302,18 +242,19 @@ export default defineSchema({
v.literal("current"), v.literal("current"),
v.literal("superseded") v.literal("superseded")
), ),
version: v.number(), createdBy: v.string(),
workId: v.id("works"), createdAt: v.number(),
}).index("by_work_and_version", ["workId", "version"]), }).index("by_work_and_version", ["workId", "version"]),
workSlices: defineTable({ workSlices: defineTable({
createdAt: v.optional(v.number()), workId: v.id("works"),
designVersion: v.number(), designVersion: v.number(),
sliceId: v.string(),
ordinal: v.number(),
title: v.string(),
objective: v.string(), objective: v.string(),
observableBehavior: v.string(), observableBehavior: v.string(),
ordinal: v.number(),
payloadJson: v.string(), payloadJson: v.string(),
sliceId: v.string(),
status: v.union( status: v.union(
v.literal("planned"), v.literal("planned"),
v.literal("ready"), v.literal("ready"),
@@ -321,27 +262,16 @@ export default defineSchema({
v.literal("completed"), v.literal("completed"),
v.literal("blocked") v.literal("blocked")
), ),
title: v.string(), }).index("by_work_and_designVersion", ["workId", "designVersion"]),
workId: v.id("works"),
})
.index("by_workId_and_designVersion", ["workId", "designVersion"])
.index("by_workId_and_designVersion_and_sliceId", [
"workId",
"designVersion",
"sliceId",
]),
workRuns: defineTable({ workRuns: defineTable({
baseRevision: v.optional(v.string()), workId: v.id("works"),
candidateRevision: v.optional(v.string()), sliceId: v.optional(v.string()),
createdAt: v.number(), status: v.union(
designVersion: v.optional(v.number()), v.literal("ready"),
endedAt: v.optional(v.number()), v.literal("running"),
kitId: v.string(), v.literal("terminal"),
kitVersion: v.string(), v.literal("cancelled")
environmentId: v.optional(v.string()),
executionKind: v.optional(
v.union(v.literal("simulated"), v.literal("real"))
), ),
scenario: v.union( scenario: v.union(
v.literal("success"), v.literal("success"),
@@ -350,142 +280,42 @@ export default defineSchema({
v.literal("permanent-failure"), v.literal("permanent-failure"),
v.literal("cancelled") v.literal("cancelled")
), ),
sliceId: v.optional(v.string()), kitId: v.string(),
sliceRowId: v.optional(v.id("workSlices")), kitVersion: v.string(),
createdAt: v.number(),
startedAt: v.optional(v.number()), startedAt: v.optional(v.number()),
status: v.union( endedAt: v.optional(v.number()),
v.literal("ready"), terminalClassification: v.optional(v.string()),
v.literal("running"),
v.literal("terminal"),
v.literal("cancelled")
),
terminalClassification: v.optional(attemptClassification),
terminalSummary: v.optional(v.string()), terminalSummary: v.optional(v.string()),
workflowId: v.optional(v.string()),
workId: v.id("works"),
}).index("by_work_and_createdAt", ["workId", "createdAt"]), }).index("by_work_and_createdAt", ["workId", "createdAt"]),
workAttempts: defineTable({ workAttempts: defineTable({
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"), runId: v.id("workRuns"),
startedAt: v.optional(v.number()), workId: v.id("works"),
number: v.number(),
status: v.union( status: v.union(
v.literal("queued"), v.literal("queued"),
v.literal("claimed"), v.literal("claimed"),
v.literal("running"), v.literal("running"),
v.literal("terminal") 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()), summary: v.optional(v.string()),
workId: v.id("works"), }).index("by_run_and_number", ["runId", "number"]),
workspaceKey: v.optional(v.string()),
})
.index("by_runId_and_number", ["runId", "number"])
.index("by_status_and_leaseExpiresAt", ["status", "leaseExpiresAt"]),
workAttemptEvents: defineTable({ workAttemptEvents: defineTable({
attemptId: v.id("workAttempts"), attemptId: v.id("workAttempts"),
sequence: v.number(),
kind: v.string(), kind: v.string(),
message: v.string(), message: v.string(),
metadataJson: v.string(), metadataJson: v.string(),
occurredAt: v.number(), occurredAt: v.number(),
sequence: v.number(),
}).index("by_attempt_and_sequence", ["attemptId", "sequence"]), }).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). // Flue persistence stores (schema/format version 4).
// ----------------------------------------------------------------- // -----------------------------------------------------------------
@@ -552,38 +382,38 @@ export default defineSchema({
// Durable evidence that a submission attempt started and has not yet // Durable evidence that a submission attempt started and has not yet
// settled. Append-once by (submissionId, attemptId). // settled. Append-once by (submissionId, attemptId).
flueAttemptMarkers: defineTable({ flueAttemptMarkers: defineTable({
submissionId: v.string(),
attemptId: v.string(), attemptId: v.string(),
createdAt: v.number(), createdAt: v.number(),
submissionId: v.string(),
}) })
.index("by_submissionId_and_attemptId", ["submissionId", "attemptId"]) .index("by_submissionId_and_attemptId", ["submissionId", "attemptId"])
.index("by_submissionId", ["submissionId"]), .index("by_submissionId", ["submissionId"]),
// Conversation-stream metadata: one row per stream path. // Conversation-stream metadata: one row per stream path.
flueConversationStreams: defineTable({ flueConversationStreams: defineTable({
closed: v.boolean(), path: v.string(),
createdAt: v.number(),
identityJson: v.string(), identityJson: v.string(),
incarnation: v.string(), incarnation: v.string(),
nextOffset: v.number(),
nextProducerSequence: v.number(),
path: v.string(),
producerEpoch: v.number(),
producerId: v.optional(v.string()), producerId: v.optional(v.string()),
producerEpoch: v.number(),
nextProducerSequence: v.number(),
nextOffset: v.number(),
closed: v.boolean(),
createdAt: v.number(),
}).index("by_path", ["path"]), }).index("by_path", ["path"]),
// Conversation-stream batches: one row per appended batch. Offset is // Conversation-stream batches: one row per appended batch. Offset is
// 0-based; `seq` is the row's position in the stream. // 0-based; `seq` is the row's position in the stream.
flueConversationBatches: defineTable({ flueConversationBatches: defineTable({
appendedAt: v.number(),
attemptId: v.optional(v.string()),
path: v.string(), path: v.string(),
producerEpoch: v.number(), seq: v.number(),
producerId: v.string(), producerId: v.string(),
producerEpoch: v.number(),
producerSequence: v.number(), producerSequence: v.number(),
recordsJson: v.string(), recordsJson: v.string(),
seq: v.number(),
submissionId: v.optional(v.string()), submissionId: v.optional(v.string()),
attemptId: v.optional(v.string()),
appendedAt: v.number(),
}) })
.index("by_path_and_seq", ["path", "seq"]) .index("by_path_and_seq", ["path", "seq"])
.index("by_path_producer_epoch_producerSequence", [ .index("by_path_producer_epoch_producerSequence", [
@@ -595,41 +425,41 @@ export default defineSchema({
// Event-stream metadata: one row per stream path. // Event-stream metadata: one row per stream path.
flueEventStreams: defineTable({ flueEventStreams: defineTable({
path: v.string(),
nextSeq: v.number(),
closed: v.boolean(), closed: v.boolean(),
createdAt: v.number(), createdAt: v.number(),
nextSeq: v.number(),
path: v.string(),
}).index("by_path", ["path"]), }).index("by_path", ["path"]),
// Event-stream entries: one row per appended event. `onceKey` carries the // Event-stream entries: one row per appended event. `onceKey` carries the
// idempotency key for `appendEventOnce` and is null for plain appends. // idempotency key for `appendEventOnce` and is null for plain appends.
flueEventEntries: defineTable({ flueEventEntries: defineTable({
appendedAt: v.number(),
dataJson: v.string(),
onceKey: v.optional(v.string()),
path: v.string(), path: v.string(),
seq: v.number(), 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_seq", ["path", "seq"])
.index("by_path_and_onceKey", ["path", "onceKey"]), .index("by_path_and_onceKey", ["path", "onceKey"]),
// Workflow run records. // Workflow run records.
flueRuns: defineTable({ 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(), runId: v.string(),
startedAt: v.string(), workflowName: v.string(),
status: v.union( status: v.union(
v.literal("active"), v.literal("active"),
v.literal("completed"), v.literal("completed"),
v.literal("errored") v.literal("errored")
), ),
startedAt: v.string(),
inputJson: v.optional(v.string()),
traceCarrierJson: v.optional(v.string()), traceCarrierJson: v.optional(v.string()),
workflowName: 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()),
}) })
.index("by_runId", ["runId"]) .index("by_runId", ["runId"])
.index("by_startedAt_and_runId", ["startedAt", "runId"]) .index("by_startedAt_and_runId", ["startedAt", "runId"])
@@ -649,15 +479,15 @@ export default defineSchema({
// Immutable attachment bytes. Identity is (streamPath, attachmentId); reads // Immutable attachment bytes. Identity is (streamPath, attachmentId); reads
// are additionally scoped by conversationId. // are additionally scoped by conversationId.
flueAttachments: defineTable({ flueAttachments: defineTable({
attachmentId: v.string(), streamPath: v.string(),
bytes: v.bytes(),
conversationId: v.string(), conversationId: v.string(),
createdAt: v.number(), attachmentId: v.string(),
digest: v.string(),
filename: v.optional(v.string()),
mimeType: v.string(), mimeType: v.string(),
size: v.number(), size: v.number(),
streamPath: v.string(), digest: v.string(),
filename: v.optional(v.string()),
bytes: v.bytes(),
createdAt: v.number(),
}) })
.index("by_streamPath", ["streamPath"]) .index("by_streamPath", ["streamPath"])
.index("by_streamPath_and_attachmentId", ["streamPath", "attachmentId"]) .index("by_streamPath_and_attachmentId", ["streamPath", "attachmentId"])

View File

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

View File

@@ -1,108 +0,0 @@
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",
]);
});
});

View File

@@ -1,370 +0,0 @@
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 };
},
});

View File

@@ -1,5 +1,5 @@
import { convexTest } from "convex-test"; import { convexTest } from "convex-test";
import { anyApi, makeFunctionReference } from "convex/server"; import { anyApi } from "convex/server";
import { describe, expect, test } from "vitest"; import { describe, expect, test } from "vitest";
import type { Id } from "./_generated/dataModel"; import type { Id } from "./_generated/dataModel";
@@ -15,11 +15,6 @@ const modules = {
}; };
const identity = { tokenIdentifier: "https://convex.test|exec-user" }; const identity = { tokenIdentifier: "https://convex.test|exec-user" };
const api = anyApi; const api = anyApi;
const claimAttemptRef = makeFunctionReference<
"mutation",
{ attemptId: Id<"workAttempts">; leaseMs: number; owner: string },
{ number: number } | null
>("workExecution:claimAttempt");
const makeTest = () => convexTest({ modules, schema }); const makeTest = () => convexTest({ modules, schema });
type TestT = ReturnType<typeof makeTest>; type TestT = ReturnType<typeof makeTest>;
@@ -31,7 +26,7 @@ type Scenario =
| "permanent-failure" | "permanent-failure"
| "cancelled"; | "cancelled";
const seedReadyWork = async (options: { secondSlice?: boolean } = {}) => { const seedReadyWork = async () => {
const t = convexTest({ modules, schema }); const t = convexTest({ modules, schema });
const { workId } = await t.withIdentity(identity).run(async (ctx) => { const { workId } = await t.withIdentity(identity).run(async (ctx) => {
const organizationId = await ctx.db.insert("organizations", { const organizationId = await ctx.db.insert("organizations", {
@@ -79,7 +74,6 @@ const seedReadyWork = async (options: { secondSlice?: boolean } = {}) => {
workId: id, workId: id,
}); });
await ctx.db.insert("workSlices", { await ctx.db.insert("workSlices", {
createdAt: 1,
designVersion: 1, designVersion: 1,
objective: "execute the slice", objective: "execute the slice",
observableBehavior: "terminal run", observableBehavior: "terminal run",
@@ -90,20 +84,6 @@ const seedReadyWork = async (options: { secondSlice?: boolean } = {}) => {
title: "first", title: "first",
workId: id, 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 { workId: id };
}); });
return { t, workId }; return { t, workId };
@@ -123,14 +103,12 @@ const attemptNumber = async (
const attempt = await t.run(async (ctx) => const attempt = await t.run(async (ctx) =>
ctx.db ctx.db
.query("workAttempts") .query("workAttempts")
.withIndex("by_runId_and_number", (q) => .withIndex("by_run_and_number", (q) =>
q.eq("runId", runId).eq("number", number) q.eq("runId", runId).eq("number", number)
) )
.unique() .unique()
); );
if (!attempt) { if (!attempt) throw new Error(`attempt ${number} not found`);
throw new Error(`attempt ${number} not found`);
}
return attempt._id; return attempt._id;
}; };
@@ -138,21 +116,15 @@ const snapshot = async (t: TestT, workId: Id<"works">, runId: Id<"workRuns">) =>
t.run(async (ctx) => ({ t.run(async (ctx) => ({
attempts: await ctx.db attempts: await ctx.db
.query("workAttempts") .query("workAttempts")
.withIndex("by_runId_and_number", (q) => q.eq("runId", runId)) .withIndex("by_run_and_number", (q) => q.eq("runId", runId))
.collect(), .collect(),
run: await ctx.db.get(runId), run: await ctx.db.get(runId),
slice: await ctx.db slice: await ctx.db
.query("workSlices") .query("workSlices")
.withIndex("by_workId_and_designVersion", (q) => .withIndex("by_work_and_designVersion", (q) =>
q.eq("workId", workId).eq("designVersion", 1) q.eq("workId", workId).eq("designVersion", 1)
) )
.first(), .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), work: await ctx.db.get(workId),
})); }));
@@ -172,77 +144,6 @@ describe("simulated execution resolver", () => {
expect(state.run?.terminalClassification).toBe("Succeeded"); expect(state.run?.terminalClassification).toBe("Succeeded");
expect(state.attempts).toHaveLength(1); expect(state.attempts).toHaveLength(1);
expect(state.slice?.status).toBe("completed"); 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 () => { test("transient failure retries within the kit budget and then succeeds", async () => {

View File

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

View File

@@ -1,80 +0,0 @@
"use node";
import { env } from "@code/env/convex";
import { decodeWorkAttemptExecutionResult } from "@code/primitives/execution-runtime";
import { ConvexError, v } from "convex/values";
import { Effect } from "effect";
import { internal } from "./_generated/api";
import { internalAction } from "./_generated/server";
import { decryptCredential } from "./gitConnections";
const backendUrl = () => env.AGENT_BACKEND_URL ?? env.FLUE_URL;
export const executeAttempt = internalAction({
args: { attemptId: v.id("workAttempts") },
handler: async (ctx, args) => {
const running = await ctx.runMutation(
internal.workExecutionWorkflow.markAttemptRunning,
args
);
if (!running) {
throw new ConvexError("Attempt is no longer runnable");
}
const context = await ctx.runQuery(
internal.workExecutionWorkflow.executionContext,
args
);
const credential = await decryptCredential(
context.connection.credentialCiphertext,
context.connection.credentialIv
);
const response = await fetch(
`${backendUrl()}/internal/work-attempts/execute`,
{
body: JSON.stringify({
attemptId: String(context.attempt._id),
auth: {
credential,
provider: context.connection.provider,
serverUrl: context.connection.serverUrl,
username: context.connection.username,
},
baseBranch: context.project.defaultBranch ?? "main",
prompt: context.prompt,
repositoryUrl: context.project.sourceUrl,
runId: String(context.run._id),
workId: String(context.work._id),
workspaceKey: context.attempt.workspaceKey,
}),
headers: {
authorization: `Bearer ${env.FLUE_DB_TOKEN}`,
"content-type": "application/json",
},
method: "POST",
}
);
const payload = (await response.json()) as unknown;
if (!response.ok) {
throw new ConvexError(
typeof payload === "object" && payload && "error" in payload
? String(payload.error)
: `Agent backend returned ${response.status}`
);
}
return await Effect.runPromise(decodeWorkAttemptExecutionResult(payload));
},
});
export const cancelAttempt = internalAction({
args: { workspaceKey: v.string() },
handler: async (_ctx, args) => {
await fetch(
`${backendUrl()}/internal/work-attempts/${encodeURIComponent(args.workspaceKey)}/cancel`,
{
headers: { authorization: `Bearer ${env.FLUE_DB_TOKEN}` },
method: "POST",
}
);
},
});

View File

@@ -1,111 +0,0 @@
import { convexTest } from "convex-test";
import { anyApi } from "convex/server";
import { describe, expect, test } from "vitest";
import schema from "./schema";
const modules = import.meta.glob("./**/*.ts");
const api = anyApi;
describe("real work execution persistence", () => {
test("records revisions, activity, diff, and terminal state atomically", async () => {
const t = convexTest({ modules, schema });
const seeded = 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,
});
const workId = await ctx.db.insert("works", {
createdAt: 1,
objective: "Implement a real slice",
organizationId,
projectId,
status: "executing",
title: "Real execution",
updatedAt: 1,
});
const sliceRowId = await ctx.db.insert("workSlices", {
designVersion: 1,
objective: "Change the repo",
observableBehavior: "A diff exists",
ordinal: 0,
payloadJson: "{}",
sliceId: "slice-1",
status: "running",
title: "Implementation",
workId,
});
const runId = await ctx.db.insert("workRuns", {
createdAt: 1,
designVersion: 1,
executionKind: "real",
kitId: "coding-v0",
kitVersion: "1",
scenario: "success",
sliceId: "slice-1",
sliceRowId,
status: "running",
workId,
});
const attemptId = await ctx.db.insert("workAttempts", {
number: 1,
runId,
status: "running",
workId,
workspaceKey: "workspace-1",
});
return { attemptId, runId, workId };
});
await t.mutation(api.workExecutionWorkflow.completeAttempt, {
attemptId: seeded.attemptId,
result: {
baseRevision: "base123",
candidateRevision: "candidate456",
changedFiles: ["src/index.ts"],
diff: "+export const ready = true;",
environmentId: "workspace-1",
events: [
{
kind: "runtime.completed",
message: "completed",
metadata: {},
occurredAt: 2,
sequence: 0,
},
],
summary: "Changed one file",
},
});
const state = await t.run(async (ctx) => ({
artifacts: await ctx.db.query("workArtifacts").collect(),
attempt: await ctx.db.get(seeded.attemptId),
run: await ctx.db.get(seeded.runId),
work: await ctx.db.get(seeded.workId),
}));
expect(state.attempt?.classification).toBe("Succeeded");
expect(state.run).toMatchObject({
baseRevision: "base123",
candidateRevision: "candidate456",
terminalClassification: "Succeeded",
});
expect(state.work?.status).toBe("completed");
expect(state.artifacts[0]).toMatchObject({
kind: "diff",
sourceRevision: "candidate456",
});
});
});

View File

@@ -1,403 +0,0 @@
import { defaultCodingKitV0 } from "@code/primitives/resolver";
import { WorkflowManager } from "@convex-dev/workflow";
import type { WorkflowId } from "@convex-dev/workflow";
import { ConvexError, v } from "convex/values";
import { components, internal } from "./_generated/api";
import type { Doc, Id } from "./_generated/dataModel";
import { internalMutation, internalQuery, mutation } from "./_generated/server";
import type { MutationCtx } from "./_generated/server";
import { requireProjectMember } from "./authz";
export const workflow = new WorkflowManager(components.workflow);
const resolveReadySlice = async (
ctx: MutationCtx,
work: Doc<"works">,
sliceId?: string
) => {
if (work.designVersion === 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", work.designVersion!)
)
.collect();
const slice = sliceId
? slices.find((candidate) => candidate.sliceId === sliceId)
: slices
.sort((left, right) => left.ordinal - right.ordinal)
.find((candidate) => candidate.status === "ready");
if (!slice || slice.status !== "ready") {
throw new ConvexError("Only the next ready slice can be executed");
}
return slice;
};
export const execute = workflow
.define({ args: { attemptId: v.id("workAttempts") } })
.handler(async (step, args): Promise<void> => {
try {
const result = await step.runAction(
internal.workExecutionAgent.executeAttempt,
args,
{ retry: true }
);
await step.runMutation(internal.workExecutionWorkflow.completeAttempt, {
attemptId: args.attemptId,
result: {
...result,
changedFiles: [...result.changedFiles],
events: result.events.map((item) => ({
...item,
metadata: { ...item.metadata },
})),
},
});
} catch (error) {
await step.runMutation(internal.workExecutionWorkflow.failAttempt, {
attemptId: args.attemptId,
summary: error instanceof Error ? error.message : "Execution failed",
});
}
});
export const startExecution = mutation({
args: {
sliceId: v.optional(v.string()),
workId: v.id("works"),
},
handler: async (
ctx,
args
): Promise<{
attemptId: Id<"workAttempts">;
runId: Id<"workRuns">;
workflowId: WorkflowId;
}> => {
const work = await ctx.db.get(args.workId);
if (!work) {
throw new ConvexError("Work not found");
}
await requireProjectMember(ctx, work.projectId);
if (work.status !== "ready") {
throw new ConvexError("Work must be Ready before execution");
}
if (
work.definitionApprovalVersion !== work.definitionVersion ||
work.designApprovalVersion !== work.designVersion
) {
throw new ConvexError("Execution requires exact approved versions");
}
const project = await ctx.db.get(work.projectId);
if (!project?.gitConnectionId) {
throw new ConvexError("Connect Git credentials to this project first");
}
const slice = await resolveReadySlice(ctx, work, args.sliceId);
const createdAt = Date.now();
const runId = await ctx.db.insert("workRuns", {
createdAt,
designVersion: slice.designVersion,
executionKind: "real",
kitId: defaultCodingKitV0.id,
kitVersion: defaultCodingKitV0.version,
scenario: "success",
sliceId: slice.sliceId,
sliceRowId: slice._id,
status: "running",
workId: work._id,
});
const workspaceKey = `work-${work._id}-run-${runId}`;
const attemptId = await ctx.db.insert("workAttempts", {
number: 1,
runId,
status: "queued",
workId: work._id,
workspaceKey,
});
const workflowId: WorkflowId = await workflow.start(
ctx,
internal.workExecutionWorkflow.execute,
{ attemptId }
);
await ctx.db.patch(runId, { startedAt: createdAt, workflowId });
await ctx.db.patch(slice._id, { status: "running" });
await ctx.db.patch(work._id, { status: "executing", updatedAt: createdAt });
await ctx.db.insert("workEvents", {
createdAt,
idempotencyKey: `real-run-started:${runId}`,
kind: "run.started",
referenceId: String(runId),
workId: work._id,
});
return { attemptId, runId, workflowId };
},
});
export const executionContext = internalQuery({
args: { attemptId: v.id("workAttempts") },
handler: async (ctx, args) => {
const attempt = await ctx.db.get(args.attemptId);
if (!attempt) {
throw new ConvexError("Attempt not found");
}
const run = await ctx.db.get(attempt.runId);
const work = await ctx.db.get(attempt.workId);
if (!run || !work || !run.sliceRowId) {
throw new ConvexError("Execution records are incomplete");
}
const [slice, project] = await Promise.all([
ctx.db.get(run.sliceRowId),
ctx.db.get(work.projectId),
]);
if (!slice || !project?.gitConnectionId) {
throw new ConvexError("Project execution configuration is incomplete");
}
const connection = await ctx.db.get(project.gitConnectionId);
if (!connection) {
throw new ConvexError("Git connection not found");
}
return {
attempt,
connection,
project,
prompt: [
`Implement this approved Zopu slice: ${slice.title}`,
`Objective: ${slice.objective}`,
`Observable behavior: ${slice.observableBehavior}`,
"Inspect the repository instructions first. Make focused changes and run relevant checks.",
].join("\n\n"),
run,
work,
};
},
});
export const markAttemptRunning = internalMutation({
args: { attemptId: v.id("workAttempts") },
handler: async (ctx, args) => {
const attempt = await ctx.db.get(args.attemptId);
if (!attempt || attempt.status !== "queued") {
return false;
}
await ctx.db.patch(attempt._id, {
startedAt: Date.now(),
status: "running",
});
return true;
},
});
const settleSliceAndWork = async (
ctx: MutationCtx,
run: Doc<"workRuns">,
succeeded: boolean
) => {
if (!run.sliceRowId) {
return;
}
const slice = await ctx.db.get(run.sliceRowId);
if (!slice) {
return;
}
await ctx.db.patch(slice._id, { status: succeeded ? "completed" : "ready" });
const work = await ctx.db.get(run.workId);
if (!work) {
return;
}
let status: Doc<"works">["status"] = succeeded ? "completed" : "failed";
if (succeeded) {
const slices = await ctx.db
.query("workSlices")
.withIndex("by_workId_and_designVersion", (q) =>
q.eq("workId", work._id).eq("designVersion", slice.designVersion)
)
.collect();
const next = slices
.sort((left, right) => left.ordinal - right.ordinal)
.find((candidate) => candidate.status === "planned");
if (next) {
await ctx.db.patch(next._id, { status: "ready" });
status = "ready";
}
}
await ctx.db.patch(work._id, { status, updatedAt: Date.now() });
};
export const completeAttempt = internalMutation({
args: {
attemptId: v.id("workAttempts"),
result: v.object({
baseRevision: v.string(),
candidateRevision: v.string(),
changedFiles: v.array(v.string()),
diff: v.string(),
environmentId: v.string(),
events: v.array(
v.object({
kind: v.string(),
message: v.string(),
metadata: v.record(v.string(), v.string()),
occurredAt: v.number(),
sequence: v.number(),
})
),
summary: v.string(),
}),
},
handler: async (ctx, args) => {
const attempt = await ctx.db.get(args.attemptId);
if (!attempt || attempt.status === "terminal") {
return;
}
const run = await ctx.db.get(attempt.runId);
const work = await ctx.db.get(attempt.workId);
if (!run || !work) {
throw new ConvexError("Execution records not found");
}
for (const item of args.result.events) {
await ctx.db.insert("workAttemptEvents", {
attemptId: attempt._id,
kind: item.kind,
message: item.message,
metadataJson: JSON.stringify(item.metadata),
occurredAt: item.occurredAt,
sequence: item.sequence,
});
}
const endedAt = Date.now();
await ctx.db.patch(attempt._id, {
classification: "Succeeded",
endedAt,
status: "terminal",
summary: args.result.summary,
});
await ctx.db.patch(run._id, {
baseRevision: args.result.baseRevision,
candidateRevision: args.result.candidateRevision,
endedAt,
environmentId: args.result.environmentId,
status: "terminal",
terminalClassification: "Succeeded",
terminalSummary: args.result.summary,
});
await ctx.db.insert("workArtifacts", {
attemptId: attempt._id,
createdAt: endedAt,
designVersion: run.designVersion,
environmentId: args.result.environmentId,
idempotencyKey: `real-diff:${attempt._id}`,
kind: "diff",
metadataJson: JSON.stringify({ changedFiles: args.result.changedFiles }),
organizationId: work.organizationId,
producer: "agentos-codex",
projectId: work.projectId,
provenanceJson: JSON.stringify({
baseRevision: args.result.baseRevision,
}),
runId: run._id,
sliceId: run.sliceId,
sourceRevision: args.result.candidateRevision,
title: "Implementation diff",
verificationStatus: "unverified",
workId: work._id,
...(args.result.diff.length > 0
? {
uri: `data:text/plain;charset=utf-8,${encodeURIComponent(args.result.diff.slice(0, 50_000))}`,
}
: {}),
});
await settleSliceAndWork(ctx, run, true);
await ctx.db.insert("workEvents", {
createdAt: endedAt,
idempotencyKey: `real-run-completed:${run._id}`,
kind: "run.completed",
payloadJson: JSON.stringify({ classification: "Succeeded" }),
referenceId: String(run._id),
workId: work._id,
});
},
});
export const failAttempt = internalMutation({
args: { attemptId: v.id("workAttempts"), summary: v.string() },
handler: async (ctx, args) => {
const attempt = await ctx.db.get(args.attemptId);
if (!attempt || attempt.status === "terminal") {
return;
}
const run = await ctx.db.get(attempt.runId);
if (!run) {
return;
}
const endedAt = Date.now();
await ctx.db.patch(attempt._id, {
classification: "PermanentFailure",
endedAt,
status: "terminal",
summary: args.summary,
});
await ctx.db.patch(run._id, {
endedAt,
status: "terminal",
terminalClassification: "PermanentFailure",
terminalSummary: args.summary,
});
await settleSliceAndWork(ctx, run, false);
},
});
export const cancelExecution = 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");
}
await requireProjectMember(ctx, (await ctx.db.get(run.workId))!.projectId);
if (run.status !== "running") {
return { cancelled: false };
}
if (run.workflowId) {
await workflow.cancel(ctx, run.workflowId as WorkflowId);
}
const attempts = await ctx.db
.query("workAttempts")
.withIndex("by_runId_and_number", (q) => q.eq("runId", run._id))
.collect();
const attempt = attempts.find(
(candidate) => candidate.status !== "terminal"
);
if (attempt?.workspaceKey) {
await ctx.scheduler.runAfter(
0,
internal.workExecutionAgent.cancelAttempt,
{
workspaceKey: attempt.workspaceKey,
}
);
await ctx.db.patch(attempt._id, {
classification: "Cancelled",
endedAt: Date.now(),
status: "terminal",
summary: "Execution cancelled",
});
}
await ctx.db.patch(run._id, {
endedAt: Date.now(),
status: "cancelled",
terminalClassification: "Cancelled",
terminalSummary: "Execution cancelled",
});
if (run.sliceRowId) {
await ctx.db.patch(run.sliceRowId, { status: "ready" });
}
const work = await ctx.db.get(run.workId);
if (work) {
await ctx.db.patch(work._id, { status: "ready", updatedAt: Date.now() });
}
return { cancelled: true };
},
});

View File

@@ -30,10 +30,10 @@ describe("Work planning and execution commands", () => {
name: "Personal", name: "Personal",
}); });
await ctx.db.insert("organizationMembers", { await ctx.db.insert("organizationMembers", {
createdAt: 1,
organizationId, organizationId,
role: "owner",
userId: identity.tokenIdentifier, userId: identity.tokenIdentifier,
role: "owner",
createdAt: 1,
}); });
const projectId = await ctx.db.insert("projects", { const projectId = await ctx.db.insert("projects", {
createdAt: 1, createdAt: 1,
@@ -46,12 +46,12 @@ describe("Work planning and execution commands", () => {
updatedAt: 1, updatedAt: 1,
}); });
const workId = await ctx.db.insert("works", { const workId = await ctx.db.insert("works", {
createdAt: 1,
objective: "Prove the flow",
organizationId, organizationId,
projectId, projectId,
status: "proposed",
title: "Executable work", title: "Executable work",
objective: "Prove the flow",
status: "proposed",
createdAt: 1,
updatedAt: 1, updatedAt: 1,
}); });
return { projectId, workId }; return { projectId, workId };
@@ -63,67 +63,67 @@ describe("Work planning and execution commands", () => {
const definition = await t const definition = await t
.withIdentity(identity) .withIdentity(identity)
.mutation(api.workPlanning.saveDefinitionProposal, { .mutation(api.workPlanning.saveDefinitionProposal, {
workId: fixture.workId,
payloadJson: JSON.stringify({ payloadJson: JSON.stringify({
acceptanceCriteria: ["terminal"], problem: "Need a proof",
affectedUsers: ["user"],
assumptions: [],
constraints: [],
desiredOutcome: "A terminal simulation", desiredOutcome: "A terminal simulation",
affectedUsers: ["user"],
inScope: ["fake"], inScope: ["fake"],
outOfScope: [], outOfScope: [],
problem: "Need a proof", acceptanceCriteria: ["terminal"],
constraints: [],
assumptions: [],
questions: [], questions: [],
requiredArtifacts: ["events"],
risk: "low", risk: "low",
requiredArtifacts: ["events"],
}), }),
workId: fixture.workId,
}); });
await t await t
.withIdentity(identity) .withIdentity(identity)
.mutation(api.workPlanning.approveDefinition, { .mutation(api.workPlanning.approveDefinition, {
version: definition.version,
workId: fixture.workId, workId: fixture.workId,
version: definition.version,
}); });
const design = await t const design = await t
.withIdentity(identity) .withIdentity(identity)
.mutation(api.workPlanning.saveDesignProposal, { .mutation(api.workPlanning.saveDesignProposal, {
workId: fixture.workId,
payloadJson: JSON.stringify({ payloadJson: JSON.stringify({
architectureSummary: "fake",
callFlowDelta: [],
concerns: [],
evidenceRequirements: ["event"],
fileTreeDelta: [],
impactMap: { files: [], modules: [], risks: [], summary: "small" }, impactMap: { files: [], modules: [], risks: [], summary: "small" },
invariants: ["terminal"], architectureSummary: "fake",
fileTreeDelta: [],
callFlowDelta: [],
keyInterfaces: [], keyInterfaces: [],
invariants: ["terminal"],
concerns: [],
slices: [ slices: [
{ {
codeBoundaries: ["runtime"],
dependsOn: [],
evidenceRequirements: ["event"],
id: "slice-1", id: "slice-1",
title: "fake",
objective: "fake", objective: "fake",
observableBehavior: "event", observableBehavior: "event",
reviewRequired: false, codeBoundaries: ["runtime"],
title: "fake", evidenceRequirements: ["event"],
verification: ["assert"], verification: ["assert"],
reviewRequired: false,
dependsOn: [],
}, },
], ],
evidenceRequirements: ["event"],
tradeoffs: [], tradeoffs: [],
}), }),
workId: fixture.workId,
}); });
await t.withIdentity(identity).mutation(api.workPlanning.approveDesign, { await t.withIdentity(identity).mutation(api.workPlanning.approveDesign, {
workId: fixture.workId,
definitionVersion: definition.version, definitionVersion: definition.version,
designVersion: design.version, designVersion: design.version,
workId: fixture.workId,
}); });
const started = await t const started = await t
.withIdentity(identity) .withIdentity(identity)
.mutation(api.workExecution.startSimulatedExecution, { .mutation(api.workExecution.startSimulatedExecution, {
workId: fixture.workId,
scenario: "success", scenario: "success",
sliceId: "slice-1", sliceId: "slice-1",
workId: fixture.workId,
}); });
expect(started.runId).toBeTruthy(); expect(started.runId).toBeTruthy();
const state = await t.run(async (ctx) => ctx.db.get(fixture.workId)); const state = await t.run(async (ctx) => ctx.db.get(fixture.workId));
@@ -137,7 +137,7 @@ describe("Work planning and execution commands", () => {
const completed = await t.run(async (ctx) => ({ const completed = await t.run(async (ctx) => ({
attempt: await ctx.db attempt: await ctx.db
.query("workAttempts") .query("workAttempts")
.withIndex("by_runId_and_number", (q) => q.eq("runId", started.runId)) .withIndex("by_run_and_number", (q) => q.eq("runId", started.runId))
.unique(), .unique(),
events: await ctx.db.query("workAttemptEvents").collect(), events: await ctx.db.query("workAttemptEvents").collect(),
run: await ctx.db.get(started.runId as Id<"workRuns">), run: await ctx.db.get(started.runId as Id<"workRuns">),
@@ -164,154 +164,7 @@ const validDefinitionPayload = () => ({
risk: "low", 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", () => { 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 () => { test("a Definition cannot be revised while a Run is executing", async () => {
const t = convexTest({ modules, schema }); const t = convexTest({ modules, schema });
const { workId } = await t.withIdentity(identity).run(async (ctx) => { const { workId } = await t.withIdentity(identity).run(async (ctx) => {

View File

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

View File

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

View File

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

View File

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

View File

@@ -14,8 +14,6 @@ const agentEnvSchema = z.object({
GITEA_TOKEN: z.string().min(1).optional(), GITEA_TOKEN: z.string().min(1).optional(),
GITEA_URL: z.url().default("https://git.openputer.com"), GITEA_URL: z.url().default("https://git.openputer.com"),
OPENROUTER_API_KEY: z.string().min(1).optional(), OPENROUTER_API_KEY: z.string().min(1).optional(),
RIVET_ENDPOINT: z.url(),
RIVET_PUBLIC_ENDPOINT: z.url().optional(),
}); });
export type AgentEnv = z.infer<typeof agentEnvSchema>; export type AgentEnv = z.infer<typeof agentEnvSchema>;

View File

@@ -5,15 +5,10 @@ export const env = createEnv({
emptyStringAsUndefined: true, emptyStringAsUndefined: true,
runtimeEnv: process.env, runtimeEnv: process.env,
server: { server: {
AGENT_BACKEND_URL: z.url().optional(),
CONVEX_SITE_URL: z.url(), CONVEX_SITE_URL: z.url(),
FLUE_DB_TOKEN: z.string().min(1), FLUE_DB_TOKEN: z.string().min(1),
FLUE_URL: z.url().optional(),
GITEA_TOKEN: z.string().min(1).optional(), GITEA_TOKEN: z.string().min(1).optional(),
GITEA_URL: z.url().default("https://git.openputer.com"), GITEA_URL: z.url().default("https://git.openputer.com"),
GITHUB_CLIENT_ID: z.string().min(1).optional(),
GITHUB_CLIENT_SECRET: z.string().min(1).optional(),
GIT_CREDENTIAL_ENCRYPTION_KEY: z.string().min(43).optional(),
NATIVE_APP_URL: z.string().min(1).default("code://"), NATIVE_APP_URL: z.string().min(1).default("code://"),
SITE_URL: z.url(), SITE_URL: z.url(),
}, },

View File

@@ -6,7 +6,6 @@
"exports": { "exports": {
".": "./src/index.ts", ".": "./src/index.ts",
"./agent-os": "./src/agent-os.ts", "./agent-os": "./src/agent-os.ts",
"./execution-runtime": "./src/execution-runtime.ts",
"./git": "./src/git.ts", "./git": "./src/git.ts",
"./git-local-runtime": "./src/git-local-runtime.ts", "./git-local-runtime": "./src/git-local-runtime.ts",
"./git-remote-runtime": "./src/git-remote-runtime.ts", "./git-remote-runtime": "./src/git-remote-runtime.ts",
@@ -17,7 +16,6 @@
"./signal": "./src/signal.ts", "./signal": "./src/signal.ts",
"./smoke": "./src/smoke.ts", "./smoke": "./src/smoke.ts",
"./work": "./src/work.ts", "./work": "./src/work.ts",
"./work-artifact": "./src/work-artifact.ts",
"./work-definition": "./src/work-definition.ts", "./work-definition": "./src/work-definition.ts",
"./work-design": "./src/work-design.ts", "./work-design": "./src/work-design.ts",
"./work-lifecycle": "./src/work-lifecycle.ts", "./work-lifecycle": "./src/work-lifecycle.ts",
@@ -37,7 +35,7 @@
}, },
"devDependencies": { "devDependencies": {
"@code/config": "workspace:*", "@code/config": "workspace:*",
"@types/node": "catalog:", "@types/node": "^22.13.14",
"typescript": "catalog:", "typescript": "catalog:",
"vitest": "catalog:" "vitest": "catalog:"
} }

View File

@@ -0,0 +1,13 @@
{
"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"
},

View File

@@ -1,38 +0,0 @@
import { Effect } from "effect";
import { describe, expect, test } from "vitest";
import {
decodeGitConnectionInput,
decodeWorkAttemptExecutionResult,
} from "./execution-runtime";
describe("execution runtime contracts", () => {
test("accepts one authenticated Gitea connection", async () => {
const decoded = await Effect.runPromise(
decodeGitConnectionInput({
credential: "secret",
credentialKind: "token",
provider: "gitea",
serverUrl: "https://git.example.com",
username: "zopu",
})
);
expect(decoded.provider).toBe("gitea");
});
test("requires exact base and candidate revisions", async () => {
await expect(
Effect.runPromise(
decodeWorkAttemptExecutionResult({
baseRevision: "",
candidateRevision: "abc123",
changedFiles: [],
diff: "",
environmentId: "workspace-1",
events: [],
summary: "done",
})
)
).rejects.toMatchObject({ reason: "InvalidInput" });
});
});

View File

@@ -1,155 +0,0 @@
/* eslint-disable max-classes-per-file -- execution boundary errors live with their schemas. */
import { Effect, Schema } from "effect";
const Text = Schema.String.check(
Schema.makeFilter((value) => value.trim().length > 0, {
expected: "a non-empty string",
})
);
const HttpUrl = Schema.String.check(
Schema.makeFilter(
(value) => {
try {
const url = new URL(value);
return url.protocol === "http:" || url.protocol === "https:";
} catch {
return false;
}
},
{ expected: "an HTTP URL" }
)
);
export const GitProvider = Schema.Literals(["github", "gitea"]);
export type GitProvider = typeof GitProvider.Type;
export const GitCredentialKind = Schema.Literals(["oauth", "token"]);
export type GitCredentialKind = typeof GitCredentialKind.Type;
export const GitConnectionInput = Schema.Struct({
credential: Text,
credentialKind: GitCredentialKind,
provider: GitProvider,
serverUrl: HttpUrl,
username: Schema.optional(Text),
});
export type GitConnectionInput = typeof GitConnectionInput.Type;
export const GitConnectionView = Schema.Struct({
connectedAt: Schema.Number,
credentialKind: GitCredentialKind,
id: Text,
provider: GitProvider,
serverUrl: HttpUrl,
username: Schema.optional(Text),
});
export type GitConnectionView = typeof GitConnectionView.Type;
export const RepositoryExecutionAuth = Schema.Struct({
credential: Text,
provider: GitProvider,
serverUrl: HttpUrl,
username: Schema.optional(Text),
});
export type RepositoryExecutionAuth = typeof RepositoryExecutionAuth.Type;
export const ExecutionEventKind = Schema.Literals([
"runtime.preparing",
"repository.cloning",
"repository.ready",
"harness.started",
"harness.progress",
"harness.log",
"repository.changed",
"runtime.completed",
"runtime.failed",
"runtime.cancelled",
]);
export type ExecutionEventKind = typeof ExecutionEventKind.Type;
export const ExecutionEvent = Schema.Struct({
kind: ExecutionEventKind,
message: Text,
metadata: Schema.Record(Schema.String, Schema.String),
occurredAt: Schema.Number,
sequence: Schema.Int,
});
export type ExecutionEvent = typeof ExecutionEvent.Type;
export const WorkAttemptExecutionInput = Schema.Struct({
attemptId: Text,
auth: RepositoryExecutionAuth,
baseBranch: Text,
prompt: Text,
repositoryUrl: HttpUrl,
runId: Text,
workId: Text,
workspaceKey: Text,
});
export type WorkAttemptExecutionInput = typeof WorkAttemptExecutionInput.Type;
export const WorkAttemptExecutionResult = Schema.Struct({
baseRevision: Text,
candidateRevision: Text,
changedFiles: Schema.Array(Text),
diff: Schema.String,
environmentId: Text,
events: Schema.Array(ExecutionEvent),
summary: Text,
});
export type WorkAttemptExecutionResult = typeof WorkAttemptExecutionResult.Type;
export const WorkAttemptExecutionErrorReason = Schema.Literals([
"Authentication",
"Cancelled",
"HarnessFailed",
"InvalidInput",
"ProviderUnavailable",
"RepositoryFailed",
"Timeout",
]);
export type WorkAttemptExecutionErrorReason =
typeof WorkAttemptExecutionErrorReason.Type;
export class WorkAttemptExecutionError extends Schema.TaggedErrorClass<WorkAttemptExecutionError>()(
"WorkAttemptExecutionError",
{
message: Schema.String,
reason: WorkAttemptExecutionErrorReason,
retryable: Schema.Boolean,
}
) {}
const invalidInput = (message: string) =>
new WorkAttemptExecutionError({
message,
reason: "InvalidInput",
retryable: false,
});
export const decodeGitConnectionInput = (input: unknown) =>
Schema.decodeUnknownEffect(GitConnectionInput)(input).pipe(
Effect.mapError((cause) => invalidInput(cause.message))
);
export const decodeWorkAttemptExecutionInput = (input: unknown) =>
Schema.decodeUnknownEffect(WorkAttemptExecutionInput)(input).pipe(
Effect.mapError((cause) => invalidInput(cause.message))
);
export const decodeWorkAttemptExecutionResult = (input: unknown) =>
Schema.decodeUnknownEffect(WorkAttemptExecutionResult)(input).pipe(
Effect.mapError((cause) => invalidInput(cause.message))
);
export interface SandboxRuntime {
readonly name: string;
readonly execute: (
input: WorkAttemptExecutionInput,
signal?: AbortSignal
) => Effect.Effect<WorkAttemptExecutionResult, WorkAttemptExecutionError>;
readonly cancel: (
workspaceKey: string
) => Effect.Effect<void, WorkAttemptExecutionError>;
}

View File

@@ -1,6 +1,5 @@
// oxlint-disable-next-line no-barrel-file -- The package root intentionally exposes its public modules. // oxlint-disable-next-line no-barrel-file -- The package root intentionally exposes its public modules.
export * from "./agent-os"; export * from "./agent-os";
export * from "./execution-runtime";
export * from "./git"; export * from "./git";
export * from "./git-local-runtime"; export * from "./git-local-runtime";
export * from "./git-remote-runtime"; export * from "./git-remote-runtime";
@@ -11,7 +10,6 @@ export * from "./project-work";
export * from "./signal"; export * from "./signal";
export * from "./smoke"; export * from "./smoke";
export * from "./work"; export * from "./work";
export * from "./work-artifact";
export * from "./work-definition"; export * from "./work-definition";
export * from "./work-design"; export * from "./work-design";
export * from "./work-lifecycle"; export * from "./work-lifecycle";

View File

@@ -8,12 +8,7 @@ const Text = Schema.String.check(
}) })
); );
export const RunStatus = Schema.Literals([ export const RunStatus = Schema.Literals(["ready", "running", "terminal"]);
"ready",
"running",
"terminal",
"cancelled",
]);
export type RunStatus = typeof RunStatus.Type; export type RunStatus = typeof RunStatus.Type;
export const AttemptStatus = Schema.Literals([ export const AttemptStatus = Schema.Literals([
"queued", "queued",

View File

@@ -1,58 +0,0 @@
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();
});
});

View File

@@ -1,141 +0,0 @@
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);

View File

@@ -59,8 +59,8 @@ export class DefinitionError extends Schema.TaggedErrorClass<DefinitionError>()(
} }
) {} ) {}
export const decodeDefinition = Effect.fn("Work.decodeDefinition")( export const validateDefinition = Effect.fn("Work.validateDefinition")(
function* decodeDefinition(input: unknown) { function* validateDefinition(input: unknown) {
const definition = yield* Schema.decodeUnknownEffect(WorkDefinition)( const definition = yield* Schema.decodeUnknownEffect(WorkDefinition)(
input input
).pipe( ).pipe(
@@ -69,25 +69,6 @@ export const decodeDefinition = Effect.fn("Work.decodeDefinition")(
new DefinitionError({ message: cause.message, reason: "Invalid" }) 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( const blocked = definition.questions.some(
(question) => question.status === "open" && question.impact === "high" (question) => question.status === "open" && question.impact === "high"
); );

View File

@@ -59,7 +59,6 @@ export class DesignError extends Schema.TaggedErrorClass<DesignError>()(
"Invalid", "Invalid",
"NoObservableSlices", "NoObservableSlices",
"DefinitionMismatch", "DefinitionMismatch",
"InvalidSliceGraph",
]), ]),
} }
) {} ) {}
@@ -74,10 +73,8 @@ export const validateDesignPacket = Effect.fn("Work.validateDesignPacket")(
); );
if ( if (
design.slices.length === 0 || design.slices.length === 0 ||
design.slices.length > 4 ||
design.slices.some( design.slices.some(
(slice) => (slice) =>
slice.codeBoundaries.length === 0 ||
slice.observableBehavior.trim().length === 0 || slice.observableBehavior.trim().length === 0 ||
slice.evidenceRequirements.length === 0 || slice.evidenceRequirements.length === 0 ||
slice.verification.length === 0 slice.verification.length === 0
@@ -91,24 +88,6 @@ 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; return design;
} }
); );

View File

@@ -3,7 +3,7 @@ import { describe, expect, test } from "vitest";
import { FakeHarnessLive } from "./harness-runtime"; import { FakeHarnessLive } from "./harness-runtime";
import { defaultCodingKitV0, resolveOutcome, shouldRetry } from "./resolver"; import { defaultCodingKitV0, resolveOutcome, shouldRetry } from "./resolver";
import { decodeDefinition, validateDefinition } from "./work-definition"; import { validateDefinition } from "./work-definition";
import { validateDesignPacket } from "./work-design"; import { validateDesignPacket } from "./work-design";
import { canTransitionWork } from "./work-lifecycle"; import { canTransitionWork } from "./work-lifecycle";
@@ -37,25 +37,11 @@ const probeOutcome = (
describe("work resolution contracts", () => { describe("work resolution contracts", () => {
test("high-impact open questions block definition approval", async () => { test("high-impact open questions block definition approval", async () => {
await expect(
Effect.runPromise(decodeDefinition(definition))
).resolves.toEqual(definition);
await expect( await expect(
Effect.runPromise(validateDefinition(definition)) Effect.runPromise(validateDefinition(definition))
).rejects.toThrow(/High-impact/u); ).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 () => { test("observable slices and terminal fake outcomes are enforced", async () => {
const approved = await Effect.runPromise( const approved = await Effect.runPromise(
validateDefinition({ ...definition, questions: [] }) validateDefinition({ ...definition, questions: [] })

View File

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

View File

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

View File

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