iceBear67
a968ad297c
Forward-port the six local patches onto the two new monorepo syncs (SOURCE_REV e6a67a54 -> 84ae1223). Resolved 0001 against the runtime full_version() injection and 0005 against the Option-wrapped UserPromptSubmit hook payload.
2397 lines
96 KiB
Diff
2397 lines
96 KiB
Diff
From 0000000000000000000000000000000000000000 Mon Sep 17 00:00:00 2001
|
|
From: iceBear67 <icebear67@sfclub.cc>
|
|
Date: Sat, 15 Aug 2026 09:09:50 +0000
|
|
Subject: [PATCH] Add /rc remote control bridge to grok-glance
|
|
|
|
Mirror the running session to a grok-glance server over ACP, without
|
|
disturbing the terminal: no restart, no headless switch, no degraded TUI.
|
|
|
|
The TUI is already an ACP client -- it holds an AcpAgentTx to send to the
|
|
agent and an AcpClientRx to receive from it. That pair is the tee point, so
|
|
remote control needs no changes to the agent runtime, the session actor or
|
|
the shell. It is also transport-agnostic: connect() and connect_via_leader()
|
|
yield the same AcpConnection, so one implementation covers both modes.
|
|
|
|
Toward glance the roles invert -- grok is the Agent, glance is the Client --
|
|
which makes glance a stock ACP client rather than something bespoke.
|
|
|
|
Both notification rails are mirrored: the stable acp::SessionUpdate and the
|
|
xAI ExtNotification rail, whose ~60 variants carry the streaming deltas and
|
|
turn boundaries that make a transcript readable. Forwarding only the former
|
|
would give glance a transcript with the middle missing. _meta is preserved
|
|
verbatim so eventId/promptId/chunkId/isReplay still order and dedup the
|
|
stream on the far side.
|
|
|
|
The three interaction requests -- request_permission, ask_user_question and
|
|
exit_plan_mode -- are raced rather than forwarded. This is the same set
|
|
grok's own leader broadcasts for first-answer-wins arbitration; handling
|
|
only permissions would strand a remote user the moment the agent asked a
|
|
question. The race is a channel swap: lift the real response_tx out, hand
|
|
the TUI a substitute, and let whichever side answers first win. The loser's
|
|
dialog closes by itself.
|
|
|
|
Failure is always local-only. A glance that is unreachable, slow or killed
|
|
mid-turn reconnects with backoff and replays from a bounded ring; it can
|
|
never take the local session down with it.
|
|
|
|
Remote actions stay attributable: a remote cancel sets cancelTrigger to
|
|
Client("glance") -- an existing variant that maps to StopGesture, the same
|
|
class as Esc -- rather than faking a keypress and lying in the session log.
|
|
Remote prompts carry clientIdentifier for the same reason.
|
|
|
|
diff --git a/Cargo.lock b/Cargo.lock
|
|
index 8fc3009c68e60a71ed400d7b3a2fb318c7455956..bcd7f081a49128e010d3b3122038002b7e58b9ca 100644
|
|
--- a/Cargo.lock
|
|
+++ b/Cargo.lock
|
|
@@ -13737,6 +13737,7 @@ dependencies = [
|
|
"dunce",
|
|
"enum_delegate",
|
|
"flate2",
|
|
+ "futures-util",
|
|
"git2",
|
|
"image",
|
|
"indexmap",
|
|
@@ -13765,6 +13766,7 @@ dependencies = [
|
|
"tempfile",
|
|
"textwrap",
|
|
"tokio",
|
|
+ "tokio-tungstenite 0.27.0",
|
|
"tokio-util",
|
|
"toml",
|
|
"toml_edit 0.22.27",
|
|
diff --git a/crates/codegen/xai-grok-pager/Cargo.toml b/crates/codegen/xai-grok-pager/Cargo.toml
|
|
index 86d9d26dd190714acd133643dc73ce6e332c1e63..cc56f87f878c18533dc595f40f3ece490d3131d9 100644
|
|
--- a/crates/codegen/xai-grok-pager/Cargo.toml
|
|
+++ b/crates/codegen/xai-grok-pager/Cargo.toml
|
|
@@ -92,6 +92,11 @@ xai-grok-update = { path = "../xai-grok-update" }
|
|
# ACP
|
|
agent-client-protocol = { workspace = true }
|
|
xai-acp-lib = { workspace = true }
|
|
+# `/rc` remote control: outbound WebSocket to a grok-glance server. Same TLS
|
|
+# feature as xai-grok-voice and xai-grok-workspace, which is also what this
|
|
+# crate already pulls in transitively via xai-grok-voice — no new rustls weight.
|
|
+tokio-tungstenite = { workspace = true, features = ["rustls-tls-webpki-roots"] }
|
|
+futures-util = { workspace = true }
|
|
xai-file-utils = { path = "../xai-file-utils" }
|
|
# Crash-recovery registry of open TUI sessions.
|
|
xai-grok-active-sessions = { workspace = true }
|
|
diff --git a/crates/codegen/xai-grok-pager/docs/user-guide/04-slash-commands.md b/crates/codegen/xai-grok-pager/docs/user-guide/04-slash-commands.md
|
|
index 723efc5b91874535777cca954e62547106b0d11f..6149b9fb68a86cbcc78b27d5998f9c1e27f62937 100644
|
|
--- a/crates/codegen/xai-grok-pager/docs/user-guide/04-slash-commands.md
|
|
+++ b/crates/codegen/xai-grok-pager/docs/user-guide/04-slash-commands.md
|
|
@@ -103,6 +103,34 @@ Rename the current session. Alias: `/title`.
|
|
|
|
`--auto` unpins a manual title and lets auto-titling resume. It applies to Build sessions only — chat conversations have no local auto-titler. It must be the only argument (`/rename --auto Something` is an error). A session cannot be named `--auto` via this command; use the dashboard rename editor (`Ctrl+R`) for that pathological case.
|
|
|
|
+### `/rc`
|
|
+
|
|
+Mirror the current session to a [grok-glance](https://github.com/user/grok-glance) server so you can watch and drive it from a browser.
|
|
+
|
|
+```
|
|
+/rc toggle
|
|
+/rc on connect (aliases: start, connect)
|
|
+/rc off disconnect (aliases: stop, disconnect)
|
|
+/rc status report the link
|
|
+```
|
|
+
|
|
+The terminal is not affected. `/rc` does not restart the session, switch it to headless, or hand it to another process — the TUI stays exactly as interactive as it was, and glance becomes a second view of the same session.
|
|
+
|
|
+From either side you can follow the turn stream, send a prompt, interrupt a running turn, and answer tool-permission requests, questions, and plan approvals. **Both sides can answer; whoever answers first wins**, and the other side's dialog closes by itself.
|
|
+
|
|
+Configure the server in `[remote_control]`:
|
|
+
|
|
+```toml
|
|
+[remote_control]
|
|
+url = "wss://glance.example.com/api/acp/agent"
|
|
+api_key = "glance_sk_..."
|
|
+auto_start = false # connect on startup without typing /rc
|
|
+```
|
|
+
|
|
+`GROK_RC_URL` and `GROK_RC_API_KEY` override the file, which is worth using — a bearer token in a plaintext config file is a poor default. Mint the key with `glance apikey add <name>` on the server.
|
|
+
|
|
+The link is scoped to this session and lasts until you quit; nothing is persisted. If glance is unreachable the connection retries with backoff and `/rc status` says why — a failed or dropped link never affects the local session, and `/rc off` clears a failed one.
|
|
+
|
|
---
|
|
|
|
## Model and Mode
|
|
diff --git a/crates/codegen/xai-grok-pager/src/app/acp_handler/mod.rs b/crates/codegen/xai-grok-pager/src/app/acp_handler/mod.rs
|
|
index 87667c5843cca2c4d2f56d41bc67103bd8296e85..5dea13dda409c058a0d4a1841eb882fef6e96fd6 100644
|
|
--- a/crates/codegen/xai-grok-pager/src/app/acp_handler/mod.rs
|
|
+++ b/crates/codegen/xai-grok-pager/src/app/acp_handler/mod.rs
|
|
@@ -153,6 +153,16 @@ use workflow_ingest::*;
|
|
/// background agent must still land in its own scrollback so the user sees
|
|
/// the full turn after switching back.
|
|
pub(crate) fn handle(msg: AcpClientMessage, app: &mut AppView) -> bool {
|
|
+ // `/rc` tee. Every inbound ACP message funnels through here -- the event
|
|
+ // loop calls this both on `recv` and inside its drain batch -- so this one
|
|
+ // line mirrors the whole stream to grok-glance. It returns the message
|
|
+ // unchanged except for interactive reverse-requests, whose reply channel is
|
|
+ // swapped for a race between the terminal and the browser. When remote
|
|
+ // control is off it is a plain move. See [`crate::rc`].
|
|
+ let msg = match app.rc.as_ref() {
|
|
+ Some(rc) => rc.intercept(msg),
|
|
+ None => msg,
|
|
+ };
|
|
match msg {
|
|
AcpClientMessage::SessionNotification(notif) => {
|
|
let mut meta = NotificationMeta::from_json(notif.request.meta.as_ref());
|
|
diff --git a/crates/codegen/xai-grok-pager/src/app/actions.rs b/crates/codegen/xai-grok-pager/src/app/actions.rs
|
|
index ac4b7057e3a24800969cbae01271bfa76fae469c..ce40302b71943058a974c38274dea3903bc13aeb 100644
|
|
--- a/crates/codegen/xai-grok-pager/src/app/actions.rs
|
|
+++ b/crates/codegen/xai-grok-pager/src/app/actions.rs
|
|
@@ -30,6 +30,22 @@ pub enum SwitchModelError {
|
|
/// Any other failure (network, auth, server error, etc.).
|
|
Other(String),
|
|
}
|
|
+/// What `/rc` was asked to do.
|
|
+///
|
|
+/// One `Action` with a sub-enum rather than four variants: turning remote
|
|
+/// control on, off, and explaining it share all their preconditions (a live
|
|
+/// session, a resolvable `[remote_control]` config) and belong in one handler.
|
|
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
|
+pub enum RemoteControlAction {
|
|
+ /// Bare `/rc` — on if off, off if on.
|
|
+ Toggle,
|
|
+ /// `/rc on` — idempotent; reports the current state if already connected.
|
|
+ On,
|
|
+ /// `/rc off` — close the glance link. The session itself is untouched.
|
|
+ Off,
|
|
+ /// `/rc status` — where the bridge is, without changing it.
|
|
+ Status,
|
|
+}
|
|
/// Synchronous, side-effect-free user intent.
|
|
///
|
|
/// Produced by [`super::input`] from key/mouse events.
|
|
@@ -717,6 +733,11 @@ pub enum Action {
|
|
/// tasks as a system block (`/tasks`). The surface minimal mode uses in
|
|
/// place of the `TasksPane`.
|
|
ShowTasks,
|
|
+ /// `/rc` — mirror this session to a grok-glance server so it can also be
|
|
+ /// watched and driven from a browser. Deliberately additive: the terminal
|
|
+ /// session keeps running exactly as it was, and nothing here can restart it,
|
|
+ /// switch it to headless, or hand it over.
|
|
+ RemoteControl(RemoteControlAction),
|
|
/// Show the current plan: preview popover if exists, toast if not.
|
|
ShowPlan,
|
|
/// Enter plan mode. If a description is provided, also start a turn
|
|
diff --git a/crates/codegen/xai-grok-pager/src/app/app_view.rs b/crates/codegen/xai-grok-pager/src/app/app_view.rs
|
|
index abbbf08a7239da006a7bf14a749230ec5398b276..b336028865a9c0c58dd0de4ea1197050c267df49 100644
|
|
--- a/crates/codegen/xai-grok-pager/src/app/app_view.rs
|
|
+++ b/crates/codegen/xai-grok-pager/src/app/app_view.rs
|
|
@@ -1229,6 +1229,18 @@ pub struct AppView {
|
|
/// combinations are unrepresentable; production mutates it only through the
|
|
/// `AppView::voice_*` transition methods.
|
|
pub voice_state: VoiceState,
|
|
+ /// Live `/rc` bridge to grok-glance, `None` while remote control is off.
|
|
+ /// Lives here rather than in the event loop because the tee runs on the ACP
|
|
+ /// receive path, which is reached with `&mut AppView` and nothing else.
|
|
+ pub(crate) rc: Option<crate::rc::RcHandle>,
|
|
+ /// Last state the bridge reported. Separate from [`Self::rc`] so `/rc
|
|
+ /// status` can still explain a bridge that gave up (bad API key) after its
|
|
+ /// task exited and the handle went away.
|
|
+ pub(crate) rc_state: crate::rc::RcState,
|
|
+ /// Parked by `/rc`, consumed at the top of the event loop, which is the only
|
|
+ /// place that can own the bridge's event receiver. Boxed to keep this large
|
|
+ /// struct's size unchanged. See [`crate::rc::RcStart`].
|
|
+ pub(crate) rc_pending_start: Option<Box<crate::rc::RcStart>>,
|
|
}
|
|
/// Reshow window elapsed? None/0 = never. Unparseable ack fails open (show).
|
|
fn privacy_banner_reshow_elapsed(acked_at: &str, reshow_days: Option<u64>) -> bool {
|
|
@@ -1667,6 +1679,9 @@ impl AppView {
|
|
voice_auth: None,
|
|
voice_cmd_tx: None,
|
|
voice_state: VoiceState::Idle,
|
|
+ rc: None,
|
|
+ rc_state: crate::rc::RcState::Off,
|
|
+ rc_pending_start: None,
|
|
}
|
|
}
|
|
/// Seed `deferred_model_switch` from CLI `-m`. The CLI effort token is
|
|
diff --git a/crates/codegen/xai-grok-pager/src/app/app_view_tests.rs b/crates/codegen/xai-grok-pager/src/app/app_view_tests.rs
|
|
index 8de13f2db0bf9dd4f0630bfc8b099ada608c4a38..b456dac46ea39af343f4f0fd84ce7012e67fc97d 100644
|
|
--- a/crates/codegen/xai-grok-pager/src/app/app_view_tests.rs
|
|
+++ b/crates/codegen/xai-grok-pager/src/app/app_view_tests.rs
|
|
@@ -308,6 +308,9 @@ pub(crate) fn test_app() -> AppView {
|
|
voice_auth: None,
|
|
voice_cmd_tx: None,
|
|
voice_state: VoiceState::Idle,
|
|
+ rc: None,
|
|
+ rc_state: crate::rc::RcState::Off,
|
|
+ rc_pending_start: None,
|
|
}
|
|
}
|
|
pub(crate) fn test_app_with_agent() -> AppView {
|
|
diff --git a/crates/codegen/xai-grok-pager/src/app/dispatch/mod.rs b/crates/codegen/xai-grok-pager/src/app/dispatch/mod.rs
|
|
index ac6d52ad658122ffefe10ec7391924c558f78860..d46d5c8b2ff31f94627843f029735e24c1aaa132 100644
|
|
--- a/crates/codegen/xai-grok-pager/src/app/dispatch/mod.rs
|
|
+++ b/crates/codegen/xai-grok-pager/src/app/dispatch/mod.rs
|
|
@@ -27,6 +27,7 @@ mod notes;
|
|
mod permissions;
|
|
mod prompt;
|
|
mod queue;
|
|
+pub(crate) mod rc;
|
|
mod rewind;
|
|
mod router;
|
|
mod session;
|
|
diff --git a/crates/codegen/xai-grok-pager/src/app/dispatch/rc.rs b/crates/codegen/xai-grok-pager/src/app/dispatch/rc.rs
|
|
new file mode 100644
|
|
index 0000000000000000000000000000000000000000..43a3fb0c4694c7da7741b683f5286c71da554d3c
|
|
--- /dev/null
|
|
+++ b/crates/codegen/xai-grok-pager/src/app/dispatch/rc.rs
|
|
@@ -0,0 +1,271 @@
|
|
+//! `/rc` -- remote control: start, stop, and report the grok-glance bridge.
|
|
+//!
|
|
+//! This module is the whole TUI-facing surface of remote control. Everything
|
|
+//! below it ([`crate::rc`]) is protocol and transport and knows nothing about
|
|
+//! views; everything above it is one slash command.
|
|
+//!
|
|
+//! Starting is deliberately two-step -- the handler parks an [`RcStart`] and the
|
|
+//! event loop spawns from it -- because the bridge's event receiver has to live
|
|
+//! as a local of the event loop. See [`RcStart`] for why.
|
|
+
|
|
+use crate::app::actions::{Effect, RemoteControlAction};
|
|
+use crate::app::app_view::{ActiveView, AppView};
|
|
+use crate::rc::{RcEvent, RcStart, RcState, SessionMeta};
|
|
+use crate::scrollback::block::RenderBlock;
|
|
+
|
|
+pub(super) fn dispatch_remote_control(app: &mut AppView, what: RemoteControlAction) -> Vec<Effect> {
|
|
+ match what {
|
|
+ RemoteControlAction::Status => {
|
|
+ let text = status_text(app);
|
|
+ report(app, text)
|
|
+ }
|
|
+ RemoteControlAction::Off => stop(app),
|
|
+ RemoteControlAction::On => start(app),
|
|
+ RemoteControlAction::Toggle => {
|
|
+ if app.rc.is_some() || app.rc_pending_start.is_some() {
|
|
+ stop(app)
|
|
+ } else {
|
|
+ start(app)
|
|
+ }
|
|
+ }
|
|
+ }
|
|
+}
|
|
+
|
|
+fn start(app: &mut AppView) -> Vec<Effect> {
|
|
+ if app.rc.is_some() || app.rc_pending_start.is_some() {
|
|
+ let text = status_text(app);
|
|
+ return report(app, format!("Remote control is already on.\n{text}"));
|
|
+ }
|
|
+
|
|
+ // The bridge mirrors one session; without one there is nothing to mirror and
|
|
+ // glance would show an empty shell.
|
|
+ if app.active_session_id().is_none() {
|
|
+ return report(
|
|
+ app,
|
|
+ "Remote control needs a live session. Start one, then run /rc.".to_owned(),
|
|
+ );
|
|
+ }
|
|
+
|
|
+ let settings = match crate::rc::load_settings() {
|
|
+ Ok(settings) => settings,
|
|
+ Err(why) => return report(app, why),
|
|
+ };
|
|
+
|
|
+ let url = settings.url.clone();
|
|
+ let meta = session_meta(app);
|
|
+ app.rc_pending_start = Some(Box::new(RcStart { settings, meta }));
|
|
+ app.rc_state = RcState::Connecting;
|
|
+ report(
|
|
+ app,
|
|
+ format!(
|
|
+ "Remote control: connecting to {url}\n\
|
|
+ This terminal stays fully interactive -- glance becomes a second view of the \
|
|
+ same session, not a replacement for it. Prompts, interrupts and tool approvals \
|
|
+ work from either side; whichever answers first wins.\n\
|
|
+ /rc off closes the link."
|
|
+ ),
|
|
+ )
|
|
+}
|
|
+
|
|
+fn stop(app: &mut AppView) -> Vec<Effect> {
|
|
+ let was_pending = app.rc_pending_start.take().is_some();
|
|
+ let handle = app.rc.take();
|
|
+ // A bridge that gave up has already dropped its handle, but `rc_state` still
|
|
+ // holds the reason. That is not "already off" -- `/rc off` is how the user
|
|
+ // clears it.
|
|
+ let was_failed = app.rc_state.is_active() && handle.is_none() && !was_pending;
|
|
+ app.rc_state = RcState::Off;
|
|
+
|
|
+ if let Some(handle) = handle {
|
|
+ // Fire-and-forget: the bridge closes its socket and exits, and the event
|
|
+ // loop drops its receiver when the channel closes.
|
|
+ handle.shutdown();
|
|
+ return report(
|
|
+ app,
|
|
+ "Remote control: off. The glance link is closed; this session is unaffected."
|
|
+ .to_owned(),
|
|
+ );
|
|
+ }
|
|
+ if was_pending {
|
|
+ return report(app, "Remote control: cancelled before connecting.".to_owned());
|
|
+ }
|
|
+ if was_failed {
|
|
+ return report(
|
|
+ app,
|
|
+ "Remote control: off. The failed link is cleared; /rc tries again.".to_owned(),
|
|
+ );
|
|
+ }
|
|
+ report(app, "Remote control is already off.".to_owned())
|
|
+}
|
|
+
|
|
+fn status_text(app: &AppView) -> String {
|
|
+ match (&app.rc, &app.rc_state) {
|
|
+ (None, RcState::Off) => {
|
|
+ "Remote control: off. Run /rc to mirror this session to glance.".to_owned()
|
|
+ }
|
|
+ // The bridge gave up (rejected key, unusable URL) and its handle is gone,
|
|
+ // but the reason still matters -- reporting a bare "off" here would hide it.
|
|
+ (None, state) => format!("Remote control: {} (link closed)", state.summary()),
|
|
+ (Some(handle), state) => {
|
|
+ format!("Remote control: {}\nGlance: {}", state.summary(), handle.url())
|
|
+ }
|
|
+ }
|
|
+}
|
|
+
|
|
+/// Session identity for glance's session list, so a browser can label the
|
|
+/// session without asking the agent anything.
|
|
+fn session_meta(app: &AppView) -> SessionMeta {
|
|
+ let session_id = app.active_session_id().map(str::to_owned);
|
|
+ let (title, model) = match app.active_view {
|
|
+ ActiveView::Agent(id) => match app.agents.get(&id) {
|
|
+ Some(agent) => (
|
|
+ agent.generated_session_title.clone(),
|
|
+ agent.session.models.current_model_name(),
|
|
+ ),
|
|
+ None => (None, None),
|
|
+ },
|
|
+ _ => (None, None),
|
|
+ };
|
|
+ SessionMeta {
|
|
+ session_id,
|
|
+ cwd: Some(app.cwd.display().to_string()),
|
|
+ title,
|
|
+ model,
|
|
+ hostname: hostname(),
|
|
+ version: Some(xai_grok_version::VERSION.to_owned()),
|
|
+ }
|
|
+}
|
|
+
|
|
+/// Best-effort host label for the session list. Purely cosmetic -- a user with
|
|
+/// several machines connected needs to tell them apart.
|
|
+fn hostname() -> Option<String> {
|
|
+ std::env::var("HOSTNAME")
|
|
+ .ok()
|
|
+ .or_else(|| std::env::var("COMPUTERNAME").ok())
|
|
+ .map(|h| h.trim().to_owned())
|
|
+ .filter(|h| !h.is_empty())
|
|
+}
|
|
+
|
|
+/// Bridge -> TUI. Returns whether the view needs a redraw.
|
|
+///
|
|
+/// Called from the event loop's lowest-priority arm, mirroring
|
|
+/// [`crate::voice::handle_voice_event`].
|
|
+pub(crate) fn handle_rc_event(app: &mut AppView, event: RcEvent) -> bool {
|
|
+ match event {
|
|
+ RcEvent::State(state) => {
|
|
+ let announce = announcement(&app.rc_state, &state);
|
|
+ app.rc_state = state;
|
|
+ match announce {
|
|
+ Some(text) => {
|
|
+ report(app, text);
|
|
+ true
|
|
+ }
|
|
+ // Reconnect churn is not worth a scrollback line each time; the
|
|
+ // status line already carries it.
|
|
+ None => true,
|
|
+ }
|
|
+ }
|
|
+
|
|
+ // Glance answered a permission / question / plan approval first. Close
|
|
+ // the terminal's modal so the user is not staring at a dialog whose
|
|
+ // request has already been resolved.
|
|
+ RcEvent::InteractionResolvedRemotely { tool_call_id } => {
|
|
+ let mut dismissed = false;
|
|
+ for agent in app.agents.values_mut() {
|
|
+ if agent.dismiss_resolved_interaction(&tool_call_id) {
|
|
+ dismissed = true;
|
|
+ }
|
|
+ }
|
|
+ if dismissed {
|
|
+ tracing::info!(%tool_call_id, "rc: interaction answered from glance");
|
|
+ }
|
|
+ dismissed
|
|
+ }
|
|
+
|
|
+ RcEvent::Notice(text) => {
|
|
+ report(app, format!("Remote control: {text}"));
|
|
+ true
|
|
+ }
|
|
+ }
|
|
+}
|
|
+
|
|
+/// Which state transitions are worth a scrollback line.
|
|
+///
|
|
+/// Connecting -> Connected and anything -> Failed are events the user acts on.
|
|
+/// A reconnect loop is not: it can fire every few seconds during an outage, and
|
|
+/// filling the transcript with it would bury the session's actual output.
|
|
+fn announcement(previous: &RcState, next: &RcState) -> Option<String> {
|
|
+ match (previous, next) {
|
|
+ (_, RcState::Failed { detail }) => Some(format!(
|
|
+ "Remote control failed: {detail}\nThe session is unaffected. Fix the setting and run /rc again."
|
|
+ )),
|
|
+ (RcState::Connected { .. }, RcState::Connected { .. }) => None,
|
|
+ (_, RcState::Connected { .. }) => {
|
|
+ Some("Remote control: connected to glance.".to_owned())
|
|
+ }
|
|
+ (RcState::Reconnecting { .. }, RcState::Reconnecting { .. }) => None,
|
|
+ (_, RcState::Reconnecting { detail, .. }) => Some(format!(
|
|
+ "Remote control: lost the glance link ({detail}); reconnecting in the background."
|
|
+ )),
|
|
+ (_, RcState::Off) => Some("Remote control: off.".to_owned()),
|
|
+ (_, RcState::Connecting) => None,
|
|
+ }
|
|
+}
|
|
+
|
|
+/// Push a system block onto the top-level agent, as `/tasks` and friends do.
|
|
+/// Falls back to a toast on session-less surfaces so a message is never lost.
|
|
+fn report(app: &mut AppView, text: String) -> Vec<Effect> {
|
|
+ if let ActiveView::Agent(id) = app.active_view
|
|
+ && let Some(agent) = app.agents.get_mut(&id)
|
|
+ {
|
|
+ agent.scrollback.push_block(RenderBlock::system(text));
|
|
+ } else {
|
|
+ app.show_toast(&text);
|
|
+ }
|
|
+ vec![]
|
|
+}
|
|
+
|
|
+#[cfg(test)]
|
|
+mod tests {
|
|
+ use super::*;
|
|
+
|
|
+ #[test]
|
|
+ fn reconnect_churn_is_not_announced() {
|
|
+ let first = RcState::Reconnecting {
|
|
+ attempt: 1,
|
|
+ detail: "closed".into(),
|
|
+ };
|
|
+ let second = RcState::Reconnecting {
|
|
+ attempt: 2,
|
|
+ detail: "closed".into(),
|
|
+ };
|
|
+ assert!(announcement(&RcState::Connected { viewers: None }, &first).is_some());
|
|
+ assert!(
|
|
+ announcement(&first, &second).is_none(),
|
|
+ "an outage must not write one scrollback line per retry"
|
|
+ );
|
|
+ }
|
|
+
|
|
+ #[test]
|
|
+ fn viewer_count_changes_are_silent_but_failures_are_not() {
|
|
+ let one = RcState::Connected { viewers: Some(1) };
|
|
+ let two = RcState::Connected { viewers: Some(2) };
|
|
+ assert!(announcement(&one, &two).is_none());
|
|
+
|
|
+ let failed = RcState::Failed {
|
|
+ detail: "HTTP 401".into(),
|
|
+ };
|
|
+ let text = announcement(&one, &failed).expect("a failure must be reported");
|
|
+ assert!(text.contains("HTTP 401"), "the reason has to survive: {text}");
|
|
+ assert!(
|
|
+ text.contains("session is unaffected"),
|
|
+ "the user must be told the local session is fine: {text}"
|
|
+ );
|
|
+ }
|
|
+
|
|
+ #[test]
|
|
+ fn connecting_is_silent_because_the_command_already_said_so() {
|
|
+ assert!(announcement(&RcState::Off, &RcState::Connecting).is_none());
|
|
+ assert!(announcement(&RcState::Connecting, &RcState::Connected { viewers: None }).is_some());
|
|
+ }
|
|
+}
|
|
diff --git a/crates/codegen/xai-grok-pager/src/app/dispatch/router.rs b/crates/codegen/xai-grok-pager/src/app/dispatch/router.rs
|
|
index 86df4292ad50f986a716298ef85c0a73bd9e12ab..1d2e07109f57220660c6737290dd949fa29e5f73 100644
|
|
--- a/crates/codegen/xai-grok-pager/src/app/dispatch/router.rs
|
|
+++ b/crates/codegen/xai-grok-pager/src/app/dispatch/router.rs
|
|
@@ -48,6 +48,7 @@ use super::prompt::{
|
|
};
|
|
use super::queue;
|
|
use super::queue::dispatch_drain_queue;
|
|
+use super::rc::dispatch_remote_control;
|
|
use super::rewind::{
|
|
dispatch_inline_edit_submit, dispatch_rewind, dispatch_rewind_cancel_offer,
|
|
dispatch_rewind_confirm, dispatch_rewind_confirm_never_ask, dispatch_rewind_dismiss,
|
|
@@ -152,6 +153,12 @@ pub(crate) fn dispatch(action: Action, app: &mut AppView) -> Vec<Effect> {
|
|
if let Some(tx) = &app.voice_cmd_tx {
|
|
let _ = tx.try_send(xai_grok_voice::VoiceCommand::Shutdown);
|
|
}
|
|
+ // Close the glance socket on the way out, so a viewer sees the
|
|
+ // session end rather than a link that simply stops responding.
|
|
+ // Fire-and-forget, like voice: quitting must not wait on the network.
|
|
+ if let Some(rc) = &app.rc {
|
|
+ rc.shutdown();
|
|
+ }
|
|
let mut effects = unregister_all_active_sessions(app);
|
|
effects.push(Effect::Quit);
|
|
effects
|
|
@@ -1022,6 +1029,7 @@ pub(crate) fn dispatch(action: Action, app: &mut AppView) -> Vec<Effect> {
|
|
Action::ManageBilling => dispatch_manage_billing(app),
|
|
Action::ShowQueue => dispatch_show_queue(app),
|
|
Action::ShowTasks => dispatch_show_tasks(app),
|
|
+ Action::RemoteControl(what) => dispatch_remote_control(app, what),
|
|
Action::ShowPlan => dispatch_show_plan(app),
|
|
Action::EnterPlanMode { description } => dispatch_enter_plan_mode(app, description),
|
|
Action::SetPlanMode(kind) => set_plan_mode(app, kind),
|
|
diff --git a/crates/codegen/xai-grok-pager/src/app/dispatch/tests/mod.rs b/crates/codegen/xai-grok-pager/src/app/dispatch/tests/mod.rs
|
|
index 63d41e80a915f5ce67ecbeca00669628e568ede8..c41f28667fca9caf278155a8f211139e66b6eee9 100644
|
|
--- a/crates/codegen/xai-grok-pager/src/app/dispatch/tests/mod.rs
|
|
+++ b/crates/codegen/xai-grok-pager/src/app/dispatch/tests/mod.rs
|
|
@@ -296,6 +296,9 @@ fn test_app() -> AppView {
|
|
voice_auth: None,
|
|
voice_cmd_tx: None,
|
|
voice_state: VoiceState::Idle,
|
|
+ rc: None,
|
|
+ rc_state: crate::rc::RcState::Off,
|
|
+ rc_pending_start: None,
|
|
}
|
|
}
|
|
/// Build a default `AgentSession` for
|
|
diff --git a/crates/codegen/xai-grok-pager/src/app/event_loop.rs b/crates/codegen/xai-grok-pager/src/app/event_loop.rs
|
|
index d0f8962eb2d298072d4a232c7ec50b5ddec5ba37..30cca1ed9792f09785d3227d424a170e2cd8d111 100644
|
|
--- a/crates/codegen/xai-grok-pager/src/app/event_loop.rs
|
|
+++ b/crates/codegen/xai-grok-pager/src/app/event_loop.rs
|
|
@@ -1737,6 +1737,12 @@ pub(crate) async fn run(
|
|
let mut voice_rx = None::<tokio::sync::mpsc::Receiver<xai_grok_voice::VoiceEvent>>;
|
|
let voice_auth_factory = connection.auth_manager.clone();
|
|
|
|
+ // `/rc` bridge events. Owned by this loop, for the same reason as
|
|
+ // `voice_rx`: an `AppView`-held receiver would need `&mut app` inside the
|
|
+ // `select!` below, where it would collide with every arm that mutates the
|
|
+ // app. `None` until `/rc` parks an `RcStart` (see `crate::rc::RcStart`).
|
|
+ let mut rc_rx = None::<tokio::sync::mpsc::UnboundedReceiver<crate::rc::RcEvent>>;
|
|
+
|
|
// Animation tick: only scheduled when there are running entries.
|
|
let mut tick_interval = tick_interval;
|
|
let mut animation_tick_at: Option<Instant> = None;
|
|
@@ -2158,6 +2164,19 @@ pub(crate) async fn run(
|
|
presenter.request_presentation(&mut app, terminal, false);
|
|
}
|
|
|
|
+ // Lazy `/rc` bridge, armed by the dispatcher one iteration ago. Spawned
|
|
+ // here rather than in the dispatcher because `rc_rx` must be a local of
|
|
+ // this loop. Dialing happens inside the task, so this never blocks: the
|
|
+ // socket comes up (or does not) as an `RcEvent` later, and the session
|
|
+ // carries on either way.
|
|
+ if let Some(start) = app.rc_pending_start.take() {
|
|
+ let (handle, event_rx) =
|
|
+ crate::rc::start(start.settings, start.meta, app.acp_tx.clone());
|
|
+ tracing::info!(url = handle.url(), "rc: remote control bridge starting");
|
|
+ app.rc = Some(handle);
|
|
+ rc_rx = Some(event_rx);
|
|
+ }
|
|
+
|
|
// Stop voice if the user has left the recording session (see method).
|
|
app.enforce_voice_session_bound();
|
|
|
|
@@ -3053,6 +3072,37 @@ pub(crate) async fn run(
|
|
presenter.request(false);
|
|
}
|
|
|
|
+ // `/rc` bridge events: connection state, notices, and interactions
|
|
+ // that glance answered before the terminal did. Low frequency by
|
|
+ // construction — a connect, a disconnect, the occasional approval —
|
|
+ // so it cannot starve the arms below it. Kept above voice so a modal
|
|
+ // closing because the browser answered is never queued behind a hot
|
|
+ // mic's interim transcripts.
|
|
+ rc_ev = async {
|
|
+ match rc_rx.as_mut() {
|
|
+ Some(rx) => rx.recv().await,
|
|
+ None => std::future::pending().await,
|
|
+ }
|
|
+ } => {
|
|
+ match rc_ev {
|
|
+ Some(rc_ev) => {
|
|
+ if dispatch::rc::handle_rc_event(&mut app, rc_ev) {
|
|
+ presenter.request(false);
|
|
+ }
|
|
+ }
|
|
+ // The bridge task exited: `/rc off`, quit, or it gave up on a
|
|
+ // rejected key. `rc_state` already holds the reason, so drop
|
|
+ // the handle and revert to pending() (no hot-loop on None).
|
|
+ None => {
|
|
+ rc_rx = None;
|
|
+ if app.rc.take().is_some() {
|
|
+ tracing::info!("rc: bridge task ended");
|
|
+ presenter.request(false);
|
|
+ }
|
|
+ }
|
|
+ }
|
|
+ }
|
|
+
|
|
// Voice STT — DELIBERATELY THE LAST (lowest-priority) arm. In a
|
|
// biased select, an arm that is ready on most iterations masks every
|
|
// arm below it. A hot mic (toggle capture stays open across pauses)
|
|
diff --git a/crates/codegen/xai-grok-pager/src/lib.rs b/crates/codegen/xai-grok-pager/src/lib.rs
|
|
index 2747f2a7a35f6c4fa444cc830cb06e5049e4e594..a8309e6b5ecfd7c992afaafbda0e10ad370486a4 100644
|
|
--- a/crates/codegen/xai-grok-pager/src/lib.rs
|
|
+++ b/crates/codegen/xai-grok-pager/src/lib.rs
|
|
@@ -44,6 +44,8 @@ pub mod notifications;
|
|
pub mod obf;
|
|
pub mod plugin_cmd;
|
|
pub mod pty_wrap;
|
|
+/// `/rc` remote control: mirrors the live session to a grok-glance server.
|
|
+pub(crate) mod rc;
|
|
pub mod recent_dirs;
|
|
pub mod scrollback;
|
|
pub mod search;
|
|
diff --git a/crates/codegen/xai-grok-pager/src/rc/mod.rs b/crates/codegen/xai-grok-pager/src/rc/mod.rs
|
|
new file mode 100644
|
|
index 0000000000000000000000000000000000000000..402a021fea440914acba09b9c80e2401f79d8e40
|
|
--- /dev/null
|
|
+++ b/crates/codegen/xai-grok-pager/src/rc/mod.rs
|
|
@@ -0,0 +1,338 @@
|
|
+//! `/rc` -- remote control: mirror the live session to a grok-glance server.
|
|
+//!
|
|
+//! # Why this is a tee and not another entry point
|
|
+//!
|
|
+//! `grok` already speaks ACP four ways (`agent stdio`, `serve`, `leader`,
|
|
+//! `headless`), and every one of them *replaces* the TUI. Remote control must
|
|
+//! not: the terminal session stays exactly as interactive as it was, and the
|
|
+//! browser becomes a second peer on the same session rather than its successor.
|
|
+//!
|
|
+//! What makes that cheap is that the TUI is *itself* an ACP client. It holds an
|
|
+//! [`AcpAgentTx`](xai_acp_lib::AcpAgentTx) to send to the agent and an
|
|
+//! `AcpClientRx` to receive from it. Everything the terminal knows arrives on
|
|
+//! that receiver. So remote control needs no changes to the agent runtime, the
|
|
+//! session actor, or the shell -- only a tee on the receive path and a clone of
|
|
+//! the send half:
|
|
+//!
|
|
+//! ```text
|
|
+//! ┌─────────────┐
|
|
+//! agent ── AcpClientMessage ─▶│ intercept │─▶ acp_handler ─▶ TUI
|
|
+//! └──────┬──────┘
|
|
+//! │ copy
|
|
+//! ▼
|
|
+//! bridge task ──WS──▶ glance ──▶ browser
|
|
+//! │
|
|
+//! agent ◀── acp_send(prompt/cancel) ───────┘
|
|
+//! ```
|
|
+//!
|
|
+//! The tee is transport-agnostic: both `connect()` and `connect_via_leader()`
|
|
+//! yield the same `AcpConnection`, so `/rc` behaves identically whether or not
|
|
+//! the session is leader-hosted.
|
|
+//!
|
|
+//! # First responder wins
|
|
+//!
|
|
+//! Interactive reverse-requests (permission, `ask_user_question`,
|
|
+//! `exit_plan_mode`) carry their own reply channel. The tee lifts the real
|
|
+//! `response_tx` out, hands the TUI a substitute, forwards a copy to glance, and
|
|
+//! races the two answers -- whichever arrives first is the answer, and the other
|
|
+//! side is told to retract its dialog. Neither side is privileged.
|
|
+//!
|
|
+//! This mirrors what the leader already does for multi-pane sessions
|
|
+//! (`is_interaction_request` + `InteractionResolved` broadcast); the same three
|
|
+//! methods are covered, so a `/rc` user is not stranded the moment the agent
|
|
+//! asks a question instead of requesting a permission.
|
|
+//!
|
|
+//! # Failure is always local-only
|
|
+//!
|
|
+//! Every failure path here degrades to "the terminal keeps working". The bridge
|
|
+//! runs in its own task, reconnects with backoff, and the tee's fast path when
|
|
+//! remote control is off is a plain move.
|
|
+
|
|
+mod protocol;
|
|
+mod ring;
|
|
+mod tee;
|
|
+mod transport;
|
|
+
|
|
+use std::sync::atomic::{AtomicU64, Ordering};
|
|
+
|
|
+use tokio::sync::{mpsc, oneshot};
|
|
+use xai_acp_lib::AcpAgentTx;
|
|
+
|
|
+/// Client identifier stamped on prompts and cancels that originate in the
|
|
+/// browser, so the session log attributes them to glance rather than silently
|
|
+/// impersonating the terminal.
|
|
+pub(crate) const RC_CLIENT_ID: &str = "glance";
|
|
+
|
|
+/// What `/rc` needs to dial out, resolved from `[remote_control]` + env.
|
|
+#[derive(Clone)]
|
|
+pub(crate) struct RcSettings {
|
|
+ pub(crate) url: String,
|
|
+ pub(crate) api_key: String,
|
|
+ pub(crate) replay_buffer: usize,
|
|
+}
|
|
+
|
|
+/// Hand-written so the bearer token cannot reach a log line, a panic message,
|
|
+/// or a crash report through a stray `{:?}`.
|
|
+impl std::fmt::Debug for RcSettings {
|
|
+ fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
|
+ f.debug_struct("RcSettings")
|
|
+ .field("url", &self.url)
|
|
+ .field("api_key", &"<redacted>")
|
|
+ .field("replay_buffer", &self.replay_buffer)
|
|
+ .finish()
|
|
+ }
|
|
+}
|
|
+
|
|
+/// Session identity, sent in the `initialize` reply so glance can label the
|
|
+/// session without asking the agent anything.
|
|
+#[derive(Clone, Debug, Default)]
|
|
+pub(crate) struct SessionMeta {
|
|
+ pub(crate) session_id: Option<String>,
|
|
+ pub(crate) cwd: Option<String>,
|
|
+ pub(crate) title: Option<String>,
|
|
+ pub(crate) model: Option<String>,
|
|
+ pub(crate) hostname: Option<String>,
|
|
+ pub(crate) version: Option<String>,
|
|
+}
|
|
+
|
|
+impl SessionMeta {
|
|
+ fn to_json(&self) -> serde_json::Value {
|
|
+ serde_json::json!({
|
|
+ "sessionId": self.session_id,
|
|
+ "cwd": self.cwd,
|
|
+ "title": self.title,
|
|
+ "model": self.model,
|
|
+ "hostname": self.hostname,
|
|
+ "version": self.version,
|
|
+ })
|
|
+ }
|
|
+}
|
|
+
|
|
+/// A start request parked on [`AppView`](crate::app::AppView) by `/rc` and
|
|
+/// consumed at the top of the event loop.
|
|
+///
|
|
+/// The bridge cannot be spawned from the dispatch handler because its event
|
|
+/// receiver has to live as a local of the event loop -- holding it on `AppView`
|
|
+/// would put a `&mut app` borrow inside the main `select!`, where it would
|
|
+/// collide with every arm that mutates the app. Voice parks
|
|
+/// `VoiceState::ColdStart` for exactly the same reason.
|
|
+pub(crate) struct RcStart {
|
|
+ pub(crate) settings: RcSettings,
|
|
+ pub(crate) meta: SessionMeta,
|
|
+}
|
|
+
|
|
+/// Resolve `[remote_control]` from the layered config plus env overrides.
|
|
+///
|
|
+/// Returns a message fit to print in the scrollback rather than an error type:
|
|
+/// every failure here is "you have not configured this yet", and the useful
|
|
+/// response is a sentence naming the missing key.
|
|
+pub(crate) fn load_settings() -> Result<RcSettings, String> {
|
|
+ use xai_grok_shell::agent::config::RemoteControlConfig;
|
|
+
|
|
+ // A missing file or a missing section is not an error: `GROK_RC_URL` +
|
|
+ // `GROK_RC_API_KEY` alone are a supported way to configure this, and the
|
|
+ // defaults resolve those.
|
|
+ let config = xai_grok_shell::config::load_effective_config()
|
|
+ .ok()
|
|
+ .and_then(|root| root.get("remote_control").cloned())
|
|
+ .and_then(|value| value.try_into::<RemoteControlConfig>().ok())
|
|
+ .unwrap_or_default();
|
|
+
|
|
+ match (config.resolved_url(), config.resolved_api_key()) {
|
|
+ (Some(url), Some(api_key)) => Ok(RcSettings {
|
|
+ url,
|
|
+ api_key,
|
|
+ replay_buffer: config.replay_buffer_len(),
|
|
+ }),
|
|
+ (None, Some(_)) => Err("remote control needs a URL: set `url` under \
|
|
+ [remote_control] in ~/.grok/config.toml, or export GROK_RC_URL"
|
|
+ .into()),
|
|
+ (Some(_), None) => Err("remote control needs an API key: set `api_key` under \
|
|
+ [remote_control] in ~/.grok/config.toml, or export GROK_RC_API_KEY"
|
|
+ .into()),
|
|
+ (None, None) => Err("remote control is not configured. Add [remote_control] with \
|
|
+ `url` and `api_key` to ~/.grok/config.toml (or export GROK_RC_URL and \
|
|
+ GROK_RC_API_KEY), then run /rc again."
|
|
+ .into()),
|
|
+ }
|
|
+}
|
|
+
|
|
+/// Where the bridge is, for the status line and `/rc status`.
|
|
+#[derive(Clone, Debug, PartialEq, Eq, Default)]
|
|
+pub(crate) enum RcState {
|
|
+ #[default]
|
|
+ Off,
|
|
+ Connecting,
|
|
+ Connected {
|
|
+ /// Browsers currently attached, as reported by glance. `None` until it
|
|
+ /// says -- distinct from `Some(0)`, which means "connected, nobody
|
|
+ /// watching".
|
|
+ viewers: Option<usize>,
|
|
+ },
|
|
+ /// Lost the socket and retrying. The session is unaffected.
|
|
+ Reconnecting {
|
|
+ attempt: u32,
|
|
+ detail: String,
|
|
+ },
|
|
+ /// Gave up, or was rejected in a way retrying cannot fix (bad API key).
|
|
+ Failed {
|
|
+ detail: String,
|
|
+ },
|
|
+}
|
|
+
|
|
+impl RcState {
|
|
+ pub(crate) fn is_active(&self) -> bool {
|
|
+ !matches!(self, Self::Off)
|
|
+ }
|
|
+
|
|
+ /// One-line summary for `/rc status` and the footer.
|
|
+ pub(crate) fn summary(&self) -> String {
|
|
+ match self {
|
|
+ Self::Off => "off".into(),
|
|
+ Self::Connecting => "connecting…".into(),
|
|
+ Self::Connected { viewers: None } => "connected".into(),
|
|
+ Self::Connected { viewers: Some(0) } => "connected (no viewers)".into(),
|
|
+ Self::Connected { viewers: Some(1) } => "connected (1 viewer)".into(),
|
|
+ Self::Connected { viewers: Some(n) } => format!("connected ({n} viewers)"),
|
|
+ Self::Reconnecting { attempt, detail } => {
|
|
+ format!("reconnecting (attempt {attempt}): {detail}")
|
|
+ }
|
|
+ Self::Failed { detail } => format!("failed: {detail}"),
|
|
+ }
|
|
+ }
|
|
+}
|
|
+
|
|
+/// Bridge -> event loop. Delivered on the `/rc` arm of the main `select!`.
|
|
+#[derive(Debug)]
|
|
+pub(crate) enum RcEvent {
|
|
+ State(RcState),
|
|
+ /// Glance answered an interaction before the terminal did: retract the
|
|
+ /// local modal for this tool call.
|
|
+ InteractionResolvedRemotely { tool_call_id: String },
|
|
+ /// Something worth a line in the scrollback (remote prompt accepted, remote
|
|
+ /// stop, a rejected request). Not an error path -- errors go through
|
|
+ /// [`RcState::Failed`].
|
|
+ Notice(String),
|
|
+}
|
|
+
|
|
+/// Event loop / tee -> bridge. One channel so ordering between a mirrored
|
|
+/// update and the interaction that follows it is preserved.
|
|
+pub(crate) enum ToBridge {
|
|
+ /// A mirrored notification, pre-framed as the exact text frame to send.
|
|
+ Frame(String),
|
|
+ /// An interactive reverse-request that needs an answer from either side.
|
|
+ Interaction(Interaction),
|
|
+ /// The terminal answered first -- retract glance's dialog and drop the
|
|
+ /// pending entry so a late remote answer is ignored.
|
|
+ InteractionSettled { id: u64 },
|
|
+ Shutdown,
|
|
+}
|
|
+
|
|
+/// One interaction in flight, as handed to the bridge.
|
|
+pub(crate) struct Interaction {
|
|
+ /// JSON-RPC id used toward glance; also the key of the pending table.
|
|
+ pub(crate) id: u64,
|
|
+ /// ACP method name, forwarded verbatim so glance sees real ACP.
|
|
+ pub(crate) method: &'static str,
|
|
+ pub(crate) params: serde_json::Value,
|
|
+ /// Identifies the modal on both sides; what a retraction is keyed by.
|
|
+ pub(crate) tool_call_id: Option<String>,
|
|
+ /// Resolved with glance's raw `result` when it answers first.
|
|
+ pub(crate) remote_tx: oneshot::Sender<serde_json::Value>,
|
|
+}
|
|
+
|
|
+/// Handle held by `AppView` while remote control is on.
|
|
+pub(crate) struct RcHandle {
|
|
+ to_bridge: mpsc::UnboundedSender<ToBridge>,
|
|
+ events: mpsc::UnboundedSender<RcEvent>,
|
|
+ url: String,
|
|
+ next_id: AtomicU64,
|
|
+}
|
|
+
|
|
+impl RcHandle {
|
|
+ pub(crate) fn url(&self) -> &str {
|
|
+ &self.url
|
|
+ }
|
|
+
|
|
+ /// True once the bridge task has gone away. The tee checks this so a dead
|
|
+ /// bridge degrades to a plain passthrough instead of accumulating sends
|
|
+ /// into a closed channel.
|
|
+ pub(crate) fn is_closed(&self) -> bool {
|
|
+ self.to_bridge.is_closed()
|
|
+ }
|
|
+
|
|
+ fn next_id(&self) -> u64 {
|
|
+ self.next_id.fetch_add(1, Ordering::Relaxed)
|
|
+ }
|
|
+
|
|
+ /// Best-effort send. A closed channel means the bridge task exited; the
|
|
+ /// session is unaffected and the next `RcEvent` will report the state.
|
|
+ fn send(&self, msg: ToBridge) {
|
|
+ let _ = self.to_bridge.send(msg);
|
|
+ }
|
|
+
|
|
+ /// Ask the bridge to close the socket and exit. Fire-and-forget, called
|
|
+ /// from `/rc off` and from `Action::Quit`.
|
|
+ pub(crate) fn shutdown(&self) {
|
|
+ self.send(ToBridge::Shutdown);
|
|
+ }
|
|
+}
|
|
+
|
|
+/// Start the bridge task and return the handle plus the event stream the main
|
|
+/// loop selects on.
|
|
+///
|
|
+/// Dialing happens inside the task, so this never blocks the TUI: `/rc` returns
|
|
+/// immediately in [`RcState::Connecting`] and the outcome arrives as an event.
|
|
+pub(crate) fn start(
|
|
+ settings: RcSettings,
|
|
+ meta: SessionMeta,
|
|
+ acp_tx: AcpAgentTx,
|
|
+) -> (RcHandle, mpsc::UnboundedReceiver<RcEvent>) {
|
|
+ let (to_bridge_tx, to_bridge_rx) = mpsc::unbounded_channel();
|
|
+ let (event_tx, event_rx) = mpsc::unbounded_channel();
|
|
+
|
|
+ let handle = RcHandle {
|
|
+ to_bridge: to_bridge_tx,
|
|
+ events: event_tx.clone(),
|
|
+ url: settings.url.clone(),
|
|
+ next_id: AtomicU64::new(1),
|
|
+ };
|
|
+
|
|
+ tokio::spawn(transport::run(settings, meta, acp_tx, to_bridge_rx, event_tx));
|
|
+
|
|
+ (handle, event_rx)
|
|
+}
|
|
+
|
|
+#[cfg(test)]
|
|
+mod tests {
|
|
+ use super::*;
|
|
+
|
|
+ #[test]
|
|
+ fn state_summaries_distinguish_no_viewers_from_unknown() {
|
|
+ // `Some(0)` is "connected, nobody watching"; `None` is "glance has not
|
|
+ // told us yet". Collapsing them would make `/rc status` lie right after
|
|
+ // connecting.
|
|
+ assert_eq!(RcState::Connected { viewers: None }.summary(), "connected");
|
|
+ assert_eq!(
|
|
+ RcState::Connected { viewers: Some(0) }.summary(),
|
|
+ "connected (no viewers)"
|
|
+ );
|
|
+ assert_eq!(
|
|
+ RcState::Connected { viewers: Some(2) }.summary(),
|
|
+ "connected (2 viewers)"
|
|
+ );
|
|
+ }
|
|
+
|
|
+ #[test]
|
|
+ fn only_off_is_inactive() {
|
|
+ assert!(!RcState::Off.is_active());
|
|
+ assert!(RcState::Connecting.is_active());
|
|
+ assert!(
|
|
+ RcState::Failed {
|
|
+ detail: "bad key".into()
|
|
+ }
|
|
+ .is_active(),
|
|
+ "a failed bridge is still 'on' -- /rc must report it, not look off"
|
|
+ );
|
|
+ }
|
|
+}
|
|
diff --git a/crates/codegen/xai-grok-pager/src/rc/protocol.rs b/crates/codegen/xai-grok-pager/src/rc/protocol.rs
|
|
new file mode 100644
|
|
index 0000000000000000000000000000000000000000..4bba548cc7789aee2c5af941ce68e2d86ef3a127
|
|
--- /dev/null
|
|
+++ b/crates/codegen/xai-grok-pager/src/rc/protocol.rs
|
|
@@ -0,0 +1,165 @@
|
|
+//! JSON-RPC 2.0 framing for the grok <-> glance link.
|
|
+//!
|
|
+//! The link is ACP with the roles inverted relative to the TUI: over this
|
|
+//! socket **grok is the Agent and glance is the Client**. That is deliberate --
|
|
+//! it makes glance a stock ACP client that happens to talk to a session it did
|
|
+//! not create, instead of a bespoke protocol only grok can speak.
|
|
+//!
|
|
+//! Only the framing lives here. What each method *means* is in
|
|
+//! [`transport`](super::transport).
|
|
+
|
|
+use serde::Deserialize;
|
|
+
|
|
+pub(crate) const JSONRPC_VERSION: &str = "2.0";
|
|
+
|
|
+// --- methods glance may call on us -----------------------------------------
|
|
+
|
|
+/// Handshake. Answered with our capabilities plus session metadata, so glance
|
|
+/// can label the session without a second round-trip.
|
|
+pub(crate) const M_INITIALIZE: &str = "initialize";
|
|
+/// The one session this bridge mirrors. `/rc` is session-scoped by design.
|
|
+pub(crate) const M_SESSION_LIST: &str = "session/list";
|
|
+/// Send a prompt as if it had been typed in the terminal.
|
|
+pub(crate) const M_SESSION_PROMPT: &str = "session/prompt";
|
|
+/// Interrupt the running turn.
|
|
+pub(crate) const M_SESSION_CANCEL: &str = "session/cancel";
|
|
+/// Re-send the replay ring (used after a browser reattaches).
|
|
+pub(crate) const M_RC_REPLAY: &str = "x.ai/rc/replay";
|
|
+
|
|
+// --- methods we call on glance ---------------------------------------------
|
|
+
|
|
+/// Mirrored standard session update.
|
|
+pub(crate) const M_SESSION_UPDATE: &str = "session/update";
|
|
+/// Bridge status: connection state, replay bounds, session metadata.
|
|
+pub(crate) const M_RC_STATUS: &str = "x.ai/rc/status";
|
|
+/// An interaction we forwarded was answered in the terminal first; glance must
|
|
+/// retract its dialog. The mirror image of grok's own `InteractionResolved`.
|
|
+pub(crate) const M_RC_INTERACTION_CANCELLED: &str = "x.ai/rc/interaction_cancelled";
|
|
+
|
|
+// --- error codes ------------------------------------------------------------
|
|
+
|
|
+pub(crate) const E_METHOD_NOT_FOUND: i32 = -32601;
|
|
+pub(crate) const E_INVALID_PARAMS: i32 = -32602;
|
|
+pub(crate) const E_INTERNAL: i32 = -32603;
|
|
+
|
|
+/// A frame arriving from glance.
|
|
+///
|
|
+/// One struct for all three JSON-RPC shapes, disambiguated by which fields are
|
|
+/// present: `method` + `id` is a request, `method` alone is a notification,
|
|
+/// `result`/`error` is a response to something we sent. Parsing permissively
|
|
+/// and classifying afterwards means a frame we do not understand produces a
|
|
+/// JSON-RPC error rather than killing the socket -- which would take the
|
|
+/// mirror down over a cosmetic disagreement.
|
|
+#[derive(Debug, Deserialize)]
|
|
+pub(crate) struct InboundFrame {
|
|
+ #[serde(default)]
|
|
+ pub(crate) id: Option<serde_json::Value>,
|
|
+ #[serde(default)]
|
|
+ pub(crate) method: Option<String>,
|
|
+ #[serde(default)]
|
|
+ pub(crate) params: Option<serde_json::Value>,
|
|
+ #[serde(default)]
|
|
+ pub(crate) result: Option<serde_json::Value>,
|
|
+ #[serde(default)]
|
|
+ pub(crate) error: Option<serde_json::Value>,
|
|
+}
|
|
+
|
|
+impl InboundFrame {
|
|
+ /// A response carries no `method` and has exactly one of `result`/`error`.
|
|
+ pub(crate) fn is_response(&self) -> bool {
|
|
+ self.method.is_none() && (self.result.is_some() || self.error.is_some())
|
|
+ }
|
|
+}
|
|
+
|
|
+/// Serialise a request we are sending to glance.
|
|
+pub(crate) fn request(id: u64, method: &str, params: serde_json::Value) -> String {
|
|
+ encode(serde_json::json!({
|
|
+ "jsonrpc": JSONRPC_VERSION,
|
|
+ "id": id,
|
|
+ "method": method,
|
|
+ "params": params,
|
|
+ }))
|
|
+}
|
|
+
|
|
+/// Serialise a notification we are sending to glance.
|
|
+pub(crate) fn notification(method: &str, params: serde_json::Value) -> String {
|
|
+ encode(serde_json::json!({
|
|
+ "jsonrpc": JSONRPC_VERSION,
|
|
+ "method": method,
|
|
+ "params": params,
|
|
+ }))
|
|
+}
|
|
+
|
|
+/// Serialise a successful reply to one of glance's requests.
|
|
+pub(crate) fn response(id: serde_json::Value, result: serde_json::Value) -> String {
|
|
+ encode(serde_json::json!({
|
|
+ "jsonrpc": JSONRPC_VERSION,
|
|
+ "id": id,
|
|
+ "result": result,
|
|
+ }))
|
|
+}
|
|
+
|
|
+/// Serialise a failed reply. Unsupported methods get a real `method not found`
|
|
+/// rather than a faked success -- glance must be able to tell what this bridge
|
|
+/// can actually do.
|
|
+pub(crate) fn error_response(id: serde_json::Value, code: i32, message: impl Into<String>) -> String {
|
|
+ encode(serde_json::json!({
|
|
+ "jsonrpc": JSONRPC_VERSION,
|
|
+ "id": id,
|
|
+ "error": { "code": code, "message": message.into() },
|
|
+ }))
|
|
+}
|
|
+
|
|
+/// `serde_json::to_string` on a `Value` built from owned data cannot fail
|
|
+/// (no borrowed keys, no non-string map keys, no custom `Serialize`), so the
|
|
+/// fallback here is unreachable in practice. It exists so a framing bug can
|
|
+/// never panic the bridge and take the mirror down mid-turn.
|
|
+fn encode(value: serde_json::Value) -> String {
|
|
+ serde_json::to_string(&value).unwrap_or_else(|e| {
|
|
+ tracing::error!(error = %e, "rc: failed to encode frame");
|
|
+ format!(
|
|
+ r#"{{"jsonrpc":"{JSONRPC_VERSION}","method":"{M_RC_STATUS}","params":{{"error":"encode failed"}}}}"#
|
|
+ )
|
|
+ })
|
|
+}
|
|
+
|
|
+#[cfg(test)]
|
|
+mod tests {
|
|
+ use super::*;
|
|
+
|
|
+ #[test]
|
|
+ fn classifies_the_three_frame_shapes() {
|
|
+ let req: InboundFrame =
|
|
+ serde_json::from_str(r#"{"jsonrpc":"2.0","id":1,"method":"session/prompt"}"#).unwrap();
|
|
+ assert!(!req.is_response());
|
|
+ assert_eq!(req.method.as_deref(), Some("session/prompt"));
|
|
+
|
|
+ let notif: InboundFrame =
|
|
+ serde_json::from_str(r#"{"jsonrpc":"2.0","method":"session/cancel"}"#).unwrap();
|
|
+ assert!(!notif.is_response());
|
|
+ assert!(notif.id.is_none());
|
|
+
|
|
+ let resp: InboundFrame =
|
|
+ serde_json::from_str(r#"{"jsonrpc":"2.0","id":7,"result":{"ok":true}}"#).unwrap();
|
|
+ assert!(resp.is_response());
|
|
+ }
|
|
+
|
|
+ #[test]
|
|
+ fn an_error_reply_is_still_a_response() {
|
|
+ // Glance refusing a permission forward must resolve our pending entry,
|
|
+ // not be mistaken for a request and answered with `method not found`.
|
|
+ let resp: InboundFrame =
|
|
+ serde_json::from_str(r#"{"jsonrpc":"2.0","id":7,"error":{"code":-32603}}"#).unwrap();
|
|
+ assert!(resp.is_response());
|
|
+ }
|
|
+
|
|
+ #[test]
|
|
+ fn request_frame_round_trips() {
|
|
+ let frame = request(3, M_SESSION_UPDATE, serde_json::json!({"a": 1}));
|
|
+ let parsed: serde_json::Value = serde_json::from_str(&frame).unwrap();
|
|
+ assert_eq!(parsed["jsonrpc"], JSONRPC_VERSION);
|
|
+ assert_eq!(parsed["id"], 3);
|
|
+ assert_eq!(parsed["method"], M_SESSION_UPDATE);
|
|
+ assert_eq!(parsed["params"]["a"], 1);
|
|
+ }
|
|
+}
|
|
diff --git a/crates/codegen/xai-grok-pager/src/rc/ring.rs b/crates/codegen/xai-grok-pager/src/rc/ring.rs
|
|
new file mode 100644
|
|
index 0000000000000000000000000000000000000000..89060b8237b536046880dd6a53a864fb4387beb1
|
|
--- /dev/null
|
|
+++ b/crates/codegen/xai-grok-pager/src/rc/ring.rs
|
|
@@ -0,0 +1,79 @@
|
|
+//! Bounded replay buffer for the glance bridge.
|
|
+//!
|
|
+//! A browser that attaches mid-session, or reattaches after glance restarted,
|
|
+//! has missed everything the agent emitted so far. The ring holds the recent
|
|
+//! tail so it can be caught up without the agent replaying anything -- the
|
|
+//! local session is never asked to do work on the remote's behalf.
|
|
+//!
|
|
+//! Frames are stored **already serialised**, for two reasons: replay is then a
|
|
+//! straight write with no re-encoding, and a reattaching viewer receives frames
|
|
+//! byte-identical to the ones it missed, so its `_meta.eventId` dedup behaves
|
|
+//! exactly as it would have live.
|
|
+
|
|
+use std::collections::VecDeque;
|
|
+
|
|
+pub(crate) struct ReplayRing {
|
|
+ frames: VecDeque<String>,
|
|
+ capacity: usize,
|
|
+ /// Frames evicted since start. A non-zero value tells glance its history is
|
|
+ /// truncated rather than complete, which is the difference between "the
|
|
+ /// session started here" and "you are missing the beginning".
|
|
+ dropped: u64,
|
|
+}
|
|
+
|
|
+impl ReplayRing {
|
|
+ pub(crate) fn new(capacity: usize) -> Self {
|
|
+ // A zero capacity would make `push` an infinite pop loop.
|
|
+ let capacity = capacity.max(1);
|
|
+ Self {
|
|
+ frames: VecDeque::with_capacity(capacity.min(256)),
|
|
+ capacity,
|
|
+ dropped: 0,
|
|
+ }
|
|
+ }
|
|
+
|
|
+ pub(crate) fn push(&mut self, frame: String) {
|
|
+ while self.frames.len() >= self.capacity {
|
|
+ self.frames.pop_front();
|
|
+ self.dropped += 1;
|
|
+ }
|
|
+ self.frames.push_back(frame);
|
|
+ }
|
|
+
|
|
+ pub(crate) fn iter(&self) -> impl Iterator<Item = &String> {
|
|
+ self.frames.iter()
|
|
+ }
|
|
+
|
|
+ pub(crate) fn len(&self) -> usize {
|
|
+ self.frames.len()
|
|
+ }
|
|
+
|
|
+ pub(crate) fn dropped(&self) -> u64 {
|
|
+ self.dropped
|
|
+ }
|
|
+}
|
|
+
|
|
+#[cfg(test)]
|
|
+mod tests {
|
|
+ use super::ReplayRing;
|
|
+
|
|
+ #[test]
|
|
+ fn evicts_oldest_and_counts_the_loss() {
|
|
+ let mut ring = ReplayRing::new(3);
|
|
+ for i in 0..5 {
|
|
+ ring.push(format!("f{i}"));
|
|
+ }
|
|
+ let kept: Vec<_> = ring.iter().cloned().collect();
|
|
+ assert_eq!(kept, vec!["f2", "f3", "f4"]);
|
|
+ assert_eq!(ring.len(), 3);
|
|
+ assert_eq!(ring.dropped(), 2, "the two evicted frames must be reported");
|
|
+ }
|
|
+
|
|
+ #[test]
|
|
+ fn zero_capacity_is_clamped_not_fatal() {
|
|
+ // `replay_buffer = 0` in config must degrade to "keep one", not hang.
|
|
+ let mut ring = ReplayRing::new(0);
|
|
+ ring.push("only".into());
|
|
+ assert_eq!(ring.len(), 1);
|
|
+ }
|
|
+}
|
|
diff --git a/crates/codegen/xai-grok-pager/src/rc/tee.rs b/crates/codegen/xai-grok-pager/src/rc/tee.rs
|
|
new file mode 100644
|
|
index 0000000000000000000000000000000000000000..6605a554530279cd1a34a1d8a7f76af83f740286
|
|
--- /dev/null
|
|
+++ b/crates/codegen/xai-grok-pager/src/rc/tee.rs
|
|
@@ -0,0 +1,352 @@
|
|
+//! The tee: copy what the agent says to glance, and race what it asks.
|
|
+//!
|
|
+//! [`RcHandle::intercept`] sits at the top of
|
|
+//! [`acp_handler::handle`](crate::app::acp_handler::handle), which is the single
|
|
+//! place every inbound `AcpClientMessage` passes through -- the event loop calls
|
|
+//! it both on `recv` and inside its drain batch, so one insertion covers both.
|
|
+//!
|
|
+//! Two rules govern everything here:
|
|
+//!
|
|
+//! 1. **The original message is passed through unchanged.** Mirroring copies;
|
|
+//! it never rewrites, reorders, or drops. `_meta` in particular is forwarded
|
|
+//! verbatim -- `eventId`, `promptId`, `chunkId` and `isReplay` are what let a
|
|
+//! viewer dedup and order the stream the same way the TUI does.
|
|
+//! 2. **Nothing here can block or fail the local session.** Sends are
|
|
+//! best-effort on unbounded channels, the race runs in a spawned task, and a
|
|
+//! dead bridge degrades to a plain move.
|
|
+
|
|
+use agent_client_protocol as acp;
|
|
+use tokio::sync::oneshot;
|
|
+use xai_acp_lib::{AcpArgs, AcpClientMessage, AcpResult};
|
|
+
|
|
+use super::{Interaction, RcEvent, RcHandle, ToBridge, protocol};
|
|
+
|
|
+/// Ext methods that are interactive reverse-requests rather than notifications.
|
|
+///
|
|
+/// Deliberately the same set the leader singles out in `is_interaction_request`
|
|
+/// (`leader/server.rs`): a `/rc` user who could approve tool calls but not
|
|
+/// answer a question or leave plan mode would be stranded the first time the
|
|
+/// agent did either.
|
|
+fn is_interaction_ext_method(method: &str) -> bool {
|
|
+ matches!(method, "x.ai/ask_user_question" | "x.ai/exit_plan_mode")
|
|
+}
|
|
+
|
|
+/// Which side answered first. Computed inside `select!` so the handler arms
|
|
+/// touch neither receiver -- the loser is used *after* the statement, once its
|
|
+/// borrow has been released.
|
|
+enum Winner<T> {
|
|
+ Local(AcpResult<T>),
|
|
+ /// The terminal dropped the request without answering.
|
|
+ LocalGone,
|
|
+ Remote(serde_json::Value),
|
|
+ /// Glance went away mid-interaction.
|
|
+ RemoteGone,
|
|
+}
|
|
+
|
|
+impl RcHandle {
|
|
+ /// Mirror `msg` to glance and return it for the TUI.
|
|
+ ///
|
|
+ /// The hot path (streaming deltas while remote control is off, or after the
|
|
+ /// bridge died) is a match and a move.
|
|
+ pub(crate) fn intercept(&self, msg: AcpClientMessage) -> AcpClientMessage {
|
|
+ if self.is_closed() {
|
|
+ return msg;
|
|
+ }
|
|
+ match msg {
|
|
+ AcpClientMessage::SessionNotification(args) => {
|
|
+ match serde_json::to_value(&args.request) {
|
|
+ Ok(params) => self.send(ToBridge::Frame(protocol::notification(
|
|
+ protocol::M_SESSION_UPDATE,
|
|
+ params,
|
|
+ ))),
|
|
+ Err(e) => tracing::warn!(error = %e, "rc: could not mirror session/update"),
|
|
+ }
|
|
+ AcpClientMessage::SessionNotification(args)
|
|
+ }
|
|
+
|
|
+ // The xAI rail. grok emits ~60 grok-specific variants here (tool-call
|
|
+ // deltas, subagents, retries, turn boundaries); a mirror carrying only
|
|
+ // `session/update` would show glance a transcript with the streaming
|
|
+ // taken out. Same predicate the pager itself uses to recognise them.
|
|
+ AcpClientMessage::ExtNotification(args)
|
|
+ if crate::acp::is_session_update_ext_method(args.request.method.as_ref()) =>
|
|
+ {
|
|
+ match serde_json::from_str::<serde_json::Value>(args.request.params.get()) {
|
|
+ Ok(params) => {
|
|
+ let method = args.request.method.to_string();
|
|
+ self.send(ToBridge::Frame(protocol::notification(&method, params)));
|
|
+ }
|
|
+ Err(e) => tracing::warn!(error = %e, "rc: could not mirror ext notification"),
|
|
+ }
|
|
+ AcpClientMessage::ExtNotification(args)
|
|
+ }
|
|
+
|
|
+ AcpClientMessage::RequestPermission(args) => self.race_permission(args),
|
|
+
|
|
+ AcpClientMessage::ExtMethod(args)
|
|
+ if is_interaction_ext_method(args.request.method.as_ref()) =>
|
|
+ {
|
|
+ self.race_ext(args)
|
|
+ }
|
|
+
|
|
+ // Everything else is local business. In the pager's default
|
|
+ // configuration `fs/*` and `terminal/*` are not even advertised, so
|
|
+ // this arm is mostly the non-interactive ext methods.
|
|
+ other => other,
|
|
+ }
|
|
+ }
|
|
+
|
|
+ fn race_permission(
|
|
+ &self,
|
|
+ args: AcpArgs<acp::RequestPermissionRequest>,
|
|
+ ) -> AcpClientMessage {
|
|
+ let AcpArgs {
|
|
+ request,
|
|
+ response_tx,
|
|
+ } = args;
|
|
+
|
|
+ let params = match serde_json::to_value(&request) {
|
|
+ Ok(params) => params,
|
|
+ Err(e) => {
|
|
+ // Un-mirrorable: hand it to the terminal untouched rather than
|
|
+ // holding a request nobody can answer.
|
|
+ tracing::warn!(error = %e, "rc: could not forward permission request");
|
|
+ return AcpClientMessage::RequestPermission(AcpArgs {
|
|
+ request,
|
|
+ response_tx,
|
|
+ });
|
|
+ }
|
|
+ };
|
|
+ let tool_call_id = request.tool_call.tool_call_id.0.to_string();
|
|
+
|
|
+ let (tui_tx, tui_rx) = oneshot::channel();
|
|
+ let id = self.begin_interaction(
|
|
+ acp::CLIENT_METHOD_NAMES.session_request_permission,
|
|
+ params,
|
|
+ Some(tool_call_id.clone()),
|
|
+ response_tx,
|
|
+ tui_rx,
|
|
+ |value| {
|
|
+ serde_json::from_value::<acp::RequestPermissionResponse>(value)
|
|
+ .map_err(|e| e.to_string())
|
|
+ },
|
|
+ );
|
|
+ tracing::debug!(id, %tool_call_id, "rc: permission request mirrored");
|
|
+
|
|
+ AcpClientMessage::RequestPermission(AcpArgs {
|
|
+ request,
|
|
+ response_tx: tui_tx,
|
|
+ })
|
|
+ }
|
|
+
|
|
+ fn race_ext(&self, args: AcpArgs<acp::ExtRequest>) -> AcpClientMessage {
|
|
+ let AcpArgs {
|
|
+ request,
|
|
+ response_tx,
|
|
+ } = args;
|
|
+
|
|
+ let params = match serde_json::from_str::<serde_json::Value>(request.params.get()) {
|
|
+ Ok(params) => params,
|
|
+ Err(e) => {
|
|
+ tracing::warn!(error = %e, method = %request.method, "rc: could not forward ext request");
|
|
+ return AcpClientMessage::ExtMethod(AcpArgs {
|
|
+ request,
|
|
+ response_tx,
|
|
+ });
|
|
+ }
|
|
+ };
|
|
+ // Both interactive ext requests carry `toolCallId`; it is what a
|
|
+ // retraction is keyed by on either side. Absent, the race still works --
|
|
+ // only the automatic dismissal of the loser's modal is lost.
|
|
+ let tool_call_id = params
|
|
+ .get("toolCallId")
|
|
+ .and_then(|v| v.as_str())
|
|
+ .map(str::to_owned);
|
|
+
|
|
+ // `method` is `Arc<str>` on the request, but the pending table wants a
|
|
+ // `&'static str`. Both interaction methods are known constants, so map
|
|
+ // rather than leak.
|
|
+ let method: &'static str = match request.method.as_ref() {
|
|
+ "x.ai/ask_user_question" => "x.ai/ask_user_question",
|
|
+ other => {
|
|
+ debug_assert_eq!(other, "x.ai/exit_plan_mode");
|
|
+ "x.ai/exit_plan_mode"
|
|
+ }
|
|
+ };
|
|
+
|
|
+ let (tui_tx, tui_rx) = oneshot::channel();
|
|
+ let id = self.begin_interaction(
|
|
+ method,
|
|
+ params,
|
|
+ tool_call_id.clone(),
|
|
+ response_tx,
|
|
+ tui_rx,
|
|
+ |value| {
|
|
+ serde_json::value::to_raw_value(&value)
|
|
+ .map(|raw| acp::ExtResponse::new(raw.into()))
|
|
+ .map_err(|e| e.to_string())
|
|
+ },
|
|
+ );
|
|
+ tracing::debug!(id, method, ?tool_call_id, "rc: ext interaction mirrored");
|
|
+
|
|
+ AcpClientMessage::ExtMethod(AcpArgs {
|
|
+ request,
|
|
+ response_tx: tui_tx,
|
|
+ })
|
|
+ }
|
|
+
|
|
+ /// Register an interaction with the bridge and spawn the task that races the
|
|
+ /// two answers. Returns the id used on the wire.
|
|
+ fn begin_interaction<T, F>(
|
|
+ &self,
|
|
+ method: &'static str,
|
|
+ params: serde_json::Value,
|
|
+ tool_call_id: Option<String>,
|
|
+ real_tx: oneshot::Sender<AcpResult<T>>,
|
|
+ tui_rx: oneshot::Receiver<AcpResult<T>>,
|
|
+ decode: F,
|
|
+ ) -> u64
|
|
+ where
|
|
+ T: Send + 'static,
|
|
+ F: FnOnce(serde_json::Value) -> Result<T, String> + Send + 'static,
|
|
+ {
|
|
+ let id = self.next_id();
|
|
+ let (remote_tx, remote_rx) = oneshot::channel();
|
|
+ self.send(ToBridge::Interaction(Interaction {
|
|
+ id,
|
|
+ method,
|
|
+ params,
|
|
+ tool_call_id: tool_call_id.clone(),
|
|
+ remote_tx,
|
|
+ }));
|
|
+
|
|
+ let to_bridge = self.to_bridge.clone();
|
|
+ let events = self.events.clone();
|
|
+ tokio::spawn(async move {
|
|
+ let mut tui_rx = tui_rx;
|
|
+ let mut remote_rx = remote_rx;
|
|
+
|
|
+ let winner = tokio::select! {
|
|
+ local = &mut tui_rx => match local {
|
|
+ Ok(answer) => Winner::Local(answer),
|
|
+ Err(_) => Winner::LocalGone,
|
|
+ },
|
|
+ remote = &mut remote_rx => match remote {
|
|
+ Ok(value) => Winner::Remote(value),
|
|
+ Err(_) => Winner::RemoteGone,
|
|
+ },
|
|
+ };
|
|
+
|
|
+ match winner {
|
|
+ // The terminal answered first. Forward it and retract the
|
|
+ // browser's dialog; a late remote answer finds no pending entry
|
|
+ // and is dropped.
|
|
+ Winner::Local(answer) => {
|
|
+ let _ = real_tx.send(answer);
|
|
+ let _ = to_bridge.send(ToBridge::InteractionSettled { id });
|
|
+ }
|
|
+
|
|
+ // The terminal dropped the request without answering (session
|
|
+ // teardown, or a question displaced by a newer one). Answering on
|
|
+ // its behalf would be wrong, so give glance the rest of the window.
|
|
+ Winner::LocalGone => match remote_rx.await {
|
|
+ Ok(value) => {
|
|
+ finish_remote(value, real_tx, decode, &events, tool_call_id);
|
|
+ }
|
|
+ Err(_) => drop(real_tx),
|
|
+ },
|
|
+
|
|
+ Winner::Remote(value) => {
|
|
+ if let Some(real_tx) =
|
|
+ finish_remote(value, real_tx, decode, &events, tool_call_id)
|
|
+ {
|
|
+ // The remote answer did not decode, so it cannot resolve
|
|
+ // the agent's request. The terminal is still showing the
|
|
+ // modal -- let it stay authoritative.
|
|
+ match tui_rx.await {
|
|
+ Ok(answer) => {
|
|
+ let _ = real_tx.send(answer);
|
|
+ let _ = to_bridge.send(ToBridge::InteractionSettled { id });
|
|
+ }
|
|
+ Err(_) => drop(real_tx),
|
|
+ }
|
|
+ }
|
|
+ }
|
|
+
|
|
+ // Glance disconnected mid-interaction. Nothing to retract; the
|
|
+ // terminal was never told anything was in flight.
|
|
+ Winner::RemoteGone => match tui_rx.await {
|
|
+ Ok(answer) => {
|
|
+ let _ = real_tx.send(answer);
|
|
+ }
|
|
+ Err(_) => drop(real_tx),
|
|
+ },
|
|
+ }
|
|
+ });
|
|
+
|
|
+ id
|
|
+ }
|
|
+}
|
|
+
|
|
+/// Decode and deliver a remote answer.
|
|
+///
|
|
+/// Returns `Some(real_tx)` when the answer was unusable, handing the reply
|
|
+/// channel back so the caller can fall back to the terminal. A malformed remote
|
|
+/// answer must never resolve the agent's request -- `RequestPermissionOutcome`
|
|
+/// is `#[non_exhaustive]`, so an outcome this build does not know about is a
|
|
+/// real possibility and has to fail closed rather than panic or guess.
|
|
+fn finish_remote<T, F>(
|
|
+ value: serde_json::Value,
|
|
+ real_tx: oneshot::Sender<AcpResult<T>>,
|
|
+ decode: F,
|
|
+ events: &tokio::sync::mpsc::UnboundedSender<RcEvent>,
|
|
+ tool_call_id: Option<String>,
|
|
+) -> Option<oneshot::Sender<AcpResult<T>>>
|
|
+where
|
|
+ F: FnOnce(serde_json::Value) -> Result<T, String>,
|
|
+{
|
|
+ match decode(value) {
|
|
+ Ok(answer) => {
|
|
+ let _ = real_tx.send(Ok(answer));
|
|
+ if let Some(tool_call_id) = tool_call_id {
|
|
+ let _ = events.send(RcEvent::InteractionResolvedRemotely { tool_call_id });
|
|
+ }
|
|
+ None
|
|
+ }
|
|
+ Err(e) => {
|
|
+ tracing::warn!(error = %e, "rc: remote answer rejected");
|
|
+ let _ = events.send(RcEvent::Notice(format!(
|
|
+ "glance sent an answer this build could not read ({e}); answer in the terminal"
|
|
+ )));
|
|
+ Some(real_tx)
|
|
+ }
|
|
+ }
|
|
+}
|
|
+
|
|
+#[cfg(test)]
|
|
+mod tests {
|
|
+ use super::*;
|
|
+
|
|
+ #[test]
|
|
+ fn interaction_ext_methods_match_the_leaders_set() {
|
|
+ assert!(is_interaction_ext_method("x.ai/ask_user_question"));
|
|
+ assert!(is_interaction_ext_method("x.ai/exit_plan_mode"));
|
|
+ // Notifications must not be raced -- they have no reply to win.
|
|
+ assert!(!is_interaction_ext_method("x.ai/session_notification"));
|
|
+ assert!(!is_interaction_ext_method("x.ai/toggle_plan_mode"));
|
|
+ }
|
|
+
|
|
+ /// The two rails the mirror must carry. If upstream renames either, this
|
|
+ /// fails here rather than silently producing a transcript with the
|
|
+ /// streaming removed.
|
|
+ #[test]
|
|
+ fn both_notification_rails_are_recognised() {
|
|
+ assert!(crate::acp::is_session_update_ext_method(
|
|
+ "x.ai/session_notification"
|
|
+ ));
|
|
+ assert!(crate::acp::is_session_update_ext_method("x.ai/session/update"));
|
|
+ assert!(!crate::acp::is_session_update_ext_method(
|
|
+ "x.ai/ask_user_question"
|
|
+ ));
|
|
+ }
|
|
+}
|
|
diff --git a/crates/codegen/xai-grok-pager/src/rc/transport.rs b/crates/codegen/xai-grok-pager/src/rc/transport.rs
|
|
new file mode 100644
|
|
index 0000000000000000000000000000000000000000..98e7e45a9428953c29b14c78b33566740219612a
|
|
--- /dev/null
|
|
+++ b/crates/codegen/xai-grok-pager/src/rc/transport.rs
|
|
@@ -0,0 +1,639 @@
|
|
+//! The bridge task: one WebSocket to glance, for the life of `/rc`.
|
|
+//!
|
|
+//! Runs entirely off the event loop. Its only couplings to the TUI are the
|
|
+//! [`ToBridge`] channel it drains, the [`RcEvent`] channel it reports on, and a
|
|
+//! clone of [`AcpAgentTx`] -- which is how a remote prompt or stop reaches the
|
|
+//! agent by exactly the same path the terminal uses, rather than being injected
|
|
+//! into the TUI and replayed.
|
|
+//!
|
|
+//! Disconnection is routine, not exceptional: glance restarting must never be
|
|
+//! visible in the terminal beyond a status line. So the socket is dialed with
|
|
+//! backoff, traffic keeps flowing into the replay ring while it is down, and
|
|
+//! reconnecting replays the ring instead of asking the session to re-emit
|
|
+//! anything.
|
|
+
|
|
+use std::collections::HashMap;
|
|
+use std::time::Duration;
|
|
+
|
|
+use agent_client_protocol as acp;
|
|
+use futures_util::{SinkExt, StreamExt};
|
|
+use tokio::sync::{mpsc, oneshot};
|
|
+use tokio_tungstenite::tungstenite::Message;
|
|
+use tokio_tungstenite::tungstenite::client::IntoClientRequest;
|
|
+use xai_acp_lib::{AcpAgentTx, acp_send};
|
|
+
|
|
+use super::{
|
|
+ Interaction, RC_CLIENT_ID, RcEvent, RcSettings, RcState, SessionMeta, ToBridge, protocol,
|
|
+ ring::ReplayRing,
|
|
+};
|
|
+
|
|
+/// Long enough that a glance restart or a laptop lid-close is ridden out, short
|
|
+/// enough that a reconnect is not perceived as a hang.
|
|
+const BACKOFF_MIN: Duration = Duration::from_secs(1);
|
|
+const BACKOFF_MAX: Duration = Duration::from_secs(30);
|
|
+const CONNECT_TIMEOUT: Duration = Duration::from_secs(15);
|
|
+
|
|
+/// An interaction forwarded to glance and awaiting its answer.
|
|
+struct Pending {
|
|
+ remote_tx: oneshot::Sender<serde_json::Value>,
|
|
+ tool_call_id: Option<String>,
|
|
+}
|
|
+
|
|
+enum PumpOutcome {
|
|
+ /// `/rc off`, `Action::Quit`, or the handle was dropped.
|
|
+ Shutdown,
|
|
+ /// Socket died. Retry.
|
|
+ Disconnected(String),
|
|
+}
|
|
+
|
|
+enum AbsorbOutcome {
|
|
+ Shutdown,
|
|
+ Elapsed,
|
|
+}
|
|
+
|
|
+pub(super) async fn run(
|
|
+ settings: RcSettings,
|
|
+ meta: SessionMeta,
|
|
+ acp_tx: AcpAgentTx,
|
|
+ mut to_bridge: mpsc::UnboundedReceiver<ToBridge>,
|
|
+ events: mpsc::UnboundedSender<RcEvent>,
|
|
+) {
|
|
+ let mut ring = ReplayRing::new(settings.replay_buffer);
|
|
+ let mut attempt: u32 = 0;
|
|
+
|
|
+ loop {
|
|
+ let _ = events.send(RcEvent::State(if attempt == 0 {
|
|
+ RcState::Connecting
|
|
+ } else {
|
|
+ RcState::Reconnecting {
|
|
+ attempt,
|
|
+ detail: "dialing".into(),
|
|
+ }
|
|
+ }));
|
|
+
|
|
+ match dial(&settings).await {
|
|
+ Ok(ws) => {
|
|
+ let _ = events.send(RcEvent::State(RcState::Connected { viewers: None }));
|
|
+ tracing::info!(url = %settings.url, "rc: connected to glance");
|
|
+
|
|
+ let outcome = pump(
|
|
+ ws,
|
|
+ &mut to_bridge,
|
|
+ &mut ring,
|
|
+ &acp_tx,
|
|
+ &meta,
|
|
+ &events,
|
|
+ &settings,
|
|
+ )
|
|
+ .await;
|
|
+
|
|
+ match outcome {
|
|
+ PumpOutcome::Shutdown => break,
|
|
+ PumpOutcome::Disconnected(detail) => {
|
|
+ tracing::info!(%detail, "rc: glance link dropped");
|
|
+ // Back to the floor of the backoff curve: a link that
|
|
+ // worked until a second ago has earned a fast retry,
|
|
+ // whatever the attempt count was before it came up.
|
|
+ attempt = 1;
|
|
+ let _ = events.send(RcEvent::State(RcState::Reconnecting {
|
|
+ attempt,
|
|
+ detail,
|
|
+ }));
|
|
+ }
|
|
+ }
|
|
+ }
|
|
+ Err(DialError::Fatal(detail)) => {
|
|
+ // A rejected key or an unusable URL will be rejected identically
|
|
+ // forever. Retrying would only produce a status line that
|
|
+ // flickers between "reconnecting" and "failed" until the user
|
|
+ // gives up reading it.
|
|
+ tracing::warn!(%detail, "rc: giving up on glance");
|
|
+ let _ = events.send(RcEvent::State(RcState::Failed { detail }));
|
|
+ break;
|
|
+ }
|
|
+ Err(DialError::Retry(detail)) => {
|
|
+ attempt = attempt.saturating_add(1);
|
|
+ let _ = events.send(RcEvent::State(RcState::Reconnecting {
|
|
+ attempt,
|
|
+ detail,
|
|
+ }));
|
|
+ }
|
|
+ }
|
|
+
|
|
+ // Keep draining while disconnected: the tee has an unbounded channel and
|
|
+ // a session streaming through a long outage would otherwise grow it
|
|
+ // without limit. Frames land in the ring so the reconnect replays them.
|
|
+ match absorb(&mut to_bridge, &mut ring, backoff(attempt)).await {
|
|
+ AbsorbOutcome::Shutdown => break,
|
|
+ AbsorbOutcome::Elapsed => {}
|
|
+ }
|
|
+ }
|
|
+
|
|
+ let _ = events.send(RcEvent::State(RcState::Off));
|
|
+ tracing::info!("rc: bridge stopped");
|
|
+}
|
|
+
|
|
+fn backoff(attempt: u32) -> Duration {
|
|
+ let shift = attempt.saturating_sub(1).min(5);
|
|
+ (BACKOFF_MIN * 2u32.saturating_pow(shift)).min(BACKOFF_MAX)
|
|
+}
|
|
+
|
|
+enum DialError {
|
|
+ /// Worth retrying: server down, DNS blip, network change.
|
|
+ Retry(String),
|
|
+ /// Will fail the same way every time: bad URL, rejected credentials.
|
|
+ Fatal(String),
|
|
+}
|
|
+
|
|
+type Ws = tokio_tungstenite::WebSocketStream<
|
|
+ tokio_tungstenite::MaybeTlsStream<tokio::net::TcpStream>,
|
|
+>;
|
|
+
|
|
+async fn dial(settings: &RcSettings) -> Result<Ws, DialError> {
|
|
+ let mut request = settings
|
|
+ .url
|
|
+ .as_str()
|
|
+ .into_client_request()
|
|
+ .map_err(|e| DialError::Fatal(format!("bad remote_control.url: {e}")))?;
|
|
+
|
|
+ // ACP has no transport auth of its own, so the API key rides the upgrade.
|
|
+ request.headers_mut().insert(
|
|
+ "Authorization",
|
|
+ format!("Bearer {}", settings.api_key)
|
|
+ .parse()
|
|
+ .map_err(|_| DialError::Fatal("remote_control.api_key is not a valid header".into()))?,
|
|
+ );
|
|
+
|
|
+ let connect = tokio::time::timeout(CONNECT_TIMEOUT, tokio_tungstenite::connect_async(request));
|
|
+ match connect.await {
|
|
+ Err(_) => Err(DialError::Retry("connect timed out".into())),
|
|
+ Ok(Ok((ws, _response))) => Ok(ws),
|
|
+ Ok(Err(e)) => Err(classify_dial_error(e)),
|
|
+ }
|
|
+}
|
|
+
|
|
+/// An HTTP rejection at upgrade time is the server's verdict on our credentials
|
|
+/// or URL; anything else is a transport problem that may clear on its own.
|
|
+fn classify_dial_error(e: tokio_tungstenite::tungstenite::Error) -> DialError {
|
|
+ use tokio_tungstenite::tungstenite::Error as WsError;
|
|
+ match &e {
|
|
+ WsError::Http(response) => {
|
|
+ let status = response.status();
|
|
+ let detail = format!("glance rejected the connection: HTTP {status}");
|
|
+ if status.is_client_error() {
|
|
+ DialError::Fatal(detail)
|
|
+ } else {
|
|
+ DialError::Retry(detail)
|
|
+ }
|
|
+ }
|
|
+ WsError::Url(_) => DialError::Fatal(format!("bad remote_control.url: {e}")),
|
|
+ _ => DialError::Retry(e.to_string()),
|
|
+ }
|
|
+}
|
|
+
|
|
+/// Drain the tee into the ring for `budget`, so an outage costs history rather
|
|
+/// than memory. Returns early only for shutdown.
|
|
+async fn absorb(
|
|
+ to_bridge: &mut mpsc::UnboundedReceiver<ToBridge>,
|
|
+ ring: &mut ReplayRing,
|
|
+ budget: Duration,
|
|
+) -> AbsorbOutcome {
|
|
+ let deadline = tokio::time::Instant::now() + budget;
|
|
+ loop {
|
|
+ tokio::select! {
|
|
+ _ = tokio::time::sleep_until(deadline) => return AbsorbOutcome::Elapsed,
|
|
+ msg = to_bridge.recv() => match msg {
|
|
+ // Handle dropped: nothing left to mirror for.
|
|
+ None | Some(ToBridge::Shutdown) => return AbsorbOutcome::Shutdown,
|
|
+ Some(ToBridge::Frame(frame)) => ring.push(frame),
|
|
+ // Dropping `remote_tx` is the signal the race task waits on: it
|
|
+ // stops waiting for a glance that is not there and lets the
|
|
+ // terminal answer normally.
|
|
+ Some(ToBridge::Interaction(_)) | Some(ToBridge::InteractionSettled { .. }) => {}
|
|
+ },
|
|
+ }
|
|
+ }
|
|
+}
|
|
+
|
|
+#[allow(clippy::too_many_arguments)]
|
|
+async fn pump(
|
|
+ ws: Ws,
|
|
+ to_bridge: &mut mpsc::UnboundedReceiver<ToBridge>,
|
|
+ ring: &mut ReplayRing,
|
|
+ acp_tx: &AcpAgentTx,
|
|
+ meta: &SessionMeta,
|
|
+ events: &mpsc::UnboundedSender<RcEvent>,
|
|
+ settings: &RcSettings,
|
|
+) -> PumpOutcome {
|
|
+ let (mut sink, mut stream) = ws.split();
|
|
+
|
|
+ // A writer task, rather than sending inline: `session/prompt` blocks for the
|
|
+ // whole turn, so the handlers that need to reply must be able to run
|
|
+ // concurrently with the mirror. They share this channel instead of the sink.
|
|
+ let (out_tx, mut out_rx) = mpsc::unbounded_channel::<Message>();
|
|
+ let mut writer = tokio::spawn(async move {
|
|
+ while let Some(message) = out_rx.recv().await {
|
|
+ if let Err(e) = sink.send(message).await {
|
|
+ return e.to_string();
|
|
+ }
|
|
+ }
|
|
+ let _ = sink.close().await;
|
|
+ "closed".to_string()
|
|
+ });
|
|
+
|
|
+ // Catch a viewer up before anything new arrives: the ring is what a browser
|
|
+ // attaching mid-session (or after a glance restart) sees as history.
|
|
+ let _ = out_tx.send(status_frame(meta, ring));
|
|
+ for frame in ring.iter() {
|
|
+ if out_tx.send(Message::Text(frame.clone().into())).is_err() {
|
|
+ return PumpOutcome::Disconnected("writer stopped during replay".into());
|
|
+ }
|
|
+ }
|
|
+
|
|
+ let mut pending: HashMap<u64, Pending> = HashMap::new();
|
|
+
|
|
+ let outcome = loop {
|
|
+ tokio::select! {
|
|
+ biased;
|
|
+
|
|
+ // The socket died under the writer. Notice it here rather than
|
|
+ // waiting for the read half to time out.
|
|
+ writer_result = &mut writer => {
|
|
+ let detail = writer_result.unwrap_or_else(|e| format!("writer panicked: {e}"));
|
|
+ break PumpOutcome::Disconnected(detail);
|
|
+ }
|
|
+
|
|
+ inbound = stream.next() => match inbound {
|
|
+ None => break PumpOutcome::Disconnected("glance closed the connection".into()),
|
|
+ Some(Err(e)) => break PumpOutcome::Disconnected(e.to_string()),
|
|
+ Some(Ok(Message::Close(_))) => {
|
|
+ break PumpOutcome::Disconnected("glance closed the connection".into());
|
|
+ }
|
|
+ Some(Ok(Message::Text(text))) => {
|
|
+ handle_inbound(
|
|
+ &text, &mut pending, &out_tx, acp_tx, meta, events, settings, ring,
|
|
+ );
|
|
+ }
|
|
+ // Ping/pong are answered by tungstenite; binary frames are not
|
|
+ // part of this protocol.
|
|
+ Some(Ok(_)) => {}
|
|
+ },
|
|
+
|
|
+ outgoing = to_bridge.recv() => match outgoing {
|
|
+ None | Some(ToBridge::Shutdown) => break PumpOutcome::Shutdown,
|
|
+
|
|
+ Some(ToBridge::Frame(frame)) => {
|
|
+ ring.push(frame.clone());
|
|
+ if out_tx.send(Message::Text(frame.into())).is_err() {
|
|
+ break PumpOutcome::Disconnected("writer stopped".into());
|
|
+ }
|
|
+ }
|
|
+
|
|
+ Some(ToBridge::Interaction(interaction)) => {
|
|
+ let Interaction { id, method, params, tool_call_id, remote_tx } = interaction;
|
|
+ let frame = protocol::request(id, method, params);
|
|
+ pending.insert(id, Pending { remote_tx, tool_call_id });
|
|
+ if out_tx.send(Message::Text(frame.into())).is_err() {
|
|
+ break PumpOutcome::Disconnected("writer stopped".into());
|
|
+ }
|
|
+ }
|
|
+
|
|
+ // The terminal won the race. Removing the entry drops `remote_tx`,
|
|
+ // which is what makes a late answer from glance a silent no-op.
|
|
+ Some(ToBridge::InteractionSettled { id }) => {
|
|
+ if let Some(entry) = pending.remove(&id) {
|
|
+ let frame = protocol::notification(
|
|
+ protocol::M_RC_INTERACTION_CANCELLED,
|
|
+ serde_json::json!({ "id": id, "toolCallId": entry.tool_call_id }),
|
|
+ );
|
|
+ if out_tx.send(Message::Text(frame.into())).is_err() {
|
|
+ break PumpOutcome::Disconnected("writer stopped".into());
|
|
+ }
|
|
+ }
|
|
+ }
|
|
+ },
|
|
+ }
|
|
+ };
|
|
+
|
|
+ // Every still-pending interaction loses its remote half here, so each race
|
|
+ // task falls back to the terminal instead of waiting on a socket that is gone.
|
|
+ drop(pending);
|
|
+ drop(out_tx);
|
|
+ writer.abort();
|
|
+
|
|
+ outcome
|
|
+}
|
|
+
|
|
+fn status_frame(meta: &SessionMeta, ring: &ReplayRing) -> Message {
|
|
+ Message::Text(
|
|
+ protocol::notification(
|
|
+ protocol::M_RC_STATUS,
|
|
+ serde_json::json!({
|
|
+ "session": meta.to_json(),
|
|
+ "replay": { "frames": ring.len(), "dropped": ring.dropped() },
|
|
+ }),
|
|
+ )
|
|
+ .into(),
|
|
+ )
|
|
+}
|
|
+
|
|
+#[allow(clippy::too_many_arguments)]
|
|
+fn handle_inbound(
|
|
+ text: &str,
|
|
+ pending: &mut HashMap<u64, Pending>,
|
|
+ out_tx: &mpsc::UnboundedSender<Message>,
|
|
+ acp_tx: &AcpAgentTx,
|
|
+ meta: &SessionMeta,
|
|
+ events: &mpsc::UnboundedSender<RcEvent>,
|
|
+ settings: &RcSettings,
|
|
+ ring: &ReplayRing,
|
|
+) {
|
|
+ let frame: protocol::InboundFrame = match serde_json::from_str(text) {
|
|
+ Ok(frame) => frame,
|
|
+ Err(e) => {
|
|
+ tracing::warn!(error = %e, "rc: unparseable frame from glance");
|
|
+ return;
|
|
+ }
|
|
+ };
|
|
+
|
|
+ if frame.is_response() {
|
|
+ resolve_pending(frame, pending);
|
|
+ return;
|
|
+ }
|
|
+
|
|
+ let Some(method) = frame.method.clone() else {
|
|
+ tracing::warn!("rc: frame from glance has neither method nor result");
|
|
+ return;
|
|
+ };
|
|
+ let params = frame.params.clone().unwrap_or(serde_json::Value::Null);
|
|
+
|
|
+ // `id` present => a request expecting a reply; absent => a notification.
|
|
+ let reply_id = frame.id.clone();
|
|
+ let reply = |result: Result<serde_json::Value, (i32, String)>| {
|
|
+ let Some(id) = reply_id.clone() else { return };
|
|
+ let frame = match result {
|
|
+ Ok(value) => protocol::response(id, value),
|
|
+ Err((code, message)) => protocol::error_response(id, code, message),
|
|
+ };
|
|
+ let _ = out_tx.send(Message::Text(frame.into()));
|
|
+ };
|
|
+
|
|
+ match method.as_str() {
|
|
+ protocol::M_INITIALIZE => reply(Ok(serde_json::json!({
|
|
+ "protocolVersion": 1,
|
|
+ "agentCapabilities": {
|
|
+ // This bridge mirrors one existing session. It cannot create or
|
|
+ // load others, and says so rather than failing the calls later.
|
|
+ "loadSession": false,
|
|
+ "promptCapabilities": { "image": false, "audio": false, "embeddedContext": false },
|
|
+ },
|
|
+ "_meta": {
|
|
+ "session": meta.to_json(),
|
|
+ "remoteControl": {
|
|
+ "replayBuffer": settings.replay_buffer,
|
|
+ "frames": ring.len(),
|
|
+ "dropped": ring.dropped(),
|
|
+ },
|
|
+ },
|
|
+ }))),
|
|
+
|
|
+ protocol::M_SESSION_LIST => reply(Ok(serde_json::json!({
|
|
+ "sessions": meta.session_id.as_ref().map(|_| vec![meta.to_json()]).unwrap_or_default(),
|
|
+ }))),
|
|
+
|
|
+ protocol::M_RC_REPLAY => {
|
|
+ for frame in ring.iter() {
|
|
+ if out_tx.send(Message::Text(frame.clone().into())).is_err() {
|
|
+ break;
|
|
+ }
|
|
+ }
|
|
+ reply(Ok(serde_json::json!({
|
|
+ "frames": ring.len(),
|
|
+ "dropped": ring.dropped(),
|
|
+ })));
|
|
+ }
|
|
+
|
|
+ protocol::M_SESSION_PROMPT => {
|
|
+ let Some(session_id) = meta.session_id.clone() else {
|
|
+ reply(Err((
|
|
+ protocol::E_INTERNAL,
|
|
+ "no session is attached to this bridge".into(),
|
|
+ )));
|
|
+ return;
|
|
+ };
|
|
+ let blocks = match prompt_blocks(¶ms) {
|
|
+ Ok(blocks) => blocks,
|
|
+ Err(e) => {
|
|
+ reply(Err((protocol::E_INVALID_PARAMS, e)));
|
|
+ return;
|
|
+ }
|
|
+ };
|
|
+ spawn_prompt(session_id, blocks, acp_tx.clone(), out_tx.clone(), reply_id, events.clone());
|
|
+ }
|
|
+
|
|
+ protocol::M_SESSION_CANCEL => {
|
|
+ let Some(session_id) = meta.session_id.clone() else {
|
|
+ reply(Err((
|
|
+ protocol::E_INTERNAL,
|
|
+ "no session is attached to this bridge".into(),
|
|
+ )));
|
|
+ return;
|
|
+ };
|
|
+ spawn_cancel(session_id, acp_tx.clone(), events.clone());
|
|
+ reply(Ok(serde_json::json!({})));
|
|
+ }
|
|
+
|
|
+ // Glance telling us how many browsers are watching. Purely cosmetic --
|
|
+ // it drives the `/rc status` line, nothing load-bearing.
|
|
+ "x.ai/rc/viewers" => {
|
|
+ let viewers = params.get("count").and_then(|v| v.as_u64()).map(|n| n as usize);
|
|
+ let _ = events.send(RcEvent::State(RcState::Connected { viewers }));
|
|
+ }
|
|
+
|
|
+ unknown => {
|
|
+ // A real `method not found`, never a faked success: glance has to be
|
|
+ // able to tell what this bridge actually implements.
|
|
+ tracing::debug!(%unknown, "rc: unsupported method from glance");
|
|
+ reply(Err((
|
|
+ protocol::E_METHOD_NOT_FOUND,
|
|
+ format!("Method not found: {unknown}"),
|
|
+ )));
|
|
+ }
|
|
+ }
|
|
+}
|
|
+
|
|
+/// Route a JSON-RPC response back to the race task waiting on it.
|
|
+fn resolve_pending(frame: protocol::InboundFrame, pending: &mut HashMap<u64, Pending>) {
|
|
+ let Some(id) = frame.id.as_ref().and_then(serde_json::Value::as_u64) else {
|
|
+ tracing::warn!("rc: response from glance with no usable id");
|
|
+ return;
|
|
+ };
|
|
+ let Some(entry) = pending.remove(&id) else {
|
|
+ // Expected whenever the terminal won: we already retracted the dialog
|
|
+ // and dropped the entry, and glance's answer lost the race.
|
|
+ tracing::debug!(id, "rc: answer for an interaction already settled");
|
|
+ return;
|
|
+ };
|
|
+ match frame.result {
|
|
+ Some(result) => {
|
|
+ let _ = entry.remote_tx.send(result);
|
|
+ }
|
|
+ None => {
|
|
+ // Glance declined. Dropping `remote_tx` hands the interaction back
|
|
+ // to the terminal, which is still showing the modal.
|
|
+ tracing::debug!(id, error = ?frame.error, "rc: glance declined an interaction");
|
|
+ }
|
|
+ }
|
|
+}
|
|
+
|
|
+/// Accept either ACP-standard `prompt: ContentBlock[]` or a plain `text`.
|
|
+fn prompt_blocks(params: &serde_json::Value) -> Result<Vec<acp::ContentBlock>, String> {
|
|
+ if let Some(prompt) = params.get("prompt") {
|
|
+ return serde_json::from_value(prompt.clone())
|
|
+ .map_err(|e| format!("invalid prompt blocks: {e}"));
|
|
+ }
|
|
+ let text = params
|
|
+ .get("text")
|
|
+ .and_then(|v| v.as_str())
|
|
+ .ok_or_else(|| "expected `prompt` (ContentBlock[]) or `text` (string)".to_string())?;
|
|
+ if text.trim().is_empty() {
|
|
+ return Err("prompt text is empty".into());
|
|
+ }
|
|
+ serde_json::from_value(serde_json::json!([{ "type": "text", "text": text }]))
|
|
+ .map_err(|e| format!("could not build text block: {e}"))
|
|
+}
|
|
+
|
|
+/// A prompt blocks for the whole turn, so it runs in its own task and replies
|
|
+/// when the turn ends.
|
|
+fn spawn_prompt(
|
|
+ session_id: String,
|
|
+ blocks: Vec<acp::ContentBlock>,
|
|
+ acp_tx: AcpAgentTx,
|
|
+ out_tx: mpsc::UnboundedSender<Message>,
|
|
+ reply_id: Option<serde_json::Value>,
|
|
+ events: mpsc::UnboundedSender<RcEvent>,
|
|
+) {
|
|
+ tokio::spawn(async move {
|
|
+ // `clientIdentifier` is what makes the turn attributable to glance in
|
|
+ // the session log, and what tells the TUI this is another client's turn
|
|
+ // to view rather than one of its own to drive.
|
|
+ let prompt_id = format!("glance-{}", uuid::Uuid::new_v4());
|
|
+ let meta = serde_json::json!({
|
|
+ "clientIdentifier": RC_CLIENT_ID,
|
|
+ "promptId": prompt_id,
|
|
+ });
|
|
+ let request = acp::PromptRequest::new(acp::SessionId::new(session_id), blocks)
|
|
+ .meta(meta.as_object().cloned());
|
|
+
|
|
+ let _ = events.send(RcEvent::Notice("prompt received from glance".into()));
|
|
+ let result = acp_send(request, &acp_tx).await;
|
|
+
|
|
+ let Some(id) = reply_id else { return };
|
|
+ let frame = match result {
|
|
+ Ok(response) => match serde_json::to_value(&response) {
|
|
+ Ok(value) => protocol::response(id, value),
|
|
+ Err(e) => protocol::error_response(id, protocol::E_INTERNAL, e.to_string()),
|
|
+ },
|
|
+ Err(e) => protocol::error_response(id, protocol::E_INTERNAL, e.to_string()),
|
|
+ };
|
|
+ let _ = out_tx.send(Message::Text(frame.into()));
|
|
+ });
|
|
+}
|
|
+
|
|
+fn spawn_cancel(session_id: String, acp_tx: AcpAgentTx, events: mpsc::UnboundedSender<RcEvent>) {
|
|
+ tokio::spawn(async move {
|
|
+ // `CancelTrigger::from_client("glance")` classifies as `StopGesture` --
|
|
+ // the same class as Esc, so the runtime treats it identically -- while
|
|
+ // telemetry still records who interrupted. Sending `esc` would work and
|
|
+ // would lie.
|
|
+ let meta = serde_json::json!({
|
|
+ "cancelTrigger": RC_CLIENT_ID,
|
|
+ "cancelSubagents": true,
|
|
+ });
|
|
+ let request = acp::CancelNotification::new(acp::SessionId::new(session_id))
|
|
+ .meta(meta.as_object().cloned());
|
|
+
|
|
+ match acp_send(request, &acp_tx).await {
|
|
+ Ok(()) => {
|
|
+ let _ = events.send(RcEvent::Notice("turn interrupted from glance".into()));
|
|
+ }
|
|
+ Err(e) => {
|
|
+ tracing::warn!(error = %e, "rc: remote cancel failed");
|
|
+ }
|
|
+ }
|
|
+ });
|
|
+}
|
|
+
|
|
+#[cfg(test)]
|
|
+mod tests {
|
|
+ use super::*;
|
|
+
|
|
+ #[test]
|
|
+ fn backoff_grows_then_caps() {
|
|
+ assert_eq!(backoff(1), BACKOFF_MIN);
|
|
+ assert_eq!(backoff(2), Duration::from_secs(2));
|
|
+ assert_eq!(backoff(4), Duration::from_secs(8));
|
|
+ // Capped, so a long outage does not turn into a multi-minute blind spot.
|
|
+ assert_eq!(backoff(20), BACKOFF_MAX);
|
|
+ }
|
|
+
|
|
+ #[test]
|
|
+ fn backoff_handles_attempt_zero() {
|
|
+ // `attempt` is 0 on the very first dial; the shift must not underflow.
|
|
+ assert_eq!(backoff(0), BACKOFF_MIN);
|
|
+ }
|
|
+
|
|
+ #[test]
|
|
+ fn prompt_accepts_text_shorthand_and_acp_blocks() {
|
|
+ let from_text = prompt_blocks(&serde_json::json!({ "text": "hello" })).unwrap();
|
|
+ assert_eq!(from_text.len(), 1);
|
|
+
|
|
+ let from_blocks = prompt_blocks(&serde_json::json!({
|
|
+ "prompt": [{ "type": "text", "text": "hi" }]
|
|
+ }))
|
|
+ .unwrap();
|
|
+ assert_eq!(from_blocks.len(), 1);
|
|
+ }
|
|
+
|
|
+ #[test]
|
|
+ fn prompt_rejects_empty_and_missing() {
|
|
+ assert!(prompt_blocks(&serde_json::json!({})).is_err());
|
|
+ assert!(prompt_blocks(&serde_json::json!({ "text": " " })).is_err());
|
|
+ }
|
|
+
|
|
+ #[test]
|
|
+ fn a_settled_interaction_ignores_a_late_answer() {
|
|
+ let mut pending: HashMap<u64, Pending> = HashMap::new();
|
|
+ let (tx, mut rx) = oneshot::channel();
|
|
+ pending.insert(1, Pending { remote_tx: tx, tool_call_id: None });
|
|
+
|
|
+ // The terminal won: the entry is removed, dropping the sender.
|
|
+ pending.remove(&1);
|
|
+
|
|
+ // Glance's late answer finds nothing to resolve, and the race task sees
|
|
+ // a closed channel rather than a second answer.
|
|
+ let frame: protocol::InboundFrame =
|
|
+ serde_json::from_str(r#"{"jsonrpc":"2.0","id":1,"result":{"outcome":"cancelled"}}"#)
|
|
+ .unwrap();
|
|
+ resolve_pending(frame, &mut pending);
|
|
+ assert!(rx.try_recv().is_err());
|
|
+ }
|
|
+
|
|
+ #[test]
|
|
+ fn a_declined_interaction_falls_back_to_the_terminal() {
|
|
+ let mut pending: HashMap<u64, Pending> = HashMap::new();
|
|
+ let (tx, mut rx) = oneshot::channel();
|
|
+ pending.insert(2, Pending { remote_tx: tx, tool_call_id: None });
|
|
+
|
|
+ let frame: protocol::InboundFrame = serde_json::from_str(
|
|
+ r#"{"jsonrpc":"2.0","id":2,"error":{"code":-32603,"message":"no viewer"}}"#,
|
|
+ )
|
|
+ .unwrap();
|
|
+ resolve_pending(frame, &mut pending);
|
|
+
|
|
+ // No value delivered, and the channel is closed -- which is exactly what
|
|
+ // makes the race task wait on the terminal instead.
|
|
+ assert!(rx.try_recv().is_err());
|
|
+ assert!(pending.is_empty());
|
|
+ }
|
|
+}
|
|
diff --git a/crates/codegen/xai-grok-pager/src/slash/commands/mod.rs b/crates/codegen/xai-grok-pager/src/slash/commands/mod.rs
|
|
index a838cc6d558a032897f0e112c2591dec56b18fec..8ca138f738c4b9a5bb03cff2ca5532b7e656283d 100644
|
|
--- a/crates/codegen/xai-grok-pager/src/slash/commands/mod.rs
|
|
+++ b/crates/codegen/xai-grok-pager/src/slash/commands/mod.rs
|
|
@@ -47,6 +47,7 @@ pub mod plan;
|
|
pub mod plugin;
|
|
pub mod privacy;
|
|
pub mod queue;
|
|
+pub mod rc;
|
|
pub mod recap;
|
|
pub mod release_notes;
|
|
pub mod remember;
|
|
@@ -142,6 +143,7 @@ pub fn builtin_commands() -> Vec<Arc<dyn SlashCommand>> {
|
|
Arc::new(usage::UsageCommand),
|
|
Arc::new(queue::QueueCommand),
|
|
Arc::new(tasks::TasksCommand),
|
|
+ Arc::new(rc::RcCommand),
|
|
Arc::new(release_notes::ReleaseNotesCommand),
|
|
Arc::new(tutorial::TutorialCommand),
|
|
Arc::new(config_agents::ConfigAgentsCommand),
|
|
diff --git a/crates/codegen/xai-grok-pager/src/slash/commands/rc.rs b/crates/codegen/xai-grok-pager/src/slash/commands/rc.rs
|
|
new file mode 100644
|
|
index 0000000000000000000000000000000000000000..41cb21d2d0cc3c3e56f7a7fc75488c83d60b0c2e
|
|
--- /dev/null
|
|
+++ b/crates/codegen/xai-grok-pager/src/slash/commands/rc.rs
|
|
@@ -0,0 +1,108 @@
|
|
+//! `/rc` -- remote control: mirror this session to a grok-glance server.
|
|
+//!
|
|
+//! The point of `/rc` is what it does *not* do. It does not restart the session,
|
|
+//! switch it to headless, hand it to another process, or degrade the terminal in
|
|
+//! any way. The TUI stays exactly as interactive as it was; glance simply
|
|
+//! becomes a second view of the same session, reachable from a browser.
|
|
+//!
|
|
+//! From either side you can watch the turn stream, send a prompt, interrupt a
|
|
+//! running turn, and answer tool-permission requests, questions, and plan
|
|
+//! approvals. Both sides can answer -- whoever answers first wins, and the other
|
|
+//! side's dialog closes by itself.
|
|
+//!
|
|
+//! Configuration lives in `[remote_control]` (`url`, `api_key`), with
|
|
+//! `GROK_RC_URL` / `GROK_RC_API_KEY` overriding it -- a bearer token in a
|
|
+//! plaintext config file is a poor default.
|
|
+//!
|
|
+//! **Scope.** This session only, and only until you quit. Nothing is persisted.
|
|
+
|
|
+use crate::app::actions::{Action, RemoteControlAction};
|
|
+use crate::slash::command::{CommandExecCtx, CommandResult, SlashCommand};
|
|
+
|
|
+/// Remote control toggle via `/rc`.
|
|
+pub struct RcCommand;
|
|
+
|
|
+impl SlashCommand for RcCommand {
|
|
+ fn name(&self) -> &str {
|
|
+ "rc"
|
|
+ }
|
|
+
|
|
+ fn description(&self) -> &str {
|
|
+ "Remote control: mirror this session to grok-glance"
|
|
+ }
|
|
+
|
|
+ fn usage(&self) -> &str {
|
|
+ "/rc [on|off|status]"
|
|
+ }
|
|
+
|
|
+ fn takes_args(&self) -> bool {
|
|
+ true
|
|
+ }
|
|
+
|
|
+ /// Bare `/rc` toggles, which is the common case.
|
|
+ fn args_required(&self) -> bool {
|
|
+ false
|
|
+ }
|
|
+
|
|
+ fn arg_placeholder(&self) -> Option<&str> {
|
|
+ Some("on|off|status")
|
|
+ }
|
|
+
|
|
+ /// There is nothing to mirror without a session.
|
|
+ fn session_scoped(&self) -> bool {
|
|
+ true
|
|
+ }
|
|
+
|
|
+ fn run(&self, ctx: &mut CommandExecCtx, args: &str) -> CommandResult {
|
|
+ if ctx.session_id.is_none() {
|
|
+ return CommandResult::Error("No active session".to_string());
|
|
+ }
|
|
+ match parse(args) {
|
|
+ Some(what) => CommandResult::Action(Action::RemoteControl(what)),
|
|
+ None => CommandResult::Error(format!(
|
|
+ "Unknown argument {:?}. Usage: /rc [on|off|status]",
|
|
+ args.trim()
|
|
+ )),
|
|
+ }
|
|
+ }
|
|
+}
|
|
+
|
|
+/// `stop` and `start` are accepted alongside `off`/`on` because both are
|
|
+/// plausible guesses, and being told "unknown argument" for a synonym of the
|
|
+/// thing you meant is a poor trade for strictness.
|
|
+fn parse(args: &str) -> Option<RemoteControlAction> {
|
|
+ match args.trim().to_ascii_lowercase().as_str() {
|
|
+ "" => Some(RemoteControlAction::Toggle),
|
|
+ "on" | "start" | "connect" => Some(RemoteControlAction::On),
|
|
+ "off" | "stop" | "disconnect" => Some(RemoteControlAction::Off),
|
|
+ "status" | "info" => Some(RemoteControlAction::Status),
|
|
+ _ => None,
|
|
+ }
|
|
+}
|
|
+
|
|
+#[cfg(test)]
|
|
+mod tests {
|
|
+ use super::*;
|
|
+
|
|
+ #[test]
|
|
+ fn bare_rc_toggles() {
|
|
+ assert_eq!(parse(""), Some(RemoteControlAction::Toggle));
|
|
+ assert_eq!(parse(" "), Some(RemoteControlAction::Toggle));
|
|
+ }
|
|
+
|
|
+ #[test]
|
|
+ fn subcommands_are_case_insensitive_and_accept_synonyms() {
|
|
+ assert_eq!(parse("OFF"), Some(RemoteControlAction::Off));
|
|
+ assert_eq!(parse(" stop "), Some(RemoteControlAction::Off));
|
|
+ assert_eq!(parse("On"), Some(RemoteControlAction::On));
|
|
+ assert_eq!(parse("status"), Some(RemoteControlAction::Status));
|
|
+ }
|
|
+
|
|
+ /// A typo must not silently toggle: `/rc of` turning remote control *on*
|
|
+ /// would be the worst possible reading of the user's intent.
|
|
+ #[test]
|
|
+ fn an_unknown_argument_is_an_error_not_a_toggle() {
|
|
+ assert_eq!(parse("of"), None);
|
|
+ assert_eq!(parse("nonsense"), None);
|
|
+ }
|
|
+}
|
|
diff --git a/crates/codegen/xai-grok-shell/src/session/slash_commands.rs b/crates/codegen/xai-grok-shell/src/session/slash_commands.rs
|
|
index 3f017f2da6765fc680fbe431d0e776fdd5b2122f..f93c675530b0c72d0e611faeddf617decb39dc4d 100644
|
|
--- a/crates/codegen/xai-grok-shell/src/session/slash_commands.rs
|
|
+++ b/crates/codegen/xai-grok-shell/src/session/slash_commands.rs
|
|
@@ -495,6 +495,7 @@ pub const PAGER_COMMAND_KEYS: &[&str] = &[
|
|
"privacy",
|
|
"queue",
|
|
"quit",
|
|
+ "rc",
|
|
"recap",
|
|
"release-notes",
|
|
"reload-plugins",
|