diff --git a/CHANGELOG.md b/CHANGELOG.md index 6e66bb0b..27a0b765 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,15 @@ that section and uses it as the GitHub release description, which the app displa as-is. The install instructions block (after the `` marker) is added by `release.yml` and stays **only** on the GitHub page — it does not appear in the app. +## v2.7.0 + +- New: a **customizable conversation side panel** — choose its widgets (git status, stats with the total tokens of every agent, linked conversations, plan usage, the machine it runs on…), reorder and fold them, or start from a preset. It can be resized, and can fit its content. +- New: an opt-in **telemetry deck** in the side panel — live gauges, clocks and the token mix of the running agent. +- New: a cleaner **title bar** — underlined view tabs and one tray for voice, sound and Caffeinate. +- New: an adjustable **conversation width**, and Display settings sorted into clearer cards. +- Fix: the **turn duration** now times the whole run, background tasks included, instead of resetting. +- Fix: **artifact icons** show again and older versions are reachable; **less battery drain** at rest (idle wake-ups, remote servers, wake word). + ## v2.6.0 - New: **per-tool permissions for MCP tools** — Allow / Ask / Block each tool, and choose where it applies (this conversation, the repository, or everywhere). Extensions now live under Settings → Claude Code. diff --git a/CLAUDE.md b/CLAUDE.md index 777826a0..dbf18186 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -196,6 +196,7 @@ Encapsulation stricte (cf. « Patterns ») : un module = un service, swappable s - **id stable** (UUID PK persistée) ≠ **handle live** (`session-N`, mémoire, non persisté) : front keyé par id stable en LECTURE, handle résolu à l'envoi. - Spawn **paresseux** (rien au démarrage ; historique lu du transcript ; `--resume` si `sessionId`). Teardown **sans orphelins** (`process_group(0)`, `kill(-pid)`, échelle EOF→SIGTERM→SIGKILL, kill-all borné au quit). - **Rewind / Fork** (`history.rs`) : rewind TRONQUE le transcript on-disk (destructif), fork COPIE sous nouveau `session_id`, toujours à une frontière de **prompt humain** (jamais de `tool_use` orphelin), puis re-spawn `--resume`. ⚠️ Toute mutation du transcript exige un arrêt **SYNCHRONE** (`SessionHandle::shutdown_and_wait()`, PAS `shutdown()`), sinon course contre un writer vivant → corruption. ⚠️ Tours live = id synthétique (`user_N`) absent du disque → ciblage par TEXTE (`prompt_match_key`, miroir Rust/TS) + index d'occurrence. Codex : voie native par turn id (`thread/fork{lastTurnId}`). +- **Sessions REMOTE (SSH) — sobriété** (audit batterie 28/09) : keepalive `ServerAliveInterval=10` ×3 par conversation (coupure vue en ~30 s — décision : ne pas redescendre à 5 s, c'était 2× plus de réveils radio au repos). Reconnexion : backoff 1 s doublé, plafond 30 s pendant 10 échecs puis ≤ 5 min (`next_reconnect_delay`), court-circuité par `reconnect_remote_sessions` (front : `online` / focus / visibilité quand un lien est down, `MachineHealthHost`) et par un envoi pendant la coupure (`SessionCommand::ReconnectNow`). Santé des serveurs : la sonde ambiante est `machine_reachability` (`ssh … true`, même classification/raison que `diagnose` via `unreachable_after`) — le diagnostic complet reste réservé au panneau Réglages ; ⚠️ JAMAIS de sonde vers une machine à laquelle une session est attachée (`attachedMachineIds` : son lien prouve déjà la joignabilité, filé via `recordReachability`). Caffeinate (Light) ne compte que les agents LOCAUX (`localConversations`) : un agent distant continue sous `flightdeckd` pendant que le Mac dort. Piste ouverte : une connexion ssh par serveur (`ControlMaster`, tâche ed7ec860). **Worktrees** - Outils natifs `EnterWorktree`/`ExitWorktree` interceptés dans `useGlobalSessionEvents`. Convention `.claude/worktrees/`. cwd non figé ; association conv↔worktree par longest-prefix. @@ -333,8 +334,7 @@ L'app HÉBERGE deux serveurs MCP. Module `src-tauri/src/appmcp/` (hub, router JS Agent vocal in-app + déclenchement mains-libres, 100 % local pour le réveil. **OpenAI strictement optionnel** : sans clé l'app marche, la clé ne déverrouille que le vocal. Front `src/voice/` + Rust `src-tauri/src/voice/` (SEUL détenteur de la clé OpenAI) + `src-tauri/src/wake/` (wake word local). - **Voix (OpenAI Realtime, WebRTC/GA)** : `voice/realtime.ts` = session hors React (comme `termManager` : les composants rendent l'état de `voiceStore`, ne possèdent pas la connexion). Modèle 2 couches : **mode armé** (session ouverte, micro FERMÉ, exchange nul → reste up pendant le travail) ↔ **micro** ouvert/fermé DANS la session. Rust forge des **secrets éphémères** (`/v1/realtime/client_secrets`) → le webview ne voit jamais la clé (Keychain « Flight Deck OpenAI » suffixé par identité de build). Outils = sous-ensemble du catalogue appmcp (`VOICE_TOOL_NAMES`, dont `get_pending_request`/`answer_request` pour répondre aux questions/permissions par la voix) + `end_call`. **Annonces proactives** = mêmes events settled que les notifs OS (`announce.ts`, file plafonnée, drain séquentiel, jamais par-dessus une réponse active). Garde coût : auto-close sur silence RÉEL (réarmé sur l'activité vocale VAD, jamais mid-monologue). Un refus micro sur une annonce dégrade en « speak-only » (ne détruit pas la session). Réglages → Control carte **« Ground Control »** : clé, annonces, PTT (⌘ droit / ⌘⇧V), seuil VAD, « Answer by voice » (opt-in ON), wake word. - **Wake word** (`wake/`, openWakeWord + Silero VAD via `ort`/ONNX statique, modèles bundlés `include_bytes!`) : phrases `alexa`/`hey_jarvis` (pré-entraînées, défaut) + **« Ground Control »** (custom, entraîné LOCALEMENT via macOS `say`, sans cloud). Pipeline en détection continue → `WakeWordEvent` → `VoiceHost` ouvre le micro. -- ⚠️ Verrous release V2 encore ouverts : `getUserMedia` en WKWebView à confirmer ; robustesse du wake sur voix humaine réelle (accent FR) non prouvée ; follow-ups CPU (re-gate duty-cycle) / faux positifs du wake. Détail complet : mémoire `voice-agent-realtime-design` (+ `wake-word-custom-phrase-training`, `flightdeck-v2-crm-structure`). - +- ⚠️ Verrous release V2 encore ouverts : `getUserMedia` en WKWebView à confirmer ; robustesse du wake sur voix humaine réelle (accent FR) non prouvée ; follow-ups CPU (re-gate duty-cycle — tenté puis RETIRÉ le 28/09 : son rattrapage au début de la parole est invérifiable sans vrai micro ; en place : callbacks micro regroupés par pas de 80 ms, worker QoS UTILITY, wake word EN PAUSE sans clé OpenAI) / faux positifs du wake. Détail complet : mémoire `voice-agent-realtime-design` (+ `wake-word-custom-phrase-training`, `flightdeck-v2-crm-structure`). ## [GENERATED] Associated Project Contexts @@ -460,4 +460,4 @@ Alexandre Josien et Armand Mounsi, deux ingénieurs informatique freelances trav ## Ressources techniques - Abonnement Max Claude Code - Clé API OpenAI -- Hébergement Railway +- Hébergement Railway \ No newline at end of file diff --git a/package.json b/package.json index e504e482..ed8b987a 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "tosse-code", - "version": "2.6.0", + "version": "2.7.0", "private": true, "type": "module", "scripts": { diff --git a/src-tauri/Cargo.lock b/src-tauri/Cargo.lock index 4eeae080..d6859459 100644 --- a/src-tauri/Cargo.lock +++ b/src-tauri/Cargo.lock @@ -5233,7 +5233,7 @@ checksum = "756daf9b1013ebe47a8776667b466417e2d4c5679d441c26230efd9ef78692db" [[package]] name = "tosse-code" -version = "2.6.0" +version = "2.7.0" dependencies = [ "base64 0.22.1", "cpal", diff --git a/src-tauri/Cargo.toml b/src-tauri/Cargo.toml index 1fbbe986..56e93ea8 100644 --- a/src-tauri/Cargo.toml +++ b/src-tauri/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "tosse-code" -version = "2.6.0" +version = "2.7.0" description = "Flight Deck — desktop app to pilot Claude Code" authors = ["Tosse"] edition = "2021" @@ -129,8 +129,9 @@ specta-typescript = "=0.0.9" # agent's microphone (see `wake/mod.rs`). 100% on-device: NO network, NO cloud, # the captured audio never leaves the Mac. `cpal` captures the default input # device; `ort` (ONNX Runtime) runs the openWakeWord pipeline (melspectrogram → -# shared speech embedding → per-phrase classifier), gated by a Silero VAD so the -# neural nets run ONLY on speech (the battery guard). All five ONNX models are +# shared speech embedding → per-phrase classifier) plus a Silero VAD, which vetoes +# a fire unless it heard speech. The whole pipeline runs on every 80 ms step while +# the wake word is on — no silence-based battery guard. All six ONNX models are # bundled into the binary (`include_bytes!` from `assets/wake/`), so the feature # works offline on first launch — no download, no external asset host. `ort` is an # RC, pinned EXACTLY like the specta/tauri-specta RCs above. diff --git a/src-tauri/src/bootstrap/orchestrator.rs b/src-tauri/src/bootstrap/orchestrator.rs index 94eb0a8b..17ecd208 100644 --- a/src-tauri/src/bootstrap/orchestrator.rs +++ b/src-tauri/src/bootstrap/orchestrator.rs @@ -2015,12 +2015,7 @@ pub(crate) async fn diagnose(machine: &MachineRecord, known_hosts: Option<&str>) // The `Err` is deliberately discarded — it embeds the raw offending value (see // `validate_ssh_user`/`validate_address_value`'s own docs) and this reason // string is user-facing. Same discipline as `TransportError::InvalidRemoteTarget`. - return ServerDiagnosis { - state: DiagnosisState::Failed { - reason: "this server's saved connection details are not valid — remove and re-add it".to_string(), - }, - ..ServerDiagnosis::unreachable() - }; + return invalid_connection_details(); } cmd.arg(diagnose_script()); // A wedged remote shell (stuck lock, hung `flightdeckd`) must not hang this @@ -2032,6 +2027,29 @@ pub(crate) async fn diagnose(machine: &MachineRecord, known_hosts: Option<&str>) Ok(Ok(out)) if out.status.success() => { parse_diagnosis(&String::from_utf8_lossy(&out.stdout), true) } + failed => unreachable_after(&machine.host, failed).await, + } +} + +/// The diagnosis of a machine whose saved `user`/`host` no longer pass validation — +/// shared by [`diagnose`] and [`probe_reachability`], so both name it in the same words. +fn invalid_connection_details() -> ServerDiagnosis { + ServerDiagnosis { + state: DiagnosisState::Failed { + reason: "this server's saved connection details are not valid — remove and re-add it".to_string(), + }, + ..ServerDiagnosis::unreachable() + } +} + +/// Classify an ssh round trip that did NOT succeed — shared by [`diagnose`] and +/// [`probe_reachability`], so the ambient probe says WHY in exactly the words the full +/// diagnosis in the server panel uses. +async fn unreachable_after( + host: &str, + failed: Result, tokio::time::error::Elapsed>, +) -> ServerDiagnosis { + match failed { // ssh itself failed (a non-zero exit — OpenSSH's own convention for exit 255, // though anything non-zero here means the same thing for THIS script, which // never returns non-zero on its own). Thread the stderr this used to discard @@ -2042,7 +2060,7 @@ pub(crate) async fn diagnose(machine: &MachineRecord, known_hosts: Option<&str>) String::from_utf8_lossy(&out.stderr).lines().map(str::to_string).collect(); let issue = ssh_link::classify_transport_close(out.status.code(), &stderr_lines) .unwrap_or(SshLinkIssue::Unreachable); - let tailscale_off = tailscale_off_locally_if_relevant(&machine.host).await; + let tailscale_off = tailscale_off_locally_if_relevant(host).await; ServerDiagnosis::unreachable_with(issue, tailscale_off) } // No `Output` at all: the round trip either couldn't even be spawned, or the @@ -2051,12 +2069,65 @@ pub(crate) async fn diagnose(machine: &MachineRecord, known_hosts: Option<&str>) // state (a wedged/hung remote shell on a tailnet host is exactly the shape // this clause exists for). Ok(Err(_)) | Err(_) => { - let tailscale_off = tailscale_off_locally_if_relevant(&machine.host).await; + let tailscale_off = tailscale_off_locally_if_relevant(host).await; ServerDiagnosis::unreachable_with(SshLinkIssue::Unreachable, tailscale_off) } } } +/// The one fact the AMBIENT machine-health probe needs: can this Mac reach the server +/// right now, and if not, why. See [`probe_reachability`]. +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, Type)] +pub struct MachineReachability { + pub reachable: bool, + /// The diagnosis's own wording for why, when unreachable — the SAME string a full + /// [`diagnose`] would put in [`DiagnosisState::Failed`]. `None` when reachable. + pub reason: Option, +} + +impl MachineReachability { + fn reached() -> Self { + Self { reachable: true, reason: None } + } +} + +impl From<&ServerDiagnosis> for MachineReachability { + fn from(d: &ServerDiagnosis) -> Self { + let reason = match &d.state { + DiagnosisState::Failed { reason } if !d.reachable => Some(reason.clone()), + _ => None, + }; + Self { reachable: d.reachable, reason } + } +} + +/// "Can I talk to this machine?" — the question the sidebar mark, the Flight Deck lane +/// header and the composer band answer, and NOTHING more. +/// +/// ⚠️ Why not [`diagnose`]: the ambient loop used to run the full diagnosis every 90 s +/// per server, i.e. a fresh ssh handshake PLUS a dozen server-side commands — `systemctl`, +/// `loginctl`, `flightdeckd --version`/`status`, and `claude --version` + `claude auth +/// status`, two Node.js start-ups on the server — to paint a glyph that reads +/// `reachable` and nothing else (see `store/machineHealth.ts`). The server panel still +/// runs [`diagnose`]: it is the one surface that shows the rest. +/// +/// Same ssh options, same timeout, same failure classification as [`diagnose`] (shared +/// through [`unreachable_after`]), so a verdict never depends on which probe produced it. +pub(crate) async fn probe_reachability(machine: &MachineRecord, known_hosts: Option<&str>) -> MachineReachability { + let mut cmd = crate::ipc::commands::keyed_ssh_options(machine.port, machine.identity_file.as_deref(), known_hosts); + cmd.arg("-T"); + if crate::ipc::commands::push_ssh_destination(&mut cmd, &machine.user, &machine.host).is_err() { + return MachineReachability::from(&invalid_connection_details()); + } + // `true`, not an empty command: ssh with no command opens an interactive login + // shell and waits on it. Exit 0 means ssh connected, authenticated and ran it. + cmd.arg("true"); + match tokio::time::timeout(SSH_ROUND_TRIP_TIMEOUT, cmd.output()).await { + Ok(Ok(out)) if out.status.success() => MachineReachability::reached(), + failed => MachineReachability::from(&unreachable_after(&machine.host, failed).await), + } +} + /// [`tailscale::local_status`], gated on [`tailscale::host_looks_like_tailnet`] first /// so a LAN/public server's diagnosis never pays for the subprocess at all — folds /// into the one signal [`ServerDiagnosis::tailscale_off_locally`] ever carries: @@ -2106,6 +2177,16 @@ pub async fn machine_diagnose(app: tauri::AppHandle, machine_id: String) -> Resu Ok(with_bundled_version(&app, diagnose(&machine, known_hosts.as_deref()).await)) } +/// The ambient health probe — see [`probe_reachability`] for why it is not +/// [`machine_diagnose`]. +#[tauri::command] +#[specta::specta] +pub async fn machine_reachability(app: tauri::AppHandle, machine_id: String) -> Result { + let machine = machine_by_id(&app, &machine_id)?; + let known_hosts = known_hosts_path(&app); + Ok(probe_reachability(&machine, known_hosts.as_deref()).await) +} + // ============================================================================ // repair // ============================================================================ @@ -2967,6 +3048,43 @@ mod tests { assert_eq!(not_true.state, DiagnosisState::Failed { reason: "could not reach the server".to_string() }); } + /// The ambient probe must say WHY in the full diagnosis's own words — the mark's + /// tooltip and the server panel can never disagree on the reason. + #[test] + fn reachability_carries_the_diagnosis_reason_verbatim() { + for (issue, tailscale_off) in [ + (SshLinkIssue::KeyRefused, None), + (SshLinkIssue::HostKeyChanged, None), + (SshLinkIssue::Unreachable, None), + (SshLinkIssue::Unreachable, Some(true)), + ] { + let d = ServerDiagnosis::unreachable_with(issue, tailscale_off); + let DiagnosisState::Failed { reason } = &d.state else { panic!("unreachable is Failed") }; + assert_eq!( + MachineReachability::from(&d), + MachineReachability { reachable: false, reason: Some(reason.clone()) }, + ); + } + let invalid = MachineReachability::from(&invalid_connection_details()); + assert!(!invalid.reachable); + assert_eq!( + invalid.reason.as_deref(), + Some("this server's saved connection details are not valid — remove and re-add it"), + ); + assert_eq!(MachineReachability::reached(), MachineReachability { reachable: true, reason: None }); + } + + /// A REACHABLE server whose daemon is stopped is `Failed` too — but it must never + /// read as unreachable through the ambient probe's shape (same prudence as + /// `store/machineHealth.ts`, which only ever looks at `reachable`). + #[test] + fn reachability_of_a_reachable_failed_diagnosis_has_no_reason() { + let mut d = ServerDiagnosis::unreachable(); + d.reachable = true; + d.state = DiagnosisState::Failed { reason: "flightdeckd is not running".to_string() }; + assert_eq!(MachineReachability::from(&d), MachineReachability { reachable: true, reason: None }); + } + #[test] fn plain_unreachable_delegates_to_unreachable_with() { let d = ServerDiagnosis::unreachable(); diff --git a/src-tauri/src/git/mod.rs b/src-tauri/src/git/mod.rs index 2920e387..0f67316c 100644 --- a/src-tauri/src/git/mod.rs +++ b/src-tauri/src/git/mod.rs @@ -483,7 +483,10 @@ fn parse_worktree_list(porcelain: &str) -> Vec { /// Status of a single worktree: dirtiness (from `git status --porcelain`) and /// the ahead/behind count against the branch's upstream (`None` when unset). pub fn worktree_status(worktree_path: &str) -> Result { - let porcelain = run_git(worktree_path, &["status", "--porcelain"])?; + // ⚠️ `--no-optional-locks` (top-level, before the subcommand): a read-only status must + // not take `.git/index.lock` from under the agent's own `git add` / `git commit` — see + // `status::STATUS_ARGS`. + let porcelain = run_git(worktree_path, &["--no-optional-locks", "status", "--porcelain"])?; let mut status = WorktreeStatus::default(); for line in porcelain.lines() { if line.is_empty() { diff --git a/src-tauri/src/git/status.rs b/src-tauri/src/git/status.rs index 3126a9c8..7c1a644f 100644 --- a/src-tauri/src/git/status.rs +++ b/src-tauri/src/git/status.rs @@ -37,20 +37,36 @@ pub struct GitStatus { pub head: Option, /// Upstream ref short name (e.g. `origin/main`); `None` when unset. pub upstream: Option, - /// Commits ahead of upstream. + /// Commits ahead of upstream. ⚠️ Also `0` when there is NO upstream or it is GONE: + /// read it only when `upstream` is set and `upstream_gone` is false. pub ahead: u32, - /// Commits behind upstream. + /// Commits behind upstream. Same caveat as `ahead`. pub behind: u32, + /// An upstream is configured but its commit is missing — typically the remote branch + /// was deleted and pruned. git then prints `branch.upstream` WITHOUT `branch.ab`, so + /// `ahead`/`behind` are not counts but the struct's defaults: without this flag a gone + /// upstream reads as a fake "in sync" 0/0. + pub upstream_gone: bool, /// The branch has no commits yet (unborn HEAD). pub unborn: bool, /// Changed entries (staged + unstaged + untracked), in git's order. pub files: Vec, } -/// Status of the working tree at `cwd`. Uses porcelain v2 with `-z` so filenames -/// with spaces/newlines and rename pairs parse unambiguously. +/// Arguments of the status read. Porcelain v2 with `-z` so filenames with +/// spaces/newlines and rename pairs parse unambiguously. +/// +/// ⚠️ `--no-optional-locks` comes FIRST (a top-level git option, so it must precede the +/// subcommand): a plain `git status` opportunistically refreshes the index and writes it +/// back under `.git/index.lock`. This read runs in the BACKGROUND (panel refreshes, turn +/// ends) while the agent may be running `git add` / `git commit` in the same checkout; +/// holding that lock, even briefly, makes the AGENT's command fail with "Unable to create +/// …/index.lock: File exists". A read must never break a write. +const STATUS_ARGS: &[&str] = &["--no-optional-locks", "status", "--porcelain=v2", "--branch", "-z"]; + +/// Status of the working tree at `cwd` (see [`STATUS_ARGS`]). pub fn status(cwd: &str) -> Result { - let out = run_git(cwd, &["status", "--porcelain=v2", "--branch", "-z"])?; + let out = run_git(cwd, STATUS_ARGS)?; Ok(parse_status_v2(&out)) } @@ -75,13 +91,16 @@ pub fn diff_worktree(cwd: &str, path: &str, orig_path: Option<&str>) -> Result GitStatus { let mut st = GitStatus::default(); + // git prints `branch.ab` only "if upstream is set and the commit is present": an + // upstream with no `branch.ab` is a GONE upstream (see `GitStatus::upstream_gone`). + let mut saw_ab = false; let mut tokens = out.split('\0'); while let Some(tok) = tokens.next() { if tok.is_empty() { continue; } if let Some(rest) = tok.strip_prefix("# ") { - parse_branch_header(rest, &mut st); + saw_ab |= parse_branch_header(rest, &mut st); continue; } match tok.as_bytes()[0] { @@ -119,10 +138,14 @@ fn parse_status_v2(out: &str) -> GitStatus { _ => {} } } + st.upstream_gone = st.upstream.is_some() && !saw_ab; st } -fn parse_branch_header(rest: &str, st: &mut GitStatus) { +/// Apply one `# branch.*` header line. Returns whether it was the `branch.ab` line: +/// the caller needs to know it was SEEN, not only its values, to tell a gone upstream +/// from a genuine `+0 -0`. +fn parse_branch_header(rest: &str, st: &mut GitStatus) -> bool { if let Some(oid) = rest.strip_prefix("branch.oid ") { if oid == "(initial)" { st.unborn = true; @@ -146,7 +169,9 @@ fn parse_branch_header(rest: &str, st: &mut GitStatus) { .next() .and_then(|s| s.trim_start_matches('-').parse().ok()) .unwrap_or(0); + return true; } + false } /// `1 ` — 8 fixed fields then the path. @@ -221,9 +246,44 @@ mod tests { assert_eq!(st.upstream.as_deref(), Some("origin/main")); assert_eq!(st.ahead, 2); assert_eq!(st.behind, 1); + assert!(!st.upstream_gone); assert!(!st.unborn); } + // Verified against git 2.50: once the remote branch is deleted and pruned, the + // header keeps `branch.upstream` but drops `branch.ab`. + #[test] + fn upstream_without_ab_is_gone_not_in_sync() { + let out = "# branch.oid abc123\0# branch.head feat\0# branch.upstream origin/feat\0"; + let st = parse_status_v2(out); + assert_eq!(st.upstream.as_deref(), Some("origin/feat")); + assert!(st.upstream_gone, "an upstream with no branch.ab is gone"); + assert_eq!((st.ahead, st.behind), (0, 0), "the counts stay at their defaults"); + } + + #[test] + fn in_sync_upstream_is_not_gone() { + let out = + "# branch.oid abc123\0# branch.head main\0# branch.upstream origin/main\0# branch.ab +0 -0\0"; + let st = parse_status_v2(out); + assert!(!st.upstream_gone, "a real +0 -0 is in sync, not gone"); + } + + #[test] + fn no_upstream_is_not_gone() { + let out = "# branch.oid abc123\0# branch.head local\0"; + let st = parse_status_v2(out); + assert_eq!(st.upstream, None); + assert!(!st.upstream_gone, "gone needs an upstream to be gone from"); + } + + #[test] + fn status_read_takes_no_optional_locks() { + // A top-level option: only valid BEFORE the subcommand. + assert_eq!(STATUS_ARGS.first(), Some(&"--no-optional-locks")); + assert_eq!(STATUS_ARGS.get(1), Some(&"status")); + } + #[test] fn unborn_branch_sets_flag() { let out = "# branch.oid (initial)\0# branch.head main\0"; diff --git a/src-tauri/src/ipc/commands.rs b/src-tauri/src/ipc/commands.rs index 4388d87a..27e065e4 100644 --- a/src-tauri/src/ipc/commands.rs +++ b/src-tauri/src/ipc/commands.rs @@ -16,8 +16,8 @@ use crate::supervisor::codex::{self, CodexServer}; use crate::supervisor::control::{self, PermissionDecision, PermissionMode}; use crate::supervisor::history::{self, DiskConversation, IndexedConversation, SearchHit}; use crate::supervisor::model::{ - ContextFill, ConversationItem, GoalState, SlashCommand, WorkflowJournal, WorkflowPhase, - WorkflowRun, + ContextFill, ConversationItem, GoalState, SessionUsage, SlashCommand, WorkflowJournal, + WorkflowPhase, WorkflowRun, }; use crate::supervisor::session::{self, InitialControls, SessionHandle}; use crate::supervisor::transport::{ImageAttachment, SpawnConfig}; @@ -133,6 +133,17 @@ impl Sessions { claude } + /// Snapshot the live REMOTE (SSH) handles — the only sessions with a link to lose. + fn remote_handles(&self) -> Vec { + self.inner + .lock() + .unwrap() + .values() + .filter(|s| s.is_remote) + .map(|s| s.handle.clone()) + .collect() + } + /// Whether any session is still registered (still tearing down or live). pub fn is_empty(&self) -> bool { self.inner.lock().unwrap().is_empty() @@ -456,6 +467,47 @@ pub fn codex_available() -> bool { crate::supervisor::codex::codex_available() } +/// This Mac's name as the user knows it — the "Computer Name" of System Settings → General → +/// Sharing (`scutil --get ComputerName`, e.g. « MacBook Pro d'Alexandre »), else the host name. +/// For the conversation side panel's Machine row, which says WHERE a local conversation runs. +/// Read once per app run (a machine is not renamed under a running app often enough to poll), +/// off the main thread. `None` only when neither can be read — the row then says « This Mac ». +#[tauri::command] +#[specta::specta] +pub async fn local_machine_name() -> Option { + static NAME: std::sync::OnceLock> = std::sync::OnceLock::new(); + if let Some(name) = NAME.get() { + return name.clone(); + } + let name = tokio::task::spawn_blocking(read_local_machine_name).await.ok().flatten(); + NAME.get_or_init(|| name).clone() +} + +fn read_local_machine_name() -> Option { + let clean = |s: &str| { + let s = s.trim(); + (!s.is_empty()).then(|| s.to_string()) + }; + let computer_name = std::process::Command::new("/usr/sbin/scutil") + .args(["--get", "ComputerName"]) + .output() + .ok() + .filter(|o| o.status.success()) + .and_then(|o| clean(&String::from_utf8_lossy(&o.stdout))); + computer_name.or_else(|| { + let mut buf = [0u8; 256]; + // SAFETY: the buffer outlives the call and its length is passed; gethostname writes a + // NUL-terminated name within it (or truncates) and reports failure through its return. + let rc = unsafe { libc::gethostname(buf.as_mut_ptr().cast(), buf.len()) }; + if rc != 0 { + return None; + } + let end = buf.iter().position(|&b| b == 0).unwrap_or(buf.len()); + // A host name reads better without its local-network suffix (« alex-mbp.local »). + clean(String::from_utf8_lossy(&buf[..end]).trim_end_matches(".local")) + }) +} + /// Whether a usable `claude` binary is installed on this machine. Powers the proactive /// "Claude CLI not found" surfaces (composer bar + Settings → Accounts) so the absence /// is shown BEFORE the first message fails — the twin of [`codex_available`]. Cheap: a @@ -2028,6 +2080,26 @@ pub async fn load_session_context(session_id: String) -> Result Result, String> { + tokio::task::spawn_blocking(move || match backend { + Backend::Claude => crate::supervisor::history::load_session_usage(&session_id), + Backend::Codex => crate::supervisor::codex::load_thread_usage(&session_id), + }) + .await + .map_err(|e| e.to_string()) +} + /// Read a conversation's active `/goal` (Claude Code's native goal feature) from its on-disk /// transcript. The CLI writes goal state as `attachment` lines that are DISK-ONLY (never on the /// live stream), so the UI polls this at conversation load and on each turn edge to know whether a @@ -2611,6 +2683,19 @@ pub async fn generate_message_summary( .map_err(|e| e.to_string()) } +/// Retry every REMOTE session's lost link NOW rather than at the end of its backoff — +/// fired by the front when the Mac's network comes back or the user returns to the app +/// (see `SessionCommand::ReconnectNow`). Cheap and idempotent: an attached session, which +/// is not waiting out a backoff, ignores it. A session that closed meanwhile is skipped. +#[tauri::command] +#[specta::specta] +pub async fn reconnect_remote_sessions(sessions: tauri::State<'_, Sessions>) -> Result<(), String> { + for handle in sessions.remote_handles() { + let _ = handle.reconnect_now().await; + } + Ok(()) +} + /// Interrupt the current turn (without killing the process). #[tauri::command] #[specta::specta] @@ -5623,19 +5708,48 @@ pub async fn voice_agent_status() -> Result Result { - tokio::task::spawn_blocking(move || crate::voice::set_key(&key)) +pub async fn set_voice_agent_key( + app: tauri::AppHandle, + key: String, +) -> Result { + let status = tokio::task::spawn_blocking(move || crate::voice::set_key(&key)) .await - .map_err(|e| format!("keychain task failed: {e}"))? + .map_err(|e| format!("keychain task failed: {e}"))??; + reapply_wake_after_key_change(&app, status.configured).await; + Ok(status) } /// Forget the stored OpenAI key (absent item = success). #[tauri::command] #[specta::specta] -pub async fn clear_voice_agent_key() -> Result { - tokio::task::spawn_blocking(crate::voice::clear_key) +pub async fn clear_voice_agent_key( + app: tauri::AppHandle, +) -> Result { + let status = tokio::task::spawn_blocking(crate::voice::clear_key) .await - .map_err(|e| format!("keychain task failed: {e}"))? + .map_err(|e| format!("keychain task failed: {e}"))??; + reapply_wake_after_key_change(&app, status.configured).await; + Ok(status) +} + +/// The wake word is held off while no OpenAI key is stored +/// (`crate::wake::require_voice_key`), so a key change re-applies it: removing +/// the key stops an always-on microphone whose every detection would now be +/// dropped, adding one resumes a wake word the user left enabled. Best-effort — +/// the key change itself already succeeded, and the wake status carries its own +/// outcome for Settings to show. +async fn reapply_wake_after_key_change(app: &tauri::AppHandle, key_configured: bool) { + let cfg = load_wake_config(app, &app.state::()); + let wake = (*app.state::>()).clone(); + // Nothing to pause or resume for a wake word the user never enabled, and a + // detector already running with a key stored needs no restart. + if !cfg.enabled || (key_configured && wake.status().running) { + return; + } + let apply = move || wake.apply(crate::wake::require_voice_key(cfg, key_configured)); + if let Err(e) = tokio::task::spawn_blocking(apply).await { + eprintln!("[wake] re-applying the wake word after a key change failed: {e}"); + } } /// Mint a short-lived Realtime client secret for ONE voice session — the only @@ -5684,6 +5798,7 @@ pub fn load_wake_config(app: &tauri::AppHandle, store: &Store) -> crate::wake::W sensitivity, debug_capture: read(WAKE_DEBUG_CAPTURE_KEY).as_deref() == Some("1"), debug_dir: wake_debug_dir(app), + paused: None, // decided at apply time (`crate::wake::require_voice_key`) } } @@ -5743,9 +5858,14 @@ pub async fn set_wake_word_config( cfg }; let wake = (*app.state::>()).clone(); - tokio::task::spawn_blocking(move || wake.apply(cfg)) - .await - .map_err(|e| format!("wake apply task failed: {e}")) + tokio::task::spawn_blocking(move || { + // No OpenAI key → held off, choice kept (`require_voice_key`). The Keychain + // is only read when it can matter. + let has_key = cfg.enabled && crate::voice::status().configured; + wake.apply(crate::wake::require_voice_key(cfg, has_key)) + }) + .await + .map_err(|e| format!("wake apply task failed: {e}")) } /// A compact, bounded directory tree for AGENT orientation (the `browse_folders` diff --git a/src-tauri/src/ipc/events.rs b/src-tauri/src/ipc/events.rs index 09ddda5f..5ba7163c 100644 --- a/src-tauri/src/ipc/events.rs +++ b/src-tauri/src/ipc/events.rs @@ -9,13 +9,6 @@ use crate::supervisor::model::{ }; use crate::usage::PlanUsage; -/// Emitted periodically by a Rust timer. Proves Rust -> React (typed event). -#[derive(Debug, Clone, Serialize, Deserialize, Type, Event)] -pub struct TickEvent { - pub seq: u32, - pub message: String, -} - /// A session's lifecycle / identity changed. #[derive(Debug, Clone, Serialize, Deserialize, Type, Event)] pub struct SessionStateEvent { diff --git a/src-tauri/src/lib.rs b/src-tauri/src/lib.rs index eb5ada11..c7f962bb 100644 --- a/src-tauri/src/lib.rs +++ b/src-tauri/src/lib.rs @@ -35,10 +35,10 @@ use ipc::commands::{ generate_conversation_title, generate_message_summary, get_plan_usage, git_branches, git_commit, git_commit_file_diff, git_commit_files, git_diff, git_fetch, git_log, git_pull, git_push, git_status, - interrupt_session, list_disk_conversations, list_extensions, list_marketplaces, + interrupt_session, local_machine_name, reconnect_remote_sessions, list_disk_conversations, list_extensions, list_marketplaces, list_plugin_contents, get_output_style, set_output_style, mcp_permission_rules, apply_session_overrides, fetch_global_mcp_status, - list_worktrees, load_persisted_state, load_session_context, load_session_goal, + list_worktrees, load_persisted_state, load_session_context, load_session_goal, load_session_usage, load_session_history, load_subagent_transcript, load_workflow_journal, load_workflow_phases, load_workflow_run, unwatch_workflow_journal, watch_workflow_journal, @@ -89,7 +89,8 @@ use ipc::commands::{ }; use bootstrap::connect::bootstrap_forget_host_key; use bootstrap::orchestrator::{ - bootstrap_cancel, bootstrap_resume, bootstrap_server, machine_diagnose, machine_repair, BootstrapSessions, + bootstrap_cancel, bootstrap_resume, bootstrap_server, machine_diagnose, machine_reachability, machine_repair, + BootstrapSessions, ServerLocks, }; use bootstrap::server_setup::{cancel_claude_login, restart_claude_login, start_claude_login, submit_claude_login_code}; @@ -99,7 +100,7 @@ use ipc::events::{ SessionCommandsEvent, SessionExtensionsChangedEvent, SessionMessageEvent, SessionPermissionEvent, SessionPermissionResolvedEvent, SessionRemoteControlEvent, SessionStateEvent, SessionSummaryEvent, SessionTaskEvent, SessionTitleEvent, BootstrapProgressEvent, HostKeyFingerprintEvent, ServerLoginPromptEvent, ServerLoginResultEvent, - TerminalExitEvent, TerminalOutputEvent, TickEvent, + TerminalExitEvent, TerminalOutputEvent, TosseCrmEvent, TosseLiveStateEvent, WakeWordEvent, WorkflowJournalEvent, }; use tauri_specta::{collect_commands, collect_events, Builder, Event}; @@ -195,6 +196,7 @@ fn ipc_builder() -> Builder { ping, spawn_session, claude_available, + local_machine_name, codex_available, codex_list_models, codex_list_skills, @@ -260,6 +262,7 @@ fn ipc_builder() -> Builder { fetch_known_agents, load_session_history, load_session_context, + load_session_usage, load_session_goal, check_rewind_target, rewind_conversation, @@ -285,6 +288,7 @@ fn ipc_builder() -> Builder { generate_conversation_title, generate_message_summary, interrupt_session, + reconnect_remote_sessions, mcp_status, mcp_toggle, mcp_reconnect, @@ -388,6 +392,7 @@ fn ipc_builder() -> Builder { bootstrap_resume, bootstrap_cancel, machine_diagnose, + machine_reachability, machine_repair, artifact_host_show, artifact_host_set_bounds, @@ -397,7 +402,6 @@ fn ipc_builder() -> Builder { artifact_host_open_claude_url, ]) .events(collect_events![ - TickEvent, SessionStateEvent, SessionMessageEvent, SessionPermissionEvent, @@ -840,7 +844,10 @@ pub fn run() { if cfg.enabled { let wake = wake.clone(); std::thread::spawn(move || { - let _ = wake.apply(cfg); + // Without an OpenAI key every detection is dropped, so the + // mic stays off — paused, choice kept (`require_voice_key`). + let has_key = voice::status().configured; + let _ = wake.apply(wake::require_voice_key(cfg, has_key)); }); } } @@ -854,24 +861,6 @@ pub fn run() { let cfg = ipc::commands::load_remote_config(&app.state::()); tauri::async_runtime::spawn(async move { hub.apply_remote(cfg).await }); } - - // Rust timer: emit a TickEvent every second (Rust -> React) — kept as - // a heartbeat / proof of the outbound event leg. - let handle = app.handle().clone(); - std::thread::spawn(move || { - let mut seq = 0u32; - loop { - std::thread::sleep(std::time::Duration::from_secs(1)); - seq += 1; - let ev = TickEvent { - seq, - message: format!("tick #{seq}"), - }; - if TickEvent::emit(&ev, &handle).is_err() { - break; // window closed - } - } - }); Ok(()) }) .build(tauri::generate_context!()) diff --git a/src-tauri/src/supervisor/assembler.rs b/src-tauri/src/supervisor/assembler.rs index 90c7aadc..216a0abb 100644 --- a/src-tauri/src/supervisor/assembler.rs +++ b/src-tauri/src/supervisor/assembler.rs @@ -18,8 +18,9 @@ use serde_json::Value; use super::control; use super::model::{ - BackgroundTask, BackgroundTaskKind, BackgroundTaskStatus, ConversationItem, NormalizedBlock, - RateLimitSnapshot, RemoteControlState, RemoteLinkState, RetryState, SessionEvent, SessionStatePayload, + BackgroundTask, BackgroundTaskKind, BackgroundTaskStatus, ConversationItem, ModelTokenUsage, + NormalizedBlock, RateLimitSnapshot, RemoteControlState, RemoteLinkState, RetryState, SessionEvent, + SessionStatePayload, SessionUsage, TokenUsage, }; use super::protocol::{ AssistantMsg, CliMessage, RateLimitMsg, ResultMsg, StreamEventMsg, SystemMsg, @@ -76,6 +77,14 @@ pub struct Assembler { /// end-of-turn (`ingest_result`) so it can never swallow a real user turn (those only /// arrive AFTER a `result`, never mid-turn). skill_invocation_pending: bool, + /// The CLI's CUMULATIVE `result.duration_api_ms` as of the previous `result` — the + /// baseline a turn's own model time is measured from. The wire value is a running + /// per-SESSION total (verified live, claude 2.1.283: 2.1s → 6.0s → 7.1s over three + /// plain prompts) that is even restored across `--resume` (7.1s → 7.9s after a + /// re-spawn), NOT a per-turn figure. `Some(0)` for a brand-new session (the counter + /// starts at zero — see [`Assembler::mark_fresh_session`]); `None` while unknown (a + /// resumed or re-attached process, until its first `result` sets it). + api_ms_baseline: Option, } /// The last-announced friendly labels for the three controls (see [`Assembler`]). @@ -108,6 +117,27 @@ impl Assembler { Self::default() } + /// This process started a brand-new session (no `--resume`, no re-attach), so the + /// CLI's cumulative model-time counter starts at zero and the FIRST turn's model time + /// is known too. Left unset otherwise: a resumed session's restored total cannot be + /// known before its first `result`, whose model time is then reported as unknown. + pub fn mark_fresh_session(&mut self) { + self.api_ms_baseline = Some(0); + } + + /// The model time spent since the previous `result` — this turn's share of the CLI's + /// cumulative `duration_api_ms` (see [`Self::api_ms_baseline`]). `None` when the + /// baseline is unknown, or when the counter went BACKWARDS (a re-spawned process that + /// restored an older total): the delta would be meaningless, so the turn reports none + /// and the new total becomes the baseline. A `result` without the field leaves the + /// baseline untouched. + fn turn_api_ms(&mut self, cumulative: Option) -> Option { + let total = cumulative?; + let turn = self.api_ms_baseline.and_then(|base| total.checked_sub(base)); + self.api_ms_baseline = Some(total); + turn + } + /// Read-only view of the current session state. pub fn state(&self) -> &SessionStatePayload { &self.state @@ -683,16 +713,21 @@ impl Assembler { // have `parent_tool_use_id` set and their own window — never let them // clobber the conversation's context meter. if se.parent_tool_use_id.is_none() { - if let Some(used) = event - .get("message") - .and_then(|m| m.get("usage")) - .and_then(context_used_from_usage) - { + let usage = event.get("message").and_then(|m| m.get("usage")); + if let Some(used) = usage.and_then(context_used_from_usage) { if self.state.context_tokens != Some(used) { self.state.context_tokens = Some(used); state_changed = true; } } + // Its breakdown, from the SAME object (so the two never disagree). The + // output is only a stub here; the call's `message_delta` completes it. + if let Some(breakdown) = usage.and_then(token_usage_from) { + if self.state.context_usage != Some(breakdown) { + self.state.context_usage = Some(breakdown); + state_changed = true; + } + } } if state_changed { out.push(SessionEvent::State(self.state.clone())); @@ -754,8 +789,23 @@ impl Assembler { } } } - // content_block_stop, message_delta, message_stop carry no incremental - // text we surface yet. + // A root call's `message_delta` carries its final `output_tokens` — the one figure + // of its usage that `message_start` could not know yet. One per model call, so this + // is one state event per call, not per token. + "message_delta" if se.parent_tool_use_id.is_none() => { + let output = event + .get("usage") + .and_then(|u| u.get("output_tokens")) + .and_then(Value::as_u64); + if let (Some(output), Some(usage)) = (output, self.state.context_usage.as_mut()) { + if usage.output != output { + usage.output = output; + out.push(SessionEvent::State(self.state.clone())); + } + } + } + // content_block_stop, message_stop, and a sub-agent's message_delta carry nothing + // we surface yet. _ => {} } } @@ -1080,6 +1130,9 @@ impl Assembler { if let Some(used) = context_used_from_usage(final_usage) { self.state.context_tokens = Some(used); } + if let Some(breakdown) = token_usage_from(final_usage) { + self.state.context_usage = Some(breakdown); + } // Authoritative window for THIS session's model (distinguishes 200k vs 1M). // Only updates when the result reports the session model's own entry — a // sub-agent-only turn returns None and keeps the last known window. @@ -1088,6 +1141,13 @@ impl Assembler { { self.state.context_window = Some(window); } + // What the whole session has consumed, every agent included: the CLI's cumulative + // `modelUsage` snapshot, REPLACED (never summed, never max'd — a `/clear` legitimately + // lowers it). A result without one (an absent/empty map) keeps the last known value + // rather than reading as "nothing spent". Rides the State event pushed just below. + if let Some(usage) = session_usage_from_model_usage(&r.model_usage, r.total_cost_usd) { + self.state.session_usage = Some(usage); + } out.push(SessionEvent::Item(ConversationItem::TurnResult { subtype: r.subtype.clone(), is_error: r.is_error, @@ -1098,8 +1158,11 @@ impl Assembler { total_cost_usd: r.total_cost_usd, num_turns: r.num_turns, duration_ms: r.duration_ms, - duration_api_ms: r.duration_api_ms, + duration_api_ms: self.turn_api_ms(r.duration_api_ms), ttft_ms: r.ttft_ms, + // The TOP-LEVEL usage: the turn's aggregate over all its calls — what it consumed — + // where the context reading above takes the LAST call's. + usage: token_usage_from(&r.usage), })); out.push(SessionEvent::State(self.state.clone())); } @@ -1364,6 +1427,79 @@ pub(crate) fn context_used_from_usage(usage: &Value) -> Option { Some(input.unwrap_or(0) + cache_creation.unwrap_or(0) + cache_read.unwrap_or(0)) } +/// The four token counts of a `usage` object, broken down (see [`TokenUsage`]). `None` when +/// it carries none of them — an empty/`null` usage must not zero a known breakdown, exactly as +/// [`context_used_from_usage`] refuses to zero a known fill. Its prompt part always sums to +/// what that function returns for the same object. +pub(crate) fn token_usage_from(usage: &Value) -> Option { + let field = |k: &str| usage.get(k).and_then(Value::as_u64); + let input = field("input_tokens"); + let cache_creation = field("cache_creation_input_tokens"); + let cache_read = field("cache_read_input_tokens"); + let output = field("output_tokens"); + if input.is_none() && cache_creation.is_none() && cache_read.is_none() && output.is_none() { + return None; + } + Some(TokenUsage { + input: input.unwrap_or(0), + cache_creation: cache_creation.unwrap_or(0), + cache_read: cache_read.unwrap_or(0), + output: output.unwrap_or(0), + }) +} + +/// A session's cumulative, all-agent spend (see [`SessionUsage`]) from a `modelUsage` map — +/// `result.modelUsage` live, or the `modelUsage` of the transcript's `cost-state` line (the +/// same camelCase shape). Each model's `inputTokens + cacheCreationInputTokens + +/// cacheReadInputTokens + outputTokens` is added into the total; `thinkingTokens` is NOT, it is +/// already inside `outputTokens`. `cost_usd` is the caller's session cost (`total_cost_usd` / +/// `totalCostUSD`), passed through as is. +/// +/// `None` when the map is absent, not an object, or names no model with a single token count: +/// an empty snapshot must not overwrite a known total with zeros. +pub(crate) fn session_usage_from_model_usage( + model_usage: &Value, + cost_usd: Option, +) -> Option { + let obj = model_usage.as_object()?; + let mut per_model: Vec = obj + .iter() + .filter_map(|(model, entry)| { + let field = |k: &str| entry.get(k).and_then(Value::as_u64); + let input = field("inputTokens"); + let cache_creation = field("cacheCreationInputTokens"); + let cache_read = field("cacheReadInputTokens"); + let output = field("outputTokens"); + if input.is_none() && cache_creation.is_none() && cache_read.is_none() && output.is_none() { + return None; + } + Some(ModelTokenUsage { + model: model.clone(), + usage: TokenUsage { + input: input.unwrap_or(0), + cache_creation: cache_creation.unwrap_or(0), + cache_read: cache_read.unwrap_or(0), + output: output.unwrap_or(0), + }, + cost_usd: entry.get("costUSD").and_then(Value::as_f64), + }) + }) + .collect(); + if per_model.is_empty() { + return None; + } + let size = |u: &TokenUsage| u.input + u.cache_creation + u.cache_read + u.output; + // Largest first; the id breaks a tie so the order never depends on the map's. + per_model.sort_by(|a, b| size(&b.usage).cmp(&size(&a.usage)).then_with(|| a.model.cmp(&b.model))); + let total = per_model.iter().fold(TokenUsage::default(), |acc, m| TokenUsage { + input: acc.input.saturating_add(m.usage.input), + cache_creation: acc.cache_creation.saturating_add(m.usage.cache_creation), + cache_read: acc.cache_read.saturating_add(m.usage.cache_read), + output: acc.output.saturating_add(m.usage.output), + }); + Some(SessionUsage { total, cost_usd, per_model }) +} + /// The AUTHORITATIVE context-window size for the session's own model, read from a /// `result.modelUsage` map. This is the only reliable source of the window: the /// number distinguishes e.g. Opus-200k from Opus-1M, which the model NAME cannot @@ -1443,6 +1579,9 @@ mod tests { #[test] fn assembles_fixture_into_normalized_events() { let mut asm = Assembler::new(); + // The capture is a brand-new session's first turn: its cumulative model time IS + // this turn's. + asm.mark_fresh_session(); let mut streamed_text = String::new(); let mut saw_model = false; let mut saw_session_id = false; @@ -1759,6 +1898,71 @@ mod tests { assert_eq!(asm.state().context_window, Some(1_000_000)); } + #[test] + fn context_usage_breaks_down_the_same_fill_from_the_fixture() { + let mut asm = Assembler::new(); + for line in CAPTURE.lines().map(str::trim).filter(|l| !l.is_empty()) { + asm.ingest(&serde_json::from_str::(line).unwrap()); + } + let u = asm.state().context_usage.expect("the fixture reports usage"); + assert_eq!((u.input, u.cache_creation, u.cache_read), (5069, 9061, 15626)); + // The breakdown and the fill come from the same object: they can never disagree. + assert_eq!(Some(u.input + u.cache_creation + u.cache_read), asm.state().context_tokens); + } + + #[test] + fn a_root_call_s_output_is_completed_by_its_message_delta() { + let line = |s: &str| serde_json::from_str::(s).unwrap(); + let mut asm = Assembler::new(); + asm.ingest(&line( + r#"{"type":"stream_event","event":{"type":"message_start","message":{"id":"m1","usage":{"input_tokens":40,"cache_creation_input_tokens":300,"cache_read_input_tokens":9000,"output_tokens":1}}},"session_id":"s"}"#, + )); + let at_start = asm.state().context_usage.unwrap(); + assert_eq!((at_start.input, at_start.cache_creation, at_start.cache_read), (40, 300, 9000)); + // A SUB-AGENT's delta must not overwrite the conversation's own call. + asm.ingest(&line( + r#"{"type":"stream_event","event":{"type":"message_delta","usage":{"output_tokens":7777}},"parent_tool_use_id":"tu_sub","session_id":"s"}"#, + )); + assert_eq!(asm.state().context_usage.unwrap().output, 1); + let evs = asm.ingest(&line( + r#"{"type":"stream_event","event":{"type":"message_delta","usage":{"output_tokens":512}},"session_id":"s"}"#, + )); + assert_eq!(asm.state().context_usage.unwrap().output, 512); + assert!(evs.iter().any(|e| matches!(e, SessionEvent::State(_))), "the new output reaches the UI"); + } + + #[test] + fn a_turn_reports_what_it_consumed_while_the_context_reads_its_last_call() { + let result = serde_json::json!({ + "type": "result", "subtype": "success", "is_error": false, "result": "ok", + "stop_reason": "end_turn", "session_id": "s", "uuid": "u", + "usage": { + "input_tokens": 2100, "cache_read_input_tokens": 18000, + "cache_creation_input_tokens": 500, "output_tokens": 900, + "iterations": [ + {"input_tokens": 100, "cache_read_input_tokens": 0, "cache_creation_input_tokens": 500, "output_tokens": 300}, + {"input_tokens": 2000, "cache_read_input_tokens": 18000, "cache_creation_input_tokens": 0, "output_tokens": 600} + ] + } + }); + let mut asm = Assembler::new(); + let evs = asm.ingest(&serde_json::from_value::(result).unwrap()); + // The window holds the LAST call's prompt… + assert_eq!( + asm.state().context_usage, + Some(TokenUsage { input: 2000, cache_creation: 0, cache_read: 18000, output: 600 }) + ); + // …while the turn consumed the aggregate. + let usage = evs.iter().find_map(|e| match e { + SessionEvent::Item(ConversationItem::TurnResult { usage, .. }) => Some(*usage), + _ => None, + }); + assert_eq!( + usage, + Some(Some(TokenUsage { input: 2100, cache_creation: 500, cache_read: 18000, output: 900 })) + ); + } + #[test] fn context_helpers_sum_and_pick_window() { let usage = serde_json::json!({ @@ -1806,6 +2010,90 @@ mod tests { assert_eq!(context_window_from_model_usage(&Value::Null, Some("x")), None); } + #[test] + fn session_usage_sums_every_model_and_never_adds_thinking() { + let model_usage = serde_json::json!({ + "claude-haiku-4-5": {"inputTokens": 514, "outputTokens": 11, "costUSD": 0.0006}, + "claude-opus-5[1m]": { + "inputTokens": 5069, "outputTokens": 900, "thinkingTokens": 400, + "cacheReadInputTokens": 15626, "cacheCreationInputTokens": 9061, + "costUSD": 0.12, "contextWindow": 1000000 + }, + // A model the map lists without a single count adds nothing and is not listed. + "claude-sonnet-4-6": {"contextWindow": 200000} + }); + let u = session_usage_from_model_usage(&model_usage, Some(0.1206)).expect("known"); + // `thinkingTokens` is already inside `outputTokens`: 900 + 11, not 1311. + assert_eq!( + u.total, + TokenUsage { input: 5069 + 514, cache_creation: 9061, cache_read: 15626, output: 911 } + ); + assert_eq!(u.cost_usd, Some(0.1206)); + // Largest first; the helper Haiku is IN the total. + let models: Vec<&str> = u.per_model.iter().map(|m| m.model.as_str()).collect(); + assert_eq!(models, ["claude-opus-5[1m]", "claude-haiku-4-5"]); + assert_eq!(u.per_model[1].cost_usd, Some(0.0006)); + // Nothing to read → None, so a result without a map never zeroes a known total. + assert_eq!(session_usage_from_model_usage(&Value::Null, Some(1.0)), None); + assert_eq!(session_usage_from_model_usage(&serde_json::json!({}), None), None); + assert_eq!( + session_usage_from_model_usage(&serde_json::json!({"m": {"contextWindow": 1}}), None), + None + ); + } + + #[test] + fn fixture_result_carries_the_all_model_session_usage() { + let mut asm = Assembler::new(); + let mut last = None; + for line in CAPTURE.lines().map(str::trim).filter(|l| !l.is_empty()) { + let msg: CliMessage = serde_json::from_str(line).unwrap(); + for ev in asm.ingest(&msg) { + if let SessionEvent::State(s) = ev { + last = Some(s.session_usage); + } + } + } + // The fixture's result: Opus 5069/9061/15626/5 + a helper Haiku 514/11. + let u = last.flatten().expect("the result's state carries the session usage"); + assert_eq!( + u.total, + TokenUsage { input: 5069 + 514, cache_creation: 9061, cache_read: 15626, output: 5 + 11 } + ); + assert_eq!(u.cost_usd, Some(0.124462)); + assert_eq!(u.per_model.len(), 2); + } + + #[test] + fn session_usage_is_replaced_by_each_result_never_summed() { + let result = |input: u64, cost: f64| -> CliMessage { + serde_json::from_value(serde_json::json!({ + "type": "result", "subtype": "success", "is_error": false, + "total_cost_usd": cost, + "usage": {"input_tokens": 1, "output_tokens": 1}, + "modelUsage": {"claude-opus-5": {"inputTokens": input, "outputTokens": 10}} + })) + .unwrap() + }; + let mut asm = Assembler::new(); + asm.ingest(&result(100, 0.5)); + // The second result carries the RUNNING total (100 + 50): it replaces, never adds. + asm.ingest(&result(150, 0.8)); + let u = asm.state().session_usage.clone().unwrap(); + assert_eq!(u.total.input, 150); + assert_eq!(u.cost_usd, Some(0.8)); + // A `/clear` resets the CLI's counter: the value goes DOWN, and is taken as is. + asm.ingest(&result(20, 0.05)); + assert_eq!(asm.state().session_usage.as_ref().unwrap().total.input, 20); + // A result without a map keeps the last known total. + let bare: CliMessage = serde_json::from_value(serde_json::json!({ + "type": "result", "subtype": "error_during_execution", "is_error": true + })) + .unwrap(); + asm.ingest(&bare); + assert_eq!(asm.state().session_usage.as_ref().unwrap().total.input, 20); + } + #[test] fn result_context_uses_last_iteration_not_aggregate() { let result = serde_json::json!({ @@ -1840,6 +2128,69 @@ mod tests { assert_eq!(asm.state().context_window, Some(1_000_000)); } + /// A `result` line carrying only the timing fields under test. + fn timed_result(duration_ms: u64, cumulative_api_ms: u64) -> CliMessage { + serde_json::from_value(serde_json::json!({ + "type": "result", + "subtype": "success", + "is_error": false, + "result": "ok", + "session_id": "s", + "uuid": "u", + "duration_ms": duration_ms, + "duration_api_ms": cumulative_api_ms, + })) + .unwrap() + } + + /// The (duration_ms, duration_api_ms) of the TurnResult a `result` normalizes to. + fn turn_timing(asm: &mut Assembler, msg: &CliMessage) -> (Option, Option) { + asm.ingest(msg) + .into_iter() + .find_map(|ev| match ev { + SessionEvent::Item(ConversationItem::TurnResult { + duration_ms, + duration_api_ms, + .. + }) => Some((duration_ms, duration_api_ms)), + _ => None, + }) + .expect("a result normalizes to a TurnResult") + } + + /// `result.duration_api_ms` is a running per-session TOTAL on the wire (values from a + /// live probe of claude 2.1.283, three prompts in one process). Each turn must report + /// its OWN share, or every footer after the first shows the whole session's model time. + #[test] + fn turn_model_time_is_the_delta_of_the_cumulative_counter() { + let mut asm = Assembler::new(); + asm.mark_fresh_session(); + assert_eq!(turn_timing(&mut asm, &timed_result(1265, 2128)), (Some(1265), Some(2128))); + assert_eq!(turn_timing(&mut asm, &timed_result(3907, 6016)), (Some(3907), Some(3888))); + assert_eq!(turn_timing(&mut asm, &timed_result(1115, 7086)), (Some(1115), Some(1070))); + } + + /// A resumed process restores the session's earlier total, which we cannot know + /// before its first `result`: that turn's model time is unknown (never the whole + /// restored total), the next ones are exact again. + #[test] + fn turn_model_time_is_unknown_on_the_first_result_of_a_resumed_process() { + let mut asm = Assembler::new(); // not marked fresh: a --resume spawn + assert_eq!(turn_timing(&mut asm, &timed_result(863, 7894)), (Some(863), None)); + assert_eq!(turn_timing(&mut asm, &timed_result(900, 8800)), (Some(900), Some(906))); + } + + /// A counter that went BACKWARDS (a re-spawned process restored an older total) yields + /// no model time for that turn instead of an underflow, and re-bases on the new total. + #[test] + fn turn_model_time_survives_a_counter_that_went_backwards() { + let mut asm = Assembler::new(); + asm.mark_fresh_session(); + assert_eq!(turn_timing(&mut asm, &timed_result(1000, 5000)).1, Some(5000)); + assert_eq!(turn_timing(&mut asm, &timed_result(1000, 3000)).1, None); + assert_eq!(turn_timing(&mut asm, &timed_result(1000, 3500)).1, Some(500)); + } + const TASKS_CAPTURE: &str = include_str!("fixtures/capture_tasks.jsonl"); /// Feed the captured task lifecycle through the assembler and collect the final diff --git a/src-tauri/src/supervisor/codex/history.rs b/src-tauri/src/supervisor/codex/history.rs index 8987042f..97ea25f0 100644 --- a/src-tauri/src/supervisor/codex/history.rs +++ b/src-tauri/src/supervisor/codex/history.rs @@ -40,7 +40,7 @@ use super::server::CLIENT_NAME_INTERNAL; use crate::supervisor::history::{ self, DiskConversation, IndexedConversation, EXCERPT_CHARS, HEAD_SCAN_LINES, INDEX_BODY_CAP, }; -use crate::supervisor::model::{ConversationItem, NormalizedBlock}; +use crate::supervisor::model::{ConversationItem, NormalizedBlock, SessionUsage, TokenUsage}; /// Find the rollout file for `thread_id` under `/sessions` (nested /// `YYYY/MM/DD/`). The thread id is the tail of the filename @@ -100,6 +100,50 @@ pub fn load_thread_history(thread_id: &str) -> Vec { parse_rollout(&path) } +/// Read a Codex thread's cumulative token usage from its rollout — the cold seed of the side +/// panel's session total, the disk twin of the live `thread/tokenUsage/updated` `total`. The +/// source is the LAST `event_msg` of `type:"token_count"` that carries +/// `info.total_token_usage` (Codex also writes rate-limit-only `token_count`s with a null +/// `info`, which are skipped). ⚠️ This thread only — collab sub-agents are threads of their own. +/// `None` when the rollout has no such line or cannot be found: unknown, never zero. +pub fn load_thread_usage(thread_id: &str) -> Option { + load_thread_usage_in(&codex_sessions_dir()?, thread_id) +} + +/// [`load_thread_usage`] against an explicit sessions dir — the testable core (no env). +fn load_thread_usage_in(sessions_dir: &Path, thread_id: &str) -> Option { + let path = find_rollout(sessions_dir, thread_id)?; + history::read_last_matching_line(&path, "\"token_count\"", |line| { + let payload = line.get("payload")?; + if payload.get("type").and_then(Value::as_str) != Some("token_count") { + return None; + } + let total = rollout_total_usage(payload.get("info")?.get("total_token_usage")?)?; + Some(SessionUsage { total, cost_usd: None, per_model: Vec::new() }) + }) +} + +/// A rollout `total_token_usage` (snake_case) on Claude's [`TokenUsage`] shape — the mapping +/// the live side applies to the camelCase `tokenUsage.total`: the cached input is INSIDE +/// `input_tokens` (split out as `cache_read`), the reasoning is inside `output_tokens`, and +/// `total_tokens` is their sum (so it is not read). `None` when it reports no counts at all. +fn rollout_total_usage(u: &Value) -> Option { + let field = |k: &str| u.get(k).and_then(Value::as_u64); + let input = field("input_tokens"); + let cached = field("cached_input_tokens"); + let output = field("output_tokens"); + if input.is_none() && cached.is_none() && output.is_none() { + return None; + } + let cached = cached.unwrap_or(0); + Some(TokenUsage { + input: input.unwrap_or(0).saturating_sub(cached), + cache_creation: 0, + cache_read: cached, + output: output.unwrap_or(0), + }) +} + // ---- Disk listing + search index (the Codex analogue of the Claude scan in // `crate::supervisor::history`) --------------------------------------------------------- // @@ -2004,4 +2048,43 @@ mod tests { "an unreadable rollout must surface a history_error notice, not an empty conversation" ); } + + #[test] + fn thread_usage_seed_reads_the_last_token_count_total() { + let base = std::env::temp_dir().join(format!("tosse-codex-usage-{}", std::process::id())); + std::fs::remove_dir_all(&base).ok(); + let id = "01a0def6-bc25-7612-abd0-086272d73197"; + let count = |input: u64, cached: u64, output: u64| { + line( + "event_msg", + json!({ "type": "token_count", "info": { + "total_token_usage": { "input_tokens": input, "cached_input_tokens": cached, "cache_write_input_tokens": 0, "output_tokens": output, "reasoning_output_tokens": 717, "total_tokens": input + output }, + "last_token_usage": { "input_tokens": 1, "cached_input_tokens": 0, "output_tokens": 1, "total_tokens": 2 }, + "model_context_window": 258400 } }), + ) + }; + write_rollout( + &base, + "26", + &format!("rollout-2026-07-26T20-25-07-{id}"), + &[ + line("session_meta", json!({ "id": id, "cwd": "/r" })), + count(1000, 800, 20), + // The latest total wins (verified shape: 74,353 + 920 = 75,273). + count(74_353, 67_840, 920), + // A rate-limit-only count (null info) after it is not a total. + line("event_msg", json!({ "type": "token_count", "info": null, "rate_limits": {} })), + ], + ); + let u = load_thread_usage_in(&base, id).expect("a total is on disk"); + assert_eq!( + u.total, + TokenUsage { input: 74_353 - 67_840, cache_creation: 0, cache_read: 67_840, output: 920 } + ); + // Summing back to the rollout's own total_tokens: nothing double-counted. + assert_eq!(u.total.input + u.total.cache_read + u.total.output, 75_273); + assert_eq!(u.cost_usd, None); + assert_eq!(load_thread_usage_in(&base, "nope"), None); + std::fs::remove_dir_all(&base).ok(); + } } diff --git a/src-tauri/src/supervisor/codex/mod.rs b/src-tauri/src/supervisor/codex/mod.rs index 75b46f24..46381f45 100644 --- a/src-tauri/src/supervisor/codex/mod.rs +++ b/src-tauri/src/supervisor/codex/mod.rs @@ -22,7 +22,9 @@ mod transport; pub use accounts::{CodexAccountStatus, CodexLoginStart}; pub use config::list_extensions; pub use extensions::{CodexHooksSnapshot, CodexPluginsLive}; -pub use history::{build_codex_search_index, list_codex_disk_conversations, load_thread_history}; +pub use history::{ + build_codex_search_index, list_codex_disk_conversations, load_thread_history, load_thread_usage, +}; pub use protocol::{CodexControls, CodexForkResult, CodexModel, CodexServiceTier, CodexSkill}; pub use server::{CodexError, CodexServer}; pub use session::spawn_session; diff --git a/src-tauri/src/supervisor/codex/session.rs b/src-tauri/src/supervisor/codex/session.rs index 3f1e874a..9750eba3 100644 --- a/src-tauri/src/supervisor/codex/session.rs +++ b/src-tauri/src/supervisor/codex/session.rs @@ -23,7 +23,7 @@ use crate::supervisor::control::PermissionDecision; use crate::supervisor::model::{ BackgroundTask, BackgroundTaskKind, BackgroundTaskStatus, ConversationItem, McpAuthResult, McpServerLive, NormalizedBlock, PermissionRequestPayload, RemoteControlState, SessionEmitter, - SessionStatePayload, + SessionStatePayload, SessionUsage, TokenUsage, }; use crate::supervisor::session::{InitialControls, SessionCommand, SessionError, SessionHandle}; use crate::supervisor::transport::{ImageAttachment, SpawnConfig, TransportError}; @@ -1008,13 +1008,26 @@ impl CodexCore { // reasoning across EVERY turn, which only ever grows — so the ring would creep // toward "near full" turn after turn even on a mostly-empty window (a gpt-5.6 // conversation reading "near max" at ~350k against its own window is this bug). - if let Some(used) = usage - .and_then(|u| u.get("last")) - .and_then(|l| l.get("inputTokens")) - .and_then(Value::as_u64) - { + let last = usage.and_then(|u| u.get("last")); + if let Some(used) = last.and_then(|l| l.get("inputTokens")).and_then(Value::as_u64) { self.state.context_tokens = Some(used); } + // The same last turn, broken down onto Claude's shape. OpenAI counts CACHED input + // INSIDE `inputTokens` (a subset, not an addition) and has no separate cache + // write — so `input` is the uncached remainder and `cache_creation` stays 0, + // which keeps `input + cache_read` equal to the fill set just above. + if let Some(breakdown) = last.and_then(codex_token_usage) { + self.state.context_usage = Some(breakdown); + } + // What the THREAD has consumed over its whole life — the one figure this push + // carries that is a running total (see the ring above for why it is not the + // fill). Same mapping as `last`, cached input split out of the input. Replaced, + // never summed. ⚠️ This thread only: collab sub-agents run as their own threads, + // whose pushes the demux drops — the UI labels it « this thread ». + if let Some(total) = usage.and_then(|u| u.get("total")).and_then(codex_token_usage) { + self.state.session_usage = + Some(SessionUsage { total, cost_usd: None, per_model: Vec::new() }); + } self.push_state(); } "error" => { @@ -1598,6 +1611,9 @@ impl CodexCore { // rider + TTFT are Claude-only; None keeps that part of the footer honest. duration_api_ms: None, ttft_ms: None, + // Codex reports usage per THREAD (`thread/tokenUsage/updated`), never per turn: + // there is no honest per-turn consumption to attach here. + usage: None, }); self.streaming_ids.clear(); self.carded.clear(); @@ -2490,9 +2506,43 @@ fn join_reasoning(summary: &[String], content: &[String]) -> String { .join("\n\n") } +/// A Codex `tokenUsage.last` object on Claude's [`TokenUsage`] shape. OpenAI counts CACHED +/// input as a SUBSET of `inputTokens` (not an addition, unlike Anthropic's cache fields), and +/// has no separate cache write: so `input` is the uncached remainder, the cached part is +/// `cache_read`, and `cache_creation` stays 0 — which keeps `input + cache_read` equal to the +/// context fill read from the same object. `None` when it reports no counts at all. +fn codex_token_usage(last: &Value) -> Option { + let field = |k: &str| last.get(k).and_then(Value::as_u64); + let input = field("inputTokens"); + let cached = field("cachedInputTokens"); + let output = field("outputTokens"); + if input.is_none() && cached.is_none() && output.is_none() { + return None; + } + let cached = cached.unwrap_or(0); + Some(TokenUsage { + input: input.unwrap_or(0).saturating_sub(cached), + cache_creation: 0, + cache_read: cached, + output: output.unwrap_or(0), + }) +} + #[cfg(test)] mod tests { use super::*; + + #[test] + fn codex_token_usage_counts_cached_input_inside_input_not_on_top() { + // OpenAI: 12_000 input of which 9_000 served from cache → 3_000 fresh + 9_000 cached, + // summing back to the 12_000 fill; no cache write on this backend. + let last = serde_json::json!({ "inputTokens": 12_000, "cachedInputTokens": 9_000, "outputTokens": 640 }); + assert_eq!( + codex_token_usage(&last), + Some(TokenUsage { input: 3_000, cache_creation: 0, cache_read: 9_000, output: 640 }) + ); + assert_eq!(codex_token_usage(&serde_json::json!({})), None); + } use crate::supervisor::model::{BackgroundTask, SlashCommand}; use std::sync::Mutex; @@ -2673,6 +2723,14 @@ mod tests { Some(42_000), "ring = last turn's input (current occupancy), NOT the 950k cumulative total" ); + // …while the cumulative total is exactly what the session total reads, cached input + // split out of the input (900k of which 0 cached) and reasoning left inside the output. + let usage = st.session_usage.expect("the thread total is carried"); + assert_eq!( + usage.total, + TokenUsage { input: 900_000, cache_creation: 0, cache_read: 0, output: 40_000 } + ); + assert_eq!(usage.cost_usd, None, "Codex reports no cost"); } #[test] diff --git a/src-tauri/src/supervisor/history.rs b/src-tauri/src/supervisor/history.rs index 1d89a1bb..b9020c92 100644 --- a/src-tauri/src/supervisor/history.rs +++ b/src-tauri/src/supervisor/history.rs @@ -23,8 +23,10 @@ use serde::{Deserialize, Serialize}; use serde_json::{json, Value}; use specta::Type; -use super::assembler::{context_used_from_usage, normalize_blocks}; -use super::model::{ContextFill, ConversationItem, GoalState}; +use super::assembler::{ + context_used_from_usage, normalize_blocks, session_usage_from_model_usage, token_usage_from, +}; +use super::model::{ContextFill, ConversationItem, GoalState, SessionUsage}; /// Claude's config dir: `$CLAUDE_CONFIG_DIR` if set, else `$HOME/.claude`. Shared /// with [`super::subagents`], which reads the sibling task-artifact directories. @@ -154,9 +156,15 @@ fn load_context_fill_in(config_dir: &Path, session_id: &str) -> ContextFill { } // The freshest main-thread usage wins (largest accumulated context); keep // scanning so the LAST one sticks. - if let Some(used) = message.get("usage").and_then(context_used_from_usage) { + let usage = message.get("usage"); + if let Some(used) = usage.and_then(context_used_from_usage) { fill.context_tokens = Some(used); } + // Its breakdown, from the same line — so a reopened conversation shows what its fill + // is made of before a single new turn runs. + if let Some(breakdown) = usage.and_then(token_usage_from) { + fill.context_usage = Some(breakdown); + } } fill } @@ -273,6 +281,102 @@ fn load_active_goal_in(config_dir: &Path, session_id: &str) -> Option active } +/// Read a conversation's cumulative, all-agent token spend from its on-disk transcript — the +/// seed of the side panel's session total on reopen, before the first live `result` refreshes +/// it (see [`SessionUsage`]). +/// +/// The source is the transcript's LAST `{"type":"cost-state", modelUsage, totalCostUSD, …}` +/// line: a DISK-ONLY record the CLI writes when its process exits (and before a resume/fork +/// switch), and the very line it restores its own counters from on `--resume`. So the seed is +/// the value a resumed process will carry on from — or, when the transcript has none (the +/// process was killed before it could write one, an older CLI), `None`: the total is unknown, +/// never zero. A session whose process is still running has no fresh line yet; its last one +/// (if any) is "as of the last close". +/// +/// Cheap by construction: the line is normally the file's LAST, so only the tail is read, with +/// a raw-substring pre-filter before any JSON parse; the whole file is read only when the tail +/// has no such line (see [`read_last_matching_line`]). +pub fn load_session_usage(session_id: &str) -> Option { + let dir = claude_config_dir()?; + load_session_usage_in(&dir, session_id) +} + +fn load_session_usage_in(config_dir: &Path, session_id: &str) -> Option { + let path = find_transcript(config_dir, session_id)?; + read_last_matching_line(&path, "\"cost-state\"", |entry| { + if entry.get("type").and_then(Value::as_str) != Some("cost-state") { + return None; + } + let cost = entry.get("totalCostUSD").and_then(Value::as_f64); + session_usage_from_model_usage(entry.get("modelUsage")?, cost) + }) +} + +/// How much of a JSON-lines file's END [`read_last_matching_line`] reads before falling back to +/// the whole file. A `cost-state` or `token_count` line weighs about a kilobyte; the room is for +/// whatever a session appended after it (a resumed turn's first lines). +const TAIL_READ_BYTES: u64 = 256 * 1024; + +/// The first line, scanning a JSON-lines file BACKWARDS, that contains `marker` verbatim and +/// that `pick` accepts once parsed — what a reader of the LAST record of some kind needs. +/// +/// Reads the file's last [`TAIL_READ_BYTES`] first, and the whole file only when that tail +/// holds no accepted line. The tail's first line may be cut in half: it simply fails to parse +/// and is skipped, and the full read then sees it whole. `marker` is a PRE-FILTER only (a line +/// without it is never parsed; a line with it is still checked by `pick`), so a message that +/// merely mentions the marker cannot pass for the record. An absent or unreadable file is +/// `None` — logged when it is a real IO error, since the caller shows "unknown" either way. +pub(crate) fn read_last_matching_line( + path: &Path, + marker: &str, + pick: impl Fn(&Value) -> Option, +) -> Option { + use std::io::{Read, Seek, SeekFrom}; + let scan = |content: &str| { + content + .lines() + .rev() + .filter(|line| line.contains(marker)) + .find_map(|line| serde_json::from_str::(line.trim()).ok().and_then(|v| pick(&v))) + }; + let log = |e: &std::io::Error| { + if e.kind() != std::io::ErrorKind::NotFound { + eprintln!("[history] cannot read {}: {e}", path.display()); + } + }; + let mut file = match std::fs::File::open(path) { + Ok(file) => file, + Err(e) => { + log(&e); + return None; + } + }; + let len = file.metadata().map(|m| m.len()).unwrap_or(0); + if len > TAIL_READ_BYTES { + let mut tail = Vec::with_capacity(TAIL_READ_BYTES as usize); + let read = file + .seek(SeekFrom::Start(len - TAIL_READ_BYTES)) + .and_then(|_| file.read_to_end(&mut tail)); + match read { + // Lossy: the cut can land inside a multi-byte character, which only ever spoils + // the cut (first) line — the one that is incomplete anyway. + Ok(_) => { + if let Some(found) = scan(&String::from_utf8_lossy(&tail)) { + return Some(found); + } + } + Err(e) => log(&e), + } + } + match std::fs::read_to_string(path) { + Ok(content) => scan(&content), + Err(e) => { + log(&e); + None + } + } +} + /// The inner text of a `user` line whose content is a single `…` /// (what a LOCAL slash command like `/goal` prints back). `None` for any other user line. fn local_command_stdout(entry: &Value) -> Option { @@ -1930,6 +2034,78 @@ mod tests { assert_eq!(fill.context_window, None); } + #[test] + fn session_usage_seed_reads_the_last_cost_state_line() { + let base = std::env::temp_dir().join(format!("tosse-usage-{}", std::process::id())); + std::fs::remove_dir_all(&base).ok(); + let sid = "11111111-2222-3333-4444-555555555555"; + write_transcript( + &base, + "-some-cwd", + sid, + &[ + // An OLDER close of the session… + r#"{"type":"cost-state","sessionId":"x","totalCostUSD":0.5,"modelUsage":{"claude-opus-5":{"inputTokens":10,"outputTokens":20,"thinkingTokens":5,"cacheReadInputTokens":300,"cacheCreationInputTokens":40,"costUSD":0.5}}}"#, + r#"{"type":"user","uuid":"u1","message":{"role":"user","content":"hi"}}"#, + // A message that merely MENTIONS the record is not the record. + r#"{"type":"user","uuid":"u2","message":{"role":"user","content":"what is \"cost-state\"?"}}"#, + // …then the latest one: it wins, both models counted, thinking not added. + r#"{"type":"cost-state","sessionId":"x","totalCostUSD":1.25,"modelUsage":{"claude-haiku-4-5":{"inputTokens":2228,"outputTokens":31,"thinkingTokens":0,"cacheReadInputTokens":0,"cacheCreationInputTokens":0,"costUSD":0.002},"claude-opus-5":{"inputTokens":12,"outputTokens":607,"thinkingTokens":242,"cacheReadInputTokens":1000,"cacheCreationInputTokens":79877,"costUSD":1.248}}}"#, + r#"{"type":"user","uuid":"u3","message":{"role":"user","content":"resumed"}}"#, + ], + ); + let u = load_session_usage_in(&base, sid).expect("a cost-state line is on disk"); + assert_eq!( + u.total, + crate::supervisor::model::TokenUsage { + input: 2228 + 12, + cache_creation: 79877, + cache_read: 1000, + output: 31 + 607, + } + ); + assert_eq!(u.cost_usd, Some(1.25)); + assert_eq!(u.per_model[0].model, "claude-opus-5"); + + // No record at all (a killed process, an older CLI): unknown, never zero. + let none = "66666666-2222-3333-4444-555555555555"; + write_transcript( + &base, + "-some-cwd", + none, + &[r#"{"type":"user","uuid":"u1","message":{"role":"user","content":"hi"}}"#], + ); + assert_eq!(load_session_usage_in(&base, none), None); + // No transcript at all (a remote conversation, a new one). + assert_eq!(load_session_usage_in(&base, "missing"), None); + std::fs::remove_dir_all(&base).ok(); + } + + #[test] + fn last_matching_line_falls_back_to_the_whole_file_past_the_tail() { + let base = std::env::temp_dir().join(format!("tosse-tail-{}", std::process::id())); + std::fs::remove_dir_all(&base).ok(); + std::fs::create_dir_all(&base).unwrap(); + let path = base.join("t.jsonl"); + let mut f = std::fs::File::create(&path).unwrap(); + writeln!(f, r#"{{"type":"cost-state","n":1}}"#).unwrap(); + // Far more than the tail window after the record, none of it matching. + let filler = format!(r#"{{"type":"user","pad":"{}"}}"#, "x".repeat(1000)); + for _ in 0..(TAIL_READ_BYTES / 1000 + 50) { + writeln!(f, "{filler}").unwrap(); + } + drop(f); + let pick = |v: &Value| v.get("n").and_then(Value::as_u64); + assert_eq!(read_last_matching_line(&path, "\"cost-state\"", pick), Some(1)); + // A record inside the tail is found without the full read, and the LAST one wins. + let mut f = std::fs::OpenOptions::new().append(true).open(&path).unwrap(); + writeln!(f, r#"{{"type":"cost-state","n":2}}"#).unwrap(); + writeln!(f, "{filler}").unwrap(); + drop(f); + assert_eq!(read_last_matching_line(&path, "\"cost-state\"", pick), Some(2)); + std::fs::remove_dir_all(&base).ok(); + } + #[test] fn active_goal_tracks_set_unmet_achieved_and_clear() { let base = std::env::temp_dir().join(format!("tosse-goal-{}", std::process::id())); diff --git a/src-tauri/src/supervisor/model.rs b/src-tauri/src/supervisor/model.rs index 8a7f04af..7587603c 100644 --- a/src-tauri/src/supervisor/model.rs +++ b/src-tauri/src/supervisor/model.rs @@ -72,6 +72,12 @@ pub struct SessionStatePayload { /// `input + cache_creation + cache_read` (from `message_start` live, then the /// `result`). `None` until the first turn reports usage. Drives the context ring. pub context_tokens: Option, + /// The same last model call, broken down: what `context_tokens` is made of (fresh input, + /// cache written, cache read) plus the tokens it generated. `None` until a call reports + /// usage. Set wherever `context_tokens` is, from the same `usage` object — the two never + /// disagree — and its `output` completed by the call's `message_delta`. Drives the + /// telemetry deck's token breakdown. + pub context_usage: Option, /// Size of the active model's context window (from `result.modelUsage[…].contextWindow`, /// e.g. 200k or 1M for Opus in 1M mode). `None` until a `result` reports it; once /// known it is kept across turns that omit it. The ring's denominator. @@ -80,6 +86,57 @@ pub struct SessionStatePayload { /// the CLI emits one. NOTE: the stream only carries status + reset, NOT a usage /// percentage — that lives behind the `/api/oauth/usage` endpoint (separate task). pub rate_limit: Option, + /// What the WHOLE session has consumed so far, every agent included — see + /// [`SessionUsage`]. Claude: the latest `result.modelUsage` (a cumulative snapshot the CLI + /// re-sends in full at each turn end). Codex: the latest `thread/tokenUsage/updated` + /// `total` (this thread only). `None` until the first of those arrives; once known it is + /// REPLACED by each newer snapshot, never summed with it. + /// + /// `serde(default)` keeps it OPTIONAL on the TypeScript side (`session_usage?:`), so the + /// hand-written state literals of older tests and mocks stay valid without it. + #[serde(default)] + pub session_usage: Option, +} + +/// A session's CUMULATIVE token spend, as the CLI itself counts it. +/// +/// ⚠️ ONE snapshot, never a sum. On Claude it is `result.modelUsage` summed over its models: +/// every model call the CLI's query pipeline made — the main loop, `Task` sub-agents at every +/// depth, sidechains, compaction and Workflow agents (the binary's own schema text), the +/// permission classifier excepted. Each `result` carries the running total, so the latest +/// one REPLACES the previous: adding results would count turn 1 N times. `result.usage` (the +/// main loop's turn only) and every per-agent "tokens" figure of the wire (`task_notification`, +/// ``, a workflow manifest's `totalTokens`) are SUBSETS or a different unit +/// (an agent's last call ≈ its final context size) — none of them may ever be added to this. +/// +/// Claude restores it on `--resume` from the transcript's last `cost-state` line (written when +/// the process exits), which is also what [`super::history::load_session_usage`] reads to seed +/// a reopened conversation. A mid-session `/clear` resets it: the value can go DOWN. +/// +/// Codex: the thread's lifetime `tokenUsage.total` — collab sub-agents run as SEPARATE threads +/// and are NOT in it; no cost, no per-model split. +#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize, Type)] +pub struct SessionUsage { + /// Every model's usage added up. `thinkingTokens` is already INSIDE `output` (never added + /// again); on Codex the cached input is inside the input, split out as `cache_read`. + pub total: TokenUsage, + /// The CLI's cost ESTIMATE for the same calls at API list prices (`result.total_cost_usd` + /// live, `totalCostUSD` on disk) — cumulative like the tokens, and not a bill on a plan. + /// `None` when the CLI reports none (Codex never does). + pub cost_usd: Option, + /// The same total split per model (a helper Haiku next to the conversation's Opus…), + /// largest first. Empty on Codex, which reports the thread as a whole. + pub per_model: Vec, +} + +/// One model's share of a [`SessionUsage`]. +#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize, Type)] +pub struct ModelTokenUsage { + /// The model id as the CLI keys it (it may carry a `[1m]` suffix). + pub model: String, + pub usage: TokenUsage, + /// This model's share of the cost estimate (`costUSD`), when reported. + pub cost_usd: Option, } /// One selectable model, as the RUNNING session reports it via the `list_models` @@ -319,6 +376,21 @@ pub struct RemoteControlState { pub struct ContextFill { pub context_tokens: Option, pub context_window: Option, + /// The breakdown of `context_tokens` (see `SessionState::context_usage`), from the same + /// transcript line. + pub context_usage: Option, +} + +/// The four token counts of one model call's `usage` (or a turn's aggregate): the prompt as +/// the API bills it — fresh `input`, `cache_creation` (prompt written to the cache), +/// `cache_read` (prompt served from it) — and the `output` generated. Their prompt part sums +/// to the context occupancy: `input + cache_creation + cache_read`. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize, Type)] +pub struct TokenUsage { + pub input: u64, + pub cache_creation: u64, + pub cache_read: u64, + pub output: u64, } /// The active `/goal` of a conversation (Claude Code's native goal feature: Claude keeps @@ -436,10 +508,17 @@ pub enum ConversationItem { total_cost_usd: Option, num_turns: Option, duration_ms: Option, - /// Cumulative model/API time this turn (the "N s of model" breakdown). + /// Model/API time spent during THIS turn (the "N s of model" breakdown), derived + /// from the CLI's cumulative per-session counter by the assembler. `None` when + /// unknown — the first turn of a resumed process (see + /// `Assembler::api_ms_baseline`) — or on a backend without the breakdown (Codex). duration_api_ms: Option, /// Time-to-first-token this turn; captured but not yet surfaced in the UI. ttft_ms: Option, + /// What the turn consumed: `result.usage`, the aggregate over every model call the + /// turn made (not its last call — that is `SessionState::context_usage`). `None` when + /// the result carries none (Codex reports usage per thread, not per turn). + usage: Option, }, /// A non-conversational notice surfaced in the timeline. Two families: /// - informational: `control_change` (a confirmed model/effort/mode move), diff --git a/src-tauri/src/supervisor/session.rs b/src-tauri/src/supervisor/session.rs index 8b2cae2e..3dc55d45 100644 --- a/src-tauri/src/supervisor/session.rs +++ b/src-tauri/src/supervisor/session.rs @@ -124,6 +124,11 @@ pub enum SessionCommand { /// `/compact` text command on the app-server). Fire-and-forget — a failure is /// surfaced by the Codex actor as a timeline notice. Compact, + /// Cut a REMOTE session's reconnect backoff short and retry now: the Mac's network + /// just came back, or the user just returned to the app. Only meaningful while the + /// actor is waiting out a backoff (see `run_actor`); a no-op everywhere else — an + /// attached remote session, a local one, Codex. + ReconnectNow, /// Tear the session down. `ack`, when present, is fired by the actor ONLY after the /// process is fully reaped (the graceful EOF→SIGTERM→SIGKILL ladder has run), so a /// caller can wait for the `claude` process to ACTUALLY be gone — required before any @@ -503,6 +508,12 @@ impl SessionHandle { self.send(SessionCommand::Compact).await } + /// Retry a remote session's lost link NOW instead of at the end of its backoff — see + /// [`SessionCommand::ReconnectNow`]. Fire-and-forget. + pub async fn reconnect_now(&self) -> Result<(), SessionError> { + self.send(SessionCommand::ReconnectNow).await + } + /// Request teardown WITHOUT waiting for the process to be reaped (the quit path uses /// this and then polls [`Sessions::is_empty`]). For a REMOTE session this only /// DETACHES — the server-side `claude` keeps running (quitting the app must never @@ -561,7 +572,13 @@ pub fn spawn_session( appmcp: Option>, ) -> Result { let (transport, msg_rx) = Transport::spawn(cfg.clone()).map_err(SessionError::Spawn)?; - let core = SessionCore::new(id.clone(), initial, emitter, transport.outbound(), appmcp); + let mut core = SessionCore::new(id.clone(), initial, emitter, transport.outbound(), appmcp); + // A brand-new session's cumulative model-time counter starts at zero, so even its + // first turn's model time is exact. A resumed / re-attached one restores an unknown + // total first (see `Assembler::api_ms_baseline`). + if cfg.resume.is_none() && cfg.attach.is_none() { + core.assembler.mark_fresh_session(); + } let (cmd_tx, cmd_rx) = mpsc::channel(64); tokio::spawn(run_actor(core, transport, msg_rx, cmd_rx, on_exit, cfg)); Ok(SessionHandle { id, cmd_tx }) @@ -1022,7 +1039,7 @@ async fn run_actor( } else { // Still in the same outage (the previous attempt spawned ssh but // never got an fd_attach): escalate the backoff. - delay = (delay * 2).min(std::time::Duration::from_secs(30)); + delay = next_reconnect_delay(delay, outage_attempts); } // A6: past the threshold against the SAME candidate, try the next // recorded address (see `next_candidate_after_failure`'s doc for the @@ -1046,6 +1063,13 @@ async fn run_actor( break 'outer false; } None => break 'outer false, + // The network came back, or the user returned to the app: retry + // now rather than sleep out a backoff that can reach minutes. + Some(SessionCommand::ReconnectNow) => {} + // A send while offline still fails (send_failed) — but the user is + // plainly here and wants THIS conversation: retry now too, so the + // next send can land instead of waiting out the backoff. + Some(cmd @ SessionCommand::SendUser { .. }) => core.on_command(cmd), Some(cmd) => { core.on_command(cmd); continue; } }, } @@ -1092,7 +1116,7 @@ async fn run_actor( } Err(e) => { eprintln!("[session] remote reconnect failed (retrying in {delay:?}): {e}"); - delay = (delay * 2).min(std::time::Duration::from_secs(30)); + delay = next_reconnect_delay(delay, outage_attempts); // A6: the spawn itself (forking the local ssh client) never even // got off the ground — mirrors the `!attach_seen` failure signal // above, against whichever candidate `cfg2.remote.host` (== the @@ -1410,6 +1434,31 @@ fn reattach_cursor_delta(lines_seen: u64, first_unparseable_offset: Option, } } +/// Failed reconnect attempts, in one outage, before the backoff turns patient. At the +/// quick cap (doubling from 1 s, then every 30 s) that is ~3 minutes: long enough to ride +/// out a blip, a Wi-Fi hand-off or a server reboot at full speed. +const QUICK_RECONNECT_ATTEMPTS: u32 = 10; +/// Backoff ceiling while the outage is young. +const QUICK_RECONNECT_CAP: std::time::Duration = std::time::Duration::from_secs(30); +/// Backoff ceiling once it is not. A long outage is a plane, a laptop off the network, a +/// server switched off — every attempt forks `ssh` and dials for up to 10 s, per remote +/// conversation, for as long as it lasts. The two moments the link is likely back — the +/// Mac's network returning, the user coming back to the app — cut the wait short through +/// [`SessionCommand::ReconnectNow`], and so does a send; this ceiling is only the floor +/// of how fast an outage nobody is looking at gets noticed. +const PATIENT_RECONNECT_CAP: std::time::Duration = std::time::Duration::from_secs(300); + +/// The next backoff of a remote session's reconnect loop, after `outage_attempts` +/// failed attempts in the current outage. Pure — see [`QUICK_RECONNECT_ATTEMPTS`]. +fn next_reconnect_delay(delay: std::time::Duration, outage_attempts: u32) -> std::time::Duration { + let cap = if outage_attempts >= QUICK_RECONNECT_ATTEMPTS { + PATIENT_RECONNECT_CAP + } else { + QUICK_RECONNECT_CAP + }; + (delay * 2).min(cap) +} + /// A6 — sustained reconnect failure address rotation: how many CONSECUTIVE /// attempts against the SAME candidate must fail before `run_actor` tries the /// next recorded address. Small on purpose: a dead preferred address should not @@ -2421,6 +2470,9 @@ impl SessionCore { // slash-command turn the composer sends directly), so there's nothing to do // on the control channel here. SessionCommand::Compact => {} + // Only acted on inside `run_actor`'s backoff wait; outside an outage there is + // nothing to reconnect. + SessionCommand::ReconnectNow => {} // Shutdown is handled in the run loop (breaks before reaching here). SessionCommand::Shutdown { .. } => {} } @@ -2452,6 +2504,23 @@ mod tests { use crate::supervisor::model::{ConversationItem, PermissionRequestPayload, SessionStatePayload}; use serde_json::json; + /// Quick while the outage is young (a blip, a reboot), patient once it is not (a + /// plane): 1 s doubling to 30 s for the first attempts, then up to 5 min. + #[test] + fn reconnect_backoff_turns_patient_after_a_long_outage() { + let s = std::time::Duration::from_secs; + let mut delay = s(1); + let mut seen = Vec::new(); + for attempt in 1..=14 { + delay = next_reconnect_delay(delay, attempt); + seen.push(delay.as_secs()); + } + assert_eq!(seen, vec![2, 4, 8, 16, 30, 30, 30, 30, 30, 60, 120, 240, 300, 300]); + // The attempt count, not the delay, decides the ceiling. + assert_eq!(next_reconnect_delay(s(30), QUICK_RECONNECT_ATTEMPTS - 1), QUICK_RECONNECT_CAP); + assert_eq!(next_reconnect_delay(s(30), QUICK_RECONNECT_ATTEMPTS), s(60)); + } + /// Serialises tests that mutate the process-wide `TOSSE_SSH_BIN` env var — /// the SAME crate-wide lock `transport::tests` uses for its own direct /// `push_remote_title`/`run_remote_stop` tests (see diff --git a/src-tauri/src/supervisor/transport.rs b/src-tauri/src/supervisor/transport.rs index c303c65d..c645be3c 100644 --- a/src-tauri/src/supervisor/transport.rs +++ b/src-tauri/src/supervisor/transport.rs @@ -919,10 +919,16 @@ impl Transport { // A REAL network cut (wifi off, cable pulled) sends no FIN: without // keepalives the ssh client hangs on a dead TCP connection for many // minutes and the actor never sees the EOF that triggers its - // auto-reconnect. 5s probes × 3 misses → a dead link is detected in - // ~15s and the reattach/replay path takes over. + // auto-reconnect. 10s probes × 3 misses → a dead link is detected in + // ~30s and the reattach/replay path takes over. + // + // ⚠️ Every live remote conversation holds its OWN ssh, and each probe is + // an encrypted round trip that wakes the Wi-Fi radio (and Tailscale's + // tunnel) even when nothing else is happening. 5s was twice the traffic + // at rest for 15s of detection nobody needed (decided 28/09: ~30s is the + // accepted time to notice a cut). .arg("-o") - .arg("ServerAliveInterval=5") + .arg("ServerAliveInterval=10") .arg("-o") .arg("ServerAliveCountMax=3") .arg("-o") diff --git a/src-tauri/src/tosse/sse.rs b/src-tauri/src/tosse/sse.rs index 42e4d02a..b7323911 100644 --- a/src-tauri/src/tosse/sse.rs +++ b/src-tauri/src/tosse/sse.rs @@ -33,21 +33,24 @@ //! Against production (2026-08-11), an idle stream ends CLEANLY — a finished body, not an //! error — after **12.0 s**, reproducibly, over HTTP/1.1. It is an INACTIVITY timeout: a //! probe that received a `task:updated` at ~12 s stayed connected until 24 s, i.e. the clock -//! restarts on every byte. The server's own `: keepalive` is documented at 15 s, so it never -//! gets the chance to fire and the connection it exists to hold open is dropped first. +//! restarts on every byte. The server's own `: keepalive` was sent every 15 s, so it never +//! got the chance to fire and the connection it exists to hold open was dropped first. +//! +//! ✅ FIXED server-side (2026-09-28, CRM_max `bd6705f` + `a8456e5`). The culprit was NOT the +//! Railway edge (the first guess): it was `Bun.serve`'s own `idleTimeout` (10 s by default, +//! checked in ~4 s steps — hence ~12 s), which closes any request that stays quiet. The SSE +//! route now opts out of it (`server.timeout(req, 0)`) and the keepalive went to 10 s; +//! production logs show no more `/api/v1/sse` endings. The recycle handling below stays as +//! the safety net it always was — a redeploy or a proxy change can end a stream cleanly. //! //! Two consequences the loop is built around: //! - A clean end after a connection that RAN ([`ESTABLISHED_AFTER`]) **and delivered** is not //! a failure. It is recycled silently: no backoff, no state change, no indicator flicker //! every 12 s. An idle timeout never qualifies, however long the socket stayed open — see //! [`ended`], which is what keeps `Error` reachable for a stream that goes quiet. -//! - The refetch owed per connection is throttled ON THE FRONT, or a recycle every 12 s -//! would become the very polling this feature exists to remove. -//! -//! The real fix belongs to the CRM: a keepalive shorter than that timeout (~10 s) would hold -//! one connection open indefinitely and make all of this moot. Worth doing — it would also -//! close the ~200 ms gap each recycle leaves, which is the only window where a change can be -//! missed until the next refetch. +//! - A recycle owes the front NO refetch (`TosseLiveHost`): only a real outage — the state +//! left `live` — does. Refetching for every recycle was a sweep every minute for as long +//! as the window stayed visible: the very polling this feature exists to remove. use serde::{Deserialize, Serialize}; use specta::Type; @@ -64,7 +67,7 @@ const SSE_PATH: &str = "/api/v1/sse"; /// How long a silent socket is allowed to stay silent before we treat it as dead. /// -/// The server sends a `: keepalive` comment every 15 s, so three missed ones is a stream +/// The server sends a `: keepalive` comment every 10 s, so five missed ones is a stream /// that is no longer being served — a TCP connection can otherwise sit "open" for many /// minutes after the peer is gone, and a frozen view that claims to be live is the worst /// outcome this feature can produce. @@ -1000,7 +1003,7 @@ mod tests { let mut parser = SseParser::new(); let mut saw_bytes = false; - // Bounded: the server sends `connected` at once, then a keepalive every 15 s. + // Bounded: the server sends `connected` at once, then a keepalive every 10 s. let deadline = std::time::Instant::now() + std::time::Duration::from_secs(20); while std::time::Instant::now() < deadline { match tokio::time::timeout(std::time::Duration::from_secs(20), resp.chunk()).await { diff --git a/src-tauri/src/wake/capture.rs b/src-tauri/src/wake/capture.rs index cef2f2d0..0b632ffe 100644 --- a/src-tauri/src/wake/capture.rs +++ b/src-tauri/src/wake/capture.rs @@ -1,22 +1,28 @@ //! Microphone capture for the wake-word detector — the ONLY place `cpal` is used. //! //! Opens the default input device, downmixes to mono, resamples to the engine's -//! 16 kHz, and streams the samples (as `[-1, 1]` f32) to the worker thread over an -//! mpsc channel. Inference NEVER runs in the audio callback (a real-time thread); -//! the callback only converts + forwards, the worker does the ONNX work. +//! 16 kHz, and hands the samples (as `[-1, 1]` f32) to the worker thread over an +//! mpsc channel — BATCHED into whole detection steps (80 ms), so the worker wakes +//! ~12 times a second instead of once per audio callback (~94 a second for a +//! 512-frame buffer at 48 kHz). Inference NEVER runs in the audio callback (a +//! real-time thread): the callback only converts, batches and forwards, reusing +//! its buffers — the one allocation left is the batch it hands off, once a step. //! //! ⚠️ This is a SECOND consumer of the microphone, running alongside the voice //! agent's WebRTC `getUserMedia`. Whether macOS + WKWebView allow both to hold the //! input device at once is the open risk validated at the first `/build-app` — if //! it does not, detection moves into the webview instead. The rest of the design -//! is independent of that outcome. +//! is independent of that outcome. (That shared device is also why the device's +//! own buffer size is left at its default: batching here costs the other consumer +//! nothing, a bigger hardware buffer might.) use std::sync::mpsc::Sender; use cpal::traits::{DeviceTrait, HostTrait, StreamTrait}; -use cpal::{SampleFormat, Stream}; +use cpal::{SampleFormat, Stream, StreamError}; -use super::engine::SAMPLE_RATE; +use super::engine::{CHUNK, SAMPLE_RATE}; +use super::WorkerMsg; /// A running capture. Dropping it stops and releases the microphone (the macOS /// orange indicator goes off). `cpal::Stream` is `!Send`, so this lives on the @@ -26,9 +32,12 @@ pub struct Capture { } impl Capture { - /// Start capturing; every resampled mono chunk is sent on `tx`. Errors carry a - /// human message (surfaced as the wake status `error`), e.g. no microphone. - pub fn start(tx: Sender>) -> Result { + /// Start capturing; the audio goes out on `tx` one detection step at a time. + /// A stream failure that ends the capture (the device went away) is sent on the + /// same channel, so the worker stops — and says so — instead of waiting forever + /// on a dead microphone. Errors carry a human message (surfaced as the wake + /// status `error`), e.g. no microphone. + pub fn start(tx: Sender) -> Result { let host = cpal::default_host(); let device = host .default_input_device() @@ -41,34 +50,35 @@ impl Capture { let channels = supported.config().channels as usize; let config: cpal::StreamConfig = supported.config(); - let mut resampler = Resampler::new(src_rate); - let err_fn = |e| eprintln!("[wake] audio stream error: {e}"); + let mut pipe = Pipe::new(src_rate, channels, tx.clone()); + let err_fn = move |e: StreamError| { + eprintln!("[wake] audio stream error: {e}"); + // A vanished device never delivers again, so the worker is told and + // stops; the status then says why instead of claiming to listen. Other + // (backend) errors leave the stream running: logged only. + if matches!(e, StreamError::DeviceNotAvailable) { + let _ = tx.send(WorkerMsg::Fault(format!( + "the microphone went away ({e}) — turn the wake word off and on to listen again" + ))); + } + }; let stream = match sample_format { SampleFormat::F32 => device.build_input_stream( &config, - move |data: &[f32], _: &_| { - let mono = downmix(data, channels, |s| s); - forward(&mut resampler, &mono, &tx); - }, + move |data: &[f32], _: &_| pipe.push(data, |s| s), err_fn, None, ), SampleFormat::I16 => device.build_input_stream( &config, - move |data: &[i16], _: &_| { - let mono = downmix(data, channels, |s| s as f32 / 32768.0); - forward(&mut resampler, &mono, &tx); - }, + move |data: &[i16], _: &_| pipe.push(data, |s| s as f32 / 32768.0), err_fn, None, ), SampleFormat::U16 => device.build_input_stream( &config, - move |data: &[u16], _: &_| { - let mono = downmix(data, channels, |s| (s as f32 - 32768.0) / 32768.0); - forward(&mut resampler, &mono, &tx); - }, + move |data: &[u16], _: &_| pipe.push(data, |s| (s as f32 - 32768.0) / 32768.0), err_fn, None, ), @@ -86,26 +96,83 @@ impl Capture { } } -/// Resample `mono` to 16 kHz and send it — dropping the (rare) case where a -/// receiver has already gone away. -fn forward(resampler: &mut Resampler, mono: &[f32], tx: &Sender>) { - let mut out = Vec::new(); - resampler.process(mono, &mut out); - if !out.is_empty() { - let _ = tx.send(out); +/// Everything the audio callback does, with its buffers kept across calls: +/// downmix → resample → batch into steps → send. +struct Pipe { + channels: usize, + resampler: Resampler, + /// Downmix scratch, reused: it grows to the callback size once, then stays. + mono: Vec, + batch: StepBatcher, + tx: Sender, +} + +impl Pipe { + fn new(src_rate: u32, channels: usize, tx: Sender) -> Self { + Self { + channels, + resampler: Resampler::new(src_rate), + mono: Vec::new(), + batch: StepBatcher::new(), + tx, + } + } + + fn push(&mut self, data: &[T], conv: impl Fn(T) -> f32) { + downmix_into(data, self.channels, conv, &mut self.mono); + let before = self.batch.pending.len(); + self.resampler.process(&self.mono, &mut self.batch.pending); + let added = self.batch.pending.len() - before; + if let Some(batch) = self.batch.completed(added) { + // A gone receiver means the worker is stopping; nothing to do. + let _ = self.tx.send(WorkerMsg::Audio(batch)); + } } } -/// Average interleaved channels into a mono `f32` frame, applying `conv` to each -/// raw sample first. `channels == 1` is the common (already-mono) fast path. -fn downmix(data: &[T], channels: usize, conv: impl Fn(T) -> f32) -> Vec { +/// Room for one step plus the callback that completes it, so the batch buffer +/// never grows mid-callback. +const BATCH_CAPACITY: usize = CHUNK * 2; + +/// Holds resampled audio until it completes a detection step, then hands the +/// whole batch over. The engine counts steps from the first sample it receives — +/// the first sample batched here — so a batch goes out in the very callback that +/// completes a step: the worker wakes once per step, and no step is decided later +/// than when every callback was sent on its own. +struct StepBatcher { + pending: Vec, + /// Samples received since the last step boundary. + into_step: usize, +} + +impl StepBatcher { + fn new() -> Self { + Self { pending: Vec::with_capacity(BATCH_CAPACITY), into_step: 0 } + } + + /// Account for `added` samples just appended to `pending`; returns the batch to + /// send when they completed a step. + fn completed(&mut self, added: usize) -> Option> { + self.into_step += added; + if self.into_step < CHUNK { + return None; + } + self.into_step %= CHUNK; + Some(std::mem::replace(&mut self.pending, Vec::with_capacity(BATCH_CAPACITY))) + } +} + +/// Average interleaved channels into mono `f32` samples in `out` (cleared first), +/// applying `conv` to each raw sample. `channels == 1` is the common (already-mono) +/// fast path. +fn downmix_into(data: &[T], channels: usize, conv: impl Fn(T) -> f32, out: &mut Vec) { + out.clear(); let ch = channels.max(1); if ch == 1 { - return data.iter().map(|&s| conv(s)).collect(); + out.extend(data.iter().map(|&s| conv(s))); + return; } - data.chunks(ch) - .map(|frame| frame.iter().map(|&s| conv(s)).sum::() / ch as f32) - .collect() + out.extend(data.chunks(ch).map(|frame| frame.iter().map(|&s| conv(s)).sum::() / ch as f32)); } /// A tiny streaming linear resampler (device rate → 16 kHz). Carries a fractional @@ -166,7 +233,53 @@ mod tests { #[test] fn downmix_averages_stereo() { let stereo = [1.0f32, 3.0, 2.0, 4.0]; // L,R,L,R - assert_eq!(downmix(&stereo, 2, |s| s), vec![2.0, 3.0]); + let mut out = vec![9.0; 7]; // stale content from a previous callback + downmix_into(&stereo, 2, |s| s, &mut out); + assert_eq!(out, vec![2.0, 3.0]); + } + + /// The worker must wake once per detection step, in the callback that + /// completes it — and batching must never drop, duplicate or reorder a + /// sample, whatever the callback sizes (including one bigger than a step). + #[test] + fn batches_go_out_once_per_step_and_lose_nothing() { + let mut batcher = StepBatcher::new(); + let mut sent: Vec> = Vec::new(); + let mut total = 0usize; + let mut completed_steps = 0usize; + let sizes = [171usize, 170, 171, 512, 1365, 3000, 160, 171]; + for (i, &n) in sizes.iter().cycle().take(400).enumerate() { + batcher.pending.extend((total..total + n).map(|v| v as f32)); + total += n; + let batch = batcher.completed(n); + // A batch goes out exactly when this callback completed a new step. + assert_eq!(batch.is_some(), total / CHUNK > completed_steps, "callback {i}"); + completed_steps = total / CHUNK; + sent.extend(batch); + } + let mut flat: Vec = sent.concat(); + flat.extend_from_slice(&batcher.pending); + assert_eq!(flat.len(), total); + assert!(flat.iter().enumerate().all(|(i, &v)| v == i as f32), "order and content kept"); + } + + /// At the common Mac cadence (512-frame callbacks at 48 kHz → ~171 samples at + /// 16 kHz, ~94 callbacks a second) the worker wakes 12.5 times a second. + #[test] + fn the_worker_wakes_once_per_step_at_the_usual_mac_cadence() { + let mut pipe_rate = Resampler::new(48_000); + let mut batcher = StepBatcher::new(); + let mut wakeups = 0usize; + let callbacks = 94 * 10; // ~10 s + for _ in 0..callbacks { + let before = batcher.pending.len(); + pipe_rate.process(&[0.0f32; 512], &mut batcher.pending); + let added = batcher.pending.len() - before; + wakeups += batcher.completed(added).is_some() as usize; + } + let seconds = callbacks as f64 * 512.0 / 48_000.0; + let per_second = wakeups as f64 / seconds; + assert!((12.0..=13.0).contains(&per_second), "{per_second:.2} wakeups/s"); } #[test] diff --git a/src-tauri/src/wake/debug.rs b/src-tauri/src/wake/debug.rs index 2c738015..9cdcedb8 100644 --- a/src-tauri/src/wake/debug.rs +++ b/src-tauri/src/wake/debug.rs @@ -94,8 +94,8 @@ pub fn write_capture( "audio_samples": audio.len(), "audio_file": wav_path.file_name().map(|n| n.to_string_lossy().to_string()), // Oldest step first, ending with the one that fired. `vad` is Silero's - // speech probability (diagnostic only today — it gates nothing) and `rms` - // the level of the window that step scored. + // speech probability — the fire-time veto reads its peak over the last 16 + // steps — and `rms` the level of the window that step scored. "trace": detection .trace .iter() diff --git a/src-tauri/src/wake/engine.rs b/src-tauri/src/wake/engine.rs index 281ec8a7..cffd1a83 100644 --- a/src-tauri/src/wake/engine.rs +++ b/src-tauri/src/wake/engine.rs @@ -3,10 +3,12 @@ //! Two stacked models, both bundled into the binary (`include_bytes!`), so the //! feature works offline on first launch: //! -//! 1. **Silero VAD** gates everything: the neural nets below run ONLY while it -//! hears speech (the battery guard — silence costs ~one tiny VAD inference -//! per 512 samples, not the full stack). VAD failures degrade to "always -//! active" rather than killing the feature. +//! 1. **Silero VAD**, on every 512 samples. It does not decide what runs: it +//! VETOES a fire unless it heard speech somewhere in the trailing ~1.3 s +//! (`VAD_VETO_MIN`). There is no silence-based battery guard — the whole +//! neural pipeline below runs on every step, speech or silence, for as long +//! as the wake word is on. A VAD failure fails OPEN (reported as certain +//! speech, so the veto never blocks) rather than killing the feature. //! //! 2. **openWakeWord** proper, a three-stage pipeline on 16 kHz mono audio: //! audio → melspectrogram (32-bin) → shared speech embedding (96-dim) → @@ -22,6 +24,14 @@ //! classifier scores them. This is a faithful streaming approximation — the exact //! recall is tuned by the sensitivity threshold and validated with a real mic //! (the /build-app spike), not asserted headlessly. +//! +//! ⚠️ The melspectrogram is recomputed over the whole trailing window on purpose, +//! not incrementally over the new audio the way openWakeWord's own streaming code +//! does: the model floors every value at 80 dB below the loudest value of ITS +//! input, so a quiet frame's value depends on everything computed with it. An +//! incremental version would feed the classifier features this pipeline was never +//! tuned on, to save ~13 µs a step (the model's cost is almost all fixed per-call +//! overhead). `the_melspectrogram_depends_on_its_whole_window` pins the reason. use ort::session::Session; use ort::value::Tensor; @@ -30,8 +40,10 @@ use ort::value::Tensor; /// input down to this before anything here runs. pub const SAMPLE_RATE: u32 = 16_000; -/// One detection step per 80 ms of new audio (openWakeWord's step). -const CHUNK: usize = 1280; +/// One detection step per 80 ms of new audio (openWakeWord's step). `capture.rs` +/// batches the microphone into whole steps, so the worker thread wakes once per +/// step instead of once per audio callback. +pub const CHUNK: usize = 1280; /// Mel frames per embedding window, and the embedding stride. const EMB_FRAMES: usize = 76; /// Number of consecutive embeddings the classifier scores. @@ -95,8 +107,8 @@ const DEBUG_AUDIO_SAMPLES: usize = SAMPLE_RATE as usize * 4; pub struct StepTrace { /// The classifier's probability for this step. pub score: f32, - /// Silero's PEAK speech probability over the audio that fed this step. - /// DIAGNOSTIC ONLY — the VAD gates nothing today (see `feed`). + /// Silero's PEAK speech probability over the audio that fed this step — what + /// the fire-time veto reads over the trailing window (see `VAD_VETO_MIN`). pub vad: f32, /// RMS of the trailing mel window, in [0, 1]. Near zero means the step scored /// (near-)silence — which no spoken phrase can do, so a fire there points at @@ -196,7 +208,9 @@ fn run_single(session: &mut Session, shape: Vec, data: Vec) -> Result< fn build_session(bytes: &[u8]) -> Result { Session::builder() .map_err(|e| e.to_string())? - // A background listener has no business grabbing every core. + // A background listener has no business grabbing every core. One intra-op + // thread also means inference runs ON the worker thread, so the worker's + // QoS class (see `run_worker`) is what schedules it. .with_intra_threads(1) .map_err(|e| e.to_string())? .commit_from_memory(bytes) @@ -205,7 +219,7 @@ fn build_session(bytes: &[u8]) -> Result { /// The Silero VAD, kept separate so its failures never take the wake pipeline /// down: if the model errors even once, `disabled` latches and the caller treats -/// every frame as speech (higher CPU, still functional). +/// every frame as speech (the veto then never blocks; detection still works). struct Vad { session: Session, state: Vec, @@ -230,7 +244,7 @@ impl Vad { /// Feed new audio; return the PEAK speech probability across the frames this /// call completed (0.0 when it completed none — callers accumulate with `max`, /// for which that is a no-op). Returning the probability rather than a boolean - /// is what lets it ride in the trace and, later, veto a fire. On any model + /// is what lets it ride in the trace and veto a fire. On any model /// error the VAD latches off and reports certain speech (fail-open — never /// silence the feature over a VAD glitch). fn speech_prob(&mut self, samples: &[f32]) -> f32 { @@ -363,21 +377,19 @@ impl Engine { /// moment the configured phrase is detected (at most once per spoken phrase, /// thanks to the cooldown). `None` otherwise. pub fn feed(&mut self, samples: &[f32]) -> Option { - // VAD is computed for diagnostics only — it does NOT gate the pipeline. + // The VAD feeds the fire-time veto only — it does NOT gate the pipeline. // openWakeWord needs a CONTINUOUS rolling window of embeddings (~2 s); a // hard VAD gate that ran the stack only during speech (and cleared the // embedding buffer on silence) mis-anchored that window and starved the // classifier of the 16 consecutive embeddings it needs — so a short // utterance never fired. Run the stack every step. // - // Reading this probability as a FIRE-TIME VETO would not touch the rolling - // window, and is the next move against false positives. ⚠️ When that lands, - // veto on the peak across the TRAILING WINDOW, never on the firing step's - // own 80 ms: the classifier scores ~2 s of context, so it typically fires - // one or two steps AFTER the phrase ends, on new audio that is already + // The veto reads the peak across the TRAILING WINDOW, never the firing + // step's own 80 ms: the classifier scores ~2 s of context, so it typically + // fires one or two steps AFTER the phrase ends, on new audio that is already // silent. Measured on a `say "Alexa"` clip: the step that scores 1.000 // carries vad=0.01, while the steps holding the phrase itself read ~1.0. - // A naive per-step veto would therefore reject every real detection. + // A per-step veto would therefore reject every real detection. self.vad_peak = self.vad_peak.max(self.vad.speech_prob(samples)); self.audio.extend_from_slice(samples); @@ -637,9 +649,9 @@ mod tests { /// Push a synthetic tone through mel → embedding → classifier directly, /// asserting every ONNX run succeeds and the shapes line up end-to-end. The - /// streaming path logs-and-swallows inference errors (and VAD gates the stack - /// out of the silence test), so a shape bug could hide until a live mic — this - /// catches it headlessly. Detection ACCURACY still needs a real mic. + /// streaming path logs-and-swallows inference errors, so a shape bug could + /// hide until a live mic — this catches it headlessly. Detection ACCURACY + /// still needs a real mic. #[test] fn pipeline_shapes_line_up_end_to_end() { let mut mel = build_session(MEL_MODEL).unwrap(); @@ -790,6 +802,70 @@ mod tests { ); } + /// The capture now hands the engine one batch per 80 ms step instead of every + /// ~10 ms callback. Batching must change no DECISION: the same fires and the + /// same suppressed candidates (debug on, so the gates' rejections show too), + /// with bit-identical scores. The batches are cut exactly the way `capture.rs` + /// cuts them — flushed in the callback that completes a step — so each one + /// completes at most ONE step. That matters: `feed` returns only the last + /// candidate of a call, so a call spanning several steps could swallow a + /// debug-only suppressed candidate (fires are safe — the cooldown follows them). + #[test] + fn batching_the_microphone_changes_no_decision() { + let alexa = parse_wav_i16(SPEECH_WAV); + let tone = |n: usize| pseudo_noise(n, 0.002); + let second = SAMPLE_RATE as usize; + let audio: Vec = [tone(second), alexa.clone(), tone(second * 2), alexa, tone(second)] + .concat(); + let callback = 171; // one 512-frame 48 kHz CoreAudio callback, resampled + let run = |batched: bool| { + let mut engine = Engine::new(DEFAULT_PHRASE, 0.5, true).expect("engine builds"); + let mut out = Vec::new(); + let (mut pending, mut into_step) = (Vec::new(), 0usize); + for c in audio.chunks(callback) { + pending.extend_from_slice(c); + into_step += c.len(); + if batched && into_step < CHUNK { + continue; // the batcher holds it until a step completes + } + into_step %= CHUNK; + out.extend(engine.feed(&std::mem::take(&mut pending))); + } + out.into_iter().map(|d| (d.suppressed_by, d.score.to_bits())).collect::>() + }; + let per_callback = run(false); + assert!( + per_callback.iter().any(|(gate, _)| gate.is_none()), + "the fixture must fire, or this proves nothing: {per_callback:?}" + ); + assert_eq!(run(true), per_callback, "batching per step changed a decision"); + } + + /// Why the melspectrogram is recomputed over the whole trailing window every + /// step instead of incrementally over the new audio (openWakeWord's own + /// streaming — and the "obvious" optimisation): the model floors every value at + /// 80 dB below the loudest value of ITS INPUT, so a quiet frame computed next to + /// speech is not the frame computed next to silence. Pinned so the change is not + /// made by accident: it would feed the classifier features the pipeline was + /// never tuned on (measured: scores moved by up to 5e-3). + #[test] + fn the_melspectrogram_depends_on_its_whole_window() { + let mut mel = build_session(MEL_MODEL).unwrap(); + // Digital silence, then speech: far more than 80 dB inside one window. + let mut audio = vec![0.0f32; 1760]; + audio.extend(parse_wav_i16(SPEECH_WAV)); + audio.resize(MEL_WINDOW_SAMPLES, 0.0); + let scaled: Vec = audio.iter().map(|s| s * 32768.0).collect(); + let (_, full) = + run_single(&mut mel, vec![1, MEL_WINDOW_SAMPLES as i64], scaled.clone()).unwrap(); + // The same first 8 frames (all silence), computed on their own. + let (_, alone) = run_single(&mut mel, vec![1, 1632], scaled[..1632].to_vec()).unwrap(); + let floor = full.iter().copied().fold(f32::NEG_INFINITY, f32::max) - 80.0; + assert!(full.iter().all(|&v| v >= floor), "nothing sits below the window's floor"); + assert!(full.iter().any(|&v| v == floor), "the silence was lifted exactly TO it"); + assert_ne!(alone[..], full[..alone.len()], "the same frames computed alone differ"); + } + /// A muted or dead input device delivers values the mel stage turns non-finite. /// Those steps must be DROPPED, never scored: a NaN through the classifier can /// come back as a confident hit. @@ -919,6 +995,106 @@ mod tests { eprintln!("[corpus] {} files: {fired} FIRED, suppressed {suppressed:?}", paths.len()); } + /// CPU time consumed by the CALLING thread, in seconds. Every session is built + /// with one intra-op thread, so the ONNX work runs on the thread that calls + /// `run` — this is the engine's real cost, without the rest of the test binary. + fn thread_cpu_secs() -> f64 { + let mut ts = libc::timespec { tv_sec: 0, tv_nsec: 0 }; + // SAFETY: `ts` is a valid, writable timespec for the duration of the call. + unsafe { libc::clock_gettime(libc::CLOCK_THREAD_CPUTIME_ID, &mut ts) }; + ts.tv_sec as f64 + ts.tv_nsec as f64 * 1e-9 + } + + /// Diagnostic (ignored) micro-benchmark: CPU per second of audio through the + /// whole engine, in silence (room tone) and in continuous speech (the fixture + /// looped), fed per ~10 ms callback (171) and per 80 ms step (what the capture + /// now sends). `WAKE_BENCH_QOS=utility` runs it at the worker's QoS class. + /// cargo test --release --lib wake::engine::tests::bench_engine_cpu -- --ignored --nocapture + #[test] + #[ignore] + fn bench_engine_cpu() { + #[cfg(target_os = "macos")] + if std::env::var("WAKE_BENCH_QOS").as_deref() == Ok("utility") { + super::super::lower_thread_qos(); + eprintln!("[bench] running at UTILITY QoS"); + } + let seconds = 30usize; + let n = SAMPLE_RATE as usize * seconds; + let clip = parse_wav_i16(SPEECH_WAV); + let speech: Vec = clip.iter().copied().cycle().take(n).collect(); + let cases = [("silence (room tone)", pseudo_noise(n, 0.002)), ("speech (fixture looped)", speech)]; + for (label, audio) in &cases { + for chunk in [171usize, CHUNK] { + let mut engine = Engine::new(DEFAULT_PHRASE, 0.5, false).expect("engine builds"); + for c in pseudo_noise(SAMPLE_RATE as usize * 3, 0.002).chunks(chunk) { + engine.feed(c); // warm-up: prime the buffers and sessions, not measured + } + let t0 = thread_cpu_secs(); + let mut slowest = std::time::Duration::ZERO; + for c in audio.chunks(chunk) { + let w0 = std::time::Instant::now(); + engine.feed(c); + slowest = slowest.max(w0.elapsed()); + } + let cpu = thread_cpu_secs() - t0; + eprintln!( + "[bench] {label:<24} chunk={chunk:>4}: {:.2} ms CPU per audio second \ + ({:.2}% of one core), slowest call {:.1} ms", + cpu * 1000.0 / seconds as f64, + cpu * 100.0 / seconds as f64, + slowest.as_secs_f64() * 1000.0, + ); + } + } + } + + /// Diagnostic (ignored): the cost of each stage on its own, per call and per + /// second of audio at the rate the engine calls it. Tells where the time goes. + /// cargo test --release --lib wake::engine::tests::bench_stage_costs -- --ignored --nocapture + #[test] + #[ignore] + fn bench_stage_costs() { + let reps = 300usize; + let scaled: Vec = + pseudo_noise(MEL_WINDOW_SAMPLES, 0.05).iter().map(|s| s * 32768.0).collect(); + let mut mel = build_session(MEL_MODEL).unwrap(); + let mut embedding = build_session(EMBEDDING_MODEL).unwrap(); + let mut classifier = build_session(classifier_bytes(DEFAULT_PHRASE)).unwrap(); + let mut vad = Vad::new().unwrap(); + let steps_per_s = SAMPLE_RATE as f64 / CHUNK as f64; + let vad_per_s = SAMPLE_RATE as f64 / VAD_FRAME as f64; + let time = |label: &str, per_s: f64, f: &mut dyn FnMut()| { + f(); // warm-up + let t0 = thread_cpu_secs(); + for _ in 0..reps { + f(); + } + let per_call = (thread_cpu_secs() - t0) / reps as f64; + eprintln!( + "[stage] {label:<34} {:>8.1} µs/call × {per_s:>5.2}/s = {:>6.2} ms per audio second", + per_call * 1e6, + per_call * per_s * 1000.0 + ); + }; + time("mel over the trailing window", steps_per_s, &mut || { + run_single(&mut mel, vec![1, scaled.len() as i64], scaled.clone()).unwrap(); + }); + let emb_in = vec![1.0f32; EMB_FRAMES * MEL_BINS]; + time("embedding (76 frames)", steps_per_s, &mut || { + run_single(&mut embedding, vec![1, EMB_FRAMES as i64, MEL_BINS as i64, 1], emb_in.clone()) + .unwrap(); + }); + let cls_in = vec![0.1f32; CLASSIFIER_EMBEDDINGS * 96]; + time("classifier (16 embeddings)", steps_per_s, &mut || { + run_single(&mut classifier, vec![1, CLASSIFIER_EMBEDDINGS as i64, 96], cls_in.clone()) + .unwrap(); + }); + let frame = vec![0.01f32; VAD_FRAME + VAD_CONTEXT]; + time("silero VAD (576 samples)", vad_per_s, &mut || { + vad.run_frame(&frame).unwrap(); + }); + } + /// Minimal PCM-16 WAV reader: locate the `data` subchunk, decode i16 LE → f32. fn parse_wav_i16(bytes: &[u8]) -> Vec { let pos = bytes diff --git a/src-tauri/src/wake/mod.rs b/src-tauri/src/wake/mod.rs index c1a6759d..4b1623ac 100644 --- a/src-tauri/src/wake/mod.rs +++ b/src-tauri/src/wake/mod.rs @@ -8,16 +8,22 @@ //! Everything is on-device — no network, no cloud, the audio never leaves the Mac. //! //! Optional and OFF by default (like the voice agent it serves): with the toggle -//! off nothing captures the microphone and this module costs nothing. The status -//! is HONEST — `running`/`error` are the real post-apply state of the detector, -//! not a switch that lies (mirroring the voice-bridge honest-toggle rule). +//! off nothing captures the microphone and this module costs nothing. It also +//! stays off without an OpenAI key (`require_voice_key`): the wake word only ever +//! opens the voice agent, so without one every detection would be thrown away. The +//! status is HONEST — `running`/`error` are the real post-apply state of the +//! detector, not a switch that lies (mirroring the voice-bridge honest-toggle rule). +//! +//! Cost while on: the full neural pipeline runs on every 80 ms step, speech or +//! silence — there is no silence-based battery guard (the VAD only vetoes fires). +//! What is kept cheap is around it: the worker wakes once per step (the capture +//! batches), runs at UTILITY QoS (efficiency cores), and sleeps when no audio comes. mod capture; mod debug; mod engine; use std::path::PathBuf; -use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::mpsc; use std::sync::{Arc, Mutex}; use std::thread::JoinHandle; @@ -49,6 +55,11 @@ pub struct WakeConfig { /// data dir) so this module stays free of Tauri types, exactly like /// `on_detect`. `None` while debug capture is off. pub debug_dir: Option, + /// Why the detector must stay OFF even though the user enabled it — today only + /// "no OpenAI key" (see `require_voice_key`). The choice itself is kept: the + /// status reads enabled-but-not-running with this as the reason, and the + /// detector comes back by itself once the reason is gone. + pub paused: Option, } impl Default for WakeConfig { @@ -59,10 +70,28 @@ impl Default for WakeConfig { sensitivity: 0.5, debug_capture: false, debug_dir: None, + paused: None, } } } +/// The reason a detector enabled without an OpenAI key stays off. +pub const PAUSED_NO_VOICE_KEY: &str = + "paused — no OpenAI key. The wake word only opens the voice agent, so the microphone \ + stays off until a key is added."; + +/// Hold an enabled detector off while no OpenAI key is stored. The wake word only +/// ever opens the voice agent, and without a key the front drops every detection +/// (`shouldFireWake`) — so an always-on microphone would buy nothing but battery +/// drain and a lit mic indicator. `key_configured` comes from the caller (reading +/// the Keychain spawns `/usr/bin/security`; this module stays free of it). +pub fn require_voice_key(mut config: WakeConfig, key_configured: bool) -> WakeConfig { + if config.enabled && !key_configured { + config.paused = Some(PAUSED_NO_VOICE_KEY.to_string()); + } + config +} + /// One selectable wake phrase for the Settings picker. #[derive(Debug, Clone, Serialize, Deserialize, specta::Type)] pub struct WakePhrase { @@ -123,7 +152,12 @@ struct Inner { /// Shared with the worker thread, which is where captures are actually /// written — `status()` reads whatever the last attempt left here. debug_error: Arc>>, - stop: Option>, + /// Why a detector that started has since died (the microphone went away) — + /// written by the worker, read by `status()`, so the status stops claiming a + /// dead detector runs. + fault: Arc>>, + /// The controller's line to the worker: a `Stop` here ends it. + stop: Option>, handle: Option>, } @@ -152,6 +186,7 @@ impl WakeController { running: false, error: None, debug_error: Arc::new(Mutex::new(None)), + fault: Arc::new(Mutex::new(None)), stop: None, handle: None, }), @@ -191,6 +226,7 @@ impl WakeController { }, }; let debug_error = Arc::new(Mutex::new(debug_reason)); + let fault = Arc::new(Mutex::new(None)); { let mut inner = self.inner.lock().unwrap(); inner.phrase = phrase.clone(); @@ -198,18 +234,23 @@ impl WakeController { inner.enabled = config.enabled; inner.debug_capture = debug_capture; inner.debug_dir = config.debug_dir.clone(); - inner.error = None; + // Paused = enabled, deliberately not running, and saying why. + inner.error = config.paused.clone().filter(|_| config.enabled); inner.debug_error = debug_error.clone(); + inner.fault = fault.clone(); inner.running = false; } - if !config.enabled { + if !config.enabled || config.paused.is_some() { return self.status(); } let on_detect = self.on_detect.lock().unwrap().clone(); - let stop = Arc::new(AtomicBool::new(false)); + // ONE channel carries the audio (from the capture), a fault (from the + // capture) and the stop request (from here), so the worker can simply + // block on it — see `run_worker`. + let (tx, inbox) = mpsc::channel::(); let (report_tx, report_rx) = mpsc::channel::>(); - let stop_worker = stop.clone(); + let link = WorkerLink { capture_tx: tx.clone(), inbox, fault, report: report_tx }; let phrase_worker = phrase.clone(); let debug = DebugSink { enabled: debug_capture, @@ -218,14 +259,12 @@ impl WakeController { }; let handle = std::thread::Builder::new() .name("wake-detector".into()) - .spawn(move || { - run_worker(phrase_worker, sensitivity, on_detect, debug, stop_worker, report_tx) - }) + .spawn(move || run_worker(phrase_worker, sensitivity, on_detect, debug, link)) .ok(); { let mut inner = self.inner.lock().unwrap(); - inner.stop = Some(stop); + inner.stop = Some(tx); inner.handle = handle; } @@ -257,7 +296,9 @@ impl WakeController { (inner.stop.take(), inner.handle.take()) }; if let Some(stop) = stop { - stop.store(true, Ordering::SeqCst); + // A worker that already exited (failed start, dead mic) has dropped its + // end; nothing to tell it then. + let _ = stop.send(WorkerMsg::Stop); } if let Some(handle) = handle { let _ = handle.join(); @@ -267,15 +308,18 @@ impl WakeController { pub fn status(&self) -> WakeStatus { let inner = self.inner.lock().unwrap(); - // Read the shared capture error into a local FIRST: its guard is a - // temporary that would otherwise outlive `inner` at the end of the block. + // Read the shared errors into locals FIRST: their guards are temporaries + // that would otherwise outlive `inner` at the end of the block. let debug_error = inner.debug_error.lock().unwrap().clone(); + let fault = inner.fault.lock().unwrap().clone(); WakeStatus { enabled: inner.enabled, phrase: inner.phrase.clone(), sensitivity: inner.sensitivity, - running: inner.running, - error: inner.error.clone(), + // A detector whose microphone went away is not running, whatever it + // was when it started. + running: inner.running && fault.is_none(), + error: fault.or_else(|| inner.error.clone()), phrases: phrase_catalogue(), debug_capture: inner.debug_capture, debug_dir: inner.debug_dir.as_ref().map(|p| p.display().to_string()), @@ -314,6 +358,47 @@ impl DebugSink { } } +/// What reaches the worker thread, all on one channel so the worker can BLOCK on +/// it: it wakes for audio (once per detection step), a dead stream or a stop +/// request, and otherwise sleeps — no timeout, no polling, even when the +/// microphone goes quiet. +enum WorkerMsg { + /// A batch of 16 kHz mono audio — about one detection step (`capture.rs`). + Audio(Vec), + /// The capture died (the device went away): stop, and record why. + Fault(String), + /// The controller wants the worker gone (disable, reconfigure, quit). + Stop, +} + +/// The worker's end of the plumbing, kept together so `run_worker`'s signature +/// stays readable. +struct WorkerLink { + /// Handed to the capture, which sends audio and faults on it. + capture_tx: mpsc::Sender, + inbox: mpsc::Receiver, + /// Where a mid-run failure is recorded for `status()`. + fault: Arc>>, + /// The start-up outcome, which `apply` waits on. + report: mpsc::Sender>, +} + +/// Move the calling thread to the UTILITY QoS class, which macOS schedules on the +/// efficiency cores. Every ONNX session runs one intra-op thread, so the inference +/// happens ON this thread and inherits the class. A step's work is ~1.3 ms on a +/// performance core (measured), against an 80 ms budget — room to spare for the +/// slower efficiency cores. +#[cfg(target_os = "macos")] +fn lower_thread_qos() { + // SAFETY: only changes the calling thread's own QoS class; no pointers. + let rc = unsafe { + libc::pthread_set_qos_class_self_np(libc::qos_class_t::QOS_CLASS_UTILITY, 0) + }; + if rc != 0 { + eprintln!("[wake] could not lower the detector's QoS class (error {rc}) — it runs at the default class"); + } +} + /// The detector worker: build the engine + open the mic, report the outcome, then /// pump audio through the engine until asked to stop. Owns the `!Send` capture /// stream and the `&mut`-driven ONNX sessions, so neither ever crosses a thread. @@ -322,13 +407,15 @@ fn run_worker( sensitivity: f32, on_detect: Option>, debug: DebugSink, - stop: Arc, - report: mpsc::Sender>, + link: WorkerLink, ) { - let (tx, rx) = mpsc::channel::>(); + #[cfg(target_os = "macos")] + lower_thread_qos(); + + let WorkerLink { capture_tx, inbox, fault, report } = link; let setup = (|| { let engine = Engine::new(&phrase, sensitivity, debug.enabled)?; - let capture = Capture::start(tx)?; + let capture = Capture::start(capture_tx)?; Ok::<_, String>((engine, capture)) })(); let (mut engine, _capture) = match setup { @@ -344,37 +431,41 @@ fn run_worker( } }; - while !stop.load(Ordering::SeqCst) { - match rx.recv_timeout(Duration::from_millis(100)) { - Ok(chunk) => { - if let Some(detection) = engine.feed(&chunk) { - match detection.suppressed_by { - None => eprintln!( - "[wake] DETECTED phrase={} score={:.3} → firing event", - engine.phrase(), - detection.score - ), - // Only reachable while debug capture is on, and only ever - // dumped — a suppressed candidate must never reach the app. - Some(gate) => eprintln!( - "[wake] suppressed phrase={} score={:.3} by {gate}", - engine.phrase(), - detection.score - ), - } - // Dump BEFORE firing: the callback hops into the webview and - // opens a microphone, and the evidence for a false positive is - // worth more than a few ms of trigger latency. - debug.record(engine.phrase(), sensitivity, &detection); - if detection.fired() { - if let Some(cb) = &on_detect { - cb(engine.phrase(), detection.score); - } - } - } + loop { + // Blocks until something arrives: audio (~12×/s), a fault, or a stop. + let chunk = match inbox.recv() { + Ok(WorkerMsg::Audio(chunk)) => chunk, + Ok(WorkerMsg::Fault(reason)) => { + eprintln!("[wake] detector STOPPED: {reason}"); + *fault.lock().unwrap() = Some(reason); + break; + } + // `Err` = the controller's sender is gone too: nobody left to serve. + Ok(WorkerMsg::Stop) | Err(_) => break, + }; + let Some(detection) = engine.feed(&chunk) else { continue }; + match detection.suppressed_by { + None => eprintln!( + "[wake] DETECTED phrase={} score={:.3} → firing event", + engine.phrase(), + detection.score + ), + // Only reachable while debug capture is on, and only ever dumped — a + // suppressed candidate must never reach the app. + Some(gate) => eprintln!( + "[wake] suppressed phrase={} score={:.3} by {gate}", + engine.phrase(), + detection.score + ), + } + // Dump BEFORE firing: the callback hops into the webview and opens a + // microphone, and the evidence for a false positive is worth more than a + // few ms of trigger latency. + debug.record(engine.phrase(), sensitivity, &detection); + if detection.fired() { + if let Some(cb) = &on_detect { + cb(engine.phrase(), detection.score); } - Err(mpsc::RecvTimeoutError::Timeout) => continue, - Err(mpsc::RecvTimeoutError::Disconnected) => break, // capture died } } // `_capture` drops here → mic released (orange indicator off). @@ -394,6 +485,40 @@ mod tests { assert_eq!(s, 0.0); } + /// Without an OpenAI key the wake word is useless (the front drops every + /// detection), so an enabled detector must not hold the microphone — but it + /// must not forget the user's choice either, and must say why it is off. + #[test] + fn an_enabled_detector_without_a_voice_key_is_paused_not_started() { + let enabled = WakeConfig { enabled: true, ..WakeConfig::default() }; + assert!(require_voice_key(enabled.clone(), true).paused.is_none(), "a key: runs"); + let disabled = WakeConfig::default(); + assert!(require_voice_key(disabled, false).paused.is_none(), "off stays plain off"); + + let ctrl = WakeController::new(); + let st = ctrl.apply(require_voice_key(enabled, false)); + assert!(st.enabled, "the user's choice is kept"); + assert!(!st.running, "no worker, no microphone"); + assert_eq!(st.error.as_deref(), Some(PAUSED_NO_VOICE_KEY), "and the reason is legible"); + assert!(ctrl.inner.lock().unwrap().handle.is_none(), "no thread was spawned"); + } + + /// A detector that started and then lost its microphone must stop claiming to + /// run — the fault the worker records wins over the start-up state. + #[test] + fn a_fault_after_start_reads_as_not_running_with_the_reason() { + let ctrl = WakeController::new(); + { + let mut inner = ctrl.inner.lock().unwrap(); + inner.enabled = true; + inner.running = true; + *inner.fault.lock().unwrap() = Some("the microphone went away".into()); + } + let st = ctrl.status(); + assert!(!st.running); + assert_eq!(st.error.as_deref(), Some("the microphone went away")); + } + #[test] fn disabled_apply_reports_not_running_without_error() { let ctrl = WakeController::new(); diff --git a/src-tauri/tauri.conf.json b/src-tauri/tauri.conf.json index 0d69d8a4..fb170ca1 100644 --- a/src-tauri/tauri.conf.json +++ b/src-tauri/tauri.conf.json @@ -1,7 +1,7 @@ { "$schema": "https://schema.tauri.app/config/2", "productName": "Flight Deck", - "version": "2.6.0", + "version": "2.7.0", "identifier": "com.tosse.desktop", "build": { "beforeDevCommand": "pnpm dev", diff --git a/src/App.tsx b/src/App.tsx index fe153043..41f7e94f 100644 --- a/src/App.tsx +++ b/src/App.tsx @@ -10,8 +10,6 @@ import { EditorToggle } from "./features/editor/EditorToggle"; import { FlightDeck } from "./features/flightdeck/FlightDeck"; import { FlightDeckReplyModal } from "./features/flightdeck/FlightDeckReplyModal"; import { useFlightdeckModal } from "./features/flightdeck/flightdeckModalStore"; -import { SoundToggle } from "./features/notifications/SoundToggle"; -import { CaffeinateToggle } from "./features/power/CaffeinateToggle"; import { CaffeinateHost } from "./features/power/CaffeinateHost"; import { MachineHealthHost } from "./features/machines/MachineHealthHost"; import { AutoAccountSwitchHost } from "./features/settings/AutoAccountSwitchHost"; @@ -44,7 +42,6 @@ import { useThreadJump } from "./store/threadJump"; import { useGlobalSessionEvents } from "./ipc/useGlobalSessionEvents"; import { AppControlHost } from "./agent/AppControlHost"; import { VoiceHost } from "./voice/VoiceHost"; -import { VoiceMicToggle, VoiceModeToggle } from "./voice/VoiceToggle"; import { startUpdaterAutoCheck } from "./store/updater"; import { startClaudeCliAutoCheck } from "./store/claudeCliUpdate"; import { initNotifications } from "./notifications/notify"; @@ -59,7 +56,9 @@ import { import { useDisplay } from "./store/display"; import { useNotifications } from "./store/notifications"; import { useSettingsUi } from "./store/settingsUi"; -import { NavBtn, TosseCrmMark, Win } from "./ui/kit"; +import { NavBtn, Win } from "./ui/kit"; +import { SystemTray } from "./ui/SystemTray"; +import { useWidgetOn } from "./store/sidePanelWidgetsStore"; import { runAppAction } from "./ui/appActions"; import { ACTION_BINDINGS, @@ -107,6 +106,16 @@ export default function App() { // With the conversation side panel on, the header carries ACTIONS only: the task chip, // the worktree indicator and the stream control move into the panel. const sidePanel = useDisplay((s) => s.conversationSidePanel); + // …each only while its panel widget is shown: a widget switched off in the panel gives its + // header home back, so hiding it never makes the task, the worktree or the stream unreachable. + // (Hooks first, then the `&&`: a short-circuit would make the hook call conditional.) + const taskWidget = useWidgetOn("task"); + const worktreeWidget = useWidgetOn("worktree"); + const streamWidget = useWidgetOn("stream"); + const taskInPanel = sidePanel && taskWidget; + const worktreeInPanel = sidePanel && worktreeWidget; + const streamInPanel = sidePanel && streamWidget; + const titleBarMotion = useDisplay((s) => s.titleBarAnimations); // The IDE tab is conditional too, on its display preference alone (Settings → General → // Display). Off → no tab, no "Open in IDE" entry points, and the view is never mounted. @@ -271,6 +280,7 @@ export default function App() { return ( changeView("conversation")} /> } label="TOSSE" on={view === "tosse"} title="TOSSE tasks (⌘3)" @@ -312,7 +319,6 @@ export default function App() { {/* Last, so it keeps ⌘4 whether or not the conditional TOSSE tab is showing. */} {ideAvailable ? ( - {/* Always visible (both views): mute/unmute the notification chime on the - spot, without opening Settings. Also bound to ⌘⇧M. */} - {/* Voice agent: arm/disarm the session, then open/close the mic within - it. Both render nothing until an OpenAI key is configured. */} - - - - {/* Always visible (both views): arm/disarm Caffeinate (keep the Mac awake). */} - + {/* Always visible (every view): the app-wide switches — voice session + mic, + notification sound (also ⌘⇧M), Caffeinate — folded into one tray. */} + {view === "conversation" && activeRepo ? ( <> {/* App-wide toggles | this conversation's actions. */} {active ?