fix(regression): finalize all callers and classify HTTP evidence safely

tools/regression_coverage.py

@@ -148,12 +148,15 @@ def route_inventory(catalog, root=None):
 
 def http_observations(paths, inventory):
     """Map Hypercorn access records to declared routes without retaining request URLs."""
-    patterns = []
+    patterns, unsupported = [], []
     for route in inventory['routes']:
         method, path = route['route'].split(' ', 1)
         parts = re.split(r'(<[^>]+>)', path)
+        if any(part.startswith('<') and ':' in part and not part.startswith(('<int:', '<path:')) for part in parts):
+            unsupported.append(route['route'])
+            continue
         pattern = ''.join(
-            ('.+' if part.startswith('<path:') else '[^/]+')
+            ('.+' if part.startswith('<path:') else r'\d+' if part.startswith('<int:') else '[^/]+')
             if part.startswith('<') and part.endswith('>')
             else re.escape(part)
             for part in parts
@@ -166,13 +169,18 @@ def http_observations(paths, inventory):
     for path in paths:
         try:
             with Path(path).open() as stream:
-                for line in stream:
+                for line_number, line in enumerate(stream, 1):
                     match = access.search(line)
                     if not match:
                         continue
                     method, target, status = match.groups()
                     records += 1
-                    request_path = urlsplit(target).path
+                    try:
+                        request_path = urlsplit(target).path
+                    except ValueError:
+                        errors.append(f'{path}:{line_number}: invalid request target')
+                        unclassified[f'{method} {status}'] += 1
+                        continue
                     route = next(
                         (
                             name
@@ -194,6 +202,7 @@ def http_observations(paths, inventory):
         'records': records,
         'routes': [{'route': route, 'statuses': dict(statuses)} for route, statuses in sorted(matched.items())],
         'unclassified': dict(unclassified),
+        'unsupported_templates': unsupported,
         'evidence_errors': errors,
     }
 

tools/regression_job.py

@@ -403,8 +403,7 @@ class Job:
             self.state['phase'] = 'traffic'
             self.save()
             summary = suite.run()
-            summary['passed'] = not summary['errors']
-            suite.write('summary.json', summary)
+            suite.finish(summary)
             result.update(completed=True, suite=summary, errors=list(summary['errors']))
             self.state['phase'] = 'finished'
             self.save()

tools/regression_suite.py

@@ -436,6 +436,15 @@ class Suite:
         path.chmod(0o600)
         return str(path)
 
+    def finish(self, result):
+        if self.profile.get('coverage_suite'):
+            from regression_coverage import write_report
+
+            coverage = write_report([self.output], self.output)
+            result['errors'].extend(coverage['evidence_errors'])
+        result['passed'] = not result['errors']
+        self.write('summary.json', result)
+
     def call(self, args, body=None, *, defer_bot_readback=False):
         command = [*self.cli, *args]
         if body is not None:
@@ -781,13 +790,7 @@ def main():
                         result['errors'].extend(suite.cleanup())
                     except (RuntimeError, ValueError, OSError, KeyError) as exc:
                         result['errors'].append(f'cleanup failed: {exc}')
-            if profile.get('coverage_suite'):
-                from regression_coverage import write_report
-
-                coverage = write_report([suite.output], suite.output)
-                result['errors'].extend(coverage['evidence_errors'])
-            result['passed'] = not result['errors']
-            suite.write('summary.json', result)
+            suite.finish(result)
         print(json.dumps(result, ensure_ascii=False, indent=2))
         if not result['passed']:
             raise SystemExit(1)

tools/test_regression_coverage.py

@@ -191,6 +191,46 @@ class CoverageTest(unittest.TestCase):
                 self.assertNotIn(value, encoded)
             self.assertTrue(http_observations([path.with_name('missing')], inventory)['evidence_errors'])
 
+    def test_http_converter_and_invalid_url_do_not_create_false_hits_or_leaks(self):
+        inventory = {'routes': [{'route': 'GET /keys/<int:id>'}]}
+        with tempfile.TemporaryDirectory() as directory:
+            path = Path(directory) / 'access.log'
+            path.write_text(
+                '\n'.join(
+                    [
+                        'GET /keys/not-an-int 1.1 404',
+                        'GET http://[private-tenant]/?token=secret 1.1 400',
+                        'GET /keys/42 1.1 200',
+                    ]
+                )
+            )
+            result = http_observations([path], inventory)
+            self.assertEqual(result['routes'], [{'route': 'GET /keys/<int:id>', 'statuses': {'200': 1}}])
+            self.assertEqual(result['unclassified'], {'GET 404': 1, 'GET 400': 1})
+            self.assertEqual(len(result['evidence_errors']), 1)
+            self.assertNotIn('private-tenant', json.dumps(result))
+            self.assertNotIn('secret', json.dumps(result))
+
+    def test_suite_finalization_rejects_missing_evidence_for_direct_callers(self):
+        from regression_suite import Suite
+
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            profile = {
+                'version': 1,
+                'coverage_suite': 'smoke',
+                'smoke': {},
+                'cli': {'config': 'unused'},
+                'cases': [{'name': 'chat'}],
+            }
+            suite = Suite(profile, root / 'input.json', root)
+            suite.write('profile.json', profile)
+            result = {'errors': []}
+            suite.finish(result)
+            self.assertFalse(result['passed'])
+            self.assertTrue((root / 'coverage.json').exists())
+            self.assertFalse(json.loads((root / 'summary.json').read_text())['passed'])
+
     def test_new_routes_are_discovered_without_adding_test_cases(self):
         with tempfile.TemporaryDirectory() as directory:
             root = Path(directory)

tools/test_regression_job.py

@@ -110,6 +110,7 @@ class JobTests(unittest.TestCase):
                 cli.chmod(0o700)
                 manifest = {'lbctl': str(cli), 'deployment': {'version': 'v4.11.0', 'port': 15351}}
                 suite = profile()
+                suite['coverage_suite'] = 'smoke'
                 suite['resources'] = [{'id': 'chat', 'kind': 'pipeline', 'body': {'name': 'QA', 'config': {'ai': {}}}}]
                 suite['cases'][0].update(args=['pipeline', 'run', '${chat}'], expect='TEST_OK')
                 suite['phases'][0].update(duration_seconds=0.5, rps=4, workers=2)
@@ -117,6 +118,7 @@ class JobTests(unittest.TestCase):
                 job = ControlledJob(manifest, suite, root / 'profile.json', root / 'success', health)
                 result = job.run()
                 self.assertTrue(result['passed'], result)
+                self.assertTrue((job.output / 'suite/coverage.json').exists())
                 self.assertEqual(result['phase'], 'cleaned')
                 self.assertFalse((job.deploy / 'data').exists())
                 self.assertFalse((job.private / 'api-key').exists())