Migrate legacy research record status columns

docs/implementation-checkpoints.md

@@ -1,5 +1,15 @@
 # Implementation Checkpoints
 
+## 104 Legacy Research Database Migration
+
+Status: complete; CI and production merge are release gates.
+
+Added an idempotent startup migration for research databases created before the active/retracted lifecycle was introduced. Existing `research_subjects`, `research_evidence`, and `research_iocs` tables receive a non-null `status` column with the default `active`, preserving all existing rows. This repairs the Mission Control Research Cases endpoint on long-lived SQLite deployments while leaving fresh databases unchanged.
+
+Verification covers an exact legacy-schema fixture, repeated database initialization, the case-list query, and the live pilot database. The live endpoint now returns an empty valid case list instead of `no such column: s.status`.
+
+Verification: the full Core suite passed with 394 tests, 14 existing warnings, and 4 subtests. MkDocs strict build and repository diff checks passed. All eight live scoped registry monitors then completed successfully with zero monitor failures.
+
 ## 103 Local Codex Research Investigation Pipeline
 
 Status: implementation complete; local service activation and end-to-end operator acceptance are release verification gates.

soc_store.py

@@ -51,6 +51,12 @@ def connect(db_path: str | None = None) -> sqlite3.Connection:
     return connection
 
 
+def _ensure_column(connection: sqlite3.Connection, table: str, column: str, definition: str) -> None:
+    columns = {str(row["name"]) for row in connection.execute(f"PRAGMA table_info({table})").fetchall()}
+    if column not in columns:
+        connection.execute(f"ALTER TABLE {table} ADD COLUMN {column} {definition}")
+
+
 def init_db(db_path: str | None = None) -> None:
     with closing(connect(db_path)) as connection:
         connection.executescript(
@@ -807,6 +813,9 @@ def init_db(db_path: str | None = None) -> None:
                 ON registry_snapshots (collector_id, created_at DESC);
             """
         )
+        for table in ("research_subjects", "research_evidence", "research_iocs"):
+            _ensure_column(connection, table, "status", "TEXT NOT NULL DEFAULT 'active'")
+        connection.commit()
 
 
 def _existing_state(connection: sqlite3.Connection, finding_id: str) -> Dict[str, str] | None:

tests/test_research_cases.py

@@ -3,6 +3,7 @@
 import io
 import hashlib
 import json
+import sqlite3
 import sys
 import tempfile
 import unittest
@@ -44,6 +45,45 @@
 
 
 class ResearchCaseTests(unittest.TestCase):
+    def test_init_db_migrates_legacy_research_status_columns(self):
+        with tempfile.TemporaryDirectory() as temp_dir:
+            db_path = str(Path(temp_dir) / "legacy.db")
+            with sqlite3.connect(db_path) as connection:
+                connection.executescript(
+                    """
+                    CREATE TABLE research_subjects (
+                        subject_id TEXT PRIMARY KEY, case_id TEXT NOT NULL, subject_type TEXT NOT NULL,
+                        ecosystem TEXT NOT NULL, name TEXT NOT NULL, version TEXT NOT NULL,
+                        publisher TEXT NOT NULL, metadata_json TEXT NOT NULL, created_at TEXT NOT NULL
+                    );
+                    CREATE TABLE research_evidence (
+                        evidence_id TEXT PRIMARY KEY, case_id TEXT NOT NULL, evidence_type TEXT NOT NULL,
+                        title TEXT NOT NULL, locator TEXT NOT NULL, sha256 TEXT NOT NULL,
+                        provenance TEXT NOT NULL, notes TEXT NOT NULL, collected_at TEXT NOT NULL,
+                        created_at TEXT NOT NULL, metadata_json TEXT NOT NULL
+                    );
+                    CREATE TABLE research_iocs (
+                        ioc_id TEXT PRIMARY KEY, case_id TEXT NOT NULL, ioc_type TEXT NOT NULL,
+                        value TEXT NOT NULL, confidence INTEGER NOT NULL, first_seen TEXT,
+                        last_seen TEXT, source_evidence_id TEXT, tags_json TEXT NOT NULL,
+                        created_at TEXT NOT NULL
+                    );
+                    """
+                )
+
+            soc_store.init_db(db_path)
+
+            with soc_store.connect(db_path) as connection:
+                for table in ("research_subjects", "research_evidence", "research_iocs"):
+                    columns = {str(row["name"]) for row in connection.execute(f"PRAGMA table_info({table})")}
+                    self.assertIn("status", columns)
+                    status_column = next(
+                        row for row in connection.execute(f"PRAGMA table_info({table})")
+                        if str(row["name"]) == "status"
+                    )
+                    self.assertEqual(str(status_column["dflt_value"]), "'active'")
+            self.assertEqual(list_cases(db_path=db_path), [])
+
     def _build_ready_case(self, db_path: str) -> dict:
         case = create_case(
             title="Typosquatted payment package investigation",