diff --git a/.changeset/execution-observer-foundation.md b/.changeset/execution-observer-foundation.md new file mode 100644 index 0000000000..4a4098b4ba --- /dev/null +++ b/.changeset/execution-observer-foundation.md @@ -0,0 +1,14 @@ +--- +"@executor-js/sdk": minor +"@executor-js/execution": minor +"@executor-js/api": minor +--- + +Add the execution-observer foundation. The execution engine now emits a typed +lifecycle stream (`ExecutionStarted`/`Finished`, `ToolCallStarted`/`Finished`, +`InteractionStarted`/`Resolved`), plugins can subscribe via the new +`plugin.runtime.executionObserver` hook, and `makeExecutionStack` composes every +registered plugin's observer onto the engine. Behaviour is unchanged when no +plugin observes, making this the opt-in seam the execution-history and +execution-metrics plugins build on. Also exposes `Executor.owner` and enriches +the `mcp.execute` span with the run id and trigger. diff --git a/apps/local/src/app.ts b/apps/local/src/app.ts index c565c14d4b..fca5bf1d34 100644 --- a/apps/local/src/app.ts +++ b/apps/local/src/app.ts @@ -11,6 +11,7 @@ import { import { withExecutionAnalytics } from "@executor-js/analytics"; import { createExecutionEngine } from "@executor-js/execution"; import { makeQuickJsExecutor } from "@executor-js/runtime-quickjs"; +import { composeExecutionObservers, type AnyPlugin } from "@executor-js/sdk"; import { localAnalytics } from "./analytics"; import { getExecutorBundle, type LocalExecutor } from "./executor"; @@ -53,7 +54,10 @@ import { ErrorCaptureLive } from "./observability"; * `HostConfig`/`CodeExecutorProvider` seams — the fixed executor is the whole * execution model. */ -const localFixedExecutionLayer = (executor: LocalExecutor): Layer.Layer => +const localFixedExecutionLayer = ( + executor: LocalExecutor, + plugins: readonly AnyPlugin[], +): Layer.Layer => Layer.succeed(FixedExecutionProvider)({ executor, // This engine serves the HTTP executions API (`executor call`/`resume`, @@ -62,6 +66,8 @@ const localFixedExecutionLayer = (executor: LocalExecutor): Layer.Layer decorate( - createExecutionEngine({ executor, codeExecutor }), + createExecutionEngine({ executor, codeExecutor, observer }), { accountId, organizationId, diff --git a/packages/core/execution/src/engine-observer.test.ts b/packages/core/execution/src/engine-observer.test.ts new file mode 100644 index 0000000000..7c7c7dc37d --- /dev/null +++ b/packages/core/execution/src/engine-observer.test.ts @@ -0,0 +1,215 @@ +import { describe, expect, it } from "@effect/vitest"; +import { Deferred, Effect, Fiber, Predicate, Schema } from "effect"; + +import { createExecutor, definePlugin, ElicitationResponse, tool } from "@executor-js/sdk"; +import type { ExecutionEvent, ExecutionObserver } from "@executor-js/sdk"; +import { makeTestConfig } from "@executor-js/sdk/testing"; +import type { CodeExecutor, ExecuteResult } from "@executor-js/codemode-core"; + +import { createExecutionEngine } from "./engine"; + +const emptyPlugin = definePlugin(() => ({ + id: "observer-test" as const, + storage: () => ({}), + staticIntegrations: () => [], +})); + +const approvalPlugin = definePlugin(() => ({ + id: "observer-approval-test" as const, + storage: () => ({}), + staticIntegrations: () => [ + { + id: "approval.ctl", + kind: "control" as const, + name: "Approval Ctl", + tools: [ + tool({ + name: "run", + description: "Requires approval", + annotations: { requiresApproval: true } as const, + inputSchema: Schema.toStandardSchemaV1(Schema.toStandardJSONSchemaV1(Schema.Struct({}))), + execute: () => Effect.succeed("ran"), + }), + ], + }, + ], +})); + +const makeExecutor = () => createExecutor(makeTestConfig({ plugins: [emptyPlugin()] as const })); + +const makeApprovalExecutor = () => + createExecutor(makeTestConfig({ plugins: [approvalPlugin()] as const })); + +// A code executor that issues one builtin tool call (tools.search) and then +// completes, enough to exercise the full event sequence. +const toolCallingExecutor: CodeExecutor = { + execute: (_code, invoker) => + invoker + .invoke({ path: "search", args: { query: "anything" } }) + .pipe(Effect.as({ result: "ok", logs: [] } satisfies ExecuteResult), Effect.orDie), +}; + +const approvalCallingExecutor: CodeExecutor = { + execute: (_code, invoker) => + invoker + .invoke({ path: "approval.ctl.run", args: {} }) + .pipe(Effect.as({ result: "ok", logs: [] } satisfies ExecuteResult), Effect.orDie), +}; + +const collectingObserver = () => { + const events: ExecutionEvent[] = []; + const observer: ExecutionObserver = { + handle: (event) => Effect.sync(() => void events.push(event)), + }; + return { events, observer }; +}; + +describe("execution engine observer emission", () => { + it.effect("emits the full lifecycle for a completed run with a tool call", () => + Effect.gen(function* () { + const executor = yield* makeExecutor(); + const { events, observer } = collectingObserver(); + const engine = createExecutionEngine({ + executor, + codeExecutor: toolCallingExecutor, + observer, + }); + + const result = yield* engine.executeWithPause("noop", { trigger: { kind: "test" } }); + expect(result.status).toBe("completed"); + + // First event opens the run, last closes it; tool calls land in between. + // `.find` with isTagged narrows each result, so the assertions read the + // typed fields directly via optional chaining (no conditional blocks). + const started = events.find((e) => Predicate.isTagged(e, "ExecutionStarted")); + const finished = events.find((e) => Predicate.isTagged(e, "ExecutionFinished")); + const toolStarted = events.find((e) => Predicate.isTagged(e, "ToolCallStarted")); + const toolFinished = events.find((e) => Predicate.isTagged(e, "ToolCallFinished")); + + expect(Predicate.isTagged(events[0], "ExecutionStarted")).toBe(true); + expect(Predicate.isTagged(events[events.length - 1], "ExecutionFinished")).toBe(true); + + expect(started?.trigger?.kind).toBe("test"); + expect(started?.owner.tenant).toBeDefined(); + expect(toolStarted).toBeDefined(); + expect(finished?.status).toBe("completed"); + + // Tool-call events share the run's executionId and carry the path. + expect(toolFinished?.path).toBe("search"); + expect(toolFinished?.status).toBe("completed"); + expect(toolFinished?.executionId).toBe(started?.executionId); + }), + ); + + it.effect("emits inline interaction events when execute handles elicitation", () => + Effect.gen(function* () { + const executor = yield* makeApprovalExecutor(); + const { events, observer } = collectingObserver(); + const engine = createExecutionEngine({ + executor, + codeExecutor: approvalCallingExecutor, + observer, + }); + + const result = yield* engine.execute("noop", { + trigger: { kind: "test" }, + onElicitation: () => Effect.succeed(ElicitationResponse.make({ action: "accept" })), + }); + expect(result.result).toBe("ok"); + + const started = events.find((e) => Predicate.isTagged(e, "ExecutionStarted")); + const interactionStarted = events.find((e) => Predicate.isTagged(e, "InteractionStarted")); + const interactionResolved = events.find((e) => Predicate.isTagged(e, "InteractionResolved")); + + expect(interactionStarted?.executionId).toBe(started?.executionId); + expect(interactionResolved?.executionId).toBe(started?.executionId); + expect(interactionResolved?.interactionId).toBe(interactionStarted?.interactionId); + expect(interactionStarted?.context.request.message).toContain("approval"); + expect(interactionResolved?.status).toBe("accepted"); + expect(interactionResolved?.response?.action).toBe("accept"); + }), + ); + + it.effect("records a ToolResult failure envelope as a failed tool call", () => + Effect.gen(function* () { + const executor = yield* makeExecutor(); + const { events, observer } = collectingObserver(); + // Invoke a path that resolves to `ToolResult.fail` on the success channel + // (an unknown tool is the cheapest such case). + const failingEnvelopeExecutor: CodeExecutor = { + execute: (code, invoker) => + invoker + .invoke({ path: "nope.missing.tool", args: {} }) + .pipe(Effect.as({ result: null, logs: [] } satisfies ExecuteResult), Effect.orDie), + }; + const engine = createExecutionEngine({ + executor, + codeExecutor: failingEnvelopeExecutor, + observer, + }); + yield* engine.executeWithPause("noop"); + + const toolFinished = events.find((e) => Predicate.isTagged(e, "ToolCallFinished")); + expect(toolFinished?.status).toBe("failed"); + expect(toolFinished?.error).toMatch(/^tool_not_found: /); + // The envelope itself stays attached for inspection. + expect(toolFinished?.result).toMatchObject({ ok: false }); + const finished = events.find((e) => Predicate.isTagged(e, "ExecutionFinished")); + expect(finished?.status).toBe("completed"); + }), + ); + + it.effect("carries emitted output on ExecutionFinished", () => + Effect.gen(function* () { + const executor = yield* makeExecutor(); + const { events, observer } = collectingObserver(); + const emittingExecutor: CodeExecutor = { + execute: () => + Effect.succeed({ + result: null, + output: [{ type: "content", content: { hello: "world" } }], + logs: ["[log] hi"], + } satisfies ExecuteResult), + }; + const engine = createExecutionEngine({ executor, codeExecutor: emittingExecutor, observer }); + yield* engine.executeWithPause("emit({ hello: 'world' })"); + + const finished = events.find((e) => Predicate.isTagged(e, "ExecutionFinished")); + expect(finished?.status).toBe("completed"); + expect(finished?.result).toBeNull(); + expect(finished?.output).toEqual([{ type: "content", content: { hello: "world" } }]); + expect(finished?.logs).toEqual(["[log] hi"]); + }), + ); + + it.effect("closes an interrupted run as `interrupted` via the exit finalizer", () => + Effect.gen(function* () { + const executor = yield* makeExecutor(); + const { events, observer } = collectingObserver(); + const started = yield* Deferred.make(); + // Never completes on its own; only engine.shutdown interrupts it. + const hangingExecutor: CodeExecutor = { + execute: () => Deferred.succeed(started, undefined).pipe(Effect.andThen(Effect.never)), + }; + const engine = createExecutionEngine({ executor, codeExecutor: hangingExecutor, observer }); + const run = yield* Effect.forkChild(engine.executeWithPause("while (true) {}")); + yield* Deferred.await(started); + yield* engine.shutdown; + yield* Fiber.await(run); + + const finished = events.find((e) => Predicate.isTagged(e, "ExecutionFinished")); + expect(finished?.status).toBe("interrupted"); + expect(finished?.result).toBeUndefined(); + expect(Predicate.isTagged(events[events.length - 1], "ExecutionFinished")).toBe(true); + }), + ); + + it.effect("does nothing observable when no observer is configured", () => + Effect.gen(function* () { + const executor = yield* makeExecutor(); + const engine = createExecutionEngine({ executor, codeExecutor: toolCallingExecutor }); + const result = yield* engine.executeWithPause("noop"); + expect(result.status).toBe("completed"); + }), + ); +}); diff --git a/packages/core/execution/src/engine.ts b/packages/core/execution/src/engine.ts index 8bb9bda071..4e2351442a 100644 --- a/packages/core/execution/src/engine.ts +++ b/packages/core/execution/src/engine.ts @@ -1,5 +1,5 @@ import { Deferred, Effect, Fiber, Predicate, Queue, Ref } from "effect"; -import type * as Cause from "effect/Cause"; +import * as Cause from "effect/Cause"; import * as Exit from "effect/Exit"; import type { @@ -8,6 +8,23 @@ import type { ElicitationResponse, ElicitationHandler, ElicitationContext, + ExecutionObserver, + ExecutionTrigger, +} from "@executor-js/sdk/core"; +import { + ExecutionId, + ExecutionInteractionId, + ExecutionToolCallId, + ExecutionFinished, + ExecutionStarted, + InteractionResolved, + InteractionStarted, + ToolCallFinished, + ToolCallStarted, + emitExecutionEvent, + isToolResult, + noopExecutionObserver, + withExecutionObserver, } from "@executor-js/sdk/core"; import { CurrentOrgWriteAccess, type OrgWriteAccessState } from "@executor-js/sdk/core"; import { CodeExecutionError } from "@executor-js/codemode-core"; @@ -31,6 +48,26 @@ export type ExecutionEngineConfig; readonly toolDiscoveryProvider?: ToolDiscoveryProvider; + /** Optional sink for execution lifecycle events. Defaults to a no-op, so a + * host that registers no observer pays only for constructing the events. */ + readonly observer?: ExecutionObserver; +}; + +/** Per-run options shared by both execute paths. */ +export type ExecutionRunOptions = { + /** What kicked off this run (e.g. `mcp.tool`, `api.http`); recorded on the + * `ExecutionStarted` event for downstream attribution. */ + readonly trigger?: ExecutionTrigger; +}; + +export type PausableExecutionOptions = ExecutionRunOptions & { + /** Treat the caller as the human approver and resolve every elicitation inline. */ + readonly autoApprove?: boolean; +}; + +/** Options for the inline-elicitation execute path. */ +export type InlineExecutionOptions = ExecutionRunOptions & { + readonly onElicitation: ElicitationHandler; }; export type ExecutionResult = @@ -477,7 +514,7 @@ export type ExecutionEngine */ readonly execute: ( code: string, - options: { readonly onElicitation: ElicitationHandler }, + options: InlineExecutionOptions, ) => Effect.Effect; /** @@ -493,7 +530,7 @@ export type ExecutionEngine */ readonly executeWithPause: ( code: string, - options?: { readonly autoApprove?: boolean }, + options?: PausableExecutionOptions, ) => Effect.Effect; /** @@ -598,6 +635,170 @@ export const createExecutionEngine = ExecutionId.make(`exec_${crypto.randomUUID()}`); + const makeToolCallId = (): ExecutionToolCallId => + ExecutionToolCallId.make(`tc_${crypto.randomUUID()}`); + const makeInteractionId = (): ExecutionInteractionId => + ExecutionInteractionId.make(`ix_${crypto.randomUUID()}`); + + const interactionStatusFromAction = (action: ResumeResponse["action"]) => + action === "accept" ? "accepted" : action === "decline" ? "declined" : "cancelled"; + + const finishFromResult = (executionId: ExecutionId, result: ExecuteResult): ExecutionFinished => + new ExecutionFinished({ + executionId, + owner, + status: result.error ? "failed" : "completed", + result: result.result, + output: result.output, + error: result.error, + logs: result.logs, + completedAt: new Date(), + }); + + // An interrupt-only cause is the run being torn down from outside (client + // abort, host backstop, sandbox shutdown), not the code failing: record it as + // `interrupted` so history never shows it as a code failure. + const finishFromCause = (executionId: ExecutionId, cause: Cause.Cause): ExecutionFinished => + new ExecutionFinished({ + executionId, + owner, + status: Cause.hasInterruptsOnly(cause) ? "interrupted" : "failed", + error: Cause.pretty(cause), + completedAt: new Date(), + }); + + /** Emit the terminal event for a sandbox run from its exit. Attached with + * `Effect.onExit` so it runs as a finalizer — uninterruptibly — and an + * interrupted run still closes in every observer instead of dangling as + * "running" forever. */ + const observeFinish = + (executionId: ExecutionId) => + (exit: Exit.Exit): Effect.Effect => + Exit.isSuccess(exit) + ? emitExecutionEvent(finishFromResult(executionId, exit.value)) + : emitExecutionEvent(finishFromCause(executionId, exit.cause)); + + /** Expected tool failures ride the success channel as `ToolResult.fail` + * envelopes; surface them to observers as failed calls (with the result kept + * for inspection) so history does not record an upstream 4xx as success. */ + const toolCallFinishedFromResult = ( + executionId: ExecutionId, + toolCallId: ExecutionToolCallId, + path: string, + result: unknown, + ): ToolCallFinished => + isToolResult(result) && !result.ok + ? new ToolCallFinished({ + executionId, + toolCallId, + owner, + path, + status: "failed", + result, + error: `${result.error.code}: ${result.error.message}`, + completedAt: new Date(), + }) + : new ToolCallFinished({ + executionId, + toolCallId, + owner, + path, + status: "completed", + result, + completedAt: new Date(), + }); + + /** Wrap an invoker so each tool call brackets `ToolCallStarted`/`Finished`. */ + const observeToolCalls = ( + executionId: ExecutionId, + inner: SandboxToolInvoker, + ): SandboxToolInvoker => ({ + invoke: (call) => + Effect.gen(function* () { + const toolCallId = makeToolCallId(); + yield* emitExecutionEvent( + new ToolCallStarted({ + executionId, + toolCallId, + owner, + path: call.path, + args: call.args, + startedAt: new Date(), + }), + ); + return yield* inner.invoke(call).pipe( + Effect.tap((result) => + emitExecutionEvent( + toolCallFinishedFromResult(executionId, toolCallId, call.path, result), + ), + ), + Effect.tapCause((cause) => + emitExecutionEvent( + new ToolCallFinished({ + executionId, + toolCallId, + owner, + path: call.path, + status: "failed", + error: Cause.pretty(cause), + completedAt: new Date(), + }), + ), + ), + ); + }), + }); + + /** Wrap an inline elicitation handler so it brackets `InteractionStarted`/ + * `Resolved`. The pausable path emits these directly (see below). */ + const observeInlineElicitation = + (executionId: ExecutionId, handler: ElicitationHandler): ElicitationHandler => + (ctx) => + Effect.gen(function* () { + const interactionId = makeInteractionId(); + yield* emitExecutionEvent( + new InteractionStarted({ + executionId, + interactionId, + owner, + context: ctx, + startedAt: new Date(), + }), + ); + return yield* handler(ctx).pipe( + Effect.tap((response) => + emitExecutionEvent( + new InteractionResolved({ + executionId, + interactionId, + owner, + status: interactionStatusFromAction(response.action), + response, + completedAt: new Date(), + }), + ), + ), + Effect.tapCause((cause) => + emitExecutionEvent( + new InteractionResolved({ + executionId, + interactionId, + owner, + status: "failed", + error: Cause.pretty(cause), + completedAt: new Date(), + }), + ), + ), + ); + }); + /** * Race a running fiber against the pause queue. Returns when either * the fiber completes or an elicitation handler fires (whichever @@ -642,7 +843,7 @@ export const createExecutionEngine = >(); @@ -677,6 +897,7 @@ export const createExecutionEngine = = { id, @@ -688,19 +909,57 @@ export const createExecutionEngine = + emitExecutionEvent( + new InteractionResolved({ + executionId, + interactionId, + owner, + status: interactionStatusFromAction(response.action), + response, + completedAt: new Date(), + }), + ), + ), + Effect.tapCause((cause) => + emitExecutionEvent( + new InteractionResolved({ + executionId, + interactionId, + owner, + status: "failed", + error: Cause.pretty(cause), + completedAt: new Date(), + }), + ), + ), + ); }); - const invoker = makeFullInvoker( - executor, - { onElicitation: elicitationHandler }, - toolDiscoveryProvider, + const invoker = observeToolCalls( + executionId, + makeFullInvoker(executor, { onElicitation: elicitationHandler }, toolDiscoveryProvider), ); fiber = yield* Effect.forkDetach( - codeExecutor.execute(code, invoker).pipe(Effect.withSpan("executor.code.exec")), + codeExecutor + .execute(code, invoker) + .pipe(Effect.withSpan("executor.code.exec"), Effect.onExit(observeFinish(executionId))), ); liveSandboxFibers.add(fiber); @@ -819,22 +1078,39 @@ export const createExecutionEngine = runInlineExecution(code, options).pipe(observeExecution), + executeWithPause: (code, options) => + startPausableExecution(code, options).pipe(observeExecution), + resume: (executionId, response) => + resumeExecution(executionId, response).pipe(observeExecution), shutdown, isExecutionSettled: (executionId) => Effect.sync(() => settledExecutionIds.has(executionId)), getPausedExecution: (executionId) => diff --git a/packages/core/sdk/src/execution-observer.test.ts b/packages/core/sdk/src/execution-observer.test.ts new file mode 100644 index 0000000000..c23612bbf9 --- /dev/null +++ b/packages/core/sdk/src/execution-observer.test.ts @@ -0,0 +1,141 @@ +import { describe, expect, it } from "@effect/vitest"; +import { Cause, Effect, Exit } from "effect"; + +import { Subject, Tenant } from "./ids"; +import { + ExecutionFinished, + ExecutionId, + composeExecutionObservers, + definePlugin, + emitExecutionEvent, + withExecutionObserver, +} from "./index"; + +const owner = { tenant: Tenant.make("tenant_test"), subject: Subject.make("subject_test") }; + +let calls: string[] = []; + +const observingPlugin = (id: string, asyncBoundary = false) => + definePlugin(() => ({ + id, + storage: () => ({}), + extension: () => ({ label: id }), + runtime: { + executionObserver: (self: { label: string }) => ({ + handle: () => + (asyncBoundary ? Effect.promise(() => Promise.resolve()) : Effect.void).pipe( + Effect.flatMap(() => Effect.sync(() => calls.push(self.label))), + ), + }), + }, + })); + +const failingPlugin = definePlugin(() => ({ + id: "failing" as const, + storage: () => ({}), + extension: () => ({ label: "failing" }), + runtime: { + executionObserver: () => ({ + handle: () => Effect.die("observer failed"), + }), + }, +})); + +const interruptingPlugin = definePlugin(() => ({ + id: "interrupting" as const, + storage: () => ({}), + extension: () => ({ label: "interrupting" }), + runtime: { + executionObserver: () => ({ + handle: () => Effect.interrupt, + }), + }, +})); + +const finishedEvent = () => + new ExecutionFinished({ + executionId: ExecutionId.make("exec_test"), + owner, + status: "completed", + result: "ok", + completedAt: new Date(), + }); + +describe("composeExecutionObservers", () => { + it.effect("emits events to the scoped observer", () => + Effect.gen(function* () { + calls = []; + yield* emitExecutionEvent(finishedEvent()).pipe( + withExecutionObserver({ + handle: () => Effect.sync(() => calls.push("observed")), + }), + ); + + expect(calls).toEqual(["observed"]); + }), + ); + + it.effect("dispatches observers sequentially and isolates failures", () => + Effect.gen(function* () { + calls = []; + const first = observingPlugin("first", true)(); + const failing = failingPlugin(); + const last = observingPlugin("last")(); + const observer = composeExecutionObservers([first, failing, last] as const, { + first: { label: "first" }, + failing: { label: "failing" }, + last: { label: "last" }, + }); + + // The failing plugin dies mid-dispatch; the others must still observe. + yield* observer.handle(finishedEvent()); + + expect(calls).toEqual(["first", "last"]); + }), + ); + + it.effect("preserves interrupts from scoped observers", () => + Effect.gen(function* () { + const exit = yield* Effect.exit( + emitExecutionEvent(finishedEvent()).pipe( + withExecutionObserver({ + handle: () => Effect.interrupt, + }), + ), + ); + + expect(Exit.isFailure(exit)).toBe(true); + if (!Exit.isFailure(exit)) return; + expect(Cause.hasInterrupts(exit.cause)).toBe(true); + }), + ); + + it.effect("preserves interrupts from composed plugin observers", () => + Effect.gen(function* () { + calls = []; + const interrupting = interruptingPlugin(); + const last = observingPlugin("last")(); + const observer = composeExecutionObservers([interrupting, last] as const, { + interrupting: { label: "interrupting" }, + last: { label: "last" }, + }); + + const exit = yield* Effect.exit(observer.handle(finishedEvent())); + + expect(Exit.isFailure(exit)).toBe(true); + if (!Exit.isFailure(exit)) return; + expect(Cause.hasInterrupts(exit.cause)).toBe(true); + expect(calls).toEqual([]); + }), + ); + + it.effect("returns a no-op observer when no plugin registers one", () => + Effect.gen(function* () { + const plain = definePlugin(() => ({ id: "plain", storage: () => ({}) }))(); + const observer = composeExecutionObservers([plain] as const, { plain: {} }); + + // No observer registered: handling is a no-op and never throws. + yield* observer.handle(finishedEvent()); + }), + ); +}); diff --git a/packages/core/sdk/src/execution-observer.ts b/packages/core/sdk/src/execution-observer.ts new file mode 100644 index 0000000000..ddb077383e --- /dev/null +++ b/packages/core/sdk/src/execution-observer.ts @@ -0,0 +1,212 @@ +import { Context, Data, Effect, Schema } from "effect"; +import * as Cause from "effect/Cause"; + +import type { ElicitationContext, ElicitationResponse } from "./elicitation"; +import type { AnyPlugin, OwnerBinding, PluginExtensions } from "./plugin"; + +/* The execution-observer contract: a pull-model lifecycle stream the engine + * emits as it runs code. Plugins opt in via `plugin.runtime.executionObserver` + * and receive every event; sinks (history, metrics, tracing) are built on top. + * Emission is dispatched to all registered observers with per-observer error + * logging, so an observer can never break an execution. */ + +export const ExecutionId = Schema.String.pipe(Schema.brand("ExecutionId")); +export type ExecutionId = typeof ExecutionId.Type; + +export const ExecutionToolCallId = Schema.String.pipe(Schema.brand("ExecutionToolCallId")); +export type ExecutionToolCallId = typeof ExecutionToolCallId.Type; + +export const ExecutionInteractionId = Schema.String.pipe(Schema.brand("ExecutionInteractionId")); +export type ExecutionInteractionId = typeof ExecutionInteractionId.Type; + +export type ExecutionTrigger = { + readonly kind: string; + readonly metadata?: Record; +}; + +export type ToolCallStatus = "completed" | "failed"; +export type InteractionStatus = "accepted" | "declined" | "cancelled" | "failed"; +/** + * How a run ended. `interrupted` is the run being torn down from outside + * (client abort, host backstop timeout, sandbox shutdown) rather than the code + * finishing or failing on its own; it carries no result and no logs. + */ +export type ExecutionStatus = "completed" | "failed" | "interrupted"; + +/** + * One item the code sent to the user through `emit()`. Mirrors the sandbox's + * `ExecuteOutputItem` structurally so the observer contract stays free of the + * kernel package: `content` is an arbitrary JSON value; `file` is a file + * reference (never inline bytes). + */ +export type ExecutionOutputItem = + | { readonly type: "content"; readonly content: unknown } + | { readonly type: "file"; readonly file: unknown }; + +export class ExecutionStarted extends Data.TaggedClass("ExecutionStarted")<{ + readonly executionId: ExecutionId; + readonly owner: OwnerBinding; + readonly code: string; + readonly trigger?: ExecutionTrigger; + readonly startedAt: Date; +}> {} + +export class ToolCallStarted extends Data.TaggedClass("ToolCallStarted")<{ + readonly executionId: ExecutionId; + readonly toolCallId: ExecutionToolCallId; + readonly owner: OwnerBinding; + readonly path: string; + readonly args: unknown; + readonly startedAt: Date; +}> {} + +export class ToolCallFinished extends Data.TaggedClass("ToolCallFinished")<{ + readonly executionId: ExecutionId; + readonly toolCallId: ExecutionToolCallId; + readonly owner: OwnerBinding; + readonly path: string; + readonly status: ToolCallStatus; + readonly result?: unknown; + readonly error?: string; + readonly completedAt: Date; +}> {} + +export class InteractionStarted extends Data.TaggedClass("InteractionStarted")<{ + readonly executionId: ExecutionId; + readonly interactionId: ExecutionInteractionId; + readonly owner: OwnerBinding; + readonly context: ElicitationContext; + readonly startedAt: Date; +}> {} + +export class InteractionResolved extends Data.TaggedClass("InteractionResolved")<{ + readonly executionId: ExecutionId; + readonly interactionId: ExecutionInteractionId; + readonly owner: OwnerBinding; + readonly status: InteractionStatus; + readonly response?: ElicitationResponse; + readonly error?: string; + readonly completedAt: Date; +}> {} + +export class ExecutionFinished extends Data.TaggedClass("ExecutionFinished")<{ + readonly executionId: ExecutionId; + readonly owner: OwnerBinding; + readonly status: ExecutionStatus; + /** The code's return value (`null`/absent when it only emitted). */ + readonly result?: unknown; + /** Everything the code sent to the user through `emit()`, in order. */ + readonly output?: readonly ExecutionOutputItem[]; + readonly error?: string; + readonly logs?: readonly string[]; + readonly completedAt: Date; +}> {} + +export type ExecutionEvent = + | ExecutionStarted + | ToolCallStarted + | ToolCallFinished + | InteractionStarted + | InteractionResolved + | ExecutionFinished; + +export interface ExecutionObserver { + readonly handle: (event: ExecutionEvent) => Effect.Effect; +} + +export const noopExecutionObserver: ExecutionObserver = { + handle: () => Effect.void, +}; + +const currentExecutionObserver = Context.Reference( + "@executor-js/sdk/ExecutionObserver", + { defaultValue: () => noopExecutionObserver }, +); + +type ExecutionEventName = ExecutionEvent["_tag"]; + +const executionEventName = (event: ExecutionEvent): ExecutionEventName => { + // oxlint-disable-next-line executor/no-manual-tag-check -- boundary: logging uses the Data.TaggedClass discriminant as an event name + return event._tag; +}; + +const logExecutionObserverFailure = ( + event: ExecutionEvent, + cause: Cause.Cause, + pluginId?: string, +): Effect.Effect => + Effect.logWarning("execution observer failed", { + cause: Cause.pretty(cause), + event: executionEventName(event), + ...(pluginId ? { pluginId } : {}), + }); + +const handleExecutionObserverCause = ( + event: ExecutionEvent, + cause: Cause.Cause, + pluginId?: string, +): Effect.Effect => + Cause.hasInterrupts(cause) + ? Effect.interrupt + : logExecutionObserverFailure(event, cause, pluginId); + +/** Emit an execution lifecycle event to the observer installed in the current + * Effect context. Defaults to a no-op when no observer is installed. */ +export const emitExecutionEvent = (event: ExecutionEvent): Effect.Effect => + Effect.service(currentExecutionObserver).pipe( + Effect.flatMap((observer) => observer.handle(event)), + ); + +/** Install an execution observer for the scoped Effect. Non-interrupt observer + * failures are logged and isolated; interrupt causes still propagate as + * cancellation. */ +export const withExecutionObserver = + (observer: ExecutionObserver) => + (effect: Effect.Effect): Effect.Effect => + effect.pipe( + Effect.provideService(currentExecutionObserver, { + handle: (event) => + observer + .handle(event) + .pipe(Effect.catchCause((cause) => handleExecutionObserverCause(event, cause))), + }), + ); + +/** Collect every plugin's `runtime.executionObserver` and fan each event to + * all of them, logging per-observer errors. Returns the no-op observer when no + * plugin registers one, the common opt-out case. */ +export const composeExecutionObservers = ( + plugins: TPlugins, + extensions: PluginExtensions, +): ExecutionObserver => { + const observers: { readonly pluginId: string; readonly observer: ExecutionObserver }[] = + []; + + for (const plugin of plugins) { + const observer = plugin.runtime?.executionObserver?.( + extensions[plugin.id as keyof PluginExtensions] as never, + ); + if (observer) { + observers.push({ pluginId: plugin.id, observer }); + } + } + + if (observers.length === 0) { + return noopExecutionObserver; + } + + return { + handle: (event) => + Effect.forEach( + observers, + ({ pluginId, observer }) => + observer + .handle(event) + .pipe( + Effect.catchCause((cause) => handleExecutionObserverCause(event, cause, pluginId)), + ), + // Preserve plugin order so observers see deterministic sequencing. + { discard: true }, + ), + }; +}; diff --git a/packages/core/sdk/src/executor.ts b/packages/core/sdk/src/executor.ts index 6916233cb5..3342a2364b 100644 --- a/packages/core/sdk/src/executor.ts +++ b/packages/core/sdk/src/executor.ts @@ -535,6 +535,11 @@ export type Executor = { ) => Effect.Effect; readonly close: () => Effect.Effect; + + /** The (tenant, subject) this executor acts as. Surfaced so engine-level + * machinery (e.g. execution observers) can attribute work to an owner + * without re-threading identity through every call site. */ + readonly owner: OwnerBinding; } & PluginExtensions; // --------------------------------------------------------------------------- @@ -7080,6 +7085,7 @@ export const createExecutor = => value as Executor; diff --git a/packages/core/sdk/src/index.ts b/packages/core/sdk/src/index.ts index d8ac973134..664447c8ac 100644 --- a/packages/core/sdk/src/index.ts +++ b/packages/core/sdk/src/index.ts @@ -235,6 +235,30 @@ export { type InvokeOptions, } from "./elicitation"; +// Execution observers: the engine lifecycle stream history/metrics/tracing build on. +export { + ExecutionId, + ExecutionToolCallId, + ExecutionInteractionId, + ExecutionStarted, + ToolCallStarted, + ToolCallFinished, + InteractionStarted, + InteractionResolved, + ExecutionFinished, + noopExecutionObserver, + composeExecutionObservers, + emitExecutionEvent, + withExecutionObserver, + type ExecutionTrigger, + type ToolCallStatus, + type InteractionStatus, + type ExecutionStatus, + type ExecutionOutputItem, + type ExecutionEvent, + type ExecutionObserver, +} from "./execution-observer"; + // Blob store — the plugin-facing contract (`BlobStore`/`PluginBlobStore`) // plus the platform-neutral backends (`makeFumaBlobStore` default, // `makeInMemoryBlobStore` for tests). Platform-specific backends live with diff --git a/packages/core/sdk/src/plugin.ts b/packages/core/sdk/src/plugin.ts index 2ace32f891..7c8020d53f 100644 --- a/packages/core/sdk/src/plugin.ts +++ b/packages/core/sdk/src/plugin.ts @@ -50,6 +50,7 @@ import type { InvalidConnectionInputError, OrgWriteDeniedError, } from "./errors"; +import type { ExecutionObserver } from "./execution-observer"; import type { OAuthService } from "./oauth-client"; import type { CredentialProvider, ProviderEntry } from "./provider"; import type { PluginStorageConfig, PluginStorageFacade } from "./plugin-storage"; @@ -841,6 +842,13 @@ export interface PluginSpec< | ((ctx: PluginCtx) => readonly CredentialProvider[]) | ((ctx: PluginCtx) => Effect.Effect); + /** Runtime hooks invoked while the engine executes code. `executionObserver` + * receives this plugin's extension and returns an observer for every + * {@link ExecutionEvent}, the seam history/metrics sinks build on. */ + readonly runtime?: { + readonly executionObserver?: (self: NoInfer) => ExecutionObserver; + }; + readonly close?: () => Effect.Effect; }