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

feat: wire real AgentOS execution

Browse Source
This commit is contained in:
-Puter
2026-07-28 16:52:49 +05:30
parent 4d9f6da41b
commit 4edd456b5b
17 changed files with 1090 additions and 308 deletions
Download Patch File Download Diff File Expand all files Collapse all files

24
packages/backend/convex/gitConnectionData.ts
View File

@@ -3,6 +3,21 @@ import { ConvexError, v } from "convex/values";
import { internalMutation, mutation, query } from "./_generated/server";
import { requireCurrentOrganization, requireProjectMember } from "./authz";
// Provider ("forge") that owns a repository host. A project's source must be
// served by the same forge whose credentials are attached, so a Gitea token is
// never offered to GitHub and vice versa. Unknown hosts return null, leaving
// the caller free to attach without a forge constraint.
export const forgeForHost = (host: string): "github" | "gitea" | null => {
const normalized = host.toLowerCase();
if (normalized === "github.com" || normalized.endsWith(".githost.com")) {
return "github";
}
if (normalized === "git.openputer.com") {
return "gitea";
}
return null;
};
export const persist = internalMutation({
args: {
credentialCiphertext: v.string(),
@@ -110,6 +125,15 @@ export const attachToProject = mutation({
if (!connection || connection.organizationId !== organizationId) {
throw new ConvexError("Git connection not found");
}
const project = await ctx.db.get(args.projectId);
if (project) {
const expected = forgeForHost(project.sourceHost);
if (expected && connection.provider !== expected) {
throw new ConvexError(
`Git credential provider (${connection.provider}) does not match this project's forge (${expected})`
);
}
}
await ctx.db.patch(args.projectId, {
gitConnectionId: connection._id,
updatedAt: Date.now(),

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

@@ -368,6 +368,17 @@ export default defineSchema({
workAttempts: defineTable({
classification: v.optional(attemptClassification),
endedAt: v.optional(v.number()),
failureReason: v.optional(
v.union(
v.literal("Authentication"),
v.literal("Cancelled"),
v.literal("HarnessFailed"),
v.literal("InvalidInput"),
v.literal("ProviderUnavailable"),
v.literal("RepositoryFailed"),
v.literal("Timeout")
)
),
leaseExpiresAt: v.optional(v.number()),
leaseOwner: v.optional(v.string()),
number: v.number(),

31
packages/backend/convex/workExecutionAgent.ts
View File

@@ -1,7 +1,11 @@
"use node";
import { env } from "@code/env/convex";
import { decodeWorkAttemptExecutionResult } from "@code/primitives/execution-runtime";
import {
WorkAttemptExecutionError,
decodeWorkAttemptExecutionFailure,
decodeWorkAttemptExecutionResult,
} from "@code/primitives/execution-runtime";
import { ConvexError, v } from "convex/values";
import { Effect } from "effect";
@@ -56,23 +60,36 @@ export const executeAttempt = internalAction({
);
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}`
// Decode the agent's classified failure envelope and re-throw as the
// typed runtime error so the workflow handler maps reason/retryable to
// a durable attempt classification. A malformed envelope falls back to
// an InvalidInput failure (non-retryable).
const failure = await Effect.runPromise(
decodeWorkAttemptExecutionFailure(payload)
);
throw new WorkAttemptExecutionError(failure.error);
}
return await Effect.runPromise(decodeWorkAttemptExecutionResult(payload));
},
});
export const cancelAttempt = internalAction({
args: { workspaceKey: v.string() },
args: {
attemptId: v.string(),
workspaceKey: v.string(),
},
handler: async (_ctx, args) => {
// The workspace key remains the URL path segment (workspace identity),
// while the body carries the attemptId so the runtime can target the
// codex session `codex-${attemptId}` for cancellation.
await fetch(
`${backendUrl()}/internal/work-attempts/${encodeURIComponent(args.workspaceKey)}/cancel`,
{
headers: { authorization: `Bearer ${env.FLUE_DB_TOKEN}` },
body: JSON.stringify({ attemptId: args.attemptId }),
headers: {
authorization: `Bearer ${env.FLUE_DB_TOKEN}`,
"content-type": "application/json",
},
method: "POST",
}
);

430
packages/backend/convex/workExecutionWorkflow.test.ts
View File

@@ -7,94 +7,116 @@ import schema from "./schema";
const modules = import.meta.glob("./**/*.ts");
const api = anyApi;
type TestContext = ReturnType<typeof convexTest>;
interface Seeded {
attemptId: string;
runId: string;
t: TestContext;
workId: string;
}
// Seed an in-flight real attempt so each mutation under test starts from a
// realistic running state: Work "executing", Run "running", Attempt "running".
const seedRunningAttempt = async (
overrides: {
attemptStatus?: "queued" | "claimed" | "running" | "terminal";
runStatus?: "ready" | "running" | "terminal" | "cancelled";
} = {}
): Promise<Seeded> => {
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://github.com/puter/zopu",
organizationId,
repositoryPath: "puter/zopu",
sourceHost: "github.com",
sourceUrl: "https://github.com/puter/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: overrides.runStatus ?? "running",
workId,
});
const attemptId = await ctx.db.insert("workAttempts", {
number: 1,
runId,
status: overrides.attemptStatus ?? "running",
workId,
workspaceKey: "workspace-1",
});
return { attemptId, runId, workId };
});
return { ...seeded, t };
};
const successResult = {
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",
};
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 };
});
const { t, ...seeded } = await seedRunningAttempt();
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",
},
result: successResult,
});
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),
attempt: await ctx.db.get(seeded.attemptId as any),
run: await ctx.db.get(seeded.runId as any),
work: await ctx.db.get(seeded.workId as any),
}));
expect(state.attempt?.classification).toBe("Succeeded");
expect(state.run).toMatchObject({
@@ -109,3 +131,251 @@ describe("real work execution persistence", () => {
});
});
});
describe("classified failure mapping", () => {
test("maps a transient runtime reason to a retryable classification", async () => {
const { t, ...seeded } = await seedRunningAttempt();
await t.mutation(api.workExecutionWorkflow.failAttempt, {
attemptId: seeded.attemptId,
reason: "ProviderUnavailable",
retryable: true,
summary: "model provider timed out",
});
const attempt = await t.run((ctx) => ctx.db.get(seeded.attemptId as any));
expect(attempt?.classification).toBe("RetryableFailure");
expect(attempt?.failureReason).toBe("ProviderUnavailable");
});
test("queues a new attempt without terminalizing a retryable run", async () => {
const { t, ...seeded } = await seedRunningAttempt();
await t.mutation(api.workExecutionWorkflow.failAttempt, {
attemptId: seeded.attemptId,
reason: "ProviderUnavailable",
retryable: true,
summary: "provider unavailable",
});
const state = await t.run(async (ctx) => ({
attempts: await ctx.db.query("workAttempts").collect(),
run: await ctx.db.get(seeded.runId as any),
work: await ctx.db.get(seeded.workId as any),
}));
expect(state.attempts).toHaveLength(2);
expect(state.attempts[1]).toMatchObject({
number: 2,
status: "queued",
workspaceKey: "workspace-1",
});
expect(state.run?.status).toBe("running");
expect(state.work?.status).toBe("executing");
});
test("maps an authentication reason to a permanent failure", async () => {
const { t, ...seeded } = await seedRunningAttempt();
await t.mutation(api.workExecutionWorkflow.failAttempt, {
attemptId: seeded.attemptId,
reason: "Authentication",
retryable: false,
summary: "token rejected",
});
const state = await t.run(async (ctx) => ({
attempt: await ctx.db.get(seeded.attemptId as any),
run: await ctx.db.get(seeded.runId as any),
work: await ctx.db.get(seeded.workId as any),
}));
expect(state.attempt?.classification).toBe("PermanentFailure");
expect(state.attempt?.failureReason).toBe("Authentication");
expect(state.run?.terminalClassification).toBe("PermanentFailure");
expect(state.work?.status).toBe("failed");
});
});
describe("cancellation fencing", () => {
test("a cancelled attempt cannot later settle success", async () => {
const { t, ...seeded } = await seedRunningAttempt();
// Simulate cancellation marking the attempt/run terminal as Cancelled.
await t.run(async (ctx) => {
await ctx.db.patch(seeded.attemptId as any, {
classification: "Cancelled",
endedAt: 5,
status: "terminal",
summary: "Execution cancelled",
});
await ctx.db.patch(seeded.runId as any, {
endedAt: 5,
status: "cancelled",
terminalClassification: "Cancelled",
terminalSummary: "Execution cancelled",
});
});
// A late-arriving completion must be ignored.
await t.mutation(api.workExecutionWorkflow.completeAttempt, {
attemptId: seeded.attemptId,
result: successResult,
});
const attempt = await t.run((ctx) => ctx.db.get(seeded.attemptId as any));
expect(attempt?.classification).toBe("Cancelled");
expect(attempt?.status).toBe("terminal");
});
test("a late failure does not overwrite a cancelled run", async () => {
const { t, ...seeded } = await seedRunningAttempt();
await t.run(async (ctx) => {
await ctx.db.patch(seeded.attemptId as any, {
classification: "Cancelled",
endedAt: 5,
status: "terminal",
});
await ctx.db.patch(seeded.runId as any, { status: "cancelled" });
});
await t.mutation(api.workExecutionWorkflow.failAttempt, {
attemptId: seeded.attemptId,
reason: "HarnessFailed",
retryable: true,
summary: "late failure",
});
const attempt = await t.run((ctx) => ctx.db.get(seeded.attemptId as any));
expect(attempt?.classification).toBe("Cancelled");
});
});
describe("empty-change rejection", () => {
test("a no-op result with no changed files fails as permanent", async () => {
const { t, ...seeded } = await seedRunningAttempt();
await t.mutation(api.workExecutionWorkflow.completeAttempt, {
attemptId: seeded.attemptId,
result: {
...successResult,
changedFiles: [],
candidateRevision: "base123",
},
});
const state = await t.run(async (ctx) => ({
attempt: await ctx.db.get(seeded.attemptId as any),
work: await ctx.db.get(seeded.workId as any),
}));
expect(state.attempt?.classification).toBe("PermanentFailure");
expect(state.attempt?.failureReason).toBe("InvalidInput");
expect(state.work?.status).toBe("failed");
});
test("identical base and candidate revisions are rejected", async () => {
const { t, ...seeded } = await seedRunningAttempt();
await t.mutation(api.workExecutionWorkflow.completeAttempt, {
attemptId: seeded.attemptId,
result: {
...successResult,
baseRevision: "same",
candidateRevision: "same",
},
});
const attempt = await t.run((ctx) => ctx.db.get(seeded.attemptId as any));
expect(attempt?.classification).toBe("PermanentFailure");
});
});
describe("attempt re-entry", () => {
test("markAttemptRunning re-claims an already-running attempt", async () => {
const { t, ...seeded } = await seedRunningAttempt();
// First claim (already "running" from seed) must still succeed so a
// workflow replay can re-enter the same live attempt.
const first = await t.mutation(
api.workExecutionWorkflow.markAttemptRunning,
{ attemptId: seeded.attemptId }
);
expect(first).toBe(true);
const attempt = await t.run((ctx) => ctx.db.get(seeded.attemptId as any));
expect(attempt?.status).toBe("running");
expect(attempt?.leaseExpiresAt).toBeGreaterThan(0);
});
test("a terminal attempt is not re-runnable", async () => {
const { t, ...seeded } = await seedRunningAttempt({
attemptStatus: "terminal",
});
const result = await t.mutation(
api.workExecutionWorkflow.markAttemptRunning,
{ attemptId: seeded.attemptId }
);
expect(result).toBe(false);
});
test("a cancelled run cannot re-enter a queued attempt", async () => {
const { t, ...seeded } = await seedRunningAttempt({
attemptStatus: "queued",
runStatus: "cancelled",
});
const result = await t.mutation(
api.workExecutionWorkflow.markAttemptRunning,
{ attemptId: seeded.attemptId }
);
expect(result).toBe(false);
const attempt = await t.run((ctx) => ctx.db.get(seeded.attemptId as any));
expect(attempt?.status).toBe("queued");
});
});
describe("forge credential validation", () => {
const identity = { tokenIdentifier: "https://convex.test|forge-user" };
const seedForgableProject = async (
provider: "github" | "gitea",
sourceHost: string
) => {
const t = convexTest({ modules, schema }).withIdentity(identity);
const ids = await t.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://${sourceHost}/puter/zopu`,
organizationId,
repositoryPath: "puter/zopu",
sourceHost,
sourceUrl: `https://${sourceHost}/puter/zopu`,
updatedAt: 1,
});
const connectionId = await ctx.db.insert("gitConnections", {
connectedAt: 1,
credentialCiphertext: "x",
credentialIv: "y",
credentialKind: provider === "github" ? "oauth" : "token",
organizationId,
provider,
serverUrl: `https://${sourceHost}`,
updatedAt: 1,
username: "zopu",
});
return { connectionId, projectId };
});
return { t, ids };
};
test("rejects a git connection whose provider mismatches the project forge", async () => {
// Gitea token attached to a GitHub-hosted project.
const { t, ids } = await seedForgableProject("gitea", "github.com");
await expect(
t.mutation(api.gitConnectionData.attachToProject, {
connectionId: ids.connectionId,
projectId: ids.projectId,
})
).rejects.toThrow(/does not match this project's forge/u);
});
test("accepts a matching provider for the project forge", async () => {
const { t, ids } = await seedForgableProject("github", "github.com");
const result = await t.mutation(api.gitConnectionData.attachToProject, {
connectionId: ids.connectionId,
projectId: ids.projectId,
});
expect(result).toMatchObject({ attached: true });
});
});

267
packages/backend/convex/workExecutionWorkflow.ts
View File

@@ -1,4 +1,7 @@
import { defaultCodingKitV0 } from "@code/primitives/resolver";
import { WorkAttemptExecutionError } from "@code/primitives/execution-runtime";
import type { WorkAttemptExecutionErrorReason } from "@code/primitives/execution-runtime";
import { defaultCodingKitV0, resolveOutcome } from "@code/primitives/resolver";
import type { AttemptClassification } from "@code/primitives/resolver";
import { WorkflowManager } from "@convex-dev/workflow";
import type { WorkflowId } from "@convex-dev/workflow";
import { ConvexError, v } from "convex/values";
@@ -8,9 +11,68 @@ import type { Doc, Id } from "./_generated/dataModel";
import { internalMutation, internalQuery, mutation } from "./_generated/server";
import type { MutationCtx } from "./_generated/server";
import { requireProjectMember } from "./authz";
import { forgeForHost } from "./gitConnectionData";
export const workflow = new WorkflowManager(components.workflow);
// Lease window for a running real attempt. Long enough to outlast a single
// agent turn; the workflow's own retry/lease reconciliation covers crashes.
const LEASE_MS = 5 * 60_000;
// Runtime failure reason -> durable attempt classification. The Convex
// workflow stores classifications, not provider-specific reasons, so the
// resolver and Work lifecycle stay forge-agnostic. `retryable` from the
// runtime is carried alongside and consulted by the kit retry policy.
const FAILURE_REASON_CLASSIFICATION = {
Authentication: "PermanentFailure",
Cancelled: "Cancelled",
HarnessFailed: "RetryableFailure",
InvalidInput: "PermanentFailure",
ProviderUnavailable: "RetryableFailure",
RepositoryFailed: "PermanentFailure",
Timeout: "RetryableFailure",
} as const satisfies Record<
WorkAttemptExecutionErrorReason,
AttemptClassification
>;
export const classifyFailure = (
reason: WorkAttemptExecutionErrorReason
): AttemptClassification => FAILURE_REASON_CLASSIFICATION[reason];
// Normalize an error thrown from the agent action into the durable failure
// triple (reason/retryable/summary). WorkAttemptExecutionError is the
// classified runtime failure; anything else is a Convex/infrastructure error
// treated as a transient HarnessFailed so the workflow retry policy decides.
const toExecutionFailure = (
error: unknown
): {
message: string;
reason: WorkAttemptExecutionErrorReason;
retryable: boolean;
} =>
error instanceof WorkAttemptExecutionError
? {
message: error.message,
reason: error.reason,
retryable: error.retryable,
}
: {
message: error instanceof Error ? error.message : "Execution failed",
reason: "HarnessFailed",
retryable: true,
};
const failureReasonValues = v.union(
v.literal("Authentication"),
v.literal("Cancelled"),
v.literal("HarnessFailed"),
v.literal("InvalidInput"),
v.literal("ProviderUnavailable"),
v.literal("RepositoryFailed"),
v.literal("Timeout")
);
const resolveReadySlice = async (
ctx: MutationCtx,
work: Doc<"works">,
@@ -36,6 +98,34 @@ const resolveReadySlice = async (
return slice;
};
// Deployment prerequisite gate for real execution. Rejects before any
// workflow/sandbox starts when the project lacks an attached, forge-matched
// Git connection with a cloneable source URL and default branch. Returns the
// validated connection so the caller can pass it to the agent unchanged.
const validateProjectDeployment = async (
ctx: MutationCtx,
work: Doc<"works">
): Promise<Doc<"gitConnections">> => {
const project = await ctx.db.get(work.projectId);
if (!project) {
throw new ConvexError("Project not found");
}
if (!project.gitConnectionId) {
throw new ConvexError("Connect Git credentials to this project first");
}
const connection = await ctx.db.get(project.gitConnectionId);
if (!connection) {
throw new ConvexError("Git connection not found");
}
const expected = forgeForHost(project.sourceHost);
if (expected && connection.provider !== expected) {
throw new ConvexError(
`Git credential provider (${connection.provider}) does not match this project's forge (${expected})`
);
}
return connection;
};
export const execute = workflow
.define({ args: { attemptId: v.id("workAttempts") } })
.handler(async (step, args): Promise<void> => {
@@ -57,9 +147,12 @@ export const execute = workflow
},
});
} catch (error) {
const failure = toExecutionFailure(error);
await step.runMutation(internal.workExecutionWorkflow.failAttempt, {
attemptId: args.attemptId,
summary: error instanceof Error ? error.message : "Execution failed",
reason: failure.reason,
retryable: failure.retryable,
summary: failure.message,
});
}
});
@@ -91,10 +184,7 @@ export const startExecution = mutation({
) {
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");
}
await validateProjectDeployment(ctx, work);
const slice = await resolveReadySlice(ctx, work, args.sliceId);
const createdAt = Date.now();
const runId = await ctx.db.insert("workRuns", {
@@ -159,6 +249,12 @@ export const executionContext = internalQuery({
if (!connection) {
throw new ConvexError("Git connection not found");
}
const expectedForge = forgeForHost(project.sourceHost);
if (expectedForge && connection.provider !== expectedForge) {
throw new ConvexError(
`Git credential provider (${connection.provider}) does not match this project's forge (${expectedForge})`
);
}
return {
attempt,
connection,
@@ -179,11 +275,16 @@ 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") {
if (!attempt || attempt.status === "terminal") {
return false;
}
const run = await ctx.db.get(attempt.runId);
if (!run || run.status !== "running") {
return false;
}
await ctx.db.patch(attempt._id, {
startedAt: Date.now(),
leaseExpiresAt: Date.now() + LEASE_MS,
startedAt: attempt.startedAt ?? Date.now(),
status: "running",
});
return true;
@@ -193,7 +294,8 @@ export const markAttemptRunning = internalMutation({
const settleSliceAndWork = async (
ctx: MutationCtx,
run: Doc<"workRuns">,
succeeded: boolean
succeeded: boolean,
workStatus: Doc<"works">["status"]
) => {
if (!run.sliceRowId) {
return;
@@ -202,12 +304,18 @@ const settleSliceAndWork = async (
if (!slice) {
return;
}
await ctx.db.patch(slice._id, { status: succeeded ? "completed" : "ready" });
let sliceStatus: "blocked" | "completed" | "ready" = "ready";
if (succeeded) {
sliceStatus = "completed";
} else if (workStatus === "blocked") {
sliceStatus = "blocked";
}
await ctx.db.patch(slice._id, { status: sliceStatus });
const work = await ctx.db.get(run.workId);
if (!work) {
return;
}
let status: Doc<"works">["status"] = succeeded ? "completed" : "failed";
let status = workStatus;
if (succeeded) {
const slices = await ctx.db
.query("workSlices")
@@ -249,6 +357,8 @@ export const completeAttempt = internalMutation({
},
handler: async (ctx, args) => {
const attempt = await ctx.db.get(args.attemptId);
// Cancellation fencing: a terminal or cancelled attempt/run must never
// later settle success, even if the agent action raced to completion.
if (!attempt || attempt.status === "terminal") {
return;
}
@@ -257,6 +367,33 @@ export const completeAttempt = internalMutation({
if (!run || !work) {
throw new ConvexError("Execution records not found");
}
if (run.status === "terminal" || run.status === "cancelled") {
return;
}
// Empty-change rejection: a no-op result (no changed files, or identical
// base/candidate revisions) is not a successful implementation. Fail it
// as a permanent InvalidInput so the resolver does not mark Work done.
const noOp =
args.result.changedFiles.length === 0 ||
args.result.baseRevision === args.result.candidateRevision;
if (noOp) {
await ctx.db.patch(attempt._id, {
classification: "PermanentFailure",
endedAt: Date.now(),
failureReason: "InvalidInput",
leaseExpiresAt: undefined,
status: "terminal",
summary: "Execution produced no repository changes",
});
await ctx.db.patch(run._id, {
endedAt: Date.now(),
status: "terminal",
terminalClassification: "PermanentFailure",
terminalSummary: "Execution produced no repository changes",
});
await settleSliceAndWork(ctx, run, false, "failed");
return;
}
for (const item of args.result.events) {
await ctx.db.insert("workAttemptEvents", {
attemptId: attempt._id,
@@ -271,6 +408,7 @@ export const completeAttempt = internalMutation({
await ctx.db.patch(attempt._id, {
classification: "Succeeded",
endedAt,
leaseExpiresAt: undefined,
status: "terminal",
summary: args.result.summary,
});
@@ -309,7 +447,7 @@ export const completeAttempt = internalMutation({
}
: {}),
});
await settleSliceAndWork(ctx, run, true);
await settleSliceAndWork(ctx, run, true, "completed");
await ctx.db.insert("workEvents", {
createdAt: endedAt,
idempotencyKey: `real-run-completed:${run._id}`,
@@ -321,31 +459,107 @@ export const completeAttempt = internalMutation({
},
});
export const failAttempt = internalMutation({
args: { attemptId: v.id("workAttempts"), summary: v.string() },
export const launchAttemptWorkflow = internalMutation({
args: { attemptId: v.id("workAttempts") },
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) {
if (!run || run.status !== "running") {
return;
}
const workflowId: WorkflowId = await workflow.start(
ctx,
internal.workExecutionWorkflow.execute,
args
);
await ctx.db.patch(run._id, { workflowId });
},
});
export const failAttempt = internalMutation({
args: {
attemptId: v.id("workAttempts"),
reason: failureReasonValues,
retryable: v.boolean(),
summary: v.string(),
},
handler: async (ctx, args) => {
const attempt = await ctx.db.get(args.attemptId);
// Cancellation fencing: once an attempt or run is terminal/cancelled, a
// late failure from the workflow catch block must not overwrite the
// settled state. This also prevents a cancelled attempt from settling
// as a different classification after cancelExecution ran.
if (!attempt || attempt.status === "terminal") {
return;
}
const run = await ctx.db.get(attempt.runId);
if (!run || run.status === "terminal" || run.status === "cancelled") {
return;
}
const classification = classifyFailure(args.reason);
const endedAt = Date.now();
await ctx.db.patch(attempt._id, {
classification: "PermanentFailure",
classification,
endedAt,
failureReason: args.reason,
leaseExpiresAt: undefined,
status: "terminal",
summary: args.summary,
});
const outcome = resolveOutcome(
{ classification, retryable: args.retryable, summary: args.summary },
attempt.number,
defaultCodingKitV0.retryPolicy
);
await ctx.db.insert("resolverDecisions", {
attemptId: attempt._id,
attemptNumber: attempt.number,
classification,
createdAt: endedAt,
decision: outcome.kind,
...(outcome.kind === "terminal"
? { resultingWorkStatus: outcome.workStatus }
: {}),
runId: run._id,
summary: args.summary,
workId: attempt.workId,
});
if (outcome.kind === "retry") {
const nextAttemptId = await ctx.db.insert("workAttempts", {
number: attempt.number + 1,
runId: run._id,
status: "queued",
workId: attempt.workId,
workspaceKey: attempt.workspaceKey,
});
await ctx.scheduler.runAfter(
0,
internal.workExecutionWorkflow.launchAttemptWorkflow,
{ attemptId: nextAttemptId }
);
return;
}
await ctx.db.patch(run._id, {
endedAt,
status: "terminal",
terminalClassification: "PermanentFailure",
terminalClassification: classification,
terminalSummary: args.summary,
});
await settleSliceAndWork(ctx, run, false);
await settleSliceAndWork(ctx, run, false, outcome.workStatus);
const work = await ctx.db.get(run.workId);
if (work) {
await ctx.db.insert("workEvents", {
createdAt: endedAt,
idempotencyKey: `real-run-failed:${run._id}`,
kind: "run.completed",
payloadJson: JSON.stringify({ classification }),
referenceId: String(run._id),
workId: work._id,
});
}
},
});
@@ -375,15 +589,30 @@ export const cancelExecution = mutation({
0,
internal.workExecutionAgent.cancelAttempt,
{
attemptId: String(attempt._id),
workspaceKey: attempt.workspaceKey,
}
);
const cancelledAt = Date.now();
await ctx.db.patch(attempt._id, {
classification: "Cancelled",
endedAt: Date.now(),
endedAt: cancelledAt,
failureReason: "Cancelled",
leaseExpiresAt: undefined,
status: "terminal",
summary: "Execution cancelled",
});
await ctx.db.insert("resolverDecisions", {
attemptId: attempt._id,
attemptNumber: attempt.number,
classification: "Cancelled",
createdAt: cancelledAt,
decision: "terminal",
resultingWorkStatus: "ready",
runId: run._id,
summary: "Execution cancelled",
workId: run.workId,
});
}
await ctx.db.patch(run._id, {
endedAt: Date.now(),