agentmux_srv/backend/blockcontroller/core.rs
1// Copyright 2026, AgentMux Corp.
2// SPDX-License-Identifier: Apache-2.0
3
4//! Shared helpers used by all block controllers (persistent, subprocess, acp).
5//!
6//! Extracted to avoid copy-paste across the four controller files. Each
7//! controller keeps only its transport-specific logic (PTY / stream-json / ACP)
8//! and delegates the common mechanics here.
9
10use std::collections::HashMap;
11use std::sync::Arc;
12
13use crate::backend::eventbus::EventBus;
14use crate::backend::storage::store::Store;
15
16/// Block metadata key for the persisted agent session ID.
17/// Used by the "My Agents" reattach path on the frontend: after a tab reload
18/// the picker reads `block.meta["agent:sessionid"]` and passes it as the
19/// `--resume` value so the CLI picks up the prior conversation.
20pub(crate) const META_SESSION_ID: &str = "agent:sessionid";
21
22/// Block metadata key for the last classified agent failure.
23/// Written on every non-zero / in-band-error exit; cleared on clean success.
24/// The frontend reads this on pane mount so the recovery banner survives
25/// tab switches and page reloads without requiring the MPS event to be
26/// received in real time.
27pub(crate) const META_LAST_FAILURE: &str = "agent:last_failure";
28
29/// Expand the working directory, create it if missing, set it on `cmd`, and
30/// apply all env-var overrides from `env_vars`.
31///
32/// Call this from each controller's spawn routine BEFORE any transport-specific
33/// `Command` flags (e.g. `CREATE_NO_WINDOW` on Windows, which must come after
34/// the env setup in some implementations).
35pub(crate) fn apply_working_dir(
36 cmd: &mut tokio::process::Command,
37 block_id: &str,
38 working_dir: &str,
39 env_vars: &HashMap<String, String>,
40) {
41 if !working_dir.is_empty() {
42 let expanded_dir = expand_home_dir(working_dir);
43 let dir_path = std::path::Path::new(&expanded_dir);
44 if !dir_path.exists() {
45 if let Err(e) = std::fs::create_dir_all(dir_path) {
46 tracing::warn!(
47 block_id = %block_id,
48 dir = %expanded_dir,
49 error = %e,
50 "failed to create working directory",
51 );
52 }
53 }
54 if dir_path.exists() {
55 cmd.current_dir(&expanded_dir);
56 // Warn loudly if the agent workspace contains a nested .git (e.g.
57 // an unintended `git clone` inside ~/.agentmux/agents/). A stale
58 // nested clone can confuse agents into reading old code and waste
59 // gigabytes of disk. Single fs::metadata call — no directory walk.
60 let looks_like_agent_workspace = expanded_dir.contains("/.agentmux/agents/")
61 || expanded_dir.contains("\\.agentmux\\agents\\");
62 if looks_like_agent_workspace {
63 let git_dir = dir_path.join(".git");
64 if git_dir.exists() {
65 tracing::warn!(
66 block_id = %block_id,
67 cwd = %expanded_dir,
68 ".git detected inside agent workspace — this is usually \
69 an unintended nested clone and can waste gigabytes of \
70 disk. Clean up with: rm -rf {}/.git",
71 expanded_dir,
72 );
73 }
74 }
75 }
76 }
77 // Strip this instance's identity before applying the caller's overlay.
78 // `tokio::process::Command` inherits srv's environment wholesale, so an
79 // agent pane — and every process the agent launches from its shell tool —
80 // otherwise receives this instance's channel, data dir and cache dir.
81 // See `backend::pane_env`.
82 crate::backend::pane_env::sanitize_process_command(cmd);
83
84 for (k, v) in env_vars {
85 let expanded = crate::backend::base::expand_home_dir_safe(v);
86 cmd.env(k, expanded.to_string_lossy().as_ref());
87 }
88}
89
90/// Expand a leading `~/` or bare `~` to the user's home directory.
91pub(crate) fn expand_home_dir(dir: &str) -> String {
92 if dir.starts_with("~/") || dir == "~" {
93 if let Some(home) = dirs::home_dir() {
94 return home
95 .join(dir.trim_start_matches("~/"))
96 .to_string_lossy()
97 .to_string();
98 }
99 }
100 dir.to_string()
101}
102
103/// Persist a newly captured session ID to block metadata and broadcast the
104/// `waveobj:update` event so the frontend reflects the change immediately.
105///
106/// Every controller that captures a session id goes through this one path
107/// (persistent/stdout_reader.rs and stderr_reader.rs, subprocess/host_spawn.rs
108/// and container_spawn.rs, acp.rs, app_server_controller.rs); ACP used to only set
109/// `inner.session_id` in memory.
110///
111/// Also writes through to the matching `db_agent_instances` row's
112/// `session_id` column (SPEC_PANE_CLOSE_REOPEN_CONTINUITY_GUARANTEE_2026_07_27.md
113/// §4.1). Without this, that column was set to `""` at instance creation and
114/// never updated again in production — `ListRecentSessionsCommand` prefers
115/// the local row over the shared registry whenever both exist, so the "My
116/// Agents" picker's reattach flow (`AgentPicker.tsx`'s `handleReattach`) was
117/// handed an empty session id even mid-session, and — per
118/// `agent-model.ts`'s own documented invariant that an empty
119/// `continueSessionId` clears `agent:sessionid` on the new block — silently
120/// started a genuinely fresh conversation instead of resuming. Best-effort:
121/// logs and continues on failure, since the block-meta write above (the
122/// live-turn source of truth) already succeeded.
123///
124/// No-ops silently when `mstore` is `None` (e.g. in unit tests that don't wire
125/// up a store).
126pub(crate) fn persist_session_id(
127 block_id: &str,
128 sid: &str,
129 mstore: &Option<Arc<Store>>,
130 event_bus: &Option<Arc<EventBus>>,
131) {
132 let Some(ref store) = mstore else {
133 return;
134 };
135 let oref_str = format!("block:{}", block_id);
136 let mut meta_update = crate::backend::obj::MetaMapType::new();
137 meta_update.insert(
138 META_SESSION_ID.to_string(),
139 serde_json::Value::String(sid.to_string()),
140 );
141 match crate::server::service::update_object_meta(store, &oref_str, &meta_update) {
142 Err(e) => {
143 tracing::warn!(
144 block_id = %block_id,
145 error = %e,
146 "failed to persist agent:sessionid",
147 );
148 }
149 Ok(_) => {
150 sync_instance_session_id(store, block_id, sid);
151
152 let Some(ref event_bus) = event_bus else {
153 return;
154 };
155 broadcast_block_update(store, event_bus, block_id);
156 }
157 }
158}
159
160/// Write `sid` into the `db_agent_instances` row for `block_id`, if one
161/// exists. Best-effort — no-ops (with a debug log, not a warn — most turns
162/// on a block with no matching instance row are expected, e.g. terminal
163/// panes never have one) when there's nothing to update.
164fn sync_instance_session_id(store: &Arc<Store>, block_id: &str, sid: &str) {
165 let instance = match store.instance_get_by_block_id(block_id) {
166 Ok(Some(i)) => i,
167 Ok(None) => return,
168 Err(e) => {
169 tracing::debug!(
170 block_id = %block_id,
171 error = %e,
172 "sync_instance_session_id: instance lookup failed"
173 );
174 return;
175 }
176 };
177 let upd = crate::backend::storage::InstanceUpdate {
178 session_id: Some(sid.to_string()),
179 ..Default::default()
180 };
181 if let Err(e) = store.instance_update_partial(&instance.id, &upd) {
182 tracing::warn!(
183 block_id = %block_id,
184 instance_id = %instance.id,
185 error = %e,
186 "sync_instance_session_id: failed to write session_id"
187 );
188 }
189}
190
191/// Persist or clear the last agent failure in block metadata.
192///
193/// Pass `Some(failure)` on a failed exit to write `agent:last_failure` into the
194/// block's meta so the pane can recover the recovery banner on any future load
195/// without needing the ephemeral MPS event. Pass `None` on a clean exit to
196/// remove the key (setting it to JSON null triggers `merge_meta`'s delete path).
197/// Broadcasts a `waveobj:update` so active frontend subscribers see the change
198/// immediately via the block atom, not just on next full load.
199pub(crate) fn persist_last_failure(
200 block_id: &str,
201 failure: Option<&crate::agents::failure::AgentFailure>,
202 mstore: &Option<Arc<Store>>,
203 event_bus: &Option<Arc<EventBus>>,
204) {
205 let Some(ref store) = mstore else {
206 return;
207 };
208 // On a clean exit (failure=None), only write null (which merge_meta uses to
209 // delete the key) if the key actually exists — otherwise every successful
210 // turn would trigger a redundant DB write + waveobj:update broadcast.
211 if failure.is_none() {
212 let key_exists = store
213 .must_get::<crate::backend::obj::Block>(block_id)
214 .ok()
215 .and_then(|b| b.meta.get(META_LAST_FAILURE).cloned())
216 .map(|v| !v.is_null())
217 .unwrap_or(false);
218 if !key_exists {
219 return;
220 }
221 }
222 // Every classified agent failure gets exactly one log line, here at the
223 // shared choke point rather than at the call sites. Deliberately does not
224 // enumerate or count them: a draft of the accompanying spec said "three",
225 // review found more, a recount said "nine", and review caught that the
226 // recount was wrong too. Every path funnels through this function, so this
227 // covers them all — including the ones nobody counted correctly, and the
228 // next one added.
229 //
230 // Before this, a classified failure left NO trace in the srv log at all:
231 // none of those sites logged, and the only logged classification was
232 // `agents/runner.rs`'s "agent run failed", which is a different path that a
233 // controller-driven turn never takes. That made a whole class of question
234 // unanswerable from logs — "did my cron agent hit a 429 overnight?" had no
235 // answer, and an empty grep looked like "it never happens" rather than
236 // "it is never recorded". See
237 // docs/specs/SPEC_HEADLESS_TRANSIENT_RETRY_2026_08_31.md §Phase 0.
238 //
239 // Logged only for `Some` — this is also called with `None` after every
240 // clean turn, and that is not an event worth a line.
241 if let Some(f) = failure {
242 tracing::warn!(
243 target: "agent-failure",
244 block_id = %block_id,
245 code = ?f.code,
246 retryable = f.retryable,
247 "agent failure classified: {}",
248 f.title,
249 );
250 }
251
252 let oref_str = format!("block:{}", block_id);
253 let mut meta_update = crate::backend::obj::MetaMapType::new();
254 let val = match failure {
255 Some(f) => match serde_json::to_value(f) {
256 Ok(v) => v,
257 Err(e) => {
258 tracing::warn!(
259 block_id = %block_id,
260 error = %e,
261 "failed to serialize AgentFailure for agent:last_failure — skipping meta write",
262 );
263 return;
264 }
265 },
266 None => serde_json::Value::Null, // null → merge_meta removes the key
267 };
268 meta_update.insert(META_LAST_FAILURE.to_string(), val);
269 match crate::server::service::update_object_meta(store, &oref_str, &meta_update) {
270 Err(e) => {
271 tracing::warn!(
272 block_id = %block_id,
273 error = %e,
274 "failed to persist agent:last_failure",
275 );
276 }
277 Ok(_) => {
278 let Some(ref bus) = event_bus else {
279 return;
280 };
281 broadcast_block_update(store, bus, block_id);
282 }
283 }
284}
285
286/// Broadcast the block's current state as a `waveobj:update` on `block:<id>`,
287/// so an open pane sees a meta write live. Does nothing if the block can't be read.
288pub(crate) fn broadcast_block_update(store: &Store, event_bus: &EventBus, block_id: &str) {
289 let Ok(block) = store.must_get::<crate::backend::obj::Block>(block_id) else {
290 return;
291 };
292 let update = crate::backend::obj::MuxObjUpdate {
293 updatetype: "update".into(),
294 otype: "block".into(),
295 oid: block_id.to_string(),
296 obj: Some(crate::backend::obj::mux_obj_to_value(&block)),
297 };
298 event_bus.broadcast_event(&crate::backend::eventbus::EventBus::typed_ws_event::<crate::backend::mps_events::MuxObjUpdated>(&update, format!("block:{block_id}")));
299}
300
301#[cfg(test)]
302mod tests {
303 use super::*;
304 use crate::backend::obj::{Block, MetaMapType};
305 use crate::backend::storage::store::{AgentDefinition, AgentInstance, InstanceStatus};
306
307 /// Minimal definition satisfying `db_agent_instances`'s FK on
308 /// `definition_id` — field values otherwise irrelevant to these tests.
309 fn sample_agent(id: &str) -> AgentDefinition {
310 AgentDefinition {
311 conversation_visibility: crate::backend::storage::agents::default_conversation_visibility(),
312 id: id.to_string(),
313 slug: id.to_string(),
314 name: id.to_string(),
315 icon: "✦".to_string(),
316 provider: "claude".to_string(),
317 description: String::new(),
318 working_directory: String::new(),
319 shell: String::new(),
320 provider_flags: String::new(),
321 auto_start: 0,
322 restart_on_crash: 0,
323 idle_timeout_minutes: 0,
324 created_at: 0,
325 agent_type: "host".to_string(),
326 environment: String::new(),
327 agent_bus_id: String::new(),
328 is_seeded: 0,
329 accounts: String::new(),
330 parent_id: String::new(),
331 branch_label: String::new(),
332 updated_at: 0,
333 user_hidden: 0,
334 container_image: String::new(),
335 container_volumes: "[]".to_string(),
336 container_name: String::new(),
337 use_ambient_login: 0,
338 auto_continue_enabled: 0,
339 model_vendor_base_url: String::new(),
340 memory_id: String::new(),
341 }
342 }
343
344 /// SPEC_PANE_CLOSE_REOPEN_CONTINUITY_GUARANTEE_2026_07_27.md §4.1: this
345 /// is the end-to-end assertion that `persist_session_id` (the live-turn
346 /// call site every controller uses) keeps `db_agent_instances.session_id`
347 /// current, not just the block's own `agent:sessionid` meta — closing the
348 /// gap where "My Agents"'s reattach flow saw a permanently-empty local
349 /// session id.
350 #[test]
351 fn persist_session_id_writes_through_to_the_instance_row() {
352 let store = Arc::new(Store::open_in_memory().unwrap());
353 let block_id = "55555555-5555-5555-5555-555555555555";
354 let mut block = Block {
355 oid: block_id.to_string(),
356 parentoref: String::new(),
357 version: 1,
358 runtimeopts: None,
359 stickers: None,
360 meta: {
361 let mut m = MetaMapType::new();
362 m.insert("view".to_string(), serde_json::json!("agent"));
363 m
364 },
365 subblockids: None,
366 };
367 store.insert(&mut block).unwrap();
368
369 let mut def = sample_agent("def-live");
370 store.agent_def_insert(&mut def).unwrap();
371
372 let inst = AgentInstance {
373 id: "inst-live".to_string(),
374 definition_id: "def-live".to_string(),
375 parent_instance_id: String::new(),
376 block_id: block_id.to_string(),
377 session_id: String::new(),
378 status: InstanceStatus::Running.as_str().to_string(),
379 github_context: String::new(),
380 started_at: 1000,
381 ended_at: 0,
382 created_at: 1000,
383 identity_id: String::new(),
384 memory_id: String::new(),
385 instance_name: String::new(),
386 working_directory: String::new(),
387 display_hidden: false,
388 };
389 store.instance_create(&inst).unwrap();
390
391 persist_session_id(block_id, "sess-abc", &Some(store.clone()), &None);
392
393 // Block meta: the pre-existing behavior.
394 let updated_block: Block = store.get(block_id).unwrap().unwrap();
395 assert_eq!(
396 updated_block.meta.get(META_SESSION_ID).and_then(|v| v.as_str()),
397 Some("sess-abc"),
398 );
399
400 // Instance row: the new write-through.
401 let updated_instance = store.instance_get_by_block_id(block_id).unwrap().unwrap();
402 assert_eq!(updated_instance.session_id, "sess-abc");
403 }
404
405 /// A block with no matching instance row (e.g. a terminal pane) must not
406 /// cause persist_session_id to error or panic — sync_instance_session_id
407 /// is best-effort and silently no-ops.
408 #[test]
409 fn persist_session_id_is_a_noop_for_a_block_with_no_instance_row() {
410 let store = Arc::new(Store::open_in_memory().unwrap());
411 let block_id = "66666666-6666-6666-6666-666666666666";
412 let mut block = Block {
413 oid: block_id.to_string(),
414 parentoref: String::new(),
415 version: 1,
416 runtimeopts: None,
417 stickers: None,
418 meta: MetaMapType::new(),
419 subblockids: None,
420 };
421 store.insert(&mut block).unwrap();
422
423 // Must not panic.
424 persist_session_id(block_id, "sess-xyz", &Some(store.clone()), &None);
425
426 let updated_block: Block = store.get(block_id).unwrap().unwrap();
427 assert_eq!(
428 updated_block.meta.get(META_SESSION_ID).and_then(|v| v.as_str()),
429 Some("sess-xyz"),
430 "block meta write still happens regardless of instance-row presence",
431 );
432 }
433}