From 7a4d74afdefb5dfdef9fd7f0ee4bebf6e0b023e5 Mon Sep 17 00:00:00 2001 From: Ben Reilly Date: Tue, 26 May 2026 12:44:24 -0400 Subject: [PATCH] control plane: `inspect` / `tail` / `open` from a second terminal MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A running `gamechat --realtime` process now binds a tiny Unix domain socket and exposes JSON-over-newline RPCs so a second `gamechat` invocation can query active sub-agents and stream their progress. Subcommands: gamechat inspect # table of slug, provider, status, last msg gamechat tail # cursor-based 750ms polling stream gamechat open # print `claude --resume ` gamechat open --launch # macOS: open Terminal via osascript `open` is claude-only — codex has no documented resume flag, so we error with a clear pointer to `tail` for codex slugs. Multiple gamechat processes? Pass `--pid ` or `--socket `. The new `src/control/` module is fully isolated: it depends only on a read-only handle to `ProgressStore` and never touches the voice loop or the orchestrator. Plumbing changes to make this possible: - `ProgressStore` gained `session_id` per slug, a sequence counter on buffered entries, `snapshot_all()` (non-rate-limited), and `entries_after(slug, cursor)` for cursor-based tail - `ProgressReporter::set_session_id()`; `ClaudeClient` calls it once it parses the session id from `claude -p` stream-json - `OrchestratorJobManager::progress_store()` accessor + register slugs with the real provider name instead of the hardcoded "orchestrator" - `voice_loop` spawns the control server best-effort at startup 41 tests pass, including 5 new in `control::server` (one end-to-end socket roundtrip) and 4 in `control::client::resume_command`. Co-Authored-By: Claude Opus 4.7 (1M context) --- README.md | 46 +++- src/control/client.rs | 281 +++++++++++++++++++++++ src/control/mod.rs | 50 ++++ src/control/protocol.rs | 53 +++++ src/control/runtime_dir.rs | 85 +++++++ src/control/server.rs | 364 ++++++++++++++++++++++++++++++ src/main.rs | 87 +++++++ src/orchestrator/claude/client.rs | 3 + src/orchestrator/interface.rs | 4 + src/orchestrator/jobs.rs | 11 +- src/orchestrator/progress.rs | 178 ++++++++++++++- src/voice_loop/mod.rs | 14 ++ 12 files changed, 1163 insertions(+), 13 deletions(-) create mode 100644 src/control/client.rs create mode 100644 src/control/mod.rs create mode 100644 src/control/protocol.rs create mode 100644 src/control/runtime_dir.rs create mode 100644 src/control/server.rs diff --git a/README.md b/README.md index 0715775..7ec25b0 100644 --- a/README.md +++ b/README.md @@ -8,8 +8,9 @@ Voice-driven supervisor for [Claude Code] and [Codex]. You talk to a low-latency │ tool: delegate_to_orchestrator(slug, intent, …) │ tool: sub_agent_progress(slug) ▼ - OrchestratorJobManager - │ one worker task per slug, ordered within a slug, + OrchestratorJobManager ◀── unix socket ── gamechat inspect / tail / open + │ one worker task per slug, (other terminal) + │ ordered within a slug, │ concurrent across slugs ▼ ┌───────┴────────┐ @@ -19,7 +20,7 @@ Voice-driven supervisor for [Claude Code] and [Codex]. You talk to a low-latency ## Architecture -There is exactly **one realtime voice loop** and an **async worker pool** for background agent jobs. Those are the two halves of the binary. +There is exactly **one realtime voice loop**, an **async worker pool** for background agent jobs, and a tiny **control plane** that exposes job state to other terminals. Three concerns, three modules. ### The voice loop (`src/voice_loop/`) @@ -44,6 +45,10 @@ On startup it sends a `session.update` that registers two tools — `delegate_to Adding a third backend means implementing `Provider` and wiring one match arm in `main.rs`. The voice loop doesn't know which agent is on the other end. +### The control plane (`src/control/`) + +The realtime process binds a Unix domain socket at startup and serves a tiny JSON-over-newline protocol. A second `gamechat` invocation reconnects to that socket for `inspect`, `tail`, or `open`. The control module never touches the voice loop or the orchestrator internals — it depends only on `Arc`, so the live conversation is unaffected by inspection. See **Inspecting active sub-agents** below for the user-facing surface. + ## Install ```sh @@ -102,6 +107,14 @@ gamechat --once "summarize the last 5 commits" --slug summarize_commits # Dump the session.update JSON without connecting. gamechat --print-realtime-config + +# Drop a personality on top of the orchestrator instructions. +gamechat --realtime --preset jarvis + +# From another terminal, while a realtime session is running: +gamechat inspect # list every active sub-agent +gamechat tail refactor_docs # stream that slug's progress +gamechat open refactor_docs # print `claude --resume ` for that slug ``` | Flag | Default | Description | @@ -162,6 +175,27 @@ For something durable, write `~/.config/gamechat/settings.json`: Resolution order, highest to lowest: CLI flag → top-level `voice` / `persona` in settings → preset's value → built-in default. Custom presets override built-ins of the same name. The orchestrator-delegation instructions are always sent — personas are appended to that base, never replace it. Run `gamechat --preset jarvis --print-realtime-config` to see the exact `instructions` and `voice` that will be sent. +## Inspecting active sub-agents + +While a `gamechat --realtime` session is running, open another terminal and ask it what's going on: + +```sh +gamechat inspect # list every active sub-agent + status +gamechat tail refactor_docs # stream that slug's progress buffer +gamechat open refactor_docs # print `claude --resume ` +gamechat open refactor_docs --launch # macOS: pop a new Terminal window into Claude Code +``` + +| Subcommand | Purpose | +|------------|---------| +| `inspect` | Table view of every registered slug: provider, status, elapsed seconds, last message. | +| `tail ` | Polls the progress buffer at ~750 ms and prints any new entries until the job finishes. | +| `open ` | Prints `claude --resume ` for that slug. With `--launch` on macOS, pops the command into a fresh Terminal window via `osascript`. | + +Each subcommand accepts `--pid ` or `--socket ` if multiple gamechat instances are running. `open` only works against the `claude` provider — codex has no documented resume flag, so `open ` errors out and points you at `tail` instead. + +Under the hood, the realtime process binds a Unix domain socket at `$TMPDIR/gamechat-$USER/$pid.sock` (or `$XDG_RUNTIME_DIR` on Linux) and serves a tiny JSON-over-newline protocol (one request per connection). The socket is auto-removed when gamechat exits cleanly; stale entries from crashes are GC'd on the next `inspect` based on `kill(pid, 0)` probes. All of this lives in `src/control/` — entirely separate from the voice loop and orchestrator, depending only on a read-only handle to `ProgressStore`. + ## Repository layout ``` @@ -174,6 +208,12 @@ gamechat/ │ │ ├── session.rs # session.update JSON + tool defs │ │ ├── settings.rs # voice/persona presets, settings.json loader │ │ └── audio.rs # cpal input/output, resampling +│ ├── control/ # `inspect`/`tail`/`open` — isolated control plane +│ │ ├── mod.rs # subcommand routing +│ │ ├── protocol.rs # JSON-over-UDS wire types +│ │ ├── runtime_dir.rs # socket directory discovery +│ │ ├── server.rs # in-process Unix socket listener +│ │ └── client.rs # subcommand implementations │ └── orchestrator/ │ ├── interface.rs # Provider / Session traits, public types │ ├── jobs.rs # OrchestratorJobManager + per-slug workers diff --git a/src/control/client.rs b/src/control/client.rs new file mode 100644 index 0000000..30e076c --- /dev/null +++ b/src/control/client.rs @@ -0,0 +1,281 @@ +//! Client side of the control protocol — used by the `inspect`, `tail`, and +//! `open` subcommands. Connects to a Unix socket and prints results. + +use super::protocol::{Request, Response}; +use super::runtime_dir::{discover_sockets, socket_path_for_pid}; +use super::{ControlSubcommand, ControlTarget}; +use crate::orchestrator::progress::JobStatus; +use std::path::{Path, PathBuf}; +use std::time::Duration; +use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; +use tokio::net::UnixStream; +use tokio::time::sleep; + +pub(super) async fn run( + subcommand: ControlSubcommand, + target: ControlTarget, +) -> Result<(), String> { + let socket = resolve_socket(&target)?; + match subcommand { + ControlSubcommand::Inspect => run_inspect(&socket).await, + ControlSubcommand::Tail { slug } => run_tail(&socket, &slug).await, + ControlSubcommand::Open { slug, launch } => run_open(&socket, &slug, launch).await, + } +} + +fn resolve_socket(target: &ControlTarget) -> Result { + if let Some(path) = target.socket.as_ref() { + return Ok(path.clone()); + } + if let Some(pid) = target.pid { + return socket_path_for_pid(pid); + } + let sockets = discover_sockets()?; + match sockets.len() { + 0 => Err( + "no running gamechat instance found. Start one with `gamechat --realtime`, or pass --pid .".to_string(), + ), + 1 => Ok(sockets.into_iter().next().unwrap()), + _ => { + let names: Vec = sockets + .iter() + .filter_map(|p| p.file_stem().and_then(|s| s.to_str()).map(|s| s.to_string())) + .collect(); + Err(format!( + "multiple gamechat instances found (pids: {}). Pass --pid to choose one.", + names.join(", ") + )) + } + } +} + +async fn send_request(socket: &Path, request: Request) -> Result { + let stream = UnixStream::connect(socket) + .await + .map_err(|e| format!("failed to connect to {}: {e}", socket.display()))?; + let (read_half, mut write_half) = stream.into_split(); + let mut payload = serde_json::to_string(&request) + .map_err(|e| format!("failed to serialize request: {e}"))?; + payload.push('\n'); + write_half + .write_all(payload.as_bytes()) + .await + .map_err(|e| format!("failed to send request: {e}"))?; + write_half + .shutdown() + .await + .map_err(|e| format!("failed to half-close request: {e}"))?; + + let mut reader = BufReader::new(read_half); + let mut line = String::new(); + reader + .read_line(&mut line) + .await + .map_err(|e| format!("failed to read response: {e}"))?; + serde_json::from_str::(line.trim()) + .map_err(|e| format!("invalid response: {e}: {line}")) +} + +async fn run_inspect(socket: &Path) -> Result<(), String> { + let hello = send_request(socket, Request::Hello).await?; + if let Response::Hello { pid, version, provider } = &hello { + println!("gamechat pid={pid} version={version} provider={provider}"); + println!("socket {}", socket.display()); + println!(); + } + let list = send_request(socket, Request::List).await?; + match list { + Response::List { slugs } => { + if slugs.is_empty() { + println!("(no active sub-agents)"); + return Ok(()); + } + println!( + "{:<24} {:<10} {:<9} {:>9} {}", + "SLUG", "PROVIDER", "STATUS", "ELAPSED", "LAST" + ); + for s in slugs { + let last = truncate_for_table(&s.last_message, 60); + println!( + "{:<24} {:<10} {:<9} {:>8}s {}", + truncate_for_table(&s.slug, 24), + truncate_for_table(&s.provider, 10), + status_label(s.status), + s.elapsed_seconds.round() as u64, + last + ); + } + } + Response::Error { message } => return Err(message), + other => return Err(format!("unexpected response: {other:?}")), + } + Ok(()) +} + +async fn run_tail(socket: &Path, slug: &str) -> Result<(), String> { + let mut cursor: Option = None; + let mut first_pass = true; + loop { + let resp = send_request( + socket, + Request::Tail { + slug: slug.to_string(), + after_seq: cursor, + }, + ) + .await?; + match resp { + Response::Tail { + entries, + next_cursor, + status, + done, + } => { + if first_pass && entries.is_empty() { + println!("(no buffered entries yet — waiting for activity, Ctrl-C to stop)"); + first_pass = false; + } + for entry in &entries { + println!("{entry}"); + } + if !entries.is_empty() { + first_pass = false; + } + cursor = next_cursor; + if done { + eprintln!("-- {} {} --", slug, status_label(status)); + return Ok(()); + } + } + Response::Error { message } => return Err(message), + other => return Err(format!("unexpected response: {other:?}")), + } + sleep(Duration::from_millis(750)).await; + } +} + +async fn run_open(socket: &Path, slug: &str, launch: bool) -> Result<(), String> { + let resp = send_request( + socket, + Request::Resume { + slug: slug.to_string(), + }, + ) + .await?; + let (provider, session_id) = match resp { + Response::Resume { + provider, + session_id, + .. + } => (provider, session_id), + Response::Error { message } => return Err(message), + other => return Err(format!("unexpected response: {other:?}")), + }; + let cmd = resume_command(&provider, session_id.as_deref(), slug)?; + if launch { + launch_in_terminal(&cmd)?; + println!("launched: {cmd}"); + } else { + println!("{cmd}"); + println!("(run this in a terminal, or re-run with --launch on macOS)"); + } + Ok(()) +} + +fn resume_command( + provider: &str, + session_id: Option<&str>, + slug: &str, +) -> Result { + match provider { + "claude" => { + let session_id = session_id.ok_or_else(|| { + format!( + "no resume id for slug {slug:?} yet — the claude agent hasn't replied once. Wait for it to make progress, then retry." + ) + })?; + Ok(format!("claude --resume {session_id}")) + } + "codex" => Err(format!( + "`open` is not supported for codex slugs: the codex CLI has no documented resume flag, so there's no UI to reopen. Use `gamechat tail {slug}` to follow its progress instead." + )), + other => Err(format!( + "`open` is not supported for provider {other:?}. Only claude exposes a resume command." + )), + } +} + +#[cfg(target_os = "macos")] +fn launch_in_terminal(cmd: &str) -> Result<(), String> { + // `osascript` is the most reliable way to open a new Terminal window with + // a pre-typed command on macOS. + let script = format!( + "tell application \"Terminal\" to do script \"{}\"", + cmd.replace('\\', "\\\\").replace('"', "\\\"") + ); + let status = std::process::Command::new("osascript") + .arg("-e") + .arg(&script) + .status() + .map_err(|e| format!("failed to invoke osascript: {e}"))?; + if !status.success() { + return Err(format!("osascript exited with {status}")); + } + Ok(()) +} + +#[cfg(not(target_os = "macos"))] +fn launch_in_terminal(_cmd: &str) -> Result<(), String> { + Err("--launch is only implemented on macOS; copy the printed command instead".to_string()) +} + +fn status_label(status: JobStatus) -> &'static str { + match status { + JobStatus::Queued => "queued", + JobStatus::Running => "running", + JobStatus::Completed => "done", + JobStatus::Failed => "failed", + } +} + +fn truncate_for_table(s: &str, max: usize) -> String { + let mut chars = s.chars(); + let head: String = chars.by_ref().take(max).collect(); + if chars.next().is_some() { + format!("{}…", &head[..head.len().saturating_sub(1)]) + } else { + head + } +} + +#[cfg(test)] +mod tests { + use super::resume_command; + + #[test] + fn claude_with_session_id_returns_resume_command() { + let cmd = resume_command("claude", Some("sess-xyz"), "refactor_docs").unwrap(); + assert_eq!(cmd, "claude --resume sess-xyz"); + } + + #[test] + fn claude_without_session_id_errors_clearly() { + let err = resume_command("claude", None, "refactor_docs").unwrap_err(); + assert!(err.contains("refactor_docs")); + assert!(err.contains("hasn't replied")); + } + + #[test] + fn codex_returns_unsupported_error() { + let err = resume_command("codex", Some("conv-abc"), "scan_repo").unwrap_err(); + assert!(err.contains("not supported for codex")); + assert!(err.contains("gamechat tail scan_repo")); + } + + #[test] + fn unknown_provider_returns_unsupported_error() { + let err = resume_command("mystery", Some("id"), "slug").unwrap_err(); + assert!(err.contains("mystery")); + assert!(err.contains("Only claude")); + } +} diff --git a/src/control/mod.rs b/src/control/mod.rs new file mode 100644 index 0000000..4032849 --- /dev/null +++ b/src/control/mod.rs @@ -0,0 +1,50 @@ +//! Control plane for inspecting a running `gamechat --realtime` process from +//! another terminal. +//! +//! Self-contained: only depends on `Arc` from the orchestrator +//! and never touches the voice loop, audio, or the realtime websocket. The +//! voice loop calls [`spawn_server`] once at startup; everything else lives +//! inside this module. +//! +//! Wire protocol: newline-delimited JSON over a Unix domain socket at +//! `$RUNTIME_DIR/$pid.sock`. One request per connection. See +//! [`protocol`] for the shape of each message. +//! +//! CLI surface (handled by [`run_cli`]): +//! +//! - `gamechat inspect [--pid N]` — list all active sub-agents. +//! - `gamechat tail [--pid N]` — follow a slug's progress buffer. +//! - `gamechat open [--pid N] [--launch]` — print (or launch) the +//! resume command for the underlying agent UI. + +mod client; +mod protocol; +mod runtime_dir; +mod server; + +use std::path::PathBuf; + +pub(crate) use server::spawn_server; + +/// Subcommand parsed from `gamechat`'s argv. +#[derive(Debug)] +pub(crate) enum ControlSubcommand { + Inspect, + Tail { slug: String }, + Open { slug: String, launch: bool }, +} + +/// Optional connection target. `pid` picks a specific running gamechat; if +/// `None` the client looks for exactly one socket in the runtime dir. +#[derive(Debug, Default)] +pub(crate) struct ControlTarget { + pub pid: Option, + pub socket: Option, +} + +pub(crate) async fn run_cli( + subcommand: ControlSubcommand, + target: ControlTarget, +) -> Result<(), String> { + client::run(subcommand, target).await +} diff --git a/src/control/protocol.rs b/src/control/protocol.rs new file mode 100644 index 0000000..2cca9eb --- /dev/null +++ b/src/control/protocol.rs @@ -0,0 +1,53 @@ +//! Wire protocol for the gamechat control socket. +//! +//! Newline-delimited JSON: one [`Request`] in, one [`Response`] back. The +//! socket closes immediately after the response — clients reconnect for the +//! next call (including the polling tail loop). Simple, stateless, and +//! avoids any subscription bookkeeping inside the server. + +use crate::orchestrator::progress::{JobStatus, SlugSummary}; +use serde::{Deserialize, Serialize}; + +#[derive(Debug, Serialize, Deserialize)] +#[serde(tag = "kind", rename_all = "snake_case")] +pub(crate) enum Request { + /// Server identity + a brief "alive" handshake used by `inspect`. + Hello, + /// Snapshot of every known slug. + List, + /// Pull buffered progress entries newer than `after_seq`. `after_seq = + /// None` returns everything currently buffered. + Tail { + slug: String, + after_seq: Option, + }, + /// Look up the resume target for a slug — provider + session id. + Resume { slug: String }, +} + +#[derive(Debug, Serialize, Deserialize)] +#[serde(tag = "kind", rename_all = "snake_case")] +pub(crate) enum Response { + Hello { + pid: u32, + version: String, + provider: String, + }, + List { + slugs: Vec, + }, + Tail { + entries: Vec, + next_cursor: Option, + status: JobStatus, + done: bool, + }, + Resume { + slug: String, + provider: String, + session_id: Option, + }, + Error { + message: String, + }, +} diff --git a/src/control/runtime_dir.rs b/src/control/runtime_dir.rs new file mode 100644 index 0000000..46a7b63 --- /dev/null +++ b/src/control/runtime_dir.rs @@ -0,0 +1,85 @@ +//! Where gamechat puts its control sockets. +//! +//! macOS doesn't set `XDG_RUNTIME_DIR`, so we fall back to `$TMPDIR`. The +//! returned directory always contains the username so multiple users on the +//! same machine never collide. Socket paths look like +//! `/gamechat-/.sock`. + +use std::path::PathBuf; + +pub(crate) fn runtime_dir() -> Result { + let base = std::env::var("XDG_RUNTIME_DIR") + .ok() + .map(PathBuf::from) + .or_else(|| std::env::var("TMPDIR").ok().map(PathBuf::from)) + .unwrap_or_else(|| PathBuf::from("/tmp")); + let user = std::env::var("USER") + .or_else(|_| std::env::var("LOGNAME")) + .unwrap_or_else(|_| "default".to_string()); + Ok(base.join(format!("gamechat-{user}"))) +} + +pub(crate) fn socket_path_for_pid(pid: u32) -> Result { + Ok(runtime_dir()?.join(format!("{pid}.sock"))) +} + +/// Scan the runtime dir for `*.sock` files. Stale entries (no matching live +/// process) are filtered out. +pub(crate) fn discover_sockets() -> Result, String> { + let dir = runtime_dir()?; + let read = match std::fs::read_dir(&dir) { + Ok(read) => read, + Err(err) if err.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()), + Err(err) => return Err(format!("failed to read runtime dir {}: {err}", dir.display())), + }; + let mut sockets: Vec = Vec::new(); + for entry in read.flatten() { + let path = entry.path(); + if path.extension().and_then(|s| s.to_str()) != Some("sock") { + continue; + } + // Drop sockets whose pid no longer exists; the server unlinks on + // graceful shutdown, but crashes leave files behind. + if let Some(pid) = path + .file_stem() + .and_then(|s| s.to_str()) + .and_then(|s| s.parse::().ok()) + { + if !pid_is_alive(pid) { + let _ = std::fs::remove_file(&path); + continue; + } + } + sockets.push(path); + } + sockets.sort(); + Ok(sockets) +} + +#[cfg(unix)] +fn pid_is_alive(pid: u32) -> bool { + // Sending signal 0 is a no-op probe that returns success iff the process + // exists and we have permission to signal it. + unsafe { libc_kill(pid as i32, 0) == 0 || std::io::Error::last_os_error().raw_os_error() == Some(libc::EPERM) } +} + +#[cfg(not(unix))] +fn pid_is_alive(_pid: u32) -> bool { + true +} + +#[cfg(unix)] +unsafe fn libc_kill(pid: i32, sig: i32) -> i32 { + extern "C" { + fn kill(pid: i32, sig: i32) -> i32; + } + unsafe { kill(pid, sig) } +} + +// libc is in std on unix via libc crate; we don't want to pull libc as a +// dep, so duplicate the EPERM constant here. +#[cfg(unix)] +#[allow(non_snake_case)] +mod libc { + pub const EPERM: i32 = 1; +} diff --git a/src/control/server.rs b/src/control/server.rs new file mode 100644 index 0000000..cdf5723 --- /dev/null +++ b/src/control/server.rs @@ -0,0 +1,364 @@ +//! Unix socket server task that exposes a running gamechat's progress state. +//! +//! Wraps the same `Arc` the voice loop already owns. Each +//! incoming connection reads a single JSON line, computes a response, writes +//! it, and closes. No subscription state, no streaming — clients reconnect +//! for the next call (the polling tail loop relies on this). + +use super::protocol::{Request, Response}; +use super::runtime_dir::socket_path_for_pid; +use crate::orchestrator::progress::{JobStatus, ProgressStore, SlugSummary}; +use std::path::PathBuf; +use std::sync::Arc; +use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; +use tokio::net::{UnixListener, UnixStream}; + +/// Handle returned to the voice loop; dropping it removes the socket file. +pub(crate) struct ServerHandle { + socket_path: PathBuf, +} + +impl Drop for ServerHandle { + fn drop(&mut self) { + let _ = std::fs::remove_file(&self.socket_path); + } +} + +/// Start the control server. Best-effort: if the runtime dir can't be created +/// or the socket can't be bound, returns Err and the caller logs a warning. +pub(crate) fn spawn_server( + store: Arc, + provider_name: &'static str, +) -> Result { + let pid = std::process::id(); + let socket_path = socket_path_for_pid(pid)?; + if let Some(parent) = socket_path.parent() { + std::fs::create_dir_all(parent) + .map_err(|e| format!("failed to create runtime dir {}: {e}", parent.display()))?; + } + // Clean up any stale socket from a previous crash with the same pid (unlikely). + let _ = std::fs::remove_file(&socket_path); + let listener = UnixListener::bind(&socket_path) + .map_err(|e| format!("failed to bind control socket {}: {e}", socket_path.display()))?; + eprintln!( + "control socket listening pid={pid} provider={provider_name} path={}", + socket_path.display() + ); + let store_for_task = Arc::clone(&store); + tokio::spawn(async move { + accept_loop(listener, store_for_task, provider_name).await; + }); + Ok(ServerHandle { socket_path }) +} + +async fn accept_loop(listener: UnixListener, store: Arc, provider_name: &'static str) { + loop { + match listener.accept().await { + Ok((stream, _addr)) => { + let store = Arc::clone(&store); + tokio::spawn(async move { + if let Err(err) = handle_connection(stream, store, provider_name).await { + eprintln!("control connection error: {err}"); + } + }); + } + Err(err) => { + eprintln!("control socket accept failed: {err}"); + break; + } + } + } +} + +async fn handle_connection( + stream: UnixStream, + store: Arc, + provider_name: &'static str, +) -> Result<(), String> { + let (read_half, mut write_half) = stream.into_split(); + let mut reader = BufReader::new(read_half); + let mut line = String::new(); + let n = reader + .read_line(&mut line) + .await + .map_err(|e| format!("failed to read request: {e}"))?; + if n == 0 { + return Ok(()); + } + let response = match serde_json::from_str::(line.trim()) { + Ok(request) => handle_request(request, &store, provider_name), + Err(err) => Response::Error { + message: format!("invalid request: {err}"), + }, + }; + let mut payload = serde_json::to_string(&response) + .map_err(|e| format!("failed to serialize response: {e}"))?; + payload.push('\n'); + write_half + .write_all(payload.as_bytes()) + .await + .map_err(|e| format!("failed to write response: {e}"))?; + write_half + .flush() + .await + .map_err(|e| format!("failed to flush response: {e}"))?; + Ok(()) +} + +fn handle_request( + request: Request, + store: &ProgressStore, + provider_name: &'static str, +) -> Response { + match request { + Request::Hello => Response::Hello { + pid: std::process::id(), + version: env!("CARGO_PKG_VERSION").to_string(), + provider: provider_name.to_string(), + }, + Request::List => Response::List { + slugs: store.snapshot_all(), + }, + Request::Tail { slug, after_seq } => match store.entries_after(&slug, after_seq) { + Some(batch) => Response::Tail { + entries: batch.entries, + next_cursor: batch.next_cursor, + status: batch.status, + done: batch.done, + }, + None => Response::Error { + message: format!("unknown slug: {slug}"), + }, + }, + Request::Resume { slug } => { + let summary = store + .snapshot_all() + .into_iter() + .find(|s: &SlugSummary| s.slug == slug); + match summary { + Some(s) => Response::Resume { + slug: s.slug, + provider: s.provider, + session_id: s.session_id, + }, + None => Response::Error { + message: format!("unknown slug: {slug}"), + }, + } + } + } +} + +// Suppress an "unused" warning for JobStatus until the client uses it via +// the deserialized Tail response. +#[allow(dead_code)] +fn _force_use_job_status(_: JobStatus) {} + +#[cfg(test)] +mod tests { + use super::*; + use crate::orchestrator::progress::ProgressStore; + use std::sync::Arc; + use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; + + #[tokio::test] + async fn end_to_end_list_and_tail_over_socket() { + let store = Arc::new(ProgressStore::new()); + store.register_job("e2e", "claude"); + store.set_running("e2e"); + store.push_progress("e2e", "step one"); + store.set_session_id("e2e", "abc123"); + + // Bind a temporary socket inside the test's tmpdir to avoid colliding + // with the runtime dir layout. + let dir = tempdir_test(); + let socket_path = dir.join("test.sock"); + let listener = UnixListener::bind(&socket_path).unwrap(); + + let store_for_task = Arc::clone(&store); + tokio::spawn(async move { accept_loop(listener, store_for_task, "claude").await }); + + // List → should see our slug + session_id. + let resp = roundtrip(&socket_path, Request::List).await; + match resp { + Response::List { slugs } => { + assert_eq!(slugs.len(), 1); + assert_eq!(slugs[0].slug, "e2e"); + assert_eq!(slugs[0].provider, "claude"); + assert_eq!(slugs[0].session_id.as_deref(), Some("abc123")); + } + other => panic!("expected List, got {other:?}"), + } + + // Tail starting from None returns everything; then cursor-based call + // returns only newly-pushed entries. + let resp = roundtrip( + &socket_path, + Request::Tail { + slug: "e2e".into(), + after_seq: None, + }, + ) + .await; + let cursor = match resp { + Response::Tail { + entries, + next_cursor, + .. + } => { + assert_eq!(entries, vec!["step one"]); + next_cursor + } + other => panic!("expected Tail, got {other:?}"), + }; + + store.push_progress("e2e", "step two"); + let resp = roundtrip( + &socket_path, + Request::Tail { + slug: "e2e".into(), + after_seq: cursor, + }, + ) + .await; + match resp { + Response::Tail { entries, .. } => { + assert_eq!(entries, vec!["step two"]); + } + other => panic!("expected Tail, got {other:?}"), + } + + // Resume returns the session id we attached. + let resp = roundtrip( + &socket_path, + Request::Resume { + slug: "e2e".into(), + }, + ) + .await; + match resp { + Response::Resume { session_id, .. } => { + assert_eq!(session_id.as_deref(), Some("abc123")); + } + other => panic!("expected Resume, got {other:?}"), + } + } + + async fn roundtrip(socket: &PathBuf, request: Request) -> Response { + let stream = UnixStream::connect(socket).await.expect("connect"); + let (read_half, mut write_half) = stream.into_split(); + let mut line = serde_json::to_string(&request).unwrap(); + line.push('\n'); + write_half.write_all(line.as_bytes()).await.unwrap(); + write_half.shutdown().await.unwrap(); + let mut reader = BufReader::new(read_half); + let mut response_line = String::new(); + reader.read_line(&mut response_line).await.unwrap(); + serde_json::from_str(response_line.trim()).expect("valid response") + } + + fn tempdir_test() -> PathBuf { + let base = std::env::temp_dir().join(format!( + "gamechat-test-{}-{}", + std::process::id(), + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_nanos() + )); + std::fs::create_dir_all(&base).unwrap(); + base + } + + #[test] + fn list_handler_returns_summary() { + let store = Arc::new(ProgressStore::new()); + store.register_job("a", "claude"); + store.set_running("a"); + store.push_progress("a", "step"); + store.set_session_id("a", "abc"); + + let resp = handle_request(Request::List, &store, "claude"); + match resp { + Response::List { slugs } => { + assert_eq!(slugs.len(), 1); + assert_eq!(slugs[0].slug, "a"); + assert_eq!(slugs[0].session_id.as_deref(), Some("abc")); + } + other => panic!("expected List response, got {other:?}"), + } + } + + #[test] + fn resume_handler_returns_session_id() { + let store = Arc::new(ProgressStore::new()); + store.register_job("solo", "claude"); + store.set_session_id("solo", "sess-xyz"); + let resp = handle_request( + Request::Resume { + slug: "solo".into(), + }, + &store, + "claude", + ); + match resp { + Response::Resume { + slug, + provider, + session_id, + } => { + assert_eq!(slug, "solo"); + assert_eq!(provider, "claude"); + assert_eq!(session_id.as_deref(), Some("sess-xyz")); + } + other => panic!("expected Resume, got {other:?}"), + } + } + + #[test] + fn tail_handler_streams_entries() { + let store = Arc::new(ProgressStore::new()); + store.register_job("t", "claude"); + store.push_progress("t", "one"); + store.push_progress("t", "two"); + + let resp = handle_request( + Request::Tail { + slug: "t".into(), + after_seq: None, + }, + &store, + "claude", + ); + match resp { + Response::Tail { + entries, + next_cursor, + done, + .. + } => { + assert_eq!(entries, vec!["one", "two"]); + assert!(next_cursor.is_some()); + assert!(!done); + } + other => panic!("expected Tail, got {other:?}"), + } + } + + #[test] + fn unknown_slug_in_tail_is_error() { + let store = Arc::new(ProgressStore::new()); + let resp = handle_request( + Request::Tail { + slug: "ghost".into(), + after_seq: None, + }, + &store, + "claude", + ); + match resp { + Response::Error { message } => assert!(message.contains("ghost")), + other => panic!("expected Error, got {other:?}"), + } + } +} diff --git a/src/main.rs b/src/main.rs index eb5efec..602500a 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,7 +1,9 @@ +mod control; mod orchestrator; mod types; mod voice_loop; +use control::{ControlSubcommand, ControlTarget}; use orchestrator::OrchestratorProvider; use std::path::PathBuf; use types::{DelegateToOrchestratorArgs, VoiceUpdate}; @@ -17,6 +19,13 @@ async fn main() { async fn run() -> Result<(), String> { load_dotenv(); + + // Subcommand short-circuit: `gamechat inspect|tail|open …` route entirely + // through the control module and never touch the realtime flag plumbing. + if let Some((subcommand, target)) = parse_control_subcommand()? { + return control::run_cli(subcommand, target).await; + } + let args = CliArgs::parse()?; if args.print_realtime_config { @@ -202,6 +211,80 @@ fn next_value(args: &mut impl Iterator, name: &str) -> Result Result, String> { + let mut iter = std::env::args().skip(1); + let Some(first) = iter.next() else { + return Ok(None); + }; + let subcommand_name = first.as_str(); + if !matches!(subcommand_name, "inspect" | "tail" | "open") { + return Ok(None); + } + let mut positional: Vec = Vec::new(); + let mut target = ControlTarget::default(); + let mut launch = false; + while let Some(arg) = iter.next() { + match arg.as_str() { + "--pid" => { + let value = next_value(&mut iter, "--pid")?; + let pid: u32 = value + .parse() + .map_err(|e| format!("--pid expected a number: {e}"))?; + target.pid = Some(pid); + } + "--socket" => { + target.socket = Some(PathBuf::from(next_value(&mut iter, "--socket")?)); + } + "--launch" => launch = true, + "--help" | "-h" => return Err(control_usage()), + other if other.starts_with("--") => { + return Err(format!("unknown argument for {subcommand_name}: {other}")); + } + other => positional.push(other.to_string()), + } + } + let subcommand = match subcommand_name { + "inspect" => { + if !positional.is_empty() { + return Err(format!( + "inspect takes no positional arguments, got: {}", + positional.join(" ") + )); + } + ControlSubcommand::Inspect + } + "tail" => { + let slug = positional + .into_iter() + .next() + .ok_or_else(|| "tail requires a slug, e.g. `gamechat tail refactor_docs`".to_string())?; + ControlSubcommand::Tail { slug } + } + "open" => { + let slug = positional + .into_iter() + .next() + .ok_or_else(|| "open requires a slug, e.g. `gamechat open refactor_docs`".to_string())?; + ControlSubcommand::Open { slug, launch } + } + _ => unreachable!(), + }; + Ok(Some((subcommand, target))) +} + +fn control_usage() -> String { + "usage: + gamechat inspect [--pid N | --socket PATH] + gamechat tail [--pid N | --socket PATH] + gamechat open [--pid N | --socket PATH] [--launch] + +Inspect a running gamechat realtime session from another terminal. With no +target flag the client connects to the only running gamechat instance and +errors if more than one is present. +" + .to_string() +} + fn build_orchestrator_provider( provider: &str, codex_bin: Option, @@ -224,6 +307,10 @@ fn usage() -> String { gamechat --once \"user asked for deeper work\" gamechat --print-realtime-config + gamechat inspect List active sub-agents in a running session. + gamechat tail Stream a sub-agent's progress buffer. + gamechat open [--launch] Print (or launch on macOS) the resume command. + options: --realtime Start the live voice loop (mic + speakers). --once Run a single delegation and print the VoiceUpdate JSON. diff --git a/src/orchestrator/claude/client.rs b/src/orchestrator/claude/client.rs index 3c54f17..d86bfb4 100644 --- a/src/orchestrator/claude/client.rs +++ b/src/orchestrator/claude/client.rs @@ -160,6 +160,9 @@ impl ClaudeClient { .map_err(|e| format!("claude output was not valid utf-8: {e}"))?; let (reply, session_id) = parse_claude_stream_output(&stdout_text); if let Some(session_id) = session_id { + if let Some(reporter) = progress.as_ref() { + reporter.set_session_id(&session_id); + } self.session_id = Some(session_id); } let reply = reply.unwrap_or(stdout_text); diff --git a/src/orchestrator/interface.rs b/src/orchestrator/interface.rs index 8302f23..a453fbf 100644 --- a/src/orchestrator/interface.rs +++ b/src/orchestrator/interface.rs @@ -103,6 +103,10 @@ impl OrchestratorProvider { } } + pub(crate) fn name(&self) -> &'static str { + self.inner.name() + } + pub(crate) async fn open_session(&self, slug: &str) -> Result { let session = self.inner.open_session(slug).await?; eprintln!( diff --git a/src/orchestrator/jobs.rs b/src/orchestrator/jobs.rs index f60d67c..e917760 100644 --- a/src/orchestrator/jobs.rs +++ b/src/orchestrator/jobs.rs @@ -29,10 +29,12 @@ pub(crate) struct OrchestratorJobManager { job_tx: mpsc::UnboundedSender, event_rx: mpsc::UnboundedReceiver, progress: Arc, + provider_name: &'static str, } impl OrchestratorJobManager { pub(crate) fn spawn(provider: OrchestratorProvider) -> Self { + let provider_name = provider.name(); let (job_tx, job_rx) = mpsc::unbounded_channel(); let (event_tx, event_rx) = mpsc::unbounded_channel(); let progress = Arc::new(ProgressStore::new()); @@ -42,14 +44,19 @@ impl OrchestratorJobManager { event_tx, Arc::clone(&progress), )); - eprintln!("orchestrator job manager started"); + eprintln!("orchestrator job manager started provider={provider_name}"); Self { job_tx, event_rx, progress, + provider_name, } } + pub(crate) fn progress_store(&self) -> Arc { + Arc::clone(&self.progress) + } + pub(crate) fn enqueue(&self, job: OrchestratorJob) -> Result<(), String> { eprintln!( "orchestrator job enqueue job={} slug={} urgency={} intent_preview={}", @@ -58,7 +65,7 @@ impl OrchestratorJobManager { job.args.urgency, preview(&job.args.user_intent) ); - self.progress.register_job(&job.slug, "orchestrator"); + self.progress.register_job(&job.slug, self.provider_name); self.progress .push_progress(&job.slug, &format!("Queued background job {}.", job.id)); self.job_tx diff --git a/src/orchestrator/progress.rs b/src/orchestrator/progress.rs index e1e04df..2334837 100644 --- a/src/orchestrator/progress.rs +++ b/src/orchestrator/progress.rs @@ -2,7 +2,7 @@ use std::collections::{HashMap, VecDeque}; use std::sync::{Arc, Mutex}; use std::time::Instant; -use serde::Serialize; +use serde::{Deserialize, Serialize}; /// Maximum number of progress entries retained per job. const MAX_BUFFER_ENTRIES: usize = 20; @@ -17,7 +17,7 @@ pub(crate) const DEFAULT_WINDOW_SIZE: usize = 1000; // ── Public types ──────────────────────────────────────────────────────── -#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "snake_case")] pub(crate) enum JobStatus { Queued, @@ -37,6 +37,31 @@ impl std::fmt::Display for JobStatus { } } +/// Lightweight per-slug summary used by the control socket. Unlike +/// [`ProgressSnapshot`], this never consumes the per-slug rate-limit window. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub(crate) struct SlugSummary { + pub slug: String, + pub provider: String, + pub status: JobStatus, + pub last_message: String, + pub elapsed_seconds: f64, + pub session_id: Option, + /// Sequence number of the most recent buffered entry, or `None` if the + /// buffer is empty. Clients use this as the initial tail cursor. + pub last_seq: Option, +} + +/// Batch returned by [`ProgressStore::entries_after`]: any newly-buffered +/// entries together with the updated cursor. +#[derive(Debug, Clone, Serialize)] +pub(crate) struct TailBatch { + pub entries: Vec, + pub next_cursor: Option, + pub status: JobStatus, + pub done: bool, +} + /// Read-only snapshot returned to callers that query a job's progress. #[derive(Debug, Clone, Serialize)] pub(crate) struct ProgressSnapshot { @@ -71,6 +96,10 @@ impl ProgressReporter { pub(crate) fn push(&self, message: &str) { self.store.push_progress(&self.slug, message); } + + pub(crate) fn set_session_id(&self, id: &str) { + self.store.set_session_id(&self.slug, id); + } } // ── Internal tracking ────────────────────────────────────────────────── @@ -80,13 +109,21 @@ struct JobProgress { provider: String, /// The most recent single progress line (e.g. last agent reply). last_message: String, - /// Bounded ring of recent progress entries (newest last). - recent_buffer: VecDeque, + /// Bounded ring of recent progress entries (newest last). Each carries a + /// monotonically increasing sequence number so cursor-based tail readers + /// can stream only new content. + recent_buffer: VecDeque<(u64, String)>, + /// Next sequence number to assign on `push_progress`. + next_seq: u64, last_updated: Instant, /// Wall-clock origin for `elapsed_seconds`. Set on `register_job` and /// re-stamped on `set_running` so the voice model gets time-in-flight /// rather than time-in-queue once the job actually starts. started_at: Instant, + /// Backend-reported conversation/session id (e.g. Claude Code's session + /// id), once known. Used by external inspection clients to resume the + /// underlying agent UI. + session_id: Option, } struct ProgressInner { @@ -121,8 +158,10 @@ impl ProgressStore { provider: provider.to_string(), last_message: String::new(), recent_buffer: VecDeque::new(), + next_seq: 0, last_updated: now, started_at: now, + session_id: None, }, ); } @@ -152,7 +191,9 @@ impl ProgressStore { if progress.recent_buffer.len() >= MAX_BUFFER_ENTRIES { progress.recent_buffer.pop_front(); } - progress.recent_buffer.push_back(trimmed.to_string()); + let seq = progress.next_seq; + progress.next_seq = progress.next_seq.wrapping_add(1); + progress.recent_buffer.push_back((seq, trimmed.to_string())); progress.last_updated = Instant::now(); } } @@ -168,7 +209,9 @@ impl ProgressStore { if progress.recent_buffer.len() >= MAX_BUFFER_ENTRIES { progress.recent_buffer.pop_front(); } - progress.recent_buffer.push_back(trimmed.to_string()); + let seq = progress.next_seq; + progress.next_seq = progress.next_seq.wrapping_add(1); + progress.recent_buffer.push_back((seq, trimmed.to_string())); } progress.last_updated = Instant::now(); } @@ -187,6 +230,64 @@ impl ProgressStore { } } + /// Attach (or update) the backend session id for a slug. Used by external + /// inspection clients to launch the underlying agent UI on the right + /// conversation. + pub fn set_session_id(&self, slug: &str, session_id: &str) { + let mut inner = self.inner.lock().expect("progress store mutex poisoned"); + if let Some(progress) = inner.jobs.get_mut(slug) { + progress.session_id = Some(session_id.to_string()); + } + } + + /// Non-rate-limited summary of every known slug. Intended for the control + /// socket, not the voice model — does not consume the per-slug rate-limit + /// window. + pub fn snapshot_all(&self) -> Vec { + let inner = self.inner.lock().expect("progress store mutex poisoned"); + let mut entries: Vec = inner + .jobs + .iter() + .map(|(slug, p)| SlugSummary { + slug: slug.clone(), + provider: p.provider.clone(), + status: p.status, + last_message: p.last_message.clone(), + elapsed_seconds: p.started_at.elapsed().as_secs_f64(), + session_id: p.session_id.clone(), + last_seq: p.recent_buffer.back().map(|(seq, _)| *seq), + }) + .collect(); + entries.sort_by(|a, b| a.slug.cmp(&b.slug)); + entries + } + + /// Return buffered entries whose sequence number is greater than + /// `after_seq`. `None` if the slug is unknown. The returned cursor is the + /// largest sequence number returned (or `after_seq` if nothing new). + pub fn entries_after(&self, slug: &str, after_seq: Option) -> Option { + let inner = self.inner.lock().expect("progress store mutex poisoned"); + let progress = inner.jobs.get(slug)?; + let mut entries: Vec = Vec::new(); + let mut max_seq = after_seq; + for (seq, msg) in &progress.recent_buffer { + let include = match after_seq { + Some(cursor) => *seq > cursor, + None => true, + }; + if include { + entries.push(msg.clone()); + max_seq = Some(max_seq.map_or(*seq, |existing| existing.max(*seq))); + } + } + Some(TailBatch { + entries, + next_cursor: max_seq, + status: progress.status, + done: matches!(progress.status, JobStatus::Completed | JobStatus::Failed), + }) + } + /// Query the progress of a job. Returns `None` if the job is unknown. /// Applies rate limiting: when the caller polls within `RATE_LIMIT_SECS`, /// the snapshot still carries the cached fields but `rate_limited` is set @@ -238,7 +339,7 @@ impl ProgressStore { /// Concatenate buffer entries and truncate to `max_chars`, keeping the /// tail (most recent) content. -fn truncate_buffer(buffer: &VecDeque, max_chars: usize) -> String { +fn truncate_buffer(buffer: &VecDeque<(u64, String)>, max_chars: usize) -> String { if buffer.is_empty() { return String::new(); } @@ -247,7 +348,7 @@ fn truncate_buffer(buffer: &VecDeque, max_chars: usize) -> String { let joined = buffer .iter() .enumerate() - .map(|(i, entry)| { + .map(|(i, (_, entry))| { if i == 0 { entry.clone() } else { @@ -416,6 +517,7 @@ mod tests { .recent_buffer .front() .unwrap() + .1 .starts_with("entry 5") ); assert!( @@ -423,7 +525,67 @@ mod tests { .recent_buffer .back() .unwrap() + .1 .starts_with("entry 24") ); } + + #[test] + fn snapshot_all_returns_sorted_summaries() { + let store = ProgressStore::new(); + store.register_job("zeta", "claude"); + store.register_job("alpha", "codex"); + store.set_running("alpha"); + store.push_progress("alpha", "step one"); + store.set_session_id("alpha", "sess-1"); + + let summaries = store.snapshot_all(); + assert_eq!(summaries.len(), 2); + assert_eq!(summaries[0].slug, "alpha"); + assert_eq!(summaries[0].session_id.as_deref(), Some("sess-1")); + assert_eq!(summaries[0].status, JobStatus::Running); + assert_eq!(summaries[1].slug, "zeta"); + assert_eq!(summaries[1].session_id, None); + } + + #[test] + fn entries_after_streams_only_new_lines() { + let store = ProgressStore::new(); + store.register_job("tailme", "claude"); + store.set_running("tailme"); + store.push_progress("tailme", "alpha"); + store.push_progress("tailme", "bravo"); + + let first = store.entries_after("tailme", None).unwrap(); + assert_eq!(first.entries, vec!["alpha", "bravo"]); + let cursor = first.next_cursor; + assert!(cursor.is_some()); + + // No new entries — empty batch, cursor unchanged. + let stale = store.entries_after("tailme", cursor).unwrap(); + assert!(stale.entries.is_empty()); + assert_eq!(stale.next_cursor, cursor); + + store.push_progress("tailme", "charlie"); + let next = store.entries_after("tailme", cursor).unwrap(); + assert_eq!(next.entries, vec!["charlie"]); + assert!(next.next_cursor > cursor); + } + + #[test] + fn entries_after_unknown_slug_is_none() { + let store = ProgressStore::new(); + assert!(store.entries_after("nope", None).is_none()); + } + + #[test] + fn entries_after_flags_done_on_terminal_status() { + let store = ProgressStore::new(); + store.register_job("finis", "claude"); + store.set_running("finis"); + store.set_completed("finis", "all done"); + let batch = store.entries_after("finis", None).unwrap(); + assert!(batch.done); + assert_eq!(batch.status, JobStatus::Completed); + } } diff --git a/src/voice_loop/mod.rs b/src/voice_loop/mod.rs index e915ad3..32ece11 100644 --- a/src/voice_loop/mod.rs +++ b/src/voice_loop/mod.rs @@ -31,9 +31,23 @@ pub(crate) struct RealtimeRunConfig { } pub(crate) async fn run_realtime_voice(config: RealtimeRunConfig) -> Result<(), String> { + let provider_name = config.orchestrator_provider.name(); let orchestrator_jobs = OrchestratorJobManager::spawn(config.orchestrator_provider); let orchestrator_bridge = OrchestratorBridge::new(); + // Best-effort: keep the realtime loop running even if the control socket + // fails to bind (read-only headless environments, etc). + let _control_handle = match crate::control::spawn_server( + orchestrator_jobs.progress_store(), + provider_name, + ) { + Ok(handle) => Some(handle), + Err(err) => { + eprintln!("control socket disabled: {err}"); + None + } + }; + let playback = Arc::new(Mutex::new(PlaybackBuffer::new())); let (_output_stream, output_rate) = start_output_stream(Arc::clone(&playback))?; let (mic_tx, mut mic_rx) = mpsc::unbounded_channel::();