From 2d99af034c440e397566c57549948fea38c8fb9b Mon Sep 17 00:00:00 2001 From: Jacob Stephens Date: Mon, 19 Oct 2026 23:55:03 +0000 Subject: [PATCH 1/3] Keep the Console Inbox current with server push (#818) --- Cargo.lock | 26 +++++ Cargo.toml | 3 +- console/README.md | 15 ++- console/tests/inbox.test.cjs | 161 +++++++++++++++++++++++++- src/console.rs | 102 +++++++++++----- src/console/assets/console.css | 2 + src/console/assets/console.js | 113 ++++++++++++++++-- src/console/assets/favicon-urgent.svg | 1 + src/console/assets/favicon.svg | 1 + src/console/inbox.html | 8 +- src/console/live.rs | 40 +++++++ src/console/row.html | 4 +- src/logs/status/inbox.rs | 53 ++++++++- tests/console.rs | 131 +++++++++++++++++++++ 14 files changed, 610 insertions(+), 50 deletions(-) create mode 100644 src/console/assets/favicon-urgent.svg create mode 100644 src/console/assets/favicon.svg create mode 100644 src/console/live.rs diff --git a/Cargo.lock b/Cargo.lock index 6b1ad193..20982ec6 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1609,6 +1609,7 @@ dependencies = [ "signal-hook", "tempfile", "tokio", + "tokio-stream", "toml", "toml_edit", "tower-layer", @@ -1688,6 +1689,31 @@ dependencies = [ "tokio", ] +[[package]] +name = "tokio-stream" +version = "0.1.19" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a3d06f0b082ba57c26b79407372e57cf2a1e28124f78e9479fe80322cf53420b" +dependencies = [ + "futures-core", + "pin-project-lite", + "tokio", + "tokio-util", +] + +[[package]] +name = "tokio-util" +version = "0.7.20" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e464cf451ba96ebfc6f9b6542f17ee8b8956e33f1e40d9690624e59d7a7f8a4b" +dependencies = [ + "bytes", + "futures-core", + "futures-sink", + "pin-project-lite", + "tokio", +] + [[package]] name = "toml" version = "1.1.6+spec-1.1.0" diff --git a/Cargo.toml b/Cargo.toml index 9fdea940..0a774b5d 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -31,7 +31,8 @@ signal-hook = "0.4.4" tempfile = "3.27.0" toml = "1.1.6" toml_edit = "0.25.15" -tokio = { version = "1.53.1", features = ["rt", "net", "time"] } +tokio = { version = "1.53.1", features = ["rt", "net", "time", "sync"] } +tokio-stream = { version = "0.1.18", features = ["sync"] } [dev-dependencies] tower-layer = "0.3.3" diff --git a/console/README.md b/console/README.md index 8079a2a9..b7668084 100644 --- a/console/README.md +++ b/console/README.md @@ -27,6 +27,15 @@ theme = "focus-auto" # also focus-light or focus-dark ``` The header names this machine and links to those addresses, without fetching -them. The first page pairs with `thirdshift status`. A refresh reads the current -records; server-push updates are a later Ticket. Replacing the installed binary -makes the Console re-execute itself with its original arguments. +them. The Inbox and its `/events` stream pair with `thirdshift status`. One +server observer checks records and log lengths every second; the page morphs +after a change notice, preserving the selected row, folds and scroll. If the +stream drops or is unavailable, the page polls every five seconds until it +reconnects. The heartbeat only beats for observed log growth, stays flat while +Silent and respects reduced motion. The tab counts Needs you and its favicon +marks urgent items (Slow alone is not urgent). + +Folds and the last-look time stay in this browser. Commands started since the +previous visit are marked new, including those arriving while the page is open. +Missing or malformed storage hides nothing. Replacing the installed binary +closes open streams and re-executes the Console with its original arguments. diff --git a/console/tests/inbox.test.cjs b/console/tests/inbox.test.cjs index 088b8f3b..b1d92e1b 100644 --- a/console/tests/inbox.test.cjs +++ b/console/tests/inbox.test.cjs @@ -2,7 +2,7 @@ const { test, before, after } = require("node:test"); const assert = require("node:assert/strict"); const { spawn } = require("node:child_process"); const { once } = require("node:events"); -const { mkdtemp, mkdir, writeFile, rm } = require("node:fs/promises"); +const { mkdtemp, mkdir, readFile, writeFile, appendFile, utimes, rm } = require("node:fs/promises"); const { tmpdir } = require("node:os"); const { join, resolve } = require("node:path"); const { createInterface } = require("node:readline"); @@ -110,3 +110,162 @@ test("explicit Focus light and dark override the system preference", async t => } } finally { await writeFile(config, "[console]\nport = 0\n"); } }); + +test("push morphs keep the selected row, folds and scroll without a reload", async t => { + const page = await pageFor(t, { viewport: { width: 1440, height: 900 } }); + const errors = []; + page.on("pageerror", error => errors.push(error.message)); + await page.goto(origin); + const row = page.locator('.command').filter({ has: page.locator('.subject', { hasText: /^acme\/widgets#1$/ }) }); + await page.locator('[data-group="done"] > summary').click(); + await row.click(); + await page.evaluate(() => { + window.visitMarker = "same document"; + window.selectedRow = document.activeElement; + document.querySelector('.commands').scrollTop = 120; + }); + const path = join(home, '.thirdshift/logs/acme/widgets/commands/issue/1-fixture.command.json'); + const fixture = JSON.parse(await readFile(path, 'utf8')); + fixture.issue.title = 'Advanced while you watched'; + await writeFile(path, JSON.stringify(fixture)); + await page.getByText('Advanced while you watched', { exact: true }).waitFor(); + assert.equal(await page.evaluate(() => document.activeElement === window.selectedRow && document.activeElement.matches('.command')), true); + assert.equal(await page.evaluate(() => window.visitMarker), "same document"); + assert.equal(await page.locator('[data-group="done"][open]').count(), 1); + assert.equal(await page.locator('[data-group="needs you"][open]').count(), 1); + assert.equal(await page.locator('[data-group="needs you"] .commands').evaluate(el => el.scrollTop), 120); + assert.deepEqual(errors, []); +}); + +test("tab title and favicon follow Needs you through pushes", async t => { + const page = await pageFor(t); + await page.goto(origin); + assert.equal(await page.title(), '10 need you · thirdshift Console'); + assert.equal(await page.locator('link[rel="icon"]').getAttribute('href'), '/assets/favicon-urgent.svg'); + const dir = join(home, '.thirdshift/logs/acme/widgets/commands/issue'); + const fixtures = []; + try { + for (let number = 1; number <= 10; number++) { + const path = join(dir, `${number}-fixture.command.json`); + const original = await readFile(path, 'utf8'); + fixtures.push([path, original]); + const fixture = JSON.parse(original); + fixture.outcome = 'merged'; + await writeFile(path, JSON.stringify(fixture)); + } + await page.waitForFunction(() => document.title === '0 need you · thirdshift Console'); + assert.equal(await page.locator('link[rel="icon"]').getAttribute('href'), '/assets/favicon.svg'); + } finally { for (const [path, original] of fixtures) await writeFile(path, original); } +}); + +test("new since the last visit survives browser storage and clears after looking", async t => { + const page = await pageFor(t); + await page.goto(origin); + await page.waitForFunction(() => localStorage.getItem('thirdshift:console:last-look') !== null); + await page.goto('about:blank'); + const dir = join(home, '.thirdshift/logs/acme/widgets/commands/issue'); + const path = join(dir, '201-fixture.command.json'); + try { + const fixture = JSON.parse(await readFile(join(dir, '1-fixture.command.json'), 'utf8')); + fixture.issue.number = 201; + fixture.started_at = fixture.ended_at = new Date().toISOString(); + fixture.cause = 'New problem after the last visit'; + await writeFile(path, JSON.stringify(fixture)); + await page.goto(origin); + const row = page.locator('.command').filter({ has: page.locator('.subject', { hasText: /^acme\/widgets#201$/ }) }); + await row.locator('.new').waitFor(); + assert.equal(await page.locator('.new:visible').count(), 1); + await page.reload(); + assert.equal(await page.locator('.new:visible').count(), 0); + } finally { await rm(path, { force: true }); } +}); + +test("cutting the open stream falls back to polling and catches up", async t => { + const page = await pageFor(t); + await page.addInitScript(() => { + const NativeSource = EventSource; + window.EventSource = class extends NativeSource { + constructor(...args) { super(...args); window.inboxStream = this; } + }; + }); + await page.goto(origin); + await page.waitForFunction(() => window.inboxStream?.readyState === EventSource.OPEN); + await page.evaluate(() => { + window.inboxStream.close(); + window.inboxStream.dispatchEvent(new Event('error')); + }); + const path = join(home, '.thirdshift/logs/acme/widgets/commands/issue/1-fixture.command.json'); + const original = await readFile(path, 'utf8'); + try { + const fixture = JSON.parse(original); + fixture.issue.title = 'Caught up by polling'; + await writeFile(path, JSON.stringify(fixture)); + await page.getByText('Caught up by polling', { exact: true }).waitFor({ timeout: 12000 }); + assert.equal(await page.evaluate(() => window.inboxStream.readyState), 2); + } finally { await writeFile(path, original); } +}); + +for (const storage of ['blocked', 'malformed']) { + test(`live Inbox works with ${storage} browser storage`, async t => { + const page = await pageFor(t); + const errors = []; + page.on('pageerror', error => errors.push(error.message)); + await page.addInitScript(mode => { + if (mode === 'blocked') Object.defineProperty(window, 'localStorage', { get() { throw new Error('Storage blocked'); } }); + else { + localStorage.setItem('thirdshift:console:folded', 'false'); + localStorage.setItem('thirdshift:console:last-look', '{broken'); + } + }, storage); + await page.goto(origin); + const path = join(home, '.thirdshift/logs/acme/widgets/commands/issue/1-fixture.command.json'); + const original = await readFile(path, 'utf8'); + try { + const fixture = JSON.parse(original); + fixture.issue.title = `Still live with ${storage} storage`; + await writeFile(path, JSON.stringify(fixture)); + await page.getByText(fixture.issue.title, { exact: true }).waitFor(); + assert.equal(await page.locator('[data-group="needs you"][open]').count(), 1); + assert.deepEqual(errors, []); + } finally { await writeFile(path, original); } + }); +} + +test("heartbeat beats only for server-observed log growth and is flat when Silent", { skip: process.platform !== 'linux' }, async t => { + // This fixture borrows the Console's real native process identity. HTTP + // coverage uses an actual held Run on both Linux and macOS. + const stat = await readFile(`/proc/${server.pid}/stat`, 'utf8'); + const start = `${(await readFile('/proc/sys/kernel/random/boot_id', 'utf8')).trim()}:${stat.slice(stat.lastIndexOf(')') + 2).split(' ')[19]}`; + const dir = join(home, '.thirdshift/logs/acme/widgets/commands/issue'); + const path = join(dir, '202-fixture.command.json'); + const log = join(dir, '202-fixture.log'); + const session = join(home, 'heartbeat-session.log'); + const fixture = JSON.parse(await readFile(join(dir, '1-fixture.command.json'), 'utf8')); + Object.assign(fixture, { kind: 'Run', process_id: server.pid, process_start: start, outcome: null, ended_at: null, session_log: session }); + fixture.issue.number = 202; + const began = new Date(Date.now() - 20 * 60 * 1000).toISOString(); + fixture.stages = [{ name: 'implement', began_at: began, session_log: session }]; + await writeFile(path, JSON.stringify(fixture)); + await writeFile(log, 'command'); + await writeFile(session, 'session'); + try { + const page = await pageFor(t); + await page.goto(origin); + await page.locator('[data-group="working"] > summary').click(); + const row = page.locator('.command').filter({ has: page.locator('.subject', { hasText: /^acme\/widgets#202$/ }) }); + const id = await row.getAttribute('id'); + assert.equal(await row.locator('.heartbeat').evaluate(dot => dot.getAnimations().length), 0); + await appendFile(session, ' growth'); + await page.waitForFunction(id => document.getElementById(id).querySelector('.heartbeat').getAnimations().some(animation => animation.playState === 'running'), id); + await page.waitForFunction(id => document.getElementById(id).querySelector('.heartbeat').getAnimations().length === 0, id); + const old = new Date(Date.now() - 20 * 60 * 1000); + await utimes(log, old, old); + await utimes(session, old, old); + await page.waitForFunction(id => document.getElementById(id).dataset.state === 'Silent', id); + assert.equal(await row.locator('.heartbeat').evaluate(dot => dot.getAnimations().length), 0); + await page.emulateMedia({ reducedMotion: 'reduce' }); + await appendFile(log, ' more'); + await page.waitForFunction(id => document.getElementById(id).dataset.logging === 'true', id); + assert.equal(await row.locator('.heartbeat').evaluate(dot => dot.getAnimations().length), 0); + } finally { await Promise.all([path, log, session].map(path => rm(path, { force: true }))); } +}); diff --git a/src/console.rs b/src/console.rs index 22cd4f21..8834948e 100644 --- a/src/console.rs +++ b/src/console.rs @@ -1,4 +1,5 @@ //! The machine's Console. Async execution is confined to this module. +mod live; mod restart; use std::io::Write; @@ -23,12 +24,26 @@ static ASSETS: Dir<'_> = include_dir!("$CARGO_MANIFEST_DIR/src/console/assets"); struct PageRoute { path: &'static str, cli: &'static str, + view: View, } -const PAGES: &[PageRoute] = &[PageRoute { - path: "/", - cli: "thirdshift status", -}]; +enum View { + Inbox, + Changes, +} + +const PAGES: &[PageRoute] = &[ + PageRoute { + path: "/", + cli: "thirdshift status", + view: View::Inbox, + }, + PageRoute { + path: "/events", + cli: "thirdshift status", + view: View::Changes, + }, +]; pub fn serve(no_sign_in: bool) -> Result<()> { ensure!( @@ -47,8 +62,15 @@ pub fn serve(no_sign_in: bool) -> Result<()> { println!("http://{}", listener.local_addr()?); std::io::stdout().flush()?; let watched = executable.clone(); - axum::serve(listener, router()) - .with_graceful_shutdown(async move { watched.replaced().await }) + let (changes, observer) = live::observe(render_inbox("thirdshift status").await?); + axum::serve(listener, router(changes)) + .with_graceful_shutdown(async move { + watched.replaced().await; + // Dropping the sole sender closes every SSE response before + // graceful shutdown waits for the connections to finish. + observer.abort(); + let _ = observer.await; + }) .await .context("Console server failed")?; Ok::<_, anyhow::Error>(()) @@ -58,14 +80,20 @@ pub fn serve(no_sign_in: bool) -> Result<()> { executable.exec() } -fn router() -> Router { +fn router(changes: tokio::sync::watch::Receiver) -> Router { let mut router = Router::new(); for page in PAGES { let cli = page.cli; - router = router.route(page.path, get(move || inbox(cli))); + router = router.route( + page.path, + match page.view { + View::Inbox => get(move || inbox(cli)), + View::Changes => get(live::events), + }, + ); } router = router.route("/assets/{*path}", get(asset)); - router + router.with_state(changes) } #[derive(Template)] @@ -79,6 +107,17 @@ struct InboxPage { } impl InboxPage { + fn needs_you(&self) -> usize { + self.inbox.sections[0].rows.len() + } + + fn urgent(&self) -> bool { + self.inbox.sections[0] + .rows + .iter() + .any(|row| row.state != "Slow") + } + fn row<'a>(&self, row: &'a Row) -> RowTemplate<'a> { RowTemplate { row } } @@ -97,28 +136,35 @@ impl RowTemplate<'_> { } async fn inbox(cli: &'static str) -> Response { - // The shared observations and templates stay blocking, off the async executor. - match tokio::task::spawn_blocking(move || -> Result { - let config = UserConfig::load()?; - Ok(InboxPage { - hostname: crate::host::name().unwrap_or_else(|| "unknown".into()), - consoles: config.console.other_consoles.clone(), - theme: config.console.theme.clone(), cli, - inbox: status::inbox(&config)?, - }.render()?) - }).await { - Ok(Ok(html)) => ( + match render_inbox(cli).await { + Ok(html) => ( [(header::CONTENT_SECURITY_POLICY, "default-src 'self'; script-src 'self'; style-src 'self'; font-src 'self'; connect-src 'self'; img-src 'self' data:; object-src 'none'; base-uri 'none'; frame-ancestors 'none'"), (header::CACHE_CONTROL, "no-store"), (header::X_CONTENT_TYPE_OPTIONS, "nosniff")], Html(html) ).into_response(), - error => { - eprintln!("thirdshift Console: can't read Inbox: {error:?}"); + Err(error) => { + eprintln!("thirdshift Console: can't read Inbox: {error:#}"); (StatusCode::INTERNAL_SERVER_ERROR, "Can't read the Inbox; check the Console's stderr.").into_response() } } } +async fn render_inbox(cli: &'static str) -> Result { + // The shared observations and templates stay blocking, off the async executor. + tokio::task::spawn_blocking(move || -> Result { + let config = UserConfig::load()?; + Ok(InboxPage { + hostname: crate::host::name().unwrap_or_else(|| "unknown".into()), + consoles: config.console.other_consoles.clone(), + theme: config.console.theme.clone(), + cli, + inbox: status::inbox(&config)?, + } + .render()?) + }) + .await? +} + async fn asset(Path(path): Path) -> Response { let Some(file) = ASSETS.get_file(&path) else { return StatusCode::NOT_FOUND.into_response(); @@ -129,6 +175,7 @@ async fn asset(Path(path): Path) -> Response { { Some("css") => "text/css; charset=utf-8", Some("js") => "text/javascript; charset=utf-8", + Some("svg") => "image/svg+xml", Some("woff2") => "font/woff2", _ => "text/plain; charset=utf-8", }; @@ -151,6 +198,11 @@ mod tests { use std::sync::{Arc, Mutex}; use tower_layer::Layer; + fn router() -> Router { + let (_, changes) = tokio::sync::watch::channel(0); + super::router(changes) + } + // Router::route_layer walks every registered method service, including // routes added directly instead of through PAGES. Nothing is dispatched. #[derive(Clone)] @@ -203,14 +255,12 @@ mod tests { #[test] fn every_console_route_has_a_cli_equivalent() { - assert_cli_parity(super::router()); + assert_cli_parity(router()); } #[test] #[should_panic(expected = "registered routes must all have a CLI pair")] fn parity_rejects_a_route_registered_without_a_cli_pair() { - assert_cli_parity( - super::router().route("/unpaired", get(|| async { "page with no CLI pair" })), - ); + assert_cli_parity(router().route("/unpaired", get(|| async { "page with no CLI pair" }))); } } diff --git a/src/console/assets/console.css b/src/console/assets/console.css index 4e0522d6..97b38d73 100644 --- a/src/console/assets/console.css +++ b/src/console/assets/console.css @@ -35,8 +35,10 @@ summary .count { margin-left: 12px; color: var(--dim); font-size: 13px; } .commands, .children, .plan { list-style: none; padding: 0; margin: 0; } .commands { max-height: 300px; overflow: auto; } .command { padding: 8px 10px; margin-bottom: 4px; border-radius: var(--radius); color: var(--dim); overflow-wrap: anywhere; } +.command:focus { outline: 2px solid var(--head); outline-offset: -2px; } .row-head { display: flex; flex-wrap: wrap; align-items: baseline; gap: 4px 10px; } .state { color: var(--text); } +.heartbeat { display: inline-block; width: 6px; height: 6px; border-radius: 50%; background: var(--working); opacity: .4; } .age { margin-left: auto; white-space: nowrap; } .issue-title { flex: 1 1 160px; } .command p { margin: 4px 0 0; } diff --git a/src/console/assets/console.js b/src/console/assets/console.js index cc45e621..776feca5 100644 --- a/src/console/assets/console.js +++ b/src/console/assets/console.js @@ -1,12 +1,107 @@ -// Folded sections are a per-browser convenience. A fresh visit opens Needs you. -const key = "thirdshift:console:folded"; -let folded = {}; -try { folded = JSON.parse(localStorage.getItem(key) ?? "{}"); } catch { /* storage is optional */ } +// Browser preferences are conveniences; malformed or unavailable storage hides nothing. +function stored(key, fallback) { + try { return JSON.parse(localStorage.getItem(key)) ?? fallback; } catch { return fallback; } +} +function remember(key, value) { + try { localStorage.setItem(key, JSON.stringify(value)); } catch { /* storage is optional */ } +} +const visitKey = "thirdshift:console:last-look"; +const visit = stored(visitKey, Date.now()); +const lastLook = typeof visit === "number" && Number.isFinite(visit) && visit >= 0 && visit <= Date.now() ? visit : Date.now(); +function markNew() { + for (const row of document.querySelectorAll(".command")) { + row.querySelector(":scope > .row-head > .new").hidden = Number(row.dataset.startedMs) <= lastLook; + } +} +markNew(); +remember(visitKey, Date.now()); +window.addEventListener("pagehide", () => remember(visitKey, Date.now())); +const foldKey = "thirdshift:console:folded"; +let folded = stored(foldKey, {}); +if (typeof folded !== "object" || Array.isArray(folded)) folded = {}; for (const section of document.querySelectorAll("details[data-group]")) { - if (typeof folded?.[section.dataset.group] === "boolean") section.open = folded[section.dataset.group]; - section.addEventListener("toggle", () => { - folded ??= {}; - folded[section.dataset.group] = section.open; - try { localStorage.setItem(key, JSON.stringify(folded)); } catch { /* storage is optional */ } + if (typeof folded[section.dataset.group] === "boolean") section.open = folded[section.dataset.group]; +} +document.addEventListener("toggle", event => { + const section = event.target; + if (!section.matches("details[data-group]")) return; + folded[section.dataset.group] = section.open; + remember(foldKey, folded); +}, true); +document.addEventListener("click", event => { + event.target.closest(".command")?.focus({ preventScroll: true }); +}); + +function growth() { + return new Map([...document.querySelectorAll(".command")].map(row => [row.id, row.dataset.logGrowth.split(":").map(Number)])); +} + +function morph(html) { + const next = new DOMParser().parseFromString(html, "text/html"); + const current = document.getElementById("inbox"); + const replacement = next.getElementById("inbox"); + if (!replacement) throw new Error("Missing Inbox"); + const active = document.activeElement; + const focused = active?.id || active?.closest("details")?.id; + const wasSummary = active?.matches("summary"); + const scroll = [...current.querySelectorAll("details")].map(section => [section.id, section.querySelector(".commands").scrollTop]); + const position = [scrollX, scrollY]; + const previousGrowth = growth(); + Idiomorph.morph(current, replacement, { + callbacks: { + beforeAttributeUpdated(name, node) { + if (name === "open" && node.matches("details[data-group]")) return false; + } + } }); + const selected = document.getElementById(focused); + (wasSummary ? selected?.querySelector("summary") : selected)?.focus({ preventScroll: true }); + for (const [id, top] of scroll) document.getElementById(id).querySelector(".commands").scrollTop = top; + window.scrollTo(...position); + document.title = next.title; + document.querySelector('link[rel="icon"]').href = next.querySelector('link[rel="icon"]').getAttribute("href"); + markNew(); + for (const row of document.querySelectorAll('.command[data-logging="true"]')) { + const previous = previousGrowth.get(row.id) ?? [0, 0]; + const increased = row.dataset.logGrowth.split(":").some((size, index) => Number(size) > previous[index]); + if (increased && !matchMedia("(prefers-reduced-motion: reduce)").matches) { + row.querySelector(":scope > .row-head > .heartbeat")?.animate([ + { transform: "scale(1)", opacity: 0.4 }, + { transform: "scale(1.8)", opacity: 1, offset: 0.3 }, + { transform: "scale(1)", opacity: 0.4 } + ], { duration: 900 }); + } + } +} + +// Coalesce notices, but fetch again if a change arrives during the request. +let refreshing = false, pending = false; +async function refresh() { + pending = true; + if (refreshing) return; + refreshing = true; + try { + while (pending) { + pending = false; + const response = await fetch("/", { cache: "no-store" }); + if (!response.ok) throw new Error("Inbox unavailable"); + morph(await response.text()); + } + } catch { + startPolling(); + } finally { refreshing = false; } } +let polling; +function startPolling() { + polling ??= setInterval(refresh, 5000); +} +if (typeof EventSource === "function") { + const stream = new EventSource("/events"); + stream.addEventListener("commands", refresh); + stream.addEventListener("error", startPolling); + stream.addEventListener("open", () => { + clearInterval(polling); + polling = undefined; + refresh(); + }); +} else { startPolling(); } diff --git a/src/console/assets/favicon-urgent.svg b/src/console/assets/favicon-urgent.svg new file mode 100644 index 00000000..cfea7e8e --- /dev/null +++ b/src/console/assets/favicon-urgent.svg @@ -0,0 +1 @@ + diff --git a/src/console/assets/favicon.svg b/src/console/assets/favicon.svg new file mode 100644 index 00000000..e5fa9f5b --- /dev/null +++ b/src/console/assets/favicon.svg @@ -0,0 +1 @@ + diff --git a/src/console/inbox.html b/src/console/inbox.html index a9bfdfd7..06d6bc48 100644 --- a/src/console/inbox.html +++ b/src/console/inbox.html @@ -4,8 +4,8 @@ - Inbox · {{ hostname }} · thirdshift - + {{ self.needs_you() }} need you · thirdshift Console + @@ -16,11 +16,11 @@

thirdshift / Console

{{ hostname }}

{% if !consoles.is_empty() %}{% endif %}
-
+

Inbox

{{ cli }}
{% match inbox.warning %}{% when Some with (warning) %}

Warning: {{ warning }}

{% when None %}{% endmatch %} {% for section in inbox.sections %} -
+
{{ section.title }}{{ section.rows.len() }} {% if section.rows.is_empty() && section.references.is_empty() %}

Nothing here.

{% endif %} {# Only the rendered, auto-escaped RowTemplate is trusted as HTML. #} diff --git a/src/console/live.rs b/src/console/live.rs new file mode 100644 index 00000000..618f4376 --- /dev/null +++ b/src/console/live.rs @@ -0,0 +1,40 @@ +//! One reader per Console; slow browsers only need the latest change notice. +use std::{convert::Infallible, time::Duration}; + +use axum::extract::State; +use axum::response::{ + IntoResponse, Response, Sse, + sse::{Event, KeepAlive}, +}; +use tokio::sync::watch; +use tokio_stream::{StreamExt, wrappers::WatchStream}; + +pub(super) fn observe(initial: String) -> (watch::Receiver, tokio::task::JoinHandle<()>) { + let (sender, changes) = watch::channel(0); + let observer = tokio::spawn(async move { + let mut previous = initial; + loop { + tokio::time::sleep(Duration::from_secs(1)).await; + match super::render_inbox("thirdshift status").await { + Ok(html) if html != previous => { + previous = html; + sender.send_modify(|revision| *revision += 1); + } + Ok(_) => {} + Err(error) => eprintln!("thirdshift Console: can't observe Inbox: {error:#}"), + } + } + }); + (changes, observer) +} + +pub(super) async fn events(State(changes): State>) -> Response { + // WatchStream emits immediately, so every reconnect also catches up. + let stream = WatchStream::new(changes) + .map(|_| Ok::<_, Infallible>(Event::default().event("commands").data("changed"))); + ( + [("x-accel-buffering", "no"), ("cache-control", "no-store")], + Sse::new(stream).keep_alive(KeepAlive::default()), + ) + .into_response() +} diff --git a/src/console/row.html b/src/console/row.html index 31f68311..61eb6418 100644 --- a/src/console/row.html +++ b/src/console/row.html @@ -1,5 +1,5 @@ -
  • -
    {{ row.state }} {{ row.subject }} {{ row.title }}{{ row.age }}
    +
  • +
    {% if row.live %}{% endif %}{{ row.state }} {{ row.subject }} {{ row.title }}{{ row.age }}

    {{ row.now_line }}

    {% if !row.harness.is_empty() %}

    {{ row.harness }}

    {% endif %} {% if !row.pull_request.is_empty() %}

    {{ row.pull_request }}

    {% endif %} diff --git a/src/logs/status/inbox.rs b/src/logs/status/inbox.rs index 9901f0f7..513bfd7c 100644 --- a/src/logs/status/inbox.rs +++ b/src/logs/status/inbox.rs @@ -1,11 +1,12 @@ //! Owned Inbox presentation for the Console, from the same observations as status. use anyhow::Result; use chrono::{DateTime, Local}; +use std::path::Path; use super::{age, stage_text, ticket_text}; use crate::config::UserConfig; use crate::logs::command_record::Issue; -use crate::logs::reading::{Command, Entry, Group, Snapshot}; +use crate::logs::reading::{Command, Entry, Group, Health, Snapshot}; pub(crate) struct Inbox { pub sections: Vec
    , @@ -22,6 +23,11 @@ pub(crate) struct Section { } pub(crate) struct Row { + pub id: String, + pub started_ms: i64, + pub live: bool, + pub logging: bool, + pub log_growth: String, pub subject: String, pub title: String, pub state: String, @@ -48,11 +54,19 @@ pub(crate) fn inbox(config: &UserConfig) -> Result { .filter(|item| item.group == group) .map(|item| { let mut row = match item.command { - Entry::Record(command) => row(command, now), + Entry::Record(command) => row(command, now, None), Entry::History(command) => { let (subject, title) = subject(&command.repository, &command.kind, command.issue.as_ref()); Row { + id: identity(&format!( + "history:{}:{}:{}", + command.repository, command.kind, command.started_at + )), + started_ms: timestamp(&command.started_at), + live: false, + logging: false, + log_growth: String::new(), subject, title, state: command.state.into(), @@ -139,10 +153,21 @@ fn subject(repository: &str, name: &str, issue: Option<&Issue>) -> (String, Stri } } -fn row(command: &Command, now: DateTime) -> Row { +fn row(command: &Command, now: DateTime, outer_log: Option<&Path>) -> Row { let facts = &command.record; + let log = outer_log.unwrap_or(&command.command_log); + let live = command.state == "live"; let (subject, title) = subject(&facts.repository, &facts.name, facts.issue.as_ref()); Row { + id: identity(&command.record_path.to_string_lossy()), + started_ms: timestamp(&facts.started_at), + live, + logging: live && !matches!(command.health, Some(Health::Silent { .. })), + log_growth: format!( + "{}:{}", + log_length(Some(log)), + log_length(facts.session_log.as_deref()) + ), subject, title, state: command.state.into(), @@ -166,11 +191,31 @@ fn row(command: &Command, now: DateTime) -> Row { .dispatched .iter() .chain(&command.children) - .map(|child| row(child, now)) + .map(|child| row(child, now, Some(log))) .collect(), } } +fn log_length(path: Option<&Path>) -> u64 { + path.and_then(|path| std::fs::metadata(path).ok()) + .map_or(0, |file| file.len()) +} + +fn timestamp(value: &str) -> i64 { + DateTime::parse_from_rfc3339(value).map_or(0, |time| time.timestamp_millis()) +} + +fn identity(value: &str) -> String { + // Encode the exact storage identity: issue numbers and timestamps alone + // can collide, and DOM IDs cannot contain whitespace. + let encoded: String = value + .as_bytes() + .iter() + .map(|byte| format!("{byte:02x}")) + .collect(); + format!("command-{encoded}") +} + fn elapsed(time: &str, now: DateTime) -> String { DateTime::parse_from_rfc3339(time) .map(|time| age(now.signed_duration_since(time).num_seconds().max(0))) diff --git a/tests/console.rs b/tests/console.rs index e8455372..c1c3ead5 100644 --- a/tests/console.rs +++ b/tests/console.rs @@ -1,4 +1,6 @@ //! The Console's public seam: the binary, fixture records and HTTP. +mod support; + use std::fs; use std::io::{BufRead, BufReader}; use std::path::Path; @@ -16,6 +18,21 @@ struct Console { reader: Option>, } +struct HeldFixture { + run: Option, + release: std::path::PathBuf, +} + +impl Drop for HeldFixture { + fn drop(&mut self) { + // Release and reap the fixture even when a regression assertion fails. + let _ = fs::write(&self.release, ""); + if let Some(run) = self.run.take() { + run.finish(); + } + } +} + impl Console { fn start(binary: &Path, home: &Path) -> Self { let mut child = Command::new(binary) @@ -104,6 +121,119 @@ fn write_record(home: &Path, number: u64, record: &Value) { .unwrap(); } +#[test] +fn command_changes_push_events_over_http() { + let home = tempfile::tempdir().unwrap(); + configure(home.path()); + let console = Console::start(Path::new(env!("CARGO_BIN_EXE_thirdshift")), home.path()); + let response = console.get("/events"); + assert_eq!(response.status(), 200); + assert_eq!(response.headers()["content-type"], "text/event-stream"); + assert_eq!(response.headers()["x-accel-buffering"], "no"); + let mut stream = BufReader::new(response); + fn changed(stream: &mut impl BufRead) { + let mut event = String::new(); + loop { + let mut line = String::new(); + assert!(stream.read_line(&mut line).unwrap() > 0, "stream ended"); + if line.trim().is_empty() && event.contains("event: commands") { + assert!(event.contains("data: changed"), "{event}"); + return; + } + if line.trim().is_empty() { + event.clear(); + } else { + event.push_str(&line); + } + } + } + changed(&mut stream); // Reconnects always catch up, even if an event was missed. + let mut fixture = record(home.path(), 81, None, 0); + changed(&mut stream); + assert!(console.get("/").text().unwrap().contains("acme/widgets#81")); + fixture["stages"][0]["name"] = json!("CI watch"); + write_record(home.path(), 81, &fixture); + changed(&mut stream); + assert!( + console + .get("/") + .text() + .unwrap() + .contains("died at CI watch") + ); + fixture["outcome"] = json!("merged"); + fixture["ended_at"] = json!(chrono::Local::now().to_rfc3339()); + write_record(home.path(), 81, &fixture); + changed(&mut stream); + assert!(console.get("/").text().unwrap().contains(">merged

    ")); +} + +#[test] +fn heartbeat_observes_log_lengths_and_is_flat_for_silent_runs() { + let scenario = support::Scenario::new(); + scenario.user_config_is("[console]\nport = 0\n"); + scenario.agent_does(&format!( + "touch '{}'\nwhile ! test -f '{}'; do sleep 0.02; done\nexit 1\n", + scenario.path("started").display(), + scenario.path("release").display() + )); + let _held = HeldFixture { + run: Some(scenario.run_until(&[&scenario.issue_url(7)], &[], "started")), + release: scenario.path("release"), + }; + let dir = scenario.path("home/.thirdshift/logs/acme/widgets/commands/issue"); + let path = fs::read_dir(dir) + .unwrap() + .map(|entry| entry.unwrap().path()) + .find(|path| path.to_string_lossy().ends_with(".command.json")) + .unwrap(); + let mut fixture: Value = serde_json::from_slice(&fs::read(&path).unwrap()).unwrap(); + let session = std::path::PathBuf::from(fixture["session_log"].as_str().unwrap()); + let log = path.with_file_name( + path.file_name() + .unwrap() + .to_str() + .unwrap() + .replace(".command.json", ".log"), + ); + let began = (chrono::Local::now() - chrono::Duration::minutes(20)).to_rfc3339(); + fixture["stages"] = json!([{"name": "implement", "began_at": began, "session_log": session}]); + fs::write(&path, serde_json::to_vec(&fixture).unwrap()).unwrap(); + fs::write(&log, "a").unwrap(); + fs::write(&session, "ab").unwrap(); + let console = Console::start( + Path::new(env!("CARGO_BIN_EXE_thirdshift")), + &scenario.path("home"), + ); + let html = console.get("/").text().unwrap(); + assert!(html.contains("data-log-growth=\"1:2\""), "{html}"); + assert!(html.contains("data-logging=\"true\"")); + fs::write(&session, "abcde").unwrap(); + assert!( + console + .get("/") + .text() + .unwrap() + .contains("data-log-growth=\"1:5\"") + ); + for file in [&log, &session] { + fs::File::options() + .write(true) + .open(file) + .unwrap() + .set_times( + fs::FileTimes::new() + .set_modified(std::time::SystemTime::now() - Duration::from_secs(20 * 60)), + ) + .unwrap(); + } + let html = console.get("/").text().unwrap(); + assert!(html.contains("data-state=\"Silent\"")); + assert!(html.contains("data-logging=\"false\"")); + assert!(html.contains("data-log-growth=\"1:5\"")); + drop(console); +} + #[test] fn inbox_over_http_matches_status_and_embeds_only_local_resources() { let home = tempfile::tempdir().unwrap(); @@ -291,6 +421,7 @@ fn replacing_the_binary_executes_the_replacement_with_the_original_arguments() { assert!(copied.success()); let console = Console::start(&binary, home.path()); assert_eq!(console.get("/").status(), 200); + let _open_stream = console.get("/events"); let replacement = home.path().join("replacement"); fs::write(&replacement, "#!/bin/sh\nprintf '%s\\n' \"$@\" > \"$HOME/reexec-args\"\nprintf 'replacement executed\\n'\n").unwrap(); fs::set_permissions(&replacement, fs::Permissions::from_mode(0o755)).unwrap(); From 4dde3f72e03a4ad2aa2ef2f65f3cab247c5060a2 Mon Sep 17 00:00:00 2001 From: Jacob Stephens Date: Tue, 20 Oct 2026 00:00:31 +0000 Subject: [PATCH 2/3] Preserve dispatch and folded row selection after a push --- console/tests/inbox.test.cjs | 46 ++++++++++++++++++++++++++++++++++ src/console/assets/console.css | 2 +- src/console/assets/console.js | 23 +++++++++++++++-- src/logs/status/inbox.rs | 11 +++++++- 4 files changed, 78 insertions(+), 4 deletions(-) diff --git a/console/tests/inbox.test.cjs b/console/tests/inbox.test.cjs index b1d92e1b..4e94fa91 100644 --- a/console/tests/inbox.test.cjs +++ b/console/tests/inbox.test.cjs @@ -269,3 +269,49 @@ test("heartbeat beats only for server-observed log growth and is flat when Silen assert.equal(await row.locator('.heartbeat').evaluate(dot => dot.getAnimations().length), 0); } finally { await Promise.all([path, log, session].map(path => rm(path, { force: true }))); } }); + +test("a push keeps a Pass dispatch selected instead of its parent", async t => { + const dir = join(home, '.thirdshift/logs/acme/widgets/commands/security'); + await mkdir(dir, { recursive: true }); + const path = join(dir, '20261019T110000+0000~security.command.json'); + const now = new Date().toISOString(); + const fixture = { + version: 1, kind: 'Security run', name: 'secure-fixture', repository: 'acme/widgets', hostname: 'fixture', + issue: null, process_id: 0, process_start: 'gone', harness: 'codex', model: null, effort: null, + started_at: now, ended_at: now, stages: [], session_log: null, outcome: 'failed', cause: 'Pass failure', issue_branch: null, pull_request: null + }; + fixture.dispatched = [{ ...fixture, kind: 'Run', name: 'issue-303', issue: { number: 303, url: 'https://github.com/acme/widgets/issues/303', title: 'Dispatch before push' } }]; + await writeFile(path, JSON.stringify(fixture)); + try { + const page = await pageFor(t); + await page.goto(origin); + const child = page.locator('.command').filter({ has: page.locator(':scope > .row-head > .subject', { hasText: /^acme\/widgets#303$/ }) }); + await child.click(); + fixture.dispatched[0].issue.title = 'Dispatch after push'; + await writeFile(path, JSON.stringify(fixture)); + await page.getByText('Dispatch after push', { exact: true }).waitFor(); + assert.equal(await page.evaluate(() => document.activeElement.querySelector(':scope > .row-head > .subject')?.textContent), 'acme/widgets#303'); + const ids = await page.locator('.command').evaluateAll(rows => rows.map(row => row.id)); + assert.equal(new Set(ids).size, ids.length); + } finally { await rm(path, { force: true }); } +}); + +test("selection survives a move into folded Done and regains focus when opened", async t => { + const path = join(home, '.thirdshift/logs/acme/widgets/commands/issue/1-fixture.command.json'); + const original = await readFile(path, 'utf8'); + try { + const page = await pageFor(t); + await page.goto(origin); + const row = page.locator('.command').filter({ has: page.locator(':scope > .row-head > .subject', { hasText: /^acme\/widgets#1$/ }) }); + await row.click(); + const id = await row.getAttribute('id'); + const fixture = JSON.parse(original); + fixture.outcome = 'merged'; + await writeFile(path, JSON.stringify(fixture)); + await page.waitForFunction(id => document.getElementById(id)?.closest('details').dataset.group === 'done', id); + assert.equal(await page.evaluate(() => document.querySelector('.command[aria-current="true"]')?.id ?? document.activeElement.id), id); + assert.equal(await page.locator('[data-group="done"][open]').count(), 0); + await page.locator('[data-group="done"] > summary').click(); + await page.waitForFunction(id => document.activeElement.id === id, id); + } finally { await writeFile(path, original); } +}); diff --git a/src/console/assets/console.css b/src/console/assets/console.css index 97b38d73..7ce2bf6c 100644 --- a/src/console/assets/console.css +++ b/src/console/assets/console.css @@ -35,7 +35,7 @@ summary .count { margin-left: 12px; color: var(--dim); font-size: 13px; } .commands, .children, .plan { list-style: none; padding: 0; margin: 0; } .commands { max-height: 300px; overflow: auto; } .command { padding: 8px 10px; margin-bottom: 4px; border-radius: var(--radius); color: var(--dim); overflow-wrap: anywhere; } -.command:focus { outline: 2px solid var(--head); outline-offset: -2px; } +.command:focus, .command[aria-current="true"] { outline: 2px solid var(--head); outline-offset: -2px; } .row-head { display: flex; flex-wrap: wrap; align-items: baseline; gap: 4px 10px; } .state { color: var(--text); } .heartbeat { display: inline-block; width: 6px; height: 6px; border-radius: 50%; background: var(--working); opacity: .4; } diff --git a/src/console/assets/console.js b/src/console/assets/console.js index 776feca5..c92ab711 100644 --- a/src/console/assets/console.js +++ b/src/console/assets/console.js @@ -18,6 +18,12 @@ remember(visitKey, Date.now()); window.addEventListener("pagehide", () => remember(visitKey, Date.now())); const foldKey = "thirdshift:console:folded"; let folded = stored(foldKey, {}); +let selectedRowId; +function selectRow(row) { + document.querySelector('.command[aria-current="true"]')?.removeAttribute("aria-current"); + selectedRowId = row?.id; + row?.setAttribute("aria-current", "true"); +} if (typeof folded !== "object" || Array.isArray(folded)) folded = {}; for (const section of document.querySelectorAll("details[data-group]")) { if (typeof folded[section.dataset.group] === "boolean") section.open = folded[section.dataset.group]; @@ -27,13 +33,22 @@ document.addEventListener("toggle", event => { if (!section.matches("details[data-group]")) return; folded[section.dataset.group] = section.open; remember(foldKey, folded); + const selected = document.getElementById(selectedRowId); + if (section.open && selected && section.contains(selected)) selected.focus({ preventScroll: true }); }, true); +document.addEventListener("focusin", event => { + const row = event.target.closest(".command"); + if (row) selectRow(row); +}); document.addEventListener("click", event => { event.target.closest(".command")?.focus({ preventScroll: true }); }); +function logLengths(row) { + return row.dataset.logGrowth.split(":").map(Number); +} function growth() { - return new Map([...document.querySelectorAll(".command")].map(row => [row.id, row.dataset.logGrowth.split(":").map(Number)])); + return new Map([...document.querySelectorAll(".command")].map(row => [row.id, logLengths(row)])); } function morph(html) { @@ -46,6 +61,7 @@ function morph(html) { const wasSummary = active?.matches("summary"); const scroll = [...current.querySelectorAll("details")].map(section => [section.id, section.querySelector(".commands").scrollTop]); const position = [scrollX, scrollY]; + const selection = selectedRowId; const previousGrowth = growth(); Idiomorph.morph(current, replacement, { callbacks: { @@ -56,6 +72,9 @@ function morph(html) { }); const selected = document.getElementById(focused); (wasSummary ? selected?.querySelector("summary") : selected)?.focus({ preventScroll: true }); + // Focus cannot enter a folded section. Keep the logical row selected so + // it remains selected, and regains focus, when that section is reopened. + selectRow(document.getElementById(selection)); for (const [id, top] of scroll) document.getElementById(id).querySelector(".commands").scrollTop = top; window.scrollTo(...position); document.title = next.title; @@ -63,7 +82,7 @@ function morph(html) { markNew(); for (const row of document.querySelectorAll('.command[data-logging="true"]')) { const previous = previousGrowth.get(row.id) ?? [0, 0]; - const increased = row.dataset.logGrowth.split(":").some((size, index) => Number(size) > previous[index]); + const increased = logLengths(row).some((size, index) => size > previous[index]); if (increased && !matchMedia("(prefers-reduced-motion: reduce)").matches) { row.querySelector(":scope > .row-head > .heartbeat")?.animate([ { transform: "scale(1)", opacity: 0.4 }, diff --git a/src/logs/status/inbox.rs b/src/logs/status/inbox.rs index 513bfd7c..b51d2542 100644 --- a/src/logs/status/inbox.rs +++ b/src/logs/status/inbox.rs @@ -159,7 +159,16 @@ fn row(command: &Command, now: DateTime, outer_log: Option<&Path>) -> Row let live = command.state == "live"; let (subject, title) = subject(&facts.repository, &facts.name, facts.issue.as_ref()); Row { - id: identity(&command.record_path.to_string_lossy()), + // A Pass's dispatched Runs share its storage path. Their immutable + // Command facts distinguish them without tying identity to ordering. + id: identity(&format!( + "{}\0{}\0{}\0{}\0{}", + command.record_path.to_string_lossy(), + facts.kind, + facts.name, + facts.issue.as_ref().map_or("", |issue| issue.url.as_str()), + facts.started_at + )), started_ms: timestamp(&facts.started_at), live, logging: live && !matches!(command.health, Some(Health::Silent { .. })), From 8c662ab78eb53b8d5367441003319995f67fb6f8 Mon Sep 17 00:00:00 2001 From: Jacob Stephens Date: Tue, 20 Oct 2026 00:02:24 +0000 Subject: [PATCH 3/3] Measure Console CLI parity after attaching router state --- src/console.rs | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/src/console.rs b/src/console.rs index 8834948e..85bf4ceb 100644 --- a/src/console.rs +++ b/src/console.rs @@ -224,8 +224,14 @@ mod tests { fn assert_cli_parity(router: Router) { // Measure the GET registration footprint through axum's public layer - // interface, so this does not depend on its implicit HEAD/fallback count. - let get_methods = registered_methods(Router::new().route("/known", get(|| async {}))); + // interface after attaching state, as the production router does, so + // this does not depend on its implicit HEAD/fallback count. + let (_, changes) = tokio::sync::watch::channel(0_u64); + let get_methods = registered_methods( + Router::>::new() + .route("/known", get(|| async {})) + .with_state(changes), + ); let expected = (super::PAGES.len() + 1) * get_methods; // one static-resource route assert_eq!( registered_methods(router),