diff --git a/console/README.md b/console/README.md index a4da1397b..f0530f56f 100644 --- a/console/README.md +++ b/console/README.md @@ -340,6 +340,16 @@ line, cut off at the pane's edge, with the whole command on hover and in the copy. Session trace links also stay on one line, trimmed from the start so the end of each path, which tells them apart, still shows. +After a build on Linux, `python3 console/reproductions/peek-cache-contention.py` +checks that parsing another Command's cold Session trace does not delay cached +Command details. + +Historical Session signals are cached in memory and in disposable +`*.signals-cache.json` files beside their Session traces, with per-trace +`*.signals-cache.lock` files coordinating CLI processes. Appends, replacements +and binary changes invalidate saved signals. A read-only logs directory still +works by parsing without saving a cache. Command records are never rewritten. + Zoom fits a parent span or a padded window around a call. The sticky axis keeps the zoom notice and Zoom out control visible; the slowest calls re-rank within the window. Folds, selection, zoom and Peek disclosures survive live refresh. diff --git a/console/reproductions/peek-cache-contention.py b/console/reproductions/peek-cache-contention.py new file mode 100644 index 000000000..a996328f4 --- /dev/null +++ b/console/reproductions/peek-cache-contention.py @@ -0,0 +1,99 @@ +#!/usr/bin/env python3 +"""After cargo build, check cached details during another Command's cold parse. + +Linux only: IN_OPEN pins the overlapping requests to actual Session trace IO. +The 128 MiB OTLP fixture makes lock contention visible even on a fast machine. +""" + +import concurrent.futures +import ctypes +import json +import os +import pathlib +import select +import subprocess +import tempfile +import time +import urllib.request + + +def check(): + binary = str(pathlib.Path(__file__).resolve().parents[2] / 'target/debug/thirdshift') + with tempfile.TemporaryDirectory(prefix='thirdshift-peek-contention-') as directory: + task_home = pathlib.Path(directory) + config = task_home / '.thirdshift' + records = config / 'logs/acme/widgets/commands/issue' + records.mkdir(parents=True) + (config / 'config.toml').write_text('[console]\nport = 0\n') + for number in [1, 2]: + trace = records / f'{number}-fixture.logs.jsonl' + if number == 1: + trace.write_text('{"resourceLogs":[]}\n') + else: + line = json.dumps({'resourceLogs': [{'scopeLogs': [{'logRecords': [ + {'body': {'stringValue': 'x' * (8 * 1024 * 1024)}} + ]}]}]}) + '\n' + with trace.open('w') as output: + for _ in range(16): + output.write(line) + record = dict( + version=1, kind='Run', repository='acme/widgets', + name=f'acme/widgets#{number}', + issue=dict(number=number, url=f'https://github.com/acme/widgets/issues/{number}', title='Widgets'), + hostname='factory', process_id=0, process_start='gone', harness='claude', + started_at='2026-10-19T11:00:00-04:00', ended_at='2026-10-19T11:01:00-04:00', + stages=[dict(name='implement', began_at='2026-10-19T11:00:00-04:00', session_log=str(trace))], + outcome='merged', + ) + (records / f'{number}-fixture.command.json').write_text(json.dumps(record)) + env = dict(os.environ, HOME=directory) + server = subprocess.Popen( + [binary, 'console', '--no-sign-in'], env=env, + stdout=subprocess.PIPE, text=True, + ) + try: + assert select.select([server.stdout], [], [], 10)[0], 'Console startup timed out' + origin = server.stdout.readline().strip() + assert origin.startswith('http://127.0.0.1:'), origin + + def peek(number): + began = time.monotonic() + with urllib.request.urlopen( + origin + '/peek?command=widgets%23' + str(number), timeout=45, + ) as response: + assert response.status == 200 + response.read() + return time.monotonic() - began + + print('warm selected details seconds', peek(1), flush=True) + libc = ctypes.CDLL(None, use_errno=True) + descriptor = libc.inotify_init1(os.O_NONBLOCK | os.O_CLOEXEC) + assert descriptor >= 0 + try: + watched = libc.inotify_add_watch( + descriptor, os.fsencode(records / '2-fixture.logs.jsonl'), 0x20, # IN_OPEN + ) + assert watched >= 0 + with concurrent.futures.ThreadPoolExecutor() as executor: + cold = executor.submit(peek, 2) + assert select.select([descriptor], [], [], 15)[0], 'cold reader did not open trace' + os.read(descriptor, 8192) + warm_seconds = peek(1) + cold_seconds = cold.result(timeout=45) + print('cold details seconds', cold_seconds, flush=True) + print('cached unrelated details seconds', warm_seconds, flush=True) + assert warm_seconds < 1, f'cached details took {warm_seconds:.3f}s during an unrelated parse' + finally: + os.close(descriptor) + finally: + server.terminate() + try: + server.wait(timeout=10) + except subprocess.TimeoutExpired: + server.kill() + server.wait() + server.stdout.close() + + +if __name__ == '__main__': + check() diff --git a/console/tests/inbox.test.cjs b/console/tests/inbox.test.cjs index e9c8d8b86..6858ee933 100644 --- a/console/tests/inbox.test.cjs +++ b/console/tests/inbox.test.cjs @@ -66,6 +66,32 @@ async function pageFor(t, options) { return page; } +test('Command details times out a held request and can be opened again', async t => { + const page = await pageFor(t); + await page.clock.install(); + let release; + const held = new Promise(resolve => { release = resolve; }); + await page.route('**/peek?*', async route => { + await held; + await route.continue(); + }); + try { + await page.goto(`${origin}/?page=security`); + const requested = page.waitForRequest('**/peek?*'); + await page.locator('.command .state').first().click(); + await requested; + await page.clock.fastForward(15000); + await page.getByText('Details unavailable. Reload to try again.', { exact: true }) + .waitFor({ timeout: 2000 }); + } finally { + release(); + await page.unrouteAll({ behavior: 'wait' }); + } + await page.locator('[data-close-peek]').click(); + await page.locator('.command .state').first().click(); + await page.locator('#peek dt').first().waitFor(); +}); + // The Inbox shows one section, a folder, at a time; Details opens from a row. const folder = (page, group) => page.locator(`#inbox .folders a[data-group="${group}"]`); const shownGroup = page => page.locator('#inbox > .group:not([hidden])'); diff --git a/site/architecture/index.html b/site/architecture/index.html index e3df33439..9b8224873 100644 --- a/site/architecture/index.html +++ b/site/architecture/index.html @@ -219,6 +219,7 @@

Architecture

thirdshift experiment --release <commit> compares the content commit with the last release tag on touched suites. The release script waits for rechecks and asks permission for a fail or inconclusive Verdict before pushing. Reruns reuse the Experiment, including before opening a pushed branch's PR. It refuses different main content before pushing or merging and checks the merged tree before tagging, so the release cannot contain unmeasured changes. No touched suite or a last release without the Trial flag carries on with a note. Releasing.

Each Ticket run's own process saves its Run metrics at completion in its Command record, including Ticket runs and Base fixes: outcome, Repairs, CI rounds, Unaddressed findings by Standards, Spec and Security, stage times against their Usual ranges, and minutes by stage and in total. Unaddressed findings counts retain Delivery's final PR body and the Security review's count, so later edits do not change them. Each Session stage keeps the review's Standards and Spec finding counts from its summary, with missing counts left missing. thirdshift show <command> --json includes these retained facts. The Console reads the same details, showing each Ticket run's available metrics and marking unavailable signals as Missing. thirdshift metrics <command> [--json] reads a Ticket run with its children; without a Command it groups weekly trends by repository, Harness, requested Model and Effort, Ticket run kind and session kind. Week and repository filters narrow that view. Historical metrics are computed from Command records without writing back, and missing signals stay missing. No combined score is produced.

Run metrics and Experiments share one Session usage module for tokens by observed Model and Harness-reported cost. Its normalized-event and retained-span readers resolve call amendments and partial totals, retaining fields a total omits. Input includes cache reads and writes. Reported Session cost wins; otherwise each Model's reported cost replaces its call costs. The metrics reader separately counts tool errors, permission denials, killed work and repeated tool calls within that session. Completion metrics sum these by session kind, and weekly Session trends split by the Harness version from each trace's resource. Saved completion metrics stay fixed. Historical metrics also read converted Session traces, leaving missing traces and unreported measurements missing.

+

Command details reconstruct Session signals only for the selected Command and its nested Ticket runs and dispatches; all records still contribute to Usual ranges and health. Weekly metrics read signals for every Command. A bounded memory cache retains derived signals by the resolved Session trace's path, length, modification time and file identity, checking the identity again after parsing. Disposable *.signals-cache.json files beside each Session trace also retain them across CLI processes and Console restarts. Per-trace locks coordinate readers. Appends, replacements and a different binary invalidate the saved signals; an unwritable cache falls back to parsing. Historical Command records are never rewritten.

Confirmed Bug origins are joined when metrics are read, from local Scoring checkpoints under ~/.thirdshift/evals/scoring/. The latest window's decision for each bug wins. Its retained local Ticket run identity selects the exact Ticket run; cross-host pull-request attribution selects the latest local Ticket run for that URL, as in Scoring, so Continuations never multiply a count. The Scoring data covers both hosts; metrics read no remote records. Late confirmations count in the Ticket run's start week, corrections move the count to the settled pull request, and rejected or unsettled origins count for nothing. Command details in show --json and the Console share this projection and reconstruct available historical metrics. Command records are never rewritten.

The same reader checks each session's shell commands against one code-owned destructive-command pattern list. Force pushes, shared-branch history rewrites and recursive deletes outside the Worktree are retained with their commands in Run metrics, with flag counts in weekly Session trends. This check is advisory: it never blocks or fails a Ticket run.

    diff --git a/site/how-it-works/index.html b/site/how-it-works/index.html index 4f1545e40..6ffde1fc1 100644 --- a/site/how-it-works/index.html +++ b/site/how-it-works/index.html @@ -160,6 +160,7 @@

    How thirdshift works

    Run metrics and Experiments share Session usage rules. Partial totals replace only reported fields, preserving known input, cache counts and cost. Input includes cache reads and writes. A reported Session cost wins; otherwise each Model's reported cost replaces its call costs. Missing measurements stay missing, and saved completion metrics stay fixed.

    CLI log following and the Console keep reading through the captured writer's exit, including final output written after a Command's outcome is recorded. An exact Session trace finishes when its writer exits or its owning record moves to a later Session.

    The Console shows this machine's Commands, opening on its Inbox. Beside the CLI, you can watch their stages and logs there. Open Command details for a Ticket run's metrics, including available historical signals and late Bug-origin confirmations; unavailable signals say Missing. Stages keeps each Spec run with its Ticket runs, newest-started groups first, filling the window's width beside any open Command details: scroll right for older runs, or switch from All runs to By repository for one grid per repository, each folded to its name and run count until opened. When a later Command takes up an issue, the earlier row stays in Resolved for inbox.days from that Command's start, saying why it left.

    +

    Command details and thirdshift show <command> --json read historical Session signals only for that Command and its nested work. The CLI and Console reuse derived signals while each Session trace is unchanged, including across process restarts, reading it again after an append, replacement or binary update. If a details request has not finished within 10 seconds, the pane says “Details unavailable. Reload to try again.”

    The Console reads ready rows' pull requests and Needs you rows' issues and pull requests from GitHub in the background once a minute, even with no pages open. A changed answer wakes open Inbox pages; an unchanged answer leaves them idle. thirdshift status uses the same kept answer and reads GitHub itself only when it is missing or five minutes old. An ended Needs you row leaves once its issue closes or its pull request merges; a Triage row also leaves when needs-triage is removed. Architect and Security runs use the issue and pull request of the Spec run or Ticket run they dispatched. A pull request closed unmerged leaves a failure waiting while its issue is open. Live Silent and Slow rows wait until their Commands end. A merged or closed pull request takes its row out of Ready for review. Each exit says why, with Resolved counted from when GitHub shows it settled. Rows GitHub can speak for keep waiting however old they are. Reopening brings back a row still in Done or Resolved; History keeps the last facts and is not read again. Dismissals hold through GitHub changes; Undo uses GitHub's latest kept facts. Rows with no issue or pull request, no Command record, no read yet or nothing GitHub can find still age out, straight to History.

    When GitHub cannot answer, each row keeps the group its last facts give it; a row never read still goes by age. After five minutes without a good read, the Console and status warn with the last-answer time and latest failure. A successful read clears that warning. With fewer than 500 GraphQL points or 500 REST requests left, both skip reads until that quota resets and say why, leaving the rest to the factory. status --json gives the last-answer time and latest failure. These limits are fixed, with no User config settings.

    status --json also gives each row's kept GitHub facts and when they were read. History keeps those observation times with its facts. A dismissed row stays in Resolved for inbox.days from its dismissal; a later GitHub settlement does not extend it.

    diff --git a/src/console/assets/console.js b/src/console/assets/console.js index 73211304a..6bee94f89 100644 --- a/src/console/assets/console.js +++ b/src/console/assets/console.js @@ -334,6 +334,11 @@ async function loadPeek(url) { pendingPeek?.abort(); const request = new AbortController(); pendingPeek = request; + let timedOut = false; + const timeout = setTimeout(() => { + timedOut = true; + request.abort(); + }, 10000); try { const response = await fetch(url, { signal: request.signal, cache: 'no-store' }); if (!response.ok) throw new Error('Details unavailable'); @@ -349,8 +354,11 @@ async function loadPeek(url) { if (content.querySelector('[data-live="true"]')) startUpdates(); else stopPeekUpdates(); } catch (error) { - if (!request.signal.aborted && error.name !== 'AbortError') content.textContent = 'Details unavailable. Reload to try again.'; + if (pendingPeek === request && (timedOut || (!request.signal.aborted && error.name !== 'AbortError'))) { + content.textContent = 'Details unavailable. Reload to try again.'; + } } finally { + clearTimeout(timeout); if (pendingPeek === request) pendingPeek = null; } } diff --git a/src/logs/reading.rs b/src/logs/reading.rs index c11be8992..f550fbd11 100644 --- a/src/logs/reading.rs +++ b/src/logs/reading.rs @@ -62,15 +62,38 @@ impl Snapshot { } pub(super) fn load_metrics(root: &Path, now: DateTime) -> Result { - Self::load_metrics_with_silence(root, now, std::time::Duration::from_secs(600)) + let mut snapshot = Self::capture_records( + Arc::new(inventory::Native), + root, + now, + std::time::Duration::from_secs(600), + true, + )?; + for entry in &mut snapshot.commands { + if let Entry::Record(command) = entry { + command.add_session_signals(None); + } + } + Ok(snapshot) } - pub(super) fn load_metrics_with_silence( + pub(super) fn load_details( root: &Path, now: DateTime, silent: std::time::Duration, + input: &str, ) -> Result { - Self::capture_records(Arc::new(inventory::Native), root, now, silent, true) + let mut snapshot = + Self::capture_records(Arc::new(inventory::Native), root, now, silent, true)?; + if let Observation::Record(selected) = snapshot.resolve(input)?.observation { + let identity = CommandIdentity::of(selected); + for entry in &mut snapshot.commands { + if let Entry::Record(command) = entry { + command.add_session_signals(Some(&identity)); + } + } + } + Ok(snapshot) } fn capture( @@ -257,6 +280,22 @@ pub(super) struct CurrentStage { } impl Command { + /// Enrich this Command when `selected` is every Command (`None`), or this + /// Command's exact identity. A match also enriches nested Ticket + /// runs and dispatches. Path alone is not enough: a dispatch shares its + /// parent's record path. Usual ranges and health already include every record. + fn add_session_signals(&mut self, selected: Option<&CommandIdentity>) { + let enrich = selected.is_none_or(|identity| identity.matches(self)); + if enrich && let Some(mut metrics) = self.record.run_metrics.take() { + metrics.add_session_signals(&self.record, cache::session_signals); + self.record.run_metrics = Some(metrics); + } + let nested = if enrich { None } else { selected }; + for child in self.dispatched.iter_mut().chain(&mut self.children) { + child.add_session_signals(nested); + } + } + fn new( inventory: &Arc, record: CommandRecord, @@ -328,12 +367,6 @@ fn read_records( if record.run_metrics.is_none() && record.outcome.is_some() { record.run_metrics = Some(super::run_metrics::RunMetrics::historical(record)); } - if let Some(mut metrics) = record.run_metrics.take() { - metrics.add_session_signals(record, |log| { - super::session_signals::read(log).ok().flatten() - }); - record.run_metrics = Some(metrics); - } } } let record = &mut tree.command; diff --git a/src/logs/reading/attribution.rs b/src/logs/reading/attribution.rs index 4d0b769fa..89dd4c768 100644 --- a/src/logs/reading/attribution.rs +++ b/src/logs/reading/attribution.rs @@ -15,6 +15,10 @@ pub(in crate::logs) struct CommandIdentity { } impl CommandIdentity { + pub(in crate::logs) fn matches(&self, command: &Command) -> bool { + self.record_path == command.record_path && self.name == command.record.name + } + pub(in crate::logs) fn of(command: &Command) -> Self { Self { record_path: command.record_path.clone(), diff --git a/src/logs/reading/cache.rs b/src/logs/reading/cache.rs index 2e6d9868b..4586cd731 100644 --- a/src/logs/reading/cache.rs +++ b/src/logs/reading/cache.rs @@ -1,16 +1,19 @@ -//! Bounded, process-local caches of retained facts. Liveness and health are -//! sampled afresh; only file contents/parsed records survive between requests. +//! Bounded caches of retained facts. Liveness and health are sampled afresh; +//! derived Session signals also survive between CLI processes. +mod retained_signals; use std::collections::BTreeMap; use std::fs; use std::path::{Path, PathBuf}; -use std::sync::{Arc, LazyLock, Mutex}; +use std::sync::{Arc, LazyLock, Mutex, Weak}; use std::time::{Instant, SystemTime}; use anyhow::{Context, Result}; +use serde::{Deserialize, Serialize}; use crate::logs::command_record::CommandTree; +use crate::logs::session_signals::SessionSignals; -#[derive(PartialEq)] +#[derive(PartialEq, Serialize, Deserialize)] struct Identity { len: u64, modified: Option, @@ -42,6 +45,7 @@ struct Retained { identity: Identity, value: T, used: Instant, + bytes: u64, } struct Cache { @@ -59,39 +63,60 @@ impl Cache { fn read(&mut self, path: &Path, load: impl FnOnce(&Path) -> Result) -> Result { let identity = Identity::read(path)?; + if let Some(value) = self.get(path, &identity) { + return Ok(value); + } + let value = load(path)?; + // Don't retain a partial in-place write. A later request retries it. + if Identity::read(path)? == identity { + let bytes = identity.len; + self.insert(path, identity, value.clone(), bytes); + } + Ok(value) + } + + /// Return a cached value. A different identity for this path is forgotten + /// so the caller can load and insert the current one. + fn get(&mut self, path: &Path, identity: &Identity) -> Option { if let Some(retained) = self.entries.get_mut(path) - && retained.identity == identity + && retained.identity == *identity { retained.used = Instant::now(); - return Ok(retained.value.clone()); + return Some(retained.value.clone()); } if let Some(old) = self.entries.remove(path) { - self.bytes -= old.identity.len; + self.bytes -= old.bytes; } - let value = load(path)?; - // Don't retain a partial in-place write. A later request retries it. + None + } + + fn insert(&mut self, path: &Path, identity: Identity, value: T, bytes: u64) { const BUDGET: u64 = 64 * 1024 * 1024; - if identity.len <= BUDGET && Identity::read(path)? == identity { - while self.bytes + identity.len > BUDGET || self.entries.len() >= 8192 { - let oldest = self - .entries - .iter() - .min_by_key(|(_, entry)| entry.used) - .map(|(path, _)| path.clone()) - .expect("nonempty cache over budget"); - self.bytes -= self.entries.remove(&oldest).unwrap().identity.len; - } - self.bytes += identity.len; - self.entries.insert( - path.to_path_buf(), - Retained { - identity, - value: value.clone(), - used: Instant::now(), - }, - ); + if bytes > BUDGET { + return; } - Ok(value) + if let Some(old) = self.entries.remove(path) { + self.bytes -= old.bytes; + } + while self.bytes + bytes > BUDGET || self.entries.len() >= 8192 { + let oldest = self + .entries + .iter() + .min_by_key(|(_, entry)| entry.used) + .map(|(path, _)| path.clone()) + .expect("nonempty cache over budget"); + self.bytes -= self.entries.remove(&oldest).unwrap().bytes; + } + self.bytes += bytes; + self.entries.insert( + path.to_path_buf(), + Retained { + identity, + value, + used: Instant::now(), + bytes, + }, + ); } } @@ -125,3 +150,74 @@ pub(super) fn text(path: &Path) -> Result> { Err(error) => Err(error).with_context(|| format!("could not read {}", path.display())), } } + +/// Retain only the derived signals, keyed by the resolved Session trace's +/// identity. Large source traces must not consume the summary cache's budget. +pub(super) fn session_signals(log: &Path) -> Option { + struct Signals { + cache: Cache>, + readers: BTreeMap>>, + } + static SIGNALS: LazyLock> = LazyLock::new(|| { + Mutex::new(Signals { + cache: Cache::new(), + readers: BTreeMap::new(), + }) + }); + let path = crate::logs::session::path(log); + let identity = Identity::read(&path).ok()?; + let reader = { + let mut signals = SIGNALS.lock().unwrap(); + if let Some(value) = signals.cache.get(&path, &identity) { + return value; + } + // Coordinate readers of this trace only. Parsing a cold trace must not + // block cached details for unrelated Commands. Weak entries disappear + // on the next miss once their readers have finished. + signals + .readers + .retain(|_, reader| reader.strong_count() > 0); + match signals.readers.get(&path).and_then(Weak::upgrade) { + Some(reader) => reader, + None => { + let reader = Arc::new(Mutex::new(())); + signals + .readers + .insert(path.clone(), Arc::downgrade(&reader)); + reader + } + } + }; + let _reading = reader.lock().unwrap(); + // A preceding reader or writer may have changed the facts while we waited. + let identity = Identity::read(&path).ok()?; + if let Some(value) = SIGNALS.lock().unwrap().cache.get(&path, &identity) { + return value; + } + // The file lock also coordinates cold readers in separate CLI processes. + // An unwritable cache is only a missed optimization, never missing evidence. + let retained = retained_signals::RetainedSignals::lock(&path).ok(); + let identity = Identity::read(&path).ok()?; + let cached = retained.as_ref().and_then(|cache| cache.read(&identity)); + let persist = cached.is_none(); + let value = match cached { + Some(value) => value, + // Readers expose a parse error as Missing. Retain that observation too; + // repairing the trace changes its identity and retries the parser. + None => crate::logs::session_signals::read(&path).ok().flatten(), + }; + if Identity::read(&path).ok().as_ref() == Some(&identity) { + if persist && let Some(retained) = retained { + let _ = retained.write(&identity, &value); + } + let bytes = serde_json::to_vec(&value) + .expect("Session signals serialize") + .len() as u64; + SIGNALS + .lock() + .unwrap() + .cache + .insert(&path, identity, value.clone(), bytes); + } + value +} diff --git a/src/logs/reading/cache/retained_signals.rs b/src/logs/reading/cache/retained_signals.rs new file mode 100644 index 000000000..5683a107d --- /dev/null +++ b/src/logs/reading/cache/retained_signals.rs @@ -0,0 +1,82 @@ +//! Disposable derived evidence beside a Session trace, never in its Command +//! record. Atomic replacement and a per-trace lock keep CLI readers coordinated. +use std::fs::{self, File}; +use std::io; +use std::path::{Path, PathBuf}; +use std::sync::LazyLock; + +use serde::{Deserialize, Serialize}; + +use super::Identity; +use crate::logs::session_signals::SessionSignals; + +#[derive(PartialEq, Serialize, Deserialize)] +struct Executable { + path: PathBuf, + identity: Identity, +} + +#[derive(Serialize, Deserialize)] +struct Saved { + executable: Executable, + trace: Identity, + signals: Option, +} + +pub(super) struct RetainedSignals { + path: PathBuf, + // Locks a separate stable inode, since the saved value is replaced atomically. + _lock: File, +} + +impl RetainedSignals { + pub(super) fn lock(trace: &Path) -> io::Result { + let lock = File::options() + .read(true) + .write(true) + .create(true) + .truncate(false) + .open(trace.with_extension("signals-cache.lock"))?; + lock.lock()?; + Ok(Self { + path: trace.with_extension("signals-cache.json"), + _lock: lock, + }) + } + + pub(super) fn read(&self, identity: &Identity) -> Option> { + let executable = executable()?; + let saved: Saved = serde_json::from_slice(&fs::read(&self.path).ok()?).ok()?; + (saved.trace == *identity && saved.executable == *executable).then_some(saved.signals) + } + + pub(super) fn write( + &self, + identity: &Identity, + signals: &Option, + ) -> anyhow::Result<()> { + // Include the exact executable identity so a change to signal semantics + // cannot silently reuse values derived by a previous binary. + let Some(executable) = executable() else { + return Ok(()); + }; + let temporary = tempfile::NamedTempFile::new_in(self.path.parent().unwrap())?; + serde_json::to_writer( + &temporary, + &serde_json::json!({ + "executable": executable, "trace": identity, "signals": signals + }), + )?; + temporary.persist(&self.path)?; + Ok(()) + } +} + +fn executable() -> Option<&'static Executable> { + static EXECUTABLE: LazyLock> = LazyLock::new(|| { + let path = std::env::current_exe().ok()?; + let identity = Identity::read(&path).ok()?; + Some(Executable { path, identity }) + }); + EXECUTABLE.as_ref() +} diff --git a/src/logs/reading/metrics_tests.rs b/src/logs/reading/metrics_tests.rs new file mode 100644 index 000000000..76f0c7a03 --- /dev/null +++ b/src/logs/reading/metrics_tests.rs @@ -0,0 +1,255 @@ +//! Observe actual trace opens at the Command-details and Snapshot seams. +use std::ffi::CString; +use std::io::Read; +use std::os::fd::FromRawFd; +use std::os::unix::ffi::OsStrExt; + +use super::*; + +struct Opens { + file: fs::File, + paths: BTreeMap, +} + +impl Opens { + fn new(paths: &[PathBuf]) -> Self { + // SAFETY: the successful descriptor is owned exactly once by File. + let fd = unsafe { libc::inotify_init1(libc::IN_NONBLOCK | libc::IN_CLOEXEC) }; + assert!(fd >= 0, "{}", std::io::Error::last_os_error()); + let file = unsafe { fs::File::from_raw_fd(fd) }; + let paths = paths + .iter() + .map(|path| { + let name = CString::new(path.as_os_str().as_bytes()).unwrap(); + // SAFETY: name is NUL terminated and fd remains owned above. + let wd = unsafe { libc::inotify_add_watch(fd, name.as_ptr(), libc::IN_OPEN) }; + assert!(wd >= 0, "{}", std::io::Error::last_os_error()); + (wd, path.clone()) + }) + .collect(); + Self { file, paths } + } + + fn take(&mut self) -> BTreeSet { + let mut opened = BTreeSet::new(); + let mut bytes = [0_u8; 8192]; + loop { + match self.file.read(&mut bytes) { + Ok(0) => break, + Ok(len) => { + let mut offset = 0; + while offset < len { + // SAFETY: inotify returns complete events; byte arrays + // need not have the alignment of inotify_event. + let event = unsafe { + bytes + .as_ptr() + .add(offset) + .cast::() + .read_unaligned() + }; + if event.mask & libc::IN_OPEN != 0 { + opened.insert(self.paths[&event.wd].clone()); + } + offset += std::mem::size_of::() + event.len as usize; + } + } + Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => break, + Err(error) => panic!("{error}"), + } + } + opened + } +} + +fn historical(root: &Path, number: u64) -> PathBuf { + let log = root.join(format!( + "acme/widgets/commands/issue/{number}-trace.logs.jsonl" + )); + fs::create_dir_all(log.parent().unwrap()).unwrap(); + fs::write(&log, "{\"resourceLogs\":[]}\n").unwrap(); + let mut past = record(number); + past["process_id"] = 0.into(); + past["outcome"] = "merged".into(); + past["ended_at"] = "2026-10-19T11:01:00-04:00".into(); + past["stages"] = json!([{ + "name":"implement", "began_at":"2026-10-19T11:00:00-04:00", "session_log":log + }]); + write(root, &format!("{number}-fixture.command.json"), &past); + log +} + +#[test] +fn command_details_only_read_selected_historical_session_traces() { + for count in [1, 2, 562] { + let dir = tempfile::tempdir().unwrap(); + let traces: Vec<_> = (1..=count).map(|n| historical(dir.path(), n)).collect(); + let mut opens = Opens::new(&traces); + let mut config = + crate::config::UserConfig::parse("", &dir.path().join("config.toml"), dir.path()) + .unwrap(); + config.logs_dir = dir.path().into(); + let inspection = crate::logs::stop::inspect(&config, "widgets#1").unwrap(); + assert_eq!( + inspection.details["run_metrics"]["session_signals"]["implement"]["observed_sessions"], + 1 + ); + assert_eq!( + opens.take(), + [traces[0].clone()].into(), + "Command details read unrelated historical Session traces with {count} records" + ); + } +} + +#[test] +fn metrics_reuse_unchanged_session_traces_and_refresh_changed_traces() { + use std::io::Write; + + let dir = tempfile::tempdir().unwrap(); + let traces: Vec<_> = (1..=4).map(|n| historical(dir.path(), n)).collect(); + // The source can exceed the entire cache budget; only its compact derived + // signals should count toward that budget. + let mut large = fs::File::create(&traces[3]).unwrap(); + let padding = vec![b' '; 1024 * 1024]; + for _ in 0..65 { + large.write_all(&padding).unwrap(); + } + large.write_all(b"{\"resourceLogs\":[]}\n").unwrap(); + drop(large); + let mut opens = Opens::new(&traces); + let first = Snapshot::load_metrics(dir.path(), now()).unwrap(); + assert_eq!(opens.take(), traces.iter().cloned().collect()); + let second = Snapshot::load_metrics(dir.path(), now()).unwrap(); + assert!( + opens.take().is_empty(), + "unchanged Session traces were reparsed" + ); + assert_eq!( + serde_json::to_value(first.resolve("widgets#1").unwrap().observation).unwrap(), + serde_json::to_value(second.resolve("widgets#1").unwrap().observation).unwrap() + ); + + let amendment = |version: &str| { + serde_json::to_string(&json!({ + "resourceLogs": [{"resource": {"attributes": [ + {"key": "thirdshift.harness.version", "value": {"stringValue": version}} + ]}, "scopeLogs": []}] + })) + .unwrap() + + "\n" + }; + fs::OpenOptions::new() + .append(true) + .open(&traces[0]) + .unwrap() + .write_all(amendment("1.0").as_bytes()) + .unwrap(); + opens.take(); + let appended = Snapshot::load_metrics(dir.path(), now()).unwrap(); + assert_eq!(opens.take(), [traces[0].clone()].into()); + let signals = |snapshot: &Snapshot| { + serde_json::to_value(snapshot.resolve("widgets#1").unwrap().observation).unwrap() + ["run_metrics"]["stages"][0]["session_signals"].clone() + }; + assert_eq!(signals(&appended)["harness_version"], "1.0"); + + // Same length and mtime, but a new inode must invalidate the old signals. + let modified = fs::metadata(&traces[0]).unwrap().modified().unwrap(); + let replacement = dir.path().join("replacement.logs.jsonl"); + fs::write( + &replacement, + "{\"resourceLogs\":[]}\n".to_owned() + &amendment("2.0"), + ) + .unwrap(); + fs::File::open(&replacement) + .unwrap() + .set_times(fs::FileTimes::new().set_modified(modified)) + .unwrap(); + fs::rename(replacement, &traces[0]).unwrap(); + let mut replaced_opens = Opens::new(&traces); + let replaced = Snapshot::load_metrics(dir.path(), now()).unwrap(); + assert_eq!(replaced_opens.take(), [traces[0].clone()].into()); + assert_eq!(signals(&replaced)["harness_version"], "2.0"); +} + +#[test] +fn command_details_keep_dispatch_and_child_signals_and_other_commands_health() { + for (selected, expected) in [("widgets#1", vec![1, 2, 3]), ("widgets#2", vec![2, 3])] { + let dir = tempfile::tempdir().unwrap(); + let traces: Vec<_> = (1..=5).map(|n| historical(dir.path(), n)).collect(); + let path = |n| { + dir.path().join(format!( + "acme/widgets/commands/issue/{n}-fixture.command.json" + )) + }; + let read = |n| serde_json::from_slice::(&fs::read(path(n)).unwrap()).unwrap(); + let mut outer = read(1); + outer["dispatched"] = json!([read(2)]); + fs::write(path(1), serde_json::to_vec(&outer).unwrap()).unwrap(); + fs::remove_file(path(2)).unwrap(); + let mut child = read(3); + child["parent_record"] = json!(path(1)); + child["parent_dispatch"] = "https://github.com/acme/widgets/issues/2".into(); + fs::write(path(3), serde_json::to_vec(&child).unwrap()).unwrap(); + let mut live = record(5); + live["stages"] = json!([{"name": "implement", "began_at": "2026-10-19T11:00:00-04:00"}]); + fs::write(path(5), serde_json::to_vec(&live).unwrap()).unwrap(); + + let before = Snapshot::load(dir.path(), now()).unwrap(); + let mut opens = Opens::new(&traces); + let details = Snapshot::load_details( + dir.path(), + now(), + std::time::Duration::from_secs(600), + selected, + ) + .unwrap(); + assert_eq!( + opens.take(), + expected.iter().map(|n| traces[n - 1].clone()).collect() + ); + for number in [1, 2, 3] { + let value = serde_json::to_value( + details + .resolve(&format!("widgets#{number}")) + .unwrap() + .observation, + ) + .unwrap(); + if expected.contains(&number) { + assert_eq!( + value["run_metrics"]["session_signals"]["implement"]["observed_sessions"], + 1 + ); + } + } + let observation = |snapshot: &Snapshot| { + serde_json::to_value(snapshot.resolve("widgets#5").unwrap().observation).unwrap() + }; + assert_eq!( + observation(&before)["health"], + observation(&details)["health"] + ); + assert_eq!( + observation(&before)["usual_ranges"], + observation(&details)["usual_ranges"] + ); + assert!(!observation(&before)["usual_ranges"]["implement"].is_null()); + } +} + +#[test] +fn saved_session_signals_do_not_read_the_historical_trace() { + let dir = tempfile::tempdir().unwrap(); + let trace = historical(dir.path(), 1); + let snapshot = Snapshot::load_metrics(dir.path(), now()).unwrap(); + let value = serde_json::to_value(snapshot.resolve("widgets#1").unwrap().observation).unwrap(); + write(dir.path(), "1-fixture.command.json", &value); + fs::write(&trace, "invalid trace, saved signals must win\n").unwrap(); + let mut opens = Opens::new(&[trace]); + let snapshot = Snapshot::load_metrics(dir.path(), now()).unwrap(); + assert!(opens.take().is_empty()); + let saved = serde_json::to_value(snapshot.resolve("widgets#1").unwrap().observation).unwrap(); + assert_eq!(saved["run_metrics"], value["run_metrics"]); +} diff --git a/src/logs/reading/tests.rs b/src/logs/reading/tests.rs index 0390e5d92..9b3d4b7ad 100644 --- a/src/logs/reading/tests.rs +++ b/src/logs/reading/tests.rs @@ -14,6 +14,10 @@ mod inventory_tests; #[path = "reader_tests.rs"] mod reader_tests; +#[cfg(target_os = "linux")] +#[path = "metrics_tests.rs"] +mod metrics_tests; + fn now() -> DateTime { DateTime::parse_from_rfc3339("2026-10-19T12:00:00-04:00") .unwrap() diff --git a/src/logs/status/inspect.rs b/src/logs/status/inspect.rs index 391afd81a..7cf646c19 100644 --- a/src/logs/status/inspect.rs +++ b/src/logs/status/inspect.rs @@ -96,10 +96,11 @@ fn print_details(command: &Command, indent: &str) -> Result<()> { pub fn show(input: &str, json: bool) -> Result<()> { let config = UserConfig::load()?; - let snapshot = Snapshot::load_metrics_with_silence( + let snapshot = Snapshot::load_details( &config.logs_dir, Local::now(), config.silent_interval, + input, )?; let command = snapshot.resolve(input)?.observation; if json { diff --git a/src/logs/stop.rs b/src/logs/stop.rs index 532c0d30b..629c79166 100644 --- a/src/logs/stop.rs +++ b/src/logs/stop.rs @@ -118,10 +118,11 @@ pub(crate) struct Inspection { /// Availability follows the CLI's child refusal and captured process identity. pub(crate) fn inspect(config: &UserConfig, input: &str) -> Result { - let snapshot = Snapshot::load_metrics_with_silence( + let snapshot = Snapshot::load_details( &config.logs_dir, chrono::Local::now(), config.silent_interval, + input, )?; let selected = snapshot.resolve(input)?; let pending = selected diff --git a/tests/metrics.rs b/tests/metrics.rs index ee7722618..84aec09ad 100644 --- a/tests/metrics.rs +++ b/tests/metrics.rs @@ -7,6 +7,131 @@ use support::Scenario; const LOGS: &str = "home/.thirdshift/logs/acme/widgets/commands/issue"; +#[cfg(target_os = "linux")] +struct TraceOpens(fs::File); + +#[cfg(target_os = "linux")] +impl TraceOpens { + fn new(log: &std::path::Path) -> Self { + use std::os::fd::FromRawFd; + use std::os::unix::ffi::OsStrExt; + // SAFETY: File owns the successful descriptor; name is NUL terminated. + let fd = unsafe { libc::inotify_init1(libc::IN_NONBLOCK | libc::IN_CLOEXEC) }; + assert!(fd >= 0); + let file = unsafe { fs::File::from_raw_fd(fd) }; + let name = std::ffi::CString::new(log.as_os_str().as_bytes()).unwrap(); + assert!(unsafe { libc::inotify_add_watch(fd, name.as_ptr(), libc::IN_OPEN) } >= 0); + Self(file) + } + + fn take(&mut self) -> bool { + use std::io::Read; + let mut buffer = [0; 1024]; + match self.0.read(&mut buffer) { + Ok(len) => len > 0, + Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => false, + Err(error) => panic!("{error}"), + } + } +} + +#[cfg(target_os = "linux")] +#[test] +fn unchanged_malformed_session_signals_are_not_reparsed_and_repair_refreshes_them() { + let scenario = Scenario::new(); + let log = trace(&scenario, "malformed", "claude", "1.0", json!([])); + fs::write(&log, "{\"resourceLogs\":[]}\n{invalid}\n").unwrap(); + let mut past = record(7); + past["stages"][0]["session_log"] = json!(log); + write(&scenario, "7-fixture.command.json", &past); + let mut opens = TraceOpens::new(&log); + let first = metrics(&scenario, &["metrics", "widgets#7", "--json"]); + assert!(first["command"]["metrics"]["stages"][0]["session_signals"].is_null()); + assert!(opens.take()); + let second = metrics(&scenario, &["metrics", "widgets#7", "--json"]); + assert!( + !opens.take(), + "unchanged malformed Session trace was reparsed" + ); + assert_eq!(first, second); + trace(&scenario, "malformed", "claude", "2.0", json!([])); + opens.take(); + let repaired = metrics(&scenario, &["metrics", "widgets#7", "--json"]); + assert!(opens.take()); + assert_eq!( + repaired["command"]["metrics"]["stages"][0]["session_signals"]["harness_version"], + "2.0" + ); +} + +#[cfg(target_os = "linux")] +#[test] +fn unchanged_session_signals_survive_between_cli_processes() { + use std::io::Write; + + let scenario = Scenario::new(); + let log = trace(&scenario, "historical", "claude", "1.0", json!([])); + let mut past = record(7); + past["stages"][0]["session_log"] = json!(log); + let record = write(&scenario, "7-fixture.command.json", &past); + let original_record = fs::read(&record).unwrap(); + let original_trace = fs::read_to_string(&log).unwrap(); + let mut watcher = TraceOpens::new(&log); + let first = metrics(&scenario, &["metrics", "widgets#7", "--json"]); + assert!(watcher.take()); + let second = metrics(&scenario, &["metrics", "widgets#7", "--json"]); + assert!( + !watcher.take(), + "a second CLI process reparsed an unchanged historical Session trace" + ); + assert_eq!(first, second); + + let amendment = |version: &str| { + serde_json::to_string(&json!({"resourceLogs": [{"resource": {"attributes": [ + {"key": "thirdshift.harness.version", "value": {"stringValue": version}} + ]}, "scopeLogs": []}]})) + .unwrap() + + "\n" + }; + fs::OpenOptions::new() + .append(true) + .open(&log) + .unwrap() + .write_all(amendment("2.0").as_bytes()) + .unwrap(); + watcher.take(); + let appended = metrics(&scenario, &["metrics", "widgets#7", "--json"]); + assert!(watcher.take()); + assert_eq!( + appended["command"]["metrics"]["stages"][0]["session_signals"]["harness_version"], + "2.0" + ); + + // Replacing a trace with the same length and mtime must still refresh it. + let modified = fs::metadata(&log).unwrap().modified().unwrap(); + let replacement = log.with_extension("replacement"); + fs::write(&replacement, original_trace + &amendment("3.0")).unwrap(); + fs::File::open(&replacement) + .unwrap() + .set_times(fs::FileTimes::new().set_modified(modified)) + .unwrap(); + fs::rename(replacement, &log).unwrap(); + watcher = TraceOpens::new(&log); + let replaced = metrics(&scenario, &["metrics", "widgets#7", "--json"]); + assert!(watcher.take()); + assert_eq!( + replaced["command"]["metrics"]["stages"][0]["session_signals"]["harness_version"], + "3.0" + ); + let shown = metrics(&scenario, &["show", "widgets#7", "--json"]); + assert!(!watcher.take()); + assert_eq!( + shown["command"]["run_metrics"]["session_signals"], + replaced["command"]["metrics"]["session_signals"] + ); + assert_eq!(fs::read(record).unwrap(), original_record); +} + fn trace( scenario: &Scenario, name: &str, diff --git a/tests/skill.rs b/tests/skill.rs index 6bc9a8aaa..7e15999e5 100644 --- a/tests/skill.rs +++ b/tests/skill.rs @@ -413,7 +413,7 @@ fn partial_muse_and_opencode_messages_are_not_success() { fs::remove_file(&executable).unwrap(); fs::write( &executable, - format!("#!/bin/sh\ncat <<'JSON'\n{event}\nJSON\nexit 0\n"), + format!("#!/bin/sh\ncat > /dev/null\ncat <<'JSON'\n{event}\nJSON\nexit 0\n"), ) .unwrap(); fs::set_permissions(&executable, fs::Permissions::from_mode(0o755)).unwrap();