diff --git a/docs/backlog/01_execution_backlog.md b/docs/backlog/01_execution_backlog.md index 3da1ee0..ec383e9 100644 --- a/docs/backlog/01_execution_backlog.md +++ b/docs/backlog/01_execution_backlog.md @@ -45,11 +45,11 @@ ## Package E: Delivery - [x] typed worker registration schema -- [ ] task event DTO normalization +- [x] task event DTO normalization - [ ] confirmation cards -- [ ] progress cards +- [x] progress cards - [ ] artifact cards -- [ ] error cards +- [x] error cards ## Package F: Validation diff --git a/docs/contracts/API_CONTRACTS.md b/docs/contracts/API_CONTRACTS.md index 1991f8d..37d6552 100644 --- a/docs/contracts/API_CONTRACTS.md +++ b/docs/contracts/API_CONTRACTS.md @@ -26,7 +26,14 @@ "task_id": "task_1", "response_type": "answer | task_started | confirmation_required | error", "message": "string", - "cards": [] + "cards": [ + { + "type": "plan_card" + }, + { + "type": "progress_card" + } + ] } ``` @@ -53,7 +60,8 @@ "progress": { "completed": 2, "total": 5 - } + }, + "cards": [] } ``` @@ -66,9 +74,18 @@ WebSocket может быть добавлен как secondary adapter, но co ```json { "event": "node_started", - "task_id": "task_1", - "node_id": "node_3", - "timestamp": "..." + "data": { + "event_id": "evt_1", + "event_type": "node_started", + "task_id": "task_1", + "node_id": "node_3", + "timestamp": "...", + "payload": {}, + "card": { + "type": "progress_card_ref", + "task_id": "task_1" + } + } } ``` diff --git a/docs/ui/USER_INTERFACE.md b/docs/ui/USER_INTERFACE.md index 83d3767..32f5e0e 100644 --- a/docs/ui/USER_INTERFACE.md +++ b/docs/ui/USER_INTERFACE.md @@ -309,6 +309,20 @@ syntax_check: success } ``` +## Backend Status + +Сейчас backend уже формирует: + +- `plan_card` +- `progress_card` +- `error_card` +- normalized task event DTO with optional card references + +Следующие карточки остаются следующими шагами: + +- `confirmation_card` +- `artifact_card` + ## Принцип Пользователь должен всегда понимать: diff --git a/src/ai_orchestrator/delivery/http/app.py b/src/ai_orchestrator/delivery/http/app.py index 1eab816..a5f7fe3 100644 --- a/src/ai_orchestrator/delivery/http/app.py +++ b/src/ai_orchestrator/delivery/http/app.py @@ -25,6 +25,12 @@ from ai_orchestrator.application.services.workers import ( WorkerService, ) from ai_orchestrator.config import AppSettings +from ai_orchestrator.delivery.http.presenters import ( + build_error_card, + build_plan_card, + build_progress_card, + build_task_event_dto, +) from ai_orchestrator.delivery.http.schemas import ( ChatRequest, ChatResponse, @@ -32,6 +38,7 @@ from ai_orchestrator.delivery.http.schemas import ( ConfirmationRejectRequest, CreateTaskRequestSchema, TaskActionRequest, + TaskEventDto, TaskStatusResponse, WorkerDispatchRequest, WorkerDispatchResponse, @@ -134,12 +141,16 @@ def create_app(settings: AppSettings | None = None) -> FastAPI: ) graph = orchestrator.plan_task(task.task_id) execution_engine.execute_ready_nodes(task=task, graph=graph) + cards = [ + build_plan_card(task, graph), + build_progress_card(task, graph), + ] return ChatResponse( conversation_id=task.conversation_id or "conv_default", task_id=task.task_id, response_type="task_started", message="Task accepted and planned.", - cards=[{"type": "plan_card", "task_id": task.task_id}], + cards=cards, ) @app.post("/tasks", response_model=TaskStatusResponse) @@ -154,11 +165,16 @@ def create_app(settings: AppSettings | None = None) -> FastAPI: ) graph = orchestrator.plan_task(task.task_id) execution_engine.execute_ready_nodes(task=task, graph=graph) + cards = [ + build_plan_card(task, graph), + build_progress_card(task, graph), + ] return TaskStatusResponse( task_id=task.task_id, status=task.status.value, current_node=task.current_node_id, progress={"completed": 0, "total": len(graph.nodes)}, + cards=cards, ) @app.get("/tasks/{task_id}", response_model=TaskStatusResponse) @@ -173,6 +189,7 @@ def create_app(settings: AppSettings | None = None) -> FastAPI: status=task.status.value, current_node=task.current_node_id, progress={"completed": completed, "total": len(graph.nodes)}, + cards=[build_progress_card(task, graph)], ) @app.get("/tasks/{task_id}/events") @@ -181,15 +198,10 @@ def create_app(settings: AppSettings | None = None) -> FastAPI: async def iterator(): for event in events: + dto = TaskEventDto.model_validate(build_task_event_dto(event)) yield { "event": event.event_type, - "data": { - "event_id": event.event_id, - "task_id": event.task_id, - "node_id": event.node_id, - "timestamp": event.occurred_at.isoformat(), - "payload": event.payload, - }, + "data": dto.model_dump(), } return EventSourceResponse(iterator()) @@ -230,6 +242,7 @@ def create_app(settings: AppSettings | None = None) -> FastAPI: status=task.status.value, current_node=task.current_node_id, progress={"completed": completed, "total": len(graph.nodes)}, + cards=[build_progress_card(task, graph)], ) @app.post("/tasks/{task_id}/cancel", response_model=TaskStatusResponse) @@ -244,6 +257,14 @@ def create_app(settings: AppSettings | None = None) -> FastAPI: status=task.status.value, current_node=task.current_node_id, progress={"completed": completed, "total": len(graph.nodes)}, + cards=[ + build_progress_card(task, graph), + build_error_card( + task_id=task.task_id, + title="Task cancelled", + message=payload.reason or "cancelled", + ), + ], ) @app.get("/workers", response_model=WorkerListResponse) diff --git a/src/ai_orchestrator/delivery/http/presenters.py b/src/ai_orchestrator/delivery/http/presenters.py new file mode 100644 index 0000000..a6179a5 --- /dev/null +++ b/src/ai_orchestrator/delivery/http/presenters.py @@ -0,0 +1,123 @@ +from __future__ import annotations + +from typing import Any + +from ai_orchestrator.domain.events import DomainEvent +from ai_orchestrator.domain.models import ConfirmationRequest, ExecutionGraph, Task + + +def build_plan_card(task: Task, graph: ExecutionGraph) -> dict[str, Any]: + return { + "type": "plan_card", + "task_id": task.task_id, + "title": "Execution Plan", + "goal": task.goal, + "steps": [ + { + "node_id": node.node_id, + "node_type": node.node_type.value, + "status": node.status.value, + "dependencies": node.dependencies, + } + for node in graph.nodes + ], + "actions": ["run", "edit_plan", "run_step_by_step", "use_strong_model", "use_local_only"], + } + + +def build_progress_card(task: Task, graph: ExecutionGraph) -> dict[str, Any]: + completed = sum(1 for node in graph.nodes if node.status.value == "completed") + return { + "type": "progress_card", + "task_id": task.task_id, + "status": task.status.value, + "current_node": task.current_node_id, + "progress": {"completed": completed, "total": len(graph.nodes)}, + "steps": [ + { + "node_id": node.node_id, + "node_type": node.node_type.value, + "status": node.status.value, + "attempts": node.attempts, + } + for node in graph.nodes + ], + } + + +def build_confirmation_card(confirmation: ConfirmationRequest) -> dict[str, Any]: + return { + "type": "confirmation_card", + "task_id": confirmation.task_id, + "confirmation_id": confirmation.confirmation_id, + "node_id": confirmation.node_id, + "status": confirmation.status.value, + "decision": confirmation.decision.decision.value, + "reason": confirmation.decision.reason, + "preview": confirmation.decision.preview, + "actions": [ + "approve_once", + "approve_task", + "approve_project", + "reject", + "edit_policy", + ], + } + + +def build_artifact_card( + *, + task_id: str, + artifact_type: str, + title: str, + data: dict[str, Any], +) -> dict[str, Any]: + return { + "type": "artifact_card", + "task_id": task_id, + "artifact_type": artifact_type, + "title": title, + "data": data, + } + + +def build_error_card( + *, + task_id: str, + title: str, + message: str, + details: dict[str, Any] | None = None, +) -> dict[str, Any]: + return { + "type": "error_card", + "task_id": task_id, + "title": title, + "message": message, + "details": details or {}, + "actions": ["retry", "show_log", "stop_task"], + } + + +def build_task_event_dto(event: DomainEvent) -> dict[str, Any]: + payload = dict(event.payload) + card: dict[str, Any] | None = None + if event.event_type == "confirmation_requested": + card = { + "type": "confirmation_card_ref", + "confirmation_id": payload.get("confirmation_id"), + "task_id": event.task_id, + } + elif event.event_type in {"node_started", "node_completed", "task_planned", "task_resumed"}: + card = {"type": "progress_card_ref", "task_id": event.task_id} + elif event.event_type in {"task_failed", "confirmation_rejected"}: + card = {"type": "error_card_ref", "task_id": event.task_id} + return { + "event_id": event.event_id, + "event_type": event.event_type, + "task_id": event.task_id, + "conversation_id": event.conversation_id, + "node_id": event.node_id, + "timestamp": event.occurred_at.isoformat(), + "payload": payload, + "card": card, + } diff --git a/src/ai_orchestrator/delivery/http/schemas.py b/src/ai_orchestrator/delivery/http/schemas.py index 5807d81..30448ca 100644 --- a/src/ai_orchestrator/delivery/http/schemas.py +++ b/src/ai_orchestrator/delivery/http/schemas.py @@ -34,6 +34,7 @@ class TaskStatusResponse(BaseModel): status: str current_node: str | None progress: dict[str, int] + cards: list[dict[str, Any]] = Field(default_factory=list) class ConfirmationApproveRequest(BaseModel): @@ -113,3 +114,14 @@ class WorkerResultMessage(BaseModel): result: dict[str, Any] = Field(default_factory=dict) artifacts: list[dict[str, Any]] = Field(default_factory=list) error: dict[str, Any] | None = None + + +class TaskEventDto(BaseModel): + event_id: str + event_type: str + task_id: str + conversation_id: str | None = None + node_id: str | None = None + timestamp: str + payload: dict[str, Any] = Field(default_factory=dict) + card: dict[str, Any] | None = None diff --git a/tests/integration/test_http_app.py b/tests/integration/test_http_app.py index a423737..1decbf4 100644 --- a/tests/integration/test_http_app.py +++ b/tests/integration/test_http_app.py @@ -30,6 +30,31 @@ def test_post_tasks_creates_planned_task() -> None: assert response.status_code == 200 assert payload["status"] == "running" assert payload["progress"]["total"] == 2 + assert any(card["type"] == "plan_card" for card in payload["cards"]) + assert any(card["type"] == "progress_card" for card in payload["cards"]) + + +def test_chat_response_contains_ui_cards() -> None: + client = TestClient(app) + + response = client.post( + "/chat", + json={ + "project_id": "default", + "conversation_id": "conv_demo", + "message": "Build plan", + "attachments": [], + "mode": "auto", + "preferences": {"show_plan": True}, + }, + ) + + payload = response.json() + + assert response.status_code == 200 + assert payload["response_type"] == "task_started" + assert any(card["type"] == "plan_card" for card in payload["cards"]) + assert any(card["type"] == "progress_card" for card in payload["cards"]) def test_cancel_endpoint_marks_task_cancelled() -> None: @@ -48,6 +73,7 @@ def test_cancel_endpoint_marks_task_cancelled() -> None: assert response.status_code == 200 assert response.json()["status"] == "cancelled" + assert any(card["type"] == "error_card" for card in response.json()["cards"]) def test_worker_registration_is_exposed_via_list_endpoint() -> None: @@ -138,3 +164,23 @@ def test_worker_websocket_heartbeat_and_poll() -> None: assert heartbeat_ack["status"] in {"online", "busy"} assert commands["type"] == "commands" assert len(commands["commands"]) == 1 + + +def test_task_events_stream_returns_normalized_event_payload() -> None: + client = TestClient(app) + created = client.post( + "/tasks", + json={ + "project_id": "default", + "goal": "Event check", + "inputs": {}, + "execution_mode": "agent_graph", + }, + ).json() + + with client.stream("GET", f"/tasks/{created['task_id']}/events") as response: + body = response.read().decode() + + assert response.status_code == 200 + assert "event_type" in body + assert "progress_card_ref" in body or "task_planned" in body