From 338faf4b811e7b0ec126d8c1aaf1806e1fc7736a Mon Sep 17 00:00:00 2001 From: Federico Jaramillo Martinez Date: Sun, 12 Jul 2026 09:22:19 +0200 Subject: [PATCH] perf: speed up chat loading and resume --- .changeset/faster-chat-loading.md | 5 + package-lock.json | 276 +++++++++++++++++- package.json | 1 + .../appShell/browserResumeController.test.ts | 135 +++++++++ .../src/appShell/browserResumeController.ts | 98 +++++++ src/client/src/components/ChatView.test.ts | 188 +++++++++++- src/client/src/components/ChatView.ts | 28 +- src/client/src/components/PiWebApp.ts | 44 +-- .../controllers/activityController.test.ts | 43 +++ .../src/controllers/activityController.ts | 8 +- .../sessionController.refresh.test.ts | 96 ++++++ .../src/controllers/sessionController.ts | 64 +++- .../controllers/trailingRefreshCoordinator.ts | 57 ++++ src/server/app.compression.test.ts | 90 ++++++ src/server/app.ts | 8 + src/server/browserMessageProjection.test.ts | 64 ++++ src/server/browserMessageProjection.ts | 59 ++++ src/server/machines/machineClient.test.ts | 32 ++ src/server/machines/machineClient.ts | 30 +- src/server/realtime/sessionEventHub.test.ts | 16 + src/server/realtime/sessionEventHub.ts | 3 +- .../piSessionService.lifecycle.test.ts | 164 ++++++++++- src/server/sessions/piSessionService.ts | 102 ++++++- src/server/sessions/sessionRoutes.test.ts | 33 ++- src/server/sessions/sessionRoutes.ts | 4 +- 25 files changed, 1565 insertions(+), 83 deletions(-) create mode 100644 .changeset/faster-chat-loading.md create mode 100644 src/client/src/appShell/browserResumeController.test.ts create mode 100644 src/client/src/appShell/browserResumeController.ts create mode 100644 src/client/src/controllers/sessionController.refresh.test.ts create mode 100644 src/client/src/controllers/trailingRefreshCoordinator.ts create mode 100644 src/server/app.compression.test.ts create mode 100644 src/server/browserMessageProjection.test.ts create mode 100644 src/server/browserMessageProjection.ts diff --git a/.changeset/faster-chat-loading.md b/.changeset/faster-chat-loading.md new file mode 100644 index 0000000..826311a --- /dev/null +++ b/.changeset/faster-chat-loading.md @@ -0,0 +1,5 @@ +--- +"@jmfederico/pi-web": patch +--- + +Improve chat loading and resume performance by sharing duplicate session work, compressing browser responses, trimming unused thinking signatures, and lazily rendering closed technical-event groups. diff --git a/package-lock.json b/package-lock.json index a30b33c..8a79114 100644 --- a/package-lock.json +++ b/package-lock.json @@ -22,6 +22,7 @@ "@codemirror/legacy-modes": "^6.5.3", "@codemirror/state": "^6.7.1", "@codemirror/view": "^6.43.6", + "@fastify/compress": "^9.0.0", "@fastify/static": "^9.3.0", "@fastify/websocket": "^11.3.0", "@xterm/addon-fit": "^0.11.0", @@ -3656,6 +3657,62 @@ "integrity": "sha512-NM8/P9n3XjXhIZn1lLhkFaACTOURQXjWhV4BA/RnOv8xvgqtqpAX9IO4mRQxSx1Rlo4tqzeqb0sOlruaOy3dug==", "license": "MIT" }, + "node_modules/@fastify/compress": { + "version": "9.0.0", + "resolved": "https://registry.npmjs.org/@fastify/compress/-/compress-9.0.0.tgz", + "integrity": "sha512-PZRg+ut5xd/ubsGPWfoPNryoCOtEdHboIWpDieTUHov1gKdLitF8mRmT3JbqNnRbelQXSNXUsIpakAEKR6AcTQ==", + "funding": [ + { + "type": "github", + "url": "https://github.com/sponsors/fastify" + }, + { + "type": "opencollective", + "url": "https://opencollective.com/fastify" + } + ], + "license": "MIT", + "dependencies": { + "@fastify/accept-negotiator": "^2.0.0", + "fastify-plugin": "^5.0.0", + "mime-db": "^1.52.0", + "minipass": "^7.0.4", + "peek-stream": "^1.1.3", + "readable-stream": "^4.5.2" + } + }, + "node_modules/@fastify/compress/node_modules/fastify-plugin": { + "version": "5.1.0", + "resolved": "https://registry.npmjs.org/fastify-plugin/-/fastify-plugin-5.1.0.tgz", + "integrity": "sha512-FAIDA8eovSt5qcDgcBvDuX/v0Cjz0ohGhENZ/wpc3y+oZCY2afZ9Baqql3g/lC+OHRnciQol4ww7tuthOb9idw==", + "funding": [ + { + "type": "github", + "url": "https://github.com/sponsors/fastify" + }, + { + "type": "opencollective", + "url": "https://opencollective.com/fastify" + } + ], + "license": "MIT" + }, + "node_modules/@fastify/compress/node_modules/readable-stream": { + "version": "4.7.0", + "resolved": "https://registry.npmjs.org/readable-stream/-/readable-stream-4.7.0.tgz", + "integrity": "sha512-oIGGmcpTLwPga8Bn6/Z75SVaH1z5dUut2ibSyAMVhmUggWpmDn2dapB0n7f8nwaSiRtepAsfJyfXIO5DCVAODg==", + "license": "MIT", + "dependencies": { + "abort-controller": "^3.0.0", + "buffer": "^6.0.3", + "events": "^3.3.0", + "process": "^0.11.10", + "string_decoder": "^1.3.0" + }, + "engines": { + "node": "^12.22.0 || ^14.17.0 || >=16.0.0" + } + }, "node_modules/@fastify/error": { "version": "4.2.0", "resolved": "https://registry.npmjs.org/@fastify/error/-/error-4.2.0.tgz", @@ -5835,6 +5892,18 @@ "addons/*" ] }, + "node_modules/abort-controller": { + "version": "3.0.0", + "resolved": "https://registry.npmjs.org/abort-controller/-/abort-controller-3.0.0.tgz", + "integrity": "sha512-h8lQ8tacZYnR3vNQTgibj+tODHI5/+l06Au2Pcriv/Gmet0eaj4TwWH41sO9wnHDiQsEj19q0drzdWdeAHtweg==", + "license": "MIT", + "dependencies": { + "event-target-shim": "^5.0.0" + }, + "engines": { + "node": ">=6.5" + } + }, "node_modules/abstract-logging": { "version": "2.0.1", "resolved": "https://registry.npmjs.org/abstract-logging/-/abstract-logging-2.0.1.tgz", @@ -6019,7 +6088,6 @@ "version": "1.5.1", "resolved": "https://registry.npmjs.org/base64-js/-/base64-js-1.5.1.tgz", "integrity": "sha512-AKpaYlHn8t4SVbOHCy+b5+KKgvR4vrsD8vbvrbiQJps7fKDTkjkDry6ji0rUJjC0kzbNePLwzxq8iypo41qeWA==", - "dev": true, "funding": [ { "type": "github", @@ -6091,6 +6159,30 @@ "node": ">=8" } }, + "node_modules/buffer": { + "version": "6.0.3", + "resolved": "https://registry.npmjs.org/buffer/-/buffer-6.0.3.tgz", + "integrity": "sha512-FTiCpNxtwiZZHEZbcbTIcZjERVICn9yq/pDFkTl95/AxzD1naBctN7YO68riM/gLSDY7sdrMby8hofADYuuqOA==", + "funding": [ + { + "type": "github", + "url": "https://github.com/sponsors/feross" + }, + { + "type": "patreon", + "url": "https://www.patreon.com/feross" + }, + { + "type": "consulting", + "url": "https://feross.org/support" + } + ], + "license": "MIT", + "dependencies": { + "base64-js": "^1.3.1", + "ieee754": "^1.2.1" + } + }, "node_modules/buffer-equal-constant-time": { "version": "1.0.1", "resolved": "https://registry.npmjs.org/buffer-equal-constant-time/-/buffer-equal-constant-time-1.0.1.tgz", @@ -6098,6 +6190,12 @@ "dev": true, "license": "BSD-3-Clause" }, + "node_modules/buffer-from": { + "version": "1.1.2", + "resolved": "https://registry.npmjs.org/buffer-from/-/buffer-from-1.1.2.tgz", + "integrity": "sha512-E+XQCRwSbaaiChtv6k6Dwgc+bx+Bs6vuKJHHl5kox/BaKbhiXzqQOwK4cO22yElGp2OCmjwVhT3HmxgyPGnJfQ==", + "license": "MIT" + }, "node_modules/chai": { "version": "6.2.2", "resolved": "https://registry.npmjs.org/chai/-/chai-6.2.2.tgz", @@ -6148,6 +6246,12 @@ "url": "https://opencollective.com/express" } }, + "node_modules/core-util-is": { + "version": "1.0.3", + "resolved": "https://registry.npmjs.org/core-util-is/-/core-util-is-1.0.3.tgz", + "integrity": "sha512-ZQBvi1DcpJ4GDqanjucZ2Hj3wEO5pZDS89BWbkcrvdxksJorwUDDZamX9ldFkp9aw2lmBDLgkObEA4DWNJ9FYQ==", + "license": "MIT" + }, "node_modules/crelt": { "version": "1.0.6", "resolved": "https://registry.npmjs.org/crelt/-/crelt-1.0.6.tgz", @@ -6631,6 +6735,24 @@ "node": ">=0.10.0" } }, + "node_modules/event-target-shim": { + "version": "5.0.1", + "resolved": "https://registry.npmjs.org/event-target-shim/-/event-target-shim-5.0.1.tgz", + "integrity": "sha512-i/2XbnSz/uxRCU6+NdVJgKWDTM427+MqYbkQzD321DuCQJUqOuJKIA0IM2+W2xtYHdKOmZ4dR6fExsd4SXL+WQ==", + "license": "MIT", + "engines": { + "node": ">=6" + } + }, + "node_modules/events": { + "version": "3.3.0", + "resolved": "https://registry.npmjs.org/events/-/events-3.3.0.tgz", + "integrity": "sha512-mQw+2fkQbALzQ7V0MY0IqdnXNOeTtP4r0lN9z7AAawCXgqea7bDii20AYrIBrFd/Hx0M2Ocz6S111CaFkUcb0Q==", + "license": "MIT", + "engines": { + "node": ">=0.8.x" + } + }, "node_modules/expect-type": { "version": "1.3.0", "resolved": "https://registry.npmjs.org/expect-type/-/expect-type-1.3.0.tgz", @@ -7315,6 +7437,26 @@ "url": "https://opencollective.com/express" } }, + "node_modules/ieee754": { + "version": "1.2.1", + "resolved": "https://registry.npmjs.org/ieee754/-/ieee754-1.2.1.tgz", + "integrity": "sha512-dcyqhDvX1C46lXZcVqCpK+FtMRQVdIMN6/Df5js2zouUsqG7I6sFxitIC+7KYK29KdXOLHdu9zL4sFnoVQnqaA==", + "funding": [ + { + "type": "github", + "url": "https://github.com/sponsors/feross" + }, + { + "type": "patreon", + "url": "https://www.patreon.com/feross" + }, + { + "type": "consulting", + "url": "https://feross.org/support" + } + ], + "license": "BSD-3-Clause" + }, "node_modules/ignore": { "version": "7.0.5", "resolved": "https://registry.npmjs.org/ignore/-/ignore-7.0.5.tgz", @@ -7406,6 +7548,12 @@ "node": ">=0.10.0" } }, + "node_modules/isarray": { + "version": "1.0.0", + "resolved": "https://registry.npmjs.org/isarray/-/isarray-1.0.0.tgz", + "integrity": "sha512-VLghIWNM6ELQzo7zwmcg0NmTVyWKYjvIeM83yjp0wRDTmUnrM678fQbcKBo6n2CJEF0szoG//ytg+TKla89ALQ==", + "license": "MIT" + }, "node_modules/isexe": { "version": "2.0.0", "resolved": "https://registry.npmjs.org/isexe/-/isexe-2.0.0.tgz", @@ -8054,6 +8202,15 @@ "node": ">=10.0.0" } }, + "node_modules/mime-db": { + "version": "1.54.0", + "resolved": "https://registry.npmjs.org/mime-db/-/mime-db-1.54.0.tgz", + "integrity": "sha512-aU5EJuIN2WDemCcAp2vFBfp/m4EAhWJnUNSSw0ixs7/kXbd6Pg64EmwJkNdFhB8aWt1sH2CTXrLxo/iAGV3oPQ==", + "license": "MIT", + "engines": { + "node": ">= 0.6" + } + }, "node_modules/minimatch": { "version": "10.2.5", "resolved": "https://registry.npmjs.org/minimatch/-/minimatch-10.2.5.tgz", @@ -8471,6 +8628,59 @@ "dev": true, "license": "MIT" }, + "node_modules/peek-stream": { + "version": "1.1.3", + "resolved": "https://registry.npmjs.org/peek-stream/-/peek-stream-1.1.3.tgz", + "integrity": "sha512-FhJ+YbOSBb9/rIl2ZeE/QHEsWn7PqNYt8ARAY3kIgNGOk13g9FGyIY6JIl/xB/3TFRVoTv5as0l11weORrTekA==", + "license": "MIT", + "dependencies": { + "buffer-from": "^1.0.0", + "duplexify": "^3.5.0", + "through2": "^2.0.3" + } + }, + "node_modules/peek-stream/node_modules/duplexify": { + "version": "3.7.1", + "resolved": "https://registry.npmjs.org/duplexify/-/duplexify-3.7.1.tgz", + "integrity": "sha512-07z8uv2wMyS51kKhD1KsdXJg5WQ6t93RneqRxUHnskXVtlYYkLqM0gqStQZ3pj073g687jPCHrqNfCzawLYh5g==", + "license": "MIT", + "dependencies": { + "end-of-stream": "^1.0.0", + "inherits": "^2.0.1", + "readable-stream": "^2.0.0", + "stream-shift": "^1.0.0" + } + }, + "node_modules/peek-stream/node_modules/readable-stream": { + "version": "2.3.8", + "resolved": "https://registry.npmjs.org/readable-stream/-/readable-stream-2.3.8.tgz", + "integrity": "sha512-8p0AUk4XODgIewSi0l8Epjs+EVnWiK7NoDIEGU0HhE7+ZyY8D1IMY7odu5lRrFXGg71L15KG8QrPmum45RTtdA==", + "license": "MIT", + "dependencies": { + "core-util-is": "~1.0.0", + "inherits": "~2.0.3", + "isarray": "~1.0.0", + "process-nextick-args": "~2.0.0", + "safe-buffer": "~5.1.1", + "string_decoder": "~1.1.1", + "util-deprecate": "~1.0.1" + } + }, + "node_modules/peek-stream/node_modules/safe-buffer": { + "version": "5.1.2", + "resolved": "https://registry.npmjs.org/safe-buffer/-/safe-buffer-5.1.2.tgz", + "integrity": "sha512-Gd2UZBJDkXlY7GbJxfsE8/nvKkUEU1G38c1siN6QP6a9PT9MmHB8GnpscSmMJSoF8LOIrt8ud/wPtojys4G6+g==", + "license": "MIT" + }, + "node_modules/peek-stream/node_modules/string_decoder": { + "version": "1.1.1", + "resolved": "https://registry.npmjs.org/string_decoder/-/string_decoder-1.1.1.tgz", + "integrity": "sha512-n/ShnvDi6FHbbVfviro+WojiFzv+s8MPMHBczVePfUpDJLwoLT0ht1l4YwBCbi8pJAveEEdnkHyPyTP/mzRfwg==", + "license": "MIT", + "dependencies": { + "safe-buffer": "~5.1.0" + } + }, "node_modules/picocolors": { "version": "1.1.1", "resolved": "https://registry.npmjs.org/picocolors/-/picocolors-1.1.1.tgz", @@ -8593,6 +8803,21 @@ "url": "https://github.com/prettier/prettier?sponsor=1" } }, + "node_modules/process": { + "version": "0.11.10", + "resolved": "https://registry.npmjs.org/process/-/process-0.11.10.tgz", + "integrity": "sha512-cdGef/drWFoydD1JsMzuFf8100nZl+GT+yacc2bEced5f9Rjk4z+WtFUTBu9PhOi9j/jfmBPu0mMEY4wIdAF8A==", + "license": "MIT", + "engines": { + "node": ">= 0.6.0" + } + }, + "node_modules/process-nextick-args": { + "version": "2.0.1", + "resolved": "https://registry.npmjs.org/process-nextick-args/-/process-nextick-args-2.0.1.tgz", + "integrity": "sha512-3ouUOpQhtgrbOa17J7+uxOTpITYWaGP7/AhoR3+A+/1e9skrzelGi/dXzEYyvbxubEF6Wn2ypscTKiKJFFn1ag==", + "license": "MIT" + }, "node_modules/process-warning": { "version": "5.0.0", "resolved": "https://registry.npmjs.org/process-warning/-/process-warning-5.0.0.tgz", @@ -9203,6 +9428,46 @@ "integrity": "sha512-P4nbQYQfePJxRSmY+v/KINxVucm4NF3p3s7pJveMTtom52FR4YGltUQLB8idDXwDDWW+eYrWDFbuzUnjoWHF7g==", "license": "MIT" }, + "node_modules/through2": { + "version": "2.0.5", + "resolved": "https://registry.npmjs.org/through2/-/through2-2.0.5.tgz", + "integrity": "sha512-/mrRod8xqpA+IHSLyGCQ2s8SPHiCDEeQJSep1jqLYeEUClOFG2Qsh+4FU6G9VeqpZnGW/Su8LQGc4YKni5rYSQ==", + "license": "MIT", + "dependencies": { + "readable-stream": "~2.3.6", + "xtend": "~4.0.1" + } + }, + "node_modules/through2/node_modules/readable-stream": { + "version": "2.3.8", + "resolved": "https://registry.npmjs.org/readable-stream/-/readable-stream-2.3.8.tgz", + "integrity": "sha512-8p0AUk4XODgIewSi0l8Epjs+EVnWiK7NoDIEGU0HhE7+ZyY8D1IMY7odu5lRrFXGg71L15KG8QrPmum45RTtdA==", + "license": "MIT", + "dependencies": { + "core-util-is": "~1.0.0", + "inherits": "~2.0.3", + "isarray": "~1.0.0", + "process-nextick-args": "~2.0.0", + "safe-buffer": "~5.1.1", + "string_decoder": "~1.1.1", + "util-deprecate": "~1.0.1" + } + }, + "node_modules/through2/node_modules/safe-buffer": { + "version": "5.1.2", + "resolved": "https://registry.npmjs.org/safe-buffer/-/safe-buffer-5.1.2.tgz", + "integrity": "sha512-Gd2UZBJDkXlY7GbJxfsE8/nvKkUEU1G38c1siN6QP6a9PT9MmHB8GnpscSmMJSoF8LOIrt8ud/wPtojys4G6+g==", + "license": "MIT" + }, + "node_modules/through2/node_modules/string_decoder": { + "version": "1.1.1", + "resolved": "https://registry.npmjs.org/string_decoder/-/string_decoder-1.1.1.tgz", + "integrity": "sha512-n/ShnvDi6FHbbVfviro+WojiFzv+s8MPMHBczVePfUpDJLwoLT0ht1l4YwBCbi8pJAveEEdnkHyPyTP/mzRfwg==", + "license": "MIT", + "dependencies": { + "safe-buffer": "~5.1.0" + } + }, "node_modules/tinybench": { "version": "2.9.0", "resolved": "https://registry.npmjs.org/tinybench/-/tinybench-2.9.0.tgz", @@ -9688,6 +9953,15 @@ } } }, + "node_modules/xtend": { + "version": "4.0.2", + "resolved": "https://registry.npmjs.org/xtend/-/xtend-4.0.2.tgz", + "integrity": "sha512-LKYU1iAXJXUgAXn9URjiu+MWhyUXHsvfp7mcuYm9dSUKK0/CjtrUwFAxD82/mCWbtLsGjFIad0wIsod4zrTAEQ==", + "license": "MIT", + "engines": { + "node": ">=0.4" + } + }, "node_modules/yaml": { "version": "2.9.0", "resolved": "https://registry.npmjs.org/yaml/-/yaml-2.9.0.tgz", diff --git a/package.json b/package.json index 8330ea6..115719b 100644 --- a/package.json +++ b/package.json @@ -65,6 +65,7 @@ "@codemirror/legacy-modes": "^6.5.3", "@codemirror/state": "^6.7.1", "@codemirror/view": "^6.43.6", + "@fastify/compress": "^9.0.0", "@fastify/static": "^9.3.0", "@fastify/websocket": "^11.3.0", "@xterm/addon-fit": "^0.11.0", diff --git a/src/client/src/appShell/browserResumeController.test.ts b/src/client/src/appShell/browserResumeController.test.ts new file mode 100644 index 0000000..e5ff3b0 --- /dev/null +++ b/src/client/src/appShell/browserResumeController.test.ts @@ -0,0 +1,135 @@ +import { describe, expect, it, vi } from "vitest"; +import { BrowserResumeController } from "./browserResumeController"; + +function deferred(): { promise: Promise; resolve: (value: T) => void } { + let resolveDeferred: ((value: T) => void) | undefined; + const promise = new Promise((resolve) => { resolveDeferred = resolve; }); + if (resolveDeferred === undefined) throw new Error("Deferred promise was not initialized"); + return { promise, resolve: resolveDeferred }; +} + +function frameHarness() { + const frames: { callback: () => void; canceled: boolean }[] = []; + return { + scheduleFrame: (callback: () => void) => { + const frame = { callback, canceled: false }; + frames.push(frame); + return { cancel: () => { frame.canceled = true; } }; + }, + pendingCount: () => frames.filter((frame) => !frame.canceled).length, + runNext: () => { + const frame = frames.shift(); + if (frame === undefined) throw new Error("No scheduled frame"); + if (!frame.canceled) frame.callback(); + }, + }; +} + +describe("BrowserResumeController", () => { + it("batches overlapping focus and visible signals into one app refresh", async () => { + const windowTarget = new EventTarget(); + const documentTarget = new EventTarget(); + const frames = frameHarness(); + const refreshGate = deferred(); + const refreshStarted = deferred(); + const refreshCompleted = deferred(); + const onResumeSignal = vi.fn(); + let visible = true; + let refreshCalls = 0; + const controller = new BrowserResumeController({ + onResumeSignal, + refreshAfterResume: async () => { + refreshCalls += 1; + refreshStarted.resolve(undefined); + await refreshGate.promise; + refreshCompleted.resolve(undefined); + }, + onRefreshError: (error) => { throw error; }, + }, { + windowTarget, + documentTarget, + isDocumentVisible: () => visible, + scheduleFrame: frames.scheduleFrame, + }); + controller.connect(); + + windowTarget.dispatchEvent(new Event("focus")); + documentTarget.dispatchEvent(new Event("visibilitychange")); + windowTarget.dispatchEvent(new Event("focus")); + + expect(onResumeSignal).toHaveBeenCalledTimes(3); + expect(frames.pendingCount()).toBe(1); + expect(refreshCalls).toBe(0); + + frames.runNext(); + await refreshStarted.promise; + expect(refreshCalls).toBe(1); + + visible = false; + documentTarget.dispatchEvent(new Event("visibilitychange")); + expect(onResumeSignal).toHaveBeenCalledTimes(3); + expect(frames.pendingCount()).toBe(0); + + refreshGate.resolve(undefined); + await refreshCompleted.promise; + windowTarget.dispatchEvent(new Event("focus")); + expect(frames.pendingCount()).toBe(1); + controller.disconnect(); + frames.runNext(); + await Promise.resolve(); + windowTarget.dispatchEvent(new Event("focus")); + expect(onResumeSignal).toHaveBeenCalledTimes(4); + expect(refreshCalls).toBe(1); + }); + + it("runs one trailing refresh when another resume arrives during active work", async () => { + const windowTarget = new EventTarget(); + const documentTarget = new EventTarget(); + const frames = frameHarness(); + const firstGate = deferred(); + const secondGate = deferred(); + const firstStarted = deferred(); + const secondStarted = deferred(); + const secondCompleted = deferred(); + let refreshCalls = 0; + const controller = new BrowserResumeController({ + onResumeSignal: () => undefined, + refreshAfterResume: async () => { + refreshCalls += 1; + if (refreshCalls === 1) { + firstStarted.resolve(undefined); + await firstGate.promise; + return; + } + secondStarted.resolve(undefined); + await secondGate.promise; + secondCompleted.resolve(undefined); + }, + onRefreshError: (error) => { throw error; }, + }, { + windowTarget, + documentTarget, + isDocumentVisible: () => true, + scheduleFrame: frames.scheduleFrame, + }); + controller.connect(); + + windowTarget.dispatchEvent(new Event("focus")); + frames.runNext(); + await firstStarted.promise; + + documentTarget.dispatchEvent(new Event("visibilitychange")); + windowTarget.dispatchEvent(new Event("focus")); + expect(frames.pendingCount()).toBe(1); + frames.runNext(); + expect(refreshCalls).toBe(1); + + firstGate.resolve(undefined); + await secondStarted.promise; + expect(refreshCalls).toBe(2); + + secondGate.resolve(undefined); + await secondCompleted.promise; + controller.disconnect(); + }); +}); diff --git a/src/client/src/appShell/browserResumeController.ts b/src/client/src/appShell/browserResumeController.ts new file mode 100644 index 0000000..638c1bf --- /dev/null +++ b/src/client/src/appShell/browserResumeController.ts @@ -0,0 +1,98 @@ +import { TrailingRefreshCoordinator } from "../controllers/trailingRefreshCoordinator"; + +interface BrowserEventTarget { + addEventListener(type: string, listener: EventListener): void; + removeEventListener(type: string, listener: EventListener): void; +} + +interface ScheduledFrame { + cancel(): void; +} + +export interface BrowserResumeCallbacks { + onResumeSignal(): void; + refreshAfterResume(): void | Promise; + onRefreshError(error: unknown): void; +} + +export interface BrowserResumeControllerOptions { + windowTarget?: BrowserEventTarget | undefined; + documentTarget?: BrowserEventTarget | undefined; + isDocumentVisible?: (() => boolean) | undefined; + scheduleFrame?: ((callback: () => void) => ScheduledFrame) | undefined; +} + +/** Owns browser resume listeners and batches focus/visibility refreshes per frame. */ +export class BrowserResumeController { + private readonly windowTarget: BrowserEventTarget | undefined; + private readonly documentTarget: BrowserEventTarget | undefined; + private readonly isDocumentVisible: () => boolean; + private readonly scheduleFrame: (callback: () => void) => ScheduledFrame; + private readonly refreshes = new TrailingRefreshCoordinator<"browser-resume">(); + private scheduledRefresh: ScheduledFrame | undefined; + private connected = false; + + constructor(private readonly callbacks: BrowserResumeCallbacks, options: BrowserResumeControllerOptions = {}) { + this.windowTarget = options.windowTarget ?? browserWindowTarget(); + this.documentTarget = options.documentTarget ?? browserDocumentTarget(); + this.isDocumentVisible = options.isDocumentVisible ?? documentIsVisible; + this.scheduleFrame = options.scheduleFrame ?? scheduleBrowserFrame; + } + + connect(): void { + if (this.connected) return; + this.connected = true; + this.windowTarget?.addEventListener("focus", this.onFocus); + this.documentTarget?.addEventListener("visibilitychange", this.onVisibilityChange); + } + + disconnect(): void { + if (!this.connected) return; + this.connected = false; + this.windowTarget?.removeEventListener("focus", this.onFocus); + this.documentTarget?.removeEventListener("visibilitychange", this.onVisibilityChange); + this.scheduledRefresh?.cancel(); + this.scheduledRefresh = undefined; + } + + private readonly onFocus: EventListener = () => { + this.handleResumeSignal(); + }; + + private readonly onVisibilityChange: EventListener = () => { + if (this.isDocumentVisible()) this.handleResumeSignal(); + }; + + private handleResumeSignal(): void { + this.callbacks.onResumeSignal(); + if (this.scheduledRefresh !== undefined) return; + this.scheduledRefresh = this.scheduleFrame(() => { + this.scheduledRefresh = undefined; + if (!this.connected) return; + void this.refreshes.request("browser-resume", async () => { + if (this.connected) await this.callbacks.refreshAfterResume(); + }).catch((error: unknown) => { this.callbacks.onRefreshError(error); }); + }); + } +} + +function browserWindowTarget(): BrowserEventTarget | undefined { + return typeof window === "undefined" ? undefined : window; +} + +function browserDocumentTarget(): BrowserEventTarget | undefined { + return typeof document === "undefined" ? undefined : document; +} + +function documentIsVisible(): boolean { + return typeof document === "undefined" || document.visibilityState === "visible"; +} + +function scheduleBrowserFrame(callback: () => void): ScheduledFrame { + if (typeof window !== "undefined" && typeof window.requestAnimationFrame === "function") { + const frame = window.requestAnimationFrame(() => { callback(); }); + return { cancel: () => { window.cancelAnimationFrame(frame); } }; + } + const timer = globalThis.setTimeout(callback, 0); + return { cancel: () => { globalThis.clearTimeout(timer); } }; +} diff --git a/src/client/src/components/ChatView.test.ts b/src/client/src/components/ChatView.test.ts index 3aa8484..df3b167 100644 --- a/src/client/src/components/ChatView.test.ts +++ b/src/client/src/components/ChatView.test.ts @@ -1,5 +1,7 @@ +import type { TemplateResult } from "lit"; import { describe, expect, it } from "vitest"; -import { chatMessageMetadataLabel, chatQueuedMessageSections } from "./ChatView"; +import type { ChatLine } from "./shared"; +import { ChatView, chatMessageMetadataLabel, chatQueuedMessageSections } from "./ChatView"; describe("chatQueuedMessageSections", () => { it("labels client-side pending-start sends separately from server queued messages", () => { @@ -35,3 +37,187 @@ describe("chatMessageMetadataLabel", () => { })).toBe(`${formattedTimestamp} ยท provider/model`); }); }); + +describe("ChatView technical-event groups", () => { + const messages: ChatLine[] = [ + { role: "assistant", parts: [{ type: "toolCall", toolName: "read", summary: "inspect a file" }] }, + { role: "tool", parts: [{ type: "toolExecution", toolName: "read", summary: "inspect a file", status: "success", resultText: "large result" }] }, + ]; + + it("defers a closed body while retaining native disclosure and group scroll anchors", () => { + const view = new ChatView(); + view.sessionId = "session-1"; + const bodyCalls = observeGroupBodyRenders(view); + + const closed = renderMessageGroup(view, messages, 40, 41, false); + + expect(bodyCalls).toEqual([]); + expect(templateStaticMarkup(closed)).toContain(""); + expect(templateStaticMarkup(closed)).toContain('aria-hidden="true"'); + expect(templateValuesAfterMarker(closed, "?open=")).toEqual([false]); + expect(templateValuesAfterMarker(closed, "data-scroll-anchor-id=")).toEqual(["g:40"]); + expect(templateValuesAfterMarker(closed, "data-marker-id=")).toEqual(["g:41"]); + }); + + // Direct handler extraction keeps this node-environment test focused on the + // native details toggle wiring without introducing a component-wide DOM shim. + it("renders an opened body with event anchors and removes it when closed again", () => { + const view = new ChatView(); + view.sessionId = "session-1"; + const bodyCalls = observeGroupBodyRenders(view); + const initiallyClosed = renderMessageGroup(view, messages, 40, 41, false); + + dispatchDetailsToggle(templateEventHandler(initiallyClosed, "@toggle="), true); + const opened = renderMessageGroup(view, messages, 40, 41, false); + + expect(bodyCalls).toEqual([{ messages, startIndex: 40 }]); + expect(templateValuesAfterMarker(opened, "?open=")).toEqual([true]); + expect(templateValuesAfterMarker(opened, "data-scroll-anchor-id=")).toEqual(["g:40", "e:40", "e:41"]); + + bodyCalls.length = 0; + dispatchDetailsToggle(templateEventHandler(opened, "@toggle="), false); + const closedAgain = renderMessageGroup(view, messages, 40, 41, false); + + expect(bodyCalls).toEqual([]); + expect(templateValuesAfterMarker(closedAgain, "?open=")).toEqual([false]); + expect(templateValuesAfterMarker(closedAgain, "data-scroll-anchor-id=")).toEqual(["g:40"]); + }); + + it("renders a live tail body by default", () => { + const view = new ChatView(); + view.sessionId = "session-1"; + const bodyCalls = observeGroupBodyRenders(view); + + const live = renderMessageGroup(view, messages, 40, 41, true); + + expect(bodyCalls).toEqual([{ messages, startIndex: 40 }]); + expect(templateValuesAfterMarker(live, "?open=")).toEqual([true]); + expect(templateValues(live)).toContain("msg event-group live"); + expect(templateValues(live)).toContain("live events"); + }); +}); + +interface GroupBodyRenderCall { + messages: ChatLine[]; + startIndex: number; +} + +type RenderMessageGroup = (this: ChatView, messages: ChatLine[], startIndex: number, endIndex: number, defaultOpen: boolean) => TemplateResult; +type RenderMessageGroupBody = (this: ChatView, messages: ChatLine[], startIndex: number) => TemplateResult; +type TemplateEventHandler = (event: Event) => void; + +function renderMessageGroup(view: ChatView, messages: ChatLine[], startIndex: number, endIndex: number, defaultOpen: boolean): TemplateResult { + const method: unknown = Reflect.get(view, "renderMessageGroup"); + if (!isRenderMessageGroup(method)) throw new Error("ChatView.renderMessageGroup is not callable"); + return method.call(view, messages, startIndex, endIndex, defaultOpen); +} + +function observeGroupBodyRenders(view: ChatView): GroupBodyRenderCall[] { + const method: unknown = Reflect.get(view, "renderMessageGroupBody"); + if (!isRenderMessageGroupBody(method)) throw new Error("ChatView.renderMessageGroupBody is not callable"); + const calls: GroupBodyRenderCall[] = []; + const observed: RenderMessageGroupBody = function (messages, startIndex) { + calls.push({ messages, startIndex }); + return method.call(this, messages, startIndex); + }; + if (!Reflect.set(view, "renderMessageGroupBody", observed)) throw new Error("Could not observe ChatView.renderMessageGroupBody"); + return calls; +} + +function isRenderMessageGroup(value: unknown): value is RenderMessageGroup { + return typeof value === "function"; +} + +function isRenderMessageGroupBody(value: unknown): value is RenderMessageGroupBody { + return typeof value === "function"; +} + +function templateEventHandler(template: TemplateResult, marker: string): TemplateEventHandler { + const strings = templateStrings(template); + const values = templateValues(template); + for (let index = 0; index < values.length; index += 1) { + const value = values[index]; + if (strings[index]?.includes(marker) === true && isTemplateEventHandler(value)) return value; + } + throw new Error(`Expected template event handler after ${marker}`); +} + +function isTemplateEventHandler(value: unknown): value is TemplateEventHandler { + return typeof value === "function"; +} + +function dispatchDetailsToggle(handler: TemplateEventHandler, open: boolean): void { + const hadDetailsElement = Reflect.has(globalThis, "HTMLDetailsElement"); + const previousDetailsElement = Reflect.get(globalThis, "HTMLDetailsElement"); + class StubDetailsElement extends EventTarget { + constructor(readonly open: boolean) { + super(); + } + } + Reflect.set(globalThis, "HTMLDetailsElement", StubDetailsElement); + try { + const details = new StubDetailsElement(open); + details.addEventListener("toggle", (event) => { handler(event); }); + details.dispatchEvent(new Event("toggle")); + } finally { + if (hadDetailsElement) Reflect.set(globalThis, "HTMLDetailsElement", previousDetailsElement); + else Reflect.deleteProperty(globalThis, "HTMLDetailsElement"); + } +} + +function templateStaticMarkup(template: TemplateResult): string { + const chunks: string[] = []; + visit(template); + return chunks.join(""); + + function visit(value: unknown): void { + if (Array.isArray(value)) { + for (const item of value) visit(item); + return; + } + if (!isTemplateResult(value)) return; + chunks.push(...templateStrings(value)); + for (const child of templateValues(value)) visit(child); + } +} + +function templateValuesAfterMarker(template: TemplateResult, marker: string): unknown[] { + const matches: unknown[] = []; + visit(template); + return matches; + + function visit(value: unknown): void { + if (Array.isArray(value)) { + for (const item of value) visit(item); + return; + } + if (!isTemplateResult(value)) return; + const strings = templateStrings(value); + const values = templateValues(value); + for (let index = 0; index < values.length; index += 1) { + if (strings[index]?.includes(marker) === true) matches.push(values[index]); + visit(values[index]); + } + } +} + +function templateStrings(template: TemplateResult): readonly string[] { + const strings = Reflect.get(template, "strings"); + if (!isStringArray(strings)) throw new Error("TemplateResult strings were unavailable"); + return strings; +} + +function templateValues(template: TemplateResult): readonly unknown[] { + const values = Reflect.get(template, "values"); + if (!Array.isArray(values)) throw new Error("TemplateResult values were unavailable"); + return values.map((value: unknown) => value); +} + +function isTemplateResult(value: unknown): value is TemplateResult { + return typeof value === "object" && value !== null && isStringArray(Reflect.get(value, "strings")) && Array.isArray(Reflect.get(value, "values")); +} + +function isStringArray(value: unknown): value is string[] { + return Array.isArray(value) && value.every((item: unknown) => typeof item === "string"); +} diff --git a/src/client/src/components/ChatView.ts b/src/client/src/components/ChatView.ts index 0e67e7d..8cc549e 100644 --- a/src/client/src/components/ChatView.ts +++ b/src/client/src/components/ChatView.ts @@ -391,21 +391,27 @@ export class ChatView extends LitElement { ${defaultOpen ? "live events" : "events"} ${summarizeChatGroup(messages)} -
- ${messages.map((message, offset) => { - const toolOnly = this.isToolExecutionOnlyMessage(message); - return html` -
- ${toolOnly ? null : this.renderMessageHeader(message, `${String(startIndex)}:${String(offset)}`)} - ${message.parts.map((part) => this.renderPart(part, message))} -
- `; - })} -
+ ${open ? this.renderMessageGroupBody(messages, startIndex) : null} `; } + private renderMessageGroupBody(messages: ChatLine[], startIndex: number) { + return html` +
+ ${messages.map((message, offset) => { + const toolOnly = this.isToolExecutionOnlyMessage(message); + return html` +
+ ${toolOnly ? null : this.renderMessageHeader(message, `${String(startIndex)}:${String(offset)}`)} + ${message.parts.map((part) => this.renderPart(part, message))} +
+ `; + })} +
+ `; + } + private renderScrollMarker(markerId: string) { return html``; } diff --git a/src/client/src/components/PiWebApp.ts b/src/client/src/components/PiWebApp.ts index 2626f9d..6e8a8d5 100644 --- a/src/client/src/components/PiWebApp.ts +++ b/src/client/src/components/PiWebApp.ts @@ -30,6 +30,7 @@ import { loadExternalPlugins } from "../plugins/external"; import { PluginRegistry, installPluginRuntimeScope, installWorkspacePanelScope } from "../plugins/registry"; import { queryNamespace, readNamespacedString, setNamespacedQueryKey } from "../namespacedQueryArgs"; import { AppShellController } from "../appShell/appShellController"; +import { BrowserResumeController } from "../appShell/browserResumeController"; import { NavigationSectionsController, type NavigationSection } from "../appShell/navigationState"; import { PanelCollapseController, mainViewClass } from "../appShell/panelCollapseController"; import { PanelResizeController, type PanelResizeConstraints, type ResizablePanelSide } from "../appShell/panelResizeController"; @@ -148,6 +149,11 @@ export class PiWebApp extends LitElement { private readonly machineNavigation = new SessionStorageMachineNavigationMemory(); private readonly terminalSelection = new SessionStorageTerminalSelectionMemory(); private readonly appShell = new AppShellController(this); + private readonly browserResume = new BrowserResumeController({ + onResumeSignal: () => { this.handleBrowserResumeSignal(); }, + refreshAfterResume: () => this.refreshAfterBrowserResume(), + onRefreshError: (error) => { console.warn("Failed to refresh after browser resume", error); }, + }); private readonly panelCollapse = new PanelCollapseController(this); private readonly panelResize = new PanelResizeController(this); private readonly navigationSections = new NavigationSectionsController( @@ -191,24 +197,6 @@ export class PiWebApp extends LitElement { this.appShell.repairViewportPosition(); this.retryPendingRemoteRouteRestoreSoon(); }; - private readonly onFocus = () => { - this.appShell.repairViewportPosition(); - void this.sessions.refreshSelectedSession(); - this.schedulePiWebStatusRefresh(); - void this.refreshMachineActivities(); - void this.refreshWorkspaceDeletionRuns(); - this.retryPendingRemoteRouteRestoreSoon(); - }; - private readonly onVisibilityChange = () => { - if (document.visibilityState === "visible") { - this.appShell.repairViewportPosition(); - void this.sessions.refreshSelectedSession(); - this.schedulePiWebStatusRefresh(); - void this.refreshMachineActivities(); - void this.refreshWorkspaceDeletionRuns(); - this.retryPendingRemoteRouteRestoreSoon(); - } - }; private readonly onSystemLightThemeChange = () => { if (this.themePreference.auto) this.applyPreferredTheme(false); }; @@ -232,8 +220,7 @@ export class PiWebApp extends LitElement { super.connectedCallback(); window.addEventListener("popstate", this.onPopState); window.addEventListener("pageshow", this.onPageShow); - window.addEventListener("focus", this.onFocus); - document.addEventListener("visibilitychange", this.onVisibilityChange); + this.browserResume.connect(); window.addEventListener("keydown", this.onKeyDown, GLOBAL_SHORTCUT_LISTENER_OPTIONS); this.systemLightThemeMedia?.addEventListener("change", this.onSystemLightThemeChange); this.applyPreferredTheme(false); @@ -248,8 +235,7 @@ export class PiWebApp extends LitElement { override disconnectedCallback(): void { window.removeEventListener("popstate", this.onPopState); window.removeEventListener("pageshow", this.onPageShow); - window.removeEventListener("focus", this.onFocus); - document.removeEventListener("visibilitychange", this.onVisibilityChange); + this.browserResume.disconnect(); window.removeEventListener("keydown", this.onKeyDown, GLOBAL_SHORTCUT_LISTENER_OPTIONS); this.systemLightThemeMedia?.removeEventListener("change", this.onSystemLightThemeChange); this.keyboard.reset(); @@ -294,6 +280,20 @@ export class PiWebApp extends LitElement { await this.refreshWorkspaceDeletionRuns(); } + private handleBrowserResumeSignal(): void { + this.appShell.repairViewportPosition(); + this.schedulePiWebStatusRefresh(); + this.retryPendingRemoteRouteRestoreSoon(); + } + + private async refreshAfterBrowserResume(): Promise { + await Promise.all([ + this.sessions.refreshSelectedSession(), + this.refreshMachineActivities(), + this.refreshWorkspaceDeletionRuns(), + ]); + } + private schedulePiWebStatusRefresh(delayMs = PI_WEB_STATUS_DEFER_MS): void { this.clearScheduledPiWebStatusRefresh(); this.piWebStatusDeferredTimer = window.setTimeout(() => { diff --git a/src/client/src/controllers/activityController.test.ts b/src/client/src/controllers/activityController.test.ts index 589b784..a51d248 100644 --- a/src/client/src/controllers/activityController.test.ts +++ b/src/client/src/controllers/activityController.test.ts @@ -12,6 +12,13 @@ function snapshot(...workspaces: WorkspaceActivity[]): WorkspaceActivityResponse return { workspaces, generatedAt: "now" }; } +function deferred(): { promise: Promise; resolve: (value: T) => void } { + let resolveDeferred: ((value: T) => void) | undefined; + const promise = new Promise((resolve) => { resolveDeferred = resolve; }); + if (resolveDeferred === undefined) throw new Error("Deferred promise was not initialized"); + return { promise, resolve: resolveDeferred }; +} + describe("ActivityController", () => { it("stores workspace activity under the requested machine", async () => { let state: AppState = { ...initialAppState(), selectedMachine: { id: "remote", name: "Remote", kind: "remote", createdAt: "now", updatedAt: "now" } }; @@ -29,6 +36,42 @@ describe("ActivityController", () => { }); }); + it("shares duplicate requests and runs one trailing refresh requested during the active fetch", async () => { + const firstSnapshot = deferred(); + const trailingSnapshot = deferred(); + const trailingStarted = deferred(); + let calls = 0; + let state: AppState = { ...initialAppState(), selectedMachine: { id: "local", name: "Local", kind: "local", createdAt: "now", updatedAt: "now" } }; + const controller = new ActivityController(() => state, (patch) => { state = { ...state, ...patch }; }, { + api: { + workspaceActivity: () => { + calls += 1; + if (calls === 2) trailingStarted.resolve(undefined); + return calls === 1 ? firstSnapshot.promise : trailingSnapshot.promise; + }, + }, + }); + + const first = controller.refresh("local"); + const duplicate = controller.refresh("local"); + await Promise.resolve(); + + expect(calls).toBe(1); + + const later = controller.refresh("local"); + const laterDuplicate = controller.refresh("local"); + firstSnapshot.resolve(snapshot(activity("/stale"))); + await trailingStarted.promise; + + expect(calls).toBe(2); + + trailingSnapshot.resolve(snapshot(activity("/fresh"))); + await Promise.all([first, duplicate, later, laterDuplicate]); + + expect(calls).toBe(2); + expect(state.workspaceActivities).toEqual({ "/fresh": activity("/fresh") }); + }); + it("applies live activity updates to the owning machine only", () => { let state: AppState = { ...initialAppState(), selectedMachine: { id: "local", name: "Local", kind: "local", createdAt: "now", updatedAt: "now" } }; const controller = new ActivityController(() => state, (patch) => { state = { ...state, ...patch }; }); diff --git a/src/client/src/controllers/activityController.ts b/src/client/src/controllers/activityController.ts index d50b34e..df6532a 100644 --- a/src/client/src/controllers/activityController.ts +++ b/src/client/src/controllers/activityController.ts @@ -1,6 +1,7 @@ import { activityApi as defaultApi, type WorkspaceActivity, type WorkspaceActivityResponse } from "../api"; import { isWorkspaceActivityActive } from "../../../shared/activity"; import { selectedMachineId, type GetState, type SetState } from "./types"; +import { TrailingRefreshCoordinator } from "./trailingRefreshCoordinator"; export interface ActivityControllerDependencies { api?: Pick; @@ -8,13 +9,16 @@ export interface ActivityControllerDependencies { export class ActivityController { private readonly api: Pick; + private readonly refreshes = new TrailingRefreshCoordinator(); constructor(private readonly getState: GetState, private readonly setState: SetState, deps: ActivityControllerDependencies = {}) { this.api = deps.api ?? defaultApi; } - async refresh(machineId = selectedMachineId(this.getState())): Promise { - this.applyMachineActivitySnapshot(machineId, indexWorkspaceActivities(await this.api.workspaceActivity(machineId))); + refresh(machineId = selectedMachineId(this.getState())): Promise { + return this.refreshes.request(machineId, async () => { + this.applyMachineActivitySnapshot(machineId, indexWorkspaceActivities(await this.api.workspaceActivity(machineId))); + }); } applyWorkspaceActivity(activity: WorkspaceActivity, machineId = selectedMachineId(this.getState())): void { diff --git a/src/client/src/controllers/sessionController.refresh.test.ts b/src/client/src/controllers/sessionController.refresh.test.ts new file mode 100644 index 0000000..46dc640 --- /dev/null +++ b/src/client/src/controllers/sessionController.refresh.test.ts @@ -0,0 +1,96 @@ +import { describe, expect, it } from "vitest"; +import { initialAppState } from "../appState"; +import { SessionController } from "./sessionController"; +import { defaultApi, deferred, FakeSocket, oldSession, replacementSession, sessionLookupId, status, workspace, type AppState, type MessagePage, type SessionStatus } from "./sessionController.testSupport"; + +function page(text: string, total: number): MessagePage { + return { messages: [{ role: "assistant", content: text }], start: 0, total }; +} + +describe("SessionController selected-session refresh", () => { + it("shares same-turn requests and runs one trailing refresh requested during the active fetch", async () => { + const firstPage = deferred(); + const firstStatus = deferred(); + const trailingPage = deferred(); + const trailingStatus = deferred(); + const trailingStarted = deferred(); + let messageCalls = 0; + let statusCalls = 0; + let state: AppState = { ...initialAppState(), selectedWorkspace: workspace, selectedSession: oldSession, sessions: [oldSession] }; + const api: typeof defaultApi = { + ...defaultApi, + messages: () => { + messageCalls += 1; + if (messageCalls === 2) trailingStarted.resolve(undefined); + return messageCalls === 1 ? firstPage.promise : trailingPage.promise; + }, + status: () => { + statusCalls += 1; + return statusCalls === 1 ? firstStatus.promise : trailingStatus.promise; + }, + }; + const controller = new SessionController( + () => state, + (patch) => { state = { ...state, ...patch }; }, + () => undefined, + undefined, + { api, socket: new FakeSocket() }, + ); + + const first = controller.refreshSelectedSession(); + const duplicate = controller.refreshSelectedSession(); + await Promise.resolve(); + + expect(messageCalls).toBe(1); + expect(statusCalls).toBe(1); + + const later = controller.refreshSelectedSession(); + const laterDuplicate = controller.refreshSelectedSession(); + firstPage.resolve(page("stale", 1)); + firstStatus.resolve({ ...status(oldSession.id), messageCount: 1 }); + await trailingStarted.promise; + + expect(messageCalls).toBe(2); + expect(statusCalls).toBe(2); + + trailingPage.resolve(page("fresh", 2)); + trailingStatus.resolve({ ...status(oldSession.id), messageCount: 2 }); + await Promise.all([first, duplicate, later, laterDuplicate]); + + expect(messageCalls).toBe(2); + expect(statusCalls).toBe(2); + expect(state.messages).toEqual([{ role: "assistant", parts: [{ type: "text", text: "fresh" }] }]); + expect(state.status?.messageCount).toBe(2); + }); + + it("does not apply an older refresh after the user selects another session", async () => { + const stalePage = deferred(); + const staleStatus = deferred(); + const replacementPage = page("replacement", 1); + let state: AppState = { ...initialAppState(), selectedWorkspace: workspace, selectedSession: oldSession, sessions: [oldSession, replacementSession] }; + const api: typeof defaultApi = { + ...defaultApi, + messages: (session) => sessionLookupId(session) === oldSession.id ? stalePage.promise : Promise.resolve(replacementPage), + status: (session) => sessionLookupId(session) === oldSession.id ? staleStatus.promise : Promise.resolve(status(replacementSession.id)), + thinkingLevels: () => Promise.resolve({ levels: [] }), + }; + const controller = new SessionController( + () => state, + (patch) => { state = { ...state, ...patch }; }, + () => undefined, + undefined, + { api, socket: new FakeSocket() }, + ); + + const staleRefresh = controller.refreshSelectedSession(); + await Promise.resolve(); + await controller.selectSession(replacementSession, { updateUrl: false }); + stalePage.resolve(page("old response", 1)); + staleStatus.resolve({ ...status(oldSession.id), messageCount: 1 }); + await staleRefresh; + + expect(state.selectedSession?.id).toBe(replacementSession.id); + expect(state.messages).toEqual([{ role: "assistant", parts: [{ type: "text", text: "replacement" }] }]); + expect(state.status?.sessionId).toBe(replacementSession.id); + }); +}); diff --git a/src/client/src/controllers/sessionController.ts b/src/client/src/controllers/sessionController.ts index d9ec288..4249e13 100644 --- a/src/client/src/controllers/sessionController.ts +++ b/src/client/src/controllers/sessionController.ts @@ -14,6 +14,7 @@ import { PI_WEB_CAPABILITIES, supportsPiWebCapability } from "../../../shared/ca import type { PromptAttachmentDelivery } from "../../../shared/apiTypes"; import { InMemorySessionSelectionMemory, markSessionArchived, markSessionsArchived, selectPreferredSession, selectionAfterArchivingSession, selectionAfterArchivingSessions, shouldDeselectAfterArchivedCollapse, type SessionSelectionMemory } from "./sessionSelection"; import { selectedMachineId, type GetState, type SetState, type UpdateUrl } from "./types"; +import { TrailingRefreshCoordinator } from "./trailingRefreshCoordinator"; const MESSAGE_PAGE_SIZE = 100; const BULK_FALLBACK_CONCURRENCY = 4; @@ -60,6 +61,12 @@ interface SuppressedCreatedSession { machineId: string; } +interface SelectedSessionRefreshTarget { + session: SessionInfo; + machineId: string; + selectionSeq: number; +} + export class SessionController { private readonly socket: SessionEventSocket; private readonly api: typeof defaultApi; @@ -74,6 +81,7 @@ export class SessionController { private pendingQueuedSendSeq = 0; private readonly pendingSessionStarts = new Map(); private readonly suppressedCreatedSessions = new Map(); + private readonly selectedSessionRefreshes = new TrailingRefreshCoordinator(); constructor( private readonly getState: GetState, @@ -95,6 +103,7 @@ export class SessionController { } dispose() { + this.selectionSeq += 1; this.socket.close(); this.clearPendingUpdates(); } @@ -163,6 +172,7 @@ export class SessionController { isReceivingPartialStream: false, status: session.archived === true ? undefined : this.getState().sessionStatuses[session.id], activity: session.archived === true ? undefined : this.getState().sessionActivities[session.id], + availableThinkingLevels: [], }); try { if (session.archived === true) { @@ -180,11 +190,9 @@ export class SessionController { () => { void this.refreshSelectedSession(session.id); }, selectedMachineId(this.getState()), ); - const [page, status] = await Promise.all([this.api.messages(session, { limit: MESSAGE_PAGE_SIZE }, selectedMachineId(this.getState())), this.api.status(session, selectedMachineId(this.getState()))]); - if (seq !== this.selectionSeq || this.getState().selectedSession?.id !== session.id) return; - const history = this.transcripts.mergeHistory(transcriptKey, page); - this.setState({ ...history, isLoadingEarlierMessages: false, ...this.setStreamCatchup(status.isStreaming ? session.id : undefined), status, activity: this.getState().sessionActivities[session.id], availableThinkingLevels: [] }); - this.applyStatus(status); + const machineId = selectedMachineId(this.getState()); + await this.requestSelectedSessionRefresh({ session, machineId, selectionSeq: seq }); + if (!this.isCurrentRefreshTarget({ session, machineId, selectionSeq: seq })) return; void this.refreshAvailableThinkingLevels(); for (const event of buffered) this.applyEvent(event); this.socket.setHandler((event) => { this.applyEvent(event); }); @@ -725,24 +733,48 @@ export class SessionController { } } - async refreshSelectedSession(sessionId = this.getState().selectedSession?.id): Promise { + refreshSelectedSession(sessionId = this.getState().selectedSession?.id): Promise { const session = this.getState().selectedSession; - if (sessionId === undefined || session?.id !== sessionId || session.archived === true || isClientPendingStartSessionInfo(session)) return; - try { + if (sessionId === undefined || session?.id !== sessionId || session.archived === true || isClientPendingStartSessionInfo(session)) return Promise.resolve(); + const target: SelectedSessionRefreshTarget = { + session, + machineId: selectedMachineId(this.getState()), + selectionSeq: this.selectionSeq, + }; + return this.requestSelectedSessionRefresh(target).catch((error: unknown) => { + if (this.isCurrentRefreshTarget(target)) this.setState({ error: String(error) }); + }); + } + + private requestSelectedSessionRefresh(target: SelectedSessionRefreshTarget): Promise { + const key = machineSessionKey(target.machineId, target.session.id); + return this.selectedSessionRefreshes.request(key, async () => { + if (!this.isCurrentRefreshTarget(target)) return; this.flushPendingUpdates(); - const [page, status] = await Promise.all([this.api.messages(session, { limit: MESSAGE_PAGE_SIZE }, selectedMachineId(this.getState())), this.api.status(session, selectedMachineId(this.getState()))]); - if (this.getState().selectedSession?.id !== sessionId) return; - const history = this.transcripts.mergeHistory(this.sessionCacheKey(sessionId), page); + const [page, status] = await Promise.all([ + this.api.messages(target.session, { limit: MESSAGE_PAGE_SIZE }, target.machineId), + this.api.status(target.session, target.machineId), + ]); + if (!this.isCurrentRefreshTarget(target)) return; + const history = this.transcripts.mergeHistory(key, page); this.setState({ ...history, status, - activity: this.getState().sessionActivities[sessionId], - ...this.setStreamCatchup(status.isStreaming ? sessionId : undefined), + activity: this.getState().sessionActivities[target.session.id], + ...this.setStreamCatchup(status.isStreaming ? target.session.id : undefined), }); this.applyStatus(status); - } catch (error) { - if (this.getState().selectedSession?.id === sessionId) this.setState({ error: String(error) }); - } + }); + } + + private isCurrentRefreshTarget(target: SelectedSessionRefreshTarget): boolean { + const state = this.getState(); + const selected = state.selectedSession; + return target.selectionSeq === this.selectionSeq + && selectedMachineId(state) === target.machineId + && selected?.id === target.session.id + && selected.archived !== true + && !isClientPendingStartSessionInfo(selected); } private applyBulkSessionFailures(action: string, failures: readonly string[]): void { diff --git a/src/client/src/controllers/trailingRefreshCoordinator.ts b/src/client/src/controllers/trailingRefreshCoordinator.ts new file mode 100644 index 0000000..555f889 --- /dev/null +++ b/src/client/src/controllers/trailingRefreshCoordinator.ts @@ -0,0 +1,57 @@ +interface PendingRefresh { + promise: Promise; + latestRefresh: () => Promise; + started: boolean; + trailing: boolean; +} + +/** + * Shares refresh work requested in the same task and collapses requests made + * during an active refresh into one trailing pass, without losing later passes. + */ +export class TrailingRefreshCoordinator { + private readonly pendingByKey = new Map(); + + request(key: Key, refresh: () => Promise): Promise { + const existing = this.pendingByKey.get(key); + if (existing !== undefined) { + existing.latestRefresh = refresh; + if (existing.started) existing.trailing = true; + return existing.promise; + } + + const pending: PendingRefresh = { + promise: Promise.resolve(), + latestRefresh: refresh, + started: false, + trailing: false, + }; + pending.promise = Promise.resolve() + .then(async () => { + let latestError: unknown; + let latestFailed: boolean; + do { + pending.trailing = false; + const runRefresh = pending.latestRefresh; + pending.started = true; + latestFailed = false; + try { + await runRefresh(); + } catch (error) { + latestError = error; + latestFailed = true; + } + } while (this.hasTrailingRequest(pending)); + if (latestFailed) throw latestError; + }) + .finally(() => { + if (this.pendingByKey.get(key) === pending) this.pendingByKey.delete(key); + }); + this.pendingByKey.set(key, pending); + return pending.promise; + } + + private hasTrailingRequest(pending: PendingRefresh): boolean { + return pending.trailing; + } +} diff --git a/src/server/app.compression.test.ts b/src/server/app.compression.test.ts new file mode 100644 index 0000000..f87cb4e --- /dev/null +++ b/src/server/app.compression.test.ts @@ -0,0 +1,90 @@ +import { Readable } from "node:stream"; +import { gunzipSync } from "node:zlib"; +import { describe, expect, it, vi } from "vitest"; +import { appTestContext, fakeRemoteClient, registerAppTestHooks } from "./app.testSupport.js"; + +registerAppTestHooks(); + +describe("browser-facing HTTP compression", () => { + it("negotiates compression for large local-machine API responses", async () => { + const marker = "local transcript content ".repeat(256); + appTestContext.piWebConfig = { + plugins: { fake: { settings: { marker } } }, + }; + + const compressed = await appTestContext.app.inject({ + method: "GET", + url: "/api/machines/local/config", + headers: { "accept-encoding": "gzip" }, + }); + const identity = await appTestContext.app.inject({ + method: "GET", + url: "/api/machines/local/config", + headers: { "accept-encoding": "identity" }, + }); + + expect(compressed.statusCode).toBe(200); + expect(compressed.headers["content-encoding"]).toBe("gzip"); + expect(compressed.headers["content-length"]).toBeUndefined(); + expect(compressed.headers.vary).toContain("accept-encoding"); + expect(gunzipJson(compressed)).toMatchObject({ effectiveConfig: { plugins: { fake: { settings: { marker } } } } }); + + expect(identity.statusCode).toBe(200); + expect(identity.headers["content-encoding"]).toBeUndefined(); + expect(identity.json()).toMatchObject({ effectiveConfig: { plugins: { fake: { settings: { marker } } } } }); + }); + + it("negotiates compression after streaming a remote-machine API response", async () => { + const addResponse = await appTestContext.app.inject({ + method: "POST", + url: "/api/machines", + payload: { name: "Remote", baseUrl: "https://remote.example.test/" }, + }); + const remote = addResponse.json<{ id: string }>(); + const projects = Array.from({ length: 64 }, (_, index) => ({ + id: `p-${String(index)}`, + name: `Remote project ${String(index)}`, + path: `/repos/project-${String(index)}`, + createdAt: "2026-07-11T00:00:00.000Z", + })); + const body = JSON.stringify(projects); + const request = vi.fn(() => Promise.resolve({ + statusCode: 200, + headers: { + "content-type": "application/json", + "content-length": String(Buffer.byteLength(body)), + }, + body: Readable.from([body]), + })); + appTestContext.remoteClient = fakeRemoteClient({ request }); + const url = `/api/machines/${remote.id}/projects`; + + const compressed = await appTestContext.app.inject({ + method: "GET", + url, + headers: { "accept-encoding": "gzip" }, + }); + const identity = await appTestContext.app.inject({ + method: "GET", + url, + headers: { "accept-encoding": "identity" }, + }); + + expect(compressed.statusCode).toBe(200); + expect(compressed.headers["content-encoding"]).toBe("gzip"); + expect(compressed.headers["content-length"]).toBeUndefined(); + expect(compressed.headers.vary).toContain("accept-encoding"); + expect(gunzipJson(compressed)).toEqual(projects); + + expect(identity.statusCode).toBe(200); + expect(identity.headers["content-encoding"]).toBeUndefined(); + expect(identity.json()).toEqual(projects); + expect(request).toHaveBeenNthCalledWith(1, "GET", "/api/projects", undefined); + expect(request).toHaveBeenNthCalledWith(2, "GET", "/api/projects", undefined); + }); +}); + +function gunzipJson(response: { rawPayload: Buffer }): unknown { + const value: unknown = JSON.parse(gunzipSync(response.rawPayload).toString("utf8")); + return value; +} diff --git a/src/server/app.ts b/src/server/app.ts index aeb4913..8d28f45 100644 --- a/src/server/app.ts +++ b/src/server/app.ts @@ -2,6 +2,7 @@ import { existsSync } from "node:fs"; import { dirname, join } from "node:path"; import { fileURLToPath } from "node:url"; import Fastify, { type FastifyInstance, type FastifyServerOptions } from "fastify"; +import fastifyCompress from "@fastify/compress"; import fastifyStatic from "@fastify/static"; import fastifyWebsocket from "@fastify/websocket"; import { ProjectStore } from "./storage/projectStore.js"; @@ -120,6 +121,13 @@ function registerLocalFileSuggestionRoutes(app: FastifyInstance, projects: Proje export async function buildApp(deps: AppDependencies = {}): Promise { const app = Fastify({ logger: deps.logger ?? true, ...(deps.bodyLimit === undefined ? {} : { bodyLimit: deps.bodyLimit }) }); + // Vite proxies development API requests here, while production and machine-scoped + // API requests already terminate here, so this is the shared browser HTTP edge. + await app.register(fastifyCompress, { + globalCompression: true, + globalDecompression: false, + threshold: 1024, + }); await app.register(fastifyWebsocket); const projects = deps.projects ?? new ProjectService(new ProjectStore()); diff --git a/src/server/browserMessageProjection.test.ts b/src/server/browserMessageProjection.test.ts new file mode 100644 index 0000000..d41d546 --- /dev/null +++ b/src/server/browserMessageProjection.test.ts @@ -0,0 +1,64 @@ +import { describe, expect, it } from "vitest"; +import { normalizeMessage } from "../client/src/chatMessages.js"; +import type { MessagePage } from "../shared/apiTypes.js"; +import { projectBrowserMessage, projectBrowserMessageResponse, projectBrowserSessionEvent } from "./browserMessageProjection.js"; + +function signedAssistantMessage() { + return { + role: "assistant", + content: [ + { type: "thinking", thinking: "private chain", thinkingSignature: "opaque-provider-payload", redacted: true }, + { type: "text", text: "visible answer", textSignature: "text-metadata" }, + { type: "toolCall", name: "read", arguments: { thinkingSignature: "ordinary nested argument" }, thoughtSignature: "tool-metadata" }, + ], + model: "model-1", + }; +} + +describe("browser message projection", () => { + it("omits only thinking-block signatures without mutating runtime messages", () => { + const message = signedAssistantMessage(); + + const projected = projectBrowserMessage(message); + + expect(projected).toEqual({ + role: "assistant", + content: [ + { type: "thinking", thinking: "private chain", redacted: true }, + { type: "text", text: "visible answer", textSignature: "text-metadata" }, + { type: "toolCall", name: "read", arguments: { thinkingSignature: "ordinary nested argument" }, thoughtSignature: "tool-metadata" }, + ], + model: "model-1", + }); + expect(message.content[0]).toEqual({ type: "thinking", thinking: "private chain", thinkingSignature: "opaque-provider-payload", redacted: true }); + expect(normalizeMessage(projected)).toEqual(normalizeMessage(message)); + }); + + it("projects both paged and legacy array history responses", () => { + const message = signedAssistantMessage(); + const page: MessagePage = { messages: [message], start: 4, total: 5 }; + + expect(projectBrowserMessageResponse(page)).toEqual({ + messages: [{ ...message, content: [{ type: "thinking", thinking: "private chain", redacted: true }, ...message.content.slice(1)] }], + start: 4, + total: 5, + }); + expect(projectBrowserMessageResponse([message])).toEqual([ + { ...message, content: [{ type: "thinking", thinking: "private chain", redacted: true }, ...message.content.slice(1)] }, + ]); + expect(page.messages[0]).toBe(message); + }); + + it("projects final-message events but leaves other event shapes untouched", () => { + const message = signedAssistantMessage(); + const finalEvent = { type: "message.end" as const, message }; + const appendEvent = { type: "message.append" as const, message }; + + expect(projectBrowserSessionEvent(finalEvent)).toEqual({ + type: "message.end", + message: { ...message, content: [{ type: "thinking", thinking: "private chain", redacted: true }, ...message.content.slice(1)] }, + }); + expect(projectBrowserSessionEvent(appendEvent)).toBe(appendEvent); + expect(finalEvent.message).toBe(message); + }); +}); diff --git a/src/server/browserMessageProjection.ts b/src/server/browserMessageProjection.ts new file mode 100644 index 0000000..54f110c --- /dev/null +++ b/src/server/browserMessageProjection.ts @@ -0,0 +1,59 @@ +import type { MessagePage, SessionUiEvent } from "../shared/apiTypes.js"; + +/** + * Remove provider-only thinking data at the browser transport boundary. The + * runtime message remains unchanged because only affected messages and content + * blocks are copied. + */ +export function projectBrowserMessage(message: unknown): unknown { + if (!isRecord(message)) return message; + const originalContent = message["content"]; + if (!isUnknownArray(originalContent)) return message; + + const content = mapChanged(originalContent, (part) => { + if (!isRecord(part) || part["type"] !== "thinking" || !Object.hasOwn(part, "thinkingSignature")) return part; + const projected = { ...part }; + delete projected["thinkingSignature"]; + return projected; + }); + + return content === originalContent ? message : { ...message, content }; +} + +export function projectBrowserMessageResponse(response: unknown[] | MessagePage): unknown[] | MessagePage { + if (Array.isArray(response)) return mapChanged(response, projectBrowserMessage); + const messages = mapChanged(response.messages, projectBrowserMessage); + return messages === response.messages ? response : { ...response, messages }; +} + +export function projectBrowserSessionEvent(event: SessionUiEvent): SessionUiEvent { + if (event.type !== "message.end" || event.message === undefined) return event; + const message = projectBrowserMessage(event.message); + return message === event.message ? event : { ...event, message }; +} + +function mapChanged(values: T[], project: (value: T) => T): T[] { + let projectedValues: T[] | undefined; + let index = 0; + for (const value of values) { + const projected = project(value); + if (projectedValues === undefined) { + if (projected === value) { + index += 1; + continue; + } + projectedValues = values.slice(0, index); + } + projectedValues.push(projected); + index += 1; + } + return projectedValues ?? values; +} + +function isUnknownArray(value: unknown): value is unknown[] { + return Array.isArray(value); +} + +function isRecord(value: unknown): value is Record { + return typeof value === "object" && value !== null && !Array.isArray(value); +} diff --git a/src/server/machines/machineClient.test.ts b/src/server/machines/machineClient.test.ts index ed15e92..adcf79a 100644 --- a/src/server/machines/machineClient.test.ts +++ b/src/server/machines/machineClient.test.ts @@ -29,6 +29,38 @@ describe("RemoteMachineClient", () => { expect(new Headers(init.headers).get("content-type")).toBe("application/json"); expect(init.body).toBe(JSON.stringify({ cwd: "/repo" })); }); + + it("requests compression for the remote hop even when configured headers use different casing", async () => { + const fetchImpl = vi.fn(() => Promise.resolve(new Response("ok", { status: 200 }))); + const client = new RemoteMachineClient({ + baseUrl: "https://remote.example.test/", + headers: { "Accept-Encoding": "identity" }, + }, fetchImpl); + + await client.request("GET", "/api/projects"); + + const { init } = onlyFetchCall(fetchImpl); + expect(new Headers(init.headers).get("accept-encoding")).toBe("gzip, deflate"); + }); + + it("removes stale representation headers after Fetch decodes a compressed response", async () => { + const fetchImpl = vi.fn(() => Promise.resolve(new Response(JSON.stringify({ ok: true }), { + status: 200, + headers: { + "content-type": "application/json", + "content-encoding": "gzip", + "content-length": "31", + }, + }))); + const client = new RemoteMachineClient({ baseUrl: "https://remote.example.test/" }, fetchImpl); + + const response = await client.requestJson("GET", "/api/projects"); + + expect(response.body).toEqual({ ok: true }); + expect(response.headers["content-type"]).toBe("application/json"); + expect(response.headers["content-encoding"]).toBeUndefined(); + expect(response.headers["content-length"]).toBeUndefined(); + }); }); function fetchInputUrl(input: RequestInfo | URL): string { diff --git a/src/server/machines/machineClient.ts b/src/server/machines/machineClient.ts index 57b2b96..cc743e6 100644 --- a/src/server/machines/machineClient.ts +++ b/src/server/machines/machineClient.ts @@ -28,6 +28,8 @@ export interface MachineClient { export const DEFAULT_REMOTE_REQUEST_TIMEOUT_MS = 30_000; export const DEFAULT_REMOTE_HEALTH_TIMEOUT_MS = 3_000; +const REMOTE_RESPONSE_ACCEPT_ENCODING = "gzip, deflate"; + const BLOCKED_CONFIGURED_HEADER_NAMES = new Set([ "host", "connection", @@ -57,7 +59,7 @@ export class RemoteMachineClient implements MachineClient { const response = await this.fetchResponse(method, path, body, options); return { statusCode: response.status, - headers: headersToRecord(response.headers), + headers: decodedResponseHeaders(response.headers), ...(response.body === null ? {} : { body: readableFromWebResponseBody(response.body) }), }; } @@ -68,7 +70,7 @@ export class RemoteMachineClient implements MachineClient { const parsed: unknown = text === "" ? undefined : JSON.parse(text); return { statusCode: response.status, - headers: headersToRecord(response.headers), + headers: decodedResponseHeaders(response.headers), body: parsed, }; } @@ -100,12 +102,12 @@ export class RemoteMachineClient implements MachineClient { } } - private requestHeaders(body: unknown, options: MachineRequestOptions): HeadersInit { - return { - ...this.remoteHeaders(), - accept: "*/*", - ...(body === undefined ? {} : { "content-type": options.contentType ?? defaultContentTypeForBody(body) }), - }; + private requestHeaders(body: unknown, options: MachineRequestOptions): Headers { + const headers = new Headers(this.remoteHeaders()); + headers.set("accept", "*/*"); + headers.set("accept-encoding", REMOTE_RESPONSE_ACCEPT_ENCODING); + if (body !== undefined) headers.set("content-type", options.contentType ?? defaultContentTypeForBody(body)); + return headers; } private remoteHeaders(): Record { @@ -145,8 +147,16 @@ function filterConfiguredHeaders(headers: Record | undefined): R return Object.fromEntries(Object.entries(headers).filter(([key]) => !BLOCKED_CONFIGURED_HEADER_NAMES.has(key.toLowerCase()))); } -function headersToRecord(headers: Headers): Record { - return Object.fromEntries(headers.entries()); +function decodedResponseHeaders(headers: Headers): Record { + const values: Record = Object.fromEntries(headers.entries()); + const contentEncoding = values["content-encoding"]; + if (contentEncoding !== undefined && contentEncoding !== "identity") { + // Fetch decodes response bodies but retains headers for the encoded wire + // representation. The outer HTTP edge must negotiate and frame the decoded body. + delete values["content-encoding"]; + delete values["content-length"]; + } + return values; } function serializeRequestBody(method: string, body: unknown): NonNullable | undefined { diff --git a/src/server/realtime/sessionEventHub.test.ts b/src/server/realtime/sessionEventHub.test.ts index a3e3182..6397759 100644 --- a/src/server/realtime/sessionEventHub.test.ts +++ b/src/server/realtime/sessionEventHub.test.ts @@ -22,6 +22,22 @@ describe("SessionEventHub", () => { expect(otherSocket.send).not.toHaveBeenCalled(); }); + it("omits thinking signatures from final-message payloads without mutating source events", () => { + const hub = new SessionEventHub(); + const socket = new FakeSocket(); + hub.add("s1", socket); + const thinkingBlock = { type: "thinking", thinking: "private chain", thinkingSignature: "opaque-provider-payload", redacted: true }; + const message = { role: "assistant", content: [thinkingBlock, { type: "text", text: "visible answer" }] }; + + hub.publish("s1", { type: "message.end", message }); + + expect(socket.send).toHaveBeenCalledWith(JSON.stringify({ + type: "message.end", + message: { role: "assistant", content: [{ type: "thinking", thinking: "private chain", redacted: true }, { type: "text", text: "visible answer" }] }, + })); + expect(thinkingBlock.thinkingSignature).toBe("opaque-provider-payload"); + }); + it("removes session sockets on close and skips non-open sockets", () => { const hub = new SessionEventHub(); const closed = new FakeSocket(); diff --git a/src/server/realtime/sessionEventHub.ts b/src/server/realtime/sessionEventHub.ts index 77ca5df..d38ed42 100644 --- a/src/server/realtime/sessionEventHub.ts +++ b/src/server/realtime/sessionEventHub.ts @@ -1,4 +1,5 @@ import type { GlobalSessionEvent, RealtimeEvent, SessionUiEvent } from "../../shared/apiTypes.js"; +import { projectBrowserSessionEvent } from "../browserMessageProjection.js"; export interface RealtimeSocket { readonly OPEN: number; @@ -29,7 +30,7 @@ export class SessionEventHub { } publish(sessionId: string, event: SessionUiEvent): void { - const payload = JSON.stringify(event); + const payload = JSON.stringify(projectBrowserSessionEvent(event)); for (const socket of this.socketsBySession.get(sessionId) ?? []) { if (socket.readyState === socket.OPEN) socket.send(payload); } diff --git a/src/server/sessions/piSessionService.lifecycle.test.ts b/src/server/sessions/piSessionService.lifecycle.test.ts index 12afa09..67f5e6a 100644 --- a/src/server/sessions/piSessionService.lifecycle.test.ts +++ b/src/server/sessions/piSessionService.lifecycle.test.ts @@ -2,8 +2,18 @@ import { mkdtemp, rm, writeFile } from "node:fs/promises"; import { tmpdir } from "node:os"; import { join } from "node:path"; import { describe, expect, it, vi } from "vitest"; -import { PiSessionService, type PiAgentSession } from "./piSessionService.js"; -import { CapturingSessionEventHub, fakeRuntime, fakeSessionManager, runtimeCreator, sessionGateway, sessionRecord, sessionRef, type RuntimeCreator } from "./piSessionService.testSupport.js"; +import { PiSessionService, type PiAgentSession, type PiSessionRuntime } from "./piSessionService.js"; +import { CapturingSessionEventHub, emptyArchiveStore, fakeRuntime, fakeSessionManager, runtimeCreator, sessionGateway, sessionRecord, sessionRef, type RuntimeCreator } from "./piSessionService.testSupport.js"; + +function deferred() { + let resolve!: (value: T | PromiseLike) => void; + let reject!: (reason?: unknown) => void; + const promise = new Promise((promiseResolve, promiseReject) => { + resolve = promiseResolve; + reject = promiseReject; + }); + return { promise, resolve, reject }; +} describe("PiSessionService lifecycle, listing, and reload", () => { it("starts sessions through an injected runtime creator", async () => { @@ -85,6 +95,156 @@ describe("PiSessionService lifecycle, listing, and reload", () => { await service.dispose(); }); + it("shares one runtime when concurrent cold lookups resolve to the same session", async () => { + const sessionId = "single-flight-session"; + const createStarted = deferred(); + const releaseCreate = deferred(); + const winnerUnsubscribe = vi.fn(); + const loserUnsubscribe = vi.fn(); + const winnerSubscribe = vi.fn(() => winnerUnsubscribe); + const loserSubscribe = vi.fn(() => loserUnsubscribe); + const winner = fakeRuntime(sessionId, { + sessionManager: fakeSessionManager("/workspace", { + getSessionId: () => sessionId, + getBranch: () => [{ type: "message", message: { role: "user", content: "shared runtime" } }], + }), + subscribe: winnerSubscribe, + }); + const loser = fakeRuntime(sessionId, { + sessionManager: fakeSessionManager("/workspace", { getSessionId: () => sessionId }), + subscribe: loserSubscribe, + }); + const runtimes = [winner.runtime, loser.runtime]; + let createCalls = 0; + const createAgentRuntime: RuntimeCreator = async () => { + const runtime = runtimes[createCalls]; + createCalls += 1; + createStarted.resolve(); + await releaseCreate.promise; + if (runtime === undefined) throw new Error("unexpected runtime creation"); + return runtime; + }; + const gateway = sessionGateway([sessionRecord(sessionId)]); + const open = vi.spyOn(gateway, "open"); + const service = new PiSessionService(new CapturingSessionEventHub(), { + archiveStore: emptyArchiveStore(), + createAgentRuntime, + sessionManager: gateway, + heartbeatIntervalMs: 60_000, + }); + + const messagesPromise = service.messages(sessionRef(sessionId)); + await createStarted.promise; + const statusPromise = service.status(sessionRef("single-flight")); + await new Promise((resolve) => setImmediate(resolve)); + const callsWhileOpening = createCalls; + releaseCreate.resolve(); + + const [messages, status] = await Promise.all([messagesPromise, statusPromise]); + const activeCount = service.activeCount(); + await service.dispose(); + + expect(callsWhileOpening).toBe(1); + expect(createCalls).toBe(1); + expect(open).toHaveBeenCalledOnce(); + expect(activeCount).toBe(1); + expect(messages).toEqual([{ role: "user", content: "shared runtime" }]); + expect(status).toMatchObject({ sessionId }); + expect(winnerSubscribe).toHaveBeenCalledOnce(); + expect(winnerUnsubscribe).toHaveBeenCalledOnce(); + expect(winner.calls.dispose).toBe(1); + expect(loserSubscribe).not.toHaveBeenCalled(); + expect(loserUnsubscribe).not.toHaveBeenCalled(); + expect(loser.calls.dispose).toBe(0); + }); + + it("clears a failed pending open so the session can be retried", async () => { + const sessionId = "retry-open-session"; + const bindStarted = deferred(); + const bindResult = deferred(); + const openingError = new Error("extension binding failed"); + const failed = fakeRuntime(sessionId, { + bindExtensions: () => { + bindStarted.resolve(); + return bindResult.promise; + }, + }); + const retried = fakeRuntime(sessionId); + const runtimes = [failed.runtime, retried.runtime]; + let createCalls = 0; + const createAgentRuntime: RuntimeCreator = () => { + const runtime = runtimes[createCalls]; + createCalls += 1; + return runtime === undefined + ? Promise.reject(new Error("unexpected runtime creation")) + : Promise.resolve(runtime); + }; + const service = new PiSessionService(new CapturingSessionEventHub(), { + archiveStore: emptyArchiveStore(), + createAgentRuntime, + sessionManager: sessionGateway([sessionRecord(sessionId)]), + heartbeatIntervalMs: 60_000, + }); + + const messagesPromise = service.messages(sessionRef(sessionId)); + await bindStarted.promise; + const statusPromise = service.status(sessionRef("retry-open")); + await new Promise((resolve) => setImmediate(resolve)); + const callsWhileOpening = createCalls; + const failedLookups = Promise.allSettled([messagesPromise, statusPromise]); + bindResult.reject(openingError); + + const outcomes = await failedLookups; + expect(callsWhileOpening).toBe(1); + expect(outcomes).toHaveLength(2); + for (const outcome of outcomes) { + expect(outcome.status).toBe("rejected"); + if (outcome.status === "rejected") expect(outcome.reason).toBe(openingError); + } + expect(service.activeCount()).toBe(0); + expect(failed.calls.abort).toBe(1); + expect(failed.calls.dispose).toBe(1); + + await expect(service.status(sessionRef(sessionId))).resolves.toMatchObject({ sessionId }); + expect(createCalls).toBe(2); + expect(service.activeCount()).toBe(1); + + await service.dispose(); + expect(retried.calls.dispose).toBe(1); + }); + + it("waits for an in-flight open before disposing the service", async () => { + const sessionId = "dispose-opening-session"; + const createStarted = deferred(); + const runtimeResult = deferred(); + const fake = fakeRuntime(sessionId); + const service = new PiSessionService(new CapturingSessionEventHub(), { + archiveStore: emptyArchiveStore(), + createAgentRuntime: () => { + createStarted.resolve(); + return runtimeResult.promise; + }, + sessionManager: sessionGateway([sessionRecord(sessionId)]), + heartbeatIntervalMs: 60_000, + }); + + const statusPromise = service.status(sessionRef(sessionId)); + await createStarted.promise; + let disposeSettled = false; + const disposePromise = service.dispose().then(() => { disposeSettled = true; }); + await new Promise((resolve) => setImmediate(resolve)); + const settledWhileOpening = disposeSettled; + runtimeResult.resolve(fake.runtime); + + await expect(statusPromise).resolves.toMatchObject({ sessionId }); + await disposePromise; + + expect(settledWhileOpening).toBe(false); + expect(service.activeCount()).toBe(0); + expect(fake.calls.abort).toBe(1); + expect(fake.calls.dispose).toBe(1); + }); + it("binds extensions again when the SDK runtime replaces the active session", async () => { const hub = new CapturingSessionEventHub(); const fake = fakeRuntime("session-1"); diff --git a/src/server/sessions/piSessionService.ts b/src/server/sessions/piSessionService.ts index 3affafe..69712aa 100644 --- a/src/server/sessions/piSessionService.ts +++ b/src/server/sessions/piSessionService.ts @@ -31,7 +31,7 @@ import { attachmentsToInlineImages, saveAttachmentsToWorkspace } from "./attachm import { parsePromptAttachments } from "../../shared/promptAttachments.js"; import type { SavedPromptAttachment, SessionBulkArchiveResponse, SessionBulkDeleteArchivedResponse, SessionBulkFailure, SessionBulkMutationRef } from "../../shared/apiTypes.js"; -import { cwdPathsEqual } from "../workingDirectory.js"; +import { canonicalizeStoredCwd, cwdPathsEqual } from "../workingDirectory.js"; import type { WorkspaceActivityService } from "../activity/workspaceActivityService.js"; import { createSpawnSessionToolDefinition, type SpawnSessionInvocation, type SpawnSessionResult } from "./spawnSessionTool.js"; import { createSubsessionToolDefinitions, type SpawnSubsessionInvocation, type SpawnSubsessionResult, type SubsessionCheckResult, type SubsessionReadQuery, type SubsessionReadResult, type SubsessionStatus, type SubsessionSummary, type SubsessionToolDeps } from "./spawnSubsessionTool.js"; @@ -261,6 +261,11 @@ export interface PiSessionRuntime { dispose(): Promise; } +interface PendingSessionOpen { + sessionId: string; + promise: Promise>; +} + interface CreateAgentRuntimeOptions { cwd: string; agentDir: string; @@ -404,6 +409,7 @@ export interface PiSessionServiceDependencies { export class PiSessionService { private readonly active = new Map>(); + private readonly pendingSessionOpens = new Map(); private readonly activities = new Map(); private readonly heartbeat: NodeJS.Timeout; private readonly commandService: SessionCommandService; @@ -533,8 +539,11 @@ export class PiSessionService { async dispose(): Promise { clearInterval(this.heartbeat); this.clearCompactionDrainTimers(); + const pendingOpens = this.pendingSessionOpenPromises(); + if (pendingOpens.length > 0) await Promise.allSettled(pendingOpens); const activeSessions = Array.from(new Set(this.active.values())); this.active.clear(); + this.pendingSessionOpens.clear(); this.activities.clear(); this.compactionPromptQueues.clear(); this.authLossWarnings.clear(); @@ -546,8 +555,11 @@ export class PiSessionService { await Promise.all(activeSessions.map(async (active) => { active.unsubscribe(); this.workspaceActivity?.removeSession(active.runtime.session.sessionId, active.runtime.session.sessionManager.getCwd()); - await active.runtime.session.abort(); - await active.runtime.dispose(); + try { + await active.runtime.session.abort(); + } finally { + await active.runtime.dispose(); + } })); } @@ -1540,6 +1552,8 @@ export class PiSessionService { } private async closeActive(sessionId: string): Promise { + const pendingOpens = this.pendingSessionOpenPromises(sessionId); + if (pendingOpens.length > 0) await Promise.allSettled(pendingOpens); const active = this.active.get(sessionId); if (!active) return; this.active.delete(sessionId); @@ -1573,13 +1587,49 @@ export class PiSessionService { if (active !== undefined) return active; const archived = await this.getArchived(ref); - if (archived?.archivePath !== undefined) return this.create(this.sessionManager.open(archived.archivePath), archived.cwd); + if (archived?.archivePath !== undefined) { + const { archivePath } = archived; + return this.openExistingSession( + archived.sessionId, + archived.cwd, + () => this.sessionManager.open(archivePath), + ); + } const match = isPiSessionRef(ref) ? (await this.sessionManager.list(ref.cwd)).find((s) => s.id === ref.id || s.id.startsWith(ref.id)) : (await this.sessionManager.listAll?.() ?? []).find((s) => s.id === ref || s.id.startsWith(ref)); if (!match) throw new Error("Session not found"); - return this.create(this.sessionManager.open(match.path), match.cwd); + return this.openExistingSession(match.id, match.cwd, () => this.sessionManager.open(match.path)); + } + + private openExistingSession( + sessionId: string, + cwd: string, + openSessionManager: () => PiSessionManager, + ): Promise> { + const active = this.activeForLookup({ id: sessionId, cwd }); + if (active !== undefined) return Promise.resolve(active); + + const key = JSON.stringify([canonicalizeStoredCwd(cwd), sessionId]); + const existing = this.pendingSessionOpens.get(key); + if (existing !== undefined) return existing.promise; + + const pending: PendingSessionOpen = { + sessionId, + promise: this.create(openSessionManager(), cwd), + }; + pending.promise = pending.promise.finally(() => { + if (this.pendingSessionOpens.get(key) === pending) this.pendingSessionOpens.delete(key); + }); + this.pendingSessionOpens.set(key, pending); + return pending.promise; + } + + private pendingSessionOpenPromises(sessionId?: string): Promise>[] { + return [...this.pendingSessionOpens.values()] + .filter((pending) => sessionId === undefined || pending.sessionId === sessionId) + .map((pending) => pending.promise); } private async getArchived(ref: PiSessionLookup): Promise { @@ -1613,18 +1663,40 @@ export class PiSessionService { delegationToolsEnabled, ...(options.initialModel === undefined ? {} : { initialModel: options.initialModel }), }); - await this.bindSessionExtensions(runtime.session); const active: ActiveSession = { runtime, unsubscribe: noop }; - this.bindRuntime(active); - runtime.setRebindSession(async (session) => { - await this.bindSessionExtensions(session); + try { + await this.bindSessionExtensions(runtime.session); this.bindRuntime(active); - await this.recoverSubsessionTrackingForOpenedSession(session); - }); - this.active.set(runtime.session.sessionId, active); - await this.recoverSubsessionTrackingForOpenedSession(runtime.session); - this.publishStatus(runtime.session); - return active; + runtime.setRebindSession(async (session) => { + await this.bindSessionExtensions(session); + this.bindRuntime(active); + await this.recoverSubsessionTrackingForOpenedSession(session); + }); + this.active.set(runtime.session.sessionId, active); + await this.recoverSubsessionTrackingForOpenedSession(runtime.session); + this.publishStatus(runtime.session); + return active; + } catch (error: unknown) { + active.unsubscribe(); + let removedActive = false; + for (const [sessionId, candidate] of this.active.entries()) { + if (candidate !== active) continue; + this.active.delete(sessionId); + this.activities.delete(sessionId); + this.clearAuthLossWarningsForSession(sessionId); + this.clearCompactionPromptQueue(sessionId); + removedActive = true; + } + if (removedActive) { + this.workspaceActivity?.removeSession(runtime.session.sessionId, runtime.session.sessionManager.getCwd()); + } + try { + await runtime.session.abort(); + } finally { + await runtime.dispose(); + } + throw error; + } } private async bindSessionExtensions(session: PiAgentSession): Promise { diff --git a/src/server/sessions/sessionRoutes.test.ts b/src/server/sessions/sessionRoutes.test.ts index abef639..8142442 100644 --- a/src/server/sessions/sessionRoutes.test.ts +++ b/src/server/sessions/sessionRoutes.test.ts @@ -2,7 +2,7 @@ import { resolve } from "node:path"; import Fastify, { type FastifyInstance } from "fastify"; import fastifyWebsocket from "@fastify/websocket"; import { afterEach, beforeEach, describe, expect, it } from "vitest"; -import type { SessionBulkArchiveResponse, SessionBulkDeleteArchivedResponse, SessionBulkMutationRef, SessionCleanupExecuteResponse, SessionCleanupPreviewResponse } from "../../shared/apiTypes.js"; +import type { MessagePage, SessionBulkArchiveResponse, SessionBulkDeleteArchivedResponse, SessionBulkMutationRef, SessionCleanupExecuteResponse, SessionCleanupPreviewResponse } from "../../shared/apiTypes.js"; import { SessionEventHub } from "../realtime/sessionEventHub.js"; import { PiSessionService, type PiSessionManagerGateway, type PiSessionRef } from "./piSessionService.js"; import { registerSessionRoutes } from "./sessionRoutes.js"; @@ -55,6 +55,32 @@ describe("session routes", () => { } }); + it("omits thinking signatures from browser history without mutating service messages", async () => { + const routeApp = Fastify({ logger: false }); + await routeApp.register(fastifyWebsocket); + const eventHub = new SessionEventHub(); + const routeService = new CapturingRouteSessionService(eventHub); + const thinkingBlock = { type: "thinking", thinking: "private chain", thinkingSignature: "opaque-provider-payload", redacted: true }; + const message = { role: "assistant", content: [thinkingBlock, { type: "text", text: "visible answer" }] }; + routeService.messagesResponse = { messages: [message], start: 0, total: 1 }; + registerSessionRoutes(routeApp, routeService, eventHub); + + try { + const response = await routeApp.inject({ method: "GET", url: "/sessions/session-1/messages?limit=20" }); + + expect(response.statusCode).toBe(200); + expect(response.json()).toEqual({ + messages: [{ role: "assistant", content: [{ type: "thinking", thinking: "private chain", redacted: true }, { type: "text", text: "visible answer" }] }], + start: 0, + total: 1, + }); + expect(thinkingBlock.thinkingSignature).toBe("opaque-provider-payload"); + } finally { + await routeService.dispose(); + await routeApp.close(); + } + }); + it("forwards prompt attachments and supports the save-attachments route", async () => { const routeApp = Fastify({ logger: false }); await routeApp.register(fastifyWebsocket); @@ -226,6 +252,7 @@ describe("session routes", () => { class CapturingRouteSessionService extends PiSessionService { readonly calls: unknown[] = []; readonly reloadCalls: (string | PiSessionRef)[] = []; + messagesResponse: unknown[] | MessagePage = []; readonly cleanupPreviewCalls: NormalizedSessionCleanupRequest[] = []; readonly cleanupCalls: NormalizedSessionCleanupRequest[] = []; readonly bulkArchiveCalls: SessionBulkMutationRef[][] = []; @@ -262,6 +289,10 @@ class CapturingRouteSessionService extends PiSessionService { return Promise.resolve(); } + override messages(): Promise { + return Promise.resolve(this.messagesResponse); + } + override status(lookup: string | PiSessionRef) { this.calls.push(lookup); return Promise.resolve({ diff --git a/src/server/sessions/sessionRoutes.ts b/src/server/sessions/sessionRoutes.ts index 3aeef8a..cc57600 100644 --- a/src/server/sessions/sessionRoutes.ts +++ b/src/server/sessions/sessionRoutes.ts @@ -1,5 +1,6 @@ import type { FastifyInstance } from "fastify"; import type { SessionBulkMutationRequest, SessionBulkMutationRef, SessionCleanupRequest } from "../../shared/apiTypes.js"; +import { projectBrowserMessageResponse } from "../browserMessageProjection.js"; import { normalizeRequestCwd } from "../workingDirectory.js"; import type { SessionEventHub } from "../realtime/sessionEventHub.js"; import type { PiSessionRef, PiSessionService } from "./piSessionService.js"; @@ -83,7 +84,8 @@ export function registerSessionRoutes(app: FastifyInstance, sessions: PiSessionS app.get<{ Params: { sessionId: string }; Querystring: MessageQuery }>(`${prefix}/sessions/:sessionId/messages`, async (request, reply) => { try { const page = { ...optionalField("before", optionalNumber(request.query.before)), ...optionalField("limit", optionalNumber(request.query.limit)) }; - return await sessions.messages(sessionLookupFromQuery(request.params.sessionId, request.query), page); + const messages = await sessions.messages(sessionLookupFromQuery(request.params.sessionId, request.query), page); + return projectBrowserMessageResponse(messages); } catch (error) { return reply.code(404).send({ error: errorMessage(error) }); }