diff --git a/.changeset/quiet-rejected-reviews.md b/.changeset/quiet-rejected-reviews.md new file mode 100644 index 000000000..8d18362a4 --- /dev/null +++ b/.changeset/quiet-rejected-reviews.md @@ -0,0 +1,5 @@ +--- +"hunkdiff": patch +--- + +Stop repeatedly resubmitting rejected reviews to an unchanged healthy session daemon while preserving recovery after review replacement or daemon changes. diff --git a/packages/hunk/src/session/broker/brokerClient.test.ts b/packages/hunk/src/session/broker/brokerClient.test.ts index e08b70c12..47b22a980 100644 --- a/packages/hunk/src/session/broker/brokerClient.test.ts +++ b/packages/hunk/src/session/broker/brokerClient.test.ts @@ -9,9 +9,11 @@ import { createTestSessionSnapshot, } from "../../../../../test/helpers/session-daemon-fixtures"; import { HUNK_SESSION_API_VERSION, HUNK_SESSION_DAEMON_VERSION } from "../protocol"; +import { MAX_REGISTRATION_FILES } from "@hunk/session-broker-core"; import { SESSION_BROKER_LIFECYCLE_DEFECT_MESSAGE, SessionBroker, + SessionBrokerAuthenticator, createSessionBrokerDaemon, type SessionBrokerSocketLike, } from "@hunk/session-broker"; @@ -70,7 +72,8 @@ interface SessionBrokerConnectionTestDouble { interface SessionBrokerClientTestAccess { connection: Partial | null; credentials: DeepPartialTest | null; - waitingForIncumbentExit: boolean; + incumbentWait: { kind: "build-skew" | "payload"; fingerprint: string | null } | null; + snapshot: ReturnType; restartIncompatibleDaemon?: unknown; connect(config: ResolvedSessionBrokerConfig): void; ensureDaemonAndConnect(attempt?: unknown): Promise; @@ -173,6 +176,7 @@ function installWebSocketObserverTest( sent: string[]; emitOpen: () => void; emitClose: (code?: number, reason?: string) => void; + emitMessage: (data: unknown) => void; }> = []; let constructionAttempts = 0; @@ -204,6 +208,10 @@ function installWebSocketObserverTest( this.onopen?.(); } + emitMessage(data: unknown) { + this.onmessage?.({ data }); + } + emitClose(code = 1000, reason = "") { this.readyState = 3; this.onclose?.({ code, reason }); @@ -242,6 +250,102 @@ function prepareDirectConnectTest(client: SessionBrokerClient) { }; } +/** Authenticate observed sockets with real signed hellos while keeping retry time deterministic. */ +async function createRefusalLifecycleTest(withMetadata = false, subscribe = false) { + const runtimeDir = mkdtempSync(join(tmpdir(), "hunk-payload-refusal-")); + process.env.XDG_RUNTIME_DIR = runtimeDir; + const nativeFetch = globalThis.fetch; + const clock = new DeterministicLifecycleClockTest(); + const webSockets = installWebSocketObserverTest(); + const messages = captureBrokerWarningsTest(); + let healthChecks = 0; + let probes = 0; + globalThis.fetch = (async () => { + healthChecks += 1; + return Response.json({ ok: true }); + }) as unknown as typeof fetch; + const client = new SessionBrokerClient(createRegistration(), createSnapshot(), { + reconnectDelayMs: 10, + lifecycleClock: clock, + probeDaemonStatus: async () => { + probes += 1; + return { kind: "unsupported" }; + }, + }); + const notices: Array = []; + if (subscribe) client.subscribeConnectionNotice((notice) => notices.push(notice)); + const config = prepareDirectConnectTest(client); + const writeMetadata = (pid: number) => { + const { metadataPath } = resolveSessionBrokerRuntimePaths(config); + mkdirSync(join(metadataPath, ".."), { recursive: true }); + writeFileSync( + metadataPath, + JSON.stringify({ + pid, + host: config.host, + port: config.port, + command: "hunk-test", + args: ["daemon", "serve"], + launchedAt: new Date(pid).toISOString(), + launchedByPid: pid, + launchCwd: runtimeDir, + }), + ); + }; + const cleanup = () => { + client.stop(); + globalThis.fetch = nativeFetch; + webSockets.restoreTest(); + rmSync(runtimeDir, { recursive: true, force: true }); + }; + try { + const credentials = await loadOrCreateHunkSessionBrokerCredentials(); + clientTestAccess(client).credentials = credentials; + clientTestAccess(client).ensureDaemonAvailable = async () => {}; + const authenticator = new SessionBrokerAuthenticator({ + appId: "dev.hunk", + appRevision: HUNK_SESSION_DAEMON_VERSION, + generation: "payload-refusal-test", + daemonIdentity: credentials.daemonIdentity, + credentials: [credentials.producer], + }); + const authenticate = async (index: number) => { + const socket = webSockets.sockets[index]!; + socket.emitOpen(); + const init = JSON.parse(socket.sent[0]!); + const challenge = await authenticator.issueChallenge( + init.hello, + `${config.wsOrigin}/session`, + ); + socket.emitMessage(JSON.stringify({ type: "hello-challenge", challenge })); + await waitUntil("signed producer proof", () => socket.sent.length === 2, 1_000, 1); + const proof = JSON.parse(socket.sent[1]!); + const result = await authenticator.completeProducerHello(proof.proof, `connection-${index}`); + socket.emitMessage(JSON.stringify({ type: "hello-ack", ack: result.ack })); + await waitUntil("authenticated registration", () => socket.sent.length === 3, 1_000, 1); + expect(JSON.parse(socket.sent[2]!).type).toBe("register"); + }; + if (withMetadata) writeMetadata(100); + clientTestAccess(client).connect(config); + await authenticate(0); + return { + client, + clock, + webSockets, + messages, + notices, + authenticate, + writeMetadata, + cleanup, + healthChecks: () => healthChecks, + probes: () => probes, + }; + } catch (error) { + cleanup(); + throw error; + } +} + afterEach(() => { restoreWebSocketObserverTest?.(); @@ -535,10 +639,9 @@ describe("Hunk session daemon client", () => { } }); - // Intent: a daemon that accepted the hello but refused the registration one parser deeper is - // a version skew the user must hear about; today that close is silent. The client keeps - // reconnecting so a daemon restart recovers the window without relaunching it. - test("warns and keeps reconnecting when the daemon rejects the registration after the hello", () => { + // Authenticated payload refusals keep the reconnect scheduler, but wait for a replacement + // review or daemon before resending the rejected payload. + test("warns and waits when the daemon rejects the registration after the hello", () => { const webSockets = installWebSocketObserverTest(); const messages = captureBrokerWarningsTest(); const client = new SessionBrokerClient(createRegistration(), createSnapshot()); @@ -554,7 +657,7 @@ describe("Hunk session daemon client", () => { authenticated: true, }); expect(directive).toEqual({ reconnect: true }); - expect(clientTestAccess(client).waitingForIncumbentExit).toBe(false); + expect(clientTestAccess(client).incumbentWait?.kind).toBe("payload"); expect(client.getConnectionState()).toEqual({ status: "disconnected", notice: HUNK_DAEMON_REGISTRATION_REJECTED_MESSAGE, @@ -568,6 +671,348 @@ describe("Hunk session daemon client", () => { } }); + for (const reason of ["Incompatible session registration.", "Incompatible session snapshot."]) { + for (const metadata of [false, true]) { + for (const subscribe of [false, true]) { + test(`quiesces ${reason} metadata=${metadata} subscriber=${subscribe}`, async () => { + const fixture = await createRefusalLifecycleTest(metadata, subscribe); + const { client, clock, webSockets, messages, notices } = fixture; + try { + webSockets.sockets[0]!.emitClose(1008, reason); + const wait = clientTestAccess(client).incumbentWait; + expect(wait?.kind).toBe("payload"); + expect(wait?.fingerprint === null).toBe(!metadata); + for (let deadline = 0; deadline < 4; deadline += 1) { + await clock.advanceByTestAsync(10); + await clock.flushMicrotasksTest(30); + } + expect(fixture.healthChecks()).toBe(4); + expect(fixture.probes()).toBe(0); + expect(webSockets.sockets).toHaveLength(1); + expect(webSockets.sockets[0]!.sent).toHaveLength(3); + expect(clientTestAccess(client).incumbentWait).toBe(wait); + expect(clock.pendingCountTest()).toBe(1); + expect(client.getConnectionState()).toMatchObject({ + notice: HUNK_DAEMON_REGISTRATION_REJECTED_MESSAGE, + }); + expect(messages).toEqual( + subscribe ? [] : [`[session:broker] ${HUNK_DAEMON_REGISTRATION_REJECTED_MESSAGE}`], + ); + if (subscribe) + expect(notices).toEqual([null, HUNK_DAEMON_REGISTRATION_REJECTED_MESSAGE]); + + // UI-only updates cannot repair either refused envelope. Failed explicit replacement + // must retain the previous local pair and the exact wait ownership. + const snapshot = createSnapshot(); + client.updateSnapshot(snapshot); + const registration = client.getRegistration(); + expect(() => + client.replaceSession( + { ...registration, sessionId: "invalid-replacement" }, + createSnapshot(), + ), + ).toThrow(); + expect(client.getRegistration()).toBe(registration); + expect(clientTestAccess(client).snapshot).toBe(snapshot); + expect(clientTestAccess(client).incumbentWait).toBe(wait); + await clock.advanceByTestAsync(10); + await clock.flushMicrotasksTest(30); + expect(webSockets.sockets).toHaveLength(1); + + const replacement = { + ...registration, + info: { ...registration.info, title: "Smaller review" }, + }; + const replacementSnapshot = createSnapshot(); + client.replaceSession(replacement, replacementSnapshot); + expect(client.getRegistration()).toBe(replacement); + expect(clientTestAccess(client).snapshot).toBe(replacementSnapshot); + expect(clientTestAccess(client).incumbentWait).toBeNull(); + expect(client.getConnectionState()).toMatchObject({ + notice: HUNK_DAEMON_REGISTRATION_REJECTED_MESSAGE, + }); + await clock.advanceByTestAsync(10); + expect(webSockets.sockets).toHaveLength(2); + await fixture.authenticate(1); + expect(client.getConnectionState()).toEqual({ status: "connected" }); + expect(JSON.parse(webSockets.sockets[1]!.sent[2]!).registration.info.title).toBe( + "Smaller review", + ); + } finally { + fixture.cleanup(); + } + }); + } + } + } + + test("recovers an actual daemon file-limit rejection after explicit smaller-review replacement", async () => { + const runtimeDir = mkdtempSync(join(tmpdir(), "hunk-file-limit-recovery-")); + process.env.XDG_RUNTIME_DIR = runtimeDir; + process.env.HUNK_MCP_HOST = "127.0.0.1"; + delete process.env.HUNK_MCP_DISABLE; + const listener = createServer(); + await new Promise((resolve, reject) => { + listener.once("error", reject); + listener.listen(0, "127.0.0.1", resolve); + }); + const address = listener.address(); + const port = typeof address === "object" && address ? address.port : 0; + await new Promise((resolve) => listener.close(() => resolve())); + process.env.HUNK_MCP_PORT = String(port); + const clock = new DeterministicLifecycleClockTest(); + const registration = createRegistration(); + const oversized = { + ...registration, + info: { + ...registration.info, + files: Array.from({ length: MAX_REGISTRATION_FILES + 1 }, (_, index) => + createTestSessionReviewFile({ path: `file-${index}.ts` }), + ), + }, + }; + const client = new SessionBrokerClient(oversized, createSnapshot(), { + reconnectDelayMs: 10, + lifecycleClock: clock, + }); + const notices: Array = []; + client.subscribeConnectionNotice((notice) => notices.push(notice)); + const messages = captureBrokerWarningsTest(); + let server: Awaited> | undefined; + try { + server = await serveHunkSessionBrokerDaemon({ idleTimeoutMs: 0 }); + await client.start(); + await waitUntil( + "actual file-limit refusal", + () => clientTestAccess(client).incumbentWait?.kind === "payload", + ); + const connection = clientTestAccess(client).connection; + const refusedNotices = [...notices]; + for (let deadline = 0; deadline < 3; deadline += 1) { + clock.advanceByTest(10); + await waitUntil( + "health-only poll settlement", + () => clock.pendingCountTest() === 1, + 2_000, + 1, + ); + } + expect(notices).toEqual(refusedNotices); + expect(notices.at(-1)).toBe(HUNK_DAEMON_REGISTRATION_REJECTED_MESSAGE); + expect(messages).toEqual([]); + expect(await createHttpHunkSessionCliClient().listSessions()).toHaveLength(0); + + client.replaceSession(registration, createSnapshot()); + clock.advanceByTest(10); + await waitUntil( + "accepted smaller review", + async () => (await createHttpHunkSessionCliClient().listSessions()).length === 1, + 3_000, + 10, + ); + expect(clientTestAccess(client).connection).toBe(connection); + expect(client.getConnectionState()).toEqual({ status: "connected" }); + expect(messages).toEqual([]); + } finally { + client.stop(); + server?.stop(true); + if (server) await server.stopped; + rmSync(runtimeDir, { recursive: true, force: true }); + } + }, 10_000); + + test("retries a refused review when the healthy daemon fingerprint changes", async () => { + const fixture = await createRefusalLifecycleTest(true, true); + try { + fixture.webSockets.sockets[0]!.emitClose(1008, "Incompatible session registration."); + await fixture.clock.advanceByTestAsync(10); + await fixture.clock.flushMicrotasksTest(30); + expect(fixture.webSockets.sockets).toHaveLength(1); + fixture.writeMetadata(200); + await fixture.clock.advanceByTestAsync(10); + await fixture.clock.flushMicrotasksTest(30); + expect(fixture.webSockets.sockets).toHaveLength(2); + await fixture.authenticate(1); + expect(fixture.client.getConnectionState()).toEqual({ status: "connected" }); + } finally { + fixture.cleanup(); + } + }); + + for (const authority of ["replaceSession", "stop"] as const) { + test(`${authority} fences pending payload-refusal health`, async () => { + const fixture = await createRefusalLifecycleTest(false, true); + const health = createDeferredTest(); + const started = createDeferredTest(); + globalThis.fetch = (async () => { + started.resolve(); + return health.promise; + }) as unknown as typeof fetch; + try { + fixture.webSockets.sockets[0]!.emitClose(1008, "Incompatible session snapshot."); + fixture.clock.advanceByTest(10); + await settleWithinTestTimeout(started.promise); + if (authority === "stop") fixture.client.stop(); + else fixture.client.replaceSession(fixture.client.getRegistration(), createSnapshot()); + health.resolve(Response.json({ ok: true })); + await fixture.clock.flushMicrotasksTest(40); + expect(fixture.webSockets.sockets).toHaveLength(authority === "stop" ? 1 : 2); + expect(fixture.messages).toEqual([]); + if (authority === "replaceSession") + expect(clientTestAccess(fixture.client).incumbentWait).toBeNull(); + else expect(fixture.clock.pendingCountTest()).toBe(0); + } finally { + fixture.cleanup(); + } + }); + } + + test("late unhealthy results cannot clear a newer payload refusal", async () => { + const fixture = await createRefusalLifecycleTest(false, true); + const health = createDeferredTest(); + const started = createDeferredTest(); + globalThis.fetch = (async () => { + started.resolve(); + return health.promise; + }) as unknown as typeof fetch; + try { + fixture.webSockets.sockets[0]!.emitClose(1008, "Incompatible session registration."); + fixture.clock.advanceByTest(10); + await settleWithinTestTimeout(started.promise); + clientTestAccess(fixture.client).connection!.options!.resolveClose!({ + code: 1008, + reason: "Incompatible session snapshot.", + authenticated: true, + }); + const newerWait = clientTestAccess(fixture.client).incumbentWait; + health.resolve(new Response("unavailable", { status: 503 })); + await fixture.clock.flushMicrotasksTest(40); + expect(clientTestAccess(fixture.client).incumbentWait).toBe(newerWait); + expect(fixture.webSockets.sockets).toHaveLength(1); + expect(fixture.messages).toEqual([]); + expect(fixture.clock.pendingCountTest()).toBe(1); + } finally { + fixture.cleanup(); + } + }); + + test("transient closes reconnect, but replacement does not release a later pre-authentication wait", async () => { + const fixture = await createRefusalLifecycleTest(false, true); + try { + fixture.webSockets.sockets[0]!.emitClose(1006); + await fixture.clock.advanceByTestAsync(10); + expect(fixture.webSockets.sockets).toHaveLength(2); + fixture.webSockets.sockets[1]!.emitClose( + 1008, + "Session broker authentication required; upgrade Hunk.", + ); + const wait = clientTestAccess(fixture.client).incumbentWait; + fixture.client.replaceSession(fixture.client.getRegistration(), createSnapshot()); + expect(wait?.kind).toBe("build-skew"); + expect(clientTestAccess(fixture.client).incumbentWait).toBe(wait); + await fixture.clock.advanceByTestAsync(10); + await fixture.clock.flushMicrotasksTest(30); + expect(fixture.webSockets.sockets).toHaveLength(2); + } finally { + fixture.cleanup(); + } + }); + + test("payload refusal ownership precedes reentrant replacement and stale skew probes", async () => { + const clock = new DeterministicLifecycleClockTest(); + const webSockets = installWebSocketObserverTest(); + const probe = createDeferredTest(); + const client = new SessionBrokerClient(createRegistration(), createSnapshot(), { + lifecycleClock: clock, + probeDaemonStatus: () => probe.promise, + }); + const config = prepareDirectConnectTest(client); + client.subscribeConnectionNotice((notice) => { + if (notice === HUNK_DAEMON_REGISTRATION_REJECTED_MESSAGE) { + expect(clientTestAccess(client).incumbentWait?.kind).toBe("payload"); + client.replaceSession(client.getRegistration(), createSnapshot()); + } + }); + try { + clientTestAccess(client).connect(config); + const resolveClose = clientTestAccess(client).connection!.options!.resolveClose!; + resolveClose({ + code: 1008, + reason: "Malformed session broker protocol.", + authenticated: false, + }); + resolveClose({ + code: 1008, + reason: "Incompatible session registration.", + authenticated: true, + }); + expect(clientTestAccess(client).incumbentWait).toBeNull(); + // A newer payload refusal can also remain owned while an old skew probe finishes. + resolveClose({ code: 1008, reason: "Incompatible session snapshot.", authenticated: true }); + expect(clientTestAccess(client).incumbentWait?.kind).toBe("payload"); + probe.resolve(statusProbeTest(HUNK_SESSION_DAEMON_VERSION + 1)); + await clock.flushMicrotasksTest(); + expect(client.getConnectionState()).toMatchObject({ + notice: HUNK_DAEMON_REGISTRATION_REJECTED_MESSAGE, + }); + } finally { + client.stop(); + webSockets.restoreTest(); + } + }); + + for (const previousRefusal of [false, true]) { + test(`preserves a reentrant replacement refusal with previousRefusal=${previousRefusal}`, () => { + const clock = new DeterministicLifecycleClockTest(); + const webSockets = installWebSocketObserverTest(); + const client = new SessionBrokerClient(createRegistration(), createSnapshot(), { + lifecycleClock: clock, + }); + client.subscribeConnectionNotice(() => {}); + const config = prepareDirectConnectTest(client); + try { + clientTestAccess(client).connect(config); + const connection = clientTestAccess(client).connection; + const resolveClose = connection?.options?.resolveClose; + if (!connection || !resolveClose) throw new Error("Expected a live connection."); + if (previousRefusal) { + resolveClose({ + code: 1008, + reason: "Incompatible session registration.", + authenticated: true, + }); + } + const previousWait = clientTestAccess(client).incumbentWait; + let replacementWait: SessionBrokerClientTestAccess["incumbentWait"] = null; + connection.replaceSession = () => { + resolveClose({ + code: 1008, + reason: "Incompatible session snapshot.", + authenticated: true, + }); + replacementWait = clientTestAccess(client).incumbentWait; + }; + const registration = { + ...client.getRegistration(), + info: { ...client.getRegistration().info, title: "Replacement review" }, + }; + const snapshot = createSnapshot(); + client.replaceSession(registration, snapshot); + expect(client.getRegistration()).toBe(registration); + expect(clientTestAccess(client).snapshot).toBe(snapshot); + expect(replacementWait).not.toBeNull(); + expect(replacementWait).not.toBe(previousWait); + expect(clientTestAccess(client).incumbentWait).toBe(replacementWait); + expect(client.getConnectionState()).toMatchObject({ + notice: HUNK_DAEMON_REGISTRATION_REJECTED_MESSAGE, + }); + } finally { + client.stop(); + webSockets.restoreTest(); + } + }); + } + test("keeps its previous registration when the live connection rejects replacement", () => { const registration = createRegistration(); const client = new SessionBrokerClient(registration, createSnapshot()); @@ -725,8 +1170,8 @@ describe("Hunk session daemon client", () => { ).toBe(true); await Bun.sleep(60); expect(websocketOpens).toBe(2); - expect(clientTestAccess(client).waitingForIncumbentExit).toBe(true); - expect(clientTestAccess(skewedClient).waitingForIncumbentExit).toBe(true); + expect(clientTestAccess(client).incumbentWait?.kind).toBe("build-skew"); + expect(clientTestAccess(skewedClient).incumbentWait?.kind).toBe("build-skew"); } finally { client.stop(); skewedClient.stop(); @@ -1443,7 +1888,7 @@ describe("Hunk session daemon client", () => { 1008, "Session broker authentication required; upgrade Hunk.", ); - expect(clientTestAccess(client).waitingForIncumbentExit).toBe(true); + expect(clientTestAccess(client).incumbentWait?.kind).toBe("build-skew"); clock.advanceByTest(10); await settleWithinTestTimeout(fetchStarted.promise); @@ -1457,7 +1902,7 @@ describe("Hunk session daemon client", () => { await clock.flushMicrotasksTest(); clock.advanceByTest(20); - expect(clientTestAccess(client).waitingForIncumbentExit).toBe(true); + expect(clientTestAccess(client).incumbentWait?.kind).toBe("build-skew"); expect(messages).toEqual([`[session:broker] ${HUNK_DAEMON_UPGRADE_WAIT_MESSAGE}`]); expect(webSockets.sockets).toHaveLength(authority === "stop" ? 1 : 2); } finally { diff --git a/packages/hunk/src/session/broker/brokerClient.ts b/packages/hunk/src/session/broker/brokerClient.ts index 42f5157dd..6e4000c2b 100644 --- a/packages/hunk/src/session/broker/brokerClient.ts +++ b/packages/hunk/src/session/broker/brokerClient.ts @@ -93,6 +93,10 @@ interface ClientConnectionGeneration { readonly id: symbol; } +type IncumbentWait = + | { kind: "build-skew"; fingerprint: string | null; buildKnown: boolean } + | { kind: "payload"; fingerprint: string | null }; + type StartupLifecycleState = | { status: "idle" } | { @@ -157,10 +161,7 @@ export class SessionBrokerClient { private connectionGeneration: ClientConnectionGeneration | null = null; private lastConnectionWarning: string | null = null; private credentials: HunkSessionBrokerCredentials | null = null; - private waitingForIncumbentExit = false; - private incumbentLaunchFingerprint: string | null = null; - /** Whether the current incumbent has given a definitive answer about its build. */ - private incumbentBuildKnown = false; + private incumbentWait: IncumbentWait | null = null; private connectionState: HunkDaemonConnectionState = { status: "connected" }; private readonly noticeListeners = new Set<(notice: string | null) => void>(); private readonly lifecycleClock: SessionBrokerLifecycleClock; @@ -252,14 +253,18 @@ export class SessionBrokerClient { } /** Ask the daemon which build it is and refine the refused-hello notice by direction. */ - private refineRefusalNotice(config: ResolvedSessionBrokerConfig, isCurrent: () => boolean) { + private refineRefusalNotice( + config: ResolvedSessionBrokerConfig, + wait: Extract, + isCurrent: () => boolean, + ) { const probe = this.timing.probeDaemonStatus ?? probeHunkSessionDaemonAdminStatus; void probe(config).then( (result) => { - if (!isCurrent() || !this.waitingForIncumbentExit) return; + if (!isCurrent() || this.incumbentWait !== wait) return; // "Unavailable" is the one answer worth asking again for: it means the probe itself did // not land. A daemon that refuses the admin scope has answered definitively. - this.incumbentBuildKnown = result.kind !== "unavailable"; + wait.buildKnown = result.kind !== "unavailable"; const { direction, notice } = daemonSkewNotice(result); this.setConnectionState({ status: "disconnected", notice, direction }); }, @@ -275,9 +280,13 @@ export class SessionBrokerClient { ) { // Let the connection validate/send first. If it throws, the client keeps // serving the previous registration and snapshot as one coherent pair. + const wait = this.incumbentWait; this.connection?.replaceSession(registration, snapshot); this.registration = registration; this.snapshot = snapshot; + // Release only the refusal this replacement retries, after the local pair commits. A newer + // reentrant refusal must survive; snapshot updates alone do not release either payload refusal. + if (wait?.kind === "payload" && this.incumbentWait === wait) this.incumbentWait = null; } private resolveConfig() { @@ -401,21 +410,27 @@ export class SessionBrokerClient { ); if (!isCommitAuthorized()) return; } - if (this.waitingForIncumbentExit) { + const wait = this.incumbentWait; + if (wait) { const healthy = await isSessionBrokerHealthy(config); if (!isCommitAuthorized()) return; - if (healthy) { + // A replacement can release payload waiting without changing the socket generation. + // Never let an older health result reinstate that wait or clear a newer refusal. + const currentWait = this.incumbentWait; + if (currentWait) { const currentFingerprint = readSessionBrokerLaunchFingerprint(config); if (!isCommitAuthorized()) return; - // Owner-private metadata is only a generation-change hint. The signed hello remains the - // sole compatibility and identity authority, and unchanged/malformed metadata causes - // health-only polling so skewed waiters cannot keep the incumbent active. - if (currentFingerprint === this.incumbentLaunchFingerprint) { - throw new Error(HUNK_DAEMON_UPGRADE_WAIT_MESSAGE); + // Metadata is only a generation-change hint. Null-to-null remains quiescent too; + // the signed hello still owns compatibility and identity after a retry. + if (currentWait !== wait || (healthy && currentFingerprint === wait.fingerprint)) { + throw new Error( + currentWait.kind === "payload" + ? HUNK_DAEMON_REGISTRATION_REJECTED_MESSAGE + : HUNK_DAEMON_UPGRADE_WAIT_MESSAGE, + ); } + this.incumbentWait = null; } - if (!isCommitAuthorized()) return; - this.waitingForIncumbentExit = false; } await this.ensureDaemonAvailable(config, isCommitAuthorized); if (!isCommitAuthorized()) return; @@ -428,25 +443,31 @@ export class SessionBrokerClient { // One incumbent, one probe: a running daemon's build cannot change, so re-asking on // every reconnect would only churn caller sessions and overwrite the refined notice // with the generic one. A different launch fingerprint means a different daemon. - const sameIncumbent = - this.waitingForIncumbentExit && this.incumbentLaunchFingerprint === fingerprint; - this.waitingForIncumbentExit = true; - this.incumbentLaunchFingerprint = fingerprint; - if (!sameIncumbent) { - this.incumbentBuildKnown = false; + const previousWait = this.incumbentWait; + const wait: Extract = + previousWait?.kind === "build-skew" && previousWait.fingerprint === fingerprint + ? previousWait + : { kind: "build-skew", fingerprint, buildKnown: false }; + this.incumbentWait = wait; + if (wait !== previousWait) { this.setConnectionState({ status: "disconnected", notice: HUNK_DAEMON_UPGRADE_WAIT_MESSAGE, direction: "unknown", }); - this.refineRefusalNotice(config, () => isConnectionCurrent(brokerGeneration)); - } else if (!this.incumbentBuildKnown) { + this.refineRefusalNotice(config, wait, () => isConnectionCurrent(brokerGeneration)); + } else if (!wait.buildKnown) { // A probe that never landed would otherwise leave this window on the generic notice // for the incumbent's whole life — including when the daemon is the newer build and // closing older windows cannot help. Retry without disturbing the notice on screen. - this.refineRefusalNotice(config, () => isConnectionCurrent(brokerGeneration)); + this.refineRefusalNotice(config, wait, () => isConnectionCurrent(brokerGeneration)); } } else if (isRegistrationRejection(event)) { + // Publish ownership before subscribers can synchronously replace the review or stop. + this.incumbentWait = { + kind: "payload", + fingerprint: readSessionBrokerLaunchFingerprint(config), + }; this.setConnectionState({ status: "disconnected", notice: HUNK_DAEMON_REGISTRATION_REJECTED_MESSAGE, @@ -459,14 +480,21 @@ export class SessionBrokerClient { }, onConnected: (brokerGeneration) => { if (!isConnectionCurrent(brokerGeneration)) return; - this.waitingForIncumbentExit = false; - this.incumbentLaunchFingerprint = null; - this.incumbentBuildKnown = false; + this.incumbentWait = null; this.lastConnectionWarning = null; this.setConnectionState({ status: "connected" }); }, onWarning: (message, brokerGeneration) => { - if (isConnectionCurrent(brokerGeneration)) this.warnUnavailable(message); + if (!isConnectionCurrent(brokerGeneration)) return; + // Health-only payload polls retain the actionable notice without repeating it in the + // console or suggesting that a build upgrade will fix review-size limits. + if ( + this.incumbentWait?.kind === "payload" && + (message === HUNK_DAEMON_REGISTRATION_REJECTED_MESSAGE || + message === HUNK_DAEMON_UPGRADE_WAIT_MESSAGE) + ) + return; + this.warnUnavailable(message); }, onDefect: this.timing.onDefect, }); diff --git a/packages/hunk/src/session/client/daemonMessages.ts b/packages/hunk/src/session/client/daemonMessages.ts index 32e9c810d..dc0136c1c 100644 --- a/packages/hunk/src/session/client/daemonMessages.ts +++ b/packages/hunk/src/session/client/daemonMessages.ts @@ -29,5 +29,6 @@ export const HUNK_DAEMON_CLIENT_NEWER_MESSAGE = `Session daemon is ${HUNK_BUILD_ export const HUNK_DAEMON_CLIENT_OLDER_MESSAGE = "Session daemon is newer; relaunch this window (notes are lost)."; -/** Covers a refused registration after the daemon accepted the window's hello. */ -export const HUNK_DAEMON_REGISTRATION_REJECTED_MESSAGE = `Session daemon rejected this window. Run ${HUNK_DAEMON_RESTART_COMMAND}.`; +/** Suggests reducing a refused review without implying restart can bypass payload limits. */ +export const HUNK_DAEMON_REGISTRATION_REJECTED_MESSAGE = + "Session daemon rejected this review. Try a smaller review."; diff --git a/packages/hunk/src/session/client/daemonSkew.test.ts b/packages/hunk/src/session/client/daemonSkew.test.ts index 1ad8d0c8a..2df06b6df 100644 --- a/packages/hunk/src/session/client/daemonSkew.test.ts +++ b/packages/hunk/src/session/client/daemonSkew.test.ts @@ -66,7 +66,7 @@ describe("daemon skew notices", () => { ], [ HUNK_DAEMON_REGISTRATION_REJECTED_MESSAGE, - "Session daemon rejected this window. Run `hunk daemon restart`.", + "Session daemon rejected this review. Try a smaller review.", ], ] as const)( "fits the complete sticky notice into an 80-column status line: %s",