Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion cmux-tui/crates/acpmux/docs/acpmux-schema.json
Original file line number Diff line number Diff line change
Expand Up @@ -123,7 +123,7 @@
"SessionDefaults": {"model": "string?", "effort": "string?", "policy": "string?", "prefer": "string[]? (profiles to use for a family, in order; first installed wins)", "env": "{key: value}? (${cwd}, ${home}, ${model}, ~/ expand)", "note": "keyed by family or profile"},
"HarnessProfile": {"kind": "acp|claude-stdio", "argv": "string[] (${model} = the requested model at spawn)", "env": "{key: value}", "family": "string? (derived from argv when absent)", "models": "(string | {id, name})[] declared catalog, shown ahead of the reported one", "model": "string?", "effort": "string?", "policy": "string?", "fallback": "string? (profile to move to on limit or auth errors)", "description": "string?"},
"SessionSummary": {"sessionId": "string", "name": "string", "harness": "string", "family": "string|null", "cwd": "string", "status": "idle|ready|running|waiting|disconnected|closed", "peer": "string? (host name for remote sessions)", "stateSeq": "number (bumps on status, permission and turn changes)", "unread": "boolean (a turn ended while no attached client showed the chat; cleared when one shows it)", "shown": "boolean (some attached client shows the chat: _acpmux/attach visible, _acpmux/view)", "indicator": "input|working|failed|lost|done|null (the status dot every client draws; first match wins: input = a permission or question waits, working = a turn runs, failed = the last turn failed, lost = the agent process is gone (disconnected or unreachable) and the last turn did not answer, done = the last turn answered (seen is client-side), null = nothing to show)", "lastTurn": "{turnId, promptId, status, stopReason?, errorText?, errorSource?, detail?, endedAt} | null", "lane": "{state: working|input|idle|failed|lost, since: number|null, error: string|null, step: string|null, title: string|null, stalledSince: number|null} | null (a coordinator's view of the session; null when closed: working since its turn started (stalledSince: since when a working turn has had no tool call in flight and no sign from its model for stalledTurnMs, default 10 minutes, 0 off), input since it began waiting, idle since its last turn ended, failed with that turn's errorText, lost when the agent process or peer is gone; step is the running turn's prompt, else the agent's last reply line, one line of at most 240 characters; title is the session's title, else its latest prompt, cut the same way; error, step and title have secrets masked, e.g. [GitHub Personal Access Token], and error is cut to 3072 characters; the summary's other text (title, lastPrompt, preview, lastMessage.text, the turn's and queued prompts, lastTurn.errorText) is raw for every client except a proven peer daemon (peer token), which reads it masked and cut to 16384 characters and may not rewrite a queued prompt that holds a secret)", "inputSince": "number|null (when the session began waiting on the person)", "attached": "number (clients attached now)", "pendingPermissions": "number", "queued": "number", "turnCount": "number", "lastSeq": "number", "lastPrompt": "string|null", "preview": "string|null", "model": "string|null", "currentModeId": "string|null", "policy": "string|null", "rules": "boolean", "tools": "ToolRow[] (the chat's tools: cmux's own, then what the harness reported)", "toolsMutable": "boolean (_acpmux/set_tools can turn its MCP servers off: Claude Code only)", "tags": "{key: value}", "createdAt": "number", "updatedAt": "number", "queue": "[{promptId, turnId, client, prompt, queuedAt, held, images: [{id, mimeType, name: string|null}]}] prompts waiting for the running turn (images: what each sends, in order; _acpmux/attachment reads each by id before it runs)", "turn": "{startedAt, client, prompt, turnId, promptId, turnSeq, stalledSince} | null (the running turn; stalledSince as lane.stalledSince)", "enforcement": "Enforcement", "gitBranch": "string|null (the branch the session's cwd has checked out now; null on a detached HEAD or outside a repository. Live while a client holds the chat (a git record on each change); otherwise the last branch this daemon read, else the one the session started on)"},
"SessionDetail": "SessionSummary plus modes, configOptions, models, agentInfo, agentCapabilities, usage, contextBreakdown: {categories: [{id?, name, tokens, kind: used|free|buffer|deferred}], memoryFiles: [{path, type, tokens}], mcpTools: [{name, serverName, tokens}], skills?: [{name, tokens}], totalTokens, maxTokens}|null (Claude Code's /context after the chat's last turn, also sent then as session_info_update _meta.acpmux.contextBreakdown; each list at most 50 rows, the largest; null for any other harness, after a failover, or until Claude Code first answers get_context_usage; a turn whose get_context_usage Claude Code refuses keeps the last breakdown), pending: [{permissionId, request, groupId, turnId}] (groupId and turnId are string|null), rulesJson, gitHead: string|null (the cwd's full HEAD commit when the session started or forked; null outside a repository; gitBranch is SessionSummary's, the branch checked out now), totals: TurnUsage|null (the sum of this session's own turns; null until a harness reports any; a fork starts its own)",
"SessionDetail": "SessionSummary plus modes, configOptions, models, agentInfo, agentCapabilities, usage, contextBreakdown: {categories: [{id?, name, tokens, kind: used|free|buffer|deferred}], memoryFiles: [{path, type, tokens}], mcpTools: [{name, serverName, tokens}], skills?: [{name, tokens}], totalTokens, maxTokens}|null (Claude Code's /context after the chat's last turn, also sent then as session_info_update _meta.acpmux.contextBreakdown; each list at most 50 rows, the largest; null for any other harness, after a failover, or until Claude Code first answers get_context_usage; a turn whose get_context_usage Claude Code refuses keeps the last breakdown), usageLimits: {plan?: string, windows: [{id: session|weekly|weekly_opus|weekly_sonnet, usedPercent: number 0-100, resetsAt?: unix seconds}]}|null (the account's plan usage limits, from Claude Code's rate_limit_event (five_hour is session, seven_day weekly, seven_day_opus weekly_opus, seven_day_sonnet weekly_sonnet; overage windows are left out), each window its newest value, a window past its cap 100; also sent as session_info_update _meta.acpmux.usageLimits with every window that Claude Code process reported; plan is not set yet, as Claude Code's stream names none; null for any other harness, before the first event, or after a harness failover), pending: [{permissionId, request, groupId, turnId}] (groupId and turnId are string|null), rulesJson, gitHead: string|null (the cwd's full HEAD commit when the session started or forked; null outside a repository; gitBranch is SessionSummary's, the branch checked out now), totals: TurnUsage|null (the sum of this session's own turns; null until a harness reports any; a fork starts its own)",
"TurnUsage": {"costUsd": "number? (USD, the harness's estimate)", "inputTokens": "number?", "outputTokens": "number?", "cacheReadTokens": "number?", "cacheWriteTokens": "number?", "note": "Only what the harness reported: Claude Code reports all of them (its tokens cover the main loop, its cost includes subagents); other harnesses report none, so the fields are absent, never 0. A turn's costUsd is the rise in Claude Code's running total, and is absent when that cannot be told: a failed turn, a total that went down, the first turn of a fork through an earlier turn or of a session older than this field."},
"Process": {"id": "string (<sessionId>/<harness task id>: Claude's task id, Codex's processId)", "sessionId": "string", "subagentSessionId": "string|null (the subagent whose call started it)", "toolCallId": "string (Codex: the commandExecution item id)", "taskId": "string (Codex: the processId)", "harness": "string (the session's harness family: claude, codex)", "command": "string", "cwd": "string", "startedAt": "number (unix ms)", "endedAt": "number|null", "state": "running|exited|failed|killed|lost (lost: it ended with its harness, unseen)", "exitCode": "number|null (Codex reports it when the process exits on its own; Claude Code does not)", "signal": "null", "reason": "person|agent|harness_exit|null (why it stopped)", "detail": "string|null (the harness's own words on its end)", "origin": "background|auto_backgrounded|user_backgrounded (user_backgrounded: moved there by _acpmux/process/background, Ctrl-B)", "outputBytes": "number|null (its output file's length when listed; null while the harness named no file, and for Codex)", "outputFile": "string? (the file the harness writes its output to; read it with _acpmux/process/tail)"},
"process_update": {"process": "Process", "note": "Logged when a process starts or ends; a daemon rebuilds the registry from these and the harness updates."},
Expand Down
19 changes: 13 additions & 6 deletions cmux-tui/crates/acpmux/src/agent.rs
Original file line number Diff line number Diff line change
Expand Up @@ -195,6 +195,14 @@ struct CodexLink {
order: Mutex<()>,
}

/// A session update whose news is only the agent process's own to give: Claude Code's /context
/// breakdown, its account's usage limits. One a replaced process sends is not the chat's.
pub fn process_state(params: &Value) -> bool {
params
.pointer("/update/_meta/acpmux")
.is_some_and(|m| m.get("contextBreakdown").is_some() || m.get("usageLimits").is_some())
}

impl ChildAgent {
/// Ends every process in the agent's job (Windows; killpg on Unix).
#[cfg(windows)]
Expand Down Expand Up @@ -517,8 +525,8 @@ impl ChildAgent {
None
}

/// Mark the agent replaced (a failover) or stopping: a /context breakdown it still sends,
/// answering a turn that ended before, is dropped (`tap_live`). Waits for a breakdown being
/// Mark the agent replaced (a failover) or stopping: a /context breakdown or usage limits it
/// still sends, answering a turn that ended before, are dropped (`tap_live`). Waits for a breakdown being
/// logged right now, so what the caller logs next comes after it.
pub fn retire(&self) {
*self.retired.lock().unwrap_or_else(|e| e.into_inner()) = true;
Expand All @@ -529,14 +537,13 @@ impl ChildAgent {
*self.retired.lock().unwrap_or_else(|e| e.into_inner())
}

/// Runs `log` for `m`, unless `m` is a /context breakdown update from a retired agent (None).
/// Runs `log` for `m`, unless `m` is a /context breakdown or usage limits update
/// (`process_state`) from a retired agent (None).
/// For such an update the retire lock is held across the check and `log`.
fn tap_live<R>(&self, m: &Message, log: impl FnOnce() -> R) -> Option<R> {
let context = matches!(m, Message::Notification { method, params }
if method == crate::rpc::method::SESSION_UPDATE
&& params
.as_ref()
.is_some_and(|p| p.pointer("/update/_meta/acpmux/contextBreakdown").is_some()));
&& params.as_ref().is_some_and(process_state));
if !context {
return Some(log());
}
Expand Down
2 changes: 2 additions & 0 deletions cmux-tui/crates/acpmux/src/claude_stdio/inbound.rs
Original file line number Diff line number Diff line change
Expand Up @@ -729,6 +729,8 @@ impl Translator {
_ => {}
}
}
// The account's plan usage limits, newest per window (`usage_limits.rs`).
"rate_limit_event" => out.extend(self.usage_limits_update(line).await),
"system" if sub.starts_with("task_") => {
if let Some(update) = self.background_shell_task(sub, line, &sid).await {
out.push(update);
Expand Down
8 changes: 8 additions & 0 deletions cmux-tui/crates/acpmux/src/claude_stdio/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -244,9 +244,13 @@ mod subagent_route;
mod subagent_tests;
#[cfg(test)]
mod tests;
mod usage_limits;
#[cfg(test)]
mod usage_limits_tests;
#[cfg(test)]
mod workflow_tests;
mod workflows;
pub use usage_limits::merge_usage_limits;

/// The internal ACP request that stops one of Claude Code's background
/// tasks (`hub/processes.rs`), `{sessionId, taskId}`: Claude Code's
Expand Down Expand Up @@ -468,6 +472,9 @@ pub struct Translator {
/// and those in the background (`run_in_background`, or moved there) until
/// their `task_notification` or Claude's exit ends them.
background_shells: Mutex<HashMap<String, BackgroundShell>>,
/// The plan usage limit windows Claude Code reported, newest each, by id
/// (`usage_limits.rs`).
usage_limits: Mutex<std::collections::BTreeMap<&'static str, Value>>,
}

/// A running shell: the session its Bash call ran in, its command, Claude's
Expand Down Expand Up @@ -555,6 +562,7 @@ impl Translator {
streamed_subagents: Mutex::new(HashSet::new()),
edits: Mutex::new(HashMap::new()),
announced: Mutex::new(HashMap::new()),
usage_limits: Mutex::new(Default::default()),
})
}

Expand Down
105 changes: 105 additions & 0 deletions cmux-tui/crates/acpmux/src/claude_stdio/usage_limits.rs
Original file line number Diff line number Diff line change
@@ -0,0 +1,105 @@
//! The account's plan usage limits, from Claude Code's `rate_limit_event`
//! lines: how much of each limit window (the 5-hour session, the week) is
//! used and when it resets, for the pane's Context card. Sent as a
//! `session_info_update` with `_meta.acpmux.usageLimits`, and kept as the
//! session detail's `usageLimits` (`hub/views.rs`).
//!
//! Claude Code 2.1.307 writes `{"type": "rate_limit_event", "rate_limit_info":
//! {status, rateLimitType?, utilization?, resetsAt?, unifiedWindows?: {five_hour?,
//! seven_day?, seven_day_overage_included?: {utilization, resetsAt}}, ...}}`
//! when a window's rounded percentage or reset time moves. `utilization` is the
//! fraction of the window used (0-1; above 1 once usage runs past the cap) and
//! `resetsAt` unix seconds. The stream names no plan, so `plan` is never set.

use super::*;
use std::collections::BTreeMap;

/// Claude Code's limit windows and the ids acpmux names them by. The overage
/// windows (`overage`, `seven_day_overage_included`) are extra paid usage, not
/// a plan limit, and are left out.
const WINDOWS: &[(&str, &str)] = &[
("five_hour", "session"),
("seven_day", "weekly"),
("seven_day_opus", "weekly_opus"),
("seven_day_sonnet", "weekly_sonnet"),
];

fn window_id(claude: &str) -> Option<&'static str> {
WINDOWS.iter().find(|(name, _)| *name == claude).map(|(_, id)| *id)
}

/// One window as acpmux sends it: `usedPercent` (0-100, two decimals, a window
/// past its cap reads 100) and `resetsAt` when Claude Code gave one. None
/// without a usable utilization.
fn window(id: &str, utilization: Option<&Value>, resets_at: Option<&Value>) -> Option<Value> {
let used = utilization?.as_f64().filter(|u| u.is_finite())?;
let percent = ((used * 100.0).clamp(0.0, 100.0) * 100.0).round() / 100.0;
let mut w = json!({"id": id, "usedPercent": percent});
if let Some(at) = resets_at.and_then(Value::as_i64).filter(|at| *at > 0) {
w["resetsAt"] = json!(at);
}
Some(w)
}

/// The windows one `rate_limit_info` reports, by id: the limiting window it
/// names (`rateLimitType` with `utilization`), then each of `unifiedWindows`,
/// which wins for a window both name. Empty for anything malformed.
pub(super) fn usage_windows(info: &Value) -> BTreeMap<&'static str, Value> {
let mut found = BTreeMap::new();
if let Some(id) = info.get("rateLimitType").and_then(Value::as_str).and_then(window_id)
&& let Some(w) = window(id, info.get("utilization"), info.get("resetsAt"))
{
found.insert(id, w);
}
for (name, w) in info.get("unifiedWindows").and_then(Value::as_object).into_iter().flatten() {
if let Some(id) = window_id(name)
&& let Some(w) = window(id, w.get("utilization"), w.get("resetsAt"))
{
found.insert(id, w);
}
}
found
}

/// The saved `usageLimits` with a newer one's windows in it: each window the
/// newer one names replaces the saved one of its id, and the others stay (a
/// restarted Claude Code knows only what it saw since). Its `plan`, if any,
/// wins.
pub fn merge_usage_limits(saved: Option<&Value>, newer: &Value) -> Value {
let mut windows: Vec<Value> =
saved.and_then(|s| s.get("windows")).and_then(Value::as_array).cloned().unwrap_or_default();
for w in newer.get("windows").and_then(Value::as_array).into_iter().flatten() {
let Some(id) = w.get("id").and_then(Value::as_str) else { continue };
match windows.iter_mut().find(|old| old.get("id").and_then(Value::as_str) == Some(id)) {
Some(old) => *old = w.clone(),
None => windows.push(w.clone()),
}
}
let mut merged = json!({"windows": windows});
if let Some(plan) = newer.get("plan").or_else(|| saved?.get("plan")).filter(|p| p.is_string()) {
merged["plan"] = plan.clone();
}
merged
}

impl Translator {
/// A `rate_limit_event` as a `session_info_update` carrying every window
/// this Claude Code reported so far, each its newest value. None when the
/// event names no window acpmux shows. Always the chat's own session: the
/// limits are the account's, whichever stream (a subagent's) reported them.
pub(super) async fn usage_limits_update(&self, line: &Value) -> Option<Message> {
let found = usage_windows(line.get("rate_limit_info")?);
if found.is_empty() {
return None;
}
let mut kept = self.usage_limits.lock().await;
kept.extend(found);
let windows: Vec<Value> = kept.values().cloned().collect();
let update = json!({
"sessionUpdate": "session_info_update",
"_meta": {"acpmux": {"usageLimits": {"windows": windows}}},
});
let params = json!({"sessionId": self.acp_session_id, "update": update});
Some(Message::notification(method::SESSION_UPDATE, params))
}
}
Loading
Loading