diff --git a/package-lock.json b/package-lock.json index 3ff6dab6..0b2a5a17 100644 --- a/package-lock.json +++ b/package-lock.json @@ -8,13 +8,13 @@ "name": "pear-by-agent-relay", "version": "1.0.0", "dependencies": { - "@agent-relay/cloud": "^9.1.6", + "@agent-relay/cloud": "^9.2.0", "@agent-relay/factory": "^0.1.13", - "@agent-relay/fleet": "^9.1.6", - "@agent-relay/harness-driver": "^9.1.6", - "@agent-relay/harnesses": "^9.1.6", - "@agent-relay/integration-prompts": "^9.1.6", - "@agent-relay/sdk": "^9.1.6", + "@agent-relay/fleet": "^9.2.0", + "@agent-relay/harness-driver": "^9.2.0", + "@agent-relay/harnesses": "^9.2.0", + "@agent-relay/integration-prompts": "^9.2.0", + "@agent-relay/sdk": "^9.2.0", "@agentworkforce/deploy": "^4.0.2", "@relayburn/sdk": "^4.0.0", "@relaycast/sdk": "^5.0.7", @@ -24,7 +24,7 @@ "@xterm/addon-web-links": "^0.11.0", "@xterm/addon-webgl": "^0.18.0", "@xterm/xterm": "^5.5.0", - "agent-relay": "^9.1.6", + "agent-relay": "^9.2.0", "agentworkforce": "^4.0.2", "ai-hist": "^0.2.3", "allotment": "^1.0.9", @@ -44,7 +44,7 @@ "pear": "bin/pear.mjs" }, "devDependencies": { - "@agent-relay/evals": "^9.1.6", + "@agent-relay/evals": "^9.2.0", "@playwright/test": "^1.57.0", "@tailwindcss/vite": "^4.0.0", "@testing-library/react": "^16.3.2", @@ -158,9 +158,9 @@ "integrity": "sha512-h6YkyIl0DZSIeVTqgPZGgN+E2I0COcF7XjljVKGzZGLUKzmBKhLROeyiy9efz/YIpApat3xjLZ4Ac20+EsydTA==" }, "node_modules/@agent-relay/broker-darwin-arm64": { - "version": "9.1.6", - "resolved": "https://registry.npmjs.org/@agent-relay/broker-darwin-arm64/-/broker-darwin-arm64-9.1.6.tgz", - "integrity": "sha512-BqiyMpdjM2CAhg4ZdNYPYpxmSNVQZu0Vss/1TPPe7OJ9tQAdyrq0QLWBOgqeS/fuvyVz6NK5z8N3GYa8wxk6LA==", + "version": "9.2.0", + "resolved": "https://registry.npmjs.org/@agent-relay/broker-darwin-arm64/-/broker-darwin-arm64-9.2.0.tgz", + "integrity": "sha512-+9FbIr/u0bcRRO7ZTe1FYf6z5WWVTxkXFCQCEKth4ZetnZIz5m+iHi2ICAX2pXqWX0ZiX+RyVYni9wguN/V0Rg==", "cpu": [ "arm64" ], @@ -171,9 +171,9 @@ ] }, "node_modules/@agent-relay/broker-darwin-x64": { - "version": "9.1.6", - "resolved": "https://registry.npmjs.org/@agent-relay/broker-darwin-x64/-/broker-darwin-x64-9.1.6.tgz", - "integrity": "sha512-D+5lfcUK5IW0D3FbBLpCIsCbnS0IMY3MVQqZOpYEL3Cq8pbf0hWII+xlQZvolFvol/CKHYk43lmoJebPIdeaFQ==", + "version": "9.2.0", + "resolved": "https://registry.npmjs.org/@agent-relay/broker-darwin-x64/-/broker-darwin-x64-9.2.0.tgz", + "integrity": "sha512-2oGoER0Hjz4IJxyaeoQRzEqsV7KnKYlxHbEP/6Hn++E4UF3NzIJr1Xmoq96WsEd3HjJdK+vyhBSUnpkQObhjRg==", "cpu": [ "x64" ], @@ -184,9 +184,9 @@ ] }, "node_modules/@agent-relay/broker-linux-arm64": { - "version": "9.1.6", - "resolved": "https://registry.npmjs.org/@agent-relay/broker-linux-arm64/-/broker-linux-arm64-9.1.6.tgz", - "integrity": "sha512-guKf/+dQfX1YP2v69SnVNgfnoWXPSZ8MgPoRv/4X8WUQPKou/WZk64SMUodH5LYvuj1uqFoYqWYZoxogrAQ45Q==", + "version": "9.2.0", + "resolved": "https://registry.npmjs.org/@agent-relay/broker-linux-arm64/-/broker-linux-arm64-9.2.0.tgz", + "integrity": "sha512-9jdtQd6tG8H9GU35wawobVVz8odVNai2vdjUF3lidq29s5b1mMo2/gwpGXxU6kBIMoS5LxM8N8QOsSTnbZh+7g==", "cpu": [ "arm64" ], @@ -197,9 +197,9 @@ ] }, "node_modules/@agent-relay/broker-linux-x64": { - "version": "9.1.6", - "resolved": "https://registry.npmjs.org/@agent-relay/broker-linux-x64/-/broker-linux-x64-9.1.6.tgz", - "integrity": "sha512-HACxKqgX7thqgmyXpmUt6hZgncyxwQwNW7UJ7JzNp4h6eMcjin28RwK/5CCmefKBXthdAeSYazOPwZsC7BZcEQ==", + "version": "9.2.0", + "resolved": "https://registry.npmjs.org/@agent-relay/broker-linux-x64/-/broker-linux-x64-9.2.0.tgz", + "integrity": "sha512-JAlDPXOobSlHNHgz/1OfxWtrXUzG92Fk8063/RKeWkYrbsXcALImkeTMuHAZUCYwp3UpMrX8y9GdAJHnjrOjbQ==", "cpu": [ "x64" ], @@ -210,9 +210,9 @@ ] }, "node_modules/@agent-relay/broker-win32-x64": { - "version": "9.1.6", - "resolved": "https://registry.npmjs.org/@agent-relay/broker-win32-x64/-/broker-win32-x64-9.1.6.tgz", - "integrity": "sha512-TCC01e2PQB0LDbr5eP412vOVkk+nnPcrYDU8ba8SqDojRI1RhQ2fOthabkJLw2X5FydYQtrgmOyo5q8rjzdFgg==", + "version": "9.2.0", + "resolved": "https://registry.npmjs.org/@agent-relay/broker-win32-x64/-/broker-win32-x64-9.2.0.tgz", + "integrity": "sha512-cPAavCzd+llBoSbsQuz+8M0aHpXGimfK6XQass4ceflbI1VJOuqgs3/YM0C5Opach6pYra4wFQTK+E0ihyVa5A==", "cpu": [ "x64" ], @@ -223,11 +223,11 @@ ] }, "node_modules/@agent-relay/cloud": { - "version": "9.1.6", - "resolved": "https://registry.npmjs.org/@agent-relay/cloud/-/cloud-9.1.6.tgz", - "integrity": "sha512-1F2yHEs3f6t18CmXDVpcmIee6/ANBNMoarzPukC246bL4BfaRKpqTS/DoqtaJYY8mv2yBVt86f/cy88R0qO55Q==", + "version": "9.2.0", + "resolved": "https://registry.npmjs.org/@agent-relay/cloud/-/cloud-9.2.0.tgz", + "integrity": "sha512-dToFV5tDKLxZ8WyBXfamrQXD7RoR4ue/j5R7rJwVg0/sENhDQm0dgxs7kmJSg4m/YPTtVaqabvS4osxp2xXEOw==", "dependencies": { - "@agent-relay/config": "9.1.6", + "@agent-relay/config": "9.2.0", "@aws-sdk/client-s3": "3.1020.0", "ignore": "^7.0.5", "tar": "^7.5.10" @@ -237,23 +237,23 @@ } }, "node_modules/@agent-relay/config": { - "version": "9.1.6", - "resolved": "https://registry.npmjs.org/@agent-relay/config/-/config-9.1.6.tgz", - "integrity": "sha512-ssBZR8DvvKiXO9TYC32DaoW+YGx4Anjz2WbRia+NH7fxRz2TXuAaD29Eb+/nOjVNrfgOtEMzHR9Zqz4dWmSU6w==", + "version": "9.2.0", + "resolved": "https://registry.npmjs.org/@agent-relay/config/-/config-9.2.0.tgz", + "integrity": "sha512-mLZw/cUdYmbF4hp2BOUycQ2MYMu/CpBt5IH7qnNQr5qYYnBg7a+Pon2lQ0c5r7LVoOu7CfsCussrIvIBMD+vGw==", "dependencies": { "zod": "^3.23.8", "zod-to-json-schema": "^3.23.1" } }, "node_modules/@agent-relay/evals": { - "version": "9.1.6", - "resolved": "https://registry.npmjs.org/@agent-relay/evals/-/evals-9.1.6.tgz", - "integrity": "sha512-zc/Am+ck+9tbzglChqrZQlqkD8MEFWUfy0RRJOPRkpoRUTbV/Z9MloRSmjafwp0/jQaY7KUbC3QTXGmQ4jxCoA==", + "version": "9.2.0", + "resolved": "https://registry.npmjs.org/@agent-relay/evals/-/evals-9.2.0.tgz", + "integrity": "sha512-v81TZt149Y6otIhmgH6/IOTro29Z7gn3lNTnhmS499npGqlalqQ8nXyrw1IM9DyQLHs9+Ez8cmRb0bi3VDfuEw==", "dev": true, "license": "Apache-2.0", "dependencies": { - "@agent-relay/harness-driver": "9.1.6", - "@agent-relay/integration-prompts": "9.1.6" + "@agent-relay/harness-driver": "9.2.0", + "@agent-relay/integration-prompts": "9.2.0" } }, "node_modules/@agent-relay/events": { @@ -317,65 +317,65 @@ } }, "node_modules/@agent-relay/fleet": { - "version": "9.1.6", - "resolved": "https://registry.npmjs.org/@agent-relay/fleet/-/fleet-9.1.6.tgz", - "integrity": "sha512-wAEaJuL+hYlk9zvkE89wZ/AtqxzVYGtETpUDvnXAYvLKCFp5wGFDaGnUmeVLdWNRBYdV+RYD7ggpnpxHkw0jsA==", + "version": "9.2.0", + "resolved": "https://registry.npmjs.org/@agent-relay/fleet/-/fleet-9.2.0.tgz", + "integrity": "sha512-6ka9QkSYYULbDjqE7XGbeRXA62U0UIs3oeCPYiEng6qTBuAtvR+oHgYGsYRoVywxznVmmDbXnLOv7uf8dn2SzA==", "license": "Apache-2.0", "dependencies": { - "@agent-relay/harness-driver": "9.1.6", - "@agent-relay/harnesses": "9.1.6", - "@agent-relay/sdk": "9.1.6", + "@agent-relay/harness-driver": "9.2.0", + "@agent-relay/harnesses": "9.2.0", + "@agent-relay/sdk": "9.2.0", "zod": "^3.23.8" } }, "node_modules/@agent-relay/harness-driver": { - "version": "9.1.6", - "resolved": "https://registry.npmjs.org/@agent-relay/harness-driver/-/harness-driver-9.1.6.tgz", - "integrity": "sha512-3lZKbkeFLm9nQ4jxrTOG1vmpMTKoCLhLLh4jrhJNO5PzWw2CJ218gUGB4U+2EWpnB+vVT8CtPzardWlqgUsZ4g==", + "version": "9.2.0", + "resolved": "https://registry.npmjs.org/@agent-relay/harness-driver/-/harness-driver-9.2.0.tgz", + "integrity": "sha512-Ww6R4j/EIEzf3h3H5TgG40w5x01T+ThC8/Ti5UWpyYIXIdIiKSCBKbHqk6IbYkbHhM9WheS44PyxtjvyPD9KRw==", "license": "Apache-2.0", "dependencies": { - "@agent-relay/sdk": "9.1.6", + "@agent-relay/sdk": "9.2.0", "ws": "^8.18.3", "zod": "^3.23.8" }, "optionalDependencies": { - "@agent-relay/broker-darwin-arm64": "9.1.6", - "@agent-relay/broker-darwin-x64": "9.1.6", - "@agent-relay/broker-linux-arm64": "9.1.6", - "@agent-relay/broker-linux-x64": "9.1.6", - "@agent-relay/broker-win32-x64": "9.1.6" + "@agent-relay/broker-darwin-arm64": "9.2.0", + "@agent-relay/broker-darwin-x64": "9.2.0", + "@agent-relay/broker-linux-arm64": "9.2.0", + "@agent-relay/broker-linux-x64": "9.2.0", + "@agent-relay/broker-win32-x64": "9.2.0" } }, "node_modules/@agent-relay/harnesses": { - "version": "9.1.6", - "resolved": "https://registry.npmjs.org/@agent-relay/harnesses/-/harnesses-9.1.6.tgz", - "integrity": "sha512-ZZh1cRoWUZLQinhKk/VlLUJzvy5IM6PTD+z3d4uUG3rRImWLSJF+iCQxiFfTzGsnz4yU03hJYNTJeL8W9QmMTw==", + "version": "9.2.0", + "resolved": "https://registry.npmjs.org/@agent-relay/harnesses/-/harnesses-9.2.0.tgz", + "integrity": "sha512-txZpB4c6cqJZgEeV//I9ln05vpAoAiAYz5oDEYpqqZC500Ef2mZLtW22nnRhLaVpSWu1eh8cFW1CqmnjIa3iQg==", "license": "Apache-2.0", "dependencies": { - "@agent-relay/harness-driver": "9.1.6", - "@agent-relay/sdk": "9.1.6" + "@agent-relay/harness-driver": "9.2.0", + "@agent-relay/sdk": "9.2.0" } }, "node_modules/@agent-relay/integration-prompts": { - "version": "9.1.6", - "resolved": "https://registry.npmjs.org/@agent-relay/integration-prompts/-/integration-prompts-9.1.6.tgz", - "integrity": "sha512-iboLPz3ZBd231DfL93nGPk3eQw5epn0cnhPKxit3epevpBQGNY8h+jBpv1pjTVqwMyQ9vLUK5xATKIhvRtf9sg==", + "version": "9.2.0", + "resolved": "https://registry.npmjs.org/@agent-relay/integration-prompts/-/integration-prompts-9.2.0.tgz", + "integrity": "sha512-GzUxmoGhwwTEaJzmzCPOttqFT1NuEyzT/Tf0y3MNQ69S+VoGCjeFdL54cTSlGoR0DIuED5KqwoWY731GPZsWFQ==", "license": "Apache-2.0" }, "node_modules/@agent-relay/sdk": { - "version": "9.1.6", - "resolved": "https://registry.npmjs.org/@agent-relay/sdk/-/sdk-9.1.6.tgz", - "integrity": "sha512-70VRkM/6KIyMIXR3NETDM9u4R1MWKRyg1V69EJQVBYSLM0+rOr6796ZZhSdlmriv+YiGBvsew5vet0ZT2NmRsw==", + "version": "9.2.0", + "resolved": "https://registry.npmjs.org/@agent-relay/sdk/-/sdk-9.2.0.tgz", + "integrity": "sha512-TvC7RBJj1FpJlKCtVMyVintZc49m6m3mrTtbMTeRIGc/6iU+MmHx25oziO41O3lQHJEryJWm10Bfqyg7AyPD2w==", "dependencies": { "@relaycast/sdk": "^5.0.5" } }, "node_modules/@agent-relay/utils": { - "version": "9.1.6", - "resolved": "https://registry.npmjs.org/@agent-relay/utils/-/utils-9.1.6.tgz", - "integrity": "sha512-zh0G6hdQQf7+vb5cxEbKftu1N5qCLpifnfZkIdbmIj21sGOHXHeY6CQEDVhPo9+NXO8sXr/B+vxNHtbkRVxt2w==", + "version": "9.2.0", + "resolved": "https://registry.npmjs.org/@agent-relay/utils/-/utils-9.2.0.tgz", + "integrity": "sha512-YXh8kBJL2fbzIMJk5wlCOmejc30+FhEVOnVe0zPBS59F6wiFSSDz7o64SZK6w6Crmywu3zHAX1YK7ONALbO2tA==", "dependencies": { - "@agent-relay/config": "9.1.6", + "@agent-relay/config": "9.2.0", "compare-versions": "^6.1.1" } }, @@ -1578,6 +1578,7 @@ }, "node_modules/@clack/prompts/node_modules/is-unicode-supported": { "version": "1.3.0", + "extraneous": true, "inBundle": true, "license": "MIT", "engines": { @@ -4885,6 +4886,15 @@ "@relayfile/sdk": ">=0.6.0 <1" } }, + "node_modules/@relayfile/client": { + "version": "0.10.19", + "resolved": "https://registry.npmjs.org/@relayfile/client/-/client-0.10.19.tgz", + "integrity": "sha512-1hoaPaMc/jB2sMq/a5gbEZ7vGHPAx8WaN23Jr03h8DTowmCeoqwu+cliMWRASo90VmdatJyfjTzQc8csHIXL2g==", + "license": "Apache-2.0", + "engines": { + "node": ">=18" + } + }, "node_modules/@relayfile/core": { "version": "0.10.15", "resolved": "https://registry.npmjs.org/@relayfile/core/-/core-0.10.15.tgz", @@ -7608,19 +7618,20 @@ } }, "node_modules/agent-relay": { - "version": "9.1.6", - "resolved": "https://registry.npmjs.org/agent-relay/-/agent-relay-9.1.6.tgz", - "integrity": "sha512-S5lZS1Rn4RkDsKFso1rf/X5LSjjWbSMd+oMc6/YBPCQ4mM2fH9JVMI0jcggLUCpvuuR8FB3Wghl0c33BLAR4dw==", + "version": "9.2.0", + "resolved": "https://registry.npmjs.org/agent-relay/-/agent-relay-9.2.0.tgz", + "integrity": "sha512-Uj/HrJDkGiAdJb/ZuzzjCIckzGjXt4PrZ4drrlgjVMU/isbUIeH/vp5/TPAeeFawopnOeTWggU3cNPySTckOzA==", "license": "Apache-2.0", "dependencies": { - "@agent-relay/cloud": "9.1.6", - "@agent-relay/config": "9.1.6", - "@agent-relay/fleet": "9.1.6", - "@agent-relay/harness-driver": "9.1.6", - "@agent-relay/sdk": "9.1.6", - "@agent-relay/utils": "9.1.6", + "@agent-relay/cloud": "9.2.0", + "@agent-relay/config": "9.2.0", + "@agent-relay/fleet": "9.2.0", + "@agent-relay/harness-driver": "9.2.0", + "@agent-relay/sdk": "9.2.0", + "@agent-relay/utils": "9.2.0", "@modelcontextprotocol/sdk": "^1.0.0", "@relaycast/sdk": "^5.0.5", + "@relayfile/client": "^0.10.19", "@relayflows/cli": "^1.0.1", "@xterm/headless": "^6.0.0", "commander": "^12.1.0", diff --git a/package.json b/package.json index 9d1ab4a0..5c2ba53d 100644 --- a/package.json +++ b/package.json @@ -35,13 +35,13 @@ "lint": "eslint ." }, "dependencies": { - "@agent-relay/cloud": "^9.1.6", + "@agent-relay/cloud": "^9.2.0", "@agent-relay/factory": "^0.1.13", - "@agent-relay/fleet": "^9.1.6", - "@agent-relay/harness-driver": "^9.1.6", - "@agent-relay/harnesses": "^9.1.6", - "@agent-relay/integration-prompts": "^9.1.6", - "@agent-relay/sdk": "^9.1.6", + "@agent-relay/fleet": "^9.2.0", + "@agent-relay/harness-driver": "^9.2.0", + "@agent-relay/harnesses": "^9.2.0", + "@agent-relay/integration-prompts": "^9.2.0", + "@agent-relay/sdk": "^9.2.0", "@agentworkforce/deploy": "^4.0.2", "@relayburn/sdk": "^4.0.0", "@relaycast/sdk": "^5.0.7", @@ -51,7 +51,7 @@ "@xterm/addon-web-links": "^0.11.0", "@xterm/addon-webgl": "^0.18.0", "@xterm/xterm": "^5.5.0", - "agent-relay": "^9.1.6", + "agent-relay": "^9.2.0", "agentworkforce": "^4.0.2", "ai-hist": "^0.2.3", "allotment": "^1.0.9", @@ -71,7 +71,7 @@ "protobufjs": "8.5.0" }, "devDependencies": { - "@agent-relay/evals": "^9.1.6", + "@agent-relay/evals": "^9.2.0", "@playwright/test": "^1.57.0", "@tailwindcss/vite": "^4.0.0", "@testing-library/react": "^16.3.2", diff --git a/src/main/broker.test.ts b/src/main/broker.test.ts index 147e8545..7198cf30 100644 --- a/src/main/broker.test.ts +++ b/src/main/broker.test.ts @@ -636,6 +636,11 @@ describe('BrokerManager local + cloud coexistence', () => { if (originalPlatformDescriptor) { Object.defineProperty(process, 'platform', originalPlatformDescriptor) } + // Guaranteed cleanup for tests that stub global fetch (mintObserverToken + // specs below) — relying on a `vi.unstubAllGlobals()` call at the end of + // each test body would leak the stub into later tests if an earlier + // assertion in that test throws first. + vi.unstubAllGlobals() }) it('keeps the local session alive when a cloud sandbox attaches', async () => { @@ -832,6 +837,217 @@ describe('BrokerManager local + cloud coexistence', () => { await manager.shutdown() }) + it('mints an observer token via a direct POST to the broker\'s local HTTP API, then caches it', async () => { + const manager = new BrokerManager() + await startLocal(manager) + + const fetchMock = vi.fn(async () => ({ + ok: true, + status: 200, + json: async () => ({ token: 'ot_live_abc', id: 'obs-1' }), + text: async () => '' + })) + vi.stubGlobal('fetch', fetchMock) + + const result = await manager.mintObserverToken(PROJECT_ID) + expect(result).toEqual({ token: 'ot_live_abc', id: 'obs-1' }) + expect(fetchMock).toHaveBeenCalledTimes(1) + expect(fetchMock).toHaveBeenCalledWith( + 'http://127.0.0.1:4242/api/observer-token', + expect.objectContaining({ method: 'POST' }) + ) + + // A second mint for the same project reuses the cached token instead of + // spamming the broker with fresh unrevoked tokens. + const second = await manager.mintObserverToken(PROJECT_ID) + expect(second).toEqual(result) + expect(fetchMock).toHaveBeenCalledTimes(1) + + await manager.shutdown() + }) + + it('coalesces concurrent mints for the same project onto a single in-flight request', async () => { + const manager = new BrokerManager() + await startLocal(manager) + + let resolveFetch: ((value: unknown) => void) | undefined + const fetchMock = vi.fn(() => new Promise((resolve) => { + resolveFetch = resolve + })) + vi.stubGlobal('fetch', fetchMock) + + // Two "clicks" before the first POST has resolved — without coalescing, + // each would race past the (still-empty) cache-miss check and mint a + // separate, unrevoked token. + const first = manager.mintObserverToken(PROJECT_ID) + const second = manager.mintObserverToken(PROJECT_ID) + + // mintObserverTokenUncached awaits session.client.getSession() before + // calling fetch, so give that microtask a turn before asserting. + await vi.waitFor(() => expect(fetchMock).toHaveBeenCalledTimes(1)) + + resolveFetch?.({ + ok: true, + status: 200, + json: async () => ({ token: 'ot_live_abc', id: 'obs-1' }) + }) + + await expect(first).resolves.toEqual({ token: 'ot_live_abc', id: 'obs-1' }) + await expect(second).resolves.toEqual({ token: 'ot_live_abc', id: 'obs-1' }) + expect(fetchMock).toHaveBeenCalledTimes(1) + + // Once settled, a later mint reuses the now-populated cache rather than + // the (already-cleared) in-flight entry. + await expect(manager.mintObserverToken(PROJECT_ID)).resolves.toEqual({ token: 'ot_live_abc', id: 'obs-1' }) + expect(fetchMock).toHaveBeenCalledTimes(1) + + await manager.shutdown() + }) + + it('allows a fresh mint after a coalesced in-flight request fails', async () => { + const manager = new BrokerManager() + await startLocal(manager) + + let rejectFetch: ((reason: unknown) => void) | undefined + const failingFetchMock = vi.fn(() => new Promise((_resolve, reject) => { + rejectFetch = reject + })) + vi.stubGlobal('fetch', failingFetchMock) + + const first = manager.mintObserverToken(PROJECT_ID) + const second = manager.mintObserverToken(PROJECT_ID) + await vi.waitFor(() => expect(failingFetchMock).toHaveBeenCalledTimes(1)) + + rejectFetch?.(new Error('network down')) + await expect(first).rejects.toThrow(/network down/) + await expect(second).rejects.toThrow(/network down/) + + const fetchMock = vi.fn(async () => ({ + ok: true, + status: 200, + json: async () => ({ token: 'ot_live_retry', id: 'obs-retry' }) + })) + vi.stubGlobal('fetch', fetchMock) + + await expect(manager.mintObserverToken(PROJECT_ID)).resolves.toEqual({ token: 'ot_live_retry', id: 'obs-retry' }) + expect(fetchMock).toHaveBeenCalledTimes(1) + + await manager.shutdown() + }) + + it('surfaces a distinct, honest error when the broker predates the observer-token route (404) instead of falling back to the insecure workspace-key link', async () => { + const manager = new BrokerManager() + await startLocal(manager) + + vi.stubGlobal('fetch', vi.fn(async () => ({ + ok: false, + status: 404, + text: async () => 'not found' + }))) + + await expect(manager.mintObserverToken(PROJECT_ID)).rejects.toThrow( + /Observer links require a newer broker version/ + ) + + await manager.shutdown() + }) + + it('throws a readable error when the broker rejects the observer-token request', async () => { + const manager = new BrokerManager() + await startLocal(manager) + + vi.stubGlobal('fetch', vi.fn(async () => ({ + ok: false, + status: 500, + text: async () => 'boom' + }))) + + await expect(manager.mintObserverToken(PROJECT_ID)).rejects.toThrow(/HTTP 500/) + + await manager.shutdown() + }) + + it('mints a fresh observer token after the project session is torn down and restarted', async () => { + const manager = new BrokerManager() + await startLocal(manager) + + vi.stubGlobal('fetch', vi.fn(async () => ({ + ok: true, + status: 200, + json: async () => ({ token: 'ot_live_abc', id: 'obs-1' }) + }))) + await manager.mintObserverToken(PROJECT_ID) + + await manager.shutdown(PROJECT_ID) + await startLocal(manager) + + const fetchMock = vi.fn(async () => ({ + ok: true, + status: 200, + json: async () => ({ token: 'ot_live_def', id: 'obs-2' }) + })) + vi.stubGlobal('fetch', fetchMock) + + const result = await manager.mintObserverToken(PROJECT_ID) + expect(result).toEqual({ token: 'ot_live_def', id: 'obs-2' }) + expect(fetchMock).toHaveBeenCalledTimes(1) + + await manager.shutdown() + }) + + it('does not let a mint left in flight by a dropped session poison the cache for a restarted session', async () => { + const manager = new BrokerManager() + await startLocal(manager) + + let resolveStaleFetch: ((value: unknown) => void) | undefined + const staleFetchMock = vi.fn(() => new Promise((resolve) => { + resolveStaleFetch = resolve + })) + vi.stubGlobal('fetch', staleFetchMock) + + // Mint starts against the first session but its POST never resolves yet. + const stalePromise = manager.mintObserverToken(PROJECT_ID) + await vi.waitFor(() => expect(staleFetchMock).toHaveBeenCalledTimes(1)) + + // The session is torn down mid-mint (e.g. joinWorkspace's shutdown path) + // and a new one starts for the same project id — a workspace switch + // while the old mint is still in flight. + await manager.shutdown(PROJECT_ID) + await startLocal(manager) + + const freshFetchMock = vi.fn(async () => ({ + ok: true, + status: 200, + json: async () => ({ token: 'ot_live_fresh', id: 'obs-fresh' }) + })) + vi.stubGlobal('fetch', freshFetchMock) + + // Without clearing observerTokenMints (and bumping the generation) in + // dropSession, this would reuse the stale in-flight promise from the + // dropped session instead of minting fresh against the new one. + const freshResult = await manager.mintObserverToken(PROJECT_ID) + expect(freshResult).toEqual({ token: 'ot_live_fresh', id: 'obs-fresh' }) + expect(freshFetchMock).toHaveBeenCalledTimes(1) + + // Now let the stale mint (against the already-dropped session) resolve + // late, simulating the race the fix guards against. + resolveStaleFetch?.({ + ok: true, + status: 200, + json: async () => ({ token: 'ot_live_stale', id: 'obs-stale' }) + }) + await expect(stalePromise).resolves.toEqual({ token: 'ot_live_stale', id: 'obs-stale' }) + + // The late resolution must not have overwritten the cache with the + // previous workspace's token — a later mint still returns the fresh, + // current-session token from cache (no additional fetch). + const cachedResult = await manager.mintObserverToken(PROJECT_ID) + expect(cachedResult).toEqual({ token: 'ot_live_fresh', id: 'obs-fresh' }) + expect(freshFetchMock).toHaveBeenCalledTimes(1) + + await manager.shutdown() + }) + it('emits a cloud workspace mismatch event when the sandbox ignores the sent key', async () => { const manager = new BrokerManager() const win = createMockWindow() @@ -2407,6 +2623,29 @@ describe('BrokerManager broker event ingress validation', () => { warnSpy.mockRestore() await manager.shutdown() }) + + it('forwards replay_gap events to the renderer and logs a distinct warning', async () => { + const warnSpy = vi.spyOn(console, 'warn').mockImplementation(() => undefined) + const manager = new BrokerManager() + const win = createMockWindow() + const local = await startLocalWithWindow(manager, win) + const listener = local.onEvent.mock.calls.at(-1)?.[0] + + listener?.({ kind: 'replay_gap', requestedSinceSeq: 10, oldestAvailable: 42, seq: 100 }) + + // Forwarded like any other event — the renderer's reconciliation + // pipeline (agent-store.ts / use-message-reconciliation.ts) is what + // actually reacts to it. + expect( + brokerEventSends(win).some((payload) => (payload as { kind?: string }).kind === 'replay_gap') + ).toBe(true) + expect(warnSpy).toHaveBeenCalledWith( + expect.stringContaining('Replay gap on project') + ) + + warnSpy.mockRestore() + await manager.shutdown() + }) }) describe('classifyBrokerEvent', () => { @@ -2510,11 +2749,25 @@ describe('classifyBrokerEvent', () => { 'delivery_queued', 'channel_subscribed', 'agent_idle', - 'agent_blocked_on_send' + 'agent_blocked_on_send', + 'replay_gap' ]) { expect(KNOWN_BROKER_EVENT_KINDS.has(kind)).toBe(true) } }) + + it('classifies a replay_gap event as valid and preserves its fields', () => { + const result = classifyBrokerEvent({ + kind: 'replay_gap', + requestedSinceSeq: 10, + oldestAvailable: 42, + seq: 100 + }) + expect(result.status).toBe('valid') + if (result.status === 'valid') { + expect(result.event).toMatchObject({ requestedSinceSeq: 10, oldestAvailable: 42, seq: 100 }) + } + }) }) describe('BrokerManager spawnAgent CLI preflight', () => { diff --git a/src/main/broker.ts b/src/main/broker.ts index c6f95540..4cd4dba0 100644 --- a/src/main/broker.ts +++ b/src/main/broker.ts @@ -374,6 +374,10 @@ function isPositiveInteger(value: unknown): value is number { } const BROKER_DETAILS_TIMEOUT_MS = 3_000 +// Guards the direct fetch to the broker's own local /api/observer-token — +// without this, a wedged/unresponsive broker would hang the IPC promise (and +// the renderer's "Minting" UI) indefinitely. +const OBSERVER_TOKEN_REQUEST_TIMEOUT_MS = 5_000 const COMMIT_DRAFT_MAX_DIFF_CHARS = 80_000 const COMMIT_DRAFT_TIMEOUT_MS = 180_000 const DEFAULT_DELIVERY_CONFIRMATION_TIMEOUT_MS = 15_000 @@ -1152,6 +1156,11 @@ interface BrokerClientInternals { } } +export interface ObserverTokenResult { + token: string + id: string +} + type BrokerOperationPriority = 'live' | 'normal' | 'cleanup' class BrokerOperationQueue { @@ -1249,6 +1258,32 @@ export class BrokerManager { private queuedReleases = new Map() private releaseQueue: string[] = [] private activeReleases = 0 + // Minted Relaycast observer tokens (scoped, read-only) keyed by projectId. + // Cached for the life of the session's broker connection so re-opening the + // "Join as observer" UI doesn't spam the workspace with fresh unrevoked + // tokens; cleared in dropSession() whenever that project's session goes + // away (including the join-a-different-workspace shutdown+restart path). + private observerTokens = new Map() + // In-flight mint promises keyed by projectId, so a double-click (or two + // BrokerDetailsPage instances) racing the cache-miss check in + // mintObserverToken() before the first POST resolves coalesce onto one + // request instead of each minting a separate, unrevoked token. Same + // "keyed in-flight-promise, clear in finally guarded by identity" shape as + // SpawnCoordinator.coalesce. Cleared in dropSession() alongside + // `observerTokens` — otherwise a session dropped mid-mint (e.g. joinWorkspace + // rejoining a different workspace) would let a restarted session for the + // same projectId reuse the old session's in-flight promise and cache ITS + // result, pointing the new session's observer link at the old workspace. + private observerTokenMints = new Map>() + // Bumped per project every time its session(s) drop (see dropSession()). + // Same "generation" guard idiom as `eventStreamGeneration` elsewhere in + // this file: a mint captures the generation in flight at the time it + // started, and refuses to populate `observerTokens` if the generation has + // since moved on — belt-and-suspenders alongside clearing the in-flight + // map entry, for the case where some other caller is still awaiting the + // stale promise directly (not through `observerTokenMints`) when the drop + // happens. + private observerTokenGenerations = new Map() get cwd(): string | null { return this.sessions.values().next().value?.cwd || null @@ -1694,6 +1729,107 @@ export class BrokerManager { } } + /** + * Mint a scoped, read-only Relaycast observer token (`ot_live_...`) for a + * project's broker, so "Join as observer" links can stop handing out the + * full workspace API key (which grants full read/write/spawn access). + * + * Calls the broker's own local `/api/observer-token` endpoint directly — + * same pattern as `getClientApiKey`/`getClientBaseUrl` reaching into the + * harness-driver client's transport internals elsewhere in this file — since + * Pear pins a published `@agent-relay/harness-driver` version that predates + * a first-class SDK method for this. + * + * Cached per project for the life of the session (see `observerTokens` + * field) so repeatedly opening the observer-link UI doesn't mint a fresh, + * unrevoked token every time. + */ + async mintObserverToken(projectId: string): Promise { + const normalizedProjectId = projectId.trim() + if (!normalizedProjectId) { + throw new Error('Project id is required') + } + + const cached = this.observerTokens.get(normalizedProjectId) + if (cached) return cached + + const inFlight = this.observerTokenMints.get(normalizedProjectId) + if (inFlight) return inFlight + + const generation = this.observerTokenGenerations.get(normalizedProjectId) || 0 + const mintPromise = this.mintObserverTokenUncached(normalizedProjectId, generation).finally(() => { + if (this.observerTokenMints.get(normalizedProjectId) === mintPromise) { + this.observerTokenMints.delete(normalizedProjectId) + } + }) + this.observerTokenMints.set(normalizedProjectId, mintPromise) + return mintPromise + } + + private async mintObserverTokenUncached( + normalizedProjectId: string, + generation: number + ): Promise { + const session = this.getSessionForProject(normalizedProjectId) + const baseUrl = getClientBaseUrl(session.client) + if (!baseUrl) { + throw new Error('Agent Relay broker URL is unavailable for this project') + } + const apiKey = getClientApiKey(session.client) + + const metadata = await session.client.getSession().catch(() => undefined) + const body: { workspaceId?: string } = {} + if (metadata?.default_workspace_id) { + body.workspaceId = metadata.default_workspace_id + } + + const headers: Record = { 'Content-Type': 'application/json' } + if (apiKey) headers['X-API-Key'] = apiKey + + let response: Response + try { + response = await fetch(`${baseUrl}/api/observer-token`, { + method: 'POST', + headers, + body: JSON.stringify(body), + signal: AbortSignal.timeout(OBSERVER_TOKEN_REQUEST_TIMEOUT_MS) + }) + } catch (err) { + throw new Error(`Failed to reach broker to mint observer token: ${toErrorMessage(err)}`) + } + + if (!response.ok) { + // 404 specifically means this broker predates the /api/observer-token + // route (e.g. the locked @agent-relay/broker-* dependency hasn't been + // bumped yet) — surface that distinctly rather than a generic HTTP + // error, and never fall back to the old workspace-key-based link here: + // that would silently reintroduce the full-access-key leak this + // endpoint exists to close. + if (response.status === 404) { + throw new Error('Observer links require a newer broker version — restart Pear after upgrading.') + } + const text = await response.text().catch(() => '') + throw new Error(`Broker rejected observer-token request (HTTP ${response.status})${text ? `: ${text}` : ''}`) + } + + const parsed = await response.json().catch(() => undefined) as + | { token?: string; id?: string; workspaceId?: string; workspaceAlias?: string } + | undefined + if (!parsed?.token || !parsed?.id) { + throw new Error('Broker returned an invalid observer-token response') + } + + const result: ObserverTokenResult = { token: parsed.token, id: parsed.id } + // Only cache if this project's session hasn't dropped/restarted since the + // mint started — otherwise this could be a stale, previous-workspace + // mint resolving after dropSession() already bumped the generation, which + // would poison the cache for the new session's callers. + if ((this.observerTokenGenerations.get(normalizedProjectId) || 0) === generation) { + this.observerTokens.set(normalizedProjectId, result) + } + return result + } + private async ensureLocalFleetSidecar(session: BrokerSession): Promise { if (session.cloudSandboxId) return if (session.fleetSidecar && session.fleetSidecarCwd === session.cwd) return @@ -2403,6 +2539,16 @@ export class BrokerManager { this.publishBrokerEvent(sessionKey, projectId, win, forwarded) + if (forwarded.kind === 'replay_gap') { + // The broker's replay buffer no longer has everything between + // requestedSinceSeq and its current seq — some events (possibly a + // `relay_inbound` chat message) were silently dropped. `forwarded` is + // still published above like any other event, so the renderer's + // reconciliation pipeline (agent-store.ts / use-message-reconciliation.ts) + // can react to it and resync canonical history. + console.warn(`[broker] Replay gap on project ${projectId}: requested since seq ${forwarded.requestedSinceSeq}, oldest available ${forwarded.oldestAvailable}, now at seq ${forwarded.seq}`) + } + if (event.kind === 'agent_spawned' && event.name) { this.rememberAgentSession(event.name, sessionKey) if (event.parent) { @@ -4091,6 +4237,17 @@ export class BrokerManager { private dropSession(sessionKey: string, options: { disconnectOnly: boolean }): void { this.inputStreamManager.closeInputStreamsForSession(sessionKey) this.clearBrokerTimeoutCountsForProject(projectIdFromSessionKey(sessionKey)) + // A dropped session may be reconnecting to a different workspace (e.g. + // joinWorkspace's shutdown+restart), so any cached observer token for + // this project would be scoped to the wrong workspace afterward. Also + // drop any in-flight mint for this project and bump its generation, so a + // mint that was in flight against the now-dropped session can't populate + // `observerTokens` for whatever new session starts next (see the + // generation check in mintObserverTokenUncached). + const droppedProjectId = projectIdFromSessionKey(sessionKey) + this.observerTokens.delete(droppedProjectId) + this.observerTokenMints.delete(droppedProjectId) + this.observerTokenGenerations.set(droppedProjectId, (this.observerTokenGenerations.get(droppedProjectId) || 0) + 1) const session = this.sessions.get(sessionKey) if (!session) return diff --git a/src/main/ipc-handlers.ts b/src/main/ipc-handlers.ts index 876ce7f5..5d7a03d8 100644 --- a/src/main/ipc-handlers.ts +++ b/src/main/ipc-handlers.ts @@ -671,6 +671,10 @@ export function registerIpcHandlers(): void { } }) + ipcMain.handle('broker:mint-observer-token', async (_, projectId: string) => { + return brokerManager.mintObserverToken(projectId) + }) + ipcMain.handle('broker:auto-fix-runtime', async ( event, projectId: string, diff --git a/src/preload/index.ts b/src/preload/index.ts index cbf4f52d..6a7f9475 100644 --- a/src/preload/index.ts +++ b/src/preload/index.ts @@ -60,6 +60,7 @@ import type { IntegrationEventTelemetrySnapshot, IntegrationOption, IntegrationsEvent, + ObserverTokenResult, PearAPI, PendingRelayMessage, ProactiveAgentBinding, @@ -227,6 +228,8 @@ const api = { invoke('broker:sync-channels', projectId, channels), joinWorkspace: (projectId: string, cwd: string, name: string, channels: string[] | undefined, workspaceKey: string) => invoke('broker:join-workspace', projectId, cwd, name, channels, workspaceKey), + mintObserverToken: (projectId: string) => + invoke('broker:mint-observer-token', projectId), autoFixRuntime: ( projectId: string, cwd: string, diff --git a/src/renderer/src/components/broker/BrokerDetailsPage.tsx b/src/renderer/src/components/broker/BrokerDetailsPage.tsx index f5e71482..4f6c10e1 100644 --- a/src/renderer/src/components/broker/BrokerDetailsPage.tsx +++ b/src/renderer/src/components/broker/BrokerDetailsPage.tsx @@ -101,9 +101,8 @@ function compactValue(value: string | undefined, fallback = 'n/a'): string { return value?.trim() || fallback } -function observerUrlForWorkspaceKey(workspaceKey: string | undefined): string | undefined { - const key = workspaceKey?.trim() - return key ? `https://agentrelay.com/observer?key=${encodeURIComponent(key)}` : undefined +function observerUrlForToken(token: string): string { + return `https://agentrelay.com/observer?key=${encodeURIComponent(token)}` } function getPrimaryRelaycastWorkspace( @@ -410,19 +409,66 @@ function CompactMetaRow({ ) } -function ObserverLink({ href }: { href: string }): React.ReactNode { +type ObserverLinkState = + | { status: 'idle' } + | { status: 'minting' } + | { status: 'ready'; url: string } + | { status: 'error'; message: string } + +/** + * "Join as observer" now mints a scoped, read-only Relaycast observer token + * (`ot_live_...`) via IPC instead of reusing the full workspace API key, so + * the resulting link can no longer be used to spawn/write to the workspace. + * Minting is a network round-trip (main process -> broker's own + * `/api/observer-token`), so this needs loading/error states rather than a + * URL that's synchronously always available. The main process caches the + * minted token per project, so re-clicking after the first successful mint + * is effectively instant and doesn't spam the workspace with fresh tokens. + */ +function ObserverLinkAction({ projectId }: { projectId: string }): React.ReactNode { + const [state, setState] = useState({ status: 'idle' }) + + const handleClick = async (): Promise => { + if (state.status === 'ready') { + window.open(state.url, '_blank', 'noreferrer') + return + } + setState({ status: 'minting' }) + try { + const result = await pear.broker.mintObserverToken(projectId) + const url = observerUrlForToken(result.token) + setState({ status: 'ready', url }) + window.open(url, '_blank', 'noreferrer') + } catch (err) { + setState({ status: 'error', message: err instanceof Error ? err.message : String(err) }) + } + } + + const minting = state.status === 'minting' + return ( - - - Join - + + + {state.status === 'ready' && } + {state.status === 'error' && ( + + {state.message} + + )} + ) } @@ -431,7 +477,6 @@ function BrokerMetadataSummary({ broker }: { broker: BrokerDetails }): React.Rea const primaryWorkspace = getPrimaryRelaycastWorkspace(broker) const workspaceId = broker.relaycast?.defaultWorkspaceId || primaryWorkspace?.workspaceId const workspaceKey = broker.relaycast?.workspaceKey || broker.session?.workspaceKey - const observerUrl = observerUrlForWorkspaceKey(workspaceKey) const workspaceAlias = primaryWorkspace?.workspaceAlias || undefined const selfAgent = primaryWorkspace ? `${primaryWorkspace.selfName} / ${primaryWorkspace.selfAgentId}` @@ -460,7 +505,16 @@ function BrokerMetadataSummary({ broker }: { broker: BrokerDetails }): React.Rea label="Workspace key" value={compactValue(workspaceKey)} copyValue={workspaceKey} - action={observerUrl ? : undefined} + action={workspaceKey + ? ( + // Keyed on workspaceKey so a rejoin/restart that changes the + // underlying workspace (same projectId) remounts this fresh — + // otherwise React would preserve the 'ready' state across the + // re-render and keep showing a stale minted token/URL from the + // old workspace. + + ) + : undefined} /> { expect(reconciler.schedule).toHaveBeenCalledTimes(1) expect(reconciler.schedule).toHaveBeenCalledWith('human-message-sent') }) + + it('schedules reconciliation after a replay-gap timestamp changes', () => { + const reconciler = { schedule: vi.fn() } + + hooks.scheduleReplayGapReconciliation({ + lastReplayGapAt: 0, + reconciler + }) + hooks.scheduleReplayGapReconciliation({ + lastReplayGapAt: 1_717_000_000_000, + reconciler + }) + + expect(reconciler.schedule).toHaveBeenCalledTimes(1) + expect(reconciler.schedule).toHaveBeenCalledWith('replay-gap') + }) + + it('does not schedule reconciliation when no replay gap has occurred yet', () => { + const reconciler = { schedule: vi.fn() } + + hooks.scheduleReplayGapReconciliation({ lastReplayGapAt: 0, reconciler }) + + expect(reconciler.schedule).not.toHaveBeenCalled() + }) }) describe('createMessageReconciler', () => { @@ -523,6 +547,13 @@ describe('createMessageReconciler', () => { expect(source).toMatch(/\[lastHumanMessageSentAt,\s*reconciler\]/) }) + it('wires replay-gap events into the reconciliation hook', () => { + const source = hooks.useMessageReconciliation.toString() + expect(source).toContain('s.lastReplayGapAt') + expect(source).toMatch(/scheduleReplayGapReconciliation\([\s\S]*lastReplayGapAt[\s\S]*reconciler/) + expect(source).toMatch(/\[lastReplayGapAt,\s*reconciler\]/) + }) + it('wires active chat rooms into periodic reconciliation', () => { const source = hooks.useMessageReconciliation.toString() expect(source).toContain('ACTIVE_ROOM_RECONCILE_POLL_MS') diff --git a/src/renderer/src/hooks/use-message-reconciliation.ts b/src/renderer/src/hooks/use-message-reconciliation.ts index 2249a02c..60ccd7db 100644 --- a/src/renderer/src/hooks/use-message-reconciliation.ts +++ b/src/renderer/src/hooks/use-message-reconciliation.ts @@ -84,6 +84,23 @@ export function scheduleHumanMessageSentReconciliation(input: { } } +// A `replay_gap` event means the broker's dashboard-WS replay buffer couldn't +// cover a reconnect, so some events (possibly a dropped `relay_inbound` chat +// message) never reached the renderer. There's no single channel/DM target on +// the gap event itself, so — like the other reconnect-signal handlers below +// (broker-status, window-focus, etc.) — this just resyncs whatever room is +// currently active via `reconciler.schedule`, which reuses the same debounced +// `reconcileMessages` IPC round trip and single shared timer as every other +// trigger, so repeated gaps in a short window collapse into one resync. +export function scheduleReplayGapReconciliation(input: { + lastReplayGapAt: number + reconciler: Pick +}): void { + if (input.lastReplayGapAt > 0) { + input.reconciler.schedule('replay-gap') + } +} + function normalizeChannelName(value: string | undefined): string | null { const normalized = value?.trim().replace(/^#/, '') return normalized || null @@ -323,6 +340,7 @@ export function useMessageReconciliation(): void { const tabs = useUIStore((s) => s.tabs) const brokerStatus = useAgentStore((s) => s.brokerStatus) const lastHumanMessageSentAt = useAgentStore((s) => s.lastHumanMessageSentAt) + const lastReplayGapAt = useAgentStore((s) => s.lastReplayGapAt) const activeTab = getActiveTab(tabs, activeTabId) const activeRoomKey = activeTab?.kind === 'channel' ? `channel:${activeTab.projectId || activeProjectId || ''}:${normalizeChannelName(activeTab.channelName) || ''}` @@ -376,6 +394,10 @@ export function useMessageReconciliation(): void { scheduleHumanMessageSentReconciliation({ lastHumanMessageSentAt, reconciler }) }, [lastHumanMessageSentAt, reconciler]) + useEffect(() => { + scheduleReplayGapReconciliation({ lastReplayGapAt, reconciler }) + }, [lastReplayGapAt, reconciler]) + useEffect(() => { const scheduleAfterRefresh = (reason: string): void => { refreshEventStream(reason) diff --git a/src/renderer/src/lib/ipc-mock.ts b/src/renderer/src/lib/ipc-mock.ts index 162ac205..dc301db3 100644 --- a/src/renderer/src/lib/ipc-mock.ts +++ b/src/renderer/src/lib/ipc-mock.ts @@ -60,6 +60,7 @@ import type { IntegrationEventTelemetrySnapshot, IntegrationOption, IntegrationsEvent, + ObserverTokenResult, PearAPI, PendingRelayMessage, ProactiveAgentBinding, @@ -888,6 +889,10 @@ export const pearMock: PearAPI = { emitBrokerStatus({ projectId, status: 'connected' }) return true }, + mintObserverToken: async (projectId: string): Promise => ({ + token: `ot_live_mock_${projectId}`, + id: `mock-observer-token-${projectId}` + }), autoFixRuntime: async () => ({ removed: [] }), connectCloud: async () => 'mock-cloud', spawnAgent: async (projectId: string, input: BrokerSpawnAgentInput): Promise => { diff --git a/src/renderer/src/stores/agent-store.test.ts b/src/renderer/src/stores/agent-store.test.ts index 89f4a995..66c2e1d9 100644 --- a/src/renderer/src/stores/agent-store.test.ts +++ b/src/renderer/src/stores/agent-store.test.ts @@ -138,3 +138,51 @@ describe('agent-store broker snapshots', () => { expect(state.activeAgentKey).toBe(getAgentKey('project-1', 'claude-1')) }) }) + +describe('agent-store replay gap handling', () => { + beforeEach(() => { + vi.stubGlobal('localStorage', { + getItem: () => null, + setItem: () => {}, + removeItem: () => {}, + clear: () => {}, + key: () => null, + length: 0 + }) + useAgentStore.getState().clearAll() + }) + + afterEach(() => { + useAgentStore.getState().clearAll() + vi.unstubAllGlobals() + }) + + it('bumps lastReplayGapAt on a replay_gap broker event', () => { + expect(useAgentStore.getState().lastReplayGapAt).toBe(0) + + useAgentStore.getState().handleBrokerEvent({ + kind: 'replay_gap', + requestedSinceSeq: 10, + oldestAvailable: 42, + seq: 100, + projectId: 'project-1' + }) + + expect(useAgentStore.getState().lastReplayGapAt).toBeGreaterThan(0) + }) + + it('leaves agents and messages untouched for a replay_gap event', () => { + const before = useAgentStore.getState() + + useAgentStore.getState().handleBrokerEvent({ + kind: 'replay_gap', + requestedSinceSeq: 1, + oldestAvailable: 2, + seq: 3 + }) + + const after = useAgentStore.getState() + expect(after.agents).toBe(before.agents) + expect(after.messages).toBe(before.messages) + }) +}) diff --git a/src/renderer/src/stores/agent-store.ts b/src/renderer/src/stores/agent-store.ts index 279681c0..e8a95dc4 100644 --- a/src/renderer/src/stores/agent-store.ts +++ b/src/renderer/src/stores/agent-store.ts @@ -725,6 +725,13 @@ interface AgentState { brokerErrors: BrokerErrorEntry[] brokerEvents: BrokerEventRecord[] lastHumanMessageSentAt: number + // Bumped whenever a `replay_gap` event is observed (broker dashboard WS + // replay buffer couldn't cover a reconnect, so some events — possibly a + // `relay_inbound` chat message — may have been silently dropped). Wired + // into use-message-reconciliation.ts the same way lastHumanMessageSentAt + // is, so a gap triggers a debounced resync of the active room's canonical + // history via the existing reconcileMessages pipeline. + lastReplayGapAt: number setActiveAgentKey: (key: string | null) => void markAgentActive: (projectId: string | undefined, name: string) => void @@ -769,6 +776,7 @@ export const useAgentStore = create()(subscribeWithSelector((set, ge brokerErrors: [], brokerEvents: [], lastHumanMessageSentAt: 0, + lastReplayGapAt: 0, setActiveAgentKey: (key) => set({ activeAgentKey: key }), @@ -1262,6 +1270,16 @@ export const useAgentStore = create()(subscribeWithSelector((set, ge return { ...clearPendingDeliveries(a), activity: 'idle', currentState: 'idle' } }) })) + } else if (kind === 'replay_gap') { + // The broker's dashboard-WS replay buffer couldn't cover a reconnect — + // some events (possibly a dropped relay_inbound chat message) between + // requestedSinceSeq and oldestAvailable never reached us. There's no + // single agent/message to patch here, so just bump the timestamp; + // use-message-reconciliation.ts watches it the same way it watches + // lastHumanMessageSentAt, and schedules a debounced resync of the + // active room's canonical history via the existing reconcileMessages + // IPC round trip. + set({ lastReplayGapAt: Date.now() }) } }, @@ -1479,7 +1497,8 @@ export const useAgentStore = create()(subscribeWithSelector((set, ge brokerError: null, brokerErrors: [], brokerEvents: [], - lastHumanMessageSentAt: 0 + lastHumanMessageSentAt: 0, + lastReplayGapAt: 0 }) }, diff --git a/src/shared/schemas/broker-events.ts b/src/shared/schemas/broker-events.ts index 73ea08ff..ef2b5065 100644 --- a/src/shared/schemas/broker-events.ts +++ b/src/shared/schemas/broker-events.ts @@ -336,6 +336,26 @@ const agentPermanentlyDead = z }) .passthrough() +/** + * Sent over the broker's local dashboard WebSocket (the same connection + * `relay_inbound`/chat events flow over) when a reconnect requested replay + * from a `sinceSeq` the broker's in-memory replay buffer no longer has — + * e.g. because a burst of other events (like `worker_stream` chunks) + * evicted it first. Any event between `requestedSinceSeq` and + * `oldestAvailable` was silently dropped; the reconciliation flow (see + * `agent-store.ts` / `use-message-reconciliation.ts`) treats this as a + * signal to refetch canonical channel/DM history rather than trusting the + * live event stream for that gap. + */ +const replayGap = z + .object({ + kind: z.literal('replay_gap'), + requestedSinceSeq: z.number(), + oldestAvailable: z.number(), + seq: z.number() + }) + .passthrough() + /** * Discriminated union of every broker event `kind` the SDK emits and the app * forwards. Membership defines the "known" set — see `classifyBrokerEvent`. @@ -372,7 +392,8 @@ export const BrokerEventSchema = z.discriminatedUnion('kind', [ agentBlockedOnSend, agentRestarting, agentRestarted, - agentPermanentlyDead + agentPermanentlyDead, + replayGap ]) export type ValidatedBrokerEvent = z.infer @@ -410,6 +431,7 @@ export type AgentBlockedOnSendEvent = Extract export type AgentRestartedEvent = Extract export type AgentPermanentlyDeadEvent = Extract +export type ReplayGapEvent = Extract /** * The shape every forwarded broker event satisfies: an object carrying a diff --git a/src/shared/types/ipc.ts b/src/shared/types/ipc.ts index cb082d48..8ee94392 100644 --- a/src/shared/types/ipc.ts +++ b/src/shared/types/ipc.ts @@ -327,6 +327,16 @@ export interface BrokerEventRecord { } } +/** + * A scoped, read-only Relaycast observer token (`ot_live_...`) minted via the + * broker's local `/api/observer-token` endpoint — used to build "Join as + * observer" links without handing out the full workspace API key. + */ +export interface ObserverTokenResult { + token: string + id: string +} + export interface BrokerSpawnAgentInput { name: string cli: string @@ -919,6 +929,7 @@ export interface PearAPI { channels: string[] | undefined, workspaceKey: string ) => Promise + mintObserverToken: (projectId: string) => Promise autoFixRuntime: ( projectId: string, cwd: string,