Add verified reference lookup and immutable consumer receipts

Assistant: codex
Assistant-Model: gpt-6.1-sol
Assistant-Session: 01a1045b-933d-7aa1-b2db-104557487a79

tests/test_reference_consume.py

@@ -0,0 +1,176 @@
+"""Exercise qualified lookup, immutable event production and owner compatibility."""
+from copy import deepcopy
+import hashlib
+import json
+from pathlib import Path
+import shutil
+import subprocess
+import sys
+
+import pytest
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT / 'scripts'))
+import reference_consume as consume
+
+
+@pytest.fixture(scope='module')
+def published():
+    return consume.published()
+
+
+@pytest.fixture
+def decision():
+    return json.loads((ROOT / 'feedback/decision-input.example.json').read_text())
+
+
+def test_lookup_explains_filters_and_preserves_exact_read_revisions(published):
+    index, pin, manifest = published
+    result = consume.lookup(index, pin, manifest, ['CAND-0007'], ['recoverability'])
+    candidate, = result['candidates']
+    assert result['status'] == 'matches'
+    assert candidate['id'] == 'CAND-0007'
+    assert candidate['classification']['forces']['worsens']
+    assert candidate['gaps'] and candidate['maturity'] == 'hypothesis'
+    assert candidate['contract']['revision'] == manifest['corpus_revision']
+    assert f'/commit/{manifest["corpus_revision"]}/' in candidate['contract']['url']
+    assert all(e['locator']['revision'] == pin['packaging_revision'] for e in candidate['evidence'])
+    assert all(e['status'] == 'proposed' and e['condition'] and e['unknown'] for e in result['relations'])
+    assert all(v['revision'] == pin['packaging_revision'] for v in result['entrypoints'].values())
+    assert result['reference']['manifest_sha256'] == pin['manifest_sha256']
+    assert consume.lookup(index, pin, manifest, ['CAND-0007'], ['telemetry-quality'])['status'] == 'no-match'
+
+
+def test_missing_context_abstains_and_responsibility_filters_intersect(published):
+    index, pin, manifest = published
+    result = consume.lookup(index, pin, manifest)
+    assert result['status'] == 'needs-context' and result['candidates'] == result['relations'] == []
+    result = consume.lookup(index, pin, manifest, responsibilities=['mutation'], forces=['recoverability'])
+    assert result['candidates']
+    assert all('mutation' in c['classification']['responsibilities'] and
+               'recoverability' in c['classification']['forces']['improves'] for c in result['candidates'])
+
+
+@pytest.mark.parametrize('filters', [
+    {'candidates': ['CAND-0018']}, {'forces': ['magic']}, {'responsibilities': ['magic']},
+])
+def test_unknown_or_unpublished_identity_is_not_silently_selected(published, filters):
+    with pytest.raises(ValueError, match='Unknown'):
+        consume.lookup(*published, **filters)
+
+
+def test_live_research_is_independent_and_tampered_export_fails(tmp_path, published):
+    shutil.copytree(ROOT / 'reference', tmp_path / 'reference')
+    (tmp_path / '.git').symlink_to(ROOT / '.git', target_is_directory=True)
+    live = tmp_path / 'research/candidates/MutationOutcomeTracking.md'
+    live.parent.mkdir(parents=True)
+    live.write_text('Legitimate later research outside the frozen release.\n')
+    assert consume.published(tmp_path) == published
+    export = tmp_path / 'reference/0.1/index.json'
+    export.write_text('{}\n')
+    with pytest.raises(ValueError, match='Published export changed'):
+        consume.published(tmp_path)
+    with pytest.raises(ValueError, match='Unknown reference version'):
+        consume.published(tmp_path, '9.9')
+
+
+def test_receipt_keeps_unknown_outcome_and_pins_verified_reference(published, decision):
+    index, pin, _ = published
+    original = deepcopy(decision)
+    event = consume.receipt(decision, index, pin)
+    assert decision == original
+    assert event['kind'] == 'synthetic'
+    assert event['outcome']['value'] == 'unknown'
+    assert event['outcome']['observation'] is None
+    assert event['outcome']['measured_decision_minutes'] is None
+    assert event['reference']['packaging_revision'] == pin['packaging_revision']
+    assert 'origin' not in event
+
+
+@pytest.mark.parametrize('mutate', [
+    lambda d: d.update(kind=''),
+    lambda d: d.update(recorded_at='2999-01-01T00:00:00Z'),
+    lambda d: d.update(recorded_at='2026-01-01T00:00:00'),
+    lambda d: d.update(previous_event_sha256='invalid'),
+    lambda d: d.update(origin={'verified': True}),
+    lambda d: d['consumer'].update(revision='main'),
+    lambda d: d['consumer'].update(run_ref=''),
+    lambda d: d['context'].update(considered=['CAND-0018']),
+    lambda d: d['context'].update(selected=['CAND-0007']),
+    lambda d: d['context'].update(considered=['CAND-0007', 'CAND-0007']),
+    lambda d: d['decision'].update(action='applied'),
+    lambda d: d['outcome'].update(value='helpful'),
+    lambda d: d['outcome'].update(measured_decision_minutes=-1),
+    lambda d: d['outcome'].update(measured_rework_minutes=float('nan')),
+])
+def test_invalid_producer_claims_or_provenance_are_rejected(published, decision, mutate):
+    mutate(decision)
+    with pytest.raises(ValueError):
+        consume.receipt(decision, *published[:2])
+
+
+def test_followup_and_observed_result_retain_lineage(published, decision):
+    decision['previous_event_sha256'] = 'a' * 64
+    decision['event_id'] = 'fixture-outcome-02'
+    decision['outcome'].update(value='mixed', basis='observed-result', stage='test',
+                               observation='Disclosed fixture result; no consumer use.', evidence_refs=['fixture:test-01'])
+    event = consume.receipt(decision, *published[:2])
+    assert event['previous_event_sha256'] == 'a' * 64
+    assert event['outcome']['value'] == 'mixed'
+
+
+def test_immutable_write_returns_exact_collector_index_entry(tmp_path, published, decision):
+    event = consume.receipt(decision, *published[:2])
+    path = 'research/pattern-use/fixture/event-01.json'
+    entry = consume.write_receipt(event, path, tmp_path)
+    raw = (tmp_path / path).read_bytes()
+    assert json.loads(raw) == event
+    assert entry == {'path': path, 'sha256': hashlib.sha256(raw).hexdigest()}
+    with pytest.raises(FileExistsError):
+        consume.write_receipt(event, path, tmp_path)
+    assert (tmp_path / path).read_bytes() == raw
+
+
+@pytest.mark.parametrize('path', ['/tmp/event.json', '../event.json', 'x//event.json', 'x/./event.json', 'x\\event.json'])
+def test_unsafe_paths_are_rejected_before_writing(tmp_path, published, decision, path):
+    with pytest.raises(ValueError, match='Unsafe receipt path'):
+        consume.write_receipt(consume.receipt(decision, *published[:2]), path, tmp_path)
+    assert list(tmp_path.iterdir()) == []
+
+
+def test_symlink_escape_is_refused(tmp_path, published, decision):
+    (tmp_path / 'out').symlink_to(ROOT, target_is_directory=True)
+    with pytest.raises(ValueError, match='escapes consumer root'):
+        consume.write_receipt(consume.receipt(decision, *published[:2]), 'out/event.json', tmp_path)
+
+
+def test_cli_generates_owner_accepted_event_and_index(tmp_path):
+    result = subprocess.run([sys.executable, str(ROOT / 'scripts/reference_consume.py'), '--version', '0.1', 'receipt',
+                             '--input', str(ROOT / 'feedback/decision-input.example.json'), '--output', 'use/event.json'],
+                            cwd=tmp_path, capture_output=True, text=True, check=True)
+    entry = json.loads(result.stdout)
+    assert entry['sha256'] == hashlib.sha256((tmp_path / entry['path']).read_bytes()).hexdigest()
+    owner = ROOT.parent / 'activity-core'
+    python = owner / '.venv/bin/python'
+    if not python.exists():
+        pytest.skip('Actual activity-core contract check requires the sibling owner environment')
+    script = '''import json, sys
+sys.path.insert(0, sys.argv[1])
+from activity_core.context_resolvers.frontend_patterns import validate_event
+event = json.load(open(sys.argv[2]))
+validate_event(event, event['consumer']['repo'])
+'''
+    subprocess.run([str(python), '-B', '-c', script, str(owner / 'src'), str(tmp_path / entry['path'])],
+                   check=True, capture_output=True, text=True)
+
+
+def test_invalid_cli_decision_cannot_create_a_receipt(tmp_path, decision):
+    decision['decision']['action'] = 'applied'  # Selection remains empty.
+    source = tmp_path / 'decision.json'
+    source.write_text(json.dumps(decision))
+    result = subprocess.run([sys.executable, str(ROOT / 'scripts/reference_consume.py'), 'receipt',
+                             '--input', str(source), '--output', 'use/event.json'],
+                            cwd=tmp_path, capture_output=True, text=True)
+    assert result.returncode == 1
+    assert 'Action/selection mismatch' in result.stderr
+    assert not (tmp_path / 'use').exists()