From 5a755ada28fa8da9d435f2e1f14697e83f4aa6db Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Wed, 25 Feb 2026 18:01:07 -0800 Subject: [PATCH] nits --- apps/server/src/wsServer.ts | 57 ++++++++++++++++-------------- apps/web/src/session-logic.test.ts | 4 +-- apps/web/src/session-logic.ts | 1 + 3 files changed, 33 insertions(+), 29 deletions(-) diff --git a/apps/server/src/wsServer.ts b/apps/server/src/wsServer.ts index 4182551518fc6ae95c12be16219e904ac79ef93d..2afceede57daaa793dd0e99ad5b722df85e128f2 100644 --- a/apps/server/src/wsServer.ts +++ b/apps/server/src/wsServer.ts @@ -183,23 +183,22 @@ export const createServer = Effect.fn(function* (options: ServerOptions = {}): E }); } - const onTerminalEvent = (event: TerminalEvent) => - Effect.gen(function* () { - const push: WsPush = { - type: "push", - channel: WS_CHANNELS.terminalEvent, - data: event, - }; - const message = yield* Schema.encodeEffect(Schema.fromJsonString(WsPush))(push); - let recipients = 0; - for (const client of yield* Ref.get(clients)) { - if (client.readyState === client.OPEN) { - client.send(message); - recipients += 1; - } + const onTerminalEvent = Effect.fnUntraced(function* (event: TerminalEvent) { + const push: WsPush = { + type: "push", + channel: WS_CHANNELS.terminalEvent, + data: event, + }; + const message = yield* Schema.encodeEffect(Schema.fromJsonString(WsPush))(push); + let recipients = 0; + for (const client of yield* Ref.get(clients)) { + if (client.readyState === client.OPEN) { + client.send(message); + recipients += 1; } - logOutgoingPush(push, recipients); - }).pipe(Effect.runPromise); + } + logOutgoingPush(push, recipients); + }); // HTTP server — serves static files or redirects to Vite dev server const httpServer = http.createServer((req, res) => { @@ -289,9 +288,7 @@ export const createServer = Effect.fn(function* (options: ServerOptions = {}): E const { openInEditor } = yield* Open; const subscriptionsScope = yield* Scope.make("sequential"); - yield* Effect.addFinalizer(() => - Scope.close(subscriptionsScope, Exit.void).pipe(Effect.orElseSucceed(() => undefined)), - ); + yield* Effect.addFinalizer(() => Scope.close(subscriptionsScope, Exit.void)); yield* Stream.runForEach(orchestrationEngine.streamDomainEvents, (event) => Effect.gen(function* () { @@ -312,7 +309,7 @@ export const createServer = Effect.fn(function* (options: ServerOptions = {}): E }), ).pipe(Effect.forkIn(subscriptionsScope)); - yield* orchestrationReactor.start.pipe(Scope.provide(subscriptionsScope)); + yield* Scope.provide(orchestrationReactor.start, subscriptionsScope); if (autoBootstrapProjectFromCwd) { const snapshot = yield* projectionReadModelQuery.getSnapshot(); @@ -332,7 +329,7 @@ export const createServer = Effect.fn(function* (options: ServerOptions = {}): E } } - const routeRequest = Effect.fn(function* (request: WebSocketRequest) { + const routeRequest = Effect.fnUntraced(function* (request: WebSocketRequest) { switch (request.body._tag) { case ORCHESTRATION_WS_METHODS.getSnapshot: return yield* projectionReadModelQuery.getSnapshot(); @@ -573,18 +570,24 @@ export const createServer = Effect.fn(function* (options: ServerOptions = {}): E ), ); - const unsubscribeTerminalEvents = yield* terminalManager.subscribe(onTerminalEvent); + const unsubscribeTerminalEvents = yield* terminalManager.subscribe( + (event) => void Effect.runPromise(onTerminalEvent(event)), + ); yield* Effect.addFinalizer(() => - Effect.sync(() => { - unsubscribeTerminalEvents(); - }), + Effect.all([Effect.sync(() => unsubscribeTerminalEvents()), terminalManager.dispose()]), ); - yield* Effect.addFinalizer(terminalManager.dispose); yield* NodeHttpServer.make(() => httpServer, listenOptions); yield* Effect.addFinalizer(() => - Effect.all([closeAllClients, closeWebSocketServer.pipe(Effect.orElseSucceed(() => undefined))]), + Effect.all([ + closeAllClients, + closeWebSocketServer.pipe( + Effect.catch((error) => + Effect.logWarning("failed to close web socket server", { cause: error }), + ), + ), + ]), ); return httpServer; diff --git a/apps/web/src/session-logic.test.ts b/apps/web/src/session-logic.test.ts index 3ecee7d63d0cf92a6b283ef0514c1f29a61ea813..3377268d2a35783e49d5a38b4fd1a4165f0366e8 100644 --- a/apps/web/src/session-logic.test.ts +++ b/apps/web/src/session-logic.test.ts @@ -69,7 +69,7 @@ describe("derivePendingApprovals", () => { }); describe("deriveWorkLogEntries", () => { - it("returns entries in chronological order", () => { + it("omits tool started entries and keeps completed entries", () => { const activities: OrchestrationThreadActivity[] = [ makeActivity({ id: "tool-complete", @@ -86,7 +86,7 @@ describe("deriveWorkLogEntries", () => { ]; const entries = deriveWorkLogEntries(activities, undefined); - expect(entries.map((entry) => entry.id)).toEqual(["tool-start", "tool-complete"]); + expect(entries.map((entry) => entry.id)).toEqual(["tool-complete"]); }); it("filters by turn id when provided", () => { diff --git a/apps/web/src/session-logic.ts b/apps/web/src/session-logic.ts index 9b24a83bbb5571e405e585f6739d6a69147bcd1d..114c2272e8ded1362ae5fb79f68c36ef8c20d310 100644 --- a/apps/web/src/session-logic.ts +++ b/apps/web/src/session-logic.ts @@ -127,6 +127,7 @@ export function deriveWorkLogEntries( ); return ordered .filter((activity) => (latestTurnId ? activity.turnId === latestTurnId : true)) + .filter((activity) => activity.kind !== "tool.started") .filter((activity) => activity.summary !== "Checkpoint captured") .map((activity) => { const payload = -- 2.54.0