Struct FleetManager 

pub struct FleetManager { /* private fields */ }

The fleet control plane.

Implementations

impl FleetManager

pub async fn new(config: FleetConfig, store: Arc<dyn Store>) -> Result<Self>

Build a manager, loading the persisted registry from a default SqliteConfigStore in config.data_dir (the same dir as the event store). For an injected config backend (e.g. Postgres), use Self::with_config_store.

pub async fn with_config_store( config: FleetConfig, store: Arc<dyn Store>, config_store: Arc<dyn ConfigStore>, ) -> Result<Self>

Build a manager with an explicit ConfigStore and the standalone EventBus/Ownership seams.

pub async fn with_seams( config: FleetConfig, store: Arc<dyn Store>, config_store: Arc<dyn ConfigStore>, bus: Arc<dyn EventBus>, ownership: Arc<dyn Ownership>, ) -> Result<Self>

Build a manager with every topology seam injected. Standalone passes InProcessBus + SelfOwnsAll; clustered (Phase 2d) passes DistributedBus + LeasedOwnership.

pub fn begin_shutdown(&self)

Signal a graceful shutdown: the poll loop finishes its in-flight cycle and returns, and attach tasks stop reading between frames. Idempotent.

Uses send_replace so the signal sticks even if no task has subscribed yet (plain send is a no-op when there are no receivers).

pub fn subscribe(&self, stream_key: &str) -> BusSubscription

Subscribe to one stream’s live event tail (see crate::EventBus). Watchers of a single agent pass the agent id (its stream key); repo/fleet watchers pass repo:<name> / fleet.

pub fn store(&self) -> Arc<dyn Store>

The shared event store. Observability reads (agent history/SSE) route here rather than through the fleet backend, so any FleetProvider (local or k8s) that emits to this store serves the same read path. (#76)

pub fn bus(&self) -> Arc<dyn EventBus>

The shared event bus (SSE subscribe routes here). See Self::store. (#76)

pub async fn snapshot(&self) -> FleetSnapshot

A clone of the current fleet snapshot, with each repo’s provider config joined in from the registry so a single read reflects any set_config.

pub fn metrics(&self) -> MetricsSnapshot

A snapshot of prosperod’s operational counters (active_attaches is read live from the running attach set).

pub async fn readiness(&self) -> Readiness

Aggregate readiness: store-writability (the ready gate) plus a summary of per-repo health. Used by the /readyz endpoint to distinguish liveness from readiness.

pub async fn history( &self, stream_key: &str, from_seq: u64, ) -> Result<Vec<FleetEvent>>

Replay a stream’s history from the store, with seq >= from_seq. Callers watching a single agent pass the agent id, which is that agent’s stream key (see crate::event::stream_key_for); repo/fleet-level history uses the repo:<name> / fleet keys.

pub async fn prune_older_than(&self, max_age: Duration) -> Result<u64>

Delete persisted events older than max_age. Returns the count removed. Backs the daemon’s age-based retention loop (#4).

pub async fn add_workspace( &self, name: impl Into<String>, root: impl Into<PathBuf>, ) -> Result<()>

Register a workspace and persist the registry. Triggers an immediate poll.

pub async fn add_repo( &self, name: impl Into<String>, root: impl Into<PathBuf>, ) -> Result<()>

Back-compat alias for Self::add_workspace: a single-repo workspace.

pub async fn add_repo_with_config( &self, name: impl Into<String>, root: impl Into<PathBuf>, config: RepoProviderConfig, ) -> Result<()>

Back-compat alias for Self::add_workspace_with_config.

pub async fn add_workspace_with_config( &self, name: impl Into<String>, root: impl Into<PathBuf>, config: RepoProviderConfig, ) -> Result<()>

Register a workspace with an initial provider config.

pub async fn repo_config(&self, repo: &str) -> Option<RepoProviderConfig>

The stored provider config for a repo, if registered.

pub async fn remove_repo(&self, name: &str) -> Result<bool>

Unregister a repo and persist the registry.

pub async fn ensure_config_for(&self, repo: &str) -> Result<EnsureConfig>

Build the EnsureConfig for a repo, resolving its env overlay from the global default + the repo’s stored provider config + prosperod’s env.

pub async fn set_repo_config_registry_only( &self, repo: &str, config: RepoProviderConfig, ) -> Result<()>

Update a repo’s provider config in the registry only (no restart).

pub async fn spawn_agent(&self, repo: &str, req: SpawnRequest) -> Result<String>

Launch a new agent under repo. Returns the new agent id.

pub async fn spawn_agent_with_socket( &self, repo: &str, req: SpawnRequest, ) -> Result<(String, Endpoint)>

Launch a new agent under repo, returning both its id and the per-agent endpoint client.spawn already handed back — so callers that need it (e.g. LocalFleet::ensure_agent) don’t have to issue a redundant Attach to re-derive it.

pub async fn kill_agent(&self, agent_id: &str) -> Result<()>

Kill an agent (resolving its repo from the snapshot).

pub async fn send_agent_input( &self, agent_id: &str, input: AttachInbound, ) -> Result<()>

Send an inbound control frame to an interactive agent. Rejects if the agent is unknown (AgentNotFound), terminal, or was not spawned interactive (InvalidState).

The state gate reads the last poll snapshot (up to one poll interval stale); caliband remains authoritative, so a just-terminated agent may pass the gate and fail at attach/send_inbound instead.

pub async fn respawn_agent(&self, agent_id: &str) -> Result<String>

Respawn an agent; returns the new id.

pub async fn drain_agent(&self, agent_id: &str, timeout: Duration) -> Result<()>

Minimal graceful drain (P1): send EndInput (fleet.rs:650), best-effort (the agent may not be interactive, or may already be terminal — either way drain still proceeds), then poll Self::snapshot up to timeout for the agent to reach a terminal AgentStatus, then unconditionally Self::kill_agent. Full checkpoint-drain is P2/operator territory — this just avoids yanking an agent mid-turn when the caller can wait a bit.

pub fn watch_changes(&self) -> BoxStream<'static, FleetChange>

Observe fleet changes as they happen: an initial burst of Discovered (one per currently-known agent) and WorkspaceHealth (one per repo) built from the current Self::snapshot, followed by a live FleetChange feed translated from the bus’s EventKind diffs — the same ones reconcile already computes (fleet.rs:811); reconcile itself is untouched.

Subscribes to the bus (via Self::subscribe_all) before reading the snapshot, mirroring InProcessBus::subscribe’s own eager-registration discipline, so no event published in the gap between “read snapshot” and “start the live feed” is lost.

pub async fn rm_agent(&self, agent_id: &str, force: bool) -> Result<()>

Remove an agent from caliban’s registry.

On success the agent is optimistically dropped from the served snapshot so GET /api/fleet reflects the removal immediately, rather than continuing to list it for up to one poll interval until the next poll reconciles (#123). The next poll remains authoritative and idempotent.

pub async fn poll_all_once(&self)

Poll every registered repo once. Refreshes the registry from the shared config store first so a clustered replica picks up repos a peer added/removed/reconfigured between cycles.

pub async fn poll_repo_once(&self, repo: &str)

Poll one repo: list agents, reconcile against the snapshot, emit diffs, and start attach tasks for newly-active agents. Failures degrade the repo to Unreachable rather than propagating.

pub async fn cached_client_names(&self) -> Vec<String>

Names of repos with a cached control client (test/observability helper).

pub fn is_attached(&self, agent_id: &str) -> bool

Whether a per-agent attach task is currently registered (test/obs helper).

pub async fn restart_caliband(&self, repo: &str) -> Result<()>

Gracefully shut down a repo’s caliband daemon and drop its cached client so the next access re-runs discovery (respawning with the current env).

pub async fn set_repo_config( &self, repo: &str, config: RepoProviderConfig, ) -> Result<()>

Persist a repo’s provider config and restart its caliband to apply it.

pub async fn run(self)

Run the background poll loop until Self::begin_shutdown is signalled.

Each iteration runs a complete poll cycle (never abandoned mid-append), then waits the interval. A shutdown signal stops scheduling new polls and returns after the in-flight cycle finishes — so the daemon can drain cleanly rather than being killed mid-iteration.

Trait Implementations

impl Clone for FleetManager

fn clone(&self) -> FleetManager

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.