seed_142_fix_dependency_directions.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# tools/seed_142_fix_dependency_directions.py
 
# OCT-LINT: disable=tests_exist reason="Covered by oc_chat/tests/registry/seed/test_dependency_directions.py"
# OCT-LINT: disable=no_hardcoded_secrets reason="Roadmap item ids + edge kinds are not secrets"
"""
Dependency-direction corrections (Plan 142 prose-preservation Phase 4, GATE4 Part A).
 
Purpose
-------
The proper render exposed INVERTED ``depends_on`` edges in the P1.7-ingested seed
(a dependent "blocked by" a *later / not-started* predecessor — chronologically
impossible). This tool FIXES the clearly-inverted edges (each evidence-backed,
see ``tools/142_dependency_direction_audit.md``); the genuinely-ambiguous ones are
left untouched and FLAGGED in the audit for an operator decision.
 
The final, operator-confirmed edge set (GATE4 direction fixes + the GATE4 ADDENDUM
cleanup bundle). ``FIXES`` re-points ``depends_on``; ``STATUS_FIXES`` corrects a
stale status:
- ``M2.depends_on -> [M1]``     (was [M3,M7.4] inverted; + legacy "M2 Depends on: M1").
- ``M3.depends_on -> [M2]`` RETIRED at Plan 151 T0 (operator M2->M3 gate waiver).
- ``M7.depends_on -> [M2, M5]`` (legacy "M7 Depends on: M2 + M5"; M7.4 nested -> phase-level).
- ``G.4a.depends_on -> []`` + ``G.5a.depends_on -> [G.4a]`` (inverted vs G.5a's prose).
- ``B.depends_on -> []`` · ``C.depends_on -> []`` · ``M11-GATE.depends_on -> []``
  (3 spurious chronologically-impossible edges removed).
- ``M11-GATE.status blocked -> done`` (legacy shows M11-GATE ✅ done at Round 24).
 
Edge ``kind`` stays ``gates`` (``gates``/``blocks`` render identically as a 🔒
blocker). No item is added or removed (fidelity preserved); only ``depends_on`` +
the one status are re-pointed. Deterministic + idempotent.
 
Responsibilities
----------------
- ``apply(source_dir)`` — load each affected record, re-point ``depends_on``,
  re-emit canonically; ``FIXES`` is the declarative edit set.
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 - fix lifecycle
    L3 - per-record edit
    L4 - per-edge detail
 
Contracts
---------
- Deterministic + idempotent; writes only the 5 affected ``source`` records;
  never touches the live legacy roadmap; every record re-validates + canonicalises.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- oc_chat.registry.regenerate (imported for the contracts bootstrap it runs)
- option_c_registry_contracts (DependencyEdge - the corrected depends_on edges)
- oc_chat.registry.source.canonical (load_record and format_record_text - reads each affected record and re-emits it canonically)
- oc_chat.registry.source.scanner (scan_source_tree - finds the phase and roadmap item records by stem)
"""
from __future__ import annotations
 
import sys
from dataclasses import dataclass, field
from pathlib import Path
 
# Force the WORKTREE-local oc_chat ahead of the editable main-repo install.
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
 
from oc_diagnostics import _dbg
 
import oc_chat.registry.regenerate  # noqa: F401  (contracts bootstrap)
import option_c_registry_contracts as rc
 
from oc_chat.registry.source.canonical import format_record_text, load_record
from oc_chat.registry.source.scanner import scan_source_tree
 
_dbg("OC_CHAT.MAIN", "module", "loaded", 4)
 
#: record-id -> the corrected ``depends_on`` predecessor ids (kind ``gates``).
#: The final, operator-confirmed edge set (GATE4 + the GATE4 ADDENDUM bundle).
FIXES: dict[str, list[str]] = {
    "M2": ["M1"],        # was [M3, M7.4] inverted; + legacy "M2 Depends on: M1" (addendum)
    # "M3": ["M2"] RETIRED (Plan 151 T0, 2026-07-16): the operator-ratified
    # M2->M3 gate WAIVER (151 dispatch pin, branch A; waiver record
    # 2026-07-16-151-t0-m2-m3-gate-waiver) removed M3's depends_on gate so the
    # migration could start with M2 non-terminal. Re-adding the edge here made
    # this one-shot fixer NON-idempotent against the committed seed (it
    # re-locked M3 on every run — caught at 151 T2 when the suite mutated the
    # working tree). The 142-era evidence for the other edges is unchanged.
    "M7": ["M2", "M5"],  # legacy "M7 Depends on: M2 + M5" (M7.4 nested -> phase-level; +M5 addendum)
    "G.4a": [],          # was [G.5a] — inverted vs G.5a's own prose
    "G.5a": ["G.4a"],    # G.5a body: "Depends on G.4a cleanup"
    "B": [],             # drop spurious B<-M11 (chronologically impossible, no prose link)
    "C": [],             # drop spurious C<-G.5a + C<-M11 (same)
    "M11-GATE": [],      # drop spurious M11-GATE<-M11 (it ran mid-M11, never blocked-on-complete-M11)
}
 
#: record-id -> corrected ``status`` (the §9-3b-style status correction).
STATUS_FIXES: dict[str, str] = {
    # Legacy Git-Workflow table shows M11-GATE ✅ done at Round 24 (merge 970d48a);
    # the seed's "blocked" was the stale-status artifact behind the spurious M11 edge.
    "M11-GATE": "done",
}
 
 
@dataclass
class FixReport:
    records_changed: int = 0
    edits: list = field(default_factory=list)  # (record_id, change_description)
 
 
def _edges(ids: list[str]) -> list[rc.DependencyEdge]:
    return [rc.DependencyEdge(id=item_id, kind="gates") for item_id in ids]
 
 
def _before(model: object) -> list[str]:
    out: list[str] = []
    for edge in getattr(model, "depends_on", []) or []:
        out.append(edge.id if isinstance(edge, rc.DependencyEdge) else str(edge))
    return out
 
 
def apply(source_dir: Path) -> FixReport:
    """Apply the declarative ``depends_on`` + ``status`` fixes to the source records."""
    func = "apply"
    report = FixReport()
    by_stem = {sc.stem: sc for sc in scan_source_tree(Path(source_dir)) if sc.kind in ("phase", "roadmap_item")}
    record_ids = list(dict.fromkeys([*FIXES, *STATUS_FIXES]))  # union, stable order
    for record_id in record_ids:
        scanned = by_stem.get(record_id)
        if scanned is None:
            _dbg("OC_CHAT.MAIN", func, f"record {record_id} not found — skipped", 2)
            continue
        model, body = load_record(scanned)
        update: dict[str, object] = {}
        change: list[str] = []
        if record_id in FIXES:
            before, after = _before(model), list(FIXES[record_id])
            if before != after:
                update["depends_on"] = _edges(after)
                change.append(f"depends_on {before} -> {after}")
        if record_id in STATUS_FIXES:
            before_status = getattr(model, "status", None)
            if before_status != STATUS_FIXES[record_id]:
                update["status"] = STATUS_FIXES[record_id]
                change.append(f"status {before_status} -> {STATUS_FIXES[record_id]}")
        if not update:
            continue
        updated = model.model_copy(update=update)
        scanned.path.write_text(
            format_record_text(updated, body=body), encoding="utf-8", newline=""
        )
        report.records_changed += 1
        report.edits.append((record_id, "; ".join(change)))
        _dbg("OC_CHAT.MAIN", func, f"{record_id}: {'; '.join(change)}", 3)
    _dbg("OC_CHAT.MAIN", func, f"changed {report.records_changed} records", 2)
    return report
 
 
def _main() -> int:
    func = "_main"
    repo = Path(__file__).resolve().parents[1]
    report = apply(repo / "oc_chat" / "roadmap" / "source")
    _dbg("OC_CHAT.MAIN", func, f"records_changed={report.records_changed}", 2)
    for record_id, change in report.edits:
        print(f"{record_id}: {change}")
    print(f"records_changed={report.records_changed}")
    return 0
 
 
if __name__ == "__main__":
    raise SystemExit(_main())