From 704dabf7862f8642c73b6846d539c2641326559a Mon Sep 17 00:00:00 2001 From: cafitac Date: Fri, 1 May 2026 21:28:00 +0900 Subject: [PATCH] refactor: map task lifecycle events to adapter DTOs --- .../pr-03-task-event-mapping.md | 8 + .../orchestrator-adapter-architecture.md | 3 +- .dev/refactor/roadmap.md | 2 +- CHANGELOG.md | 1 + hermit_agent/orchestrators/__init__.py | 8 + hermit_agent/orchestrators/events.py | 166 ++++++++++++++++++ tests/test_orchestrator_event_mapping.py | 159 +++++++++++++++++ 7 files changed, 345 insertions(+), 2 deletions(-) create mode 100644 hermit_agent/orchestrators/events.py create mode 100644 tests/test_orchestrator_event_mapping.py diff --git a/.dev/refactor/hermes-executor-readiness/pr-03-task-event-mapping.md b/.dev/refactor/hermes-executor-readiness/pr-03-task-event-mapping.md index 1d9626d..690579b 100644 --- a/.dev/refactor/hermes-executor-readiness/pr-03-task-event-mapping.md +++ b/.dev/refactor/hermes-executor-readiness/pr-03-task-event-mapping.md @@ -121,6 +121,14 @@ Expected: ## Acceptance criteria +Status: implemented in PR-03. + +- Added `hermit_agent/orchestrators/events.py` with pure mapping helpers for Gateway SSE events, existing channel actions, and task status payloads. +- Exported the helpers from `hermit_agent.orchestrators` for later adapter extraction slices. +- Covered submitted/running/progress/waiting/done/error/cancelled and unknown-event behavior in `tests/test_orchestrator_event_mapping.py`. +- Existing runtime delivery behavior remains unchanged; no MCP/channel dispatch code was rewired. + +Acceptance criteria: - Existing runtime delivery behavior is unchanged. - Current event/prompt metadata can be represented in neutral DTOs. - Mapping helpers are tested and defensive against unknown payloads. diff --git a/.dev/refactor/orchestrator-adapter-architecture.md b/.dev/refactor/orchestrator-adapter-architecture.md index 0294fa8..b8cc610 100644 --- a/.dev/refactor/orchestrator-adapter-architecture.md +++ b/.dev/refactor/orchestrator-adapter-architecture.md @@ -126,10 +126,11 @@ Current test anchor: - `tests/test_orchestrator_contracts.py` verifies immutable DTO shape, stable status/event enum values, and the structural lifecycle shape of `OrchestratorAdapter`. - `tests/test_hermes_orchestrator_adapter.py` verifies `HermesMcpAdapter` maps existing Hermes install, live-smoke, and doctor helpers into `AdapterInstallResult` / `AdapterHealth` without mutating behavior. - `tests/test_orchestrator_prompt_mapping.py` verifies current runtime prompts can round-trip through the neutral `InteractivePrompt` / `PromptReply` DTOs while preserving method/request/thread metadata. +- `tests/test_orchestrator_event_mapping.py` verifies current Gateway SSE events, channel actions, and task status payloads can map into neutral `TaskEvent` DTOs without rewiring delivery behavior. Next extraction candidates: - start equivalent Claude/Codex install and health wrappers after the Hermes wrapper shape proves stable -- map task lifecycle/status payloads into `TaskEvent` before moving channel delivery behind adapters +- keep MCP server as the canonical runtime boundary unless a later smoke proves an adapter-owned runtime path is safer ## Non-goals diff --git a/.dev/refactor/roadmap.md b/.dev/refactor/roadmap.md index 80e1191..bac37d9 100644 --- a/.dev/refactor/roadmap.md +++ b/.dev/refactor/roadmap.md @@ -97,7 +97,7 @@ Acceptance: ### R2.2 Move orchestrator-specific code behind adapters -Status: first Hermes wrapper added as `HermesMcpAdapter`; it maps existing install/doctor/live-smoke helpers to adapter DTOs but does not yet change CLI dispatch or MCP task runtime paths. Prompt mapping helpers now round-trip current runtime `InteractivePrompt` values through the neutral DTOs as another behavior-preserving extraction step. +Status: first Hermes wrapper added as `HermesMcpAdapter`; it maps existing install/doctor/live-smoke helpers to adapter DTOs but does not yet change CLI dispatch or MCP task runtime paths. Prompt mapping helpers now round-trip current runtime `InteractivePrompt` values through the neutral `InteractivePrompt` / `PromptReply` DTOs. Task event mapping helpers now convert current Gateway SSE events, channel actions, and task status payloads into neutral `TaskEvent` DTOs as another behavior-preserving extraction step. Likely areas: - `hermit_agent/mcp_channel.py` diff --git a/CHANGELOG.md b/CHANGELOG.md index 9f64fa6..6d84d92 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -45,6 +45,7 @@ - Added an orchestrator-neutral adapter contract scaffold under `hermit_agent.orchestrators` so Claude Code, Codex, and Hermes integration work can converge on shared task, event, prompt, health, and install DTOs before larger runtime refactors. - Added a thin `HermesMcpAdapter` wrapper that maps existing Hermes print/fix/live-smoke and doctor helpers onto the shared adapter DTOs without rewiring current CLI or MCP runtime behavior. - Added adapter prompt mapping helpers that round-trip the current runtime `InteractivePrompt` shape through the neutral `InteractivePrompt` / `PromptReply` DTOs without changing interactive delivery behavior. +- Added pure task event mapping helpers that convert existing Gateway SSE events, channel actions, and task status payloads into neutral `TaskEvent` DTOs without changing MCP/channel delivery behavior. - Started tracking the long-term refactor roadmap under `.dev/refactor/**` while leaving other `.dev/*` scratch files ignored. - Removed warning-producing test mocks for setup coroutines so the suite no longer emits unawaited `init_db` / `create_api_key` runtime warnings. - Aligned contributor and harness docs on `.venv/bin/python -m pytest tests/` and added a repo hygiene regression so stale `.venv/bin/pytest` shebangs are not reintroduced as the documented path. diff --git a/hermit_agent/orchestrators/__init__.py b/hermit_agent/orchestrators/__init__.py index 32e9587..5065c5f 100644 --- a/hermit_agent/orchestrators/__init__.py +++ b/hermit_agent/orchestrators/__init__.py @@ -13,6 +13,11 @@ TaskHandle, TaskRequest, ) +from .events import ( + channel_action_to_task_event, + sse_event_to_task_event, + task_status_payload_to_task_event, +) from .hermes import HermesMcpAdapter from .prompts import ( adapter_prompt_to_runtime_prompt, @@ -34,6 +39,9 @@ "TaskHandle", "TaskRequest", "adapter_prompt_to_runtime_prompt", + "channel_action_to_task_event", "prompt_reply_from_answer", "runtime_prompt_to_adapter_prompt", + "sse_event_to_task_event", + "task_status_payload_to_task_event", ] diff --git a/hermit_agent/orchestrators/events.py b/hermit_agent/orchestrators/events.py new file mode 100644 index 0000000..d0bee5d --- /dev/null +++ b/hermit_agent/orchestrators/events.py @@ -0,0 +1,166 @@ +"""Pure mapping helpers for orchestrator-neutral task lifecycle events.""" + +from __future__ import annotations + +from collections.abc import Mapping +from typing import Any + +from .contracts import TaskEvent, TaskEventKind + + +def _copy_payload(payload: Mapping[str, Any]) -> dict[str, Any]: + return dict(payload) + + +def _string_value(value: Any, default: str = "") -> str: + if value is None: + return default + return str(value) + + +def _options_tuple(value: Any) -> tuple[str, ...]: + if value is None: + return () + if isinstance(value, str): + return (value,) + try: + return tuple(str(item) for item in value) + except TypeError: + return (str(value),) + + +def _prompt_payload(event: Mapping[str, Any]) -> dict[str, Any]: + payload = _copy_payload(event) + event_type = _string_value(event.get("type") or event.get("kind") or event.get("status"), "waiting") + payload["prompt_kind"] = event_type + if "options" in payload: + payload["options"] = _options_tuple(payload.get("options")) + if "tool" in payload and "tool_name" not in payload: + payload["tool_name"] = payload.pop("tool") + if event_type == "waiting" and "tool_name" not in payload: + payload["tool_name"] = "ask" + if event_type == "permission_ask" and "tool_name" not in payload: + payload["tool_name"] = "bash" + return payload + + +def sse_event_to_task_event(task_id: str, event: Mapping[str, Any]) -> TaskEvent | None: + """Convert a Gateway SSE payload into a neutral TaskEvent. + + This is intentionally pure and side-effect free. It mirrors the existing + runtime event shapes without changing how MCP/channel delivery works. + Unknown/non-lifecycle streaming events return None so callers can ignore + them without inventing misleading lifecycle states. + """ + + event_type = _string_value(event.get("type")) + payload = _copy_payload(event) + + if event_type in {"submitted"}: + return TaskEvent(task_id=task_id, kind=TaskEventKind.SUBMITTED, payload=payload) + if event_type in {"reply_ack", "running"}: + return TaskEvent(task_id=task_id, kind=TaskEventKind.RUNNING, payload=payload) + if event_type in {"progress", "status", "tool_use", "tool_result", "model_changed"}: + return TaskEvent( + task_id=task_id, + kind=TaskEventKind.PROGRESS, + message=_string_value(event.get("message") or event.get("detail") or event.get("content")), + payload=payload, + ) + if event_type in {"waiting", "permission_ask"}: + question = _string_value(event.get("question")) + return TaskEvent( + task_id=task_id, + kind=TaskEventKind.WAITING, + message=question, + payload=_prompt_payload(event), + ) + if event_type == "done": + result = _string_value(event.get("result")) + return TaskEvent(task_id=task_id, kind=TaskEventKind.DONE, message=result, payload=payload) + if event_type == "error": + message = _string_value(event.get("message")) + return TaskEvent(task_id=task_id, kind=TaskEventKind.ERROR, message=message, payload=payload) + if event_type == "cancelled": + message = _string_value(event.get("message"), "Task cancelled") or "Task cancelled" + return TaskEvent(task_id=task_id, kind=TaskEventKind.CANCELLED, message=message, payload=payload) + return None + + +def channel_action_to_task_event(task_id: str, action: Any) -> TaskEvent | None: + """Convert an existing ChannelAction-like object into a neutral TaskEvent.""" + + action_kind = _string_value(getattr(action, "kind", "")) + if action_kind == "prompt": + question = _string_value(getattr(action, "question", "")) + return TaskEvent( + task_id=task_id, + kind=TaskEventKind.WAITING, + message=question, + payload={ + "question": question, + "options": _options_tuple(getattr(action, "options", ())), + "prompt_kind": _string_value(getattr(action, "prompt_kind", "")), + "tool_name": _string_value(getattr(action, "tool", "")), + "method": _string_value(getattr(action, "method", "")), + }, + ) + if action_kind == "done": + message = _string_value(getattr(action, "message", "")) + return TaskEvent(task_id=task_id, kind=TaskEventKind.DONE, message=message, payload={"message": message}) + if action_kind == "error": + message = _string_value(getattr(action, "message", "")) + return TaskEvent(task_id=task_id, kind=TaskEventKind.ERROR, message=message, payload={"message": message}) + if action_kind == "running": + return TaskEvent(task_id=task_id, kind=TaskEventKind.RUNNING, payload={}) + if action_kind == "cancelled": + message = _string_value(getattr(action, "message", ""), "Task cancelled") or "Task cancelled" + return TaskEvent(task_id=task_id, kind=TaskEventKind.CANCELLED, message=message, payload={"message": message}) + return None + + +def task_status_payload_to_task_event(payload: Mapping[str, Any]) -> TaskEvent | None: + """Convert Gateway/MCP task status payloads into neutral TaskEvent values.""" + + task_id = _string_value(payload.get("task_id")) + if not task_id: + return None + + status = _string_value(payload.get("status")) + copied = _copy_payload(payload) + if "options" in copied: + copied["options"] = _options_tuple(copied.get("options")) + + if status == "waiting": + prompt_kind = _string_value(payload.get("kind") or payload.get("prompt_kind") or "waiting") + copied["prompt_kind"] = prompt_kind + return TaskEvent( + task_id=task_id, + kind=TaskEventKind.WAITING, + message=_string_value(payload.get("question")), + payload=copied, + ) + if status == "running": + return TaskEvent(task_id=task_id, kind=TaskEventKind.RUNNING, payload=copied) + if status == "done": + return TaskEvent( + task_id=task_id, + kind=TaskEventKind.DONE, + message=_string_value(payload.get("result")), + payload=copied, + ) + if status == "error": + return TaskEvent( + task_id=task_id, + kind=TaskEventKind.ERROR, + message=_string_value(payload.get("message")), + payload=copied, + ) + if status == "cancelled": + return TaskEvent( + task_id=task_id, + kind=TaskEventKind.CANCELLED, + message=_string_value(payload.get("message"), "Task cancelled") or "Task cancelled", + payload=copied, + ) + return None diff --git a/tests/test_orchestrator_event_mapping.py b/tests/test_orchestrator_event_mapping.py new file mode 100644 index 0000000..270e5ab --- /dev/null +++ b/tests/test_orchestrator_event_mapping.py @@ -0,0 +1,159 @@ +from __future__ import annotations + +from hermit_agent.channels_core.event_adapters import ChannelAction +from hermit_agent.orchestrators import ( + TaskEvent, + TaskEventKind, + channel_action_to_task_event, + sse_event_to_task_event, + task_status_payload_to_task_event, +) + + +def test_sse_running_and_progress_events_map_to_neutral_task_events_without_mutation(): + running_payload = {"type": "reply_ack", "extra": {"nested": True}} + progress_payload = {"type": "progress", "message": 123, "step": "lint"} + + running = sse_event_to_task_event("task-1", running_payload) + progress = sse_event_to_task_event("task-1", progress_payload) + + assert running == TaskEvent( + task_id="task-1", + kind=TaskEventKind.RUNNING, + payload={"type": "reply_ack", "extra": {"nested": True}}, + ) + assert progress == TaskEvent( + task_id="task-1", + kind=TaskEventKind.PROGRESS, + message="123", + payload={"type": "progress", "message": 123, "step": "lint"}, + ) + assert running_payload == {"type": "reply_ack", "extra": {"nested": True}} + + +def test_sse_waiting_event_preserves_prompt_metadata(): + event = { + "type": "permission_ask", + "question": "Run tests?", + "options": ["Yes", "No"], + "tool_name": "bash", + "method": "item/commandExecution/requestApproval", + "request_id": "req-1", + } + + mapped = sse_event_to_task_event("task-2", event) + + assert mapped == TaskEvent( + task_id="task-2", + kind=TaskEventKind.WAITING, + message="Run tests?", + payload={ + "type": "permission_ask", + "question": "Run tests?", + "options": ("Yes", "No"), + "tool_name": "bash", + "method": "item/commandExecution/requestApproval", + "request_id": "req-1", + "prompt_kind": "permission_ask", + }, + ) + + +def test_sse_done_error_cancelled_and_unknown_events_are_defensive(): + assert sse_event_to_task_event("task-3", {"type": "done", "result": "complete", "tokens": 7}) == TaskEvent( + task_id="task-3", + kind=TaskEventKind.DONE, + message="complete", + payload={"type": "done", "result": "complete", "tokens": 7}, + ) + assert sse_event_to_task_event("task-3", {"type": "error", "message": "boom"}) == TaskEvent( + task_id="task-3", + kind=TaskEventKind.ERROR, + message="boom", + payload={"type": "error", "message": "boom"}, + ) + assert sse_event_to_task_event("task-3", {"type": "cancelled"}) == TaskEvent( + task_id="task-3", + kind=TaskEventKind.CANCELLED, + message="Task cancelled", + payload={"type": "cancelled"}, + ) + assert sse_event_to_task_event("task-3", {"type": "streaming", "token": "partial"}) is None + + +def test_channel_action_mapping_covers_existing_action_surface(): + assert channel_action_to_task_event( + "task-4", + ChannelAction( + kind="prompt", + question="Need input", + options=("A", "B"), + prompt_kind="waiting", + tool="ask", + method="item/tool/requestUserInput", + ), + ) == TaskEvent( + task_id="task-4", + kind=TaskEventKind.WAITING, + message="Need input", + payload={ + "question": "Need input", + "options": ("A", "B"), + "prompt_kind": "waiting", + "tool_name": "ask", + "method": "item/tool/requestUserInput", + }, + ) + assert channel_action_to_task_event("task-4", ChannelAction(kind="done", message="ok")) == TaskEvent( + task_id="task-4", + kind=TaskEventKind.DONE, + message="ok", + payload={"message": "ok"}, + ) + assert channel_action_to_task_event("task-4", ChannelAction(kind="error", message="boom")) == TaskEvent( + task_id="task-4", + kind=TaskEventKind.ERROR, + message="boom", + payload={"message": "boom"}, + ) + assert channel_action_to_task_event("task-4", ChannelAction(kind="running")) == TaskEvent( + task_id="task-4", + kind=TaskEventKind.RUNNING, + payload={}, + ) + + +def test_task_status_payload_mapping_matches_gateway_status_responses(): + waiting = task_status_payload_to_task_event( + { + "task_id": "task-5", + "status": "waiting", + "kind": "permission_ask", + "question": "Approve?", + "options": ["Yes", "No"], + "tool_name": "bash", + } + ) + done = task_status_payload_to_task_event({"task_id": "task-5", "status": "done", "result": "finished"}) + + assert waiting == TaskEvent( + task_id="task-5", + kind=TaskEventKind.WAITING, + message="Approve?", + payload={ + "task_id": "task-5", + "status": "waiting", + "kind": "permission_ask", + "question": "Approve?", + "options": ("Yes", "No"), + "tool_name": "bash", + "prompt_kind": "permission_ask", + }, + ) + assert done == TaskEvent( + task_id="task-5", + kind=TaskEventKind.DONE, + message="finished", + payload={"task_id": "task-5", "status": "done", "result": "finished"}, + ) + assert task_status_payload_to_task_event({"status": "running"}) is None