feat: add durable AgentOS slice execution
This commit is contained in:
17
packages/backend/convex/_generated/api.d.ts
vendored
17
packages/backend/convex/_generated/api.d.ts
vendored
@@ -11,7 +11,10 @@
|
||||
import type * as auth from "../auth.js";
|
||||
import type * as authz from "../authz.js";
|
||||
import type * as conversationMessages from "../conversationMessages.js";
|
||||
import type * as crons from "../crons.js";
|
||||
import type * as fluePersistence from "../fluePersistence.js";
|
||||
import type * as gitConnectionData from "../gitConnectionData.js";
|
||||
import type * as gitConnections from "../gitConnections.js";
|
||||
import type * as healthCheck from "../healthCheck.js";
|
||||
import type * as http from "../http.js";
|
||||
import type * as organizations from "../organizations.js";
|
||||
@@ -19,6 +22,11 @@ import type * as privateData from "../privateData.js";
|
||||
import type * as projects from "../projects.js";
|
||||
import type * as publicGit from "../publicGit.js";
|
||||
import type * as signalRouting from "../signalRouting.js";
|
||||
import type * as workArtifacts from "../workArtifacts.js";
|
||||
import type * as workExecution from "../workExecution.js";
|
||||
import type * as workExecutionAgent from "../workExecutionAgent.js";
|
||||
import type * as workExecutionWorkflow from "../workExecutionWorkflow.js";
|
||||
import type * as workPlanning from "../workPlanning.js";
|
||||
import type * as works from "../works.js";
|
||||
|
||||
import type {
|
||||
@@ -31,7 +39,10 @@ declare const fullApi: ApiFromModules<{
|
||||
auth: typeof auth;
|
||||
authz: typeof authz;
|
||||
conversationMessages: typeof conversationMessages;
|
||||
crons: typeof crons;
|
||||
fluePersistence: typeof fluePersistence;
|
||||
gitConnectionData: typeof gitConnectionData;
|
||||
gitConnections: typeof gitConnections;
|
||||
healthCheck: typeof healthCheck;
|
||||
http: typeof http;
|
||||
organizations: typeof organizations;
|
||||
@@ -39,6 +50,11 @@ declare const fullApi: ApiFromModules<{
|
||||
projects: typeof projects;
|
||||
publicGit: typeof publicGit;
|
||||
signalRouting: typeof signalRouting;
|
||||
workArtifacts: typeof workArtifacts;
|
||||
workExecution: typeof workExecution;
|
||||
workExecutionAgent: typeof workExecutionAgent;
|
||||
workExecutionWorkflow: typeof workExecutionWorkflow;
|
||||
workPlanning: typeof workPlanning;
|
||||
works: typeof works;
|
||||
}>;
|
||||
|
||||
@@ -70,4 +86,5 @@ export declare const internal: FilterApi<
|
||||
|
||||
export declare const components: {
|
||||
betterAuth: import("@convex-dev/better-auth/_generated/component.js").ComponentApi<"betterAuth">;
|
||||
workflow: import("@convex-dev/workflow/_generated/component.js").ComponentApi<"workflow">;
|
||||
};
|
||||
|
||||
@@ -25,10 +25,14 @@ import type { DataModel } from "./dataModel.js";
|
||||
* Typesafe environment variables declared in `convex.config.ts`.
|
||||
*/
|
||||
type Env = {
|
||||
readonly AGENT_BACKEND_URL: string | undefined;
|
||||
readonly FLUE_DB_TOKEN: string;
|
||||
readonly FLUE_URL: string | undefined;
|
||||
readonly GITEA_TOKEN: string | undefined;
|
||||
readonly GITEA_URL: string | undefined;
|
||||
readonly GITHUB_CLIENT_ID: string | undefined;
|
||||
readonly GITHUB_CLIENT_SECRET: string | undefined;
|
||||
readonly GIT_CREDENTIAL_ENCRYPTION_KEY: string | undefined;
|
||||
readonly NATIVE_APP_URL: string | undefined;
|
||||
readonly SITE_URL: string;
|
||||
};
|
||||
|
||||
@@ -30,6 +30,15 @@ const createAuth = (ctx: GenericCtx<DataModel>) =>
|
||||
jwksRotateOnTokenGenerationError: true,
|
||||
}),
|
||||
],
|
||||
socialProviders:
|
||||
env.GITHUB_CLIENT_ID && env.GITHUB_CLIENT_SECRET
|
||||
? {
|
||||
github: {
|
||||
clientId: env.GITHUB_CLIENT_ID,
|
||||
clientSecret: env.GITHUB_CLIENT_SECRET,
|
||||
},
|
||||
}
|
||||
: {},
|
||||
trustedOrigins: [siteUrl, nativeAppUrl, "exp://"],
|
||||
});
|
||||
|
||||
|
||||
@@ -1,17 +1,23 @@
|
||||
import betterAuth from "@convex-dev/better-auth/convex.config";
|
||||
import workflow from "@convex-dev/workflow/convex.config.js";
|
||||
import { defineApp } from "convex/server";
|
||||
import { v } from "convex/values";
|
||||
|
||||
const app = defineApp({
|
||||
env: {
|
||||
AGENT_BACKEND_URL: v.optional(v.string()),
|
||||
FLUE_DB_TOKEN: v.string(),
|
||||
FLUE_URL: v.optional(v.string()),
|
||||
GITEA_TOKEN: v.optional(v.string()),
|
||||
GITEA_URL: v.optional(v.string()),
|
||||
GITHUB_CLIENT_ID: v.optional(v.string()),
|
||||
GITHUB_CLIENT_SECRET: v.optional(v.string()),
|
||||
GIT_CREDENTIAL_ENCRYPTION_KEY: v.optional(v.string()),
|
||||
NATIVE_APP_URL: v.optional(v.string()),
|
||||
SITE_URL: v.string(),
|
||||
},
|
||||
});
|
||||
app.use(betterAuth);
|
||||
app.use(workflow);
|
||||
|
||||
export default app;
|
||||
|
||||
119
packages/backend/convex/gitConnectionData.ts
Normal file
119
packages/backend/convex/gitConnectionData.ts
Normal file
@@ -0,0 +1,119 @@
|
||||
import { ConvexError, v } from "convex/values";
|
||||
|
||||
import { internalMutation, mutation, query } from "./_generated/server";
|
||||
import { requireCurrentOrganization, requireProjectMember } from "./authz";
|
||||
|
||||
export const persist = internalMutation({
|
||||
args: {
|
||||
credentialCiphertext: v.string(),
|
||||
credentialIv: v.string(),
|
||||
credentialKind: v.union(v.literal("oauth"), v.literal("token")),
|
||||
provider: v.union(v.literal("github"), v.literal("gitea")),
|
||||
serverUrl: v.string(),
|
||||
userId: v.string(),
|
||||
username: v.optional(v.string()),
|
||||
},
|
||||
handler: async (ctx, args) => {
|
||||
const organization = await ctx.db
|
||||
.query("organizations")
|
||||
.withIndex("by_createdBy_and_kind", (q) =>
|
||||
q.eq("createdBy", args.userId).eq("kind", "personal")
|
||||
)
|
||||
.unique();
|
||||
if (!organization) {
|
||||
throw new ConvexError("Organization not found");
|
||||
}
|
||||
const existing = await ctx.db
|
||||
.query("gitConnections")
|
||||
.withIndex("by_organizationId_and_provider_and_serverUrl", (q) =>
|
||||
q
|
||||
.eq("organizationId", organization._id)
|
||||
.eq("provider", args.provider)
|
||||
.eq("serverUrl", args.serverUrl)
|
||||
)
|
||||
.unique();
|
||||
const timestamp = Date.now();
|
||||
if (existing) {
|
||||
await ctx.db.patch(existing._id, {
|
||||
credentialCiphertext: args.credentialCiphertext,
|
||||
credentialIv: args.credentialIv,
|
||||
credentialKind: args.credentialKind,
|
||||
updatedAt: timestamp,
|
||||
username: args.username,
|
||||
});
|
||||
return existing._id;
|
||||
}
|
||||
return await ctx.db.insert("gitConnections", {
|
||||
connectedAt: timestamp,
|
||||
credentialCiphertext: args.credentialCiphertext,
|
||||
credentialIv: args.credentialIv,
|
||||
credentialKind: args.credentialKind,
|
||||
organizationId: organization._id,
|
||||
provider: args.provider,
|
||||
serverUrl: args.serverUrl,
|
||||
updatedAt: timestamp,
|
||||
username: args.username,
|
||||
});
|
||||
},
|
||||
});
|
||||
|
||||
export const list = query({
|
||||
args: {},
|
||||
handler: async (ctx) => {
|
||||
const { organizationId } = await requireCurrentOrganization(ctx);
|
||||
const connections = await ctx.db
|
||||
.query("gitConnections")
|
||||
.withIndex("by_organizationId", (q) =>
|
||||
q.eq("organizationId", organizationId)
|
||||
)
|
||||
.collect();
|
||||
return connections.map((connection) => ({
|
||||
connectedAt: connection.connectedAt,
|
||||
credentialKind: connection.credentialKind,
|
||||
id: String(connection._id),
|
||||
provider: connection.provider,
|
||||
serverUrl: connection.serverUrl,
|
||||
username: connection.username,
|
||||
}));
|
||||
},
|
||||
});
|
||||
|
||||
export const getForProject = query({
|
||||
args: { projectId: v.id("projects") },
|
||||
handler: async (ctx, args) => {
|
||||
await requireProjectMember(ctx, args.projectId);
|
||||
const project = await ctx.db.get(args.projectId);
|
||||
const connection = project?.gitConnectionId
|
||||
? await ctx.db.get(project.gitConnectionId)
|
||||
: null;
|
||||
return connection
|
||||
? {
|
||||
connectedAt: connection.connectedAt,
|
||||
credentialKind: connection.credentialKind,
|
||||
id: String(connection._id),
|
||||
provider: connection.provider,
|
||||
serverUrl: connection.serverUrl,
|
||||
username: connection.username,
|
||||
}
|
||||
: null;
|
||||
},
|
||||
});
|
||||
|
||||
export const attachToProject = mutation({
|
||||
args: {
|
||||
connectionId: v.id("gitConnections"),
|
||||
projectId: v.id("projects"),
|
||||
},
|
||||
handler: async (ctx, args) => {
|
||||
const { organizationId } = await requireProjectMember(ctx, args.projectId);
|
||||
const connection = await ctx.db.get(args.connectionId);
|
||||
if (!connection || connection.organizationId !== organizationId) {
|
||||
throw new ConvexError("Git connection not found");
|
||||
}
|
||||
await ctx.db.patch(args.projectId, {
|
||||
gitConnectionId: connection._id,
|
||||
updatedAt: Date.now(),
|
||||
});
|
||||
return { attached: true };
|
||||
},
|
||||
});
|
||||
124
packages/backend/convex/gitConnections.ts
Normal file
124
packages/backend/convex/gitConnections.ts
Normal file
@@ -0,0 +1,124 @@
|
||||
"use node";
|
||||
|
||||
import { env } from "@code/env/convex";
|
||||
import { decodeGitConnectionInput } from "@code/primitives/execution-runtime";
|
||||
import { ConvexError, v } from "convex/values";
|
||||
import { Effect } from "effect";
|
||||
|
||||
import { internal } from "./_generated/api";
|
||||
import type { Id } from "./_generated/dataModel";
|
||||
import { action } from "./_generated/server";
|
||||
import { authComponent, createAuth } from "./auth";
|
||||
|
||||
const encryptionKey = async (): Promise<CryptoKey> => {
|
||||
if (!env.GIT_CREDENTIAL_ENCRYPTION_KEY) {
|
||||
throw new ConvexError("Git credential encryption is not configured");
|
||||
}
|
||||
const bytes = Buffer.from(env.GIT_CREDENTIAL_ENCRYPTION_KEY, "base64url");
|
||||
if (bytes.byteLength !== 32) {
|
||||
throw new ConvexError("Git credential encryption key must be 32 bytes");
|
||||
}
|
||||
return await crypto.subtle.importKey("raw", bytes, "AES-GCM", false, [
|
||||
"encrypt",
|
||||
"decrypt",
|
||||
]);
|
||||
};
|
||||
|
||||
const encryptCredential = async (credential: string) => {
|
||||
const iv = crypto.getRandomValues(new Uint8Array(12));
|
||||
const encrypted = await crypto.subtle.encrypt(
|
||||
{ iv, name: "AES-GCM" },
|
||||
await encryptionKey(),
|
||||
new TextEncoder().encode(credential)
|
||||
);
|
||||
return {
|
||||
credentialCiphertext: Buffer.from(encrypted).toString("base64url"),
|
||||
credentialIv: Buffer.from(iv).toString("base64url"),
|
||||
};
|
||||
};
|
||||
|
||||
export const decryptCredential = async (
|
||||
credentialCiphertext: string,
|
||||
credentialIv: string
|
||||
): Promise<string> => {
|
||||
const decrypted = await crypto.subtle.decrypt(
|
||||
{
|
||||
iv: Buffer.from(credentialIv, "base64url"),
|
||||
name: "AES-GCM",
|
||||
},
|
||||
await encryptionKey(),
|
||||
Buffer.from(credentialCiphertext, "base64url")
|
||||
);
|
||||
return new TextDecoder().decode(decrypted);
|
||||
};
|
||||
|
||||
export const connectGitea = action({
|
||||
args: {
|
||||
serverUrl: v.string(),
|
||||
token: v.string(),
|
||||
username: v.optional(v.string()),
|
||||
},
|
||||
handler: async (
|
||||
ctx,
|
||||
args
|
||||
): Promise<{ connectionId: Id<"gitConnections"> }> => {
|
||||
const userId = await ctx.auth.getUserIdentity().then((identity) => {
|
||||
if (!identity) {
|
||||
throw new ConvexError("Authentication required");
|
||||
}
|
||||
return identity.tokenIdentifier;
|
||||
});
|
||||
const connection = await Effect.runPromise(
|
||||
decodeGitConnectionInput({
|
||||
credential: args.token,
|
||||
credentialKind: "token",
|
||||
provider: "gitea",
|
||||
serverUrl: args.serverUrl,
|
||||
username: args.username,
|
||||
})
|
||||
);
|
||||
const encrypted = await encryptCredential(connection.credential);
|
||||
const connectionId = await ctx.runMutation(
|
||||
internal.gitConnectionData.persist,
|
||||
{
|
||||
...encrypted,
|
||||
credentialKind: connection.credentialKind,
|
||||
provider: connection.provider,
|
||||
serverUrl: connection.serverUrl,
|
||||
userId,
|
||||
username: connection.username,
|
||||
}
|
||||
);
|
||||
return { connectionId };
|
||||
},
|
||||
});
|
||||
|
||||
export const connectGithub = action({
|
||||
args: {},
|
||||
handler: async (ctx): Promise<{ connectionId: Id<"gitConnections"> }> => {
|
||||
const identity = await ctx.auth.getUserIdentity();
|
||||
if (!identity) {
|
||||
throw new ConvexError("Authentication required");
|
||||
}
|
||||
const { auth, headers } = await authComponent.getAuth(createAuth, ctx);
|
||||
const token = await auth.api.getAccessToken({
|
||||
body: { providerId: "github" },
|
||||
headers,
|
||||
});
|
||||
if (!token.accessToken) {
|
||||
throw new ConvexError("GitHub account is not connected");
|
||||
}
|
||||
const encrypted = await encryptCredential(token.accessToken);
|
||||
const connectionId = await ctx.runMutation(
|
||||
internal.gitConnectionData.persist,
|
||||
{
|
||||
...encrypted,
|
||||
credentialKind: "oauth",
|
||||
provider: "github",
|
||||
serverUrl: "https://github.com",
|
||||
userId: identity.tokenIdentifier,
|
||||
}
|
||||
);
|
||||
return { connectionId };
|
||||
},
|
||||
});
|
||||
@@ -44,10 +44,28 @@ export default defineSchema({
|
||||
.index("by_userId", ["userId"])
|
||||
.index("by_organizationId", ["organizationId"])
|
||||
.index("by_organizationId_and_userId", ["organizationId", "userId"]),
|
||||
gitConnections: defineTable({
|
||||
connectedAt: v.number(),
|
||||
credentialCiphertext: v.string(),
|
||||
credentialIv: v.string(),
|
||||
credentialKind: v.union(v.literal("oauth"), v.literal("token")),
|
||||
organizationId: v.id("organizations"),
|
||||
provider: v.union(v.literal("github"), v.literal("gitea")),
|
||||
serverUrl: v.string(),
|
||||
updatedAt: v.number(),
|
||||
username: v.optional(v.string()),
|
||||
})
|
||||
.index("by_organizationId", ["organizationId"])
|
||||
.index("by_organizationId_and_provider_and_serverUrl", [
|
||||
"organizationId",
|
||||
"provider",
|
||||
"serverUrl",
|
||||
]),
|
||||
projects: defineTable({
|
||||
createdAt: v.number(),
|
||||
defaultBranch: v.optional(v.string()),
|
||||
description: v.optional(v.string()),
|
||||
gitConnectionId: v.optional(v.id("gitConnections")),
|
||||
name: v.string(),
|
||||
normalizedSourceUrl: v.string(),
|
||||
organizationId: v.id("organizations"),
|
||||
@@ -314,11 +332,17 @@ export default defineSchema({
|
||||
]),
|
||||
|
||||
workRuns: defineTable({
|
||||
baseRevision: v.optional(v.string()),
|
||||
candidateRevision: v.optional(v.string()),
|
||||
createdAt: v.number(),
|
||||
designVersion: v.optional(v.number()),
|
||||
endedAt: v.optional(v.number()),
|
||||
kitId: v.string(),
|
||||
kitVersion: v.string(),
|
||||
environmentId: v.optional(v.string()),
|
||||
executionKind: v.optional(
|
||||
v.union(v.literal("simulated"), v.literal("real"))
|
||||
),
|
||||
scenario: v.union(
|
||||
v.literal("success"),
|
||||
v.literal("transient-failure-then-success"),
|
||||
@@ -337,6 +361,7 @@ export default defineSchema({
|
||||
),
|
||||
terminalClassification: v.optional(attemptClassification),
|
||||
terminalSummary: v.optional(v.string()),
|
||||
workflowId: v.optional(v.string()),
|
||||
workId: v.id("works"),
|
||||
}).index("by_work_and_createdAt", ["workId", "createdAt"]),
|
||||
|
||||
@@ -356,6 +381,7 @@ export default defineSchema({
|
||||
),
|
||||
summary: v.optional(v.string()),
|
||||
workId: v.id("works"),
|
||||
workspaceKey: v.optional(v.string()),
|
||||
})
|
||||
.index("by_runId_and_number", ["runId", "number"])
|
||||
.index("by_status_and_leaseExpiresAt", ["status", "leaseExpiresAt"]),
|
||||
|
||||
80
packages/backend/convex/workExecutionAgent.ts
Normal file
80
packages/backend/convex/workExecutionAgent.ts
Normal file
@@ -0,0 +1,80 @@
|
||||
"use node";
|
||||
|
||||
import { env } from "@code/env/convex";
|
||||
import { decodeWorkAttemptExecutionResult } from "@code/primitives/execution-runtime";
|
||||
import { ConvexError, v } from "convex/values";
|
||||
import { Effect } from "effect";
|
||||
|
||||
import { internal } from "./_generated/api";
|
||||
import { internalAction } from "./_generated/server";
|
||||
import { decryptCredential } from "./gitConnections";
|
||||
|
||||
const backendUrl = () => env.AGENT_BACKEND_URL ?? env.FLUE_URL;
|
||||
|
||||
export const executeAttempt = internalAction({
|
||||
args: { attemptId: v.id("workAttempts") },
|
||||
handler: async (ctx, args) => {
|
||||
const running = await ctx.runMutation(
|
||||
internal.workExecutionWorkflow.markAttemptRunning,
|
||||
args
|
||||
);
|
||||
if (!running) {
|
||||
throw new ConvexError("Attempt is no longer runnable");
|
||||
}
|
||||
const context = await ctx.runQuery(
|
||||
internal.workExecutionWorkflow.executionContext,
|
||||
args
|
||||
);
|
||||
const credential = await decryptCredential(
|
||||
context.connection.credentialCiphertext,
|
||||
context.connection.credentialIv
|
||||
);
|
||||
const response = await fetch(
|
||||
`${backendUrl()}/internal/work-attempts/execute`,
|
||||
{
|
||||
body: JSON.stringify({
|
||||
attemptId: String(context.attempt._id),
|
||||
auth: {
|
||||
credential,
|
||||
provider: context.connection.provider,
|
||||
serverUrl: context.connection.serverUrl,
|
||||
username: context.connection.username,
|
||||
},
|
||||
baseBranch: context.project.defaultBranch ?? "main",
|
||||
prompt: context.prompt,
|
||||
repositoryUrl: context.project.sourceUrl,
|
||||
runId: String(context.run._id),
|
||||
workId: String(context.work._id),
|
||||
workspaceKey: context.attempt.workspaceKey,
|
||||
}),
|
||||
headers: {
|
||||
authorization: `Bearer ${env.FLUE_DB_TOKEN}`,
|
||||
"content-type": "application/json",
|
||||
},
|
||||
method: "POST",
|
||||
}
|
||||
);
|
||||
const payload = (await response.json()) as unknown;
|
||||
if (!response.ok) {
|
||||
throw new ConvexError(
|
||||
typeof payload === "object" && payload && "error" in payload
|
||||
? String(payload.error)
|
||||
: `Agent backend returned ${response.status}`
|
||||
);
|
||||
}
|
||||
return await Effect.runPromise(decodeWorkAttemptExecutionResult(payload));
|
||||
},
|
||||
});
|
||||
|
||||
export const cancelAttempt = internalAction({
|
||||
args: { workspaceKey: v.string() },
|
||||
handler: async (_ctx, args) => {
|
||||
await fetch(
|
||||
`${backendUrl()}/internal/work-attempts/${encodeURIComponent(args.workspaceKey)}/cancel`,
|
||||
{
|
||||
headers: { authorization: `Bearer ${env.FLUE_DB_TOKEN}` },
|
||||
method: "POST",
|
||||
}
|
||||
);
|
||||
},
|
||||
});
|
||||
111
packages/backend/convex/workExecutionWorkflow.test.ts
Normal file
111
packages/backend/convex/workExecutionWorkflow.test.ts
Normal file
@@ -0,0 +1,111 @@
|
||||
import { convexTest } from "convex-test";
|
||||
import { anyApi } from "convex/server";
|
||||
import { describe, expect, test } from "vitest";
|
||||
|
||||
import schema from "./schema";
|
||||
|
||||
const modules = import.meta.glob("./**/*.ts");
|
||||
const api = anyApi;
|
||||
|
||||
describe("real work execution persistence", () => {
|
||||
test("records revisions, activity, diff, and terminal state atomically", async () => {
|
||||
const t = convexTest({ modules, schema });
|
||||
const seeded = await t.run(async (ctx) => {
|
||||
const organizationId = await ctx.db.insert("organizations", {
|
||||
createdAt: 1,
|
||||
createdBy: "user",
|
||||
kind: "personal",
|
||||
name: "Personal",
|
||||
});
|
||||
const projectId = await ctx.db.insert("projects", {
|
||||
createdAt: 1,
|
||||
name: "Zopu",
|
||||
normalizedSourceUrl: "https://example.com/zopu",
|
||||
organizationId,
|
||||
repositoryPath: "puter/zopu",
|
||||
sourceHost: "example.com",
|
||||
sourceUrl: "https://example.com/zopu",
|
||||
updatedAt: 1,
|
||||
});
|
||||
const workId = await ctx.db.insert("works", {
|
||||
createdAt: 1,
|
||||
objective: "Implement a real slice",
|
||||
organizationId,
|
||||
projectId,
|
||||
status: "executing",
|
||||
title: "Real execution",
|
||||
updatedAt: 1,
|
||||
});
|
||||
const sliceRowId = await ctx.db.insert("workSlices", {
|
||||
designVersion: 1,
|
||||
objective: "Change the repo",
|
||||
observableBehavior: "A diff exists",
|
||||
ordinal: 0,
|
||||
payloadJson: "{}",
|
||||
sliceId: "slice-1",
|
||||
status: "running",
|
||||
title: "Implementation",
|
||||
workId,
|
||||
});
|
||||
const runId = await ctx.db.insert("workRuns", {
|
||||
createdAt: 1,
|
||||
designVersion: 1,
|
||||
executionKind: "real",
|
||||
kitId: "coding-v0",
|
||||
kitVersion: "1",
|
||||
scenario: "success",
|
||||
sliceId: "slice-1",
|
||||
sliceRowId,
|
||||
status: "running",
|
||||
workId,
|
||||
});
|
||||
const attemptId = await ctx.db.insert("workAttempts", {
|
||||
number: 1,
|
||||
runId,
|
||||
status: "running",
|
||||
workId,
|
||||
workspaceKey: "workspace-1",
|
||||
});
|
||||
return { attemptId, runId, workId };
|
||||
});
|
||||
|
||||
await t.mutation(api.workExecutionWorkflow.completeAttempt, {
|
||||
attemptId: seeded.attemptId,
|
||||
result: {
|
||||
baseRevision: "base123",
|
||||
candidateRevision: "candidate456",
|
||||
changedFiles: ["src/index.ts"],
|
||||
diff: "+export const ready = true;",
|
||||
environmentId: "workspace-1",
|
||||
events: [
|
||||
{
|
||||
kind: "runtime.completed",
|
||||
message: "completed",
|
||||
metadata: {},
|
||||
occurredAt: 2,
|
||||
sequence: 0,
|
||||
},
|
||||
],
|
||||
summary: "Changed one file",
|
||||
},
|
||||
});
|
||||
|
||||
const state = await t.run(async (ctx) => ({
|
||||
artifacts: await ctx.db.query("workArtifacts").collect(),
|
||||
attempt: await ctx.db.get(seeded.attemptId),
|
||||
run: await ctx.db.get(seeded.runId),
|
||||
work: await ctx.db.get(seeded.workId),
|
||||
}));
|
||||
expect(state.attempt?.classification).toBe("Succeeded");
|
||||
expect(state.run).toMatchObject({
|
||||
baseRevision: "base123",
|
||||
candidateRevision: "candidate456",
|
||||
terminalClassification: "Succeeded",
|
||||
});
|
||||
expect(state.work?.status).toBe("completed");
|
||||
expect(state.artifacts[0]).toMatchObject({
|
||||
kind: "diff",
|
||||
sourceRevision: "candidate456",
|
||||
});
|
||||
});
|
||||
});
|
||||
403
packages/backend/convex/workExecutionWorkflow.ts
Normal file
403
packages/backend/convex/workExecutionWorkflow.ts
Normal file
@@ -0,0 +1,403 @@
|
||||
import { defaultCodingKitV0 } from "@code/primitives/resolver";
|
||||
import { WorkflowManager } from "@convex-dev/workflow";
|
||||
import type { WorkflowId } from "@convex-dev/workflow";
|
||||
import { ConvexError, v } from "convex/values";
|
||||
|
||||
import { components, internal } from "./_generated/api";
|
||||
import type { Doc, Id } from "./_generated/dataModel";
|
||||
import { internalMutation, internalQuery, mutation } from "./_generated/server";
|
||||
import type { MutationCtx } from "./_generated/server";
|
||||
import { requireProjectMember } from "./authz";
|
||||
|
||||
export const workflow = new WorkflowManager(components.workflow);
|
||||
|
||||
const resolveReadySlice = async (
|
||||
ctx: MutationCtx,
|
||||
work: Doc<"works">,
|
||||
sliceId?: string
|
||||
) => {
|
||||
if (work.designVersion === undefined) {
|
||||
throw new ConvexError("Work has no approved Design to execute");
|
||||
}
|
||||
const slices = await ctx.db
|
||||
.query("workSlices")
|
||||
.withIndex("by_workId_and_designVersion", (q) =>
|
||||
q.eq("workId", work._id).eq("designVersion", work.designVersion!)
|
||||
)
|
||||
.collect();
|
||||
const slice = sliceId
|
||||
? slices.find((candidate) => candidate.sliceId === sliceId)
|
||||
: slices
|
||||
.sort((left, right) => left.ordinal - right.ordinal)
|
||||
.find((candidate) => candidate.status === "ready");
|
||||
if (!slice || slice.status !== "ready") {
|
||||
throw new ConvexError("Only the next ready slice can be executed");
|
||||
}
|
||||
return slice;
|
||||
};
|
||||
|
||||
export const execute = workflow
|
||||
.define({ args: { attemptId: v.id("workAttempts") } })
|
||||
.handler(async (step, args): Promise<void> => {
|
||||
try {
|
||||
const result = await step.runAction(
|
||||
internal.workExecutionAgent.executeAttempt,
|
||||
args,
|
||||
{ retry: true }
|
||||
);
|
||||
await step.runMutation(internal.workExecutionWorkflow.completeAttempt, {
|
||||
attemptId: args.attemptId,
|
||||
result: {
|
||||
...result,
|
||||
changedFiles: [...result.changedFiles],
|
||||
events: result.events.map((item) => ({
|
||||
...item,
|
||||
metadata: { ...item.metadata },
|
||||
})),
|
||||
},
|
||||
});
|
||||
} catch (error) {
|
||||
await step.runMutation(internal.workExecutionWorkflow.failAttempt, {
|
||||
attemptId: args.attemptId,
|
||||
summary: error instanceof Error ? error.message : "Execution failed",
|
||||
});
|
||||
}
|
||||
});
|
||||
|
||||
export const startExecution = mutation({
|
||||
args: {
|
||||
sliceId: v.optional(v.string()),
|
||||
workId: v.id("works"),
|
||||
},
|
||||
handler: async (
|
||||
ctx,
|
||||
args
|
||||
): Promise<{
|
||||
attemptId: Id<"workAttempts">;
|
||||
runId: Id<"workRuns">;
|
||||
workflowId: WorkflowId;
|
||||
}> => {
|
||||
const work = await ctx.db.get(args.workId);
|
||||
if (!work) {
|
||||
throw new ConvexError("Work not found");
|
||||
}
|
||||
await requireProjectMember(ctx, work.projectId);
|
||||
if (work.status !== "ready") {
|
||||
throw new ConvexError("Work must be Ready before execution");
|
||||
}
|
||||
if (
|
||||
work.definitionApprovalVersion !== work.definitionVersion ||
|
||||
work.designApprovalVersion !== work.designVersion
|
||||
) {
|
||||
throw new ConvexError("Execution requires exact approved versions");
|
||||
}
|
||||
const project = await ctx.db.get(work.projectId);
|
||||
if (!project?.gitConnectionId) {
|
||||
throw new ConvexError("Connect Git credentials to this project first");
|
||||
}
|
||||
const slice = await resolveReadySlice(ctx, work, args.sliceId);
|
||||
const createdAt = Date.now();
|
||||
const runId = await ctx.db.insert("workRuns", {
|
||||
createdAt,
|
||||
designVersion: slice.designVersion,
|
||||
executionKind: "real",
|
||||
kitId: defaultCodingKitV0.id,
|
||||
kitVersion: defaultCodingKitV0.version,
|
||||
scenario: "success",
|
||||
sliceId: slice.sliceId,
|
||||
sliceRowId: slice._id,
|
||||
status: "running",
|
||||
workId: work._id,
|
||||
});
|
||||
const workspaceKey = `work-${work._id}-run-${runId}`;
|
||||
const attemptId = await ctx.db.insert("workAttempts", {
|
||||
number: 1,
|
||||
runId,
|
||||
status: "queued",
|
||||
workId: work._id,
|
||||
workspaceKey,
|
||||
});
|
||||
const workflowId: WorkflowId = await workflow.start(
|
||||
ctx,
|
||||
internal.workExecutionWorkflow.execute,
|
||||
{ attemptId }
|
||||
);
|
||||
await ctx.db.patch(runId, { startedAt: createdAt, workflowId });
|
||||
await ctx.db.patch(slice._id, { status: "running" });
|
||||
await ctx.db.patch(work._id, { status: "executing", updatedAt: createdAt });
|
||||
await ctx.db.insert("workEvents", {
|
||||
createdAt,
|
||||
idempotencyKey: `real-run-started:${runId}`,
|
||||
kind: "run.started",
|
||||
referenceId: String(runId),
|
||||
workId: work._id,
|
||||
});
|
||||
return { attemptId, runId, workflowId };
|
||||
},
|
||||
});
|
||||
|
||||
export const executionContext = internalQuery({
|
||||
args: { attemptId: v.id("workAttempts") },
|
||||
handler: async (ctx, args) => {
|
||||
const attempt = await ctx.db.get(args.attemptId);
|
||||
if (!attempt) {
|
||||
throw new ConvexError("Attempt not found");
|
||||
}
|
||||
const run = await ctx.db.get(attempt.runId);
|
||||
const work = await ctx.db.get(attempt.workId);
|
||||
if (!run || !work || !run.sliceRowId) {
|
||||
throw new ConvexError("Execution records are incomplete");
|
||||
}
|
||||
const [slice, project] = await Promise.all([
|
||||
ctx.db.get(run.sliceRowId),
|
||||
ctx.db.get(work.projectId),
|
||||
]);
|
||||
if (!slice || !project?.gitConnectionId) {
|
||||
throw new ConvexError("Project execution configuration is incomplete");
|
||||
}
|
||||
const connection = await ctx.db.get(project.gitConnectionId);
|
||||
if (!connection) {
|
||||
throw new ConvexError("Git connection not found");
|
||||
}
|
||||
return {
|
||||
attempt,
|
||||
connection,
|
||||
project,
|
||||
prompt: [
|
||||
`Implement this approved Zopu slice: ${slice.title}`,
|
||||
`Objective: ${slice.objective}`,
|
||||
`Observable behavior: ${slice.observableBehavior}`,
|
||||
"Inspect the repository instructions first. Make focused changes and run relevant checks.",
|
||||
].join("\n\n"),
|
||||
run,
|
||||
work,
|
||||
};
|
||||
},
|
||||
});
|
||||
|
||||
export const markAttemptRunning = internalMutation({
|
||||
args: { attemptId: v.id("workAttempts") },
|
||||
handler: async (ctx, args) => {
|
||||
const attempt = await ctx.db.get(args.attemptId);
|
||||
if (!attempt || attempt.status !== "queued") {
|
||||
return false;
|
||||
}
|
||||
await ctx.db.patch(attempt._id, {
|
||||
startedAt: Date.now(),
|
||||
status: "running",
|
||||
});
|
||||
return true;
|
||||
},
|
||||
});
|
||||
|
||||
const settleSliceAndWork = async (
|
||||
ctx: MutationCtx,
|
||||
run: Doc<"workRuns">,
|
||||
succeeded: boolean
|
||||
) => {
|
||||
if (!run.sliceRowId) {
|
||||
return;
|
||||
}
|
||||
const slice = await ctx.db.get(run.sliceRowId);
|
||||
if (!slice) {
|
||||
return;
|
||||
}
|
||||
await ctx.db.patch(slice._id, { status: succeeded ? "completed" : "ready" });
|
||||
const work = await ctx.db.get(run.workId);
|
||||
if (!work) {
|
||||
return;
|
||||
}
|
||||
let status: Doc<"works">["status"] = succeeded ? "completed" : "failed";
|
||||
if (succeeded) {
|
||||
const slices = await ctx.db
|
||||
.query("workSlices")
|
||||
.withIndex("by_workId_and_designVersion", (q) =>
|
||||
q.eq("workId", work._id).eq("designVersion", slice.designVersion)
|
||||
)
|
||||
.collect();
|
||||
const next = slices
|
||||
.sort((left, right) => left.ordinal - right.ordinal)
|
||||
.find((candidate) => candidate.status === "planned");
|
||||
if (next) {
|
||||
await ctx.db.patch(next._id, { status: "ready" });
|
||||
status = "ready";
|
||||
}
|
||||
}
|
||||
await ctx.db.patch(work._id, { status, updatedAt: Date.now() });
|
||||
};
|
||||
|
||||
export const completeAttempt = internalMutation({
|
||||
args: {
|
||||
attemptId: v.id("workAttempts"),
|
||||
result: v.object({
|
||||
baseRevision: v.string(),
|
||||
candidateRevision: v.string(),
|
||||
changedFiles: v.array(v.string()),
|
||||
diff: v.string(),
|
||||
environmentId: v.string(),
|
||||
events: v.array(
|
||||
v.object({
|
||||
kind: v.string(),
|
||||
message: v.string(),
|
||||
metadata: v.record(v.string(), v.string()),
|
||||
occurredAt: v.number(),
|
||||
sequence: v.number(),
|
||||
})
|
||||
),
|
||||
summary: v.string(),
|
||||
}),
|
||||
},
|
||||
handler: async (ctx, args) => {
|
||||
const attempt = await ctx.db.get(args.attemptId);
|
||||
if (!attempt || attempt.status === "terminal") {
|
||||
return;
|
||||
}
|
||||
const run = await ctx.db.get(attempt.runId);
|
||||
const work = await ctx.db.get(attempt.workId);
|
||||
if (!run || !work) {
|
||||
throw new ConvexError("Execution records not found");
|
||||
}
|
||||
for (const item of args.result.events) {
|
||||
await ctx.db.insert("workAttemptEvents", {
|
||||
attemptId: attempt._id,
|
||||
kind: item.kind,
|
||||
message: item.message,
|
||||
metadataJson: JSON.stringify(item.metadata),
|
||||
occurredAt: item.occurredAt,
|
||||
sequence: item.sequence,
|
||||
});
|
||||
}
|
||||
const endedAt = Date.now();
|
||||
await ctx.db.patch(attempt._id, {
|
||||
classification: "Succeeded",
|
||||
endedAt,
|
||||
status: "terminal",
|
||||
summary: args.result.summary,
|
||||
});
|
||||
await ctx.db.patch(run._id, {
|
||||
baseRevision: args.result.baseRevision,
|
||||
candidateRevision: args.result.candidateRevision,
|
||||
endedAt,
|
||||
environmentId: args.result.environmentId,
|
||||
status: "terminal",
|
||||
terminalClassification: "Succeeded",
|
||||
terminalSummary: args.result.summary,
|
||||
});
|
||||
await ctx.db.insert("workArtifacts", {
|
||||
attemptId: attempt._id,
|
||||
createdAt: endedAt,
|
||||
designVersion: run.designVersion,
|
||||
environmentId: args.result.environmentId,
|
||||
idempotencyKey: `real-diff:${attempt._id}`,
|
||||
kind: "diff",
|
||||
metadataJson: JSON.stringify({ changedFiles: args.result.changedFiles }),
|
||||
organizationId: work.organizationId,
|
||||
producer: "agentos-codex",
|
||||
projectId: work.projectId,
|
||||
provenanceJson: JSON.stringify({
|
||||
baseRevision: args.result.baseRevision,
|
||||
}),
|
||||
runId: run._id,
|
||||
sliceId: run.sliceId,
|
||||
sourceRevision: args.result.candidateRevision,
|
||||
title: "Implementation diff",
|
||||
verificationStatus: "unverified",
|
||||
workId: work._id,
|
||||
...(args.result.diff.length > 0
|
||||
? {
|
||||
uri: `data:text/plain;charset=utf-8,${encodeURIComponent(args.result.diff.slice(0, 50_000))}`,
|
||||
}
|
||||
: {}),
|
||||
});
|
||||
await settleSliceAndWork(ctx, run, true);
|
||||
await ctx.db.insert("workEvents", {
|
||||
createdAt: endedAt,
|
||||
idempotencyKey: `real-run-completed:${run._id}`,
|
||||
kind: "run.completed",
|
||||
payloadJson: JSON.stringify({ classification: "Succeeded" }),
|
||||
referenceId: String(run._id),
|
||||
workId: work._id,
|
||||
});
|
||||
},
|
||||
});
|
||||
|
||||
export const failAttempt = internalMutation({
|
||||
args: { attemptId: v.id("workAttempts"), summary: v.string() },
|
||||
handler: async (ctx, args) => {
|
||||
const attempt = await ctx.db.get(args.attemptId);
|
||||
if (!attempt || attempt.status === "terminal") {
|
||||
return;
|
||||
}
|
||||
const run = await ctx.db.get(attempt.runId);
|
||||
if (!run) {
|
||||
return;
|
||||
}
|
||||
const endedAt = Date.now();
|
||||
await ctx.db.patch(attempt._id, {
|
||||
classification: "PermanentFailure",
|
||||
endedAt,
|
||||
status: "terminal",
|
||||
summary: args.summary,
|
||||
});
|
||||
await ctx.db.patch(run._id, {
|
||||
endedAt,
|
||||
status: "terminal",
|
||||
terminalClassification: "PermanentFailure",
|
||||
terminalSummary: args.summary,
|
||||
});
|
||||
await settleSliceAndWork(ctx, run, false);
|
||||
},
|
||||
});
|
||||
|
||||
export const cancelExecution = mutation({
|
||||
args: { runId: v.id("workRuns") },
|
||||
handler: async (ctx, args) => {
|
||||
const run = await ctx.db.get(args.runId);
|
||||
if (!run) {
|
||||
throw new ConvexError("Run not found");
|
||||
}
|
||||
await requireProjectMember(ctx, (await ctx.db.get(run.workId))!.projectId);
|
||||
if (run.status !== "running") {
|
||||
return { cancelled: false };
|
||||
}
|
||||
if (run.workflowId) {
|
||||
await workflow.cancel(ctx, run.workflowId as WorkflowId);
|
||||
}
|
||||
const attempts = await ctx.db
|
||||
.query("workAttempts")
|
||||
.withIndex("by_runId_and_number", (q) => q.eq("runId", run._id))
|
||||
.collect();
|
||||
const attempt = attempts.find(
|
||||
(candidate) => candidate.status !== "terminal"
|
||||
);
|
||||
if (attempt?.workspaceKey) {
|
||||
await ctx.scheduler.runAfter(
|
||||
0,
|
||||
internal.workExecutionAgent.cancelAttempt,
|
||||
{
|
||||
workspaceKey: attempt.workspaceKey,
|
||||
}
|
||||
);
|
||||
await ctx.db.patch(attempt._id, {
|
||||
classification: "Cancelled",
|
||||
endedAt: Date.now(),
|
||||
status: "terminal",
|
||||
summary: "Execution cancelled",
|
||||
});
|
||||
}
|
||||
await ctx.db.patch(run._id, {
|
||||
endedAt: Date.now(),
|
||||
status: "cancelled",
|
||||
terminalClassification: "Cancelled",
|
||||
terminalSummary: "Execution cancelled",
|
||||
});
|
||||
if (run.sliceRowId) {
|
||||
await ctx.db.patch(run.sliceRowId, { status: "ready" });
|
||||
}
|
||||
const work = await ctx.db.get(run.workId);
|
||||
if (work) {
|
||||
await ctx.db.patch(work._id, { status: "ready", updatedAt: Date.now() });
|
||||
}
|
||||
return { cancelled: true };
|
||||
},
|
||||
});
|
||||
@@ -771,13 +771,41 @@ export const listForProject = query({
|
||||
.eq("designVersion", work.designVersion ?? 0)
|
||||
)
|
||||
.collect();
|
||||
const runs = await ctx.db
|
||||
const runRows = await ctx.db
|
||||
.query("workRuns")
|
||||
.withIndex("by_work_and_createdAt", (q: any) =>
|
||||
q.eq("workId", work._id)
|
||||
)
|
||||
.order("desc")
|
||||
.take(10);
|
||||
const runs = await Promise.all(
|
||||
runRows.map(async (run) => {
|
||||
const attempts = await ctx.db
|
||||
.query("workAttempts")
|
||||
.withIndex("by_runId_and_number", (q) => q.eq("runId", run._id))
|
||||
.collect();
|
||||
const eventsByAttempt = await Promise.all(
|
||||
attempts.map((attempt) =>
|
||||
ctx.db
|
||||
.query("workAttemptEvents")
|
||||
.withIndex("by_attempt_and_sequence", (q: any) =>
|
||||
q.eq("attemptId", attempt._id)
|
||||
)
|
||||
.collect()
|
||||
)
|
||||
);
|
||||
const attemptEvents = eventsByAttempt
|
||||
.flat()
|
||||
.sort((left, right) => left.occurredAt - right.occurredAt);
|
||||
const artifacts = await ctx.db
|
||||
.query("workArtifacts")
|
||||
.withIndex("by_runId_and_createdAt", (q) =>
|
||||
q.eq("runId", run._id)
|
||||
)
|
||||
.collect();
|
||||
return { ...run, artifacts, attemptEvents, attempts };
|
||||
})
|
||||
);
|
||||
const events = await ctx.db
|
||||
.query("workEvents")
|
||||
.withIndex("by_work_and_createdAt", (q: any) =>
|
||||
|
||||
Reference in New Issue
Block a user