diff --git a/apps/mobile/src/App.tsx b/apps/mobile/src/App.tsx index 6287a05956519..b4bec531a6366 100644 --- a/apps/mobile/src/App.tsx +++ b/apps/mobile/src/App.tsx @@ -20,6 +20,8 @@ import { } from "@react-navigation/native"; import { pipe } from "effect/Function"; import { DEFAULT_HOSTED_APP_URL } from "@t3tools/shared/connectAuth"; +import * as Effect from "effect/Effect"; +import { mobileInteraction } from "./connection/background-activity-scopes"; import { RegistryContext } from "@effect/atom-react"; import { ProjectChipIconRasterizer } from "./native/ProjectChipIconImages"; @@ -243,7 +245,10 @@ function AppContent() { <> - + Effect.runSync(mobileInteraction.record)} + > diff --git a/apps/mobile/src/components/ComposerEditor.tsx b/apps/mobile/src/components/ComposerEditor.tsx index f7118b95b425f..46f378cc17c98 100644 --- a/apps/mobile/src/components/ComposerEditor.tsx +++ b/apps/mobile/src/components/ComposerEditor.tsx @@ -4,6 +4,8 @@ import { useAtomValue } from "@effect/atom-react"; import { AsyncResult } from "effect/reactivity"; import { settlePromise, squashAtomCommandFailure } from "@t3tools/client-runtime/state/runtime"; import { useEffect, useState } from "react"; +import * as Effect from "effect/Effect"; +import { mobileInteraction } from "../connection/background-activity-scopes"; import type { EnvironmentId } from "@t3tools/contracts"; import { encodeComposerContextFragment } from "@t3tools/shared/composerContextClipboard"; import { collectComposerContextReferences } from "@t3tools/shared/composerContextReferences"; @@ -69,6 +71,7 @@ export function ComposerEditor({ // oxlint-disable-next-line t3code/no-unreasoned-state-sync -- releases the attachment file leases held for native text undo when the history is replaced or unmounts useEffect(() => () => contextHistory.dispose(), [contextHistory]); const changeText = (text: string) => { + Effect.runSync(mobileInteraction.record); const restored = contextHistory.restore( text, draftKey ? getComposerDraftSnapshot(draftKey) : draft, diff --git a/apps/mobile/src/connection/background-activity-scopes.ts b/apps/mobile/src/connection/background-activity-scopes.ts index c390461cb22bf..e62ff16b71d6a 100644 --- a/apps/mobile/src/connection/background-activity-scopes.ts +++ b/apps/mobile/src/connection/background-activity-scopes.ts @@ -1,6 +1,7 @@ import type { EnvironmentRpcSubscriptionObservation } from "@t3tools/client-runtime/rpc"; import { type BackgroundScope, type EnvironmentId, WS_METHODS } from "@t3tools/contracts"; import * as Effect from "effect/Effect"; +import * as Clock from "effect/Clock"; interface RetainedScope { readonly environmentId: EnvironmentId; @@ -8,6 +9,22 @@ interface RetainedScope { refCount: number; } +let lastInteractionNanos: bigint | undefined; +export const mobileInteraction = { + record: Clock.monotonicTimeNanos.pipe( + Effect.tap((now) => + Effect.sync(() => { + lastInteractionNanos = now; + }), + ), + ), + isRecent: Clock.monotonicTimeNanos.pipe( + Effect.map( + (now) => lastInteractionNanos !== undefined && now - lastInteractionNanos < 60_000_000_000n, + ), + ), +}; + const retainedScopes = new Map(); const listeners = new Set<() => void>(); diff --git a/apps/mobile/src/connection/background-activity.test.ts b/apps/mobile/src/connection/background-activity.test.ts index 170ba4efa9c90..339d991a2698f 100644 --- a/apps/mobile/src/connection/background-activity.test.ts +++ b/apps/mobile/src/connection/background-activity.test.ts @@ -1,14 +1,26 @@ import { EnvironmentId, WS_METHODS } from "@t3tools/contracts"; import { describe, expect, it } from "@effect/vitest"; import * as Effect from "effect/Effect"; +import * as TestClock from "effect/testing/TestClock"; import { onRetainedMobileBackgroundScopesChange, + mobileInteraction, observeMobileBackgroundActivitySubscription, retainedMobileBackgroundScopes, } from "./background-activity-scopes"; describe("mobile background activity", () => { + it.effect("expires interaction recency while the app remains foregrounded", () => + Effect.gen(function* () { + yield* mobileInteraction.record; + expect(yield* mobileInteraction.isRecent).toBe(true); + yield* TestClock.adjust("2 minutes"); + expect(yield* mobileInteraction.isRecent).toBe(false); + yield* mobileInteraction.record; + expect(yield* mobileInteraction.isRecent).toBe(true); + }), + ); it.effect("retains VCS demand only while the mobile subscription is active", () => Effect.gen(function* () { const environmentId = EnvironmentId.make("mobile-environment"); diff --git a/apps/mobile/src/connection/background-activity.ts b/apps/mobile/src/connection/background-activity.ts index 66eb2b10e0263..f5d9f033e07aa 100644 --- a/apps/mobile/src/connection/background-activity.ts +++ b/apps/mobile/src/connection/background-activity.ts @@ -18,6 +18,7 @@ import { AppState } from "react-native"; import * as MobileStorage from "../persistence/mobile-storage"; import { + mobileInteraction, observeMobileBackgroundActivitySubscription, onRetainedMobileBackgroundScopesChange, retainedMobileBackgroundScopes, @@ -54,10 +55,13 @@ export const layerReporter = Layer.effectDiscard( const reportRequests = yield* Queue.sliding(1); const requestReport = () => Queue.offerUnsafe(reportRequests, undefined); let appState = AppState.currentState; + yield* mobileInteraction.record; + const runSync = Effect.runSyncWith(yield* Effect.context()); const report = Effect.gen(function* () { const observedAtMs = yield* Clock.currentTimeMillis; const active = appState === "active"; + const recentlyInteracted = active && (yield* mobileInteraction.isRecent); const entries = yield* SubscriptionRef.get(registry.entries); yield* Effect.forEach( entries.keys(), @@ -71,7 +75,7 @@ export const layerReporter = Layer.effectDiscard( clientKind: "mobile", visible: active, focused: active, - recentlyInteracted: active, + recentlyInteracted, appState: normalizeAppState(appState), scopes: [ ...BASELINE_SCOPES, @@ -93,6 +97,7 @@ export const layerReporter = Layer.effectDiscard( Effect.sync(() => { const removeScopeListener = onRetainedMobileBackgroundScopesChange(requestReport); const subscription = AppState.addEventListener("change", (nextState) => { + if (nextState === "active" && appState !== "active") runSync(mobileInteraction.record); appState = nextState; requestReport(); }); diff --git a/apps/mobile/src/features/settings/SettingsServerControlsRouteScreen.tsx b/apps/mobile/src/features/settings/SettingsServerControlsRouteScreen.tsx index 793e3ae75412c..e2d14a30efd91 100644 --- a/apps/mobile/src/features/settings/SettingsServerControlsRouteScreen.tsx +++ b/apps/mobile/src/features/settings/SettingsServerControlsRouteScreen.tsx @@ -272,6 +272,9 @@ function ServerSettingsDetail(props: { readonly page: SettingsPage }) { RUNTIME_MODE_CHOICES, targets.map((target) => target.environment.serverConfig.managedPolicy), ); + const supportsAutomaticUpdates = targets.every( + (target) => target.environment.serverConfig.environment.capabilities.automaticUpdates === true, + ); const disabledFor = (key: string) => disabled || managed(key) !== undefined || @@ -542,6 +545,28 @@ function ServerSettingsDetail(props: { readonly page: SettingsPage }) { disabled={disabledFor("enableProviderUpdateChecks")} onValueChange={(value) => write({ enableProviderUpdateChecks: value })} /> + + write({ automaticUpdates: value })} + /> + { ), ), ); + + it.effect("an interaction resets the quiet clock and a heartbeat does not", () => + Effect.gen(function* () { + const policy = yield* BackgroundPolicy.BackgroundPolicy; + const sessionId = AuthSessionId.make("session-1"); + yield* TestClock.adjust("20 minutes"); + assert.isTrue( + Duration.isGreaterThanOrEqualTo( + yield* policy.sinceLastClientInteraction, + Duration.minutes(20), + ), + ); + + yield* policy.reportClientActivity( + sessionId, + RpcClientId.make(1), + makeReport({ recentlyInteracted: false }), + ); + assert.isTrue( + Duration.isGreaterThanOrEqualTo( + yield* policy.sinceLastClientInteraction, + Duration.minutes(20), + ), + ); + + yield* policy.reportClientActivity(sessionId, RpcClientId.make(1), makeReport()); + yield* TestClock.adjust("1 minute"); + assert.isTrue( + Duration.isLessThan(yield* policy.sinceLastClientInteraction, Duration.minutes(2)), + ); + }).pipe(Effect.provide(layerFor(nominalHostPower))), + ); }); diff --git a/apps/server/src/background/BackgroundPolicy.ts b/apps/server/src/background/BackgroundPolicy.ts index 331b4c4f3365b..1559ac4c2bbba 100644 --- a/apps/server/src/background/BackgroundPolicy.ts +++ b/apps/server/src/background/BackgroundPolicy.ts @@ -12,8 +12,10 @@ import { resolveServerBackgroundActivitySettings, type ResolvedBackgroundActivitySettings, } from "@t3tools/shared/backgroundActivitySettings"; +import * as Clock from "effect/Clock"; import * as DateTime from "effect/DateTime"; import * as Context from "effect/Context"; +import * as Duration from "effect/Duration"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as Ref from "effect/Ref"; @@ -34,6 +36,8 @@ export class BackgroundPolicy extends Context.Service< rpcClientId: RpcClientId, ) => Effect.Effect; readonly snapshot: Effect.Effect; + /** Time since recent client interaction, measured on the monotonic clock. */ + readonly sinceLastClientInteraction: Effect.Effect; readonly shouldRunScopeWork: (scope: BackgroundScope) => Effect.Effect; readonly shouldRunOpportunisticWork: Effect.Effect; } @@ -196,6 +200,7 @@ export const make = Effect.fn("background.policy.make")(function* () { const hostPowerMonitor = yield* HostPowerMonitor.HostPowerMonitor; const serverSettings = yield* ServerSettings.ServerSettingsService; const leasesRef = yield* Ref.make(new Map()); + const lastClientInteractionNanosRef = yield* Ref.make(yield* Clock.monotonicTimeNanos); const backgroundActivitySettings = serverSettings.getSettings.pipe( Effect.map(resolveServerBackgroundActivitySettings), @@ -233,6 +238,9 @@ export const make = Effect.fn("background.policy.make")(function* () { expiresAt, }; yield* Ref.update(leasesRef, (leases) => upsertClientActivityLease(leases, lease, now)); + if (input.recentlyInteracted) { + yield* Ref.set(lastClientInteractionNanosRef, yield* Clock.monotonicTimeNanos); + } }); const removeRpcClient: BackgroundPolicy["Service"]["removeRpcClient"] = ( @@ -289,6 +297,11 @@ export const make = Effect.fn("background.policy.make")(function* () { reportClientActivity, removeRpcClient, snapshot, + sinceLastClientInteraction: Effect.zipWith( + Clock.monotonicTimeNanos, + Ref.get(lastClientInteractionNanosRef), + (now, last) => Duration.nanos(now - last), + ), shouldRunScopeWork, shouldRunOpportunisticWork, }); diff --git a/apps/server/src/environment/ServerEnvironment.ts b/apps/server/src/environment/ServerEnvironment.ts index 9e29b0fdedc95..ae6c3c8b6c3ca 100644 --- a/apps/server/src/environment/ServerEnvironment.ts +++ b/apps/server/src/environment/ServerEnvironment.ts @@ -270,6 +270,7 @@ export const make = Effect.gen(function* () { environmentIcon: true, projectCloneTracking: true, projectOrganization: true, + automaticUpdates: true, ignoredFileListing: true, triage: true, oneTimeSchedules: true, diff --git a/apps/server/src/mcp/toolkits/environment/handlers.ts b/apps/server/src/mcp/toolkits/environment/handlers.ts index fb1716885d6e8..cb2a2f4f5dec5 100644 --- a/apps/server/src/mcp/toolkits/environment/handlers.ts +++ b/apps/server/src/mcp/toolkits/environment/handlers.ts @@ -13,6 +13,7 @@ export function preferences(settings: ServerSettings) { defaultThreadEnvMode, newWorktreesStartFromOrigin, enableProviderUpdateChecks, + automaticUpdates, backgroundActivity, sourceControlWritingStyle, } = settings; @@ -21,6 +22,7 @@ export function preferences(settings: ServerSettings) { defaultThreadEnvMode, newWorktreesStartFromOrigin, enableProviderUpdateChecks, + automaticUpdates, backgroundActivity: { profile: backgroundActivity.profile }, sourceControlWritingStyle: { ...sourceControlWritingStyle, diff --git a/apps/server/src/mcp/toolkits/environment/tools.ts b/apps/server/src/mcp/toolkits/environment/tools.ts index 9e10380295de2..e23b889d9381f 100644 --- a/apps/server/src/mcp/toolkits/environment/tools.ts +++ b/apps/server/src/mcp/toolkits/environment/tools.ts @@ -19,6 +19,7 @@ const PreferenceFields = { defaultThreadEnvMode: ServerSettings.fields.defaultThreadEnvMode, newWorktreesStartFromOrigin: ServerSettings.fields.newWorktreesStartFromOrigin, enableProviderUpdateChecks: ServerSettings.fields.enableProviderUpdateChecks, + automaticUpdates: ServerSettings.fields.automaticUpdates, backgroundActivity: Schema.Struct({ profile: BackgroundActivityProfileSelection }), sourceControlWritingStyle: Schema.Struct({ mode: Schema.String, @@ -62,6 +63,7 @@ const EnvironmentPreferencesTool = Tool.make("t3_environment_preferences_update" defaultThreadEnvMode: ServerSettingsPatch.fields.defaultThreadEnvMode, newWorktreesStartFromOrigin: ServerSettingsPatch.fields.newWorktreesStartFromOrigin, enableProviderUpdateChecks: ServerSettingsPatch.fields.enableProviderUpdateChecks, + automaticUpdates: ServerSettingsPatch.fields.automaticUpdates, backgroundActivity: Schema.optionalKey(Schema.Struct({ profile: BackgroundActivityProfile })), sourceControlWritingStyle: ServerSettingsPatch.fields.sourceControlWritingStyle, }), diff --git a/apps/server/src/orchestration-v2/ProjectionRecovery.test.ts b/apps/server/src/orchestration-v2/ProjectionRecovery.test.ts index 6b1388f0ccfec..06585aeadb2ce 100644 --- a/apps/server/src/orchestration-v2/ProjectionRecovery.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionRecovery.test.ts @@ -510,6 +510,8 @@ it.effect("selects unfinished recovery work without reading settled thread histo JSON.stringify(plan), ); } + // Only runs count: queued runs and idle background items do not block an update. + assert.deepEqual(yield* projections.getRecoveryThreadIds("active-work"), [blocked]); const recoveryState = yield* projections.getRuntimeRecoveryProjection(queued); assert.deepEqual( recoveryState.runs.map((run) => run.id), diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 3d33f76b7a494..781562bf7218a 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -170,6 +170,9 @@ export type ProjectionRecoveryKind = | "queued-runs" | "runtime" | "restart-continuations" + // Runs a restart or CLI replacement would cut off: preparing, starting, + // running or waiting (a pending approval or question). + | "active-work" | "subagent-results" | "delegated-completions"; @@ -4217,6 +4220,11 @@ export const layer: Layer.Layer = THEN json_extract(payload_json, '$.restartContinuationOfRunId') IS NOT NULL ELSE 0 END `; + case "active-work": + return sql` + SELECT thread_id FROM orchestration_v2_projection_runs + WHERE status IN ('preparing', 'starting', 'running', 'waiting') + `; case "runtime": return sql` WITH pending_provider_threads AS MATERIALIZED ( diff --git a/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts b/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts index d1b5168258874..c9fa0b703e1e6 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts @@ -18,6 +18,7 @@ import { RunAttemptId, RunId, ThreadId, + TurnItemId, ProjectId, type OrchestrationV2ThreadProjection, OrchestrationV2DomainEvent, @@ -33,6 +34,7 @@ import * as Option from "effect/Option"; import * as Path from "effect/Path"; import * as PlatformError from "effect/PlatformError"; import * as Schema from "effect/Schema"; +import * as Stream from "effect/Stream"; import * as ServerConfig from "../config.ts"; import { withWorkspaceLease } from "../workspace/workspaceLease.ts"; @@ -41,6 +43,7 @@ import * as GitVcsDriver from "../vcs/GitVcsDriver.ts"; import { layerRealGit } from "./ProviderTurnStartService.testkit.ts"; import * as ProjectService from "../project/ProjectService.ts"; import * as ProviderAuthService from "../provider/ProviderAuthService.ts"; +import * as ProviderInstallLock from "../provider/ProviderInstallLock.ts"; import * as ServerSettings from "../serverSettings.ts"; import * as ContextHandoffService from "./ContextHandoffService.ts"; import { withInlineContext } from "./ContextHandoffDelivery.ts"; @@ -193,6 +196,10 @@ function startWithMissingWorktree( }), Layer.mock(RunExecutionService.RunExecutionServiceV2)({ startRootRun }), Layer.mock(RuntimePolicy.RuntimePolicyV2)({}), + Layer.mock(ProviderInstallLock.ProviderInstallLock)({ + untilIdle: Effect.void, + observe: Effect.succeed(() => Effect.succeed(false)), + }), ), ), ); @@ -269,6 +276,7 @@ it.each([ function makeLocalCommandHarness(input: { readonly text: string; + readonly driver?: ProviderDriverKind; readonly worktreePath?: string; readonly worktreesDir?: string; readonly workspaceRoot?: string; @@ -278,6 +286,10 @@ function makeLocalCommandHarness(input: { readonly previousMessages?: ReadonlyArray; readonly logoutFailure?: string; readonly openFailure?: unknown; + readonly installOverlap?: boolean; + readonly installIdle?: Effect.Effect; + readonly installObservation?: ProviderInstallLock.ProviderInstallLock["Service"]["observe"]; + readonly sessionEvents?: ProviderAdapter.ProviderAdapterV2SessionRuntime["events"]; readonly ensureThreadFailure?: unknown; readonly historyReadFailureAfterFallback?: unknown; readonly resumeFailure?: ProviderAdapter.ProviderAdapterV2Error; @@ -337,7 +349,7 @@ function makeLocalCommandHarness(input: { }; const providerThread: OrchestrationV2ThreadProjection["providerThreads"][number] = { id: providerThreadId, - driver: ProviderDriverKind.make("codex"), + driver: input.driver ?? ProviderDriverKind.make("codex"), providerInstanceId: newInstanceId, providerSessionId, appThreadId: threadId, @@ -802,6 +814,7 @@ function makeLocalCommandHarness(input: { resumeThread: () => Effect.succeed(sideChatThread), forkThread, startTurn, + events: input.sessionEvents ?? Stream.empty, }; const open = vi.fn(() => input.sideChat !== undefined @@ -835,6 +848,8 @@ function makeLocalCommandHarness(input: { } as never) : Effect.die("A local command must not open a native session."), ); + const release = vi.fn(() => Effect.void); + const releaseUnlessBusy = vi.fn(() => Effect.succeed(true)); const startRootRun = vi.fn< (input: RunExecutionService.RunExecutionServiceV2StartRootRunInput) => Effect.Effect >((runInput) => @@ -982,6 +997,8 @@ function makeLocalCommandHarness(input: { }), Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({ open, + release, + releaseUnlessBusy, get: () => Effect.succeed( input.previousTurn?.liveSession === false @@ -1003,6 +1020,12 @@ function makeLocalCommandHarness(input: { Layer.mock(RuntimePolicy.RuntimePolicyV2)({ resolve: resolveRuntimePolicy, }), + Layer.mock(ProviderInstallLock.ProviderInstallLock)({ + untilIdle: input.installIdle ?? Effect.void, + observe: + input.installObservation ?? + Effect.succeed(() => Effect.succeed(input.installOverlap === true)), + }), ), ), ); @@ -1017,6 +1040,8 @@ function makeLocalCommandHarness(input: { createWorktree, open, fallbackEnsureThread, + release, + releaseUnlessBusy, writeIfRunCurrent, startRootRun, startTurn, @@ -1658,6 +1683,233 @@ effectIt.effect("terminalizes a starting run when its provider session cannot op }), ); +effectIt.effect("retries a session launch once after an overlapping provider install", () => + Effect.gen(function* () { + const installFinished = yield* Deferred.make(); + const launchFailed = yield* Deferred.make(); + let waits = 0; + const harness = makeLocalCommandHarness({ + text: "Continue", + installOverlap: true, + installIdle: Effect.suspend(() => + ++waits === 1 ? Effect.void : Deferred.await(installFinished), + ), + sideChat: { contextRun: "delivered" }, + }); + harness.open.mockImplementationOnce(() => + Deferred.succeed(launchFailed, undefined).pipe( + Effect.andThen( + Effect.fail( + new ProviderSessionManager.ProviderSessionOpenError({ + instanceId: harness.newInstanceId, + providerSessionId: ProviderSessionId.make("new-provider-session"), + cause: new Error("spawn codex ENOENT"), + }), + ), + ), + ), + ); + + const pending = yield* harness.start.pipe(Effect.forkChild({ startImmediately: true })); + yield* Deferred.await(launchFailed); + expect(harness.open).toHaveBeenCalledOnce(); + expect(harness.projection().runs.at(-1)?.status).toBe("starting"); + yield* Deferred.succeed(installFinished, undefined); + yield* Fiber.join(pending); + + expect(harness.open).toHaveBeenCalledTimes(2); + expect(harness.startTurn).toHaveBeenCalledOnce(); + expect(harness.projection().runs.at(-1)?.status).toBe("running"); + expect(harness.projection().turnItems.some((item) => item.type === "error")).toBe(false); + }), +); + +effectIt.effect.each([ + "waited install", + "retried install", + "usage limit", + "initialized", + "other thread", + "api retry", +] as const)("preserves a session failure after %s", (scenario) => + Effect.gen(function* () { + let idle = false; + let generation = 0; + let events = Stream.empty as ProviderAdapter.ProviderAdapterV2SessionRuntime["events"]; + const harness = makeLocalCommandHarness({ + text: "Continue", + sideChat: { contextRun: "delivered" }, + sessionEvents: Stream.suspend(() => events), + installIdle: Effect.sync(() => void (idle = true)), + installObservation: Effect.sync(() => { + const observed = generation; + const overlapped = scenario !== "waited install" || !idle; + return () => + Effect.succeed(scenario === "retried install" ? observed !== generation : overlapped); + }), + }); + const failure = { + class: scenario === "usage limit" ? ("usage_limit" as const) : ("provider_error" as const), + message: "provider refused this turn", + code: null, + retryable: scenario === "api retry", + resetAt: "2026-09-04T13:00:00Z", + }; + // Claude's api_retry: the provider answered, so the launch is not an installer failure. + const event: ProviderAdapter.ProviderAdapterV2Event = + scenario === "api retry" + ? ({ + type: "turn_item.updated", + driver: ProviderDriverKind.make("claudeAgent"), + turnItem: { + id: TurnItemId.make("turn-item-api-retry"), + threadId: ThreadId.make("thread-native-account-command"), + runId: RunId.make("run-native-account-command"), + providerTurnId: ProviderTurnId.make("failed-turn"), + ordinal: 1, + type: "error", + status: "running", + failure, + retry: { attempt: 1, maxAttempts: 10, retryDelayMs: 500 }, + }, + } as ProviderAdapter.ProviderAdapterV2Event) + : { + type: "turn.terminal", + driver: ProviderDriverKind.make("codex"), + providerThreadId: ProviderThreadId.make( + scenario === "other thread" ? "another-provider-thread" : "new-provider-thread", + ), + providerTurnId: ProviderTurnId.make("failed-turn"), + runOrdinal: 2, + failureItemOrdinal: 1, + status: "failed", + failure, + threadDisposition: "reusable", + }; + events = Stream.make(event); + if (scenario === "retried install") { + harness.open.mockImplementationOnce(() => + Effect.sync(() => void generation++).pipe( + Effect.andThen( + Effect.fail( + new ProviderSessionManager.ProviderSessionOpenError({ + instanceId: harness.newInstanceId, + providerSessionId: ProviderSessionId.make("new-provider-session"), + cause: new Error("spawn codex ENOENT"), + }), + ), + ), + ), + ); + } + harness.startRootRun.mockImplementation((input) => + Effect.gen(function* () { + if (scenario === "initialized") yield* input.session.startTurn(input as never); + const result = yield* Stream.runCollect(input.session.events).pipe(Effect.result); + expect(result._tag).toBe("Success"); + if (result._tag === "Success") expect(result.success).toEqual([event]); + }).pipe(Effect.orDie), + ); + + yield* harness.start; + + expect(harness.release).not.toHaveBeenCalled(); + expect(harness.releaseUnlessBusy).not.toHaveBeenCalled(); + }), +); + +effectIt.effect("settles a launch that still fails after the install retry", () => + Effect.gen(function* () { + const harness = makeLocalCommandHarness({ + text: "Continue", + installOverlap: true, + openFailure: new Error("spawn codex ENOENT"), + }); + + yield* harness.start; + + expect(harness.open).toHaveBeenCalledTimes(2); + expect(harness.projection().runs.at(-1)?.status).toBe("failed"); + expect(harness.projection().attempts[0]?.status).toBe("failed"); + expect(harness.projection().nodes[0]?.status).toBe("failed"); + expect(harness.projection().turnItems).toMatchObject([ + { type: "error", failure: { message: "spawn codex ENOENT" } }, + ]); + }), +); + +effectIt.effect.each(["ensure", "queued", "queued terminal"] as const)( + "releases an overlapping install's failed session without resending the prompt after %s", + (stage) => + Effect.gen(function* () { + // Claude's query-exit path reports the dead CLI as a transport-error terminal. + const transportTerminal: ProviderAdapter.ProviderAdapterV2Event = { + type: "turn.terminal", + driver: ProviderDriverKind.make("claudeAgent"), + providerThreadId: ProviderThreadId.make("new-provider-thread"), + providerTurnId: ProviderTurnId.make("failed-turn"), + runOrdinal: 2, + failureItemOrdinal: 1, + status: "failed", + failure: { + class: "transport_error", + message: "Claude Code process exited", + code: null, + retryable: false, + }, + threadDisposition: "reusable", + }; + const sessionEvents: ProviderAdapter.ProviderAdapterV2SessionRuntime["events"] = + stage === "queued terminal" + ? Stream.make(transportTerminal) + : Stream.fail( + new ProviderAdapter.ProviderAdapterEventStreamError({ + driver: ProviderDriverKind.make("claudeAgent"), + providerSessionId: ProviderSessionId.make("new-provider-session"), + cause: new Error("provider exited before initialization"), + }), + ); + const harness = makeLocalCommandHarness({ + text: "Continue", + installOverlap: true, + ...(stage === "ensure" + ? { ensureThreadFailure: new Error("provider exited before initialization") } + : { + driver: ProviderDriverKind.make("claudeAgent"), + sideChat: { contextRun: "delivered" as const }, + sessionEvents, + }), + }); + if (stage !== "ensure") { + harness.startRootRun.mockImplementation((input) => + input.session + .startTurn(input as never) + .pipe( + Effect.andThen(Stream.runDrain(input.session.events)), + Effect.result, + Effect.asVoid, + ), + ); + } + + yield* harness.start; + + expect(harness.open).toHaveBeenCalledOnce(); + expect(harness.startTurn).toHaveBeenCalledTimes(stage === "ensure" ? 0 : 1); + expect(harness.releaseUnlessBusy).toHaveBeenCalledWith( + expect.objectContaining({ + reason: "runtime_error", + ownProviderThreadIds: [ProviderThreadId.make("new-provider-thread")], + }), + ); + if (stage !== "ensure") return; + expect(harness.projection().runs.at(-1)?.status).toBe("failed"); + expect(harness.projection().turnItems).toMatchObject([ + { type: "error", failure: { message: "Codex was updating on this machine. Try again." } }, + ]); + }), +); + effectIt.effect("leaves the run starting when a session-open failure will be retried", () => Effect.gen(function* () { const harness = makeLocalCommandHarness({ @@ -1740,38 +1992,41 @@ effectIt.effect("does not overwrite a run interrupted while its provider session }), ); -effectIt.effect("fails the run instead of replacing a native thread another client holds", () => - Effect.gen(function* () { - const detail = "This conversation is open in another Codex client."; - const harness = makeLocalCommandHarness({ - text: "Continue", - resumeFailure: new ProviderAdapter.ProviderAdapterThreadInUseError({ - driver: ProviderDriverKind.make("codex"), - providerThreadId: ProviderThreadId.make("new-provider-thread"), - detail, - cause: new Error("thread native-resume-thread already has an active writer"), - }), - }); +effectIt.effect.each([false, true])( + "fails the run instead of replacing a native thread another client holds (install overlap %s)", + (installOverlapped) => + Effect.gen(function* () { + const detail = "This conversation is open in another Codex client."; + const harness = makeLocalCommandHarness({ + text: "Continue", + installObservation: Effect.succeed(() => Effect.succeed(installOverlapped)), + resumeFailure: new ProviderAdapter.ProviderAdapterThreadInUseError({ + driver: ProviderDriverKind.make("codex"), + providerThreadId: ProviderThreadId.make("new-provider-thread"), + detail, + cause: new Error("thread native-resume-thread already has an active writer"), + }), + }); - const exit = yield* Effect.exit(harness.start); + const exit = yield* Effect.exit(harness.start); - expect(harness.fallbackEnsureThread).not.toHaveBeenCalled(); - expect(exit._tag).toBe("Success"); - expect(harness.startRootRun).not.toHaveBeenCalled(); - const projection = harness.projection(); - expect(projection.providerThreads.at(-1)?.nativeThreadRef?.nativeId).toBe( - "native-resume-thread", - ); - expect(projection.contextHandoffs).toEqual([]); - expect(projection.runs.at(-1)).toMatchObject({ status: "failed", startedAt: null }); - expect(projection.turnItems).toMatchObject([ - { - type: "error", - title: "Provider turn failed to start", - failure: { class: "provider_error", message: detail }, - }, - ]); - }), + expect(harness.fallbackEnsureThread).not.toHaveBeenCalled(); + expect(exit._tag).toBe("Success"); + expect(harness.startRootRun).not.toHaveBeenCalled(); + const projection = harness.projection(); + expect(projection.providerThreads.at(-1)?.nativeThreadRef?.nativeId).toBe( + "native-resume-thread", + ); + expect(projection.contextHandoffs).toEqual([]); + expect(projection.runs.at(-1)).toMatchObject({ status: "failed", startedAt: null }); + expect(projection.turnItems).toMatchObject([ + { + type: "error", + title: "Provider turn failed to start", + failure: { class: "provider_error", message: detail }, + }, + ]); + }), ); effectIt.effect("does not mistake a failed state read for a superseded run", () => @@ -1849,3 +2104,52 @@ for (const previousMessages of [[], ["/compact", " /COMPACT "]]) { }), ); } + +effectIt.effect("holds a starting run back while a provider install is in flight", () => + Effect.gen(function* () { + const installDone = yield* Deferred.make(); + const reads: Array = []; + const layer = ProviderTurnStart.layer.pipe( + Layer.provide( + Layer.mergeAll( + Layer.mock(ContextHandoffService.ContextHandoffServiceV2)({}), + Layer.mock(EventSink.EventSinkV2)({}), + IdAllocator.layer, + layerWorktreeLocation(), + FileSystem.layerNoop({}), + Layer.mock(GitWorkflow.GitWorkflowService)({}), + Layer.mock(ProjectService.ProjectService)({}), + Layer.mock(ProjectionStore.ProjectionStoreV2)({ + getThreadRecords: () => Effect.succeed({ thread: { worktreePath: null } } as never), + getTurnStartContext: () => + Effect.sync(() => { + reads.push("context"); + return { runs: [] } as never; + }), + }), + Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({}), + Layer.mock(ProviderAuthService.ProviderAuthService)({}), + Layer.mock(RunExecutionService.RunExecutionServiceV2)({}), + Layer.mock(RuntimePolicy.RuntimePolicyV2)({}), + Layer.mock(ProviderInstallLock.ProviderInstallLock)({ + untilIdle: Deferred.await(installDone), + observe: Effect.succeed(() => Effect.succeed(false)), + }), + ), + ), + ); + const start = Effect.gen(function* () { + yield* (yield* ProviderTurnStart.ProviderTurnStartServiceV2).start({ + threadId: ThreadId.make("thread_start_waits_for_install"), + runId: RunId.make("run_start_waits_for_install"), + }); + }).pipe(Effect.provide(layer)); + const fiber = yield* start.pipe(Effect.ignore, Effect.forkChild); + yield* Effect.yieldNow; + expect(reads).toEqual([]); + + yield* Deferred.succeed(installDone, undefined); + yield* Fiber.join(fiber); + expect(reads).toEqual(["context"]); + }), +); diff --git a/apps/server/src/orchestration-v2/ProviderTurnStartService.testkit.ts b/apps/server/src/orchestration-v2/ProviderTurnStartService.testkit.ts index fb3fb72fd76e3..28051ce6365f8 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnStartService.testkit.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnStartService.testkit.ts @@ -19,8 +19,9 @@ import * as Layer from "effect/Layer"; import * as GitWorkflow from "../git/GitWorkflowService.ts"; import * as ProjectService from "../project/ProjectService.ts"; +import * as ProviderInstallLock from "../provider/ProviderInstallLock.ts"; -export const layer = Layer.merge( +export const layer = Layer.mergeAll( Layer.mock(GitWorkflow.GitWorkflowService)({ pruneWorktrees: () => Effect.void, createWorktree: () => Effect.succeed({} as never), @@ -28,6 +29,10 @@ export const layer = Layer.merge( Layer.mock(ProjectService.ProjectService)({ getById: () => Effect.succeedNone, }), + Layer.mock(ProviderInstallLock.ProviderInstallLock)({ + untilIdle: Effect.void, + observe: Effect.succeed(() => Effect.succeed(false)), + }), ); // Keep repository detection, branch selection and worktree creation on the real Git path. diff --git a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts index 54b16d7aa2aef..50dda57be0fd1 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts @@ -6,6 +6,7 @@ import { type OrchestrationV2DomainEvent, type OrchestrationV2ContextHandoff, type OrchestrationV2ExecutionNode, + type OrchestrationV2ProviderFailure, type OrchestrationV2ProviderThread, type OrchestrationV2Run, type OrchestrationV2RunAttempt, @@ -24,12 +25,14 @@ import * as Option from "effect/Option"; import * as Path from "effect/Path"; import * as Predicate from "effect/Predicate"; import * as Schema from "effect/Schema"; +import * as Stream from "effect/Stream"; import * as ServerConfig from "../config.ts"; import { withWorkspaceLease } from "../workspace/workspaceLease.ts"; import * as GitWorkflowService from "../git/GitWorkflowService.ts"; import * as ProjectService from "../project/ProjectService.ts"; import * as ProviderAuthService from "../provider/ProviderAuthService.ts"; +import * as ProviderInstallLock from "../provider/ProviderInstallLock.ts"; import * as ServerSettings from "../serverSettings.ts"; import { makeIsManagedWorktreePath } from "../worktreesDirectory.ts"; import * as EventSink from "./EventSink.ts"; @@ -54,6 +57,7 @@ import { ContextHandoffBackgroundWorkError, knownProviderFailureMessage, makeProviderFailure, + ProviderUpdatingError, } from "@t3tools/provider-core/server/failure"; import * as RunExecutionService from "./RunExecutionService.ts"; import * as RuntimePolicy from "./RuntimePolicy.ts"; @@ -115,6 +119,10 @@ const refusedBeforePrompt = (error: unknown): boolean => PROMPT_REFUSAL_TAGS.some((tag) => Predicate.isTagged(error, tag)) || (Predicate.hasProperty(error, "cause") && refusedBeforePrompt(error.cause)); +// Claude reports a CLI that exits mid-launch as a transport error. +const isLaunchFailureClass = (failureClass: OrchestrationV2ProviderFailure["class"]) => + failureClass === "provider_error" || failureClass === "transport_error"; + export interface ProviderTurnStartServiceV2Shape { readonly start: (input: { readonly threadId: ThreadId; @@ -141,6 +149,7 @@ export const layer: Layer.Layer< | GitWorkflowService.GitWorkflowService | ProjectService.ProjectService | ProviderAuthService.ProviderAuthService + | ProviderInstallLock.ProviderInstallLock | ProjectionStore.ProjectionStoreV2 | ProviderSessionManager.ProviderSessionManagerV2 | RunExecutionService.RunExecutionServiceV2 @@ -157,6 +166,7 @@ export const layer: Layer.Layer< const gitWorkflow = yield* GitWorkflowService.GitWorkflowService; const projects = yield* ProjectService.ProjectService; const providerAuth = yield* ProviderAuthService.ProviderAuthService; + const installLock = yield* ProviderInstallLock.ProviderInstallLock; const projectionStore = yield* ProjectionStore.ProjectionStoreV2; const providerSessions = yield* ProviderSessionManager.ProviderSessionManagerV2; const runExecution = yield* RunExecutionService.RunExecutionServiceV2; @@ -341,6 +351,9 @@ export const layer: Layer.Layer< readonly willRetry?: boolean; }) { const { runId } = input; + // Another server can acquire an install lock after this wait completes. + yield* installLock.untilIdle; + let installedSinceAdmission = yield* installLock.observe; const projection = yield* projectionStore.getTurnStartContext(input.threadId, runId); const run = projection.runs.find((candidate) => candidate.id === runId); if (run === undefined) { @@ -786,7 +799,7 @@ export const layer: Layer.Layer< }); if (!repaired.committed) return; } - const sessionResult = yield* Effect.result( + const openSession = Effect.suspend(() => providerSessions.open({ threadId: projection.thread.id, providerSessionId, @@ -806,6 +819,16 @@ export const layer: Layer.Layer< }), }), ); + let sessionResult = yield* Effect.result(openSession); + if ( + sessionResult._tag === "Failure" && + (yield* installedSinceAdmission(providerThread.driver)) + ) { + yield* installLock.untilIdle; + if (!(yield* isCurrentAttemptInStatus("starting"))) return; + sessionResult = yield* Effect.result(openSession); + installedSinceAdmission = yield* installLock.observe; + } if (sessionResult._tag === "Failure") { const failure = sessionResult.failure; const requiresSignIn = @@ -820,7 +843,123 @@ export const layer: Layer.Layer< }); return; } - const session = sessionResult.success; + const openedSession = sessionResult.success; + let launching = true; + const installFailure = Effect.fn("ProviderTurnStartService.installFailure")(function* ( + error: ProviderAdapter.ProviderAdapterV2Error, + ) { + // Another client holding the native thread is not an install failure. + if ( + !launching || + error._tag === "ProviderAdapterThreadInUseError" || + !(yield* installedSinceAdmission(providerThread.driver)) + ) { + return error; + } + launching = false; + const updating = new ProviderUpdatingError({ + driver: providerThread.driver, + newNativeThread: providerThread.nativeThreadRef === null, + cause: error, + }); + // A prompt may already have reached the provider; release rather than replay it, but + // leave a shared session alone while another thread on it is still working. + yield* providerSessions + .releaseUnlessBusy({ + providerSessionId, + ownProviderThreadIds: [providerThread.id], + reason: "runtime_error", + detail: updating.message, + }) + .pipe(Effect.ignore); + return new ProviderAdapter.ProviderAdapterEventStreamError({ + driver: providerThread.driver, + providerSessionId, + cause: updating, + }); + }); + const guardLaunch = ( + effect: Effect.Effect, + initializes = false, + ) => + effect.pipe( + Effect.catch((error) => installFailure(error).pipe(Effect.flatMap(Effect.fail))), + Effect.tap(() => + Effect.sync(() => { + // Claude queues the prompt before its SDK reader initializes. + if (initializes && providerThread.driver !== "claudeAgent") launching = false; + }), + ), + ); + const guardEvents = (events: typeof openedSession.events) => + Stream.suspend(() => + events.pipe( + Stream.mapEffect((event) => + Effect.gen(function* () { + if ( + launching && + ((event.type === "turn.terminal" && + event.providerThreadId === providerThread.id && + event.runOrdinal === run.ordinal && + event.status === "failed" && + isLaunchFailureClass(event.failure.class)) || + (event.type === "turn_item.updated" && + event.turnItem.runId === run.id && + event.turnItem.type === "error" && + event.turnItem.retry === undefined && + isLaunchFailureClass(event.turnItem.failure.class))) && + (yield* installedSinceAdmission(providerThread.driver)) + ) { + const error = new ProviderAdapter.ProviderAdapterEventStreamError({ + driver: providerThread.driver, + providerSessionId, + cause: event, + }); + return yield* installFailure(error).pipe(Effect.flatMap(Effect.fail)); + } + if ( + (event.type === "turn.terminal" && + event.providerThreadId === providerThread.id && + event.runOrdinal === run.ordinal) || + (event.type === "turn_item.updated" && + event.turnItem.runId === run.id && + // A retrying API error means the provider is up and answering. + (event.turnItem.type === "error" + ? event.turnItem.retry !== undefined + : event.turnItem.type !== "system_notice")) + ) { + launching = false; + } + return event; + }), + ), + Stream.catch((error) => + Stream.unwrap(installFailure(error).pipe(Effect.map(Stream.fail))), + ), + ), + ); + const session: ProviderAdapter.ProviderAdapterV2SessionRuntime = { + ...openedSession, + ensureThread: (input) => guardLaunch(openedSession.ensureThread(input)), + resumeThread: (input) => guardLaunch(openedSession.resumeThread(input)), + startTurn: (input) => guardLaunch(openedSession.startTurn(input), true), + ...(openedSession.compactThread === undefined + ? {} + : { + compactThread: (input) => guardLaunch(openedSession.compactThread!(input), true), + }), + events: guardEvents(openedSession.events), + ...(openedSession.subscribeEvents === undefined + ? {} + : { + subscribeEvents: openedSession.subscribeEvents.pipe( + Effect.map((subscription) => ({ + ...subscription, + events: guardEvents(subscription.events), + })), + ), + }), + }; if (!(yield* isCurrentAttemptInStatus("starting"))) return; const loadFromProvider = Effect.fn("ProviderTurnStartService.loadFromProvider")(function* ( load: Effect.Effect, diff --git a/apps/server/src/orchestration-v2/RunExecutionService.test.ts b/apps/server/src/orchestration-v2/RunExecutionService.test.ts index 5bfb4fe34dc6a..a376f3ef9e9eb 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.test.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.test.ts @@ -3070,6 +3070,43 @@ it.effect.each([ }), ); +it.effect.each([false, true])( + "recovers an update failure with new native thread %s", + (newNativeThread) => + Effect.gen(function* () { + const nativeThreadRef = { + driver, + nativeId: "native:update-launch", + strength: "strong" as const, + }; + const { observed, finalEvents } = yield* captureRootRunTermination({ + key: `update-launch:${newNativeThread}`, + providerThread: { firstRunOrdinal: 1, nativeThreadRef }, + shouldFinalizeRun: () => Effect.succeed(true), + events: (ids) => + Stream.make({ + type: "turn.terminal", + status: "failed", + driver, + providerThreadId: ids.providerThreadId, + providerTurnId: ids.rootProviderTurnId, + runOrdinal: 1, + failureItemOrdinal: 101, + threadDisposition: "reusable", + failure: { + class: "provider_error", + message: "Codex was updating on this machine. Try again.", + code: newNativeThread ? "provider_updating_new_thread" : "provider_updating", + retryable: null, + }, + }), + }); + assert.deepEqual(observed, ["run:failed", "pull-requests-refreshed"]); + const thread = finalEvents.find((event) => event.type === "provider-thread.updated")?.payload; + assert.deepEqual(thread?.nativeThreadRef, newNativeThread ? null : nativeThreadRef); + }), +); + it.effect("fails the run when its event stream ends without a root terminal", () => Effect.gen(function* () { const { observed, finalEvents } = yield* captureRootRunTermination({ @@ -3136,6 +3173,7 @@ it.effect("keeps completed runs completed when pull request refresh fails", () = function captureRootRunTermination(input: { readonly key: string; + readonly providerThread?: Partial; readonly shouldFinalizeRun: () => Effect.Effect; readonly rejectTerminalWrite?: boolean; readonly hasUnpairedRunInterruptRequest?: () => Effect.Effect; @@ -3313,6 +3351,7 @@ function captureRootRunTermination(input: { providerThread: { id: ids.providerThreadId, driver, + ...input.providerThread, } as OrchestrationV2ProviderThread, attempt: { id: ids.attemptId, diff --git a/apps/server/src/orchestration-v2/RunExecutionService.ts b/apps/server/src/orchestration-v2/RunExecutionService.ts index 66c0d2143e9aa..9d2150377a24e 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.ts @@ -730,6 +730,10 @@ export const layer: Layer.Layer< const finalizedProviderThread: OrchestrationV2ProviderThread = { ...input.providerThread, status: finalProviderThreadStatus(input.terminal.threadDisposition), + ...(input.terminal.status === "failed" && + input.terminal.failure.code === "provider_updating_new_thread" + ? { nativeThreadRef: null, nativeMetadata: null } + : {}), updatedAt: completedAt, }; const runEventId = yield* allocateEventId(); diff --git a/apps/server/src/orchestration-v2/runtimeLayer.ts b/apps/server/src/orchestration-v2/runtimeLayer.ts index ba1dadad1864c..58a4650220165 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.ts @@ -45,6 +45,7 @@ import * as CommandOutputHub from "./CommandOutputHub.ts"; import * as ProviderRuntimeRecoveryService from "./ProviderRuntimeRecoveryService.ts"; import * as ProviderSwitchService from "./ProviderSwitchService.ts"; import * as ProviderTurnControlService from "./ProviderTurnControlService.ts"; +import * as ProviderInstallLock from "../provider/ProviderInstallLock.ts"; import * as ProviderTurnStartService from "./ProviderTurnStartService.ts"; import * as RunExecutionService from "./RunExecutionService.ts"; import * as RunFinalizationService from "./RunFinalizationService.ts"; @@ -169,6 +170,7 @@ const layerProviderTurnStartServiceProvided = ProviderTurnStartService.layer.pip layerEventSink, IdAllocator.layer, ProjectionStore.layer, + ProviderInstallLock.layer, layerProviderSessionManagerProvided, layerProviderAuthServiceProvided, layerRunExecutionServiceProvided, diff --git a/apps/server/src/provider/ProviderInstallLock.ts b/apps/server/src/provider/ProviderInstallLock.ts new file mode 100644 index 0000000000000..2cdbbc9fde6fb --- /dev/null +++ b/apps/server/src/provider/ProviderInstallLock.ts @@ -0,0 +1,369 @@ +import * as HostProcess from "@t3tools/shared/HostProcess"; +import * as Clock from "effect/Clock"; +import * as Context from "effect/Context"; +import * as SecureRandom from "effect/crypto/SecureRandom"; +import * as Duration from "effect/Duration"; +import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as Path from "effect/Path"; +import type { PlatformError } from "effect/PlatformError"; +import * as Ref from "effect/Ref"; +import * as Schema from "effect/Schema"; + +// The wrapper's go deadline + install backstop stay below this cap; the +// generation handshake fences owners paused before creating their wrapper. +const MAX_LOCK_AGE = Duration.hours(1); +const IDLE_POLL_INTERVAL = Duration.seconds(1); + +export class ProviderInstallLockError extends Schema.TaggedError()( + "ProviderInstallLockError", + { cause: Schema.Defect() }, +) { + override get message(): string { + return "Could not use the provider install lock directory."; + } +} + +/** + * A lock on running a provider installer, shared by every T3 server on this + * host. Two servers must never run `npm i -g` on the same prefix at once. + */ +export class ProviderInstallLock extends Context.Service< + ProviderInstallLock, + { + /** + * Runs `run` while holding the lock for `lockKey`, or returns `None` without + * running when another process holds it. `run` calls `recordChildPid` with the + * installer's pid: a detached installer outlives this server, and the lock + * stays held until both `run` completes and the installer exits. + */ + readonly tryRun: ( + lockKey: string, + run: ( + recordChildPid: (pid: number) => Effect.Effect, + ) => Effect.Effect, + provider?: string, + ) => Effect.Effect, E | ProviderInstallLockError, R>; + /** Completes once no installer holds any lock, or when the lock directory cannot be read. */ + readonly untilIdle: Effect.Effect; + /** Whether any lock is held right now. A failed read is an error, not "free". */ + readonly anyHeld: Effect.Effect; + /** Detects installers held now or published since this snapshot, including completed installs. */ + readonly observe: Effect.Effect<(provider: string) => Effect.Effect>; + } +>()("t3/provider/ProviderInstallLock") {} + +const isProviderInstallLockError = Schema.is(ProviderInstallLockError); + +function isPidAlive(pid: number): boolean { + try { + process.kill(pid, 0); + return true; + } catch (cause) { + // EPERM: the process exists but belongs to someone else. + return (cause as NodeJS.ErrnoException).code === "EPERM"; + } +} + +function parseGeneration(name: string) { + const match = /^(.+\.lock)\.([1-9]\d*)$/.exec(name); + if (!match) return undefined; + const generation = Number(match[2]); + return Number.isSafeInteger(generation) ? { key: match[1]!, generation, name } : undefined; +} + +const make = Effect.gen(function* () { + const fileSystem = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const random = yield* SecureRandom.SecureRandom; + const environment = yield* HostProcess.Environment; + const homeDirectory = yield* HostProcess.HomeDirectory; + const cacheHome = environment.XDG_CACHE_HOME || path.join(homeDirectory, ".cache"); + const directory = path.join(cacheHome, "t3", "provider-install"); + + const toLockError = (cause: unknown) => + isProviderInstallLockError(cause) ? cause : new ProviderInstallLockError({ cause }); + const ignoreNotFound = + (fallback: B) => + (effect: Effect.Effect): Effect.Effect => + effect.pipe( + Effect.catchTags({ + PlatformError: (error) => + error.reason._tag === "NotFound" ? Effect.succeed(fallback) : Effect.fail(error), + }), + ); + + const readRecord = Effect.fn("ProviderInstallLock.readRecord")(function* (file: string) { + const info = yield* fileSystem.stat(file); + const modifiedAt = Option.match(info.mtime, { + onNone: () => 0, + onSome: (date) => date.getTime(), + }); + const expired = + (yield* Clock.currentTimeMillis) - modifiedAt >= Duration.toMillis(MAX_LOCK_AGE); + // The first line holds pids; the second identifies the owning run, including within one server. + const contents = yield* fileSystem.readFileString(file); + const [pids = ""] = contents.split("\n"); + const recorded = pids.split(" ").map(Number); + const valid = (pid: number) => Number.isSafeInteger(pid) && pid > 0; + const live = recorded.some((pid) => valid(pid) && isPidAlive(pid)); + return { + contents, + modifiedAt, + held: !expired && pids !== "released" && (live || recorded.some((pid) => !valid(pid))), + expiredLive: expired && live, + provider: contents.split("\n")[2], + }; + }); + const isHeld = (file: string) => readRecord(file).pipe(Effect.map((record) => record.held)); + + /** Publishes a complete record exclusively; a paused writer must never look abandoned. */ + const createExclusively = Effect.fn("ProviderInstallLock.createExclusively")(function* ( + file: string, + contents: string, + owner: string, + ) { + const temporary = `${file}.${owner}.tmp`; + return yield* Effect.gen(function* () { + const handle = yield* fileSystem.open(temporary, { flag: "wx" }); + yield* handle.writeAll(new TextEncoder().encode(contents)); + }).pipe( + Effect.scoped, + // Unlike rename, link cannot overwrite another contender's generation. + Effect.andThen(fileSystem.link(temporary, file)), + Effect.as(true), + Effect.ensuring(fileSystem.remove(temporary, { force: true }).pipe(Effect.ignore)), + Effect.catchTags({ + PlatformError: (error) => + error.reason._tag === "AlreadyExists" ? Effect.succeed(false) : Effect.fail(error), + }), + ); + }); + + const generations = Effect.gen(function* () { + const names = yield* fileSystem.readDirectory(directory).pipe(ignoreNotFound([])); + return names.flatMap((name) => { + const parsed = parseGeneration(name); + return parsed ? [parsed] : []; + }); + }); + + const replace = Effect.fn("ProviderInstallLock.replace")(function* ( + file: string, + next: string, + owner: string, + ) { + const temporary = `${file}.${owner}.tmp`; + return yield* fileSystem.writeFileString(temporary, next).pipe( + Effect.andThen(fileSystem.rename(temporary, file)), + Effect.onError(() => fileSystem.remove(temporary, { force: true }).pipe(Effect.ignore)), + ); + }); + const historyFile = (provider: string) => + path.join(directory, `${encodeURIComponent(provider)}.installed`); + + /** + * Only exclusive creation elects an owner. The highest generation is retained + * even after release, so a delayed reclaimer cannot reuse its number. + */ + const acquire = Effect.fn("ProviderInstallLock.acquire")(function* ( + key: string, + contents: string, + owner: string, + ) { + yield* fileSystem.makeDirectory(directory, { recursive: true }); + // Older servers wrote an unnumbered lock with the same age cap. + if (yield* isHeld(path.join(directory, key)).pipe(ignoreNotFound(false))) return undefined; + const previous = (yield* generations).filter((entry) => entry.key === key); + const highest = previous.reduce((max, entry) => Math.max(max, entry.generation), 0); + const previousFile = highest > 0 ? path.join(directory, `${key}.${highest}`) : undefined; + const previousRecord = + previousFile === undefined + ? undefined + : yield* readRecord(previousFile).pipe(ignoreNotFound(undefined)); + if (previousRecord?.held) return undefined; + const next = highest + 1; + const file = path.join(directory, `${key}.${next}`); + if (!(yield* createExclusively(file, contents, owner))) return undefined; + return yield* Effect.gen(function* () { + // Our directory snapshot may predate a successor that already pruned this + // number. A recreated lower generation must never run an installer. + if ((yield* generations).some((entry) => entry.key === key && entry.generation > next)) { + yield* fileSystem.remove(file, { force: true }); + return undefined; + } + if ( + previousFile !== undefined && + (previousRecord === undefined || previousRecord.expiredLive) + ) { + // On one host these fs operations are linearizable: owner writes N then + // reads the directory; successor links N+1 then reads N. At least one + // sees the other, so both can back off but cannot both start installers. + const current = yield* readRecord(previousFile).pipe(ignoreNotFound(undefined)); + if ( + current !== undefined && + (previousRecord === undefined || + current.contents !== previousRecord.contents || + current.modifiedAt !== previousRecord.modifiedAt) + ) { + // Restore N as highest; a released N+1 would hide its resumed owner. + yield* fileSystem.remove(file, { force: true }); + return undefined; + } + } + // Publish before scanning: concurrent owners of different keys cannot both miss each other. + if (yield* heldExcept(file)) { + yield* fileSystem.remove(file, { force: true }); + return undefined; + } + // A rejected successor can disappear after another contender snapshots it. + // That contender must not skip the resumed owner below the missing generation. + for (const entry of previous) { + if ( + entry.generation < highest && + (yield* isHeld(path.join(directory, entry.name)).pipe(ignoreNotFound(false))) + ) { + yield* fileSystem.remove(file, { force: true }); + return undefined; + } + } + const provider = contents.split("\n")[2]; + // A shared install key can change providers when its old generations are pruned. + if (provider) yield* replace(historyFile(provider), owner, owner); + yield* Effect.forEach( + previous, + (entry) => + fileSystem.remove(path.join(directory, entry.name), { force: true }).pipe(Effect.ignore), + { discard: true }, + ); + return file; + }).pipe(Effect.onError(() => fileSystem.remove(file, { force: true }).pipe(Effect.ignore))); + }); + + const tryRun = Effect.fn("ProviderInstallLock.tryRun")(function* ( + lockKey: string, + run: ( + recordChildPid: (pid: number) => Effect.Effect, + ) => Effect.Effect, + provider?: string, + ): Effect.fn.Return, E | ProviderInstallLockError, R> { + const key = `${encodeURIComponent(lockKey)}.lock`; + const owner = yield* random.uuidV4().pipe(Effect.mapError(toLockError)); + const contents = `${process.pid}\n${owner}\n${provider ?? ""}`; + // Written aside and renamed over the lock, so a failed write (disk full) + // leaves the previous complete record instead of an empty, held file. + return yield* Effect.acquireUseRelease( + acquire(key, contents, owner).pipe( + Effect.mapError(toLockError), + Effect.flatMap((acquired) => + acquired !== undefined + ? Ref.make<{ file: string; pid: number | undefined; contents: string }>({ + file: acquired, + pid: undefined, + contents, + }).pipe(Effect.asSome) + : Effect.succeedNone, + ), + ), + (ownerState) => + Option.isNone(ownerState) + ? Effect.succeedNone + : run((pid) => + Effect.gen(function* () { + const state = yield* Ref.get(ownerState.value); + if ((yield* fileSystem.readFileString(state.file)) !== state.contents) { + return yield* toLockError(new Error("Provider install lock ownership changed")); + } + const next = `${process.pid} ${pid}\n${owner}\n${provider ?? ""}`; + yield* replace(state.file, next, owner); + yield* Ref.set(ownerState.value, { ...state, pid, contents: next }); + const generation = parseGeneration(path.basename(state.file))!; + if ( + (yield* generations).some( + (entry) => entry.key === key && entry.generation > generation.generation, + ) + ) { + return yield* toLockError(new Error("Provider install lock ownership expired")); + } + }).pipe(Effect.mapError(toLockError), Effect.uninterruptible), + ).pipe(Effect.asSome), + (ownerState) => + Option.isNone(ownerState) + ? Effect.void + : Ref.get(ownerState.value).pipe( + Effect.flatMap((state) => + Effect.gen(function* () { + if ((yield* fileSystem.readFileString(state.file)) !== state.contents) return; + // Once `run` ends, only a surviving detached installer keeps the lock held. + yield* replace( + state.file, + `${state.pid !== undefined && isPidAlive(state.pid) ? state.pid : "released"}\n${owner}\n${provider ?? ""}`, + owner, + ); + }), + ), + Effect.ignore, + ), + ); + }); + + const heldExcept = Effect.fn("ProviderInstallLock.heldExcept")(function* ( + excluded: string | undefined, + ) { + const names = yield* fileSystem.readDirectory(directory).pipe(ignoreNotFound([])); + // A provisional successor must not hide a resumed older owner from turn starts. + for (const name of names) { + if (name.endsWith(".lock") || parseGeneration(name) !== undefined) { + const file = path.join(directory, name); + if (file !== excluded && (yield* isHeld(file).pipe(ignoreNotFound(false)))) return true; + } + } + return false; + }, Effect.mapError(toLockError)); + const anyHeld = heldExcept(undefined); + + const untilIdle: Effect.Effect = Effect.suspend(() => + anyHeld.pipe( + Effect.orElseSucceed(() => false), + Effect.flatMap((held) => + held ? Effect.sleep(IDLE_POLL_INTERVAL).pipe(Effect.andThen(untilIdle)) : Effect.void, + ), + ), + ); + + const observe = Effect.gen(function* () { + const history = new Map(); + for (const name of yield* fileSystem.readDirectory(directory).pipe(ignoreNotFound([]))) { + if (!name.endsWith(".installed")) continue; + const record = yield* fileSystem + .readFileString(path.join(directory, name)) + .pipe(ignoreNotFound(undefined)); + if (record !== undefined) history.set(name, record); + } + const before = new Set((yield* generations).map((entry) => entry.name)); + return (provider: string) => + Effect.gen(function* () { + const file = historyFile(provider); + const record = yield* fileSystem.readFileString(file).pipe(ignoreNotFound(undefined)); + if (record !== undefined && record !== history.get(path.basename(file))) return true; + for (const entry of yield* generations) { + const record = yield* readRecord(path.join(directory, entry.name)).pipe( + ignoreNotFound(undefined), + ); + if ( + record !== undefined && + record.provider === provider && + (record.held || !before.has(entry.name)) + ) + return true; + } + return false; + }).pipe(Effect.orElseSucceed(() => false)); + }).pipe(Effect.orElseSucceed(() => (_provider: string) => Effect.succeed(false))); + + return ProviderInstallLock.of({ tryRun, untilIdle, anyHeld, observe }); +}); + +export const layer = Layer.effect(ProviderInstallLock, make); diff --git a/apps/server/src/provider/agentContextBudget.json b/apps/server/src/provider/agentContextBudget.json index 50ce63aa8a2b3..f98ece6eb95a2 100644 --- a/apps/server/src/provider/agentContextBudget.json +++ b/apps/server/src/provider/agentContextBudget.json @@ -1,5 +1,5 @@ { "claudeSystemAppendBytes": 5547, "codexAdditionalContextBytes": 7277, - "mcpToolListJsonBytes": 48217 + "mcpToolListJsonBytes": 48255 } diff --git a/apps/server/src/provider/providerMaintenanceRunner.test.ts b/apps/server/src/provider/providerMaintenanceRunner.test.ts index adf53c5267203..8ce35ddf405ba 100644 --- a/apps/server/src/provider/providerMaintenanceRunner.test.ts +++ b/apps/server/src/provider/providerMaintenanceRunner.test.ts @@ -1,27 +1,48 @@ +// @effect-diagnostics nodeBuiltinImport:off - real detached installers and lock files are Node boundaries import { describe, it, assert } from "@effect/vitest"; import { ProviderDriverKind, ProviderInstanceId, + ThreadId, type ServerProvider, type ServerProviderUpdateState, } from "@t3tools/contracts"; import { ServerProviderUpdateError } from "@t3tools/contracts"; import * as Cause from "effect/Cause"; import * as Deferred from "effect/Deferred"; +import * as Duration from "effect/Duration"; import * as Effect from "effect/Effect"; import * as Exit from "effect/Exit"; import * as Fiber from "effect/Fiber"; import * as Layer from "effect/Layer"; import * as Ref from "effect/Ref"; import * as Schema from "effect/Schema"; -import * as Sink from "effect/Sink"; import * as Stream from "effect/Stream"; import * as TestClock from "effect/testing/TestClock"; import { HttpClient, HttpClientResponse } from "effect/http"; -import { ChildProcessSpawner } from "effect/process"; +import * as NodeCrypto from "@effect/platform-node/NodeCrypto"; +import * as NodeServices from "@effect/platform-node/NodeServices"; import * as HostProcess from "@t3tools/shared/HostProcess"; import { SpawnExecutableResolution } from "@t3tools/shared/shell"; - +import * as FileSystem from "effect/FileSystem"; +import * as Option from "effect/Option"; +import * as PlatformError from "effect/PlatformError"; +import * as Logger from "effect/Logger"; +import * as Scope from "effect/Scope"; +import * as NodeChildProcess from "node:child_process"; +import * as NodeEvents from "node:events"; +import * as NodePath from "node:path"; +import * as NodeStream from "node:stream"; +import * as NodeVM from "node:vm"; + +import * as BackgroundPolicy from "../background/BackgroundPolicy.ts"; +import * as ServerConfig from "../config.ts"; +import * as ProjectionStore from "../orchestration-v2/ProjectionStore.ts"; +import * as ServerSettings from "../serverSettings.ts"; +import * as ProviderAutoUpdater from "../updates/ProviderAutoUpdater.ts"; +import * as UpdateWindow from "../updates/UpdateWindow.ts"; + +import * as ProviderInstallLock from "./ProviderInstallLock.ts"; import * as ProviderRegistry from "./ProviderRegistry.ts"; import * as ModelManifest from "./ModelManifest.ts"; import * as ProviderMaintenanceRunner from "./providerMaintenanceRunner.ts"; @@ -32,6 +53,7 @@ import { type ProviderMaintenanceCapabilities, } from "@t3tools/provider-core/server/maintenanceResolver"; const isServerProviderUpdateError = Schema.is(ServerProviderUpdateError); +const isProviderInstallLockError = Schema.is(ProviderInstallLock.ProviderInstallLockError); const CODEX_DRIVER = ProviderDriverKind.make("codex"); const NATIVE_CLI_DRIVER = ProviderDriverKind.make("nativeCli"); @@ -39,7 +61,6 @@ const OPENCODE_DRIVER = ProviderDriverKind.make("opencode"); const CODEX_INSTANCE_ID = ProviderInstanceId.make("codex"); const NATIVE_CLI_INSTANCE_ID = ProviderInstanceId.make("nativeCli"); const OPENCODE_INSTANCE_ID = ProviderInstanceId.make("opencode"); -const encoder = new TextEncoder(); // Pin Linux so raw command/args assertions bypass Windows shim resolution. const layerNonWindowsPlatform = Layer.succeed(HostProcess.Platform, "linux"); @@ -105,27 +126,7 @@ const layerLatestVersionHttpClient = (version: string) => ), ); -function mockHandle(result: { - readonly stdout?: string; - readonly stderr?: string; - readonly code?: number; - readonly exitCode?: Effect.Effect; -}) { - return ChildProcessSpawner.makeHandle({ - pid: ChildProcessSpawner.ProcessId(1), - exitCode: result.exitCode ?? Effect.succeed(ChildProcessSpawner.ExitCode(result.code ?? 0)), - isRunning: Effect.succeed(false), - kill: () => Effect.void, - unref: Effect.succeed(Effect.void), - stdin: Sink.drain, - stdout: Stream.make(encoder.encode(result.stdout ?? "")), - stderr: Stream.make(encoder.encode(result.stderr ?? "")), - all: Stream.empty, - getInputFd: () => Sink.drain, - getOutputFd: () => Stream.empty, - }); -} - +/** Stands in for the detached installer: `handler` decides what the installer does. */ function layerMockSpawner( handler: ( command: string, @@ -135,22 +136,25 @@ function layerMockSpawner( readonly stdout?: string; readonly stderr?: string; readonly code?: number; - readonly exitCode?: Effect.Effect; + readonly exitCode?: Effect.Effect; + readonly progress?: ReadonlyArray; }, ) { - return Layer.succeed( - ChildProcessSpawner.ChildProcessSpawner, - ChildProcessSpawner.make((command) => { - const childProcess = command as unknown as { - readonly command: string; - readonly args: ReadonlyArray; - readonly options: { readonly env?: NodeJS.ProcessEnv | undefined }; - }; - return Effect.succeed( - mockHandle(handler(childProcess.command, childProcess.args, childProcess.options)), - ); - }), - ); + return Layer.succeed(ProviderMaintenanceRunner.ProviderInstaller, { + run: (input) => + Effect.gen(function* () { + const result = handler(input.command, input.args, { env: input.env }); + for (const line of result.progress ?? []) yield* input.onProgress?.(line) ?? Effect.void; + return { + stdout: result.stdout ?? "", + stderr: result.stderr ?? "", + exitCode: result.exitCode ? yield* result.exitCode : (result.code ?? 0), + timedOut: false, + stdoutTruncated: false, + stderrTruncated: false, + }; + }), + }); } function makeRegistry( @@ -227,17 +231,49 @@ function makeRegistry( }); } +/** Locks live under `/t3/provider-install`; servers sharing a home share their locks. */ +// The install lock needs SecureRandom, which NodeServices does not provide. +const NodeTestServices = Layer.mergeAll(NodeServices.layer, NodeCrypto.layer); + +const layerLockAt = (cacheHome: string) => + ProviderInstallLock.layer.pipe( + Layer.provide( + Layer.mergeAll( + NodeServices.layer, + NodeCrypto.layer, + Layer.succeed(HostProcess.Environment, { XDG_CACHE_HOME: cacheHome }), + ), + ), + ); + +// Generic updater fixtures use synthetic versions. Keep their compatibility +// unknown so real harness minimums do not bypass the command under test. +const defaultManifest: ModelManifest.ModelManifestData = { + version: 1, + currentModels: {}, + compatibility: [CODEX_DRIVER, OPENCODE_DRIVER].map((driver) => ({ + driver, + t3CodeRange: ">=0.0.42", + ranges: [], + })), +}; + const makeTestRunner = ( registry: ProviderRegistry.ProviderRegistry["Service"], - manifest: ModelManifest.ModelManifestData = { - version: 1, - currentModels: {}, - compatibility: [CODEX_DRIVER, OPENCODE_DRIVER].map((driver) => ({ - driver, - t3CodeRange: ">=0.0.42", - ranges: [], - })), - }, + manifest: ModelManifest.ModelManifestData = defaultManifest, + latestVersions: Layer.Layer = ProviderLatestVersions.layer, +) => + Effect.gen(function* () { + const lockHome = yield* FileSystem.FileSystem.use((fileSystem) => + fileSystem.makeTempDirectoryScoped({ prefix: "t3-install-lock-" }), + ).pipe(Effect.provide(NodeTestServices)); + return yield* makeRunnerWithLock(registry, manifest, layerLockAt(lockHome), latestVersions); + }); + +const makeRunnerWithLock = ( + registry: ProviderRegistry.ProviderRegistry["Service"], + manifest: ModelManifest.ModelManifestData, + lock: Layer.Layer, latestVersions: Layer.Layer = ProviderLatestVersions.layer, ) => Effect.service(ProviderMaintenanceRunner.ProviderMaintenanceRunner).pipe( @@ -245,6 +281,7 @@ const makeTestRunner = ( ProviderMaintenanceRunner.layer.pipe( Layer.provide( Layer.mergeAll( + lock, Layer.succeed(ProviderRegistry.ProviderRegistry, registry), Layer.succeed(ModelManifest.ModelManifest, { current: Effect.succeed(manifest), @@ -766,7 +803,6 @@ describe("providerMaintenanceRunner", () => { // binary behind that later updates mistake for an install. it.effect.each([ { runFor: "6 minutes", status: "succeeded", message: "Provider updated." }, - { runFor: "31 minutes", status: "failed", message: "Update timed out after 30 minutes." }, ] as const)("lets a slow installer run for $runFor", ({ runFor, status, message }) => { const started = Deferred.makeUnsafe(); return Effect.gen(function* () { @@ -788,7 +824,7 @@ describe("providerMaintenanceRunner", () => { layerMockSpawner(() => ({ exitCode: Deferred.succeed(started, undefined).pipe( Effect.andThen(Effect.sleep(runFor)), - Effect.as(ChildProcessSpawner.ExitCode(0)), + Effect.as(0), ), })), ), @@ -833,9 +869,7 @@ describe("providerMaintenanceRunner", () => { startedLatch.resolve(); return { stdout: "updated", - exitCode: Effect.promise(() => release).pipe( - Effect.as(ChildProcessSpawner.ExitCode(0)), - ), + exitCode: Effect.promise(() => release).pipe(Effect.as(0)), }; }), ), @@ -897,48 +931,100 @@ describe("providerMaintenanceRunner", () => { ), ); - it.effect("resolves npm to a .cmd shim from the provider's PATH on win32", () => { + it.effect("gates a detached wrapper carrying the resolved win32 .cmd command", () => { + const recorded: Array = []; const captured: Array<{ readonly command: string; readonly args: ReadonlyArray; - readonly shell: boolean | string | undefined; - readonly env: NodeJS.ProcessEnv | undefined; + readonly options: NodeChildProcess.SpawnOptions; }> = []; - return Effect.gen(function* () { - const { registry } = yield* makeRegistry(baseProvider, { - installedVersionAfterUpdate: "1.0.0", + const spawnProcess = (( + command: string, + args: ReadonlyArray, + options: NodeChildProcess.SpawnOptions, + ) => { + captured.push({ command, args, options }); + const stdin = new NodeStream.Writable({ + write(chunk, _encoding, callback) { + assert.deepStrictEqual(recorded, [4242]); + assert.strictEqual(chunk.toString(), "g"); + callback(); + }, }); - const runner = yield* makeTestRunner({ - ...registry, - getProviderMaintenanceCapabilitiesForInstance: () => - Effect.succeed( - makeProviderMaintenanceCapabilities({ - provider: CODEX_DRIVER, - packageName: "@openai/codex", - updateExecutable: "npm", - updateArgs: ["install", "-g", "@openai/codex@latest"], - updateLockKey: "npm-global", - env: { PATH: "C:\\fake\\npm" }, - }), - ), + const child = Object.assign(new NodeEvents.EventEmitter(), { + pid: 4242, + stdin, + unref: () => {}, }); + stdin.once("finish", () => child.emit("exit", 0)); + setImmediate(() => { + child.emit("spawn"); + }); + return child; + }) as unknown as typeof NodeChildProcess.spawn; + return Effect.gen(function* () { + const result = yield* ProviderMaintenanceRunner.runDetachedInstall( + { + command: "npm", + args: ["install", "-g", "@openai/codex@latest"], + env: { PATH: "C:\\fake\\npm" }, + recordChildPid: (pid) => Effect.sync(() => void recorded.push(pid)), + }, + spawnProcess, + ); - const result = yield* runner.updateProvider(CODEX_DRIVER); - + // resolveSpawnCommand resolves `npm` to the `.cmd` shim and routes it through cmd.exe. assert.strictEqual(captured.length, 1); const call = captured[0]; assert.ok(call, "expected the spawner to be invoked once"); assert.strictEqual( - call.command, + JSON.parse(call.args[2]!).command, '^"C:\\fake\\npm\\npm.cmd^" ^"install^" ^"-g^" ^"@openai/codex@latest^"', ); - assert.deepEqual(call.args, []); - assert.strictEqual(call.shell, true); - assert.deepStrictEqual(call.env, { PATH: "C:\\fake\\npm" }); - assert.strictEqual(result.providers[0]?.updateState?.status, "succeeded"); + assert.strictEqual(call.command, process.execPath); + assert.strictEqual(call.options.shell, undefined); + assert.strictEqual(JSON.parse(call.args[2]!).shell, true); + assert.strictEqual(call.options.detached, true); + assert.strictEqual(call.options.windowsHide, true); + assert.deepEqual(JSON.parse(call.args[2]!).args, []); + assert.strictEqual(call.options.stdio?.[0], "pipe"); + assert.deepStrictEqual(recorded, [4242]); + assert.strictEqual(result.exitCode, 0); + const stdin = Object.assign(new NodeEvents.EventEmitter(), { destroy() {} }); + const installer = Object.assign(new NodeEvents.EventEmitter(), { + stdout: new NodeEvents.EventEmitter(), + stderr: new NodeEvents.EventEmitter(), + }); + const timers = new Set<() => void>(); + const wrapperProcess = { + argv: ["node", call.args[2]!], + env: {}, + stdin, + platform: "linux", + pid: 4242, + exitCode: undefined as number | undefined, + }; + NodeVM.runInNewContext(call.args[1]!, { + process: wrapperProcess, + performance: { now: () => 0 }, + require: (name: string) => + name === "node:child_process" ? { spawn: () => installer } : {}, + setTimeout: (callback: () => void) => { + timers.add(callback); + return callback; + }, + clearTimeout: (callback: () => void) => timers.delete(callback), + }); + stdin.emit("data", Buffer.from("g")); + installer.emit("exit", 0); + assert.strictEqual(timers.size, 1, "descendants holding pipes must retain the backstop"); + assert.strictEqual(wrapperProcess.exitCode, 0); + installer.emit("close", 0); + assert.strictEqual(timers.size, 0); }).pipe( Effect.provide( Layer.mergeAll( + NodeServices.layer, Layer.succeed(HostProcess.Platform, "win32"), // Only the provider's PATH has npm. Layer.succeed(HostProcess.Environment, { @@ -950,33 +1036,987 @@ describe("providerMaintenanceRunner", () => { ? "C:\\fake\\npm\\npm.cmd" : undefined, ), - layerLatestVersionHttpClient("1.0.0"), - Layer.succeed( - ChildProcessSpawner.ChildProcessSpawner, - ChildProcessSpawner.make((command) => { - const childProcess = command as unknown as { - readonly command: string; - readonly args: ReadonlyArray; - readonly options: { - readonly shell?: boolean | string | undefined; - readonly env?: NodeJS.ProcessEnv | undefined; - }; - }; - captured.push({ - command: childProcess.command, - args: childProcess.args, - shell: childProcess.options.shell, - env: childProcess.options.env, - }); - return Effect.succeed(mockHandle({ stdout: "updated" })); + ), + ), + ); + }); + + it.live("runs a native installer without external Node in a standalone server", () => + Effect.gen(function* () { + const installer = yield* ProviderMaintenanceRunner.ProviderInstaller; + const recorded: Array = []; + const result = yield* installer.run({ + command: "/bin/sh", + args: ["-c", "printf native-updated; printf native-progress >&2"], + env: { PATH: "/no-node" }, + recordChildPid: (pid) => Effect.sync(() => void recorded.push(pid)), + }); + assert.strictEqual(result.exitCode, 0); + assert.strictEqual(result.stdout, "native-updated"); + assert.strictEqual(result.stderr, "native-progress"); + assert.strictEqual(recorded.length, 1); + assert.isAbove(recorded[0]!, 0); + }).pipe( + Effect.provide(ProviderMaintenanceRunner.layerInstaller), + Effect.provideService(HostProcess.IsExecutable, true), + Effect.provide(NodeTestServices), + ), + ); + + describe("background installs", () => { + const npmCapabilities = (lockKey = "npm-global:/fixture") => + makeProviderMaintenanceCapabilities({ + provider: CODEX_DRIVER, + packageName: "@openai/codex", + updateExecutable: "npm", + updateArgs: ["install", "-g", "@openai/codex@latest"], + updateLockKey: lockKey, + }); + /** A registry whose codex version follows `version()`, so an install can change it. */ + const makeVersionedRegistry = ( + version: () => string, + capabilities: ProviderMaintenanceCapabilities = npmCapabilities(), + ) => + Effect.gen(function* () { + const { registry, updateStatesRef } = yield* makeRegistry(); + const current = Effect.sync(() => [{ ...baseProvider, version: version() }]); + return { + updateStatesRef, + registry: { + ...registry, + getProviders: current, + refreshInstance: () => current, + getProviderMaintenanceCapabilitiesForInstance: () => Effect.succeed(capabilities), + }, + }; + }); + const logs: Array = []; + const layerCaptureLogs = Logger.layer([ + Logger.make(({ message }) => void logs.push(JSON.stringify(message))), + ]); + + it.effect("installs the pinned release silently and publishes no update state", () => { + const calls: Array = []; + let installed = false; + return Effect.gen(function* () { + const { registry, updateStatesRef } = yield* makeVersionedRegistry(() => + installed ? "9.9.9" : "0.0.0", + ); + const runner = yield* makeTestRunner(registry); + + const result = yield* runner.updateProvider({ + provider: CODEX_DRIVER, + targetVersion: "9.9.9", + background: true, + }); + + assert.deepStrictEqual(calls, ["install -g @openai/codex@9.9.9"]); + assert.deepStrictEqual(yield* Ref.get(updateStatesRef), []); + assert.strictEqual(result.providers[0]?.updateState, undefined); + }).pipe( + Effect.provide( + Layer.mergeAll( + layerNonWindowsPlatform, + layerLatestVersionHttpClient("9.9.9"), + layerMockSpawner((_command, args) => { + calls.push(args.join(" ")); + installed = true; + return {}; }), ), ), - ), + ); + }); + + it.effect("skips when the installed version already reaches the target", () => { + const calls: Array = []; + return Effect.gen(function* () { + for (const installedVersion of ["9.9.9", "10.0.0-beta.1"]) { + const { registry } = yield* makeVersionedRegistry(() => installedVersion); + const runner = yield* makeTestRunner(registry); + yield* runner.updateProvider({ + provider: CODEX_DRIVER, + targetVersion: "9.9.9", + background: true, + }); + } + // OpenCode reports its helper as `version` and the CLI as `installedVersion`. + const { registry } = yield* makeVersionedRegistry(() => "1.0.0"); + const helperBehind = Effect.succeed([ + { ...baseProvider, version: "1.0.0", installedVersion: "9.9.9" }, + ]); + const runner = yield* makeTestRunner({ + ...registry, + getProviders: helperBehind, + refreshInstance: () => helperBehind, + }); + yield* runner.updateProvider({ + provider: CODEX_DRIVER, + targetVersion: "9.9.9", + background: true, + }); + assert.deepStrictEqual(calls, []); + }).pipe( + Effect.provide( + Layer.mergeAll( + layerNonWindowsPlatform, + layerLatestVersionHttpClient("9.9.9"), + layerMockSpawner((_command, args) => { + calls.push(args.join(" ")); + return {}; + }), + ), + ), + ); + }); + + it.effect( + "refuses a broken release, an unpinnable package manager, and a closed window", + () => { + const calls: Array = []; + const manifest: ModelManifest.ModelManifestData = { + version: 1, + currentModels: {}, + compatibility: [ + { + driver: "codex", + t3CodeRange: ">=0.0.42", + ranges: [{ range: ">2.0.0", status: "broken" }], + }, + ], + }; + return Effect.gen(function* () { + const target = { provider: CODEX_DRIVER, background: true } as const; + const make = (capabilities?: ProviderMaintenanceCapabilities) => + makeVersionedRegistry(() => "0.0.0", capabilities).pipe( + Effect.flatMap(({ registry }) => makeTestRunner(registry, manifest)), + ); + yield* (yield* make()).updateProvider({ ...target, targetVersion: "3.0.0" }); + yield* (yield* make(npmCapabilities("unknown-lock-key"))).updateProvider({ + ...target, + targetVersion: "1.0.0", + }); + yield* (yield* make()).updateProvider({ + ...target, + targetVersion: "1.0.0", + beforeInstall: Effect.succeed(false), + }); + assert.deepStrictEqual(calls, []); + }).pipe( + Effect.provide( + Layer.mergeAll( + layerNonWindowsPlatform, + layerLatestVersionHttpClient("3.0.0"), + layerMockSpawner((_command, args) => { + calls.push(args.join(" ")); + return {}; + }), + ), + ), + ); + }, + ); + + it.effect.each([ + ["codex-native", "9.9.9"], + ["codex-native", "10.0.0"], + ["codex-native", null], + ["homebrew", "9.9.9"], + ] as const)( + "vets an unpinned installer actual latest %s under the lock", + ([lockKey, latest]) => { + const calls: Array = []; + let installed = false; + return Effect.gen(function* () { + const { registry } = yield* makeVersionedRegistry( + () => (installed ? "9.9.9" : "0.0.0"), + makeProviderMaintenanceCapabilities({ + provider: CODEX_DRIVER, + packageName: "@openai/codex", + updateExecutable: "codex", + updateArgs: ["update"], + updateLockKey: lockKey, + lookupLatestVersion: Effect.succeed(latest), + }), + ); + const runner = yield* makeTestRunner(registry, { + version: 1, + currentModels: {}, + compatibility: [ + { + driver: "codex", + t3CodeRange: ">=0.0.42", + ranges: [{ range: ">9.9.9", status: "broken" }], + }, + ], + }); + yield* runner.updateProvider({ + provider: CODEX_DRIVER, + targetVersion: "9.9.9", + background: true, + }); + assert.deepStrictEqual(calls, latest === "9.9.9" ? ["update"] : []); + }).pipe( + Effect.provide( + Layer.mergeAll( + layerNonWindowsPlatform, + layerLatestVersionHttpClient("9.9.9"), + layerMockSpawner((_command, args, options) => { + assert.strictEqual( + options.env?.HOMEBREW_NO_AUTO_UPDATE, + lockKey === "homebrew" ? "1" : undefined, + ); + calls.push(args.join(" ")); + installed = true; + return {}; + }), + ), + ), + ); + }, ); + + it.effect("logs an install that failed or left the version unchanged", () => { + logs.length = 0; + let code = 1; + return Effect.gen(function* () { + const { registry, updateStatesRef } = yield* makeVersionedRegistry(() => "0.0.0"); + const runner = yield* makeTestRunner(registry); + const target = { provider: CODEX_DRIVER, targetVersion: "9.9.9", background: true }; + + yield* runner.updateProvider(target); + code = 0; + yield* runner.updateProvider(target); + + assert.deepStrictEqual(yield* Ref.get(updateStatesRef), []); + const text = logs.join("\n"); + assert.include(text, '"status":"failed"'); + assert.include(text, '"status":"unchanged"'); + }).pipe( + Effect.provide( + Layer.mergeAll( + layerNonWindowsPlatform, + layerLatestVersionHttpClient("9.9.9"), + layerCaptureLogs, + layerMockSpawner(() => ({ code })), + ), + ), + ); + }); + + it.effect("tells a manual update another server holds the install lock", () => { + const calls: Array = []; + return Effect.gen(function* () { + const home = yield* FileSystem.FileSystem.use((fileSystem) => + fileSystem.makeTempDirectoryScoped({ prefix: "t3-install-lock-" }), + ).pipe(Effect.provide(NodeTestServices)); + const { registry } = yield* makeVersionedRegistry( + () => "0.0.0", + npmCapabilities("npm-global"), + ); + const otherServer = yield* Effect.service(ProviderInstallLock.ProviderInstallLock).pipe( + Effect.provide(layerLockAt(home)), + ); + const runner = yield* makeRunnerWithLock(registry, defaultManifest, layerLockAt(home)); + + const blocked = yield* otherServer.tryRun("npm-global", () => + runner.updateProvider(CODEX_DRIVER), + ); + + assert.strictEqual( + Option.getOrThrow(blocked).providers[0]?.updateState?.message, + "Another T3 server is already updating this provider.", + ); + assert.deepStrictEqual(calls, []); + }).pipe( + Effect.provide( + Layer.mergeAll( + layerNonWindowsPlatform, + layerLatestVersionHttpClient("0.0.0"), + layerMockSpawner((_command, args) => { + calls.push(args.join(" ")); + return {}; + }), + ), + ), + ); + }); }); }); +describe("ProviderInstallLock", () => { + const withHome = Effect.gen(function* () { + const home = yield* FileSystem.FileSystem.use((fileSystem) => + fileSystem.makeTempDirectoryScoped({ prefix: "t3-install-lock-" }), + ).pipe(Effect.provide(NodeTestServices)); + const instance = () => + Effect.service(ProviderInstallLock.ProviderInstallLock).pipe( + Effect.provide(layerLockAt(home)), + ); + return { + home, + directory: NodePath.join(home, "t3", "provider-install"), + first: yield* instance(), + second: yield* instance(), + }; + }); + const lockFile = (directory: string, key: string) => + NodePath.join(directory, `${encodeURIComponent(key)}.lock`); + + it.effect("excludes every installer key on the host", () => + Effect.gen(function* () { + const { first, second } = yield* withHome; + const installed = yield* second.observe; + const held = yield* first.tryRun( + "npm-global:/prefix", + () => + Effect.gen(function* () { + const sameKey = yield* second.tryRun("npm-global:/prefix", () => Effect.succeed(1)); + const otherKey = yield* second.tryRun("homebrew", () => Effect.succeed(2)); + assert.isTrue(yield* installed("codex")); + assert.isFalse(yield* installed("claudeAgent")); + return { sameKey, otherKey, anyHeld: yield* second.anyHeld }; + }), + "codex", + ); + assert.deepStrictEqual(Option.getOrThrow(held), { + sameKey: Option.none(), + otherKey: Option.none(), + anyHeld: true, + }); + assert.isFalse(yield* second.anyHeld); + assert.isTrue(yield* installed("codex")); + const afterInstall = yield* second.observe; + assert.isFalse(yield* afterInstall("codex")); + yield* first.tryRun("npm-global:/prefix", () => Effect.void, "claudeAgent"); + assert.isTrue(yield* installed("codex")); + assert.isTrue(yield* installed("claudeAgent")); + assert.isFalse(yield* afterInstall("codex")); + assert.isTrue(yield* afterInstall("claudeAgent")); + }).pipe(Effect.provide(NodeTestServices)), + ); + + it.live("holds the lock through verification after the installer exits", () => + Effect.gen(function* () { + const { first, second } = yield* withHome; + const installer = yield* Effect.service(ProviderMaintenanceRunner.ProviderInstaller).pipe( + Effect.provide(ProviderMaintenanceRunner.layerInstaller), + ); + yield* first.tryRun("k", (recordChildPid) => + Effect.gen(function* () { + const result = yield* installer.run({ + command: process.execPath, + args: ["-e", ""], + recordChildPid, + }); + assert.strictEqual(result.exitCode, 0); + assert.isTrue(yield* second.anyHeld); + assert.deepStrictEqual(yield* second.tryRun("k", () => Effect.void), Option.none()); + }), + ); + assert.isFalse(yield* second.anyHeld); + }).pipe(Effect.provide(NodeTestServices)), + ); + + it.effect("expires legacy and numbered records even when their pids are still alive", () => + Effect.gen(function* () { + const { directory, first, second } = yield* withHome; + const fileSystem = yield* FileSystem.FileSystem; + yield* fileSystem.makeDirectory(directory, { recursive: true }); + for (const file of [lockFile(directory, "legacy"), `${lockFile(directory, "k")}.1`]) { + yield* fileSystem.writeFileString(file, String(process.pid)); + yield* fileSystem.utimes(file, 0, 0); + } + yield* TestClock.adjust(Duration.minutes(59)); + assert.isTrue(yield* second.anyHeld); + assert.deepStrictEqual(yield* second.tryRun("k", () => Effect.void), Option.none()); + assert.deepStrictEqual(yield* second.tryRun("legacy", () => Effect.void), Option.none()); + yield* TestClock.adjust(Duration.minutes(1)); + assert.isFalse(yield* second.anyHeld); + yield* second.untilIdle; + assert.deepStrictEqual(yield* first.tryRun("k", () => Effect.succeed(1)), Option.some(1)); + assert.deepStrictEqual( + yield* first.tryRun("legacy", () => Effect.succeed(2)), + Option.some(2), + ); + }).pipe(Effect.provide(NodeTestServices)), + ); + + it.effect.each(["write", "published"] as const)( + "removes a newly created lock after a %s failure", + (stage) => + Effect.gen(function* () { + const fileSystem = yield* FileSystem.FileSystem; + const home = yield* fileSystem.makeTempDirectoryScoped({ prefix: "t3-install-lock-" }); + const failure = PlatformError.systemError({ + _tag: "Unknown", + module: "FileSystem", + method: "writeAll", + description: "No space left on device", + }); + let failRead = false; + const failingFileSystem = FileSystem.FileSystem.of({ + ...fileSystem, + link: (from, to) => + fileSystem.link(from, to).pipe( + Effect.tap(() => + Effect.sync(() => { + failRead = stage === "published"; + }), + ), + ), + readDirectory: (directory) => { + if (!failRead) return fileSystem.readDirectory(directory); + failRead = false; + return Effect.fail(failure); + }, + open: (file, options) => + fileSystem + .open(file, options) + .pipe( + Effect.map((handle) => + stage === "write" ? { ...handle, writeAll: () => Effect.fail(failure) } : handle, + ), + ), + }); + const lock = yield* Effect.service(ProviderInstallLock.ProviderInstallLock).pipe( + Effect.provide(ProviderInstallLock.layer), + Effect.provideService(FileSystem.FileSystem, failingFileSystem), + Effect.provideService(HostProcess.Environment, { XDG_CACHE_HOME: home }), + ); + const result = yield* lock.tryRun("k", () => Effect.die("must not run")).pipe(Effect.exit); + assert.isTrue(Exit.isFailure(result)); + if (Exit.isFailure(result)) { + assert.strictEqual(isProviderInstallLockError(Cause.squash(result.cause)), true); + } + assert.isFalse( + yield* fileSystem.exists( + `${lockFile(NodePath.join(home, "t3", "provider-install"), "k")}.1`, + ), + ); + assert.isFalse(yield* lock.anyHeld); + }).pipe(Effect.provide(NodeTestServices)), + ); + + it.effect.each(["contents", "mtime", "reappeared"] as const)( + "abandons an age takeover after its predecessor changes (%s)", + (change) => + Effect.gen(function* () { + const { home, directory } = yield* withHome; + const fileSystem = yield* FileSystem.FileSystem; + const previous = `${lockFile(directory, "k")}.1`; + const contents = `${process.pid}\npaused-owner`; + yield* fileSystem.makeDirectory(directory, { recursive: true }); + yield* fileSystem.writeFileString(previous, contents); + yield* fileSystem.utimes(previous, 0, 0); + yield* TestClock.adjust(Duration.hours(1)); + let inspected = false; + const lock = yield* Effect.service(ProviderInstallLock.ProviderInstallLock).pipe( + Effect.provide(ProviderInstallLock.layer), + Effect.provideService( + FileSystem.FileSystem, + FileSystem.FileSystem.of({ + ...fileSystem, + stat: (file) => { + if (file !== previous || change !== "reappeared" || inspected) { + return fileSystem.stat(file); + } + inspected = true; + return fileSystem.remove(file).pipe(Effect.andThen(fileSystem.stat(file))); + }, + link: (from, to) => + fileSystem.link(from, to).pipe( + Effect.andThen( + change === "mtime" + ? fileSystem.utimes(previous, 3600, 3600) + : fileSystem.writeFileString(previous, `${process.pid}\nresumed-owner`).pipe( + // A contents change must fence even if timestamp precision hides it. + Effect.andThen(fileSystem.utimes(previous, 0, 0)), + ), + ), + ), + }), + ), + Effect.provideService(HostProcess.Environment, { XDG_CACHE_HOME: home }), + ); + assert.deepStrictEqual( + yield* lock.tryRun("k", () => Effect.die("must not install")), + Option.none(), + ); + assert.deepStrictEqual(yield* fileSystem.readDirectory(directory), ["k.lock.1"]); + assert.strictEqual( + yield* fileSystem.readFileString(previous), + change === "mtime" ? contents : `${process.pid}\nresumed-owner`, + ); + if (change === "mtime") { + assert.isTrue(yield* lock.anyHeld); + assert.deepStrictEqual(yield* lock.tryRun("k", () => Effect.void), Option.none()); + } + }).pipe(Effect.provide(NodeTestServices)), + ); + + it.effect("does not skip a resumed owner after another successor abandons its generation", () => + Effect.gen(function* () { + const { home, directory, first } = yield* withHome; + const fileSystem = yield* FileSystem.FileSystem; + const resumed = `${lockFile(directory, "k")}.1`; + const abandoned = `${lockFile(directory, "k")}.2`; + yield* fileSystem.makeDirectory(directory, { recursive: true }); + yield* fileSystem.writeFileString(resumed, `${process.pid}\nresumed-owner`); + yield* fileSystem.writeFileString(abandoned, "released\nabandoning-successor"); + assert.isTrue(yield* first.anyHeld); + yield* fileSystem.writeFileString(abandoned, `${process.pid}\nabandoning-successor`); + let snapshotted = false; + const lock = yield* Effect.service(ProviderInstallLock.ProviderInstallLock).pipe( + Effect.provide(ProviderInstallLock.layer), + Effect.provideService( + FileSystem.FileSystem, + FileSystem.FileSystem.of({ + ...fileSystem, + readDirectory: (file) => + fileSystem.readDirectory(file).pipe( + Effect.tap(() => { + if (snapshotted) return Effect.void; + snapshotted = true; + return fileSystem.remove(abandoned); + }), + ), + }), + ), + Effect.provideService(HostProcess.Environment, { XDG_CACHE_HOME: home }), + ); + assert.deepStrictEqual( + yield* lock.tryRun("k", () => Effect.die("must not install")), + Option.none(), + ); + assert.deepStrictEqual(yield* fileSystem.readDirectory(directory), ["k.lock.1"]); + assert.isTrue(yield* lock.anyHeld); + }).pipe(Effect.provide(NodeTestServices)), + ); + + it.effect("reclaims a lock left by a process that is gone", () => + Effect.gen(function* () { + const { directory, first, second } = yield* withHome; + const fileSystem = yield* FileSystem.FileSystem; + const deadPid = NodeChildProcess.spawnSync(process.execPath, ["-e", ""]).pid; + yield* fileSystem.makeDirectory(directory, { recursive: true }); + yield* fileSystem.writeFileString(lockFile(directory, "k"), String(deadPid)); + + assert.isFalse(yield* second.anyHeld); + assert.deepStrictEqual( + yield* first.tryRun("k", () => Effect.succeed("ran")), + Option.some("ran"), + ); + }).pipe(Effect.provide(NodeTestServices)), + ); + + it.effect("honors a live legacy lock and retains only the highest released generation", () => + Effect.gen(function* () { + const { directory, first, second } = yield* withHome; + const fileSystem = yield* FileSystem.FileSystem; + const legacy = lockFile(directory, "k"); + yield* fileSystem.makeDirectory(directory, { recursive: true }); + yield* fileSystem.writeFileString(legacy, String(process.pid)); + assert.isTrue(yield* first.anyHeld); + assert.deepStrictEqual(yield* second.tryRun("k", () => Effect.void), Option.none()); + yield* fileSystem.remove(legacy); + for (let generation = 1; generation <= 3; generation += 1) { + assert.isTrue(Option.isSome(yield* first.tryRun("k", () => Effect.void))); + assert.deepStrictEqual(yield* fileSystem.readDirectory(directory), [ + `k.lock.${generation}`, + ]); + assert.strictEqual( + (yield* fileSystem.readFileString(`${legacy}.${generation}`)).split("\n")[0], + "released", + ); + assert.isFalse(yield* second.anyHeld); + } + }).pipe(Effect.provide(NodeTestServices)), + ); + + it.live("does not start an installer when recording its pid fails", () => + Effect.gen(function* () { + const { home, directory, second } = yield* withHome; + const fileSystem = yield* FileSystem.FileSystem; + const marker = NodePath.join(home, "must-not-install"); + let failRename = true; + const lock = yield* Effect.service(ProviderInstallLock.ProviderInstallLock).pipe( + Effect.provide(ProviderInstallLock.layer), + Effect.provideService( + FileSystem.FileSystem, + FileSystem.FileSystem.of({ + ...fileSystem, + rename: (from, to) => { + if (!failRename) return fileSystem.rename(from, to); + failRename = false; + return Effect.fail( + PlatformError.systemError({ + _tag: "Unknown", + module: "FileSystem", + method: "rename", + description: "No space left on device", + }), + ); + }, + }), + ), + Effect.provideService(HostProcess.Environment, { XDG_CACHE_HOME: home }), + ); + const wrapperExited = yield* Deferred.make(); + const result = yield* lock + .tryRun("k", (recordChildPid) => + ProviderMaintenanceRunner.runDetachedInstall( + { + command: process.execPath, + args: ["-e", "require('node:fs').writeFileSync(process.argv[1],'bad')", marker], + recordChildPid, + }, + (( + command: string, + args: ReadonlyArray, + options: NodeChildProcess.SpawnOptions, + ) => { + const child = NodeChildProcess.spawn(command, args, options); + child.once("exit", () => Deferred.doneUnsafe(wrapperExited, Effect.void)); + return child; + }) as typeof NodeChildProcess.spawn, + ), + ) + .pipe(Effect.exit); + assert.isTrue(Exit.isFailure(result)); + yield* Deferred.await(wrapperExited); + assert.isFalse(yield* fileSystem.exists(marker)); + assert.isFalse(yield* second.anyHeld); + assert.deepStrictEqual(yield* fileSystem.readDirectory(directory), ["k.lock.1"]); + assert.isTrue(Option.isSome(yield* second.tryRun("k", () => Effect.void))); + }).pipe(Effect.provide(NodeTestServices)), + ); + + it.effect("two servers reclaiming the same stale lock leave exactly one owner", () => + Effect.gen(function* () { + const { directory, first, second } = yield* withHome; + const fileSystem = yield* FileSystem.FileSystem; + const deadPid = NodeChildProcess.spawnSync(process.execPath, ["-e", ""]).pid; + yield* fileSystem.makeDirectory(directory, { recursive: true }); + yield* fileSystem.writeFileString(lockFile(directory, "k"), String(deadPid)); + const runs = yield* Ref.make(0); + yield* fileSystem.writeFileString(`${lockFile(directory, "k")}.reclaim`, String(deadPid)); + yield* fileSystem.utimes(`${lockFile(directory, "k")}.reclaim`, 0, 0); + // The winner holds the lock until the other reclaimer has finished its attempt. + const loserDone = yield* Deferred.make(); + const attempt = (lock: typeof first) => + lock + .tryRun("k", () => + Ref.update(runs, (count) => count + 1).pipe(Effect.andThen(Deferred.await(loserDone))), + ) + .pipe( + Effect.tap((result) => + Option.isNone(result) ? Deferred.succeed(loserDone, undefined) : Effect.void, + ), + ); + + const results = yield* Effect.all([attempt(first), attempt(second)], { + concurrency: "unbounded", + }); + + assert.strictEqual(results.filter(Option.isSome).length, 1); + assert.strictEqual(yield* Ref.get(runs), 1); + }).pipe(Effect.provide(NodeTestServices)), + ); + + it.live("a detached installer outlives the scope that ran it and keeps its lock", () => + Effect.gen(function* () { + const fileSystem = yield* FileSystem.FileSystem; + const { directory, first, second } = yield* withHome; + const marker = NodePath.join(NodePath.dirname(directory), "done"); + const installer = yield* Effect.service(ProviderMaintenanceRunner.ProviderInstaller).pipe( + Effect.provide(ProviderMaintenanceRunner.layerInstaller), + ); + const started = yield* Deferred.make(); + const scope = yield* Scope.make(); + yield* first + .tryRun("k", (recordChildPid) => + installer.run({ + command: process.execPath, + args: [ + "-e", + "console.log('hello');setTimeout(()=>require('node:fs').writeFileSync(process.argv[1],'finished'),1500)", + marker, + ], + recordChildPid: (pid) => + recordChildPid(pid).pipe(Effect.andThen(Deferred.succeed(started, undefined))), + }), + ) + .pipe(Effect.forkIn(scope)); + yield* Deferred.await(started); + + // Closing the scope interrupts the runner; the installer keeps going and keeps the lock. + yield* Scope.close(scope, Exit.void); + assert.isTrue(yield* second.anyHeld); + assert.deepStrictEqual(yield* second.tryRun("k", () => Effect.void), Option.none()); + + // The wrapper releases the lock only after the installer exits. + yield* second.untilIdle; + assert.strictEqual(yield* fileSystem.readFileString(marker), "finished"); + assert.isFalse(yield* second.anyHeld); + }).pipe(Effect.provide(NodeTestServices)), + ); + + it.live("captures installer output and reports its latest line", () => + Effect.gen(function* () { + const fileSystem = yield* FileSystem.FileSystem; + let maxLogSize = 0; + const installer = yield* Effect.service(ProviderMaintenanceRunner.ProviderInstaller).pipe( + Effect.provide(ProviderMaintenanceRunner.layerInstaller), + Effect.provideService( + FileSystem.FileSystem, + FileSystem.FileSystem.of({ + ...fileSystem, + readFile: (file) => + file.includes("t3-provider-install-") + ? Effect.die("installer logs require bounded reads") + : fileSystem.readFile(file), + open: (file, options) => + fileSystem.open(file, options).pipe( + Effect.map((handle) => + file.includes("t3-provider-install-") && options?.flag === "r" + ? { + ...handle, + seek: (offset, from) => handle.seek(offset, from), + stat: handle.stat.pipe( + Effect.tap((info) => + Effect.sync(() => { + maxLogSize = Math.max(maxLogSize, Number(info.size)); + }), + ), + ), + read: (bytes) => { + assert.isAtMost(bytes.length, 10_000); + return handle.read(bytes); + }, + } + : handle, + ), + ), + }), + ), + ); + const lines: Array = []; + const result = yield* installer.run({ + command: process.execPath, + args: [ + "-e", + "process.stdout.write('x'.repeat(2*1024*1024));console.log('\\nhello');setTimeout(()=>{},1300)", + ], + onProgress: (line) => Effect.sync(() => void lines.push(line)), + recordChildPid: () => Effect.void, + }); + assert.strictEqual(result.exitCode, 0); + assert.isTrue(result.stdout.endsWith("\nhello\n")); + assert.isTrue(result.stdoutTruncated); + assert.isAbove(maxLogSize, 10_000); + assert.isAtMost(maxLogSize, 1_048_576); + assert.include(lines, "hello"); + }).pipe(Effect.provide(NodeTestServices)), + ); +}); + +describe("ProviderAutoUpdater", () => { + const harness = (input: { + readonly automaticUpdates?: boolean; + readonly window: { open: boolean }; + readonly latest: { version: string }; + readonly twoInstances?: boolean; + }) => + Effect.gen(function* () { + const calls: Array<{ + readonly targetVersion?: string | undefined; + readonly beforeInstall?: Effect.Effect | undefined; + }> = []; + const { registry } = yield* makeRegistry({ ...baseProvider, version: "1.0.0" }); + yield* Layer.build( + ProviderAutoUpdater.layer.pipe( + Layer.provide( + Layer.mergeAll( + Layer.succeed(ServerConfig.ServerConfig, { devUrl: undefined } as never), + Layer.succeed(ProviderRegistry.ProviderRegistry, { + ...registry, + getProviders: input.twoInstances + ? Effect.map(registry.getProviders, (providers) => [ + ...providers, + { ...providers[0]!, instanceId: ProviderInstanceId.make("codex-two") }, + ]) + : registry.getProviders, + }), + Layer.succeed(ProviderMaintenanceRunner.ProviderMaintenanceRunner, { + updateProvider: (target) => + Effect.sync(() => { + if (typeof target !== "string") calls.push(target); + return { providers: [] }; + }), + }), + Layer.succeed(UpdateWindow.UpdateWindow, { + isOpen: () => Effect.sync(() => input.window.open), + runIfOpen: (effect) => + input.window.open ? Effect.asSome(effect) : Effect.succeedNone, + }), + ServerSettings.layerTest({ automaticUpdates: input.automaticUpdates ?? true }), + ProviderLatestVersions.layer, + Layer.succeed( + HttpClient.HttpClient, + HttpClient.make((request) => + Effect.succeed( + HttpClientResponse.fromWeb( + request, + Response.json({ version: input.latest.version }), + ), + ), + ), + ), + ), + ), + ), + ); + const advance = (duration: Duration.Duration) => TestClock.adjust(duration); + return { calls, advance }; + }); + + it.effect("attempts one outdated instance per pass", () => + Effect.gen(function* () { + const { calls, advance } = yield* harness({ + window: { open: true }, + latest: { version: "2.0.0" }, + twoInstances: true, + }); + yield* advance(Duration.minutes(1)); + assert.strictEqual(calls.length, 0); + yield* advance(Duration.minutes(1)); + assert.deepStrictEqual( + calls.map((call) => call.targetVersion), + ["2.0.0"], + ); + yield* advance(Duration.minutes(10)); + assert.strictEqual(calls.length, 2); + }), + ); + + it.effect("does nothing when automatic updates are off", () => + Effect.gen(function* () { + const { calls, advance } = yield* harness({ + automaticUpdates: false, + window: { open: true }, + latest: { version: "2.0.0" }, + }); + yield* advance(Duration.minutes(30)); + assert.strictEqual(calls.length, 0); + }), + ); + + it.effect("waits for the window, and tries again once a postponed install's window reopens", () => + Effect.gen(function* () { + const window = { open: false }; + const { calls, advance } = yield* harness({ window, latest: { version: "2.0.0" } }); + yield* advance(Duration.minutes(12)); + assert.strictEqual(calls.length, 0); + + window.open = true; + yield* advance(Duration.minutes(10)); + assert.strictEqual(calls.length, 1); + + // The window closed again before the installer started: that is not a failed try. + window.open = false; + assert.isFalse(yield* calls[0]!.beforeInstall!); + window.open = true; + yield* advance(Duration.minutes(10)); + assert.strictEqual(calls.length, 2); + }), + ); + + it.effect("tries a release once per six hours but a newer release at once", () => + Effect.gen(function* () { + const latest = { version: "2.0.0" }; + const { calls, advance } = yield* harness({ window: { open: true }, latest }); + yield* advance(Duration.minutes(2)); + yield* advance(Duration.minutes(10)); + assert.strictEqual(calls.length, 1); + + latest.version = "2.1.0"; + yield* advance(Duration.minutes(70)); + assert.deepStrictEqual( + calls.map((call) => call.targetVersion), + ["2.0.0", "2.1.0"], + ); + + yield* advance(Duration.hours(7)); + assert.deepStrictEqual( + calls.map((call) => call.targetVersion), + ["2.0.0", "2.1.0", "2.1.0"], + ); + }), + ); +}); + +describe("UpdateWindow", () => { + const makeWindow = (state: { + quietFor: Duration.Duration; + active: ReadonlyArray | "unreadable"; + installing: boolean; + }) => + Effect.service(UpdateWindow.UpdateWindow).pipe( + Effect.provide( + UpdateWindow.layer.pipe( + Layer.provide( + Layer.mergeAll( + Layer.mock(BackgroundPolicy.BackgroundPolicy)({ + sinceLastClientInteraction: Effect.sync(() => state.quietFor), + }), + Layer.mock(ProjectionStore.ProjectionStoreV2)({ + getRecoveryThreadIds: () => + state.active === "unreadable" + ? Effect.fail(new ProjectionStore.ProjectionStoreSetupError({ cause: "x" })) + : Effect.succeed(state.active), + }), + Layer.mock(ProviderInstallLock.ProviderInstallLock)({ + anyHeld: Effect.sync(() => state.installing), + }), + ), + ), + ), + ), + ); + + it.effect( + "is open only when clients are quiet, no run is active, and no install is running", + () => + Effect.gen(function* () { + const state = { + quietFor: Duration.minutes(2), + active: [] as ReadonlyArray | "unreadable", + installing: false, + }; + const window = yield* makeWindow(state); + assert.isTrue(yield* window.isOpen()); + + state.quietFor = Duration.seconds(119); + assert.isFalse(yield* window.isOpen()); + state.quietFor = Duration.minutes(2); + + state.active = [ThreadId.make("thread:busy")]; + assert.isFalse(yield* window.isOpen()); + state.active = "unreadable"; + assert.isFalse(yield* window.isOpen()); + state.active = []; + + state.installing = true; + assert.isFalse(yield* window.isOpen()); + assert.isTrue(yield* window.isOpen({ ignoreInstallLocks: true })); + assert.deepStrictEqual(yield* window.runIfOpen(Effect.succeed("ran")), Option.none()); + state.installing = false; + assert.deepStrictEqual(yield* window.runIfOpen(Effect.succeed("ran")), Option.some("ran")); + }), + ); +}); + it.effect.each([ { mode: "default", diff --git a/apps/server/src/provider/providerMaintenanceRunner.ts b/apps/server/src/provider/providerMaintenanceRunner.ts index d3fe7fdf86739..7b57d057b0902 100644 --- a/apps/server/src/provider/providerMaintenanceRunner.ts +++ b/apps/server/src/provider/providerMaintenanceRunner.ts @@ -1,3 +1,4 @@ +// @effect-diagnostics nodeBuiltinImport:off - the detached installer needs a raw spawn and a log fd import { defaultInstanceIdForDriver, ProviderDriverKind, @@ -7,24 +8,34 @@ import { type ServerProviderUpdatedPayload, type ServerProviderUpdateState, } from "@t3tools/contracts"; +import * as HostProcess from "@t3tools/shared/HostProcess"; +import { resolveNodeExecutable } from "@t3tools/shared/nodeRuntime"; +import { compareSemverVersions } from "@t3tools/shared/semver"; import { resolveSpawnCommand } from "@t3tools/shared/shell"; import * as Cause from "effect/Cause"; import * as Context from "effect/Context"; import * as DateTime from "effect/DateTime"; +import * as Deferred from "effect/Deferred"; import * as Duration from "effect/Duration"; import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; +import * as Path from "effect/Path"; import * as Ref from "effect/Ref"; import * as Schedule from "effect/Schedule"; import * as Schema from "effect/Schema"; import * as Stream from "effect/Stream"; -import { HttpClient } from "effect/http"; import { ChildProcess, ChildProcessSpawner } from "effect/process"; +import { collectUint8StreamText } from "@t3tools/provider-core/server/collectStreamText"; +import { HttpClient } from "effect/http"; +import * as NodeChildProcess from "node:child_process"; +import * as NodeFS from "node:fs"; import * as NodeUtil from "node:util"; import * as ModelManifest from "./ModelManifest.ts"; import { resolveProviderCompatibility } from "./providerCompatibility.ts"; +import * as ProviderInstallLock from "./ProviderInstallLock.ts"; import * as ProviderRegistry from "./ProviderRegistry.ts"; import { makeProviderMaintenanceCommandCoordinator } from "./providerMaintenanceCommandCoordinator.ts"; import { @@ -37,15 +48,17 @@ import { } from "@t3tools/provider-core/server/maintenanceResolver"; import * as ProviderLatestVersions from "@t3tools/provider-core/server/ProviderLatestVersions"; import type { ProviderMaintenanceCapabilities } from "@t3tools/provider-core/server/maintenanceResolver"; -import { collectUint8StreamText } from "@t3tools/provider-core/server/collectStreamText"; +import { signalProcessGroup } from "@t3tools/provider-core/server/processGroup"; const isServerProviderUpdateError = Schema.is(ServerProviderUpdateError); -// Native updaters (`claude update`) download a ~300 MB binary without printing -// progress, and killing one mid-download leaves a truncated binary that the -// next run treats as installed. Allow a slow connection to finish. -const UPDATE_TIMEOUT_MINUTES = 30; -const UPDATE_TIMEOUT_MS = UPDATE_TIMEOUT_MINUTES * 60_000; +// An installer is never cut off mid-write for taking long; this only frees the +// install lock and held turn starts from one that hangs. +const INSTALL_BACKSTOP = Duration.minutes(30); +const INSTALL_GO_TIMEOUT = Duration.seconds(30); +// Only these may run their own updater without a pinned `package@version`. +const SELF_UPDATING_LOCK_KEY = /^(?:homebrew|grok|muse:.*|.+-native)$/; const UPDATE_OUTPUT_MAX_BYTES = 10_000; +const INSTALL_LOG_MAX_BYTES = 1_048_576; // Sample installer output because every progress update resends the provider list to all clients. const UPDATE_PROGRESS_INTERVAL = Duration.seconds(1); const UPDATE_PROGRESS_MAX_LENGTH = 200; @@ -69,6 +82,14 @@ export interface ProviderMaintenanceRunnerShape { readonly provider: ProviderDriverKind; readonly instanceId?: ProviderInstanceId | undefined; readonly targetVersion?: string | undefined; + /** + * Install `targetVersion` unattended: publishes no update state, refuses + * anything but a pinned (or self-updating) install, and skips when the + * installed version already reaches the target. + */ + readonly background?: boolean | undefined; + /** Last check before a background install starts, run under the install lock. */ + readonly beforeInstall?: Effect.Effect | undefined; }, ) => Effect.Effect; } @@ -94,117 +115,377 @@ interface VerifiedProviderRefresh { const nowIso = Effect.map(DateTime.now, DateTime.formatIso); -const runProviderMaintenanceCommandWithSpawner = Effect.fn("ProviderMaintenanceRunner.runCommand")( - function* (input: { - readonly spawner: ChildProcessSpawner.ChildProcessSpawner["Service"]; - readonly command: string; - readonly args: ReadonlyArray; - readonly env?: NodeJS.ProcessEnv; - readonly onProgress?: (line: string) => Effect.Effect; - }) { - const collectCommandResult = Effect.fn("ProviderMaintenanceRunner.collectCommandResult")( - function* () { - const env = input.env ? { env: input.env, extendEnv: true } : {}; - // Windows batch shims require shell dispatch through resolveSpawnCommand. - const resolved = yield* resolveSpawnCommand(input.command, input.args, env); - const child = yield* input.spawner - .spawn( - ChildProcess.make(resolved.command, resolved.args, { - shell: resolved.shell, - ...env, - }), - ) - .pipe( - Effect.mapError( - (cause) => - new ProviderMaintenanceCommandError({ - message: `Failed to run update command ${input.command}: ${cause.message}`, - cause, - }), - ), +export interface ProviderInstallInput { + readonly command: string; + readonly args: ReadonlyArray; + readonly env?: NodeJS.ProcessEnv | undefined; + readonly onProgress?: ((line: string) => Effect.Effect) | undefined; + /** Must record the gated wrapper's pid before allowing it to start the installer. */ + readonly recordChildPid: (pid: number) => Effect.Effect; +} + +/** Runs one provider installer to completion. */ +export class ProviderInstaller extends Context.Service< + ProviderInstaller, + { + readonly run: ( + input: ProviderInstallInput, + ) => Effect.Effect; + } +>()("t3/provider/providerMaintenanceRunner/ProviderInstaller") {} + +const noop = () => {}; + +const encodeInstallWrapperInput = Schema.encodeSync( + Schema.fromJsonString( + Schema.Struct({ + command: Schema.String, + args: Schema.Array(Schema.String), + shell: Schema.Boolean, + runAsNode: Schema.optional(Schema.String), + }), + ), +); + +// EOF or the go deadline aborts without touching the install target. The wrapper +// owns the installer pipes, so a server exit cannot break their reader. +const INSTALL_WRAPPER = ` +const goDeadline = performance.now() + ${Duration.toMillis(INSTALL_GO_TIMEOUT)}; +const { spawn } = require("node:child_process"); +const { command, args, shell, runAsNode } = JSON.parse(process.argv[1]); +let started = false; +const goTimeout = setTimeout(() => process.exit(1), ${Duration.toMillis(INSTALL_GO_TIMEOUT)}); +process.stdin.once("end", () => { if (!started) process.exit(1); }); +process.stdin.once("data", (data) => { + if (data[0] !== 103 || performance.now() >= goDeadline) process.exit(1); + started = true; + clearTimeout(goTimeout); + process.stdin.destroy(); + const env = { ...process.env }; + if (runAsNode === undefined) delete env.ELECTRON_RUN_AS_NODE; + else env.ELECTRON_RUN_AS_NODE = runAsNode; + const fs = require("node:fs"); + let logBytes = 0; + const appendLog = (chunk) => { + try { + if (logBytes + chunk.length > ${INSTALL_LOG_MAX_BYTES}) { + const tail = Buffer.alloc(Math.min(logBytes, ${UPDATE_OUTPUT_MAX_BYTES + 1})); + fs.readSync(1, tail, 0, tail.length, logBytes - tail.length); + fs.ftruncateSync(1, 0); + logBytes = fs.writeSync(1, tail); + } + logBytes += fs.writeSync(1, chunk); + } catch {} + }; + const installer = spawn(command, args, { + shell, windowsHide: true, stdio: ["ignore", "pipe", "pipe"], env, + }); + installer.stdout.on("data", appendLog); + installer.stderr.on("data", appendLog); + // This backstop survives the server that normally enforces the same limit. + let timedOut = false; + const backstop = setTimeout(() => { + timedOut = true; + if (process.platform === "win32") { + const killer = spawn("taskkill", ["/PID", String(installer.pid), "/T", "/F"], { + windowsHide: true, stdio: "ignore", + }); + killer.once("error", () => process.exit(1)); + killer.once("exit", () => process.exit(1)); + } else { + try { process.kill(-process.pid, "SIGKILL"); } + catch (error) { console.error(error); process.exit(1); } + } + }, ${Duration.toMillis(INSTALL_BACKSTOP)}); + installer.once("error", (error) => { + console.error(error); + process.exitCode = 1; + }); + installer.once("exit", (code) => { + process.exitCode = timedOut ? 1 : code ?? 1; + }); + installer.once("close", () => clearTimeout(backstop)); +}); +`; + +/** + * Runs an installer as a detached process whose output goes to a log file. It + * is not tied to this server: quitting the server neither kills the installer + * nor breaks its stdio (no EPIPE), so an npm extract is never cut in half. Only + * the backstop ends it, and a result is returned once it exits. + */ +export const runDetachedInstall = Effect.fn("ProviderMaintenanceRunner.runDetachedInstall")( + function* ( + input: ProviderInstallInput, + spawnProcess: typeof NodeChildProcess.spawn = NodeChildProcess.spawn, + ) { + const fileSystem = yield* FileSystem.FileSystem; + const platform = yield* HostProcess.Platform; + const environment = yield* HostProcess.Environment; + const childEnvironment = { ...environment, ...input.env }; + const node = yield* resolveNodeExecutable("Provider CLI updates", childEnvironment); + // Windows batch shims need shell dispatch through resolveSpawnCommand. + const resolved = yield* resolveSpawnCommand(input.command, input.args, { + ...(input.env ? { env: input.env } : {}), + extendEnv: true, + }); + const toError = (cause: unknown) => + new ProviderMaintenanceCommandError({ + message: `Failed to run update command ${input.command}: ${cause instanceof Error ? cause.message : "unknown error"}`, + cause, + }); + const logPath = yield* fileSystem.makeTempFileScoped({ prefix: "t3-provider-install-" }); + // The child gets its own copy of this fd, so it is closed here once the child has spawned. + const logFd = yield* Effect.acquireRelease( + Effect.sync(() => NodeFS.openSync(logPath, "a+")), + (fd) => Effect.sync(() => NodeFS.closeSync(fd)), + ); + + // Listeners are registered in the spawn tick: Node emits `spawn` and `exit` + // asynchronously, and a late listener would wait forever. + const { child, spawned, exited } = yield* Effect.acquireRelease( + Effect.try({ + try: () => { + const child = spawnProcess( + node, + [ + "-e", + INSTALL_WRAPPER, + encodeInstallWrapperInput({ + ...resolved, + runAsNode: childEnvironment.ELECTRON_RUN_AS_NODE, + }), + ], + { + detached: true, + windowsHide: true, + stdio: ["pipe", logFd, logFd], + env: { + ...childEnvironment, + ...(process.versions.electron ? { ELECTRON_RUN_AS_NODE: "1" } : {}), + }, + }, ); - yield* Effect.addFinalizer(() => child.kill().pipe(Effect.ignore)); - - const pendingProgress = yield* Ref.make(null); - const onProgress = input.onProgress; - if (onProgress) { - yield* Ref.getAndSet(pendingProgress, null).pipe( - Effect.flatMap((line) => (line === null ? Effect.void : onProgress(line))), - Effect.repeat(Schedule.spaced(UPDATE_PROGRESS_INTERVAL)), - Effect.forkScoped, + const spawned = Deferred.makeUnsafe(); + const exited = Deferred.makeUnsafe(); + child.on("error", (cause) => Deferred.doneUnsafe(spawned, Effect.fail(toError(cause)))); + child.once("spawn", () => Deferred.doneUnsafe(spawned, Effect.succeed(child.pid ?? 0))); + child.once("exit", (code) => Deferred.doneUnsafe(exited, Effect.succeed(code))); + child.stdin!.on("error", noop); + child.unref(); + return { child, spawned, exited }; + }, + catch: toError, + }), + ({ child }) => Effect.sync(() => child.stdin!.destroy()), + ); + const pid = yield* Deferred.await(spawned); + yield* input.recordChildPid(pid).pipe( + Effect.andThen( + Effect.callback((resume) => { + child.stdin!.end("g", (error?: Error | null) => + resume(error ? Effect.fail(toError(error)) : Effect.void), ); - } - const trackProgress = (stream: Stream.Stream) => { - if (!onProgress) { - return stream; - } - const decoder = new TextDecoder(); - let partialLine = ""; - return stream.pipe( - Stream.tap((chunk) => { - const split = splitOutputLines(partialLine, decoder.decode(chunk, { stream: true })); - partialLine = split.partialLine; - const line = - toProgressLine(split.partialLine) ?? - split.lines.map(toProgressLine).findLast((value) => value !== null); - return line ? Ref.set(pendingProgress, line) : Effect.void; - }), + }), + ), + Effect.uninterruptible, + ); + + const readLog = Effect.gen(function* () { + const file = yield* fileSystem.open(logPath, { flag: "r" }); + const info = yield* file.stat; + const limit = BigInt(UPDATE_OUTPUT_MAX_BYTES); + const offset = info.size > limit ? info.size - limit : 0n; + yield* file.seek(offset, "start"); + const bytes = new Uint8Array(Number(info.size - offset)); + let length = 0; + while (length < bytes.length) { + const read = Number(yield* file.read(bytes.subarray(length))); + if (read === 0) break; + length += read; + } + return { bytes: bytes.subarray(0, length), truncated: offset > 0n }; + }).pipe( + Effect.scoped, + Effect.orElseSucceed(() => ({ bytes: new Uint8Array(), truncated: false })), + ); + const onProgress = input.onProgress; + if (onProgress) { + yield* readLog.pipe( + Effect.map(({ bytes }) => { + const split = splitOutputLines("", new TextDecoder().decode(bytes)); + // A progress bar that redraws with a leading `\r` keeps its newest + // frame unterminated, so that frame is the latest status. + return ( + toProgressLine(split.partialLine) ?? + split.lines.map(toProgressLine).findLast((value) => value !== null) ?? + null ); - }; + }), + Effect.flatMap((line) => (line === null ? Effect.void : onProgress(line))), + Effect.repeat(Schedule.spaced(UPDATE_PROGRESS_INTERVAL)), + Effect.forkScoped, + ); + } - const [stdout, stderr, exitCode] = yield* Effect.all( - [ - collectUint8StreamText({ - stream: trackProgress(child.stdout), - maxBytes: UPDATE_OUTPUT_MAX_BYTES, - }), - collectUint8StreamText({ - stream: trackProgress(child.stderr), - maxBytes: UPDATE_OUTPUT_MAX_BYTES, - }), - child.exitCode, - ], - { concurrency: "unbounded" }, - ).pipe( + const code = yield* Deferred.await(exited).pipe(Effect.timeoutOption(INSTALL_BACKSTOP)); + if (Option.isNone(code)) { + yield* Effect.sync(() => { + if (platform === "win32") { + spawnProcess("taskkill", ["/PID", String(pid), "/T", "/F"], { + windowsHide: true, + stdio: "ignore", + }).on("error", noop); + } else { + signalProcessGroup(pid, "SIGKILL"); + } + }).pipe(Effect.ignore); + } + const output = yield* readLog; + return { + stdout: new TextDecoder().decode(output.bytes), + stderr: "", + exitCode: Option.isSome(code) ? code.value : null, + timedOut: Option.isNone(code), + stdoutTruncated: output.truncated, + stderrTruncated: false, + } satisfies ProviderMaintenanceCommandResult; + }, + Effect.scoped, +); + +const runForegroundInstall = Effect.fn("ProviderMaintenanceRunner.runCommand")(function* (input: { + readonly spawner: ChildProcessSpawner.ChildProcessSpawner["Service"]; + readonly command: string; + readonly args: ReadonlyArray; + readonly env?: NodeJS.ProcessEnv | undefined; + readonly onProgress?: ((line: string) => Effect.Effect) | undefined; + readonly recordChildPid: ProviderInstallInput["recordChildPid"]; +}) { + const collectCommandResult = Effect.fn("ProviderMaintenanceRunner.collectCommandResult")( + function* () { + const env = input.env ? { env: input.env, extendEnv: true } : {}; + // Windows batch shims require shell dispatch through resolveSpawnCommand. + const resolved = yield* resolveSpawnCommand(input.command, input.args, env); + const child = yield* input.spawner + .spawn( + ChildProcess.make(resolved.command, resolved.args, { + shell: resolved.shell, + ...env, + }), + ) + .pipe( Effect.mapError( (cause) => new ProviderMaintenanceCommandError({ - message: cause instanceof Error ? cause.message : "Update command failed to run.", + message: `Failed to run update command ${input.command}: ${cause.message}`, cause, }), ), ); + yield* Effect.addFinalizer(() => child.kill().pipe(Effect.ignore)); - return { - stdout: stdout.text, - stderr: stderr.text, - exitCode: Number(exitCode), - timedOut: false, - stdoutTruncated: stdout.truncated, - stderrTruncated: stderr.truncated, - } satisfies ProviderMaintenanceCommandResult; - }, - ); + yield* input.recordChildPid(Number(child.pid)); - return yield* collectCommandResult().pipe( - Effect.scoped, - Effect.timeoutOption(Duration.millis(UPDATE_TIMEOUT_MS)), - Effect.map((result) => - Option.match(result, { - onSome: (value) => value, - onNone: () => - ({ - stdout: "", - stderr: "", - exitCode: null, - timedOut: true, - stdoutTruncated: false, - stderrTruncated: false, - }) satisfies ProviderMaintenanceCommandResult, - }), + const pendingProgress = yield* Ref.make(null); + const onProgress = input.onProgress; + if (onProgress) { + yield* Ref.getAndSet(pendingProgress, null).pipe( + Effect.flatMap((line) => (line === null ? Effect.void : onProgress(line))), + Effect.repeat(Schedule.spaced(UPDATE_PROGRESS_INTERVAL)), + Effect.forkScoped, + ); + } + const trackProgress = (stream: Stream.Stream) => { + if (!onProgress) { + return stream; + } + const decoder = new TextDecoder(); + let partialLine = ""; + return stream.pipe( + Stream.tap((chunk) => { + const split = splitOutputLines(partialLine, decoder.decode(chunk, { stream: true })); + partialLine = split.partialLine; + const line = + toProgressLine(split.partialLine) ?? + split.lines.map(toProgressLine).findLast((value) => value !== null); + return line ? Ref.set(pendingProgress, line) : Effect.void; + }), + ); + }; + + const [stdout, stderr, exitCode] = yield* Effect.all( + [ + collectUint8StreamText({ + stream: trackProgress(child.stdout), + maxBytes: UPDATE_OUTPUT_MAX_BYTES, + }), + collectUint8StreamText({ + stream: trackProgress(child.stderr), + maxBytes: UPDATE_OUTPUT_MAX_BYTES, + }), + child.exitCode, + ], + { concurrency: "unbounded" }, + ).pipe( + Effect.mapError( + (cause) => + new ProviderMaintenanceCommandError({ + message: cause instanceof Error ? cause.message : "Update command failed to run.", + cause, + }), + ), + ); + + return { + stdout: stdout.text, + stderr: stderr.text, + exitCode: Number(exitCode), + timedOut: false, + stdoutTruncated: stdout.truncated, + stderrTruncated: stderr.truncated, + } satisfies ProviderMaintenanceCommandResult; + }, + ); + + return yield* collectCommandResult().pipe( + Effect.scoped, + Effect.timeoutOption(INSTALL_BACKSTOP), + Effect.map((result) => + Option.match(result, { + onSome: (value) => value, + onNone: () => + ({ + stdout: "", + stderr: "", + exitCode: null, + timedOut: true, + stdoutTruncated: false, + stderrTruncated: false, + }) satisfies ProviderMaintenanceCommandResult, + }), + ), + ); +}); + +export const layerInstaller = Layer.effect( + ProviderInstaller, + Effect.gen(function* () { + const fileSystem = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const spawner = yield* ChildProcessSpawner.ChildProcessSpawner; + return ProviderInstaller.of({ + run: Effect.fn("ProviderInstaller.run")((input) => + runDetachedInstall(input).pipe( + Effect.catchTags({ + NodeRuntimeUnavailableError: () => runForegroundInstall({ ...input, spawner }), + }), + Effect.provideService(FileSystem.FileSystem, fileSystem), + Effect.provideService(Path.Path, path), + ), ), - ); - }, + }); + }), ); export function splitOutputLines( @@ -266,7 +547,7 @@ function lastOutputLine(result: ProviderMaintenanceCommandResult): string | null function failureMessage(result: ProviderMaintenanceCommandResult): string { if (result.timedOut) { - return `Update timed out after ${UPDATE_TIMEOUT_MINUTES} minutes.`; + return `Update timed out after ${Duration.toMinutes(INSTALL_BACKSTOP)} minutes.`; } const summary = result.exitCode !== null && result.exitCode !== 0 @@ -304,20 +585,10 @@ function makeUpdateState(input: { export const make = Effect.fn("ProviderMaintenanceRunner.make")(function* () { const providerRegistry = yield* ProviderRegistry.ProviderRegistry; const manifestService = yield* ModelManifest.ModelManifest; - const spawner = yield* ChildProcessSpawner.ChildProcessSpawner; + const installer = yield* ProviderInstaller; + const installLock = yield* ProviderInstallLock.ProviderInstallLock; const httpClient = yield* HttpClient.HttpClient; const latestVersions = yield* ProviderLatestVersions.ProviderLatestVersions; - const runMaintenanceCommand = ( - update: ProviderMaintenanceCommandAction, - onProgress: (line: string) => Effect.Effect, - ) => - runProviderMaintenanceCommandWithSpawner({ - spawner, - command: update.executable, - args: update.args, - onProgress, - ...(update.env ? { env: update.env } : {}), - }); const commandCoordinator = yield* makeProviderMaintenanceCommandCoordinator({ makeAlreadyRunningError: () => new ServerProviderUpdateError({ @@ -390,6 +661,8 @@ export const make = Effect.fn("ProviderMaintenanceRunner.make")(function* () { ? defaultInstanceIdForDriver(provider) : (target.instanceId ?? defaultInstanceIdForDriver(provider)); const targetVersion = typeof target === "string" ? undefined : target.targetVersion; + const background = typeof target !== "string" && target.background === true; + const beforeInstall = typeof target === "string" ? undefined : target.beforeInstall; const targetKey = `instance:${instanceId}`; const capabilities = yield* providerRegistry.getProviderMaintenanceCapabilitiesForInstance( instanceId, @@ -403,12 +676,15 @@ export const make = Effect.fn("ProviderMaintenanceRunner.make")(function* () { }); } + // A background install is silent: nobody asked for it, so clients never see its state. const setUpdateState = (state: ServerProviderUpdateState | null) => - providerRegistry.setProviderMaintenanceActionState({ - instanceId, - action: "update", - state, - }); + background + ? providerRegistry.getProviders + : providerRegistry.setProviderMaintenanceActionState({ + instanceId, + action: "update", + state, + }); const setQueuedState = setUpdateState( makeUpdateState({ status: "queued", @@ -421,11 +697,20 @@ export const make = Effect.fn("ProviderMaintenanceRunner.make")(function* () { const runProviderUpdate = Effect.fn("ProviderMaintenanceRunner.runProviderUpdate")( function* () { const finish = (state: ServerProviderUpdateState) => - setUpdateState(state).pipe(Effect.map((providers) => ({ providers }))); + (background + ? (state.status === "failed" ? Effect.logWarning : Effect.logInfo)( + "Background provider update finished", + { instanceId, status: state.status, message: state.message }, + ) + : Effect.void + ).pipe( + Effect.andThen(setUpdateState(state)), + Effect.map((providers) => ({ providers })), + ); const startedAtRef = yield* Ref.make(null); const runCommandAndVerify = Effect.fn("ProviderMaintenanceRunner.runCommandAndVerify")( - function* () { + function* (recordChildPid: ProviderInstallInput["recordChildPid"]) { const startedAt = yield* nowIso; yield* Ref.set(startedAtRef, startedAt); const setRunningMessage = (message: string) => @@ -445,7 +730,8 @@ export const make = Effect.fn("ProviderMaintenanceRunner.make")(function* () { provider, { fresh: true }, ); - if (!fresh.update || fresh.update.lockKey !== update.lockKey) { + const freshUpdate = fresh.update; + if (!freshUpdate || freshUpdate.lockKey !== update.lockKey) { return yield* finish( makeUpdateState({ status: "failed", @@ -457,17 +743,25 @@ export const make = Effect.fn("ProviderMaintenanceRunner.make")(function* () { } const manifest = yield* manifestService.current; + const pinnedCommand = + targetVersion !== undefined + ? makeTargetedProviderUpdateAction(fresh, targetVersion) + : null; // Check compatibility against what an unpinned command would install now. const candidateVersion = - targetVersion ?? - (yield* invalidateLatestProviderVersion(fresh).pipe( - Effect.andThen(resolveLatestProviderVersion(fresh)), - Effect.provideService(HttpClient.HttpClient, httpClient), - Effect.provideService( - ProviderLatestVersions.ProviderLatestVersions, - latestVersions, - ), - )); + background && !pinnedCommand + ? fresh.lookupLatestVersion + ? yield* fresh.lookupLatestVersion + : null + : (targetVersion ?? + (yield* invalidateLatestProviderVersion(fresh).pipe( + Effect.andThen(resolveLatestProviderVersion(fresh)), + Effect.provideService(HttpClient.HttpClient, httpClient), + Effect.provideService( + ProviderLatestVersions.ProviderLatestVersions, + latestVersions, + ), + ))); const advisory = resolveProviderCompatibility(manifest.compatibility, provider, candidateVersion) ?? resolveProviderCompatibility( @@ -475,16 +769,25 @@ export const make = Effect.fn("ProviderMaintenanceRunner.make")(function* () { provider, candidateVersion, ); - const command = - targetVersion !== undefined - ? makeTargetedProviderUpdateAction(fresh, targetVersion) + // Unattended installs run a pinned `package@version`; only a native + // self-updater or Homebrew may run unpinned. + const command = background + ? (pinnedCommand ?? + (SELF_UPDATING_LOCK_KEY.test(freshUpdate.lockKey) ? freshUpdate : null)) + : targetVersion !== undefined + ? pinnedCommand : resolveLatestProviderUpdateAction(fresh, candidateVersion); - const rejected = - targetVersion !== undefined + const incompatible = + advisory?.status === "broken" || advisory?.status === "unsupported"; + const rejected = background + ? incompatible || + targetVersion === undefined || + (!pinnedCommand && (!fresh.lookupLatestVersion || !candidateVersion)) + : targetVersion !== undefined ? !command || advisory?.recommendedVersion !== targetVersion || advisory.status !== "supported" - : advisory?.status === "broken" || advisory?.status === "unsupported"; + : incompatible; if (rejected || !command) { return yield* finish( makeUpdateState({ @@ -498,7 +801,41 @@ export const make = Effect.fn("ProviderMaintenanceRunner.make")(function* () { }), ); } - + const compareVersions = fresh.compareVersions ?? compareSemverVersions; + if (background && candidateVersion) { + // Re-probe under the lock: another server or a manual update may have won. + const current = (yield* providerRegistry.refreshInstance(instanceId)).find( + (candidate) => candidate.instanceId === instanceId, + ); + const currentVersion = + current?.installedVersion !== undefined + ? current.installedVersion + : current?.version; + if ( + !current?.enabled || + !current.installed || + (currentVersion && compareVersions(currentVersion, candidateVersion) >= 0) + ) { + return yield* finish( + makeUpdateState({ + status: "unchanged", + startedAt, + finishedAt: yield* nowIso, + message: "No newer provider version is available.", + }), + ); + } + if (beforeInstall && !(yield* beforeInstall)) { + return yield* finish( + makeUpdateState({ + status: "unchanged", + startedAt, + finishedAt: yield* nowIso, + message: "Background update postponed until nothing is in use.", + }), + ); + } + } if (targetVersion === undefined) { const prior = (yield* providerRegistry.getProviders).find( (candidate) => candidate.instanceId === instanceId, @@ -534,7 +871,17 @@ export const make = Effect.fn("ProviderMaintenanceRunner.make")(function* () { } yield* setRunningMessage(`Running ${describeCommand(command)}`); - const result = yield* runMaintenanceCommand(command, setRunningMessage); + const result = yield* installer.run({ + command: command.executable, + args: command.args, + env: + background && freshUpdate.lockKey === "homebrew" + ? { ...command.env, HOMEBREW_NO_AUTO_UPDATE: "1" } + : command.env, + // Nobody watches a background install, so its output is not sampled. + onProgress: background ? undefined : setRunningMessage, + recordChildPid, + }); const finishedAt = yield* nowIso; if (result.timedOut || result.exitCode !== 0) { yield* Effect.logWarning("Provider update command failed", { @@ -584,8 +931,12 @@ export const make = Effect.fn("ProviderMaintenanceRunner.make")(function* () { (verifiedProvider.status === "error" && (verifiedProvider.auth.status !== "unauthenticated" || !verifiedProvider.version)) || - (targetVersion !== undefined && - installedVersion?.replace(/^v/, "") !== targetVersion) + (background + ? !installedVersion || + !candidateVersion || + compareVersions(installedVersion, candidateVersion) < 0 + : targetVersion !== undefined && + installedVersion?.replace(/^v/, "") !== targetVersion) ); }); const stillOutdated = @@ -623,7 +974,28 @@ export const make = Effect.fn("ProviderMaintenanceRunner.make")(function* () { }, ); - return yield* runCommandAndVerify().pipe(Effect.catchCause(recordFailedUpdate)); + // Held until the installer exits, so two servers never run the same installer at once. + return yield* installLock.tryRun(update.lockKey, runCommandAndVerify, provider).pipe( + Effect.flatMap( + Option.match({ + onSome: Effect.succeed, + onNone: () => + nowIso.pipe( + Effect.flatMap((finishedAt) => + finish( + makeUpdateState({ + status: "failed", + startedAt: null, + finishedAt, + message: "Another T3 server is already updating this provider.", + }), + ), + ), + ), + }), + ), + Effect.catchCause(recordFailedUpdate), + ); }, ); @@ -646,6 +1018,16 @@ export const make = Effect.fn("ProviderMaintenanceRunner.make")(function* () { ); }); + // A previous server may have quit while its detached installer kept running. + // Wait for it, then pick up the version it installed. + yield* installLock.anyHeld.pipe( + Effect.flatMap((held) => + held ? installLock.untilIdle.pipe(Effect.andThen(providerRegistry.refresh())) : Effect.void, + ), + Effect.ignore, + Effect.forkScoped, + ); + return ProviderMaintenanceRunner.of({ updateProvider, }); diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index 47190f60dcf51..59a426c159b71 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -199,6 +199,10 @@ import * as CursorUsageAccounts from "@t3tools/provider-cursor/server/CursorUsag import * as UsageService from "./usage/UsageService.ts"; import * as CodexUsageHistory from "./usage/CodexUsageHistory.ts"; import * as RuntimeLayer from "./orchestration-v2/runtimeLayer.ts"; +import * as ProviderInstallLock from "./provider/ProviderInstallLock.ts"; +import * as ProviderMaintenanceRunner from "./provider/providerMaintenanceRunner.ts"; +import * as ProviderAutoUpdater from "./updates/ProviderAutoUpdater.ts"; +import * as UpdateWindow from "./updates/UpdateWindow.ts"; import * as ProjectStore from "./orchestration-v2/ProjectStore.ts"; import * as ThreadSearch from "./orchestration-v2/ThreadSearch.ts"; import * as ThreadSettlementService from "./orchestration-v2/ThreadSettlementService.ts"; @@ -641,7 +645,17 @@ const layerServerRestart = Layer.effect( }), ); +// One maintenance runner for RPCs and background updates, so both queue +// behind the same per-installer lock. +const layerProviderUpdates = ProviderAutoUpdater.layer.pipe( + Layer.provide(UpdateWindow.layer.pipe(Layer.provide(ProjectionStoreV2.layer))), + Layer.provideMerge( + ProviderMaintenanceRunner.layer.pipe(Layer.provide(ProviderMaintenanceRunner.layerInstaller)), + ), +); + const layerRuntimeCoreDependenciesBase = Layer.mergeAll( + layerProviderUpdates, AgentAwarenessRelay.layer, HeldHooksWaker.layer, layerThreadSettlementWorker, @@ -1222,6 +1236,8 @@ const layerMakeServer = Layer.unwrap( Layer.provideMerge(FetchHttpClient.layer), Layer.provide(VcsProcess.layer), Layer.provideMerge(AgentScopeLive.layer), + // Provider installs and turn starts coordinate with every T3 server on the host. + Layer.provideMerge(ProviderInstallLock.layer), Layer.provideMerge(layerPlatformServices), ); }), diff --git a/apps/server/src/serverSettings.test.ts b/apps/server/src/serverSettings.test.ts index 347b43dd6f44c..297e20315c3d8 100644 --- a/apps/server/src/serverSettings.test.ts +++ b/apps/server/src/serverSettings.test.ts @@ -2373,6 +2373,8 @@ it.layer(Layer.mergeAll(NodeServices.layer, NodeCrypto.layer))("server settings" const settings = yield* serverSettings.getSettings; assert.isFalse(settings.projectSettingsFolded); assert.isFalse(settings.continueThreadsAfterServerUpdate); + // An unreadable file may hold an opt-out, so nothing installs unattended on a guess. + assert.isFalse(settings.automaticUpdates); assert.deepEqual(settings.projectSettingsOverrides, {}); assert.equal(yield* fileSystem.readFileString(serverConfig.settingsPath), broken); diff --git a/apps/server/src/serverSettings.ts b/apps/server/src/serverSettings.ts index 251becba12f6c..1dd7db2cf05dc 100644 --- a/apps/server/src/serverSettings.ts +++ b/apps/server/src/serverSettings.ts @@ -805,8 +805,13 @@ const make = Effect.gen(function* () { } if (decoded._tag === "Failure" || legacySettings._tag === "Failure") { const failure = decoded._tag === "Failure" ? decoded : legacySettings; - // The file may hold an opt-out we can no longer read; never resume threads on a guess. - settings = { ...managed.defaults, continueThreadsAfterServerUpdate: false }; + // The file may hold an opt-out we can no longer read; never resume threads or install + // updates on a guess. + settings = { + ...managed.defaults, + continueThreadsAfterServerUpdate: false, + automaticUpdates: false, + }; if (failure._tag === "Failure") { parseFailure = Option.some(Cause.squash(failure.cause)); yield* Effect.logWarning("failed to parse settings.json, using defaults", { diff --git a/apps/server/src/updates/ProviderAutoUpdater.ts b/apps/server/src/updates/ProviderAutoUpdater.ts new file mode 100644 index 0000000000000..3e859fdfaa2b7 --- /dev/null +++ b/apps/server/src/updates/ProviderAutoUpdater.ts @@ -0,0 +1,129 @@ +import type { ProviderInstanceId } from "@t3tools/contracts"; +import { enrichProviderSnapshotWithVersionAdvisory } from "@t3tools/provider-core/server/maintenanceResolver"; +import * as ProviderLatestVersions from "@t3tools/provider-core/server/ProviderLatestVersions"; +import * as Clock from "effect/Clock"; +import * as Duration from "effect/Duration"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as Schedule from "effect/Schedule"; +import { HttpClient } from "effect/http"; + +import * as ServerConfig from "../config.ts"; +import * as ProviderRegistry from "../provider/ProviderRegistry.ts"; +import * as ProviderMaintenanceRunner from "../provider/providerMaintenanceRunner.ts"; +import { forkParked } from "../serverActivation.ts"; +import * as ServerSettings from "../serverSettings.ts"; +import * as UpdateWindow from "./UpdateWindow.ts"; + +/** An instance is tried once per release, then again only after this long. */ +const RETRY_AFTER = Duration.hours(6); +const FIRST_PASS_DELAY = Duration.minutes(2); +const PASS_INTERVAL = Duration.minutes(10); + +/** + * Installs provider CLI updates in the background when `automaticUpdates` is + * on, one instance at a time, only while the update window is open. + */ +export const layer = Layer.effectDiscard( + Effect.gen(function* () { + // A dev server shares the host's CLIs with the real install; leave them to it. + if ((yield* ServerConfig.ServerConfig).devUrl !== undefined) return; + const providers = yield* ProviderRegistry.ProviderRegistry; + const runner = yield* ProviderMaintenanceRunner.ProviderMaintenanceRunner; + const updateWindow = yield* UpdateWindow.UpdateWindow; + const serverSettings = yield* ServerSettings.ServerSettingsService; + const httpClient = yield* HttpClient.HttpClient; + const latestVersions = yield* ProviderLatestVersions.ProviderLatestVersions; + // `instanceId@version` to when it was last tried, on the monotonic clock. + const attempts = new Map(); + + const enabled = serverSettings.getSettings.pipe( + Effect.map((settings) => settings.automaticUpdates && settings.enableProviderUpdateChecks), + Effect.orElseSucceed(() => false), + ); + + /** Chooses what to install for one instance and spends its retry budget, or returns undefined. */ + const pick = Effect.fn("ProviderAutoUpdater.pick")(function* (instanceId: ProviderInstanceId) { + if (!(yield* enabled)) return undefined; + const current = (yield* providers.getProviders).find( + (candidate) => candidate.instanceId === instanceId, + ); + const updateStatus = current?.updateState?.status; + if ( + !current?.enabled || + !current.installed || + updateStatus === "queued" || + updateStatus === "running" + ) { + return undefined; + } + // A headless service has no client refreshing advisories, so look up the newest release here. + const capabilities = yield* providers.getProviderMaintenanceCapabilitiesForInstance( + instanceId, + current.driver, + ); + const advisory = (yield* enrichProviderSnapshotWithVersionAdvisory(current, capabilities, { + enableProviderUpdateChecks: true, + }).pipe( + Effect.provideService(HttpClient.HttpClient, httpClient), + Effect.provideService(ProviderLatestVersions.ProviderLatestVersions, latestVersions), + )).versionAdvisory; + const targetVersion = advisory?.latestVersion; + if (advisory?.status !== "behind_latest" || !advisory.canUpdate || !targetVersion) { + return undefined; + } + const nowMs = Number(yield* Clock.monotonicTimeNanos) / 1_000_000; + for (const [key, attemptedAt] of attempts) { + if (nowMs - attemptedAt >= Duration.toMillis(RETRY_AFTER)) attempts.delete(key); + } + const attemptKey = `${instanceId}@${targetVersion}`; + if (attempts.has(attemptKey)) return undefined; + attempts.set(attemptKey, nowMs); + yield* Effect.logInfo("Updating provider in the background", { + instanceId, + fromVersion: current.version, + targetVersion, + }); + return { provider: current.driver, targetVersion, attemptKey }; + }); + + const pass = Effect.gen(function* () { + if (!(yield* enabled)) return; + for (const { instanceId } of yield* providers.getProviders) { + // Rechecked per install: the previous one may have run for minutes. + const picked = yield* updateWindow.runIfOpen(pick(instanceId)); + if (Option.isNone(picked)) return; + if (picked.value === undefined) continue; + const { provider, targetVersion, attemptKey } = picked.value; + // Runs under the install lock, which would close the window on itself. + const beforeInstall = Effect.gen(function* () { + const allowed = + (yield* enabled) && (yield* updateWindow.isOpen({ ignoreInstallLocks: true })); + // Postponed, not failed: the next open window gets a fresh try. + if (!allowed) attempts.delete(attemptKey); + return allowed; + }).pipe(Effect.catchCause(() => Effect.succeed(false))); + yield* runner + .updateProvider({ provider, instanceId, targetVersion, background: true, beforeInstall }) + .pipe( + Effect.catchCause((cause) => + Effect.logWarning("Background provider update failed", { instanceId, cause }), + ), + ); + return; + } + }).pipe( + Effect.catchCause((cause) => + Effect.logWarning("Provider auto-update pass failed", { cause }), + ), + Effect.withSpan("updates.ProviderAutoUpdater.pass"), + ); + + yield* forkParked( + Effect.sleep(FIRST_PASS_DELAY).pipe( + Effect.andThen(pass.pipe(Effect.repeat(Schedule.spaced(PASS_INTERVAL)))), + ), + ); + }), +); diff --git a/apps/server/src/updates/UpdateWindow.ts b/apps/server/src/updates/UpdateWindow.ts new file mode 100644 index 0000000000000..219e96ac4972d --- /dev/null +++ b/apps/server/src/updates/UpdateWindow.ts @@ -0,0 +1,67 @@ +import * as Context from "effect/Context"; +import * as Duration from "effect/Duration"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; + +import * as BackgroundPolicy from "../background/BackgroundPolicy.ts"; +import * as ProjectionStore from "../orchestration-v2/ProjectionStore.ts"; +import * as ProviderInstallLock from "../provider/ProviderInstallLock.ts"; + +/** How long every client must be untouched before an update may run. */ +const QUIET_PERIOD = Duration.minutes(2); + +/** + * Decides when a background update may replace something without anyone + * noticing: no client was used for 2 minutes, no run is active, and no + * provider installer is running on this host. + */ +export class UpdateWindow extends Context.Service< + UpdateWindow, + { + /** + * Whether the window is open. A failed read counts as closed: guessing + * "idle" could cut off a run. A caller that already holds an install lock + * passes `ignoreInstallLocks`, or its own lock would close the window. + */ + readonly isOpen: (options?: { + readonly ignoreInstallLocks?: boolean; + }) => Effect.Effect; + /** Runs `effect` if the window is open, otherwise returns `None` without running it. */ + readonly runIfOpen: ( + effect: Effect.Effect, + ) => Effect.Effect, E, R>; + } +>()("t3/updates/UpdateWindow") {} + +const make = Effect.gen(function* () { + const backgroundPolicy = yield* BackgroundPolicy.BackgroundPolicy; + const projections = yield* ProjectionStore.ProjectionStoreV2; + const installLock = yield* ProviderInstallLock.ProviderInstallLock; + + const isOpen: UpdateWindow["Service"]["isOpen"] = Effect.fn("updates.UpdateWindow.isOpen")( + function* (options) { + if (Duration.isLessThan(yield* backgroundPolicy.sinceLastClientInteraction, QUIET_PERIOD)) { + return false; + } + // Preparing, starting, running, or waiting on an approval or question. + const activeThreadIds = yield* projections.getRecoveryThreadIds("active-work").pipe( + Effect.tapError((cause) => + Effect.logWarning("Update window could not read active runs", { cause }), + ), + Effect.option, + ); + if (Option.isNone(activeThreadIds) || activeThreadIds.value.length > 0) return false; + if (options?.ignoreInstallLocks) return true; + // Covers manual updates and installs started by other servers on this host. + return !(yield* installLock.anyHeld.pipe(Effect.orElseSucceed(() => true))); + }, + ); + + const runIfOpen: UpdateWindow["Service"]["runIfOpen"] = (effect) => + isOpen().pipe(Effect.flatMap((open) => (open ? Effect.asSome(effect) : Effect.succeedNone))); + + return UpdateWindow.of({ isOpen, runIfOpen }); +}); + +export const layer = Layer.effect(UpdateWindow, make); diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 1c204481350d6..e3393d732893f 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -3530,6 +3530,8 @@ export const layer = Layer.unwrap( const serverBrowser = yield* ServerBrowser.ServerBrowser; const serverSelfUpdate = yield* ServerSelfUpdate.ServerSelfUpdate; const npmRelease = yield* NpmRelease.NpmRelease; + // Shared with background updates so every caller queues behind one lock per installer. + const providerMaintenanceRunner = yield* ProviderMaintenanceRunner.ProviderMaintenanceRunner; const pullRequests = yield* PullRequestService.PullRequestService; const reviewGuides = yield* PullRequestReviewGuideService.PullRequestReviewGuideService; const peerLinks = yield* PeerLinks.PeerLinks; @@ -3609,7 +3611,12 @@ export const layer = Layer.unwrap( Layer.provide(DefectReporter.layer), Layer.provide(Layer.succeed(SqlClient.SqlClient, sql)), Layer.provide(AgentSessionScanner.layer), - Layer.provide(ProviderMaintenanceRunner.layer), + Layer.provide( + Layer.succeed( + ProviderMaintenanceRunner.ProviderMaintenanceRunner, + providerMaintenanceRunner, + ), + ), Layer.provide(Layer.succeed(NpmRelease.NpmRelease, npmRelease)), Layer.provide(Layer.succeed(ServerSelfUpdate.ServerSelfUpdate, serverSelfUpdate)), // Share server-lifetime PR caches so WS mutations invalidate every client's HTTP diffs. diff --git a/apps/web/src/components/settings/SettingsPanels.tsx b/apps/web/src/components/settings/SettingsPanels.tsx index a37bf1581990e..1e43b10423205 100644 --- a/apps/web/src/components/settings/SettingsPanels.tsx +++ b/apps/web/src/components/settings/SettingsPanels.tsx @@ -545,6 +545,9 @@ export function useSettingsRestore(onRestored?: () => void) { ...(settings.enableProviderUpdateChecks !== DEFAULT_UNIFIED_SETTINGS.enableProviderUpdateChecks ? ["Provider update checks"] : []), + ...(settings.automaticUpdates !== DEFAULT_UNIFIED_SETTINGS.automaticUpdates + ? ["Update automatically"] + : []), ...(settings.continueThreadsAfterServerUpdate !== DEFAULT_UNIFIED_SETTINGS.continueThreadsAfterServerUpdate ? ["Continue threads after restarts"] @@ -668,6 +671,7 @@ export function useSettingsRestore(onRestored?: () => void) { snoozePresets: DEFAULT_UNIFIED_SETTINGS.snoozePresets, responseStreamingMode: DEFAULT_UNIFIED_SETTINGS.responseStreamingMode, enableProviderUpdateChecks: DEFAULT_UNIFIED_SETTINGS.enableProviderUpdateChecks, + automaticUpdates: DEFAULT_UNIFIED_SETTINGS.automaticUpdates, continueThreadsAfterServerUpdate: DEFAULT_UNIFIED_SETTINGS.continueThreadsAfterServerUpdate, backgroundActivity: DEFAULT_UNIFIED_SETTINGS.backgroundActivity, backgroundActivityProfile: DEFAULT_UNIFIED_SETTINGS.backgroundActivityProfile, @@ -1953,6 +1957,12 @@ export function MaintenanceSettingsPanel() { const isEnvironmentScope = scope.environmentIds.length === 1 && environmentId !== null; const hasServerTargets = connectedEnvironments.length > 0; + const supportsAutomaticUpdates = + connectedEnvironments.length > 0 && + connectedEnvironments.every( + (target) => target.serverConfig?.environment.capabilities.automaticUpdates === true, + ); + return ( @@ -1985,6 +1995,40 @@ export function MaintenanceSettingsPanel() { /> } /> + + updateSettings({ automaticUpdates: DEFAULT_UNIFIED_SETTINGS.automaticUpdates }) + } + /> + ) : null + } + control={ + updateSettings({ automaticUpdates: Boolean(checked) })} + aria-label="Update automatically" + /> + } + /> + Effect.sync(() => { @@ -234,10 +235,10 @@ export const layer = Layer.effectDiscard( window.removeEventListener("focus", requestReport); window.removeEventListener("blur", requestReport); window.removeEventListener("online", requestReport); - window.removeEventListener("pointermove", recordInteraction); - window.removeEventListener("keydown", recordInteraction); - window.removeEventListener("wheel", recordInteraction); - window.removeEventListener("touchstart", recordInteraction); + window.removeEventListener("pointermove", recordInteraction, interactionListenerOptions); + window.removeEventListener("keydown", recordInteraction, interactionListenerOptions); + window.removeEventListener("wheel", recordInteraction, interactionListenerOptions); + window.removeEventListener("touchstart", recordInteraction, interactionListenerOptions); }), ); diff --git a/docs/internals/providers.md b/docs/internals/providers.md index 634b83980a540..e0c22d94e51c2 100644 --- a/docs/internals/providers.md +++ b/docs/internals/providers.md @@ -112,6 +112,12 @@ Ownership is cached per instance and re-read immediately before an update runs. changed since the advisory, and reports success only when the refreshed provider is still installed with a readable, current version. +Background updates (`automaticUpdates`) use the same runner but install the checked release as +`package@version`; only native self-updaters and Homebrew run unpinned. A lock file per `lockKey` +in the user cache directory is shared by every T3 server on the host, and the installer runs +detached with its output in a log file, so quitting the server never cuts an install in half. A turn +start waits for any held lock before it reads its context, so a CLI is never replaced under a launch. + ## Protocol traps Codex async questions arrive as notifications and are answered with a new user message. There is diff --git a/docs/user/updating.md b/docs/user/updating.md index 7e62c7bf580fc..837321aba8a71 100644 --- a/docs/user/updating.md +++ b/docs/user/updating.md @@ -168,6 +168,15 @@ Machines without npm check the public npm registry directly. ## Update providers +**Settings → Maintenance → Update automatically** is on by default. With +**Provider update checks** also on, each environment installs provider CLI +updates in the background once no client has been used for 2 minutes and no +thread is running or waiting for you. Only CLIs that support one-click updates +are updated, each to the release T3 Code checked. A failed update is retried +after 6 hours or when a newer version is published. While `settings.json` +cannot be read, automatic updates stay off, and a server started for +development never updates provider CLIs. + **Settings → Providers** shows provider updates for the selected environment. **Update all** updates every outdated provider on every connected environment at once. Hover it to see which providers it will update. A provider is left out diff --git a/packages/contracts/src/environment.ts b/packages/contracts/src/environment.ts index 5e9cca70752b6..560014c2c5854 100644 --- a/packages/contracts/src/environment.ts +++ b/packages/contracts/src/environment.ts @@ -142,6 +142,8 @@ export const ExecutionEnvironmentCapabilities = Schema.Struct({ serverInstallation: ForwardCompatibleOptional(ServerInstallation), serverSelfUpdateProgress: Schema.optionalKey(Schema.Boolean), serverUpdateThreadContinuation: Schema.optionalKey(Schema.Boolean), + /** Server honors the automaticUpdates setting. */ + automaticUpdates: Schema.optionalKey(Schema.Boolean), /** Only explicit false skips Live Activity seeding; older servers may still publish. */ agentActivityPublishing: Schema.optionalKey(Schema.Boolean), projectCloneTracking: Schema.optionalKey(Schema.Boolean), diff --git a/packages/contracts/src/settings.ts b/packages/contracts/src/settings.ts index cd112d84bf9f4..397804377e49f 100644 --- a/packages/contracts/src/settings.ts +++ b/packages/contracts/src/settings.ts @@ -1352,6 +1352,8 @@ export const ServerSettings = Schema.Struct({ Schema.withDecodingDefault(Effect.succeed("paragraph" as const)), ), enableProviderUpdateChecks: Schema.Boolean.pipe(Schema.withDecodingDefault(Effect.succeed(true))), + /** Install provider CLI updates in the background once nothing is running and nobody is using the app. */ + automaticUpdates: Schema.Boolean.pipe(Schema.withDecodingDefault(Effect.succeed(true))), continueThreadsAfterServerUpdate: Schema.Boolean.pipe( Schema.withDecodingDefault(Effect.succeed(true)), ), @@ -1801,6 +1803,7 @@ export const ServerSettingsPatch = Schema.Struct({ fastWorktreeCheckout: Schema.optionalKey(Schema.Boolean), responseStreamingMode: Schema.optionalKey(ResponseStreamingMode), enableProviderUpdateChecks: Schema.optionalKey(Schema.Boolean), + automaticUpdates: Schema.optionalKey(Schema.Boolean), continueThreadsAfterServerUpdate: Schema.optionalKey(Schema.Boolean), enableAgentBrowserAccess: Schema.optionalKey(Schema.Boolean), cloudSpeechCredentials: Schema.optionalKey( diff --git a/packages/provider-core/src/server/failure.ts b/packages/provider-core/src/server/failure.ts index edaf0aa97a743..cf19e2d677f06 100644 --- a/packages/provider-core/src/server/failure.ts +++ b/packages/provider-core/src/server/failure.ts @@ -16,6 +16,19 @@ import * as Cause from "effect/Cause"; import * as IdAllocator from "./IdAllocator.ts"; import type { ProviderAdapterFailureReason } from "./ProviderAdapter.ts"; import * as Schema from "effect/Schema"; +import { + PROVIDER_DISPLAY_NAMES, + ProviderDriverKind as ProviderDriverKindSchema, +} from "@t3tools/contracts"; + +export class ProviderUpdatingError extends Schema.TaggedError()( + "ProviderUpdatingError", + { driver: ProviderDriverKindSchema, newNativeThread: Schema.Boolean, cause: Schema.Defect() }, +) { + override get message() { + return `${PROVIDER_DISPLAY_NAMES[this.driver] ?? this.driver} was updating on this machine. Try again.`; + } +} export class ContextHandoffBudgetError extends Schema.TaggedError()( "ContextHandoffBudgetError", @@ -78,6 +91,7 @@ const PROVIDER_FAILURE_REASONS: Record = { function curatedCauseMessages(cause: unknown): { readonly known: string | undefined; readonly category: string | undefined; + readonly code?: string; } { const seen = new Set(); let category: (typeof PROVIDER_FAILURE_CATEGORIES)[string]; @@ -130,6 +144,15 @@ function curatedCauseMessages(cause: unknown): { } break; } + case "ProviderUpdatingError": + return { + known: stringField(cause, "message"), + category: categoryMessage(), + code: + (cause as Record).newNativeThread === true + ? "provider_updating_new_thread" + : "provider_updating", + }; case "AcpProcessExitedError": { const exit = `provider process exited${processExitStatus(cause)}`; const stage = category === undefined ? `The ${exit}` : `${category.summary}: the ${exit}`; @@ -266,7 +289,7 @@ export function makeProviderFailure(input: { const rawMessage = input.message ?? curated.known ?? curated.category ?? DEFAULT_PROVIDER_FAILURE_MESSAGE; const message = boundedText(rawMessage, MAX_PROVIDER_FAILURE_MESSAGE_LENGTH); - const rawCode = input.code ?? stringField(input.cause, "code") ?? null; + const rawCode = input.code ?? curated.code ?? stringField(input.cause, "code") ?? null; const code = rawCode === null ? null : boundedText(rawCode, MAX_PROVIDER_FAILURE_CODE_LENGTH) || null; diff --git a/packages/provider-opencode/src/server/driver.test.ts b/packages/provider-opencode/src/server/driver.test.ts index 665f4785459d9..ebe4e34c66220 100644 --- a/packages/provider-opencode/src/server/driver.test.ts +++ b/packages/provider-opencode/src/server/driver.test.ts @@ -589,9 +589,9 @@ it.effect.skipIf(HostProcess.Platform.defaultValue() === "win32")( const native = executable(".opencode", "bin", "opencode"); for (const generation of ["v1", "v2"] as const) { assert.isNull((yield* resolveUpdate(generation, launcher)).update); - assert.deepStrictEqual((yield* resolveUpdate(generation, native)).update?.args, [ - "upgrade", - ]); + const update = (yield* resolveUpdate(generation, native)).update; + assert.deepStrictEqual(update?.args, ["upgrade", "--method", "curl"]); + assert.strictEqual(update?.env?.PATH, `${NodePath.dirname(native)}${NodePath.delimiter}`); } }), ); diff --git a/packages/provider-opencode/src/server/driver.ts b/packages/provider-opencode/src/server/driver.ts index 0da1a04ff1fd7..4d27c658bde62 100644 --- a/packages/provider-opencode/src/server/driver.ts +++ b/packages/provider-opencode/src/server/driver.ts @@ -84,7 +84,17 @@ export const openCodeUpdateFor = (generation: ProbedOpenCode["generation"]) => provider: DRIVER_KIND, npmPackageName: generation === "v2" ? "@opencode/cli" : "opencode-ai", nativeUpdate: { - args: ["upgrade"], + args: ["upgrade", "--method", "curl"], + env: ({ resolvedCommandPath, env, platform }) => ({ + // The curl installer checks `opencode` on PATH before replacing its native binary. + PATH: + resolvedCommandPath.slice( + 0, + Math.max(resolvedCommandPath.lastIndexOf("/"), resolvedCommandPath.lastIndexOf("\\")), + ) + + (platform === "win32" ? ";" : ":") + + (env.PATH ?? env.Path ?? env.path ?? ""), + }), isCommandPath: isOpenCodeNativeCommandPath, requiresCommandPath: true, }, diff --git a/packages/shared/src/nodeRuntime.ts b/packages/shared/src/nodeRuntime.ts index 7b3e9398682cc..9d3bd39725b35 100644 --- a/packages/shared/src/nodeRuntime.ts +++ b/packages/shared/src/nodeRuntime.ts @@ -12,6 +12,7 @@ const NodeRuntimeFeature = Schema.Literals([ "Device automation", "Antigravity", "Antigravity sign-in", + "Provider CLI updates", ]); export const nodeRuntimeUnavailableMessage = (feature: typeof NodeRuntimeFeature.Type): string =>