NK-WP-0035-T04: independent receiver readback helper for the retained hourly capture

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Assistant: claude-code
Assistant-Model: opus
Assistant-Process: 696317@bnt-lap001
Assistant-Session: bab9e511-0915-45b3-926a-2fab552edff6

tools/emission-cadence-profile/receiver_readback.py

@@ -0,0 +1,114 @@
+#!/usr/bin/env python3
+"""Independent receiver readback for a retained cadence capture (NK-WP-0035-T04).
+
+Reads the capture's `capture-provenance.json` (its `receiver_readback` request:
+source, tenant, stream instance, window and event IDs) and `events.jsonl`, then
+asks audit-core directly, under the operator's own reader token:
+
+- `GET /v1/reconciliation` for per-class counts in the same window and instance;
+- `GET /v1/events/<id>` for every captured event ID.
+
+The output is a value-safe receipt with the counts from both sides, a presence
+check per event ID and a verdict (`agree` or `diverge`). It keeps no event
+payloads, and the token never enters the receipt. Agreement proves neither
+completeness nor that King's Guard accepts the stream (KG-WP-0005-T03).
+
+Reach the ClusterIP receiver with your own port-forward, for example
+`kubectl -n audit-core port-forward svc/audit-core 18080:8080`.
+"""
+import argparse
+import collections
+import json
+import os
+import stat
+import sys
+import urllib.error
+import urllib.parse
+import urllib.request
+from pathlib import Path
+
+SCHEMA = 'nk-wp-0035-receiver-readback/v1'
+
+
+def load_capture(directory):
+    provenance = json.loads((directory / 'capture-provenance.json').read_text())
+    request = provenance['receiver_readback']
+    events = [json.loads(line) for line in (directory / 'events.jsonl').read_text().splitlines() if line.strip()]
+    if any(e.get('stream_instance_id') != request['stream_instance_id'] for e in events):
+        raise ValueError('capture events span another stream instance')
+    return request, collections.Counter(e['event_class'] for e in events)
+
+
+def read_token(path):
+    info = path.lstat()
+    if not stat.S_ISREG(info.st_mode) or info.st_mode & 0o077:
+        raise ValueError('token file must be a regular file readable only by its owner (0600)')
+    token = path.read_text().strip()
+    if not token or any(c.isspace() for c in token):
+        raise ValueError('token file must hold exactly one bearer token')
+    return token
+
+
+def readback(request, captured, get):
+    """get(path_with_query) -> (status, json). Returns the value-safe receipt."""
+    query = urllib.parse.urlencode({'source': request['source'], 'tenant': request['tenant'],
+                                    'since': request['since'], 'until': request['until'],
+                                    'stream_instance_id': request['stream_instance_id']})
+    status, body = get('/v1/reconciliation?' + query)
+    findings = []
+    if status != 200:
+        findings.append(f'reconciliation returned {status}: {body.get("error", "")}')
+        receiver = {}
+    else:
+        receiver = {row['class']: row['count'] for row in body.get('counts', [])}
+        for cls in sorted(set(receiver) | set(captured)):
+            if receiver.get(cls, 0) != captured.get(cls, 0):
+                findings.append(f'class {cls}: captured {captured.get(cls, 0)}, receiver {receiver.get(cls, 0)}')
+    presence = {}
+    for event_id in request['event_ids']:
+        status, body = get('/v1/events/' + urllib.parse.quote(event_id, safe=''))
+        presence[event_id] = status == 200
+        if status != 200:
+            findings.append(f'event {event_id[:12]}… returned {status}')
+        elif (body.get('source'), body.get('tenant')) not in ((request['source'], request['tenant']), (None, None)):
+            findings.append(f'event {event_id[:12]}… source/tenant differs')
+    return {'schema': SCHEMA, 'request': request, 'captured_counts': dict(captured),
+            'receiver_counts': receiver, 'event_presence': presence,
+            'verdict': 'agree' if not findings else 'diverge', 'findings': findings,
+            'payloads_retained': False, 'token_retained': False,
+            'means': 'agreement proves neither completeness nor King\'s Guard acceptance'}
+
+
+def http_get(base, token):
+    def get(path):
+        req = urllib.request.Request(base.rstrip('/') + path, headers={'Authorization': 'Bearer ' + token})
+        try:
+            with urllib.request.urlopen(req, timeout=15) as response:
+                return response.status, json.loads(response.read() or b'{}')
+        except urllib.error.HTTPError as exc:
+            try:
+                return exc.code, json.loads(exc.read() or b'{}')
+            except ValueError:
+                return exc.code, {}
+    return get
+
+
+def main(argv=None):
+    p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
+    p.add_argument('--capture-dir', type=Path, required=True)
+    p.add_argument('--base-url', default='http://127.0.0.1:18080')
+    p.add_argument('--token-file', type=Path, required=True)
+    p.add_argument('--output', type=Path, required=True)
+    args = p.parse_args(argv)
+    request, captured = load_capture(args.capture_dir)
+    receipt = readback(request, captured, http_get(args.base_url, read_token(args.token_file)))
+    fd = os.open(args.output, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o644)
+    with os.fdopen(fd, 'w') as out:
+        json.dump(receipt, out, indent=2, sort_keys=True)
+        out.write('\n')
+    print(f"{receipt['verdict']}: {len(receipt['findings'])} finding(s); receipt {args.output}")
+    return 0 if receipt['verdict'] == 'agree' else 1
+
+
+if __name__ == '__main__':
+    sys.exit(main())

tools/emission-cadence-profile/tests/test_receiver_readback.py

@@ -0,0 +1,92 @@
+"""Synthetic tests for the NK-WP-0035-T04 receiver readback helper; no network, no tokens."""
+import importlib.util
+import json
+import os
+from pathlib import Path
+
+import pytest
+
+SOURCE = Path(__file__).parents[1] / 'receiver_readback.py'
+spec = importlib.util.spec_from_file_location('receiver_readback', SOURCE)
+rb = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(rb)
+
+CAPTURE = Path(__file__).parents[3] / 'docs/evidence/2026-10-15-cadence-adoption/hourly-observation-capture'
+
+
+def fake_receiver(counts, present, source='qonto-assistant', tenant='tenant:friendly:binky', recon_status=200):
+    calls = []
+
+    def get(path):
+        calls.append(path)
+        if path.startswith('/v1/reconciliation?'):
+            if recon_status != 200:
+                return recon_status, {'error': 'source_not_allowed'}
+            return 200, {'counts': [{'class': k, 'count': v} for k, v in counts.items()]}
+        event_id = path.rsplit('/', 1)[1]
+        if event_id in present:
+            return 200, {'source': source, 'tenant': tenant, 'id': event_id}
+        return 404, {'error': 'not_found'}
+    return get, calls
+
+
+def test_retained_capture_loads():
+    request, captured = rb.load_capture(CAPTURE)
+    assert len(request['event_ids']) == 3
+    assert sum(captured.values()) == 3
+
+
+def test_agreement():
+    request, captured = rb.load_capture(CAPTURE)
+    get, calls = fake_receiver(dict(captured), set(request['event_ids']))
+    receipt = rb.readback(request, captured, get)
+    assert receipt['verdict'] == 'agree' and receipt['findings'] == []
+    assert 'stream_instance_id=' in calls[0] and len(calls) == 4
+    assert receipt['token_retained'] is False and receipt['payloads_retained'] is False
+
+
+@pytest.mark.parametrize('case', ['missing_event', 'count_differs', 'extra_class', 'forbidden', 'wrong_tenant'])
+def test_divergence(case):
+    request, captured = rb.load_capture(CAPTURE)
+    counts, present = dict(captured), set(request['event_ids'])
+    kwargs = {}
+    if case == 'missing_event':
+        present.discard(request['event_ids'][0])
+    elif case == 'count_differs':
+        cls = next(iter(counts))
+        counts[cls] += 1
+    elif case == 'extra_class':
+        counts['audit.allow'] = 1
+    elif case == 'forbidden':
+        kwargs['recon_status'] = 403
+    else:
+        kwargs['tenant'] = 'tenant:other'
+    get, _ = fake_receiver(counts, present, **kwargs)
+    receipt = rb.readback(request, captured, get)
+    assert receipt['verdict'] == 'diverge' and receipt['findings']
+
+
+def test_token_file_must_be_private(tmp_path):
+    token = tmp_path / 'token'
+    token.write_text('abc.def.ghi\n')
+    os.chmod(token, 0o644)
+    with pytest.raises(ValueError):
+        rb.read_token(token)
+    os.chmod(token, 0o600)
+    assert rb.read_token(token) == 'abc.def.ghi'
+    token.write_text('two tokens\n')
+    with pytest.raises(ValueError):
+        rb.read_token(token)
+
+
+def test_receipt_never_contains_token(tmp_path, monkeypatch):
+    token = tmp_path / 'token'
+    token.write_text('SENTINEL-TOKEN-VALUE')
+    os.chmod(token, 0o600)
+    request, captured = rb.load_capture(CAPTURE)
+    get, _ = fake_receiver(dict(captured), set(request['event_ids']))
+    monkeypatch.setattr(rb, 'http_get', lambda base, tok: get)
+    out = tmp_path / 'receipt.json'
+    assert rb.main(['--capture-dir', str(CAPTURE), '--token-file', str(token), '--output', str(out)]) == 0
+    assert 'SENTINEL' not in out.read_text()
+    assert json.loads(out.read_text())['verdict'] == 'agree'

workplans/NK-WP-0035-emission-cadence-security-profile.md

@@ -431,6 +431,17 @@ The next step needs a human: an independent receiver readback under a fresh
 attended operator OIDC login. King's Guard then compares counts, sequence,
 coverage and lag. This capture is not acceptance, and the profile stays proposed.
 
+2026-10-23: the readback is now one attended command. The helper
+`tools/emission-cadence-profile/receiver_readback.py` (6 synthetic tests) reads
+the retained capture's `receiver_readback` request. It asks audit-core
+`/v1/reconciliation` (same source, tenant, window and instance) and
+`/v1/events/<id>` for each of the three IDs, under the operator's own 0600 token
+file. It writes a value-safe `agree`/`diverge` receipt without payloads or the
+token. A class-name mapping difference between source and receiver shows up as
+`diverge`; it is never hidden. The operator runs it through their own
+port-forward. The receipt then goes to King's Guard (KG-WP-0005-T03) and
+qonto-assistant (QONTO-WP-0006-T03).
+
 ## Automate staged receiver derivation without sibling mutations
 
 ```task