From 2e9e84d5372871efc0b0fa3ae3b9024bb03ec9ac Mon Sep 17 00:00:00 2001 From: marcmantei <8666222+marcmantei@users.noreply.github.com> Date: Thu, 12 Mar 2026 17:10:34 +0100 Subject: [PATCH 1/8] feat: add dynamic project registry (Phase 1) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add a SQLite-backed registry that auto-discovers GitHub repositories via `gh repo list`, syncs periodically in the background, and optionally auto-clones new repos. Per-repo config overrides (model, enabled) are preserved across syncs. New modules: - registry/store.rs — RegistryStore CRUD (upsert, list, overrides, link) - registry/sync.rs — discovery via gh CLI, reconciliation, auto-clone - api/registry.rs — 5 REST endpoints for repos, overrides, sync, status Config: [defaults.registry] section with github_owners, clone_base_dir, sync_interval_secs, auto_clone, exclude_patterns. Wired into RuntimeConfig (hot-reloadable), AgentDeps, ApiState, and main agent init loop with background sync task. 9 new tests, all 633 tests pass. Refs: marcmantei/spacebot#1 Generated with [Claude Code](https://claude.ai/code) via [Happy](https://happy.engineering) Co-Authored-By: Claude Co-Authored-By: Happy --- migrations/20260312000001_registry.sql | 36 ++ src/api.rs | 1 + src/api/agents.rs | 6 + src/api/registry.rs | 185 +++++++++ src/api/server.rs | 13 +- src/api/state.rs | 20 + src/config/load.rs | 24 +- src/config/runtime.rs | 5 + src/config/toml_schema.rs | 14 + src/config/types.rs | 37 ++ src/lib.rs | 2 + src/main.rs | 24 ++ src/registry.rs | 8 + src/registry/store.rs | 539 +++++++++++++++++++++++++ src/registry/sync.rs | 340 ++++++++++++++++ 15 files changed, 1251 insertions(+), 3 deletions(-) create mode 100644 migrations/20260312000001_registry.sql create mode 100644 src/api/registry.rs create mode 100644 src/registry.rs create mode 100644 src/registry/store.rs create mode 100644 src/registry/sync.rs diff --git a/migrations/20260312000001_registry.sql b/migrations/20260312000001_registry.sql new file mode 100644 index 000000000..2895cbdc4 --- /dev/null +++ b/migrations/20260312000001_registry.sql @@ -0,0 +1,36 @@ +-- Dynamic project registry: auto-discovered GitHub repositories. +-- Tracks repos from `gh repo list` with per-repo config overrides. + +CREATE TABLE IF NOT EXISTS registry_repos ( + id TEXT PRIMARY KEY, + agent_id TEXT NOT NULL, + owner TEXT NOT NULL, + name TEXT NOT NULL, + full_name TEXT NOT NULL, + description TEXT NOT NULL DEFAULT '', + default_branch TEXT NOT NULL DEFAULT 'main', + is_archived INTEGER NOT NULL DEFAULT 0, + is_fork INTEGER NOT NULL DEFAULT 0, + visibility TEXT NOT NULL DEFAULT 'private', + language TEXT, + local_path TEXT, + clone_url TEXT NOT NULL, + ssh_url TEXT NOT NULL DEFAULT '', + -- Per-repo overrides (NULL = inherit from agent defaults) + worker_model TEXT, + enabled INTEGER NOT NULL DEFAULT 1, + -- Linkage to existing projects table + project_id TEXT, + -- Sync metadata + last_synced_at TIMESTAMP, + created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, + updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, + FOREIGN KEY (project_id) REFERENCES projects(id) ON DELETE SET NULL +); + +CREATE UNIQUE INDEX IF NOT EXISTS idx_registry_repos_full_name + ON registry_repos(agent_id, full_name); +CREATE INDEX IF NOT EXISTS idx_registry_repos_agent + ON registry_repos(agent_id); +CREATE INDEX IF NOT EXISTS idx_registry_repos_enabled + ON registry_repos(agent_id, enabled); diff --git a/src/api.rs b/src/api.rs index 0bd99aec9..5bd1add1f 100644 --- a/src/api.rs +++ b/src/api.rs @@ -20,6 +20,7 @@ mod models; mod opencode_proxy; mod projects; mod providers; +mod registry; mod secrets; mod server; mod settings; diff --git a/src/api/agents.rs b/src/api/agents.rs index ce75842c1..2f75b12e5 100644 --- a/src/api/agents.rs +++ b/src/api/agents.rs @@ -417,6 +417,8 @@ pub(super) async fn trigger_warmup( let (event_tx, memory_event_tx) = crate::create_process_event_buses(); let project_store = std::sync::Arc::new(crate::projects::ProjectStore::new(sqlite_pool.clone())); + let registry_store = + std::sync::Arc::new(crate::registry::RegistryStore::new(sqlite_pool.clone())); let deps = crate::AgentDeps { agent_id: Arc::from(agent_id.as_str()), memory_search, @@ -431,6 +433,7 @@ pub(super) async fn trigger_warmup( sandbox, task_store, project_store, + registry_store, links: Arc::new(arc_swap::ArcSwap::from_pointee(Vec::new())), agent_names: Arc::new(std::collections::HashMap::new()), humans: Arc::new(arc_swap::ArcSwap::from_pointee(humans)), @@ -760,6 +763,8 @@ pub async fn create_agent_internal( ); let project_store = std::sync::Arc::new(crate::projects::ProjectStore::new(db.sqlite.clone())); + let registry_store = + std::sync::Arc::new(crate::registry::RegistryStore::new(db.sqlite.clone())); // Inject active project root paths into the sandbox allowlist. crate::projects::refresh_sandbox_project_paths(&project_store, &arc_agent_id, &sandbox).await; @@ -771,6 +776,7 @@ pub async fn create_agent_internal( mcp_manager: mcp_manager.clone(), task_store: task_store.clone(), project_store: project_store.clone(), + registry_store: registry_store.clone(), cron_tool: None, runtime_config: runtime_config.clone(), event_tx: event_tx.clone(), diff --git a/src/api/registry.rs b/src/api/registry.rs new file mode 100644 index 000000000..f6687e672 --- /dev/null +++ b/src/api/registry.rs @@ -0,0 +1,185 @@ +//! REST API handlers for the dynamic project registry. + +use super::state::ApiState; + +use axum::Json; +use axum::extract::{Query, State}; +use axum::http::StatusCode; +use serde::{Deserialize, Serialize}; +use std::sync::Arc; + +use crate::registry::store::RegistryRepo; +use crate::registry::sync::{SyncResult, SyncStatus, sync_registry}; + +// --------------------------------------------------------------------------- +// Query / request types +// --------------------------------------------------------------------------- + +#[derive(Deserialize)] +pub(super) struct AgentQuery { + agent_id: String, +} + +#[derive(Deserialize)] +pub(super) struct RepoListQuery { + agent_id: String, + #[serde(default)] + enabled_only: bool, +} + +#[derive(Deserialize)] +pub(super) struct RepoQuery { + agent_id: String, + full_name: String, +} + +#[derive(Deserialize)] +pub(super) struct UpdateRepoOverridesBody { + agent_id: String, + full_name: String, + /// Set to `Some(Some("model"))` to set, `Some(None)` to clear, `None` to leave unchanged. + worker_model: Option>, + enabled: Option, +} + +// --------------------------------------------------------------------------- +// Response types +// --------------------------------------------------------------------------- + +#[derive(Serialize)] +pub(super) struct RepoListResponse { + repos: Vec, + total: usize, +} + +#[derive(Serialize)] +pub(super) struct SyncResponse { + result: SyncResult, +} + +#[derive(Serialize)] +pub(super) struct StatusResponse { + status: SyncStatus, +} + +// --------------------------------------------------------------------------- +// Handlers +// --------------------------------------------------------------------------- + +/// GET /api/registry/repos — list all registry repos for an agent. +pub(super) async fn list_registry_repos( + State(state): State>, + Query(query): Query, +) -> Result, StatusCode> { + let stores = state.registry_stores.load(); + let store = stores + .get(&query.agent_id) + .ok_or(StatusCode::NOT_FOUND)?; + + let repos = store + .list_repos(&query.agent_id, query.enabled_only) + .await + .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?; + + let total = repos.len(); + Ok(Json(RepoListResponse { repos, total })) +} + +/// GET /api/registry/repos/detail — get a single repo by full_name. +pub(super) async fn get_registry_repo( + State(state): State>, + Query(query): Query, +) -> Result, StatusCode> { + let stores = state.registry_stores.load(); + let store = stores + .get(&query.agent_id) + .ok_or(StatusCode::NOT_FOUND)?; + + let repo = store + .get_by_full_name(&query.agent_id, &query.full_name) + .await + .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)? + .ok_or(StatusCode::NOT_FOUND)?; + + Ok(Json(repo)) +} + +/// PUT /api/registry/repos/overrides — update per-repo overrides. +pub(super) async fn update_repo_overrides( + State(state): State>, + Json(body): Json, +) -> Result, StatusCode> { + let stores = state.registry_stores.load(); + let store = stores + .get(&body.agent_id) + .ok_or(StatusCode::NOT_FOUND)?; + + let repo = store + .set_overrides( + &body.agent_id, + &body.full_name, + body.worker_model, + body.enabled, + ) + .await + .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)? + .ok_or(StatusCode::NOT_FOUND)?; + + Ok(Json(repo)) +} + +/// POST /api/registry/sync — trigger a manual sync. +pub(super) async fn trigger_sync( + State(state): State>, + Query(query): Query, +) -> Result, StatusCode> { + let stores = state.registry_stores.load(); + let store = stores + .get(&query.agent_id) + .ok_or(StatusCode::NOT_FOUND)?; + + let configs = state.runtime_configs.load(); + let runtime_config = configs + .get(&query.agent_id) + .ok_or(StatusCode::NOT_FOUND)?; + + let registry_config = runtime_config.registry.load(); + + if !registry_config.enabled { + return Err(StatusCode::BAD_REQUEST); + } + + // Update sync status + let status_map = state.registry_sync_status.load(); + if let Some(status) = status_map.get(&query.agent_id) { + status.store(Arc::new(SyncStatus::Syncing)); + } + + let result = sync_registry(store, &query.agent_id, ®istry_config) + .await + .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?; + + // Update sync status + if let Some(status) = status_map.get(&query.agent_id) { + status.store(Arc::new(SyncStatus::Completed { + at: chrono::Utc::now().to_rfc3339(), + result: result.clone(), + })); + } + + Ok(Json(SyncResponse { result })) +} + +/// GET /api/registry/status — get current sync status. +pub(super) async fn registry_status( + State(state): State>, + Query(query): Query, +) -> Result, StatusCode> { + let status_map = state.registry_sync_status.load(); + let status = status_map + .get(&query.agent_id) + .ok_or(StatusCode::NOT_FOUND)?; + + let current = status.load().as_ref().clone(); + Ok(Json(StatusResponse { status: current })) +} diff --git a/src/api/server.rs b/src/api/server.rs index 4d4451d38..1de5121c0 100644 --- a/src/api/server.rs +++ b/src/api/server.rs @@ -3,8 +3,8 @@ use super::state::ApiState; use super::{ agents, bindings, channels, config, cortex, cron, factory, ingest, links, mcp, memories, - messaging, models, opencode_proxy, projects, providers, secrets, settings, skills, ssh, system, - tasks, tools, webchat, workers, + messaging, models, opencode_proxy, projects, providers, registry, secrets, settings, skills, + ssh, system, tasks, tools, webchat, workers, }; use axum::Json; @@ -202,6 +202,15 @@ pub async fn start_http_server( .route("/agents/skills/upload", post(skills::upload_skill)) .route("/agents/skills/remove", delete(skills::remove_skill)) .route("/agents/tools", get(tools::list_tools)) + // Registry: dynamic project discovery + .route("/registry/repos", get(registry::list_registry_repos)) + .route("/registry/repos/detail", get(registry::get_registry_repo)) + .route( + "/registry/repos/overrides", + put(registry::update_repo_overrides), + ) + .route("/registry/sync", post(registry::trigger_sync)) + .route("/registry/status", get(registry::registry_status)) // Secret store management .route("/secrets/status", get(secrets::secrets_status)) .route("/secrets", get(secrets::list_secrets)) diff --git a/src/api/state.rs b/src/api/state.rs index 9500617e1..9938bce4b 100644 --- a/src/api/state.rs +++ b/src/api/state.rs @@ -12,6 +12,7 @@ use crate::memory::{EmbeddingModel, MemorySearch}; use crate::messaging::MessagingManager; use crate::messaging::webchat::WebChatAdapter; use crate::projects::ProjectStore; +use crate::registry::{RegistryStore, SyncStatus}; use crate::prompts::PromptEngine; use crate::tasks::TaskStore; use crate::update::SharedUpdateStatus; @@ -84,6 +85,10 @@ pub struct ApiState { pub task_stores: arc_swap::ArcSwap>>, /// Per-agent project stores for project/repo/worktree CRUD operations. pub project_stores: arc_swap::ArcSwap>>, + /// Per-agent registry stores for auto-discovered GitHub repos. + pub registry_stores: arc_swap::ArcSwap>>, + /// Per-agent registry sync status. + pub registry_sync_status: arc_swap::ArcSwap>>>, /// Per-agent RuntimeConfig for reading live hot-reloaded configuration. pub runtime_configs: ArcSwap>>, /// Per-agent MCP managers for status and reconnect APIs. @@ -310,6 +315,8 @@ impl ApiState { cron_schedulers: arc_swap::ArcSwap::from_pointee(HashMap::new()), task_stores: arc_swap::ArcSwap::from_pointee(HashMap::new()), project_stores: arc_swap::ArcSwap::from_pointee(HashMap::new()), + registry_stores: arc_swap::ArcSwap::from_pointee(HashMap::new()), + registry_sync_status: arc_swap::ArcSwap::from_pointee(HashMap::new()), runtime_configs: ArcSwap::from_pointee(HashMap::new()), mcp_managers: ArcSwap::from_pointee(HashMap::new()), sandboxes: ArcSwap::from_pointee(HashMap::new()), @@ -756,6 +763,19 @@ impl ApiState { self.project_stores.store(Arc::new(stores)); } + /// Set the registry stores for all agents. + pub fn set_registry_stores(&self, stores: HashMap>) { + self.registry_stores.store(Arc::new(stores)); + } + + /// Set the registry sync status trackers for all agents. + pub fn set_registry_sync_status( + &self, + status: HashMap>>, + ) { + self.registry_sync_status.store(Arc::new(status)); + } + /// Set the runtime configs for all agents. pub fn set_runtime_configs(&self, configs: HashMap>) { self.runtime_configs.store(Arc::new(configs)); diff --git a/src/config/load.rs b/src/config/load.rs index a4b44882c..5cf9a8761 100644 --- a/src/config/load.rs +++ b/src/config/load.rs @@ -15,7 +15,7 @@ use super::{ CoalesceConfig, CompactionConfig, Config, CortexConfig, CronDef, DefaultsConfig, DiscordConfig, DiscordInstanceConfig, EmailConfig, EmailInstanceConfig, GroupDef, HumanDef, IngestionConfig, LinkDef, LlmConfig, McpServerConfig, McpTransport, MemoryPersistenceConfig, MessagingConfig, - MetricsConfig, OpenCodeConfig, ProjectsConfig, ProviderConfig, SignalConfig, + MetricsConfig, OpenCodeConfig, ProjectsConfig, ProviderConfig, RegistryConfig, SignalConfig, SignalInstanceConfig, SlackCommandConfig, SlackConfig, SlackInstanceConfig, TelegramConfig, TelegramInstanceConfig, TelemetryConfig, TwitchConfig, TwitchInstanceConfig, WarmupConfig, WebhookConfig, normalize_adapter, validate_named_messaging_adapters, @@ -1610,6 +1610,28 @@ impl Config { } }) .unwrap_or_else(|| base_defaults.projects.clone()), + registry: toml + .defaults + .registry + .map(|r| { + let base = &base_defaults.registry; + RegistryConfig { + enabled: r.enabled.unwrap_or(base.enabled), + github_owners: r.github_owners.unwrap_or_else(|| base.github_owners.clone()), + clone_base_dir: r + .clone_base_dir + .map(std::path::PathBuf::from) + .unwrap_or_else(|| base.clone_base_dir.clone()), + sync_interval_secs: r + .sync_interval_secs + .unwrap_or(base.sync_interval_secs), + auto_clone: r.auto_clone.unwrap_or(base.auto_clone), + exclude_patterns: r + .exclude_patterns + .unwrap_or_else(|| base.exclude_patterns.clone()), + } + }) + .unwrap_or_else(|| base_defaults.registry.clone()), }; let mut agents: Vec = toml diff --git a/src/config/runtime.rs b/src/config/runtime.rs index 58c36ce47..c1c9ef648 100644 --- a/src/config/runtime.rs +++ b/src/config/runtime.rs @@ -77,6 +77,8 @@ pub struct RuntimeConfig { pub sandbox: Arc>, /// Projects workspace management configuration. pub projects: ArcSwap, + /// Registry configuration for auto-discovering GitHub repos. + pub registry: ArcSwap, /// Shared browser state for persistent sessions. /// /// When `browser.persist_session = true`, all workers share this handle so @@ -141,6 +143,7 @@ impl RuntimeConfig { secrets: ArcSwap::from_pointee(None), sandbox: Arc::new(ArcSwap::from_pointee(agent_config.sandbox.clone())), projects: ArcSwap::from_pointee(agent_config.projects.clone()), + registry: ArcSwap::from_pointee(agent_config.registry.clone()), shared_browser: if agent_config.browser.persist_session { Some(crate::tools::browser::new_shared_browser_handle()) } else { @@ -294,6 +297,8 @@ impl RuntimeConfig { new_sandbox.project_paths = existing_project_paths; self.sandbox.store(Arc::new(new_sandbox)); self.projects.store(Arc::new(resolved.projects.clone())); + self.registry + .store(Arc::new(config.defaults.registry.clone())); let old_opencode = self.opencode.load().as_ref().clone(); let new_opencode = config.defaults.opencode.clone(); diff --git a/src/config/toml_schema.rs b/src/config/toml_schema.rs index a3a833484..509083d79 100644 --- a/src/config/toml_schema.rs +++ b/src/config/toml_schema.rs @@ -294,6 +294,20 @@ pub(super) struct TomlDefaultsConfig { pub(super) opencode: Option, pub(super) worker_log_mode: Option, pub(super) projects: Option, + pub(super) registry: Option, +} + +#[derive(Deserialize)] +pub(super) struct TomlRegistryConfig { + #[serde(default)] + pub(super) enabled: Option, + #[serde(default)] + pub(super) github_owners: Option>, + pub(super) clone_base_dir: Option, + pub(super) sync_interval_secs: Option, + pub(super) auto_clone: Option, + #[serde(default)] + pub(super) exclude_patterns: Option>, } #[derive(Deserialize, Default)] diff --git a/src/config/types.rs b/src/config/types.rs index fe9d31463..86d54df98 100644 --- a/src/config/types.rs +++ b/src/config/types.rs @@ -550,6 +550,8 @@ pub struct DefaultsConfig { pub worker_log_mode: crate::settings::WorkerLogMode, /// Projects workspace management defaults. pub projects: ProjectsConfig, + /// Registry configuration for auto-discovering GitHub repos. + pub registry: RegistryConfig, } impl std::fmt::Debug for DefaultsConfig { @@ -581,6 +583,7 @@ impl std::fmt::Debug for DefaultsConfig { .field("opencode", &self.opencode) .field("worker_log_mode", &self.worker_log_mode) .field("projects", &self.projects) + .field("registry", &self.registry) .finish() } } @@ -995,6 +998,36 @@ impl Default for ProjectsConfig { } } +/// Registry configuration for auto-discovering GitHub repositories. +#[derive(Debug, Clone)] +pub struct RegistryConfig { + /// Whether the registry sync is enabled. + pub enabled: bool, + /// GitHub user/org names to discover repos from. + pub github_owners: Vec, + /// Base directory for auto-cloning repos. + pub clone_base_dir: std::path::PathBuf, + /// Sync interval in seconds (default: 3600 = 1 hour). + pub sync_interval_secs: u64, + /// Whether to auto-clone newly discovered repos. + pub auto_clone: bool, + /// Repos to exclude from the registry (exact match or trailing `*` glob). + pub exclude_patterns: Vec, +} + +impl Default for RegistryConfig { + fn default() -> Self { + Self { + enabled: false, + github_owners: Vec::new(), + clone_base_dir: std::path::PathBuf::from("/tmp/spacebot-repos"), + sync_interval_secs: 3600, + auto_clone: false, + exclude_patterns: Vec::new(), + } + } +} + /// Current warmup lifecycle state. #[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] #[serde(rename_all = "snake_case")] @@ -1198,6 +1231,8 @@ pub struct ResolvedAgentConfig { pub sandbox: crate::sandbox::SandboxConfig, /// Projects workspace management settings. pub projects: ProjectsConfig, + /// Registry configuration for auto-discovering GitHub repos. + pub registry: RegistryConfig, /// Number of messages to fetch from the platform when a new channel is created. pub history_backfill_count: usize, pub cron: Vec, @@ -1229,6 +1264,7 @@ impl Default for DefaultsConfig { opencode: OpenCodeConfig::default(), worker_log_mode: crate::settings::WorkerLogMode::default(), projects: ProjectsConfig::default(), + registry: RegistryConfig::default(), } } } @@ -1300,6 +1336,7 @@ impl AgentConfig { .projects .clone() .unwrap_or_else(|| defaults.projects.clone()), + registry: defaults.registry.clone(), history_backfill_count: defaults.history_backfill_count, cron: self.cron.clone(), } diff --git a/src/lib.rs b/src/lib.rs index 4ed49856d..aae4f6bc6 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -22,6 +22,7 @@ pub mod openai_auth; pub mod opencode; pub mod projects; pub mod prompts; +pub mod registry; pub mod sandbox; pub mod secrets; pub mod self_awareness; @@ -386,6 +387,7 @@ pub struct AgentDeps { pub mcp_manager: Arc, pub task_store: Arc, pub project_store: Arc, + pub registry_store: Arc, pub cron_tool: Option, pub runtime_config: Arc, pub event_tx: tokio::sync::broadcast::Sender, diff --git a/src/main.rs b/src/main.rs index 0679c27f3..2559da7db 100644 --- a/src/main.rs +++ b/src/main.rs @@ -2528,6 +2528,7 @@ async fn initialize_agents( spacebot::memory::MemoryStore::with_agent_id(db.sqlite.clone(), &agent_config.id); let task_store = Arc::new(spacebot::tasks::TaskStore::new(db.sqlite.clone())); let project_store = Arc::new(spacebot::projects::ProjectStore::new(db.sqlite.clone())); + let registry_store = Arc::new(spacebot::registry::RegistryStore::new(db.sqlite.clone())); let embedding_table = spacebot::memory::EmbeddingTable::open_or_create(&db.lance) .await .with_context(|| { @@ -2635,6 +2636,7 @@ async fn initialize_agents( mcp_manager, task_store: task_store.clone(), project_store: project_store.clone(), + registry_store: registry_store.clone(), cron_tool: None, runtime_config, event_tx, @@ -2707,6 +2709,8 @@ async fn initialize_agents( let mut mcp_managers = std::collections::HashMap::new(); let mut task_stores = std::collections::HashMap::new(); let mut project_stores = std::collections::HashMap::new(); + let mut registry_stores = std::collections::HashMap::new(); + let mut registry_sync_statuses = std::collections::HashMap::new(); let mut agent_workspaces = std::collections::HashMap::new(); let mut agent_identity_dirs = std::collections::HashMap::new(); let mut agent_data_dirs = std::collections::HashMap::new(); @@ -2720,6 +2724,24 @@ async fn initialize_agents( mcp_managers.insert(agent_id.to_string(), agent.deps.mcp_manager.clone()); task_stores.insert(agent_id.to_string(), agent.deps.task_store.clone()); project_stores.insert(agent_id.to_string(), agent.deps.project_store.clone()); + registry_stores.insert(agent_id.to_string(), agent.deps.registry_store.clone()); + // Registry sync: start background loop for each agent + { + let sync_status = std::sync::Arc::new( + arc_swap::ArcSwap::from_pointee(spacebot::registry::SyncStatus::default()), + ); + registry_sync_statuses + .insert(agent_id.to_string(), sync_status.clone()); + let reg_store = agent.deps.registry_store.as_ref().clone(); + let reg_agent_id = agent_id.to_string(); + let reg_runtime_config = agent.deps.runtime_config.clone(); + tokio::spawn(spacebot::registry::sync::registry_sync_loop( + reg_store, + reg_agent_id, + reg_runtime_config, + sync_status, + )); + } agent_workspaces.insert(agent_id.to_string(), agent.config.workspace.clone()); agent_identity_dirs.insert(agent_id.to_string(), agent.config.identity_dir.clone()); agent_data_dirs.insert(agent_id.to_string(), agent.config.data_dir.clone()); @@ -2744,6 +2766,8 @@ async fn initialize_agents( api_state.set_mcp_managers(mcp_managers); api_state.set_task_stores(task_stores); api_state.set_project_stores(project_stores); + api_state.set_registry_stores(registry_stores); + api_state.set_registry_sync_status(registry_sync_statuses); api_state.set_runtime_configs(runtime_configs); api_state.set_agent_workspaces(agent_workspaces); api_state.set_agent_identity_dirs(agent_identity_dirs); diff --git a/src/registry.rs b/src/registry.rs new file mode 100644 index 000000000..e418394ff --- /dev/null +++ b/src/registry.rs @@ -0,0 +1,8 @@ +//! Dynamic project registry: auto-discovers GitHub repositories and keeps +//! a persistent index of repos with per-repo config overrides. + +pub mod store; +pub mod sync; + +pub use store::{RegistryRepo, RegistryStore}; +pub use sync::{SyncResult, SyncStatus}; diff --git a/src/registry/store.rs b/src/registry/store.rs new file mode 100644 index 000000000..038408beb --- /dev/null +++ b/src/registry/store.rs @@ -0,0 +1,539 @@ +//! SQLite-backed registry store for auto-discovered GitHub repositories. + +use crate::error::Result; + +use anyhow::Context as _; +use serde::{Deserialize, Serialize}; +use sqlx::{Row as _, SqlitePool}; + +/// A GitHub repository tracked in the registry. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct RegistryRepo { + pub id: String, + pub agent_id: String, + pub owner: String, + pub name: String, + pub full_name: String, + pub description: String, + pub default_branch: String, + pub is_archived: bool, + pub is_fork: bool, + pub visibility: String, + pub language: Option, + pub local_path: Option, + pub clone_url: String, + pub ssh_url: String, + /// Per-repo model override (e.g., "anthropic/claude-sonnet-4"). + pub worker_model: Option, + /// Whether the agent should process events for this repo. + pub enabled: bool, + /// Link to the existing projects table. + pub project_id: Option, + pub last_synced_at: Option, + pub created_at: String, + pub updated_at: String, +} + +/// Input for upserting a repo discovered from `gh repo list`. +#[derive(Debug, Clone)] +pub struct UpsertRepoInput { + pub agent_id: String, + pub owner: String, + pub name: String, + pub full_name: String, + pub description: String, + pub default_branch: String, + pub is_archived: bool, + pub is_fork: bool, + pub visibility: String, + pub language: Option, + pub clone_url: String, + pub ssh_url: String, +} + +/// SQLite-backed registry store. +#[derive(Debug, Clone)] +pub struct RegistryStore { + pool: SqlitePool, +} + +impl RegistryStore { + pub fn new(pool: SqlitePool) -> Self { + Self { pool } + } + + /// Insert or update a discovered repo. Preserves user-set overrides + /// (worker_model, enabled) on conflict. + pub async fn upsert_repo(&self, input: UpsertRepoInput) -> Result { + let id = uuid::Uuid::new_v4().to_string(); + + sqlx::query( + r#" + INSERT INTO registry_repos (id, agent_id, owner, name, full_name, description, + default_branch, is_archived, is_fork, visibility, language, clone_url, ssh_url, + last_synced_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, CURRENT_TIMESTAMP) + ON CONFLICT (agent_id, full_name) DO UPDATE SET + description = excluded.description, + default_branch = excluded.default_branch, + is_archived = excluded.is_archived, + is_fork = excluded.is_fork, + visibility = excluded.visibility, + language = excluded.language, + clone_url = excluded.clone_url, + ssh_url = excluded.ssh_url, + last_synced_at = CURRENT_TIMESTAMP, + updated_at = CURRENT_TIMESTAMP + "#, + ) + .bind(&id) + .bind(&input.agent_id) + .bind(&input.owner) + .bind(&input.name) + .bind(&input.full_name) + .bind(&input.description) + .bind(&input.default_branch) + .bind(input.is_archived) + .bind(input.is_fork) + .bind(&input.visibility) + .bind(&input.language) + .bind(&input.clone_url) + .bind(&input.ssh_url) + .execute(&self.pool) + .await + .context("failed to upsert registry repo")?; + + Ok(self + .get_by_full_name(&input.agent_id, &input.full_name) + .await? + .context("repo not found after upsert")?) + } + + /// Get a repo by its full name (e.g., "marcmantei/ChargePilot-Launch"). + pub async fn get_by_full_name( + &self, + agent_id: &str, + full_name: &str, + ) -> Result> { + let row = sqlx::query( + "SELECT * FROM registry_repos WHERE agent_id = ? AND full_name = ?", + ) + .bind(agent_id) + .bind(full_name) + .fetch_optional(&self.pool) + .await + .context("failed to fetch registry repo by full_name")?; + + row.map(|r| row_to_registry_repo(&r)).transpose() + } + + /// Get a repo by its ID. + pub async fn get_by_id(&self, id: &str) -> Result> { + let row = sqlx::query("SELECT * FROM registry_repos WHERE id = ?") + .bind(id) + .fetch_optional(&self.pool) + .await + .context("failed to fetch registry repo by id")?; + + row.map(|r| row_to_registry_repo(&r)).transpose() + } + + /// List all repos for an agent, optionally filtering by enabled status. + pub async fn list_repos( + &self, + agent_id: &str, + enabled_only: bool, + ) -> Result> { + let rows = if enabled_only { + sqlx::query( + "SELECT * FROM registry_repos WHERE agent_id = ? AND enabled = 1 ORDER BY full_name ASC", + ) + .bind(agent_id) + .fetch_all(&self.pool) + .await + .context("failed to list enabled registry repos")? + } else { + sqlx::query( + "SELECT * FROM registry_repos WHERE agent_id = ? ORDER BY full_name ASC", + ) + .bind(agent_id) + .fetch_all(&self.pool) + .await + .context("failed to list registry repos")? + }; + + rows.iter().map(row_to_registry_repo).collect() + } + + /// Update per-repo overrides. + pub async fn set_overrides( + &self, + agent_id: &str, + full_name: &str, + worker_model: Option>, + enabled: Option, + ) -> Result> { + let existing = self.get_by_full_name(agent_id, full_name).await?; + let Some(existing) = existing else { + return Ok(None); + }; + + let worker_model = match worker_model { + Some(v) => v, + None => existing.worker_model, + }; + let enabled = enabled.unwrap_or(existing.enabled); + + sqlx::query( + r#" + UPDATE registry_repos + SET worker_model = ?, enabled = ?, updated_at = CURRENT_TIMESTAMP + WHERE agent_id = ? AND full_name = ? + "#, + ) + .bind(&worker_model) + .bind(enabled) + .bind(agent_id) + .bind(full_name) + .execute(&self.pool) + .await + .context("failed to update registry repo overrides")?; + + self.get_by_full_name(agent_id, full_name).await + } + + /// Link a registry repo to an existing project. + pub async fn link_project( + &self, + agent_id: &str, + full_name: &str, + project_id: Option<&str>, + ) -> Result<()> { + sqlx::query( + "UPDATE registry_repos SET project_id = ?, updated_at = CURRENT_TIMESTAMP WHERE agent_id = ? AND full_name = ?", + ) + .bind(project_id) + .bind(agent_id) + .bind(full_name) + .execute(&self.pool) + .await + .context("failed to link registry repo to project")?; + Ok(()) + } + + /// Set the local_path for a repo (after cloning). + pub async fn set_local_path( + &self, + agent_id: &str, + full_name: &str, + local_path: &str, + ) -> Result<()> { + sqlx::query( + "UPDATE registry_repos SET local_path = ?, updated_at = CURRENT_TIMESTAMP WHERE agent_id = ? AND full_name = ?", + ) + .bind(local_path) + .bind(agent_id) + .bind(full_name) + .execute(&self.pool) + .await + .context("failed to set registry repo local_path")?; + Ok(()) + } + + /// Mark repos not seen in the latest sync as archived. + pub async fn mark_absent_as_archived( + &self, + agent_id: &str, + seen_full_names: &[String], + ) -> Result { + if seen_full_names.is_empty() { + return Ok(0); + } + + // Build a parameterized IN clause + let placeholders: Vec<&str> = seen_full_names.iter().map(|_| "?").collect(); + let in_clause = placeholders.join(", "); + let query = format!( + "UPDATE registry_repos SET is_archived = 1, updated_at = CURRENT_TIMESTAMP \ + WHERE agent_id = ? AND full_name NOT IN ({}) AND is_archived = 0", + in_clause + ); + + let mut q = sqlx::query(&query).bind(agent_id); + for name in seen_full_names { + q = q.bind(name); + } + + let result = q + .execute(&self.pool) + .await + .context("failed to mark absent repos as archived")?; + + Ok(result.rows_affected()) + } + + /// Delete a repo from the registry. + pub async fn delete_repo(&self, agent_id: &str, full_name: &str) -> Result { + let result = sqlx::query( + "DELETE FROM registry_repos WHERE agent_id = ? AND full_name = ?", + ) + .bind(agent_id) + .bind(full_name) + .execute(&self.pool) + .await + .context("failed to delete registry repo")?; + + Ok(result.rows_affected() > 0) + } +} + +// Row mapping + +fn row_to_registry_repo(row: &sqlx::sqlite::SqliteRow) -> Result { + Ok(RegistryRepo { + id: row.try_get("id").context("missing id")?, + agent_id: row.try_get("agent_id").context("missing agent_id")?, + owner: row.try_get("owner").context("missing owner")?, + name: row.try_get("name").context("missing name")?, + full_name: row.try_get("full_name").context("missing full_name")?, + description: row.try_get("description").context("missing description")?, + default_branch: row + .try_get("default_branch") + .context("missing default_branch")?, + is_archived: row + .try_get::("is_archived") + .context("missing is_archived")?, + is_fork: row + .try_get::("is_fork") + .context("missing is_fork")?, + visibility: row + .try_get("visibility") + .context("missing visibility")?, + language: row.try_get("language").unwrap_or(None), + local_path: row.try_get("local_path").unwrap_or(None), + clone_url: row.try_get("clone_url").context("missing clone_url")?, + ssh_url: row.try_get("ssh_url").context("missing ssh_url")?, + worker_model: row.try_get("worker_model").unwrap_or(None), + enabled: row + .try_get::("enabled") + .context("missing enabled")?, + project_id: row.try_get("project_id").unwrap_or(None), + last_synced_at: row.try_get("last_synced_at").unwrap_or(None), + created_at: row.try_get("created_at").context("missing created_at")?, + updated_at: row.try_get("updated_at").context("missing updated_at")?, + }) +} + +// Tests + +#[cfg(test)] +mod tests { + use super::*; + + async fn setup_pool() -> SqlitePool { + let pool = SqlitePool::connect("sqlite::memory:") + .await + .expect("failed to create in-memory pool"); + sqlx::migrate!("./migrations") + .run(&pool) + .await + .expect("failed to run migrations"); + pool + } + + fn sample_input() -> UpsertRepoInput { + UpsertRepoInput { + agent_id: "main".into(), + owner: "marcmantei".into(), + name: "ChargePilot-Launch".into(), + full_name: "marcmantei/ChargePilot-Launch".into(), + description: "ChargePilot launch site".into(), + default_branch: "main".into(), + is_archived: false, + is_fork: false, + visibility: "private".into(), + language: Some("TypeScript".into()), + clone_url: "https://github.com/marcmantei/ChargePilot-Launch.git".into(), + ssh_url: "git@github.com:marcmantei/ChargePilot-Launch.git".into(), + } + } + + #[tokio::test] + async fn upsert_and_retrieve() { + let pool = setup_pool().await; + let store = RegistryStore::new(pool); + + let repo = store.upsert_repo(sample_input()).await.unwrap(); + assert_eq!(repo.full_name, "marcmantei/ChargePilot-Launch"); + assert_eq!(repo.language.as_deref(), Some("TypeScript")); + assert!(repo.enabled); + assert!(!repo.is_archived); + + // Retrieve by full name + let found = store + .get_by_full_name("main", "marcmantei/ChargePilot-Launch") + .await + .unwrap() + .expect("should find repo"); + assert_eq!(found.id, repo.id); + } + + #[tokio::test] + async fn upsert_preserves_overrides() { + let pool = setup_pool().await; + let store = RegistryStore::new(pool); + + store.upsert_repo(sample_input()).await.unwrap(); + + // Set a worker_model override + store + .set_overrides( + "main", + "marcmantei/ChargePilot-Launch", + Some(Some("anthropic/claude-sonnet-4".into())), + Some(false), + ) + .await + .unwrap(); + + // Upsert again (simulating re-sync) + let mut input = sample_input(); + input.description = "Updated description".into(); + store.upsert_repo(input).await.unwrap(); + + // Overrides should be preserved + let repo = store + .get_by_full_name("main", "marcmantei/ChargePilot-Launch") + .await + .unwrap() + .unwrap(); + assert_eq!( + repo.worker_model.as_deref(), + Some("anthropic/claude-sonnet-4") + ); + assert!(!repo.enabled); + assert_eq!(repo.description, "Updated description"); + } + + #[tokio::test] + async fn list_repos_filter() { + let pool = setup_pool().await; + let store = RegistryStore::new(pool); + + store.upsert_repo(sample_input()).await.unwrap(); + + let mut input2 = sample_input(); + input2.name = "other-repo".into(); + input2.full_name = "marcmantei/other-repo".into(); + store.upsert_repo(input2).await.unwrap(); + + // Disable one + store + .set_overrides("main", "marcmantei/other-repo", None, Some(false)) + .await + .unwrap(); + + let all = store.list_repos("main", false).await.unwrap(); + assert_eq!(all.len(), 2); + + let enabled = store.list_repos("main", true).await.unwrap(); + assert_eq!(enabled.len(), 1); + assert_eq!(enabled[0].full_name, "marcmantei/ChargePilot-Launch"); + } + + #[tokio::test] + async fn mark_absent_as_archived() { + let pool = setup_pool().await; + let store = RegistryStore::new(pool); + + store.upsert_repo(sample_input()).await.unwrap(); + + let mut input2 = sample_input(); + input2.name = "removed-repo".into(); + input2.full_name = "marcmantei/removed-repo".into(); + store.upsert_repo(input2).await.unwrap(); + + // Only ChargePilot-Launch was seen in this sync + let archived = store + .mark_absent_as_archived( + "main", + &["marcmantei/ChargePilot-Launch".into()], + ) + .await + .unwrap(); + assert_eq!(archived, 1); + + let repo = store + .get_by_full_name("main", "marcmantei/removed-repo") + .await + .unwrap() + .unwrap(); + assert!(repo.is_archived); + } + + #[tokio::test] + async fn delete_repo() { + let pool = setup_pool().await; + let store = RegistryStore::new(pool); + + store.upsert_repo(sample_input()).await.unwrap(); + let deleted = store + .delete_repo("main", "marcmantei/ChargePilot-Launch") + .await + .unwrap(); + assert!(deleted); + + let found = store + .get_by_full_name("main", "marcmantei/ChargePilot-Launch") + .await + .unwrap(); + assert!(found.is_none()); + } + + #[tokio::test] + async fn link_project() { + let pool = setup_pool().await; + let store = RegistryStore::new(pool.clone()); + + store.upsert_repo(sample_input()).await.unwrap(); + + // Create a real project to link to (satisfies FK constraint). + let project_store = crate::projects::ProjectStore::new(pool); + let project = project_store + .create_project(crate::projects::store::CreateProjectInput { + agent_id: "main".into(), + name: "ChargePilot".into(), + description: String::new(), + icon: String::new(), + tags: vec![], + root_path: "/home/sira/ChargePilot-Launch".into(), + settings: serde_json::Value::Object(Default::default()), + }) + .await + .unwrap(); + + store + .link_project("main", "marcmantei/ChargePilot-Launch", Some(&project.id)) + .await + .unwrap(); + + let repo = store + .get_by_full_name("main", "marcmantei/ChargePilot-Launch") + .await + .unwrap() + .unwrap(); + assert_eq!(repo.project_id.as_deref(), Some(project.id.as_str())); + + // Unlinking should also work. + store + .link_project("main", "marcmantei/ChargePilot-Launch", None) + .await + .unwrap(); + let repo = store + .get_by_full_name("main", "marcmantei/ChargePilot-Launch") + .await + .unwrap() + .unwrap(); + assert!(repo.project_id.is_none()); + } +} diff --git a/src/registry/sync.rs b/src/registry/sync.rs new file mode 100644 index 000000000..fd7a3452c --- /dev/null +++ b/src/registry/sync.rs @@ -0,0 +1,340 @@ +//! Registry sync: discovers GitHub repos via `gh repo list` and reconciles +//! against the local registry store. + +use super::store::{RegistryStore, UpsertRepoInput}; +use crate::config::RegistryConfig; +use crate::error::Result; + +use anyhow::Context as _; +use arc_swap::ArcSwap; +use serde::Deserialize; +use std::path::Path; +use std::sync::Arc; +use tokio::time::Duration; + +/// Result of a single sync pass. +#[derive(Debug, Clone, Default, serde::Serialize)] +pub struct SyncResult { + pub repos_found: usize, + pub new: usize, + pub updated: usize, + pub archived: usize, + pub cloned: usize, + pub errors: Vec, +} + +/// Current sync status. +#[derive(Debug, Clone, serde::Serialize)] +#[serde(tag = "state", rename_all = "snake_case")] +pub enum SyncStatus { + Idle, + Syncing, + Failed { + error: String, + at: String, + }, + Completed { + at: String, + result: SyncResult, + }, +} + +impl Default for SyncStatus { + fn default() -> Self { + Self::Idle + } +} + +/// JSON shape returned by `gh repo list --json`. +#[derive(Debug, Deserialize)] +struct GhRepo { + name: String, + #[serde(default)] + description: String, + #[serde(rename = "defaultBranchRef")] + default_branch_ref: Option, + #[serde(rename = "isArchived", default)] + is_archived: bool, + #[serde(rename = "isFork", default)] + is_fork: bool, + #[serde(default)] + visibility: String, + #[serde(rename = "primaryLanguage")] + primary_language: Option, + #[serde(default)] + url: String, + #[serde(rename = "sshUrl", default)] + ssh_url: String, +} + +#[derive(Debug, Deserialize)] +struct GhBranchRef { + name: String, +} + +#[derive(Debug, Deserialize)] +struct GhLanguage { + name: String, +} + +/// Discover repos for a single GitHub owner using the `gh` CLI. +async fn discover_github_repos(owner: &str) -> Result> { + let output = tokio::process::Command::new("gh") + .args([ + "repo", + "list", + owner, + "--json", + "name,description,defaultBranchRef,isArchived,isFork,visibility,primaryLanguage,url,sshUrl", + "--limit", + "200", + ]) + .output() + .await + .context("failed to run `gh repo list`")?; + + if !output.status.success() { + let stderr = String::from_utf8_lossy(&output.stderr); + return Err(anyhow::anyhow!("`gh repo list {}` failed: {}", owner, stderr.trim()).into()); + } + + let repos: Vec = + serde_json::from_slice(&output.stdout).context("failed to parse gh repo list output")?; + Ok(repos) +} + +/// Check if a repo matches any exclude pattern (simple glob with `*` support). +fn matches_exclude(full_name: &str, patterns: &[String]) -> bool { + for pattern in patterns { + if pattern == full_name { + return true; + } + // Support trailing wildcard: "owner/*-test" not supported, but "owner/prefix*" is. + if let Some(prefix) = pattern.strip_suffix('*') { + if full_name.starts_with(prefix) { + return true; + } + } + } + false +} + +/// Run a single sync pass: discover repos, upsert into store, mark absent repos. +pub async fn sync_registry( + store: &RegistryStore, + agent_id: &str, + config: &RegistryConfig, +) -> Result { + let mut result = SyncResult::default(); + let mut all_seen: Vec = Vec::new(); + + for owner in &config.github_owners { + match discover_github_repos(owner).await { + Ok(repos) => { + for gh_repo in repos { + let full_name = format!("{}/{}", owner, gh_repo.name); + + // Apply exclude patterns + if matches_exclude(&full_name, &config.exclude_patterns) { + continue; + } + + all_seen.push(full_name.clone()); + result.repos_found += 1; + + let was_known = store + .get_by_full_name(agent_id, &full_name) + .await? + .is_some(); + + let input = UpsertRepoInput { + agent_id: agent_id.to_string(), + owner: owner.clone(), + name: gh_repo.name.clone(), + full_name: full_name.clone(), + description: gh_repo.description, + default_branch: gh_repo + .default_branch_ref + .map(|b| b.name) + .unwrap_or_else(|| "main".into()), + is_archived: gh_repo.is_archived, + is_fork: gh_repo.is_fork, + visibility: gh_repo.visibility.to_lowercase(), + language: gh_repo.primary_language.map(|l| l.name), + clone_url: gh_repo.url.clone(), + ssh_url: gh_repo.ssh_url, + }; + + store.upsert_repo(input).await?; + + if was_known { + result.updated += 1; + } else { + result.new += 1; + + // Auto-clone if configured + if config.auto_clone && !gh_repo.is_archived { + let clone_dir = config.clone_base_dir.join(&gh_repo.name); + if !clone_dir.exists() { + match auto_clone_repo(&full_name, &clone_dir).await { + Ok(()) => { + store + .set_local_path( + agent_id, + &full_name, + &clone_dir.to_string_lossy(), + ) + .await?; + result.cloned += 1; + tracing::info!( + repo = %full_name, + path = %clone_dir.display(), + "auto-cloned new repo" + ); + } + Err(e) => { + let msg = + format!("failed to clone {}: {}", full_name, e); + tracing::warn!("{}", msg); + result.errors.push(msg); + } + } + } else { + // Directory exists, just record the path + store + .set_local_path( + agent_id, + &full_name, + &clone_dir.to_string_lossy(), + ) + .await?; + } + } + } + } + } + Err(e) => { + let msg = format!("failed to discover repos for {}: {}", owner, e); + tracing::warn!("{}", msg); + result.errors.push(msg); + } + } + } + + // Mark repos not seen in this sync as archived + if !all_seen.is_empty() { + result.archived = store + .mark_absent_as_archived(agent_id, &all_seen) + .await? as usize; + } + + Ok(result) +} + +/// Clone a repo using `gh repo clone`. +async fn auto_clone_repo(full_name: &str, target_dir: &Path) -> Result<()> { + let output = tokio::process::Command::new("gh") + .args([ + "repo", + "clone", + full_name, + &target_dir.to_string_lossy(), + ]) + .output() + .await + .context("failed to run `gh repo clone`")?; + + if !output.status.success() { + let stderr = String::from_utf8_lossy(&output.stderr); + return Err(anyhow::anyhow!("gh repo clone failed: {}", stderr.trim()).into()); + } + Ok(()) +} + +/// Background loop that periodically syncs the registry. +/// +/// Reads the current `RegistryConfig` from `runtime_config` on each iteration +/// so hot-reloads take effect without restarting. +pub async fn registry_sync_loop( + store: RegistryStore, + agent_id: String, + runtime_config: Arc, + status: Arc>, +) { + // Initial delay to let the agent fully start up. + tokio::time::sleep(Duration::from_secs(30)).await; + + loop { + let current_config = runtime_config.registry.load(); + if !current_config.enabled { + tokio::time::sleep(Duration::from_secs(60)).await; + continue; + } + + let sync_interval = current_config.sync_interval_secs; + + status.store(Arc::new(SyncStatus::Syncing)); + tracing::info!(agent_id = %agent_id, "starting registry sync"); + + match sync_registry(&store, &agent_id, ¤t_config).await { + Ok(result) => { + tracing::info!( + agent_id = %agent_id, + found = result.repos_found, + new = result.new, + archived = result.archived, + cloned = result.cloned, + errors = result.errors.len(), + "registry sync completed" + ); + status.store(Arc::new(SyncStatus::Completed { + at: chrono::Utc::now().to_rfc3339(), + result, + })); + } + Err(e) => { + tracing::error!(agent_id = %agent_id, error = %e, "registry sync failed"); + status.store(Arc::new(SyncStatus::Failed { + error: e.to_string(), + at: chrono::Utc::now().to_rfc3339(), + })); + } + } + + tokio::time::sleep(Duration::from_secs(sync_interval)).await; + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_matches_exclude() { + assert!(matches_exclude("marcmantei/liralot-config", &["marcmantei/liralot-config".into()])); + assert!(matches_exclude("marcmantei/test-repo", &["marcmantei/test*".into()])); + assert!(!matches_exclude("marcmantei/ChargePilot", &["marcmantei/liralot-config".into()])); + assert!(!matches_exclude("other/repo", &["marcmantei/*".into()])); + } + + #[test] + fn test_gh_repo_deserialization() { + let json = r#"[ + { + "name": "test-repo", + "description": "A test", + "defaultBranchRef": {"name": "main"}, + "isArchived": false, + "isFork": false, + "visibility": "PRIVATE", + "primaryLanguage": {"name": "Rust"}, + "url": "https://github.com/owner/test-repo", + "sshUrl": "git@github.com:owner/test-repo.git" + } + ]"#; + let repos: Vec = serde_json::from_str(json).unwrap(); + assert_eq!(repos.len(), 1); + assert_eq!(repos[0].name, "test-repo"); + assert_eq!(repos[0].primary_language.as_ref().unwrap().name, "Rust"); + } +} From f32bbd4bf72978d828de42f27a172e399b2aece8 Mon Sep 17 00:00:00 2001 From: marcmantei <8666222+marcmantei@users.noreply.github.com> Date: Thu, 12 Mar 2026 17:10:34 +0100 Subject: [PATCH 2/8] feat: add notification on new repo discovery Send a summary message via MessagingManager when registry sync discovers new repos or archives removed ones. Configurable via `notification_target` in [defaults.registry] (e.g. "telegram:1285309093"). Generated with [Claude Code](https://claude.ai/code) via [Happy](https://happy.engineering) Co-Authored-By: Claude Co-Authored-By: Happy --- src/config/load.rs | 3 ++ src/config/toml_schema.rs | 1 + src/config/types.rs | 4 ++ src/main.rs | 2 + src/registry/store.rs | 16 ++++++++ src/registry/sync.rs | 79 +++++++++++++++++++++++++++++++++++++++ tests/bulletin.rs | 1 + tests/context_dump.rs | 1 + 8 files changed, 107 insertions(+) diff --git a/src/config/load.rs b/src/config/load.rs index 5cf9a8761..ac06b4ef6 100644 --- a/src/config/load.rs +++ b/src/config/load.rs @@ -1629,6 +1629,9 @@ impl Config { exclude_patterns: r .exclude_patterns .unwrap_or_else(|| base.exclude_patterns.clone()), + notification_target: r + .notification_target + .or_else(|| base.notification_target.clone()), } }) .unwrap_or_else(|| base_defaults.registry.clone()), diff --git a/src/config/toml_schema.rs b/src/config/toml_schema.rs index 509083d79..aa8a6da81 100644 --- a/src/config/toml_schema.rs +++ b/src/config/toml_schema.rs @@ -308,6 +308,7 @@ pub(super) struct TomlRegistryConfig { pub(super) auto_clone: Option, #[serde(default)] pub(super) exclude_patterns: Option>, + pub(super) notification_target: Option, } #[derive(Deserialize, Default)] diff --git a/src/config/types.rs b/src/config/types.rs index 86d54df98..1d8bfffc4 100644 --- a/src/config/types.rs +++ b/src/config/types.rs @@ -1013,6 +1013,9 @@ pub struct RegistryConfig { pub auto_clone: bool, /// Repos to exclude from the registry (exact match or trailing `*` glob). pub exclude_patterns: Vec, + /// Optional delivery target for notifications when repos are discovered/archived. + /// Format: "adapter:target" (e.g., "telegram:1285309093", "discord:123456789"). + pub notification_target: Option, } impl Default for RegistryConfig { @@ -1024,6 +1027,7 @@ impl Default for RegistryConfig { sync_interval_secs: 3600, auto_clone: false, exclude_patterns: Vec::new(), + notification_target: None, } } } diff --git a/src/main.rs b/src/main.rs index 2559da7db..f185e53e8 100644 --- a/src/main.rs +++ b/src/main.rs @@ -2735,11 +2735,13 @@ async fn initialize_agents( let reg_store = agent.deps.registry_store.as_ref().clone(); let reg_agent_id = agent_id.to_string(); let reg_runtime_config = agent.deps.runtime_config.clone(); + let reg_messaging = agent.deps.messaging_manager.clone(); tokio::spawn(spacebot::registry::sync::registry_sync_loop( reg_store, reg_agent_id, reg_runtime_config, sync_status, + reg_messaging, )); } agent_workspaces.insert(agent_id.to_string(), agent.config.workspace.clone()); diff --git a/src/registry/store.rs b/src/registry/store.rs index 038408beb..f0fe2b33f 100644 --- a/src/registry/store.rs +++ b/src/registry/store.rs @@ -272,6 +272,22 @@ impl RegistryStore { Ok(result.rows_affected()) } + /// Get the full names of all non-archived repos for an agent. + pub async fn get_non_archived_names(&self, agent_id: &str) -> Result> { + let rows = sqlx::query( + "SELECT full_name FROM registry_repos WHERE agent_id = ? AND is_archived = 0", + ) + .bind(agent_id) + .fetch_all(&self.pool) + .await + .context("failed to fetch non-archived repo names")?; + + rows.iter() + .map(|r| r.try_get("full_name").context("missing full_name")) + .collect::, _>>() + .map_err(Into::into) + } + /// Delete a repo from the registry. pub async fn delete_repo(&self, agent_id: &str, full_name: &str) -> Result { let result = sqlx::query( diff --git a/src/registry/sync.rs b/src/registry/sync.rs index fd7a3452c..9ec3857a0 100644 --- a/src/registry/sync.rs +++ b/src/registry/sync.rs @@ -4,6 +4,9 @@ use super::store::{RegistryStore, UpsertRepoInput}; use crate::config::RegistryConfig; use crate::error::Result; +use crate::messaging::MessagingManager; +use crate::messaging::target::parse_delivery_target; +use crate::OutboundResponse; use anyhow::Context as _; use arc_swap::ArcSwap; @@ -21,6 +24,10 @@ pub struct SyncResult { pub archived: usize, pub cloned: usize, pub errors: Vec, + /// Names of newly discovered repos (for notifications). + pub new_repos: Vec, + /// Names of repos newly marked as archived (for notifications). + pub archived_repos: Vec, } /// Current sync status. @@ -171,6 +178,7 @@ pub async fn sync_registry( result.updated += 1; } else { result.new += 1; + result.new_repos.push(full_name.clone()); // Auto-clone if configured if config.auto_clone && !gh_repo.is_archived { @@ -223,9 +231,19 @@ pub async fn sync_registry( // Mark repos not seen in this sync as archived if !all_seen.is_empty() { + // Snapshot non-archived names before marking, so we can report which ones changed. + let before = store.get_non_archived_names(agent_id).await?; result.archived = store .mark_absent_as_archived(agent_id, &all_seen) .await? as usize; + if result.archived > 0 { + let seen_set: std::collections::HashSet<&str> = + all_seen.iter().map(|s| s.as_str()).collect(); + result.archived_repos = before + .into_iter() + .filter(|name| !seen_set.contains(name.as_str())) + .collect(); + } } Ok(result) @@ -260,6 +278,7 @@ pub async fn registry_sync_loop( agent_id: String, runtime_config: Arc, status: Arc>, + messaging_manager: Option>, ) { // Initial delay to let the agent fully start up. tokio::time::sleep(Duration::from_secs(30)).await; @@ -287,6 +306,17 @@ pub async fn registry_sync_loop( errors = result.errors.len(), "registry sync completed" ); + + // Send notification if there are new or archived repos. + if !result.new_repos.is_empty() || !result.archived_repos.is_empty() { + send_sync_notification( + ¤t_config, + &messaging_manager, + &result, + ) + .await; + } + status.store(Arc::new(SyncStatus::Completed { at: chrono::Utc::now().to_rfc3339(), result, @@ -305,6 +335,55 @@ pub async fn registry_sync_loop( } } +/// Build and send a notification about new/archived repos after a sync pass. +async fn send_sync_notification( + config: &RegistryConfig, + messaging_manager: &Option>, + result: &SyncResult, +) { + let target_str = match &config.notification_target { + Some(t) => t, + None => return, + }; + let mm = match messaging_manager { + Some(m) => m, + None => return, + }; + let target = match parse_delivery_target(target_str) { + Some(t) => t, + None => { + tracing::warn!( + target = %target_str, + "invalid notification_target for registry sync" + ); + return; + } + }; + + let mut lines = Vec::new(); + lines.push("📦 Registry sync update:".to_string()); + if !result.new_repos.is_empty() { + lines.push(format!("\nNew repos ({}):", result.new_repos.len())); + for name in &result.new_repos { + lines.push(format!(" + {}", name)); + } + } + if !result.archived_repos.is_empty() { + lines.push(format!("\nArchived repos ({}):", result.archived_repos.len())); + for name in &result.archived_repos { + lines.push(format!(" − {}", name)); + } + } + let text = lines.join("\n"); + + if let Err(e) = mm + .broadcast(&target.adapter, &target.target, OutboundResponse::Text(text)) + .await + { + tracing::warn!(error = %e, "failed to send registry sync notification"); + } +} + #[cfg(test)] mod tests { use super::*; diff --git a/tests/bulletin.rs b/tests/bulletin.rs index f7e6942ab..cf7b09f66 100644 --- a/tests/bulletin.rs +++ b/tests/bulletin.rs @@ -113,6 +113,7 @@ async fn bootstrap_deps() -> anyhow::Result { mcp_manager, task_store, project_store: Arc::new(spacebot::projects::ProjectStore::new(db.sqlite.clone())), + registry_store: Arc::new(spacebot::registry::RegistryStore::new(db.sqlite.clone())), cron_tool: None, runtime_config, event_tx, diff --git a/tests/context_dump.rs b/tests/context_dump.rs index 4fad60d8f..08159b8ac 100644 --- a/tests/context_dump.rs +++ b/tests/context_dump.rs @@ -112,6 +112,7 @@ async fn bootstrap_deps() -> anyhow::Result<(spacebot::AgentDeps, spacebot::conf mcp_manager, task_store, project_store: Arc::new(spacebot::projects::ProjectStore::new(db.sqlite.clone())), + registry_store: Arc::new(spacebot::registry::RegistryStore::new(db.sqlite.clone())), cron_tool: None, runtime_config, event_tx, From e3608f4c10757c34298ff718d2497eebbcea3b2e Mon Sep 17 00:00:00 2001 From: marcmantei <8666222+marcmantei@users.noreply.github.com> Date: Thu, 12 Mar 2026 17:48:42 +0100 Subject: [PATCH 3/8] fix: defer registry sync loop spawn until messaging_manager is available The sync loop was spawned before messaging_manager was set on AgentDeps, so notifications were never sent. Now spawns after init, reading the sync status from ApiState. Generated with [Claude Code](https://claude.ai/code) via [Happy](https://happy.engineering) Co-Authored-By: Claude Co-Authored-By: Happy --- src/main.rs | 31 +++++++++++++++++++------------ 1 file changed, 19 insertions(+), 12 deletions(-) diff --git a/src/main.rs b/src/main.rs index f185e53e8..b90c3da7f 100644 --- a/src/main.rs +++ b/src/main.rs @@ -2725,24 +2725,13 @@ async fn initialize_agents( task_stores.insert(agent_id.to_string(), agent.deps.task_store.clone()); project_stores.insert(agent_id.to_string(), agent.deps.project_store.clone()); registry_stores.insert(agent_id.to_string(), agent.deps.registry_store.clone()); - // Registry sync: start background loop for each agent + // Registry sync: prepare status (spawn deferred until messaging_manager is set) { let sync_status = std::sync::Arc::new( arc_swap::ArcSwap::from_pointee(spacebot::registry::SyncStatus::default()), ); registry_sync_statuses .insert(agent_id.to_string(), sync_status.clone()); - let reg_store = agent.deps.registry_store.as_ref().clone(); - let reg_agent_id = agent_id.to_string(); - let reg_runtime_config = agent.deps.runtime_config.clone(); - let reg_messaging = agent.deps.messaging_manager.clone(); - tokio::spawn(spacebot::registry::sync::registry_sync_loop( - reg_store, - reg_agent_id, - reg_runtime_config, - sync_status, - reg_messaging, - )); } agent_workspaces.insert(agent_id.to_string(), agent.config.workspace.clone()); agent_identity_dirs.insert(agent_id.to_string(), agent.config.identity_dir.clone()); @@ -3215,6 +3204,24 @@ async fn initialize_agents( let store = Arc::new(spacebot::cron::CronStore::new(agent.db.sqlite.clone())); agent.deps.messaging_manager = Some(messaging_manager.clone()); + // Registry sync: spawn background loop now that messaging_manager is available + { + let statuses = api_state.registry_sync_status.load(); + if let Some(sync_status) = statuses.get(&agent_id.to_string()) { + let reg_store = agent.deps.registry_store.as_ref().clone(); + let reg_agent_id = agent_id.to_string(); + let reg_runtime_config = agent.deps.runtime_config.clone(); + let reg_messaging = agent.deps.messaging_manager.clone(); + tokio::spawn(spacebot::registry::sync::registry_sync_loop( + reg_store, + reg_agent_id, + reg_runtime_config, + sync_status.clone(), + reg_messaging, + )); + } + } + // Seed cron jobs from config into the database for cron_def in &agent.config.cron { let cron_config = spacebot::cron::CronConfig { From da94ac22d30bef6ecb0b8ee9527a12af919c3b91 Mon Sep 17 00:00:00 2001 From: marcmantei <8666222+marcmantei@users.noreply.github.com> Date: Thu, 12 Mar 2026 19:57:29 +0100 Subject: [PATCH 4/8] fix: correct default model + worker retrigger relay 1. Fix cortex profile 404: default routing model was "anthropic/claude-sonnet-4" (doesn't exist), now uses "anthropic/claude-sonnet-4-20250514". 2. Fix worker outcome reporting: when a webhook-triggered worker completes and Lira relays the result via send_message_to_another_channel (e.g. to Telegram), the retrigger system didn't recognize it as a successful relay because only the reply tool set the replied_flag. Now SendMessageTool also sets the flag on successful delivery. Generated with [Claude Code](https://claude.ai/code) via [Happy](https://happy.engineering) Co-Authored-By: Claude Co-Authored-By: Happy --- src/llm/routing.rs | 4 ++-- src/tools.rs | 17 +++++++++------- src/tools/send_message_to_another_channel.rs | 21 ++++++++++++++++++++ 3 files changed, 33 insertions(+), 9 deletions(-) diff --git a/src/llm/routing.rs b/src/llm/routing.rs index eb677a8c8..f3e3aace7 100644 --- a/src/llm/routing.rs +++ b/src/llm/routing.rs @@ -36,7 +36,7 @@ pub struct RoutingConfig { impl Default for RoutingConfig { fn default() -> Self { - Self::for_model("anthropic/claude-sonnet-4".into()) + Self::for_model("anthropic/claude-sonnet-4-20250514".into()) } } @@ -163,7 +163,7 @@ pub fn is_context_overflow_error(error_message: &str) -> bool { /// each provider sane defaults so things work out of the box. pub fn defaults_for_provider(provider: &str) -> RoutingConfig { match provider { - "anthropic" => RoutingConfig::for_model("anthropic/claude-sonnet-4".into()), + "anthropic" => RoutingConfig::for_model("anthropic/claude-sonnet-4-20250514".into()), "openrouter" => { let channel: String = "openrouter/anthropic/claude-sonnet-4-20250514".into(); let worker: String = "openrouter/anthropic/claude-haiku-4.5-20250514".into(); diff --git a/src/tools.rs b/src/tools.rs index 677bd3e85..a062854ea 100644 --- a/src/tools.rs +++ b/src/tools.rs @@ -389,13 +389,16 @@ pub async fn add_channel_tools( .cloned() .unwrap_or_else(|| state.deps.agent_id.to_string()); handle - .add_tool(SendMessageTool::new( - messaging_manager.clone(), - state.channel_store.clone(), - state.conversation_logger.clone(), - send_message_display_name, - current_adapter.clone(), - )) + .add_tool( + SendMessageTool::new( + messaging_manager.clone(), + state.channel_store.clone(), + state.conversation_logger.clone(), + send_message_display_name, + current_adapter.clone(), + ) + .with_replied_flag(replied_flag.clone()), + ) .await?; } handle diff --git a/src/tools/send_message_to_another_channel.rs b/src/tools/send_message_to_another_channel.rs index b2232f637..b95a9a40a 100644 --- a/src/tools/send_message_to_another_channel.rs +++ b/src/tools/send_message_to_another_channel.rs @@ -4,12 +4,14 @@ use crate::ChannelId; use crate::conversation::ChannelStore; use crate::conversation::history::ConversationLogger; use crate::messaging::MessagingManager; +use crate::tools::reply::RepliedFlag; use rig::completion::ToolDefinition; use rig::tool::Tool; use schemars::JsonSchema; use serde::{Deserialize, Serialize}; use std::sync::Arc; +use std::sync::atomic::Ordering; /// Check if a string is a valid UUID format. /// Accepts standard UUID format: 8-4-4-4-12 hexadecimal digits. @@ -30,6 +32,7 @@ pub struct SendMessageTool { conversation_logger: ConversationLogger, agent_display_name: String, current_adapter: Option, + replied_flag: Option, } impl std::fmt::Debug for SendMessageTool { @@ -52,6 +55,20 @@ impl SendMessageTool { conversation_logger, agent_display_name, current_adapter, + replied_flag: None, + } + } + + /// Attach a replied flag so successful sends mark the retrigger as relayed. + pub fn with_replied_flag(mut self, flag: RepliedFlag) -> Self { + self.replied_flag = Some(flag); + self + } + + /// Mark the replied flag (if present) to indicate a message was delivered. + fn mark_replied(&self) { + if let Some(flag) = &self.replied_flag { + flag.store(true, Ordering::Relaxed); } } } @@ -174,6 +191,7 @@ impl Tool for SendMessageTool { "message sent via explicit signal: prefix" ); + self.mark_replied(); return Ok(SendMessageOutput { success: true, target: target.target, @@ -208,6 +226,7 @@ impl Tool for SendMessageTool { "message sent via implicit Signal shorthand" ); + self.mark_replied(); return Ok(SendMessageOutput { success: true, target: target.target, @@ -241,6 +260,7 @@ impl Tool for SendMessageTool { ); // Email targets don't have a channel to log to. + self.mark_replied(); return Ok(SendMessageOutput { success: true, target: explicit_target.target, @@ -303,6 +323,7 @@ impl Tool for SendMessageTool { "message sent to channel and logged to destination history" ); + self.mark_replied(); Ok(SendMessageOutput { success: true, target: channel.display_name.unwrap_or_else(|| channel.id.clone()), From 044f9d89e3f65ee5d5817493237dea4a01927141 Mon Sep 17 00:00:00 2001 From: marcmantei <8666222+marcmantei@users.noreply.github.com> Date: Thu, 12 Mar 2026 20:34:48 +0100 Subject: [PATCH 5/8] style: cargo fmt Generated with [Claude Code](https://claude.ai/code) via [Happy](https://happy.engineering) Co-Authored-By: Claude Co-Authored-By: Happy --- src/api/registry.rs | 20 ++++----------- src/api/state.rs | 7 ++--- src/config/load.rs | 8 +++--- src/main.rs | 9 +++---- src/registry/store.rs | 49 ++++++++++++++--------------------- src/registry/sync.rs | 59 ++++++++++++++++++++----------------------- 6 files changed, 62 insertions(+), 90 deletions(-) diff --git a/src/api/registry.rs b/src/api/registry.rs index f6687e672..94b18eaf3 100644 --- a/src/api/registry.rs +++ b/src/api/registry.rs @@ -72,9 +72,7 @@ pub(super) async fn list_registry_repos( Query(query): Query, ) -> Result, StatusCode> { let stores = state.registry_stores.load(); - let store = stores - .get(&query.agent_id) - .ok_or(StatusCode::NOT_FOUND)?; + let store = stores.get(&query.agent_id).ok_or(StatusCode::NOT_FOUND)?; let repos = store .list_repos(&query.agent_id, query.enabled_only) @@ -91,9 +89,7 @@ pub(super) async fn get_registry_repo( Query(query): Query, ) -> Result, StatusCode> { let stores = state.registry_stores.load(); - let store = stores - .get(&query.agent_id) - .ok_or(StatusCode::NOT_FOUND)?; + let store = stores.get(&query.agent_id).ok_or(StatusCode::NOT_FOUND)?; let repo = store .get_by_full_name(&query.agent_id, &query.full_name) @@ -110,9 +106,7 @@ pub(super) async fn update_repo_overrides( Json(body): Json, ) -> Result, StatusCode> { let stores = state.registry_stores.load(); - let store = stores - .get(&body.agent_id) - .ok_or(StatusCode::NOT_FOUND)?; + let store = stores.get(&body.agent_id).ok_or(StatusCode::NOT_FOUND)?; let repo = store .set_overrides( @@ -134,14 +128,10 @@ pub(super) async fn trigger_sync( Query(query): Query, ) -> Result, StatusCode> { let stores = state.registry_stores.load(); - let store = stores - .get(&query.agent_id) - .ok_or(StatusCode::NOT_FOUND)?; + let store = stores.get(&query.agent_id).ok_or(StatusCode::NOT_FOUND)?; let configs = state.runtime_configs.load(); - let runtime_config = configs - .get(&query.agent_id) - .ok_or(StatusCode::NOT_FOUND)?; + let runtime_config = configs.get(&query.agent_id).ok_or(StatusCode::NOT_FOUND)?; let registry_config = runtime_config.registry.load(); diff --git a/src/api/state.rs b/src/api/state.rs index 9938bce4b..c9f32037e 100644 --- a/src/api/state.rs +++ b/src/api/state.rs @@ -12,8 +12,8 @@ use crate::memory::{EmbeddingModel, MemorySearch}; use crate::messaging::MessagingManager; use crate::messaging::webchat::WebChatAdapter; use crate::projects::ProjectStore; -use crate::registry::{RegistryStore, SyncStatus}; use crate::prompts::PromptEngine; +use crate::registry::{RegistryStore, SyncStatus}; use crate::tasks::TaskStore; use crate::update::SharedUpdateStatus; use crate::{ProcessEvent, ProcessId}; @@ -769,10 +769,7 @@ impl ApiState { } /// Set the registry sync status trackers for all agents. - pub fn set_registry_sync_status( - &self, - status: HashMap>>, - ) { + pub fn set_registry_sync_status(&self, status: HashMap>>) { self.registry_sync_status.store(Arc::new(status)); } diff --git a/src/config/load.rs b/src/config/load.rs index ac06b4ef6..0f4755ca2 100644 --- a/src/config/load.rs +++ b/src/config/load.rs @@ -1617,14 +1617,14 @@ impl Config { let base = &base_defaults.registry; RegistryConfig { enabled: r.enabled.unwrap_or(base.enabled), - github_owners: r.github_owners.unwrap_or_else(|| base.github_owners.clone()), + github_owners: r + .github_owners + .unwrap_or_else(|| base.github_owners.clone()), clone_base_dir: r .clone_base_dir .map(std::path::PathBuf::from) .unwrap_or_else(|| base.clone_base_dir.clone()), - sync_interval_secs: r - .sync_interval_secs - .unwrap_or(base.sync_interval_secs), + sync_interval_secs: r.sync_interval_secs.unwrap_or(base.sync_interval_secs), auto_clone: r.auto_clone.unwrap_or(base.auto_clone), exclude_patterns: r .exclude_patterns diff --git a/src/main.rs b/src/main.rs index b90c3da7f..0ac6a6560 100644 --- a/src/main.rs +++ b/src/main.rs @@ -2727,11 +2727,10 @@ async fn initialize_agents( registry_stores.insert(agent_id.to_string(), agent.deps.registry_store.clone()); // Registry sync: prepare status (spawn deferred until messaging_manager is set) { - let sync_status = std::sync::Arc::new( - arc_swap::ArcSwap::from_pointee(spacebot::registry::SyncStatus::default()), - ); - registry_sync_statuses - .insert(agent_id.to_string(), sync_status.clone()); + let sync_status = std::sync::Arc::new(arc_swap::ArcSwap::from_pointee( + spacebot::registry::SyncStatus::default(), + )); + registry_sync_statuses.insert(agent_id.to_string(), sync_status.clone()); } agent_workspaces.insert(agent_id.to_string(), agent.config.workspace.clone()); agent_identity_dirs.insert(agent_id.to_string(), agent.config.identity_dir.clone()); diff --git a/src/registry/store.rs b/src/registry/store.rs index f0fe2b33f..65e14a9ea 100644 --- a/src/registry/store.rs +++ b/src/registry/store.rs @@ -115,14 +115,12 @@ impl RegistryStore { agent_id: &str, full_name: &str, ) -> Result> { - let row = sqlx::query( - "SELECT * FROM registry_repos WHERE agent_id = ? AND full_name = ?", - ) - .bind(agent_id) - .bind(full_name) - .fetch_optional(&self.pool) - .await - .context("failed to fetch registry repo by full_name")?; + let row = sqlx::query("SELECT * FROM registry_repos WHERE agent_id = ? AND full_name = ?") + .bind(agent_id) + .bind(full_name) + .fetch_optional(&self.pool) + .await + .context("failed to fetch registry repo by full_name")?; row.map(|r| row_to_registry_repo(&r)).transpose() } @@ -153,13 +151,11 @@ impl RegistryStore { .await .context("failed to list enabled registry repos")? } else { - sqlx::query( - "SELECT * FROM registry_repos WHERE agent_id = ? ORDER BY full_name ASC", - ) - .bind(agent_id) - .fetch_all(&self.pool) - .await - .context("failed to list registry repos")? + sqlx::query("SELECT * FROM registry_repos WHERE agent_id = ? ORDER BY full_name ASC") + .bind(agent_id) + .fetch_all(&self.pool) + .await + .context("failed to list registry repos")? }; rows.iter().map(row_to_registry_repo).collect() @@ -290,14 +286,12 @@ impl RegistryStore { /// Delete a repo from the registry. pub async fn delete_repo(&self, agent_id: &str, full_name: &str) -> Result { - let result = sqlx::query( - "DELETE FROM registry_repos WHERE agent_id = ? AND full_name = ?", - ) - .bind(agent_id) - .bind(full_name) - .execute(&self.pool) - .await - .context("failed to delete registry repo")?; + let result = sqlx::query("DELETE FROM registry_repos WHERE agent_id = ? AND full_name = ?") + .bind(agent_id) + .bind(full_name) + .execute(&self.pool) + .await + .context("failed to delete registry repo")?; Ok(result.rows_affected() > 0) } @@ -322,9 +316,7 @@ fn row_to_registry_repo(row: &sqlx::sqlite::SqliteRow) -> Result { is_fork: row .try_get::("is_fork") .context("missing is_fork")?, - visibility: row - .try_get("visibility") - .context("missing visibility")?, + visibility: row.try_get("visibility").context("missing visibility")?, language: row.try_get("language").unwrap_or(None), local_path: row.try_get("local_path").unwrap_or(None), clone_url: row.try_get("clone_url").context("missing clone_url")?, @@ -471,10 +463,7 @@ mod tests { // Only ChargePilot-Launch was seen in this sync let archived = store - .mark_absent_as_archived( - "main", - &["marcmantei/ChargePilot-Launch".into()], - ) + .mark_absent_as_archived("main", &["marcmantei/ChargePilot-Launch".into()]) .await .unwrap(); assert_eq!(archived, 1); diff --git a/src/registry/sync.rs b/src/registry/sync.rs index 9ec3857a0..426001623 100644 --- a/src/registry/sync.rs +++ b/src/registry/sync.rs @@ -2,11 +2,11 @@ //! against the local registry store. use super::store::{RegistryStore, UpsertRepoInput}; +use crate::OutboundResponse; use crate::config::RegistryConfig; use crate::error::Result; use crate::messaging::MessagingManager; use crate::messaging::target::parse_delivery_target; -use crate::OutboundResponse; use anyhow::Context as _; use arc_swap::ArcSwap; @@ -36,14 +36,8 @@ pub struct SyncResult { pub enum SyncStatus { Idle, Syncing, - Failed { - error: String, - at: String, - }, - Completed { - at: String, - result: SyncResult, - }, + Failed { error: String, at: String }, + Completed { at: String, result: SyncResult }, } impl Default for SyncStatus { @@ -201,8 +195,7 @@ pub async fn sync_registry( ); } Err(e) => { - let msg = - format!("failed to clone {}: {}", full_name, e); + let msg = format!("failed to clone {}: {}", full_name, e); tracing::warn!("{}", msg); result.errors.push(msg); } @@ -233,9 +226,7 @@ pub async fn sync_registry( if !all_seen.is_empty() { // Snapshot non-archived names before marking, so we can report which ones changed. let before = store.get_non_archived_names(agent_id).await?; - result.archived = store - .mark_absent_as_archived(agent_id, &all_seen) - .await? as usize; + result.archived = store.mark_absent_as_archived(agent_id, &all_seen).await? as usize; if result.archived > 0 { let seen_set: std::collections::HashSet<&str> = all_seen.iter().map(|s| s.as_str()).collect(); @@ -252,12 +243,7 @@ pub async fn sync_registry( /// Clone a repo using `gh repo clone`. async fn auto_clone_repo(full_name: &str, target_dir: &Path) -> Result<()> { let output = tokio::process::Command::new("gh") - .args([ - "repo", - "clone", - full_name, - &target_dir.to_string_lossy(), - ]) + .args(["repo", "clone", full_name, &target_dir.to_string_lossy()]) .output() .await .context("failed to run `gh repo clone`")?; @@ -309,12 +295,7 @@ pub async fn registry_sync_loop( // Send notification if there are new or archived repos. if !result.new_repos.is_empty() || !result.archived_repos.is_empty() { - send_sync_notification( - ¤t_config, - &messaging_manager, - &result, - ) - .await; + send_sync_notification(¤t_config, &messaging_manager, &result).await; } status.store(Arc::new(SyncStatus::Completed { @@ -369,7 +350,10 @@ async fn send_sync_notification( } } if !result.archived_repos.is_empty() { - lines.push(format!("\nArchived repos ({}):", result.archived_repos.len())); + lines.push(format!( + "\nArchived repos ({}):", + result.archived_repos.len() + )); for name in &result.archived_repos { lines.push(format!(" − {}", name)); } @@ -377,7 +361,11 @@ async fn send_sync_notification( let text = lines.join("\n"); if let Err(e) = mm - .broadcast(&target.adapter, &target.target, OutboundResponse::Text(text)) + .broadcast( + &target.adapter, + &target.target, + OutboundResponse::Text(text), + ) .await { tracing::warn!(error = %e, "failed to send registry sync notification"); @@ -390,9 +378,18 @@ mod tests { #[test] fn test_matches_exclude() { - assert!(matches_exclude("marcmantei/liralot-config", &["marcmantei/liralot-config".into()])); - assert!(matches_exclude("marcmantei/test-repo", &["marcmantei/test*".into()])); - assert!(!matches_exclude("marcmantei/ChargePilot", &["marcmantei/liralot-config".into()])); + assert!(matches_exclude( + "marcmantei/liralot-config", + &["marcmantei/liralot-config".into()] + )); + assert!(matches_exclude( + "marcmantei/test-repo", + &["marcmantei/test*".into()] + )); + assert!(!matches_exclude( + "marcmantei/ChargePilot", + &["marcmantei/liralot-config".into()] + )); assert!(!matches_exclude("other/repo", &["marcmantei/*".into()])); } From c3319e7e46b479c1a93214162a50459697032ebb Mon Sep 17 00:00:00 2001 From: marcmantei <8666222+marcmantei@users.noreply.github.com> Date: Thu, 12 Mar 2026 20:43:21 +0100 Subject: [PATCH 6/8] fix: clippy collapsible_if + derive Default for SyncStatus Generated with [Claude Code](https://claude.ai/code) via [Happy](https://happy.engineering) Co-Authored-By: Claude Co-Authored-By: Happy --- src/registry/sync.rs | 27 ++++++++++++++------------- 1 file changed, 14 insertions(+), 13 deletions(-) diff --git a/src/registry/sync.rs b/src/registry/sync.rs index 426001623..0c291e54c 100644 --- a/src/registry/sync.rs +++ b/src/registry/sync.rs @@ -31,19 +31,20 @@ pub struct SyncResult { } /// Current sync status. -#[derive(Debug, Clone, serde::Serialize)] +#[derive(Debug, Clone, Default, serde::Serialize)] #[serde(tag = "state", rename_all = "snake_case")] pub enum SyncStatus { + #[default] Idle, Syncing, - Failed { error: String, at: String }, - Completed { at: String, result: SyncResult }, -} - -impl Default for SyncStatus { - fn default() -> Self { - Self::Idle - } + Failed { + error: String, + at: String, + }, + Completed { + at: String, + result: SyncResult, + }, } /// JSON shape returned by `gh repo list --json`. @@ -111,10 +112,10 @@ fn matches_exclude(full_name: &str, patterns: &[String]) -> bool { return true; } // Support trailing wildcard: "owner/*-test" not supported, but "owner/prefix*" is. - if let Some(prefix) = pattern.strip_suffix('*') { - if full_name.starts_with(prefix) { - return true; - } + if let Some(prefix) = pattern.strip_suffix('*') + && full_name.starts_with(prefix) + { + return true; } } false From 39c6f7c55e8dbb2c8500f3c1791fc311237fa1a2 Mon Sep 17 00:00:00 2001 From: marcmantei <8666222+marcmantei@users.noreply.github.com> Date: Thu, 12 Mar 2026 21:26:24 +0100 Subject: [PATCH 7/8] feat: dynamic project context from registry + webhook filtering MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The agent now uses the registry as its dynamic project list: 1. System prompt injection: enabled registry repos are appended to the project context so the agent knows about all discovered repos (name, path, language, model override, default branch). 2. Webhook event filtering: events from repos not in the registry or with enabled=false are dropped before reaching the agent. This replaces the need for a hardcoded project table in ROLE.md. Combined with the GitHub App (all repos) and registry sync (auto-discover), this completes the zero-config flow: new repo → auto-discovered → webhooks delivered → agent reacts. Generated with [Claude Code](https://claude.ai/code) via [Happy](https://happy.engineering) Co-Authored-By: Claude Co-Authored-By: Happy --- src/agent/channel.rs | 68 +++++++++++++++++++++++++++++++++++++++++--- src/main.rs | 36 +++++++++++++++++++++++ 2 files changed, 100 insertions(+), 4 deletions(-) diff --git a/src/agent/channel.rs b/src/agent/channel.rs index 6f992b20c..0e275c075 100644 --- a/src/agent/channel.rs +++ b/src/agent/channel.rs @@ -2051,11 +2051,12 @@ impl Channel { } }; - if projects.is_empty() { - return None; - } + let mut contexts = Vec::new(); + + // Collect project_ids that are linked to registry repos so we can + // avoid duplicating them when we append unlinked registry repos below. + let mut linked_project_ids = std::collections::HashSet::new(); - let mut contexts = Vec::with_capacity(projects.len()); for project in &projects { let repos = match store.list_repos(&project.id).await { Ok(repos) => repos, @@ -2105,6 +2106,65 @@ impl Channel { }) .collect(), }); + linked_project_ids.insert(project.id.clone()); + } + + // Append enabled registry repos that are NOT already linked to a project, + // so the agent knows about all dynamically discovered repos. + if let Ok(registry_repos) = self + .deps + .registry_store + .list_repos(&self.deps.agent_id, true) + .await + { + for reg_repo in registry_repos { + // Skip repos already represented via a linked project. + if reg_repo + .project_id + .as_ref() + .is_some_and(|pid| linked_project_ids.contains(pid)) + { + continue; + } + // Skip archived repos. + if reg_repo.is_archived { + continue; + } + let root_path = reg_repo + .local_path + .clone() + .unwrap_or_else(|| format!("(not cloned) {}", reg_repo.clone_url)); + let mut desc_parts = Vec::new(); + if let Some(ref lang) = reg_repo.language { + desc_parts.push(lang.clone()); + } + if let Some(ref model) = reg_repo.worker_model { + desc_parts.push(format!("model: {model}")); + } + contexts.push(ProjectContext { + name: reg_repo.full_name.clone(), + root_path, + description: if desc_parts.is_empty() { + None + } else { + Some(desc_parts.join(", ")) + }, + tags: Vec::new(), + repos: vec![ProjectRepoContext { + name: reg_repo.name.clone(), + path: reg_repo + .local_path + .unwrap_or_else(|| reg_repo.clone_url.clone()), + default_branch: reg_repo.default_branch.clone(), + remote_url: Some(reg_repo.clone_url), + }], + worktrees: Vec::new(), + }); + } + } + + if contexts.is_empty() { + return None; } match prompt_engine.render_projects_context(contexts) { diff --git a/src/main.rs b/src/main.rs index 0ac6a6560..6bcfb5ffd 100644 --- a/src/main.rs +++ b/src/main.rs @@ -2003,6 +2003,42 @@ async fn run( let conversation_id = message.conversation_id.clone(); + // Filter webhook events against the dynamic registry. + // conversation_id format: "webhook:github:{owner/repo}" + if let Some(repo_full_name) = conversation_id + .strip_prefix("webhook:github:") + { + if let Some(agent) = agents.get(&agent_id) { + let registry = &agent.deps.registry_store; + match registry.get_by_full_name(&agent_id, repo_full_name).await { + Ok(Some(reg_repo)) if !reg_repo.enabled => { + tracing::debug!( + repo = %repo_full_name, + "webhook event dropped: repo disabled in registry" + ); + continue; + } + Ok(None) => { + tracing::debug!( + repo = %repo_full_name, + "webhook event dropped: repo not in registry" + ); + continue; + } + Err(error) => { + tracing::warn!( + repo = %repo_full_name, + %error, + "registry lookup failed, allowing webhook event" + ); + } + Ok(Some(_)) => { + // Repo is enabled, proceed. + } + } + } + } + // Find or create a channel for this conversation if !active_channels.contains_key(&conversation_id) { let Some(agent) = agents.get(&agent_id) else { From f355180b072bb70d81a34a61c0916428e771d61c Mon Sep 17 00:00:00 2001 From: Marc Mantei <8666222+marcmantei@users.noreply.github.com> Date: Thu, 12 Mar 2026 22:05:07 +0100 Subject: [PATCH 8/8] Fix clippy collapsible_if in webhook registry filter Generated with [Claude Code](https://claude.ai/code) via [Happy](https://happy.engineering) Co-Authored-By: Claude Co-Authored-By: Happy --- src/main.rs | 53 ++++++++++++++++++++++++++--------------------------- 1 file changed, 26 insertions(+), 27 deletions(-) diff --git a/src/main.rs b/src/main.rs index 6bcfb5ffd..0bb87929e 100644 --- a/src/main.rs +++ b/src/main.rs @@ -2007,34 +2007,33 @@ async fn run( // conversation_id format: "webhook:github:{owner/repo}" if let Some(repo_full_name) = conversation_id .strip_prefix("webhook:github:") + && let Some(agent) = agents.get(&agent_id) { - if let Some(agent) = agents.get(&agent_id) { - let registry = &agent.deps.registry_store; - match registry.get_by_full_name(&agent_id, repo_full_name).await { - Ok(Some(reg_repo)) if !reg_repo.enabled => { - tracing::debug!( - repo = %repo_full_name, - "webhook event dropped: repo disabled in registry" - ); - continue; - } - Ok(None) => { - tracing::debug!( - repo = %repo_full_name, - "webhook event dropped: repo not in registry" - ); - continue; - } - Err(error) => { - tracing::warn!( - repo = %repo_full_name, - %error, - "registry lookup failed, allowing webhook event" - ); - } - Ok(Some(_)) => { - // Repo is enabled, proceed. - } + let registry = &agent.deps.registry_store; + match registry.get_by_full_name(&agent_id, repo_full_name).await { + Ok(Some(reg_repo)) if !reg_repo.enabled => { + tracing::debug!( + repo = %repo_full_name, + "webhook event dropped: repo disabled in registry" + ); + continue; + } + Ok(None) => { + tracing::debug!( + repo = %repo_full_name, + "webhook event dropped: repo not in registry" + ); + continue; + } + Err(error) => { + tracing::warn!( + repo = %repo_full_name, + %error, + "registry lookup failed, allowing webhook event" + ); + } + Ok(Some(_)) => { + // Repo is enabled, proceed. } } }