From 57a222de2bf0876e6b6ef1e626cbbfdfa7967776 Mon Sep 17 00:00:00 2001 From: Kit Langton Date: Sat, 11 Jul 2026 22:24:57 -0400 Subject: [PATCH] refactor(server): narrow event feed scope --- packages/core/src/event-logger.ts | 23 +++++---- packages/server/script/bench-event-feed.ts | 59 ---------------------- packages/server/test/event-feed.test.ts | 8 +++ specs/v2/event-stream-architecture.md | 20 ++------ 4 files changed, 23 insertions(+), 87 deletions(-) delete mode 100644 packages/server/script/bench-event-feed.ts diff --git a/packages/core/src/event-logger.ts b/packages/core/src/event-logger.ts index c9942a8100..f07c96320e 100644 --- a/packages/core/src/event-logger.ts +++ b/packages/core/src/event-logger.ts @@ -1,22 +1,23 @@ export * as EventLogger from "./event-logger" -import { Effect, Layer, Stream } from "effect" -import { Agent } from "@opencode-ai/schema/agent" -import { Catalog } from "@opencode-ai/schema/catalog" -import { Command } from "@opencode-ai/schema/command" -import { Config } from "@opencode-ai/schema/config" +import { Effect, Layer } from "effect" import { makeGlobalNode } from "./effect/app-node" import { EventV2 } from "./event" +const Types = new Set([ + "agent.updated", + "catalog.updated", + "command.updated", + "config.updated", +]) + export const layer = Layer.effectDiscard( Effect.gen(function* () { const events = yield* EventV2.Service - yield* events - .subscribe([Agent.Event.Updated, Catalog.Event.Updated, Command.Event.Updated, Config.Event.Updated]) - .pipe( - Stream.runForEach((event) => Effect.logInfo("event", { event })), - Effect.forkScoped({ startImmediately: true }), - ) + const unsubscribe = yield* events.listen((event) => + Types.has(event.type) ? Effect.logInfo("event", { event }) : Effect.void, + ) + yield* Effect.addFinalizer(() => unsubscribe) }), ) diff --git a/packages/server/script/bench-event-feed.ts b/packages/server/script/bench-event-feed.ts deleted file mode 100644 index f2d24bd2fd..0000000000 --- a/packages/server/script/bench-event-feed.ts +++ /dev/null @@ -1,59 +0,0 @@ -import { OpenCodeEvent } from "@opencode-ai/protocol/groups/event" -import { EventV2 } from "@opencode-ai/core/event" -import { AbsolutePath } from "@opencode-ai/core/schema" -import { DateTime } from "effect" -import { EventFeed } from "../src/event-feed" - -const clients = Number(process.argv[2] ?? 10) -const events = Number(process.argv[3] ?? 1_000) -const mode = process.argv[4] ?? "current" -const runs = 9 -const event: OpenCodeEvent = { - id: EventV2.ID.make("evt_benchmark"), - created: DateTime.makeUnsafe(Date.now()), - type: "mcp.status.changed", - location: { directory: AbsolutePath.make("/tmp/opencode-benchmark") }, - metadata: { - output: "x".repeat(8_192), - nested: Array.from({ length: 32 }, (_, index) => ({ index, value: `value-${index}` })), - }, - data: { server: "benchmark" }, -} - -function current() { - let bytes = 0 - for (let index = 0; index < events; index++) { - for (let client = 0; client < clients; client++) bytes += EventFeed.frame(event).length - } - return bytes -} - -function shared() { - let bytes = 0 - for (let index = 0; index < events; index++) { - const encoded = EventFeed.frame(event) - for (let client = 0; client < clients; client++) bytes += encoded.length - } - return bytes -} - -const benchmark = mode === "shared" ? shared : current -benchmark() - -const samples = Array.from({ length: runs }, () => { - Bun.gc(true) - const start = performance.now() - const bytes = benchmark() - return { duration: performance.now() - start, bytes } -}) -const durations = samples.map((sample) => sample.duration).toSorted((a, b) => a - b) -const median = durations[Math.floor(durations.length / 2)] -const absoluteDeviations = durations.map((duration) => Math.abs(duration - median)).toSorted((a, b) => a - b) -const mad = absoluteDeviations[Math.floor(absoluteDeviations.length / 2)] - -console.log(`mode=${mode} clients=${clients} events=${events} runs=${runs}`) -console.log( - `median=${median.toFixed(3)}ms mad=${mad.toFixed(3)}ms best=${durations[0].toFixed(3)}ms worst=${durations[durations.length - 1].toFixed(3)}ms`, -) -console.log(`METRIC event_feed_${mode}_ms=${median.toFixed(3)}`) -console.log(`METRIC event_feed_${mode}_mad_ms=${mad.toFixed(3)}`) diff --git a/packages/server/test/event-feed.test.ts b/packages/server/test/event-feed.test.ts index 03b626ad6f..1ec271a7cc 100644 --- a/packages/server/test/event-feed.test.ts +++ b/packages/server/test/event-feed.test.ts @@ -1,6 +1,7 @@ import { describe, expect, test } from "bun:test" import { AgentV2 } from "@opencode-ai/core/agent" import { EventV2 } from "@opencode-ai/core/event" +import { OpenCodeEvent } from "@opencode-ai/protocol/groups/event" import { DateTime, Deferred, Effect, Exit, Fiber, Option, Schema, Stream } from "effect" import { EventFeed } from "../src/event-feed" @@ -35,6 +36,13 @@ function makeSource() { } describe("EventFeed", () => { + test("preserves the public SSE frame encoding", () => { + const payload = event("wire") + expect(EventFeed.frame(payload)).toBe( + `data: ${JSON.stringify(Schema.encodeUnknownSync(OpenCodeEvent)(payload))}\n\n`, + ) + }) + test("encodes once and delivers the same frame to every subscriber", async () => { let encodes = 0 const result = await Effect.runPromise( diff --git a/specs/v2/event-stream-architecture.md b/specs/v2/event-stream-architecture.md index 5ba47d8f23..74a40df4fb 100644 --- a/specs/v2/event-stream-architecture.md +++ b/specs/v2/event-stream-architecture.md @@ -5,7 +5,7 @@ The public HTTP event stream uses one Server-scoped encoded feed with one independently bounded queue per connection. ```text -Core EventV2.subscribe() +Core EventV2.listen() | | one global subscription v @@ -165,10 +165,6 @@ If one accepted public event cannot be encoded: Keeping current clients connected would create a silent gap. Permanently terminating the feed would poison future connections. -### Source termination - -Unexpected Core source termination logs the cause and closes current subscribers. Normal Server-scope interruption is not logged as an error. - ## HTTP And Code Generation Protocol remains unchanged: @@ -196,23 +192,13 @@ Because method, path, schema, and wire representation do not change: ## Core Cleanup -The Server no longer uses `EventV2.liveBounded`, so Core removes that helper and its transport-specific overflow error. - -`EventLogger` moves from callback `listen` to a scoped `events.subscribe()` consumer. - -The deprecated `listen` interface remains temporarily for V1 compatibility and tests. Removing the callback fan-out completely requires migrating those callers separately and must not be conflated with this Server transport optimization. +The Server no longer uses `EventV2.liveBounded`, so Core removes that dead helper and its transport-specific overflow error. The feed registers one observer through the existing `listen` interface; other listeners are unchanged. Transactional projector registration is unrelated and remains unchanged. ## Benchmark -The disposable benchmark reproduces the current per-connection schema/JSON/SSE encoding path with a representative 8 KiB public event. It uses one warmup and nine measured runs; median is the primary metric and median absolute deviation is reported. - -Command from `packages/server`: - -```sh -bun run script/bench-event-feed.ts 1000 -``` +The disposable benchmark reproduced the previous per-connection schema/JSON/SSE encoding path with a representative 8 KiB public event. It used one warmup and nine measured runs; median was the primary metric and median absolute deviation was reported. The benchmark was intentionally not committed because it isolated the removed encoding boundary rather than exercising the complete HTTP stack. Results on Apple Silicon with Bun 1.3.14: