rune/scripts/scaffold_oracle.py

#!/usr/bin/env python3
"""scaffold_oracle.py — freeze/verify wrapper for the scaffold dichotomy
wave (contract BUILDER-2026-10-13T10:42:56Z-179836-factory-gap-sc).

The shared harness (scripts/freeze_truth.py) imports ONE module and replays
JSON-native cases against it. rune.scaffold's clock lives in the module-local
`now` bindings (sc.now from structural_store import; ss.now for the
structural-store internals) — no module-global `datetime` — so the harness's
built-in clock pin can't reach it. This wrapper pins BOTH bindings from
RUNE_FROZEN_NOW (set by the envelope's own frozen_now during --verify), pins
the structural core to the Python oracle (RUNE_STRUCTURAL_CORE=python), and
exposes tiny JSON-native capture helpers:

  * init_case          — Scaffold.init through a relative store path (IO-bearing;
                         the store file is written under the caller's CWD).
  * prd/maintenance/bind/observe/reconcile_case — the mutating Scaffold methods
                         on a doc passed by value; result + mutated doc captured
                         (the harness expected carries the wrapper's return).
  * repo_state_from_git_case — the git subprocess seam REPLACED by a scripted
                         fake (arg): no real IO, byte-deterministic.
  * *_error_case       — the oracle's error texts (unknown placement,
                         unbound repo observation) captured without raising.

The CLI-level operator surface (project-init/project-show/contract-*) and the
nested argv helpers (_scaffold_from_args/_cmd_args) are covered by the live
parity sweep + Go arg-convention tests instead (nested functions are not
module attributes the harness can resolve).
"""
import os
import sys
from pathlib import Path

REPO = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(REPO / "src"))

os.environ["RUNE_STRUCTURAL_CORE"] = "python"  # deterministic freeze/verify mode

FROZEN = os.environ.get("RUNE_FROZEN_NOW", "")
if FROZEN:
    import rune.scaffold as sc
    import rune.structural_store as ss

    sc.now = lambda: FROZEN
    ss.now = lambda: FROZEN

from rune import scaffold as sc  # noqa: E402  (Scaffold re-export for harness recv)
import json  # noqa: E402


class _ShellResult:
    def __init__(self, stdout, returncode):
        self.stdout = stdout
        self.returncode = returncode
        self.stderr = ""


class _FakeGit:
    """subprocess module replacement scripted per call ({"stdout",
    "returncode", "raise"} — "timeout"/"oserror" raise exactly the branches
    the oracle catches). run() is the attribute scaffold.py uses; the
    TimeoutExpired attribute must exist too — the oracle's except clause
    references subprocess.TimeoutExpired THROUGH the patched binding."""

    import subprocess as _sp  # noqa: E402

    TimeoutExpired = _sp.TimeoutExpired

    def __init__(self, script):
        self.steps = list(script)

    def run(self, *args, **kwargs):
        spec = self.steps.pop(0)
        if spec.get("raise") == "timeout":
            raise self._sp.TimeoutExpired(args[0] if args else "git", 15)
        if spec.get("raise") == "oserror":
            raise OSError("git unavailable")
        return _ShellResult(spec.get("stdout", ""), spec.get("returncode", 0))


def _fresh_doc():
    return {
        "schema_version": 1,
        "generated_at": FROZEN,
        "entities": [],
        "relationships": [],
        "observations": [],
    }


def init_case(store_dir, project_id, purpose="truth battery", requirement=""):
    """Scaffold.init on a relative store path (writes <store_dir>/current.json
    under the caller's CWD — the verify run happens in a scratch CWD)."""
    s = sc.Scaffold(store_dir, project_id)
    doc = s.init(purpose=purpose, requirement=requirement)
    return {"result": doc}


def set_prd_case(doc, project_id, prd_ref):
    s = sc.Scaffold(".", project_id)
    e = s.set_prd(doc, prd_ref)
    return {"result": e, "doc": doc}


def add_maintenance_item_case(doc, name, **kw):
    """kwargs pass through to the oracle — its signature defaults
    (approval=user, priority=normal, status=open) stay the authority."""
    s = sc.Scaffold(".", "gap-project")
    try:
        e = s.add_maintenance_item(doc, name, **kw)
    except TypeError as ex:
        return {"result": None, "error": str(ex), "doc": doc}
    return {"result": e, "doc": doc}


def bind_repo_case(doc, **kw):
    """kwargs pass through to the oracle — its signature defaults
    (placement=EXISTING_PROJECT_REPO, worktree_strategy=per-contract) stay
    the authority."""
    s = sc.Scaffold(".", "gap-project")
    try:
        e = s.bind_repo(doc, **kw)
    except ValueError as ex:
        return {"result": None, "error": str(ex), "doc": doc}
    return {"result": e, "doc": doc}


def observe_repo_case(doc, owner, name, **kw):
    s = sc.Scaffold(".", "gap-project")
    try:
        e = s.observe_repo(doc, owner, name, **kw)
    except ValueError as ex:
        return {"result": None, "error": str(ex), "doc": doc}
    return {"result": e, "doc": doc}


def repo_state_from_git_case(doc, owner, name, repo_root, script):
    s = sc.Scaffold(".", "gap-project")
    sc.subprocess = _FakeGit(script)
    try:
        e = s.repo_state_from_git(doc, owner, name, repo_root)
    except ValueError as ex:
        return {"result": None, "error": str(ex), "doc": doc}
    return {"result": e, "doc": doc}


def reconcile_case(doc):
    s = sc.Scaffold(".", "gap-project")
    out = s.reconcile(doc)
    return {"result": out, "doc": doc}