feat(execution): add execution observer foundation

.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.

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<FixedExecutionProvider> =>
+const localFixedExecutionLayer = (
+  executor: LocalExecutor,
+  plugins: readonly AnyPlugin[],
+): Layer.Layer<FixedExecutionProvider> =>
   Layer.succeed(FixedExecutionProvider)({
     executor,
     // This engine serves the HTTP executions API (`executor call`/`resume`,
@@ -62,6 +66,8 @@ const localFixedExecutionLayer = (executor: LocalExecutor): Layer.Layer<FixedExe
       createExecutionEngine({
         executor,
         codeExecutor: makeQuickJsExecutor(),
+        // Local bypasses makeExecutionStack, so compose plugin observers here.
+        observer: composeExecutionObservers(plugins, executor),
       }),
       localAnalytics,
       { plane: "api", toolkit: false },
@@ -95,7 +101,7 @@ export const makeLocalApiHandler = async (token: string): Promise<LocalApiHandle
   // 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);
 
   // The authoritative identity gate for the typed `/api`: validates the boot
   // bearer token and resolves the one local Principal. The Bun shell

apps/local/src/main.ts

@@ -6,6 +6,7 @@ import { artifactUrlFor } from "@executor-js/host-mcp/create-artifact";
 import { loadMcpAppsShellHtml } from "@executor-js/mcp-apps-shell";
 import { smokeRenderArtifact } from "@executor-js/mcp-apps-shell/smoke-render";
 import { makeQuickJsExecutor } from "@executor-js/runtime-quickjs";
+import { composeExecutionObservers } from "@executor-js/sdk";
 import { localAnalytics } from "./analytics";
 import { makeLocalApiHandler } from "./app";
 import { createExecutorHandle, disposeExecutor, getExecutorBundle } from "./executor";
@@ -86,14 +87,15 @@ export const createServerHandlers = async (token: string): Promise<ServerHandler
     // 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, webBaseUrl } = await getExecutorBundle();
+    const { executor, plugins, webBaseUrl } = await getExecutorBundle();
     // Both engines below serve MCP endpoints, so the wrap binds the "mcp"
     // plane structurally; the toolkit-scoped engine additionally marks
     // `toolkit` (the slug itself is a user label and never recorded).
     const engine = withExecutionAnalytics(
       createExecutionEngine({
         executor,
         codeExecutor: makeQuickJsExecutor(),
+        observer: composeExecutionObservers(plugins, executor),
       }),
       localAnalytics,
       { plane: "mcp", toolkit: false },
@@ -148,6 +150,7 @@ export const createServerHandlers = async (token: string): Promise<ServerHandler
           createExecutionEngine({
             executor: handle.executor,
             codeExecutor: makeQuickJsExecutor(),
+            observer: composeExecutionObservers(handle.plugins, handle.executor),
           }),
           localAnalytics,
           { plane: "mcp", toolkit: true },

packages/core/api/src/server/execution-stack.ts

@@ -27,6 +27,7 @@ import { Context, Effect, Layer } from "effect";
 import type * as Cause from "effect/Cause";
 
 import type { McpResource } from "@executor-js/host-mcp";
+import { composeExecutionObservers } from "@executor-js/sdk";
 import type { AnyPlugin, Executor, ExecutorConfig, StorageFailure } from "@executor-js/sdk";
 import {
   createExecutionEngine,
@@ -139,9 +140,11 @@ export const makeExecutionStack = <
     const { decorate } = yield* EngineDecorator.asEffect().pipe(
       Effect.withSpan("executor.stack.decorator"),
     );
+    const { plugins } = yield* PluginsProvider;
+    const observer = composeExecutionObservers(plugins() as TPlugins, executor);
     const engine = yield* Effect.sync(() =>
       decorate(
-        createExecutionEngine({ executor, codeExecutor }),
+        createExecutionEngine({ executor, codeExecutor, observer }),
         {
           accountId,
           organizationId,

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<void>();
+      // 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");
+    }),
+  );
+});