app/transcript_to_tasks.py

"""Pure mapper: OpenClaw Gateway `chat` event envelope -> TaskMutation list.

Phase 2.1 of `specs/realtime-progress-via-gateway-ws.md`. No I/O, no asyncio,
no logging. Caller (the realtime feed orchestrator) is responsible for
applying the mutations to a `ProgressTracker` and for managing subscription
lifecycle. This module is the deterministic translation layer alone.

Wire-shape source: `specs/m1-t13-ws-tap-evidence/raw-envelopes.jsonl` (live
capture, 2026-05-13) + the gateway sanitizer at
`~/.nvm/versions/node/v24.15.0/lib/node_modules/openclaw/dist/chat-display-projection-BSsHGnx6.js`.
"""
from __future__ import annotations

from dataclasses import dataclass, field
from typing import Any, Literal


Status = Literal["loading", "complete"]


@dataclass(frozen=True)
class AddTask:
    """Append a new task to the open workflow.

    `tool_call_id`, when present, is remembered by the orchestrator so a
    later `CompleteTask(tool_call_id=...)` can correlate to the right task
    index without the translator owning the index space.
    """

    title: str
    icon: str | None
    tool_call_id: str | None = None


@dataclass(frozen=True)
class CompleteTask:
    """Flip the task that was created for `tool_call_id` to status=complete."""

    tool_call_id: str
    title: str | None = None


@dataclass(frozen=True)
class MarkRunTerminal:
    """Run ended (state=error | aborted). Close all loading tasks and append
    a final task carrying the terminal reason (suffix). The orchestrator
    must also tear down the per-turn feed after applying this mutation.
    """

    suffix: AddTask | None = None


# Tools that exist on every session but are internal plumbing — too noisy
# to render as user-facing progress steps. Source: bridge tool catalog +
# `docs/tools/subagents.md`. Allowlist is conservative.
_INTERNAL_TOOLS_SKIP: frozenset[str] = frozenset({
    "sessions_send",
    "sessions_history",
    "sessions_yield",
    "session_status",
    "subagents",  # the polling helper, not sessions_spawn
})


@dataclass
class PerTurnState:
    """Mutable state carried across translator calls within one turn.

    - `open_tool_uses`: maps tool_use id -> tool name. Populated when an
      `AddTask` is emitted with a non-None tool_call_id. Cleared when the
      matching `CompleteTask` is emitted.
    - `seen_session_keys`: tracks which subscribed session keys we've
      already seen events for — used by the orchestrator to decide when
      to open a transitive child subscription. The translator only
      records; the orchestrator owns the subscribe calls.
    """

    open_tool_uses: dict[str, str] = field(default_factory=dict)
    seen_session_keys: set[str] = field(default_factory=set)
    parent_session_key: str | None = None


def _basename(path: Any) -> str:
    if not isinstance(path, str) or not path:
        return ""
    return path.rstrip("/").split("/")[-1] or path


def _first_words(text: str, n: int) -> str:
    parts = text.strip().split()
    return " ".join(parts[:n])


def _hostname(url: Any) -> str:
    if not isinstance(url, str):
        return ""
    rest = url.split("//", 1)[-1]
    return rest.split("/", 1)[0]


def _title_and_icon(name: str, inp: Any) -> tuple[str, str]:
    """Map a `tool_use` name + input to a user-facing (title, icon).

    Icon allowlist: only `bolt` and `agent` are validated by ChatKit
    `CustomTask` construction; see M1b decision log in
    `specs/realtime-progress-events.md`.
    """
    inp = inp if isinstance(inp, dict) else {}
    if name == "Read":
        return f"Reading {_basename(inp.get('file_path'))}".strip(), "bolt"
    if name == "Write":
        return f"Writing {_basename(inp.get('file_path'))}".strip(), "bolt"
    if name == "Edit":
        return f"Editing {_basename(inp.get('file_path'))}".strip(), "bolt"
    if name == "Bash":
        cmd = inp.get("command") or ""
        head = cmd
        if isinstance(cmd, str) and cmd.startswith("cd ") and "&&" in cmd:
            head = cmd.split("&&", 1)[1].strip()
        return f"Running: {_first_words(head, 8)}".strip(), "bolt"
    if name == "Grep":
        pat = str(inp.get("pattern") or "")[:40]
        return f'Searching: "{pat}"', "bolt"
    if name == "Glob":
        pat = str(inp.get("pattern") or "")
        return f'Finding files: "{pat}"', "bolt"
    if name in ("WebSearch", "web_search"):
        q = str(inp.get("query") or "")[:40]
        return f'Searching the web: "{q}"', "bolt"
    if name == "WebFetch":
        return f"Fetching {_hostname(inp.get('url'))}".strip(), "bolt"
    if name == "sessions_spawn":
        agent = inp.get("agentId") or inp.get("agent") or "agent"
        task = str(inp.get("task") or "")[:60]
        title = f"Delegating to {agent}: {task}" if task else f"Delegating to {agent}"
        return title, "agent"
    return f"Calling {name}", "bolt"


def _normalize_block_type(t: Any) -> str:
    """Lowercase + strip + collapse separators. Recognises both the WS `chat`
    broadcast shape (`tool_use`, `tool_call`) AND the `sessions_history`
    transcript shape (`toolcall`, no separator) — see spec §6.1b.
    """
    if not isinstance(t, str):
        return ""
    return t.strip().lower().replace("-", "_")


def _extract_tool_use_id(block: dict[str, Any]) -> str | None:
    """tool_use blocks carry id in `id`; tool_result blocks carry the
    matching id in `tool_use_id` / `toolUseId` / `tool_call_id` / `id`."""
    for key in ("id", "tool_use_id", "toolUseId", "tool_call_id"):
        v = block.get(key)
        if isinstance(v, str) and v:
            return v
    return None


def _extract_tool_result_id(block: dict[str, Any]) -> str | None:
    for key in ("tool_use_id", "toolUseId", "tool_call_id", "id"):
        v = block.get(key)
        if isinstance(v, str) and v:
            return v
    return None


def translate(
    envelope: dict[str, Any],
    state: PerTurnState,
) -> list[AddTask | CompleteTask | MarkRunTerminal]:
    """Map one inbound Gateway WS frame to zero-or-more TaskMutations.

    Inputs we accept and handle:
    - frame.type == "event" AND frame.event == "chat"
      - payload.state ∈ {"delta"}: ignored (no mutations)
      - payload.state == "final": dispatch by role/content
      - payload.state ∈ {"error", "aborted"}: emit MarkRunTerminal

    Everything else (other event names, non-event frames, malformed
    payloads) returns []. The translator NEVER raises.
    """
    try:
        if not isinstance(envelope, dict):
            return []
        if envelope.get("type") != "event":
            return []
        if envelope.get("event") != "chat":
            return []
        payload = envelope.get("payload") or {}
        if not isinstance(payload, dict):
            return []

        # Track session-key observation for orchestrator's child-subscribe.
        sk = payload.get("sessionKey")
        if isinstance(sk, str):
            state.seen_session_keys.add(sk)

        state_str = payload.get("state")

        if state_str == "error":
            err = payload.get("errorMessage")
            err_str = err if isinstance(err, str) and err else "unknown"
            return [
                MarkRunTerminal(
                    suffix=AddTask(
                        title=f"Run errored: {err_str[:80]}",
                        icon="bolt",
                    )
                )
            ]
        if state_str == "aborted":
            return [MarkRunTerminal(suffix=AddTask(title="Aborted", icon="bolt"))]
        if state_str != "final":
            # delta / unknown / None — ignore for now (Phase 2.1 scope).
            return []

        message = payload.get("message") or {}
        if not isinstance(message, dict):
            return []
        role = message.get("role")
        role_str = role.lower() if isinstance(role, str) else ""

        mutations: list[AddTask | CompleteTask | MarkRunTerminal] = []

        # Content array dispatch (Anthropic-shape canonical).
        content = message.get("content")
        if isinstance(content, list):
            for block in content:
                if not isinstance(block, dict):
                    continue
                btype = _normalize_block_type(block.get("type"))

                if btype in ("tool_use", "tool_call", "toolcall"):
                    name = block.get("name") or block.get("tool_name") or "tool"
                    if not isinstance(name, str):
                        continue
                    if name in _INTERNAL_TOOLS_SKIP:
                        continue
                    tool_id = _extract_tool_use_id(block)
                    if tool_id and tool_id in state.open_tool_uses:
                        # Idempotent: a re-broadcast of the same final
                        # (e.g. after a reconnect replay) must not double-add.
                        continue
                    # WS shape uses `input`; transcript shape uses `arguments`
                    # (spec §6.1b). Prefer input when present.
                    inp = block.get("input")
                    if inp is None:
                        inp = block.get("arguments")
                    title, icon = _title_and_icon(name, inp)
                    mutations.append(
                        AddTask(title=title, icon=icon, tool_call_id=tool_id)
                    )
                    if tool_id:
                        state.open_tool_uses[tool_id] = name

                elif btype == "tool_result":
                    tool_id = _extract_tool_result_id(block)
                    if tool_id and tool_id in state.open_tool_uses:
                        mutations.append(CompleteTask(tool_call_id=tool_id))
                        state.open_tool_uses.pop(tool_id, None)

        # Message-level tool-result form (role=tool|toolresult|function,
        # toolCallId at message root rather than in a content block).
        if role_str in ("tool", "toolresult", "function"):
            tcid = None
            for key in ("toolCallId", "tool_call_id", "toolUseId", "tool_use_id"):
                v = message.get(key)
                if isinstance(v, str) and v:
                    tcid = v
                    break
            if tcid and tcid in state.open_tool_uses:
                # Avoid duplicate complete if the content array already emitted one.
                already_completed = any(
                    isinstance(m, CompleteTask) and m.tool_call_id == tcid
                    for m in mutations
                )
                if not already_completed:
                    mutations.append(CompleteTask(tool_call_id=tcid))
                    state.open_tool_uses.pop(tcid, None)

        return mutations
    except Exception:
        # Translator contract: never raise. Caller can't drop the
        # subscription on a single bad envelope.
        return []