From 2852809fd4227a812282c2da1f24629b2663ff5b Mon Sep 17 00:00:00 2001
From: Dennis <denniswu501@gmail.com>
Date: Sun, 4 Oct 2026 20:19:06 -0700
Subject: [PATCH 01/16] Measure branch coverage for the hallmark package

Configure coverage.py to measure the hallmark package with branch
coverage and repository-relative paths, so data from separate CI jobs on
Linux and macOS can be combined. Add pytest-cov and pytest-timeout to the
dev dependency group and document `pytest --cov` for local runs.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 .gitignore     |  2 ++
 README.rst     |  4 ++--
 pyproject.toml | 15 ++++++++++++++-
 3 files changed, 18 insertions(+), 3 deletions(-)

diff --git a/.gitignore b/.gitignore
index 3a2b4eb..7e64860 100644
--- a/.gitignore
+++ b/.gitignore
@@ -7,6 +7,8 @@ venv/
 .pytest_cache/
 .ruff_cache/
 .coverage
+.coverage.*
+coverage.xml
 htmlcov/
 build/
 dist/
diff --git a/README.rst b/README.rst
index 04b526b..3bab969 100644
--- a/README.rst
+++ b/README.rst
@@ -135,8 +135,8 @@ version, operating system, and a minimal example without credentials or private
 
 From a development checkout, install the validation tools and run::
 
-    python -m pip install -e . pytest ruff sphinx
-    python -m pytest
+    python -m pip install -e . --group dev --group lint --group doc
+    python -m pytest --cov
     ruff check --select E,F mod test
     python -m sphinx -b html -W doc doc/_build
 
diff --git a/pyproject.toml b/pyproject.toml
index b5ab0d7..637320a 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -41,7 +41,7 @@ Repository = "https://github.com/l6a/hallmark"
 Issues = "https://github.com/l6a/hallmark/issues"
 
 [dependency-groups]
-dev = ["pytest"]
+dev = ["pytest", "pytest-cov", "pytest-timeout"]
 doc = ["sphinx"]
 lint = ["ruff"]
 
@@ -62,3 +62,16 @@ markers = [
     "ssh_client: exercise the actual local OpenSSH SFTP command parser",
     "ssh_integration: disposable loopback sshd (HALLMARK_RUN_SSH_TESTS=1 required)",
 ]
+
+[tool.coverage.run]
+source_pkgs = ["hallmark"]
+branch = true
+# Store repository-relative paths so data from Linux and macOS jobs combines.
+relative_files = true
+
+[tool.coverage.paths]
+hallmark = ["mod/hallmark", "*/mod/hallmark", "*/site-packages/hallmark"]
+
+[tool.coverage.report]
+show_missing = true
+skip_covered = true

From efb2bc971784fa8b1a3fb1b1cc93f068b23c849c Mon Sep 17 00:00:00 2001
From: Dennis <denniswu501@gmail.com>
Date: Sun, 4 Oct 2026 20:20:40 -0700
Subject: [PATCH 02/16] Run the whole suite with SSH tests in every CI job

Replace the separate build and ssh-transport jobs with one test matrix
that installs the OpenSSH server and runs the complete suite with
HALLMARK_RUN_SSH_TESTS=1 on Python 3.9-3.14 (Linux) and 3.9/3.14 (macOS).
Run CI for pull requests into any branch so stacked feature branches are
tested, lint once in its own job, update the actions to their current
majors, and check the built distributions with twine.

Move the ruff exclusions to the top-level table so `ruff check .` no
longer parses the shell cells in the demo notebooks.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 .github/workflows/ci.yml | 106 ++++++++++++++++++++++-----------------
 doc/usecase.rst          |   4 +-
 pyproject.toml           |   3 +-
 3 files changed, 65 insertions(+), 48 deletions(-)

diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 17a74fc..dcf2570 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -1,4 +1,5 @@
-# This workflow will install Python dependencies, run tests and lint with a variety of Python versions
+# Lints, tests every supported Python version on Linux and macOS, and checks
+# the built distributions and documentation.
 # For more information see: https://docs.github.com/en/actions/automating-builds-and-tests/building-and-testing-python
 
 name: CI
@@ -6,74 +7,89 @@ name: CI
 on:
   push:
     branches: [ "main", "dev" ]
+  # Pull requests into any branch, including stacked feature branches.
   pull_request:
-    branches: [ "main", "dev" ]
+  workflow_dispatch:
+
+permissions:
+  contents: read
+
+concurrency:
+  group: ci-${{ github.event.pull_request.number || github.ref }}
+  cancel-in-progress: ${{ github.event_name == 'pull_request' }}
 
 jobs:
-  build:
+  lint:
+    runs-on: ubuntu-24.04
+    steps:
+    - uses: actions/checkout@v7
+      with:
+        persist-credentials: false
+    - uses: actions/setup-python@v7
+      with:
+        python-version: "3.13"
+    - name: Install ruff
+      run: python -m pip install ruff
+    - name: Lint with ruff
+      run: ruff check .
 
-    runs-on: ubuntu-latest
+  # Every job runs the whole suite, including the loopback OpenSSH tests, so
+  # each Python version and both operating systems exercise every module.
+  test:
+    runs-on: ${{ matrix.os }}
+    timeout-minutes: 25
     strategy:
       fail-fast: false
       matrix:
-        python-version: ["3.9", "3.10","3.11","3.12","3.13","3.14"]
-
+        # Pinned because Python 3.9 builds may not exist for newer images.
+        os: [ubuntu-24.04]
+        python-version: ["3.9", "3.10", "3.11", "3.12", "3.13", "3.14"]
+        include:
+        - os: macos-latest
+          python-version: "3.9"
+        - os: macos-latest
+          python-version: "3.14"
     steps:
-    - uses: actions/checkout@v4
+    - uses: actions/checkout@v7
+      with:
+        persist-credentials: false
     - name: Set up Python ${{ matrix.python-version }}
-      uses: actions/setup-python@v3
+      uses: actions/setup-python@v7
       with:
         python-version: ${{ matrix.python-version }}
-    - name: Install dependencies
+    - name: Install the OpenSSH server
+      if: runner.os == 'Linux'
+      run: sudo apt-get update && sudo apt-get install -y --no-install-recommends openssh-server
+    - name: Install Hallmark and test tools
       run: |
         python -m pip install --upgrade pip
-        python -m pip install ruff pytest
-        python -m pip install -e .
-    - name: Lint with ruff
-      run: |
-        ruff check --select E,F --exclude ./demo
+        python -m pip install -e . --group dev
     - name: Test with pytest
-      run: |
-        pytest
+      env:
+        HALLMARK_RUN_SSH_TESTS: "1"
+      run: python -m pytest --timeout=600 -ra --durations=15
 
   distribution-and-docs:
-    runs-on: ubuntu-latest
+    runs-on: ubuntu-24.04
     steps:
-    - uses: actions/checkout@v4
+    - uses: actions/checkout@v7
       with:
         fetch-depth: 0
-    - uses: actions/setup-python@v5
+        persist-credentials: false
+    - uses: actions/setup-python@v7
       with:
-        python-version: "3.12"
-    - name: Build source and wheel distributions
+        python-version: "3.13"
+    - name: Build and check source and wheel distributions
       run: |
-        python -m pip install build
+        python -m pip install build twine
         python -m build
+        python -m twine check --strict dist/*
     - name: Install the wheel and documentation tools
       run: python -m pip install dist/*.whl sphinx
-    - name: Check the installed CLI
-      run: hm --help
+    - name: Check both installed commands
+      run: |
+        hm --help
+        hallmark --help
       working-directory: ${{ runner.temp }}
     - name: Build documentation with warnings as errors
       run: python -m sphinx -b html -W --keep-going doc "${{ runner.temp }}/hallmark-docs"
-
-  ssh-transport:
-    runs-on: ${{ matrix.os }}
-    strategy:
-      fail-fast: false
-      matrix:
-        os: [ubuntu-latest, macos-latest]
-    steps:
-    - uses: actions/checkout@v4
-    - uses: actions/setup-python@v3
-      with:
-        python-version: "3.9"
-    - name: Install Linux SSH server
-      if: runner.os == 'Linux'
-      run: sudo apt-get update && sudo apt-get install -y openssh-server
-    - name: Install Hallmark and tests
-      run: python -m pip install -e . pytest
-    - name: Exercise disposable OpenSSH server and client
-      env:
-        HALLMARK_RUN_SSH_TESTS: "1"
-      run: python -m pytest test/test_ssh_integration.py test/test_transport.py -q
diff --git a/doc/usecase.rst b/doc/usecase.rst
index c6fd20e..be22692 100644
--- a/doc/usecase.rst
+++ b/doc/usecase.rst
@@ -357,5 +357,5 @@ SSH configuration and known-hosts files.
 They do not connect to a lab server or change the user's SSH trust files.
 Enabling the tests requires the OpenSSH tools to be installed; missing
 tools cause a test failure.
-CI installs the server and runs these tests separately from the
-Python-version matrix.
+CI installs the server and runs the complete suite, these tests included,
+in every Linux and macOS test job.
diff --git a/pyproject.toml b/pyproject.toml
index 637320a..6f78fb2 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -50,10 +50,11 @@ where = ["mod"]
 
 [tool.ruff]
 target-version = "py39"
+# Excluded at the top level so the shell cells in demo notebooks are not parsed.
+extend-exclude = ["doc", "demo"]
 
 [tool.ruff.lint]
 select = ["E", "F"]
-exclude = ["doc", "demo"]
 
 [tool.pytest.ini_options]
 testpaths = ["test"]

From aa6f5960c8efdff6f44ff65be4db819e12f24bc6 Mon Sep 17 00:00:00 2001
From: Dennis <denniswu501@gmail.com>
Date: Sun, 4 Oct 2026 20:25:25 -0700
Subject: [PATCH 03/16] Treat exhausted prompt input consistently across Click
 versions in tests

Click 8.1, the newest release for Python 3.9, makes CliRunner read
exhausted input as an empty line instead of raising EOFError, so an empty
answer to the clone download prompt declines rather than aborts there.
Real terminals abort on end of input with both versions; only the test
harness differs. Expect the empty-line behaviour when Click is older than
8.2 so the Python 3.9 CI job tests the same outcomes.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 test/test_cli.py | 10 +++++++++-
 1 file changed, 9 insertions(+), 1 deletion(-)

diff --git a/test/test_cli.py b/test/test_cli.py
index df11e45..8083ffd 100644
--- a/test/test_cli.py
+++ b/test/test_cli.py
@@ -15,7 +15,7 @@
 
 import os
 import requests
-import importlib
+import importlib.metadata
 import pytest
 import pandas as pd
 from pathlib        import Path
@@ -30,6 +30,10 @@
 from hallmark.remote.plan import DownloadItem, DownloadPlan
 from hallmark.utils import use_working_directory
 
+_CLICK_RUNNER_SIGNALS_EOF = tuple(
+    int(part) for part in importlib.metadata.version("click").split(".")[:2]
+) >= (8, 2)
+
 cli_module = importlib.import_module("hallmark.cli")
 
 # files to create for testing, with a variety of a and i values to test regex encoding
@@ -915,6 +919,10 @@ def capture_get(url, **kwargs):
     assert "1 file(s); 4 bytes" in result.output
     assert "Download these files? [y/N]" in result.output
     assert "Source: https://example.test/data/" in result.output
+    # Click 8.1 (the newest for Python 3.9) reads exhausted CliRunner input
+    # as an empty line instead of end of input, so "" acts like "\n" there.
+    if answer == "" and not _CLICK_RUNNER_SIGNALS_EOF:
+        answer = "\n"
     if answer == "y\n":
         assert result.exit_code == 0, result.output
         assert requests_made == ["https://example.test/data/tiny.fits"]

From a4c4ccfde6333f1e06c6212c784ede602df83ca3 Mon Sep 17 00:00:00 2001
From: Dennis <denniswu501@gmail.com>
Date: Sun, 4 Oct 2026 20:25:34 -0700
Subject: [PATCH 04/16] Combine coverage across jobs and enforce a floor

Each test job writes coverage data under its OS and Python version and
uploads it. A coverage job combines them, writes the report to the job
summary, uploads XML and HTML reports, and fails below 90%. Some branches
only run on one Python version or operating system, so no single job
sees the total. The floor lives in the workflow rather than in
pyproject.toml because pytest-cov would apply it to each partial run.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 .github/workflows/ci.yml | 48 +++++++++++++++++++++++++++++++++++++++-
 1 file changed, 47 insertions(+), 1 deletion(-)

diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index dcf2570..d6a2040 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -67,7 +67,53 @@ jobs:
     - name: Test with pytest
       env:
         HALLMARK_RUN_SSH_TESTS: "1"
-      run: python -m pytest --timeout=600 -ra --durations=15
+        COVERAGE_FILE: .coverage.${{ runner.os }}-py${{ matrix.python-version }}
+      run: python -m pytest --cov --cov-report= --timeout=600 -ra --durations=15
+    - name: Upload coverage data
+      uses: actions/upload-artifact@v7
+      with:
+        name: coverage-${{ runner.os }}-py${{ matrix.python-version }}
+        path: .coverage.${{ runner.os }}-py${{ matrix.python-version }}
+        include-hidden-files: true
+        if-no-files-found: error
+        retention-days: 7
+
+  # Combines coverage from every test job. Some branches only run on one
+  # Python version or operating system, so no single job sees the total.
+  coverage:
+    needs: test
+    runs-on: ubuntu-24.04
+    steps:
+    - uses: actions/checkout@v7
+      with:
+        persist-credentials: false
+    - uses: actions/setup-python@v7
+      with:
+        python-version: "3.13"
+    - name: Install coverage
+      run: python -m pip install coverage
+    - uses: actions/download-artifact@v8
+      with:
+        pattern: coverage-*
+        merge-multiple: true
+    - name: Combine and report coverage
+      run: |
+        coverage combine
+        coverage report
+        coverage report --format=markdown >> "$GITHUB_STEP_SUMMARY"
+        coverage xml
+        coverage html
+    - name: Upload the combined report
+      uses: actions/upload-artifact@v7
+      with:
+        name: coverage-report
+        path: |
+          coverage.xml
+          htmlcov/
+    # The floor lives here rather than in pyproject.toml because pytest-cov
+    # would apply it to each partial run.
+    - name: Enforce the coverage floor
+      run: coverage report --fail-under=90
 
   distribution-and-docs:
     runs-on: ubuntu-24.04

From 015f0351a5d6b83b156de41b17765833132be3af Mon Sep 17 00:00:00 2001
From: Dennis <denniswu501@gmail.com>
Date: Sun, 4 Oct 2026 20:26:09 -0700
Subject: [PATCH 05/16] Upload combined coverage to Codecov

Upload the combined coverage.xml from the coverage job, verified with
OIDC so no repository secret is needed; pull requests from forks upload
tokenless. The upload cannot fail CI, and codecov.yml marks Codecov's
statuses informational, so the floor in the workflow remains the only
gate. Add a coverage badge and describe the coverage workflow in the
contribution guide.

The Codecov GitHub App must be installed on the lanisrc organization for
uploads to be accepted.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 .github/workflows/ci.yml | 14 ++++++++++++++
 README.rst               |  4 ++++
 codecov.yml              | 25 +++++++++++++++++++++++++
 doc/CONTRIBUTING.md      |  1 +
 4 files changed, 44 insertions(+)
 create mode 100644 codecov.yml

diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index d6a2040..36e1b88 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -83,6 +83,10 @@ jobs:
   coverage:
     needs: test
     runs-on: ubuntu-24.04
+    permissions:
+      contents: read
+      # Lets Codecov verify uploads with OIDC instead of a stored token.
+      id-token: write
     steps:
     - uses: actions/checkout@v7
       with:
@@ -110,6 +114,16 @@ jobs:
         path: |
           coverage.xml
           htmlcov/
+    # Codecov needs its GitHub App installed on the organization. Until then
+    # the upload fails without failing CI; the floor below is the gate.
+    # Pull requests from forks cannot request OIDC tokens and upload tokenless.
+    - name: Upload coverage to Codecov
+      uses: codecov/codecov-action@303a32d7a59b442fa8d48b6a1cc6825c09c847a5 # v7.1.1
+      with:
+        files: coverage.xml
+        disable_search: true
+        fail_ci_if_error: false
+        use_oidc: ${{ !github.event.pull_request.head.repo.fork }}
     # The floor lives here rather than in pyproject.toml because pytest-cov
     # would apply it to each partial run.
     - name: Enforce the coverage floor
diff --git a/README.rst b/README.rst
index 3bab969..f2e118e 100644
--- a/README.rst
+++ b/README.rst
@@ -3,6 +3,10 @@
 
     Reproducibility is the |hallmark|_ of the scientific method.
 
+..  image:: https://codecov.io/gh/lanisrc/hallmark/graph/badge.svg?branch=main
+    :target: https://codecov.io/gh/lanisrc/hallmark
+    :alt: Coverage
+
 Modern science has become so complex that many science projects rely
 on multiple software packages to work in unison, resulting in networks of
 data products along the analyses.
diff --git a/codecov.yml b/codecov.yml
new file mode 100644
index 0000000..7671050
--- /dev/null
+++ b/codecov.yml
@@ -0,0 +1,25 @@
+# Codecov reports are informational; CI enforces the coverage floor in
+# .github/workflows/ci.yml from the combined report it uploads.
+codecov:
+  notify:
+    after_n_builds: 1
+coverage:
+  precision: 1
+  round: down
+  status:
+    project:
+      default:
+        target: auto
+        threshold: 1%
+        informational: true
+    patch:
+      default:
+        target: 80%
+        informational: true
+comment:
+  layout: "condensed_header, diff, flags, files"
+  after_n_builds: 1
+ignore:
+  - "test/**"
+  - "demo/**"
+  - "doc/**"
diff --git a/doc/CONTRIBUTING.md b/doc/CONTRIBUTING.md
index dce1cbf..f30a85f 100644
--- a/doc/CONTRIBUTING.md
+++ b/doc/CONTRIBUTING.md
@@ -28,6 +28,7 @@
 
 * If the implementation of a feature or fix is less clear than the desired outcome, write unit or end-to-end tests first to define the expected functionality. Then write the code to make those tests pass.
 * If the implementation is as clear as (or clearer than) the goal, implement the solution first. Then write tests to ensure the solution covers edge cases and realistic user scenarios. The tests may reveal places where the implementation needs to be fixed, expanded, or simplified.
+* Run `pytest --cov` to see which lines the tests miss, or `pytest --cov --cov-report=html` for a browsable report in `htmlcov/`. CI runs the whole suite, including the loopback SSH tests, on every supported Python version, combines coverage from every job, and fails if the total drops below the floor set in `.github/workflows/ci.yml`. Codecov comments on each pull request with the coverage of its changed lines; those comments are informational.
 
 ## Data Management
 

From 84135ad772536dce7b266ab206715837c1f3453f Mon Sep 17 00:00:00 2001
From: Dennis <denniswu501@gmail.com>
Date: Sun, 4 Oct 2026 20:26:42 -0700
Subject: [PATCH 06/16] Share the loopback SSH server fixture

Move the disposable OpenSSH server fixture from test_ssh_integration.py
into conftest.py, unchanged, so survey workflow and fault-injection tests
can serve data over the same loopback server.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 test/conftest.py             | 107 +++++++++++++++++++++++++++++++++++
 test/test_ssh_integration.py | 104 ----------------------------------
 2 files changed, 107 insertions(+), 104 deletions(-)

diff --git a/test/conftest.py b/test/conftest.py
index 9aba900..f1dd825 100644
--- a/test/conftest.py
+++ b/test/conftest.py
@@ -1,10 +1,18 @@
+import getpass
+import os
 from pathlib import Path
+import shutil
+import socket
+import subprocess
+import time
+from urllib.parse import quote
 
 import pytest
 
 from hallmark import ParaFrame, Repo
 from hallmark.remote.download import _select_remote_config, execute_download_plan
 from hallmark.remote.plan import DownloadItem, DownloadPlan
+from hallmark.transport.ssh import SshBackend
 
 
 Standard_files = [
@@ -108,3 +116,102 @@ def download_selection(repo, output, selected, max_workers=4, show_progress=Fals
         backend_options=remote.get("backend_options") or {})
     return execute_download_plan(repo, plan, approved=True, max_workers=max_workers,
                                  show_progress=show_progress)
+
+
+@pytest.fixture
+def ssh_server(tmp_path, monkeypatch, request):
+    """Start a loopback SSH server with temporary keys and configuration."""
+    if os.environ.get("HALLMARK_RUN_SSH_TESTS") != "1":
+        pytest.skip("Set HALLMARK_RUN_SSH_TESTS=1 for disposable loopback SSH tests")
+    sshd = shutil.which("sshd") or "/usr/sbin/sshd"
+    if not Path(sshd).exists():
+        pytest.fail("Opted-in SSH tests require openssh-server")
+    for name in ("host", "client", "second", "untrusted"):
+        subprocess.run(
+            ["ssh-keygen", "-q", "-t", "ed25519", "-N", "", "-f", str(tmp_path / name)],
+            check=True,
+        )
+    authorized = tmp_path / "authorized"
+    authorized.write_text(
+        (tmp_path / "client.pub").read_text() + (tmp_path / "second.pub").read_text()
+    )
+    with socket.socket() as sock:
+        sock.bind(("127.0.0.1", 0))
+        port = sock.getsockname()[1]
+    config = tmp_path / "sshd_config"
+    config.write_text(f"""ListenAddress 127.0.0.1
+Port {port}
+HostKey {tmp_path}/host
+PidFile {tmp_path}/sshd.pid
+AuthorizedKeysFile {authorized}
+StrictModes no
+UsePAM no
+PasswordAuthentication no
+KbdInteractiveAuthentication no
+PermitRootLogin prohibit-password
+MaxSessions 4
+Subsystem sftp internal-sftp
+""")
+    mode = getattr(request, "param", None)
+    if mode == "sftp-only":
+        config.write_text(config.read_text() + "ForceCommand internal-sftp\n")
+    elif mode == "no-sftp":
+        config.write_text(
+            config.read_text().replace("Subsystem sftp internal-sftp\n", "")
+        )
+    elif mode == "max-one":
+        config.write_text(config.read_text().replace("MaxSessions 4", "MaxSessions 1"))
+    log = (tmp_path / "sshd.log").open("wb")
+    daemon = subprocess.Popen(
+        [sshd, "-D", "-e", "-f", str(config)], stderr=log, start_new_session=True
+    )
+    known = tmp_path / "known_hosts"
+    known.write_text(f"[127.0.0.1]:{port} " + (tmp_path / "host.pub").read_text())
+    client_config = tmp_path / "ssh_config"
+    client_config.write_text(f"""Host hm-test
+    HostName 127.0.0.1
+Host *
+    User {getpass.getuser()}
+    Port {port}
+    IdentityFile {tmp_path}/client
+    IdentityAgent none
+    IdentitiesOnly yes
+    UserKnownHostsFile {known}
+    GlobalKnownHostsFile /dev/null
+""")
+    original = SshBackend._ssh_options
+    monkeypatch.setattr(
+        SshBackend,
+        "_ssh_options",
+        lambda self, **kw: ["-F", str(client_config)] + original(self, **kw),
+    )
+    monkeypatch.delenv("HALLMARK_AUTH_FILE", raising=False)
+    monkeypatch.setenv("XDG_CONFIG_HOME", str(tmp_path / "xdg"))
+    root = tmp_path / "export"
+    root.mkdir()
+    try:
+        deadline = time.monotonic() + 5
+        while True:
+            if daemon.poll() is not None:
+                log.flush()
+                pytest.fail((tmp_path / "sshd.log").read_text())
+            try:
+                with socket.create_connection(("127.0.0.1", port), timeout=0.1):
+                    break
+            except OSError:
+                if time.monotonic() >= deadline:
+                    pytest.fail("Local SSH test server did not start")
+                time.sleep(0.05)
+        yield {
+            "root": root,
+            "port": port,
+            "known": known,
+            "base": tmp_path,
+            "config": client_config,
+            "daemon": daemon,
+            "url": "ssh://hm-test" + quote(str(root), safe="/") + "/",
+        }
+    finally:
+        daemon.terminate()
+        daemon.wait(timeout=5)
+        log.close()
diff --git a/test/test_ssh_integration.py b/test/test_ssh_integration.py
index c5b272b..319e665 100644
--- a/test/test_ssh_integration.py
+++ b/test/test_ssh_integration.py
@@ -4,13 +4,8 @@
 from dataclasses import replace
 import getpass
 import hashlib
-import os
 from pathlib import Path
-import shutil
-import socket
-import subprocess
 import time
-from urllib.parse import quote
 
 from click.testing import CliRunner
 import pytest
@@ -28,105 +23,6 @@
 pytestmark = pytest.mark.ssh_integration
 
 
-@pytest.fixture
-def ssh_server(tmp_path, monkeypatch, request):
-    """Start a loopback SSH server with temporary keys and configuration."""
-    if os.environ.get("HALLMARK_RUN_SSH_TESTS") != "1":
-        pytest.skip("Set HALLMARK_RUN_SSH_TESTS=1 for disposable loopback SSH tests")
-    sshd = shutil.which("sshd") or "/usr/sbin/sshd"
-    if not Path(sshd).exists():
-        pytest.fail("Opted-in SSH tests require openssh-server")
-    for name in ("host", "client", "second", "untrusted"):
-        subprocess.run(
-            ["ssh-keygen", "-q", "-t", "ed25519", "-N", "", "-f", str(tmp_path / name)],
-            check=True,
-        )
-    authorized = tmp_path / "authorized"
-    authorized.write_text(
-        (tmp_path / "client.pub").read_text() + (tmp_path / "second.pub").read_text()
-    )
-    with socket.socket() as sock:
-        sock.bind(("127.0.0.1", 0))
-        port = sock.getsockname()[1]
-    config = tmp_path / "sshd_config"
-    config.write_text(f"""ListenAddress 127.0.0.1
-Port {port}
-HostKey {tmp_path}/host
-PidFile {tmp_path}/sshd.pid
-AuthorizedKeysFile {authorized}
-StrictModes no
-UsePAM no
-PasswordAuthentication no
-KbdInteractiveAuthentication no
-PermitRootLogin prohibit-password
-MaxSessions 4
-Subsystem sftp internal-sftp
-""")
-    mode = getattr(request, "param", None)
-    if mode == "sftp-only":
-        config.write_text(config.read_text() + "ForceCommand internal-sftp\n")
-    elif mode == "no-sftp":
-        config.write_text(
-            config.read_text().replace("Subsystem sftp internal-sftp\n", "")
-        )
-    elif mode == "max-one":
-        config.write_text(config.read_text().replace("MaxSessions 4", "MaxSessions 1"))
-    log = (tmp_path / "sshd.log").open("wb")
-    daemon = subprocess.Popen(
-        [sshd, "-D", "-e", "-f", str(config)], stderr=log, start_new_session=True
-    )
-    known = tmp_path / "known_hosts"
-    known.write_text(f"[127.0.0.1]:{port} " + (tmp_path / "host.pub").read_text())
-    client_config = tmp_path / "ssh_config"
-    client_config.write_text(f"""Host hm-test
-    HostName 127.0.0.1
-Host *
-    User {getpass.getuser()}
-    Port {port}
-    IdentityFile {tmp_path}/client
-    IdentityAgent none
-    IdentitiesOnly yes
-    UserKnownHostsFile {known}
-    GlobalKnownHostsFile /dev/null
-""")
-    original = SshBackend._ssh_options
-    monkeypatch.setattr(
-        SshBackend,
-        "_ssh_options",
-        lambda self, **kw: ["-F", str(client_config)] + original(self, **kw),
-    )
-    monkeypatch.delenv("HALLMARK_AUTH_FILE", raising=False)
-    monkeypatch.setenv("XDG_CONFIG_HOME", str(tmp_path / "xdg"))
-    root = tmp_path / "export"
-    root.mkdir()
-    try:
-        deadline = time.monotonic() + 5
-        while True:
-            if daemon.poll() is not None:
-                log.flush()
-                pytest.fail((tmp_path / "sshd.log").read_text())
-            try:
-                with socket.create_connection(("127.0.0.1", port), timeout=0.1):
-                    break
-            except OSError:
-                if time.monotonic() >= deadline:
-                    pytest.fail("Local SSH test server did not start")
-                time.sleep(0.05)
-        yield {
-            "root": root,
-            "port": port,
-            "known": known,
-            "base": tmp_path,
-            "config": client_config,
-            "daemon": daemon,
-            "url": "ssh://hm-test" + quote(str(root), safe="/") + "/",
-        }
-    finally:
-        daemon.terminate()
-        daemon.wait(timeout=5)
-        log.close()
-
-
 @pytest.mark.parametrize("scheme", ["ssh", "sftp"])
 def test_parallel_exact_files_and_master_cleanup(ssh_server, tmp_path, scheme):
     root = ssh_server["root"]

From 8b0e99bc1875020bbc711fc227a2760950a4e231 Mon Sep 17 00:00:00 2001
From: Dennis <denniswu501@gmail.com>
Date: Sun, 4 Oct 2026 20:31:10 -0700
Subject: [PATCH 07/16] Serve synthetic survey releases from a local HTTP
 server

Add a real threaded HTTP server for tests that serves a directory on
127.0.0.1 with python, Apache or CyVerse style listings, records each
request, and applies scripted per-path faults: status codes, resets,
short or unannounced-length bodies, flipped bytes, stalls, slow drips,
redirects and pages without listings. Unlike MockServer, it exercises
the real Requests and urllib3 code paths.

Add deterministic generators for scaled-down DESI healpix, EHT UVFITS,
simulation export and telescope layouts, a CliRunner helper, and
fixtures that start servers and refuse non-loopback HTTP requests.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 test/conftest.py    |  43 +++++-
 test/http_server.py | 330 ++++++++++++++++++++++++++++++++++++++++++++
 test/survey.py      | 170 +++++++++++++++++++++++
 3 files changed, 542 insertions(+), 1 deletion(-)
 create mode 100644 test/http_server.py
 create mode 100644 test/survey.py

diff --git a/test/conftest.py b/test/conftest.py
index f1dd825..983ff92 100644
--- a/test/conftest.py
+++ b/test/conftest.py
@@ -5,14 +5,16 @@
 import socket
 import subprocess
 import time
-from urllib.parse import quote
+from urllib.parse import quote, urlsplit
 
 import pytest
+import requests
 
 from hallmark import ParaFrame, Repo
 from hallmark.remote.download import _select_remote_config, execute_download_plan
 from hallmark.remote.plan import DownloadItem, DownloadPlan
 from hallmark.transport.ssh import SshBackend
+from http_server import DatasetHTTPServer
 
 
 Standard_files = [
@@ -215,3 +217,42 @@ def ssh_server(tmp_path, monkeypatch, request):
         daemon.terminate()
         daemon.wait(timeout=5)
         log.close()
+
+
+@pytest.fixture
+def http_server(monkeypatch):
+    """Start loopback HTTP servers for directories; stop them at teardown.
+
+    Call the fixture value with a directory and an optional listing style.
+    Proxies are disabled so requests reach 127.0.0.1 directly.
+    """
+    for name in ("HTTP_PROXY", "HTTPS_PROXY", "ALL_PROXY",
+                 "http_proxy", "https_proxy", "all_proxy"):
+        monkeypatch.delenv(name, raising=False)
+    monkeypatch.setenv("NO_PROXY", "127.0.0.1,localhost")
+    servers = []
+
+    def start(root, listing_style="apache"):
+        server = DatasetHTTPServer(root, listing_style=listing_style).start()
+        servers.append(server)
+        return server
+
+    try:
+        yield start
+    finally:
+        for server in servers:
+            server.close()
+
+
+@pytest.fixture
+def loopback_only(monkeypatch):
+    """Fail any HTTP request that would leave this machine."""
+    send = requests.adapters.HTTPAdapter.send
+
+    def guarded(self, request, *args, **kwargs):
+        host = urlsplit(request.url).hostname
+        if host not in {"127.0.0.1", "localhost", "::1"}:
+            raise AssertionError(f"Test attempted a non-loopback request to {host}")
+        return send(self, request, *args, **kwargs)
+
+    monkeypatch.setattr(requests.adapters.HTTPAdapter, "send", guarded)
diff --git a/test/http_server.py b/test/http_server.py
new file mode 100644
index 0000000..40fb779
--- /dev/null
+++ b/test/http_server.py
@@ -0,0 +1,330 @@
+"""Serve a directory over loopback HTTP, with scripted faults, for tests.
+
+Unlike ``mock_server.MockServer``, this is a real server: requests,
+urllib3 and Hallmark's HTTP backend exercise their actual network code,
+including status handling, short reads, resets and timeouts.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+import fnmatch
+import html
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+import os
+from pathlib import Path
+import socket
+import struct
+import threading
+import time
+from urllib.parse import quote, unquote, urlsplit
+
+
+@dataclass(frozen=True)
+class Status:
+    """Respond with an HTTP status and an empty body."""
+    code: int
+    headers: tuple = ()
+
+
+@dataclass(frozen=True)
+class ResetAfter:
+    """Announce the full length, send ``nbytes``, then reset the connection."""
+    nbytes: int
+
+
+@dataclass(frozen=True)
+class ShortBody:
+    """Announce the full length, send ``nbytes``, then close cleanly."""
+    nbytes: int
+
+
+@dataclass(frozen=True)
+class NoLengthTruncate:
+    """Omit Content-Length, send ``nbytes``, then close cleanly."""
+    nbytes: int
+
+
+@dataclass(frozen=True)
+class BitFlip:
+    """Serve the file with one byte inverted at ``offset``."""
+    offset: int = 0
+
+
+@dataclass(frozen=True)
+class Stall:
+    """Wait before responding, or after the headers when ``after_headers``."""
+    seconds: float
+    after_headers: bool = False
+
+
+@dataclass(frozen=True)
+class SlowDrip:
+    """Send the body in ``chunk``-byte pieces ``delay`` seconds apart."""
+    chunk: int = 1024
+    delay: float = 0.05
+
+
+@dataclass(frozen=True)
+class Redirect:
+    """Redirect to ``location`` (an absolute or server-relative URL)."""
+    location: str
+    code: int = 302
+
+
+@dataclass(frozen=True)
+class NotAListing:
+    """Serve a directory as an ordinary page without listing markers."""
+
+
+@dataclass
+class _Rule:
+    pattern: str
+    fault: object
+    method: str
+    remaining: int | None
+
+
+@dataclass
+class FaultPlan:
+    """Faults to apply to matching request paths, consumed per request.
+
+    Paths are relative to the served root, without a leading slash;
+    directories end with ``/``. Rules are keyed by path rather than by a
+    global request count so concurrent download workers stay deterministic.
+    """
+    _rules: list = field(default_factory=list)
+    _lock: threading.Lock = field(default_factory=threading.Lock)
+
+    def add(self, pattern, fault, *, times=1, method="GET"):
+        """Apply ``fault`` to the next ``times`` matches (None: every match)."""
+        with self._lock:
+            self._rules.append(_Rule(pattern, fault, method, times))
+
+    def clear(self):
+        with self._lock:
+            self._rules.clear()
+
+    def take(self, method, path):
+        with self._lock:
+            for rule in self._rules:
+                if (rule.method in {method, "*"} and rule.remaining != 0
+                        and fnmatch.fnmatchcase(path, rule.pattern)):
+                    if rule.remaining is not None:
+                        rule.remaining -= 1
+                    return rule.fault
+        return None
+
+
+_MONTHS = ("Jan", "Feb", "Mar", "Apr", "May", "Jun",
+           "Jul", "Aug", "Sep", "Oct", "Nov", "Dec")
+
+
+def _listing(style, request_path, children):
+    """Render ``children`` as (name, is_directory, size) in a server's style."""
+    title = html.escape(request_path)
+    if style == "python":
+        items = "".join(
+            f'<li><a href="{quote(name)}{"/" if is_dir else ""}">'
+            f'{html.escape(name)}{"/" if is_dir else ""}</a></li>\n'
+            for name, is_dir, _ in children)
+        return (f'<!DOCTYPE HTML>\n<html lang="en"><head><meta charset="utf-8">'
+                f"<title>Directory listing for {title}</title></head>\n<body>\n"
+                f"<h1>Directory listing for {title}</h1>\n<hr>\n<ul>\n{items}"
+                "</ul>\n<hr>\n</body>\n</html>\n")
+    if style == "apache":
+        stamp = time.gmtime(1759600000)
+        date = (f"{stamp.tm_mday:02d}-{_MONTHS[stamp.tm_mon - 1]}-{stamp.tm_year} "
+                f"{stamp.tm_hour:02d}:{stamp.tm_min:02d}")
+        rows = "".join(
+            f'<a href="{quote(name)}{"/" if is_dir else ""}">'
+            f'{html.escape(name)}{"/" if is_dir else ""}</a>'
+            f'    {date}  {"-" if is_dir else size}\n'
+            for name, is_dir, size in children)
+        return (f"<html><head><title>Index of {title}</title></head><body>\n"
+                f"<h1>Index of {title}</h1><pre>"
+                '<a href="?C=N;O=D">Name</a>  <a href="?C=M;O=A">Last modified</a>'
+                '  <a href="?C=S;O=A">Size</a>\n<hr>'
+                '<a href="../">Parent Directory</a>\n'
+                f"{rows}<hr></pre></body></html>\n")
+    if style == "cyverse":
+        rows = "".join(
+            f'<tr class="object {"collection" if is_dir else "data-object"}">'
+            f'<td class="name"><a href="{quote(name)}{"/" if is_dir else ""}">'
+            f"{html.escape(name)}</a></td>"
+            f'<td class="size">{"" if is_dir else size}</td></tr>'
+            for name, is_dir, size in children)
+        return f"<html><body><table><tbody>{rows}</tbody></table></body></html>"
+    raise ValueError(f"Unknown listing style {style!r}")
+
+
+class _Handler(BaseHTTPRequestHandler):
+    protocol_version = "HTTP/1.1"
+    # Idle keep-alive connections end so the server can shut down promptly.
+    timeout = 5
+
+    def log_message(self, format, *args):
+        pass
+
+    def do_HEAD(self):
+        self._respond(head=True)
+
+    def do_GET(self):
+        self._respond(head=False)
+
+    def _respond(self, head):
+        server = self.server.owner
+        request_path = unquote(urlsplit(self.path).path, errors="strict")
+        relative = request_path.lstrip("/")
+        with server._lock:
+            server.requests.append((self.command, relative))
+        fault = server.faults.take(self.command, relative)
+        if isinstance(fault, Status):
+            return self._send_empty(fault.code, dict(fault.headers))
+        if isinstance(fault, Redirect):
+            return self._send_empty(fault.code, {"Location": fault.location})
+        if isinstance(fault, Stall) and not fault.after_headers:
+            server.release.wait(fault.seconds)
+
+        target = (server.root / relative).resolve()
+        if server.root.resolve() not in (target, *target.parents):
+            return self._send_empty(404)
+        if target.is_dir():
+            if not request_path.endswith("/"):
+                return self._send_empty(301, {"Location": request_path + "/"})
+            if isinstance(fault, NotAListing):
+                body = b"<html><body><p>Welcome</p></body></html>"
+            else:
+                children = sorted(
+                    (child.name, child.is_dir(),
+                     None if child.is_dir() else child.stat().st_size)
+                    for child in target.iterdir())
+                body = _listing(server.listing_style, request_path,
+                                children).encode("utf-8")
+            return self._send_body(body, "text/html; charset=utf-8", head, fault)
+        if not target.is_file():
+            return self._send_empty(404)
+        body = target.read_bytes()
+        if isinstance(fault, BitFlip) and body:
+            index = min(fault.offset, len(body) - 1)
+            body = body[:index] + bytes([body[index] ^ 0xFF]) + body[index + 1:]
+        return self._send_body(body, "application/octet-stream", head, fault)
+
+    def _send_empty(self, code, headers=None):
+        self.send_response(code)
+        for name, value in (headers or {}).items():
+            self.send_header(name, value)
+        self.send_header("Content-Length", "0")
+        self.end_headers()
+
+    def _send_body(self, body, content_type, head, fault):
+        server = self.server.owner
+        self.send_response(200)
+        self.send_header("Content-Type", content_type)
+        if isinstance(fault, NoLengthTruncate):
+            self.send_header("Connection", "close")
+            self.close_connection = True
+        else:
+            self.send_header("Content-Length", str(len(body)))
+        self.end_headers()
+        if head:
+            return
+        if isinstance(fault, Stall) and fault.after_headers:
+            self.wfile.flush()
+            server.release.wait(fault.seconds)
+        if isinstance(fault, (ResetAfter, ShortBody, NoLengthTruncate)):
+            self.wfile.write(body[:fault.nbytes])
+            self.wfile.flush()
+            self.close_connection = True
+            if isinstance(fault, ResetAfter):
+                # A zero linger timeout makes close() send RST instead of FIN.
+                self.connection.setsockopt(
+                    socket.SOL_SOCKET, socket.SO_LINGER, struct.pack("ii", 1, 0))
+            return
+        if isinstance(fault, SlowDrip):
+            for start in range(0, len(body), fault.chunk):
+                self.wfile.write(body[start:start + fault.chunk])
+                self.wfile.flush()
+                if server.release.wait(fault.delay):
+                    break
+            return
+        self.wfile.write(body)
+
+
+class _Server(ThreadingHTTPServer):
+    daemon_threads = True
+
+    def handle_error(self, request, client_address):
+        # Clients that abandon faulted transfers are expected; stay quiet.
+        pass
+
+
+class DatasetHTTPServer:
+    """A loopback HTTP server for a directory, with faults and a request log.
+
+    Args:
+        root: Directory to serve.
+        listing_style: ``"python"`` (http.server), ``"apache"`` (with sizes)
+            or ``"cyverse"`` (CyVerse WebDAV table rows with sizes).
+    """
+
+    def __init__(self, root, listing_style="python"):
+        self.root = Path(root)
+        self.root.mkdir(parents=True, exist_ok=True)
+        self.listing_style = listing_style
+        self.faults = FaultPlan()
+        self.requests = []
+        # Set at shutdown so stalled and dripping responses end promptly.
+        self.release = threading.Event()
+        self._lock = threading.Lock()
+        self._httpd = _Server(("127.0.0.1", 0), _Handler)
+        self._httpd.owner = self
+        self._thread = threading.Thread(target=self._httpd.serve_forever,
+                                        kwargs={"poll_interval": 0.05}, daemon=True)
+
+    @property
+    def port(self):
+        return self._httpd.server_address[1]
+
+    def url(self, path=""):
+        """Return the URL for ``path`` relative to the served root."""
+        return f"http://127.0.0.1:{self.port}/" + quote(path, safe="/")
+
+    def gets(self, prefix=""):
+        """Paths requested with GET, in order, optionally under ``prefix``."""
+        with self._lock:
+            return [path for method, path in self.requests
+                    if method == "GET" and path.startswith(prefix)]
+
+    def payload_gets(self):
+        """GET requests for files rather than directory listings."""
+        return [path for path in self.gets()
+                if path and not path.endswith("/")
+                and (self.root / path).is_file()]
+
+    def start(self):
+        self._thread.start()
+        return self
+
+    def close(self):
+        self.release.set()
+        self._httpd.shutdown()
+        self._httpd.server_close()
+        self._thread.join(timeout=5)
+
+    def __enter__(self):
+        return self.start()
+
+    def __exit__(self, *exc):
+        self.close()
+
+
+def write_file(root, relative, data):
+    """Write ``data`` (bytes or text) at ``root/relative``, creating folders."""
+    path = Path(root) / relative
+    path.parent.mkdir(parents=True, exist_ok=True)
+    if isinstance(data, str):
+        data = data.encode("utf-8")
+    path.write_bytes(data)
+    os.utime(path, (1759600000, 1759600000))
+    return path
diff --git a/test/survey.py b/test/survey.py
new file mode 100644
index 0000000..40b61f5
--- /dev/null
+++ b/test/survey.py
@@ -0,0 +1,170 @@
+"""Synthetic survey data releases and a CLI helper for workflow tests.
+
+The layouts follow real releases, scaled down to a few kilobytes:
+DESI DR1 healpix directories with sha256 manifests, EHT UVFITS files on
+CyVerse, a simulation export and a telescope tree. Contents are
+deterministic for a seed so failures reproduce.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import random
+
+from click.testing import CliRunner
+
+from hallmark.cli import hallmark
+from hallmark.utils import use_working_directory
+from http_server import write_file
+
+DESI_DARK = "public/dr1/spectro/redux/iron/healpix/main/dark"
+EHT_UVFITS = "EHTC_FirstM87Results_Apr2019/uvfits"
+
+
+def _fits_like(rng, size):
+    """Bytes that start like a FITS header, padded to ``size``."""
+    header = b"SIMPLE  =                    T".ljust(80) + b"END".ljust(80)
+    return header + rng.randbytes(max(0, size - len(header)))
+
+
+def _write_manifest(root, directory, names, files, name):
+    """Write a sha256sum-style manifest for ``names`` in ``directory``."""
+    lines = "".join(
+        f"{hashlib.sha256(files[f'{directory}/{n}']).hexdigest()}  {n}\n"
+        for n in names)
+    write_file(root, f"{directory}/{name}", lines)
+
+
+def make_desi_like_release(root, pixels=((230, 23040), (230, 23041)), seed=0):
+    """Write a DESI-like healpix release under ``root``.
+
+    Each pixel directory has redrock and qso_mgii FITS-like files, an
+    exposure CSV and a sha256 manifest named like the real release.
+
+    Returns:
+        dict: Release-relative path to contents, excluding manifests.
+    """
+    rng = random.Random(seed)
+    files = {}
+    for group, pixel in pixels:
+        directory = f"{DESI_DARK}/{group}/{pixel}"
+        names = [f"redrock-main-dark-{pixel}.fits",
+                 f"qso_mgii-main-dark-{pixel}.fits",
+                 f"hpixexp-main-dark-{pixel}.csv"]
+        for name in names:
+            if name.endswith(".csv"):
+                data = f"EXPID,NIGHT\n{rng.randrange(10**5)},20210514\n".encode()
+            else:
+                data = _fits_like(rng, 2880 + rng.randrange(2880))
+            write_file(root, f"{directory}/{name}", data)
+            files[f"{directory}/{name}"] = data
+        _write_manifest(root, directory, names, files,
+                        f"redux_iron_healpix_main_dark_{group}_{pixel}.sha256sum")
+    return files
+
+
+def make_eht_like_release(root, seed=0):
+    """Write EHT-like UVFITS files without a checksum manifest.
+
+    Returns:
+        dict: Release-relative path to contents.
+    """
+    rng = random.Random(seed)
+    files = {}
+    for day in (95, 96, 100, 101):
+        for band in ("lo", "hi"):
+            path = (f"{EHT_UVFITS}/SR1_M87_2017_{day:03d}_{band}"
+                    "_hops_netcal_StokesI.uvfits")
+            files[path] = _fits_like(rng, 2880 + rng.randrange(1440))
+            write_file(root, path, files[path])
+    return files
+
+
+def make_lab_export(root, runs=2, frames=3, seed=0):
+    """Write a simulation export laid out as ``run{run}/frame{frame}.h5``.
+
+    Returns:
+        dict: Export-relative path to contents.
+    """
+    rng = random.Random(seed)
+    files = {}
+    for run in range(1, runs + 1):
+        for frame in range(frames):
+            path = f"run{run}/frame{frame}.h5"
+            files[path] = b"\x89HDF\r\n\x1a\n" + rng.randbytes(512)
+            write_file(root, path, files[path])
+    return files
+
+
+def make_telescope_tree(root, sites=("ALMA", "SMA"), years=(2024, 2025),
+                        days=(95, 96), seed=0):
+    """Write observations laid out as ``{site}/{year:d}/{day:d}.fits``.
+
+    Returns:
+        dict: Worktree-relative path to contents.
+    """
+    rng = random.Random(seed)
+    files = {}
+    for site in sites:
+        for year in years:
+            for day in days:
+                path = f"{site}/{year}/{day}.fits"
+                files[path] = _fits_like(rng, 2880)
+                write_file(root, path, files[path])
+    return files
+
+
+def apply_dr_update(root, files, pixel=(230, 23042), seed=1):
+    """Reprocess one DESI-like file and add a pixel, updating manifests.
+
+    Returns:
+        tuple: The changed path and the paths of the new pixel's files.
+    """
+    rng = random.Random(seed)
+    changed = sorted(path for path in files if path.endswith(".fits"))[0]
+    files[changed] = _fits_like(rng, len(files[changed]))
+    write_file(root, changed, files[changed])
+    directory = changed.rsplit("/", 1)[0]
+    group, number = directory.split("/")[-2:]
+    names = sorted(path.rsplit("/", 1)[1] for path in files
+                   if path.rsplit("/", 1)[0] == directory)
+    _write_manifest(root, directory, names, files,
+                    f"redux_iron_healpix_main_dark_{group}_{number}.sha256sum")
+    added = make_desi_like_release(root, pixels=(pixel,), seed=seed)
+    files.update(added)
+    return changed, sorted(added)
+
+
+def apply_silent_corruption(root, path):
+    """Change a published file without updating its manifest."""
+    target = root / path
+    data = bytearray(target.read_bytes())
+    data[-1] ^= 0xFF
+    target.write_bytes(bytes(data))
+
+
+class Hm:
+    """Run ``hm`` commands in-process through Click's test runner."""
+
+    def __init__(self):
+        self.runner = CliRunner()
+
+    def __call__(self, *args, cwd, input=None, expect=0):
+        """Run ``hm ARGS`` in ``cwd`` and check its exit code.
+
+        Args:
+            cwd: Directory to run in.
+            input (str, optional): Text for prompts, such as ``"y\\n"``.
+            expect (int | None): Required exit code, or None to skip the check.
+
+        Returns:
+            click.testing.Result: The completed invocation.
+        """
+        with use_working_directory(cwd):
+            result = self.runner.invoke(hallmark, [str(arg) for arg in args],
+                                        input=input)
+        if expect is not None and result.exit_code != expect:
+            raise AssertionError(
+                f"hm {' '.join(map(str, args))} exited {result.exit_code}, "
+                f"expected {expect}:\n{result.output}") from result.exception
+        return result

From 9a927c8127ae2984f3de0e7d04566f9a49bb48a6 Mon Sep 17 00:00:00 2001
From: Dennis <denniswu501@gmail.com>
Date: Sun, 4 Oct 2026 20:34:18 -0700
Subject: [PATCH 08/16] Test survey workflows from the documented use cases

Add end-to-end tests for each workflow in doc/usecase.rst: telescope
ingest and update, a bare simulation catalog with explicit downloads
over SSH, isolated worktrees, nightly programmatic ingest, in-memory
ParaFrame selection and private lab data from a named SSH remote.

Also test DESI-like and EHT-like releases over loopback HTTP: published
sha256 checksums and listing sizes are recorded, URL-pattern adds read
no payloads, downloads honour --filter, --fmt and --output, a data
release update is rescanned and refreshed, a silently corrupted file is
rejected while the rest succeed, and Git catalogs and published
snapshots clone with a download prompt. A loopback guard keeps every
request on this machine.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 pyproject.toml      |   1 +
 test/survey.py      |   4 +-
 test/test_survey.py | 305 ++++++++++++++++++++++++++++++++++++++++++++
 3 files changed, 308 insertions(+), 2 deletions(-)
 create mode 100644 test/test_survey.py

diff --git a/pyproject.toml b/pyproject.toml
index 6f78fb2..86b76b9 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -62,6 +62,7 @@ addopts = "--strict-markers"
 markers = [
     "ssh_client: exercise the actual local OpenSSH SFTP command parser",
     "ssh_integration: disposable loopback sshd (HALLMARK_RUN_SSH_TESTS=1 required)",
+    "survey: end-to-end survey workflows against loopback HTTP and SSH servers",
 ]
 
 [tool.coverage.run]
diff --git a/test/survey.py b/test/survey.py
index 40b61f5..99fdc4f 100644
--- a/test/survey.py
+++ b/test/survey.py
@@ -115,13 +115,13 @@ def make_telescope_tree(root, sites=("ALMA", "SMA"), years=(2024, 2025),
 
 
 def apply_dr_update(root, files, pixel=(230, 23042), seed=1):
-    """Reprocess one DESI-like file and add a pixel, updating manifests.
+    """Reprocess one redrock file and add a pixel, updating manifests.
 
     Returns:
         tuple: The changed path and the paths of the new pixel's files.
     """
     rng = random.Random(seed)
-    changed = sorted(path for path in files if path.endswith(".fits"))[0]
+    changed = sorted(path for path in files if "/redrock-" in path)[0]
     files[changed] = _fits_like(rng, len(files[changed]))
     write_file(root, changed, files[changed])
     directory = changed.rsplit("/", 1)[0]
diff --git a/test/test_survey.py b/test/test_survey.py
new file mode 100644
index 0000000..9ba290a
--- /dev/null
+++ b/test/test_survey.py
@@ -0,0 +1,305 @@
+"""End-to-end survey workflows against loopback HTTP and SSH servers.
+
+Scenarios follow doc/usecase.rst and the public-data examples in
+doc/private_data.rst, using scaled-down synthetic releases. No request
+leaves the machine.
+"""
+
+import hashlib
+import shutil
+
+import pandas as pd
+import pytest
+
+from hallmark import ParaFrame, Repo
+from survey import (DESI_DARK, EHT_UVFITS, Hm, apply_dr_update,
+                    apply_silent_corruption, make_desi_like_release,
+                    make_eht_like_release, make_lab_export, make_telescope_tree)
+
+pytestmark = [pytest.mark.survey, pytest.mark.usefixtures("loopback_only")]
+
+TELESCOPE_FMT = "{site}/{year:d}/{day:d}.fits"
+DESI_PATTERN = "{group:d}/{pixel:d}/redrock-main-dark-{p:d}.fits"
+
+
+@pytest.fixture
+def hm(tmp_path, monkeypatch):
+    """Run hm with no local authentication profiles."""
+    monkeypatch.delenv("HALLMARK_AUTH_FILE", raising=False)
+    monkeypatch.setenv("XDG_CONFIG_HOME", str(tmp_path / "xdg"))
+    return Hm()
+
+
+@pytest.fixture
+def desi(tmp_path, http_server):
+    """A DESI-like release served with Apache-style listings."""
+    www = tmp_path / "www"
+    files = make_desi_like_release(www)
+    return {"www": www, "files": files, "server": http_server(www)}
+
+
+def _catalog(repo_path):
+    return pd.read_csv(repo_path / ".hm" / "data.tsv", sep="\t", dtype=str,
+                       keep_default_na=False)
+
+
+def _assert_downloaded(worktree, files, prefix=""):
+    for path, data in files.items():
+        relative = path[len(prefix):] if prefix else path
+        assert (worktree / relative).read_bytes() == data, relative
+
+
+# 1. CLI: standard repository ingest
+def test_telescope_ingest_commit_and_update(tmp_path, hm):
+    obs = tmp_path / "obs"
+    hm("init", obs, cwd=tmp_path)
+    files = make_telescope_tree(obs)
+    added = hm("add", TELESCOPE_FMT, cwd=obs)
+    assert all(path in added.output for path in files)
+    hm("commit", "-m", "Initial observation ingest", cwd=obs)
+    assert "SMA/2025/96.fits" not in hm("status", cwd=obs).output
+
+    (obs / "SMA/2025/96.fits").write_bytes(b"recalibrated")
+    assert "SMA/2025/96.fits" in hm("status", cwd=obs).output
+    hm("add", ".", cwd=obs)
+    hm("commit", "-m", "Recalibrate SMA day 96", cwd=obs)
+
+    log = hm("log", cwd=obs).output
+    assert log.index("Recalibrate SMA day 96") < log.index("Initial observation ingest")
+    info = hm("info", cwd=obs).output
+    assert str(obs.resolve()) in info or str(obs) in info
+    # Local catalogs record the checksum and the pattern's fields per file.
+    rows = _catalog(obs).set_index(["site", "year", "day"])
+    assert rows.loc[("SMA", "2025", "96"), "sha1"] == hashlib.sha1(
+        b"recalibrated").hexdigest()
+    assert len(rows) == len(files)
+
+
+# 2. CLI: bare repository with explicit downloads
+@pytest.mark.ssh_integration
+def test_bare_simulation_catalog_downloads_to_explicit_output(
+        tmp_path, hm, ssh_server):
+    files = make_lab_export(ssh_server["root"])
+    hm("init", "sim.hm", cwd=tmp_path)
+    sim = tmp_path / "sim.hm"
+    hm("add", ssh_server["url"], "--fmt", "run{run:d}/frame{frame:d}.h5", cwd=sim)
+    hm("commit", "-m", "Record remote catalog", cwd=sim)
+    assert "sim.hm" in hm("info", cwd=sim).output
+
+    preview = hm("download", "--all", "--output", "../outputs", "--dry-run", cwd=sim)
+    assert f"{len(files)} file(s)" in preview.output
+    assert not (tmp_path / "outputs").exists()
+    hm("download", "--all", "--output", "../outputs", cwd=sim, input="y\n")
+    _assert_downloaded(tmp_path / "outputs", files)
+
+    # The downloaded files can become a standard repository of their own.
+    outputs = tmp_path / "outputs"
+    hm("init", outputs, cwd=tmp_path)
+    hm("add", "run{run:d}/frame{frame:d}.h5", cwd=outputs)
+    hm("commit", "-m", "Version the downloaded frames", cwd=outputs)
+    assert len(_catalog(outputs)) == len(files)
+
+
+# 3. Python: branch-isolated analysis with multiple worktrees
+def test_second_worktree_keeps_observations_isolated(tmp_path):
+    repo = Repo.init(tmp_path / "obs")
+    make_telescope_tree(repo.worktree)
+    repo.add(TELESCOPE_FMT)
+    repo.commit("Initial observation ingest")
+
+    assert repo.add_worktree("obs2")
+    second = Repo(tmp_path / "obs2")
+    make_telescope_tree(second.worktree, sites=("APEX",), seed=7)
+    second.add(TELESCOPE_FMT)
+    second.commit("Observation ingest on branch obs2")
+
+    assert second.dothm.active_branch.name == "obs2"
+    assert repo.dothm.active_branch.name != "obs2"
+    assert not (repo.worktree / "APEX").exists()
+    assert "APEX" not in set(repo.state.data.site)
+    assert "APEX" in set(second.state.data.site)
+
+
+# 4. Python: programmatic repository updates
+def test_nightly_programmatic_ingest_keeps_every_version(tmp_path):
+    repo = Repo.init(tmp_path / "obs")
+    versions = set()
+    for night, day in enumerate((95, 96, 97)):
+        make_telescope_tree(repo.worktree, sites=("ALMA",), years=(2025,),
+                            days=(day,), seed=night)
+        # Each night also reprocesses the first observation.
+        (repo.worktree / "ALMA/2025/95.fits").write_bytes(f"pass {night}".encode())
+        repo.add(TELESCOPE_FMT)
+        repo.commit(f"Nightly ingest {night}")
+        versions.update(repo.state.data.sha1)
+    assert len(repo.state.data) == 3
+    assert all(repo.objects.contains(sha1) for sha1 in versions)
+    assert all(f"Nightly ingest {night}" in repo.log() for night in range(3))
+
+
+# 5. Python: in-memory state workflows
+def test_paraframe_selects_files_without_a_repository(tmp_path):
+    data = tmp_path / "data"
+    make_telescope_tree(data)
+    selected = ParaFrame.parse(TELESCOPE_FMT, base_path=data).filter(year=2025)
+    assert sorted(selected.path) == sorted(
+        path for path in (str(p.relative_to(data)) for p in data.rglob("*.fits"))
+        if "/2025/" in path)
+    assert not (data / ".hm").exists()
+
+
+# 6. CLI: private lab data over SSH
+@pytest.mark.ssh_integration
+def test_private_lab_data_from_named_ssh_remote(tmp_path, hm, ssh_server):
+    files = make_lab_export(ssh_server["root"], runs=1)
+    lab = tmp_path / "lab"
+    hm("init", lab, cwd=tmp_path)
+    hm("add", ssh_server["url"], "--fmt", "run{run:d}/frame{frame:d}.h5", cwd=lab)
+    hm("commit", "-m", "Catalog lab export", cwd=lab)
+    hm("set-config", "--remote-name", "campus", "--remote-url", ssh_server["url"],
+       cwd=lab)
+    preview = hm("download", "--remote", "campus", "--all", "--dry-run", cwd=lab)
+    assert "Source: ssh://hm-test" in preview.output
+    assert not (lab / "run1").exists()
+    hm("download", "--remote", "campus", "--all", cwd=lab, input="y\n")
+    _assert_downloaded(lab, files)
+
+
+def test_desi_like_pixel_records_published_checksums(tmp_path, hm, desi):
+    pixel_dir = f"{DESI_DARK}/230/23040/"
+    repo = tmp_path / "desi"
+    hm("init", repo, cwd=tmp_path)
+    hm("add", desi["server"].url(pixel_dir), "--filter", "redrock-*.fits", cwd=repo)
+    rows = _catalog(repo)
+    assert list(rows.path) == ["redrock-main-dark-23040.fits"]
+    data = desi["files"][pixel_dir + "redrock-main-dark-23040.fits"]
+    assert rows.loc[0, "checksum_algorithm"] == "sha256"
+    assert rows.loc[0, "checksum"] == hashlib.sha256(data).hexdigest()
+    assert rows.loc[0, "size_bytes"] == str(len(data))
+    hm("commit", "-m", "Catalog redrock file", cwd=repo)
+    hm("download", "--all", cwd=repo, input="y\n")
+    assert (repo / "redrock-main-dark-23040.fits").read_bytes() == data
+
+
+def test_url_pattern_add_catalogs_without_payload_requests(tmp_path, hm, desi):
+    server = desi["server"]
+    repo = tmp_path / "desi"
+    hm("init", repo, cwd=tmp_path)
+    hm("add", server.url(DESI_DARK + "/") + DESI_PATTERN, cwd=repo)
+    rows = _catalog(repo)
+    assert list(rows.pixel) == ["23040", "23041"]
+    assert set(rows.group) == {"230"}
+    payloads = [path for path in server.payload_gets()
+                if not path.endswith(".sha256sum")]
+    assert payloads == []
+
+    hm("commit", "-m", "Catalog redrock files", cwd=repo)
+    hm("download", "--all", "--filter", "230/23041/*", cwd=repo, input="y\n")
+    assert server.payload_gets()[-1].endswith("redrock-main-dark-23041.fits")
+    assert not (repo / "230/23040").exists()
+    hm("download", "--all", "--fmt", "{group:d}/23040/{name}", "--output", "../out",
+       cwd=repo, input="y\n")
+    _assert_downloaded(tmp_path / "out", {
+        path: data for path, data in desi["files"].items()
+        if path.endswith("redrock-main-dark-23040.fits")}, prefix=DESI_DARK + "/")
+
+
+def test_eht_like_cyverse_listing_records_sizes(tmp_path, hm, http_server):
+    www = tmp_path / "www"
+    files = make_eht_like_release(www)
+    server = http_server(www, listing_style="cyverse")
+    repo = tmp_path / "eht"
+    hm("init", repo, cwd=tmp_path)
+    hm("add", server.url(EHT_UVFITS + "/"), "--filter", "SR1_M87_2017_095_lo_*",
+       cwd=repo)
+    rows = _catalog(repo)
+    name = "SR1_M87_2017_095_lo_hops_netcal_StokesI.uvfits"
+    assert list(rows.path) == [name]
+    assert rows.loc[0, "size_bytes"] == str(len(files[f"{EHT_UVFITS}/{name}"]))
+    hm("commit", "-m", "Catalog M87 day 95", cwd=repo)
+    hm("download", "--all", cwd=repo, input="y\n")
+    assert (repo / name).read_bytes().startswith(b"SIMPLE")
+
+
+def test_data_release_update_rescans_and_refreshes(tmp_path, hm, desi):
+    server, www, files = desi["server"], desi["www"], desi["files"]
+    url = server.url(DESI_DARK + "/") + DESI_PATTERN
+    repo = tmp_path / "desi"
+    hm("init", repo, cwd=tmp_path)
+    hm("add", url, cwd=repo)
+    hm("commit", "-m", "DR1 catalog", cwd=repo)
+    hm("download", "--all", cwd=repo, input="y\n")
+
+    changed, added = apply_dr_update(www, files)
+    hm("add", url, cwd=repo)
+    rows = _catalog(repo).set_index("path")
+    relative = changed[len(DESI_DARK) + 1:]
+    assert rows.loc[relative, "checksum"] == hashlib.sha256(files[changed]).hexdigest()
+    assert "230/23042/redrock-main-dark-23042.fits" in rows.index
+    hm("commit", "-m", "DR1 reprocessing", cwd=repo)
+
+    # Replacing a file with different contents means deleting it first.
+    (repo / relative).unlink()
+    hm("download", "--all", cwd=repo, input="y\n")
+    _assert_downloaded(repo, {path: data for path, data in files.items()
+                              if path.endswith(".fits") and "redrock" in path},
+                       prefix=DESI_DARK + "/")
+
+
+def test_silently_corrupted_release_file_is_rejected(tmp_path, hm, desi):
+    server, www, files = desi["server"], desi["www"], desi["files"]
+    repo = tmp_path / "desi"
+    hm("init", repo, cwd=tmp_path)
+    hm("add", server.url(DESI_DARK + "/") + DESI_PATTERN, cwd=repo)
+    hm("commit", "-m", "DR1 catalog", cwd=repo)
+    corrupted = f"{DESI_DARK}/230/23041/redrock-main-dark-23041.fits"
+    apply_silent_corruption(www, corrupted)
+
+    result = hm("download", "--all", cwd=repo, input="y\n", expect=1)
+    assert "1 file" in result.output
+    assert not (repo / "230/23041/redrock-main-dark-23041.fits").exists()
+    assert not list(repo.rglob("*.part"))
+    intact = f"{DESI_DARK}/230/23040/redrock-main-dark-23040.fits"
+    assert (repo / "230/23040/redrock-main-dark-23040.fits").read_bytes() == \
+        files[intact]
+
+
+def test_clone_git_catalog_prompts_then_downloads(tmp_path, hm, desi):
+    source = tmp_path / "desi"
+    hm("init", source, cwd=tmp_path)
+    hm("add", desi["server"].url(DESI_DARK + "/") + DESI_PATTERN, cwd=source)
+    hm("commit", "-m", "DR1 catalog", cwd=source)
+
+    declined = hm("clone", source, "later", cwd=tmp_path, input="n\n")
+    assert "2 file(s)" in declined.output
+    assert "Download these files? [y/N]" in declined.output
+    assert not (tmp_path / "later/230").exists()
+    assert len(_catalog(tmp_path / "later")) == 2
+
+    hm("clone", source, "copy", cwd=tmp_path, input="y\n")
+    _assert_downloaded(tmp_path / "copy", {
+        path: data for path, data in desi["files"].items()
+        if "redrock" in path}, prefix=DESI_DARK + "/")
+    assert "DR1 catalog" in hm("log", cwd=tmp_path / "copy").output
+
+    hm("clone", source, "catalog-only.hm", "--no-download", cwd=tmp_path)
+    assert (tmp_path / "catalog-only.hm/data.tsv").is_file()
+    hm("clone", source, "refused.hm", cwd=tmp_path, expect=1)
+    assert not (tmp_path / "refused.hm").exists()
+
+
+def test_clone_published_catalog_snapshot_over_http(tmp_path, hm, desi, http_server):
+    source = tmp_path / "desi"
+    hm("init", source, cwd=tmp_path)
+    hm("add", desi["server"].url(DESI_DARK + "/") + DESI_PATTERN, cwd=source)
+    hm("commit", "-m", "DR1 catalog", cwd=source)
+    published = tmp_path / "published" / "catalogs" / "desi"
+    published.mkdir(parents=True)
+    for name in ("config.yml", "meta.yml", "data.tsv"):
+        shutil.copy(source / ".hm" / name, published / name)
+    catalogs = http_server(tmp_path / "published")
+
+    hm("clone", catalogs.url("catalogs/desi/"), "copy", cwd=tmp_path, input="y\n")
+    _assert_downloaded(tmp_path / "copy", {
+        path: data for path, data in desi["files"].items()
+        if "redrock" in path}, prefix=DESI_DARK + "/")

From 86aa45f00a931d58ab239e7769609caffe44629a Mon Sep 17 00:00:00 2001
From: Dennis <denniswu501@gmail.com>
Date: Sun, 4 Oct 2026 20:49:45 -0700
Subject: [PATCH 09/16] Inject network faults into discovery and downloads

Serve releases from the loopback HTTP server while it resets
connections, truncates bodies, returns 404/429/5xx, stalls, flips bytes
and redirects. Every faulted file must fail alone, leaving no partial
destination or temporary file and no URL in the error, while the other
files arrive intact; downloads must complete on a rerun; and a failed
rescan, including an unreadable SSH folder, must leave the committed
catalog and index untouched.

chaos.py adds a snapshot of a repository's tracked state and an
invariant check for corruption (unopenable repositories, objects that
do not match their checksums, committed files without objects, git
fsck errors) and leftover temporary files.

A body truncated without Content-Length is still accepted when the
catalog records the file's size but no checksum; that test is a strict
expected failure until the download size is verified.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 pyproject.toml             |   1 +
 test/chaos.py              | 118 +++++++++++++++++++++
 test/test_chaos_network.py | 205 +++++++++++++++++++++++++++++++++++++
 3 files changed, 324 insertions(+)
 create mode 100644 test/chaos.py
 create mode 100644 test/test_chaos_network.py

diff --git a/pyproject.toml b/pyproject.toml
index 86b76b9..761da80 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -63,6 +63,7 @@ markers = [
     "ssh_client: exercise the actual local OpenSSH SFTP command parser",
     "ssh_integration: disposable loopback sshd (HALLMARK_RUN_SSH_TESTS=1 required)",
     "survey: end-to-end survey workflows against loopback HTTP and SSH servers",
+    "chaos: fault injection, crash consistency and seeded random operations",
 ]
 
 [tool.coverage.run]
diff --git a/test/chaos.py b/test/chaos.py
new file mode 100644
index 0000000..ea5bd0b
--- /dev/null
+++ b/test/chaos.py
@@ -0,0 +1,118 @@
+"""Helpers for fault-injection ("chaos monkey") tests.
+
+``check_invariants`` describes what must hold for a repository after any
+operation, failed or not. Problems are grouped by severity:
+
+* corruption: the repository cannot be opened, an object does not match
+  its checksum, a committed file has no object, or Git reports damage;
+* debris: temporary files that a finished operation left behind.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+import hashlib
+from io import StringIO
+from pathlib import Path
+import subprocess
+
+import pandas as pd
+
+from hallmark import Repo
+
+STATE_FILES = ("config.yml", "meta.yml", "data.tsv")
+DEBRIS_PATTERNS = ("**/.*.part", "**/.*.tmp", "**/.hallmark-checkout-*")
+
+
+@dataclass
+class Violations:
+    """Invariant violations found in a repository."""
+    corruption: list = field(default_factory=list)
+    debris: list = field(default_factory=list)
+
+    def __bool__(self):
+        return bool(self.corruption or self.debris)
+
+    def __str__(self):
+        lines = [f"corruption: {item}" for item in self.corruption]
+        lines += [f"debris: {item}" for item in self.debris]
+        return "\n".join(lines) or "no violations"
+
+
+def _git(dothm, *args):
+    return subprocess.run(["git", "-C", str(dothm), *args], check=True,
+                          capture_output=True, text=True).stdout
+
+
+def snapshot_hm(repo_path):
+    """Capture a repository's tracked state: state files, index and refs."""
+    repo_path = Path(repo_path)
+    dothm = repo_path if repo_path.name.endswith(".hm") else repo_path / ".hm"
+    return {
+        "files": {name: (dothm / name).read_bytes() for name in STATE_FILES
+                  if (dothm / name).is_file()},
+        "index": _git(dothm, "ls-files", "--stage"),
+        "refs": _git(dothm, "for-each-ref", "--format=%(refname) %(objectname)"),
+        "head": subprocess.run(["git", "-C", str(dothm), "symbolic-ref", "-q", "HEAD"],
+                               capture_output=True, text=True).stdout.strip()
+        or _git(dothm, "rev-parse", "HEAD").strip(),
+    }
+
+
+def check_invariants(repo_path, *, fsck=True):
+    """Check a repository for corruption and leftover temporary files.
+
+    Args:
+        repo_path: Worktree or bare ``.hm`` path.
+        fsck (bool): Also run ``git fsck``, which is slower.
+
+    Returns:
+        Violations: Empty (falsy) when every invariant holds.
+    """
+    found = Violations()
+    try:
+        repo = Repo(repo_path)
+        repo.status()
+    except Exception as exc:
+        found.corruption.append(
+            f"repository does not open: {type(exc).__name__}: {exc}")
+        return found
+
+    objects = repo.objects.root
+    if objects.is_dir():
+        for path in objects.glob("??/*"):
+            if path.name.startswith("."):
+                found.debris.append(str(path))
+                continue
+            digest = hashlib.sha1(path.read_bytes()).hexdigest()
+            if digest != path.parent.name + path.name:
+                found.corruption.append(f"object {path} has checksum {digest}")
+
+    dothm = Path(repo.dothm.path)
+    for head in repo.dothm.heads:
+        try:
+            text = _git(dothm, "show", f"{head.name}:data.tsv")
+        except subprocess.CalledProcessError:
+            continue
+        frame = pd.read_csv(StringIO(text), sep="\t", dtype=str,
+                            keep_default_na=False)
+        if "sha1" in frame.columns:
+            for sha1 in frame.sha1:
+                if sha1 and not repo.objects.contains(sha1):
+                    found.corruption.append(
+                        f"branch {head.name} references missing object {sha1}")
+
+    if fsck:
+        result = subprocess.run(["git", "-C", str(dothm), "fsck", "--no-dangling"],
+                                capture_output=True, text=True)
+        if result.returncode != 0:
+            found.corruption.append(f"git fsck: {result.stdout}{result.stderr}")
+
+    roots = {Path(repo_path)}
+    if repo.worktree is not None:
+        roots.add(Path(repo.worktree))
+    for root in roots:
+        for pattern in DEBRIS_PATTERNS:
+            found.debris.extend(str(path) for path in root.glob(pattern))
+    found.debris = sorted(set(found.debris))
+    return found
diff --git a/test/test_chaos_network.py b/test/test_chaos_network.py
new file mode 100644
index 0000000..dedf20c
--- /dev/null
+++ b/test/test_chaos_network.py
@@ -0,0 +1,205 @@
+"""Network faults during remote scans and downloads.
+
+A real loopback HTTP server misbehaves on chosen paths. Each failed file
+must leave no partial destination or temporary file, other files must
+still arrive intact, and a failed scan must leave the catalog unchanged.
+"""
+
+import hashlib
+import os
+
+import pytest
+
+from chaos import check_invariants, snapshot_hm
+from hallmark import Repo
+import hallmark.transport.http as http_transport
+from http_server import (BitFlip, NoLengthTruncate, NotAListing, Redirect,
+                         ResetAfter, ShortBody, SlowDrip, Stall, Status, write_file)
+from survey import Hm
+
+pytestmark = [pytest.mark.chaos, pytest.mark.usefixtures("loopback_only")]
+
+FILES = {f"rel/f{i}.bin": os.urandom(40000) for i in range(4)}
+
+# Faults that the HTTP transport detects without any checksum.
+TRANSPORT_FAULTS = {
+    "reset": ResetAfter(1000),
+    "short-body": ShortBody(1000),
+    "500": Status(500),
+    "503": Status(503),
+    "429": Status(429, (("Retry-After", "1"),)),
+    "404": Status(404),
+    "stall-before-headers": Stall(5),
+    "stall-mid-body": Stall(5, after_headers=True),
+}
+# Faults that only a published checksum can detect.
+CONTENT_FAULTS = {
+    "bit-flip": BitFlip(100),
+    "unannounced-truncation": NoLengthTruncate(1000),
+}
+
+
+@pytest.fixture(autouse=True)
+def fast_timeouts(monkeypatch):
+    monkeypatch.setattr(http_transport, "REMOTE_REQUEST_TIMEOUT", (2, 0.5))
+
+
+@pytest.fixture
+def release(tmp_path, http_server):
+    """Serve FILES; ``publish_checksums`` adds a sha256 manifest."""
+    www = tmp_path / "www"
+    for path, data in FILES.items():
+        write_file(www, path, data)
+
+    def serve(publish_checksums=True, listing_style="apache"):
+        if publish_checksums:
+            write_file(www, "rel/sha256sum.txt", "".join(
+                f"{hashlib.sha256(data).hexdigest()}  {path.split('/')[1]}\n"
+                for path, data in FILES.items()))
+        return http_server(www, listing_style=listing_style)
+
+    return serve
+
+
+def _catalog_repo(path, server):
+    repo = Repo.init(path)
+    repo.add(server.url("rel/") + "{name}.bin")
+    repo.commit("Catalog release")
+    return repo
+
+
+def _download_all(repo, max_workers=2):
+    plan = repo.plan_download(all_files=True)
+    return repo.download(plan, approved=True, max_workers=max_workers)
+
+
+def _assert_only_faulted_file_failed(repo, result, faulted, server):
+    assert result["failed"] == 1, result
+    assert result["succeeded"] == len(FILES) - 1, result
+    for path, data in FILES.items():
+        destination = repo.worktree / path.split("/", 1)[1]
+        if path == faulted:
+            assert not destination.exists()
+        else:
+            assert destination.read_bytes() == data
+    errors = " ".join(map(str, dict(result["errors"]).values())
+                      if isinstance(result["errors"], dict) else result["errors"])
+    assert server.url("rel/") not in errors
+    problems = check_invariants(repo.worktree)
+    assert not problems, problems
+
+
+@pytest.mark.parametrize("publish_checksums", [True, False],
+                         ids=["checksums", "no-checksums"])
+@pytest.mark.parametrize("fault", TRANSPORT_FAULTS.values(), ids=TRANSPORT_FAULTS)
+def test_transport_fault_fails_only_that_file(tmp_path, release, fault,
+                                              publish_checksums):
+    server = release(publish_checksums)
+    repo = _catalog_repo(tmp_path / "repo", server)
+    server.faults.add("rel/f1.bin", fault)
+    result = _download_all(repo)
+    _assert_only_faulted_file_failed(repo, result, "rel/f1.bin", server)
+
+
+@pytest.mark.parametrize("fault", CONTENT_FAULTS.values(), ids=CONTENT_FAULTS)
+def test_published_checksum_rejects_damaged_content(tmp_path, release, fault):
+    server = release(publish_checksums=True)
+    repo = _catalog_repo(tmp_path / "repo", server)
+    server.faults.add("rel/f2.bin", fault)
+    result = _download_all(repo)
+    _assert_only_faulted_file_failed(repo, result, "rel/f2.bin", server)
+
+
+@pytest.mark.xfail(strict=True, reason=(
+    "Bug: a body truncated without Content-Length is accepted when the "
+    "catalog records the file size but no checksum"))
+def test_catalogued_size_rejects_unannounced_truncation(tmp_path, release):
+    server = release(publish_checksums=False, listing_style="apache")
+    repo = _catalog_repo(tmp_path / "repo", server)
+    server.faults.add("rel/f3.bin", NoLengthTruncate(1000))
+    result = _download_all(repo)
+    _assert_only_faulted_file_failed(repo, result, "rel/f3.bin", server)
+
+
+def test_concurrent_faults_fail_independently(tmp_path, release):
+    server = release(publish_checksums=True)
+    repo = _catalog_repo(tmp_path / "repo", server)
+    server.faults.add("rel/f0.bin", ResetAfter(10))
+    server.faults.add("rel/f2.bin", BitFlip(0))
+    server.faults.add("rel/f3.bin", SlowDrip(chunk=4096, delay=0.01))
+    result = _download_all(repo, max_workers=4)
+    assert (result["succeeded"], result["failed"]) == (2, 2), result
+    assert (repo.worktree / "f3.bin").read_bytes() == FILES["rel/f3.bin"]
+    assert not check_invariants(repo.worktree)
+
+
+def test_rerun_after_transient_failure_completes(tmp_path, release):
+    server = release(publish_checksums=True)
+    repo = _catalog_repo(tmp_path / "repo", server)
+    server.faults.add("rel/f1.bin", Status(503), times=1)
+    assert _download_all(repo)["failed"] == 1
+    plan = repo.plan_download(file_paths=["f1.bin"])
+    assert repo.download(plan, approved=True)["succeeded"] == 1
+    for path, data in FILES.items():
+        assert (repo.worktree / path.split("/", 1)[1]).read_bytes() == data
+    assert not check_invariants(repo.worktree)
+
+
+SCAN_FAULTS = {
+    "listing-500": ("rel/sub/", Status(500)),
+    "listing-reset": ("rel/sub/", ResetAfter(10)),
+    "listing-stall": ("rel/sub/", Stall(5)),
+    "not-a-listing": ("rel/sub/", NotAListing()),
+    "redirect-off-host": ("rel/sub/", Redirect("http://127.0.0.2:9/elsewhere/")),
+    "redirect-outside-root": ("rel/sub/", Redirect("/other/")),
+    "manifest-500": ("rel/sub/sha256sum.txt", Status(500)),
+    "manifest-reset": ("rel/sub/sha256sum.txt", ResetAfter(5)),
+}
+
+
+@pytest.mark.parametrize("pattern, fault", SCAN_FAULTS.values(), ids=SCAN_FAULTS)
+def test_failed_rescan_keeps_catalog_and_reports_cleanly(
+        tmp_path, http_server, monkeypatch, pattern, fault):
+    www = tmp_path / "www"
+    for name in ("top.bin", "sub/a.bin", "sub/b.bin"):
+        write_file(www, f"rel/{name}", os.urandom(64))
+    write_file(www, "rel/sub/sha256sum.txt", "".join(
+        f"{hashlib.sha256((www / 'rel/sub' / n).read_bytes()).hexdigest()}  {n}\n"
+        for n in ("a.bin", "b.bin")))
+    server = http_server(www)
+    monkeypatch.delenv("HALLMARK_AUTH_FILE", raising=False)
+    hm = Hm()
+    repo = tmp_path / "repo"
+    hm("init", repo, cwd=tmp_path)
+    url = server.url("rel/") + "{name}.bin"
+    hm("add", url, cwd=repo)
+    hm("commit", "-m", "Catalog release", cwd=repo)
+    before = snapshot_hm(repo)
+
+    server.faults.add(pattern, fault)
+    result = hm("add", url, cwd=repo, expect=None)
+    assert result.exit_code != 0, result.output
+    assert "Traceback" not in result.output
+    assert snapshot_hm(repo) == before
+    assert not check_invariants(repo)
+
+
+@pytest.mark.ssh_integration
+def test_unreadable_remote_folder_stops_ssh_scan(tmp_path, ssh_server):
+    if hasattr(os, "geteuid") and os.geteuid() == 0:
+        pytest.skip("root can read folders without permission")
+    root = ssh_server["root"]
+    for name in ("a/1.h5", "b/2.h5"):
+        write_file(root, name, os.urandom(32))
+    repo = Repo.init(tmp_path / "repo")
+    repo.add(ssh_server["url"] + "{group}/{n:d}.h5")
+    repo.commit("Catalog export")
+    before = snapshot_hm(repo.worktree)
+    (root / "b").chmod(0)
+    try:
+        with pytest.raises(Exception) as raised:
+            repo.add(ssh_server["url"] + "{group}/{n:d}.h5")
+    finally:
+        (root / "b").chmod(0o755)
+    assert "Traceback" not in str(raised.value)
+    assert snapshot_hm(repo.worktree) == before

From 79745d43a81bd72470ac421295d2a48da35bf66f Mon Sep 17 00:00:00 2001
From: Dennis <denniswu501@gmail.com>
Date: Sun, 4 Oct 2026 20:49:57 -0700
Subject: [PATCH 10/16] Inject storage faults into state writes, commits and
 downloads

Fail the Nth file replacement or write with ENOSPC or EIO while staging,
committing and downloading, and make an output folder read-only. A
failed commit must keep the branch pointer, leave no corrupt object and
succeed when retried; a download that runs out of space or permission
must fail only the affected file.

Dothm.save_state writes config.yml, meta.yml and data.tsv one at a time,
so a failure after the first write leaves state files from different
generations; those cases are strict expected failures until the writes
are made atomic together.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 test/chaos.py              |  70 ++++++++++++++++++++
 test/test_chaos_storage.py | 128 +++++++++++++++++++++++++++++++++++++
 2 files changed, 198 insertions(+)
 create mode 100644 test/test_chaos_storage.py

diff --git a/test/chaos.py b/test/chaos.py
index ea5bd0b..fc15745 100644
--- a/test/chaos.py
+++ b/test/chaos.py
@@ -11,12 +11,16 @@
 from __future__ import annotations
 
 from dataclasses import dataclass, field
+import errno
 import hashlib
 from io import StringIO
+import os
+import pathlib
 from pathlib import Path
 import subprocess
 
 import pandas as pd
+import pytest
 
 from hallmark import Repo
 
@@ -39,6 +43,72 @@ def __str__(self):
         return "\n".join(lines) or "no violations"
 
 
+def skip_if_root():
+    """Skip permission tests when running as root, which ignores modes."""
+    if hasattr(os, "geteuid") and os.geteuid() == 0:
+        pytest.skip("root ignores file permissions")
+
+
+def fail_nth_replace(monkeypatch, nth, *, code=errno.ENOSPC, match=lambda target: True):
+    """Make the ``nth`` matching ``Path.replace`` call raise ``OSError(code)``.
+
+    Patch the method on the class: on Python 3.9 and 3.10, pathlib calls a
+    reference to ``os.replace`` captured at import, so patching ``os`` misses it.
+
+    Returns:
+        list: Targets of the matching calls, in order, for assertions.
+    """
+    original = pathlib.Path.replace
+    calls = []
+
+    def replace(self, target):
+        if match(Path(target)):
+            calls.append(Path(target))
+            if len(calls) == nth:
+                raise OSError(code, os.strerror(code), str(target))
+        return original(self, target)
+
+    monkeypatch.setattr(pathlib.Path, "replace", replace)
+    return calls
+
+
+class _FailingWriter:
+    """File proxy whose writes fail once ``limit`` bytes have been written."""
+
+    def __init__(self, handle, limit, code):
+        self._handle, self._limit, self._code = handle, limit, code
+        self._written = 0
+
+    def write(self, data):
+        if self._written + len(data) > self._limit:
+            raise OSError(self._code, os.strerror(self._code))
+        self._written += len(data)
+        return self._handle.write(data)
+
+    def __getattr__(self, name):
+        return getattr(self._handle, name)
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, *exc):
+        return self._handle.__exit__(*exc)
+
+
+def fail_writes_after(monkeypatch, limit, *, code=errno.ENOSPC,
+                      match=lambda path: True):
+    """Make binary writes to matching paths fail after ``limit`` bytes."""
+    original = pathlib.Path.open
+
+    def open_(self, mode="r", *args, **kwargs):
+        handle = original(self, mode, *args, **kwargs)
+        if "w" in mode and "b" in mode and match(self):
+            return _FailingWriter(handle, limit, code)
+        return handle
+
+    monkeypatch.setattr(pathlib.Path, "open", open_)
+
+
 def _git(dothm, *args):
     return subprocess.run(["git", "-C", str(dothm), *args], check=True,
                           capture_output=True, text=True).stdout
diff --git a/test/test_chaos_storage.py b/test/test_chaos_storage.py
new file mode 100644
index 0000000..ed468bb
--- /dev/null
+++ b/test/test_chaos_storage.py
@@ -0,0 +1,128 @@
+"""Local storage faults: full disks, I/O errors and read-only folders.
+
+Writes fail partway through staging, committing and downloading. Each
+failure must surface as an error and leave the repository consistent:
+no corrupt objects, no partial files and no half-updated state.
+"""
+
+import errno
+import hashlib
+import os
+
+import pytest
+
+from chaos import (check_invariants, fail_nth_replace, fail_writes_after,
+                   skip_if_root, snapshot_hm)
+from hallmark import Repo
+from http_server import write_file
+from survey import make_telescope_tree
+
+pytestmark = [pytest.mark.chaos, pytest.mark.usefixtures("loopback_only")]
+
+FMT = "{site}/{year:d}/{day:d}.fits"
+
+
+def _in_dothm(target):
+    return ".hm" in target.parts
+
+
+@pytest.mark.parametrize("nth", [
+    1,
+    pytest.param(2, marks=pytest.mark.xfail(strict=True, reason=(
+        "Bug: Dothm.save_state writes config.yml, meta.yml and data.tsv "
+        "separately, so a failed later write leaves mixed generations"))),
+    pytest.param(3, marks=pytest.mark.xfail(strict=True, reason=(
+        "Bug: Dothm.save_state writes config.yml, meta.yml and data.tsv "
+        "separately, so a failed later write leaves mixed generations"))),
+])
+def test_failed_state_write_keeps_one_generation(tmp_path, monkeypatch, nth):
+    reference = Repo.init(tmp_path / "reference")
+    make_telescope_tree(reference.worktree)
+    reference.add(FMT)
+    new = snapshot_hm(reference.worktree)["files"]
+
+    repo = Repo.init(tmp_path / "repo")
+    make_telescope_tree(repo.worktree)
+    old = snapshot_hm(repo.worktree)
+    calls = fail_nth_replace(monkeypatch, nth, match=_in_dothm)
+    with pytest.raises(OSError):
+        repo.add(FMT)
+    monkeypatch.undo()
+    assert len(calls) == nth
+
+    after = snapshot_hm(repo.worktree)
+    generations = {"old" if after["files"][name] == old["files"][name] else
+                   "new" if after["files"][name] == new[name] else "other"
+                   for name in after["files"]}
+    assert generations in ({"old"}, {"new"}), generations
+    assert after["index"] == old["index"]
+    assert not check_invariants(repo.worktree)
+
+
+@pytest.mark.parametrize("code", [errno.ENOSPC, errno.EIO])
+@pytest.mark.parametrize("nth", [1, 3])
+def test_failed_object_store_keeps_commit_retryable(tmp_path, monkeypatch, nth, code):
+    repo = Repo.init(tmp_path / "repo")
+    files = make_telescope_tree(repo.worktree)
+    repo.add(FMT)
+    before = snapshot_hm(repo.worktree)
+
+    objects = repo.objects.root
+    calls = fail_nth_replace(monkeypatch, nth, code=code,
+                             match=lambda target: objects in target.parents)
+    with pytest.raises(OSError):
+        repo.commit("Ingest")
+    monkeypatch.undo()
+    assert len(calls) == nth
+    assert snapshot_hm(repo.worktree)["refs"] == before["refs"]
+    assert not check_invariants(repo.worktree)
+
+    repo = Repo(repo.worktree)
+    repo.commit("Ingest")
+    assert all(repo.objects.contains(hashlib.sha1(data).hexdigest())
+               for data in files.values())
+    assert not check_invariants(repo.worktree)
+
+
+@pytest.fixture
+def served_release(tmp_path, http_server):
+    www = tmp_path / "www"
+    files = {f"rel/d{i}/f.bin": os.urandom(20000) for i in range(3)}
+    for path, data in files.items():
+        write_file(www, path, data)
+    write_file(www, "rel/sha256sum.txt", "".join(
+        f"{hashlib.sha256(data).hexdigest()}  {path[4:]}\n"
+        for path, data in files.items()))
+    server = http_server(www)
+    repo = Repo.init(tmp_path / "repo")
+    repo.add(server.url("rel/") + "d{i:d}/f.bin")
+    repo.commit("Catalog release")
+    return repo, files
+
+
+def test_disk_full_mid_download_fails_only_that_file(monkeypatch, served_release):
+    repo, files = served_release
+    fail_writes_after(monkeypatch, 5000,
+                      match=lambda path: path.name.startswith(".f.bin.")
+                      and path.parent.name == "d1")
+    result = repo.download(repo.plan_download(all_files=True), approved=True)
+    monkeypatch.undo()
+    assert (result["succeeded"], result["failed"]) == (2, 1), result
+    assert not (repo.worktree / "d1/f.bin").exists()
+    assert (repo.worktree / "d0/f.bin").read_bytes() == files["rel/d0/f.bin"]
+    assert not check_invariants(repo.worktree)
+
+
+def test_read_only_output_folder_fails_only_its_files(served_release):
+    skip_if_root()
+    repo, files = served_release
+    locked = repo.worktree / "d2"
+    locked.mkdir()
+    locked.chmod(0o555)
+    try:
+        result = repo.download(repo.plan_download(all_files=True), approved=True)
+    finally:
+        locked.chmod(0o755)
+    assert (result["succeeded"], result["failed"]) == (2, 1), result
+    assert not (locked / "f.bin").exists()
+    assert not check_invariants(repo.worktree)

From 7b4bc0ba9ca9d48ddf25efc16378882e27c67984 Mon Sep 17 00:00:00 2001
From: Dennis <denniswu501@gmail.com>
Date: Sun, 4 Oct 2026 20:49:57 -0700
Subject: [PATCH 11/16] Check crash consistency with self-killing subprocesses

Run hm in a child process that sends itself SIGKILL at a chosen call
while staging, committing, checking out, adding a worktree or
downloading. The parent requires the kill to have happened, then checks
that the repository is not corrupt and that rerunning the command
finishes the job. An interrupted download must never leave a partial
destination and must keep its unfinished data in a separate file.

Two known recovery bugs are strict expected failures: a checkout killed
after moving files to its backup folder cannot be rerun, and
add_worktree skips restoring files when its destination already has .hm.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 test/chaos_child.py      |  60 +++++++++++++
 test/test_chaos_crash.py | 184 +++++++++++++++++++++++++++++++++++++++
 2 files changed, 244 insertions(+)
 create mode 100644 test/chaos_child.py
 create mode 100644 test/test_chaos_crash.py

diff --git a/test/chaos_child.py b/test/chaos_child.py
new file mode 100644
index 0000000..3c9d956
--- /dev/null
+++ b/test/chaos_child.py
@@ -0,0 +1,60 @@
+"""Run ``hm`` in a process that kills itself at a chosen point.
+
+Used by test_chaos_crash.py; pytest does not collect this module.
+``HALLMARK_CHAOS_KILL`` holds a JSON object:
+
+* ``target``: ``"module:Class.method"`` or ``"module:function"`` to wrap;
+* ``nth``: which matching call triggers the kill (1-based);
+* ``when``: ``"before"`` or ``"after"`` the wrapped call runs;
+* ``match``: optional text that must appear in the call's arguments.
+
+The process sends itself SIGKILL, so no cleanup handlers run, as in a
+power cut or an out-of-memory kill. Arguments are passed to ``hm``, or
+``--api CODE`` runs Python with ``Repo`` imported, in the current folder.
+"""
+
+import importlib
+import json
+import os
+import signal
+import sys
+
+
+def _install(spec):
+    module_name, _, qualname = spec["target"].partition(":")
+    owner = importlib.import_module(module_name)
+    *parents, attribute = qualname.split(".")
+    for name in parents:
+        owner = getattr(owner, name)
+    original = getattr(owner, attribute)
+    count = 0
+
+    def wrapper(*args, **kwargs):
+        nonlocal count
+        if spec.get("match", "") in " ".join(map(str, args)):
+            count += 1
+            if count == spec["nth"] and spec.get("when", "before") == "before":
+                os.kill(os.getpid(), signal.SIGKILL)
+            result = original(*args, **kwargs)
+            if count == spec["nth"] and spec.get("when") == "after":
+                os.kill(os.getpid(), signal.SIGKILL)
+            return result
+        return original(*args, **kwargs)
+
+    setattr(owner, attribute, wrapper)
+
+
+def main(argv):
+    spec = os.environ.get("HALLMARK_CHAOS_KILL")
+    if spec:
+        _install(json.loads(spec))
+    if argv[:1] == ["--api"]:
+        from hallmark import Repo
+        exec(argv[1], {"Repo": Repo})
+        return 0
+    from hallmark.cli import hallmark
+    return hallmark.main(args=argv, prog_name="hm")
+
+
+if __name__ == "__main__":
+    sys.exit(main(sys.argv[1:]))
diff --git a/test/test_chaos_crash.py b/test/test_chaos_crash.py
new file mode 100644
index 0000000..04d14dd
--- /dev/null
+++ b/test/test_chaos_crash.py
@@ -0,0 +1,184 @@
+"""Crash consistency: hm is killed with SIGKILL partway through a command.
+
+A killed command may leave temporary files, but never a corrupt
+repository, and running the same command again must finish the job.
+"""
+
+import hashlib
+import json
+import os
+from pathlib import Path
+import signal
+import subprocess
+import sys
+import time
+
+import pytest
+
+from chaos import check_invariants, snapshot_hm
+from hallmark import Repo
+from http_server import SlowDrip, write_file
+from survey import make_telescope_tree
+
+pytestmark = [
+    pytest.mark.chaos,
+    pytest.mark.usefixtures("loopback_only"),
+    pytest.mark.skipif(not hasattr(signal, "SIGKILL"), reason="needs SIGKILL"),
+]
+
+CHILD = str(Path(__file__).with_name("chaos_child.py"))
+FMT = "{site}/{year:d}/{day:d}.fits"
+
+
+def _run(cwd, args, kill=None, input=None):
+    env = dict(os.environ)
+    env.pop("HALLMARK_CHAOS_KILL", None)
+    if kill is not None:
+        env["HALLMARK_CHAOS_KILL"] = json.dumps(kill)
+    return subprocess.run([sys.executable, CHILD, *args], cwd=cwd, env=env,
+                          input=input, capture_output=True, text=True, timeout=120)
+
+
+def _killed(cwd, args, kill, input=None):
+    result = _run(cwd, args, kill, input)
+    # Otherwise the kill point was never reached and the test proves nothing.
+    assert result.returncode == -signal.SIGKILL, result.stdout + result.stderr
+    problems = check_invariants(cwd)
+    assert not problems.corruption, problems
+    return result
+
+
+def _rerun(cwd, args, input=None):
+    result = _run(cwd, args, input=input)
+    assert result.returncode == 0, result.stdout + result.stderr
+    problems = check_invariants(cwd)
+    assert not problems, problems
+    return result
+
+
+def _replace(nth, match, when="after"):
+    return {"target": "pathlib:Path.replace", "nth": nth, "when": when,
+            "match": match}
+
+
+@pytest.fixture
+def telescope(tmp_path):
+    repo = Repo.init(tmp_path / "obs")
+    files = make_telescope_tree(repo.worktree)
+    return repo, files
+
+
+@pytest.mark.parametrize("nth", [1, 2, 3])
+def test_killed_add_recovers_on_rerun(tmp_path, telescope, nth):
+    repo, _ = telescope
+    reference = Repo.init(tmp_path / "reference")
+    make_telescope_tree(reference.worktree)
+    reference.add(FMT)
+
+    _killed(repo.worktree, ["add", FMT], _replace(nth, "/.hm/"))
+    _rerun(repo.worktree, ["add", FMT])
+    assert snapshot_hm(repo.worktree)["files"] == \
+        snapshot_hm(reference.worktree)["files"]
+
+
+@pytest.mark.parametrize("kill", [
+    _replace(1, "/objects/"),
+    _replace(5, "/objects/"),
+    {"target": "git.index.base:IndexFile.commit", "nth": 1, "when": "before"},
+], ids=["first-object", "fifth-object", "before-git-commit"])
+def test_killed_commit_recovers_on_rerun(telescope, kill):
+    repo, files = telescope
+    repo.add(FMT)
+    before = snapshot_hm(repo.worktree)["refs"]
+    _killed(repo.worktree, ["commit", "-m", "Ingest"], kill)
+    assert snapshot_hm(repo.worktree)["refs"] == before
+    _rerun(repo.worktree, ["commit", "-m", "Ingest"])
+    repo = Repo(repo.worktree)
+    assert "Ingest" in repo.log()
+    assert all(repo.objects.contains(hashlib.sha1(data).hexdigest())
+               for data in files.values())
+
+
+@pytest.fixture
+def two_branches(telescope):
+    """Commit main, then branch exp with two files changed; return to main."""
+    repo, files = telescope
+    repo.add(FMT)
+    repo.commit("main data")
+    assert _run(repo.worktree, ["checkout", "exp"]).returncode == 0
+    for path in ("ALMA/2024/95.fits", "SMA/2025/96.fits"):
+        (repo.worktree / path).write_bytes(b"exp version of " + path.encode())
+    exp = Repo(repo.worktree)
+    exp.add(FMT)
+    exp.commit("exp data")
+    assert _run(repo.worktree, ["checkout", "main"]).returncode == 0
+    return repo, files
+
+
+@pytest.mark.parametrize("nth", [
+    pytest.param(1, marks=pytest.mark.xfail(strict=True, reason=(
+        "Bug: a checkout killed after moving a file to its backup folder "
+        "cannot be rerun; the file is reported missing"))),
+    pytest.param(3, marks=pytest.mark.xfail(strict=True, reason=(
+        "Bug: a checkout killed after moving a file to its backup folder "
+        "cannot be rerun; the file is reported missing"))),
+])
+def test_killed_checkout_can_be_completed(two_branches, nth):
+    repo, _ = two_branches
+    _killed(repo.worktree, ["checkout", "exp"], _replace(nth, ".fits"))
+    _rerun(repo.worktree, ["checkout", "exp"])
+    assert (repo.worktree / "SMA/2025/96.fits").read_bytes() == \
+        b"exp version of SMA/2025/96.fits"
+
+
+@pytest.mark.xfail(strict=True, reason=(
+    "Bug: add_worktree returns early when the destination already has .hm, "
+    "so a rerun after a crash leaves the worktree without its files"))
+def test_killed_add_worktree_is_completed_by_rerun(telescope):
+    repo, files = telescope
+    repo.add(FMT)
+    repo.commit("main data")
+    code = 'Repo(".").add_worktree("obs2")'
+    # Killed after Git created the linked worktree, before restoring files.
+    _killed(repo.worktree, ["--api", code], {
+        "target": "hallmark.repo.objects:Objects.restore", "nth": 1,
+        "when": "before"})
+    _rerun(repo.worktree, ["--api", code])
+    second = repo.worktree.parent / "obs2"
+    for path, data in files.items():
+        assert (second / path).read_bytes() == data
+
+
+def test_killed_download_keeps_partial_data_separate(tmp_path, http_server):
+    www = tmp_path / "www"
+    data = os.urandom(400_000)
+    write_file(www, "rel/big.bin", data)
+    write_file(www, "rel/sha256sum.txt",
+               f"{hashlib.sha256(data).hexdigest()}  big.bin\n")
+    server = http_server(www)
+    repo = Repo.init(tmp_path / "repo")
+    repo.add(server.url("rel/") + "{name}.bin")
+    repo.commit("Catalog")
+    server.faults.add("rel/big.bin", SlowDrip(chunk=4096, delay=0.02), times=1)
+
+    env = dict(os.environ)
+    env.pop("HALLMARK_CHAOS_KILL", None)
+    child = subprocess.Popen([sys.executable, CHILD, "download", "--all"],
+                             cwd=repo.worktree, env=env, stdin=subprocess.PIPE,
+                             stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
+    child.stdin.write(b"y\n")
+    child.stdin.close()
+    deadline = time.monotonic() + 30
+    parts = repo.worktree.glob
+    while not any(path.stat().st_size for path in parts(".big.bin.*.part")):
+        assert child.poll() is None and time.monotonic() < deadline, \
+            "download finished or stalled before it could be interrupted"
+        time.sleep(0.01)
+    child.kill()
+    child.wait(timeout=10)
+
+    assert not (repo.worktree / "big.bin").exists()
+    assert list(repo.worktree.glob(".big.bin.*.part"))
+    assert not check_invariants(repo.worktree).corruption
+    _run(repo.worktree, ["download", "--all"], input="y\n")
+    assert (repo.worktree / "big.bin").read_bytes() == data

From c6a01f3aa548c8a6ec1e9cd07db4f5cac3a33508 Mon Sep 17 00:00:00 2001
From: Dennis <denniswu501@gmail.com>
Date: Sun, 4 Oct 2026 21:00:51 -0700
Subject: [PATCH 12/16] Run seeded random operation sequences against
 repository invariants

A seeded "monkey" edits, stages, commits and switches branches while
injecting full-disk and I/O errors, and a second one republishes,
rescans and downloads a release while the server misbehaves. Instead of
modelling every rule, they check what must always hold: operations
succeed or raise a documented error; a failed operation changes neither
the repository nor the worktree; checked-out files match their
catalogued checksums; nothing is ever corrupt; and downloaded files are
always versions the server published. A failure prints its seed, recent
operations and the command that reproduces it. HALLMARK_CHAOS_SEED and
HALLMARK_CHAOS_STEPS widen the search.

The monkey found that checkout before the first commit raises a raw
GitCommandError, recorded as a strict expected failure. Known bugs are
listed where the monkey steers around them.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 test/test_chaos_monkey.py | 254 ++++++++++++++++++++++++++++++++++++++
 1 file changed, 254 insertions(+)
 create mode 100644 test/test_chaos_monkey.py

diff --git a/test/test_chaos_monkey.py b/test/test_chaos_monkey.py
new file mode 100644
index 0000000..7275840
--- /dev/null
+++ b/test/test_chaos_monkey.py
@@ -0,0 +1,254 @@
+"""Seeded random operation sequences checked against repository invariants.
+
+A "monkey" edits files, stages, commits, switches branches and downloads
+while faults are injected at random. It does not model every rule;
+instead it checks properties that must always hold:
+
+* operations succeed or raise a documented error;
+* a failed operation changes neither the repository nor the worktree;
+* after a checkout, every tracked file matches its catalogued checksum;
+* the repository is never corrupt, and successful operations leave no
+  temporary files;
+* a downloaded file always equals a version the server published.
+
+Set HALLMARK_CHAOS_SEED (comma-separated) and HALLMARK_CHAOS_STEPS to
+explore further; nightly CI runs many more steps with fresh seeds.
+"""
+
+import errno
+import hashlib
+import os
+import random
+
+import pytest
+
+from chaos import check_invariants, fail_nth_replace, snapshot_hm
+from hallmark import Repo
+from hallmark.error import CheckoutError
+import hallmark.transport.http as http_transport
+from hallmark.transport.base import DownloadError
+from hallmark.repo.config import row_to_path
+from http_server import (BitFlip, NoLengthTruncate, ResetAfter, ShortBody, Status,
+                         write_file)
+
+pytestmark = [pytest.mark.chaos, pytest.mark.usefixtures("loopback_only")]
+
+SEEDS = [int(seed) for seed in
+         os.environ.get("HALLMARK_CHAOS_SEED", "101,202,303").split(",") if seed]
+STEPS = int(os.environ.get("HALLMARK_CHAOS_STEPS", "40"))
+FAULT_RATE = float(os.environ.get("HALLMARK_CHAOS_FAULT_RATE", "0.2"))
+FMT = "{site}/{year:d}/{day:d}.fits"
+# Known bugs the monkey must not trip over until they are fixed:
+# * Dothm.save_state is not atomic when a pattern change rewrites
+#   config.yml (test_chaos_storage.py), so the pattern never changes here
+#   and the add that first sets it runs without faults.
+# * A truncated body without Content-Length is accepted when no checksum
+#   is published (test_chaos_network.py), so every file has a checksum.
+# * Checkout before the first commit raises a raw GitCommandError
+#   (test_checkout_before_first_commit_reports_a_documented_error), so the
+#   monkey commits once before switching branches.
+EXPECTED_ERRORS = (CheckoutError, DownloadError, FileNotFoundError, OSError,
+                   RuntimeError, ValueError)
+
+
+class Monkey:
+    """Seeded operation log with a reproducible failure report."""
+
+    def __init__(self, seed, nodeid):
+        self.seed, self.nodeid = seed, nodeid
+        self.rng = random.Random(seed)
+        self.log = []
+
+    def fail(self, step, message):
+        recent = "\n".join(f"  {line}" for line in self.log[-30:])
+        pytest.fail(
+            f"seed {self.seed}, step {step}: {message}\nLast operations:\n{recent}\n"
+            f"Reproduce: HALLMARK_CHAOS_SEED={self.seed} "
+            f"HALLMARK_CHAOS_STEPS={step + 1} pytest '{self.nodeid}'",
+            pytrace=False)
+
+
+def _worktree_files(root):
+    return {path.relative_to(root).as_posix(): path.read_bytes()
+            for path in root.rglob("*.fits") if ".hm" not in path.parts}
+
+
+def _tracked_files_match_catalog(repo):
+    """Return paths whose bytes differ from the current branch's catalog."""
+    fmt = repo.state.config["data"][0]["fmt"]
+    wrong = []
+    for _, row in repo.state.data.iterrows():
+        path = repo.worktree / row_to_path(row, fmt)
+        if (not path.is_file()
+                or hashlib.sha1(path.read_bytes()).hexdigest() != row["sha1"]):
+            wrong.append(str(path.relative_to(repo.worktree)))
+    return wrong
+
+
+@pytest.mark.parametrize("seed", SEEDS)
+def test_local_monkey_keeps_repository_consistent(tmp_path, monkeypatch, request,
+                                                   seed):
+    monkey = Monkey(seed, request.node.nodeid)
+    rng = monkey.rng
+    root = tmp_path / "obs"
+    Repo.init(root)
+    branches = {"main"}
+    patterned = committed = False
+
+    for step in range(STEPS):
+        repo = Repo(root)
+        name = rng.choice(["write", "write", "delete", "add", "add", "commit",
+                           "commit", "checkout"])
+        if name == "checkout" and not committed:
+            name = "commit"
+        path = f"{rng.choice(['ALMA', 'SMA'])}/{rng.choice([2024, 2025])}/" \
+               f"{rng.randrange(1, 4)}.fits"
+        target = rng.choice(sorted(branches) + [f"b{step}"])
+        fault = None
+        if name in {"add", "commit", "checkout"} and rng.random() < FAULT_RATE:
+            fault = (rng.randrange(1, 4), rng.choice([errno.ENOSPC, errno.EIO]))
+            if name == "add" and not patterned:
+                fault = None
+        monkey.log.append(f"{step}: {name} path={path} target={target} "
+                          f"fault={fault}")
+
+        if name == "write":
+            write_file(root, path, rng.randbytes(rng.randrange(1, 200)))
+            continue
+        if name == "delete":
+            files = sorted(_worktree_files(root))
+            if files:
+                (root / rng.choice(files)).unlink()
+            continue
+
+        before, before_files = snapshot_hm(root), _worktree_files(root)
+        if fault is not None:
+            fail_nth_replace(monkeypatch, fault[0], code=fault[1])
+        try:
+            if name == "add":
+                repo.add(FMT if not patterned else ".")
+                patterned = True
+            elif name == "commit":
+                committed = repo.commit(f"step {step}") or committed
+            else:
+                repo.checkout(target)
+        except EXPECTED_ERRORS as exc:
+            monkeypatch.undo()
+            monkey.log[-1] += f" -> {type(exc).__name__}: {exc}"
+            after = snapshot_hm(root)
+            if name == "commit":
+                # Objects stored before the failure are harmless extras.
+                unchanged = after["refs"] == before["refs"] and \
+                    after["files"] == before["files"]
+            else:
+                unchanged = after == before
+            if not unchanged:
+                monkey.fail(step, f"failed {name} changed repository state")
+            if _worktree_files(root) != before_files and name != "commit":
+                monkey.fail(step, f"failed {name} changed worktree files")
+            problems = check_invariants(root, fsck=False)
+            if problems.corruption:
+                monkey.fail(step, str(problems))
+            continue
+        except Exception as exc:
+            monkeypatch.undo()
+            monkey.fail(step, f"undocumented {type(exc).__name__}: {exc}")
+        monkeypatch.undo()
+        monkey.log[-1] += " -> ok"
+        if name == "checkout":
+            branches.add(target)
+            wrong = _tracked_files_match_catalog(Repo(root))
+            if wrong:
+                monkey.fail(step, f"checkout left wrong tracked files: {wrong}")
+        problems = check_invariants(root, fsck=step % 10 == 9)
+        if problems:
+            monkey.fail(step, str(problems))
+    problems = check_invariants(root)
+    if problems.corruption:
+        monkey.fail(STEPS, str(problems))
+
+
+@pytest.mark.xfail(strict=True, reason=(
+    "Bug: before the first commit, checkout raises GitCommandError because "
+    "init leaves data.tsv uncommitted"))
+def test_checkout_before_first_commit_reports_a_documented_error(tmp_path):
+    repo = Repo.init(tmp_path / "obs")
+    try:
+        repo.checkout("main")
+    except EXPECTED_ERRORS:
+        pass
+
+
+@pytest.mark.parametrize("seed", SEEDS)
+def test_remote_monkey_never_lands_unpublished_bytes(tmp_path, monkeypatch,
+                                                      http_server, request, seed):
+    monkeypatch.setattr(http_transport, "REMOTE_REQUEST_TIMEOUT", (2, 0.5))
+    monkey = Monkey(seed, request.node.nodeid)
+    rng = monkey.rng
+    www = tmp_path / "www"
+    names = [f"d{i}/f{j}.bin" for i in range(2) for j in range(3)]
+    published = {}
+
+    def write_manifest(directory):
+        write_file(www, f"rel/{directory}/sha256sum.txt", "".join(
+            f"{hashlib.sha256((www / 'rel' / n).read_bytes()).hexdigest()}  "
+            f"{n.split('/')[1]}\n" for n in names if n.startswith(directory + "/")))
+
+    def publish(name, stale=False):
+        data = rng.randbytes(rng.randrange(1000, 20000))
+        write_file(www, f"rel/{name}", data)
+        published.setdefault(name, set()).add(hashlib.sha256(data).hexdigest())
+        if not stale:
+            write_manifest(name.split("/")[0])
+
+    for name in names:
+        publish(name, stale=True)
+    for directory in ("d0", "d1"):
+        write_manifest(directory)
+    server = http_server(www)
+    url = server.url("rel/") + "{d}/{f}.bin"
+    repo = Repo.init(tmp_path / "repo")
+    repo.add(url)
+    repo.commit("Catalog")
+    faults = [ResetAfter(100), ShortBody(100), Status(500), Status(404),
+              BitFlip(7), NoLengthTruncate(100)]
+
+    for step in range(STEPS):
+        name = rng.choice(["publish", "publish-stale", "rescan", "commit",
+                           "download", "download", "delete"])
+        target = rng.choice(names)
+        monkey.log.append(f"{step}: {name} {target}")
+        try:
+            if name == "publish":
+                publish(target)
+            elif name == "publish-stale":
+                publish(target, stale=True)
+            elif name == "rescan":
+                repo.add(url)
+            elif name == "commit":
+                repo.commit(f"step {step}")
+            elif name == "delete":
+                (repo.worktree / target).unlink(missing_ok=True)
+            else:
+                if rng.random() < 0.5:
+                    server.faults.add(f"rel/{target}", rng.choice(faults))
+                result = repo.download(repo.plan_download(all_files=True),
+                                       approved=True, max_workers=3)
+                monkey.log[-1] += f" -> {result['succeeded']} ok, " \
+                                  f"{result['failed']} failed"
+        except EXPECTED_ERRORS as exc:
+            monkey.log[-1] += f" -> {type(exc).__name__}: {exc}"
+        except Exception as exc:
+            monkey.fail(step, f"undocumented {type(exc).__name__}: {exc}")
+        finally:
+            server.faults.clear()
+            repo = Repo(repo.worktree)
+
+        for name_ in names:
+            path = repo.worktree / name_
+            if path.is_file() and hashlib.sha256(path.read_bytes()).hexdigest() \
+                    not in published[name_]:
+                monkey.fail(step, f"{name_} holds bytes the server never published")
+        problems = check_invariants(repo.worktree, fsck=False)
+        if problems.corruption or (name.startswith("download") and problems.debris):
+            monkey.fail(step, str(problems))

From 29134c60d143d0805cffd278720741e8285dbb30 Mon Sep 17 00:00:00 2001
From: Dennis <denniswu501@gmail.com>
Date: Sun, 4 Oct 2026 21:00:51 -0700
Subject: [PATCH 13/16] Add opt-in live survey tests against DESI and CyVerse

With HALLMARK_RUN_LIVE_TESTS=1, follow the public-collection examples
in doc/private_data.rst: catalog DESI DR1 healpix files and verify the
downloads against the release's sha256 manifest, and catalog and
download EHT M87 UVFITS files from CyVerse, including through a URL
pattern. Every plan is checked before approval so a run downloads a few
megabytes at most. Tests skip, rather than fail, when an archive is
down or times out.

CyVerse lists sizes such as "554K", which Hallmark does not record, so
CyVerse downloads are limited to single, exactly named files.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 pyproject.toml           |   1 +
 test/test_survey_live.py | 132 +++++++++++++++++++++++++++++++++++++++
 2 files changed, 133 insertions(+)
 create mode 100644 test/test_survey_live.py

diff --git a/pyproject.toml b/pyproject.toml
index 761da80..a38cdd9 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -64,6 +64,7 @@ markers = [
     "ssh_integration: disposable loopback sshd (HALLMARK_RUN_SSH_TESTS=1 required)",
     "survey: end-to-end survey workflows against loopback HTTP and SSH servers",
     "chaos: fault injection, crash consistency and seeded random operations",
+    "survey_live: tiny downloads from public archives (HALLMARK_RUN_LIVE_TESTS=1 required)",
 ]
 
 [tool.coverage.run]
diff --git a/test/test_survey_live.py b/test/test_survey_live.py
new file mode 100644
index 0000000..0e5f954
--- /dev/null
+++ b/test/test_survey_live.py
@@ -0,0 +1,132 @@
+"""Tiny end-to-end downloads from real public survey archives.
+
+These follow the public-collection examples in doc/private_data.rst
+against DESI DR1 and the EHT M87 release on CyVerse. They need network
+access, so they only run with HALLMARK_RUN_LIVE_TESTS=1 (nightly in CI).
+Every plan is checked before approval so a run downloads at most a few
+megabytes. CyVerse lists sizes such as "554K" that Hallmark does not
+record, so its downloads are limited to single, exactly named files.
+"""
+
+import hashlib
+import os
+
+import pytest
+import requests
+
+from hallmark import Repo
+
+pytestmark = [
+    pytest.mark.survey_live,
+    pytest.mark.skipif(os.environ.get("HALLMARK_RUN_LIVE_TESTS") != "1",
+                       reason="Set HALLMARK_RUN_LIVE_TESTS=1 to reach public archives"),
+]
+
+DESI = ("https://data.desi.lbl.gov/public/dr1/spectro/redux/iron/"
+        "healpix/main/dark/230/23040/")
+EHT = ("https://data.cyverse.org/dav-anon/iplant/commons/"
+       "cyverse_curated/EHTC_FirstM87Results_Apr2019/uvfits/")
+MAX_TARGET_BYTES = 2_000_000
+MAX_RUN_BYTES = 10_000_000
+_downloaded = []
+
+
+@pytest.fixture(scope="module")
+def archive_available():
+    """Skip, rather than fail, when an archive itself is down or overloaded.
+
+    The probe uses plain Requests with Hallmark's timeouts, so a failure
+    after a successful probe points at Hallmark rather than the archive.
+    """
+    checked = {}
+
+    def check(url):
+        if url not in checked:
+            try:
+                status = requests.get(url, timeout=(10, 30)).status_code
+                checked[url] = (status < 500, f"HTTP {status}")
+            except requests.RequestException as exc:
+                checked[url] = (False, type(exc).__name__)
+        available, detail = checked[url]
+        if not available:
+            pytest.skip(f"{url} is unavailable ({detail})")
+
+    return check
+
+
+def _download(repo, plan):
+    """Download a plan, skipping when the archive times out or errors."""
+    result = repo.download(plan, approved=True)
+    errors = " ".join(map(str, result["errors"]))
+    if result["failed"] and all(
+            "ReadTimeout" in str(error) or "ConnectTimeout" in str(error)
+            or "HTTP 5" in str(error) for error in result["errors"]):
+        pytest.skip(f"The archive failed during the download: {errors}")
+    return result
+
+
+def _approve_small(plan, expected_files, sizes_listed=True):
+    """Refuse to download anything unexpectedly large.
+
+    Without listed sizes, only a single file the test names exactly is
+    allowed, and its size is checked after the download.
+    """
+    assert plan.file_count == expected_files, plan.summary()
+    if sizes_listed:
+        assert plan.unknown_size_count == 0, plan.summary()
+    else:
+        assert plan.file_count == 1, plan.summary()
+    assert plan.known_bytes <= MAX_TARGET_BYTES, plan.summary()
+    assert sum(_downloaded) + plan.known_bytes <= MAX_RUN_BYTES
+    _downloaded.append(plan.known_bytes or MAX_TARGET_BYTES)
+
+
+def test_desi_dr1_files_match_published_checksums(tmp_path, archive_available):
+    archive_available(DESI)
+    repo = Repo.init(tmp_path / "desi")
+    repo.add(DESI, filter=["redrock-main-dark-23040.fits",
+                           "hpixexp-main-dark-23040.csv"])
+    rows = repo.state.data.set_index("path")
+    assert set(rows.checksum_algorithm) == {"sha256"}
+    repo.commit("Record remote catalog")
+
+    plan = repo.plan_download(all_files=True)
+    _approve_small(plan, expected_files=2)
+    result = _download(repo, plan)
+    assert (result["succeeded"], result["failed"]) == (2, 0), result
+    for path, row in rows.iterrows():
+        data = (repo.worktree / path).read_bytes()
+        assert hashlib.sha256(data).hexdigest() == row.checksum
+        assert len(data) == int(row.size_bytes)
+
+
+def test_eht_m87_uvfits_downloads_from_cyverse(tmp_path, archive_available):
+    archive_available(EHT)
+    name = "SR1_M87_2017_095_lo_hops_netcal_StokesI.uvfits"
+    repo = Repo.init(tmp_path / "eht")
+    repo.add(EHT, filter=name)
+    assert list(repo.state.data.path) == [name]
+    repo.commit("Record remote catalog")
+
+    plan = repo.plan_download(all_files=True)
+    _approve_small(plan, expected_files=1, sizes_listed=False)
+    result = _download(repo, plan)
+    assert (result["succeeded"], result["failed"]) == (1, 0), result
+    data = (repo.worktree / name).read_bytes()
+    assert data.startswith(b"SIMPLE")
+    assert len(data) <= MAX_TARGET_BYTES
+
+
+def test_eht_url_pattern_catalogs_every_day_and_band(tmp_path, archive_available):
+    archive_available(EHT)
+    repo = Repo.init(tmp_path / "eht")
+    repo.add(EHT + "SR1_M87_2017_{day:d}_{band}_hops_netcal_StokesI.uvfits")
+    rows = repo.state.data
+    assert set(rows.band) == {"lo", "hi"}
+    assert {95, 96, 100, 101} <= {int(day) for day in rows.day}
+    repo.commit("Record remote catalog")
+
+    plan = repo.plan_download(all_files=True, filter="*_100_lo_*")
+    _approve_small(plan, expected_files=1, sizes_listed=False)
+    result = _download(repo, plan)
+    assert (result["succeeded"], result["failed"]) == (1, 0), result

From 5a0eb926f9302b810e0930039f7638b37f65e8d8 Mon Sep 17 00:00:00 2001
From: Dennis <denniswu501@gmail.com>
Date: Sun, 4 Oct 2026 21:00:51 -0700
Subject: [PATCH 14/16] Run extended chaos and live survey tests nightly

Every night, run the chaos tests on Linux (Python 3.9 and 3.14) and
macOS with fresh seeds and 2000 random operations per seed, keeping the
failed repositories as artifacts, and run the live survey tests.
Scheduled and manual runs start only from the default branch, so pull
requests that change these tests or the workflow run it too, with fewer
operations. Document how to run and extend these tests and how to
record the bugs they find.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 .github/workflows/nightly.yml | 113 ++++++++++++++++++++++++++++++++++
 doc/CONTRIBUTING.md           |   8 +++
 2 files changed, 121 insertions(+)
 create mode 100644 .github/workflows/nightly.yml

diff --git a/.github/workflows/nightly.yml b/.github/workflows/nightly.yml
new file mode 100644
index 0000000..394804d
--- /dev/null
+++ b/.github/workflows/nightly.yml
@@ -0,0 +1,113 @@
+# Runs the chaos tests with many more random operations and fresh seeds,
+# and downloads a few small files from public survey archives.
+
+name: Nightly
+
+on:
+  schedule:
+  - cron: "23 6 * * *"
+  workflow_dispatch:
+    inputs:
+      steps:
+        description: Random operations per seed
+        default: "2000"
+      seeds:
+        description: Comma-separated seeds (blank derives them from the run)
+        default: ""
+  # Scheduled and manual runs only start from the default branch, so pull
+  # requests that change these tests or this workflow run it too.
+  pull_request:
+    paths:
+    - .github/workflows/nightly.yml
+    - test/chaos.py
+    - test/chaos_child.py
+    - test/http_server.py
+    - test/survey.py
+    - test/test_chaos_*.py
+    - test/test_survey*.py
+
+permissions:
+  contents: read
+
+concurrency:
+  group: nightly-${{ github.event.pull_request.number || github.ref }}
+  cancel-in-progress: true
+
+jobs:
+  chaos:
+    runs-on: ${{ matrix.os }}
+    timeout-minutes: 45
+    strategy:
+      fail-fast: false
+      matrix:
+        include:
+        - os: ubuntu-24.04
+          python-version: "3.9"
+          lane: 1
+        - os: ubuntu-24.04
+          python-version: "3.14"
+          lane: 2
+        - os: macos-latest
+          python-version: "3.14"
+          lane: 3
+    steps:
+    - uses: actions/checkout@v7
+      with:
+        persist-credentials: false
+    - uses: actions/setup-python@v7
+      with:
+        python-version: ${{ matrix.python-version }}
+    - name: Install the OpenSSH server
+      if: runner.os == 'Linux'
+      run: sudo apt-get update && sudo apt-get install -y --no-install-recommends openssh-server
+    - name: Install Hallmark and test tools
+      run: |
+        python -m pip install --upgrade pip
+        python -m pip install -e . --group dev
+    # Inputs reach the shell through the environment, never inline.
+    - name: Choose seeds and steps
+      env:
+        INPUT_SEEDS: ${{ inputs.seeds }}
+        INPUT_STEPS: ${{ inputs.steps }}
+        RUN_ID: ${{ github.run_id }}
+        LANE: ${{ matrix.lane }}
+        EVENT: ${{ github.event_name }}
+      run: |
+        seeds="${INPUT_SEEDS:-${RUN_ID}${LANE}1,${RUN_ID}${LANE}2,${RUN_ID}${LANE}3}"
+        if [ "$EVENT" = pull_request ]; then default_steps=300; else default_steps=2000; fi
+        steps="${INPUT_STEPS:-$default_steps}"
+        [[ "$seeds" =~ ^[0-9,]+$ && "$steps" =~ ^[0-9]+$ ]] || { echo "Invalid seeds or steps"; exit 1; }
+        echo "HALLMARK_CHAOS_SEED=$seeds" >> "$GITHUB_ENV"
+        echo "HALLMARK_CHAOS_STEPS=$steps" >> "$GITHUB_ENV"
+        echo "Chaos seeds: \`$seeds\`, steps per seed: $steps" >> "$GITHUB_STEP_SUMMARY"
+    - name: Run the chaos tests
+      env:
+        HALLMARK_RUN_SSH_TESTS: "1"
+      run: python -m pytest -m chaos -ra --timeout=1800 --basetemp="$RUNNER_TEMP/pytest"
+    - name: Keep failed repositories for inspection
+      if: failure()
+      uses: actions/upload-artifact@v7
+      with:
+        name: chaos-${{ runner.os }}-py${{ matrix.python-version }}
+        path: ${{ runner.temp }}/pytest
+        include-hidden-files: true
+        retention-days: 14
+
+  survey-live:
+    runs-on: ubuntu-24.04
+    timeout-minutes: 20
+    steps:
+    - uses: actions/checkout@v7
+      with:
+        persist-credentials: false
+    - uses: actions/setup-python@v7
+      with:
+        python-version: "3.13"
+    - name: Install Hallmark and test tools
+      run: |
+        python -m pip install --upgrade pip
+        python -m pip install -e . --group dev
+    - name: Download small files from public survey archives
+      env:
+        HALLMARK_RUN_LIVE_TESTS: "1"
+      run: python -m pytest -m survey_live -ra --timeout=600
diff --git a/doc/CONTRIBUTING.md b/doc/CONTRIBUTING.md
index f30a85f..e6f0f0b 100644
--- a/doc/CONTRIBUTING.md
+++ b/doc/CONTRIBUTING.md
@@ -30,6 +30,14 @@
 * If the implementation is as clear as (or clearer than) the goal, implement the solution first. Then write tests to ensure the solution covers edge cases and realistic user scenarios. The tests may reveal places where the implementation needs to be fixed, expanded, or simplified.
 * Run `pytest --cov` to see which lines the tests miss, or `pytest --cov --cov-report=html` for a browsable report in `htmlcov/`. CI runs the whole suite, including the loopback SSH tests, on every supported Python version, combines coverage from every job, and fails if the total drops below the floor set in `.github/workflows/ci.yml`. Codecov comments on each pull request with the coverage of its changed lines; those comments are informational.
 
+## Fault Injection and Survey Workflow Tests
+
+
+* `pytest -m survey` runs end-to-end workflows from `doc/usecase.rst` against scaled-down DESI-like and EHT-like releases on a loopback HTTP server, and over a loopback SSH server when `HALLMARK_RUN_SSH_TESTS=1`. These tests refuse any request that would leave the machine.
+* `pytest -m chaos` injects network faults, full disks, I/O errors and process kills, and runs seeded random operation sequences (the "monkey") against repository invariants. Set `HALLMARK_CHAOS_SEED` (comma-separated) and `HALLMARK_CHAOS_STEPS` to explore further; a failing monkey prints the command that reproduces it.
+* `HALLMARK_RUN_LIVE_TESTS=1 pytest -m survey_live` downloads a few megabytes from DESI DR1 and the EHT release on CyVerse. CI runs it, and the chaos tests with many more steps, every night.
+* When these tests expose a bug, reduce it to a small deterministic test marked `xfail(strict=True)` with the reason and issue, and fix it on its own branch. Strict expected failures fail CI once the bug is fixed, so the fix must remove the marker.
+
 ## Data Management
 
 

From ee2120f3ac18b87180efdd07a3e2c1998fd03af0 Mon Sep 17 00:00:00 2001
From: Dennis <denniswu501@gmail.com>
Date: Sun, 4 Oct 2026 21:11:56 -0700
Subject: [PATCH 15/16] Link expected failures to their issues

Name the issue tracking each bug in its strict expected-failure reason:
#54 truncated downloads, #55 state-file atomicity, #56 interrupted
checkouts, #57 interrupted worktree creation and #58 checkout before the
first commit.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 test/test_chaos_crash.py   | 6 +++---
 test/test_chaos_monkey.py  | 2 +-
 test/test_chaos_network.py | 2 +-
 test/test_chaos_storage.py | 4 ++--
 4 files changed, 7 insertions(+), 7 deletions(-)

diff --git a/test/test_chaos_crash.py b/test/test_chaos_crash.py
index 04d14dd..b98da14 100644
--- a/test/test_chaos_crash.py
+++ b/test/test_chaos_crash.py
@@ -118,10 +118,10 @@ def two_branches(telescope):
 @pytest.mark.parametrize("nth", [
     pytest.param(1, marks=pytest.mark.xfail(strict=True, reason=(
         "Bug: a checkout killed after moving a file to its backup folder "
-        "cannot be rerun; the file is reported missing"))),
+        "cannot be rerun; the file is reported missing (#56)"))),
     pytest.param(3, marks=pytest.mark.xfail(strict=True, reason=(
         "Bug: a checkout killed after moving a file to its backup folder "
-        "cannot be rerun; the file is reported missing"))),
+        "cannot be rerun; the file is reported missing (#56)"))),
 ])
 def test_killed_checkout_can_be_completed(two_branches, nth):
     repo, _ = two_branches
@@ -133,7 +133,7 @@ def test_killed_checkout_can_be_completed(two_branches, nth):
 
 @pytest.mark.xfail(strict=True, reason=(
     "Bug: add_worktree returns early when the destination already has .hm, "
-    "so a rerun after a crash leaves the worktree without its files"))
+    "so a rerun after a crash leaves the worktree without its files (#57)"))
 def test_killed_add_worktree_is_completed_by_rerun(telescope):
     repo, files = telescope
     repo.add(FMT)
diff --git a/test/test_chaos_monkey.py b/test/test_chaos_monkey.py
index 7275840..4351014 100644
--- a/test/test_chaos_monkey.py
+++ b/test/test_chaos_monkey.py
@@ -170,7 +170,7 @@ def test_local_monkey_keeps_repository_consistent(tmp_path, monkeypatch, request
 
 @pytest.mark.xfail(strict=True, reason=(
     "Bug: before the first commit, checkout raises GitCommandError because "
-    "init leaves data.tsv uncommitted"))
+    "init leaves data.tsv uncommitted (#58)"))
 def test_checkout_before_first_commit_reports_a_documented_error(tmp_path):
     repo = Repo.init(tmp_path / "obs")
     try:
diff --git a/test/test_chaos_network.py b/test/test_chaos_network.py
index dedf20c..a7e8e3e 100644
--- a/test/test_chaos_network.py
+++ b/test/test_chaos_network.py
@@ -112,7 +112,7 @@ def test_published_checksum_rejects_damaged_content(tmp_path, release, fault):
 
 @pytest.mark.xfail(strict=True, reason=(
     "Bug: a body truncated without Content-Length is accepted when the "
-    "catalog records the file size but no checksum"))
+    "catalog records the file size but no checksum (#54)"))
 def test_catalogued_size_rejects_unannounced_truncation(tmp_path, release):
     server = release(publish_checksums=False, listing_style="apache")
     repo = _catalog_repo(tmp_path / "repo", server)
diff --git a/test/test_chaos_storage.py b/test/test_chaos_storage.py
index ed468bb..473ecd5 100644
--- a/test/test_chaos_storage.py
+++ b/test/test_chaos_storage.py
@@ -30,10 +30,10 @@ def _in_dothm(target):
     1,
     pytest.param(2, marks=pytest.mark.xfail(strict=True, reason=(
         "Bug: Dothm.save_state writes config.yml, meta.yml and data.tsv "
-        "separately, so a failed later write leaves mixed generations"))),
+        "separately, so a failed later write leaves mixed generations (#55)"))),
     pytest.param(3, marks=pytest.mark.xfail(strict=True, reason=(
         "Bug: Dothm.save_state writes config.yml, meta.yml and data.tsv "
-        "separately, so a failed later write leaves mixed generations"))),
+        "separately, so a failed later write leaves mixed generations (#55)"))),
 ])
 def test_failed_state_write_keeps_one_generation(tmp_path, monkeypatch, nth):
     reference = Repo.init(tmp_path / "reference")

From 07120f0f6148c369b68a3f35a3905acc5eee7432 Mon Sep 17 00:00:00 2001
From: Dennis <denniswu501@gmail.com>
Date: Sun, 4 Oct 2026 21:15:36 -0700
Subject: [PATCH 16/16] Skip live survey tests when an archive times out during
 a scan

CyVerse listings can take longer than Hallmark's read timeout even after
the availability probe succeeds. Treat timeouts and server errors during
the catalog scan, as during downloads, as the archive being unavailable,
so the nightly job reports Hallmark failures rather than archive load.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 test/test_survey_live.py | 29 ++++++++++++++++++++++-------
 1 file changed, 22 insertions(+), 7 deletions(-)

diff --git a/test/test_survey_live.py b/test/test_survey_live.py
index 0e5f954..d8bba9b 100644
--- a/test/test_survey_live.py
+++ b/test/test_survey_live.py
@@ -15,6 +15,7 @@
 import requests
 
 from hallmark import Repo
+from hallmark.transport.base import DownloadError
 
 pytestmark = [
     pytest.mark.survey_live,
@@ -54,13 +55,27 @@ def check(url):
     return check
 
 
+def _archive_failure(error):
+    """Whether an error means the archive timed out or failed on its side."""
+    return any(sign in str(error)
+               for sign in ("ReadTimeout", "ConnectTimeout", "HTTP 5"))
+
+
+def _scan(repo, url, **options):
+    """Catalog ``url``, skipping when the archive times out or errors."""
+    try:
+        return repo.add(url, **options)
+    except DownloadError as exc:
+        if _archive_failure(exc):
+            pytest.skip(f"The archive failed during the scan: {exc}")
+        raise
+
+
 def _download(repo, plan):
     """Download a plan, skipping when the archive times out or errors."""
     result = repo.download(plan, approved=True)
     errors = " ".join(map(str, result["errors"]))
-    if result["failed"] and all(
-            "ReadTimeout" in str(error) or "ConnectTimeout" in str(error)
-            or "HTTP 5" in str(error) for error in result["errors"]):
+    if result["failed"] and all(map(_archive_failure, result["errors"])):
         pytest.skip(f"The archive failed during the download: {errors}")
     return result
 
@@ -84,8 +99,8 @@ def _approve_small(plan, expected_files, sizes_listed=True):
 def test_desi_dr1_files_match_published_checksums(tmp_path, archive_available):
     archive_available(DESI)
     repo = Repo.init(tmp_path / "desi")
-    repo.add(DESI, filter=["redrock-main-dark-23040.fits",
-                           "hpixexp-main-dark-23040.csv"])
+    _scan(repo, DESI, filter=["redrock-main-dark-23040.fits",
+                              "hpixexp-main-dark-23040.csv"])
     rows = repo.state.data.set_index("path")
     assert set(rows.checksum_algorithm) == {"sha256"}
     repo.commit("Record remote catalog")
@@ -104,7 +119,7 @@ def test_eht_m87_uvfits_downloads_from_cyverse(tmp_path, archive_available):
     archive_available(EHT)
     name = "SR1_M87_2017_095_lo_hops_netcal_StokesI.uvfits"
     repo = Repo.init(tmp_path / "eht")
-    repo.add(EHT, filter=name)
+    _scan(repo, EHT, filter=name)
     assert list(repo.state.data.path) == [name]
     repo.commit("Record remote catalog")
 
@@ -120,7 +135,7 @@ def test_eht_m87_uvfits_downloads_from_cyverse(tmp_path, archive_available):
 def test_eht_url_pattern_catalogs_every_day_and_band(tmp_path, archive_available):
     archive_available(EHT)
     repo = Repo.init(tmp_path / "eht")
-    repo.add(EHT + "SR1_M87_2017_{day:d}_{band}_hops_netcal_StokesI.uvfits")
+    _scan(repo, EHT + "SR1_M87_2017_{day:d}_{band}_hops_netcal_StokesI.uvfits")
     rows = repo.state.data
     assert set(rows.band) == {"lo", "hi"}
     assert {95, 96, 100, 101} <= {int(day) for day in rows.day}