ss-tools/backend/src/api/routes/dashboard_testing/scenario_automation.py

# #region Api.ScenarioAutomation.Routes [C:4] [TYPE Module] [SEMANTICS scenario,automation,api,schedule,trigger,policy,metrics,rbac]
# @defgroup Api Scenario automation management routes (SCAUTO-FR-010/011).
# @BRIEF Persisted CRUD for schedules, trigger rules and policies, notification list,
#       operational metrics, and the direct scenario run trigger.
# @RELATION DEPENDS_ON -> [ScenarioAutomation.Schedule]
# @RELATION DEPENDS_ON -> [ScenarioAutomation.Metrics.Aggregate]
# @RELATION DEPENDS_ON -> [ScenarioExecution.Runner.Start]
# @RELATION DEPENDS_ON -> [ScenarioExecution.Runner.QueuedDispatch]
# @RELATION DEPENDS_ON -> [ScenarioExecution.EnvironmentPolicy.Resolve]
# @INVARIANT Mutations require scenario:automation MANAGE; direct triggers require TRIGGER
#       (+ automation PROD and scenario RUN_PROD for PROD-classified environments). PROD classification is server-owned
#       (ConfigManager stage/is_production) — never derived from the environment name.
# @INVARIANT 046 routes pass a server-owned automation origin to 044. A revision with a human
#            step is rejected as manual-run-only before a ScenarioRun, gate, or notification exists.
# @INVARIANT All operational reads (schedules/trigger-rules/policies/notifications/metrics/retention)
#            require an authenticated principal holding scenario:automation READ plus per-object
#            scenario-ownership ACL (SEC-01, DG-2 2026-09-12): anonymous -> 401 via the OAuth2
#            bearer dependency, lacking READ -> 403, foreign rows are filtered out of every
#            collection projection so existence never leaks (foreign object -> 404 semantics on
#            a collection surface); admin roles bypass the ACL filter and see all rows.
#            Schedule/trigger-rule persistence additionally runs the shared revision-bound
#            eligibility guard before any row is written (DEF-02) — REST and MCP never diverge.
# @RATIONALE DG-2 (046 T019): reads admit an authenticated principal of ANY type holding
#            scenario:automation READ with per-object ACL; mutations stay human-only. Policy
#            objects carry no scenario ownership, so a policy is hidden from a non-admin only
#            when it is referenced exclusively by foreign scenarios — unreferenced or own-
#            referenced policies stay visible to keep the management surface functional.
# @REJECTED Authenticated-only reads without a READ grant were rejected (DG-2 supersedes the
#           pre-2026-09-12 note) — operational config is not public-to-any-user data.
# @REJECTED Hiding unreferenced policies from non-admin READ holders was rejected — policies are
#           shared named configuration; breaking the policy picker for legitimate READ holders
#           buys no secrecy (a policy carries no scenario payload).
# @REJECTED A parallel scheduler was rejected — this surface persists configuration; APScheduler
#           job registration stays server-owned via the 037 framework.
# @REJECTED str(environment_id).startswith("prod") PROD classification was rejected — a client-name
#           heuristic that is spoofable by environment naming; replaced by ConfigManager lookup.
from __future__ import annotations

from typing import Any

from fastapi import APIRouter, Depends, Header, HTTPException, status
from pydantic import BaseModel, Field

from src.dependencies import get_config_manager, get_current_user, get_db, get_scheduler_service, has_permission
from src.models.scenario_automation import AutomationPolicy, ScenarioNotificationEvent, ScenarioRetentionDeletion, ScenarioSchedule, ScenarioTriggerRule
from src.models.scenario_registry import ScenarioRegistryEntry
from src.services.dashboard_testing.automation.metrics import automation_metrics
from src.services.dashboard_testing.automation.eligibility import assert_automation_eligible
from src.services.dashboard_testing.automation.poisoned import unquarantine_scenario_automation
from src.services.dashboard_testing.automation.poisoned_store import get_poisoned_store
from src.services.dashboard_testing.automation.retention import tier_limits
from src.services.dashboard_testing.automation.trigger import dispatch_trigger_event
from src.services.dashboard_testing.execution.environment_policy import (
    resolve_environment_execution_policy,
)
from src.services.dashboard_testing.execution.runner import start_run

router = APIRouter(prefix="/api/scenario-automation", tags=["scenario-automation"])
_DB = Depends(get_db)
_USER = Depends(get_current_user)
_READ = Depends(has_permission("scenario:automation", "READ"))
_MANAGE = Depends(has_permission("scenario:automation", "MANAGE"))
_TRIGGER = Depends(has_permission("scenario:automation", "TRIGGER"))
_CONFIG_MANAGER = Depends(get_config_manager)

TRIGGER_TYPES = {"deploy_to_preprod", "release_created", "etl_completed", "api"}
MISSED_POLICIES = {"skip", "run_latest", "queue_all"}


# #region Api.ScenarioAutomation.ReadAcl [C:3] [TYPE Block] [SEMANTICS scenario,automation,api,rbac,acl]
# @ingroup Api
# @BRIEF Per-object read ACL for automation collections: admin sees all; any other READ holder
#        sees only rows whose scenario registry entry they own (owner_id/owner_username).
# @POST _visible_scenario_ids returns None for admin (no filter) or a set of owned scenario ids;
#       rows pointing at scenarios without a registry entry are hidden from non-admins (their
#       ownership cannot be proven, so their existence must not leak).
def _is_admin(user) -> bool:
    return any(getattr(role, "is_admin", False) for role in (getattr(user, "roles", None) or []))


def _visible_scenario_ids(db, user) -> set[str] | None:
    if _is_admin(user):
        return None
    user_id = str(getattr(user, "id", "") or "")
    username = str(getattr(user, "username", "") or "")
    rows = (
        db.query(ScenarioRegistryEntry.scenario_id)
        .filter(
            (ScenarioRegistryEntry.owner_id == user_id)
            | (ScenarioRegistryEntry.owner_username == username)
        )
        .all()
    )
    return {row[0] for row in rows}


def _visible_policy_ids(db, visible: set[str] | None) -> set[str] | None:
    """None for admin; otherwise policies NOT referenced exclusively by foreign scenarios."""
    if visible is not None:
        referenced = {
            row[0]
            for row in db.query(ScenarioSchedule.policy_id).filter(ScenarioSchedule.policy_id.isnot(None)).all()
        } | {
            row[0]
            for row in db.query(ScenarioTriggerRule.policy_id).filter(ScenarioTriggerRule.policy_id.isnot(None)).all()
        }
        own_referenced = {
            row[0]
            for row in db.query(ScenarioSchedule.policy_id)
            .filter(ScenarioSchedule.policy_id.isnot(None), ScenarioSchedule.scenario_id.in_(visible))
            .all()
        } | {
            row[0]
            for row in db.query(ScenarioTriggerRule.policy_id)
            .filter(ScenarioTriggerRule.policy_id.isnot(None), ScenarioTriggerRule.scenario_id.in_(visible))
            .all()
        }
        foreign_only = referenced - own_referenced
        return {
            row[0] for row in db.query(AutomationPolicy.id).all()
        } - foreign_only
    return None
# #endregion Api.ScenarioAutomation.ReadAcl


# #region Api.ScenarioAutomation.ScheduleRequest [C:1] [TYPE Class] [SEMANTICS scenario,automation,api,schedule]
# @ingroup Api
class ScheduleRequest(BaseModel):
    scenario_id: str
    environment_id: str
    cron_expr: str
    revision_policy: str = "current"
    revision_id: str | None = None
    timezone: str = "UTC"
    policy_id: str | None = None
    missed_execution_policy: str = "skip"
    max_instances: int = Field(default=1, ge=1)
    misfire_grace_time: int = Field(default=300, ge=0)
    enabled: bool = True
# #endregion Api.ScenarioAutomation.ScheduleRequest


# #region Api.ScenarioAutomation.TriggerRuleRequest [C:1] [TYPE Class] [SEMANTICS scenario,automation,api,trigger]
# @ingroup Api
class TriggerRuleRequest(BaseModel):
    scenario_id: str
    environment_id: str
    trigger: str
    revision_policy: str = "current"
    revision_id: str | None = None
    policy_id: str | None = None
    enabled: bool = True
# #endregion Api.ScenarioAutomation.TriggerRuleRequest


# #region Api.ScenarioAutomation.PolicyRequest [C:1] [TYPE Class] [SEMANTICS scenario,automation,api,policy]
# @ingroup Api
class PolicyRequest(BaseModel):
    name: str
    enabled: bool = True
    workload_class: str = "scenario_smoke"
    max_concurrent_per_env: int = Field(default=1, ge=1)
    dedup_window_seconds: int = Field(default=0, ge=0)
    overlap_rule: str = "block"
    retention_days: int = Field(default=30, ge=1)
    prod_gate_required: bool = False
    on_repeated_failure: str = "alert"
# #endregion Api.ScenarioAutomation.PolicyRequest


# #region Api.ScenarioAutomation.QuarantineReleaseRequest [C:1] [TYPE Class] [SEMANTICS scenario,automation,api,quarantine,recovery]
# @ingroup Api
class QuarantineReleaseRequest(BaseModel):
    environment_id: str
    expected_version: int = Field(..., ge=1)
# #endregion Api.ScenarioAutomation.QuarantineReleaseRequest


# #region Api.ScenarioAutomation.ApiTriggerRequest [C:1] [TYPE Class] [SEMANTICS scenario,automation,api,trigger,run]
# @ingroup Api
class ApiTriggerRequest(BaseModel):
    environment_id: str
    revision_id: str | None = None
    params: dict[str, Any] = Field(default_factory=dict)
# #endregion Api.ScenarioAutomation.ApiTriggerRequest


# #region Api.ScenarioAutomation.EventDispatchRequest [C:1] [TYPE Class] [SEMANTICS scenario,automation,api,trigger,event]
# @ingroup Api
# @BRIEF Server-owned typed event envelope accepted by the 046 trigger dispatcher.
class EventDispatchRequest(BaseModel):
    type: str
    fingerprint: str
    overlap: bool = False
# #endregion Api.ScenarioAutomation.EventDispatchRequest


def _validate_rule_trigger(trigger: str) -> None:
    if trigger not in TRIGGER_TYPES:
        raise HTTPException(status_code=422, detail={"code": "INVALID_TRIGGER", "detail": f"trigger must be one of {sorted(TRIGGER_TYPES)}"})


def _validate_missed_policy(policy: str) -> None:
    if policy not in MISSED_POLICIES:
        raise HTTPException(status_code=422, detail={"code": "INVALID_MISSED_POLICY", "detail": f"missed_execution_policy must be one of {sorted(MISSED_POLICIES)}"})


# #region Api.ScenarioAutomation.Schedules [C:3] [TYPE Block] [SEMANTICS scenario,automation,api,schedule,crud]
# @ingroup Api
# @BRIEF Persist schedule configuration and register/remove its server-owned scheduler job.
# @RELATION DEPENDS_ON -> [ScenarioAutomation.Schedule.DeriveApsParams]
# @INVARIANT Schedule CRUD may register work but never claims or executes a queued ScenarioRun;
#            the separate 044 queued dispatcher remains the sole initial execution authority.
@router.get("/schedules")
def list_schedules(db=_DB, current_user=_READ):
    query = db.query(ScenarioSchedule).order_by(ScenarioSchedule.created_at.desc())
    visible = _visible_scenario_ids(db, current_user)
    if visible is not None:
        query = query.filter(ScenarioSchedule.scenario_id.in_(visible))
    return query.all()


@router.post("/schedules", status_code=status.HTTP_201_CREATED)
def create_schedule(body: ScheduleRequest, db=_DB, _perm=_MANAGE, _user=_USER):
    _validate_missed_policy(body.missed_execution_policy)
    try:
        assert_automation_eligible(db, body.scenario_id, body.revision_id)
    except ValueError as exc:
        raise HTTPException(status_code=409, detail={"code": str(exc)}) from exc
    schedule = ScenarioSchedule(**body.model_dump())
    # Owner principal: scheduled runs execute as this user so the deployment binding's principal
    # fingerprint (sha256(actor)) matches; trigger_source stays "scheduled".
    schedule.created_by = str(_user.id)
    db.add(schedule)
    db.commit()
    db.refresh(schedule)
    if schedule.enabled:
        get_scheduler_service().add_scenario_job(
            schedule_id=schedule.id,
            scenario_id=schedule.scenario_id,
            revision_policy=schedule.revision_policy,
            revision_id=schedule.revision_id,
            environment_id=schedule.environment_id,
            cron_expr=schedule.cron_expr,
            timezone=schedule.timezone,
            policy_id=schedule.policy_id,
            misfire_grace_time=schedule.misfire_grace_time,
            max_instances=schedule.max_instances,
            missed_execution_policy=schedule.missed_execution_policy,
        )
    return schedule


@router.patch("/schedules/{schedule_id}")
def update_schedule(schedule_id: str, body: ScheduleRequest, db=_DB, _perm=_MANAGE, _user=_USER):
    schedule = db.query(ScenarioSchedule).filter(ScenarioSchedule.id == schedule_id).first()
    if schedule is None:
        raise HTTPException(status_code=404, detail={"code": "SCHEDULE_NOT_FOUND"})
    _validate_missed_policy(body.missed_execution_policy)
    try:
        assert_automation_eligible(db, body.scenario_id, body.revision_id)
    except ValueError as exc:
        raise HTTPException(status_code=409, detail={"code": str(exc)}) from exc
    for key, value in body.model_dump(exclude_unset=True).items():
        setattr(schedule, key, value)
    db.commit()
    db.refresh(schedule)
    if schedule.enabled:
        get_scheduler_service().add_scenario_job(
            schedule_id=schedule.id,
            scenario_id=schedule.scenario_id,
            revision_policy=schedule.revision_policy,
            revision_id=schedule.revision_id,
            environment_id=schedule.environment_id,
            cron_expr=schedule.cron_expr,
            timezone=schedule.timezone,
            policy_id=schedule.policy_id,
            misfire_grace_time=schedule.misfire_grace_time,
            max_instances=schedule.max_instances,
            missed_execution_policy=schedule.missed_execution_policy,
        )
    else:
        get_scheduler_service().remove_scenario_job(schedule.id)
    return schedule


@router.delete("/schedules/{schedule_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_schedule(schedule_id: str, db=_DB, _perm=_MANAGE):
    schedule = db.query(ScenarioSchedule).filter(ScenarioSchedule.id == schedule_id).first()
    if schedule is None:
        raise HTTPException(status_code=404, detail={"code": "SCHEDULE_NOT_FOUND"})
    get_scheduler_service().remove_scenario_job(schedule.id)
    db.delete(schedule)
    db.commit()
    return None
# #endregion Api.ScenarioAutomation.Schedules


# #region Api.ScenarioAutomation.TriggerRules [C:3] [TYPE Block] [SEMANTICS scenario,automation,api,trigger,crud]
# @ingroup Api
@router.get("/trigger-rules")
def list_trigger_rules(db=_DB, current_user=_READ):
    query = db.query(ScenarioTriggerRule).order_by(ScenarioTriggerRule.created_at.desc())
    visible = _visible_scenario_ids(db, current_user)
    if visible is not None:
        query = query.filter(ScenarioTriggerRule.scenario_id.in_(visible))
    return query.all()


@router.post("/trigger-rules", status_code=status.HTTP_201_CREATED)
def create_trigger_rule(body: TriggerRuleRequest, db=_DB, _perm=_MANAGE, _user=_USER):
    _validate_rule_trigger(body.trigger)
    try:
        assert_automation_eligible(db, body.scenario_id, body.revision_id)
    except ValueError as exc:
        raise HTTPException(status_code=409, detail={"code": str(exc)}) from exc
    rule = ScenarioTriggerRule(**body.model_dump())
    db.add(rule)
    db.commit()
    db.refresh(rule)
    return rule


@router.patch("/trigger-rules/{rule_id}")
def update_trigger_rule(rule_id: str, body: TriggerRuleRequest, db=_DB, _perm=_MANAGE, _user=_USER):
    rule = db.query(ScenarioTriggerRule).filter(ScenarioTriggerRule.id == rule_id).first()
    if rule is None:
        raise HTTPException(status_code=404, detail={"code": "TRIGGER_RULE_NOT_FOUND"})
    _validate_rule_trigger(body.trigger)
    try:
        assert_automation_eligible(db, body.scenario_id, body.revision_id)
    except ValueError as exc:
        raise HTTPException(status_code=409, detail={"code": str(exc)}) from exc
    for key, value in body.model_dump(exclude_unset=True).items():
        setattr(rule, key, value)
    db.commit()
    db.refresh(rule)
    return rule


@router.delete("/trigger-rules/{rule_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_trigger_rule(rule_id: str, db=_DB, _perm=_MANAGE):
    rule = db.query(ScenarioTriggerRule).filter(ScenarioTriggerRule.id == rule_id).first()
    if rule is None:
        raise HTTPException(status_code=404, detail={"code": "TRIGGER_RULE_NOT_FOUND"})
    db.delete(rule)
    db.commit()
    return None
# #endregion Api.ScenarioAutomation.TriggerRules


# #region Api.ScenarioAutomation.Policies [C:3] [TYPE Block] [SEMANTICS scenario,automation,api,policy,crud]
# @ingroup Api
@router.get("/policies")
def list_policies(db=_DB, current_user=_READ):
    query = db.query(AutomationPolicy).order_by(AutomationPolicy.created_at.desc())
    visible = _visible_policy_ids(db, _visible_scenario_ids(db, current_user))
    if visible is not None:
        query = query.filter(AutomationPolicy.id.in_(visible))
    return query.all()


@router.post("/policies", status_code=status.HTTP_201_CREATED)
def create_policy(body: PolicyRequest, db=_DB, _perm=_MANAGE):
    policy = AutomationPolicy(**body.model_dump())
    db.add(policy)
    db.commit()
    db.refresh(policy)
    return policy


@router.patch("/policies/{policy_id}")
def update_policy(policy_id: str, body: PolicyRequest, db=_DB, _perm=_MANAGE):
    policy = db.query(AutomationPolicy).filter(AutomationPolicy.id == policy_id).first()
    if policy is None:
        raise HTTPException(status_code=404, detail={"code": "POLICY_NOT_FOUND"})
    for key, value in body.model_dump(exclude_unset=True).items():
        setattr(policy, key, value)
    db.commit()
    db.refresh(policy)
    return policy


@router.delete("/policies/{policy_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_policy(policy_id: str, db=_DB, _perm=_MANAGE):
    policy = db.query(AutomationPolicy).filter(AutomationPolicy.id == policy_id).first()
    if policy is None:
        raise HTTPException(status_code=404, detail={"code": "POLICY_NOT_FOUND"})
    db.delete(policy)
    db.commit()
    return None
# #endregion Api.ScenarioAutomation.Policies


# #region Api.ScenarioAutomation.Notifications [C:2] [TYPE Function] [SEMANTICS scenario,automation,api,notifications]
# @ingroup Api
@router.get("/notifications")
def list_notifications(limit: int = 100, db=_DB, current_user=_READ):
    query = db.query(ScenarioNotificationEvent).order_by(ScenarioNotificationEvent.created_at.desc())
    visible = _visible_scenario_ids(db, current_user)
    if visible is not None:
        query = query.filter(ScenarioNotificationEvent.scenario_id.in_(visible))
    return query.limit(limit).all()
# #endregion Api.ScenarioAutomation.Notifications


# #region Api.ScenarioAutomation.Quarantine [C:4] [TYPE Block] [SEMANTICS scenario,automation,api,quarantine,dlq,recovery,operator]
# @ingroup Api
# @BRIEF Operator-only read and CAS release of poisoned-run quarantines recorded by the durable failure counter.
# @RELATION CALLS -> [ScenarioAutomation.Poisoned.Unquarantine]
# @RELATION DEPENDS_ON -> [ScenarioAutomation.PoisonedStore]
# @POST A released quarantine re-enables matching schedules and resets the identical-failure counter.
# @INVARIANT Recovery requires scenario:automation MANAGE and a matching quarantine version; an unknown or stale
#            release changes no schedule and no counter.
@router.get("/quarantine")
def list_quarantines(_user=_USER):
    return get_poisoned_store().list_quarantines()


@router.post("/quarantine/{scenario_id}/release")
def release_quarantine(
    scenario_id: str,
    body: QuarantineReleaseRequest,
    db=_DB,
    _perm=_MANAGE,
    current_user=_USER,
):
    store = get_poisoned_store()
    actor = str(getattr(current_user, "id", "") or getattr(current_user, "username", "") or "")
    try:
        result = unquarantine_scenario_automation(
            db,
            store,
            scenario_id=scenario_id,
            environment_id=body.environment_id,
            expected_version=body.expected_version,
            actor=actor,
        )
        db.commit()
    except ValueError as exc:
        db.rollback()
        code = str(exc)
        raise HTTPException(
            status_code=404 if code == "NOT_QUARANTINED" else 409,
            detail={"code": code},
        ) from exc
    return result
# #endregion Api.ScenarioAutomation.Quarantine


# #region Api.ScenarioAutomation.EventDispatch [C:3] [TYPE Function] [SEMANTICS scenario,automation,api,event,trigger,persistence]
# @ingroup Api
# @BRIEF Persist policy-allowed event-triggered runs without request-time execution.
# @RELATION CALLS -> [ScenarioAutomation.Trigger.Dispatch]
# @RELATION DEPENDS_ON -> [ScenarioExecution.Runner.QueuedDispatch]
# @POST Returns accepted run ids only after persistence; adapter execution waits for the server dispatcher CAS.
# @INVARIANT A malformed automated human plan is refused before creation; this HTTP route never
#            fabricates manual authority or invokes a walker.
# @INVARIANT ConfigManager classifies every persisted trigger target before start; an event payload
#            cannot provide an is_prod/environment_class override or create an unknown target.
@router.post("/events/dispatch", status_code=status.HTTP_202_ACCEPTED)
def api_dispatch_event(
    body: EventDispatchRequest,
    db=_DB,
    _perm=_TRIGGER,
    config_manager=_CONFIG_MANAGER,
):
    if body.type not in TRIGGER_TYPES:
        raise HTTPException(status_code=422, detail={"code": "INVALID_TRIGGER"})
    try:
        created = dispatch_trigger_event(
            db,
            {"type": body.type, "fingerprint": body.fingerprint, "overlap": body.overlap},
            config_manager=config_manager,
        )
        db.commit()
        return {"run_ids": created, "count": len(created)}
    except ValueError as exc:
        db.rollback()
        if str(exc) == "ENVIRONMENT_NOT_CONFIGURED":
            raise HTTPException(
                status_code=422,
                detail={"code": str(exc), "detail": str(exc)},
            ) from exc
        raise
# #endregion Api.ScenarioAutomation.EventDispatch


# #region Api.ScenarioAutomation.Metrics [C:3] [TYPE Function] [SEMANTICS scenario,automation,api,metrics]
# @ingroup Api
# @BRIEF Operational metrics over persisted schedules, trigger rules, runs and notifications.
@router.get("/metrics")
def metrics(db=_DB, current_user=_READ):
    visible = _visible_scenario_ids(db, current_user)
    schedules_q = db.query(ScenarioSchedule)
    rules_q = db.query(ScenarioTriggerRule)
    notifications_q = db.query(ScenarioNotificationEvent)
    from src.models.scenario_run import ScenarioRun

    runs_q = db.query(ScenarioRun)
    if visible is not None:
        schedules_q = schedules_q.filter(ScenarioSchedule.scenario_id.in_(visible))
        rules_q = rules_q.filter(ScenarioTriggerRule.scenario_id.in_(visible))
        notifications_q = notifications_q.filter(ScenarioNotificationEvent.scenario_id.in_(visible))
        runs_q = runs_q.filter(ScenarioRun.scenario_id.in_(visible))
    schedules = schedules_q.all()
    rules = rules_q.all()
    notifications = notifications_q.all()
    runs = runs_q.all()
    return automation_metrics(
        schedules=[{"enabled": s.enabled} for s in schedules],
        trigger_rules=[{"enabled": r.enabled} for r in rules],
        runs=[
            {"status": r.status, "trigger_source": r.trigger_source, "started_at": r.created_at}
            for r in runs
        ],
        notifications=[
            {"event_type": n.event_type, "scenario_id": n.scenario_id, "run_id": n.run_id, "payload": n.payload}
            for n in notifications
        ],
    )
# #endregion Api.ScenarioAutomation.Metrics


# #region Api.ScenarioAutomation.RetentionDefaults [C:3] [TYPE Function] [SEMANTICS scenario,automation,api,retention,tiers,deletion,receipts]
# @ingroup Api
# @BRIEF Expose the canonical layered retention tier horizons plus per-object deletion receipts.
# @POST Returns tiers and the ACL-filtered deletion receipts; totals never leak foreign rows.
# @INVARIANT Receipts with a null scenario_id are admin-only (fail-closed ACL); a non-admin never
#            sees a foreign receipt row nor its count.
@router.get("/retention")
def retention_defaults(db=_DB, current_user=_READ):
    from src.services.dashboard_testing.automation.deletions import retention_receipt_to_dict

    query = db.query(ScenarioRetentionDeletion).order_by(ScenarioRetentionDeletion.created_at.desc())
    visible = _visible_scenario_ids(db, current_user)
    if visible is not None:
        query = query.filter(ScenarioRetentionDeletion.scenario_id.in_(visible))
    receipts = query.limit(200).all()
    return {
        "tiers": tier_limits(),
        "deletions": [retention_receipt_to_dict(receipt) for receipt in receipts],
        "deletions_total": len(receipts),
    }
# #endregion Api.ScenarioAutomation.RetentionDefaults


# #region Api.ScenarioAutomation.DirectTrigger [C:4] [TYPE Function] [SEMANTICS scenario,automation,api,trigger,run,idempotency]
# @ingroup Api
# @BRIEF External API run trigger (SCAUTO-FR-011) — starts a scenario run with an
#       Idempotency-Key; PROD-classified environments (server-owned ConfigManager
#       classification, never a name heuristic) require automation PROD + scenario RUN_PROD.
# @PRE Idempotency-Key header is required; scenario_id path parameter names the scenario;
#      the environment must be configured server-side or the trigger is refused (422).
# @POST Returns 202 with run_id/status; 409 IDEMPOTENCY_KEY_REUSED on changed request.
# @POST A human-containing revision returns 409 AUTOMATION_INELIGIBLE_HUMAN_STEP and creates no run.
# @RELATION CALLS -> [ScenarioExecution.Runner.Start]
# @RELATION CALLS -> [ScenarioExecution.EnvironmentPolicy.Resolve]
# @RELATION DEPENDS_ON -> [ScenarioExecution.Runner.QueuedDispatch]
# @SIDE_EFFECT Commits only a permitted ScenarioRun; rejection rolls back without a run, gate, notification, or queue row.
# @INVARIANT API automation origin is server-owned and cannot consume or substitute a HumanCheckpoint or ActionApprovalGate.
# @INVARIANT HTTP trigger/replay is persistence-only: its accepted row remains queued/pending until
#            the separate server dispatcher wins the queued->running status CAS.
# @INVARIANT PROD classification and RUN_PROD validation complete before start_run; a valid
#            automation intent persists pending_approval plus its ActionApprovalGate, never queued.
@router.post("/scenarios/{scenario_id}/trigger", status_code=status.HTTP_202_ACCEPTED)
def api_trigger_scenario(
    scenario_id: str,
    body: ApiTriggerRequest,
    idempotency_key: str = Header(..., alias="Idempotency-Key"),
    db=_DB,
    current_user=_USER,
    _perm=_TRIGGER,
    config_manager=_CONFIG_MANAGER,
):
    try:
        environment_policy = resolve_environment_execution_policy(
            body.environment_id, config_manager
        )
    except ValueError as exc:
        raise HTTPException(
            status_code=422,
            detail={"code": str(exc), "detail": str(exc)},
        ) from exc
    if environment_policy.is_prod:
        has_permission("scenario:automation", "PROD")(current_user=current_user)
        has_permission("scenario", "RUN_PROD")(current_user=current_user)
    revision_id = body.revision_id or ""
    try:
        run = start_run(
            db,
            scenario_id,
            revision_id,
            body.params,
            body.environment_id,
            actor=str(getattr(current_user, "id", "") or getattr(current_user, "username", "")),
            idempotency_key=idempotency_key,
            config_manager=config_manager,
            trigger_source="api",
        )
        db.commit()
    except PermissionError as exc:
        db.rollback()
        raise HTTPException(status_code=403, detail={"code": "PROD_APPROVAL_REQUIRED", "detail": str(exc)}) from exc
    except ValueError as exc:
        db.rollback()
        code = (
            "AUTOMATION_INELIGIBLE_HUMAN_STEP"
            if str(exc) == "AUTOMATION_INELIGIBLE_HUMAN_STEP"
            else "IDEMPOTENCY_KEY_REUSED" if "IDEMPOTENCY" in str(exc) else "RUN_START_CONFLICT"
        )
        raise HTTPException(status_code=409, detail={"code": code, "detail": str(exc)}) from exc
    return {"run_id": run.id, "status": run.status, "scenario_id": scenario_id}
# #endregion Api.ScenarioAutomation.DirectTrigger
# #endregion Api.ScenarioAutomation.Routes