History

wip durable execution

Rhys Sullivan committed 8 months ago

Commit ffe801e1 parent e50769019 files · +1334 −136Browse files

Unified · Split

19 files
  1. apps/executor/src/cli/main.ts +5 −0
  2. apps/executor/src/cli/pending-interaction-output.test.ts +1 −0
  3. apps/executor/src/server/server.test.ts +5 −0
  4. packages/codemode-mcp/src/mcp-tools.ts +40 −5
  5. packages/control-plane/drizzle/20260312233000_execution_steps_and_private_interaction_responses/migration.sql +25 −0
  6. packages/control-plane/src/api/executions/api.ts +2 −0
  7. packages/control-plane/src/persistence/index.ts +2 −0
  8. packages/control-plane/src/persistence/repos/execution-steps-repo.ts +89 −0
  9. packages/control-plane/src/persistence/repos/index.ts +1 −0
  10. packages/control-plane/src/persistence/schema.ts +40 −0
  11. packages/control-plane/src/runtime/execution-mcp-resume.test.ts +5 −7
  12. packages/control-plane/src/runtime/execution-replay.test.ts +271 −0
  13. packages/control-plane/src/runtime/execution-service.ts +668 −89
  14. packages/control-plane/src/runtime/live-execution.ts +8 −0
  15. packages/control-plane/src/runtime/local-operations.ts +63 −12
  16. packages/control-plane/src/runtime/source-auth-service.ts +44 −14
  17. packages/control-plane/src/schema/ids.ts +4 −0
  18. packages/control-plane/src/schema/models/execution.ts +43 −0
  19. packages/executor-mcp/src/index.ts +18 −9
apps/executor/src/cli/main.ts+5−0
@@ -857,6 +857,9 @@ const printUrlInteraction = (input: { · 856 unmodified lines
857 }857 }
858 });858 });
859 859
860const executionInteractionMode = (): "live_form" | "detach" =>
861 process.stdin.isTTY && process.stdout.isTTY ? "live_form" : "detach";
862
860const promptInteraction = (input: {863const promptInteraction = (input: {
861 interaction: ExecutionInteraction;864 interaction: ExecutionInteraction;
862 shouldOpenUrls: boolean;865 shouldOpenUrls: boolean;
@@ -1105,6 +1108,7 @@ const driveExecution = (input: { · 242 unmodified lines
1105 },1108 },
1106 payload: {1109 payload: {
1107 responseJson,1110 responseJson,
1111 interactionMode: executionInteractionMode(),
1108 },1112 },
1109 });1113 });
1110 }1114 }
@@ -1271,6 +1275,7 @@ const callCommand = Command.make( · 160 unmodified lines
1271 },1275 },
1272 payload: {1276 payload: {
1273 code: resolvedCode,1277 code: resolvedCode,
1278 interactionMode: executionInteractionMode(),
1274 },1279 },
1275 });1280 });
1276 1281
apps/executor/src/cli/pending-interaction-output.test.ts+1−0
@@ -30,6 +30,7 @@ const makeInteraction = (patch: Partial<ExecutionInteraction> = {}): ExecutionIn · 29 unmodified lines
30 },30 },
31 }),31 }),
32 responseJson: null,32 responseJson: null,
33 responsePrivateJson: null,
33 createdAt: 1,34 createdAt: 1,
34 updatedAt: 1,35 updatedAt: 1,
35 ...patch,36 ...patch,
apps/executor/src/server/server.test.ts+5−0
@@ -954,6 +954,7 @@ describe("local-executor-server", () => { · 953 unmodified lines
954 },954 },
955 payload: {955 payload: {
956 code: 'return await tools.demo.gated_echo({ value: "from-daemon" });',956 code: 'return await tools.demo.gated_echo({ value: "from-daemon" });',
957 interactionMode: "live_form",
957 },958 },
958 });959 });
959 960
@@ -970,6 +971,7 @@ describe("local-executor-server", () => { · 10 unmodified lines
970 executionId: created.execution.id,971 executionId: created.execution.id,
971 },972 },
972 payload: {973 payload: {
974 interactionMode: "live_form",
973 responseJson: JSON.stringify({975 responseJson: JSON.stringify({
974 action: "accept",976 action: "accept",
975 content: {977 content: {
@@ -1036,6 +1038,7 @@ describe("local-executor-server", () => { · 60 unmodified lines
1036 },1038 },
1037 payload: {1039 payload: {
1038 code: `return await tools.demo.gated_echo({ value: "${value}" });`,1040 code: `return await tools.demo.gated_echo({ value: "${value}" });`,
1041 interactionMode: "live_form",
1039 },1042 },
1040 });1043 });
1041 1044
@@ -1048,6 +1051,7 @@ describe("local-executor-server", () => { · 6 unmodified lines
1048 executionId: created.execution.id,1051 executionId: created.execution.id,
1049 },1052 },
1050 payload: {1053 payload: {
1054 interactionMode: "live_form",
1051 responseJson: JSON.stringify({1055 responseJson: JSON.stringify({
1052 action: "accept",1056 action: "accept",
1053 content: {1057 content: {
@@ -1226,6 +1230,7 @@ describe("local-executor-server", () => { · 172 unmodified lines
1226 },1230 },
1227 payload: {1231 payload: {
1228 code: `return await tools.executor.sources.add({ endpoint: ${JSON.stringify(oauthServer.endpoint)}, name: "Axiom", namespace: "axiom" });`,1232 code: `return await tools.executor.sources.add({ endpoint: ${JSON.stringify(oauthServer.endpoint)}, name: "Axiom", namespace: "axiom" });`,
1233 interactionMode: "live",
1229 },1234 },
1230 });1235 });
1231 1236
packages/codemode-mcp/src/mcp-tools.ts+40−5
@@ -69,6 +69,27 @@ export class McpToolsError extends Data.TaggedError("McpToolsError")<{ · 68 unmodified lines
69 details: string | null;69 details: string | null;
70}> {}70}> {}
71 71
72const EXECUTION_SUSPENDED_SENTINEL = "__EXECUTION_SUSPENDED__";
73
74const causeText = (cause: unknown): string => {
75 if (cause instanceof Error) {
76 return `${cause.message}\n${cause.stack ?? ""}`;
77 }
78
79 if (typeof cause === "string") {
80 return cause;
81 }
82
83 try {
84 return JSON.stringify(cause);
85 } catch {
86 return String(cause);
87 }
88};
89
90const isExecutionSuspendedCause = (cause: unknown): boolean =>
91 causeText(cause).includes(EXECUTION_SUSPENDED_SENTINEL);
92
72const toDetails = (cause: unknown): string =>93const toDetails = (cause: unknown): string =>
73 cause instanceof Error ? cause.message : String(cause);94 cause instanceof Error ? cause.message : String(cause);
74 95
@@ -140,11 +161,13 @@ const resolveElicitationResponse = (input: { · 65 unmodified lines
140 elicitation: input.elicitation,161 elicitation: input.elicitation,
141 }).pipe(162 }).pipe(
142 Effect.mapError((cause) =>163 Effect.mapError((cause) =>
143 new McpToolsError({164 isExecutionSuspendedCause(cause)
144 stage: "call_tool",165 ? (cause as McpToolsError)
145 message: `Failed resolving elicitation for ${input.toolName}`,166 : new McpToolsError({
146 details: toDetails(cause),167 stage: "call_tool",
147 })),168 message: `Failed resolving elicitation for ${input.toolName}`,
169 details: toDetails(cause),
170 })),
148 );171 );
149 172
150const installMcpElicitationHandler = (input: {173const installMcpElicitationHandler = (input: {
@@ -196,6 +219,10 @@ const installMcpElicitationHandler = (input: { · 45 unmodified lines
196 ),219 ),
197 Effect.map(toMcpElicitationResponse),220 Effect.map(toMcpElicitationResponse),
198 Effect.catchAll((error) => {221 Effect.catchAll((error) => {
222 if (isExecutionSuspendedCause(error)) {
223 return Effect.fail(error);
224 }
225
199 console.error(226 console.error(
200 `[mcp-tools] elicitation failed for ${input.toolName}, treating as cancel:`,227 `[mcp-tools] elicitation failed for ${input.toolName}, treating as cancel:`,
201 error instanceof Error ? error.message : String(error),228 error instanceof Error ? error.message : String(error),
@@ -318,6 +345,10 @@ const runMcpListToolsEffect = (input: { · 116 unmodified lines
318 return attempt.right;345 return attempt.right;
319 }346 }
320 347
348 if (isExecutionSuspendedCause(attempt.left)) {
349 return yield* Effect.fail(attempt.left as McpToolsError);
350 }
351
321 if (352 if (
322 input.mcpDiscoveryElicitation353 input.mcpDiscoveryElicitation
323 && isUrlElicitationRequiredError(attempt.left)354 && isUrlElicitationRequiredError(attempt.left)
@@ -382,6 +413,10 @@ const runMcpToolCallEffect = (input: { · 58 unmodified lines
382 return attempt.right;413 return attempt.right;
383 }414 }
384 415
416 if (isExecutionSuspendedCause(attempt.left)) {
417 return yield* Effect.fail(attempt.left as McpToolsError);
418 }
419
385 if (420 if (
386 onElicitation421 onElicitation
387 && isUrlElicitationRequiredError(attempt.left)422 && isUrlElicitationRequiredError(attempt.left)
packages/control-plane/drizzle/20260312233000_execution_steps_and_private_interaction_responses/migration.sql+25−0
@@ -0,0 +1,25 @@
1ALTER TABLE "execution_interactions"
2ADD COLUMN "response_private_json" text;--> statement-breakpoint
3
4CREATE TABLE "execution_steps" (
5 "id" text PRIMARY KEY NOT NULL,
6 "execution_id" text NOT NULL,
7 "sequence" bigint NOT NULL,
8 "kind" text NOT NULL,
9 "status" text NOT NULL,
10 "path" text NOT NULL,
11 "args_json" text NOT NULL,
12 "result_json" text,
13 "error_text" text,
14 "interaction_id" text,
15 "created_at" bigint NOT NULL,
16 "updated_at" bigint NOT NULL,
17 CONSTRAINT "execution_steps_kind_check" CHECK ("execution_steps"."kind" in ('tool_call')),
18 CONSTRAINT "execution_steps_status_check" CHECK ("execution_steps"."status" in ('pending', 'waiting', 'completed', 'failed'))
19);--> statement-breakpoint
20
21CREATE UNIQUE INDEX "execution_steps_execution_sequence_idx"
22ON "execution_steps" ("execution_id", "sequence");--> statement-breakpoint
23
24CREATE INDEX "execution_steps_execution_updated_idx"
25ON "execution_steps" ("execution_id", "updated_at", "id");
packages/control-plane/src/api/executions/api.ts+2−0
@@ -17,12 +17,14 @@ import { TrimmedNonEmptyStringSchema } from "../string-schemas"; · 16 unmodified lines
17 17
18export const CreateExecutionPayloadSchema = Schema.Struct({18export const CreateExecutionPayloadSchema = Schema.Struct({
19 code: TrimmedNonEmptyStringSchema,19 code: TrimmedNonEmptyStringSchema,
20 interactionMode: Schema.optional(Schema.Literal("live", "live_form", "detach")),
20});21});
21 22
22export type CreateExecutionPayload = typeof CreateExecutionPayloadSchema.Type;23export type CreateExecutionPayload = typeof CreateExecutionPayloadSchema.Type;
23 24
24export const ResumeExecutionPayloadSchema = Schema.Struct({25export const ResumeExecutionPayloadSchema = Schema.Struct({
25 responseJson: Schema.optional(Schema.String),26 responseJson: Schema.optional(Schema.String),
27 interactionMode: Schema.optional(Schema.Literal("live", "live_form", "detach")),
26});28});
27 29
28export type ResumeExecutionPayload = typeof ResumeExecutionPayloadSchema.Type;30export type ResumeExecutionPayload = typeof ResumeExecutionPayloadSchema.Type;
packages/control-plane/src/persistence/index.ts+2−0
@@ -10,6 +10,7 @@ import { · 9 unmodified lines
10 createAuthLeasesRepo,10 createAuthLeasesRepo,
11 createCodeMigrationsRepo,11 createCodeMigrationsRepo,
12 createExecutionInteractionsRepo,12 createExecutionInteractionsRepo,
13 createExecutionStepsRepo,
13 createExecutionsRepo,14 createExecutionsRepo,
14 createSecretMaterialsRepo,15 createSecretMaterialsRepo,
15 createSourceAuthSessionsRepo,16 createSourceAuthSessionsRepo,
@@ -64,6 +65,7 @@ const createRows = (client: DrizzleClient, tables: DrizzleTables = drizzleSchema · 48 unmodified lines
64 sourceAuthSessions: createSourceAuthSessionsRepo(client, tables),65 sourceAuthSessions: createSourceAuthSessionsRepo(client, tables),
65 executions: createExecutionsRepo(client, tables),66 executions: createExecutionsRepo(client, tables),
66 executionInteractions: createExecutionInteractionsRepo(client, tables),67 executionInteractions: createExecutionInteractionsRepo(client, tables),
68 executionSteps: createExecutionStepsRepo(client, tables),
67});69});
68 70
69export type SqlControlPlaneRows = ReturnType<typeof createRows>;71export type SqlControlPlaneRows = ReturnType<typeof createRows>;
packages/control-plane/src/persistence/repos/execution-steps-repo.ts+89−0
@@ -0,0 +1,89 @@
1import {
2 type ExecutionStep,
3 ExecutionStepSchema,
4} from "#schema";
5import * as Option from "effect/Option";
6import { Schema } from "effect";
7import { and, desc, eq } from "drizzle-orm";
8
9import type { DrizzleClient } from "../client";
10import type { DrizzleTables } from "../schema";
11import { firstOption } from "./shared";
12
13const decodeExecutionStep = Schema.decodeUnknownSync(ExecutionStepSchema);
14
15export const createExecutionStepsRepo = (
16 client: DrizzleClient,
17 tables: DrizzleTables,
18) => ({
19 getByExecutionAndSequence: (
20 executionId: ExecutionStep["executionId"],
21 sequence: ExecutionStep["sequence"],
22 ) =>
23 client.use("rows.execution_steps.get_by_execution_and_sequence", async (db) => {
24 const rows = await db
25 .select()
26 .from(tables.executionStepsTable)
27 .where(
28 and(
29 eq(tables.executionStepsTable.executionId, executionId),
30 eq(tables.executionStepsTable.sequence, sequence),
31 ),
32 )
33 .limit(1);
34
35 const row = firstOption(rows);
36 return Option.isSome(row)
37 ? Option.some(decodeExecutionStep(row.value))
38 : Option.none<ExecutionStep>();
39 }),
40
41 listByExecutionId: (executionId: ExecutionStep["executionId"]) =>
42 client.use("rows.execution_steps.list_by_execution_id", async (db) => {
43 const rows = await db
44 .select()
45 .from(tables.executionStepsTable)
46 .where(eq(tables.executionStepsTable.executionId, executionId))
47 .orderBy(
48 tables.executionStepsTable.sequence,
49 desc(tables.executionStepsTable.updatedAt),
50 );
51
52 return rows.map((row) => decodeExecutionStep(row));
53 }),
54
55 insert: (step: ExecutionStep) =>
56 client.use("rows.execution_steps.insert", async (db) => {
57 await db.insert(tables.executionStepsTable).values(step);
58 }),
59
60 deleteByExecutionId: (executionId: ExecutionStep["executionId"]) =>
61 client.use("rows.execution_steps.delete_by_execution_id", async (db) => {
62 await db
63 .delete(tables.executionStepsTable)
64 .where(eq(tables.executionStepsTable.executionId, executionId));
65 }),
66
67 updateByExecutionAndSequence: (
68 executionId: ExecutionStep["executionId"],
69 sequence: ExecutionStep["sequence"],
70 patch: Partial<Omit<ExecutionStep, "id" | "executionId" | "sequence" | "createdAt">>,
71 ) =>
72 client.use("rows.execution_steps.update_by_execution_and_sequence", async (db) => {
73 const rows = await db
74 .update(tables.executionStepsTable)
75 .set(patch)
76 .where(
77 and(
78 eq(tables.executionStepsTable.executionId, executionId),
79 eq(tables.executionStepsTable.sequence, sequence),
80 ),
81 )
82 .returning();
83
84 const row = firstOption(rows);
85 return Option.isSome(row)
86 ? Option.some(decodeExecutionStep(row.value))
87 : Option.none<ExecutionStep>();
88 }),
89});
packages/control-plane/src/persistence/repos/index.ts+1−0
@@ -13,3 +13,4 @@ export { createSecretMaterialsRepo } from "./secret-materials-repo"; · 12 unmodified lines
13export { createSourceAuthSessionsRepo } from "./source-auth-sessions-repo";13export { createSourceAuthSessionsRepo } from "./source-auth-sessions-repo";
14export { createExecutionsRepo } from "./executions-repo";14export { createExecutionsRepo } from "./executions-repo";
15export { createExecutionInteractionsRepo } from "./execution-interactions-repo";15export { createExecutionInteractionsRepo } from "./execution-interactions-repo";
16export { createExecutionStepsRepo } from "./execution-steps-repo";
packages/control-plane/src/persistence/schema.ts+40−0
@@ -26,6 +26,7 @@ export const tableNames = { · 25 unmodified lines
26 sourceAuthSessions: "source_auth_sessions",26 sourceAuthSessions: "source_auth_sessions",
27 executions: "executions",27 executions: "executions",
28 executionInteractions: "execution_interactions",28 executionInteractions: "execution_interactions",
29 executionSteps: "execution_steps",
29} as const;30} as const;
30 31
31export const accountsTable = pgTable(32export const accountsTable = pgTable(
@@ -473,6 +474,7 @@ export const executionInteractionsTable = pgTable( · 441 unmodified lines
473 purpose: text("purpose").notNull(),474 purpose: text("purpose").notNull(),
474 payloadJson: text("payload_json").notNull(),475 payloadJson: text("payload_json").notNull(),
475 responseJson: text("response_json"),476 responseJson: text("response_json"),
477 responsePrivateJson: text("response_private_json"),
476 createdAt: bigint("created_at", { mode: "number" }).notNull(),478 createdAt: bigint("created_at", { mode: "number" }).notNull(),
477 updatedAt: bigint("updated_at", { mode: "number" }).notNull(),479 updatedAt: bigint("updated_at", { mode: "number" }).notNull(),
478 },480 },
@@ -489,6 +491,43 @@ export const executionInteractionsTable = pgTable( · 10 unmodified lines
489 ],491 ],
490);492);
491 493
494export const executionStepsTable = pgTable(
495 tableNames.executionSteps,
496 {
497 id: text("id").notNull().primaryKey(),
498 executionId: text("execution_id").notNull(),
499 sequence: bigint("sequence", { mode: "number" }).notNull(),
500 kind: text("kind").notNull(),
501 status: text("status").notNull(),
502 path: text("path").notNull(),
503 argsJson: text("args_json").notNull(),
504 resultJson: text("result_json"),
505 errorText: text("error_text"),
506 interactionId: text("interaction_id"),
507 createdAt: bigint("created_at", { mode: "number" }).notNull(),
508 updatedAt: bigint("updated_at", { mode: "number" }).notNull(),
509 },
510 (table) => [
511 uniqueIndex("execution_steps_execution_sequence_idx").on(
512 table.executionId,
513 table.sequence,
514 ),
515 index("execution_steps_execution_updated_idx").on(
516 table.executionId,
517 table.updatedAt,
518 table.id,
519 ),
520 check(
521 "execution_steps_kind_check",
522 sql`${table.kind} in ('tool_call')`,
523 ),
524 check(
525 "execution_steps_status_check",
526 sql`${table.status} in ('pending', 'waiting', 'completed', 'failed')`,
527 ),
528 ],
529);
530
492export const drizzleSchema = {531export const drizzleSchema = {
493 accountsTable,532 accountsTable,
494 sourcesTable,533 sourcesTable,
@@ -505,6 +544,7 @@ export const drizzleSchema = { · 10 unmodified lines
505 sourceAuthSessionsTable,544 sourceAuthSessionsTable,
506 executionsTable,545 executionsTable,
507 executionInteractionsTable,546 executionInteractionsTable,
547 executionStepsTable,
508};548};
509 549
510export type DrizzleTables = typeof drizzleSchema;550export type DrizzleTables = typeof drizzleSchema;
packages/control-plane/src/runtime/execution-mcp-resume.test.ts+5−7
@@ -280,7 +280,7 @@ const makeMcpExecutionResolver = ( · 279 unmodified lines
280 >);280 >);
281 281
282describe("execution-mcp-resume", () => {282describe("execution-mcp-resume", () => {
283 it.scoped("pauses on MCP elicitation and resumes the same live execution", () =>283 it.scoped("keeps live form elicitation resumable without replaying the run", () =>
284 Effect.gen(function* () {284 Effect.gen(function* () {
285 const mcpServer = yield* makeFormElicitationServer;285 const mcpServer = yield* makeFormElicitationServer;
286 const runtime = yield* Effect.acquireRelease(286 const runtime = yield* Effect.acquireRelease(
@@ -306,6 +306,7 @@ describe("execution-mcp-resume", () => { · 19 unmodified lines
306 },306 },
307 payload: {307 payload: {
308 code: 'return await tools.source.form.gated_echo({ value: "from-control-plane" });',308 code: 'return await tools.source.form.gated_echo({ value: "from-control-plane" });',
309 interactionMode: "live_form",
309 },310 },
310 }),311 }),
311 );312 );
@@ -339,6 +340,7 @@ describe("execution-mcp-resume", () => { · 27 unmodified lines
339 executionId: created.execution.id,340 executionId: created.execution.id,
340 },341 },
341 payload: {342 payload: {
343 interactionMode: "live_form",
342 responseJson: JSON.stringify({344 responseJson: JSON.stringify({
343 action: "accept",345 action: "accept",
344 content: {346 content: {
@@ -349,14 +351,10 @@ describe("execution-mcp-resume", () => { · 4 unmodified lines
349 }),351 }),
350 );352 );
351 353
352 expect(resumed.execution.id).toBe(created.execution.id);
353 expect(resumed.execution.status).toBe("completed");354 expect(resumed.execution.status).toBe("completed");
354 expect(resumed.pendingInteraction).toBeNull();355 expect(resumed.pendingInteraction).toBeNull();
355 expect(resumed.execution.resultJson).toBe(356 expect(resumed.execution.resultJson).toContain("approved:from-control-plane");
356 JSON.stringify({357
357 content: [{ type: "text", text: "approved:from-control-plane" }],
358 }),
359 );
360 }),358 }),
361 60_000,359 60_000,
362 );360 );
packages/control-plane/src/runtime/execution-replay.test.ts+271−0
@@ -0,0 +1,271 @@
1import { describe, expect, it } from "@effect/vitest";
2import * as Effect from "effect/Effect";
3import * as Schema from "effect/Schema";
4
5import { makeToolInvokerFromTools } from "@executor/codemode-core";
6import { makeDenoSubprocessExecutor } from "@executor/runtime-deno-subprocess";
7
8import {
9 createSqlControlPlaneRuntime,
10 type ResolveExecutionEnvironment,
11} from "./index";
12import { withControlPlaneClient } from "./test-http-client";
13
14const gatedEchoElicitation = {
15 message: "Approve gated echo?",
16 requestedSchema: {
17 type: "object",
18 properties: {
19 approve: {
20 type: "boolean",
21 title: "Approve",
22 },
23 },
24 required: ["approve"],
25 },
26} as const;
27
28const makeReplayResolver = (countedCalls: { value: number }): ResolveExecutionEnvironment =>
29 ({ onElicitation }) =>
30 Effect.succeed({
31 executor: makeDenoSubprocessExecutor(),
32 toolInvoker: makeToolInvokerFromTools({
33 onElicitation,
34 tools: {
35 "counted.add": {
36 description: "Adds numbers and counts calls",
37 inputSchema: Schema.standardSchemaV1(
38 Schema.Struct({
39 a: Schema.optional(Schema.Number),
40 b: Schema.optional(Schema.Number),
41 }),
42 ),
43 execute: ({ a, b }: { a?: number; b?: number }) => {
44 countedCalls.value += 1;
45 return { sum: (a ?? 0) + (b ?? 0) };
46 },
47 },
48 "gated.echo": {
49 tool: {
50 description: "Echoes only after approval",
51 inputSchema: Schema.standardSchemaV1(
52 Schema.Struct({
53 value: Schema.String,
54 approve: Schema.optional(Schema.Boolean),
55 }),
56 ),
57 execute: ({ value, approve }: { value: string; approve?: boolean }) => ({
58 text: approve === true ? `approved:${value}` : `denied:${value}`,
59 }),
60 },
61 metadata: {
62 interaction: "required",
63 elicitation: gatedEchoElicitation,
64 },
65 },
66 },
67 }),
68 });
69
70const createRuntime = (countedCalls: { value: number }) =>
71 Effect.acquireRelease(
72 createSqlControlPlaneRuntime({
73 localDataDir: ":memory:",
74 executionResolver: makeReplayResolver(countedCalls),
75 }),
76 (runtime) => Effect.promise(() => runtime.close()).pipe(Effect.orDie),
77 );
78
79describe("execution replay", () => {
80 it.scoped("replays completed proxy calls instead of re-executing them", () =>
81 Effect.gen(function* () {
82 const countedCalls = { value: 0 };
83 const runtime = yield* createRuntime(countedCalls);
84 const installation = runtime.localInstallation;
85
86 const created = yield* withControlPlaneClient(
87 {
88 runtime,
89 accountId: installation.accountId,
90 },
91 (client) =>
92 client.executions.create({
93 path: {
94 workspaceId: installation.workspaceId,
95 },
96 payload: {
97 code: [
98 "const sum = await tools.counted.add({ a: 20, b: 22 });",
99 'const echo = await tools.gated.echo({ value: "from-replay" });',
100 "return { sum, echo };",
101 ].join("\n"),
102 },
103 }),
104 );
105
106 expect(created.execution.status).toBe("waiting_for_interaction");
107 expect(countedCalls.value).toBe(1);
108
109 const resumed = yield* withControlPlaneClient(
110 {
111 runtime,
112 accountId: installation.accountId,
113 },
114 (client) =>
115 client.executions.resume({
116 path: {
117 workspaceId: installation.workspaceId,
118 executionId: created.execution.id,
119 },
120 payload: {
121 responseJson: JSON.stringify({
122 action: "accept",
123 content: {
124 approve: true,
125 },
126 }),
127 },
128 }),
129 );
130
131 expect(resumed.execution.status).toBe("completed");
132 expect(countedCalls.value).toBe(1);
133 expect(resumed.execution.resultJson).toBe(
134 JSON.stringify({
135 sum: { sum: 42 },
136 echo: { text: "approved:from-replay" },
137 }),
138 );
139
140 const completedSteps = yield* runtime.persistence.rows.executionSteps.listByExecutionId(
141 created.execution.id,
142 );
143 expect(completedSteps).toHaveLength(0);
144 }),
145 );
146
147 it.scoped("keeps live form interactions in the same run when requested", () =>
148 Effect.gen(function* () {
149 const countedCalls = { value: 0 };
150 const runtime = yield* createRuntime(countedCalls);
151 const installation = runtime.localInstallation;
152
153 const created = yield* withControlPlaneClient(
154 {
155 runtime,
156 accountId: installation.accountId,
157 },
158 (client) =>
159 client.executions.create({
160 path: {
161 workspaceId: installation.workspaceId,
162 },
163 payload: {
164 interactionMode: "live_form",
165 code: [
166 "const nonce = String(Math.random());",
167 "return await tools.gated.echo({ value: nonce });",
168 ].join("\n"),
169 },
170 }),
171 );
172
173 expect(created.execution.status).toBe("waiting_for_interaction");
174
175 const resumed = yield* withControlPlaneClient(
176 {
177 runtime,
178 accountId: installation.accountId,
179 },
180 (client) =>
181 client.executions.resume({
182 path: {
183 workspaceId: installation.workspaceId,
184 executionId: created.execution.id,
185 },
186 payload: {
187 interactionMode: "live_form",
188 responseJson: JSON.stringify({
189 action: "accept",
190 content: {
191 approve: true,
192 },
193 }),
194 },
195 }),
196 );
197
198 expect(resumed.execution.status).toBe("completed");
199 expect(resumed.execution.resultJson).toContain("approved:");
200 }),
201 );
202
203 it.scoped("fails loudly when replayed code reaches a different proxy call", () =>
204 Effect.gen(function* () {
205 const countedCalls = { value: 0 };
206 const runtime = yield* createRuntime(countedCalls);
207 const installation = runtime.localInstallation;
208
209 const created = yield* withControlPlaneClient(
210 {
211 runtime,
212 accountId: installation.accountId,
213 },
214 (client) =>
215 client.executions.create({
216 path: {
217 workspaceId: installation.workspaceId,
218 },
219 payload: {
220 code: [
221 "await tools.counted.add({ a: 1, b: 2 });",
222 'return await tools.gated.echo({ value: "mismatch" });',
223 ].join("\n"),
224 },
225 }),
226 );
227
228 expect(created.execution.status).toBe("waiting_for_interaction");
229 expect(countedCalls.value).toBe(1);
230
231 yield* runtime.persistence.rows.executions.update(created.execution.id, {
232 code: [
233 "await tools.counted.add({ a: 1, b: 999 });",
234 'return await tools.gated.echo({ value: "mismatch" });',
235 ].join("\n"),
236 updatedAt: Date.now(),
237 });
238
239 const resumed = yield* withControlPlaneClient(
240 {
241 runtime,
242 accountId: installation.accountId,
243 },
244 (client) =>
245 client.executions.resume({
246 path: {
247 workspaceId: installation.workspaceId,
248 executionId: created.execution.id,
249 },
250 payload: {
251 responseJson: JSON.stringify({
252 action: "accept",
253 content: {
254 approve: true,
255 },
256 }),
257 },
258 }),
259 );
260
261 expect(resumed.execution.status).toBe("failed");
262 expect(resumed.execution.errorText).toContain("Durable execution mismatch");
263 expect(countedCalls.value).toBe(1);
264
265 const failedSteps = yield* runtime.persistence.rows.executionSteps.listByExecutionId(
266 created.execution.id,
267 );
268 expect(failedSteps).toHaveLength(0);
269 }),
270 );
271});
packages/control-plane/src/runtime/execution-service.ts+668−89
@@ -3,13 +3,20 @@ import { · 2 unmodified lines
3 ControlPlaneNotFoundError,3 ControlPlaneNotFoundError,
4 ControlPlaneStorageError,4 ControlPlaneStorageError,
5} from "../api/errors";5} from "../api/errors";
6import type {
7 ElicitationResponse,
8 OnElicitation,
9 ToolInvocationContext,
10 ToolInvoker,
11} from "@executor/codemode-core";
6import type {12import type {
7 CreateExecutionPayload,13 CreateExecutionPayload,
8 ResumeExecutionPayload,14 ResumeExecutionPayload,
9} from "../api/executions/api";15} from "../api/executions/api";
10import type { ToolInvoker } from "@executor/codemode-core";
11import {16import {
12 ExecutionIdSchema,17 ExecutionIdSchema,
18 ExecutionInteractionIdSchema,
19 ExecutionStepIdSchema,
13 type AccountId,20 type AccountId,
14 type Execution,21 type Execution,
15 type ExecutionEnvelope,22 type ExecutionEnvelope,
@@ -21,12 +28,10 @@ import * as Effect from "effect/Effect"; · 5 unmodified lines
21import * as Option from "effect/Option";28import * as Option from "effect/Option";
22import * as Schema from "effect/Schema";29import * as Schema from "effect/Schema";
23 30
24import {31import { type ResolveExecutionEnvironment } from "./execution-state";
25 type ResolveExecutionEnvironment,
26 ResumeUnsupportedError,
27} from "./execution-state";
28import {32import {
29 LiveExecutionManagerService,33 LiveExecutionManagerService,
34 sanitizePersistedElicitationResponse,
30 type LiveExecutionManager,35 type LiveExecutionManager,
31} from "./live-execution";36} from "./live-execution";
32import {37import {
@@ -46,6 +51,12 @@ const executionOps = { · 13 unmodified lines
46 resume: operationErrors("executions.resume"),51 resume: operationErrors("executions.resume"),
47} as const;52} as const;
48 53
54type InteractionMode = NonNullable<CreateExecutionPayload["interactionMode"]>;
55
56const EXECUTION_SUSPENDED_SENTINEL = "__EXECUTION_SUSPENDED__";
57
58const DEFAULT_INTERACTION_MODE: InteractionMode = "detach";
59
49const serializeJson = (value: unknown): string | null => {60const serializeJson = (value: unknown): string | null => {
50 if (value === undefined) {61 if (value === undefined) {
51 return null;62 return null;
@@ -54,6 +65,17 @@ const serializeJson = (value: unknown): string | null => { · 2 unmodified lines
54 return JSON.stringify(value);65 return JSON.stringify(value);
55};66};
56 67
68const serializeRequiredJson = (value: unknown): string =>
69 JSON.stringify(value === undefined ? null : value);
70
71const parseStoredJson = (value: string | null): unknown => {
72 if (value === null) {
73 return undefined;
74 }
75
76 return JSON.parse(value);
77};
78
57const ElicitationActionSchema = Schema.Literal("accept", "decline", "cancel");79const ElicitationActionSchema = Schema.Literal("accept", "decline", "cancel");
58 80
59const ElicitationResponseSchema = Schema.Struct({81const ElicitationResponseSchema = Schema.Struct({
@@ -88,12 +110,119 @@ const withExecutionInvocationContext = (input: { · 28 unmodified lines
88 typeof context?.callId === "string" && context.callId.length > 0110 typeof context?.callId === "string" && context.callId.length > 0
89 ? context.callId111 ? context.callId
90 : `call_${String(sequence)}`,112 : `call_${String(sequence)}`,
113 executionStepSequence: sequence,
91 },114 },
92 });115 });
93 },116 },
94 };117 };
95};118};
96 119
120const executionStepSequenceFromContext = (
121 context: ToolInvocationContext | undefined,
122): number | null => {
123 const value = context?.executionStepSequence;
124 return typeof value === "number" && Number.isSafeInteger(value) && value > 0
125 ? value
126 : null;
127};
128
129const executionStepIdFor = (
130 executionId: Execution["id"],
131 sequence: number,
132) => ExecutionStepIdSchema.make(`${executionId}:step:${String(sequence)}`);
133
134const interactionIdSuffixForRequest = (input: {
135 interactionId: string;
136 path: string;
137 context?: ToolInvocationContext;
138}) => {
139 const callId =
140 typeof input.context?.callId === "string" && input.context.callId.length > 0
141 ? input.context.callId
142 : null;
143
144 return callId === null ? input.interactionId : `${callId}:${input.path}`;
145};
146
147const executionInteractionIdForRequest = (input: {
148 executionId: Execution["id"];
149 interactionId: string;
150 path: string;
151 context?: ToolInvocationContext;
152}) => {
153 return ExecutionInteractionIdSchema.make(
154 `${input.executionId}:${interactionIdSuffixForRequest(input)}`,
155 );
156};
157
158const resolveInteractionMode = (
159 value: CreateExecutionPayload["interactionMode"] | ResumeExecutionPayload["interactionMode"],
160): InteractionMode =>
161 value === "live" || value === "live_form" ? value : DEFAULT_INTERACTION_MODE;
162
163const createExecutionSuspendedError = (input: {
164 executionId: Execution["id"];
165 interactionId: string;
166}): Error =>
167 new Error(
168 `${EXECUTION_SUSPENDED_SENTINEL}:${input.executionId}:${input.interactionId}`,
169 );
170
171const isExecutionSuspendedValue = (value: unknown): boolean => {
172 if (value instanceof Error) {
173 return value.message.includes(EXECUTION_SUSPENDED_SENTINEL);
174 }
175
176 return typeof value === "string" && value.includes(EXECUTION_SUSPENDED_SENTINEL);
177};
178
179const decodeStoredElicitationResponse = (input: {
180 interactionId: string;
181 responseJson: string | null;
182}) =>
183 Effect.try({
184 try: () => {
185 if (input.responseJson === null) {
186 throw new Error(
187 `Interaction ${input.interactionId} has no stored response`,
188 );
189 }
190
191 return JSON.parse(input.responseJson);
192 },
193 catch: (error) =>
194 error instanceof Error ? error : new Error(String(error)),
195 }).pipe(
196 Effect.flatMap((decoded) =>
197 decodeElicitationResponse(decoded).pipe(
198 Effect.mapError((error) => new Error(String(error))),
199 )
200 ),
201 );
202
203const verifyStoredStepMatches = (input: {
204 executionId: Execution["id"];
205 sequence: number;
206 expectedPath: string;
207 expectedArgsJson: string;
208 actualPath: string;
209 actualArgsJson: string;
210}) => {
211 if (
212 input.expectedPath === input.actualPath
213 && input.expectedArgsJson === input.actualArgsJson
214 ) {
215 return;
216 }
217
218 throw new Error(
219 [
220 `Durable execution mismatch for ${input.executionId} at tool step ${String(input.sequence)}.`,
221 `Expected ${input.expectedPath}(${input.expectedArgsJson}) but replay reached ${input.actualPath}(${input.actualArgsJson}).`,
222 ].join(" "),
223 );
224};
225
97const fetchExecution = (226const fetchExecution = (
98 store: ControlPlaneStoreShape,227 store: ControlPlaneStoreShape,
99 input: {228 input: {
@@ -141,6 +270,482 @@ const fetchExecutionEnvelope = ( · 41 unmodified lines
141 };270 };
142 });271 });
143 272
273const suspendExecutionForInteraction = (input: {
274 rows: ControlPlaneStoreShape;
275 executionId: Execution["id"];
276 liveExecutionManager: LiveExecutionManager;
277 request: Parameters<OnElicitation>[0];
278 interactionId: ExecutionEnvelope["pendingInteraction"] extends infer T
279 ? T extends { id: infer I }
280 ? I
281 : never
282 : never;
283}) =>
284 Effect.gen(function* () {
285 const now = Date.now();
286 const existing = yield* input.rows.executionInteractions.getById(input.interactionId);
287 const stepSequence = executionStepSequenceFromContext(input.request.context);
288
289 if (Option.isSome(existing) && existing.value.status !== "pending") {
290 return yield* decodeStoredElicitationResponse({
291 interactionId: input.interactionId,
292 responseJson:
293 existing.value.responsePrivateJson ?? existing.value.responseJson,
294 });
295 }
296
297 if (Option.isNone(existing)) {
298 yield* input.rows.executionInteractions.insert({
299 id: ExecutionInteractionIdSchema.make(input.interactionId),
300 executionId: input.executionId,
301 status: "pending",
302 kind: input.request.elicitation.mode === "url" ? "url" : "form",
303 purpose: "elicitation",
304 payloadJson:
305 serializeJson({
306 path: input.request.path,
307 sourceKey: input.request.sourceKey,
308 args: input.request.args,
309 context: input.request.context,
310 elicitation: input.request.elicitation,
311 }) ?? "{}",
312 responseJson: null,
313 responsePrivateJson: null,
314 createdAt: now,
315 updatedAt: now,
316 });
317 }
318
319 if (stepSequence !== null) {
320 yield* input.rows.executionSteps.updateByExecutionAndSequence(
321 input.executionId,
322 stepSequence,
323 {
324 status: "waiting",
325 interactionId: ExecutionInteractionIdSchema.make(input.interactionId),
326 updatedAt: now,
327 },
328 );
329 }
330
331 yield* input.rows.executions.update(input.executionId, {
332 status: "waiting_for_interaction",
333 updatedAt: now,
334 });
335 yield* input.liveExecutionManager.publishState({
336 executionId: input.executionId,
337 state: "waiting_for_interaction",
338 });
339
340 return yield* Effect.fail(
341 createExecutionSuspendedError({
342 executionId: input.executionId,
343 interactionId: input.interactionId,
344 }),
345 );
346 });
347
348const createHybridOnElicitation = (input: {
349 rows: ControlPlaneStoreShape;
350 executionId: Execution["id"];
351 liveExecutionManager: LiveExecutionManager;
352 interactionMode: InteractionMode;
353}): OnElicitation => {
354 const liveOnElicitation = input.liveExecutionManager.createOnElicitation({
355 rows: input.rows,
356 executionId: input.executionId,
357 });
358
359 return (request) =>
360 Effect.gen(function* () {
361 const interactionIdSuffix = interactionIdSuffixForRequest({
362 interactionId: request.interactionId,
363 path: request.path,
364 context: request.context,
365 });
366 const interactionId = executionInteractionIdForRequest({
367 executionId: input.executionId,
368 interactionId: request.interactionId,
369 path: request.path,
370 context: request.context,
371 });
372 const existing = yield* input.rows.executionInteractions.getById(interactionId);
373
374 if (Option.isSome(existing) && existing.value.status !== "pending") {
375 return yield* decodeStoredElicitationResponse({
376 interactionId,
377 responseJson:
378 existing.value.responsePrivateJson ?? existing.value.responseJson,
379 });
380 }
381
382 const allowLiveWait =
383 input.interactionMode === "live"
384 || (input.interactionMode === "live_form" && request.elicitation.mode !== "url");
385
386 if (Option.isNone(existing) && allowLiveWait) {
387 const stepSequence = executionStepSequenceFromContext(request.context);
388 if (stepSequence !== null) {
389 yield* input.rows.executionSteps.updateByExecutionAndSequence(
390 input.executionId,
391 stepSequence,
392 {
393 status: "waiting",
394 interactionId: ExecutionInteractionIdSchema.make(interactionId),
395 updatedAt: Date.now(),
396 },
397 );
398 }
399
400 return yield* liveOnElicitation({
401 ...request,
402 interactionId: interactionIdSuffix,
403 });
404 }
405
406 return yield* suspendExecutionForInteraction({
407 rows: input.rows,
408 executionId: input.executionId,
409 liveExecutionManager: input.liveExecutionManager,
410 request,
411 interactionId,
412 });
413 });
414};
415
416const createReplayToolInvoker = (input: {
417 rows: ControlPlaneStoreShape;
418 executionId: Execution["id"];
419 toolInvoker: ToolInvoker;
420}): ToolInvoker => ({
421 invoke: ({ path, args, context }) =>
422 Effect.gen(function* () {
423 const stepSequence = executionStepSequenceFromContext(context);
424 if (stepSequence === null) {
425 return yield* input.toolInvoker.invoke({ path, args, context });
426 }
427
428 const argsJson = serializeRequiredJson(args);
429 const existing = yield* input.rows.executionSteps.getByExecutionAndSequence(
430 input.executionId,
431 stepSequence,
432 );
433
434 if (Option.isSome(existing)) {
435 verifyStoredStepMatches({
436 executionId: input.executionId,
437 sequence: stepSequence,
438 expectedPath: existing.value.path,
439 expectedArgsJson: existing.value.argsJson,
440 actualPath: path,
441 actualArgsJson: argsJson,
442 });
443
444 if (existing.value.status === "completed") {
445 return parseStoredJson(existing.value.resultJson);
446 }
447
448 if (existing.value.status === "failed") {
449 return yield* Effect.fail(
450 new Error(
451 existing.value.errorText
452 ?? `Stored tool step ${String(stepSequence)} failed`,
453 ),
454 );
455 }
456 } else {
457 const now = Date.now();
458 yield* input.rows.executionSteps.insert({
459 id: executionStepIdFor(input.executionId, stepSequence),
460 executionId: input.executionId,
461 sequence: stepSequence,
462 kind: "tool_call",
463 status: "pending",
464 path,
465 argsJson,
466 resultJson: null,
467 errorText: null,
468 interactionId: null,
469 createdAt: now,
470 updatedAt: now,
471 });
472 }
473
474 try {
475 const value = yield* input.toolInvoker.invoke({ path, args, context });
476 const updatedAt = Date.now();
477
478 yield* input.rows.executionSteps.updateByExecutionAndSequence(
479 input.executionId,
480 stepSequence,
481 {
482 status: "completed",
483 resultJson: serializeJson(value),
484 errorText: null,
485 updatedAt,
486 },
487 );
488
489 return value;
490 } catch (error) {
491 const updatedAt = Date.now();
492
493 if (isExecutionSuspendedValue(error)) {
494 yield* input.rows.executionSteps.updateByExecutionAndSequence(
495 input.executionId,
496 stepSequence,
497 {
498 status: "waiting",
499 updatedAt,
500 },
501 );
502
503 return yield* Effect.fail(error);
504 }
505
506 yield* input.rows.executionSteps.updateByExecutionAndSequence(
507 input.executionId,
508 stepSequence,
509 {
510 status: "failed",
511 errorText: error instanceof Error ? error.message : String(error),
512 updatedAt,
513 },
514 );
515
516 return yield* Effect.fail(error);
517 }
518 }),
519});
520
521const persistExecutionOutcome = (input: {
522 rows: ControlPlaneStoreShape;
523 liveExecutionManager: LiveExecutionManager;
524 executionId: Execution["id"];
525 outcome: {
526 result: unknown;
527 error?: string;
528 logs?: string[];
529 };
530}) =>
531 Effect.gen(function* () {
532 if (isExecutionSuspendedValue(input.outcome.error)) {
533 return;
534 }
535
536 if (input.outcome.error) {
537 const [execution, pendingInteraction] = yield* Effect.all([
538 input.rows.executions.getById(input.executionId),
539 input.rows.executionInteractions.getPendingByExecutionId(input.executionId),
540 ]);
541
542 if (
543 Option.isSome(execution)
544 && execution.value.status === "waiting_for_interaction"
545 && Option.isSome(pendingInteraction)
546 ) {
547 return;
548 }
549 }
550
551 const completedAt = Date.now();
552 const updated = yield* input.rows.executions.update(input.executionId, {
553 status: input.outcome.error ? "failed" : "completed",
554 resultJson: serializeJson(input.outcome.result),
555 errorText: input.outcome.error ?? null,
556 logsJson: serializeJson(input.outcome.logs ?? null),
557 completedAt,
558 updatedAt: completedAt,
559 });
560
561 if (Option.isNone(updated)) {
562 yield* input.liveExecutionManager.clearRun(input.executionId);
563 return;
564 }
565
566 yield* input.liveExecutionManager.finishRun({
567 executionId: input.executionId,
568 state: updated.value.status === "completed" ? "completed" : "failed",
569 });
570
571 yield* input.rows.executionSteps.deleteByExecutionId(input.executionId);
572 });
573
574const persistExecutionFailure = (input: {
575 rows: ControlPlaneStoreShape;
576 liveExecutionManager: LiveExecutionManager;
577 executionId: Execution["id"];
578 error: string;
579}) =>
580 Effect.gen(function* () {
581 const completedAt = Date.now();
582 const updated = yield* input.rows.executions.update(input.executionId, {
583 status: "failed",
584 errorText: input.error,
585 completedAt,
586 updatedAt: completedAt,
587 });
588
589 if (Option.isNone(updated)) {
590 yield* input.liveExecutionManager.clearRun(input.executionId);
591 return;
592 }
593
594 yield* input.liveExecutionManager.finishRun({
595 executionId: input.executionId,
596 state: "failed",
597 });
598
599 yield* input.rows.executionSteps.deleteByExecutionId(input.executionId);
600 });
601
602const runExecutionAttemptWithDependencies = (
603 store: ControlPlaneStoreShape,
604 executionResolver: ResolveExecutionEnvironment,
605 liveExecutionManager: LiveExecutionManager,
606 execution: Execution,
607 interactionMode: InteractionMode,
608) =>
609 executionResolver({
610 workspaceId: execution.workspaceId,
611 accountId: execution.createdByAccountId,
612 executionId: execution.id,
613 onElicitation: createHybridOnElicitation({
614 rows: store,
615 executionId: execution.id,
616 liveExecutionManager,
617 interactionMode,
618 }),
619 }).pipe(
620 Effect.map((environment) => ({
621 executor: environment.executor,
622 toolInvoker: withExecutionInvocationContext({
623 executionId: execution.id,
624 toolInvoker: createReplayToolInvoker({
625 rows: store,
626 executionId: execution.id,
627 toolInvoker: environment.toolInvoker,
628 }),
629 }),
630 })),
631 Effect.flatMap(({ executor, toolInvoker }) =>
632 executor.execute(execution.code, toolInvoker)
633 ),
634 Effect.flatMap((outcome) =>
635 persistExecutionOutcome({
636 rows: store,
637 liveExecutionManager,
638 executionId: execution.id,
639 outcome,
640 })
641 ),
642 Effect.catchAll((error) =>
643 persistExecutionFailure({
644 rows: store,
645 liveExecutionManager,
646 executionId: execution.id,
647 error: error instanceof Error ? error.message : String(error),
648 }).pipe(
649 Effect.catchAll(() => liveExecutionManager.clearRun(execution.id)),
650 )
651 ),
652 );
653
654const forkExecutionAttemptWithDependencies = (
655 store: ControlPlaneStoreShape,
656 executionResolver: ResolveExecutionEnvironment,
657 liveExecutionManager: LiveExecutionManager,
658 execution: Execution,
659 interactionMode: InteractionMode,
660) =>
661 Effect.sync(() => {
662 Effect.runFork(
663 runExecutionAttemptWithDependencies(
664 store,
665 executionResolver,
666 liveExecutionManager,
667 execution,
668 interactionMode,
669 ),
670 );
671 });
672
673const submitExecutionInteractionResponseWithDependencies = (
674 store: ControlPlaneStoreShape,
675 executionResolver: ResolveExecutionEnvironment,
676 liveExecutionManager: LiveExecutionManager,
677 input: {
678 executionId: ExecutionId;
679 response: ElicitationResponse;
680 interactionMode: InteractionMode;
681 },
682) =>
683 Effect.gen(function* () {
684 const execution = yield* store.executions.getById(input.executionId);
685 if (Option.isNone(execution)) {
686 return false;
687 }
688
689 const pendingInteraction = yield* store.executionInteractions.getPendingByExecutionId(
690 input.executionId,
691 );
692 if (Option.isNone(pendingInteraction)) {
693 return false;
694 }
695
696 if (
697 execution.value.status !== "waiting_for_interaction"
698 && execution.value.status !== "failed"
699 ) {
700 return false;
701 }
702
703 const now = Date.now();
704 const steps = yield* store.executionSteps.listByExecutionId(input.executionId);
705 const waitingStep = [...steps]
706 .reverse()
707 .find((step) => step.interactionId === pendingInteraction.value.id);
708
709 if (waitingStep) {
710 yield* store.executionSteps.updateByExecutionAndSequence(
711 input.executionId,
712 waitingStep.sequence,
713 {
714 status: "waiting",
715 errorText: null,
716 updatedAt: now,
717 },
718 );
719 }
720
721 yield* store.executionInteractions.update(pendingInteraction.value.id, {
722 status: input.response.action === "cancel" ? "cancelled" : "resolved",
723 responseJson: serializeJson(
724 sanitizePersistedElicitationResponse(input.response),
725 ),
726 responsePrivateJson: serializeJson(input.response),
727 updatedAt: now,
728 });
729
730 const updated = yield* store.executions.update(input.executionId, {
731 status: "running",
732 updatedAt: now,
733 });
734 if (Option.isNone(updated)) {
735 return false;
736 }
737
738 yield* forkExecutionAttemptWithDependencies(
739 store,
740 executionResolver,
741 liveExecutionManager,
742 updated.value,
743 input.interactionMode,
744 );
745
746 return true;
747 });
748
144const createExecutionWithDependencies = (749const createExecutionWithDependencies = (
145 store: ControlPlaneStoreShape,750 store: ControlPlaneStoreShape,
146 executionResolver: ResolveExecutionEnvironment,751 executionResolver: ResolveExecutionEnvironment,
@@ -190,84 +795,15 @@ const createExecutionWithDependencies = ( · 43 unmodified lines
190 );795 );
191 }796 }
192 797
193 const environment = yield* executionResolver({
194 workspaceId: input.workspaceId,
195 accountId: input.createdByAccountId,
196 executionId: execution.id,
197 onElicitation: liveExecutionManager.createOnElicitation({
198 rows: store,
199 executionId: execution.id,
200 }),
201 }).pipe(
202 Effect.mapError((error) =>
203 executionOps.create.child("environment").unknownStorage(
204 error,
205 "Execution environment resolution failed",
206 ),
207 ),
208 );
209
210 const nextState = yield* liveExecutionManager.registerStateWaiter(execution.id);798 const nextState = yield* liveExecutionManager.registerStateWaiter(execution.id);
211 const toolInvoker = withExecutionInvocationContext({
212 executionId: execution.id,
213 toolInvoker: environment.toolInvoker,
214 });
215 799
216 yield* Effect.sync(() => {800 yield* forkExecutionAttemptWithDependencies(
217 Effect.runFork(801 store,
218 environment.executor.execute(code, toolInvoker).pipe(802 executionResolver,
219 Effect.flatMap((outcome) => {803 liveExecutionManager,
220 const completedAt = Date.now();804 running.value,
221 return executionOps.create.child("complete").mapStorage(805 resolveInteractionMode(input.payload.interactionMode),
222 store.executions.update(execution.id, {806 );
223 status: outcome.error ? "failed" : "completed",
224 resultJson: serializeJson(outcome.result),
225 errorText: outcome.error ?? null,
226 logsJson: serializeJson(outcome.logs ?? null),
227 completedAt,
228 updatedAt: completedAt,
229 }),
230 ).pipe(
231 Effect.flatMap((updated) =>
232 Option.isNone(updated)
233 ? Effect.fail(
234 executionOps.create.notFound(
235 "Execution not found after completion",
236 `executionId=${execution.id}`,
237 ),
238 )
239 : Effect.succeed(updated.value),
240 ),
241 Effect.flatMap((updated) =>
242 liveExecutionManager.finishRun({
243 executionId: execution.id,
244 state: updated.status === "completed" ? "completed" : "failed",
245 }).pipe(Effect.as(updated)),
246 ),
247 );
248 }),
249 Effect.catchAll((error) => {
250 const completedAt = Date.now();
251 return executionOps.create.child("complete_error").mapStorage(
252 store.executions.update(execution.id, {
253 status: "failed",
254 errorText: error instanceof Error ? error.message : String(error),
255 completedAt,
256 updatedAt: completedAt,
257 }),
258 ).pipe(
259 Effect.zipRight(
260 liveExecutionManager.finishRun({
261 executionId: execution.id,
262 state: "failed",
263 }),
264 ),
265 Effect.catchAll(() => liveExecutionManager.clearRun(execution.id)),
266 );
267 }),
268 ),
269 );
270 });
271 807
272 yield* Deferred.await(nextState);808 yield* Deferred.await(nextState);
273 809
@@ -308,6 +844,27 @@ export const getExecution = (input: { · 34 unmodified lines
308 })844 })
309 );845 );
310 846
847export const submitExecutionInteractionResponse = (input: {
848 executionId: ExecutionId;
849 response: ElicitationResponse;
850 interactionMode?: InteractionMode;
851}) =>
852 Effect.gen(function* () {
853 const store = yield* ControlPlaneStore;
854 const executionResolver = yield* RuntimeExecutionResolverService;
855 const liveExecutionManager = yield* LiveExecutionManagerService;
856
857 return yield* submitExecutionInteractionResponseWithDependencies(
858 store,
859 executionResolver,
860 liveExecutionManager,
861 {
862 ...input,
863 interactionMode: input.interactionMode ?? DEFAULT_INTERACTION_MODE,
864 },
865 );
866 });
867
311export const resumeExecution = (input: {868export const resumeExecution = (input: {
312 workspaceId: WorkspaceId;869 workspaceId: WorkspaceId;
313 executionId: ExecutionId;870 executionId: ExecutionId;
@@ -316,6 +873,7 @@ export const resumeExecution = (input: { · 2 unmodified lines
316}) =>873}) =>
317 Effect.gen(function* () {874 Effect.gen(function* () {
318 const store = yield* ControlPlaneStore;875 const store = yield* ControlPlaneStore;
876 const executionResolver = yield* RuntimeExecutionResolverService;
319 const liveExecutionManager = yield* LiveExecutionManagerService;877 const liveExecutionManager = yield* LiveExecutionManagerService;
320 878
321 const existing = yield* fetchExecutionEnvelope(store, {879 const existing = yield* fetchExecutionEnvelope(store, {
@@ -324,7 +882,13 @@ export const resumeExecution = (input: { · 2 unmodified lines
324 operation: "executions.resume",882 operation: "executions.resume",
325 });883 });
326 884
327 if (existing.execution.status !== "waiting_for_interaction") {885 if (
886 existing.execution.status !== "waiting_for_interaction"
887 && !(
888 existing.execution.status === "failed"
889 && existing.pendingInteraction !== null
890 )
891 ) {
328 return yield* Effect.fail(892 return yield* Effect.fail(
329 executionOps.resume.badRequest(893 executionOps.resume.badRequest(
330 "Execution is not waiting for interaction",894 "Execution is not waiting for interaction",
@@ -353,23 +917,38 @@ export const resumeExecution = (input: { · 22 unmodified lines
353 String(error),917 String(error),
354 ),918 ),
355 ),919 ),
356 ),920 )
357 ),921 ),
358 );922 );
359 923
360 const nextState = yield* liveExecutionManager.registerStateWaiter(input.executionId);924 const nextState = yield* liveExecutionManager.registerStateWaiter(input.executionId);
361 const resumed = yield* liveExecutionManager.resolveInteraction({925 const resumedLive = yield* liveExecutionManager.resolveInteraction({
362 executionId: input.executionId,926 executionId: input.executionId,
363 response,927 response,
364 });928 });
365 929
366 if (!resumed) {930 if (!resumedLive) {
367 return yield* Effect.fail(931 const resumed = yield* executionOps.resume.child("submit_interaction").mapStorage(
368 executionOps.resume.badRequest(932 submitExecutionInteractionResponseWithDependencies(
369 "Resume is unavailable for this execution",933 store,
370 `executionId=${input.executionId} mode=${new ResumeUnsupportedError({ executionId: input.executionId })._tag}`,934 executionResolver,
935 liveExecutionManager,
936 {
937 executionId: input.executionId,
938 response,
939 interactionMode: resolveInteractionMode(input.payload.interactionMode),
940 },
371 ),941 ),
372 );942 );
943
944 if (!resumed) {
945 return yield* Effect.fail(
946 executionOps.resume.badRequest(
947 "Resume is unavailable for this execution",
948 `executionId=${input.executionId}`,
949 ),
950 );
951 }
373 }952 }
374 953
375 yield* Deferred.await(nextState);954 yield* Deferred.await(nextState);
packages/control-plane/src/runtime/live-execution.ts+8−0
@@ -27,6 +27,10 @@ type LiveRunEntry = { · 26 unmodified lines
27};27};
28 28
29type LiveExecutionManagerShape = {29type LiveExecutionManagerShape = {
30 publishState: (input: {
31 executionId: Execution["id"];
32 state: VisibleExecutionState;
33 }) => Effect.Effect<void>;
30 registerStateWaiter: (34 registerStateWaiter: (
31 executionId: Execution["id"],35 executionId: Execution["id"],
32 ) => Effect.Effect<Deferred.Deferred<VisibleExecutionState>>;36 ) => Effect.Effect<Deferred.Deferred<VisibleExecutionState>>;
@@ -138,6 +142,8 @@ export const createLiveExecutionManager = () => { · 105 unmodified lines
138 });142 });
139 143
140 const manager = {144 const manager = {
145 publishState,
146
141 registerStateWaiter: (executionId) =>147 registerStateWaiter: (executionId) =>
142 Effect.gen(function* () {148 Effect.gen(function* () {
143 const waiter = yield* Deferred.make<VisibleExecutionState>();149 const waiter = yield* Deferred.make<VisibleExecutionState>();
@@ -168,6 +174,7 @@ export const createLiveExecutionManager = () => { · 24 unmodified lines
168 elicitation: input.elicitation,174 elicitation: input.elicitation,
169 }) ?? "{}",175 }) ?? "{}",
170 responseJson: null,176 responseJson: null,
177 responsePrivateJson: null,
171 createdAt: now,178 createdAt: now,
172 updatedAt: now,179 updatedAt: now,
173 };180 };
@@ -194,6 +201,7 @@ export const createLiveExecutionManager = () => { · 20 unmodified lines
194 yield* rows.executionInteractions.update(interaction.id, {201 yield* rows.executionInteractions.update(interaction.id, {
195 status: resolved.action === "cancel" ? "cancelled" : "resolved",202 status: resolved.action === "cancel" ? "cancelled" : "resolved",
196 responseJson: serializeJson(sanitizePersistedElicitationResponse(resolved)),203 responseJson: serializeJson(sanitizePersistedElicitationResponse(resolved)),
204 responsePrivateJson: serializeJson(resolved),
197 updatedAt: resolvedAt,205 updatedAt: resolvedAt,
198 });206 });
199 yield* rows.executions.update(executionId, {207 yield* rows.executions.update(executionId, {
packages/control-plane/src/runtime/local-operations.ts+63−12
@@ -13,6 +13,7 @@ import * as Effect from "effect/Effect"; · 12 unmodified lines
13import * as Option from "effect/Option";13import * as Option from "effect/Option";
14 14
15import { loadLocalInstallation } from "./local-installation";15import { loadLocalInstallation } from "./local-installation";
16import { submitExecutionInteractionResponse } from "./execution-service";
16import { LiveExecutionManagerService } from "./live-execution";17import { LiveExecutionManagerService } from "./live-execution";
17import {18import {
18 operationErrors,19 operationErrors,
@@ -241,12 +242,28 @@ export const submitSourceCredentialInteraction = (input: { · 222 unmodified lines
241 });242 });
242 243
243 if (!resumed) {244 if (!resumed) {
244 return yield* Effect.fail(245 const persisted = yield* submitExecutionInteractionResponse({
245 localOps.sourceCredentialSubmit.badRequest(246 executionId: interaction.executionId,
246 "Source credential request is no longer resumable",247 response: {
247 `interactionId=${interaction.interactionId}`,248 action: "cancel",
249 },
250 }).pipe(
251 Effect.mapError((error) =>
252 localOps.sourceCredentialSubmit.unknownStorage(
253 error,
254 `Failed resuming execution for interaction ${interaction.interactionId}`,
255 )
248 ),256 ),
249 );257 );
258
259 if (!persisted) {
260 return yield* Effect.fail(
261 localOps.sourceCredentialSubmit.badRequest(
262 "Source credential request is no longer resumable",
263 `interactionId=${interaction.interactionId}`,
264 ),
265 );
266 }
250 }267 }
251 268
252 return {269 return {
@@ -266,12 +283,29 @@ export const submitSourceCredentialInteraction = (input: { · 13 unmodified lines
266 });283 });
267 284
268 if (!resumed) {285 if (!resumed) {
269 return yield* Effect.fail(286 const persisted = yield* submitExecutionInteractionResponse({
270 localOps.sourceCredentialSubmit.badRequest(287 executionId: interaction.executionId,
271 "Source credential request is no longer resumable",288 response: {
272 `interactionId=${interaction.interactionId}`,289 action: "accept",
290 content: createSourceCredentialSelectionNoneContent(),
291 },
292 }).pipe(
293 Effect.mapError((error) =>
294 localOps.sourceCredentialSubmit.unknownStorage(
295 error,
296 `Failed resuming execution for interaction ${interaction.interactionId}`,
297 )
273 ),298 ),
274 );299 );
300
301 if (!persisted) {
302 return yield* Effect.fail(
303 localOps.sourceCredentialSubmit.badRequest(
304 "Source credential request is no longer resumable",
305 `interactionId=${interaction.interactionId}`,
306 ),
307 );
308 }
275 }309 }
276 310
277 return {311 return {
@@ -314,12 +348,29 @@ export const submitSourceCredentialInteraction = (input: { · 36 unmodified lines
314 });348 });
315 349
316 if (!resumed) {350 if (!resumed) {
317 return yield* Effect.fail(351 const persisted = yield* submitExecutionInteractionResponse({
318 localOps.sourceCredentialSubmit.badRequest(352 executionId: interaction.executionId,
319 "Source credential request is no longer resumable",353 response: {
320 `interactionId=${interaction.interactionId}`,354 action: "accept",
355 content: createSourceCredentialSelectionBearerContent(tokenRef),
356 },
357 }).pipe(
358 Effect.mapError((error) =>
359 localOps.sourceCredentialSubmit.unknownStorage(
360 error,
361 `Failed resuming execution for interaction ${interaction.interactionId}`,
362 )
321 ),363 ),
322 );364 );
365
366 if (!persisted) {
367 return yield* Effect.fail(
368 localOps.sourceCredentialSubmit.badRequest(
369 "Source credential request is no longer resumable",
370 `interactionId=${interaction.interactionId}`,
371 ),
372 );
373 }
323 }374 }
324 375
325 return {376 return {
packages/control-plane/src/runtime/source-auth-service.ts+44−14
@@ -42,6 +42,7 @@ import * as Schema from "effect/Schema"; · 41 unmodified lines
42 42
43import {43import {
44 LiveExecutionManagerService,44 LiveExecutionManagerService,
45 sanitizePersistedElicitationResponse,
45 type LiveExecutionManager,46 type LiveExecutionManager,
46} from "./live-execution";47} from "./live-execution";
47import {48import {
@@ -302,6 +303,7 @@ const mergeOauth2PkceSourceAuthSessionData = (input: { · 254 unmodified lines
302 303
303 304
304const completeLiveInteraction = (input: {305const completeLiveInteraction = (input: {
306 rows: SqlControlPlaneRows;
305 liveExecutionManager: LiveExecutionManager;307 liveExecutionManager: LiveExecutionManager;
306 session: SourceAuthSession;308 session: SourceAuthSession;
307 response: {309 response: {
@@ -314,24 +316,50 @@ const completeLiveInteraction = (input: { · 6 unmodified lines
314 return;316 return;
315 }317 }
316 318
317 yield* input.liveExecutionManager.resolveInteraction({319 const response =
320 input.response.action === "accept"
321 ? { action: "accept" as const }
322 : {
323 action: "cancel" as const,
324 ...(input.response.reason
325 ? {
326 content: {
327 reason: input.response.reason,
328 },
329 }
330 : {}),
331 };
332
333 const resumed = yield* input.liveExecutionManager.resolveInteraction({
318 executionId: input.session.executionId,334 executionId: input.session.executionId,
319 response:335 response,
320 input.response.action === "accept"
321 ? { action: "accept" }
322 : {
323 action: "cancel",
324 ...(input.response.reason
325 ? {
326 content: {
327 reason: input.response.reason,
328 },
329 }
330 : {}),
331 },
332 });336 });
337
338 if (!resumed) {
339 const pendingInteraction = yield* input.rows.executionInteractions
340 .getPendingByExecutionId(input.session.executionId);
341
342 if (Option.isSome(pendingInteraction)) {
343 yield* input.rows.executionInteractions.update(pendingInteraction.value.id, {
344 status: response.action === "cancel" ? "cancelled" : "resolved",
345 responseJson: serializeJson(
346 sanitizePersistedElicitationResponse(response),
347 ),
348 responsePrivateJson: serializeJson(response),
349 updatedAt: Date.now(),
350 });
351 }
352 }
333 });353 });
334 354
355const serializeJson = (value: unknown): string | null => {
356 if (value === undefined) {
357 return null;
358 }
359
360 return JSON.stringify(value);
361};
362
335const updateSourceStatus = (rows: SqlControlPlaneRows, source: Source, input: {363const updateSourceStatus = (rows: SqlControlPlaneRows, source: Source, input: {
336 actorAccountId?: AccountId | null;364 actorAccountId?: AccountId | null;
337 status: Source["status"];365 status: Source["status"];
@@ -2006,6 +2034,7 @@ export const createRuntimeSourceAuthService = (input: { · 1668 unmodified lines
2006 resolveSecretMaterial,2034 resolveSecretMaterial,
2007 });2035 });
2008 yield* completeLiveInteraction({2036 yield* completeLiveInteraction({
2037 rows: input.rows,
2009 liveExecutionManager: input.liveExecutionManager,2038 liveExecutionManager: input.liveExecutionManager,
2010 session,2039 session,
2011 response: {2040 response: {
@@ -2193,6 +2222,7 @@ export const createRuntimeSourceAuthService = (input: { · 181 unmodified lines
2193 });2222 });
2194 2223
2195 yield* completeLiveInteraction({2224 yield* completeLiveInteraction({
2225 rows: input.rows,
2196 liveExecutionManager: input.liveExecutionManager,2226 liveExecutionManager: input.liveExecutionManager,
2197 session,2227 session,
2198 response: {2228 response: {
packages/control-plane/src/schema/ids.ts+4−0
@@ -31,6 +31,9 @@ export const ExecutionIdSchema = Schema.String.pipe(Schema.brand("ExecutionId")) · 30 unmodified lines
31export const ExecutionInteractionIdSchema = Schema.String.pipe(31export const ExecutionInteractionIdSchema = Schema.String.pipe(
32 Schema.brand("ExecutionInteractionId"),32 Schema.brand("ExecutionInteractionId"),
33);33);
34export const ExecutionStepIdSchema = Schema.String.pipe(
35 Schema.brand("ExecutionStepId"),
36);
34 37
35export type AccountId = typeof AccountIdSchema.Type;38export type AccountId = typeof AccountIdSchema.Type;
36export type WorkspaceId = typeof WorkspaceIdSchema.Type;39export type WorkspaceId = typeof WorkspaceIdSchema.Type;
@@ -47,3 +50,4 @@ export type SecretMaterialId = typeof SecretMaterialIdSchema.Type; · 10 unmodified lines
47export type PolicyId = typeof PolicyIdSchema.Type;50export type PolicyId = typeof PolicyIdSchema.Type;
48export type ExecutionId = typeof ExecutionIdSchema.Type;51export type ExecutionId = typeof ExecutionIdSchema.Type;
49export type ExecutionInteractionId = typeof ExecutionInteractionIdSchema.Type;52export type ExecutionInteractionId = typeof ExecutionInteractionIdSchema.Type;
53export type ExecutionStepId = typeof ExecutionStepIdSchema.Type;
packages/control-plane/src/schema/models/execution.ts+43−0
@@ -7,6 +7,7 @@ import { Schema } from "effect"; · 6 unmodified lines
7 7
8import {8import {
9 executionInteractionsTable,9 executionInteractionsTable,
10 executionStepsTable,
10 executionsTable,11 executionsTable,
11} from "../../persistence/schema";12} from "../../persistence/schema";
12import { TimestampMsSchema } from "../common";13import { TimestampMsSchema } from "../common";
@@ -14,6 +15,7 @@ import { · 1 unmodified line
14 AccountIdSchema,15 AccountIdSchema,
15 ExecutionIdSchema,16 ExecutionIdSchema,
16 ExecutionInteractionIdSchema,17 ExecutionInteractionIdSchema,
18 ExecutionStepIdSchema,
17 WorkspaceIdSchema,19 WorkspaceIdSchema,
18} from "../ids";20} from "../ids";
19 21
@@ -67,6 +69,7 @@ const executionInteractionSchemaOverrides = { · 47 unmodified lines
67 status: ExecutionInteractionStatusSchema,69 status: ExecutionInteractionStatusSchema,
68 purpose: Schema.String,70 purpose: Schema.String,
69 responseJson: Schema.NullOr(Schema.String),71 responseJson: Schema.NullOr(Schema.String),
72 responsePrivateJson: Schema.NullOr(Schema.String),
70 createdAt: TimestampMsSchema,73 createdAt: TimestampMsSchema,
71 updatedAt: TimestampMsSchema,74 updatedAt: TimestampMsSchema,
72} as const;75} as const;
@@ -91,8 +94,48 @@ export const ExecutionEnvelopeSchema = Schema.Struct({ · 18 unmodified lines
91 pendingInteraction: Schema.NullOr(ExecutionInteractionSchema),94 pendingInteraction: Schema.NullOr(ExecutionInteractionSchema),
92});95});
93 96
97export const ExecutionStepKindSchema = Schema.Literal("tool_call");
98
99export const ExecutionStepStatusSchema = Schema.Literal(
100 "pending",
101 "waiting",
102 "completed",
103 "failed",
104);
105
106const executionStepSchemaOverrides = {
107 id: ExecutionStepIdSchema,
108 executionId: ExecutionIdSchema,
109 sequence: Schema.Number,
110 kind: ExecutionStepKindSchema,
111 status: ExecutionStepStatusSchema,
112 interactionId: Schema.NullOr(ExecutionInteractionIdSchema),
113 resultJson: Schema.NullOr(Schema.String),
114 errorText: Schema.NullOr(Schema.String),
115 createdAt: TimestampMsSchema,
116 updatedAt: TimestampMsSchema,
117} as const;
118
119export const ExecutionStepSchema = createSelectSchema(
120 executionStepsTable,
121 executionStepSchemaOverrides,
122);
123
124export const ExecutionStepInsertSchema = createInsertSchema(
125 executionStepsTable,
126 executionStepSchemaOverrides,
127);
128
129export const ExecutionStepUpdateSchema = createUpdateSchema(
130 executionStepsTable,
131 executionStepSchemaOverrides,
132);
133
94export type ExecutionStatus = typeof ExecutionStatusSchema.Type;134export type ExecutionStatus = typeof ExecutionStatusSchema.Type;
95export type Execution = typeof ExecutionSchema.Type;135export type Execution = typeof ExecutionSchema.Type;
96export type ExecutionInteractionStatus = typeof ExecutionInteractionStatusSchema.Type;136export type ExecutionInteractionStatus = typeof ExecutionInteractionStatusSchema.Type;
97export type ExecutionInteraction = typeof ExecutionInteractionSchema.Type;137export type ExecutionInteraction = typeof ExecutionInteractionSchema.Type;
98export type ExecutionEnvelope = typeof ExecutionEnvelopeSchema.Type;138export type ExecutionEnvelope = typeof ExecutionEnvelopeSchema.Type;
139export type ExecutionStepKind = typeof ExecutionStepKindSchema.Type;
140export type ExecutionStepStatus = typeof ExecutionStepStatusSchema.Type;
141export type ExecutionStep = typeof ExecutionStepSchema.Type;
packages/executor-mcp/src/index.ts+18−9
@@ -117,6 +117,9 @@ const supportsManagedElicitation = (server: McpServer): boolean => { · 116 unmodified lines
117 return Boolean(capabilities?.elicitation?.form) && Boolean(capabilities?.elicitation?.url);117 return Boolean(capabilities?.elicitation?.form) && Boolean(capabilities?.elicitation?.url);
118};118};
119 119
120const interactionModeForServer = (server: McpServer): "live_form" | "detach" =>
121 supportsManagedElicitation(server) ? "live_form" : "detach";
122
120type CatalogLike = {123type CatalogLike = {
121 listNamespaces: (input: { limit: number }) => Effect.Effect<124 listNamespaces: (input: { limit: number }) => Effect.Effect<
122 ReadonlyArray<{ namespace: string; displayName?: string }>,125 ReadonlyArray<{ namespace: string; displayName?: string }>,
@@ -318,6 +321,7 @@ const driveExecutionWithElicitation = async (input: { · 195 unmodified lines
318 executionId: current.execution.id as never,321 executionId: current.execution.id as never,
319 payload: {322 payload: {
320 responseJson: JSON.stringify(response),323 responseJson: JSON.stringify(response),
324 interactionMode: interactionModeForServer(input.server),
321 },325 },
322 resumedByAccountId: input.accountId as never,326 resumedByAccountId: input.accountId as never,
323 }),327 }),
@@ -340,6 +344,7 @@ const driveExecutionWithElicitation = async (input: { · 16 unmodified lines
340 executionId: current.execution.id as never,344 executionId: current.execution.id as never,
341 payload: {345 payload: {
342 responseJson: JSON.stringify(response),346 responseJson: JSON.stringify(response),
347 interactionMode: interactionModeForServer(input.server),
343 },348 },
344 resumedByAccountId: input.accountId as never,349 resumedByAccountId: input.accountId as never,
345 }),350 }),
@@ -382,14 +387,15 @@ const driveExecutionWithoutElicitation = async (input: { · 36 unmodified lines
382 387
383 current = await runControlPlane(388 current = await runControlPlane(
384 input.runtime,389 input.runtime,
385 resumeExecution({390 resumeExecution({
386 workspaceId: input.workspaceId as never,391 workspaceId: input.workspaceId as never,
387 executionId: current.execution.id as never,392 executionId: current.execution.id as never,
388 payload: {393 payload: {
389 responseJson: JSON.stringify(response),394 responseJson: JSON.stringify(response),
390 },395 interactionMode: "detach",
391 resumedByAccountId: input.accountId as never,396 },
392 }),397 resumedByAccountId: input.accountId as never,
398 }),
393 );399 );
394 response = undefined;400 response = undefined;
395 }401 }
@@ -424,7 +430,10 @@ const createExecutorMcpServer = async (config: { · 28 unmodified lines
424 config.runtime,430 config.runtime,
425 createExecution({431 createExecution({
426 workspaceId,432 workspaceId,
427 payload: { code },433 payload: {
434 code,
435 interactionMode: interactionModeForServer(server),
436 },
428 createdByAccountId: accountId,437 createdByAccountId: accountId,
429 }),438 }),
430 );439 );