projection.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/core/projection.py
 
"""
Incremental projection of the M2 event stream into readable state — plan 190 T2.
 
Purpose
-------
The UI synthesis (§2/§4/§5) calls the projection/reducer "the keystone
deliverable": the thing that turns the append-only ``TurnEvent`` stream
into the state a reader wants, *before* any UI exists. One fold serves
three consumers that would otherwise each grow their own ad-hoc
accounting — a live dashboard, verification-backpressure observability,
and cache governance.
 
The fold is deliberately plain: :meth:`EventProjection.feed` takes one
event and updates internal counters; :meth:`EventProjection.snapshot`
returns an immutable value. Because it is pure and incremental, the same
events fed live and replayed from the durable NDJSON log produce the
identical snapshot — the property plan 190 T3 witnesses, and the reason
``/trace``, ``/packet`` and replay can share one implementation.
 
Responsibilities
----------------
- Fold cost, token and cache/ctx gauge inputs from
  ``provider_budget_consumed``.
- Track the live tool lifecycle (started → progress → completed) keyed by
  ``tool_call_id``, plus proposals and permission decisions.
- Track the sub-agent board, session/provider markers, prompt-render
  layer hashes, and error / compaction markers.
- Return all of it as one frozen :class:`ProjectionSnapshot`.
 
What this module is NOT
-----------------------
No I/O, no presenter or ``oc_ui`` import, and **no bus attachment** — the
module never subscribes to anything. A caller that wants live state
attaches ``projection.feed`` to its own bus subscription; a caller
replaying a log calls ``feed`` in a loop. Keeping attachment outside is
what lets the same object serve both without a code path difference.
 
Plan 188's ``core/presenter_events.py`` is untouched and unrelated: it
translates events into presenter calls. Unifying the two is a named
follow-up, not this module's job.
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 — lifecycle events (none emitted here)
    L3 — semantic details (none emitted here)
    L4 — module-load coverage marker
 
Contracts
---------
- ``feed`` never raises on an unrecognised ``kind``: the v0.1 enum is
  closed, but a replayed log may predate or postdate this build, and a
  reducer that dies on an unknown line is useless for forensics.
- ``snapshot`` returns a value: taking it does not freeze the reducer, and
  feeding more events does not mutate a snapshot already returned.
- Every number reported is one the stream actually carries. Where the
  current 20 kinds cannot express something a reader wants, this module
  reports nothing rather than an inferred figure. The three such gaps are
  documented on :class:`ProjectionSnapshot` and asserted by the tests.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- option_c_agent_contracts (lazy; TurnEvent type hint for the feed method)
"""
from __future__ import annotations
 
from dataclasses import dataclass, field
from typing import TYPE_CHECKING, Any, Optional
 
from oc_diagnostics import _dbg
 
if TYPE_CHECKING:  # pragma: no cover - typing only
    from option_c_agent_contracts import TurnEvent
 
_dbg("OC_CHAT.MAIN", "projection", "module", 4)
 
 
#: Tool names that drive the plan / queue workflow. Used ONLY to offer a
#: filtered view of real tool activity (see ``plan_tool_calls``) — never to
#: synthesise plan state, which the v0.1 enum cannot express.
_PLAN_TOOL_NAMES: frozenset = frozenset({
    "create_plan", "mark_step", "get_status", "stop_plan",
})
 
 
@dataclass(frozen=True, slots=True)
class CostView:
    """The cost + gauge inputs derivable from ``provider_budget_consumed``."""
 
    total_usd: float = 0.0
    # OCT-LINT: disable=no_hardcoded_secrets reason="Dataclass field docstring prose trips the Shannon entropy heuristic (L98, 'Sum of the ``cost_usd`` of the samples that carried one.', 4.50 bits/char - re-measured for plan 203, which rewrote this docstring; the prior waiver still quoted the pre-203 text 'Sum of the published samples...' at 4.60 bits/char); no secret material - this module is a pure in-memory fold holding no credentials, tokens, URLs or connection strings. File-scoped because that is the only waiver scope oct implements (quality_gate.py discards the line number)."
    """Sum of the ``cost_usd`` of the samples that carried one.
 
    EXACT when :attr:`unpriced_samples` is 0; a LOWER BOUND when it is not —
    and, since plan 203, the reader can tell which. Plan 190 had to document
    this as an unconditional lower bound: ``cost_usd`` was a non-optional
    ``float``, so an unpriceable sample could not be published at all and its
    absence was indistinguishable from no spend. The contract now types it
    ``Optional[float]``, the unpriceable sample publishes with ``None``, and
    the count of those lands in :attr:`unpriced_samples` instead of vanishing.
    """
    input_tokens: int = 0
    output_tokens: int = 0
    samples: int = 0
    unpriced_samples: int = 0
    """How many of :attr:`samples` published a ``cost_usd`` of ``None``.
 
    The unknown-vs-zero distinction: a sample that reported ``0.0`` is
    priced-and-free and is NOT counted here, while one that reported ``None``
    could not be priced. Non-zero means :attr:`total_usd` understates real
    spend by an amount this fold cannot know — which is a fact about the
    stream, not a defect in it.
    """
    max_input_tokens: int = 0
    """The ctx gauge input: the LARGEST single prompt seen, not the sum.
 
    ``ctx:%`` asks how close one request came to the context window, so a
    running total would be meaningless for it.
    """
 
 
@dataclass(frozen=True, slots=True)
class ToolCallView:
    """One tool call's live lifecycle, keyed by ``tool_call_id``."""
 
    tool_call_id: str
    tool_name: str
    status: str
    """``"started"`` → ``"running"`` (a progress event arrived) → ``"completed"``."""
    outcome: Optional[str] = None
    progress: tuple = ()
    result_artifact_id: Optional[str] = None
 
 
@dataclass(frozen=True, slots=True)
class ProposedCallView:
    """A proposal, deliberately WITHOUT a ``tool_call_id``.
 
    ``tool_call_proposed`` carries a tool name and the requested args;
    the call id does not exist until the provider assigns one, so neither
    its payload nor its envelope can carry it. The reducer records the
    proposal and declines to guess which started call it became — ADR-5's
    ordering makes a positional guess *usually* right, which is exactly
    what makes it dangerous to encode as fact.
    """
 
    tool_name: str
    turn_id: Optional[str] = None
 
 
@dataclass(frozen=True, slots=True)
class PermissionDecisionView:
    """A permission decision, deliberately WITHOUT a tool linkage.
 
    LIMITATION (plan 190, documented not faked): ``permission_decision``
    sets ``permission_decision_id`` on payload and envelope, but no
    ``tool_call_id`` on either — so a decision cannot be attributed to the
    call it gated. Ordering within a turn is suggestive, not evidence.
    """
 
    permission_decision_id: str
    decision: str
    decided_by: str
    turn_id: Optional[str] = None
 
 
@dataclass(frozen=True, slots=True)
class SubagentView:
    """One row of the sub-agent board, keyed by ``subagent_session_id``."""
 
    subagent_session_id: str
    status: str
    outcome: Optional[str] = None
 
 
@dataclass(frozen=True, slots=True)
class CompactionView:
    """Compaction markers.
 
    STEP-0 found no compaction publisher in the product, so a live stream
    will not carry these today. The fold handles them anyway: a replayed
    log from another build must not break the reducer.
    """
 
    started: int = 0
    completed: int = 0
    active: bool = False
    last_reason: Optional[str] = None
 
 
@dataclass(frozen=True, slots=True)
class ProjectionSnapshot:
    """The folded state at one point in the stream — an immutable value.
 
    Three things a reader may want are NOT here, because the current 20
    kinds cannot express them and an inferred number is worse than none:
 
    1. **Cache hit ratio.** ``provider_budget_consumed`` carries
       ``input_tokens`` / ``output_tokens`` but no hit/miss split, so the
       ``cache:%`` gauge cannot be derived from the stream (the ctx gauge
       input can — see :attr:`CostView.max_input_tokens`).
    2. **Plan / queue state.** The v0.1 enum has no plan or queue kinds.
       :attr:`plan_tool_calls` is a filter over real tool events — it
       reports plan-tool *activity*, and is not a claim about plan state.
    3. **Proposal → call and permission → call linkage.** See
       :class:`ProposedCallView` and :class:`PermissionDecisionView`.
    """
 
    events_seen: int = 0
    session_id: Optional[str] = None
    session_status: Optional[str] = None
    session_outcome: Optional[str] = None
    turn_ids: tuple = ()
    cost: CostView = field(default_factory=CostView)
    prompt_versions: tuple = ()
    layer_hashes: dict = field(default_factory=dict)
    tool_calls: tuple = ()
    proposed_calls: tuple = ()
    permission_decisions: tuple = ()
    subagents: tuple = ()
    plan_tool_calls: tuple = ()
    errors: tuple = ()
    branch_summaries: tuple = ()
    hooks: tuple = ()
    compaction: CompactionView = field(default_factory=CompactionView)
    rate_limited: int = 0
    retried: int = 0
    user_inputs: int = 0
    assistant_messages: int = 0
 
 
class EventProjection:
    """Fold a ``TurnEvent`` stream into a :class:`ProjectionSnapshot`.
 
    Usage is two lines, live or replayed::
 
        proj = EventProjection()
        bus.subscribe("*", proj.feed)      # live  — attachment is the CALLER's
        for event in replayed: proj.feed(event)   # replayed
 
    Both produce the same snapshot for the same events, which is the
    property plan 190 T3 witnesses end-to-end.
    """
 
    def __init__(self) -> None:
        self._events = 0
        self._session_id: Optional[str] = None
        self._session_status: Optional[str] = None
        self._session_outcome: Optional[str] = None
        self._turn_ids: dict = {}
        self._usd = 0.0
        self._in_tokens = 0
        self._out_tokens = 0
        self._samples = 0
        self._unpriced = 0
        self._max_in = 0
        self._prompt_versions: list = []
        self._layer_hashes: dict = {}
        self._tools: dict = {}
        self._proposed: list = []
        self._permissions: list = []
        self._subagents: dict = {}
        self._plan_tools: list = []
        self._errors: list = []
        self._branch_summaries: list = []
        self._hooks: list = []
        self._compaction_started = 0
        self._compaction_completed = 0
        self._compaction_reason: Optional[str] = None
        self._rate_limited = 0
        self._retried = 0
        self._user_inputs = 0
        self._assistant_messages = 0
 
    # -- the fold ------------------------------------------------------------
    def feed(self, event: "TurnEvent") -> None:
        """Fold one event into the running state.
 
        Unrecognised kinds advance the counter and are otherwise ignored
        (see the module Contracts) — a reducer that dies on an unknown
        line cannot be used to investigate the log that contains it.
        """
        self._events += 1
        kind = getattr(event, "kind", "")
        payload = getattr(event, "payload", None)
 
        sid = getattr(event, "session_id", None)
        if sid and self._session_id is None:
            self._session_id = sid
        turn_id = getattr(event, "turn_id", None)
        if turn_id:
            self._turn_ids[turn_id] = True
 
        handler = _HANDLERS.get(kind)
        if handler is not None:
            handler(self, event, payload)
 
    # -- per-kind handlers ---------------------------------------------------
    def _on_session_started(self, _event: Any, _payload: Any) -> None:
        self._session_status = "started"
 
    def _on_session_ended(self, _event: Any, payload: Any) -> None:
        self._session_status = "ended"
        self._session_outcome = getattr(payload, "outcome", None)
 
    def _on_user_input(self, _event: Any, _payload: Any) -> None:
        self._user_inputs += 1
 
    def _on_assistant_message(self, _event: Any, _payload: Any) -> None:
        self._assistant_messages += 1
 
    def _on_prompt_rendered(self, _event: Any, payload: Any) -> None:
        version = getattr(payload, "prompt_version", None)
        if version:
            self._prompt_versions.append(version)
        hashes = getattr(payload, "layer_hashes", None)
        if hashes:
            # Last render wins: cache governance asks "what is the prefix
            # NOW", and a superseded hash is history, not state.
            self._layer_hashes = dict(hashes)
 
    def _on_budget(self, _event: Any, payload: Any) -> None:
        # Plan 203: None is UNKNOWN, not zero. The previous ``or 0.0`` folded
        # both into the same number, which is the coercion that made a running
        # total an unqualified lower bound. A missing attribute reads as
        # unknown too — a foreign build that never set it has told us nothing.
        cost = getattr(payload, "cost_usd", None)
        if cost is None:
            self._unpriced += 1
        else:
            self._usd += float(cost)
        inp = int(getattr(payload, "input_tokens", 0) or 0)
        self._in_tokens += inp
        self._out_tokens += int(getattr(payload, "output_tokens", 0) or 0)
        self._max_in = max(self._max_in, inp)
        self._samples += 1
 
    def _on_rate_limited(self, _event: Any, _payload: Any) -> None:
        self._rate_limited += 1
 
    def _on_retried(self, _event: Any, _payload: Any) -> None:
        self._retried += 1
 
    def _on_proposed(self, event: Any, payload: Any) -> None:
        self._proposed.append(ProposedCallView(
            tool_name=getattr(payload, "tool_name", ""),
            turn_id=getattr(event, "turn_id", None),
        ))
 
    def _on_permission(self, event: Any, payload: Any) -> None:
        self._permissions.append(PermissionDecisionView(
            permission_decision_id=getattr(payload, "permission_decision_id", ""),
            decision=getattr(payload, "decision", ""),
            decided_by=getattr(payload, "decided_by", ""),
            turn_id=getattr(event, "turn_id", None),
        ))
 
    def _on_tool_started(self, _event: Any, payload: Any) -> None:
        call_id = getattr(payload, "tool_call_id", "")
        name = getattr(payload, "tool_name", "")
        self._tools[call_id] = ToolCallView(
            tool_call_id=call_id, tool_name=name, status="started")
        if name in _PLAN_TOOL_NAMES:
            self._plan_tools.append(name)
 
    def _on_tool_progress(self, _event: Any, payload: Any) -> None:
        call_id = getattr(payload, "tool_call_id", "")
        node = self._tools.get(call_id)
        message = getattr(payload, "message", "")
        if node is None:
            # Progress without a start: a truncated or mid-stream log. Record
            # the call rather than dropping evidence that it ran.
            self._tools[call_id] = ToolCallView(
                tool_call_id=call_id, tool_name="", status="running",
                progress=(message,))
            return
        self._tools[call_id] = ToolCallView(
            tool_call_id=node.tool_call_id, tool_name=node.tool_name,
            status="running" if node.status != "completed" else node.status,
            outcome=node.outcome, progress=node.progress + (message,),
            result_artifact_id=node.result_artifact_id)
 
    def _on_tool_completed(self, _event: Any, payload: Any) -> None:
        call_id = getattr(payload, "tool_call_id", "")
        node = self._tools.get(call_id)
        outcome = getattr(payload, "outcome", None)
        artifact = getattr(payload, "result_artifact_id", None)
        if node is None:
            self._tools[call_id] = ToolCallView(
                tool_call_id=call_id, tool_name="", status="completed",
                outcome=outcome, result_artifact_id=artifact)
            return
        self._tools[call_id] = ToolCallView(
            tool_call_id=node.tool_call_id, tool_name=node.tool_name,
            status="completed", outcome=outcome, progress=node.progress,
            result_artifact_id=artifact)
 
    def _on_subagent_started(self, _event: Any, payload: Any) -> None:
        said = getattr(payload, "subagent_session_id", "")
        self._subagents[said] = SubagentView(
            subagent_session_id=said, status="started")
 
    def _on_subagent_completed(self, _event: Any, payload: Any) -> None:
        said = getattr(payload, "subagent_session_id", "")
        self._subagents[said] = SubagentView(
            subagent_session_id=said, status="completed",
            outcome=getattr(payload, "outcome", None))
 
    def _on_compaction_started(self, _event: Any, payload: Any) -> None:
        self._compaction_started += 1
        self._compaction_reason = getattr(payload, "reason", None)
 
    def _on_compaction_completed(self, _event: Any, _payload: Any) -> None:
        self._compaction_completed += 1
 
    def _on_branch_summary(self, _event: Any, payload: Any) -> None:
        self._branch_summaries.append(getattr(payload, "summary", ""))
 
    def _on_error(self, _event: Any, payload: Any) -> None:
        self._errors.append(getattr(payload, "error_summary", ""))
 
    def _on_hook(self, _event: Any, payload: Any) -> None:
        self._hooks.append((getattr(payload, "hook_name", ""),
                            int(getattr(payload, "exit_code", 0) or 0)))
 
    # -- the read ------------------------------------------------------------
    def snapshot(self) -> ProjectionSnapshot:
        """Return the folded state as an immutable value.
 
        Copies every mutable container, so the returned snapshot is
        unaffected by later :meth:`feed` calls.
        """
        return ProjectionSnapshot(
            events_seen=self._events,
            session_id=self._session_id,
            session_status=self._session_status,
            session_outcome=self._session_outcome,
            turn_ids=tuple(self._turn_ids),
            cost=CostView(
                total_usd=round(self._usd, 10),
                input_tokens=self._in_tokens,
                output_tokens=self._out_tokens,
                samples=self._samples,
                unpriced_samples=self._unpriced,
                max_input_tokens=self._max_in,
            ),
            prompt_versions=tuple(self._prompt_versions),
            layer_hashes=dict(self._layer_hashes),
            tool_calls=tuple(self._tools.values()),
            proposed_calls=tuple(self._proposed),
            permission_decisions=tuple(self._permissions),
            subagents=tuple(self._subagents.values()),
            plan_tool_calls=tuple(self._plan_tools),
            errors=tuple(self._errors),
            branch_summaries=tuple(self._branch_summaries),
            hooks=tuple(self._hooks),
            compaction=CompactionView(
                started=self._compaction_started,
                completed=self._compaction_completed,
                active=self._compaction_started > self._compaction_completed,
                last_reason=self._compaction_reason,
            ),
            rate_limited=self._rate_limited,
            retried=self._retried,
            user_inputs=self._user_inputs,
            assistant_messages=self._assistant_messages,
        )
 
 
#: Kind → bound-method dispatch, built once. A dict beats an if/elif chain
#: here because the fold runs per event on the hot path of a live stream.
_HANDLERS: dict = {
    "session_started": EventProjection._on_session_started,
    "session_ended": EventProjection._on_session_ended,
    "user_input_received": EventProjection._on_user_input,
    "assistant_message": EventProjection._on_assistant_message,
    "prompt_rendered": EventProjection._on_prompt_rendered,
    "provider_rate_limited": EventProjection._on_rate_limited,
    "provider_retried": EventProjection._on_retried,
    "provider_budget_consumed": EventProjection._on_budget,
    "tool_call_proposed": EventProjection._on_proposed,
    "permission_decision": EventProjection._on_permission,
    "tool_call_started": EventProjection._on_tool_started,
    "tool_call_progress": EventProjection._on_tool_progress,
    "tool_call_completed": EventProjection._on_tool_completed,
    "subagent_started": EventProjection._on_subagent_started,
    "subagent_completed": EventProjection._on_subagent_completed,
    "compaction_started": EventProjection._on_compaction_started,
    "compaction_completed": EventProjection._on_compaction_completed,
    "branch_summary": EventProjection._on_branch_summary,
    "error": EventProjection._on_error,
    "hook_executed": EventProjection._on_hook,
}
 
__all__ = [
    "CompactionView",
    "CostView",
    "EventProjection",
    "PermissionDecisionView",
    "ProjectionSnapshot",
    "ProposedCallView",
    "SubagentView",
    "ToolCallView",
]