extract.py
file-level
1
files
1
commits
0
hotspots
0
🧊 dead
0
💥 blast risk
| 1 | """Extract and format the paste-ready fence for ``ok next`` (§ONS.5 / §NXP.3).""" |
| 2 | |
| 3 | from __future__ import annotations |
| 4 | |
| 5 | from collections.abc import Callable |
| 6 | from dataclasses import dataclass |
| 7 | from datetime import datetime, timezone |
| 8 | from pathlib import Path |
| 9 | |
| 10 | from tools.governance_hygiene.next_regen import extract_paste_fence_body |
| 11 | |
| 12 | |
| 13 | # Exact heading — one space before the Unicode em dash (§ONS.5.3). |
| 14 | CURRENT_NEXT_HEADING = "## CURRENT NEXT — paste this" |
| 15 | |
| 16 | # Provenance line template — separator is space, U+00B7, space (§NXP.3.2). |
| 17 | PROVENANCE_LINE_TEMPLATE = ( |
| 18 | "**Source:** `{repo_name}` · `{repo_root_abs}` · `{doc_rel}` · lane `{lane}` · read `{read_at}`" |
| 19 | ) |
| 20 | PROVENANCE_SEPARATOR = " · " |
| 21 | |
| 22 | PASTE_HEADING = "### Paste-ready prompt" |
| 23 | |
| 24 | REASON_HANDOVER_MISSING = "handover_missing" |
| 25 | REASON_HANDOVER_UNREADABLE = "handover_unreadable" |
| 26 | REASON_HEADING_MISSING = "heading_missing" |
| 27 | REASON_FENCE_MISSING = "fence_missing" |
| 28 | REASON_FENCE_EMPTY = "fence_empty" |
| 29 | REASON_MODEL_MISSING = "model_missing" |
| 30 | REASON_REPO_ROOT_UNRESOLVED = "repo_root_unresolved" |
| 31 | |
| 32 | FAIL_CLOSED_REASONS = frozenset( |
| 33 | { |
| 34 | REASON_HANDOVER_MISSING, |
| 35 | REASON_HANDOVER_UNREADABLE, |
| 36 | REASON_HEADING_MISSING, |
| 37 | REASON_FENCE_MISSING, |
| 38 | REASON_FENCE_EMPTY, |
| 39 | REASON_MODEL_MISSING, |
| 40 | REASON_REPO_ROOT_UNRESOLVED, |
| 41 | } |
| 42 | ) |
| 43 | |
| 44 | # Injectable clock seam (§NXP.3.4) — production uses real UTC; tests pin. |
| 45 | _Clock = Callable[[], str] |
| 46 | _clock: _Clock | None = None |
| 47 | |
| 48 | |
| 49 | def utc_read_at() -> str: |
| 50 | """Return UTC ISO-8601 ``YYYY-MM-DDTHH:MM:SSZ`` at second precision.""" |
| 51 | return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") |
| 52 | |
| 53 | |
| 54 | def set_read_at_clock(clock: _Clock | None) -> None: |
| 55 | """Install or clear the injectable ``read_at`` clock (§NXP.3.4).""" |
| 56 | global _clock |
| 57 | _clock = clock |
| 58 | |
| 59 | |
| 60 | def read_at_now() -> str: |
| 61 | """Return ``read_at`` via the injected clock, else the real UTC clock.""" |
| 62 | if _clock is not None: |
| 63 | return _clock() |
| 64 | return utc_read_at() |
| 65 | |
| 66 | |
| 67 | def absolute_repo_root(repo_root: Path) -> str | None: |
| 68 | """Return absolute POSIX repo root, or ``None`` when unresolved (§NXP.3.6).""" |
| 69 | try: |
| 70 | resolved = repo_root.resolve() |
| 71 | except (OSError, RuntimeError): |
| 72 | return None |
| 73 | if not resolved.is_absolute(): |
| 74 | return None |
| 75 | return resolved.as_posix() |
| 76 | |
| 77 | |
| 78 | @dataclass(frozen=True) |
| 79 | class CurrentNextResult: |
| 80 | """Successful extract of the paste-ready fence body.""" |
| 81 | |
| 82 | path: str |
| 83 | lane: str | None |
| 84 | fence: str |
| 85 | heading: str = CURRENT_NEXT_HEADING |
| 86 | |
| 87 | |
| 88 | @dataclass(frozen=True) |
| 89 | class CurrentNextError: |
| 90 | """Fail-closed extract outcome (§ONS.5.7 / §NXP.3.6).""" |
| 91 | |
| 92 | reason: str |
| 93 | detail: str |
| 94 | path: str | None = None |
| 95 | lane: str | None = None |
| 96 | |
| 97 | @property |
| 98 | def message(self) -> str: |
| 99 | return f"next: {self.reason} — {self.detail}" |
| 100 | |
| 101 | |
| 102 | def format_provenance_line( |
| 103 | *, |
| 104 | repo_name: str | None, |
| 105 | repo_root_abs: str, |
| 106 | doc_rel: str, |
| 107 | lane: str | None, |
| 108 | read_at: str, |
| 109 | ) -> str: |
| 110 | """Render the single provenance line (§NXP.3.2).""" |
| 111 | name = (repo_name or "").strip() or "unknown" |
| 112 | lane_label = lane if lane else "-" |
| 113 | return PROVENANCE_LINE_TEMPLATE.format( |
| 114 | repo_name=name, |
| 115 | repo_root_abs=repo_root_abs, |
| 116 | doc_rel=doc_rel, |
| 117 | lane=lane_label, |
| 118 | read_at=read_at, |
| 119 | ) |
| 120 | |
| 121 | |
| 122 | def extract_current_next( |
| 123 | handover_path: Path, |
| 124 | *, |
| 125 | repo_relative_path: str, |
| 126 | lane: str | None, |
| 127 | ) -> CurrentNextResult | CurrentNextError: |
| 128 | """Read ``handover_path`` and return the paste fence or a closed reason. |
| 129 | |
| 130 | Order of fail-closed reasons is frozen (§ONS.5.7). Does not invent a fence |
| 131 | from roadmap, chat memory, or planned regen bytes. |
| 132 | """ |
| 133 | if not handover_path.exists(): |
| 134 | return CurrentNextError( |
| 135 | reason=REASON_HANDOVER_MISSING, |
| 136 | detail=f"handover not found: {repo_relative_path}", |
| 137 | path=repo_relative_path, |
| 138 | lane=lane, |
| 139 | ) |
| 140 | |
| 141 | try: |
| 142 | text = handover_path.read_text(encoding="utf-8") |
| 143 | except (OSError, UnicodeDecodeError) as exc: |
| 144 | return CurrentNextError( |
| 145 | reason=REASON_HANDOVER_UNREADABLE, |
| 146 | detail=f"cannot read handover as UTF-8: {exc}", |
| 147 | path=repo_relative_path, |
| 148 | lane=lane, |
| 149 | ) |
| 150 | |
| 151 | # Heading check before fence extract so a stray ``` elsewhere cannot win. |
| 152 | if PASTE_HEADING not in text: |
| 153 | return CurrentNextError( |
| 154 | reason=REASON_HEADING_MISSING, |
| 155 | detail="missing '### Paste-ready prompt' heading (KH1 H7)", |
| 156 | path=repo_relative_path, |
| 157 | lane=lane, |
| 158 | ) |
| 159 | |
| 160 | body = extract_paste_fence_body(text) |
| 161 | if body is None: |
| 162 | return CurrentNextError( |
| 163 | reason=REASON_FENCE_MISSING, |
| 164 | detail="paste-ready heading present but no fenced body", |
| 165 | path=repo_relative_path, |
| 166 | lane=lane, |
| 167 | ) |
| 168 | |
| 169 | if not body.strip(): |
| 170 | return CurrentNextError( |
| 171 | reason=REASON_FENCE_EMPTY, |
| 172 | detail="paste-ready fence body is empty", |
| 173 | path=repo_relative_path, |
| 174 | lane=lane, |
| 175 | ) |
| 176 | |
| 177 | if "Model:" not in body: |
| 178 | return CurrentNextError( |
| 179 | reason=REASON_MODEL_MISSING, |
| 180 | detail="paste-ready fence body lacks 'Model:' (KH1 H8)", |
| 181 | path=repo_relative_path, |
| 182 | lane=lane, |
| 183 | ) |
| 184 | |
| 185 | return CurrentNextResult( |
| 186 | path=repo_relative_path, |
| 187 | lane=lane, |
| 188 | fence=body, |
| 189 | heading=CURRENT_NEXT_HEADING, |
| 190 | ) |
| 191 | |
| 192 | |
| 193 | def format_current_next( |
| 194 | result: CurrentNextResult, |
| 195 | *, |
| 196 | repo_name: str | None, |
| 197 | repo_root_abs: str, |
| 198 | read_at: str, |
| 199 | ) -> str: |
| 200 | """Return human stdout bytes for a successful extract (§NXP.3.1). |
| 201 | |
| 202 | Twelve-step layout (narrow supersede of §ONS.5.4). Trailing newline on the |
| 203 | last line is included. Fence body bytes are unchanged. |
| 204 | """ |
| 205 | body = result.fence |
| 206 | if not body.endswith("\n"): |
| 207 | body = body + "\n" |
| 208 | provenance = format_provenance_line( |
| 209 | repo_name=repo_name, |
| 210 | repo_root_abs=repo_root_abs, |
| 211 | doc_rel=result.path, |
| 212 | lane=result.lane, |
| 213 | read_at=read_at, |
| 214 | ) |
| 215 | return f"{CURRENT_NEXT_HEADING}\n\n{provenance}\n\n```text\n{body}```\n" |