feat(execution): add execution observer foundation

[aryasaatvik]

Summary

Add an Effect-native execution lifecycle observer that plugins can use for durable history, metrics, indexing, or cache maintenance without coupling those products to the execution engine. The hook complements existing spans and OpenTelemetry rather than replacing them.

Contract

type ExecutionEvent =
  | ExecutionStarted
  | ToolCallStarted
  | ToolCallFinished
  | InteractionStarted
  | InteractionResolved
  | ExecutionFinished;

definePlugin(() => ({
  runtime: {
    executionObserver: (extension) => ({ handle: (event) => Effect.void }),
  },
}));

execute / executeWithPause / resume
  -> install the composed observer in Effect context
  -> emit execution, tool-call, and interaction events
  -> dispatch sequentially to plugin observers
  -> isolate and log non-interrupt observer failures

• Observer context survives inline and detached pause/resume fibers.
• Interrupts remain cancellations; ordinary observer failures cannot fail execution.
• Shared API and local execution stacks use the same plugin hook.
• Owner metadata preserves execution attribution.

Validation

• Focused SDK observer tests: 5 passed.
• Engine lifecycle tests: 3 passed across inline and pause/resume execution.
• SDK, execution, API, and local package suites and typechecks passed.
• Format, lint, and all 36 applicable GitHub CI checks passed at 9f9373469.

Execution history delivery map

Prerequisites:

• feat(execution): add execution observer foundation #1097 — Execution observer foundation 👈 current
• feat(fumadb): add aggregate and keyset query support #1119 — Aggregate and keyset queries

Follow-up PR-sized diffs:

• Core persistence and read API
  • Runs console UI
    • Cloudflare service-token actors

The fork also uses this observer for execution metrics, which remains a separate consumer. View the complete fork comparison.

[greptile-apps]

Greptile Summary

This PR introduces the execution observer foundation: a typed lifecycle event stream (ExecutionStarted/Finished, ToolCallStarted/Finished, InteractionStarted/Resolved) emitted by the execution engine, a plugin hook (runtime.executionObserver) for opt-in subscription, and composition helpers that fan events to all registered plugin observers while isolating non-interrupt failures and propagating interrupt causes correctly.

• Observer contract (execution-observer.ts): Data.TaggedClass events with stable _tag discriminants, a Context.Reference-backed dispatch channel with interrupt-aware error isolation via handleExecutionObserverCause, and composeExecutionObservers for sequential per-plugin fan-out.
• Engine wiring (engine.ts): Both runInlineExecution and startPausableExecution now bracket their full lifecycle with ExecutionStarted/Finished events; tool calls and elicitations are wrapped with matching Started/Finished/Resolved pairs. The daemon fiber inherits the observer context from the forkDetach scope so post-pause emissions flow to the same observer.
• Integration points (execution-stack.ts, app.ts, main.ts): makeExecutionStack composes plugin observers before passing them to the engine; both local bypass paths receive the same treatment so no execution surface is unobserved.

Confidence Score: 5/5

The change is additive and opt-out by default; executions with no registered observer pay only a no-op context lookup, and the error isolation design prevents any observer from breaking a live execution.

The interrupt-propagation and failure-isolation logic is correct and well-tested. The daemon fiber correctly inherits the observer context through forkDetach. Both execution paths emit a complete and symmetric event sequence. The two findings are narrow edge cases under concurrent interruption that do not affect the common path.

No files require special attention for merge safety.

Important Files Changed

Filename	Overview
packages/core/sdk/src/execution-observer.ts	New file defining the full observer contract — tagged event classes, Context.Reference-backed dispatch, interrupt-aware error isolation, and plugin fan-out composition. Clean and well-structured.
packages/core/execution/src/engine.ts	Wires ExecutionStarted/Finished, ToolCallStarted/Finished, and InteractionStarted/Resolved into both execution paths; daemon fiber inherits observer context correctly via forkDetach within the withExecutionObserver scope.
packages/core/sdk/src/execution-observer.test.ts	Covers the core dispatch, failure isolation, interrupt propagation, and no-observer cases well; uses module-level mutable calls array that each test resets manually — works today but fragile under concurrent test runners.
packages/core/execution/src/engine-observer.test.ts	Covers the full lifecycle on both the pausable and inline elicitation paths; validates event ordering, ID threading, and no-op behavior when no observer is registered.
packages/core/api/src/server/execution-stack.ts	Adds composeExecutionObservers to makeExecutionStack with the correct phantom-type recovery cast; change is minimal and correctly threaded.
packages/core/sdk/src/executor.ts	Exposes executor.owner (ownerBinding) on the Executor type so engine machinery can attribute events without re-threading identity.
packages/core/sdk/src/plugin.ts	Adds optional runtime.executionObserver hook to PluginSpec; clean additive change with no breaking surface.

Sequence Diagram

%%{init: {'theme': 'neutral'}}%%
sequenceDiagram
    participant Caller
    participant Engine
    participant DaemonFiber
    participant Observer

    Caller->>Engine: executeWithPause(code, options)
    Engine->>Observer: ExecutionStarted
    Engine->>DaemonFiber: forkDetach (inherits Observer context)
    DaemonFiber->>Observer: ToolCallStarted
    DaemonFiber->>Observer: ToolCallFinished
    DaemonFiber->>Observer: InteractionStarted
    DaemonFiber-->>Engine: paused (Deferred)
    Engine-->>Caller: PausedExecution

    Caller->>Engine: resume(executionId, response)
    Engine->>DaemonFiber: Deferred.succeed(response)
    DaemonFiber->>Observer: InteractionResolved
    DaemonFiber->>Observer: ExecutionFinished
    Engine-->>Caller: ExecutionResult

Loading

%%{init: {'theme': 'base', 'themeVariables': {"darkMode": true, "background": "#0d1117", "primaryColor": "#21262d", "primaryTextColor": "#e6edf3", "primaryBorderColor": "#8b949e", "lineColor": "#8b949e", "textColor": "#e6edf3", "edgeLabelBackground": "#161b22", "actorBkg": "#21262d", "actorBorder": "#8b949e", "actorTextColor": "#e6edf3", "actorLineColor": "#8b949e", "signalColor": "#8b949e", "signalTextColor": "#e6edf3", "noteBkgColor": "#373320", "noteBorderColor": "#d4a72c", "noteTextColor": "#f0e6c0", "labelBoxBkgColor": "#21262d", "labelBoxBorderColor": "#8b949e", "labelTextColor": "#e6edf3", "loopTextColor": "#e6edf3", "activationBkgColor": "#30363d", "activationBorderColor": "#8b949e"}}}%%
sequenceDiagram
    participant Caller
    participant Engine
    participant DaemonFiber
    participant Observer

    Caller->>Engine: executeWithPause(code, options)
    Engine->>Observer: ExecutionStarted
    Engine->>DaemonFiber: forkDetach (inherits Observer context)
    DaemonFiber->>Observer: ToolCallStarted
    DaemonFiber->>Observer: ToolCallFinished
    DaemonFiber->>Observer: InteractionStarted
    DaemonFiber-->>Engine: paused (Deferred)
    Engine-->>Caller: PausedExecution

    Caller->>Engine: resume(executionId, response)
    Engine->>DaemonFiber: Deferred.succeed(response)
    DaemonFiber->>Observer: InteractionResolved
    DaemonFiber->>Observer: ExecutionFinished
    Engine-->>Caller: ExecutionResult

Loading

_{Reviews (5): Last reviewed commit: "refactor(execution): scope observer disp..." | Re-trigger Greptile}