fix: fail on required output errors

README.md

@@ -120,6 +120,12 @@ Each run creates a timestamped subdirectory under the output directory containin
 | `{provider}_ip_ranges.json` | Validated provider ranges and provenance used by this run, from either a live source or a fresh cache |
 | `evidence/dns/` | dig or nslookup output per flagged domain (when `--evidence` is set) |
 
+Required output failures terminate the run instead of being logged and ignored.
+`allocator-targets-v1.json` is cleared when a run starts and published with an
+atomic replace only after processing succeeds, so a partial run cannot leave an
+older actionable document looking current. Non-actionable partial files may
+remain for diagnosis.
+
 ## AWS Lambda deployment
 
 DNSResolver can run as an S3-triggered Lambda. The full deployment walkthrough (ECR image, IAM, triggers) lives in [`docs/LAMBDA.md`](docs/LAMBDA.md). Note: the maintained Lambda packaging is produced in a separate project; the handler here is the reference entry point.

ROADMAP.md

@@ -31,9 +31,9 @@ then confirm GitHub jobs `test`, `gitleaks`, and `trufflehog` before squash merg
 | 4 | Fail closed on incomplete provider catalogues | Merged, PR 163 | AWS/GCP/Azure use bounded retries, validation, provenance, freshness-limited integrity-checked snapshots, and explicit states; no DNS processing occurs with an unusable provider. |
 | 5 | Publish the versioned allocator target document | Merged, PR 164 | A successful run atomically publishes schema-valid provider-aware JSON, groups repeated metadata, excludes wildcard observations, and retains the legacy pipe output. |
 | 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 | Implemented on `fix/preserve-cloud-attribution`; PR pending | 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. |
+| 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 | Make output publication atomic and observable | Deferred; not an initial-workability gate | 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. |
+| 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. |
 | 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. |
@@ -46,15 +46,16 @@ then confirm GitHub jobs `test`, `gitleaks`, and `trufflehog` before squash merg
   individual work-item PR.
 - No matcher-performance refactor inside attribution item 7; correctness is frozen
   first so optimisation has a trustworthy equivalence oracle.
-- No separate security-hardening phase. Items 8-10 are optional reliability and
-  performance follow-ups, deferred until measured operational need.
+- No standalone security-hardening phase is planned here. Items 9-10 are
+  operational reliability improvements; item 8 is a performance change that
+  remains deferred until measured need.
 - No claim that unit coverage proves resolver behavior; real DNS, real provider
   catalogues, consumer validation, and actual merge CI remain separate gates.
 
 ### Delivery state — 2026-08-14
 
-Item 7 is implemented on `fix/preserve-cloud-attribution` and ready for its
-focused pull request.
+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.
 
 - Focused suite: 91 tests passed.
 - Full local CI: Ruff and format passed; baseline 5 passed; full suite 310 passed
@@ -75,6 +76,11 @@ focused pull request.
 - The live gate may be delegated using `docs/EXTERNAL-ACCEPTANCE.md`; its report
   must keep automated, catalogue, system-resolver, public-resolver, and allocator
   evidence separate.
+- Item 9 focused suite: 55 tests passed. Full local CI: 315 tests passed at 95%
+  coverage for `classes` and `imports`; Ruff and formatting passed. A production
+  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.
 
 > **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/allocator_contract.py

@@ -5,6 +5,8 @@
 import os
 from pathlib import Path
 
+from classes.custom_exceptions import OutputWriteError
+
 CONTRACT_VERSION = "1.0"
 SUPPORTED_PROVIDERS = {"aws", "gcp", "azure"}
 WILDCARD_PREFIXES = ("WILDCARD|", "WILDCARD_ZONE|")
@@ -13,7 +15,7 @@
 def publish_allocator_targets(csp_path, output_dir):
     """Write allocator-targets-v1.json atomically and return its records."""
     destination = Path(output_dir) / "allocator-targets-v1.json"
-    destination.unlink(missing_ok=True)
+    clear_allocator_targets(output_dir)
     grouped = {}
     for line_number, line in enumerate(_lines(csp_path), start=1):
         if line.startswith(WILDCARD_PREFIXES):
@@ -56,12 +58,29 @@ def publish_allocator_targets(csp_path, output_dir):
             json.dump(targets, handle, indent=2)
             handle.write("\n")
         os.replace(temporary, destination)
+    except OSError as error:
+        raise OutputWriteError(
+            f"Unable to publish allocator targets to {destination}: {error}"
+        ) from error
     finally:
-        if temporary.exists():
-            temporary.unlink()
+        _remove_output(temporary)
     return targets
 
 
+def clear_allocator_targets(output_dir):
+    """Remove actionable output and its temporary file before a run starts."""
+    destination = Path(output_dir) / "allocator-targets-v1.json"
+    _remove_output(destination)
+    _remove_output(destination.with_suffix(".json.tmp"))
+
+
+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
+
+
 def _lines(path):
     try:
         with open(path, encoding="utf-8") as handle:

classes/custom_exceptions.py

@@ -12,3 +12,7 @@ class InvalidNameserversError(Exception):
 
 class ProviderCatalogueError(RuntimeError):
     """One or more required cloud provider catalogues are unusable."""
+
+
+class OutputWriteError(RuntimeError):
+    """A required run output could not be created or written."""

classes/output_manager.py

@@ -3,6 +3,8 @@
 
 import aiofiles
 
+from classes.custom_exceptions import OutputWriteError
+
 
 class OutputManager:
     """Creates the output directory structure and handles async file writes."""
@@ -44,12 +46,16 @@ def _create(self, path):
             else:
                 with open(path, "w", encoding="utf-8"):
                     pass
-        except (IOError, OSError) as e:
-            self._logger.error("Unable to create %s: %s", path, e)
+        except OSError as error:
+            message = f"Unable to create required output {path}: {error}"
+            self._logger.error(message)
+            raise OutputWriteError(message) from error
 
     async def write_to_file(self, file_path, content):
         try:
             async with aiofiles.open(file_path, "a", encoding="utf-8") as f:
                 await f.write(content + "\n")
-        except Exception as e:
-            self._logger.error("Failed to write to %s: %s", file_path, e)
+        except OSError as error:
+            message = f"Failed to write required output {file_path}: {error}"
+            self._logger.error(message)
+            raise OutputWriteError(message) from error

imports/cloud_service_provider_checks.py

@@ -1,6 +1,8 @@
 import functools
 import ipaddress
 
+from classes.custom_exceptions import OutputWriteError
+
 
 @functools.lru_cache(maxsize=None)
 def parse_network(cidr):
@@ -167,8 +169,13 @@ def log_and_write(
                 if message in written_lines:
                     continue
 
-                with open(file_path, "a", encoding="utf-8") as file:
-                    file.write(message + "\n")
+                try:
+                    with open(file_path, "a", encoding="utf-8") as file:
+                        file.write(message + "\n")
+                except OSError as error:
+                    raise OutputWriteError(
+                        f"Failed to write CSP attribution to {file_path}: {error}"
+                    ) from error
                 written_lines.add(message)
                 domain_context.log_info(message)
                 wrote_any = True

resolver.py

@@ -4,9 +4,12 @@
 
 from tqdm import tqdm
 
-from classes.allocator_contract import publish_allocator_targets
+from classes.allocator_contract import (
+    clear_allocator_targets,
+    publish_allocator_targets,
+)
 from classes.csp_ip_addresses import CSPIPAddresses
-from classes.custom_exceptions import ProviderCatalogueError
+from classes.custom_exceptions import OutputWriteError, ProviderCatalogueError
 from classes.dns_handler import DNSHandler
 from classes.environment_manager import EnvironmentManager
 from classes.run_summary import RunSummary
@@ -25,6 +28,7 @@ async def run(env_manager):
     both the CLI entrypoint (resolver.py) and the Lambda entrypoint (lambda_handler.py)
     can share the same logic.
     """
+    clear_allocator_targets(env_manager.output_dir)
     catalogues = {
         "gcp": fetch_google_cloud_ip_ranges(
             env_manager.output_dir, env_manager.extreme
@@ -102,6 +106,11 @@ async def bounded_process(domain, pbar, final_retry):
 
         failed = []
         for domain, result in zip(domains_to_process, results):
+            if isinstance(result, OutputWriteError):
+                env_manager.log_error(
+                    "Required output failure processing %s: %s", domain, result
+                )
+                raise result
             if isinstance(result, Exception):
                 env_manager.log_error(
                     "Unhandled exception processing %s: %s", domain, result

tests/test_allocator_publisher.py

@@ -1,10 +1,12 @@
 import json
 from pathlib import Path
+from unittest.mock import patch
 
 import pytest
 from jsonschema import Draft202012Validator, FormatChecker
 
 from classes.allocator_contract import publish_allocator_targets
+from classes.custom_exceptions import OutputWriteError
 
 REPO_ROOT = Path(__file__).resolve().parents[1]
 SCHEMA = json.loads(
@@ -93,3 +95,40 @@ def test_publisher_preserves_distinct_provider_regions(tmp_path):
         "ap-southeast-1",
         "us-east-1",
     }
+
+
+@pytest.mark.parametrize("failure", ["open", "write", "replace"])
+def test_publication_failure_removes_stale_and_temporary_files(tmp_path, failure):
+    matches = write_matches(
+        tmp_path,
+        [
+            "api.example.com|192.0.2.10|aws|ap-southeast-1|EC2|192.0.2.0/24|ap-southeast-1"
+        ],
+    )
+    destination = tmp_path / "allocator-targets-v1.json"
+    temporary = tmp_path / "allocator-targets-v1.json.tmp"
+    destination.write_text('[{"stale": true}]', encoding="utf-8")
+
+    if failure == "open":
+        real_open = open
+
+        def fail_output_open(path, mode="r", *args, **kwargs):
+            if Path(path) == temporary and "w" in mode:
+                raise OSError("disk full")
+            return real_open(path, mode, *args, **kwargs)
+
+        failure_patch = patch("builtins.open", side_effect=fail_output_open)
+    elif failure == "write":
+        failure_patch = patch(
+            "classes.allocator_contract.json.dump", side_effect=OSError("disk full")
+        )
+    else:
+        failure_patch = patch(
+            "classes.allocator_contract.os.replace", side_effect=OSError("read only")
+        )
+
+    with failure_patch, pytest.raises(OutputWriteError):
+        publish_allocator_targets(matches, tmp_path)
+
+    assert not destination.exists()
+    assert not temporary.exists()

tests/test_cloud_service_provider_checks.py

@@ -11,6 +11,7 @@
 import pytest
 
 from classes.allocator_contract import publish_allocator_targets
+from classes.custom_exceptions import OutputWriteError
 from classes.domain_processing_context import DomainProcessingContext
 from imports.cloud_service_provider_checks import (
     get_ip_matches,
@@ -292,6 +293,26 @@ def guarded_open(file, mode="r", *args, **kwargs):
     assert "example.com|34.1.2.3|gcp|" in out_file.read_text()
 
 
+def test_log_and_write_raises_explicit_output_error(tmp_path, ctx, monkeypatch):
+    out_file = tmp_path / "csp.txt"
+    output_files = {"standard": {"csp": str(out_file)}}
+
+    def fail_open(*args, **kwargs):
+        raise OSError("disk full")
+
+    monkeypatch.setattr("builtins.open", fail_open)
+
+    with pytest.raises(OutputWriteError, match="Failed to write CSP attribution"):
+        log_and_write(
+            "gcp",
+            {"34.1.2.3": {"34.0.0.0/8"}},
+            "example.com",
+            output_files,
+            ctx,
+            set(),
+        )
+
+
 def test_log_and_write_substring_line_not_suppressed(tmp_path, ctx):
     """A different line that happens to be a substring of an existing line
     must NOT be treated as a duplicate (the old file.read()-substring check

tests/test_output_manager.py

@@ -2,15 +2,15 @@
 Tests for OutputManager.
 
 Covers: output-dir creation, the evidence-dir-vs-file branch in _create,
-the write error path (aiofiles.open patched to raise), and evidence file
-layout when evidence=True.
+explicit create/write failures, and evidence file layout when evidence=True.
 """
 
 import os
 from unittest.mock import patch
 
 import pytest
 
+from classes.custom_exceptions import OutputWriteError
 from classes.output_manager import OutputManager
 
 
@@ -37,30 +37,33 @@ def test_no_evidence_key_when_evidence_disabled(tmp_path):
     assert "evidence" not in om.output_files
 
 
-def test_create_write_error_is_logged_not_raised(tmp_path):
+def test_create_write_error_is_raised(tmp_path):
     om = OutputManager(str(tmp_path), "20260101_000000")
-    with patch("builtins.open", side_effect=OSError("disk full")):
-        # Should not raise even though every _create call would fail internally.
-        om._create(str(tmp_path / "some_file.txt"))
-    # No assertion beyond "did not raise" is required by _create's contract,
-    # but confirm the logger recorded the failure.
+    target = str(tmp_path / "some_file.txt")
+    with (
+        patch("builtins.open", side_effect=OSError("disk full")),
+        pytest.raises(OutputWriteError, match="Unable to create"),
+    ):
+        om._create(target)
 
 
 def test_create_logs_error_on_failure(tmp_path):
     om = OutputManager(str(tmp_path), "20260101_000000")
     with patch("builtins.open", side_effect=OSError("disk full")):
         with patch.object(om._logger, "error") as mock_error:
-            om._create(str(tmp_path / "some_file.txt"))
+            with pytest.raises(OutputWriteError):
+                om._create(str(tmp_path / "some_file.txt"))
             mock_error.assert_called_once()
 
 
 @pytest.mark.asyncio
-async def test_write_to_file_error_logged_not_raised(tmp_path):
+async def test_write_to_file_error_is_raised(tmp_path):
     om = OutputManager(str(tmp_path), "20260101_000000")
     target = str(tmp_path / "out.txt")
     with patch("classes.output_manager.aiofiles.open", side_effect=OSError("boom")):
         with patch.object(om._logger, "error") as mock_error:
-            await om.write_to_file(target, "some content")
+            with pytest.raises(OutputWriteError, match="Failed to write"):
+                await om.write_to_file(target, "some content")
             mock_error.assert_called_once()