895 lines
27 KiB
TypeScript
895 lines
27 KiB
TypeScript
import { env } from "@code/env/convex";
|
|
import { WorkAttemptExecutionError } from "@code/primitives/execution-runtime";
|
|
import type { WorkAttemptExecutionErrorReason } from "@code/primitives/execution-runtime";
|
|
import {
|
|
CREDENTIAL_FRESHNESS_MS,
|
|
providerForHost as forgeForHost,
|
|
} from "@code/primitives/git-provider";
|
|
import { defaultCodingKitV0, resolveOutcome } from "@code/primitives/resolver";
|
|
import type { AttemptClassification } from "@code/primitives/resolver";
|
|
import type { WorkEventKind } from "@code/primitives/work";
|
|
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,
|
|
query,
|
|
} from "./_generated/server";
|
|
import type { MutationCtx } from "./_generated/server";
|
|
import { requireProjectMember } from "./authz";
|
|
|
|
/** Dual auth: service token for agent calls, user auth for web calls. */
|
|
const authorizeExecution = async (
|
|
ctx: MutationCtx,
|
|
token: string | undefined,
|
|
projectId: Id<"projects">
|
|
) => {
|
|
if (token !== undefined) {
|
|
if (token !== env.FLUE_DB_TOKEN) {
|
|
throw new ConvexError("Invalid agent control token");
|
|
}
|
|
return;
|
|
}
|
|
await requireProjectMember(ctx, projectId);
|
|
};
|
|
|
|
export const workflow = new WorkflowManager(components.workflow);
|
|
|
|
const LEASE_MS = 5 * 60_000;
|
|
const now = (): number => Date.now();
|
|
|
|
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];
|
|
|
|
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 appendWorkEvent = async (
|
|
ctx: MutationCtx,
|
|
workId: Id<"works">,
|
|
kind: WorkEventKind,
|
|
idempotencyKey: string,
|
|
referenceId?: string,
|
|
payloadJson?: string
|
|
) => {
|
|
const existing = await ctx.db
|
|
.query("workEvents")
|
|
.withIndex("by_work_and_idempotencyKey", (q) =>
|
|
q.eq("workId", workId).eq("idempotencyKey", idempotencyKey)
|
|
)
|
|
.unique();
|
|
if (existing) {
|
|
return existing._id;
|
|
}
|
|
return await ctx.db.insert("workEvents", {
|
|
createdAt: now(),
|
|
idempotencyKey,
|
|
kind,
|
|
workId,
|
|
...(referenceId ? { referenceId } : {}),
|
|
...(payloadJson ? { payloadJson } : {}),
|
|
});
|
|
};
|
|
|
|
const resolveReadySlice = async (
|
|
ctx: MutationCtx,
|
|
work: Doc<"works">,
|
|
sliceId?: string
|
|
) => {
|
|
if (work.designVersion === undefined) {
|
|
throw new ConvexError("Work has no current 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;
|
|
};
|
|
|
|
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 provider (${expected})`
|
|
);
|
|
}
|
|
if (connection.state !== undefined && connection.state !== "active") {
|
|
throw new ConvexError(
|
|
`Git connection is ${connection.state}; verify or reconnect credentials before execution`
|
|
);
|
|
}
|
|
const currentTime = now();
|
|
if (
|
|
connection.lastVerifiedAt === undefined ||
|
|
currentTime - connection.lastVerifiedAt > CREDENTIAL_FRESHNESS_MS
|
|
) {
|
|
throw new ConvexError(
|
|
"Git credentials have not been verified recently; run a connection health check before execution"
|
|
);
|
|
}
|
|
return connection;
|
|
};
|
|
|
|
const settleSliceAndWork = async (
|
|
ctx: MutationCtx,
|
|
run: Doc<"workRuns">,
|
|
succeeded: boolean,
|
|
workStatus: Doc<"works">["status"]
|
|
) => {
|
|
if (!run.sliceRowId) {
|
|
return;
|
|
}
|
|
const slice = await ctx.db.get(run.sliceRowId);
|
|
if (!slice) {
|
|
return;
|
|
}
|
|
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 = workStatus;
|
|
if (succeeded && workStatus === "completed") {
|
|
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: now() });
|
|
};
|
|
|
|
// --- Workflow definition ---
|
|
|
|
export const execute = workflow
|
|
.define({ args: { attemptId: v.id("workAttempts") } })
|
|
.handler(async (step, args): Promise<void> => {
|
|
try {
|
|
const result = await step.runAction(
|
|
internal.agentBackend.executeAttempt,
|
|
args,
|
|
{ retry: true }
|
|
);
|
|
await step.runMutation(internal.workExecution.completeAttempt, {
|
|
attemptId: args.attemptId,
|
|
result: {
|
|
...result,
|
|
changedFiles: [...result.changedFiles],
|
|
events: result.events.map((item: any) => ({
|
|
...item,
|
|
metadata: { ...item.metadata },
|
|
})),
|
|
},
|
|
});
|
|
} catch (error) {
|
|
const failure = toExecutionFailure(error);
|
|
await step.runMutation(internal.workExecution.failAttempt, {
|
|
attemptId: args.attemptId,
|
|
reason: failure.reason,
|
|
retryable: failure.retryable,
|
|
summary: failure.message,
|
|
});
|
|
}
|
|
});
|
|
|
|
// --- Public commands ---
|
|
|
|
export const start = mutation({
|
|
args: {
|
|
organizationId: v.optional(v.id("organizations")),
|
|
sliceId: v.optional(v.string()),
|
|
token: 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");
|
|
}
|
|
if (
|
|
args.token !== undefined &&
|
|
args.organizationId !== work.organizationId
|
|
) {
|
|
throw new ConvexError("Organization mismatch");
|
|
}
|
|
await authorizeExecution(ctx, args.token, work.projectId);
|
|
if (work.status !== "ready") {
|
|
throw new ConvexError("Work must be Ready before execution");
|
|
}
|
|
if (
|
|
work.definitionVersion === undefined ||
|
|
work.designVersion === undefined
|
|
) {
|
|
throw new ConvexError("Work must have a current Definition and Design");
|
|
}
|
|
await validateProjectDeployment(ctx, work);
|
|
const slice = await resolveReadySlice(ctx, work, args.sliceId);
|
|
const createdAt = now();
|
|
const runId = await ctx.db.insert("workRuns", {
|
|
createdAt,
|
|
designVersion: slice.designVersion,
|
|
kitId: defaultCodingKitV0.id,
|
|
kitVersion: defaultCodingKitV0.version,
|
|
sliceId: slice.sliceId,
|
|
sliceRowId: slice._id,
|
|
status: "running",
|
|
workId: work._id,
|
|
});
|
|
const attemptId = await ctx.db.insert("workAttempts", {
|
|
number: 1,
|
|
runId,
|
|
status: "queued",
|
|
workId: work._id,
|
|
workspaceKey: `work-${work._id}-attempt-pending`,
|
|
});
|
|
const workspaceKey = `work-${work._id}-attempt-${attemptId}`;
|
|
await ctx.db.patch(attemptId, { workspaceKey });
|
|
const workflowId: WorkflowId = await workflow.start(
|
|
ctx,
|
|
internal.workExecution.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 appendWorkEvent(
|
|
ctx,
|
|
work._id,
|
|
"run.started",
|
|
`run-started:${runId}`,
|
|
String(runId)
|
|
);
|
|
return { attemptId, runId, workflowId };
|
|
},
|
|
});
|
|
|
|
export const cancel = mutation({
|
|
args: {
|
|
organizationId: v.optional(v.id("organizations")),
|
|
runId: v.id("workRuns"),
|
|
token: v.optional(v.string()),
|
|
},
|
|
handler: async (ctx, args) => {
|
|
const run = await ctx.db.get(args.runId);
|
|
if (!run) {
|
|
throw new ConvexError("Run not found");
|
|
}
|
|
const work = await ctx.db.get(run.workId);
|
|
if (!work) {
|
|
throw new ConvexError("Work not found");
|
|
}
|
|
if (
|
|
args.token !== undefined &&
|
|
args.organizationId !== work.organizationId
|
|
) {
|
|
throw new ConvexError("Organization mismatch");
|
|
}
|
|
await authorizeExecution(ctx, args.token, work.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 activeAttempt = attempts.find(
|
|
(candidate) => candidate.status !== "terminal"
|
|
);
|
|
if (activeAttempt?.workspaceKey) {
|
|
await ctx.scheduler.runAfter(0, internal.agentBackend.cancelAttempt, {
|
|
attemptId: String(activeAttempt._id),
|
|
workspaceKey: activeAttempt.workspaceKey,
|
|
});
|
|
const cancelledAt = now();
|
|
await ctx.db.patch(activeAttempt._id, {
|
|
classification: "Cancelled",
|
|
endedAt: cancelledAt,
|
|
failureReason: "Cancelled",
|
|
leaseExpiresAt: undefined,
|
|
status: "terminal",
|
|
summary: "Execution cancelled",
|
|
});
|
|
await ctx.db.insert("resolverDecisions", {
|
|
attemptId: activeAttempt._id,
|
|
attemptNumber: activeAttempt.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: now(),
|
|
status: "cancelled",
|
|
terminalClassification: "Cancelled",
|
|
terminalSummary: "Execution cancelled",
|
|
});
|
|
if (run.sliceRowId) {
|
|
await ctx.db.patch(run.sliceRowId, { status: "ready" });
|
|
}
|
|
await ctx.db.patch(work._id, { status: "ready", updatedAt: now() });
|
|
return { cancelled: true };
|
|
},
|
|
});
|
|
|
|
export const retry = mutation({
|
|
args: {
|
|
organizationId: v.optional(v.id("organizations")),
|
|
runId: v.id("workRuns"),
|
|
token: v.optional(v.string()),
|
|
},
|
|
handler: async (ctx, args) => {
|
|
const run = await ctx.db.get(args.runId);
|
|
if (!run) {
|
|
throw new ConvexError("Run not found");
|
|
}
|
|
const work = await ctx.db.get(run.workId);
|
|
if (!work) {
|
|
throw new ConvexError("Work not found");
|
|
}
|
|
if (
|
|
args.token !== undefined &&
|
|
args.organizationId !== work.organizationId
|
|
) {
|
|
throw new ConvexError("Organization mismatch");
|
|
}
|
|
await authorizeExecution(ctx, args.token, work.projectId);
|
|
if (run.status !== "terminal" && run.status !== "cancelled") {
|
|
throw new ConvexError("Only a terminal or cancelled Run can be retried");
|
|
}
|
|
if (run.terminalClassification === "Succeeded") {
|
|
throw new ConvexError("A successful Run cannot be retried");
|
|
}
|
|
const attempts = await ctx.db
|
|
.query("workAttempts")
|
|
.withIndex("by_runId_and_number", (q) => q.eq("runId", run._id))
|
|
.collect();
|
|
const nextNumber = attempts.length + 1;
|
|
const attemptId = await ctx.db.insert("workAttempts", {
|
|
number: nextNumber,
|
|
runId: run._id,
|
|
status: "queued",
|
|
workId: work._id,
|
|
workspaceKey: `work-${work._id}-attempt-pending`,
|
|
});
|
|
const workspaceKey = `work-${work._id}-attempt-${attemptId}`;
|
|
await ctx.db.patch(attemptId, { workspaceKey });
|
|
await ctx.db.patch(run._id, {
|
|
endedAt: undefined,
|
|
status: "running",
|
|
terminalClassification: undefined,
|
|
terminalSummary: undefined,
|
|
});
|
|
await ctx.db.patch(work._id, { status: "executing", updatedAt: now() });
|
|
const workflowId: WorkflowId = await workflow.start(
|
|
ctx,
|
|
internal.workExecution.execute,
|
|
{ attemptId }
|
|
);
|
|
await ctx.db.patch(run._id, { workflowId });
|
|
return { runId: run._id };
|
|
},
|
|
});
|
|
|
|
// --- Internal mutations ---
|
|
|
|
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");
|
|
}
|
|
const expectedForge = forgeForHost(project.sourceHost);
|
|
if (expectedForge && connection.provider !== expectedForge) {
|
|
throw new ConvexError(
|
|
`Git credential provider (${connection.provider}) does not match this project's provider (${expectedForge})`
|
|
);
|
|
}
|
|
if (connection.state !== undefined && connection.state !== "active") {
|
|
throw new ConvexError(
|
|
`Git connection is ${connection.state}; execution is blocked`
|
|
);
|
|
}
|
|
return {
|
|
attempt,
|
|
connection,
|
|
project,
|
|
prompt: [
|
|
`Implement this 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 === "terminal") {
|
|
return false;
|
|
}
|
|
const run = await ctx.db.get(attempt.runId);
|
|
if (!run || run.status !== "running") {
|
|
return false;
|
|
}
|
|
await ctx.db.patch(attempt._id, {
|
|
leaseExpiresAt: now() + LEASE_MS,
|
|
startedAt: attempt.startedAt ?? now(),
|
|
status: "running",
|
|
});
|
|
return true;
|
|
},
|
|
});
|
|
|
|
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");
|
|
}
|
|
if (run.status === "terminal" || run.status === "cancelled") {
|
|
return;
|
|
}
|
|
// Empty-change rejection: a no-op result is not a successful implementation.
|
|
const noOp =
|
|
args.result.changedFiles.length === 0 ||
|
|
args.result.baseRevision === args.result.candidateRevision;
|
|
if (noOp) {
|
|
await ctx.db.patch(attempt._id, {
|
|
classification: "PermanentFailure",
|
|
endedAt: now(),
|
|
failureReason: "InvalidInput",
|
|
leaseExpiresAt: undefined,
|
|
status: "terminal",
|
|
summary: "Execution produced no repository changes",
|
|
});
|
|
await ctx.db.patch(run._id, {
|
|
endedAt: 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,
|
|
kind: item.kind,
|
|
message: item.message,
|
|
metadataJson: JSON.stringify(item.metadata),
|
|
occurredAt: item.occurredAt,
|
|
sequence: item.sequence,
|
|
});
|
|
}
|
|
const endedAt = now();
|
|
await ctx.db.patch(attempt._id, {
|
|
classification: "Succeeded",
|
|
endedAt,
|
|
leaseExpiresAt: undefined,
|
|
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: `diff:${attempt._id}`,
|
|
kind: "diff",
|
|
metadataJson: JSON.stringify({ changedFiles: args.result.changedFiles }),
|
|
organizationId: work.organizationId,
|
|
producer: "agentos",
|
|
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))}`,
|
|
}
|
|
: {}),
|
|
});
|
|
// A successful AgentOS response alone must never mark Work complete.
|
|
// Move to verifying so an independent step can confirm the candidate.
|
|
await settleSliceAndWork(ctx, run, false, "verifying");
|
|
await appendWorkEvent(
|
|
ctx,
|
|
work._id,
|
|
"verification.started",
|
|
`verification-started:${run._id}`,
|
|
String(run._id)
|
|
);
|
|
},
|
|
});
|
|
|
|
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 || run.status !== "running") {
|
|
return;
|
|
}
|
|
const workflowId: WorkflowId = await workflow.start(
|
|
ctx,
|
|
internal.workExecution.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);
|
|
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 = now();
|
|
await ctx.db.patch(attempt._id, {
|
|
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.workExecution.launchAttemptWorkflow,
|
|
{ attemptId: nextAttemptId }
|
|
);
|
|
return;
|
|
}
|
|
await ctx.db.patch(run._id, {
|
|
endedAt,
|
|
status: "terminal",
|
|
terminalClassification: classification,
|
|
terminalSummary: args.summary,
|
|
});
|
|
await settleSliceAndWork(ctx, run, false, outcome.workStatus);
|
|
const work = await ctx.db.get(run.workId);
|
|
if (work) {
|
|
await appendWorkEvent(
|
|
ctx,
|
|
work._id,
|
|
"run.completed",
|
|
`run-failed:${run._id}`,
|
|
String(run._id),
|
|
JSON.stringify({ classification })
|
|
);
|
|
}
|
|
},
|
|
});
|
|
|
|
// --- Lease reconciliation (cron target) ---
|
|
|
|
export const reconcileExpiredAttempts = internalMutation({
|
|
args: {},
|
|
handler: async (ctx) => {
|
|
const timestamp = now();
|
|
const [claimed, running] = await Promise.all([
|
|
ctx.db
|
|
.query("workAttempts")
|
|
.withIndex("by_status_and_leaseExpiresAt", (q) =>
|
|
q.eq("status", "claimed").lt("leaseExpiresAt", timestamp)
|
|
)
|
|
.collect(),
|
|
ctx.db
|
|
.query("workAttempts")
|
|
.withIndex("by_status_and_leaseExpiresAt", (q) =>
|
|
q.eq("status", "running").lt("leaseExpiresAt", timestamp)
|
|
)
|
|
.collect(),
|
|
]);
|
|
const active = [...claimed, ...running];
|
|
let reconciled = 0;
|
|
for (const attempt of active) {
|
|
if (
|
|
(attempt.status === "claimed" || attempt.status === "running") &&
|
|
attempt.leaseExpiresAt !== undefined &&
|
|
attempt.leaseExpiresAt < now()
|
|
) {
|
|
await ctx.db.patch(attempt._id, {
|
|
classification: "Blocked",
|
|
endedAt: now(),
|
|
leaseExpiresAt: undefined,
|
|
leaseOwner: undefined,
|
|
status: "terminal",
|
|
summary: "Attempt lease expired and was reconciled",
|
|
});
|
|
const run = await ctx.db.get(attempt.runId);
|
|
await appendWorkEvent(
|
|
ctx,
|
|
attempt.workId,
|
|
"attempt.reconciled",
|
|
`attempt-reconciled:${attempt._id}`,
|
|
String(attempt._id),
|
|
JSON.stringify({ attemptNumber: attempt.number })
|
|
);
|
|
reconciled += 1;
|
|
if (!run) {
|
|
continue;
|
|
}
|
|
const attempts = await ctx.db
|
|
.query("workAttempts")
|
|
.withIndex("by_runId_and_number", (q) => q.eq("runId", run._id))
|
|
.collect();
|
|
const withinBudget =
|
|
attempts.length < defaultCodingKitV0.retryPolicy.maxAttempts;
|
|
const retryRun = run.status === "running" && withinBudget;
|
|
await ctx.db.insert("resolverDecisions", {
|
|
attemptId: attempt._id,
|
|
attemptNumber: attempt.number,
|
|
classification: "Blocked",
|
|
createdAt: now(),
|
|
decision: retryRun ? "retry" : "terminal",
|
|
resultingWorkStatus: retryRun ? 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: retryRun ? "retry" : "terminal",
|
|
})
|
|
);
|
|
if (retryRun) {
|
|
const nextAttemptId = await ctx.db.insert("workAttempts", {
|
|
number: attempts.length + 1,
|
|
runId: run._id,
|
|
status: "queued",
|
|
workId: attempt.workId,
|
|
workspaceKey: attempt.workspaceKey,
|
|
});
|
|
await ctx.scheduler.runAfter(
|
|
0,
|
|
internal.workExecution.launchAttemptWorkflow,
|
|
{ attemptId: nextAttemptId }
|
|
);
|
|
} else {
|
|
await ctx.db.patch(run._id, {
|
|
endedAt: now(),
|
|
status: "terminal",
|
|
terminalClassification: "Blocked",
|
|
terminalSummary: "Run reconciled after expired lease",
|
|
});
|
|
const work = await ctx.db.get(run.workId);
|
|
if (work) {
|
|
await ctx.db.patch(work._id, {
|
|
status: "blocked",
|
|
updatedAt: now(),
|
|
});
|
|
}
|
|
}
|
|
}
|
|
}
|
|
return { reconciled };
|
|
},
|
|
});
|
|
|
|
export const listRunEvents = query({
|
|
args: { attemptId: v.id("workAttempts") },
|
|
handler: async (ctx, args) => {
|
|
const attempt = await ctx.db.get(args.attemptId);
|
|
if (!attempt) {
|
|
return [];
|
|
}
|
|
const work = await ctx.db.get(attempt.workId);
|
|
if (!work) {
|
|
return [];
|
|
}
|
|
return await ctx.db
|
|
.query("workAttemptEvents")
|
|
.withIndex("by_attempt_and_sequence", (q) =>
|
|
q.eq("attemptId", attempt._id)
|
|
)
|
|
.collect();
|
|
},
|
|
});
|