diff --git a/apps/server/src/policy/ManagedPolicy.test.ts b/apps/server/src/policy/ManagedPolicy.test.ts index d6c7710b57ead..bd743d4e0de1a 100644 --- a/apps/server/src/policy/ManagedPolicy.test.ts +++ b/apps/server/src/policy/ManagedPolicy.test.ts @@ -193,7 +193,15 @@ it.layer(NodeServices.layer)("ManagedPolicy.load", (it) => { ]); // The host variable that replaces Codex's launch arguments is ignored too. const host = { PATH: "/usr/bin", T3CODE_CODEX_LAUNCH_ARGS: "-c mcp_servers.x.command=sh" }; - assert.deepEqual(providerHostEnvironment(policy.summary, host), { PATH: "/usr/bin" }); + const view = providerHostEnvironment(policy.summary, host); + assert.deepEqual({ ...view }, { PATH: "/usr/bin" }); + host.PATH = "/late/bin"; + Object.assign(host, { NEW_VARIABLE: "late", t3code_codex_launch_args: "hidden" }); + assert.deepEqual({ ...view }, { PATH: "/late/bin", NEW_VARIABLE: "late" }); + assert.isUndefined(view.T3CODE_CODEX_LAUNCH_ARGS); + assert.isFalse("t3code_codex_launch_args" in view); + assert.isFalse(Reflect.set(view, "PATH", "/replaced")); + assert.equal(host.PATH, "/late/bin"); assert.strictEqual(providerHostEnvironment(undefined, host), host); }), ); diff --git a/apps/server/src/policy/managedProviders.ts b/apps/server/src/policy/managedProviders.ts index 059f6aeb4b782..91fde8a56b63b 100644 --- a/apps/server/src/policy/managedProviders.ts +++ b/apps/server/src/policy/managedProviders.ts @@ -162,13 +162,29 @@ export const providerHostEnvironment = ( environment: NodeJS.ProcessEnv, ): NodeJS.ProcessEnv => { if (summary === undefined) return environment; - const next = { ...environment }; - for (const name of Object.keys(next)) { - if (LAUNCH_OVERRIDE_VARIABLES.some((override) => override === name.toUpperCase())) { - delete next[name]; - } - } - return next; + const allowed = (name: string | symbol) => + typeof name !== "string" || + !LAUNCH_OVERRIDE_VARIABLES.some((override) => override === name.toUpperCase()); + return new Proxy( + {}, + { + get: (_, name) => (allowed(name) ? Reflect.get(environment, name) : undefined), + has: (_, name) => allowed(name) && Reflect.has(environment, name), + ownKeys: () => Reflect.ownKeys(environment).filter(allowed), + getOwnPropertyDescriptor: (_, name) => + allowed(name) && Reflect.has(environment, name) + ? { + configurable: true, + enumerable: true, + value: Reflect.get(environment, name), + writable: false, + } + : undefined, + set: () => false, + deleteProperty: () => false, + defineProperty: () => false, + }, + ); }; /** Each driver's home variable, and the settings that rewrite it at launch. */ diff --git a/apps/server/src/provider/CodexProvider.ts b/apps/server/src/provider/CodexProvider.ts index 2e56eede84cd6..c2f2a43a7ee9b 100644 --- a/apps/server/src/provider/CodexProvider.ts +++ b/apps/server/src/provider/CodexProvider.ts @@ -736,6 +736,15 @@ function accountProbeStatus({ account, rateLimits }: CodexAppServerProviderSnaps return { status: "ready", auth }; } +const CODEX_PROBE_FAILED_PREFIX = "Codex app-server provider probe failed: "; + +/** The CLI spawned but neither the probe nor `codex --version` ran, e.g. its launcher lacks node. */ +export const isCodexLaunchFailure = (provider: ServerProvider): boolean => + provider.installed && + provider.status === "error" && + provider.version === null && + provider.message?.startsWith(CODEX_PROBE_FAILED_PREFIX) === true; + export const checkCodexProviderStatus = Effect.fn("checkCodexProviderStatus")(function* ( codexSettings: CodexSettings, probe: (input: { @@ -815,7 +824,7 @@ export const checkCodexProviderStatus = Effect.fn("checkCodexProviderStatus")(fu status: "error", auth: { status: "unknown" }, message: installed - ? `Codex app-server provider probe failed: ${error.message}.` + ? `${CODEX_PROBE_FAILED_PREFIX}${error.message}.` : `Could not start Codex CLI (\`${codexSettings.binaryPath}\`). Check Settings → Providers → Codex → Binary path on the server.` + (codexSettings.binaryPath === "codex" ? " Installing ChatGPT or Codex desktop may not add codex to PATH." diff --git a/apps/server/src/provider/ProviderHostPathWatcher.ts b/apps/server/src/provider/ProviderHostPathWatcher.ts new file mode 100644 index 0000000000000..b77d9ec39e31d --- /dev/null +++ b/apps/server/src/provider/ProviderHostPathWatcher.ts @@ -0,0 +1,18 @@ +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Stream from "effect/Stream"; + +import * as DesktopTelemetryReceiver from "../resourceTelemetry/DesktopTelemetryReceiver.ts"; +import * as ProviderRegistry from "./ProviderRegistry.ts"; + +/** The desktop's shell PATH can arrive after drivers were built and checked against a bare one. */ +export const layer = Layer.effectDiscard( + Effect.gen(function* () { + const receiver = yield* DesktopTelemetryReceiver.DesktopTelemetryReceiver; + const providers = yield* ProviderRegistry.ProviderRegistry; + yield* receiver.pathChanges.pipe( + Stream.runForEach(() => providers.rebuildMissingAndRefresh), + Effect.forkScoped, + ); + }), +); diff --git a/apps/server/src/provider/ProviderInstanceRegistry.test.ts b/apps/server/src/provider/ProviderInstanceRegistry.test.ts index 95866780189ac..226eba3b7b7eb 100644 --- a/apps/server/src/provider/ProviderInstanceRegistry.test.ts +++ b/apps/server/src/provider/ProviderInstanceRegistry.test.ts @@ -957,6 +957,23 @@ describe("ProviderInstanceRegistry secret environment", () => { }).pipe(Effect.scoped); }); + it.effect("does not rebuild an instance replaced after selection", () => { + const fixture = setup({ [secret.reference]: "sk-live" }); + return Effect.gen(function* () { + const { registry, mutator } = yield* fixture.make; + const selected = yield* registry.listInstances; + fixture.vault.set(secret.reference, "sk-rotated"); + yield* mutator.reloadSecrets(); + const replacement = yield* registry.getInstance(instanceId); + expect(replacement).not.toBe(selected[0]); + yield* mutator.rebuild(selected); + expect(yield* registry.getInstance(instanceId)).toBe(replacement); + expect(fixture.created).toHaveLength(2); + yield* mutator.rebuild([replacement!]); + expect(fixture.created).toHaveLength(3); + }).pipe(Effect.scoped); + }); + it.effect("reads a config field typed SettingsSecretValue and rebuilds when it rotates", () => { const apiKey: SecretReference = { ...secret, reference: "op://Private/Gateway/key" }; const fixture = setup( diff --git a/apps/server/src/provider/ProviderInstanceRegistry.ts b/apps/server/src/provider/ProviderInstanceRegistry.ts index 3f4df0eb35d18..b00c5b41e296d 100644 --- a/apps/server/src/provider/ProviderInstanceRegistry.ts +++ b/apps/server/src/provider/ProviderInstanceRegistry.ts @@ -542,6 +542,22 @@ export const makeProviderInstanceRegistry = Effect.fn( Effect.provideContext(driverContext), ); + const rebuild: ProviderInstanceRegistryMutator.ProviderInstanceRegistryMutatorShape["rebuild"] = + Effect.fn("ProviderInstanceRegistry.rebuild")( + function* (instances) { + const live = yield* Ref.get(entries); + const ids = new Set( + instances + .filter((instance) => live.get(instance.instanceId)?.instance === instance) + .map((instance) => instance.instanceId), + ); + if (ids.size === 0) return; + yield* reconcileWithR(yield* Ref.get(configMapRef), ids); + }, + (effect) => semaphore.withPermits(1)(effect), + Effect.provideContext(driverContext), + ); + // Reading secrets can wait on a vault unlock, so the server does not wait for it to start. yield* semaphore.withPermits(1)( Ref.set(configMapRef, input.configMap).pipe( @@ -592,6 +608,7 @@ export const makeProviderInstanceRegistry = Effect.fn( const mutator: ProviderInstanceRegistryMutator.ProviderInstanceRegistryMutatorShape = { reconcile, reloadSecrets, + rebuild, }; return { registry, mutator }; diff --git a/apps/server/src/provider/ProviderInstanceRegistryHydration.test.ts b/apps/server/src/provider/ProviderInstanceRegistryHydration.test.ts index dc44f0324a6b0..0ba6245e98290 100644 --- a/apps/server/src/provider/ProviderInstanceRegistryHydration.test.ts +++ b/apps/server/src/provider/ProviderInstanceRegistryHydration.test.ts @@ -165,6 +165,7 @@ describe("layerSettingsWatcher", () => { const mutator = Layer.succeed(ProviderInstanceRegistryMutator, { reconcile: () => Queue.offer(reconciled, undefined), reloadSecrets: () => Effect.sync(() => void reloads++), + rebuild: () => Effect.void, }); yield* Layer.build( layerSettingsWatcher(Stream.fromQueue(changes), initial).pipe(Layer.provide(mutator)), diff --git a/apps/server/src/provider/ProviderInstanceRegistryMutator.ts b/apps/server/src/provider/ProviderInstanceRegistryMutator.ts index 35eb52d3f2906..2a5659cf6a3d6 100644 --- a/apps/server/src/provider/ProviderInstanceRegistryMutator.ts +++ b/apps/server/src/provider/ProviderInstanceRegistryMutator.ts @@ -1,5 +1,6 @@ /** Reconcile closes removed/replaced scopes before creating replacements: at most one live instance per id. */ import type { ProviderInstanceConfigMap, ProviderInstanceId } from "@t3tools/contracts"; +import type { ProviderInstance } from "@t3tools/provider-core/server/driver"; import * as Context from "effect/Context"; import type * as Effect from "effect/Effect"; @@ -7,6 +8,8 @@ export interface ProviderInstanceRegistryMutatorShape { readonly reconcile: (configMap: ProviderInstanceConfigMap) => Effect.Effect; /** Re-reads the secrets of every instance, or one, and rebuilds those whose secrets changed. */ readonly reloadSecrets: (instanceId?: ProviderInstanceId) => Effect.Effect; + /** Rebuilds these instances from their current config, so their drivers read the environment again. */ + readonly rebuild: (instances: ReadonlyArray) => Effect.Effect; } export class ProviderInstanceRegistryMutator extends Context.Service< diff --git a/apps/server/src/provider/ProviderRegistry.test.ts b/apps/server/src/provider/ProviderRegistry.test.ts index 5c53f76237b6a..df37c55c9c485 100644 --- a/apps/server/src/provider/ProviderRegistry.test.ts +++ b/apps/server/src/provider/ProviderRegistry.test.ts @@ -68,6 +68,7 @@ import { } from "./providerStatusCache.ts"; import { COMPACT_SLASH_COMMAND } from "@t3tools/provider-core/server/snapshotProbe"; import type { + AnyProviderDriver, ProviderInstance, ProviderWorkspaceSnapshot, } from "@t3tools/provider-core/server/driver"; @@ -76,6 +77,10 @@ import * as ProviderInstanceRegistryMutator from "./ProviderInstanceRegistryMuta import * as ProviderRegistry from "./ProviderRegistry.ts"; import { mergeProviderInstanceEnvironment } from "@t3tools/provider-core/server/instanceEnvironment"; import { makeManualOnlyProviderMaintenanceCapabilities } from "@t3tools/provider-core/server/maintenanceResolver"; +import * as SecretResolver from "@t3tools/secret-provider-core/server/SecretResolver"; +import * as ProviderHostPathWatcher from "./ProviderHostPathWatcher.ts"; +import * as DesktopTelemetryReceiver from "../resourceTelemetry/DesktopTelemetryReceiver.ts"; +import { providerHostEnvironment } from "../policy/managedProviders.ts"; vi.mock("@anthropic-ai/claude-agent-sdk", { spy: true }); const decodeServerProvider = Schema.decodeSync(ServerProvider); @@ -1156,6 +1161,186 @@ it.layer( ); }); + it.effect.each([ + "replay", + "in-flight missing", + "in-flight installed", + "launcher", + "codex timeout", + "other error", + ] as const)("recovers a copied driver environment after late PATH (%s)", (mode) => + Effect.gen(function* () { + // "launcher": the Codex CLI is found on the early PATH, but its check fails until the + // late PATH brings the node its npm launcher runs. "other error" is the same failure on a + // driver whose errors may be auth or network, so it must not be rebuilt. "codex timeout" is a + // Codex error other than a launcher failure, which must not be rebuilt either. + const failsEarly = + mode === "launcher" || mode === "codex timeout" || mode === "other error"; + const replay = mode === "replay" || failsEarly; + const host = { PATH: "/early/bin", T3CODE_CODEX_LAUNCH_ARGS: "hidden" }; + const environment = providerHostEnvironment( + { organization: "PATH test", lockedSettings: [] }, + host, + ); + const updates = yield* DesktopTelemetryReceiver.makeDesktopShellEnvironmentUpdates( + host, + "linux", + ); + const entered = yield* Deferred.make(); + const release = yield* Deferred.make(); + const finished = yield* Deferred.make(); + const paths: Array = []; + let first = true; + const driverKind = ProviderDriverKind.make( + mode === "launcher" || mode === "codex timeout" ? "codex" : "pi", + ); + const instanceId = ProviderInstanceId.make("late-path"); + const driver: AnyProviderDriver = { + driverKind, + metadata: { displayName: "PATH fixture" }, + configSchema: Schema.Unknown, + defaultConfig: () => ({}), + create: (input) => + Effect.gen(function* () { + const copied = { ...environment }; + paths.push(copied.PATH); + const late = copied.PATH?.includes("/late/bin") === true; + const installed = mode === "in-flight installed" || failsEarly || late; + const ready = failsEarly ? late : installed; + const snapshot: ServerProvider = { + instanceId, + driver: driverKind, + enabled: true, + installed: false, + status: "warning", + auth: { status: "unknown" }, + version: null, + checkedAt: "2026-10-03T00:00:00.000Z", + models: [], + slashCommands: [], + skills: [], + }; + const current = yield* Ref.make(snapshot); + return { + instanceId: input.instanceId, + driverKind, + enabled: true, + displayName: undefined, + continuationIdentity: { driverKind, continuationKey: `${driverKind}:late-path` }, + snapshot: { + resolveMaintenance: () => + Effect.succeed( + makeManualOnlyProviderMaintenanceCapabilities({ + provider: driverKind, + packageName: null, + }), + ), + getSnapshot: Ref.get(current), + refresh: Effect.gen(function* () { + if (first && !replay) { + first = false; + yield* Deferred.succeed(entered, undefined); + yield* Deferred.await(release); + } + const checked = { + ...snapshot, + installed, + status: ready ? ("ready" as const) : ("error" as const), + ...(ready + ? {} + : { + message: + mode === "codex timeout" + ? "Timed out while checking Codex app-server provider status." + : "Codex app-server provider probe failed: transport failed.", + }), + }; + yield* Ref.set(current, checked); + return checked; + }), + streamChanges: Stream.empty, + applyUsageLimits: () => Effect.void, + }, + orchestrationAdapter: {} as ProviderInstance["orchestrationAdapter"], + textGeneration: {} as ProviderInstance["textGeneration"], + } satisfies ProviderInstance; + }), + }; + const built = yield* ProviderInstanceRegistry.makeProviderInstanceRegistry({ + drivers: [driver], + configMap: { [instanceId]: { driver: driverKind, enabled: true } }, + }).pipe(Effect.provide(Layer.mock(SecretResolver.SecretResolver)({}))); + yield* Effect.gen(function* () { + const registry = yield* ProviderRegistry.ProviderRegistry; + const running = replay ? undefined : yield* registry.refresh().pipe(Effect.forkScoped); + if (running) yield* Deferred.await(entered); + yield* updates.apply({ + version: 1, + type: "shellEnvironment", + platform: "linux", + discovered: true, + environment: { PATH: "/early/bin:/late/bin" }, + }); + expect(paths).toEqual(["/early/bin"]); + if (replay) { + yield* Layer.build( + ProviderHostPathWatcher.layer.pipe( + Layer.provide( + DesktopTelemetryReceiver.layerTest({ pathChanges: updates.pathChanges }), + ), + Layer.provide( + Layer.succeed(ProviderRegistry.ProviderRegistry, { + ...registry, + rebuildMissingAndRefresh: registry.rebuildMissingAndRefresh.pipe( + Effect.tap(() => Deferred.succeed(finished, undefined)), + ), + }), + ), + ), + ); + } else { + yield* registry.rebuildMissingAndRefresh.pipe( + Effect.tap(() => Deferred.succeed(finished, undefined)), + Effect.forkScoped({ startImmediately: true }), + ); + expect(paths).toEqual(["/early/bin"]); + } + yield* Deferred.succeed(release, undefined); + if (running) yield* Fiber.join(running); + yield* Deferred.await(finished); + expect( + (yield* registry.getProviders).find((row) => row.instanceId === instanceId)?.installed, + ).toBe(true); + expect(paths).toEqual( + mode === "in-flight installed" || mode === "codex timeout" || mode === "other error" + ? ["/early/bin"] + : ["/early/bin", "/early/bin:/late/bin"], + ); + }).pipe( + Effect.provide( + ProviderRegistry.layer.pipe( + Layer.provide( + Layer.succeed(ProviderInstanceRegistry.ProviderInstanceRegistry, built.registry), + ), + Layer.provide( + Layer.succeed( + ProviderInstanceRegistryMutator.ProviderInstanceRegistryMutator, + built.mutator, + ), + ), + ), + ), + ); + }).pipe( + Effect.scoped, + Effect.provide( + ServerConfig.layerTest(process.cwd(), { prefix: "t3-late-path-" }).pipe( + Layer.provideMerge(Layer.mergeAll(NodeServices.layer, NodeCrypto.layer)), + ), + ), + ), + ); + describe("Codex model inventories", () => { const cachedProvider = { instanceId: ProviderInstanceId.make("codex-personal"), diff --git a/apps/server/src/provider/ProviderRegistry.ts b/apps/server/src/provider/ProviderRegistry.ts index a0ea5c78fbc61..d3a843ae9a1a8 100644 --- a/apps/server/src/provider/ProviderRegistry.ts +++ b/apps/server/src/provider/ProviderRegistry.ts @@ -47,6 +47,7 @@ import { latestProviderVersionCacheKeys, } from "@t3tools/provider-core/server/maintenanceResolver"; import type { ProviderSnapshotSource } from "./builtInProviderCatalog.ts"; +import { isCodexLaunchFailure } from "./CodexProvider.ts"; export type ProviderMaintenanceActionKind = "update"; @@ -69,6 +70,13 @@ export class ProviderRegistry extends Context.Service< instanceId?: ProviderInstanceId, ) => Effect.Effect>; + /** + * Rebuilds and re-checks enabled instances not yet found installed, plus Codex CLIs that cannot run, + * after the host environment changed. Drivers copy the environment when built, so only a rebuild + * sees a later PATH. + */ + readonly rebuildMissingAndRefresh: Effect.Effect>; + /** Only refreshed instances can verify an update; the published list still includes every row. */ readonly refreshAfterUpdate: ( provider: ProviderDriverKind, @@ -708,6 +716,41 @@ export const layer = Layer.effect( return yield* instanceId === undefined ? refresh() : refreshInstance(instanceId); }); + // Instances whose check passed keep running; restarting them could end live sessions. A Codex + // CLI that spawns but cannot run (its npm launcher lacks node) is rebuilt too. Auth, network, + // and timeout errors are not, since a rebuild would close their sessions. + const rebuildMissingAndRefresh = Effect.fn("rebuildMissingAndRefresh")(function* () { + const instances = yield* instanceRegistry.listInstances; + yield* refresh().pipe(Effect.catchCause(recoverRefreshFailure)); + const missing = yield* Effect.filter(instances, (instance) => + Effect.gen(function* () { + // ACP also reports installed:false for setup states unrelated to executable discovery. + if ( + !instance.enabled || + instance.driverKind === "acpRegistry" || + !(yield* isCurrentInstance(instance)) + ) + return false; + const snapshot = yield* instance.snapshot.getSnapshot; + return ( + !snapshot.installed || + (instance.driverKind === ProviderDriverKind.make("codex") && + isCodexLaunchFailure(snapshot)) + ); + }), + ); + if (missing.length === 0) return yield* Ref.get(providersRef); + yield* instanceRegistryMutator.rebuild(missing); + // A shared pass already in flight may hold the retired instances, so probe each rebuilt one. + yield* Effect.forEach( + missing, + (instance) => + refreshInstance(instance.instanceId).pipe(Effect.catchCause(recoverRefreshFailure)), + { concurrency: "unbounded", discard: true }, + ); + return yield* Ref.get(providersRef); + }); + const refreshAfterUpdate = Effect.fn("refreshAfterUpdate")(function* ( provider: ProviderDriverKind, selectedInstanceId: ProviderInstanceId, @@ -1105,6 +1148,9 @@ export const layer = Layer.effect( refreshInstance(instanceId).pipe(Effect.catchCause(recoverRefreshFailure)), reloadSecretsAndRefresh: (instanceId?: ProviderInstanceId) => reloadSecretsAndRefresh(instanceId).pipe(Effect.catchCause(recoverRefreshFailure)), + rebuildMissingAndRefresh: rebuildMissingAndRefresh().pipe( + Effect.catchCause(recoverRefreshFailure), + ), refreshWorkspaceSnapshot: (input) => refreshWorkspaceSnapshot(input).pipe(Effect.catchCause(recoverRefreshFailure)), refreshAfterUpdate, diff --git a/apps/server/src/provider/providerMaintenanceRunner.test.ts b/apps/server/src/provider/providerMaintenanceRunner.test.ts index adf53c5267203..ddc3a6bd90272 100644 --- a/apps/server/src/provider/providerMaintenanceRunner.test.ts +++ b/apps/server/src/provider/providerMaintenanceRunner.test.ts @@ -198,6 +198,7 @@ function makeRegistry( refresh: () => Ref.get(providersRef), refreshInstance: () => Ref.get(providersRef), reloadSecretsAndRefresh: () => Ref.get(providersRef), + rebuildMissingAndRefresh: Ref.get(providersRef), refreshAfterUpdate: () => { const version = options?.installedVersionAfterUpdate; if (version === undefined) { diff --git a/apps/server/src/provider/testUtils/providerRegistryMock.ts b/apps/server/src/provider/testUtils/providerRegistryMock.ts index 3ae8079414a9c..7df89c4fd88a2 100644 --- a/apps/server/src/provider/testUtils/providerRegistryMock.ts +++ b/apps/server/src/provider/testUtils/providerRegistryMock.ts @@ -12,6 +12,7 @@ const makeProviderRegistryMock = ( refresh: () => Effect.succeed(providers), refreshInstance: () => Effect.succeed(providers), reloadSecretsAndRefresh: () => Effect.succeed(providers), + rebuildMissingAndRefresh: Effect.succeed(providers), refreshAfterUpdate: () => Effect.succeed(providers), refreshWorkspaceSnapshot: () => Effect.succeed(providers), invalidateModelCaches: () => Effect.void, diff --git a/apps/server/src/resourceTelemetry/DesktopTelemetryReceiver.ts b/apps/server/src/resourceTelemetry/DesktopTelemetryReceiver.ts index 7d9de87ecfd98..c769e5e95796f 100644 --- a/apps/server/src/resourceTelemetry/DesktopTelemetryReceiver.ts +++ b/apps/server/src/resourceTelemetry/DesktopTelemetryReceiver.ts @@ -175,6 +175,8 @@ export class DesktopTelemetryReceiver extends Context.Service< never, Scope.Scope >; + /** Emits when a desktop shell environment changes this process's PATH; replays the last one. */ + readonly pathChanges: Stream.Stream; readonly health: Effect.Effect; readonly subscribeHealth: Effect.Effect< { @@ -395,12 +397,31 @@ export function requireDesktopTelemetryWriteProgress( : Effect.fail(new DesktopTelemetryControlStalled({ fd, remainingBytes })); } +export const makeDesktopShellEnvironmentUpdates = Effect.fn( + "DesktopTelemetryReceiver.makeDesktopShellEnvironmentUpdates", +)(function* (environment: NodeJS.ProcessEnv, platform: NodeJS.Platform) { + const applyShellEnvironment = makeApplyDesktopShellEnvironmentSnapshot(environment, platform); + // Replayed, because the provider watcher may subscribe after the snapshot arrived. + const pathChanges = yield* Effect.acquireRelease( + PubSub.sliding({ capacity: 1, replay: 1 }), + PubSub.shutdown, + ); + const readPath = () => environment.PATH ?? environment.Path; + // Untraced: applyShellEnvironment already carries this operation's span. + const apply = Effect.fnUntraced(function* (message: DesktopShellEnvironmentSnapshot) { + const before = readPath(); + yield* applyShellEnvironment(message); + if (readPath() !== before) yield* PubSub.publish(pathChanges, undefined); + }); + return { apply, pathChanges: Stream.fromPubSub(pathChanges) }; +}); + /** @public */ export const make = Effect.fn("resourceTelemetry.desktopTelemetryReceiver.make")(function* () { const config = yield* ServerConfig.ServerConfig; const environment = yield* HostProcess.Environment; const platform = yield* HostProcess.Platform; - const applyShellSnapshot = makeApplyDesktopShellEnvironmentSnapshot(environment, platform); + const shellEnvironment = yield* makeDesktopShellEnvironmentUpdates(environment, platform); const serverSettings = yield* ServerSettings.ServerSettingsService; const latest = yield* Ref.make(Option.none()); const receiverStartedAt = yield* DateTime.now; @@ -597,7 +618,7 @@ export const make = Effect.fn("resourceTelemetry.desktopTelemetryReceiver.make") } if (message.type === "shellEnvironment") { - return recordContact.pipe(Effect.andThen(applyShellSnapshot(message))); + return recordContact.pipe(Effect.andThen(shellEnvironment.apply(message))); } if (message.type === "desktopUpdateStatus") { @@ -704,6 +725,7 @@ export const make = Effect.fn("resourceTelemetry.desktopTelemetryReceiver.make") return DesktopTelemetryReceiver.of({ latest: Ref.get(latest), changes: Stream.fromPubSub(changes), + pathChanges: shellEnvironment.pathChanges, subscribe: snapshotMutex.withPermits(1)( Effect.gen(function* () { const initial = yield* Ref.get(latest); @@ -767,6 +789,7 @@ export const layerTest = ( DesktopTelemetryReceiver.of({ latest, changes, + pathChanges: overrides.pathChanges ?? Stream.empty, subscribe: overrides.subscribe ?? latest.pipe( diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index 47190f60dcf51..338df12dc4695 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -75,6 +75,7 @@ import { BUILT_IN_SOURCE_CONTROL_DRIVER_LIST } from "./sourceControl/sourceContr import * as TextGeneration from "./textGeneration/TextGeneration.ts"; import * as SecretResolver from "./secrets/SecretResolver.ts"; import * as ProviderInstanceRegistryHydration from "./provider/ProviderInstanceRegistryHydration.ts"; +import * as ProviderHostPathWatcher from "./provider/ProviderHostPathWatcher.ts"; import * as ProjectSettingsOverrideSweep from "./provider/ProjectSettingsOverrideSweep.ts"; import * as TerminalManager from "./terminal/Manager.ts"; import * as McpHttpServer from "./mcp/McpHttpServer.ts"; @@ -643,6 +644,7 @@ const layerServerRestart = Layer.effect( const layerRuntimeCoreDependenciesBase = Layer.mergeAll( AgentAwarenessRelay.layer, + ProviderHostPathWatcher.layer.pipe(Layer.provide(layerDesktopTelemetryReceiver)), HeldHooksWaker.layer, layerThreadSettlementWorker, layerIdleCompactionWorker,