1use std::collections::BTreeMap;
44use std::fs;
45use std::path::{Path, PathBuf};
46
47use agentstategraph::{CommitOptions, Repository};
48use agentstategraph_core::IntentCategory;
49use serde_json::Value;
50
51use crate::error::{AsdError, Result};
52use crate::paths;
53use crate::repair::drop_orphaned_edge_refs;
54use crate::schema::{ASD_SCHEMA_VERSION, EffectDecl, LedgerEntry, Rebind, Symbol};
55use crate::search_fts::SearchFtsDb;
56
57const SIDECAR_REL_ROOT: &str = ".asd/v1";
59
60#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
65#[serde(rename_all = "kebab-case")]
66pub enum SidecarState {
67 Missing,
69 Present,
71 Hydrated,
73 FreshReset,
75}
76
77pub fn outdated_hooks(root: &Path) -> Option<String> {
85 let hooks = root.join(".asd/hooks");
86 let stale: Vec<&str> = ["post-checkout", "post-merge"]
87 .into_iter()
88 .filter(|name| {
89 fs::read_to_string(hooks.join(name)).is_ok_and(|script| {
90 script
91 .lines()
92 .any(|l| !l.trim_start().starts_with('#') && l.contains("asd hydrate"))
93 })
94 })
95 .collect();
96 (!stale.is_empty()).then(|| {
97 format!(
98 "this project's {} hook{} still run{} `asd hydrate`, reloading the whole ledger \
99 from .asd/v1 on every checkout — run `asd init` to install the current hooks \
100 (they import committed conclusions instead), and commit .asd/hooks",
101 stale.join(" and "),
102 if stale.len() == 1 { "" } else { "s" },
103 if stale.len() == 1 { "s" } else { "" },
104 )
105 })
106}
107
108pub fn sidecar_lifecycle_state(dir: &Path) -> SidecarState {
112 let root = dir.join(SIDECAR_REL_ROOT);
113 if !root.exists() {
114 return SidecarState::Missing;
115 }
116 let meta = root.join("meta");
117 if meta.join("fresh-reset").exists() {
118 return SidecarState::FreshReset;
119 }
120 if meta.join("hydrated-at").exists() {
121 return SidecarState::Hydrated;
122 }
123 SidecarState::Present
124}
125
126pub fn mark_fresh_reset(dir: &Path) -> Result<()> {
129 let sentinel = dir.join(SIDECAR_REL_ROOT).join("meta").join("fresh-reset");
130 if let Some(parent) = sentinel.parent() {
131 fs::create_dir_all(parent)?;
132 }
133 write_text_atomic(&sentinel, &format!("{}\n", chrono::Utc::now().to_rfc3339()))
134}
135
136#[derive(Debug, Clone)]
139pub struct SyncSummary {
140 pub effects_written: usize,
141 pub ledger_entries_written: usize,
142 pub symbols_written: usize,
143 pub rebinds_synced: usize,
144 pub schema_version: String,
145 pub pruned: usize,
147 pub moved_entries_removed: usize,
150}
151
152#[derive(Debug, Clone)]
156pub struct HydrateSummary {
157 pub effects_loaded: usize,
158 pub ledger_entries_loaded: usize,
159 pub symbols_loaded: usize,
160 pub ledger_entries_skipped: usize,
164 pub rebinds_replayed: usize,
165 pub missing_schema_version: bool,
166 pub symbols_skipped: usize,
171 pub blobs_rejected: usize,
175 pub refs_dropped: usize,
178}
179
180pub fn sync_to_dir(repo: &Repository, ref_name: &str, dir: &Path) -> Result<SyncSummary> {
191 let root = dir.join(SIDECAR_REL_ROOT);
192 let effects_dir = root.join("effects");
193 let ledger_dir = root.join("ledger");
194 let symbols_dir = root.join("symbols");
195 let rebinds_dir = root.join("rebinds");
196 let meta_dir = root.join("meta");
197
198 fs::create_dir_all(&effects_dir)?;
199 fs::create_dir_all(&ledger_dir)?;
200 fs::create_dir_all(&symbols_dir)?;
201 fs::create_dir_all(&rebinds_dir)?;
202 fs::create_dir_all(&meta_dir)?;
203
204 let mut effects_written = 0usize;
206 let effects_prefix = format!("{}/effects", paths::ASD_ROOT);
207 if let Ok(serde_json::Value::Object(map)) = repo.get_tree(ref_name, &effects_prefix) {
208 let sorted: BTreeMap<_, _> = map.into_iter().collect();
210 for (symbol_id, value) in sorted {
211 let decl: EffectDecl = serde_json::from_value(value)?;
214 let out = effects_dir.join(format!("{symbol_id}.json"));
215 write_json_atomic(&out, &decl)?;
216 effects_written += 1;
217 }
218 }
219
220 let mut ledger_entries_written = 0usize;
222 let mut moved_entries_removed = 0usize;
223 let ledger_prefix = format!("{}/ledger", paths::ASD_ROOT);
224 if let Ok(serde_json::Value::Object(by_symbol)) = repo.get_tree(ref_name, &ledger_prefix) {
225 moved_entries_removed = remove_moved_entries(&ledger_dir, &by_symbol)?;
226 let sorted_syms: BTreeMap<_, _> = by_symbol.into_iter().collect();
227 for (symbol_id, bucket) in sorted_syms {
228 let serde_json::Value::Object(entries) = bucket else {
229 continue;
230 };
231 if entries.is_empty() {
232 continue;
233 }
234 let sym_dir = ledger_dir.join(&symbol_id);
235 fs::create_dir_all(&sym_dir)?;
236 let sorted_entries: BTreeMap<_, _> = entries.into_iter().collect();
237 for (entry_id, entry_val) in sorted_entries {
238 let entry: LedgerEntry = serde_json::from_value(entry_val)?;
239 let out = sym_dir.join(format!("{entry_id}.json"));
240 write_json_atomic(&out, &entry)?;
241 ledger_entries_written += 1;
242 }
243 }
244 }
245
246 let mut symbols_written = 0usize;
251 let qname_prefix = format!("{}/index/by-qname", paths::ASD_ROOT);
252 if let Ok(serde_json::Value::Object(map)) = repo.get_tree(ref_name, &qname_prefix) {
253 let sorted: BTreeMap<_, _> = map.into_iter().collect();
254 for (qname, value) in sorted {
255 let sym: Symbol = serde_json::from_value(value)?;
256 let out = symbols_dir.join(format!("{qname}.json"));
257 write_json_atomic(&out, &sym)?;
258 symbols_written += 1;
259 }
260 }
261
262 let mut rebinds_synced = 0usize;
264 let rebinds_prefix = format!("{}/rebinds", paths::ASD_ROOT);
265 if let Ok(serde_json::Value::Object(map)) = repo.get_tree(ref_name, &rebinds_prefix) {
266 let sorted: BTreeMap<_, _> = map.into_iter().collect();
267 for (from_symbol_id, value) in sorted {
268 let rebind: Rebind = serde_json::from_value(value)?;
269 let out = rebinds_dir.join(format!("{from_symbol_id}.json"));
270 write_json_atomic(&out, &rebind)?;
271 rebinds_synced += 1;
272 }
273 }
274
275 let sv_path = meta_dir.join("schema-version");
277 write_text_atomic(&sv_path, &format!("{ASD_SCHEMA_VERSION}\n"))?;
278
279 Ok(SyncSummary {
280 effects_written,
281 ledger_entries_written,
282 symbols_written,
283 rebinds_synced,
284 schema_version: ASD_SCHEMA_VERSION.to_string(),
285 pruned: 0,
286 moved_entries_removed,
287 })
288}
289
290fn remove_moved_entries(
296 ledger_dir: &Path,
297 by_symbol: &serde_json::Map<String, Value>,
298) -> Result<usize> {
299 if !ledger_dir.is_dir() {
300 return Ok(0);
301 }
302 let stored = crate::ledger_dupes::locations(by_symbol);
303 let mut removed = 0usize;
304 for sym_entry in fs::read_dir(ledger_dir)? {
305 let sym_dir = sym_entry?.path();
306 if !sym_dir.is_dir() {
307 continue;
308 }
309 let Some(symbol_id) = sym_dir
310 .file_name()
311 .and_then(|n| n.to_str())
312 .map(str::to_string)
313 else {
314 continue;
315 };
316 for file_entry in fs::read_dir(&sym_dir)? {
317 let path = file_entry?.path();
318 if !is_json_file(&path) {
319 continue;
320 }
321 let Some(entry_id) = path.file_stem().and_then(|n| n.to_str()) else {
322 continue;
323 };
324 if stored
325 .get(entry_id)
326 .is_some_and(|symbols| !symbols.contains(&symbol_id))
327 {
328 fs::remove_file(&path)?;
329 removed += 1;
330 }
331 }
332 if fs::read_dir(&sym_dir)?.next().is_none() {
333 fs::remove_dir(&sym_dir)?;
334 }
335 }
336 Ok(removed)
337}
338
339pub fn prune_sidecar(repo: &Repository, ref_name: &str, dir: &Path) -> Result<usize> {
345 let root = dir.join(SIDECAR_REL_ROOT);
346 if !root.exists() {
347 return Ok(0);
348 }
349
350 let mut pruned = 0usize;
351
352 let live_symbol_ids: std::collections::HashSet<String> = {
354 let prefix = format!("{}/effects", paths::ASD_ROOT);
355 match repo.get_tree(ref_name, &prefix) {
356 Ok(serde_json::Value::Object(map)) => map.into_iter().map(|(k, _)| k).collect(),
357 _ => std::collections::HashSet::new(),
358 }
359 };
360
361 let live_qnames: std::collections::HashSet<String> = {
362 let prefix = format!("{}/index/by-qname", paths::ASD_ROOT);
363 match repo.get_tree(ref_name, &prefix) {
364 Ok(serde_json::Value::Object(map)) => map.into_iter().map(|(k, _)| k).collect(),
365 _ => std::collections::HashSet::new(),
366 }
367 };
368
369 let live_ledger_symbol_ids: std::collections::HashSet<String> = {
370 let prefix = format!("{}/ledger", paths::ASD_ROOT);
371 match repo.get_tree(ref_name, &prefix) {
372 Ok(serde_json::Value::Object(map)) => map.into_iter().map(|(k, _)| k).collect(),
373 _ => std::collections::HashSet::new(),
374 }
375 };
376
377 let live_rebind_ids: std::collections::HashSet<String> = {
378 let prefix = format!("{}/rebinds", paths::ASD_ROOT);
379 match repo.get_tree(ref_name, &prefix) {
380 Ok(serde_json::Value::Object(map)) => map.into_iter().map(|(k, _)| k).collect(),
381 _ => std::collections::HashSet::new(),
382 }
383 };
384
385 pruned += prune_flat_dir(&root.join("effects"), &live_symbol_ids)?;
387
388 pruned += prune_flat_dir(&root.join("symbols"), &live_qnames)?;
390
391 pruned += prune_flat_dir(&root.join("rebinds"), &live_rebind_ids)?;
393
394 let ledger_dir = root.join("ledger");
398 if ledger_dir.is_dir() {
399 for entry in fs::read_dir(&ledger_dir)? {
400 let entry = entry?;
401 let sym_dir = entry.path();
402 if !sym_dir.is_dir() {
403 continue;
404 }
405 let sym_id = sym_dir
406 .file_name()
407 .and_then(|n| n.to_str())
408 .unwrap_or("")
409 .to_string();
410
411 if !live_ledger_symbol_ids.contains(&sym_id) {
412 fs::remove_dir_all(&sym_dir)?;
414 pruned += 1;
415 } else {
416 let live_entries: std::collections::HashSet<String> = {
418 let prefix = format!("{}/ledger/{}", paths::ASD_ROOT, sym_id);
419 match repo.get_tree(ref_name, &prefix) {
420 Ok(serde_json::Value::Object(m)) => m.into_iter().map(|(k, _)| k).collect(),
421 _ => std::collections::HashSet::new(),
422 }
423 };
424 for file_entry in fs::read_dir(&sym_dir)? {
425 let file_entry = file_entry?;
426 let file_path = file_entry.path();
427 if !is_json_file(&file_path) {
428 continue;
429 }
430 let stem = file_path
431 .file_stem()
432 .and_then(|n| n.to_str())
433 .unwrap_or("")
434 .to_string();
435 if !live_entries.contains(&stem) {
436 fs::remove_file(&file_path)?;
437 pruned += 1;
438 }
439 }
440 if fs::read_dir(&sym_dir)?.next().is_none() {
442 fs::remove_dir(&sym_dir)?;
443 }
444 }
445 }
446 }
447
448 Ok(pruned)
449}
450
451fn prune_flat_dir(dir: &Path, live_keys: &std::collections::HashSet<String>) -> Result<usize> {
453 if !dir.is_dir() {
454 return Ok(0);
455 }
456 let mut removed = 0usize;
457 for entry in fs::read_dir(dir)? {
458 let entry = entry?;
459 let path = entry.path();
460 if !is_json_file(&path) {
461 continue;
462 }
463 let stem = path
464 .file_stem()
465 .and_then(|n| n.to_str())
466 .unwrap_or("")
467 .to_string();
468 if !live_keys.contains(&stem) {
469 fs::remove_file(&path)?;
470 removed += 1;
471 }
472 }
473 Ok(removed)
474}
475
476pub fn hydrate_from_dir(
483 repo: &Repository,
484 ref_name: &str,
485 dir: &Path,
486 agent_id: &str,
487) -> Result<HydrateSummary> {
488 hydrate_from_dir_with_cache(repo, None, ref_name, dir, agent_id)
489}
490
491pub fn hydrate_from_dir_with_cache(
494 repo: &Repository,
495 fts: Option<&SearchFtsDb>,
496 ref_name: &str,
497 dir: &Path,
498 agent_id: &str,
499) -> Result<HydrateSummary> {
500 let root = dir.join(SIDECAR_REL_ROOT);
501 if !root.exists() {
502 return Err(AsdError::Other(format!(
503 "no sidecar found at {} — did you mean to run `asd sync` first?",
504 root.display()
505 )));
506 }
507
508 let effects_dir = root.join("effects");
509 let ledger_dir = root.join("ledger");
510 let symbols_dir = root.join("symbols");
511 let rebinds_dir = root.join("rebinds");
512 let meta_dir = root.join("meta");
513
514 let mut symbols_loaded = 0usize;
531 let mut symbols_skipped = 0usize;
532 let mut blobs_rejected = 0usize;
533 if symbols_dir.is_dir() {
534 let mut by_qname: serde_json::Map<String, Value> = repo
536 .get_tree(ref_name, "/asd/v1/index/by-qname")
537 .ok()
538 .and_then(|v| v.as_object().cloned())
539 .unwrap_or_default();
540
541 let live_fps: std::collections::HashSet<String> = by_qname
546 .values()
547 .filter_map(|v| v.get("symbol_fp")?.as_str().map(|s| s.to_string()))
548 .collect();
549
550 let mut by_code: BTreeMap<String, serde_json::Map<String, Value>> = {
552 let existing = repo
553 .get_tree(ref_name, "/asd/v1/code")
554 .ok()
555 .and_then(|v| v.as_object().cloned())
556 .unwrap_or_default();
557 existing
558 .into_iter()
559 .filter_map(|(lang, subtree)| subtree.as_object().cloned().map(|m| (lang, m)))
560 .collect()
561 };
562
563 for entry in fs::read_dir(&symbols_dir)? {
564 let entry = entry?;
565 let path = entry.path();
566 if !is_json_file(&path) {
567 continue;
568 }
569 let text = match fs::read_to_string(&path) {
570 Ok(t) => t,
571 Err(e) => {
572 eprintln!("asd hydrate: skipping unreadable {}: {}", path.display(), e);
573 blobs_rejected += 1;
574 continue;
575 }
576 };
577 let sym: Symbol = match serde_json::from_str(&text) {
578 Ok(s) => s,
579 Err(e) => {
580 eprintln!(
581 "asd hydrate: skipping malformed symbol {}: {}",
582 path.display(),
583 e
584 );
585 blobs_rejected += 1;
586 continue;
587 }
588 };
589
590 if let Some(existing_val) = by_qname.get(&sym.qname) {
593 if let Ok(existing_sym) = serde_json::from_value::<Symbol>(existing_val.clone()) {
594 if existing_sym.symbol_fp == sym.symbol_fp {
595 symbols_skipped += 1;
596 continue; }
598 }
600 }
601
602 if live_fps.contains(&sym.symbol_fp) {
608 symbols_skipped += 1;
609 continue;
610 }
611
612 let sym_val = serde_json::to_value(&sym)?;
613 let code_key = format!("{}/{}", paths::clean(&sym.file), sym.symbol_fp);
614 by_qname.insert(sym.qname.clone(), sym_val.clone());
615 by_code
616 .entry(sym.language.clone())
617 .or_default()
618 .insert(code_key, sym_val);
619 symbols_loaded += 1;
620 }
621
622 if symbols_loaded > 0 {
623 let code_tree: serde_json::Map<String, Value> = by_code
624 .into_iter()
625 .map(|(lang, subtree)| (lang, Value::Object(subtree)))
626 .collect();
627 let spec = repo
628 .speculate(ref_name, Some("asd-hydrate-symbols".into()))
629 .map_err(|e| AsdError::Other(e.to_string()))?;
630 repo.spec_set_json(spec, "/asd/v1/index/by-qname", &Value::Object(by_qname))
631 .map_err(|e| AsdError::Other(e.to_string()))?;
632 if !code_tree.is_empty() {
633 repo.spec_set_json(spec, "/asd/v1/code", &Value::Object(code_tree))
634 .map_err(|e| AsdError::Other(e.to_string()))?;
635 }
636 let opts = CommitOptions::new(
637 agent_id,
638 IntentCategory::Checkpoint,
639 format!("asd hydrate: {} symbols", symbols_loaded),
640 );
641 repo.commit_speculation(spec, opts)
642 .map_err(|e| AsdError::Other(e.to_string()))?;
643 }
644 }
645
646 let mut effects_loaded = 0usize;
650 if effects_dir.is_dir() {
651 let mut by_effects: serde_json::Map<String, Value> = repo
652 .get_tree(ref_name, "/asd/v1/effects")
653 .ok()
654 .and_then(|v| v.as_object().cloned())
655 .unwrap_or_default();
656
657 for entry in fs::read_dir(&effects_dir)? {
658 let entry = entry?;
659 let path = entry.path();
660 if !is_json_file(&path) {
661 continue;
662 }
663 let text = match fs::read_to_string(&path) {
664 Ok(t) => t,
665 Err(e) => {
666 eprintln!("asd hydrate: skipping unreadable {}: {}", path.display(), e);
667 blobs_rejected += 1;
668 continue;
669 }
670 };
671 let decl: EffectDecl = match serde_json::from_str(&text) {
672 Ok(d) => d,
673 Err(e) => {
674 eprintln!(
675 "asd hydrate: skipping malformed effect {}: {}",
676 path.display(),
677 e
678 );
679 blobs_rejected += 1;
680 continue;
681 }
682 };
683 let val = serde_json::to_value(&decl)?;
684 by_effects.insert(decl.symbol_id.clone(), val);
685 effects_loaded += 1;
686 }
687
688 if effects_loaded > 0 {
689 let spec = repo
690 .speculate(ref_name, Some("asd-hydrate-effects".into()))
691 .map_err(|e| AsdError::Other(e.to_string()))?;
692 repo.spec_set_json(spec, "/asd/v1/effects", &Value::Object(by_effects))
693 .map_err(|e| AsdError::Other(e.to_string()))?;
694 let opts = CommitOptions::new(
695 agent_id,
696 IntentCategory::Checkpoint,
697 format!("asd hydrate: {} effects", effects_loaded),
698 );
699 repo.commit_speculation(spec, opts)
700 .map_err(|e| AsdError::Other(e.to_string()))?;
701 }
702 }
703
704 let ledger = hydrate_ledger(
709 repo,
710 ref_name,
711 &ledger_dir,
712 &rebinds_dir,
713 agent_id,
714 &mut blobs_rejected,
715 )?;
716 if let Some(fts) = fts {
717 for entry in &ledger.changed {
718 fts.upsert_ledger_entry(entry, ref_name)
719 .map_err(|e| AsdError::Other(format!("ledger cache: {e}")))?;
720 }
721 }
722 let ledger_entries_loaded = ledger.loaded;
723 let ledger_entries_skipped = ledger.skipped;
724 let rebinds_replayed = ledger.rebinds_replayed;
725
726 let missing_schema_version = !meta_dir.join("schema-version").is_file();
727
728 let refs_dropped = drop_orphaned_edge_refs(repo, ref_name, agent_id).unwrap_or_else(|e| {
734 eprintln!("asd hydrate: edge-ref cleanup failed: {}", e);
735 0
736 });
737
738 if refs_dropped > 0 {
739 eprintln!(
740 "asd hydrate: dropped {} orphaned call-graph ref(s) — run `asd repair` for details",
741 refs_dropped
742 );
743 }
744
745 let meta_dir = root.join("meta");
748 let _ = write_text_atomic(
753 &meta_dir.join("hydrated-at"),
754 &format!("{}\n", chrono::Utc::now().to_rfc3339()),
755 );
756 let _ = fs::remove_file(meta_dir.join("fresh-reset"));
759
760 Ok(HydrateSummary {
761 effects_loaded,
762 ledger_entries_loaded,
763 symbols_loaded,
764 ledger_entries_skipped,
765 rebinds_replayed,
766 missing_schema_version,
767 symbols_skipped,
768 blobs_rejected,
769 refs_dropped,
770 })
771}
772
773#[derive(Default)]
775struct LedgerHydrate {
776 loaded: usize,
777 skipped: usize,
778 rebinds_replayed: usize,
779 changed: Vec<LedgerEntry>,
781}
782
783struct SidecarCopy {
785 symbol_id: String,
786 entry: LedgerEntry,
787 written: Option<std::time::SystemTime>,
788}
789
790fn read_sidecar_ledger(
793 ledger_dir: &Path,
794 blobs_rejected: &mut usize,
795) -> Result<BTreeMap<String, Vec<SidecarCopy>>> {
796 let mut copies: BTreeMap<String, Vec<SidecarCopy>> = BTreeMap::new();
797 if !ledger_dir.is_dir() {
798 return Ok(copies);
799 }
800 for sym_entry in fs::read_dir(ledger_dir)? {
801 let sym_path = sym_entry?.path();
802 if !sym_path.is_dir() {
803 continue;
804 }
805 for file_entry in fs::read_dir(&sym_path)? {
806 let file_path = file_entry?.path();
807 if !is_json_file(&file_path) {
808 continue;
809 }
810 let text = match fs::read_to_string(&file_path) {
811 Ok(t) => t,
812 Err(e) => {
813 eprintln!(
814 "asd hydrate: skipping unreadable {}: {}",
815 file_path.display(),
816 e
817 );
818 *blobs_rejected += 1;
819 continue;
820 }
821 };
822 let entry: LedgerEntry = match serde_json::from_str(&text) {
823 Ok(e) => e,
824 Err(e) => {
825 eprintln!(
826 "asd hydrate: skipping malformed ledger entry {}: {}",
827 file_path.display(),
828 e
829 );
830 *blobs_rejected += 1;
831 continue;
832 }
833 };
834 let written = fs::metadata(&file_path).and_then(|m| m.modified()).ok();
835 copies
836 .entry(entry.entry_id.clone())
837 .or_default()
838 .push(SidecarCopy {
839 symbol_id: entry.symbol_id.clone(),
840 entry,
841 written,
842 });
843 }
844 }
845 Ok(copies)
846}
847
848fn hydrate_ledger(
866 repo: &Repository,
867 ref_name: &str,
868 ledger_dir: &Path,
869 rebinds_dir: &Path,
870 agent_id: &str,
871 blobs_rejected: &mut usize,
872) -> Result<LedgerHydrate> {
873 let copies = read_sidecar_ledger(ledger_dir, blobs_rejected)?;
874 let mut rebinds: Vec<Rebind> = Vec::new();
875 if rebinds_dir.is_dir() {
876 for entry in fs::read_dir(rebinds_dir)? {
877 let path = entry?.path();
878 if !is_json_file(&path) {
879 continue;
880 }
881 match fs::read_to_string(&path)
882 .map_err(|e| e.to_string())
883 .and_then(|t| serde_json::from_str::<Rebind>(&t).map_err(|e| e.to_string()))
884 {
885 Ok(rebind) => rebinds.push(rebind),
886 Err(e) => {
887 eprintln!("asd hydrate: skipping rebind {}: {}", path.display(), e);
888 *blobs_rejected += 1;
889 }
890 }
891 }
892 }
893 rebinds.sort_by_key(|r| r.at);
895 if copies.is_empty() && rebinds.is_empty() {
896 return Ok(LedgerHydrate::default());
897 }
898
899 let (spec, fork) = crate::subtree::speculate_at_head(repo, ref_name, "asd-hydrate-ledger")?;
900 let staged = (|| -> Result<LedgerHydrate> {
901 let rebinds_root = format!("{}/rebinds", paths::ASD_ROOT);
902 let mut ledger = crate::subtree::read_seed(repo, &fork, &paths::ledger_root())?;
903 let mut index = crate::subtree::read_seed(repo, &fork, &paths::ledger_index_root())?;
904 let mut records = crate::subtree::read_seed(repo, &fork, &rebinds_root)?;
905 let live: std::collections::HashSet<String> =
906 crate::subtree::read_seed(repo, &fork, &format!("{}/index/by-qname", paths::ASD_ROOT))?
907 .values()
908 .filter_map(|v| v.get("symbol_id")?.as_str().map(str::to_string))
909 .collect();
910 let stored = crate::ledger_dupes::locations(&ledger);
911 let mut out = LedgerHydrate::default();
912 let mut ledger_changed = false;
913
914 let put = |ledger: &mut serde_json::Map<String, Value>,
915 symbol_id: &str,
916 entry: &LedgerEntry|
917 -> Result<()> {
918 let Value::Object(entries) = ledger
919 .entry(symbol_id.to_string())
920 .or_insert_with(|| Value::Object(Default::default()))
921 else {
922 return Err(AsdError::Other(format!(
923 "ledger node for {symbol_id} is not a map"
924 )));
925 };
926 entries.insert(entry.entry_id.clone(), serde_json::to_value(entry)?);
927 Ok(())
928 };
929
930 for (entry_id, mut copies) in copies {
931 if let Some(symbols) = stored.get(&entry_id) {
932 for copy in copies {
933 let current = symbols
934 .contains(©.symbol_id)
935 .then(|| ledger.get(©.symbol_id)?.get(&entry_id))
936 .flatten()
937 .and_then(|v| serde_json::from_value::<LedgerEntry>(v.clone()).ok());
938 match current {
939 Some(current)
940 if crate::conclusions_export::revised_at(©.entry)
941 > crate::conclusions_export::revised_at(¤t) =>
942 {
943 put(&mut ledger, ©.symbol_id, ©.entry)?;
944 ledger_changed = true;
945 out.loaded += 1;
946 out.changed.push(copy.entry);
947 }
948 _ => out.skipped += 1,
949 }
950 }
951 continue;
952 }
953 copies.sort_by(|a, b| {
954 crate::ledger_dupes::claim(&b.symbol_id, &b.entry, &live)
955 .cmp(&crate::ledger_dupes::claim(&a.symbol_id, &a.entry, &live))
956 .then(b.written.cmp(&a.written))
957 .then(a.symbol_id.cmp(&b.symbol_id))
958 });
959 let mut copies = copies.into_iter();
960 let Some(keep) = copies.next() else {
961 continue;
962 };
963 out.skipped += copies.len();
964 put(&mut ledger, &keep.symbol_id, &keep.entry)?;
965 index.insert(entry_id, Value::String(keep.symbol_id.clone()));
966 ledger_changed = true;
967 out.loaded += 1;
968 out.changed.push(keep.entry);
969 }
970
971 let rebound_at = format!(
972 "rebound-at:{}",
973 chrono::Utc::now().format("%Y-%m-%dT%H:%M:%SZ")
974 );
975 let mut records_changed = false;
976 for rebind in &rebinds {
977 let mut replayed = false;
978 if !records.contains_key(&rebind.from_symbol_id) {
979 records.insert(rebind.from_symbol_id.clone(), serde_json::to_value(rebind)?);
980 records_changed = true;
981 replayed = true;
982 }
983 if rebind.from_symbol_id != rebind.to_symbol_id
986 && let Some(Value::Object(entries)) = ledger.remove(&rebind.from_symbol_id)
987 {
988 for (entry_id, value) in entries {
989 let mut entry: LedgerEntry = serde_json::from_value(value).map_err(|e| {
990 AsdError::Other(format!(
991 "ledger entry {entry_id} under {}: {e}",
992 rebind.from_symbol_id
993 ))
994 })?;
995 entry.symbol_id = rebind.to_symbol_id.clone();
996 entry
997 .tags
998 .retain(|t| t != "orphaned" && !t.starts_with("orphaned-at:"));
999 entry.tags.push(rebound_at.clone());
1000 put(&mut ledger, &rebind.to_symbol_id, &entry)?;
1001 index.insert(entry_id, Value::String(rebind.to_symbol_id.clone()));
1002 out.changed.push(entry);
1003 ledger_changed = true;
1004 replayed = true;
1005 }
1006 }
1007 out.rebinds_replayed += usize::from(replayed);
1008 }
1009
1010 let write = |path: &str, map: serde_json::Map<String, Value>| {
1011 repo.spec_set_json(spec, path, &Value::Object(map))
1012 .map_err(|e| AsdError::Other(e.to_string()))
1013 };
1014 if ledger_changed {
1015 write(&paths::ledger_root(), ledger)?;
1016 write(&paths::ledger_index_root(), index)?;
1017 }
1018 if records_changed {
1019 write(&rebinds_root, records)?;
1020 }
1021 Ok(out)
1022 })();
1023 match staged {
1024 Ok(out) if out.changed.is_empty() && out.rebinds_replayed == 0 => {
1025 let _ = repo.discard_speculation(spec);
1026 Ok(out)
1027 }
1028 Ok(out) => {
1029 let opts = CommitOptions::new(
1030 agent_id,
1031 IntentCategory::Checkpoint,
1032 format!(
1033 "asd hydrate: {} ledger entr{}, {} rebind(s) ({} sidecar cop{} already in the store)",
1034 out.loaded,
1035 if out.loaded == 1 { "y" } else { "ies" },
1036 out.rebinds_replayed,
1037 out.skipped,
1038 if out.skipped == 1 { "y" } else { "ies" },
1039 ),
1040 );
1041 repo.commit_speculation(spec, opts)
1042 .map_err(|e| AsdError::Other(e.to_string()))?;
1043 Ok(out)
1044 }
1045 Err(e) => {
1046 let _ = repo.discard_speculation(spec);
1047 Err(e)
1048 }
1049 }
1050}
1051
1052fn is_json_file(p: &Path) -> bool {
1053 p.is_file() && p.extension().and_then(|s| s.to_str()) == Some("json")
1054}
1055
1056fn write_json_atomic<T: serde::Serialize>(path: &Path, value: &T) -> Result<()> {
1059 let bytes = serde_json::to_vec_pretty(value)?;
1060 write_bytes_atomic(path, &bytes)
1061}
1062
1063fn write_text_atomic(path: &Path, text: &str) -> Result<()> {
1064 write_bytes_atomic(path, text.as_bytes())
1065}
1066
1067fn write_bytes_atomic(path: &Path, bytes: &[u8]) -> Result<()> {
1068 let tmp = tmp_path_for(path);
1069 if let Some(parent) = tmp.parent() {
1070 fs::create_dir_all(parent)?;
1071 }
1072 fs::write(&tmp, bytes)?;
1073 fs::rename(&tmp, path)?;
1074 Ok(())
1075}
1076
1077fn tmp_path_for(path: &Path) -> PathBuf {
1078 let mut os = path.as_os_str().to_owned();
1079 os.push(".tmp");
1080 PathBuf::from(os)
1081}