From 9168b16692c61723dac1c05cdbb6c004f758cd66 Mon Sep 17 00:00:00 2001 From: fumi Date: Tue, 14 Jul 2026 08:43:29 +0900 Subject: [PATCH 1/4] feat(api): add lossless forward message pagination Add --after-id to the team messages endpoint so polling clients can fetch rows strictly after a cursor. Forward pages select the oldest matching rows before emitting them in ascending order, preventing gaps when more than the requested limit arrives between polls; backward history keeps its existing newest-page behavior. Reject simultaneous before/after cursors and cover both cursor exclusivity and multi-page forward ordering in the API tests. --- scripts/api.sh | 24 +++++++++++++++++++++--- tests/test_api.bats | 21 +++++++++++++++++++++ 2 files changed, 42 insertions(+), 3 deletions(-) diff --git a/scripts/api.sh b/scripts/api.sh index e4e77f09..2578333e 100755 --- a/scripts/api.sh +++ b/scripts/api.sh @@ -26,7 +26,8 @@ set -euo pipefail # Usage: # api.sh get teams # api.sh get teams members -# api.sh get teams messages [--agent ] [--limit N] [--before-id ] +# api.sh get teams messages [--agent ] [--limit N] +# [--before-id | --after-id ] # # Output is always JSONL — one JSON object per line, UTF-8, no # pretty-printing — for every resource, including `teams` (a uniform @@ -95,12 +96,13 @@ get_members() { get_messages() { local team="$1" shift - local agent="" limit=30 before_id="" + local agent="" limit=30 before_id="" after_id="" while [ $# -gt 0 ]; do case "$1" in --agent) agent="${2:?--agent needs a value}"; shift 2 ;; --limit) limit="${2:?--limit needs a value}"; shift 2 ;; --before-id) before_id="${2:?--before-id needs a value}"; shift 2 ;; + --after-id) after_id="${2:?--after-id needs a value}"; shift 2 ;; *) echo "Unknown option: $1" >&2; exit 1 ;; esac done @@ -108,6 +110,11 @@ get_messages() { # same guard history.sh uses for LIMIT. case "$limit" in ''|*[!0-9]*) limit=30 ;; esac case "$before_id" in ''|*[!0-9]*) before_id="" ;; esac + case "$after_id" in ''|*[!0-9]*) after_id="" ;; esac + if [ -n "$before_id" ] && [ -n "$after_id" ]; then + echo "--before-id and --after-id are mutually exclusive" >&2 + exit 1 + fi local db; db="$(agmsg_db_path)" if [ ! -f "$db" ]; then @@ -123,6 +130,9 @@ get_messages() { if [ -n "$before_id" ]; then where="$where AND id<$before_id" fi + if [ -n "$after_id" ]; then + where="$where AND id>$after_id" + fi # Inner query takes the most recent `limit` by id DESC, outer re-sorts # ASC — oldest-first output, same ordering contract §2.1 of the driver @@ -133,6 +143,14 @@ get_messages() { # a decimal STRING (not a JSON number) so a future UUIDv7/Redis-stream-id # driver doesn't change this field's JSON type — a consumer parsing id as # a string today needs no change once that lands. + # Forward polling must take the OLDEST rows after the cursor. Taking the + # most recent rows would silently skip messages whenever more than `limit` + # arrive between polls. Backward/history queries keep the existing newest-N + # behavior, with both paths emitting oldest-first JSONL. + local page_order="ORDER BY id DESC LIMIT $limit" + if [ -n "$after_id" ]; then + page_order="ORDER BY id ASC LIMIT $limit" + fi agmsg_sqlite "$db" " SELECT json_object( 'type', 'message_sent', @@ -143,7 +161,7 @@ get_messages() { 'body', body, 'at', created_at ) FROM ( - SELECT * FROM messages WHERE $where ORDER BY id DESC LIMIT $limit + SELECT * FROM messages WHERE $where $page_order ) ORDER BY id ASC; " } diff --git a/tests/test_api.bats b/tests/test_api.bats index 316eba52..c9f49acf 100644 --- a/tests/test_api.bats +++ b/tests/test_api.bats @@ -143,6 +143,27 @@ json_valid_line() { [ "$(json_field "$output" body)" = "one" ] } +@test "api: get teams messages --after-id pages forward without skipping" { + bash "$SCRIPTS/send.sh" testteam alice bob "one" + local cursor + cursor="$(json_field "$(bash "$SCRIPTS/api.sh" get teams testteam messages --limit 1)" id)" + bash "$SCRIPTS/send.sh" testteam alice bob "two" + bash "$SCRIPTS/send.sh" testteam alice bob "three" + bash "$SCRIPTS/send.sh" testteam alice bob "four" + + run bash "$SCRIPTS/api.sh" get teams testteam messages --after-id "$cursor" --limit 2 + [ "$status" -eq 0 ] + [ "$(echo "$output" | wc -l | tr -d ' ')" -eq 2 ] + [ "$(json_field "$(echo "$output" | sed -n 1p)" body)" = "two" ] + [ "$(json_field "$(echo "$output" | sed -n 2p)" body)" = "three" ] +} + +@test "api: get teams messages rejects before and after cursors together" { + run bash "$SCRIPTS/api.sh" get teams testteam messages --before-id 3 --after-id 1 + [ "$status" -ne 0 ] + [[ "$output" =~ "mutually exclusive" ]] +} + @test "api: get teams messages emits valid JSON for a body containing a quote" { bash "$SCRIPTS/send.sh" testteam alice bob "she said \"hi\"" run bash "$SCRIPTS/api.sh" get teams testteam messages From 76a7db784be97766423ac37e08405b60efa3cfdd Mon Sep 17 00:00:00 2001 From: fumi Date: Tue, 14 Jul 2026 09:42:46 +0900 Subject: [PATCH 2/4] feat(app): merge local and remote message sources Allow Team Room to combine the implicit local agmsg installation with read-only installations reached over SSH. Sources are configured in ~/.agmsg/config/app-sources.json and queried through the public api.sh interface with the system SSH client; the app never mounts, copies, or synchronizes messages.db and does not store credentials. Keep history pages and live cursors independent per source, deduplicate messages with source-qualified ids, and merge them in stable timestamp order. Modern remotes use lossless --after-id polling while older cores fall back to refreshing their latest 100 messages. Show source badges and remote members for filtering while keeping spawn, rename, leave, and compose operations local-only. Add source validation, Rust and frontend coverage, and configuration documentation. --- app/README.md | 36 ++++ app/src-tauri/src/agmsg.rs | 341 ++++++++++++++++++++++++++++- app/src-tauri/src/lib.rs | 4 + app/src/App.css | 13 ++ app/src/App.tsx | 379 +++++++++++++++++++++++++++------ app/src/messageSources.test.ts | 51 +++++ app/src/messageSources.ts | 44 ++++ 7 files changed, 797 insertions(+), 71 deletions(-) create mode 100644 app/src/messageSources.test.ts create mode 100644 app/src/messageSources.ts 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..eb533e5c 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, 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 2f56d495..e6efc125 100644 --- a/app/src-tauri/src/lib.rs +++ b/app/src-tauri/src/lib.rs @@ -595,6 +595,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 794f474f..434eba0e 100644 --- a/app/src/App.css +++ b/app/src/App.css @@ -1365,3 +1365,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 6e8d84f4..36a07ea9 100644 --- a/app/src/App.tsx +++ b/app/src/App.tsx @@ -42,16 +42,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; @@ -141,10 +147,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 [windows, setWindows] = useState([]); const [active, setActive] = useState("room"); @@ -268,6 +276,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). @@ -300,14 +320,19 @@ export default function App() { windowsRef.current = windows; // 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. @@ -321,11 +346,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) => @@ -338,17 +380,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(() => { @@ -417,7 +497,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); @@ -430,7 +511,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 @@ -447,33 +528,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); @@ -500,7 +627,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]); // Load-more-on-scroll-up: fetch the page older than the currently-oldest // loaded message and prepend it, restoring the scroll position afterward @@ -508,30 +718,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 @@ -572,6 +818,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); @@ -850,7 +1097,7 @@ export default function App() { agentType: APP_USER_TYPE, project, }); - await loadTeams(); + await loadTeams(sources); setTeam(name); setModal(null); }, @@ -870,7 +1117,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); }, @@ -1067,7 +1314,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) })); @@ -1271,9 +1518,10 @@ export default function App() {
    {others.map((m) => (
  • { + if (m.read_only) return; e.preventDefault(); e.stopPropagation(); closeAllMenus(); @@ -1292,11 +1540,16 @@ export default function App() { )} @@ -1462,10 +1716,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}
    ))} @@ -1753,8 +2010,8 @@ export default function App() {