CCR-2026-0029/0030: activity-core ops_run worker token paths (proposed)

ACTIVITY-WP-0039-T02. Add read on the two exact worker-token paths to
workload-kv-read-activity-core-eso, one CCR per worker identity. Adds
scripts/openbao-policy-sync.sh, a silent attended policy writer that refuses
undeclared live drift and verifies on readback. Not applied; CCRs await
platform-operator and activity-core-owner approval.

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

Assistant: claude-code
Assistant-Model: opus
Assistant-Process: 150322@bnt-lap001
Assistant-Session: 16a7b788-374e-4915-a1df-fc87ffd9a5e4

tests/test_openbao_policy_sync.py

@@ -0,0 +1,65 @@
+import importlib.util
+import json
+from pathlib import Path
+
+import pytest
+
+spec = importlib.util.spec_from_file_location(
+    'sync', Path(__file__).resolve().parents[1] / 'scripts/openbao_policy_sync.py')
+m = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(m)
+
+NAME = 'workload-kv-read-activity-core-eso'
+DECLARED = (m.POLICY_DIR / (NAME + '.hcl')).read_text()
+
+
+def store(initial):
+    live = {'rules': initial}
+    writes = []
+
+    def write(name, path):
+        writes.append(name)
+        live['rules'] = Path(path).read_text()
+    return live, writes, (lambda name: live['rules']), write
+
+
+def test_applies_when_live_is_the_expected_prior_version():
+    live, writes, read, write = store('path "old" {}')
+    result = m.sync(NAME, m.digest('path "old" {}'), read, write)
+    assert result['status'] == 'applied' and writes == [NAME]
+    assert m.digest(live['rules']) == m.digest(DECLARED)
+
+
+def test_already_current_writes_nothing():
+    _, writes, read, write = store(DECLARED)
+    assert m.sync(NAME, 'unused', read, write)['status'] == 'already_current'
+    assert writes == []
+
+
+def test_refuses_undeclared_drift():
+    _, writes, read, write = store('path "someone-else" {}')
+    with pytest.raises(m.Refused, match='live_policy_drifted'):
+        m.sync(NAME, m.digest('path "old" {}'), read, write)
+    assert writes == []
+
+
+def test_readback_mismatch_is_refused():
+    live = {'rules': 'path "old" {}'}
+    with pytest.raises(m.Refused, match='readback_mismatch'):
+        m.sync(NAME, m.digest('path "old" {}'), lambda n: live['rules'], lambda n, p: None)
+
+
+def test_new_paths_are_exact_and_read_only():
+    for worker in ('rein-aharness-railiance01', 'rein-aharness-metered-railiance01'):
+        block = 'path "platform/data/workloads/activity-core/ops-run-workers/%s" {\n  capabilities = ["read"]\n}' % worker
+        assert block in DECLARED
+    assert 'ops-run-workers/*' not in DECLARED and '"list"' not in DECLARED
+
+
+def test_refuses_outside_envelope_and_bad_args(tmp_path, monkeypatch):
+    monkeypatch.setenv('HOME', str(tmp_path))
+    receipt = tmp_path / 'r.json'
+    assert m.main(['--policy', NAME, '--expect-live-sha256', 'x', '--receipt', str(receipt)]) == 1
+    assert json.loads(receipt.read_text())['status'] == 'attended_envelope_required'
+    with pytest.raises(SystemExit):
+        m.parse(['--policy', '../etc', '--expect-live-sha256', 'x', '--receipt', 'r'])