diff --git a/.changeset/execution-metrics-plugin.md b/.changeset/execution-metrics-plugin.md new file mode 100644 index 0000000000..9728adf762 --- /dev/null +++ b/.changeset/execution-metrics-plugin.md @@ -0,0 +1,13 @@ +--- +"@executor-js/plugin-execution-metrics": minor +--- + +Add the `@executor-js/plugin-execution-metrics` plugin — an opt-in metrics sink +built on the ExecutionObserver foundation. The base plugin drives Effect Metric +counters and an execution-duration histogram from the event stream; `./api`'s +`executionMetricsHttpPlugin` exposes them as an auth-gated `GET +/execution-metrics/metrics` Prometheus scrape for long-lived hosts; and +`./cloudflare`'s `createWaeMetricsObserver` writes Workers Analytics Engine data +points on the Cloudflare host (per-isolate Effect metrics being meaningless on a +Worker fleet). The base `.` entry never loads `@executor-js/api` or +`@cloudflare/workers-types`. diff --git a/apps/host-cloudflare/package.json b/apps/host-cloudflare/package.json index eacf0335a1..a4b1bf9fc7 100644 --- a/apps/host-cloudflare/package.json +++ b/apps/host-cloudflare/package.json @@ -24,6 +24,7 @@ "@executor-js/fumadb": "workspace:*", "@executor-js/host-mcp": "workspace:*", "@executor-js/plugin-encrypted-secrets": "workspace:*", + "@executor-js/plugin-execution-metrics": "workspace:*", "@executor-js/plugin-graphql": "workspace:*", "@executor-js/plugin-mcp": "workspace:*", "@executor-js/plugin-openapi": "workspace:*", diff --git a/apps/host-cloudflare/src/app.ts b/apps/host-cloudflare/src/app.ts index b664e8a32d..7a5e50096f 100644 --- a/apps/host-cloudflare/src/app.ts +++ b/apps/host-cloudflare/src/app.ts @@ -33,7 +33,7 @@ import { preloadQuickJs } from "./quickjs"; export const makeCloudflareApp = async (env: CloudflareEnv) => { const config = loadConfig(env); - const plugins = makeCloudflarePlugins(config.secretKey); + const plugins = makeCloudflarePlugins(config.secretKey, env.ANALYTICS); // Load the Workers-compatible (WASM-inlined) QuickJS variant before any // executor is built — the default variant can't fetch its .wasm on Workers. diff --git a/apps/host-cloudflare/src/config.ts b/apps/host-cloudflare/src/config.ts index c372513792..f01968cd84 100644 --- a/apps/host-cloudflare/src/config.ts +++ b/apps/host-cloudflare/src/config.ts @@ -1,4 +1,5 @@ import type { D1Database, DurableObjectNamespace, R2Bucket } from "@cloudflare/workers-types"; +import type { AnalyticsEngineDataset } from "@executor-js/plugin-execution-metrics/cloudflare"; import { isValidOrgSlug } from "@executor-js/api"; @@ -18,6 +19,10 @@ export interface CloudflareEnv { readonly DB: D1Database; /** R2 bucket binding — holds values too large for a D1 row (~1-2MB cap). */ readonly BLOBS?: R2Bucket; + /** Workers Analytics Engine binding — opt-in sink for execution metrics. When + * bound (uncomment `analytics_engine_datasets` in wrangler.jsonc), each + * finished execution/tool call writes a data point; absent, metrics are off. */ + readonly ANALYTICS?: AnalyticsEngineDataset; /** MCP session Durable Object namespace — one addressable isolate per MCP * session (the DO id IS the session id), so a session survives across the * Worker's stateless isolates. */ diff --git a/apps/host-cloudflare/src/plugins.ts b/apps/host-cloudflare/src/plugins.ts index 941c807950..279f4ec728 100644 --- a/apps/host-cloudflare/src/plugins.ts +++ b/apps/host-cloudflare/src/plugins.ts @@ -2,6 +2,12 @@ import { openApiHttpPlugin } from "@executor-js/plugin-openapi/api"; import { mcpHttpPlugin } from "@executor-js/plugin-mcp/api"; import { graphqlHttpPlugin } from "@executor-js/plugin-graphql/api"; import { encryptedSecretsPlugin } from "@executor-js/plugin-encrypted-secrets"; +import { executionMetricsPlugin } from "@executor-js/plugin-execution-metrics"; +import { + createWaeMetricsObserver, + type AnalyticsEngineDataset, +} from "@executor-js/plugin-execution-metrics/cloudflare"; +import { noopExecutionObserver } from "@executor-js/sdk"; // --------------------------------------------------------------------------- // The Cloudflare host's plugin list — the same protocol/provider plugins as @@ -12,14 +18,25 @@ import { encryptedSecretsPlugin } from "@executor-js/plugin-encrypted-secrets"; // // `dangerouslyAllowStdioMCP` is false: a multi-user instance must not let a user // spawn arbitrary stdio MCP processes. +// +// Execution metrics ship to Workers Analytics Engine — opt-in via the wrangler +// `ANALYTICS` binding. The plugin is always in the tuple (it has no tables/API, +// so the shape is stable), but its observer is a no-op until `env.ANALYTICS` is +// bound. Effect's Metric registry is per-isolate (meaningless on a Worker +// fleet), so the local Prometheus scrape is deliberately NOT mounted here; WAE +// is the durable sink. To enable: uncomment `analytics_engine_datasets` in +// wrangler.jsonc. // --------------------------------------------------------------------------- -export const makeCloudflarePlugins = (secretKey: string) => +export const makeCloudflarePlugins = (secretKey: string, analytics?: AnalyticsEngineDataset) => [ openApiHttpPlugin(), mcpHttpPlugin({ dangerouslyAllowStdioMCP: false }), graphqlHttpPlugin(), encryptedSecretsPlugin({ key: secretKey }), + executionMetricsPlugin({ + observer: () => (analytics ? createWaeMetricsObserver(analytics) : noopExecutionObserver), + }), ] as const; export type CloudflarePlugins = ReturnType; diff --git a/apps/host-cloudflare/wrangler.jsonc b/apps/host-cloudflare/wrangler.jsonc index aec20421ff..c7eee826a1 100644 --- a/apps/host-cloudflare/wrangler.jsonc +++ b/apps/host-cloudflare/wrangler.jsonc @@ -33,6 +33,13 @@ "bucket_name": "executor-blobs", }, ], + // Workers Analytics Engine — the opt-in sink for execution metrics. Uncomment + // to enable: each finished execution/tool call writes a data point (queryable + // via Cloudflare's GraphQL Analytics API). The plugin is inert until this + // binding is present. Requires a paid Workers plan. + // "analytics_engine_datasets": [ + // { "binding": "ANALYTICS", "dataset": "executor_executions" }, + // ], // The MCP session Durable Object: one addressable isolate per MCP session (the // DO id IS the session id) so a session survives across the Worker's stateless // isolates — without it, `tools/list` after `initialize` can land on a fresh diff --git a/bun.lock b/bun.lock index b0e11e0592..a50dbc31d5 100644 --- a/bun.lock +++ b/bun.lock @@ -167,6 +167,7 @@ "@executor-js/fumadb": "workspace:*", "@executor-js/host-mcp": "workspace:*", "@executor-js/plugin-encrypted-secrets": "workspace:*", + "@executor-js/plugin-execution-metrics": "workspace:*", "@executor-js/plugin-graphql": "workspace:*", "@executor-js/plugin-mcp": "workspace:*", "@executor-js/plugin-openapi": "workspace:*", @@ -839,6 +840,29 @@ "react", ], }, + "packages/plugins/execution-metrics": { + "name": "@executor-js/plugin-execution-metrics", + "version": "0.0.0", + "dependencies": { + "@executor-js/sdk": "workspace:*", + "effect": "catalog:", + }, + "devDependencies": { + "@cloudflare/workers-types": "^4.20250620.0", + "@effect/vitest": "catalog:", + "@executor-js/api": "workspace:*", + "@types/node": "catalog:", + "bun-types": "catalog:", + "tsup": "catalog:", + "vitest": "catalog:", + }, + "peerDependencies": { + "@executor-js/api": "workspace:*", + }, + "optionalPeers": [ + "@executor-js/api", + ], + }, "packages/plugins/file-secrets": { "name": "@executor-js/plugin-file-secrets", "version": "1.5.10", @@ -1608,6 +1632,8 @@ "@executor-js/plugin-execution-history": ["@executor-js/plugin-execution-history@workspace:packages/plugins/execution-history"], + "@executor-js/plugin-execution-metrics": ["@executor-js/plugin-execution-metrics@workspace:packages/plugins/execution-metrics"], + "@executor-js/plugin-file-secrets": ["@executor-js/plugin-file-secrets@workspace:packages/plugins/file-secrets"], "@executor-js/plugin-graphql": ["@executor-js/plugin-graphql@workspace:packages/plugins/graphql"], diff --git a/packages/plugins/execution-metrics/CHANGELOG.md b/packages/plugins/execution-metrics/CHANGELOG.md new file mode 100644 index 0000000000..3d1d7847f1 --- /dev/null +++ b/packages/plugins/execution-metrics/CHANGELOG.md @@ -0,0 +1 @@ +# @executor-js/plugin-execution-metrics diff --git a/packages/plugins/execution-metrics/package.json b/packages/plugins/execution-metrics/package.json new file mode 100644 index 0000000000..a43bb47362 --- /dev/null +++ b/packages/plugins/execution-metrics/package.json @@ -0,0 +1,38 @@ +{ + "name": "@executor-js/plugin-execution-metrics", + "version": "0.0.0", + "private": true, + "type": "module", + "exports": { + ".": "./src/sdk/index.ts", + "./api": "./src/api/index.ts", + "./cloudflare": "./src/cloudflare/index.ts" + }, + "scripts": { + "build": "tsup && (tsc --declaration --emitDeclarationOnly --outDir dist --rootDir src || true)", + "typecheck": "tsgo --noEmit", + "test": "vitest run", + "test:watch": "vitest" + }, + "dependencies": { + "@executor-js/sdk": "workspace:*", + "effect": "catalog:" + }, + "devDependencies": { + "@cloudflare/workers-types": "^4.20250620.0", + "@effect/vitest": "catalog:", + "@executor-js/api": "workspace:*", + "@types/node": "catalog:", + "bun-types": "catalog:", + "tsup": "catalog:", + "vitest": "catalog:" + }, + "peerDependencies": { + "@executor-js/api": "workspace:*" + }, + "peerDependenciesMeta": { + "@executor-js/api": { + "optional": true + } + } +} diff --git a/packages/plugins/execution-metrics/src/api/group.ts b/packages/plugins/execution-metrics/src/api/group.ts new file mode 100644 index 0000000000..da2a259e4a --- /dev/null +++ b/packages/plugins/execution-metrics/src/api/group.ts @@ -0,0 +1,28 @@ +import { HttpApiEndpoint, HttpApiGroup, HttpApiSchema } from "effect/unstable/httpapi"; +import { Schema } from "effect"; +import { InternalError } from "@executor-js/sdk/shared"; + +// --------------------------------------------------------------------------- +// Responses +// +// The scrape endpoint returns Prometheus text-exposition, not JSON. +// `HttpApiSchema.asText()` flips the success body to `text/plain` so the +// renderer's string is written verbatim. +// --------------------------------------------------------------------------- + +const PrometheusResponse = Schema.String.pipe(HttpApiSchema.asText()); + +// --------------------------------------------------------------------------- +// Group — the execution-metrics HTTP surface. +// +// One GET endpoint. Auth-gated automatically by the host's execution-stack +// middleware, like every other plugin route. `InternalError` is the shared +// opaque 500 surface translated at the HTTP edge by `capture`. +// --------------------------------------------------------------------------- + +export const ExecutionMetricsGroup = HttpApiGroup.make("execution-metrics").add( + HttpApiEndpoint.get("metrics", "/execution-metrics/metrics", { + success: PrometheusResponse, + error: InternalError, + }), +); diff --git a/packages/plugins/execution-metrics/src/api/handlers.ts b/packages/plugins/execution-metrics/src/api/handlers.ts new file mode 100644 index 0000000000..60b132f8e3 --- /dev/null +++ b/packages/plugins/execution-metrics/src/api/handlers.ts @@ -0,0 +1,27 @@ +import { HttpApiBuilder } from "effect/unstable/httpapi"; + +import { addGroup } from "@executor-js/api"; +import { ExecutionMetricsGroup } from "./group"; +import { renderPrometheus } from "./prometheus"; + +// --------------------------------------------------------------------------- +// Composed API — core + execution-metrics group +// --------------------------------------------------------------------------- + +const ExecutorApiWithMetrics = addGroup(ExecutionMetricsGroup); + +// --------------------------------------------------------------------------- +// Handlers +// +// The scrape reads the global Effect Metric registry directly (no plugin +// extension state), so there is no `extensionService` to require — the handler +// just renders the snapshot. `renderPrometheus` has no error channel, so no +// `capture` translation is needed; the group's `InternalError` covers defects +// the edge middleware captures. +// --------------------------------------------------------------------------- + +export const ExecutionMetricsHandlers = HttpApiBuilder.group( + ExecutorApiWithMetrics, + "execution-metrics", + (handlers) => handlers.handle("metrics", () => renderPrometheus), +); diff --git a/packages/plugins/execution-metrics/src/api/index.ts b/packages/plugins/execution-metrics/src/api/index.ts new file mode 100644 index 0000000000..298a0d6c30 --- /dev/null +++ b/packages/plugins/execution-metrics/src/api/index.ts @@ -0,0 +1,24 @@ +import { definePlugin } from "@executor-js/sdk/core"; + +import { executionMetricsPlugin, type ExecutionMetricsPluginOptions } from "../sdk/plugin"; +import { ExecutionMetricsGroup } from "./group"; +import { ExecutionMetricsHandlers } from "./handlers"; + +export { ExecutionMetricsGroup } from "./group"; +export { ExecutionMetricsHandlers } from "./handlers"; +export { renderPrometheus } from "./prometheus"; + +// HTTP-augmented variant of `executionMetricsPlugin`. The returned plugin +// carries the HTTP `routes` and `handlers` so a host can mount the Prometheus +// scrape (`GET /execution-metrics/metrics`). The scrape reads the global Effect +// Metric registry, so there is no plugin extension state and no +// `extensionService`. Hosts that compose an `HttpApi` import this; SDK-only +// consumers stay on `@executor-js/plugin-execution-metrics` and never load +// `@executor-js/api`. +export const executionMetricsHttpPlugin = definePlugin( + (options?: ExecutionMetricsPluginOptions) => ({ + ...executionMetricsPlugin(options), + routes: () => ExecutionMetricsGroup, + handlers: () => ExecutionMetricsHandlers, + }), +); diff --git a/packages/plugins/execution-metrics/src/api/prometheus.test.ts b/packages/plugins/execution-metrics/src/api/prometheus.test.ts new file mode 100644 index 0000000000..dee6aea730 --- /dev/null +++ b/packages/plugins/execution-metrics/src/api/prometheus.test.ts @@ -0,0 +1,109 @@ +import { describe, expect, it } from "@effect/vitest"; +import { Effect } from "effect"; + +import { + ExecutionFinished, + ExecutionId, + ExecutionStarted, + ExecutionToolCallId, + Subject, + Tenant, + ToolCallFinished, + ToolCallStarted, +} from "@executor-js/sdk/core"; + +import { createExecutionMetricsObserver } from "../sdk/observer"; +import { renderPrometheus } from "./prometheus"; + +const owner = { tenant: Tenant.make("tenant_test"), subject: Subject.make("subject_test") }; + +const executionId = ExecutionId.make("exec_metrics_test"); +const toolCallId = ExecutionToolCallId.make("toolcall_metrics_test"); + +const startedAt = new Date("2026-06-13T00:00:00.000Z"); +const completedAt = new Date("2026-06-13T00:00:00.250Z"); + +describe("execution-metrics observer + renderPrometheus", () => { + it.effect("records counters/histogram a Prometheus scrape can expose", () => + Effect.gen(function* () { + const observer = createExecutionMetricsObserver(); + + yield* observer.handle( + new ExecutionStarted({ + executionId, + owner, + code: "console.log('hi')", + trigger: { kind: "manual" }, + startedAt, + }), + ); + yield* observer.handle( + new ToolCallStarted({ + executionId, + toolCallId, + owner, + path: "github.issues.create", + args: {}, + startedAt, + }), + ); + yield* observer.handle( + new ToolCallFinished({ + executionId, + toolCallId, + owner, + path: "github.issues.create", + status: "completed", + completedAt, + }), + ); + yield* observer.handle( + new ExecutionFinished({ + executionId, + owner, + status: "completed", + result: "ok", + completedAt, + }), + ); + + const text = yield* renderPrometheus; + + // Counter + histogram series names are exposed. + expect(text).toContain("# TYPE executor_execution_started_total counter"); + expect(text).toContain("# TYPE executor_execution_completed_total counter"); + expect(text).toContain("# TYPE executor_tool_call_started_total counter"); + expect(text).toContain("# TYPE executor_tool_call_completed_total counter"); + expect(text).toContain("# TYPE executor_execution_duration_ms histogram"); + expect(text).toContain("executor_execution_duration_ms_bucket"); + expect(text).toContain('le="+Inf"'); + + // Effect's histogram buckets already carry the terminal `+Inf` boundary, + // so the renderer must emit it exactly once per series — a duplicate + // `+Inf` bucket line is a duplicate label set and a scrape rejects it. + const infBuckets = text + .split("\n") + .filter( + (line) => + line.startsWith("executor_execution_duration_ms_bucket{") && line.includes('le="+Inf"'), + ); + expect(infBuckets).toHaveLength(1); + + // The trigger_kind attribute rides the tagged series. + expect(text).toContain('trigger_kind="manual"'); + + // At least one started execution was counted (global registry — value ≥ 1). + const startedLine = text + .split("\n") + .find( + (line) => + line.startsWith("executor_execution_started_total{") && + line.includes('trigger_kind="manual"'), + ); + const startedValue = startedLine + ? Number(startedLine.slice(startedLine.lastIndexOf(" ") + 1)) + : 0; + expect(startedValue).toBeGreaterThanOrEqual(1); + }), + ); +}); diff --git a/packages/plugins/execution-metrics/src/api/prometheus.ts b/packages/plugins/execution-metrics/src/api/prometheus.ts new file mode 100644 index 0000000000..1da67b9028 --- /dev/null +++ b/packages/plugins/execution-metrics/src/api/prometheus.ts @@ -0,0 +1,118 @@ +import { Effect, Match, Metric } from "effect"; + +// --------------------------------------------------------------------------- +// Prometheus text-exposition renderer. +// +// Snapshots the global Effect Metric registry (`Metric.snapshot`) and renders +// the v0.0.4 text format: one HELP/TYPE header pair per metric name, then a +// sample line per attribute-set. Histograms expand to cumulative `_bucket` +// series (with a terminal `le="+Inf"`), `_count`, and `_sum`. +// --------------------------------------------------------------------------- + +const sanitizeName = (raw: string): string => raw.replace(/[^a-zA-Z0-9_:]/g, "_"); + +const escapeLabelValue = (raw: string): string => + raw.replace(/\\/g, "\\\\").replace(/\n/g, "\\n").replace(/"/g, '\\"'); + +const formatValue = (value: bigint | number): string => { + if (typeof value === "bigint") return value.toString(10); + if (value === Number.POSITIVE_INFINITY) return "+Inf"; + if (value === Number.NEGATIVE_INFINITY) return "-Inf"; + return String(value); +}; + +const formatLabels = ( + attributes: Metric.Metric.AttributeSet, + extra?: Readonly>, +): string => { + const pairs = Object.entries(attributes).map( + ([key, value]) => `${sanitizeName(key)}="${escapeLabelValue(String(value))}"`, + ); + if (extra) { + for (const [key, value] of Object.entries(extra)) { + pairs.push(`${sanitizeName(key)}="${escapeLabelValue(value)}"`); + } + } + return pairs.length === 0 ? "" : `{${pairs.join(",")}}`; +}; + +const emitHeader = ( + name: string, + type: "counter" | "gauge" | "histogram" | "summary", + description: string, + seen: Set, +): readonly string[] => { + if (seen.has(name)) return []; + seen.add(name); + const safeDescription = description.replace(/\\/g, "\\\\").replace(/\n/g, "\\n"); + return [`# HELP ${name} ${safeDescription || "(no description)"}`, `# TYPE ${name} ${type}`]; +}; + +const formatPair = (metric: Metric.Metric.Snapshot, seen: Set): readonly string[] => { + const name = sanitizeName(metric.id); + const description = metric.description ?? ""; + const attributes = metric.attributes ?? {}; + return Match.value(metric).pipe( + Match.when({ type: "Counter" }, (metric) => { + const state = metric.state; + return [ + ...emitHeader(name, "counter", description, seen), + `${name}${formatLabels(attributes)} ${formatValue(state.count)}`, + ]; + }), + Match.when({ type: "Gauge" }, (metric) => { + const state = metric.state; + return [ + ...emitHeader(name, "gauge", description, seen), + `${name}${formatLabels(attributes)} ${formatValue(state.value)}`, + ]; + }), + Match.when({ type: "Histogram" }, (metric) => { + const state = metric.state; + const lines = [...emitHeader(name, "histogram", description, seen)]; + // Effect's HistogramState.buckets already ends with the terminal + // `Infinity` boundary (formatted as `le="+Inf"`), whose cumulative count + // equals the total observation count. Emitting it from the loop covers + // Prometheus's required `+Inf` bucket — appending a second `+Inf` line + // would duplicate the series and the scrape would be rejected. + for (const [upperBound, count] of state.buckets) { + lines.push( + `${name}_bucket${formatLabels(attributes, { le: formatValue(upperBound) })} ${count}`, + ); + } + lines.push(`${name}_count${formatLabels(attributes)} ${state.count}`); + lines.push(`${name}_sum${formatLabels(attributes)} ${formatValue(state.sum)}`); + return lines; + }), + Match.when({ type: "Summary" }, (metric) => { + const state = metric.state; + const lines = [...emitHeader(name, "summary", description, seen)]; + for (const [quantile, value] of state.quantiles) { + lines.push( + `${name}${formatLabels(attributes, { quantile: formatValue(quantile) })} ${formatValue( + value ?? Number.NaN, + )}`, + ); + } + lines.push(`${name}_count${formatLabels(attributes)} ${state.count}`); + lines.push(`${name}_sum${formatLabels(attributes)} ${formatValue(state.sum)}`); + return lines; + }), + Match.when({ type: "Frequency" }, (metric) => { + const state = metric.state; + const lines = [...emitHeader(name, "counter", description, seen)]; + for (const [bucket, count] of state.occurrences) { + lines.push(`${name}${formatLabels(attributes, { bucket })} ${formatValue(count)}`); + } + return lines; + }), + Match.exhaustive, + ); +}; + +/** Snapshot the global Effect Metric registry and render Prometheus text. */ +export const renderPrometheus: Effect.Effect = Effect.map(Metric.snapshot, (snapshot) => { + const seen = new Set(); + const lines = snapshot.flatMap((pair) => formatPair(pair, seen)); + return lines.length === 0 ? "\n" : `${lines.join("\n")}\n`; +}); diff --git a/packages/plugins/execution-metrics/src/cloudflare/index.ts b/packages/plugins/execution-metrics/src/cloudflare/index.ts new file mode 100644 index 0000000000..399cdb80bf --- /dev/null +++ b/packages/plugins/execution-metrics/src/cloudflare/index.ts @@ -0,0 +1,91 @@ +import { Effect, Predicate } from "effect"; + +import { type ExecutionEvent, type ExecutionObserver } from "@executor-js/sdk/core"; + +import { rememberStart } from "../sdk/bounded-map"; + +/** + * Minimal structural view of a Workers Analytics Engine binding. The full type + * lives in `@cloudflare/workers-types` as a global ambient `interface + * AnalyticsEngineDataset`, which isn't a named export and would require pulling + * the Workers global lib into this package's `types`. A local interface keeps + * the Cloudflare entry self-contained; a real binding satisfies it structurally. + */ +export interface AnalyticsEngineDataset { + writeDataPoint(event: { + readonly blobs?: ReadonlyArray; + readonly doubles?: ReadonlyArray; + readonly indexes?: ReadonlyArray; + }): void; +} + +/** + * Build an {@link ExecutionObserver} that writes one Analytics Engine data + * point per finished execution / tool call. `writeDataPoint` is synchronous, + * fire-and-forget (no `await`, no `ctx.waitUntil`), so the observer wraps each + * call in `Effect.sync`. Errors are swallowed by the engine's per-observer + * suppression — an observer must never break the execution it watches. + * + * Layout: + * - `ExecutionFinished` → blobs `[ "execution", status, trigger? ]`, doubles + * `[ durationMs ]`, indexes `[ executionId ]`. + * - `ToolCallFinished` → blobs `[ "tool_call", status, path ]`, doubles + * `[ durationMs ]`, indexes `[ executionId ]`. + */ +export const createWaeMetricsObserver = (analytics: AnalyticsEngineDataset): ExecutionObserver => { + const executionStarts = new Map< + string, + { readonly startedAt: number; readonly trigger?: string } + >(); + const toolCallStarts = new Map(); + + return { + handle: (event: ExecutionEvent) => { + if (Predicate.isTagged(event, "ExecutionStarted")) { + return Effect.sync(() => { + rememberStart(executionStarts, event.executionId, { + startedAt: event.startedAt.getTime(), + trigger: event.trigger?.kind, + }); + }); + } + + if (Predicate.isTagged(event, "ToolCallStarted")) { + return Effect.sync(() => { + rememberStart(toolCallStarts, event.toolCallId, event.startedAt.getTime()); + }); + } + + if (Predicate.isTagged(event, "ExecutionFinished")) { + return Effect.sync(() => { + const started = executionStarts.get(event.executionId); + executionStarts.delete(event.executionId); + const durationMs = started + ? Math.max(0, event.completedAt.getTime() - started.startedAt) + : 0; + analytics.writeDataPoint({ + blobs: ["execution", event.status, started?.trigger ?? ""], + doubles: [durationMs], + indexes: [event.executionId], + }); + }); + } + + if (Predicate.isTagged(event, "ToolCallFinished")) { + return Effect.sync(() => { + const startedAt = toolCallStarts.get(event.toolCallId); + toolCallStarts.delete(event.toolCallId); + const durationMs = + startedAt !== undefined ? Math.max(0, event.completedAt.getTime() - startedAt) : 0; + analytics.writeDataPoint({ + blobs: ["tool_call", event.status, event.path], + doubles: [durationMs], + indexes: [event.executionId], + }); + }); + } + + return Effect.void; + }, + }; +}; diff --git a/packages/plugins/execution-metrics/src/sdk/bounded-map.ts b/packages/plugins/execution-metrics/src/sdk/bounded-map.ts new file mode 100644 index 0000000000..123cd3e618 --- /dev/null +++ b/packages/plugins/execution-metrics/src/sdk/bounded-map.ts @@ -0,0 +1,17 @@ +// Both observer sinks track per-execution start times in a Map so a Finished +// event can compute the elapsed duration. An execution killed before its +// Finished event (process signal, VM timeout, hard shutdown) would otherwise +// leak its entry forever in a long-lived host. `rememberStart` bounds the Map: +// when it's at capacity and the key is new, the oldest entry (insertion-order +// first) is evicted — so a steady stream of orphaned starts can't grow without +// limit. + +const MAX_TRACKED = 10_000; + +export const rememberStart = (map: Map, key: K, value: V): void => { + if (map.size >= MAX_TRACKED && !map.has(key)) { + const oldest = map.keys().next().value; + if (oldest !== undefined) map.delete(oldest); + } + map.set(key, value); +}; diff --git a/packages/plugins/execution-metrics/src/sdk/index.ts b/packages/plugins/execution-metrics/src/sdk/index.ts new file mode 100644 index 0000000000..b7e0e3732b --- /dev/null +++ b/packages/plugins/execution-metrics/src/sdk/index.ts @@ -0,0 +1,2 @@ +export { executionMetricsPlugin, type ExecutionMetricsPluginOptions } from "./plugin"; +export { createExecutionMetricsObserver } from "./observer"; diff --git a/packages/plugins/execution-metrics/src/sdk/observer.ts b/packages/plugins/execution-metrics/src/sdk/observer.ts new file mode 100644 index 0000000000..c95dddc1a3 --- /dev/null +++ b/packages/plugins/execution-metrics/src/sdk/observer.ts @@ -0,0 +1,124 @@ +import { Effect, Metric, Predicate } from "effect"; + +import { + type ExecutionEvent, + type ExecutionId, + type ExecutionObserver, +} from "@executor-js/sdk/core"; + +import { rememberStart } from "./bounded-map"; + +// --------------------------------------------------------------------------- +// Metric definitions — registered in the global Effect Metric registry the +// first time this module is loaded. The Prometheus scrape (`../api/prometheus`) +// and the Effect OTLP/Prometheus exporters read the same registry. +// --------------------------------------------------------------------------- + +const executionsStarted = Metric.counter("executor.execution.started_total", { + description: "Executor executions started.", + incremental: true, +}); + +const executionsCompleted = Metric.counter("executor.execution.completed_total", { + description: "Executor executions completed.", + incremental: true, +}); + +const executionsFailed = Metric.counter("executor.execution.failed_total", { + description: "Executor executions failed.", + incremental: true, +}); + +const executionDuration = Metric.histogram("executor.execution.duration_ms", { + description: "Executor execution duration in milliseconds.", + boundaries: Metric.exponentialBoundaries({ start: 1, factor: 2, count: 16 }), +}); + +const toolCallsStarted = Metric.counter("executor.tool_call.started_total", { + description: "Executor tool calls started.", + incremental: true, +}); + +const toolCallsCompleted = Metric.counter("executor.tool_call.completed_total", { + description: "Executor tool calls completed.", + incremental: true, +}); + +const toolCallsFailed = Metric.counter("executor.tool_call.failed_total", { + description: "Executor tool calls failed.", + incremental: true, +}); + +const triggerAttributes = (triggerKind: string | undefined): Metric.Metric.AttributeSet => + triggerKind ? { trigger_kind: triggerKind } : {}; + +/** + * Build an {@link ExecutionObserver} that drives the global Effect Metric + * registry from the execution lifecycle stream. Counters track executions and + * tool calls (started / completed / failed); a histogram records execution + * duration tagged by `trigger_kind`. Start timestamps are buffered per + * execution so {@link ExecutionFinished} can compute the elapsed duration. + */ +export const createExecutionMetricsObserver = (): ExecutionObserver => { + const executionStarts = new Map< + ExecutionId, + { readonly startedAt: Date; readonly trigger?: string } + >(); + + const updateCounter = ( + counter: Metric.Counter, + attributes: Metric.Metric.AttributeSet = {}, + ): Effect.Effect => Metric.update(Metric.withAttributes(counter, attributes), 1); + + const handleExecutionStarted = ( + event: Extract, + ) => + Effect.sync(() => { + rememberStart(executionStarts, event.executionId, { + startedAt: event.startedAt, + trigger: event.trigger?.kind, + }); + }).pipe( + Effect.andThen(updateCounter(executionsStarted, triggerAttributes(event.trigger?.kind))), + ); + + const handleExecutionFinished = ( + event: Extract, + ) => + Effect.gen(function* () { + const started = executionStarts.get(event.executionId); + if (started) { + executionStarts.delete(event.executionId); + const durationMs = Math.max(0, event.completedAt.getTime() - started.startedAt.getTime()); + yield* Metric.update( + Metric.withAttributes(executionDuration, triggerAttributes(started.trigger)), + durationMs, + ); + } + + const attributes = triggerAttributes(started?.trigger); + return yield* event.status === "completed" + ? updateCounter(executionsCompleted, attributes) + : updateCounter(executionsFailed, attributes); + }); + + 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" + ? updateCounter(toolCallsCompleted) + : updateCounter(toolCallsFailed); + } + return Effect.void; + }, + }; +}; diff --git a/packages/plugins/execution-metrics/src/sdk/plugin.ts b/packages/plugins/execution-metrics/src/sdk/plugin.ts new file mode 100644 index 0000000000..1769bebca4 --- /dev/null +++ b/packages/plugins/execution-metrics/src/sdk/plugin.ts @@ -0,0 +1,26 @@ +import { definePlugin, type ExecutionObserver } from "@executor-js/sdk/core"; + +import { createExecutionMetricsObserver } from "./observer"; + +export interface ExecutionMetricsPluginOptions { + /** Override the observer factory. Defaults to the Effect Metric sink + * ({@link createExecutionMetricsObserver}). A host on Cloudflare can pass a + * Workers Analytics Engine observer here instead. */ + readonly observer?: () => ExecutionObserver; +} + +/** + * Opt-in execution-metrics sink. Registers an {@link ExecutionObserver} via + * `runtime.executionObserver` so the engine fans every {@link ExecutionEvent} + * to it. The base plugin carries NO HTTP surface — import + * `@executor-js/plugin-execution-metrics/api` for the Prometheus scrape, and + * never load `@executor-js/api` from here. + */ +export const executionMetricsPlugin = definePlugin((options?: ExecutionMetricsPluginOptions) => ({ + id: "execution-metrics" as const, + packageName: "@executor-js/plugin-execution-metrics", + storage: () => ({}), + runtime: { + executionObserver: options?.observer ?? createExecutionMetricsObserver, + }, +})); diff --git a/packages/plugins/execution-metrics/tsconfig.json b/packages/plugins/execution-metrics/tsconfig.json new file mode 100644 index 0000000000..eebc1e6f14 --- /dev/null +++ b/packages/plugins/execution-metrics/tsconfig.json @@ -0,0 +1,24 @@ +{ + "compilerOptions": { + "target": "ES2022", + "module": "ESNext", + "moduleResolution": "Bundler", + "strict": true, + "skipLibCheck": true, + "lib": ["ES2022"], + "types": ["bun-types", "node"], + "noUnusedLocals": true, + "noImplicitOverride": true, + "plugins": [ + { + "name": "@effect/language-service", + "ignoreEffectSuggestionsInTscExitCode": true, + "ignoreEffectWarningsInTscExitCode": true, + "diagnosticSeverity": { + "preferSchemaOverJson": "off" + } + } + ] + }, + "include": ["src/**/*.ts"] +} diff --git a/packages/plugins/execution-metrics/tsup.config.ts b/packages/plugins/execution-metrics/tsup.config.ts new file mode 100644 index 0000000000..21cb020a31 --- /dev/null +++ b/packages/plugins/execution-metrics/tsup.config.ts @@ -0,0 +1,14 @@ +import { defineConfig } from "tsup"; + +export default defineConfig({ + entry: { + index: "src/sdk/index.ts", + api: "src/api/index.ts", + cloudflare: "src/cloudflare/index.ts", + }, + format: ["esm"], + dts: false, + sourcemap: true, + clean: true, + external: [/^@executor-js\//, /^effect/, /^@effect\//], +}); diff --git a/packages/plugins/execution-metrics/vitest.config.ts b/packages/plugins/execution-metrics/vitest.config.ts new file mode 100644 index 0000000000..5bfa2d586e --- /dev/null +++ b/packages/plugins/execution-metrics/vitest.config.ts @@ -0,0 +1,8 @@ +import { defineConfig } from "vitest/config"; + +export default defineConfig({ + test: { + include: ["src/**/*.test.ts"], + passWithNoTests: true, + }, +});