_common.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/registry/render/roadmap/_common.py
 
# OCT-LINT: disable=tests_exist reason="Covered by oc_chat/tests/registry/roadmap/test_render_common.py"
"""
Shared plumbing for the living roadmap GeneratedView set (Plan 142 P1.6, §6).
 
Purpose
-------
The pieces every roadmap GeneratedView shares: the **parameterized generated
header** (build + parse), the canonical **renderer-id** set, the **glyph/label
maps** for the §7 status/priority enums, and the **render guard** that forbids
reading a runtime-excluded table or an EXCLUDED column from a source-only
projection (Registry Source & Storage Spec V0.22 §3.2 / §6 / §7 — AC §8.30).
The crash-safe atomic writer is reused from ``registry/render/_common.py``
(the §6 shared ``atomic_write_bytes``).
 
Responsibilities
----------------
- ``build_generated_header`` / ``parse_generated_header`` — the §6 four-line
  HTML-comment header carrying ``source_hash`` / ``logical_state_hash`` /
  ``renderer``.
- ``status_glyph`` / ``status_label`` / ``priority_glyph`` — the §7 enum→glyph
  mapping (single source for every renderer; AC §8.7).
- ``assert_render_query_safe`` / ``guarded_query`` — reject a render query that
  reads a runtime-excluded table or an EXCLUDED column (AC §8.30).
- ``atomic_write_text`` — atomic UTF-8 (LF, no BOM) write via the shared writer.
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 - header lifecycle
    L3 - parsed header fields
    L4 - per-query guard probes
 
Contracts
---------
- The header is output-only metadata; ``render --check`` parses it and treats a
  drifted/absent header (a hand-edit) as ``E_ROADMAP_UNTRACKED_EDIT`` drift.
- A source-only GeneratedView reads only INCLUDED columns of source-rebuildable
  tables; the guard fails closed on any runtime-excluded table/column token.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- oc_chat.registry.render._common (the shared atomic writer)
"""
from __future__ import annotations
 
import posixpath
import re
import sqlite3
from pathlib import Path
from typing import Optional
 
from oc_diagnostics import _dbg
 
from oc_chat.registry.render._common import atomic_write_bytes
 
_dbg("OC_CHAT.MAIN", "module", "loaded", 4)
 
#: The four canonical artifact-renderer ids (§6).
RENDERER_LEAN = "roadmap-lean-v4"
RENDERER_EXTENDED = "roadmap-extended-v3"
RENDERER_CHANGELOG = "roadmap-changelog-v1"
RENDERER_ROOT_CHANGELOG = "changelog-product-v2"
RENDERER_IDS = frozenset(
    {RENDERER_LEAN, RENDERER_EXTENDED, RENDERER_CHANGELOG, RENDERER_ROOT_CHANGELOG}
)
 
#: The current rendered living-set filenames (un-versioned, fixed) — the single
#: source of truth for both ``ARTIFACTS`` and the generated footer's companion
#: links (so the footer never names a stale versioned file like ROADMAP-v3.38.md).
LIVING_SET_FILENAMES: dict[str, str] = {
    "roadmap": "ROADMAP.md",
    "roadmap_extended": "ROADMAP-EXTENDED.md",
    "roadmap_changelog": "ROADMAP-CHANGELOG.md",
    "root_changelog": "CHANGELOG.md",
}
 
#: Human label for each companion link in the generated footer.
_FOOTER_LABELS: dict[str, str] = {
    "roadmap": "Lean roadmap",
    "roadmap_extended": "Extended record",
    "roadmap_changelog": "Roadmap changelog",
    "root_changelog": "Product changelog",
}
 
#: Each living view's write-dir relative to the repo root — ``artifact_paths`` writes
#: the three roadmap views to ``oc_chat/roadmap/`` and the root changelog to the repo
#: root, so the footer's companion links must be artifact-relative, not flat.
_ARTIFACT_DIRS: dict[str, str] = {
    "roadmap": "oc_chat/roadmap",
    "roadmap_extended": "oc_chat/roadmap",
    "roadmap_changelog": "oc_chat/roadmap",
    "root_changelog": "",
}
 
 
def _companion_link(self_key: str, target_key: str) -> str:
    """The artifact-relative link from ``self_key``'s file to ``target_key``'s file:
    a same-dir companion stays flat; a roadmap view → the root changelog becomes
    ``../../CHANGELOG.md``; the root changelog → a roadmap view becomes
    ``oc_chat/roadmap/ROADMAP*.md`` (purely lexical via ``posixpath``)."""
    target = posixpath.join(_ARTIFACT_DIRS[target_key], LIVING_SET_FILENAMES[target_key])
    return posixpath.relpath(target, _ARTIFACT_DIRS[self_key] or ".")
 
 
def generated_footer(self_key: str) -> list[str]:
    """The generated footer for a living view: a do-not-hand-edit provenance note
    + **artifact-relative** companion links to the *current* rendered set (never a
    stale versioned filename, and resolvable from this artifact's own write-dir).
    ``self_key`` is the artifact being rendered (excluded from its own list)."""
    companions = " · ".join(
        f"[{_FOOTER_LABELS[key]}]({_companion_link(self_key, key)})"
        for key in LIVING_SET_FILENAMES
        if key != self_key
    )
    return [
        "---",
        "",
        "_Generated from the registry source (`oc_chat/roadmap/source`) — do not "
        "hand-edit; edit the source records and run `oct agent roadmap render` to "
        f"regenerate. Companion views: {companions}._",
    ]
 
#: Tables runtime-excluded from the §3.2 logical hash — a source-only
#: GeneratedView must never read these (AC §8.30). A future view that renders
#: plan-link traceability must carry a runtime-checkpoint-hash header instead.
RUNTIME_EXCLUDED_TABLES = frozenset(
    {
        "roadmap_plan_links",
        "roadmap_item_events",
        "agents",
        "rounds",
        "plans",
        "operation_log",
        "auth_events",
    }
)
 
#: Columns EXCLUDED from the §3.2 logical hash (volatile timestamps, runtime
#: provenance, the v2-deferred prose_path) — a renderer must not emit them, since
#: they differ across rebuilds and would break the byte-stable-render guarantee.
EXCLUDED_COLUMNS = frozenset(
    {
        "created_at",
        "updated_at",
        "last_updated_at",
        "closed_round_no",
        "created_by_slot",
        "prose_path",
    }
)
 
#: §7 status token → glyph (source == store tokens are canonical).
_STATUS_GLYPH = {
    "todo": "⬜",
    "in_progress": "🔄",
    "done": "✅",
    "blocked": "🔒",
    "deferred": "↩",  # overridden to 🚫 when there is no successor (see status_glyph)
    "superseded": "↩",
}
 
#: §7 status token → render label.
_STATUS_LABEL = {
    "todo": "Not started",
    "in_progress": "In progress",
    "done": "Complete",
    "blocked": "Blocked",
    "deferred": "Deferred",
    "superseded": "Superseded",
}
 
#: §7 priority token → glyph (4-level; store token is ``lower`` not ``low``).
_PRIORITY_GLYPH = {
    "critical": "🔴",
    "high": "🟠",
    "medium": "🟡",
    "lower": "🟢",
}
 
#: Terminal item statuses (Codex V7 F3) — all three are terminal for the
#: ``current``-pointer derivation (AC §8.35).
TERMINAL_STATUSES = frozenset({"done", "deferred", "superseded"})
 
_HEADER_OPEN = "<!-- GENERATED by oct agent render; do not hand-edit."
_HEADER_CLOSE = "-->"
_HEADER_RE = re.compile(
    # OCT-LINT: disable=no_hardcoded_secrets reason="Table/column names + header tokens are schema constants, not secrets"
    r"<!-- GENERATED by oct agent render; do not hand-edit\.\s*"
    r"source_hash=(?P<source_hash>\S+)\s+"
    r"logical_state_hash=(?P<logical_state_hash>\S+)\s+"
    r"renderer=(?P<renderer>\S+)\s*-->",
)
_IDENTIFIER_RE = re.compile(r"[A-Za-z_][A-Za-z0-9_]*")
 
 
class RenderGuardError(RuntimeError):
    """A render query touched a runtime-excluded table or EXCLUDED column."""
 
 
def status_glyph(status: str, *, has_successor: bool = False) -> str:
    """Return the §7 glyph for ``status``.
 
    ``deferred`` with a named successor renders ``↩`` (redirect); ``deferred``
    with no successor renders ``🚫`` (no successor). ``superseded`` always has a
    ``superseded_by`` and renders ``↩``.
    """
    if status == "deferred":
        return "↩" if has_successor else "🚫"
    return _STATUS_GLYPH.get(status, "⬜")
 
 
def status_label(status: str, *, successor: Optional[str] = None,
                 superseded_by: Optional[str] = None) -> str:
    """Return the §7 render label (with the redirect target when present)."""
    if status == "deferred":
        return f"Deferred → {successor}" if successor else "Deferred (no successor)"
    if status == "superseded" and superseded_by:
        return f"Superseded by {superseded_by}"
    return _STATUS_LABEL.get(status, status)
 
 
def priority_glyph(priority: Optional[str]) -> str:
    """Return the §7 priority glyph; ``—`` for null (not applicable)."""
    if priority is None:
        return "—"
    return _PRIORITY_GLYPH.get(priority, "—")
 
 
def build_generated_header(
    *, source_hash: str, logical_state_hash: str, renderer: str
) -> str:
    """Return the §6 four-line generated header (HTML comment)."""
    if renderer not in RENDERER_IDS:
        raise ValueError(f"unknown renderer id {renderer!r}")
    return (
        f"{_HEADER_OPEN}\n"
        f"source_hash={source_hash}\n"
        f"logical_state_hash={logical_state_hash}\n"
        f"renderer={renderer} {_HEADER_CLOSE}"
    )
 
 
def parse_generated_header(text: str) -> Optional[dict[str, str]]:
    """Parse the §6 header from rendered ``text``; ``None`` if absent/malformed.
 
    A ``None`` result is itself drift (a hand-edit that dropped the header).
    """
    func = "parse_generated_header"
    match = _HEADER_RE.search(text)
    if match is None:
        return None
    fields = match.groupdict()
    _dbg("OC_CHAT.MAIN", func, f"renderer={fields['renderer']}", 3)
    return fields
 
 
def assert_render_query_safe(sql: str) -> None:
    """Raise ``RenderGuardError`` if ``sql`` reads a runtime-excluded table or
    EXCLUDED column (AC §8.30 — the source-only render guard / column-scope guard).
    """
    tokens = {token.lower() for token in _IDENTIFIER_RE.findall(sql)}
    bad_tables = tokens & RUNTIME_EXCLUDED_TABLES
    if bad_tables:
        raise RenderGuardError(
            "source-only render query reads runtime-excluded table(s) "
            f"{sorted(bad_tables)}; a runtime-checkpoint-hash header is required"
        )
    bad_columns = tokens & EXCLUDED_COLUMNS
    if bad_columns:
        raise RenderGuardError(
            "render query references §3.2 EXCLUDED column(s) "
            f"{sorted(bad_columns)}; these differ across rebuilds"
        )
 
 
def guarded_query(
    conn: sqlite3.Connection, sql: str, params: tuple = ()
) -> sqlite3.Cursor:
    """Run a render SELECT through ``assert_render_query_safe`` first (§8.30)."""
    assert_render_query_safe(sql)
    return conn.execute(sql, params)
 
 
def atomic_write_text(path: Path, text: str) -> Path:
    """Atomic-write ``text`` as UTF-8 (LF, no BOM) via the shared §6 writer."""
    path = Path(path)
    atomic_write_bytes(path, text.encode("utf-8"))
    return path
 
 
__all__ = [
    "RENDERER_LEAN",
    "RENDERER_EXTENDED",
    "RENDERER_CHANGELOG",
    "RENDERER_ROOT_CHANGELOG",
    "RENDERER_IDS",
    "LIVING_SET_FILENAMES",
    "generated_footer",
    "RUNTIME_EXCLUDED_TABLES",
    "EXCLUDED_COLUMNS",
    "TERMINAL_STATUSES",
    "RenderGuardError",
    "status_glyph",
    "status_label",
    "priority_glyph",
    "build_generated_header",
    "parse_generated_header",
    "assert_render_query_safe",
    "guarded_query",
    "atomic_write_text",
]