← Files APODICTIC Development EditorARCHIVED FILE
scripts/annotation_export.py
106 KB · Nov 3, 2026 · 18:57 UTC
#!/usr/bin/env python3
"""annotation-export — render the gated annotation manifest into other formats. Increment 1: Obsidian.
`manifest + snapshot → Obsidian-native Markdown`: each finding becomes a native footnote `[^<finding_id>]`
at its anchor locus, whose definition carries the **verbatim** manifest comment. No plugin (Obsidian
renders footnotes natively; CriticMarkup needs one). A pure projection of the gated manifest — it invents
nothing; the reverse transform (delete the manifest-keyed `[^id]` refs + the trailing definition block)
reproduces the snapshot byte-for-byte (the A2 discipline). The Obsidian copy is built from the SNAPSHOT
(clean ATX headings), not the CriticMarkup annotated copy.
Validator (`obsidian-export <run_folder>`):
O1 (ERROR) round-trip — manifest-keyed, two-sided. Precondition: the snapshot and every comment are free
of the footnote-ref sigil '[^' and no snapshot line is footnote-definition-shaped. Round-trip: strip
the exact `[^<finding_id>]` refs (manifest id set, never a wildcard) + the trailing `[^<id>]:` block
→ snapshot byte-for-byte.
O2 (ERROR) footnote resolution — every `[^id]` ref has exactly one definition and vice versa; the id set
equals the manifest's annotation finding_ids (the A4 forward+inverse multiset, on footnotes).
O3 (ERROR) comment fidelity — each definition body equals its manifest comment byte-for-byte (the A5
analog: relocate, never re-author).
Usage:
annotation_export.py obsidian <run_folder> # writes obsidian/<Project>_..._Obsidian_...md
annotation_export.py obsidian-export <run_folder> [--strict] # validate (O1-O3), print
annotation_export.py --self-test
Exit: 0 clean, 1 ERROR, 2 usage.
"""
import glob
import io
import os
import re
import sys
import zipfile
try:
import annotation_manifest as am
except ImportError:
am = None
_FN_REF = "[^" # footnote-ref sigil (the two-sided precondition)
_FN_DEF_RE = re.compile(r"^\[\^[^\]]+\]:") # a footnote-definition-shaped line
# The trailing definition block: a separator newline + one-or-more `[^id]: …` lines to end of text.
_DEF_BLOCK_RE = re.compile(r"\n(?:\[\^[^\]]+\]:[^\n]*\n)+\Z")
# Increment 2 (letter cross-links).
_CROSSLINKED_GLOB = "*_Crosslinked_Letter_*.md"
_FINDING_MARKER_RE = re.compile(r"<!--\s*finding:\s*(F-[A-Za-z0-9]+-[0-9]{2,})\s*-->")
_CM_SPAN_RE = re.compile(r"\{>>.*?<<\}", re.DOTALL) # a CriticMarkup span (the crosslinked back-link)
_WIKILINK_RE = re.compile(r"\[\[[^\]]*\]\]") # an Obsidian wikilink (added by Inc 2)
_WIKILINK_SIGIL = "[[" # the O5 two-sided precondition sigil
def _read(path):
try:
with open(path, encoding="utf-8") as fh:
return fh.read()
except (OSError, UnicodeDecodeError):
return None
def _newest(paths):
return max(paths, key=os.path.getmtime) if paths else None
def _insertion_offset(anchor, snapshot, nl_at, chap_l, sec_l):
"""The character offset a footnote ref is spliced at — mirroring annotation_manifest.render:
quote → end offset; chapter/section/line-range → end of the anchored line; document → end of line 1."""
kind = anchor.get("kind")
if kind == "quote":
m = re.match(r"(\d+)-(\d+)$", str(anchor.get("value", "")))
return int(m.group(2)) if m else 0
if kind == "document":
return nl_at.get(1, len(snapshot))
line = am.anchor_line(anchor, snapshot, chap_l, sec_l)
return nl_at.get(1, len(snapshot)) if line is None else nl_at.get(line, len(snapshot))
def _fwd_wikilink(letter_basename, fid):
"""The exact forward wikilink appended to a copy footnote definition (Increment 2)."""
return " [[%s#^%s|→ letter §%s]]" % (letter_basename, fid, fid)
def build_obsidian(manifest_obj, snapshot, letter_basename=None):
"""-> (obsidian_text_or_None, errs). Pure projection of the gated manifest; errs is the O1 precondition.
When letter_basename is set (Increment 2), each footnote definition appends a forward wikilink to the
letter entry — the verbatim comment is unchanged; only the exact wikilink is added."""
annotations = manifest_obj.get("annotations") if isinstance(manifest_obj.get("annotations"), list) else []
errs = []
# O1 precondition (two-sided, mirroring render/A2): no footnote sigil in the snapshot or any comment,
# and no snapshot line already shaped like a footnote definition (else the reverse transform is ambiguous).
if _FN_REF in snapshot:
errs.append("O1 precondition: the snapshot already contains a footnote-ref sigil '[^' — the reverse "
"transform would not be reversible")
for ln in snapshot.split("\n"):
if _FN_DEF_RE.match(ln):
errs.append("O1 precondition: the snapshot contains a footnote-definition-shaped line: %r" % ln[:50])
break
for an in annotations:
if not isinstance(an, dict):
continue
c = an.get("comment") or ""
if _FN_REF in c:
errs.append("O1 precondition: comment for %s contains '[^'" % an.get("finding_id"))
if "\n" in c or "\r" in c:
errs.append("O1 precondition: comment for %s is multi-line — a footnote definition is one "
"line (the trailing-block round-trip would break)" % an.get("finding_id"))
if errs:
return None, errs
if not annotations:
return snapshot, [] # nothing to annotate — the copy IS the snapshot
chap_n, sec_n, chap_l, sec_l = am.heading_index(snapshot)
nl_at, ln = {}, 1
for i, ch in enumerate(snapshot):
if ch == "\n":
nl_at[ln] = i
ln += 1
# Splice refs in DESCENDING (offset, finding_id) so each insertion sits to the right of every
# not-yet-inserted one (never perturbing offsets); co-located refs end up adjacent ([^A][^B]),
# which renders in Reading view and round-trips exactly (no inserted separator to strip).
inserts = [(_insertion_offset(an.get("anchor") or {}, snapshot, nl_at, chap_l, sec_l), an.get("finding_id"))
for an in annotations if isinstance(an, dict)]
out = snapshot
# am.fid_key normalizes a malformed (non-string / non-hashable) finding_id so the (offset, fid) sort
# cannot crash on a tie comparing a non-string id (the sibling of the chapter-value guard below).
for off, fid in sorted(inserts, key=lambda t: (t[0], am.fid_key(t[1]) or ""), reverse=True):
off = max(0, min(off, len(out)))
out = out[:off] + ("[^%s]" % fid) + out[off:]
# Definition block (deterministic: sorted by finding_id), each line the VERBATIM manifest comment,
# plus (Inc 2) the exact forward wikilink to the letter entry.
def _def(an):
fid = an.get("finding_id")
tail = _fwd_wikilink(letter_basename, fid) if letter_basename else ""
return "[^%s]: %s%s\n" % (fid, an.get("comment"), tail)
defs = "".join(_def(an) for an in sorted((a for a in annotations if isinstance(a, dict)),
key=lambda a: am.fid_key(a.get("finding_id")) or ""))
return out + "\n" + defs, []
def reverse_obsidian(obsidian_text, finding_ids):
"""The A2-analog inverse: strip the trailing definition block, then the exact manifest-keyed refs."""
body = _DEF_BLOCK_RE.sub("", obsidian_text)
for fid in finding_ids:
body = body.replace("[^%s]" % fid, "")
return body
def check_obsidian(manifest_obj, snapshot, obsidian_text, letter_basename=None):
"""O1-O3 over an emitted Obsidian copy. Returns (errs, warns). When letter_basename is set (Inc 2),
O3 expects each definition to equal `comment` + the exact forward wikilink."""
errs = []
annotations = manifest_obj.get("annotations") if isinstance(manifest_obj.get("annotations"), list) else []
# am.fid_key: a non-hashable list/object id must not crash the O2 `manifest_set[fid]` keying; str()
# coercion matches build_obsidian's `"[^%s]" % fid`-rendered footnote ref, so the multiset still ties.
ids = [am.fid_key(an.get("finding_id")) for an in annotations if isinstance(an, dict)]
# O1 (authoritative) — exact-build equality: the on-disk copy must equal a fresh deterministic build
# from the gated manifest + snapshot, byte-for-byte. The copy is fully determined by them, so this pins
# prose, every footnote ref POSITION, the verbatim definitions (+ Inc-2 wikilinks), and forbids any
# authored content — including an injected orphan trailing `[^X]: …` definition that the round-trip
# strips and O2 (a def with no ref is not a manifest id) would both miss (the H1/D1 discipline).
built, _perrs = build_obsidian(manifest_obj, snapshot, letter_basename=letter_basename)
if built is not None and obsidian_text != built:
errs.append("O1 copy integrity: the on-disk Obsidian copy is not byte-identical to a fresh build "
"from the gated manifest + snapshot (mutated prose, a moved/altered footnote, or "
"authored content such as a smuggled orphan definition)")
# O1 — round-trip to source (manifest-keyed strip == snapshot).
if reverse_obsidian(obsidian_text, ids) != snapshot:
errs.append("O1 round-trip: stripping the [^id] refs + the trailing definition block does NOT "
"reproduce the snapshot byte-for-byte (prose was altered, or a ref/def is malformed)")
# O2 — footnote resolution: refs in the body ↔ definitions ↔ manifest id set (bijection, multiset).
body = _DEF_BLOCK_RE.sub("", obsidian_text)
ref_counts = {}
for m in re.finditer(r"\[\^([^\]]+)\]", body):
ref_counts[m.group(1)] = ref_counts.get(m.group(1), 0) + 1
def_ids = _DEF_BLOCK_RE.search(obsidian_text)
def_lines = []
if def_ids:
for dl in def_ids.group(0).strip("\n").split("\n"):
dm = re.match(r"\[\^([^\]]+)\]:\s?(.*)$", dl)
if dm:
def_lines.append((dm.group(1), dm.group(2)))
def_counts = {}
for fid, _b in def_lines:
def_counts[fid] = def_counts.get(fid, 0) + 1
manifest_set = {}
for fid in ids:
manifest_set[fid] = manifest_set.get(fid, 0) + 1
for fid, n in sorted(ref_counts.items()):
if n != 1:
errs.append("O2 footnote resolution: ref [^%s] appears %d times (need exactly 1)" % (fid, n))
if fid not in manifest_set:
errs.append("O2 footnote resolution: ref [^%s] is not a manifest finding_id (un-manifested footnote)" % fid)
for fid, n in sorted(def_counts.items()):
if n != 1:
errs.append("O2 footnote resolution: definition [^%s]: appears %d times (need exactly 1)" % (fid, n))
for fid in sorted(manifest_set):
if ref_counts.get(fid, 0) != 1:
errs.append("O2 footnote resolution: manifest finding %s has %d refs (need exactly 1)"
% (fid, ref_counts.get(fid, 0)))
if def_counts.get(fid, 0) != 1:
errs.append("O2 footnote resolution: manifest finding %s has %d definitions (need exactly 1)"
% (fid, def_counts.get(fid, 0)))
# O3 — comment fidelity: each definition body equals the manifest comment (Inc 1) or the comment +
# the EXACT forward wikilink (Inc 2). No room for authored text around the verbatim comment.
# am.fid_key: a non-hashable finding_id must not crash this O3 dict-key; the str() form matches the
# parsed `def_lines` fid (which came from the `"[^%s]" % fid`-rendered ref), so the lookup still ties.
comment_of = {am.fid_key(an.get("finding_id")): an.get("comment") for an in annotations if isinstance(an, dict)}
for fid, body_text in def_lines:
if fid not in comment_of:
continue
# Only a STRING comment is the verbatim body. `or ""` neutralizes a FALSY comment (None/""),
# but a TRUTHY non-string comment (a JSON list/dict — kept as-is by parse_manifest) would slip
# through and `list + str` would crash on this line. Coerce any non-string comment to "" so a
# malformed comment yields the same clean O3 fidelity error as None does, never a traceback.
manifest_comment = comment_of[fid] if isinstance(comment_of[fid], str) else ""
expected = manifest_comment + (_fwd_wikilink(letter_basename, fid) if letter_basename else "")
if body_text != expected:
errs.append("O3 comment fidelity: definition for %s is not the verbatim manifest comment%s "
"(relocate, never re-author)" % (fid, " + exact forward wikilink" if letter_basename else ""))
return errs, []
# ---------------------------------------------------------------- Increment 2: the letter
def _heading_slug(raw):
"""A copy heading's match text with any appended footnote ref stripped (Obsidian strips it from the
anchor slug — web-verified): `Chapter 9[^F-RR-01]` -> `Chapter 9`."""
return re.sub(r"\[\^[^\]]+\]", "", raw).strip()
def _heading_text_of(anchor, snapshot):
"""For a chapter/section anchor, the snapshot heading's text (e.g. 'Chapter 9'); else None."""
kind, val = anchor.get("kind"), anchor.get("value")
_cn, _sn, chap_l, sec_l = am.heading_index(snapshot)
if kind == "chapter":
# a non-string chapter value is unhashable -> chap_l.get(val) would crash (sibling of the
# anchor_line guard; the section branch already str()-coerces).
lineno = chap_l.get(val) if isinstance(val, str) else None
elif kind == "section":
lineno = sec_l.get(str(val).strip().lower())
else:
return None
if not lineno:
return None
for (l, raw, _norm) in am.atx_headings(snapshot):
if l == lineno:
return raw
return None
def _reverse_wikilink(anchor, snapshot, copy_basename):
"""The letter→copy back-link for an anchor -> (wikilink, is_file_level). A heading link when the
anchor resolves to a heading whose text is safe in an Obsidian `#fragment` (no `] [ | #`, which would
break the wikilink syntax / the O5 strip); otherwise a file-level link (W1)."""
heading = _heading_text_of(anchor, snapshot)
if heading is not None and not any(c in heading for c in "][|#"):
return "[[%s#%s]]" % (copy_basename, heading), False
return "[[%s]]" % copy_basename, True
def build_obsidian_letter(crosslinked_text, manifest_obj, snapshot, copy_basename):
"""-> (letter_text_or_None, errs, warns). Project the gated crosslinked letter: append ` ^F-id` block
ids to finding lines + convert each CriticMarkup back-link span to a `[[copy#heading]]` wikilink
(file-level for line-range/quote/document — W1). Pure projection; the editorial prose is untouched (O5)."""
errs, warns = [], []
# O5 two-sided precondition: the editorial prose (letter minus CriticMarkup spans) must carry no
# wikilink sigil and no line-terminal block id, so the strip is unambiguous (mirrors render/A2).
prose = am.reverse_transform(crosslinked_text)
if _WIKILINK_SIGIL in prose:
errs.append("O5 precondition: the editorial letter already contains '[[' — the reverse transform "
"would be ambiguous")
for ln in prose.split("\n"):
if re.search(r" \^\S+$", ln):
errs.append("O5 precondition: the editorial letter has a line-terminal ' ^token' block id")
break
if errs:
return None, errs, warns
# am.fid_key: a non-hashable finding_id must not crash this dict-key; the lookup fid comes from the
# crosslinked marker regex (a well-formed F-id string), so the str() form still resolves it.
anchor_of = {am.fid_key(an.get("finding_id")): (an.get("anchor") or {})
for an in (manifest_obj.get("annotations") or []) if isinstance(an, dict)}
out_lines = []
for line in crosslinked_text.split("\n"):
m = _FINDING_MARKER_RE.search(line)
if not m:
out_lines.append(line)
continue
fid = m.group(1)
anchor = anchor_of.get(fid, {})
wl, file_level = _reverse_wikilink(anchor, snapshot, copy_basename)
if file_level:
warns.append("W1 file-level back-link: %s anchor (%s) has no Obsidian-linkable heading"
% (fid, anchor.get("kind")))
new_line, n = _CM_SPAN_RE.subn(wl, line, count=1) # replace the back-link span in place
if n == 0:
new_line = line + wl # no span (uncited) — append
out_lines.append(new_line + " ^%s" % fid) # block id at end of the single-line block
return "\n".join(out_lines), errs, warns
def strip_obsidian_letter(text, finding_ids, copy_basename=None):
"""Strip Inc-2 additions (the [[…]] wikilinks + line-terminal ` ^<fid>`) — the O5 inverse. When
copy_basename is given the wikilink strip is **manifest-keyed** to that copy target (`[[copy]]` /
`[[copy#heading]]`) — never a `[[…]]` wildcard — so an injected wikilink to ANOTHER target survives
the strip and is caught by the O5 prose compare (the firewall's manifest-keyed-strip discipline)."""
if copy_basename is not None:
t = re.sub(r"\[\[%s(?:#[^\]]*)?\]\]" % re.escape(copy_basename), "", text)
else:
t = _WIKILINK_RE.sub("", text)
for fid in finding_ids:
t = t.replace(" ^%s" % fid, "")
return t
def check_obsidian_letter(crosslinked_text, obsidian_letter, copy_text, copy_basename, letter_basename,
manifest_obj, snapshot):
"""O4 (link resolution) + O5 (letter prose fidelity) over the emitted Obsidian letter + copy."""
errs, warns = [], []
ids = [an.get("finding_id") for an in (manifest_obj.get("annotations") or []) if isinstance(an, dict)]
# O5 (authoritative) — exact-build equality: the on-disk letter must equal a fresh build from the gated
# crosslinked letter + manifest + snapshot + copy basename, byte-for-byte. The letter is fully determined
# by those inputs, so this pins the editorial prose, every block id, and every back-link wikilink, and
# forbids any authored content. This is the firewall guarantee: it closes the wildcard-strip hole where
# `_WIKILINK_RE.sub("")` deleted an INJECTED `[[Injected_Note]]` before the prose compare, so a tampered
# on-disk letter passed (the H1/D1 discipline). The strip-based check below remains a granular diagnostic.
built, _berrs, _bw = build_obsidian_letter(crosslinked_text, manifest_obj, snapshot, copy_basename)
if built is not None and obsidian_letter != built:
errs.append("O5 letter integrity: the on-disk Obsidian letter is not byte-identical to a fresh build "
"from the gated crosslinked letter + manifest + snapshot (an injected/altered wikilink or "
"block id, or mutated editorial prose)")
# O5 — letter prose fidelity (diagnostic): strip the Obsidian additions == strip the crosslinked
# CriticMarkup. Manifest-keyed to the copy target (not a `[[…]]` wildcard), so an injected wikilink to a
# foreign target survives the strip and trips this compare too.
if strip_obsidian_letter(obsidian_letter, ids, copy_basename) != am.reverse_transform(crosslinked_text):
errs.append("O5 letter prose fidelity: stripping the Obsidian letter's additions does not "
"reproduce the crosslinked letter's editorial prose (the projection mutated prose)")
# O4 forward — every copy `[[letter#^id]]` resolves to a `^id` block id in the letter.
letter_block_ids = set(re.findall(r" \^(F-[A-Za-z0-9]+-[0-9]{2,})$", obsidian_letter, re.MULTILINE))
fwd = re.compile(r"\[\[%s#\^(F-[A-Za-z0-9]+-[0-9]{2,})(?:\|[^\]]*)?\]\]" % re.escape(letter_basename))
for m in fwd.finditer(copy_text):
if m.group(1) not in letter_block_ids:
errs.append("O4 link resolution: copy forward link [[%s#^%s]] has no matching ^%s block id "
"in the letter" % (letter_basename, m.group(1), m.group(1)))
# O4 reverse — PER-FINDING correctness (not mere set-membership): each finding line's back-link must
# be EXACTLY the expected `[[copy#<that finding's heading>]]` (or file-level), so a real-but-wrong
# heading — or a smuggled `|display` label — fails. Defense-in-depth: any heading fragment must also
# resolve to a real copy heading (footnote refs stripped).
copy_headings = set(_heading_slug(raw) for (_l, raw, _n) in am.atx_headings(copy_text))
# am.fid_key: a non-hashable finding_id must not crash this dict-key (sibling of build_obsidian_letter).
anchor_of = {am.fid_key(an.get("finding_id")): (an.get("anchor") or {})
for an in (manifest_obj.get("annotations") or []) if isinstance(an, dict)}
for line in obsidian_letter.split("\n"):
fm = _FINDING_MARKER_RE.search(line)
if not fm:
continue
fid = fm.group(1)
expected, _fl = _reverse_wikilink(anchor_of.get(fid, {}), snapshot, copy_basename)
if expected not in line:
errs.append("O4 link resolution: %s back-link is not the expected %s "
"(wrong heading, or a smuggled display label)" % (fid, expected))
rm = re.search(r"\[\[%s#([^\]|]+)\]\]" % re.escape(copy_basename), line)
if rm and rm.group(1).strip() not in copy_headings:
errs.append("O4 link resolution: letter reverse link [[%s#%s]] matches no copy heading"
% (copy_basename, rm.group(1).strip()))
return errs, warns
# ---------------------------------------------------------------- Increment 3: read-only HTML
_HTML_STYLE = (
"body{max-width:46em;margin:2rem auto;padding:0 1rem;"
"font-family:Georgia,'Times New Roman',serif;line-height:1.5;color:#222}\n"
"pre.manuscript{white-space:pre-wrap;font-family:inherit;font-size:1rem;margin:0}\n"
".fnref{font-size:.75em;color:#a00;vertical-align:super}\n"
".fnref a,.backref{text-decoration:none;color:#a00}\n"
"section.findings{margin-top:2rem;border-top:1px solid #ccc;padding-top:1rem;font-size:.95em}\n"
"section.findings h2{font-size:1.1rem}\n"
"section.findings li{margin:.4rem 0}\n"
)
def _html_escape(s):
"""Element-content escaping: `&` FIRST, then `<`, `>` (the order is load-bearing for the exact inverse)."""
return (s or "").replace("&", "&").replace("<", "<").replace(">", ">")
def _html_unescape_exact(s):
"""The EXACT inverse of _html_escape (mirror order: `<`, `>`, then `&` LAST). NOT a general decoder."""
return (s or "").replace("<", "<").replace(">", ">").replace("&", "&")
def _html_marker(fid):
return '<sup class="fnref" id="ref-%s"><a href="#fn-%s">[%s]</a></sup>' % (fid, fid, fid)
def _html_backref(fid):
return '<a class="backref" href="#ref-%s">↩</a>' % fid
_PRE_RE = re.compile(r'<pre class="manuscript">(.*)</pre>', re.DOTALL)
def build_html(manifest_obj, snapshot):
"""-> (html_text_or_None, errs). Pure projection: a self-contained read-only HTML — the snapshot in a
faithful <pre> (escaped, markers spliced between escaped segments) + a findings list with bidirectional
anchors. errs is the two-sided precondition."""
annotations = [a for a in (manifest_obj.get("annotations") or []) if isinstance(a, dict)]
errs = []
# Two-sided precondition (by construction, since escaping removes every `<`): neither the escaped
# snapshot nor any escaped comment can contain the literal marker prefix `<sup`.
if "<sup" in _html_escape(snapshot):
errs.append("H1 precondition: the escaped snapshot contains '<sup' (impossible after escaping)")
for a in annotations:
if "<sup" in _html_escape(a.get("comment") or ""):
errs.append("H1 precondition: escaped comment for %s contains '<sup'" % a.get("finding_id"))
if errs:
return None, errs
project = manifest_obj.get("project", "Manuscript")
chap_n, sec_n, chap_l, sec_l = am.heading_index(snapshot)
nl_at, ln = {}, 1
for i, ch in enumerate(snapshot):
if ch == "\n":
nl_at[ln] = i
ln += 1
# group markers by raw-snapshot offset (co-located markers render adjacent, deterministic by id)
by_off = {}
for a in annotations:
off = _insertion_offset(a.get("anchor") or {}, snapshot, nl_at, chap_l, sec_l)
by_off.setdefault(max(0, min(off, len(snapshot))), []).append(a.get("finding_id"))
# escape each prose segment between offsets; splice literal markers between (never escaped, no drift)
parts, prev = [], 0
for off in sorted(by_off):
parts.append(_html_escape(snapshot[prev:off]))
# am.fid_key in the SORT KEY: a malformed (non-string / non-hashable) finding_id must not crash the
# co-located-marker ordering — the HTML sibling of build_obsidian's guard (the marker itself is
# `%s`-rendered, so a coerced id and the raw id emit the same text and still round-trip).
for fid in sorted(by_off[off], key=lambda f: am.fid_key(f) or ""):
parts.append(_html_marker(fid))
prev = off
parts.append(_html_escape(snapshot[prev:]))
pre_content = "".join(parts)
lis = "".join('<li id="fn-%s">%s %s</li>\n'
% (a.get("finding_id"), _html_escape(a.get("comment")), _html_backref(a.get("finding_id")))
for a in sorted(annotations, key=lambda x: am.fid_key(x.get("finding_id")) or ""))
html = ('<!DOCTYPE html>\n<html lang="en">\n<head>\n<meta charset="utf-8">\n'
'<title>%s — Annotated Manuscript</title>\n<style>\n%s</style>\n</head>\n<body>\n'
'<main><pre class="manuscript">%s</pre></main>\n'
'<section class="findings">\n<h2>Findings</h2>\n<ol>\n%s</ol>\n</section>\n</body>\n</html>\n'
% (_html_escape(project), _HTML_STYLE, pre_content, lis))
return html, []
def reverse_html(pre_content, finding_ids):
"""The A2-analog inverse: delete the exact manifest-keyed markers, then the exact 3-entity unescape."""
t = pre_content
for fid in finding_ids:
t = t.replace(_html_marker(fid), "")
return _html_unescape_exact(t)
def check_html(manifest_obj, snapshot, html_text):
"""H1-H3 over an emitted HTML export. Returns (errs, warns).
H1 is the AUTHORITATIVE lock: the on-disk HTML must equal a fresh build from the gated manifest +
snapshot byte-for-byte — the artifact is fully determined by them, so equality pins prose, comments,
marker POSITIONS, structure, and forbids any authored content (in or outside the <pre>/findings).
H2/H3 (and the round-trip) remain as granular diagnostics, but H1-equality is the firewall guarantee."""
errs = []
annotations = [a for a in (manifest_obj.get("annotations") or []) if isinstance(a, dict)]
# am.fid_key: a non-hashable finding_id must not crash the H2 `manifest_set[fid]` dict-key nor the
# round-trip marker strip; the str() form matches build_html's `%s`-rendered marker, so it still ties.
ids = [am.fid_key(a.get("finding_id")) for a in annotations]
# H1 (authoritative) — exact-build equality: the on-disk artifact == a fresh deterministic build.
expected, _perrs = build_html(manifest_obj, snapshot)
if expected is not None and html_text != expected:
errs.append("H1 artifact integrity: the on-disk HTML is not byte-identical to a fresh build from "
"the gated manifest + snapshot — prose / comment / marker-position / structure drift, "
"or authored content outside the manuscript or findings")
n_pre = len(re.findall(r'<pre class="manuscript">', html_text)) # count the OPENING tag (greedy-safe)
pres = _PRE_RE.findall(html_text)
if n_pre != 1 or len(pres) != 1:
errs.append('H1 round-trip: expected exactly one <pre class="manuscript">…</pre> (found %d)' % n_pre)
return errs, []
pre_content = pres[0]
# H1 — round-trip: strip manifest-keyed markers + exact unescape == snapshot.
if reverse_html(pre_content, ids) != snapshot:
errs.append("H1 round-trip: stripping the markers + exact HTML-unescape does NOT reproduce the "
"snapshot byte-for-byte (prose altered, or a marker/escape is malformed)")
# H2 — anchor resolution: <sup id="ref-fid"> ↔ <li id="fn-fid"> bijection == manifest id set.
ref_ids = re.findall(r'<sup class="fnref" id="ref-(F-[A-Za-z0-9]+-[0-9]{2,})">', pre_content)
fn_ids = re.findall(r'<li id="fn-(F-[A-Za-z0-9]+-[0-9]{2,})">', html_text)
manifest_set = {}
for fid in ids:
manifest_set[fid] = manifest_set.get(fid, 0) + 1
for label, found in (("marker", ref_ids), ("finding", fn_ids)):
counts = {}
for fid in found:
counts[fid] = counts.get(fid, 0) + 1
for fid, n in sorted(counts.items()):
if n != 1:
errs.append("H2 anchor resolution: %s id %s appears %d times (need exactly 1)" % (label, fid, n))
if fid not in manifest_set:
errs.append("H2 anchor resolution: %s id %s is not a manifest finding_id (un-manifested)" % (label, fid))
for fid in sorted(manifest_set):
if counts.get(fid, 0) != 1:
errs.append("H2 anchor resolution: manifest finding %s has %d %s id(s) (need exactly 1)"
% (fid, counts.get(fid, 0), label))
# H3 — comment fidelity: each <li> content == escape(comment) + " " + exact back-ref.
# am.fid_key: a non-hashable finding_id must not crash this dict-key (the H3 sibling of O3's guard).
comment_of = {am.fid_key(a.get("finding_id")): a.get("comment") for a in annotations}
for m in re.finditer(r'<li id="fn-(F-[A-Za-z0-9]+-[0-9]{2,})">(.*?)</li>', html_text, re.DOTALL):
fid, content = m.group(1), m.group(2)
if fid in comment_of:
expected = "%s %s" % (_html_escape(comment_of[fid]), _html_backref(fid))
if content != expected:
errs.append("H3 comment fidelity: <li> for %s is not the verbatim manifest comment + the "
"exact back-ref (relocate, never re-author)" % fid)
return errs, []
# ---------------------------------------------------------------- Increment 4: DOCX (→ Google Docs)
# A .docx is a ZIP of OOXML parts. XML element-content escaping reuses _html_escape/_html_unescape_exact
# (the same exact `& < >` 3-entity pair). The model assembles fixed boilerplate around copied bytes.
_W_NS = "http://schemas.openxmlformats.org/wordprocessingml/2006/main"
_DOCX_DATE_FALLBACK = "2026-01-01T12:00:00Z" # only when the runlabel carries no YYYY-MM-DD date
_XMLDECL = '<?xml version="1.0" encoding="UTF-8" standalone="yes"?>\n'
_DOCX_CT = (_XMLDECL +
'<Types xmlns="http://schemas.openxmlformats.org/package/2006/content-types">'
'<Default Extension="rels" ContentType="application/vnd.openxmlformats-package.relationships+xml"/>'
'<Default Extension="xml" ContentType="application/xml"/>'
'<Override PartName="/word/document.xml" ContentType="application/vnd.openxmlformats-officedocument.wordprocessingml.document.main+xml"/>'
'<Override PartName="/word/comments.xml" ContentType="application/vnd.openxmlformats-officedocument.wordprocessingml.comments+xml"/>'
'<Override PartName="/word/styles.xml" ContentType="application/vnd.openxmlformats-officedocument.wordprocessingml.styles+xml"/>'
'<Override PartName="/word/settings.xml" ContentType="application/vnd.openxmlformats-officedocument.wordprocessingml.settings+xml"/>'
'</Types>')
_DOCX_ROOT_RELS = (_XMLDECL +
'<Relationships xmlns="http://schemas.openxmlformats.org/package/2006/relationships">'
'<Relationship Id="rId1" Type="http://schemas.openxmlformats.org/officeDocument/2006/relationships/officeDocument" Target="word/document.xml"/>'
'</Relationships>')
_DOCX_DOC_RELS = (_XMLDECL +
'<Relationships xmlns="http://schemas.openxmlformats.org/package/2006/relationships">'
'<Relationship Id="rId1" Type="http://schemas.openxmlformats.org/officeDocument/2006/relationships/comments" Target="comments.xml"/>'
'<Relationship Id="rId2" Type="http://schemas.openxmlformats.org/officeDocument/2006/relationships/styles" Target="styles.xml"/>'
'<Relationship Id="rId3" Type="http://schemas.openxmlformats.org/officeDocument/2006/relationships/settings" Target="settings.xml"/>'
'</Relationships>')
_DOCX_STYLES = (_XMLDECL +
'<w:styles xmlns:w="%s">'
'<w:docDefaults><w:rPrDefault><w:rPr/></w:rPrDefault><w:pPrDefault><w:pPr/></w:pPrDefault></w:docDefaults>'
'<w:style w:type="paragraph" w:default="1" w:styleId="Normal"><w:name w:val="Normal"/></w:style>'
'</w:styles>' % _W_NS)
_DOCX_SETTINGS = _XMLDECL + '<w:settings xmlns:w="%s"/>' % _W_NS
_WT_RE = re.compile(r'<w:t xml:space="preserve">(.*?)</w:t>', re.DOTALL)
_WP_RE = re.compile(r"<w:p>(.*?)</w:p>", re.DOTALL)
def _line_bounds(snapshot):
"""1-based lineno -> (start_offset, end_offset_of_content) for each snapshot line."""
line_start, line_end, off, lineno = {}, {}, 0, 1
for line in snapshot.split("\n"):
line_start[lineno], line_end[lineno] = off, off + len(line)
off += len(line) + 1
lineno += 1
return line_start, line_end
def _docx_span(anchor, snapshot, line_start, line_end, chap_l, sec_l):
"""An explicit (start, end) char span for an anchor (commentRange needs a span, not a point)."""
kind, val = anchor.get("kind"), anchor.get("value")
if kind == "quote":
m = re.match(r"(\d+)-(\d+)$", str(val))
if m:
return int(m.group(1)), int(m.group(2))
elif kind == "chapter":
# a non-string chapter value is unhashable -> chap_l.get(val) would crash (sibling of the
# anchor_line guard; the section branch already str()-coerces).
ln = chap_l.get(val) if isinstance(val, str) else None
if ln:
return line_start[ln], line_end[ln]
elif kind == "section":
ln = sec_l.get(str(val).strip().lower())
if ln:
return line_start[ln], line_end[ln]
elif kind == "line-range":
m = re.match(r"(\d+)-(\d+)$", str(val))
if m and int(m.group(1)) in line_start and int(m.group(2)) in line_end:
return line_start[int(m.group(1))], line_end[int(m.group(2))]
return line_start.get(1, 0), line_end.get(1, 0) # document / fallback -> wrap line 1
def _docx_body(snapshot, spans):
"""document.xml body: one <w:p> per snapshot line; each span wrapped commentRangeStart/End + reference.
Ranges sharing an offset NEST rather than cross: at a point, opens go in ascending id (lowest id is
outermost — `sorted(finding_id)` order), and closes go in reverse-open order (latest-opened first), so a
co-located pair {0,1} emits `start0 start1 … end1 ref1 end0 ref0` — proper LIFO nesting Word/GDocs anchor
cleanly, not the crossing `start0 start1 … end0 ref0 end1 ref1`."""
starts, ends = {}, {}
start_of = {}
for s, e, wid in spans:
starts.setdefault(s, []).append(wid)
ends.setdefault(e, []).append(wid)
start_of[wid] = s
def run(text):
return '<w:r><w:t xml:space="preserve">%s</w:t></w:r>' % _html_escape(text) if text else ""
lines = snapshot.split("\n")
if lines and lines[-1] == "":
lines = lines[:-1] # drop the trailing empty (the final newline is not a paragraph)
out, off = [], 0
for line in lines:
ls, le = off, off + len(line)
inner, prev = [], ls
for o in sorted(o for o in set(list(starts) + list(ends)) if ls <= o <= le):
if o > prev:
inner.append(run(snapshot[prev:o]))
# close in reverse-open order (latest-opened first): sort by (start offset, id) DESCENDING so
# co-located/co-terminating ranges nest instead of cross (Codex #111).
for wid in sorted(ends.get(o, []), key=lambda w: (start_of[w], w), reverse=True):
inner.append('<w:commentRangeEnd w:id="%d"/><w:r><w:commentReference w:id="%d"/></w:r>' % (wid, wid))
for wid in sorted(starts.get(o, [])):
inner.append('<w:commentRangeStart w:id="%d"/>' % wid)
prev = o
if le > prev:
inner.append(run(snapshot[prev:le]))
out.append("<w:p>" + "".join(inner) + "</w:p>")
off = le + 1
return "".join(out)
def _deterministic_zip(parts):
"""A byte-stable .docx: ZIP_STORED, explicit ZipInfo per part with pinned date / create_system /
external_attr, fixed order, seekable buffer (so the same bytes emit on every machine)."""
buf = io.BytesIO()
with zipfile.ZipFile(buf, "w") as zf:
for name, data in parts:
zi = zipfile.ZipInfo(name, date_time=(1980, 1, 1, 0, 0, 0))
zi.compress_type = zipfile.ZIP_STORED
zi.create_system = 3 # Unix (else host-dependent: 0 on Windows)
zi.external_attr = 0o644 << 16 # else open(mode='w') rewrites it
zf.writestr(zi, data.encode("utf-8"))
return buf.getvalue()
def _docx_date(manifest_obj):
"""The comment `w:date` — the run's OWN date (the manifest runlabel's `YYYY-MM-DD`), at **noon UTC** so
it never rolls back a day in a western timezone (a midnight-UTC stamp displays as the day before). It is a
deterministic function of the gated manifest, NOT the wall clock, so the committed .docx stays byte-stable
(the `--check-all` byte-identity gate still holds). Falls back to a fixed literal if the runlabel carries
no date prefix OR the prefix is not a real calendar date (so we never emit a malformed `xsd:dateTime`)."""
import datetime as _dt # used only to VALIDATE the date string — a pure parse, never reads the clock
m = re.match(r"(\d{4}-\d{2}-\d{2})", str(manifest_obj.get("runlabel") or ""))
if m:
try:
_dt.date.fromisoformat(m.group(1)) # reject shape-valid-but-impossible dates (e.g. 2026-13-45)
return m.group(1) + "T12:00:00Z"
except ValueError:
pass
return _DOCX_DATE_FALLBACK
def build_docx(manifest_obj, snapshot):
"""-> (docx_bytes_or_None, errs). Pure projection: the snapshot in <w:p>/<w:t> with each finding's
span wrapped as an anchored comment carrying the verbatim comment. errs is the single-line precondition."""
annotations = [a for a in (manifest_obj.get("annotations") or []) if isinstance(a, dict)]
errs = []
for a in annotations:
c = a.get("comment") or ""
if "\n" in c or "\r" in c:
errs.append("D precondition: comment for %s is multi-line — a <w:t> is one line" % a.get("finding_id"))
if errs:
return None, errs
# am.fid_key: a non-hashable finding_id must not crash the sort key nor the idmap dict-key/lookup
# (the DOCX sibling of the Obsidian/HTML guards); the comment-id is the idmap index, not the id itself.
ordered = sorted(annotations, key=lambda a: am.fid_key(a.get("finding_id")) or "")
idmap = {am.fid_key(a.get("finding_id")): i for i, a in enumerate(ordered)}
line_start, line_end = _line_bounds(snapshot)
_cn, _sn, chap_l, sec_l = am.heading_index(snapshot)
spans = [(*_docx_span(a.get("anchor") or {}, snapshot, line_start, line_end, chap_l, sec_l),
idmap[am.fid_key(a.get("finding_id"))]) for a in ordered]
document_xml = '%s<w:document xmlns:w="%s"><w:body>%s</w:body></w:document>' % (
_XMLDECL, _W_NS, _docx_body(snapshot, spans))
docx_date = _docx_date(manifest_obj)
comments = "".join(
'<w:comment w:id="%d" w:author="APODICTIC" w:date="%s" w:initials="AP">'
'<w:p><w:r><w:t xml:space="preserve">%s</w:t></w:r></w:p></w:comment>'
% (idmap[am.fid_key(a.get("finding_id"))], docx_date, _html_escape(a.get("comment"))) for a in ordered)
comments_xml = '%s<w:comments xmlns:w="%s">%s</w:comments>' % (_XMLDECL, _W_NS, comments)
parts = [("[Content_Types].xml", _DOCX_CT), ("_rels/.rels", _DOCX_ROOT_RELS),
("word/document.xml", document_xml), ("word/comments.xml", comments_xml),
("word/styles.xml", _DOCX_STYLES), ("word/settings.xml", _DOCX_SETTINGS),
("word/_rels/document.xml.rels", _DOCX_DOC_RELS)]
return _deterministic_zip(parts), []
def check_docx(manifest_obj, snapshot, docx_bytes):
"""D1-D3 over an emitted .docx. Returns (errs, warns)."""
errs = []
annotations = [a for a in (manifest_obj.get("annotations") or []) if isinstance(a, dict)]
# D1 (authoritative) — the on-disk .docx == a fresh deterministic build, byte-for-byte.
expected, _perrs = build_docx(manifest_obj, snapshot)
if expected is not None and docx_bytes != expected:
errs.append("D1 artifact integrity: the on-disk .docx is not byte-identical to a fresh build from "
"the gated manifest + snapshot (text / comment / structure drift, or authored content)")
try:
zf = zipfile.ZipFile(io.BytesIO(docx_bytes))
document_xml = zf.read("word/document.xml").decode("utf-8")
comments_xml = zf.read("word/comments.xml").decode("utf-8")
except (zipfile.BadZipFile, KeyError, UnicodeDecodeError) as exc:
return errs + ["D1: cannot read the .docx OOXML parts — %s" % exc], []
# D2 — text round-trip: <w:t> text, exact-unescaped, one <w:p> per line, one trailing newline.
para_texts = [_html_unescape_exact("".join(_WT_RE.findall(p))) for p in _WP_RE.findall(document_xml)]
if (("\n".join(para_texts) + "\n") if para_texts else "") != snapshot:
errs.append("D2 text round-trip: the document.xml <w:t> text does not reproduce the snapshot "
"byte-for-byte")
# D3 — comment resolution + fidelity (rebuild the sorted(finding_id)->N map).
# am.fid_key: mirror build_docx's coercion so a non-hashable finding_id cannot crash the rebuilt map.
ordered = sorted(annotations, key=lambda a: am.fid_key(a.get("finding_id")) or "")
idmap = {am.fid_key(a.get("finding_id")): i for i, a in enumerate(ordered)}
expected_ids = set(idmap.values())
starts = [int(x) for x in re.findall(r'<w:commentRangeStart w:id="(\d+)"/>', document_xml)]
ends = [int(x) for x in re.findall(r'<w:commentRangeEnd w:id="(\d+)"/>', document_xml)]
refs = [int(x) for x in re.findall(r'<w:commentReference w:id="(\d+)"/>', document_xml)]
comment_text, comment_ids = {}, []
for cm in re.finditer(r'<w:comment w:id="(\d+)"[^>]*>(.*?)</w:comment>', comments_xml, re.DOTALL):
comment_ids.append(int(cm.group(1)))
comment_text[int(cm.group(1))] = _html_unescape_exact("".join(_WT_RE.findall(cm.group(2))))
def _exactly_once(lst): # every manifest id appears EXACTLY once, no missing / duplicate / extra
seen = {}
for x in lst:
seen[x] = seen.get(x, 0) + 1
return set(seen) == expected_ids and all(v == 1 for v in seen.values())
if not (_exactly_once(starts) and _exactly_once(ends) and _exactly_once(refs) and _exactly_once(comment_ids)):
errs.append("D3 comment resolution: the commentRangeStart/End/Reference ids and the comments.xml "
"ids must each contain every manifest finding {0..n-1} exactly once (no missing, "
"duplicate, or un-manifested id)")
for a in ordered:
wid = idmap[am.fid_key(a.get("finding_id"))]
if comment_text.get(wid) != a.get("comment"):
errs.append("D3 comment fidelity: comment %d is not the verbatim manifest comment for %s "
"(relocate, never re-author)" % (wid, a.get("finding_id")))
return errs, []
# ---------------------------------------------------------------- Increment 5: PDF (proofing target)
# A .pdf is a hand-written stack of PDF objects — header, body, xref, trailer — NO external libs. The
# manuscript prose renders as base-14 Helvetica text; each finding drops an inline [<finding_id>] marker
# at its anchor locus (the HTML <sup> precedent, without a font that can superscript) and its verbatim
# comment lands in a trailing Findings section. Byte-determinism (the DOCX discipline, for a different
# binary): NO /Info dict (so no /CreationDate or /ModDate), NO /ID array, NO stream compression (so no
# zlib-version drift), all text emitted as octal-escaped ASCII (so no locale/encoding drift), a fixed
# object order, and xref offsets computed from byte lengths — no wall clock, no random anywhere.
_PDF_FONT_SIZE = 11
_PDF_LEADING = 14
_PDF_X = 72
_PDF_Y_START = 734 # first T* moves down one LEADING -> baseline 720
_PDF_PAGE_W = 612 # US Letter
_PDF_PAGE_H = 792
_PDF_LINES_PER_PAGE = 45 # (720 - 45*14) = 90 >= 72 bottom margin
_PDF_WRAP = 90 # findings comments wrap at a raw char boundary (concatenation is exact)
_PDF_TITLE_SUFFIX = " — Annotated Manuscript"
_PDF_FINDINGS_HEADER = "Findings"
_PDF_MARKER_RE = re.compile(r"\[(F-[A-Za-z0-9]+-[0-9]{2,})\]")
def _pdf_marker(fid):
return "[%s]" % fid
def _pdf_comment_str(annotation):
"""The verbatim comment as a string; a non-string (malformed) comment coerces to '' (never crashes)."""
c = annotation.get("comment")
return c if isinstance(c, str) else ""
def _pdf_enc_ok(s):
"""True iff every char is WinAnsi(CP1252)-encodable — the base-14 Helvetica codomain."""
try:
(s if isinstance(s, str) else "").encode("cp1252")
return True
except (UnicodeEncodeError, LookupError):
return False
def _pdf_escape_string(s):
"""A PDF string-literal body: CP1252 bytes, with `( ) \\` and every non-printable/high byte emitted as
an octal `\\ddd` escape — so the whole content stream stays ASCII and byte-deterministic. The font's
/WinAnsiEncoding interprets the bytes; a `\\247` renders as § matching the CP1252 encode."""
out = []
for byte in s.encode("cp1252"):
if byte == 0x28:
out.append("\\(")
elif byte == 0x29:
out.append("\\)")
elif byte == 0x5C:
out.append("\\\\")
elif 32 <= byte <= 126:
out.append(chr(byte))
else:
out.append("\\%03o" % byte)
return "".join(out)
def _pdf_string_literals(stream_text):
"""Parse every `( … )` string literal in a content stream, in order, back to a unicode str (the exact
inverse of _pdf_escape_string: octal + `\\( \\) \\\\` unescape, then CP1252-decode). The build emits a
`(…)` literal ONLY for shown text (each followed by ` Tj`), so this yields exactly the shown strings."""
out, i, n = [], 0, len(stream_text)
while i < n:
if stream_text[i] != "(":
i += 1
continue
i += 1
depth, buf = 1, []
while i < n and depth > 0:
c = stream_text[i]
if c == "\\":
nxt = stream_text[i + 1] if i + 1 < n else ""
if nxt in "01234567":
j, octs = i + 1, ""
while j < n and len(octs) < 3 and stream_text[j] in "01234567":
octs += stream_text[j]
j += 1
buf.append(int(octs, 8) & 0xFF)
i = j
continue
simple = {"n": 10, "r": 13, "t": 9, "b": 8, "f": 12, "(": 40, ")": 41, "\\": 92}
if nxt in simple:
buf.append(simple[nxt])
elif nxt == "\n":
pass # line continuation -> nothing
else:
buf.append(ord(nxt) & 0xFF) # `\x` -> x
i += 2
continue
if c == "(":
depth += 1
buf.append(40)
i += 1
continue
if c == ")":
depth -= 1
if depth == 0:
i += 1
break
buf.append(41)
i += 1
continue
buf.append(ord(c) & 0xFF)
i += 1
out.append(bytes(buf).decode("cp1252", errors="replace"))
return out
def _pdf_content_streams(pdf_bytes):
"""The content-stream bodies, in file (= page) order."""
s = pdf_bytes.decode("latin-1")
return re.findall(r"stream\r?\n(.*?)\r?\nendstream", s, re.DOTALL)
def _snapshot_lines(snapshot):
"""The snapshot's lines with the final-newline's trailing '' dropped (one line -> one shown text run)."""
lines = snapshot.split("\n")
if lines and lines[-1] == "":
lines = lines[:-1]
return lines
def _pdf_wrap(s, width):
"""Wrap at a raw char boundary (never a word/space boundary) so the chunks concatenate back to `s`
byte-for-byte — the exact-fidelity discipline (D3/H3/O3), independent of any word model."""
return [s[i:i + width] for i in range(0, len(s), width)] or [""]
def _pdf_marked(snapshot, annotations, chap_l, sec_l, nl_at):
"""The snapshot with a `[<finding_id>]` marker spliced at each anchor locus (descending (offset, fid)
splice, so co-located markers land adjacent `[A][B]` and never perturb a not-yet-inserted offset)."""
inserts = [(_insertion_offset(a.get("anchor") or {}, snapshot, nl_at, chap_l, sec_l), a.get("finding_id"))
for a in annotations]
out = snapshot
for off, fid in sorted(inserts, key=lambda t: (t[0], am.fid_key(t[1]) or ""), reverse=True):
off = max(0, min(off, len(out)))
out = out[:off] + _pdf_marker(fid) + out[off:]
return out
def _pdf_content(page_items):
"""A page content stream. Each item is ('show', text) or ('skip',); every item advances one line (T*),
so a snapshot blank line shows as `() Tj` (round-trippable) while a structural spacer shows nothing."""
out = ["BT", "/F1 %d Tf" % _PDF_FONT_SIZE, "%d TL" % _PDF_LEADING, "%d %d Td" % (_PDF_X, _PDF_Y_START)]
for it in page_items:
out.append("T*")
if it[0] == "show":
out.append("(%s) Tj" % _pdf_escape_string(it[1]))
out.append("ET")
return "\n".join(out) + "\n"
def build_pdf(manifest_obj, snapshot):
"""-> (pdf_bytes_or_None, errs). Pure projection: the snapshot prose as Helvetica text with a
`[<finding_id>]` marker at each locus, then a Findings section of the verbatim comments. errs is the
two-sided precondition (WinAnsi-encodable; no marker sigil already in the snapshot; single-line comment)."""
annotations = [a for a in (manifest_obj.get("annotations") or []) if isinstance(a, dict)]
project = manifest_obj.get("project", "Manuscript")
project = project if isinstance(project, str) else "Manuscript"
title = project + _PDF_TITLE_SUFFIX
errs = []
# Precondition (two-sided, the O1/D discipline): every rendered string must be WinAnsi-encodable, no
# comment may be multi-line (it reads as clutter — forbid it like DOCX), and the snapshot must not
# already carry a manifest marker (else the P2 manifest-keyed strip would not be reversible).
if not _pdf_enc_ok(snapshot):
errs.append("P precondition: the snapshot has a character outside WinAnsi (CP1252) — not renderable "
"in a base-14 Helvetica PDF without an embedded font")
if not _pdf_enc_ok(title):
errs.append("P precondition: the project title has a non-WinAnsi character")
for a in annotations:
c = _pdf_comment_str(a)
if not _pdf_enc_ok(c):
errs.append("P precondition: comment for %s has a non-WinAnsi character" % a.get("finding_id"))
if "\n" in c or "\r" in c:
errs.append("P precondition: comment for %s is multi-line" % a.get("finding_id"))
if _pdf_marker(a.get("finding_id")) in snapshot:
errs.append("P precondition: the snapshot already contains the marker %s — the reverse "
"transform would not be reversible" % _pdf_marker(a.get("finding_id")))
if errs:
return None, errs
chap_n, sec_n, chap_l, sec_l = am.heading_index(snapshot)
nl_at, ln = {}, 1
for i, ch in enumerate(snapshot):
if ch == "\n":
nl_at[ln] = i
ln += 1
marked_lines = _snapshot_lines(_pdf_marked(snapshot, annotations, chap_l, sec_l, nl_at))
ordered = sorted(annotations, key=lambda a: am.fid_key(a.get("finding_id")) or "")
# The flat, deterministic sequence of shown strings: [title] + manuscript(L) + [header] + finding chunks.
# Structural spacers advance the cursor (T*) but show nothing, so they never enter the shown sequence.
items = [("show", title), ("skip",)]
items += [("show", m) for m in marked_lines]
items += [("skip",), ("show", _PDF_FINDINGS_HEADER), ("skip",)]
for a in ordered:
items += [("show", chunk) for chunk in _pdf_wrap(_pdf_comment_str(a), _PDF_WRAP)]
items.append(("skip",))
pages = [items[i:i + _PDF_LINES_PER_PAGE] for i in range(0, len(items), _PDF_LINES_PER_PAGE)] or [[]]
page_count = len(pages)
font_num = 3 + 2 * page_count
objs = {}
objs[1] = b"<< /Type /Catalog /Pages 2 0 R >>"
kids = " ".join("%d 0 R" % (3 + 2 * p) for p in range(page_count))
objs[2] = ("<< /Type /Pages /Kids [%s] /Count %d >>" % (kids, page_count)).encode("ascii")
for p in range(page_count):
page_num, cont_num = 3 + 2 * p, 4 + 2 * p
objs[page_num] = (
"<< /Type /Page /Parent 2 0 R /MediaBox [0 0 %d %d] "
"/Resources << /Font << /F1 %d 0 R >> >> /Contents %d 0 R >>"
% (_PDF_PAGE_W, _PDF_PAGE_H, font_num, cont_num)).encode("ascii")
data = _pdf_content(pages[p]).encode("ascii")
objs[cont_num] = b"<< /Length %d >>\nstream\n" % len(data) + data + b"\nendstream"
objs[font_num] = b"<< /Type /Font /Subtype /Type1 /BaseFont /Helvetica /Encoding /WinAnsiEncoding >>"
buf = bytearray(b"%PDF-1.4\n%\xe2\xe3\xcf\xd3\n") # the binary-comment 2nd line marks the file binary
offsets = {}
for num in range(1, font_num + 1):
offsets[num] = len(buf)
buf += b"%d 0 obj\n" % num + objs[num] + b"\nendobj\n"
xref_off = len(buf)
n_entries = font_num + 1
buf += b"xref\n0 %d\n" % n_entries
buf += b"0000000000 65535 f\r\n"
for num in range(1, font_num + 1):
buf += b"%010d 00000 n\r\n" % offsets[num]
buf += (b"trailer\n<< /Size %d /Root 1 0 R >>\nstartxref\n%d\n%%%%EOF\n" % (n_entries, xref_off))
return bytes(buf), []
def check_pdf(manifest_obj, snapshot, pdf_bytes):
"""P1-P3 over an emitted .pdf. Returns (errs, warns).
P1 (authoritative) — the on-disk .pdf == a fresh deterministic build byte-for-byte (the artifact is
fully determined by the gated manifest + snapshot; equality pins prose, marker positions, comment text,
structure, and forbids any authored content — the DOCX-D1 discipline, for a text-native PDF).
P2/P3 independently parse the PDF bytes as granular diagnostics."""
errs = []
annotations = [a for a in (manifest_obj.get("annotations") or []) if isinstance(a, dict)]
expected, _perrs = build_pdf(manifest_obj, snapshot)
if expected is not None and pdf_bytes != expected:
errs.append("P1 artifact integrity: the on-disk .pdf is not byte-identical to a fresh build from "
"the gated manifest + snapshot (prose / marker-position / comment / structure drift, "
"or authored content)")
shown = []
for cs in _pdf_content_streams(pdf_bytes):
shown += _pdf_string_literals(cs)
ids = [am.fid_key(a.get("finding_id")) for a in annotations]
lines = _snapshot_lines(snapshot)
L = len(lines)
# Structure: shown[0] = title, shown[1:1+L] = manuscript, shown[1+L] = header, shown[2+L:] = findings.
if len(shown) < 2 + L:
return errs + ["P2 text round-trip: the PDF carries too few text runs (%d) to hold the manuscript "
"(%d lines) + the Findings header" % (len(shown), L)], []
manuscript = shown[1:1 + L]
# P2 — text round-trip: join the manuscript runs, strip the manifest-keyed markers, == snapshot.
reconstructed = "\n".join(manuscript) + "\n"
for fid in ids:
reconstructed = reconstructed.replace(_pdf_marker(fid), "")
if reconstructed != snapshot:
errs.append("P2 text round-trip: the PDF manuscript text (markers stripped) does not reproduce the "
"snapshot byte-for-byte")
# P3 — marker resolution: the manuscript markers ↔ the manifest finding_id set (bijection, multiset).
marker_counts = {}
for m in manuscript:
for fid in _PDF_MARKER_RE.findall(m):
marker_counts[fid] = marker_counts.get(fid, 0) + 1
manifest_set = {}
for fid in ids:
manifest_set[fid] = manifest_set.get(fid, 0) + 1
for fid, ncount in sorted(marker_counts.items()):
if ncount != 1:
errs.append("P3 marker resolution: marker [%s] appears %d times (need exactly 1)" % (fid, ncount))
if fid not in manifest_set:
errs.append("P3 marker resolution: marker [%s] is not a manifest finding_id (un-manifested)" % fid)
for fid in sorted(manifest_set):
if marker_counts.get(fid, 0) != 1:
errs.append("P3 marker resolution: manifest finding %s has %d markers (need exactly 1)"
% (fid, marker_counts.get(fid, 0)))
# P3 — comment fidelity: each finding's chunk group (re-derived) concatenates to the verbatim comment.
findings = shown[2 + L:]
pos = 0
for a in sorted(annotations, key=lambda x: am.fid_key(x.get("finding_id")) or ""):
c = _pdf_comment_str(a)
chunks = _pdf_wrap(c, _PDF_WRAP)
got = findings[pos:pos + len(chunks)]
pos += len(chunks)
if "".join(got) != c:
errs.append("P3 comment fidelity: the comment for %s is not the verbatim manifest comment "
"(relocate, never re-author)" % a.get("finding_id"))
return errs, []
def _runlabel_of(path):
base = os.path.basename(path or "")
for infix in ("_Manuscript_Snapshot_", "_Annotation_Manifest_"):
if infix in base:
return os.path.splitext(base.split(infix)[-1])[0] or "run"
return os.path.splitext(base)[0] or "run"
def _resolve(folder):
"""-> (manifest_obj_or_None, snapshot_or_None, project, runlabel, crosslinked_text_or_None, err)."""
man = _newest(glob.glob(os.path.join(folder, am._MANIFEST_GLOB)))
snap = _newest(glob.glob(os.path.join(folder, am._SNAPSHOT_GLOB)))
if not man:
return None, None, None, None, None, "no %s in %s" % (am._MANIFEST_GLOB, folder)
if not snap:
return None, None, None, None, None, "no %s in %s" % (am._SNAPSHOT_GLOB, folder)
obj, merrs = am.parse_manifest(_read(man))
# A present-but-non-object manifest block (a JSON array / scalar) parses to a non-None, non-dict
# obj; guarding on `obj is None` alone let it reach obj.get("project") below — and the build_* /
# check_* manifest_obj.get() sites downstream — and crash with AttributeError. Treat ANY non-dict
# as an invalid manifest so every public generate/validate path fails cleanly, not with a traceback.
if not isinstance(obj, dict):
return None, None, None, None, None, "manifest invalid — %s" % (
merrs[0] if merrs else "manifest block is not a JSON object")
snapshot = am.normalize_snapshot(_read(snap) or "")
project = os.path.basename(snap).split("_Manuscript_Snapshot_")[0] or obj.get("project", "Manuscript")
# The crosslinked letter (Increment 2 input) — optional; its presence enables the letter projection.
cl_path = _newest(glob.glob(os.path.join(folder, _CROSSLINKED_GLOB)))
crosslinked = _read(cl_path) if cl_path else None
return obj, snapshot, project, _runlabel_of(snap), crosslinked, None
def generate(folder):
"""Write obsidian/<copy> (and, when a crosslinked letter is present, obsidian/<letter> + forward
wikilinks in the copy — Increment 2). Returns (code, lines)."""
obj, snapshot, project, runlabel, crosslinked, err = _resolve(folder)
if err:
return 2, ["obsidian-export: %s" % err]
copy_basename = "%s_Annotated_Manuscript_%s" % (project, runlabel)
letter_basename = "%s_Crosslinked_Letter_%s" % (project, runlabel)
letter_bn = letter_basename if crosslinked else None
copy_text, errs = build_obsidian(obj, snapshot, letter_basename=letter_bn)
if errs:
return 1, ["obsidian-export: " + e for e in errs] + ["obsidian-export: FAIL (O1 precondition)"]
outdir = os.path.join(folder, "obsidian")
os.makedirs(outdir, exist_ok=True)
with open(os.path.join(outdir, copy_basename + ".md"), "w", encoding="utf-8", newline="") as fh:
fh.write(copy_text)
wrote = [copy_basename + ".md"]
if crosslinked:
letter_text, lerrs, _lwarns = build_obsidian_letter(crosslinked, obj, snapshot, copy_basename)
if lerrs:
return 1, ["obsidian-export: " + e for e in lerrs] + ["obsidian-export: FAIL (O5 precondition)"]
with open(os.path.join(outdir, letter_basename + ".md"), "w", encoding="utf-8", newline="") as fh:
fh.write(letter_text)
wrote.append(letter_basename + ".md")
return 0, ["obsidian-export: wrote " + ", ".join("obsidian/" + w for w in wrote)]
def run(paths, strict=False):
if len(paths) < 1 or not os.path.isdir(paths[0]):
return 2, ["obsidian-export: usage: obsidian-export <run_folder>"]
obj, snapshot, _p, _r, crosslinked, err = _resolve(paths[0])
if err:
return 2, ["obsidian-export: %s" % err]
# Validate the ON-DISK export artifacts, never a regenerate — so a hand-edited copy/letter is caught
# (mirrors how annotated-manuscript gates the committed copy's bytes — the A2 discipline).
copy_path = _newest(glob.glob(os.path.join(paths[0], "obsidian", "*_Annotated_Manuscript_*.md")))
if not copy_path:
return 2, ["obsidian-export: no obsidian/*_Annotated_Manuscript_*.md found "
"(run `annotation_export.py obsidian <run_folder>` first)"]
copy_text = _read(copy_path)
if copy_text is None:
return 2, ["obsidian-export: cannot read %s" % copy_path]
copy_basename = os.path.splitext(os.path.basename(copy_path))[0]
letter_path = _newest(glob.glob(os.path.join(paths[0], "obsidian", _CROSSLINKED_GLOB)))
letter_basename = os.path.splitext(os.path.basename(letter_path))[0] if letter_path else None
# The two-sided build precondition the artifacts must have satisfied.
_expected, perrs = build_obsidian(obj, snapshot, letter_basename=letter_basename)
if perrs:
return 1, ["obsidian-export: " + e for e in perrs] + ["obsidian-export: FAIL (O1 precondition)"]
errs, _w = check_obsidian(obj, snapshot, copy_text, letter_basename=letter_basename)
gates = "O1 round-trip + O2 footnote resolution + O3 comment fidelity"
if letter_path and not crosslinked:
# An on-disk Obsidian letter exists but its gated crosslinked source is absent, so O4/O5 cannot run.
# Refuse to PASS rather than silently skip — else deleting the gated source and leaving a tampered
# obsidian/ letter in place would ride through unvalidated (a firewall bypass).
errs.append("an exported Obsidian letter (obsidian/%s) is present but its gated crosslinked source "
"(%s) is absent — refusing to PASS an unvalidatable letter"
% (os.path.basename(letter_path), _CROSSLINKED_GLOB))
elif letter_path and crosslinked:
letter_text = _read(letter_path) or ""
e_l, _wl = check_obsidian_letter(crosslinked, letter_text, copy_text, copy_basename,
letter_basename, obj, snapshot)
errs = errs + e_l
gates += " + O4 link resolution + O5 letter prose fidelity"
lines = ["obsidian-export: %d finding(s); validating obsidian/%s%s"
% (len(obj.get("annotations") or []), os.path.basename(copy_path),
" + " + os.path.basename(letter_path) if letter_path else "")]
for e in errs:
lines.append(" ERROR: %s" % e)
if errs:
lines.append("obsidian-export: FAIL (%d error(s))" % len(errs))
return 1, lines
lines.append("obsidian-export: PASS (%s)" % gates)
return 0, lines
def generate_html(folder):
"""Write html/<Project>_Annotated_Manuscript_<runlabel>.html. Returns (code, lines)."""
obj, snapshot, project, runlabel, _cl, err = _resolve(folder)
if err:
return 2, ["html-export: %s" % err]
html, errs = build_html(obj, snapshot)
if errs:
return 1, ["html-export: " + e for e in errs] + ["html-export: FAIL (H1 precondition)"]
outdir = os.path.join(folder, "html")
os.makedirs(outdir, exist_ok=True)
out_path = os.path.join(outdir, "%s_Annotated_Manuscript_%s.html" % (project, runlabel))
with open(out_path, "w", encoding="utf-8", newline="") as fh:
fh.write(html)
return 0, ["html-export: wrote html/%s" % os.path.basename(out_path)]
def run_html(paths):
"""Validate the ON-DISK html/<copy>.html (H1-H3), never a regenerate."""
if len(paths) < 1 or not os.path.isdir(paths[0]):
return 2, ["html-export: usage: html-export <run_folder>"]
obj, snapshot, _p, _r, _cl, err = _resolve(paths[0])
if err:
return 2, ["html-export: %s" % err]
html_path = _newest(glob.glob(os.path.join(paths[0], "html", "*_Annotated_Manuscript_*.html")))
if not html_path:
return 2, ["html-export: no html/*_Annotated_Manuscript_*.html found "
"(run `annotation_export.py html <run_folder>` first)"]
html_text = _read(html_path)
if html_text is None:
return 2, ["html-export: cannot read %s" % html_path]
_exp, perrs = build_html(obj, snapshot)
if perrs:
return 1, ["html-export: " + e for e in perrs] + ["html-export: FAIL (H1 precondition)"]
errs, _w = check_html(obj, snapshot, html_text)
lines = ["html-export: %d finding(s); validating html/%s"
% (len(obj.get("annotations") or []), os.path.basename(html_path))]
for e in errs:
lines.append(" ERROR: %s" % e)
if errs:
lines.append("html-export: FAIL (%d error(s))" % len(errs))
return 1, lines
lines.append("html-export: PASS (H1 round-trip + H2 anchor resolution + H3 comment fidelity)")
return 0, lines
def _read_bytes(path):
try:
with open(path, "rb") as fh:
return fh.read()
except OSError:
return None
def generate_docx(folder):
"""Write docx/<Project>_Annotated_Manuscript_<runlabel>.docx. Returns (code, lines)."""
obj, snapshot, project, runlabel, _cl, err = _resolve(folder)
if err:
return 2, ["docx-export: %s" % err]
docx, errs = build_docx(obj, snapshot)
if errs:
return 1, ["docx-export: " + e for e in errs] + ["docx-export: FAIL (D precondition)"]
outdir = os.path.join(folder, "docx")
os.makedirs(outdir, exist_ok=True)
out_path = os.path.join(outdir, "%s_Annotated_Manuscript_%s.docx" % (project, runlabel))
with open(out_path, "wb") as fh:
fh.write(docx)
return 0, ["docx-export: wrote docx/%s" % os.path.basename(out_path)]
def run_docx(paths):
"""Validate the ON-DISK docx/<copy>.docx (D1-D3), never a regenerate."""
if len(paths) < 1 or not os.path.isdir(paths[0]):
return 2, ["docx-export: usage: docx-export <run_folder>"]
obj, snapshot, _p, _r, _cl, err = _resolve(paths[0])
if err:
return 2, ["docx-export: %s" % err]
docx_path = _newest(glob.glob(os.path.join(paths[0], "docx", "*_Annotated_Manuscript_*.docx")))
if not docx_path:
return 2, ["docx-export: no docx/*_Annotated_Manuscript_*.docx found "
"(run `annotation_export.py docx <run_folder>` first)"]
docx_bytes = _read_bytes(docx_path)
if docx_bytes is None:
return 2, ["docx-export: cannot read %s" % docx_path]
_exp, perrs = build_docx(obj, snapshot)
if perrs:
return 1, ["docx-export: " + e for e in perrs] + ["docx-export: FAIL (D precondition)"]
errs, _w = check_docx(obj, snapshot, docx_bytes)
lines = ["docx-export: %d finding(s); validating docx/%s"
% (len(obj.get("annotations") or []), os.path.basename(docx_path))]
for e in errs:
lines.append(" ERROR: %s" % e)
if errs:
lines.append("docx-export: FAIL (%d error(s))" % len(errs))
return 1, lines
lines.append("docx-export: PASS (D1 artifact integrity + D2 text round-trip + D3 comment resolution)")
return 0, lines
def generate_pdf(folder):
"""Write pdf/<Project>_Annotated_Manuscript_<runlabel>.pdf. Returns (code, lines)."""
obj, snapshot, project, runlabel, _cl, err = _resolve(folder)
if err:
return 2, ["pdf-export: %s" % err]
pdf, errs = build_pdf(obj, snapshot)
if errs:
return 1, ["pdf-export: " + e for e in errs] + ["pdf-export: FAIL (P precondition)"]
outdir = os.path.join(folder, "pdf")
os.makedirs(outdir, exist_ok=True)
out_path = os.path.join(outdir, "%s_Annotated_Manuscript_%s.pdf" % (project, runlabel))
with open(out_path, "wb") as fh:
fh.write(pdf)
return 0, ["pdf-export: wrote pdf/%s" % os.path.basename(out_path)]
def run_pdf(paths):
"""Validate the ON-DISK pdf/<copy>.pdf (P1-P3), never a regenerate."""
if len(paths) < 1 or not os.path.isdir(paths[0]):
return 2, ["pdf-export: usage: pdf-export <run_folder>"]
obj, snapshot, _p, _r, _cl, err = _resolve(paths[0])
if err:
return 2, ["pdf-export: %s" % err]
pdf_path = _newest(glob.glob(os.path.join(paths[0], "pdf", "*_Annotated_Manuscript_*.pdf")))
if not pdf_path:
return 2, ["pdf-export: no pdf/*_Annotated_Manuscript_*.pdf found "
"(run `annotation_export.py pdf <run_folder>` first)"]
pdf_bytes = _read_bytes(pdf_path)
if pdf_bytes is None:
return 2, ["pdf-export: cannot read %s" % pdf_path]
_exp, perrs = build_pdf(obj, snapshot)
if perrs:
return 1, ["pdf-export: " + e for e in perrs] + ["pdf-export: FAIL (P precondition)"]
errs, _w = check_pdf(obj, snapshot, pdf_bytes)
lines = ["pdf-export: %d finding(s); validating pdf/%s"
% (len(obj.get("annotations") or []), os.path.basename(pdf_path))]
for e in errs:
lines.append(" ERROR: %s" % e)
if errs:
lines.append("pdf-export: FAIL (%d error(s))" % len(errs))
return 1, lines
lines.append("pdf-export: PASS (P1 artifact integrity + P2 text round-trip + P3 comment resolution)")
return 0, lines
# ---------------------------------------------------------------- self-test
def run_self_test():
import json as _j
import tempfile
import shutil
rc = {"v": 0}
def chk(name, cond):
print(" %s: %s" % (name, "OK" if cond else "FAIL"))
if not cond:
rc["v"] = 1
snap = am.normalize_snapshot(
"# Chapter 1\n"
"The lighthouse stood unlit for forty years.\n"
"She counted the hours.\n"
"# Chapter 9\n"
"Three days collapsed here.\n")
quote = "The lighthouse stood unlit for forty years."
qs = snap.find(quote)
def ann(fid, anchor, comment):
return {"finding_id": fid, "anchor": anchor, "comment": comment}
obj = {"schema": am._SCHEMA_ID, "project": "T", "runlabel": "r",
"snapshot_path": "T_Manuscript_Snapshot_r.md", "snapshot_sha256": am.sha256(snap),
"snapshot_line_count": am.line_count(snap),
"annotations": [
ann("F-QT-01", {"kind": "quote", "value": "%d-%d" % (qs, qs + len(quote)), "quote": quote},
"[Must-Fix · F-QT-01] flat reveal — fix class: stage it. (See letter §F-QT-01.)"),
ann("F-CH-01", {"kind": "chapter", "value": "Ch 9"},
"[Should-Fix · F-CH-01] pacing seam — fix class: add a beat. (See letter §F-CH-01.)"),
ann("F-DOC-01", {"kind": "document", "value": ""},
"[Could-Fix · F-DOC-01] soft opening — fix class: sharpen. (See letter §F-DOC-01.)"),
ann("F-NEG-01", {"kind": "chapter", "value": "Ch 1"},
"[Should-Fix · F-NEG-01] POV wobble — fix class: hold POV. (See letter §F-NEG-01.)"),
]}
text, errs = build_obsidian(obj, snap)
chk("build_no_precondition_errs", not errs and text is not None)
chk("ref_at_quote_end", (quote + "[^F-QT-01]") in text)
chk("ref_on_chapter_line", "# Chapter 9[^F-CH-01]" in text)
# F-DOC-01 (document, end of line 1) + F-NEG-01 (Ch 1, end of line 1) are CO-LOCATED -> adjacent refs.
chk("co_located_adjacent", "# Chapter 1[^F-DOC-01][^F-NEG-01]" in text)
chk("definition_block_verbatim",
"[^F-QT-01]: [Must-Fix · F-QT-01] flat reveal — fix class: stage it. (See letter §F-QT-01.)" in text)
# O1 round-trip: strip refs + def block -> snapshot.
chk("round_trip_to_snapshot", reverse_obsidian(text, [a["finding_id"] for a in obj["annotations"]]) == snap)
# the validator passes on a clean export.
e, _w = check_obsidian(obj, snap, text)
chk("check_clean", e == [])
# O1 fires on prose mutation (tamper a body char).
tampered = text.replace("Three days collapsed here.", "Three days collapsed HERE.")
e, _w = check_obsidian(obj, snap, tampered)
chk("o1_fires_on_mutation", any("O1 round-trip" in x for x in e))
# O3 fires on a re-authored comment.
reauth = text.replace("flat reveal", "REWRITTEN")
e, _w = check_obsidian(obj, snap, reauth)
chk("o3_fires_on_reauthor", any("O3 comment fidelity" in x for x in e))
# regression: a TRUTHY non-string comment (a JSON list/dict, kept as-is by parse_manifest) must
# not crash the O3 line. `or ""` only neutralizes a FALSY comment; pre-fix `list + str` raised
# TypeError on the edited line. build_obsidian str()-coerces the comment (survives), so the copy
# reaches check_obsidian; the malformed comment must surface as a clean O3 error, never a traceback.
for _badc in (["a", "b"], {"k": "v"}):
obj_bc = {"schema": am._SCHEMA_ID, "project": "T", "runlabel": "r",
"snapshot_path": "T_Manuscript_Snapshot_r.md", "snapshot_sha256": am.sha256(snap),
"snapshot_line_count": am.line_count(snap),
"annotations": [ann("F-X-01", {"kind": "chapter", "value": "Ch 1"}, _badc)]}
_tbc, _ebc = build_obsidian(obj_bc, snap)
try:
e_bc, _wbc = check_obsidian(obj_bc, snap, _tbc)
chk("o3_nonstring_comment_no_crash_%s" % type(_badc).__name__,
_tbc is not None and any("O3 comment fidelity" in x for x in e_bc))
except TypeError:
chk("o3_nonstring_comment_no_crash_%s" % type(_badc).__name__, False)
# sibling regression: a non-string chapter ANCHOR value (unhashable) must not crash build_obsidian
# via _insertion_offset -> anchor_line, nor check_obsidian via _heading_text_of. Pre-fix it raised
# TypeError: unhashable type at chap_l.get(val).
obj_bv = {"schema": am._SCHEMA_ID, "project": "T", "runlabel": "r",
"snapshot_path": "T_Manuscript_Snapshot_r.md", "snapshot_sha256": am.sha256(snap),
"snapshot_line_count": am.line_count(snap),
"annotations": [ann("F-X-02", {"kind": "chapter", "value": [1, 2]}, "c")]}
try:
_tbv, _ebv = build_obsidian(obj_bv, snap)
check_obsidian(obj_bv, snap, _tbv)
chk("nonstring_chapter_value_no_crash", _tbv is not None)
except TypeError:
chk("nonstring_chapter_value_no_crash", False)
# sibling regression (finding_id): a non-hashable finding_id (list) must not crash the (offset, fid)
# ref sort / the definition-block sort in build_obsidian, nor the O2 `manifest_set[fid]` keying in
# check_obsidian; and a mixed int/str id set must not crash those sorts. am.fid_key normalizes both.
obj_bf = {"schema": am._SCHEMA_ID, "project": "T", "runlabel": "r",
"snapshot_path": "T_Manuscript_Snapshot_r.md", "snapshot_sha256": am.sha256(snap),
"snapshot_line_count": am.line_count(snap),
"annotations": [ann([1, 2], {"kind": "chapter", "value": "Ch 1"}, "c"),
ann("F-S-01", {"kind": "chapter", "value": "Ch 1"}, "c2")]}
try:
_tbf, _ebf = build_obsidian(obj_bf, snap)
check_obsidian(obj_bf, snap, _tbf)
chk("nonhashable_mixed_finding_id_no_crash", _tbf is not None)
except TypeError:
chk("nonhashable_mixed_finding_id_no_crash", False)
# O2 fires on an un-manifested (authored) footnote ref+def smuggled in.
smuggled = text.replace("Three days collapsed here.", "Three days collapsed here.[^F-EVIL-01]")
smuggled = smuggled.rstrip("\n") + "\n[^F-EVIL-01]: authored note\n"
e, _w = check_obsidian(obj, snap, smuggled)
chk("o2_fires_on_unmanifested", any("O2 footnote resolution" in x and "F-EVIL-01" in x for x in e))
# O1 precondition: a snapshot already carrying '[^' is refused.
bad_snap = am.normalize_snapshot("# Ch 1\nan array[^index] reference\n")
_t, perrs = build_obsidian({"annotations": [ann("F-X-01", {"kind": "document", "value": ""}, "c")]}, bad_snap)
chk("precondition_snapshot_sigil", any("snapshot already contains" in x for x in perrs))
# O1 precondition: a comment carrying '[^' is refused.
_t, perrs = build_obsidian({"annotations": [ann("F-X-01", {"kind": "document", "value": ""}, "see [^1]")]}, snap)
chk("precondition_comment_sigil", any("comment for F-X-01 contains" in x for x in perrs))
# O1 precondition: a multi-line comment is refused (a footnote definition is one line).
_t, perrs = build_obsidian({"annotations": [ann("F-X-01", {"kind": "document", "value": ""}, "line one\nline two")]}, snap)
chk("precondition_comment_multiline", any("is multi-line" in x for x in perrs))
# determinism.
chk("deterministic", build_obsidian(obj, snap)[0] == build_obsidian(obj, snap)[0])
# empty manifest -> copy is the snapshot, round-trips trivially.
t0, _e = build_obsidian({"annotations": []}, snap)
chk("empty_is_snapshot", t0 == snap)
# --- Increment 2: the letter cross-links ---
copy_bn, letter_bn = "T_Annotated_Manuscript_r", "T_Crosslinked_Letter_r"
copy2, _e = build_obsidian(obj, snap, letter_basename=letter_bn)
chk("inc2_copy_forward_wikilink",
"(See letter §F-CH-01.) [[T_Crosslinked_Letter_r#^F-CH-01|→ letter §F-CH-01]]" in copy2)
chk("inc2_copy_round_trips",
reverse_obsidian(copy2, [a["finding_id"] for a in obj["annotations"]]) == snap)
chk("inc2_copy_o3_clean", check_obsidian(obj, snap, copy2, letter_basename=letter_bn)[0] == [])
# O3 fires if the forward wikilink is tampered (authored text in the definition).
chk("inc2_o3_fires_on_bad_wikilink",
any("O3" in x for x in check_obsidian(obj, snap, copy2.replace("→ letter §F-CH-01", "→ EVIL"), letter_basename=letter_bn)[0]))
crosslinked = ("# T — Editorial Letter\n\n## What Needs Work\n\n"
"The reveal lands flat. <!-- finding: F-QT-01 -->{>>→ marked-up copy: F-QT-01 @ quote:%d-%d<<}\n\n"
"The pacing collapses in chapter nine. <!-- finding: F-CH-01 -->{>>→ marked-up copy: F-CH-01 @ chapter:Ch 9<<}\n\n"
"The opening is soft. <!-- finding: F-DOC-01 -->{>>→ marked-up copy: F-DOC-01 @ document:<<}\n\n"
"The POV wobbles. <!-- finding: F-NEG-01 -->{>>→ marked-up copy: F-NEG-01 @ chapter:Ch 1<<}\n"
% (qs, qs + len(quote)))
letter2, lerrs, lwarns = build_obsidian_letter(crosslinked, obj, snap, copy_bn)
chk("inc2_letter_no_precond_errs", not lerrs)
chk("inc2_letter_block_id", "<!-- finding: F-CH-01 -->[[T_Annotated_Manuscript_r#Chapter 9]] ^F-CH-01" in letter2)
chk("inc2_letter_heading_text_not_token", "#Chapter 9]]" in letter2 and "#Ch 9]]" not in letter2)
chk("inc2_letter_file_level_doc", "[[T_Annotated_Manuscript_r]] ^F-DOC-01" in letter2 and len(lwarns) >= 1)
ids = [a["finding_id"] for a in obj["annotations"]]
chk("inc2_o5_prose_fidelity", strip_obsidian_letter(letter2, ids, copy_bn) == am.reverse_transform(crosslinked))
chk("inc2_o4_o5_clean",
check_obsidian_letter(crosslinked, letter2, copy2, copy_bn, letter_bn, obj, snap)[0] == [])
# Codex #109 (THE reproduced firewall hole): an INJECTED `[[Injected_Note]]` in the on-disk letter — with
# no whitespace delta — used to PASS because the wildcard `_WIKILINK_RE.sub("")` stripped it before the
# prose compare. The authoritative O5 exact-build equality (and the now manifest-keyed strip) must catch it.
# NO leading space — the exact zero-whitespace-delta injection from the bug report, so the OLD wildcard
# strip would have removed it cleanly (reproducing the prose, a false PASS); only the authoritative lock
# + the manifest-keyed strip catch it. (A leading space would let the old strip leave a residual and pass
# for the wrong reason — a vacuous test.)
_injected = letter2.replace("# T — Editorial Letter", "# T — Editorial Letter[[Injected_Note]]", 1)
chk("inc2_o5_catches_injected_wikilink",
any("O5" in x for x in check_obsidian_letter(crosslinked, _injected, copy2, copy_bn, letter_bn, obj, snap)[0]))
chk("inc2_o5_keyed_strip_keeps_foreign_wikilink", # the keyed strip leaves a foreign wikilink in place
strip_obsidian_letter(_injected, ids, copy_bn) != am.reverse_transform(crosslinked))
# Codex #109 analog on the COPY: an injected orphan trailing `[^X]: …` definition (no in-body ref) used
# to slip past O1's trailing-block strip + O2 (a def with no ref is not a manifest id). Authoritative O1
# exact-build equality catches it.
_orphan = copy2.rstrip("\n") + "\n[^F-EVIL-02]: an authored orphan definition\n"
chk("inc2_copy_catches_orphan_definition",
any("O1 copy integrity" in x for x in check_obsidian(obj, snap, _orphan, letter_basename=letter_bn)[0]))
# O4 forward fires when a letter block id is missing (copy forward link dangles).
chk("inc2_o4_forward_fires",
any("O4" in x for x in check_obsidian_letter(crosslinked, letter2.replace(" ^F-CH-01", ""),
copy2, copy_bn, letter_bn, obj, snap)[0]))
# O5 fires on letter prose mutation.
chk("inc2_o5_fires_on_mutation",
any("O5" in x for x in check_obsidian_letter(crosslinked, letter2.replace("pacing collapses", "pacing COLLAPSES"),
copy2, copy_bn, letter_bn, obj, snap)[0]))
# O5 precondition: an editorial letter already containing '[[' is refused.
_t, lerrs2, _w = build_obsidian_letter(crosslinked.replace("The reveal lands flat.", "See [[elsewhere]]."),
obj, snap, copy_bn)
chk("inc2_o5_precondition_wikilink", any("already contains '[['" in x for x in lerrs2))
# P2-2: O4-reverse is PER-FINDING — a real-but-wrong heading fails (set-membership would pass).
chk("inc2_o4_reverse_wrong_heading_fires",
any("O4" in x for x in check_obsidian_letter(
crosslinked, letter2.replace("#Chapter 9]]", "#Chapter 1]]"), copy2, copy_bn, letter_bn, obj, snap)[0]))
# P3: a smuggled `|display` label on a reverse link fails (the expected link has no `|`).
chk("inc2_o4_reverse_display_label_fires",
any("O4" in x for x in check_obsidian_letter(
crosslinked, letter2.replace("#Chapter 9]]", "#Chapter 9|SMUGGLED]]"), copy2, copy_bn, letter_bn, obj, snap)[0]))
# P2-1: a section heading with an unsafe char (`]`) degrades to a file-level link (not a malformed one).
snap_br = am.normalize_snapshot("# Orientation [draft]\nProse here.\n")
obj_br = {"annotations": [ann("F-SEC-01", {"kind": "section", "value": "Orientation [draft]"},
"[Should-Fix · F-SEC-01] x — fix class: y. (See letter §F-SEC-01.)")]}
cl_br = "# L\n\nThing. <!-- finding: F-SEC-01 -->{>>→ marked-up copy: F-SEC-01 @ section:Orientation [draft]<<}\n"
lt_br, _e, w_br = build_obsidian_letter(cl_br, obj_br, snap_br, "C")
chk("inc2_unsafe_heading_file_level", "[[C]] ^F-SEC-01" in lt_br and "[[C#" not in lt_br and len(w_br) >= 1)
# generate() end-to-end from a run folder.
d = tempfile.mkdtemp()
try:
with open(os.path.join(d, "T_Manuscript_Snapshot_r.md"), "w", encoding="utf-8", newline="") as fh:
fh.write(snap)
with open(os.path.join(d, "T_Annotation_Manifest_r.md"), "w", encoding="utf-8", newline="") as fh:
fh.write("<!-- apodictic:annotation\n%s\n-->" % _j.dumps(obj))
chk("generate_writes", generate(d)[0] == 0 and os.path.isfile(
os.path.join(d, "obsidian", "T_Annotated_Manuscript_r.md")))
chk("run_validates", run([d])[0] == 0)
# run() must validate the ON-DISK copy: tampering the emitted file is caught (not a regenerate).
copy_p = os.path.join(d, "obsidian", "T_Annotated_Manuscript_r.md")
good = open(copy_p, encoding="utf-8").read()
open(copy_p, "w", encoding="utf-8", newline="").write(good.replace("Three days collapsed here.", "Three days collapsed THERE."))
chk("run_catches_disk_prose_mutation", run([d])[0] == 1)
open(copy_p, "w", encoding="utf-8", newline="").write(good.replace("pacing seam", "AN INVENTED CLAIM"))
chk("run_catches_disk_comment_reauthor", run([d])[0] == 1)
open(copy_p, "w", encoding="utf-8", newline="").write(
good.replace("Three days collapsed here.", "Three days collapsed here.[^F-EVIL-01]").rstrip("\n")
+ "\n[^F-EVIL-01]: authored note\n")
chk("run_catches_disk_unmanifested_footnote", run([d])[0] == 1)
open(copy_p, "w", encoding="utf-8", newline="").write(good)
chk("run_passes_after_restore", run([d])[0] == 0)
# a missing on-disk export is a usage error, not a false PASS.
os.remove(copy_p)
chk("run_no_copy_is_usage", run([d])[0] == 2)
finally:
shutil.rmtree(d, ignore_errors=True)
# Increment 2 end-to-end: a run folder WITH a crosslinked letter -> generate writes copy + letter,
# run validates both (O1-O5), and tampering the on-disk letter is caught.
d2 = tempfile.mkdtemp()
try:
with open(os.path.join(d2, "T_Manuscript_Snapshot_r.md"), "w", encoding="utf-8", newline="") as fh:
fh.write(snap)
with open(os.path.join(d2, "T_Annotation_Manifest_r.md"), "w", encoding="utf-8", newline="") as fh:
fh.write("<!-- apodictic:annotation\n%s\n-->" % _j.dumps(obj))
with open(os.path.join(d2, "T_Crosslinked_Letter_r.md"), "w", encoding="utf-8", newline="") as fh:
fh.write(crosslinked)
code, _l = generate(d2)
chk("inc2_generate_both", code == 0
and os.path.isfile(os.path.join(d2, "obsidian", "T_Annotated_Manuscript_r.md"))
and os.path.isfile(os.path.join(d2, "obsidian", "T_Crosslinked_Letter_r.md")))
chk("inc2_run_validates_both", run([d2])[0] == 0)
letter_p = os.path.join(d2, "obsidian", "T_Crosslinked_Letter_r.md")
good_l = open(letter_p, encoding="utf-8").read()
open(letter_p, "w", encoding="utf-8", newline="").write(good_l.replace("pacing collapses", "pacing COLLAPSES"))
chk("inc2_run_catches_letter_mutation", run([d2])[0] == 1)
# Codex #109 end-to-end: inject a `[[Injected_Note]]` (zero whitespace delta) into the EXPORTED
# letter file and run the validator — it must FAIL (it returned PASS before the authoritative O5 lock).
open(letter_p, "w", encoding="utf-8", newline="").write(
good_l.replace("# T — Editorial Letter", "# T — Editorial Letter[[Injected_Note]]", 1))
chk("inc2_run_catches_disk_injected_wikilink", run([d2])[0] == 1)
# Codex #109 P2: deleting the gated crosslinked SOURCE while leaving the tampered obsidian/ letter
# must NOT silently skip O4/O5 — it must FAIL (a present-but-unvalidatable letter is refused).
src_p = os.path.join(d2, "T_Crosslinked_Letter_r.md")
os.remove(src_p)
chk("inc2_run_refuses_letter_without_gated_source", run([d2])[0] == 1)
open(src_p, "w", encoding="utf-8", newline="").write(crosslinked) # restore the gated source
open(letter_p, "w", encoding="utf-8", newline="").write(good_l)
chk("inc2_run_passes_after_restore", run([d2])[0] == 0)
finally:
shutil.rmtree(d2, ignore_errors=True)
# --- Increment 3: read-only HTML ---
html, herrs = build_html(obj, snap)
chk("html_no_precond_errs", not herrs and html is not None)
chk("html_one_pre", len(_PRE_RE.findall(html)) == 1)
chk("html_marker_at_anchor",
'<sup class="fnref" id="ref-F-QT-01"><a href="#fn-F-QT-01">[F-QT-01]</a></sup>' in html)
chk("html_finding_li",
'<li id="fn-F-CH-01">[Should-Fix · F-CH-01] pacing seam — fix class: add a beat. (See letter §F-CH-01.) '
'<a class="backref" href="#ref-F-CH-01">↩</a></li>' in html)
chk("html_check_clean", check_html(obj, snap, html)[0] == [])
chk("html_round_trip", reverse_html(_PRE_RE.findall(html)[0], [a["finding_id"] for a in obj["annotations"]]) == snap)
chk("html_h1_fires_on_mutation",
any("H1" in x for x in check_html(obj, snap, html.replace("Three days collapsed here.", "Three days THERE."))[0]))
chk("html_h3_fires_on_reauthor",
any("H3" in x for x in check_html(obj, snap, html.replace("pacing seam", "INVENTED"))[0]))
chk("html_h2_fires_on_unmanifested",
any("H2" in x for x in check_html(obj, snap, html.replace(
"</ol>", '<li id="fn-F-EVIL-01">x <a class="backref" href="#ref-F-EVIL-01">↩</a></li>\n</ol>'))[0]))
# H1-equality (authoritative) catches authored prose OUTSIDE the <pre>/findings (the standalone gate
# was weaker than the whole-document lock — build-review P2).
chk("html_h1_fires_on_authored_prose_outside_pre",
any("H1 artifact integrity" in x for x in check_html(
obj, snap, html.replace("</main>", "</main>\n<p>EDITOR-INVENTED PROSE</p>"))[0]))
# H1-equality catches marker OFFSET-DRIFT (a literal-replace round-trip would not — build-review P2).
_drift = html.replace('forty years.<sup class="fnref" id="ref-F-QT-01">',
'forty<sup class="fnref" id="ref-F-QT-01"> years.', 1) \
if 'forty years.<sup class="fnref" id="ref-F-QT-01">' in html else html
chk("html_h1_fires_on_offset_drift",
_drift == html or any("H1" in x for x in check_html(obj, snap, _drift)[0]))
# HOSTILE (mandated): HTML metachars `& < >` UPSTREAM of an anchor (catches offset drift) + literal
# `& <` in a comment (catches double-escape + the exact inverse) — the canonical fixture has none.
q = "The lighthouse stood unlit for forty years."
snap_h = am.normalize_snapshot("# A & B <tag> here\n" + q + "\n")
qh = snap_h.find(q)
obj_h = {"project": "H & <co>", "annotations": [ann(
"F-QT-01", {"kind": "quote", "value": "%d-%d" % (qh, qh + len(q)), "quote": q},
"[Must-Fix · F-QT-01] the line uses & and <x> markup — fix class: stage it. (See letter §F-QT-01.)")]}
html_h, _e = build_html(obj_h, snap_h)
chk("html_hostile_round_trips", reverse_html(_PRE_RE.findall(html_h)[0], ["F-QT-01"]) == snap_h)
chk("html_hostile_check_clean", check_html(obj_h, snap_h, html_h)[0] == [])
chk("html_hostile_escapes_prose", "# A & B <tag> here" in html_h)
chk("html_hostile_escapes_comment", "uses & and <x> markup" in html_h)
chk("html_hostile_no_offset_drift", (q + '<sup class="fnref" id="ref-F-QT-01">') in html_h)
# generate_html + run_html end-to-end; on-disk tampering caught.
d3 = tempfile.mkdtemp()
try:
with open(os.path.join(d3, "T_Manuscript_Snapshot_r.md"), "w", encoding="utf-8", newline="") as fh:
fh.write(snap)
with open(os.path.join(d3, "T_Annotation_Manifest_r.md"), "w", encoding="utf-8", newline="") as fh:
fh.write("<!-- apodictic:annotation\n%s\n-->" % _j.dumps(obj))
chk("html_generate_writes", generate_html(d3)[0] == 0
and os.path.isfile(os.path.join(d3, "html", "T_Annotated_Manuscript_r.html")))
chk("html_run_validates", run_html([d3])[0] == 0)
hp = os.path.join(d3, "html", "T_Annotated_Manuscript_r.html")
gh = open(hp, encoding="utf-8").read()
open(hp, "w", encoding="utf-8", newline="").write(gh.replace("Three days collapsed here.", "Three days THERE."))
chk("html_run_catches_disk_mutation", run_html([d3])[0] == 1)
open(hp, "w", encoding="utf-8", newline="").write(gh)
chk("html_run_passes_after_restore", run_html([d3])[0] == 0)
finally:
shutil.rmtree(d3, ignore_errors=True)
# --- Increment 4: DOCX (→ Google Docs) ---
docx_bytes, derrs = build_docx(obj, snap)
chk("docx_no_precond_errs", not derrs and docx_bytes is not None)
chk("docx_is_zip", docx_bytes[:2] == b"PK")
chk("docx_deterministic", build_docx(obj, snap)[0] == docx_bytes) # byte-identical across builds
chk("docx_check_clean", check_docx(obj, snap, docx_bytes)[0] == [])
z = zipfile.ZipFile(io.BytesIO(docx_bytes))
docxml = z.read("word/document.xml").decode("utf-8")
cxml = z.read("word/comments.xml").decode("utf-8")
chk("docx_seven_parts", len(z.namelist()) == 7)
chk("docx_comment_anchored",
'<w:commentRangeStart w:id="0"/>' in docxml and '<w:commentReference w:id="0"/>' in docxml
and '<w:commentRangeEnd w:id="0"/>' in docxml)
chk("docx_comment_verbatim", "pacing seam — fix class: add a beat. (See letter §F-CH-01.)" in cxml)
paras = [_html_unescape_exact("".join(_WT_RE.findall(p))) for p in _WP_RE.findall(docxml)]
chk("docx_text_round_trip", ("\n".join(paras) + "\n") == snap)
# Codex #111: co-located ranges (F-DOC-01 wid 1 + F-NEG-01 wid 2 both span line 1) must NEST, not cross:
# opens ascending (1 then 2), closes reverse-open (2 then 1) -> start1 start2 … end2 ref2 end1 ref1.
_nested = ('<w:commentRangeStart w:id="1"/><w:commentRangeStart w:id="2"/>' in docxml
and ('<w:commentRangeEnd w:id="2"/><w:r><w:commentReference w:id="2"/></w:r>'
'<w:commentRangeEnd w:id="1"/><w:r><w:commentReference w:id="1"/></w:r>') in docxml)
_crossing = ('<w:commentRangeEnd w:id="1"/><w:r><w:commentReference w:id="1"/></w:r>'
'<w:commentRangeEnd w:id="2"/>') in docxml
chk("docx_colocated_ranges_nest_not_cross", _nested and not _crossing)
# Comment w:date is the run's OWN date (manifest runlabel) at noon UTC — deterministic (not wall-clock)
# and no timezone day-rollback. The main obj's runlabel "r" has no date -> fixed fallback; a YYYY-MM-DD
# runlabel -> that date at noon.
chk("docx_date_fallback", ('w:date="%s"' % _DOCX_DATE_FALLBACK) in cxml)
# Use a date DISTINCT from the fallback so this proves DERIVATION (a stub that always returns the
# fallback would otherwise pass — the fallback literal coincides with the 2026-01-01 fixture date).
def _cxml_for(runlabel):
b = build_docx({"runlabel": runlabel, "annotations": obj["annotations"]}, snap)[0]
return zipfile.ZipFile(io.BytesIO(b)).read("word/comments.xml").decode("utf-8")
chk("docx_date_from_runlabel",
'w:date="2025-07-04T12:00:00Z"' in _cxml_for("2025-07-04")
and _DOCX_DATE_FALLBACK not in _cxml_for("2025-07-04"))
chk("docx_date_with_suffix_runlabel", # `YYYY-MM-DD` prefix is honored even with a trailing suffix
'w:date="2025-07-04T12:00:00Z"' in _cxml_for("2025-07-04-v2"))
chk("docx_date_rejects_invalid", # shape-valid but impossible date -> fallback, never a bad dateTime
_DOCX_DATE_FALLBACK in _cxml_for("2026-13-45") and "2026-13-45" not in _cxml_for("2026-13-45"))
chk("docx_date_deterministic", _cxml_for("2025-07-04") == _cxml_for("2025-07-04"))
def _rezip_with(override):
return _deterministic_zip([(n, override.get(n, z.read(n).decode("utf-8"))) for n in z.namelist()])
chk("docx_d1d2_fire_on_body_mutation",
any("D1" in x or "D2" in x for x in check_docx(obj, snap, _rezip_with(
{"word/document.xml": docxml.replace("Three days collapsed here.", "Three days COLLAPSED here.")}))[0]))
chk("docx_d3_fires_on_reauthor",
any("D1" in x or "D3" in x for x in check_docx(obj, snap, _rezip_with(
{"word/comments.xml": cxml.replace("pacing seam", "INVENTED")}))[0]))
# D3 multiset (build-review P3): a DUPLICATED range id must fail D3's resolution check standalone
# (a set-based check would collapse it). Inject a second commentRangeStart id=0.
_dup = docxml.replace('<w:commentRangeStart w:id="0"/>',
'<w:commentRangeStart w:id="0"/><w:commentRangeStart w:id="0"/>', 1)
chk("docx_d3_multiset_catches_dup_id",
any("D3 comment resolution" in x for x in check_docx(obj, snap, _rezip_with({"word/document.xml": _dup}))[0]))
_t, dperrs = build_docx(
{"annotations": [ann("F-X-01", {"kind": "document", "value": ""}, "line\nline2")]}, snap)
chk("docx_precond_multiline", any("multi-line" in x for x in dperrs))
d4 = tempfile.mkdtemp()
try:
with open(os.path.join(d4, "T_Manuscript_Snapshot_r.md"), "w", encoding="utf-8", newline="") as fh:
fh.write(snap)
with open(os.path.join(d4, "T_Annotation_Manifest_r.md"), "w", encoding="utf-8", newline="") as fh:
fh.write("<!-- apodictic:annotation\n%s\n-->" % _j.dumps(obj))
chk("docx_generate_writes", generate_docx(d4)[0] == 0
and os.path.isfile(os.path.join(d4, "docx", "T_Annotated_Manuscript_r.docx")))
chk("docx_run_validates", run_docx([d4])[0] == 0)
dp = os.path.join(d4, "docx", "T_Annotated_Manuscript_r.docx")
with open(dp, "wb") as fh:
fh.write(_rezip_with({"word/document.xml": docxml.replace("Three days collapsed here.", "X")}))
chk("docx_run_catches_disk_tamper", run_docx([d4])[0] == 1)
finally:
shutil.rmtree(d4, ignore_errors=True)
# --- Increment 5: PDF (proofing target) ---
pdf_bytes, pperrs = build_pdf(obj, snap)
chk("pdf_no_precond_errs", not pperrs and pdf_bytes is not None)
chk("pdf_is_pdf", pdf_bytes[:5] == b"%PDF-")
chk("pdf_deterministic", build_pdf(obj, snap)[0] == pdf_bytes) # byte-identical across builds
chk("pdf_check_clean", check_pdf(obj, snap, pdf_bytes)[0] == [])
_pshown = []
for _cs in _pdf_content_streams(pdf_bytes):
_pshown += _pdf_string_literals(_cs)
_Lp = len(_snapshot_lines(snap))
_pman = _pshown[1:1 + _Lp]
chk("pdf_marker_on_chapter_line", "# Chapter 9[F-CH-01]" in _pman)
chk("pdf_marker_at_quote_end", any("forty years.[F-QT-01]" in m for m in _pman))
# F-DOC-01 (document, end of line 1) + F-NEG-01 (Ch 1, end of line 1) are CO-LOCATED -> adjacent markers.
chk("pdf_colocated_adjacent", "# Chapter 1[F-DOC-01][F-NEG-01]" in _pman)
_precon = "\n".join(_pman) + "\n"
for _fid in [a["finding_id"] for a in obj["annotations"]]:
_precon = _precon.replace("[%s]" % _fid, "")
chk("pdf_round_trip", _precon == snap)
chk("pdf_comment_verbatim_extractable",
"flat reveal" in "".join(_pshown[2 + _Lp:]) and "pacing seam" in "".join(_pshown[2 + _Lp:]))
# P1/P2 fire on a manuscript prose mutation (equal-length byte edit keeps xref valid so P2/P3 can parse).
_pmut = pdf_bytes.replace(b"Three days collapsed here.", b"Three days collapsed THERE.")
chk("pdf_p1_fires_on_mutation",
_pmut != pdf_bytes and any("P1" in x or "P2" in x for x in check_pdf(obj, snap, _pmut)[0]))
# P1/P3 fire on a re-authored comment.
_preauth = pdf_bytes.replace(b"flat reveal", b"REWRITTEN!!")
chk("pdf_p3_fires_on_reauthor",
_preauth != pdf_bytes and any("P1" in x or "P3" in x for x in check_pdf(obj, snap, _preauth)[0]))
# P precondition: a snapshot already carrying a manifest marker is refused (irreversible strip).
_t, _pe = build_pdf({"annotations": [ann("F-QT-01", {"kind": "document", "value": ""}, "c")]},
am.normalize_snapshot("# Ch\nsee [F-QT-01] here\n"))
chk("pdf_precond_marker_in_snapshot", any("already contains the marker" in x for x in _pe))
# P precondition: a non-WinAnsi (CP1252) character in a comment is refused (no embedded font).
_t, _pe = build_pdf({"annotations": [ann("F-X-01", {"kind": "document", "value": ""}, "bad \U0001f600 char")]}, snap)
chk("pdf_precond_non_winansi", any("non-WinAnsi" in x for x in _pe))
# P precondition: a multi-line comment is refused.
_t, _pe = build_pdf({"annotations": [ann("F-X-01", {"kind": "document", "value": ""}, "line\nline2")]}, snap)
chk("pdf_precond_multiline", any("multi-line" in x for x in _pe))
# HOSTILE (mandated, like HTML/DOCX): PDF-string metachars `( ) \` + non-ASCII `§` UPSTREAM of an
# anchor (catches offset/escape drift) and in a comment (catches the exact escape/unescape inverse) —
# the canonical fixture has none.
_pq = "The lighthouse stood unlit for forty years."
_psnap = am.normalize_snapshot("# A (paren) \\slash § here\n" + _pq + "\n")
_pqi = _psnap.find(_pq)
_pobj = {"project": "P (co) §", "annotations": [ann(
"F-QT-01", {"kind": "quote", "value": "%d-%d" % (_pqi, _pqi + len(_pq)), "quote": _pq},
"[Must-Fix · F-QT-01] uses ( ) \\ and § chars — fix. (See letter §F-QT-01.)")]}
_ppdf, _pe = build_pdf(_pobj, _psnap)
chk("pdf_hostile_no_precond", not _pe and _ppdf is not None)
chk("pdf_hostile_check_clean", check_pdf(_pobj, _psnap, _ppdf)[0] == [])
_hshown = []
for _cs in _pdf_content_streams(_ppdf):
_hshown += _pdf_string_literals(_cs)
_hman = _hshown[1:1 + len(_snapshot_lines(_psnap))]
chk("pdf_hostile_round_trips", ("\n".join(_hman) + "\n").replace("[F-QT-01]", "") == _psnap)
# generate_pdf + run_pdf end-to-end; on-disk tampering caught.
d6 = tempfile.mkdtemp()
try:
with open(os.path.join(d6, "T_Manuscript_Snapshot_r.md"), "w", encoding="utf-8", newline="") as fh:
fh.write(snap)
with open(os.path.join(d6, "T_Annotation_Manifest_r.md"), "w", encoding="utf-8", newline="") as fh:
fh.write("<!-- apodictic:annotation\n%s\n-->" % _j.dumps(obj))
chk("pdf_generate_writes", generate_pdf(d6)[0] == 0
and os.path.isfile(os.path.join(d6, "pdf", "T_Annotated_Manuscript_r.pdf")))
chk("pdf_run_validates", run_pdf([d6])[0] == 0)
_pp = os.path.join(d6, "pdf", "T_Annotated_Manuscript_r.pdf")
_gp = _read_bytes(_pp)
with open(_pp, "wb") as fh:
fh.write(_gp.replace(b"Three days collapsed here.", b"Three days collapsed THERE."))
chk("pdf_run_catches_disk_tamper", run_pdf([d6])[0] == 1)
with open(_pp, "wb") as fh:
fh.write(_gp)
chk("pdf_run_passes_after_restore", run_pdf([d6])[0] == 0)
finally:
shutil.rmtree(d6, ignore_errors=True)
# regression (Codex #141 round-2): a present-but-non-object manifest block (a JSON array) must not
# reach obj.get() in _resolve / the build_* manifest_obj.get() sites — every public generate
# entrypoint returns a clean non-zero error, never an AttributeError traceback.
d5 = tempfile.mkdtemp()
try:
with open(os.path.join(d5, "T_Manuscript_Snapshot_r.md"), "w", encoding="utf-8", newline="") as fh:
fh.write(snap)
with open(os.path.join(d5, "T_Annotation_Manifest_r.md"), "w", encoding="utf-8", newline="") as fh:
fh.write("<!-- apodictic:annotation\n[1, 2, 3]\n-->")
try:
nonobj_ok = (generate(d5)[0] != 0 and generate_html(d5)[0] != 0 and generate_docx(d5)[0] != 0
and generate_pdf(d5)[0] != 0)
except AttributeError:
nonobj_ok = False
chk("nonobject_manifest_no_crash", nonobj_ok)
finally:
shutil.rmtree(d5, ignore_errors=True)
# sibling regression (HTML / DOCX / letter): a non-hashable finding_id (a JSON list/object) must not
# crash the HTML marker sort + H2/H3 dict-keys, the DOCX idmap dict-key/lookup, nor the letter
# anchor_of dict-keys — the build_obsidian guard generalized to every export surface. am.fid_key SSoT.
obj_nh = {"schema": am._SCHEMA_ID, "project": "T", "runlabel": "r",
"snapshot_path": "T_Manuscript_Snapshot_r.md", "snapshot_sha256": am.sha256(snap),
"snapshot_line_count": am.line_count(snap),
"annotations": [ann([1, 2], {"kind": "chapter", "value": "Ch 1"}, "c"),
ann("F-S-01", {"kind": "chapter", "value": "Ch 1"}, "c2")]}
try:
_h, _he = build_html(obj_nh, snap)
check_html(obj_nh, snap, _h or "")
_d, _de = build_docx(obj_nh, snap)
check_docx(obj_nh, snap, _d or b"")
_pn, _pne = build_pdf(obj_nh, snap)
check_pdf(obj_nh, snap, _pn or b"")
chk("html_docx_nonhashable_finding_id_no_crash", _h is not None and _d is not None and _pn is not None)
except TypeError:
chk("html_docx_nonhashable_finding_id_no_crash", False)
try:
_lt, _le, _lw = build_obsidian_letter(crosslinked, obj_nh, snap, copy_bn)
check_obsidian_letter(crosslinked, _lt or "", copy2, copy_bn, letter_bn, obj_nh, snap)
chk("letter_nonhashable_finding_id_no_crash", _lt is not None)
except TypeError:
chk("letter_nonhashable_finding_id_no_crash", False)
print("Self-test: %s" % ("PASS" if rc["v"] == 0 else "FAIL"))
return rc["v"]
def main(argv):
if "--self-test" in argv:
return run_self_test()
if am is None:
print("obsidian-export: annotation_manifest unavailable (same-dir import failed)")
return 2
args = [a for a in argv[1:] if not a.startswith("--")]
# generate modes
for verb, gen in (("obsidian", generate), ("html", generate_html), ("docx", generate_docx),
("pdf", generate_pdf)):
if args and args[0] == verb:
rest = [a for a in args[1:]]
if len(rest) != 1 or not os.path.isdir(rest[0]):
print("Usage: annotation_export.py %s <run_folder>" % verb)
return 2
code, lines = gen(rest[0])
for ln in lines:
print(ln)
return code
# validate modes
for verb, val in (("html-export", run_html), ("docx-export", run_docx), ("pdf-export", run_pdf)):
if args and args[0] == verb:
code, lines = val(args[1:])
for ln in lines:
print(ln)
return code
paths = [a for a in args if a != "obsidian-export"]
if not paths:
print("Usage: annotation_export.py obsidian|html|docx|pdf <run_folder> | "
"obsidian-export|html-export|docx-export|pdf-export <run_folder> | --self-test")
return 2
code, lines = run(paths, strict="--strict" in argv)
for ln in lines:
print(ln)
return code
if __name__ == "__main__":
sys.exit(main(sys.argv))
SHA-256: 37d65a67d464dd87a24ed7d1b51061fce9ff59198340f1ddcfa0f00bc35e9016