diff --git a/packages/core/execution/src/engine.ts b/packages/core/execution/src/engine.ts index 31e1fbc953..ed84c20e10 100644 --- a/packages/core/execution/src/engine.ts +++ b/packages/core/execution/src/engine.ts @@ -21,8 +21,9 @@ import { InteractionStarted, ToolCallFinished, ToolCallStarted, - ignoreExecutionObserverErrors, + emitExecutionEvent, noopExecutionObserver, + 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 = ExecutionId.make(`exec_${crypto.randomUUID()}`); @@ -505,7 +505,7 @@ export const createExecutionEngine = Effect.gen(function* () { const toolCallId = makeToolCallId(); - yield* emit( + yield* emitExecutionEvent( new ToolCallStarted({ executionId, toolCallId, @@ -517,7 +517,7 @@ export const createExecutionEngine = - emit( + emitExecutionEvent( new ToolCallFinished({ executionId, toolCallId, @@ -530,7 +530,7 @@ export const createExecutionEngine = - emit( + emitExecutionEvent( new ToolCallFinished({ executionId, toolCallId, @@ -553,7 +553,7 @@ export const createExecutionEngine = Effect.gen(function* () { const interactionId = makeInteractionId(); - yield* emit( + yield* emitExecutionEvent( new InteractionStarted({ executionId, interactionId, @@ -564,7 +564,7 @@ export const createExecutionEngine = - emit( + emitExecutionEvent( new InteractionResolved({ executionId, interactionId, @@ -576,7 +576,7 @@ export const createExecutionEngine = - emit( + emitExecutionEvent( new InteractionResolved({ executionId, interactionId, @@ -635,7 +635,7 @@ export const createExecutionEngine = - emit( + emitExecutionEvent( new InteractionResolved({ executionId, interactionId, @@ -704,7 +704,7 @@ export const createExecutionEngine = - emit( + emitExecutionEvent( new InteractionResolved({ executionId, interactionId, @@ -725,8 +725,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))), ), ); @@ -823,7 +823,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))), ); }); 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 d8536f19fd..c23612bbf9 100644 --- a/packages/core/sdk/src/execution-observer.test.ts +++ b/packages/core/sdk/src/execution-observer.test.ts @@ -1,21 +1,31 @@ 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, + emitExecutionEvent, + withExecutionObserver, +} 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 +41,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 +62,23 @@ const finishedEvent = () => }); describe("composeExecutionObservers", () => { - it.effect("fans an event to every plugin observer and isolates failures", () => + it.effect("emits events to the scoped observer", () => Effect.gen(function* () { calls = []; - const first = observingPlugin("first")(); + 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 = []; + const first = observingPlugin("first", true)(); const failing = failingPlugin(); const last = observingPlugin("last")(); const observer = composeExecutionObservers([first, failing, last] as const, { @@ -53,19 +87,54 @@ 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 scoped observers", () => + Effect.gen(function* () { + const exit = yield* Effect.exit( + emitExecutionEvent(finishedEvent()).pipe( + withExecutionObserver({ + handle: () => Effect.interrupt, + }), + ), + ); + + 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..407fa1f5c5 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 { Context, Data, Effect, 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,76 @@ export const noopExecutionObserver: ExecutionObserver = { handle: () => Effect.void, }; -/** Wrap an observer so any failure (defect or expected error) is swallowed — - * an observer must never propagate into the execution it observes. */ -export const ignoreExecutionObserverErrors = ( - observer: ExecutionObserver, -): ExecutionObserver => ({ - handle: (event) => observer.handle(event).pipe(Effect.catchCause(() => Effect.void)), -}); +const currentExecutionObserver = Context.Reference( + "@executor-js/sdk/ExecutionObserver", + { defaultValue: () => noopExecutionObserver }, +); + +type ExecutionEventName = ExecutionEvent["_tag"]; + +const executionEventName = (event: ExecutionEvent): ExecutionEventName => { + // 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 = ( + event: ExecutionEvent, + cause: Cause.Cause, + pluginId?: string, +): Effect.Effect => + Effect.logWarning("execution observer failed", { + cause: Cause.pretty(cause), + event: executionEventName(event), + ...(pluginId ? { pluginId } : {}), + }); + +const handleExecutionObserverCause = ( + event: ExecutionEvent, + cause: Cause.Cause, + pluginId?: string, +): Effect.Effect => + Cause.hasInterrupts(cause) + ? Effect.interrupt + : logExecutionObserverFailure(event, cause, pluginId); + +/** 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, 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 = ( plugins: TPlugins, extensions: PluginExtensions, ): ExecutionObserver => { - const observers: 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(observer); + observers.push({ pluginId: plugin.id, observer }); } } @@ -152,11 +200,14 @@ export const composeExecutionObservers = 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 }, ), }; }; diff --git a/packages/core/sdk/src/index.ts b/packages/core/sdk/src/index.ts index 7adbbfad05..a8599ea2b5 100644 --- a/packages/core/sdk/src/index.ts +++ b/packages/core/sdk/src/index.ts @@ -181,8 +181,9 @@ export { InteractionResolved, ExecutionFinished, noopExecutionObserver, - ignoreExecutionObserverErrors, composeExecutionObservers, + emitExecutionEvent, + withExecutionObserver, type ExecutionTrigger, type ExecutionActor, type ToolCallStatus, 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 => { - 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().pipe( + Match.withReturnType>(), + 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(); return { - handle: (event: ExecutionEvent) => { - if (Predicate.isTagged(event, "ExecutionStarted")) { - return Effect.sync(() => { + handle: Match.type().pipe( + Match.withReturnType>(), + 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().pipe( + Match.withReturnType>(), + 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, + ), }; };