|
| 1 | +import * as Deferred from "effect/Deferred"; |
1 | 2 | import * as Exit from "effect/Exit"; |
2 | 3 | import * as Layer from "effect/Layer"; |
| 4 | +import * as Logger from "effect/Logger"; |
3 | 5 | import * as Path from "effect/Path"; |
4 | 6 | import * as Effect from "effect/Effect"; |
| 7 | +import * as Queue from "effect/Queue"; |
5 | 8 | import * as Ref from "effect/Ref"; |
| 9 | +import * as References from "effect/References"; |
6 | 10 | import * as Scope from "effect/Scope"; |
7 | 11 | import { ChildProcess, ChildProcessSpawner } from "effect/process"; |
8 | 12 |
|
9 | 13 | import * as NodeServices from "@effect/platform-node/NodeServices"; |
10 | | -import { assert, it } from "@effect/vitest"; |
| 14 | +import { assert, expect, it } from "@effect/vitest"; |
11 | 15 |
|
12 | 16 | import * as CodexClient from "./client.ts"; |
| 17 | +import { makeInMemoryStdio } from "./_internal/stdio.ts"; |
13 | 18 |
|
14 | 19 | const mockPeerPath = Effect.map(Effect.service(Path.Path), (path) => |
15 | 20 | path.join(import.meta.dirname, "../test/fixtures/codex-app-server-mock-peer.ts"), |
@@ -155,4 +160,52 @@ it.layer(NodeServices.layer)("effect-codex-app-server client", (it) => { |
155 | 160 | assert.equal(initialized.userAgent, "mock-codex-app-server"); |
156 | 161 | }), |
157 | 162 | ); |
| 163 | + |
| 164 | + it.effect("warns once about a notification that fails to decode and keeps dispatching", () => |
| 165 | + Effect.gen(function* () { |
| 166 | + const secret = "codex-notification-secret-sentinel"; |
| 167 | + const { stdio, input } = yield* makeInMemoryStdio(); |
| 168 | + const warnings: Array<unknown> = []; |
| 169 | + const client = yield* CodexClient.make(stdio).pipe( |
| 170 | + Effect.provide( |
| 171 | + Logger.layer([ |
| 172 | + Logger.make(({ logLevel, fiber }) => { |
| 173 | + if (logLevel === "Warn") |
| 174 | + warnings.push(fiber.getRef(References.CurrentLogAnnotations)); |
| 175 | + }), |
| 176 | + ]), |
| 177 | + ), |
| 178 | + ); |
| 179 | + const delivered = yield* Deferred.make<unknown>(); |
| 180 | + yield* client.handleServerNotification("item/agentMessage/delta", (payload) => |
| 181 | + Deferred.succeed(delivered, payload).pipe(Effect.asVoid), |
| 182 | + ); |
| 183 | + |
| 184 | + const send = (params: unknown) => |
| 185 | + Queue.offer( |
| 186 | + input, |
| 187 | + new TextEncoder().encode( |
| 188 | + `${JSON.stringify({ method: "item/agentMessage/delta", params })}\n`, |
| 189 | + ), |
| 190 | + ); |
| 191 | + const malformed = { itemId: "item-1", threadId: "thread-1", turnId: secret }; |
| 192 | + yield* send(malformed); |
| 193 | + yield* send(malformed); |
| 194 | + yield* send({ delta: "ok", itemId: "item-1", threadId: "thread-1", turnId: "turn-1" }); |
| 195 | + |
| 196 | + assert.deepEqual(yield* Deferred.await(delivered), { |
| 197 | + delta: "ok", |
| 198 | + itemId: "item-1", |
| 199 | + threadId: "thread-1", |
| 200 | + turnId: "turn-1", |
| 201 | + }); |
| 202 | + assert.lengthOf(warnings, 1); |
| 203 | + expect(warnings[0]).toMatchObject({ |
| 204 | + operation: "decode-notification-payload", |
| 205 | + method: "item/agentMessage/delta", |
| 206 | + issueKinds: expect.arrayContaining(["MissingKey"]), |
| 207 | + }); |
| 208 | + assert.notInclude(JSON.stringify(warnings), secret); |
| 209 | + }), |
| 210 | + ); |
158 | 211 | }); |
0 commit comments