From c15ba819cdf2cff55446559fca26554bda6e79e2 Mon Sep 17 00:00:00 2001
From: Saatvik Arya <aryasaatvik@gmail.com>
Date: Wed, 24 Jun 2026 15:12:30 +0530
Subject: [PATCH 1/4] fix(execution): mirror observer upstream hardening

---
 .../core/sdk/src/execution-observer.test.ts   | 69 ++++++++++++++++---
 packages/core/sdk/src/execution-observer.ts   | 67 ++++++++++++++----
 2 files changed, 114 insertions(+), 22 deletions(-)

diff --git a/packages/core/sdk/src/execution-observer.test.ts b/packages/core/sdk/src/execution-observer.test.ts
index d8536f19fd..f9cb6b571c 100644
--- a/packages/core/sdk/src/execution-observer.test.ts
+++ b/packages/core/sdk/src/execution-observer.test.ts
@@ -1,21 +1,30 @@
 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") };
 
 let calls: string[] = [];
 
-const observingPlugin = (id: string) =>
+const observingPlugin = (id: string, asyncBoundary = false) =>
   definePlugin(() => ({
     id,
     storage: () => ({}),
     extension: () => ({ label: id }),
     runtime: {
       executionObserver: (self: { label: string }) => ({
-        handle: () => Effect.sync(() => calls.push(self.label)),
+        handle: () =>
+          (asyncBoundary ? Effect.promise(() => Promise.resolve()) : Effect.void).pipe(
+            Effect.flatMap(() => Effect.sync(() => calls.push(self.label))),
+          ),
       }),
     },
   }));
@@ -31,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"),
@@ -41,10 +61,10 @@ const finishedEvent = () =>
   });
 
 describe("composeExecutionObservers", () => {
-  it.effect("fans an event to every plugin observer and isolates failures", () =>
+  it.effect("dispatches observers sequentially and isolates failures", () =>
     Effect.gen(function* () {
       calls = [];
-      const first = observingPlugin("first")();
+      const first = observingPlugin("first", true)();
       const failing = failingPlugin();
       const last = observingPlugin("last")();
       const observer = composeExecutionObservers([first, failing, last] as const, {
@@ -53,19 +73,52 @@ describe("composeExecutionObservers", () => {
         last: { label: "last" },
       });
 
-      // The failing plugin dies mid-fan; the others must still observe.
+      // The failing plugin dies mid-dispatch; the others must still observe.
       yield* observer.handle(finishedEvent());
 
       expect(calls).toEqual(["first", "last"]);
     }),
   );
 
+  it.effect("preserves interrupts from 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: () => ({}) }))();
       const observer = composeExecutionObservers([plain] as const, { plain: {} });
 
-      // No observer registered → handling is a silent no-op, never throws.
+      // 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
index e0ec8898ac..c940dadd66 100644
--- a/packages/core/sdk/src/execution-observer.ts
+++ b/packages/core/sdk/src/execution-observer.ts
@@ -1,4 +1,5 @@
-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";
 import type { AnyPlugin, OwnerBinding, PluginExtensions } from "./plugin";
@@ -6,8 +7,8 @@ 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. */
+ * 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;
@@ -118,29 +119,64 @@ export const noopExecutionObserver: ExecutionObserver = {
   handle: () => Effect.void,
 };
 
-/** Wrap an observer so any failure (defect or expected error) is swallowed —
+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<unknown>,
+  pluginId?: string,
+): Effect.Effect<void> =>
+  Effect.logWarning("execution observer failed", {
+    cause: Cause.pretty(cause),
+    event: executionEventName(event),
+    ...(pluginId ? { pluginId } : {}),
+  });
+
+const handleExecutionObserverCause = (
+  event: ExecutionEvent,
+  cause: Cause.Cause<unknown>,
+  pluginId?: string,
+): Effect.Effect<void> =>
+  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 = (
   observer: ExecutionObserver<unknown>,
 ): ExecutionObserver => ({
-  handle: (event) => observer.handle(event).pipe(Effect.catchCause(() => Effect.void)),
+  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, suppressing per-observer errors. Returns the no-op observer
- *  when no plugin registers one — the common (opt-out) case. */
+ *  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 = <TPlugins extends readonly AnyPlugin[]>(
   plugins: TPlugins,
   extensions: PluginExtensions<TPlugins>,
 ): ExecutionObserver => {
-  const observers: ExecutionObserver<unknown>[] = [];
+  const observers: { readonly pluginId: string; readonly observer: ExecutionObserver<unknown> }[] =
+    [];
 
   for (const plugin of plugins) {
     const observer = plugin.runtime?.executionObserver?.(
       extensions[plugin.id as keyof PluginExtensions<TPlugins>] as never,
     );
     if (observer) {
-      observers.push(observer);
+      observers.push({ pluginId: plugin.id, observer });
     }
   }
 
@@ -152,11 +188,14 @@ export const composeExecutionObservers = <TPlugins extends readonly AnyPlugin[]>
     handle: (event) =>
       Effect.forEach(
         observers,
-        (observer) => observer.handle(event).pipe(Effect.catchCause(() => Effect.void)),
-        // 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" },
+        ({ pluginId, observer }) =>
+          observer
+            .handle(event)
+            .pipe(
+              Effect.catchCause((cause) => handleExecutionObserverCause(event, cause, pluginId)),
+            ),
+        // Preserve plugin order so observers see deterministic sequencing.
+        { discard: true },
       ),
   };
 };

From a0963006358ea2733e5f165fd7c6195fcefd5bde Mon Sep 17 00:00:00 2001
From: Saatvik Arya <aryasaatvik@gmail.com>
Date: Wed, 24 Jun 2026 16:04:40 +0530
Subject: [PATCH 2/4] 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 31e1fbc953..1dd9472a3d 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 { CodeExecutionError } from "@executor-js/codemode-core";
 import type { CodeExecutor, ExecuteResult, SandboxToolInvoker } from "@executor-js/codemode-core";
@@ -464,7 +464,7 @@ export const createExecutionEngine = <E extends Cause.YieldableError = CodeExecu
 
   // 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 observer = wrapExecutionObserver(config.observer ?? noopExecutionObserver);
   const emit = observer.handle;
   const owner = executor.owner;
 
diff --git a/packages/core/sdk/src/execution-observer.test.ts b/packages/core/sdk/src/execution-observer.test.ts
index f9cb6b571c..9483fb3abc 100644
--- a/packages/core/sdk/src/execution-observer.test.ts
+++ b/packages/core/sdk/src/execution-observer.test.ts
@@ -7,7 +7,7 @@ import {
   ExecutionId,
   composeExecutionObservers,
   definePlugin,
-  ignoreExecutionObserverErrors,
+  wrapExecutionObserver,
 } from "./index";
 
 const owner = { tenant: Tenant.make("tenant_test"), subject: Subject.make("subject_test") };
@@ -82,7 +82,7 @@ describe("composeExecutionObservers", () => {
 
   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 c940dadd66..c26b6d02ed 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";
@@ -122,12 +122,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 = (
@@ -150,11 +146,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<unknown>,
-): ExecutionObserver => ({
+/** Wrap an observer so non-interrupt failures are logged and isolated while
+ *  interrupt causes still propagate as cancellation. */
+export const wrapExecutionObserver = (observer: ExecutionObserver<unknown>): ExecutionObserver => ({
   handle: (event) =>
     observer
       .handle(event)
diff --git a/packages/core/sdk/src/index.ts b/packages/core/sdk/src/index.ts
index 7adbbfad05..b58d045988 100644
--- a/packages/core/sdk/src/index.ts
+++ b/packages/core/sdk/src/index.ts
@@ -181,8 +181,8 @@ export {
   InteractionResolved,
   ExecutionFinished,
   noopExecutionObserver,
-  ignoreExecutionObserverErrors,
   composeExecutionObservers,
+  wrapExecutionObserver,
   type ExecutionTrigger,
   type ExecutionActor,
   type ToolCallStatus,

From 256c906bb15c2846a37a6e09f68c2c3a523a41d0 Mon Sep 17 00:00:00 2001
From: Saatvik Arya <aryasaatvik@gmail.com>
Date: Wed, 24 Jun 2026 16:32:38 +0530
Subject: [PATCH 3/4] 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 1dd9472a3d..ed84c20e10 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 { CodeExecutionError } from "@executor-js/codemode-core";
 import type { CodeExecutor, ExecuteResult, SandboxToolInvoker } from "@executor-js/codemode-core";
@@ -462,10 +463,9 @@ export const createExecutionEngine = <E extends Cause.YieldableError = CodeExecu
     }
   };
 
-  // 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 = wrapExecutionObserver(config.observer ?? noopExecutionObserver);
-  const emit = observer.handle;
+  // Observer wiring is scoped around public execution methods so detached
+  // execution fibers inherit the observer context for pause/resume events.
+  const observeExecution = withExecutionObserver(config.observer ?? noopExecutionObserver);
   const owner = executor.owner;
 
   const makeExecutionId = (): ExecutionId => ExecutionId.make(`exec_${crypto.randomUUID()}`);
@@ -505,7 +505,7 @@ export const createExecutionEngine = <E extends Cause.YieldableError = CodeExecu
     invoke: (call) =>
       Effect.gen(function* () {
         const toolCallId = makeToolCallId();
-        yield* emit(
+        yield* emitExecutionEvent(
           new ToolCallStarted({
             executionId,
             toolCallId,
@@ -517,7 +517,7 @@ export const createExecutionEngine = <E extends Cause.YieldableError = CodeExecu
         );
         return yield* inner.invoke(call).pipe(
           Effect.tap((result) =>
-            emit(
+            emitExecutionEvent(
               new ToolCallFinished({
                 executionId,
                 toolCallId,
@@ -530,7 +530,7 @@ export const createExecutionEngine = <E extends Cause.YieldableError = CodeExecu
             ),
           ),
           Effect.tapCause((cause) =>
-            emit(
+            emitExecutionEvent(
               new ToolCallFinished({
                 executionId,
                 toolCallId,
@@ -553,7 +553,7 @@ export const createExecutionEngine = <E extends Cause.YieldableError = CodeExecu
     (ctx) =>
       Effect.gen(function* () {
         const interactionId = makeInteractionId();
-        yield* emit(
+        yield* emitExecutionEvent(
           new InteractionStarted({
             executionId,
             interactionId,
@@ -564,7 +564,7 @@ export const createExecutionEngine = <E extends Cause.YieldableError = CodeExecu
         );
         return yield* handler(ctx).pipe(
           Effect.tap((response) =>
-            emit(
+            emitExecutionEvent(
               new InteractionResolved({
                 executionId,
                 interactionId,
@@ -576,7 +576,7 @@ export const createExecutionEngine = <E extends Cause.YieldableError = CodeExecu
             ),
           ),
           Effect.tapCause((cause) =>
-            emit(
+            emitExecutionEvent(
               new InteractionResolved({
                 executionId,
                 interactionId,
@@ -635,7 +635,7 @@ export const createExecutionEngine = <E extends Cause.YieldableError = CodeExecu
     // resume handle below so a run with N interactions is one execution with N
     // interaction events.
     const executionId = makeExecutionId();
-    yield* emit(
+    yield* emitExecutionEvent(
       new ExecutionStarted({
         executionId,
         owner,
@@ -675,7 +675,7 @@ export const createExecutionEngine = <E extends Cause.YieldableError = CodeExecu
         };
         pausedExecutions.set(id, paused);
 
-        yield* emit(
+        yield* emitExecutionEvent(
           new InteractionStarted({
             executionId,
             interactionId,
@@ -692,7 +692,7 @@ export const createExecutionEngine = <E extends Cause.YieldableError = CodeExecu
         // pair — mirroring observeInlineElicitation on the inline path.
         return yield* Deferred.await(responseDeferred).pipe(
           Effect.tap((response) =>
-            emit(
+            emitExecutionEvent(
               new InteractionResolved({
                 executionId,
                 interactionId,
@@ -704,7 +704,7 @@ export const createExecutionEngine = <E extends Cause.YieldableError = CodeExecu
             ),
           ),
           Effect.tapCause((cause) =>
-            emit(
+            emitExecutionEvent(
               new InteractionResolved({
                 executionId,
                 interactionId,
@@ -725,8 +725,8 @@ export const createExecutionEngine = <E extends Cause.YieldableError = CodeExecu
     fiber = yield* Effect.forkDetach(
       codeExecutor.execute(code, invoker).pipe(
         Effect.withSpan("executor.code.exec"),
-        Effect.tap((result) => 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))),
       ),
     );
 
@@ -823,7 +823,7 @@ export const createExecutionEngine = <E extends Cause.YieldableError = CodeExecu
     });
 
     const executionId = makeExecutionId();
-    yield* emit(
+    yield* emitExecutionEvent(
       new ExecutionStarted({
         executionId,
         owner,
@@ -847,15 +847,17 @@ export const createExecutionEngine = <E extends Cause.YieldableError = CodeExecu
     );
     return yield* codeExecutor.execute(code, invoker).pipe(
       Effect.withSpan("executor.code.exec"),
-      Effect.tap((result) => 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))),
     );
   });
 
   return {
-    execute: runInlineExecution,
-    executeWithPause: startPausableExecution,
-    resume: resumeExecution,
+    execute: (code, options) => runInlineExecution(code, options).pipe(observeExecution),
+    executeWithPause: (code, options) =>
+      startPausableExecution(code, options).pipe(observeExecution),
+    resume: (executionId, response) =>
+      resumeExecution(executionId, response).pipe(observeExecution),
     getPausedExecution: (executionId) =>
       Effect.sync(() => pausedExecutions.get(executionId) ?? null),
     pausedExecutionCount: () => Effect.sync(() => pausedExecutions.size),
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 c26b6d02ed..407fa1f5c5 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";
@@ -119,6 +119,11 @@ export const noopExecutionObserver: ExecutionObserver = {
   handle: () => Effect.void,
 };
 
+const currentExecutionObserver = Context.Reference<ExecutionObserver>(
+  "@executor-js/sdk/ExecutionObserver",
+  { defaultValue: () => noopExecutionObserver },
+);
+
 type ExecutionEventName = ExecutionEvent["_tag"];
 
 const executionEventName = (event: ExecutionEvent): ExecutionEventName => {
@@ -146,14 +151,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<unknown>): 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<void> =>
+  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<unknown>) =>
+  <A, E, R>(effect: Effect.Effect<A, E, R>): Effect.Effect<A, E, R> =>
+    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 b58d045988..a8599ea2b5 100644
--- a/packages/core/sdk/src/index.ts
+++ b/packages/core/sdk/src/index.ts
@@ -182,7 +182,8 @@ export {
   ExecutionFinished,
   noopExecutionObserver,
   composeExecutionObservers,
-  wrapExecutionObserver,
+  emitExecutionEvent,
+  withExecutionObserver,
   type ExecutionTrigger,
   type ExecutionActor,
   type ToolCallStatus,

From 4f5a49940abd1f39a058c232a9f3bbfb5f96b4cc Mon Sep 17 00:00:00 2001
From: Saatvik Arya <aryasaatvik@gmail.com>
Date: Wed, 24 Jun 2026 17:17:21 +0530
Subject: [PATCH 4/4] refactor(plugins): match execution events exhaustively

---
 .../execution-history/src/sdk/store.ts        | 23 +++++----
 .../execution-metrics/src/cloudflare/index.ts | 47 +++++++++----------
 .../execution-metrics/src/sdk/observer.ts     | 31 ++++++------
 3 files changed, 48 insertions(+), 53 deletions(-)

diff --git a/packages/plugins/execution-history/src/sdk/store.ts b/packages/plugins/execution-history/src/sdk/store.ts
index 802a4fda38..df70089a31 100644
--- a/packages/plugins/execution-history/src/sdk/store.ts
+++ b/packages/plugins/execution-history/src/sdk/store.ts
@@ -1,4 +1,4 @@
-import { Effect, Option, Predicate, Schema } from "effect";
+import { Effect, Match, Option, Predicate, Schema } from "effect";
 
 import {
   type ExecutionEvent,
@@ -658,17 +658,16 @@ export const makeExecutionHistoryStore = (deps: StorageDeps): ExecutionHistorySt
     );
   };
 
-  const handleEvent = (event: ExecutionEvent): Effect.Effect<void, StorageFailure> => {
-    if (Predicate.isTagged(event, "ExecutionStarted")) return onExecutionStarted(event);
-    if (Predicate.isTagged(event, "ToolCallStarted")) return onToolCallStarted(event);
-    if (Predicate.isTagged(event, "ToolCallFinished")) return onToolCallFinished(event);
-    if (Predicate.isTagged(event, "InteractionStarted")) return onInteractionStarted(event);
-    if (Predicate.isTagged(event, "InteractionResolved")) return onInteractionResolved(event);
-    // Explicit guard, not a fallthrough: a future ExecutionEvent variant must
-    // not be silently recorded as a finished run.
-    if (Predicate.isTagged(event, "ExecutionFinished")) return onExecutionFinished(event);
-    return Effect.void;
-  };
+  const handleEvent = Match.type<ExecutionEvent>().pipe(
+    Match.withReturnType<Effect.Effect<void, StorageFailure>>(),
+    Match.tag("ExecutionStarted", onExecutionStarted),
+    Match.tag("ToolCallStarted", onToolCallStarted),
+    Match.tag("ToolCallFinished", onToolCallFinished),
+    Match.tag("InteractionStarted", onInteractionStarted),
+    Match.tag("InteractionResolved", onInteractionResolved),
+    Match.tag("ExecutionFinished", onExecutionFinished),
+    Match.exhaustive,
+  );
 
   const computeMeta = (
     options: ExecutionHistoryListOptions,
diff --git a/packages/plugins/execution-metrics/src/cloudflare/index.ts b/packages/plugins/execution-metrics/src/cloudflare/index.ts
index 399cdb80bf..686f4bf54d 100644
--- a/packages/plugins/execution-metrics/src/cloudflare/index.ts
+++ b/packages/plugins/execution-metrics/src/cloudflare/index.ts
@@ -1,4 +1,4 @@
-import { Effect, Predicate } from "effect";
+import { Effect, Match } from "effect";
 
 import { type ExecutionEvent, type ExecutionObserver } from "@executor-js/sdk/core";
 
@@ -40,24 +40,23 @@ export const createWaeMetricsObserver = (analytics: AnalyticsEngineDataset): Exe
   const toolCallStarts = new Map<string, number>();
 
   return {
-    handle: (event: ExecutionEvent) => {
-      if (Predicate.isTagged(event, "ExecutionStarted")) {
-        return Effect.sync(() => {
+    handle: Match.type<ExecutionEvent>().pipe(
+      Match.withReturnType<Effect.Effect<void>>(),
+      Match.tag("ExecutionStarted", (event) =>
+        Effect.sync(() => {
           rememberStart(executionStarts, event.executionId, {
             startedAt: event.startedAt.getTime(),
             trigger: event.trigger?.kind,
           });
-        });
-      }
-
-      if (Predicate.isTagged(event, "ToolCallStarted")) {
-        return Effect.sync(() => {
+        }),
+      ),
+      Match.tag("ToolCallStarted", (event) =>
+        Effect.sync(() => {
           rememberStart(toolCallStarts, event.toolCallId, event.startedAt.getTime());
-        });
-      }
-
-      if (Predicate.isTagged(event, "ExecutionFinished")) {
-        return Effect.sync(() => {
+        }),
+      ),
+      Match.tag("ExecutionFinished", (event) =>
+        Effect.sync(() => {
           const started = executionStarts.get(event.executionId);
           executionStarts.delete(event.executionId);
           const durationMs = started
@@ -68,11 +67,10 @@ export const createWaeMetricsObserver = (analytics: AnalyticsEngineDataset): Exe
             doubles: [durationMs],
             indexes: [event.executionId],
           });
-        });
-      }
-
-      if (Predicate.isTagged(event, "ToolCallFinished")) {
-        return Effect.sync(() => {
+        }),
+      ),
+      Match.tag("ToolCallFinished", (event) =>
+        Effect.sync(() => {
           const startedAt = toolCallStarts.get(event.toolCallId);
           toolCallStarts.delete(event.toolCallId);
           const durationMs =
@@ -82,10 +80,11 @@ export const createWaeMetricsObserver = (analytics: AnalyticsEngineDataset): Exe
             doubles: [durationMs],
             indexes: [event.executionId],
           });
-        });
-      }
-
-      return Effect.void;
-    },
+        }),
+      ),
+      Match.tag("InteractionStarted", () => Effect.void),
+      Match.tag("InteractionResolved", () => Effect.void),
+      Match.exhaustive,
+    ),
   };
 };
diff --git a/packages/plugins/execution-metrics/src/sdk/observer.ts b/packages/plugins/execution-metrics/src/sdk/observer.ts
index c95dddc1a3..0f15bebc75 100644
--- a/packages/plugins/execution-metrics/src/sdk/observer.ts
+++ b/packages/plugins/execution-metrics/src/sdk/observer.ts
@@ -1,4 +1,4 @@
-import { Effect, Metric, Predicate } from "effect";
+import { Effect, Match, Metric } from "effect";
 
 import {
   type ExecutionEvent,
@@ -103,22 +103,19 @@ export const createExecutionMetricsObserver = (): ExecutionObserver => {
     });
 
   return {
-    handle: (event) => {
-      if (Predicate.isTagged(event, "ExecutionStarted")) {
-        return handleExecutionStarted(event);
-      }
-      if (Predicate.isTagged(event, "ExecutionFinished")) {
-        return handleExecutionFinished(event);
-      }
-      if (Predicate.isTagged(event, "ToolCallStarted")) {
-        return updateCounter(toolCallsStarted);
-      }
-      if (Predicate.isTagged(event, "ToolCallFinished")) {
-        return event.status === "completed"
+    handle: Match.type<ExecutionEvent>().pipe(
+      Match.withReturnType<Effect.Effect<void>>(),
+      Match.tag("ExecutionStarted", handleExecutionStarted),
+      Match.tag("ExecutionFinished", handleExecutionFinished),
+      Match.tag("ToolCallStarted", () => updateCounter(toolCallsStarted)),
+      Match.tag("ToolCallFinished", (event) =>
+        event.status === "completed"
           ? updateCounter(toolCallsCompleted)
-          : updateCounter(toolCallsFailed);
-      }
-      return Effect.void;
-    },
+          : updateCounter(toolCallsFailed),
+      ),
+      Match.tag("InteractionStarted", () => Effect.void),
+      Match.tag("InteractionResolved", () => Effect.void),
+      Match.exhaustive,
+    ),
   };
 };