def verify_ingress_observers(
report: Mapping[str, Any], manifest: Mapping[str, Any], host_attempts: Mapping[str, Any],
) -> dict[str, Any]:
"""Return bound/incomplete/unaccounted_ingress/not_comparable, without I/O.
Label/window incompatibility takes precedence, then unaccounted ingress,
then incomplete evidence. All defects retain issue codes. ``bound`` means
equality only within the declared native-entry observation population.
Reader provenance is required; operator-supplied native_rows are not used.
Native files must be canonical absolute paths identical to reader-resolved
files. Relative paths and symlink aliases are conservatively rejected.
"""
issues: list[dict[str, str]] = []
counts: list[dict[str, Any]] = []
def issue(code: str, outcome: str = "incomplete") -> None:
issues.append({"code": code, "outcome": outcome})
def verify() -> None:
if not all(isinstance(value, Mapping) for value in (report, manifest, host_attempts)):
issue("malformed_input")
return
if set(host_attempts) != {"labels", "rows"}:
issue("malformed_host_attempts")
if manifest.get("ingress_inventory_attested") is not True:
issue("missing_ingress_inventory_attestation")
if (report.get("completeness") != "operator_attested"
or report.get("issues") != []):
issue("incomplete_report")
for field in ("release_version", "deployment_id"):
if not _label(report.get(field)) or report.get(field) != manifest.get(field):
issue("report_labels_mismatch", "not_comparable")
if not _labels(manifest.get("host_labels")) or not _labels(host_attempts.get("labels")):
issue("invalid_host_labels", "not_comparable")
elif manifest["host_labels"] != host_attempts["labels"]:
issue("host_labels_mismatch", "not_comparable")
window = report["window"]
start, end = _utc(window["start"]), _utc(window["end"])
if start >= end or window.get("bounds") != "[start,end)":
issue("invalid_window", "not_comparable")
return
if (start, end) != (_utc(manifest["start"]), _utc(manifest["end"])):
issue("report_window_mismatch", "not_comparable")
if any(time.minute or time.second or time.microsecond for time in (start, end)):
issue("non_hour_aligned_window", "not_comparable")
return
hours = (end - start) // timedelta(hours=1)
try:
bindings, observers = _inventory(manifest)
except (KeyError, TypeError, ValueError):
issue("invalid_surface_bindings")
return
capture_issues, *_ = capture_boundary_evidence(manifest, start=start, end=end)
if capture_issues:
issue("invalid_manifest_capture_boundaries")
expected_files = manifest.get("expected_files")
if (not _files(expected_files) or any(
name not in expected_files for spec in observers.values() for name in spec["files"]
)):
issue("native_file_inventory_mismatch")
if observers.keys() - set(bindings.values()):
issue("unclaimed_journal", "unaccounted_ingress")
rows = report.get("native_entry_journal_counts")
if (report.get("native_entry_journal_count_source") != "selected_jsonl_rows"
or not isinstance(rows, list)):
issue("missing_reader_native_counts")
return
native = {}
for row in rows:
if (not isinstance(row, Mapping)
or set(row) != {"replica", "journal_id", "entrypoint",
"declared_file_count", "resolved_files", "count"}
or not _label(row["replica"]) or not _label(row["journal_id"])
or not _count(row["count"]) or not _count(row["declared_file_count"])):
issue("malformed_native_count")
continue
observer = row["replica"], row["journal_id"]
if observer not in observers:
issue("undeclared_journal", "unaccounted_ingress")
continue
if observer in native:
issue("duplicate_native_count")
continue
spec = observers[observer]
if (row["entrypoint"] != spec["entrypoint"]
or row["declared_file_count"] != len(spec["files"])):
issue("native_observer_binding_mismatch")
if (not _files(row["resolved_files"])
or sorted(row["resolved_files"]) != sorted(spec["files"])):
issue("native_file_identity_mismatch")
native[observer] = row["count"]
if set(native) != set(observers):
issue("missing_journal_count")
if (not _count(report.get("native_entry_count"))
or sum(native.values()) != report["native_entry_count"]):
issue("native_count_total_mismatch")
for field in ("native_capture_failure_count", "native_capture_in_flight_end",
"unknown_native_identity_count"):
if type(report.get(field)) is not int or report[field] != 0:
issue("unknown_or_positive_observer_loss")
rows = host_attempts["rows"]
if not isinstance(rows, list):
issue("malformed_host_rows")
return
attempts: Counter = Counter()
seen = set()
covered: Counter = Counter()
for row in rows:
if (not isinstance(row, Mapping)
or set(row) != {"replica", "surface", "hour", "attempts", "lost_attempts"}
or not _label(row["replica"]) or not _label(row["surface"])
or not _count(row["attempts"])):
issue("malformed_host_row")
continue
if type(row["lost_attempts"]) is not int or row["lost_attempts"] != 0:
issue("unknown_or_positive_host_loss")
surface = row["replica"], row["surface"]
try:
hour = _utc(row["hour"])
except (ValueError, TypeError, OverflowError):
issue("invalid_host_hour", "not_comparable")
continue
if hour.minute or hour.second or hour.microsecond or not start <= hour < end:
issue("host_hour_outside_window", "not_comparable")
continue
if surface not in bindings:
issue("undeclared_surface", "unaccounted_ingress")
continue
key = *surface, hour
if key in seen:
issue("duplicate_surface_hour")
continue
seen.add(key)
covered[surface] += 1
attempts[bindings[surface]] += row["attempts"]
if any(covered[surface] != hours for surface in bindings):
issue("missing_surface_hour")
incomplete_observers = {
observer for surface, observer in bindings.items() if covered[surface] != hours
}
for observer in sorted(observers):
observed, admitted = native.get(observer), attempts[observer]
counts.append({"replica": observer[0], "journal_id": observer[1],
"attempts": admitted, "native_entries": observed})
if observer in incomplete_observers:
continue
if observed is None or observed < admitted:
issue("missing_native_entries")
elif observed > admitted:
issue("extra_native_entries", "unaccounted_ingress")
try:
verify()
except (KeyError, TypeError, ValueError, OverflowError, AttributeError):
issue("malformed_input")
outcomes = {item["outcome"] for item in issues}
outcome = next((value for value in ("not_comparable", "unaccounted_ingress", "incomplete")
if value in outcomes), "bound")
unique = sorted({(item["code"], item["outcome"]) for item in issues})
return {"outcome": outcome, "authority": AUTHORITY, "journals": counts,
"issues": [{"code": code, "outcome": status} for code, status in unique]}