From b80f32ebb2ef2f5858965df2cc7d810c87ba120a Mon Sep 17 00:00:00 2001 From: Saatvik Arya Date: Sat, 13 Jun 2026 22:21:49 +0530 Subject: [PATCH 1/6] feat(sdk): add ExecutionObserver contract, events, and plugin runtime hook The pull-model execution lifecycle stream the engine emits: six tagged events (ExecutionStarted/Finished, ToolCallStarted/Finished, InteractionStarted/ Resolved), the ExecutionObserver interface, and composeExecutionObservers which fans each event to every plugin's runtime.executionObserver with per-observer error suppression. Events carry owner: OwnerBinding (tenant + subject) since the v2 restructure removed ScopeId. Adds the optional PluginSpec.runtime hook. --- .../core/sdk/src/execution-observer.test.ts | 72 +++++++++ packages/core/sdk/src/execution-observer.ts | 140 ++++++++++++++++++ packages/core/sdk/src/index.ts | 22 +++ packages/core/sdk/src/plugin.ts | 8 + 4 files changed, 242 insertions(+) create mode 100644 packages/core/sdk/src/execution-observer.test.ts create mode 100644 packages/core/sdk/src/execution-observer.ts 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..d8536f19fd --- /dev/null +++ b/packages/core/sdk/src/execution-observer.test.ts @@ -0,0 +1,72 @@ +import { describe, expect, it } from "@effect/vitest"; +import { Effect } from "effect"; + +import { Subject, Tenant } from "./ids"; +import { ExecutionFinished, ExecutionId, composeExecutionObservers, definePlugin } from "./index"; + +const owner = { tenant: Tenant.make("tenant_test"), subject: Subject.make("subject_test") }; + +let calls: string[] = []; + +const observingPlugin = (id: string) => + definePlugin(() => ({ + id, + storage: () => ({}), + extension: () => ({ label: id }), + runtime: { + executionObserver: (self: { label: string }) => ({ + handle: () => 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 finishedEvent = () => + new ExecutionFinished({ + executionId: ExecutionId.make("exec_test"), + owner, + status: "completed", + result: "ok", + completedAt: new Date(), + }); + +describe("composeExecutionObservers", () => { + it.effect("fans an event to every plugin observer and isolates failures", () => + Effect.gen(function* () { + calls = []; + const first = observingPlugin("first")(); + 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-fan; the others must still observe. + yield* observer.handle(finishedEvent()); + + expect(calls).toEqual(["first", "last"]); + }), + ); + + 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 silent no-op, 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..905391ff5b --- /dev/null +++ b/packages/core/sdk/src/execution-observer.ts @@ -0,0 +1,140 @@ +import { Data, Effect, Schema } from "effect"; + +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 fanned to all registered observers with per-observer error + * suppression, 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"; +export type ExecutionStatus = "completed" | "failed"; + +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; + readonly result?: unknown; + 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, +}; + +/** Wrap an observer so any failure (defect or expected error) is swallowed — + * an observer must never propagate into the execution it observes. */ +export const ignoreExecutionObserverErrors = ( + observer: ExecutionObserver, +): ExecutionObserver => ({ + handle: (event) => observer.handle(event).pipe(Effect.catchCause(() => Effect.void)), +}); + +/** Collect every plugin's `runtime.executionObserver` and fan each event to + * all of them, suppressing 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: ExecutionObserver[] = []; + + for (const plugin of plugins) { + const observer = plugin.runtime?.executionObserver?.( + extensions[plugin.id as keyof PluginExtensions] as never, + ); + if (observer) { + observers.push(observer); + } + } + + if (observers.length === 0) { + return noopExecutionObserver; + } + + return { + handle: (event) => + Effect.forEach( + observers, + (observer) => observer.handle(event).pipe(Effect.catchCause(() => Effect.void)), + { discard: true }, + ), + }; +}; diff --git a/packages/core/sdk/src/index.ts b/packages/core/sdk/src/index.ts index 8d6a95258f..4b822b4310 100644 --- a/packages/core/sdk/src/index.ts +++ b/packages/core/sdk/src/index.ts @@ -167,6 +167,28 @@ 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, + ignoreExecutionObserverErrors, + composeExecutionObservers, + type ExecutionTrigger, + type ToolCallStatus, + type InteractionStatus, + type ExecutionStatus, + 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 d48041f82b..6ab97f45ba 100644 --- a/packages/core/sdk/src/plugin.ts +++ b/packages/core/sdk/src/plugin.ts @@ -38,6 +38,7 @@ import type { IntegrationRemovalNotAllowedError, InvalidConnectionInputError, } 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"; @@ -499,6 +500,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 that is fanned + * every {@link ExecutionEvent} — the seam history/metrics sinks build on. */ + readonly runtime?: { + readonly executionObserver?: (self: NoInfer) => ExecutionObserver; + }; + readonly close?: () => Effect.Effect; } From 963558f145a2bdbb98e0abceb416f9259c06976a Mon Sep 17 00:00:00 2001 From: Saatvik Arya Date: Sat, 13 Jun 2026 22:21:49 +0530 Subject: [PATCH 2/6] feat(execution): emit lifecycle events from the engine and tag the run span createExecutionEngine takes an optional observer and emits the six ExecutionEvents around a run: ExecutionStarted (with trigger), tool calls bracketed via an invoker wrapper, elicitations via InteractionStarted/Resolved on both the inline and pause/resume paths, and ExecutionFinished on the forked fiber. The observer defaults to no-op, so hosts that register none are unaffected. Each run mints its own ExecutionId, distinct from the per-pause resume handle. execute/executeWithPause gain a trigger option, and the mcp.execute span is annotated with the run id + trigger so traces correlate to a run. Exposes Executor.owner so events can attribute work without re-threading identity. --- .../execution/src/engine-observer.test.ts | 81 ++++++ packages/core/execution/src/engine.ts | 263 ++++++++++++++++-- packages/core/sdk/src/executor.ts | 6 + 3 files changed, 332 insertions(+), 18 deletions(-) create mode 100644 packages/core/execution/src/engine-observer.test.ts 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..4c46ab8667 --- /dev/null +++ b/packages/core/execution/src/engine-observer.test.ts @@ -0,0 +1,81 @@ +import { describe, expect, it } from "@effect/vitest"; +import { Effect, Predicate } from "effect"; + +import { createExecutor, definePlugin } 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: () => ({}), + staticSources: () => [], +})); + +const makeExecutor = () => createExecutor(makeTestConfig({ plugins: [emptyPlugin()] 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 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("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 5dee95e036..711fe8ebe2 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 } from "effect"; -import type * as Cause from "effect/Cause"; +import * as Cause from "effect/Cause"; import type { Executor, @@ -7,6 +7,21 @@ import type { ElicitationResponse, ElicitationHandler, ElicitationContext, + ExecutionObserver, + ExecutionTrigger, +} from "@executor-js/sdk/core"; +import { + ExecutionId, + ExecutionInteractionId, + ExecutionToolCallId, + ExecutionFinished, + ExecutionStarted, + InteractionResolved, + InteractionStarted, + ToolCallFinished, + ToolCallStarted, + ignoreExecutionObserverErrors, + noopExecutionObserver, } from "@executor-js/sdk/core"; import { CodeExecutionError } from "@executor-js/codemode-core"; import type { CodeExecutor, ExecuteResult, SandboxToolInvoker } from "@executor-js/codemode-core"; @@ -29,6 +44,21 @@ 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; +}; + +/** Options for the inline-elicitation execute path. */ +export type InlineExecutionOptions = ExecutionRunOptions & { + readonly onElicitation: ElicitationHandler; }; export type ExecutionResult = @@ -341,7 +371,7 @@ export type ExecutionEngine */ readonly execute: ( code: string, - options: { readonly onElicitation: ElicitationHandler }, + options: InlineExecutionOptions, ) => Effect.Effect; /** @@ -349,7 +379,10 @@ export type ExecutionEngine * Use this when the host doesn't support inline elicitation. * Returns either a completed result or a paused execution that can be resumed. */ - readonly executeWithPause: (code: string) => Effect.Effect; + readonly executeWithPause: ( + code: string, + options?: ExecutionRunOptions, + ) => Effect.Effect; /** * Resume a paused execution. Returns a completed result, a new pause, or @@ -379,6 +412,134 @@ export const createExecutionEngine = >(); let nextId = 0; + // Observer wiring. `emit` never fails (errors are swallowed per-observer) and + // resolves to a no-op when no plugin registered one — the opt-out default. + const observer = ignoreExecutionObserverErrors(config.observer ?? noopExecutionObserver); + const emit = observer.handle; + const owner = executor.owner; + + const makeExecutionId = (): ExecutionId => 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, + error: result.error, + logs: result.logs, + completedAt: new Date(), + }); + + const finishFromCause = (executionId: ExecutionId, cause: Cause.Cause): ExecutionFinished => + new ExecutionFinished({ + executionId, + owner, + status: "failed", + error: Cause.pretty(cause), + 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* emit( + new ToolCallStarted({ + executionId, + toolCallId, + owner, + path: call.path, + args: call.args, + startedAt: new Date(), + }), + ); + return yield* inner.invoke(call).pipe( + Effect.tap((result) => + emit( + new ToolCallFinished({ + executionId, + toolCallId, + owner, + path: call.path, + status: "completed", + result, + completedAt: new Date(), + }), + ), + ), + Effect.tapCause((cause) => + emit( + 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* emit( + new InteractionStarted({ + executionId, + interactionId, + owner, + context: ctx, + startedAt: new Date(), + }), + ); + return yield* handler(ctx).pipe( + Effect.tap((response) => + emit( + new InteractionResolved({ + executionId, + interactionId, + owner, + status: interactionStatusFromAction(response.action), + response, + completedAt: new Date(), + }), + ), + ), + Effect.tapCause((cause) => + emit( + 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 @@ -411,12 +572,33 @@ export const createExecutionEngine = >(); @@ -428,6 +610,7 @@ export const createExecutionEngine = (); const id = `exec_${++nextId}`; + const interactionId = makeInteractionId(); const paused: InternalPausedExecution = { id, @@ -438,19 +621,42 @@ export const createExecutionEngine = emit(finishFromResult(executionId, result))), + Effect.tapCause((cause) => emit(finishFromCause(executionId, cause))), + ), ); return (yield* awaitCompletionOrPause(fiber, pauseQueue)) as ExecutionResult; @@ -486,20 +692,41 @@ export const createExecutionEngine = emit(finishFromResult(executionId, result))), + Effect.tapCause((cause) => emit(finishFromCause(executionId, cause))), ); - return yield* codeExecutor.execute(code, invoker).pipe(Effect.withSpan("executor.code.exec")); }); return { diff --git a/packages/core/sdk/src/executor.ts b/packages/core/sdk/src/executor.ts index c8d578bf8b..a90003ef92 100644 --- a/packages/core/sdk/src/executor.ts +++ b/packages/core/sdk/src/executor.ts @@ -310,6 +310,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; export interface ExecutorDb { @@ -3124,6 +3129,7 @@ export const createExecutor = => value as Executor; From 8c216a708e22b47ee0c3aed399048fdc07debf15 Mon Sep 17 00:00:00 2001 From: Saatvik Arya Date: Sat, 13 Jun 2026 22:21:49 +0530 Subject: [PATCH 3/6] feat(api): compose plugin execution observers in makeExecutionStack The shared (user, org) -> { executor, engine } factory composes every registered plugin's runtime.executionObserver and passes it to the engine. Cloud, self-host, host-cloudflare and the MCP Durable Objects all build through this factory, so one seam wires the observer for every HTTP and DO path. No plugin observing = no-op. --- packages/core/api/src/server/execution-stack.ts | 13 ++++++++++++- 1 file changed, 12 insertions(+), 1 deletion(-) diff --git a/packages/core/api/src/server/execution-stack.ts b/packages/core/api/src/server/execution-stack.ts index 6e58801fcf..ebfe63bf63 100644 --- a/packages/core/api/src/server/execution-stack.ts +++ b/packages/core/api/src/server/execution-stack.ts @@ -26,6 +26,7 @@ import { Context, Effect, Layer } from "effect"; import type * as Cause from "effect/Cause"; +import { composeExecutionObservers } from "@executor-js/sdk"; import type { AnyPlugin, Executor, StorageFailure } from "@executor-js/sdk"; import { createExecutionEngine, @@ -108,7 +109,17 @@ export const makeExecutionStack = < ); const codeExecutor = yield* CodeExecutorProvider; const { decorate } = yield* EngineDecorator; - const engine = decorate(createExecutionEngine({ executor, codeExecutor }), { + + // Compose every registered plugin's runtime.executionObserver, bound to the + // executor's own extensions. Resolves to a no-op when no plugin observes — + // so a stack with no history/metrics plugin is unaffected. `plugins()` is + // the erased AnyPlugin[]; the caller's TPlugins phantom recovers the tuple. + const { plugins } = yield* PluginsProvider; + // PluginsProvider erases the tuple to AnyPlugin[]; recover the caller's + // TPlugins phantom so the extensions arg (the executor) lines up. + const observer = composeExecutionObservers(plugins() as TPlugins, executor); + + const engine = decorate(createExecutionEngine({ executor, codeExecutor, observer }), { accountId, organizationId, organizationName, From 6327dcfc4d760f55b03fd82315984d76e11598e9 Mon Sep 17 00:00:00 2001 From: Saatvik Arya Date: Sat, 13 Jun 2026 22:21:49 +0530 Subject: [PATCH 4/6] feat(local): compose execution observers in local's engine sites Local's fixed-execution seam and in-process MCP handler build the engine directly rather than through makeExecutionStack, so compose the observer at both. Other hosts inherit it from the shared factory. --- apps/local/src/app.ts | 10 ++++++++-- apps/local/src/main.ts | 5 ++++- 2 files changed, 12 insertions(+), 3 deletions(-) diff --git a/apps/local/src/app.ts b/apps/local/src/app.ts index f9bddc6e86..4510997e64 100644 --- a/apps/local/src/app.ts +++ b/apps/local/src/app.ts @@ -9,6 +9,7 @@ import { } from "@executor-js/api/server"; import { createExecutionEngine } from "@executor-js/execution"; import { makeQuickJsExecutor } from "@executor-js/runtime-quickjs"; +import { composeExecutionObservers, type AnyPlugin } from "@executor-js/sdk"; import { getExecutorBundle, type LocalExecutor } from "./executor"; import { localIdentityLayer } from "./identity"; @@ -50,12 +51,17 @@ 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, engine: createExecutionEngine({ executor, codeExecutor: makeQuickJsExecutor(), + // Local bypasses makeExecutionStack, so compose plugin observers here. + observer: composeExecutionObservers(plugins, executor), }), // The executor IS its own plugin-extension map (`executor[pluginId]`); the // fixed middleware reads `executor[id]` to satisfy each plugin's @@ -86,7 +92,7 @@ export const makeLocalApiHandler = async (): Promise => { // Layer is the `fixedExecution` seam declaration AND lives in `boot` so the // fixed middleware's residual `FixedExecutionProvider` resolves there — exactly // as self-host declares `db: SelfHostDbProvider` and puts the handle in `boot`. - const fixedExecution = localFixedExecutionLayer(executor); + const fixedExecution = localFixedExecutionLayer(executor, plugins); const { toWebHandler } = ExecutorApp.make({ plugins, diff --git a/apps/local/src/main.ts b/apps/local/src/main.ts index bce001a3a6..cf5122323e 100644 --- a/apps/local/src/main.ts +++ b/apps/local/src/main.ts @@ -2,6 +2,7 @@ import { Context, Effect, Layer, ManagedRuntime } from "effect"; import { createExecutionEngine } from "@executor-js/execution"; import { makeQuickJsExecutor } from "@executor-js/runtime-quickjs"; +import { composeExecutionObservers } from "@executor-js/sdk"; import { makeLocalApiHandler } from "./app"; import { getExecutorBundle } from "./executor"; import { createMcpRequestHandler, type McpRequestHandler } from "./mcp"; @@ -61,10 +62,12 @@ export const createServerHandlers = async (): Promise => { // engine instance (the browser-approval + stdio surface is local-only and not // part of the shared API). Reuse the shared boot bundle so the MCP executor is // byte-identical to the one the API serves. - const { executor } = await getExecutorBundle(); + const { executor, plugins } = await getExecutorBundle(); const engine = createExecutionEngine({ executor, codeExecutor: makeQuickJsExecutor(), + // Local's in-process MCP also bypasses makeExecutionStack — compose here. + observer: composeExecutionObservers(plugins, executor), }); const mcp = createMcpRequestHandler({ engine }); From 11d40f4026c76115f68d0c2bcddf4cc6e35d1e3f Mon Sep 17 00:00:00 2001 From: Saatvik Arya Date: Sat, 13 Jun 2026 22:21:49 +0530 Subject: [PATCH 5/6] chore: add changeset for the execution-observer foundation --- .changeset/execution-observer-foundation.md | 14 ++++++++++++++ 1 file changed, 14 insertions(+) create mode 100644 .changeset/execution-observer-foundation.md diff --git a/.changeset/execution-observer-foundation.md b/.changeset/execution-observer-foundation.md new file mode 100644 index 0000000000..4d117c72b5 --- /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 — this is 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. From 8b57284f27f1824421641519720fadb852ea0c2b Mon Sep 17 00:00:00 2001 From: Saatvik Arya Date: Sun, 14 Jun 2026 01:55:09 +0530 Subject: [PATCH 6/6] fix(execution): parallelize observer fan-out and balance interaction events on interrupt MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Addresses Greptile review on #12: - composeExecutionObservers fans events to observers with concurrency: unbounded, so a slow sink (DB-backed history) no longer serializes behind another (metrics push). Per-observer error isolation is unchanged. - The pausable elicitation path now emits InteractionResolved on the failure/interrupt branch too (via tapCause), so a fiber killed while blocked on the response Deferred still pairs its InteractionStarted — matching the inline path. --- packages/core/execution/src/engine.ts | 41 +++++++++++++++------ packages/core/sdk/src/execution-observer.ts | 5 ++- 2 files changed, 33 insertions(+), 13 deletions(-) diff --git a/packages/core/execution/src/engine.ts b/packages/core/execution/src/engine.ts index 711fe8ebe2..b67151118e 100644 --- a/packages/core/execution/src/engine.ts +++ b/packages/core/execution/src/engine.ts @@ -632,19 +632,36 @@ export const createExecutionEngine = + emit( + new InteractionResolved({ + executionId, + interactionId, + owner, + status: interactionStatusFromAction(response.action), + response, + completedAt: new Date(), + }), + ), + ), + Effect.tapCause((cause) => + emit( + new InteractionResolved({ + executionId, + interactionId, + owner, + status: "failed", + error: Cause.pretty(cause), + completedAt: new Date(), + }), + ), + ), ); - return response; }); const invoker = observeToolCalls( diff --git a/packages/core/sdk/src/execution-observer.ts b/packages/core/sdk/src/execution-observer.ts index 905391ff5b..2c19bcb0f5 100644 --- a/packages/core/sdk/src/execution-observer.ts +++ b/packages/core/sdk/src/execution-observer.ts @@ -134,7 +134,10 @@ export const composeExecutionObservers = Effect.forEach( observers, (observer) => observer.handle(event).pipe(Effect.catchCause(() => Effect.void)), - { discard: true }, + // Fan out in parallel — a slow sink (e.g. a DB-backed history observer) + // must not serialize behind another (e.g. a metrics push). Per-observer + // error isolation is preserved by the catchCause above. + { discard: true, concurrency: "unbounded" }, ), }; };