agentmux_srv/migrations/
m0007_agents_consolidate.rs

1// Copyright 2026, AgentMux Corp.
2// SPDX-License-Identifier: Apache-2.0
3
4use std::sync::Arc;
5use crate::backend::storage::agents_consolidate::consolidate_looks_incomplete;
6use crate::backend::storage::store::Store;
7use super::{Migration, MigrationContext, MigrationError, MigrationScope, VerifyOutcome};
8
9pub struct M0007AgentsConsolidate;
10
11impl Migration for M0007AgentsConsolidate {
12    fn id(&self) -> &'static str { "0007_agents_consolidate" }
13    fn scope(&self) -> MigrationScope { MigrationScope::Channel }
14    fn description(&self) -> &'static str { "Consolidate db_agent_definitions + db_agent_instances into db_agents" }
15
16    fn up(&self, ctx: &MigrationContext) -> Result<(), MigrationError> {
17        if !ctx.channel_store_path.exists() {
18            return Ok(());
19        }
20        let mstore = Arc::new(
21            Store::open(&ctx.channel_store_path)
22                .map_err(|e| MigrationError(format!("agents_consolidate: open mstore: {}", e)))?,
23        );
24        let stats = mstore
25            .run_agents_consolidate(Some(&ctx.data_dir))
26            .map_err(|e| MigrationError(format!("agents_consolidate: {}", e)))?;
27        // Phase 0c hardening: log the outcome instead of discarding it —
28        // `already_done` in particular used to be silently thrown away,
29        // making a "marker present, wrote nothing" run indistinguishable
30        // from a real backfill in the logs. See
31        // docs/specs/SPEC_MIGRATION_SYSTEM_HARDENING_2026_08_03.md Phase 0c.
32        tracing::info!(
33            already_done = stats.already_done,
34            templates_inserted = stats.templates_inserted,
35            user_defs_inserted = stats.user_defs_inserted,
36            instances_as_clone_inserted = stats.instances_as_clone_inserted,
37            instances_folded_into_def = stats.instances_folded_into_def,
38            "m0007_agents_consolidate: outcome",
39        );
40        Ok(())
41    }
42
43    /// Phase 1 doctor check — the incident this spec was written about
44    /// (`SPEC_MIGRATION_SYSTEM_HARDENING_2026_08_03` §1, F1/F2): a marker
45    /// said "done" while `db_agents` was suspected empty.
46    ///
47    /// The decision is NOT re-derived here: it is Phase 0b's
48    /// `consolidate_looks_incomplete` (`backend/storage/agents_consolidate.rs`),
49    /// the same rule the marker gate and `m0000_bootstrap` already use — one
50    /// definition of "incomplete", so the three cannot drift (reagent P2 on
51    /// #3058). This only adds the counts as evidence, through a read-only
52    /// connection.
53    fn verify(&self, ctx: &MigrationContext) -> VerifyOutcome {
54        let conn = match super::runner::open_readonly(&ctx.channel_store_path) {
55            Ok(Some(c)) => c,
56            Ok(None) => return VerifyOutcome::Ok("no channel store on this data dir; nothing to consolidate".into()),
57            Err(e) => return VerifyOutcome::Error(e),
58        };
59        let incomplete = match consolidate_looks_incomplete(&conn) {
60            Ok(b) => b,
61            // The rule queries all three tables; a store where one is missing
62            // is recorded-applied-but-schema-absent, which IS an inconsistency.
63            Err(e) => return VerifyOutcome::Error(format!("consolidate check: {}", e)),
64        };
65        let count = |t: &str| super::runner::table_count(&conn, t);
66        match (count("db_agent_definitions"), count("db_agent_instances"), count("db_agents")) {
67            (Ok(defs), Ok(instances), Ok(agents)) => consolidate_report(incomplete, defs, instances, agents),
68            (Err(e), _, _) | (_, Err(e), _) | (_, _, Err(e)) => VerifyOutcome::Error(e),
69        }
70    }
71}
72
73/// Formats the verdict `consolidate_looks_incomplete` reached, with the counts
74/// as evidence. Formatting only — the rule lives in one place.
75pub(super) fn consolidate_report(incomplete: bool, defs: i64, instances: i64, agents: i64) -> VerifyOutcome {
76    if incomplete {
77        return VerifyOutcome::Mismatch(format!(
78            "db_agent_definitions={} db_agent_instances={} but db_agents={} — recorded applied, wrote nothing",
79            defs, instances, agents
80        ));
81    }
82    if defs + instances == 0 {
83        return VerifyOutcome::Ok(format!("no legacy rows to consolidate (db_agents={})", agents));
84    }
85    VerifyOutcome::Ok(format!(
86        "db_agents={} (legacy: definitions={}, instances={})",
87        agents, defs, instances
88    ))
89}
90
91#[cfg(test)]
92mod tests {
93    use super::*;
94    use crate::backend::storage::store::Store;
95
96    #[test]
97    fn incomplete_verdict_reports_the_incident_shape_with_counts() {
98        match consolidate_report(true, 3, 2, 0) {
99            VerifyOutcome::Mismatch(m) => {
100                assert!(m.contains("db_agent_definitions=3"));
101                assert!(m.contains("db_agents=0"));
102            }
103            other => panic!("expected Mismatch, got {:?}", other),
104        }
105    }
106
107    #[test]
108    fn complete_verdicts_pass_and_carry_counts() {
109        assert!(matches!(consolidate_report(false, 0, 0, 0), VerifyOutcome::Ok(_)));
110        match consolidate_report(false, 4, 1, 5) {
111            VerifyOutcome::Ok(s) => assert!(s.contains("db_agents=5")),
112            other => panic!("expected Ok, got {:?}", other),
113        }
114    }
115
116    #[test]
117    fn verify_on_a_fresh_channel_store_passes_and_on_no_store_passes_without_creating_one() {
118        let dir = tempfile::tempdir().unwrap();
119        let data_dir = dir.path().join("data");
120        let path = data_dir.join("db").join("objects.db");
121        let ctx = MigrationContext {
122            home: dir.path().to_path_buf(),
123            data_dir: data_dir.clone(),
124            shared_store_path: dir.path().join("shared").join("store.db"),
125            channel_store_path: path.clone(),
126        };
127        // No store: nothing to consolidate, and verify must not create one.
128        assert!(matches!(M0007AgentsConsolidate.verify(&ctx), VerifyOutcome::Ok(_)));
129        assert!(!path.exists());
130
131        // A fresh schema: no legacy rows, empty target — that is consistent.
132        std::fs::create_dir_all(path.parent().unwrap()).unwrap();
133        drop(Store::open(&path).expect("create channel store"));
134        match M0007AgentsConsolidate.verify(&ctx) {
135            VerifyOutcome::Ok(s) => assert!(s.contains("no legacy rows"), "{}", s),
136            other => panic!("expected Ok, got {:?}", other),
137        }
138    }
139}
agentmux_srv/migrations/m0007_agents_consolidate.rs

agentmux_srv/migrations/
m0007_agents_consolidate.rs
