session_postmortem.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/scripts/session_postmortem.py
 
"""
The session miner — what actually happened in a plan's dogfood sessions.
 
Purpose
-------
Worker sessions produce friction evidence — errors, denials, retry loops,
truncations, workarounds — that evaporates when the session ends. This module
mines it mechanically, per the ratified design
``docs/architecture/AGENT_DISPATCH_IDENTITY_DESIGN_2026-08-26.md`` §4.7: it
reads the plan's oc_chat dogfood sessions (session-store content — turns, tool
calls and their outcomes — plus the event streams) and emits a per-plan
findings pair at ``docs/reports/postmortems/<NNN>-postmortem.md`` and
``<NNN>-postmortem.json``.
 
The miner never concludes and never files anything. Findings become backlog
items only through operator decision at the gate (§4.7, design decision D4);
nothing here edits roadmap source, a plan, or any document beyond its own two
output files.
 
Responsibilities
----------------
- :func:`discover_sessions` — resolve which sessions belong to the plan. An
  explicit ``--session`` list takes precedence; otherwise the store is searched
  for sessions whose directory name, ``session_name`` or ``active_plan_id``
  references the plan id. What was searched and what matched is recorded in the
  output, so a miss is legible rather than silent.
- :func:`mine_session` — extract findings in the eight pinned categories
  (:data:`CATEGORIES`) from the chat log and the event stream.
- :func:`build_report` / :func:`render_json` / :func:`render_markdown` — the
  two emitters. They carry identical findings and their totals agree by
  construction: the markdown counts are read off the same ``by_category`` map
  the JSON publishes.
- :func:`main` — the ``python -m oc_chat.scripts.session_postmortem <NNN>`` CLI.
 
Read-only contract (the plan-212 miner's rule)
----------------------------------------------
Every input is opened for reading only. The module deliberately does NOT use
:mod:`oc_chat.core.persistence`'s directory helpers (``_sessions_root``,
``_session_dir``, ``_archive_dir``) because those call ``_ensure_dir`` and so
bring a directory into being as a side effect of asking about it; the local
:func:`_latest_kind` mirrors ``_glob_latest`` without that. The ONLY writes are
the two postmortem files, each written via a temp file plus ``os.replace`` so a
crash mid-emit cannot leave a torn report — and cannot touch an input either
way.
 
Diagnostics
-----------
Domain: OC_CHAT.REPLAY
    Shared with the offline replay reader deliberately: this is the same
    read-only-analysis-over-the-event-substrate surface, and the domain table in
    ``.option_c/debug_config.json`` gates on exact literals — an unlisted domain
    is silently threshold 0, and that file is outside this plan's scope.
Levels:
    L2 — CLI invocation, discovery outcome, emit.
    L3 — per-session substrate detail.
    L4 — module-load coverage marker.
 
Contracts
---------
- Exit 0 on success including the empty case (no sessions found still emits an
  honest postmortem naming what was searched); 2 on user error; 1 unexpected.
- Every category in :data:`CATEGORIES` appears in both outputs, zero included —
  an empty section is information, not omission.
- Output carries no wall-clock stamp, so re-running over unchanged inputs
  produces byte-identical files (a mined report is a function of its inputs).
- Findings are lenses, not a partition: one message can raise findings in more
  than one category, but each (locus, signal) raises at most one finding, under
  the precedence documented on each detector.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- oc_chat.core.event_bootstrap (_slug / _contained_alias_dir — reused rather
  than copied because alias validation and root containment are security-
  relevant path handling that must not drift)
- oc_chat.core.replay (read_events_from_dir — the tolerant NDJSON slot reader)
- oc_chat.core.persistence (DEFAULT_SESSIONS_DIR and the reserved dir names)
"""
from __future__ import annotations
 
import argparse
import hashlib
import json
import os
import re
import sys
from dataclasses import dataclass, field
from datetime import datetime
from pathlib import Path
from typing import Any, Optional, Sequence
 
from oc_diagnostics import _dbg
 
from oc_chat.core.event_bootstrap import _contained_alias_dir, _slug
from oc_chat.core.persistence import (
    ARCHIVE_DIR_NAME,
    DEFAULT_SESSIONS_DIR,
    LEGACY_DIR_NAME,
)
from oc_chat.core.replay import read_events_from_dir
 
_dbg("OC_CHAT.REPLAY", "session_postmortem", "module", 4)
 
#: The output contract version carried in the JSON sidecar.
SCHEMA_VERSION = "postmortem/v1"
 
#: The pinned category list, exactly design §4.7's, in the order both emitters
#: render. Order is part of the contract: a reader comparing two postmortems
#: should not have to re-sort them.
CATEGORIES: tuple[str, ...] = (
    "errors",
    "warnings",
    "tool_call_failures_and_denials",
    "truncations",
    "retry_loops",
    "missing_tool_workarounds",
    "cwd_anchor_drift",
    "blocked_time_gaps",
)
 
#: Human headings for the markdown sections.
CATEGORY_TITLES: dict[str, str] = {
    "errors": "Errors",
    "warnings": "Warnings",
    "tool_call_failures_and_denials": "Tool-call failures and denials",
    "truncations": "Truncations",
    "retry_loops": "Retry loops",
    "missing_tool_workarounds": "Missing-tool workarounds",
    "cwd_anchor_drift": "cwd / anchor drift",
    "blocked_time_gaps": "Blocked-time gaps",
}
 
#: A plan id that is safe as a filename component by construction. The id
#: reaches the output path, so validating it here is the path-traversal guard,
#: not merely input hygiene. Lettered plan numbers (``148g``) are real history.
PLAN_ID_RE = re.compile(r"^[0-9]{1,4}[a-z]?$")
 
#: One-line excerpts are bounded so a finding stays readable and so a runaway
#: payload cannot turn the report into a transcript dump.
EXCERPT_MAX_CHARS = 200
 
#: Default wall-clock silence, in seconds, that counts as blocked time. Ten
#: minutes between two consecutive events in one session is not thinking.
DEFAULT_GAP_SECONDS = 600.0
 
#: Two consecutive identical calls already ARE a retry; the run length is the
#: number of calls, so a finding names a run of >= 2.
_RETRY_RUN_MIN = 2
 
#: Completion outcomes that mean the tool call worked.
_OK_OUTCOMES = frozenset({"ok", "success"})
 
#: Denial spellings across the permission surfaces.
_DENY_DECISIONS = frozenset({"deny", "denied", "reject", "rejected"})
 
#: Substrings that mark a truncation, drawn from the strings the runtime
#: actually emits (``core/api_context.py`` footers, ``core/react_loop.py``
#: notices) rather than from a guess at what truncation looks like.
_TRUNCATION_MARKERS: tuple[str, ...] = (
    "[output truncated at",
    "[assistant reply truncated for API context",
    "[Answer truncated",
    "[Tool call(s) truncated",
    "[File truncated at limit]",
    "[File was truncated at read limit]",
    "[truncated for summariser]",
)
 
#: Substrings that mark the anchor class of failure — a command run against the
#: wrong directory, a path that is not where the caller assumed, a git command
#: outside a repository.
_DRIFT_MARKERS: tuple[str, ...] = (
    "No such file or directory",
    "The system cannot find the file specified",
    "The system cannot find the path specified",
    "cannot find the path",
    "fatal: not a git repository",
    "is not recognized as an internal or external command",
    "No such directory",
    "cwd does not exist",
    "outside the sandbox root",
)
 
#: Substrings that mark the model reaching for a tool that is not there.
_MISSING_TOOL_MARKERS: tuple[str, ...] = (
    "Unknown tool",
    "unknown tool",
    "is not a known tool",
    "No such tool",
    "tool not found",
)
 
#: Shell programs whose use through ``run_command`` duplicates a first-class
#: tool. Reaching for these is the classic missing-tool workaround: the model
#: shells out because it did not find (or could not use) the native tool.
_SHELL_WORKAROUND_COMMANDS: dict[str, str] = {
    "cat": "read_file",
    "type": "read_file",
    "head": "read_file",
    "tail": "read_file",
    "more": "read_file",
    "grep": "search_files",
    "rg": "search_files",
    "findstr": "search_files",
    "ls": "list_dir",
    "dir": "list_dir",
    "find": "search_files",
    "tree": "list_dir",
}
 
 
class _UserError(Exception):
    """Raised for user-facing errors that map to exit code 2."""
 
 
@dataclass(frozen=True)
class Finding:
    """One mined observation: its category, where it came from, what it said."""
 
    category: str
    session: str
    locus: str
    excerpt: str
 
    def to_json(self) -> dict[str, str]:
        """Return the JSON projection of this finding."""
        return {
            "category": self.category,
            "session": self.session,
            "locus": self.locus,
            "excerpt": self.excerpt,
        }
 
 
@dataclass
class SessionRecord:
    """One discovered session and the substrates found for it."""
 
    name: str
    directory: Path
    match_reason: str
    session_id: str = ""
    meta_path: Optional[Path] = None
    chat_log_path: Optional[Path] = None
    event_dirs: tuple[Path, ...] = ()
 
    def input_paths(self) -> list[Path]:
        """Return every file this record will be read from, in a stable order."""
        paths: list[Path] = []
        if self.meta_path is not None:
            paths.append(self.meta_path)
        if self.chat_log_path is not None:
            paths.append(self.chat_log_path)
        for event_dir in self.event_dirs:
            try:
                paths.extend(sorted(p for p in event_dir.iterdir() if p.is_file()))
            except OSError:
                continue  # a dir that vanished between discovery and read is not an input
        return paths
 
 
@dataclass
class Discovery:
    """What the miner searched, what it examined, and what it matched."""
 
    mode: str
    sessions_dir: str
    event_root: str
    examined: list[str] = field(default_factory=list)
    records: list[SessionRecord] = field(default_factory=list)
 
 
# ─── Small read-only helpers ─────────────────────────────────────────────────
 
def _excerpt(value: Any) -> str:
    """Collapse *value* to one bounded line, verbatim apart from whitespace."""
    text = value if isinstance(value, str) else json.dumps(value, default=str, sort_keys=True)
    flat = " ".join(text.split())
    if len(flat) > EXCERPT_MAX_CHARS:
        return flat[: EXCERPT_MAX_CHARS - 1] + "…"
    return flat
 
 
def _read_json(path: Path) -> Optional[Any]:
    """Read *path* as JSON, tolerating a BOM. Returns ``None`` when unreadable."""
    func = "_read_json"
    try:
        raw = path.read_bytes()
    except OSError as exc:
        _dbg("OC_CHAT.REPLAY", func, f"EXTERNAL_ERROR: unreadable input {path}: {exc}", 2)
        return None
    if raw.startswith(b"\xef\xbb\xbf"):
        raw = raw[3:]
    try:
        return json.loads(raw.decode("utf-8", errors="replace"))
    except json.JSONDecodeError as exc:
        _dbg("OC_CHAT.REPLAY", func, f"VALIDATION_ERROR: malformed JSON in {path}: {exc}", 2)
        return None
 
 
def _sha256(path: Path) -> str:
    """Return the hex SHA-256 of *path*, or ``"unreadable"`` when it cannot be read."""
    func = "_sha256"
    digest = hashlib.sha256()
    try:
        with open(path, "rb") as handle:
            for chunk in iter(lambda: handle.read(65536), b""):
                digest.update(chunk)
    except OSError as exc:
        _dbg("OC_CHAT.REPLAY", func, f"EXTERNAL_ERROR: cannot hash {path}: {exc}", 2)
        return "unreadable"
    return digest.hexdigest()
 
 
def _latest_kind(session_dir: Path, kind: str) -> Optional[Path]:
    """Return the current ``*_{kind}.json`` in *session_dir*, or ``None``.
 
    Mirrors ``persistence._glob_latest`` (most-recently-modified wins) WITHOUT
    its raise-on-miss and, crucially, without the ``_ensure_dir`` side effect
    the surrounding persistence path helpers carry.
    """
    try:
        matches = sorted(session_dir.glob(f"*_{kind}.json"), key=lambda p: p.stat().st_mtime)
    except OSError:
        return None
    return matches[-1] if matches else None
 
 
def _canonical_args(value: Any) -> str:
    """Return a stable string form of tool arguments for identity comparison."""
    if isinstance(value, str):
        try:
            value = json.loads(value)
        except json.JSONDecodeError:
            return " ".join(value.split())
    try:
        return json.dumps(value, sort_keys=True, default=str)
    except (TypeError, ValueError):
        return repr(value)
 
 
def _parse_ts(value: Any) -> Optional[datetime]:
    """Parse an RFC3339 timestamp; return ``None`` for anything unparseable."""
    if not isinstance(value, str) or not value:
        return None
    text = value[:-1] + "+00:00" if value.endswith("Z") else value
    try:
        return datetime.fromisoformat(text)
    except ValueError:
        return None
 
 
def _first_marker(text: str, markers: Sequence[str]) -> Optional[str]:
    """Return the first marker in *markers* present in *text*, else ``None``."""
    for marker in markers:
        if marker in text:
            return marker
    return None
 
 
# ─── Discovery ───────────────────────────────────────────────────────────────
 
def _plan_tokens(plan_id: str) -> tuple[str, ...]:
    """Return the spellings of *plan_id* a session name or metadata may carry."""
    return (plan_id, f"plan-{plan_id}", f"plan_{plan_id}", f"plan {plan_id}")
 
 
def _name_references_plan(name: str, plan_id: str) -> bool:
    """True when *name* references *plan_id* as a token, not as a substring.
 
    ``2240`` and ``1224`` must not answer for plan ``224``, so the id has to sit
    on a token boundary — the separators a session name actually uses are
    hyphen, underscore, space and dot.
    """
    pattern = re.compile(rf"(?:^|[^0-9a-z]){re.escape(plan_id)}(?:$|[^0-9a-z])", re.IGNORECASE)
    return bool(pattern.search(name))
 
 
def _match_reason(plan_id: str, dir_name: str, meta: Optional[dict]) -> Optional[str]:
    """Return why this session matches *plan_id*, or ``None`` when it does not."""
    session_name = ""
    active_plan = ""
    if isinstance(meta, dict):
        session_name = str(meta.get("session_name") or "")
        active_plan = str(meta.get("active_plan_id") or "")
    if active_plan and (active_plan == plan_id or _name_references_plan(active_plan, plan_id)):
        return f"metadata active_plan_id={active_plan!r}"
    if _name_references_plan(dir_name, plan_id):
        return f"directory name {dir_name!r} references the plan id"
    if session_name and _name_references_plan(session_name, plan_id):
        return f"metadata session_name={session_name!r}"
    for token in _plan_tokens(plan_id)[1:]:
        if token.lower() in dir_name.lower():
            return f"directory name {dir_name!r} contains {token!r}"
    return None
 
 
def _event_dirs_for(name: str, meta: Optional[dict], event_root: Path) -> tuple[Path, ...]:
    """Return the existing event directories for a session, in merge order.
 
    Mirrors ``event_bootstrap.event_read_dirs`` from the persisted metadata
    rather than from a live ``ChatSession``, because constructing one to ask a
    question about a stored session would be a write-capable object in a
    read-only tool.
    """
    func = "_event_dirs_for"
    out: list[Path] = []
    seen: set[str] = set()
 
    def _add(path: Optional[Path]) -> None:
        if path is None:
            return
        try:
            if not path.is_dir():
                return
            key = str(path.resolve(strict=False)).casefold()
        except OSError:
            return
        if key in seen:
            return
        seen.add(key)
        out.append(path)
 
    session_id = ""
    aliases: list[str] = []
    if isinstance(meta, dict):
        session_id = str(meta.get("session_id") or "")
        raw_aliases = meta.get("legacy_event_slugs")
        if isinstance(raw_aliases, (list, tuple)):
            aliases = [a for a in raw_aliases if isinstance(a, str)]
 
    name_slug = _slug(name)
    id_key = session_id or name_slug
    if name_slug != id_key:
        _add(event_root / name_slug)
    for alias in aliases:
        _add(_contained_alias_dir(event_root, alias))
    _add(event_root / id_key)
    _dbg("OC_CHAT.REPLAY", func, f"{name}: {len(out)} event dir(s)", 3)
    return tuple(out)
 
 
def _record_for_dir(
    session_dir: Path,
    reason: str,
    event_root: Path,
    meta: Any = None,
    meta_path: Optional[Path] = None,
) -> SessionRecord:
    """Build a :class:`SessionRecord` for an existing session directory.
 
    *meta* / *meta_path* let a caller that has already read the metadata (the
    store search does, to decide the match) hand it over rather than have the
    file opened a second time.
    """
    if meta_path is None:
        meta_path = _latest_kind(session_dir, "meta")
        meta = _read_json(meta_path) if meta_path is not None else None
    meta_dict = meta if isinstance(meta, dict) else None
    name = session_dir.name
    if meta_dict and isinstance(meta_dict.get("session_name"), str) and meta_dict["session_name"]:
        name = meta_dict["session_name"]
    return SessionRecord(
        name=name,
        directory=session_dir,
        match_reason=reason,
        session_id=str((meta_dict or {}).get("session_id") or ""),
        meta_path=meta_path,
        chat_log_path=_latest_kind(session_dir, "chat_log"),
        event_dirs=_event_dirs_for(name, meta_dict, event_root),
    )
 
 
def _resolve_explicit(raw: str) -> Path:
    """Resolve one ``--session`` argument to a session directory.
 
    Accepts the directory itself or either of its two sidecar files, because a
    caller who has just been reading a ``*_meta.json`` should be able to paste
    that path.
    """
    path = Path(raw)
    if path.is_file():
        path = path.parent
    if not path.is_dir():
        raise _UserError(f"--session path is not a session directory: {raw}")
    return path
 
 
def discover_sessions(
    plan_id: str,
    sessions_dir: Path,
    event_root: Path,
    explicit: Sequence[str] = (),
) -> Discovery:
    """Resolve the sessions for *plan_id*, recording what was searched.
 
    An explicit list wins outright: the operator naming sessions is a stronger
    signal than any name heuristic, and honouring it silently alongside a store
    search would make the report's provenance ambiguous.
    """
    func = "discover_sessions"
    if explicit:
        records = [
            _record_for_dir(_resolve_explicit(raw), "named explicitly with --session", event_root)
            for raw in explicit
        ]
        _dbg("OC_CHAT.REPLAY", func, f"explicit mode: {len(records)} session(s)", 2)
        return Discovery(
            mode="explicit",
            sessions_dir=str(sessions_dir),
            event_root=str(event_root),
            examined=[str(r.directory) for r in records],
            records=records,
        )
 
    examined: list[str] = []
    records: list[SessionRecord] = []
    if sessions_dir.is_dir():
        for child in sorted(sessions_dir.iterdir()):
            if not child.is_dir() or child.name in (ARCHIVE_DIR_NAME, LEGACY_DIR_NAME):
                continue
            examined.append(child.name)
            meta_path = _latest_kind(child, "meta")
            meta = _read_json(meta_path) if meta_path is not None else None
            reason = _match_reason(plan_id, child.name, meta if isinstance(meta, dict) else None)
            if reason is not None:
                records.append(_record_for_dir(child, reason, event_root, meta, meta_path))
    _dbg(
        "OC_CHAT.REPLAY", func,
        f"store search: examined {len(examined)}, matched {len(records)}", 2,
    )
    return Discovery(
        mode="store-search",
        sessions_dir=str(sessions_dir),
        event_root=str(event_root),
        examined=examined,
        records=records,
    )
 
 
# ─── Category extraction — the chat-log substrate ────────────────────────────
 
def _tool_payload(message: dict) -> Optional[dict]:
    """Return the parsed dict a ``role:"tool"`` message carries, if it is one."""
    content = message.get("content")
    if not isinstance(content, str) or not content.strip():
        return None
    try:
        parsed = json.loads(content)
    except json.JSONDecodeError:
        return None
    return parsed if isinstance(parsed, dict) else None
 
 
def _outcome_finding(
    session: str, locus: str, tool_name: str, payload: dict, raw: str,
) -> Optional[Finding]:
    """Return the ONE outcome-family finding for a tool result, if any.
 
    Precedence, most specific first: a missing dependency is a missing-tool
    workaround; an error or a non-zero return code is drift when its text names
    the anchor class of failure, otherwise a tool-call failure; a pending
    confirmation is a denial in the same sense the design means it (the call
    did not run).
    """
    if payload.get("missing_dependency"):
        return Finding(
            "missing_tool_workarounds", session, locus,
            _excerpt(f"{tool_name}: missing_dependency={payload['missing_dependency']!r} "
                     f"hint={payload.get('install_hint', '')!r}"),
        )
    error = payload.get("error")
    if isinstance(error, str) and error.strip():
        category = "cwd_anchor_drift" if _first_marker(error, _DRIFT_MARKERS) else "tool_call_failures_and_denials"
        return Finding(category, session, locus, _excerpt(f"{tool_name}: {error}"))
    if payload.get("requires_confirmation"):
        return Finding(
            "tool_call_failures_and_denials", session, locus,
            _excerpt(f"{tool_name}: awaiting confirmation for "
                     f"{payload.get('command', payload.get('message', ''))}"),
        )
    returncode = payload.get("returncode")
    if isinstance(returncode, int) and returncode != 0:
        stderr = payload.get("stderr") if isinstance(payload.get("stderr"), str) else ""
        detail = stderr or raw
        category = "cwd_anchor_drift" if _first_marker(detail, _DRIFT_MARKERS) else "errors"
        return Finding(category, session, locus, _excerpt(f"{tool_name}: rc={returncode} {stderr}"))
    return None
 
 
def _warning_findings(session: str, locus: str, tool_name: str, payload: dict) -> list[Finding]:
    """Return the warning-family findings for a tool result.
 
    A declared ``warnings`` entry is one finding each; stderr text alongside a
    SUCCESSFUL command is the undeclared form of the same thing (a failing
    command's stderr is already reported by the outcome family, so counting it
    here too would double-report one event).
    """
    out: list[Finding] = []
    warnings = payload.get("warnings")
    if isinstance(warnings, (list, tuple)):
        for warning in warnings:
            if warning:
                out.append(Finding("warnings", session, locus, _excerpt(f"{tool_name}: {warning}")))
    stderr = payload.get("stderr")
    if payload.get("returncode") == 0 and isinstance(stderr, str) and stderr.strip():
        out.append(Finding("warnings", session, locus, _excerpt(f"{tool_name}: stderr {stderr}")))
    return out
 
 
def _shell_workaround(tool_name: str, arguments: Any) -> Optional[str]:
    """Return the native tool a shelled-out command duplicates, else ``None``."""
    if tool_name != "run_command":
        return None
    args = arguments
    if isinstance(args, str):
        try:
            args = json.loads(args)
        except json.JSONDecodeError:
            return None
    if not isinstance(args, dict):
        return None
    command = args.get("command")
    if not isinstance(command, str) or not command.strip():
        return None
    base = command.split()[0].rsplit("/", 1)[-1].rsplit("\\", 1)[-1].lower()
    base = base[:-4] if base.endswith(".exe") else base
    return _SHELL_WORKAROUND_COMMANDS.get(base)
 
 
def _mine_chat_log(record: SessionRecord, messages: list) -> list[Finding]:
    """Extract findings from one session's chat-log message list."""
    func = "_mine_chat_log"
    findings: list[Finding] = []
    call_sequence: list[tuple[str, str, int]] = []
 
    for index, message in enumerate(messages):
        if not isinstance(message, dict):
            continue
        role = message.get("role")
        locus = f"chat_log[message {index}]"
        raw_content = message.get("content") if isinstance(message.get("content"), str) else ""
 
        if role == "tool":
            tool_name = str(message.get("name") or "?")
            locus = f"chat_log[message {index}] tool={tool_name}"
            payload = _tool_payload(message)
            if payload is not None:
                outcome = _outcome_finding(record.name, locus, tool_name, payload, raw_content)
                if outcome is not None:
                    findings.append(outcome)
                findings.extend(_warning_findings(record.name, locus, tool_name, payload))
                if payload.get("truncated") is True:
                    findings.append(Finding(
                        "truncations", record.name, locus,
                        _excerpt(f"{tool_name}: truncated=true total_size={payload.get('total_size')}"),
                    ))
                    raw_content = ""  # already reported; do not double-count via a marker
            marker = _first_marker(raw_content, _TRUNCATION_MARKERS)
            if marker is not None:
                findings.append(Finding("truncations", record.name, locus, _excerpt(f"{tool_name}: {marker}")))
            missing = _first_marker(raw_content, _MISSING_TOOL_MARKERS)
            if missing is not None:
                findings.append(Finding(
                    "missing_tool_workarounds", record.name, locus, _excerpt(raw_content),
                ))
 
        elif role == "assistant":
            marker = _first_marker(raw_content, _TRUNCATION_MARKERS)
            if marker is not None:
                findings.append(Finding("truncations", record.name, locus, _excerpt(marker)))
            for call in message.get("tool_calls") or []:
                if not isinstance(call, dict):
                    continue
                function = call.get("function") if isinstance(call.get("function"), dict) else {}
                name = str(function.get("name") or call.get("name") or "?")
                arguments = function.get("arguments", call.get("arguments"))
                call_sequence.append((name, _canonical_args(arguments), index))
                native = _shell_workaround(name, arguments)
                if native is not None:
                    findings.append(Finding(
                        "missing_tool_workarounds", record.name,
                        f"chat_log[message {index}] tool={name}",
                        _excerpt(f"shelled out where {native} exists: {_canonical_args(arguments)}"),
                    ))
 
    findings.extend(_retry_findings(record.name, call_sequence, "chat_log"))
    _dbg("OC_CHAT.REPLAY", func, f"{record.name}: {len(findings)} chat-log finding(s)", 3)
    return findings
 
 
def _retry_findings(session: str, sequence: list[tuple[str, str, int]], substrate: str) -> list[Finding]:
    """Return one finding per maximal run of >= 2 identical consecutive calls."""
    findings: list[Finding] = []
    start = 0
    while start < len(sequence):
        end = start + 1
        while end < len(sequence) and sequence[end][:2] == sequence[start][:2]:
            end += 1
        run = end - start
        if run >= _RETRY_RUN_MIN:
            name, args, first_locus = sequence[start]
            findings.append(Finding(
                "retry_loops", session,
                f"{substrate}[{first_locus}..{sequence[end - 1][2]}]",
                _excerpt(f"{run}x identical consecutive call to {name}({args})"),
            ))
        start = end
    return findings
 
 
# ─── Category extraction — the event substrate ───────────────────────────────
 
def _event_locus(event: Any) -> str:
    """Return a stable, readable locus string for one event."""
    turn = (getattr(event, "turn_id", None) or "")[:8]
    return f"events[{getattr(event, 'event_id', '?')}]" + (f" turn={turn}" if turn else "")
 
 
def _mine_events(record: SessionRecord, events: list, gap_seconds: float) -> list[Finding]:
    """Extract findings from one session's ordered event stream."""
    func = "_mine_events"
    findings: list[Finding] = []
    proposals: list[tuple[str, str, int]] = []
 
    for index, event in enumerate(events):
        payload = getattr(event, "payload", None)
        kind = getattr(event, "kind", "")
        locus = _event_locus(event)
 
        if kind == "error":
            findings.append(Finding(
                "errors", record.name, locus, _excerpt(getattr(payload, "error_summary", "") or "error"),
            ))
        elif kind == "hook_executed" and getattr(payload, "exit_code", 0):
            findings.append(Finding(
                "errors", record.name, locus,
                _excerpt(f"hook {getattr(payload, 'hook_name', '?')} exited "
                         f"{getattr(payload, 'exit_code', '?')}"),
            ))
        elif kind == "session_ended":
            outcome = str(getattr(payload, "outcome", "") or "")
            if outcome and outcome not in _OK_OUTCOMES:
                findings.append(Finding(
                    "errors", record.name, locus, _excerpt(f"session ended with outcome={outcome}"),
                ))
        elif kind == "tool_call_completed":
            outcome = str(getattr(payload, "outcome", "") or "")
            if outcome and outcome not in _OK_OUTCOMES:
                summary = getattr(payload, "error_summary", None) or ""
                findings.append(Finding(
                    "tool_call_failures_and_denials", record.name, locus,
                    _excerpt(f"outcome={outcome} {summary}".strip()),
                ))
        elif kind == "permission_decision":
            decision = str(getattr(payload, "decision", "") or "").lower()
            final = str(getattr(payload, "final_outcome", "") or "").lower()
            if decision in _DENY_DECISIONS or final in _DENY_DECISIONS:
                reason = getattr(payload, "denial_reason_code", None) or ""
                findings.append(Finding(
                    "tool_call_failures_and_denials", record.name, locus,
                    _excerpt(f"decision={decision or final} by "
                             f"{getattr(payload, 'decided_by', '?')} {reason}".strip()),
                ))
        elif kind == "provider_retried":
            findings.append(Finding(
                "retry_loops", record.name, locus,
                _excerpt(f"provider {getattr(payload, 'provider_name', '?')} retry attempt "
                         f"{getattr(payload, 'attempt', '?')}"),
            ))
        elif kind == "provider_rate_limited":
            findings.append(Finding(
                "retry_loops", record.name, locus,
                _excerpt(f"provider {getattr(payload, 'provider_name', '?')} rate limited; "
                         f"retry_after={getattr(payload, 'retry_after_seconds', None)}"),
            ))
        elif kind == "tool_call_proposed":
            proposals.append((
                str(getattr(payload, "tool_name", "") or "?"),
                _canonical_args(getattr(payload, "requested_args", {}) or {}),
                index,
            ))
        elif kind == "assistant_message":
            marker = _first_marker(str(getattr(payload, "text", "") or ""), _TRUNCATION_MARKERS)
            if marker is not None:
                findings.append(Finding("truncations", record.name, locus, _excerpt(marker)))
 
    findings.extend(_retry_findings(record.name, proposals, "events"))
    findings.extend(_gap_findings(record.name, events, gap_seconds))
    _dbg("OC_CHAT.REPLAY", func, f"{record.name}: {len(findings)} event finding(s)", 3)
    return findings
 
 
def _gap_findings(session: str, events: list, gap_seconds: float) -> list[Finding]:
    """Return one finding per pair of consecutive events separated by blocked time."""
    findings: list[Finding] = []
    previous: Optional[tuple[datetime, Any]] = None
    for event in events:
        stamp = _parse_ts(getattr(event, "ts", None))
        if stamp is None:
            continue
        if previous is not None:
            delta = (stamp - previous[0]).total_seconds()
            if delta >= gap_seconds:
                findings.append(Finding(
                    "blocked_time_gaps", session,
                    f"{_event_locus(previous[1])} -> {_event_locus(event)}",
                    _excerpt(f"{delta:.0f}s of silence between "
                             f"{getattr(previous[1], 'kind', '?')} and {getattr(event, 'kind', '?')}"),
                ))
        previous = (stamp, event)
    return findings
 
 
def mine_session(record: SessionRecord, gap_seconds: float = DEFAULT_GAP_SECONDS) -> list[Finding]:
    """Mine one session across both substrates, in category-stable order."""
    func = "mine_session"
    findings: list[Finding] = []
 
    if record.chat_log_path is not None:
        chat_log = _read_json(record.chat_log_path)
        messages = chat_log.get("messages") if isinstance(chat_log, dict) else None
        if isinstance(messages, list):
            findings.extend(_mine_chat_log(record, messages))
 
    events: list = []
    for event_dir in record.event_dirs:
        read, _skipped = read_events_from_dir(event_dir)
        events.extend(read)
    if record.session_id:
        events = [e for e in events if getattr(e, "session_id", "") == record.session_id]
    events.sort(key=lambda e: (str(getattr(e, "ts", "")),))
    if events:
        findings.extend(_mine_events(record, events, gap_seconds))
 
    _dbg("OC_CHAT.REPLAY", func, f"{record.name}: {len(findings)} finding(s) total", 2)
    return findings
 
 
# ─── Report assembly + emitters ──────────────────────────────────────────────
 
def build_report(plan_id: str, discovery: Discovery, gap_seconds: float = DEFAULT_GAP_SECONDS) -> dict:
    """Mine every discovered session and assemble the report structure."""
    func = "build_report"
    findings: list[Finding] = []
    sessions: list[dict] = []
    inputs: list[dict] = []
 
    for record in discovery.records:
        record_findings = mine_session(record, gap_seconds)
        findings.extend(record_findings)
        sessions.append({
            "session": record.name,
            "directory": str(record.directory),
            "match_reason": record.match_reason,
            "session_id": record.session_id,
            "substrates": {
                "metadata": str(record.meta_path) if record.meta_path else None,
                "chat_log": str(record.chat_log_path) if record.chat_log_path else None,
                "event_dirs": [str(d) for d in record.event_dirs],
            },
            "finding_count": len(record_findings),
        })
        for path in record.input_paths():
            inputs.append({
                "path": str(path),
                "sha256": _sha256(path),
                "bytes": path.stat().st_size if path.exists() else 0,
            })
 
    by_category = {category: 0 for category in CATEGORIES}
    for finding in findings:
        by_category[finding.category] = by_category.get(finding.category, 0) + 1
 
    ordered = sorted(findings, key=lambda f: (CATEGORIES.index(f.category), f.session, f.locus))
    _dbg("OC_CHAT.REPLAY", func, f"plan {plan_id}: {len(ordered)} finding(s)", 2)
    return {
        "schema": SCHEMA_VERSION,
        "plan_id": plan_id,
        "discovery": {
            "mode": discovery.mode,
            "searched": {
                "sessions_dir": discovery.sessions_dir,
                "event_root": discovery.event_root,
                "session_dirs_examined": discovery.examined,
            },
            "matched": sessions,
        },
        "inputs": inputs,
        "totals": {"findings": len(ordered), "by_category": by_category},
        "findings": [f.to_json() for f in ordered],
        "gap_seconds": gap_seconds,
    }
 
 
def render_json(report: dict) -> str:
    """Render *report* as the JSON sidecar (stable key order, trailing newline)."""
    return json.dumps(report, indent=2, sort_keys=False, ensure_ascii=False) + "\n"
 
 
def render_markdown(report: dict) -> str:
    """Render *report* as the markdown view of the SAME findings.
 
    Counts come from ``totals.by_category`` rather than from a second pass over
    the findings, so the two outputs cannot disagree by construction.
    """
    plan_id = report["plan_id"]
    discovery = report["discovery"]
    searched = discovery["searched"]
    totals = report["totals"]
    lines: list[str] = [
        f"# Plan {plan_id} — session postmortem",
        "",
        f"Mined by `python -m oc_chat.scripts.session_postmortem {plan_id}` "
        f"({report['schema']}). Read-only over every input; findings are candidates for "
        "operator judgement at the gate, never filed automatically.",
        "",
        "## What was searched",
        "",
        f"- Discovery mode: `{discovery['mode']}`",
        f"- Session store: `{searched['sessions_dir']}`",
        f"- Event root: `{searched['event_root']}`",
        f"- Session directories examined: {len(searched['session_dirs_examined'])}"
        + (f" ({', '.join(searched['session_dirs_examined'])})" if searched["session_dirs_examined"] else ""),
        f"- Blocked-time threshold: {report['gap_seconds']:.0f}s",
        "",
        "## What matched",
        "",
    ]
    if not discovery["matched"]:
        lines += [
            f"No session matched plan {plan_id}. Nothing was mined; this postmortem records the "
            "search, not a finding of health.",
            "",
        ]
    else:
        lines.append("| Session | Match reason | Chat log | Event dirs | Findings |")
        lines.append("|---|---|---|---|---|")
        for entry in discovery["matched"]:
            substrates = entry["substrates"]
            lines.append(
                f"| `{entry['session']}` | {entry['match_reason']} | "
                f"{'yes' if substrates['chat_log'] else 'no'} | "
                f"{len(substrates['event_dirs'])} | {entry['finding_count']} |"
            )
        lines.append("")
 
    lines += [
        "## Findings",
        "",
        f"Total: {totals['findings']} across {len(CATEGORIES)} categories. "
        "A category with zero findings is listed as zero — an empty section is information, "
        "not omission.",
        "",
    ]
    by_category = totals["by_category"]
    findings = report["findings"]
    for category in CATEGORIES:
        count = by_category.get(category, 0)
        lines.append(f"### {CATEGORY_TITLES[category]} — {count} finding(s)")
        lines.append("")
        if count == 0:
            lines.append("_None._")
            lines.append("")
            continue
        for finding in findings:
            if finding["category"] != category:
                continue
            lines.append(f"- `{finding['session']}` @ `{finding['locus']}` — {finding['excerpt']}")
        lines.append("")
 
    lines += [
        "## Inputs (read-only)",
        "",
        f"{len(report['inputs'])} file(s) were opened for reading. Their digests are recorded so a "
        "reader can prove the miner left the store byte-identical.",
        "",
    ]
    for entry in report["inputs"]:
        lines.append(f"- `{entry['path']}` — sha256 `{entry['sha256']}` ({entry['bytes']} bytes)")
    lines.append("")
    return "\n".join(lines)
 
 
def _write_atomic(path: Path, text: str) -> None:
    """Write *text* to *path* via a temp file + ``os.replace``.
 
    A crash mid-emit must not leave a torn report. Inputs are untouched either
    way — the miner only ever writes here.
    """
    func = "_write_atomic"
    path.parent.mkdir(parents=True, exist_ok=True)
    tmp = path.with_name(path.name + ".tmp")
    tmp.write_text(text, encoding="utf-8", newline="\n")
    os.replace(tmp, path)
    _dbg("OC_CHAT.REPLAY", func, f"wrote {path}", 2)
 
 
def emit(report: dict, out_dir: Path) -> tuple[Path, Path]:
    """Write the markdown + JSON pair for *report*; return both paths."""
    plan_id = report["plan_id"]
    md_path = out_dir / f"{plan_id}-postmortem.md"
    json_path = out_dir / f"{plan_id}-postmortem.json"
    _write_atomic(md_path, render_markdown(report))
    _write_atomic(json_path, render_json(report))
    return md_path, json_path
 
 
# ─── CLI ─────────────────────────────────────────────────────────────────────
 
def default_out_dir() -> Path:
    """Return the committed postmortem home, resolved from this file's location."""
    return Path(__file__).resolve().parents[1] / "docs" / "reports" / "postmortems"
 
 
def _build_parser() -> argparse.ArgumentParser:
    """Build the ``session_postmortem`` argument parser."""
    parser = argparse.ArgumentParser(
        prog="python -m oc_chat.scripts.session_postmortem",
        description="Mine a plan's oc_chat dogfood sessions into a postmortem (read-only).",
    )
    parser.add_argument("plan_id", help="the plan number, e.g. 224 (lettered ids such as 148g are accepted)")
    parser.add_argument("--session", action="append", default=[], metavar="PATH",
                        help="explicit session directory (repeatable); takes precedence over the store search")
    parser.add_argument("--sessions-dir", default=None, help="session store root to search")
    # OCT-LINT: disable=no_hardcoded_secrets reason="argparse --help prose mis-flagged as a high-entropy string; not a secret"
    parser.add_argument("--event-root", default=None, help="event-log root (default: OC_CHAT_EVENT_DIR or <cwd>/.oc_chat/events)")
    parser.add_argument("--out-dir", default=None, help="where to write the two postmortem files")
    parser.add_argument("--gap-seconds", type=float, default=DEFAULT_GAP_SECONDS,
                        help=f"blocked-time threshold in seconds (default {DEFAULT_GAP_SECONDS:.0f})")
    parser.add_argument("--quiet", action="store_true", help="suppress the stdout summary")
    return parser
 
 
def _resolve_event_root(raw: Optional[str]) -> Path:
    """Resolve the event root: the flag, else the env override, else the default."""
    if raw:
        return Path(raw)
    env_dir = os.environ.get("OC_CHAT_EVENT_DIR")
    return Path(env_dir) if env_dir else Path.cwd() / ".oc_chat" / "events"
 
 
def main(argv: Optional[Sequence[str]] = None) -> int:
    """Entry point. Returns the process exit code (0 ok, 1 error, 2 user error)."""
    func = "main"
    for stream in (sys.stdout, sys.stderr):
        try:
            stream.reconfigure(errors="replace")
        except (AttributeError, ValueError):
            pass  # non-reconfigurable stream (tests, pipes wrapped by harnesses)
    args = _build_parser().parse_args(argv)
    try:
        plan_id = str(args.plan_id).strip()
        if not PLAN_ID_RE.match(plan_id):
            raise _UserError(f"plan id must match {PLAN_ID_RE.pattern}; got {args.plan_id!r}")
        if args.gap_seconds <= 0:
            raise _UserError("--gap-seconds must be positive")
        sessions_dir = Path(args.sessions_dir) if args.sessions_dir else Path(DEFAULT_SESSIONS_DIR)
        event_root = _resolve_event_root(args.event_root)
        out_dir = Path(args.out_dir) if args.out_dir else default_out_dir()
        _dbg("OC_CHAT.REPLAY", func, f"plan={plan_id} store={sessions_dir} events={event_root}", 2)
 
        discovery = discover_sessions(plan_id, sessions_dir, event_root, args.session)
        report = build_report(plan_id, discovery, args.gap_seconds)
        md_path, json_path = emit(report, out_dir)
        if not args.quiet:
            print(f"plan {plan_id}: {len(discovery.records)} session(s) matched, "
                  f"{report['totals']['findings']} finding(s)")
            print(f"  {md_path}")
            print(f"  {json_path}")
        return 0
    except _UserError as exc:
        print(f"error: {exc}", file=sys.stderr)
        return 2
    except Exception as exc:  # noqa: BLE001 — CLI boundary: no stack trace to the user
        _dbg("OC_CHAT.REPLAY", func, f"SYSTEM_ERROR: {type(exc).__name__}: {exc}", 1, override=True)
        print(f"error: {type(exc).__name__}: {exc}", file=sys.stderr)
        return 1
 
 
if __name__ == "__main__":  # pragma: no cover
    sys.exit(main())