def report_deployment_evidence(
paths: Iterable[str | Path], *, release_version: str, deployment_id: str,
start: datetime, end: datetime, manifest: DeploymentManifest | None = None,
) -> dict[str, Any]:
"""Read only the supplied JSONL files within the half-open UTC window.
Invalid request labels/windows raise ValueError. File/record defects are
reported as issues while valid positive observations remain visible.
completeness is at most operator_attested; status never implies zero proof.
"""
if not _label(release_version) or not _label(deployment_id):
raise ValueError("explicit bounded release/deployment labels required")
start, end = _utc(start), _utc(end)
if start >= end:
raise ValueError("start must precede end")
if isinstance(paths, (str, Path)):
raise ValueError("paths must be an iterable of paths, not a scalar str or Path")
files = [Path(path) for path in paths]
from symfonic.platform.capture_boundary import (
capture_boundary_evidence,
native_journal_count_rows,
native_journal_file_bindings,
)
from symfonic.platform.deployment_observation_groups import observation_groups
from symfonic.platform.dispatch_capture import dispatch_capture_evidence, dispatch_count_issues
from symfonic.platform.fallback_capture import (
fallback_capture_bindings,
fallback_capture_status,
fallback_kind_issues,
)
from symfonic.platform.package_version import package_version_report
from symfonic.platform.route_attempt_capture import (
route_attempt_capture_evidence,
route_attempt_report,
)
capture_issues, failures, active_end, native_failures, native_active_end = (
capture_boundary_evidence(manifest, start=start, end=end)
)
binding_issues, native_bindings = native_journal_file_bindings(
manifest, {_resolve(path) for path in files}, _resolve,
)
dispatch_result = dispatch_capture_evidence(
manifest, {_resolve(path) for path in files}, _resolve, start=start, end=end,
)
fallback_issues, fallback_bound = fallback_capture_bindings(
manifest, {_resolve(path) for path in files}, _resolve,
)
attempt_capture = route_attempt_capture_evidence(
manifest, {_resolve(path) for path in files}, _resolve, start=start, end=end,
)
(
dispatch_issues, dispatch_failures, dispatch_active_end,
dispatch_bindings, expected_dispatch,
) = dispatch_result
fallback_issues += fallback_kind_issues(
fallback_bound, {*native_bindings, *dispatch_bindings, *attempt_capture[3]},
)
issues: list[dict[str, Any]] = [
{"code": code} for code in (
_manifest_issues(manifest, files, release_version, deployment_id, start, end)
+ capture_issues + binding_issues + dispatch_issues + fallback_issues
)
]
if not files:
issues.append({"code": "missing_files"})
if len({_resolve(p) for p in files}) != len(files):
raise ValueError("duplicate file paths")
fallback: Counter[tuple[str, str, str]] = Counter()
native: Counter[str] = Counter()
native_by_journal: Counter[tuple[str, str]] = Counter()
refusals: Counter[tuple[str, str, str, str]] = Counter()
dispatch: Counter[tuple[str, str, str, str, str]] = Counter()
dispatch_by_path: Counter[Path] = Counter()
attempts: Counter[tuple[str, str]] = Counter()
attempts_by_path: Counter[Path] = Counter()
versions: Counter[str | None] = Counter()
unknown_identity = unknown_fallback_attribution = excluded = 0
physical_files: set[tuple[int, int]] = set()
dispatch_binding_issues: set[Path] = set()
attempt_binding_issues: set[Path] = set()
fallback_strays: set[Path] = set()
fallback_foreign: set[Path] = set()
unreadable: set[Path] = set()
for path in files:
try:
with path.open("rb") as stream:
identity = os.fstat(stream.fileno())
physical = (identity.st_dev, identity.st_ino)
if physical in physical_files:
raise ValueError("duplicate physical capture file")
physical_files.add(physical)
binding_issue_recorded = False
for number, line in enumerate(stream, 1):
if not line.endswith(b"\n"):
issues.append({"code": "truncated_line", "path": str(path), "line": number})
continue
try:
row = json.loads(line)
except (ValueError, TypeError, UnicodeDecodeError, RecursionError):
issues.append({"code": "invalid_json", "path": str(path), "line": number})
continue
try:
kind, observed = _record(row)
except ValueError as exc:
issues.append({"code": str(exc), "path": str(path), "line": number})
continue
if (fallback_bound and kind != "kernel_fallback_attribution.v1"
and _resolve(path) in fallback_bound - fallback_foreign):
issues.append({"code": "fallback_journal_kind_mismatch",
"path": str(path), "line": number})
fallback_foreign.add(_resolve(path))
if (row["release_version"], row["deployment_id"]) != (
release_version, deployment_id,
) or not start <= observed < end:
excluded += 1
continue
versions[row.get("package_version")] += 1
if kind == "kernel_route_attempt.v1":
resolved = _resolve(path)
if (resolved not in attempt_capture[3]
and resolved not in attempt_binding_issues):
issues.append({"code": "route_attempt_journal_binding_mismatch",
"path": str(path), "line": number})
attempt_binding_issues.add(resolved)
attempts[(row["entrypoint"], row["capability"])] += 1
attempts_by_path[resolved] += 1
elif kind == "invocation_dispatch_outcome.v1":
resolved = _resolve(path)
if (resolved not in dispatch_bindings
and resolved not in dispatch_binding_issues):
issues.append({"code": "dispatch_journal_binding_mismatch",
"path": str(path), "line": number})
dispatch_binding_issues.add(resolved)
dispatch[(row["outcome"], row["entrypoint"], row["capability"],
row["reason_code"], row["selected_route"])] += 1
dispatch_by_path[resolved] += 1
elif kind in {"native_composition_refusal.v1", "legacy_invocation_refusal.v1"}:
refusals[(row["boundary"], row["surface"], row["reason_code"],
row["approval_reference"])] += 1
elif kind == "kernel_native_entry.v1":
owner = native_bindings.get(_resolve(path))
if (manifest is not None and not binding_issue_recorded
and (owner is None or owner[2] != row["entrypoint"])):
issues.append({"code": "native_journal_binding_mismatch",
"path": str(path), "line": number})
binding_issue_recorded = True
if owner is not None and owner[2] == row["entrypoint"]:
native_by_journal[owner[:2]] += 1
native[row["entrypoint"]] += 1
unknown_identity += not row["identity_known"]
else:
resolved = _resolve(path)
if (fallback_bound is not None and resolved not in fallback_bound
and resolved not in fallback_strays):
issues.append({"code": "fallback_journal_binding_mismatch",
"path": str(path), "line": number})
fallback_strays.add(resolved)
fallback[(row["entrypoint"], row["capability"], row["reason_code"])] += 1
unknown_fallback_attribution += any(
row[key] == "unknown"
for key in ("entrypoint", "capability", "reason_code")
)
except OSError:
issues.append({"code": "unreadable_file", "path": str(path)})
unreadable.add(_resolve(path))
total_native, total_fallback = sum(native.values()), sum(fallback.values())
total_dispatch = sum(dispatch.values())
attempt_report, attempt_issues = route_attempt_report(
attempts, attempts_by_path, attempt_capture,
declared=bool(manifest and manifest.get("expected_route_attempt_journals")),
)
issues.extend({"code": code} for code in attempt_issues)
count_issues, delivered = dispatch_count_issues(expected_dispatch, dispatch_by_path)
if manifest is not None and manifest.get("expected_dispatch_journals") and not total_dispatch:
count_issues.append("empty_dispatch_journal")
issues.extend({"code": code} for code in count_issues)
if not total_native and not total_fallback and not total_dispatch and not attempts:
issues.append({"code": "empty_observed_population"})
if unknown_identity:
issues.append({"code": "unknown_native_identity"})
if unknown_fallback_attribution:
issues.append({"code": "unknown_fallback_attribution"})
version_fields, version_issues = package_version_report(versions, manifest)
issues.extend({"code": code} for code in version_issues)
# Non-empty bindings imply _journal_inventory validated this manifest shape.
native_journal_counts = (
native_journal_count_rows(manifest, native_by_journal, native_bindings)
if manifest is not None and native_bindings and not issues else None
)
return {
"release_version": release_version, "deployment_id": deployment_id,
"window": {"start": start.isoformat(), "end": end.isoformat(), "bounds": "[start,end)"},
"status": ("positive_observations" if total_native or total_fallback or total_dispatch
or attempts
else "not_measurable"),
"completeness": "incomplete" if issues else "operator_attested",
**observation_groups(fallback, native, dispatch, refusals),
**attempt_report,
**version_fields,
"native_entry_journal_count_source": (
"selected_jsonl_rows" if native_journal_counts is not None else None
),
"native_entry_journal_counts": native_journal_counts,
"operator_exercised_population": (
manifest["exercised_population"] if manifest is not None
and type(manifest.get("exercised_population")) is int
and manifest["exercised_population"] > 0 else None
),
"unknown_native_identity_count": unknown_identity,
"unknown_fallback_attribution_count": unknown_fallback_attribution,
"capture_failure_count": failures,
"capture_in_flight_end": active_end,
"native_capture_failure_count": native_failures,
"native_capture_in_flight_end": native_active_end,
"dispatch_capture_failure_count": dispatch_failures,
"dispatch_capture_in_flight_end": dispatch_active_end,
"dispatch_capture_delivered_count": delivered,
"fallback_capture_status": fallback_capture_status(
fallback_bound, fallback_issues, stray_rows=bool(fallback_strays),
foreign_rows=bool(fallback_foreign),
failures=failures, active_end=active_end,
unreadable=bool(fallback_bound and unreadable & fallback_bound),
),
"dispatch_capture_status": (
"healthy" if dispatch_failures == 0 and dispatch_active_end == 0
and not dispatch_issues and not count_issues else "incomplete"
),
"excluded_record_count": excluded,
"authority": "positive_observations_only; operator_attestation_is_not_executable_proof",
"issues": issues,
}