From 85633a512408a223ba97ff304fff611360861e90 Mon Sep 17 00:00:00 2001 From: -Puter <22245429+puterhimself@users.noreply.github.com> Date: Tue, 28 Jul 2026 00:07:35 +0530 Subject: [PATCH] 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. --- package.json | 2 +- packages/backend/convex/crons.ts | 19 ++ packages/backend/convex/workExecution.test.ts | 274 ++++++++++++++++++ packages/backend/convex/workExecution.ts | 204 +++++++++++-- packages/backend/convex/workPlanning.test.ts | 118 +++++++- packages/backend/convex/workPlanning.ts | 42 ++- packages/primitives/src/resolver.ts | 38 ++- packages/primitives/src/work-lifecycle.ts | 10 +- .../primitives/src/work-resolution.test.ts | 57 +++- 9 files changed, 717 insertions(+), 47 deletions(-) create mode 100644 packages/backend/convex/crons.ts create mode 100644 packages/backend/convex/workExecution.test.ts diff --git a/package.json b/package.json index 9c32f09..33ae6f2 100644 --- a/package.json +++ b/package.json @@ -45,7 +45,7 @@ "dev": "vp run -r dev", "build": "vp run -r build", "check-types": "vp run -r check-types", - "check": "ultracite check package.json vite.config.ts apps/web/package.json apps/web/src/root.tsx apps/web/src/index.css apps/web/src/components/chat apps/web/src/components/slice-one apps/web/src/hooks/chat apps/web/src/hooks/slice-one apps/web/src/lib/chat apps/web/src/lib/slice-one apps/web/src/routes.ts apps/web/src/routes/app/dashboard/page.tsx apps/web/src/routes/app/mobile/page.tsx packages/agents/package.json packages/agents/src/agents/zopu.ts packages/agents/src/app.ts packages/agents/src/auth.ts packages/agents/src/tools/slice-one.ts packages/backend/package.json packages/backend/convex/conversationMessages.ts packages/backend/convex/conversationMessages.test.ts packages/backend/convex/projects.ts packages/backend/convex/schema.ts packages/backend/convex/signalRouting.ts packages/backend/convex/signalRouting.test.ts packages/backend/convex/works.ts packages/backend/convex/works.test.ts packages/primitives/src/work.ts packages/primitives/src/work.test.ts", + "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": "vp lint", "format": "vp fmt", "staged": "vp staged", diff --git a/packages/backend/convex/crons.ts b/packages/backend/convex/crons.ts new file mode 100644 index 0000000..7be4c15 --- /dev/null +++ b/packages/backend/convex/crons.ts @@ -0,0 +1,19 @@ +import { cronJobs, makeFunctionReference } from "convex/server"; + +// Reconciles attempts whose worker lease expired (crash/restart). The target +// is referenced by string so this file does not depend on regenerated codegen. +const reconcileRef = makeFunctionReference< + "mutation", + Record, + { reconciled: number } | null +>("workExecution:reconcileExpiredAttempts"); + +const crons = cronJobs(); + +crons.interval( + "reconcile expired work attempts", + { seconds: 30 }, + reconcileRef +); + +export default crons; diff --git a/packages/backend/convex/workExecution.test.ts b/packages/backend/convex/workExecution.test.ts new file mode 100644 index 0000000..59b343d --- /dev/null +++ b/packages/backend/convex/workExecution.test.ts @@ -0,0 +1,274 @@ +import { convexTest } from "convex-test"; +import { anyApi } from "convex/server"; +import { describe, expect, test } from "vitest"; + +import type { Id } from "./_generated/dataModel"; +import schema from "./schema"; + +// Execution-only module set: avoids loading workPlanning (and its env read) so +// these tests run without Convex env vars. +const modules = { + "./_generated/api.ts": () => import("./_generated/api"), + "./_generated/server.ts": () => import("./_generated/server"), + "./authz.ts": () => import("./authz"), + "./workExecution.ts": () => import("./workExecution"), +}; +const identity = { tokenIdentifier: "https://convex.test|exec-user" }; +const api = anyApi; + +const makeTest = () => convexTest({ modules, schema }); +type TestT = ReturnType; + +type Scenario = + | "success" + | "transient-failure-then-success" + | "needs-input" + | "permanent-failure" + | "cancelled"; + +const seedReadyWork = async () => { + const t = convexTest({ modules, schema }); + const { workId } = await t.withIdentity(identity).run(async (ctx) => { + const organizationId = await ctx.db.insert("organizations", { + createdAt: 1, + createdBy: identity.tokenIdentifier, + kind: "personal", + name: "Personal", + }); + await ctx.db.insert("organizationMembers", { + createdAt: 1, + organizationId, + role: "owner", + userId: identity.tokenIdentifier, + }); + const projectId = await ctx.db.insert("projects", { + createdAt: 1, + name: "Zopu", + normalizedSourceUrl: "https://example.com/zopu", + organizationId, + repositoryPath: "puter/zopu", + sourceHost: "example.com", + sourceUrl: "https://example.com/zopu", + updatedAt: 1, + }); + const id = await ctx.db.insert("works", { + createdAt: 1, + definitionApprovalVersion: 1, + definitionVersion: 1, + designApprovalVersion: 1, + designVersion: 1, + objective: "Prove execution", + organizationId, + projectId, + status: "ready", + title: "Executable work", + updatedAt: 1, + }); + await ctx.db.insert("designPackets", { + createdAt: 1, + createdBy: "test", + definitionVersion: 1, + payloadJson: "{}", + status: "current", + version: 1, + workId: id, + }); + await ctx.db.insert("workSlices", { + designVersion: 1, + objective: "execute the slice", + observableBehavior: "terminal run", + ordinal: 0, + payloadJson: "{}", + sliceId: "slice-1", + status: "ready", + title: "first", + workId: id, + }); + return { workId: id }; + }); + return { t, workId }; +}; + +const runAttempt = async ( + t: TestT, + attemptId: Id<"workAttempts">, + scenario: Scenario +) => t.action(api.workExecution.executeFakeAttempt, { attemptId, scenario }); + +const attemptNumber = async ( + t: TestT, + runId: Id<"workRuns">, + number: number +): Promise> => { + const attempt = await t.run(async (ctx) => + ctx.db + .query("workAttempts") + .withIndex("by_run_and_number", (q) => + q.eq("runId", runId).eq("number", number) + ) + .unique() + ); + if (!attempt) throw new Error(`attempt ${number} not found`); + return attempt._id; +}; + +const snapshot = async (t: TestT, workId: Id<"works">, runId: Id<"workRuns">) => + t.run(async (ctx) => ({ + attempts: await ctx.db + .query("workAttempts") + .withIndex("by_run_and_number", (q) => q.eq("runId", runId)) + .collect(), + run: await ctx.db.get(runId), + slice: await ctx.db + .query("workSlices") + .withIndex("by_work_and_designVersion", (q) => + q.eq("workId", workId).eq("designVersion", 1) + ) + .first(), + work: await ctx.db.get(workId), + })); + +describe("simulated execution resolver", () => { + test("success settles Work as completed and marks the slice completed", async () => { + const { t, workId } = await seedReadyWork(); + const started = await t + .withIdentity(identity) + .mutation(api.workExecution.startSimulatedExecution, { + scenario: "success", + sliceId: "slice-1", + workId, + }); + await runAttempt(t, started.attemptId, "success"); + const state = await snapshot(t, workId, started.runId); + expect(state.work?.status).toBe("completed"); + expect(state.run?.terminalClassification).toBe("Succeeded"); + expect(state.attempts).toHaveLength(1); + expect(state.slice?.status).toBe("completed"); + }); + + test("transient failure retries within the kit budget and then succeeds", async () => { + const { t, workId } = await seedReadyWork(); + const started = await t + .withIdentity(identity) + .mutation(api.workExecution.startSimulatedExecution, { + scenario: "transient-failure-then-success", + workId, + }); + await runAttempt(t, started.attemptId, "transient-failure-then-success"); + // resolver auto-scheduled attempt 2; drive it manually in the test runtime + const second = await attemptNumber(t, started.runId, 2); + await runAttempt(t, second, "transient-failure-then-success"); + const state = await snapshot(t, workId, started.runId); + expect(state.work?.status).toBe("completed"); + expect(state.run?.terminalClassification).toBe("Succeeded"); + expect(state.attempts).toHaveLength(2); + expect(state.attempts[0]?.classification).toBe("RetryableFailure"); + }); + + test("permanent failure settles Work as failed, not silently runnable", async () => { + const { t, workId } = await seedReadyWork(); + const started = await t + .withIdentity(identity) + .mutation(api.workExecution.startSimulatedExecution, { + scenario: "permanent-failure", + workId, + }); + await runAttempt(t, started.attemptId, "permanent-failure"); + const state = await snapshot(t, workId, started.runId); + expect(state.work?.status).toBe("failed"); + expect(state.run?.terminalClassification).toBe("PermanentFailure"); + }); + + test("needs-input surfaces a blocker instead of looping", async () => { + const { t, workId } = await seedReadyWork(); + const started = await t + .withIdentity(identity) + .mutation(api.workExecution.startSimulatedExecution, { + scenario: "needs-input", + workId, + }); + await runAttempt(t, started.attemptId, "needs-input"); + const state = await snapshot(t, workId, started.runId); + expect(state.work?.status).toBe("needs-input"); + expect(state.run?.terminalClassification).toBe("NeedsInput"); + }); + + test("cancellation terminates the attempt and returns Work to ready", async () => { + const { t, workId } = await seedReadyWork(); + const started = await t + .withIdentity(identity) + .mutation(api.workExecution.startSimulatedExecution, { + scenario: "success", + workId, + }); + await t + .withIdentity(identity) + .mutation(api.workExecution.cancelSimulatedExecution, { + runId: started.runId, + }); + const state = await snapshot(t, workId, started.runId); + expect(state.work?.status).toBe("ready"); + expect(state.run?.status).toBe("cancelled"); + expect(state.slice?.status).toBe("ready"); + }); + + test("a slice that is not part of the current approved Design is rejected", async () => { + const { t, workId } = await seedReadyWork(); + await expect( + t + .withIdentity(identity) + .mutation(api.workExecution.startSimulatedExecution, { + scenario: "success", + sliceId: "not-a-real-slice", + workId, + }) + ).rejects.toThrow(/current approved Design/u); + }); + + test("manual retry restarts a failed Run", async () => { + const { t, workId } = await seedReadyWork(); + const started = await t + .withIdentity(identity) + .mutation(api.workExecution.startSimulatedExecution, { + scenario: "permanent-failure", + workId, + }); + await runAttempt(t, started.attemptId, "permanent-failure"); + expect((await snapshot(t, workId, started.runId)).work?.status).toBe( + "failed" + ); + const retried = await t + .withIdentity(identity) + .mutation(api.workExecution.retrySimulatedExecution, { + runId: started.runId, + }); + await runAttempt(t, retried.attemptId as Id<"workAttempts">, "success"); + const state = await snapshot(t, workId, started.runId); + expect(state.work?.status).toBe("completed"); + expect(retried.number).toBe(2); + }); + + test("reconciling an expired lease resumes within budget", async () => { + const { t, workId } = await seedReadyWork(); + const started = await t + .withIdentity(identity) + .mutation(api.workExecution.startSimulatedExecution, { + scenario: "success", + workId, + }); + // simulate a worker that claimed but died before finishing + await t.run(async (ctx) => { + await ctx.db.patch(started.attemptId, { + leaseExpiresAt: Date.now() - 1000, + leaseOwner: "dead-worker", + status: "claimed", + }); + }); + await t.mutation(api.workExecution.reconcileExpiredAttempts, {}); + const after = await snapshot(t, workId, started.runId); + expect(after.attempts.find((a) => a.number === 1)?.status).toBe("terminal"); + expect( + after.attempts.find((a) => a.number === 2 && a.status === "queued") + ).toBeTruthy(); + }); +}); diff --git a/packages/backend/convex/workExecution.ts b/packages/backend/convex/workExecution.ts index 4d68c25..4385e2f 100644 --- a/packages/backend/convex/workExecution.ts +++ b/packages/backend/convex/workExecution.ts @@ -2,12 +2,16 @@ import { FakeHarnessLive, type FakeScenario, } from "@code/primitives/harness-runtime"; -import { defaultCodingKitV0 } from "@code/primitives/resolver"; +import { + type AttemptClassification, + defaultCodingKitV0, + resolveOutcome, +} from "@code/primitives/resolver"; import { makeFunctionReference } from "convex/server"; import { ConvexError, v } from "convex/values"; import { Effect } from "effect"; -import type { Id } from "./_generated/dataModel"; +import type { Doc, Id } from "./_generated/dataModel"; import { internalAction, internalMutation, @@ -37,16 +41,17 @@ const checkpointRef = makeFunctionReference< }, unknown >("workExecution:checkpointAttempt"); -const completeRef = makeFunctionReference< +const finishRef = makeFunctionReference< "mutation", { attemptId: Id<"workAttempts">; classification: string; owner: string; + retryable: boolean; summary: string; }, unknown ->("workExecution:completeAttempt"); +>("workExecution:finishAttempt"); const now = () => Date.now(); const requireWorkForMember = async (ctx: any, workId: Id<"works">) => { @@ -81,6 +86,54 @@ const appendWorkEvent = async ( }); }; +// Slice the Run targets for the currently approved Design. `sliceId` is +// optional only to let the resolver default to the first ready slice. +const resolveRunSlice = async ( + ctx: any, + work: Doc<"works">, + sliceId?: string +) => { + const currentDesignVersion = work.designVersion; + if (currentDesignVersion === undefined) + throw new ConvexError("Work has no approved Design to execute"); + const slices = await ctx.db + .query("workSlices") + .withIndex("by_work_and_designVersion", (q: any) => + q.eq("workId", work._id).eq("designVersion", currentDesignVersion) + ) + .collect(); + if (slices.length === 0) + throw new ConvexError("No slices found for the current approved Design"); + const slice = sliceId + ? slices.find((row: any) => row.sliceId === sliceId) + : slices.sort((a: any, b: any) => a.ordinal - b.ordinal)[0]; + if (!slice) + throw new ConvexError( + "Requested slice does not belong to the current approved Design" + ); + return slice as Doc<"workSlices">; +}; + +const setSliceStatus = async ( + ctx: any, + workId: Id<"works">, + sliceId: string | undefined, + status: any +) => { + if (!sliceId) return; + const slice = ( + await ctx.db + .query("workSlices") + .withIndex("by_work_and_designVersion", (q: any) => + q.eq("workId", workId) + ) + .collect() + ).find((row: any) => row.sliceId === sliceId); + if (slice) await ctx.db.patch(slice._id, { status }); +}; + +const RETRYABLE_WORK_STATUSES = ["failed", "needs-input", "blocked", "ready"]; + export const startSimulatedExecution = mutation({ args: { workId: v.id("works"), @@ -104,10 +157,11 @@ export const startSimulatedExecution = mutation({ throw new ConvexError( "Execution requires exact approved Definition and Design versions" ); + const slice = await resolveRunSlice(ctx, work, args.sliceId); const createdAt = now(); const runId = await ctx.db.insert("workRuns", { workId: work._id, - sliceId: args.sliceId, + sliceId: slice.sliceId, status: "ready", scenario: args.scenario, kitId: defaultCodingKitV0.id, @@ -120,7 +174,11 @@ export const startSimulatedExecution = mutation({ number: 1, status: "queued", }); - await ctx.db.patch(work._id, { status: "executing", updatedAt: createdAt }); + await ctx.db.patch(work._id, { + status: "executing", + updatedAt: createdAt, + }); + await ctx.db.patch(slice._id, { status: "running" }); await ctx.db.patch(runId, { status: "running", startedAt: createdAt }); await appendWorkEvent( ctx, @@ -163,6 +221,7 @@ export const cancelSimulatedExecution = mutation({ classification: "Cancelled", summary: "Simulation cancelled", }); + await setSliceStatus(ctx, work._id, run.sliceId, "ready"); await ctx.db.patch(work._id, { status: "ready", updatedAt: now() }); await appendWorkEvent( ctx, @@ -175,6 +234,8 @@ export const cancelSimulatedExecution = mutation({ }, }); +// User-initiated restart of a terminal Run. Unlike the automatic resolver +// retry, this is an explicit decision and may exceed the kit budget. export const retrySimulatedExecution = mutation({ args: { runId: v.id("workRuns") }, handler: async (ctx, args) => { @@ -183,6 +244,8 @@ export const retrySimulatedExecution = mutation({ const work = await requireWorkForMember(ctx, run.workId); if (run.status !== "terminal") throw new ConvexError("Only terminal Runs can be retried"); + if (!RETRYABLE_WORK_STATUSES.includes(work.status)) + throw new ConvexError("Work is not in a retryable state"); const attempts = await ctx.db .query("workAttempts") .withIndex("by_run_and_number", (q: any) => q.eq("runId", run._id)) @@ -202,6 +265,7 @@ export const retrySimulatedExecution = mutation({ terminalSummary: undefined, }); await ctx.db.patch(work._id, { status: "executing", updatedAt: now() }); + await setSliceStatus(ctx, work._id, run.sliceId, "running"); await ctx.scheduler.runAfter(0, executeRef, { attemptId, scenario: run.scenario, @@ -274,11 +338,16 @@ export const checkpointAttempt = internalMutation({ }, }); -export const completeAttempt = internalMutation({ +// Single resolver mutation: record the attempt outcome, then either schedule +// the next attempt within the kit retry policy or settle the Run/Work at the +// terminal classification. Replaces the old completeAttempt that always reset +// Work to "ready" and ignored the retry policy. +export const finishAttempt = internalMutation({ args: { attemptId: v.id("workAttempts"), owner: v.string(), classification: v.string(), + retryable: v.boolean(), summary: v.string(), }, handler: async (ctx, args) => { @@ -288,7 +357,7 @@ export const completeAttempt = internalMutation({ attempt.leaseOwner !== args.owner || attempt.status === "terminal" ) - return false; + return null; const endedAt = now(); await ctx.db.patch(attempt._id, { status: "terminal", @@ -297,8 +366,44 @@ export const completeAttempt = internalMutation({ summary: args.summary, leaseExpiresAt: undefined, }); + const outcome = { + classification: args.classification as AttemptClassification, + retryable: args.retryable, + summary: args.summary, + }; + const resolution = resolveOutcome( + outcome, + attempt.number, + defaultCodingKitV0.retryPolicy + ); const run = await ctx.db.get(attempt.runId); - if (!run) return false; + if (!run) return null; + await appendWorkEvent( + ctx, + attempt.workId, + "attempt.completed", + `attempt-completed:${attempt._id}`, + String(attempt._id), + JSON.stringify({ + classification: args.classification, + retried: resolution.kind === "retry", + summary: args.summary, + }) + ); + if (resolution.kind === "retry") { + // Keep Run running and Work executing; spin the next attempt. + const nextAttemptId = await ctx.db.insert("workAttempts", { + runId: run._id, + workId: attempt.workId, + number: attempt.number + 1, + status: "queued", + }); + await ctx.scheduler.runAfter(0, executeRef, { + attemptId: nextAttemptId, + scenario: run.scenario, + }); + return { nextAttemptId, retried: true }; + } await ctx.db.patch(run._id, { status: "terminal", endedAt, @@ -308,22 +413,16 @@ export const completeAttempt = internalMutation({ const work = await ctx.db.get(run.workId); if (work) await ctx.db.patch(work._id, { - status: args.classification === "NeedsInput" ? "needs-input" : "ready", + status: resolution.workStatus, updatedAt: endedAt, }); - if (work) - await ctx.db.insert("workEvents", { - workId: work._id, - kind: "attempt.completed", - idempotencyKey: `attempt-completed:${attempt._id}`, - referenceId: String(attempt._id), - payloadJson: JSON.stringify({ - classification: args.classification, - summary: args.summary, - }), - createdAt: endedAt, - }); - return true; + await setSliceStatus( + ctx, + attempt.workId, + run.sliceId, + args.classification === "Succeeded" ? "completed" : "ready" + ); + return { retried: false }; }, }); @@ -355,22 +454,26 @@ export const executeFakeAttempt = internalAction({ for (const item of events) await ctx.runMutation(checkpointRef, { attemptId: args.attemptId, - owner, - sequence: item.sequence, kind: item.kind, message: item.message, metadataJson: JSON.stringify(item.metadata), + owner, + sequence: item.sequence, }); - await ctx.runMutation(completeRef, { + await ctx.runMutation(finishRef, { attemptId: args.attemptId, - owner, classification: outcome.classification, + owner, + retryable: outcome.retryable, summary: outcome.summary, }); return null; }, }); +// Bounded recovery: a claimed/running attempt whose lease expired means the +// worker died mid-flight. Terminate that attempt, then either resume within +// the kit budget or settle the Run/Work so nothing stays "running" forever. export const reconcileExpiredAttempts = internalMutation({ args: {}, handler: async (ctx) => { @@ -383,17 +486,56 @@ export const reconcileExpiredAttempts = internalMutation({ attempt.leaseExpiresAt < now() ) { await ctx.db.patch(attempt._id, { - status: "queued", + status: "terminal", + endedAt: now(), + classification: "Blocked", + summary: "Attempt lease expired and was reconciled", leaseOwner: undefined, leaseExpiresAt: undefined, }); const run = await ctx.db.get(attempt.runId); - if (run) + await appendWorkEvent( + ctx, + attempt.workId, + "attempt.reconciled", + `attempt-reconciled:${attempt._id}`, + String(attempt._id), + JSON.stringify({ attemptNumber: attempt.number }) + ); + reconciled += 1; + if (!run) continue; + const attempts = await ctx.db + .query("workAttempts") + .withIndex("by_run_and_number", (q: any) => q.eq("runId", run._id)) + .collect(); + const withinBudget = + attempts.length < defaultCodingKitV0.retryPolicy.maxAttempts; + if (run.status === "running" && withinBudget) { + const nextAttemptId = await ctx.db.insert("workAttempts", { + runId: run._id, + workId: attempt.workId, + number: attempts.length + 1, + status: "queued", + }); await ctx.scheduler.runAfter(0, executeRef, { - attemptId: attempt._id, + attemptId: nextAttemptId, scenario: run.scenario, }); - reconciled += 1; + } else { + await ctx.db.patch(run._id, { + status: "terminal", + endedAt: now(), + terminalClassification: "Blocked", + terminalSummary: "Run reconciled after expired lease", + }); + const work = await ctx.db.get(run.workId); + if (work) + await ctx.db.patch(work._id, { + status: "blocked", + updatedAt: now(), + }); + await setSliceStatus(ctx, attempt.workId, run.sliceId, "ready"); + } } } return { reconciled }; diff --git a/packages/backend/convex/workPlanning.test.ts b/packages/backend/convex/workPlanning.test.ts index c824ac9..7cc959f 100644 --- a/packages/backend/convex/workPlanning.test.ts +++ b/packages/backend/convex/workPlanning.test.ts @@ -143,9 +143,125 @@ describe("Work planning and execution commands", () => { run: await ctx.db.get(started.runId as Id<"workRuns">), work: await ctx.db.get(fixture.workId), })); - expect(completed.work?.status).toBe("ready"); + expect(completed.work?.status).toBe("completed"); expect(completed.run?.terminalClassification).toBe("Succeeded"); expect(completed.attempt?.status).toBe("terminal"); expect(completed.events.length).toBeGreaterThan(0); }); }); + +const validDefinitionPayload = () => ({ + acceptanceCriteria: ["terminal outcome"], + affectedUsers: ["user"], + assumptions: [], + constraints: [], + desiredOutcome: "A terminal simulation", + inScope: ["fake"], + outOfScope: [], + problem: "Need a proof", + questions: [], + requiredArtifacts: ["events"], + risk: "low", +}); + +describe("Work planning guards", () => { + test("a Definition cannot be revised while a Run is executing", async () => { + const t = convexTest({ modules, schema }); + const { workId } = await t.withIdentity(identity).run(async (ctx) => { + const organizationId = await ctx.db.insert("organizations", { + createdAt: 1, + createdBy: identity.tokenIdentifier, + kind: "personal", + name: "Personal", + }); + await ctx.db.insert("organizationMembers", { + createdAt: 1, + organizationId, + role: "owner", + userId: identity.tokenIdentifier, + }); + const projectId = await ctx.db.insert("projects", { + createdAt: 1, + name: "Zopu", + normalizedSourceUrl: "https://example.com/zopu", + organizationId, + repositoryPath: "puter/zopu", + sourceHost: "example.com", + sourceUrl: "https://example.com/zopu", + updatedAt: 1, + }); + const id = await ctx.db.insert("works", { + createdAt: 1, + definitionApprovalVersion: 1, + definitionVersion: 1, + designApprovalVersion: 1, + designVersion: 1, + objective: "Prove the guard", + organizationId, + projectId, + status: "executing", + title: "Guarded work", + updatedAt: 1, + }); + return { workId: id }; + }); + await expect( + t + .withIdentity(identity) + .mutation(api.workPlanning.saveDefinitionProposal, { + payloadJson: JSON.stringify(validDefinitionPayload()), + workId, + }) + ).rejects.toThrow(/while a Run is executing/u); + }); + + test("planner questions are persisted as durable Work questions", async () => { + const t = convexTest({ modules, schema }); + const { workId } = await t.withIdentity(identity).run(async (ctx) => { + const organizationId = await ctx.db.insert("organizations", { + createdAt: 1, + createdBy: identity.tokenIdentifier, + kind: "personal", + name: "Personal", + }); + const projectId = await ctx.db.insert("projects", { + createdAt: 1, + name: "Zopu", + normalizedSourceUrl: "https://example.com/zopu", + organizationId, + repositoryPath: "puter/zopu", + sourceHost: "example.com", + sourceUrl: "https://example.com/zopu", + updatedAt: 1, + }); + const id = await ctx.db.insert("works", { + createdAt: 1, + definitionVersion: 1, + objective: "Prove questions", + organizationId, + projectId, + status: "awaiting-definition-approval", + title: "Questionable work", + updatedAt: 1, + }); + return { workId: id }; + }); + await t.mutation(api.workPlanning.submitQuestion, { + questionJson: JSON.stringify({ + alternatives: [], + id: "q1", + impact: "medium", + prompt: "Which runtime?", + status: "open", + }), + token: "test-token", + workId, + }); + const questions = await t.run(async (ctx) => + ctx.db.query("workQuestions").collect() + ); + expect(questions).toHaveLength(1); + expect(questions[0]?.questionId).toBe("q1"); + expect(questions[0]?.definitionVersion).toBe(1); + }); +}); diff --git a/packages/backend/convex/workPlanning.ts b/packages/backend/convex/workPlanning.ts index f29590b..2f937a2 100644 --- a/packages/backend/convex/workPlanning.ts +++ b/packages/backend/convex/workPlanning.ts @@ -1,11 +1,12 @@ import { - validateDefinition, type WorkDefinition, + validateDefinition, + WorkQuestion, } from "@code/primitives/work-definition"; import { validateDesignPacket } from "@code/primitives/work-design"; import { makeFunctionReference } from "convex/server"; import { ConvexError, v } from "convex/values"; -import { Effect } from "effect"; +import { Effect, Schema } from "effect"; import type { Doc, Id } from "./_generated/dataModel"; import { @@ -129,6 +130,8 @@ const saveDefinition = async ( payloadJson: string, createdBy: string ) => { + if (work.status === "executing") + throw new ConvexError("Cannot revise Work while a Run is executing"); const decoded = await Effect.runPromise( validateDefinition({ ...objectPayload(payloadJson), @@ -352,6 +355,8 @@ const saveDesign = async ( payloadJson: string, createdBy: string ) => { + if (work.status === "executing") + throw new ConvexError("Cannot revise Work while a Run is executing"); if (work.definitionApprovalVersion !== work.definitionVersion) throw new ConvexError("Design must bind an approved Definition version"); const decoded = await Effect.runPromise( @@ -372,6 +377,15 @@ const saveDesign = async ( for (const row of prior) if (row.status === "current") await ctx.db.patch(row._id, { status: "superseded" }); + const designApprovals = await ctx.db + .query("workApprovals") + .withIndex("by_work_and_kind", (q: any) => + q.eq("workId", work._id).eq("kind", "design") + ) + .collect(); + for (const approval of designApprovals) + if (approval.status === "active") + await ctx.db.patch(approval._id, { status: "invalidated" }); const designId = await ctx.db.insert("designPackets", { workId: work._id, version: decoded.version, @@ -496,7 +510,29 @@ export const submitQuestion = mutation({ handler: async (ctx, args) => { if (args.token !== env.FLUE_DB_TOKEN) throw new ConvexError("Invalid agent control token"); - return { accepted: Boolean(JSON.parse(args.questionJson)) }; + const work = await requireWork(ctx, args.workId); + if (work.definitionVersion === undefined) + throw new ConvexError("Work has no Definition to attach a question to"); + const question = await Effect.runPromise( + Schema.decodeUnknownEffect(WorkQuestion)(JSON.parse(args.questionJson)) + ).catch((error: unknown) => { + throw new ConvexError( + error instanceof Error ? error.message : "Invalid question" + ); + }); + await ctx.db.insert("workQuestions", { + workId: work._id, + definitionVersion: work.definitionVersion, + questionId: question.id, + prompt: question.prompt, + impact: question.impact, + recommendation: question.recommendation, + alternativesJson: JSON.stringify(question.alternatives), + status: question.status, + answer: question.answer, + createdAt: Date.now(), + }); + return { accepted: true }; }, }); diff --git a/packages/primitives/src/resolver.ts b/packages/primitives/src/resolver.ts index 0454d89..5fc014d 100644 --- a/packages/primitives/src/resolver.ts +++ b/packages/primitives/src/resolver.ts @@ -1,5 +1,7 @@ import { Schema } from "effect"; +import type { WorkStatus } from "./work-lifecycle"; + const Text = Schema.String.check( Schema.makeFilter((value) => value.trim().length > 0, { expected: "a non-empty string", @@ -48,9 +50,6 @@ export const CodingKitV0 = Schema.Struct({ }); export type CodingKitV0 = typeof CodingKitV0.Type; -export const isTerminalClassification = (_: AttemptClassification): true => - true; - export const shouldRetry = ( outcome: AttemptOutcome, attemptNumber: number, @@ -60,6 +59,39 @@ export const shouldRetry = ( attemptNumber < policy.maxAttempts && policy.retryableClassifications.includes(outcome.classification); +// Terminal Work status the resolver settles on once it will run no further +// attempt. RetryableFailure lands here only after the retry budget is gone. +export const WORK_STATUS_FOR_OUTCOME: Readonly< + Record +> = { + Blocked: "blocked", + BudgetExhausted: "failed", + Cancelled: "ready", + NeedsInput: "needs-input", + PermanentFailure: "failed", + RetryableFailure: "failed", + Succeeded: "completed", + VerificationFailed: "blocked", +}; + +export type Resolution = + | { readonly kind: "retry" } + | { readonly kind: "terminal"; readonly workStatus: WorkStatus }; + +// Resolver decision after an attempt finishes: retry within policy, or settle +// Work at the terminal status mapped from the attempt classification. +export const resolveOutcome = ( + outcome: AttemptOutcome, + attemptNumber: number, + policy: RetryPolicy +): Resolution => + shouldRetry(outcome, attemptNumber, policy) + ? { kind: "retry" } + : { + kind: "terminal", + workStatus: WORK_STATUS_FOR_OUTCOME[outcome.classification], + }; + export const defaultCodingKitV0: CodingKitV0 = { harness: "fake", id: "coding-kit-v0", diff --git a/packages/primitives/src/work-lifecycle.ts b/packages/primitives/src/work-lifecycle.ts index d20a63c..fe9555c 100644 --- a/packages/primitives/src/work-lifecycle.ts +++ b/packages/primitives/src/work-lifecycle.ts @@ -20,14 +20,10 @@ export const WORK_TRANSITIONS: Readonly< Record > = { "awaiting-definition-approval": ["designing", "defining"], - "awaiting-design-approval": [ - "ready", - "designing", - "awaiting-definition-approval", - ], + "awaiting-design-approval": ["ready", "designing"], blocked: ["ready", "executing", "cancelled"], - cancelled: ["ready", "proposed"], - completed: ["proposed", "defining"], + cancelled: ["ready"], + completed: [], defining: ["awaiting-definition-approval", "proposed"], designing: ["awaiting-design-approval", "awaiting-definition-approval"], executing: [ diff --git a/packages/primitives/src/work-resolution.test.ts b/packages/primitives/src/work-resolution.test.ts index 4648b09..537862f 100644 --- a/packages/primitives/src/work-resolution.test.ts +++ b/packages/primitives/src/work-resolution.test.ts @@ -2,7 +2,7 @@ import { Effect } from "effect"; import { describe, expect, test } from "vitest"; import { FakeHarnessLive } from "./harness-runtime"; -import { defaultCodingKitV0, shouldRetry } from "./resolver"; +import { defaultCodingKitV0, resolveOutcome, shouldRetry } from "./resolver"; import { validateDefinition } from "./work-definition"; import { validateDesignPacket } from "./work-design"; import { canTransitionWork } from "./work-lifecycle"; @@ -30,6 +30,11 @@ const definition = { version: 1, }; +const probeOutcome = ( + classification: Parameters[0]["classification"], + retryable = false +) => ({ classification, retryable, summary: "probe" }); + describe("work resolution contracts", () => { test("high-impact open questions block definition approval", async () => { await expect( @@ -103,3 +108,53 @@ describe("work resolution contracts", () => { expect(canTransitionWork("ready", "executing")).toBe(true); }); }); + +describe("resolver decisions", () => { + const policy = defaultCodingKitV0.retryPolicy; + + test("retries a retryable failure while the budget remains", () => { + expect( + resolveOutcome(probeOutcome("RetryableFailure", true), 1, policy) + ).toEqual({ kind: "retry" }); + }); + + test("settles RetryableFailure as failed once the budget is exhausted", () => { + expect( + resolveOutcome(probeOutcome("RetryableFailure", true), 3, policy) + ).toEqual({ + kind: "terminal", + workStatus: "failed", + }); + }); + + test("maps each terminal classification to its Work status", () => { + expect(resolveOutcome(probeOutcome("Succeeded"), 1, policy)).toEqual({ + kind: "terminal", + workStatus: "completed", + }); + expect(resolveOutcome(probeOutcome("PermanentFailure"), 1, policy)).toEqual( + { + kind: "terminal", + workStatus: "failed", + } + ); + expect(resolveOutcome(probeOutcome("NeedsInput"), 1, policy)).toEqual({ + kind: "terminal", + workStatus: "needs-input", + }); + expect(resolveOutcome(probeOutcome("Blocked"), 1, policy)).toEqual({ + kind: "terminal", + workStatus: "blocked", + }); + expect(resolveOutcome(probeOutcome("Cancelled"), 1, policy)).toEqual({ + kind: "terminal", + workStatus: "ready", + }); + }); + + test("completed and cancelled are not silently reopened", () => { + expect(canTransitionWork("completed", "defining")).toBe(false); + expect(canTransitionWork("cancelled", "proposed")).toBe(false); + expect(canTransitionWork("ready", "executing")).toBe(true); + }); +});