From 50de16cb5636d303a0a3db721e050bcdb286d1c2 Mon Sep 17 00:00:00 2001 From: bk-agent-01 Date: Sat, 26 Sep 2026 08:37:10 +0000 Subject: [PATCH] feat(server): event feed for followers GET /api/orchestration/events?after=&limit=<1..1000>&wait=<0..25> returns events after a cursor in sequence order, shaped like the WebSocket replay. An empty read is held until the engine publishes a newer event or the wait elapses (woken by the live event stream, no polling). A cursor ahead of the log gets 410 cursor-invalid. This lets the Linear bridge stop polling the full shell every 2 s. Co-Authored-By: Claude Opus 5.5 --- .../eventFeedHttp.expbkt3.test.ts | 243 ++++++++++++++++++ .../orchestration/eventFeedHttp.expbkt3.ts | 242 +++++++++++++++++ .../orchestration/forkRouteAuth.expbkt3.ts | 45 ++++ apps/server/src/server.ts | 3 + 4 files changed, 533 insertions(+) create mode 100644 apps/server/src/orchestration/eventFeedHttp.expbkt3.test.ts create mode 100644 apps/server/src/orchestration/eventFeedHttp.expbkt3.ts create mode 100644 apps/server/src/orchestration/forkRouteAuth.expbkt3.ts diff --git a/apps/server/src/orchestration/eventFeedHttp.expbkt3.test.ts b/apps/server/src/orchestration/eventFeedHttp.expbkt3.test.ts new file mode 100644 index 0000000000000..d5bc1b36a03ed --- /dev/null +++ b/apps/server/src/orchestration/eventFeedHttp.expbkt3.test.ts @@ -0,0 +1,243 @@ +import { + AuthSessionId, + CommandId, + type OrchestrationEvent, + ProjectId, + ProviderInstanceId, + UserId, +} from "@t3tools/contracts"; +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { NodeHttpServer } from "@effect/platform-node"; +import { assert, describe, it } from "@effect/vitest"; +import * as Effect from "effect/Effect"; +import * as Fiber from "effect/Fiber"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import { TestClock } from "effect/testing"; +import { HttpClient, HttpClientRequest, HttpRouter } from "effect/unstable/http"; + +import { EnvironmentAuth } from "../auth/EnvironmentAuth.ts"; +import { ServerConfig } from "../config.ts"; +import { OrchestrationCommandReceiptRepositoryLive } from "../persistence/Layers/OrchestrationCommandReceipts.ts"; +import { OrchestrationEventStoreLive } from "../persistence/Layers/OrchestrationEventStore.ts"; +import { SqlitePersistenceMemory } from "../persistence/Layers/Sqlite.ts"; +import * as RepositoryIdentityResolver from "../project/RepositoryIdentityResolver.ts"; +import { + EVENT_FEED_PATH, + eventFeedRouteLayer, + parseEventFeedQuery, + readEventFeedPage, +} from "./eventFeedHttp.expbkt3.ts"; +import { OrchestrationEngineLive } from "./Layers/OrchestrationEngine.ts"; +import { OrchestrationProjectionPipelineLive } from "./Layers/ProjectionPipeline.ts"; +import { OrchestrationProjectionSnapshotQueryLive } from "./Layers/ProjectionSnapshotQuery.ts"; +import { OrchestrationAccessControl } from "./Services/AccessControl.ts"; +import { OrchestrationEngineService } from "./Services/OrchestrationEngine.ts"; +import * as ThreadBackgroundLiveness from "./ThreadBackgroundLiveness.ts"; +import * as ThreadPlanProgress from "./ThreadPlanProgress.ts"; + +const hiddenProjectId = ProjectId.make("project-hidden"); + +const engineLayer = () => + OrchestrationEngineLive.pipe( + Layer.provideMerge(OrchestrationProjectionSnapshotQueryLive), + Layer.provide(ThreadBackgroundLiveness.layer), + Layer.provide(ThreadPlanProgress.layer), + Layer.provide(OrchestrationProjectionPipelineLive), + Layer.provide(OrchestrationEventStoreLive), + Layer.provide(OrchestrationCommandReceiptRepositoryLive), + Layer.provideMerge(RepositoryIdentityResolver.layer), + Layer.provideMerge(SqlitePersistenceMemory), + Layer.provideMerge(ServerConfig.layerTest(process.cwd(), { prefix: "t3-event-feed-" })), + Layer.provideMerge(NodeServices.layer), + // Team-mode actors see every project except `project-hidden`. + Layer.provideMerge( + Layer.succeed(OrchestrationAccessControl, { + actorFor: (subject: string) => + subject === "team-user" ? Option.some(UserId.make("user-team")) : Option.none(), + canAccessThread: () => Effect.succeed(true), + canAccessProject: (_userId: UserId, projectId: ProjectId) => + Effect.succeed(projectId !== hiddenProjectId), + canTransferThreadOwnership: () => Effect.succeed(false), + canTransferProjectOwnership: () => Effect.succeed(false), + }), + ), + ); + +const createProject = (id: string) => + Effect.gen(function* () { + const engine = yield* OrchestrationEngineService; + return yield* engine.dispatch({ + type: "project.create", + commandId: CommandId.make(`cmd-${id}`), + projectId: ProjectId.make(id), + title: id, + workspaceRoot: `/tmp/${id}`, + defaultModelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5-codex" }, + createdAt: "2026-09-26T00:00:00.000Z", + }); + }); + +const sequences = (events: ReadonlyArray) => + events.map((event) => event.sequence); + +describe("parseEventFeedQuery", () => { + it("applies defaults and bounds", () => { + assert.deepStrictEqual(parseEventFeedQuery(new URLSearchParams("after=0")), { + after: 0, + limit: 500, + waitSeconds: 0, + }); + assert.deepStrictEqual(parseEventFeedQuery(new URLSearchParams("after=7&limit=1000&wait=25")), { + after: 7, + limit: 1000, + waitSeconds: 25, + }); + assert.isString(parseEventFeedQuery(new URLSearchParams(""))); + assert.isString(parseEventFeedQuery(new URLSearchParams("after=-1"))); + assert.isString(parseEventFeedQuery(new URLSearchParams("after=1.5"))); + assert.isString(parseEventFeedQuery(new URLSearchParams("after=0&limit=0"))); + assert.isString(parseEventFeedQuery(new URLSearchParams("after=0&limit=1001"))); + assert.isString(parseEventFeedQuery(new URLSearchParams("after=0&wait=26"))); + }); +}); + +describe("readEventFeedPage", () => { + it.effect("returns events after the cursor in sequence order, up to the limit", () => + Effect.gen(function* () { + for (const id of ["project-a", "project-b", "project-c"]) yield* createProject(id); + + const page = yield* readEventFeedPage({ after: 0, limit: 2, waitSeconds: 0 }, null); + assert.strictEqual(page._tag, "Page"); + if (page._tag !== "Page") return; + assert.deepStrictEqual(sequences(page.events), [1, 2]); + assert.strictEqual(page.nextAfter, 2); + assert.strictEqual(page.headSequence, 3); + + const rest = yield* readEventFeedPage({ after: 2, limit: 500, waitSeconds: 0 }, null); + assert.isTrue(rest._tag === "Page" && rest.nextAfter === 3 && rest.events.length === 1); + + const empty = yield* readEventFeedPage({ after: 3, limit: 500, waitSeconds: 0 }, null); + assert.isTrue(empty._tag === "Page" && empty.events.length === 0 && empty.nextAfter === 3); + }).pipe(Effect.provide(engineLayer())), + ); + + it.effect("rejects a cursor ahead of the log", () => + Effect.gen(function* () { + yield* createProject("project-a"); + const result = yield* readEventFeedPage({ after: 9, limit: 500, waitSeconds: 0 }, null); + assert.deepStrictEqual(result, { _tag: "CursorInvalid", headSequence: 1 }); + }).pipe(Effect.provide(engineLayer())), + ); + + it.effect("holds an empty read until the next event is committed", () => + Effect.gen(function* () { + yield* createProject("project-a"); + const waiting = yield* Effect.forkChild( + readEventFeedPage({ after: 1, limit: 500, waitSeconds: 25 }, null), + ); + for (let tick = 0; tick < 20; tick += 1) yield* Effect.yieldNow; + assert.isUndefined(waiting.pollUnsafe(), "the read must be parked, not answered empty"); + yield* createProject("project-b"); + const page = yield* Fiber.join(waiting); + assert.isTrue(page._tag === "Page"); + if (page._tag !== "Page") return; + assert.deepStrictEqual(sequences(page.events), [2]); + }).pipe(Effect.provide(engineLayer())), + ); + + it.effect("returns an empty page when the wait elapses", () => + Effect.gen(function* () { + yield* createProject("project-a"); + const waiting = yield* Effect.forkChild( + readEventFeedPage({ after: 1, limit: 500, waitSeconds: 2 }, null), + ); + for (let tick = 0; tick < 50 && waiting.pollUnsafe() === undefined; tick += 1) { + yield* TestClock.adjust("1 second"); + } + const page = yield* Fiber.join(waiting); + assert.isTrue(page._tag === "Page" && page.events.length === 0 && page.nextAfter === 1); + }).pipe(Effect.provide(engineLayer())), + ); + + it.effect("drops events a team actor cannot see but still advances the cursor", () => + Effect.gen(function* () { + yield* createProject("project-a"); + yield* createProject("project-hidden"); + const page = yield* readEventFeedPage( + { after: 0, limit: 500, waitSeconds: 0 }, + UserId.make("user-team"), + ); + assert.isTrue(page._tag === "Page"); + if (page._tag !== "Page") return; + assert.deepStrictEqual(sequences(page.events), [1]); + assert.strictEqual(page.nextAfter, 2); + }).pipe(Effect.provide(engineLayer())), + ); +}); + +describe("GET /api/orchestration/events", () => { + const scopesByToken: Record> = { + reader: ["orchestration:read"], + "no-scope": [], + }; + const authLayer = Layer.succeed(EnvironmentAuth, { + authenticateHttpRequest: (request: { + readonly headers: Record; + }) => { + const token = request.headers.authorization?.replace(/^Bearer /, "") ?? ""; + const scopes = scopesByToken[token]; + return scopes === undefined + ? Effect.die(new Error("unexpected token")) + : Effect.succeed({ + sessionId: AuthSessionId.make(`session-${token}`), + userId: null, + subject: token, + method: "bearer-access-token", + scopes, + }); + }, + } as unknown as EnvironmentAuth["Service"]); + + const request = (query: string, token = "reader") => + Effect.gen(function* () { + const client = yield* HttpClient.HttpClient; + const response = yield* client.execute( + HttpClientRequest.get(`${EVENT_FEED_PATH}?${query}`).pipe( + HttpClientRequest.bearerToken(token), + ), + ); + return { status: response.status, body: (yield* response.json) as Record }; + }); + + it.effect("serves pages, rejects bad input, stale cursors and missing scope", () => + Effect.scoped( + Effect.gen(function* () { + yield* HttpRouter.serve(eventFeedRouteLayer, { + disableListenLog: true, + disableLogger: true, + }).pipe(Layer.build); + yield* createProject("project-a"); + yield* createProject("project-b"); + + const ok = yield* request("after=0&limit=1"); + assert.strictEqual(ok.status, 200); + assert.strictEqual(ok.body.nextAfter, 1); + assert.strictEqual(ok.body.headSequence, 2); + const events = ok.body.events as ReadonlyArray>; + assert.deepStrictEqual( + events.map((event) => [event.sequence, event.type, event.aggregateId]), + [[1, "project.created", "project-a"]], + ); + + assert.strictEqual((yield* request("limit=5")).status, 400); + const stale = yield* request("after=99"); + assert.deepStrictEqual(stale, { + status: 410, + body: { error: "cursor-invalid", headSequence: 2 }, + }); + assert.strictEqual((yield* request("after=0", "no-scope")).status, 403); + }), + ).pipe(Effect.provide(Layer.mergeAll(NodeHttpServer.layerTest, authLayer, engineLayer()))), + ); +}); diff --git a/apps/server/src/orchestration/eventFeedHttp.expbkt3.ts b/apps/server/src/orchestration/eventFeedHttp.expbkt3.ts new file mode 100644 index 0000000000000..67d18483a9a39 --- /dev/null +++ b/apps/server/src/orchestration/eventFeedHttp.expbkt3.ts @@ -0,0 +1,242 @@ +/** + * T3-CUSTOM(expbkt3): `GET /api/orchestration/events` — an ordered event feed + * for followers such as the Linear bridge. + * + * The bridge used to poll the full shell (~620 KB) every 2 s and full thread + * histories on top, about a quarter of the server's only event loop. A + * follower instead keeps a cursor and long-polls: + * + * GET /api/orchestration/events?after=&limit=<1..1000>&wait=<0..25> + * + * Events come from the sequence primary key in ascending order, shaped exactly + * as the WebSocket `replayEvents` RPC returns them (project repository identity + * attached, team-mode visibility applied, activity payloads projected). When + * nothing is newer than `after`, the request is held until the engine + * publishes a newer event or `wait` seconds pass; there is no polling loop. + * + * `410 cursor-invalid` means the cursor is ahead of the log (the database was + * replaced): re-read the shell and restart from its `snapshotSequence`. The + * event log is never pruned, so an old cursor is always served. + * + * Contract: ~/perf-audit-bkt3/80-api-contract.md, section 1. + */ +import { + AuthOrchestrationReadScope, + OrchestrationEvent, + ProjectId, + ThreadId, +} from "@t3tools/contracts"; +import * as Duration from "effect/Duration"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as Schema from "effect/Schema"; +import * as Stream from "effect/Stream"; +import { HttpRouter, HttpServerRequest, HttpServerResponse } from "effect/unstable/http"; + +import * as RepositoryIdentityResolver from "../project/RepositoryIdentityResolver.ts"; +import { projectActivityEvent } from "./ActivityPayloadProjection.ts"; +import { authenticateForkRoute, catchForkRouteAuthErrors } from "./forkRouteAuth.expbkt3.ts"; +import { OrchestrationAccessControl } from "./Services/AccessControl.ts"; +import { OrchestrationEngineService } from "./Services/OrchestrationEngine.ts"; +import { ProjectionSnapshotQuery } from "./Services/ProjectionSnapshotQuery.ts"; + +export const EVENT_FEED_PATH = "/api/orchestration/events"; +export const EVENT_FEED_DEFAULT_LIMIT = 500; +export const EVENT_FEED_MAX_LIMIT = 1000; +export const EVENT_FEED_MAX_WAIT_SECONDS = 25; + +export interface EventFeedQuery { + readonly after: number; + readonly limit: number; + readonly waitSeconds: number; +} + +const parseIntegerParam = ( + params: URLSearchParams, + name: string, + bounds: { readonly min: number; readonly max: number; readonly fallback?: number }, +): number | string => { + const raw = params.get(name); + if (raw === null || raw === "") { + return bounds.fallback ?? `${name} is required`; + } + if (!/^\d+$/.test(raw)) return `${name} must be a non-negative integer`; + const value = Number(raw); + if (!Number.isSafeInteger(value) || value < bounds.min || value > bounds.max) { + return `${name} must be between ${bounds.min} and ${bounds.max}`; + } + return value; +}; + +/** Parses the query string; a string result is the reason it was rejected. */ +export const parseEventFeedQuery = (params: URLSearchParams): EventFeedQuery | string => { + const after = parseIntegerParam(params, "after", { min: 0, max: Number.MAX_SAFE_INTEGER }); + if (typeof after === "string") return after; + const limit = parseIntegerParam(params, "limit", { + min: 1, + max: EVENT_FEED_MAX_LIMIT, + fallback: EVENT_FEED_DEFAULT_LIMIT, + }); + if (typeof limit === "string") return limit; + const waitSeconds = parseIntegerParam(params, "wait", { + min: 0, + max: EVENT_FEED_MAX_WAIT_SECONDS, + fallback: 0, + }); + if (typeof waitSeconds === "string") return waitSeconds; + return { after, limit, waitSeconds }; +}; + +const encodeEvents = Schema.encodeEffect(Schema.Array(OrchestrationEvent)); + +export type EventFeedResult = + | { + readonly _tag: "Page"; + readonly events: ReadonlyArray; + readonly nextAfter: number; + readonly headSequence: number; + } + | { readonly _tag: "CursorInvalid"; readonly headSequence: number }; + +/** + * Reads one page of the feed for `actorUserId` (null = unrestricted operator). + * `nextAfter` advances past events the actor may not see, so a filtered + * follower never re-reads them. + */ +export const readEventFeedPage = Effect.fn("orchestration.eventFeed.read")(function* ( + query: EventFeedQuery, + actorUserId: Parameters[0] | null, +) { + const engine = yield* OrchestrationEngineService; + const headBefore = yield* engine.latestSequence; + if (query.after > headBefore) { + return { _tag: "CursorInvalid", headSequence: headBefore } satisfies EventFeedResult; + } + + const readPage = Stream.runCollect(engine.readEvents(query.after, query.limit)).pipe( + Effect.map((events) => Array.from(events)), + ); + const raw = yield* Effect.scoped( + Effect.gen(function* () { + // Subscribe before reading, so an event committed between the read and + // the wait still wakes this request. + const live = yield* engine.subscribeDomainEvents; + const page = yield* readPage; + if (page.length > 0 || query.waitSeconds === 0) return page; + const woke = yield* live.pipe( + Stream.filter((event) => event.sequence > query.after), + Stream.runHead, + Effect.timeoutOption(Duration.seconds(query.waitSeconds)), + ); + return Option.isSome(woke) ? yield* readPage : page; + }), + ); + + const visible = yield* filterVisible(raw, actorUserId); + const events = yield* Effect.forEach(visible, attachRepositoryIdentity, { concurrency: 4 }); + return { + _tag: "Page", + events: events.map(projectActivityEvent), + nextAfter: raw.at(-1)?.sequence ?? query.after, + headSequence: Math.max(yield* engine.latestSequence, raw.at(-1)?.sequence ?? 0), + } satisfies EventFeedResult; +}); + +const filterVisible = Effect.fnUntraced(function* ( + events: ReadonlyArray, + actorUserId: Parameters[0] | null, +) { + if (actorUserId === null) return events; + const accessControl = yield* OrchestrationAccessControl; + const allowed = new Map(); + const visible: OrchestrationEvent[] = []; + for (const event of events) { + const key = `${event.aggregateKind}:${event.aggregateId}`; + let canSee = allowed.get(key); + if (canSee === undefined) { + canSee = yield* ( + event.aggregateKind === "thread" + ? accessControl.canAccessThread(actorUserId, ThreadId.make(event.aggregateId)) + : accessControl.canAccessProject(actorUserId, ProjectId.make(event.aggregateId)) + ).pipe(Effect.orElseSucceed(() => false)); + allowed.set(key, canSee); + } + if (canSee) visible.push(event); + } + return visible; +}); + +/** Mirrors the WebSocket replay: project events carry the resolved repository identity. */ +const attachRepositoryIdentity = ( + event: OrchestrationEvent, +): Effect.Effect< + OrchestrationEvent, + never, + RepositoryIdentityResolver.RepositoryIdentityResolver | ProjectionSnapshotQuery +> => + Effect.gen(function* () { + const resolver = yield* RepositoryIdentityResolver.RepositoryIdentityResolver; + if (event.type === "project.created") { + const repositoryIdentity = yield* resolver.resolve(event.payload.workspaceRoot); + return { ...event, payload: { ...event.payload, repositoryIdentity } }; + } + if (event.type === "project.meta-updated") { + const projectionSnapshotQuery = yield* ProjectionSnapshotQuery; + const workspaceRoot = + event.payload.workspaceRoot ?? + Option.match(yield* projectionSnapshotQuery.getProjectShellById(event.payload.projectId), { + onNone: () => null, + onSome: (project) => project.workspaceRoot, + }); + if (workspaceRoot === null) return event; + const repositoryIdentity = yield* resolver.resolve(workspaceRoot); + return { ...event, payload: { ...event.payload, repositoryIdentity } }; + } + return event; + }).pipe(Effect.orElseSucceed(() => event)); + +const jsonError = (status: number, body: Record) => + HttpServerResponse.jsonUnsafe(body, { status }); + +const eventFeedHandler = Effect.gen(function* () { + const session = yield* authenticateForkRoute(AuthOrchestrationReadScope); + const request = yield* HttpServerRequest.HttpServerRequest; + const url = HttpServerRequest.toURL(request); + const query = parseEventFeedQuery( + Option.isSome(url) ? url.value.searchParams : new URLSearchParams(), + ); + if (typeof query === "string") { + return jsonError(400, { error: "invalid-request", detail: query }); + } + const accessControl = yield* OrchestrationAccessControl; + const actorUserId = Option.getOrNull(accessControl.actorFor(session.subject, session.userId)); + const result = yield* readEventFeedPage(query, actorUserId); + if (result._tag === "CursorInvalid") { + return jsonError(410, { error: "cursor-invalid", headSequence: result.headSequence }); + } + return HttpServerResponse.jsonUnsafe({ + events: yield* encodeEvents(result.events), + nextAfter: result.nextAfter, + headSequence: result.headSequence, + }); +}).pipe( + catchForkRouteAuthErrors, + Effect.catch((error) => + Effect.logWarning("orchestration event feed failed", { error }).pipe( + Effect.as(jsonError(500, { error: "internal-error" })), + ), + ), +); + +/** The access-control service is captured once, when the route layer is built. */ +export const eventFeedRouteLayer = Layer.unwrap( + Effect.gen(function* () { + const accessControl = yield* OrchestrationAccessControl; + return HttpRouter.add( + "GET", + EVENT_FEED_PATH, + eventFeedHandler.pipe(Effect.provideService(OrchestrationAccessControl, accessControl)), + ); + }), +); diff --git a/apps/server/src/orchestration/forkRouteAuth.expbkt3.ts b/apps/server/src/orchestration/forkRouteAuth.expbkt3.ts new file mode 100644 index 0000000000000..d21168b6b1408 --- /dev/null +++ b/apps/server/src/orchestration/forkRouteAuth.expbkt3.ts @@ -0,0 +1,45 @@ +/** + * T3-CUSTOM(expbkt3): bearer auth for fork-owned raw HTTP routes. + * + * Same credential handling as upstream's `authenticateRawRouteWithScope` in + * `http.ts`, but it returns the authenticated session so a route can resolve + * the acting user. Failures are the upstream typed errors; routes turn them + * into responses with `catchForkRouteAuthErrors`. + */ +import type { AuthEnvironmentScope } from "@t3tools/contracts"; +import * as Effect from "effect/Effect"; +import { HttpServerRequest, HttpServerRespondable } from "effect/unstable/http"; + +import * as EnvironmentAuth from "../auth/EnvironmentAuth.ts"; +import { + failEnvironmentAuthInvalid, + failEnvironmentInternal, + failEnvironmentScopeRequired, +} from "../auth/http.ts"; + +export const authenticateForkRoute = (scope: AuthEnvironmentScope) => + Effect.gen(function* () { + const request = yield* HttpServerRequest.HttpServerRequest; + const serverAuth = yield* EnvironmentAuth.EnvironmentAuth; + const session = yield* serverAuth.authenticateHttpRequest(request).pipe( + Effect.catchIf(EnvironmentAuth.isServerAuthCredentialError, (error) => + failEnvironmentAuthInvalid( + EnvironmentAuth.serverAuthCredentialReason(error), + EnvironmentAuth.serverAuthDpopFailureReason(error), + ), + ), + Effect.catchIf(EnvironmentAuth.isServerAuthInternalError, (error) => + failEnvironmentInternal("internal_error", error), + ), + ); + if (!session.scopes.includes(scope)) { + return yield* failEnvironmentScopeRequired(scope); + } + return session; + }); + +export const catchForkRouteAuthErrors = Effect.catchTags({ + EnvironmentAuthInvalidError: HttpServerRespondable.toResponse, + EnvironmentInternalError: HttpServerRespondable.toResponse, + EnvironmentScopeRequiredError: HttpServerRespondable.toResponse, +}); diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index f45898d631b3c..2dd893dc3b48d 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -78,6 +78,8 @@ import * as PlanReviewDocuments from "./persistence/PlanReviewDocuments.ts"; import * as AgentUiRenders from "./persistence/AgentUiRenders.ts"; import { plannotatorProxyRouteLayer } from "./plannotator/http.ts"; // T3-CUSTOM(expbkt3): END +// T3-CUSTOM(expbkt3): event feed for followers such as the Linear bridge. +import { eventFeedRouteLayer } from "./orchestration/eventFeedHttp.expbkt3.ts"; import * as DeviceService from "./device/DeviceService.ts"; import { deviceHubProxyRouteLayer } from "./device/DeviceHubProxy.ts"; import * as PreviewManager from "./preview/Manager.ts"; @@ -779,6 +781,7 @@ const PlannotatorAndMcpRoutesLive = Layer.mergeAll( plannotatorProxyRouteLayer, mcpUpstreamProxyRouteLayer, McpHttpServer.layer, + eventFeedRouteLayer, ).pipe( // One registry instance authenticates both the native and upstream MCP // routes; separate instances would not recognize each other's run tokens.