From 03d78201573b2df370b0864d8b7e0474ec9c9209 Mon Sep 17 00:00:00 2001 From: Saatvik Arya Date: Tue, 23 Jun 2026 20:52:29 +0530 Subject: [PATCH 1/8] feat(execution): add execution observer foundation --- .changeset/execution-observer-foundation.md | 14 + apps/local/src/app.ts | 10 +- apps/local/src/main.ts | 5 +- .../core/api/src/server/execution-stack.ts | 5 +- .../execution/src/engine-observer.test.ts | 81 +++++ packages/core/execution/src/engine.ts | 284 ++++++++++++++++-- .../core/sdk/src/execution-observer.test.ts | 75 +++++ packages/core/sdk/src/execution-observer.ts | 162 ++++++++++ packages/core/sdk/src/executor.ts | 6 + packages/core/sdk/src/index.ts | 22 ++ packages/core/sdk/src/plugin.ts | 8 + 11 files changed, 647 insertions(+), 25 deletions(-) create mode 100644 .changeset/execution-observer-foundation.md create mode 100644 packages/core/execution/src/engine-observer.test.ts create mode 100644 packages/core/sdk/src/execution-observer.test.ts create mode 100644 packages/core/sdk/src/execution-observer.ts 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..924ec569e2 --- /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 8bb9bda071..8a76be7887 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,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 { CurrentOrgWriteAccess, type OrgWriteAccessState } from "@executor-js/sdk/core"; import { CodeExecutionError } from "@executor-js/codemode-core"; @@ -31,6 +46,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 +512,7 @@ export type ExecutionEngine */ readonly execute: ( code: string, - options: { readonly onElicitation: ElicitationHandler }, + options: InlineExecutionOptions, ) => Effect.Effect; /** @@ -493,7 +528,7 @@ export type ExecutionEngine */ readonly executeWithPause: ( code: string, - options?: { readonly autoApprove?: boolean }, + options?: PausableExecutionOptions, ) => Effect.Effect; /** @@ -598,6 +633,134 @@ 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, + 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 @@ -642,7 +805,7 @@ export const createExecutionEngine = >(); @@ -677,6 +859,7 @@ export const createExecutionEngine = = { id, @@ -688,19 +871,59 @@ 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(), + }), + ), + ), + ); }); - 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.tap((result) => emit(finishFromResult(executionId, result))), + Effect.tapCause((cause) => emit(finishFromCause(executionId, cause))), + ), ); liveSandboxFibers.add(fiber); @@ -819,22 +1042,41 @@ export const createExecutionEngine = emit(finishFromResult(executionId, result))), + Effect.tapCause((cause) => emit(finishFromCause(executionId, cause))), ); - const result = yield* codeExecutor - .execute(code, invoker) - .pipe(Effect.withSpan("executor.code.exec")); yield* annotateExecuteOutcome(result); return result; }); 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..d18e647ea4 --- /dev/null +++ b/packages/core/sdk/src/execution-observer.test.ts @@ -0,0 +1,75 @@ +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, 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 finishedEvent = () => + new ExecutionFinished({ + executionId: ExecutionId.make("exec_test"), + owner, + status: "completed", + result: "ok", + completedAt: new Date(), + }); + +describe("composeExecutionObservers", () => { + 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("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..a8ba7fa81a --- /dev/null +++ b/packages/core/sdk/src/execution-observer.ts @@ -0,0 +1,162 @@ +import { 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"; +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, +}; + +const logExecutionObserverFailure = ( + event: ExecutionEvent, + cause: Cause.Cause, + pluginId?: string, +): Effect.Effect => + Effect.logWarning("execution observer failed", { + cause: Cause.pretty(cause), + event: event._tag, + ...(pluginId ? { pluginId } : {}), + }); + +/** Wrap an observer so any failure (defect or expected error) is logged, and + * an observer must never propagate into the execution it observes. */ +export const ignoreExecutionObserverErrors = ( + observer: ExecutionObserver, +): ExecutionObserver => ({ + handle: (event) => + observer + .handle(event) + .pipe(Effect.catchCause((cause) => logExecutionObserverFailure(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) => logExecutionObserverFailure(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..3aba1a20e5 100644 --- a/packages/core/sdk/src/index.ts +++ b/packages/core/sdk/src/index.ts @@ -235,6 +235,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 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; } From 989dc9dc23e66769079449730096a59bc5d063a7 Mon Sep 17 00:00:00 2001 From: Saatvik Arya Date: Tue, 23 Jun 2026 20:54:38 +0530 Subject: [PATCH 2/8] fix(sdk): avoid manual observer tag logging --- packages/core/sdk/src/execution-observer.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/packages/core/sdk/src/execution-observer.ts b/packages/core/sdk/src/execution-observer.ts index a8ba7fa81a..7ff38a9eb7 100644 --- a/packages/core/sdk/src/execution-observer.ts +++ b/packages/core/sdk/src/execution-observer.ts @@ -107,7 +107,7 @@ const logExecutionObserverFailure = ( ): Effect.Effect => Effect.logWarning("execution observer failed", { cause: Cause.pretty(cause), - event: event._tag, + event: event.constructor.name, ...(pluginId ? { pluginId } : {}), }); From 0a360179c899f6a3245c4f645010dda19fd91169 Mon Sep 17 00:00:00 2001 From: Saatvik Arya Date: Wed, 24 Jun 2026 12:35:22 +0530 Subject: [PATCH 3/8] fix(execution): address observer review feedback --- .../execution/src/engine-observer.test.ts | 64 ++++++++++++++++++- packages/core/sdk/src/execution-observer.ts | 15 ++++- 2 files changed, 75 insertions(+), 4 deletions(-) diff --git a/packages/core/execution/src/engine-observer.test.ts b/packages/core/execution/src/engine-observer.test.ts index 924ec569e2..03b0ad4351 100644 --- a/packages/core/execution/src/engine-observer.test.ts +++ b/packages/core/execution/src/engine-observer.test.ts @@ -1,7 +1,7 @@ import { describe, expect, it } from "@effect/vitest"; -import { Effect, Predicate } from "effect"; +import { Effect, Predicate, Schema } from "effect"; -import { createExecutor, definePlugin } from "@executor-js/sdk"; +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"; @@ -14,8 +14,32 @@ const emptyPlugin = definePlugin(() => ({ staticSources: () => [], })); +const approvalPlugin = definePlugin(() => ({ + id: "observer-approval-test" as const, + storage: () => ({}), + staticSources: () => [ + { + 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 = { @@ -25,6 +49,13 @@ const toolCallingExecutor: CodeExecutor = { .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 = { @@ -70,6 +101,35 @@ describe("execution engine observer emission", () => { }), ); + 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("does nothing observable when no observer is configured", () => Effect.gen(function* () { const executor = yield* makeExecutor(); diff --git a/packages/core/sdk/src/execution-observer.ts b/packages/core/sdk/src/execution-observer.ts index 7ff38a9eb7..3cc397cb00 100644 --- a/packages/core/sdk/src/execution-observer.ts +++ b/packages/core/sdk/src/execution-observer.ts @@ -1,4 +1,4 @@ -import { Data, Effect, Schema } from "effect"; +import { Data, Effect, Predicate, Schema } from "effect"; import * as Cause from "effect/Cause"; import type { ElicitationContext, ElicitationResponse } from "./elicitation"; @@ -100,6 +100,17 @@ export const noopExecutionObserver: ExecutionObserver = { handle: () => Effect.void, }; +type ExecutionEventName = ExecutionEvent["_tag"]; + +const executionEventName = (event: ExecutionEvent): ExecutionEventName => { + if (Predicate.isTagged(event, "ExecutionStarted")) return "ExecutionStarted"; + if (Predicate.isTagged(event, "ToolCallStarted")) return "ToolCallStarted"; + if (Predicate.isTagged(event, "ToolCallFinished")) return "ToolCallFinished"; + if (Predicate.isTagged(event, "InteractionStarted")) return "InteractionStarted"; + if (Predicate.isTagged(event, "InteractionResolved")) return "InteractionResolved"; + return "ExecutionFinished"; +}; + const logExecutionObserverFailure = ( event: ExecutionEvent, cause: Cause.Cause, @@ -107,7 +118,7 @@ const logExecutionObserverFailure = ( ): Effect.Effect => Effect.logWarning("execution observer failed", { cause: Cause.pretty(cause), - event: event.constructor.name, + event: executionEventName(event), ...(pluginId ? { pluginId } : {}), }); From 6d05786b4145cf3089919e738aa0ae85a47bec0b Mon Sep 17 00:00:00 2001 From: Saatvik Arya Date: Wed, 24 Jun 2026 15:10:05 +0530 Subject: [PATCH 4/8] fix(execution): preserve observer interrupts --- .../core/sdk/src/execution-observer.test.ts | 54 ++++++++++++++++++- packages/core/sdk/src/execution-observer.ts | 13 ++++- 2 files changed, 63 insertions(+), 4 deletions(-) diff --git a/packages/core/sdk/src/execution-observer.test.ts b/packages/core/sdk/src/execution-observer.test.ts index d18e647ea4..f9cb6b571c 100644 --- a/packages/core/sdk/src/execution-observer.test.ts +++ b/packages/core/sdk/src/execution-observer.test.ts @@ -1,8 +1,14 @@ import { describe, expect, it } from "@effect/vitest"; -import { Effect } from "effect"; +import { Cause, Effect, Exit } from "effect"; import { Subject, Tenant } from "./ids"; -import { ExecutionFinished, ExecutionId, composeExecutionObservers, definePlugin } from "./index"; +import { + ExecutionFinished, + ExecutionId, + composeExecutionObservers, + definePlugin, + ignoreExecutionObserverErrors, +} from "./index"; const owner = { tenant: Tenant.make("tenant_test"), subject: Subject.make("subject_test") }; @@ -34,6 +40,17 @@ const failingPlugin = definePlugin(() => ({ }, })); +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"), @@ -63,6 +80,39 @@ describe("composeExecutionObservers", () => { }), ); + it.effect("preserves interrupts from isolated observers", () => + Effect.gen(function* () { + const observer = ignoreExecutionObserverErrors({ + handle: () => Effect.interrupt, + }); + + 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); + }), + ); + + 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: () => ({}) }))(); diff --git a/packages/core/sdk/src/execution-observer.ts b/packages/core/sdk/src/execution-observer.ts index 3cc397cb00..8cf01d7317 100644 --- a/packages/core/sdk/src/execution-observer.ts +++ b/packages/core/sdk/src/execution-observer.ts @@ -122,6 +122,15 @@ const logExecutionObserverFailure = ( ...(pluginId ? { pluginId } : {}), }); +const handleExecutionObserverCause = ( + event: ExecutionEvent, + cause: Cause.Cause, + pluginId?: string, +): Effect.Effect => + Cause.hasInterrupts(cause) + ? Effect.interrupt + : logExecutionObserverFailure(event, cause, pluginId); + /** Wrap an observer so any failure (defect or expected error) is logged, and * an observer must never propagate into the execution it observes. */ export const ignoreExecutionObserverErrors = ( @@ -130,7 +139,7 @@ export const ignoreExecutionObserverErrors = ( handle: (event) => observer .handle(event) - .pipe(Effect.catchCause((cause) => logExecutionObserverFailure(event, cause))), + .pipe(Effect.catchCause((cause) => handleExecutionObserverCause(event, cause))), }); /** Collect every plugin's `runtime.executionObserver` and fan each event to @@ -164,7 +173,7 @@ export const composeExecutionObservers = observer .handle(event) .pipe( - Effect.catchCause((cause) => logExecutionObserverFailure(event, cause, pluginId)), + Effect.catchCause((cause) => handleExecutionObserverCause(event, cause, pluginId)), ), // Preserve plugin order so observers see deterministic sequencing. { discard: true }, From a1db54edf1562e7dc0417fac1b7541d5744ea957 Mon Sep 17 00:00:00 2001 From: Saatvik Arya Date: Wed, 24 Jun 2026 16:06:09 +0530 Subject: [PATCH 5/8] fix(execution): clarify observer wrapper contract --- packages/core/execution/src/engine.ts | 4 ++-- .../core/sdk/src/execution-observer.test.ts | 4 ++-- packages/core/sdk/src/execution-observer.ts | 18 ++++++------------ packages/core/sdk/src/index.ts | 2 +- 4 files changed, 11 insertions(+), 17 deletions(-) diff --git a/packages/core/execution/src/engine.ts b/packages/core/execution/src/engine.ts index 8a76be7887..104acf5734 100644 --- a/packages/core/execution/src/engine.ts +++ b/packages/core/execution/src/engine.ts @@ -21,8 +21,8 @@ import { InteractionStarted, ToolCallFinished, ToolCallStarted, - ignoreExecutionObserverErrors, noopExecutionObserver, + wrapExecutionObserver, } from "@executor-js/sdk/core"; import { CurrentOrgWriteAccess, type OrgWriteAccessState } from "@executor-js/sdk/core"; import { CodeExecutionError } from "@executor-js/codemode-core"; @@ -635,7 +635,7 @@ export const createExecutionEngine = { it.effect("preserves interrupts from isolated observers", () => Effect.gen(function* () { - const observer = ignoreExecutionObserverErrors({ + const observer = wrapExecutionObserver({ handle: () => Effect.interrupt, }); diff --git a/packages/core/sdk/src/execution-observer.ts b/packages/core/sdk/src/execution-observer.ts index 8cf01d7317..e8a34bdbe9 100644 --- a/packages/core/sdk/src/execution-observer.ts +++ b/packages/core/sdk/src/execution-observer.ts @@ -1,4 +1,4 @@ -import { Data, Effect, Predicate, Schema } from "effect"; +import { Data, Effect, Schema } from "effect"; import * as Cause from "effect/Cause"; import type { ElicitationContext, ElicitationResponse } from "./elicitation"; @@ -103,12 +103,8 @@ export const noopExecutionObserver: ExecutionObserver = { type ExecutionEventName = ExecutionEvent["_tag"]; const executionEventName = (event: ExecutionEvent): ExecutionEventName => { - if (Predicate.isTagged(event, "ExecutionStarted")) return "ExecutionStarted"; - if (Predicate.isTagged(event, "ToolCallStarted")) return "ToolCallStarted"; - if (Predicate.isTagged(event, "ToolCallFinished")) return "ToolCallFinished"; - if (Predicate.isTagged(event, "InteractionStarted")) return "InteractionStarted"; - if (Predicate.isTagged(event, "InteractionResolved")) return "InteractionResolved"; - return "ExecutionFinished"; + // 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 = ( @@ -131,11 +127,9 @@ const handleExecutionObserverCause = ( ? Effect.interrupt : logExecutionObserverFailure(event, cause, pluginId); -/** Wrap an observer so any failure (defect or expected error) is logged, and - * an observer must never propagate into the execution it observes. */ -export const ignoreExecutionObserverErrors = ( - observer: ExecutionObserver, -): ExecutionObserver => ({ +/** Wrap an observer so non-interrupt failures are logged and isolated while + * interrupt causes still propagate as cancellation. */ +export const wrapExecutionObserver = (observer: ExecutionObserver): ExecutionObserver => ({ handle: (event) => observer .handle(event) diff --git a/packages/core/sdk/src/index.ts b/packages/core/sdk/src/index.ts index 3aba1a20e5..6c46ece370 100644 --- a/packages/core/sdk/src/index.ts +++ b/packages/core/sdk/src/index.ts @@ -247,8 +247,8 @@ export { InteractionResolved, ExecutionFinished, noopExecutionObserver, - ignoreExecutionObserverErrors, composeExecutionObservers, + wrapExecutionObserver, type ExecutionTrigger, type ToolCallStatus, type InteractionStatus, From c412d0a9d1292299f76d8aef298c58a98b34e581 Mon Sep 17 00:00:00 2001 From: Saatvik Arya Date: Wed, 24 Jun 2026 16:32:38 +0530 Subject: [PATCH 6/8] refactor(execution): scope observer dispatch --- packages/core/execution/src/engine.ts | 48 ++++++++++--------- .../core/sdk/src/execution-observer.test.ts | 30 +++++++++--- packages/core/sdk/src/execution-observer.ts | 36 ++++++++++---- packages/core/sdk/src/index.ts | 3 +- 4 files changed, 77 insertions(+), 40 deletions(-) diff --git a/packages/core/execution/src/engine.ts b/packages/core/execution/src/engine.ts index 104acf5734..9ecb6ce196 100644 --- a/packages/core/execution/src/engine.ts +++ b/packages/core/execution/src/engine.ts @@ -21,8 +21,9 @@ import { InteractionStarted, ToolCallFinished, ToolCallStarted, + emitExecutionEvent, noopExecutionObserver, - wrapExecutionObserver, + withExecutionObserver, } from "@executor-js/sdk/core"; import { CurrentOrgWriteAccess, type OrgWriteAccessState } from "@executor-js/sdk/core"; import { CodeExecutionError } from "@executor-js/codemode-core"; @@ -633,10 +634,9 @@ export const createExecutionEngine = ExecutionId.make(`exec_${crypto.randomUUID()}`); @@ -676,7 +676,7 @@ export const createExecutionEngine = Effect.gen(function* () { const toolCallId = makeToolCallId(); - yield* emit( + yield* emitExecutionEvent( new ToolCallStarted({ executionId, toolCallId, @@ -688,7 +688,7 @@ export const createExecutionEngine = - emit( + emitExecutionEvent( new ToolCallFinished({ executionId, toolCallId, @@ -701,7 +701,7 @@ export const createExecutionEngine = - emit( + emitExecutionEvent( new ToolCallFinished({ executionId, toolCallId, @@ -724,7 +724,7 @@ export const createExecutionEngine = Effect.gen(function* () { const interactionId = makeInteractionId(); - yield* emit( + yield* emitExecutionEvent( new InteractionStarted({ executionId, interactionId, @@ -735,7 +735,7 @@ export const createExecutionEngine = - emit( + emitExecutionEvent( new InteractionResolved({ executionId, interactionId, @@ -747,7 +747,7 @@ export const createExecutionEngine = - emit( + emitExecutionEvent( new InteractionResolved({ executionId, interactionId, @@ -829,7 +829,7 @@ export const createExecutionEngine = - emit( + emitExecutionEvent( new InteractionResolved({ executionId, interactionId, @@ -900,7 +900,7 @@ export const createExecutionEngine = - emit( + emitExecutionEvent( new InteractionResolved({ executionId, interactionId, @@ -921,8 +921,8 @@ export const createExecutionEngine = emit(finishFromResult(executionId, result))), - Effect.tapCause((cause) => emit(finishFromCause(executionId, cause))), + Effect.tap((result) => emitExecutionEvent(finishFromResult(executionId, result))), + Effect.tapCause((cause) => emitExecutionEvent(finishFromCause(executionId, cause))), ), ); liveSandboxFibers.add(fiber); @@ -1050,7 +1050,7 @@ export const createExecutionEngine = emit(finishFromResult(executionId, result))), - Effect.tapCause((cause) => emit(finishFromCause(executionId, cause))), + Effect.tap((result) => emitExecutionEvent(finishFromResult(executionId, result))), + Effect.tapCause((cause) => emitExecutionEvent(finishFromCause(executionId, cause))), ); yield* annotateExecuteOutcome(result); return result; @@ -1104,9 +1104,11 @@ 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 index 9483fb3abc..c23612bbf9 100644 --- a/packages/core/sdk/src/execution-observer.test.ts +++ b/packages/core/sdk/src/execution-observer.test.ts @@ -7,7 +7,8 @@ import { ExecutionId, composeExecutionObservers, definePlugin, - wrapExecutionObserver, + emitExecutionEvent, + withExecutionObserver, } from "./index"; const owner = { tenant: Tenant.make("tenant_test"), subject: Subject.make("subject_test") }; @@ -61,6 +62,19 @@ const finishedEvent = () => }); 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 = []; @@ -80,13 +94,15 @@ describe("composeExecutionObservers", () => { }), ); - it.effect("preserves interrupts from isolated observers", () => + it.effect("preserves interrupts from scoped observers", () => Effect.gen(function* () { - const observer = wrapExecutionObserver({ - handle: () => Effect.interrupt, - }); - - const exit = yield* Effect.exit(observer.handle(finishedEvent())); + const exit = yield* Effect.exit( + emitExecutionEvent(finishedEvent()).pipe( + withExecutionObserver({ + handle: () => Effect.interrupt, + }), + ), + ); expect(Exit.isFailure(exit)).toBe(true); if (!Exit.isFailure(exit)) return; diff --git a/packages/core/sdk/src/execution-observer.ts b/packages/core/sdk/src/execution-observer.ts index e8a34bdbe9..93ec676904 100644 --- a/packages/core/sdk/src/execution-observer.ts +++ b/packages/core/sdk/src/execution-observer.ts @@ -1,4 +1,4 @@ -import { Data, Effect, Schema } from "effect"; +import { Context, Data, Effect, Schema } from "effect"; import * as Cause from "effect/Cause"; import type { ElicitationContext, ElicitationResponse } from "./elicitation"; @@ -100,6 +100,11 @@ 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 => { @@ -127,14 +132,27 @@ const handleExecutionObserverCause = ( ? Effect.interrupt : logExecutionObserverFailure(event, cause, pluginId); -/** Wrap an observer so non-interrupt failures are logged and isolated while - * interrupt causes still propagate as cancellation. */ -export const wrapExecutionObserver = (observer: ExecutionObserver): ExecutionObserver => ({ - handle: (event) => - observer - .handle(event) - .pipe(Effect.catchCause((cause) => handleExecutionObserverCause(event, cause))), -}); +/** 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 diff --git a/packages/core/sdk/src/index.ts b/packages/core/sdk/src/index.ts index 6c46ece370..abaf76f551 100644 --- a/packages/core/sdk/src/index.ts +++ b/packages/core/sdk/src/index.ts @@ -248,7 +248,8 @@ export { ExecutionFinished, noopExecutionObserver, composeExecutionObservers, - wrapExecutionObserver, + emitExecutionEvent, + withExecutionObserver, type ExecutionTrigger, type ToolCallStatus, type InteractionStatus, From a3be7ac7b8b2eeb25c2599eaa636188c3f632589 Mon Sep 17 00:00:00 2001 From: Saatvik Arya Date: Thu, 27 Aug 2026 21:53:24 +0530 Subject: [PATCH 7/8] test(execution): update observer integration fixtures --- packages/core/execution/src/engine-observer.test.ts | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/packages/core/execution/src/engine-observer.test.ts b/packages/core/execution/src/engine-observer.test.ts index 03b0ad4351..45c10b137a 100644 --- a/packages/core/execution/src/engine-observer.test.ts +++ b/packages/core/execution/src/engine-observer.test.ts @@ -11,13 +11,13 @@ import { createExecutionEngine } from "./engine"; const emptyPlugin = definePlugin(() => ({ id: "observer-test" as const, storage: () => ({}), - staticSources: () => [], + staticIntegrations: () => [], })); const approvalPlugin = definePlugin(() => ({ id: "observer-approval-test" as const, storage: () => ({}), - staticSources: () => [ + staticIntegrations: () => [ { id: "approval.ctl", kind: "control" as const, From 7decefbca1014f27cf640704295d787c014fc27c Mon Sep 17 00:00:00 2001 From: Saatvik Arya Date: Mon, 7 Sep 2026 00:17:50 +0530 Subject: [PATCH 8/8] feat(sdk,execution): carry output, interruption, and envelope failures to observers The observer contract lagged the engine: `ExecuteResult.output` (everything the code sent through `emit()`) never reached `ExecutionFinished`, so an emit-only run recorded no result; expected tool failures ride the success channel as `ToolResult.fail` envelopes since #826, so every upstream 4xx was observed as a completed call; and the finish event was emitted from inside the execution fiber, so an interrupt (client abort, host backstop, sandbox shutdown) tore the run down before any observer learned it had ended. - `ExecutionFinished` gains `output` and the `interrupted` status; the new `ExecutionOutputItem` mirrors the sandbox shape structurally so the sdk stays free of the kernel package. - The engine emits the finish event from `Effect.onExit`, which runs as an uninterruptible finalizer; an interrupt-only cause maps to `interrupted`. - `ToolCallFinished` reports `failed` with `code: message` when the result is a failure envelope, keeping the envelope attached for inspection. Claude-Session: https://claude.ai/code/session_016tQXmEycQ2gmmW7LJt6Nhz --- .../execution/src/engine-observer.test.ts | 76 ++++++++++++++++++- packages/core/execution/src/engine.ts | 74 +++++++++++++----- packages/core/sdk/src/execution-observer.ts | 20 ++++- packages/core/sdk/src/index.ts | 1 + 4 files changed, 149 insertions(+), 22 deletions(-) diff --git a/packages/core/execution/src/engine-observer.test.ts b/packages/core/execution/src/engine-observer.test.ts index 45c10b137a..7c7c7dc37d 100644 --- a/packages/core/execution/src/engine-observer.test.ts +++ b/packages/core/execution/src/engine-observer.test.ts @@ -1,5 +1,5 @@ import { describe, expect, it } from "@effect/vitest"; -import { Effect, Predicate, Schema } from "effect"; +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"; @@ -130,6 +130,80 @@ describe("execution engine observer emission", () => { }), ); + 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(); diff --git a/packages/core/execution/src/engine.ts b/packages/core/execution/src/engine.ts index 9ecb6ce196..4e2351442a 100644 --- a/packages/core/execution/src/engine.ts +++ b/packages/core/execution/src/engine.ts @@ -22,6 +22,7 @@ import { ToolCallFinished, ToolCallStarted, emitExecutionEvent, + isToolResult, noopExecutionObserver, withExecutionObserver, } from "@executor-js/sdk/core"; @@ -654,20 +655,65 @@ export const createExecutionEngine = ): ExecutionFinished => new ExecutionFinished({ executionId, owner, - status: "failed", + 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, @@ -689,15 +735,7 @@ export const createExecutionEngine = emitExecutionEvent( - new ToolCallFinished({ - executionId, - toolCallId, - owner, - path: call.path, - status: "completed", - result, - completedAt: new Date(), - }), + toolCallFinishedFromResult(executionId, toolCallId, call.path, result), ), ), Effect.tapCause((cause) => @@ -919,11 +957,9 @@ export const createExecutionEngine = emitExecutionEvent(finishFromResult(executionId, result))), - Effect.tapCause((cause) => emitExecutionEvent(finishFromCause(executionId, cause))), - ), + codeExecutor + .execute(code, invoker) + .pipe(Effect.withSpan("executor.code.exec"), Effect.onExit(observeFinish(executionId))), ); liveSandboxFibers.add(fiber); @@ -1072,11 +1108,9 @@ export const createExecutionEngine = emitExecutionEvent(finishFromResult(executionId, result))), - Effect.tapCause((cause) => emitExecutionEvent(finishFromCause(executionId, cause))), - ); + const result = yield* codeExecutor + .execute(code, invoker) + .pipe(Effect.withSpan("executor.code.exec"), Effect.onExit(observeFinish(executionId))); yield* annotateExecuteOutcome(result); return result; }); diff --git a/packages/core/sdk/src/execution-observer.ts b/packages/core/sdk/src/execution-observer.ts index 93ec676904..ddb077383e 100644 --- a/packages/core/sdk/src/execution-observer.ts +++ b/packages/core/sdk/src/execution-observer.ts @@ -26,7 +26,22 @@ export type ExecutionTrigger = { export type ToolCallStatus = "completed" | "failed"; export type InteractionStatus = "accepted" | "declined" | "cancelled" | "failed"; -export type ExecutionStatus = "completed" | "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; @@ -78,7 +93,10 @@ 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; diff --git a/packages/core/sdk/src/index.ts b/packages/core/sdk/src/index.ts index abaf76f551..664447c8ac 100644 --- a/packages/core/sdk/src/index.ts +++ b/packages/core/sdk/src/index.ts @@ -254,6 +254,7 @@ export { type ToolCallStatus, type InteractionStatus, type ExecutionStatus, + type ExecutionOutputItem, type ExecutionEvent, type ExecutionObserver, } from "./execution-observer";