From 8927b192f805d8f0e1cf31572736173e0160bbb7 Mon Sep 17 00:00:00 2001 From: Juan Date: Sun, 10 May 2026 15:38:44 +0200 Subject: [PATCH] feat(api): implementa /executions con invoke + HITL approve/reject + persistencia MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - POST /agents/{name}/invoke, GET /executions, GET /executions/{trace_id}, POST /executions/{trace_id}/{approve,reject}. - execution_index.json persiste trace_id → (agent_name, version) para reconstruir el contexto en approve/reject tras un reinicio. - Refactor del orchestrator: split _snapshot → _build_execution + _snapshot, nuevo método público snapshot() (lee estado sin avanzar). Sustituye a los helpers _snapshot_execution/_build_graph_for_snapshot del plan, incompatibles con el checkpointer por-llamada. - Aliases Annotated[...] de dependencias movidos a deps.py. - Fixture mínima tests/fixtures/policies/default/. Co-Authored-By: Claude Opus 4.7 --- core/src/agentforge_core/api/agents.py | 9 +- core/src/agentforge_core/api/deps.py | 10 + core/src/agentforge_core/api/executions.py | 211 +++++++++++++++++- core/src/agentforge_core/main.py | 1 + .../agentforge_core/runtime/orchestrator.py | 72 +++--- tests/fixtures/policies/default/index.yaml | 8 + .../policies/default/versions/v1.yaml | 6 + tests/unit/test_api_executions.py | 124 ++++++++++ tests/unit/test_runtime_orchestrator.py | 18 ++ 9 files changed, 420 insertions(+), 39 deletions(-) create mode 100644 tests/fixtures/policies/default/index.yaml create mode 100644 tests/fixtures/policies/default/versions/v1.yaml create mode 100644 tests/unit/test_api_executions.py diff --git a/core/src/agentforge_core/api/agents.py b/core/src/agentforge_core/api/agents.py index 6ca0828..41aa31e 100644 --- a/core/src/agentforge_core/api/agents.py +++ b/core/src/agentforge_core/api/agents.py @@ -2,19 +2,14 @@ from __future__ import annotations -from typing import Annotated +from fastapi import APIRouter, HTTPException -from fastapi import APIRouter, Depends, HTTPException - -from agentforge_core.api.deps import get_registry +from agentforge_core.api.deps import RegistryDep from agentforge_core.domain.agent import AgentDefinition, AgentVersionMeta -from agentforge_core.registry.repository import FileSystemAgentRegistry from agentforge_core.registry.versioning import DiffResult router = APIRouter() -RegistryDep = Annotated[FileSystemAgentRegistry, Depends(get_registry)] - @router.get("", response_model=list[AgentDefinition]) def list_agents(registry: RegistryDep) -> list[AgentDefinition]: diff --git a/core/src/agentforge_core/api/deps.py b/core/src/agentforge_core/api/deps.py index b8e8931..a087202 100644 --- a/core/src/agentforge_core/api/deps.py +++ b/core/src/agentforge_core/api/deps.py @@ -8,6 +8,9 @@ tests llaman a `.cache_clear()` para reconstruirlos contra un `DATA_DIR` tempora from __future__ import annotations from functools import lru_cache +from typing import Annotated + +from fastapi import Depends from agentforge_core.config import Settings from agentforge_core.guardrails.base import GuardrailEngine @@ -53,3 +56,10 @@ def get_orchestrator() -> AgentOrchestrator: engine=get_guardrail_engine(), data_dir=settings.data_dir, ) + + +# Aliases `Annotated[...]` para inyección en los routers (evitan `Depends()` en defaults). +SettingsDep = Annotated[Settings, Depends(get_settings)] +RegistryDep = Annotated[FileSystemAgentRegistry, Depends(get_registry)] +PolicyStoreDep = Annotated[FileSystemPolicyStore, Depends(get_policy_store)] +OrchestratorDep = Annotated[AgentOrchestrator, Depends(get_orchestrator)] diff --git a/core/src/agentforge_core/api/executions.py b/core/src/agentforge_core/api/executions.py index 87a2df3..c49f204 100644 --- a/core/src/agentforge_core/api/executions.py +++ b/core/src/agentforge_core/api/executions.py @@ -1,5 +1,212 @@ -"""Router stub de /executions — se implementa en una task posterior.""" +"""Router /executions y POST /agents/{name}/invoke: ejecución, HITL approve/reject, listado.""" -from fastapi import APIRouter +from __future__ import annotations + +import json +from pathlib import Path +from uuid import UUID + +from fastapi import APIRouter, HTTPException +from pydantic import BaseModel, Field + +from agentforge_core.api.deps import ( + OrchestratorDep, + PolicyStoreDep, + RegistryDep, + SettingsDep, +) +from agentforge_core.api.persistence import ( + append_execution, + append_violation, + read_execution_summaries, +) +from agentforge_core.domain.agent import AgentDefinition +from agentforge_core.domain.execution import AgentExecution, AgentExecutionSummary +from agentforge_core.domain.policy import PolicyDefinition +from agentforge_core.registry.policy_store import FileSystemPolicyStore +from agentforge_core.registry.repository import FileSystemAgentRegistry +from agentforge_core.runtime.orchestrator import AgentOrchestrator router = APIRouter() +# POST /agents/{name}/invoke vive aquí pero se monta en main bajo el prefijo /agents. +invoke_router = APIRouter() + +_TERMINAL_STATUSES = {"completed", "failed", "blocked_by_guardrail"} + + +class InvokeRequest(BaseModel): + input: str + version: str | None = None + + +class ApproveRequest(BaseModel): + approved_action_ids: list[str] = Field(default_factory=list) + comment: str | None = None + + +class RejectRequest(BaseModel): + reason: str + + +# --- Índice trace_id → (agent_name, version), persistido en disco ---------------------- +# /approve y /reject solo reciben el trace_id; necesitan reconstruir qué agente/versión +# ejecutó. Se persiste para sobrevivir a reinicios del proceso con HITL pendiente. + + +def _index_path(data_dir: Path) -> Path: + return data_dir / "execution_index.json" + + +def _load_index(data_dir: Path) -> dict[str, dict[str, str]]: + path = _index_path(data_dir) + if not path.exists(): + return {} + data: dict[str, dict[str, str]] = json.loads(path.read_text(encoding="utf-8")) + return data + + +def _record_execution(data_dir: Path, trace_id: str, agent_name: str, version: str) -> None: + data_dir.mkdir(parents=True, exist_ok=True) + index = _load_index(data_dir) + index[trace_id] = {"agent_name": agent_name, "version": version} + _index_path(data_dir).write_text( + json.dumps(index, ensure_ascii=False, indent=2), encoding="utf-8" + ) + + +def _resolve( + registry: FileSystemAgentRegistry, + policies: FileSystemPolicyStore, + data_dir: Path, + trace_id: UUID, +) -> tuple[AgentDefinition, PolicyDefinition]: + """Recupera (agent_def, policy) de una ejecución registrada; 404 si no existe.""" + meta = _load_index(data_dir).get(str(trace_id)) + if meta is None: + raise HTTPException(status_code=404, detail="execution not found") + try: + agent_def = registry.get_version(meta["agent_name"], meta["version"]) + policy = policies.get_policy(agent_def.guardrails[0]) + except (FileNotFoundError, IndexError, KeyError) as exc: + raise HTTPException( + status_code=500, detail=f"execution refers to missing config: {exc}" + ) from exc + return agent_def, policy + + +async def _ensure_awaiting( + orchestrator: AgentOrchestrator, + agent_def: AgentDefinition, + policy: PolicyDefinition, + trace_id: UUID, +) -> None: + snap = await orchestrator.snapshot(agent_def=agent_def, policy=policy, trace_id=trace_id) + if snap is None: + raise HTTPException(status_code=404, detail="execution not found") + if snap.status != "awaiting_approval": + raise HTTPException( + status_code=409, + detail=f"execution status '{snap.status}' does not allow approve/reject", + ) + + +# --- Endpoints ------------------------------------------------------------------------- + + +@invoke_router.post("/{name}/invoke", response_model=AgentExecution) +async def invoke_agent( + name: str, + body: InvokeRequest, + registry: RegistryDep, + policies: PolicyStoreDep, + orchestrator: OrchestratorDep, + settings: SettingsDep, +) -> AgentExecution: + try: + agent_def = registry.get_agent(name, body.version) + except FileNotFoundError as exc: + raise HTTPException(status_code=404, detail=str(exc)) from exc + if not agent_def.guardrails: + raise HTTPException(status_code=422, detail="agent has no policy attached") + try: + policy = policies.get_policy(agent_def.guardrails[0]) + except FileNotFoundError as exc: + raise HTTPException(status_code=422, detail=str(exc)) from exc + + execution = await orchestrator.invoke( + agent_def=agent_def, policy=policy, user_input=body.input + ) + _record_execution( + settings.data_dir, str(execution.trace_id), agent_def.name, agent_def.version + ) + if execution.status in _TERMINAL_STATUSES: + append_execution(settings.data_dir, execution) + for violation in execution.violations: + append_violation(settings.data_dir, violation) + return execution + + +@router.get("", response_model=list[AgentExecutionSummary]) +def list_executions(settings: SettingsDep) -> list[AgentExecutionSummary]: + return read_execution_summaries(settings.data_dir) + + +@router.get("/{trace_id}", response_model=AgentExecution) +async def get_execution( + trace_id: UUID, + registry: RegistryDep, + policies: PolicyStoreDep, + orchestrator: OrchestratorDep, + settings: SettingsDep, +) -> AgentExecution: + agent_def, policy = _resolve(registry, policies, settings.data_dir, trace_id) + execution = await orchestrator.snapshot(agent_def=agent_def, policy=policy, trace_id=trace_id) + if execution is None: + raise HTTPException(status_code=404, detail="execution not found") + return execution + + +@router.post("/{trace_id}/approve", response_model=AgentExecution) +async def approve_execution( + trace_id: UUID, + body: ApproveRequest, + registry: RegistryDep, + policies: PolicyStoreDep, + orchestrator: OrchestratorDep, + settings: SettingsDep, +) -> AgentExecution: + agent_def, policy = _resolve(registry, policies, settings.data_dir, trace_id) + await _ensure_awaiting(orchestrator, agent_def, policy, trace_id) + execution = await orchestrator.resume( + agent_def=agent_def, + policy=policy, + trace_id=trace_id, + decision={ + "approved_action_ids": body.approved_action_ids, + "comment": body.comment, + "rejected": False, + }, + ) + append_execution(settings.data_dir, execution) + return execution + + +@router.post("/{trace_id}/reject", response_model=AgentExecution) +async def reject_execution( + trace_id: UUID, + body: RejectRequest, + registry: RegistryDep, + policies: PolicyStoreDep, + orchestrator: OrchestratorDep, + settings: SettingsDep, +) -> AgentExecution: + agent_def, policy = _resolve(registry, policies, settings.data_dir, trace_id) + await _ensure_awaiting(orchestrator, agent_def, policy, trace_id) + execution = await orchestrator.resume( + agent_def=agent_def, + policy=policy, + trace_id=trace_id, + decision={"approved_action_ids": [], "rejected": True, "reason": body.reason}, + ) + append_execution(settings.data_dir, execution) + return execution diff --git a/core/src/agentforge_core/main.py b/core/src/agentforge_core/main.py index 018d575..1d8efad 100644 --- a/core/src/agentforge_core/main.py +++ b/core/src/agentforge_core/main.py @@ -27,6 +27,7 @@ def create_app() -> FastAPI: from agentforge_core.api import agents, executions, policies, violations app.include_router(agents.router, prefix="/agents", tags=["agents"]) + app.include_router(executions.invoke_router, prefix="/agents", tags=["agents"]) app.include_router(executions.router, prefix="/executions", tags=["executions"]) app.include_router(policies.router, prefix="/policies", tags=["policies"]) app.include_router(violations.router, prefix="/violations", tags=["violations"]) diff --git a/core/src/agentforge_core/runtime/orchestrator.py b/core/src/agentforge_core/runtime/orchestrator.py index 6e9e817..aed377a 100644 --- a/core/src/agentforge_core/runtime/orchestrator.py +++ b/core/src/agentforge_core/runtime/orchestrator.py @@ -1,10 +1,11 @@ """Orchestrator: envuelve build_graph + ainvoke/resume + serialización a ``AgentExecution``. Es el único punto de entrada al runtime: el router de FastAPI lo usa para lanzar -ejecuciones y reanudar pausas Human-in-the-Loop. Cada llamada abre su propio +ejecuciones (``invoke``), reanudar pausas Human-in-the-Loop (``resume``) y leer el +estado actual sin avanzarlo (``snapshot``). Cada llamada abre su propio ``AsyncSqliteSaver`` (context manager) sobre ``data_dir/checkpoints.sqlite``, así que -el estado de un ``awaiting_approval`` sobrevive a un reinicio del proceso: basta crear -otro ``AgentOrchestrator`` apuntando al mismo ``data_dir`` y llamar a ``resume``. +un ``awaiting_approval`` sobrevive a un reinicio del proceso: basta crear otro +``AgentOrchestrator`` apuntando al mismo ``data_dir`` y llamar a ``resume``. """ from __future__ import annotations @@ -36,7 +37,7 @@ def _thread_config(trace_id: UUID) -> dict[str, Any]: class AgentOrchestrator: - """Punto único de entrada para invocar agentes y reanudar pausas HITL.""" + """Punto único de entrada para invocar agentes, reanudar HITL y leer su estado.""" def __init__( self, *, provider: LLMProvider, engine: GuardrailEngine, data_dir: Path @@ -45,6 +46,17 @@ class AgentOrchestrator: self._engine = engine self._data_dir = data_dir + def _build_graph( + self, agent_def: AgentDefinition, policy: PolicyDefinition, checkpointer: Any + ) -> Any: + return build_graph( + agent_def=agent_def, + policy=policy, + provider=self._provider, + engine=self._engine, + checkpointer=checkpointer, + ) + async def invoke( self, *, @@ -56,7 +68,6 @@ class AgentOrchestrator: """Lanza una ejecución. Si hay acciones que requieren aprobación, devuelve ``status="awaiting_approval"`` y el grafo queda pausado en el checkpointer.""" tid = trace_id or uuid4() - started_at = datetime.now(UTC) initial: dict[str, Any] = { "trace_id": str(tid), "agent_name": agent_def.name, @@ -74,20 +85,14 @@ class AgentOrchestrator: "final_output": None, } async with build_checkpointer(self._data_dir) as checkpointer: - graph = build_graph( - agent_def=agent_def, - policy=policy, - provider=self._provider, - engine=self._engine, - checkpointer=checkpointer, - ) + graph = self._build_graph(agent_def, policy, checkpointer) crashed = False try: await graph.ainvoke(initial, config=_thread_config(tid)) except Exception: log.exception("agent_invoke_failed", trace_id=str(tid)) crashed = True - return await self._snapshot(graph, agent_def, tid, started_at, crashed=crashed) + return await self._snapshot(graph, agent_def, tid, crashed=crashed) async def resume( self, @@ -98,34 +103,40 @@ class AgentOrchestrator: decision: dict[str, Any], ) -> AgentExecution: """Reanuda una ejecución pausada en HITL con la decisión del operador.""" - started_at = datetime.now(UTC) async with build_checkpointer(self._data_dir) as checkpointer: - graph = build_graph( - agent_def=agent_def, - policy=policy, - provider=self._provider, - engine=self._engine, - checkpointer=checkpointer, - ) + graph = self._build_graph(agent_def, policy, checkpointer) crashed = False try: await graph.ainvoke(Command(resume=decision), config=_thread_config(trace_id)) except Exception: log.exception("agent_resume_failed", trace_id=str(trace_id)) crashed = True - return await self._snapshot(graph, agent_def, trace_id, started_at, crashed=crashed) + return await self._snapshot(graph, agent_def, trace_id, crashed=crashed) + + async def snapshot( + self, *, agent_def: AgentDefinition, policy: PolicyDefinition, trace_id: UUID + ) -> AgentExecution | None: + """Lee el estado actual de una ejecución del checkpointer, sin avanzarla. + + Devuelve ``None`` si no hay estado guardado para ese ``trace_id``. + """ + async with build_checkpointer(self._data_dir) as checkpointer: + graph = self._build_graph(agent_def, policy, checkpointer) + state = await graph.aget_state(_thread_config(trace_id)) + if not state.values: + return None + return self._build_execution(state, agent_def, trace_id, crashed=False) async def _snapshot( - self, - graph: Any, - agent_def: AgentDefinition, - trace_id: UUID, - started_at: datetime, - *, - crashed: bool, + self, graph: Any, agent_def: AgentDefinition, trace_id: UUID, *, crashed: bool ) -> AgentExecution: - """Lee el estado del checkpointer y lo serializa a ``AgentExecution``.""" state = await graph.aget_state(_thread_config(trace_id)) + return self._build_execution(state, agent_def, trace_id, crashed=crashed) + + def _build_execution( + self, state: Any, agent_def: AgentDefinition, trace_id: UUID, *, crashed: bool + ) -> AgentExecution: + """Mapea un ``StateSnapshot`` de LangGraph a un ``AgentExecution``.""" values: dict[str, Any] = state.values or {} status: str = values.get("status", "running") if crashed and status not in _TERMINAL_STATUSES: @@ -147,6 +158,7 @@ class AgentOrchestrator: else None ) error = values.get("error") or ("internal_error" if crashed else None) + started_at = decision_path[0].timestamp if decision_path else datetime.now(UTC) finished_at = datetime.now(UTC) if status in _TERMINAL_STATUSES else None return AgentExecution( trace_id=trace_id, diff --git a/tests/fixtures/policies/default/index.yaml b/tests/fixtures/policies/default/index.yaml new file mode 100644 index 0000000..98b8a6b --- /dev/null +++ b/tests/fixtures/policies/default/index.yaml @@ -0,0 +1,8 @@ +name: default +versions: + - id: v1 + hash: testhashp1 + author: Juan + message: fixture mínima + created_at: 2026-04-01T00:00:00Z +active_version: v1 diff --git a/tests/fixtures/policies/default/versions/v1.yaml b/tests/fixtures/policies/default/versions/v1.yaml new file mode 100644 index 0000000..399323f --- /dev/null +++ b/tests/fixtures/policies/default/versions/v1.yaml @@ -0,0 +1,6 @@ +name: default +version: v1 +description: Política mínima para tests del core (sin validadores) +input_validators: [] +output_validators: [] +on_validator_error: fail_closed diff --git a/tests/unit/test_api_executions.py b/tests/unit/test_api_executions.py new file mode 100644 index 0000000..7d80638 --- /dev/null +++ b/tests/unit/test_api_executions.py @@ -0,0 +1,124 @@ +"""Tests del router /executions: invoke, get, list, approve, reject.""" + +from pathlib import Path + +import pytest +from fastapi.testclient import TestClient + +from agentforge_core.api import deps +from agentforge_core.main import create_app + +FIXTURES = Path(__file__).parent.parent / "fixtures" + + +@pytest.fixture(autouse=True) +def patch_settings(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("DATA_DIR", str(tmp_path)) + monkeypatch.setenv("AGENTS_DIR", str(FIXTURES / "agents")) + monkeypatch.setenv("POLICIES_DIR", str(FIXTURES / "policies")) + monkeypatch.setenv("LLM_PROVIDER", "mock") + for fn in ( + deps.get_settings, + deps.get_registry, + deps.get_policy_store, + deps.get_orchestrator, + deps.get_llm_provider, + deps.get_guardrail_engine, + ): + fn.cache_clear() + + +@pytest.fixture +def client() -> TestClient: + return TestClient(create_app()) + + +def test_invoke_devuelve_execution_completada(client: TestClient) -> None: + r = client.post("/agents/incident_analyzer/invoke", json={"input": "degradación MOS pool SBC"}) + assert r.status_code == 200 + body = r.json() + assert body["status"] == "completed" + assert body["agent_name"] == "incident_analyzer" + assert body["agent_version"] == "v1" + assert body["needs_human_for"] is None + + +def test_invoke_agente_inexistente_404(client: TestClient) -> None: + r = client.post("/agents/inexistente/invoke", json={"input": "x"}) + assert r.status_code == 404 + + +def test_invoke_riesgo_alto_pausa_hitl(client: TestClient) -> None: + r = client.post("/agents/incident_analyzer/invoke", json={"input": "caída registros sip"}) + assert r.status_code == 200 + body = r.json() + assert body["status"] == "awaiting_approval" + assert body["needs_human_for"] + assert {a["id"] for a in body["needs_human_for"]} == {"act-1"} + + +def test_get_execution_existe_tras_invoke(client: TestClient) -> None: + trace_id = client.post( + "/agents/incident_analyzer/invoke", json={"input": "degradación MOS pool SBC"} + ).json()["trace_id"] + r = client.get(f"/executions/{trace_id}") + assert r.status_code == 200 + assert r.json()["trace_id"] == trace_id + + +def test_get_execution_inexistente_404(client: TestClient) -> None: + r = client.get("/executions/00000000-0000-0000-0000-000000000000") + assert r.status_code == 404 + + +def test_list_executions_incluye_la_completada(client: TestClient) -> None: + client.post("/agents/incident_analyzer/invoke", json={"input": "degradación MOS pool SBC"}) + r = client.get("/executions") + assert r.status_code == 200 + rows = r.json() + assert len(rows) == 1 + assert rows[0]["status"] == "completed" + assert rows[0]["n_proposed_actions"] == 1 + + +def test_approve_resume_completa_ejecucion(client: TestClient) -> None: + invoked = client.post( + "/agents/incident_analyzer/invoke", json={"input": "caída registros sip"} + ).json() + trace_id = invoked["trace_id"] + action_ids = [a["id"] for a in invoked["needs_human_for"]] + r = client.post( + f"/executions/{trace_id}/approve", + json={"approved_action_ids": action_ids, "comment": "OK adelante"}, + ) + assert r.status_code == 200 + body = r.json() + assert body["status"] == "completed" + assert {a["id"] for a in body["final_output"]["approved_actions"]} == {"act-1"} + + +def test_reject_marca_failed(client: TestClient) -> None: + trace_id = client.post( + "/agents/incident_analyzer/invoke", json={"input": "caída registros sip"} + ).json()["trace_id"] + r = client.post(f"/executions/{trace_id}/reject", json={"reason": "no procede ahora"}) + assert r.status_code == 200 + body = r.json() + assert body["status"] == "failed" + assert "rejected_by_human" in body["error"] + + +def test_approve_sobre_estado_no_pausado_da_409(client: TestClient) -> None: + trace_id = client.post( + "/agents/incident_analyzer/invoke", json={"input": "degradación MOS pool SBC"} + ).json()["trace_id"] + r = client.post(f"/executions/{trace_id}/approve", json={"approved_action_ids": []}) + assert r.status_code == 409 + + +def test_approve_trace_id_desconocido_da_404(client: TestClient) -> None: + r = client.post( + "/executions/11111111-1111-1111-1111-111111111111/approve", + json={"approved_action_ids": []}, + ) + assert r.status_code == 404 diff --git a/tests/unit/test_runtime_orchestrator.py b/tests/unit/test_runtime_orchestrator.py index 6c28452..26dc3d9 100644 --- a/tests/unit/test_runtime_orchestrator.py +++ b/tests/unit/test_runtime_orchestrator.py @@ -2,6 +2,7 @@ from datetime import UTC, datetime from pathlib import Path +from uuid import uuid4 import pytest @@ -64,6 +65,23 @@ async def test_invoke_camino_feliz(orchestrator: AgentOrchestrator) -> None: ] +async def test_snapshot_relee_estado_sin_avanzar(orchestrator: AgentOrchestrator) -> None: + agent = _agent() + paused = await orchestrator.invoke( + agent_def=agent, policy=_policy(), user_input="caída registros sip" + ) + again = await orchestrator.snapshot(agent_def=agent, policy=_policy(), trace_id=paused.trace_id) + assert again is not None + assert again.status == "awaiting_approval" + assert again.trace_id == paused.trace_id + # Idempotente: leer no añade pasos al decision_path. + assert [s.step for s in again.decision_path] == [s.step for s in paused.decision_path] + + +async def test_snapshot_trace_id_desconocido_devuelve_none(orchestrator: AgentOrchestrator) -> None: + assert await orchestrator.snapshot(agent_def=_agent(), policy=_policy(), trace_id=uuid4()) is None + + async def test_invoke_pausa_hitl_y_resume_aprueba(orchestrator: AgentOrchestrator) -> None: agent = _agent() paused = await orchestrator.invoke(