From dc1d3f93202ccbd32b552e48bc32e4a39c9c774a Mon Sep 17 00:00:00 2001 From: danljungstrom Date: Sun, 2 Aug 2026 21:06:05 +0000 Subject: [PATCH 1/7] fix(server-presence): clear session active on explicit stop disconnect An accepted stop kills the runner, but the resulting publisher disconnect only recorded observer loss and left `active` set. Archive is gated on `active === false`, so it returned 409 session-active for the full 10-minute presence fence while the client retry budget is 75s, making archive-after-stop impossible to complete from the UI. Record intent only when a runner accepts a stop, and treat the disconnect that follows as the termination it is. A superseded fence still protects a successor publisher, and a failed stop records nothing so an incidental disconnect cannot end a live session. Co-Authored-By: Claude Opus 5 (1M context) --- apps/server/sources/app/api/socket.ts | 14 ++++++ .../api/socket/rpcHandler.integration.spec.ts | 28 +++++++++++ .../sources/app/api/socket/rpcHandler.ts | 38 ++++++++++++++- ...blisherPresence.sqlite.integration.spec.ts | 38 +++++++++++++++ .../app/presence/sessionPublisherPresence.ts | 47 +++++++++++++++++++ 5 files changed, 164 insertions(+), 1 deletion(-) diff --git a/apps/server/sources/app/api/socket.ts b/apps/server/sources/app/api/socket.ts index f91d52f4ea..4f78692645 100644 --- a/apps/server/sources/app/api/socket.ts +++ b/apps/server/sources/app/api/socket.ts @@ -349,6 +349,20 @@ export function startSocket(app: Fastify) { decrementWebSocketConnection(connection.connectionType); if (connection.connectionType === 'session-scoped') { void sessionPublisherPresence.forgetDisconnectedPublisher({ socket }).then(async (result) => { + // A disconnect that completes an explicit stop ends the session, so + // participants have to learn it went inactive now rather than when the + // presence timeout fence eventually expires. + if (result.status === 'closed' && 'participantCursors' in result) { + await publishSessionPublisherLifecycleUpdate({ + sessionId: connection.sessionId, + participantCursors: result.participantCursors, + active: false, + activeAt: result.activeAt.getTime(), + ...(result.projection ? { projection: result.projection } : {}), + ...(result.turnProjection ?? {}), + }); + return; + } if (result.status !== 'applied') return; await publishSessionPublisherLifecycleUpdate({ sessionId: connection.sessionId, diff --git a/apps/server/sources/app/api/socket/rpcHandler.integration.spec.ts b/apps/server/sources/app/api/socket/rpcHandler.integration.spec.ts index 4c98c841ab..64fe4c125b 100644 --- a/apps/server/sources/app/api/socket/rpcHandler.integration.spec.ts +++ b/apps/server/sources/app/api/socket/rpcHandler.integration.spec.ts @@ -658,6 +658,34 @@ describe("rpcHandler", () => { ); }); + it("records no explicit-stop intent when a session stop RPC has no reachable runner", async () => { + vi.resetModules(); + const { rpcHandler } = await import("./rpcHandler"); + const socket = createFakeSocket(); + const markExplicitStopRequested = vi.fn(); + + rpcHandler("user-1", socket as any, new Map() as any, new Map() as any, { + io: {} as any, + redisRegistry: { enabled: false }, + sessionPublisherPresence: { + captureExplicitMachineStop: vi.fn(), + finalizeExplicitMachineStop: vi.fn(), + markExplicitStopRequested, + } as any, + }); + + const handler = getSocketHandler(socket, SOCKET_RPC_EVENTS.CALL); + const callback = vi.fn(); + await handler({ method: `sess_1:${RPC_METHODS.KILL_SESSION}`, params: {} }, callback); + + // The stop never reached a runner, so nothing may later read a disconnect as an + // intentional termination — that would end a session whose runner is still alive. + expect(callback).toHaveBeenCalledWith( + expect.objectContaining({ ok: false, errorCode: RPC_ERROR_CODES.METHOD_NOT_AVAILABLE }), + ); + expect(markExplicitStopRequested).not.toHaveBeenCalled(); + }); + it("uses Redis RPC registry + io.emitWithAck when enabled", async () => { vi.resetModules(); const targetSocketId = "target-socket"; diff --git a/apps/server/sources/app/api/socket/rpcHandler.ts b/apps/server/sources/app/api/socket/rpcHandler.ts index c88ea3aabe..a41b196a25 100644 --- a/apps/server/sources/app/api/socket/rpcHandler.ts +++ b/apps/server/sources/app/api/socket/rpcHandler.ts @@ -130,6 +130,32 @@ function readExplicitMachineStopRequest(method: string, value: unknown): Readonl return { machineId, sessionId: trimmedSessionId }; } +/** + * Session-scoped stop (`:killSession`). The runner answers this one directly, + * so unlike the machine-scoped stop there is no result the server can fence on — the only + * later signal is the publisher disconnect, which needs the recorded intent to be read as + * an intentional termination rather than an incidental drop. + */ +function readSessionScopedStopSessionId(method: string): string | null { + const separatorIndex = method.indexOf(':'); + if (separatorIndex <= 0) return null; + if (method.slice(separatorIndex + 1) !== RPC_METHODS.KILL_SESSION) return null; + const sessionId = method.slice(0, separatorIndex).trim(); + if (!sessionId || sessionId.length > MAX_RPC_METHOD_NAME_LENGTH) return null; + return sessionId; +} + +/** + * A stop the runner accepted: either proven termination or an acknowledged request. + * A transport error, refusal, or missing method is not acceptance. + */ +function isAcceptedStopResponse(targetResponse: unknown): boolean { + const strict = StopSessionResultSchema.safeParse(targetResponse); + if (strict.success) return strict.data.status === "stopped" || strict.data.status === "requested"; + if (!targetResponse || typeof targetResponse !== "object") return false; + return (targetResponse as { success?: unknown }).success === true; +} + function revalidatePrivilegedRpcTargetCompatibility(socket: Socket, method: string) { if (!resolveSocketRpcProviderStartingMethod(method)) return null; const socketData = readHappierSocketData(socket); @@ -170,7 +196,7 @@ export function rpcHandler( redisRegistry: RpcRedisRegistryConfig; sessionPublisherPresence?: Pick< ReturnType, - "captureExplicitMachineStop" | "finalizeExplicitMachineStop" + "captureExplicitMachineStop" | "finalizeExplicitMachineStop" | "markExplicitStopRequested" >; }, ) { @@ -328,6 +354,7 @@ export function rpcHandler( let { targetUserId, targetSocket } = targetResolution; const explicitMachineStopRequest = readExplicitMachineStopRequest(method, rpcAuthorization); + const sessionScopedStopSessionId = readSessionScopedStopSessionId(method); if (method.endsWith(`:${RPC_METHODS.STOP_SESSION}`) && !explicitMachineStopRequest) { callback?.({ ok: false, @@ -351,6 +378,15 @@ export function rpcHandler( }; const forwardTargetResponse = async (targetResponse: unknown) => { const forwarded = forwardedRpcTargetResponse({ method, targetResponse }); + // Only a stop the runner accepted may explain a later disconnect. Recording + // intent for an attempt that failed would let an unrelated incidental + // disconnect end a session whose runner is still alive. + const acceptedStopSessionId = sessionScopedStopSessionId ?? explicitMachineStopRequest?.sessionId ?? null; + if (acceptedStopSessionId && isAcceptedStopResponse(targetResponse)) { + ctx.sessionPublisherPresence?.markExplicitStopRequested({ + sessionId: acceptedStopSessionId, + }); + } if (!explicitMachineStopRequest || explicitMachineStopCapture?.status !== "captured") { return forwarded; } diff --git a/apps/server/sources/app/presence/sessionPublisherPresence.sqlite.integration.spec.ts b/apps/server/sources/app/presence/sessionPublisherPresence.sqlite.integration.spec.ts index da8ba0444b..d0321fa390 100644 --- a/apps/server/sources/app/presence/sessionPublisherPresence.sqlite.integration.spec.ts +++ b/apps/server/sources/app/presence/sessionPublisherPresence.sqlite.integration.spec.ts @@ -454,6 +454,44 @@ describe("session publisher presence on SQLite", () => { }); }); + it("closes the exact publisher on disconnect after an explicit stop was requested", async () => { + const seeded = await seed(); + const presence = createSessionPublisherPresence({ now: () => new Date(seeded.fence.getTime() + 10) }); + const socket = {}; + const registered = await presence.registerPublisher({ + socket, + binding: seeded.binding, + completeActivitySnapshot: { state: "active", activeCount: 1 }, + }); + if (registered.status !== "registered") throw new Error("expected registration"); + + // An explicit stop was requested but the runner died before it could prove + // physical termination, so only the disconnect reaches the server. + presence.markExplicitStopRequested({ sessionId: seeded.binding.sessionId }); + + const disconnected = await presence.forgetDisconnectedPublisher({ socket }); + // A terminal close carries participant cursors; the bare already-closed marker + // does not, so the fanout is what proves the session actually ended here. + if (!("participantCursors" in disconnected)) throw new Error("expected a terminal close with fanout"); + expect(disconnected.status).toBe("closed"); + expect(disconnected.participantCursors.map((cursor) => cursor.accountId).sort()) + .toEqual(seeded.participantIds); + await expect(db.session.findUniqueOrThrow({ + where: { id: seeded.binding.sessionId }, + select: { + active: true, + lastActiveAt: true, + runtimeActivityState: true, + runtimeActivityActiveCount: true, + }, + })).resolves.toEqual({ + active: false, + lastActiveAt: registered.committedFence, + runtimeActivityState: "unknown", + runtimeActivityActiveCount: 0, + }); + }); + it("does not let a completed explicit stop close a successor publisher that registered meanwhile", async () => { const seeded = await seed(); let now = new Date(seeded.fence.getTime() + 10); diff --git a/apps/server/sources/app/presence/sessionPublisherPresence.ts b/apps/server/sources/app/presence/sessionPublisherPresence.ts index dfc8d41c45..f947224839 100644 --- a/apps/server/sources/app/presence/sessionPublisherPresence.ts +++ b/apps/server/sources/app/presence/sessionPublisherPresence.ts @@ -95,12 +95,45 @@ function publisherIntentKey(binding: PublisherBinding, snapshot: SessionRuntimeA class RegistrationContentionError extends Error {} +/** + * How long an explicit stop request stays able to explain a publisher disconnect. + * Comfortably longer than a stop round trip, far shorter than the presence timeout + * fence so a stale intent can never close an unrelated later publisher. + */ +const EXPLICIT_STOP_REQUEST_TTL_MS = 2 * 60 * 1000; + export function createSessionPublisherPresence(options: Readonly<{ now?: () => Date }> = {}) { const now = options.now ?? (() => new Date()); const registrations = new WeakMap(); const registrationAttempts = new WeakMap(); const closeResults = new WeakMap>(); const operationTails = new WeakMap>(); + // Sessions with an explicit stop in flight. A stop that never proves physical + // termination still reaches the server only as a publisher disconnect, so the + // intent has to survive from the stop request until that disconnect arrives. + const explicitStopRequestedAtBySessionId = new Map(); + + const markExplicitStopRequested = (params: Readonly<{ sessionId: string }>): void => { + const at = now().getTime(); + // Intents are consumed by the disconnect they explain, but a stop whose publisher + // never disconnects leaves one behind. Sweep on write so the map cannot grow + // unbounded on a long-lived server. + for (const [sessionId, requestedAt] of explicitStopRequestedAtBySessionId) { + if (at - requestedAt > EXPLICIT_STOP_REQUEST_TTL_MS) { + explicitStopRequestedAtBySessionId.delete(sessionId); + } + } + explicitStopRequestedAtBySessionId.set(params.sessionId, at); + }; + + const consumeExplicitStopRequest = (sessionId: string): boolean => { + const requestedAt = explicitStopRequestedAtBySessionId.get(sessionId); + if (requestedAt === undefined) return false; + explicitStopRequestedAtBySessionId.delete(sessionId); + // A stop request only explains a disconnect that follows it closely. Beyond the + // window the disconnect is incidental and must not close a later publisher. + return now().getTime() - requestedAt <= EXPLICIT_STOP_REQUEST_TTL_MS; + }; const serialize = async (socket: object, operation: () => Promise): Promise => { const prior = operationTails.get(socket) ?? Promise.resolve(); @@ -490,11 +523,25 @@ export function createSessionPublisherPresence(options: Readonly<{ now?: () => D resolveCurrentPublisher, runAsCurrentPublisher, publishSnapshot, + markExplicitStopRequested, forgetDisconnectedPublisher: async (params: Readonly<{ socket: object }>) => await serialize(params.socket, async () => { try { const registration = registrations.get(params.socket); if (!registration) return { status: "unregistered" } as const; if (closeResults.has(params.socket)) return { status: "closed" } as const; + // An explicit stop is an intentional termination, so the disconnect it + // produces ends the session now. Without this the row stays active until + // the presence timeout fence expires, which blocks archive for that whole + // window. A superseded fence means a successor publisher owns the session, + // so fall through and only record observer loss. + if (consumeExplicitStopRequest(registration.binding.sessionId)) { + const closed = await closeBindingAtFence({ + binding: registration.binding, + committedFence: registration.committedFence, + mutationId: `explicit-stop-disconnect:${registration.committedFence.getTime()}`, + }); + if (closed.status === "closed") return closed; + } return await inTx(async (tx) => { const session = await tx.session.findUnique({ where: { id: registration.binding.sessionId }, From dd6eb494cef6c057bd01db139459c839bcbcf707 Mon Sep 17 00:00:00 2001 From: danljungstrom Date: Sun, 2 Aug 2026 21:40:55 +0000 Subject: [PATCH 2/7] fix(server-presence): persist explicit-stop intent on the session row The intent lived in a process-local Map inside `createSessionPublisherPresence`. With `redisRegistry.enabled` the stop RPC and the runner's publisher socket can be served by different instances, so the disconnect handler never saw the intent and the fix degraded to the pre-existing behaviour: archive stays 409 `session-active` for the full 10-minute presence fence. Move the intent to `Session.stopRequestedAt`. `markExplicitStopRequested` writes the timestamp; `forgetDisconnectedPublisher` reads and clears it inside the transaction it already opens, so the consume and the close commit together. `closeBindingAtFence` is split into an in-transaction owner plus a wrapper so the disconnect path reuses it without nesting a second transaction. The TTL is unchanged in meaning and is now a comparison against the persisted timestamp, which also retires the unbounded-map sweep. The intent write is awaited in `rpcHandler` because it is now durable: the publisher disconnect that reads it can arrive as soon as the runner begins tearing down. Co-Authored-By: Claude Opus 5 (1M context) --- .../migration.sql | 4 + .../migration.sql | 4 + apps/server/prisma/mysql/schema.prisma | 4 + apps/server/prisma/schema.prisma | 4 + .../migration.sql | 4 + apps/server/prisma/sqlite/schema.prisma | 4 + .../sources/app/api/socket/rpcHandler.ts | 5 +- ...blisherPresence.sqlite.integration.spec.ts | 36 +++++- .../app/presence/sessionPublisherPresence.ts | 109 +++++++++++------- 9 files changed, 130 insertions(+), 44 deletions(-) create mode 100644 apps/server/prisma/migrations/20260802220000_add_session_stop_requested_at/migration.sql create mode 100644 apps/server/prisma/mysql/migrations/20260802220000_add_session_stop_requested_at/migration.sql create mode 100644 apps/server/prisma/sqlite/migrations/20260802220000_add_session_stop_requested_at/migration.sql diff --git a/apps/server/prisma/migrations/20260802220000_add_session_stop_requested_at/migration.sql b/apps/server/prisma/migrations/20260802220000_add_session_stop_requested_at/migration.sql new file mode 100644 index 0000000000..6cbdc9b8a6 --- /dev/null +++ b/apps/server/prisma/migrations/20260802220000_add_session_stop_requested_at/migration.sql @@ -0,0 +1,4 @@ +-- Durable explicit-stop intent: the disconnect that completes a stop can land on a +-- different server instance than the RPC that accepted it. + +ALTER TABLE "Session" ADD COLUMN "stopRequestedAt" TIMESTAMP(3); diff --git a/apps/server/prisma/mysql/migrations/20260802220000_add_session_stop_requested_at/migration.sql b/apps/server/prisma/mysql/migrations/20260802220000_add_session_stop_requested_at/migration.sql new file mode 100644 index 0000000000..c0cec2be44 --- /dev/null +++ b/apps/server/prisma/mysql/migrations/20260802220000_add_session_stop_requested_at/migration.sql @@ -0,0 +1,4 @@ +-- Durable explicit-stop intent: the disconnect that completes a stop can land on a +-- different server instance than the RPC that accepted it. + +ALTER TABLE `Session` ADD COLUMN `stopRequestedAt` DATETIME(3) NULL; diff --git a/apps/server/prisma/mysql/schema.prisma b/apps/server/prisma/mysql/schema.prisma index 8a904248a0..02c8dd3c8d 100644 --- a/apps/server/prisma/mysql/schema.prisma +++ b/apps/server/prisma/mysql/schema.prisma @@ -321,6 +321,10 @@ model Session { runtimeActivityRevision BigInt @default(0) meaningfulActivityAt DateTime? active Boolean @default(false) + // When a runner accepted an explicit stop. The disconnect that completes the stop can + // reach a different server instance than the RPC that requested it, so the intent has + // to live on the session row rather than in the accepting instance's memory. + stopRequestedAt DateTime? archivedAt DateTime? lastActiveAt DateTime @default(now()) createdAt DateTime @default(now()) diff --git a/apps/server/prisma/schema.prisma b/apps/server/prisma/schema.prisma index 2893fd7a4e..eec15563ff 100644 --- a/apps/server/prisma/schema.prisma +++ b/apps/server/prisma/schema.prisma @@ -320,6 +320,10 @@ model Session { runtimeActivityRevision BigInt @default(0) meaningfulActivityAt DateTime? active Boolean @default(false) + // When a runner accepted an explicit stop. The disconnect that completes the stop can + // reach a different server instance than the RPC that requested it, so the intent has + // to live on the session row rather than in the accepting instance's memory. + stopRequestedAt DateTime? archivedAt DateTime? lastActiveAt DateTime @default(now()) createdAt DateTime @default(now()) diff --git a/apps/server/prisma/sqlite/migrations/20260802220000_add_session_stop_requested_at/migration.sql b/apps/server/prisma/sqlite/migrations/20260802220000_add_session_stop_requested_at/migration.sql new file mode 100644 index 0000000000..d91de5e28b --- /dev/null +++ b/apps/server/prisma/sqlite/migrations/20260802220000_add_session_stop_requested_at/migration.sql @@ -0,0 +1,4 @@ +-- Durable explicit-stop intent: the disconnect that completes a stop can land on a +-- different server instance than the RPC that accepted it. + +ALTER TABLE "Session" ADD COLUMN "stopRequestedAt" DATETIME; diff --git a/apps/server/prisma/sqlite/schema.prisma b/apps/server/prisma/sqlite/schema.prisma index 46be91f1b8..83d6c585f4 100644 --- a/apps/server/prisma/sqlite/schema.prisma +++ b/apps/server/prisma/sqlite/schema.prisma @@ -321,6 +321,10 @@ model Session { runtimeActivityRevision BigInt @default(0) meaningfulActivityAt DateTime? active Boolean @default(false) + // When a runner accepted an explicit stop. The disconnect that completes the stop can + // reach a different server instance than the RPC that requested it, so the intent has + // to live on the session row rather than in the accepting instance's memory. + stopRequestedAt DateTime? archivedAt DateTime? lastActiveAt DateTime @default(now()) createdAt DateTime @default(now()) diff --git a/apps/server/sources/app/api/socket/rpcHandler.ts b/apps/server/sources/app/api/socket/rpcHandler.ts index a41b196a25..81fef5a7d8 100644 --- a/apps/server/sources/app/api/socket/rpcHandler.ts +++ b/apps/server/sources/app/api/socket/rpcHandler.ts @@ -383,7 +383,10 @@ export function rpcHandler( // disconnect end a session whose runner is still alive. const acceptedStopSessionId = sessionScopedStopSessionId ?? explicitMachineStopRequest?.sessionId ?? null; if (acceptedStopSessionId && isAcceptedStopResponse(targetResponse)) { - ctx.sessionPublisherPresence?.markExplicitStopRequested({ + // The intent is durable, so it has to be committed before the caller + // learns the stop was accepted — the publisher disconnect that reads it + // can arrive as soon as the runner starts tearing down. + await ctx.sessionPublisherPresence?.markExplicitStopRequested({ sessionId: acceptedStopSessionId, }); } diff --git a/apps/server/sources/app/presence/sessionPublisherPresence.sqlite.integration.spec.ts b/apps/server/sources/app/presence/sessionPublisherPresence.sqlite.integration.spec.ts index d0321fa390..b088140f42 100644 --- a/apps/server/sources/app/presence/sessionPublisherPresence.sqlite.integration.spec.ts +++ b/apps/server/sources/app/presence/sessionPublisherPresence.sqlite.integration.spec.ts @@ -467,7 +467,7 @@ describe("session publisher presence on SQLite", () => { // An explicit stop was requested but the runner died before it could prove // physical termination, so only the disconnect reaches the server. - presence.markExplicitStopRequested({ sessionId: seeded.binding.sessionId }); + await presence.markExplicitStopRequested({ sessionId: seeded.binding.sessionId }); const disconnected = await presence.forgetDisconnectedPublisher({ socket }); // A terminal close carries participant cursors; the bare already-closed marker @@ -492,6 +492,40 @@ describe("session publisher presence on SQLite", () => { }); }); + it("closes the publisher on disconnect when another server instance accepted the stop", async () => { + const seeded = await seed(); + const now = () => new Date(seeded.fence.getTime() + 10); + // With a Redis RPC registry the stop call and the runner's publisher socket land on + // different server instances, so the intent only reaches the disconnect if it is + // durable rather than held in the accepting instance's memory. + const publisherInstance = createSessionPublisherPresence({ now }); + const rpcInstance = createSessionPublisherPresence({ now }); + const socket = {}; + const registered = await publisherInstance.registerPublisher({ + socket, + binding: seeded.binding, + completeActivitySnapshot: { state: "active", activeCount: 1 }, + }); + if (registered.status !== "registered") throw new Error("expected registration"); + + await rpcInstance.markExplicitStopRequested({ sessionId: seeded.binding.sessionId }); + + const disconnected = await publisherInstance.forgetDisconnectedPublisher({ socket }); + if (!("participantCursors" in disconnected)) throw new Error("expected a terminal close with fanout"); + expect(disconnected.status).toBe("closed"); + expect(disconnected.participantCursors.map((cursor) => cursor.accountId).sort()) + .toEqual(seeded.participantIds); + await expect(db.session.findUniqueOrThrow({ + where: { id: seeded.binding.sessionId }, + select: { active: true, lastActiveAt: true, stopRequestedAt: true }, + })).resolves.toEqual({ + active: false, + lastActiveAt: registered.committedFence, + // Consumed by the disconnect it explained, so it cannot end a later publisher. + stopRequestedAt: null, + }); + }); + it("does not let a completed explicit stop close a successor publisher that registered meanwhile", async () => { const seeded = await seed(); let now = new Date(seeded.fence.getTime() + 10); diff --git a/apps/server/sources/app/presence/sessionPublisherPresence.ts b/apps/server/sources/app/presence/sessionPublisherPresence.ts index f947224839..9a50f4a42e 100644 --- a/apps/server/sources/app/presence/sessionPublisherPresence.ts +++ b/apps/server/sources/app/presence/sessionPublisherPresence.ts @@ -19,7 +19,7 @@ import { writeSessionRuntimeActivityObserverLossInTx, writeSessionRuntimeActivityProjectionInTx, } from "@/app/session/runtimeActivity/writeProjection"; -import { inTx } from "@/storage/inTx"; +import { inTx, type Tx } from "@/storage/inTx"; import { blockInheritedProviderDeliveryClaims } from "@/app/session/pending/providerDeliveryClaimStaleness"; import { applyLatestSessionTurnEndInTx } from "@/app/session/sessionWriteService"; @@ -102,37 +102,45 @@ class RegistrationContentionError extends Error {} */ const EXPLICIT_STOP_REQUEST_TTL_MS = 2 * 60 * 1000; +/** + * Read-and-clear the durable stop intent for the disconnect that is being handled. + * Clearing is unconditional: an intent older than the window no longer explains a + * disconnect, and leaving it behind would let a much later incidental drop end a + * session whose runner is alive. + */ +async function consumeExplicitStopRequestInTx(params: Readonly<{ + tx: Tx; + sessionId: string; + stopRequestedAt: Date | null; + at: Date; +}>): Promise { + if (params.stopRequestedAt === null) return false; + await params.tx.session.updateMany({ + where: { id: params.sessionId, stopRequestedAt: params.stopRequestedAt }, + data: { stopRequestedAt: null }, + }); + return params.at.getTime() - params.stopRequestedAt.getTime() <= EXPLICIT_STOP_REQUEST_TTL_MS; +} + export function createSessionPublisherPresence(options: Readonly<{ now?: () => Date }> = {}) { const now = options.now ?? (() => new Date()); const registrations = new WeakMap(); const registrationAttempts = new WeakMap(); const closeResults = new WeakMap>(); const operationTails = new WeakMap>(); - // Sessions with an explicit stop in flight. A stop that never proves physical - // termination still reaches the server only as a publisher disconnect, so the - // intent has to survive from the stop request until that disconnect arrives. - const explicitStopRequestedAtBySessionId = new Map(); - - const markExplicitStopRequested = (params: Readonly<{ sessionId: string }>): void => { - const at = now().getTime(); - // Intents are consumed by the disconnect they explain, but a stop whose publisher - // never disconnects leaves one behind. Sweep on write so the map cannot grow - // unbounded on a long-lived server. - for (const [sessionId, requestedAt] of explicitStopRequestedAtBySessionId) { - if (at - requestedAt > EXPLICIT_STOP_REQUEST_TTL_MS) { - explicitStopRequestedAtBySessionId.delete(sessionId); - } - } - explicitStopRequestedAtBySessionId.set(params.sessionId, at); - }; - const consumeExplicitStopRequest = (sessionId: string): boolean => { - const requestedAt = explicitStopRequestedAtBySessionId.get(sessionId); - if (requestedAt === undefined) return false; - explicitStopRequestedAtBySessionId.delete(sessionId); - // A stop request only explains a disconnect that follows it closely. Beyond the - // window the disconnect is incidental and must not close a later publisher. - return now().getTime() - requestedAt <= EXPLICIT_STOP_REQUEST_TTL_MS; + // A stop that never proves physical termination reaches the server only as a publisher + // disconnect, and with a Redis RPC registry that disconnect can land on a different + // instance than the accepting call. The intent lives on the session row so it survives + // the hop from stop request to disconnect. + const markExplicitStopRequested = async (params: Readonly<{ sessionId: string }>): Promise => { + const at = now(); + await inTx(async (tx) => { + await tx.session.updateMany({ + where: { id: params.sessionId }, + data: { stopRequestedAt: at }, + }); + }); }; const serialize = async (socket: object, operation: () => Promise): Promise => { @@ -230,11 +238,13 @@ export function createSessionPublisherPresence(options: Readonly<{ now?: () => D } }; - const closeBindingAtFence = async (params: Readonly<{ + const closeBindingAtFenceInTx = async (params: Readonly<{ + tx: Tx; binding: PublisherBinding; committedFence: Date; mutationId: string; - }>): Promise => await inTx(async (tx): Promise => { + }>): Promise => { + const tx = params.tx; const session = await tx.session.findUnique({ where: { id: params.binding.sessionId }, select: { active: true, archivedAt: true, lastActiveAt: true }, @@ -291,7 +301,15 @@ export function createSessionPublisherPresence(options: Readonly<{ now?: () => D } : {}), }; - }); + }; + + const closeBindingAtFence = async (params: Readonly<{ + binding: PublisherBinding; + committedFence: Date; + mutationId: string; + }>): Promise => await inTx( + async (tx) => await closeBindingAtFenceInTx({ tx, ...params }), + ); const captureExplicitMachineStop = async (params: Readonly<{ binding: PublisherBinding; @@ -529,29 +547,36 @@ export function createSessionPublisherPresence(options: Readonly<{ now?: () => D const registration = registrations.get(params.socket); if (!registration) return { status: "unregistered" } as const; if (closeResults.has(params.socket)) return { status: "closed" } as const; - // An explicit stop is an intentional termination, so the disconnect it - // produces ends the session now. Without this the row stays active until - // the presence timeout fence expires, which blocks archive for that whole - // window. A superseded fence means a successor publisher owns the session, - // so fall through and only record observer loss. - if (consumeExplicitStopRequest(registration.binding.sessionId)) { - const closed = await closeBindingAtFence({ - binding: registration.binding, - committedFence: registration.committedFence, - mutationId: `explicit-stop-disconnect:${registration.committedFence.getTime()}`, - }); - if (closed.status === "closed") return closed; - } return await inTx(async (tx) => { const session = await tx.session.findUnique({ where: { id: registration.binding.sessionId }, - select: { active: true, archivedAt: true, lastActiveAt: true }, + select: { active: true, archivedAt: true, lastActiveAt: true, stopRequestedAt: true }, }); if (!session) return { status: "rejected", reason: "not_found" } as const; if (!await hasCurrentSessionScopedMachineAccessInTx({ tx, ...registration.binding })) { return { status: "rejected", reason: "unauthorized" } as const; } if (session.archivedAt !== null) return { status: "rejected", reason: "archived" } as const; + // An explicit stop is an intentional termination, so the disconnect it + // produces ends the session now. Without this the row stays active until + // the presence timeout fence expires, which blocks archive for that whole + // window. A superseded fence means a successor publisher owns the session, + // so fall through and only record observer loss. + const explicitStopRequested = await consumeExplicitStopRequestInTx({ + tx, + sessionId: registration.binding.sessionId, + stopRequestedAt: session.stopRequestedAt, + at: now(), + }); + if (explicitStopRequested) { + const closed = await closeBindingAtFenceInTx({ + tx, + binding: registration.binding, + committedFence: registration.committedFence, + mutationId: `explicit-stop-disconnect:${registration.committedFence.getTime()}`, + }); + if (closed.status === "closed") return closed; + } if (session.lastActiveAt.getTime() !== registration.committedFence.getTime()) { return { status: "rejected", reason: "superseded" } as const; } From 41fcedccccd1b298ea3780e2395b46ec8dfd298a Mon Sep 17 00:00:00 2001 From: danljungstrom Date: Sun, 2 Aug 2026 22:02:18 +0000 Subject: [PATCH 3/7] fix(server-presence): invalidate stop intent when a successor publisher registers MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `stopRequestedAt` is keyed by session, not by publisher, so a stop that never killed its runner stayed readable after a successor took the session over. The successor's own incidental disconnect then consumed it and closed a session whose runner never agreed to stop — its fence was current, so nothing else rejected the close. A registering publisher is a new binding, so clear the intent in the update `registerOnce` already performs. No extra query, and the fence check keeps covering the opposite direction (a predecessor's stale fence). Co-Authored-By: Claude Opus 5 (1M context) --- ...blisherPresence.sqlite.integration.spec.ts | 39 +++++++++++++++++++ .../app/presence/sessionPublisherPresence.ts | 5 ++- 2 files changed, 43 insertions(+), 1 deletion(-) diff --git a/apps/server/sources/app/presence/sessionPublisherPresence.sqlite.integration.spec.ts b/apps/server/sources/app/presence/sessionPublisherPresence.sqlite.integration.spec.ts index b088140f42..f911fe48da 100644 --- a/apps/server/sources/app/presence/sessionPublisherPresence.sqlite.integration.spec.ts +++ b/apps/server/sources/app/presence/sessionPublisherPresence.sqlite.integration.spec.ts @@ -526,6 +526,45 @@ describe("session publisher presence on SQLite", () => { }); }); + it("does not let a successor publisher's disconnect consume a stop intent recorded for its predecessor", async () => { + const seeded = await seed(); + let now = new Date(seeded.fence.getTime() + 10); + const presence = createSessionPublisherPresence({ now: () => now }); + const predecessor = {}; + const first = await presence.registerPublisher({ + socket: predecessor, + binding: seeded.binding, + completeActivitySnapshot: { state: "active", activeCount: 1 }, + }); + if (first.status !== "registered") throw new Error("expected first registration"); + + // The stop was accepted but never killed the runner, so its intent is still on the + // row when a successor takes the session over. + await presence.markExplicitStopRequested({ sessionId: seeded.binding.sessionId }); + + now = new Date(first.committedFence.getTime() + 10); + const successorSocket = {}; + const successor = await presence.registerPublisher({ + socket: successorSocket, + binding: seeded.binding, + completeActivitySnapshot: { state: "active", activeCount: 1 }, + }); + if (successor.status !== "registered") throw new Error("expected successor registration"); + + // The successor's own disconnect is incidental. Reading the predecessor's intent + // here would end a session whose runner never agreed to stop. + const disconnected = await presence.forgetDisconnectedPublisher({ socket: successorSocket }); + expect(disconnected.status).not.toBe("closed"); + await expect(db.session.findUniqueOrThrow({ + where: { id: seeded.binding.sessionId }, + select: { active: true, lastActiveAt: true, stopRequestedAt: true }, + })).resolves.toEqual({ + active: true, + lastActiveAt: successor.committedFence, + stopRequestedAt: null, + }); + }); + it("does not let a completed explicit stop close a successor publisher that registered meanwhile", async () => { const seeded = await seed(); let now = new Date(seeded.fence.getTime() + 10); diff --git a/apps/server/sources/app/presence/sessionPublisherPresence.ts b/apps/server/sources/app/presence/sessionPublisherPresence.ts index 9a50f4a42e..830ef0604e 100644 --- a/apps/server/sources/app/presence/sessionPublisherPresence.ts +++ b/apps/server/sources/app/presence/sessionPublisherPresence.ts @@ -178,7 +178,10 @@ export function createSessionPublisherPresence(options: Readonly<{ now?: () => D archivedAt: null, lastActiveAt: session.lastActiveAt, }, - data: { active: true, lastActiveAt: committedFence }, + // A registering publisher is a new binding, so any stop intent left by a + // predecessor no longer explains anything. Clearing it here is what keeps + // the successor's own later disconnect from being read as that stop. + data: { active: true, lastActiveAt: committedFence, stopRequestedAt: null }, }); if (updated.count === 0) throw new RegistrationContentionError(); const participantCursors = await markSessionParticipantsChanged({ tx, sessionId: binding.sessionId }); From c4865579ef026f3fdd6ef2adb9386966caa808ca Mon Sep 17 00:00:00 2001 From: danljungstrom Date: Sun, 2 Aug 2026 22:45:26 +0000 Subject: [PATCH 4/7] fix(server-rpc): run the stop lifecycle on forwards without an acknowledgement MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `forwardTargetResponse` owns explicit-stop intent recording and machine-scoped stop finalization, but all three forward sites invoked it only inside `if (callback)`. Socket.IO permits a caller to emit without an acknowledgement, and such a stop was forwarded to the runner and accepted while the server recorded nothing — leaving the session `active` for the full presence fence. Evaluate the forwarded response on every successful forward and call back only when the caller supplied one. Note that `callback?.(await forward(...))` does not fix this: optional-call short-circuiting skips argument evaluation, so the lifecycle never runs. The result is bound to a local first. Not reachable from first-party clients today — every stop path uses `emitWithAck` — so this closes the contract rather than an observed incident. Co-Authored-By: Claude Opus 5 (1M context) --- .../api/socket/rpcHandler.integration.spec.ts | 31 +++++++++++++++++++ .../sources/app/api/socket/rpcHandler.ts | 21 +++++++------ 2 files changed, 43 insertions(+), 9 deletions(-) diff --git a/apps/server/sources/app/api/socket/rpcHandler.integration.spec.ts b/apps/server/sources/app/api/socket/rpcHandler.integration.spec.ts index 64fe4c125b..1882193545 100644 --- a/apps/server/sources/app/api/socket/rpcHandler.integration.spec.ts +++ b/apps/server/sources/app/api/socket/rpcHandler.integration.spec.ts @@ -686,6 +686,37 @@ describe("rpcHandler", () => { expect(markExplicitStopRequested).not.toHaveBeenCalled(); }); + it("records an accepted stop even when the caller sends no acknowledgement callback", async () => { + vi.resetModules(); + const { rpcHandler } = await import("./rpcHandler"); + const method = `sess_1:${RPC_METHODS.KILL_SESSION}`; + const markExplicitStopRequested = vi.fn(); + const targetEmitWithAck = vi.fn().mockResolvedValue({ status: "requested" }); + const targetSocket = createFakeSocket({ + id: "runner-socket", + timeout: vi.fn(() => ({ emitWithAck: targetEmitWithAck })) as any, + }); + const callerSocket = createFakeSocket({ id: "caller-socket" }); + + rpcHandler("user-1", callerSocket as any, new Map([[method, targetSocket]]) as any, new Map() as any, { + io: {} as any, + redisRegistry: { enabled: false }, + sessionPublisherPresence: { + captureExplicitMachineStop: vi.fn(), + finalizeExplicitMachineStop: vi.fn(), + markExplicitStopRequested, + } as any, + }); + + // Socket.IO lets a caller emit without an acknowledgement. The runner still accepted + // the stop, so the disconnect it produces has to stay explainable — otherwise the + // session holds `active` for the full presence fence and archive keeps returning 409. + await getSocketHandler(callerSocket, SOCKET_RPC_EVENTS.CALL)({ method, params: {} }); + + expect(targetEmitWithAck).toHaveBeenCalledTimes(1); + expect(markExplicitStopRequested).toHaveBeenCalledWith({ sessionId: "sess_1" }); + }); + it("uses Redis RPC registry + io.emitWithAck when enabled", async () => { vi.resetModules(); const targetSocketId = "target-socket"; diff --git a/apps/server/sources/app/api/socket/rpcHandler.ts b/apps/server/sources/app/api/socket/rpcHandler.ts index 81fef5a7d8..1050f6ba84 100644 --- a/apps/server/sources/app/api/socket/rpcHandler.ts +++ b/apps/server/sources/app/api/socket/rpcHandler.ts @@ -376,6 +376,9 @@ export function rpcHandler( } return allRpcListeners.get(targetUserId)?.get(method) ?? null; }; + // Owns the explicit-stop lifecycle — intent recording and machine-stop + // finalization — not just the shape of the caller's response. Every successful + // forward must run it, including one the caller made without an acknowledgement. const forwardTargetResponse = async (targetResponse: unknown) => { const forwarded = forwardedRpcTargetResponse({ method, targetResponse }); // Only a stop the runner accepted may explain a later disconnect. Recording @@ -526,9 +529,11 @@ export function rpcHandler( SOCKET_RPC_EVENTS.REQUEST, buildForwardedRequest(), ); - if (callback) { - callback(await forwardTargetResponse(response)); - } + // Evaluated before the optional call: `callback?.(await ...)` would + // short-circuit its argument and skip the stop lifecycle entirely + // when the caller emitted without an acknowledgement. + const forwardedResponse = await forwardTargetResponse(response); + callback?.(forwardedResponse); return; } if (!targetSocketId) { @@ -624,9 +629,8 @@ export function rpcHandler( } const response = Array.isArray(responses) ? responses[0] : responses; - if (callback) { - callback(await forwardTargetResponse(response)); - } + const forwardedResponse = await forwardTargetResponse(response); + callback?.(forwardedResponse); return; } @@ -691,9 +695,8 @@ export function rpcHandler( buildForwardedRequest(), ); - if (callback) { - callback(await forwardTargetResponse(response)); - } + const forwardedResponse = await forwardTargetResponse(response); + callback?.(forwardedResponse); } catch (error) { const errorMsg = error instanceof Error ? error.message : 'RPC call failed'; From 10b837c92af586efbdfcf65e201a6636e4c1cd64 Mon Sep 17 00:00:00 2001 From: Leeroy Brun Date: Mon, 3 Aug 2026 18:51:24 +0200 Subject: [PATCH 5/7] feat(rpc): expose daemon-proven stop metadata --- .../cli/src/api/apiMachine.transports.test.ts | 33 +++++++- apps/cli/src/api/apiMachine.ts | 2 + ...MachineRpcTransportAcknowledgement.test.ts | 35 ++++++++ ...ojectMachineRpcTransportAcknowledgement.ts | 22 +++++ .../cli/src/api/rpc/RpcHandlerManager.test.ts | 39 +++++++++ apps/cli/src/api/rpc/RpcHandlerManager.ts | 80 +++++++++++++++++-- apps/cli/src/api/rpc/types.ts | 16 ++-- apps/cli/src/api/types.ts | 9 +-- packages/protocol/src/socketRpc.test.ts | 35 ++++++++ packages/protocol/src/socketRpc.ts | 35 ++++++++ 10 files changed, 285 insertions(+), 21 deletions(-) create mode 100644 apps/cli/src/api/machine/projectMachineRpcTransportAcknowledgement.test.ts create mode 100644 apps/cli/src/api/machine/projectMachineRpcTransportAcknowledgement.ts create mode 100644 packages/protocol/src/socketRpc.test.ts diff --git a/apps/cli/src/api/apiMachine.transports.test.ts b/apps/cli/src/api/apiMachine.transports.test.ts index 55474e57a5..daaa9379c9 100644 --- a/apps/cli/src/api/apiMachine.transports.test.ts +++ b/apps/cli/src/api/apiMachine.transports.test.ts @@ -12,7 +12,7 @@ import { import { logger } from '@/ui/logger'; import type { Machine } from './types'; -const { configurationMock, mockAxiosGet, mockAxiosIsAxiosError, mockAxiosPost, mockIo } = vi.hoisted(() => ({ +const { configurationMock, mockAxiosGet, mockAxiosIsAxiosError, mockAxiosPost, mockIo, rpcHandlerConfigs } = vi.hoisted(() => ({ configurationMock: { apiServerUrl: 'http://localhost:3005', activeServerDir: '', @@ -29,6 +29,7 @@ const { configurationMock, mockAxiosGet, mockAxiosIsAxiosError, mockAxiosPost, m emitWithAck: vi.fn(), io: { on: vi.fn() }, })), + rpcHandlerConfigs: [] as Array>, })); vi.mock('socket.io-client', () => ({ @@ -63,6 +64,9 @@ vi.mock('@/rpc/handlers/machineFileBrowser/registerMachineFileBrowserHandlers', vi.mock('./machine/rpcHandlers', () => ({ registerMachineRpcHandlers: vi.fn() })); vi.mock('./rpc/RpcHandlerManager', () => ({ RpcHandlerManager: class { + constructor(config: Record) { + rpcHandlerConfigs.push(config); + } registerHandler() {} onSocketConnect() {} onSocketDisconnect() {} @@ -88,6 +92,7 @@ describe('ApiMachineClient transports', () => { mockAxiosPost.mockResolvedValue({ status: 200, data: { success: true, applied: true } }); mockAxiosGet.mockResolvedValue({ status: 200, data: { machine: null } }); bindApiSessionSocketMock(mockIo, createApiSessionSocketStub()); + rpcHandlerConfigs.length = 0; }); afterEach(() => { @@ -123,6 +128,32 @@ describe('ApiMachineClient transports', () => { expect(opts.autoConnect).toBe(false); }); + it('configures machine RPC to project only strict completed-stop transport proof', async () => { + const mod = await import('./apiMachine'); + const { RPC_METHODS } = await import('@happier-dev/protocol/rpc'); + + new mod.ApiMachineClient('fake-token', { + id: 'test-machine', + encryptionKey: new Uint8Array(32), + encryptionVariant: 'legacy', + metadata: null, + metadataVersion: 0, + daemonState: null, + daemonStateVersion: 0, + }); + + const projector = rpcHandlerConfigs.at(-1)?.projectTransportAcknowledgement; + expect(projector).toBeTypeOf('function'); + expect((projector as (input: { method: string; result: unknown }) => unknown)({ + method: `test-machine:${RPC_METHODS.STOP_SESSION}`, + result: { status: 'stopped' }, + })).toEqual({ kind: 'session.stop', status: 'stopped' }); + expect((projector as (input: { method: string; result: unknown }) => unknown)({ + method: `test-machine:${RPC_METHODS.STOP_SESSION}`, + result: { status: 'requested' }, + })).toBeNull(); + }); + it('serializes machine refresh errors without dumping axios request details', async () => { const mod = await import('./apiMachine'); diff --git a/apps/cli/src/api/apiMachine.ts b/apps/cli/src/api/apiMachine.ts index 5863fac35a..d8980f81c1 100644 --- a/apps/cli/src/api/apiMachine.ts +++ b/apps/cli/src/api/apiMachine.ts @@ -48,6 +48,7 @@ import { recoverDaemonTerminalSessionMutationJournals } from './session/mutation import type { DaemonToServerEvents, ServerToDaemonEvents } from './machine/socketTypes'; import { authorizeMachineRpcRequest } from './machine/machineRpcAuthorization'; +import { projectMachineRpcTransportAcknowledgement } from './machine/projectMachineRpcTransportAcknowledgement'; import { registerMachineRpcHandlers, type MachineRpcHandlerDeps, type MachineRpcHandlers } from './machine/rpcHandlers'; import { resolveMachineRpcWorkingDirectory } from './machine/resolveMachineRpcWorkingDirectory'; import type { Socket } from 'socket.io-client'; @@ -255,6 +256,7 @@ export class ApiMachineClient { } }, authorizeRequest: authorizeMachineRpcRequest, + projectTransportAcknowledgement: projectMachineRpcTransportAcknowledgement, }); const machineRpcWorkingDirectory = resolveMachineRpcWorkingDirectory(); diff --git a/apps/cli/src/api/machine/projectMachineRpcTransportAcknowledgement.test.ts b/apps/cli/src/api/machine/projectMachineRpcTransportAcknowledgement.test.ts new file mode 100644 index 0000000000..439192d2da --- /dev/null +++ b/apps/cli/src/api/machine/projectMachineRpcTransportAcknowledgement.test.ts @@ -0,0 +1,35 @@ +import { describe, expect, it } from 'vitest'; +import { RPC_METHODS } from '@happier-dev/protocol/rpc'; + +import { projectMachineRpcTransportAcknowledgement } from './projectMachineRpcTransportAcknowledgement'; + +describe('projectMachineRpcTransportAcknowledgement', () => { + it('projects proof only for a strict completed stop result', () => { + expect(projectMachineRpcTransportAcknowledgement({ + method: `machine-1:${RPC_METHODS.STOP_SESSION}`, + result: { status: 'stopped' }, + })).toEqual({ kind: 'session.stop', status: 'stopped' }); + + for (const result of [ + { status: 'requested' }, + { status: 'not_found' }, + { status: 'incomplete', reason: 'runner_exit_timeout' }, + ]) { + expect(projectMachineRpcTransportAcknowledgement({ + method: `machine-1:${RPC_METHODS.STOP_SESSION}`, + result, + })).toBeNull(); + } + }); + + it('does not project proof for another method or a lookalike result', () => { + expect(projectMachineRpcTransportAcknowledgement({ + method: 'machine-1:other-method', + result: { status: 'stopped' }, + })).toBeNull(); + expect(projectMachineRpcTransportAcknowledgement({ + method: `machine-1:${RPC_METHODS.STOP_SESSION}`, + result: { status: 'stopped', extra: true }, + })).toBeNull(); + }); +}); diff --git a/apps/cli/src/api/machine/projectMachineRpcTransportAcknowledgement.ts b/apps/cli/src/api/machine/projectMachineRpcTransportAcknowledgement.ts new file mode 100644 index 0000000000..d9158e20a7 --- /dev/null +++ b/apps/cli/src/api/machine/projectMachineRpcTransportAcknowledgement.ts @@ -0,0 +1,22 @@ +import { StopSessionResultSchema } from '@happier-dev/protocol'; +import { + RPC_METHODS, + resolveSocketRpcSessionWriteAuthorizationMethod, +} from '@happier-dev/protocol/rpc'; +import type { SocketRpcTransportAcknowledgementV1 } from '@happier-dev/protocol/socketRpc'; + +export function projectMachineRpcTransportAcknowledgement(input: Readonly<{ + method: string; + result: unknown; +}>): SocketRpcTransportAcknowledgementV1 | null { + if ( + resolveSocketRpcSessionWriteAuthorizationMethod(input.method) + !== RPC_METHODS.STOP_SESSION + ) { + return null; + } + const parsed = StopSessionResultSchema.safeParse(input.result); + return parsed.success && parsed.data.status === 'stopped' + ? { kind: 'session.stop', status: 'stopped' } + : null; +} diff --git a/apps/cli/src/api/rpc/RpcHandlerManager.test.ts b/apps/cli/src/api/rpc/RpcHandlerManager.test.ts index b9467d553c..17dd5f268c 100644 --- a/apps/cli/src/api/rpc/RpcHandlerManager.test.ts +++ b/apps/cli/src/api/rpc/RpcHandlerManager.test.ts @@ -139,6 +139,45 @@ describe('RpcHandlerManager.handleRequest (plaintext)', () => { }); describe('RpcHandlerManager.handleRequest (encrypted)', () => { + it('keeps the encrypted result opaque while exposing requested transport acknowledgement metadata', async () => { + const encryptionKey = new Uint8Array(32).fill(3); + const rpc = new RpcHandlerManager({ + scopePrefix: 'machine-1', + encryptionKey, + encryptionVariant: 'dataKey', + logger: () => {}, + projectTransportAcknowledgement: ({ method, result }) => ( + method === 'machine-1:stop-session' + && typeof result === 'object' + && result !== null + && (result as { status?: unknown }).status === 'stopped' + ? { kind: 'session.stop' as const, status: 'stopped' as const } + : null + ), + }); + + rpc.registerHandler('stop-session', async () => ({ status: 'stopped' })); + + const res = await rpc.handleRequest({ + method: 'machine-1:stop-session', + params: encodeBase64(encrypt(encryptionKey, 'dataKey', { sessionId: 'sess_1' })), + transportResponseEnvelopeVersion: 1, + }); + + expect(res).toEqual({ + v: 1, + result: expect.any(String), + acknowledgement: { kind: 'session.stop', status: 'stopped' }, + }); + expect( + decrypt( + encryptionKey, + 'dataKey', + decodeBase64((res as { result: string }).result), + ), + ).toEqual({ status: 'stopped' }); + }); + it('rejects encrypted requests when the authorization hook rejects the decrypted params', async () => { const encryptionKey = new Uint8Array(32).fill(5); const authorizeRequest = vi.fn(() => ({ diff --git a/apps/cli/src/api/rpc/RpcHandlerManager.ts b/apps/cli/src/api/rpc/RpcHandlerManager.ts index 50662f77f2..ba0225def5 100644 --- a/apps/cli/src/api/rpc/RpcHandlerManager.ts +++ b/apps/cli/src/api/rpc/RpcHandlerManager.ts @@ -13,7 +13,11 @@ import { type RpcAuthorizationResult, } from './types'; import { Socket } from 'socket.io-client'; -import { SOCKET_RPC_EVENTS } from '@happier-dev/protocol/socketRpc'; +import { + SOCKET_RPC_EVENTS, + SOCKET_RPC_TRANSPORT_RESPONSE_ENVELOPE_VERSION_V1, + type SocketRpcTransportAcknowledgementV1, +} from '@happier-dev/protocol/socketRpc'; import { RPC_ERROR_CODES, RPC_ERROR_MESSAGES, @@ -30,6 +34,7 @@ export class RpcHandlerManager { private readonly logger: (message: string, data?: any) => void; private readonly onRegistrationError: RpcHandlerConfig['onRegistrationError']; private readonly authorizeRequest: RpcHandlerConfig['authorizeRequest']; + private readonly projectTransportAcknowledgement: RpcHandlerConfig['projectTransportAcknowledgement']; private socket: Socket | null = null; private inFlightRequestCount = 0; private idleResolvers = new Set<() => void>(); @@ -42,6 +47,7 @@ export class RpcHandlerManager { this.logger = config.logger || ((msg, data) => defaultLogger.debug(msg, data)); this.onRegistrationError = config.onRegistrationError; this.authorizeRequest = config.authorizeRequest; + this.projectTransportAcknowledgement = config.projectTransportAcknowledgement; } private encodeResponse(response: unknown): unknown { @@ -83,7 +89,7 @@ export class RpcHandlerManager { if (!handler) { this.logger('[RPC] [ERROR] Method not found', { method: request.method }); const errorResponse = { error: RPC_ERROR_MESSAGES.METHOD_NOT_FOUND, errorCode: RPC_ERROR_CODES.METHOD_NOT_FOUND }; - return this.encodeResponse(errorResponse); + return this.encodeTransportResponse(request, errorResponse); } // Decrypt the incoming params (unless session is plaintext). @@ -96,7 +102,7 @@ export class RpcHandlerManager { const errorResponse = { error: 'Invalid RPC params', }; - return this.encodeResponse(errorResponse); + return this.encodeTransportResponse(request, errorResponse); } const authorizationResult: RpcAuthorizationResult = this.authorizeRequest @@ -107,7 +113,7 @@ export class RpcHandlerManager { }) : { ok: true }; if (authorizationResult.ok !== true) { - return this.encodeResponse({ + return this.encodeTransportResponse(request, { error: authorizationResult.error, ...(authorizationResult.errorCode ? { errorCode: authorizationResult.errorCode } : {}), }); @@ -119,9 +125,20 @@ export class RpcHandlerManager { this.logger('[RPC] Handler returned', { method: request.method, hasResult: result !== undefined }); // Encrypt and return the response - const response = this.encodeResponse(result); - if (this.encryptionMode !== 'plain' && typeof response === 'string') { - this.logger('[RPC] Sending encrypted response', { method: request.method, responseLength: response.length }); + const acknowledgement = this.projectAcknowledgement(request, decryptedParams, result); + const response = this.encodeTransportResponse(request, result, acknowledgement); + if (this.encryptionMode !== 'plain') { + const encodedResult = request.transportResponseEnvelopeVersion + === SOCKET_RPC_TRANSPORT_RESPONSE_ENVELOPE_VERSION_V1 + && response + && typeof response === 'object' + && !Array.isArray(response) + ? (response as { result?: unknown }).result + : response; + this.logger('[RPC] Sending encrypted response', { + method: request.method, + responseLength: typeof encodedResult === 'string' ? encodedResult.length : 0, + }); } return response; } catch (error) { @@ -129,7 +146,7 @@ export class RpcHandlerManager { const errorResponse = { error: error instanceof Error ? error.message : 'Unknown error' }; - return this.encodeResponse(errorResponse); + return this.encodeTransportResponse(request, errorResponse); } finally { this.finishInFlightRequest(); } @@ -224,6 +241,53 @@ export class RpcHandlerManager { return `${this.scopePrefix}:${method}`; } + private encodeTransportResponse( + request: RpcRequest, + result: unknown, + acknowledgement: SocketRpcTransportAcknowledgementV1 | null = null, + ): unknown { + const encodedResult = this.encodeResponse(result); + if ( + request.transportResponseEnvelopeVersion + !== SOCKET_RPC_TRANSPORT_RESPONSE_ENVELOPE_VERSION_V1 + ) { + return encodedResult; + } + return { + v: SOCKET_RPC_TRANSPORT_RESPONSE_ENVELOPE_VERSION_V1, + result: encodedResult, + ...(acknowledgement ? { acknowledgement } : {}), + }; + } + + private projectAcknowledgement( + request: RpcRequest, + params: unknown, + result: unknown, + ): SocketRpcTransportAcknowledgementV1 | null { + if ( + request.transportResponseEnvelopeVersion + !== SOCKET_RPC_TRANSPORT_RESPONSE_ENVELOPE_VERSION_V1 + || !this.projectTransportAcknowledgement + ) { + return null; + } + try { + return this.projectTransportAcknowledgement({ + method: request.method, + params, + result, + ...(request.authorization ? { authorization: request.authorization } : {}), + }); + } catch (error) { + this.logger('[RPC] Transport acknowledgement projection failed', { + method: request.method, + error, + }); + return null; + } + } + private beginInFlightRequest(): void { this.inFlightRequestCount += 1; } diff --git a/apps/cli/src/api/rpc/types.ts b/apps/cli/src/api/rpc/types.ts index a757e2bced..b15bb46dae 100644 --- a/apps/cli/src/api/rpc/types.ts +++ b/apps/cli/src/api/rpc/types.ts @@ -1,4 +1,8 @@ import type { SocketRpcAuthorizationContext } from '@happier-dev/protocol'; +import type { + SocketRpcRequestPayload, + SocketRpcTransportAcknowledgementV1, +} from '@happier-dev/protocol/socketRpc'; /** * Common RPC types and interfaces for both session and machine clients @@ -34,11 +38,7 @@ export type RpcHandlerMap = Map; /** * RPC request data from server */ -export interface RpcRequest { - method: string; - params: unknown; - authorization?: SocketRpcAuthorizationContext; -} +export type RpcRequest = SocketRpcRequestPayload; /** * RPC response callback @@ -60,6 +60,12 @@ export interface RpcHandlerConfig { params: unknown; authorization?: SocketRpcAuthorizationContext; }>) => RpcAuthorizationResult | Promise; + projectTransportAcknowledgement?: (request: Readonly<{ + method: string; + params: unknown; + result: unknown; + authorization?: SocketRpcAuthorizationContext; + }>) => SocketRpcTransportAcknowledgementV1 | null; } export type RpcAuthorizationResult = diff --git a/apps/cli/src/api/types.ts b/apps/cli/src/api/types.ts index 7449612bb7..14ee6ab81c 100644 --- a/apps/cli/src/api/types.ts +++ b/apps/cli/src/api/types.ts @@ -1,6 +1,6 @@ import { z } from 'zod' import { UsageSchema } from '@/api/usage' -import { SOCKET_RPC_EVENTS } from '@happier-dev/protocol/socketRpc' +import { SOCKET_RPC_EVENTS, type SocketRpcRequestPayload as ProtocolSocketRpcRequestPayload } from '@happier-dev/protocol/socketRpc' import { SentFromSchema } from '@happier-dev/protocol' import type { ExecutionRunPublicState } from '@happier-dev/protocol' import type { @@ -19,7 +19,6 @@ import type { SessionRollbackRangesV1, SessionUsageLimitRecoveryV1, SessionTerminalMetadata, - SocketRpcAuthorizationContext, SessionMessageRole, ProviderSessionInfoV1, SessionRuntimeActivityState, @@ -140,11 +139,7 @@ export type UpdateMachineBody = Extract export const SessionBroadcastSchema = SessionBroadcastContainerSchema export type SessionBroadcast = SessionBroadcastContainer -export interface SocketRpcRequestPayload { - method: string - params: unknown - authorization?: SocketRpcAuthorizationContext -} +export type SocketRpcRequestPayload = ProtocolSocketRpcRequestPayload export interface SocketRpcCallPayload extends SocketRpcRequestPayload { timeoutMs?: number diff --git a/packages/protocol/src/socketRpc.test.ts b/packages/protocol/src/socketRpc.test.ts new file mode 100644 index 0000000000..cc449bd83d --- /dev/null +++ b/packages/protocol/src/socketRpc.test.ts @@ -0,0 +1,35 @@ +import { describe, expect, it } from 'vitest'; + +import { + SOCKET_RPC_TRANSPORT_RESPONSE_ENVELOPE_VERSION_V1, + SocketRpcTransportResponseEnvelopeV1Schema, +} from './socketRpc'; + +describe('SocketRpcTransportResponseEnvelopeV1Schema', () => { + it('accepts an opaque result with a strict stopped-session acknowledgement', () => { + expect(SocketRpcTransportResponseEnvelopeV1Schema.parse({ + v: SOCKET_RPC_TRANSPORT_RESPONSE_ENVELOPE_VERSION_V1, + result: 'opaque-encrypted-result', + acknowledgement: { kind: 'session.stop', status: 'stopped' }, + })).toEqual({ + v: 1, + result: 'opaque-encrypted-result', + acknowledgement: { kind: 'session.stop', status: 'stopped' }, + }); + }); + + it('rejects unrecognized acknowledgement claims', () => { + expect(SocketRpcTransportResponseEnvelopeV1Schema.safeParse({ + v: 1, + result: 'opaque-encrypted-result', + acknowledgement: { kind: 'session.stop', status: 'requested' }, + }).success).toBe(false); + }); + + it('requires the original result even when stopped proof is present', () => { + expect(SocketRpcTransportResponseEnvelopeV1Schema.safeParse({ + v: 1, + acknowledgement: { kind: 'session.stop', status: 'stopped' }, + }).success).toBe(false); + }); +}); diff --git a/packages/protocol/src/socketRpc.ts b/packages/protocol/src/socketRpc.ts index aea683bed1..7eda5a1473 100644 --- a/packages/protocol/src/socketRpc.ts +++ b/packages/protocol/src/socketRpc.ts @@ -1,3 +1,7 @@ +import { z } from 'zod'; + +import type { SocketRpcAuthorizationContext } from './rpc.js'; + export const SOCKET_RPC_EVENTS = { REGISTER: 'rpc-register', REGISTERED: 'rpc-registered', @@ -11,6 +15,37 @@ export const SOCKET_RPC_EVENTS = { export type SocketRpcEvent = (typeof SOCKET_RPC_EVENTS)[keyof typeof SOCKET_RPC_EVENTS]; +export type SocketRpcRequestPayload = Readonly<{ + method: string; + params: unknown; + authorization?: SocketRpcAuthorizationContext; + transportResponseEnvelopeVersion?: 1; +}>; + +export const SOCKET_RPC_TRANSPORT_RESPONSE_ENVELOPE_VERSION_V1 = 1 as const; + +export const SocketRpcTransportAcknowledgementV1Schema = z.discriminatedUnion('kind', [ + z.object({ + kind: z.literal('session.stop'), + status: z.literal('stopped'), + }).strict(), +]); + +export type SocketRpcTransportAcknowledgementV1 = + z.infer; + +export const SocketRpcTransportResponseEnvelopeV1Schema = z.object({ + v: z.literal(SOCKET_RPC_TRANSPORT_RESPONSE_ENVELOPE_VERSION_V1), + result: z.unknown(), + acknowledgement: SocketRpcTransportAcknowledgementV1Schema.optional(), +}).strict().refine( + (value) => Object.prototype.hasOwnProperty.call(value, 'result'), + { path: ['result'], message: 'result is required' }, +); + +export type SocketRpcTransportResponseEnvelopeV1 = + z.infer; + export type SocketRpcTargetFailureV1 = Readonly<{ type: 'socket-rpc-target-failure-v1'; errorCode: string; From 9ee9ed2c1053ef3495edd0d6c8d20444179b80d3 Mon Sep 17 00:00:00 2001 From: Leeroy Brun Date: Mon, 3 Aug 2026 18:51:35 +0200 Subject: [PATCH 6/7] fix(server-presence): finalize only daemon-proven stops --- .../migration.sql | 4 - .../migration.sql | 4 - apps/server/prisma/mysql/schema.prisma | 4 - apps/server/prisma/schema.prisma | 4 - .../migration.sql | 4 - apps/server/prisma/sqlite/schema.prisma | 4 - apps/server/sources/app/api/socket.ts | 14 - ...er.explicitStop.sqlite.integration.spec.ts | 21 +- .../api/socket/rpcHandler.integration.spec.ts | 359 ++++++++++++++++-- .../sources/app/api/socket/rpcHandler.ts | 148 +++++--- ...blisherPresence.sqlite.integration.spec.ts | 112 +----- .../app/presence/sessionPublisherPresence.ts | 88 +---- 12 files changed, 450 insertions(+), 316 deletions(-) delete mode 100644 apps/server/prisma/migrations/20260802220000_add_session_stop_requested_at/migration.sql delete mode 100644 apps/server/prisma/mysql/migrations/20260802220000_add_session_stop_requested_at/migration.sql delete mode 100644 apps/server/prisma/sqlite/migrations/20260802220000_add_session_stop_requested_at/migration.sql diff --git a/apps/server/prisma/migrations/20260802220000_add_session_stop_requested_at/migration.sql b/apps/server/prisma/migrations/20260802220000_add_session_stop_requested_at/migration.sql deleted file mode 100644 index 6cbdc9b8a6..0000000000 --- a/apps/server/prisma/migrations/20260802220000_add_session_stop_requested_at/migration.sql +++ /dev/null @@ -1,4 +0,0 @@ --- Durable explicit-stop intent: the disconnect that completes a stop can land on a --- different server instance than the RPC that accepted it. - -ALTER TABLE "Session" ADD COLUMN "stopRequestedAt" TIMESTAMP(3); diff --git a/apps/server/prisma/mysql/migrations/20260802220000_add_session_stop_requested_at/migration.sql b/apps/server/prisma/mysql/migrations/20260802220000_add_session_stop_requested_at/migration.sql deleted file mode 100644 index c0cec2be44..0000000000 --- a/apps/server/prisma/mysql/migrations/20260802220000_add_session_stop_requested_at/migration.sql +++ /dev/null @@ -1,4 +0,0 @@ --- Durable explicit-stop intent: the disconnect that completes a stop can land on a --- different server instance than the RPC that accepted it. - -ALTER TABLE `Session` ADD COLUMN `stopRequestedAt` DATETIME(3) NULL; diff --git a/apps/server/prisma/mysql/schema.prisma b/apps/server/prisma/mysql/schema.prisma index 02c8dd3c8d..8a904248a0 100644 --- a/apps/server/prisma/mysql/schema.prisma +++ b/apps/server/prisma/mysql/schema.prisma @@ -321,10 +321,6 @@ model Session { runtimeActivityRevision BigInt @default(0) meaningfulActivityAt DateTime? active Boolean @default(false) - // When a runner accepted an explicit stop. The disconnect that completes the stop can - // reach a different server instance than the RPC that requested it, so the intent has - // to live on the session row rather than in the accepting instance's memory. - stopRequestedAt DateTime? archivedAt DateTime? lastActiveAt DateTime @default(now()) createdAt DateTime @default(now()) diff --git a/apps/server/prisma/schema.prisma b/apps/server/prisma/schema.prisma index eec15563ff..2893fd7a4e 100644 --- a/apps/server/prisma/schema.prisma +++ b/apps/server/prisma/schema.prisma @@ -320,10 +320,6 @@ model Session { runtimeActivityRevision BigInt @default(0) meaningfulActivityAt DateTime? active Boolean @default(false) - // When a runner accepted an explicit stop. The disconnect that completes the stop can - // reach a different server instance than the RPC that requested it, so the intent has - // to live on the session row rather than in the accepting instance's memory. - stopRequestedAt DateTime? archivedAt DateTime? lastActiveAt DateTime @default(now()) createdAt DateTime @default(now()) diff --git a/apps/server/prisma/sqlite/migrations/20260802220000_add_session_stop_requested_at/migration.sql b/apps/server/prisma/sqlite/migrations/20260802220000_add_session_stop_requested_at/migration.sql deleted file mode 100644 index d91de5e28b..0000000000 --- a/apps/server/prisma/sqlite/migrations/20260802220000_add_session_stop_requested_at/migration.sql +++ /dev/null @@ -1,4 +0,0 @@ --- Durable explicit-stop intent: the disconnect that completes a stop can land on a --- different server instance than the RPC that accepted it. - -ALTER TABLE "Session" ADD COLUMN "stopRequestedAt" DATETIME; diff --git a/apps/server/prisma/sqlite/schema.prisma b/apps/server/prisma/sqlite/schema.prisma index 83d6c585f4..46be91f1b8 100644 --- a/apps/server/prisma/sqlite/schema.prisma +++ b/apps/server/prisma/sqlite/schema.prisma @@ -321,10 +321,6 @@ model Session { runtimeActivityRevision BigInt @default(0) meaningfulActivityAt DateTime? active Boolean @default(false) - // When a runner accepted an explicit stop. The disconnect that completes the stop can - // reach a different server instance than the RPC that requested it, so the intent has - // to live on the session row rather than in the accepting instance's memory. - stopRequestedAt DateTime? archivedAt DateTime? lastActiveAt DateTime @default(now()) createdAt DateTime @default(now()) diff --git a/apps/server/sources/app/api/socket.ts b/apps/server/sources/app/api/socket.ts index 4f78692645..f91d52f4ea 100644 --- a/apps/server/sources/app/api/socket.ts +++ b/apps/server/sources/app/api/socket.ts @@ -349,20 +349,6 @@ export function startSocket(app: Fastify) { decrementWebSocketConnection(connection.connectionType); if (connection.connectionType === 'session-scoped') { void sessionPublisherPresence.forgetDisconnectedPublisher({ socket }).then(async (result) => { - // A disconnect that completes an explicit stop ends the session, so - // participants have to learn it went inactive now rather than when the - // presence timeout fence eventually expires. - if (result.status === 'closed' && 'participantCursors' in result) { - await publishSessionPublisherLifecycleUpdate({ - sessionId: connection.sessionId, - participantCursors: result.participantCursors, - active: false, - activeAt: result.activeAt.getTime(), - ...(result.projection ? { projection: result.projection } : {}), - ...(result.turnProjection ?? {}), - }); - return; - } if (result.status !== 'applied') return; await publishSessionPublisherLifecycleUpdate({ sessionId: connection.sessionId, diff --git a/apps/server/sources/app/api/socket/rpcHandler.explicitStop.sqlite.integration.spec.ts b/apps/server/sources/app/api/socket/rpcHandler.explicitStop.sqlite.integration.spec.ts index 3a702b491e..45696ec59e 100644 --- a/apps/server/sources/app/api/socket/rpcHandler.explicitStop.sqlite.integration.spec.ts +++ b/apps/server/sources/app/api/socket/rpcHandler.explicitStop.sqlite.integration.spec.ts @@ -69,7 +69,19 @@ describe("explicit machine stop RPC on SQLite", () => { const method = `${seeded.machineId}:${RPC_METHODS.STOP_SESSION}`; const targetEmitWithAck = vi.fn(async () => { await params.beforeResponse?.(); - return params.targetResponse; + const isStopped = ( + params.targetResponse + && typeof params.targetResponse === "object" + && !Array.isArray(params.targetResponse) + && (params.targetResponse as { status?: unknown }).status === "stopped" + ); + return { + v: 1, + result: params.targetResponse, + ...(isStopped + ? { acknowledgement: { kind: "session.stop", status: "stopped" } } + : {}), + }; }); const target = createFakeSocket({ id: "target-socket", @@ -117,6 +129,7 @@ describe("explicit machine stop RPC on SQLite", () => { kind: "session.write", sessionId: result.session.id, }, + transportResponseEnvelopeVersion: 1, }, ); expect(result.callback).toHaveBeenCalledWith({ @@ -166,7 +179,11 @@ describe("explicit machine stop RPC on SQLite", () => { }); if (successor.status !== "registered") throw new Error("expected successor registration"); successorFence = successor.committedFence; - return { status: "stopped" }; + return { + v: 1, + result: { status: "stopped" }, + acknowledgement: { kind: "session.stop", status: "stopped" }, + }; }, })), }); diff --git a/apps/server/sources/app/api/socket/rpcHandler.integration.spec.ts b/apps/server/sources/app/api/socket/rpcHandler.integration.spec.ts index 1882193545..ecacb16216 100644 --- a/apps/server/sources/app/api/socket/rpcHandler.integration.spec.ts +++ b/apps/server/sources/app/api/socket/rpcHandler.integration.spec.ts @@ -127,6 +127,34 @@ describe("rpcHandler", () => { } }); + it("rejects a machine-scoped socket registering another machine's RPC prefix", async () => { + vi.resetModules(); + dbMockFns.machineFindFirst.mockResolvedValue({ + id: "machine-1", + revokedAt: null, + replacedByMachineId: null, + }); + const { rpcHandler } = await import("./rpcHandler"); + const socket = createFakeSocket({ + data: { clientType: "machine-scoped", machineId: "machine-1" }, + }); + const listeners = new Map(); + + rpcHandler("user-1", socket as any, listeners, new Map(), { + io: {} as any, + redisRegistry: { enabled: false }, + }); + + const method = `machine-2:${RPC_METHODS.STOP_SESSION}`; + await getSocketHandler(socket, SOCKET_RPC_EVENTS.REGISTER)({ method }); + + expect(listeners.has(method)).toBe(false); + expect(socket.emit).toHaveBeenCalledWith(SOCKET_RPC_EVENTS.ERROR, { + type: "register", + error: "Forbidden", + }); + }); + it.each([ { method: `machine-1:${RPC_METHODS.SPAWN_HAPPY_SESSION}`, @@ -658,63 +686,326 @@ describe("rpcHandler", () => { ); }); - it("records no explicit-stop intent when a session stop RPC has no reachable runner", async () => { + it("finalizes daemon-proven machine stop even when the caller sends no acknowledgement callback", async () => { vi.resetModules(); + dbMockFns.machineFindFirst.mockResolvedValue({ + id: "machine-1", + revokedAt: null, + replacedByMachineId: null, + }); + dbMockFns.sessionFindUnique.mockResolvedValue({ + accountId: "user-1", + active: true, + lastActiveAt: new Date(1_000), + }); const { rpcHandler } = await import("./rpcHandler"); - const socket = createFakeSocket(); - const markExplicitStopRequested = vi.fn(); + const method = `machine-1:${RPC_METHODS.STOP_SESSION}`; + const capturedTarget = { + binding: { accountId: "user-1", machineId: "machine-1", sessionId: "sess_1" }, + committedFence: new Date(1_000), + }; + const finalizeExplicitMachineStop = vi.fn().mockResolvedValue({ status: "already_inactive" }); + const targetEmitWithAck = vi.fn().mockResolvedValue({ + v: 1, + result: "opaque-e2ee-result", + acknowledgement: { kind: "session.stop", status: "stopped" }, + }); + const targetSocket = createFakeSocket({ + id: "daemon-socket", + data: { clientType: "machine-scoped", machineId: "machine-1" }, + timeout: vi.fn(() => ({ emitWithAck: targetEmitWithAck })) as any, + }); + const callerSocket = createFakeSocket({ id: "caller-socket" }); - rpcHandler("user-1", socket as any, new Map() as any, new Map() as any, { + rpcHandler("user-1", callerSocket as any, new Map([[method, targetSocket]]) as any, new Map() as any, { io: {} as any, redisRegistry: { enabled: false }, sessionPublisherPresence: { - captureExplicitMachineStop: vi.fn(), - finalizeExplicitMachineStop: vi.fn(), - markExplicitStopRequested, + captureExplicitMachineStop: vi.fn().mockResolvedValue({ status: "captured", target: capturedTarget }), + finalizeExplicitMachineStop, } as any, }); - const handler = getSocketHandler(socket, SOCKET_RPC_EVENTS.CALL); - const callback = vi.fn(); - await handler({ method: `sess_1:${RPC_METHODS.KILL_SESSION}`, params: {} }, callback); + await getSocketHandler(callerSocket, SOCKET_RPC_EVENTS.CALL)({ + method, + params: "opaque-e2ee-params", + authorization: { kind: "session.write", sessionId: "sess_1" }, + }); - // The stop never reached a runner, so nothing may later read a disconnect as an - // intentional termination — that would end a session whose runner is still alive. - expect(callback).toHaveBeenCalledWith( - expect.objectContaining({ ok: false, errorCode: RPC_ERROR_CODES.METHOD_NOT_AVAILABLE }), + expect(targetEmitWithAck).toHaveBeenCalledTimes(1); + expect(finalizeExplicitMachineStop).toHaveBeenCalledWith({ target: capturedTarget }); + }); + + it("finalizes an encrypted machine stop from authenticated transport proof", async () => { + vi.resetModules(); + dbMockFns.machineFindFirst.mockResolvedValue({ + id: "machine-1", + revokedAt: null, + replacedByMachineId: null, + }); + dbMockFns.sessionFindUnique.mockResolvedValue({ + accountId: "user-1", + active: true, + lastActiveAt: new Date(1_000), + }); + const { rpcHandler } = await import("./rpcHandler"); + const method = `machine-1:${RPC_METHODS.STOP_SESSION}`; + const encryptedResult = "opaque-e2ee-result"; + const targetEmitWithAck = vi.fn().mockResolvedValue({ + v: 1, + result: encryptedResult, + acknowledgement: { kind: "session.stop", status: "stopped" }, + }); + const targetSocket = createFakeSocket({ + id: "daemon-socket", + data: { clientType: "machine-scoped", machineId: "machine-1" }, + timeout: vi.fn(() => ({ emitWithAck: targetEmitWithAck })) as any, + }); + const callerSocket = createFakeSocket({ id: "caller-socket" }); + const capturedTarget = { + binding: { accountId: "user-1", machineId: "machine-1", sessionId: "sess_1" }, + committedFence: new Date(1_000), + }; + const captureExplicitMachineStop = vi.fn().mockResolvedValue({ + status: "captured", + target: capturedTarget, + }); + const finalizeExplicitMachineStop = vi.fn().mockResolvedValue({ status: "already_inactive" }); + + rpcHandler( + "user-1", + callerSocket as any, + new Map([[method, targetSocket]]) as any, + new Map() as any, + { + io: {} as any, + redisRegistry: { enabled: false }, + sessionPublisherPresence: { + captureExplicitMachineStop, + finalizeExplicitMachineStop, + } as any, + }, ); - expect(markExplicitStopRequested).not.toHaveBeenCalled(); + + const callback = vi.fn(); + await getSocketHandler(callerSocket, SOCKET_RPC_EVENTS.CALL)({ + method, + params: "opaque-e2ee-params", + authorization: { kind: "session.write", sessionId: "sess_1" }, + }, callback); + + expect(targetEmitWithAck).toHaveBeenCalledWith(SOCKET_RPC_EVENTS.REQUEST, { + method, + params: "opaque-e2ee-params", + authorization: { kind: "session.write", sessionId: "sess_1" }, + transportResponseEnvelopeVersion: 1, + }); + expect(finalizeExplicitMachineStop).toHaveBeenCalledWith({ target: capturedTarget }); + expect(callback).toHaveBeenCalledWith({ ok: true, result: encryptedResult }); }); - it("records an accepted stop even when the caller sends no acknowledgement callback", async () => { + it("forwards an older daemon's raw encrypted result without treating it as stop proof", async () => { vi.resetModules(); + dbMockFns.machineFindFirst.mockResolvedValue({ + id: "machine-1", + revokedAt: null, + replacedByMachineId: null, + }); + dbMockFns.sessionFindUnique.mockResolvedValue({ + accountId: "user-1", + active: true, + lastActiveAt: new Date(1_000), + }); const { rpcHandler } = await import("./rpcHandler"); - const method = `sess_1:${RPC_METHODS.KILL_SESSION}`; - const markExplicitStopRequested = vi.fn(); - const targetEmitWithAck = vi.fn().mockResolvedValue({ status: "requested" }); + const method = `machine-1:${RPC_METHODS.STOP_SESSION}`; + const encryptedResult = "legacy-opaque-e2ee-result"; const targetSocket = createFakeSocket({ - id: "runner-socket", + id: "daemon-socket", + data: { clientType: "machine-scoped", machineId: "machine-1" }, + timeout: vi.fn(() => ({ emitWithAck: vi.fn().mockResolvedValue(encryptedResult) })) as any, + }); + const callerSocket = createFakeSocket({ id: "caller-socket" }); + const finalizeExplicitMachineStop = vi.fn(); + + rpcHandler( + "user-1", + callerSocket as any, + new Map([[method, targetSocket]]) as any, + new Map() as any, + { + io: {} as any, + redisRegistry: { enabled: false }, + sessionPublisherPresence: { + captureExplicitMachineStop: vi.fn().mockResolvedValue({ + status: "captured", + target: { + binding: { accountId: "user-1", machineId: "machine-1", sessionId: "sess_1" }, + committedFence: new Date(1_000), + }, + }), + finalizeExplicitMachineStop, + } as any, + }, + ); + + const callback = vi.fn(); + await getSocketHandler(callerSocket, SOCKET_RPC_EVENTS.CALL)({ + method, + params: "opaque-e2ee-params", + authorization: { kind: "session.write", sessionId: "sess_1" }, + }, callback); + + expect(finalizeExplicitMachineStop).not.toHaveBeenCalled(); + expect(callback).toHaveBeenCalledWith({ ok: true, result: encryptedResult }); + }); + + it("does not accept stop proof from a different machine-scoped responder", async () => { + vi.resetModules(); + dbMockFns.machineFindFirst.mockResolvedValue({ + id: "machine-2", + revokedAt: null, + replacedByMachineId: null, + }); + dbMockFns.sessionFindUnique.mockResolvedValue({ + accountId: "user-1", + active: true, + lastActiveAt: new Date(1_000), + }); + const { rpcHandler } = await import("./rpcHandler"); + const method = `machine-1:${RPC_METHODS.STOP_SESSION}`; + const targetEmitWithAck = vi.fn().mockResolvedValue({ + v: 1, + result: "opaque-e2ee-result", + acknowledgement: { kind: "session.stop", status: "stopped" }, + }); + const wrongMachineSocket = createFakeSocket({ + id: "wrong-machine-socket", + data: { clientType: "machine-scoped", machineId: "machine-2" }, timeout: vi.fn(() => ({ emitWithAck: targetEmitWithAck })) as any, }); const callerSocket = createFakeSocket({ id: "caller-socket" }); + const finalizeExplicitMachineStop = vi.fn(); - rpcHandler("user-1", callerSocket as any, new Map([[method, targetSocket]]) as any, new Map() as any, { - io: {} as any, - redisRegistry: { enabled: false }, - sessionPublisherPresence: { - captureExplicitMachineStop: vi.fn(), - finalizeExplicitMachineStop: vi.fn(), - markExplicitStopRequested, - } as any, + rpcHandler( + "user-1", + callerSocket as any, + new Map([[method, wrongMachineSocket]]) as any, + new Map() as any, + { + io: {} as any, + redisRegistry: { enabled: false }, + sessionPublisherPresence: { + captureExplicitMachineStop: vi.fn().mockResolvedValue({ + status: "captured", + target: { + binding: { accountId: "user-1", machineId: "machine-1", sessionId: "sess_1" }, + committedFence: new Date(1_000), + }, + }), + finalizeExplicitMachineStop, + } as any, + }, + ); + + const callback = vi.fn(); + await getSocketHandler(callerSocket, SOCKET_RPC_EVENTS.CALL)({ + method, + params: "opaque-e2ee-params", + authorization: { kind: "session.write", sessionId: "sess_1" }, + }, callback); + + expect(targetEmitWithAck).not.toHaveBeenCalled(); + expect(finalizeExplicitMachineStop).not.toHaveBeenCalled(); + expect(callback).toHaveBeenCalledWith(expect.objectContaining({ + ok: false, + errorCode: RPC_ERROR_CODES.METHOD_NOT_AVAILABLE, + })); + }); + + it("uses the Redis-selected current daemon instead of a stale same-machine local listener for stop proof", async () => { + vi.resetModules(); + const targetSocketId = "current-daemon-socket"; + const hmget = vi.fn().mockResolvedValue([targetSocketId]); + const evalFn = vi.fn(); + const multi = vi.fn(() => ({ hset: () => ({ expire: () => ({ exec: vi.fn() }) }) })); + vi.doMock("@/storage/redis/redis", () => ({ + getRedisClient: () => ({ hmget, eval: evalFn, multi }), + })); + dbMockFns.machineFindFirst.mockResolvedValue({ + id: "machine-1", + revokedAt: null, + replacedByMachineId: null, + }); + dbMockFns.sessionFindUnique.mockResolvedValue({ + accountId: "user-1", + active: true, + lastActiveAt: new Date(1_000), }); - // Socket.IO lets a caller emit without an acknowledgement. The runner still accepted - // the stop, so the disconnect it produces has to stay explainable — otherwise the - // session holds `active` for the full presence fence and archive keeps returning 409. - await getSocketHandler(callerSocket, SOCKET_RPC_EVENTS.CALL)({ method, params: {} }); + try { + const { rpcHandler } = await import("./rpcHandler"); + const method = `machine-1:${RPC_METHODS.STOP_SESSION}`; + const staleLocalEmitWithAck = vi.fn().mockResolvedValue({ + v: 1, + result: "stale-result", + acknowledgement: { kind: "session.stop", status: "stopped" }, + }); + const staleLocalSocket = createFakeSocket({ + id: "stale-local-daemon-socket", + data: { clientType: "machine-scoped", machineId: "machine-1" }, + timeout: vi.fn(() => ({ emitWithAck: staleLocalEmitWithAck })) as any, + }); + const currentRemoteSocket = createFakeSocket({ + id: targetSocketId, + data: { clientType: "machine-scoped", machineId: "machine-1" }, + }); + const remoteEmitWithAck = vi.fn().mockResolvedValue([{ + v: 1, + result: "current-result", + acknowledgement: { kind: "session.stop", status: "stopped" }, + }]); + const to = vi.fn(() => ({ emitWithAck: remoteEmitWithAck })); + const timeout = vi.fn(() => ({ to })); + const fetchSockets = vi.fn().mockResolvedValue([currentRemoteSocket]); + const io = { timeout, in: vi.fn(() => ({ fetchSockets })) } as any; + const callerSocket = createFakeSocket({ id: "caller-socket" }); + const finalizeExplicitMachineStop = vi.fn().mockResolvedValue({ status: "already_inactive" }); + + rpcHandler( + "user-1", + callerSocket as any, + new Map([[method, staleLocalSocket]]) as any, + new Map() as any, + { + io, + redisRegistry: { enabled: true, instanceId: "instance-1", ttlSeconds: 120 }, + sessionPublisherPresence: { + captureExplicitMachineStop: vi.fn().mockResolvedValue({ + status: "captured", + target: { + binding: { accountId: "user-1", machineId: "machine-1", sessionId: "sess_1" }, + committedFence: new Date(1_000), + }, + }), + finalizeExplicitMachineStop, + } as any, + }, + ); - expect(targetEmitWithAck).toHaveBeenCalledTimes(1); - expect(markExplicitStopRequested).toHaveBeenCalledWith({ sessionId: "sess_1" }); + const callback = vi.fn(); + await getSocketHandler(callerSocket, SOCKET_RPC_EVENTS.CALL)({ + method, + params: "opaque-e2ee-params", + authorization: { kind: "session.write", sessionId: "sess_1" }, + }, callback); + + expect(staleLocalEmitWithAck).not.toHaveBeenCalled(); + expect(remoteEmitWithAck).toHaveBeenCalledTimes(1); + expect(callback).toHaveBeenCalledWith({ ok: true, result: "current-result" }); + expect(finalizeExplicitMachineStop).toHaveBeenCalledTimes(1); + } finally { + vi.doUnmock("@/storage/redis/redis"); + } }); it("uses Redis RPC registry + io.emitWithAck when enabled", async () => { diff --git a/apps/server/sources/app/api/socket/rpcHandler.ts b/apps/server/sources/app/api/socket/rpcHandler.ts index 1050f6ba84..f326af3115 100644 --- a/apps/server/sources/app/api/socket/rpcHandler.ts +++ b/apps/server/sources/app/api/socket/rpcHandler.ts @@ -9,8 +9,11 @@ import { RPC_ERROR_MESSAGES, type SocketRpcAuthorizationContext, } from "@happier-dev/protocol/rpc"; -import { StopSessionResultSchema } from "@happier-dev/protocol"; -import { SOCKET_RPC_EVENTS } from "@happier-dev/protocol/socketRpc"; +import { + SOCKET_RPC_EVENTS, + SOCKET_RPC_TRANSPORT_RESPONSE_ENVELOPE_VERSION_V1, + SocketRpcTransportResponseEnvelopeV1Schema, +} from "@happier-dev/protocol/socketRpc"; import { checkSessionAccess, requireAccessLevel } from "@/app/share/accessControl"; import { resolveRpcForwardTimeoutMs } from "./rpcForwardTimeout"; import { resolveRpcMethodAvailabilityGraceMs, resolveRpcMethodAvailabilityPollMs } from "./rpcMethodAvailabilityGrace"; @@ -114,6 +117,20 @@ function readMachineIdPrefix(method: string): string | null { return method.slice(0, separatorIndex); } +function canRegisterMachineScopedRpcMethod(socket: Socket, method: string): boolean { + const machineId = readMachineScopedSocketMachineId(socket); + if (!machineId) return true; + const methodMachineId = readMachineIdPrefix(method); + return methodMachineId === null || methodMachineId === machineId; +} + +function isExplicitMachineStopTargetSocket(params: Readonly<{ + socket: Socket; + request: Readonly<{ machineId: string }>; +}>): boolean { + return readMachineScopedSocketMachineId(params.socket) === params.request.machineId; +} + function readExplicitMachineStopRequest(method: string, value: unknown): Readonly<{ machineId: string; sessionId: string; @@ -130,32 +147,6 @@ function readExplicitMachineStopRequest(method: string, value: unknown): Readonl return { machineId, sessionId: trimmedSessionId }; } -/** - * Session-scoped stop (`:killSession`). The runner answers this one directly, - * so unlike the machine-scoped stop there is no result the server can fence on — the only - * later signal is the publisher disconnect, which needs the recorded intent to be read as - * an intentional termination rather than an incidental drop. - */ -function readSessionScopedStopSessionId(method: string): string | null { - const separatorIndex = method.indexOf(':'); - if (separatorIndex <= 0) return null; - if (method.slice(separatorIndex + 1) !== RPC_METHODS.KILL_SESSION) return null; - const sessionId = method.slice(0, separatorIndex).trim(); - if (!sessionId || sessionId.length > MAX_RPC_METHOD_NAME_LENGTH) return null; - return sessionId; -} - -/** - * A stop the runner accepted: either proven termination or an acknowledged request. - * A transport error, refusal, or missing method is not acceptance. - */ -function isAcceptedStopResponse(targetResponse: unknown): boolean { - const strict = StopSessionResultSchema.safeParse(targetResponse); - if (strict.success) return strict.data.status === "stopped" || strict.data.status === "requested"; - if (!targetResponse || typeof targetResponse !== "object") return false; - return (targetResponse as { success?: unknown }).success === true; -} - function revalidatePrivilegedRpcTargetCompatibility(socket: Socket, method: string) { if (!resolveSocketRpcProviderStartingMethod(method)) return null; const socketData = readHappierSocketData(socket); @@ -196,7 +187,7 @@ export function rpcHandler( redisRegistry: RpcRedisRegistryConfig; sessionPublisherPresence?: Pick< ReturnType, - "captureExplicitMachineStop" | "finalizeExplicitMachineStop" | "markExplicitStopRequested" + "captureExplicitMachineStop" | "finalizeExplicitMachineStop" >; }, ) { @@ -227,7 +218,10 @@ export function rpcHandler( return; } - if (!canRegisterSessionScopedRpcMethod({ socket, method })) { + if ( + !canRegisterSessionScopedRpcMethod({ socket, method }) + || !canRegisterMachineScopedRpcMethod(socket, method) + ) { socket.emit(SOCKET_RPC_EVENTS.ERROR, { type: 'register', error: 'Forbidden' }); return; } @@ -354,7 +348,6 @@ export function rpcHandler( let { targetUserId, targetSocket } = targetResolution; const explicitMachineStopRequest = readExplicitMachineStopRequest(method, rpcAuthorization); - const sessionScopedStopSessionId = readSessionScopedStopSessionId(method); if (method.endsWith(`:${RPC_METHODS.STOP_SESSION}`) && !explicitMachineStopRequest) { callback?.({ ok: false, @@ -369,6 +362,12 @@ export function rpcHandler( method, params: callParams, ...(rpcAuthorization ? { authorization: rpcAuthorization } : {}), + ...(explicitMachineStopRequest + ? { + transportResponseEnvelopeVersion: + SOCKET_RPC_TRANSPORT_RESPONSE_ENVELOPE_VERSION_V1, + } + : {}), }); const lookupInMemoryTargetSocket = (): Socket | null => { if (targetUserId === userId && !allRpcListeners.has(userId)) { @@ -376,28 +375,23 @@ export function rpcHandler( } return allRpcListeners.get(targetUserId)?.get(method) ?? null; }; - // Owns the explicit-stop lifecycle — intent recording and machine-stop - // finalization — not just the shape of the caller's response. Every successful - // forward must run it, including one the caller made without an acknowledgement. + // Explicit machine-stop lifecycle work is independent from whether the caller + // supplied an acknowledgement callback, so every successful forward runs it. const forwardTargetResponse = async (targetResponse: unknown) => { - const forwarded = forwardedRpcTargetResponse({ method, targetResponse }); - // Only a stop the runner accepted may explain a later disconnect. Recording - // intent for an attempt that failed would let an unrelated incidental - // disconnect end a session whose runner is still alive. - const acceptedStopSessionId = sessionScopedStopSessionId ?? explicitMachineStopRequest?.sessionId ?? null; - if (acceptedStopSessionId && isAcceptedStopResponse(targetResponse)) { - // The intent is durable, so it has to be committed before the caller - // learns the stop was accepted — the publisher disconnect that reads it - // can arrive as soon as the runner starts tearing down. - await ctx.sessionPublisherPresence?.markExplicitStopRequested({ - sessionId: acceptedStopSessionId, - }); - } + const envelope = explicitMachineStopRequest + ? SocketRpcTransportResponseEnvelopeV1Schema.safeParse(targetResponse) + : null; + const targetResult = envelope?.success ? envelope.data.result : targetResponse; + const forwarded = forwardedRpcTargetResponse({ method, targetResponse: targetResult }); if (!explicitMachineStopRequest || explicitMachineStopCapture?.status !== "captured") { return forwarded; } - const stopResult = StopSessionResultSchema.safeParse(targetResponse); - if (!stopResult.success || stopResult.data.status !== "stopped") return forwarded; + const didProveStopped = ( + envelope?.success + && envelope.data.acknowledgement?.kind === "session.stop" + && envelope.data.acknowledgement.status === "stopped" + ); + if (!didProveStopped) return forwarded; const presence = ctx.sessionPublisherPresence; if (!presence) { return { @@ -489,7 +483,17 @@ export function rpcHandler( targetSocketId = awaited.targetSocketId; targetSocket = awaited.targetSocket ?? targetSocket; } - const fallbackSocket = targetSocket ?? lookupInMemoryTargetSocket(); + const fallbackCandidate = targetSocket ?? lookupInMemoryTargetSocket(); + const fallbackSocket = ( + fallbackCandidate + && ( + !explicitMachineStopRequest + || !targetSocketId + || fallbackCandidate.id === targetSocketId + ) + ) + ? fallbackCandidate + : null; if (fallbackSocket && fallbackSocket.connected) { if (fallbackSocket === socket) { if (callback) { @@ -502,6 +506,20 @@ export function rpcHandler( } const fallbackMachineId = readMachineScopedSocketMachineId(fallbackSocket); + if ( + explicitMachineStopRequest + && !isExplicitMachineStopTargetSocket({ + socket: fallbackSocket, + request: explicitMachineStopRequest, + }) + ) { + callback?.({ + ok: false, + error: RPC_ERROR_MESSAGES.METHOD_NOT_AVAILABLE, + errorCode: RPC_ERROR_CODES.METHOD_NOT_AVAILABLE, + }); + return; + } if (fallbackMachineId) { const fallbackMachine = await validateCurrentMachineSocket({ accountId: targetUserId, @@ -560,7 +578,7 @@ export function rpcHandler( } attemptedTargetSocketId = targetSocketId; - if (resolveSocketRpcProviderStartingMethod(method)) { + if (resolveSocketRpcProviderStartingMethod(method) || explicitMachineStopRequest) { const currentTargets = await ctx.io.in(targetSocketId).fetchSockets(); const currentTarget = currentTargets.find((candidate) => candidate.id === targetSocketId); if (!currentTarget) { @@ -573,6 +591,20 @@ export function rpcHandler( return; } const currentMachineId = readMachineScopedSocketMachineId(currentTarget as unknown as Socket); + if ( + explicitMachineStopRequest + && !isExplicitMachineStopTargetSocket({ + socket: currentTarget as unknown as Socket, + request: explicitMachineStopRequest, + }) + ) { + callback?.({ + ok: false, + error: RPC_ERROR_MESSAGES.METHOD_NOT_AVAILABLE, + errorCode: RPC_ERROR_CODES.METHOD_NOT_AVAILABLE, + }); + return; + } if (!currentMachineId) { const upgradeRequired = buildPrivilegedRpcUpgradeRequiredResponse( currentTarget as unknown as Socket, @@ -666,6 +698,20 @@ export function rpcHandler( } const targetMachineId = readMachineScopedSocketMachineId(targetSocket); + if ( + explicitMachineStopRequest + && !isExplicitMachineStopTargetSocket({ + socket: targetSocket, + request: explicitMachineStopRequest, + }) + ) { + callback?.({ + ok: false, + error: RPC_ERROR_MESSAGES.METHOD_NOT_AVAILABLE, + errorCode: RPC_ERROR_CODES.METHOD_NOT_AVAILABLE, + }); + return; + } if (targetMachineId) { const targetMachine = await validateCurrentMachineSocket({ accountId: targetUserId, diff --git a/apps/server/sources/app/presence/sessionPublisherPresence.sqlite.integration.spec.ts b/apps/server/sources/app/presence/sessionPublisherPresence.sqlite.integration.spec.ts index f911fe48da..0d3504641e 100644 --- a/apps/server/sources/app/presence/sessionPublisherPresence.sqlite.integration.spec.ts +++ b/apps/server/sources/app/presence/sessionPublisherPresence.sqlite.integration.spec.ts @@ -452,117 +452,9 @@ describe("session publisher presence on SQLite", () => { runtimeActivityState: "unknown", runtimeActivityActiveCount: 0, }); - }); - - it("closes the exact publisher on disconnect after an explicit stop was requested", async () => { - const seeded = await seed(); - const presence = createSessionPublisherPresence({ now: () => new Date(seeded.fence.getTime() + 10) }); - const socket = {}; - const registered = await presence.registerPublisher({ - socket, - binding: seeded.binding, - completeActivitySnapshot: { state: "active", activeCount: 1 }, - }); - if (registered.status !== "registered") throw new Error("expected registration"); - - // An explicit stop was requested but the runner died before it could prove - // physical termination, so only the disconnect reaches the server. - await presence.markExplicitStopRequested({ sessionId: seeded.binding.sessionId }); - - const disconnected = await presence.forgetDisconnectedPublisher({ socket }); - // A terminal close carries participant cursors; the bare already-closed marker - // does not, so the fanout is what proves the session actually ended here. - if (!("participantCursors" in disconnected)) throw new Error("expected a terminal close with fanout"); - expect(disconnected.status).toBe("closed"); - expect(disconnected.participantCursors.map((cursor) => cursor.accountId).sort()) - .toEqual(seeded.participantIds); - await expect(db.session.findUniqueOrThrow({ - where: { id: seeded.binding.sessionId }, - select: { - active: true, - lastActiveAt: true, - runtimeActivityState: true, - runtimeActivityActiveCount: true, - }, - })).resolves.toEqual({ - active: false, - lastActiveAt: registered.committedFence, - runtimeActivityState: "unknown", - runtimeActivityActiveCount: 0, - }); - }); - it("closes the publisher on disconnect when another server instance accepted the stop", async () => { - const seeded = await seed(); - const now = () => new Date(seeded.fence.getTime() + 10); - // With a Redis RPC registry the stop call and the runner's publisher socket land on - // different server instances, so the intent only reaches the disconnect if it is - // durable rather than held in the accepting instance's memory. - const publisherInstance = createSessionPublisherPresence({ now }); - const rpcInstance = createSessionPublisherPresence({ now }); - const socket = {}; - const registered = await publisherInstance.registerPublisher({ - socket, - binding: seeded.binding, - completeActivitySnapshot: { state: "active", activeCount: 1 }, - }); - if (registered.status !== "registered") throw new Error("expected registration"); - - await rpcInstance.markExplicitStopRequested({ sessionId: seeded.binding.sessionId }); - - const disconnected = await publisherInstance.forgetDisconnectedPublisher({ socket }); - if (!("participantCursors" in disconnected)) throw new Error("expected a terminal close with fanout"); - expect(disconnected.status).toBe("closed"); - expect(disconnected.participantCursors.map((cursor) => cursor.accountId).sort()) - .toEqual(seeded.participantIds); - await expect(db.session.findUniqueOrThrow({ - where: { id: seeded.binding.sessionId }, - select: { active: true, lastActiveAt: true, stopRequestedAt: true }, - })).resolves.toEqual({ - active: false, - lastActiveAt: registered.committedFence, - // Consumed by the disconnect it explained, so it cannot end a later publisher. - stopRequestedAt: null, - }); - }); - - it("does not let a successor publisher's disconnect consume a stop intent recorded for its predecessor", async () => { - const seeded = await seed(); - let now = new Date(seeded.fence.getTime() + 10); - const presence = createSessionPublisherPresence({ now: () => now }); - const predecessor = {}; - const first = await presence.registerPublisher({ - socket: predecessor, - binding: seeded.binding, - completeActivitySnapshot: { state: "active", activeCount: 1 }, - }); - if (first.status !== "registered") throw new Error("expected first registration"); - - // The stop was accepted but never killed the runner, so its intent is still on the - // row when a successor takes the session over. - await presence.markExplicitStopRequested({ sessionId: seeded.binding.sessionId }); - - now = new Date(first.committedFence.getTime() + 10); - const successorSocket = {}; - const successor = await presence.registerPublisher({ - socket: successorSocket, - binding: seeded.binding, - completeActivitySnapshot: { state: "active", activeCount: 1 }, - }); - if (successor.status !== "registered") throw new Error("expected successor registration"); - - // The successor's own disconnect is incidental. Reading the predecessor's intent - // here would end a session whose runner never agreed to stop. - const disconnected = await presence.forgetDisconnectedPublisher({ socket: successorSocket }); - expect(disconnected.status).not.toBe("closed"); - await expect(db.session.findUniqueOrThrow({ - where: { id: seeded.binding.sessionId }, - select: { active: true, lastActiveAt: true, stopRequestedAt: true }, - })).resolves.toEqual({ - active: true, - lastActiveAt: successor.committedFence, - stopRequestedAt: null, - }); + await expect(presence.finalizeExplicitMachineStop({ target: captured.target })) + .resolves.toEqual({ status: "already_inactive" }); }); it("does not let a completed explicit stop close a successor publisher that registered meanwhile", async () => { diff --git a/apps/server/sources/app/presence/sessionPublisherPresence.ts b/apps/server/sources/app/presence/sessionPublisherPresence.ts index 830ef0604e..eaaebcfdb4 100644 --- a/apps/server/sources/app/presence/sessionPublisherPresence.ts +++ b/apps/server/sources/app/presence/sessionPublisherPresence.ts @@ -19,7 +19,7 @@ import { writeSessionRuntimeActivityObserverLossInTx, writeSessionRuntimeActivityProjectionInTx, } from "@/app/session/runtimeActivity/writeProjection"; -import { inTx, type Tx } from "@/storage/inTx"; +import { inTx } from "@/storage/inTx"; import { blockInheritedProviderDeliveryClaims } from "@/app/session/pending/providerDeliveryClaimStaleness"; import { applyLatestSessionTurnEndInTx } from "@/app/session/sessionWriteService"; @@ -95,33 +95,6 @@ function publisherIntentKey(binding: PublisherBinding, snapshot: SessionRuntimeA class RegistrationContentionError extends Error {} -/** - * How long an explicit stop request stays able to explain a publisher disconnect. - * Comfortably longer than a stop round trip, far shorter than the presence timeout - * fence so a stale intent can never close an unrelated later publisher. - */ -const EXPLICIT_STOP_REQUEST_TTL_MS = 2 * 60 * 1000; - -/** - * Read-and-clear the durable stop intent for the disconnect that is being handled. - * Clearing is unconditional: an intent older than the window no longer explains a - * disconnect, and leaving it behind would let a much later incidental drop end a - * session whose runner is alive. - */ -async function consumeExplicitStopRequestInTx(params: Readonly<{ - tx: Tx; - sessionId: string; - stopRequestedAt: Date | null; - at: Date; -}>): Promise { - if (params.stopRequestedAt === null) return false; - await params.tx.session.updateMany({ - where: { id: params.sessionId, stopRequestedAt: params.stopRequestedAt }, - data: { stopRequestedAt: null }, - }); - return params.at.getTime() - params.stopRequestedAt.getTime() <= EXPLICIT_STOP_REQUEST_TTL_MS; -} - export function createSessionPublisherPresence(options: Readonly<{ now?: () => Date }> = {}) { const now = options.now ?? (() => new Date()); const registrations = new WeakMap(); @@ -129,20 +102,6 @@ export function createSessionPublisherPresence(options: Readonly<{ now?: () => D const closeResults = new WeakMap>(); const operationTails = new WeakMap>(); - // A stop that never proves physical termination reaches the server only as a publisher - // disconnect, and with a Redis RPC registry that disconnect can land on a different - // instance than the accepting call. The intent lives on the session row so it survives - // the hop from stop request to disconnect. - const markExplicitStopRequested = async (params: Readonly<{ sessionId: string }>): Promise => { - const at = now(); - await inTx(async (tx) => { - await tx.session.updateMany({ - where: { id: params.sessionId }, - data: { stopRequestedAt: at }, - }); - }); - }; - const serialize = async (socket: object, operation: () => Promise): Promise => { const prior = operationTails.get(socket) ?? Promise.resolve(); const result = prior.catch(() => {}).then(operation); @@ -178,10 +137,7 @@ export function createSessionPublisherPresence(options: Readonly<{ now?: () => D archivedAt: null, lastActiveAt: session.lastActiveAt, }, - // A registering publisher is a new binding, so any stop intent left by a - // predecessor no longer explains anything. Clearing it here is what keeps - // the successor's own later disconnect from being read as that stop. - data: { active: true, lastActiveAt: committedFence, stopRequestedAt: null }, + data: { active: true, lastActiveAt: committedFence }, }); if (updated.count === 0) throw new RegistrationContentionError(); const participantCursors = await markSessionParticipantsChanged({ tx, sessionId: binding.sessionId }); @@ -241,13 +197,11 @@ export function createSessionPublisherPresence(options: Readonly<{ now?: () => D } }; - const closeBindingAtFenceInTx = async (params: Readonly<{ - tx: Tx; + const closeBindingAtFence = async (params: Readonly<{ binding: PublisherBinding; committedFence: Date; mutationId: string; - }>): Promise => { - const tx = params.tx; + }>): Promise => await inTx(async (tx): Promise => { const session = await tx.session.findUnique({ where: { id: params.binding.sessionId }, select: { active: true, archivedAt: true, lastActiveAt: true }, @@ -258,6 +212,7 @@ export function createSessionPublisherPresence(options: Readonly<{ now?: () => D } if (session.archivedAt !== null) return { status: "rejected", reason: "archived" }; if (session.lastActiveAt.getTime() !== params.committedFence.getTime()) return { status: "superseded" }; + if (!session.active) return { status: "already_inactive" }; const turnResult = await applyLatestSessionTurnEndInTx({ tx, sessionId: params.binding.sessionId, @@ -304,15 +259,7 @@ export function createSessionPublisherPresence(options: Readonly<{ now?: () => D } : {}), }; - }; - - const closeBindingAtFence = async (params: Readonly<{ - binding: PublisherBinding; - committedFence: Date; - mutationId: string; - }>): Promise => await inTx( - async (tx) => await closeBindingAtFenceInTx({ tx, ...params }), - ); + }); const captureExplicitMachineStop = async (params: Readonly<{ binding: PublisherBinding; @@ -544,7 +491,6 @@ export function createSessionPublisherPresence(options: Readonly<{ now?: () => D resolveCurrentPublisher, runAsCurrentPublisher, publishSnapshot, - markExplicitStopRequested, forgetDisconnectedPublisher: async (params: Readonly<{ socket: object }>) => await serialize(params.socket, async () => { try { const registration = registrations.get(params.socket); @@ -553,33 +499,13 @@ export function createSessionPublisherPresence(options: Readonly<{ now?: () => D return await inTx(async (tx) => { const session = await tx.session.findUnique({ where: { id: registration.binding.sessionId }, - select: { active: true, archivedAt: true, lastActiveAt: true, stopRequestedAt: true }, + select: { active: true, archivedAt: true, lastActiveAt: true }, }); if (!session) return { status: "rejected", reason: "not_found" } as const; if (!await hasCurrentSessionScopedMachineAccessInTx({ tx, ...registration.binding })) { return { status: "rejected", reason: "unauthorized" } as const; } if (session.archivedAt !== null) return { status: "rejected", reason: "archived" } as const; - // An explicit stop is an intentional termination, so the disconnect it - // produces ends the session now. Without this the row stays active until - // the presence timeout fence expires, which blocks archive for that whole - // window. A superseded fence means a successor publisher owns the session, - // so fall through and only record observer loss. - const explicitStopRequested = await consumeExplicitStopRequestInTx({ - tx, - sessionId: registration.binding.sessionId, - stopRequestedAt: session.stopRequestedAt, - at: now(), - }); - if (explicitStopRequested) { - const closed = await closeBindingAtFenceInTx({ - tx, - binding: registration.binding, - committedFence: registration.committedFence, - mutationId: `explicit-stop-disconnect:${registration.committedFence.getTime()}`, - }); - if (closed.status === "closed") return closed; - } if (session.lastActiveAt.getTime() !== registration.committedFence.getTime()) { return { status: "rejected", reason: "superseded" } as const; } From a83b18793d9ddd2bd136e7d708209779ba2fa26b Mon Sep 17 00:00:00 2001 From: Leeroy Brun Date: Mon, 3 Aug 2026 19:13:23 +0200 Subject: [PATCH 7/7] refactor(server-rpc): centralize stop target checks --- .../sources/app/api/socket/rpcHandler.ts | 71 +++++++++---------- packages/protocol/src/socketRpc.test.ts | 24 +++++++ 2 files changed, 59 insertions(+), 36 deletions(-) diff --git a/apps/server/sources/app/api/socket/rpcHandler.ts b/apps/server/sources/app/api/socket/rpcHandler.ts index f326af3115..0fa8a0f265 100644 --- a/apps/server/sources/app/api/socket/rpcHandler.ts +++ b/apps/server/sources/app/api/socket/rpcHandler.ts @@ -131,6 +131,23 @@ function isExplicitMachineStopTargetSocket(params: Readonly<{ return readMachineScopedSocketMachineId(params.socket) === params.request.machineId; } +function readExplicitMachineStopTargetMismatch(params: Readonly<{ + socket: Socket; + request: Readonly<{ machineId: string }> | null; +}>): Readonly<{ ok: false; error: string; errorCode: string }> | null { + if (!params.request || isExplicitMachineStopTargetSocket({ + socket: params.socket, + request: params.request, + })) { + return null; + } + return { + ok: false, + error: RPC_ERROR_MESSAGES.METHOD_NOT_AVAILABLE, + errorCode: RPC_ERROR_CODES.METHOD_NOT_AVAILABLE, + }; +} + function readExplicitMachineStopRequest(method: string, value: unknown): Readonly<{ machineId: string; sessionId: string; @@ -506,18 +523,12 @@ export function rpcHandler( } const fallbackMachineId = readMachineScopedSocketMachineId(fallbackSocket); - if ( - explicitMachineStopRequest - && !isExplicitMachineStopTargetSocket({ - socket: fallbackSocket, - request: explicitMachineStopRequest, - }) - ) { - callback?.({ - ok: false, - error: RPC_ERROR_MESSAGES.METHOD_NOT_AVAILABLE, - errorCode: RPC_ERROR_CODES.METHOD_NOT_AVAILABLE, - }); + const fallbackStopTargetMismatch = readExplicitMachineStopTargetMismatch({ + socket: fallbackSocket, + request: explicitMachineStopRequest, + }); + if (fallbackStopTargetMismatch) { + callback?.(fallbackStopTargetMismatch); return; } if (fallbackMachineId) { @@ -591,18 +602,12 @@ export function rpcHandler( return; } const currentMachineId = readMachineScopedSocketMachineId(currentTarget as unknown as Socket); - if ( - explicitMachineStopRequest - && !isExplicitMachineStopTargetSocket({ - socket: currentTarget as unknown as Socket, - request: explicitMachineStopRequest, - }) - ) { - callback?.({ - ok: false, - error: RPC_ERROR_MESSAGES.METHOD_NOT_AVAILABLE, - errorCode: RPC_ERROR_CODES.METHOD_NOT_AVAILABLE, - }); + const currentStopTargetMismatch = readExplicitMachineStopTargetMismatch({ + socket: currentTarget as unknown as Socket, + request: explicitMachineStopRequest, + }); + if (currentStopTargetMismatch) { + callback?.(currentStopTargetMismatch); return; } if (!currentMachineId) { @@ -698,18 +703,12 @@ export function rpcHandler( } const targetMachineId = readMachineScopedSocketMachineId(targetSocket); - if ( - explicitMachineStopRequest - && !isExplicitMachineStopTargetSocket({ - socket: targetSocket, - request: explicitMachineStopRequest, - }) - ) { - callback?.({ - ok: false, - error: RPC_ERROR_MESSAGES.METHOD_NOT_AVAILABLE, - errorCode: RPC_ERROR_CODES.METHOD_NOT_AVAILABLE, - }); + const stopTargetMismatch = readExplicitMachineStopTargetMismatch({ + socket: targetSocket, + request: explicitMachineStopRequest, + }); + if (stopTargetMismatch) { + callback?.(stopTargetMismatch); return; } if (targetMachineId) { diff --git a/packages/protocol/src/socketRpc.test.ts b/packages/protocol/src/socketRpc.test.ts index cc449bd83d..e0d1db542b 100644 --- a/packages/protocol/src/socketRpc.test.ts +++ b/packages/protocol/src/socketRpc.test.ts @@ -32,4 +32,28 @@ describe('SocketRpcTransportResponseEnvelopeV1Schema', () => { acknowledgement: { kind: 'session.stop', status: 'stopped' }, }).success).toBe(false); }); + + it('rejects unsupported envelope versions', () => { + expect(SocketRpcTransportResponseEnvelopeV1Schema.safeParse({ + v: 2, + result: 'opaque-encrypted-result', + }).success).toBe(false); + }); + + it('rejects unknown envelope and acknowledgement fields', () => { + expect(SocketRpcTransportResponseEnvelopeV1Schema.safeParse({ + v: 1, + result: 'opaque-encrypted-result', + extra: true, + }).success).toBe(false); + expect(SocketRpcTransportResponseEnvelopeV1Schema.safeParse({ + v: 1, + result: 'opaque-encrypted-result', + acknowledgement: { + kind: 'session.stop', + status: 'stopped', + extra: true, + }, + }).success).toBe(false); + }); });