diffview.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_ui/terminal/diffview.py
 
"""
Diff-first rendering model for the mutation tools (plan 188 T1).
 
Purpose
-------
Every comparator in the field renders an edit as a **diff**; OC Chat
rendered ``write_file`` / ``edit_file`` results as a truncated JSON
blob. This module owns the one shared, framework-free model that turns
a mutation tool's *call arguments* plus its *result payload* into a
renderable unified diff, so all three terminal presenters
(:class:`~oc_ui.terminal.presenter.ConsolePresenter`,
:class:`~oc_ui.terminal.rich_presenter.RichPresenter`, and the
:class:`~oc_ui.terminal.enhanced_presenter.EnhancedPresenter` that
subclasses it) show the same thing and degrade the same way.
 
Nothing here changes the tool contract. The diff is derived at RENDER
time from data the render lane already carries: the presenter sees the
arguments in :meth:`~oc_ui.presenter_base.Presenter.show_tool_call`
before the tool runs and the result in
:meth:`~oc_ui.presenter_base.Presenter.show_tool_result` after — this
module just pairs them (:class:`MutationArgsLedger`) and folds them
into a diff.
 
**What the render lane can and cannot know.** ``edit_file`` carries
both sides of the change (``old_string`` / ``new_string``) plus the
real ``match_line``, so its diff is a true, file-anchored unified diff.
``write_file`` carries only the NEW content — the previous contents are
gone by the time the result arrives — so its render is the full written
content as additions, and the header says exactly that rather than
implying a comparison that was never made.
 
Responsibilities
----------------
- :class:`MutationArgsLedger` — FIFO pairing of ``show_tool_call``
  arguments with the matching ``show_tool_result``.
- :class:`DiffLine` / :class:`DiffRender` — the pure, presenter-neutral
  diff model.
- :func:`build_mutation_diff` — the honesty-gated constructor:
  returns ``None`` (caller falls back to the legacy JSON line) for a
  non-mutation tool, an error / gate / confirmation result, unparseable
  or absent arguments, or a call/result pair whose paths disagree.
- :func:`diff_header_line` — the one-line summary above the body.
- :func:`render_diff_plain` — ANSI (or plain) line renderer used by
  the console presenter; Rich builds its own renderable from the same
  :class:`DiffRender`.
 
Diagnostics
-----------
Domain: OC_UI.TERMINAL
Levels:
    L2 — lifecycle events
    L3 — semantic details (guard rejections)
    L4 — deep tracing
 
Contracts
---------
- Framework-free: no ``rich`` / ``prompt_toolkit`` import here, exactly
  like :mod:`oc_ui.terminal.statusline` — the plain console must render
  a diff with no optional dependency installed.
- :func:`build_mutation_diff` NEVER guesses. Any missing or
  contradictory input returns ``None`` so the caller degrades to the
  previous rendering rather than showing a diff that might be wrong.
- Rendered body text is passed through the same secret redaction the
  read path already applies (:func:`oc_chat.tools._sandbox.redact_secrets`),
  so a diff is not a secret side-channel. Redaction is best-effort: an
  import failure degrades to unredacted text rather than losing the
  render (the content is already on the operator's own disk).
- Truncation is explicit: :attr:`DiffRender.truncated` is set, the
  hidden line count is carried, and
  :attr:`DiffRender.full_content_hint` names the on-disk file where the
  complete content lives.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- oc_chat.tools._sandbox (lazy; redact_secrets for rendered diff body text)
"""
from __future__ import annotations
 
import difflib
import json
import re
from dataclasses import dataclass
from typing import Any, Iterable
 
from oc_diagnostics import _dbg
 
_dbg("OC_UI.TERMINAL", "module", "loaded", 4)
 
 
#: The tools whose results render as a diff. Deliberately explicit — a new
#: mutating tool renders as JSON until it is added here WITH a build rule.
MUTATION_TOOLS: tuple[str, ...] = ("write_file", "edit_file")
 
#: How many body lines a diff shows before truncating. Beyond this the render
#: says so and points at the file on disk (which holds the complete content).
DEFAULT_MAX_DIFF_LINES: int = 60
 
#: Unified-diff context lines around each hunk.
_CONTEXT_LINES: int = 3
 
#: ``@@ -a,b +c,d @@`` — rewritten so the offsets are REAL file line numbers.
_HUNK_RE = re.compile(r"^@@ -(\d+)(?:,(\d+))? \+(\d+)(?:,(\d+))? @@(.*)$")
 
 
@dataclass(frozen=True, slots=True)
class DiffLine:
    """One rendered diff line.
 
    *kind* is one of ``"add"`` / ``"del"`` / ``"ctx"`` / ``"hunk"`` /
    ``"meta"``; *text* already carries the unified-diff prefix
    (``+`` / ``-`` / space) so a renderer only has to choose a colour.
    """
 
    kind: str
    text: str
 
 
@dataclass(frozen=True, slots=True)
class DiffRender:
    """A presenter-neutral diff ready to be styled and printed."""
 
    tool: str
    """The mutation tool that produced it (``write_file`` / ``edit_file``)."""
    path: str
    """The file the change landed in, as the render lane knows it."""
    lines: tuple[DiffLine, ...]
    """The (possibly truncated) diff body."""
    added: int
    """Total added lines in the WHOLE change, before truncation."""
    removed: int
    """Total removed lines in the WHOLE change, before truncation."""
    truncated: bool
    """``True`` when :attr:`lines` shows only part of the change."""
    hidden: int
    """Body lines not shown; ``0`` when :attr:`truncated` is ``False``."""
    full_content_hint: str
    """Where the complete content lives (the file on disk)."""
    detail: str = ""
    """Extra header detail (bytes written, replacement count + anchor line)."""
 
 
class MutationArgsLedger:
    """FIFO pairing of mutation-tool call arguments with their results.
 
    The ReAct loop announces every tool call (``show_tool_call``) for an
    iteration *before* dispatching, then renders every result
    (``show_tool_result``) **in the same order** — so a per-tool-name
    queue pairs them exactly, with no change to the loop. Only
    :data:`MUTATION_TOOLS` are recorded, so the ledger stays tiny.
 
    Pairing is still VERIFIED downstream: :func:`build_mutation_diff`
    compares the recorded ``path`` against the path the result echoes and
    refuses to render on disagreement. A mis-pair degrades to the legacy
    JSON line; it never renders a wrong diff.
    """
 
    __slots__ = ("_pending",)
 
    def __init__(self) -> None:
        self._pending: dict[str, list[str]] = {}
 
    def record(self, tool: str, arguments: str | None) -> None:
        """Remember *arguments* for the next result of *tool* (mutation tools only)."""
        if tool not in MUTATION_TOOLS or not arguments:
            return
        self._pending.setdefault(tool, []).append(arguments)
 
    def take(self, tool: str) -> str | None:
        """Pop the oldest recorded arguments for *tool*, or ``None``."""
        queue = self._pending.get(tool)
        if not queue:
            return None
        return queue.pop(0)
 
    def clear(self) -> None:
        """Drop every pending entry — called at turn start so a tool call that
        never produced a result (an exception mid-dispatch) cannot mis-pair
        with a later one."""
        self._pending.clear()
 
 
def _redact(text: str) -> str:
    """Apply the shared secret redaction, degrading to *text* if unavailable."""
    func = "_redact"
    try:
        from oc_chat.tools._sandbox import redact_secrets
    except Exception as exc:  # noqa: BLE001 — display must never fail on an import
        _dbg("OC_UI.TERMINAL", func, f"redaction unavailable: {type(exc).__name__}", 3)
        return text
    try:
        return redact_secrets(text)
    except Exception as exc:  # noqa: BLE001 — same: never lose the render
        _dbg("OC_UI.TERMINAL", func, f"redaction failed: {type(exc).__name__}", 3)
        return text
 
 
def _parse_args(arguments: str | None) -> dict | None:
    """Parse a tool-call ``arguments`` JSON string into a dict, else ``None``."""
    if not arguments:
        return None
    try:
        parsed = json.loads(arguments)
    except (json.JSONDecodeError, TypeError, ValueError):
        return None
    return parsed if isinstance(parsed, dict) else None
 
 
def _norm_path(value: object) -> str:
    """Normalise a path for comparison: forward slashes, no leading ``./``."""
    text = str(value or "").replace("\\", "/").strip()
    while text.startswith("./"):
        text = text[2:]
    return text
 
 
def _paths_agree(arg_path: object, result_paths: Iterable[object]) -> bool:
    """True when the recorded call path matches a path the result echoes.
 
    Exact (normalised) match first; a basename match is accepted as the
    lenient tier because ``write_file``'s ``_tmp`` redirect and the sandbox
    clamp can legitimately rewrite the directory part. Anything else is a
    mis-pair and the caller refuses to render.
    """
    a = _norm_path(arg_path)
    if not a:
        return False
    candidates = [_norm_path(p) for p in result_paths if p]
    if not candidates:
        return False
    if a in candidates:
        return True
    a_base = a.rsplit("/", 1)[-1]
    return any(a_base == c.rsplit("/", 1)[-1] for c in candidates)
 
 
def _fmt_bytes(n: int) -> str:
    """Compact byte count: ``29 B`` / ``1.2 KB`` / ``3.4 MB``."""
    if n >= 1_000_000:
        return f"{n / 1_000_000:.1f} MB"
    if n >= 1_000:
        return f"{n / 1_000:.1f} KB"
    return f"{n} B"
 
 
def _classify(line: str) -> str:
    """Map a raw unified-diff line to a :class:`DiffLine` kind."""
    if line.startswith("@@"):
        return "hunk"
    if line.startswith("+"):
        return "add"
    if line.startswith("-"):
        return "del"
    return "ctx"
 
 
def _reanchor_hunk(line: str, offset: int) -> str:
    """Shift a ``@@`` hunk header by *offset* so it names REAL file lines."""
    m = _HUNK_RE.match(line)
    if m is None:
        return line
    old_start, old_len, new_start, new_len, tail = m.groups()
    old = f"{int(old_start) + offset}" + (f",{old_len}" if old_len else "")
    new = f"{int(new_start) + offset}" + (f",{new_len}" if new_len else "")
    return f"@@ -{old} +{new} @@{tail or ''}"
 
 
def _truncate(body: list[DiffLine], max_lines: int) -> tuple[tuple[DiffLine, ...], bool, int]:
    """Clip *body* to *max_lines*, reporting whether (and how much) was hidden."""
    if max_lines <= 0 or len(body) <= max_lines:
        return tuple(body), False, 0
    return tuple(body[:max_lines]), True, len(body) - max_lines
 
 
def _build_edit_diff(args: dict, result: dict, max_lines: int) -> DiffRender | None:
    """Both sides are present — a true, file-anchored unified diff."""
    old_string = args.get("old_string")
    new_string = args.get("new_string")
    if not isinstance(old_string, str) or not isinstance(new_string, str):
        return None
 
    path = _norm_path(result.get("actual_path") or result.get("edited") or args.get("path"))
    old_lines = _redact(old_string).splitlines()
    new_lines = _redact(new_string).splitlines()
 
    raw = list(difflib.unified_diff(old_lines, new_lines, n=_CONTEXT_LINES, lineterm=""))
    # difflib prepends ``---``/``+++`` file headers; this module renders its own.
    raw = [ln for ln in raw if not (ln.startswith("---") or ln.startswith("+++"))]
    if not raw:
        return None
 
    try:
        anchor = int(result.get("match_line") or 1)
    except (TypeError, ValueError):
        anchor = 1
    offset = max(0, anchor - 1)
 
    body: list[DiffLine] = []
    added = removed = 0
    for line in raw:
        kind = _classify(line)
        if kind == "hunk":
            body.append(DiffLine("hunk", _reanchor_hunk(line, offset)))
            continue
        if kind == "add":
            added += 1
        elif kind == "del":
            removed += 1
        body.append(DiffLine(kind, line))
 
    lines, truncated, hidden = _truncate(body, max_lines)
    replacements = result.get("replacements")
    detail = f"@ line {anchor}"
    if isinstance(replacements, int) and replacements > 1:
        detail = f"{replacements} replacements, first @ line {anchor}"
    return DiffRender(
        tool="edit_file", path=path, lines=lines, added=added, removed=removed,
        truncated=truncated, hidden=hidden, full_content_hint=path, detail=detail,
    )
 
 
def _build_write_diff(args: dict, result: dict, max_lines: int) -> DiffRender | None:
    """Only the NEW side reaches the render lane — show it as additions."""
    content = args.get("content")
    if not isinstance(content, str):
        return None
 
    path = _norm_path(result.get("actual_path") or result.get("written") or args.get("path"))
    lines_text = _redact(content).splitlines()
    body = [DiffLine("add", "+" + line) for line in lines_text]
    lines, truncated, hidden = _truncate(body, max_lines)
 
    size = result.get("size_bytes")
    detail = "full content shown as additions"
    if isinstance(size, int):
        detail = f"{_fmt_bytes(size)} · full content shown as additions"
    return DiffRender(
        tool="write_file", path=path, lines=lines, added=len(body), removed=0,
        truncated=truncated, hidden=hidden, full_content_hint=path, detail=detail,
    )
 
 
def build_mutation_diff(
    tool: str,
    arguments: str | None,
    result: Any,
    *,
    max_lines: int = DEFAULT_MAX_DIFF_LINES,
) -> DiffRender | None:
    """Build a :class:`DiffRender` for a mutation tool, or ``None`` to degrade.
 
    ``None`` — meaning "render this the old way" — is returned for every
    case where a diff would be a guess: a tool this module has no rule
    for, a result that is an error / gate / pending-confirmation rather
    than an applied change, arguments that are absent or unparseable, and
    a call/result pair whose paths disagree (a mis-pairing).
    """
    func = "build_mutation_diff"
    if tool not in MUTATION_TOOLS:
        return None
    if not isinstance(result, dict):
        return None
    # Not an applied change: an error, a gate block, or a pending confirmation.
    if result.get("error") is not None or result.get("gate") or result.get("requires_confirmation"):
        _dbg("OC_UI.TERMINAL", func, f"{tool}: non-applied result; no diff", 3)
        return None
 
    args = _parse_args(arguments)
    if args is None:
        _dbg("OC_UI.TERMINAL", func, f"{tool}: no usable arguments; no diff", 3)
        return None
 
    result_paths = (result.get("original_path"), result.get("actual_path"),
                    result.get("written"), result.get("edited"), result.get("path"))
    if not _paths_agree(args.get("path"), result_paths):
        _dbg("OC_UI.TERMINAL", func, f"{tool}: call/result path mismatch; no diff", 3)
        return None
 
    if tool == "edit_file":
        return _build_edit_diff(args, result, max_lines)
    return _build_write_diff(args, result, max_lines)
 
 
#: Argument fields whose payload the diff below already shows in full, so the
#: pre-dispatch announce line renders them as a size marker instead.
_ELIDED_ARG_FIELDS: tuple[str, ...] = ("content", "old_string", "new_string")
 
 
def _elide(value: str) -> str:
    """``<12 lines, 1.2 KB>`` — a size marker standing in for a bulky payload."""
    n_lines = len(value.splitlines()) or (1 if value else 0)
    return f"<{n_lines} line(s), {_fmt_bytes(len(value.encode('utf-8', 'replace')))}>"
 
 
def summarize_mutation_call(tool: str, arguments: str | None) -> str | None:
    """Compact ``show_tool_call`` argument rendering for a mutation tool.
 
    ``show_tool_call`` fires BEFORE dispatch, so a large ``write_file``
    used to dump its entire ``content`` into scrollback and *then* render
    the diff underneath — the announce line defeated the diff. For
    :data:`MUTATION_TOOLS` the bulky fields (:data:`_ELIDED_ARG_FIELDS`)
    become size markers; every other field renders normally, so the path,
    the expected count and any future flag stay visible.
 
    Returns ``None`` for a non-mutation tool or unparseable arguments —
    the caller then renders the arguments exactly as before.
    """
    if tool not in MUTATION_TOOLS:
        return None
    args = _parse_args(arguments)
    if args is None:
        return None
    parts: list[str] = []
    for key, value in args.items():
        if key in _ELIDED_ARG_FIELDS and isinstance(value, str):
            parts.append(f"{key}={_elide(value)}")
        else:
            parts.append(f"{key}={json.dumps(value, ensure_ascii=False)}")
    return ", ".join(parts)
 
 
def diff_header_line(diff: DiffRender) -> str:
    """The one-line summary rendered above the diff body."""
    counts = f"+{diff.added}"
    if diff.removed:
        counts += f" -{diff.removed}"
    detail = f" · {diff.detail}" if diff.detail else ""
    return f"{diff.tool} · {diff.path}  ({counts}{detail})"
 
 
def diff_footer_line(diff: DiffRender) -> str | None:
    """The truncation notice, or ``None`` when the whole diff is shown."""
    if not diff.truncated:
        return None
    return (f"… +{diff.hidden} more line(s) not shown — "
            f"full content on disk at {diff.full_content_hint}")
 
 
# ─── Plain-ANSI rendering (the console presenter; Rich builds its own) ────────
 
_ANSI = {
    "reset": "\033[0m",
    "dim": "\033[2m",
    "add": "\033[32m",
    "del": "\033[31m",
    "hunk": "\033[36m",
}
 
 
def render_diff_plain(diff: DiffRender | None, *, color: bool = True) -> list[str]:
    """Render *diff* as a list of terminal lines (ANSI-styled when *color*).
 
    Returns ``[]`` for ``None`` so a caller can branch on emptiness without a
    second null check.
    """
    if diff is None:
        return []
 
    def style(key: str, text: str) -> str:
        if not color:
            return text
        return f"{_ANSI[key]}{text}{_ANSI['reset']}"
 
    out: list[str] = [style("dim", diff_header_line(diff))]
    for line in diff.lines:
        if line.kind == "add":
            out.append(style("add", line.text))
        elif line.kind == "del":
            out.append(style("del", line.text))
        elif line.kind == "hunk":
            out.append(style("hunk", line.text))
        else:
            out.append(line.text)
    footer = diff_footer_line(diff)
    if footer:
        out.append(style("dim", footer))
    return out
 
 
__all__ = [
    "MUTATION_TOOLS",
    "DEFAULT_MAX_DIFF_LINES",
    "DiffLine",
    "DiffRender",
    "MutationArgsLedger",
    "build_mutation_diff",
    "diff_header_line",
    "diff_footer_line",
    "render_diff_plain",
    "summarize_mutation_call",
]