DataOps-platform/tests/integration/test_agent_model_offline.py

"""Opt-in V54 deterministic model-offline degradation acceptance."""

import os

import pytest
from sqlalchemy import create_engine, text

from app.core.common.identifiers import new_governance_uid
from app.core.mcp.identity import AgentIdentity
from app.core.mcp.persistence import PostgresAuditSink
from app.core.orchestration.agent.planner import SchedulingPlanner

pytestmark = pytest.mark.integration
DATABASE_URL = "postgresql://dataops:dataops-test-password@127.0.0.1:15432/dataops"


class OfflineModel:
    provider = "deepseek"
    model_name = "deepseek-chat"

    def generate(self, **_kwargs):
        raise TimeoutError("the model provider is unavailable")


class ContextTools:
    def call_tool(self, name, _arguments):
        return {
            "get_sla_constraints": {"deadline": "08:00"},
            "list_datasource_capabilities": {
                "items": [],
                "count": 0,
                "truncated": False,
            },
            "get_execution_history": {
                "items": [{"status": "success"}],
                "count": 1,
                "truncated": False,
            },
        }[name]


class ForbiddenSchedulingTools:
    def call_tool(self, name, _arguments):
        raise AssertionError(f"offline model must not call scheduling tool {name}")


class ForbiddenVerifier:
    def verify(self, *_args, **_kwargs):
        raise AssertionError("offline model must not start Canary verification")


def test_model_offline_keeps_fixed_schedule_and_persists_trace():
    if os.environ.get("RUN_V54_AGENT_L3") != "1":
        pytest.skip("set RUN_V54_AGENT_L3=1 for the V54 Agent acceptance")

    engine = create_engine(DATABASE_URL, pool_pre_ping=True)
    correlation_id = new_governance_uid()
    dataflow_uid = new_governance_uid()
    source_uid = new_governance_uid()
    identity = AgentIdentity(
        subject="v54-model-offline",
        roles=frozenset({"scheduler"}),
        business_domains=frozenset({"sales"}),
        environments=frozenset({"test"}),
        correlation_id=correlation_id,
    )
    planner = SchedulingPlanner(
        model=OfflineModel(),
        context_tools=ContextTools(),
        scheduling_tools=ForbiddenSchedulingTools(),
        canary_verifier=ForbiddenVerifier(),
        audit=PostgresAuditSink(engine),
        model_timeout_seconds=1,
    )

    try:
        result = planner.run(
            identity,
            {
                "business_goal": "Keep the published schedule stable",
                "dataflow_uid": dataflow_uid,
                "business_domain": "sales",
                "environment": "test",
                "available_window": {
                    "start": "2026-07-20T06:00:00+08:00",
                    "end": "2026-07-20T08:00:00+08:00",
                },
                "authorized_resources": [source_uid],
                "canary_inputs": {},
            },
        )

        assert result == {
            "status": "degraded",
            "mode": "fixed_schedule",
            "reason_code": "model_unavailable",
            "action": "continue_published_schedule",
            "formal_engine_changed": False,
            "correlation_id": correlation_id,
        }
        with engine.connect() as connection:
            row = (
                connection.execute(
                    text(
                        "SELECT actor_subject, model_provider, model_name, "
                        "prompt_version, schema_version, decision, decision_detail "
                        "FROM public.workflow_plan_audits "
                        "WHERE correlation_id = CAST(:id AS uuid) "
                        "AND action = 'ai_planning_closed_loop'"
                    ),
                    {"id": correlation_id},
                )
                .mappings()
                .one()
            )
        assert row["actor_subject"] == "v54-model-offline"
        assert row["model_provider"] == "deepseek"
        assert row["schema_version"] == "v54"
        assert row["decision"] == "failed"
        assert row["decision_detail"]["reason_code"] == "model_unavailable"
        assert "provider is unavailable" not in str(row["decision_detail"])
    finally:
        with engine.begin() as connection:
            connection.execute(
                text(
                    "DELETE FROM public.workflow_plan_audits "
                    "WHERE correlation_id = CAST(:id AS uuid)"
                ),
                {"id": correlation_id},
            )
        engine.dispose()