diff --git a/apps/ai/package.json b/apps/ai/package.json index 8aec98c71..b03889b2f 100644 --- a/apps/ai/package.json +++ b/apps/ai/package.json @@ -18,6 +18,7 @@ "@effect-agent/sandbox": "0.1.0-beta.74", "@effect/ai-openai-compat": "catalog:effect", "@effect/ai-openrouter": "catalog:effect", + "@maple-dev/effect-sdk": "workspace:*", "@maple/db": "workspace:*", "@maple/domain": "workspace:*", "@maple/infra": "workspace:*", @@ -33,7 +34,6 @@ "@effect-agent/testing": "0.1.0-beta.74", "@effect/language-service": "catalog:effect", "@effect/vitest": "catalog:effect", - "@maple-dev/effect-sdk": "workspace:*", "@types/node": "catalog:tooling", "ai": "^6.0.196", "gpt-tokenizer": "^3.0.1", diff --git a/apps/ai/src/mcp/expected-failures.test.ts b/apps/ai/src/mcp/expected-failures.test.ts index 4c677e126..8559a06c2 100644 --- a/apps/ai/src/mcp/expected-failures.test.ts +++ b/apps/ai/src/mcp/expected-failures.test.ts @@ -43,4 +43,20 @@ describe("MCP expected failures", () => { expect(offenders.map((path) => path.slice(SRC_ROOT.length))).toEqual([]) }) + + // The check above only sees a tracer built through `MapleCloudflareSDK.make`. + // This Worker's request-facing tracer is built by `WorkerTelemetry` in the + // init instead, and it is the one serving the PUBLIC `/mcp` transport — the + // surface these identifiers were written for. It lost them once already, in + // the move out of apps/api, because nothing tied the two files together. + it("spreads the MCP identifiers into this Worker's own telemetry", () => { + const offenders = sourceFiles(SRC_ROOT).filter((path) => { + const source = readFileSync(path, "utf8") + return ( + source.includes("WorkerTelemetry({") && !source.includes("MCP_ANTICIPATED_ERROR_IDENTIFIERS") + ) + }) + + expect(offenders.map((path) => path.slice(SRC_ROOT.length))).toEqual([]) + }) }) diff --git a/apps/ai/src/worker.ts b/apps/ai/src/worker.ts index c74d3358d..fb0194007 100644 --- a/apps/ai/src/worker.ts +++ b/apps/ai/src/worker.ts @@ -46,11 +46,14 @@ import { } from "@maple/infra/env" import { WorkerTelemetry } from "@maple/infra/worker-telemetry" import * as Cloudflare from "alchemy/Cloudflare" +import * as AlchemyTelemetry from "alchemy/Telemetry" import { Context, Effect, Layer } from "effect" import { ChatSessionLive, ChatSessionObject } from "@ai/chat/ChatSession" +import { MCP_ANTICIPATED_ERROR_IDENTIFIERS } from "@ai/mcp/expected-failures" import InvestigationFanoutWorkflow from "@ai/workflows/InvestigationFanoutWorkflow" import { aiPorts, AiBindingLayers, bindAiClients } from "@ai/worker/bindings" import { buildApp, makeFetch } from "@ai/worker/http" +import { AiObservabilityLive } from "@ai/worker/observability" /** * The AI worker's resource bindings, split from the `Config`-sourced env so @@ -169,7 +172,22 @@ export default MapleAi.make( // implementation; yielding the class above is what forces this to run, // so the class reaches the generated entry's exports. ChatSessionLive, - WorkerTelemetry({ serviceName: "maple-ai" }), + WorkerTelemetry({ + serviceName: "maple-ai", + // Both carried over from apps/api with the surfaces they describe. + // `dropSpanNames` keeps the MCP server's notification spans out of + // export; the MCP identifiers are what keep an expected 400/401 — + // a tool call that does not decode, a missing credential — exporting + // with an `Ok` status and no exception event, per CLAUDE.md's rule + // that only 5xx is an `Error` span. `chat/turn-runner.ts` and the + // fan-out Workflow pass the same set to their own tracers; this is + // the public `/mcp` transport's. + dropSpanNames: ["McpServer/Notifications."], + anticipatedErrorIdentifiers: MCP_ANTICIPATED_ERROR_IDENTIFIERS, + }), + // The references the bridge's `HttpMiddleware.tracer` reads, built into + // every event beside the SDK; they cannot live in the app graph. + AlchemyTelemetry.layer(AiObservabilityLive), ), ), ), diff --git a/apps/ai/src/worker/observability.ts b/apps/ai/src/worker/observability.ts new file mode 100644 index 000000000..eb04d1da4 --- /dev/null +++ b/apps/ai/src/worker/observability.ts @@ -0,0 +1,22 @@ +import { Layer } from "effect" +import { HttpMiddleware } from "effect/unstable/http" + +/** + * The one reference `HttpMiddleware.tracer` reads that this Worker has to set + * itself. The bridge's tracer runs outside the app graph, so it cannot live + * there — the Worker registers this beside the SDK exporters, exactly as + * `apps/api/src/http/api-observability.ts` does for the api. + * + * Only the filter. Header redaction is left at Effect's defaults + * (`authorization`, `cookie`, `set-cookie`, `x-api-key`), which already cover + * every credential reaching this Worker: the provider webhook signatures api + * adds to its own list are received on api's routes and never forwarded here. + * + * `/health` and `OPTIONS` are skipped for the reason CLAUDE.md gives for the + * api: this Worker's traces land in Maple's own org, so a liveness probe that + * spans is self-traffic nothing reads. + */ +export const AiObservabilityLive = Layer.succeed( + HttpMiddleware.TracerDisabledWhen, + (request: { url: string; method: string }) => request.url === "/health" || request.method === "OPTIONS", +) diff --git a/apps/ai/src/workflows/InvestigationFanoutWorkflow.ts b/apps/ai/src/workflows/InvestigationFanoutWorkflow.ts index f622786bc..6648ff382 100644 --- a/apps/ai/src/workflows/InvestigationFanoutWorkflow.ts +++ b/apps/ai/src/workflows/InvestigationFanoutWorkflow.ts @@ -1,9 +1,12 @@ /** - * The investigation fan-out Workflow, in alchemy's form: yielded from the api + * The investigation fan-out Workflow, in alchemy's form: yielded from the AI * Worker's init, which binds it as `InvestigationFanoutWorkflow`, registers the * physical workflow and exports the class from the generated entry. N lens * agents run in parallel, then one validator promotes a single cause and * records why each rival lost. + * + * The api and alerting Workers bind the same physical workflow cross-script + * under this class name — see `@maple/domain/investigation-fanout`. */ import { ChatSessionObject } from "@ai/chat/ChatSession" import { MCP_ANTICIPATED_ERROR_IDENTIFIERS } from "@ai/mcp/expected-failures" @@ -34,7 +37,11 @@ export default class InvestigationFanoutWorkflow extends Cloudflare.Workflow()( ApiBindingLayers, Cloudflare.Workers.CronEventSourceLive, Cloudflare.Queues.EventSourceLive, - WorkerTelemetry({ - serviceName: "maple-api", - dropSpanNames: ["McpServer/Notifications."], - }), + // No `dropSpanNames`: the MCP server's notification spans are maple-ai's + // to drop now, and its telemetry config is where that option lives. + WorkerTelemetry({ serviceName: "maple-api" }), // The references the bridge's `HttpMiddleware.tracer` reads, built into // every event beside the SDK; they cannot live in the app graph. AlchemyTelemetry.layer(ApiObservabilityLive), diff --git a/apps/api/src/worker/ai-forward.test.ts b/apps/api/src/worker/ai-forward.test.ts new file mode 100644 index 000000000..2728b8025 --- /dev/null +++ b/apps/api/src/worker/ai-forward.test.ts @@ -0,0 +1,106 @@ +import { assert, describe, it } from "@effect/vitest" +import { Effect } from "effect" +import { HttpServerRequest } from "effect/unstable/http" +import { forwardsToAi, forwardToAi, isCloudflareFetcher } from "./ai-forward" + +/** + * The predicate IS the contract between the two Workers: a path it misses 404s + * from api's own router, and one it over-matches never reaches api's routes at + * all. Both failures are silent in a typecheck, hence the table. + */ +describe("forwardsToAi", () => { + it("forwards the MCP transport and the chat surfaces", () => { + for (const path of [ + "/mcp", + "/mcp/", + "/mcp/anything", + "/api/chat/sessions/org:tab/history", + "/api/chat/sessions/org:tab/events", + "/api/chat/sessions/org:tab/messages", + "/api/chat/sessions/org:tab/abort", + "/internal/chat/apply", + ]) { + assert.isTrue(forwardsToAi(path), path) + } + }) + + it("keeps every other path on api, including neighbours of the forwarded ones", () => { + for (const path of [ + "/", + "/health", + "/mcp-something", + "/mcpx", + "/.well-known/oauth-protected-resource/mcp", + "/oauth/authorize", + "/api/chatter", + "/api/chat", + "/internal/chatty", + "/internal/ai-sessions", + "/v2/errors", + ]) { + assert.isFalse(forwardsToAi(path), path) + } + }) +}) + +describe("isCloudflareFetcher", () => { + it("accepts a binding exposing fetch and rejects everything else", () => { + assert.isTrue(isCloudflareFetcher({ fetch: () => new Response("ok") })) + for (const value of [undefined, null, {}, "AI_WORKER", { fetch: "nope" }]) { + assert.isFalse(isCloudflareFetcher(value), JSON.stringify(value)) + } + }) +}) + +/** + * SAFETY: the forward calls `fetch` and nothing else, so the stub implements + * that and is asserted to the binding's type rather than growing a `connect` + * this path never reaches. + */ +const echoFetcher = (seen: Array) => + ({ + fetch: (input: RequestInfo | URL) => { + if (input instanceof Request) seen.push(input) + return Promise.resolve(new Response("forwarded", { status: 207 })) + }, + }) as Fetcher + +describe("forwardToAi", () => { + it.effect("hands maple-ai this hop's span as traceparent, and keeps the rest of the request", () => + Effect.gen(function* () { + const seen: Array = [] + const request = HttpServerRequest.fromWeb( + new Request("https://api.maple.dev/mcp", { + method: "POST", + headers: { authorization: "Bearer key", "content-type": "application/json" }, + body: '{"method":"tools/list"}', + }), + ) + + const response = yield* forwardToAi(echoFetcher(seen), request) + const span = yield* Effect.currentSpan + + assert.strictEqual(response.status, 207) + const forwarded = seen[0] + assert.isDefined(forwarded) + assert.strictEqual(forwarded.method, "POST") + assert.strictEqual(forwarded.headers.get("authorization"), "Bearer key") + assert.strictEqual(new URL(forwarded.url).host, "api.maple.dev") + // Parented to api's own server span, so the two Workers share one trace + // and the service map gains the edge between them. + assert.strictEqual(forwarded.headers.get("traceparent"), `00-${span.traceId}-${span.spanId}-01`) + assert.strictEqual(yield* Effect.promise(() => forwarded.text()), '{"method":"tools/list"}') + }).pipe(Effect.withSpan("test-root")), + ) + + it.effect("forwards unchanged when nothing is tracing", () => + Effect.gen(function* () { + const seen: Array = [] + const request = HttpServerRequest.fromWeb(new Request("https://api.maple.dev/mcp")) + + yield* forwardToAi(echoFetcher(seen), request) + + assert.isNull(seen[0]?.headers.get("traceparent") ?? null) + }), + ) +}) diff --git a/apps/api/src/worker/ai-forward.ts b/apps/api/src/worker/ai-forward.ts new file mode 100644 index 000000000..216dda00f --- /dev/null +++ b/apps/api/src/worker/ai-forward.ts @@ -0,0 +1,102 @@ +// BOUNDARY: This module owns unparsed external values and narrows them before domain use. +/** + * Forwarding the agent surfaces to maple-ai. + * + * The api keeps the hostname and hands `/mcp` and the chat paths to the AI + * Worker over a service binding, which is what keeps `/mcp`'s OAuth issuer and + * its RFC 8707 resource identifiers on this origin — moving them would + * invalidate every registered MCP client. + * + * Its own module for two reasons: the path predicate is the contract between + * the two Workers and is worth testing without building a route graph, and the + * binding arrives as an unparsed `env` value that has to be narrowed rather + * than asserted. + */ +import * as Cloudflare from "alchemy/Cloudflare" +import { Effect, Option } from "effect" +import { HttpServerRequest, HttpServerResponse } from "effect/unstable/http" +import { API_CORS_RESPONSE_HEADERS } from "../http/api-cors" + +/** + * The paths maple-ai serves. + * + * `/mcp` is matched exactly rather than by prefix so a future `/mcp-something` + * on this origin is not silently swallowed. Kept in step with + * `MapleAiApi` and the AI Worker's raw routers: anything this misses 404s from + * api's own router, and anything it over-matches never reaches api's routes at + * all. + */ +export const forwardsToAi = (path: string): boolean => + path === "/mcp" || + path.startsWith("/mcp/") || + path.startsWith("/api/chat/") || + path.startsWith("/internal/chat/") + +/** + * A service binding off `env`, narrowed rather than cast. + * + * Only `fetch` is checked, because that is all the forward calls; a binding + * that is present but not a fetcher is then a logged 503 instead of a defect + * inside alchemy's adapter. + */ +export const isCloudflareFetcher = (value: unknown): value is Fetcher => + typeof value === "object" && value !== null && typeof (value as { fetch?: unknown }).fetch === "function" + +/** maple-ai could not be reached. CORS headers included: the dashboard's chat calls this from a browser. */ +export const aiUnavailableResponse = (): HttpServerResponse.HttpServerResponse => + HttpServerResponse.text("maple-ai is unavailable", { + status: 503, + headers: API_CORS_RESPONSE_HEADERS, + }) + +/** The W3C header carrying this hop's span, sampled flag included, as `apps/web`'s fetch wrapper writes it. */ +const traceparentOf = (span: { readonly traceId: string; readonly spanId: string }): string => + `00-${span.traceId}-${span.spanId}-01` + +/** + * Forward one request to maple-ai and answer with its response. + * + * Byte-transparent but for one header: the same method, the original `Host` + * (which is what keeps `/mcp`'s OAuth `resource_metadata` pointing at this + * origin's well-known), every other header, and both bodies as streams. The + * chat tail is an open `text/event-stream`, so buffering either side would turn + * a live transcript into a hang. + * + * The exception is `traceparent`, which is replaced with THIS hop's server + * span. Without it maple-ai starts a new root trace and one `/mcp` call reads + * as two unrelated traces with no edge between the Workers on the service map. + * Replacing rather than preserving is what parents ai's span to api's: a client + * that sent its own `traceparent` is already this span's parent, so the trace + * stays whole either way. + * + * The one expected failure is a request that cannot be rendered to the web + * shape, which on workerd means never — the bridge's request already IS one, so + * `toWeb` hands it back untouched. It is answered as the same 503 an absent + * binding gets rather than widened into this handler's error channel, where it + * would not match the bridge's `HttpEffect`. A binding call that rejects stays a + * defect the bridge renders and reports, exactly as before the split. + */ +export const forwardToAi = ( + fetcher: Fetcher, + request: HttpServerRequest.HttpServerRequest, +): Effect.Effect => + Effect.gen(function* () { + const web = yield* HttpServerRequest.toWeb(request) + const span = yield* Effect.option(Effect.currentSpan) + const forwarded = Option.match(span, { + onNone: () => web, + onSome: (current) => { + const headers = new Headers(web.headers) + headers.set("traceparent", traceparentOf(current)) + return new Request(web, { headers }) + }, + }) + return yield* Cloudflare.fromCloudflareFetcher(fetcher).fetch(HttpServerRequest.fromWeb(forwarded)) + }).pipe( + Effect.catch((error) => + Effect.logError("Forwarding to the AI worker failed", error).pipe( + Effect.annotateLogs({ method: request.method, path: request.url }), + Effect.as(aiUnavailableResponse()), + ), + ), + ) diff --git a/apps/api/src/worker/http.ts b/apps/api/src/worker/http.ts index 0f92085c9..37514c566 100644 --- a/apps/api/src/worker/http.ts +++ b/apps/api/src/worker/http.ts @@ -9,6 +9,7 @@ import { HttpRouter, HttpServerRequest, HttpServerResponse } from "effect/unstab import * as Etag from "effect/unstable/http/Etag" import * as HttpPlatform from "effect/unstable/http/HttpPlatform" import { API_CORS_RESPONSE_HEADERS, apiCorsPreflightResponse } from "../http/api-cors" +import { aiUnavailableResponse, forwardsToAi, forwardToAi, isCloudflareFetcher } from "./ai-forward" import { v2WorkerUnavailableResponse } from "../http/v2-worker-unavailable" import type { MapleDbConnection } from "../platform/bindings" import { layerPg } from "../platform/DatabasePgLive" @@ -121,16 +122,6 @@ const bridgeHandler = ( >, ): HttpEffect => handler as HttpEffect -/** - * The paths maple-ai serves. `/mcp` is matched exactly rather than by prefix so - * a future `/mcp-something` on this origin is not silently swallowed. - */ -const forwardsToAi = (path: string): boolean => - path === "/mcp" || - path.startsWith("/mcp/") || - path.startsWith("/api/chat/") || - path.startsWith("/internal/chat/") - const pathOf = (url: string): string => { const query = url.indexOf("?") return query === -1 ? url : url.slice(0, query) @@ -214,26 +205,17 @@ export const makeFetch = (app: Effect.Effect, ports: Layer. // The agent surfaces moved to maple-ai; this origin keeps serving them. // Ahead of the route graph on purpose — that is the whole point of the // split, so a `/mcp` call no longer builds `AllRoutes` and `ApiAuthLive`. - // - // The forward must stay byte-transparent: the same method, the original - // `Host` (which is what keeps `/mcp`'s OAuth `resource_metadata` pointing - // at this origin's well-known), every header, and both bodies as streams. - // The chat tail is an open `text/event-stream`, so buffering either side - // would turn a live transcript into a hang. + // What the forward preserves, and the one header it replaces, is spelled + // out in `ai-forward.ts`. if (forwardsToAi(path)) { const aiWorker = (yield* Cloudflare.WorkerEnvironment).AI_WORKER - if (aiWorker === undefined) { + if (!isCloudflareFetcher(aiWorker)) { yield* Effect.logError("AI worker binding is missing").pipe( Effect.annotateLogs({ method: request.method, path }), ) - return HttpServerResponse.text("maple-ai is unavailable", { - status: 503, - headers: API_CORS_RESPONSE_HEADERS, - }) + return aiUnavailableResponse() } - return yield* Cloudflare.fromCloudflareFetcher( - aiWorker as Parameters[0], - ).fetch(request) + return yield* forwardToAi(aiWorker, request) } const startedAt = yield* Clock.currentTimeMillis diff --git a/bun.lock b/bun.lock index ce8e637f3..3d48f7716 100644 --- a/bun.lock +++ b/bun.lock @@ -34,6 +34,7 @@ "@effect-agent/sandbox": "0.1.0-beta.74", "@effect/ai-openai-compat": "catalog:effect", "@effect/ai-openrouter": "catalog:effect", + "@maple-dev/effect-sdk": "workspace:*", "@maple/db": "workspace:*", "@maple/domain": "workspace:*", "@maple/infra": "workspace:*", @@ -49,7 +50,6 @@ "@effect-agent/testing": "0.1.0-beta.74", "@effect/language-service": "catalog:effect", "@effect/vitest": "catalog:effect", - "@maple-dev/effect-sdk": "workspace:*", "@types/node": "catalog:tooling", "ai": "^6.0.196", "gpt-tokenizer": "^3.0.1", diff --git a/packages/domain/src/investigation-fanout.ts b/packages/domain/src/investigation-fanout.ts index 12ec3b0f1..8c0b899b7 100644 --- a/packages/domain/src/investigation-fanout.ts +++ b/packages/domain/src/investigation-fanout.ts @@ -1,10 +1,10 @@ /** * The investigation fan-out Workflow's contract, as its two callers see it. * - * The api Worker hosts the Workflow as an alchemy class and binds it under - * the class name; the alerting Worker binds the same physical workflow + * The AI Worker hosts the Workflow as an alchemy class and binds it under the + * class name; the api and alerting Workers bind the same physical workflow * cross-script under the SAME key, because the services that start - * investigations (`AlertsService`, `ErrorsService`, …) run in both Workers and + * investigations (`AlertsService`, `ErrorsService`, …) run in those Workers and * read the binding by this one name. */ import type { IssueSeverity } from "./http/errors"