agentmux_srv/backend/agent_session/session_io.rs
1// Copyright 2026, AgentMux Corp.
2// SPDX-License-Identifier: Apache-2.0
3
4//! Read/write/append snapshot state for the `agent:<defId>:current` zone,
5//! including the global-mirror normalization + one-shot heal.
6
7use crate::backend::storage::filestore::FileStore;
8
9use super::global_store::global_transcript_store;
10use super::helpers::{ensure_file, write_zone_file};
11use super::zone_naming::{agent_current_zone, is_valid_definition_id, validate_and_current};
12
13// ---------------------------------------------------------------------------
14// File names within an agent session zone (mirrors per-block zone shape)
15// ---------------------------------------------------------------------------
16
17/// Full UI snapshot (JSON). Frontend reads this on pane mount.
18pub const SNAPSHOT_FILE: &str = "output.state.json";
19/// Raw NDJSON stream for crash-recovery replay.
20pub const OUTPUT_FILE: &str = "output";
21/// Receive-time sidecar for `output`: one NDJSON `{"off":<byte offset of the
22/// appended batch>,"ms":<unix ms>}` record per append batch. A pure addition —
23/// `output`'s own format and every existing parser are untouched; a missing or
24/// corrupt sidecar degrades reads to "no stamps" (today's behavior). Gives
25/// replayed transcript nodes real timestamps (hover peek after restore, day
26/// separators). Spec:
27/// SPEC_AGENT_PANE_SESSION_SCOPED_SCROLLBACK_AND_AGENT_HISTORY_VIEW_2026_08_09.md §4.4.
28pub const TSIDX_FILE: &str = "output.tsidx";
29
30/// Write `content` to `output.state.json` in `agent:<defId>:current`.
31/// Idempotent — creates the file if missing, overwrites otherwise.
32///
33/// The per-channel write is primary (preserving all existing same-channel
34/// behaviour); the snapshot overlay is *additionally* mirrored into the GLOBAL
35/// transcript store (when installed) so a cross-channel open — and the
36/// migrated-agent backfill — find a coherent zone (overlay + `output`
37/// together). Mirroring is best-effort: a global-store failure is logged, never
38/// propagated. See `docs/analysis/ANALYSIS_CROSS_CHANNEL_CONVERSATION_HISTORY_2026_06_14.md`.
39pub fn write_session_state(
40 filestore: &FileStore,
41 definition_id: &str,
42 content: &[u8],
43) -> Result<(), String> {
44 let zone = validate_and_current(definition_id)?;
45 write_zone_file(filestore, &zone, SNAPSHOT_FILE, content)?;
46 if let Some(gfs) = global_transcript_store() {
47 // The GLOBAL mirror must be AGENT-anchored, not channel-anchored.
48 // `sourceBlockId` names the LOCAL block whose `output` a restore reads,
49 // and that id only exists in the writing channel. A different channel
50 // opening the agent would read from a block it doesn't have, and the read
51 // fallback (`global_output_source`) — which resolves the agent zone via
52 // that block's LOCAL meta — can't anchor it, so history renders empty.
53 // Strip it to "" in the global copy so a cross-channel open anchors on its
54 // own fresh local block (which maps to the agent). The per-channel copy
55 // above keeps the real id for same-channel restore. See
56 // docs/retro/retro-legacy-agent-history-cross-channel-2026-06-16.md.
57 let global_content = normalize_snapshot_for_global(content);
58 if let Err(e) = write_zone_file(gfs, &zone, SNAPSHOT_FILE, &global_content) {
59 tracing::warn!(zone = %zone, error = %e, "global transcripts: snapshot mirror failed");
60 }
61 }
62 Ok(())
63}
64
65/// Strip `sourceBlockId` to "" for the GLOBAL (cross-channel) snapshot mirror.
66///
67/// In a local snapshot `sourceBlockId` names the block whose per-block `output` a
68/// restore reads. That id is channel-scoped, so a global copy carrying it is only
69/// usable by the writing channel — every other channel's reader can't anchor it.
70/// "" is the agent-anchored sentinel that makes the reader fall back to the opening
71/// channel's own block. Best-effort: returns the input unchanged if it isn't a
72/// JSON object.
73pub fn normalize_snapshot_for_global(content: &[u8]) -> Vec<u8> {
74 let Ok(mut v) = serde_json::from_slice::<serde_json::Value>(content) else {
75 return content.to_vec();
76 };
77 if let Some(obj) = v.as_object_mut() {
78 if obj.contains_key("sourceBlockId") {
79 obj.insert(
80 "sourceBlockId".to_string(),
81 serde_json::Value::String(String::new()),
82 );
83 }
84 }
85 let result = serde_json::to_vec(&v).unwrap_or_else(|_| content.to_vec());
86 // Invariant G1: global snapshot must NEVER carry a non-empty sourceBlockId.
87 // A non-empty id is channel-local and breaks cross-channel opens.
88 debug_assert!(
89 serde_json::from_slice::<serde_json::Value>(&result)
90 .ok()
91 .and_then(|v| v.get("sourceBlockId").and_then(|s| s.as_str()).map(|s| s.is_empty()))
92 .unwrap_or(true),
93 "normalize_snapshot_for_global: G1 violated — sourceBlockId was not stripped"
94 );
95 result
96}
97
98/// One-shot heal for global snapshots poisoned before the normalize-on-mirror fix
99/// (a channel-local `sourceBlockId` was mirrored into `agent:<defId>:current`,
100/// breaking cross-channel opens). For each `def_id`, rewrite its global snapshot's
101/// `sourceBlockId` to "" iff it isn't already. Idempotent and cheap (one small
102/// JSON per agent); returns the number healed. Best-effort per agent.
103pub fn heal_global_snapshot_source_block_ids(gfs: &FileStore, def_ids: &[String]) -> usize {
104 let mut healed = 0;
105 for def_id in def_ids {
106 if !is_valid_definition_id(def_id) {
107 continue;
108 }
109 let zone = agent_current_zone(def_id);
110 let bytes = match gfs.read_file(&zone, SNAPSHOT_FILE) {
111 Ok(Some(b)) => b,
112 _ => continue, // no global snapshot for this agent
113 };
114 // Only rewrite when currently poisoned (non-empty sourceBlockId).
115 let poisoned = serde_json::from_slice::<serde_json::Value>(&bytes)
116 .ok()
117 .and_then(|v| {
118 v.get("sourceBlockId")
119 .and_then(|s| s.as_str())
120 .map(|s| !s.is_empty())
121 })
122 .unwrap_or(false);
123 if !poisoned {
124 continue;
125 }
126 let fixed = normalize_snapshot_for_global(&bytes);
127 if write_zone_file(gfs, &zone, SNAPSHOT_FILE, &fixed).is_ok() {
128 healed += 1;
129 tracing::info!(zone = %zone, "global transcripts: healed poisoned snapshot sourceBlockId");
130 }
131 }
132 healed
133}
134
135/// Append `line` (with a trailing newline added if not present) to
136/// `output` in `agent:<defId>:current`. Creates the file if missing.
137pub fn append_session_output(
138 filestore: &FileStore,
139 definition_id: &str,
140 line: &str,
141) -> Result<u64, String> {
142 let zone = validate_and_current(definition_id)?;
143 // Normalize to NDJSON: each line ends with exactly one '\n'.
144 let mut buf = line.as_bytes().to_vec();
145 if !buf.ends_with(b"\n") {
146 buf.push(b'\n');
147 }
148 ensure_file(filestore, &zone, OUTPUT_FILE)?;
149 filestore
150 .append_data(&zone, OUTPUT_FILE, &buf)
151 .map_err(|e| format!("append_data: {e}"))?;
152 Ok(buf.len() as u64)
153}
154
155/// Read `output.state.json` from `agent:<defId>:current`. Returns
156/// `Ok((None, None))` when the zone doesn't exist — that's the
157/// "fresh agent, nothing to restore" path and is NOT an error.
158///
159/// Reads the GLOBAL store first (preferred). The global copy always holds an
160/// agent-anchored snapshot (`sourceBlockId = ""`), which works in any channel.
161/// Per-channel is the fallback for old builds that pre-date the global store.
162/// Symmetric with the `blockfile:read_range` fallback in
163/// `app_api/blockfile.rs`.
164/// See `docs/specs/SPEC_AGENT_GLOBAL_PORTABILITY_2026-06-16.md` invariant G2.
165pub fn read_session_state(
166 filestore: &FileStore,
167 definition_id: &str,
168) -> Result<(Option<String>, Option<i64>), String> {
169 let zone = validate_and_current(definition_id)?;
170 // Read the global store first: it always holds the agent-anchored
171 // (sourceBlockId="") snapshot. The per-channel copy carries the
172 // writing channel's local block id, which is stale the moment any
173 // OTHER block opens the agent (same channel, different tab/build).
174 // Preferring global ensures cross-channel opens never get a foreign
175 // sourceBlockId — the root cause of the blank-pane regression.
176 // Per-channel is the fallback for pre-global-store builds only.
177 if let Some(gfs) = global_transcript_store() {
178 if let Some(found) = read_snapshot_from(gfs, &zone)? {
179 return Ok((Some(found.0), Some(found.1)));
180 }
181 }
182 if let Some(found) = read_snapshot_from(filestore, &zone)? {
183 return Ok((Some(found.0), Some(found.1)));
184 }
185 Ok((None, None))
186}
187
188/// Read `output.state.json` from `zone` in `store`. `Ok(None)` when absent.
189pub(super) fn read_snapshot_from(store: &FileStore, zone: &str) -> Result<Option<(String, i64)>, String> {
190 let stat = store
191 .stat(zone, SNAPSHOT_FILE)
192 .map_err(|e| format!("stat: {e}"))?;
193 let Some(file) = stat else {
194 return Ok(None);
195 };
196 let bytes = store
197 .read_file(zone, SNAPSHOT_FILE)
198 .map_err(|e| format!("read_file: {e}"))?
199 .unwrap_or_default();
200 Ok(Some((String::from_utf8_lossy(&bytes).into_owned(), file.modts)))
201}