feat(okf): in-import cross-link graph — extract/resolve/reject links, dangling-link signal (T5a/A, TDD, +10)

src/llm_ingestion_guard/okf.py

@@ -39,15 +39,20 @@ __all__ = [
     "stamp_concept",
     "format_log_entry",
     "import_bundle",
+    "extract_link_targets",
+    "resolve_link",
+    "link_graph",
     "Origin",
     "Channel",
     "ProvenanceStamp",
     "ConceptResult",
     "BundleResult",
+    "LinkGraphResult",
     "OKFError",
     "OKFFrontmatterError",
     "OKFPathError",
     "OKFResourceError",
+    "OKFLinkError",
 ]
 
 _FENCE = "---"
@@ -76,6 +81,10 @@ class OKFResourceError(OKFError):
     """A ``resource`` URL is not on the https allowlist."""
 
 
+class OKFLinkError(OKFError):
+    """A cross-link target is unsafe (dangerous scheme or bundle escape)."""
+
+
 _URL_SCHEME_RE = re.compile(r"^([A-Za-z][A-Za-z0-9+.\-]*):")
 
 
@@ -369,6 +378,122 @@ def _most_severe(dispositions):
     return worst
 
 
+# --- T5a / A: cross-link graph (in-import) -----------------------------------
+# The persisted cross-run graph (B) that would catch "plant a link now, write
+# the poisoned target in a LATER run" (§7.2) is deferred to stream 2, where the
+# consumer that owns the corpus decides where the durable graph state lives.
+# This in-import graph resolves links within a single bundle merge.
+
+_MD_LINK_RE = re.compile(r"\[[^\]]*\]\(\s*([^)\s]+)")
+# Active-content schemes are refused in a link, mirroring the resource gate (T3).
+_DANGEROUS_LINK_SCHEMES = frozenset({"javascript", "data", "vbscript", "file", "blob"})
+
+
+@dataclass(frozen=True)
+class LinkGraphResult:
+    """In-import cross-link resolution over one bundle.
+
+    ``dangling`` — ``(from_id, target_concept_id)`` for in-bundle ``.md`` links
+    whose target concept is **not present** in the bundle: the dormant-injection
+    signal of §7.2 (a link planted to a not-yet-written concept). ``rejected`` —
+    ``(from_id, target, reason)`` for links refused outright (dangerous scheme or
+    bundle escape). ``resolved`` — ``(from_id, target_concept_id)`` for links to
+    concepts present in the bundle.
+    """
+
+    dangling: tuple
+    rejected: tuple
+    resolved: tuple
+
+
+def extract_link_targets(body):
+    """Return the destinations of markdown ``[text](target)`` links in ``body``."""
+    return _MD_LINK_RE.findall(body)
+
+
+def resolve_link(target, from_concept_id):
+    """Resolve one link target to an in-bundle concept-ID, or reject it (T5a).
+
+    Returns the target concept-ID for an in-bundle ``.md`` link (bundle-absolute
+    ``/x.md`` or relative ``./x.md`` / ``../y.md``, resolved against the linking
+    concept's directory). Returns ``None`` for an external ``http(s)``/other
+    non-active link (not a concept edge) and for non-``.md`` targets. Raises
+    :class:`OKFLinkError` for an active-content scheme or a ``..`` escape past the
+    bundle root.
+    """
+    candidate = target.strip().split("#", 1)[0].split("?", 1)[0]
+    if not candidate:
+        return None
+
+    scheme_match = _URL_SCHEME_RE.match(candidate)
+    if scheme_match:
+        scheme = scheme_match.group(1).lower()
+        if scheme in _DANGEROUS_LINK_SCHEMES:
+            raise OKFLinkError("link uses a dangerous scheme %r: %r" % (scheme, target))
+        return None  # external (http/https/mailto/…): not an in-bundle concept edge
+
+    if not candidate.endswith(".md"):
+        return None  # not a concept-document link (asset, anchor, …)
+
+    if candidate.startswith("/"):
+        normalized = _normalize_bundle_path(candidate[1:])
+    else:
+        from_dir = from_concept_id.rsplit("/", 1)[0] if "/" in from_concept_id else ""
+        joined = from_dir + "/" + candidate if from_dir else candidate
+        normalized = _normalize_bundle_path(joined)
+
+    return normalized[: -len(".md")]
+
+
+def link_graph(bundle):
+    """Resolve every cross-link in ``bundle`` against the concepts it contains.
+
+    ``bundle`` maps concept path to document text (as :func:`import_bundle`). Only
+    the body is scanned for links. See :class:`LinkGraphResult` for the outcome.
+    """
+    present = {p[: -len(".md")] for p in bundle if p.endswith(".md")}
+    dangling, rejected, resolved = [], [], []
+
+    for path in sorted(bundle):
+        if not path.endswith(".md"):
+            continue
+        from_id = path[: -len(".md")]
+        try:
+            _frontmatter, body = parse_frontmatter(bundle[path])
+        except OKFFrontmatterError:
+            body = bundle[path]  # unparseable frontmatter is T2's reject, not ours
+
+        for target in extract_link_targets(body):
+            try:
+                concept_id = resolve_link(target, from_id)
+            except OKFLinkError as exc:
+                rejected.append((from_id, target, str(exc)))
+                continue
+            if concept_id is None:
+                continue
+            if concept_id in present:
+                resolved.append((from_id, concept_id))
+            else:
+                dangling.append((from_id, concept_id))
+
+    return LinkGraphResult(tuple(dangling), tuple(rejected), tuple(resolved))
+
+
+def _normalize_bundle_path(path):
+    """Normalize a ``/``-separated bundle path; raise if it escapes the root."""
+    parts = []
+    for segment in path.split("/"):
+        if segment in ("", "."):
+            continue
+        if segment == "..":
+            if not parts:
+                raise OKFLinkError("link target escapes the bundle root: %r" % path)
+            parts.pop()
+        else:
+            parts.append(segment)
+    return "/".join(parts)
+
+
 def _parse_flat(fm_lines):
     result = {}
     i = 0

tests/test_okf.py

@@ -26,11 +26,15 @@ from llm_ingestion_guard.okf import (
     trust_for,
     format_log_entry,
     import_bundle,
+    extract_link_targets,
+    resolve_link,
+    link_graph,
     Origin,
     Channel,
     OKFFrontmatterError,
     OKFPathError,
     OKFResourceError,
+    OKFLinkError,
 )
 from llm_ingestion_guard.report import Report
 from llm_ingestion_guard.disposition import Trust, Disposition
@@ -378,3 +382,65 @@ def test_import_bundle_records_origin_channel_on_stamp():
     assert stamp.trust is Trust.TRUSTED
     assert stamp.origin is Origin.INTERNAL
     assert stamp.channel is Channel.MANUAL
+
+
+# --- T5a/A: cross-link extraction, target validation, in-import resolution ----
+# OKF links are markdown `.md` paths, bundle-absolute (`/x.md`, recommended) or
+# relative (`./x.md`); verified against SPEC.md. In-import graph only (A); the
+# cross-run persisted graph (B) is deferred to stream 2 (see docs/PLAN.md).
+
+def test_extract_link_targets_pulls_markdown_destinations():
+    body = "See [users](/tables/users.md) and [orders](./orders.md) for detail."
+    assert extract_link_targets(body) == ["/tables/users.md", "./orders.md"]
+
+
+def test_resolve_link_bundle_absolute_to_concept_id():
+    assert resolve_link("/tables/customers.md", "docs/intro") == "tables/customers"
+
+
+def test_resolve_link_relative_to_concept_id():
+    assert resolve_link("./other.md", "tables/users") == "tables/other"
+
+
+def test_resolve_link_relative_parent_stays_in_bundle():
+    assert resolve_link("../ops/runbook.md", "tables/users") == "ops/runbook"
+
+
+def test_resolve_link_external_https_is_not_a_concept_edge():
+    assert resolve_link("https://example.com/page", "tables/users") is None
+
+
+def test_resolve_link_rejects_dangerous_scheme():
+    with pytest.raises(OKFLinkError):
+        resolve_link("javascript:alert(1)", "tables/users")
+
+
+def test_resolve_link_rejects_bundle_escape():
+    with pytest.raises(OKFLinkError):
+        resolve_link("../../etc/passwd.md", "tables/users")
+
+
+def test_link_graph_flags_dangling_link():
+    # a/main links to a not-yet-existent b/target -> dormant-injection signal (§7.2)
+    bundle = {
+        "a/main.md": "---\ntype: t\n---\nSee [later](/b/target.md).\n",
+        "a/other.md": "---\ntype: t\n---\nNothing linked.\n",
+    }
+    graph = link_graph(bundle)
+    assert ("a/main", "b/target") in graph.dangling
+
+
+def test_link_graph_resolves_present_target():
+    bundle = {
+        "a/main.md": "---\ntype: t\n---\nSee [here](/b/target.md).\n",
+        "b/target.md": "---\ntype: t\n---\nThe target concept.\n",
+    }
+    graph = link_graph(bundle)
+    assert ("a/main", "b/target") in graph.resolved
+    assert graph.dangling == ()
+
+
+def test_link_graph_records_rejected_dangerous_link():
+    bundle = {"a/main.md": "---\ntype: t\n---\n[x](javascript:alert(1))\n"}
+    graph = link_graph(bundle)
+    assert any(from_id == "a/main" for from_id, _target, _reason in graph.rejected)