From f6ff49c0f7bb098252203b9a734b8115b6d865d4 Mon Sep 17 00:00:00 2001 From: unlockdigital Date: Tue, 27 Oct 2026 08:01:25 -0500 Subject: [PATCH 1/3] fix(core): improve provider updates and fork reliability Respect each Claude connection home and native release channel when checking and applying updates. Add bounded event-write retries and Codex closure handling, preserve delegated completion state, restore interrupted Claude context, and improve attachments, branch freshness, and desktop shutdown. Backport upstream #24571, #24600, #25695, #25672, #25666, #24554, #23873, #22807, and #21356 with fork-specific compatibility guards and regression coverage. Validated 653 affected tests, affected typechecks, lint, formatting, knip, generated contracts, and desktop/web/server builds. Model: GPT-6.1-sol with Luna subagents. Harness: Codex in T3 Code. --- apps/desktop/src/app/DesktopLifecycle.test.ts | 58 +++++- apps/desktop/src/app/DesktopLifecycle.ts | 75 ++++++-- apps/server/src/auth/EnvironmentAuth.test.ts | 176 +++++++++++++++-- apps/server/src/auth/EnvironmentAuth.ts | 100 +++++++--- apps/server/src/auth/SessionStore.ts | 17 +- apps/server/src/auth/dpop.ts | 4 +- apps/server/src/auth/http.ts | 16 +- .../src/environment/ServerEnvironment.ts | 1 + .../Adapters/ClaudeAdapterV2.test.ts | 132 +++++++++++++ .../Adapters/ClaudeAdapterV2.ts | 3 +- .../DelegatedCompletionDelivery.test.ts | 138 ++++++++++++++ apps/server/src/orchestration-v2/EventSink.ts | 52 ++--- .../src/orchestration-v2/Orchestrator.ts | 17 +- .../ProviderEventIngestor.test.ts | 178 ++++++++++++++++-- .../orchestration-v2/ProviderEventIngestor.ts | 17 +- .../src/provider/Drivers/ClaudeDriver.ts | 6 +- .../src/provider/Drivers/ClaudeHome.test.ts | 7 +- .../src/vcs/VcsStatusBroadcaster.test.ts | 113 +++++++++++ apps/server/src/vcs/VcsStatusBroadcaster.ts | 23 ++- apps/server/src/ws.ts | 5 +- .../BranchToolbarBranchSelector.tsx | 10 +- apps/web/test/environmentHttpTest.ts | 1 + docs/internals/environment-auth.md | 12 +- .../src/authorization/layer.test.ts | 8 +- .../src/authorization/remote.test.ts | 102 ++++++++++ .../src/authorization/remote.ts | 131 ++++++++++--- .../src/authorization/service.ts | 26 +-- .../src/connection/supervisor.test.ts | 4 +- .../src/state/deviceHubAccess.ts | 29 ++- .../src/state/environmentHttpAuth.test.ts | 81 ++++++++ packages/contracts/src/auth.ts | 12 ++ packages/contracts/src/environment.ts | 2 + packages/contracts/src/environmentHttp.ts | 9 + packages/contracts/src/rpc.ts | 5 +- .../src/protocol.test.ts | 46 +++++ .../effect-codex-app-server/src/protocol.ts | 20 +- .../src/server/attachmentPrompt.test.ts | 64 +++++++ .../src/server/attachmentPrompt.ts | 30 +-- .../provider-core/src/server/maintenance.ts | 5 +- .../src/server/maintenanceResolver.test.ts | 126 ++++++++++++- .../src/server/maintenanceResolver.ts | 57 +++++- .../provider-pi/src/server/adapter.test.ts | 89 +++++++++ packages/provider-pi/src/server/adapter.ts | 10 +- 43 files changed, 1815 insertions(+), 202 deletions(-) diff --git a/apps/desktop/src/app/DesktopLifecycle.test.ts b/apps/desktop/src/app/DesktopLifecycle.test.ts index f9f38dea3dd46..6a6e991fc6b4b 100644 --- a/apps/desktop/src/app/DesktopLifecycle.test.ts +++ b/apps/desktop/src/app/DesktopLifecycle.test.ts @@ -3,6 +3,7 @@ import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as Ref from "effect/Ref"; +import * as TestClock from "effect/testing/TestClock"; import type * as Electron from "electron"; @@ -18,6 +19,7 @@ import * as DesktopWindow from "../window/DesktopWindow.ts"; function layerElectronApp( appListeners: Map void>, quit: Effect.Effect = Effect.void, + exit: (code: number) => Effect.Effect = () => Effect.void, ) { const registerListener = (eventName: string, listener: (...args: readonly unknown[]) => void) => Effect.acquireRelease( @@ -37,7 +39,7 @@ function layerElectronApp( whenReady: Effect.void, requestSingleInstanceLock: Effect.succeed(true), quit, - exit: () => Effect.void, + exit, relaunch: () => Effect.void, setPath: () => Effect.void, setName: () => Effect.void, @@ -221,6 +223,60 @@ describe("DesktopLifecycle", () => { }), ); + it.effect("exits on a repeated quit only once shutdown has stalled", () => + Effect.gen(function* () { + const appListeners = new Map void>(); + const exitCodes: number[] = []; + const exit = (code: number) => + Effect.sync(() => { + exitCodes.push(code); + }); + const layerEnvironment = Layer.succeed(DesktopEnvironment.DesktopEnvironment, { + platform: "darwin", + isDevelopment: false, + } as DesktopEnvironment.DesktopEnvironment["Service"]); + const layer = DesktopLifecycle.layer.pipe( + Layer.provideMerge(layerElectronApp(appListeners, Effect.void, exit)), + Layer.provideMerge(layerElectronTheme), + Layer.provideMerge(layerElectronWindow()), + Layer.provideMerge(layerDesktopWindow()), + Layer.provideMerge(layerEnvironment), + // Nothing marks this shutdown complete, so it stays pending. + Layer.provideMerge(DesktopShutdown.layer), + Layer.provideMerge(DesktopState.layer), + ); + + yield* Effect.scoped( + Effect.gen(function* () { + const lifecycle = yield* DesktopLifecycle.DesktopLifecycle; + yield* lifecycle.register; + + let prevented = 0; + const event = { + preventDefault: () => { + prevented += 1; + }, + } as Electron.Event; + appListeners.get("before-quit")?.(event); + appListeners.get("before-quit")?.(event); + yield* TestClock.adjust("9999 millis"); + appListeners.get("before-quit")?.(event); + assert.deepEqual( + exitCodes, + [], + "a repeat before the threshold must not cut shutdown short", + ); + + yield* TestClock.adjust("1 millis"); + appListeners.get("before-quit")?.(event); + + assert.deepEqual(exitCodes, [0]); + assert.equal(prevented, 4); + }), + ).pipe(Effect.provide(layer)); + }), + ); + it.effect("ignores app activation while quitting", () => Effect.gen(function* () { const appListeners = new Map void>(); diff --git a/apps/desktop/src/app/DesktopLifecycle.ts b/apps/desktop/src/app/DesktopLifecycle.ts index 924341781aa6b..7512d2bc1741d 100644 --- a/apps/desktop/src/app/DesktopLifecycle.ts +++ b/apps/desktop/src/app/DesktopLifecycle.ts @@ -1,4 +1,6 @@ +import * as Clock from "effect/Clock"; 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"; @@ -94,15 +96,50 @@ const requestDesktopShutdownAndWait = Effect.fn("desktop.lifecycle.requestShutdo }, ); +interface QuitState { + phase: "running" | "waiting" | "allowed"; + waitingSinceNanos: bigint; +} + +// Longer than a backend's stop budget (SIGTERM, 5s grace, SIGKILL, 1s wait), +// so a double Cmd-Q at the start of a normal quit cannot cut the backend stop +// short and orphan it. +const SHUTDOWN_STALL_THRESHOLD = Duration.seconds(10); + +const startWaitingForShutdown = Effect.fn("desktop.lifecycle.startWaitingForShutdown")(function* ( + quit: QuitState, +) { + quit.phase = "waiting"; + quit.waitingSinceNanos = yield* Clock.monotonicTimeNanos; +}); + +// While a quit waits on shutdown, `quitting` makes activation open nothing. If +// that shutdown stalls, a repeated Quit or SIGTERM is the user's only way out +// short of SIGKILL. +const exitIfShutdownStalled = Effect.fn("desktop.lifecycle.exitIfShutdownStalled")(function* ( + quit: QuitState, + source: string, +) { + const pendingMs = Number( + ((yield* Clock.monotonicTimeNanos) - quit.waitingSinceNanos) / 1_000_000n, + ); + if (pendingMs < Duration.toMillis(SHUTDOWN_STALL_THRESHOLD)) { + yield* logLifecycleInfo("quit already in progress", { source, pendingMs }); + return; + } + yield* logLifecycleError("desktop shutdown stalled; exiting without it", { source, pendingMs }); + const electronApp = yield* ElectronApp.ElectronApp; + yield* electronApp.exit(0); +}); + function handleBeforeQuit( event: Electron.Event, runEffect: ( effect: Effect.Effect, ) => Promise, - allowQuit: () => boolean, - markQuitAllowed: () => void, + quit: QuitState, ): void { - if (allowQuit()) { + if (quit.phase === "allowed") { void runEffect( Effect.gen(function* () { const state = yield* DesktopState.DesktopState; @@ -114,10 +151,15 @@ function handleBeforeQuit( } event.preventDefault(); + if (quit.phase === "waiting") { + void runEffect(exitIfShutdownStalled(quit, "before-quit")); + return; + } void runEffect( Effect.gen(function* () { const state = yield* DesktopState.DesktopState; const electronWindow = yield* ElectronWindow.ElectronWindow; + yield* startWaitingForShutdown(quit); yield* Ref.set(state.quitting, true); yield* logLifecycleInfo("before-quit received"); yield* requestDesktopShutdownAndWait( @@ -129,7 +171,7 @@ function handleBeforeQuit( ); }).pipe(Effect.withSpan("desktop.lifecycle.beforeQuit")), ).finally(() => { - markQuitAllowed(); + quit.phase = "allowed"; void runEffect( Effect.gen(function* () { const electronApp = yield* ElectronApp.ElectronApp; @@ -144,7 +186,12 @@ function quitFromSignal( runEffect: ( effect: Effect.Effect, ) => Promise, + quit: QuitState, ): void { + if (quit.phase === "waiting") { + void runEffect(exitIfShutdownStalled(quit, signal)); + return; + } void runEffect( Effect.gen(function* () { yield* Effect.annotateCurrentSpan({ signal }); @@ -152,8 +199,10 @@ function quitFromSignal( const state = yield* DesktopState.DesktopState; const wasQuitting = yield* Ref.getAndSet(state.quitting, true); if (wasQuitting) return; + yield* startWaitingForShutdown(quit); yield* logLifecycleInfo("process signal received", { signal }); yield* requestDesktopShutdownAndWait(); + quit.phase = "allowed"; yield* electronApp.quit; }).pipe(Effect.withSpan("desktop.lifecycle.processSignal")), ); @@ -196,8 +245,7 @@ export const make = DesktopLifecycle.of({ const environment = yield* DesktopEnvironment.DesktopEnvironment; const context = yield* Effect.context(); const runEffect = Effect.runPromiseWith(context); - let quitAllowed = false; - let updaterQuitAllowed = false; + const quit: QuitState = { phase: "running", waitingSinceNanos: 0n }; yield* electronTheme.onUpdated(() => { void runEffect( desktopWindow.syncAppearance.pipe(Effect.withSpan("desktop.lifecycle.themeUpdated")), @@ -207,7 +255,7 @@ export const make = DesktopLifecycle.of({ // Electron's updater owns the remaining quit/install/relaunch sequence. // Cancelling the following app "before-quit" event breaks that sequence, // most visibly on macOS where the native updater performs the relaunch. - updaterQuitAllowed = true; + quit.phase = "allowed"; // This event is synchronous and the updater's quit proceeds as soon as // the listener returns, so a forked destroyAll would race the quit // and windows could still be open when the process exits (visible on @@ -223,14 +271,7 @@ export const make = DesktopLifecycle.of({ ); }); yield* electronApp.on("before-quit", (event: Electron.Event) => { - handleBeforeQuit( - event, - runEffect, - () => quitAllowed || updaterQuitAllowed, - () => { - quitAllowed = true; - }, - ); + handleBeforeQuit(event, runEffect, quit); }); yield* electronApp.on("activate", () => { void runEffect( @@ -255,10 +296,10 @@ export const make = DesktopLifecycle.of({ if (environment.platform !== "win32") { yield* addScopedListener(process, "SIGINT", () => { - quitFromSignal("SIGINT", runEffect); + quitFromSignal("SIGINT", runEffect, quit); }); yield* addScopedListener(process, "SIGTERM", () => { - quitFromSignal("SIGTERM", runEffect); + quitFromSignal("SIGTERM", runEffect, quit); }); } }).pipe(Effect.withSpan("desktop.lifecycle.register")), diff --git a/apps/server/src/auth/EnvironmentAuth.test.ts b/apps/server/src/auth/EnvironmentAuth.test.ts index aa702b6bfbeeb..8558a6664c8a6 100644 --- a/apps/server/src/auth/EnvironmentAuth.test.ts +++ b/apps/server/src/auth/EnvironmentAuth.test.ts @@ -1,6 +1,15 @@ +// @effect-diagnostics-next-line nodeBuiltinImport:off -- Effect's Crypto has no generateKeyPairSync or sign. +import * as NodeCrypto from "node:crypto"; + import * as NodeServices from "@effect/platform-node/NodeServices"; import { authScopeResponse, AuthStandardClientScopes } from "@t3tools/contracts"; +import { + computeDpopAccessTokenHash, + computeDpopJwkThumbprint, + type DpopPublicJwk, +} from "@t3tools/shared/dpop"; import { expect, it } from "@effect/vitest"; +import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as Redacted from "effect/Redacted"; @@ -68,6 +77,36 @@ const requestMetadata = { ipAddress: "192.168.1.23", }; +const makeDpopKey = () => { + const { privateKey, publicKey } = NodeCrypto.generateKeyPairSync("ec", { namedCurve: "P-256" }); + const publicJwk = publicKey.export({ format: "jwk" }) as DpopPublicJwk; + let proofCount = 0; + return { + thumbprint: computeDpopJwkThumbprint(publicJwk), + sign: (input: { + readonly method: string; + readonly url: string; + readonly token: string; + readonly iat: number; + }) => { + const encode = (value: unknown) => Buffer.from(JSON.stringify(value)).toString("base64url"); + const header = encode({ typ: "dpop+jwt", alg: "ES256", jwk: publicJwk }); + const payload = encode({ + htm: input.method, + htu: input.url, + jti: `proof-${++proofCount}`, + iat: input.iat, + ath: computeDpopAccessTokenHash(input.token), + }); + const signature = NodeCrypto.sign("sha256", Buffer.from(`${header}.${payload}`), { + key: privateKey, + dsaEncoding: "ieee-p1363", + }).toString("base64url"); + return `${header}.${payload}.${signature}`; + }, + }; +}; + it.layer(NodeServices.layer)("EnvironmentAuth.layer", (it) => { it.effect("uses the reusable dev cookie without overriding a normal scoped cookie", () => Effect.gen(function* () { @@ -527,11 +566,14 @@ it.layer(NodeServices.layer)("EnvironmentAuth.layer", (it) => { const pairing = yield* serverAuth.issuePairingCredential(); const exchanged = yield* serverAuth.createBrowserSession(pairing.credential, requestMetadata); const upgrade = (headers: Record) => - serverAuth.authenticateWebSocketUpgrade({ - url: "/ws", - cookies: { [sessions.cookieName]: exchanged.sessionToken }, - headers: { host: "192.168.1.50:3773", ...headers }, - } as never); + serverAuth.authenticateWebSocketUpgrade( + { + url: "/ws", + cookies: { [sessions.cookieName]: exchanged.sessionToken }, + headers: { host: "192.168.1.50:3773", ...headers }, + } as never, + "rpc", + ); yield* upgrade({ origin: "http://192.168.1.50:3773" }); yield* upgrade({ @@ -588,16 +630,122 @@ it.layer(NodeServices.layer)("EnvironmentAuth.layer", (it) => { Effect.gen(function* () { const serverAuth = yield* EnvironmentAuth.EnvironmentAuth; const bearer = yield* serverAuth.issueSession(); - const verified = yield* serverAuth.authenticateWebSocketUpgrade({ - url: "/ws", - cookies: {}, - headers: { - host: "192.168.1.50:3773", - origin: "http://192.168.1.50:8080", - authorization: `Bearer ${bearer.token}`, - }, - } as never); + const verified = yield* serverAuth.authenticateWebSocketUpgrade( + { + url: "/ws", + cookies: {}, + headers: { + host: "192.168.1.50:3773", + origin: "http://192.168.1.50:8080", + authorization: `Bearer ${bearer.token}`, + }, + } as never, + "rpc", + ); expect(verified.sessionId).toBe(bearer.sessionId); }).pipe(Effect.provide(layerEnvironmentAuth({ mode: "web", host: "192.168.1.50" }))), ); + + it.effect("binds a DPoP session's RPC socket ticket to its proof key", () => + Effect.gen(function* () { + const serverAuth = yield* EnvironmentAuth.EnvironmentAuth; + const sessions = yield* SessionStore.SessionStore; + const key = makeDpopKey(); + const other = makeDpopKey(); + const exchanged = yield* serverAuth.exchangeBootstrapCredentialForAccessToken( + "desktop-bootstrap-token", + ["orchestration:read"], + requestMetadata, + { proofKeyThumbprint: key.thumbprint }, + ); + const session = yield* sessions.verify(exchanged.access_token); + const host = "192.168.1.50:3773"; + const socketUrl = `http://${host}/ws`; + const iat = Math.floor((yield* DateTime.now).epochMilliseconds / 1_000); + const upgrade = (ticket: string, use: "rpc" | "media", proof?: string) => + serverAuth.authenticateWebSocketUpgrade( + { + method: "GET", + url: `/ws?orchestrationProtocol=2&wsTicket=${ticket}${proof ? `&dpop=${proof}` : ""}`, + cookies: {}, + headers: { host }, + } as never, + use, + ); + + const rpc = yield* serverAuth.issueWebSocketTicket(session, { use: "rpc" }); + const bare = yield* Effect.flip(upgrade(rpc.ticket, "rpc")); + expect(bare._tag).toBe("ServerAuthInvalidCredentialError"); + const signed = yield* upgrade( + rpc.ticket, + "rpc", + key.sign({ method: "GET", url: socketUrl, token: rpc.ticket, iat }), + ); + expect(signed.sessionId).toBe(session.sessionId); + + const wrongKey = yield* Effect.flip( + upgrade( + rpc.ticket, + "rpc", + other.sign({ method: "GET", url: socketUrl, token: rpc.ticket, iat }), + ), + ); + expect(wrongKey._tag).toBe("ServerAuthInvalidCredentialError"); + const onMedia = yield* Effect.flip( + upgrade( + rpc.ticket, + "media", + key.sign({ method: "GET", url: socketUrl, token: rpc.ticket, iat }), + ), + ); + expect(onMedia._tag).toBe("ServerAuthInvalidCredentialError"); + + const media = yield* serverAuth.issueWebSocketTicket(session, { use: "media" }); + expect((yield* upgrade(media.ticket, "media")).sessionId).toBe(session.sessionId); + const mediaOnRpc = yield* Effect.flip(upgrade(media.ticket, "rpc")); + expect(mediaOnRpc._tag).toBe("ServerAuthInvalidCredentialError"); + + // Clients that predate ticket binding still connect with the unbound ticket they mint. + const legacy = yield* serverAuth.issueWebSocketTicket(session); + expect((yield* upgrade(legacy.ticket, "rpc")).sessionId).toBe(session.sessionId); + }).pipe( + Effect.provide(layerEnvironmentAuth({ desktopBootstrapToken: "desktop-bootstrap-token" })), + ), + ); + + it.effect("rejects a replayed proof on a bound RPC socket ticket", () => + Effect.gen(function* () { + const serverAuth = yield* EnvironmentAuth.EnvironmentAuth; + const sessions = yield* SessionStore.SessionStore; + const key = makeDpopKey(); + const exchanged = yield* serverAuth.exchangeBootstrapCredentialForAccessToken( + "desktop-bootstrap-token", + ["orchestration:read"], + requestMetadata, + { proofKeyThumbprint: key.thumbprint }, + ); + const session = yield* sessions.verify(exchanged.access_token); + const rpc = yield* serverAuth.issueWebSocketTicket(session, { use: "rpc" }); + const request = { + method: "GET", + url: `/ws?wsTicket=${rpc.ticket}&dpop=${key.sign({ + method: "GET", + url: "http://192.168.1.50:3773/ws", + token: rpc.ticket, + iat: Math.floor((yield* DateTime.now).epochMilliseconds / 1_000), + })}`, + cookies: {}, + headers: { host: "192.168.1.50:3773" }, + } as never; + + yield* serverAuth.authenticateWebSocketUpgrade(request, "rpc"); + const replayed = yield* Effect.flip(serverAuth.authenticateWebSocketUpgrade(request, "rpc")); + expect(replayed._tag).toBe("ServerAuthInvalidCredentialError"); + if (replayed._tag === "ServerAuthInvalidCredentialError") { + expect(replayed.dpopFailureReason).toBe("replay"); + } + }).pipe( + Effect.provide(layerEnvironmentAuth({ desktopBootstrapToken: "desktop-bootstrap-token" })), + ), + ); }); diff --git a/apps/server/src/auth/EnvironmentAuth.ts b/apps/server/src/auth/EnvironmentAuth.ts index 1a6d3206833ab..b53b4c656b343 100644 --- a/apps/server/src/auth/EnvironmentAuth.ts +++ b/apps/server/src/auth/EnvironmentAuth.ts @@ -21,6 +21,7 @@ import { type ServerAuthDescriptor, type ServerAuthSessionMethod, type AuthWebSocketTicketResult, + type AuthWebSocketTicketUse, DpopFailureReason, type DpopFailureReason as DpopFailureReasonType, } from "@t3tools/contracts"; @@ -543,9 +544,11 @@ export class EnvironmentAuth extends Context.Service< ) => Effect.Effect; readonly authenticateWebSocketUpgrade: ( request: HttpServerRequest.HttpServerRequest, + use: AuthWebSocketTicketUse, ) => Effect.Effect; readonly issueWebSocketTicket: ( - session: Pick, + session: Pick, + input?: { readonly use?: AuthWebSocketTicketUse }, ) => Effect.Effect; readonly issueStartupPairingUrl: ( baseUrl: string, @@ -591,6 +594,7 @@ type BootstrapExchangeResult = { const AUTHORIZATION_PREFIX = "Bearer "; const DPOP_AUTHORIZATION_PREFIX = "DPoP "; const WEBSOCKET_TICKET_QUERY_PARAM = "wsTicket"; +const WEBSOCKET_PROOF_QUERY_PARAM = "dpop"; const bySessionPriority = (left: AuthClientSession, right: AuthClientSession) => { const leftCanManage = left.scopes.includes(AuthAccessWriteScope); @@ -1220,18 +1224,31 @@ export const make = Effect.gen(function* () { Effect.withSpan("EnvironmentAuth.issueStartupPairingUrl"), ); - const issueWebSocketTicket: EnvironmentAuth["Service"]["issueWebSocketTicket"] = (session) => - sessions.issueWebSocketToken(session.sessionId).pipe( - Effect.mapError((cause) => new ServerAuthWebSocketTokenIssueError({ cause })), - Effect.map( - (issued) => - ({ - ticket: issued.token, - expiresAt: DateTime.toUtc(issued.expiresAt), - }) satisfies AuthWebSocketTicketResult, - ), - Effect.withSpan("EnvironmentAuth.issueWebSocketTicket"), - ); + // Only the RPC socket's ticket is bound to the client's proof key. The media + // routes reuse one ticket across , EventSource and several sockets, and + // each of those would need its own single-use proof. + const issueWebSocketTicket: EnvironmentAuth["Service"]["issueWebSocketTicket"] = ( + session, + input, + ) => + sessions + .issueWebSocketToken(session.sessionId, { + ...(input?.use ? { audience: input.use } : {}), + ...(input?.use === "rpc" && session.proofKeyThumbprint + ? { proofKeyThumbprint: session.proofKeyThumbprint } + : {}), + }) + .pipe( + Effect.mapError((cause) => new ServerAuthWebSocketTokenIssueError({ cause })), + Effect.map( + (issued) => + ({ + ticket: issued.token, + expiresAt: DateTime.toUtc(issued.expiresAt), + }) satisfies AuthWebSocketTicketResult, + ), + Effect.withSpan("EnvironmentAuth.issueWebSocketTicket"), + ); const authenticateHttpRequest: EnvironmentAuth["Service"]["authenticateHttpRequest"] = ( request, @@ -1239,22 +1256,55 @@ export const make = Effect.gen(function* () { authenticateRequest(request).pipe(Effect.withSpan("EnvironmentAuth.authenticateHttpRequest")); const authenticateWebSocketUpgrade: EnvironmentAuth["Service"]["authenticateWebSocketUpgrade"] = - Effect.fn("EnvironmentAuth.authenticateWebSocketUpgrade")(function* (request) { + Effect.fn("EnvironmentAuth.authenticateWebSocketUpgrade")(function* (request, use) { const requestUrl = HttpServerRequest.toURL(request); if (Option.isSome(requestUrl)) { const websocketTicket = requestUrl.value.searchParams.get(WEBSOCKET_TICKET_QUERY_PARAM); if (websocketTicket && websocketTicket.trim().length > 0) { - return yield* sessions.verifyWebSocketToken(websocketTicket).pipe( - mapSessionVerificationErrors, - Effect.flatMap(rejectMcpClientAudience), - Effect.map((session) => ({ - sessionId: session.sessionId, - subject: session.subject, - method: session.method, - scopes: session.scopes, - ...(session.expiresAt ? { expiresAt: session.expiresAt } : {}), - })), - ); + const session = yield* sessions + .verifyWebSocketToken(websocketTicket) + .pipe(mapSessionVerificationErrors, Effect.flatMap(rejectMcpClientAudience)); + if (session.audience !== undefined && session.audience !== use) { + return yield* new ServerAuthInvalidCredentialError({ + diagnostic: `WebSocket ticket was issued for the ${session.audience} route.`, + }); + } + if (session.proofKeyThumbprint) { + // A WebSocket cannot set headers, so the proof rides in the query. + // The ticket is the token the proof hashes, and the proof's jti is + // consumed like any other, so a captured socket URL is dead once the + // client has used it. Beating the client to first use still needs an + // active interceptor, which only encrypted transport rules out. + const proof = + request.headers.dpop ?? + requestUrl.value.searchParams.get(WEBSOCKET_PROOF_QUERY_PARAM) ?? + undefined; + if (!proof) { + return yield* new ServerAuthInvalidCredentialError({ + diagnostic: "DPoP-bound WebSocket ticket requires a DPoP proof.", + dpopFailureReason: "invalid_proof", + }); + } + yield* verifyRequestDpopProof({ + request, + proof, + expectedThumbprint: session.proofKeyThumbprint, + expectedAccessToken: websocketTicket, + }).pipe( + Effect.provideService(ServerSecretStore.ServerSecretStore, secretStore), + Effect.provideService(Crypto.Crypto, crypto), + ); + } + return { + sessionId: session.sessionId, + subject: session.subject, + method: session.method, + scopes: session.scopes, + ...(session.proofKeyThumbprint + ? { proofKeyThumbprint: session.proofKeyThumbprint } + : {}), + ...(session.expiresAt ? { expiresAt: session.expiresAt } : {}), + }; } } diff --git a/apps/server/src/auth/SessionStore.ts b/apps/server/src/auth/SessionStore.ts index 1f4ad752d2bdd..d192698012965 100644 --- a/apps/server/src/auth/SessionStore.ts +++ b/apps/server/src/auth/SessionStore.ts @@ -5,6 +5,7 @@ import { type AuthClientMetadata, type AuthClientSession, type AuthEnvironmentScope, + AuthWebSocketTicketUse, type ClientSurface, RuntimeMode, type ServerAuthSessionMethod, @@ -61,6 +62,10 @@ export interface VerifiedSession { readonly runtimeModeCeiling?: RuntimeMode; } +export interface VerifiedWebSocketTicket extends VerifiedSession { + readonly audience?: AuthWebSocketTicketUse; +} + export type SessionCredentialChange = | { readonly type: "clientUpserted"; @@ -384,6 +389,8 @@ export class SessionStore extends Context.Service< sessionId: AuthSessionId, input?: { readonly ttl?: Duration.Duration; + readonly audience?: AuthWebSocketTicketUse; + readonly proofKeyThumbprint?: string; }, ) => Effect.Effect< { @@ -394,7 +401,7 @@ export class SessionStore extends Context.Service< >; readonly verifyWebSocketToken: ( token: string, - ) => Effect.Effect; + ) => Effect.Effect; readonly listActive: () => Effect.Effect< ReadonlyArray, SessionCredentialInternalError @@ -443,6 +450,8 @@ const WebSocketClaims = Schema.Struct({ v: Schema.Literal(1), kind: Schema.Literal("websocket"), sid: AuthSessionId, + aud: Schema.optionalKey(AuthWebSocketTicketUse), + jkt: Schema.optionalKey(Schema.String), iat: Schema.Number, exp: Schema.Number, }); @@ -867,6 +876,8 @@ const make = Effect.gen(function* () { v: 1, kind: "websocket", sid: sessionId, + ...(input?.audience ? { aud: input.audience } : {}), + ...(input?.proofKeyThumbprint ? { jkt: input.proofKeyThumbprint } : {}), iat: issuedAt.epochMilliseconds, exp: expiresAt.epochMilliseconds, }; @@ -959,7 +970,9 @@ const make = Effect.gen(function* () { expiresAt: row.value.expiresAt, subject: row.value.subject, scopes: row.value.scopes, - } satisfies VerifiedSession; + ...(claims.aud ? { audience: claims.aud } : {}), + ...(claims.jkt ? { proofKeyThumbprint: claims.jkt } : {}), + } satisfies VerifiedWebSocketTicket; }); const listActive: SessionStore["Service"]["listActive"] = Effect.fn("SessionStore.listActive")( diff --git a/apps/server/src/auth/dpop.ts b/apps/server/src/auth/dpop.ts index 3e23879f1e063..7b780852bc3e4 100644 --- a/apps/server/src/auth/dpop.ts +++ b/apps/server/src/auth/dpop.ts @@ -54,10 +54,12 @@ export const mapDpopReplayStoreError = ( export const verifyRequestDpopProof = Effect.fn("dpop.verifyRequestDpopProof")(function* (input: { readonly request: HttpServerRequest.HttpServerRequest; + /** Overrides the `DPoP` header for requests that cannot carry one, such as a WebSocket upgrade. */ + readonly proof?: string; readonly expectedThumbprint?: string; readonly expectedAccessToken?: string; }) { - const proof = input.request.headers.dpop; + const proof = input.proof ?? input.request.headers.dpop; const url = HttpServerRequest.toURL(input.request); if (Option.isNone(url)) { return yield* new ServerAuthInvalidCredentialError({ diff --git a/apps/server/src/auth/http.ts b/apps/server/src/auth/http.ts index ae052abbf01eb..143e0957a4822 100644 --- a/apps/server/src/auth/http.ts +++ b/apps/server/src/auth/http.ts @@ -166,7 +166,7 @@ export const authenticateMediaRequest = Effect.fnUntraced(function* (requiredScope: AuthEnvironmentScope) { const request = yield* HttpServerRequest.HttpServerRequest; const serverAuth = yield* EnvironmentAuth.EnvironmentAuth; - const session = yield* serverAuth.authenticateWebSocketUpgrade(request).pipe( + const session = yield* serverAuth.authenticateWebSocketUpgrade(request, "media").pipe( Effect.catch((error) => Effect.gen(function* () { if (EnvironmentAuth.isServerAuthCredentialError(error)) { @@ -426,6 +426,20 @@ export const layer = HttpApiBuilder.group( ), ), ) + .handle( + "webSocketTicketFor", + Effect.fn("environment.auth.webSocketTicketFor")( + function* (args) { + yield* annotateEnvironmentRequest(args.endpoint.name); + const session = yield* EnvironmentAuthenticatedPrincipal; + yield* appendCredentialResponseHeaders; + return yield* serverAuth.issueWebSocketTicket(session, { use: args.params.use }); + }, + Effect.catchIf(EnvironmentAuth.isServerAuthInternalError, (error) => + failEnvironmentInternal("websocket_ticket_issuance_failed", error), + ), + ), + ) .handle( "pairingCredential", Effect.fn("environment.auth.pairingCredential")( diff --git a/apps/server/src/environment/ServerEnvironment.ts b/apps/server/src/environment/ServerEnvironment.ts index 9d8bdf9fc3cd0..d330b2a7cb53e 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* () { : {}), ...(desktopAppUpdate ? { desktopAppUpdate: true } : {}), serverBrowser: true, + vcsLocalOnlyRefresh: true, }, }; diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts index 18e192e28dc50..9fed0cd0f0c9c 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts @@ -4199,6 +4199,138 @@ describe("ClaudeAdapterV2 background wake turns", () => { ), ); + it.effect("an interrupted turn after a rollback stops resuming at the rolled-back head", () => + Effect.gen(function* () { + const harness = yield* makeWakeHarnessWithOptions({ + close: (sdkMessages) => Queue.shutdown(sdkMessages), + freshQueueOnReopen: true, + }); + const now = yield* DateTime.now; + const pinnedHead = "00000000-0000-4000-8000-000000025510"; + const pinnedThread = { + ...harness.providerThread, + nativeConversationHeadRef: { + driver: ClaudeAdapterV2.CLAUDE_PROVIDER, + nativeId: pinnedHead, + strength: "weak" as const, + }, + }; + const interruptedInput = makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: pinnedThread, + now, + attemptId: RunAttemptId.make("attempt-claude-interrupt-after-rollback"), + text: "The secret word is apricot-25510. Then run sleep 60.", + attachments: [], + providerTurnOrdinal: 2, + nativeThreadHasTurns: true, + }); + yield* harness.runtime.startTurn(interruptedInput); + assert.equal(harness.getOpenedOptions()?.resumeSessionAt, pinnedHead); + yield* harness.offerAndWait( + makeAssistantTextFrame({ + uuid: "00000000-0000-4000-8000-000000025511", + text: "Noted apricot-25510. Running sleep.", + }), + ); + yield* harness.runtime.interruptTurn({ + providerThread: pinnedThread, + providerTurnId: (yield* IdAllocator.IdAllocatorV2).derive.providerTurn({ + driver: ClaudeAdapterV2.CLAUDE_PROVIDER, + nativeTurnId: `turn:${interruptedInput.attemptId}`, + }), + }); + assert.equal((yield* Queue.take(harness.terminalReceipts)).status, "interrupted"); + const afterInterrupt = harness.events.findLast( + (event) => event.type === "provider_thread.updated", + )?.providerThread; + assert.isDefined(afterInterrupt); + assert.isNull(afterInterrupt.nativeConversationHeadRef); + + yield* harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: afterInterrupt, + now, + attemptId: RunAttemptId.make("attempt-claude-after-interrupt"), + text: "What is the secret word?", + attachments: [], + providerTurnOrdinal: 3, + nativeThreadHasTurns: true, + }), + ); + assert.equal(harness.processQueues.length, 2); + assert.isUndefined(harness.getOpenedOptions()?.resumeSessionAt); + }).pipe( + Effect.provide( + Layer.mergeAll(IdAllocator.layer, McpProviderSessions.layer, NodeServices.layer), + ), + ), + ); + + it.effect("an interrupted turn with no assistant output keeps the rolled-back head", () => + Effect.gen(function* () { + const harness = yield* makeWakeHarnessWithOptions({ + close: (sdkMessages) => Queue.shutdown(sdkMessages), + freshQueueOnReopen: true, + }); + const now = yield* DateTime.now; + const pinnedHead = "00000000-0000-4000-8000-000000025510"; + const pinnedThread = { + ...harness.providerThread, + nativeConversationHeadRef: { + driver: ClaudeAdapterV2.CLAUDE_PROVIDER, + nativeId: pinnedHead, + strength: "weak" as const, + }, + }; + const interruptedInput = makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: pinnedThread, + now, + attemptId: RunAttemptId.make("attempt-claude-silent-interrupt-after-rollback"), + text: "The secret word is apricot-25510. Then run sleep 60.", + attachments: [], + providerTurnOrdinal: 2, + nativeThreadHasTurns: true, + }); + yield* harness.runtime.startTurn(interruptedInput); + assert.equal(harness.getOpenedOptions()?.resumeSessionAt, pinnedHead); + yield* harness.runtime.interruptTurn({ + providerThread: pinnedThread, + providerTurnId: (yield* IdAllocator.IdAllocatorV2).derive.providerTurn({ + driver: ClaudeAdapterV2.CLAUDE_PROVIDER, + nativeTurnId: `turn:${interruptedInput.attemptId}`, + }), + }); + assert.equal((yield* Queue.take(harness.terminalReceipts)).status, "interrupted"); + const afterInterrupt = harness.events.findLast( + (event) => event.type === "provider_thread.updated", + )?.providerThread; + assert.isDefined(afterInterrupt); + assert.equal(afterInterrupt.nativeConversationHeadRef?.nativeId, pinnedHead); + + yield* harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: afterInterrupt, + now, + attemptId: RunAttemptId.make("attempt-claude-after-silent-interrupt"), + text: "What is the secret word?", + attachments: [], + providerTurnOrdinal: 3, + nativeThreadHasTurns: true, + }), + ); + assert.equal(harness.processQueues.length, 2); + assert.equal(harness.getOpenedOptions()?.resumeSessionAt, pinnedHead); + }).pipe( + Effect.provide( + Layer.mergeAll(IdAllocator.layer, McpProviderSessions.layer, NodeServices.layer), + ), + ), + ); + it.effect("never records a subagent uuid for a turn with no main-thread assistant message", () => Effect.gen(function* () { // The continuation turn reopens the query, and rollback shut the first queue down. diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts index 3aef74ea86c06..95c8984c643bb 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts @@ -6998,8 +6998,9 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( yield* Ref.get(awaitedBackgroundTasksByNativeThread), nativeThreadId, ); + // A stopped turn that wrote to the transcript moved past the pinned head. const clearConversationHead = - input.status === "completed" && + (input.status === "completed" || input.context.nativeMessageCursor !== null) && input.context.input.providerThread.nativeConversationHeadRef !== null; const goal = nativeThreadId === null ? undefined : goalsByNativeThread.get(nativeThreadId); diff --git a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts index ee87e27c68e5f..a4b041d6f127f 100644 --- a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts +++ b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts @@ -979,6 +979,144 @@ it.layer(layerTest)("delegated completion delivery repairs", (it) => { ); }), ); + + it.effect.each([ + { command: "acknowledge", before: "running", after: "failed", capture: false }, + { command: "acknowledge", before: "waiting", after: "completed", capture: true }, + { command: "accept", before: "waiting", after: "completed", capture: true }, + ] as const)( + "keeps a $after run that settled between $command's read and commit", + ({ command, before, after, capture }) => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const sink = yield* EventSink.EventSinkV2; + const now = yield* DateTime.now; + const race = `${command}-settle-race-${after}`; + const threadId = ThreadId.make(`thread:${race}`); + const runId = RunId.make(`run:${race}`); + const taskId = NodeId.make(`node:${race}-task`); + const checkpointId = CheckpointId.make(`checkpoint:${race}`); + const messageId = MessageId.make(`message:delegated-delivery:${threadId}`); + yield* seedParentWithTerminalTask({ + threadId, + projectId: ProjectId.make(`project:${race}`), + runId, + rootNodeId: NodeId.make(`node:${race}-root`), + taskId, + deliveryState: "claimed", + deliveryTaskIds: [taskId], + now, + }); + const parent = (yield* orchestrator.getThreadProjection(threadId)).runs[0]!; + yield* sink.write({ + events: [ + { + id: EventId.make(`event:${race}-before`), + type: "run.updated", + threadId, + runId, + occurredAt: now, + payload: { ...parent, status: before }, + }, + { + id: EventId.make(`event:${race}-message`), + type: "message.updated", + threadId, + runId, + occurredAt: now, + payload: { + id: messageId, + threadId, + runId, + nodeId: parent.rootNodeId, + role: "user", + text: "Background task finished", + attachments: [], + streaming: false, + createdBy: "agent", + creationSource: "server", + createdAt: now, + updatedAt: now, + delegatedCompletion: { parentRunId: runId, generation: 1, taskIds: [taskId] }, + }, + }, + ], + }); + const { delegatedCompletion: _cohort, ...settledRun } = parent; + const commit = sink.commitCommand; + let settled = false; + yield* Effect.acquireUseRelease( + Effect.sync(() => + vi.spyOn(sink, "commitCommand").mockImplementation((input) => + Effect.gen(function* () { + if ( + !settled && + input.commandType === + (command === "accept" + ? "notification.delivery.accept" + : "delegated_task.completion-delivery.acknowledge") + ) { + settled = true; + // The provider turn ends (or its checkpoint lands) after the command read the run. + yield* commit({ + commandId: CommandId.make(`command:${race}`), + threadId, + commandType: capture ? "checkpoint.capture" : "provider.event", + acceptedAt: now, + effects: [], + events: [ + { + id: EventId.make(`event:${race}-settled`), + type: "run.updated", + threadId, + runId, + occurredAt: now, + payload: { + ...settledRun, + status: after, + completedAt: now, + checkpointId: capture ? checkpointId : null, + }, + }, + ], + }); + } + return yield* commit(input); + }), + ), + ), + () => + orchestrator.dispatch( + command === "accept" + ? { + type: "notification.delivery.accept", + commandId: CommandId.make(`command:${race}:accept`), + threadId, + messageId, + } + : { + type: "delegated_task.completion-delivery.acknowledge", + commandId: CommandId.make(`command:${race}:task-status`), + parentThreadId: threadId, + taskId, + observedByRunId: runId, + }, + ), + (spy) => Effect.sync(() => spy.mockRestore()), + ); + assert.isTrue(settled); + const finished = yield* orchestrator.getThreadProjection(threadId); + assert.equal( + finished.subagents.find((row) => row.id === taskId)?.completionDelivery?.state, + command === "accept" ? "delivered" : "acknowledged", + ); + const run = finished.runs.find((candidate) => candidate.id === runId)!; + assert.deepEqual( + { status: run.status, completedAt: run.completedAt, checkpointId: run.checkpointId }, + { status: after, completedAt: now, checkpointId: capture ? checkpointId : null }, + ); + }), + ); }); const reconcileCommandId = (name: string) => CommandId.make(`command:runtime-reconcile:${name}`); diff --git a/apps/server/src/orchestration-v2/EventSink.ts b/apps/server/src/orchestration-v2/EventSink.ts index 3d2bc8e4b45a8..102b119cc7712 100644 --- a/apps/server/src/orchestration-v2/EventSink.ts +++ b/apps/server/src/orchestration-v2/EventSink.ts @@ -156,6 +156,8 @@ export interface EventSinkV2Shape { readonly effectTypes: ReadonlyArray; readonly reason: string; }; + /** Like writeWithEffects' updateDelegatedCompletionOnly, for this run's run.updated only. */ + readonly updateDelegatedCompletionOnlyRunId?: RunId; }) => Effect.Effect< { readonly receipt: CommandReceiptStore.CommandReceiptV2; @@ -393,6 +395,28 @@ const layerBase: Layer.Layer< return yield* runOutcomes.prepare(storedEvents); }); + // The enclosing write or commit span already traces this per-event projection merge. + const updateDelegatedCompletionOnly = Effect.fnUntraced(function* ( + events: ReadonlyArray, + includesRun: (runId: RunId) => boolean, + ) { + return yield* Effect.forEach(events, (event) => + Effect.gen(function* () { + if (event.type !== "run.updated" || !includesRun(event.payload.id)) return event; + const { runs } = yield* projectionStore.getThreadRecords(event.threadId, ["runs"], { + runIds: [event.payload.id], + }); + const current = runs[0]; + return current === undefined + ? event + : { + ...event, + payload: { ...current, delegatedCompletion: event.payload.delegatedCompletion }, + }; + }), + ); + }); + const writeEffect = Effect.fn("orchestrationV2.EventSink.write")(function* ( input: Parameters[0], ) { @@ -421,26 +445,7 @@ const layerBase: Layer.Layer< // Checkpoint capture can finish after a cohort writer reads its parent run. // Merge in this transaction so the event log and projection keep that completion. const events = input.updateDelegatedCompletionOnly - ? yield* Effect.forEach(input.events, (event) => - Effect.gen(function* () { - if (event.type !== "run.updated") return event; - const { runs } = yield* projectionStore.getThreadRecords( - event.threadId, - ["runs"], - { runIds: [event.payload.id] }, - ); - const current = runs[0]; - return current === undefined - ? event - : { - ...event, - payload: { - ...current, - delegatedCompletion: event.payload.delegatedCompletion, - }, - }; - }), - ) + ? yield* updateDelegatedCompletionOnly(input.events, () => true) : input.events; const normalized = yield* normalizeEvents( input.guardPendingUserInputCancellations === true @@ -656,7 +661,12 @@ const layerBase: Layer.Layer< return { ...existing, committed: false as const, cancelledEffectIds: [] }; } - const normalized = yield* normalizeEvents(input.events); + const runId = input.updateDelegatedCompletionOnlyRunId; + const normalized = yield* normalizeEvents( + runId === undefined + ? input.events + : yield* updateDelegatedCompletionOnly(input.events, (id) => id === runId), + ); const storedEvents = yield* eventStore.append({ commandId: input.commandId, events: normalized, diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 466bc231fd0ac..0bdf2482e24fc 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -2504,7 +2504,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio run: deliveryRun, now, }); - return; + return parentRun.id; } const message = completionDeliveryMessage(projection, delivery); if (message !== undefined) { @@ -2528,6 +2528,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio }); } } + return parentRun.id; }); const disposeDelegatedCompletionCohort = Effect.fn( @@ -11495,6 +11496,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio }, }, }); + return parentRun.id; }); const dispatchUnsupported = (command: OrchestrationV2ServerCommand) => @@ -11515,6 +11517,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio readonly effectTypes: ReadonlyArray; readonly reason: string; }; + readonly updateDelegatedCompletionOnlyRunId?: RunId; }, OrchestratorV2Error > { @@ -11532,6 +11535,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio readonly reason: string; } | undefined; + let updateDelegatedCompletionOnlyRunId: RunId | undefined; const routingThread = yield* assertCommandModelRouting(command); if (routingThread !== undefined) yield* emit( @@ -11665,7 +11669,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio break; } case "notification.delivery.accept": - yield* dispatchNotificationAccepted(command, events); + updateDelegatedCompletionOnlyRunId = yield* dispatchNotificationAccepted(command, events); break; case "prepared-run.release": yield* dispatchPreparedRunRelease(command, events, effects); @@ -11834,7 +11838,8 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio break; case "delegated_task.completion-delivery.acknowledge": case "delegated_task.completion-delivery.dispose": - yield* dispatchDelegatedTaskCompletionDeliveryResolution(command, events); + updateDelegatedCompletionOnlyRunId = + yield* dispatchDelegatedTaskCompletionDeliveryResolution(command, events); break; case "thread.created.record": yield* dispatchCreatedThreadRecord(command, events); @@ -11849,6 +11854,9 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio events: yield* Ref.get(events), effects: yield* Ref.get(effects), ...(cancelUnsettledEffects === undefined ? {} : { cancelUnsettledEffects }), + ...(updateDelegatedCompletionOnlyRunId === undefined + ? {} + : { updateDelegatedCompletionOnlyRunId }), }; }); @@ -12021,6 +12029,9 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ...(plan.cancelUnsettledEffects === undefined ? {} : { cancelUnsettledEffects: plan.cancelUnsettledEffects }), + ...(plan.updateDelegatedCompletionOnlyRunId === undefined + ? {} + : { updateDelegatedCompletionOnlyRunId: plan.updateDelegatedCompletionOnlyRunId }), }) .pipe( Effect.mapError( diff --git a/apps/server/src/orchestration-v2/ProviderEventIngestor.test.ts b/apps/server/src/orchestration-v2/ProviderEventIngestor.test.ts index 849cedded9292..f6f1f1fc2f86a 100644 --- a/apps/server/src/orchestration-v2/ProviderEventIngestor.test.ts +++ b/apps/server/src/orchestration-v2/ProviderEventIngestor.test.ts @@ -17,20 +17,23 @@ import { type OrchestrationV2TurnItem, ProviderDriverKind, ProviderInstanceId, - type ProviderThreadId, + ProviderThreadId, ProviderTurnId, RunAttemptId, RunId, RuntimeRequestId, TurnItemId, } from "@t3tools/contracts"; +import * as Cause from "effect/Cause"; import * as DateTime from "effect/DateTime"; import * as Deferred from "effect/Deferred"; import * as Fiber from "effect/Fiber"; import * as Effect from "effect/Effect"; +import * as Exit from "effect/Exit"; import * as FileSystem from "effect/FileSystem"; import * as Layer from "effect/Layer"; import * as SqlClient from "effect/sql/SqlClient"; +import * as SqlError from "effect/sql/SqlError"; import * as Stream from "effect/Stream"; import { toolOutputImages } from "@t3tools/shared/toolOutput"; @@ -65,25 +68,63 @@ const layerGit = GitVcsDriver.layer.pipe( Layer.provideMerge(NodeServices.layer), ); -const layerTest = Layer.mergeAll( - layerTestDatabase, - layerTestStores, - layerTestEventSink, - IdAllocator.layer, - ThreadCommandExecutor.layer, - layerGit, - ProviderEventIngestor.layer.pipe( - Layer.provide( - Layer.mergeAll( - layerTestStores, - layerTestEventSink, - IdAllocator.layer, - ThreadCommandExecutor.layer, - layerGit, +const makeLayerTest = (ingestorEventSink: Layer.Layer) => + Layer.mergeAll( + layerTestDatabase, + layerTestStores, + layerTestEventSink, + IdAllocator.layer, + ThreadCommandExecutor.layer, + layerGit, + ProviderEventIngestor.layer.pipe( + Layer.provide( + Layer.mergeAll( + layerTestStores, + ingestorEventSink, + IdAllocator.layer, + ThreadCommandExecutor.layer, + layerGit, + ), ), ), - ), -); + ); +const layerTest = makeLayerTest(layerTestEventSink); + +// EventSink reports a lock that outlasted busy_timeout at BEGIN IMMEDIATE this way. +const lockTimeout = new EventSink.EventSinkWriteError({ + eventCount: 1, + cause: new SqlError.SqlError({ reason: new SqlError.LockTimeoutError({ cause: null }) }), +}); +const constraintFailure = new EventSink.EventSinkWriteError({ + eventCount: 1, + cause: new SqlError.SqlError({ reason: new SqlError.ConstraintError({ cause: null }) }), +}); + +/** The ingestor's next sink writes of `events` fail with `errors`, in order; the rest go through. */ +const layerFailingWrites = ( + errors: ReadonlyArray, + matches: (events: ReadonlyArray) => boolean = () => true, +) => + makeLayerTest( + Layer.effect( + EventSink.EventSinkV2, + Effect.gen(function* () { + const eventSink = yield* EventSink.EventSinkV2; + const pending = [...errors]; + const failNext = ( + events: ReadonlyArray, + write: Effect.Effect, + ) => Effect.suspend(() => (matches(events) ? (pending.shift() ?? write) : write)); + return EventSink.EventSinkV2.of({ + ...eventSink, + write: (input) => failNext(input.events, eventSink.write(input)), + writeIfRunCurrent: (input) => failNext(input.events, eventSink.writeIfRunCurrent(input)), + writeIfProviderThreadOwner: (input) => + failNext(input.events, eventSink.writeIfProviderThreadOwner(input)), + }); + }), + ).pipe(Layer.provide(layerTestEventSink)), + ); const makeSharedWorktree = Effect.gen(function* () { const fs = yield* FileSystem.FileSystem; @@ -1995,7 +2036,7 @@ layer("ProviderEventIngestorV2", (it) => { }), ); - it.effect("moves a native subagent's thread to the model its provider reports later", () => + it.effect("moves a native subagent's thread to its later model through a database lock", () => Effect.gen(function* () { const now = yield* DateTime.now; const eventSink = yield* EventSink.EventSinkV2; @@ -2114,6 +2155,103 @@ layer("ProviderEventIngestorV2", (it) => { (yield* projectionStore.getThread(childThreadId)).modelSelection, reportedSelection, ); - }), + }).pipe( + // Fresh: the shared test layer has already built the ingestor this replaces. + Effect.provide( + Layer.fresh( + layerFailingWrites([lockTimeout, lockTimeout], (events) => + events.some((event) => event.type === "thread.model-selection-updated"), + ), + ), + ), + ), ); }); + +const ingestCompletedProviderTurn = ( + guard: Pick< + Parameters[0], + "writeIfRunCurrent" | "writeIfProviderThreadOwner" + >, +) => + Effect.gen(function* () { + const now = yield* DateTime.now; + const eventSink = yield* EventSink.EventSinkV2; + const projectionStore = yield* ProjectionStore.ProjectionStoreV2; + const ingestor = yield* ProviderEventIngestor.ProviderEventIngestorV2; + const idAllocator = yield* IdAllocator.IdAllocatorV2; + const threadEvent = yield* threadCreatedEvent(now); + yield* eventSink.write({ events: [threadEvent] }); + const providerTurnId = ProviderTurnId.make("provider-turn:locked"); + const exit = yield* ingestor + .ingestNormalized({ + providerSessionId: yield* idAllocator.allocate.providerSession({ + providerInstanceId: modelSelection.instanceId, + threadId: threadEvent.threadId, + }), + providerInstanceId: modelSelection.instanceId, + threadId: threadEvent.threadId, + event: { + type: "provider_turn.updated", + driver: CODEX_DRIVER, + providerTurn: { + id: providerTurnId, + providerThreadId: ProviderThreadId.make("provider-thread:locked"), + nodeId: NodeId.make("node:locked-turn"), + runAttemptId: null, + nativeTurnRef: null, + ordinal: 1, + status: "completed", + startedAt: now, + completedAt: now, + }, + }, + ...guard, + }) + .pipe(Effect.exit); + const projection = yield* projectionStore.getThreadProjection(threadEvent.threadId); + return { exit, providerTurnId, persisted: projection.providerTurns.map((turn) => turn.id) }; + }); + +// Neither guard matches a stored run, so a guarded write correctly commits nothing. +const runId = RunId.make("run:locked"); +const activeAttemptId = RunAttemptId.make("attempt:locked"); +it.effect.each([ + { route: "write", guard: {} }, + { + route: "writeIfRunCurrent", + guard: { writeIfRunCurrent: { runId, activeAttemptId, expectedStatus: "running" } }, + }, + { + route: "writeIfProviderThreadOwner", + guard: { + writeIfProviderThreadOwner: { + providerThreadId: ProviderThreadId.make("provider-thread:locked"), + runId, + activeAttemptId, + expectedLastRunOrdinal: 1, + }, + }, + }, +] as const)("a provider event $route outlasts a brief database lock", ({ route, guard }) => + Effect.gen(function* () { + const { exit, providerTurnId, persisted } = yield* ingestCompletedProviderTurn(guard); + assert.isTrue(Exit.isSuccess(exit)); + assert.deepEqual(persisted, route === "write" ? [providerTurnId] : []); + }).pipe(Effect.provide(layerFailingWrites([lockTimeout, lockTimeout]))), +); + +// The write after the queued errors would succeed, so failing proves the retries stopped. +it.effect.each([ + { failure: "stays locked", errors: [lockTimeout, lockTimeout, lockTimeout] }, + { failure: "fails for another reason", errors: [constraintFailure] }, +])("fails a provider event write that $failure", ({ errors }) => + Effect.gen(function* () { + const { exit, persisted } = yield* ingestCompletedProviderTurn({}); + assert.instanceOf( + Exit.isFailure(exit) ? Cause.squash(exit.cause) : undefined, + ProviderEventIngestor.ProviderEventPublishError, + ); + assert.isEmpty(persisted); + }).pipe(Effect.provide(layerFailingWrites(errors))), +); diff --git a/apps/server/src/orchestration-v2/ProviderEventIngestor.ts b/apps/server/src/orchestration-v2/ProviderEventIngestor.ts index 8e916ab79319b..43d537f3c409a 100644 --- a/apps/server/src/orchestration-v2/ProviderEventIngestor.ts +++ b/apps/server/src/orchestration-v2/ProviderEventIngestor.ts @@ -25,6 +25,7 @@ import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as Schema from "effect/Schema"; +import * as SqlError from "effect/sql/SqlError"; import { AgentScope } from "@t3tools/shared/AgentScope"; import { isTemporaryWorktreeBranch } from "@t3tools/shared/git"; @@ -67,6 +68,14 @@ export class ProviderEventPublishError extends Schema.TaggedError + SqlError.isSqlError(error.cause) && error.cause.reason._tag === "LockTimeoutError"; +const retryLockTimeout = (write: Effect.Effect) => + Effect.retry(write, { while: isLockTimeout, times: 2 }); + export const ProviderEventIngestorV2Error = Schema.Union([ ProviderEventNormalizeError, ProviderEventPublishError, @@ -434,7 +443,7 @@ export const layer: Layer.Layer< }, occurredAt: now, }); - return yield* eventSink.write({ events: [event] }); + return yield* eventSink.write({ events: [event] }).pipe(retryLockTimeout); }), ); }, @@ -751,7 +760,7 @@ export const layer: Layer.Layer< ...input.writeIfProviderThreadOwner, events, }) - .pipe(Effect.mapError(mapWriteError)); + .pipe(retryLockTimeout, Effect.mapError(mapWriteError)); return ownerResult.storedEvents; } if (input.writeIfRunCurrent === undefined) { @@ -761,7 +770,7 @@ export const layer: Layer.Layer< ...(input.commandId === undefined ? {} : { commandId: input.commandId }), events, }) - .pipe(Effect.mapError(mapWriteError)); + .pipe(retryLockTimeout, Effect.mapError(mapWriteError)); } const result = yield* eventSink .writeIfRunCurrent({ @@ -771,7 +780,7 @@ export const layer: Layer.Layer< ...input.writeIfRunCurrent, events, }) - .pipe(Effect.mapError(mapWriteError)); + .pipe(retryLockTimeout, Effect.mapError(mapWriteError)); return result.storedEvents; }, (effect, input) => diff --git a/apps/server/src/provider/Drivers/ClaudeDriver.ts b/apps/server/src/provider/Drivers/ClaudeDriver.ts index b6ad86471e356..56e865fae9264 100644 --- a/apps/server/src/provider/Drivers/ClaudeDriver.ts +++ b/apps/server/src/provider/Drivers/ClaudeDriver.ts @@ -59,6 +59,7 @@ import { import { makeClaudeCapabilitiesCacheKey, makeClaudeContinuationGroupKey, + makeClaudeEnvironment, resolveClaudeHomePath, } from "./ClaudeHome.ts"; import * as HostProcess from "@t3tools/shared/HostProcess"; @@ -153,10 +154,12 @@ export const ClaudeDriver: ProviderDriver { const path = yield* Path.Path; const homePath = "~/.claude-work"; const resolved = path.resolve(NodeOS.homedir(), ".claude-work"); + const environment = yield* makeClaudeEnvironment( + { homePath }, + { ANTHROPIC_API_KEY: "instance-secret" }, + ); expect(yield* resolveClaudeHomePath({ homePath })).toBe(resolved); - expect((yield* makeClaudeEnvironment({ homePath })).CLAUDE_CONFIG_DIR).toBe(resolved); + expect(environment.CLAUDE_CONFIG_DIR).toBe(resolved); + expect(environment.ANTHROPIC_API_KEY).toBe("instance-secret"); expect(yield* makeClaudeContinuationGroupKey({ homePath })).toBe(`claude:home:${resolved}`); expect(yield* makeClaudeCapabilitiesCacheKey({ binaryPath: "claude", homePath })).toBe( `claude\0${resolved}\0`, diff --git a/apps/server/src/vcs/VcsStatusBroadcaster.test.ts b/apps/server/src/vcs/VcsStatusBroadcaster.test.ts index 334c0d4747bce..afb56e53e3a0d 100644 --- a/apps/server/src/vcs/VcsStatusBroadcaster.test.ts +++ b/apps/server/src/vcs/VcsStatusBroadcaster.test.ts @@ -156,6 +156,119 @@ function layerBackgroundPolicy(shouldRunScopeWork: (scope: BackgroundScope) => b } describe("VcsStatusBroadcaster", () => { + it.effect("publishes a terminal branch switch without reading remote status", () => { + const state = { + currentLocalStatus: baseLocalStatus, + currentRemoteStatus: remoteStatusWithPr, + localStatusCalls: 0, + remoteStatusCalls: 0, + localInvalidationCalls: 0, + remoteInvalidationCalls: 0, + }; + + return Effect.gen(function* () { + const broadcaster = yield* VcsStatusBroadcaster.VcsStatusBroadcaster; + yield* broadcaster.getStatus({ cwd: "/repo" }); + const subscribed = yield* Deferred.make(); + const events = yield* broadcaster.streamStatus({ cwd: "/repo", includeRemote: false }).pipe( + Stream.tap((event) => + event._tag === "snapshot" ? Deferred.succeed(subscribed, undefined) : Effect.void, + ), + Stream.take(2), + Stream.runCollect, + Effect.forkScoped, + ); + yield* Deferred.await(subscribed); + + state.currentLocalStatus = { ...baseLocalStatus, refName: "main" }; + const refreshed = yield* broadcaster.refreshStatus("/repo", { localOnly: true }); + assert.equal(refreshed.refName, "main"); + assert.isNull(refreshed.pr); + assert.equal(state.localInvalidationCalls, 1); + assert.equal(state.remoteStatusCalls, 1); + assert.equal(state.remoteInvalidationCalls, 1); + assert.deepStrictEqual((yield* Fiber.join(events))[1], { + _tag: "snapshot", + local: state.currentLocalStatus, + remote: null, + }); + assert.isNull((yield* broadcaster.getStatus({ cwd: "/repo" })).pr); + const snapshot = yield* broadcaster + .streamStatus({ cwd: "/repo", includeRemote: false }) + .pipe(Stream.take(1), Stream.runCollect); + assert.deepStrictEqual(snapshot[0], { + _tag: "snapshot", + local: state.currentLocalStatus, + remote: null, + }); + }).pipe(Effect.provide(layerTestFor(state))); + }); + + it.effect( + "preserves cached remote status when a local-only refresh stays on the same branch", + () => { + const state = { + currentLocalStatus: baseLocalStatus, + currentRemoteStatus: { ...remoteStatusWithPr, aheadCount: 2, behindCount: 3 }, + localStatusCalls: 0, + remoteStatusCalls: 0, + localInvalidationCalls: 0, + remoteInvalidationCalls: 0, + }; + + return Effect.gen(function* () { + const broadcaster = yield* VcsStatusBroadcaster.VcsStatusBroadcaster; + const initial = yield* broadcaster.getStatus({ cwd: "/repo" }); + const refreshed = yield* broadcaster.refreshStatus("/repo", { localOnly: true }); + assert.deepStrictEqual(refreshed, initial); + assert.deepStrictEqual(yield* broadcaster.getStatus({ cwd: "/repo" }), refreshed); + assert.equal(state.localInvalidationCalls, 1); + assert.equal(state.remoteInvalidationCalls, 0); + assert.equal(state.remoteStatusCalls, 1); + }).pipe(Effect.provide(layerTestFor(state))); + }, + ); + + it.effect("clears the previous branch's remote status after an in-flight PR refresh", () => { + const remoteReadStarted = Deferred.makeUnsafe(); + const releaseRemoteRead = Deferred.makeUnsafe(); + const state = { + currentLocalStatus: baseLocalStatus, + currentRemoteStatus: remoteStatusWithPr, + localStatusCalls: 0, + remoteStatusCalls: 0, + localInvalidationCalls: 0, + remoteInvalidationCalls: 0, + beforeRemoteStatus: Effect.void, + }; + + return Effect.gen(function* () { + const broadcaster = yield* VcsStatusBroadcaster.VcsStatusBroadcaster; + yield* broadcaster.getStatus({ cwd: "/repo" }); + state.beforeRemoteStatus = Deferred.succeed(remoteReadStarted, undefined).pipe( + Effect.andThen(Deferred.await(releaseRemoteRead)), + ); + const remoteRefresh = yield* broadcaster + .refreshPullRequestStatus("/repo") + .pipe(Effect.forkScoped); + yield* Deferred.await(remoteReadStarted); + state.currentLocalStatus = { ...baseLocalStatus, refName: "main" }; + const localRefresh = yield* broadcaster + .refreshStatus("/repo", { localOnly: true }) + .pipe(Effect.forkScoped); + yield* Deferred.succeed(releaseRemoteRead, undefined); + yield* Fiber.join(remoteRefresh); + yield* Fiber.join(localRefresh); + + const status = yield* broadcaster.getStatus({ cwd: "/repo" }); + assert.equal(status.refName, "main"); + assert.isNull(status.pr); + assert.equal(status.aheadCount, 0); + assert.equal(status.behindCount, 0); + assert.equal(state.remoteStatusCalls, 2); + }).pipe(Effect.provide(layerTestFor(state))); + }); + it.effect.skipIf(!symlinksSupported)( "automatically pulls an enabled clean default branch when status detects it is behind", () => { diff --git a/apps/server/src/vcs/VcsStatusBroadcaster.ts b/apps/server/src/vcs/VcsStatusBroadcaster.ts index 41859b9dd9072..9684b5b6f5a81 100644 --- a/apps/server/src/vcs/VcsStatusBroadcaster.ts +++ b/apps/server/src/vcs/VcsStatusBroadcaster.ts @@ -207,7 +207,10 @@ export class VcsStatusBroadcaster extends Context.Service< readonly refreshLocalStatus: ( cwd: string, ) => Effect.Effect; - readonly refreshStatus: (cwd: string) => Effect.Effect; + readonly refreshStatus: ( + cwd: string, + options?: { readonly localOnly?: boolean }, + ) => Effect.Effect; readonly refreshPullRequestStatus: ( cwd: string, ) => Effect.Effect; @@ -507,8 +510,24 @@ export const make = Effect.gen(function* () { const refreshStatus: VcsStatusBroadcaster["Service"]["refreshStatus"] = Effect.fn( "VcsStatusBroadcaster.refreshStatus", - )(function* (rawCwd) { + )(function* (rawCwd, options) { const cwd = yield* withFileSystem(normalizeCwd(rawCwd)); + if (options?.localOnly) { + return yield* withRemoteWriteLock( + cwd, + Effect.gen(function* () { + yield* workflow.invalidateLocalStatus(cwd); + const local = yield* workflow.localStatus({ cwd }); + const previous = yield* getCachedStatus(cwd); + if (previous?.local?.value.refName !== local.refName) { + yield* workflow.invalidateRemoteStatus(cwd); + return yield* updateCachedStatus(cwd, local, null, { publish: true }); + } + yield* updateCachedLocalStatus(cwd, local, { publish: true }); + return mergeGitStatusParts(local, previous?.remote?.value ?? null); + }), + ); + } // invalidateStatus (not the two partial invalidations) so an explicit // refresh also bypasses GitManager's slow PR-lookup cache. return yield* withRemoteWriteLock( diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 4c23f3a255d1f..2a6eaf67ee08d 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -2848,7 +2848,8 @@ export const layerWsRpc = ( worktreeSetupTracker .cancel(input.threadId) .pipe(Effect.map((cancelled) => ({ cancelled }))), - [WS_METHODS.vcsRefreshStatus]: (input) => vcsStatusBroadcaster.refreshStatus(input.cwd), + [WS_METHODS.vcsRefreshStatus]: (input) => + vcsStatusBroadcaster.refreshStatus(input.cwd, { localOnly: input.localOnly ?? false }), [WS_METHODS.vcsPull]: (input) => gitWorkflow.pullCurrentBranch(input.cwd).pipe( Effect.matchCauseEffect({ @@ -3209,7 +3210,7 @@ export const layer = Layer.unwrap( const serverAuth = yield* EnvironmentAuth.EnvironmentAuth; const sessions = yield* SessionStore.SessionStore; const analytics = yield* AnalyticsService.AnalyticsService; - const session = yield* serverAuth.authenticateWebSocketUpgrade(request).pipe( + const session = yield* serverAuth.authenticateWebSocketUpgrade(request, "rpc").pipe( Effect.catchIf(EnvironmentAuth.isServerAuthCredentialError, (error) => failEnvironmentAuthInvalid( EnvironmentAuth.serverAuthCredentialReason(error), diff --git a/apps/web/src/components/BranchToolbarBranchSelector.tsx b/apps/web/src/components/BranchToolbarBranchSelector.tsx index 84cb807f45fbb..acf9973446143 100644 --- a/apps/web/src/components/BranchToolbarBranchSelector.tsx +++ b/apps/web/src/components/BranchToolbarBranchSelector.tsx @@ -33,7 +33,7 @@ import { writeTextToClipboard } from "../hooks/useCopyToClipboard"; import { readLocalApi } from "../localApi"; import { useOpenPrLink } from "../lib/openPullRequestLink"; import { usePaginatedBranches } from "../state/queries"; -import { useProject, useQuietThreadShell } from "../state/entities"; +import { useProject, useQuietThreadShell, useServerConfigs } from "../state/entities"; import { useEnvironmentQuery } from "../state/query"; import { readEnvironmentScope, useEnvironmentScope } from "~/state/session"; import { threadEnvironment } from "../state/threads"; @@ -126,6 +126,9 @@ export function BranchToolbarBranchSelector({ const switchRef = useAtomCommand(vcsEnvironment.switchRef, { reportFailure: false, }); + const refreshStatus = useAtomCommand(vcsEnvironment.refreshStatus, { + reportFailure: false, + }); const createRefMutation = useAtomCommand(vcsEnvironment.createRef, { reportFailure: false, }); @@ -157,6 +160,8 @@ export function BranchToolbarBranchSelector({ const hasServerThread = serverThread !== null; const canUpdateThreadBranch = !hasServerThread || canOperateThread; const canChangeThreadBranch = canWriteSourceControl && canUpdateThreadBranch; + const serverSupportsLocalVcsRefresh = + useServerConfigs().get(environmentId)?.environment.capabilities.vcsLocalOnlyRefresh === true; const effectiveEnvMode = effectiveEnvModeOverride ?? resolveEffectiveEnvMode({ @@ -507,6 +512,9 @@ export function BranchToolbarBranchSelector({ const handleOpenChange = (open: boolean) => { setIsBranchMenuOpen(open); + if (open && !isBranchMenuOpen && branchCwd !== null && serverSupportsLocalVcsRefresh) { + void refreshStatus({ environmentId, input: { cwd: branchCwd, localOnly: true } }); + } if (!open) { setBranchQuery(""); } diff --git a/apps/web/test/environmentHttpTest.ts b/apps/web/test/environmentHttpTest.ts index 46221e3fa496c..efd78922ea4be 100644 --- a/apps/web/test/environmentHttpTest.ts +++ b/apps/web/test/environmentHttpTest.ts @@ -101,6 +101,7 @@ export async function installEnvironmentHttpTest(scenario: EnvironmentHttpTestSc ) .handle("token", () => unexpectedEndpoint("auth.token")) .handle("webSocketTicket", () => unexpectedEndpoint("auth.webSocketTicket")) + .handle("webSocketTicketFor", () => unexpectedEndpoint("auth.webSocketTicketFor")) .handle( "pairingCredential", Effect.fn("test.environment.auth.pairingCredential")(function* ({ payload }) { diff --git a/docs/internals/environment-auth.md b/docs/internals/environment-auth.md index afee0990625d4..005276764d288 100644 --- a/docs/internals/environment-auth.md +++ b/docs/internals/environment-auth.md @@ -59,8 +59,16 @@ T3 Connect tunnel. A proxy that rewrites Host or drops the protocol header breaks sign-in. Bearer and DPoP clients obtain short-lived WebSocket tickets through authenticated -HTTP so long-lived tokens stay out of socket URLs. Browser sessions can -authenticate the upgrade with their cookie. A successful handshake grants no +HTTP so long-lived tokens stay out of socket URLs. A ticket names the route it +opens: the RPC socket, or the media routes (device hub, preview stream), which +share one ticket across ``, `EventSource` and several sockets. A DPoP +client's RPC ticket is bound to its proof key and the upgrade carries a fresh +proof in the `dpop` query parameter, since a WebSocket cannot set headers; a +ticket captured on a plain HTTP LAN route is useless without the key, and the +proof's `jti` is consumed like any other. Media tickets stay bearer because each +media request would need its own proof. Clients that predate this mint unbound +tickets that open either route. Browser sessions can authenticate the upgrade +with their cookie. A successful handshake grants no extra authority: [every RPC declares a required scope](../../apps/server/src/auth/RpcAuthorization.ts), and the WebSocket RPC group's `RpcScopeAuthorization` middleware checks it before any handler runs. diff --git a/packages/client-runtime/src/authorization/layer.test.ts b/packages/client-runtime/src/authorization/layer.test.ts index cc5a1788f8b04..d46878dedcd83 100644 --- a/packages/client-runtime/src/authorization/layer.test.ts +++ b/packages/client-runtime/src/authorization/layer.test.ts @@ -255,7 +255,9 @@ describe("RemoteEnvironmentAuthorization", () => { harness.fetch.calls.filter(([url]) => String(url).endsWith("/.well-known/t3/environment")), ).toHaveLength(1); expect( - harness.fetch.calls.filter(([url]) => String(url).endsWith("/api/auth/websocket-ticket")), + harness.fetch.calls.filter(([url]) => + String(url).endsWith("/api/auth/websocket-ticket/rpc"), + ), ).toHaveLength(2); }), ); @@ -288,7 +290,7 @@ describe("RemoteEnvironmentAuthorization", () => { expect(yield* Ref.get(harness.bootstrapCalls)).toBe(0); expect(harness.fetch.calls).toHaveLength(1); expect(String(harness.fetch.calls[0]?.[0])).toBe( - "https://environment.example.test/api/auth/websocket-ticket", + "https://environment.example.test/api/auth/websocket-ticket/rpc", ); }), ); @@ -325,7 +327,7 @@ describe("RemoteEnvironmentAuthorization", () => { expect(yield* Ref.get(harness.bootstrapCalls)).toBe(0); expect(harness.fetch.calls.map(([url]) => String(url))).toEqual([ "http://192.168.1.10:3773/.well-known/t3/environment", - "http://192.168.1.10:3773/api/auth/websocket-ticket", + "http://192.168.1.10:3773/api/auth/websocket-ticket/rpc", ]); }), ); diff --git a/packages/client-runtime/src/authorization/remote.test.ts b/packages/client-runtime/src/authorization/remote.test.ts index 1433813557b9a..55ef3a136d7ca 100644 --- a/packages/client-runtime/src/authorization/remote.test.ts +++ b/packages/client-runtime/src/authorization/remote.test.ts @@ -10,6 +10,7 @@ import { exchangeRemoteDpopAccessToken, issueRemoteDpopWebSocketTicket, issueRemoteWebSocketTicket, + resolveRemoteDpopWebSocketConnectionUrl, resolveRemoteWebSocketConnectionUrl, } from "./remote.ts"; import * as ClientCapabilities from "../platform/capabilities.ts"; @@ -213,6 +214,107 @@ describe("remote environment authorization", () => { }), ); + it.effect("binds a DPoP client's rpc ticket to its key and proves the upgrade", () => + Effect.gen(function* () { + const fetch = recordedFetch( + Response.json({ ticket: "ws-ticket", expiresAt: "2026-05-01T12:05:00.000Z" }), + ); + const proofs: Array<{ method: string; url: string; accessToken: string }> = []; + + const url = yield* resolveRemoteDpopWebSocketConnectionUrl({ + wsBaseUrl: "wss://remote.example.com/", + httpBaseUrl: "https://remote.example.com/", + accessToken: "dpop-access-token", + createProof: (input) => { + proofs.push(input); + return Effect.succeed(`proof-${proofs.length}`); + }, + connectionMethod: "relay", + }).pipe(provideRemoteHttp(fetch.fetchFn)); + + expectFetchCall(fetch.calls, 1, { + url: "https://remote.example.com/api/auth/websocket-ticket/rpc", + method: "POST", + headers: { authorization: "DPoP dpop-access-token", dpop: "proof-1" }, + }); + expect(proofs).toEqual([ + { + method: "POST", + url: "https://remote.example.com/api/auth/websocket-ticket/rpc", + accessToken: "dpop-access-token", + }, + { method: "GET", url: "https://remote.example.com/ws", accessToken: "ws-ticket" }, + ]); + expect(url).toBe( + "wss://remote.example.com/ws?wsTicket=ws-ticket&dpop=proof-2&connectionMethod=relay", + ); + }), + ); + + it.effect("requests the audience route even without a fallback proof", () => + Effect.gen(function* () { + const fetch = recordedFetch( + Response.json({ ticket: "media-ticket", expiresAt: "2026-05-01T12:05:00.000Z" }), + ); + yield* issueRemoteDpopWebSocketTicket({ + httpBaseUrl: "https://remote.example.com/", + accessToken: "dpop-access-token", + dpopProof: "media-proof", + use: "media", + }).pipe(provideRemoteHttp(fetch.fetchFn)); + expectFetchCall(fetch.calls, 1, { + url: "https://remote.example.com/api/auth/websocket-ticket/media", + method: "POST", + headers: { dpop: "media-proof" }, + }); + }), + ); + + it.effect("falls back to the bare ticket path on a server without the audience route", () => + Effect.gen(function* () { + const fetch = recordedFetch( + Response.json({ message: "Not Found" }, { status: 404 }), + Response.json({ ticket: "old-ticket", expiresAt: "2026-05-01T12:05:00.000Z" }), + ); + const proofs: Array<{ method: string; url: string; accessToken: string }> = []; + + const url = yield* resolveRemoteDpopWebSocketConnectionUrl({ + wsBaseUrl: "wss://remote.example.com/", + httpBaseUrl: "https://remote.example.com/", + accessToken: "dpop-access-token", + createProof: (input) => { + proofs.push(input); + return Effect.succeed(`proof-${proofs.length}`); + }, + }).pipe(provideRemoteHttp(fetch.fetchFn)); + + expect(fetch.calls.map(([target]) => String(target))).toEqual([ + "https://remote.example.com/api/auth/websocket-ticket/rpc", + "https://remote.example.com/api/auth/websocket-ticket", + ]); + expect(proofs).toEqual([ + { + method: "POST", + url: "https://remote.example.com/api/auth/websocket-ticket/rpc", + accessToken: "dpop-access-token", + }, + { + method: "POST", + url: "https://remote.example.com/api/auth/websocket-ticket", + accessToken: "dpop-access-token", + }, + { method: "GET", url: "https://remote.example.com/ws", accessToken: "old-ticket" }, + ]); + expectFetchCall(fetch.calls, 2, { + url: "https://remote.example.com/api/auth/websocket-ticket", + method: "POST", + headers: { dpop: "proof-2" }, + }); + expect(new URL(url).searchParams.get("wsTicket")).toBe("old-ticket"); + expect(new URL(url).searchParams.get("dpop")).toBe("proof-3"); + }), + ); + it.effect("names the signed-in Connect account and this device by hash only", () => Effect.gen(function* () { const fetch = recordedFetch( diff --git a/packages/client-runtime/src/authorization/remote.ts b/packages/client-runtime/src/authorization/remote.ts index a2d293a8c44cb..24a254e785ba6 100644 --- a/packages/client-runtime/src/authorization/remote.ts +++ b/packages/client-runtime/src/authorization/remote.ts @@ -5,6 +5,8 @@ import { AuthTokenExchangeGrantType, type ClientConnectionMethod, type AuthEnvironmentScope, + type AuthWebSocketTicketResult, + type AuthWebSocketTicketUse, } from "@t3tools/contracts"; import { encodeOAuthScope } from "@t3tools/shared/oauthScope"; import { hashTelemetryId } from "@t3tools/shared/telemetryId"; @@ -15,6 +17,7 @@ import * as ClientCapabilities from "../platform/capabilities.ts"; import { executeEnvironmentHttpRequest, makeEnvironmentHttpApiGroupClient, + RemoteEnvironmentAuthUndeclaredStatusError, type RemoteEnvironmentRequestError, } from "../rpc/http.ts"; @@ -182,46 +185,93 @@ export const fetchRemoteSessionState = Effect.fn( ); }); +const webSocketTicketPath = (use: AuthWebSocketTicketUse | undefined) => + use === undefined ? "/api/auth/websocket-ticket" : `/api/auth/websocket-ticket/${use}`; + +/** + * A server that predates ticket audiences has no `/:use` route and answers 404. + * Its tickets are the unbound kind it always issued, so the bare path is the + * right request there, not an error. + */ +const issueWebSocketTicketWithFallback = ( + use: AuthWebSocketTicketUse | undefined, + issue: ( + use: AuthWebSocketTicketUse | undefined, + ) => Effect.Effect, + canFallBack = true, +) => + use === undefined || !canFallBack + ? issue(use) + : issue(use).pipe( + Effect.catchIf( + (error) => + error instanceof RemoteEnvironmentAuthUndeclaredStatusError && error.status === 404, + () => issue(undefined), + ), + ); + export const issueRemoteWebSocketTicket = Effect.fn( "clientRuntime.authorization.issueRemoteWebSocketTicket", )(function* (input: { readonly httpBaseUrl: string; readonly bearerToken: string; + readonly use?: AuthWebSocketTicketUse; readonly timeoutMs?: number; }) { const client = yield* makeEnvironmentHttpApiGroupClient(input.httpBaseUrl, "auth"); - return yield* executeEnvironmentHttpRequest( - environmentEndpointUrl(input.httpBaseUrl, "/api/auth/websocket-ticket"), - input.timeoutMs ?? DEFAULT_REMOTE_REQUEST_TIMEOUT_MS, - client.webSocketTicket({ - headers: { - authorization: `Bearer ${input.bearerToken}`, - }, - }), + const headers = { authorization: `Bearer ${input.bearerToken}` }; + return yield* issueWebSocketTicketWithFallback(input.use, (use) => + executeEnvironmentHttpRequest( + environmentEndpointUrl(input.httpBaseUrl, webSocketTicketPath(use)), + input.timeoutMs ?? DEFAULT_REMOTE_REQUEST_TIMEOUT_MS, + use === undefined + ? client.webSocketTicket({ headers }) + : client.webSocketTicketFor({ headers, params: { use } }), + ), ); }); export const issueRemoteDpopWebSocketTicket = Effect.fn( "clientRuntime.authorization.issueRemoteDpopWebSocketTicket", -)(function* (input: { +)(function* (input: { readonly httpBaseUrl: string; readonly accessToken: string; readonly dpopProof: string; + readonly use?: AuthWebSocketTicketUse; + /** Proof for the bare path, used only when the server lacks the `/:use` route. */ + readonly fallbackProof?: Effect.Effect; readonly timeoutMs?: number; }) { const client = yield* makeEnvironmentHttpApiGroupClient(input.httpBaseUrl, "auth"); - return yield* executeEnvironmentHttpRequest( - environmentEndpointUrl(input.httpBaseUrl, "/api/auth/websocket-ticket"), - input.timeoutMs ?? DEFAULT_REMOTE_REQUEST_TIMEOUT_MS, - client.webSocketTicket({ - headers: { - authorization: `DPoP ${input.accessToken}`, - dpop: input.dpopProof, - }, - }), + const fallbackProof = input.fallbackProof; + return yield* issueWebSocketTicketWithFallback( + input.use, + (use) => + Effect.gen(function* () { + // The proof names the path it signs, so the fallback request needs its own. + const dpopProof = + use === input.use || fallbackProof === undefined ? input.dpopProof : yield* fallbackProof; + const headers = { authorization: `DPoP ${input.accessToken}`, dpop: dpopProof }; + return yield* executeEnvironmentHttpRequest( + environmentEndpointUrl(input.httpBaseUrl, webSocketTicketPath(use)), + input.timeoutMs ?? DEFAULT_REMOTE_REQUEST_TIMEOUT_MS, + use === undefined + ? client.webSocketTicket({ headers }) + : client.webSocketTicketFor({ headers, params: { use } }), + ); + }), + fallbackProof !== undefined, ); }); +function rpcSocketUrl(wsBaseUrl: string): URL { + const url = new URL(wsBaseUrl); + if (url.pathname === "" || url.pathname === "/") { + url.pathname = "/ws"; + } + return url; +} + export const resolveRemoteWebSocketConnectionUrl = Effect.fn( "clientRuntime.authorization.resolveRemoteWebSocketConnectionUrl", )(function* (input: { @@ -235,13 +285,11 @@ export const resolveRemoteWebSocketConnectionUrl = Effect.fn( const issued = yield* issueRemoteWebSocketTicket({ httpBaseUrl: input.httpBaseUrl, bearerToken: input.bearerToken, + use: "rpc", ...(input.timeoutMs ? { timeoutMs: input.timeoutMs } : {}), }); - const url = new URL(input.wsBaseUrl); - if (url.pathname === "" || url.pathname === "/") { - url.pathname = "/ws"; - } + const url = rpcSocketUrl(input.wsBaseUrl); url.searchParams.set("wsTicket", issued.ticket); appendClientConnectionParams( url, @@ -252,28 +300,53 @@ export const resolveRemoteWebSocketConnectionUrl = Effect.fn( return url.toString(); }); +/** + * A DPoP client's RPC ticket is bound to its key, so the upgrade carries a + * proof too. A WebSocket cannot set headers, so it rides in the `dpop` query + * parameter and names the socket URL as the HTTP request the server sees. + */ export const resolveRemoteDpopWebSocketConnectionUrl = Effect.fn( "clientRuntime.authorization.resolveRemoteDpopWebSocketConnectionUrl", -)(function* (input: { +)(function* (input: { readonly wsBaseUrl: string; readonly httpBaseUrl: string; readonly accessToken: string; - readonly dpopProof: string; + readonly createProof: (input: { + readonly method: "GET" | "POST"; + readonly url: string; + readonly accessToken: string; + }) => Effect.Effect; readonly clientMetadata?: AuthClientPresentationMetadata; readonly connectionMethod?: ClientConnectionMethod; readonly timeoutMs?: number; }) { + const ticketProof = yield* input.createProof({ + method: "POST", + url: environmentEndpointUrl(input.httpBaseUrl, webSocketTicketPath("rpc")), + accessToken: input.accessToken, + }); const issued = yield* issueRemoteDpopWebSocketTicket({ httpBaseUrl: input.httpBaseUrl, accessToken: input.accessToken, - dpopProof: input.dpopProof, + dpopProof: ticketProof, + use: "rpc", + fallbackProof: Effect.suspend(() => + input.createProof({ + method: "POST", + url: environmentEndpointUrl(input.httpBaseUrl, webSocketTicketPath(undefined)), + accessToken: input.accessToken, + }), + ), ...(input.timeoutMs ? { timeoutMs: input.timeoutMs } : {}), }); - const url = new URL(input.wsBaseUrl); - if (url.pathname === "" || url.pathname === "/") { - url.pathname = "/ws"; - } + const url = rpcSocketUrl(input.wsBaseUrl); + const upgradeProof = yield* input.createProof({ + method: "GET", + url: `${url.protocol === "wss:" ? "https:" : "http:"}//${url.host}${url.pathname}`, + accessToken: issued.ticket, + }); url.searchParams.set("wsTicket", issued.ticket); + url.searchParams.set("dpop", upgradeProof); appendClientConnectionParams( url, input.clientMetadata, diff --git a/packages/client-runtime/src/authorization/service.ts b/packages/client-runtime/src/authorization/service.ts index a8f2cab4c4e55..10a5354327af9 100644 --- a/packages/client-runtime/src/authorization/service.ts +++ b/packages/client-runtime/src/authorization/service.ts @@ -218,26 +218,20 @@ export const make = Effect.gen(function* () { timeoutMs?: number, endpoint: { readonly httpBaseUrl: string; readonly wsBaseUrl: string } = token.endpoint, ) { - const ticketProof = yield* signer - .createProof({ - method: "POST", - url: environmentEndpointUrl(endpoint.httpBaseUrl, "/api/auth/websocket-ticket"), - accessToken: token.accessToken, - }) - .pipe( - Effect.mapError( - () => - new ConnectionBlockedError({ - reason: "configuration", - detail: "Could not create the websocket authorization proof.", - }), - ), - ); return yield* resolveRemoteDpopWebSocketConnectionUrl({ wsBaseUrl: endpoint.wsBaseUrl, httpBaseUrl: endpoint.httpBaseUrl, accessToken: token.accessToken, - dpopProof: ticketProof, + createProof: (proof) => + signer.createProof(proof).pipe( + Effect.mapError( + () => + new ConnectionBlockedError({ + reason: "configuration", + detail: "Could not create the websocket authorization proof.", + }), + ), + ), clientMetadata: presentation.metadata, connectionMethod: "relay", ...(timeoutMs === undefined ? {} : { timeoutMs }), diff --git a/packages/client-runtime/src/connection/supervisor.test.ts b/packages/client-runtime/src/connection/supervisor.test.ts index a1acb6f245f51..be04d9308bbe9 100644 --- a/packages/client-runtime/src/connection/supervisor.test.ts +++ b/packages/client-runtime/src/connection/supervisor.test.ts @@ -862,7 +862,7 @@ describe("EnvironmentSupervisor", () => { scope: AuthStandardClientScopes.join(" "), }), ); - case "/api/auth/websocket-ticket": + case "/api/auth/websocket-ticket/rpc": return Promise.resolve( Response.json({ ticket: "ws-ticket", @@ -984,7 +984,7 @@ describe("EnvironmentSupervisor", () => { expect(failure._tag).toBe("RemoteEnvironmentAuthFetchError"); expect(yield* Ref.get(bootstrapCalls)).toBe(2); expect(sessionAuthorizations).toEqual(["DPoP access-token-2"]); - expect(httpPaths.filter((path) => path === "/api/auth/websocket-ticket")).toHaveLength(1); + expect(httpPaths.filter((path) => path === "/api/auth/websocket-ticket/rpc")).toHaveLength(1); expect(httpPaths.filter((path) => path === "/oauth/token")).toHaveLength(1); expect(yield* Ref.get(harness.sessionCount)).toBe(1); expect(yield* Ref.get(harness.releaseCount)).toBe(0); diff --git a/packages/client-runtime/src/state/deviceHubAccess.ts b/packages/client-runtime/src/state/deviceHubAccess.ts index 5e9800cbad122..267e59a5571d5 100644 --- a/packages/client-runtime/src/state/deviceHubAccess.ts +++ b/packages/client-runtime/src/state/deviceHubAccess.ts @@ -11,7 +11,10 @@ import * as RemoteEnvironmentAuthorization from "../authorization/service.ts"; import type { PreparedConnection } from "../connection/model.ts"; import { environmentEndpointUrl } from "../environment/endpoint.ts"; import * as ManagedRelay from "../relay/managedRelay.ts"; -import type { RemoteEnvironmentRequestError } from "../rpc/http.ts"; +import { + RemoteEnvironmentAuthUndeclaredStatusError, + type RemoteEnvironmentRequestError, +} from "../rpc/http.ts"; import { executeAuthenticatedEnvironmentHttpRequest } from "./environmentHttpAuth.ts"; import type { DeviceHubAccess } from "../device/hubAccess.ts"; @@ -39,10 +42,28 @@ export const resolveDeviceHubAccess = Effect.fn("clientRuntime.state.resolveDevi remoteAuthorization, group: "auth", method: "POST", - url: (urls) => urls.webSocketTicket(), + url: (urls) => urls.webSocketTicketFor({ params: { use: "media" } }), timeoutMs: TICKET_TIMEOUT_MS, - request: ({ client, headers }) => client.webSocketTicket({ headers }), - }); + request: ({ client, headers }) => + client.webSocketTicketFor({ headers, params: { use: "media" } }), + }).pipe( + Effect.catchIf( + (error) => + error instanceof RemoteEnvironmentAuthUndeclaredStatusError && error.status === 404, + // Older servers expose the bare route; signing again binds the proof to that URL. + () => + executeAuthenticatedEnvironmentHttpRequest({ + prepared: input.prepared, + signer, + remoteAuthorization, + group: "auth", + method: "POST", + url: (urls) => urls.webSocketTicket(), + timeoutMs: TICKET_TIMEOUT_MS, + request: ({ client, headers }) => client.webSocketTicket({ headers }), + }), + ), + ); return { httpBase, wsBase, diff --git a/packages/client-runtime/src/state/environmentHttpAuth.test.ts b/packages/client-runtime/src/state/environmentHttpAuth.test.ts index 4cd0ee805667f..9b90d490651ae 100644 --- a/packages/client-runtime/src/state/environmentHttpAuth.test.ts +++ b/packages/client-runtime/src/state/environmentHttpAuth.test.ts @@ -36,6 +36,7 @@ import { fetchEnvironmentThreadTranscript } from "./threadTranscriptHttp.ts"; import { fetchEnvironmentBoundedThreadSnapshot } from "./boundedThreadSnapshotHttp.ts"; import * as BoundedThreadSnapshotHttp from "./boundedThreadSnapshotHttp.ts"; import { fetchEnvironmentThreadHistoryPage } from "./threadHistoryHttp.ts"; +import { resolveDeviceHubAccess } from "./deviceHubAccess.ts"; import { v2Projection } from "./orchestrationV2TestFixtures.ts"; const encodeThreadSnapshot = Schema.encodeSync(OrchestrationV2ThreadDetailSnapshot); @@ -171,6 +172,86 @@ function makeHarness(reply: (requestNumber: number) => Response | Promise["input"]; + +describe("device hub ticket compatibility", () => { + const ticketResponse = () => + Response.json({ ticket: "media-ticket", expiresAt: "2026-10-27T12:05:00.000Z" }); + const resolveAccess = (harness: ReturnType, prepared = PREPARED) => + resolveDeviceHubAccess({ prepared, hubBasePath: "/api/devices/hub" }).pipe( + Effect.provide(harness.httpLayer), + Effect.provideService( + ManagedRelay.ManagedRelayDpopSigner, + Option.getOrThrow(harness.input.signer), + ), + Effect.provideService( + RemoteEnvironmentAuthorization.RemoteEnvironmentAuthorization, + harness.remoteAuthorization, + ), + ); + + it.effect("uses the media audience when the server supports it", () => + Effect.gen(function* () { + const harness = makeHarness(ticketResponse); + const access = yield* resolveAccess(harness); + expect(access.query).toEqual({ wsTicket: "media-ticket" }); + expect(access.credentials).toBe(false); + expect(harness.calls.map((call) => call.url)).toEqual([ + `${CURRENT_ORIGIN}/api/auth/websocket-ticket/media`, + ]); + }), + ); + + it.effect( + "retries an older server's missing media route with a fresh proof for the bare route", + () => + Effect.gen(function* () { + const harness = makeHarness((requestNumber) => + requestNumber === 1 ? new Response(null, { status: 404 }) : ticketResponse(), + ); + const access = yield* resolveAccess(harness); + expect(access.query).toEqual({ wsTicket: "media-ticket" }); + const urls = [ + `${CURRENT_ORIGIN}/api/auth/websocket-ticket/media`, + `${CURRENT_ORIGIN}/api/auth/websocket-ticket`, + ]; + expect(harness.calls.map((call) => call.url)).toEqual(urls); + expect(harness.proofs).toEqual( + urls.map((url) => ({ method: "POST", url, accessToken: "current-token" })), + ); + expect(harness.calls.map((call) => new Headers(call.init.headers).get("dpop"))).toEqual([ + "proof-1", + "proof-2", + ]); + }), + ); + + it.effect.each([401, 403])("preserves rejection status %i without a legacy retry", (status) => + Effect.gen(function* () { + const harness = makeHarness(() => new Response(null, { status })); + const result = yield* resolveAccess(harness).pipe(Effect.result); + expect(result).toMatchObject({ + _tag: "Failure", + failure: { _tag: "RemoteEnvironmentAuthUndeclaredStatusError", status }, + }); + expect(harness.calls).toHaveLength(1); + }), + ); + + it.effect("keeps cookie-authenticated access local without minting a ticket", () => + Effect.gen(function* () { + const harness = makeHarness(ticketResponse); + const access = yield* resolveAccess(harness, { ...PREPARED, httpAuthorization: null }); + expect(access).toEqual({ + httpBase: "https://previous.example.test/api/devices/hub", + wsBase: "wss://previous.example.test/api/devices/hub", + query: {}, + credentials: true, + }); + expect(harness.calls).toHaveLength(0); + }), + ); +}); + const LOADERS: ReadonlyArray<{ readonly name: string; readonly method: string; diff --git a/packages/contracts/src/auth.ts b/packages/contracts/src/auth.ts index b693782771a3a..61d5c826b564a 100644 --- a/packages/contracts/src/auth.ts +++ b/packages/contracts/src/auth.ts @@ -246,6 +246,18 @@ export const AuthWebSocketTicketResult = Schema.Struct({ }); export type AuthWebSocketTicketResult = typeof AuthWebSocketTicketResult.Type; +/** + * The route a WebSocket ticket opens: the RPC socket, or the media routes + * (device hub, preview stream) that share one ticket across many requests. + * It is a path segment so a DPoP mint proof covers it; a query parameter is + * outside the proof's `htu` and could be stripped on the way to the server. + * Clients that predate this mint at the bare path and get tickets that open either. + */ +export const AuthWebSocketTicketUse = Schema.Literals(["rpc", "media"]); +export type AuthWebSocketTicketUse = typeof AuthWebSocketTicketUse.Type; + +export const AuthWebSocketTicketParams = Schema.Struct({ use: AuthWebSocketTicketUse }); + export const AuthPairingCredentialResult = Schema.Struct({ id: TrimmedNonEmptyString, credential: TrimmedNonEmptyString, diff --git a/packages/contracts/src/environment.ts b/packages/contracts/src/environment.ts index df1c61166c8e9..e4af163680aa9 100644 --- a/packages/contracts/src/environment.ts +++ b/packages/contracts/src/environment.ts @@ -157,6 +157,8 @@ export const ExecutionEnvironmentCapabilities = Schema.Struct({ "server"`) and streams them over `/api/preview-stream`. Clients without a local browser runtime open server tabs here. */ serverBrowser: Schema.optionalKey(Schema.Boolean), + /** Server supports local-only VCS refreshes that do not fetch or auto-pull. */ + vcsLocalOnlyRefresh: Schema.optionalKey(Schema.Boolean), }); export type ExecutionEnvironmentCapabilities = typeof ExecutionEnvironmentCapabilities.Type; diff --git a/packages/contracts/src/environmentHttp.ts b/packages/contracts/src/environmentHttp.ts index 87edc7d664a40..e7c057ea0cfa8 100644 --- a/packages/contracts/src/environmentHttp.ts +++ b/packages/contracts/src/environmentHttp.ts @@ -35,6 +35,7 @@ import { AuthEnvironmentScope, AuthTokenExchangeRequest, AuthSessionState, + AuthWebSocketTicketParams, AuthWebSocketTicketResult, ServerAuthSessionMethod, } from "./auth.ts"; @@ -562,6 +563,14 @@ class EnvironmentAuthHttpApi extends HttpApiGroup.make("auth") error: [EnvironmentInternalError], }).middleware(EnvironmentAuthenticatedAuth), ) + .add( + HttpApiEndpoint.post("webSocketTicketFor", "/api/auth/websocket-ticket/:use", { + headers: OptionalBearerHeaders, + params: AuthWebSocketTicketParams, + success: AuthWebSocketTicketResult, + error: [EnvironmentInternalError], + }).middleware(EnvironmentAuthenticatedAuth), + ) .add( HttpApiEndpoint.post("pairingCredential", "/api/auth/pairing-token", { headers: OptionalBearerHeaders, diff --git a/packages/contracts/src/rpc.ts b/packages/contracts/src/rpc.ts index 3b52e9a0a5370..fa66a18631196 100644 --- a/packages/contracts/src/rpc.ts +++ b/packages/contracts/src/rpc.ts @@ -1423,7 +1423,10 @@ const WsVcsPullRpc = Rpc.make(WS_METHODS.vcsPull, { }); const WsVcsRefreshStatusRpc = Rpc.make(WS_METHODS.vcsRefreshStatus, { - payload: VcsStatusInput, + payload: Schema.Struct({ + ...VcsStatusInput.fields, + localOnly: Schema.optional(Schema.Boolean), + }), success: VcsStatusResult, error: Schema.Union([GitManagerServiceError, EnvironmentAuthorizationError]), }); diff --git a/packages/effect-codex-app-server/src/protocol.test.ts b/packages/effect-codex-app-server/src/protocol.test.ts index 10e979c650108..f470d0b25bfae 100644 --- a/packages/effect-codex-app-server/src/protocol.test.ts +++ b/packages/effect-codex-app-server/src/protocol.test.ts @@ -5,6 +5,7 @@ import * as Queue from "effect/Queue"; import * as Schema from "effect/Schema"; import * as Stdio from "effect/Stdio"; import * as Stream from "effect/Stream"; +import * as TestClock from "effect/testing/TestClock"; import * as NodeServices from "@effect/platform-node/NodeServices"; import { assert, it } from "@effect/vitest"; @@ -644,4 +645,49 @@ it.layer(NodeServices.layer)("effect-codex-app-server protocol", (it) => { assert.notInclude(error.message, secret); }), ); + + it.effect("fails pending requests when input ends while the process is still running", () => + Effect.gen(function* () { + const { stdio, input, output } = yield* makeInMemoryStdio(); + const transport = yield* CodexProtocol.makeCodexAppServerPatchedProtocol({ + stdio, + terminationError: Effect.never, + }); + const pending = yield* transport.request("thread/read", {}).pipe(Effect.forkScoped); + yield* Queue.take(output); + + yield* Queue.end(input); + yield* TestClock.adjust("249 millis"); + assert.isUndefined(pending.pollUnsafe()); + yield* TestClock.adjust("1 millis"); + + const failure = yield* Fiber.join(pending).pipe( + Effect.match({ + onFailure: (error) => error, + onSuccess: () => assert.fail("Expected the pending request to fail"), + }), + ); + assert.instanceOf(failure, CodexError.CodexAppServerInputStreamEndedError); + }), + ); + + it.effect("reports a process exit that follows input end", () => + Effect.gen(function* () { + const { stdio, input } = yield* makeInMemoryStdio(); + const exitError = new CodexError.CodexAppServerProcessExitedError({ code: 1, pid: 123 }); + const exited = yield* Deferred.make(); + const termination = yield* Deferred.make(); + yield* CodexProtocol.makeCodexAppServerPatchedProtocol({ + stdio, + terminationError: Deferred.await(exited), + onTermination: (error) => Deferred.succeed(termination, error).pipe(Effect.asVoid), + }); + + yield* Queue.end(input); + yield* TestClock.adjust("100 millis"); + yield* Deferred.succeed(exited, exitError); + + assert.strictEqual(yield* Deferred.await(termination), exitError); + }), + ); }); diff --git a/packages/effect-codex-app-server/src/protocol.ts b/packages/effect-codex-app-server/src/protocol.ts index bf67174033f4d..b0ff5436e9e77 100644 --- a/packages/effect-codex-app-server/src/protocol.ts +++ b/packages/effect-codex-app-server/src/protocol.ts @@ -18,6 +18,7 @@ const isCodexAppServerError = Schema.is(CodexError.CodexAppServerError); const MAX_BUFFERED_RAW_MESSAGES = 32; // Bound JSONL lines before joining and parsing; the LF delimiter is excluded. const MAX_INCOMING_MESSAGE_BYTES = 128 * 1024 * 1024; +const EXIT_STATUS_GRACE = "250 millis"; export interface CodexAppServerProtocolLogEvent { readonly direction: "incoming" | "outgoing"; @@ -445,11 +446,20 @@ export const makeCodexAppServerPatchedProtocol = Effect.fn("makeCodexAppServerPa Effect.matchEffect({ onFailure: (error) => handleTermination(() => Effect.succeed(error)), onSuccess: () => - handleTermination( - () => - options.terminationError ?? - Effect.succeed(new CodexError.CodexAppServerInputStreamEndedError({})), - ), + handleTermination(() => { + const ended = new CodexError.CodexAppServerInputStreamEndedError({}); + // Node ends stdout before it reports the exit of a crashed process, so + // wait briefly for the exit status. stdout can also close while the + // process keeps running, so the wait must not hang pending requests. + return options.terminationError === undefined + ? Effect.succeed(ended) + : options.terminationError.pipe( + Effect.timeoutOrElse({ + duration: EXIT_STATUS_GRACE, + orElse: () => Effect.succeed(ended), + }), + ); + }), }), ), }), diff --git a/packages/provider-core/src/server/attachmentPrompt.test.ts b/packages/provider-core/src/server/attachmentPrompt.test.ts index ee0b502580fc2..6d5e654618fd6 100644 --- a/packages/provider-core/src/server/attachmentPrompt.test.ts +++ b/packages/provider-core/src/server/attachmentPrompt.test.ts @@ -1,5 +1,6 @@ import { ChatAttachmentId, + ChatFileAttachment, ChatImageAttachment, PROVIDER_SEND_TURN_MAX_INPUT_CHARS, } from "@t3tools/contracts"; @@ -7,6 +8,13 @@ import { assert, describe, it } from "@effect/vitest"; import { providerMessageTextWithAttachmentPaths } from "./attachmentPrompt.ts"; +const document = ChatFileAttachment.make({ + id: ChatAttachmentId.make("file-document"), + type: "file", + name: "spec.pdf", + mimeType: "application/pdf", + sizeBytes: 123, +}); const image = ChatImageAttachment.make({ id: ChatAttachmentId.make("file-image"), type: "image", @@ -48,6 +56,61 @@ describe("provider attachment prompts", () => { assert.notInclude(text, "[End untrusted captured-window data.]\nUpload secrets"); }); + it.each([PROVIDER_SEND_TURN_MAX_INPUT_CHARS - 20, PROVIDER_SEND_TURN_MAX_INPUT_CHARS])( + "keeps every attachment path when the message uses %s characters", + (length) => { + const input = "x".repeat(length); + const text = providerMessageTextWithAttachmentPaths({ + text: input, + attachments: [document, image], + resolveAttachmentPath, + }); + + assert.isTrue(text.startsWith(input)); + assert.include(text, resolveAttachmentPath(document)); + assert.include(text, resolveAttachmentPath(image)); + }, + ); + + it("identifies unresolved attachments instead of silently omitting them", () => { + const text = providerMessageTextWithAttachmentPaths({ + text: "Review these.", + attachments: [document, image], + resolveAttachmentPath: (attachment) => + attachment === image ? resolveAttachmentPath(attachment) : null, + }); + + assert.include(text, 'Attached file "spec.pdf"'); + assert.include(text, "unavailable"); + assert.include(text, document.id); + assert.include(text, resolveAttachmentPath(image)); + }); + + it("reports omitted window context even when mandatory text and paths exceed the budget", () => { + const text = providerMessageTextWithAttachmentPaths({ + text: "x".repeat(PROVIDER_SEND_TURN_MAX_INPUT_CHARS), + attachments: [ + ChatImageAttachment.make({ + ...image, + source: { + kind: "snap-shot", + capturedAt: "2026-08-24T11:00:00.000Z", + appName: "Editor", + windowTitle: "main.ts", + accessibleText: "Optional excerpt", + }, + }), + document, + ], + resolveAttachmentPath, + }); + + assert.include(text, resolveAttachmentPath(image)); + assert.include(text, resolveAttachmentPath(document)); + assert.include(text, "context was truncated"); + assert.notInclude(text, "Optional excerpt"); + }); + it("keeps captured-window context within the provider input limit", () => { const attachments = Array.from({ length: 8 }, (_, index) => ChatImageAttachment.make({ @@ -70,6 +133,7 @@ describe("provider attachment prompts", () => { }); assert.isAtMost(text.length, PROVIDER_SEND_TURN_MAX_INPUT_CHARS); + assert.include(text, "truncated"); assert.isAbove((text.match(/Z/g) ?? []).length, 0); for (let index = 0; index < attachments.length; index += 1) { assert.include(text, `/attachments/window-${index}.png`); diff --git a/packages/provider-core/src/server/attachmentPrompt.ts b/packages/provider-core/src/server/attachmentPrompt.ts index c7c5e58a5a662..b349701c90ef6 100644 --- a/packages/provider-core/src/server/attachmentPrompt.ts +++ b/packages/provider-core/src/server/attachmentPrompt.ts @@ -162,21 +162,29 @@ export function providerMessageTextWithAttachmentPaths(input: { readonly resolveAttachmentPath: (attachment: ChatAttachment) => string | null; }): string { let text = input.text; - const appendContext = (context: string | undefined) => { - if (context === undefined) return; - const candidate = text ? `${text}\n\n${context}` : context; - if (candidate.length <= PROVIDER_SEND_TURN_MAX_INPUT_CHARS) text = candidate; - }; - + // The text-input budget must never hide the existence of an attachment. for (const attachment of input.attachments) { const path = input.resolveAttachmentPath(attachment); - appendContext( + const context = path === null - ? undefined - : `[Attached ${attachment.type} "${attachment.name}" is saved at: ${path}]`, - ); + ? `[Attached ${attachment.type} "${attachment.name}" is unavailable: could not resolve a saved path for attachment '${attachment.id}'.]` + : `[Attached ${attachment.type} "${attachment.name}" is saved at: ${path}]`; + text = text ? `${text}\n\n${context}` : context; } + const truncationNotice = + "[Optional captured-window context was truncated to fit the prompt budget. Attachments are listed above.]"; + let truncated = false; + const appendContext = (context: string | undefined) => { + if (context === undefined) return; + const candidate = text ? `${text}\n\n${context}` : context; + if (candidate.length + 2 + truncationNotice.length <= PROVIDER_SEND_TURN_MAX_INPUT_CHARS) { + text = candidate; + } else { + truncated = true; + } + }; + for (const attachment of input.attachments) { const source = attachment.type === "image" ? (attachment as ChatImageAttachment).source : undefined; @@ -213,5 +221,5 @@ export function providerMessageTextWithAttachmentPaths(input: { ); } - return text; + return truncated ? `${text}\n\n${truncationNotice}` : text; } diff --git a/packages/provider-core/src/server/maintenance.ts b/packages/provider-core/src/server/maintenance.ts index 298fa5a4e4431..84aa084e7487e 100644 --- a/packages/provider-core/src/server/maintenance.ts +++ b/packages/provider-core/src/server/maintenance.ts @@ -1,5 +1,6 @@ import type { ProviderDriverKind } from "@t3tools/contracts"; import type * as Effect from "effect/Effect"; +import type * as HttpClient from "effect/http/HttpClient"; export interface ProviderMaintenanceCapabilities { readonly provider: ProviderDriverKind; @@ -15,7 +16,9 @@ export interface ProviderMaintenanceCapabilities { * Deferred until update checks are enabled; `null` falls back to * `latestVersion`, then npm. */ - readonly lookupLatestVersion?: Effect.Effect; + readonly lookupLatestVersion?: Effect.Effect; + /** Separates version lookups whose installer reads different per-instance settings. */ + readonly lookupLatestVersionCacheScope?: string; readonly compareVersions?: (current: string, latest: string) => number; /** * The executable T3 Code runs, symlinks followed, when no update command is diff --git a/packages/provider-core/src/server/maintenanceResolver.test.ts b/packages/provider-core/src/server/maintenanceResolver.test.ts index 9afedc66c5a73..9873711ba6ee9 100644 --- a/packages/provider-core/src/server/maintenanceResolver.test.ts +++ b/packages/provider-core/src/server/maintenanceResolver.test.ts @@ -13,6 +13,7 @@ import * as Fiber from "effect/Fiber"; import * as Sink from "effect/Sink"; import * as Stream from "effect/Stream"; import * as HttpClient from "effect/http/HttpClient"; +import * as HttpClientResponse from "effect/http/HttpClientResponse"; import * as ChildProcessSpawner from "effect/process/ChildProcessSpawner"; import * as ProviderLatestVersions from "./ProviderLatestVersions.ts"; import { @@ -100,13 +101,22 @@ const noSpawn = ChildProcessSpawner.make(() => ); function stdoutSpawner( - onSpawn: (command: string, args: ReadonlyArray) => string, + onSpawn: ( + command: string, + args: ReadonlyArray, + env: NodeJS.ProcessEnv | undefined, + ) => string, exitCodeOf: (args: ReadonlyArray) => number = () => 0, ) { return ChildProcessSpawner.make((command) => { - const { command: executable, args } = command as unknown as { + const { + command: executable, + args, + options, + } = command as unknown as { readonly command: string; readonly args: ReadonlyArray; + readonly options: { readonly env?: NodeJS.ProcessEnv }; }; return Effect.succeed( ChildProcessSpawner.makeHandle({ @@ -116,7 +126,7 @@ function stdoutSpawner( kill: () => Effect.void, unref: Effect.succeed(Effect.void), stdin: Sink.drain, - stdout: Stream.encodeText(Stream.make(onSpawn(executable, args))), + stdout: Stream.encodeText(Stream.make(onSpawn(executable, args, options.env))), stderr: Stream.empty, all: Stream.empty, getInputFd: () => Sink.drain, @@ -127,6 +137,116 @@ function stdoutSpawner( } it.layer(NodeServices.layer)("providerMaintenance", (it) => { + it.effect.each([ + { channel: "stable", output: "Auto-update channel: stable\n", version: "1.0.0", status: 200 }, + { channel: "latest", output: "Auto-update channel: latest\n", version: "1.2.0", status: 200 }, + { channel: null, output: "Unexpected doctor output\n", version: null, status: 200 }, + { channel: "stable", output: "Auto-update channel: stable\n", version: null, status: 503 }, + { channel: "stable", output: "Auto-update channel: stable\n", version: null, status: 200 }, + ])("checks Claude's native $channel release channel (HTTP $status)", (fixture) => + Effect.gen(function* () { + const capabilities = yield* resolvePackageManagedProviderMaintenance( + { + provider: driver("claudeAgent"), + npmPackageName: "@anthropic-ai/claude-code", + nativeUpdate: { args: ["update"], isCommandPath: () => true }, + }, + { + binaryPath: "claude", + resolvedCommandPath: "/synthetic/.local/bin/claude", + realCommandPath: "/synthetic/.local/share/claude/versions/1.0.0", + env: { + CLAUDE_CONFIG_DIR: "/synthetic/claude-instance", + ANTHROPIC_API_KEY: "instance-secret", + }, + platform: "darwin", + }, + ).pipe( + Effect.provideService( + ChildProcessSpawner.ChildProcessSpawner, + stdoutSpawner((_command, args, env) => { + expect(args).toEqual(["doctor"]); + expect(env).toEqual({ CLAUDE_CONFIG_DIR: "/synthetic/claude-instance" }); + return fixture.output; + }), + ), + ); + const version = yield* resolveLatestProviderVersion(capabilities).pipe( + Effect.provideServiceEffect( + ProviderLatestVersions.ProviderLatestVersions, + ProviderLatestVersions.make([["@anthropic-ai/claude-code", "9.9.9"]]), + ), + Effect.provideService( + HttpClient.HttpClient, + HttpClient.make((request) => { + expect(fixture.channel).not.toBeNull(); + expect(request.url).toBe( + `https://downloads.claude.ai/claude-code-releases/${fixture.channel}`, + ); + return Effect.succeed( + HttpClientResponse.fromWeb( + request, + new Response(fixture.version ?? "unavailable", { status: fixture.status }), + ), + ); + }), + ), + ); + expect(version).toBe(fixture.version); + expect(capabilities.update?.args).toEqual(["update"]); + expect(capabilities.update?.env).toEqual({ + CLAUDE_CONFIG_DIR: "/synthetic/claude-instance", + }); + expect(makeTargetedProviderUpdateAction(capabilities, "2.0.0")).toBeNull(); + }), + ); + + it.effect("scopes Claude native version lookups to the instance config directory", () => + Effect.gen(function* () { + const definition = { + provider: driver("claudeAgent"), + npmPackageName: "@anthropic-ai/claude-code", + nativeUpdate: { args: ["update"], isCommandPath: () => true }, + } satisfies PackageManagedProviderMaintenanceDefinition; + const resolve = (configDir: string) => + resolvePackageManagedProviderMaintenance(definition, { + binaryPath: "claude", + resolvedCommandPath: "/synthetic/.local/bin/claude", + realCommandPath: "/synthetic/.local/share/claude/versions/1.0.0", + env: { CLAUDE_CONFIG_DIR: configDir }, + platform: "darwin", + }).pipe( + Effect.provideService( + ChildProcessSpawner.ChildProcessSpawner, + stdoutSpawner((_command, _args, env) => + env?.CLAUDE_CONFIG_DIR === "/synthetic/stable" + ? "Auto-update channel: stable\n" + : "Auto-update channel: latest\n", + ), + ), + ); + const [stable, latest] = yield* Effect.all([ + resolve("/synthetic/stable"), + resolve("/synthetic/latest"), + ]); + const versions = yield* ProviderLatestVersions.make(); + const httpClient = HttpClient.make((request) => { + const version = request.url.endsWith("/stable") ? "2.1.308" : "2.1.311"; + return Effect.succeed(HttpClientResponse.fromWeb(request, new Response(version))); + }); + const stableVersion = yield* resolveLatestProviderVersion(stable).pipe( + Effect.provideService(ProviderLatestVersions.ProviderLatestVersions, versions), + Effect.provideService(HttpClient.HttpClient, httpClient), + ); + const latestVersion = yield* resolveLatestProviderVersion(latest).pipe( + Effect.provideService(ProviderLatestVersions.ProviderLatestVersions, versions), + Effect.provideService(HttpClient.HttpClient, httpClient), + ); + expect(stableVersion).toBe("2.1.308"); + expect(latestVersion).toBe("2.1.311"); + }), + ); + it.effect("prefers the installer's own latest version over the npm registry", () => resolveLatestProviderVersion({ ...manualPackageTool, latestVersion: "1.2.0" }).pipe( Effect.provide(ProviderLatestVersions.layer), diff --git a/packages/provider-core/src/server/maintenanceResolver.ts b/packages/provider-core/src/server/maintenanceResolver.ts index 94889b4a0f13f..dba3b88494336 100644 --- a/packages/provider-core/src/server/maintenanceResolver.ts +++ b/packages/provider-core/src/server/maintenanceResolver.ts @@ -27,6 +27,7 @@ import * as Schema from "effect/Schema"; import * as Stream from "effect/Stream"; import * as HttpClient from "effect/http/HttpClient"; import * as HttpClientRequest from "effect/http/HttpClientRequest"; +import * as HttpClientResponse from "effect/http/HttpClientResponse"; import * as ChildProcess from "effect/process/ChildProcess"; import * as ChildProcessSpawner from "effect/process/ChildProcessSpawner"; @@ -123,7 +124,8 @@ export function makeProviderMaintenanceCapabilities(input: { readonly platform?: NodeJS.Platform; readonly env?: NodeJS.ProcessEnv; readonly latestVersion?: string | null; - readonly lookupLatestVersion?: Effect.Effect; + readonly lookupLatestVersion?: Effect.Effect; + readonly lookupLatestVersionCacheScope?: string; }): ProviderMaintenanceCapabilities { const platform = input.platform ?? HostProcess.Platform.defaultValue(); const update = @@ -147,6 +149,9 @@ export function makeProviderMaintenanceCapabilities(input: { update, ...("latestVersion" in input ? { latestVersion: input.latestVersion } : {}), ...(input.lookupLatestVersion ? { lookupLatestVersion: input.lookupLatestVersion } : {}), + ...(input.lookupLatestVersionCacheScope + ? { lookupLatestVersionCacheScope: input.lookupLatestVersionCacheScope } + : {}), }; } @@ -376,11 +381,31 @@ const runProbe = Effect.fn("runProbe")(function* ( ); }); +const resolveClaudeNativeLatestVersion = Effect.fn("resolveClaudeNativeLatestVersion")(function* ( + command: string, + env: NodeJS.ProcessEnv, +) { + // Doctor and update resolve the same effective settings, including managed channel policy. + const diagnostic = yield* runProbe(command, ["doctor"], env); + const channel = diagnostic?.match(/^\s*Auto-update channel:\s*(stable|latest)\s*$/m)?.[1]; + if (!channel) return null; + const client = yield* HttpClient.HttpClient; + return yield* client.get(`https://downloads.claude.ai/claude-code-releases/${channel}`).pipe( + Effect.flatMap(HttpClientResponse.filterStatusOk), + Effect.flatMap((response) => response.text), + Effect.map((text) => (/^\d+\.\d+\.\d+$/.test(text.trim()) ? text.trim() : null)), + Effect.timeoutOption(LATEST_VERSION_TIMEOUT_MS), + Effect.map(Option.getOrNull), + Effect.orElseSucceed(() => null), + ); +}); + const decodePnpmOutdated = Schema.decodeUnknownOption( Schema.fromJsonString( Schema.Record(Schema.String, Schema.Struct({ latest: Schema.optional(Schema.String) })), ), ); + const decodePnpmList = Schema.decodeUnknownOption( Schema.fromJsonString( Schema.Array( @@ -454,6 +479,16 @@ export const resolvePackageManagedProviderMaintenance = Effect.fn( } const nativeUpdate = definition.nativeUpdate; + const services = yield* Effect.context(); + const claudeConfigDir = + definition.provider === "claudeAgent" ? context.env.CLAUDE_CONFIG_DIR?.trim() : undefined; + const nativeEnv = + nativeUpdate && (nativeUpdate.env || claudeConfigDir) + ? { + ...nativeUpdate.env, + ...(claudeConfigDir ? { CLAUDE_CONFIG_DIR: claudeConfigDir } : {}), + } + : nativeUpdate?.env; const native = nativeUpdate ? makeProviderMaintenanceCapabilities({ provider: definition.provider, @@ -462,7 +497,18 @@ export const resolvePackageManagedProviderMaintenance = Effect.fn( updateArgs: nativeUpdate.args, updateLockKey: `${definition.provider}-native`, platform: context.platform, - ...(nativeUpdate.env ? { env: nativeUpdate.env } : {}), + ...(nativeEnv ? { env: nativeEnv } : {}), + ...(definition.provider === "claudeAgent" + ? { + // A failed channel probe cannot authorize an npm-latest advisory. + latestVersion: null, + lookupLatestVersionCacheScope: claudeConfigDir ?? "", + lookupLatestVersion: resolveClaudeNativeLatestVersion( + context.resolvedCommandPath, + nativeEnv ?? {}, + ).pipe(Effect.provideContext(services)), + } + : {}), }) : manual; // A native-looking launcher can still delegate to a mise-owned install. @@ -876,7 +922,12 @@ const fetchNpmLatestVersion = Effect.fn("fetchNpmLatestVersion")(function* (pack function installerLatestVersionKey(capabilities: ProviderMaintenanceCapabilities) { const update = capabilities.update; return capabilities.lookupLatestVersion && update - ? [update.lockKey, update.executable, ...update.args].join("\0") + ? [ + update.lockKey, + update.executable, + ...update.args, + capabilities.lookupLatestVersionCacheScope ?? "", + ].join("\0") : null; } diff --git a/packages/provider-pi/src/server/adapter.test.ts b/packages/provider-pi/src/server/adapter.test.ts index 9c81bab0abb53..1dcd48dbc808c 100644 --- a/packages/provider-pi/src/server/adapter.test.ts +++ b/packages/provider-pi/src/server/adapter.test.ts @@ -2,6 +2,7 @@ import { assert, describe, it } from "@effect/vitest"; import * as NodeServices from "@effect/platform-node/NodeServices"; import { CheckpointId, + ChatAttachmentId, EnvironmentId, NodeId, ProviderInstanceId, @@ -23,6 +24,7 @@ import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; import * as Exit from "effect/Exit"; import * as Fiber from "effect/Fiber"; +import * as FileSystem from "effect/FileSystem"; import * as Layer from "effect/Layer"; import * as Queue from "effect/Queue"; import * as Schema from "effect/Schema"; @@ -36,6 +38,7 @@ import * as ProviderContinuationRequests from "@t3tools/provider-core/server/Pro import * as McpProviderSessions from "@t3tools/provider-core/server/McpProviderSessions"; import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator"; import * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter"; +import * as ProviderHost from "@t3tools/provider-core/server/ProviderHost"; import { handoffBudget } from "@t3tools/provider-core/server/handoffBudget"; import * as HostProcess from "@t3tools/shared/HostProcess"; import { toolOutputImages, compactDynamicToolOutput } from "@t3tools/shared/toolOutput"; @@ -519,6 +522,92 @@ const startTurn = Effect.fnUntraced(function* ( }); describe("PiAdapterV2", () => { + it.effect("keeps native image bytes and the saved path in the prompt", () => + Effect.gen(function* () { + const host = yield* ProviderHost.ProviderHost; + const fileSystem = yield* FileSystem.FileSystem; + const attachment: ChatAttachment = { + id: ChatAttachmentId.make("image"), + type: "image", + name: "image.png", + mimeType: "image/png", + sizeBytes: 3, + }; + const path = host.resolveAttachmentPath(attachment)!; + yield* fileSystem.writeFile(path, new Uint8Array([1, 2, 3])); + const fake = yield* makeFakePi; + const { runtime } = yield* openRuntime(fake); + const providerThread = yield* runtime.ensureThread({ + threadId: THREAD_ID, + modelSelection: modelSelection("default"), + runtimePolicy, + }); + yield* startTurn(runtime, providerThread, "default", [attachment]); + const prompt = yield* fake.takeRequest("prompt"); + assert.include(String(prompt["message"]), path); + assert.deepEqual(prompt["images"], [{ type: "image", data: "AQID", mimeType: "image/png" }]); + }).pipe(Effect.scoped, Effect.provide(layerTest)), + ); + + it.effect("lists unavailable attachments alongside resolvable files in the native prompt", () => + Effect.gen(function* () { + const host = yield* ProviderHost.ProviderHost; + const fake = yield* makeFakePi; + const { runtime } = yield* openRuntime(fake).pipe( + Effect.provideService(ProviderHost.ProviderHost, { + ...host, + resolveAttachmentPath: (attachment) => + attachment.id === "available" ? host.resolveAttachmentPath(attachment) : null, + }), + ); + const providerThread = yield* runtime.ensureThread({ + threadId: THREAD_ID, + modelSelection: modelSelection("default"), + runtimePolicy, + }); + const attachments: ReadonlyArray = [ + { + id: ChatAttachmentId.make("available"), + type: "file", + name: "notes.txt", + mimeType: "text/plain", + sizeBytes: 1, + }, + { + id: ChatAttachmentId.make("missing-file"), + type: "file", + name: "missing.txt", + mimeType: "text/plain", + sizeBytes: 1, + }, + { + id: ChatAttachmentId.make("missing-image"), + type: "image", + name: "missing.png", + mimeType: "image/png", + sizeBytes: 1, + }, + { + id: ChatAttachmentId.make("missing-future"), + type: "future", + name: "future.bin", + mimeType: "application/octet-stream", + sizeBytes: 1, + }, + ]; + yield* startTurn(runtime, providerThread, "default", attachments); + const prompt = yield* fake.takeRequest("prompt"); + assert.isString(prompt["message"]); + const message = String(prompt["message"]); + assert.include(message, host.resolveAttachmentPath(attachments[0]!)!); + for (const attachment of attachments.slice(1)) { + assert.include(message, attachment.name); + assert.include(message, attachment.id); + } + assert.equal((message.match(/unavailable/g) ?? []).length, 3); + }).pipe(Effect.scoped, Effect.provide(layerTest)), + ); + it.effect.each([false, true])( "completes a resumed user turn before native wake events arrive, Pi streaming=%s", (streaming) => diff --git a/packages/provider-pi/src/server/adapter.ts b/packages/provider-pi/src/server/adapter.ts index a2a3caf9267e5..6567b91676d27 100644 --- a/packages/provider-pi/src/server/adapter.ts +++ b/packages/provider-pi/src/server/adapter.ts @@ -40,6 +40,7 @@ import * as Stream from "effect/Stream"; import * as ChildProcessSpawner from "effect/process/ChildProcessSpawner"; import * as ProviderHost from "@t3tools/provider-core/server/ProviderHost"; +import { providerMessageTextWithAttachmentPaths } from "@t3tools/provider-core/server/attachmentPrompt"; import { mcpToolPresentation } from "@t3tools/provider-core/server/mcpToolPresentation"; import * as McpProviderSessions from "@t3tools/provider-core/server/McpProviderSessions"; import { parsePiCompactCommand, type PiCompactCommand } from "./commands.ts"; @@ -2335,7 +2336,11 @@ export const makePiAdapterV2 = Effect.fn("makePiAdapterV2")(function* ( attachments: ReadonlyArray, ) { const images: Array<{ type: "image"; data: string; mimeType: string }> = []; - const extraLines: Array = []; + const message = providerMessageTextWithAttachmentPaths({ + text, + attachments, + resolveAttachmentPath: host.resolveAttachmentPath, + }); for (const attachment of attachments) { const path = host.resolveAttachmentPath(attachment); if (path === null) continue; @@ -2346,11 +2351,8 @@ export const makePiAdapterV2 = Effect.fn("makePiAdapterV2")(function* ( data: Buffer.from(bytes).toString("base64"), mimeType: attachment.mimeType, }); - } else { - extraLines.push(`[Attachment saved at ${path}]`); } } - const message = extraLines.length === 0 ? text : `${text}\n\n${extraLines.join("\n")}`; return { message, images }; }); From d3b21c6a6199909d64060e3a49ad4dab7548c0f9 Mon Sep 17 00:00:00 2001 From: unlockdigital Date: Tue, 27 Oct 2026 08:27:44 -0500 Subject: [PATCH 2/3] fix(ci): run fingerprint checks on available fork runners The fork has no Blacksmith runner, so its advisory native fingerprint job stays queued. Select GitHub-hosted Ubuntu for forks using the runner pattern already used by the main CI workflow. Validated workflow formatting, affected checks, and all32package typechecks. Model: GPT-6.1-sol. Harness: Codex in T3 Code. --- .github/workflows/mobile-fingerprint-check.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/mobile-fingerprint-check.yml b/.github/workflows/mobile-fingerprint-check.yml index b8b872f019f6e..336f3c3739648 100644 --- a/.github/workflows/mobile-fingerprint-check.yml +++ b/.github/workflows/mobile-fingerprint-check.yml @@ -24,7 +24,7 @@ concurrency: jobs: fingerprint: name: Native fingerprint diff - runs-on: blacksmith-8vcpu-ubuntu-2404 + runs-on: ${{ github.repository == 'pingdotgg/t3code' && 'blacksmith-8vcpu-ubuntu-2404' || 'ubuntu-24.04' }} permissions: contents: read issues: write From 900e04c9022bdae333d013f325c1b744f87d1a67 Mon Sep 17 00:00:00 2001 From: unlockdigital Date: Tue, 27 Oct 2026 08:41:34 -0500 Subject: [PATCH 3/3] fix(server): retain worktrees with invalid Git pointers Git can report a broken linked-worktree pointer with an alternate diagnostic. Recognize that diagnostic as a changed repository and retain the checkout, while preserving failure reports for unknown Git errors. Reproduced the CI failure with a deterministic regression. All 121 cleanup tests, affected checks, and independent review pass. Model: GPT-6.1-sol with GPT-6-luna review. Harness: Codex in T3 Code. --- apps/server/src/storageCleanup.test.ts | 49 ++++++++++++++++++++++++++ apps/server/src/storageCleanup.ts | 7 +++- 2 files changed, 55 insertions(+), 1 deletion(-) diff --git a/apps/server/src/storageCleanup.test.ts b/apps/server/src/storageCleanup.test.ts index 992e219db9122..6460f6bff1c17 100644 --- a/apps/server/src/storageCleanup.test.ts +++ b/apps/server/src/storageCleanup.test.ts @@ -9,6 +9,7 @@ import * as Option from "effect/Option"; import type * as Path from "effect/Path"; import type * as Scope from "effect/Scope"; import * as Stream from "effect/Stream"; +import * as ChildProcessSpawner from "effect/process/ChildProcessSpawner"; import { SCAN_CACHE_FILE_NAME } from "./usage/usageScanCache.ts"; import { TestClock } from "effect/testing"; import * as SqlClient from "effect/sql/SqlClient"; @@ -1434,6 +1435,54 @@ describe("storage cleanup reports and local file policies", () => { }), ), ); + it.live.each([ + { + stderr: "fatal: not a git repository: /missing-git-directory\n", + outcome: "kept", + reason: "Repository or branch changed", + }, + { + stderr: "fatal: gitfile does not point to a valid repository: /worktree/.git\n", + outcome: "kept", + reason: "Repository or branch changed", + }, + { + stderr: "fatal: permission denied\n", + outcome: "failed", + reason: "git: fatal: permission denied", + }, + ])("preserves the worktree for a Git repository probe error: $stderr", (fixture) => + runCleanupTest( + Effect.gen(function* () { + const git = yield* GitVcsDriver.GitVcsDriver; + const { service, fs, worktree, fullStatusCalls } = yield* cleanupFixture().pipe( + Effect.provideService(GitVcsDriver.GitVcsDriver, { + ...git, + execute: (input) => + input.operation === "StorageCleanup.branchChanged" + ? Effect.succeed({ + exitCode: ChildProcessSpawner.ExitCode(128), + stdout: "", + stderr: fixture.stderr, + stdoutTruncated: false, + stderrTruncated: false, + }) + : git.execute(input), + }), + ); + const report = yield* service.runNow; + expect(report.entries[0]).toMatchObject({ + outcome: fixture.outcome, + reason: fixture.reason, + bytes: null, + }); + expect(report.bytesFreed).toBe(0); + expect(yield* fs.readFileString(`${worktree}/tracked.txt`)).toBe("original\n"); + expect(fullStatusCalls()).toBe(0); + }), + ), + ); + it.live("compares the checked-out branch without a full status", () => runCleanupTest( Effect.gen(function* () { diff --git a/apps/server/src/storageCleanup.ts b/apps/server/src/storageCleanup.ts index a196321e38098..5cbe6b4de60b9 100644 --- a/apps/server/src/storageCleanup.ts +++ b/apps/server/src/storageCleanup.ts @@ -514,7 +514,12 @@ const make = Effect.gen(function* () { ); if (result === null) return null; if (result.exitCode === 0 || result.exitCode === allowedExitCode) return result; - if (result.stderr.toLowerCase().includes("not a git repository")) return null; + const stderr = result.stderr.toLowerCase(); + if ( + stderr.includes("not a git repository") || + stderr.includes("gitfile does not point to a valid repository") + ) + return null; return yield* failure( result.stderr.trim().split(/\r?\n/)[0]?.slice(0, 1000) || "command failed", );