feat: authorized one-hop graph expansion with server-bounded examination (#151)

STATUS.md

@@ -170,6 +170,7 @@ Follow the ADR for its exact evidence boundary.
 | [0086](./docs/decisions/0086-report-worker-batches-and-compose-source-wide-cycles.md) | Emit privacy-shaped worker batch progress and compose source-wide scan/status from existing exact Control calls |
 | [0091](./docs/decisions/0091-reconcile-connector-acl-freshness-at-acceptance.md) | Admit the credential-free Feishu Docs twin through the leased connector runner, preserve lease-reconciled ACL freshness provenance, and atomically isolate or fix one Article ACL with a database Policy Epoch advance |
 | [0092](./docs/decisions/0092-authorize-feishu-subjects-and-bound-mirrored-freshness.md) | Recompute Feishu subject grants from engine-owned mappings, bind artifacts to exact Articles, and expire Mirrored Feishu evidence under one closed five-minute Runtime profile |
+| [0093](./docs/decisions/0093-activate-leased-rich-markdown-and-revision-link-graph.md) | Activate lease-selected rich Markdown v3 publication and one bounded authorized Revision-graph hop |
 
 ADR-0065 extends the active File Provider boundary from a flat root to
 deterministic recursive discovery of canonical nested Markdown paths. Each
@@ -197,8 +198,8 @@ bounded File scan over an explicitly configured anchored root, accept every new
 provider page, schedule only changed upserts, reconcile accepted current-scan
 upsert pages missing durable jobs, and hand those jobs to the existing
 autonomous worker. Real-PostgreSQL fixture evidence covers exact unchanged
-replay, interrupted scheduling recovery, one-note addition, aggregate
-compilation refusal, delete observation without delete execution, and
+replay, interrupted scheduling recovery, one-note addition, delete observation
+without delete execution, and
 384-dimensional Fragment publication. This does not claim that the maintainer's
 private corpus has run; it activates no watcher, alternate publisher, new
 tombstone authority, or network operation.
@@ -240,6 +241,17 @@ This does **not** activate a live Feishu client, tenant credentials, external
 network calls, production subject-mapping administration, or Feishu delivery.
 Those surfaces remain `NOT_ACTIVE`.
 
+ADR-0093 activates the exact File-import worker's rich Markdown v3 compiler
+subprocess and immutable content-free Revision link edges. Runtime follows one
+outgoing or backlink hop only from authorized main-path projections, verifies
+same-Article/current-Revision lineage before inheritance, re-authorizes every
+cross-Article candidate through the unchanged Kernel, and admits only relevant
+authorized neighbours into the existing ranking competition. Registered
+PostgreSQL and generated-SDK evidence proves a denied neighbour leaves no
+tenant-visible or retained decision trace. Recursive graph traversal,
+historical edge backfill, external-URI expansion, and graph-carried authority
+remain `NOT_ACTIVE`.
+
 ### Wire contract, SDK, and trusted delivery
 
 | ADR | Activates |

adapters/http/ui_api.py

@@ -37,7 +37,6 @@
 from engine.persistence.role_guard import assert_control_role, assert_runtime_role
 from engine.runtime.actor import CurrentMembershipVerification
 from engine.supply import (
-    ACTIVE_FILE_IMPORT_MARKDOWN_CONFIG_VERSION,
     CompilationFailure,
     MarkdownCompilerConfig,
     ParsedDocument,
@@ -739,7 +738,7 @@ def preview_import(
             raw = roots.read(FileRootRef(root_ref), import_path)
             outcome = compile_markdown(
                 raw,
-                MarkdownCompilerConfig(ACTIVE_FILE_IMPORT_MARKDOWN_CONFIG_VERSION),
+                MarkdownCompilerConfig("markdown-config-v1"),
             )
         except (LookupError, RuntimeError, TypeError, ValueError):
             raise UiApiUnavailable from None
@@ -761,7 +760,7 @@ def preview_import(
             ).encode("utf-8")
         ).hexdigest()
         payload: dict[str, object] = {
-            "configVersion": ACTIVE_FILE_IMPORT_MARKDOWN_CONFIG_VERSION,
+            "configVersion": "markdown-config-v1",
             "contentLength": len(raw),
             "contentSha256": hashlib.sha256(raw).hexdigest(),
             "fragmentDigest": fragment_digest,
@@ -805,7 +804,7 @@ def confirm_import(
                 or len(fragment_digest) != 64
                 or type(content_length) is not int
                 or content_length < 0
-                or config_version != ACTIVE_FILE_IMPORT_MARKDOWN_CONFIG_VERSION
+                or config_version != "markdown-config-v1"
             ):
                 raise ValueError
         except (KeyError, TypeError, ValueError):

applications/control.py

@@ -574,7 +574,6 @@ def _scan_report_document(report: FileScanReport) -> dict[str, object]:
     return {
         "advancedCursor": report.advanced_cursor,
         "changesAccepted": report.changes_accepted,
-        "compilationRefusals": report.compilation_refusals,
         "deletesObserved": report.deletes_observed,
         "importsScheduled": report.imports_scheduled,
         "pathsObserved": report.paths_observed,
@@ -606,9 +605,6 @@ def _multi_scan_report_json(report: MultiSourceScanReport) -> str:
                 "changesAccepted": sum(
                     source.changes_accepted for source in sources
                 ),
-                "compilationRefusals": sum(
-                    source.compilation_refusals for source in sources
-                ),
                 "deletesObserved": sum(
                     source.deletes_observed for source in sources
                 ),

applications/file_scan.py

@@ -14,7 +14,6 @@
 from sqlalchemy import Engine
 
 from adapters.file_source import FileChangeProvider, FileRootRegistry
-from adapters.parsers.markdown import compile_markdown
 from applications.file_root_configuration import required_environment
 from applications.operator_authentication import (
     CONTROL_OPERATOR_SECRET_ENV,
@@ -50,11 +49,6 @@
     SourceRef,
 )
 from engine.persistence import PostgreSQLControlStore
-from engine.supply import (
-    ACTIVE_FILE_IMPORT_MARKDOWN_CONFIG_VERSION,
-    CompilationFailure,
-    MarkdownCompilerConfig,
-)
 
 PROVIDER_SIGNING_KEY_ENV = "CONTEXT_ENGINE_FILE_CHANGE_PROVIDER_SIGNING_KEY_HEX"
 CHECKPOINT_SIGNING_KEY_ENV = "CONTEXT_ENGINE_FILE_CHANGE_CHECKPOINT_SIGNING_KEY_HEX"
@@ -76,7 +70,6 @@ class FileScanReport:
     changes_accepted: int
     imports_scheduled: int
     deletes_observed: int
-    compilation_refusals: int
     advanced_cursor: str | None
     scan_bound: int
 
@@ -140,7 +133,6 @@ def scan_file_source(
         source_ref=source_ref,
     )
     imports_scheduled = 0
-    compilation_refusals = 0
     reconciled_page_refs: set[str] = set()
     for pending in progress.pending_change_schedules:
         scheduled = _schedule_page(
@@ -154,16 +146,14 @@ def scan_file_source(
             audience=audience,
         )
         imports_scheduled += len(scheduled.changes)
-        compilation_refusals += sum(
-            _compilation_refused(
+        for scheduled_change in scheduled.changes:
+            _verify_accepted_content_identity(
                 roots,
                 manifest,
-                change.path.value,
-                change.content_sha256,
-                change.content_length,
+                scheduled_change.path.value,
+                scheduled_change.content_sha256,
+                scheduled_change.content_length,
             )
-            for change in scheduled.changes
-        )
         reconciled_page_refs.add(pending.page_ref)
     source = FileChangeSource(
         organization_id,
@@ -233,7 +223,6 @@ def scan_file_source(
                 changes_accepted=0,
                 imports_scheduled=imports_scheduled,
                 deletes_observed=0,
-                compilation_refusals=compilation_refusals,
                 advanced_cursor=baseline.reference.checkpoint_ref,
                 scan_bound=baseline.reference.scan_bound,
             )
@@ -279,16 +268,14 @@ def scan_file_source(
                 in {candidate.path.value for candidate in novel_upserts}
             }
             imports_scheduled += len(scheduled_changes)
-            compilation_refusals += sum(
-                _compilation_refused(
+            for path, scheduled_change in scheduled_changes.items():
+                _verify_accepted_content_identity(
                     roots,
                     manifest,
                     path,
-                    change.content_sha256,
-                    change.content_length,
+                    scheduled_change.content_sha256,
+                    scheduled_change.content_length,
                 )
-                for path, change in scheduled_changes.items()
-            )
         advanced_cursor = accepted.checkpoint_ref
         if accepted.next_cursor is None:
             break
@@ -300,7 +287,6 @@ def scan_file_source(
         changes_accepted=changes_accepted,
         imports_scheduled=imports_scheduled,
         deletes_observed=deletes_observed,
-        compilation_refusals=compilation_refusals,
         advanced_cursor=advanced_cursor,
         scan_bound=roots._limits.max_baseline_entries,
     )
@@ -505,13 +491,13 @@ def _replays_complete_baseline(
     )
 
 
-def _compilation_refused(
+def _verify_accepted_content_identity(
     roots: FileRootRegistry,
     manifest: SourceManifest,
     path: str,
     expected_sha256: str,
     expected_length: int,
-) -> int:
+) -> None:
     try:
         payload = roots.read(
             manifest.active_version.root_ref,
@@ -524,8 +510,6 @@ def _compilation_refused(
         or hashlib.sha256(payload).hexdigest() != expected_sha256
     ):
         raise SourceScanRefused
-    outcome = compile_markdown(
-        payload,
-        MarkdownCompilerConfig(ACTIVE_FILE_IMPORT_MARKDOWN_CONFIG_VERSION),
-    )
-    return int(type(outcome) is CompilationFailure)
+    # Production rich compilation is owned by the exact leased Supply worker.
+    # Scan retains only the accepted-byte identity preflight and cannot invoke
+    # or predict the runner's durable refusal classification.

applications/leased_compiler_runner.py

@@ -0,0 +1,74 @@
+"""Pure rich-Markdown transform selected by an exact leased Supply worker."""
+
+from __future__ import annotations
+
+import argparse
+import base64
+import json
+import sys
+from typing import Never, cast
+
+from adapters.parsers.ragflow_markdown import compile_rich_markdown
+from engine.supply.markdown import (
+    CompilationFailure,
+    CompilationFailureCode,
+    CompilationOutcome,
+    MarkdownCompilerConfig,
+    ParsedDocument,
+    canonicalize_parsed_document,
+)
+
+
+class _ClosedArgumentParser(argparse.ArgumentParser):
+    def error(self, message: str) -> Never:
+        raise SystemExit("leased compiler runner arguments are invalid")
+
+
+def _boundary_failure() -> CompilationFailure:
+    return CompilationFailure(
+        code=CompilationFailureCode.UNSUPPORTED_DOCUMENT_SHAPE,
+        position=None,
+    )
+
+
+def _failure_document(failure: CompilationFailure) -> dict[str, object]:
+    return {"code": failure.code.value}
+
+
+def _emit(outcome: CompilationOutcome) -> None:
+    if type(outcome) is ParsedDocument:
+        envelope: dict[str, object] = {
+            "outcome": "parsed",
+            "document": base64.b64encode(
+                canonicalize_parsed_document(outcome)
+            ).decode("ascii"),
+        }
+    else:
+        assert type(outcome) is CompilationFailure
+        envelope = {"outcome": "failure", "failure": _failure_document(outcome)}
+    sys.stdout.write(json.dumps(envelope, sort_keys=True, separators=(",", ":")))
+
+
+def main() -> None:
+    parser = _ClosedArgumentParser(description=__doc__)
+    parser.add_argument("--compile-leased", action="store_true")
+    parser.add_argument("--config", required=True)
+    parser.add_argument("--token-ceiling", required=True, type=int)
+    args = parser.parse_args()
+    if not args.compile_leased:
+        raise SystemExit("leased compiler runner arguments are invalid")
+    try:
+        outcome = compile_rich_markdown(
+            sys.stdin.buffer.read(),
+            MarkdownCompilerConfig(
+                cast(str, args.config),
+                token_ceiling=cast(int, args.token_ceiling),
+            ),
+        )
+    except Exception:
+        outcome = _boundary_failure()
+    _emit(outcome)
+
+
+if __name__ == "__main__":
+    main()