Add append-only package writer with atomic validated publication

README.md

@@ -1,6 +1,6 @@
 # VianoBugAnalysis
 
-Project-independent foundations for reading and validating long-lived ZIP case
+Project-independent foundations for reading, validating and extending ZIP case
 packages. The authoritative design is [docs/architecture.md](docs/architecture.md).
 The implemented v1 contract is documented in
 [docs/package-format-v1.md](docs/package-format-v1.md).
@@ -13,5 +13,5 @@ python -m venv .venv
 .\.venv\Scripts\python.exe -m pytest
 ```
 
-This milestone includes no package writer, lifecycle worker, CaseBuilder, LLM,
+This milestone includes no lifecycle worker, CaseBuilder, LLM,
 case interpretation, hypothesis generation or investigation tools.

docs/package-format-v1.md

@@ -9,8 +9,8 @@ Exactly one root `manifest.json` is the authoritative entry point.
 
 `format` is `viano-bug-case-package`; `format_version` is integer `1`.
 `package_id` is an opaque nonblank string identifying the case package throughout
-its lifetime. Enriching and emitting it again preserves this identity and, in a
-future writer, increases `package_generation` (a nonnegative integer).
+its lifetime. Enriching and emitting it again preserves this identity and
+increases `package_generation` (a nonnegative integer).
 `case_id` identifies the case. `analysis_revision` is a separate nonnegative
 integer, initially zero; it is independent of package generation.
 The reader does not derive or increment any of these values.
@@ -63,9 +63,9 @@ non-UTF-8 manifests and nonstandard numeric constants are rejected.
 feedback and revisions are appended. Only explicitly designated index/pointer
 files (`manifest.json`, `analysis/current.json`) may later be updated. Unknown
 files are preserved unchanged. Validator and reader never write, extract, delete
-or rewrite anything. Historical append-only and output-superset comparisons,
-atomic generation and generation increments belong to a future writer; they
-cannot be proven from one standalone incoming archive.
+or rewrite anything. The writer checks append-only and output-superset preservation
+against its input. Historical preservation cannot be proven from one standalone
+incoming archive.
 
 ## ZIP acceptance and configurable security policy
 
@@ -123,3 +123,82 @@ Read access outside the context is a programming error (`RuntimeError`). Invalid
 packages and missing member requests raise `PackageReadError` with structured
 details. `read_bytes` is bounded by policy but materializes a single file in memory;
 validation itself streams. No artifact interpretation is performed.
+
+## Writer API and publication contract
+
+```python
+from viano_bug_analysis.package import PackageWriter, PackageWriteError
+
+try:
+    result = PackageWriter().write(
+        "input.any-name",
+        "next-generation.any-name",
+        additions={"case/revisions/0001/case.json": b'{"example": true}'},
+        index_updates={"analysis/current.json": b'{"revision": 1}'},
+        manifest_updates={
+            "state": "VALIDATED",
+            "routing": {"destination": "cases"},
+            "artifacts": {"case": "case/revisions/0001/case.json"},
+        },
+    )
+except PackageWriteError as error:
+    issues = error.result.issues
+```
+
+`additions` accepts new file paths and bytes only. Existing files cannot be
+overwritten, even with identical bytes. Unknown input files and explicit directory
+entries are carried forward. ZIP comments and retained entry metadata are copied;
+compressed representations may change, but retained file bytes must be identical.
+Existing `analysis/revisions/<revision>/` namespaces are frozen, including empty
+directories: this conservative foundation does not infer whether a revision is
+still active. New revisions use new namespaces. New reproductions and feedback
+use new paths. There is no deletion operation.
+
+`index_updates` can create or replace only `analysis/current.json`. `manifest.json`
+is managed exclusively by the writer. Index bytes are opaque at this milestone;
+no interpretation or current-index schema is implemented.
+
+`manifest_updates` accepts only `state`, `routing`, `analysis_revision`, `artifacts`
+and `history`. Object updates merge recursively, preserving nested extension
+fields. Identity, format, project and creation time remain unchanged. The writer
+sets generation to input generation plus one, sets updated time to the later of
+the current UTC time and the previous updated time, and maintains SHA-256/size
+records for every output file except the manifest. Existing integrity-record
+extensions are retained. Analysis revision and standard history counters cannot
+decrease; they are otherwise supplied explicitly, never inferred from generation
+or artifact names. No lifecycle transition policy or automatic history event is
+introduced. Existing `history/events.jsonl` cannot be replaced: a future event
+writer needs an explicit byte-prefix append contract; for now use a new event
+artifact and update `events_ref` if needed.
+
+The source is fully validated before staging. Output uses ZIP_STORED so highly
+compressible new data does not create archives rejected by compression-ratio
+defaults. The same configurable `PackageLimits` apply to input and output. This
+may require a larger archive-byte policy for a previously compressed input.
+Copying and preservation comparison stream in bounded chunks; additions are
+caller-provided bytes. The source must remain unchanged throughout the call.
+
+Staging occurs in an exclusively created temporary file in the destination's
+existing parent directory. The ZIP is closed, flushed and fsynced, then fully
+validated. Its manifest must equal the planned manifest; every input entry must
+survive, and every protected file is compared byte for byte. The target is only
+published after these checks. Normal failures remove staging files and leave
+the input and existing target unchanged. A process crash may leave an unpublished
+staging file; this milestone does not implement recovery jobs.
+An exceptional cleanup failure before publication raises a structured error;
+after publication it emits a ResourceWarning and still returns success, since
+the valid target has already been committed. A staging file may remain in that
+case and can be removed by the caller.
+
+By default an existing target is rejected. Atomic no-clobber publication uses a
+same-directory hard link, preventing concurrent writers from silently replacing
+each other; the destination filesystem must support hard links. `overwrite=True`
+explicitly permits atomic replacement using `os.replace`. Input aliases (including
+hard links), symbolic-link targets and in-place output are rejected. Temporary
+files are cleaned up after publication. This is atomic visibility with a flushed
+file, not a claim of portable directory-metadata durability across power loss.
+Destination directories and input files must be controlled by the caller; this
+milestone does not implement case locking or protection from concurrent external
+filesystem mutation. Every successful write, including one without additions,
+produces exactly one new generation. Job-level retry idempotence is deferred to
+the later lifecycle layer.

pyproject.toml

@@ -5,7 +5,7 @@ build-backend = "setuptools.build_meta"
 [project]
 name = "viano-bug-analysis"
 version = "0.1.0"
-description = "Project-independent case package validation and reading"
+description = "Project-independent case package validation, reading and writing"
 requires-python = ">=3.11"
 dependencies = ["jsonschema>=4.18,<5"]
 

src/viano_bug_analysis/package/__init__.py

@@ -1,7 +1,8 @@
-from .errors import PackageReadError, ValidationIssue, ValidationResult
+from .errors import PackageReadError, PackageWriteError, ValidationIssue, ValidationResult
 from .limits import PackageLimits
 from .reader import PackageReader
 from .validator import PackageValidator
+from .writer import PackageWriter
 
 __all__ = ["PackageReader", "PackageValidator", "PackageLimits", "PackageReadError",
-           "ValidationIssue", "ValidationResult"]
+           "ValidationIssue", "ValidationResult", "PackageWriter", "PackageWriteError"]

src/viano_bug_analysis/package/errors.py

@@ -24,3 +24,11 @@ class PackageReadError(Exception):
     def __init__(self, result: ValidationResult):
         self.result = result
         super().__init__("; ".join(f"{i.code}: {i.message}" for i in result.issues))
+
+
+class PackageWriteError(Exception):
+    """Controlled write rejection, with the same issue contract as validation."""
+
+    def __init__(self, result: ValidationResult):
+        self.result = result
+        super().__init__("; ".join(f"{i.code}: {i.message}" for i in result.issues))

src/viano_bug_analysis/package/writer.py

@@ -0,0 +1,203 @@
+import copy
+import hashlib
+import json
+import os
+import tempfile
+import warnings
+import zipfile
+from collections.abc import Mapping
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any
+
+from .errors import PackageReadError, PackageWriteError, ValidationResult
+from .limits import PackageLimits
+from .manifest import manifest_issues, safe_path
+from .reader import PackageReader
+from .validator import ARCHIVE_ERRORS, CHUNK_SIZE, PackageValidator
+
+MUTABLE_INDEX = "analysis/current.json"
+UPDATE_FIELDS = frozenset({"state", "routing", "analysis_revision", "artifacts", "history"})
+HISTORY_COUNTERS = ("analysis_revisions", "reproductions", "evidence", "handoffs", "feedback")
+
+
+def _merge(base: dict, updates: Mapping) -> None:
+    """Merge objects without dropping unrecognized nested extension fields."""
+    for key, value in updates.items():
+        if isinstance(value, Mapping) and isinstance(base.get(key), dict):
+            _merge(base[key], value)
+        else:
+            base[key] = copy.deepcopy(value)
+
+
+class PackageWriter:
+    """Append artifacts, verify preservation, then atomically publish a generation."""
+
+    def __init__(self, limits: PackageLimits | None = None):
+        self.validator = PackageValidator(limits)
+
+    def _reject(self, code: str, message: str, path: str | None = None):
+        raise PackageWriteError(self.validator._failure(code, message, path))
+
+    def write(
+        self,
+        source: str | os.PathLike[str],
+        destination: str | os.PathLike[str],
+        *,
+        additions: Mapping[str, bytes] | None = None,
+        index_updates: Mapping[str, bytes] | None = None,
+        manifest_updates: Mapping[str, Any] | None = None,
+        overwrite: bool = False,
+    ) -> ValidationResult:
+        """Return validated output metadata. Input is never replaced, even by alias."""
+        temporary = None
+        published = False
+        source_path, target = Path(source), Path(destination)
+        additions = dict(additions or {})
+        index_updates = dict(index_updates or {})
+        updates = dict(manifest_updates or {})
+        try:
+            self._check_target(source_path, target, overwrite)
+            with PackageReader(source_path, self.validator.limits) as reader:
+                if updates.keys() - UPDATE_FIELDS:
+                    self._reject("MANIFEST_UPDATE_FORBIDDEN", "Only lifecycle, routing, revision, artifact and history updates are allowed")
+                self._check_changes(reader, additions, index_updates)
+                manifest = reader.manifest
+                old_manifest = reader.manifest
+                _merge(manifest, updates)
+                manifest["package_generation"] = int(old_manifest["package_generation"]) + 1
+                previous_time = datetime.fromisoformat(old_manifest["updated_at"].upper().replace("Z", "+00:00"))
+                manifest["updated_at"] = max(datetime.now(timezone.utc), previous_time).isoformat()
+                issues = manifest_issues(manifest)
+                if issues:
+                    raise PackageWriteError(ValidationResult(tuple(issues)))
+                if manifest["analysis_revision"] < old_manifest["analysis_revision"]:
+                    self._reject("REVISION_REGRESSION", "Analysis revision cannot decrease")
+                for key in HISTORY_COUNTERS:
+                    if manifest["history"][key] < old_manifest["history"][key]:
+                        self._reject("HISTORY_REGRESSION", "History counters cannot decrease", key)
+
+                changes = additions | index_updates
+                for name, data in changes.items():
+                    record = manifest["integrity"].setdefault(name, {})
+                    record.update(sha256=hashlib.sha256(data).hexdigest(), size=len(data))
+
+                # Hash all retained files, including previously unindexed unknown data.
+                for name in reader.files:
+                    if name == "manifest.json" or name in changes:
+                        continue
+                    digest = hashlib.sha256()
+                    size = 0
+                    with reader._archive.open(name) as stream:
+                        while chunk := stream.read(CHUNK_SIZE):
+                            size += len(chunk)
+                            if size > self.validator.limits.max_file_bytes:
+                                self._reject("FILE_LIMIT", "Input changed or exceeds configured limit", name)
+                            digest.update(chunk)
+                    record = manifest["integrity"].setdefault(name, {})
+                    record.update(sha256=digest.hexdigest(), size=size)
+                manifest_bytes = json.dumps(manifest, ensure_ascii=False, allow_nan=False, indent=2).encode("utf-8")
+                if len(manifest_bytes) > min(self.validator.limits.max_manifest_bytes, self.validator.limits.max_file_bytes):
+                    self._reject("FILE_LIMIT", "Output manifest exceeds configured limit", "manifest.json")
+                total = len(manifest_bytes) + sum(len(data) for data in changes.values())
+                total += sum(info.file_size for info in reader._archive.infolist() if info.filename != "manifest.json" and info.filename not in changes)
+                if total > self.validator.limits.max_total_bytes:
+                    self._reject("TOTAL_LIMIT", "Output exceeds configured expanded size")
+                if total > self.validator.limits.max_archive_bytes:
+                    self._reject("ARCHIVE_LIMIT", "Stored output exceeds configured archive size")
+                if len(reader._archive.infolist()) + len(additions) + sum(name not in reader.files for name in index_updates) > self.validator.limits.max_entries:
+                    self._reject("ENTRY_LIMIT", "Output has too many entries")
+
+                with tempfile.NamedTemporaryFile(prefix=".vbcase-", suffix=".tmp", dir=target.parent, delete=False) as staging:
+                    temporary = Path(staging.name)
+                    with zipfile.ZipFile(staging, "w", compression=zipfile.ZIP_STORED) as output:
+                        output.comment = reader._archive.comment
+                        for info in reader._archive.infolist():
+                            if info.filename == "manifest.json" or info.filename in changes:
+                                continue
+                            retained = copy.copy(info)
+                            retained.compress_type = zipfile.ZIP_STORED
+                            with reader._archive.open(info) as original, output.open(retained, "w", force_zip64=True) as copied:
+                                size = 0
+                                while chunk := original.read(CHUNK_SIZE):
+                                    size += len(chunk)
+                                    if size > self.validator.limits.max_file_bytes:
+                                        self._reject("FILE_LIMIT", "Input changed or exceeds configured limit", info.filename)
+                                    copied.write(chunk)
+                        for name, data in changes.items():
+                            output.writestr(name, data)
+                        output.writestr("manifest.json", manifest_bytes)
+                    staging.flush()
+                    os.fsync(staging.fileno())
+                result = self.validator.validate(temporary)
+                if not result.valid:
+                    raise PackageWriteError(result)
+                if result.manifest != manifest:
+                    self._reject("MANIFEST_CHANGED", "Output manifest differs from planned generation")
+                self._verify_preservation(reader, temporary, set(index_updates))
+                self._check_target(source_path, target, overwrite)
+                if overwrite:
+                    os.replace(temporary, target)
+                else:
+                    # Atomic no-clobber publication, including competing writers.
+                    os.link(temporary, target)
+                published = True
+                return result
+        except PackageReadError as error:
+            raise PackageWriteError(error.result) from error
+        except ARCHIVE_ERRORS + (UnicodeError, TypeError, RecursionError) as error:
+            raise PackageWriteError(self.validator._failure("WRITE_ERROR", str(error))) from error
+        finally:
+            if temporary is not None:
+                try:
+                    temporary.unlink(missing_ok=True)
+                except OSError as error:
+                    if published:
+                        warnings.warn(f"Package published, but staging cleanup failed: {error}", ResourceWarning)
+                    else:
+                        raise PackageWriteError(self.validator._failure("CLEANUP_ERROR", str(error))) from error
+
+    def _check_target(self, source: Path, target: Path, overwrite: bool) -> None:
+        if source.resolve() == target.resolve() or (target.exists() and source.exists() and os.path.samefile(source, target)):
+            self._reject("INPUT_TARGET_ALIAS", "Output must not replace input")
+        if target.is_symlink():
+            self._reject("TARGET_SYMLINK", "Output cannot be a symbolic link")
+        if target.exists() and not overwrite:
+            self._reject("TARGET_EXISTS", "Output already exists; explicitly opt into replacement")
+
+    def _check_changes(self, reader: PackageReader, additions: dict, indexes: dict) -> None:
+        if additions.keys() & indexes.keys():
+            self._reject("ENTRY_COLLISION", "Path supplied twice")
+        revisions = {info.filename.split("/")[2] for info in reader._archive.infolist()
+                     if info.filename.startswith("analysis/revisions/") and len(info.filename.split("/")) > 3}
+        for name, data in (additions | indexes).items():
+            if not safe_path(name) or name == "manifest.json":
+                self._reject("UNSAFE_PATH", "Unsafe or reserved artifact path", name)
+            if not isinstance(data, bytes):
+                self._reject("ARTIFACT_TYPE", "Artifacts must be bytes", name)
+            if len(data) > self.validator.limits.max_file_bytes:
+                self._reject("FILE_LIMIT", "Artifact exceeds configured limit", name)
+            if name in indexes and name != MUTABLE_INDEX:
+                self._reject("INDEX_UPDATE_FORBIDDEN", "Only analysis/current.json can be updated", name)
+            if name in additions and (name in reader.files or name == MUTABLE_INDEX):
+                self._reject("ARTIFACT_EXISTS", "Use a new artifact path; index changes require index_updates", name)
+            parts = name.split("/")
+            if len(parts) > 3 and parts[:2] == ["analysis", "revisions"] and parts[2] in revisions:
+                self._reject("REVISION_IMMUTABLE", "Existing revision namespaces are frozen", name)
+
+    def _verify_preservation(self, reader: PackageReader, temporary: Path, indexes: set[str]) -> None:
+        with zipfile.ZipFile(temporary) as output:
+            names = set(output.namelist())
+            for info in reader._archive.infolist():
+                name = info.filename
+                if name not in names:
+                    self._reject("PRESERVATION_FAILED", "Input entry disappeared", name)
+                if name == "manifest.json" or name in indexes:
+                    continue
+                with reader._archive.open(info) as old, output.open(name) as new:
+                    while True:
+                        before, after = old.read(CHUNK_SIZE), new.read(CHUNK_SIZE)
+                        if before != after:
+                            self._reject("PRESERVATION_FAILED", "Input bytes changed", name)
+                        if not before:
+                            break

tests/test_package_writer.py

@@ -0,0 +1,283 @@
+import hashlib
+import os
+import zipfile
+
+import pytest
+
+from viano_bug_analysis.package import (
+    PackageLimits, PackageReader, PackageValidator, PackageWriteError, PackageWriter,
+)
+
+
+def assert_rejected(writer, source, target, code=None, **kwargs):
+    before = source.read_bytes()
+    target_before = target.read_bytes() if target.exists() else None
+    with pytest.raises(PackageWriteError) as error:
+        writer.write(source, target, **kwargs)
+    if code:
+        assert error.value.result.issues[0].code == code
+    assert source.read_bytes() == before
+    assert (target.read_bytes() if target.exists() else None) == target_before
+    assert not list(target.parent.glob(".vbcase-*.tmp"))
+
+
+def test_full_roundtrip(manifest, make_package, tmp_path):
+    manifest["extension"] = {"unknown": [1, 2]}
+    manifest["routing"]["custom"] = "preserve"
+    manifest["artifacts"]["vendor"] = "unknown/binary"
+    originals = {"original/report.json": b'{"report":"generic"}',
+                 "unknown/binary": bytes(range(256)), "empty/": b"",
+                 "analysis/revisions/0001/data.json": b"old revision",
+                 "analysis/current.json": b'{"revision":1}'}
+    source = make_package(entries=originals)
+    before = source.read_bytes()
+    target = tmp_path / "renamed.anything"
+    result = PackageWriter().write(
+        source, target,
+        additions={"analysis/revisions/0002/data.json": b"new revision"},
+        index_updates={"analysis/current.json": b'{"revision":2}'},
+        manifest_updates={"state": "VALIDATED", "routing": {"destination": "cases"},
+                          "analysis_revision": 2, "history": {"analysis_revisions": 2},
+                          "artifacts": {"case": "analysis/revisions/0002/data.json"}},
+    )
+    assert result.valid and PackageValidator().validate(target).valid
+    assert source.read_bytes() == before
+    with PackageReader(target) as reader:
+        updated = reader.manifest
+        for field in ("package_id", "case_id", "created_at", "project", "extension"):
+            assert updated[field] == manifest[field]
+        assert updated["package_generation"] == 1
+        assert updated["analysis_revision"] == 2
+        assert updated["routing"]["custom"] == "preserve"
+        assert updated["artifacts"]["vendor"] == "unknown/binary"
+        assert updated["updated_at"] >= manifest["updated_at"]
+        for name, data in originals.items():
+            if name != "analysis/current.json" and not name.endswith("/"):
+                assert reader.read_bytes(name) == data
+                assert updated["integrity"][name]["sha256"] == hashlib.sha256(data).hexdigest()
+    with zipfile.ZipFile(target) as archive:
+        assert "empty/" in archive.namelist()
+    next_target = tmp_path / "generation2"
+    PackageWriter().write(target, next_target)
+    with PackageReader(next_target) as reader:
+        assert reader.manifest["package_generation"] == 2
+        assert reader.manifest["analysis_revision"] == 2
+
+
+@pytest.mark.parametrize("name", ["original/report", "unknown/file", "analysis/revisions/0001/data"])
+def test_immutable_existing_files(name, make_package, tmp_path):
+    source = make_package(entries={name: b"old"})
+    assert_rejected(PackageWriter(), source, tmp_path / "out", "ARTIFACT_EXISTS", additions={name: b"new"})
+
+
+def test_frozen_revision_namespace(make_package, tmp_path):
+    source = make_package(entries={"analysis/revisions/0001/data": b"old"})
+    assert_rejected(PackageWriter(), source, tmp_path / "out", "REVISION_IMMUTABLE", additions={"analysis/revisions/0001/new": b"new"})
+
+
+@pytest.mark.parametrize("name", ["original/report", "unknown/file", "analysis/revisions/0001/data", "manifest.json"])
+def test_index_allowlist(name, make_package, tmp_path):
+    source = make_package(entries={name: b"old"} if name != "manifest.json" else {})
+    assert_rejected(PackageWriter(), source, tmp_path / "out", index_updates={name: b"new"})
+
+
+@pytest.mark.parametrize("name", ["../escape", "/abs", "x\\y", "manifest.json", "a:stream"])
+def test_unsafe_additions(name, make_package, tmp_path):
+    assert_rejected(PackageWriter(), make_package(), tmp_path / "out", "UNSAFE_PATH", additions={name: b"new"})
+
+
+@pytest.mark.parametrize("entries,addition", [({"a": b"x"}, "a/b"), ({"a/b": b"x"}, "a"), ({"A": b"x"}, "a"), ({"dir/": b""}, "dir")])
+def test_colliding_paths(entries, addition, make_package, tmp_path):
+    assert_rejected(PackageWriter(), make_package(entries=entries), tmp_path / "out", "ENTRY_COLLISION", additions={addition: b"new"})
+
+
+@pytest.mark.parametrize("field,value", [("package_id", "new"), ("case_id", "new"), ("package_generation", 99), ("created_at", "2026-01-01T00:00:00Z"), ("integrity", {}), ("project", {}), ("format_version", 2)])
+def test_reserved_manifest_fields(field, value, make_package, tmp_path):
+    assert_rejected(PackageWriter(), make_package(), tmp_path / "out", "MANIFEST_UPDATE_FORBIDDEN", manifest_updates={field: value})
+
+
+@pytest.mark.parametrize("updates", [{"state": "WRONG"}, {"artifacts": {"case": "absent"}}, {"routing": {"destination": "wrong"}}, {"analysis_revision": -1}])
+def test_invalid_manifest_update(updates, make_package, tmp_path):
+    assert_rejected(PackageWriter(), make_package(), tmp_path / "out", manifest_updates=updates)
+
+
+def test_revision_and_history_cannot_decrease(manifest, make_package, tmp_path):
+    manifest["analysis_revision"] = 2
+    manifest["history"]["evidence"] = 3
+    source = make_package()
+    assert_rejected(PackageWriter(), source, tmp_path / "out", "REVISION_REGRESSION", manifest_updates={"analysis_revision": 1})
+    assert_rejected(PackageWriter(), source, tmp_path / "out", "HISTORY_REGRESSION", manifest_updates={"history": {"evidence": 2}})
+
+
+def test_invalid_input(make_package, tmp_path):
+    assert_rejected(PackageWriter(), make_package(raw=b"{}"), tmp_path / "out")
+
+
+def test_input_alias_and_hardlink(make_package, tmp_path):
+    source = make_package()
+    assert_rejected(PackageWriter(), source, source, "INPUT_TARGET_ALIAS", overwrite=True)
+    alias = tmp_path / "alias"
+    os.link(source, alias)
+    assert_rejected(PackageWriter(), source, alias, "INPUT_TARGET_ALIAS", overwrite=True)
+
+
+def test_explicit_target_replacement(make_package, tmp_path):
+    source, target = make_package(), tmp_path / "out"
+    target.write_bytes(b"old target")
+    assert_rejected(PackageWriter(), source, target, "TARGET_EXISTS")
+    assert PackageWriter().write(source, target, overwrite=True).valid
+
+
+def test_failure_preserves_existing_target(make_package, tmp_path):
+    source, target = make_package(), tmp_path / "out"
+    target.write_bytes(b"old target")
+    assert_rejected(PackageWriter(), source, target, overwrite=True, manifest_updates={"artifacts": {"case": "absent"}})
+
+
+@pytest.mark.parametrize("field,value", [("max_file_bytes", 1024), ("max_total_bytes", 2000), ("max_entries", 1), ("max_archive_bytes", 2000), ("max_manifest_bytes", 800)])
+def test_output_limits(field, value, make_package, tmp_path):
+    limits = PackageLimits(**{field: value})
+    assert_rejected(PackageWriter(limits), make_package(), tmp_path / "out", additions={"new": b"x" * 3000})
+
+
+def test_compressible_artifact_is_valid_with_default_policy(make_package, tmp_path):
+    source, target = make_package(), tmp_path / "out"
+    assert PackageWriter().write(source, target, additions={"new": b"x" * 100_000}).valid
+
+
+def test_integrity_extensions_retained(manifest, make_package, tmp_path):
+    manifest["integrity"]["analysis/current.json"] = {"size": 3, "sha256": hashlib.sha256(b"old").hexdigest(), "vendor": "preserve"}
+    source = make_package(entries={"analysis/current.json": b"old"})
+    result = PackageWriter().write(source, tmp_path / "out", index_updates={"analysis/current.json": b"new"})
+    assert result.manifest["integrity"]["analysis/current.json"]["vendor"] == "preserve"
+
+
+def test_atomic_replace_failure(make_package, tmp_path, monkeypatch):
+    source, target = make_package(), tmp_path / "out"
+    target.write_bytes(b"old")
+    def fail(*args):
+        raise OSError("simulated rename failure")
+    monkeypatch.setattr(os, "replace", fail)
+    assert_rejected(PackageWriter(), source, target, "WRITE_ERROR", overwrite=True)
+
+
+def test_competing_target_not_clobbered(make_package, tmp_path, monkeypatch):
+    source, target = make_package(), tmp_path / "out"
+    original_link = os.link
+    def race(src, dst):
+        target.write_bytes(b"competitor")
+        original_link(src, dst)
+    monkeypatch.setattr(os, "link", race)
+    with pytest.raises(PackageWriteError):
+        PackageWriter().write(source, target)
+    assert target.read_bytes() == b"competitor"
+    assert not list(tmp_path.glob(".vbcase-*.tmp"))
+
+
+def test_preservation_check_detects_valid_tampered_output(make_package, tmp_path, monkeypatch):
+    source = make_package(entries={"unknown": b"original"})
+    writer = PackageWriter()
+    original_validate = writer.validator.validate
+    def tamper(path):
+        with zipfile.ZipFile(path) as archive:
+            entries = {name: archive.read(name) for name in archive.namelist()}
+        import json
+        doc = json.loads(entries["manifest.json"])
+        entries["unknown"] = b"tampered"
+        doc["integrity"]["unknown"].update(size=8, sha256=hashlib.sha256(b"tampered").hexdigest())
+        entries["manifest.json"] = json.dumps(doc).encode()
+        with zipfile.ZipFile(path, "w") as archive:
+            for name, data in entries.items():
+                archive.writestr(name, data)
+        return original_validate(path)
+    monkeypatch.setattr(writer.validator, "validate", tamper)
+    assert_rejected(writer, source, tmp_path / "out", "MANIFEST_CHANGED")
+
+
+def test_preservation_check_detects_unlisted_file_drop(make_package, tmp_path, monkeypatch):
+    source = make_package(entries={"unknown": b"original"})
+    writer = PackageWriter()
+    original_verify = writer._verify_preservation
+    def drop(reader, path, indexes):
+        with zipfile.ZipFile(path) as archive:
+            entries = {name: archive.read(name) for name in archive.namelist() if name != "unknown"}
+        with zipfile.ZipFile(path, "w") as archive:
+            for name, data in entries.items():
+                archive.writestr(name, data)
+        original_verify(reader, path, indexes)
+    monkeypatch.setattr(writer, "_verify_preservation", drop)
+    assert_rejected(writer, source, tmp_path / "out", "PRESERVATION_FAILED")
+
+
+def test_frozen_empty_revision_directory(make_package, tmp_path):
+    source = make_package(entries={"analysis/revisions/0001/": b""})
+    assert_rejected(PackageWriter(), source, tmp_path / "out", "REVISION_IMMUTABLE", additions={"analysis/revisions/0001/new": b"new"})
+
+
+def test_missing_target_parent(make_package, tmp_path):
+    source = make_package()
+    with pytest.raises(PackageWriteError):
+        PackageWriter().write(source, tmp_path / "absent" / "out")
+
+
+def test_new_index_and_type_validation(make_package, tmp_path):
+    source = make_package()
+    assert PackageWriter().write(source, tmp_path / "out", index_updates={"analysis/current.json": b"{}"}).valid
+    assert_rejected(PackageWriter(), source, tmp_path / "other", "ARTIFACT_TYPE", additions={"new": "string"})
+
+
+def test_future_timestamp_does_not_regress(manifest, make_package, tmp_path):
+    manifest["updated_at"] = "2099-01-01T00:00:00Z"
+    result = PackageWriter().write(make_package(), tmp_path / "out")
+    from datetime import datetime
+    assert datetime.fromisoformat(result.manifest["updated_at"]) == datetime.fromisoformat("2099-01-01T00:00:00+00:00")
+
+
+def test_fsync_failure(make_package, tmp_path, monkeypatch):
+    source, target = make_package(), tmp_path / "out"
+    target.write_bytes(b"old")
+    def fail(*args):
+        raise OSError("simulated disk failure")
+    monkeypatch.setattr(os, "fsync", fail)
+    assert_rejected(PackageWriter(), source, target, "WRITE_ERROR", overwrite=True)
+
+
+def test_byte_comparison_detects_changed_file(make_package, tmp_path, monkeypatch):
+    source = make_package(entries={"unknown": b"original"})
+    writer = PackageWriter()
+    original_verify = writer._verify_preservation
+    def change(reader, path, indexes):
+        with zipfile.ZipFile(path) as archive:
+            entries = {name: archive.read(name) for name in archive.namelist()}
+        entries["unknown"] = b"changed"
+        with zipfile.ZipFile(path, "w") as archive:
+            for name, data in entries.items():
+                archive.writestr(name, data)
+        original_verify(reader, path, indexes)
+    monkeypatch.setattr(writer, "_verify_preservation", change)
+    assert_rejected(writer, source, tmp_path / "out", "PRESERVATION_FAILED")
+
+
+def test_zip_comments_preserved(make_package, tmp_path):
+    source = make_package(entries={"unknown": b"data"})
+    with zipfile.ZipFile(source, "a") as archive:
+        archive.comment = b"vendor archive comment"
+    PackageWriter().write(source, tmp_path / "out")
+    with zipfile.ZipFile(tmp_path / "out") as archive:
+        assert archive.comment == b"vendor archive comment"
+
+
+def test_cleanup_failure_after_publication_reports_success(make_package, tmp_path, monkeypatch):
+    from pathlib import Path
+    source = make_package()
+    original_unlink = Path.unlink
+    def fail(path, *args, **kwargs):
+        if path.name.startswith(".vbcase-"):
+            raise OSError("simulated cleanup failure")
+        return original_unlink(path, *args, **kwargs)
+    monkeypatch.setattr(Path, "unlink", fail)
+    with pytest.warns(ResourceWarning, match="Package published"):
+        assert PackageWriter().write(source, tmp_path / "out").valid
+    assert PackageValidator().validate(tmp_path / "out").valid
+    for path in tmp_path.glob(".vbcase-*.tmp"):
+        original_unlink(path)