docs: align runner guide with factory pattern

docs/adding-a-runner.md

@@ -138,6 +138,7 @@ Copy the Claude pattern: create a small dataclass to hold streaming state.
 
 Common things to track:
 
+- `factory`: `EventFactory` instance for creating Takopi events and tracking resume
 - `pending_actions`: map tool_use_id → `Action` so tool results can complete them
 - `last_assistant_text`: fallback for final answer if the engine omits it
 - `note_seq`: counter used by `JsonlSubprocessRunner.note_event(...)`
@@ -145,8 +146,11 @@ Common things to track:
 ```py
 from dataclasses import dataclass, field
 
+from ..events import EventFactory
+
 @dataclass
 class PiStreamState:
+    factory: EventFactory = field(default_factory=lambda: EventFactory(ENGINE))
     pending_actions: dict[str, Action] = field(default_factory=dict)
     last_assistant_text: str | None = None
     note_seq: int = 0
@@ -228,39 +232,30 @@ Use this mapping (mirrors Claude’s approach):
 Claude keeps translation logic in a standalone function (`translate_claude_event(...)`).
 This makes it easy to unit test without spawning a subprocess.
 
-Do the same for Pi:
+Do the same for Pi. Use pattern matching against msgspec shapes, and rely on the
+`EventFactory` (as in Codex/Claude) to standardize event creation:
 
 ```py
 def translate_pi_event(
-    event: dict[str, Any],
+    event: pi_schema.PiEvent,
     *,
     title: str,
     state: PiStreamState,
+    factory: EventFactory,
 ) -> list[TakopiEvent]:
-    etype = event.get("type")
+    match event:
-
+        case pi_schema.SessionStart(session_id=session_id, model=model):
-    if etype == "session.start":
-        session_id = event.get("session_id")
             if not session_id:
                 return []
-        model = event.get("model")
             event_title = str(model) if model else title
-        return [
+            token = ResumeToken(engine=ENGINE, value=session_id)
-            StartedEvent(
+            return [factory.started(token, title=event_title)]
-                engine=ENGINE,
-                resume=ResumeToken(engine=ENGINE, value=str(session_id)),
-                title=event_title,
-            )
-        ]
 
-    if etype == "tool.use":
+        case pi_schema.ToolUse(id=tool_id, name=name, input=tool_input):
-        tool_id = event.get("id")
+            if not tool_id:
-        if not isinstance(tool_id, str) or not tool_id:
                 return []
-        name = str(event.get("name") or "tool")
+            tool_input = tool_input or {}
-        tool_input = event.get("input")
+            name = str(name or "tool")
-        if not isinstance(tool_input, dict):
-            tool_input = {}
 
             # Keep titles short and friendly.
             # (Claude uses takopi.utils.paths.relativize_command / relativize_path)
@@ -277,11 +272,19 @@ def translate_pi_event(
                 detail={"name": name, "input": tool_input},
             )
             state.pending_actions[action.id] = action
-        return [ActionEvent(engine=ENGINE, action=action, phase="started")]
+            return [
+                factory.action_started(
+                    action_id=action.id,
+                    kind=action.kind,
+                    title=action.title,
+                    detail=action.detail,
+                )
+            ]
 
-    if etype == "tool.result":
+        case pi_schema.ToolResult(
-        tool_use_id = event.get("tool_use_id")
+            tool_use_id=tool_use_id, content=content, is_error=is_error
-        if not isinstance(tool_use_id, str) or not tool_use_id:
+        ):
+            if not tool_use_id:
                 return []
             action = state.pending_actions.pop(tool_use_id, None)
             if action is None:
@@ -292,61 +295,54 @@ def translate_pi_event(
                     detail={},
                 )
 
-        is_error = event.get("is_error") is True
+            result_text = (
-        content = event.get("content")
+                ""
-        result_text = "" if content is None else (content if isinstance(content, str) else str(content))
+                if content is None
-
+                else (content if isinstance(content, str) else str(content))
+            )
             detail = dict(action.detail)
-        detail.update({"result_preview": result_text, "is_error": is_error})
+            detail.update(
+                {"result_preview": result_text, "is_error": bool(is_error)}
+            )
 
             return [
-            ActionEvent(
+                factory.action_completed(
-                engine=ENGINE,
+                    action_id=action.id,
-                action=Action(
-                    id=action.id,
                     kind=action.kind,
                     title=action.title,
+                    ok=not bool(is_error),
                     detail=detail,
-                ),
-                phase="completed",
-                ok=not is_error,
                 )
             ]
 
-    if etype == "final":
+        case pi_schema.Final(session_id=session_id, ok=ok, answer=answer, error=error):
-        ok = event.get("ok") is True
+            answer = answer or ""
-        answer = event.get("answer")
-        if not isinstance(answer, str):
-            answer = ""
             if ok and not answer and state.last_assistant_text:
                 answer = state.last_assistant_text
 
-        session_id = event.get("session_id")
             resume = (
-            ResumeToken(engine=ENGINE, value=str(session_id)) if session_id else None
+                ResumeToken(engine=ENGINE, value=session_id) if session_id else None
             )
 
-        error = None
+            if ok:
-        if not ok:
+                return [factory.completed_ok(answer=answer, resume=resume)]
-            err = event.get("error")
-            error = str(err) if err else "pi run failed"
 
+            error_text = str(error) if error else "pi run failed"
             return [
-            CompletedEvent(
+                factory.completed_error(
-                engine=ENGINE,
+                    error=error_text,
-                ok=ok,
                     answer=answer,
                     resume=resume,
-                error=error,
                 )
             ]
 
+        case _:
             return []
 ```
 
 This is intentionally close to Claude’s structure:
 
-- Parse `type`
+- Match on the msgspec event type
 - Handle “init/session start” first
 - Emit action-start and action-complete events
 - Emit a final `CompletedEvent`
@@ -377,17 +373,14 @@ import logging
 import re
 from dataclasses import dataclass
 from pathlib import Path
-from typing import Any, cast
+from typing import Any
 
 from ..backends import EngineBackend, EngineConfig
 from ..model import (
-    CompletedEvent,
     EngineId,
     ResumeToken,
-    StartedEvent,
     TakopiEvent,
 )
-import msgspec
 
 from ..runner import JsonlSubprocessRunner, ResumeTokenMixin, Runner
 from ..schemas import pi as pi_schema
@@ -458,21 +451,25 @@ class PiRunner(ResumeTokenMixin, JsonlSubprocessRunner):
         raw: bytes,
         line: bytes,
         state: PiStreamState,
-    ) -> dict[str, Any] | None:
+    ) -> pi_schema.PiEvent | None:
         _ = raw, state
-        event = pi_schema.decode_event(line)
+        return pi_schema.decode_event(line)
-        return cast(dict[str, Any], msgspec.to_builtins(event))
 
     def translate(
         self,
-        data: dict[str, Any],
+        data: pi_schema.PiEvent,
         *,
         state: PiStreamState,
         resume: ResumeToken | None,
         found_session: ResumeToken | None,
     ) -> list[TakopiEvent]:
         _ = resume, found_session
-        return translate_pi_event(data, title=self.session_title, state=state)
+        return translate_pi_event(
+            data,
+            title=self.session_title,
+            state=state,
+            factory=state.factory,
+        )
 ```
 
 Notes: