hermes-hub/tests/test_workflow_service_a30.py

from __future__ import annotations

import json

import pytest

from antigravity_provider.router.router_config import (
    RolePolicy,
    RouterConfig,
    RouterProfileConfig,
    save_router_config,
)
from antigravity_provider.router.workflow_service import WorkflowService


# Роль в фикстурах названа test-developer намеренно: "developer" — это
# псевдоним канонической роли developer-1, и при загрузке конфигурации он
# переименовывается. Тест проверяет механику workflow, а не работу псевдонимов,
# поэтому имя взято такое же нейтральное, как у соседнего test-reviewer.
@pytest.fixture
def workflow_service(tmp_path, monkeypatch):
    monkeypatch.setenv("HERMES_HOME", str(tmp_path))
    config = RouterConfig(
        profiles={
            "account-a": RouterProfileConfig(
                profile_id="account-a",
                provider="openai-codex",
                preferred_models=["model-real-from-config"],
            )
        },
        roles={"test-developer": RolePolicy(role_name="test-developer", preferred_chain=["account-a"])},
    )
    save_router_config(config)
    return WorkflowService(tmp_path / "workflow_state.json")


def test_router_roles_migrate_to_agents_and_create_real_files(workflow_service, tmp_path):
    snapshot = workflow_service.snapshot()
    assert "test-developer" in [agent["id"] for agent in snapshot["agents"]]
    agent = next(item for item in snapshot["agents"] if item["id"] == "test-developer")
    assert agent["agent_file"] == "agents/test-developer.md"
    assert agent["agent_file_exists"] is True
    assert (tmp_path / "agents" / "test-developer.md").is_file()
    assert agent["execution_config"]["account"] == "account-a"
    assert agent["execution_config"]["model"] == "model-real-from-config"


def test_create_update_file_and_restart_persistence(workflow_service, tmp_path):
    created = workflow_service.create_agent({
        "name": "Security Reviewer",
        "role": "security-reviewer",
        "account": "account-a",
        "model": "model-real-from-config",
        "description": "Проверяет безопасность",
    })
    workflow_service.save_agent_file(created.id, "# Security\n\nOnly measured facts.")
    workflow_service.update_agent(created.id, {"timeout": 91, "position": {"x": 33, "y": 44}})

    restarted = WorkflowService(tmp_path / "workflow_state.json")
    item = next(agent for agent in restarted.snapshot()["agents"] if agent["id"] == created.id)
    assert item["timeout"] == 91
    assert item["position"] == {"x": 33, "y": 44}
    assert restarted.read_agent_file(created.id)["content"].endswith("Only measured facts.")


def test_delete_requires_explicit_confirmation_when_referenced(workflow_service):
    workflow_service.create_agent({"name": "Test Reviewer", "role": "test-reviewer", "account": "account-a"})
    workflow_service.save_workflow({
        "start_agent_id": "test-developer",
        "max_iterations": 3,
        "edges": [{"id": "review", "source": "test-developer", "target": "test-reviewer", "condition": "SUCCESS"}],
    })
    warning = workflow_service.delete_agent("test-reviewer")
    assert warning["confirmation_required"] is True
    assert warning["consequences"]["workflow_edges"] == ["review"]
    result = workflow_service.delete_agent("test-reviewer", force=True)
    assert result["deleted"] is True
    assert workflow_service.workflow.edges == []


def test_cycles_are_valid_and_iteration_limit_is_persisted(workflow_service):
    workflow_service.create_agent({"name": "Test Reviewer", "role": "test-reviewer", "account": "account-a"})
    definition = workflow_service.save_workflow({
        "start_agent_id": "test-developer",
        "max_iterations": 2,
        "edges": [
            {"source": "test-developer", "target": "test-reviewer", "condition": "SUCCESS"},
            {"source": "test-reviewer", "target": "test-developer", "condition": "REVIEW_FAILED"},
        ],
    })
    assert definition.max_iterations == 2
    assert definition.edges[1].condition == "REVIEW_FAILED"
    payload = json.loads(workflow_service.state_path.read_text(encoding="utf-8"))
    assert payload["workflow"]["max_iterations"] == 2


def test_invalid_edge_and_unknown_model_are_rejected(workflow_service):
    with pytest.raises(ValueError, match="отсутствующего агента"):
        workflow_service.save_workflow({"edges": [{"source": "test-developer", "target": "missing"}]})
    with pytest.raises(ValueError, match="не доступна"):
        workflow_service.update_agent("test-developer", {"account": "account-a", "model": "invented-model"})


def test_interrupted_run_is_reported_not_silently_completed(workflow_service, tmp_path):
    workflow_service.run.update({"id": "run-1", "status": "running", "current_agent_id": "test-developer"})
    workflow_service._save()
    restarted = WorkflowService(tmp_path / "workflow_state.json")
    assert restarted.run["status"] == "interrupted"
    assert "перезапуском" in restarted.run["error"]
    assert restarted.events[-1].type == "WORKFLOW_INTERRUPTED"


def test_live_cycle_stops_with_explicit_iteration_limit_event(workflow_service, monkeypatch):
    workflow_service.create_agent({"name": "Loop Reviewer", "role": "loop-reviewer", "account": "account-a"})
    workflow_service.save_workflow({
        "start_agent_id": "test-developer",
        "max_iterations": 2,
        "edges": [
            {"source": "test-developer", "target": "loop-reviewer", "condition": "SUCCESS"},
            {"source": "loop-reviewer", "target": "test-developer", "condition": "REVIEW_FAILED"},
        ],
    })

    class FakeEngine:
        def reload_config(self):
            return None

        def route_request(self, request, role=None, session_id=None):
            status = "REVIEW_FAILED" if role == "loop-reviewer" else "SUCCESS"
            return {
                "choices": [{"message": {"content": status}}],
                "router_metadata": {
                    "provider": "measured-provider",
                    "profile_id": "account-a",
                    "selection_trace": {"selected_model": "model-real-from-config"},
                },
            }

    monkeypatch.setattr("antigravity_provider.router.router_engine.get_router_engine", lambda: FakeEngine())
    workflow_service.start("Проверить реальный цикл")
    thread = workflow_service._thread
    assert thread is not None
    thread.join(timeout=3)
    assert not thread.is_alive(), "Workflow execution thread leaked after join"

    assert workflow_service.run["status"] == "failed"
    assert workflow_service.run["error"] == "Достигнут предел итераций: 2"
    assert any(event.type == "WORKFLOW_MAX_ITERATIONS" for event in workflow_service.events)


def test_provider_error_text_reaches_run_and_events(workflow_service, monkeypatch):
    provider_text = "Provider Error: authentication token missing for account-a"

    class ErrorEngine:
        def reload_config(self):
            return None

        def route_request(self, request, role=None, session_id=None):
            return {"choices": [{"message": {"content": f"ERROR\n{provider_text}"}}]}

    monkeypatch.setattr("antigravity_provider.router.router_engine.get_router_engine", lambda: ErrorEngine())
    workflow_service.workflow.start_agent_id = "test-developer"
    workflow_service.workflow.edges = []
    workflow_service.start("Проверить ошибку")
    thread = workflow_service._thread
    assert thread is not None
    thread.join(timeout=3)
    assert not thread.is_alive(), "Workflow execution thread leaked after join"

    assert workflow_service.run["status"] == "failed"
    assert provider_text in workflow_service.run["error"]
    assert any(provider_text in (event.error or "") for event in workflow_service.events)
    assert workflow_service.snapshot()["agents"][0]["runtime_state"] == "error"