Agent computer: terminal (tmux + drives + tabs), Obsidian vault, UI polish
Terminal app (xterm ⇄ WebSocket ⇄ per-agent themed container):
- zsh + oh-my-zsh + powerlevel10k image (agent-terminal), runs as uid 65532 to
share read-write ownership of the file-drive volume with the server.
- Interactive PTY in cm-sandbox (bollard exec tty/attach + resize) + a
TerminalManager; ticket-authed WS bridge routed straight to the backend via a
Traefik PathRegexp(/ws) rule. MOTD greets the user by name.
- tmux resumable sessions; multi-tab (one tmux session per tab, same container),
drag-to-reorder, rename, and a Save that persists named tabs to the server
(terminal_tabs, migration 0014) so they survive logout / a new device.
- Files drives mounted per-agent (subpath) at ~/drives/{documents,received,
shared}; a reconciler keeps the Files app's index in sync with terminal writes.
Storage moved to a shared `filedata` volume (CLAWMATES_STORAGE__DATA_DIR).
Obsidian vault (a markdown "second brain" per agent):
- New `vault` FileDrive (migration 0015) mounted into the terminal at ~/obsidian;
a file-content read route; a purple Obsidian tile + a vault viewer app.
Computer UI:
- Draggable computer-panel width (min = phone preset) keeping the size presets.
- Green Terminal glyph, "Claw Chat" → "Chat", colored gradient-outline app icons.
- Agent page: avatar↔activity-grid spacing + larger, uniform section fonts with
colored section-tinted tag chips.
Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
This commit is contained in:
Omar Sobh
co-authored by
Claude Opus 4.8
Claude Opus 4.8
parent
671df7c622
commit
e61724ff82
@@ -1,11 +1,93 @@
|
||||
use std::collections::HashMap;
|
||||
use std::path::Path;
|
||||
|
||||
use axum::extract::{Query, State};
|
||||
use axum::Json;
|
||||
use cm_domain::{AgentId, FileDrive, FileNode};
|
||||
use cm_domain::{AgentId, FileDrive, FileNode, WorkspaceId};
|
||||
use serde::Deserialize;
|
||||
use sqlx::PgPool;
|
||||
|
||||
use crate::routes::claws::workspace_agent;
|
||||
use crate::{ApiError, AppState, Authed};
|
||||
|
||||
/// Recursively collect files under `dir` as (path-relative-to-dir → size).
|
||||
fn walk_files<'a>(
|
||||
dir: std::path::PathBuf,
|
||||
base: std::path::PathBuf,
|
||||
out: &'a mut HashMap<String, u64>,
|
||||
) -> std::pin::Pin<Box<dyn std::future::Future<Output = ()> + Send + 'a>> {
|
||||
Box::pin(async move {
|
||||
let mut rd = match tokio::fs::read_dir(&dir).await {
|
||||
Ok(rd) => rd,
|
||||
Err(_) => return,
|
||||
};
|
||||
while let Ok(Some(entry)) = rd.next_entry().await {
|
||||
let path = entry.path();
|
||||
let Ok(ft) = entry.file_type().await else { continue };
|
||||
if ft.is_dir() {
|
||||
walk_files(path, base.clone(), out).await;
|
||||
} else if ft.is_file() {
|
||||
if let Ok(rel) = path.strip_prefix(&base) {
|
||||
let rel = rel.to_string_lossy().replace('\\', "/");
|
||||
let size = entry.metadata().await.map(|m| m.len()).unwrap_or(0);
|
||||
out.insert(rel, size);
|
||||
}
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
/// Reconcile `file_nodes` with what's actually in the drive's blob directory, so
|
||||
/// files the Terminal created/edited/removed are reflected in the Files app
|
||||
/// (Local backend only; the blob layout is `{root}/{ws}/{drive}/{scope}/{path}`).
|
||||
async fn reconcile_drive(
|
||||
pool: &PgPool,
|
||||
root: &Path,
|
||||
ws: WorkspaceId,
|
||||
drive: FileDrive,
|
||||
agent: AgentId,
|
||||
) {
|
||||
let scope = if drive.is_agent_scoped() {
|
||||
agent.to_string()
|
||||
} else {
|
||||
"shared".to_string()
|
||||
};
|
||||
let dir = root.join(format!("{ws}/{}/{scope}", drive.as_str()));
|
||||
let mut disk = HashMap::new();
|
||||
walk_files(dir, root.join(format!("{ws}/{}/{scope}", drive.as_str())), &mut disk).await;
|
||||
|
||||
let db = match cm_db::repo::files::list(pool, ws, drive, agent).await {
|
||||
Ok(d) => d,
|
||||
Err(_) => return,
|
||||
};
|
||||
// Upsert files present on disk but new/changed in the index.
|
||||
for (path, size) in &disk {
|
||||
let existing = db.iter().find(|n| &n.path == path);
|
||||
if existing.map(|n| n.size as u64) == Some(*size) {
|
||||
continue;
|
||||
}
|
||||
let node = FileNode {
|
||||
id: existing.map(|n| n.id).unwrap_or_else(uuid::Uuid::now_v7),
|
||||
workspace_id: ws,
|
||||
agent_id: drive.is_agent_scoped().then_some(agent),
|
||||
drive,
|
||||
path: path.clone(),
|
||||
size: *size as i64,
|
||||
blob_ref: format!("{ws}/{}/{scope}/{path}", drive.as_str()),
|
||||
};
|
||||
let _ = cm_db::repo::files::upsert(pool, &node).await;
|
||||
}
|
||||
// Drop index rows whose blob no longer exists on disk.
|
||||
for n in &db {
|
||||
if n.blob_ref.is_empty() {
|
||||
continue;
|
||||
}
|
||||
if !root.join(&n.blob_ref).exists() {
|
||||
let _ = cm_db::repo::files::delete(pool, n.id).await;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
pub struct FilesQuery {
|
||||
#[serde(rename = "clawId")]
|
||||
@@ -30,6 +112,9 @@ pub async fn openclaw_files(
|
||||
if !drive.is_agent_scoped() {
|
||||
return Err(ApiError::NotFound); // shared drive has its own route
|
||||
}
|
||||
if let Some(root) = &state.file_root {
|
||||
reconcile_drive(&state.pool, root, user.workspace_id, drive, agent.id).await;
|
||||
}
|
||||
let nodes = cm_db::repo::files::list(&state.pool, user.workspace_id, drive, agent.id).await?;
|
||||
Ok(Json(nodes))
|
||||
}
|
||||
@@ -40,6 +125,44 @@ pub struct SharedQuery {
|
||||
claw_id: AgentId,
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
pub struct ContentQuery {
|
||||
#[serde(rename = "clawId")]
|
||||
claw_id: AgentId,
|
||||
drive: String,
|
||||
path: String,
|
||||
}
|
||||
|
||||
/// GET /api/openclaw/files/content?clawId=&drive=&path= — a file's text content
|
||||
/// from a drive (used by the Obsidian vault viewer to render a note).
|
||||
pub async fn file_content(
|
||||
State(state): State<AppState>,
|
||||
Authed(user): Authed,
|
||||
Query(query): Query<ContentQuery>,
|
||||
) -> Result<String, ApiError> {
|
||||
let agent = workspace_agent(&state, &user, query.claw_id).await?;
|
||||
let drive: FileDrive = query.drive.parse().map_err(|_| ApiError::NotFound)?;
|
||||
let scope = if drive.is_agent_scoped() {
|
||||
agent.id.to_string()
|
||||
} else {
|
||||
"shared".to_string()
|
||||
};
|
||||
let key = format!(
|
||||
"{}/{}/{}/{}",
|
||||
user.workspace_id,
|
||||
drive.as_str(),
|
||||
scope,
|
||||
query.path
|
||||
);
|
||||
let bytes = state
|
||||
.runtime
|
||||
.blob()
|
||||
.get(&key)
|
||||
.await
|
||||
.map_err(|_| ApiError::NotFound)?;
|
||||
Ok(String::from_utf8_lossy(&bytes).to_string())
|
||||
}
|
||||
|
||||
/// GET /api/shared-drive/files?clawId= — the team-wide ClawDrive (§7.4).
|
||||
pub async fn shared_files(
|
||||
State(state): State<AppState>,
|
||||
@@ -47,6 +170,9 @@ pub async fn shared_files(
|
||||
Query(query): Query<SharedQuery>,
|
||||
) -> Result<Json<Vec<FileNode>>, ApiError> {
|
||||
let agent = workspace_agent(&state, &user, query.claw_id).await?;
|
||||
if let Some(root) = &state.file_root {
|
||||
reconcile_drive(&state.pool, root, user.workspace_id, FileDrive::Shared, agent.id).await;
|
||||
}
|
||||
let nodes =
|
||||
cm_db::repo::files::list(&state.pool, user.workspace_id, FileDrive::Shared, agent.id)
|
||||
.await?;
|
||||
|
||||
@@ -21,4 +21,5 @@ pub mod slack;
|
||||
pub mod structure;
|
||||
pub mod team;
|
||||
pub mod teams;
|
||||
pub mod terminal;
|
||||
pub mod topology;
|
||||
|
||||
@@ -0,0 +1,302 @@
|
||||
//! The Terminal computer app: an interactive PTY (`zsh -l`) in the agent's
|
||||
//! themed terminal container, bridged to the browser's xterm.js over a
|
||||
//! WebSocket. Because a browser WS handshake can't carry the bearer header, the
|
||||
//! client first POSTs (through the authed proxy) for a short-lived single-use
|
||||
//! ticket, then opens the WS with `?ticket=`. The WS itself is routed straight
|
||||
//! to this server by the edge (Traefik), bypassing the HTTP-only Next proxy.
|
||||
|
||||
use std::collections::HashMap;
|
||||
use std::sync::{Arc, Mutex};
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
use axum::extract::ws::{Message, WebSocket, WebSocketUpgrade};
|
||||
use axum::extract::{Path, Query, State};
|
||||
use axum::http::StatusCode;
|
||||
use axum::response::{IntoResponse, Response};
|
||||
use axum::Json;
|
||||
use cm_db::repo::audit::Actor;
|
||||
use cm_domain::{AgentId, WorkspaceId};
|
||||
use futures::{SinkExt, StreamExt};
|
||||
use serde::{Deserialize, Serialize};
|
||||
use serde_json::json;
|
||||
use tokio::io::AsyncWriteExt;
|
||||
|
||||
use crate::routes::claws::workspace_agent;
|
||||
use crate::{ApiError, AppState, Authed};
|
||||
|
||||
const TICKET_TTL: Duration = Duration::from_secs(30);
|
||||
|
||||
/// In-memory single-use WS tickets (token → agent + greeting label + expiry).
|
||||
/// Tiny + ephemeral; a ticket is consumed on redeem, expired ones swept lazily.
|
||||
struct Ticket {
|
||||
agent_id: AgentId,
|
||||
workspace_id: WorkspaceId,
|
||||
/// Display name to greet the user with in the terminal MOTD.
|
||||
label: String,
|
||||
expires: Instant,
|
||||
}
|
||||
|
||||
#[derive(Clone, Default)]
|
||||
pub struct TerminalTickets {
|
||||
inner: Arc<Mutex<HashMap<String, Ticket>>>,
|
||||
}
|
||||
|
||||
impl TerminalTickets {
|
||||
pub fn issue(&self, agent_id: AgentId, workspace_id: WorkspaceId, label: String) -> String {
|
||||
let token = uuid::Uuid::new_v4().simple().to_string();
|
||||
let mut g = self.inner.lock().expect("tickets lock");
|
||||
let now = Instant::now();
|
||||
g.retain(|_, t| t.expires > now);
|
||||
g.insert(
|
||||
token.clone(),
|
||||
Ticket {
|
||||
agent_id,
|
||||
workspace_id,
|
||||
label,
|
||||
expires: now + TICKET_TTL,
|
||||
},
|
||||
);
|
||||
token
|
||||
}
|
||||
|
||||
/// Consume a ticket; returns its (workspace, greeting label) iff valid for this agent.
|
||||
pub fn redeem(&self, token: &str, agent_id: AgentId) -> Option<(WorkspaceId, String)> {
|
||||
let now = Instant::now();
|
||||
let mut g = self.inner.lock().expect("tickets lock");
|
||||
g.retain(|_, t| t.expires > now);
|
||||
match g.remove(token) {
|
||||
Some(t) if t.agent_id == agent_id && t.expires > now => Some((t.workspace_id, t.label)),
|
||||
_ => None,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Serialize)]
|
||||
pub struct TicketResponse {
|
||||
pub ticket: String,
|
||||
}
|
||||
|
||||
/// `POST /api/terminal/{id}/ticket` — owner-gated; mints a short-lived ticket.
|
||||
pub async fn ticket(
|
||||
State(state): State<AppState>,
|
||||
Authed(user): Authed,
|
||||
Path(agent_id): Path<AgentId>,
|
||||
) -> Result<Json<TicketResponse>, ApiError> {
|
||||
// Tenant isolation: a foreign agent looks non-existent.
|
||||
workspace_agent(&state, &user, agent_id).await?;
|
||||
// Greet by display name, falling back to the email local-part.
|
||||
let label = match cm_db::repo::users::get(&state.pool, user.user_id).await {
|
||||
Ok(u) if !u.display_name.trim().is_empty() => u.display_name.trim().to_string(),
|
||||
Ok(u) => u
|
||||
.email
|
||||
.split('@')
|
||||
.next()
|
||||
.filter(|s| !s.is_empty())
|
||||
.unwrap_or("there")
|
||||
.to_string(),
|
||||
Err(_) => "there".to_string(),
|
||||
};
|
||||
let ticket = state
|
||||
.terminal_tickets
|
||||
.issue(agent_id, user.workspace_id, label);
|
||||
cm_db::repo::audit::append(
|
||||
&state.pool,
|
||||
user.workspace_id,
|
||||
Actor::User(user.user_id),
|
||||
"terminal.ticket_issued",
|
||||
"agent",
|
||||
&agent_id.to_string(),
|
||||
json!({}),
|
||||
)
|
||||
.await?;
|
||||
Ok(Json(TicketResponse { ticket }))
|
||||
}
|
||||
|
||||
/// GET /api/terminal/{id}/tabs — the user's saved tab layout (or null).
|
||||
pub async fn get_tabs(
|
||||
State(state): State<AppState>,
|
||||
Authed(user): Authed,
|
||||
Path(agent_id): Path<AgentId>,
|
||||
) -> Result<Json<serde_json::Value>, ApiError> {
|
||||
workspace_agent(&state, &user, agent_id).await?;
|
||||
let layout = cm_db::repo::terminal_tabs::get(&state.pool, user.user_id, agent_id).await?;
|
||||
Ok(Json(layout.unwrap_or(serde_json::Value::Null)))
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
pub struct SaveTabsRequest {
|
||||
layout: serde_json::Value,
|
||||
}
|
||||
|
||||
/// PUT /api/terminal/{id}/tabs — save (replace) the user's tab layout, so the
|
||||
/// named tabs + their tmux sessions survive logout / a new device.
|
||||
pub async fn save_tabs(
|
||||
State(state): State<AppState>,
|
||||
Authed(user): Authed,
|
||||
Path(agent_id): Path<AgentId>,
|
||||
Json(body): Json<SaveTabsRequest>,
|
||||
) -> Result<StatusCode, ApiError> {
|
||||
workspace_agent(&state, &user, agent_id).await?;
|
||||
cm_db::repo::terminal_tabs::upsert(
|
||||
&state.pool,
|
||||
user.workspace_id,
|
||||
user.user_id,
|
||||
agent_id,
|
||||
&body.layout,
|
||||
)
|
||||
.await?;
|
||||
cm_db::repo::audit::append(
|
||||
&state.pool,
|
||||
user.workspace_id,
|
||||
Actor::User(user.user_id),
|
||||
"terminal.tabs_saved",
|
||||
"agent",
|
||||
&agent_id.to_string(),
|
||||
json!({}),
|
||||
)
|
||||
.await?;
|
||||
Ok(StatusCode::NO_CONTENT)
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
pub struct WsQuery {
|
||||
ticket: String,
|
||||
/// tmux session name → one per terminal tab (same container, shared drives).
|
||||
#[serde(default)]
|
||||
session: Option<String>,
|
||||
}
|
||||
|
||||
/// Keep tmux-session names safe ([A-Za-z0-9_-], <=24) — they all share the
|
||||
/// agent's one container, so different names = independent tabs.
|
||||
fn sanitize_session(raw: Option<&str>) -> String {
|
||||
let s: String = raw
|
||||
.unwrap_or("main")
|
||||
.chars()
|
||||
.filter(|c| c.is_ascii_alphanumeric() || *c == '-' || *c == '_')
|
||||
.take(24)
|
||||
.collect();
|
||||
if s.is_empty() {
|
||||
"main".to_string()
|
||||
} else {
|
||||
s
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
struct ControlMsg {
|
||||
#[serde(rename = "type")]
|
||||
kind: String,
|
||||
#[serde(default)]
|
||||
cols: u16,
|
||||
#[serde(default)]
|
||||
rows: u16,
|
||||
}
|
||||
|
||||
/// `GET /api/terminal/{id}/ws?ticket=…` — upgrades to a WebSocket bridged to an
|
||||
/// interactive PTY. Ticket-authed (no bearer header on a browser WS handshake).
|
||||
pub async fn ws(
|
||||
State(state): State<AppState>,
|
||||
Path(agent_id): Path<AgentId>,
|
||||
Query(q): Query<WsQuery>,
|
||||
upgrade: WebSocketUpgrade,
|
||||
) -> Response {
|
||||
let (workspace_id, label) = match state.terminal_tickets.redeem(&q.ticket, agent_id) {
|
||||
Some(v) => v,
|
||||
None => return (StatusCode::UNAUTHORIZED, "invalid or expired ticket").into_response(),
|
||||
};
|
||||
let tm = match state.runtime.terminals() {
|
||||
Some(tm) => tm,
|
||||
None => {
|
||||
return (
|
||||
StatusCode::SERVICE_UNAVAILABLE,
|
||||
"terminal engine unavailable",
|
||||
)
|
||||
.into_response()
|
||||
}
|
||||
};
|
||||
// MOTD greeting injected into the PTY session (read by the container .zshrc).
|
||||
let env = vec![format!("CLAWMATES_USER={label}")];
|
||||
let session = sanitize_session(q.session.as_deref());
|
||||
upgrade.on_upgrade(move |socket| bridge(socket, tm, workspace_id, agent_id, env, session))
|
||||
}
|
||||
|
||||
/// Pumps bytes both ways between the WebSocket and the container PTY. Binary
|
||||
/// frames are stdin; text frames are control JSON ({"type":"resize",cols,rows}).
|
||||
async fn bridge(
|
||||
socket: WebSocket,
|
||||
tm: Arc<cm_runtime::TerminalManager>,
|
||||
workspace_id: WorkspaceId,
|
||||
agent_id: AgentId,
|
||||
env: Vec<String>,
|
||||
tmux_session: String,
|
||||
) {
|
||||
let (mut ws_tx, mut ws_rx) = socket.split();
|
||||
let mut session = match tm
|
||||
.attach(workspace_id, agent_id, 80, 24, &env, &tmux_session)
|
||||
.await
|
||||
{
|
||||
Ok(s) => s,
|
||||
Err(e) => {
|
||||
let _ = ws_tx
|
||||
.send(Message::Text(
|
||||
format!("\r\n\x1b[31mterminal error: {e}\x1b[0m\r\n").into(),
|
||||
))
|
||||
.await;
|
||||
tm.detach(agent_id).await;
|
||||
return;
|
||||
}
|
||||
};
|
||||
let exec_id = session.exec_id.clone();
|
||||
|
||||
// PTY → WS: stream the shell's combined TTY output as binary frames.
|
||||
let pty_to_ws = async {
|
||||
while let Some(chunk) = session.output.next().await {
|
||||
match chunk {
|
||||
Ok(bytes) => {
|
||||
if ws_tx.send(Message::Binary(bytes.into())).await.is_err() {
|
||||
break;
|
||||
}
|
||||
}
|
||||
Err(_) => break,
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
// WS → PTY: binary = keystrokes; text = control (resize).
|
||||
let ws_to_pty = async {
|
||||
while let Some(msg) = ws_rx.next().await {
|
||||
let msg = match msg {
|
||||
Ok(m) => m,
|
||||
Err(_) => break,
|
||||
};
|
||||
match msg {
|
||||
Message::Binary(b) => {
|
||||
if session.input.write_all(&b).await.is_err() {
|
||||
break;
|
||||
}
|
||||
let _ = session.input.flush().await;
|
||||
}
|
||||
Message::Text(t) => {
|
||||
if let Ok(ctrl) = serde_json::from_str::<ControlMsg>(t.as_str()) {
|
||||
if ctrl.kind == "resize" && ctrl.cols > 0 && ctrl.rows > 0 {
|
||||
let _ = tm.resize(&exec_id, ctrl.cols, ctrl.rows).await;
|
||||
continue;
|
||||
}
|
||||
}
|
||||
if session.input.write_all(t.as_bytes()).await.is_err() {
|
||||
break;
|
||||
}
|
||||
let _ = session.input.flush().await;
|
||||
}
|
||||
Message::Close(_) => break,
|
||||
_ => {}
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
tokio::select! {
|
||||
_ = pty_to_ws => {}
|
||||
_ = ws_to_pty => {}
|
||||
}
|
||||
tm.detach(agent_id).await;
|
||||
}
|
||||
Reference in New Issue
Block a user