feat: publish run observations and manifest

README.md

@@ -115,6 +115,8 @@ Each run creates a timestamped subdirectory under the output directory containin
 | `takeover_candidates_*.txt` | `DANGLING\|origin\|target\|category\|recommendation\|evidence\|hops\|chain` — the chain records the full CNAME path (`a -> b -> c`), so the claimable hop is visible without re-resolving. Plus `NS_TAKEOVER\|` lines for unresolvable nameservers |
 | `csp_matches_*.txt` | One handoff record per provider-published attribution: `domain\|ip\|provider\|region\|service\|prefix\|border_group`. One address can produce several records. Prefixed `WILDCARD\|` when the resolution was a catch-all. See [Cloud IP attribution](#cloud-ip-attribution) |
 | `allocator-targets-v1.json` | Versioned provider-aware allocator handoff. Groups service, prefix, and border-group metadata per provider/hostname/address/region and excludes wildcard observations; one address may produce multiple records when a provider publishes multiple regions. |
+| `dns-observations-v1.json` | Versioned excluded-observation feed for wildcard, dangling CNAME, NS, unresolved, and provider-catalogue-incomplete results. |
+| `run-manifest-v1.json` | Final completion marker recording run status, provider catalogue provenance, and the published contract documents. |
 | `environment_results_*.json` | Run metadata (command, external IP, Docker status) |
 | `provider_catalogues.json` | AWS, GCP, and Azure catalogue status, source, retrieval time, snapshot identifier, and any failure reason |
 | `{provider}_ip_ranges.json` | Validated provider ranges and provenance used by this run, from either a live source or a fresh cache |
@@ -145,6 +147,7 @@ lambda_handler.py        — Lambda entry point → run(env_manager)
 ├── DomainProcessingContext  — per-domain state (domain name, resolver, CSP IPs)
 ├── CSPIPAddresses           — value object holding fetched AWS/GCP/Azure IP ranges
 ├── DomainCategoriser        — regex-based classification of dangling CNAME targets
+├── run_contract.py          — excluded-observation and final run-manifest publisher
 └── domain_processor.py     — orchestrates DNS → CSP checks per domain
 ```
 
@@ -157,8 +160,8 @@ to the unresolved output only after both resolvers fail on the final configured
 
 ## Cloud IP attribution
 
-The current text files are the v2 CLI output. The versioned JSON handoff planned
-for provider-aware allocators is defined separately in the
+The current text files are the v2 CLI output. The published versioned JSON
+handoff for provider-aware allocators is defined separately in the
 [allocator contract](docs/ALLOCATOR-CONTRACT.md); its schemas, examples, and
 documentation field tables are validated together in CI.
 

ROADMAP.md

@@ -33,8 +33,8 @@ then confirm GitHub jobs `test`, `gitleaks`, and `trufflehog` before squash merg
 | 6 | Pin Lambda input to the triggering object version | Merged, PR 165 | The reference handler reads the exact S3 object version from the event and rejects incomplete version information. This does not reopen Lambda expansion. |
 | 7 | Preserve every provider attribution | Merged, PR 169 | Duplicate services, overlapping prefixes, identical CIDRs across providers, and multiple provider-published regions survive catalogue parsing, matching, pipe output, and JSON publication without changing the schema. Legacy scalar cache snapshots remain readable. |
 | 8 | Replace linear cloud-range matching with an indexed matcher | Deferred until measured need | A deterministic benchmark is defined before implementation; outputs are byte-for-byte equivalent to item 7; measured runtime and memory are reported at representative scale. |
-| 9 | Fail explicitly on required output errors | Implemented on `fix/fail-on-output-errors`; PR pending | Required writes cannot be swallowed; partial runs cannot leave a stale actionable document; injected open/write/replace failures produce a nonzero, explicit failure with regression tests. |
-| 10 | Complete observation/run-manifest publication | Deferred; not an initial-workability gate | The checked observation and manifest contracts are emitted by real runs; manifest state reflects provider completeness and publication outcome; actionable output is null for incomplete/failed runs. |
+| 9 | Fail explicitly on required output errors | Merged, PR 170 | Required writes cannot be swallowed; partial runs cannot leave a stale actionable document; injected open/write/replace failures produce a nonzero, explicit failure with regression tests. |
+| 10 | Publish observations and the run manifest | Implemented on `feat/publish-run-contracts`; PR pending | The checked observation and manifest contracts are emitted by real runs; manifest state reflects provider completeness and publication outcome; actionable output is null for incomplete/failed runs. |
 | 11 | Validate the real allocator consumer end to end | Planned, cross-repository | The AWS consumer ingests a current DNSResolver document unchanged; GCP/Azure route only to provider-aware implementations or are explicitly rejected; a synthetic authorized fixture proves no provider is misrouted. |
 | 12 | Measure and calibrate large-run behavior | Planned last | A repeatable representative benchmark replaces the unmeasured README scalability claim; resource limits and operational guidance reflect measured results. |
 
@@ -54,8 +54,8 @@ then confirm GitHub jobs `test`, `gitleaks`, and `trufflehog` before squash merg
 
 ### Delivery state — 2026-08-14
 
-Item 7 was squash-merged in PR 169. Item 9 is implemented on
-`fix/fail-on-output-errors` and ready for its focused pull request.
+Items 7 and 9 were squash-merged in PRs 169 and 170. Item 10 is implemented on
+`feat/publish-run-contracts` and ready for its focused pull request.
 
 - Focused suite: 91 tests passed.
 - Full local CI: Ruff and format passed; baseline 5 passed; full suite 310 passed
@@ -81,6 +81,13 @@ Item 7 was squash-merged in PR 169. Item 9 is implemented on
   system-resolver run fetched all three current catalogues, resolved both public
   inputs, emitted 45 attribution records, and published 20 schema-valid targets
   without leaving a temporary allocator document.
+- Item 10 focused suite: 32 tests passed. Full local CI: 320 tests passed at 95%
+  coverage for `classes` and `imports`. A production system-resolver run published
+  20 actionable targets, an empty but schema-valid observation document, and a
+  schema-valid complete manifest referencing both documents; no temporary run
+  document remained. A reserved `example.com` negative control separately
+  published one schema-valid unresolved observation, an empty actionable array,
+  and a complete manifest.
 
 > **On its derivation.** The plan below came from [`REVIEW.md`](REVIEW.md) (2026-07-15), which
 > assessed the tool against a misread goal — cloud attribution treated as a supporting attribute

classes/run_contract.py

@@ -0,0 +1,191 @@
+"""Publish non-actionable DNS observations and the run completion manifest."""
+
+import json
+import os
+from pathlib import Path
+
+from classes.custom_exceptions import OutputWriteError
+
+CONTRACT_VERSION = "1.0"
+OBSERVATIONS_FILENAME = "dns-observations-v1.json"
+MANIFEST_FILENAME = "run-manifest-v1.json"
+TARGETS_FILENAME = "allocator-targets-v1.json"
+
+
+def clear_run_documents(output_dir):
+    """Remove stale run documents and their temporary files."""
+    for filename in (OBSERVATIONS_FILENAME, MANIFEST_FILENAME):
+        destination = Path(output_dir) / filename
+        _remove_output(destination)
+        _remove_output(destination.with_suffix(destination.suffix + ".tmp"))
+
+
+def publish_dns_observations(
+    output_files,
+    output_dir,
+    provider_failures=None,
+    hostnames=None,
+):
+    """Build the excluded-observation contract from the run's text outputs."""
+    grouped = {}
+    standard = output_files.get("standard", {})
+
+    for line in _lines(standard.get("resolved")):
+        if line.startswith("WILDCARD|"):
+            _add_observation(
+                grouped,
+                line,
+                "WILDCARD",
+                "wildcard",
+                "DNS answer matched the zone wildcard probe",
+            )
+        elif line.startswith("WILDCARD_ZONE|"):
+            _add_observation(
+                grouped,
+                line,
+                "WILDCARD_ZONE",
+                "wildcard_zone",
+                "Parent zone answered random-label probes",
+            )
+
+    for line in _lines(standard.get("takeover")):
+        fields = line.split("|")
+        if line.startswith("DANGLING|") and len(fields) >= 3:
+            _merge_observation(
+                grouped,
+                fields[1],
+                "dangling_cname",
+                [],
+                [f"CNAME target did not resolve: {fields[2]}"],
+            )
+        elif line.startswith("NS_TAKEOVER|") and len(fields) >= 3:
+            _merge_observation(
+                grouped,
+                fields[1],
+                "ns_takeover",
+                [],
+                [f"Nameserver did not resolve: {fields[2]}"],
+            )
+        else:
+            raise ValueError("Malformed takeover observation output")
+
+    unresolved_prefix = "DNS resolution error for "
+    for line in _lines(standard.get("unresolved")):
+        if not line.startswith(unresolved_prefix):
+            raise ValueError("Malformed unresolved observation output")
+        hostname, separator, reason = line[len(unresolved_prefix) :].partition(": ")
+        if not separator or not hostname or not reason:
+            raise ValueError("Malformed unresolved observation output")
+        _merge_observation(grouped, hostname, "unresolved", [], [reason])
+
+    if provider_failures:
+        reasons = [
+            f"{provider} provider catalogue unusable: {reason}"
+            for provider, reason in sorted(provider_failures.items())
+        ]
+        for hostname in sorted(set(hostnames or [])):
+            _merge_observation(
+                grouped,
+                hostname,
+                "provider_catalog_incomplete",
+                [],
+                reasons,
+            )
+
+    observations = []
+    for (hostname, kind), entry in sorted(grouped.items()):
+        observations.append(
+            {
+                "contract_version": CONTRACT_VERSION,
+                "hostname": hostname,
+                "kind": kind,
+                "ips": sorted(entry["ips"]),
+                "reasons": sorted(entry["reasons"]),
+                "actionability": "excluded",
+            }
+        )
+
+    _write_json_atomic(observations, Path(output_dir) / OBSERVATIONS_FILENAME)
+    return observations
+
+
+def publish_run_manifest(
+    output_dir,
+    run_id,
+    status,
+    started_at,
+    completed_at,
+    catalogues,
+):
+    """Publish the final run state after its referenced outputs are settled."""
+    if status not in {"complete", "incomplete", "failed"}:
+        raise ValueError(f"Unknown run status: {status}")
+    if set(catalogues) != {"aws", "gcp", "azure"}:
+        raise ValueError("Run manifest requires AWS, GCP, and Azure catalogues")
+
+    manifest = {
+        "contract_version": CONTRACT_VERSION,
+        "run_id": run_id,
+        "status": status,
+        "started_at": started_at,
+        "completed_at": completed_at,
+        "provider_catalogs": {
+            provider: catalogues[provider].manifest_entry()
+            for provider in ("aws", "gcp", "azure")
+        },
+        "outputs": {
+            "actionable_targets": TARGETS_FILENAME if status == "complete" else None,
+            "observations": OBSERVATIONS_FILENAME,
+        },
+    }
+    _write_json_atomic(manifest, Path(output_dir) / MANIFEST_FILENAME)
+    return manifest
+
+
+def _add_observation(grouped, line, marker, kind, reason):
+    fields = line.split("|")
+    if len(fields) < 3 or fields[0] != marker or not fields[1]:
+        raise ValueError(f"Malformed {kind} observation output")
+    _merge_observation(grouped, fields[1], kind, fields[2:], [reason])
+
+
+def _merge_observation(grouped, hostname, kind, ips, reasons):
+    entry = grouped.setdefault((hostname, kind), {"ips": set(), "reasons": set()})
+    entry["ips"].update(ip for ip in ips if ip)
+    entry["reasons"].update(reason for reason in reasons if reason)
+
+
+def _lines(path):
+    if not path:
+        return []
+    try:
+        with open(path, encoding="utf-8") as handle:
+            return [line.strip() for line in handle if line.strip()]
+    except OSError as error:
+        raise ValueError(
+            f"Unable to read observation output {path}: {error}"
+        ) from error
+
+
+def _write_json_atomic(document, destination):
+    temporary = destination.with_suffix(destination.suffix + ".tmp")
+    _remove_output(destination)
+    _remove_output(temporary)
+    try:
+        with open(temporary, "w", encoding="utf-8") as handle:
+            json.dump(document, handle, indent=2)
+            handle.write("\n")
+        os.replace(temporary, destination)
+    except OSError as error:
+        raise OutputWriteError(
+            f"Unable to publish run document {destination}: {error}"
+        ) from error
+    finally:
+        _remove_output(temporary)
+
+
+def _remove_output(path):
+    try:
+        path.unlink(missing_ok=True)
+    except OSError as error:
+        raise OutputWriteError(f"Unable to remove output {path}: {error}") from error

docs/ALLOCATOR-CONTRACT.md

@@ -28,6 +28,12 @@ provider-published region, and excludes
 unusable, or a CSP record is malformed, no allocator-target document is
 published.
 
+Every run also publishes `dns-observations-v1.json` and writes
+`run-manifest-v1.json` last as its completion marker. A complete manifest points
+to the actionable document. Incomplete or failed manifests set that path to
+`null`; a failed-state manifest is best effort when the underlying failure also
+prevents output writes.
+
 ## Documents
 
 | Document | Schema | Purpose |

resolver.py

@@ -1,6 +1,8 @@
 import asyncio
 import json
+from datetime import datetime, timezone
 from pathlib import Path
+from uuid import uuid4
 
 from tqdm import tqdm
 
@@ -12,6 +14,11 @@
 from classes.custom_exceptions import OutputWriteError, ProviderCatalogueError
 from classes.dns_handler import DNSHandler
 from classes.environment_manager import EnvironmentManager
+from classes.run_contract import (
+    clear_run_documents,
+    publish_dns_observations,
+    publish_run_manifest,
+)
 from classes.run_summary import RunSummary
 from imports.cloud_ip_ranges import (
     fetch_aws_ip_ranges,
@@ -28,14 +35,47 @@ async def run(env_manager):
     both the CLI entrypoint (resolver.py) and the Lambda entrypoint (lambda_handler.py)
     can share the same logic.
     """
+    run_id = str(uuid4())
+    started_at = _timestamp()
+    catalogues = {}
     clear_allocator_targets(env_manager.output_dir)
-    catalogues = {
+    clear_run_documents(env_manager.output_dir)
+    try:
+        await _run(env_manager, run_id, started_at, catalogues)
+    except ProviderCatalogueError:
+        raise
+    except Exception:
+        try:
+            clear_allocator_targets(env_manager.output_dir)
+            publish_dns_observations(
+                env_manager.output_files,
+                env_manager.output_dir,
+            )
+            if set(catalogues) == {"aws", "gcp", "azure"}:
+                publish_run_manifest(
+                    env_manager.output_dir,
+                    run_id=run_id,
+                    status="failed",
+                    started_at=started_at,
+                    completed_at=_timestamp(),
+                    catalogues=catalogues,
+                )
+        except Exception as contract_error:
+            env_manager.log_error(
+                "Unable to publish failed run state: %s", contract_error
+            )
+        raise
+
+
+async def _run(env_manager, run_id, started_at, catalogues):
+    fetched_catalogues = {
         "gcp": fetch_google_cloud_ip_ranges(
             env_manager.output_dir, env_manager.extreme
         ),
         "aws": fetch_aws_ip_ranges(env_manager.output_dir, env_manager.extreme),
         "azure": fetch_azure_ip_ranges(env_manager.output_dir, env_manager.extreme),
     }
+    catalogues.update(fetched_catalogues)
     status_path = Path(env_manager.output_dir) / "provider_catalogues.json"
     with open(status_path, "w", encoding="utf-8") as handle:
         json.dump(
@@ -49,10 +89,28 @@ async def run(env_manager):
 
     unusable = [name for name, catalogue in catalogues.items() if not catalogue.usable]
     if unusable:
+        env_manager.set_domains()
         details = "; ".join(
             f"{name}: {catalogues[name].error or catalogues[name].status}"
             for name in unusable
         )
+        publish_dns_observations(
+            env_manager.output_files,
+            env_manager.output_dir,
+            provider_failures={
+                name: catalogues[name].error or catalogues[name].status
+                for name in unusable
+            },
+            hostnames=env_manager.domains,
+        )
+        publish_run_manifest(
+            env_manager.output_dir,
+            run_id=run_id,
+            status="incomplete",
+            started_at=started_at,
+            completed_at=_timestamp(),
+            catalogues=catalogues,
+        )
         raise ProviderCatalogueError(
             "Required provider catalogue(s) unavailable; no domains were processed "
             f"and no actionable results were published. {details}. Status: {status_path}"
@@ -143,6 +201,7 @@ async def bounded_process(domain, pbar, final_retry):
         "Published %d actionable target(s) to allocator-targets-v1.json",
         len(targets),
     )
+    publish_dns_observations(env_manager.output_files, env_manager.output_dir)
 
     RunSummary(env_manager.output_files, env_manager.output_dir, __version__).display(
         len(env_manager.domains), len(domains_to_process)
@@ -160,6 +219,19 @@ async def bounded_process(domain, pbar, final_retry):
         env_manager.log_info("Azure IPv4 Ranges: %s", csp_ip_addresses.get_azure_ipv4())
         env_manager.log_info("Azure IPv6 Ranges: %s", csp_ip_addresses.get_azure_ipv6())
 
+    publish_run_manifest(
+        env_manager.output_dir,
+        run_id=run_id,
+        status="complete",
+        started_at=started_at,
+        completed_at=_timestamp(),
+        catalogues=catalogues,
+    )
+
+
+def _timestamp():
+    return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
+
 
 async def main_async():
     print(f"DNSResolver v{__version__}")