Build reproducible capability advisor and scheduled source capture

Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a09c77-b1a9-7eb3-b501-6cb24b458c97

tests/test_advisor.py

@@ -0,0 +1,95 @@
+import copy
+import json
+from pathlib import Path
+import tempfile
+import unittest
+
+from intelligence_radar.advisor import advise, outcome, recommend, replay
+
+REPO = Path(__file__).resolve().parents[1]
+
+
+class AdvisorTests(unittest.TestCase):
+    def setUp(self):
+        self.temp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.temp.cleanup)
+        self.root = Path(self.temp.name)
+        self.task = json.loads((REPO/'examples/task.json').read_text())
+        self.catalog = json.loads((REPO/'examples/advisor-catalog.json').read_text())
+        self.policy = json.loads((REPO/'policies/advisor-v0.2.json').read_text())
+
+    def report(self):
+        return advise(self.root, self.task, self.catalog, self.policy, '2026-09-13', True)
+
+    def test_rank_cost_and_no_unearned_sufficiency(self):
+        r = self.report()
+        self.assertEqual(r['frontier'], 'frontier-fixture')
+        self.assertEqual(r['efficient'], 'efficient-fixture')
+        self.assertIsNone(r['budget'])
+        self.assertEqual(len(r['selected_benchmarks']), 5)
+        self.assertAlmostEqual(sum(b['weight'] for b in r['selected_benchmarks']), 1)
+        efficient = next(x for x in r['profiles'] if x['id']=='efficient-fixture')
+        self.assertAlmostEqual(efficient['cost']['scenario_usd'], .07)
+
+    def test_missing_score_never_imputed(self):
+        self.catalog['results'].pop(0)
+        report = self.report()
+        self.assertEqual(report['frontier'], 'efficient-fixture')
+        row = next(x for x in report['profiles'] if x['id']=='frontier-fixture')
+        self.assertIsNone(row['fitness_proxy'])
+        self.assertEqual(len(row['missing_measurements']), 1)
+
+    def test_constraints_remove_highest_scoring_profile(self):
+        self.catalog['profiles'][0]['attributes']['requester_access'] = False
+        self.assertEqual(self.report()['frontier'], 'efficient-fixture')
+
+    def test_missing_or_conflicting_price_prevents_cheapest_claim(self):
+        self.catalog['prices'].pop(0)
+        self.assertIsNone(self.report()['efficient'])
+        self.catalog['prices'].append(copy.deepcopy(self.catalog['prices'][0]))
+        self.assertIsNone(self.report()['efficient'])
+
+    def test_stale_future_and_wrong_version_results(self):
+        for field, value in [('observed_at','2026-09-14T00:00:00Z'),
+                             ('observed_at','2025-01-01T00:00:00Z'),
+                             ('benchmark_version','other')]:
+            with self.subTest(field=field,value=value):
+                original=self.catalog['results'][0].copy()
+                self.catalog['results'][0][field]=value
+                self.assertEqual(self.report()['frontier'], 'efficient-fixture')
+                self.catalog['results'][0]=original
+
+    def test_aggregate_and_short_selection_withhold_ranking(self):
+        self.catalog['benchmarks'][0]['aggregate']=True
+        self.assertIsNone(self.report()['frontier'])
+
+    def test_calibration_duplicate_protection_and_frozen_replay(self):
+        ident, initial = recommend(self.root,self.task,self.catalog,self.policy,'2026-09-13',True)
+        value=dict(recommendation_id=ident,profile_id='budget-fixture',run_id='run-0',
+                   accepted=True,independent_verification=True,verifier_evidence='synthetic test run',
+                   actual_usd=.01,runtime_seconds=1,completed_at='2026-09-13T22:00:00Z')
+        first=outcome(self.root,value)
+        self.assertEqual(outcome(self.root,value),first)
+        with self.assertRaises(ValueError):
+            outcome(self.root,dict(value,accepted=False))
+        for i in range(1,20):
+            outcome(self.root,dict(value,run_id=f'run-{i}'))
+        self.assertEqual(self.report()['budget'],'budget-fixture')
+        self.assertEqual(replay(self.root,ident), initial)
+        self.task['acceptance']+=' Different acceptance test.'
+        self.assertIsNone(self.report()['budget'])
+
+    def test_synthetic_requires_explicit_opt_in(self):
+        with self.assertRaises(ValueError):
+            advise(self.root,self.task,self.catalog,self.policy,'2026-09-13')
+
+    def test_invalid_numeric_inputs_rejected(self):
+        self.catalog['results'][0]['raw_score']=float('nan')
+        with self.assertRaises(ValueError):
+            self.report()
+
+    def test_replay_is_independent_of_json_object_key_order(self):
+        self.task['constraints']['unavailable_attribute'] = True
+        self.task['capabilities']['coding'] = .31
+        ident, initial = recommend(self.root,self.task,self.catalog,self.policy,'2026-09-13',True)
+        self.assertEqual(replay(self.root,ident), initial)