nisse/nisse-dev/src/nisse_dev/views.py

"""Per-role views of a work item's transcript (architecture L6; Development
Plan M1.2).

The log records every role's events in one work item. A model must not read
another role's turns as its own: in the second live run the planner, shown
the executor's tool calls as assistant turns, took that work for its own
(#78). So what a role is shown is projected here, before rendering:

- its own events, unchanged;
- what the operator said, as a user message labelled with where it came from;
- the orchestrator's facts and gates, as labelled statements;
- another model role's work only where it needs it, and only as a labelled
  report or narration, never as a turn.

The projection keeps each event's id, so cursors (``after``) and compaction
ranges still line up with the log. Everything not the viewer's own becomes a
``UserMessage``: from where the viewer sits, it is all input.
"""

from dataclasses import replace

from nisse_dev.canonical import (
    AssistantText,
    DelegationCompleted,
    Fact,
    GateOpened,
    PlanSubmitted,
    StoredEvent,
    ToolCall,
    ToolResult,
    UserMessage,
)

# How much of another role's tool result a narration line keeps.
NARRATION_RESULT_CHARS = 300

# The roles whose work the viewer is shown, and how: "report" keeps only the
# handback, "narrate" keeps a line per step. Anything not listed is not shown.
_SHOWN: dict[str, dict[str, str]] = {
    "planner": {},
    "driver": {"helper": "report"},
    "executor": {"helper": "report"},
    "helper": {"executor": "narrate", "driver": "narrate"},
}


def operator_label(source: str) -> str:
    """Where an operator message came from, for its label."""
    origin, _, idle = source.partition(" (sent while idle")
    if origin.startswith("transport:"):
        channel = "Matrix"
    elif origin.startswith("gitea:pr:"):
        number = origin.split(":")[2]
        channel = f"comment on PR #{number}"
    elif origin.startswith("gitea:issue-body:"):
        channel = f"issue #{origin.split(':')[2]}"
    elif origin.startswith("gitea:issue:"):
        number = origin.split(":")[2]
        channel = f"comment on plan issue #{number}"
    elif origin.startswith("gitea:"):
        channel = "Gitea comment"
    else:
        channel = origin or "unknown"
    if idle:
        channel += ", sent while idle"
    return f"operator, {channel}"


def _said(stored: StoredEvent, label: str, text: str) -> StoredEvent:
    return replace(
        stored,
        event=UserMessage(text=f"[{label}] {text}", source=f"view:{stored.actor}"),
    )


def _narration(call: ToolCall, result: ToolResult | None) -> str:
    target = call.arguments.get("path") or call.arguments.get("command")
    step = f"{call.tool}({str(target)[:80]})" if target else call.tool
    if result is None:
        return f"{step} → no result recorded"
    outcome = "ok" if result.success else "failed"
    text = result.result.strip()
    if len(text) > NARRATION_RESULT_CHARS:
        text = text[:NARRATION_RESULT_CHARS] + " …"
    return f"{step} → {outcome}: {text}" if text else f"{step} → {outcome}"


def project(events: list[StoredEvent], viewer: str) -> list[StoredEvent]:
    """The events as ``viewer`` should be shown them, in log order."""
    shown = _SHOWN.get(viewer)
    if shown is None:
        # A role with no view defined (the expert is stateless, the
        # classifier sees only an action) reads the log as it is.
        return list(events)
    results = {
        e.event.call_id: e.event for e in events if isinstance(e.event, ToolResult)
    }
    out: list[StoredEvent] = []
    for stored in events:
        event, actor = stored.event, stored.actor
        if isinstance(event, PlanSubmitted):
            # A record for the executor's scope and the dump, not a turn:
            # the planner's own submit_plan call already carries the plan.
            if viewer == "helper":
                out.append(_said(stored, "approved plan", event.plan))
            continue
        if actor == viewer:
            out.append(stored)
        elif actor == "operator" and isinstance(event, UserMessage):
            out.append(_said(stored, operator_label(event.source), event.text))
        elif isinstance(event, Fact):
            out.append(_said(stored, "orchestrator", event.text))
        elif isinstance(event, GateOpened):
            out.append(
                _said(stored, "orchestrator", f"gate {event.gate}: {event.reason}")
            )
        elif shown.get(actor) == "report" and isinstance(event, DelegationCompleted):
            out.append(
                _said(
                    stored,
                    f"{actor} report, {event.exit_status}",
                    event.handback,
                )
            )
        elif shown.get(actor) == "narrate":
            if isinstance(event, ToolCall):
                line = _narration(event, results.get(event.call_id))
                out.append(_said(stored, actor, line))
            elif isinstance(event, AssistantText) and event.text.strip():
                out.append(_said(stored, actor, event.text.strip()))
    return out