authorgravatar for julius0216@outlook.comJulius Marminge <julius0216@outlook.com> 2026-10-05 21:37:51-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-10-05 21:37:51-07:00
log327a29ef56e215c69a5db5d0b6f8df89f04f5d5b
tree09583e199430bab8a6eb40340597fef338744e5b
parenteef36a67c37fa74905fc7feaa836e5a81daa8685
signature Signed by PGP key B5690EEEBB952194

fix(server): one-shot initializers no longer race (#16260)

Co-authored-by: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

4 files changed, 277 insertions(+), 15 deletions(-)

apps/desktop/src/backend/DesktopServerExposure.test.ts+51
......@@ -2,7 +2,9 @@ import * as NodeFileSystem from "@effect/platform-node/NodeFileSystem";
22import * as NodeHttpClient from "@effect/platform-node/NodeHttpClient";
33import * as NodeServices from "@effect/platform-node/NodeServices";
44import { assert, describe, it } from "@effect/vitest";
5import * as Deferred from "effect/Deferred";
56import * as Effect from "effect/Effect";
7import * as Fiber from "effect/Fiber";
68import * as FileSystem from "effect/FileSystem";
79import * as Layer from "effect/Layer";
810import * as Sink from "effect/Sink";
......@@ -305,6 +307,55 @@ describe("DesktopServerExposure", () => {
305307 );
306308 });
307309
310 it.effect("keeps a Tailscale Serve change made while a mode change is saving", () =>
311 Effect.gen(function* () {
312 const modeWriteStarted = yield* Deferred.make<void>();
313 const releaseModeWrite = yield* Deferred.make<void>();
314 const settingsLayer = Layer.effect(
315 DesktopAppSettings.DesktopAppSettings,
316 Effect.gen(function* () {
317 const settings = yield* DesktopAppSettings.DesktopAppSettings;
318 return DesktopAppSettings.DesktopAppSettings.of({
319 ...settings,
320 // Hold the mode write the way a slow disk would.
321 setServerExposureMode: (mode) =>
322 Deferred.succeed(modeWriteStarted, undefined).pipe(
323 Effect.andThen(Deferred.await(releaseModeWrite)),
324 Effect.andThen(settings.setServerExposureMode(mode)),
325 ),
326 });
327 }),
328 ).pipe(Layer.provide(DesktopAppSettings.layerTest()));
329
330 return yield* withHarness(
331 lanNetworkInterfaces,
332 Effect.gen(function* () {
333 const serverExposure = yield* DesktopServerExposure.DesktopServerExposure;
334 yield* serverExposure.configureFromSettings({ port: 4173 });
335
336 const modeChange = yield* serverExposure
337 .setMode("network-accessible")
338 .pipe(Effect.forkChild({ startImmediately: true }));
339 yield* Deferred.await(modeWriteStarted);
340 const tailscaleChange = yield* serverExposure
341 .setTailscaleServeEnabled({ enabled: true, port: 8443 })
342 .pipe(Effect.forkChild({ startImmediately: true }));
343 yield* Deferred.succeed(releaseModeWrite, undefined);
344 yield* Fiber.join(modeChange);
345 yield* Fiber.join(tailscaleChange);
346
347 const state = yield* serverExposure.getState;
348 assert.equal(state.mode, "network-accessible");
349 assert.equal(state.tailscaleServeEnabled, true);
350 assert.equal(state.tailscaleServePort, 8443);
351 }),
352 {},
353 undefined,
354 settingsLayer,
355 );
356 }),
357 );
358
308359 it.effect("keeps LAN and Tailscale endpoints distinct when Tailscale is enumerated first", () =>
309360 withHarness(
310361 { ...tailnetNetworkInterfaces, ...lanNetworkInterfaces },
apps/desktop/src/backend/DesktopServerExposure.ts+7-1
......@@ -17,6 +17,7 @@ import * as Layer from "effect/Layer";
1717import * as Option from "effect/Option";
1818import * as Ref from "effect/Ref";
1919import * as Schema from "effect/Schema";
20import * as Semaphore from "effect/Semaphore";
2021import * as HttpClient from "effect/http/HttpClient";
2122import * as ChildProcessSpawner from "effect/process/ChildProcessSpawner";
2223
......@@ -419,6 +420,10 @@ export const make = Effect.gen(function* () {
419420 const httpClient = yield* HttpClient.HttpClient;
420421 const desktopSettings = yield* DesktopAppSettings.DesktopAppSettings;
421422 const stateRef = yield* Ref.make(initialRuntimeState());
423 // Each change reads the runtime state, persists settings, then writes the
424 // state back. Run them one at a time so a change never writes over another
425 // with values it read before that change persisted.
426 const changePermit = yield* Semaphore.make(1);
422427
423428 // Cache the `tailscale status` spawn for the TTL. On macOS, the Mac App
424429 // Store Tailscale CLI lives inside Tailscale's sandbox container, so each
......@@ -492,7 +497,7 @@ export const make = Effect.gen(function* () {
492497 state: toContractState(resolved.state),
493498 requiresRelaunch: change.changed || requiresBackendRelaunch(previous, resolved.state),
494499 };
495 });
500 }, changePermit.withPermit);
496501
497502 const setTailscaleServeEnabled = Effect.fn("desktop.serverExposure.setTailscaleServeEnabled")(
498503 function* (input: { readonly enabled: boolean; readonly port?: number }) {
......@@ -527,6 +532,7 @@ export const make = Effect.gen(function* () {
527532 requiresRelaunch: result.changed,
528533 };
529534 },
535 changePermit.withPermit,
530536 );
531537
532538 const getAdvertisedEndpoints = Effect.gen(function* () {
apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts+187
......@@ -1653,6 +1653,193 @@ function makeCodexReplayTranscript(input: {
16531653 };
16541654}
16551655
1656function withReplayRequestId(
1657 entry: CodexReplay.CodexAppServerReplayEntry,
1658 id: number,
1659): CodexReplay.CodexAppServerReplayEntry {
1660 return entry.type !== "runtime_exit" && Predicate.isObject(entry.frame)
1661 ? { ...entry, frame: { ...entry.frame, id } }
1662 : entry;
1663}
1664
1665describe("CodexAdapterV2 session initialize", () => {
1666 const openReplaySession = (
1667 transcript: CodexReplay.CodexAppServerReplayTranscript,
1668 beforeEmitInbound?: CodexReplay.CodexAppServerReplayDriver["beforeEmitInbound"],
1669 ) =>
1670 Effect.gen(function* () {
1671 const driver = yield* CodexReplay.makeReplayDriver(
1672 transcript,
1673 beforeEmitInbound === undefined ? {} : { beforeEmitInbound },
1674 );
1675 let initializeRequests = 0;
1676 const adapter = CodexAdapterV2.makeCodexAdapterV2({
1677 instanceId: CodexAdapterV2.CODEX_DEFAULT_INSTANCE_ID,
1678 settings: DEFAULT_CODEX_SETTINGS,
1679 environment: {},
1680 clientFactory: {
1681 open: (openInput) =>
1682 Layer.build(CodexReplay.layerReplayWithDriver(driver)).pipe(
1683 Effect.flatMap((context) =>
1684 Effect.service(CodexClient.CodexAppServerClient).pipe(Effect.provide(context)),
1685 ),
1686 Effect.map(
1687 (client) =>
1688 ({
1689 ...client,
1690 request: (method, params) =>
1691 Effect.sync(() => {
1692 if (method === "initialize") initializeRequests++;
1693 }).pipe(Effect.andThen(client.request(method, params))),
1694 }) satisfies CodexClient.CodexAppServerClient["Service"],
1695 ),
1696 Effect.mapError(
1697 (cause) =>
1698 new ProviderAdapterOpenSessionError({
1699 driver: CodexAdapterV2.CODEX_DRIVER_KIND,
1700 providerSessionId: openInput.providerSessionId,
1701 cause,
1702 }),
1703 ),
1704 ),
1705 },
1706 fileSystem: yield* FileSystem.FileSystem,
1707 idAllocator: yield* IdAllocator.IdAllocatorV2,
1708 serverConfig: yield* makeReplayServerConfig(transcript.scenario).pipe(Effect.orDie),
1709 });
1710 const runtime = yield* adapter.openSession({
1711 threadId: ThreadId.make(`thread-${transcript.scenario}`),
1712 providerSessionId: ProviderSessionId.make(`provider-session-${transcript.scenario}`),
1713 modelSelection: CODEX_TEST_MODEL_SELECTION,
1714 runtimePolicy: CODEX_TEST_RUNTIME_POLICY,
1715 });
1716 return {
1717 ensureThread: (threadId: string) =>
1718 runtime.ensureThread({
1719 threadId: ThreadId.make(threadId),
1720 modelSelection: CODEX_TEST_MODEL_SELECTION,
1721 runtimePolicy: CODEX_TEST_RUNTIME_POLICY,
1722 }),
1723 initializeRequests: () => initializeRequests,
1724 };
1725 });
1726
1727 const replayPreamble = (nativeThreadId: string) =>
1728 codexReplayPreamble({ nativeThreadId, nativeTurnId: "unused", prompt: "unused" });
1729
1730 it.effect("sends one initialize when two threads start on a fresh session at once", () =>
1731 Effect.gen(function* () {
1732 const initializeAwaitingResponse = yield* Deferred.make<void>();
1733 const releaseInitialize = yield* Deferred.make<void>();
1734 // The transcript allows exactly one handshake: a second `initialize`
1735 // frame fails the replay, as Codex rejects it with "Already initialized".
1736 const session = yield* openReplaySession(
1737 makeCodexReplayTranscript({
1738 scenario: "concurrent-initialize",
1739 entries: [
1740 ...replayPreamble("concurrent-first").slice(0, 5),
1741 ...replayPreamble("concurrent-second")
1742 .slice(3, 5)
1743 .map((entry) => withReplayRequestId(entry, 3)),
1744 ],
1745 }),
1746 (entry) =>
1747 entry.label === "initialize"
1748 ? Deferred.succeed(initializeAwaitingResponse, undefined).pipe(
1749 Effect.andThen(Deferred.await(releaseInitialize)),
1750 )
1751 : Effect.void,
1752 );
1753
1754 const first = yield* session
1755 .ensureThread("thread-concurrent-first")
1756 .pipe(Effect.forkChild({ startImmediately: true }));
1757 yield* Deferred.await(initializeAwaitingResponse);
1758 // The second thread arrives while the handshake is still unanswered.
1759 const second = yield* session
1760 .ensureThread("thread-concurrent-second")
1761 .pipe(Effect.forkChild({ startImmediately: true }));
1762 yield* Deferred.succeed(releaseInitialize, undefined);
1763
1764 const providerThreads = [yield* Fiber.join(first), yield* Fiber.join(second)];
1765 assert.equal(session.initializeRequests(), 1);
1766 assert.sameMembers(
1767 providerThreads.map((providerThread) => providerThread.nativeThreadRef?.nativeId),
1768 ["concurrent-first", "concurrent-second"],
1769 );
1770 }).pipe(Effect.scoped, Effect.provide(Layer.merge(IdAllocator.layer, NodeServices.layer))),
1771 );
1772
1773 it.effect("retries initialize after a failed handshake", () =>
1774 Effect.gen(function* () {
1775 const preamble = replayPreamble("initialize-retry");
1776 const entries: Array<CodexReplay.CodexAppServerReplayEntry> = [
1777 ...preamble.slice(0, 1),
1778 {
1779 type: "emit_inbound",
1780 label: "initialize",
1781 frame: { id: 1, error: { code: -32603, message: "Codex is not ready." } },
1782 },
1783 ...preamble.slice(0, 2).map((entry) => withReplayRequestId(entry, 2)),
1784 ...preamble.slice(2, 3),
1785 ...preamble.slice(3, 5).map((entry) => withReplayRequestId(entry, 3)),
1786 ];
1787 const session = yield* openReplaySession(
1788 makeCodexReplayTranscript({ scenario: "initialize-retry", entries }),
1789 );
1790
1791 const failure = yield* session.ensureThread("thread-initialize-retry").pipe(Effect.flip);
1792 assert.equal(failure._tag, "ProviderAdapterEnsureThreadError");
1793 const providerThread = yield* session.ensureThread("thread-initialize-retry");
1794 assert.equal(providerThread.nativeThreadRef?.nativeId, "initialize-retry");
1795 assert.equal(session.initializeRequests(), 2);
1796 }).pipe(Effect.scoped, Effect.provide(Layer.merge(IdAllocator.layer, NodeServices.layer))),
1797 );
1798
1799 it.effect("completes the handshake after a caller is interrupted mid-initialize", () =>
1800 Effect.gen(function* () {
1801 const initializeAwaitingResponse = yield* Deferred.make<void>();
1802 const releaseInitialize = yield* Deferred.make<void>();
1803 const preamble = replayPreamble("initialize-interrupted");
1804 const session = yield* openReplaySession(
1805 makeCodexReplayTranscript({
1806 scenario: "initialize-interrupted",
1807 entries: [
1808 ...preamble.slice(0, 2),
1809 // Codex handled the interrupted caller's `initialize`, so it
1810 // rejects the next one.
1811 ...preamble.slice(0, 1).map((entry) => withReplayRequestId(entry, 2)),
1812 {
1813 type: "emit_inbound",
1814 label: "initialize-rejected",
1815 frame: { id: 2, error: { code: -32600, message: "Already initialized" } },
1816 },
1817 ...preamble.slice(2, 3),
1818 ...preamble.slice(3, 5).map((entry) => withReplayRequestId(entry, 3)),
1819 ],
1820 }),
1821 (entry) =>
1822 entry.label === "initialize"
1823 ? Deferred.succeed(initializeAwaitingResponse, undefined).pipe(
1824 Effect.andThen(Deferred.await(releaseInitialize)),
1825 )
1826 : Effect.void,
1827 );
1828
1829 const interrupted = yield* session
1830 .ensureThread("thread-initialize-interrupted")
1831 .pipe(Effect.forkChild({ startImmediately: true }));
1832 yield* Deferred.await(initializeAwaitingResponse);
1833 yield* Fiber.interrupt(interrupted);
1834 yield* Deferred.succeed(releaseInitialize, undefined);
1835
1836 const providerThread = yield* session.ensureThread("thread-initialize-interrupted");
1837 assert.equal(providerThread.nativeThreadRef?.nativeId, "initialize-interrupted");
1838 assert.equal(session.initializeRequests(), 2);
1839 }).pipe(Effect.scoped, Effect.provide(Layer.merge(IdAllocator.layer, NodeServices.layer))),
1840 );
1841});
1842
16561843describe("CodexAdapterV2 post-settle continuation", () => {
16571844 const awaitUntil = (predicate: () => boolean, label: string): Effect.Effect<void> =>
16581845 Effect.gen(function* () {
apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts+32-14
......@@ -1684,21 +1684,39 @@ export function makeCodexAdapterV2(adapterOptions: CodexAdapterV2Options): Provi
16841684 ),
16851685 );
16861686 const initialized = yield* Ref.make(false);
1687 const ensureInitialized = Effect.gen(function* () {
1688 const alreadyInitialized = yield* Ref.get(initialized);
1689 if (alreadyInitialized) {
1690 return;
1691 }
1687 // Threads share this app-server, and Codex rejects a second
1688 // `initialize`. Callers wait for an in-flight handshake instead of
1689 // starting their own; a failed handshake leaves the flag unset so the
1690 // next caller retries.
1691 const initializePermit = yield* Semaphore.make(1);
1692 const ensureInitialized = initializePermit.withPermit(
1693 Effect.gen(function* () {
1694 const alreadyInitialized = yield* Ref.get(initialized);
1695 if (alreadyInitialized) {
1696 return;
1697 }
16921698
1693 yield* client.request("initialize", {
1694 // Codex uses the client name as the request originator, so sessions
1695 // identify themselves exactly like the provider probe.
1696 clientInfo: buildCodexInitializeParams().clientInfo,
1697 capabilities: CODEX_CLIENT_CAPABILITIES,
1698 });
1699 yield* client.notify("initialized", undefined);
1700 yield* Ref.set(initialized, true);
1701 });
1699 yield* client
1700 .request("initialize", {
1701 // Codex uses the client name as the request originator, so sessions
1702 // identify themselves exactly like the provider probe.
1703 clientInfo: buildCodexInitializeParams().clientInfo,
1704 capabilities: CODEX_CLIENT_CAPABILITIES,
1705 })
1706 .pipe(
1707 Effect.catchTags({
1708 // A caller interrupted after its `initialize` reached Codex
1709 // leaves the app-server initialized but the flag unset.
1710 CodexAppServerRequestError: (error) =>
1711 error.code === -32600 && error.errorMessage === "Already initialized"
1712 ? Effect.void
1713 : Effect.fail(error),
1714 }),
1715 );
1716 yield* client.notify("initialized", undefined);
1717 yield* Ref.set(initialized, true);
1718 }),
1719 );
17021720 const now = yield* DateTime.now;
17031721 const session = providerSession({
17041722 providerSessionId: input.providerSessionId,