diff --git a/app/README.md b/app/README.md index 1aa18dc3..774d5c01 100644 --- a/app/README.md +++ b/app/README.md @@ -30,6 +30,42 @@ Prerequisite either way: agmsg itself installed at `~/.agents/skills/agmsg` (the app reads its DB and team config from there) — see the [main README](../README.md) for that install. +## Multiple message sources + +The Team Room can merge the local agmsg history with one or more read-only +agmsg installations reached over SSH. Each source keeps its own SQLite store; +the app queries its public `api.sh` interface and never mounts, copies, or +synchronizes `messages.db`. + +Add remote sources in `~/.agmsg/config/app-sources.json`: + +```json +{ + "sources": [ + { + "id": "mac", + "label": "Mac", + "ssh": "you@mac.local", + "base": "/Users/you/.agents/skills/agmsg" + } + ] +} +``` + +`id` must be unique and contain only letters, numbers, `-`, or `_`; `local` is +reserved for the implicit local source. `base` must be an absolute remote path. +The system `ssh` client and its existing config/agent are used with +`BatchMode=yes`; the app does not store keys or passwords. Set +`AGMSG_APP_SOURCES` to use a different config-file path. + +Messages from all sources with the selected team name are merged by timestamp +and carry a source badge. History pagination and live cursors remain independent +per source, so overlapping numeric SQLite ids do not collide. A remote core that +predates `api.sh --after-id` falls back to merging its latest 100 messages on +each poll; upgrading that core enables lossless forward-cursor polling. Remote +members are shown for filtering but are read-only: spawning, renaming, leaving, +and the composer continue to operate on the local source only. + ## Strategic core — universal stdin-inject delivery The app **owns** each spawned agent's pseudo-terminal. When a new agmsg message diff --git a/app/src-tauri/src/agmsg.rs b/app/src-tauri/src/agmsg.rs index bf9f3ff7..0b810cce 100644 --- a/app/src-tauri/src/agmsg.rs +++ b/app/src-tauri/src/agmsg.rs @@ -6,6 +6,7 @@ // read-only feed, plus the left-hand member list. use std::path::PathBuf; +use std::process::Command; use std::thread; use std::time::Duration; @@ -198,6 +199,166 @@ fn bash_command() -> Result { Ok(cmd) } +const LOCAL_SOURCE_ID: &str = "local"; +const LOCAL_SOURCE_LABEL: &str = "Local"; + +#[derive(Clone, Serialize)] +pub struct MessageSource { + pub id: String, + pub label: String, + pub read_only: bool, + #[serde(skip_serializing)] + transport: SourceTransport, +} + +#[derive(Clone)] +enum SourceTransport { + Local, + Ssh { target: String, base: String }, +} + +#[derive(Deserialize)] +struct SourcesFile { + #[serde(default)] + sources: Vec, +} + +#[derive(Deserialize)] +struct SshSourceConfig { + id: String, + label: String, + ssh: String, + base: String, +} + +fn sources_config_path() -> PathBuf { + if let Ok(over) = std::env::var("AGMSG_APP_SOURCES") { + if !over.is_empty() { + return PathBuf::from(over); + } + } + let home = home_dir_string().unwrap_or_else(|| ".".into()); + PathBuf::from(home).join(".agmsg/config/app-sources.json") +} + +fn valid_source_id(id: &str) -> bool { + !id.is_empty() + && id != LOCAL_SOURCE_ID + && id + .chars() + .all(|c| c.is_ascii_alphanumeric() || c == '-' || c == '_') +} + +fn message_sources() -> Result, String> { + let mut sources = vec![MessageSource { + id: LOCAL_SOURCE_ID.into(), + label: LOCAL_SOURCE_LABEL.into(), + read_only: false, + transport: SourceTransport::Local, + }]; + let path = sources_config_path(); + if !path.is_file() { + return Ok(sources); + } + let raw = std::fs::read_to_string(&path) + .map_err(|e| format!("Couldn't read {}: {e}", path.display()))?; + let configured: SourcesFile = serde_json::from_str(&raw) + .map_err(|e| format!("Invalid {}: {e}", path.display()))?; + let mut seen = std::collections::HashSet::from([LOCAL_SOURCE_ID.to_string()]); + for source in configured.sources { + if !valid_source_id(&source.id) { + return Err(format!( + "Invalid source id {:?} in {} (use letters, numbers, '-' or '_'; 'local' is reserved)", + source.id, + path.display() + )); + } + if !seen.insert(source.id.clone()) { + return Err(format!("Duplicate source id {:?} in {}", source.id, path.display())); + } + if source.label.trim().is_empty() + || source.ssh.trim().is_empty() + || !source.base.starts_with('/') + { + return Err(format!( + "Source {:?} in {} needs non-empty label/ssh and an absolute remote base path", + source.id, + path.display() + )); + } + sources.push(MessageSource { + id: source.id, + label: source.label, + read_only: true, + transport: SourceTransport::Ssh { + target: source.ssh, + base: source.base, + }, + }); + } + Ok(sources) +} + +fn source_by_id(id: &str) -> Result { + message_sources()? + .into_iter() + .find(|source| source.id == id) + .ok_or_else(|| format!("Unknown agmsg source: {id}")) +} + +fn shell_quote(value: &str) -> String { + format!("'{}'", value.replace('\'', "'\"'\"'")) +} + +fn ssh_command() -> Command { + let mut cmd = Command::new("ssh"); + cmd.args([ + "-o", + "BatchMode=yes", + "-o", + "ConnectTimeout=3", + "-o", + "ServerAliveInterval=2", + "-o", + "ServerAliveCountMax=1", + ]); + #[cfg(target_os = "windows")] + { + use std::os::windows::process::CommandExt; + const CREATE_NO_WINDOW: u32 = 0x08000000; + cmd.creation_flags(CREATE_NO_WINDOW); + } + if let Some(path) = crate::imported_path() { + cmd.env("PATH", path); + } + cmd +} + +fn run_source_script(source: &MessageSource, name: &str, args: &[&str]) -> Result { + match &source.transport { + SourceTransport::Local => run_script(name, args), + SourceTransport::Ssh { target, base } => { + let script = format!("{base}/scripts/{name}"); + let remote_command = std::iter::once(script.as_str()) + .chain(args.iter().copied()) + .map(shell_quote) + .collect::>() + .join(" "); + let output = ssh_command() + .arg(target) + .arg(remote_command) + .output() + .map_err(|e| format!("Couldn't run ssh for source {}: {e}", source.label))?; + if output.status.success() { + Ok(String::from_utf8_lossy(&output.stdout).into_owned()) + } else { + let stderr = String::from_utf8_lossy(&output.stderr); + Err(format!("{}: {}", source.label, stderr.trim())) + } + } + } +} + fn db_path() -> PathBuf { agmsg_base().join("db/messages.db") } @@ -212,7 +373,9 @@ fn open_ro() -> Result { #[derive(Clone, Serialize)] pub struct Message { - pub id: i64, + pub id: String, + pub source_id: String, + pub source_label: String, pub team: String, pub from: String, pub to: String, @@ -222,6 +385,9 @@ pub struct Message { #[derive(Clone, Serialize)] pub struct Member { + pub source_id: String, + pub source_label: String, + pub read_only: bool, pub name: String, /// Agent types registered under this name (claude-code, codex, ...). pub types: Vec, @@ -378,8 +544,8 @@ struct ApiMember { /// that lands, not what needs to change. `id` is a JSON *string* on the /// wire — api.sh CASTs it, since the driver interface treats every message /// id as opaque (a legacy sqlite int today, potentially a UUIDv7 or -/// Redis-stream-id tomorrow) — parsed back to `i64` below for `Message`, -/// which is a Tauri-IPC-only contract with the frontend, not agmsg's. +/// Redis-stream-id tomorrow). `Message` preserves that opaque string and +/// adds source metadata for its Tauri IPC contract with the frontend. #[derive(Deserialize)] struct ApiMessage { id: String, @@ -539,6 +705,83 @@ pub fn agmsg_teams() -> Result, String> { } /// Members of a team, via `api.sh get teams members`. +#[tauri::command] +pub fn agmsg_sources() -> Result, String> { + message_sources() +} + +#[tauri::command] +pub fn agmsg_source_teams(source_id: String) -> Result, String> { + let source = source_by_id(&source_id)?; + let raw = run_source_script(&source, "api.sh", &["get", "teams"])?; + let mut teams: Vec = + parse_jsonl::(&raw).into_iter().map(|t| t.name).collect(); + teams.sort(); + Ok(teams) +} + +#[tauri::command] +pub fn agmsg_source_members(source_id: String, team: String) -> Result, String> { + let source = source_by_id(&source_id)?; + let raw = run_source_script(&source, "api.sh", &["get", "teams", &team, "members"])?; + let mut members: Vec = parse_jsonl::(&raw) + .into_iter() + .map(|m| { + let mut types = m.types; + types.sort(); + types.dedup(); + Member { + source_id: source.id.clone(), + source_label: source.label.clone(), + read_only: source.read_only, + name: m.name, + types, + project: m.project.unwrap_or_default(), + } + }) + .collect(); + members.sort_by(|a, b| a.name.cmp(&b.name)); + Ok(members) +} + +#[tauri::command] +pub fn agmsg_source_messages( + source_id: String, + team: String, + limit: Option, + before_id: Option, + after_id: Option, +) -> Result, String> { + if before_id.is_some() && after_id.is_some() { + return Err("beforeId and afterId are mutually exclusive".into()); + } + let source = source_by_id(&source_id)?; + let limit_s = limit.unwrap_or(30).to_string(); + let mut args = vec!["get", "teams", &team, "messages", "--limit", &limit_s]; + if let Some(id) = before_id.as_deref() { + args.push("--before-id"); + args.push(id); + } + if let Some(id) = after_id.as_deref() { + args.push("--after-id"); + args.push(id); + } + let raw = run_source_script(&source, "api.sh", &args)?; + Ok(parse_jsonl::(&raw) + .into_iter() + .map(|m| Message { + id: m.id, + source_id: source.id.clone(), + source_label: source.label.clone(), + team: m.team, + from: m.from, + to: m.to, + body: m.body, + created_at: m.created_at, + }) + .collect()) +} + #[tauri::command] pub fn agmsg_members(team: String) -> Result, String> { let raw = run_script("api.sh", &["get", "teams", &team, "members"])?; @@ -548,7 +791,14 @@ pub fn agmsg_members(team: String) -> Result, String> { let mut types = m.types; types.sort(); types.dedup(); - Member { name: m.name, types, project: m.project.unwrap_or_default() } + Member { + source_id: LOCAL_SOURCE_ID.into(), + source_label: LOCAL_SOURCE_LABEL.into(), + read_only: false, + name: m.name, + types, + project: m.project.unwrap_or_default(), + } }) .collect(); members.sort_by(|a, b| a.name.cmp(&b.name)); @@ -578,14 +828,16 @@ pub fn agmsg_messages( let raw = run_script("api.sh", &args)?; Ok(parse_jsonl::(&raw) .into_iter() - .filter_map(|m| Some(Message { - id: m.id.parse().ok()?, + .map(|m| Message { + id: m.id, + source_id: LOCAL_SOURCE_ID.into(), + source_label: LOCAL_SOURCE_LABEL.into(), team: m.team, from: m.from, to: m.to, body: m.body, created_at: m.created_at, - })) + }) .collect()) } @@ -714,8 +966,11 @@ pub fn start_watcher(app: AppHandle) { Err(_) => return, }; let mapped = stmt.query_map(rusqlite::params![last_id], |r| { + let id: i64 = r.get(0)?; Ok(Message { - id: r.get(0)?, + id: id.to_string(), + source_id: LOCAL_SOURCE_ID.into(), + source_label: LOCAL_SOURCE_LABEL.into(), team: r.get(1)?, from: r.get(2)?, to: r.get(3)?, @@ -729,7 +984,9 @@ pub fn start_watcher(app: AppHandle) { } }; for m in new_rows { - last_id = m.id.max(last_id); + if let Ok(id) = m.id.parse::() { + last_id = id.max(last_id); + } let _ = app.emit("agmsg-message", m); } thread::sleep(Duration::from_millis(800)); @@ -739,7 +996,10 @@ pub fn start_watcher(app: AppHandle) { #[cfg(test)] mod tests { - use super::{agmsg_base, msys_to_native, parse_semver, run_script, to_bash_slashes}; + use super::{ + agmsg_base, message_sources, msys_to_native, parse_semver, run_script, shell_quote, + to_bash_slashes, valid_source_id, + }; use serial_test::serial; use std::io::Write; @@ -902,6 +1162,39 @@ mod tests { FakeBase { _dir: dir, _env: env } } + #[test] + fn source_ids_are_safe_config_keys() { + assert!(valid_source_id("mac-studio_2")); + assert!(!valid_source_id("")); + assert!(!valid_source_id("local")); + assert!(!valid_source_id("mac studio")); + assert!(!valid_source_id("../mac")); + } + + #[test] + fn ssh_arguments_are_single_quoted() { + assert_eq!(shell_quote("plain"), "'plain'"); + assert_eq!(shell_quote("it's safe"), "'it'\"'\"'s safe'"); + } + + #[test] + #[serial] + fn source_config_adds_read_only_ssh_sources() { + let dir = tempfile::tempdir().unwrap(); + let config = dir.path().join("sources.json"); + std::fs::write( + &config, + r#"{"sources":[{"id":"mac","label":"Mac","ssh":"you@mac","base":"/Users/you/.agents/skills/agmsg"}]}"#, + ) + .unwrap(); + let _env = EnvGuard::set("AGMSG_APP_SOURCES", &config.to_string_lossy()); + let sources = message_sources().unwrap(); + assert_eq!(sources.len(), 2); + assert_eq!(sources[0].id, "local"); + assert_eq!(sources[1].id, "mac"); + assert!(sources[1].read_only); + } + #[test] #[serial] fn agmsg_base_honors_the_env_override() { @@ -952,6 +1245,34 @@ mod tests { assert!(run_script("nope.sh", &[]).is_err()); } + #[test] + #[serial] + fn source_messages_parse_the_public_api_shape() { + let _base = fake_base(&[( + "api.sh", + r#"cat <<'JSON' +{"type":"message_sent","id":"42","team":"yuzu","from":"alice","to":"bob","body":"hello","at":"2026-07-14T00:00:00Z"} +JSON +"#, + )]); + let missing_config = _base._dir.path().join("no-sources.json"); + let _sources = EnvGuard::set("AGMSG_APP_SOURCES", &missing_config.to_string_lossy()); + + let messages = super::agmsg_source_messages( + "local".into(), + "yuzu".into(), + Some(2), + None, + None, + ) + .expect("public message API should parse"); + + assert_eq!(messages.len(), 1); + assert_eq!(messages[0].id, "42"); + assert_eq!(messages[0].source_id, "local"); + assert_eq!(messages[0].body, "hello"); + } + // --- #315 Windows spawn-path regression (runs on the windows-latest job) --- /// The core #315 guarantee, independent of bash: create_dir_all runs before diff --git a/app/src-tauri/src/lib.rs b/app/src-tauri/src/lib.rs index 0b333fef..debf4d49 100644 --- a/app/src-tauri/src/lib.rs +++ b/app/src-tauri/src/lib.rs @@ -598,6 +598,10 @@ pub fn run() { agmsg::agmsg_core_version_status, agmsg::agmsg_update_core, agmsg::agmsg_teams, + agmsg::agmsg_sources, + agmsg::agmsg_source_teams, + agmsg::agmsg_source_members, + agmsg::agmsg_source_messages, agmsg::agmsg_members, agmsg::agmsg_messages, agmsg::agmsg_send, diff --git a/app/src/App.css b/app/src/App.css index 2888297d..18e29b66 100644 --- a/app/src/App.css +++ b/app/src/App.css @@ -1433,3 +1433,16 @@ body.resizing-row { opacity: 0.4; cursor: default; } + +.source-badge { + color: var(--muted); + font-size: 0.72rem; + border: 1px solid var(--border); + border-radius: 999px; + padding: 1px 6px; +} + +.member:disabled { + cursor: default; + opacity: 0.72; +} diff --git a/app/src/App.tsx b/app/src/App.tsx index fb2f98dd..ec9bb036 100644 --- a/app/src/App.tsx +++ b/app/src/App.tsx @@ -43,16 +43,22 @@ import { type PaneRect, type SplitNode, } from "./paneTree"; +import { + isAfterIdUnsupported, + mergeMessages, + messageKey, + type MessageSource, + type SourceMember, + type SourceMessage, +} from "./messageSources"; import "./App.css"; -export type Member = { name: string; types: string[]; project: string }; -type Message = { - id: number; - team: string; - from: string; - to: string; - body: string; - created_at: string; +export type Member = SourceMember; +type Message = SourceMessage; +type SourceHistory = { + oldestId?: string; + latestId?: string; + hasMore: boolean; }; type Pane = { id: string; @@ -142,10 +148,12 @@ export default function App() { // otherwise completes silently (the outdated banner just disappears), // which read as "did that actually work?" in testing. const [coreUpdateSucceeded, setCoreUpdateSucceeded] = useState(null); + const [sources, setSources] = useState([]); const [teams, setTeams] = useState([]); const [team, setTeam] = useState(""); const [members, setMembers] = useState([]); const [messages, setMessages] = useState([]); + const [, setSourceHistoryState] = useState>({}); const [panes, setPanes] = useState([]); const [paneStatus, setPaneStatus] = useState({}); const [windows, setWindows] = useState([]); @@ -270,6 +278,18 @@ export default function App() { renameDraftRef.current = renameDraft; const seq = useRef(0); const feedRef = useRef(null); + const sourceHistoryRef = useRef>({}); + const legacyRemoteSourcesRef = useRef>(new Set()); + const updateSourceHistory = useCallback( + (updater: (current: Record) => Record) => { + setSourceHistoryState((current) => { + const next = updater(current); + sourceHistoryRef.current = next; + return next; + }); + }, + [], + ); // Set right before prepending an older history page, so the scroll-to- // bottom effect below skips that render (loadOlderMessages restores the // scroll position itself instead). @@ -305,14 +325,19 @@ export default function App() { }, []); // The app user = the member registered with the agmsg-app type (one per team). - const appUserMember = members.find((m) => m.types.includes(APP_USER_TYPE)); + const appUserMember = members.find( + (m) => m.source_id === "local" && m.types.includes(APP_USER_TYPE), + ); const appUser = appUserMember?.name ?? ""; // The team's project dir (the app-user's) — new agents default into the same place. const teamProject = appUserMember?.project ?? ""; // Everyone else is a spawnable/messageable agent. const others = members.filter((m) => !m.types.includes(APP_USER_TYPE)); + const sendableOthers = others.filter((m) => !m.read_only); // The app user's own send/receive thread. - const myThread = messages.filter((m) => m.from === appUser || m.to === appUser); + const myThread = messages.filter( + (m) => m.source_id === "local" && (m.from === appUser || m.to === appUser), + ); // Team-room member filter: keep a message when either party is a checked // member. Names not in the roster (e.g. the app-user) ride on their counterpart. @@ -326,11 +351,28 @@ export default function App() { // Cozy grouping: collapse runs of consecutive messages with the same from→to // into one header + stacked bodies (Slack/Discord style), so short bursts stay // light while long messages still line up. - const groups: { key: number; from: string; to: string; items: Message[] }[] = []; + const groups: { + key: string; + source_id: string; + source_label: string; + from: string; + to: string; + items: Message[]; + }[] = []; for (const m of roomMessages) { const last = groups[groups.length - 1]; - if (last && last.from === m.from && last.to === m.to) last.items.push(m); - else groups.push({ key: m.id, from: m.from, to: m.to, items: [m] }); + if (last && last.source_id === m.source_id && last.from === m.from && last.to === m.to) { + last.items.push(m); + } else { + groups.push({ + key: messageKey(m), + source_id: m.source_id, + source_label: m.source_label, + from: m.from, + to: m.to, + items: [m], + }); + } } const toggleMember = (name: string) => @@ -343,17 +385,55 @@ export default function App() { const selectAllMembers = () => setDeselected(new Set()); const selectNoMembers = () => setDeselected(new Set(others.map((m) => m.name))); - const loadTeams = useCallback(async () => { - const t = await invoke("agmsg_teams"); - setTeams(t); - return t; + const loadSources = useCallback(async () => { + const loaded = await invoke("agmsg_sources"); + setSources(loaded); + return loaded; }, []); - const loadMembers = useCallback(async (t: string) => { - const m = await invoke("agmsg_members", { team: t }); - setMembers(m); - return m; - }, []); + const loadTeams = useCallback( + async (activeSources: MessageSource[]) => { + const results = await Promise.allSettled( + activeSources.map((source) => + invoke("agmsg_source_teams", { sourceId: source.id }), + ), + ); + const loaded = results.flatMap((result, index) => { + if (result.status === "fulfilled") return result.value; + console.warn("Couldn't load teams from " + activeSources[index].label + ":", result.reason); + return []; + }); + if (activeSources.length > 0 && results.every((result) => result.status === "rejected")) { + throw results[0].status === "rejected" ? results[0].reason : new Error("No agmsg sources"); + } + const unique = [...new Set(loaded)].sort(); + setTeams(unique); + return unique; + }, + [], + ); + + const loadMembers = useCallback( + async (t: string) => { + const results = await Promise.allSettled( + sources.map((source) => + invoke("agmsg_source_members", { sourceId: source.id, team: t }), + ), + ); + const loaded = results.flatMap((result, index) => { + if (result.status === "fulfilled") return result.value; + console.warn("Couldn't load members from " + sources[index].label + ":", result.reason); + return []; + }); + loaded.sort( + (left, right) => + left.name.localeCompare(right.name) || left.source_id.localeCompare(right.source_id), + ); + setMembers(loaded); + return loaded; + }, + [sources], + ); // The agmsg slash-command name, for the `/ actas ` boot prompt. useEffect(() => { @@ -422,7 +502,8 @@ export default function App() { } catch (err) { console.error(err); } - const loadedTeams = await loadTeams(); + const loadedSources = await loadSources(); + const loadedTeams = await loadTeams(loadedSources); if (loadedTeams.length === 0) setModal({ kind: "team", firstRun: true }); else { const lastTeam = localStorage.getItem(LAST_TEAM_KEY); @@ -435,7 +516,7 @@ export default function App() { } } boot(); - }, [loadTeams, t]); + }, [loadSources, loadTeams, t]); // Tabs are per-team (see the Window type), so switching teams also swaps // the whole visible tab set — land on the team room rather than leaving @@ -452,33 +533,79 @@ export default function App() { if (team) localStorage.setItem(LAST_TEAM_KEY, team); }, [team]); - // On team change: load members + the most recent history page. Prompt to - // add an app-user if missing. + // On team change: load members and one independent history page from + // every configured source, then merge without trimming. Keeping each + // source's full page is what makes per-database backward cursors lossless. useEffect(() => { - if (!team) return; - setDeselected(new Set()); // reset the room filter when switching teams + if (!team || sources.length === 0) return; + let cancelled = false; + setDeselected(new Set()); setMessages([]); setHasMoreHistory(true); - invoke("agmsg_messages", { team, limit: ROOM_PAGE_SIZE }) - .then((msgs) => { - setMessages(msgs); - setHasMoreHistory(msgs.length >= ROOM_PAGE_SIZE); - }) - .catch(console.error); + + async function loadInitialHistory() { + const results = await Promise.allSettled( + sources.map((source) => + invoke("agmsg_source_messages", { + sourceId: source.id, + team, + limit: ROOM_PAGE_SIZE, + }), + ), + ); + if (cancelled) return; + + const history: Record = {}; + const loaded = results.flatMap((result, index) => { + const source = sources[index]; + if (result.status === "rejected") { + console.warn("Couldn't load messages from " + source.label + ":", result.reason); + history[source.id] = { hasMore: false }; + return []; + } + const page = result.value; + history[source.id] = { + oldestId: page[0]?.id, + latestId: page[page.length - 1]?.id, + hasMore: page.length >= ROOM_PAGE_SIZE, + }; + return page; + }); + setMessages(mergeMessages([], loaded)); + updateSourceHistory(() => history); + setHasMoreHistory(Object.values(history).some((state) => state.hasMore)); + } + + void loadInitialHistory(); loadMembers(team) - .then((m) => { - if (!m.some((x) => x.types.includes(APP_USER_TYPE))) { - setModal((cur) => cur ?? { kind: "appuser" }); + .then((loaded) => { + if (!loaded.some((member) => member.source_id === "local" && member.types.includes(APP_USER_TYPE))) { + setModal((current) => current ?? { kind: "appuser" }); } }) .catch(console.error); - }, [team, loadMembers]); + + return () => { + cancelled = true; + }; + }, [team, sources, loadMembers, updateSourceHistory]); // Live team-room updates; inject into a matching pane. useEffect(() => { const p = listen("agmsg-message", (e) => { if (e.payload.team !== team) return; - setMessages((prev) => [...prev, e.payload]); + setMessages((prev) => mergeMessages(prev, [e.payload])); + updateSourceHistory((current) => { + const local = current.local ?? { hasMore: false }; + return { + ...current, + local: { + ...local, + oldestId: local.oldestId ?? e.payload.id, + latestId: e.payload.id, + }, + }; + }); // Only inject into NON-native panes; a native (actas-booted) agent runs // its own agmsg monitor and would otherwise receive the message twice. const pane = panesRef.current.find((pn) => pn.label === e.payload.to && !pn.native); @@ -505,7 +632,90 @@ export default function App() { } }); return () => void p.then((u) => u()); - }, [team, cmdName]); + }, [team, cmdName, updateSourceHistory]); + + // Remote sources are read-only room feeds. Use a forward cursor where the + // core supports it and drain full pages before sleeping; older cores fall + // back to repeatedly merging their latest page for compatibility. + useEffect(() => { + const remoteSources = sources.filter((source) => source.read_only); + if (!team || remoteSources.length === 0) return; + + let stopped = false; + let running = false; + const poll = async () => { + if (running || stopped) return; + running = true; + const mergePage = (source: MessageSource, page: Message[]) => { + if (page.length === 0) return; + setMessages((current) => mergeMessages(current, page)); + const latestId = page[page.length - 1].id; + updateSourceHistory((current) => ({ + ...current, + [source.id]: { + ...(current[source.id] ?? { hasMore: false }), + latestId, + }, + })); + }; + + try { + for (const source of remoteSources) { + try { + if (legacyRemoteSourcesRef.current.has(source.id)) { + const page = await invoke("agmsg_source_messages", { + sourceId: source.id, + team, + limit: 100, + }); + mergePage(source, page); + continue; + } + + let afterId = sourceHistoryRef.current[source.id]?.latestId ?? "0"; + for (let pageNumber = 0; pageNumber < 20 && !stopped; pageNumber += 1) { + const page = await invoke("agmsg_source_messages", { + sourceId: source.id, + team, + limit: 100, + afterId, + }); + if (page.length === 0) break; + mergePage(source, page); + afterId = page[page.length - 1].id; + if (page.length < 100) break; + } + } catch (error) { + if (isAfterIdUnsupported(error)) { + legacyRemoteSourcesRef.current.add(source.id); + console.warn( + `${source.label} uses a legacy agmsg core; polling its latest 100 messages`, + ); + const page = await invoke("agmsg_source_messages", { + sourceId: source.id, + team, + limit: 100, + }); + mergePage(source, page); + } else { + console.warn(`Couldn't poll remote agmsg source ${source.label}:`, error); + } + } + } + } catch (error) { + console.warn("Couldn't poll remote agmsg sources:", error); + } finally { + running = false; + } + }; + + void poll(); + const timer = window.setInterval(() => void poll(), 2500); + return () => { + stopped = true; + window.clearInterval(timer); + }; + }, [team, sources, updateSourceHistory]); useEffect(() => { const stateListener = listen<{ id: string; state: RawState }>("agent-state", (event) => { @@ -531,30 +741,66 @@ export default function App() { // height, since scrollTop stays fixed while scrollHeight grows above it). const loadOlderMessages = useCallback(async () => { if (loadingHistory || !hasMoreHistory || messages.length === 0) return; + const pagedSources = sources.filter((source) => { + const state = sourceHistoryRef.current[source.id]; + return state?.hasMore && state.oldestId; + }); + if (pagedSources.length === 0) { + setHasMoreHistory(false); + return; + } + setLoadingHistory(true); - const beforeId = messages[0].id; const el = feedRef.current; const prevScrollHeight = el?.scrollHeight ?? 0; try { - const older = await invoke("agmsg_messages", { - team, - limit: ROOM_PAGE_SIZE, - beforeId, + const results = await Promise.allSettled( + pagedSources.map((source) => + invoke("agmsg_source_messages", { + sourceId: source.id, + team, + limit: ROOM_PAGE_SIZE, + beforeId: sourceHistoryRef.current[source.id].oldestId, + }), + ), + ); + const nextHistory = { ...sourceHistoryRef.current }; + const older = results.flatMap((result, index) => { + const source = pagedSources[index]; + if (result.status === "rejected") { + console.warn("Couldn't load older messages from " + source.label + ":", result.reason); + return []; + } + const page = result.value; + nextHistory[source.id] = { + ...nextHistory[source.id], + oldestId: page[0]?.id ?? nextHistory[source.id]?.oldestId, + hasMore: page.length >= ROOM_PAGE_SIZE, + }; + return page; }); + updateSourceHistory(() => nextHistory); + setHasMoreHistory(Object.values(nextHistory).some((state) => state.hasMore)); if (older.length > 0) { isPrependingRef.current = true; - setMessages((prev) => [...older, ...prev]); + setMessages((current) => mergeMessages(current, older)); requestAnimationFrame(() => { if (el) el.scrollTop += el.scrollHeight - prevScrollHeight; }); } - setHasMoreHistory(older.length >= ROOM_PAGE_SIZE); - } catch (err) { - console.error(err); + } catch (error) { + console.error(error); } finally { setLoadingHistory(false); } - }, [team, messages, loadingHistory, hasMoreHistory]); + }, [ + team, + sources, + messages.length, + loadingHistory, + hasMoreHistory, + updateSourceHistory, + ]); // useLayoutEffect (not useEffect): runs synchronously right after the DOM // updates and before the browser paints, so scrollHeight already reflects @@ -595,6 +841,7 @@ export default function App() { // targetWindowId: spawn straight into an existing tab as an extra // side-by-side pane, instead of opening a new tab (the default). async (m: Member, targetWindowId?: string) => { + if (m.read_only) return; // Don't spawn a second pane for a member that's already running — just // focus its window (one live agent per identity). const existing = panesRef.current.find((p) => p.label === m.name); @@ -884,7 +1131,7 @@ export default function App() { agentType: APP_USER_TYPE, project, }); - await loadTeams(); + await loadTeams(sources); setTeam(name); setModal(null); }, @@ -904,7 +1151,7 @@ export default function App() { async (name: string, type: string, project: string) => { await invoke("agmsg_join", { team, name, agentType: type, project }); const m = await loadMembers(team); - const added = m.find((x) => x.name === name); + const added = m.find((x) => x.source_id === "local" && x.name === name); if (added) spawnMember(added); setModal(null); }, @@ -1101,7 +1348,7 @@ export default function App() { await invoke("agmsg_update_core"); setCoreOutdated(null); setCoreUpdateSucceeded(targetVersion); - await loadTeams(); + await loadTeams(sources); } catch (err) { console.error(err); setStartupError(t("startupError.updateFailed", { error: String(err) })); @@ -1329,20 +1576,28 @@ export default function App() {
    {others.map((m) => { - const pane = panes.find( - (candidate) => - candidate.label === m.name && - windows.some( - (window) => - window.team === team && leaves(window.root).includes(candidate.id), - ), - ); + // Panes are local-only, and member names are unique per + // source rather than globally: a remote source can carry a + // member with the same name as a local one (e.g. claude-win + // on both Windows and Mac). Matching on the name alone would + // show the local pane's status on the remote twin. + const pane = m.read_only + ? undefined + : panes.find( + (candidate) => + candidate.label === m.name && + windows.some( + (window) => + window.team === team && leaves(window.root).includes(candidate.id), + ), + ); const status = pane ? (paneStatus[pane.id]?.state ?? "unknown") : null; return (
  • { + if (m.read_only) return; e.preventDefault(); e.stopPropagation(); closeAllMenus(); @@ -1365,15 +1620,21 @@ export default function App() { )} @@ -1533,10 +1794,13 @@ export default function App() { {g.from} → {g.to} + {sources.length > 1 && ( + {g.source_label} + )} {g.items[0].created_at.slice(11, 19)} {g.items.map((m) => ( -
    +
    {m.body}
    ))} @@ -1830,8 +2094,8 @@ export default function App() {