agentmux_srv/backend/storage/
agents_consolidate.rs

1// Copyright 2025-2026, AgentMux Corp.
2// SPDX-License-Identifier: Apache-2.0
3
4//! Phase 3a — one-shot backfill from `db_agent_definitions` +
5//! `db_agent_instances` into the new consolidated `db_agents` table.
6//!
7//! See `docs/specs/SPEC_AGENT_CONCEPT_CONSOLIDATION_2026_05_24.md`.
8//!
9//! Phase 3a is **write-only**: this migration populates `db_agents` so
10//! a later Phase 3b PR can flip readers over with full confidence the
11//! data is there. Phase 3c drops the old tables.
12//!
13//! Marker-file gated (`<data_dir>/migration_agents_consolidate_v1.flag`)
14//! so the backfill only runs once per data dir. Idempotent on second
15//! run (marker check short-circuits).
16//!
17//! Algorithm (per spec §"What `db_agent_instances` rows become"):
18//!
19//! 1. For each `db_agent_definitions WHERE is_seeded = 1`: INSERT a
20//!    template projection (`is_template = 1`, bindings empty).
21//!
22//! 2. For each `db_agent_definitions WHERE is_seeded = 0`: INSERT a
23//!    user-clone projection (`is_template = 0`, `parent_template_id =
24//!    parent_id`).
25//!
26//! 3. For each `db_agent_instances` row whose `definition_id` points at
27//!    a TEMPLATE: INSERT a new user-clone projection keyed by
28//!    `instance.id`, `parent_template_id = definition_id`, name +
29//!    bindings from the instance.
30//!
31//! 4. For each `db_agent_instances` row whose `definition_id` points at
32//!    an already-user-cloned definition: UPDATE the existing
33//!    user-clone projection (keyed by the definition id from pass 2) to
34//!    fold in the instance's bindings. If multiple instances point at
35//!    the same user-clone, the most-recent (`created_at` DESC) wins
36//!    and a warning is logged.
37//!
38//! Continuation rows (`parent_instance_id` non-empty) are skipped — the
39//! consolidated model has no place for them; they were the
40//! pre-Option-E continuation chain.
41
42use std::path::Path;
43
44use rusqlite::{params, Connection};
45use tracing::{info, warn};
46
47use super::error::StoreError;
48
49/// Marker filename. Lives in the data dir (one level above the `db/`
50/// subdir that holds `objects.db`).
51pub const CONSOLIDATE_MARKER: &str = "migration_agents_consolidate_v1.flag";
52
53/// Backfill statistics — useful for logs + tests.
54#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
55pub struct ConsolidateStats {
56    pub templates_inserted: usize,
57    pub user_defs_inserted: usize,
58    pub instances_as_clone_inserted: usize,
59    pub instances_folded_into_def: usize,
60    pub instances_skipped_continuation: usize,
61    pub instances_skipped_no_definition: usize,
62    pub instances_collision_warned: usize,
63    pub already_done: bool,
64}
65
66/// True when the backfill's source tables have data but its target
67/// (`db_agents`) doesn't — i.e. a marker (or a `db_migrations` stamp) claims
68/// this migration ran, but the data it's supposed to have produced isn't
69/// there. Used to stop trusting marker/stamp *existence* as proof of
70/// *effect* (see `docs/specs/SPEC_MIGRATION_SYSTEM_HARDENING_2026_08_03.md`
71/// Phase 0a/0b) — a stale marker (rebuilt/restored `objects.db` sitting next
72/// to an old flag file, or `m0000_bootstrap` stamping from a copied flag
73/// file with no matching data) is now detectable instead of silently
74/// trusted.
75///
76/// Cheap: up to three `COUNT(*)` queries, safe to call on every startup.
77///
78/// Tolerates either source table being physically absent (Phase 3e,
79/// `SPEC_AGENT_ARCHITECTURE_2026_05_27.md` — `db_agent_definitions`/
80/// `db_agent_instances` are dropped by `m0029_drop_legacy_agent_tables`):
81/// an absent table categorically has no rows, so it can never be the
82/// "source has rows but the target doesn't" shape this function detects.
83/// Load-bearing, not defensive-only — a fresh (v32+) channel is stamped
84/// `0007_agents_consolidate` applied via its own early-return no-op
85/// (`m0007_agents_consolidate.rs::up`'s `channel_store_path.exists()`
86/// guard, for a channel created before migrations ever ran), and a later
87/// `muxspect migrations --verify` doctor pass calls `m0007`'s `verify()` —
88/// which calls this — against that channel once it has real, ordinarily-
89/// populated `db_agents` data. Without this guard that doctor pass would
90/// error on "no such table" for every fresh install, not just an upgraded
91/// one that genuinely went through the old table pair.
92pub fn consolidate_looks_incomplete(conn: &Connection) -> Result<bool, StoreError> {
93    let defs = table_row_count_if_exists(conn, "db_agent_definitions")?;
94    let insts = table_row_count_if_exists(conn, "db_agent_instances")?;
95    let agents: i64 = conn.query_row("SELECT COUNT(*) FROM db_agents", [], |r| r.get(0))?;
96    Ok((defs > 0 || insts > 0) && agents == 0)
97}
98
99/// `SELECT COUNT(*) FROM <table>`, or `0` if the table doesn't exist at
100/// all — distinct from a genuinely empty table, which also counts as `0`
101/// here on purpose (both mean "no source rows to backfill").
102fn table_row_count_if_exists(conn: &Connection, table: &str) -> Result<i64, StoreError> {
103    let exists: bool = conn.query_row(
104        "SELECT EXISTS(SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = ?1)",
105        [table],
106        |row| row.get(0),
107    )?;
108    if !exists {
109        return Ok(0);
110    }
111    Ok(conn.query_row(&format!("SELECT COUNT(*) FROM {table}"), [], |r| r.get(0))?)
112}
113
114/// Run the one-shot consolidation backfill, gated by the marker file.
115///
116/// `data_dir` is the directory that holds the marker file (typically
117/// the parent of the `db/` directory). Pass `None` to skip marker
118/// gating — only intended for tests + in-memory stores.
119///
120/// Returns `Ok(stats)` on success (incl. the marker-already-present
121/// short-circuit). Failures roll back the active transaction and
122/// return the underlying SQLite error; the marker is NOT written.
123pub fn run_consolidate_migration(
124    conn: &mut Connection,
125    data_dir: Option<&Path>,
126) -> Result<ConsolidateStats, StoreError> {
127    // Marker gate. Existence alone is no longer trusted (Phase 0b): if the
128    // source tables have rows but db_agents doesn't, the marker is stale —
129    // fall through and re-run instead of silently reporting "already done."
130    if let Some(dir) = data_dir {
131        let marker = dir.join(CONSOLIDATE_MARKER);
132        if marker.exists() {
133            if !consolidate_looks_incomplete(conn)? {
134                return Ok(ConsolidateStats {
135                    already_done: true,
136                    ..Default::default()
137                });
138            }
139            warn!(
140                marker = %marker.display(),
141                "agents_consolidate: marker present but db_agents is empty while source tables have rows — \
142                 treating as stale and re-running instead of trusting it",
143            );
144        }
145    }
146
147    // Nothing to consolidate FROM if neither source table exists — a fresh
148    // (v32+) channel, or one already past `m0029_drop_legacy_agent_tables`.
149    // `up()`'s own `channel_store_path.exists()` guard already keeps this
150    // from firing for a channel created after this shipped (see
151    // `consolidate_looks_incomplete`'s doc comment for the doctor-pass path
152    // that reaches this function on an EXISTING, already-migrated channel
153    // instead).
154    let defs_exist: bool = conn.query_row(
155        "SELECT EXISTS(SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'db_agent_definitions')",
156        [],
157        |row| row.get(0),
158    )?;
159    if !defs_exist {
160        return Ok(ConsolidateStats::default());
161    }
162
163    // The backfill runs inside a transaction so a mid-flight failure
164    // leaves db_agents empty rather than half-populated. The dual-
165    // write call sites tolerate empty + idempotently upsert on the
166    // next mutation, so partial backfill state is recoverable.
167    let tx = conn.transaction()?;
168
169    let mut stats = ConsolidateStats::default();
170
171    // Pass 1 + 2 — definitions.
172    {
173        let mut def_stmt = tx.prepare(
174            "SELECT id, name, icon, provider, description,
175                    working_directory, shell, provider_flags, auto_start,
176                    restart_on_crash, idle_timeout_minutes, created_at,
177                    agent_type, environment, agent_bus_id, is_seeded,
178                    accounts, parent_id, branch_label, updated_at, slug
179             FROM db_agent_definitions",
180        )?;
181        let rows = def_stmt.query_map([], |row| {
182            Ok(DefRow {
183                id: row.get(0)?,
184                name: row.get(1)?,
185                icon: row.get(2)?,
186                provider: row.get(3)?,
187                description: row.get(4)?,
188                working_directory: row.get(5)?,
189                shell: row.get(6)?,
190                provider_flags: row.get(7)?,
191                auto_start: row.get(8)?,
192                restart_on_crash: row.get(9)?,
193                idle_timeout_minutes: row.get(10)?,
194                created_at: row.get(11)?,
195                agent_type: row.get(12)?,
196                environment: row.get(13)?,
197                agent_bus_id: row.get(14)?,
198                is_seeded: row.get(15)?,
199                accounts: row.get(16)?,
200                parent_id: row.get(17)?,
201                branch_label: row.get(18)?,
202                updated_at: row.get(19)?,
203                slug: row.get(20)?,
204            })
205        })?;
206        let mut defs: Vec<DefRow> = Vec::new();
207        for r in rows {
208            defs.push(r?);
209        }
210        drop(def_stmt);
211        for def in &defs {
212            let is_template = if def.is_seeded == 1 { 1_i64 } else { 0_i64 };
213            let parent_template_id = if def.is_seeded == 1 {
214                String::new()
215            } else {
216                def.parent_id.clone()
217            };
218            // Use INSERT OR REPLACE so a re-run after a partial state
219            // (e.g. a developer deleted the marker manually) doesn't
220            // explode on the PK; the user-clone-binding-fold path
221            // immediately below relies on definition rows existing.
222            tx.execute(
223                "INSERT OR REPLACE INTO db_agents (
224                    id, name, icon, description,
225                    is_template, parent_template_id,
226                    provider, provider_flags, shell, environment,
227                    agent_type, agent_bus_id, accounts,
228                    auto_start, restart_on_crash, idle_timeout_minutes,
229                    slug, branch_label,
230                    identity_id, memory_id, working_directory, github_context,
231                    instance_name,
232                    created_at, updated_at, is_seeded, user_hidden
233                 ) VALUES (
234                    ?1, ?2, ?3, ?4,
235                    ?5, ?6,
236                    ?7, ?8, ?9, ?10,
237                    ?11, ?12, ?13,
238                    ?14, ?15, ?16,
239                    ?17, ?18,
240                    '', '', ?19, '',
241                    '',
242                    ?20, ?21, ?22, 0
243                 )",
244                params![
245                    def.id,
246                    def.name,
247                    def.icon,
248                    def.description,
249                    is_template,
250                    parent_template_id,
251                    def.provider,
252                    def.provider_flags,
253                    def.shell,
254                    def.environment,
255                    def.agent_type,
256                    def.agent_bus_id,
257                    def.accounts,
258                    def.auto_start,
259                    def.restart_on_crash,
260                    def.idle_timeout_minutes,
261                    def.slug,
262                    def.branch_label,
263                    def.working_directory,
264                    def.created_at,
265                    def.updated_at,
266                    def.is_seeded,
267                ],
268            )?;
269            if def.is_seeded == 1 {
270                stats.templates_inserted += 1;
271            } else {
272                stats.user_defs_inserted += 1;
273            }
274        }
275    }
276
277    // Pass 3 + 4 — instances. Tolerates `db_agent_instances` not existing
278    // even though `db_agent_definitions` does (checked separately from the
279    // guard at the top of this function, which only covers "neither table
280    // exists at all") — a definition with no instance ever recorded against
281    // it is a legitimate shape (pre-consolidation agents that were defined
282    // but never launched), not a reason to skip Pass 1/2 too.
283    //
284    // Order by created_at DESC so the FIRST instance we see for a
285    // given user-cloned def is the most recent (the spec wants
286    // most-recent bindings to win on collision).
287    let insts_exist: bool = tx.query_row(
288        "SELECT EXISTS(SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'db_agent_instances')",
289        [],
290        |row| row.get(0),
291    )?;
292    let inst_rows: Vec<InstanceRow> = if !insts_exist {
293        Vec::new()
294    } else {
295        let mut stmt = tx.prepare(
296            "SELECT i.id, i.definition_id, i.parent_instance_id,
297                    i.instance_name, i.identity_id, i.memory_id,
298                    i.working_directory, i.github_context,
299                    i.created_at, i.display_hidden,
300                    d.is_seeded, d.name, d.icon, d.description,
301                    d.provider, d.provider_flags, d.shell, d.environment,
302                    d.agent_type, d.agent_bus_id, d.accounts,
303                    d.auto_start, d.restart_on_crash, d.idle_timeout_minutes,
304                    d.slug, d.branch_label
305             FROM db_agent_instances i
306             LEFT JOIN db_agent_definitions d ON d.id = i.definition_id
307             ORDER BY i.created_at DESC",
308        )?;
309        let iter = stmt.query_map([], |row| {
310            Ok(InstanceRow {
311                id: row.get(0)?,
312                definition_id: row.get(1)?,
313                parent_instance_id: row.get(2)?,
314                instance_name: row.get(3)?,
315                identity_id: row.get(4)?,
316                memory_id: row.get(5)?,
317                working_directory: row.get(6)?,
318                github_context: row.get(7)?,
319                created_at: row.get(8)?,
320                display_hidden: row.get::<_, i64>(9)? != 0,
321                def_is_seeded: row.get::<_, Option<i64>>(10)?.unwrap_or(0),
322                def_name: row.get::<_, Option<String>>(11)?.unwrap_or_default(),
323                def_icon: row.get::<_, Option<String>>(12)?.unwrap_or_default(),
324                def_description: row.get::<_, Option<String>>(13)?.unwrap_or_default(),
325                def_provider: row.get::<_, Option<String>>(14)?.unwrap_or_default(),
326                def_provider_flags: row.get::<_, Option<String>>(15)?.unwrap_or_default(),
327                def_shell: row.get::<_, Option<String>>(16)?.unwrap_or_default(),
328                def_environment: row.get::<_, Option<String>>(17)?.unwrap_or_default(),
329                def_agent_type: row
330                    .get::<_, Option<String>>(18)?
331                    .unwrap_or_else(|| "standalone".to_string()),
332                def_agent_bus_id: row.get::<_, Option<String>>(19)?.unwrap_or_default(),
333                def_accounts: row.get::<_, Option<String>>(20)?.unwrap_or_default(),
334                def_auto_start: row.get::<_, Option<i64>>(21)?.unwrap_or(0),
335                def_restart_on_crash: row.get::<_, Option<i64>>(22)?.unwrap_or(0),
336                def_idle_timeout_minutes: row.get::<_, Option<i64>>(23)?.unwrap_or(0),
337                def_slug: row.get::<_, Option<String>>(24)?.unwrap_or_default(),
338                def_branch_label: row.get::<_, Option<String>>(25)?.unwrap_or_default(),
339                def_present: row.get::<_, Option<i64>>(10)?.is_some(),
340            })
341        })?;
342        let mut out = Vec::new();
343        for r in iter {
344            out.push(r?);
345        }
346        out
347    };
348
349    // Track which user-clone def-ids already had their bindings folded.
350    // Multiple instances on the same user-clone-def → first wins (most
351    // recent because we ordered DESC); the rest get a warning.
352    let mut folded: std::collections::HashSet<String> = std::collections::HashSet::new();
353
354    for inst in &inst_rows {
355        if !inst.parent_instance_id.is_empty() {
356            stats.instances_skipped_continuation += 1;
357            continue;
358        }
359        if !inst.def_present {
360            // Orphaned instance — no definition row. The old schema's
361            // FK cascade would have removed this; if it survived,
362            // there's nothing to project against.
363            stats.instances_skipped_no_definition += 1;
364            warn!(
365                instance_id = %inst.id,
366                definition_id = %inst.definition_id,
367                "agents_consolidate: instance has no definition; skipping",
368            );
369            continue;
370        }
371        if inst.def_is_seeded == 1 {
372            // Instance of a TEMPLATE — INSERT a new user-clone row
373            // keyed by the instance id.
374            let name = if inst.instance_name.is_empty() {
375                inst.def_name.clone()
376            } else {
377                inst.instance_name.clone()
378            };
379            tx.execute(
380                "INSERT OR REPLACE INTO db_agents (
381                    id, name, icon, description,
382                    is_template, parent_template_id,
383                    provider, provider_flags, shell, environment,
384                    agent_type, agent_bus_id, accounts,
385                    auto_start, restart_on_crash, idle_timeout_minutes,
386                    slug, branch_label,
387                    identity_id, memory_id, working_directory, github_context,
388                    instance_name,
389                    created_at, updated_at, is_seeded, user_hidden
390                 ) VALUES (
391                    ?1, ?2, ?3, ?4,
392                    0, ?5,
393                    ?6, ?7, ?8, ?9,
394                    ?10, ?11, ?12,
395                    ?13, ?14, ?15,
396                    ?16, ?17,
397                    ?18, ?19, ?20, ?21,
398                    ?22,
399                    ?23, ?23, 0, ?24
400                 )",
401                params![
402                    inst.id,
403                    name,
404                    inst.def_icon,
405                    inst.def_description,
406                    inst.definition_id,
407                    inst.def_provider,
408                    inst.def_provider_flags,
409                    inst.def_shell,
410                    inst.def_environment,
411                    inst.def_agent_type,
412                    inst.def_agent_bus_id,
413                    inst.def_accounts,
414                    inst.def_auto_start,
415                    inst.def_restart_on_crash,
416                    inst.def_idle_timeout_minutes,
417                    inst.def_slug,
418                    inst.def_branch_label,
419                    inst.identity_id,
420                    inst.memory_id,
421                    inst.working_directory,
422                    inst.github_context,
423                    inst.instance_name,
424                    inst.created_at,
425                    if inst.display_hidden { 1_i64 } else { 0_i64 },
426                ],
427            )?;
428            stats.instances_as_clone_inserted += 1;
429        } else {
430            // Instance of an already-user-cloned definition — UPDATE
431            // the existing user-clone row (keyed by definition_id)
432            // to fold in the instance bindings. Collision: only the
433            // first (most recent) wins; warn on subsequent.
434            if folded.contains(&inst.definition_id) {
435                stats.instances_collision_warned += 1;
436                warn!(
437                    instance_id = %inst.id,
438                    definition_id = %inst.definition_id,
439                    "agents_consolidate: multiple instances on one user-cloned def; keeping most-recent bindings",
440                );
441                continue;
442            }
443            let name = if inst.instance_name.is_empty() {
444                inst.def_name.clone()
445            } else {
446                inst.instance_name.clone()
447            };
448            // Stamp `updated_at` with the folded instance's
449            // `created_at` (its launch moment). Without this, the
450            // backfill UPDATE leaves `updated_at` as whatever the
451            // def's edit time was — which makes `db_agents.updated_at`
452            // useless as a "most-recently-used" sort key for migrated
453            // stores, breaking the ordering invariant that the live
454            // launch write (`Store::instance_create`) maintains
455            // for new launches. Surfaced in #1110 via the
456            // first read-flip (the since-deleted `instance_get_by_name`)
457            // ordering by `updated_at DESC`.
458            //
459            // The loop iterates `ORDER BY i.created_at DESC`, so the
460            // FIRST instance per def_id is the most recent, and that's
461            // the one whose `created_at` we want imprinted as
462            // `updated_at`. Collision warns skip subsequent rows for
463            // the same def.
464            tx.execute(
465                "UPDATE db_agents SET
466                    name = ?1,
467                    identity_id = ?2,
468                    memory_id = ?3,
469                    working_directory = ?4,
470                    github_context = ?5,
471                    instance_name = ?6,
472                    user_hidden = ?7,
473                    updated_at = MAX(updated_at, ?8)
474                 WHERE id = ?9 AND is_template = 0",
475                params![
476                    name,
477                    inst.identity_id,
478                    inst.memory_id,
479                    inst.working_directory,
480                    inst.github_context,
481                    inst.instance_name,
482                    if inst.display_hidden { 1_i64 } else { 0_i64 },
483                    inst.created_at,
484                    inst.definition_id,
485                ],
486            )?;
487            folded.insert(inst.definition_id.clone());
488            stats.instances_folded_into_def += 1;
489        }
490    }
491
492    tx.commit()?;
493
494    // Marker written AFTER successful commit so a crash mid-backfill
495    // leaves the marker absent → next start retries from scratch.
496    if let Some(dir) = data_dir {
497        let marker = dir.join(CONSOLIDATE_MARKER);
498        if let Err(e) = std::fs::write(&marker, b"phase3a") {
499            // The data is in place; failing to write the marker just
500            // means the next startup will redo the work. Log + return
501            // success.
502            warn!(
503                error = %e,
504                marker = %marker.display(),
505                "agents_consolidate: failed to write marker; next startup will redo backfill",
506            );
507        }
508    }
509
510    info!(
511        templates_inserted = stats.templates_inserted,
512        user_defs_inserted = stats.user_defs_inserted,
513        instances_as_clone_inserted = stats.instances_as_clone_inserted,
514        instances_folded_into_def = stats.instances_folded_into_def,
515        instances_skipped_continuation = stats.instances_skipped_continuation,
516        instances_skipped_no_definition = stats.instances_skipped_no_definition,
517        instances_collision_warned = stats.instances_collision_warned,
518        "agents_consolidate: backfill completed",
519    );
520    Ok(stats)
521}
522
523
524/// Snapshot of one `db_agent_definitions` row, narrow projection
525/// matching what the backfill needs.
526struct DefRow {
527    id: String,
528    name: String,
529    icon: String,
530    provider: String,
531    description: String,
532    working_directory: String,
533    shell: String,
534    provider_flags: String,
535    auto_start: i64,
536    restart_on_crash: i64,
537    idle_timeout_minutes: i64,
538    created_at: i64,
539    agent_type: String,
540    environment: String,
541    agent_bus_id: String,
542    is_seeded: i64,
543    accounts: String,
544    parent_id: String,
545    branch_label: String,
546    updated_at: i64,
547    slug: String,
548}
549
550/// Snapshot of one `db_agent_instances` row plus the LEFT-JOINed
551/// definition fields the backfill copies into `db_agents`.
552struct InstanceRow {
553    id: String,
554    definition_id: String,
555    parent_instance_id: String,
556    instance_name: String,
557    identity_id: String,
558    memory_id: String,
559    working_directory: String,
560    github_context: String,
561    created_at: i64,
562    display_hidden: bool,
563    def_present: bool,
564    def_is_seeded: i64,
565    def_name: String,
566    def_icon: String,
567    def_description: String,
568    def_provider: String,
569    def_provider_flags: String,
570    def_shell: String,
571    def_environment: String,
572    def_agent_type: String,
573    def_agent_bus_id: String,
574    def_accounts: String,
575    def_auto_start: i64,
576    def_restart_on_crash: i64,
577    def_idle_timeout_minutes: i64,
578    def_slug: String,
579    def_branch_label: String,
580}
581
582#[cfg(test)]
583mod tests {
584    use super::*;
585    use crate::backend::storage::migrations::run_object_schema;
586
587    /// `db_agent_definitions`/`db_agent_instances` are gone from the flat
588    /// schema as of v32 (`OBJECT_SCHEMA_VERSION`'s v32 doc comment,
589    /// `SPEC_AGENT_ARCHITECTURE_2026_05_27.md` Phase 3e) — this fixture
590    /// stands up the pre-drop legacy shape by hand, since `run_object_schema`
591    /// no longer will. Every column `insert_def`/`insert_instance` write is
592    /// present; the FK matches what the tables actually had before the
593    /// drop, so cascade behavior this module's own logic depends on
594    /// (`run_consolidate_migration` reads both, unrelated to cascades) still
595    /// matches production history.
596    fn fresh_conn() -> Connection {
597        let conn = Connection::open_in_memory().unwrap();
598        conn.execute_batch("PRAGMA foreign_keys=ON;").unwrap();
599        run_object_schema(&conn).unwrap();
600        conn.execute_batch(
601            "CREATE TABLE db_agent_definitions (
602                id                   TEXT PRIMARY KEY,
603                slug                 TEXT NOT NULL DEFAULT '',
604                name                 TEXT NOT NULL,
605                icon                 TEXT NOT NULL DEFAULT '✦',
606                provider             TEXT NOT NULL,
607                description          TEXT NOT NULL DEFAULT '',
608                working_directory    TEXT NOT NULL DEFAULT '',
609                shell                TEXT NOT NULL DEFAULT '',
610                provider_flags       TEXT NOT NULL DEFAULT '',
611                auto_start           INTEGER NOT NULL DEFAULT 0,
612                restart_on_crash     INTEGER NOT NULL DEFAULT 0,
613                idle_timeout_minutes INTEGER NOT NULL DEFAULT 0,
614                agent_type           TEXT NOT NULL DEFAULT 'standalone',
615                environment          TEXT NOT NULL DEFAULT '',
616                agent_bus_id         TEXT NOT NULL DEFAULT '',
617                is_seeded            INTEGER NOT NULL DEFAULT 0,
618                accounts             TEXT NOT NULL DEFAULT '',
619                parent_id            TEXT NOT NULL DEFAULT '',
620                branch_label         TEXT NOT NULL DEFAULT '',
621                created_at           INTEGER NOT NULL DEFAULT 0,
622                updated_at           INTEGER NOT NULL DEFAULT 0,
623                user_hidden          INTEGER NOT NULL DEFAULT 0
624             );
625             CREATE TABLE db_agent_instances (
626                id                 TEXT PRIMARY KEY,
627                definition_id      TEXT NOT NULL,
628                parent_instance_id TEXT NOT NULL DEFAULT '',
629                block_id           TEXT NOT NULL DEFAULT '',
630                session_id         TEXT NOT NULL DEFAULT '',
631                status             TEXT NOT NULL DEFAULT 'running',
632                github_context     TEXT NOT NULL DEFAULT '',
633                identity_id        TEXT NOT NULL DEFAULT '',
634                memory_id          TEXT NOT NULL DEFAULT '',
635                instance_name      TEXT NOT NULL DEFAULT '',
636                working_directory  TEXT NOT NULL DEFAULT '',
637                display_hidden     INTEGER NOT NULL DEFAULT 0,
638                started_at         INTEGER NOT NULL DEFAULT 0,
639                ended_at           INTEGER NOT NULL DEFAULT 0,
640                created_at         INTEGER NOT NULL DEFAULT 0,
641                FOREIGN KEY (definition_id) REFERENCES db_agent_definitions(id) ON DELETE CASCADE
642             );",
643        )
644        .unwrap();
645        conn
646    }
647
648    fn insert_def(
649        conn: &Connection,
650        id: &str,
651        name: &str,
652        is_seeded: i64,
653        parent_id: &str,
654    ) {
655        conn.execute(
656            "INSERT INTO db_agent_definitions
657                (id, slug, name, icon, provider, description, working_directory, shell,
658                 provider_flags, auto_start, restart_on_crash, idle_timeout_minutes,
659                 created_at, agent_type, environment, agent_bus_id, is_seeded, accounts,
660                 parent_id, branch_label, updated_at)
661             VALUES (?1, ?2, ?3, '✦', 'claude', 'desc', '', 'bash',
662                     '', 0, 0, 0,
663                     ?4, 'standalone', '', '', ?5, '',
664                     ?6, '', ?4)",
665            params![id, id, name, 1000_i64, is_seeded, parent_id],
666        )
667        .unwrap();
668    }
669
670    fn insert_instance(
671        conn: &Connection,
672        id: &str,
673        definition_id: &str,
674        instance_name: &str,
675        identity_id: &str,
676        memory_id: &str,
677        working_directory: &str,
678        created_at: i64,
679        display_hidden: bool,
680    ) {
681        conn.execute(
682            "INSERT INTO db_agent_instances
683                (id, definition_id, parent_instance_id, block_id, session_id, status,
684                 github_context, started_at, ended_at, created_at, identity_id, memory_id,
685                 instance_name, working_directory, display_hidden)
686             VALUES (?1, ?2, '', '', '', 'running', '', ?3, 0, ?3, ?4, ?5, ?6, ?7, ?8)",
687            params![
688                id,
689                definition_id,
690                created_at,
691                identity_id,
692                memory_id,
693                instance_name,
694                working_directory,
695                if display_hidden { 1_i64 } else { 0_i64 },
696            ],
697        )
698        .unwrap();
699    }
700
701    fn count_agents(conn: &Connection, where_clause: &str) -> i64 {
702        let sql = format!("SELECT COUNT(*) FROM db_agents WHERE {where_clause}");
703        conn.query_row(&sql, [], |row| row.get(0)).unwrap()
704    }
705
706    #[test]
707    fn round_trip_template_user_clone_and_instance_lifecycle() {
708        let mut conn = fresh_conn();
709
710        // 2 templates, 1 user-cloned definition (from template tpl-1),
711        // 3 instances:
712        //   - inst-A on template tpl-1 (named "Maks")
713        //   - inst-B on template tpl-2 (unnamed)
714        //   - inst-C on user-clone-def def-u1 (named "Custom")
715        insert_def(&conn, "tpl-1", "Coder", 1, "");
716        insert_def(&conn, "tpl-2", "Reviewer", 1, "");
717        insert_def(&conn, "def-u1", "Maks-the-Coder", 0, "tpl-1");
718        insert_instance(&conn, "inst-A", "tpl-1", "Maks", "id-1", "mem-1", "/wd/a", 100, false);
719        insert_instance(&conn, "inst-B", "tpl-2", "", "", "", "/wd/b", 200, false);
720        insert_instance(&conn, "inst-C", "def-u1", "Custom", "id-2", "mem-2", "/wd/c", 300, false);
721
722        let stats = run_consolidate_migration(&mut conn, None).unwrap();
723        assert_eq!(stats.templates_inserted, 2);
724        assert_eq!(stats.user_defs_inserted, 1);
725        assert_eq!(stats.instances_as_clone_inserted, 2);
726        assert_eq!(stats.instances_folded_into_def, 1);
727        assert_eq!(stats.instances_skipped_continuation, 0);
728
729        // Templates present, is_template=1, parent_template_id empty.
730        assert_eq!(count_agents(&conn, "is_template = 1"), 2);
731        for tpl in &["tpl-1", "tpl-2"] {
732            let parent: String = conn
733                .query_row(
734                    "SELECT parent_template_id FROM db_agents WHERE id = ?1",
735                    params![tpl],
736                    |r| r.get(0),
737                )
738                .unwrap();
739            assert_eq!(parent, "");
740        }
741
742        // 3 user-clone rows: def-u1 (folded from inst-C), inst-A, inst-B.
743        assert_eq!(count_agents(&conn, "is_template = 0"), 3);
744
745        // inst-A → parent_template_id = tpl-1, name = "Maks".
746        let (parent, name, identity): (String, String, String) = conn
747            .query_row(
748                "SELECT parent_template_id, name, identity_id FROM db_agents WHERE id = 'inst-A'",
749                [],
750                |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)),
751            )
752            .unwrap();
753        assert_eq!(parent, "tpl-1");
754        assert_eq!(name, "Maks");
755        assert_eq!(identity, "id-1");
756
757        // inst-B (unnamed) → name falls back to template name "Reviewer".
758        let name: String = conn
759            .query_row(
760                "SELECT name FROM db_agents WHERE id = 'inst-B'",
761                [],
762                |r| r.get(0),
763            )
764            .unwrap();
765        assert_eq!(name, "Reviewer");
766
767        // def-u1 (user-clone) folded inst-C's bindings.
768        let (name, identity, memory): (String, String, String) = conn
769            .query_row(
770                "SELECT name, identity_id, memory_id FROM db_agents WHERE id = 'def-u1'",
771                [],
772                |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)),
773            )
774            .unwrap();
775        assert_eq!(name, "Custom"); // instance_name overrode def name
776        assert_eq!(identity, "id-2");
777        assert_eq!(memory, "mem-2");
778    }
779
780    #[test]
781    fn multiple_instances_on_one_user_clone_keeps_most_recent_bindings() {
782        let mut conn = fresh_conn();
783        insert_def(&conn, "tpl-1", "Coder", 1, "");
784        insert_def(&conn, "def-u1", "User Coder", 0, "tpl-1");
785        // Two instances on def-u1; most recent has id-RECENT.
786        insert_instance(&conn, "inst-old", "def-u1", "Old", "id-OLD", "mem-OLD", "/wd/old", 100, false);
787        insert_instance(&conn, "inst-new", "def-u1", "New", "id-RECENT", "mem-RECENT", "/wd/new", 999, false);
788
789        let stats = run_consolidate_migration(&mut conn, None).unwrap();
790        assert_eq!(stats.instances_collision_warned, 1);
791
792        let identity: String = conn
793            .query_row(
794                "SELECT identity_id FROM db_agents WHERE id = 'def-u1'",
795                [],
796                |r| r.get(0),
797            )
798            .unwrap();
799        assert_eq!(identity, "id-RECENT", "most-recent instance's bindings must win");
800    }
801
802    #[test]
803    fn fold_into_user_clone_stamps_updated_at_from_instance_created_at() {
804        // Codex P2 on PR #1110: ordering `db_agents` by `updated_at`
805        // would misbehave on migrated stores if the fold UPDATE
806        // didn't write `updated_at`. After the fix, the folded
807        // user-clone row's `updated_at` equals the most-recent
808        // instance's `created_at` (the launch moment), not the def's
809        // edit time — matching what the live launch write
810        // (`Store::instance_create`) stamps for new
811        // launches.
812        let mut conn = fresh_conn();
813        // Def created at t=1000 (insert_def uses ?4 for both
814        // created_at and updated_at).
815        insert_def(&conn, "tpl-1", "Coder", 1, "");
816        insert_def(&conn, "def-u1", "User Coder", 0, "tpl-1");
817        // Instance launched at t=5000 — later than the def.
818        insert_instance(
819            &conn, "inst-1", "def-u1", "Maks", "id-1", "mem-1", "/wd/m", 5000, false,
820        );
821
822        run_consolidate_migration(&mut conn, None).unwrap();
823
824        let updated_at: i64 = conn
825            .query_row(
826                "SELECT updated_at FROM db_agents WHERE id = 'def-u1'",
827                [],
828                |r| r.get(0),
829            )
830            .unwrap();
831        assert_eq!(
832            updated_at, 5000,
833            "folded user-clone row's updated_at must reflect the instance's launch time, \
834             not the def's edit time, so ORDER BY updated_at picks the right row"
835        );
836    }
837
838    #[test]
839    fn fold_into_user_clone_keeps_higher_updated_at() {
840        // MAX(updated_at, ?) semantics: if the def was edited AFTER
841        // the instance launched, the def's edit time should stay as
842        // updated_at (the def edit IS the most-recent touch).
843        let mut conn = fresh_conn();
844        insert_def(&conn, "tpl-1", "Coder", 1, "");
845        // Def edited at t=8000.
846        conn.execute(
847            "UPDATE db_agent_definitions SET updated_at = 8000 WHERE id = 'tpl-1'",
848            [],
849        )
850        .unwrap();
851        insert_def(&conn, "def-u1", "User Coder", 0, "tpl-1");
852        conn.execute(
853            "UPDATE db_agent_definitions SET updated_at = 8000 WHERE id = 'def-u1'",
854            [],
855        )
856        .unwrap();
857        // Now insert an OLDER instance.
858        insert_instance(
859            &conn, "inst-1", "def-u1", "Maks", "id-1", "mem-1", "/wd/m", 3000, false,
860        );
861
862        run_consolidate_migration(&mut conn, None).unwrap();
863
864        let updated_at: i64 = conn
865            .query_row(
866                "SELECT updated_at FROM db_agents WHERE id = 'def-u1'",
867                [],
868                |r| r.get(0),
869            )
870            .unwrap();
871        assert_eq!(
872            updated_at, 8000,
873            "later def edit beats older instance launch — MAX(updated_at, inst.created_at)"
874        );
875    }
876
877    #[test]
878    fn marker_short_circuits_second_run() {
879        let mut conn = fresh_conn();
880        insert_def(&conn, "tpl-1", "Coder", 1, "");
881        let tmp = tempfile::tempdir().unwrap();
882
883        let stats1 = run_consolidate_migration(&mut conn, Some(tmp.path())).unwrap();
884        assert_eq!(stats1.templates_inserted, 1);
885        assert!(!stats1.already_done);
886        assert!(tmp.path().join(CONSOLIDATE_MARKER).exists());
887
888        // Insert a NEW row after the marker; second call must NOT see
889        // it.
890        insert_def(&conn, "tpl-2", "Reviewer", 1, "");
891        let stats2 = run_consolidate_migration(&mut conn, Some(tmp.path())).unwrap();
892        assert!(stats2.already_done);
893        assert_eq!(stats2.templates_inserted, 0);
894        // db_agents only has tpl-1 — the post-marker insert is the
895        // dual-write hook's problem, not the backfill's.
896        assert_eq!(count_agents(&conn, "is_template = 1"), 1);
897    }
898
899    #[test]
900    fn skips_continuation_rows() {
901        let mut conn = fresh_conn();
902        insert_def(&conn, "tpl-1", "Coder", 1, "");
903        insert_instance(&conn, "inst-A", "tpl-1", "Original", "", "", "/wd/a", 100, false);
904        // Continuation row (parent_instance_id = inst-A).
905        conn.execute(
906            "INSERT INTO db_agent_instances
907                (id, definition_id, parent_instance_id, block_id, session_id, status,
908                 github_context, started_at, ended_at, created_at, identity_id, memory_id,
909                 instance_name, working_directory, display_hidden)
910             VALUES ('inst-cont', 'tpl-1', 'inst-A', '', '', 'running', '', 200, 0, 200,
911                     '', '', 'Original', '/wd/a', 0)",
912            [],
913        )
914        .unwrap();
915
916        let stats = run_consolidate_migration(&mut conn, None).unwrap();
917        assert_eq!(stats.instances_skipped_continuation, 1);
918        // Only inst-A and the template made it into db_agents.
919        assert_eq!(count_agents(&conn, "1 = 1"), 2);
920        assert_eq!(count_agents(&conn, "id = 'inst-cont'"), 0);
921    }
922
923    #[test]
924    fn empty_database_is_clean_noop() {
925        let mut conn = fresh_conn();
926        let stats = run_consolidate_migration(&mut conn, None).unwrap();
927        assert_eq!(stats, ConsolidateStats::default());
928        assert_eq!(count_agents(&conn, "1 = 1"), 0);
929    }
930
931    #[test]
932    fn preserves_hidden_flag_into_user_hidden() {
933        let mut conn = fresh_conn();
934        insert_def(&conn, "tpl-1", "Coder", 1, "");
935        insert_instance(&conn, "inst-H", "tpl-1", "Hidden", "", "", "/wd/h", 100, true);
936        run_consolidate_migration(&mut conn, None).unwrap();
937        let hidden: i64 = conn
938            .query_row(
939                "SELECT user_hidden FROM db_agents WHERE id = 'inst-H'",
940                [],
941                |r| r.get(0),
942            )
943            .unwrap();
944        assert_eq!(hidden, 1);
945    }
946
947    #[test]
948    fn definition_working_directory_survives_the_pass_1_2_backfill() {
949        // Regression: the Pass 1/2 INSERT (templates + user-cloned
950        // definitions with no instance to fold) used to hardcode
951        // working_directory as '' instead of binding def.working_directory,
952        // silently discarding it during consolidation — audit finding,
953        // docs/reports/REPORT_REPO_HEALTH_AUDIT_2026_07_20.md §1.2. A
954        // definition that was never instantiated (no db_agent_instances
955        // row) only ever goes through this path — Pass 3/4 (instances)
956        // never touches it — so this case wasn't covered by any existing
957        // test here.
958        let mut conn = fresh_conn();
959        conn.execute(
960            "INSERT INTO db_agent_definitions
961                (id, slug, name, icon, provider, description, working_directory, shell,
962                 provider_flags, auto_start, restart_on_crash, idle_timeout_minutes,
963                 created_at, agent_type, environment, agent_bus_id, is_seeded, accounts,
964                 parent_id, branch_label, updated_at)
965             VALUES ('tpl-wd', 'tpl-wd', 'Coder', '✦', 'claude', 'desc',
966                     '/home/user/my-project', 'bash',
967                     '', 0, 0, 0,
968                     1000, 'standalone', '', '', 1, '',
969                     '', '', 1000)",
970            [],
971        )
972        .unwrap();
973
974        run_consolidate_migration(&mut conn, None).unwrap();
975
976        let working_directory: String = conn
977            .query_row(
978                "SELECT working_directory FROM db_agents WHERE id = 'tpl-wd'",
979                [],
980                |r| r.get(0),
981            )
982            .unwrap();
983        assert_eq!(
984            working_directory, "/home/user/my-project",
985            "a definition's working_directory must survive Pass 1/2 consolidation \
986             even when it was never instantiated (no instance row to fold bindings from)"
987        );
988    }
989
990    // ── Phase 0 hardening (SPEC_MIGRATION_SYSTEM_HARDENING_2026_08_03) ──────
991
992    #[test]
993    fn consolidate_looks_incomplete_is_false_on_a_fresh_db() {
994        let conn = fresh_conn();
995        // No source rows, no db_agents rows — nothing to backfill, not "incomplete".
996        assert!(!consolidate_looks_incomplete(&conn).unwrap());
997    }
998
999    /// The scenario `consolidate_looks_incomplete`'s own doc comment names:
1000    /// a genuinely v32+ channel, born without either legacy table at all
1001    /// (not just empty — physically absent, unlike `fresh_conn`'s fixture
1002    /// above, which stands them up empty for the OTHER tests in this file
1003    /// that still need to insert legacy rows). Must not error.
1004    #[test]
1005    fn consolidate_looks_incomplete_tolerates_both_tables_being_physically_absent() {
1006        let conn = Connection::open_in_memory().unwrap();
1007        conn.execute_batch("PRAGMA foreign_keys=ON;").unwrap();
1008        run_object_schema(&conn).unwrap();
1009        assert!(!consolidate_looks_incomplete(&conn).unwrap());
1010    }
1011
1012    /// Same scenario, through `run_consolidate_migration` directly (not
1013    /// just its `consolidate_looks_incomplete` sub-check) — the doctor-pass
1014    /// path this guards against calls `verify()`, which only reads via
1015    /// `consolidate_looks_incomplete`, but `run_consolidate_migration`
1016    /// itself must also tolerate this shape rather than erroring on the
1017    /// first `SELECT ... FROM db_agent_definitions` if anything ever calls
1018    /// it directly against a table-less store.
1019    #[test]
1020    fn run_consolidate_migration_tolerates_both_tables_being_physically_absent() {
1021        let mut conn = Connection::open_in_memory().unwrap();
1022        conn.execute_batch("PRAGMA foreign_keys=ON;").unwrap();
1023        run_object_schema(&conn).unwrap();
1024        let stats = run_consolidate_migration(&mut conn, None).unwrap();
1025        assert_eq!(stats, ConsolidateStats::default());
1026    }
1027
1028    #[test]
1029    fn consolidate_looks_incomplete_is_false_once_backfilled() {
1030        let mut conn = fresh_conn();
1031        insert_def(&conn, "tpl-1", "Coder", 1, "");
1032        run_consolidate_migration(&mut conn, None).unwrap();
1033        assert!(!consolidate_looks_incomplete(&conn).unwrap());
1034    }
1035
1036    #[test]
1037    fn consolidate_looks_incomplete_is_true_when_source_has_rows_but_db_agents_is_empty() {
1038        let conn = fresh_conn();
1039        // Source rows present, but db_agents was never populated (the exact
1040        // shape a stale/orphaned marker file produces).
1041        insert_def(&conn, "tpl-1", "Coder", 1, "");
1042        assert!(consolidate_looks_incomplete(&conn).unwrap());
1043    }
1044
1045    #[test]
1046    fn stale_marker_does_not_block_a_real_backfill() {
1047        // Acceptance test for Phase 0a/0b: a data dir with a marker file
1048        // present but an empty db_agents table (simulating a rebuilt/restored
1049        // objects.db sitting next to a stale flag file) must NOT be trusted
1050        // as "already done" — the migration must detect this and actually
1051        // run the backfill instead of silently no-op'ing.
1052        let tmp = tempfile::TempDir::new().unwrap();
1053        std::fs::write(tmp.path().join(CONSOLIDATE_MARKER), b"phase3a").unwrap();
1054
1055        let mut conn = fresh_conn();
1056        insert_def(&conn, "tpl-1", "Coder", 1, "");
1057        assert_eq!(count_agents(&conn, "1=1"), 0, "db_agents must start empty for this test to be meaningful");
1058
1059        let stats = run_consolidate_migration(&mut conn, Some(tmp.path())).unwrap();
1060        assert!(!stats.already_done, "a stale marker must not short-circuit as already_done");
1061        assert_eq!(stats.templates_inserted, 1);
1062        assert_eq!(count_agents(&conn, "1=1"), 1, "db_agents must actually get populated, not silently skipped");
1063    }
1064
1065    #[test]
1066    fn genuinely_completed_marker_still_short_circuits() {
1067        // The flip side: once db_agents is correctly populated, a present
1068        // marker SHOULD short-circuit (already_done=true) — Phase 0's fix
1069        // must not make every run pay the full backfill cost forever.
1070        let tmp = tempfile::TempDir::new().unwrap();
1071        std::fs::write(tmp.path().join(CONSOLIDATE_MARKER), b"phase3a").unwrap();
1072
1073        let mut conn = fresh_conn();
1074        insert_def(&conn, "tpl-1", "Coder", 1, "");
1075        run_consolidate_migration(&mut conn, None).unwrap(); // populate db_agents directly
1076        assert_eq!(count_agents(&conn, "1=1"), 1);
1077
1078        let stats = run_consolidate_migration(&mut conn, Some(tmp.path())).unwrap();
1079        assert!(stats.already_done, "a marker backed by real data must still short-circuit");
1080        assert_eq!(count_agents(&conn, "1=1"), 1, "must not double-insert");
1081    }
1082}
agentmux_srv/backend/storage/agents_consolidate.rs

agentmux_srv/backend/storage/
agents_consolidate.rs
