Daemon: PtyTarget — exec a terminal PTY into an agent container (node-placed terminal foundation)
Generalize the daemon's PTY spawn so a session can target either the node's host shell (today) or `docker exec -it <container> tmux …` (the node-placed agent terminal, which shares the container's node-local ~/drives). A `container` (+ optional `session`) field on pty_open/webrtc_offer selects the container path; absent it, the host shell path is byte-identical to before — so the Infra node terminal is unaffected. Threads PtyTarget through open_pty + rtc handle_offer → build_peer → bridge_pty. Additive; nothing emits `container` yet. Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
This commit is contained in:
co-authored by
Claude Opus 4.8
parent
1072ec159a
commit
12212c72ac
@@ -280,7 +280,7 @@ async fn handle_frame(
|
|||||||
let cols = v.get("cols").and_then(Value::as_u64).unwrap_or(80) as u16;
|
let cols = v.get("cols").and_then(Value::as_u64).unwrap_or(80) as u16;
|
||||||
let rows = v.get("rows").and_then(Value::as_u64).unwrap_or(24) as u16;
|
let rows = v.get("rows").and_then(Value::as_u64).unwrap_or(24) as u16;
|
||||||
eprintln!("[pty] received pty_open sid={sid}");
|
eprintln!("[pty] received pty_open sid={sid}");
|
||||||
if let Err(e) = open_pty(sid, cols, rows, out.clone(), ptys.clone()).await {
|
if let Err(e) = open_pty(sid, cols, rows, PtyTarget::from_frame(&v), out.clone(), ptys.clone()).await {
|
||||||
eprintln!("[pty] open_pty FAILED sid={sid}: {e}");
|
eprintln!("[pty] open_pty FAILED sid={sid}: {e}");
|
||||||
let _ = out.send(json!({ "t": "pty_out", "sid": sid, "data": B64.encode(format!("\r\n\x1b[31m[clawmates] could not start shell: {e}\x1b[0m\r\n").as_bytes()) }).to_string());
|
let _ = out.send(json!({ "t": "pty_out", "sid": sid, "data": B64.encode(format!("\r\n\x1b[31m[clawmates] could not start shell: {e}\x1b[0m\r\n").as_bytes()) }).to_string());
|
||||||
let _ = out.send(json!({ "t": "pty_exit", "sid": sid, "error": e }).to_string());
|
let _ = out.send(json!({ "t": "pty_exit", "sid": sid, "error": e }).to_string());
|
||||||
@@ -315,7 +315,7 @@ async fn handle_frame(
|
|||||||
"webrtc_offer" => {
|
"webrtc_offer" => {
|
||||||
let sid = v.get("sid").and_then(Value::as_u64).unwrap_or(0);
|
let sid = v.get("sid").and_then(Value::as_u64).unwrap_or(0);
|
||||||
let sdp = v.get("sdp").and_then(Value::as_str).unwrap_or("").to_owned();
|
let sdp = v.get("sdp").and_then(Value::as_str).unwrap_or("").to_owned();
|
||||||
rtc::handle_offer(sid, sdp, out.clone(), ptys.clone(), peers.clone()).await;
|
rtc::handle_offer(sid, sdp, PtyTarget::from_frame(&v), out.clone(), ptys.clone(), peers.clone()).await;
|
||||||
}
|
}
|
||||||
"webrtc_ice" => {
|
"webrtc_ice" => {
|
||||||
let sid = v.get("sid").and_then(Value::as_u64).unwrap_or(0);
|
let sid = v.get("sid").and_then(Value::as_u64).unwrap_or(0);
|
||||||
@@ -338,37 +338,81 @@ type PtyParts = (
|
|||||||
Box<dyn portable_pty::Child + Send + Sync>,
|
Box<dyn portable_pty::Child + Send + Sync>,
|
||||||
);
|
);
|
||||||
|
|
||||||
/// Spawn a host login shell (tmux `-L clawmates`) in a fresh PTY.
|
/// Spawn `cmd` in a fresh PTY, returning handles for resize/read/write/child.
|
||||||
fn spawn_terminal_pty(cols: u16, rows: u16) -> Result<PtyParts, String> {
|
fn spawn_pty(cmd: CommandBuilder, cols: u16, rows: u16) -> Result<PtyParts, String> {
|
||||||
let pair = native_pty_system()
|
let pair = native_pty_system()
|
||||||
.openpty(PtySize { rows, cols, pixel_width: 0, pixel_height: 0 })
|
.openpty(PtySize { rows, cols, pixel_width: 0, pixel_height: 0 })
|
||||||
.map_err(|e| e.to_string())?;
|
.map_err(|e| e.to_string())?;
|
||||||
let child = pair
|
let child = pair.slave.spawn_command(cmd).map_err(|e| e.to_string())?;
|
||||||
.slave
|
|
||||||
.spawn_command(terminal_command())
|
|
||||||
.map_err(|e| e.to_string())?;
|
|
||||||
drop(pair.slave);
|
drop(pair.slave);
|
||||||
let reader = pair.master.try_clone_reader().map_err(|e| e.to_string())?;
|
let reader = pair.master.try_clone_reader().map_err(|e| e.to_string())?;
|
||||||
let writer = pair.master.take_writer().map_err(|e| e.to_string())?;
|
let writer = pair.master.take_writer().map_err(|e| e.to_string())?;
|
||||||
Ok((pair.master, reader, writer, child))
|
Ok((pair.master, reader, writer, child))
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Spawn a host login shell (tmux `-L clawmates`) in a fresh PTY.
|
||||||
|
fn spawn_terminal_pty(cols: u16, rows: u16) -> Result<PtyParts, String> {
|
||||||
|
spawn_pty(terminal_command(), cols, rows)
|
||||||
|
}
|
||||||
|
|
||||||
|
/// `docker exec -it` a tmux session inside an agent's container — the node-placed
|
||||||
|
/// agent terminal, which shares the container's node-local `~/drives`.
|
||||||
|
fn spawn_container_pty(container: &str, session: &str, cols: u16, rows: u16) -> Result<PtyParts, String> {
|
||||||
|
let mut c = CommandBuilder::new("docker");
|
||||||
|
for a in ["exec", "-it", container, "tmux", "new-session", "-A", "-s", session] {
|
||||||
|
c.arg(a);
|
||||||
|
}
|
||||||
|
spawn_pty(c, cols, rows)
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Where a session's PTY runs: the node's host shell, or `docker exec` into a
|
||||||
|
/// specific agent container on this node.
|
||||||
|
pub(crate) enum PtyTarget {
|
||||||
|
Host,
|
||||||
|
Container { container: String, session: String },
|
||||||
|
}
|
||||||
|
|
||||||
|
impl PtyTarget {
|
||||||
|
/// Parse from a control frame: a non-empty `container` field selects the
|
||||||
|
/// container path (with an optional `session`, default "main").
|
||||||
|
pub(crate) fn from_frame(v: &Value) -> Self {
|
||||||
|
match v.get("container").and_then(Value::as_str) {
|
||||||
|
Some(c) if !c.is_empty() => PtyTarget::Container {
|
||||||
|
container: c.to_owned(),
|
||||||
|
session: v.get("session").and_then(Value::as_str).unwrap_or("main").to_owned(),
|
||||||
|
},
|
||||||
|
_ => PtyTarget::Host,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
pub(crate) fn spawn(&self, cols: u16, rows: u16) -> Result<PtyParts, String> {
|
||||||
|
match self {
|
||||||
|
PtyTarget::Host => spawn_terminal_pty(cols, rows),
|
||||||
|
PtyTarget::Container { container, session } => spawn_container_pty(container, session, cols, rows),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
/// Spawn a host login shell in a PTY; stream its output back as pty_out frames.
|
/// Spawn a host login shell in a PTY; stream its output back as pty_out frames.
|
||||||
async fn open_pty(
|
async fn open_pty(
|
||||||
sid: u64,
|
sid: u64,
|
||||||
cols: u16,
|
cols: u16,
|
||||||
rows: u16,
|
rows: u16,
|
||||||
|
target: PtyTarget,
|
||||||
out: mpsc::UnboundedSender<String>,
|
out: mpsc::UnboundedSender<String>,
|
||||||
ptys: Ptys,
|
ptys: Ptys,
|
||||||
) -> Result<(), String> {
|
) -> Result<(), String> {
|
||||||
let (master, mut reader, writer, child) = spawn_terminal_pty(cols, rows)?;
|
let (master, mut reader, writer, child) = target.spawn(cols, rows)?;
|
||||||
ptys.lock().await.insert(sid, Pty { master, writer, child });
|
ptys.lock().await.insert(sid, Pty { master, writer, child });
|
||||||
let tmux = has_tmux();
|
let label = match &target {
|
||||||
eprintln!("[pty] open sid={sid} cols={cols} rows={rows} tmux={tmux}");
|
PtyTarget::Host => format!("host shell ({})", if has_tmux() { "tmux" } else { "login shell" }),
|
||||||
|
PtyTarget::Container { container, .. } => format!("container {container}"),
|
||||||
|
};
|
||||||
|
eprintln!("[pty] open sid={sid} cols={cols} rows={rows} target={label}");
|
||||||
// Immediate banner over the channel: if the browser shows this but no shell,
|
// Immediate banner over the channel: if the browser shows this but no shell,
|
||||||
// the relay works and the shell is the problem (vs. a dead relay → nothing).
|
// the relay works and the shell is the problem (vs. a dead relay → nothing).
|
||||||
let host = System::host_name().unwrap_or_else(|| "node".into());
|
let host = System::host_name().unwrap_or_else(|| "node".into());
|
||||||
let banner = format!("\r\n\x1b[2m[clawmates] host shell on {host} ({}) — starting…\x1b[0m\r\n", if tmux { "tmux" } else { "login shell" });
|
let banner = format!("\r\n\x1b[2m[clawmates] {label} on {host} — starting…\x1b[0m\r\n");
|
||||||
let _ = out.send(json!({ "t": "pty_out", "sid": sid, "data": B64.encode(banner.as_bytes()) }).to_string());
|
let _ = out.send(json!({ "t": "pty_out", "sid": sid, "data": B64.encode(banner.as_bytes()) }).to_string());
|
||||||
// Blocking PTY reads on a thread → base64 pty_out frames into the out channel.
|
// Blocking PTY reads on a thread → base64 pty_out frames into the out channel.
|
||||||
std::thread::spawn(move || {
|
std::thread::spawn(move || {
|
||||||
|
|||||||
@@ -27,7 +27,7 @@ use webrtc::peer_connection::peer_connection_state::RTCPeerConnectionState;
|
|||||||
use webrtc::peer_connection::sdp::session_description::RTCSessionDescription;
|
use webrtc::peer_connection::sdp::session_description::RTCSessionDescription;
|
||||||
use webrtc::peer_connection::RTCPeerConnection;
|
use webrtc::peer_connection::RTCPeerConnection;
|
||||||
|
|
||||||
use crate::{spawn_terminal_pty, Pty, Ptys};
|
use crate::{Pty, PtyTarget, Ptys};
|
||||||
|
|
||||||
/// Live WebRTC peers keyed by terminal session id.
|
/// Live WebRTC peers keyed by terminal session id.
|
||||||
pub type RtcPeers = Arc<Mutex<HashMap<u64, Arc<RTCPeerConnection>>>>;
|
pub type RtcPeers = Arc<Mutex<HashMap<u64, Arc<RTCPeerConnection>>>>;
|
||||||
@@ -38,11 +38,12 @@ pub type RtcPeers = Arc<Mutex<HashMap<u64, Arc<RTCPeerConnection>>>>;
|
|||||||
pub async fn handle_offer(
|
pub async fn handle_offer(
|
||||||
sid: u64,
|
sid: u64,
|
||||||
sdp: String,
|
sdp: String,
|
||||||
|
target: PtyTarget,
|
||||||
out: mpsc::UnboundedSender<String>,
|
out: mpsc::UnboundedSender<String>,
|
||||||
ptys: Ptys,
|
ptys: Ptys,
|
||||||
peers: RtcPeers,
|
peers: RtcPeers,
|
||||||
) {
|
) {
|
||||||
if let Err(e) = build_peer(sid, sdp, out.clone(), ptys, peers).await {
|
if let Err(e) = build_peer(sid, sdp, target, out.clone(), ptys, peers).await {
|
||||||
eprintln!("[rtc] sid={sid} offer failed: {e}");
|
eprintln!("[rtc] sid={sid} offer failed: {e}");
|
||||||
let _ = out.send(json!({ "t": "webrtc_failed", "sid": sid, "error": e }).to_string());
|
let _ = out.send(json!({ "t": "webrtc_failed", "sid": sid, "error": e }).to_string());
|
||||||
}
|
}
|
||||||
@@ -51,10 +52,12 @@ pub async fn handle_offer(
|
|||||||
async fn build_peer(
|
async fn build_peer(
|
||||||
sid: u64,
|
sid: u64,
|
||||||
sdp: String,
|
sdp: String,
|
||||||
|
target: PtyTarget,
|
||||||
out: mpsc::UnboundedSender<String>,
|
out: mpsc::UnboundedSender<String>,
|
||||||
ptys: Ptys,
|
ptys: Ptys,
|
||||||
peers: RtcPeers,
|
peers: RtcPeers,
|
||||||
) -> Result<(), String> {
|
) -> Result<(), String> {
|
||||||
|
let target = Arc::new(target);
|
||||||
let mut m = MediaEngine::default();
|
let mut m = MediaEngine::default();
|
||||||
let registry = register_default_interceptors(Registry::new(), &mut m).map_err(|e| e.to_string())?;
|
let registry = register_default_interceptors(Registry::new(), &mut m).map_err(|e| e.to_string())?;
|
||||||
let api = APIBuilder::new()
|
let api = APIBuilder::new()
|
||||||
@@ -95,16 +98,20 @@ async fn build_peer(
|
|||||||
})
|
})
|
||||||
}));
|
}));
|
||||||
|
|
||||||
// The browser creates the DataChannel; on open, bridge it to a host PTY.
|
// The browser creates the DataChannel; on open, bridge it to the target PTY
|
||||||
|
// (host shell, or `docker exec` into the agent's container).
|
||||||
let ptys_dc = ptys.clone();
|
let ptys_dc = ptys.clone();
|
||||||
pc.on_data_channel(Box::new(move |dc: Arc<RTCDataChannel>| {
|
pc.on_data_channel(Box::new(move |dc: Arc<RTCDataChannel>| {
|
||||||
let ptys_dc = ptys_dc.clone();
|
let ptys_dc = ptys_dc.clone();
|
||||||
|
let target = target.clone();
|
||||||
Box::pin(async move {
|
Box::pin(async move {
|
||||||
let dc_open = dc.clone();
|
let dc_open = dc.clone();
|
||||||
let ptys_open = ptys_dc.clone();
|
let ptys_open = ptys_dc.clone();
|
||||||
|
let target = target.clone();
|
||||||
dc.on_open(Box::new(move || {
|
dc.on_open(Box::new(move || {
|
||||||
|
let target = target.clone();
|
||||||
Box::pin(async move {
|
Box::pin(async move {
|
||||||
bridge_pty(sid, dc_open, ptys_open).await;
|
bridge_pty(sid, dc_open, ptys_open, target).await;
|
||||||
})
|
})
|
||||||
}));
|
}));
|
||||||
})
|
})
|
||||||
@@ -143,9 +150,9 @@ async fn build_peer(
|
|||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Spawn the host PTY for this session and pump it both ways over the DataChannel.
|
/// Spawn the target PTY for this session and pump it both ways over the DataChannel.
|
||||||
async fn bridge_pty(sid: u64, dc: Arc<RTCDataChannel>, ptys: Ptys) {
|
async fn bridge_pty(sid: u64, dc: Arc<RTCDataChannel>, ptys: Ptys, target: Arc<PtyTarget>) {
|
||||||
let (master, mut reader, writer, child) = match spawn_terminal_pty(120, 40) {
|
let (master, mut reader, writer, child) = match target.spawn(120, 40) {
|
||||||
Ok(parts) => parts,
|
Ok(parts) => parts,
|
||||||
Err(e) => {
|
Err(e) => {
|
||||||
let _ = dc
|
let _ = dc
|
||||||
|
|||||||
Reference in New Issue
Block a user