scanner.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/registry/source/scanner.py
 
# OCT-LINT: disable=tests_exist reason="Covered by oc_chat/tests/registry/source/test_scanner.py"
"""
Closed-world scanner for the AuthoredSource tree.
 
Purpose
-------
Enumerate ``oc_chat/roadmap/source/**`` and classify every file against the
Registry Source & Storage Spec V0.22 §3.1 allowlist, hard-failing any unknown
path and enforcing the ``phases/`` path-shape rule, while silently skipping a
hardcoded OS/editor-noise set independent of ``.gitignore``.
 
Responsibilities
----------------
- Walk the tree fail-closed (``SourceScanError`` on unreadable trees).
- Classify each recognised file into a typed ``ScannedRecord`` (entity kind +
  resolution metadata: ``track_kind``, promoted ``phase_code``, changelog kind).
- Enforce the two legal ``phases/`` shapes (depth-1 phase, depth-2 promoted
  item under an existing ``phases/<code>.md``); reject every other shape.
- Skip ``.DS_Store``/``Thumbs.db``/``*.swp``/``*~``/``*.tmp`` silently.
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 - scan lifecycle
    L3 - per-file classification
    L4 - phases path-shape resolution
 
Contracts
---------
- Read-only: never mutates the source tree.
- Output is deterministic — records sorted by POSIX relative path.
- Unknown/illegal paths raise ``SourceValidationError`` with the offending path;
  unreadable trees raise ``SourceScanError`` (``raw_source_tree_hash`` → null).
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- oc_chat.registry.source.errors (SRC_* codes + exception types)
"""
from __future__ import annotations
 
from dataclasses import dataclass, field
import fnmatch
import os
from pathlib import Path
from typing import Optional
 
from oc_diagnostics import _dbg
 
from oc_chat.registry.source.errors import (
    SRC_BAD_PHASE_PATH,
    SRC_DUP_OR_STEM_MISMATCH,
    SRC_UNKNOWN_FILE,
    SourceScanError,
    SourceValidationError,
)
 
_dbg("OC_CHAT.MAIN", "module", "loaded", 4)
 
# Hardcoded OS/editor-noise set (§3.1): skipped silently, independent of
# ``.gitignore``. Exact names + glob patterns.
NOISE_EXACT: frozenset[str] = frozenset({".DS_Store", "Thumbs.db"})
NOISE_GLOBS: tuple[str, ...] = ("*.swp", "*~", "*.tmp")
 
# The non-phase ``items/<track_kind>/`` subdirectories (§3.1).
ITEM_TRACK_KINDS: frozenset[str] = frozenset(
    {"backlog", "vision", "git_workflow", "pre_m0_era"}
)
# The ``changelog/<kind>/`` subdirectories (§3.1).
CHANGELOG_KINDS: frozenset[str] = frozenset({"roadmap", "product"})
 
 
@dataclass(frozen=True)
class ScannedRecord:
    """A recognised source file + the resolution metadata the loader needs."""
 
    kind: str  # manifest | era | track | phase | promoted_item | roadmap_item | version | framing | note
    path: Path
    rel_path: str  # POSIX relative path from the source root
    stem: str  # filename stem (validated == frontmatter id by the loader)
    track_kind: Optional[str] = None  # for roadmap_item / promoted_item
    phase_code: Optional[str] = None  # for promoted_item (immediate parent dir)
    changelog_kind: Optional[str] = None  # for version (routing subdir)
    extra: dict = field(default_factory=dict)
 
 
def _is_noise(name: str) -> bool:
    """True if ``name`` is in the hardcoded OS/editor-noise set (§3.1)."""
    if name in NOISE_EXACT:
        return True
    return any(fnmatch.fnmatch(name, pattern) for pattern in NOISE_GLOBS)
 
 
def _walk_files(root: Path, temp_dir_names: frozenset[str]) -> list[Path]:
    """Return all non-noise files under ``root`` (fail-closed on read errors)."""
 
    def _on_error(error: OSError) -> None:
        """Convert an ``os.walk`` read error into a fail-closed scan error."""
        raise SourceScanError(
            f"cannot enumerate source tree at {getattr(error, 'filename', root)!r}",
            path=str(getattr(error, "filename", root) or root),
            os_error=str(error),
            hint="check directory permissions and symlink targets under source/",
        )
 
    if not root.exists():
        raise SourceScanError(
            f"source tree does not exist: {root}",
            path=str(root),
            os_error="FileNotFoundError",
            hint="create oc_chat/roadmap/source/ or pass the correct root",
        )
    files: list[Path] = []
    for dirpath, dirnames, filenames in os.walk(root, onerror=_on_error):
        dirnames[:] = [d for d in dirnames if d not in temp_dir_names]
        for filename in filenames:
            if _is_noise(filename):
                continue
            files.append(Path(dirpath) / filename)
    return files
 
 
def _classify(root: Path, path: Path, phase_stems: frozenset[str]) -> ScannedRecord:
    """Classify one file against the §3.1 allowlist + ``phases/`` path-shape rule."""
    parts = path.relative_to(root).parts
    rel = "/".join(parts)
    name = parts[-1]
    stem = name[:-3] if name.endswith(".md") else name
 
    def unknown() -> SourceValidationError:
        """Build the SRC_UNKNOWN_FILE error for this path."""
        return SourceValidationError(
            f"unknown file under source/: {rel}",
            code=SRC_UNKNOWN_FILE,
            records=[rel],
            hint="only the §3.1 allowlist paths are recognised under source/",
        )
 
    if not name.endswith(".md"):
        raise unknown()
 
    if len(parts) == 1:
        if name == "manifest.md":
            return ScannedRecord(kind="manifest", path=path, rel_path=rel, stem="manifest")
        raise unknown()
 
    top = parts[0]
    if top == "eras" and len(parts) == 2:
        return ScannedRecord(kind="era", path=path, rel_path=rel, stem=stem)
    if top == "tracks" and len(parts) == 2:
        return ScannedRecord(kind="track", path=path, rel_path=rel, stem=stem)
    if top == "notes" and len(parts) == 2:
        # Plan 142j R7: only ``notes/<id>.md`` is legal (§3.1); a deeper
        # ``notes/<sub>/<id>.md`` falls through to SRC_UNKNOWN_FILE (mirrors the
        # len(parts) guards on tracks/items).
        return ScannedRecord(kind="note", path=path, rel_path=rel, stem=stem)
    if top == "items" and len(parts) == 3 and parts[1] in ITEM_TRACK_KINDS:
        return ScannedRecord(
            kind="roadmap_item", path=path, rel_path=rel, stem=stem, track_kind=parts[1]
        )
    if top == "changelog" and len(parts) == 3 and parts[1] in CHANGELOG_KINDS:
        return ScannedRecord(
            kind="version", path=path, rel_path=rel, stem=stem, changelog_kind=parts[1]
        )
    if top == "framing" and len(parts) == 2:
        # [NEW·V24] content-bearing framing (labeled divider / section header),
        # §3.1 allowlist ``framing/<id>.md``; stem == frontmatter id (loader).
        return ScannedRecord(kind="framing", path=path, rel_path=rel, stem=stem)
    if top == "phases":
        if len(parts) == 2:
            return ScannedRecord(kind="phase", path=path, rel_path=rel, stem=stem)
        if len(parts) == 3:
            phase_code = parts[1]
            if phase_code not in phase_stems:
                raise SourceValidationError(
                    f"promoted item {rel} has no parent phases/{phase_code}.md",
                    code=SRC_BAD_PHASE_PATH,
                    records=[rel],
                    hint="a promoted sub-item requires an existing phases/<code>.md",
                )
            return ScannedRecord(
                kind="promoted_item",
                path=path,
                rel_path=rel,
                stem=stem,
                phase_code=phase_code,
                track_kind="m_phase",
            )
        raise SourceValidationError(
            f"illegal phases/ path shape (too deep): {rel}",
            code=SRC_BAD_PHASE_PATH,
            records=[rel],
            hint="phases/ admits only <id>.md or <code>/<sub_id>.md",
        )
    raise unknown()
 
 
#: The literal allowlist-instruction file excluded from the casefold stem index
#: (DeepSeek-R12-NP2-2) — it is not an entity record.
_CASEFOLD_EXCLUDE = "README.md"
 
 
def _check_casefold_stems(root: Path, files: list[Path]) -> None:
    """Plan 142j R7 (VC-0007, layer 1): reject a **per-directory** casefold stem
    collision over every discovered file BEFORE loading records (§3.2).
 
    ``canonical.load_record`` is per-file and cannot see cross-file collisions;
    only the scanner walks directories, so the physical stem-vs-stem check is
    owned here. A case-only-different pair (e.g. ``Foo.md``/``foo.md`` in one dir)
    breaks every case-insensitive NTFS/APFS checkout ⇒ ``SRC_DUP_OR_STEM_MISMATCH``.
    The literal ``README.md`` is excluded (not an entity record).
    """
    by_dir: dict[str, dict[str, list[str]]] = {}
    for path in files:
        if path.name == _CASEFOLD_EXCLUDE:
            continue
        rel = path.relative_to(root)
        parent = rel.parent.as_posix()
        stem = path.name[:-3] if path.name.endswith(".md") else path.name
        by_dir.setdefault(parent, {}).setdefault(stem.casefold(), []).append(rel.as_posix())
    for _parent, stems in sorted(by_dir.items()):
        for _key, paths in sorted(stems.items()):
            if len(paths) > 1:
                raise SourceValidationError(
                    f"case-insensitive stem collision in one directory: {sorted(paths)}",
                    code=SRC_DUP_OR_STEM_MISMATCH,
                    records=sorted(paths),
                    hint="ids/slugs are unique case-insensitively per directory (§3.2); "
                    "rename so no two files share a stem modulo case",
                )
 
 
def observed_files(
    root: Path, *, temp_dir_names: frozenset[str] = frozenset()
) -> list[Path]:
    """All physically-observed non-noise files under ``root`` (sorted).
 
    Used by the raw-source-tree-hash preimage (§10), which includes files that
    later fail closed-world validation — so it walks independently of
    classification and is computable before any parse.
    """
 
    root = Path(root)
    return sorted(
        _walk_files(root, temp_dir_names), key=lambda p: p.relative_to(root).parts
    )
 
 
def scan_source_tree(
    root: Path, *, temp_dir_names: frozenset[str] = frozenset()
) -> list[ScannedRecord]:
    """Scan ``root`` and return the classified records (sorted by POSIX path).
 
    Raises ``SourceScanError`` if the tree cannot be enumerated, or
    ``SourceValidationError`` (``SRC_UNKNOWN_FILE`` / ``SRC_BAD_PHASE_PATH``) on
    the first offending path in sorted order.
    """
 
    func = "scan_source_tree"
    root = Path(root)
    files = sorted(_walk_files(root, temp_dir_names), key=lambda p: p.relative_to(root).parts)
    _check_casefold_stems(root, files)  # R7 layer 1: per-directory casefold uniqueness
    phase_stems = frozenset(
        path.stem
        for path in files
        if path.name.endswith(".md") and path.relative_to(root).parts[:1] == ("phases",)
        and len(path.relative_to(root).parts) == 2
    )
    records = [_classify(root, path, phase_stems) for path in files]
    _dbg("OC_CHAT.MAIN", func, f"classified {len(records)} records", 3)
    return records