codexpi/tests/test_autonomous_remediation.py

from __future__ import annotations

import json
import uuid

import pytest

from codexpi.codex_service import INTERRUPTED_WORKER_REASON, CodexManager
from codexpi.database import Database, utcnow
from codexpi.remediation_service import (
    AutonomousRemediationService,
    RemediationConflict,
    normalize_name,
    parse_self_repair_verdict,
    parse_supervision_verdict,
)


def plan_step(target: str = "NASflix") -> dict:
    return {
        "id": "step-one",
        "order": 1,
        "phase": 1,
        "phaseLabel": "Mesurer",
        "target": target,
        "aliases": ["NASFLIX"],
        "title": "Réduire la charge",
        "summary": "Mesurer puis corriger la charge du projet.",
        "risk": "code",
        "prompt": "Inspecter les traitements de fond, appliquer une correction bornée et vérifier par des tests reproductibles. " * 2,
    }


def service_env(tmp_path):
    database = Database(tmp_path / "codexpi.db")
    database.initialize()
    master_path = tmp_path / "master"
    target_path = tmp_path / "nasflix"
    master_path.mkdir()
    target_path.mkdir()
    master_id = str(uuid.uuid4())
    target_id = str(uuid.uuid4())
    database.execute(
        "INSERT INTO projects(id,name,path,owner_id,full_access,execution_group,created_at) VALUES(?,?,?,?,?,?,?)",
        (master_id, "Agent maître", str(master_path), None, 1, "__global_agent__", utcnow()),
    )
    database.execute(
        "INSERT INTO projects(id,name,path,owner_id,full_access,created_at) VALUES(?,?,?,?,?,?)",
        (target_id, "NASflix", str(target_path), "owner-1", 0, utcnow()),
    )
    codex = CodexManager(database, tmp_path / "data", "/bin/true")
    codex._spawn = lambda *args, **kwargs: 4242
    return database, AutonomousRemediationService(database, codex)


def finish_queue(database: Database, queue_id: str, content: str) -> str:
    queue = database.one("SELECT * FROM prompt_queue WHERE id=?", (queue_id,))
    database.execute("UPDATE prompt_queue SET status='done',finished_at=? WHERE id=?", (utcnow(), queue_id))
    database.execute("UPDATE conversations SET status='completed',worker_pid=NULL,updated_at=? WHERE id=?", (utcnow(), queue["conversation_id"]))
    database.execute(
        "INSERT INTO messages(id,conversation_id,role,content,metadata,created_at) VALUES(?,?,?,?,?,?)",
        (str(uuid.uuid4()), queue["conversation_id"], "assistant", content, json.dumps({"queue_id": queue_id, "actions": [], "files": []}), utcnow()),
    )
    return queue_id


def finish_active(database: Database, run_id: str, content: str) -> str:
    run = database.one("SELECT * FROM remediation_runs WHERE id=?", (run_id,))
    return finish_queue(database, run["active_queue_id"], content)


def add_self_project(database: Database, root) -> str:
    path = root / "codexpi-self"
    path.mkdir()
    project_id = str(uuid.uuid4())
    database.execute(
        "INSERT INTO projects(id,name,path,owner_id,full_access,created_at) VALUES(?,?,?,?,?,?)",
        (project_id, "CodexPi (Self)", str(path), "owner-1", 1, utcnow()),
    )
    return project_id


def test_supervision_marker_is_strict_and_accent_insensitive():
    assert normalize_name("Agent Bibliothécaire") == "agentbibliothecaire"
    verdict = parse_supervision_verdict('Rapport\nCODEXPI_SUPERVISION {"decision":"correct","feedback":"Rejouer les tests"}')
    assert verdict == {"decision": "correct", "feedback": "Rejouer les tests"}
    assert parse_supervision_verdict("approuvé sans marqueur") is None


def test_self_repair_marker_is_strict_and_bounded():
    verdict = parse_self_repair_verdict(
        'Rapport\nCODEXPI_SELF_REPAIR {"decision":"fixed","feedback":"Plan corrigé",'
        '"prompt_addendum":"Accepter une maintenance bornée."}'
    )
    assert verdict == {
        "decision": "fixed",
        "feedback": "Plan corrigé",
        "prompt_addendum": "Accepter une maintenance bornée.",
    }
    assert parse_self_repair_verdict('CODEXPI_SELF_REPAIR {"decision":"ignore"}') is None


def test_block_dispatches_codexpi_self_without_replacing_existing_threads(tmp_path):
    database, service = service_env(tmp_path)
    self_project_id = add_self_project(database, tmp_path)
    run = service.start("owner-1", [plan_step()])
    finish_active(database, run["id"], 'CODEXPI_SUPERVISION {"decision":"approve","feedback":"Plan valide"}')
    service.tick()
    target_conversation = service.detail(run["id"], "owner-1")["steps"][0]["conversation_id"]
    finish_active(database, run["id"], "La protection persistante manque.")
    service.tick()
    finish_active(
        database,
        run["id"],
        'CODEXPI_SUPERVISION {"decision":"block","feedback":"Corriger le contrat de reprise"}',
    )
    service.tick()

    blocked = service.detail(run["id"], "owner-1")
    repair = blocked["self_repair"]
    assert blocked["status"] == "BLOCKED"
    assert blocked["steps"][0]["conversation_id"] == target_conversation
    assert repair["status"] == "RUNNING"
    assert repair["project_id"] == self_project_id
    assert repair["conversation_id"] != target_conversation
    queued = database.one("SELECT content FROM prompt_queue WHERE id=?", (repair["queue_id"],))
    assert "Corriger le contrat de reprise" in queued["content"]
    assert "Ne contourne jamais" in queued["content"]
    assert any(event["kind"] == "self_repair.started" for event in blocked["events"])


def test_successful_self_repair_rebases_and_resumes_the_same_step(tmp_path):
    database, service = service_env(tmp_path)
    add_self_project(database, tmp_path)
    run = service.start("owner-1", [plan_step()])
    finish_active(database, run["id"], 'CODEXPI_SUPERVISION {"decision":"approve","feedback":"Plan valide"}')
    service.tick()
    working = service.detail(run["id"], "owner-1")
    target_conversation = working["steps"][0]["conversation_id"]
    finish_active(database, run["id"], "Le plan impose deux contraintes contradictoires.")
    service.tick()
    finish_active(
        database,
        run["id"],
        'CODEXPI_SUPERVISION {"decision":"block","feedback":"Contradiction du plan"}',
    )
    service.tick()
    repair = service.detail(run["id"], "owner-1")["self_repair"]

    finish_queue(
        database,
        repair["queue_id"],
        'Correction et tests terminés.\nCODEXPI_SELF_REPAIR '
        '{"decision":"fixed","feedback":"Le contrat distingue maintenance et perte de données.",'
        '"prompt_addendum":"Une courte maintenance réversible est autorisée sans perte de données."}',
    )
    assert service.tick()["state"] == "self_repair_processed"

    resumed = service.detail(run["id"], "owner-1")
    assert resumed["status"] == "RUNNING"
    assert resumed["stage"] == "AGENT"
    assert resumed["steps"][0]["status"] == "CORRECTING"
    assert resumed["steps"][0]["conversation_id"] == target_conversation
    assert resumed["steps"][0]["attempts"] == 2
    assert "courte maintenance réversible" in resumed["steps"][0]["prompt"]
    assert resumed["self_repair"]["status"] == "SUCCEEDED"
    assert any(event["kind"] == "self_repair.resumed" for event in resumed["events"])


def test_self_repair_can_require_the_owner_without_retrying_the_chantier(tmp_path):
    database, service = service_env(tmp_path)
    add_self_project(database, tmp_path)
    run = service.start("owner-1", [plan_step()])
    finish_active(database, run["id"], 'CODEXPI_SUPERVISION {"decision":"block","feedback":"Secret externe requis"}')
    service.tick()
    repair = service.detail(run["id"], "owner-1")["self_repair"]
    finish_queue(
        database,
        repair["queue_id"],
        'CODEXPI_SELF_REPAIR {"decision":"needs_user","feedback":"Un secret externe doit être fourni",'
        '"prompt_addendum":""}',
    )
    service.tick()

    blocked = service.detail(run["id"], "owner-1")
    assert blocked["status"] == "BLOCKED"
    assert blocked["self_repair"]["status"] == "NEEDS_USER"
    assert blocked["self_repair"]["feedback"] == "Un secret externe doit être fourni"


def test_self_repair_keeps_deployment_approval_blocked_without_completing_the_step(tmp_path):
    database, service = service_env(tmp_path)
    add_self_project(database, tmp_path)
    run = service.start("owner-1", [plan_step(target="Aloe")])
    finish_active(database, run["id"], 'CODEXPI_SUPERVISION {"decision":"approve","feedback":"Plan valide"}')
    service.tick()
    finish_active(database, run["id"], "Les manifests et les contrôles dry-run sont valides.")
    service.tick()
    finish_active(
        database,
        run["id"],
        'CODEXPI_SUPERVISION {"decision":"block","feedback":"Le déploiement et le redémarrage exigent la validation distincte du propriétaire"}',
    )
    service.tick()
    repair = service.detail(run["id"], "owner-1")["self_repair"]

    finish_queue(
        database,
        repair["queue_id"],
        'CODEXPI_SELF_REPAIR {"decision":"needs_user","feedback":"Validation propriétaire requise avant déploiement ou redémarrage",'
        '"prompt_addendum":""}',
    )
    assert service.tick()["state"] == "self_repair_processed"

    blocked = service.detail(run["id"], "owner-1")
    assert blocked["status"] == "BLOCKED"
    assert blocked["steps"][0]["status"] == "BLOCKED"
    assert blocked["steps"][0]["finished_at"] is None
    assert blocked["self_repair"]["status"] == "NEEDS_USER"
    assert blocked["self_repair"]["prompt_addendum"] == ""

    resumed = service.resume("owner-1")
    assert resumed["status"] == "RUNNING"
    assert resumed["stage"] == "AGENT"
    assert resumed["self_repair"]["status"] == "SUPERSEDED"
    assert any(event["kind"] == "self_repair.owner_resolved" for event in resumed["events"])


def test_interrupted_supervisor_restarts_review_without_replaying_project_work(tmp_path):
    database, service = service_env(tmp_path)
    run = service.start("owner-1", [plan_step(target="Aloe")])
    finish_active(database, run["id"], 'CODEXPI_SUPERVISION {"decision":"approve","feedback":"Plan valide"}')
    service.tick()
    working = service.detail(run["id"], "owner-1")
    conversation_id = working["steps"][0]["conversation_id"]
    finish_active(database, run["id"], "Déploiement réversible terminé avec mesures avant/après.")
    service.tick()
    reviewing = service.detail(run["id"], "owner-1")
    interrupted_queue_id = reviewing["active_queue_id"]
    database.execute(
        "UPDATE prompt_queue SET status='failed',error=?,finished_at=? WHERE id=?",
        (INTERRUPTED_WORKER_REASON, utcnow(), interrupted_queue_id),
    )

    assert service.tick()["state"] == "processed"

    restarted = service.detail(run["id"], "owner-1")
    assert restarted["status"] == "RUNNING"
    assert restarted["stage"] == "SUPERVISOR"
    assert restarted["active_queue_id"] != interrupted_queue_id
    assert restarted["steps"][0]["status"] == "REVIEWING"
    assert restarted["steps"][0]["attempts"] == 1
    assert restarted["steps"][0]["conversation_id"] == conversation_id
    assert restarted["self_repair"] is None
    retry = database.one("SELECT content FROM prompt_queue WHERE id=?", (restarted["active_queue_id"],))
    assert "Reprends uniquement l’inspection" in retry["content"]
    assert "Ne relance aucune modification" in retry["content"]
    assert any(event["kind"] == "master.review_restarted" for event in restarted["events"])


def test_self_repair_of_interrupted_supervisor_resumes_review_not_agent(tmp_path):
    database, service = service_env(tmp_path)
    add_self_project(database, tmp_path)
    run = service.start("owner-1", [plan_step(target="Aloe")])
    finish_active(database, run["id"], 'CODEXPI_SUPERVISION {"decision":"approve","feedback":"Plan valide"}')
    service.tick()
    finish_active(database, run["id"], "Déploiement réversible terminé avec mesures avant/après.")
    service.tick()
    reviewing = service.detail(run["id"], "owner-1")
    project_attempts = reviewing["steps"][0]["attempts"]
    service._block(run["id"], INTERRUPTED_WORKER_REASON, "step-one")
    repair = service.detail(run["id"], "owner-1")["self_repair"]
    finish_queue(
        database,
        repair["queue_id"],
        'CODEXPI_SELF_REPAIR {"decision":"fixed","feedback":"Reprise de revue bornée testée",'
        '"prompt_addendum":""}',
    )

    assert service.tick()["state"] == "self_repair_processed"

    resumed = service.detail(run["id"], "owner-1")
    assert resumed["status"] == "RUNNING"
    assert resumed["stage"] == "SUPERVISOR"
    assert resumed["steps"][0]["status"] == "REVIEWING"
    assert resumed["steps"][0]["attempts"] == project_attempts
    queued = database.one("SELECT content FROM prompt_queue WHERE id=?", (resumed["active_queue_id"],))
    assert "Reprends uniquement l’inspection" in queued["content"]


def test_unexpected_runner_crash_is_sent_to_codexpi_self(tmp_path):
    database, service = service_env(tmp_path)
    add_self_project(database, tmp_path)
    run = service.start("owner-1", [plan_step()])

    def crash(_run):
        raise RuntimeError("plantage inattendu du moteur")

    service._process = crash
    result = service.tick()

    assert result["state"] == "blocked"
    blocked = service.detail(run["id"], "owner-1")
    assert blocked["status"] == "BLOCKED"
    assert "plantage inattendu" in blocked["error"]
    assert blocked["self_repair"]["status"] == "RUNNING"
    queued = database.one("SELECT content FROM prompt_queue WHERE id=?", (blocked["self_repair"]["queue_id"],))
    assert "plantage inattendu du moteur" in queued["content"]


def test_upgrade_recovers_a_legacy_block_and_rebuilds_its_supervisor_stage(tmp_path):
    database, service = service_env(tmp_path)
    add_self_project(database, tmp_path)
    run = service.start("owner-1", [plan_step()])
    finish_active(database, run["id"], 'CODEXPI_SUPERVISION {"decision":"approve","feedback":"Plan valide"}')
    service.tick()
    finish_active(database, run["id"], "Travail projet terminé avec preuves.")
    service.tick()
    reviewing = service.detail(run["id"], "owner-1")
    assert reviewing["stage"] == "SUPERVISOR"
    queue_id = reviewing["active_queue_id"]
    queue = database.one("SELECT conversation_id FROM prompt_queue WHERE id=?", (queue_id,))
    stamp = utcnow()
    database.execute(
        "UPDATE prompt_queue SET status='failed',error=?,finished_at=? WHERE id=?",
        ("thread-store conflict: already has an active writer", stamp, queue_id),
    )
    database.execute(
        "UPDATE conversations SET status='failed',worker_pid=NULL,updated_at=? WHERE id=?",
        (stamp, queue["conversation_id"]),
    )
    database.execute(
        "UPDATE remediation_runs SET status='BLOCKED',stage='IDLE',error=?,lease_owner=NULL,lease_until=NULL," 
        "finished_at=?,updated_at=? WHERE id=?",
        ("thread-store conflict: already has an active writer", stamp, stamp, run["id"]),
    )
    database.execute(
        "UPDATE remediation_step_runs SET status='BLOCKED',feedback=?,updated_at=? WHERE run_id=? AND step_id=?",
        ("thread-store conflict", stamp, run["id"], "step-one"),
    )

    upgraded = AutonomousRemediationService(database, service.codex)
    result = upgraded.tick()

    assert result["state"] == "self_repair_recovered"
    assert result["blocked_stage"] == "SUPERVISOR"
    recovered = upgraded.detail(run["id"], "owner-1")
    assert recovered["self_repair"]["status"] == "RUNNING"
    assert recovered["self_repair"]["blocked_stage"] == "SUPERVISOR"
    assert any(event["kind"] == "self_repair.catchup_started" for event in recovered["events"])


def test_master_and_project_ping_pong_runs_until_final_report(tmp_path):
    database, service = service_env(tmp_path)
    created = service.start("owner-1", [plan_step()])
    run_id = created["id"]
    assert created["stage"] == "PLAN_REVIEW"
    assert created["steps"][0]["status"] == "PENDING"

    finish_active(database, run_id, 'Plan cohérent.\nCODEXPI_SUPERVISION {"decision":"approve","feedback":"Ordre validé"}')
    assert service.tick()["state"] == "processed"
    running = service.detail(run_id, "owner-1")
    assert running["stage"] == "AGENT"
    assert running["steps"][0]["status"] == "WORKING"
    first_conversation = running["steps"][0]["conversation_id"]

    finish_active(database, run_id, "Correction appliquée, mais preuve partielle.")
    service.tick()
    reviewing = service.detail(run_id, "owner-1")
    assert reviewing["stage"] == "SUPERVISOR"
    assert reviewing["steps"][0]["status"] == "REVIEWING"

    finish_active(database, run_id, 'Preuve insuffisante.\nCODEXPI_SUPERVISION {"decision":"correct","feedback":"Ajouter la mesure avant/après"}')
    service.tick()
    correcting = service.detail(run_id, "owner-1")
    assert correcting["stage"] == "AGENT"
    assert correcting["correction_round"] == 1
    assert correcting["steps"][0]["status"] == "CORRECTING"
    assert correcting["steps"][0]["conversation_id"] == first_conversation
    assert correcting["steps"][0]["attempts"] == 2

    finish_active(database, run_id, "Correction complète avec mesures et tests réussis.")
    service.tick()
    finish_active(database, run_id, 'Résultat vérifié.\nCODEXPI_SUPERVISION {"decision":"approve","feedback":"Mesures et tests conformes"}')
    service.tick()
    finalizing = service.detail(run_id, "owner-1")
    assert finalizing["stage"] == "FINAL_REVIEW"
    assert finalizing["steps"][0]["status"] == "APPROVED"

    finish_active(database, run_id, "Synthèse finale du chantier.")
    service.tick()
    completed = service.detail(run_id, "owner-1")
    assert completed["status"] == "COMPLETED"
    assert completed["stage"] == "IDLE"
    assert any(event["kind"] == "run.completed" for event in completed["events"])


def test_only_one_autonomous_run_can_be_active(tmp_path):
    _, service = service_env(tmp_path)
    service.start("owner-1", [plan_step()])
    with pytest.raises(RemediationConflict, match="déjà en cours"):
        service.start("owner-1", [plan_step()])


def test_blocked_run_resumes_in_the_existing_project_conversation(tmp_path):
    database, service = service_env(tmp_path)
    run = service.start("owner-1", [plan_step()])
    finish_active(database, run["id"], 'CODEXPI_SUPERVISION {"decision":"approve","feedback":"Plan valide"}')
    service.tick()
    working = service.detail(run["id"], "owner-1")
    conversation_id = working["steps"][0]["conversation_id"]
    finish_active(database, run["id"], "Accès technique indisponible.")
    service.tick()
    finish_active(database, run["id"], 'CODEXPI_SUPERVISION {"decision":"block","feedback":"Canal manquant"}')
    service.tick()
    assert service.detail(run["id"], "owner-1")["status"] == "BLOCKED"

    resumed = service.resume("owner-1")
    assert resumed["status"] == "RUNNING"
    assert resumed["stage"] == "AGENT"
    assert resumed["steps"][0]["status"] == "CORRECTING"
    assert resumed["steps"][0]["conversation_id"] == conversation_id
    assert resumed["steps"][0]["attempts"] == 2
    assert any(event["kind"] == "run.resumed" for event in resumed["events"])


def test_blocked_run_can_rebase_prompts_without_replacing_threads_or_permissions(tmp_path):
    database, service = service_env(tmp_path)
    original = plan_step()
    run = service.start("owner-1", [original])
    finish_active(database, run["id"], 'CODEXPI_SUPERVISION {"decision":"approve","feedback":"Plan valide"}')
    service.tick()
    working = service.detail(run["id"], "owner-1")
    conversation_id = working["steps"][0]["conversation_id"]
    finish_active(database, run["id"], "Certaines métriques sont refusées par le noyau.")
    service.tick()
    finish_active(database, run["id"], 'CODEXPI_SUPERVISION {"decision":"block","feedback":"Critère impossible"}')
    service.tick()
    database.execute("UPDATE remediation_runs SET correction_round=3 WHERE id=?", (run["id"],))

    revised = {**original, "prompt": original["prompt"] + " Accepter une couverture explicitement qualifiée sans modifier les droits."}
    resumed = service.resume("owner-1", [revised])

    assert resumed["status"] == "RUNNING"
    assert resumed["correction_round"] == 0
    assert resumed["steps"][0]["prompt"] == revised["prompt"]
    assert resumed["steps"][0]["conversation_id"] == conversation_id
    queued = database.one("SELECT content FROM prompt_queue WHERE id=?", (resumed["active_queue_id"],))
    assert revised["prompt"] in queued["content"]
    event = next(item for item in resumed["events"] if item["kind"] == "run.resumed")
    assert event["details"]["plan_rebased"] is True


def test_rebase_refuses_to_change_target_or_risk(tmp_path):
    database, service = service_env(tmp_path)
    run = service.start("owner-1", [plan_step()])
    finish_active(database, run["id"], 'CODEXPI_SUPERVISION {"decision":"approve","feedback":"Plan valide"}')
    service.tick()
    finish_active(database, run["id"], "Résultat incomplet.")
    service.tick()
    finish_active(database, run["id"], 'CODEXPI_SUPERVISION {"decision":"block","feedback":"Étape bloquée"}')
    service.tick()
    assert service.detail(run["id"], "owner-1")["status"] == "BLOCKED"

    changed = {**plan_step(), "risk": "production"}
    with pytest.raises(RemediationConflict, match="ordre, la cible ou le risque"):
        service.resume("owner-1", [changed])


def test_owner_can_start_and_read_the_autonomous_run_through_the_api(logged_in, monkeypatch):
    client, app, projects, _, headers = logged_in
    owner_id = str(app.state.db.one("SELECT id FROM users LIMIT 1")["id"])
    master_path = projects / "master"
    target_path = projects / "nasflix"
    master_path.mkdir()
    target_path.mkdir()
    app.state.db.execute(
        "INSERT INTO projects(id,name,path,owner_id,full_access,execution_group,created_at) VALUES(?,?,?,?,?,?,?)",
        (str(uuid.uuid4()), "Agent maître", str(master_path), None, 1, "__global_agent__", utcnow()),
    )
    app.state.db.execute(
        "INSERT INTO projects(id,name,path,owner_id,created_at) VALUES(?,?,?,?,?)",
        (str(uuid.uuid4()), "NASflix", str(target_path), owner_id, utcnow()),
    )
    monkeypatch.setattr(app.state.codex, "detect", lambda *args: "/synthetic-agent")
    monkeypatch.setattr(app.state.codex, "_spawn", lambda *args: 4242)

    assert client.get("/api/resource-remediation").json()["status"] == "NOT_STARTED"
    response = client.post("/api/resource-remediation/start", headers=headers, json={"steps": [plan_step()]})
    assert response.status_code == 201, response.text
    assert response.json()["stage"] == "PLAN_REVIEW"
    assert response.json()["master_conversation_id"]
    current = client.get("/api/resource-remediation")
    assert current.status_code == 200
    assert current.json()["status"] == "RUNNING"