1use crate::protocol::{
2 AgentId, ManagedSessionRecord, ManagedSessionState, ManagedWorktreeLeaseId,
3 ManagedWorktreeSpawnReceipt, NodeId, NodeIncarnationId, OpaqueHostPath,
4 ResolvedBundleReceipt, ResolvedContextPackReceipt, ResolvedEnvironmentProfileReceipt,
5 SessionAddress, SessionMode,
6 SessionRecordId, SpawnIdempotencyKey, WorktreeProfileId,
7 WorktreeProfileRevision, WorkspaceId,
8 MAX_NODE_TEXT_BYTES, MAX_SESSION_DISPLAY_NAME_BYTES, MAX_WORKSPACE_ROOT_BYTES,
9 NODE_STATE_SCHEMA_V1, NODE_STATE_SCHEMA_V2, NODE_STATE_SCHEMA_V3, NODE_STATE_SCHEMA_V4,
10 NODE_STATE_SCHEMA_V5,
11 NODE_STATE_SCHEMA_V6, NODE_STATE_SCHEMA_V7, NODE_STATE_SCHEMA_V8,
12 NODE_STATE_SCHEMA_V9, NODE_STATE_SCHEMA_V10,
13};
14use crate::managed_worktree::ManagedWorktreeLeaseRecord;
15use crate::materialization::{
16 opaque_to_path, path_to_opaque, MaterializationId, MaterializationOwner,
17 MaterializationOwnershipRecord, MaterializationState, MaterializedPathDeclaration,
18 MAX_SESSION_MATERIALIZATIONS,
19};
20use serde::{Deserialize, Serialize};
21use std::collections::{BTreeMap, BTreeSet};
22use std::error::Error as StdError;
23use std::fmt;
24use std::fs::{self, File, OpenOptions};
25use std::io::{self, Read, Write};
26#[cfg(windows)]
27use std::os::windows::fs::OpenOptionsExt;
28#[cfg(unix)]
29use std::os::unix::fs::{MetadataExt, OpenOptionsExt, PermissionsExt};
30use std::path::{Path, PathBuf};
31use std::sync::atomic::{AtomicU64, Ordering};
32
33const MAX_STATE_BYTES: u64 = 4 * 1024 * 1024;
34const MAX_PERSISTED_WORKSPACES: usize = 256;
35const MAX_PERSISTED_MANAGED_WORKTREE_BRANCH_BYTES: usize = 512;
36pub const MAX_MANAGED_WORKTREE_SPAWN_REPLAYS: usize = 256;
37pub const MAX_MANAGED_SESSION_RECORDS: usize = 4_096;
38const BACKUP_ROTATION_WARNING: &str = "durable-state-backup-rotation-failed";
39const CORRUPT_PRIMARY_PRESERVED_WARNING: &str = "durable-state-corrupt-primary-preserved";
40pub const DURABLE_STATE_COMMIT_FAILED: &str = "durable-state-commit-failed";
41pub const DURABLE_STATE_RECOVERY_WARNING: &str = "durable-state-recovery-required";
42pub const DURABLE_STATE_WORKSPACE_WARNING: &str =
43 "durable-state-workspace-unavailable";
44static NEXT_TEMPORARY_FILE: AtomicU64 = AtomicU64::new(1);
45
46pub type ProviderSessionSemanticKey = (String, u8, String);
47
48pub fn provider_session_semantic_key(
49 provider: &AgentId,
50 identity: &gate4agent_types::ProviderSessionIdentity,
51) -> ProviderSessionSemanticKey {
52 let key = match identity.key {
53 gate4agent_types::ProviderSessionKey::SessionId => 0,
54 gate4agent_types::ProviderSessionKey::ConversationId => 1,
55 };
56 (provider.as_str().to_owned(), key, identity.id.clone())
57}
58
59pub fn same_provider_session(
60 left_provider: &AgentId,
61 left: &gate4agent_types::ProviderSessionIdentity,
62 right_provider: &AgentId,
63 right: &gate4agent_types::ProviderSessionIdentity,
64) -> bool {
65 left_provider == right_provider && left.key == right.key && left.id == right.id
66}
67
68pub fn sanitized_persistence_summary(message: &str) -> String {
69 match message {
70 BACKUP_ROTATION_WARNING
71 | CORRUPT_PRIMARY_PRESERVED_WARNING
72 | DURABLE_STATE_COMMIT_FAILED
73 | DURABLE_STATE_RECOVERY_WARNING
74 | DURABLE_STATE_WORKSPACE_WARNING => message.to_owned(),
75 _ => DURABLE_STATE_COMMIT_FAILED.to_owned(),
76 }
77}
78
79pub struct StatePathLock {
80 file: Option<File>,
81 #[cfg(windows)]
82 path: PathBuf,
83}
84
85impl StatePathLock {
86 pub fn acquire(path: Option<&Path>) -> io::Result<Option<Self>> {
87 let Some(path) = path else {
88 return Ok(None);
89 };
90 let parent = path.parent().ok_or_else(|| {
91 io::Error::new(io::ErrorKind::InvalidInput, "durable state path has no parent")
92 })?;
93 fs::create_dir_all(parent)?;
94 #[cfg(unix)]
95 secure_state_directory(parent)?;
96 let lock_path = sibling_path(path, "lock");
97 #[cfg(windows)]
98 let file = OpenOptions::new()
99 .create(true)
100 .read(true)
101 .write(true)
102 .share_mode(0)
103 .open(&lock_path)
104 .map_err(|error| {
105 if error.raw_os_error() == Some(32) {
106 io::Error::new(
107 io::ErrorKind::WouldBlock,
108 "durable state path is already owned by another node process",
109 )
110 } else {
111 error
112 }
113 })?;
114 #[cfg(unix)]
115 let file = acquire_unix_state_lock(&lock_path)?;
116 Ok(Some(Self {
117 file: Some(file),
118 #[cfg(windows)]
119 path: lock_path,
120 }))
121 }
122}
123
124impl Drop for StatePathLock {
125 fn drop(&mut self) {
126 #[cfg(unix)]
127 if let Some(file) = self.file.as_ref() {
128 unsafe {
129 libc::flock(file.as_raw_fd(), libc::LOCK_UN);
130 }
131 }
132 drop(self.file.take());
133 #[cfg(windows)]
134 let _ = fs::remove_file(&self.path);
135 }
136}
137
138#[cfg(unix)]
139use std::os::fd::AsRawFd;
140
141#[cfg(unix)]
142fn secure_state_directory(path: &Path) -> io::Result<()> {
143 let metadata = fs::symlink_metadata(path)?;
144 if !metadata.file_type().is_dir() || metadata.uid() != unsafe { libc::geteuid() } {
145 return Err(io::Error::new(
146 io::ErrorKind::PermissionDenied,
147 "durable state directory is not owned by the current user",
148 ));
149 }
150 fs::set_permissions(path, fs::Permissions::from_mode(0o700))?;
151 let metadata = fs::symlink_metadata(path)?;
152 if metadata.permissions().mode() & 0o7777 != 0o700 {
153 return Err(io::Error::new(
154 io::ErrorKind::PermissionDenied,
155 "durable state directory must have mode 0700",
156 ));
157 }
158 Ok(())
159}
160
161#[cfg(unix)]
162fn acquire_unix_state_lock(path: &Path) -> io::Result<File> {
163 let file = OpenOptions::new()
164 .create(true)
165 .read(true)
166 .write(true)
167 .mode(0o600)
168 .custom_flags(libc::O_CLOEXEC | libc::O_NOFOLLOW)
169 .open(path)?;
170 let metadata = file.metadata()?;
171 if !metadata.file_type().is_file()
172 || metadata.uid() != unsafe { libc::geteuid() }
173 {
174 return Err(io::Error::new(
175 io::ErrorKind::PermissionDenied,
176 "durable state lock is not an owner-controlled regular file",
177 ));
178 }
179 file.set_permissions(fs::Permissions::from_mode(0o600))?;
180 let result = unsafe { libc::flock(file.as_raw_fd(), libc::LOCK_EX | libc::LOCK_NB) };
181 if result != 0 {
182 let error = io::Error::last_os_error();
183 if error.kind() == io::ErrorKind::WouldBlock {
184 return Err(io::Error::new(
185 io::ErrorKind::WouldBlock,
186 "durable state path is already owned by another node process",
187 ));
188 }
189 return Err(error);
190 }
191 Ok(file)
192}
193
194#[derive(Clone, Eq, PartialEq)]
195pub struct LoadedNodeState {
196 pub workspaces: BTreeMap<WorkspaceId, String>,
197 pub records: Vec<ManagedSessionRecord>,
198 pub managed_worktrees: Vec<ManagedWorktreeLeaseRecord>,
199 pub managed_worktree_tombstones: Vec<ManagedWorktreeLeaseRecord>,
200 pub materializations: Vec<MaterializationOwnershipRecord>,
201 pub managed_spawn_replays: Vec<ManagedWorktreeSpawnReplayRecordV10>,
202 pub warning: Option<String>,
203}
204
205impl LoadedNodeState {
206 fn empty() -> Self {
207 Self {
208 workspaces: BTreeMap::new(),
209 records: Vec::new(),
210 managed_worktrees: Vec::new(),
211 managed_worktree_tombstones: Vec::new(),
212 materializations: Vec::new(),
213 managed_spawn_replays: Vec::new(),
214 warning: None,
215 }
216 }
217}
218
219impl fmt::Debug for LoadedNodeState {
220 fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
221 formatter
222 .debug_struct("LoadedNodeState")
223 .field("workspace_count", &self.workspaces.len())
224 .field("record_count", &self.records.len())
225 .field("managed_worktree_count", &self.managed_worktrees.len())
226 .field("managed_worktree_tombstone_count", &self.managed_worktree_tombstones.len())
227 .field("materialization_count", &self.materializations.len())
228 .field("managed_spawn_replay_count", &self.managed_spawn_replays.len())
229 .field("warning", &self.warning)
230 .finish()
231 }
232}
233
234#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
235struct PersistedNodeStateV1 {
236 version: u16,
237 node_id: NodeId,
238 workspaces: Vec<PersistedWorkspaceV1>,
239 session_records: Vec<PersistedManagedSessionRecordV1>,
240}
241
242#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
243struct PersistedNodeStateV2 {
244 version: u16,
245 node_id: NodeId,
246 workspaces: Vec<PersistedWorkspaceV2>,
247 session_records: Vec<PersistedManagedSessionRecordV2>,
248}
249
250#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
251struct PersistedNodeStateV3 {
252 version: u16,
253 node_id: NodeId,
254 workspaces: Vec<PersistedWorkspaceV2>,
255 session_records: Vec<PersistedManagedSessionRecordV3>,
256}
257
258#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
259struct PersistedNodeStateV4 {
260 version: u16,
261 node_id: NodeId,
262 workspaces: Vec<PersistedWorkspaceV2>,
263 session_records: Vec<PersistedManagedSessionRecordV3>,
264 managed_worktrees: Vec<ManagedWorktreeLeaseRecord>,
265 managed_worktree_tombstones: Vec<ManagedWorktreeLeaseRecord>,
266}
267
268#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
269struct PersistedNodeStateV5 {
270 version: u16,
271 node_id: NodeId,
272 workspaces: Vec<PersistedWorkspaceV2>,
273 session_records: Vec<PersistedManagedSessionRecordV5>,
274 managed_worktrees: Vec<ManagedWorktreeLeaseRecord>,
275 managed_worktree_tombstones: Vec<ManagedWorktreeLeaseRecord>,
276}
277
278#[derive(Clone, Deserialize, Eq, PartialEq, Serialize)]
279struct PersistedNodeStateV6 {
280 version: u16,
281 node_id: NodeId,
282 workspaces: Vec<PersistedWorkspaceV2>,
283 session_records: Vec<PersistedManagedSessionRecordV5>,
284 managed_worktrees: Vec<ManagedWorktreeLeaseRecord>,
285 managed_worktree_tombstones: Vec<ManagedWorktreeLeaseRecord>,
286 materializations: Vec<PersistedMaterializationRecordV6>,
287}
288
289#[derive(Clone, Deserialize, Eq, PartialEq, Serialize)]
290struct PersistedNodeStateV7 {
291 version: u16,
292 node_id: NodeId,
293 workspaces: Vec<PersistedWorkspaceV2>,
294 session_records: Vec<ManagedSessionRecord>,
295 managed_worktrees: Vec<ManagedWorktreeLeaseRecord>,
296 managed_worktree_tombstones: Vec<ManagedWorktreeLeaseRecord>,
297 materializations: Vec<PersistedMaterializationRecordV7>,
298}
299
300#[derive(Clone, Deserialize, Eq, PartialEq, Serialize)]
301struct PersistedNodeStateV8 {
302 version: u16,
303 node_id: NodeId,
304 workspaces: Vec<PersistedWorkspaceV2>,
305 session_records: Vec<ManagedSessionRecord>,
306 managed_worktrees: Vec<ManagedWorktreeLeaseRecord>,
307 managed_worktree_tombstones: Vec<ManagedWorktreeLeaseRecord>,
308 materializations: Vec<PersistedMaterializationRecordV8>,
309}
310
311type PersistedNodeStateV9 = PersistedNodeStateV8;
312
313#[derive(Clone, Deserialize, Eq, PartialEq, Serialize)]
314#[serde(deny_unknown_fields)]
315struct PersistedNodeStateV10 {
316 version: u16,
317 node_id: NodeId,
318 workspaces: Vec<PersistedWorkspaceV2>,
319 session_records: Vec<ManagedSessionRecord>,
320 managed_worktrees: Vec<ManagedWorktreeLeaseRecord>,
321 managed_worktree_tombstones: Vec<ManagedWorktreeLeaseRecord>,
322 materializations: Vec<PersistedMaterializationRecordV8>,
323 managed_spawn_replays: Vec<ManagedWorktreeSpawnReplayRecordV10>,
324}
325
326#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
327#[serde(deny_unknown_fields)]
328pub struct ManagedWorktreeSpawnReplayRecordV10 {
329 pub idempotency_key: SpawnIdempotencyKey,
330 pub request_digest: String,
331 pub source_workspace_id: WorkspaceId,
332 pub profile_id: WorktreeProfileId,
333 pub expected_profile_revision: WorktreeProfileRevision,
334 pub state: ManagedWorktreeSpawnReplayStateV10,
335 pub created_at_unix_ms: u64,
336 pub updated_at_unix_ms: u64,
337}
338
339#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
340#[serde(tag = "state", rename_all = "kebab-case", deny_unknown_fields)]
341pub enum ManagedWorktreeSpawnReplayStateV10 {
342 Pending {
343 owner_incarnation_id: NodeIncarnationId,
344 #[serde(default, skip_serializing_if = "Option::is_none")]
345 lease_id: Option<ManagedWorktreeLeaseId>,
346 },
347 Committed {
348 receipt: ManagedWorktreeSpawnReceipt,
349 },
350}
351
352struct DecodedNodeState {
353 node_id: NodeId,
354 workspaces: Vec<(WorkspaceId, String)>,
355 session_records: Vec<ManagedSessionRecord>,
356 managed_worktrees: Vec<ManagedWorktreeLeaseRecord>,
357 managed_worktree_tombstones: Vec<ManagedWorktreeLeaseRecord>,
358 materializations: Vec<MaterializationOwnershipRecord>,
359 managed_spawn_replays: Vec<ManagedWorktreeSpawnReplayRecordV10>,
360}
361
362#[derive(Deserialize)]
363struct PersistedNodeStateHeader {
364 version: u16,
365}
366
367#[derive(Clone, Copy, Debug, Eq, PartialEq)]
368pub enum StateLoadRefusal {
369 UnsupportedSchema,
370 PathSemanticsUnsupported,
371 NodeIdentityMismatch,
372}
373
374#[derive(Debug)]
375enum StateLoadRefusalError {
376 UnsupportedSchema(u16),
377 PathSemanticsUnsupported,
378 NodeIdentityMismatch,
379}
380
381impl StateLoadRefusalError {
382 fn refusal(&self) -> StateLoadRefusal {
383 match self {
384 Self::UnsupportedSchema(_) => StateLoadRefusal::UnsupportedSchema,
385 Self::PathSemanticsUnsupported => StateLoadRefusal::PathSemanticsUnsupported,
386 Self::NodeIdentityMismatch => StateLoadRefusal::NodeIdentityMismatch,
387 }
388 }
389}
390
391impl fmt::Display for StateLoadRefusalError {
392 fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
393 match self {
394 Self::UnsupportedSchema(version) => {
395 write!(formatter, "unsupported durable state version {version}")
396 }
397 Self::PathSemanticsUnsupported => {
398 formatter.write_str("durable state path semantics are unsupported on this host")
399 }
400 Self::NodeIdentityMismatch => {
401 formatter.write_str("durable state belongs to a different node")
402 }
403 }
404 }
405}
406
407impl StdError for StateLoadRefusalError {}
408
409#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
410struct PersistedWorkspaceV1 {
411 workspace_id: WorkspaceId,
412 canonical_root: String,
413}
414
415#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
416struct PersistedWorkspaceV2 {
417 workspace_id: WorkspaceId,
418 canonical_root: OpaqueHostPath,
419}
420
421#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
422struct PersistedManagedSessionRecordV1 {
423 record_id: SessionRecordId,
424 display_name: String,
425 provider: LegacyAgentProvider,
426 mode: SessionMode,
427 state: ManagedSessionState,
428 workspace_id: WorkspaceId,
429 canonical_root: String,
430 provider_session: Option<gate4agent_types::ProviderSessionIdentity>,
431 active_session: Option<SessionAddress>,
432 created_at_unix_ms: u64,
433 updated_at_unix_ms: u64,
434 last_error: Option<String>,
435}
436
437#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
438struct PersistedManagedSessionRecordV2 {
439 record_id: SessionRecordId,
440 display_name: String,
441 provider: LegacyAgentProvider,
442 mode: SessionMode,
443 state: ManagedSessionState,
444 workspace_id: WorkspaceId,
445 canonical_root: OpaqueHostPath,
446 provider_session: Option<gate4agent_types::ProviderSessionIdentity>,
447 active_session: Option<SessionAddress>,
448 created_at_unix_ms: u64,
449 updated_at_unix_ms: u64,
450 last_error: Option<String>,
451}
452
453#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
454#[serde(rename_all = "kebab-case")]
455enum LegacyAgentProvider {
456 Claude,
457 Codex,
458 Kimi,
459}
460
461impl LegacyAgentProvider {
462 fn into_agent_id(self) -> AgentId {
463 let value = match self {
464 Self::Claude => "claude",
465 Self::Codex => "codex",
466 Self::Kimi => "kimi",
467 };
468 AgentId::new(value).expect("legacy provider IDs remain valid")
469 }
470
471 #[cfg(test)]
472 fn from_agent_id(provider: &AgentId) -> Self {
473 match provider.as_str() {
474 "claude" => Self::Claude,
475 "codex" => Self::Codex,
476 "kimi" => Self::Kimi,
477 value => panic!("{value} is not a legacy provider ID"),
478 }
479 }
480}
481
482#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
483struct PersistedManagedSessionRecordV3 {
484 record_id: SessionRecordId,
485 display_name: String,
486 provider: AgentId,
487 mode: SessionMode,
488 state: ManagedSessionState,
489 workspace_id: WorkspaceId,
490 canonical_root: OpaqueHostPath,
491 provider_session: Option<gate4agent_types::ProviderSessionIdentity>,
492 active_session: Option<SessionAddress>,
493 created_at_unix_ms: u64,
494 updated_at_unix_ms: u64,
495 last_error: Option<String>,
496}
497
498#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
499struct PersistedManagedSessionRecordV5 {
500 record_id: SessionRecordId,
501 display_name: String,
502 provider: AgentId,
503 mode: SessionMode,
504 state: ManagedSessionState,
505 workspace_id: WorkspaceId,
506 canonical_root: OpaqueHostPath,
507 provider_session: Option<gate4agent_types::ProviderSessionIdentity>,
508 active_session: Option<SessionAddress>,
509 environment_profile: Option<ResolvedEnvironmentProfileReceipt>,
510 created_at_unix_ms: u64,
511 updated_at_unix_ms: u64,
512 last_error: Option<String>,
513}
514
515#[derive(Clone, Deserialize, Eq, PartialEq, Serialize)]
516#[serde(deny_unknown_fields)]
517struct PersistedMaterializationRecordV6 {
518 materialization_id: MaterializationId,
519 environment_profile: ResolvedEnvironmentProfileReceipt,
520 owner: MaterializationOwner,
521 managed_lease_id: Option<crate::protocol::ManagedWorktreeLeaseId>,
522 state: MaterializationState,
523 root: OpaqueHostPath,
524 provider_home: OpaqueHostPath,
525 declared_paths: Vec<MaterializedPathDeclaration>,
526 created_at_unix_ms: u64,
527 updated_at_unix_ms: u64,
528}
529
530#[derive(Clone, Deserialize, Eq, PartialEq, Serialize)]
531#[serde(deny_unknown_fields)]
532struct PersistedMaterializationRecordV7 {
533 materialization_id: MaterializationId,
534 environment_profile: Option<ResolvedEnvironmentProfileReceipt>,
535 bundle: Option<ResolvedBundleReceipt>,
536 owner: MaterializationOwner,
537 managed_lease_id: Option<crate::protocol::ManagedWorktreeLeaseId>,
538 state: MaterializationState,
539 root: OpaqueHostPath,
540 provider_home: OpaqueHostPath,
541 bundle_root: OpaqueHostPath,
542 plugin_data: OpaqueHostPath,
543 declared_paths: Vec<MaterializedPathDeclaration>,
544 created_at_unix_ms: u64,
545 updated_at_unix_ms: u64,
546}
547
548#[derive(Clone, Deserialize, Eq, PartialEq, Serialize)]
549#[serde(deny_unknown_fields)]
550struct PersistedMaterializationRecordV8 {
551 materialization_id: MaterializationId,
552 environment_profile: Option<ResolvedEnvironmentProfileReceipt>,
553 bundle: Option<ResolvedBundleReceipt>,
554 context: Option<ResolvedContextPackReceipt>,
555 owner: MaterializationOwner,
556 managed_lease_id: Option<crate::protocol::ManagedWorktreeLeaseId>,
557 state: MaterializationState,
558 root: OpaqueHostPath,
559 provider_home: OpaqueHostPath,
560 bundle_root: OpaqueHostPath,
561 plugin_data: OpaqueHostPath,
562 declared_paths: Vec<MaterializedPathDeclaration>,
563 created_at_unix_ms: u64,
564 updated_at_unix_ms: u64,
565}
566
567pub fn load(path: Option<&Path>, expected_node_id: &NodeId) -> io::Result<LoadedNodeState> {
568 let Some(path) = path else {
569 return Ok(LoadedNodeState::empty());
570 };
571 let backup = sibling_path(path, "bak");
572 let pending_backup = sibling_path(path, "pending-backup");
573 if !path.exists() {
574 if !pending_backup.exists() && !backup.exists() {
575 return Ok(LoadedNodeState::empty());
576 }
577 return recover_from_copies(
578 path,
579 &pending_backup,
580 &backup,
581 expected_node_id,
582 "is missing".to_owned(),
583 );
584 }
585 match load_one(path, expected_node_id) {
586 Ok(loaded) => Ok(loaded),
587 Err(primary_error) if is_authoritative_state_error(&primary_error) => {
588 Err(primary_error)
589 }
590 Err(primary_error) => recover_from_copies(
591 path,
592 &pending_backup,
593 &backup,
594 expected_node_id,
595 format!("is invalid: {primary_error}"),
596 ),
597 }
598}
599
600fn recover_from_copies(
601 primary: &Path,
602 pending_backup: &Path,
603 backup: &Path,
604 expected_node_id: &NodeId,
605 _primary_problem: String,
606) -> io::Result<LoadedNodeState> {
607 for candidate in [pending_backup, backup] {
608 if !candidate.exists() {
609 continue;
610 }
611 match load_one(candidate, expected_node_id) {
612 Ok(mut loaded) => {
613 loaded.warning = Some(DURABLE_STATE_RECOVERY_WARNING.to_owned());
614 return Ok(loaded);
615 }
616 Err(error) if is_authoritative_state_error(&error) => {
617 return Err(error);
618 }
619 Err(_) => {}
620 }
621 }
622 let _ = primary;
623 Err(invalid_data("durable-state-recovery-failed"))
624}
625
626fn load_one(path: &Path, expected_node_id: &NodeId) -> io::Result<LoadedNodeState> {
627 let metadata = fs::metadata(path)?;
628 if metadata.len() > MAX_STATE_BYTES {
629 return Err(invalid_data("durable state exceeds the bounded file size"));
630 }
631 let mut bytes = Vec::with_capacity(metadata.len() as usize);
632 File::open(path)?
633 .take(MAX_STATE_BYTES + 1)
634 .read_to_end(&mut bytes)?;
635 if bytes.len() as u64 > MAX_STATE_BYTES {
636 return Err(invalid_data("durable state exceeds the bounded file size"));
637 }
638 let state = decode_persisted_state(&bytes)?;
639 if &state.node_id != expected_node_id {
640 return Err(io::Error::new(
641 io::ErrorKind::PermissionDenied,
642 StateLoadRefusalError::NodeIdentityMismatch,
643 ));
644 }
645 if state.workspaces.len() > MAX_PERSISTED_WORKSPACES {
646 return Err(invalid_data("durable state contains too many workspaces"));
647 }
648 if state.session_records.len() > MAX_MANAGED_SESSION_RECORDS {
649 return Err(invalid_data("durable state contains too many session records"));
650 }
651
652 let mut workspaces = BTreeMap::new();
653 let mut roots = BTreeSet::new();
654 for (workspace_id, canonical_root) in state.workspaces {
655 validate_root(&canonical_root)?;
656 if workspaces
657 .insert(workspace_id.clone(), canonical_root.clone())
658 .is_some()
659 {
660 return Err(invalid_data("durable state contains duplicate workspace IDs"));
661 }
662 if !roots.insert(crate::platform::root_identity(&canonical_root)) {
663 return Err(invalid_data("durable state contains duplicate workspace roots"));
664 }
665 }
666
667 let mut record_ids = BTreeSet::new();
668 let mut provider_sessions = BTreeSet::new();
669 let mut records = Vec::with_capacity(state.session_records.len());
670 for mut record in state.session_records {
671 if let Some(identity) = &mut record.provider_session {
672 identity.transcript_path = None;
673 }
674 validate_record(&record)?;
675 record.last_error = record
676 .last_error
677 .as_deref()
678 .map(sanitized_record_error_summary);
679 if !record_ids.insert(record.record_id.clone()) {
680 return Err(invalid_data("durable state contains duplicate session record IDs"));
681 }
682 if let Some(identity) = &mut record.provider_session {
683 let key = provider_session_semantic_key(&record.provider, identity);
684 if !provider_sessions.insert(key) {
685 return Err(invalid_data("durable state contains duplicate provider sessions"));
686 }
687 }
688 record.active_session = None;
689 record.state = if record.provider_session.is_some() {
690 ManagedSessionState::Dormant
691 } else {
692 ManagedSessionState::Unavailable
693 };
694 records.push(record);
695 }
696 Ok(LoadedNodeState {
697 workspaces,
698 records,
699 managed_worktrees: state.managed_worktrees,
700 managed_worktree_tombstones: state.managed_worktree_tombstones,
701 materializations: state.materializations,
702 managed_spawn_replays: state.managed_spawn_replays,
703 warning: None,
704 })
705}
706
707fn decode_persisted_state(bytes: &[u8]) -> io::Result<DecodedNodeState> {
708 let header: PersistedNodeStateHeader = serde_json::from_slice(bytes)
709 .map_err(|error| invalid_data(format!("durable state JSON is invalid: {error}")))?;
710 match header.version {
711 NODE_STATE_SCHEMA_V1 => {
712 let state: PersistedNodeStateV1 = serde_json::from_slice(bytes)
713 .map_err(|error| invalid_data(format!("durable state JSON is invalid: {error}")))?;
714 decode_v1_state(state)
715 }
716 NODE_STATE_SCHEMA_V2 => {
717 let state: PersistedNodeStateV2 = serde_json::from_slice(bytes)
718 .map_err(|error| invalid_data(format!("durable state JSON is invalid: {error}")))?;
719 decode_v2_state(state)
720 }
721 NODE_STATE_SCHEMA_V3 => {
722 let state: PersistedNodeStateV3 = serde_json::from_slice(bytes)
723 .map_err(|error| invalid_data(format!("durable state JSON is invalid: {error}")))?;
724 decode_v3_state(state)
725 }
726 NODE_STATE_SCHEMA_V4 => {
727 let state: PersistedNodeStateV4 = serde_json::from_slice(bytes)
728 .map_err(|error| invalid_data(format!("durable state JSON is invalid: {error}")))?;
729 decode_v4_state(state)
730 }
731 NODE_STATE_SCHEMA_V5 => {
732 let state: PersistedNodeStateV5 = serde_json::from_slice(bytes)
733 .map_err(|error| invalid_data(format!("durable state JSON is invalid: {error}")))?;
734 decode_v5_state(state)
735 }
736 NODE_STATE_SCHEMA_V6 => {
737 let state: PersistedNodeStateV6 = serde_json::from_slice(bytes)
738 .map_err(|error| invalid_data(format!("durable state JSON is invalid: {error}")))?;
739 decode_v6_state(state)
740 }
741 NODE_STATE_SCHEMA_V7 => {
742 let state: PersistedNodeStateV7 = serde_json::from_slice(bytes)
743 .map_err(|error| invalid_data(format!("durable state JSON is invalid: {error}")))?;
744 decode_v7_state(state)
745 }
746 NODE_STATE_SCHEMA_V8 => {
747 let state: PersistedNodeStateV8 = serde_json::from_slice(bytes)
748 .map_err(|error| invalid_data(format!("durable state JSON is invalid: {error}")))?;
749 decode_v8_state(state)
750 }
751 NODE_STATE_SCHEMA_V9 => {
752 let state: PersistedNodeStateV9 = serde_json::from_slice(bytes)
753 .map_err(|error| invalid_data(format!("durable state JSON is invalid: {error}")))?;
754 decode_v9_state(state)
755 }
756 NODE_STATE_SCHEMA_V10 => {
757 let state: PersistedNodeStateV10 = serde_json::from_slice(bytes)
758 .map_err(|error| invalid_data(format!("durable state JSON is invalid: {error}")))?;
759 decode_v10_state(state)
760 }
761 version => Err(io::Error::new(
762 io::ErrorKind::Unsupported,
763 StateLoadRefusalError::UnsupportedSchema(version),
764 )),
765 }
766}
767
768fn decode_v1_state(state: PersistedNodeStateV1) -> io::Result<DecodedNodeState> {
769 Ok(DecodedNodeState {
770 node_id: state.node_id,
771 workspaces: state
772 .workspaces
773 .into_iter()
774 .map(|workspace| (workspace.workspace_id, workspace.canonical_root))
775 .collect(),
776 session_records: state
777 .session_records
778 .into_iter()
779 .map(|record| {
780 Ok(ManagedSessionRecord {
781 record_id: record.record_id,
782 display_name: record.display_name,
783 provider: record.provider.into_agent_id(),
784 mode: record.mode,
785 state: record.state,
786 workspace_id: record.workspace_id,
787 canonical_root: OpaqueHostPath::utf8(record.canonical_root)
788 .map_err(|error| invalid_data(format!(
789 "durable state contains an invalid workspace path: {error}"
790 )))?,
791 provider_session: record.provider_session,
792 active_session: record.active_session,
793 environment_profile: None,
794 bundle: None,
795 context_id: None,
796 context: None,
797 exported_context: None,
798 task_binding: None,
799 created_at_unix_ms: record.created_at_unix_ms,
800 updated_at_unix_ms: record.updated_at_unix_ms,
801 last_error: record.last_error,
802 })
803 })
804 .collect::<io::Result<_>>()?,
805 managed_worktrees: Vec::new(),
806 managed_worktree_tombstones: Vec::new(),
807 materializations: Vec::new(),
808 managed_spawn_replays: Vec::new(),
809 })
810}
811
812fn decode_v2_state(state: PersistedNodeStateV2) -> io::Result<DecodedNodeState> {
813 Ok(DecodedNodeState {
814 node_id: state.node_id,
815 workspaces: state
816 .workspaces
817 .into_iter()
818 .map(|workspace| {
819 Ok((
820 workspace.workspace_id,
821 require_utf8_state_path(&workspace.canonical_root)?.to_owned(),
822 ))
823 })
824 .collect::<io::Result<_>>()?,
825 session_records: state
826 .session_records
827 .into_iter()
828 .map(|record| {
829 require_utf8_state_path(&record.canonical_root)?;
830 Ok(ManagedSessionRecord {
831 record_id: record.record_id,
832 display_name: record.display_name,
833 provider: record.provider.into_agent_id(),
834 mode: record.mode,
835 state: record.state,
836 workspace_id: record.workspace_id,
837 canonical_root: record.canonical_root,
838 provider_session: record.provider_session,
839 active_session: record.active_session,
840 environment_profile: None,
841 bundle: None,
842 context_id: None,
843 context: None,
844 exported_context: None,
845 task_binding: None,
846 created_at_unix_ms: record.created_at_unix_ms,
847 updated_at_unix_ms: record.updated_at_unix_ms,
848 last_error: record.last_error,
849 })
850 })
851 .collect::<io::Result<_>>()?,
852 managed_worktrees: Vec::new(),
853 managed_worktree_tombstones: Vec::new(),
854 materializations: Vec::new(),
855 managed_spawn_replays: Vec::new(),
856 })
857}
858
859fn decode_v3_state(state: PersistedNodeStateV3) -> io::Result<DecodedNodeState> {
860 Ok(DecodedNodeState {
861 node_id: state.node_id,
862 workspaces: state
863 .workspaces
864 .into_iter()
865 .map(|workspace| {
866 Ok((
867 workspace.workspace_id,
868 require_utf8_state_path(&workspace.canonical_root)?.to_owned(),
869 ))
870 })
871 .collect::<io::Result<_>>()?,
872 session_records: state
873 .session_records
874 .into_iter()
875 .map(|record| {
876 require_utf8_state_path(&record.canonical_root)?;
877 Ok(ManagedSessionRecord {
878 record_id: record.record_id,
879 display_name: record.display_name,
880 provider: record.provider,
881 mode: record.mode,
882 state: record.state,
883 workspace_id: record.workspace_id,
884 canonical_root: record.canonical_root,
885 provider_session: record.provider_session,
886 active_session: record.active_session,
887 environment_profile: None,
888 bundle: None,
889 context_id: None,
890 context: None,
891 exported_context: None,
892 task_binding: None,
893 created_at_unix_ms: record.created_at_unix_ms,
894 updated_at_unix_ms: record.updated_at_unix_ms,
895 last_error: record.last_error,
896 })
897 })
898 .collect::<io::Result<_>>()?,
899 managed_worktrees: Vec::new(),
900 managed_worktree_tombstones: Vec::new(),
901 materializations: Vec::new(),
902 managed_spawn_replays: Vec::new(),
903 })
904}
905
906fn decode_v4_state(state: PersistedNodeStateV4) -> io::Result<DecodedNodeState> {
907 let legacy = decode_v3_state(PersistedNodeStateV3 {
908 version: NODE_STATE_SCHEMA_V3,
909 node_id: state.node_id,
910 workspaces: state.workspaces,
911 session_records: state.session_records,
912 })?;
913 validate_managed_worktree_records(&state.managed_worktrees, false)?;
914 validate_managed_worktree_records(&state.managed_worktree_tombstones, true)?;
915 validate_managed_worktree_identity_sets(
916 &state.managed_worktrees,
917 &state.managed_worktree_tombstones,
918 )?;
919 Ok(DecodedNodeState {
920 node_id: legacy.node_id,
921 workspaces: legacy.workspaces,
922 session_records: legacy.session_records,
923 managed_worktrees: state.managed_worktrees,
924 managed_worktree_tombstones: state.managed_worktree_tombstones,
925 materializations: Vec::new(),
926 managed_spawn_replays: Vec::new(),
927 })
928}
929
930fn decode_v5_state(state: PersistedNodeStateV5) -> io::Result<DecodedNodeState> {
931 let mut environment_profiles = Vec::with_capacity(state.session_records.len());
932 let legacy_records = state
933 .session_records
934 .into_iter()
935 .map(|record| {
936 environment_profiles.push(record.environment_profile);
937 PersistedManagedSessionRecordV3 {
938 record_id: record.record_id,
939 display_name: record.display_name,
940 provider: record.provider,
941 mode: record.mode,
942 state: record.state,
943 workspace_id: record.workspace_id,
944 canonical_root: record.canonical_root,
945 provider_session: record.provider_session,
946 active_session: record.active_session,
947 created_at_unix_ms: record.created_at_unix_ms,
948 updated_at_unix_ms: record.updated_at_unix_ms,
949 last_error: record.last_error,
950 }
951 })
952 .collect();
953 let mut decoded = decode_v4_state(PersistedNodeStateV4 {
954 version: NODE_STATE_SCHEMA_V4,
955 node_id: state.node_id,
956 workspaces: state.workspaces,
957 session_records: legacy_records,
958 managed_worktrees: state.managed_worktrees,
959 managed_worktree_tombstones: state.managed_worktree_tombstones,
960 })?;
961 for (record, environment_profile) in decoded
962 .session_records
963 .iter_mut()
964 .zip(environment_profiles)
965 {
966 record.environment_profile = environment_profile;
967 }
968 Ok(decoded)
969}
970
971fn decode_v6_state(state: PersistedNodeStateV6) -> io::Result<DecodedNodeState> {
972 let mut decoded = decode_v5_state(PersistedNodeStateV5 {
973 version: NODE_STATE_SCHEMA_V5,
974 node_id: state.node_id,
975 workspaces: state.workspaces,
976 session_records: state.session_records,
977 managed_worktrees: state.managed_worktrees,
978 managed_worktree_tombstones: state.managed_worktree_tombstones,
979 })?;
980 let materializations = state.materializations.into_iter().map(|record| {
981 MaterializationOwnershipRecord::from_persisted(
982 record.materialization_id,
983 Some(record.environment_profile),
984 None,
985 None,
986 record.owner,
987 record.managed_lease_id,
988 record.state,
989 opaque_to_path(&record.root)?,
990 opaque_to_path(&record.provider_home)?,
991 opaque_to_path(&record.root)?.join("bundle"),
992 opaque_to_path(&record.root)?.join("plugin-data"),
993 record.declared_paths,
994 record.created_at_unix_ms,
995 record.updated_at_unix_ms,
996 ).map_err(|error| invalid_data(format!("durable state materialization record is invalid: {error}")))
997 }).collect::<io::Result<Vec<_>>>()?;
998 validate_materialization_records(
999 &materializations,
1000 &decoded.session_records,
1001 &decoded.managed_worktrees,
1002 &decoded.managed_worktree_tombstones,
1003 )?;
1004 decoded.materializations = materializations;
1005 Ok(decoded)
1006}
1007
1008fn decode_v7_state(state: PersistedNodeStateV7) -> io::Result<DecodedNodeState> {
1009 let session_records = state.session_records.into_iter().map(|mut record| {
1010 record.context_id = None;
1011 record.context = None;
1012 record.task_binding = None;
1013 record
1014 }).collect::<Vec<_>>();
1015 let materializations = state.materializations.into_iter().map(|record| {
1016 MaterializationOwnershipRecord::from_persisted(
1017 record.materialization_id,
1018 record.environment_profile,
1019 record.bundle,
1020 None,
1021 record.owner,
1022 record.managed_lease_id,
1023 record.state,
1024 opaque_to_path(&record.root)?,
1025 opaque_to_path(&record.provider_home)?,
1026 opaque_to_path(&record.bundle_root)?,
1027 opaque_to_path(&record.plugin_data)?,
1028 record.declared_paths,
1029 record.created_at_unix_ms,
1030 record.updated_at_unix_ms,
1031 ).map_err(|error| invalid_data(format!("durable state materialization record is invalid: {error}")))
1032 }).collect::<io::Result<Vec<_>>>()?;
1033 validate_managed_worktree_records(&state.managed_worktrees, false)?;
1034 validate_managed_worktree_records(&state.managed_worktree_tombstones, true)?;
1035 validate_managed_worktree_identity_sets(
1036 &state.managed_worktrees,
1037 &state.managed_worktree_tombstones,
1038 )?;
1039 validate_materialization_records(
1040 &materializations,
1041 &session_records,
1042 &state.managed_worktrees,
1043 &state.managed_worktree_tombstones,
1044 )?;
1045 Ok(DecodedNodeState {
1046 node_id: state.node_id,
1047 workspaces: state.workspaces.into_iter().map(|workspace| {
1048 Ok((workspace.workspace_id, require_utf8_state_path(&workspace.canonical_root)?.to_owned()))
1049 }).collect::<io::Result<_>>()?,
1050 session_records,
1051 managed_worktrees: state.managed_worktrees,
1052 managed_worktree_tombstones: state.managed_worktree_tombstones,
1053 materializations,
1054 managed_spawn_replays: Vec::new(),
1055 })
1056}
1057
1058fn decode_v8_state(mut state: PersistedNodeStateV8) -> io::Result<DecodedNodeState> {
1059 for record in &mut state.session_records {
1060 record.task_binding = None;
1061 }
1062 decode_v9_state(state)
1063}
1064
1065fn decode_v9_state(state: PersistedNodeStateV9) -> io::Result<DecodedNodeState> {
1066 let materializations = state.materializations.into_iter().map(|record| {
1067 MaterializationOwnershipRecord::from_persisted(
1068 record.materialization_id,
1069 record.environment_profile,
1070 record.bundle,
1071 record.context,
1072 record.owner,
1073 record.managed_lease_id,
1074 record.state,
1075 opaque_to_path(&record.root)?,
1076 opaque_to_path(&record.provider_home)?,
1077 opaque_to_path(&record.bundle_root)?,
1078 opaque_to_path(&record.plugin_data)?,
1079 record.declared_paths,
1080 record.created_at_unix_ms,
1081 record.updated_at_unix_ms,
1082 ).map_err(|error| invalid_data(format!("durable state materialization record is invalid: {error}")))
1083 }).collect::<io::Result<Vec<_>>>()?;
1084 validate_managed_worktree_records(&state.managed_worktrees, false)?;
1085 validate_managed_worktree_records(&state.managed_worktree_tombstones, true)?;
1086 validate_managed_worktree_identity_sets(
1087 &state.managed_worktrees,
1088 &state.managed_worktree_tombstones,
1089 )?;
1090 validate_materialization_records(
1091 &materializations,
1092 &state.session_records,
1093 &state.managed_worktrees,
1094 &state.managed_worktree_tombstones,
1095 )?;
1096 Ok(DecodedNodeState {
1097 node_id: state.node_id,
1098 workspaces: state.workspaces.into_iter().map(|workspace| {
1099 Ok((workspace.workspace_id, require_utf8_state_path(&workspace.canonical_root)?.to_owned()))
1100 }).collect::<io::Result<_>>()?,
1101 session_records: state.session_records,
1102 managed_worktrees: state.managed_worktrees,
1103 managed_worktree_tombstones: state.managed_worktree_tombstones,
1104 materializations,
1105 managed_spawn_replays: Vec::new(),
1106 })
1107}
1108
1109fn decode_v10_state(state: PersistedNodeStateV10) -> io::Result<DecodedNodeState> {
1110 validate_managed_spawn_replays(
1111 &state.managed_spawn_replays,
1112 &state.managed_worktrees,
1113 &state.managed_worktree_tombstones,
1114 )?;
1115 let mut decoded = decode_v9_state(PersistedNodeStateV9 {
1116 version: NODE_STATE_SCHEMA_V9,
1117 node_id: state.node_id,
1118 workspaces: state.workspaces,
1119 session_records: state.session_records,
1120 managed_worktrees: state.managed_worktrees,
1121 managed_worktree_tombstones: state.managed_worktree_tombstones,
1122 materializations: state.materializations,
1123 })?;
1124 decoded.managed_spawn_replays = state.managed_spawn_replays;
1125 Ok(decoded)
1126}
1127
1128fn validate_managed_spawn_replays(
1129 records: &[ManagedWorktreeSpawnReplayRecordV10],
1130 managed_worktrees: &[ManagedWorktreeLeaseRecord],
1131 managed_worktree_tombstones: &[ManagedWorktreeLeaseRecord],
1132) -> io::Result<()> {
1133 if records.len() > MAX_MANAGED_WORKTREE_SPAWN_REPLAYS {
1134 return Err(invalid_data("durable state contains too many managed spawn replays"));
1135 }
1136 let mut keys = BTreeSet::new();
1137 for record in records {
1138 let digest = record.request_digest.strip_prefix("sha256:").ok_or_else(|| {
1139 invalid_data("durable managed spawn replay request digest is invalid")
1140 })?;
1141 if digest.len() != 64
1142 || !digest
1143 .bytes()
1144 .all(|byte| byte.is_ascii_digit() || matches!(byte, b'a'..=b'f'))
1145 {
1146 return Err(invalid_data(
1147 "durable managed spawn replay request digest is invalid",
1148 ));
1149 }
1150 if record.updated_at_unix_ms < record.created_at_unix_ms {
1151 return Err(invalid_data(
1152 "durable managed spawn replay update precedes creation",
1153 ));
1154 }
1155 let (lease_id, receipt) = match &record.state {
1156 ManagedWorktreeSpawnReplayStateV10::Pending { lease_id, .. } => {
1157 (lease_id.as_ref(), None)
1158 }
1159 ManagedWorktreeSpawnReplayStateV10::Committed { receipt } => {
1160 if receipt.spawn.idempotency_key != record.idempotency_key {
1161 return Err(invalid_data(
1162 "durable managed spawn replay receipt is not linked to its idempotency key",
1163 ));
1164 }
1165 if receipt.lease.source_workspace_id != record.source_workspace_id
1166 || receipt.lease.profile_id != record.profile_id
1167 || receipt.lease.profile_revision != record.expected_profile_revision
1168 {
1169 return Err(invalid_data(
1170 "durable managed spawn replay receipt is not linked to its reservation",
1171 ));
1172 }
1173 (Some(&receipt.lease.lease_id), Some(receipt))
1174 }
1175 };
1176 if let Some(lease_id) = lease_id {
1177 let durable_lease = managed_worktrees
1178 .iter()
1179 .chain(managed_worktree_tombstones)
1180 .find(|lease| &lease.lease_id == lease_id);
1181 match durable_lease {
1182 Some(durable_lease)
1183 if durable_lease.source_workspace_id == record.source_workspace_id
1184 && durable_lease.profile_id == record.profile_id
1185 && durable_lease.profile_revision == record.expected_profile_revision
1186 && receipt.map_or(true, |receipt| {
1187 durable_lease.workspace_id == receipt.lease.workspace_id
1188 }) => {}
1189 None if receipt.is_none() => {}
1190 _ => {
1191 return Err(invalid_data(
1192 "durable managed spawn replay lease identity is inconsistent",
1193 ));
1194 }
1195 }
1196 }
1197 if !keys.insert(record.idempotency_key.clone()) {
1198 return Err(invalid_data(
1199 "durable state contains duplicate managed spawn replay keys",
1200 ));
1201 }
1202 }
1203 Ok(())
1204}
1205
1206fn validate_materialization_records(
1207 records: &[MaterializationOwnershipRecord],
1208 session_records: &[ManagedSessionRecord],
1209 managed_worktrees: &[ManagedWorktreeLeaseRecord],
1210 managed_worktree_tombstones: &[ManagedWorktreeLeaseRecord],
1211) -> io::Result<()> {
1212 if records.len() > MAX_SESSION_MATERIALIZATIONS {
1213 return Err(invalid_data("durable state contains too many materializations"));
1214 }
1215 let known_records = session_records
1216 .iter()
1217 .map(|record| (&record.record_id, record))
1218 .collect::<BTreeMap<_, _>>();
1219 let active_leases = managed_worktrees
1220 .iter()
1221 .map(|lease| (&lease.lease_id, lease))
1222 .collect::<BTreeMap<_, _>>();
1223 let tombstone_ids = managed_worktree_tombstones
1224 .iter()
1225 .map(|lease| &lease.lease_id)
1226 .collect::<BTreeSet<_>>();
1227 let mut ids = BTreeSet::new();
1228 let mut roots = BTreeSet::new();
1229 let mut owners = BTreeSet::new();
1230 for record in records {
1231 record.validate().map_err(|error| invalid_data(format!("durable state materialization record is invalid: {error}")))?;
1232 if record.context().is_some_and(|context| !context.is_valid()) {
1233 return Err(invalid_data(
1234 "durable state materialization context receipt is invalid",
1235 ));
1236 }
1237 let (owner_key, record_workspace) = match record.owner() {
1238 MaterializationOwner::Session { incarnation_id, instance_id, generation } => {
1239 (
1240 format!("session:{incarnation_id}:{}:{}", instance_id.0, generation.0),
1241 None,
1242 )
1243 }
1244 MaterializationOwner::Record { record_id } => {
1245 let managed_record = known_records.get(record_id).ok_or_else(|| {
1246 invalid_data(
1247 "durable state materialization references an unknown session record",
1248 )
1249 })?;
1250 if record.environment_profile() != managed_record.environment_profile.as_ref()
1251 || record.bundle() != managed_record.bundle.as_ref()
1252 || record.context() != managed_record.context.as_ref()
1253 {
1254 return Err(invalid_data(
1255 "durable state materialization receipt disagrees with its session record",
1256 ));
1257 }
1258 (
1259 format!("record:{}", record_id.as_str()),
1260 Some(&managed_record.workspace_id),
1261 )
1262 }
1263 };
1264 if let Some(lease_id) = record.managed_lease_id() {
1265 if tombstone_ids.contains(lease_id) {
1266 return Err(invalid_data(
1267 "durable state materialization references a removed managed worktree lease",
1268 ));
1269 }
1270 let lease = active_leases.get(lease_id).ok_or_else(|| {
1271 invalid_data(
1272 "durable state materialization references an unknown managed worktree lease",
1273 )
1274 })?;
1275 if let Some(record_workspace) = record_workspace {
1276 if record_workspace != &lease.workspace_id {
1277 return Err(invalid_data(
1278 "durable state materialization record and managed worktree lease disagree on workspace",
1279 ));
1280 }
1281 }
1282 }
1283 if !ids.insert(record.id().clone())
1284 || !roots.insert(record.root().to_path_buf())
1285 || !owners.insert(owner_key)
1286 {
1287 return Err(invalid_data("durable state contains duplicate materialization ownership"));
1288 }
1289 }
1290 for record in session_records.iter().filter(|record| {
1291 record.bundle.is_some() || record.context.is_some()
1292 }) {
1293 if !records.iter().any(|materialization| {
1294 materialization.owner()
1295 == &MaterializationOwner::Record {
1296 record_id: record.record_id.clone(),
1297 }
1298 && materialization.bundle() == record.bundle.as_ref()
1299 && materialization.environment_profile() == record.environment_profile.as_ref()
1300 && materialization.context() == record.context.as_ref()
1301 }) {
1302 return Err(invalid_data(
1303 "durable bound session record lacks exact materialization ownership",
1304 ));
1305 }
1306 }
1307 Ok(())
1308}
1309
1310fn validate_managed_worktree_identity_sets(
1311 managed_worktrees: &[ManagedWorktreeLeaseRecord],
1312 managed_worktree_tombstones: &[ManagedWorktreeLeaseRecord],
1313) -> io::Result<()> {
1314 let active_ids = managed_worktrees
1315 .iter()
1316 .map(|lease| &lease.lease_id)
1317 .collect::<BTreeSet<_>>();
1318 if managed_worktree_tombstones
1319 .iter()
1320 .any(|lease| active_ids.contains(&lease.lease_id))
1321 {
1322 return Err(invalid_data(
1323 "durable state contains duplicate managed worktree leases",
1324 ));
1325 }
1326 Ok(())
1327}
1328
1329fn validate_managed_worktree_records(
1330 records: &[ManagedWorktreeLeaseRecord],
1331 tombstones: bool,
1332) -> io::Result<()> {
1333 if records.len() > crate::protocol::MAX_MANAGED_WORKTREE_LEASES {
1334 return Err(invalid_data("durable state contains too many managed worktree leases"));
1335 }
1336 let mut lease_ids = BTreeSet::new();
1337 let mut workspace_ids = BTreeSet::new();
1338 let mut roots = BTreeSet::new();
1339 let mut branches = BTreeSet::new();
1340 for lease in records {
1341 validate_root(&lease.target_root)?;
1342 let mut session_holders = BTreeSet::new();
1343 let mut record_holders = BTreeSet::new();
1344 if lease.branch.is_empty()
1345 || lease.branch.len() > MAX_PERSISTED_MANAGED_WORKTREE_BRANCH_BYTES
1346 || !valid_git_object_id(&lease.base_commit)
1347 || lease.expected_head.as_ref().is_some_and(|head| !valid_git_object_id(head))
1348 || lease.source_workspace_id == lease.workspace_id
1349 || lease.branch.chars().any(char::is_control)
1350 || lease.base_commit.chars().any(char::is_control)
1351 || lease.created_at_unix_ms > lease.updated_at_unix_ms
1352 || tombstones != (lease.state == crate::protocol::ManagedWorktreeLeaseState::Removed)
1353 || !lease_ids.insert(lease.lease_id.clone())
1354 || !workspace_ids.insert(lease.workspace_id.clone())
1355 || !roots.insert(crate::platform::root_identity(&lease.target_root))
1356 || !branches.insert(lease.branch.clone())
1357 || lease.session_holders.len() > gate4agent_types::CONTROL_SESSIONS_MAX
1358 || lease.record_holders.len() > MAX_MANAGED_SESSION_RECORDS
1359 || lease.session_holders.iter().any(|holder| !session_holders.insert((
1360 holder.incarnation_id,
1361 holder.instance_id,
1362 holder.generation,
1363 )))
1364 || lease.record_holders.iter().any(|holder| !record_holders.insert(holder.clone()))
1365 || (lease.state == crate::protocol::ManagedWorktreeLeaseState::Removed
1366 && (!lease.session_holders.is_empty() || !lease.record_holders.is_empty()))
1367 || (matches!(
1368 lease.state,
1369 crate::protocol::ManagedWorktreeLeaseState::Allocating
1370 | crate::protocol::ManagedWorktreeLeaseState::Ready
1371 | crate::protocol::ManagedWorktreeLeaseState::Retained
1372 | crate::protocol::ManagedWorktreeLeaseState::CleanupBlocked
1373 | crate::protocol::ManagedWorktreeLeaseState::Removed
1374 ) && (!lease.session_holders.is_empty() || !lease.record_holders.is_empty()))
1375 || (lease.state == crate::protocol::ManagedWorktreeLeaseState::InUse
1376 && lease.session_holders.is_empty() && lease.record_holders.is_empty())
1377 || (matches!(
1378 lease.state,
1379 crate::protocol::ManagedWorktreeLeaseState::CleanupBlocked
1380 | crate::protocol::ManagedWorktreeLeaseState::RecoveryRequired
1381 ) != lease.cleanup_failure.is_some())
1382 {
1383 return Err(invalid_data("durable managed worktree record is invalid"));
1384 }
1385 }
1386 Ok(())
1387}
1388
1389fn valid_git_object_id(value: &str) -> bool {
1390 matches!(value.len(), 40 | 64) && value.bytes().all(|byte| byte.is_ascii_hexdigit())
1391}
1392
1393fn require_utf8_state_path(path: &OpaqueHostPath) -> io::Result<&str> {
1394 path.as_utf8().ok_or_else(|| io::Error::new(
1395 io::ErrorKind::Unsupported,
1396 StateLoadRefusalError::PathSemanticsUnsupported,
1397 ))
1398}
1399
1400fn is_authoritative_state_error(error: &io::Error) -> bool {
1401 state_load_refusal(error).is_some()
1402}
1403
1404pub fn state_load_refusal(error: &io::Error) -> Option<StateLoadRefusal> {
1405 error
1406 .get_ref()
1407 .and_then(|source| source.downcast_ref::<StateLoadRefusalError>())
1408 .map(StateLoadRefusalError::refusal)
1409}
1410
1411#[cfg(test)]
1412pub fn save(
1413 path: Option<&Path>,
1414 node_id: &NodeId,
1415 workspaces: &BTreeMap<WorkspaceId, String>,
1416 records: &[ManagedSessionRecord],
1417) -> io::Result<Option<String>> {
1418 let Some(path) = path else {
1419 return Ok(None);
1420 };
1421 if workspaces.len() > MAX_PERSISTED_WORKSPACES {
1422 return Err(invalid_data("refusing to persist too many workspaces"));
1423 }
1424 if records.len() > MAX_MANAGED_SESSION_RECORDS {
1425 return Err(invalid_data("refusing to persist too many session records"));
1426 }
1427 let mut persisted_records = Vec::with_capacity(records.len());
1428 let mut provider_sessions = BTreeSet::new();
1429 for record in records {
1430 let mut persisted = record.clone();
1431 persisted.active_session = None;
1432 if let Some(identity) = &mut persisted.provider_session {
1433 if !provider_sessions.insert(provider_session_semantic_key(
1434 &persisted.provider,
1435 identity,
1436 )) {
1437 return Err(invalid_data(
1438 "refusing to persist duplicate provider sessions",
1439 ));
1440 }
1441 identity.transcript_path = None;
1442 }
1443 persisted.last_error = persisted
1444 .last_error
1445 .as_deref()
1446 .map(sanitized_record_error_summary);
1447 validate_record(&persisted)?;
1448 persisted.state = if persisted.provider_session.is_some() {
1449 ManagedSessionState::Dormant
1450 } else {
1451 ManagedSessionState::Unavailable
1452 };
1453 persisted_records.push(persisted);
1454 }
1455 let state = PersistedNodeStateV3 {
1456 version: NODE_STATE_SCHEMA_V3,
1457 node_id: node_id.clone(),
1458 workspaces: workspaces
1459 .iter()
1460 .map(|(workspace_id, canonical_root)| {
1461 Ok(PersistedWorkspaceV2 {
1462 workspace_id: workspace_id.clone(),
1463 canonical_root: OpaqueHostPath::utf8(canonical_root.clone()).map_err(|error| {
1464 invalid_data(format!("refusing to persist an invalid workspace path: {error}"))
1465 })?,
1466 })
1467 })
1468 .collect::<io::Result<_>>()?,
1469 session_records: persisted_records
1470 .into_iter()
1471 .map(persisted_v3_record)
1472 .collect(),
1473 };
1474 let bytes = serde_json::to_vec_pretty(&state)
1475 .map_err(|error| invalid_data(format!("durable state encoding failed: {error}")))?;
1476 if bytes.len() as u64 > MAX_STATE_BYTES {
1477 return Err(invalid_data("refusing to persist oversized durable state"));
1478 }
1479 let previous_primary_valid = if !path.exists() {
1480 true
1481 } else {
1482 match load_one(path, node_id) {
1483 Ok(_) => true,
1484 Err(error) if is_authoritative_state_error(&error) => return Err(error),
1485 Err(_) => false,
1486 }
1487 };
1488 atomic_write(path, &bytes, previous_primary_valid)
1489}
1490
1491#[cfg(test)]
1492pub fn save_v4(
1493 path: Option<&Path>,
1494 node_id: &NodeId,
1495 workspaces: &BTreeMap<WorkspaceId, String>,
1496 records: &[ManagedSessionRecord],
1497 managed_worktrees: &[ManagedWorktreeLeaseRecord],
1498 managed_worktree_tombstones: &[ManagedWorktreeLeaseRecord],
1499) -> io::Result<Option<String>> {
1500 let Some(path) = path else {
1501 return Ok(None);
1502 };
1503 if workspaces.len() > MAX_PERSISTED_WORKSPACES {
1504 return Err(invalid_data("refusing to persist too many workspaces"));
1505 }
1506 if records.len() > MAX_MANAGED_SESSION_RECORDS {
1507 return Err(invalid_data("refusing to persist too many session records"));
1508 }
1509 validate_managed_worktree_records(managed_worktrees, false)?;
1510 validate_managed_worktree_records(managed_worktree_tombstones, true)?;
1511 let mut persisted_records = Vec::with_capacity(records.len());
1512 let mut provider_sessions = BTreeSet::new();
1513 for record in records {
1514 let mut persisted = record.clone();
1515 persisted.active_session = None;
1516 if let Some(identity) = &mut persisted.provider_session {
1517 if !provider_sessions.insert(provider_session_semantic_key(
1518 &persisted.provider,
1519 identity,
1520 )) {
1521 return Err(invalid_data("refusing to persist duplicate provider sessions"));
1522 }
1523 identity.transcript_path = None;
1524 }
1525 persisted.last_error = persisted.last_error.as_deref().map(sanitized_record_error_summary);
1526 validate_record(&persisted)?;
1527 persisted.state = if persisted.provider_session.is_some() {
1528 ManagedSessionState::Dormant
1529 } else {
1530 ManagedSessionState::Unavailable
1531 };
1532 persisted_records.push(persisted_v3_record(persisted));
1533 }
1534 let state = PersistedNodeStateV4 {
1535 version: NODE_STATE_SCHEMA_V4,
1536 node_id: node_id.clone(),
1537 workspaces: workspaces.iter().map(|(workspace_id, canonical_root)| {
1538 Ok(PersistedWorkspaceV2 {
1539 workspace_id: workspace_id.clone(),
1540 canonical_root: OpaqueHostPath::utf8(canonical_root.clone()).map_err(|error| {
1541 invalid_data(format!("refusing to persist an invalid workspace path: {error}"))
1542 })?,
1543 })
1544 }).collect::<io::Result<_>>()?,
1545 session_records: persisted_records,
1546 managed_worktrees: managed_worktrees.to_vec(),
1547 managed_worktree_tombstones: managed_worktree_tombstones.to_vec(),
1548 };
1549 let bytes = serde_json::to_vec_pretty(&state)
1550 .map_err(|error| invalid_data(format!("durable state encoding failed: {error}")))?;
1551 if bytes.len() as u64 > MAX_STATE_BYTES {
1552 return Err(invalid_data("refusing to persist oversized durable state"));
1553 }
1554 let previous_primary_valid = if !path.exists() {
1555 true
1556 } else {
1557 match load_one(path, node_id) {
1558 Ok(_) => true,
1559 Err(error) if is_authoritative_state_error(&error) => return Err(error),
1560 Err(_) => false,
1561 }
1562 };
1563 atomic_write(path, &bytes, previous_primary_valid)
1564}
1565
1566#[cfg(test)]
1567pub fn save_v5(
1568 path: Option<&Path>,
1569 node_id: &NodeId,
1570 workspaces: &BTreeMap<WorkspaceId, String>,
1571 records: &[ManagedSessionRecord],
1572 managed_worktrees: &[ManagedWorktreeLeaseRecord],
1573 managed_worktree_tombstones: &[ManagedWorktreeLeaseRecord],
1574) -> io::Result<Option<String>> {
1575 let Some(path) = path else {
1576 return Ok(None);
1577 };
1578 if workspaces.len() > MAX_PERSISTED_WORKSPACES {
1579 return Err(invalid_data("refusing to persist too many workspaces"));
1580 }
1581 if records.len() > MAX_MANAGED_SESSION_RECORDS {
1582 return Err(invalid_data("refusing to persist too many session records"));
1583 }
1584 validate_managed_worktree_records(managed_worktrees, false)?;
1585 validate_managed_worktree_records(managed_worktree_tombstones, true)?;
1586 let mut persisted_records = Vec::with_capacity(records.len());
1587 let mut provider_sessions = BTreeSet::new();
1588 for record in records {
1589 let mut persisted = record.clone();
1590 persisted.active_session = None;
1591 if let Some(identity) = &mut persisted.provider_session {
1592 if !provider_sessions.insert(provider_session_semantic_key(
1593 &persisted.provider,
1594 identity,
1595 )) {
1596 return Err(invalid_data("refusing to persist duplicate provider sessions"));
1597 }
1598 identity.transcript_path = None;
1599 }
1600 persisted.last_error = persisted.last_error.as_deref().map(sanitized_record_error_summary);
1601 validate_record(&persisted)?;
1602 persisted.state = if persisted.provider_session.is_some() {
1603 ManagedSessionState::Dormant
1604 } else {
1605 ManagedSessionState::Unavailable
1606 };
1607 persisted_records.push(persisted_v5_record(persisted));
1608 }
1609 let state = PersistedNodeStateV5 {
1610 version: NODE_STATE_SCHEMA_V5,
1611 node_id: node_id.clone(),
1612 workspaces: workspaces.iter().map(|(workspace_id, canonical_root)| {
1613 Ok(PersistedWorkspaceV2 {
1614 workspace_id: workspace_id.clone(),
1615 canonical_root: OpaqueHostPath::utf8(canonical_root.clone()).map_err(|error| {
1616 invalid_data(format!("refusing to persist an invalid workspace path: {error}"))
1617 })?,
1618 })
1619 }).collect::<io::Result<_>>()?,
1620 session_records: persisted_records,
1621 managed_worktrees: managed_worktrees.to_vec(),
1622 managed_worktree_tombstones: managed_worktree_tombstones.to_vec(),
1623 };
1624 let bytes = serde_json::to_vec_pretty(&state)
1625 .map_err(|error| invalid_data(format!("durable state encoding failed: {error}")))?;
1626 if bytes.len() as u64 > MAX_STATE_BYTES {
1627 return Err(invalid_data("refusing to persist oversized durable state"));
1628 }
1629 let previous_primary_valid = if !path.exists() {
1630 true
1631 } else {
1632 match load_one(path, node_id) {
1633 Ok(_) => true,
1634 Err(error) if is_authoritative_state_error(&error) => return Err(error),
1635 Err(_) => false,
1636 }
1637 };
1638 atomic_write(path, &bytes, previous_primary_valid)
1639}
1640
1641#[cfg(test)]
1642pub fn save_v7(
1643 path: Option<&Path>,
1644 node_id: &NodeId,
1645 workspaces: &BTreeMap<WorkspaceId, String>,
1646 records: &[ManagedSessionRecord],
1647 managed_worktrees: &[ManagedWorktreeLeaseRecord],
1648 managed_worktree_tombstones: &[ManagedWorktreeLeaseRecord],
1649 materializations: &[MaterializationOwnershipRecord],
1650) -> io::Result<Option<String>> {
1651 let Some(path) = path else {
1652 return Ok(None);
1653 };
1654 if records.iter().any(|record| record.context_id.is_some() || record.context.is_some())
1655 || materializations.iter().any(|record| record.context().is_some())
1656 {
1657 return Err(invalid_data(
1658 "refusing to persist context metadata in legacy durable state v7",
1659 ));
1660 }
1661 if workspaces.len() > MAX_PERSISTED_WORKSPACES {
1662 return Err(invalid_data("refusing to persist too many workspaces"));
1663 }
1664 if records.len() > MAX_MANAGED_SESSION_RECORDS {
1665 return Err(invalid_data("refusing to persist too many session records"));
1666 }
1667 validate_managed_worktree_records(managed_worktrees, false)?;
1668 validate_managed_worktree_records(managed_worktree_tombstones, true)?;
1669 validate_managed_worktree_identity_sets(
1670 managed_worktrees,
1671 managed_worktree_tombstones,
1672 )?;
1673 validate_materialization_records(
1674 materializations,
1675 records,
1676 managed_worktrees,
1677 managed_worktree_tombstones,
1678 )?;
1679 let mut persisted_records = Vec::with_capacity(records.len());
1680 let mut provider_sessions = BTreeSet::new();
1681 for record in records {
1682 let mut persisted = record.clone();
1683 persisted.active_session = None;
1684 if let Some(identity) = &mut persisted.provider_session {
1685 if !provider_sessions.insert(provider_session_semantic_key(&persisted.provider, identity)) {
1686 return Err(invalid_data("refusing to persist duplicate provider sessions"));
1687 }
1688 identity.transcript_path = None;
1689 }
1690 persisted.last_error = persisted.last_error.as_deref().map(sanitized_record_error_summary);
1691 validate_record(&persisted)?;
1692 persisted.state = if persisted.provider_session.is_some() {
1693 ManagedSessionState::Dormant
1694 } else {
1695 ManagedSessionState::Unavailable
1696 };
1697 persisted_records.push(persisted);
1698 }
1699 let state = PersistedNodeStateV7 {
1700 version: NODE_STATE_SCHEMA_V7,
1701 node_id: node_id.clone(),
1702 workspaces: workspaces.iter().map(|(workspace_id, canonical_root)| {
1703 Ok(PersistedWorkspaceV2 {
1704 workspace_id: workspace_id.clone(),
1705 canonical_root: OpaqueHostPath::utf8(canonical_root.clone()).map_err(|error| {
1706 invalid_data(format!("refusing to persist an invalid workspace path: {error}"))
1707 })?,
1708 })
1709 }).collect::<io::Result<_>>()?,
1710 session_records: persisted_records,
1711 managed_worktrees: managed_worktrees.to_vec(),
1712 managed_worktree_tombstones: managed_worktree_tombstones.to_vec(),
1713 materializations: materializations.iter().map(|record| {
1714 Ok(PersistedMaterializationRecordV7 {
1715 materialization_id: record.id().clone(),
1716 environment_profile: record.environment_profile().cloned(),
1717 bundle: record.bundle().cloned(),
1718 owner: record.owner().clone(),
1719 managed_lease_id: record.managed_lease_id().cloned(),
1720 state: record.state(),
1721 root: path_to_opaque(record.root())?,
1722 provider_home: path_to_opaque(record.provider_home())?,
1723 bundle_root: path_to_opaque(record.bundle_root())?,
1724 plugin_data: path_to_opaque(record.plugin_data())?,
1725 declared_paths: record.declared_paths().to_vec(),
1726 created_at_unix_ms: record.created_at_unix_ms(),
1727 updated_at_unix_ms: record.updated_at_unix_ms(),
1728 })
1729 }).collect::<io::Result<Vec<_>>>()?,
1730 };
1731 let bytes = serde_json::to_vec_pretty(&state)
1732 .map_err(|error| invalid_data(format!("durable state encoding failed: {error}")))?;
1733 if bytes.len() as u64 > MAX_STATE_BYTES {
1734 return Err(invalid_data("refusing to persist oversized durable state"));
1735 }
1736 let previous_primary_valid = if !path.exists() {
1737 true
1738 } else {
1739 match load_one(path, node_id) {
1740 Ok(_) => true,
1741 Err(error) if is_authoritative_state_error(&error) => return Err(error),
1742 Err(_) => false,
1743 }
1744 };
1745 atomic_write(path, &bytes, previous_primary_valid)
1746}
1747
1748#[cfg(test)]
1749pub fn save_v8(
1750 path: Option<&Path>,
1751 node_id: &NodeId,
1752 workspaces: &BTreeMap<WorkspaceId, String>,
1753 records: &[ManagedSessionRecord],
1754 managed_worktrees: &[ManagedWorktreeLeaseRecord],
1755 managed_worktree_tombstones: &[ManagedWorktreeLeaseRecord],
1756 materializations: &[MaterializationOwnershipRecord],
1757) -> io::Result<Option<String>> {
1758 if records.iter().any(|record| record.task_binding.is_some()) {
1759 return Err(invalid_data(
1760 "refusing to persist task bindings in legacy durable state v8",
1761 ));
1762 }
1763 save_current(
1764 path,
1765 node_id,
1766 workspaces,
1767 records,
1768 managed_worktrees,
1769 managed_worktree_tombstones,
1770 materializations,
1771 &[],
1772 NODE_STATE_SCHEMA_V8,
1773 )
1774}
1775
1776#[cfg(test)]
1777pub fn save_v9(
1778 path: Option<&Path>,
1779 node_id: &NodeId,
1780 workspaces: &BTreeMap<WorkspaceId, String>,
1781 records: &[ManagedSessionRecord],
1782 managed_worktrees: &[ManagedWorktreeLeaseRecord],
1783 managed_worktree_tombstones: &[ManagedWorktreeLeaseRecord],
1784 materializations: &[MaterializationOwnershipRecord],
1785) -> io::Result<Option<String>> {
1786 save_current(
1787 path,
1788 node_id,
1789 workspaces,
1790 records,
1791 managed_worktrees,
1792 managed_worktree_tombstones,
1793 materializations,
1794 &[],
1795 NODE_STATE_SCHEMA_V9,
1796 )
1797}
1798
1799pub fn save_v10(
1800 path: Option<&Path>,
1801 node_id: &NodeId,
1802 workspaces: &BTreeMap<WorkspaceId, String>,
1803 records: &[ManagedSessionRecord],
1804 managed_worktrees: &[ManagedWorktreeLeaseRecord],
1805 managed_worktree_tombstones: &[ManagedWorktreeLeaseRecord],
1806 materializations: &[MaterializationOwnershipRecord],
1807 managed_spawn_replays: &[ManagedWorktreeSpawnReplayRecordV10],
1808) -> io::Result<Option<String>> {
1809 save_current(
1810 path,
1811 node_id,
1812 workspaces,
1813 records,
1814 managed_worktrees,
1815 managed_worktree_tombstones,
1816 materializations,
1817 managed_spawn_replays,
1818 NODE_STATE_SCHEMA_V10,
1819 )
1820}
1821
1822fn save_current(
1823 path: Option<&Path>,
1824 node_id: &NodeId,
1825 workspaces: &BTreeMap<WorkspaceId, String>,
1826 records: &[ManagedSessionRecord],
1827 managed_worktrees: &[ManagedWorktreeLeaseRecord],
1828 managed_worktree_tombstones: &[ManagedWorktreeLeaseRecord],
1829 materializations: &[MaterializationOwnershipRecord],
1830 managed_spawn_replays: &[ManagedWorktreeSpawnReplayRecordV10],
1831 version: u16,
1832) -> io::Result<Option<String>> {
1833 let Some(path) = path else {
1834 return Ok(None);
1835 };
1836 if workspaces.len() > MAX_PERSISTED_WORKSPACES {
1837 return Err(invalid_data("refusing to persist too many workspaces"));
1838 }
1839 if records.len() > MAX_MANAGED_SESSION_RECORDS {
1840 return Err(invalid_data("refusing to persist too many session records"));
1841 }
1842 validate_managed_worktree_records(managed_worktrees, false)?;
1843 validate_managed_worktree_records(managed_worktree_tombstones, true)?;
1844 validate_managed_worktree_identity_sets(
1845 managed_worktrees,
1846 managed_worktree_tombstones,
1847 )?;
1848 validate_materialization_records(
1849 materializations,
1850 records,
1851 managed_worktrees,
1852 managed_worktree_tombstones,
1853 )?;
1854 if version == NODE_STATE_SCHEMA_V10 {
1855 validate_managed_spawn_replays(
1856 managed_spawn_replays,
1857 managed_worktrees,
1858 managed_worktree_tombstones,
1859 )?;
1860 } else if !managed_spawn_replays.is_empty() {
1861 return Err(invalid_data(
1862 "refusing to persist managed spawn replays in a legacy durable state",
1863 ));
1864 }
1865 let mut persisted_records = Vec::with_capacity(records.len());
1866 let mut provider_sessions = BTreeSet::new();
1867 for record in records {
1868 let mut persisted = record.clone();
1869 persisted.active_session = None;
1870 if let Some(identity) = &mut persisted.provider_session {
1871 if !provider_sessions.insert(provider_session_semantic_key(&persisted.provider, identity)) {
1872 return Err(invalid_data("refusing to persist duplicate provider sessions"));
1873 }
1874 identity.transcript_path = None;
1875 }
1876 persisted.last_error = persisted.last_error.as_deref().map(sanitized_record_error_summary);
1877 validate_record(&persisted)?;
1878 persisted.state = if persisted.provider_session.is_some() {
1879 ManagedSessionState::Dormant
1880 } else {
1881 ManagedSessionState::Unavailable
1882 };
1883 persisted_records.push(persisted);
1884 }
1885 let persisted_workspaces = workspaces.iter().map(|(workspace_id, canonical_root)| {
1886 Ok(PersistedWorkspaceV2 {
1887 workspace_id: workspace_id.clone(),
1888 canonical_root: OpaqueHostPath::utf8(canonical_root.clone()).map_err(|error| {
1889 invalid_data(format!("refusing to persist an invalid workspace path: {error}"))
1890 })?,
1891 })
1892 }).collect::<io::Result<Vec<_>>>()?;
1893 let persisted_materializations = materializations.iter().map(|record| {
1894 Ok(PersistedMaterializationRecordV8 {
1895 materialization_id: record.id().clone(),
1896 environment_profile: record.environment_profile().cloned(),
1897 bundle: record.bundle().cloned(),
1898 context: record.context().cloned(),
1899 owner: record.owner().clone(),
1900 managed_lease_id: record.managed_lease_id().cloned(),
1901 state: record.state(),
1902 root: path_to_opaque(record.root())?,
1903 provider_home: path_to_opaque(record.provider_home())?,
1904 bundle_root: path_to_opaque(record.bundle_root())?,
1905 plugin_data: path_to_opaque(record.plugin_data())?,
1906 declared_paths: record.declared_paths().to_vec(),
1907 created_at_unix_ms: record.created_at_unix_ms(),
1908 updated_at_unix_ms: record.updated_at_unix_ms(),
1909 })
1910 }).collect::<io::Result<Vec<_>>>()?;
1911 let bytes = if version == NODE_STATE_SCHEMA_V10 {
1912 serde_json::to_vec_pretty(&PersistedNodeStateV10 {
1913 version,
1914 node_id: node_id.clone(),
1915 workspaces: persisted_workspaces,
1916 session_records: persisted_records,
1917 managed_worktrees: managed_worktrees.to_vec(),
1918 managed_worktree_tombstones: managed_worktree_tombstones.to_vec(),
1919 materializations: persisted_materializations,
1920 managed_spawn_replays: managed_spawn_replays.to_vec(),
1921 })
1922 } else {
1923 serde_json::to_vec_pretty(&PersistedNodeStateV9 {
1924 version,
1925 node_id: node_id.clone(),
1926 workspaces: persisted_workspaces,
1927 session_records: persisted_records,
1928 managed_worktrees: managed_worktrees.to_vec(),
1929 managed_worktree_tombstones: managed_worktree_tombstones.to_vec(),
1930 materializations: persisted_materializations,
1931 })
1932 };
1933 let bytes = bytes.map_err(|error| {
1934 invalid_data(format!("durable state encoding failed: {error}"))
1935 })?;
1936 if bytes.len() as u64 > MAX_STATE_BYTES {
1937 return Err(invalid_data("refusing to persist oversized durable state"));
1938 }
1939 let previous_primary_valid = if !path.exists() {
1940 true
1941 } else {
1942 match load_one(path, node_id) {
1943 Ok(_) => true,
1944 Err(error) if is_authoritative_state_error(&error) => return Err(error),
1945 Err(_) => false,
1946 }
1947 };
1948 atomic_write(path, &bytes, previous_primary_valid)
1949}
1950
1951#[cfg(test)]
1952fn persisted_v3_record(record: ManagedSessionRecord) -> PersistedManagedSessionRecordV3 {
1953 PersistedManagedSessionRecordV3 {
1954 record_id: record.record_id,
1955 display_name: record.display_name,
1956 provider: record.provider,
1957 mode: record.mode,
1958 state: record.state,
1959 workspace_id: record.workspace_id,
1960 canonical_root: record.canonical_root,
1961 provider_session: record.provider_session,
1962 active_session: record.active_session,
1963 created_at_unix_ms: record.created_at_unix_ms,
1964 updated_at_unix_ms: record.updated_at_unix_ms,
1965 last_error: record.last_error,
1966 }
1967}
1968
1969#[cfg(test)]
1970fn persisted_v5_record(record: ManagedSessionRecord) -> PersistedManagedSessionRecordV5 {
1971 PersistedManagedSessionRecordV5 {
1972 record_id: record.record_id,
1973 display_name: record.display_name,
1974 provider: record.provider,
1975 mode: record.mode,
1976 state: record.state,
1977 workspace_id: record.workspace_id,
1978 canonical_root: record.canonical_root,
1979 provider_session: record.provider_session,
1980 active_session: record.active_session,
1981 environment_profile: record.environment_profile,
1982 created_at_unix_ms: record.created_at_unix_ms,
1983 updated_at_unix_ms: record.updated_at_unix_ms,
1984 last_error: record.last_error,
1985 }
1986}
1987
1988pub fn validate_display_name(display_name: &str) -> io::Result<()> {
1989 if display_name.trim().is_empty()
1990 || display_name.len() > MAX_SESSION_DISPLAY_NAME_BYTES
1991 || display_name.chars().any(char::is_control)
1992 {
1993 return Err(invalid_data(format!(
1994 "session display name must contain 1..={MAX_SESSION_DISPLAY_NAME_BYTES} bytes and no control characters",
1995 )));
1996 }
1997 Ok(())
1998}
1999
2000fn validate_record(record: &ManagedSessionRecord) -> io::Result<()> {
2001 validate_display_name(&record.display_name)?;
2002 validate_root(require_utf8_state_path(&record.canonical_root)?)?;
2003 if record.created_at_unix_ms > record.updated_at_unix_ms {
2004 return Err(invalid_data("session record timestamps are inconsistent"));
2005 }
2006 if !record.context_binding_is_valid() {
2007 return Err(invalid_data(
2008 "session record context id and receipt are not correlated",
2009 ));
2010 }
2011 if !record.task_binding_is_valid()
2012 || record.task_binding.as_ref().is_some_and(|binding| {
2013 binding.changed_at_unix_ms < record.created_at_unix_ms
2014 || binding.changed_at_unix_ms > record.updated_at_unix_ms
2015 })
2016 {
2017 return Err(invalid_data("session record task binding is invalid"));
2018 }
2019 if let Some(identity) = &record.provider_session {
2020 identity
2021 .validate()
2022 .map_err(|error| invalid_data(format!("provider session identity is invalid: {error}")))?;
2023 }
2024 if record
2025 .last_error
2026 .as_ref()
2027 .is_some_and(|message| {
2028 message.len() > MAX_NODE_TEXT_BYTES || message.chars().any(|character| character == '\0')
2029 })
2030 {
2031 return Err(invalid_data("session record error is invalid"));
2032 }
2033 Ok(())
2034}
2035
2036pub fn sanitized_record_error_summary(message: &str) -> String {
2037 let normalized = message.to_ascii_lowercase();
2038 if matches!(
2039 normalized.as_str(),
2040 "workspace-unavailable"
2041 | "provider-session-scope-conflict"
2042 | "provider-session-live-conflict"
2043 | "managed-session-capacity"
2044 | "provider-session-identity-allocation-failed"
2045 | "provider-resume-rejected"
2046 | "provider-authentication-required"
2047 | "provider-readiness-failed"
2048 | "provider-process-exited"
2049 | "provider-resume-failed"
2050 | "durable-state-commit-failed"
2051 | "provider-runtime-failed"
2052 | "environment-profile-unavailable"
2053 | "bundle-unavailable"
2054 | "bundle-binding-mismatch"
2055 | "bundle-materialization-failed"
2056 ) {
2057 return normalized;
2058 }
2059 if normalized.contains("workspace is unavailable") {
2060 "workspace-unavailable".to_owned()
2061 } else if normalized.contains("another workspace or transport") {
2062 "provider-session-scope-conflict".to_owned()
2063 } else if normalized.contains("conflicts with another live managed record") {
2064 "provider-session-live-conflict".to_owned()
2065 } else if normalized.contains("managed session capacity") {
2066 "managed-session-capacity".to_owned()
2067 } else if normalized.contains("allocate a record") {
2068 "provider-session-identity-allocation-failed".to_owned()
2069 } else if normalized.contains("resume command was rejected") {
2070 "provider-resume-rejected".to_owned()
2071 } else if normalized.contains("authentication") || normalized.contains("sign in") {
2072 "provider-authentication-required".to_owned()
2073 } else if normalized.contains("readiness") {
2074 "provider-readiness-failed".to_owned()
2075 } else if normalized.contains("exited with code")
2076 || normalized.contains("no row for root pid")
2077 {
2078 "provider-process-exited".to_owned()
2079 } else if normalized.contains("resume") {
2080 "provider-resume-failed".to_owned()
2081 } else if normalized.contains("durable state") || normalized.contains("persistence") {
2082 "durable-state-commit-failed".to_owned()
2083 } else {
2084 "provider-runtime-failed".to_owned()
2085 }
2086}
2087
2088fn validate_root(root: &str) -> io::Result<()> {
2089 if root.is_empty()
2090 || root.len() > MAX_WORKSPACE_ROOT_BYTES
2091 || root.chars().any(char::is_control)
2092 || !Path::new(root).is_absolute()
2093 {
2094 return Err(invalid_data("durable workspace root is invalid"));
2095 }
2096 Ok(())
2097}
2098
2099#[derive(Clone, Copy, Eq, PartialEq)]
2100enum AtomicWriteFault {
2101 None,
2102 BeforePrimaryCommit,
2103 BeforeBackupRotation,
2104}
2105
2106fn atomic_write(
2107 path: &Path,
2108 bytes: &[u8],
2109 previous_primary_valid: bool,
2110) -> io::Result<Option<String>> {
2111 atomic_write_with_policy(
2112 path,
2113 bytes,
2114 AtomicWriteFault::None,
2115 previous_primary_valid,
2116 )
2117}
2118
2119#[cfg(test)]
2120fn atomic_write_with_fault(
2121 path: &Path,
2122 bytes: &[u8],
2123 fault: AtomicWriteFault,
2124) -> io::Result<Option<String>> {
2125 atomic_write_with_policy(path, bytes, fault, true)
2126}
2127
2128fn atomic_write_with_policy(
2129 path: &Path,
2130 bytes: &[u8],
2131 fault: AtomicWriteFault,
2132 previous_primary_valid: bool,
2133) -> io::Result<Option<String>> {
2134 let parent = path.parent().ok_or_else(|| {
2135 io::Error::new(io::ErrorKind::InvalidInput, "durable state path has no parent")
2136 })?;
2137 fs::create_dir_all(parent)?;
2138 #[cfg(unix)]
2139 secure_state_directory(parent)?;
2140 let (temporary, mut file) = create_unique_sibling_file(path, "tmp")?;
2141 let backup = sibling_path(path, "bak");
2142 let prepared = (|| {
2143 file.write_all(bytes)?;
2144 file.flush()?;
2145 file.sync_all()
2146 })();
2147 drop(file);
2148 if let Err(error) = prepared {
2149 let _ = fs::remove_file(&temporary);
2150 return Err(error);
2151 }
2152 let pending_backup = sibling_path(path, "pending-backup");
2153 if pending_backup.exists() {
2154 let pending_result = if backup.exists() {
2155 rotate_backup_after_commit(&backup, &pending_backup, AtomicWriteFault::None)
2156 } else {
2157 fs::rename(&pending_backup, &backup)
2158 };
2159 if let Err(error) = pending_result {
2160 let _ = fs::remove_file(&temporary);
2161 return Err(error);
2162 }
2163 }
2164 let displaced = if path.exists() {
2165 let displaced = if !previous_primary_valid {
2166 unique_sibling_path(path, "corrupt")?
2167 } else if backup.exists() {
2168 pending_backup.clone()
2169 } else {
2170 backup.clone()
2171 };
2172 if let Err(error) = fs::rename(path, &displaced) {
2173 let _ = fs::remove_file(&temporary);
2174 return Err(error);
2175 }
2176 Some(displaced)
2177 } else {
2178 None
2179 };
2180 let result = (|| {
2181 if fault == AtomicWriteFault::BeforePrimaryCommit {
2182 return Err(io::Error::new(
2183 io::ErrorKind::Other,
2184 "injected failure before durable primary commit",
2185 ));
2186 }
2187 fs::rename(&temporary, path)?;
2188 sync_parent_directory(path)
2189 })();
2190 if let Err(error) = result {
2191 let _ = fs::remove_file(&temporary);
2192 if let Some(displaced) = displaced.as_deref() {
2193 if !path.exists() {
2194 fs::rename(displaced, path).map_err(|rollback_error| {
2195 io::Error::new(
2196 rollback_error.kind(),
2197 format!(
2198 "durable primary commit failed ({error}); restoring the prior primary also failed: {rollback_error}"
2199 ),
2200 )
2201 })?;
2202 }
2203 }
2204 return Err(error);
2205 }
2206
2207 let backup_warning = if let Some(displaced) = displaced.as_deref() {
2208 if displaced == pending_backup {
2209 rotate_backup_after_commit(&backup, displaced, fault)
2210 .err()
2211 .map(|_| BACKUP_ROTATION_WARNING.to_owned())
2212 } else if displaced != backup {
2213 Some(CORRUPT_PRIMARY_PRESERVED_WARNING.to_owned())
2214 } else {
2215 None
2216 }
2217 } else {
2218 None
2219 };
2220 Ok(backup_warning)
2221}
2222
2223fn rotate_backup_after_commit(
2224 backup: &Path,
2225 previous_primary: &Path,
2226 fault: AtomicWriteFault,
2227) -> io::Result<()> {
2228 if fault == AtomicWriteFault::BeforeBackupRotation {
2229 return Err(io::Error::new(
2230 io::ErrorKind::Other,
2231 "injected failure before durable backup rotation",
2232 ));
2233 }
2234 let preserved_backup = unique_sibling_path(backup, "preserved")?;
2235 replace_file(backup, previous_primary, &preserved_backup)?;
2236 let _ = fs::remove_file(preserved_backup);
2237 Ok(())
2238}
2239
2240fn replace_file(replaced: &Path, replacement: &Path, preserved: &Path) -> io::Result<()> {
2241 #[cfg(windows)]
2242 {
2243 use std::ffi::c_void;
2244 use std::os::windows::ffi::OsStrExt;
2245
2246 #[link(name = "kernel32")]
2247 extern "system" {
2248 #[link_name = "ReplaceFileW"]
2249 fn replace_file_w(
2250 replaced_file_name: *const u16,
2251 replacement_file_name: *const u16,
2252 backup_file_name: *const u16,
2253 replace_flags: u32,
2254 exclude: *mut c_void,
2255 reserved: *mut c_void,
2256 ) -> i32;
2257 }
2258
2259 let replaced = replaced
2260 .as_os_str()
2261 .encode_wide()
2262 .chain(std::iter::once(0))
2263 .collect::<Vec<_>>();
2264 let replacement = replacement
2265 .as_os_str()
2266 .encode_wide()
2267 .chain(std::iter::once(0))
2268 .collect::<Vec<_>>();
2269 let preserved = preserved
2270 .as_os_str()
2271 .encode_wide()
2272 .chain(std::iter::once(0))
2273 .collect::<Vec<_>>();
2274 let replaced = unsafe {
2277 replace_file_w(
2278 replaced.as_ptr(),
2279 replacement.as_ptr(),
2280 preserved.as_ptr(),
2281 0,
2282 std::ptr::null_mut(),
2283 std::ptr::null_mut(),
2284 )
2285 };
2286 if replaced == 0 {
2287 Err(io::Error::last_os_error())
2288 } else {
2289 Ok(())
2290 }
2291 }
2292 #[cfg(unix)]
2293 {
2294 fs::rename(replaced, preserved)?;
2295 match fs::rename(replacement, replaced) {
2296 Ok(()) => sync_parent_directory(replaced),
2297 Err(error) => {
2298 let rollback = fs::rename(preserved, replaced);
2299 match rollback {
2300 Ok(()) => Err(error),
2301 Err(rollback_error) => Err(io::Error::new(
2302 rollback_error.kind(),
2303 format!(
2304 "durable backup rotation failed ({error}); restoring the prior backup also failed: {rollback_error}"
2305 ),
2306 )),
2307 }
2308 }
2309 }
2310 }
2311}
2312
2313#[cfg(windows)]
2314fn sync_parent_directory(_path: &Path) -> io::Result<()> {
2315 Ok(())
2316}
2317
2318#[cfg(unix)]
2319fn sync_parent_directory(path: &Path) -> io::Result<()> {
2320 let parent = path.parent().ok_or_else(|| {
2321 io::Error::new(io::ErrorKind::InvalidInput, "durable state path has no parent")
2322 })?;
2323 File::open(parent)?.sync_all()
2324}
2325
2326fn create_unique_sibling_file(path: &Path, suffix: &str) -> io::Result<(PathBuf, File)> {
2327 for _ in 0..32 {
2328 let candidate = unique_sibling_candidate(path, suffix);
2329 let mut options = OpenOptions::new();
2330 options.create_new(true).write(true);
2331 #[cfg(unix)]
2332 options.mode(0o600);
2333 match options.open(&candidate) {
2334 Ok(file) => return Ok((candidate, file)),
2335 Err(error) if error.kind() == io::ErrorKind::AlreadyExists => continue,
2336 Err(error) => return Err(error),
2337 }
2338 }
2339 Err(io::Error::new(
2340 io::ErrorKind::AlreadyExists,
2341 "could not allocate a unique durable state temporary file",
2342 ))
2343}
2344
2345fn unique_sibling_path(path: &Path, suffix: &str) -> io::Result<PathBuf> {
2346 for _ in 0..32 {
2347 let candidate = unique_sibling_candidate(path, suffix);
2348 if !candidate.exists() {
2349 return Ok(candidate);
2350 }
2351 }
2352 Err(io::Error::new(
2353 io::ErrorKind::AlreadyExists,
2354 "could not allocate a unique durable state sibling path",
2355 ))
2356}
2357
2358fn unique_sibling_candidate(path: &Path, suffix: &str) -> PathBuf {
2359 let name = path
2360 .file_name()
2361 .and_then(|name| name.to_str())
2362 .unwrap_or("state-v1.json");
2363 let sequence = NEXT_TEMPORARY_FILE.fetch_add(1, Ordering::Relaxed);
2364 path.with_file_name(format!(
2365 ".{name}.{suffix}-{}-{sequence}",
2366 std::process::id(),
2367 ))
2368}
2369
2370fn sibling_path(path: &Path, suffix: &str) -> PathBuf {
2371 let name = path
2372 .file_name()
2373 .and_then(|name| name.to_str())
2374 .unwrap_or("state-v1.json");
2375 path.with_file_name(format!(".{name}.{suffix}"))
2376}
2377
2378fn invalid_data(message: impl Into<String>) -> io::Error {
2379 io::Error::new(io::ErrorKind::InvalidData, message.into())
2380}
2381
2382#[cfg(test)]
2383mod tests {
2384 use super::*;
2385 use crate::protocol::{SessionMode, SessionRecordId};
2386 use gate4agent_types::{ProviderSessionIdentity, ProviderSessionKey};
2387 use std::sync::atomic::{AtomicU64, Ordering};
2388
2389 static NEXT_TEMP: AtomicU64 = AtomicU64::new(1);
2390
2391 fn agent(value: &str) -> AgentId {
2392 AgentId::new(value).unwrap()
2393 }
2394
2395 fn temp_path(test: &str) -> PathBuf {
2396 let unique = NEXT_TEMP.fetch_add(1, Ordering::Relaxed);
2397 std::env::temp_dir()
2398 .join(format!("gate4agent-node-state-{}-{unique}", std::process::id()))
2399 .join(format!("{test}.json"))
2400 }
2401
2402 fn fixture(path: &Path) -> (NodeId, BTreeMap<WorkspaceId, String>, ManagedSessionRecord) {
2403 let node_id = NodeId::new("node-a").unwrap();
2404 let workspace_id = WorkspaceId::new("primary").unwrap();
2405 let root = path.parent().unwrap().to_string_lossy().into_owned();
2406 let workspaces = BTreeMap::from([(workspace_id.clone(), root.clone())]);
2407 let record = ManagedSessionRecord {
2408 record_id: SessionRecordId::new("session-1").unwrap(),
2409 display_name: "Claude release check".to_owned(),
2410 provider: agent("claude"),
2411 mode: SessionMode::Pty,
2412 state: ManagedSessionState::Live,
2413 workspace_id,
2414 canonical_root: OpaqueHostPath::utf8(root).unwrap(),
2415 provider_session: Some(ProviderSessionIdentity {
2416 key: ProviderSessionKey::SessionId,
2417 id: "provider-session-1".to_owned(),
2418 transcript_path: Some(r"C:\private\provider-transcript.jsonl".to_owned()),
2419 }),
2420 active_session: None,
2421 environment_profile: None,
2422 bundle: None,
2423 context_id: None,
2424 context: None,
2425 exported_context: None,
2426 task_binding: None,
2427 created_at_unix_ms: 10,
2428 updated_at_unix_ms: 11,
2429 last_error: None,
2430 };
2431 (node_id, workspaces, record)
2432 }
2433
2434 fn managed_lease(path: &Path, id: &str, workspace: &str, branch: &str) -> ManagedWorktreeLeaseRecord {
2435 ManagedWorktreeLeaseRecord {
2436 lease_id: crate::protocol::ManagedWorktreeLeaseId::new(id).unwrap(),
2437 source_workspace_id: WorkspaceId::new("primary").unwrap(),
2438 workspace_id: WorkspaceId::new(workspace).unwrap(),
2439 profile_id: crate::protocol::WorktreeProfileId::new("default").unwrap(),
2440 profile_revision: crate::protocol::WorktreeProfileRevision::new("v1").unwrap(),
2441 target_root: path.to_string_lossy().into_owned(),
2442 branch: branch.to_owned(),
2443 base_commit: "0123456789abcdef0123456789abcdef01234567".to_owned(),
2444 expected_head: Some("0123456789abcdef0123456789abcdef01234567".to_owned()),
2445 retention: crate::protocol::ManagedWorktreeRetention::RemoveWhenReleased,
2446 state: crate::protocol::ManagedWorktreeLeaseState::Ready,
2447 session_holders: Vec::new(),
2448 record_holders: Vec::new(),
2449 cleanup_failure: None,
2450 created_at_unix_ms: 10,
2451 updated_at_unix_ms: 11,
2452 }
2453 }
2454
2455 fn materialization_ownership(
2456 path: &Path,
2457 id: &str,
2458 owner: MaterializationOwner,
2459 managed_lease_id: Option<crate::protocol::ManagedWorktreeLeaseId>,
2460 ) -> MaterializationOwnershipRecord {
2461 let root = path
2462 .parent()
2463 .unwrap()
2464 .join("session-environments")
2465 .join(id);
2466 MaterializationOwnershipRecord::from_persisted(
2467 MaterializationId::new(id).unwrap(),
2468 Some(ResolvedEnvironmentProfileReceipt {
2469 profile_id: crate::protocol::SpawnEnvironmentProfileId::new("local-claude")
2470 .unwrap(),
2471 profile_revision: crate::protocol::SpawnEnvironmentProfileRevision::new("r1")
2472 .unwrap(),
2473 network_allowlist: None,
2474 browser_profile_id: None,
2475 }),
2476 None,
2477 None,
2478 owner,
2479 managed_lease_id,
2480 MaterializationState::Ready,
2481 root.clone(),
2482 root.join("home"),
2483 root.join("bundle"),
2484 root.join("plugin-data"),
2485 Vec::new(),
2486 10,
2487 11,
2488 )
2489 .unwrap()
2490 }
2491
2492 fn persisted_materialization(
2493 record: &MaterializationOwnershipRecord,
2494 ) -> PersistedMaterializationRecordV6 {
2495 PersistedMaterializationRecordV6 {
2496 materialization_id: record.id().clone(),
2497 environment_profile: record.environment_profile().cloned().unwrap(),
2498 owner: record.owner().clone(),
2499 managed_lease_id: record.managed_lease_id().cloned(),
2500 state: record.state(),
2501 root: path_to_opaque(record.root()).unwrap(),
2502 provider_home: path_to_opaque(record.provider_home()).unwrap(),
2503 declared_paths: record.declared_paths().to_vec(),
2504 created_at_unix_ms: record.created_at_unix_ms(),
2505 updated_at_unix_ms: record.updated_at_unix_ms(),
2506 }
2507 }
2508
2509 fn context_receipt(id: &str, digest_byte: char) -> ResolvedContextPackReceipt {
2510 ResolvedContextPackReceipt {
2511 id: crate::protocol::SpawnContextId::new(id).unwrap(),
2512 digest: crate::protocol::SpawnContextDigest::new(format!(
2513 "sha256:{}",
2514 digest_byte.to_string().repeat(64),
2515 ))
2516 .unwrap(),
2517 lineage: crate::protocol::ContextPackLineageReceipt {
2518 source_node_id: NodeId::new("source-node").unwrap(),
2519 source_session: SessionAddress {
2520 workspace_id: WorkspaceId::new("source-workspace").unwrap(),
2521 session: crate::protocol::SessionKey {
2522 instance_id: gate4agent_types::AgentInstanceId(7),
2523 generation: gate4agent_types::SessionGeneration(3),
2524 },
2525 },
2526 source_provider: agent("codex"),
2527 },
2528 source_message_count: 12,
2529 retained_message_count: 9,
2530 byte_len: 4096,
2531 truncated: true,
2532 }
2533 }
2534
2535 fn context_materialization(
2536 path: &Path,
2537 id: &str,
2538 record_id: SessionRecordId,
2539 context: ResolvedContextPackReceipt,
2540 ) -> MaterializationOwnershipRecord {
2541 let root = path
2542 .parent()
2543 .unwrap()
2544 .join("session-environments")
2545 .join(id);
2546 MaterializationOwnershipRecord::from_persisted(
2547 MaterializationId::new(id).unwrap(),
2548 None,
2549 None,
2550 Some(context),
2551 MaterializationOwner::Record { record_id },
2552 None,
2553 MaterializationState::Ready,
2554 root.clone(),
2555 root.join("home"),
2556 root.join("bundle"),
2557 root.join("plugin-data"),
2558 Vec::new(),
2559 10,
2560 11,
2561 )
2562 .unwrap()
2563 }
2564
2565 #[test]
2566 fn v4_roundtrip_preserves_host_only_managed_lease_identity() {
2567 let path = temp_path("v4-managed-roundtrip");
2568 std::fs::create_dir_all(path.parent().unwrap()).unwrap();
2569 let (node_id, workspaces, record) = fixture(&path);
2570 let lease = managed_lease(
2571 &path.parent().unwrap().join("managed-a"),
2572 "mw-a",
2573 "managed-a",
2574 "gate4agent/mw-a",
2575 );
2576 save_v4(
2577 Some(&path),
2578 &node_id,
2579 &workspaces,
2580 &[record],
2581 &[lease.clone()],
2582 &[],
2583 ).unwrap();
2584 let header: PersistedNodeStateHeader = serde_json::from_slice(
2585 &std::fs::read(&path).unwrap(),
2586 ).unwrap();
2587 assert_eq!(header.version, NODE_STATE_SCHEMA_V4);
2588 let loaded = load(Some(&path), &node_id).unwrap();
2589 assert_eq!(loaded.managed_worktrees, vec![lease]);
2590 assert!(loaded.managed_worktree_tombstones.is_empty());
2591 std::fs::remove_dir_all(path.parent().unwrap()).unwrap();
2592 }
2593
2594 #[test]
2595 fn v5_roundtrip_preserves_only_opaque_environment_profile_identity() {
2596 let path = temp_path("v5-environment-profile-roundtrip");
2597 std::fs::create_dir_all(path.parent().unwrap()).unwrap();
2598 let (node_id, workspaces, mut record) = fixture(&path);
2599 let environment_profile = ResolvedEnvironmentProfileReceipt {
2600 profile_id: crate::protocol::SpawnEnvironmentProfileId::new("local-claude").unwrap(),
2601 profile_revision: crate::protocol::SpawnEnvironmentProfileRevision::new(
2602 "local-claude-r1",
2603 )
2604 .unwrap(),
2605 network_allowlist: None,
2606 browser_profile_id: None,
2607 };
2608 record.environment_profile = Some(environment_profile.clone());
2609
2610 save_v5(
2611 Some(&path),
2612 &node_id,
2613 &workspaces,
2614 &[record],
2615 &[],
2616 &[],
2617 )
2618 .unwrap();
2619 let bytes = std::fs::read(&path).unwrap();
2620 let header: PersistedNodeStateHeader = serde_json::from_slice(&bytes).unwrap();
2621 assert_eq!(header.version, NODE_STATE_SCHEMA_V5);
2622 let serialized = String::from_utf8(bytes).unwrap();
2623 assert!(serialized.contains("local-claude"));
2624 assert!(!serialized.contains("ANTHROPIC_AUTH_TOKEN"));
2625 assert!(!serialized.contains("secret"));
2626
2627 let loaded = load(Some(&path), &node_id).unwrap();
2628 assert_eq!(loaded.records.len(), 1);
2629 assert_eq!(
2630 loaded.records[0].environment_profile.as_ref(),
2631 Some(&environment_profile),
2632 );
2633 std::fs::remove_dir_all(path.parent().unwrap()).unwrap();
2634 }
2635
2636 #[test]
2637 fn v4_to_v5_migration_defaults_environment_profile_to_none() {
2638 let path = temp_path("v4-to-v5-environment-profile");
2639 std::fs::create_dir_all(path.parent().unwrap()).unwrap();
2640 let (node_id, workspaces, record) = fixture(&path);
2641 save_v4(Some(&path), &node_id, &workspaces, &[record], &[], &[]).unwrap();
2642
2643 let loaded = load(Some(&path), &node_id).unwrap();
2644 assert_eq!(loaded.records.len(), 1);
2645 assert!(loaded.records[0].environment_profile.is_none());
2646 save_v5(
2647 Some(&path),
2648 &node_id,
2649 &workspaces,
2650 &loaded.records,
2651 &[],
2652 &[],
2653 )
2654 .unwrap();
2655 let header: PersistedNodeStateHeader = serde_json::from_slice(
2656 &std::fs::read(&path).unwrap(),
2657 )
2658 .unwrap();
2659 assert_eq!(header.version, NODE_STATE_SCHEMA_V5);
2660 std::fs::remove_dir_all(path.parent().unwrap()).unwrap();
2661 }
2662
2663 #[test]
2664 fn v5_to_v6_migration_starts_with_empty_materialization_registry() {
2665 let path = temp_path("v5-to-v6-materializations");
2666 std::fs::create_dir_all(path.parent().unwrap()).unwrap();
2667 let (node_id, workspaces, record) = fixture(&path);
2668 save_v5(Some(&path), &node_id, &workspaces, &[record], &[], &[]).unwrap();
2669 let loaded = load(Some(&path), &node_id).unwrap();
2670 assert!(loaded.materializations.is_empty());
2671 save_v7(
2672 Some(&path),
2673 &node_id,
2674 &workspaces,
2675 &loaded.records,
2676 &[],
2677 &[],
2678 &loaded.materializations,
2679 ).unwrap();
2680 let header: PersistedNodeStateHeader = serde_json::from_slice(&std::fs::read(&path).unwrap()).unwrap();
2681 assert_eq!(header.version, NODE_STATE_SCHEMA_V7);
2682 std::fs::remove_dir_all(path.parent().unwrap()).unwrap();
2683 }
2684
2685 #[test]
2686 fn v6_roundtrip_persists_ownership_without_secret_reference_or_value() {
2687 let path = temp_path("v6-materialization-roundtrip");
2688 std::fs::create_dir_all(path.parent().unwrap()).unwrap();
2689 let (node_id, workspaces, _) = fixture(&path);
2690 let materialization_root = path.parent().unwrap().join("session-environments").join("m-1");
2691 let ownership = MaterializationOwnershipRecord::from_persisted(
2692 MaterializationId::new("m-1").unwrap(),
2693 Some(ResolvedEnvironmentProfileReceipt {
2694 profile_id: crate::protocol::SpawnEnvironmentProfileId::new("local-claude").unwrap(),
2695 profile_revision: crate::protocol::SpawnEnvironmentProfileRevision::new("r1").unwrap(),
2696 network_allowlist: None,
2697 browser_profile_id: None,
2698 }),
2699 None,
2700 None,
2701 MaterializationOwner::Session {
2702 incarnation_id: crate::protocol::NodeIncarnationId::from_bytes([3; crate::protocol::NODE_INCARNATION_ID_BYTES]),
2703 instance_id: gate4agent_types::AgentInstanceId(9),
2704 generation: gate4agent_types::SessionGeneration(2),
2705 },
2706 None,
2707 MaterializationState::Ready,
2708 materialization_root.clone(),
2709 materialization_root.join("home"),
2710 materialization_root.join("bundle"),
2711 materialization_root.join("plugin-data"),
2712 vec![MaterializedPathDeclaration {
2713 class: crate::materialization::NodeSessionPathClass::Config,
2714 relative_path: PathBuf::from("provider/settings.json"),
2715 kind: crate::materialization::MaterializedPathKind::Generated,
2716 }],
2717 10,
2718 11,
2719 ).unwrap();
2720 save_v7(Some(&path), &node_id, &workspaces, &[], &[], &[], &[ownership.clone()]).unwrap();
2721 let bytes = std::fs::read(&path).unwrap();
2722 let serialized = String::from_utf8(bytes).unwrap();
2723 assert!(!serialized.contains("fixture-secret-reference"));
2724 assert!(!serialized.contains("fixture-secret-value"));
2725 let loaded = load(Some(&path), &node_id).unwrap();
2726 assert!(loaded.materializations == vec![ownership]);
2727 std::fs::remove_dir_all(path.parent().unwrap()).unwrap();
2728 }
2729
2730 #[test]
2731 fn v6_save_rejects_unknown_materialization_managed_lease() {
2732 let path = temp_path("v6-unknown-materialization-lease");
2733 std::fs::create_dir_all(path.parent().unwrap()).unwrap();
2734 let node_id = NodeId::new("node-a").unwrap();
2735 let ownership = materialization_ownership(
2736 &path,
2737 "m-unknown",
2738 MaterializationOwner::Session {
2739 incarnation_id: crate::protocol::NodeIncarnationId::from_bytes([
2740 3;
2741 crate::protocol::NODE_INCARNATION_ID_BYTES
2742 ]),
2743 instance_id: gate4agent_types::AgentInstanceId(9),
2744 generation: gate4agent_types::SessionGeneration(2),
2745 },
2746 Some(crate::protocol::ManagedWorktreeLeaseId::new("mw-missing").unwrap()),
2747 );
2748
2749 let error = save_v7(
2750 Some(&path),
2751 &node_id,
2752 &BTreeMap::new(),
2753 &[],
2754 &[],
2755 &[],
2756 &[ownership],
2757 )
2758 .unwrap_err();
2759 assert_eq!(error.kind(), io::ErrorKind::InvalidData);
2760 assert!(!path.exists());
2761 std::fs::remove_dir_all(path.parent().unwrap()).unwrap();
2762 }
2763
2764 #[test]
2765 fn v6_decode_rejects_tombstoned_materialization_managed_lease() {
2766 let path = temp_path("v6-tombstoned-materialization-lease");
2767 std::fs::create_dir_all(path.parent().unwrap()).unwrap();
2768 let node_id = NodeId::new("node-a").unwrap();
2769 let mut tombstone = managed_lease(
2770 &path.parent().unwrap().join("managed-a"),
2771 "mw-a",
2772 "managed-a",
2773 "gate4agent/mw-a",
2774 );
2775 tombstone.state = crate::protocol::ManagedWorktreeLeaseState::Removed;
2776 let ownership = materialization_ownership(
2777 &path,
2778 "m-tombstone",
2779 MaterializationOwner::Session {
2780 incarnation_id: crate::protocol::NodeIncarnationId::from_bytes([
2781 3;
2782 crate::protocol::NODE_INCARNATION_ID_BYTES
2783 ]),
2784 instance_id: gate4agent_types::AgentInstanceId(9),
2785 generation: gate4agent_types::SessionGeneration(2),
2786 },
2787 Some(tombstone.lease_id.clone()),
2788 );
2789 let state = PersistedNodeStateV6 {
2790 version: NODE_STATE_SCHEMA_V6,
2791 node_id: node_id.clone(),
2792 workspaces: Vec::new(),
2793 session_records: Vec::new(),
2794 managed_worktrees: Vec::new(),
2795 managed_worktree_tombstones: vec![tombstone],
2796 materializations: vec![persisted_materialization(&ownership)],
2797 };
2798 std::fs::write(&path, serde_json::to_vec(&state).unwrap()).unwrap();
2799
2800 assert_eq!(
2801 load(Some(&path), &node_id).unwrap_err().kind(),
2802 io::ErrorKind::InvalidData,
2803 );
2804 std::fs::remove_dir_all(path.parent().unwrap()).unwrap();
2805 }
2806
2807 #[test]
2808 fn v6_save_rejects_record_materialization_for_foreign_lease_workspace() {
2809 let path = temp_path("v6-record-foreign-lease-workspace");
2810 std::fs::create_dir_all(path.parent().unwrap()).unwrap();
2811 let (node_id, workspaces, record) = fixture(&path);
2812 let lease = managed_lease(
2813 &path.parent().unwrap().join("managed-a"),
2814 "mw-a",
2815 "managed-a",
2816 "gate4agent/mw-a",
2817 );
2818 let ownership = materialization_ownership(
2819 &path,
2820 "m-foreign",
2821 MaterializationOwner::Record {
2822 record_id: record.record_id.clone(),
2823 },
2824 Some(lease.lease_id.clone()),
2825 );
2826
2827 let error = save_v7(
2828 Some(&path),
2829 &node_id,
2830 &workspaces,
2831 &[record],
2832 &[lease],
2833 &[],
2834 &[ownership],
2835 )
2836 .unwrap_err();
2837 assert_eq!(error.kind(), io::ErrorKind::InvalidData);
2838 assert!(!path.exists());
2839 std::fs::remove_dir_all(path.parent().unwrap()).unwrap();
2840 }
2841
2842 #[test]
2843 fn v6_roundtrip_accepts_session_materialization_for_active_lease() {
2844 let path = temp_path("v6-session-active-materialization-lease");
2845 std::fs::create_dir_all(path.parent().unwrap()).unwrap();
2846 let node_id = NodeId::new("node-a").unwrap();
2847 let lease = managed_lease(
2848 &path.parent().unwrap().join("managed-a"),
2849 "mw-a",
2850 "managed-a",
2851 "gate4agent/mw-a",
2852 );
2853 let ownership = materialization_ownership(
2854 &path,
2855 "m-active",
2856 MaterializationOwner::Session {
2857 incarnation_id: crate::protocol::NodeIncarnationId::from_bytes([
2858 3;
2859 crate::protocol::NODE_INCARNATION_ID_BYTES
2860 ]),
2861 instance_id: gate4agent_types::AgentInstanceId(9),
2862 generation: gate4agent_types::SessionGeneration(2),
2863 },
2864 Some(lease.lease_id.clone()),
2865 );
2866
2867 save_v7(
2868 Some(&path),
2869 &node_id,
2870 &BTreeMap::new(),
2871 &[],
2872 &[lease.clone()],
2873 &[],
2874 &[ownership.clone()],
2875 )
2876 .unwrap();
2877 let loaded = load(Some(&path), &node_id).unwrap();
2878 assert_eq!(loaded.managed_worktrees, vec![lease]);
2879 assert!(loaded.materializations == vec![ownership]);
2880 std::fs::remove_dir_all(path.parent().unwrap()).unwrap();
2881 }
2882
2883 #[test]
2884 fn v6_rejects_declared_path_prefix_collisions() {
2885 let path = temp_path("v6-prefix-collision");
2886 std::fs::create_dir_all(path.parent().unwrap()).unwrap();
2887 let node_id = NodeId::new("node-a").unwrap();
2888 let materialization_root = path.parent().unwrap().join("session-environments").join("m-1");
2889 let state = PersistedNodeStateV6 {
2890 version: NODE_STATE_SCHEMA_V6,
2891 node_id: node_id.clone(),
2892 workspaces: Vec::new(),
2893 session_records: Vec::new(),
2894 managed_worktrees: Vec::new(),
2895 managed_worktree_tombstones: Vec::new(),
2896 materializations: vec![PersistedMaterializationRecordV6 {
2897 materialization_id: MaterializationId::new("m-1").unwrap(),
2898 environment_profile: ResolvedEnvironmentProfileReceipt {
2899 profile_id: crate::protocol::SpawnEnvironmentProfileId::new("local-claude").unwrap(),
2900 profile_revision: crate::protocol::SpawnEnvironmentProfileRevision::new("r1").unwrap(),
2901 network_allowlist: None,
2902 browser_profile_id: None,
2903 },
2904 owner: MaterializationOwner::Session {
2905 incarnation_id: crate::protocol::NodeIncarnationId::from_bytes([3; crate::protocol::NODE_INCARNATION_ID_BYTES]),
2906 instance_id: gate4agent_types::AgentInstanceId(9),
2907 generation: gate4agent_types::SessionGeneration(2),
2908 },
2909 managed_lease_id: None,
2910 state: MaterializationState::Ready,
2911 root: path_to_opaque(&materialization_root).unwrap(),
2912 provider_home: path_to_opaque(&materialization_root.join("home")).unwrap(),
2913 declared_paths: vec![
2914 MaterializedPathDeclaration {
2915 class: crate::materialization::NodeSessionPathClass::Config,
2916 relative_path: PathBuf::from("provider"),
2917 kind: crate::materialization::MaterializedPathKind::Generated,
2918 },
2919 MaterializedPathDeclaration {
2920 class: crate::materialization::NodeSessionPathClass::Config,
2921 relative_path: PathBuf::from("provider/settings.json"),
2922 kind: crate::materialization::MaterializedPathKind::Generated,
2923 },
2924 ],
2925 created_at_unix_ms: 10,
2926 updated_at_unix_ms: 11,
2927 }],
2928 };
2929 std::fs::write(&path, serde_json::to_vec(&state).unwrap()).unwrap();
2930 assert_eq!(load(Some(&path), &node_id).unwrap_err().kind(), io::ErrorKind::InvalidData);
2931 std::fs::remove_dir_all(path.parent().unwrap()).unwrap();
2932 }
2933
2934 #[test]
2935 fn v6_rejects_case_ambiguous_materialization_ids() {
2936 let path = temp_path("v6-case-ambiguous-materialization-id");
2937 std::fs::create_dir_all(path.parent().unwrap()).unwrap();
2938 let node_id = NodeId::new("node-a").unwrap();
2939 let first_root = path.parent().unwrap().join("session-environments").join("a");
2940 let second_root = path.parent().unwrap().join("session-environments").join("b");
2941 let first = PersistedMaterializationRecordV6 {
2942 materialization_id: MaterializationId::new("a").unwrap(),
2943 environment_profile: ResolvedEnvironmentProfileReceipt {
2944 profile_id: crate::protocol::SpawnEnvironmentProfileId::new("local-claude").unwrap(),
2945 profile_revision: crate::protocol::SpawnEnvironmentProfileRevision::new("r1").unwrap(),
2946 network_allowlist: None,
2947 browser_profile_id: None,
2948 },
2949 owner: MaterializationOwner::Session {
2950 incarnation_id: crate::protocol::NodeIncarnationId::from_bytes([3; crate::protocol::NODE_INCARNATION_ID_BYTES]),
2951 instance_id: gate4agent_types::AgentInstanceId(9),
2952 generation: gate4agent_types::SessionGeneration(2),
2953 },
2954 managed_lease_id: None,
2955 state: MaterializationState::Ready,
2956 root: path_to_opaque(&first_root).unwrap(),
2957 provider_home: path_to_opaque(&first_root.join("home")).unwrap(),
2958 declared_paths: Vec::new(),
2959 created_at_unix_ms: 10,
2960 updated_at_unix_ms: 11,
2961 };
2962 let mut second = first.clone();
2963 second.materialization_id = MaterializationId::new("b").unwrap();
2964 second.owner = MaterializationOwner::Session {
2965 incarnation_id: crate::protocol::NodeIncarnationId::from_bytes([3; crate::protocol::NODE_INCARNATION_ID_BYTES]),
2966 instance_id: gate4agent_types::AgentInstanceId(10),
2967 generation: gate4agent_types::SessionGeneration(2),
2968 };
2969 second.root = path_to_opaque(&second_root).unwrap();
2970 second.provider_home = path_to_opaque(&second_root.join("home")).unwrap();
2971 let state = PersistedNodeStateV6 {
2972 version: NODE_STATE_SCHEMA_V6,
2973 node_id: node_id.clone(),
2974 workspaces: Vec::new(),
2975 session_records: Vec::new(),
2976 managed_worktrees: Vec::new(),
2977 managed_worktree_tombstones: Vec::new(),
2978 materializations: vec![first, second],
2979 };
2980 let mut value = serde_json::to_value(state).unwrap();
2981 value["materializations"][1]["materialization_id"] = serde_json::Value::String("A".to_owned());
2982 std::fs::write(&path, serde_json::to_vec(&value).unwrap()).unwrap();
2983 assert_eq!(load(Some(&path), &node_id).unwrap_err().kind(), io::ErrorKind::InvalidData);
2984 std::fs::remove_dir_all(path.parent().unwrap()).unwrap();
2985 }
2986
2987 #[test]
2988 fn malformed_v4_duplicate_branch_is_rejected_without_startup_panic() {
2989 let path = temp_path("v4-managed-duplicate-branch");
2990 std::fs::create_dir_all(path.parent().unwrap()).unwrap();
2991 let (node_id, workspaces, record) = fixture(&path);
2992 let first = managed_lease(
2993 &path.parent().unwrap().join("managed-a"),
2994 "mw-a",
2995 "managed-a",
2996 "gate4agent/shared",
2997 );
2998 let second = managed_lease(
2999 &path.parent().unwrap().join("managed-b"),
3000 "mw-b",
3001 "managed-b",
3002 "gate4agent/shared",
3003 );
3004 let state = PersistedNodeStateV4 {
3005 version: NODE_STATE_SCHEMA_V4,
3006 node_id: node_id.clone(),
3007 workspaces: workspaces.into_iter().map(|(workspace_id, canonical_root)| {
3008 PersistedWorkspaceV2 {
3009 workspace_id,
3010 canonical_root: OpaqueHostPath::utf8(canonical_root).unwrap(),
3011 }
3012 }).collect(),
3013 session_records: vec![persisted_v3_record(record)],
3014 managed_worktrees: vec![first, second],
3015 managed_worktree_tombstones: Vec::new(),
3016 };
3017 std::fs::write(&path, serde_json::to_vec(&state).unwrap()).unwrap();
3018 assert_eq!(load(Some(&path), &node_id).unwrap_err().kind(), io::ErrorKind::InvalidData);
3019 std::fs::remove_dir_all(path.parent().unwrap()).unwrap();
3020 }
3021
3022 fn persisted_v1_record(record: ManagedSessionRecord) -> PersistedManagedSessionRecordV1 {
3023 PersistedManagedSessionRecordV1 {
3024 record_id: record.record_id,
3025 display_name: record.display_name,
3026 provider: LegacyAgentProvider::from_agent_id(&record.provider),
3027 mode: record.mode,
3028 state: record.state,
3029 workspace_id: record.workspace_id,
3030 canonical_root: record.canonical_root.as_utf8().unwrap().to_owned(),
3031 provider_session: record.provider_session,
3032 active_session: record.active_session,
3033 created_at_unix_ms: record.created_at_unix_ms,
3034 updated_at_unix_ms: record.updated_at_unix_ms,
3035 last_error: record.last_error,
3036 }
3037 }
3038
3039 fn persisted_v1_state(
3040 version: u16,
3041 node_id: NodeId,
3042 workspaces: &BTreeMap<WorkspaceId, String>,
3043 records: Vec<ManagedSessionRecord>,
3044 ) -> PersistedNodeStateV1 {
3045 PersistedNodeStateV1 {
3046 version,
3047 node_id,
3048 workspaces: workspaces
3049 .iter()
3050 .map(|(workspace_id, canonical_root)| PersistedWorkspaceV1 {
3051 workspace_id: workspace_id.clone(),
3052 canonical_root: canonical_root.clone(),
3053 })
3054 .collect(),
3055 session_records: records.into_iter().map(persisted_v1_record).collect(),
3056 }
3057 }
3058
3059 fn persisted_v2_state(
3060 node_id: NodeId,
3061 workspaces: &BTreeMap<WorkspaceId, String>,
3062 records: Vec<ManagedSessionRecord>,
3063 ) -> PersistedNodeStateV2 {
3064 PersistedNodeStateV2 {
3065 version: NODE_STATE_SCHEMA_V2,
3066 node_id,
3067 workspaces: workspaces
3068 .iter()
3069 .map(|(workspace_id, canonical_root)| PersistedWorkspaceV2 {
3070 workspace_id: workspace_id.clone(),
3071 canonical_root: OpaqueHostPath::utf8(canonical_root.clone()).unwrap(),
3072 })
3073 .collect(),
3074 session_records: records
3075 .into_iter()
3076 .map(|record| PersistedManagedSessionRecordV2 {
3077 record_id: record.record_id,
3078 display_name: record.display_name,
3079 provider: LegacyAgentProvider::from_agent_id(&record.provider),
3080 mode: record.mode,
3081 state: record.state,
3082 workspace_id: record.workspace_id,
3083 canonical_root: record.canonical_root,
3084 provider_session: record.provider_session,
3085 active_session: record.active_session,
3086 created_at_unix_ms: record.created_at_unix_ms,
3087 updated_at_unix_ms: record.updated_at_unix_ms,
3088 last_error: record.last_error,
3089 })
3090 .collect(),
3091 }
3092 }
3093
3094 #[test]
3095 fn durable_state_round_trips_without_ephemeral_active_binding() {
3096 let path = temp_path("round-trip");
3097 let (node_id, workspaces, mut record) = fixture(&path);
3098 record.provider_session.as_mut().unwrap().transcript_path =
3099 Some("C:\\private\\bearer-secret\nprovider-transcript.jsonl".to_owned());
3100 save(Some(&path), &node_id, &workspaces, &[record.clone()]).unwrap();
3101 let persisted = fs::read_to_string(&path).unwrap();
3102 assert_eq!(
3103 serde_json::from_str::<PersistedNodeStateHeader>(&persisted)
3104 .unwrap()
3105 .version,
3106 NODE_STATE_SCHEMA_V3,
3107 );
3108 assert!(!persisted.contains("provider-transcript.jsonl"));
3109 assert!(!persisted.contains(r"C:\private"));
3110 let loaded = load(Some(&path), &node_id).unwrap();
3111 assert_eq!(loaded.workspaces, workspaces);
3112 assert_eq!(loaded.records.len(), 1);
3113 assert_eq!(loaded.records[0].record_id, record.record_id);
3114 assert_eq!(loaded.records[0].state, ManagedSessionState::Dormant);
3115 assert_eq!(loaded.records[0].active_session, None);
3116 assert_eq!(
3117 loaded.records[0]
3118 .provider_session
3119 .as_ref()
3120 .unwrap()
3121 .transcript_path,
3122 None,
3123 );
3124 let _ = fs::remove_dir_all(path.parent().unwrap());
3125 }
3126
3127 #[test]
3128 fn v3_state_round_trips_open_provider_id_exactly() {
3129 let path = temp_path("v3-open-provider-round-trip");
3130 let (node_id, workspaces, mut record) = fixture(&path);
3131 record.provider = agent("third-party-agent");
3132 save(Some(&path), &node_id, &workspaces, &[record.clone()]).unwrap();
3133
3134 let persisted: PersistedNodeStateV3 =
3135 serde_json::from_slice(&fs::read(&path).unwrap()).unwrap();
3136 assert_eq!(persisted.version, NODE_STATE_SCHEMA_V3);
3137 assert_eq!(persisted.session_records.len(), 1);
3138 assert_eq!(persisted.session_records[0].provider, agent("third-party-agent"));
3139
3140 let loaded = load(Some(&path), &node_id).unwrap();
3141 assert_eq!(loaded.records.len(), 1);
3142 assert_eq!(loaded.records[0].provider, agent("third-party-agent"));
3143 assert_eq!(loaded.records[0].record_id, record.record_id);
3144 let _ = fs::remove_dir_all(path.parent().unwrap());
3145 }
3146
3147 #[test]
3148 fn semantic_provider_session_key_ignores_only_transcript_path() {
3149 let without_path = ProviderSessionIdentity {
3150 key: ProviderSessionKey::SessionId,
3151 id: "provider-session-1".to_owned(),
3152 transcript_path: None,
3153 };
3154 let with_path = ProviderSessionIdentity {
3155 transcript_path: Some(r"C:\private\provider-transcript.jsonl".to_owned()),
3156 ..without_path.clone()
3157 };
3158 assert_eq!(
3159 provider_session_semantic_key(&agent("claude"), &without_path),
3160 provider_session_semantic_key(&agent("claude"), &with_path),
3161 );
3162 assert_ne!(
3163 provider_session_semantic_key(&agent("claude"), &without_path),
3164 provider_session_semantic_key(&agent("codex"), &with_path),
3165 );
3166 let conversation = ProviderSessionIdentity {
3167 key: ProviderSessionKey::ConversationId,
3168 ..with_path
3169 };
3170 assert_ne!(
3171 provider_session_semantic_key(&agent("claude"), &without_path),
3172 provider_session_semantic_key(&agent("claude"), &conversation),
3173 );
3174 }
3175
3176 #[test]
3177 fn legacy_state_transcript_path_is_removed_from_loaded_identity() {
3178 let path = temp_path("legacy-transcript-path");
3179 let (node_id, workspaces, mut record) = fixture(&path);
3180 record.provider_session.as_mut().unwrap().transcript_path =
3181 Some("C:\\private\\bearer-secret\nprovider-transcript.jsonl".to_owned());
3182 let legacy = persisted_v1_state(
3183 NODE_STATE_SCHEMA_V1,
3184 node_id.clone(),
3185 &workspaces,
3186 vec![record],
3187 );
3188 fs::create_dir_all(path.parent().unwrap()).unwrap();
3189 fs::write(&path, serde_json::to_vec_pretty(&legacy).unwrap()).unwrap();
3190 let loaded = load(Some(&path), &node_id).unwrap();
3191 assert_eq!(loaded.records.len(), 1);
3192 assert_eq!(
3193 loaded.records[0]
3194 .provider_session
3195 .as_ref()
3196 .unwrap()
3197 .transcript_path,
3198 None,
3199 );
3200 let snapshot = serde_json::to_string(&loaded.records).unwrap();
3201 assert!(!snapshot.contains("provider-transcript.jsonl"));
3202 assert!(!snapshot.contains("bearer-secret"));
3203 assert!(!snapshot.contains(r"C:\private"));
3204 let _ = fs::remove_dir_all(path.parent().unwrap());
3205 }
3206
3207 #[test]
3208 fn v1_state_loads_losslessly_and_next_save_migrates_to_v3() {
3209 let path = temp_path("v1-migration");
3210 let backup = sibling_path(&path, "bak");
3211 let (node_id, workspaces, record) = fixture(&path);
3212 let legacy = persisted_v1_state(
3213 NODE_STATE_SCHEMA_V1,
3214 node_id.clone(),
3215 &workspaces,
3216 vec![record.clone()],
3217 );
3218 let legacy_bytes = serde_json::to_vec_pretty(&legacy).unwrap();
3219 fs::create_dir_all(path.parent().unwrap()).unwrap();
3220 fs::write(&path, &legacy_bytes).unwrap();
3221
3222 let loaded = load(Some(&path), &node_id).unwrap();
3223 assert_eq!(loaded.workspaces, workspaces);
3224 assert_eq!(loaded.records[0].record_id, record.record_id);
3225 assert_eq!(
3226 loaded.records[0].canonical_root,
3227 record.canonical_root,
3228 );
3229 save(Some(&path), &node_id, &loaded.workspaces, &loaded.records).unwrap();
3230
3231 let migrated = fs::read(&path).unwrap();
3232 let header: PersistedNodeStateHeader = serde_json::from_slice(&migrated).unwrap();
3233 assert_eq!(header.version, NODE_STATE_SCHEMA_V3);
3234 assert_eq!(fs::read(&backup).unwrap(), legacy_bytes);
3235 let reloaded = load(Some(&path), &node_id).unwrap();
3236 assert_eq!(reloaded.workspaces, workspaces);
3237 assert_eq!(reloaded.records, loaded.records);
3238 let _ = fs::remove_dir_all(path.parent().unwrap());
3239 }
3240
3241 #[test]
3242 fn v2_state_loads_losslessly_and_next_save_migrates_to_v3() {
3243 let path = temp_path("v2-migration");
3244 let backup = sibling_path(&path, "bak");
3245 let (node_id, workspaces, record) = fixture(&path);
3246 let legacy = persisted_v2_state(
3247 node_id.clone(),
3248 &workspaces,
3249 vec![record.clone()],
3250 );
3251 let legacy_bytes = serde_json::to_vec_pretty(&legacy).unwrap();
3252 fs::create_dir_all(path.parent().unwrap()).unwrap();
3253 fs::write(&path, &legacy_bytes).unwrap();
3254
3255 let loaded = load(Some(&path), &node_id).unwrap();
3256 assert_eq!(loaded.workspaces, workspaces);
3257 assert_eq!(loaded.records[0].provider, agent("claude"));
3258 assert_eq!(loaded.records[0].record_id, record.record_id);
3259 save(Some(&path), &node_id, &loaded.workspaces, &loaded.records).unwrap();
3260
3261 let migrated = fs::read(&path).unwrap();
3262 let header: PersistedNodeStateHeader = serde_json::from_slice(&migrated).unwrap();
3263 assert_eq!(header.version, NODE_STATE_SCHEMA_V3);
3264 assert_eq!(fs::read(&backup).unwrap(), legacy_bytes);
3265 assert_eq!(load(Some(&path), &node_id).unwrap().records, loaded.records);
3266 let _ = fs::remove_dir_all(path.parent().unwrap());
3267 }
3268
3269 #[test]
3270 fn save_rejects_semantic_provider_session_duplicates_before_commit() {
3271 let path = temp_path("duplicate-provider-session-save");
3272 let (node_id, workspaces, record) = fixture(&path);
3273 save(Some(&path), &node_id, &workspaces, &[record.clone()]).unwrap();
3274 let original = fs::read(&path).unwrap();
3275 let mut duplicate = record.clone();
3276 duplicate.record_id = SessionRecordId::new("session-2").unwrap();
3277 duplicate.display_name = "duplicate".to_owned();
3278 duplicate.provider_session.as_mut().unwrap().transcript_path =
3279 Some(r"D:\another\transcript.jsonl".to_owned());
3280 let error = save(
3281 Some(&path),
3282 &node_id,
3283 &workspaces,
3284 &[record, duplicate],
3285 )
3286 .unwrap_err();
3287 assert_eq!(error.kind(), io::ErrorKind::InvalidData);
3288 assert_eq!(fs::read(&path).unwrap(), original);
3289 let loaded = load(Some(&path), &node_id).unwrap();
3290 assert_eq!(loaded.records.len(), 1);
3291 let _ = fs::remove_dir_all(path.parent().unwrap());
3292 }
3293
3294 #[test]
3295 fn load_rejects_semantic_provider_session_duplicates_with_different_paths() {
3296 let path = temp_path("duplicate-provider-session-load");
3297 let (node_id, workspaces, record) = fixture(&path);
3298 let mut duplicate = record.clone();
3299 duplicate.record_id = SessionRecordId::new("session-2").unwrap();
3300 duplicate.display_name = "duplicate".to_owned();
3301 duplicate.provider_session.as_mut().unwrap().transcript_path =
3302 Some(r"D:\another\transcript.jsonl".to_owned());
3303 let legacy = persisted_v1_state(
3304 NODE_STATE_SCHEMA_V1,
3305 node_id.clone(),
3306 &workspaces,
3307 vec![record, duplicate],
3308 );
3309 fs::create_dir_all(path.parent().unwrap()).unwrap();
3310 fs::write(&path, serde_json::to_vec_pretty(&legacy).unwrap()).unwrap();
3311 let error = load(Some(&path), &node_id).unwrap_err();
3312 assert_eq!(error.kind(), io::ErrorKind::InvalidData);
3313 let _ = fs::remove_dir_all(path.parent().unwrap());
3314 }
3315
3316 #[test]
3317 fn durable_state_persists_only_allowlisted_error_summaries() {
3318 let path = temp_path("sanitized-error");
3319 let (node_id, workspaces, mut record) = fixture(&path);
3320 record.last_error = Some(
3321 r"provider failed with bearer-secret-123 at C:\private\session.jsonl".to_owned(),
3322 );
3323 save(Some(&path), &node_id, &workspaces, &[record]).unwrap();
3324 let persisted = fs::read_to_string(&path).unwrap();
3325 assert!(!persisted.contains("bearer-secret-123"));
3326 assert!(!persisted.contains(r"C:\private"));
3327 assert!(persisted.contains("provider-runtime-failed"));
3328 let loaded = load(Some(&path), &node_id).unwrap();
3329 assert_eq!(
3330 loaded.records[0].last_error.as_deref(),
3331 Some("provider-runtime-failed"),
3332 );
3333 let _ = fs::remove_dir_all(path.parent().unwrap());
3334 }
3335
3336 #[test]
3337 fn corrupt_primary_recovers_valid_backup_without_rewriting_corrupt_bytes() {
3338 let path = temp_path("recover");
3339 let (node_id, workspaces, record) = fixture(&path);
3340 save(Some(&path), &node_id, &workspaces, &[record.clone()]).unwrap();
3341 let mut renamed = record;
3342 renamed.display_name = "renamed".to_owned();
3343 save(Some(&path), &node_id, &workspaces, &[renamed]).unwrap();
3344 fs::write(&path, b"{corrupt").unwrap();
3345 let corrupt = fs::read(&path).unwrap();
3346 let loaded = load(Some(&path), &node_id).unwrap();
3347 assert_eq!(loaded.records[0].display_name, "Claude release check");
3348 assert_eq!(
3349 loaded.warning.as_deref(),
3350 Some(DURABLE_STATE_RECOVERY_WARNING),
3351 );
3352 assert!(!loaded.warning.unwrap().contains(&path.to_string_lossy().into_owned()));
3353 assert_eq!(fs::read(&path).unwrap(), corrupt);
3354 let _ = fs::remove_dir_all(path.parent().unwrap());
3355 }
3356
3357 #[test]
3358 fn newer_primary_schema_refuses_without_falling_back_to_valid_v1_backup() {
3359 let path = temp_path("newer-primary-schema");
3360 let backup = sibling_path(&path, "bak");
3361 let (node_id, workspaces, mut record) = fixture(&path);
3362 save(Some(&path), &node_id, &workspaces, &[record.clone()]).unwrap();
3363 record.display_name = "generation-2".to_owned();
3364 record.updated_at_unix_ms += 1;
3365 save(Some(&path), &node_id, &workspaces, &[record.clone()]).unwrap();
3366 load_one(&backup, &node_id).unwrap();
3367
3368 let future = persisted_v1_state(
3369 NODE_STATE_SCHEMA_V10 + 1,
3370 node_id.clone(),
3371 &workspaces,
3372 vec![record.clone()],
3373 );
3374 let future_bytes = serde_json::to_vec_pretty(&future).unwrap();
3375 fs::write(&path, &future_bytes).unwrap();
3376 let backup_bytes = fs::read(&backup).unwrap();
3377
3378 let error = load(Some(&path), &node_id).unwrap_err();
3379 assert_eq!(error.kind(), io::ErrorKind::Unsupported);
3380 assert_eq!(
3381 state_load_refusal(&error),
3382 Some(StateLoadRefusal::UnsupportedSchema),
3383 );
3384 assert_eq!(fs::read(&path).unwrap(), future_bytes);
3385 assert_eq!(fs::read(&backup).unwrap(), backup_bytes);
3386 let save_error = save(Some(&path), &node_id, &workspaces, &[record]).unwrap_err();
3387 assert_eq!(save_error.kind(), io::ErrorKind::Unsupported);
3388 assert_eq!(fs::read(&path).unwrap(), future_bytes);
3389 assert_eq!(fs::read(&backup).unwrap(), backup_bytes);
3390 let _ = fs::remove_dir_all(path.parent().unwrap());
3391 }
3392
3393 #[test]
3394 fn newer_pending_schema_refuses_without_skipping_to_older_backup() {
3395 let path = temp_path("newer-pending-schema");
3396 let backup = sibling_path(&path, "bak");
3397 let pending_backup = sibling_path(&path, "pending-backup");
3398 let (node_id, workspaces, mut record) = fixture(&path);
3399 save(Some(&path), &node_id, &workspaces, &[record.clone()]).unwrap();
3400 record.display_name = "generation-2".to_owned();
3401 record.updated_at_unix_ms += 1;
3402 save(Some(&path), &node_id, &workspaces, &[record.clone()]).unwrap();
3403 load_one(&backup, &node_id).unwrap();
3404 fs::remove_file(&path).unwrap();
3405
3406 let future = persisted_v1_state(
3407 NODE_STATE_SCHEMA_V10 + 1,
3408 node_id.clone(),
3409 &workspaces,
3410 vec![record],
3411 );
3412 let future_bytes = serde_json::to_vec_pretty(&future).unwrap();
3413 fs::write(&pending_backup, &future_bytes).unwrap();
3414 let backup_bytes = fs::read(&backup).unwrap();
3415
3416 let error = load(Some(&path), &node_id).unwrap_err();
3417 assert_eq!(error.kind(), io::ErrorKind::Unsupported);
3418 assert_eq!(
3419 state_load_refusal(&error),
3420 Some(StateLoadRefusal::UnsupportedSchema),
3421 );
3422 assert_eq!(fs::read(&pending_backup).unwrap(), future_bytes);
3423 assert_eq!(fs::read(&backup).unwrap(), backup_bytes);
3424 let _ = fs::remove_dir_all(path.parent().unwrap());
3425 }
3426
3427 #[test]
3428 fn v2_unix_bytes_primary_refuses_without_fallback_or_rewrite() {
3429 let path = temp_path("unix-bytes-primary");
3430 let backup = sibling_path(&path, "bak");
3431 let (node_id, workspaces, mut record) = fixture(&path);
3432 save(Some(&path), &node_id, &workspaces, &[record.clone()]).unwrap();
3433 record.display_name = "generation-2".to_owned();
3434 record.updated_at_unix_ms += 1;
3435 save(Some(&path), &node_id, &workspaces, &[record.clone()]).unwrap();
3436 let foreign = PersistedNodeStateV2 {
3437 version: NODE_STATE_SCHEMA_V2,
3438 node_id: node_id.clone(),
3439 workspaces: vec![PersistedWorkspaceV2 {
3440 workspace_id: WorkspaceId::new("primary").unwrap(),
3441 canonical_root: OpaqueHostPath::unix_bytes(b"/srv/repo".to_vec()).unwrap(),
3442 }],
3443 session_records: Vec::new(),
3444 };
3445 let foreign_bytes = serde_json::to_vec_pretty(&foreign).unwrap();
3446 fs::write(&path, &foreign_bytes).unwrap();
3447 let backup_bytes = fs::read(&backup).unwrap();
3448
3449 let error = load(Some(&path), &node_id).unwrap_err();
3450 assert_eq!(
3451 state_load_refusal(&error),
3452 Some(StateLoadRefusal::PathSemanticsUnsupported),
3453 );
3454 let save_error = save(Some(&path), &node_id, &workspaces, &[record]).unwrap_err();
3455 assert_eq!(
3456 state_load_refusal(&save_error),
3457 Some(StateLoadRefusal::PathSemanticsUnsupported),
3458 );
3459 assert_eq!(fs::read(&path).unwrap(), foreign_bytes);
3460 assert_eq!(fs::read(&backup).unwrap(), backup_bytes);
3461 let _ = fs::remove_dir_all(path.parent().unwrap());
3462 }
3463
3464 #[test]
3465 fn v2_unix_bytes_pending_refuses_without_skipping_older_backup() {
3466 let path = temp_path("unix-bytes-pending");
3467 let backup = sibling_path(&path, "bak");
3468 let pending_backup = sibling_path(&path, "pending-backup");
3469 let (node_id, workspaces, mut record) = fixture(&path);
3470 save(Some(&path), &node_id, &workspaces, &[record.clone()]).unwrap();
3471 record.display_name = "generation-2".to_owned();
3472 record.updated_at_unix_ms += 1;
3473 save(Some(&path), &node_id, &workspaces, &[record]).unwrap();
3474 fs::remove_file(&path).unwrap();
3475 let foreign = PersistedNodeStateV2 {
3476 version: NODE_STATE_SCHEMA_V2,
3477 node_id: node_id.clone(),
3478 workspaces: vec![PersistedWorkspaceV2 {
3479 workspace_id: WorkspaceId::new("primary").unwrap(),
3480 canonical_root: OpaqueHostPath::unix_bytes(b"/srv/repo".to_vec()).unwrap(),
3481 }],
3482 session_records: Vec::new(),
3483 };
3484 let foreign_bytes = serde_json::to_vec_pretty(&foreign).unwrap();
3485 fs::write(&pending_backup, &foreign_bytes).unwrap();
3486 let backup_bytes = fs::read(&backup).unwrap();
3487
3488 let error = load(Some(&path), &node_id).unwrap_err();
3489 assert_eq!(
3490 state_load_refusal(&error),
3491 Some(StateLoadRefusal::PathSemanticsUnsupported),
3492 );
3493 assert_eq!(fs::read(&pending_backup).unwrap(), foreign_bytes);
3494 assert_eq!(fs::read(&backup).unwrap(), backup_bytes);
3495 let _ = fs::remove_dir_all(path.parent().unwrap());
3496 }
3497
3498 #[test]
3499 fn wrong_node_primary_refuses_without_falling_back_or_rewriting() {
3500 let path = temp_path("wrong-node-primary");
3501 let backup = sibling_path(&path, "bak");
3502 let (node_id, workspaces, mut record) = fixture(&path);
3503 save(Some(&path), &node_id, &workspaces, &[record.clone()]).unwrap();
3504 record.display_name = "generation-2".to_owned();
3505 record.updated_at_unix_ms += 1;
3506 save(Some(&path), &node_id, &workspaces, &[record.clone()]).unwrap();
3507 load_one(&backup, &node_id).unwrap();
3508
3509 let wrong_node = persisted_v1_state(
3510 NODE_STATE_SCHEMA_V1,
3511 NodeId::new("node-b").unwrap(),
3512 &workspaces,
3513 vec![record.clone()],
3514 );
3515 let wrong_node_bytes = serde_json::to_vec_pretty(&wrong_node).unwrap();
3516 fs::write(&path, &wrong_node_bytes).unwrap();
3517 let backup_bytes = fs::read(&backup).unwrap();
3518
3519 let error = load(Some(&path), &node_id).unwrap_err();
3520 assert_eq!(error.kind(), io::ErrorKind::PermissionDenied);
3521 assert_eq!(
3522 state_load_refusal(&error),
3523 Some(StateLoadRefusal::NodeIdentityMismatch),
3524 );
3525 let save_error = save(Some(&path), &node_id, &workspaces, &[record]).unwrap_err();
3526 assert_eq!(save_error.kind(), io::ErrorKind::PermissionDenied);
3527 assert_eq!(fs::read(&path).unwrap(), wrong_node_bytes);
3528 assert_eq!(fs::read(&backup).unwrap(), backup_bytes);
3529 let _ = fs::remove_dir_all(path.parent().unwrap());
3530 }
3531
3532 #[test]
3533 fn wrong_node_pending_refuses_without_skipping_to_older_backup() {
3534 let path = temp_path("wrong-node-pending");
3535 let backup = sibling_path(&path, "bak");
3536 let pending_backup = sibling_path(&path, "pending-backup");
3537 let (node_id, workspaces, mut record) = fixture(&path);
3538 save(Some(&path), &node_id, &workspaces, &[record.clone()]).unwrap();
3539 record.display_name = "generation-2".to_owned();
3540 record.updated_at_unix_ms += 1;
3541 save(Some(&path), &node_id, &workspaces, &[record.clone()]).unwrap();
3542 load_one(&backup, &node_id).unwrap();
3543 fs::remove_file(&path).unwrap();
3544
3545 let wrong_node = persisted_v1_state(
3546 NODE_STATE_SCHEMA_V1,
3547 NodeId::new("node-b").unwrap(),
3548 &workspaces,
3549 vec![record],
3550 );
3551 let wrong_node_bytes = serde_json::to_vec_pretty(&wrong_node).unwrap();
3552 fs::write(&pending_backup, &wrong_node_bytes).unwrap();
3553 let backup_bytes = fs::read(&backup).unwrap();
3554
3555 let error = load(Some(&path), &node_id).unwrap_err();
3556 assert_eq!(error.kind(), io::ErrorKind::PermissionDenied);
3557 assert_eq!(
3558 state_load_refusal(&error),
3559 Some(StateLoadRefusal::NodeIdentityMismatch),
3560 );
3561 assert_eq!(fs::read(&pending_backup).unwrap(), wrong_node_bytes);
3562 assert_eq!(fs::read(&backup).unwrap(), backup_bytes);
3563 let _ = fs::remove_dir_all(path.parent().unwrap());
3564 }
3565
3566 #[test]
3567 fn authoritative_refusals_are_typed_not_inferred_from_io_kinds() {
3568 let generic_permission = io::Error::new(io::ErrorKind::PermissionDenied, "access denied");
3569 let generic_unsupported = io::Error::new(io::ErrorKind::Unsupported, "operation unsupported");
3570 assert_eq!(state_load_refusal(&generic_permission), None);
3571 assert_eq!(state_load_refusal(&generic_unsupported), None);
3572
3573 let schema = io::Error::new(
3574 io::ErrorKind::Unsupported,
3575 StateLoadRefusalError::UnsupportedSchema(NODE_STATE_SCHEMA_V3 + 1),
3576 );
3577 let node = io::Error::new(
3578 io::ErrorKind::PermissionDenied,
3579 StateLoadRefusalError::NodeIdentityMismatch,
3580 );
3581 let path = io::Error::new(
3582 io::ErrorKind::Unsupported,
3583 StateLoadRefusalError::PathSemanticsUnsupported,
3584 );
3585 assert_eq!(
3586 state_load_refusal(&schema),
3587 Some(StateLoadRefusal::UnsupportedSchema),
3588 );
3589 assert_eq!(
3590 state_load_refusal(&node),
3591 Some(StateLoadRefusal::NodeIdentityMismatch),
3592 );
3593 assert_eq!(
3594 state_load_refusal(&path),
3595 Some(StateLoadRefusal::PathSemanticsUnsupported),
3596 );
3597 }
3598
3599 #[test]
3600 fn backup_tracks_the_immediately_previous_generation_across_four_commits() {
3601 let path = temp_path("backup-generations");
3602 let backup = sibling_path(&path, "bak");
3603 let (node_id, workspaces, mut record) = fixture(&path);
3604 for generation in 1..=4 {
3605 record.display_name = format!("generation-{generation}");
3606 record.updated_at_unix_ms = 10 + generation;
3607 let warning = save(
3608 Some(&path),
3609 &node_id,
3610 &workspaces,
3611 &[record.clone()],
3612 )
3613 .unwrap();
3614 assert_eq!(warning, None);
3615 if generation > 1 {
3616 let loaded_backup = load_one(&backup, &node_id).unwrap();
3617 assert_eq!(
3618 loaded_backup.records[0].display_name,
3619 format!("generation-{}", generation - 1),
3620 );
3621 }
3622 }
3623
3624 fs::write(&path, b"{corrupt").unwrap();
3625 let recovered = load(Some(&path), &node_id).unwrap();
3626 assert_eq!(recovered.records[0].display_name, "generation-3");
3627 let _ = fs::remove_dir_all(path.parent().unwrap());
3628 }
3629
3630 #[test]
3631 fn failed_backup_rotation_preserves_old_backup_and_pending_previous_primary() {
3632 let path = temp_path("backup-rotation-failure");
3633 let backup = sibling_path(&path, "bak");
3634 let pending_backup = sibling_path(&path, "pending-backup");
3635 let (node_id, workspaces, mut record) = fixture(&path);
3636 record.display_name = "generation-1".to_owned();
3637 save(Some(&path), &node_id, &workspaces, &[record.clone()]).unwrap();
3638 record.display_name = "generation-2".to_owned();
3639 record.updated_at_unix_ms += 1;
3640 save(Some(&path), &node_id, &workspaces, &[record]).unwrap();
3641 let old_backup = fs::read(&backup).unwrap();
3642 let previous_primary = fs::read(&path).unwrap();
3643
3644 let warning = atomic_write_with_fault(
3645 &path,
3646 &previous_primary,
3647 AtomicWriteFault::BeforeBackupRotation,
3648 )
3649 .unwrap();
3650 assert_eq!(warning.as_deref(), Some(BACKUP_ROTATION_WARNING));
3651 assert_eq!(fs::read(&backup).unwrap(), old_backup);
3652 assert_eq!(fs::read(&pending_backup).unwrap(), previous_primary);
3653 fs::write(&path, b"{corrupt").unwrap();
3654 let recovered = load(Some(&path), &node_id).unwrap();
3655 assert_eq!(recovered.records[0].display_name, "generation-2");
3656 let _ = fs::remove_dir_all(path.parent().unwrap());
3657 }
3658
3659 #[test]
3660 fn failed_recovery_commit_preserves_the_valid_backup() {
3661 let path = temp_path("recover-failed-commit");
3662 let backup = sibling_path(&path, "bak");
3663 let (node_id, workspaces, record) = fixture(&path);
3664 save(Some(&path), &node_id, &workspaces, &[record.clone()]).unwrap();
3665 let mut renamed = record;
3666 renamed.display_name = "renamed".to_owned();
3667 save(Some(&path), &node_id, &workspaces, &[renamed]).unwrap();
3668 fs::write(&path, b"{corrupt").unwrap();
3669 let valid_backup = fs::read(&backup).unwrap();
3670 let corrupt_primary = fs::read(&path).unwrap();
3671
3672 let error = atomic_write_with_policy(
3673 &path,
3674 br#"{"new":"state"}"#,
3675 AtomicWriteFault::BeforePrimaryCommit,
3676 false,
3677 )
3678 .unwrap_err();
3679 assert_eq!(error.kind(), io::ErrorKind::Other);
3680 assert_eq!(fs::read(&backup).unwrap(), valid_backup);
3681 assert_eq!(fs::read(&path).unwrap(), corrupt_primary);
3682 let loaded = load(Some(&path), &node_id).unwrap();
3683 assert_eq!(loaded.records[0].display_name, "Claude release check");
3684 let _ = fs::remove_dir_all(path.parent().unwrap());
3685 }
3686
3687 #[test]
3688 fn state_path_lock_rejects_a_second_owner_until_drop() {
3689 let path = temp_path("exclusive-owner");
3690 let first = StatePathLock::acquire(Some(&path)).unwrap().unwrap();
3691 let error = match StatePathLock::acquire(Some(&path)) {
3692 Ok(_) => panic!("a second durable state owner was accepted"),
3693 Err(error) => error,
3694 };
3695 assert_eq!(error.kind(), io::ErrorKind::WouldBlock);
3696 drop(first);
3697 let second = StatePathLock::acquire(Some(&path)).unwrap().unwrap();
3698 drop(second);
3699 let _ = fs::remove_dir_all(path.parent().unwrap());
3700 }
3701
3702 #[test]
3703 fn corrupt_only_copy_fails_and_is_preserved() {
3704 let path = temp_path("corrupt-only");
3705 fs::create_dir_all(path.parent().unwrap()).unwrap();
3706 fs::write(&path, b"not-json").unwrap();
3707 let error = load(Some(&path), &NodeId::new("node-a").unwrap()).unwrap_err();
3708 assert_eq!(error.kind(), io::ErrorKind::InvalidData);
3709 assert_eq!(fs::read(&path).unwrap(), b"not-json");
3710 let _ = fs::remove_dir_all(path.parent().unwrap());
3711 }
3712
3713 #[test]
3714 fn v6_to_v7_migration_preserves_environment_receipt_and_adds_no_bundle() {
3715 let path = temp_path("v6-to-v7-bundle-migration");
3716 fs::create_dir_all(path.parent().unwrap()).unwrap();
3717 let (node_id, workspaces, mut record) = fixture(&path);
3718 let environment_profile = ResolvedEnvironmentProfileReceipt {
3719 profile_id: crate::protocol::SpawnEnvironmentProfileId::new("local-claude").unwrap(),
3720 profile_revision: crate::protocol::SpawnEnvironmentProfileRevision::new("r1").unwrap(),
3721 network_allowlist: None,
3722 browser_profile_id: None,
3723 };
3724 record.environment_profile = Some(environment_profile.clone());
3725 let ownership = materialization_ownership(
3726 &path,
3727 "legacy-environment",
3728 MaterializationOwner::Record {
3729 record_id: record.record_id.clone(),
3730 },
3731 None,
3732 );
3733 let state = PersistedNodeStateV6 {
3734 version: NODE_STATE_SCHEMA_V6,
3735 node_id: node_id.clone(),
3736 workspaces: workspaces.iter().map(|(workspace_id, root)| PersistedWorkspaceV2 {
3737 workspace_id: workspace_id.clone(),
3738 canonical_root: OpaqueHostPath::utf8(root.clone()).unwrap(),
3739 }).collect(),
3740 session_records: vec![persisted_v5_record(record)],
3741 managed_worktrees: Vec::new(),
3742 managed_worktree_tombstones: Vec::new(),
3743 materializations: vec![persisted_materialization(&ownership)],
3744 };
3745 fs::write(&path, serde_json::to_vec(&state).unwrap()).unwrap();
3746 let loaded = load(Some(&path), &node_id).unwrap();
3747 assert_eq!(loaded.records[0].environment_profile.as_ref(), Some(&environment_profile));
3748 assert!(loaded.records[0].bundle.is_none());
3749 assert!(loaded.materializations[0].bundle().is_none());
3750 save_v7(
3751 Some(&path),
3752 &node_id,
3753 &workspaces,
3754 &loaded.records,
3755 &[],
3756 &[],
3757 &loaded.materializations,
3758 )
3759 .unwrap();
3760 let header: PersistedNodeStateHeader =
3761 serde_json::from_slice(&fs::read(&path).unwrap()).unwrap();
3762 assert_eq!(header.version, NODE_STATE_SCHEMA_V7);
3763 let _ = fs::remove_dir_all(path.parent().unwrap());
3764 }
3765
3766 #[test]
3767 fn v7_roundtrip_preserves_opaque_bundle_and_rejects_receipt_mismatch() {
3768 let path = temp_path("v7-bundle-roundtrip");
3769 fs::create_dir_all(path.parent().unwrap()).unwrap();
3770 let (node_id, workspaces, mut record) = fixture(&path);
3771 let environment_profile = ResolvedEnvironmentProfileReceipt {
3772 profile_id: crate::protocol::SpawnEnvironmentProfileId::new("local-claude").unwrap(),
3773 profile_revision: crate::protocol::SpawnEnvironmentProfileRevision::new("r1").unwrap(),
3774 network_allowlist: None,
3775 browser_profile_id: None,
3776 };
3777 let bundle = ResolvedBundleReceipt {
3778 id: crate::protocol::SpawnBundleId::new("review-bundle").unwrap(),
3779 revision: crate::protocol::SpawnBundleRevision::new("r1").unwrap(),
3780 digest: crate::protocol::SpawnBundleDigest::new(format!("sha256:{}", "0".repeat(64))).unwrap(),
3781 };
3782 record.environment_profile = Some(environment_profile.clone());
3783 record.bundle = Some(bundle.clone());
3784 let root = path.parent().unwrap().join("session-environments").join("bundle-record");
3785 let ownership = MaterializationOwnershipRecord::from_persisted(
3786 MaterializationId::new("bundle-record").unwrap(),
3787 Some(environment_profile),
3788 Some(bundle.clone()),
3789 None,
3790 MaterializationOwner::Record { record_id: record.record_id.clone() },
3791 None,
3792 MaterializationState::Ready,
3793 root.clone(),
3794 root.join("home"),
3795 root.join("bundle"),
3796 root.join("plugin-data"),
3797 Vec::new(),
3798 10,
3799 11,
3800 )
3801 .unwrap();
3802 save_v7(
3803 Some(&path),
3804 &node_id,
3805 &workspaces,
3806 &[record],
3807 &[],
3808 &[],
3809 &[ownership],
3810 )
3811 .unwrap();
3812 let loaded = load(Some(&path), &node_id).unwrap();
3813 assert_eq!(loaded.records[0].bundle.as_ref(), Some(&bundle));
3814
3815 let mut value: serde_json::Value = serde_json::from_slice(&fs::read(&path).unwrap()).unwrap();
3816 value["session_records"][0]["bundle"]["digest"] =
3817 serde_json::Value::String(format!("sha256:{}", "1".repeat(64)));
3818 fs::write(&path, serde_json::to_vec(&value).unwrap()).unwrap();
3819 assert_eq!(load(Some(&path), &node_id).unwrap_err().kind(), io::ErrorKind::InvalidData);
3820 let _ = fs::remove_dir_all(path.parent().unwrap());
3821 }
3822
3823 #[test]
3824 fn v7_to_v8_migration_defaults_context_metadata_to_none() {
3825 let path = temp_path("v7-to-v8-context-migration");
3826 fs::create_dir_all(path.parent().unwrap()).unwrap();
3827 let (node_id, workspaces, record) = fixture(&path);
3828 save_v7(
3829 Some(&path),
3830 &node_id,
3831 &workspaces,
3832 &[record],
3833 &[],
3834 &[],
3835 &[],
3836 )
3837 .unwrap();
3838
3839 let loaded_v7 = load(Some(&path), &node_id).unwrap();
3840 assert!(loaded_v7.records[0].context_id.is_none());
3841 assert!(loaded_v7.records[0].context.is_none());
3842 assert!(loaded_v7.materializations.is_empty());
3843 save_v8(
3844 Some(&path),
3845 &node_id,
3846 &workspaces,
3847 &loaded_v7.records,
3848 &[],
3849 &[],
3850 &loaded_v7.materializations,
3851 )
3852 .unwrap();
3853
3854 let bytes = fs::read(&path).unwrap();
3855 let header: PersistedNodeStateHeader = serde_json::from_slice(&bytes).unwrap();
3856 assert_eq!(header.version, NODE_STATE_SCHEMA_V8);
3857 let loaded_v8 = load(Some(&path), &node_id).unwrap();
3858 assert!(loaded_v8.records[0].context_id.is_none());
3859 assert!(loaded_v8.records[0].context.is_none());
3860 let _ = fs::remove_dir_all(path.parent().unwrap());
3861 }
3862
3863 #[test]
3864 fn v8_roundtrip_preserves_only_opaque_context_metadata() {
3865 let path = temp_path("v8-context-roundtrip");
3866 fs::create_dir_all(path.parent().unwrap()).unwrap();
3867 let (node_id, workspaces, mut record) = fixture(&path);
3868 let context = context_receipt("ctx-context-roundtrip", 'a');
3869 record.context_id = Some(context.id.clone());
3870 record.context = Some(context.clone());
3871 let ownership = context_materialization(
3872 &path,
3873 "context-record",
3874 record.record_id.clone(),
3875 context.clone(),
3876 );
3877
3878 save_v8(
3879 Some(&path),
3880 &node_id,
3881 &workspaces,
3882 &[record],
3883 &[],
3884 &[],
3885 &[ownership.clone()],
3886 )
3887 .unwrap();
3888 let bytes = fs::read(&path).unwrap();
3889 let serialized: serde_json::Value = serde_json::from_slice(&bytes).unwrap();
3890 assert_eq!(serialized["version"], NODE_STATE_SCHEMA_V8);
3891 assert_eq!(
3892 serialized["session_records"][0]["context"]["id"],
3893 "ctx-context-roundtrip",
3894 );
3895 assert!(serialized["session_records"][0]["context"].get("messages").is_none());
3896 assert!(serialized["materializations"][0]["context"].get("messages").is_none());
3897 assert!(!String::from_utf8(bytes).unwrap().contains("bounded-history-message-body"));
3898
3899 let loaded = load(Some(&path), &node_id).unwrap();
3900 assert_eq!(loaded.records[0].context_id.as_ref(), Some(&context.id));
3901 assert_eq!(loaded.records[0].context.as_ref(), Some(&context));
3902 assert!(loaded.materializations == vec![ownership]);
3903 let _ = fs::remove_dir_all(path.parent().unwrap());
3904 }
3905
3906 #[test]
3907 fn v9_roundtrip_loads_with_an_empty_managed_spawn_replay_ledger() {
3908 let path = temp_path("v9-managed-spawn-migration");
3909 fs::create_dir_all(path.parent().unwrap()).unwrap();
3910 let (node_id, workspaces, record) = fixture(&path);
3911
3912 save_v9(
3913 Some(&path),
3914 &node_id,
3915 &workspaces,
3916 &[record],
3917 &[],
3918 &[],
3919 &[],
3920 )
3921 .unwrap();
3922
3923 let bytes = fs::read(&path).unwrap();
3924 let header: PersistedNodeStateHeader = serde_json::from_slice(&bytes).unwrap();
3925 assert_eq!(header.version, NODE_STATE_SCHEMA_V9);
3926 let loaded = load(Some(&path), &node_id).unwrap();
3927 assert!(loaded.managed_spawn_replays.is_empty());
3928 let _ = fs::remove_dir_all(path.parent().unwrap());
3929 }
3930
3931 #[test]
3932 fn v8_save_rejects_context_id_and_receipt_mismatch() {
3933 let path = temp_path("v8-context-id-mismatch");
3934 fs::create_dir_all(path.parent().unwrap()).unwrap();
3935 let (node_id, workspaces, mut record) = fixture(&path);
3936 let context = context_receipt("ctx-receipt", 'b');
3937 record.context_id = Some(crate::protocol::SpawnContextId::new("ctx-other").unwrap());
3938 record.context = Some(context.clone());
3939 let ownership = context_materialization(
3940 &path,
3941 "context-mismatch",
3942 record.record_id.clone(),
3943 context,
3944 );
3945
3946 let error = save_v8(
3947 Some(&path),
3948 &node_id,
3949 &workspaces,
3950 &[record],
3951 &[],
3952 &[],
3953 &[ownership],
3954 )
3955 .unwrap_err();
3956 assert_eq!(error.kind(), io::ErrorKind::InvalidData);
3957 assert!(!path.exists());
3958 let _ = fs::remove_dir_all(path.parent().unwrap());
3959 }
3960
3961 #[test]
3962 fn v8_save_rejects_mismatched_context_materialization_reference() {
3963 let path = temp_path("v8-context-materialization-mismatch");
3964 fs::create_dir_all(path.parent().unwrap()).unwrap();
3965 let (node_id, workspaces, mut record) = fixture(&path);
3966 let record_context = context_receipt("ctx-record", 'd');
3967 record.context_id = Some(record_context.id.clone());
3968 record.context = Some(record_context);
3969 let ownership = context_materialization(
3970 &path,
3971 "context-foreign",
3972 record.record_id.clone(),
3973 context_receipt("ctx-foreign", 'e'),
3974 );
3975
3976 let error = save_v8(
3977 Some(&path),
3978 &node_id,
3979 &workspaces,
3980 &[record],
3981 &[],
3982 &[],
3983 &[ownership],
3984 )
3985 .unwrap_err();
3986 assert_eq!(error.kind(), io::ErrorKind::InvalidData);
3987 assert!(!path.exists());
3988 let _ = fs::remove_dir_all(path.parent().unwrap());
3989 }
3990
3991 #[test]
3992 fn v8_load_rejects_invalid_context_receipt() {
3993 let path = temp_path("v8-context-invalid-receipt");
3994 fs::create_dir_all(path.parent().unwrap()).unwrap();
3995 let (node_id, workspaces, mut record) = fixture(&path);
3996 let context = context_receipt("ctx-valid", 'c');
3997 record.context_id = Some(context.id.clone());
3998 record.context = Some(context.clone());
3999 let ownership = context_materialization(
4000 &path,
4001 "context-invalid",
4002 record.record_id.clone(),
4003 context,
4004 );
4005 save_v8(
4006 Some(&path),
4007 &node_id,
4008 &workspaces,
4009 &[record],
4010 &[],
4011 &[],
4012 &[ownership],
4013 )
4014 .unwrap();
4015
4016 let mut invalid: serde_json::Value = serde_json::from_slice(&fs::read(&path).unwrap()).unwrap();
4017 invalid["materializations"][0]["context"]["byte_len"] = serde_json::Value::from(0);
4018 fs::write(&path, serde_json::to_vec(&invalid).unwrap()).unwrap();
4019 assert_eq!(
4020 load(Some(&path), &node_id).unwrap_err().kind(),
4021 io::ErrorKind::InvalidData,
4022 );
4023 let _ = fs::remove_dir_all(path.parent().unwrap());
4024 }
4025}