Validate immutable report identities and document historical provenance import

automation/bug_intake.py

@@ -1,6 +1,7 @@
 #!/usr/bin/env python3
 """Private, non-agent admission and leased queue for saphid/frame-control only."""
 import argparse
+from datetime import datetime, timezone
 import base64
 import binascii
 import hashlib
@@ -38,6 +39,7 @@ SCHEMA = '''CREATE TABLE IF NOT EXISTS reports (
  activity_id TEXT PRIMARY KEY, fingerprint TEXT NOT NULL, recorded INTEGER NOT NULL,
  kind TEXT NOT NULL, value TEXT NOT NULL, note TEXT NOT NULL,
  UNIQUE(fingerprint,kind,value,note));
+ CREATE INDEX IF NOT EXISTS history_activity_case ON history_activity(fingerprint,recorded);
  CREATE TABLE IF NOT EXISTS history_relations (
  left_id TEXT NOT NULL, right_id TEXT NOT NULL, note TEXT NOT NULL,
  PRIMARY KEY(left_id,right_id));
@@ -45,6 +47,16 @@ SCHEMA = '''CREATE TABLE IF NOT EXISTS reports (
  CREATE INDEX IF NOT EXISTS reports_fingerprint ON reports(fingerprint);'''
 
 
+def history_time(value):
+    if isinstance(value, (int, float)):
+        date = datetime.fromtimestamp(value, timezone.utc)
+    else:
+        date = datetime.fromisoformat(str(value).replace('Z', '+00:00'))
+        if date.tzinfo is None:
+            raise ValueError('history timestamp requires timezone')
+    return date.astimezone(timezone.utc).isoformat(timespec='milliseconds').replace('+00:00', 'Z')
+
+
 class Rejected(Exception):
     def __init__(self, status):
         self.status = status
@@ -57,7 +69,7 @@ class Queue:
             db.executescript(SCHEMA)
             # Metadata survives queue payload retention. Upgrade existing installations.
             db.execute('INSERT OR IGNORE INTO history_events SELECT id,fingerprint, '
-                       "json_extract(payload,'$.event.event'),CAST(created AS TEXT),'queue' FROM reports")
+                       "json_extract(payload,'$.event.event'),strftime('%Y-%m-%dT%H:%M:%fZ',created,'unixepoch'),'queue' FROM reports")
             db.execute('INSERT OR IGNORE INTO history_activity '
                        "SELECT 'finish:'||d.report_id,r.fingerprint,d.finished,'disposition',d.disposition,'' "
                        'FROM dispositions d JOIN reports r ON r.id=d.report_id')
@@ -135,9 +147,10 @@ class Queue:
                 raise ValueError()
             # Ignore arbitrary envelope instructions; preserve only event data for triage.
             payload = json.dumps({'repository': REPO, 'project_id': PROJECT, 'event': {'uuid': event_id, 'event': name, 'properties': props}}, allow_nan=False)
+            observed = history_time(event.get('timestamp', now))
             fingerprint = hashlib.sha256(json.dumps([name, props.get('$exception_type'), evidence],
                                                     ensure_ascii=True).encode()).hexdigest()
-        except (KeyError, TypeError, ValueError, AttributeError, RecursionError):
+        except (KeyError, TypeError, ValueError, OverflowError, AttributeError, RecursionError):
             raise Rejected(422) from None
         with self.connect() as db:
             db.execute('BEGIN IMMEDIATE')
@@ -147,7 +160,7 @@ class Queue:
             if prior and prior[0] != fingerprint:
                 raise Rejected(422)
             db.execute('INSERT OR IGNORE INTO history_events VALUES(?,?,?,?,?)',
-                       (event_id, fingerprint, name, str(event.get('timestamp', now)), 'webhook'))
+                       (event_id, fingerprint, name, observed, 'webhook'))
             if db.execute('SELECT 1 FROM reports WHERE id=? OR (fingerprint=? AND created>=?)',
                           (event_id, fingerprint, now - 7*86400)).fetchone():
                 return 'duplicate'
@@ -279,7 +292,7 @@ class Queue:
                 raise ValueError('invalid evidence')
             fingerprint = hashlib.sha256(json.dumps([name, props.get('$exception_type'), evidence],
                                                     ensure_ascii=True).encode()).hexdigest()
-            rows.append((eid, fingerprint, name, str(event['timestamp']), 'posthog-history'))
+            rows.append((eid, fingerprint, name, history_time(event['timestamp']), 'posthog-history'))
         with self.connect() as db:
             for row in rows:
                 old = db.execute('SELECT fingerprint FROM history_events WHERE event_id=?', (row[0],)).fetchone()

docs/bug-intake.md

@@ -119,3 +119,6 @@ Use the same executable/`--db` prefix with:
 - `import-history PRIVATE_JSON_FILE`: fixed repository/project envelope with `events` (UUID, event name, timestamp, scoped properties). Indexes history only, transactionally rejects conflicting UUID evidence. Re-import is idempotent. This local trusted operator tool does not replace signed webhook admission.
 
 New queue installations migrate existing admitted reports/dispositions into history. Workers must consult it before work, register their thread immediately, append artifacts/status as they happen, and check current main/releases before deciding a prior fix applies. The live scheduler instructions mirror [worker-policy.md](../automation/worker-policy.md); [AGENTS.md](../AGENTS.md) makes this discoverable to other repo agents. Block automated work if the shared ledger cannot be consulted. Cross-kind correlations need evidence and explicit `relate`; this avoids false automatic merging of unrelated reports.
+
+
+Correlation deployment evidence (2026-10-20): imported all 3,162 scoped PostHog events (3,155 exceptions, seven bug reports) into 42 exact groups; existing smoke jobs are indexed separately. Added 17 evidence-noted prior-PR references across these groups to merged PRs #61/#70/#78/#85 and their actual merge commits. These are prior context, not newly verified resolution. Original implementation thread links for those PRs were not found in accessible lxso2/lxso1/nursedroid/MacBook Pro indexes; this gap is recorded, rather than inventing thread associations. The earlier independent smoke worker is linked to its actual thread. Focused suite now has 22 passing tests, including retention/reopen, duplicate occurrence preservation, cross-kind links, transactional import conflict rejection, no-work history import and invalid-scope/artifact rejection. Private SQLite backup was taken before migration.

tests/test_bug_intake.py

@@ -103,6 +103,21 @@ class Intake(unittest.TestCase):
         with self.assertRaises(ValueError):
             self.q.history(other['uuid'])
 
+    def test_history_live_uuid_conflict_and_bad_time_fail_closed(self):
+        body = self.body()
+        self.send(body)
+        body['event']['properties']['$exception_message'] = 'different'
+        body['event']['properties']['$exception_list'][0]['value'] = 'different'
+        with self.assertRaises(m.Rejected) as error:
+            self.send(body)
+        self.assertEqual(error.exception.status, 422)
+        invalid = self.body()
+        invalid['event']['timestamp'] = 'not a timestamp'
+        with self.assertRaises(m.Rejected) as error:
+            self.send(invalid)
+        self.assertEqual(error.exception.status, 422)
+        self.assertEqual(len(self.q.history_list()), 1)
+
     def test_native_standard_signatures_and_replays(self):
         raw = json.dumps(self.body()).encode()
         good = 'v1,' + base64.b64encode(hmac.new(self.key,