| author | |
| committer | |
| log | 8fc317939f5c8bbef4afbe309ae897abbc221631 |
| tree | 3f3cb167f3e18043a3239edc2585fd376d65a54c |
| parent | 16c69ba7ffe950aa054b84e8052b03abe0f7379a |
| signature | Signed by PGP key B5690EEEBB952194 |
Co-authored-by: codex <codex@users.noreply.github.com>5 files changed, 58 insertions(+), 169 deletions(-)
apps/server/src/provider/Layers/CodexProvider.ts+8-10| ... | @@ -5,6 +5,7 @@ import * as Layer from "effect/Layer"; | ... | @@ -5,6 +5,7 @@ import * as Layer from "effect/Layer"; |
| 5 | import * as Option from "effect/Option"; | 5 | import * as Option from "effect/Option"; |
| 6 | import * as Result from "effect/Result"; | 6 | import * as Result from "effect/Result"; |
| 7 | import * as Schema from "effect/Schema"; | 7 | import * as Schema from "effect/Schema"; |
| 8 | import * as Scope from "effect/Scope"; | ||
| 8 | import * as Types from "effect/Types"; | 9 | import * as Types from "effect/Types"; |
| 9 | import { ChildProcessSpawner } from "effect/unstable/process"; | 10 | import { ChildProcessSpawner } from "effect/unstable/process"; |
| 10 | import * as CodexClient from "effect-codex-app-server/client"; | 11 | import * as CodexClient from "effect-codex-app-server/client"; |
| ... | @@ -28,7 +29,6 @@ import { | ... | @@ -28,7 +29,6 @@ import { |
| 28 | type ServerProviderDraft, | 29 | type ServerProviderDraft, |
| 29 | } from "../providerSnapshot.ts"; | 30 | } from "../providerSnapshot.ts"; |
| 30 | import { expandHomePath } from "../../pathExpansion.ts"; | 31 | import { expandHomePath } from "../../pathExpansion.ts"; |
| 31 | import { scopedSafeTeardown } from "./scopedSafeTeardown.ts"; | ||
| 32 | import packageJson from "../../../package.json" with { type: "json" }; | 32 | import packageJson from "../../../package.json" with { type: "json" }; |
| 33 | const isCodexAppServerSpawnError = Schema.is(CodexErrors.CodexAppServerSpawnError); | 33 | const isCodexAppServerSpawnError = Schema.is(CodexErrors.CodexAppServerSpawnError); |
| 34 | 34 | ||
| ... | @@ -248,12 +248,6 @@ export function buildCodexInitializeParams(): CodexSchema.V1InitializeParams { | ... | @@ -248,12 +248,6 @@ export function buildCodexInitializeParams(): CodexSchema.V1InitializeParams { |
| 248 | }; | 248 | }; |
| 249 | } | 249 | } |
| 250 | 250 | ||
| 251 | // Wrapped with `scopedSafeTeardown("codex-probe")` rather than the usual | ||
| 252 | // `Effect.scoped` so that a defect from the `Layer.build` finalizer (e.g. | ||
| 253 | // `ChildProcess.kill` throwing because the `codex app-server` child exited | ||
| 254 | // early) cannot override a successful probe body. Without this guard the | ||
| 255 | // defect bubbles past `Effect.result` in `checkCodexProviderStatus`, dies | ||
| 256 | // `refreshOneSource`, and `providersRef` never receives the snapshot. | ||
| 257 | const probeCodexAppServerProvider = Effect.fn("probeCodexAppServerProvider")(function* (input: { | 251 | const probeCodexAppServerProvider = Effect.fn("probeCodexAppServerProvider")(function* (input: { |
| 258 | readonly binaryPath: string; | 252 | readonly binaryPath: string; |
| 259 | readonly homePath?: string; | 253 | readonly homePath?: string; |
| ... | @@ -323,7 +317,7 @@ const probeCodexAppServerProvider = Effect.fn("probeCodexAppServerProvider")(fun | ... | @@ -323,7 +317,7 @@ const probeCodexAppServerProvider = Effect.fn("probeCodexAppServerProvider")(fun |
| 323 | models: appendCustomCodexModels(models, input.customModels ?? []), | 317 | models: appendCustomCodexModels(models, input.customModels ?? []), |
| 324 | skills: parseCodexSkillsListResponse(skillsResponse, input.cwd), | 318 | skills: parseCodexSkillsListResponse(skillsResponse, input.cwd), |
| 325 | } satisfies CodexAppServerProviderSnapshot; | 319 | } satisfies CodexAppServerProviderSnapshot; |
| 326 | }, scopedSafeTeardown("codex-probe")); | 320 | }); |
| 327 | 321 | ||
| 328 | const emptyCodexModelsFromSettings = (codexSettings: CodexSettings): ServerProvider["models"] => | 322 | const emptyCodexModelsFromSettings = (codexSettings: CodexSettings): ServerProvider["models"] => |
| 329 | codexSettings.customModels | 323 | codexSettings.customModels |
| ... | @@ -416,7 +410,7 @@ export const checkCodexProviderStatus = Effect.fn("checkCodexProviderStatus")(fu | ... | @@ -416,7 +410,7 @@ export const checkCodexProviderStatus = Effect.fn("checkCodexProviderStatus")(fu |
| 416 | }) => Effect.Effect< | 410 | }) => Effect.Effect< |
| 417 | CodexAppServerProviderSnapshot, | 411 | CodexAppServerProviderSnapshot, |
| 418 | CodexErrors.CodexAppServerError, | 412 | CodexErrors.CodexAppServerError, |
| 419 | ChildProcessSpawner.ChildProcessSpawner | 413 | ChildProcessSpawner.ChildProcessSpawner | Scope.Scope |
| 420 | > = probeCodexAppServerProvider, | 414 | > = probeCodexAppServerProvider, |
| 421 | environment: NodeJS.ProcessEnv = process.env, | 415 | environment: NodeJS.ProcessEnv = process.env, |
| 422 | ): Effect.fn.Return< | 416 | ): Effect.fn.Return< |
| ... | @@ -450,7 +444,11 @@ export const checkCodexProviderStatus = Effect.fn("checkCodexProviderStatus")(fu | ... | @@ -450,7 +444,11 @@ export const checkCodexProviderStatus = Effect.fn("checkCodexProviderStatus")(fu |
| 450 | cwd: process.cwd(), | 444 | cwd: process.cwd(), |
| 451 | customModels: codexSettings.customModels, | 445 | customModels: codexSettings.customModels, |
| 452 | environment, | 446 | environment, |
| 453 | }).pipe(Effect.timeoutOption(Duration.millis(AUTH_PROBE_TIMEOUT_MS)), Effect.result); | 447 | }).pipe( |
| 448 | Effect.scoped, | ||
| 449 | Effect.timeoutOption(Duration.millis(AUTH_PROBE_TIMEOUT_MS)), | ||
| 450 | Effect.result, | ||
| 451 | ); | ||
| 454 | 452 | ||
| 455 | if (Result.isFailure(probeResult)) { | 453 | if (Result.isFailure(probeResult)) { |
| 456 | const error = probeResult.failure; | 454 | const error = probeResult.failure; |
apps/server/src/provider/Layers/CodexSessionRuntime.ts+2| ... | @@ -53,6 +53,7 @@ const BENIGN_ERROR_LOG_SNIPPETS = [ | ... | @@ -53,6 +53,7 @@ const BENIGN_ERROR_LOG_SNIPPETS = [ |
| 53 | "state db missing rollout path for thread", | 53 | "state db missing rollout path for thread", |
| 54 | "state db record_discrepancy: find_thread_path_by_id_str_in_subdir, falling_back", | 54 | "state db record_discrepancy: find_thread_path_by_id_str_in_subdir, falling_back", |
| 55 | ]; | 55 | ]; |
| 56 | const CODEX_APP_SERVER_FORCE_KILL_AFTER = "2 seconds" as const; | ||
| 56 | const RECOVERABLE_THREAD_RESUME_ERROR_SNIPPETS = [ | 57 | const RECOVERABLE_THREAD_RESUME_ERROR_SNIPPETS = [ |
| 57 | "not found", | 58 | "not found", |
| 58 | "missing thread", | 59 | "missing thread", |
| ... | @@ -721,6 +722,7 @@ export const makeCodexSessionRuntime = ( | ... | @@ -721,6 +722,7 @@ export const makeCodexSessionRuntime = ( |
| 721 | ChildProcess.make(options.binaryPath, ["app-server"], { | 722 | ChildProcess.make(options.binaryPath, ["app-server"], { |
| 722 | cwd: options.cwd, | 723 | cwd: options.cwd, |
| 723 | env, | 724 | env, |
| 725 | forceKillAfter: CODEX_APP_SERVER_FORCE_KILL_AFTER, | ||
| 724 | shell: process.platform === "win32", | 726 | shell: process.platform === "win32", |
| 725 | }), | 727 | }), |
| 726 | ) | 728 | ) |
apps/server/src/provider/Layers/ProviderRegistry.test.ts+48| ... | @@ -2,6 +2,7 @@ import * as NodeServices from "@effect/platform-node/NodeServices"; | ... | @@ -2,6 +2,7 @@ import * as NodeServices from "@effect/platform-node/NodeServices"; |
| 2 | import { describe, it, assert } from "@effect/vitest"; | 2 | import { describe, it, assert } from "@effect/vitest"; |
| 3 | import * as Effect from "effect/Effect"; | 3 | import * as Effect from "effect/Effect"; |
| 4 | import * as Exit from "effect/Exit"; | 4 | import * as Exit from "effect/Exit"; |
| 5 | import * as Fiber from "effect/Fiber"; | ||
| 5 | import * as Layer from "effect/Layer"; | 6 | import * as Layer from "effect/Layer"; |
| 6 | import * as PubSub from "effect/PubSub"; | 7 | import * as PubSub from "effect/PubSub"; |
| 7 | import * as Ref from "effect/Ref"; | 8 | import * as Ref from "effect/Ref"; |
| ... | @@ -209,6 +210,31 @@ function failingSpawnerLayer(description: string) { | ... | @@ -209,6 +210,31 @@ function failingSpawnerLayer(description: string) { |
| 209 | ); | 210 | ); |
| 210 | } | 211 | } |
| 211 | 212 | ||
| 213 | function hangingScopedSpawnerLayer(killCalls: Ref.Ref<number>) { | ||
| 214 | return Layer.succeed( | ||
| 215 | ChildProcessSpawner.ChildProcessSpawner, | ||
| 216 | ChildProcessSpawner.make(() => | ||
| 217 | Effect.gen(function* () { | ||
| 218 | const handle = ChildProcessSpawner.makeHandle({ | ||
| 219 | pid: ChildProcessSpawner.ProcessId(1), | ||
| 220 | exitCode: Effect.never, | ||
| 221 | isRunning: Effect.succeed(true), | ||
| 222 | kill: () => Ref.update(killCalls, (current) => current + 1), | ||
| 223 | unref: Effect.succeed(Effect.void), | ||
| 224 | stdin: Sink.drain, | ||
| 225 | stdout: Stream.never, | ||
| 226 | stderr: Stream.never, | ||
| 227 | all: Stream.never, | ||
| 228 | getInputFd: () => Sink.drain, | ||
| 229 | getOutputFd: () => Stream.empty, | ||
| 230 | }); | ||
| 231 | yield* Effect.addFinalizer(() => handle.kill().pipe(Effect.ignore)); | ||
| 232 | return handle; | ||
| 233 | }), | ||
| 234 | ), | ||
| 235 | ); | ||
| 236 | } | ||
| 237 | |||
| 212 | const codexModelCapabilities = createModelCapabilities({ | 238 | const codexModelCapabilities = createModelCapabilities({ |
| 213 | optionDescriptors: [ | 239 | optionDescriptors: [ |
| 214 | selectDescriptor("reasoningEffort", "Reasoning", [ | 240 | selectDescriptor("reasoningEffort", "Reasoning", [ |
| ... | @@ -421,6 +447,28 @@ it.layer(Layer.mergeAll(NodeServices.layer, ServerSettingsService.layerTest(), T | ... | @@ -421,6 +447,28 @@ it.layer(Layer.mergeAll(NodeServices.layer, ServerSettingsService.layerTest(), T |
| 421 | ); | 447 | ); |
| 422 | }), | 448 | }), |
| 423 | ); | 449 | ); |
| 450 | |||
| 451 | it.effect("closes the app-server probe scope when provider status times out", () => | ||
| 452 | Effect.gen(function* () { | ||
| 453 | const killCalls = yield* Ref.make(0); | ||
| 454 | const statusFiber = yield* checkCodexProviderStatus(defaultCodexSettings).pipe( | ||
| 455 | Effect.provide(hangingScopedSpawnerLayer(killCalls)), | ||
| 456 | Effect.forkChild, | ||
| 457 | ); | ||
| 458 | |||
| 459 | yield* Effect.yieldNow; | ||
| 460 | yield* TestClock.adjust("11 seconds"); | ||
| 461 | yield* Effect.yieldNow; | ||
| 462 | |||
| 463 | const status = yield* Fiber.join(statusFiber); | ||
| 464 | assert.strictEqual(status.status, "error"); | ||
| 465 | assert.strictEqual( | ||
| 466 | status.message, | ||
| 467 | "Timed out while checking Codex app-server provider status.", | ||
| 468 | ); | ||
| 469 | assert.strictEqual(yield* Ref.get(killCalls), 1); | ||
| 470 | }), | ||
| 471 | ); | ||
| 424 | }); | 472 | }); |
| 425 | 473 | ||
| 426 | describe("ProviderRegistryLive", () => { | 474 | describe("ProviderRegistryLive", () => { |
apps/server/src/provider/Layers/scopedSafeTeardown.test.ts deleted-96| ... | @@ -1,96 +0,0 @@ | ||
| 1 | import { it } from "@effect/vitest"; | ||
| 2 | import * as Cause from "effect/Cause"; | ||
| 3 | import * as Effect from "effect/Effect"; | ||
| 4 | import * as Exit from "effect/Exit"; | ||
| 5 | import { describe, expect } from "vitest"; | ||
| 6 | |||
| 7 | import { scopedSafeTeardown } from "./scopedSafeTeardown.ts"; | ||
| 8 | |||
| 9 | describe("scopedSafeTeardown", () => { | ||
| 10 | it.effect("returns the body's value when teardown is clean", () => | ||
| 11 | Effect.gen(function* () { | ||
| 12 | const finalizers: string[] = []; | ||
| 13 | const wrapped = Effect.gen(function* () { | ||
| 14 | yield* Effect.addFinalizer(() => | ||
| 15 | Effect.sync(() => { | ||
| 16 | finalizers.push("clean"); | ||
| 17 | }), | ||
| 18 | ); | ||
| 19 | return "body-ok"; | ||
| 20 | }).pipe(scopedSafeTeardown("test")); | ||
| 21 | |||
| 22 | const value = yield* wrapped; | ||
| 23 | expect(value).toBe("body-ok"); | ||
| 24 | expect(finalizers).toEqual(["clean"]); | ||
| 25 | }), | ||
| 26 | ); | ||
| 27 | |||
| 28 | it.effect("preserves body success when a finalizer dies", () => | ||
| 29 | // The production failure mode: `Layer.build(...)` registers a finalizer | ||
| 30 | // that kills a subprocess; if the kill fails, the defect would otherwise | ||
| 31 | // override a successful probe body. | ||
| 32 | Effect.gen(function* () { | ||
| 33 | const finalizers: string[] = []; | ||
| 34 | const wrapped = Effect.gen(function* () { | ||
| 35 | yield* Effect.addFinalizer(() => | ||
| 36 | Effect.sync(() => { | ||
| 37 | finalizers.push("ran-before-die"); | ||
| 38 | }), | ||
| 39 | ); | ||
| 40 | yield* Effect.addFinalizer(() => | ||
| 41 | Effect.die(new Error("simulated subprocess kill failure")), | ||
| 42 | ); | ||
| 43 | return "body-ok"; | ||
| 44 | }).pipe(scopedSafeTeardown("test")); | ||
| 45 | |||
| 46 | const value = yield* wrapped; | ||
| 47 | expect(value).toBe("body-ok"); | ||
| 48 | // The clean finalizer still ran; teardown defect was logged + swallowed. | ||
| 49 | expect(finalizers).toEqual(["ran-before-die"]); | ||
| 50 | }), | ||
| 51 | ); | ||
| 52 | |||
| 53 | it.effect("preserves typed body failures even when teardown is clean", () => | ||
| 54 | Effect.gen(function* () { | ||
| 55 | class BodyError { | ||
| 56 | readonly _tag = "BodyError" as const; | ||
| 57 | } | ||
| 58 | const wrapped = Effect.gen(function* () { | ||
| 59 | yield* Effect.addFinalizer(() => Effect.void); | ||
| 60 | return yield* Effect.fail(new BodyError()); | ||
| 61 | }).pipe(scopedSafeTeardown("test")); | ||
| 62 | |||
| 63 | const exit = yield* Effect.exit(wrapped); | ||
| 64 | expect(Exit.isFailure(exit)).toBe(true); | ||
| 65 | if (Exit.isFailure(exit)) { | ||
| 66 | // Body's typed failure should surface, not a defect. | ||
| 67 | const squashed = Cause.squash(exit.cause); | ||
| 68 | expect(squashed).toBeInstanceOf(BodyError); | ||
| 69 | } | ||
| 70 | }), | ||
| 71 | ); | ||
| 72 | |||
| 73 | it.effect("prefers the body's typed failure over a teardown defect", () => | ||
| 74 | // Even when both the body fails AND teardown defects, the body's typed | ||
| 75 | // failure is what callers see. This matters because `Effect.result` / | ||
| 76 | // `.pipe(Effect.exit)` in callers expects a typed Failure, not a Die. | ||
| 77 | Effect.gen(function* () { | ||
| 78 | class BodyError { | ||
| 79 | readonly _tag = "BodyError" as const; | ||
| 80 | } | ||
| 81 | const wrapped = Effect.gen(function* () { | ||
| 82 | yield* Effect.addFinalizer(() => | ||
| 83 | Effect.die(new Error("simulated subprocess kill failure")), | ||
| 84 | ); | ||
| 85 | return yield* Effect.fail(new BodyError()); | ||
| 86 | }).pipe(scopedSafeTeardown("test")); | ||
| 87 | |||
| 88 | const exit = yield* Effect.exit(wrapped); | ||
| 89 | expect(Exit.isFailure(exit)).toBe(true); | ||
| 90 | if (Exit.isFailure(exit)) { | ||
| 91 | const squashed = Cause.squash(exit.cause); | ||
| 92 | expect(squashed).toBeInstanceOf(BodyError); | ||
| 93 | } | ||
| 94 | }), | ||
| 95 | ); | ||
| 96 | }); | ||
apps/server/src/provider/Layers/scopedSafeTeardown.ts deleted-63| ... | @@ -1,63 +0,0 @@ | ||
| 1 | /** | ||
| 2 | * scopedSafeTeardown — run a scope-requiring effect so that finalizer | ||
| 3 | * failures during scope close cannot override the body's Exit. | ||
| 4 | * | ||
| 5 | * Motivation | ||
| 6 | * ---------- | ||
| 7 | * The obvious pattern is `body.pipe(Effect.scoped)`: provide a fresh | ||
| 8 | * Scope, run the body, close the scope with the body's Exit. If a | ||
| 9 | * finalizer (e.g. `ChildProcess.kill` from an effect-codex-app-server | ||
| 10 | * spawn) dies during that close, the combined Exit becomes the | ||
| 11 | * finalizer's defect — even when the body already succeeded. | ||
| 12 | * | ||
| 13 | * Concretely this bit us in the Codex provider probe: a successful | ||
| 14 | * `initialize` → `account/read` → `skills/list` → `model/list` | ||
| 15 | * round-trip produced a `CodexAppServerProviderSnapshot`, but the | ||
| 16 | * `Layer.build(CodexClient.layerCommand(...))` finalizer then failed to | ||
| 17 | * kill the `codex app-server` subprocess with a `PlatformError`. The | ||
| 18 | * defect bubbled past `Effect.result` in `checkCodexProviderStatus`, | ||
| 19 | * died `refreshOneSource`, and `providersRef` never saw the snapshot. | ||
| 20 | * | ||
| 21 | * Strategy | ||
| 22 | * -------- | ||
| 23 | * 1. Make a fresh scope manually. | ||
| 24 | * 2. Run the body against that scope, capturing its Exit via | ||
| 25 | * `Effect.exit`. | ||
| 26 | * 3. Close the scope, catching any cause (typed failure *or* defect) | ||
| 27 | * with a log. | ||
| 28 | * 4. Replay the captured Exit so typed body failures still surface and | ||
| 29 | * successes still return their value. | ||
| 30 | * | ||
| 31 | * The helper deliberately logs teardown causes at `Warning` level — | ||
| 32 | * silently swallowing them is dangerous because they usually indicate a | ||
| 33 | * real bug in a downstream Layer's finalizer. | ||
| 34 | * | ||
| 35 | * @module provider/Layers/scopedSafeTeardown | ||
| 36 | */ | ||
| 37 | import * as Effect from "effect/Effect"; | ||
| 38 | import * as Exit from "effect/Exit"; | ||
| 39 | import * as Scope from "effect/Scope"; | ||
| 40 | |||
| 41 | /** | ||
| 42 | * Run `effect` with a freshly made `Scope.Scope`, guaranteeing that | ||
| 43 | * teardown failures cannot override the body's Exit. | ||
| 44 | * | ||
| 45 | * Shape matches `Effect.scoped`: takes an effect whose env includes | ||
| 46 | * `Scope.Scope`, returns one whose env excludes it. | ||
| 47 | * | ||
| 48 | * @param label Short label for the warning log emitted when teardown | ||
| 49 | * fails. Use something like `"codex-probe"`. | ||
| 50 | */ | ||
| 51 | export const scopedSafeTeardown = | ||
| 52 | (label: string) => | ||
| 53 | <A, E, R>(effect: Effect.Effect<A, E, R>): Effect.Effect<A, E, Exclude<R, Scope.Scope>> => | ||
| 54 | Effect.gen(function* () { | ||
| 55 | const scope = yield* Scope.make(); | ||
| 56 | const bodyExit = yield* effect.pipe(Effect.provideService(Scope.Scope, scope), Effect.exit); | ||
| 57 | yield* Scope.close(scope, Exit.void).pipe( | ||
| 58 | Effect.catchCause((cause) => | ||
| 59 | Effect.logWarning(`${label} teardown errored; preserving body result`, cause), | ||
| 60 | ), | ||
| 61 | ); | ||
| 62 | return yield* bodyExit; | ||
| 63 | }) as Effect.Effect<A, E, Exclude<R, Scope.Scope>>; | ||