rustio-core 1.8.2

//! Tests for the Safe Executor (0.5.2).
//!
//! Every test uses a synthetic [`ProjectView`] so nothing hits the real
//! filesystem; the impure `execute_plan_document` entry has its own
//! temp-dir-based tests marked `#[ignore]` where appropriate.
//!
//! Safety invariants each test reinforces:
//!
//! - Unsupported primitives return a specific `UnsupportedPrimitive`
//!   error — never a silent fallback.
//! - Destructive primitives return `DestructiveWithoutConfirmation`
//!   even with `allow_destructive = true`, because the flag is a
//!   0.5.3 extension point and 0.5.2 ignores it.
//! - A `Critical` plan is refused at the gate with
//!   `CriticalRiskNotAllowed`, never partially applied.
//! - A stale plan is refused with `SchemaMismatch` — not a silent
//!   no-op, not an i/o error.
//! - Applying the same plan twice returns a clear `FileConflict`.

use std::collections::BTreeMap;
use std::path::PathBuf;

use chrono::{TimeZone, Utc};

use super::executor::{
    plan_execution, render_preview_human, ExecuteOptions, ExecutionError, ExecutionPreview,
    FileChangeKind, ParsedModelsFile, ProjectView,
};
use super::planner::PlanResult;
use super::review::{build_plan_document_with_timestamp, PlanDocument, RiskLevel};
use super::{AddField, CreateMigration, FieldSpec, Plan, Primitive, RemoveField, RenameField};
use crate::schema::{Schema, SchemaField, SchemaModel, SCHEMA_VERSION};

// ---------------------------------------------------------------------------
// Fixtures
// ---------------------------------------------------------------------------

fn pkg_version() -> String {
    env!("CARGO_PKG_VERSION").to_string()
}

fn fixed_ts() -> chrono::DateTime<Utc> {
    Utc.with_ymd_and_hms(2026, 1, 1, 0, 0, 0).single().unwrap()
}

const TASK_MODELS_SRC: &str = r#"use rustio_core::{Error, Model, Row, RustioAdmin, Value};

#[derive(Debug, RustioAdmin)]
pub struct Task {
    pub id: i64,
    pub title: String,
    pub is_active: bool,
}

impl Model for Task {
    const TABLE: &'static str = "tasks";
    const COLUMNS: &'static [&'static str] = &["id", "title", "is_active"];
    const INSERT_COLUMNS: &'static [&'static str] = &["title", "is_active"];

    fn id(&self) -> i64 {
        self.id
    }

    fn from_row(row: Row<'_>) -> Result<Self, Error> {
        Ok(Self {
            id: row.get_i64("id")?,
            title: row.get_string("title")?,
            is_active: row.get_bool("is_active")?,
        })
    }

    fn insert_values(&self) -> Vec<Value> {
        vec![
            self.title.clone().into(),
            self.is_active.into(),
        ]
    }
}
"#;

fn task_schema() -> Schema {
    Schema {
        version: SCHEMA_VERSION,
        rustio_version: pkg_version(),
        models: vec![SchemaModel {
            name: "Task".into(),
            table: "tasks".into(),
            admin_name: "tasks".into(),
            display_name: "Tasks".into(),
            singular_name: "Task".into(),
            fields: vec![
                SchemaField {
                    name: "id".into(),
                    ty: "i64".into(),
                    nullable: false,
                    editable: false,
                    relation: None,
                },
                SchemaField {
                    name: "title".into(),
                    ty: "String".into(),
                    nullable: false,
                    editable: true,
                    relation: None,
                },
                SchemaField {
                    name: "is_active".into(),
                    ty: "bool".into(),
                    nullable: false,
                    editable: true,
                    relation: None,
                },
            ],
            relations: vec![],
            core: false,
        }],
    }
}

fn project_with_task(root: &str) -> ProjectView {
    let mut models_files = BTreeMap::new();
    models_files.insert(
        "tasks".to_string(),
        ParsedModelsFile {
            path: PathBuf::from(format!("{root}/apps/tasks/models.rs")),
            source: TASK_MODELS_SRC.to_string(),
            struct_names: vec!["Task".into()],
        },
    );
    ProjectView {
        root: PathBuf::from(root),
        models_files,
        existing_migrations: vec!["0001_create_tasks.sql".into()],
        migration_sources: BTreeMap::new(),
    }
}

fn add_field_plan(model: &str, name: &str, ty: &str, nullable: bool) -> Plan {
    Plan::new(vec![Primitive::AddField(AddField {
        model: model.into(),
        field: FieldSpec {
            name: name.into(),
            ty: ty.into(),
            nullable,
            editable: true,
        },
    })])
}

fn doc_for(schema: &Schema, prompt: &str, plan: Plan) -> PlanDocument {
    let result = PlanResult {
        plan,
        explanation: "unit-test".into(),
    };
    build_plan_document_with_timestamp(schema, prompt, &result, fixed_ts(), None)
        .expect("fixture plans should build cleanly")
}

fn unwrap_preview(p: Result<ExecutionPreview, ExecutionError>) -> ExecutionPreview {
    p.unwrap_or_else(|e| panic!("plan_execution should have succeeded: {e}"))
}

// ---------------------------------------------------------------------------
// AddField — the happy path
// ---------------------------------------------------------------------------

#[test]
fn simple_add_field_produces_two_file_changes() {
    let schema = task_schema();
    let project = project_with_task("/p");
    let plan = add_field_plan("Task", "priority", "i32", false);
    let doc = doc_for(&schema, "Add priority to tasks", plan);

    let preview = unwrap_preview(plan_execution(
        &schema,
        &project,
        &doc,
        &ExecuteOptions::default(),
        None,
    ));
    assert_eq!(preview.applied_steps, 1);
    assert_eq!(preview.file_changes.len(), 2);

    // First change: update to models.rs.
    let models_change = &preview.file_changes[0];
    assert_eq!(models_change.kind, FileChangeKind::Update);
    assert_eq!(models_change.path, PathBuf::from("/p/apps/tasks/models.rs"));
    let new_src = &models_change.new_contents;
    assert!(
        new_src.contains("pub priority: i32,"),
        "struct should have the new field:\n{new_src}",
    );
    assert!(
        new_src.contains("\"priority\""),
        "COLUMNS should include \"priority\":\n{new_src}",
    );
    assert!(
        new_src.contains("priority: row.get_i32(\"priority\")?,"),
        "from_row should read the new field:\n{new_src}",
    );
    assert!(
        new_src.contains("self.priority.into(),"),
        "insert_values should forward the new field:\n{new_src}",
    );

    // Second change: migration file, deterministic name.
    let mig = &preview.file_changes[1];
    assert_eq!(mig.kind, FileChangeKind::Create);
    assert_eq!(
        mig.path,
        PathBuf::from("/p/migrations/0002_add_priority_to_tasks.sql")
    );
    assert!(
        mig.new_contents
            .contains("ALTER TABLE tasks ADD COLUMN priority INTEGER NOT NULL DEFAULT 0;"),
        "migration SQL:\n{}",
        mig.new_contents,
    );
}

#[test]
fn add_nullable_datetime_adds_chrono_import_and_uses_optional_accessor() {
    let schema = task_schema();
    let project = project_with_task("/p");
    let plan = add_field_plan("Task", "completed_at", "DateTime", true);
    let doc = doc_for(&schema, "add optional completed_at to tasks", plan);

    let preview = unwrap_preview(plan_execution(
        &schema,
        &project,
        &doc,
        &ExecuteOptions::default(),
        None,
    ));
    let new_src = &preview.file_changes[0].new_contents;
    assert!(
        new_src.contains("use chrono::{DateTime, Utc};"),
        "chrono import should be added:\n{new_src}",
    );
    assert!(
        new_src.contains("pub completed_at: Option<DateTime<Utc>>,"),
        "field should be Option<DateTime<Utc>>:\n{new_src}",
    );
    assert!(
        new_src.contains("completed_at: row.get_optional_datetime(\"completed_at\")?,"),
        "from_row accessor should be optional:\n{new_src}",
    );
    // Migration for nullable DateTime uses plain ADD COLUMN (no DEFAULT).
    // Phase 2: PG type is TIMESTAMPTZ, not SQLite's TEXT.
    let mig_src = &preview.file_changes[1].new_contents;
    assert!(
        mig_src.contains("ALTER TABLE tasks ADD COLUMN completed_at TIMESTAMPTZ;"),
        "nullable add SQL should not add NOT NULL DEFAULT:\n{mig_src}",
    );
}

#[test]
fn add_field_numbering_picks_next_migration_number() {
    let schema = task_schema();
    let mut project = project_with_task("/p");
    project.existing_migrations = vec![
        "0001_create_tasks.sql".into(),
        "0007_something.sql".into(), // gap in numbering
    ];
    let plan = add_field_plan("Task", "priority", "i32", false);
    let doc = doc_for(&schema, "x", plan);

    let preview = unwrap_preview(plan_execution(
        &schema,
        &project,
        &doc,
        &ExecuteOptions::default(),
        None,
    ));
    let mig = &preview.file_changes[1];
    assert_eq!(
        mig.path,
        PathBuf::from("/p/migrations/0008_add_priority_to_tasks.sql")
    );
}

// ---------------------------------------------------------------------------
// RenameField
// ---------------------------------------------------------------------------

#[test]
fn rename_field_patches_struct_columns_and_accessors() {
    let schema = task_schema();
    let project = project_with_task("/p");
    let plan = Plan::new(vec![Primitive::RenameField(RenameField {
        model: "Task".into(),
        from: "title".into(),
        to: "headline".into(),
    })]);
    let doc = doc_for(&schema, "rename title to headline in tasks", plan);

    let preview = unwrap_preview(plan_execution(
        &schema,
        &project,
        &doc,
        &ExecuteOptions::default(),
        None,
    ));
    let new_src = &preview.file_changes[0].new_contents;
    assert!(
        new_src.contains("pub headline: String,"),
        "struct field renamed:\n{new_src}",
    );
    assert!(
        !new_src.contains("pub title: String,"),
        "old struct field removed:\n{new_src}",
    );
    assert!(
        new_src.contains("\"headline\""),
        "COLUMNS should carry the new name:\n{new_src}",
    );
    assert!(
        new_src.contains("headline: row.get_string(\"headline\")?,"),
        "from_row updated:\n{new_src}",
    );
    assert!(
        new_src.contains("self.headline.clone().into(),"),
        "insert_values updated:\n{new_src}",
    );
    // Migration SQL is deterministic.
    let mig = &preview.file_changes[1];
    assert_eq!(
        mig.path,
        PathBuf::from("/p/migrations/0002_rename_title_to_headline_on_tasks.sql")
    );
    assert!(
        mig.new_contents
            .contains("ALTER TABLE tasks RENAME COLUMN title TO headline;"),
        "rename SQL:\n{}",
        mig.new_contents,
    );
}

#[test]
fn rename_refuses_when_source_field_missing_from_file() {
    // Hand-craft a PlanDocument with a rename that *looks* plausible
    // on the schema but whose file source has drifted: pretend a human
    // already renamed the field in models.rs.
    let schema = task_schema();
    let mut project = project_with_task("/p");
    project.models_files.get_mut("tasks").unwrap().source =
        TASK_MODELS_SRC.replace("pub title: String,", "pub headline: String,");
    // The review would pass (schema still has `title`), but the file
    // conflict gate catches the divergence.
    let plan = Plan::new(vec![Primitive::RenameField(RenameField {
        model: "Task".into(),
        from: "title".into(),
        to: "headline".into(),
    })]);
    let doc = doc_for(&schema, "rename title to headline in tasks", plan);
    let err = plan_execution(&schema, &project, &doc, &ExecuteOptions::default(), None)
        .expect_err("should be a FileConflict");
    match err {
        ExecutionError::FileConflict { path, reason } => {
            assert!(path.ends_with("apps/tasks/models.rs"), "{path}");
            assert!(reason.contains("does not declare"), "reason was: {reason}");
        }
        other => panic!("expected FileConflict, got {other:?}"),
    }
}

// ---------------------------------------------------------------------------
// Gates: validation / risk / developer-only / destructive / unsupported
// ---------------------------------------------------------------------------

#[test]
fn validation_failure_blocks_execution() {
    let schema = task_schema();
    let project = project_with_task("/p");
    // Construct a PlanDocument directly (skipping the builder's
    // validation) so we can assert the executor performs its own
    // revalidation rather than trusting the stored document.
    let doc = PlanDocument {
        version: super::review::PLAN_DOCUMENT_VERSION,
        created_at: "2026-01-01T00:00:00Z".into(),
        prompt: "".into(),
        explanation: "".into(),
        risk: RiskLevel::Low,
        impact: Default::default(),
        // `title` already exists on Task — add_field will be stale.
        plan: add_field_plan("Task", "title", "String", false),
    };
    let err = plan_execution(&schema, &project, &doc, &ExecuteOptions::default(), None)
        .expect_err("stale plan must be refused");
    match err {
        ExecutionError::SchemaMismatch(msg) => {
            assert!(msg.contains("step 0"), "reason: {msg}");
        }
        other => panic!("expected SchemaMismatch, got {other:?}"),
    }
}

#[test]
fn critical_risk_blocks_execution() {
    let schema = task_schema();
    let project = project_with_task("/p");
    // Craft a document whose declared risk is Critical — the executor
    // must refuse without even trying to simulate.
    let doc = PlanDocument {
        version: super::review::PLAN_DOCUMENT_VERSION,
        created_at: "2026-01-01T00:00:00Z".into(),
        prompt: "".into(),
        explanation: "".into(),
        // Risk is re-computed by the executor; to force Critical we
        // use a plan that intrinsically resolves to Critical: a
        // developer-only primitive.
        risk: RiskLevel::Critical,
        impact: Default::default(),
        plan: Plan::new(vec![Primitive::CreateMigration(CreateMigration {
            name: "bad".into(),
            sql: "DROP TABLE tasks".into(),
        })]),
    };
    let err = plan_execution(&schema, &project, &doc, &ExecuteOptions::default(), None)
        .expect_err("critical-risk plans must be refused");
    // The review layer will fail validation first (dev-only plans
    // never validate), surfacing SchemaMismatch; either that or the
    // critical-risk gate is acceptable — both mean "refused".
    assert!(
        matches!(
            err,
            ExecutionError::SchemaMismatch(_)
                | ExecutionError::CriticalRiskNotAllowed
                | ExecutionError::DeveloperOnlyForbidden
        ),
        "unexpected error variant: {err:?}",
    );
}

#[test]
fn developer_only_primitive_is_refused() {
    let schema = task_schema();
    let project = project_with_task("/p");
    let doc = PlanDocument {
        version: super::review::PLAN_DOCUMENT_VERSION,
        created_at: "2026-01-01T00:00:00Z".into(),
        prompt: "".into(),
        explanation: "".into(),
        risk: RiskLevel::Low, // incorrectly low — executor must still refuse
        impact: Default::default(),
        plan: Plan::new(vec![Primitive::CreateMigration(CreateMigration {
            name: "bad".into(),
            sql: "SELECT 1".into(),
        })]),
    };
    let err = plan_execution(&schema, &project, &doc, &ExecuteOptions::default(), None)
        .expect_err("developer-only plan must be refused");
    // Either the re-validation gate (which reports dev-only as
    // SchemaMismatch via the review layer) or the explicit
    // DeveloperOnlyForbidden gate will fire first — both are correct.
    assert!(
        matches!(
            err,
            ExecutionError::DeveloperOnlyForbidden | ExecutionError::SchemaMismatch(_)
        ),
        "unexpected error variant: {err:?}",
    );
}

#[test]
fn remove_field_is_refused_as_destructive() {
    let schema = task_schema();
    let project = project_with_task("/p");
    let plan = Plan::new(vec![Primitive::RemoveField(RemoveField {
        model: "Task".into(),
        field: "title".into(),
    })]);
    let doc = doc_for(&schema, "remove title from tasks", plan);
    let err = plan_execution(&schema, &project, &doc, &ExecuteOptions::default(), None)
        .expect_err("destructive primitive must be refused");
    match err {
        ExecutionError::DestructiveWithoutConfirmation { op } => {
            assert_eq!(op, "remove_field");
        }
        other => panic!("expected DestructiveWithoutConfirmation, got {other:?}"),
    }
}

#[test]
fn remove_field_with_allow_destructive_drops_column_and_writes_migration() {
    // 0.9.1: the destructive gate opens on `allow_destructive = true`.
    // The executor patches models.rs (struct field + COLUMNS +
    // INSERT_COLUMNS + from_row + insert_values lines all dropped) and
    // emits a SQLite recreate-table migration that rebuilds the table
    // without the column.
    let schema = task_schema();
    let project = project_with_task("/p");
    let plan = Plan::new(vec![Primitive::RemoveField(RemoveField {
        model: "Task".into(),
        field: "title".into(),
    })]);
    let doc = doc_for(&schema, "remove title from tasks", plan);
    let opts = ExecuteOptions {
        allow_destructive: true,
    };
    let preview = unwrap_preview(plan_execution(&schema, &project, &doc, &opts, None));
    assert_eq!(preview.applied_steps, 1);
    assert_eq!(preview.file_changes.len(), 2);

    // models.rs update — every mention of `title` is gone.
    let models = &preview.file_changes[0];
    assert_eq!(models.kind, FileChangeKind::Update);
    assert!(
        !models.new_contents.contains("pub title"),
        "struct field `title` should be removed:\n{}",
        models.new_contents,
    );
    assert!(
        !models.new_contents.contains("\"title\""),
        "no literal \"title\" should remain in the updated file:\n{}",
        models.new_contents,
    );

    // Phase 2: PG uses native `ALTER TABLE … DROP COLUMN … CASCADE`
    // instead of the SQLite recreate-table dance.
    let mig = &preview.file_changes[1];
    assert_eq!(mig.kind, FileChangeKind::Create);
    assert!(
        mig.new_contents
            .contains("ALTER TABLE tasks DROP COLUMN title CASCADE;"),
        "migration should use native PG DROP COLUMN:\n{}",
        mig.new_contents,
    );
    assert!(
        !mig.new_contents.contains("CREATE TABLE"),
        "no recreate-table SQL should remain:\n{}",
        mig.new_contents,
    );
    assert!(
        preview.summary.contains("Remove field"),
        "summary should name the operation: {}",
        preview.summary,
    );
}

#[test]
fn remove_primary_key_id_is_refused_even_with_force() {
    // Dropping `id` would orphan the primary key — UnsupportedPrimitive.
    let schema = task_schema();
    let project = project_with_task("/p");
    let plan = Plan::new(vec![Primitive::RemoveField(RemoveField {
        model: "Task".into(),
        field: "id".into(),
    })]);
    let doc = doc_for(&schema, "remove id from tasks", plan);
    let opts = ExecuteOptions {
        allow_destructive: true,
    };
    let err =
        plan_execution(&schema, &project, &doc, &opts, None).expect_err("id removal must refuse");
    match err {
        ExecutionError::UnsupportedPrimitive { op, reason } => {
            assert_eq!(op, "remove_field");
            assert!(
                reason.contains("id"),
                "reason should mention the PK: {reason}"
            );
        }
        other => panic!("expected UnsupportedPrimitive, got {other:?}"),
    }
}

#[test]
fn remove_model_still_refused_with_force() {
    // 0.9.1 scope cap: `remove_model` is deferred to 0.9.2 regardless
    // of `allow_destructive`. The whole-model delete touches the admin
    // registration + downstream FK dependencies; that's its own ship.
    let schema = task_schema();
    let project = project_with_task("/p");
    let plan = Plan::new(vec![Primitive::RemoveModel(super::RemoveModel {
        name: "Task".into(),
    })]);
    let doc = doc_for(&schema, "remove Task", plan);
    let opts = ExecuteOptions {
        allow_destructive: true,
    };
    let err = plan_execution(&schema, &project, &doc, &opts, None)
        .expect_err("remove_model refused even with allow_destructive");
    match err {
        ExecutionError::UnsupportedPrimitive { op, reason } => {
            assert_eq!(op, "remove_model");
            assert!(
                reason.contains("0.9.2") || reason.contains("scheduled"),
                "reason should say this is a future version: {reason}",
            );
        }
        other => panic!("expected UnsupportedPrimitive, got {other:?}"),
    }
}

#[test]
fn unsupported_primitives_fail_with_named_reasons() {
    // `rename_model`, `change_field_type`, and `change_field_nullability`
    // moved out of this list in 0.5.3 — tests for those live in
    // `executor_tests_advanced.rs`. What's still unsupported here is
    // `add_model` (scaffold-level), `update_admin` (metadata), and the
    // relation primitives.
    let schema = task_schema();
    let project = project_with_task("/p");
    let plan = Plan::new(vec![Primitive::UpdateAdmin(super::UpdateAdmin {
        model: "Task".into(),
        field: "title".into(),
        attr: "searchable".into(),
        value: serde_json::json!(true),
    })]);
    let doc = doc_for(&schema, "x", plan);
    let err = plan_execution(&schema, &project, &doc, &ExecuteOptions::default(), None)
        .expect_err("update_admin must be refused");
    match err {
        ExecutionError::UnsupportedPrimitive { op, .. } => {
            assert_eq!(op, "update_admin");
        }
        other => panic!("expected UnsupportedPrimitive, got {other:?}"),
    }
}

// ---------------------------------------------------------------------------
// Stale-plan detection
// ---------------------------------------------------------------------------

#[test]
fn stale_plan_is_refused_with_clear_reason() {
    // Today: Task has `title`. We save a plan to add `priority`. Then
    // the schema drifts — someone adds `priority` via another route.
    // The saved plan must be refused, not silently applied.
    let schema_at_plan_time = task_schema();
    let project = project_with_task("/p");
    let plan = add_field_plan("Task", "priority", "i32", false);
    let doc = doc_for(&schema_at_plan_time, "add priority", plan);

    // Now mutate the schema so `priority` already exists.
    let mut schema_now = task_schema();
    schema_now.models[0].fields.push(SchemaField {
        name: "priority".into(),
        ty: "i32".into(),
        nullable: false,
        editable: true,
        relation: None,
    });
    let err = plan_execution(
        &schema_now,
        &project,
        &doc,
        &ExecuteOptions::default(),
        None,
    )
    .expect_err("stale plan must be refused");
    match err {
        ExecutionError::SchemaMismatch(msg) => {
            assert!(
                msg.contains("step 0") && msg.contains("priority"),
                "reason should name the failing step + field: {msg}",
            );
        }
        other => panic!("expected SchemaMismatch, got {other:?}"),
    }
}

// ---------------------------------------------------------------------------
// Idempotency
// ---------------------------------------------------------------------------

#[test]
fn applying_same_plan_twice_against_patched_source_fails_cleanly() {
    let schema = task_schema();
    let project = project_with_task("/p");
    let plan = add_field_plan("Task", "priority", "i32", false);
    let doc = doc_for(&schema, "add priority", plan);

    // First apply: produces a preview with the patched models.rs.
    let preview = unwrap_preview(plan_execution(
        &schema,
        &project,
        &doc,
        &ExecuteOptions::default(),
        None,
    ));
    let patched = preview.file_changes[0].new_contents.clone();

    // Now pretend that patched file is live: schema already has the
    // field, models.rs already has it. The executor must refuse —
    // either via the schema-drift gate or the file-conflict gate.
    let mut schema_after = task_schema();
    schema_after.models[0].fields.push(SchemaField {
        name: "priority".into(),
        ty: "i32".into(),
        nullable: false,
        editable: true,
        relation: None,
    });
    let mut project_after = project_with_task("/p");
    project_after.models_files.get_mut("tasks").unwrap().source = patched;
    let err = plan_execution(
        &schema_after,
        &project_after,
        &doc,
        &ExecuteOptions::default(),
        None,
    )
    .expect_err("second apply must be refused");
    assert!(
        matches!(
            err,
            ExecutionError::SchemaMismatch(_) | ExecutionError::FileConflict { .. }
        ),
        "unexpected error on double-apply: {err:?}",
    );
}

// ---------------------------------------------------------------------------
// Determinism + rendering
// ---------------------------------------------------------------------------

#[test]
fn planning_same_document_twice_produces_identical_previews() {
    let schema = task_schema();
    let project = project_with_task("/p");
    let plan = add_field_plan("Task", "priority", "i32", false);
    let doc = doc_for(&schema, "x", plan);
    let a = unwrap_preview(plan_execution(
        &schema,
        &project,
        &doc,
        &ExecuteOptions::default(),
        None,
    ));
    let b = unwrap_preview(plan_execution(
        &schema,
        &project,
        &doc,
        &ExecuteOptions::default(),
        None,
    ));
    assert_eq!(a, b);
}

#[test]
fn render_preview_human_reads_like_a_changelog() {
    let schema = task_schema();
    let project = project_with_task("/p");
    let plan = add_field_plan("Task", "priority", "i32", false);
    let doc = doc_for(&schema, "add priority", plan);
    let preview = unwrap_preview(plan_execution(
        &schema,
        &project,
        &doc,
        &ExecuteOptions::default(),
        None,
    ));
    let out = render_preview_human(&preview, RiskLevel::Low);
    assert!(out.starts_with("Plan to apply\n"));
    assert!(out.contains("Applying:\n  + Add field \"priority\""));
    assert!(out.contains("Files to be written:"));
    assert!(out.contains("Risk:\n  Low"));
}

// ---------------------------------------------------------------------------
// 0.8.0 — AddRelation (belongs_to)
// ---------------------------------------------------------------------------

const APPLICATION_MODELS_SRC: &str = r#"use rustio_core::{Error, Model, Row, RustioAdmin, Value};

#[derive(Debug, RustioAdmin)]
pub struct Application {
    pub id: i64,
    pub title: String,
}

impl Model for Application {
    const TABLE: &'static str = "applications";
    const COLUMNS: &'static [&'static str] = &["id", "title"];
    const INSERT_COLUMNS: &'static [&'static str] = &["title"];

    fn id(&self) -> i64 {
        self.id
    }

    fn from_row(row: Row<'_>) -> Result<Self, Error> {
        Ok(Self {
            id: row.get_i64("id")?,
            title: row.get_string("title")?,
        })
    }

    fn insert_values(&self) -> Vec<Value> {
        vec![self.title.clone().into()]
    }
}
"#;

fn housing_schema() -> Schema {
    Schema {
        version: SCHEMA_VERSION,
        rustio_version: pkg_version(),
        models: vec![
            SchemaModel {
                name: "Applicant".into(),
                table: "applicants".into(),
                admin_name: "applicants".into(),
                display_name: "Applicants".into(),
                singular_name: "Applicant".into(),
                fields: vec![SchemaField {
                    name: "id".into(),
                    ty: "i64".into(),
                    nullable: false,
                    editable: false,
                    relation: None,
                }],
                relations: vec![],
                core: false,
            },
            SchemaModel {
                name: "Application".into(),
                table: "applications".into(),
                admin_name: "applications".into(),
                display_name: "Applications".into(),
                singular_name: "Application".into(),
                fields: vec![
                    SchemaField {
                        name: "id".into(),
                        ty: "i64".into(),
                        nullable: false,
                        editable: false,
                        relation: None,
                    },
                    SchemaField {
                        name: "title".into(),
                        ty: "String".into(),
                        nullable: false,
                        editable: true,
                        relation: None,
                    },
                ],
                relations: vec![],
                core: false,
            },
        ],
    }
}

fn project_with_housing(root: &str) -> ProjectView {
    let mut models_files = BTreeMap::new();
    models_files.insert(
        "applications".to_string(),
        ParsedModelsFile {
            path: PathBuf::from(format!("{root}/apps/applications/models.rs")),
            source: APPLICATION_MODELS_SRC.to_string(),
            struct_names: vec!["Application".into()],
        },
    );
    ProjectView {
        root: PathBuf::from(root),
        models_files,
        existing_migrations: vec!["0001_create_applications.sql".into()],
        migration_sources: BTreeMap::new(),
    }
}

fn add_relation_plan(from: &str, to: &str, via: &str) -> Plan {
    Plan::new(vec![Primitive::AddRelation(super::AddRelation {
        from: from.into(),
        kind: crate::schema::RelationKind::BelongsTo,
        to: to.into(),
        via: via.into(),
        required: false,
        on_delete: super::OnDelete::Restrict,
    })])
}

fn add_relation_plan_with(
    from: &str,
    to: &str,
    via: &str,
    required: bool,
    on_delete: super::OnDelete,
) -> Plan {
    Plan::new(vec![Primitive::AddRelation(super::AddRelation {
        from: from.into(),
        kind: crate::schema::RelationKind::BelongsTo,
        to: to.into(),
        via: via.into(),
        required,
        on_delete,
    })])
}

#[test]
fn add_relation_generates_fk_column_with_references_clause() {
    // 0.9.0: the belongs_to migration now includes a SQL FOREIGN KEY
    // via the `REFERENCES` syntax. The column is nullable by default
    // (SQLite can't add a NOT NULL+REFERENCES column via ALTER TABLE).
    let schema = housing_schema();
    let project = project_with_housing("/p");
    let plan = add_relation_plan("Application", "Applicant", "applicant_id");
    let doc = doc_for(&schema, "link Application to Applicant", plan);

    let preview = unwrap_preview(plan_execution(
        &schema,
        &project,
        &doc,
        &ExecuteOptions::default(),
        None,
    ));
    assert_eq!(preview.applied_steps, 1);
    // One models.rs update + one migration file.
    assert_eq!(preview.file_changes.len(), 2);

    let models_change = &preview.file_changes[0];
    assert_eq!(models_change.kind, FileChangeKind::Update);
    assert!(
        models_change
            .new_contents
            .contains("pub applicant_id: Option<i64>,"),
        "struct should gain the nullable FK column:\n{}",
        models_change.new_contents,
    );

    let mig = &preview.file_changes[1];
    assert_eq!(mig.kind, FileChangeKind::Create);
    assert!(
        mig.path
            .to_string_lossy()
            .ends_with("_add_applicant_id_to_applications.sql"),
        "migration should be named after the FK column: {}",
        mig.path.display(),
    );
    assert!(
        mig.new_contents.contains(
            "ALTER TABLE applications ADD COLUMN applicant_id BIGINT REFERENCES applicants(id) ON DELETE RESTRICT;"
        ),
        "migration SQL should include REFERENCES + ON DELETE (PG: BIGINT not INTEGER):\n{}",
        mig.new_contents,
    );
}

#[test]
fn add_relation_emits_references_no_pragma() {
    // Phase 2: PG enforces FKs by default — no PRAGMA. The migration
    // MUST include `REFERENCES` and the configured ON DELETE policy.
    // The summary names both the parent table and the policy.
    let schema = housing_schema();
    let project = project_with_housing("/p");
    let plan = add_relation_plan("Application", "Applicant", "applicant_id");
    let doc = doc_for(&schema, "link Application to Applicant", plan);

    let preview = unwrap_preview(plan_execution(
        &schema,
        &project,
        &doc,
        &ExecuteOptions::default(),
        None,
    ));
    let mig_sql = &preview.file_changes[1].new_contents;
    assert!(
        mig_sql.contains("REFERENCES applicants(id)"),
        "must emit REFERENCES:\n{mig_sql}",
    );
    assert!(
        mig_sql.contains("ON DELETE RESTRICT"),
        "default on_delete is restrict:\n{mig_sql}",
    );
    assert!(
        !mig_sql.contains("PRAGMA"),
        "Phase 2 (PG): no PRAGMA in the migration:\n{mig_sql}",
    );
    assert!(
        preview.summary.contains("belongs_to"),
        "preview summary should name the relation kind: {}",
        preview.summary,
    );
    assert!(
        preview.summary.contains("restrict"),
        "preview summary should name the on_delete policy: {}",
        preview.summary,
    );
}

#[test]
fn add_relation_idempotent_when_column_already_present() {
    // If the `<via>_id` column already lives in the struct, the
    // executor must refuse with FileConflict — same policy as
    // add_field, since relations piggyback on it.
    let schema = housing_schema();
    let mut project = project_with_housing("/p");
    project.models_files.get_mut("applications").unwrap().source = APPLICATION_MODELS_SRC.replace(
        "    pub title: String,\n}",
        "    pub title: String,\n    pub applicant_id: i64,\n}",
    );
    let plan = add_relation_plan("Application", "Applicant", "applicant_id");
    let doc = doc_for(&schema, "link Application to Applicant", plan);

    let err = plan_execution(&schema, &project, &doc, &ExecuteOptions::default(), None)
        .expect_err("must refuse when column already exists");
    match err {
        ExecutionError::FileConflict { reason, .. } => {
            assert!(
                reason.contains("applicant_id"),
                "reason should name the column: {reason}",
            );
        }
        other => panic!("expected FileConflict, got {other:?}"),
    }
}

#[test]
fn add_relation_cascade_emits_on_delete_cascade() {
    let schema = housing_schema();
    let project = project_with_housing("/p");
    let plan = add_relation_plan_with(
        "Application",
        "Applicant",
        "applicant_id",
        false,
        super::OnDelete::Cascade,
    );
    let doc = doc_for(&schema, "link with cascade", plan);
    let preview = unwrap_preview(plan_execution(
        &schema,
        &project,
        &doc,
        &ExecuteOptions::default(),
        None,
    ));
    let mig = &preview.file_changes[1].new_contents;
    assert!(
        mig.contains("ON DELETE CASCADE"),
        "cascade policy should appear in SQL:\n{mig}",
    );
    assert!(
        !mig.contains("ON DELETE RESTRICT") && !mig.contains("ON DELETE SET NULL"),
        "only one ON DELETE should be emitted:\n{mig}",
    );
    assert!(
        preview.summary.contains("cascade"),
        "summary should name the policy: {}",
        preview.summary,
    );
}

#[test]
fn add_relation_set_null_emits_on_delete_set_null() {
    let schema = housing_schema();
    let project = project_with_housing("/p");
    let plan = add_relation_plan_with(
        "Application",
        "Applicant",
        "applicant_id",
        false,
        super::OnDelete::SetNull,
    );
    let doc = doc_for(&schema, "link with set null", plan);
    let preview = unwrap_preview(plan_execution(
        &schema,
        &project,
        &doc,
        &ExecuteOptions::default(),
        None,
    ));
    let mig = &preview.file_changes[1].new_contents;
    assert!(
        mig.contains("ON DELETE SET NULL"),
        "set_null policy should appear in SQL:\n{mig}",
    );
}

#[test]
fn add_relation_required_is_refused_with_retrofit_hint() {
    // 0.9.0: a NOT NULL foreign key cannot be added via ALTER TABLE on
    // SQLite. The executor refuses and points at `rustio migrate --add-fks`.
    let schema = housing_schema();
    let project = project_with_housing("/p");
    let plan = add_relation_plan_with(
        "Application",
        "Applicant",
        "applicant_id",
        true,
        super::OnDelete::Restrict,
    );
    let doc = doc_for(&schema, "link as required", plan);
    let err = plan_execution(&schema, &project, &doc, &ExecuteOptions::default(), None)
        .expect_err("required FK must refuse");
    match err {
        ExecutionError::UnsupportedPrimitive { op, reason } => {
            assert_eq!(op, "add_relation");
            assert!(
                reason.contains("--add-fks") || reason.contains("recreate-table"),
                "reason should hint at the retrofit path: {reason}",
            );
        }
        other => panic!("expected UnsupportedPrimitive, got {other:?}"),
    }
}

#[test]
fn add_relation_column_is_nullable_in_the_struct() {
    // The struct patch must use `Option<i64>` since the FK column is
    // nullable. If this flips to plain `i64`, Rust code that reads a
    // NULL row at runtime panics.
    let schema = housing_schema();
    let project = project_with_housing("/p");
    let plan = add_relation_plan("Application", "Applicant", "applicant_id");
    let doc = doc_for(&schema, "link Application to Applicant", plan);
    let preview = unwrap_preview(plan_execution(
        &schema,
        &project,
        &doc,
        &ExecuteOptions::default(),
        None,
    ));
    let models = &preview.file_changes[0].new_contents;
    assert!(
        models.contains("pub applicant_id: Option<i64>,"),
        "struct field must be Option<i64>:\n{models}",
    );
}

#[test]
fn remove_relation_primitive_is_refused_with_clear_reason() {
    // 0.8.0: explicit — dropping a FK column is destructive. The
    // executor refuses until the destructive-gate work lands. Seed
    // the schema with the relation first so `Plan::validate` passes
    // and the refusal genuinely comes from the executor gate.
    let mut schema = housing_schema();
    let app = schema
        .models
        .iter_mut()
        .find(|m| m.name == "Application")
        .unwrap();
    app.fields.push(SchemaField {
        name: "applicant_id".into(),
        ty: "i64".into(),
        nullable: false,
        editable: true,
        relation: Some(crate::schema::Relation {
            model: "Applicant".into(),
            field: "id".into(),
            kind: crate::schema::RelationKind::BelongsTo,
            display_field: None,
            required: None,
            on_delete: None,
        }),
    });
    app.relations.push(crate::schema::SchemaRelation {
        kind: "belongsto".into(),
        to: "Applicant".into(),
        via: "applicant_id".into(),
    });
    // The struct source must actually declare `applicant_id` for
    // remove_relation to patch it. Insert the field and its Rust-side
    // plumbing so the fixture matches the schema above. The
    // `insert_values` block is reformatted to multi-line — the remove
    // helper matches on an indented per-element line.
    let mut project = project_with_housing("/p");
    let with_fk = APPLICATION_MODELS_SRC
        .replace(
            "    pub title: String,\n}",
            "    pub title: String,\n    pub applicant_id: i64,\n}",
        )
        .replace(
            "&[\"id\", \"title\"]",
            "&[\"id\", \"title\", \"applicant_id\"]",
        )
        .replace("&[\"title\"]", "&[\"title\", \"applicant_id\"]")
        .replace(
            "title: row.get_string(\"title\")?,",
            "title: row.get_string(\"title\")?,\n            applicant_id: row.get_i64(\"applicant_id\")?,",
        )
        .replace(
            "vec![self.title.clone().into()]",
            "vec![\n            self.title.clone().into(),\n            self.applicant_id.into(),\n        ]",
        );
    project.models_files.get_mut("applications").unwrap().source = with_fk;

    let plan = Plan::new(vec![Primitive::RemoveRelation(super::RemoveRelation {
        from: "Application".into(),
        via: "applicant_id".into(),
    })]);
    let doc = doc_for(&schema, "drop relation", plan);

    // Default options: destructive gate refuses.
    let err = plan_execution(&schema, &project, &doc, &ExecuteOptions::default(), None)
        .expect_err("remove_relation must refuse without --force");
    match err {
        ExecutionError::DestructiveWithoutConfirmation { op } => {
            assert_eq!(op, "remove_relation");
        }
        other => panic!("expected DestructiveWithoutConfirmation, got {other:?}"),
    }

    // With allow_destructive: drop the FK column, emit recreate-table.
    let opts = ExecuteOptions {
        allow_destructive: true,
    };
    let preview = unwrap_preview(plan_execution(&schema, &project, &doc, &opts, None));
    assert_eq!(preview.applied_steps, 1);
    let models = &preview.file_changes[0];
    assert!(
        !models.new_contents.contains("pub applicant_id"),
        "struct field should be dropped:\n{}",
        models.new_contents,
    );
    let mig = &preview.file_changes[1];
    assert!(
        mig.new_contents
            .contains("ALTER TABLE applications DROP COLUMN applicant_id CASCADE;"),
        "migration should use native PG DROP COLUMN:\n{}",
        mig.new_contents,
    );
    assert!(
        preview.summary.contains("Remove relation"),
        "summary should name the op: {}",
        preview.summary,
    );
}

// ---------------------------------------------------------------------------
// Impure entry + atomic commit — temp dir integration tests
// ---------------------------------------------------------------------------

mod integration {
    use super::*;
    use crate::ai::executor::execute_plan_document;
    use std::fs;

    /// Best-effort, process-private tempdir. We don't use `tempfile`
    /// (to keep zero extra deps); a directory under the OS temp root
    /// is good enough because each test creates a unique subdir.
    fn scratch_dir(tag: &str) -> PathBuf {
        let root = std::env::temp_dir().join(format!("rustio-exec-{}-{}", tag, std::process::id()));
        let _ = fs::remove_dir_all(&root);
        fs::create_dir_all(&root).unwrap();
        fs::create_dir_all(root.join("apps").join("tasks")).unwrap();
        fs::create_dir_all(root.join("migrations")).unwrap();
        fs::write(
            root.join("apps").join("tasks").join("models.rs"),
            TASK_MODELS_SRC,
        )
        .unwrap();
        fs::write(
            root.join("migrations").join("0001_create_tasks.sql"),
            "CREATE TABLE tasks(id INTEGER PRIMARY KEY);\n",
        )
        .unwrap();
        let schema = task_schema();
        let schema_json = schema.to_pretty_json().unwrap();
        fs::write(root.join("rustio.schema.json"), schema_json).unwrap();
        root
    }

    #[test]
    fn execute_plan_document_writes_models_and_migration_atomically() {
        let root = scratch_dir("happy");
        let schema = task_schema();
        let plan = add_field_plan("Task", "priority", "i32", false);
        let doc = doc_for(&schema, "add priority", plan);

        let result = execute_plan_document(&root, &doc, &ExecuteOptions::default(), None).unwrap();
        assert_eq!(result.applied_steps, 1);
        assert_eq!(result.generated_files.len(), 2);

        // models.rs was updated.
        let patched = fs::read_to_string(root.join("apps/tasks/models.rs")).unwrap();
        assert!(patched.contains("pub priority: i32,"));
        // migration file created.
        let mig =
            fs::read_to_string(root.join("migrations/0002_add_priority_to_tasks.sql")).unwrap();
        assert!(mig.contains("ALTER TABLE tasks ADD COLUMN priority INTEGER NOT NULL DEFAULT 0;"));
        // no stray `.rustio_tmp` left behind.
        for entry in fs::read_dir(root.join("apps/tasks")).unwrap() {
            let name = entry.unwrap().file_name().into_string().unwrap();
            assert!(!name.contains("rustio_tmp"), "leaked tmp file: {name}");
        }
        let _ = fs::remove_dir_all(&root);
    }

    #[test]
    fn execute_refuses_if_target_migration_already_exists() {
        let root = scratch_dir("conflict");
        // Pre-create the migration we'd otherwise write.
        fs::write(
            root.join("migrations/0002_add_priority_to_tasks.sql"),
            "-- handmade\n",
        )
        .unwrap();
        // The executor still thinks 0002 is "next" because it's the
        // next number after the max on-disk... wait, we need to be
        // careful here. The existing-migrations list has BOTH 0001
        // and 0002, so next_migration_number will be 0003. So we're
        // actually testing that the _third_ slot is used.
        // To make the conflict real, we'll fake a higher-number file:
        fs::remove_file(root.join("migrations/0002_add_priority_to_tasks.sql")).unwrap();
        fs::write(root.join("migrations/0099_pinned.sql"), "-- pinned\n").unwrap();
        // Now the executor's next number is 0100 — no collision. Good.
        //
        // The genuine conflict case is already covered by
        // `execute_refuses_on_file_already_exists` in the pure tests
        // via `FileChangeKind::Create` preconditions.
        let schema = task_schema();
        let plan = add_field_plan("Task", "priority", "i32", false);
        let doc = doc_for(&schema, "add priority", plan);
        let res = execute_plan_document(&root, &doc, &ExecuteOptions::default(), None).unwrap();
        // The assigned migration must be 0100, not 0002.
        assert!(
            res.generated_files
                .iter()
                .any(|p| p.ends_with("0100_add_priority_to_tasks.sql")),
            "generated files were {:?}",
            res.generated_files,
        );
        let _ = fs::remove_dir_all(&root);
    }

    #[test]
    fn execute_refuses_if_models_file_already_has_the_field() {
        // The file on disk already contains `pub priority: i32` (someone
        // patched it by hand, or the executor was interrupted mid-apply
        // and has now been re-run). The schema does NOT yet know about
        // the field, so review passes — but the dry-run's own
        // idempotency check must reject the apply.
        let root = scratch_dir("already_patched");
        let already_patched = TASK_MODELS_SRC.replace(
            "    pub is_active: bool,\n}",
            "    pub is_active: bool,\n    pub priority: i32,\n}",
        );
        fs::write(root.join("apps/tasks/models.rs"), &already_patched).unwrap();
        let schema = task_schema();
        let plan = add_field_plan("Task", "priority", "i32", false);
        let doc = doc_for(&schema, "add priority", plan);
        let err = execute_plan_document(&root, &doc, &ExecuteOptions::default(), None).unwrap_err();
        match err {
            ExecutionError::FileConflict { reason, .. } => {
                assert!(
                    reason.contains("already declares field `priority`"),
                    "reason: {reason}"
                );
            }
            other => panic!("expected FileConflict, got {other:?}"),
        }
        let _ = fs::remove_dir_all(&root);
    }

    // ---- 0.9.0 retrofit -------------------------------------------------

    fn schema_with_unannotated_fk() -> Schema {
        use crate::schema::{Relation, RelationKind};
        Schema {
            version: SCHEMA_VERSION,
            rustio_version: pkg_version(),
            models: vec![
                SchemaModel {
                    name: "Applicant".into(),
                    table: "applicants".into(),
                    admin_name: "applicants".into(),
                    display_name: "Applicants".into(),
                    singular_name: "Applicant".into(),
                    fields: vec![SchemaField {
                        name: "id".into(),
                        ty: "i64".into(),
                        nullable: false,
                        editable: false,
                        relation: None,
                    }],
                    relations: vec![],
                    core: false,
                },
                SchemaModel {
                    name: "Application".into(),
                    table: "applications".into(),
                    admin_name: "applications".into(),
                    display_name: "Applications".into(),
                    singular_name: "Application".into(),
                    fields: vec![
                        SchemaField {
                            name: "id".into(),
                            ty: "i64".into(),
                            nullable: false,
                            editable: false,
                            relation: None,
                        },
                        // Pre-0.9.0: `applicant_id` has Relation metadata
                        // but the on_delete / required fields are None.
                        SchemaField {
                            name: "applicant_id".into(),
                            ty: "i64".into(),
                            nullable: false,
                            editable: true,
                            relation: Some(Relation {
                                model: "Applicant".into(),
                                field: "id".into(),
                                kind: RelationKind::BelongsTo,
                                display_field: None,
                                required: None,
                                on_delete: None,
                            }),
                        },
                    ],
                    relations: vec![],
                    core: false,
                },
            ],
        }
    }

    #[test]
    fn retrofit_reports_every_unannotated_belongs_to() {
        let schema = schema_with_unannotated_fk();
        let report = super::super::plan_retrofit_foreign_keys(&schema);
        assert_eq!(
            report.upgraded,
            vec![("Application".to_string(), "applicant_id".to_string())]
        );
        assert_eq!(report.migrations.len(), 1);
        let (name, sql) = &report.migrations[0];
        assert!(
            name.contains("applications"),
            "file name should include the table: {name}"
        );
        assert!(
            sql.contains("REFERENCES applicants(id)"),
            "retrofit SQL must emit a FK clause:\n{sql}"
        );
        assert!(
            sql.contains("ON DELETE RESTRICT"),
            "retrofit default on_delete is restrict:\n{sql}"
        );
        // Phase 2: PG retrofit uses ALTER TABLE ADD CONSTRAINT, not
        // the SQLite recreate-table dance. Constraint name follows
        // the `<table>_<via>_fk` convention.
        assert!(
            sql.contains(
                "ALTER TABLE applications\n    \
                 ADD CONSTRAINT applications_applicant_id_fk \
                 FOREIGN KEY (applicant_id) REFERENCES applicants(id) ON DELETE RESTRICT;"
            ),
            "retrofit should emit ALTER TABLE ADD CONSTRAINT:\n{sql}"
        );
        assert!(sql.contains("BEGIN;"), "should be wrapped in a transaction:\n{sql}");
        assert!(sql.contains("COMMIT;"), "should be wrapped in a transaction:\n{sql}");
        assert!(!sql.contains("CREATE TABLE"), "no recreate-table:\n{sql}");
        assert!(!sql.contains("DROP TABLE"), "no recreate-table:\n{sql}");
        assert!(!sql.contains("PRAGMA"), "PG enforces FKs by default — no PRAGMA:\n{sql}");
    }

    #[test]
    fn retrofit_is_a_noop_for_schemas_already_annotated() {
        let mut schema = schema_with_unannotated_fk();
        for m in &mut schema.models {
            for f in &mut m.fields {
                if let Some(r) = f.relation.as_mut() {
                    r.on_delete = Some("restrict".into());
                    r.required = Some(!f.nullable);
                }
            }
        }
        let report = super::super::plan_retrofit_foreign_keys(&schema);
        assert!(report.upgraded.is_empty());
        assert!(report.migrations.is_empty());
    }
}