Validate OpenBao snapshot evidence and record assurance closure gates

Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a06ecb-456a-71c2-b41e-0755d336e883

tests/test_recovery_evidence.py

@@ -41,3 +41,32 @@ def test_invalid_receipt_is_unavailable(tmp_path, change):
     (tmp_path / 'assurance').mkdir()
     (tmp_path / 'assurance/recovery-evidence.json').write_text(json.dumps(index))
     assert recovery_signals(NOW, tmp_path)['apps-pg.restore']['result'] == 'unavailable'
+
+
+def test_snapshot_is_stale_and_never_substitutes_for_restore():
+    signals = recovery_signals(NOW)
+    assert signals['openbao.snapshot']['observed_at'] == '2026-08-22T22:29:21Z'
+    assert 'openbao.restore' not in signals
+    contract = {'cluster_uid': 'test', 'capture_max_age_seconds': 900,
+        'signals': {'openbao.snapshot': {'owner': 'platform', 'max_age_seconds': 129600}}}
+    result = evaluate(contract, {'schema': 'railiance-platform.observation.v1',
+        'cluster_uid': 'test', 'captured_at': NOW.isoformat(),
+        'signals': {'openbao.snapshot': signals['openbao.snapshot']}}, NOW)
+    assert result['signals']['openbao.snapshot']['state'] == 'stale'
+
+
+@pytest.mark.parametrize('key,value', [('encrypted_copy_off_host', False),
+    ('hash_verified', False), ('cluster_id', 'other'), ('snapshot_sha256', 'invalid'),
+    ('created_at', '2027-01-01T00:00:00Z')])
+def test_snapshot_rejects_unverified_or_wrong_scope(tmp_path, key, value):
+    index = json.loads((ROOT / 'assurance/recovery-evidence.json').read_text())
+    entry = next(e for e in index['receipts'] if e['signal'] == 'openbao.snapshot')
+    receipt = json.loads((ROOT / entry['path']).read_text())
+    receipt[key] = value
+    path = tmp_path / entry['path']; path.parent.mkdir(parents=True)
+    path.write_text(json.dumps(receipt))
+    entry['sha256'] = hashlib.sha256(path.read_bytes()).hexdigest()
+    (tmp_path / 'assurance').mkdir()
+    index['receipts'] = [entry]
+    (tmp_path / 'assurance/recovery-evidence.json').write_text(json.dumps(index))
+    assert recovery_signals(NOW, tmp_path)['openbao.snapshot']['result'] == 'unavailable'

tests/test_service_assurance.py

@@ -148,7 +148,7 @@ class CollectorTests(unittest.TestCase):
         with patch.object(self.collector, 'query', side_effect=query), patch.object(self.collector, 'admission', return_value=baseline):
             observation = self.collector.capture()
         for name, sample in observation['signals'].items():
-            if name not in ('apps-pg.restore', 'forgejo-db.restore'):
+            if name not in ('apps-pg.restore', 'forgejo-db.restore', 'openbao.snapshot'):
                 self.assertEqual(sample['result'], 'unavailable')
         # Recorded recovery evidence is independent of failed live status reads.
         self.assertEqual(observation['signals']['apps-pg.restore']['result'], 'pass')