From 18dcebeb1c501c97f6b338a5acebad29932520a1 Mon Sep 17 00:00:00 2001 From: beasty Date: Thu, 10 Sep 2026 00:12:13 +0200 Subject: [PATCH 1/3] Show live T3 question alerts through read-only pairing --- README.md | 44 +- package.json | 2 +- plugin/property-inspector/index.html | 59 +- .../property-inspector/property-inspector.js | 207 +++++-- plugin/property-inspector/styles.css | 58 +- pnpm-lock.yaml | 6 +- scripts/build.ts | 2 +- scripts/verify-release.ts | 23 +- src/connection-store.ts | 99 ++++ src/controller.ts | 131 ++++- src/index.ts | 4 +- src/render.ts | 87 ++- src/t3-client.ts | 17 +- src/t3-error.ts | 13 + src/t3-live-client.ts | 508 ++++++++++++++++++ src/t3-protocol.ts | 122 +++++ src/types.ts | 29 +- tests/controller.test.ts | 115 +++- tests/property-inspector.test.ts | 134 +++-- tests/render.test.ts | 54 +- tests/status.test.ts | 20 +- tests/t3-live-client.test.ts | 427 +++++++++++++++ 22 files changed, 1962 insertions(+), 199 deletions(-) create mode 100644 src/connection-store.ts create mode 100644 src/t3-error.ts create mode 100644 src/t3-live-client.ts create mode 100644 src/t3-protocol.ts create mode 100644 tests/t3-live-client.test.ts diff --git a/README.md b/README.md index 97338f8..1d6e312 100644 --- a/README.md +++ b/README.md @@ -1,6 +1,6 @@ # T3 Code Status for OpenDeck -T3 Code Status adds one OpenDeck key that shows how many open T3 Code threads are working. `4/7` means that four of seven open threads currently have T3's `Starting` or `Working` state. The ring advances toward the next cache read. Press the key to refresh immediately. +T3 Code Status adds an OpenDeck key that shows how many open T3 Code threads are working. `4/7` means that four of seven open threads currently have T3's `Starting` or `Working` state. Status arrives through a live stream; a full ring means the stream is connected. Press the key to retry an offline connection. This is an unofficial community plugin. It is not part of T3 Code or OpenDeck. @@ -8,11 +8,25 @@ This is an unofficial community plugin. It is not part of T3 Code or OpenDeck. The plugin needs Node.js 20 or newer installed on the host system. OpenDeck starts JavaScript plugins with this system-wide `node` executable. You do not need pnpm for a normal installation. -1. Start T3 Code Alpha at least once, then leave it open so the cache stays current. +1. Start T3 Code and leave its server running. 2. Install `com.beastyrabbit.t3-code-status.streamDeckPlugin` through OpenDeck's plugin manager. 3. Drag `Thread status` from the `T3 Code Status` category onto a free key. +4. In T3 Code, open **Settings → Connections**. Enable **Network access** if the pairing controls are hidden, then create a **Read only** pairing link. +5. Paste the complete link into the key's **Pair an environment** field and choose **Pair / replace authorization**. Use the link within five minutes, without opening it elsewhere first. -The first cache read starts immediately. Further reads run every 60 seconds by default. If T3 Code exits or its cache cannot be read, the key retries every five seconds until it recovers, then returns to the configured interval. +Repeat pairing for each environment you want to monitor (up to 16). Connections are shared by all keys. Direct links and hosted `app.t3.codes` pairing links are accepted. Use HTTPS for remote connections; plain HTTP beyond loopback requires explicitly allowing it on your trusted private network. + +Choose **Key display** in the key's settings: + +- **Threads + questions** is the default. It shows the working/open thread count and adds a large blinking question mark while input, approval, or plan review is pending. +- **Threads only** shows the working/open thread count without question alerts. +- **Questions only** shows a faded question mark when nothing needs a response. When a thread needs input, approval, or plan review, the whole key flashes amber once per second and shows how many threads need you. + +For two separate keys, drag `Thread status` onto two keys and select **Threads only** on one and **Questions only** on the other. Each key keeps its own display mode. Waiting threads, background monitoring, and errors alone do not trigger the question alert. Loading, offline, and connection errors still appear in every display mode. + +Questions appear or clear when the stream delivers an update. There is no HTTP polling interval. Pressing a key does not answer or dismiss a question. + +Transient connection failures retry with backoff up to 30 seconds. If any paired environment is disconnected, the key shows the connection state instead of presenting incomplete totals as live. Expired or revoked authorization shows **LINK**: create a fresh read-only pairing link and paste it into the same settings panel. Re-pairing the same environment preserves your keys and replaces its credential. Current T3 bearer sessions expire after about 30 days; automatic authorization renewal is not available through this pairing flow. The normal plugin package declares these OpenDeck platforms: @@ -22,9 +36,9 @@ The normal plugin package declares these OpenDeck platforms: | macOS | 10.15 | | Windows | 10 | -Linux is the only platform tested end to end with OpenDeck so far. That test used OpenDeck 2.14.0, T3 Code Alpha 0.0.36, and a physical Stream Deck. The same bundled T3 client and cache reader also connected successfully to live T3 Code Alpha 0.0.36 data on Windows 11 and macOS 26 on Apple silicon. OpenDeck was not installed on either probe machine, so those checks do not cover its plugin process, WebSocket connection, Property Inspector, key rendering, installation, or restart behavior. +Earlier releases were tested with OpenDeck 2.14.0 and a physical Stream Deck on Linux. The new pairing/stream connection is covered by local HTTP/WebSocket integration tests; those earlier hardware checks do not validate this new connection flow on Windows or macOS. -OpenDeck and T3 Code must run as the same desktop user so the plugin can see that user's T3 runtime and cache. For a Flatpak installation of OpenDeck on Linux, Node.js must be installed outside Flatpak and available on the system `PATH`. A Node.js Flatpak is not enough. Flatpak has not been tested yet. +OpenDeck needs network access to each paired T3 server. For a Flatpak installation of OpenDeck on Linux, Node.js must be installed outside Flatpak and available on the system `PATH`. A Node.js Flatpak is not enough. Flatpak has not been tested yet. ## What the key shows @@ -33,11 +47,11 @@ OpenDeck and T3 Code must run as the same desktop user so the plugin can see tha - Red at `0/7`: none of the open threads is working. - Gray at `0/0`: there are no open threads. -Intermediate values move from red through yellow to green. The plugin counts open top-level threads from every environment in T3 Code's local cache. Child agents, archived threads, and settled threads do not count. Snoozed threads normally do not count; they reappear when they request approval or input, or when a fresh failure or completed turn after the snooze needs attention, provided they are otherwise still open. T3's `Starting` and `Working` states count as work. `Monitoring` counts as waiting. +Intermediate values move from red through yellow to green. The plugin counts open threads from every paired environment's shell stream. Archived and settled threads do not count. Snoozed threads normally do not count; they reappear when they request approval or input, or when a fresh failure or completed turn after the snooze needs attention, provided they are otherwise still open. T3's `Starting` and `Working` states count as work. `Monitoring` counts as waiting. OpenDeck also exposes the same changing status as the key's accessible label. -The Property Inspector accepts refresh intervals from 5 to 300 seconds. Its default is 60 seconds. +The former refresh-interval setting is no longer used. Each environment has one subscription regardless of how many keys use it. ## Optional automatic key placement on Linux @@ -98,17 +112,19 @@ gh attestation verify com.beastyrabbit.t3-code-status.streamDeckPlugin \ Run the same `gh attestation verify` command for the setup archive if you use it. GitHub Actions signs the build provenance with a short-lived Sigstore certificate. Protected version tags bind each immutable release to the source commit that the workflow scanned. -## How local cache access works +## Connection and credential storage + +The plugin exchanges the one-time pairing token for a bearer session with exactly `orchestration:read`, obtains short-lived WebSocket tickets, and subscribes to `orchestration.subscribeShell`. It does not read Chromium caches or import T3 account credentials. The read permission is broader than status-only access: it also permits reading files and conversations, though this plugin does not request them. -T3 Code stores a compact shell snapshot for every connected environment in its local Chromium cache. The plugin calls T3 Code's loopback environment endpoint to select the active T3 profile, then reads those snapshots directly. It does not need a T3 account, pairing code, access token, or separate server. +Credentials are stored separately from OpenDeck profiles and the plugin installation in `~/.config/opendeck-t3-code-status/connections.json`. The directory and file use owner-only permissions on Unix; Windows relies on the user's profile ACLs. This is a private credential file, not an encrypted vault. Do not share it. Pairing tokens are discarded after exchange, and credentials are never sent back to the settings panel or included in logs. -The plugin discovers the current `t3code` user-data directory as well as the legacy `T3 Code`, `T3 Code (Alpha)`, `T3 Code (Beta)`, and other `T3 Code ()` directories. If T3 Code uses a custom Electron data directory, set `T3CODE_CACHE_DIR` to the exact `IndexedDB/t3code_app_0.indexeddb.leveldb` directory. +**Forget selected environment** removes its saved credential from this plugin. Revoke the corresponding session in **T3 Settings → Connections** to remove its server authorization too. -The plugin does not interpret message, title, or prompt records. Chromium stores several object stores in the same LevelDB files, so the reader briefly holds bounded file data in memory before it filters out every non-shell record. Large shell snapshots may live in Chromium's matching external blob store; the reader resolves only the blob metadata for the exact shell key. It does not retain, log, or transmit cache contents, and it does not copy thread IDs into its own files or logs. If T3 Code is not running or has not created a cache yet, the key shows `OFF` or `ERR`, and the Property Inspector explains the state. +On reconnect the plugin checks the server's environment identity, gets a fresh ticket, and resumes from the last applied sequence. Servers without the optional completion marker receive a fresh-snapshot subscription instead. Stream state stays in memory; plugin restarts load a new snapshot. Messages and retained thread counts are bounded. -To keep malformed cache data from exhausting the OpenDeck plugin process, the reader limits files, profiles, records, table blocks, decompressed blocks, external blobs, shell keys, serialized snapshots, and estimated parser allocations. A single LevelDB file may not exceed 64 MB, and a shell snapshot or external blob may not exceed 8 MB. The key shows `ERR` when a limit is exceeded. +This uses T3's existing first-party interface, which is not a versioned public SDK. Protocol reference: T3 Code commit `e16b8b059c9f5ff6dfed1addecffb831c6aee043`, following [the maintainer's guidance](https://github.com/pingdotgg/t3code/issues/10929#issuecomment-5601670045). Future T3 changes may require a plugin update. -Pull requests are one known limitation. T3 Code keeps a pull request's open, closed, or merged state in its live renderer but does not include it in the local shell snapshot. This plugin therefore uses only the settlement state available in the cache. A thread whose automatic settlement depends only on a pull request may remain in the denominator for a while. The plugin deliberately does not request GitHub or T3 credentials to close that gap. +The plugin uses a three-day age-based settlement default and does not import per-client sidebar preferences or pull-request state. Counts can differ from a T3 sidebar configured with other settlement rules. ## Build from source @@ -127,7 +143,7 @@ pnpm package - `release/com.beastyrabbit.t3-code-status.streamDeckPlugin` - `release/com.beastyrabbit.t3-code-status-opendeck-setup.tar.gz` -The package check builds each archive twice and fails unless both copies are byte-for-byte identical. It also verifies the manifest, version, required files, archive contents, bundled dependency notices, and the isolated setup path. It rejects markers from the discarded pairing implementation and never touches the real OpenDeck configuration. +The package check builds each archive twice and fails unless both copies are byte-for-byte identical. It also verifies the manifest, version, required files, archive contents, bundled dependency notices, and the isolated setup path. It rejects the old cache reader in the runtime bundle and never touches the real OpenDeck configuration. For a local Linux development update, OpenDeck may remain open: diff --git a/package.json b/package.json index 6a3559b..23d64ff 100644 --- a/package.json +++ b/package.json @@ -36,7 +36,6 @@ "packageManager": "pnpm@11.24.0", "type": "module", "dependencies": { - "snappyjs": "^0.7.0", "ws": "^8.21.3" }, "devDependencies": { @@ -46,6 +45,7 @@ "@types/ws": "^8.18.1", "esbuild": "^0.28.2", "lefthook": "^2.1.12", + "snappyjs": "^0.7.0", "tsx": "^4.23.13", "typescript": "^7.0.2" } diff --git a/plugin/property-inspector/index.html b/plugin/property-inspector/index.html index 52ac8fa..d62241b 100644 --- a/plugin/property-inspector/index.html +++ b/plugin/property-inspector/index.html @@ -27,45 +27,48 @@

Thread status

-

Local T3 cache

+

Live T3 connection

-

Reading cache

+

Connecting

- Looking for T3 Code. No sign-in required. + Checking your paired environments.

-
-
-
-

Refresh interval

-

The ring on the key runs until the next cache read.

-
- TIMER -
- -

- Choose 5 to 300 seconds. Press the key to refresh immediately. +

+ +

In T3 Code, open Settings → Connections, enable Network access if needed, and create a Read only pairing link. Paste the complete link below.

+ + + +

+

Pair once per environment for all keys. A new link for the same environment replaces its saved authorization. T3 currently requires a new link after the 30-day session expires.

+

Read only also permits access to files and conversations. This plugin requests only the thread status stream.

+ + +

+ +

Forgetting removes the credential from this plugin. Revoke its access in T3 Settings → Connections.

+
+ +
+ + +

+ Shows thread counts and a blinking question mark when input, approval, or plan review is pending.

+

For separate keys, add Thread status twice and choose a different display on each.

+ +

Status updates arrive live. Press a key to retry an offline connection. A full ring means the stream is connected.

diff --git a/plugin/property-inspector/property-inspector.js b/plugin/property-inspector/property-inspector.js index ab29d78..b097b6a 100644 --- a/plugin/property-inspector/property-inspector.js +++ b/plugin/property-inspector/property-inspector.js @@ -1,7 +1,4 @@ const ACTION_UUID = "com.beastyrabbit.t3-code-status.overview"; -const DEFAULT_REFRESH_SECONDS = 60; -const MIN_REFRESH_SECONDS = 5; -const MAX_REFRESH_SECONDS = 300; const SETTINGS_VERSION = 1; const MAX_ACTION_INFO_CODE_UNITS = 64 * 1024; const MAX_MESSAGE_CODE_UNITS = 64 * 1024; @@ -17,6 +14,8 @@ let protocolFailed = false; let connectionErrorShown = false; let messageWindowStartedAt = 0; let messagesInWindow = 0; +let authBusy = false; +let connections = []; const elements = {}; @@ -93,50 +92,86 @@ function cacheElements() { elements.connectionState = document.getElementById("connection-state"); elements.connectionDetail = document.getElementById("connection-detail"); elements.errorMessage = document.getElementById("error-message"); - elements.refreshInput = document.getElementById("refresh-seconds"); - elements.refreshNote = document.getElementById("refresh-note"); + elements.displayMode = document.getElementById("display-mode"); + elements.displayNote = document.getElementById("display-note"); + elements.pairingLink = document.getElementById("pairing-link"); + elements.allowHttp = document.getElementById("allow-http"); + elements.pairButton = document.getElementById("pair-button"); + elements.pairingResult = document.getElementById("pairing-result"); + elements.environmentList = document.getElementById("environment-list"); + elements.environmentDetail = document.getElementById("environment-detail"); + elements.removeConnection = document.getElementById("remove-connection"); } function bindControls() { - elements.refreshInput.addEventListener("change", saveRefreshInterval); + elements.displayMode.addEventListener("change", saveDisplayMode); + elements.pairButton.addEventListener("click", () => { + const link = elements.pairingLink.value.trim(); + if (!link || authBusy) return; + if (sendToPlugin({ command: "pair", link, allowHttp: elements.allowHttp.checked })) { + elements.pairingLink.value = ""; + authBusy = true; + setSettingsEnabled(socketReady); + setTextContent(elements.pairingResult, "Pairing…"); + } + }); + elements.removeConnection.addEventListener("click", () => { + if (!elements.environmentList.value || authBusy) return; + if (sendToPlugin({ command: "removeConnection", environmentId: elements.environmentList.value })) { + authBusy = true; + setSettingsEnabled(socketReady); + } + }); + elements.environmentList.addEventListener("change", renderEnvironmentDetail); } function setSettingsEnabled(enabled) { - elements.refreshInput.disabled = !enabled; + elements.displayMode.disabled = !enabled; + elements.pairingLink.disabled = !enabled || authBusy; + elements.allowHttp.disabled = !enabled || authBusy; + elements.pairButton.disabled = !enabled || authBusy; + elements.environmentList.disabled = !enabled || authBusy || connections.length === 0; + elements.removeConnection.disabled = !enabled || authBusy || connections.length === 0; } function applySettings(settings) { - const refreshSeconds = normalizeRefreshSeconds(settings?.refreshSeconds); - elements.refreshInput.value = String(refreshSeconds); + elements.displayMode.value = normalizeDisplayMode(settings?.displayMode); + updateDisplayNote(); +} + +function normalizeDisplayMode(value) { + return value === "threads" || value === "questions" ? value : "combined"; } -function normalizeRefreshSeconds(value) { - const parsed = Number(value); - if (!Number.isFinite(parsed)) return DEFAULT_REFRESH_SECONDS; - return Math.min(MAX_REFRESH_SECONDS, Math.max(MIN_REFRESH_SECONDS, Math.round(parsed))); +function updateDisplayNote() { + const descriptions = { + combined: + "Shows thread counts and a blinking question mark when input, approval, or plan review is pending.", + threads: "Shows working threads / all open threads. This key does not blink for questions.", + questions: + "A faded question mark when clear. The whole key flashes amber while input, approval, or plan review is pending.", + }; + elements.displayNote.textContent = descriptions[normalizeDisplayMode(elements.displayMode.value)]; } -function saveRefreshInterval() { - const entered = Number(elements.refreshInput.value); - const refreshSeconds = normalizeRefreshSeconds(elements.refreshInput.value); - elements.refreshInput.value = String(refreshSeconds); - const sent = sendSocketMessage({ +function saveSettings() { + return sendSocketMessage({ event: "setSettings", context: actionContext, - payload: { refreshSeconds, settingsVersion: SETTINGS_VERSION }, + payload: { + displayMode: normalizeDisplayMode(elements.displayMode.value), + settingsVersion: SETTINGS_VERSION, + }, }); - if (!sent) { - elements.refreshNote.textContent = "Interval not saved. Check the connection to OpenDeck."; - return; - } - const wasAdjusted = !Number.isFinite(entered) || entered !== refreshSeconds; - elements.refreshNote.textContent = wasAdjusted - ? `Adjusted to ${refreshSeconds} seconds. Choose a value from 5 to 300.` - : `Refresh set to ${refreshSeconds} seconds. Press the key to refresh immediately.`; +} + +function saveDisplayMode() { + if (saveSettings()) updateDisplayNote(); + else elements.displayNote.textContent = "Display not saved. Check the connection to OpenDeck."; } function sendToPlugin(payload) { - sendSocketMessage({ + return sendSocketMessage({ action: ACTION_UUID, event: "sendToPlugin", context: actionContext, @@ -182,6 +217,15 @@ function handleSocketMessage(event) { return; } + if (message.event === "sendToPropertyInspector" && message.payload?.type === "pairingResult") { + authBusy = false; + setSettingsEnabled(socketReady); + setTextContent( + elements.pairingResult, + message.payload.error ? localizeRuntimeError(message.payload.error) : "Connection settings saved.", + ); + return; + } if (message.event !== "sendToPropertyInspector" || message.payload?.type !== "connectionStatus") return; renderConnectionStatus(message.payload); } @@ -213,9 +257,12 @@ function isBoundedContext(value) { } function renderConnectionStatus(payload) { + if (Array.isArray(payload.status?.connections)) renderEnvironments(payload.status.connections); const status = normalizeConnectionStatus(payload.status); - const errorCode = typeof payload.error === "string" ? payload.error.trim() : ""; - const hasError = errorCode.length > 0; + const rawError = payload.error ?? payload.status?.error; + const errorCode = typeof rawError === "string" ? rawError.trim() : ""; + const hasError = + errorCode.length > 0 && !["connecting", "pairing-required", "authorization-required"].includes(errorCode); setBusy(Boolean(payload.busy)); if (hasError) { @@ -228,7 +275,7 @@ function renderConnectionStatus(payload) { if (payload.busy) { setTextContent(elements.connectionState, busyLabel()); - setTextContent(elements.connectionDetail, "Reading the local T3 cache."); + setTextContent(elements.connectionDetail, "Updating the live T3 connection…"); return; } if (hasError) { @@ -242,11 +289,26 @@ function renderConnectionStatus(payload) { setTextContent(elements.connectionState, "Connected"); setTextContent(elements.connectionDetail, connectedDetail(status)); break; + case "pairing-required": + setTextContent(elements.connectionState, "Pair T3 Code"); + setTextContent(elements.connectionDetail, "Add a read-only pairing link below to start live updates."); + break; + case "authorization-required": + setTextContent(elements.connectionState, "Pairing needed"); + setTextContent( + elements.connectionDetail, + "Authorization expired or was revoked. Paste a fresh read-only pairing link.", + ); + break; + case "connecting": + setTextContent(elements.connectionState, "Connecting"); + setTextContent(elements.connectionDetail, "Synchronizing thread status…"); + break; default: setTextContent(elements.connectionState, "T3 Code offline"); setTextContent( elements.connectionDetail, - "Start T3 Code. The plugin will then read its local thread cache.", + "Check that the paired environments are running and reachable. The plugin reconnects automatically.", ); } } @@ -255,37 +317,57 @@ function runtimeErrorLabel(code) { switch (code) { case "offline": return "T3 Code offline"; - case "cache-unavailable": - return "Cache unavailable"; + case "pairing-required": + return "Pair T3 Code"; + case "authorization-required": + return "Pairing needed"; + case "connecting": + return "Connecting"; case "invalid-response": - return "Cache incompatible"; + return "Connection incompatible"; default: - return "Cache read failed"; + return "Connection unavailable"; } } function runtimeErrorDetail(code) { - return code === "offline" - ? "Start T3 Code. The plugin will then read its local thread cache." - : "Thread status cannot update until the local cache is available."; + return localizeRuntimeError(code); } function localizeRuntimeError(code) { switch (code) { + case "pairing-required": + return "Add a read-only pairing link below to start live updates."; + case "authorization-required": + return "Authorization expired or was revoked. Paste a fresh read-only pairing link."; + case "connecting": + return "Synchronizing thread status…"; + case "invalid-link": + return "Paste the complete pairing link, including its token."; + case "insecure-origin": + return "Use HTTPS, or allow HTTP only if you trust this private network."; + case "identity-mismatch": + return "This address now belongs to a different T3 environment. Create a new pairing link."; + case "storage-error": + return "The plugin could not securely read or save its credentials. Check permissions on its private configuration folder."; + case "busy": + return "Another connection change is still in progress."; case "offline": - return "T3 Code is not running or its local thread cache is unavailable."; - case "cache-unavailable": - return "The local T3 thread cache could not be opened."; + return "A paired environment is unreachable. Retrying automatically."; case "invalid-response": - return "The local T3 thread cache contains unexpected data. Update T3 Code and try again."; + return "T3 returned an unexpected response. Check plugin and T3 Code compatibility."; default: - return "The local T3 thread cache could not be read."; + return "The connection could not be completed. Try a fresh read-only pairing link."; } } function normalizeConnectionStatus(status) { if (status && typeof status === "object") { - const state = status.state === "connected" ? "connected" : "offline"; + const state = ["connected", "connecting", "pairing-required", "authorization-required"].includes( + status.state, + ) + ? status.state + : "offline"; const environments = Number(status.environments); return { state, @@ -299,7 +381,7 @@ function normalizeConnectionStatus(status) { function connectedDetail(status) { const environmentLabel = status.environments === 1 ? "1 Environment" : `${status.environments} Environments`; - return `${environmentLabel} · local cache · no sign-in`; + return `${environmentLabel} · live stream · read only`; } function setBusy(busy) { @@ -307,7 +389,38 @@ function setBusy(busy) { } function busyLabel() { - return "Reading cache"; + return "Connecting"; +} + +function renderEnvironments(items) { + const previous = elements.environmentList.value; + connections = items.filter((item) => item && typeof item.environmentId === "string").slice(0, 16); + const options = connections.map((item) => { + const option = document.createElement("option"); + option.value = item.environmentId; + option.textContent = `${item.label} · ${item.state}`; + return option; + }); + if (!options.length) { + const option = document.createElement("option"); + option.value = ""; + option.textContent = "No paired environments"; + options.push(option); + } + elements.environmentList.replaceChildren(...options); + if (connections.some((item) => item.environmentId === previous)) elements.environmentList.value = previous; + renderEnvironmentDetail(); + setSettingsEnabled(socketReady); +} + +function renderEnvironmentDetail() { + const item = connections.find((item) => item.environmentId === elements.environmentList.value); + setTextContent( + elements.environmentDetail, + item + ? `${item.origin} · Authorization expires ${new Date(item.expiresAt).toLocaleDateString()}${item.error ? ` · ${localizeRuntimeError(item.error)}` : ""}` + : "", + ); } function setVisualState(state) { diff --git a/plugin/property-inspector/styles.css b/plugin/property-inspector/styles.css index d521d5b..befc7f1 100644 --- a/plugin/property-inspector/styles.css +++ b/plugin/property-inspector/styles.css @@ -31,10 +31,66 @@ body { #0d1316; } -input { +input, +select, +button { font: inherit; } +[hidden] { + display: none; +} + +#pairing-link, +button { + width: 100%; + min-height: 37px; + margin-top: 10px; + padding: 8px; + color: var(--ink); + border: 1px solid #607985; + background: var(--panel-deep); +} + +button { + cursor: pointer; +} +button:disabled { + opacity: 0.5; + cursor: not-allowed; +} +button:focus-visible, +#pairing-link:focus-visible { + outline: 2px solid var(--focus); + outline-offset: 2px; +} +.http-option { + display: block; +} +#environment-detail { + overflow-wrap: anywhere; +} + +select { + width: 100%; + min-height: 37px; + margin-top: 10px; + padding: 7px; + color: var(--ink); + border: 1px solid #607985; + background: var(--panel-deep); +} + +select:focus-visible { + outline: 2px solid var(--focus); + outline-offset: 2px; +} + +select:disabled { + opacity: 0.6; + cursor: not-allowed; +} + .panel { width: min(100%, 390px); margin: 0 auto; diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 3cc532f..e1fade6 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -8,9 +8,6 @@ importers: .: dependencies: - snappyjs: - specifier: ^0.7.0 - version: 0.7.0 ws: specifier: ^8.21.3 version: 8.21.3 @@ -33,6 +30,9 @@ importers: lefthook: specifier: ^2.1.12 version: 2.1.12 + snappyjs: + specifier: ^0.7.0 + version: 0.7.0 tsx: specifier: ^4.23.13 version: 4.23.13 diff --git a/scripts/build.ts b/scripts/build.ts index 782abfd..0f30b04 100644 --- a/scripts/build.ts +++ b/scripts/build.ts @@ -12,7 +12,7 @@ const distRoot = resolve(projectRoot, "dist"); const outputRoot = resolve(distRoot, PLUGIN_DIRECTORY); const setupOutput = resolve(projectRoot, "dist", "setup-opendeck.mjs"); const require = createRequire(import.meta.url); -const BUNDLED_DEPENDENCIES = ["snappyjs", "ws"] as const; +const BUNDLED_DEPENDENCIES = ["ws"] as const; const PLUGIN_SOURCE_FILES = [ "manifest.json", "icons/action.svg", diff --git a/scripts/verify-release.ts b/scripts/verify-release.ts index 7eb4612..7b2531f 100644 --- a/scripts/verify-release.ts +++ b/scripts/verify-release.ts @@ -23,20 +23,7 @@ const REQUIRED_PLUGIN_FILES = [ "property-inspector/property-inspector.js", "property-inspector/styles.css", ]; -const LEGACY_AUTH_MARKERS = [ - "credential.json", - "local-pairing", - "pairing-form", - "pairing-failed", - "pairing-route", - "pairing-section", - "pairingCode", - "pairWithT3Code", - "subject_token", - "Create pairing link", - "T3 Code koppeln", - "enqueueAuthOperation", -]; +const LEGACY_CACHE_MARKERS = ["readT3ShellCache", "T3CODE_CACHE_DIR", "indexeddb.leveldb"]; interface SetupModule { setupOpenDeck: ( @@ -73,7 +60,7 @@ export async function verifyRelease(projectRoot: string): Promise { "the built plugin contains unexpected or missing files", ); await verifyManifest(resolve(builtPlugin, "manifest.json"), expectedManifestVersion); - await verifyAuthFreeBundle(builtPlugin); + await verifyLiveBundle(builtPlugin); const temporaryRoot = await mkdtemp(join(tmpdir(), "t3-code-status-release-")); try { @@ -207,7 +194,7 @@ async function verifyManifest(path: string, expectedVersion: string): Promise { +async function verifyLiveBundle(pluginRoot: string): Promise { const checkedFiles = [ "bin/plugin.cjs", "property-inspector/index.html", @@ -216,8 +203,8 @@ async function verifyAuthFreeBundle(pluginRoot: string): Promise { ]; for (const path of checkedFiles) { const contents = await readFile(resolve(pluginRoot, path), "utf8"); - for (const marker of LEGACY_AUTH_MARKERS) { - assert.equal(contents.includes(marker), false, `${path} still contains legacy auth marker ${marker}`); + for (const marker of LEGACY_CACHE_MARKERS) { + assert.equal(contents.includes(marker), false, `${path} still contains legacy cache marker ${marker}`); } } } diff --git a/src/connection-store.ts b/src/connection-store.ts new file mode 100644 index 0000000..50eec71 --- /dev/null +++ b/src/connection-store.ts @@ -0,0 +1,99 @@ +import { randomUUID } from "node:crypto"; +import { constants } from "node:fs"; +import { lstat, mkdir, open, rename, rm } from "node:fs/promises"; +import { homedir } from "node:os"; +import { dirname, join } from "node:path"; +import { LiveConnectionError, MAX_ENVIRONMENTS, originFrom, record, textValue } from "./t3-protocol.js"; + +export interface SavedConnection { + environmentId: string; + label: string; + origin: string; + token: string; + expiresAt: number; + allowHttp: boolean; +} +export interface ConnectionStore { + load(): Promise; + save(connections: SavedConnection[]): Promise; +} + +// Separate from profiles and installation files so exports/upgrades never carry credentials. +export class FileConnectionStore implements ConnectionStore { + constructor( + private readonly path = join(homedir(), ".config", "opendeck-t3-code-status", "connections.json"), + ) {} + async load(): Promise { + try { + const handle = await open(this.path, constants.O_RDONLY | constants.O_NOFOLLOW | constants.O_NONBLOCK); + try { + const stat = await handle.stat(); + if ( + !stat.isFile() || + stat.size > 256 * 1024 || + (process.platform !== "win32" && ((stat.mode & 0o077) !== 0 || stat.uid !== process.getuid?.())) + ) + throw new Error(); + const buffer = Buffer.alloc(256 * 1024 + 1); + let bytes = 0; + while (bytes < buffer.length) { + const result = await handle.read(buffer, bytes, buffer.length - bytes, null); + if (result.bytesRead === 0) break; + bytes += result.bytesRead; + } + if (bytes > 256 * 1024) throw new Error(); + const value: unknown = JSON.parse(buffer.subarray(0, bytes).toString("utf8")); + if (!Array.isArray(value) || value.length > MAX_ENVIRONMENTS) throw new Error(); + const ids = new Set(); + return value.map((item) => { + if ( + !record(item) || + !textValue(item.environmentId) || + ids.has(item.environmentId) || + !textValue(item.label, 120) || + !textValue(item.origin, 2048) || + !textValue(item.token, 8192) || + typeof item.expiresAt !== "number" || + !Number.isFinite(item.expiresAt) || + typeof item.allowHttp !== "boolean" || + originFrom(item.origin, item.allowHttp) !== item.origin + ) + throw new Error(); + ids.add(item.environmentId); + return item as unknown as SavedConnection; + }); + } finally { + await handle.close(); + } + } catch (error) { + if (record(error) && error.code === "ENOENT") return []; + throw new LiveConnectionError("storage-error"); + } + } + async save(connections: SavedConnection[]): Promise { + const temporary = `${this.path}.${randomUUID()}.tmp`; + try { + const folder = dirname(this.path); + await mkdir(folder, { recursive: true, mode: 0o700 }); + const stat = await lstat(folder); + if ( + !stat.isDirectory() || + stat.isSymbolicLink() || + (process.platform !== "win32" && (stat.uid !== process.getuid?.() || (stat.mode & 0o077) !== 0)) + ) + throw new Error(); + const handle = await open(temporary, "wx", 0o600); + try { + await handle.writeFile(JSON.stringify(connections)); + await handle.sync(); + } finally { + await handle.close(); + } + await rename(temporary, this.path); + } catch { + throw new LiveConnectionError("storage-error"); + } finally { + await rm(temporary, { force: true }).catch(() => undefined); + } + } +} diff --git a/src/controller.ts b/src/controller.ts index c8eee77..9a27025 100644 --- a/src/controller.ts +++ b/src/controller.ts @@ -4,10 +4,13 @@ import { type OpenDeckConnection, type OpenDeckEvent, } from "./opendeck.js"; -import { getAccessibleTitle, renderDashboard } from "./render.js"; -import { T3ClientError, type T3ClientSnapshot } from "./t3-client.js"; +import { getAccessibleTitle, hasOpenQuestions, renderDashboard } from "./render.js"; +import type { T3ClientSnapshot } from "./t3-client.js"; +import { T3ClientError } from "./t3-error.js"; +import { LiveConnectionError } from "./t3-protocol.js"; import { ACTION_UUID, + type ActionSettings, type ConnectionStatus, type DashboardModel, DEFAULT_REFRESH_SECONDS, @@ -16,7 +19,7 @@ import { SETTINGS_VERSION, } from "./types.js"; -const ANIMATION_INTERVAL_MS = 1_000; +const ANIMATION_INTERVAL_MS = 500; const CONNECTION_STATUS_CACHE_MS = 1_000; const COMMAND_BURST_WINDOW_MS = 250; const ERROR_RETRY_INTERVAL_MS = 5_000; @@ -24,6 +27,11 @@ const LEGACY_REFRESH_SECONDS = 15; const RING_PROGRESS_STEPS = 20; export interface T3StatusClient { + subscribe?(listener: () => void): () => void; + pair?(link: string, allowHttp: boolean): Promise; + remove?(environmentId: string): Promise; + reconnect?(): Promise; + dispose?(): Promise; getConnectionStatus(): Promise; getSnapshot(): Promise; } @@ -37,6 +45,8 @@ interface VisibleContext { context: string; cycleStartedAt: number; lastRenderedModel?: DashboardModel; + lastRenderedMode?: NormalizedSettings["displayMode"]; + lastRenderedAlertPhase?: boolean; lastRenderedProgressStep?: number; lastRenderedTitle?: string; model: DashboardModel; @@ -67,6 +77,8 @@ export class T3CodeController { private connectionStatusReportInFlight?: Promise; private connectionStatusReadAt = 0; private disposed = false; + private unsubscribe?: () => void; + private authInFlight?: Promise; private lastConnectionError?: string; private lastConnectionStatus: ConnectionStatus = { state: "offline" }; private refreshInFlight?: Promise; @@ -78,6 +90,13 @@ export class T3CodeController { ) { this.animationIntervalMs = options.animationIntervalMs ?? ANIMATION_INTERVAL_MS; this.now = options.now ?? Date.now; + this.unsubscribe = client.subscribe?.(() => { + if (this.disposed) return; + this.connectionStatusKnown = false; + this.scheduleRefresh(this.visibleContexts.keys(), true); + for (const context of this.inspectorContexts) this.pendingConnectionStatusContexts.add(context); + this.startConnectionStatusReports(); + }); } handle(event: OpenDeckEvent): void { @@ -110,6 +129,8 @@ export class T3CodeController { async dispose(): Promise { if (this.disposed) return; this.disposed = true; + this.unsubscribe?.(); + await this.client.dispose?.(); if (this.animationTimer) clearInterval(this.animationTimer); this.animationTimer = undefined; this.pendingRefresh.clear(); @@ -123,6 +144,7 @@ export class T3CodeController { this.refreshInFlight, this.connectionStatusInFlight, this.connectionStatusReportInFlight, + this.authInFlight, ]); } @@ -198,6 +220,7 @@ export class T3CodeController { const lastRefresh = this.lastForcedRefreshAt.get(context); if (lastRefresh !== undefined && now - lastRefresh < COMMAND_BURST_WINDOW_MS) return; this.lastForcedRefreshAt.set(context, now); + void this.client.reconnect?.(); this.scheduleRefresh([context], true); } @@ -224,6 +247,7 @@ export class T3CodeController { } private isDue(visible: VisibleContext, now: number): boolean { + if (this.client.subscribe) return false; return now - visible.cycleStartedAt >= this.refreshIntervalMs(visible); } @@ -237,15 +261,30 @@ export class T3CodeController { const visible = this.visibleContexts.get(context); if (!visible) return; const duration = this.refreshIntervalMs(visible); - const progress = Math.min(1, Math.max(0, (now - visible.cycleStartedAt) / duration)); + const progress = this.client.subscribe + ? visible.model.kind === "ready" + ? 1 + : 0 + : Math.min(1, Math.max(0, (now - visible.cycleStartedAt) / duration)); const progressStep = Math.min(RING_PROGRESS_STEPS, Math.floor(progress * RING_PROGRESS_STEPS)); - const title = getAccessibleTitle(visible.model); + const mode = visible.settings.displayMode; + const alertPhase = + mode !== "threads" && hasOpenQuestions(visible.model) && Math.floor(now / 500) % 2 === 0; + const title = getAccessibleTitle(visible.model, mode); const imageChanged = - visible.lastRenderedModel !== visible.model || visible.lastRenderedProgressStep !== progressStep; + visible.lastRenderedModel !== visible.model || + visible.lastRenderedProgressStep !== progressStep || + visible.lastRenderedMode !== mode || + visible.lastRenderedAlertPhase !== alertPhase; if (imageChanged) { visible.lastRenderedModel = visible.model; visible.lastRenderedProgressStep = progressStep; - this.host.setImage(context, renderDashboard(visible.model, progressStep / RING_PROGRESS_STEPS)); + visible.lastRenderedMode = mode; + visible.lastRenderedAlertPhase = alertPhase; + this.host.setImage( + context, + renderDashboard(visible.model, progressStep / RING_PROGRESS_STEPS, mode, alertPhase), + ); } if (visible.lastRenderedTitle !== title) { visible.lastRenderedTitle = title; @@ -293,7 +332,9 @@ export class T3CodeController { } catch (error) { model = modelForClientError(error); const code = clientErrorCode(error); - if (code === "offline") this.rememberConnectionStatus({ state: "offline" }); + if (this.client.subscribe) + this.rememberConnectionStatus(await this.client.getConnectionStatus(), code); + else if (code === "offline") this.rememberConnectionStatus({ state: "offline" }); else this.rememberConnectionStatus(this.lastConnectionStatus, code); this.broadcastConnectionStatus(this.lastConnectionStatus, false, this.lastConnectionError); } @@ -328,6 +369,56 @@ export class T3CodeController { private handleInspectorCommand(event: OpenDeckEvent): void { if (!event.context || !isRecord(event.payload) || typeof event.payload.command !== "string") return; switch (event.payload.command) { + case "pair": + case "removeConnection": { + if (!this.trackInspectorContext(event.context)) return; + if (this.authInFlight) { + this.host.sendToPropertyInspector(ACTION_UUID, event.context, { + type: "pairingResult", + error: "busy", + }); + return; + } + const payload = event.payload; + const context = event.context; + if ( + payload.command === "pair" && + (typeof payload.link !== "string" || payload.link.length > 16_384 || !this.client.pair) + ) + return; + if ( + payload.command === "removeConnection" && + (typeof payload.environmentId !== "string" || + payload.environmentId.length > 1024 || + !this.client.remove) + ) + return; + this.broadcastConnectionStatus(this.lastConnectionStatus, true); + this.authInFlight = (async () => { + let error: string | undefined; + try { + if (payload.command === "pair") + await this.client.pair?.(payload.link as string, payload.allowHttp === true); + else await this.client.remove?.(payload.environmentId as string); + } catch (cause) { + error = clientErrorCode(cause); + } + const status = await this.client.getConnectionStatus(); + this.rememberConnectionStatus(status); + this.broadcastConnectionStatus(status, false); + if (!this.disposed) + this.host.sendToPropertyInspector(ACTION_UUID, context, { + type: "pairingResult", + ...(error ? { error } : {}), + }); + this.scheduleRefresh(this.visibleContexts.keys(), true); + })() + .catch(() => undefined) + .finally(() => { + this.authInFlight = undefined; + }); + break; + } case "getConnectionStatus": { if (!this.trackInspectorContext(event.context)) return; this.queueConnectionStatusReport(event.context); @@ -438,31 +529,39 @@ export class T3CodeController { } } -function readSettings(payload: unknown): { refreshSeconds?: number; settingsVersion?: number } | undefined { +function readSettings(payload: unknown): ActionSettings | undefined { if (!isRecord(payload) || !isRecord(payload.settings)) return undefined; const value = payload.settings.refreshSeconds; const version = payload.settings.settingsVersion; const refreshSeconds = typeof value === "number" ? value : typeof value === "string" && value.trim() ? Number(value) : undefined; return { + displayMode: normalizeSettings({ + displayMode: payload.settings.displayMode as ActionSettings["displayMode"], + }).displayMode, ...(refreshSeconds === undefined ? {} : { refreshSeconds }), ...(typeof version === "number" ? { settingsVersion: version } : {}), }; } -function migrateSettings(settings: { refreshSeconds?: number; settingsVersion?: number } | undefined): { - refreshSeconds: number; - settingsVersion: number; -} { +function migrateSettings( + settings: ActionSettings | undefined, +): NormalizedSettings & { settingsVersion: number } { if (settings?.settingsVersion === SETTINGS_VERSION) { return { ...normalizeSettings(settings), settingsVersion: SETTINGS_VERSION }; } const refreshSeconds = settings?.refreshSeconds === LEGACY_REFRESH_SECONDS ? DEFAULT_REFRESH_SECONDS : settings?.refreshSeconds; - return { ...normalizeSettings({ refreshSeconds }), settingsVersion: SETTINGS_VERSION }; + return { ...normalizeSettings({ ...settings, refreshSeconds }), settingsVersion: SETTINGS_VERSION }; } function modelForClientError(error: unknown): DashboardModel { + if (error instanceof LiveConnectionError) { + if (error.code === "pairing-required" || error.code === "authorization-required") + return { kind: "pairing" }; + if (error.code === "connecting") return { kind: "loading" }; + return { kind: error.code === "offline" ? "offline" : "error" }; + } if (!(error instanceof T3ClientError)) return { kind: "error" }; switch (error.code) { case "offline": @@ -473,7 +572,9 @@ function modelForClientError(error: unknown): DashboardModel { } function clientErrorCode(error: unknown): string { - return error instanceof T3ClientError ? error.code : "cache-read-failed"; + return error instanceof T3ClientError || error instanceof LiveConnectionError + ? error.code + : "invalid-response"; } function isRecord(value: unknown): value is Record { diff --git a/src/index.ts b/src/index.ts index 6f31a5f..ca2545d 100644 --- a/src/index.ts +++ b/src/index.ts @@ -2,11 +2,11 @@ import { T3CodeController } from "./controller.js"; import { OpenDeckHost } from "./opendeck.js"; -import { T3Client } from "./t3-client.js"; +import { T3LiveClient } from "./t3-live-client.js"; export async function main(argumentsList = process.argv.slice(2)): Promise { const host = new OpenDeckHost(argumentsList); - const controller = new T3CodeController(host, new T3Client()); + const controller = new T3CodeController(host, new T3LiveClient()); host.onEvent((event) => controller.handle(event)); let stopping = false; diff --git a/src/render.ts b/src/render.ts index 7134537..3526c13 100644 --- a/src/render.ts +++ b/src/render.ts @@ -1,4 +1,4 @@ -import type { DashboardModel, ThreadSummary } from "./types.js"; +import type { DashboardModel, DisplayMode, ThreadSummary } from "./types.js"; const COLORS = { cyan: "#51D7E8", @@ -27,29 +27,65 @@ interface Display { label: string; } -export function getDisplay(model: DashboardModel): Display { - if (model.kind === "ready") return readyDisplay(model.summary); +export function hasOpenQuestions(model: DashboardModel): boolean { + return model.kind === "ready" && directAttentionCount(model.summary) > 0; +} + +function directAttentionCount(summary: ThreadSummary): number { + return summary.approval + summary.input + summary.plan; +} + +export function getDisplay(model: DashboardModel, mode: DisplayMode = "combined"): Display { + if (model.kind === "ready") { + if (mode === "questions") { + const count = directAttentionCount(model.summary); + return { + accent: count > 0 ? COLORS.yellow : COLORS.muted, + count: "?", + footer: count > 0 ? `${count} ${count === 1 ? "NEEDS YOU" : "NEED YOU"}` : "NO QUESTIONS", + label: "QUESTIONS", + }; + } + if (mode === "threads") { + return { + accent: runningAccent(model.summary.running, model.summary.total), + count: `${model.summary.running}/${model.summary.total}`, + footer: model.summary.total === 0 ? "NO THREADS" : "WORKING", + label: "T3 CODE", + }; + } + return readyDisplay(model.summary); + } if (model.kind === "loading") { return { accent: COLORS.cyan, count: "···", footer: "LOADING", label: "T3 CODE" }; } if (model.kind === "offline") { return { accent: COLORS.muted, count: "OFF", footer: "T3 OFFLINE", label: "T3 CODE" }; } + if (model.kind === "pairing") + return { accent: COLORS.yellow, count: "LINK", footer: "OPEN SETTINGS", label: "T3 CODE" }; return { accent: COLORS.red, count: "ERR", footer: "STATUS ERROR", label: "T3 CODE" }; } -export function getAccessibleTitle(model: DashboardModel): string { +export function getAccessibleTitle(model: DashboardModel, mode: DisplayMode = "combined"): string { if (model.kind === "loading") return "Loading T3 Code status"; if (model.kind === "offline") return "T3 Code offline"; if (model.kind === "error") return "T3 Code status error"; + if (model.kind === "pairing") return "Pair T3 Code in the OpenDeck key settings"; const { summary } = model; + const directAttention = directAttentionCount(summary); + if (mode === "questions") { + return directAttention > 0 + ? `${directAttention} ${directAttention === 1 ? "thread needs" : "threads need"} your input, approval, or plan review` + : "No T3 Code questions pending"; + } if (summary.total === 0) return "No open T3 Code threads"; + if (mode === "threads") return `${summary.running} of ${summary.total} threads working`; const working = `${summary.running} of ${summary.total} working`; if (summary.failed > 0) { - return `${working}, ${summary.failed} ${summary.failed === 1 ? "error" : "errors"}`; + return `${working}, ${summary.failed} ${summary.failed === 1 ? "error" : "errors"}${directAttention > 0 ? `, ${directAttention} need your attention` : ""}`; } - const directAttention = summary.approval + summary.input + summary.plan; if (directAttention > 0) { return `${working}, ${directAttention} ${directAttention === 1 ? "needs" : "need"} your attention`; } @@ -108,8 +144,23 @@ function interpolateColor(from: Rgb, to: Rgb, amount: number): string { return `#${channel(from[0], to[0])}${channel(from[1], to[1])}${channel(from[2], to[2])}`; } -export function renderDashboard(model: DashboardModel, progress: number): string { - const display = getDisplay(model); +export function renderDashboard( + model: DashboardModel, + progress: number, + mode: DisplayMode = "combined", + alertPhase = true, +): string { + const display = getDisplay(model, mode); + const alert = mode !== "threads" && hasOpenQuestions(model); + const questionsOnly = mode === "questions" && model.kind === "ready"; + const brightBackground = questionsOnly && alert && alertPhase; + const foreground = brightBackground ? COLORS.ink : COLORS.paper; + const accent = brightBackground ? COLORS.ink : alert ? COLORS.yellow : display.accent; + if (alert && model.kind === "ready") { + const count = directAttentionCount(model.summary); + display.footer = `${count} ${count === 1 ? "NEEDS YOU" : "NEED YOU"}`; + display.footerAccent = accent; + } const boundedProgress = Math.min(1, Math.max(0, progress)); const radius = 61; const circumference = 2 * Math.PI * radius; @@ -120,18 +171,26 @@ export function renderDashboard(model: DashboardModel, progress: number): string const countSize = display.count.length <= 3 ? 34 : Math.max(14, Math.floor(115 / display.count.length)); const footerSize = display.footer.length >= 13 ? 10 : display.footer.length >= 12 ? 11 : 12; const footerSpacing = display.footer.length >= 13 ? 0.3 : display.footer.length >= 12 ? 0.4 : 0.6; + const question = `?`; + const content = questionsOnly + ? question + : alert + ? `${question}${escapeXml(display.count)}` + : `${escapeXml(display.count)}`; return svgData(` - + - - - + ${display.label} - ${escapeXml(display.count)} + ${content} ${escapeXml(display.footer)} `); diff --git a/src/t3-client.ts b/src/t3-client.ts index 2b7eba7..445557d 100644 --- a/src/t3-client.ts +++ b/src/t3-client.ts @@ -11,8 +11,11 @@ import { sameFileIdentity, T3CacheError, } from "./t3-cache.js"; +import { T3ClientError } from "./t3-error.js"; import type { ConnectionStatus, T3ShellSnapshot, ThreadSummary } from "./types.js"; +export { T3ClientError, type T3ClientErrorCode } from "./t3-error.js"; + const DEFAULT_AUTO_SETTLE_AFTER_DAYS = 3; const MIN_AUTO_SETTLE_AFTER_DAYS = 1; const MAX_AUTO_SETTLE_AFTER_DAYS = 90; @@ -25,20 +28,6 @@ const READ_ONLY_FILE_FLAGS = ? fileConstants.O_RDONLY : fileConstants.O_RDONLY | fileConstants.O_NONBLOCK | fileConstants.O_NOFOLLOW; -export type T3ClientErrorCode = - | "offline" - | "unsafe-origin" - | "invalid-response" - | "cache-unavailable" - | "cache-read-failed"; - -export class T3ClientError extends Error { - constructor(readonly code: T3ClientErrorCode) { - super(code); - this.name = "T3ClientError"; - } -} - interface RuntimeDescriptor { origin: string; } diff --git a/src/t3-error.ts b/src/t3-error.ts new file mode 100644 index 0000000..6a22e1d --- /dev/null +++ b/src/t3-error.ts @@ -0,0 +1,13 @@ +export type T3ClientErrorCode = + | "offline" + | "unsafe-origin" + | "invalid-response" + | "cache-unavailable" + | "cache-read-failed"; + +export class T3ClientError extends Error { + constructor(readonly code: T3ClientErrorCode) { + super(code); + this.name = "T3ClientError"; + } +} diff --git a/src/t3-live-client.ts b/src/t3-live-client.ts new file mode 100644 index 0000000..96aa6a2 --- /dev/null +++ b/src/t3-live-client.ts @@ -0,0 +1,508 @@ +import WebSocket from "ws"; +import { type ConnectionStore, FileConnectionStore, type SavedConnection } from "./connection-store.js"; +import { emptySummary, summarizeSnapshot } from "./status.js"; +import type { T3ClientSnapshot } from "./t3-client.js"; +import { + LiveConnectionError, + type LiveErrorCode, + MAX_ENVIRONMENTS, + MAX_RESPONSE_BYTES, + MAX_THREADS, + parsePairingLink, + parseThread, + record, + sequence, + textValue, +} from "./t3-protocol.js"; +import type { ConnectionStatus, EnvironmentStatus, T3ThreadShell } from "./types.js"; + +interface Environment { + saved: SavedConnection; + threads: Map; + sequence?: number; + state: EnvironmentStatus["state"]; + error?: LiveErrorCode; + socket?: WebSocket; + retry?: NodeJS.Timeout; + heartbeat?: NodeJS.Timeout; + syncTimeout?: NodeJS.Timeout; + abort?: AbortController; + pong: boolean; + attempt: number; + generation: number; +} +export interface LiveClientOptions { + store?: ConnectionStore; + fetchImpl?: typeof fetch; + now?: () => number; + retryMs?: number; + heartbeatMs?: number; +} + +export class T3LiveClient { + private readonly store: ConnectionStore; + private readonly fetchImpl: typeof fetch; + private readonly now: () => number; + private readonly listeners = new Set<() => void>(); + private readonly environments = new Map(); + private readonly ready: Promise; + private readonly clockTimer: NodeJS.Timeout; + private disposed = false; + private busy = false; + private loadError?: LiveErrorCode; + private notificationTimer?: NodeJS.Timeout; + + constructor(private readonly options: LiveClientOptions = {}) { + this.store = options.store ?? new FileConnectionStore(); + this.fetchImpl = options.fetchImpl ?? fetch; + this.now = options.now ?? Date.now; + this.ready = this.restore(); + // Re-evaluate time-based snoozes/settlement locally; this makes no HTTP requests. + this.clockTimer = setInterval(() => this.emit(), 30_000); + this.clockTimer.unref(); + } + subscribe(listener: () => void): () => void { + this.listeners.add(listener); + return () => this.listeners.delete(listener); + } + private emit(): void { + if (this.disposed || this.notificationTimer) return; + // Bound inspector traffic and rendering during a burst of shell events. + this.notificationTimer = setTimeout(() => { + this.notificationTimer = undefined; + if (!this.disposed) for (const listener of this.listeners) listener(); + }, 50); + this.notificationTimer.unref(); + } + private async restore(): Promise { + try { + const saved = await this.store.load(); + if (this.disposed) return; + for (const connection of saved) this.install(connection); + } catch { + this.loadError = "storage-error"; + } + this.emit(); + } + private install(saved: SavedConnection): void { + const old = this.environments.get(saved.environmentId); + if (old) this.stop(old); + const env: Environment = { + saved, + threads: new Map(), + state: "connecting", + pong: true, + attempt: 0, + generation: 0, + }; + this.environments.set(saved.environmentId, env); + void this.connect(env); + } + async pair(link: string, allowHttp = false): Promise { + await this.ready; + if (this.busy) throw new LiveConnectionError("busy"); + if (this.disposed) return; + if (this.loadError) throw new LiveConnectionError(this.loadError); + this.busy = true; + try { + const target = parsePairingLink(link, allowHttp); + const descriptor = await this.request(target.origin, "/.well-known/t3/environment"); + if (!record(descriptor) || !textValue(descriptor.environmentId)) + throw new LiveConnectionError("invalid-response"); + if (!this.environments.has(descriptor.environmentId) && this.environments.size >= MAX_ENVIRONMENTS) + throw new LiveConnectionError("invalid-response"); + const response = await this.request(target.origin, "/oauth/token", { + method: "POST", + headers: { "content-type": "application/x-www-form-urlencoded" }, + body: new URLSearchParams({ + grant_type: "urn:ietf:params:oauth:grant-type:token-exchange", + subject_token: target.credential, + subject_token_type: "urn:t3:params:oauth:token-type:environment-bootstrap", + requested_token_type: "urn:ietf:params:oauth:token-type:access_token", + scope: "orchestration:read", + client_label: "OpenDeck T3 Code Status", + }), + }); + if ( + !record(response) || + !textValue(response.access_token, 8192) || + response.token_type !== "Bearer" || + response.scope !== "orchestration:read" || + typeof response.expires_in !== "number" || + !Number.isFinite(response.expires_in) || + response.expires_in <= 0 || + response.expires_in > 366 * 86400 + ) + throw new LiveConnectionError("invalid-response"); + const saved: SavedConnection = { + environmentId: descriptor.environmentId, + label: textValue(descriptor.label, 120) ? descriptor.label : new URL(target.origin).host, + origin: target.origin, + token: response.access_token, + expiresAt: this.now() + response.expires_in * 1000, + allowHttp, + }; + const session = await this.request(target.origin, "/api/auth/session", { + headers: { authorization: `Bearer ${saved.token}` }, + }); + if ( + !record(session) || + session.authenticated !== true || + !Array.isArray(session.scopes) || + session.scopes.length !== 1 || + session.scopes[0] !== "orchestration:read" + ) + throw new LiveConnectionError("authorization-required"); + if (this.disposed) return; + await this.store.save( + [...this.environments.values()] + .filter((env) => env.saved.environmentId !== saved.environmentId) + .map((env) => env.saved) + .concat(saved), + ); + if (!this.disposed) this.install(saved); + } finally { + this.busy = false; + this.emit(); + } + } + async remove(environmentId: string): Promise { + await this.ready; + if (this.busy) throw new LiveConnectionError("busy"); + const env = this.environments.get(environmentId); + if (!env || this.disposed) return; + this.busy = true; + try { + await this.store.save( + [...this.environments.values()].filter((item) => item !== env).map((item) => item.saved), + ); + this.stop(env); + this.environments.delete(environmentId); + } finally { + this.busy = false; + this.emit(); + } + } + async reconnect(): Promise { + await this.ready; + for (const env of this.environments.values()) { + if (env.state === "connected" || env.state === "connecting" || env.state === "authorization-required") + continue; + this.stop(env); + void this.connect(env); + } + } + async getConnectionStatus(): Promise { + await this.ready; + const connections: EnvironmentStatus[] = [...this.environments.values()].map((env) => ({ + environmentId: env.saved.environmentId, + label: env.saved.label, + origin: env.saved.origin, + expiresAt: env.saved.expiresAt, + state: env.state, + ...(env.error ? { error: env.error } : {}), + })); + const state = this.loadError + ? "error" + : connections.length === 0 + ? "pairing-required" + : connections.some((env) => env.state === "authorization-required") + ? "authorization-required" + : connections.every((env) => env.state === "connected") + ? "connected" + : connections.some((env) => env.state === "connecting") + ? "connecting" + : "offline"; + return { + state, + origin: connections[0]?.origin ?? "", + environments: connections.length, + connections, + ...(this.loadError ? { error: this.loadError } : {}), + }; + } + async getSnapshot(): Promise { + const connectionStatus = await this.getConnectionStatus(); + if (connectionStatus.state !== "connected") + throw new LiveConnectionError( + connectionStatus.state === "error" ? (this.loadError ?? "invalid-response") : connectionStatus.state, + ); + const summary = emptySummary(); + for (const env of this.environments.values()) { + const part = summarizeSnapshot( + { + snapshotSequence: env.sequence ?? 0, + threads: [...env.threads.values()], + updatedAt: new Date(this.now()).toISOString(), + }, + this.now(), + ); + for (const key of Object.keys(summary) as Array) summary[key] += part[key]; + } + return { connectionStatus, summary }; + } + async dispose(): Promise { + this.disposed = true; + clearInterval(this.clockTimer); + clearTimeout(this.notificationTimer); + for (const env of this.environments.values()) this.stop(env); + this.listeners.clear(); + await this.ready; + } + private stop(env: Environment): void { + env.generation++; + env.abort?.abort(); + env.abort = undefined; + clearTimeout(env.retry); + clearTimeout(env.syncTimeout); + clearInterval(env.heartbeat); + env.retry = undefined; + env.heartbeat = undefined; + env.syncTimeout = undefined; + const socket = env.socket; + env.socket = undefined; + socket?.terminate(); + } + private fail(env: Environment, code: LiveErrorCode): void { + this.stop(env); + if (this.disposed) return; + env.error = code; + env.state = code === "authorization-required" ? "authorization-required" : "offline"; + if (code !== "authorization-required" && code !== "identity-mismatch") { + const delay = Math.min(30_000, (this.options.retryMs ?? 1000) * 2 ** Math.min(env.attempt++, 5)); + env.retry = setTimeout(() => { + env.retry = undefined; + void this.connect(env); + }, delay); + env.retry.unref(); + } + this.emit(); + } + private async connect(env: Environment): Promise { + if (this.disposed) return; + if (this.now() >= env.saved.expiresAt) { + this.fail(env, "authorization-required"); + return; + } + const generation = ++env.generation; + const current = () => !this.disposed && generation === env.generation; + env.state = "connecting"; + env.error = undefined; + env.abort = new AbortController(); + this.emit(); + try { + const descriptor = await this.request( + env.saved.origin, + "/.well-known/t3/environment", + {}, + env.abort.signal, + ); + if (!current()) return; + if (!record(descriptor) || descriptor.environmentId !== env.saved.environmentId) + throw new LiveConnectionError("identity-mismatch"); + const marker = + record(descriptor.capabilities) && descriptor.capabilities.shellResumeCompletionMarker === true; + const ticket = await this.request( + env.saved.origin, + "/api/auth/websocket-ticket", + { method: "POST", headers: { authorization: `Bearer ${env.saved.token}` } }, + env.abort.signal, + ); + if (!current()) return; + if (!record(ticket) || !textValue(ticket.ticket, 8192)) + throw new LiveConnectionError("invalid-response"); + const url = new URL("/ws", env.saved.origin); + url.protocol = url.protocol === "https:" ? "wss:" : "ws:"; + url.searchParams.set("wsTicket", ticket.ticket); + const socket = new WebSocket(url, { + handshakeTimeout: 10_000, + maxPayload: MAX_RESPONSE_BYTES, + followRedirects: false, + }); + env.socket = socket; + socket.on("error", () => { + /* Never log socket URLs or server responses. */ + }); + socket.on("unexpected-response", (_request, response) => { + response.destroy(); + if (current()) + this.fail( + env, + response.statusCode === 401 || response.statusCode === 403 ? "authorization-required" : "offline", + ); + }); + socket.on("close", () => { + if (current()) this.fail(env, "offline"); + }); + socket.on("open", () => { + if (!current()) { + socket.terminate(); + return; + } + const payload = marker + ? { + ...(env.sequence === undefined ? {} : { afterSequence: env.sequence }), + requestCompletionMarker: true, + } + : {}; + // Without a completion marker, request a full snapshot so stale state is never shown as live. + socket.send( + JSON.stringify({ + _tag: "Request", + id: "shell", + tag: "orchestration.subscribeShell", + payload, + headers: [], + }), + ); + env.syncTimeout = setTimeout(() => { + if (current()) this.fail(env, "offline"); + }, 15_000); + env.syncTimeout.unref(); + env.pong = true; + env.heartbeat = setInterval(() => { + if (!env.pong) { + if (current()) this.fail(env, "offline"); + return; + } + env.pong = false; + if (socket.readyState === WebSocket.OPEN) socket.send(JSON.stringify({ _tag: "Ping" })); + }, this.options.heartbeatMs ?? 30_000); + env.heartbeat.unref(); + }); + socket.on("message", (data) => { + if (!current()) return; + try { + const value: unknown = JSON.parse(data.toString()); + const messages = Array.isArray(value) ? value : [value]; + if (messages.length > 10_000) throw new Error(); + for (const message of messages) { + if (!record(message)) throw new Error(); + if (message._tag === "Pong") { + env.pong = true; + continue; + } + if (message._tag !== "Chunk" || message.requestId !== "shell" || !Array.isArray(message.values)) + throw new Error(); + for (const item of message.values) this.apply(env, item, marker); + if (socket.readyState === WebSocket.OPEN) + socket.send(JSON.stringify({ _tag: "Ack", requestId: "shell" })); + } + if (env.state === "connected") this.emit(); + } catch { + env.sequence = undefined; + env.threads.clear(); + this.fail(env, "invalid-response"); + } + }); + } catch (error) { + if (current()) this.fail(env, error instanceof LiveConnectionError ? error.code : "offline"); + } + } + private apply(env: Environment, item: unknown, marker: boolean): void { + if (!record(item)) throw new Error(); + if (item.kind === "snapshot") { + const snapshot = item.snapshot; + if ( + !record(snapshot) || + !sequence(snapshot.snapshotSequence) || + !Array.isArray(snapshot.threads) || + snapshot.threads.length > MAX_THREADS + ) + throw new Error(); + const threads = new Map(); + for (const raw of snapshot.threads) { + const thread = parseThread(raw); + if (threads.has(thread.id)) throw new Error(); + threads.set(thread.id, thread); + } + env.threads = threads; + env.sequence = snapshot.snapshotSequence; + if (!marker) this.synchronized(env); + return; + } + if (item.kind === "synchronized") { + if (env.sequence === undefined) throw new Error(); + this.synchronized(env); + return; + } + if (!sequence(item.sequence) || env.sequence === undefined) throw new Error(); + if (item.sequence <= env.sequence) return; + switch (item.kind) { + case "thread-upserted": { + const thread = parseThread(item.thread); + if (!env.threads.has(thread.id) && env.threads.size >= MAX_THREADS) throw new Error(); + env.threads.set(thread.id, thread); + break; + } + case "thread-removed": + if (!textValue(item.threadId)) throw new Error(); + env.threads.delete(item.threadId); + break; + case "project-upserted": + case "project-removed": + break; + default: + throw new Error(); + } + env.sequence = item.sequence; + } + private synchronized(env: Environment): void { + env.state = "connected"; + env.error = undefined; + env.attempt = 0; + clearTimeout(env.syncTimeout); + env.syncTimeout = undefined; + } + private async request( + origin: string, + path: string, + init: RequestInit = {}, + signal?: AbortSignal, + ): Promise { + const controller = new AbortController(); + const abort = () => controller.abort(); + signal?.addEventListener("abort", abort, { once: true }); + if (signal?.aborted) controller.abort(); + const timeout = setTimeout(abort, 10_000); + try { + const response = await this.fetchImpl(new URL(path, origin), { + ...init, + redirect: "error", + signal: controller.signal, + }); + if (!response.ok) { + await response.body?.cancel(); + throw new LiveConnectionError( + response.status === 401 || response.status === 403 ? "authorization-required" : "offline", + ); + } + if (!response.body) throw new LiveConnectionError("invalid-response"); + const reader = response.body.getReader(); + const chunks: Uint8Array[] = []; + let length = 0; + try { + while (true) { + const next = await reader.read(); + if (next.done) break; + length += next.value.byteLength; + if (length > 128 * 1024) throw new LiveConnectionError("invalid-response"); + chunks.push(next.value); + } + } finally { + await reader.cancel().catch(() => undefined); + reader.releaseLock(); + } + try { + return JSON.parse(Buffer.concat(chunks).toString("utf8")); + } catch { + throw new LiveConnectionError("invalid-response"); + } + } catch (error) { + throw error instanceof LiveConnectionError ? error : new LiveConnectionError("offline"); + } finally { + clearTimeout(timeout); + signal?.removeEventListener("abort", abort); + } + } +} diff --git a/src/t3-protocol.ts b/src/t3-protocol.ts new file mode 100644 index 0000000..2648b82 --- /dev/null +++ b/src/t3-protocol.ts @@ -0,0 +1,122 @@ +import type { T3ThreadShell } from "./types.js"; + +export type LiveErrorCode = + | "pairing-required" + | "authorization-required" + | "invalid-link" + | "insecure-origin" + | "invalid-response" + | "identity-mismatch" + | "storage-error" + | "offline" + | "connecting" + | "busy"; +export class LiveConnectionError extends Error { + constructor(readonly code: LiveErrorCode) { + super(code); + } +} + +export const MAX_ENVIRONMENTS = 16; +export const MAX_THREADS = 50_000; +export const MAX_RESPONSE_BYTES = 16 * 1024 * 1024; + +export function record(value: unknown): value is Record { + return value !== null && typeof value === "object" && !Array.isArray(value); +} +export function textValue(value: unknown, max = 1024): value is string { + return typeof value === "string" && value.length > 0 && value.length <= max; +} +export function sequence(value: unknown): value is number { + return typeof value === "number" && Number.isSafeInteger(value) && value >= 0; +} +export function originFrom(value: string, allowHttp: boolean): string { + try { + const url = new URL(value); + if (url.username || url.password || !["http:", "https:"].includes(url.protocol)) throw new Error(); + if ( + url.protocol === "http:" && + !["localhost", "127.0.0.1", "[::1]"].includes(url.hostname) && + !allowHttp + ) { + throw new LiveConnectionError("insecure-origin"); + } + return url.origin; + } catch (error) { + if (error instanceof LiveConnectionError) throw error; + throw new LiveConnectionError("invalid-link"); + } +} +export function parsePairingLink(link: string, allowHttp = false): { origin: string; credential: string } { + try { + if (!textValue(link, 16_384)) throw new Error(); + const url = new URL(link.trim()); + if (!["http:", "https:"].includes(url.protocol) || url.username || url.password) throw new Error(); + const credential = new URLSearchParams(url.hash.slice(1)).get("token") || url.searchParams.get("token"); + if (!textValue(credential, 8192)) throw new Error(); + return { origin: originFrom(url.searchParams.get("host") || url.origin, allowHttp), credential }; + } catch (error) { + if (error instanceof LiveConnectionError) throw error; + throw new LiveConnectionError("invalid-link"); + } +} + +// Retain only fields used by the status display, never conversation or file data. +export function parseThread(value: unknown): T3ThreadShell { + if ( + !record(value) || + !textValue(value.id) || + !["hasPendingApprovals", "hasPendingUserInput", "hasActionableProposedPlan"].every( + (key) => typeof value[key] === "boolean", + ) + ) { + throw new LiveConnectionError("invalid-response"); + } + const date = (v: unknown): string | null => { + if (v == null) return null; + if (typeof v !== "string" || !Number.isFinite(Date.parse(v))) + throw new LiveConnectionError("invalid-response"); + return v; + }; + const session = value.session; + const turn = value.latestTurn; + if (session != null && (!record(session) || !textValue(session.status, 64))) + throw new LiveConnectionError("invalid-response"); + if ( + turn != null && + (!record(turn) || !["running", "interrupted", "completed", "error"].includes(String(turn.state))) + ) + throw new LiveConnectionError("invalid-response"); + return { + id: value.id, + interactionMode: value.interactionMode === "plan" ? "plan" : "default", + archivedAt: date(value.archivedAt), + createdAt: date(value.createdAt), + settledAt: date(value.settledAt), + settledOverride: + value.settledOverride === "settled" || value.settledOverride === "active" + ? value.settledOverride + : null, + latestUserMessageAt: date(value.latestUserMessageAt), + snoozedAt: date(value.snoozedAt), + snoozedUntil: date(value.snoozedUntil), + hasPendingApprovals: value.hasPendingApprovals as boolean, + hasPendingUserInput: value.hasPendingUserInput as boolean, + hasActionableProposedPlan: value.hasActionableProposedPlan as boolean, + backgroundLiveness: + value.backgroundLiveness === "working" || value.backgroundLiveness === "monitoring" + ? value.backgroundLiveness + : null, + session: record(session) + ? { status: session.status as string, updatedAt: date(session.updatedAt) ?? undefined } + : null, + latestTurn: record(turn) + ? { + state: turn.state as "running" | "interrupted" | "completed" | "error", + requestedAt: date(turn.requestedAt), + startedAt: date(turn.startedAt), + completedAt: date(turn.completedAt), + } + : null, + }; +} diff --git a/src/types.ts b/src/types.ts index 9af8be6..7d0188a 100644 --- a/src/types.ts +++ b/src/types.ts @@ -5,12 +5,16 @@ export const MIN_REFRESH_SECONDS = 5; export const MAX_REFRESH_SECONDS = 300; export const SETTINGS_VERSION = 1; +export type DisplayMode = "combined" | "threads" | "questions"; + export interface ActionSettings { + displayMode?: DisplayMode; refreshSeconds?: number; settingsVersion?: number; } export interface NormalizedSettings { + displayMode: DisplayMode; refreshSeconds: number; } @@ -88,16 +92,37 @@ export type DashboardModel = | { kind: "loading" } | { kind: "ready"; summary: ThreadSummary; refreshedAt: number } | { kind: "offline" } + | { kind: "pairing" } | { kind: "error" }; +export interface EnvironmentStatus { + environmentId: string; + label: string; + origin: string; + expiresAt: number; + state: "connected" | "connecting" | "offline" | "authorization-required"; + error?: string; +} + export type ConnectionStatus = | { state: "connected"; origin: string; environments: number } - | { state: "offline"; origin?: string }; + | { state: "offline"; origin?: string } + | { + state: "connected" | "connecting" | "offline" | "authorization-required" | "pairing-required" | "error"; + origin?: string; + environments?: number; + connections: EnvironmentStatus[]; + error?: string; + }; export function normalizeSettings(settings: ActionSettings | undefined): NormalizedSettings { const candidate = Number(settings?.refreshSeconds); const refreshSeconds = Number.isFinite(candidate) ? Math.min(MAX_REFRESH_SECONDS, Math.max(MIN_REFRESH_SECONDS, Math.round(candidate))) : DEFAULT_REFRESH_SECONDS; - return { refreshSeconds }; + const displayMode = + settings?.displayMode === "threads" || settings?.displayMode === "questions" + ? settings.displayMode + : "combined"; + return { displayMode, refreshSeconds }; } diff --git a/tests/controller.test.ts b/tests/controller.test.ts index d64653c..c54d7c0 100644 --- a/tests/controller.test.ts +++ b/tests/controller.test.ts @@ -10,6 +10,7 @@ import { type OpenDeckEvent, } from "../src/opendeck.js"; import { T3ClientError, type T3ClientSnapshot } from "../src/t3-client.js"; +import { LiveConnectionError } from "../src/t3-protocol.js"; import { ACTION_UUID, type ConnectionStatus, type ThreadSummary } from "../src/types.js"; interface ImageCall { @@ -87,6 +88,40 @@ class FakeClient implements T3StatusClient { } } +test("live updates reach every display immediately and interval ticks do not poll", async (context) => { + useMockedIntervals(context); + let update = () => {}; + let unsubscribed = false; + const host = new FakeHost(); + const client = new FakeClient() as FakeClient & T3StatusClient; + client.subscribe = (listener) => { + update = listener; + return () => { + unsubscribed = true; + }; + }; + const controller = new T3CodeController(host, client); + controller.handle(event("willAppear", "key-a")); + controller.handle(event("willAppear", "key-b", { settings: { displayMode: "questions" } })); + await flushMicrotasks(); + const reads = client.getSummaryCalls; + context.mock.timers.tick(300_000); + await flushMicrotasks(); + assert.equal(client.getSummaryCalls, reads); + client.onGetSummary = async () => summary({ input: 1 }); + update(); + await flushMicrotasks(); + assert.ok(host.titles.some((call) => call.context === "key-b" && call.title.includes("needs your input"))); + client.onGetSummary = async () => { + throw new LiveConnectionError("authorization-required"); + }; + update(); + await flushMicrotasks(); + assert.equal(host.titles.at(-1)?.title, "Pair T3 Code in the OpenDeck key settings"); + await controller.dispose(); + assert.equal(unsubscribed, true); +}); + function summary(overrides: Partial = {}): ThreadSummary { return { total: 6, @@ -149,6 +184,53 @@ function useMockedIntervals(context: TestContext): void { context.mock.timers.enable({ apis: ["setInterval"] }); } +test("keys blink independently of the refresh ring and stop after answers, mode changes, or removal", async (context) => { + useMockedIntervals(context); + let now = 0; + const host = new FakeHost(); + const client = new FakeClient(); + client.onGetSummary = async () => summary({ input: 1, waiting: 1 }); + const controller = new T3CodeController(host, client, { now: () => now }); + context.after(() => controller.dispose()); + for (const displayMode of ["combined", "threads", "questions"]) { + controller.handle( + event("willAppear", displayMode, { settings: { displayMode, refreshSeconds: 60, settingsVersion: 1 } }), + ); + } + await flushMicrotasks(); + const initial = Object.fromEntries( + ["combined", "threads", "questions"].map((mode) => [mode, latestSvg(host, mode)]), + ); + const titleCount = host.titles.length; + const readCount = client.getSummaryCalls; + now = 500; + context.mock.timers.tick(500); + assert.notEqual(latestSvg(host, "combined"), initial.combined); + assert.notEqual(latestSvg(host, "questions"), initial.questions); + assert.equal(latestSvg(host, "threads"), initial.threads); + assert.equal(host.titles.length, titleCount); + assert.equal(client.getSummaryCalls, readCount); + controller.handle( + event("didReceiveSettings", "combined", { + settings: { displayMode: "threads", refreshSeconds: 60, settingsVersion: 1 }, + }), + ); + assert.equal(latestSvg(host, "combined"), initial.threads); + client.onGetSummary = async () => summary(); + controller.handle(event("keyUp", "questions")); + await flushMicrotasks(); + const cleared = latestSvg(host, "questions"); + assert.match(cleared, /opacity="0.12">\?/); + now = 1_000; + context.mock.timers.tick(500); + assert.equal(latestSvg(host, "questions"), cleared); + for (const key of ["combined", "threads", "questions"]) controller.handle(event("willDisappear", key)); + const imageCount = host.images.length; + now = 10_000; + context.mock.timers.tick(9_000); + assert.equal(host.images.length, imageCount); +}); + test("willAppear renders loading immediately and then the fetched summary", async (context) => { useMockedIntervals(context); const host = new FakeHost(); @@ -189,7 +271,7 @@ test("willAppear migrates the old 15-second default to 60 seconds once", async ( await flushMicrotasks(); assert.deepEqual(host.settings, [ - { context: "key-a", settings: { refreshSeconds: 60, settingsVersion: 1 } }, + { context: "key-a", settings: { displayMode: "combined", refreshSeconds: 60, settingsVersion: 1 } }, ]); now = 15_000; context.mock.timers.tick(15_000); @@ -739,3 +821,34 @@ test("events for another action and events after disposal have no effect", async assert.deepEqual(host.images, []); assert.deepEqual(host.settings, []); }); + +test("concurrent pairing commands get a busy response and never write credentials to key settings", async () => { + const host = new FakeHost(); + const client = new FakeClient() as FakeClient & T3StatusClient; + let finish: () => void = () => {}; + let calls = 0; + client.pair = async () => { + calls++; + await new Promise((resolve) => { + finish = resolve; + }); + }; + const controller = new T3CodeController(host, client); + controller.handle( + event("sendToPlugin", "key-a", { command: "pair", link: "http://localhost/pair#token=fixture" }), + ); + controller.handle( + event("sendToPlugin", "key-b", { command: "pair", link: "http://localhost/pair#token=fixture-two" }), + ); + assert.equal(calls, 1); + assert.ok( + host.inspectorMessages.some( + (message) => message.context === "key-b" && JSON.stringify(message.payload).includes('"error":"busy"'), + ), + ); + finish(); + await flushMicrotasks(); + assert.doesNotMatch(JSON.stringify(host.settings), /fixture|token/); + assert.doesNotMatch(JSON.stringify(host.inspectorMessages), /fixture/); + await controller.dispose(); +}); diff --git a/tests/property-inspector.test.ts b/tests/property-inspector.test.ts index 4653107..30dda83 100644 --- a/tests/property-inspector.test.ts +++ b/tests/property-inspector.test.ts @@ -14,6 +14,7 @@ const inspectorHtml = await readFile( interface FakeElement { addEventListener(name: string, listener: () => void): void; + emit(name: string): void; classList: { toggle(name: string, enabled: boolean): void }; dataset: Record; disabled: boolean; @@ -40,16 +41,29 @@ function loadInspector() { "connection-state", "connection-detail", "error-message", - "refresh-seconds", - "refresh-note", + "display-mode", + "display-note", + "pairing-link", + "allow-http", + "pair-button", + "pairing-result", + "environment-list", + "environment-detail", + "remove-connection", ]) { let hidden = true; let textContent = ""; + const listeners = new Map void>(); const element: FakeElement = { - addEventListener: () => undefined, + addEventListener: (name, listener) => { + listeners.set(name, listener); + }, + emit: (name) => { + listeners.get(name)?.(); + }, classList: { toggle: () => undefined }, dataset: {}, - disabled: id === "refresh-seconds", + disabled: id === "display-mode", get hidden() { return hidden; }, @@ -66,7 +80,7 @@ function loadInspector() { this.textContentWrites += 1; }, textContentWrites: 0, - value: id === "refresh-seconds" ? "60" : "", + value: id === "display-mode" ? "60" : "", }; elements.set(id, element); } @@ -139,6 +153,37 @@ function loadInspector() { }; } +test("display changes restore saved modes", () => { + const { connect, elements, socket } = loadInspector(); + connect( + 1234, + "pi-context", + "registerPropertyInspector", + "{}", + JSON.stringify({ + context: "key-a", + payload: { settings: { displayMode: "questions", refreshSeconds: 90 } }, + }), + ); + const connection = socket(); + assert.ok(connection); + const mode = elements.get("display-mode"); + assert.ok(mode); + assert.equal(mode.disabled, true); + assert.equal(mode.value, "questions"); + connection.open(); + assert.equal(mode.disabled, false); + mode.value = "threads"; + mode.emit("change"); + assert.deepEqual(JSON.parse(connection.sent.at(-1) ?? ""), { + event: "setSettings", + context: "key-a", + payload: { displayMode: "threads", settingsVersion: 1 }, + }); + connection.emit("close"); + assert.equal(mode.disabled, true); +}); + test("the property inspector reports malformed or invalid action data", () => { for (const actionInfo of ["{", "null", "[]", '"text"', "42"]) { const { connect, elements } = loadInspector(); @@ -151,26 +196,44 @@ test("the property inspector reports malformed or invalid action data", () => { assert.equal(elements.get("error-message")?.textContent, "OpenDeck sent invalid action data."); assert.equal(elements.get("error-message")?.hidden, false); assert.equal(elements.get("status-card")?.dataset.state, "error"); - assert.equal(elements.get("refresh-seconds")?.disabled, true); + assert.equal(elements.get("display-mode")?.disabled, true); } }); -test("the refresh interval stays disabled until the OpenDeck socket is ready", () => { - assert.match(inspectorHtml, /id="refresh-seconds"[\s\S]*?disabled[\s\S]*?aria-label=/); +test("pairing sends a transient command, clears the secret, and keeps it out of key settings", () => { const { connect, elements, socket } = loadInspector(); - connect( - 1234, - "pi-context", - "registerPropertyInspector", - "{}", - JSON.stringify({ context: "action-context", payload: { settings: { refreshSeconds: 90 } } }), - ); - assert.equal(elements.get("refresh-seconds")?.disabled, true); - assert.equal(elements.get("refresh-seconds")?.value, "90"); + connect(1234, "pi-context", "registerPropertyInspector", "{}", JSON.stringify({ context: "key-a" })); const connection = socket(); assert.ok(connection); connection.open(); - assert.equal(elements.get("refresh-seconds")?.disabled, false); + const link = elements.get("pairing-link"); + const button = elements.get("pair-button"); + assert.ok(link); + assert.ok(button); + link.value = "http://127.0.0.1:3773/pair#token=fixture"; + button.emit("click"); + const message = JSON.parse(connection.sent.at(-1) ?? ""); + assert.equal(message.event, "sendToPlugin"); + assert.equal(message.payload.command, "pair"); + assert.equal(message.payload.link, "http://127.0.0.1:3773/pair#token=fixture"); + assert.equal(link.value, ""); + assert.equal(button.disabled, true); + connection.emit("message", { + data: JSON.stringify({ + event: "sendToPropertyInspector", + payload: { type: "pairingResult", error: "invalid-link" }, + }), + }); + assert.equal(button.disabled, false); + assert.equal( + elements.get("pairing-result")?.textContent, + "Paste the complete pairing link, including its token.", + ); + const mode = elements.get("display-mode"); + assert.ok(mode); + mode.value = "questions"; + mode.emit("change"); + assert.doesNotMatch(connection.sent.at(-1) ?? "", /fixture|token|pairing/); }); test("connection updates use one atomic live region for status, detail, and recovery guidance", () => { @@ -264,25 +327,28 @@ test("visible connection changes still update the live region", () => { type: "connectionStatus", }); assert.equal(state.textContent, "Connected"); - assert.equal(detail.textContent, "3 Environments · local cache · no sign-in"); + assert.equal(detail.textContent, "3 Environments · live stream · read only"); assert.equal(state.textContentWrites + detail.textContentWrites, connectedWrites + 1); sendStatus({ - error: "cache-unavailable", - status: { environments: 0, state: "offline" }, + error: "authorization-required", + status: { environments: 0, state: "authorization-required" }, type: "connectionStatus", }); - assert.equal(state.textContent, "Cache unavailable"); - assert.equal(detail.textContent, "Thread status cannot update until the local cache is available."); - assert.equal(error.textContent, "The local T3 thread cache could not be opened."); - assert.equal(error.hidden, false); + assert.equal(state.textContent, "Pairing needed"); + assert.equal( + detail.textContent, + "Authorization expired or was revoked. Paste a fresh read-only pairing link.", + ); + assert.equal(error.textContent, ""); + assert.equal(error.hidden, true); sendStatus({ status: { environments: 3, state: "connected" }, type: "connectionStatus", }); assert.equal(state.textContent, "Connected"); - assert.equal(detail.textContent, "3 Environments · local cache · no sign-in"); + assert.equal(detail.textContent, "3 Environments · live stream · read only"); assert.equal(error.textContent, ""); assert.equal(error.hidden, true); }); @@ -300,13 +366,13 @@ test("websocket errors and normal closes disable settings without duplicate reco const connection = inspector.socket(); assert.ok(connection); connection.open(); - assert.equal(inspector.elements.get("refresh-seconds")?.disabled, false); + assert.equal(inspector.elements.get("display-mode")?.disabled, false); return { ...inspector, connection }; }; const failed = connectInspector(); failed.connection.emit("error"); - assert.equal(failed.elements.get("refresh-seconds")?.disabled, true); + assert.equal(failed.elements.get("display-mode")?.disabled, true); assert.equal(failed.elements.get("connection-state")?.textContent, "OpenDeck disconnected"); assert.equal( failed.elements.get("connection-detail")?.textContent, @@ -325,7 +391,7 @@ test("websocket errors and normal closes disable settings without duplicate reco const closed = connectInspector(); closed.connection.emit("close"); - assert.equal(closed.elements.get("refresh-seconds")?.disabled, true); + assert.equal(closed.elements.get("display-mode")?.disabled, true); assert.equal(closed.elements.get("connection-state")?.textContent, "OpenDeck disconnected"); assert.equal(closed.elements.get("error-message")?.textContent, "The connection to OpenDeck was closed."); }); @@ -337,7 +403,7 @@ test("the property inspector rejects oversized startup data before opening a soc assert.equal(socket(), undefined); assert.equal(elements.get("connection-state")?.textContent, "Settings unavailable"); assert.equal(elements.get("error-message")?.textContent, "OpenDeck sent invalid or oversized action data."); - assert.equal(elements.get("refresh-seconds")?.disabled, true); + assert.equal(elements.get("display-mode")?.disabled, true); }); test("malformed bounded messages are ignored and a later valid message still renders", () => { @@ -366,7 +432,7 @@ test("malformed bounded messages are ignored and a later valid message still ren }); assert.equal(elements.get("connection-state")?.textContent, "Connected"); - assert.equal(elements.get("connection-detail")?.textContent, "2 Environments · local cache · no sign-in"); + assert.equal(elements.get("connection-detail")?.textContent, "2 Environments · live stream · read only"); assert.deepEqual(connection.closeCalls, []); }); @@ -383,12 +449,12 @@ test("an oversized websocket message closes the connection with a bounded local elements.get("error-message")?.textContent, "OpenDeck sent an oversized or unsupported settings message.", ); - assert.equal(elements.get("refresh-seconds")?.disabled, true); + assert.equal(elements.get("display-mode")?.disabled, true); connection.emit("message", { data: JSON.stringify({ event: "didReceiveSettings", payload: { settings: { refreshSeconds: 90 } } }), }); - assert.equal(elements.get("refresh-seconds")?.value, "60"); + assert.equal(elements.get("display-mode")?.value, "combined"); }); test("a websocket message flood is cut off before unbounded parsing work accumulates", () => { @@ -402,5 +468,5 @@ test("a websocket message flood is cut off before unbounded parsing work accumul assert.deepEqual(connection.closeCalls, [{ code: 1009, reason: "Settings input limit exceeded" }]); assert.equal(elements.get("error-message")?.textContent, "OpenDeck sent settings messages too quickly."); - assert.equal(elements.get("refresh-seconds")?.disabled, true); + assert.equal(elements.get("display-mode")?.disabled, true); }); diff --git a/tests/render.test.ts b/tests/render.test.ts index 7155088..77f828d 100644 --- a/tests/render.test.ts +++ b/tests/render.test.ts @@ -1,7 +1,7 @@ import assert from "node:assert/strict"; import { describe, test } from "node:test"; -import { getAccessibleTitle, getDisplay, renderDashboard } from "../src/render.js"; +import { getAccessibleTitle, getDisplay, hasOpenQuestions, renderDashboard } from "../src/render.js"; import type { DashboardModel, ThreadSummary } from "../src/types.js"; function summary(overrides: Partial = {}): ThreadSummary { @@ -165,6 +165,58 @@ function decodeSvg(dataUrl: string): string { return Buffer.from(dataUrl.slice(prefix.length), "base64").toString("utf8"); } +test("question alerts cover input, approvals and plans but exclude waiting and failures", () => { + for (const state of ["input", "approval", "plan", "waiting", "failed", "monitoring"] as const) { + const model: DashboardModel = { + kind: "ready", + refreshedAt: 0, + summary: summary({ total: 1, [state]: 1 }), + }; + const alerts = ["input", "approval", "plan"].includes(state); + assert.equal(hasOpenQuestions(model), alerts); + const bright = decodeSvg(renderDashboard(model, 0, "questions", true)); + const dim = decodeSvg(renderDashboard(model, 0, "questions", false)); + assert.equal(bright !== dim, alerts); + if (alerts) { + assert.match(bright, /]*fill="#FFBE55"/); + assert.match(dim, /]*fill="#071014"/); + assert.match(bright, />1 NEEDS YOU<\/text>/); + } else { + assert.match(bright, /opacity="0.12">\?<\/text>/); + assert.match(bright, />NO QUESTIONS<\/text>/); + } + } +}); + +test("combined alerts retain thread counts and thread-only mode never blinks", () => { + const model: DashboardModel = { + kind: "ready", + refreshedAt: 0, + summary: summary({ total: 6, running: 4, input: 1, failed: 1 }), + }; + const combined = decodeSvg(renderDashboard(model, 0, "combined", true)); + assert.match(combined, />\?<\/text>/); + assert.match(combined, />4\/6<\/text>/); + assert.match(combined, />1 NEEDS YOU<\/text>/); + assert.notEqual(combined, decodeSvg(renderDashboard(model, 0, "combined", false))); + const threads = renderDashboard(model, 0, "threads", true); + assert.equal(threads, renderDashboard(model, 0, "threads", false)); + assert.doesNotMatch(decodeSvg(threads), />\?<\/text>|NEEDS YOU|ERROR/); + assert.match(getAccessibleTitle(model, "combined"), /1 error, 1 need your attention/); + assert.equal(getAccessibleTitle(model, "threads"), "4 of 6 threads working"); + assert.match(getAccessibleTitle(model, "questions"), /1 thread needs your input/); +}); + +test("every mode shows connection failures instead of a question or a false all-clear", () => { + for (const kind of ["loading", "offline", "error"] as const) { + for (const mode of ["combined", "threads", "questions"] as const) { + assert.equal(renderDashboard({ kind }, 0, mode, true), renderDashboard({ kind }, 0, mode, false)); + assert.doesNotMatch(decodeSvg(renderDashboard({ kind }, 0, mode)), />\?<\/text>|NO QUESTIONS/); + assert.equal(getAccessibleTitle({ kind }, mode), getAccessibleTitle({ kind })); + } + } +}); + test("renderDashboard returns a self-contained 144-pixel SVG", () => { const model: DashboardModel = { kind: "ready", diff --git a/tests/status.test.ts b/tests/status.test.ts index e8d6203..2cb2447 100644 --- a/tests/status.test.ts +++ b/tests/status.test.ts @@ -449,28 +449,42 @@ test("emptySummary returns a fresh all-zero summary", () => { describe("normalizeSettings", () => { test("uses the default when the setting is absent or not finite", () => { - assert.deepEqual(normalizeSettings(undefined), { refreshSeconds: DEFAULT_REFRESH_SECONDS }); - assert.deepEqual(normalizeSettings({}), { refreshSeconds: DEFAULT_REFRESH_SECONDS }); + assert.deepEqual(normalizeSettings(undefined), { + displayMode: "combined", + refreshSeconds: DEFAULT_REFRESH_SECONDS, + }); + assert.deepEqual(normalizeSettings({}), { + displayMode: "combined", + refreshSeconds: DEFAULT_REFRESH_SECONDS, + }); assert.deepEqual(normalizeSettings({ refreshSeconds: Number.NaN }), { + displayMode: "combined", refreshSeconds: DEFAULT_REFRESH_SECONDS, }); assert.deepEqual(normalizeSettings({ refreshSeconds: Number.POSITIVE_INFINITY }), { + displayMode: "combined", refreshSeconds: DEFAULT_REFRESH_SECONDS, }); }); test("rounds values and clamps them to the supported interval", () => { - assert.deepEqual(normalizeSettings({ refreshSeconds: 21.6 }), { refreshSeconds: 22 }); + assert.deepEqual(normalizeSettings({ refreshSeconds: 21.6 }), { + displayMode: "combined", + refreshSeconds: 22, + }); assert.deepEqual(normalizeSettings({ refreshSeconds: MIN_REFRESH_SECONDS - 1 }), { + displayMode: "combined", refreshSeconds: MIN_REFRESH_SECONDS, }); assert.deepEqual(normalizeSettings({ refreshSeconds: MAX_REFRESH_SECONDS + 1 }), { + displayMode: "combined", refreshSeconds: MAX_REFRESH_SECONDS, }); }); test("accepts the numeric strings emitted by a property inspector", () => { assert.deepEqual(normalizeSettings({ refreshSeconds: "45" } as unknown as { refreshSeconds: number }), { + displayMode: "combined", refreshSeconds: 45, }); }); diff --git a/tests/t3-live-client.test.ts b/tests/t3-live-client.test.ts new file mode 100644 index 0000000..4450bd2 --- /dev/null +++ b/tests/t3-live-client.test.ts @@ -0,0 +1,427 @@ +import assert from "node:assert/strict"; +import { once } from "node:events"; +import { mkdtemp, readFile, rm, stat } from "node:fs/promises"; +import { createServer } from "node:http"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { type TestContext, test } from "node:test"; +import { type WebSocket, WebSocketServer } from "ws"; +import { type ConnectionStore, FileConnectionStore, type SavedConnection } from "../src/connection-store.js"; +import { T3LiveClient } from "../src/t3-live-client.js"; +import { parsePairingLink } from "../src/t3-protocol.js"; + +class MemoryStore implements ConnectionStore { + values: SavedConnection[] = []; + async load() { + return structuredClone(this.values); + } + async save(values: SavedConnection[]) { + this.values = structuredClone(values); + } +} +const thread = (id = "thread-1", extra = {}) => ({ + id, + interactionMode: "default", + archivedAt: null, + settledOverride: "active", + settledAt: null, + hasPendingApprovals: false, + hasPendingUserInput: false, + hasActionableProposedPlan: false, + latestTurn: null, + session: { status: "running" }, + ...extra, +}); + +async function fixture(context: TestContext, marker = true) { + const requests: Array<{ path: string; authorization?: string; body: string }> = []; + const subscriptions: Array> = []; + const sockets: WebSocket[] = []; + let rejected = false; + let environmentId = "env-1"; + let expiresIn = 30 * 86400; + let grantedScope = "orchestration:read"; + let descriptorCapability = marker; + const server = createServer(async (request, response) => { + let body = ""; + for await (const chunk of request) body += chunk; + requests.push({ path: request.url ?? "", authorization: request.headers.authorization, body }); + response.setHeader("content-type", "application/json"); + if (request.url === "/.well-known/t3/environment") + response.end( + JSON.stringify({ + environmentId, + label: "Test environment", + capabilities: { shellResumeCompletionMarker: descriptorCapability }, + }), + ); + else if (rejected) { + response.statusCode = 401; + response.end("{}"); + } else if (request.url === "/oauth/token") + response.end( + JSON.stringify({ + access_token: "fixture-bearer", + token_type: "Bearer", + scope: grantedScope, + expires_in: expiresIn, + }), + ); + else if (request.url === "/api/auth/session") + response.end(JSON.stringify({ authenticated: true, scopes: [grantedScope] })); + else if (request.url === "/api/auth/websocket-ticket") + response.end(JSON.stringify({ ticket: "fixture-ticket" })); + else { + response.statusCode = 404; + response.end("{}"); + } + }); + const ws = new WebSocketServer({ server }); + let onSubscribe = (socket: WebSocket, payload: Record) => { + if (payload.afterSequence === undefined) + send(socket, [ + { + kind: "snapshot", + snapshot: { snapshotSequence: 10, threads: [thread()], updatedAt: new Date().toISOString() }, + }, + ]); + if (descriptorCapability) send(socket, [{ kind: "synchronized" }]); + }; + ws.on("connection", (socket, request) => { + assert.equal( + new URL(request.url ?? "", "http://localhost").searchParams.get("wsTicket"), + "fixture-ticket", + ); + assert.equal(request.headers.authorization, undefined); + sockets.push(socket); + socket.on("message", (data) => { + const message = JSON.parse(data.toString()); + if (message._tag === "Ping") socket.send(JSON.stringify({ _tag: "Pong" })); + if (message._tag === "Request") { + assert.equal(message.tag, "orchestration.subscribeShell"); + subscriptions.push(message.payload); + onSubscribe(socket, message.payload); + } + }); + }); + server.listen(0, "127.0.0.1"); + await once(server, "listening"); + const address = server.address(); + assert.ok(address && typeof address !== "string"); + const origin = `http://127.0.0.1:${address.port}`; + context.after(async () => { + for (const socket of ws.clients) socket.terminate(); + ws.close(); + server.closeAllConnections(); + await new Promise((resolve) => server.close(() => resolve())); + }); + return { + origin, + link: `${origin}/pair#token=fixture-pairing`, + requests, + subscriptions, + sockets, + setRejected: (value: boolean) => { + rejected = value; + }, + setIdentity: (value: string) => { + environmentId = value; + }, + setExpiry: (value: number) => { + expiresIn = value; + }, + setScope: (value: string) => { + grantedScope = value; + }, + setMarker: (value: boolean) => { + descriptorCapability = value; + }, + onSubscribe: (callback: typeof onSubscribe) => { + onSubscribe = callback; + }, + }; +} +function send(socket: WebSocket, values: unknown[]) { + socket.send(JSON.stringify({ _tag: "Chunk", requestId: "shell", values })); +} +async function until(client: T3LiveClient, predicate: () => Promise) { + await new Promise((resolve, reject) => { + const timer = setTimeout(() => { + unsubscribe(); + reject(new Error("Expected live state was not reached")); + }, 2000); + const check = () => { + void predicate().then((ok) => { + if (ok) { + clearTimeout(timer); + unsubscribe(); + resolve(); + } + }, reject); + }; + const unsubscribe = client.subscribe(check); + check(); + }); +} +async function connected(client: T3LiveClient) { + await until(client, async () => (await client.getConnectionStatus()).state === "connected"); +} + +test("pairing exchanges only read scope, streams input changes and removals without polling", async (context) => { + const server = await fixture(context); + const store = new MemoryStore(); + const client = new T3LiveClient({ store, retryMs: 10 }); + context.after(() => client.dispose()); + assert.equal((await client.getConnectionStatus()).state, "pairing-required"); + await client.pair(server.link); + await connected(client); + assert.equal((await client.getSnapshot()).summary.running, 1); + const exchange = server.requests.find((item) => item.path === "/oauth/token"); + assert.ok(exchange); + assert.equal(new URLSearchParams(exchange.body).get("scope"), "orchestration:read"); + assert.equal(new URLSearchParams(exchange.body).get("subject_token"), "fixture-pairing"); + assert.equal(store.values.length, 1); + assert.doesNotMatch(JSON.stringify(await client.getConnectionStatus()), /fixture-bearer|fixture-pairing/); + const count = server.requests.length; + await Promise.all(Array.from({ length: 20 }, () => client.getSnapshot())); + assert.equal(server.requests.length, count); + send(firstSocket(server.sockets), [ + { kind: "thread-upserted", sequence: 12, thread: thread("thread-1", { hasPendingUserInput: true }) }, + ]); + await until(client, async () => (await client.getSnapshot()).summary.input === 1); + send(firstSocket(server.sockets), [{ kind: "thread-removed", sequence: 13, threadId: "thread-1" }]); + await until(client, async () => (await client.getSnapshot()).summary.total === 0); +}); + +test("reconnect requests a fresh ticket and resumes after the last applied sequence", async (context) => { + const server = await fixture(context); + const client = new T3LiveClient({ store: new MemoryStore(), retryMs: 10 }); + context.after(() => client.dispose()); + await client.pair(server.link); + await connected(client); + send(firstSocket(server.sockets), [ + { kind: "thread-upserted", sequence: 19, thread: thread("thread-1", { hasPendingUserInput: true }) }, + ]); + await until(client, async () => (await client.getSnapshot()).summary.input === 1); + server.onSubscribe((socket, payload) => { + assert.equal(payload.afterSequence, 19); + send(socket, [ + { kind: "thread-upserted", sequence: 19, thread: thread() }, + { kind: "thread-upserted", sequence: 21, thread: thread("thread-2") }, + { kind: "synchronized" }, + ]); + }); + firstSocket(server.sockets).terminate(); + await until( + client, + async () => + server.subscriptions.length === 2 && (await client.getConnectionStatus()).state === "connected", + ); + assert.equal((await client.getSnapshot()).summary.input, 1); + assert.equal((await client.getSnapshot()).summary.total, 2); + assert.equal(server.requests.filter((item) => item.path === "/oauth/token").length, 1); + assert.equal(server.requests.filter((item) => item.path === "/api/auth/websocket-ticket").length, 2); +}); + +test("expired sessions stop automatic retries; re-pair replaces credentials without duplicating environments", async (context) => { + const server = await fixture(context); + const store = new MemoryStore(); + let now = Date.now(); + const client = new T3LiveClient({ store, now: () => now, retryMs: 10 }); + context.after(() => client.dispose()); + await client.pair(server.link); + await connected(client); + now += 31 * 86400_000; + firstSocket(server.sockets).terminate(); + await until(client, async () => (await client.getConnectionStatus()).state === "authorization-required"); + await assert.rejects(client.getSnapshot(), { code: "authorization-required" }); + const requestCount = server.requests.length; + await client.reconnect(); + assert.equal(server.requests.length, requestCount); + await client.pair(server.link); + await connected(client); + assert.equal(store.values.length, 1); + assert.equal((await client.getSnapshot()).summary.running, 1); + await client.remove("env-1"); + assert.equal(store.values.length, 0); + assert.equal((await client.getConnectionStatus()).state, "pairing-required"); +}); + +test("old servers receive no resume options; replacement snapshots recover from stale cursors", async (context) => { + const server = await fixture(context, false); + const client = new T3LiveClient({ store: new MemoryStore(), retryMs: 10 }); + context.after(() => client.dispose()); + await client.pair(server.link); + await connected(client); + assert.deepEqual(server.subscriptions[0], {}); + server.setMarker(true); + server.onSubscribe((socket, payload) => { + assert.equal(payload.afterSequence, 10); + send(socket, [ + { + kind: "snapshot", + snapshot: { snapshotSequence: 1000, threads: [], updatedAt: new Date().toISOString() }, + }, + { kind: "synchronized" }, + ]); + }); + firstSocket(server.sockets).terminate(); + await until( + client, + async () => + server.subscriptions.length === 2 && (await client.getConnectionStatus()).state === "connected", + ); + assert.equal((await client.getSnapshot()).summary.total, 0); +}); + +test("identity mismatch never sends bearer credentials and revoked sessions require pairing", async (context) => { + const server = await fixture(context); + const store = new MemoryStore(); + const client = new T3LiveClient({ store, retryMs: 10 }); + context.after(() => client.dispose()); + await client.pair(server.link); + await connected(client); + const count = server.requests.length; + server.setIdentity("different"); + firstSocket(server.sockets).terminate(); + await until(client, async () => { + const status = await client.getConnectionStatus(); + return "connections" in status && status.connections[0]?.error === "identity-mismatch"; + }); + assert.ok(server.requests.slice(count).every((item) => !item.authorization)); + server.setIdentity("env-1"); + server.setRejected(true); + await client.reconnect(); + await until(client, async () => (await client.getConnectionStatus()).state === "authorization-required"); +}); + +test("restore credentials across plugin restarts and reject excessive scopes without replacing existing connection", async (context) => { + const server = await fixture(context); + const store = new MemoryStore(); + const first = new T3LiveClient({ store }); + await first.pair(server.link); + await connected(first); + await first.dispose(); + const client = new T3LiveClient({ store }); + context.after(() => client.dispose()); + await connected(client); + assert.equal(server.requests.filter((item) => item.path === "/oauth/token").length, 1); + server.setScope("orchestration:read orchestration:operate"); + await assert.rejects(client.pair(server.link), { code: "invalid-response" }); + assert.equal((await client.getConnectionStatus()).state, "connected"); + assert.equal(store.values.length, 1); +}); + +test("pairing links support hosted links, reject credentials in authority, and require opt-in for LAN HTTP", () => { + assert.deepEqual( + parsePairingLink("https://app.t3.codes/pair?host=https%3A%2F%2Fhost.example#token=fixture"), + { origin: "https://host.example", credential: "fixture" }, + ); + assert.throws(() => parsePairingLink("http://192.168.1.2/pair#token=fixture"), { code: "insecure-origin" }); + assert.equal(parsePairingLink("http://192.168.1.2/pair#token=fixture", true).origin, "http://192.168.1.2"); + for (const link of [ + "oops", + "https://user:pass@example.com/pair#token=fixture", + "file:///tmp/test#token=fixture", + "https://example.com/pair", + ]) + assert.throws(() => parsePairingLink(link), { code: "invalid-link" }); +}); + +test("credentials persist with private permissions, and corrupt stores fail closed", async (context) => { + const folder = await mkdtemp(join(tmpdir(), "opendeck-stream-store-")); + context.after(() => rm(folder, { recursive: true, force: true })); + const path = join(folder, "private", "connections.json"); + const store = new FileConnectionStore(path); + assert.deepEqual(await store.load(), []); + const values: SavedConnection[] = [ + { + environmentId: "env", + label: "Test", + origin: "http://127.0.0.1:3773", + token: "fixture", + expiresAt: Date.now() + 1000, + allowHttp: false, + }, + ]; + await store.save(values); + assert.deepEqual(await store.load(), values); + assert.ok((await readFile(path, "utf8")).includes("fixture")); + if (process.platform !== "win32") assert.equal((await stat(path)).mode & 0o777, 0o600); + await store.save([...values, ...values]); + await assert.rejects(store.load(), { code: "storage-error" }); +}); + +function firstSocket(sockets: WebSocket[]): WebSocket { + const socket = sockets[0]; + assert.ok(socket); + return socket; +} + +test("multiple environments aggregate independently and offline environments never produce partial live totals", async (context) => { + const a = await fixture(context); + const b = await fixture(context); + b.setIdentity("env-2"); + const store = new MemoryStore(); + const client = new T3LiveClient({ store, retryMs: 10 }); + context.after(() => client.dispose()); + await client.pair(a.link); + await connected(client); + await client.pair(b.link); + await connected(client); + assert.equal((await client.getSnapshot()).summary.total, 2); + assert.equal(a.subscriptions.length, 1); + assert.equal(b.subscriptions.length, 1); + b.setRejected(true); + firstSocket(b.sockets).terminate(); + await until(client, async () => (await client.getConnectionStatus()).state === "authorization-required"); + await assert.rejects(client.getSnapshot(), { code: "authorization-required" }); + await client.remove("env-2"); + await connected(client); + assert.equal((await client.getSnapshot()).summary.total, 1); + assert.equal(a.subscriptions.length, 1); +}); + +test("malformed stream data drops stale state and reconnects with a complete snapshot", async (context) => { + const server = await fixture(context); + const client = new T3LiveClient({ store: new MemoryStore(), retryMs: 10 }); + context.after(() => client.dispose()); + await client.pair(server.link); + await connected(client); + server.onSubscribe((socket, payload) => { + assert.equal(payload.afterSequence, undefined); + send(socket, [ + { + kind: "snapshot", + snapshot: { snapshotSequence: 22, threads: [], updatedAt: new Date().toISOString() }, + }, + { kind: "synchronized" }, + ]); + }); + send(firstSocket(server.sockets), [{ kind: "thread-upserted", sequence: 12, thread: { id: "bad" } }]); + await until( + client, + async () => + server.subscriptions.length === 2 && (await client.getConnectionStatus()).state === "connected", + ); + assert.equal((await client.getSnapshot()).summary.total, 0); +}); + +test("HTTP redirects are refused before credentials are sent to another origin", async (context) => { + const server = createServer((_request, response) => { + response.writeHead(302, { location: "http://127.0.0.1:1/secret-destination" }); + response.end(); + }); + server.listen(0, "127.0.0.1"); + await once(server, "listening"); + context.after(() => new Promise((resolve) => server.close(() => resolve()))); + const address = server.address(); + assert.ok(address && typeof address !== "string"); + const store = new MemoryStore(); + const client = new T3LiveClient({ store }); + context.after(() => client.dispose()); + await assert.rejects(client.pair(`http://127.0.0.1:${address.port}/pair#token=fixture`), { + code: "offline", + }); + assert.equal(store.values.length, 0); +}); From a3a13c5072478b3f884b6aa095ec25d2a8984471 Mon Sep 17 00:00:00 2001 From: beasty Date: Thu, 10 Sep 2026 00:16:55 +0200 Subject: [PATCH 2/3] Use fresh snapshots on reconnect and clarify authorization expiry --- README.md | 4 +- plugin/property-inspector/index.html | 2 +- src/t3-live-client.ts | 23 +++-------- tests/t3-live-client.test.ts | 62 ++++++++++++++++------------ 4 files changed, 45 insertions(+), 46 deletions(-) diff --git a/README.md b/README.md index 1d6e312..f44414b 100644 --- a/README.md +++ b/README.md @@ -26,7 +26,7 @@ For two separate keys, drag `Thread status` onto two keys and select **Threads o Questions appear or clear when the stream delivers an update. There is no HTTP polling interval. Pressing a key does not answer or dismiss a question. -Transient connection failures retry with backoff up to 30 seconds. If any paired environment is disconnected, the key shows the connection state instead of presenting incomplete totals as live. Expired or revoked authorization shows **LINK**: create a fresh read-only pairing link and paste it into the same settings panel. Re-pairing the same environment preserves your keys and replaces its credential. Current T3 bearer sessions expire after about 30 days; automatic authorization renewal is not available through this pairing flow. +Transient connection failures retry with backoff up to 30 seconds. If any paired environment is disconnected, the key shows the connection state instead of presenting incomplete totals as live. When expired or revoked authorization prevents a connection, the key shows **LINK**: create a fresh read-only pairing link and paste it into the same settings panel. Re-pairing the same environment preserves your keys and replaces its credential. Current T3 bearer sessions expire after about 30 days; automatic authorization renewal is not available through this pairing flow. An existing stream may remain connected beyond that date because T3 checks authorization when the connection opens. Its next reconnect requires a fresh pairing link. The normal plugin package declares these OpenDeck platforms: @@ -120,7 +120,7 @@ Credentials are stored separately from OpenDeck profiles and the plugin installa **Forget selected environment** removes its saved credential from this plugin. Revoke the corresponding session in **T3 Settings → Connections** to remove its server authorization too. -On reconnect the plugin checks the server's environment identity, gets a fresh ticket, and resumes from the last applied sequence. Servers without the optional completion marker receive a fresh-snapshot subscription instead. Stream state stays in memory; plugin restarts load a new snapshot. Messages and retained thread counts are bounded. +On reconnect the plugin checks the server's environment identity, gets a fresh ticket, and loads a fresh shell snapshot before continuing live updates. This also works with servers that lack optional replay-completion markers. Stream state stays in memory. Messages and retained thread counts are bounded. This uses T3's existing first-party interface, which is not a versioned public SDK. Protocol reference: T3 Code commit `e16b8b059c9f5ff6dfed1addecffb831c6aee043`, following [the maintainer's guidance](https://github.com/pingdotgg/t3code/issues/10929#issuecomment-5601670045). Future T3 changes may require a plugin update. diff --git a/plugin/property-inspector/index.html b/plugin/property-inspector/index.html index d62241b..0da1b2a 100644 --- a/plugin/property-inspector/index.html +++ b/plugin/property-inspector/index.html @@ -46,7 +46,7 @@

Thread status

-

Pair once per environment for all keys. A new link for the same environment replaces its saved authorization. T3 currently requires a new link after the 30-day session expires.

+

Pair once per environment for all keys. A new link for the same environment replaces its saved authorization. After T3's 30-day authorization expires, reconnecting requires a fresh link. An existing stream may stay connected longer.

Read only also permits access to files and conversations. This plugin requests only the thread status stream.

diff --git a/src/t3-live-client.ts b/src/t3-live-client.ts index 96aa6a2..2ccb023 100644 --- a/src/t3-live-client.ts +++ b/src/t3-live-client.ts @@ -300,8 +300,6 @@ export class T3LiveClient { if (!current()) return; if (!record(descriptor) || descriptor.environmentId !== env.saved.environmentId) throw new LiveConnectionError("identity-mismatch"); - const marker = - record(descriptor.capabilities) && descriptor.capabilities.shellResumeCompletionMarker === true; const ticket = await this.request( env.saved.origin, "/api/auth/websocket-ticket", @@ -339,19 +337,15 @@ export class T3LiveClient { socket.terminate(); return; } - const payload = marker - ? { - ...(env.sequence === undefined ? {} : { afterSequence: env.sequence }), - requestCompletionMarker: true, - } - : {}; - // Without a completion marker, request a full snapshot so stale state is never shown as live. + // A fresh snapshot on every connection avoids depending on optional replay markers. + env.sequence = undefined; + env.threads.clear(); socket.send( JSON.stringify({ _tag: "Request", id: "shell", tag: "orchestration.subscribeShell", - payload, + payload: {}, headers: [], }), ); @@ -384,7 +378,7 @@ export class T3LiveClient { } if (message._tag !== "Chunk" || message.requestId !== "shell" || !Array.isArray(message.values)) throw new Error(); - for (const item of message.values) this.apply(env, item, marker); + for (const item of message.values) this.apply(env, item); if (socket.readyState === WebSocket.OPEN) socket.send(JSON.stringify({ _tag: "Ack", requestId: "shell" })); } @@ -399,7 +393,7 @@ export class T3LiveClient { if (current()) this.fail(env, error instanceof LiveConnectionError ? error.code : "offline"); } } - private apply(env: Environment, item: unknown, marker: boolean): void { + private apply(env: Environment, item: unknown): void { if (!record(item)) throw new Error(); if (item.kind === "snapshot") { const snapshot = item.snapshot; @@ -418,11 +412,6 @@ export class T3LiveClient { } env.threads = threads; env.sequence = snapshot.snapshotSequence; - if (!marker) this.synchronized(env); - return; - } - if (item.kind === "synchronized") { - if (env.sequence === undefined) throw new Error(); this.synchronized(env); return; } diff --git a/tests/t3-live-client.test.ts b/tests/t3-live-client.test.ts index 4450bd2..489cd2f 100644 --- a/tests/t3-live-client.test.ts +++ b/tests/t3-live-client.test.ts @@ -33,7 +33,7 @@ const thread = (id = "thread-1", extra = {}) => ({ ...extra, }); -async function fixture(context: TestContext, marker = true) { +async function fixture(context: TestContext) { const requests: Array<{ path: string; authorization?: string; body: string }> = []; const subscriptions: Array> = []; const sockets: WebSocket[] = []; @@ -41,7 +41,6 @@ async function fixture(context: TestContext, marker = true) { let environmentId = "env-1"; let expiresIn = 30 * 86400; let grantedScope = "orchestration:read"; - let descriptorCapability = marker; const server = createServer(async (request, response) => { let body = ""; for await (const chunk of request) body += chunk; @@ -52,7 +51,7 @@ async function fixture(context: TestContext, marker = true) { JSON.stringify({ environmentId, label: "Test environment", - capabilities: { shellResumeCompletionMarker: descriptorCapability }, + capabilities: { terminal: true }, }), ); else if (rejected) { @@ -85,7 +84,6 @@ async function fixture(context: TestContext, marker = true) { snapshot: { snapshotSequence: 10, threads: [thread()], updatedAt: new Date().toISOString() }, }, ]); - if (descriptorCapability) send(socket, [{ kind: "synchronized" }]); }; ws.on("connection", (socket, request) => { assert.equal( @@ -133,9 +131,6 @@ async function fixture(context: TestContext, marker = true) { setScope: (value: string) => { grantedScope = value; }, - setMarker: (value: boolean) => { - descriptorCapability = value; - }, onSubscribe: (callback: typeof onSubscribe) => { onSubscribe = callback; }, @@ -193,7 +188,7 @@ test("pairing exchanges only read scope, streams input changes and removals with await until(client, async () => (await client.getSnapshot()).summary.total === 0); }); -test("reconnect requests a fresh ticket and resumes after the last applied sequence", async (context) => { +test("reconnect requests a fresh ticket and snapshot before accepting live updates", async (context) => { const server = await fixture(context); const client = new T3LiveClient({ store: new MemoryStore(), retryMs: 10 }); context.after(() => client.dispose()); @@ -204,11 +199,18 @@ test("reconnect requests a fresh ticket and resumes after the last applied seque ]); await until(client, async () => (await client.getSnapshot()).summary.input === 1); server.onSubscribe((socket, payload) => { - assert.equal(payload.afterSequence, 19); + assert.deepEqual(payload, {}); send(socket, [ - { kind: "thread-upserted", sequence: 19, thread: thread() }, - { kind: "thread-upserted", sequence: 21, thread: thread("thread-2") }, - { kind: "synchronized" }, + { + kind: "snapshot", + snapshot: { + snapshotSequence: 21, + threads: [thread("thread-2")], + updatedAt: new Date().toISOString(), + }, + }, + { kind: "thread-upserted", sequence: 21, thread: thread("ignored-duplicate") }, + { kind: "thread-upserted", sequence: 22, thread: thread("thread-3") }, ]); }); firstSocket(server.sockets).terminate(); @@ -217,7 +219,7 @@ test("reconnect requests a fresh ticket and resumes after the last applied seque async () => server.subscriptions.length === 2 && (await client.getConnectionStatus()).state === "connected", ); - assert.equal((await client.getSnapshot()).summary.input, 1); + assert.equal((await client.getSnapshot()).summary.input, 0); assert.equal((await client.getSnapshot()).summary.total, 2); assert.equal(server.requests.filter((item) => item.path === "/oauth/token").length, 1); assert.equal(server.requests.filter((item) => item.path === "/api/auth/websocket-ticket").length, 2); @@ -232,6 +234,12 @@ test("expired sessions stop automatic retries; re-pair replaces credentials with await client.pair(server.link); await connected(client); now += 31 * 86400_000; + // T3 authorizes at upgrade: an uninterrupted stream can outlive its bearer credential. + assert.equal((await client.getConnectionStatus()).state, "connected"); + send(firstSocket(server.sockets), [ + { kind: "thread-upserted", sequence: 12, thread: thread("thread-1", { hasPendingUserInput: true }) }, + ]); + await until(client, async () => (await client.getSnapshot()).summary.input === 1); firstSocket(server.sockets).terminate(); await until(client, async () => (await client.getConnectionStatus()).state === "authorization-required"); await assert.rejects(client.getSnapshot(), { code: "authorization-required" }); @@ -247,25 +255,28 @@ test("expired sessions stop automatic retries; re-pair replaces credentials with assert.equal((await client.getConnectionStatus()).state, "pairing-required"); }); -test("old servers receive no resume options; replacement snapshots recover from stale cursors", async (context) => { - const server = await fixture(context, false); +test("reconnect waits for a fresh snapshot even when no events changed", async (context) => { + const server = await fixture(context); const client = new T3LiveClient({ store: new MemoryStore(), retryMs: 10 }); context.after(() => client.dispose()); await client.pair(server.link); await connected(client); assert.deepEqual(server.subscriptions[0], {}); - server.setMarker(true); - server.onSubscribe((socket, payload) => { - assert.equal(payload.afterSequence, 10); - send(socket, [ - { - kind: "snapshot", - snapshot: { snapshotSequence: 1000, threads: [], updatedAt: new Date().toISOString() }, - }, - { kind: "synchronized" }, - ]); + server.onSubscribe((_socket, payload) => { + assert.deepEqual(payload, {}); }); firstSocket(server.sockets).terminate(); + await until(client, async () => server.subscriptions.length === 2); + assert.equal((await client.getConnectionStatus()).state, "connecting"); + await assert.rejects(client.getSnapshot(), { code: "connecting" }); + const socket = server.sockets[1]; + assert.ok(socket); + send(socket, [ + { + kind: "snapshot", + snapshot: { snapshotSequence: 1000, threads: [], updatedAt: new Date().toISOString() }, + }, + ]); await until( client, async () => @@ -395,7 +406,6 @@ test("malformed stream data drops stale state and reconnects with a complete sna kind: "snapshot", snapshot: { snapshotSequence: 22, threads: [], updatedAt: new Date().toISOString() }, }, - { kind: "synchronized" }, ]); }); send(firstSocket(server.sockets), [{ kind: "thread-upserted", sequence: 12, thread: { id: "bad" } }]); From dc14d3eff3eafeda668a3a872e13faf45f0e8ee7 Mon Sep 17 00:00:00 2001 From: beasty Date: Thu, 10 Sep 2026 00:19:04 +0200 Subject: [PATCH 3/3] Keep empty thread-only keys gray --- src/render.ts | 5 ++++- tests/render.test.ts | 2 ++ 2 files changed, 6 insertions(+), 1 deletion(-) diff --git a/src/render.ts b/src/render.ts index 3526c13..6761a3d 100644 --- a/src/render.ts +++ b/src/render.ts @@ -48,7 +48,10 @@ export function getDisplay(model: DashboardModel, mode: DisplayMode = "combined" } if (mode === "threads") { return { - accent: runningAccent(model.summary.running, model.summary.total), + accent: + model.summary.total === 0 + ? COLORS.muted + : runningAccent(model.summary.running, model.summary.total), count: `${model.summary.running}/${model.summary.total}`, footer: model.summary.total === 0 ? "NO THREADS" : "WORKING", label: "T3 CODE", diff --git a/tests/render.test.ts b/tests/render.test.ts index 77f828d..41745c9 100644 --- a/tests/render.test.ts +++ b/tests/render.test.ts @@ -189,6 +189,8 @@ test("question alerts cover input, approvals and plans but exclude waiting and f }); test("combined alerts retain thread counts and thread-only mode never blinks", () => { + const empty: DashboardModel = { kind: "ready", refreshedAt: 0, summary: summary({ total: 0, running: 0 }) }; + assert.equal(getDisplay(empty, "threads").accent, "#6F858E"); const model: DashboardModel = { kind: "ready", refreshedAt: 0,