278 lines
14 KiB
Python
278 lines
14 KiB
Python
#!/usr/bin/env python3
|
|
# #region Tooling.EvidenceLinkMigration [C:4] [TYPE Module] [SEMANTICS evidence,migration,git,closed-spans]
|
|
# @BRIEF Apply externally pinned, curated historical link units after validating the entire committed source set.
|
|
# @PRE The exact mapping digest, source commit and index are explicitly supplied; every replacement was reviewed.
|
|
# @POST Check mode writes nothing; apply changes only validated Markdown units, with no path discovery or fallback.
|
|
# @RATIONALE Whole command and backtick units preserve Markdown semantics without inserting links inside executable code.
|
|
# @REJECTED Regex discovery, stale-span adaptation and nearest-directory guesses would silently change historical meaning.
|
|
import argparse
|
|
from collections import Counter
|
|
import hashlib
|
|
import json
|
|
import os
|
|
from pathlib import Path, PurePosixPath
|
|
import posixpath
|
|
import re
|
|
import subprocess
|
|
import tempfile
|
|
from urllib.parse import unquote, urlsplit
|
|
|
|
INDEX = "specs/050-mcp-interface/evidence-index.md"
|
|
ACTIONS = frozenset({"rewrite_link_destination_only",
|
|
"replace_whole_backticked_reference_with_index_link",
|
|
"replace_whole_code_unit_with_index_link_outside_code"})
|
|
|
|
|
|
# #region Tooling.EvidenceLinkMigration.Refusal [C:1] [TYPE Class]
|
|
class MigrationRefusal(ValueError):
|
|
def __init__(self, code):
|
|
self.code = code
|
|
super().__init__(code)
|
|
# #endregion Tooling.EvidenceLinkMigration.Refusal
|
|
|
|
|
|
# #region Tooling.EvidenceLinkMigration.Digest [C:1] [TYPE Function]
|
|
def digest(value):
|
|
return hashlib.sha256(value).hexdigest()
|
|
# #endregion Tooling.EvidenceLinkMigration.Digest
|
|
|
|
|
|
# #region Tooling.EvidenceLinkMigration.Git [C:2] [TYPE Function]
|
|
def git(repo, *args):
|
|
result = subprocess.run(["git", "-C", str(repo), *args], capture_output=True)
|
|
if result.returncode:
|
|
raise MigrationRefusal("EVIDENCE_MIGRATION_GIT_UNAVAILABLE")
|
|
return result.stdout
|
|
# #endregion Tooling.EvidenceLinkMigration.Git
|
|
|
|
|
|
# #region Tooling.EvidenceLinkMigration.Source [C:3] [TYPE Function]
|
|
# @POST Only an unchanged committed regular Markdown file inside this exact worktree is read.
|
|
def source(repo, commit, path, expected):
|
|
name = PurePosixPath(path)
|
|
if name.is_absolute() or ".." in name.parts or name.suffix != ".md" or str(name) != path:
|
|
raise MigrationRefusal("EVIDENCE_MIGRATION_PATH_INVALID")
|
|
mode = git(repo, "ls-tree", commit, "--", path).split(b" ", 1)[0]
|
|
if mode != b"100644":
|
|
raise MigrationRefusal("EVIDENCE_MIGRATION_SOURCE_MODE_INVALID")
|
|
target = repo / path
|
|
if target.resolve() != target.absolute() or not target.resolve().is_relative_to(repo.resolve()):
|
|
raise MigrationRefusal("EVIDENCE_MIGRATION_PATH_INVALID")
|
|
committed = git(repo, "show", f"{commit}:{path}")
|
|
if digest(committed) != expected or target.read_bytes() != committed:
|
|
raise MigrationRefusal("EVIDENCE_MIGRATION_SOURCE_CONFLICT")
|
|
return committed.decode("utf-8")
|
|
# #endregion Tooling.EvidenceLinkMigration.Source
|
|
|
|
|
|
# #region Tooling.EvidenceLinkMigration.Span [C:3] [TYPE Function]
|
|
def span(text, start, end, expected):
|
|
if type(start) is not int or type(end) is not int or not 0 <= start < end <= len(text):
|
|
raise MigrationRefusal("EVIDENCE_MIGRATION_SPAN_INVALID")
|
|
selected = text[start:end]
|
|
if digest(selected.encode()) != expected:
|
|
raise MigrationRefusal("EVIDENCE_MIGRATION_SPAN_CONFLICT")
|
|
return selected
|
|
# #endregion Tooling.EvidenceLinkMigration.Span
|
|
|
|
|
|
# #region Tooling.EvidenceLinkMigration.Unit [C:4] [TYPE Function]
|
|
def validate_unit(unit, text, anchors, tag_commit):
|
|
action, replacement = unit["action"], unit["replacement"]
|
|
original = span(text, unit["span_start"], unit["span_end"], unit["original_sha256"])
|
|
referenced = unit["reference_anchors"]
|
|
if (action not in ACTIONS or not isinstance(replacement, str) or not replacement
|
|
or not isinstance(referenced, list) or not referenced
|
|
or any(a not in anchors or f"#{a}" not in replacement for a in referenced)):
|
|
raise MigrationRefusal("EVIDENCE_MIGRATION_REPLACEMENT_INVALID")
|
|
if action == "rewrite_link_destination_only":
|
|
expected = os.path.relpath(INDEX, str(PurePosixPath(unit["doc"]).parent)) + "#" + referenced[0]
|
|
if len(referenced) != 1 or replacement != expected:
|
|
raise MigrationRefusal("EVIDENCE_MIGRATION_REPLACEMENT_INVALID")
|
|
elif action == "replace_whole_backticked_reference_with_index_link":
|
|
destination = os.path.relpath(INDEX, str(PurePosixPath(unit["doc"]).parent)) + "#" + referenced[0]
|
|
if (len(referenced) != 1 or not original.startswith("`") or not original.endswith("`")
|
|
or replacement != f"[{original}]({destination})"):
|
|
raise MigrationRefusal("EVIDENCE_MIGRATION_REPLACEMENT_INVALID")
|
|
else:
|
|
validate_command(unit, replacement, tag_commit)
|
|
keys = []
|
|
for ref in unit["original_reference_spans"]:
|
|
span(text, ref["start"], ref["end"], ref["sha256"])
|
|
if not unit["span_start"] <= ref["start"] < ref["end"] <= unit["span_end"]:
|
|
raise MigrationRefusal("EVIDENCE_MIGRATION_REFERENCE_INVALID")
|
|
keys.append((unit["doc"], ref["start"], ref["end"], ref["sha256"]))
|
|
return keys
|
|
# #endregion Tooling.EvidenceLinkMigration.Unit
|
|
|
|
|
|
# #region Tooling.EvidenceLinkMigration.Command [C:3] [TYPE Function]
|
|
def validate_command(unit, replacement, tag_commit):
|
|
archive = "https://git.bebesh.ru/busya/ss-tools/src/tag/stage1-go/" + unit["doc"]
|
|
relative_index = os.path.relpath(INDEX, str(PurePosixPath(unit["doc"]).parent))
|
|
allowed = {archive, *(relative_index + "#" + anchor for anchor in unit["reference_anchors"])}
|
|
destinations = re.findall(r"\]\(([^)\n]+)\)", replacement)
|
|
if (unit["original_command_source_path"] != unit["doc"]
|
|
or unit["original_command_source_commit"] != tag_commit
|
|
or "```" in replacement or set(destinations) != allowed
|
|
or len(destinations) != len(allowed)):
|
|
raise MigrationRefusal("EVIDENCE_MIGRATION_REPLACEMENT_INVALID")
|
|
# #endregion Tooling.EvidenceLinkMigration.Command
|
|
|
|
|
|
# #region Tooling.EvidenceLinkMigration.LocalTarget [C:2] [TYPE Function]
|
|
def local_target(doc, value):
|
|
parsed = urlsplit(value)
|
|
if parsed.scheme or parsed.netloc:
|
|
return None, None
|
|
path = unquote(parsed.path)
|
|
resolved = posixpath.normpath(posixpath.join(str(PurePosixPath(doc).parent), path))
|
|
return resolved, unquote(parsed.fragment)
|
|
# #endregion Tooling.EvidenceLinkMigration.LocalTarget
|
|
|
|
|
|
# #region Tooling.EvidenceLinkMigration.PostScan [C:4] [TYPE Function]
|
|
# @POST No local raw-evidence target survives; every local index link names an existing exact anchor.
|
|
def post_scan(doc, text, anchors):
|
|
links = list(re.finditer(r"\[[^\]\n]*\]\(([^)\n]+)\)", text))
|
|
for link in links:
|
|
path, fragment = local_target(doc, link[1])
|
|
if path is None:
|
|
continue
|
|
if path == INDEX and fragment not in anchors:
|
|
raise MigrationRefusal("EVIDENCE_MIGRATION_POST_INDEX_INVALID")
|
|
if path == "specs/050-mcp-interface/evidence" or path.startswith("specs/050-mcp-interface/evidence/"):
|
|
raise MigrationRefusal("EVIDENCE_MIGRATION_POST_RAW_TARGET")
|
|
post_code_scan(text, links)
|
|
# #endregion Tooling.EvidenceLinkMigration.PostScan
|
|
|
|
|
|
# #region Tooling.EvidenceLinkMigration.PostCodeScan [C:3] [TYPE Function]
|
|
def post_code_scan(text, links):
|
|
for inline in re.finditer(r"(?<!`)`([^`\n]+)`(?!`)", text):
|
|
if any(link.start() <= inline.start() < inline.end() <= link.end() for link in links):
|
|
continue
|
|
if "specs/050-mcp-interface/evidence/" in inline[1] or inline[1].startswith("evidence/"):
|
|
raise MigrationRefusal("EVIDENCE_MIGRATION_POST_RAW_TARGET")
|
|
for fenced in re.finditer(r"```[^\n]*\n(.*?)```", text, re.DOTALL):
|
|
for token in re.findall(r"[^\s`\"'<>]+", fenced[1]):
|
|
if not urlsplit(token).scheme and ("specs/050-mcp-interface/evidence/" in token or token.startswith("evidence/")):
|
|
raise MigrationRefusal("EVIDENCE_MIGRATION_POST_RAW_TARGET")
|
|
# #endregion Tooling.EvidenceLinkMigration.PostCodeScan
|
|
|
|
|
|
# #region Tooling.EvidenceLinkMigration.Documents [C:4] [TYPE Function]
|
|
def prepare_documents(repo, commit, mapping, anchors):
|
|
sources = {path: source(repo, commit, path, info["sha256"])
|
|
for path, info in mapping["documents"].items()}
|
|
units_by_doc, unit_refs = {}, Counter()
|
|
for unit in mapping["edit_units"]:
|
|
path = unit["doc"]
|
|
if path not in sources or unit["source_sha256"] != mapping["documents"][path]["sha256"]:
|
|
raise MigrationRefusal("EVIDENCE_MIGRATION_SOURCE_CONFLICT")
|
|
unit_refs.update(validate_unit(unit, sources[path], anchors, mapping["peeled_tag_commit"]))
|
|
units_by_doc.setdefault(path, []).append(unit)
|
|
reference_keys = [(r["doc"], r["span_start"], r["span_end"], r["original_target_sha256"])
|
|
for r in mapping["references"]]
|
|
if (any(n != 1 for n in unit_refs.values()) or len(set(reference_keys)) != len(reference_keys)
|
|
or unit_refs != Counter(reference_keys) or set(sources) != set(units_by_doc)):
|
|
raise MigrationRefusal("EVIDENCE_MIGRATION_REFERENCE_CONFLICT")
|
|
outputs = {}
|
|
for path, units in units_by_doc.items():
|
|
ordered = sorted(units, key=lambda u: u["span_start"])
|
|
if any(a["span_end"] > b["span_start"] for a, b in zip(ordered, ordered[1:])):
|
|
raise MigrationRefusal("EVIDENCE_MIGRATION_SPAN_OVERLAP")
|
|
text = sources[path]
|
|
for unit in reversed(ordered):
|
|
text = text[:unit["span_start"]] + unit["replacement"] + text[unit["span_end"]:]
|
|
post_scan(path, text, anchors)
|
|
outputs[path] = text.encode()
|
|
return outputs
|
|
# #endregion Tooling.EvidenceLinkMigration.Documents
|
|
|
|
|
|
# #region Tooling.EvidenceLinkMigration.Prepare [C:4] [TYPE Function]
|
|
# @POST The complete plan is validated before any output file is created or replaced.
|
|
def prepare(repo, mapping_path, mapping_sha256, commit, index):
|
|
if (not isinstance(commit, str) or re.fullmatch(r"[a-f0-9]{40}", commit) is None
|
|
or git(repo, "rev-parse", "HEAD").decode().strip() != commit):
|
|
raise MigrationRefusal("EVIDENCE_MIGRATION_COMMIT_CONFLICT")
|
|
raw = Path(mapping_path).read_bytes()
|
|
if digest(raw) != mapping_sha256:
|
|
raise MigrationRefusal("EVIDENCE_MIGRATION_MAPPING_CONFLICT")
|
|
mapping = json.loads(raw)
|
|
if (mapping["schema_version"] != 1 or mapping["source_head"] != commit
|
|
or mapping["offset_unit"] != "unicode_codepoints_in_utf8_decoded_document"
|
|
or index != INDEX or mapping["index"]["path"] != INDEX):
|
|
raise MigrationRefusal("EVIDENCE_MIGRATION_CONTRACT_INVALID")
|
|
index_text = source(repo, commit, index, mapping["index"]["sha256"])
|
|
anchors = re.findall(r'<a id="([^"]+)"></a>', index_text)
|
|
if len(anchors) != len(set(anchors)) or len(anchors) != mapping["index"]["anchor_count"]:
|
|
raise MigrationRefusal("EVIDENCE_MIGRATION_INDEX_INVALID")
|
|
outputs = prepare_documents(repo, commit, mapping, set(anchors))
|
|
counts = mapping["counts"]
|
|
if (len(outputs) != counts["documents"] or len(mapping["edit_units"]) != counts["edit_units"]
|
|
or len(mapping["references"]) != counts["occurrences"]):
|
|
raise MigrationRefusal("EVIDENCE_MIGRATION_COUNTS_CONFLICT")
|
|
return outputs, counts
|
|
# #endregion Tooling.EvidenceLinkMigration.Prepare
|
|
|
|
|
|
# #region Tooling.EvidenceLinkMigration.Apply [C:3] [TYPE Function]
|
|
def apply(repo, outputs, index_sha256, anchors):
|
|
if digest((repo / INDEX).read_bytes()) != index_sha256:
|
|
raise MigrationRefusal("EVIDENCE_MIGRATION_POST_INDEX_INVALID")
|
|
for path, raw in outputs.items():
|
|
target = repo / path
|
|
temporary = None
|
|
try:
|
|
with tempfile.NamedTemporaryFile(dir=target.parent, prefix=".evidence-link-", delete=False) as stream:
|
|
temporary = Path(stream.name)
|
|
stream.write(raw)
|
|
temporary.chmod(target.stat().st_mode & 0o777)
|
|
os.replace(temporary, target)
|
|
finally:
|
|
if temporary is not None:
|
|
temporary.unlink(missing_ok=True)
|
|
if digest((repo / INDEX).read_bytes()) != index_sha256:
|
|
raise MigrationRefusal("EVIDENCE_MIGRATION_POST_INDEX_INVALID")
|
|
for path, expected in outputs.items():
|
|
actual = (repo / path).read_bytes()
|
|
if digest(actual) != digest(expected):
|
|
raise MigrationRefusal("EVIDENCE_MIGRATION_POST_READBACK_CONFLICT")
|
|
post_scan(path, actual.decode(), anchors)
|
|
# #endregion Tooling.EvidenceLinkMigration.Apply
|
|
|
|
|
|
# #region Tooling.EvidenceLinkMigration.Main [C:3] [TYPE Function]
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
for name in ("mapping", "mapping-sha256", "commit", "index"):
|
|
parser.add_argument("--" + name, required=True)
|
|
mode = parser.add_mutually_exclusive_group(required=True)
|
|
mode.add_argument("--check", action="store_true")
|
|
mode.add_argument("--apply", action="store_true")
|
|
args = parser.parse_args()
|
|
try:
|
|
repo = Path(git(Path.cwd(), "rev-parse", "--show-toplevel").decode().strip()).resolve()
|
|
outputs, counts = prepare(repo, args.mapping, args.mapping_sha256, args.commit, args.index)
|
|
if args.apply:
|
|
index_bytes = git(repo, "show", f"{args.commit}:{INDEX}")
|
|
anchors = set(re.findall(r'<a id="([^"]+)"></a>', index_bytes.decode()))
|
|
apply(repo, outputs, digest(index_bytes), anchors)
|
|
except MigrationRefusal as exc:
|
|
print(json.dumps({"status": "REFUSED", "error_code": exc.code}))
|
|
return 2
|
|
except (OSError, ValueError, KeyError, TypeError, UnicodeError):
|
|
print(json.dumps({"status": "REFUSED", "error_code": "EVIDENCE_MIGRATION_INPUT_INVALID"}))
|
|
return 2
|
|
print(json.dumps({"status": "APPLIED" if args.apply else "PASS", "commit": args.commit,
|
|
"documents": len(outputs), "edit_units": counts["edit_units"], "references": counts["occurrences"]}))
|
|
return 0
|
|
# #endregion Tooling.EvidenceLinkMigration.Main
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|
|
# #endregion Tooling.EvidenceLinkMigration
|