Artifex/tests/test_steward_v1.py

from __future__ import annotations

from pathlib import Path

from agents.steward import StewardService
from control_plane.projects.models import CommitRecord, Milestone, Project, ProjectPlan, StewardAction, StewardFinding, StewardPolicy, TaskStatus
from graph.bootstrap import champion_steward_run_graph_v1
from graph.langgraph_runtime import LangGraphRuntime
from graph.models import GraphApprovalStatus, GraphRun, GraphRunStatus
from graph.steward import steward_registry
from tests.test_m2_autonomous_loop import create_disposable_django_repo


def enrolled_project(tmp_path: Path, *, metadata: dict[str, object] | None = None, approval_requirements: dict[str, str] | None = None) -> tuple[Project, StewardService]:
    repo = create_disposable_django_repo(tmp_path)
    project = Project.objects.create(name="Stewarded", goal="Keep system healthy", repository_path=str(repo))
    policy = StewardPolicy.objects.create(
        name="Steward Test Policy",
        enabled_checks=["DEPENDENCY_DRIFT", "RUNTIME_CI", "SECURITY", "SECRET_EXPIRY", "PERFORMANCE"],
        metadata=metadata or {},
        severity_thresholds={"performance_delta_percent": 20},
        approval_requirements=approval_requirements or {"REPAIR": "CRITICAL"},
    )
    service = StewardService()
    service.enroll_project(project, policy)
    return project, service


def run_steward_graph(project: Project, service: StewardService):
    enrollment = project.steward_enrollments.get(status="ACTIVE")
    version = champion_steward_run_graph_v1()
    steward_run = service.start_run(enrollment, version)
    graph_run = GraphRun.objects.create(execution_graph_version=version, project=project, current_node=version.graph_spec["entry"], metadata={"steward_run_id": str(steward_run.id)})
    runtime = LangGraphRuntime(steward_registry(service))
    runtime.run_until_terminal_or_paused(graph_run)
    steward_run.refresh_from_db()
    graph_run.refresh_from_db()
    return steward_run, graph_run


def test_steward_enrollment_checks_and_repair_routing(tmp_path: Path) -> None:
    project, service = enrolled_project(
        tmp_path,
        metadata={"security_findings": [{"id": "CVE-1", "severity": "MEDIUM"}], "secret_expiry_metadata": [{"name": "api_cert", "expires_at": "2026-09-01", "severity": "MEDIUM"}]},
    )

    steward_run, graph_run = run_steward_graph(project, service)

    assert graph_run.status == GraphRunStatus.COMPLETE
    assert steward_run.status == "COMPLETE"
    assert steward_run.checks.count() == 5
    findings = list(steward_run.findings.order_by("finding_type"))
    assert [finding.finding_type for finding in findings] == ["SECRET_EXPIRY", "SECURITY_SIGNAL"]
    actions = list(StewardAction.objects.select_related("task", "finding"))
    assert len(actions) == 2
    assert all(action.action_type == "REPAIR" for action in actions)
    assert all(action.task and action.task.status == TaskStatus.READY for action in actions)
    assert "api_cert" in actions[0].finding.evidence.get("expiring_secrets", [{}])[0].get("name", "") or "api_cert" in actions[1].finding.evidence.get("expiring_secrets", [{}])[0].get("name", "")


def test_steward_deduplicates_findings_across_runs(tmp_path: Path) -> None:
    project, service = enrolled_project(tmp_path, metadata={"security_findings": [{"id": "CVE-1", "severity": "MEDIUM"}]})

    first_run, _ = run_steward_graph(project, service)
    second_run, _ = run_steward_graph(project, service)

    assert StewardFinding.objects.count() == 1
    finding = StewardFinding.objects.get()
    assert finding.occurrence_count == 2
    assert finding.steward_run == second_run
    assert first_run.findings.count() == 0


def test_high_severity_repair_pauses_until_approval(tmp_path: Path) -> None:
    project, service = enrolled_project(tmp_path, metadata={"security_findings": [{"id": "CVE-2", "severity": "HIGH"}]}, approval_requirements={"REPAIR": "HIGH"})

    steward_run, graph_run = run_steward_graph(project, service)

    assert graph_run.status == GraphRunStatus.PAUSED
    assert graph_run.approvals.get().status == GraphApprovalStatus.PENDING
    action = StewardAction.objects.get()
    assert action.status == "APPROVAL_REQUIRED"
    assert action.task is None

    approval = graph_run.approvals.get()
    approval.status = GraphApprovalStatus.APPROVED
    approval.decided_by = "tester"
    approval.save(update_fields=["status", "decided_by", "updated_at"])
    graph_run.status = GraphRunStatus.RUNNING
    graph_run.save(update_fields=["status", "updated_at"])
    LangGraphRuntime(steward_registry(service)).run_until_terminal_or_paused(graph_run)

    graph_run.refresh_from_db()
    steward_run.refresh_from_db()
    action.refresh_from_db()
    assert graph_run.status == GraphRunStatus.COMPLETE
    assert steward_run.status == "COMPLETE"
    assert action.status == "ROUTED"
    assert action.task is not None


def test_steward_routes_systemic_runtime_signal_to_progeny_investigation(tmp_path: Path) -> None:
    project, service = enrolled_project(tmp_path)
    enrollment = project.steward_enrollments.get()
    steward_run = service.start_run(enrollment)
    check = service._check(steward_run, "RUNTIME_CI", "FAIL", {"events": [{"event_type": "CI_FAILED"}]}, "HIGH")
    raw = service._findings_for_check(check)[0]
    finding = service.upsert_finding(steward_run, check, raw)
    finding.occurrence_count = 2
    finding.save(update_fields=["occurrence_count"])
    service.classify_findings(steward_run)

    action = service.route_findings(steward_run)[0]

    assert action.action_type == "INVESTIGATE"
    assert action.investigation is not None
    assert action.task is None


def test_repair_resolution_links_commit_to_steward_finding(tmp_path: Path) -> None:
    project, service = enrolled_project(tmp_path, metadata={"security_findings": [{"id": "CVE-1", "severity": "MEDIUM"}]})
    run_steward_graph(project, service)
    action = StewardAction.objects.select_related("task", "finding").get()
    task = action.task
    assert task is not None
    task.status = TaskStatus.COMPLETE
    task.save(update_fields=["status", "updated_at"])
    CommitRecord.objects.create(project=project, task=task, sha="abc123", branch_name="repair", message="Repair Steward finding")

    assert service.resolve_after_repair(action.finding) is True

    action.finding.refresh_from_db()
    assert action.finding.status == "RESOLVED"
    assert CommitRecord.objects.get(task=task).steward_finding == action.finding


def test_resolved_identical_findings_do_not_create_duplicate_repairs(tmp_path: Path) -> None:
    project, service = enrolled_project(tmp_path, metadata={"security_findings": [{"id": "CVE-1", "severity": "MEDIUM"}]})
    run_steward_graph(project, service)
    action = StewardAction.objects.select_related("task", "finding").get()
    task = action.task
    assert task is not None
    task.status = TaskStatus.COMPLETE
    task.save(update_fields=["status", "updated_at"])
    CommitRecord.objects.create(project=project, task=task, sha="abc123", branch_name="repair", message="Repair Steward finding")
    service.resolve_after_repair(action.finding)

    run_steward_graph(project, service)

    assert StewardFinding.objects.count() == 1
    assert StewardAction.objects.count() == 1
    assert project.tasks.filter(task_type="repair").count() == 1
    assert StewardFinding.objects.get().status == "RESOLVED"