From d82e8f31053f2b74d12e5877123726a03b277598 Mon Sep 17 00:00:00 2001 From: SAPHID Date: Sun, 11 Oct 2026 23:32:40 +0000 Subject: [PATCH 1/8] fix(server): subscribe before loading the connection access snapshot Acquire both credential change subscriptions before reading the snapshot so concurrent connection updates survive the snapshot ACK wait. Release the subscriptions with the RPC request scope. Co-Authored-By: GPT-6.1 Sol --- apps/server/src/auth/PairingGrantStore.ts | 7 + apps/server/src/auth/SessionStore.ts | 7 + apps/server/src/ws.authAccess.test.ts | 291 ++++++++++++++++++++++ apps/server/src/ws.ts | 90 ++++--- 4 files changed, 349 insertions(+), 46 deletions(-) create mode 100644 apps/server/src/ws.authAccess.test.ts diff --git a/apps/server/src/auth/PairingGrantStore.ts b/apps/server/src/auth/PairingGrantStore.ts index d3f5bf7ef8e5e..3b500472c71aa 100644 --- a/apps/server/src/auth/PairingGrantStore.ts +++ b/apps/server/src/auth/PairingGrantStore.ts @@ -19,6 +19,7 @@ import * as Option from "effect/Option"; import * as PubSub from "effect/PubSub"; import * as Ref from "effect/Ref"; import * as Schema from "effect/Schema"; +import * as Scope from "effect/Scope"; import * as Stream from "effect/Stream"; import * as ServerConfig from "../config.ts"; @@ -225,6 +226,11 @@ export class PairingGrantStore extends Context.Service< BootstrapCredentialInternalError >; readonly streamChanges: Stream.Stream; + readonly subscribeChanges: Effect.Effect< + PubSub.Subscription, + never, + Scope.Scope + >; readonly revoke: (id: string) => Effect.Effect; readonly consume: ( credential: string, @@ -628,6 +634,7 @@ export const make = Effect.gen(function* () { get streamChanges() { return Stream.fromPubSub(changesPubSub); }, + subscribeChanges: PubSub.subscribe(changesPubSub), revoke, consume, }); diff --git a/apps/server/src/auth/SessionStore.ts b/apps/server/src/auth/SessionStore.ts index 9570785d9f93d..7cd9bdfdfe5ac 100644 --- a/apps/server/src/auth/SessionStore.ts +++ b/apps/server/src/auth/SessionStore.ts @@ -19,6 +19,7 @@ import * as Layer from "effect/Layer"; import * as PubSub from "effect/PubSub"; import * as Ref from "effect/Ref"; import * as Schema from "effect/Schema"; +import * as Scope from "effect/Scope"; import * as Stream from "effect/Stream"; import * as Option from "effect/Option"; @@ -407,6 +408,11 @@ export class SessionStore extends Context.Service< SessionCredentialInternalError >; readonly streamChanges: Stream.Stream; + readonly subscribeChanges: Effect.Effect< + PubSub.Subscription, + never, + Scope.Scope + >; readonly awaitInvalidation: ( sessionId: AuthSessionId, ) => Effect.Effect; @@ -1091,6 +1097,7 @@ export const make = Effect.gen(function* () { get streamChanges() { return Stream.fromPubSub(changesPubSub); }, + subscribeChanges: PubSub.subscribe(changesPubSub), revoke, revokeAllExcept, markConnected, diff --git a/apps/server/src/ws.authAccess.test.ts b/apps/server/src/ws.authAccess.test.ts new file mode 100644 index 0000000000000..127122bc9bdc0 --- /dev/null +++ b/apps/server/src/ws.authAccess.test.ts @@ -0,0 +1,291 @@ +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { assert, it } from "@effect/vitest"; +import { + AuthAccessStreamEvent, + AuthAccessStreamError, + AuthAdministrativeScopes, + WS_METHODS, + WsRpcGroup, +} from "@t3tools/contracts"; +import * as Deferred from "effect/Deferred"; +import * as Effect from "effect/Effect"; +import * as PubSub from "effect/PubSub"; +import * as Layer from "effect/Layer"; +import * as Queue from "effect/Queue"; +import * as Schema from "effect/Schema"; +import * as Stream from "effect/Stream"; +import * as TestClock from "effect/testing/TestClock"; +import { RpcMessage, RpcSerialization, RpcServer } from "effect/rpc"; + +import * as EnvironmentAuth from "./auth/EnvironmentAuth.ts"; +import * as PairingGrantStore from "./auth/PairingGrantStore.ts"; +import * as RpcAuthorization from "./auth/RpcAuthorization.ts"; +import { RPC_REQUIRED_SCOPES } from "./auth/RpcAuthorization.ts"; +import * as SessionStore from "./auth/SessionStore.ts"; +import * as ServerSecretStore from "./auth/ServerSecretStore.ts"; +import * as ServerConfig from "./config.ts"; +import * as ServerEnvironment from "./environment/ServerEnvironment.ts"; +import * as SqlitePersistence from "./persistence/Sqlite.ts"; +import { subscribeAuthAccess, WS_RPC_SERVER_OPTIONS } from "./ws.ts"; + +const group = WsRpcGroup.omit( + ...[...WsRpcGroup.requests.keys()].filter( + ( + tag, + ): tag is Exclude => + tag !== WS_METHODS.subscribeAuthAccess, + ), +); + +const layerAuth = EnvironmentAuth.layer.pipe( + Layer.provideMerge(ServerSecretStore.layer), + Layer.provideMerge(SqlitePersistence.layerMemory), + Layer.provide(ServerEnvironment.layerIdentity), + Layer.provide(ServerConfig.layerTest(process.cwd(), { prefix: "t3-auth-access-stream-test-" })), +); + +const makeProtocol = Effect.gen(function* () { + const responses = yield* Queue.unbounded(); + const receive = yield* Deferred.make[0]>(); + const protocol = yield* RpcServer.Protocol.make((write) => + Effect.gen(function* () { + yield* Deferred.succeed(receive, write); + const serialization = yield* RpcSerialization.RpcSerialization; + return { + disconnects: yield* Queue.unbounded(), + send: (_clientId, response) => Queue.offer(responses, response), + end: () => Effect.void, + clientIds: Effect.succeed(new Set([0])), + initialMessage: Effect.succeedNone, + supportsAck: true, + supportsTransferables: false, + supportsSpanPropagation: false, + supportsNotifications: true, + codecFor: serialization.codecFor, + }; + }), + ); + return { responses, receive, protocol }; +}); + +const startRpc = Effect.fn(function* ( + stream: Stream.Stream, + { responses, receive, protocol }: Effect.Success, +) { + yield* RpcServer.make(group, WS_RPC_SERVER_OPTIONS).pipe( + Effect.provide( + Layer.mergeAll( + group.toLayerHandler(WS_METHODS.subscribeAuthAccess, () => stream), + RpcAuthorization.layer(AuthAdministrativeScopes), + ), + ), + Effect.provideService(RpcServer.Protocol, protocol), + Effect.forkScoped, + ); + return { write: yield* Deferred.await(receive), responses }; +}); + +const decodeEvents = Schema.decodeUnknownSync( + Schema.toCodecJson(Schema.Array(AuthAccessStreamEvent)), +); +const takeEvents = (responses: Queue.Queue) => + Queue.take(responses).pipe( + Effect.map((response) => { + assert.equal(response._tag, "Chunk"); + return response._tag === "Chunk" ? decodeEvents(response.values) : []; + }), + ); + +it.layer(NodeServices.layer)("subscribeAuthAccess snapshot handoff", (it) => { + it.effect.each([ + { change: "revoke link", milestone: "snapshot read" }, + { change: "revoke link", milestone: "snapshot ACK" }, + { change: "connect client", milestone: "snapshot read" }, + { change: "connect client", milestone: "snapshot ACK" }, + { change: "disconnect client", milestone: "snapshot read" }, + { change: "disconnect client", milestone: "snapshot ACK" }, + ] as const)("buffers $change while waiting for $milestone", ({ change, milestone }) => + Effect.gen(function* () { + const auth = yield* EnvironmentAuth.EnvironmentAuth; + const sessions = yield* SessionStore.SessionStore; + const current = yield* sessions.issue({ subject: "current-client" }); + const link = yield* auth.createPairingLink({ label: "Revoked during subscription" }); + if (change === "disconnect client") yield* sessions.markConnected(current.sessionId); + + const snapshotRead = yield* Deferred.make(); + const releaseSnapshot = yield* Deferred.make(); + const { responses, receive, protocol } = yield* makeProtocol; + const snapshotAuth = EnvironmentAuth.EnvironmentAuth.of({ + ...auth, + listClientSessions: (sessionId) => + auth.listClientSessions(sessionId).pipe( + Effect.tap(() => Deferred.succeed(snapshotRead, undefined)), + Effect.flatMap((snapshot) => Deferred.await(releaseSnapshot).pipe(Effect.as(snapshot))), + ), + }); + const { write } = yield* startRpc( + Stream.unwrap( + subscribeAuthAccess(current.sessionId).pipe( + Effect.provideService(EnvironmentAuth.EnvironmentAuth, snapshotAuth), + ), + ), + { responses, receive, protocol }, + ); + yield* write(0, { + _tag: "Request", + id: "1", + tag: WS_METHODS.subscribeAuthAccess, + payload: {}, + headers: [], + }); + yield* Deferred.await(snapshotRead).pipe( + Effect.raceFirst( + Queue.take(responses).pipe(Effect.flatMap((response) => Effect.die(response))), + ), + ); + const mutate = + change === "revoke link" + ? auth.revokePairingLink(link.id) + : change === "connect client" + ? sessions.markConnected(current.sessionId) + : sessions.markDisconnected(current.sessionId); + if (milestone === "snapshot read") yield* mutate; + yield* Deferred.succeed(releaseSnapshot, undefined); + const [snapshot] = yield* takeEvents(responses); + assert.equal(snapshot?.type, "snapshot"); + if (snapshot?.type !== "snapshot") return; + assert.equal(snapshot.revision, 1); + assert.deepEqual( + snapshot.payload.pairingLinks.map((value) => value.id), + [link.id], + ); + assert.equal(snapshot.payload.clientSessions[0]?.connected, change === "disconnect client"); + if (milestone === "snapshot ACK") yield* mutate; + yield* write(0, { _tag: "Ack", requestId: "1" }); + // Drain runnable fibers to attach the old handler's live streams. This does + // not advance time. An unrelated update makes lost changes fail without a timeout. + yield* TestClock.adjust(0); + const unrelated = yield* auth.createPairingLink({ label: "Unrelated update" }); + const [next] = yield* takeEvents(responses); + assert.equal(next?.revision, 2); + if (change === "revoke link") { + assert.equal(next?.type, "pairingLinkRemoved"); + if (next?.type === "pairingLinkRemoved") { + assert.equal(next.payload.id, link.id); + const remaining = snapshot.payload.pairingLinks.filter( + (value) => value.id !== next.payload.id, + ); + assert.deepEqual( + remaining.map((value) => value.id), + [], + ); + } + } else { + assert.equal(next?.type, "clientUpserted"); + if (next?.type === "clientUpserted") { + assert.equal(next.payload.sessionId, current.sessionId); + assert.isTrue(next.payload.current); + assert.equal(next.payload.connected, change === "connect client"); + assert.deepEqual(next.payload, (yield* auth.listClientSessions(current.sessionId))[0]); + } + } + yield* write(0, { _tag: "Ack", requestId: "1" }); + const [following] = yield* takeEvents(responses); + assert.equal(following?.revision, 3); + assert.equal(following?.type, "pairingLinkUpserted"); + if (following?.type === "pairingLinkUpserted") { + assert.equal(following.payload.id, unrelated.id); + const remaining = snapshot.payload.pairingLinks.filter( + (value) => change !== "revoke link" || value.id !== link.id, + ); + assert.deepEqual( + [...remaining, following.payload].map((value) => value.id).toSorted(), + (yield* auth.listPairingLinks()).map((value) => value.id).toSorted(), + ); + } + yield* write(0, { _tag: "Interrupt", requestId: "1" }); + assert.equal((yield* Queue.take(responses))._tag, "Exit"); + }).pipe(Effect.scoped, Effect.provide(Layer.mergeAll(layerAuth, RpcSerialization.layerJson))), + ); + + it.effect.each(["snapshot read", "snapshot ACK", "live tail", "snapshot failure"] as const)( + "releases both subscriptions on cancellation or failure during %s", + (milestone) => + Effect.gen(function* () { + const auth = yield* EnvironmentAuth.EnvironmentAuth; + const pairingLinks = yield* PairingGrantStore.PairingGrantStore; + const sessions = yield* SessionStore.SessionStore; + const current = yield* sessions.issue({ subject: "current-client" }); + const acquired = yield* Queue.unbounded>(); + const snapshotRead = yield* Deferred.make(); + const releaseSnapshot = yield* Deferred.make(); + const snapshotAuth = EnvironmentAuth.EnvironmentAuth.of({ + ...auth, + listClientSessions: (sessionId) => + auth.listClientSessions(sessionId).pipe( + Effect.tap(() => Deferred.succeed(snapshotRead, undefined)), + Effect.flatMap((snapshot) => + Deferred.await(releaseSnapshot).pipe( + Effect.andThen( + milestone === "snapshot failure" + ? Effect.fail( + new EnvironmentAuth.ServerAuthSessionsListError({ cause: "test" }), + ) + : Effect.succeed(snapshot), + ), + ), + ), + ), + }); + const { responses, receive, protocol } = yield* makeProtocol; + const { write } = yield* startRpc( + Stream.unwrap( + subscribeAuthAccess(current.sessionId).pipe( + Effect.provideService(EnvironmentAuth.EnvironmentAuth, snapshotAuth), + Effect.provideService(PairingGrantStore.PairingGrantStore, { + ...pairingLinks, + subscribeChanges: pairingLinks.subscribeChanges.pipe( + Effect.tap((subscription) => Queue.offer(acquired, subscription)), + ), + }), + Effect.provideService(SessionStore.SessionStore, { + ...sessions, + subscribeChanges: sessions.subscribeChanges.pipe( + Effect.tap((subscription) => Queue.offer(acquired, subscription)), + ), + }), + ), + ), + { responses, receive, protocol }, + ); + yield* write(0, { + _tag: "Request", + id: "1", + tag: WS_METHODS.subscribeAuthAccess, + payload: {}, + headers: [], + }); + const linkSubscription = yield* Queue.take(acquired); + const clientSubscription = yield* Queue.take(acquired); + yield* Deferred.await(snapshotRead); + assert.isFalse(linkSubscription.shutdownFlag.current); + assert.isFalse(clientSubscription.shutdownFlag.current); + if (milestone !== "snapshot read") { + yield* Deferred.succeed(releaseSnapshot, undefined); + if (milestone !== "snapshot failure") yield* takeEvents(responses); + } + if (milestone === "live tail") { + yield* sessions.markConnected(current.sessionId); + yield* write(0, { _tag: "Ack", requestId: "1" }); + const [event] = yield* takeEvents(responses); + assert.equal(event?.type, "clientUpserted"); + } + if (milestone !== "snapshot failure") { + yield* write(0, { _tag: "Interrupt", requestId: "1" }); + } + assert.equal((yield* Queue.take(responses))._tag, "Exit"); + assert.isTrue(linkSubscription.shutdownFlag.current); + assert.isTrue(clientSubscription.shutdownFlag.current); + }).pipe(Effect.scoped, Effect.provide(Layer.mergeAll(layerAuth, RpcSerialization.layerJson))), + ); +}); diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 375c2ac78aa1f..26124c202f962 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -603,6 +603,49 @@ const MAX_CLIENT_APP_VERSION_LENGTH = 64; const MAX_CLIENT_BROWSER_LENGTH = 64; const MAX_CLIENT_DEVICE_MODEL_LENGTH = 80; +export const subscribeAuthAccess = Effect.fn("ws.subscribeAuthAccess")(function* ( + currentSessionId: AuthSessionId, +) { + const serverAuth = yield* EnvironmentAuth.EnvironmentAuth; + const bootstrapCredentials = yield* PairingGrantStore.PairingGrantStore; + const sessions = yield* SessionStore.SessionStore; + // Buffer both stores before reading; RPC backpressure can hold the snapshot + // until its ACK, delaying consumption of the live tail. + const pairingLinkChanges = yield* bootstrapCredentials.subscribeChanges; + const clientChanges = yield* sessions.subscribeChanges; + const initialSnapshot = yield* Effect.all({ + pairingLinks: serverAuth.listPairingLinks(), + clientSessions: serverAuth.listClientSessions(currentSessionId), + }).pipe(Effect.mapError((error) => new AuthAccessStreamError({ message: error.message }))); + const revisionRef = yield* Ref.make(1); + const accessChanges: Stream.Stream< + PairingGrantStore.BootstrapCredentialChange | SessionStore.SessionCredentialChange + > = Stream.merge( + Stream.fromSubscription(pairingLinkChanges), + Stream.fromSubscription(clientChanges), + ); + + const liveEvents: Stream.Stream = accessChanges.pipe( + Stream.mapEffect((change) => + Ref.updateAndGet(revisionRef, (revision) => revision + 1).pipe( + Effect.map((revision) => toAuthAccessStreamEvent(change, revision, currentSessionId)), + ), + ), + ); + + return Stream.concat( + rpcInitialItems([ + { + version: 1 as const, + revision: 1, + type: "snapshot" as const, + payload: initialSnapshot, + }, + ]), + liveEvents, + ); +}); + export function hasCompatibleOrchestrationProtocol(url: URL): boolean { return ( url.searchParams.get(ORCHESTRATION_PROTOCOL_QUERY_PARAM) === @@ -1321,8 +1364,6 @@ const layerWsRpc = ( const sourceControlRepositories = yield* SourceControlRepositoryService.SourceControlRepositoryService; const withPullRequestViewer = pullRequests.withRoutingCredential; - const bootstrapCredentials = yield* PairingGrantStore.PairingGrantStore; - const sessions = yield* SessionStore.SessionStore; const processDiagnostics = yield* ProcessDiagnostics.ProcessDiagnostics; const hostResources = yield* HostResources.HostResources; const processResourceMonitor = yield* ProcessResourceMonitor.ProcessResourceMonitor; @@ -1667,19 +1708,6 @@ const layerWsRpc = ( yield* providerRegistry.refreshInstance(input.instanceId); return { disabled: true } as const; }); - const loadAuthAccessSnapshot = () => - Effect.all({ - pairingLinks: serverAuth.listPairingLinks(), - clientSessions: serverAuth.listClientSessions(currentSessionId), - }).pipe( - Effect.mapError( - (error) => - new AuthAccessStreamError({ - message: error.message, - }), - ), - ); - const loadServerConfig = (options: { readonly usageLimitsCommand: boolean }) => Effect.gen(function* () { const keybindingsConfig = yield* keybindings.loadConfigState; @@ -3090,37 +3118,7 @@ const layerWsRpc = ( }), ), [WS_METHODS.subscribeAuthAccess]: (_input) => - Stream.unwrap( - Effect.gen(function* () { - const initialSnapshot = yield* loadAuthAccessSnapshot(); - const revisionRef = yield* Ref.make(1); - const accessChanges: Stream.Stream< - PairingGrantStore.BootstrapCredentialChange | SessionStore.SessionCredentialChange - > = Stream.merge(bootstrapCredentials.streamChanges, sessions.streamChanges); - - const liveEvents: Stream.Stream = accessChanges.pipe( - Stream.mapEffect((change) => - Ref.updateAndGet(revisionRef, (revision) => revision + 1).pipe( - Effect.map((revision) => - toAuthAccessStreamEvent(change, revision, currentSessionId), - ), - ), - ), - ); - - return Stream.concat( - rpcInitialItems([ - { - version: 1 as const, - revision: 1, - type: "snapshot" as const, - payload: initialSnapshot, - }, - ]), - liveEvents, - ); - }), - ), + Stream.unwrap(subscribeAuthAccess(currentSessionId)), [WS_METHODS.subscribeBackgroundPolicy]: (_input) => Stream.unwrap( Effect.map(backgroundPolicy.subscribe, ({ latest, changes }) => From 40a99a57feff42158d45fde9aadcd396c6b20dd8 Mon Sep 17 00:00:00 2001 From: SAPHID Date: Mon, 12 Oct 2026 00:12:26 +0000 Subject: [PATCH 2/8] refactor(server): move auth access streaming into the auth service Co-Authored-By: GPT-6.1 Sol --- apps/server/src/auth/EnvironmentAuth.ts | 91 +++++++++++++++++++++++ apps/server/src/ws.authAccess.test.ts | 93 +++++++++++++----------- apps/server/src/ws.ts | 97 ++----------------------- 3 files changed, 150 insertions(+), 131 deletions(-) diff --git a/apps/server/src/auth/EnvironmentAuth.ts b/apps/server/src/auth/EnvironmentAuth.ts index d66cdbb62c365..149769ece4117 100644 --- a/apps/server/src/auth/EnvironmentAuth.ts +++ b/apps/server/src/auth/EnvironmentAuth.ts @@ -1,4 +1,5 @@ import { + type AuthAccessStreamEvent, AuthAccessTokenType, AuthAccessWriteScope, AuthAdministrativeScopes, @@ -33,7 +34,10 @@ import * as Duration from "effect/Duration"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; +import * as Ref from "effect/Ref"; import * as Schema from "effect/Schema"; +import * as Scope from "effect/Scope"; +import * as Stream from "effect/Stream"; import * as HttpServerRequest from "effect/http/HttpServerRequest"; import * as ServerEnvironment from "../environment/ServerEnvironment.ts"; @@ -538,6 +542,15 @@ export class EnvironmentAuth extends Context.Service< readonly listClientSessions: ( currentSessionId: AuthSessionId, ) => Effect.Effect, ServerAuthInternalError>; + /** Acquires both live subscriptions before reading the request's snapshot. */ + readonly streamAccess: (currentSessionId: AuthSessionId) => Effect.Effect< + { + readonly snapshot: Extract; + readonly changes: Stream.Stream; + }, + ServerAuthInternalError, + Scope.Scope + >; readonly revokeClientSession: ( currentSessionId: AuthSessionId, targetSessionId: AuthSessionId, @@ -586,6 +599,47 @@ export class EnvironmentAuth extends Context.Service< } >()("t3/auth/EnvironmentAuth") {} +function toAuthAccessStreamEvent( + change: PairingGrantStore.BootstrapCredentialChange | SessionStore.SessionCredentialChange, + revision: number, + currentSessionId: AuthSessionId, +): AuthAccessStreamEvent { + switch (change.type) { + case "pairingLinkUpserted": + return { + version: 1, + revision, + type: "pairingLinkUpserted", + payload: { ...change.pairingLink, ...authScopeResponse(change.pairingLink.scopes) }, + }; + case "pairingLinkRemoved": + return { + version: 1, + revision, + type: "pairingLinkRemoved", + payload: { id: change.id }, + }; + case "clientUpserted": + return { + version: 1, + revision, + type: "clientUpserted", + payload: { + ...change.clientSession, + ...authScopeResponse(change.clientSession.scopes), + current: change.clientSession.sessionId === currentSessionId, + }, + }; + case "clientRemoved": + return { + version: 1, + revision, + type: "clientRemoved", + payload: { sessionId: change.sessionId }, + }; + } +} + type BootstrapExchangeResult = { readonly response: AuthBrowserSessionResult; readonly sessionToken: string; @@ -1136,6 +1190,42 @@ export const make = Effect.gen(function* () { Effect.withSpan("EnvironmentAuth.listClientSessions"), ); + const streamAccess: EnvironmentAuth["Service"]["streamAccess"] = Effect.fn( + "EnvironmentAuth.streamAccess", + )(function* (currentSessionId) { + // Buffer both stores before reading; consumers may hold the snapshot before + // starting the live tail (for example, while waiting for an RPC ACK). + const pairingLinkChanges = yield* bootstrapCredentials.subscribeChanges; + const clientChanges = yield* sessions.subscribeChanges; + const initialSnapshot = yield* Effect.all({ + pairingLinks: listPairingLinks(), + clientSessions: listClientSessions(currentSessionId), + }); + const revisionRef = yield* Ref.make(1); + const accessChanges: Stream.Stream< + PairingGrantStore.BootstrapCredentialChange | SessionStore.SessionCredentialChange + > = Stream.merge( + Stream.fromSubscription(pairingLinkChanges), + Stream.fromSubscription(clientChanges), + ); + const changes = accessChanges.pipe( + Stream.mapEffect((change) => + Ref.updateAndGet(revisionRef, (revision) => revision + 1).pipe( + Effect.map((revision) => toAuthAccessStreamEvent(change, revision, currentSessionId)), + ), + ), + ); + return { + snapshot: { + version: 1 as const, + revision: 1, + type: "snapshot" as const, + payload: initialSnapshot, + }, + changes, + }; + }); + const revokeClientSession: EnvironmentAuth["Service"]["revokeClientSession"] = Effect.fn( "EnvironmentAuth.revokeClientSession", )(function* (currentSessionId, targetSessionId) { @@ -1300,6 +1390,7 @@ export const make = Effect.gen(function* () { revokeSession, revokeOtherSessionsExcept, listClientSessions, + streamAccess, revokeClientSession, revokeOtherClientSessions, authenticateHttpRequest, diff --git a/apps/server/src/ws.authAccess.test.ts b/apps/server/src/ws.authAccess.test.ts index 127122bc9bdc0..e4b9952f6845b 100644 --- a/apps/server/src/ws.authAccess.test.ts +++ b/apps/server/src/ws.authAccess.test.ts @@ -26,7 +26,8 @@ import * as ServerSecretStore from "./auth/ServerSecretStore.ts"; import * as ServerConfig from "./config.ts"; import * as ServerEnvironment from "./environment/ServerEnvironment.ts"; import * as SqlitePersistence from "./persistence/Sqlite.ts"; -import { subscribeAuthAccess, WS_RPC_SERVER_OPTIONS } from "./ws.ts"; +import { rpcInitialItems } from "./rpcInitialItems.ts"; +import { WS_RPC_SERVER_OPTIONS } from "./ws.ts"; const group = WsRpcGroup.omit( ...[...WsRpcGroup.requests.keys()].filter( @@ -41,7 +42,9 @@ const layerAuth = EnvironmentAuth.layer.pipe( Layer.provideMerge(ServerSecretStore.layer), Layer.provideMerge(SqlitePersistence.layerMemory), Layer.provide(ServerEnvironment.layerIdentity), - Layer.provide(ServerConfig.layerTest(process.cwd(), { prefix: "t3-auth-access-stream-test-" })), + Layer.provideMerge( + ServerConfig.layerTest(process.cwd(), { prefix: "t3-auth-access-stream-test-" }), + ), ); const makeProtocol = Effect.gen(function* () { @@ -115,18 +118,25 @@ it.layer(NodeServices.layer)("subscribeAuthAccess snapshot handoff", (it) => { const snapshotRead = yield* Deferred.make(); const releaseSnapshot = yield* Deferred.make(); const { responses, receive, protocol } = yield* makeProtocol; - const snapshotAuth = EnvironmentAuth.EnvironmentAuth.of({ - ...auth, - listClientSessions: (sessionId) => - auth.listClientSessions(sessionId).pipe( - Effect.tap(() => Deferred.succeed(snapshotRead, undefined)), - Effect.flatMap((snapshot) => Deferred.await(releaseSnapshot).pipe(Effect.as(snapshot))), - ), - }); + const snapshotAuth = yield* EnvironmentAuth.make.pipe( + Effect.provideService(SessionStore.SessionStore, { + ...sessions, + listActive: () => + sessions.listActive().pipe( + Effect.tap(() => Deferred.succeed(snapshotRead, undefined)), + Effect.flatMap((snapshot) => + Deferred.await(releaseSnapshot).pipe(Effect.as(snapshot)), + ), + ), + }), + ); const { write } = yield* startRpc( Stream.unwrap( - subscribeAuthAccess(current.sessionId).pipe( - Effect.provideService(EnvironmentAuth.EnvironmentAuth, snapshotAuth), + snapshotAuth.streamAccess(current.sessionId).pipe( + Effect.map(({ snapshot, changes }) => + Stream.concat(rpcInitialItems([snapshot]), changes), + ), + Effect.mapError((error) => new AuthAccessStreamError({ message: error.message })), ), ), { responses, receive, protocol }, @@ -212,48 +222,47 @@ it.layer(NodeServices.layer)("subscribeAuthAccess snapshot handoff", (it) => { "releases both subscriptions on cancellation or failure during %s", (milestone) => Effect.gen(function* () { - const auth = yield* EnvironmentAuth.EnvironmentAuth; const pairingLinks = yield* PairingGrantStore.PairingGrantStore; const sessions = yield* SessionStore.SessionStore; const current = yield* sessions.issue({ subject: "current-client" }); const acquired = yield* Queue.unbounded>(); const snapshotRead = yield* Deferred.make(); const releaseSnapshot = yield* Deferred.make(); - const snapshotAuth = EnvironmentAuth.EnvironmentAuth.of({ - ...auth, - listClientSessions: (sessionId) => - auth.listClientSessions(sessionId).pipe( - Effect.tap(() => Deferred.succeed(snapshotRead, undefined)), - Effect.flatMap((snapshot) => - Deferred.await(releaseSnapshot).pipe( - Effect.andThen( - milestone === "snapshot failure" - ? Effect.fail( - new EnvironmentAuth.ServerAuthSessionsListError({ cause: "test" }), - ) - : Effect.succeed(snapshot), + const snapshotAuth = yield* EnvironmentAuth.make.pipe( + Effect.provideService(PairingGrantStore.PairingGrantStore, { + ...pairingLinks, + subscribeChanges: pairingLinks.subscribeChanges.pipe( + Effect.tap((subscription) => Queue.offer(acquired, subscription)), + ), + }), + Effect.provideService(SessionStore.SessionStore, { + ...sessions, + subscribeChanges: sessions.subscribeChanges.pipe( + Effect.tap((subscription) => Queue.offer(acquired, subscription)), + ), + listActive: () => + sessions.listActive().pipe( + Effect.tap(() => Deferred.succeed(snapshotRead, undefined)), + Effect.flatMap((snapshot) => + Deferred.await(releaseSnapshot).pipe( + Effect.andThen( + milestone === "snapshot failure" + ? Effect.fail(new SessionStore.ActiveSessionsListError({ cause: "test" })) + : Effect.succeed(snapshot), + ), ), ), ), - ), - }); + }), + ); const { responses, receive, protocol } = yield* makeProtocol; const { write } = yield* startRpc( Stream.unwrap( - subscribeAuthAccess(current.sessionId).pipe( - Effect.provideService(EnvironmentAuth.EnvironmentAuth, snapshotAuth), - Effect.provideService(PairingGrantStore.PairingGrantStore, { - ...pairingLinks, - subscribeChanges: pairingLinks.subscribeChanges.pipe( - Effect.tap((subscription) => Queue.offer(acquired, subscription)), - ), - }), - Effect.provideService(SessionStore.SessionStore, { - ...sessions, - subscribeChanges: sessions.subscribeChanges.pipe( - Effect.tap((subscription) => Queue.offer(acquired, subscription)), - ), - }), + snapshotAuth.streamAccess(current.sessionId).pipe( + Effect.map(({ snapshot, changes }) => + Stream.concat(rpcInitialItems([snapshot]), changes), + ), + Effect.mapError((error) => new AuthAccessStreamError({ message: error.message })), ), ), { responses, receive, protocol }, diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 26124c202f962..73bd9f80abd42 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -24,13 +24,10 @@ import { DEFAULT_AUTOMATIC_GIT_FETCH_INTERVAL, AcpRegistryOperationError, CommandId, - authScopeResponse, AuthAccessStreamError, - type AuthAccessStreamEvent, AuthOrchestrationOperateScope, type AuthEnvironmentScope, type ScheduledTaskListResult, - AuthSessionId, ClientConnectionMethod, ClientDeviceType, ClientOs, @@ -237,7 +234,6 @@ import * as SourceControlProviderRegistry from "./sourceControl/SourceControlPro import * as GitVcsDriver from "./vcs/GitVcsDriver.ts"; import * as VcsDriverRegistry from "./vcs/VcsDriverRegistry.ts"; import * as VcsProjectConfig from "./vcs/VcsProjectConfig.ts"; -import * as PairingGrantStore from "./auth/PairingGrantStore.ts"; import * as SessionStore from "./auth/SessionStore.ts"; import { failEnvironmentAuthInvalid, failEnvironmentInternal } from "./auth/http.ts"; import * as RelayClient from "@t3tools/shared/relayClient"; @@ -553,47 +549,6 @@ const SHELL_RESUME_MAX_GAP = 1_000; // payload bytes of the range in SQL and reset with a snapshot past this budget. const ORCHESTRATION_REPLAY_PAYLOAD_BUDGET_BYTES = 8 * 1024 * 1024; -function toAuthAccessStreamEvent( - change: PairingGrantStore.BootstrapCredentialChange | SessionStore.SessionCredentialChange, - revision: number, - currentSessionId: AuthSessionId, -): AuthAccessStreamEvent { - switch (change.type) { - case "pairingLinkUpserted": - return { - version: 1, - revision, - type: "pairingLinkUpserted", - payload: { ...change.pairingLink, ...authScopeResponse(change.pairingLink.scopes) }, - }; - case "pairingLinkRemoved": - return { - version: 1, - revision, - type: "pairingLinkRemoved", - payload: { id: change.id }, - }; - case "clientUpserted": - return { - version: 1, - revision, - type: "clientUpserted", - payload: { - ...change.clientSession, - ...authScopeResponse(change.clientSession.scopes), - current: change.clientSession.sessionId === currentSessionId, - }, - }; - case "clientRemoved": - return { - version: 1, - revision, - type: "clientRemoved", - payload: { sessionId: change.sessionId }, - }; - } -} - const isClientSurface = Schema.is(ClientSurface); const isClientConnectionMethod = Schema.is(ClientConnectionMethod); const isClientDeviceType = Schema.is(ClientDeviceType); @@ -603,49 +558,6 @@ const MAX_CLIENT_APP_VERSION_LENGTH = 64; const MAX_CLIENT_BROWSER_LENGTH = 64; const MAX_CLIENT_DEVICE_MODEL_LENGTH = 80; -export const subscribeAuthAccess = Effect.fn("ws.subscribeAuthAccess")(function* ( - currentSessionId: AuthSessionId, -) { - const serverAuth = yield* EnvironmentAuth.EnvironmentAuth; - const bootstrapCredentials = yield* PairingGrantStore.PairingGrantStore; - const sessions = yield* SessionStore.SessionStore; - // Buffer both stores before reading; RPC backpressure can hold the snapshot - // until its ACK, delaying consumption of the live tail. - const pairingLinkChanges = yield* bootstrapCredentials.subscribeChanges; - const clientChanges = yield* sessions.subscribeChanges; - const initialSnapshot = yield* Effect.all({ - pairingLinks: serverAuth.listPairingLinks(), - clientSessions: serverAuth.listClientSessions(currentSessionId), - }).pipe(Effect.mapError((error) => new AuthAccessStreamError({ message: error.message }))); - const revisionRef = yield* Ref.make(1); - const accessChanges: Stream.Stream< - PairingGrantStore.BootstrapCredentialChange | SessionStore.SessionCredentialChange - > = Stream.merge( - Stream.fromSubscription(pairingLinkChanges), - Stream.fromSubscription(clientChanges), - ); - - const liveEvents: Stream.Stream = accessChanges.pipe( - Stream.mapEffect((change) => - Ref.updateAndGet(revisionRef, (revision) => revision + 1).pipe( - Effect.map((revision) => toAuthAccessStreamEvent(change, revision, currentSessionId)), - ), - ), - ); - - return Stream.concat( - rpcInitialItems([ - { - version: 1 as const, - revision: 1, - type: "snapshot" as const, - payload: initialSnapshot, - }, - ]), - liveEvents, - ); -}); - export function hasCompatibleOrchestrationProtocol(url: URL): boolean { return ( url.searchParams.get(ORCHESTRATION_PROTOCOL_QUERY_PARAM) === @@ -3118,7 +3030,14 @@ const layerWsRpc = ( }), ), [WS_METHODS.subscribeAuthAccess]: (_input) => - Stream.unwrap(subscribeAuthAccess(currentSessionId)), + Stream.unwrap( + serverAuth.streamAccess(currentSessionId).pipe( + Effect.map(({ snapshot, changes }) => + Stream.concat(rpcInitialItems([snapshot]), changes), + ), + Effect.mapError((error) => new AuthAccessStreamError({ message: error.message })), + ), + ), [WS_METHODS.subscribeBackgroundPolicy]: (_input) => Stream.unwrap( Effect.map(backgroundPolicy.subscribe, ({ latest, changes }) => From dcc2a060e69b9a7d0fd2d7a23853da2bdc27c977 Mon Sep 17 00:00:00 2001 From: SAPHID Date: Wed, 14 Oct 2026 21:15:43 +0000 Subject: [PATCH 3/8] fix(server): narrow auth access buffering to the existing handler Restore the inline workflow and keep one ACK-window regression with scoped cleanup. Use two session changes to test ordering within one source. Co-Authored-By: Claude Opus 5.5 Co-Authored-By: GPT-6.1 Sol --- apps/server/src/auth/EnvironmentAuth.ts | 91 ---- apps/server/src/auth/PairingGrantStore.ts | 7 +- apps/server/src/auth/SessionStore.ts | 7 +- apps/server/src/ws.authAccess.test.ts | 500 +++++++++++----------- apps/server/src/ws.ts | 100 ++++- 5 files changed, 355 insertions(+), 350 deletions(-) diff --git a/apps/server/src/auth/EnvironmentAuth.ts b/apps/server/src/auth/EnvironmentAuth.ts index 7d13869deae23..8a59f22fca9f9 100644 --- a/apps/server/src/auth/EnvironmentAuth.ts +++ b/apps/server/src/auth/EnvironmentAuth.ts @@ -1,5 +1,4 @@ import { - type AuthAccessStreamEvent, AuthAccessTokenType, AuthAccessWriteScope, AuthAdministrativeScopes, @@ -34,10 +33,7 @@ import * as Duration from "effect/Duration"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; -import * as Ref from "effect/Ref"; import * as Schema from "effect/Schema"; -import * as Scope from "effect/Scope"; -import * as Stream from "effect/Stream"; import * as HttpServerRequest from "effect/http/HttpServerRequest"; import * as ServerEnvironment from "../environment/ServerEnvironment.ts"; @@ -543,15 +539,6 @@ export class EnvironmentAuth extends Context.Service< readonly listClientSessions: ( currentSessionId: AuthSessionId, ) => Effect.Effect, ServerAuthInternalError>; - /** Acquires both live subscriptions before reading the request's snapshot. */ - readonly streamAccess: (currentSessionId: AuthSessionId) => Effect.Effect< - { - readonly snapshot: Extract; - readonly changes: Stream.Stream; - }, - ServerAuthInternalError, - Scope.Scope - >; readonly revokeClientSession: ( currentSessionId: AuthSessionId, targetSessionId: AuthSessionId, @@ -600,47 +587,6 @@ export class EnvironmentAuth extends Context.Service< } >()("t3/auth/EnvironmentAuth") {} -function toAuthAccessStreamEvent( - change: PairingGrantStore.BootstrapCredentialChange | SessionStore.SessionCredentialChange, - revision: number, - currentSessionId: AuthSessionId, -): AuthAccessStreamEvent { - switch (change.type) { - case "pairingLinkUpserted": - return { - version: 1, - revision, - type: "pairingLinkUpserted", - payload: { ...change.pairingLink, ...authScopeResponse(change.pairingLink.scopes) }, - }; - case "pairingLinkRemoved": - return { - version: 1, - revision, - type: "pairingLinkRemoved", - payload: { id: change.id }, - }; - case "clientUpserted": - return { - version: 1, - revision, - type: "clientUpserted", - payload: { - ...change.clientSession, - ...authScopeResponse(change.clientSession.scopes), - current: change.clientSession.sessionId === currentSessionId, - }, - }; - case "clientRemoved": - return { - version: 1, - revision, - type: "clientRemoved", - payload: { sessionId: change.sessionId }, - }; - } -} - type BootstrapExchangeResult = { readonly response: AuthBrowserSessionResult; readonly sessionToken: string; @@ -1256,42 +1202,6 @@ export const make = Effect.gen(function* () { Effect.withSpan("EnvironmentAuth.listClientSessions"), ); - const streamAccess: EnvironmentAuth["Service"]["streamAccess"] = Effect.fn( - "EnvironmentAuth.streamAccess", - )(function* (currentSessionId) { - // Buffer both stores before reading; consumers may hold the snapshot before - // starting the live tail (for example, while waiting for an RPC ACK). - const pairingLinkChanges = yield* bootstrapCredentials.subscribeChanges; - const clientChanges = yield* sessions.subscribeChanges; - const initialSnapshot = yield* Effect.all({ - pairingLinks: listPairingLinks(), - clientSessions: listClientSessions(currentSessionId), - }); - const revisionRef = yield* Ref.make(1); - const accessChanges: Stream.Stream< - PairingGrantStore.BootstrapCredentialChange | SessionStore.SessionCredentialChange - > = Stream.merge( - Stream.fromSubscription(pairingLinkChanges), - Stream.fromSubscription(clientChanges), - ); - const changes = accessChanges.pipe( - Stream.mapEffect((change) => - Ref.updateAndGet(revisionRef, (revision) => revision + 1).pipe( - Effect.map((revision) => toAuthAccessStreamEvent(change, revision, currentSessionId)), - ), - ), - ); - return { - snapshot: { - version: 1 as const, - revision: 1, - type: "snapshot" as const, - payload: initialSnapshot, - }, - changes, - }; - }); - const revokeClientSession: EnvironmentAuth["Service"]["revokeClientSession"] = Effect.fn( "EnvironmentAuth.revokeClientSession", )(function* (currentSessionId, targetSessionId) { @@ -1455,7 +1365,6 @@ export const make = Effect.gen(function* () { revokeSession, revokeOtherSessionsExcept, listClientSessions, - streamAccess, revokeClientSession, revokeOtherClientSessions, authenticateHttpRequest, diff --git a/apps/server/src/auth/PairingGrantStore.ts b/apps/server/src/auth/PairingGrantStore.ts index b4be105786767..7d34369bc03a1 100644 --- a/apps/server/src/auth/PairingGrantStore.ts +++ b/apps/server/src/auth/PairingGrantStore.ts @@ -226,8 +226,9 @@ export class PairingGrantStore extends Context.Service< BootstrapCredentialInternalError >; readonly streamChanges: Stream.Stream; + /** Acquire before reading a snapshot to buffer changes until the stream starts. */ readonly subscribeChanges: Effect.Effect< - PubSub.Subscription, + Stream.Stream, never, Scope.Scope >; @@ -637,7 +638,9 @@ export const make = Effect.gen(function* () { get streamChanges() { return Stream.fromPubSub(changesPubSub); }, - subscribeChanges: PubSub.subscribe(changesPubSub), + get subscribeChanges() { + return PubSub.subscribe(changesPubSub).pipe(Effect.map(Stream.fromSubscription)); + }, revoke, consume, }); diff --git a/apps/server/src/auth/SessionStore.ts b/apps/server/src/auth/SessionStore.ts index 9601223ddfa8d..e0c758ceedebd 100644 --- a/apps/server/src/auth/SessionStore.ts +++ b/apps/server/src/auth/SessionStore.ts @@ -408,8 +408,9 @@ export class SessionStore extends Context.Service< SessionCredentialInternalError >; readonly streamChanges: Stream.Stream; + /** Acquire before reading a snapshot to buffer changes until the stream starts. */ readonly subscribeChanges: Effect.Effect< - PubSub.Subscription, + Stream.Stream, never, Scope.Scope >; @@ -1097,7 +1098,9 @@ export const make = Effect.gen(function* () { get streamChanges() { return Stream.fromPubSub(changesPubSub); }, - subscribeChanges: PubSub.subscribe(changesPubSub), + get subscribeChanges() { + return PubSub.subscribe(changesPubSub).pipe(Effect.map(Stream.fromSubscription)); + }, revoke, revokeAllExcept, markConnected, diff --git a/apps/server/src/ws.authAccess.test.ts b/apps/server/src/ws.authAccess.test.ts index e4b9952f6845b..86e7cc8759a41 100644 --- a/apps/server/src/ws.authAccess.test.ts +++ b/apps/server/src/ws.authAccess.test.ts @@ -1,33 +1,100 @@ import * as NodeServices from "@effect/platform-node/NodeServices"; import { assert, it } from "@effect/vitest"; import { - AuthAccessStreamEvent, - AuthAccessStreamError, AuthAdministrativeScopes, + type AuthAccessStreamEvent, WS_METHODS, WsRpcGroup, } from "@t3tools/contracts"; import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; -import * as PubSub from "effect/PubSub"; import * as Layer from "effect/Layer"; import * as Queue from "effect/Queue"; -import * as Schema from "effect/Schema"; +import type * as Scope from "effect/Scope"; import * as Stream from "effect/Stream"; -import * as TestClock from "effect/testing/TestClock"; -import { RpcMessage, RpcSerialization, RpcServer } from "effect/rpc"; +import { Headers } from "effect/http"; +import { RpcMessage, RpcServer } from "effect/rpc"; import * as EnvironmentAuth from "./auth/EnvironmentAuth.ts"; import * as PairingGrantStore from "./auth/PairingGrantStore.ts"; import * as RpcAuthorization from "./auth/RpcAuthorization.ts"; import { RPC_REQUIRED_SCOPES } from "./auth/RpcAuthorization.ts"; -import * as SessionStore from "./auth/SessionStore.ts"; import * as ServerSecretStore from "./auth/ServerSecretStore.ts"; -import * as ServerConfig from "./config.ts"; +import * as SessionStore from "./auth/SessionStore.ts"; +import * as BackgroundPolicy from "./background/BackgroundPolicy.ts"; +import * as CheckpointDiffQuery from "./checkpointing/CheckpointDiffQuery.ts"; +import * as ServerSelfUpdate from "./cloud/selfUpdate.ts"; +import * as DeviceService from "./device/DeviceService.ts"; +import * as ProcessDiagnostics from "./diagnostics/ProcessDiagnostics.ts"; +import * as ProcessResourceMonitor from "./diagnostics/ProcessResourceMonitor.ts"; +import * as TraceDiagnostics from "./diagnostics/TraceDiagnostics.ts"; +import * as DirectEndpoints from "./environment/DirectEndpoints.ts"; +import * as RemoteOpenTargets from "./environment/RemoteOpenTargets.ts"; import * as ServerEnvironment from "./environment/ServerEnvironment.ts"; +import * as ServerConfig from "./config.ts"; +import * as EnvironmentTheme from "./environmentTheme.ts"; +import * as GitWorkflowService from "./git/GitWorkflowService.ts"; +import * as Keybindings from "./keybindings.ts"; +import * as McpAppRequests from "./mcpApps/McpAppRequests.ts"; +import * as OtlpEndpointCheck from "./observability/OtlpEndpointCheck.ts"; +import * as Orchestrator from "./orchestration-v2/Orchestrator.ts"; +import * as ProjectStore from "./orchestration-v2/ProjectStore.ts"; +import * as ProviderSessionManager from "./orchestration-v2/ProviderSessionManager.ts"; +import * as PullRequestSyncReactor from "./orchestration-v2/PullRequestSyncReactor.ts"; +import * as ThreadLaunchService from "./orchestration-v2/ThreadLaunchService.ts"; +import * as ThreadManagement from "./orchestration-v2/ThreadManagementService.ts"; +import * as ThreadSearch from "./orchestration-v2/ThreadSearch.ts"; +import * as OrchestrationEventStore from "./persistence/OrchestrationEventStore.ts"; import * as SqlitePersistence from "./persistence/Sqlite.ts"; -import { rpcInitialItems } from "./rpcInitialItems.ts"; -import { WS_RPC_SERVER_OPTIONS } from "./ws.ts"; +import * as PreviewAutomationBroker from "./mcp/PreviewAutomationBroker.ts"; +import * as PreviewManager from "./preview/Manager.ts"; +import * as PortScanner from "./preview/PortScanner.ts"; +import * as ServerBrowser from "./preview/ServerBrowser.ts"; +import * as ExternalLauncher from "./process/externalLauncher.ts"; +import * as AgentSessionImporter from "./project/AgentSessionImporter.ts"; +import * as AgentSessionScanner from "./project/AgentSessionScanner.ts"; +import * as ManagedProjectFolders from "./project/ManagedProjectFolders.ts"; +import * as ProjectCloneTracker from "./project/ProjectCloneTracker.ts"; +import * as ProjectEnrichment from "./project/ProjectEnrichmentService.ts"; +import * as ProjectFaviconResolver from "./project/ProjectFaviconResolver.ts"; +import * as ProjectService from "./project/ProjectService.ts"; +import * as ProjectSetupScriptRunner from "./project/ProjectSetupScriptRunner.ts"; +import * as RepositoryIdentityResolver from "./project/RepositoryIdentityResolver.ts"; +import * as WorktreeSetupTracker from "./project/WorktreeSetupTracker.ts"; +import * as AntigravityInstallation from "./provider/AntigravityInstallation.ts"; +import * as CodexInstallation from "./provider/CodexInstallation.ts"; +import * as ModelManifest from "./provider/ModelManifest.ts"; +import * as ProviderAuthService from "./provider/ProviderAuthService.ts"; +import * as ProviderInstanceRegistry from "./provider/ProviderInstanceRegistry.ts"; +import * as ProviderMaintenanceRunner from "./provider/providerMaintenanceRunner.ts"; +import * as ProviderRegistry from "./provider/ProviderRegistry.ts"; +import * as PullRequestReviewGuideService from "./pullRequest/PullRequestReviewGuideService.ts"; +import * as PullRequestService from "./pullRequest/PullRequestService.ts"; +import * as HostResources from "./resourceTelemetry/HostResources.ts"; +import * as ResourceTelemetry from "./resourceTelemetry/ResourceTelemetry.ts"; +import * as ReviewService from "./review/ReviewService.ts"; +import * as ScheduledTasks from "./scheduledTasks/ScheduledTaskService.ts"; +import * as SecretRequests from "./secrets/SecretRequests.ts"; +import * as ServerLifecycleEvents from "./serverLifecycleEvents.ts"; +import * as ServerRuntimeStartup from "./serverRuntimeStartup.ts"; +import * as ServerSettings from "./serverSettings.ts"; +import * as SourceControlDiscovery from "./sourceControl/SourceControlDiscovery.ts"; +import * as SourceControlRepositoryService from "./sourceControl/SourceControlRepositoryService.ts"; +import * as StorageCleanup from "./storageCleanup.ts"; +import * as AcpRegistryRuntimeCoordinator from "@t3tools/provider-acp-registry/server/AcpRegistryRuntimeCoordinator"; +import * as AcpRegistrySupport from "@t3tools/provider-acp-registry/server/AcpRegistrySupport"; +import * as ProviderLatestVersions from "@t3tools/provider-core/server/ProviderLatestVersions"; +import * as RelayClient from "@t3tools/shared/relayClient"; +import * as AnalyticsService from "./telemetry/AnalyticsService.ts"; +import * as TerminalManager from "./terminal/Manager.ts"; +import * as UsageLimitSources from "./usage/UsageLimitSources.ts"; +import * as UsageService from "./usage/UsageService.ts"; +import * as VcsProvisioningService from "./vcs/VcsProvisioningService.ts"; +import * as VcsStatusBroadcaster from "./vcs/VcsStatusBroadcaster.ts"; +import * as WorkspaceEntries from "./workspace/WorkspaceEntries.ts"; +import * as WorkspaceFileSystem from "./workspace/WorkspaceFileSystem.ts"; +import * as WorkspacePaths from "./workspace/WorkspacePaths.ts"; +import { layerWsRpc, WS_RPC_SERVER_OPTIONS } from "./ws.ts"; const group = WsRpcGroup.omit( ...[...WsRpcGroup.requests.keys()].filter( @@ -38,263 +105,200 @@ const group = WsRpcGroup.omit( ), ); +// Build the production RPC layer; unrelated capabilities fail if this test calls them. +const layerUnusedAuthAccessRpcServices = Layer.mergeAll( + Layer.mock(AcpRegistryRuntimeCoordinator.AcpRegistryRuntimeCoordinator)({}), + Layer.mock(AcpRegistrySupport.AcpRegistryCatalog)({}), + Layer.mock(AgentSessionImporter.AgentSessionImporter)({}), + Layer.mock(AgentSessionScanner.AgentSessionScanner)({}), + Layer.mock(AnalyticsService.AnalyticsService)({}), + Layer.mock(AntigravityInstallation.AntigravityInstallation)({ managedDirectory: "/unused" }), + Layer.mock(BackgroundPolicy.BackgroundPolicy)({}), + Layer.mock(CheckpointDiffQuery.CheckpointDiffQuery)({}), + Layer.mock(CodexInstallation.CodexInstallation)({ managedDirectory: "/unused" }), + Layer.mock(DeviceService.DeviceService)({}), + Layer.mock(DirectEndpoints.DirectEndpoints)({}), + Layer.mock(EnvironmentTheme.EnvironmentThemeService)({}), + Layer.mock(ExternalLauncher.ExternalLauncher)({}), + Layer.mock(GitWorkflowService.GitWorkflowService)({}), + Layer.mock(HostResources.HostResources)({}), + Layer.mock(Keybindings.Keybindings)({}), + Layer.mock(ManagedProjectFolders.ManagedProjectFolders)({ namedProjectsRoot: "/unused" }), + Layer.mock(McpAppRequests.McpAppRequests)({}), + Layer.mock(ModelManifest.ModelManifest)({}), + Layer.mock(OrchestrationEventStore.OrchestrationEventStore)({}), + Layer.mock(Orchestrator.OrchestratorV2)({}), + Layer.mock(OtlpEndpointCheck.OtlpEndpointCheck)({}), + Layer.mock(PortScanner.PortDiscovery)({}), + Layer.mock(PreviewManager.PreviewManager)({}), + Layer.mock(ProcessDiagnostics.ProcessDiagnostics)({}), + Layer.mock(ProcessResourceMonitor.ProcessResourceMonitor)({}), + Layer.mock(ProjectCloneTracker.ProjectCloneTracker)({}), + Layer.mock(ProjectEnrichment.ProjectEnrichmentService)({}), + Layer.mock(ProjectFaviconResolver.ProjectFaviconResolver)({}), + Layer.mock(ProjectService.ProjectService)({}), + Layer.mock(ProjectSetupScriptRunner.ProjectSetupScriptRunner)({}), + Layer.mock(ProjectStore.ProjectStoreV2)({}), + Layer.mock(ProviderAuthService.ProviderAuthService)({}), + Layer.mock(ProviderInstanceRegistry.ProviderInstanceRegistry)({}), + Layer.mock(ProviderLatestVersions.ProviderLatestVersions)({}), + Layer.mock(ProviderMaintenanceRunner.ProviderMaintenanceRunner)({}), + Layer.mock(ProviderRegistry.ProviderRegistry)({}), + Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({}), + Layer.mock(PullRequestReviewGuideService.PullRequestReviewGuideService)({}), + Layer.mock(PullRequestService.PullRequestService)({}), + Layer.mock(PullRequestSyncReactor.PullRequestSyncReactor)({}), + Layer.mock(RelayClient.RelayClient)({}), + Layer.mock(RemoteOpenTargets.RemoteOpenTargets)({}), + Layer.mock(RepositoryIdentityResolver.RepositoryIdentityResolver)({}), + Layer.mock(ResourceTelemetry.ResourceTelemetry)({}), + Layer.mock(ReviewService.ReviewService)({}), + Layer.mock(ScheduledTasks.ScheduledTaskService)({}), + Layer.mock(SecretRequests.SecretRequests)({}), + Layer.mock(ServerEnvironment.ServerEnvironment)({}), + Layer.mock(ServerLifecycleEvents.ServerLifecycleEvents)({}), + Layer.mock(ServerRuntimeStartup.ServerRuntimeStartup)({}), + Layer.mock(ServerSelfUpdate.ServerSelfUpdate)({}), + Layer.mock(ServerSettings.ServerSettingsService)({}), + Layer.mock(SourceControlDiscovery.SourceControlDiscovery)({}), + Layer.mock(SourceControlRepositoryService.SourceControlRepositoryService)({}), + Layer.mock(StorageCleanup.StorageCleanup)({}), + Layer.mock(TerminalManager.TerminalManager)({}), + Layer.mock(ThreadLaunchService.ThreadLaunchService)({}), + Layer.mock(ThreadManagement.ThreadManagementService)({}), + Layer.mock(ThreadSearch.ThreadSearch)({}), + Layer.mock(TraceDiagnostics.TraceDiagnostics)({}), + Layer.mock(UsageLimitSources.UsageLimitSources)({}), + Layer.mock(UsageService.UsageService)({}), + Layer.mock(VcsProvisioningService.VcsProvisioningService)({}), + Layer.mock(VcsStatusBroadcaster.VcsStatusBroadcaster)({}), + Layer.mock(WorkspaceEntries.WorkspaceEntries)({}), + Layer.mock(WorkspaceFileSystem.WorkspaceFileSystem)({}), + Layer.mock(WorkspacePaths.WorkspacePaths)({}), + Layer.mock(WorktreeSetupTracker.WorktreeSetupTracker)({}), +); + const layerAuth = EnvironmentAuth.layer.pipe( Layer.provideMerge(ServerSecretStore.layer), Layer.provideMerge(SqlitePersistence.layerMemory), - Layer.provide(ServerEnvironment.layerIdentity), + Layer.provideMerge(ServerEnvironment.layerIdentity), Layer.provideMerge( ServerConfig.layerTest(process.cwd(), { prefix: "t3-auth-access-stream-test-" }), ), + Layer.provideMerge(NodeServices.layer), ); -const makeProtocol = Effect.gen(function* () { - const responses = yield* Queue.unbounded(); - const receive = yield* Deferred.make[0]>(); - const protocol = yield* RpcServer.Protocol.make((write) => - Effect.gen(function* () { - yield* Deferred.succeed(receive, write); - const serialization = yield* RpcSerialization.RpcSerialization; - return { - disconnects: yield* Queue.unbounded(), - send: (_clientId, response) => Queue.offer(responses, response), - end: () => Effect.void, - clientIds: Effect.succeed(new Set([0])), - initialMessage: Effect.succeedNone, - supportsAck: true, - supportsTransferables: false, - supportsSpanPropagation: false, - supportsNotifications: true, - codecFor: serialization.codecFor, - }; - }), - ); - return { responses, receive, protocol }; -}); - -const startRpc = Effect.fn(function* ( - stream: Stream.Stream, - { responses, receive, protocol }: Effect.Success, -) { - yield* RpcServer.make(group, WS_RPC_SERVER_OPTIONS).pipe( - Effect.provide( - Layer.mergeAll( - group.toLayerHandler(WS_METHODS.subscribeAuthAccess, () => stream), - RpcAuthorization.layer(AuthAdministrativeScopes), - ), - ), - Effect.provideService(RpcServer.Protocol, protocol), - Effect.forkScoped, - ); - return { write: yield* Deferred.await(receive), responses }; -}); - -const decodeEvents = Schema.decodeUnknownSync( - Schema.toCodecJson(Schema.Array(AuthAccessStreamEvent)), -); -const takeEvents = (responses: Queue.Queue) => - Queue.take(responses).pipe( - Effect.map((response) => { - assert.equal(response._tag, "Chunk"); - return response._tag === "Chunk" ? decodeEvents(response.values) : []; - }), - ); - -it.layer(NodeServices.layer)("subscribeAuthAccess snapshot handoff", (it) => { - it.effect.each([ - { change: "revoke link", milestone: "snapshot read" }, - { change: "revoke link", milestone: "snapshot ACK" }, - { change: "connect client", milestone: "snapshot read" }, - { change: "connect client", milestone: "snapshot ACK" }, - { change: "disconnect client", milestone: "snapshot read" }, - { change: "disconnect client", milestone: "snapshot ACK" }, - ] as const)("buffers $change while waiting for $milestone", ({ change, milestone }) => +it.effect( + "keeps an access change made before the snapshot ACK, then releases both subscriptions", + () => Effect.gen(function* () { const auth = yield* EnvironmentAuth.EnvironmentAuth; + const pairingLinks = yield* PairingGrantStore.PairingGrantStore; const sessions = yield* SessionStore.SessionStore; const current = yield* sessions.issue({ subject: "current-client" }); - const link = yield* auth.createPairingLink({ label: "Revoked during subscription" }); - if (change === "disconnect client") yield* sessions.markConnected(current.sessionId); - - const snapshotRead = yield* Deferred.make(); - const releaseSnapshot = yield* Deferred.make(); - const { responses, receive, protocol } = yield* makeProtocol; - const snapshotAuth = yield* EnvironmentAuth.make.pipe( - Effect.provideService(SessionStore.SessionStore, { - ...sessions, - listActive: () => - sessions.listActive().pipe( - Effect.tap(() => Deferred.succeed(snapshotRead, undefined)), - Effect.flatMap((snapshot) => - Deferred.await(releaseSnapshot).pipe(Effect.as(snapshot)), + const subscribed = yield* Queue.unbounded(); + const released = yield* Queue.unbounded(); + // Report each live subscription on either accessor so the test fails, rather + // than hangs, against a handler that subscribes after the snapshot. + const tracked = (changes: Effect.Effect, never, Scope.Scope>) => + changes.pipe( + Effect.tap(() => Queue.offer(subscribed, undefined)), + Effect.tap(() => Effect.addFinalizer(() => Queue.offer(released, undefined))), + ); + const pairingLinkChanges = tracked(pairingLinks.subscribeChanges); + const clientChanges = tracked(sessions.subscribeChanges); + const responses = yield* Queue.unbounded(); + const exited = yield* Deferred.make(); + const server = yield* RpcServer.makeNoSerialization(group, { + ...WS_RPC_SERVER_OPTIONS, + onFromServer: (response) => + response._tag === "Chunk" + ? Effect.forEach(response.values, (event) => Queue.offer(responses, event), { + discard: true, + }) + : response._tag === "Exit" + ? Deferred.succeed(exited, undefined) + : Effect.void, + }).pipe( + Effect.provide( + Layer.mergeAll( + layerWsRpc( + { + sessionId: current.sessionId, + subject: "current-client", + method: "browser-session-cookie", + scopes: AuthAdministrativeScopes, + }, + {}, + {}, + yield* PreviewAutomationBroker.PreviewAutomationBroker, + yield* ServerBrowser.ServerBrowser, + ).pipe( + Layer.provide( + Layer.mergeAll( + layerUnusedAuthAccessRpcServices, + Layer.succeed(EnvironmentAuth.EnvironmentAuth, auth), + Layer.succeed(PairingGrantStore.PairingGrantStore, { + ...pairingLinks, + streamChanges: Stream.unwrap(pairingLinkChanges), + subscribeChanges: pairingLinkChanges, + }), + Layer.succeed(SessionStore.SessionStore, { + ...sessions, + streamChanges: Stream.unwrap(clientChanges), + subscribeChanges: clientChanges, + }), + ), ), ), - }), - ); - const { write } = yield* startRpc( - Stream.unwrap( - snapshotAuth.streamAccess(current.sessionId).pipe( - Effect.map(({ snapshot, changes }) => - Stream.concat(rpcInitialItems([snapshot]), changes), - ), - Effect.mapError((error) => new AuthAccessStreamError({ message: error.message })), + RpcAuthorization.layer(AuthAdministrativeScopes), ), ), - { responses, receive, protocol }, ); - yield* write(0, { + const requestId = RpcMessage.RequestId("auth-access"); + const acknowledge = server.write(0, { _tag: "Ack", requestId }); + yield* server.write(0, { _tag: "Request", - id: "1", + id: requestId, tag: WS_METHODS.subscribeAuthAccess, payload: {}, - headers: [], + headers: Headers.empty, }); - yield* Deferred.await(snapshotRead).pipe( - Effect.raceFirst( - Queue.take(responses).pipe(Effect.flatMap((response) => Effect.die(response))), - ), - ); - const mutate = - change === "revoke link" - ? auth.revokePairingLink(link.id) - : change === "connect client" - ? sessions.markConnected(current.sessionId) - : sessions.markDisconnected(current.sessionId); - if (milestone === "snapshot read") yield* mutate; - yield* Deferred.succeed(releaseSnapshot, undefined); - const [snapshot] = yield* takeEvents(responses); - assert.equal(snapshot?.type, "snapshot"); - if (snapshot?.type !== "snapshot") return; - assert.equal(snapshot.revision, 1); - assert.deepEqual( - snapshot.payload.pairingLinks.map((value) => value.id), - [link.id], - ); - assert.equal(snapshot.payload.clientSessions[0]?.connected, change === "disconnect client"); - if (milestone === "snapshot ACK") yield* mutate; - yield* write(0, { _tag: "Ack", requestId: "1" }); - // Drain runnable fibers to attach the old handler's live streams. This does - // not advance time. An unrelated update makes lost changes fail without a timeout. - yield* TestClock.adjust(0); - const unrelated = yield* auth.createPairingLink({ label: "Unrelated update" }); - const [next] = yield* takeEvents(responses); - assert.equal(next?.revision, 2); - if (change === "revoke link") { - assert.equal(next?.type, "pairingLinkRemoved"); - if (next?.type === "pairingLinkRemoved") { - assert.equal(next.payload.id, link.id); - const remaining = snapshot.payload.pairingLinks.filter( - (value) => value.id !== next.payload.id, - ); - assert.deepEqual( - remaining.map((value) => value.id), - [], - ); - } - } else { - assert.equal(next?.type, "clientUpserted"); - if (next?.type === "clientUpserted") { - assert.equal(next.payload.sessionId, current.sessionId); - assert.isTrue(next.payload.current); - assert.equal(next.payload.connected, change === "connect client"); - assert.deepEqual(next.payload, (yield* auth.listClientSessions(current.sessionId))[0]); - } - } - yield* write(0, { _tag: "Ack", requestId: "1" }); - const [following] = yield* takeEvents(responses); - assert.equal(following?.revision, 3); - assert.equal(following?.type, "pairingLinkUpserted"); - if (following?.type === "pairingLinkUpserted") { - assert.equal(following.payload.id, unrelated.id); - const remaining = snapshot.payload.pairingLinks.filter( - (value) => change !== "revoke link" || value.id !== link.id, - ); - assert.deepEqual( - [...remaining, following.payload].map((value) => value.id).toSorted(), - (yield* auth.listPairingLinks()).map((value) => value.id).toSorted(), - ); - } - yield* write(0, { _tag: "Interrupt", requestId: "1" }); - assert.equal((yield* Queue.take(responses))._tag, "Exit"); - }).pipe(Effect.scoped, Effect.provide(Layer.mergeAll(layerAuth, RpcSerialization.layerJson))), - ); - it.effect.each(["snapshot read", "snapshot ACK", "live tail", "snapshot failure"] as const)( - "releases both subscriptions on cancellation or failure during %s", - (milestone) => - Effect.gen(function* () { - const pairingLinks = yield* PairingGrantStore.PairingGrantStore; - const sessions = yield* SessionStore.SessionStore; - const current = yield* sessions.issue({ subject: "current-client" }); - const acquired = yield* Queue.unbounded>(); - const snapshotRead = yield* Deferred.make(); - const releaseSnapshot = yield* Deferred.make(); - const snapshotAuth = yield* EnvironmentAuth.make.pipe( - Effect.provideService(PairingGrantStore.PairingGrantStore, { - ...pairingLinks, - subscribeChanges: pairingLinks.subscribeChanges.pipe( - Effect.tap((subscription) => Queue.offer(acquired, subscription)), - ), - }), - Effect.provideService(SessionStore.SessionStore, { - ...sessions, - subscribeChanges: sessions.subscribeChanges.pipe( - Effect.tap((subscription) => Queue.offer(acquired, subscription)), - ), - listActive: () => - sessions.listActive().pipe( - Effect.tap(() => Deferred.succeed(snapshotRead, undefined)), - Effect.flatMap((snapshot) => - Deferred.await(releaseSnapshot).pipe( - Effect.andThen( - milestone === "snapshot failure" - ? Effect.fail(new SessionStore.ActiveSessionsListError({ cause: "test" })) - : Effect.succeed(snapshot), - ), - ), - ), - ), - }), - ); - const { responses, receive, protocol } = yield* makeProtocol; - const { write } = yield* startRpc( - Stream.unwrap( - snapshotAuth.streamAccess(current.sessionId).pipe( - Effect.map(({ snapshot, changes }) => - Stream.concat(rpcInitialItems([snapshot]), changes), - ), - Effect.mapError((error) => new AuthAccessStreamError({ message: error.message })), - ), - ), - { responses, receive, protocol }, - ); - yield* write(0, { - _tag: "Request", - id: "1", - tag: WS_METHODS.subscribeAuthAccess, - payload: {}, - headers: [], - }); - const linkSubscription = yield* Queue.take(acquired); - const clientSubscription = yield* Queue.take(acquired); - yield* Deferred.await(snapshotRead); - assert.isFalse(linkSubscription.shutdownFlag.current); - assert.isFalse(clientSubscription.shutdownFlag.current); - if (milestone !== "snapshot read") { - yield* Deferred.succeed(releaseSnapshot, undefined); - if (milestone !== "snapshot failure") yield* takeEvents(responses); - } - if (milestone === "live tail") { - yield* sessions.markConnected(current.sessionId); - yield* write(0, { _tag: "Ack", requestId: "1" }); - const [event] = yield* takeEvents(responses); - assert.equal(event?.type, "clientUpserted"); - } - if (milestone !== "snapshot failure") { - yield* write(0, { _tag: "Interrupt", requestId: "1" }); - } - assert.equal((yield* Queue.take(responses))._tag, "Exit"); - assert.isTrue(linkSubscription.shutdownFlag.current); - assert.isTrue(clientSubscription.shutdownFlag.current); - }).pipe(Effect.scoped, Effect.provide(Layer.mergeAll(layerAuth, RpcSerialization.layerJson))), - ); -}); + const snapshot = yield* Queue.take(responses); + assert.equal(snapshot.type, "snapshot"); + if (snapshot.type !== "snapshot") return yield* Effect.die("Missing initial snapshot"); + assert.isFalse(snapshot.payload.clientSessions[0]?.connected); + yield* sessions.markConnected(current.sessionId); + yield* acknowledge; + yield* Queue.take(subscribed); + yield* Queue.take(subscribed); + + // A later change reaches either implementation, so a handler that dropped + // the first change fails here instead of waiting for a timeout. + yield* sessions.markDisconnected(current.sessionId); + const missed = yield* Queue.take(responses); + assert.equal(missed.revision, 2); + assert.equal(missed.type, "clientUpserted"); + if (missed.type === "clientUpserted") assert.isTrue(missed.payload.connected); + yield* acknowledge; + const latest = yield* Queue.take(responses); + assert.equal(latest.revision, 3); + assert.equal(latest.type, "clientUpserted"); + if (latest.type === "clientUpserted") assert.isFalse(latest.payload.connected); + + yield* server.write(0, { _tag: "Interrupt", requestId, interruptors: [] }); + yield* Deferred.await(exited); + yield* Queue.take(released); + yield* Queue.take(released); + }).pipe( + Effect.provide( + Layer.mergeAll( + Layer.fresh(layerAuth), + Layer.mock(PreviewAutomationBroker.PreviewAutomationBroker)({}), + Layer.mock(ServerBrowser.ServerBrowser)({}), + ), + ), + Effect.scoped, + ), +); diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 702ded435e984..988ef902ebcdf 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -24,10 +24,13 @@ import { DEFAULT_AUTOMATIC_GIT_FETCH_INTERVAL, AcpRegistryOperationError, CommandId, + authScopeResponse, AuthAccessStreamError, + type AuthAccessStreamEvent, AuthOrchestrationOperateScope, type AuthEnvironmentScope, type ScheduledTaskListResult, + AuthSessionId, ClientConnectionMethod, ClientDeviceType, ClientOs, @@ -238,6 +241,7 @@ import * as SourceControlProviderRegistry from "./sourceControl/SourceControlPro import * as GitVcsDriver from "./vcs/GitVcsDriver.ts"; import * as VcsDriverRegistry from "./vcs/VcsDriverRegistry.ts"; import * as VcsProjectConfig from "./vcs/VcsProjectConfig.ts"; +import * as PairingGrantStore from "./auth/PairingGrantStore.ts"; import * as SessionStore from "./auth/SessionStore.ts"; import { failEnvironmentAuthInvalid, failEnvironmentInternal } from "./auth/http.ts"; import * as RelayClient from "@t3tools/shared/relayClient"; @@ -556,6 +560,47 @@ const SHELL_RESUME_MAX_GAP = 1_000; // payload bytes of the range in SQL and reset with a snapshot past this budget. const ORCHESTRATION_REPLAY_PAYLOAD_BUDGET_BYTES = 8 * 1024 * 1024; +function toAuthAccessStreamEvent( + change: PairingGrantStore.BootstrapCredentialChange | SessionStore.SessionCredentialChange, + revision: number, + currentSessionId: AuthSessionId, +): AuthAccessStreamEvent { + switch (change.type) { + case "pairingLinkUpserted": + return { + version: 1, + revision, + type: "pairingLinkUpserted", + payload: { ...change.pairingLink, ...authScopeResponse(change.pairingLink.scopes) }, + }; + case "pairingLinkRemoved": + return { + version: 1, + revision, + type: "pairingLinkRemoved", + payload: { id: change.id }, + }; + case "clientUpserted": + return { + version: 1, + revision, + type: "clientUpserted", + payload: { + ...change.clientSession, + ...authScopeResponse(change.clientSession.scopes), + current: change.clientSession.sessionId === currentSessionId, + }, + }; + case "clientRemoved": + return { + version: 1, + revision, + type: "clientRemoved", + payload: { sessionId: change.sessionId }, + }; + } +} + const isClientSurface = Schema.is(ClientSurface); const isClientConnectionMethod = Schema.is(ClientConnectionMethod); const isClientDeviceType = Schema.is(ClientDeviceType); @@ -1152,7 +1197,7 @@ export const subscribeOrchestrationV2Shell = Effect.fn("ws.orchestrationV2.subsc }, ); -const layerWsRpc = ( +export const layerWsRpc = ( currentSession: EnvironmentAuth.AuthenticatedSession, clientOrigin: OrchestrationClientOrigin, clientAnalyticsProps: Readonly>, @@ -1291,6 +1336,8 @@ const layerWsRpc = ( const sourceControlRepositories = yield* SourceControlRepositoryService.SourceControlRepositoryService; const withPullRequestViewer = pullRequests.withRoutingCredential; + const bootstrapCredentials = yield* PairingGrantStore.PairingGrantStore; + const sessions = yield* SessionStore.SessionStore; const processDiagnostics = yield* ProcessDiagnostics.ProcessDiagnostics; const hostResources = yield* HostResources.HostResources; const processResourceMonitor = yield* ProcessResourceMonitor.ProcessResourceMonitor; @@ -1635,6 +1682,19 @@ const layerWsRpc = ( yield* providerRegistry.refreshInstance(input.instanceId); return { disabled: true } as const; }); + const loadAuthAccessSnapshot = () => + Effect.all({ + pairingLinks: serverAuth.listPairingLinks(), + clientSessions: serverAuth.listClientSessions(currentSessionId), + }).pipe( + Effect.mapError( + (error) => + new AuthAccessStreamError({ + message: error.message, + }), + ), + ); + const loadServerConfig = (options: { readonly usageLimitsCommand: boolean }) => Effect.gen(function* () { const keybindingsConfig = yield* keybindings.loadConfigState; @@ -3046,12 +3106,38 @@ const layerWsRpc = ( ), [WS_METHODS.subscribeAuthAccess]: (_input) => Stream.unwrap( - serverAuth.streamAccess(currentSessionId).pipe( - Effect.map(({ snapshot, changes }) => - Stream.concat(rpcInitialItems([snapshot]), changes), - ), - Effect.mapError((error) => new AuthAccessStreamError({ message: error.message })), - ), + Effect.gen(function* () { + // Buffer changes through the snapshot reads and its RPC ACK wait. + const pairingLinkChanges = yield* bootstrapCredentials.subscribeChanges; + const clientChanges = yield* sessions.subscribeChanges; + const initialSnapshot = yield* loadAuthAccessSnapshot(); + const revisionRef = yield* Ref.make(1); + const accessChanges: Stream.Stream< + PairingGrantStore.BootstrapCredentialChange | SessionStore.SessionCredentialChange + > = Stream.merge(pairingLinkChanges, clientChanges); + + const liveEvents: Stream.Stream = accessChanges.pipe( + Stream.mapEffect((change) => + Ref.updateAndGet(revisionRef, (revision) => revision + 1).pipe( + Effect.map((revision) => + toAuthAccessStreamEvent(change, revision, currentSessionId), + ), + ), + ), + ); + + return Stream.concat( + rpcInitialItems([ + { + version: 1 as const, + revision: 1, + type: "snapshot" as const, + payload: initialSnapshot, + }, + ]), + liveEvents, + ); + }), ), [WS_METHODS.subscribeBackgroundPolicy]: (_input) => Stream.unwrap( From 2a93049a14f2cc9b0dd4496a388bf5e6523aa4db Mon Sep 17 00:00:00 2001 From: Julius Marminge <51714798+juliusmarminge@users.noreply.github.com> Date: Mon, 19 Oct 2026 18:24:25 -0700 Subject: [PATCH 4/8] fix(server): hold auth access subscriptions through the snapshot read The Connections access stream could drop a session or pairing-link change made while its initial snapshot was being read. Both store subscriptions are now plain scoped accessors acquired before the snapshot, and the regression test gates the snapshot read so the change lands inside that window and crosses the ACK wait. Co-Authored-By: Claude Fable 5.1 --- apps/server/src/auth/PairingGrantStore.ts | 4 +-- apps/server/src/auth/SessionStore.ts | 4 +-- apps/server/src/ws.authAccess.test.ts | 32 ++++++++++++++++++----- 3 files changed, 28 insertions(+), 12 deletions(-) diff --git a/apps/server/src/auth/PairingGrantStore.ts b/apps/server/src/auth/PairingGrantStore.ts index 4217e5afa35a2..5f17bbf1e0fe5 100644 --- a/apps/server/src/auth/PairingGrantStore.ts +++ b/apps/server/src/auth/PairingGrantStore.ts @@ -617,9 +617,7 @@ export const make = Effect.gen(function* () { get streamChanges() { return Stream.fromPubSub(changesPubSub); }, - get subscribeChanges() { - return PubSub.subscribe(changesPubSub).pipe(Effect.map(Stream.fromSubscription)); - }, + subscribeChanges: PubSub.subscribe(changesPubSub).pipe(Effect.map(Stream.fromSubscription)), revoke, consume, }); diff --git a/apps/server/src/auth/SessionStore.ts b/apps/server/src/auth/SessionStore.ts index 17487c8e1aad3..733b485910ebb 100644 --- a/apps/server/src/auth/SessionStore.ts +++ b/apps/server/src/auth/SessionStore.ts @@ -1090,9 +1090,7 @@ export const make = Effect.gen(function* () { get streamChanges() { return Stream.fromPubSub(changesPubSub); }, - get subscribeChanges() { - return PubSub.subscribe(changesPubSub).pipe(Effect.map(Stream.fromSubscription)); - }, + subscribeChanges: PubSub.subscribe(changesPubSub).pipe(Effect.map(Stream.fromSubscription)), revoke, revokeAllExcept, markConnected, diff --git a/apps/server/src/ws.authAccess.test.ts b/apps/server/src/ws.authAccess.test.ts index 86e7cc8759a41..200fd99778610 100644 --- a/apps/server/src/ws.authAccess.test.ts +++ b/apps/server/src/ws.authAccess.test.ts @@ -1,7 +1,7 @@ import * as NodeServices from "@effect/platform-node/NodeServices"; import { assert, it } from "@effect/vitest"; import { - AuthAdministrativeScopes, + AuthStandardClientScopes, type AuthAccessStreamEvent, WS_METHODS, WsRpcGroup, @@ -37,6 +37,7 @@ import * as GitWorkflowService from "./git/GitWorkflowService.ts"; import * as Keybindings from "./keybindings.ts"; import * as McpAppRequests from "./mcpApps/McpAppRequests.ts"; import * as OtlpEndpointCheck from "./observability/OtlpEndpointCheck.ts"; +import * as CommandOutputHub from "./orchestration-v2/CommandOutputHub.ts"; import * as Orchestrator from "./orchestration-v2/Orchestrator.ts"; import * as ProjectStore from "./orchestration-v2/ProjectStore.ts"; import * as ProviderSessionManager from "./orchestration-v2/ProviderSessionManager.ts"; @@ -44,6 +45,7 @@ import * as PullRequestSyncReactor from "./orchestration-v2/PullRequestSyncReact import * as ThreadLaunchService from "./orchestration-v2/ThreadLaunchService.ts"; import * as ThreadManagement from "./orchestration-v2/ThreadManagementService.ts"; import * as ThreadSearch from "./orchestration-v2/ThreadSearch.ts"; +import * as DatabaseStorage from "./persistence/DatabaseStorage.ts"; import * as OrchestrationEventStore from "./persistence/OrchestrationEventStore.ts"; import * as SqlitePersistence from "./persistence/Sqlite.ts"; import * as PreviewAutomationBroker from "./mcp/PreviewAutomationBroker.ts"; @@ -116,6 +118,8 @@ const layerUnusedAuthAccessRpcServices = Layer.mergeAll( Layer.mock(BackgroundPolicy.BackgroundPolicy)({}), Layer.mock(CheckpointDiffQuery.CheckpointDiffQuery)({}), Layer.mock(CodexInstallation.CodexInstallation)({ managedDirectory: "/unused" }), + Layer.mock(CommandOutputHub.CommandOutputHub)({}), + Layer.mock(DatabaseStorage.DatabaseStorage)({}), Layer.mock(DeviceService.DeviceService)({}), Layer.mock(DirectEndpoints.DirectEndpoints)({}), Layer.mock(EnvironmentTheme.EnvironmentThemeService)({}), @@ -189,7 +193,7 @@ const layerAuth = EnvironmentAuth.layer.pipe( ); it.effect( - "keeps an access change made before the snapshot ACK, then releases both subscriptions", + "keeps an access change made while the snapshot loads, then releases both subscriptions", () => Effect.gen(function* () { const auth = yield* EnvironmentAuth.EnvironmentAuth; @@ -207,6 +211,20 @@ it.effect( ); const pairingLinkChanges = tracked(pairingLinks.subscribeChanges); const clientChanges = tracked(sessions.subscribeChanges); + const snapshotRead = yield* Deferred.make(); + const releaseSnapshot = yield* Deferred.make(); + // Hold the snapshot after its session read so a change lands between the + // read and the snapshot reaching the client. + const listClientSessions: typeof auth.listClientSessions = (sessionId) => + auth + .listClientSessions(sessionId) + .pipe( + Effect.tap(() => + Deferred.succeed(snapshotRead, undefined).pipe( + Effect.andThen(Deferred.await(releaseSnapshot)), + ), + ), + ); const responses = yield* Queue.unbounded(); const exited = yield* Deferred.make(); const server = yield* RpcServer.makeNoSerialization(group, { @@ -227,7 +245,7 @@ it.effect( sessionId: current.sessionId, subject: "current-client", method: "browser-session-cookie", - scopes: AuthAdministrativeScopes, + scopes: AuthStandardClientScopes, }, {}, {}, @@ -237,7 +255,7 @@ it.effect( Layer.provide( Layer.mergeAll( layerUnusedAuthAccessRpcServices, - Layer.succeed(EnvironmentAuth.EnvironmentAuth, auth), + Layer.succeed(EnvironmentAuth.EnvironmentAuth, { ...auth, listClientSessions }), Layer.succeed(PairingGrantStore.PairingGrantStore, { ...pairingLinks, streamChanges: Stream.unwrap(pairingLinkChanges), @@ -251,7 +269,7 @@ it.effect( ), ), ), - RpcAuthorization.layer(AuthAdministrativeScopes), + RpcAuthorization.layer(AuthStandardClientScopes), ), ), ); @@ -265,11 +283,13 @@ it.effect( headers: Headers.empty, }); + yield* Deferred.await(snapshotRead); + yield* sessions.markConnected(current.sessionId); + yield* Deferred.succeed(releaseSnapshot, undefined); const snapshot = yield* Queue.take(responses); assert.equal(snapshot.type, "snapshot"); if (snapshot.type !== "snapshot") return yield* Effect.die("Missing initial snapshot"); assert.isFalse(snapshot.payload.clientSessions[0]?.connected); - yield* sessions.markConnected(current.sessionId); yield* acknowledge; yield* Queue.take(subscribed); yield* Queue.take(subscribed); From a7f9bdbf08fb09a7b7037ecb3764b265c222bb18 Mon Sep 17 00:00:00 2001 From: SAPHID Date: Sat, 31 Oct 2026 07:28:14 +0000 Subject: [PATCH 5/8] test(server): provide current services to auth access race fixture --- apps/server/src/ws.authAccess.test.ts | 19 +++++++++++++++++++ 1 file changed, 19 insertions(+) diff --git a/apps/server/src/ws.authAccess.test.ts b/apps/server/src/ws.authAccess.test.ts index 200fd99778610..d8b26e0a43d97 100644 --- a/apps/server/src/ws.authAccess.test.ts +++ b/apps/server/src/ws.authAccess.test.ts @@ -1,3 +1,12 @@ +import * as SecretResolver from "@t3tools/secret-provider-core/server/SecretResolver"; +import * as DocumentText from "./provider/DocumentText.ts"; +import * as ScratchThreadPromotion from "./project/ScratchThreadPromotion.ts"; +import * as PeerLinkRequests from "./peer/PeerLinkRequests.ts"; +import * as PeerLinks from "./peer/PeerLinks.ts"; +import * as GuestShareService from "./guest/GuestShareService.ts"; +import * as NpmRelease from "./cloud/npmRelease.ts"; +import * as SourceControlSignIn from "./sourceControl/SourceControlSignIn.ts"; +import * as TriageService from "./triage/TriageService.ts"; import * as NodeServices from "@effect/platform-node/NodeServices"; import { assert, it } from "@effect/vitest"; import { @@ -109,6 +118,16 @@ const group = WsRpcGroup.omit( // Build the production RPC layer; unrelated capabilities fail if this test calls them. const layerUnusedAuthAccessRpcServices = Layer.mergeAll( + Layer.mock(SecretResolver.SecretResolver)({}), + Layer.mock(DocumentText.DocumentText)({}), + Layer.mock(ScratchThreadPromotion.ScratchThreadPromotion)({}), + Layer.mock(PeerLinkRequests.PeerLinkRequests)({}), + Layer.mock(PeerLinks.PeerLinks)({}), + Layer.mock(GuestShareService.GuestShareService)({}), + Layer.mock(NpmRelease.NpmRelease)({}), + Layer.mock(SourceControlSignIn.SourceControlSignIn)({}), + Layer.mock(TriageService.TriageService)({}), + Layer.mock(AcpRegistryRuntimeCoordinator.AcpRegistryRuntimeCoordinator)({}), Layer.mock(AcpRegistrySupport.AcpRegistryCatalog)({}), Layer.mock(AgentSessionImporter.AgentSessionImporter)({}), From 135e70b3508f6dabf1f84917424105686818f59b Mon Sep 17 00:00:00 2001 From: SAPHID Date: Sat, 31 Oct 2026 07:32:36 +0000 Subject: [PATCH 6/8] test(server): adapt auth stream fixture to current browser services --- apps/server/src/ws.authAccess.test.ts | 7 ++----- 1 file changed, 2 insertions(+), 5 deletions(-) diff --git a/apps/server/src/ws.authAccess.test.ts b/apps/server/src/ws.authAccess.test.ts index d8b26e0a43d97..b7153bbfc9243 100644 --- a/apps/server/src/ws.authAccess.test.ts +++ b/apps/server/src/ws.authAccess.test.ts @@ -57,7 +57,6 @@ import * as ThreadSearch from "./orchestration-v2/ThreadSearch.ts"; import * as DatabaseStorage from "./persistence/DatabaseStorage.ts"; import * as OrchestrationEventStore from "./persistence/OrchestrationEventStore.ts"; import * as SqlitePersistence from "./persistence/Sqlite.ts"; -import * as PreviewAutomationBroker from "./mcp/PreviewAutomationBroker.ts"; import * as PreviewManager from "./preview/Manager.ts"; import * as PortScanner from "./preview/PortScanner.ts"; import * as ServerBrowser from "./preview/ServerBrowser.ts"; @@ -152,7 +151,7 @@ const layerUnusedAuthAccessRpcServices = Layer.mergeAll( Layer.mock(OrchestrationEventStore.OrchestrationEventStore)({}), Layer.mock(Orchestrator.OrchestratorV2)({}), Layer.mock(OtlpEndpointCheck.OtlpEndpointCheck)({}), - Layer.mock(PortScanner.PortDiscovery)({}), + Layer.mock(PortScanner.PortDiscovery)({ trackAgent: () => () => undefined }), Layer.mock(PreviewManager.PreviewManager)({}), Layer.mock(ProcessDiagnostics.ProcessDiagnostics)({}), Layer.mock(ProcessResourceMonitor.ProcessResourceMonitor)({}), @@ -267,9 +266,8 @@ it.effect( scopes: AuthStandardClientScopes, }, {}, - {}, - yield* PreviewAutomationBroker.PreviewAutomationBroker, yield* ServerBrowser.ServerBrowser, + () => Effect.void, ).pipe( Layer.provide( Layer.mergeAll( @@ -334,7 +332,6 @@ it.effect( Effect.provide( Layer.mergeAll( Layer.fresh(layerAuth), - Layer.mock(PreviewAutomationBroker.PreviewAutomationBroker)({}), Layer.mock(ServerBrowser.ServerBrowser)({}), ), ), From 2602ca2837ddb19820c89ab74b7fa442fc1e869f Mon Sep 17 00:00:00 2001 From: SAPHID Date: Sat, 31 Oct 2026 07:37:29 +0000 Subject: [PATCH 7/8] style(server): format auth access test services --- apps/server/src/ws.authAccess.test.ts | 5 +---- 1 file changed, 1 insertion(+), 4 deletions(-) diff --git a/apps/server/src/ws.authAccess.test.ts b/apps/server/src/ws.authAccess.test.ts index b7153bbfc9243..32e4ce5b59cd7 100644 --- a/apps/server/src/ws.authAccess.test.ts +++ b/apps/server/src/ws.authAccess.test.ts @@ -330,10 +330,7 @@ it.effect( yield* Queue.take(released); }).pipe( Effect.provide( - Layer.mergeAll( - Layer.fresh(layerAuth), - Layer.mock(ServerBrowser.ServerBrowser)({}), - ), + Layer.mergeAll(Layer.fresh(layerAuth), Layer.mock(ServerBrowser.ServerBrowser)({})), ), Effect.scoped, ), From 95bbd0c91b7d9c2cbebdbeac9c07760f33be129a Mon Sep 17 00:00:00 2001 From: SAPHID Date: Sat, 31 Oct 2026 13:50:09 +0000 Subject: [PATCH 8/8] test(auth): assert snapshot chunk boundary before acknowledgement --- apps/server/src/ws.authAccess.test.ts | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/apps/server/src/ws.authAccess.test.ts b/apps/server/src/ws.authAccess.test.ts index 32e4ce5b59cd7..92da48e6da47b 100644 --- a/apps/server/src/ws.authAccess.test.ts +++ b/apps/server/src/ws.authAccess.test.ts @@ -244,6 +244,7 @@ it.effect( ), ); const responses = yield* Queue.unbounded(); + const chunkSizes = yield* Queue.unbounded(); const exited = yield* Deferred.make(); const server = yield* RpcServer.makeNoSerialization(group, { ...WS_RPC_SERVER_OPTIONS, @@ -251,7 +252,7 @@ it.effect( response._tag === "Chunk" ? Effect.forEach(response.values, (event) => Queue.offer(responses, event), { discard: true, - }) + }).pipe(Effect.andThen(Queue.offer(chunkSizes, Array.from(response.values).length))) : response._tag === "Exit" ? Deferred.succeed(exited, undefined) : Effect.void, @@ -303,6 +304,7 @@ it.effect( yield* Deferred.await(snapshotRead); yield* sessions.markConnected(current.sessionId); yield* Deferred.succeed(releaseSnapshot, undefined); + assert.equal(yield* Queue.take(chunkSizes), 1); const snapshot = yield* Queue.take(responses); assert.equal(snapshot.type, "snapshot"); if (snapshot.type !== "snapshot") return yield* Effect.die("Missing initial snapshot");