"use node"; import { env } from "@code/env/convex"; import { WorkAttemptExecutionError, decodeWorkAttemptExecutionFailure, 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) { // 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: { 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 // matching Pi ACP session for cancellation. await fetch( `${backendUrl()}/internal/work-attempts/${encodeURIComponent(args.workspaceKey)}/cancel`, { body: JSON.stringify({ attemptId: args.attemptId }), headers: { authorization: `Bearer ${env.FLUE_DB_TOKEN}`, "content-type": "application/json", }, method: "POST", } ); }, });