From 0a883c7f564f3f9aa7405a027d477306a537e457 Mon Sep 17 00:00:00 2001 From: Ning Zhou Date: Mon, 17 Aug 2026 18:56:06 +0400 Subject: [PATCH] feat: add public source immutable landing slice --- .github/workflows/cd-staging.yml | 2 + .github/workflows/ci.yml | 2 + data_agent/api/platform_gateway_routes.py | 41 +- data_agent/platform_gateway.py | 127 +++-- data_agent/public_source_landing.py | 451 ++++++++++++++++++ data_agent/test_platform_gateway.py | 12 +- data_agent/test_public_source_landing.py | 242 ++++++++++ .../test_public_source_landing_postgres.py | 144 ++++++ ...adr-081-public-source-immutable-landing.md | 73 +++ .../public-source-landing-2026-08-17.json | 21 + docs/roadmap.md | 5 +- docs/system-of-record-matrix-2026-07-24.md | 8 +- 12 files changed, 1086 insertions(+), 42 deletions(-) create mode 100644 data_agent/public_source_landing.py create mode 100644 data_agent/test_public_source_landing.py create mode 100644 data_agent/test_public_source_landing_postgres.py create mode 100644 docs/architecture-decisions/adr-081-public-source-immutable-landing.md create mode 100644 docs/evidence/public-source-landing-2026-08-17.json diff --git a/.github/workflows/cd-staging.yml b/.github/workflows/cd-staging.yml index c5bd4d4d..824bb200 100644 --- a/.github/workflows/cd-staging.yml +++ b/.github/workflows/cd-staging.yml @@ -171,6 +171,8 @@ jobs: data_agent/test_chongqing_admission_readiness.py \ data_agent/test_chongqing_protected_admission.py \ data_agent/test_chongqing_protected_admission_workflow.py \ + data_agent/test_public_source_landing.py \ + data_agent/test_public_source_landing_postgres.py \ data_agent/test_dolphinscheduler_adapter.py \ data_agent/test_dolphinscheduler_command_consumer.py \ data_agent/test_dolphinscheduler_command_worker.py \ diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index b07e8a3b..003d9ce4 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -180,6 +180,8 @@ jobs: data_agent/test_chongqing_admission_readiness.py \ data_agent/test_chongqing_protected_admission.py \ data_agent/test_chongqing_protected_admission_workflow.py \ + data_agent/test_public_source_landing.py \ + data_agent/test_public_source_landing_postgres.py \ data_agent/test_dolphinscheduler_adapter.py \ data_agent/test_dolphinscheduler_command_consumer.py \ data_agent/test_dolphinscheduler_command_worker.py \ diff --git a/data_agent/api/platform_gateway_routes.py b/data_agent/api/platform_gateway_routes.py index 43b593e6..d09be0f8 100644 --- a/data_agent/api/platform_gateway_routes.py +++ b/data_agent/api/platform_gateway_routes.py @@ -13,7 +13,6 @@ from starlette.responses import JSONResponse from starlette.routing import Route -from .helpers import _get_user_from_request from ..platform_contracts import ( Artifact, FrameworkAttemptObservation, @@ -44,10 +43,11 @@ GatewayNotFoundError, GatewayUnavailableError, GatewayValidationError, + LandingRegistration, PlatformGateway, PlatformGatewayError, ) - +from .helpers import _get_user_from_request _TENANT_ADAPTER = TypeAdapter(TenantId) _PLATFORM_ROLES = frozenset({"admin", "platform_operator"}) @@ -544,6 +544,42 @@ async def create_artifact(request: Request) -> JSONResponse: return _gateway_error(request, exc) +async def create_landing(request: Request) -> JSONResponse: + principal = _principal(request) + if isinstance(principal, JSONResponse): + return principal + registration = await _parse(request, LandingRegistration) + if isinstance(registration, JSONResponse): + return registration + if mismatch := _tenant_matches( + request, principal, registration.resource.tenant_id + ): + return mismatch + actors = { + registration.resource_version.created_by, + registration.artifact.created_by, + } + if actors != {principal.actor_ref}: + return _error( + request, + 403, + "actor_mismatch", + "Landing version and artifact actors must match authenticated actor", + ) + try: + result = await asyncio.to_thread( + _gateway().register_landing, registration + ) + return _success( + request, + result.value, + status_code=201 if result.created else 200, + created=result.created, + ) + except PlatformGatewayError as exc: + return _gateway_error(request, exc) + + async def create_quality_result(request: Request) -> JSONResponse: principal = _principal(request) if isinstance(principal, JSONResponse): @@ -681,6 +717,7 @@ def get_platform_gateway_routes() -> list[Route]: create_dolphinscheduler_callback, methods=["POST"], ), + Route(f"{base}/landings", create_landing, methods=["POST"]), Route(f"{base}/artifacts", create_artifact, methods=["POST"]), Route(f"{base}/quality-results", create_quality_result, methods=["POST"]), Route( diff --git a/data_agent/platform_gateway.py b/data_agent/platform_gateway.py index 8b44767c..1a33212a 100644 --- a/data_agent/platform_gateway.py +++ b/data_agent/platform_gateway.py @@ -170,6 +170,47 @@ def _consistent_definition_identity(self) -> DefinitionRegistration: return self +class LandingRegistration(BaseModel): + """Atomic immutable Landing Resource + Version + input Artifact registration.""" + + model_config = ConfigDict(extra="forbid", frozen=True) + + resource: Resource + resource_version: ResourceVersion + artifact: Artifact + + @model_validator(mode="after") + def _consistent_landing_identity(self) -> LandingRegistration: + resource = self.resource + version = self.resource_version + artifact = self.artifact + if resource.resource_kind != "dataset": + raise ValueError("landing resource must use kind 'dataset'") + if resource.authority_system != "gda_landing": + raise ValueError("landing resource authority must be 'gda_landing'") + if len({resource.tenant_id, version.tenant_id, artifact.tenant_id}) != 1: + raise ValueError("landing registration tenants must match") + if resource.resource_urn != version.resource_urn: + raise ValueError("landing ResourceVersion must bind the Resource") + if artifact.resource_version_id != version.resource_version_id: + raise ValueError("landing Artifact must bind the ResourceVersion") + if artifact.content_sha256 != version.content_sha256: + raise ValueError("landing Artifact and ResourceVersion hashes must match") + if artifact.artifact_role.value != "input": + raise ValueError("landing Artifact must use the input role") + if artifact.run_id is not None: + raise ValueError("landing Artifact cannot bind a PlatformRun") + if artifact.manifest.get("schema") != "gda.public_source_landing.v1": + raise ValueError("landing Artifact must contain the public-source manifest") + if artifact.manifest.get("admission_class") != "public_open": + raise ValueError("landing Artifact must use public_open admission") + if artifact.manifest.get("resource_urn") != resource.resource_urn: + raise ValueError("landing manifest must bind the Resource URN") + if artifact.manifest.get("authority_locator") != resource.authority_locator: + raise ValueError("landing manifest must bind the authority locator") + return self + + def _json(value: Any) -> str: return json.dumps(value, ensure_ascii=True, sort_keys=True, separators=(",", ":")) @@ -1234,39 +1275,64 @@ def get_artifact(self, tenant_id: str, artifact_id: UUID) -> Artifact: raise GatewayNotFoundError("Artifact was not found") return artifact + def _put_artifact(self, connection, artifact: Artifact) -> GatewayWriteResult: + inserted = connection.execute( + text( + """ + INSERT INTO gda_control.artifact ( + tenant_id, artifact_id, artifact_key, artifact_role, + storage_uri, media_type, content_sha256, size_bytes, + run_id, resource_version_id, manifest, created_by, created_at + ) VALUES ( + :tenant_id, :artifact_id, :artifact_key, :artifact_role, + :storage_uri, :media_type, :content_sha256, :size_bytes, + :run_id, :resource_version_id, + CAST(:manifest AS jsonb), :created_by, :created_at + ) + ON CONFLICT DO NOTHING + RETURNING artifact_id + """ + ), + { + **artifact.model_dump(mode="python", exclude={"manifest"}), + "artifact_role": artifact.artifact_role.value, + "manifest": _json(artifact.manifest), + }, + ).first() + stored = self._load_artifact( + connection, artifact.tenant_id, artifact.artifact_id + ) + if stored is None or stored != artifact: + raise GatewayConflictError( + "Artifact identity already has a different payload" + ) + return GatewayWriteResult(stored, inserted is not None) + def record_artifact(self, artifact: Artifact) -> GatewayWriteResult: with self._transaction(artifact.tenant_id) as connection: - inserted = connection.execute( - text( - """ - INSERT INTO gda_control.artifact ( - tenant_id, artifact_id, artifact_key, artifact_role, - storage_uri, media_type, content_sha256, size_bytes, - run_id, resource_version_id, manifest, created_by, created_at - ) VALUES ( - :tenant_id, :artifact_id, :artifact_key, :artifact_role, - :storage_uri, :media_type, :content_sha256, :size_bytes, - :run_id, :resource_version_id, - CAST(:manifest AS jsonb), :created_by, :created_at + return self._put_artifact(connection, artifact) + + def register_landing( + self, registration: LandingRegistration + ) -> GatewayWriteResult: + """Register a staged Landing object and its ledger identity atomically.""" + with self._transaction(registration.resource.tenant_id) as connection: + resource_result = self._put_resource(connection, registration.resource) + version_result = self._put_resource_version( + connection, registration.resource_version + ) + artifact_result = self._put_artifact(connection, registration.artifact) + return GatewayWriteResult( + registration, + any( + result.created + for result in ( + resource_result, + version_result, + artifact_result, ) - ON CONFLICT DO NOTHING - RETURNING artifact_id - """ ), - { - **artifact.model_dump(mode="python", exclude={"manifest"}), - "artifact_role": artifact.artifact_role.value, - "manifest": _json(artifact.manifest), - }, - ).first() - stored = self._load_artifact( - connection, artifact.tenant_id, artifact.artifact_id ) - if stored is None or stored != artifact: - raise GatewayConflictError( - "Artifact identity already has a different payload" - ) - return GatewayWriteResult(stored, inserted is not None) @staticmethod def _metadata_fabric_binding_from_row(row) -> MetadataFabricBindingRecord: @@ -1993,6 +2059,8 @@ def build_gateway_report( 'SET LOCAL ROLE "{GATEWAY_DATABASE_ROLE}"', "SELECT set_config('app.current_tenant', :tenant, true)", "ON CONFLICT DO NOTHING", + "class LandingRegistration", + "def register_landing(", "def get_artifact(", "def _validate_run_policy_references(", "def record_attempt_and_enqueue_reconcile(", @@ -2011,6 +2079,7 @@ def build_gateway_report( 'frozenset({"admin", "platform_operator"})', '"tenant_context_required"', '"actor_mismatch"', + "create_landing", "create_dolphinscheduler_callback", "create_quality_result", "finalize_run_success", @@ -2080,7 +2149,7 @@ def build_gateway_report( "schema": GATEWAY_SCHEMA_VERSION, "status": "valid" if not errors else "invalid", "database_role": GATEWAY_DATABASE_ROLE, - "route_count": 12, + "route_count": 13, "files": files, "missing_markers": missing_markers, "errors": errors, diff --git a/data_agent/public_source_landing.py b/data_agent/public_source_landing.py new file mode 100644 index 00000000..03c438be --- /dev/null +++ b/data_agent/public_source_landing.py @@ -0,0 +1,451 @@ +"""Immutable Landing execution for explicitly public/open source bytes.""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import os +import re +import stat +from datetime import UTC, datetime +from pathlib import Path +from typing import Annotated +from urllib.parse import urlsplit +from uuid import NAMESPACE_URL, uuid4, uuid5 + +from pydantic import Field, field_validator, model_validator +from sqlalchemy import create_engine + +from .platform_contracts import ( + Artifact, + FrozenContract, + NonEmptyText, + Resource, + ResourceVersion, + Sha256, + ShortName, + TenantId, + canonical_json_bytes, + canonical_json_fingerprint, +) +from .platform_gateway import LandingRegistration, PlatformGateway + +LANDING_SCHEMA = "gda.public_source_landing.v1" +_DATASET_ID_RE = re.compile(r"^[a-z0-9][a-z0-9._-]{0,79}$") +_SAFE_SUFFIX_RE = re.compile(r"^\.[a-z0-9]{1,12}$") + +DatasetId = Annotated[ + str, + Field(min_length=1, max_length=80, pattern=r"^[a-z0-9][a-z0-9._-]{0,79}$"), +] + + +class PublicSourceLandingError(RuntimeError): + """Public source bytes cannot be staged or verified safely.""" + + +class PublicSourceLandingRequest(FrozenContract): + schema_id = "public_source_landing_request" + + tenant_id: TenantId + dataset_id: DatasetId + source_uri: NonEmptyText + license_id: ShortName + owner_ref: NonEmptyText + expected_sha256: Sha256 + media_type: NonEmptyText + created_by: NonEmptyText + created_at: datetime + + @field_validator("source_uri") + @classmethod + def _stable_public_source_uri(cls, value: str) -> str: + parts = urlsplit(value) + if parts.scheme != "https" or not parts.netloc: + raise ValueError("public source URI must use HTTPS") + if parts.username or parts.password or parts.query or parts.fragment: + raise ValueError("public source URI must be stable and credential-free") + return value + + @field_validator("created_by") + @classmethod + def _controlled_actor(cls, value: str) -> str: + if not value.startswith(("human:", "workload:")): + raise ValueError("created_by must use a human or workload identity") + return value + + @field_validator("created_at") + @classmethod + def _utc_created_at(cls, value: datetime) -> datetime: + if value.tzinfo is None or value.utcoffset() is None: + raise ValueError("created_at must include a timezone") + return value.astimezone(UTC) + + +class PublicSourceLandingResult(FrozenContract): + schema_id = "public_source_landing_result" + + registration: LandingRegistration + payload_path: str + manifest_path: str + payload_created: bool + manifest_created: bool + manifest_sha256: Sha256 + ledger_created: bool | None = None + + @model_validator(mode="after") + def _consistent_paths(self) -> PublicSourceLandingResult: + payload = Path(self.payload_path) + manifest = Path(self.manifest_path) + if not payload.is_absolute() or not manifest.is_absolute(): + raise ValueError("landing result paths must be absolute") + if payload.parent != manifest.parent: + raise ValueError("landing payload and manifest must share one version root") + return self + + +def _safe_suffix(source_path: Path) -> str: + suffix = source_path.suffix.lower() + return suffix if _SAFE_SUFFIX_RE.fullmatch(suffix) else ".bin" + + +def _sha256_file(path: Path) -> tuple[str, int]: + digest = hashlib.sha256() + size = 0 + with path.open("rb") as handle: + for chunk in iter(lambda: handle.read(1024 * 1024), b""): + digest.update(chunk) + size += len(chunk) + return digest.hexdigest(), size + + +def _copy_to_staging(source_path: Path, staging_path: Path) -> tuple[str, int]: + if source_path.is_symlink(): + raise PublicSourceLandingError("landing source cannot be a symbolic link") + flags = os.O_RDONLY + if hasattr(os, "O_NOFOLLOW"): + flags |= os.O_NOFOLLOW + try: + source_fd = os.open(source_path, flags) + except OSError as exc: + raise PublicSourceLandingError("landing source is not readable") from exc + try: + source_stat = os.fstat(source_fd) + if not stat.S_ISREG(source_stat.st_mode): + raise PublicSourceLandingError("landing source must be a regular file") + digest = hashlib.sha256() + size = 0 + with os.fdopen(source_fd, "rb", closefd=False) as source, staging_path.open( + "xb" + ) as target: + os.chmod(staging_path, 0o600) + for chunk in iter(lambda: source.read(1024 * 1024), b""): + digest.update(chunk) + size += len(chunk) + target.write(chunk) + target.flush() + os.fsync(target.fileno()) + return digest.hexdigest(), size + finally: + os.close(source_fd) + + +def _verify_payload(path: Path, expected_sha256: str, expected_size: int) -> None: + if path.is_symlink() or not path.is_file(): + raise PublicSourceLandingError("landing payload is not an immutable regular file") + actual_sha256, actual_size = _sha256_file(path) + if actual_sha256 != expected_sha256 or actual_size != expected_size: + raise PublicSourceLandingError("existing landing payload does not match its key") + + +def _install_staged_payload( + staging_path: Path, + destination: Path, + *, + expected_sha256: str, + expected_size: int, +) -> bool: + destination.parent.mkdir(parents=True, exist_ok=True, mode=0o750) + try: + os.link(staging_path, destination) + os.chmod(destination, 0o440) + return True + except FileExistsError: + _verify_payload(destination, expected_sha256, expected_size) + return False + + +def _install_immutable_bytes(destination: Path, payload: bytes) -> bool: + destination.parent.mkdir(parents=True, exist_ok=True, mode=0o750) + temporary = destination.parent / f".{destination.name}.{uuid4().hex}.part" + try: + with temporary.open("xb") as handle: + os.chmod(temporary, 0o600) + handle.write(payload) + handle.flush() + os.fsync(handle.fileno()) + try: + os.link(temporary, destination) + os.chmod(destination, 0o440) + return True + except FileExistsError: + if destination.is_symlink() or destination.read_bytes() != payload: + raise PublicSourceLandingError( + "existing landing manifest does not match the staged object" + ) from None + return False + finally: + temporary.unlink(missing_ok=True) + + +def _build_registration( + request: PublicSourceLandingRequest, + *, + payload_path: Path, + landing_root: Path, + content_sha256: str, + size_bytes: int, +) -> tuple[LandingRegistration, str, bytes]: + resource_urn = f"gda://{request.tenant_id}/dataset/{request.dataset_id}" + authority_locator = f"{request.tenant_id}/{request.dataset_id}" + object_key = payload_path.relative_to(landing_root).as_posix() + version_id = uuid5(NAMESPACE_URL, f"{resource_urn}@sha256:{content_sha256}") + artifact_id = uuid5( + NAMESPACE_URL, + f"gda-landing-artifact:{resource_urn}@sha256:{content_sha256}", + ) + manifest = { + "schema": LANDING_SCHEMA, + "admission_class": "public_open", + "tenant_id": request.tenant_id, + "dataset_id": request.dataset_id, + "resource_urn": resource_urn, + "resource_version_id": str(version_id), + "artifact_id": str(artifact_id), + "authority_locator": authority_locator, + "object_key": object_key, + "source_uri": request.source_uri, + "license_id": request.license_id, + "media_type": request.media_type, + "content_sha256": content_sha256, + "size_bytes": size_bytes, + "created_by": request.created_by, + "created_at": request.created_at.isoformat().replace("+00:00", "Z"), + "content_admission_authorized": True, + "production_ready": False, + } + manifest_sha256 = canonical_json_fingerprint(manifest) + resource = Resource( + tenant_id=request.tenant_id, + resource_urn=resource_urn, + resource_kind="dataset", + authority_system="gda_landing", + authority_locator=authority_locator, + owner_ref=request.owner_ref, + governance_ref={ + "admission_class": "public_open", + "source_uri": request.source_uri, + "license_id": request.license_id, + }, + ) + version = ResourceVersion( + tenant_id=request.tenant_id, + resource_urn=resource_urn, + resource_version_id=version_id, + version_key=f"sha256:{content_sha256[:16]}", + content_sha256=content_sha256, + authority_version_ref={ + "authority_system": "gda_landing", + "object_key": object_key, + "manifest_sha256": manifest_sha256, + }, + created_by=request.created_by, + created_at=request.created_at, + ) + artifact = Artifact( + tenant_id=request.tenant_id, + artifact_id=artifact_id, + artifact_key=f"landing:{request.dataset_id}:{content_sha256[:12]}", + artifact_role="input", + storage_uri=payload_path.as_uri(), + media_type=request.media_type, + content_sha256=content_sha256, + size_bytes=size_bytes, + resource_version_id=version_id, + manifest=manifest, + created_by=request.created_by, + created_at=request.created_at, + ) + registration = LandingRegistration( + resource=resource, + resource_version=version, + artifact=artifact, + ) + manifest_document = { + "manifest": manifest, + "manifest_sha256": manifest_sha256, + } + return registration, manifest_sha256, canonical_json_bytes(manifest_document) + b"\n" + + +def stage_public_source( + request: PublicSourceLandingRequest, + *, + source_path: Path, + landing_root: Path, +) -> PublicSourceLandingResult: + """Copy verified public bytes into a content-addressed immutable Landing.""" + if not _DATASET_ID_RE.fullmatch(request.dataset_id): + raise PublicSourceLandingError("dataset_id is not canonical") + if source_path.is_symlink(): + raise PublicSourceLandingError("landing source cannot be a symbolic link") + source_path = source_path.resolve(strict=True) + if landing_root.is_symlink(): + raise PublicSourceLandingError("landing root cannot be a symbolic link") + landing_root = landing_root.resolve() + landing_root.mkdir(parents=True, exist_ok=True, mode=0o750) + staging_root = landing_root / ".staging" + staging_root.mkdir(parents=True, exist_ok=True, mode=0o700) + staging_path = staging_root / f"{uuid4().hex}.part" + try: + content_sha256, size_bytes = _copy_to_staging(source_path, staging_path) + if content_sha256 != request.expected_sha256: + raise PublicSourceLandingError( + "source SHA-256 does not match the approved public-source input" + ) + version_root = ( + landing_root + / request.tenant_id + / request.dataset_id + / "sha256" + / content_sha256 + ) + payload_path = version_root / f"payload{_safe_suffix(source_path)}" + payload_created = _install_staged_payload( + staging_path, + payload_path, + expected_sha256=content_sha256, + expected_size=size_bytes, + ) + registration, manifest_sha256, manifest_bytes = _build_registration( + request, + payload_path=payload_path, + landing_root=landing_root, + content_sha256=content_sha256, + size_bytes=size_bytes, + ) + manifest_path = version_root / "manifest.json" + manifest_created = _install_immutable_bytes(manifest_path, manifest_bytes) + return PublicSourceLandingResult( + registration=registration, + payload_path=str(payload_path), + manifest_path=str(manifest_path), + payload_created=payload_created, + manifest_created=manifest_created, + manifest_sha256=manifest_sha256, + ) + finally: + staging_path.unlink(missing_ok=True) + + +def verify_public_source_landing(result: PublicSourceLandingResult) -> None: + """Re-read Landing bytes and prove their ledger bindings still match.""" + registration = result.registration + artifact = registration.artifact + payload_path = Path(result.payload_path) + manifest_path = Path(result.manifest_path) + _verify_payload(payload_path, artifact.content_sha256, artifact.size_bytes) + try: + document = json.loads(manifest_path.read_text(encoding="utf-8")) + except (OSError, ValueError) as exc: + raise PublicSourceLandingError("landing manifest is unreadable") from exc + manifest = document.get("manifest") + if manifest != artifact.manifest: + raise PublicSourceLandingError("landing manifest does not bind the Artifact") + if document.get("manifest_sha256") != result.manifest_sha256: + raise PublicSourceLandingError("landing manifest fingerprint does not match") + if canonical_json_fingerprint(manifest) != result.manifest_sha256: + raise PublicSourceLandingError("landing manifest content was modified") + if payload_path.as_uri() != artifact.storage_uri: + raise PublicSourceLandingError("landing payload path does not bind the Artifact URI") + + +def _parse_time(value: str) -> datetime: + parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) + if parsed.tzinfo is None or parsed.utcoffset() is None: + raise argparse.ArgumentTypeError("timestamp must include a timezone") + return parsed.astimezone(UTC) + + +def _write_result(result: PublicSourceLandingResult, output: Path | None) -> None: + rendered = json.dumps( + result.model_dump(mode="json"), + ensure_ascii=True, + indent=2, + sort_keys=True, + ) + if output is not None: + output.parent.mkdir(parents=True, exist_ok=True) + output.write_text(rendered + "\n", encoding="utf-8") + print(rendered) + + +def _build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(description=__doc__) + subparsers = parser.add_subparsers(dest="command", required=True) + stage = subparsers.add_parser("stage", help="stage and optionally register bytes") + stage.add_argument("--source-file", type=Path, required=True) + stage.add_argument("--landing-root", type=Path, required=True) + stage.add_argument("--tenant-id", required=True) + stage.add_argument("--dataset-id", required=True) + stage.add_argument("--source-uri", required=True) + stage.add_argument("--license-id", required=True) + stage.add_argument("--owner-ref", required=True) + stage.add_argument("--expected-sha256", required=True) + stage.add_argument("--media-type", required=True) + stage.add_argument("--created-by", required=True) + stage.add_argument("--created-at", type=_parse_time, required=True) + stage.add_argument("--database-url") + stage.add_argument("--output", type=Path) + verify = subparsers.add_parser("verify", help="verify a staged Landing result") + verify.add_argument("--input", type=Path, required=True) + return parser + + +def main(argv: list[str] | None = None) -> int: + args = _build_parser().parse_args(argv) + if args.command == "stage": + request = PublicSourceLandingRequest( + tenant_id=args.tenant_id, + dataset_id=args.dataset_id, + source_uri=args.source_uri, + license_id=args.license_id, + owner_ref=args.owner_ref, + expected_sha256=args.expected_sha256, + media_type=args.media_type, + created_by=args.created_by, + created_at=args.created_at, + ) + result = stage_public_source( + request, + source_path=args.source_file, + landing_root=args.landing_root, + ) + if args.database_url: + gateway = PlatformGateway(create_engine(args.database_url)) + ledger = gateway.register_landing(result.registration) + result = result.model_copy(update={"ledger_created": ledger.created}) + verify_public_source_landing(result) + _write_result(result, args.output) + return 0 + result = PublicSourceLandingResult.model_validate_json( + args.input.read_text(encoding="utf-8") + ) + verify_public_source_landing(result) + print(json.dumps({"valid": True, "manifest_sha256": result.manifest_sha256})) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/data_agent/test_platform_gateway.py b/data_agent/test_platform_gateway.py index 9b84e8d5..1de0df73 100644 --- a/data_agent/test_platform_gateway.py +++ b/data_agent/test_platform_gateway.py @@ -1,6 +1,6 @@ import asyncio import json -from datetime import datetime, timezone +from datetime import UTC, datetime from types import SimpleNamespace from unittest.mock import MagicMock, patch from uuid import UUID @@ -22,8 +22,8 @@ ) from data_agent.platform_gateway import ( COMMAND_OUTBOX_MIGRATION, - DefinitionRegistration, GATEWAY_ROLE_MIGRATION, + DefinitionRegistration, GatewayConflictError, GatewayValidationError, GatewayWriteResult, @@ -31,13 +31,12 @@ build_gateway_report, ) - TENANT = "tenant-a" ACTOR = "human:operator-1" DEFINITION_ID = UUID("00000000-0000-4000-8000-000000000010") RUN_ID = UUID("00000000-0000-4000-8000-000000000020") SOURCE_ID = UUID("00000000-0000-4000-8000-000000000030") -NOW = datetime(2026, 7, 24, 12, 0, tzinfo=timezone.utc) +NOW = datetime(2026, 7, 24, 12, 0, tzinfo=UTC) def _request(*, body=None, path=None, headers=None): @@ -513,8 +512,9 @@ def test_run_transition_rejects_negative_state_version_at_http_boundary(): def test_platform_gateway_routes_are_versioned_and_registered(): registered = routes.get_platform_gateway_routes() - assert len(registered) == 12 + assert len(registered) == 13 assert all(route.path.startswith("/api/platform/v1/") for route in registered) + assert "/api/platform/v1/landings" in {route.path for route in registered} from data_agent.frontend_api import get_frontend_api_routes @@ -526,7 +526,7 @@ def test_platform_gateway_static_contract_and_fail_closed_role(tmp_path): report = build_gateway_report() assert report["status"] == "valid" assert report["database_role"] == "gda_control_gateway" - assert report["route_count"] == 12 + assert report["route_count"] == 13 unsafe = tmp_path / "unsafe_gateway.sql" unsafe.write_text( diff --git a/data_agent/test_public_source_landing.py b/data_agent/test_public_source_landing.py new file mode 100644 index 00000000..6288d797 --- /dev/null +++ b/data_agent/test_public_source_landing.py @@ -0,0 +1,242 @@ +import asyncio +import hashlib +import json +import os +from datetime import UTC, datetime +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import MagicMock, patch + +import pytest +from pydantic import ValidationError + +from data_agent.api import platform_gateway_routes as routes +from data_agent.platform_gateway import GatewayWriteResult, LandingRegistration +from data_agent.public_source_landing import ( + PublicSourceLandingError, + PublicSourceLandingRequest, + main, + stage_public_source, + verify_public_source_landing, +) + +NOW = datetime(2026, 8, 17, 14, 0, tzinfo=UTC) +PAYLOAD = b'{"type":"FeatureCollection","features":[]}\n' +PAYLOAD_SHA256 = hashlib.sha256(PAYLOAD).hexdigest() + + +def _request() -> PublicSourceLandingRequest: + return PublicSourceLandingRequest( + tenant_id="public-demo", + dataset_id="natural-earth-countries", + source_uri=( + "https://raw.githubusercontent.com/nvkelso/natural-earth-vector/" + "v5.1.2/geojson/ne_110m_admin_0_countries.geojson" + ), + license_id="public-domain", + owner_ref="team:data-platform", + expected_sha256=PAYLOAD_SHA256, + media_type="application/geo+json", + created_by="workload:public-source-ingest", + created_at=NOW, + ) + + +def _stage(tmp_path: Path): + source = tmp_path / "countries.geojson" + source.write_bytes(PAYLOAD) + return stage_public_source( + _request(), + source_path=source, + landing_root=tmp_path / "landing", + ) + + +def _api_request(body): + request = MagicMock() + + async def read_json(): + return body + + request.json.side_effect = read_json + request.path_params = {} + request.headers = {"x-request-id": "landing-request"} + return request + + +def _user(*, tenant_id="public-demo", identifier="public-source-ingest"): + return SimpleNamespace( + identifier=identifier, + metadata={ + "role": "platform_operator", + "tenant_id": tenant_id, + "subject_type": "workload", + }, + ) + + +def test_stage_public_source_creates_and_replays_immutable_landing(tmp_path): + result = _stage(tmp_path) + verify_public_source_landing(result) + + payload = Path(result.payload_path) + manifest = Path(result.manifest_path) + assert payload.read_bytes() == PAYLOAD + assert payload.stat().st_mode & 0o777 == 0o440 + assert manifest.stat().st_mode & 0o777 == 0o440 + assert f"sha256/{PAYLOAD_SHA256}/payload.geojson" in result.payload_path + assert result.payload_created is True + assert result.manifest_created is True + assert result.ledger_created is None + + registration = result.registration + assert registration.resource.authority_system == "gda_landing" + assert registration.resource_version.content_sha256 == PAYLOAD_SHA256 + assert registration.artifact.content_sha256 == PAYLOAD_SHA256 + assert registration.artifact.run_id is None + assert registration.artifact.manifest["content_admission_authorized"] is True + assert registration.artifact.manifest["production_ready"] is False + + replay = _stage(tmp_path) + verify_public_source_landing(replay) + assert replay.registration == registration + assert replay.payload_created is False + assert replay.manifest_created is False + + +def test_stage_rejects_unapproved_or_symlinked_bytes(tmp_path): + source = tmp_path / "countries.geojson" + source.write_bytes(PAYLOAD) + bad_request = _request().model_copy(update={"expected_sha256": "0" * 64}) + with pytest.raises(PublicSourceLandingError, match="does not match"): + stage_public_source( + bad_request, + source_path=source, + landing_root=tmp_path / "landing", + ) + assert not tuple((tmp_path / "landing").glob("public-demo/**/payload*")) + + link = tmp_path / "linked.geojson" + link.symlink_to(source) + with pytest.raises(PublicSourceLandingError, match="symbolic link"): + stage_public_source( + _request(), + source_path=link, + landing_root=tmp_path / "other-landing", + ) + + +def test_verify_detects_payload_and_manifest_tampering(tmp_path): + result = _stage(tmp_path) + payload = Path(result.payload_path) + os.chmod(payload, 0o640) + payload.write_bytes(b"tampered") + with pytest.raises(PublicSourceLandingError, match="does not match its key"): + verify_public_source_landing(result) + + other = tmp_path / "other" + source = other / "countries.geojson" + source.parent.mkdir() + source.write_bytes(PAYLOAD) + result = stage_public_source( + _request(), source_path=source, landing_root=other / "landing" + ) + manifest = Path(result.manifest_path) + os.chmod(manifest, 0o640) + document = json.loads(manifest.read_text(encoding="utf-8")) + document["manifest"]["license_id"] = "unknown" + manifest.write_text(json.dumps(document), encoding="utf-8") + with pytest.raises(PublicSourceLandingError, match="does not bind"): + verify_public_source_landing(result) + + +def test_public_request_and_registration_fail_closed(tmp_path): + request_document = _request().model_dump(mode="python") + request_document["source_uri"] = "file:///private/source.geojson" + with pytest.raises(ValidationError, match="must use HTTPS"): + PublicSourceLandingRequest.model_validate(request_document) + request_document = _request().model_dump(mode="python") + request_document["created_by"] = "agent:planner" + with pytest.raises(ValidationError, match="human or workload"): + PublicSourceLandingRequest.model_validate(request_document) + + result = _stage(tmp_path) + with pytest.raises(ValidationError, match="public_open admission"): + LandingRegistration( + resource=result.registration.resource, + resource_version=result.registration.resource_version, + artifact=result.registration.artifact.model_copy( + update={ + "manifest": { + **result.registration.artifact.manifest, + "admission_class": "protected", + } + } + ), + ) + + +def test_landing_api_registers_atomically_and_enforces_actor(tmp_path): + result = _stage(tmp_path) + registration = result.registration + gateway = MagicMock() + gateway.register_landing.return_value = GatewayWriteResult(registration, True) + request = _api_request(registration.model_dump(mode="json")) + with ( + patch.object(routes, "_get_user_from_request", return_value=_user()), + patch.object(routes, "_gateway", return_value=gateway), + ): + response = asyncio.run(routes.create_landing(request)) + assert response.status_code == 201 + assert json.loads(response.body)["created"] is True + gateway.register_landing.assert_called_once_with(registration) + + request = _api_request(registration.model_dump(mode="json")) + with patch.object( + routes, + "_get_user_from_request", + return_value=_user(identifier="different-workload"), + ): + response = asyncio.run(routes.create_landing(request)) + assert response.status_code == 403 + assert json.loads(response.body)["error"]["code"] == "actor_mismatch" + + +def test_cli_stages_and_verifies_landing(tmp_path, capsys): + source = tmp_path / "countries.geojson" + source.write_bytes(PAYLOAD) + output = tmp_path / "landing-result.json" + exit_code = main( + [ + "stage", + "--source-file", + str(source), + "--landing-root", + str(tmp_path / "landing"), + "--tenant-id", + "public-demo", + "--dataset-id", + "natural-earth-countries", + "--source-uri", + _request().source_uri, + "--license-id", + "public-domain", + "--owner-ref", + "team:data-platform", + "--expected-sha256", + PAYLOAD_SHA256, + "--media-type", + "application/geo+json", + "--created-by", + "workload:public-source-ingest", + "--created-at", + "2026-08-17T14:00:00Z", + "--output", + str(output), + ] + ) + assert exit_code == 0 + assert output.is_file() + capsys.readouterr() + assert main(["verify", "--input", str(output)]) == 0 + assert json.loads(capsys.readouterr().out)["valid"] is True diff --git a/data_agent/test_public_source_landing_postgres.py b/data_agent/test_public_source_landing_postgres.py new file mode 100644 index 00000000..f99e69ab --- /dev/null +++ b/data_agent/test_public_source_landing_postgres.py @@ -0,0 +1,144 @@ +import hashlib +import os +from datetime import UTC, datetime +from pathlib import Path +from uuid import uuid4 + +import pytest +from sqlalchemy import create_engine, text + +from data_agent.platform_gateway import ( + GatewayConflictError, + LandingRegistration, + PlatformGateway, +) +from data_agent.public_source_landing import ( + PublicSourceLandingRequest, + stage_public_source, +) + +DATABASE_URL = os.environ.get("DATABASE_URL") +MIGRATIONS = tuple( + Path(__file__).resolve().parent / "migrations" / filename + for filename in ( + "092_platform_control_ledger.sql", + "093_app_user_tenant_context.sql", + "094_platform_control_gateway.sql", + ) +) +NOW = datetime(2026, 8, 17, 14, 0, tzinfo=UTC) + + +def _stage(tmp_path: Path, *, tenant: str, dataset_id: str, payload: bytes): + source = tmp_path / f"{dataset_id}.geojson" + source.write_bytes(payload) + sha256 = hashlib.sha256(payload).hexdigest() + return stage_public_source( + PublicSourceLandingRequest( + tenant_id=tenant, + dataset_id=dataset_id, + source_uri=f"https://example.org/open-data/{dataset_id}.geojson", + license_id="CC0-1.0", + owner_ref="team:data-platform", + expected_sha256=sha256, + media_type="application/geo+json", + created_by="workload:public-source-ingest", + created_at=NOW, + ), + source_path=source, + landing_root=tmp_path / "landing", + ) + + +@pytest.mark.skipif(not DATABASE_URL, reason="DATABASE_URL is not configured") +def test_postgres_registers_landing_atomically_and_replays(tmp_path): + engine = create_engine(DATABASE_URL) + tenant = f"landing-{uuid4().hex[:12]}" + try: + with engine.begin() as connection: + is_superuser = connection.exec_driver_sql( + "SELECT rolsuper FROM pg_roles WHERE rolname = current_user" + ).scalar_one() + if not is_superuser: + pytest.skip("landing gateway test requires a PostgreSQL superuser") + connection.exec_driver_sql( + """ + CREATE TABLE IF NOT EXISTS agent_app_users ( + id SERIAL PRIMARY KEY, + username VARCHAR(100) UNIQUE NOT NULL + ) + """ + ) + for migration in MIGRATIONS: + connection.execute(text(migration.read_text(encoding="utf-8"))) + + first = _stage( + tmp_path, + tenant=tenant, + dataset_id="natural-earth-countries", + payload=b'{"type":"FeatureCollection","features":[]}\n', + ) + gateway = PlatformGateway(engine) + created = gateway.register_landing(first.registration) + assert created.created is True + assert created.value == first.registration + replay = gateway.register_landing(first.registration) + assert replay.created is False + assert replay.value == first.registration + + with engine.connect() as connection: + counts = connection.execute( + text( + """ + SELECT + (SELECT count(*) FROM gda_control.resource + WHERE tenant_id = :tenant_id), + (SELECT count(*) FROM gda_control.resource_version + WHERE tenant_id = :tenant_id), + (SELECT count(*) FROM gda_control.artifact + WHERE tenant_id = :tenant_id) + """ + ), + {"tenant_id": tenant}, + ).one() + assert counts == (1, 1, 1) + + second = _stage( + tmp_path, + tenant=tenant, + dataset_id="natural-earth-rivers", + payload=b'{"type":"FeatureCollection","features":[{}]}\n', + ) + conflicting = LandingRegistration( + resource=second.registration.resource, + resource_version=second.registration.resource_version, + artifact=second.registration.artifact.model_copy( + update={"artifact_id": first.registration.artifact.artifact_id} + ), + ) + with pytest.raises(GatewayConflictError, match="different payload"): + gateway.register_landing(conflicting) + + with engine.connect() as connection: + rolled_back = connection.execute( + text( + """ + SELECT + (SELECT count(*) FROM gda_control.resource + WHERE tenant_id = :tenant_id AND resource_urn = :resource_urn), + (SELECT count(*) FROM gda_control.resource_version + WHERE tenant_id = :tenant_id + AND resource_version_id = :resource_version_id) + """ + ), + { + "tenant_id": tenant, + "resource_urn": second.registration.resource.resource_urn, + "resource_version_id": ( + second.registration.resource_version.resource_version_id + ), + }, + ).one() + assert rolled_back == (0, 0) + finally: + engine.dispose() diff --git a/docs/architecture-decisions/adr-081-public-source-immutable-landing.md b/docs/architecture-decisions/adr-081-public-source-immutable-landing.md new file mode 100644 index 00000000..4287a83a --- /dev/null +++ b/docs/architecture-decisions/adr-081-public-source-immutable-landing.md @@ -0,0 +1,73 @@ +# ADR-081: Public/Open Source Immutable Landing + +**Status**: Accepted + +**Date**: 2026-08-17 + +**Decision owners**: Platform Architecture, Data Platform, Data Governance + +## Context + +AR-2 needs a working source-to-platform entry point while the Chongqing +protected source remains blocked on external governance attestations. The +existing control ledger already owns `Resource`, `ResourceVersion` and +`Artifact`; creating another Landing registry would repeat the authority split +that AR-0 is intended to remove. + +The first executable slice must copy actual bytes, prove their checksum, make +replay idempotent, and leave an auditable binding without reading protected +source payloads or granting scheduler/provider mutation authority. + +## Options considered + +| Option | Benefit | Limitation | Decision | +|---|---|---|---| +| Add a dedicated Landing database/table | Explicit schema | Creates a second version and artifact authority | Rejected | +| Register a path and trust later ingestion | Small implementation | Does not prove bytes or protect against replacement | Rejected | +| Content-addressed local Landing plus atomic existing-ledger registration | Real byte-level behavior, deterministic replay, no new authority | Local profile is not object-store or production evidence | Adopted | + +## Decision + +Adopt `data_agent.public_source_landing` for the public/open profile: + +1. Require an HTTPS source URI without credentials, query, or fragment, an + explicit license identifier, an expected SHA-256, owner, media type, and + controlled actor. +2. Copy a regular non-symlink source file into a content-addressed path under + `//sha256//payload.` using atomic + no-overwrite installation. The manifest is installed alongside the payload + with restrictive permissions and the same no-overwrite rule. +3. Represent the Landing authority using the existing `gda_control.resource`, + `resource_version` and `artifact` rows. The gateway registers all three in + one transaction through `register_landing`; a conflict rolls back the + complete registration. +4. Expose `/api/platform/v1/landings` for an authenticated same-tenant actor + to register an already staged object. The CLI performs the local byte copy + and can invoke the same gateway transaction. +5. Keep `admission_class=public_open`, `content_admission_authorized=true` and + `production_ready=false` explicit in the manifest. This path is not allowed + to reinterpret M3-31/M3-32/M3-33 Chongqing evidence or bypass the protected + workflow. + +## Consequences + +**Positive**: the platform now has a real immutable byte Landing and ledger +identity with deterministic replay and conflict rollback, without another +registry or queue. + +**Negative**: the current implementation is a local filesystem profile. It +does not prove object-store locking, cloud identity, DataOps scheduling, +Bronze/Silver/Gold transformation, or production readiness. + +**Next**: bind this public Landing ResourceVersion to a minimal DataOps +definition and PlatformRun, then materialize a small GeoJSON/ZIP slice to the +lightweight serving profile. Separately provision the protected Chongqing +environment and real attestations. + +## Verification + +- `data_agent/test_public_source_landing.py` +- `data_agent/test_public_source_landing_postgres.py` +- Natural Earth 110m public-domain ZIP smoke: content SHA-256 + `0f243aeac8ac6cf26f0417285b0bd33ac47f1b5bdb719fd3e0df37d03ea37110`, + 214,976 bytes, replay and verify passed. diff --git a/docs/evidence/public-source-landing-2026-08-17.json b/docs/evidence/public-source-landing-2026-08-17.json new file mode 100644 index 00000000..fe0fbfff --- /dev/null +++ b/docs/evidence/public-source-landing-2026-08-17.json @@ -0,0 +1,21 @@ +{ + "schema": "gda.public_source_landing.evidence.v1", + "evidence_date": "2026-08-17", + "source_label": "natural-earth-admin0-countries", + "source_uri": "https://naturalearth.s3.amazonaws.com/110m_cultural/ne_110m_admin_0_countries.zip", + "license_id": "public-domain", + "media_type": "application/zip", + "content_sha256": "0f243aeac8ac6cf26f0417285b0bd33ac47f1b5bdb719fd3e0df37d03ea37110", + "size_bytes": 214976, + "resource_urn": "gda://public-demo/dataset/natural-earth-admin0-countries", + "resource_version_id": "905bc019-71ef-5207-8392-cd2c46b7b5b7", + "artifact_id": "464cfb4a-cb1a-5ca5-8993-905ba22443a9", + "manifest_sha256": "bd884360f4138048fff1caba84821b93aa57b9d21f1266a36cccd6d8115d8ac9", + "payload_created": true, + "replay_payload_created": false, + "replay_manifest_created": false, + "verify_valid": true, + "ledger_registered": false, + "production_ready": false, + "protected_chongqing_admission_unchanged": true +} diff --git a/docs/roadmap.md b/docs/roadmap.md index f1db2981..1bae86ff 100644 --- a/docs/roadmap.md +++ b/docs/roadmap.md @@ -445,6 +445,7 @@ AR-0 Architecture / Schema / Runtime Truth - M3-31 protected admission readiness contract：绑定 M3-28/M3-29/M3-30 的逻辑与文件 fingerprints，把 6 项 derivation、8 项 governance 和 1 项 protected attestation 固定为 15 个 fail-closed requirements；`admission_eligible=false` 前不得创建 Landing、ResourceVersion、PlatformRun、scheduler submission 或 provider mutation。 - M3-32 protected admission attestation intake：固定外部 attestation 的 M3-31 logical/file fingerprint binding、15 项逐项 SHA-256 证明、受保护 verifier/source binding、24 小时 freshness、七天 validity 上限和八项 no-payload/no-mutation checks；`evaluate` 只产生 fingerprinted readiness report，不创建任何 Landing、ResourceVersion、PlatformRun、scheduler submission 或 provider mutation authority。 - M3-33 protected admission verifier workflow：只允许从 `main` 手工触发受保护 `chongqing-admission` environment 的专用 verifier runner，消费 metadata-only secret bundle，绑定 exact M3-31 fingerprints，执行 M3-32 evaluate/verify，并以 GitHub OIDC provenance attestation 固定 input/report；workflow 不含 source scan、Landing/ResourceVersion/PlatformRun 创建、scheduler submission 或 provider client，环境/runner/15 项真实 attestation 未 provision 前仍 blocked。 +- M3-34 public/open-source immutable Landing slice：以显式 HTTPS source、license、owner、expected SHA-256 和 controlled actor 将真实 public-domain bytes 写入 content-addressed local Landing；复用现有 Resource/ResourceVersion/Artifact authority，通过单事务 gateway registration 保证幂等 replay 和冲突回滚；Natural Earth 110m smoke 已通过，public profile 仍非 production-ready，也不改变 Chongqing protected admission。 - SourceDefinition、CredentialReference、SourceCapability、SyncDefinition/Version、SyncRun、Cursor/Watermark、SchemaDriftEvent 和 Reconciliation。 - 数据库、对象存储/空间文件、HTTP/STAC 三类代表 source 的连接、凭据、连通、发现、preview、profile 和 owner 登记。 - 全量/增量微批的 Append/Overwrite/Merge 策略，以及至少一个真实 CDC 或事件流 source 通过 Flink 写入版本化 Bronze；覆盖 watermark/offset、checkpoint、迟到/乱序、源端删除、幂等、对账、重放和失败恢复。 @@ -692,7 +693,7 @@ Golden checks 至少覆盖： 3. 冻结 ResourceURN、ResourceVersion、PlatformDefinition/PlatformRun/FrameworkAttemptObservation/Artifact/LineageEvent、SubjectContext 与 storage/table/compute provider 最小合同。 4. 分阶段实现 `gda-metadata-fabric-bridge`：M1 只读 mapping/reconciliation、M2a 本地 foundation/重启连续性、M2b-1 本地三存储恢复、M2b-2 隔离 versioned/Object-Locked repository round-trip、M2b-3 本机双集群 + Kubernetes 外 COMPLIANCE repository + 独立 writer/reader、M2c-1 provider-native metrics、M2c-2 临时 OTel Collector + JSON Exporter 的双周期本地 pipeline、M2c-3 本地单 job scrape 故障检测/配置恢复/完整清理 evidence、M2c-4 绑定 source revision 的 production observability readiness contract，以及 M2d-1 本地 kindnet 跨节点 NetworkPolicy enforcement 已验证；M2c-4 当前仍有 20 项 blockers，M2d-1 也未验证生产 provider policy 或 tenant isolation。下一步完成 source host/cluster 外的生产 bucket、KMS/TLS/workload identity、source-loss recovery 与 RPO/RTO，并批准 metrics backend、retention、OTel/TLS、tenant、alert/SLO/owner 后在受保护环境验证持续采集、存储、查询、真实告警投递、runbook 响应和 provider NetworkPolicy；再推进 OIDC、upgrade/rollback、registry provenance 和 owner/runbook；之后才进入 M3 ingestion/OpenLineage/conformance。 5. 实现 `gda-orchestration-gateway`、DolphinScheduler process/task/schedule/complement/worker-group、Spark/Flink provider task adapter 和故障注入；不再开发新的 lease/queue/scheduler。 -6. M3-28 已冻结全量重庆真实源的 path-free physical/metadata admission baseline，M3-29 已建立 metadata-only extraction provenance gap baseline，M3-30 已选择 `bishan_land_use_dltb_local` 并把八项治理输入固化为 fail-closed pending records，M3-31 已将三层证据统一成 protected admission readiness contract，M3-32 已固定 protected attestation intake/evaluate/verify boundary，M3-33 已固定受保护 workflow execution/provenance boundary；下一步 provision 专用 environment/runner 并取得 15 项外部 attestation 后执行首次 protected verification，未获批准前不得 content admission。 +6. M3-28 已冻结全量重庆真实源的 path-free physical/metadata admission baseline，M3-29 已建立 metadata-only extraction provenance gap baseline，M3-30 已选择 `bishan_land_use_dltb_local` 并把八项治理输入固化为 fail-closed pending records，M3-31 已将三层证据统一成 protected admission readiness contract，M3-32 已固定 protected attestation intake/evaluate/verify boundary，M3-33 已固定受保护 workflow execution/provenance boundary，M3-34 已用 Natural Earth public-domain bytes 验证 immutable Landing 与 ledger registration path；下一步用 public Landing 绑定最小 DataOps PlatformRun 并物化轻量 serving slice，同时 provision 专用 environment/runner 并取得 Chongqing 的 15 项外部 attestation，未获批准前不得 content admission。 7. 冻结 Default Lakehouse、Cloud Managed、Lightweight Integrated profiles；以统一 Run 完成默认 MinIO/Iceberg/Spark/Flink、轻量 PostGIS/DuckDB 和 Azure 代表 adapter 的 conformance smoke。 8. 实现跨 profile 的 Raw -> ODS -> DIM/DWD -> DWS -> ADS 通用生产、质量、发布、回滚和 golden equivalence。 9. 建立 DataProductBlueprint、模型版本和 Visual/SQL/Notebook 共用 definition 的 Build 工作台，打通 preview、test、publish、approval 和 rollback。 @@ -733,7 +734,7 @@ AR-4 parity/control gate 退出前暂停以下主线扩张： |---|---|---| | AR-0 Architecture/Schema/Runtime Truth Freeze | `in_progress` | 全环境 schema/config fingerprint、迁移 fail-closed、事实清单、storage/compute/GIS serving provider profile/capability、ADR-017 benchmark、owner/SLO 和首条数据/服务验收集冻结 | | AR-1 Unified Metadata + Orchestration Control Planes | `in_progress` | controlled gateway、DolphinScheduler adapter、Metadata Fabric M1/M2a/M2b、M2c-1 provider metrics、M2c-2 本地临时 OTel pipeline、M2c-3 本地 scrape failure/recovery、M2c-4 production observability readiness contract 与 M2d-1 本地跨节点 NetworkPolicy enforcement 已验证，生产观测和生产 policy/tenant isolation 仍 blocked；下一证据是 source host/cluster 外的生产 recovery、持久 metrics backend/TLS/tenant/真实 alert delivery/SLO、OIDC、受保护 provider NetworkPolicy、升级回滚/registry provenance，以及受控 ingestion/replay 与无双写验收 | -| AR-2 Source/Ingestion + Geospatial Lakehouse Vertical Slice | `in_progress` | M3-28 admission baseline、M3-29 extraction provenance gap baseline、M3-30 first-candidate governance gate、M3-31 protected admission readiness、M3-32 attestation intake 与 M3-33 protected verifier workflow contract 已完成；checked baseline 仍为 `admission_eligible=false`，下一证据是专用 environment/runner provisioning、15 项外部 attestation 和首次 protected verifier run，随后才可设计 immutable Landing authority、content admission 和 ingestion；最终仍需三类代表源、`DriveTransfer`、大文件恢复、默认湖仓、轻量存算一体及 Azure 代表 adapter 通过 conformance 与 Raw -> ADS 验收 | +| AR-2 Source/Ingestion + Geospatial Lakehouse Vertical Slice | `in_progress` | M3-28 admission baseline、M3-29 extraction provenance gap baseline、M3-30 first-candidate governance gate、M3-31 protected admission readiness、M3-32 attestation intake、M3-33 protected verifier workflow contract 与 M3-34 public immutable Landing slice 已完成；public profile 已有真实 bytes/manifest/ledger path，但 DataOps PlatformRun、Bronze/Silver/Gold、serving 和 rollback 尚未完成；checked Chongqing baseline 仍为 `admission_eligible=false`，下一证据是 public Landing 的最小 DataOps run，以及专用 environment/runner provisioning、15 项外部 attestation 和首次 protected verifier run | | AR-3 Data Product Engineering + Governance Workbench | `planned` | Blueprint、模型、Visual/SQL/Notebook、DataOps CI/CD、质量/安全/审批共用 definition 和产品生命周期 | | AR-4 Asset/GIS Service/Spatial Experience Operations | `planned` | Service Control Plane、Features/Tiles/MVT/COG/STAC/export 及条件 legacy OGC/3D/EDR provider、Gateway/权限/缓存、原子切换/回滚、Discover/Operate/Govern 和无 LLM 多入口通过 conformance/parity/control gate | | AR-5 AgentOps Runtime + UX Uplift | `planned` | DataOps parity/control 通过；Agent bundle eval、deployment、online observation、incident/rollback 和 uplift gate | diff --git a/docs/system-of-record-matrix-2026-07-24.md b/docs/system-of-record-matrix-2026-07-24.md index e3406970..81f5c696 100644 --- a/docs/system-of-record-matrix-2026-07-24.md +++ b/docs/system-of-record-matrix-2026-07-24.md @@ -2,7 +2,7 @@ 日期：2026-08-17 -阶段：AR-0 `in_progress`；AR-1 gateway、成功终局 evidence gate、DolphinScheduler adapter sandbox POC、Metadata Fabric M1/M2、M2c-4/M2d-2 production readiness contracts、M3-1/M3-2、M3-3 local binding ledger、M3-4 local OpenLineage wire delivery、M3-5 local OpenMetadata bounded identity、M3-6 local Gravitino Basic bounded identity 与 M3-7 production identity readiness contract 已验证；AR-2 M3-28 重庆真实源 metadata-only admission baseline、M3-29 extraction provenance gap baseline、M3-30 first-candidate governance gate、M3-31 protected admission readiness、M3-32 attestation intake 与 M3-33 protected verifier workflow contract 已建立，内容准入、Landing authority、生产 provider ingestion、生产观测、生产 policy/tenant isolation、生产 identity attestation 和生产切换仍 `in_progress` +阶段：AR-0 `in_progress`；AR-1 gateway、成功终局 evidence gate、DolphinScheduler adapter sandbox POC、Metadata Fabric M1/M2、M2c-4/M2d-2 production readiness contracts、M3-1/M3-2、M3-3 local binding ledger、M3-4 local OpenLineage wire delivery、M3-5 local OpenMetadata bounded identity、M3-6 local Gravitino Basic bounded identity 与 M3-7 production identity readiness contract 已验证；AR-2 M3-28 重庆真实源 metadata-only admission baseline、M3-29 extraction provenance gap baseline、M3-30 first-candidate governance gate、M3-31 protected admission readiness、M3-32 attestation intake、M3-33 protected verifier workflow contract 与 M3-34 public immutable Landing slice 已建立，public profile 的 DataOps run/serving、重庆内容准入、生产 provider ingestion、生产观测、生产 policy/tenant isolation、生产 identity attestation 和生产切换仍 `in_progress` 适用分支：`main` @@ -21,7 +21,7 @@ | 部署配置策略 | Compose/K8s/进程环境；`platform_truth.CONFIG_SPECS` 定义关键类型与策略；DolphinScheduler worker 有默认零副本、外部 ConfigMap/Secret 驱动的 Kustomize 模板、静态 validator、staging activation preflight 和受保护的单副本 activation admission/workflow | `.env` 仅补默认；脱敏 snapshot、Secret key attestation、未扩容 Deployment、`ready_for_activation` 和未执行的 activation workflow 都是观测/变更能力 | 版本化 DeploymentProfile + secret reference；部署环境始终优先；模板、preflight 或 admission 通过都不等于环境已启用 | Platform/SRE/Security | AR-0，部分实现；worker 激活合同已验证、真实运行待审批 | | 环境发布与晋级 | publisher `31862363442`、protected verifier `31862984294` 与 staging deploy/observe `31863077257` 已将 `main@5fffc85`、GHCR digest、attested release、cluster/namespace identity 和 live revision 绑定；schema/config/runtime/health/rollout 通过，golden slice 缺失使 promotion fail closed | 旧 mainline、feature branch、CI artifact、JSON、离线 report、单独的 staging deployment 或人工批准都不能成为 production 发布权威 | 由受保护 environment 的 DeploymentRevision 绑定 OCI、provenance artifact、release manifest、golden slice 与全部 live verdict | Platform/SRE/Security/Repository Owner | AR-1 真实 staging 已部署 -> golden slice/production exit gates 待完成 | | 后台运行时清单 | `platform_truth.RUNTIME_INVENTORY` 是代码层登记；`gda_control` 已有受控 PlatformRun 写入口；DolphinScheduler adapter、tenant-scoped managed command worker 与受保护的单副本激活边界已有代码和测试，但 worker 尚未在 staging 运行；M2b recovery runner、M2c-1 provider probe、M2c-2 `_OtelPortForward`、M2c-3 failure rehearsal 与 M2d-1 NetworkPolicy rehearsal 均登记为 `local_verification_only`，不是 scheduler、worker、持续监控、生产 policy controller 或状态权威 | AST primitive report、worker status JSON、FrameworkAttemptObservation、DolphinScheduler instance state、本地 recovery/metrics/network-policy evidence | PlatformRun ledger 唯一登记最终状态；framework/provider attempt 只能回报观测；worker status 仅为进程健康投影；本地演练进程与 evidence 不得变成生产控制器、监控后端或 tenant-isolation 权威 | Platform Architecture | AR-1 adapter/worker/activation 合同已验证 -> staging 运行待审批；metadata recovery/metrics/policy runner 仅本地验证 | -| 原始文件/对象 | M3-28 已以 path-free metadata manifest 记录重庆真实源 archive/extracted fingerprints、source groups、asset profiles 与治理 blockers；M3-29/M3-30 固定派生证明缺口和首条候选的八项 pending 治理决策；M3-31/M3-32 将其绑定为 15 项 pending requirements 与 protected intake contract；M3-33 已定义 protected verifier workflow，但未 provision 或执行；`source_content_admitted=false`，因此尚无内容写入权威 | admission/provenance/governance/readiness/evaluation/workflow evidence、archive/extracted comparison、临时上传、下载缓存、预览文件 | 在专用 environment/runner 中由 15 项外部 attestation 产生受保护 verifier report，并经独立 admission decision 后，才可由 immutable Landing object URI + checksum + retention 建立内容权威；本地 scratch、checked contract 与 workflow 文件均不可替代 Landing | Data Platform | AR-2 `in_progress` | +| 原始文件/对象 | M3-28 已以 path-free metadata manifest 记录重庆真实源 archive/extracted fingerprints、source groups、asset profiles 与治理 blockers；M3-29/M3-30 固定派生证明缺口和首条候选的八项 pending 治理决策；M3-31/M3-32 将其绑定为 15 项 pending requirements 与 protected intake contract；M3-33 已定义 protected verifier workflow，但未 provision 或执行；M3-34 已用 Natural Earth public-domain ZIP 验证 content-addressed local Landing、manifest、ResourceVersion 和 input Artifact 的实际路径；重庆 `source_content_admitted=false`，public Landing 也尚未绑定 DataOps Run 或 production serving | public-source Landing bytes/manifest、Resource/ResourceVersion/Artifact、admission/provenance/governance/readiness/evaluation/workflow evidence、archive/extracted comparison、临时上传、下载缓存、预览文件 | public/open profile 可由受控 DataOps definition/Run 消费；重庆仍必须在专用 environment/runner 中由 15 项外部 attestation 产生受保护 verifier report，并经独立 admission decision 后才可建立其 immutable Landing authority；本地 scratch、checked contract 与 workflow 文件均不可替代生产 Landing | Data Platform | AR-2 `in_progress` | | 湖仓表与 snapshot | Iceberg/STAC/S3A 有局部实现，尚无通用发布权威 | STAC item、GeoParquet export | Iceberg catalog snapshot 是分析表版本权威；对象是物理内容，STAC 是发现投影 | Data Platform | AR-2 | | 在线空间数据 | PostGIS 业务表是当前编辑/查询事实，部分临时表混入 | Martin MVT、API JSON、导出文件 | 已批准 DataProductVersion 物化到 PostGIS；不能由瓦片或临时表反向定义产品版本 | GIS/Data Platform | AR-2 -> AR-4 | | 数据资产身份与版本 | `gda_control.resource/resource_version` 已实现 identity、hash、predecessor、tenant FK 和幂等 gateway 写入；`agent_data_assets`、`agent_asset_versions` 仍是兼容写路径 | UI catalog、search index、STAC | GDA ledger 管身份与版本绑定；旧行只有在 tenant、authority identity、checksum 和 version evidence 完整时才可形成 eligible plan；OpenMetadata 管治理目录，Gravitino 管技术对象映射 | Metadata Platform | AR-1 gateway 已验证 -> 生产切换待验收 | @@ -68,6 +68,7 @@ 22. M3-31 只允许绑定 M3-28/M3-29/M3-30 并记录 6 项 derivation、8 项 governance 与 1 项 protected attestation 的 pending requirements；`admission_eligible=false` 时不形成 Landing object、ResourceVersion、PlatformRun、scheduler submission、provider mutation、content admission 或 production ingestion 权威，readiness evidence 不得被 rehash 成批准。 23. M3-32 只允许受保护 verifier 消费绑定 M3-31 logical/file fingerprints 的外部 attestation；15 项 requirement、八项 no-payload/no-mutation checks、freshness、expiry 和 verifier/source binding 任一缺失即 blocked。完整 attestation 只能使 fingerprinted report 的 `admission_eligible=true`，不创建 Landing object、ResourceVersion、PlatformRun、scheduler submission、provider mutation、content admission 或 production readiness 权威；合成测试 attestation 不计入生产证据。 24. M3-33 只允许 `main` 上手工触发的受保护 `chongqing-admission` workflow 在专用 runner 中消费 metadata-only secret bundle，绑定 M3-31 exact logical/file fingerprints，执行 M3-32 evaluate/verify 并 attested/upload input/report；workflow 不读取 source payload、不调用 connector/provider/scheduler、不创建 Landing object、ResourceVersion、PlatformRun 或任何 ingestion authority。environment、runner、reviewer policy、secret rotation 与真实 15 项 attestation 未 provision 前，workflow contract 不计入 protected admission evidence。 +25. M3-34 public/open Landing 只允许显式 HTTPS source、license、owner、expected SHA-256 和 controlled actor；payload 与 manifest 以 content-addressed no-overwrite 文件写入，Resource、ResourceVersion、input Artifact 必须由同一 gateway transaction 登记并可幂等 replay。该 public profile 不创建 PlatformRun、不代表 Chongqing protected admission、不证明 object-lock、provider identity、production serving 或 production readiness。 ## 已建立的 AR-0/AR-1 entry 证据 @@ -105,6 +106,7 @@ - AR-2 M3-31 已建立 protected admission readiness contract：候选为 `bishan_land_use_dltb_local`，evidence SHA 为 `2f5ae24ab904af0eed18ee7c517ab5c4638cbdf0923c9345b0041af185d25591`，evidence file SHA 为 `c595065e152988529ff12e2301d59caebb31d2889658a676c9d1f8239e6f8372`；证据绑定 M3-28/M3-29/M3-30 的逻辑与文件 fingerprints，15 项要求全部为 `missing`，`admission_eligible=false`、`source_content_admitted=false`、`production_ready=false`。该证据只定义未来受保护 verifier 的输入，不含 source payload、绝对路径、记录值或 geometry，也不证明任何 Landing authority、ResourceVersion、Run、scheduler/provider mutation 或生产 ingestion。 - AR-2 M3-32 已建立 protected admission attestation intake/evaluate/verify contract：外部输入必须绑定 M3-31 logical/file fingerprints，逐项提供 15 个 SHA-256 attestation、八项 protected checks、verifier/source binding 和受控 freshness/expiry；checked baseline 的 `readiness_valid=true`、`attestation_valid=false`、`admission_eligible=false`，所有 content/authority/production claims 继续为 `false`。合成完整 attestation 仅覆盖 evaluator 单元测试，不计入真实准入证据。 - AR-2 M3-33 已建立 protected admission verifier workflow contract：workflow 仅从 `main` 手工触发、绑定专用 environment/runner、以 restrictive umask 解码 metadata-only attestation secret、执行 M3-32 evaluate/verify 并通过 GitHub OIDC provenance attestation 固定 input/report；checked workflow 未执行，专用环境、runner、reviewer policy、secret rotation 与真实 15 项 attestation 均未 provision，不形成 content admission 或生产 authority。 +- AR-2 M3-34 已用 Natural Earth 110m public-domain ZIP 完成 public/open immutable Landing smoke：214,976 bytes 的 content SHA、content-addressed payload、只读 manifest、ResourceVersion、input Artifact、replay 和 verify 均通过；该 smoke 的 `ledger_registered=false` 只表示 CLI 未连接生产数据库，独立 PostgreSQL 集成测试已验证同一原子 registration/replay/conflict rollback contract，public profile 仍 `production_ready=false`。 ## 下一验收证据 @@ -113,5 +115,5 @@ - 为受保护 activation 提供真实 ConfigMap snapshot、Secret key attestation、provider identity 和 reviewer approval，随后完成 managed outbox worker/provider callback 单副本实际部署、唯一 worker ID、status/lease 故障恢复和无双写证据； - 首条真实图斑链对 golden slice 的 output hash、独立质量结果/evidence、血缘、发布 revision 和 rollback 演练； - OpenMetadata/Gravitino 的 source host/cluster 外生产 backup account/bucket、KMS/TLS/workload identity、PITR/source-loss recovery、RPO/RTO、OIDC、受保护环境 provider NetworkPolicy/tenant isolation、upgrade/rollback、registry provenance、持续 metrics backend/retention/query、真实 alert delivery/SLO owner/runbook，以及受保护 PolicyDecision/Approval、双 provider 最小权限 ingestion、生产持久 binding、受保护 production OpenLineage receiver、无双写 read-back 和 conformance；M1 fixture、M2 本地 evidence/readiness contracts、M3-1 projection candidate、M3-2 local replay、M3-3 临时 binding ledger、M3-4 loopback delivery、M3-5/M3-6 本地临时 provider identity 与 M3-7 pending profile/合成 attestation 均不计入生产退出门； -- M3-29/M3-30/M3-31/M3-32/M3-33 的下一证据必须 provision 专用 `chongqing-admission` environment/runner/reviewer/rotation policy，补齐 operator/tool/command、modified/additional manifest、archive-to-working-set attestation、owner/license/retention/access/privacy-sensitivity/standard-version/DataSLO/golden-result 八项签名决策与 fresh protected attestation，并由 M3-33 workflow 按 M3-32 intake contract 重新计算 admission eligibility；metadata-only admission/provenance/governance/readiness/evaluation/workflow evidence 是不可变检查结果，不能被编辑或重解释为内容批准、Landing authority 或 ingestion authorization； +- M3-34 的下一证据是将 public Landing ResourceVersion 绑定最小 DataOps definition/PlatformRun，完成一次轻量 profile 的解包、质量、GeoJSON/STAC 或 PostGIS serving、lineage、replay 和 rollback；M3-29/M3-30/M3-31/M3-32/M3-33 的重庆下一证据仍必须 provision 专用 `chongqing-admission` environment/runner/reviewer/rotation policy，补齐 operator/tool/command、modified/additional manifest、archive-to-working-set attestation、owner/license/retention/access/privacy-sensitivity/standard-version/DataSLO/golden-result 八项签名决策与 fresh protected attestation，并由 M3-33 workflow 按 M3-32 intake contract 重新计算 admission eligibility；metadata-only admission/provenance/governance/readiness/evaluation/workflow evidence 是不可变检查结果，不能被编辑或重解释为内容批准、Landing authority 或 ingestion authorization； - DolphinScheduler/Temporal sandbox 的独立数据库、备份恢复、身份、版本和升级责任证明；DolphinScheduler standalone/H2 不计入此退出门。