diff --git a/deploy/crypto-agility-consumer@.service b/deploy/crypto-agility-consumer@.service
new file mode 100644
index 0000000..364b93f
--- /dev/null
+++ b/deploy/crypto-agility-consumer@.service
@@ -0,0 +1,36 @@
+# Installation template only. Provision the named existing service account,
+# owner-private config/bundles and verified dependencies before enabling.
+[Unit]
+Description=Crypto-agility authenticated evidence consumer (%i)
+After=local-fs.target
+StartLimitIntervalSec=300
+StartLimitBurst=5
+
+[Service]
+Type=simple
+User=%i
+Group=%i
+WorkingDirectory=/opt/crypto-agility
+ExecStart=/opt/crypto-agility/.venv/bin/python -m tools.consumer --config /etc/crypto-agility/%i/config.json --bundles /var/lib/crypto-agility-bundles/%i --state-directory /var/lib/crypto-agility-consumer/%i --interval 60
+StateDirectory=crypto-agility-consumer/%i
+StateDirectoryMode=0700
+UMask=0077
+Restart=on-failure
+RestartSec=10
+NoNewPrivileges=true
+PrivateTmp=true
+PrivateDevices=true
+ProtectSystem=strict
+ProtectHome=true
+ProtectKernelTunables=true
+ProtectKernelModules=true
+ProtectControlGroups=true
+RestrictSUIDSGID=true
+RestrictAddressFamilies=AF_UNIX
+IPAddressDeny=any
+ReadOnlyPaths=/etc/crypto-agility/%i /var/lib/crypto-agility-bundles/%i
+StandardOutput=journal
+StandardError=journal
+
+[Install]
+WantedBy=multi-user.target
diff --git a/docs/consumer-deployment.md b/docs/consumer-deployment.md
new file mode 100644
index 0000000..4162e96
--- /dev/null
+++ b/docs/consumer-deployment.md
@@ -0,0 +1,61 @@
+# Persistent authenticated consumer
+
+`python -m tools.consumer` wraps the existing monitor, authenticated ingestion, replay store and durable pending-alert outbox. It does not sign evidence, establish trust, send alerts or infer a delivery destination. Installation is prepared; production activation requires an operator-approved issuer/key status, subject/audience policy and alert destination with tested delivery and acknowledgment. A successfully running process or an output line is not proof that an observation was accepted or an alert delivered.
+
+## Inputs and publication
+
+Provision separate absolute paths for the configuration, bundle root and persistent state directory. Every input directory must be owned by the consumer account and mode 0700; input files must be owner-only regular files (0600), with exactly one link. No symlink path components, selectors, bundle directories or files are accepted. Use resolved physical paths on systems with conventional symlink ancestors. The consumer never creates input directories, alters their permissions, downloads keys or creates an issuer identity. The publisher and privileged operators are trusted to manage these private directories; another process running as that same account can modify them.
+
+A bundle contains exactly `manifest.json`, `observation.jwt` and `transcript.txt`. The manifest must explicitly use version 2 with `subject.kind` equal to `origin` and `subject.id` equal to the configured canonical HTTPS origin. A qScan repository manifest must not be relabeled as origin evidence. Device compatibility remains unknown or declared unless representative measurements exist. The JWT must bind the SHA-256 digest of the exact manifest bytes. Reading a local committed bundle does not prove that a public HTTPS endpoint currently serves those bytes; independently compare the deployed response digest during publication acceptance.
+
+Compute the bundle name as lowercase SHA-256 of UTF-8 compact JSON mapping each of the three filenames to its lowercase content SHA-256, sorted by filename, without trailing newline. For example, the hash input has the form `{"manifest.json":"<64hex>","observation.jwt":"<64hex>","transcript.txt":"<64hex>"}`. This is an immutable content commitment, not a new signature format or ledger chain identifier. The issuer transaction should retain the exact files and commitment for delivery recovery.
+
+Write files to a new private staging directory, fsync every file, rename the complete directory to its 64-hex commitment under the bundle root, then fsync the root. Do not overwrite an existing committed directory. Write the commitment plus one optional newline to a private temporary selector, fsync it, atomically rename it to `current`, and fsync the root again. Retain the previously selected directory until no reader needs it. Never publish files individually into a selected directory. The consumer reads the selector once, opens that directory without following links, reads bounded file bytes and verifies their commitment before authentication; a selector change cannot mix transactions. Modified content fails closed.
+
+## Operator configuration
+
+`config.json` contains exactly these fields. Replace placeholders with independently approved values. The example intentionally cannot activate a consumer:
+
+```json
+{
+  "version": 1,
+  "issuer": "<approved issuer>",
+  "subject": "https://example.org",
+  "audience": "<approved consumer audience>",
+  "keyId": "<approved key id>",
+  "keyStatus": {
+    "status": "unknown",
+    "checkedAt": "<UTC RFC3339 timestamp ending Z>",
+    "validUntil": "<UTC RFC3339 timestamp ending Z>",
+    "source": "<operator decision or authoritative status reference>"
+  },
+  "trust": {
+    "issuers": {
+      "<approved issuer>": {
+        "keys": {"<approved key id>": "<pinned public key in existing trust format>"},
+        "subjects": ["https://example.org"],
+        "audiences": ["<approved consumer audience>"],
+        "purposes": ["pqc-observation"],
+        "maxAgeSeconds": 3600,
+        "maxLifetimeSeconds": 3600
+      }
+    }
+  }
+}
+```
+
+The status assertion must explicitly be `active`, have a nonempty source, and cover the current time with a validity window no longer than 24 hours. This is an operator assertion, **not cryptographic proof of current nonrevocation**. A missing revocation endpoint or an old root-signed certificate does not establish this assertion. Do not fill it automatically. The configuration permits exactly one issuer/key and the exact configured subject, audience and observation purpose; the existing verifier still validates algorithm, public-key encoding, signature, claim lifetime and policy. Configuration is reread every cycle, including when a previously authenticated measurement is cached. Replace the configuration atomically; absent, malformed, stale or inactive status queues a durable failure while preserving the last baseline.
+
+## State, supervision and alert delivery
+
+Before the first run, explicitly bootstrap a new empty private state directory with `python -m tools.consumer --initialize-state --state-directory /absolute/private/state`. This refuses existing state or lock files. Interrupted bootstrap requires reviewed recovery, not an automatic retry that erases files. The supervisor requires the initialization marker and all persistent documents on every check; missing files never recreate a replay store or baseline.
+
+The state directory contains `initialized.json`, `replay.json`, `state.json`, `clock.json` and their stable locks, plus `consumer.lock` for the singleton CLI. Keep the entire directory across deployments and restarts. Never delete replay or clock state to make a rejected token pass. Restore only through an explicitly reviewed recovery procedure; restoring an old state backup can undo replay protection. The persistent clock high-water mark rejects rollback even after an input failure.
+
+The long-running process retains the existing bounded in-memory authenticated-input cache and reevaluates expiration every cycle. After restart, a previously consumed token is rejected as replay; provision a new fresh observation and JTI. An unchanged token cannot silently become current assurance again. Historical trusted measurements and pending alerts survive, allowing later valid classical evidence to reveal a hybrid regression even after rejected inputs.
+
+Reports go to stdout/journald; they are not alert-delivery receipts. The durable `state.json` outbox retains each occurrence's `deliveryId` until a separately configured delivery component confirms success. After confirmed delivery, acknowledge only that exact occurrence with `python -m tools.monitor --state /absolute/state/state.json --ack-alert <deliveryId>`. Never acknowledge merely because a report was printed or read. No delivery adapter or destination is selected here. Input failures continue producing reports and durable pending alerts; failures that prevent persistence explicitly include `deliveryPersistenceFailure`. State-directory startup failures or stdout errors stop the CLI so the supervisor can restart it. The CLI does not exit on an assessment finding, avoiding replay-producing restart loops.
+
+`deploy/crypto-agility-consumer@.service` is an installation template, not discovered production coordinates. Review its paths and existing service account, install a pinned checkout and virtual environment at the chosen location, provision approved configuration/bundles, and retain private writable state. Install dependencies from `requirements.txt` with the supported Python version; ML-DSA verification additionally needs the compatible OpenSSL runtime described in the attestation documentation. The template grants no network access and exposes no signing credentials. Validate the unit on the actual Linux host with `systemd-analyze verify` before installing; no enable/start command is authorized by the mere presence of this template.
+
+Before activation, run the zero-skip acceptance suite, authenticate a fresh exact-bound bundle, exercise a selector switch, restart/replay rejection, expiry, missing configuration, durable failure recovery and confirmed alert delivery at the operator's destination. Verify actual service ownership, filesystem permissions and persistence across a service restart. Hosted CI must cover the final committed tree; local fixture tests alone do not establish production success.
diff --git a/tests/test_consumer.py b/tests/test_consumer.py
new file mode 100644
index 0000000..83ef412
--- /dev/null
+++ b/tests/test_consumer.py
@@ -0,0 +1,264 @@
+"""Deployment boundary tests using the independent Node JWS fixture issuer."""
+import copy
+import datetime as dt
+import hashlib
+import json
+import os
+from pathlib import Path
+import subprocess
+import tempfile
+import unittest
+from urllib.parse import urlsplit
+from unittest.mock import patch
+
+ROOT = Path(__file__).resolve().parents[1]
+NOW = dt.datetime(2026, 10, 12, tzinfo=dt.timezone.utc)
+SUBJECT = 'https://example.org'
+TRANSCRIPT = b'CONNECTION ESTABLISHED\nProtocol version: TLSv1.3\nCiphersuite: TLS_AES_256_GCM_SHA384\nSignature type: RSA-PSS\nVerification: OK\nNegotiated TLS1.3 group: X25519MLKEM768\nDONE\n'
+
+
+def issue(request):
+    result = subprocess.run(['node', str(ROOT/'tests/fixtures/node_jws_issuer.mjs')], input=json.dumps(request), text=True, capture_output=True, check=True, timeout=15)
+    return json.loads(result.stdout)
+
+
+def put(path, raw):
+    path.write_bytes(raw)
+    path.chmod(0o600)
+
+
+class ConsumerTests(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.tmp.cleanup)
+        self.root = Path(self.tmp.name).resolve()
+        self.bundles, self.state = self.root/'bundles', self.root/'state'
+        self.bundles.mkdir(mode=0o700)
+        self.state.mkdir(mode=0o700)
+        from tools.consumer import initialize_state
+        initialize_state(self.state, NOW)
+        self.config_path = self.root/'config.json'
+        self.keys = issue({'operation': 'sign', 'header': {'alg': 'EdDSA', 'kid': 'probe', 'typ': 'pqc-observation+jwt'}, 'claims': {}})
+        self.config = {'version': 1, 'issuer': 'observatory', 'subject': SUBJECT, 'audience': 'monitor', 'keyId': 'probe',
+                       'keyStatus': {'status': 'active', 'checkedAt': NOW.isoformat().replace("+00:00", "Z"), 'validUntil': (NOW+dt.timedelta(hours=2)).isoformat().replace("+00:00", "Z"), 'source': 'fixture operator decision'},
+                       'trust': {'issuers': {'observatory': {'keys': {'probe': self.keys['publicJwk']['x']}, 'subjects': [SUBJECT], 'audiences': ['monitor'], 'purposes': ['pqc-observation']}}}}
+        self.save_config()
+        self.select('first')
+
+    def save_config(self):
+        put(self.config_path, json.dumps(self.config).encode())
+
+    def select(self, jti, group='X25519MLKEM768', subject=SUBJECT):
+        manifest = json.loads((ROOT/'examples/crypto-agility-v2.json').read_bytes())
+        manifest['subject'] = {'kind': 'origin', 'id': subject}
+        raw = json.dumps(manifest).encode()
+        transcript = TRANSCRIPT.replace(b'X25519MLKEM768', group.encode())
+        claims = {'iss': 'observatory', 'sub': subject, 'aud': 'monitor', 'jti': jti, 'purpose': 'pqc-observation', 'iat': int(NOW.timestamp()), 'exp': int(NOW.timestamp())+3600,
+                  'snapshotDigest': 'sha256:'+hashlib.sha256(raw).hexdigest(),
+                  'observation': {'profile': 'openssl-tls13-v1', 'observedAt': NOW.isoformat().replace("+00:00", "Z"), 'endpoint': {'host': urlsplit(subject).hostname, 'port': urlsplit(subject).port or 443, 'address': '192.0.2.1'},
+                                  'probe': {'name': 'openssl', 'version': 'OpenSSL 3.5.0', 'groups': [group], 'certificateValidation': 'pkix-hostname', 'exitCode': 0}, 'transcriptSha256': hashlib.sha256(transcript).hexdigest()}}
+        token = issue({'operation': 'sign', 'privateJwk': self.keys['privateJwk'], 'header': {'alg': 'EdDSA', 'kid': 'probe', 'typ': 'pqc-observation+jwt'}, 'claims': claims})['token']
+        values = {'manifest.json': raw, 'observation.jwt': token.encode(), 'transcript.txt': transcript}
+        commitment = json.dumps({k: hashlib.sha256(v).hexdigest() for k, v in values.items()}, sort_keys=True, separators=(',', ':')).encode()
+        identity = hashlib.sha256(commitment).hexdigest()
+        directory = self.bundles/identity
+        directory.mkdir(mode=0o700)
+        for name, value in values.items():
+            put(directory/name, value)
+        put(self.bundles/'next', (identity+'\n').encode())
+        os.replace(self.bundles/'next', self.bundles/'current')
+        self.selected = directory
+
+    def consumer(self):
+        from tools.consumer import Consumer
+        return Consumer(self.config_path, self.bundles, self.state)
+
+    def test_signed_bundle_cache_restart_replay_and_expiry(self):
+        consumer = self.consumer()
+        report = consumer.check(NOW)
+        self.assertEqual(report['measurements'][0]['status'], 'authenticated-observation', report)
+        self.assertEqual(consumer.check(NOW)['measurements'][0]['status'], 'authenticated-observation')
+        replay = self.consumer().check(NOW)
+        self.assertIn({'code': 'measurement-rejected', 'ref': 'replay'}, replay['assessment']['findings'])
+        expired = consumer.check(NOW+dt.timedelta(seconds=3601))
+        self.assertIn('measurement-expired', [f['code'] for f in expired['assessment']['findings']])
+
+    def test_switch_regression_and_failure_outbox_survive_repair(self):
+        consumer = self.consumer()
+        consumer.check(NOW)
+        self.config_path.unlink()
+        failed = consumer.check(NOW)
+        delivery = [a['deliveryId'] for a in failed['changes']['alerts'] if a['type'] == 'monitor-failure'][0]
+        self.save_config()
+        self.select('classical', 'X25519')
+        report = consumer.check(NOW)
+        self.assertIn('tls-regression', [a['type'] for a in report['changes']['alerts']])
+        self.assertIn(delivery, [a['deliveryId'] for a in report['changes']['alerts']])
+        self.assertEqual((self.state/'state.json').stat().st_mode & 0o777, 0o600)
+
+    def test_content_mutation_and_selector_traversal_fail_before_replay(self):
+        put(self.selected/'transcript.txt', TRANSCRIPT+b'changed')
+        self.assertEqual(self.consumer().check(NOW)['assessment']['status'], 'invalid')
+        self.assertEqual(json.loads((self.state/'replay.json').read_bytes())['entries'], {})
+        put(self.bundles/'current', b'../config.json')
+        self.assertEqual(self.consumer().check(NOW)['assessment']['status'], 'invalid')
+
+    def test_symlink_hardlink_and_public_files_rejected(self):
+        target = self.selected/'manifest.json'
+        original = target.read_bytes()
+        target.unlink()
+        external = self.root/'external'
+        put(external, original)
+        for kind in ('symlink', 'hardlink', 'public'):
+            with self.subTest(kind=kind):
+                if kind == 'symlink': target.symlink_to(external)
+                elif kind == 'hardlink': os.link(external, target)
+                else:
+                    put(target, original)
+                    target.chmod(0o644)
+                self.assertEqual(self.consumer().check(NOW)['assessment']['status'], 'invalid')
+                target.unlink()
+        self.assertEqual(json.loads((self.state/'replay.json').read_bytes())['entries'], {})
+
+    def test_operator_status_required_bounded_and_rechecked_after_cache(self):
+        consumer = self.consumer()
+        self.assertTrue(consumer.check(NOW)['measurements'])
+        original = copy.deepcopy(self.config)
+        for change in ({'status': 'unknown'}, {'validUntil': NOW.isoformat().replace("+00:00", "Z")}, {'checkedAt': (NOW+dt.timedelta(seconds=1)).isoformat().replace("+00:00", "Z")}, {'validUntil': (NOW+dt.timedelta(days=2)).isoformat().replace("+00:00", "Z")}, {'source': ''}):
+            with self.subTest(change=change):
+                self.config = copy.deepcopy(original)
+                self.config['keyStatus'].update(change)
+                self.save_config()
+                self.assertEqual(consumer.check(NOW)['assessment']['status'], 'invalid')
+
+    def test_subject_and_trust_scope_are_not_inferred_from_token(self):
+        self.select('other-subject', subject='https://other.example')
+        self.assertEqual(self.consumer().check(NOW)['assessment']['status'], 'invalid')
+        self.assertEqual(json.loads((self.state/'replay.json').read_bytes())['entries'], {})
+        self.config['trust']['issuers']['observatory']['subjects'].append('https://other.example')
+        self.save_config()
+        self.assertEqual(self.consumer().check(NOW)['assessment']['status'], 'invalid')
+
+    def test_selector_switch_during_read_never_mixes_transactions(self):
+        from tools import consumer as implementation
+        old = self.selected.name
+        self.select('second', 'X25519')
+        new = self.selected.name
+        put(self.bundles/'current', (old+'\n').encode())
+        read = implementation._read
+        def switch(directory, name, limit):
+            raw = read(directory, name, limit)
+            if name == 'manifest.json':
+                put(self.bundles/'next', (new+'\n').encode())
+                os.replace(self.bundles/'next', self.bundles/'current')
+            return raw
+        consumer = self.consumer()
+        with patch.object(implementation, '_read', side_effect=switch):
+            first = consumer.check(NOW)
+        self.assertEqual(first['measurements'][0]['negotiatedGroup'], 'X25519MLKEM768', first)
+        second = consumer.check(NOW)
+        self.assertIn('tls-regression', [a['type'] for a in second['changes']['alerts']])
+
+    def test_clock_rollback_rejected_across_restart_even_after_input_failure(self):
+        self.config_path.unlink()
+        self.consumer().check(NOW+dt.timedelta(seconds=5))
+        self.save_config()
+        report = self.consumer().check(NOW)
+        self.assertEqual(report['assessment']['status'], 'invalid')
+        self.assertIn('clock rollback', report['assessment']['findings'][0]['message'])
+        self.assertEqual(json.loads((self.state/'replay.json').read_bytes())['entries'], {})
+
+    def test_symlink_ancestor_rejected_and_unsafe_state_reports_nondurability(self):
+        alias = self.root/'alias'
+        alias.symlink_to(self.bundles, target_is_directory=True)
+        from tools.consumer import Consumer
+        self.assertEqual(Consumer(self.config_path, alias, self.state).check(NOW)['assessment']['status'], 'invalid')
+        self.state.chmod(0o755)
+        report = self.consumer().check(NOW)
+        self.assertIn('deliveryPersistenceFailure', report['changes'])
+
+    def test_unknown_repository_manifest_never_becomes_origin(self):
+        # A correctly committed repository declaration is still the wrong subject kind.
+        raw = (self.selected/'manifest.json').read_bytes()
+        raw = raw.replace(b'"kind": "origin"', b'"kind": "repository"')
+        values = {n: (raw if n == 'manifest.json' else (self.selected/n).read_bytes()) for n in ('manifest.json', 'observation.jwt', 'transcript.txt')}
+        identity = hashlib.sha256(json.dumps({k: hashlib.sha256(v).hexdigest() for k, v in values.items()}, sort_keys=True, separators=(',', ':')).encode()).hexdigest()
+        directory = self.bundles/identity
+        directory.mkdir(mode=0o700)
+        for name, value in values.items(): put(directory/name, value)
+        put(self.bundles/'current', (identity+'\n').encode())
+        report = self.consumer().check(NOW)
+        self.assertEqual(report['assessment']['status'], 'invalid')
+        self.assertIn('explicitly declare', report['assessment']['findings'][0]['message'])
+
+    def test_missing_initialized_state_never_resets_replay_protection(self):
+        self.consumer().check(NOW)
+        for name in ('replay.json', 'state.json', 'clock.json', 'initialized.json'):
+            with self.subTest(name=name):
+                path = self.state/name
+                raw = path.read_bytes()
+                path.unlink()
+                report = self.consumer().check(NOW)
+                self.assertEqual(report['assessment']['status'], 'invalid')
+                self.assertFalse(path.exists(), 'lost state must not be recreated')
+                put(path, raw)
+        from tools.consumer import initialize_state
+        with self.assertRaises(ValueError): initialize_state(self.state, NOW)
+
+    def test_malformed_clock_cannot_clear_high_water_mark(self):
+        for value in ({}, {'lastNow': True}, {'lastNow': float('nan')}):
+            with self.subTest(value=value):
+                raw = json.dumps(value).encode()
+                put(self.state/'clock.json', raw)
+                report = self.consumer().check(NOW)
+                self.assertEqual(report['assessment']['status'], 'invalid')
+                self.assertEqual((self.state/'clock.json').read_bytes(), raw)
+
+    def test_cli_continues_reporting_missing_configuration_without_acknowledging(self):
+        import sys
+        self.config_path.unlink()
+        result = subprocess.run([sys.executable, '-m', 'tools.consumer', '--config', str(self.config_path), '--bundles', str(self.bundles), '--state-directory', str(self.state), '--iterations', '2', '--interval', '1'], cwd=ROOT, capture_output=True, text=True, timeout=15)
+        self.assertEqual(result.returncode, 0, result.stderr)
+        reports = [json.loads(line) for line in result.stdout.splitlines()]
+        self.assertEqual(len(reports), 2)
+        self.assertTrue(all(r['assessment']['status'] == 'invalid' for r in reports))
+        self.assertEqual(reports[0]['changes']['alerts'][0]['deliveryId'], reports[1]['changes']['alerts'][0]['deliveryId'])
+
+    def test_state_loss_after_precheck_never_recreates_replay_or_baseline(self):
+        from tools import consumer as implementation
+        self.consumer().check(NOW)
+        read_bundle = implementation._bundle
+        for name in ('replay.json', 'state.json'):
+            with self.subTest(name=name):
+                path = self.state/name
+                raw = path.read_bytes()
+                def lose_file(root):
+                    values = read_bundle(root)
+                    path.unlink()
+                    return values
+                with patch.object(implementation, '_bundle', side_effect=lose_file):
+                    report = self.consumer().check(NOW)
+                self.assertNotEqual(report.get('measurements', [{}])[0].get('status') if report.get('measurements') else None, 'authenticated-observation')
+                self.assertFalse(path.exists(), 'transaction must not recreate missing initialized state')
+                put(path, raw)
+
+    def test_explicit_nondefault_tls_port_is_valid_for_local_bundle(self):
+        subject = 'https://example.org:8443'
+        self.config['subject'] = subject
+        self.config['trust']['issuers']['observatory']['subjects'] = [subject]
+        self.save_config()
+        self.select('nondefault-port', subject=subject)
+        report = self.consumer().check(NOW)
+        self.assertEqual(report.get('measurements', [{}])[0].get('status'), 'authenticated-observation', report)
+
+    def test_noncanonical_origin_policy_rejected(self):
+        for subject in ('https://example.org/', 'https://example.org:443', 'https://Example.org', 'https://user@example.org', 'https://example.org?q=1', 'https://example.org#fragment'):
+            with self.subTest(subject=subject):
+                self.config['subject'] = subject
+                self.config['trust']['issuers']['observatory']['subjects'] = [subject]
+                self.save_config()
+                self.assertEqual(self.consumer().check(NOW)['assessment']['status'], 'invalid')
+
+
+if __name__ == '__main__':
+    unittest.main()
diff --git a/tools/attestation.py b/tools/attestation.py
index 92d16da..330bb43 100644
--- a/tools/attestation.py
+++ b/tools/attestation.py
@@ -74,14 +74,15 @@ class ReplayStore:
     Fresh IDs do not permit an older issuance time to replace a newer statement
     for the same issuer/subject/audience/purpose. A clock rollback fails closed.
     """
-    def __init__(self, path, max_entries=10000):
+    def __init__(self, path, max_entries=10000, *, require_existing=False):
         self.path = Path(path)
         self.max_entries = max_entries
+        self.require_existing = require_existing
 
     def consume(self, claims, now):
         identity = hashlib.sha256(canonical([claims[k] for k in ("iss", "aud", "jti")])).hexdigest()
         scope = hashlib.sha256(canonical([claims[k] for k in ("iss", "sub", "aud", "purpose")])).hexdigest()
-        with locked_document(self.path) as (previous, save):
+        with locked_document(self.path, require_existing=self.require_existing) as (previous, save):
             state = previous if previous is not None else {"version": 1, "lastNow": now, "entries": {}, "latest": {}}
             if (state.get("version") != 1 or not isinstance(state.get("entries"), dict)
                     or not isinstance(state.get("latest"), dict)):
diff --git a/tools/consumer.py b/tools/consumer.py
new file mode 100644
index 0000000..37668e3
--- /dev/null
+++ b/tools/consumer.py
@@ -0,0 +1,240 @@
+"""Owner-provisioned immutable-bundle supervisor; no signing or alert transport."""
+import argparse
+from contextlib import contextmanager
+import datetime as dt
+import fcntl
+import hashlib
+import json
+import ipaddress
+import math
+import os
+from pathlib import Path
+import re
+import stat
+import time
+
+from tools.attestation import ReplayStore
+from urllib.parse import urlsplit
+from tools.monitor import run_once
+from tools.state import canonical, enqueue_failure, fingerprint, locked_document
+from tools.verifier import load_manifest, parse_time
+
+ROOT = Path(__file__).resolve().parents[1]
+FILES = {'manifest.json': 1048576, 'observation.jwt': 1048576, 'transcript.txt': 65536}
+
+
+def _private(info, directory=False):
+    kind = stat.S_ISDIR if directory else stat.S_ISREG
+    if (not kind(info.st_mode) or info.st_uid != os.geteuid() or info.st_mode & 0o077
+            or (not directory and info.st_nlink != 1)):
+        raise ValueError('owner-only regular file or directory required')
+
+
+@contextmanager
+def _directory(path):
+    # Traverse every component with no-follow; Path.resolve would hide symlinks.
+    path = Path(path)
+    if not path.is_absolute() or '..' in path.parts:
+        raise ValueError('absolute paths without traversal required')
+    fd = os.open('/', os.O_RDONLY | os.O_DIRECTORY)
+    try:
+        for part in path.parts[1:]:
+            child = os.open(part, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, dir_fd=fd)
+            os.close(fd)
+            fd = child
+        _private(os.fstat(fd), directory=True)
+        yield fd
+    finally:
+        os.close(fd)
+
+
+def _read(directory, name, limit):
+    fd = os.open(name, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK, dir_fd=directory)
+    with os.fdopen(fd, 'rb') as source:
+        _private(os.fstat(fd))
+        raw = source.read(limit + 1)
+        if len(raw) > limit:
+            raise ValueError('input size limit')
+        return raw
+
+
+def _origin(value):
+    """Local identity only; this performs no discovery and allows explicit TLS ports."""
+    if not value.isascii() or any(ord(c) <= 32 or ord(c) == 127 for c in value):
+        raise ValueError('invalid canonical HTTPS origin')
+    parts = urlsplit(value)
+    if (parts.scheme != 'https' or not parts.hostname or parts.username is not None or parts.password is not None
+            or parts.path or parts.query or parts.fragment or '?' in value or '#' in value):
+        raise ValueError('canonical HTTPS origin required')
+    host = parts.hostname
+    if ':' in host:
+        host = '[' + ipaddress.IPv6Address(host).compressed + ']'
+    elif not re.fullmatch(r'[a-z0-9](?:[a-z0-9.-]*[a-z0-9])?', host):
+        raise ValueError('invalid origin hostname')
+    port = parts.port
+    if port == 0:
+        raise ValueError('invalid origin port')
+    canonical_origin = 'https://' + host + (':' + str(port) if port not in (None, 443) else '')
+    if value != canonical_origin:
+        raise ValueError('canonical HTTPS origin required')
+    return value
+
+
+def _config(path, now):
+    with _directory(path.parent) as parent:
+        config = load_manifest(_read(parent, path.name, 1048576))
+    if set(config) != {'version', 'issuer', 'subject', 'audience', 'keyId', 'keyStatus', 'trust'} or type(config['version']) is not int or config['version'] != 1:
+        raise ValueError('explicit consumer configuration required')
+    for key in ('issuer', 'subject', 'audience', 'keyId'):
+        if not isinstance(config[key], str) or not 1 <= len(config[key]) <= 2048:
+            raise ValueError('invalid consumer identity')
+    if _origin(config['subject']) != config['subject']:
+        raise ValueError('canonical origin subject required')
+    status = config['keyStatus']
+    if not isinstance(status, dict) or set(status) != {'status', 'checkedAt', 'validUntil', 'source'}:
+        raise ValueError('explicit operator key status required')
+    checked, until = parse_time(status['checkedAt']), parse_time(status['validUntil'])
+    if (status['status'] != 'active' or not checked <= now < until or (until-checked).total_seconds() > 86400
+            or not isinstance(status['source'], str) or not 1 <= len(status['source'].strip()) <= 2048):
+        raise ValueError('operator key status absent, inactive, future, expired or unbounded')
+    trust = config['trust']
+    if not isinstance(trust, dict) or set(trust) != {'issuers'} or set(trust['issuers']) != {config['issuer']}:
+        raise ValueError('exact issuer trust required')
+    policy = trust['issuers'][config['issuer']]
+    if (not isinstance(policy, dict) or set(policy.get('keys', {})) != {config['keyId']}
+            or policy.get('subjects') != [config['subject']] or policy.get('audiences') != [config['audience']]
+            or policy.get('purposes') != ['pqc-observation'] or policy.get('disabled', False) is not False
+            or config['keyId'] in policy.get('revokedKeys', [])):
+        raise ValueError('exact enabled key and consumer scopes required')
+    return config
+
+
+def _bundle(root):
+    with _directory(root) as directory:
+        identity = _read(directory, 'current', 65).decode('ascii').removesuffix('\n')
+        if not re.fullmatch('[a-f0-9]{64}', identity):
+            raise ValueError('invalid bundle selector')
+        fd = os.open(identity, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, dir_fd=directory)
+        try:
+            _private(os.fstat(fd), directory=True)
+            if set(os.listdir(fd)) != set(FILES):
+                raise ValueError('complete bundle requires exactly three files')
+            values = {name: _read(fd, name, limit) for name, limit in FILES.items()}
+        finally:
+            os.close(fd)
+    commitment = canonical({name: hashlib.sha256(raw).hexdigest() for name, raw in values.items()})
+    if hashlib.sha256(commitment).hexdigest() != identity:
+        raise ValueError('bundle content commitment mismatch')
+    return values
+
+
+def initialize_state(directory, now=None):
+    """Explicit first bootstrap only; missing existing state requires recovery."""
+    directory = Path(directory)
+    now = now or dt.datetime.now(dt.timezone.utc)
+    with _directory(directory) as fd:
+        with locked_document(directory/'initialized.json') as (previous, finish):
+            if previous is not None or set(os.listdir(fd)) != {'initialized.json.lock'}:
+                raise ValueError('initialization requires an empty private directory; never reset existing state')
+            documents = {'clock.json': {'lastNow': now.timestamp()},
+                         'state.json': {'noBaseline': True, 'pendingAlerts': []},
+                         'replay.json': {'version': 1, 'lastNow': now.timestamp(), 'entries': {}, 'latest': {}}}
+            for name, value in documents.items():
+                with locked_document(directory/name) as (_, save):
+                    save(value)
+            finish({'version': 1})
+
+
+def _check_state(directory):
+    with _directory(directory) as fd:
+        marker = load_manifest(_read(fd, 'initialized.json', 128))
+        if marker != {'version': 1} or type(marker['version']) is not int:
+            raise ValueError('invalid initialization marker')
+        # Missing or unsafe files cannot trigger implicit baseline/replay reset.
+        for name in ('state.json', 'replay.json', 'clock.json'):
+            value = load_manifest(_read(fd, name, 4 * 1024 * 1024))
+            if name == 'clock.json' and (set(value) != {'lastNow'} or type(value['lastNow']) not in (int, float)
+                                         or not math.isfinite(value['lastNow'])):
+                raise ValueError('invalid consumer clock state')
+
+
+class Consumer:
+    def __init__(self, config, bundles, state_directory):
+        self.config_path, self.bundles, self.directory = map(Path, (config, bundles, state_directory))
+        self.schema = json.loads((ROOT/'schema/crypto-agility-v2.schema.json').read_bytes())
+        self.cache = {}
+
+    def check(self, now=None):
+        now = now or dt.datetime.now(dt.timezone.utc)
+        state_safe = False
+        try:
+            _check_state(self.directory)
+            state_safe = True
+            with locked_document(self.directory/'clock.json', require_existing=True) as (clock, save):
+                if now.tzinfo is None or (clock and now.timestamp() < clock['lastNow']):
+                    raise ValueError('consumer clock rollback')
+                save({'lastNow': now.timestamp()})
+            config = _config(self.config_path, now)
+            values = _bundle(self.bundles)
+            manifest = load_manifest(values['manifest.json'])
+            if manifest.get('subject', {}).get('kind') != 'origin' or manifest['subject'].get('id') != config['subject']:
+                raise ValueError('bundle must explicitly declare the configured origin subject')
+            return run_once(None, self.schema, self.directory/'state.json', now,
+                            evidence={'audience': config['audience'], 'replay_store': ReplayStore(self.directory/'replay.json', require_existing=True)},
+                            measurement_cache=self.cache, require_existing_state=True,
+                            input_bytes={'manifest': values['manifest.json'], 'token': values['observation.jwt'],
+                                         'transcript': values['transcript.txt'], 'trust': canonical(config['trust'])})
+        except (OSError, ValueError, TypeError, KeyError, AttributeError, RecursionError) as exc:
+            finding = {'code': 'consumer-input-or-state-failure', 'message': str(exc)}
+            alert = {'type': 'monitor-failure', 'severity': 'error', 'action': 'Repair consumer input or state; retain the previous baseline and pending alerts.', 'id': fingerprint(finding)}
+            changes = {'alerts': [alert], 'resolved': []}
+            try:
+                if not state_safe:
+                    raise ValueError('consumer state directory is unsafe or unavailable')
+                changes['alerts'] = enqueue_failure(self.directory/'state.json', alert, require_existing=True)
+            except (OSError, ValueError, TypeError, KeyError, RecursionError) as error:
+                changes['deliveryPersistenceFailure'] = str(error)
+            return {'subject': None, 'checkedAt': now.isoformat(), 'assessment': {'status': 'invalid', 'findings': [finding]}, 'pairs': [], 'changes': changes}
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument('--config', type=Path)
+    parser.add_argument('--bundles', type=Path)
+    parser.add_argument('--state-directory', type=Path, required=True)
+    parser.add_argument('--initialize-state', action='store_true', help='explicit first bootstrap; refuses existing state')
+    parser.add_argument('--interval', type=int, default=60)
+    parser.add_argument('--iterations', type=int, default=0, help='0 runs continuously')
+    args = parser.parse_args()
+    if not 1 <= args.interval <= 86400 or args.iterations < 0:
+        parser.error('invalid interval or iterations')
+    os.umask(0o077)
+    if args.initialize_state:
+        if args.config or args.bundles:
+            parser.error('initialize state separately from consumer inputs')
+        initialize_state(args.state_directory)
+        return 0
+    if not args.config or not args.bundles:
+        parser.error('config and bundles are required')
+    consumer = Consumer(args.config, args.bundles, args.state_directory)
+    # Separate singleton lock never unlinked/replaced. State/replay retain their own
+    # transaction locks so delivery acknowledgments remain independent.
+    with _directory(args.state_directory) as directory:
+        try:
+            fd = os.open('consumer.lock', os.O_RDWR | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600, dir_fd=directory)
+        except FileExistsError:
+            fd = os.open('consumer.lock', os.O_RDWR | os.O_NOFOLLOW | os.O_NONBLOCK, dir_fd=directory)
+    with os.fdopen(fd, 'r+b') as lock:
+        _private(os.fstat(lock.fileno()))
+        fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
+        count = 0
+        while args.iterations == 0 or count < args.iterations:
+            print(json.dumps(consumer.check(), separators=(',', ':')), flush=True)
+            count += 1
+            if args.iterations == 0 or count < args.iterations:
+                time.sleep(args.interval)
+    return 0
+
+
+if __name__ == '__main__':
+    raise SystemExit(main())
diff --git a/tools/monitor.py b/tools/monitor.py
index eb2f1e0..eeaed53 100644
--- a/tools/monitor.py
+++ b/tools/monitor.py
@@ -76,28 +76,39 @@ def _bounded_read(path, limit=1048576):
 
 
 def run_once(manifest_path, schema, state_path=None, now=None, *, evidence=None,
-             measurement_cache=None, discovery=None, origin=None, expected_subject=None):
+             measurement_cache=None, discovery=None, origin=None, expected_subject=None, input_bytes=None, require_existing_state=False):
     """Read/refresh, authenticate, assess and atomically record one check.
 
     evidence contains token/transcript/trust paths, audience and ReplayStore.
     A process-local cache reuses only identical authenticated input bytes and
     policy. Expiry is reevaluated each cycle; a restart requires a fresh JWT.
+    input_bytes optionally supplies bounded raw manifest/token/transcript/trust
+    bytes from a coherent deployment bundle; it never supplies verified results.
     """
     now = now or dt.datetime.now(dt.timezone.utc)
     try:
+        def read_input(name, path, limit=1048576):
+            if input_bytes is None:
+                return _bounded_read(path, limit)
+            value = input_bytes[name]
+            if not isinstance(value, bytes) or len(value) > limit:
+                raise ValueError("invalid bounded raw input")
+            return value
+        if input_bytes is not None and discovery is not None:
+            raise ValueError("raw inputs and discovery are mutually exclusive")
         if discovery is not None:
             fetched = discovery.fetch(origin, expected_subject, now=now)
             manifest, raw = fetched["manifest"], fetched["body"]
         else:
-            raw = _bounded_read(manifest_path)
+            raw = read_input("manifest", manifest_path)
             manifest = load_manifest(raw)
         report = snapshot(manifest, schema, now)
         measurements = []
         rejection = None
         if evidence and report["assessment"]["status"] != "invalid":
-            token = _bounded_read(evidence["token"]).decode("ascii").strip()
-            transcript = _bounded_read(evidence["transcript"], 65536)
-            trust_raw = _bounded_read(evidence["trust"])
+            token = read_input("token", evidence.get("token")).decode("ascii").strip()
+            transcript = read_input("transcript", evidence.get("transcript"), 65536)
+            trust_raw = read_input("trust", evidence.get("trust"))
             trust = load_manifest(trust_raw)
             digest = "sha256:" + hashlib.sha256(raw).hexdigest()
             cache_key = fingerprint([token, hashlib.sha256(transcript).hexdigest(),
@@ -129,7 +140,7 @@ def run_once(manifest_path, schema, state_path=None, now=None, *, evidence=None,
                 report["assessment"]["findings"].append({"code": "measurement-expired", "ref": report["subject"]})
                 report["assessment"]["status"] = "needs-review"
         if state_path and report["assessment"]["status"] != "invalid":
-            report["changes"] = update_state(state_path, report)
+            report["changes"] = update_state(state_path, report, require_existing=require_existing_state)
     except (OSError, ValueError, TypeError, KeyError, RecursionError) as exc:
         finding = {"code": "monitor-input-or-state-failure", "message": str(exc)}
         report = {"subject": None, "checkedAt": now.isoformat(),
@@ -141,7 +152,7 @@ def run_once(manifest_path, schema, state_path=None, now=None, *, evidence=None,
         report["changes"] = {"alerts": [alert], "resolved": []}
         if state_path:
             try:
-                report["changes"]["alerts"] = enqueue_failure(state_path, alert)
+                report["changes"]["alerts"] = enqueue_failure(state_path, alert, require_existing=require_existing_state)
             except (OSError, ValueError, TypeError, KeyError, RecursionError) as exc:
                 # State failure cannot promise durability. Keep an explicit output
                 # signal for the supervisor rather than silently losing the failure.
diff --git a/tools/state.py b/tools/state.py
index c3f8b53..21a624d 100644
--- a/tools/state.py
+++ b/tools/state.py
@@ -35,7 +35,7 @@ def _owned_regular(fd):
 
 
 @contextlib.contextmanager
-def locked_document(path):
+def locked_document(path, *, require_existing=False):
     """Yield (previous, save), holding a separate lock across read/replace/fsync."""
     path = Path(path).absolute()
     directory = os.open(path.parent, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW)
@@ -57,6 +57,8 @@ def locked_document(path):
         try:
             fd = os.open(path.name, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK, dir_fd=directory)
         except FileNotFoundError:
+            if require_existing:
+                raise ValueError("initialized state file is missing")
             previous = None
         else:
             with os.fdopen(fd, "rb") as source:
@@ -152,12 +154,12 @@ def _time(value):
     return parsed
 
 
-def update_state(path, report):
+def update_state(path, report, *, require_existing=False):
     if not isinstance(report.get("subject"), str) or not report["subject"]:
         raise ValueError("state subject required")
     if report.get("assessment", {}).get("status") == "invalid":
         raise ValueError("invalid assessments cannot replace monitoring state")
-    with locked_document(path) as (previous, save):
+    with locked_document(path, require_existing=require_existing) as (previous, save):
         if previous is not None and not previous.get("noBaseline", False):
             if previous.get("subject") != report["subject"]:
                 raise ValueError("state subject mismatch")
@@ -203,9 +205,9 @@ def acknowledge_alerts(path, delivery_ids):
         return len(pending) - len(remaining)
 
 
-def enqueue_failure(path, alert):
+def enqueue_failure(path, alert, *, require_existing=False):
     """Durably queue an input failure without advancing a successful baseline."""
-    with locked_document(path) as (previous, save):
+    with locked_document(path, require_existing=require_existing) as (previous, save):
         state = previous if previous is not None else {"noBaseline": True}
         pending = {a["id"]: a for a in state.get("pendingAlerts", [])}
         if alert["id"] not in pending: