From 50a220022fecc30f4fcf3f0346db639542abfd00 Mon Sep 17 00:00:00 2001 From: stone Date: Fri, 31 Jul 2026 07:28:02 +0800 Subject: [PATCH 01/15] feat: authorize one-hop revision graph expansion --- applications/file_scan.py | 40 +- applications/leased_compiler_runner.py | 86 +++ engine/persistence/configuration.py | 1 + engine/persistence/file_imports.py | 21 +- engine/persistence/membership_context.py | 62 ++ .../persistence/schema_security_manifest.yaml | 205 +++++- engine/runtime/authorized_ranking.py | 54 +- engine/runtime/construction.py | 165 ++++- engine/runtime/materialized.py | 67 ++ engine/supply/compiler_runner.py | 112 +++ engine/supply/link_graph.py | 162 +++++ engine/supply/markdown.py | 2 +- .../20260731_0045_revision_link_graph.py | 676 ++++++++++++++++++ scripts/provision_database_roles.py | 23 +- 14 files changed, 1638 insertions(+), 38 deletions(-) create mode 100644 applications/leased_compiler_runner.py create mode 100644 engine/supply/compiler_runner.py create mode 100644 engine/supply/link_graph.py create mode 100644 migrations/versions/20260731_0045_revision_link_graph.py diff --git a/applications/file_scan.py b/applications/file_scan.py index 4884382a..dae698af 100644 --- a/applications/file_scan.py +++ b/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" @@ -154,16 +148,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, @@ -279,16 +271,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 @@ -505,13 +495,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 +514,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. diff --git a/applications/leased_compiler_runner.py b/applications/leased_compiler_runner.py new file mode 100644 index 00000000..40912eda --- /dev/null +++ b/applications/leased_compiler_runner.py @@ -0,0 +1,86 @@ +"""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, + "construct": failure.construct.value if failure.construct is not None else None, + "position": ( + { + "line": failure.position.line, + "column": failure.position.column, + "byteOffset": failure.position.byte_offset, + } + if failure.position is not None + else None + ), + } + + +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() diff --git a/engine/persistence/configuration.py b/engine/persistence/configuration.py index a427902d..8eaab6d9 100644 --- a/engine/persistence/configuration.py +++ b/engine/persistence/configuration.py @@ -21,6 +21,7 @@ DELIVERY_EVIDENCE_DEFINER_ROLE = "context_engine_delivery_evidence_definer" CITATION_DEFINER_ROLE = "context_engine_citation_definer" ACCESS_POLICY_DEFINER_ROLE = "context_engine_access_policy_definer" +GRAPH_DEFINER_ROLE = "context_engine_graph_definer" WORKER_LEASE_DEFINER_ROLE = "context_engine_worker_lease_definer" FILE_DISPATCH_DEFINER_ROLE = "context_engine_file_dispatch_definer" CONTEXT_RUN_READER_DEFINER_ROLE = "context_engine_context_run_reader_definer" diff --git a/engine/persistence/file_imports.py b/engine/persistence/file_imports.py index acba3349..18441f22 100644 --- a/engine/persistence/file_imports.py +++ b/engine/persistence/file_imports.py @@ -44,7 +44,9 @@ validate_embedding_batch, worker_lease_digest, ) +from engine.supply.compiler_runner import compile_in_leased_compiler_runner from engine.supply.jobs import _require_utc +from engine.supply.link_graph import extract_revision_links _CONCURRENT_PUBLICATION_WAIT_SECONDS = 5.0 _CONCURRENT_PUBLICATION_POLL_SECONDS = 0.01 @@ -296,7 +298,11 @@ def run(self, redemption: FileImportLeaseRedemption) -> PublishedFileImport: or sha256(source).hexdigest() != redeemed.expected_content_sha256 ): raise LookupError("accepted File observation changed") - outcome = compile_markdown(source, self._config) + outcome = ( + compile_in_leased_compiler_runner(source, self._config) + if self._config.version == "markdown-config-v3" + else compile_markdown(source, self._config) + ) except LookupError: with suppress(WorkNotAvailable): self._fail(redemption.token, claims) @@ -427,10 +433,21 @@ def _publish( raise _rejection(token) requested_revision_id = self._uuid_factory() resource_ref = _resource_ref(redeemed.source_ref, redeemed.path) - if document.provenance.is_structural_v2: + if document.provenance.is_structural_v2 or document.provenance.is_rich_v3: raw_compilation_document = json.loads( canonicalize_parsed_document(document).decode("utf-8") ) + if document.provenance.is_rich_v3: + raw_compilation_document["revisionLinks"] = [ + { + "kind": link.kind.value, + "targetPath": link.target_path, + } + for link in extract_revision_links( + document, + source_path=redeemed.path.value, + ) + ] compilation_document: str | None = json.dumps( raw_compilation_document, ensure_ascii=False, diff --git a/engine/persistence/membership_context.py b/engine/persistence/membership_context.py index c3da73a6..aee0ce8a 100644 --- a/engine/persistence/membership_context.py +++ b/engine/persistence/membership_context.py @@ -65,6 +65,7 @@ MaterializedFragmentProjection, MaterializedFragmentWindowItem, MaterializedFragmentWindowRead, + MaterializedOneHopCandidate, MaterializedProjectionKind, MaterializedProjectionSession, MaterializedPublicationTrace, @@ -1019,6 +1020,67 @@ def read_fragment_window( reauthorization_refs=reauthorization_refs, ) + def discover_one_hop( + self, + anchors: tuple[CandidateRef, ...], + limit: int, + ) -> tuple[MaterializedOneHopCandidate, ...]: + """Read outgoing and backlink locators from current Revision edges.""" + + if type(anchors) is not tuple or any( + type(anchor) is not CandidateRef for anchor in anchors + ): + raise TypeError("one-hop discovery requires exact anchors") + if type(limit) is not int or not 1 <= limit <= 64: + raise ValueError("one-hop discovery requires a bounded limit") + if not anchors: + return () + rows = self._connection.execute( + text( + """ + SELECT * + FROM public.context_runtime_resolve_one_hop_graph( + CAST(:organization_ids AS uuid[]), + CAST(:source_refs AS text[]), + CAST(:resource_refs AS text[]), + CAST(:revision_ids AS uuid[]), + CAST(:fragment_refs AS text[]), + :limit + ) + """ + ), + { + "organization_ids": [anchor.organization_id for anchor in anchors], + "source_refs": [anchor.source_ref for anchor in anchors], + "resource_refs": [anchor.resource_ref for anchor in anchors], + "revision_ids": [ + _canonical_candidate_revision(anchor.revision_ref) + for anchor in anchors + ], + "fragment_refs": [anchor.fragment_ref for anchor in anchors], + "limit": limit, + }, + ) + return tuple( + MaterializedOneHopCandidate( + anchor_ref=CandidateRef( + organization_id=row.anchor_organization_id, + source_ref=row.anchor_source_ref, + resource_ref=row.anchor_resource_ref, + revision_ref=str(row.anchor_revision_id), + fragment_ref=row.anchor_fragment_ref, + ), + candidate_ref=CandidateRef( + organization_id=row.organization_id, + source_ref=row.source_ref, + resource_ref=row.resource_ref, + revision_ref=str(row.revision_id), + fragment_ref=row.fragment_ref, + ), + ) + for row in rows + ) + def _observe_active_runtime_release( connection: Connection, diff --git a/engine/persistence/schema_security_manifest.yaml b/engine/persistence/schema_security_manifest.yaml index 46f84f1c..b6077310 100644 --- a/engine/persistence/schema_security_manifest.yaml +++ b/engine/persistence/schema_security_manifest.yaml @@ -1,5 +1,5 @@ { - "manifestVersion": "42.0.0", + "manifestVersion": "43.0.0", "controlOperations": [ { "name": "register_file_source", @@ -526,7 +526,8 @@ "name": "publish_file_import", "versionedDatabaseFunctions": { "markdown-config-v1": "context_worker_publish_file_import_v2", - "markdown-config-v2": "context_worker_publish_structural_file_import_v2" + "markdown-config-v2": "context_worker_publish_structural_file_import_v2", + "markdown-config-v3": "context_worker_prepare_file_publication" }, "role": "context_engine_worker", "definerRole": "context_engine_worker_lease_definer", @@ -553,6 +554,7 @@ "context_resource", "context_revision", "file_revision_snapshot", + "revision_link_edge", "context_fragment", "revision_publication_event", "exact_phrase_candidate", @@ -995,6 +997,14 @@ ], "using": "(organization_id = (NULLIF(current_setting('app.organization_id'::text, true), ''::text))::uuid)" }, + { + "name": "membership_graph_definer_select", + "command": "SELECT", + "roles": [ + "context_engine_graph_definer" + ], + "using": "(organization_id = (NULLIF(current_setting('app.organization_id'::text, true), ''::text))::uuid)" + }, { "name": "membership_migrator_administration", "command": "ALL", @@ -1054,6 +1064,9 @@ ], "context_engine_context_run_reader_definer": [ "SELECT" + ], + "context_engine_graph_definer": [ + "SELECT organization_id, user_id, membership_id, membership_version, status, valid_from, valid_until" ] }, "partitions": [], @@ -2148,6 +2161,14 @@ ], "withCheck": "(organization_id = (NULLIF(current_setting('app.organization_id'::text, true), ''::text))::uuid)" }, + { + "name": "context_resource_graph_definer_select", + "command": "SELECT", + "roles": [ + "context_engine_graph_definer" + ], + "using": "(organization_id = (NULLIF(current_setting('app.organization_id'::text, true), ''::text))::uuid)" + }, { "name": "context_resource_file_import_definer_select", "command": "SELECT", @@ -2228,6 +2249,9 @@ "context_engine_learning": [], "context_engine_release_operator": [ "EXECUTE context_release_observe_candidate_snapshot" + ], + "context_engine_graph_definer": [ + "SELECT organization_id, source_ref, resource_ref, active_revision_id, tombstoned" ] }, "partitions": [], @@ -6897,6 +6921,14 @@ ], "withCheck": "(organization_id = (NULLIF(current_setting('app.organization_id'::text, true), ''::text))::uuid)" }, + { + "name": "context_fragment_graph_definer_select", + "command": "SELECT", + "roles": [ + "context_engine_graph_definer" + ], + "using": "(organization_id = (NULLIF(current_setting('app.organization_id'::text, true), ''::text))::uuid)" + }, { "name": "context_fragment_file_noop_definer_select", "command": "SELECT", @@ -6943,6 +6975,9 @@ ], "context_engine_citation_definer": [ "SELECT" + ], + "context_engine_graph_definer": [ + "SELECT organization_id, resource_ref, revision_id, fragment_ref, ordinal" ] }, "partitions": [], @@ -8141,6 +8176,14 @@ "context_engine_file_dispatch_definer" ], "using": "true" + }, + { + "name": "file_acquisition_graph_definer_select", + "command": "SELECT", + "roles": [ + "context_engine_graph_definer" + ], + "using": "(organization_id = (NULLIF(current_setting('app.organization_id'::text, true), ''::text))::uuid)" } ] }, @@ -8176,6 +8219,9 @@ "context_engine_worker_lease_definer": [ "SELECT", "INSERT" + ], + "context_engine_graph_definer": [ + "SELECT organization_id, source_id, acquisition_id, relative_path" ] }, "partitions": [], @@ -9274,6 +9320,14 @@ "context_engine_worker_lease_definer" ], "using": "organization_id = NULLIF(current_setting('app.organization_id', true), '')::uuid" + }, + { + "name": "file_revision_snapshot_graph_definer_select", + "command": "SELECT", + "roles": [ + "context_engine_graph_definer" + ], + "using": "(organization_id = (NULLIF(current_setting('app.organization_id'::text, true), ''::text))::uuid)" } ] }, @@ -9302,6 +9356,9 @@ "context_engine_worker_lease_definer": [ "SELECT", "INSERT" + ], + "context_engine_graph_definer": [ + "SELECT organization_id, acquisition_id, resource_ref, revision_id" ] }, "partitions": [], @@ -9318,6 +9375,150 @@ "WORKER-LEASE-007" ] }, + { + "name": "revision_link_edge", + "classification": "tenant_owned", + "nonOwnerEvidence": { + "evidenceId": "PG-ONE-HOP-GRAPH-151", + "selector": { + "table": "revision_link_edge" + } + }, + "purpose": "Immutable content-free outgoing Revision link structure; carries no content, Article authorization, Fragment ACL, or delivery decision", + "organizationColumn": "organization_id", + "organizationInclusiveKeys": [ + { + "name": "pk_revision_link_edge", + "kind": "primary_key", + "columns": [ + "organization_id", + "source_resource_ref", + "source_revision_id", + "ordinal" + ] + }, + { + "name": "uq_revision_link_edge_target", + "kind": "unique", + "columns": [ + "organization_id", + "source_resource_ref", + "source_revision_id", + "target_path" + ] + } + ], + "foreignKeys": [ + { + "name": "fk_revision_link_edge_revision_same_organization", + "columns": [ + "organization_id", + "source_resource_ref", + "source_revision_id" + ], + "references": { + "table": "context_revision", + "columns": [ + "organization_id", + "resource_ref", + "revision_id" + ] + } + } + ], + "checkConstraints": [ + { + "name": "ck_revision_link_edge_ordinal", + "expression": "ordinal >= 0" + }, + { + "name": "ck_revision_link_edge_kind", + "expression": "link_kind IN ('wikilink', 'embed', 'markdown_link')" + }, + { + "name": "ck_revision_link_edge_target_path", + "expression": "target_path is one bounded canonical Markdown path with no empty, dot, dot-dot, or backslash component" + } + ], + "rowLevelSecurity": { + "enabled": true, + "forced": true, + "policies": [ + { + "name": "revision_link_edge_graph_select", + "command": "SELECT", + "roles": [ + "context_engine_graph_definer" + ], + "using": "(organization_id = (NULLIF(current_setting('app.organization_id'::text, true), ''::text))::uuid)" + }, + { + "name": "revision_link_edge_migrator_administration", + "command": "ALL", + "roles": [ + "context_engine_migrator" + ], + "using": "true", + "withCheck": "true" + }, + { + "name": "revision_link_edge_worker_insert", + "command": "INSERT", + "roles": [ + "context_engine_worker_lease_definer" + ], + "withCheck": "(organization_id = (NULLIF(current_setting('app.organization_id'::text, true), ''::text))::uuid)" + } + ] + }, + "immutableRows": { + "trigger": "revision_link_edge_immutable", + "function": "context_content_reject_mutation", + "events": [ + "UPDATE", + "DELETE" + ], + "sqlstate": "55000" + }, + "functionOnlyMutation": { + "databaseFunction": "context_worker_prepare_file_publication", + "definerRole": "context_engine_worker_lease_definer", + "directTableMutationAllowed": false + }, + "permittedOperations": { + "context_engine_runtime": [ + "EXECUTE context_runtime_resolve_one_hop_graph" + ], + "context_engine_worker": [ + "EXECUTE context_worker_prepare_file_publication" + ], + "context_engine_worker_lease_definer": [ + "INSERT" + ], + "context_engine_graph_definer": [ + "SELECT" + ] + }, + "retention": { + "sourceContent": "none", + "authorizationDecision": "none", + "oldRevisionBackfill": "none" + }, + "partitions": [], + "securityInvariantIds": [ + "TENANT-OWNERSHIP-001", + "TENANT-FK-002", + "RLS-FAIL-CLOSED-003", + "INDEX-NOT-AUTHORITY-005" + ], + "negativeTestIds": [ + "DB-001", + "DB-004", + "DB-008", + "PG-ONE-HOP-GRAPH-151", + "SDK-ONE-HOP-GRAPH-151" + ] + }, { "name": "revision_publication_event", "classification": "tenant_owned", diff --git a/engine/runtime/authorized_ranking.py b/engine/runtime/authorized_ranking.py index f3e99c17..a4c1c194 100644 --- a/engine/runtime/authorized_ranking.py +++ b/engine/runtime/authorized_ranking.py @@ -2,6 +2,7 @@ from __future__ import annotations +import re from collections import defaultdict from dataclasses import dataclass, field from math import isfinite @@ -20,7 +21,9 @@ "NEUTRAL_FUSED_RANK", "UNIFORM_RANKER_WEIGHT", "AuthorizedRerankItem", + "GRAPH_RANKER_REF", "join_authorized_ranking", + "rank_authorized_one_hop", "select_authorized_ranking", ] @@ -74,7 +77,56 @@ def __post_init__(self) -> None: UNIFORM_RANKER_WEIGHTS: Final = RankerWeights() -HYBRID_RANKER_WEIGHTS: Final = RankerWeights({"fts": 1.0, "vector": 1.0}) +GRAPH_RANKER_REF: Final = "graph" +HYBRID_RANKER_WEIGHTS: Final = RankerWeights( + {"fts": 1.0, "vector": 1.0, GRAPH_RANKER_REF: 1.0} +) +_RELEVANCE_TOKEN = re.compile(r"[\w]+", re.UNICODE) + + +def rank_authorized_one_hop( + query_text: str, + projections: tuple[AuthorizedProjection, ...], +) -> tuple[CandidateRankEvidence, ...]: + """Rank only already-authorized graph projections by lexical relevance.""" + + if type(query_text) is not str or not query_text or query_text.isspace(): + raise ValueError("authorized graph ranking requires a nonblank query") + if type(projections) is not tuple or any( + type(projection) is not AuthorizedProjection for projection in projections + ): + raise TypeError("authorized graph ranking requires exact projections") + query_tokens = frozenset(_RELEVANCE_TOKEN.findall(query_text.casefold())) + scored: list[tuple[AuthorizedProjection, float]] = [] + for projection in projections: + _require_active_authorized_projection(projection) + body_tokens = frozenset( + _RELEVANCE_TOKEN.findall(projection.projected_body.casefold()) + ) + overlap = len(query_tokens & body_tokens) + if overlap: + scored.append((projection, overlap / len(query_tokens))) + ordered = sorted( + scored, + key=lambda item: ( + -item[1], + _candidate_sort_key(item[0].candidate_ref), + ), + ) + return tuple( + CandidateRankEvidence( + candidate_ref=projection.candidate_ref, + per_ranker=( + RankerEvidence( + ranker_ref=GRAPH_RANKER_REF, + position=position, + score=score, + ), + ), + fused_rank=position, + ) + for position, (projection, score) in enumerate(ordered, start=1) + ) @dataclass(frozen=True, slots=True, init=False) diff --git a/engine/runtime/construction.py b/engine/runtime/construction.py index b6de7911..d3537b81 100644 --- a/engine/runtime/construction.py +++ b/engine/runtime/construction.py @@ -19,6 +19,7 @@ UNIFORM_RANKER_WEIGHTS, RankerWeights, join_authorized_ranking, + rank_authorized_one_hop, select_authorized_ranking, ) from engine.runtime.budget import PackageBudget, effective_package_budget @@ -121,9 +122,11 @@ from engine.runtime.materialized import ( CandidateDiscoverySession, MaterializedFragmentLocator, + MaterializedOneHopCandidate, MaterializedProjectionSession, _close_candidate_discovery_session, _construct_candidate_discovery_session, + _discover_materialized_one_hop, _locate_materialized_fragment, _project_materialized_fragment, _read_materialized_fragment_window, @@ -231,6 +234,10 @@ class AuthorizationDecision: provenance_receipt: DecisionProvenanceReceipt projections: tuple[AuthorizedProjection, ...] _projection_scope: _AuthorizationKernelScope | None = field(repr=False) + expanded_candidate_refs: frozenset[CandidateRef] = field( + default_factory=frozenset, + repr=False, + ) _effective_budget_limits: tuple[int, int, int, int] = field( init=False, repr=False, @@ -662,6 +669,18 @@ def _decision_integrity_snapshot(decision: AuthorizationDecision) -> tuple[objec policy.effective_scope.digest, tuple(getattr(provenance, item.name) for item in fields(provenance)), projections, + tuple( + sorted( + ( + candidate.organization_id.bytes, + candidate.source_ref, + candidate.resource_ref, + candidate.revision_ref, + candidate.fragment_ref, + ) + for candidate in decision.expanded_candidate_refs + ) + ), ) @@ -776,9 +795,18 @@ def select_for_delivery( _decision_integrity_material(decision), ): raise ValueError("authorization decision integrity validation failed") + ranked_refs = frozenset( + evidence.candidate_ref for evidence in rank_evidence + ) + eligible_projections = tuple( + projection + for projection in decision.projections + if projection.candidate_ref not in decision.expanded_candidate_refs + or projection.candidate_ref in ranked_refs + ) selected = select_authorized_ranking( join_authorized_ranking( - decision.projections, + eligible_projections, rank_evidence, ranker_weights=( ranker_weights.values if ranker_weights is not None else None @@ -974,6 +1002,95 @@ def authorize_acquire( ) return decision + def authorize_one_hop( + self, + invocation: AuthenticatedInvocation, + preparation: PreparedAcquireAuthorization, + decision: AuthorizationDecision, + candidates: tuple[MaterializedOneHopCandidate, ...], + *, + projection_session: MaterializedProjectionSession, + ) -> AuthorizationDecision: + """Admit inherited and cross-Article graph candidates through the Kernel.""" + + if type(preparation) is not PreparedAcquireAuthorization: + raise TypeError("Kernel graph expansion requires prepared authorization") + if type(decision) is not AuthorizationDecision: + raise TypeError("Kernel graph expansion requires AuthorizationDecision") + if type(candidates) is not tuple or any( + type(candidate) is not MaterializedOneHopCandidate + for candidate in candidates + ): + raise TypeError("Kernel graph expansion requires exact graph candidates") + anchors = { + projection.candidate_ref: projection for projection in decision.projections + } + inherited: list[AuthorizedProjection] = [] + reauthorization_refs: list[CandidateRef] = [] + for candidate in candidates: + anchor = anchors.get(candidate.anchor_ref) + if anchor is None: + raise ValueError("graph candidate is not rooted in this decision") + candidate_ref = candidate.candidate_ref + anchor_ref = candidate.anchor_ref + same_article = ( + candidate_ref.organization_id == anchor_ref.organization_id + and candidate_ref.source_ref == anchor_ref.source_ref + and candidate_ref.resource_ref == anchor_ref.resource_ref + ) + if not same_article: + reauthorization_refs.append(candidate_ref) + continue + if candidate_ref.revision_ref != anchor_ref.revision_ref: + continue + locator = _locate_materialized_fragment( + projection_session, + candidate_ref, + ) + if locator is None or not _locator_matches_candidate( + locator, + candidate_ref, + ): + continue + projection = _project_materialized_fragment( + projection_session, + locator, + ) + if projection is None: + continue + inherited.append( + _construct_inherited_authorized_projection( + anchor=anchor, + candidate_ref=candidate_ref, + body=projection.rendered_body, + projected_field_refs=projection.projected_field_refs, + ) + ) + reauthorized, _scope = self._authorize_and_project( + invocation, + preparation.policy_receipt, + preparation.provenance_receipt, + tuple(reauthorization_refs), + projection_session, + kernel_scope=decision._projection_scope, + ) + expanded = replace( + decision, + projections=decision.projections + tuple(inherited) + reauthorized, + expanded_candidate_refs=frozenset( + projection.candidate_ref for projection in (*inherited, *reauthorized) + ), + ) + object.__setattr__( + expanded, + "_integrity_seal", + _issue_selection_authority_seal( + _kernel_selection_authority(self), + _decision_integrity_material(expanded), + ), + ) + return expanded + def authorize_open_citation( self, invocation: AuthenticatedInvocation, @@ -1282,6 +1399,8 @@ def _authorize_and_project( provenance_receipt: DecisionProvenanceReceipt, candidates: tuple[CandidateRef, ...], projection_session: MaterializedProjectionSession | None, + *, + kernel_scope: _AuthorizationKernelScope | None = None, ) -> tuple[tuple[AuthorizedProjection, ...], _AuthorizationKernelScope | None]: if not candidates or not policy_receipt.effective_scope.targets: return (), None @@ -1290,7 +1409,8 @@ def _authorize_and_project( "candidate discovery requires same-transaction projection session" ) - kernel_scope = _open_authorization_kernel_scope() + selected_kernel_scope = kernel_scope or _open_authorization_kernel_scope() + owns_scope = kernel_scope is None try: projections = [] ordered_candidates = sorted( @@ -1321,7 +1441,7 @@ def _authorize_and_project( if field_projection is None: continue projection = _construct_authorized_projection( - kernel_scope=kernel_scope, + kernel_scope=selected_kernel_scope, candidate_ref=candidate, body=field_projection.rendered_body, projected_field_refs=(field_projection.projected_field_refs), @@ -1344,9 +1464,10 @@ def _authorize_and_project( ), ) projections.append(projection) - return tuple(projections), kernel_scope + return tuple(projections), selected_kernel_scope except BaseException: - _close_authorization_kernel_scope(kernel_scope) + if owns_scope: + _close_authorization_kernel_scope(selected_kernel_scope) raise @@ -1730,6 +1851,40 @@ def resolve( invocation.user_actor.materialized_projection_session ), ) + projection_session = invocation.user_actor.materialized_projection_session + if projection_session is not None and decision.projections: + one_hop = _discover_materialized_one_hop( + projection_session, + decision.projections, + min(64, self._candidate_submission_limit), + ) + main_candidate_refs = set(candidate_refs) + one_hop = tuple( + item + for item in one_hop + if item.candidate_ref not in main_candidate_refs + ) + if one_hop: + main_projection_count = len(decision.projections) + decision = self._kernel.authorize_one_hop( + invocation, + preparation, + decision, + one_hop, + projection_session=projection_session, + ) + graph_evidence = rank_authorized_one_hop( + request.need.query, + tuple( + projection + for projection in decision.projections[ + main_projection_count: + ] + if projection.candidate_ref + in decision.expanded_candidate_refs + ), + ) + rank_evidence = rank_evidence + graph_evidence else: assert isinstance(request, OpenCitation) citation_session = invocation.user_actor.citation_open_session diff --git a/engine/runtime/materialized.py b/engine/runtime/materialized.py index 7b08c523..a03aaceb 100644 --- a/engine/runtime/materialized.py +++ b/engine/runtime/materialized.py @@ -13,6 +13,7 @@ from engine.runtime.evidence import ( MAX_PROJECTED_FIELD_REF_LENGTH, MAX_PROJECTED_FIELD_REFS, + AuthorizedProjection, CandidateRef, validate_projected_field_refs, ) @@ -307,6 +308,21 @@ def __post_init__(self) -> None: ) +@dataclass(frozen=True, slots=True) +class MaterializedOneHopCandidate: + """One content-free graph candidate retaining its authorized root.""" + + anchor_ref: CandidateRef = field(repr=False) + candidate_ref: CandidateRef = field(repr=False) + + def __post_init__(self) -> None: + if ( + type(self.anchor_ref) is not CandidateRef + or type(self.candidate_ref) is not CandidateRef + ): + raise TypeError("one-hop graph lineage requires exact CandidateRef values") + + class MaterializedProjectionPort(Protocol): """Narrow operations executed by the owning current database transaction.""" @@ -349,6 +365,15 @@ def read_fragment_window( ) -> MaterializedFragmentWindowRead: ... +@runtime_checkable +class _MaterializedOneHopPort(Protocol): + def discover_one_hop( + self, + anchors: tuple[CandidateRef, ...], + limit: int, + ) -> tuple[MaterializedOneHopCandidate, ...]: ... + + class _MaterializedProjectionScope: """Private lifetime token owned by one current UserActor transaction.""" @@ -880,6 +905,48 @@ def _project_materialized_fragment( return projection +def _discover_materialized_one_hop( + session: MaterializedProjectionSession, + anchors: tuple[AuthorizedProjection, ...], + limit: int, +) -> tuple[MaterializedOneHopCandidate, ...]: + """Generate one content-free graph hop from exact authorized anchors only.""" + + from engine.runtime.evidence import _require_active_authorized_projection + + _require_active_materialized_projection_session(session) + if type(anchors) is not tuple or any( + type(anchor) is not AuthorizedProjection for anchor in anchors + ): + raise TypeError("one-hop expansion requires authorized projections") + if type(limit) is not int or not 1 <= limit <= 64: + raise ValueError("one-hop expansion requires a bounded limit") + for anchor in anchors: + _require_active_authorized_projection(anchor) + if not anchors: + return () + if not isinstance(session._port, _MaterializedOneHopPort): + return () + discovered = session._port.discover_one_hop( + tuple(anchor.candidate_ref for anchor in anchors), + limit, + ) + if ( + type(discovered) is not tuple + or len(discovered) > limit + or any(type(item) is not MaterializedOneHopCandidate for item in discovered) + ): + raise TypeError("one-hop expansion must return bounded ranked refs") + anchor_refs = {anchor.candidate_ref for anchor in anchors} + if any(item.anchor_ref not in anchor_refs for item in discovered): + raise ValueError("one-hop expansion roots must be authorized anchors") + if any(item.candidate_ref in anchor_refs for item in discovered): + raise ValueError("one-hop expansion cannot return an anchor") + if len({item.candidate_ref for item in discovered}) != len(discovered): + raise ValueError("one-hop expansion candidates must be unique") + return discovered + + def _read_materialized_fragment_window( session: MaterializedProjectionSession, anchor: MaterializedFragmentLocator, diff --git a/engine/supply/compiler_runner.py b/engine/supply/compiler_runner.py new file mode 100644 index 00000000..52ecbe7c --- /dev/null +++ b/engine/supply/compiler_runner.py @@ -0,0 +1,112 @@ +"""Leased parent boundary for the pure rich-Markdown compiler subprocess.""" + +from __future__ import annotations + +import base64 +import json +import subprocess +import sys +from typing import Final, cast + +from engine.supply.markdown import ( + CompilationFailure, + CompilationFailureCode, + CompilationOutcome, + MarkdownCompilerConfig, + SourcePoint, + UnsupportedConstruct, + deserialize_parsed_document, +) + +_RUNNER_MODULE: Final = "applications.leased_compiler_runner" +COMPILER_RUNNER_TIMEOUT_SECONDS: Final = 30.0 + + +def _boundary_failure() -> CompilationFailure: + return CompilationFailure( + code=CompilationFailureCode.UNSUPPORTED_DOCUMENT_SHAPE, + position=None, + ) + + +def _failure_from_document(value: object) -> CompilationFailure: + if type(value) is not dict: + raise ValueError("runner failure must be an object") + document = cast(dict[str, object], value) + position_value = document["position"] + position = None + if type(position_value) is dict: + point = cast(dict[str, object], position_value) + position = SourcePoint( + line=cast(int, point["line"]), + column=cast(int, point["column"]), + byte_offset=cast(int, point["byteOffset"]), + ) + construct_value = document["construct"] + return CompilationFailure( + code=CompilationFailureCode(cast(str, document["code"])), + position=position, + construct=( + UnsupportedConstruct(cast(str, construct_value)) + if construct_value is not None + else None + ), + ) + + +def compile_in_leased_compiler_runner( + source: bytes, + config: MarkdownCompilerConfig, +) -> CompilationOutcome: + """Run the pure child selected by an already-verified leased worker.""" + + if type(source) is not bytes: + raise TypeError("compiler-runner source must be exact bytes") + if type(config) is not MarkdownCompilerConfig or config.token_ceiling is None: + raise TypeError("compiler-runner requires rich Markdown config") + try: + completed = subprocess.run( + [ + sys.executable, + "-m", + _RUNNER_MODULE, + "--compile-leased", + "--config", + config.version, + "--token-ceiling", + str(config.token_ceiling), + ], + input=source, + capture_output=True, + check=False, + env={}, + timeout=COMPILER_RUNNER_TIMEOUT_SECONDS, + ) + except Exception: + return _boundary_failure() + if completed.returncode != 0: + return _boundary_failure() + try: + envelope = json.loads(completed.stdout) + except (UnicodeDecodeError, json.JSONDecodeError): + return _boundary_failure() + if type(envelope) is not dict: + return _boundary_failure() + document = cast(dict[str, object], envelope) + if document.get("outcome") == "parsed": + encoded = document.get("document") + if type(encoded) is not str: + return _boundary_failure() + try: + return deserialize_parsed_document(base64.b64decode(encoded, validate=True)) + except Exception: + return _boundary_failure() + if document.get("outcome") == "failure": + try: + return _failure_from_document(document.get("failure")) + except Exception: + return _boundary_failure() + return _boundary_failure() + + +__all__ = ["compile_in_leased_compiler_runner"] diff --git a/engine/supply/link_graph.py b/engine/supply/link_graph.py new file mode 100644 index 00000000..25e5da54 --- /dev/null +++ b/engine/supply/link_graph.py @@ -0,0 +1,162 @@ +"""Deterministic content-free link structure derived from one rich Revision.""" + +from __future__ import annotations + +import posixpath +import re +from dataclasses import dataclass, field +from enum import StrEnum +from pathlib import PurePosixPath +from urllib.parse import unquote, urlsplit + +from engine.supply.markdown import ParsedDocument, SectionKind + +_INLINE_CODE = re.compile(r"`+[^`\r\n]*`+") +_WIKILINK = re.compile(r"(?P!)?\[\[(?P[^]\r\n]+)]]") +_MARKDOWN_LINK = re.compile(r"(?[^()\r\n]+)\)") +_REFERENCE_DEFINITION = re.compile( + r'''^ {0,3}\[(?P