Map task lifecycle events to adapter DTOs

.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.

.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
 

.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`

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.

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",
 ]

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

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