"""Cooperative authoritative reader for DIRECT_STABLE_PATH_SET_V1.
|
|
The stable paths are intentionally readable by ordinary filesystem tools, but
|
only this double-read protocol may announce them as the current generation.
|
It never returns a partial payload: lock presence yields RETRY and any identity
|
or exact-set mismatch yields STOP_NOT_CURRENT.
|
"""
|
|
from __future__ import annotations
|
|
import os
|
from pathlib import Path
|
from typing import Any
|
|
from . import publisher as base
|
|
|
READER_SCHEMA = "SHARED_CONTENT_PUBLISHER_AUTHORITATIVE_READER_V1"
|
|
|
def _early_paths(config: dict[str, Any], config_path: Path) -> tuple[Path, Path, Path, Path]:
|
if config.get("schema_version") != base.DIRECT_CONFIG_SCHEMA:
|
raise base.PublisherError("READER_CONFIG_SCHEMA", str(config.get("schema_version")))
|
commit = config.get("commit")
|
roots = config.get("roots")
|
expected = config.get("expected")
|
if not isinstance(commit, dict) or commit.get("strategy") != base.DIRECT_COMMIT_STRATEGY:
|
raise base.PublisherError("READER_COMMIT_STRATEGY", str(commit))
|
if not isinstance(roots, dict) or not isinstance(expected, dict):
|
raise base.PublisherError("READER_CONFIG_TYPE", "roots/expected")
|
operation_root = Path(os.path.abspath(base._need_string(roots.get("operation_root"), "roots.operation_root")))
|
base._physical_chain(operation_root, final_kind="dir")
|
if base._norm(operation_root) != base._norm(base._need_string(roots.get("resolved_root"), "roots.resolved_root")):
|
raise base.PublisherError("READER_ROOT_RESOLUTION", str(operation_root))
|
if os.path.commonpath([base._norm(operation_root), base._norm(config_path)]) != base._norm(operation_root):
|
raise base.PublisherError("READER_CONFIG_OUTSIDE_ROOT", str(config_path))
|
lock_path = base._inside(operation_root, base._safe_relative(roots.get("lock_path"), "roots.lock_path"))
|
case_index = base._inside(operation_root, base._safe_relative(roots.get("case_current_index_path"), "roots.case_current_index_path"))
|
candidate = expected.get("candidate")
|
if not isinstance(candidate, dict):
|
raise base.PublisherError("READER_CONFIG_TYPE", "expected.candidate")
|
rows = base._parse_artifact_rows(candidate.get("rows"), "expected.candidate.rows")
|
manifest_rel = base._safe_relative(candidate.get("manifest_self_formal_relative_path"), "expected.candidate.manifest_self_formal_relative_path")
|
manifest_matches = [row for row in rows if row.formal_relative_path.casefold() == manifest_rel.casefold()]
|
if len(manifest_matches) != 1:
|
raise base.PublisherError("READER_MANIFEST_ROW", str(len(manifest_matches)))
|
current_manifest = base._inside(operation_root, manifest_matches[0].formal_relative_path)
|
return operation_root, lock_path, case_index, current_manifest
|
|
|
def _identity_or_stop(path: Path, label: str) -> tuple[int, str]:
|
try:
|
return base._identity(path)
|
except base.PublisherError:
|
raise
|
except Exception as exc:
|
raise base.PublisherError("READER_IDENTITY", f"{label}:{type(exc).__name__}:{exc}", exit_code=20) from exc
|
|
|
def read_direct_current(config_path: Path, *, include_payload: bool = False) -> dict[str, Any]:
|
"""Return authoritative identities, or a retry/stop status with no payload."""
|
|
config_path = Path(os.path.abspath(config_path))
|
config, config_data = base._load_config(config_path)
|
operation_root, lock_path, case_index, current_manifest = _early_paths(config, config_path)
|
common = {
|
"schema_version": READER_SCHEMA,
|
"commit_strategy": base.DIRECT_COMMIT_STRATEGY,
|
"config_bytes": len(config_data),
|
"config_sha256": base._sha_bytes(config_data),
|
"resolved_root": os.path.abspath(operation_root),
|
}
|
if base._exists(lock_path):
|
return {**common, "status": "RETRY_PUBLISH_IN_PROGRESS", "exit_code": 75, "payload_returned": False, "error_code": "PUBLISH_LOCK_PRESENT"}
|
try:
|
before_case = _identity_or_stop(case_index, "case_current_index")
|
before_manifest = _identity_or_stop(current_manifest, "current_manifest")
|
# Full strict validation is intentionally between the two identity reads.
|
# It validates every declared stable target, current-manifest coverage,
|
# link/evidence closure, exact scope sets, and the complete receipt chain.
|
from .direct_stable_path import (
|
_destination_exact_set,
|
_validate_direct_config,
|
_verify_candidate_destinations,
|
)
|
|
plan = _validate_direct_config(config, config_path, config_data)
|
if plan.lifecycle != "REPLAY":
|
raise base.PublisherError("READER_NOT_COMMITTED", plan.lifecycle, exit_code=20)
|
_verify_candidate_destinations(plan, "READER_TARGET_SET")
|
payloads: dict[str, bytes] = {}
|
for row in plan.candidate_rows:
|
data = base._read_bytes(plan.destination_paths[row.member_id])
|
if (len(data), base._sha_bytes(data)) != (row.bytes, row.sha256):
|
raise base.PublisherError("READER_PAYLOAD_IDENTITY", row.member_id, exit_code=20)
|
payloads[row.formal_relative_path] = data
|
after_case = _identity_or_stop(case_index, "case_current_index")
|
after_manifest = _identity_or_stop(current_manifest, "current_manifest")
|
if base._exists(lock_path):
|
return {**common, "status": "RETRY_PUBLISH_IN_PROGRESS", "exit_code": 75, "payload_returned": False, "error_code": "PUBLISH_LOCK_PRESENT_AFTER_READ"}
|
if before_case != after_case or before_manifest != after_manifest:
|
return {**common, "status": "STOP_NOT_CURRENT", "exit_code": 20, "payload_returned": False, "error_code": "CURRENT_IDENTITY_DRIFT"}
|
return {
|
**common,
|
"status": "CURRENT_READ_COMPLETE",
|
"exit_code": 0,
|
"payload_returned": include_payload,
|
"current_announced": True,
|
"case_current_index_bytes": after_case[0],
|
"case_current_index_sha256": after_case[1],
|
"current_manifest_bytes": after_manifest[0],
|
"current_manifest_sha256": after_manifest[1],
|
"destination_count": len(plan.candidate_rows),
|
"destination_set_sha256": _destination_exact_set(plan.candidate_rows),
|
**({"payloads": payloads} if include_payload else {}),
|
}
|
except base.PublisherError as exc:
|
if base._exists(lock_path):
|
return {**common, "status": "RETRY_PUBLISH_IN_PROGRESS", "exit_code": 75, "payload_returned": False, "error_code": "PUBLISH_LOCK_PRESENT_DURING_READ"}
|
return {**common, "status": "STOP_NOT_CURRENT", "exit_code": 20, "payload_returned": False, "error_code": exc.code, "detail": exc.detail}
|
|
|
def read_direct_current_snapshot(config_path: Path) -> dict[str, Any]:
|
"""Python API returning the full frozen byte snapshot only on success."""
|
|
return read_direct_current(config_path, include_payload=True)
|
