fix(sensor): support the current Warp database schema

Sensor/adr_sensor/parsers/warp_parser.py

@@ -64,6 +64,7 @@ def parse_all(self) -> List[AgentEvent]:
             conn.row_factory = sqlite3.Row
 
             conversations = self._get_all_conversations(conn)
+            has_ai_blocks = self._has_table(conn, "ai_blocks")
             print(f"[WARP] Found {len(conversations)} conversations")
 
             recent_conversations = self._filter_recent_conversations(conversations)
@@ -74,7 +75,7 @@ def parse_all(self) -> List[AgentEvent]:
             for conversation in recent_conversations:
                 conversation_id = conversation["conversation_id"]
                 try:
-                    exchanges = self._get_conversation_exchanges(conn, conversation_id)
+                    exchanges = self._get_conversation_exchanges(conn, conversation_id, has_ai_blocks)
                     entry = self._create_entry_from_exchanges(conversation_id, exchanges)
                     if entry and entry.has_meaningful_content():
                         entries.append(entry)
@@ -93,9 +94,9 @@ def _get_all_conversations(self, conn) -> List[Dict]:
         """Get all conversation ids and timestamps from the database.
 
         Deliberately excludes the conversation_data column: it is not used
-        anywhere in this parser (exchange content comes from ai_queries /
-        ai_blocks instead), so fetching it here would be wasted I/O for
-        every conversation on every run.
+        anywhere in this parser (exchange content comes from ai_queries and,
+        for legacy schemas, ai_blocks), so fetching it here would be wasted
+        I/O for every conversation on every run.
         """
         cursor = conn.cursor()
         cursor.execute("""
@@ -127,15 +128,26 @@ def _filter_recent_conversations(self, conversations: List[Dict]) -> List[Dict]:
 
         return recent
 
-    def _get_conversation_exchanges(self, conn, conversation_id: str) -> List[Dict]:
-        """Get all exchanges for a conversation with LLM output."""
+    def _has_table(self, conn, table_name: str) -> bool:
+        """Return whether the connected database contains ``table_name``."""
+        return (
+            conn.execute("SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = ?", (table_name,)).fetchone()
+            is not None
+        )
+
+    def _get_conversation_exchanges(
+        self, conn, conversation_id: str, has_ai_blocks: bool
+    ) -> List[Dict]:
+        """Get all exchanges, including legacy LLM output when available."""
         cursor = conn.cursor()
+        llm_output_column = "b.output" if has_ai_blocks else "NULL"
+        ai_blocks_join = "LEFT JOIN ai_blocks b ON q.exchange_id = b.exchange_id" if has_ai_blocks else ""
         cursor.execute(
-            """
+            f"""
             SELECT q.exchange_id, q.conversation_id, q.start_ts, q.input, q.working_directory,
-                   q.output_status, q.model_id, b.output as llm_output
+                   q.output_status, q.model_id, {llm_output_column} as llm_output
             FROM ai_queries q
-            LEFT JOIN ai_blocks b ON q.exchange_id = b.exchange_id
+            {ai_blocks_join}
             WHERE q.conversation_id = ?
             ORDER BY q.start_ts ASC
         """,
@@ -154,13 +166,18 @@ def _create_entry_from_exchanges(
             sorted_exchanges = sorted(exchanges, key=lambda x: x["start_ts"], reverse=True)
             most_recent = sorted_exchanges[0]
             timestamp = normalize_timestamp(most_recent["start_ts"])
+            model_id = most_recent.get("model_id")
+            if isinstance(model_id, str):
+                parsed_model_id = self._parse_json_safely(model_id)
+                if isinstance(parsed_model_id, str):
+                    model_id = parsed_model_id
 
             entry = AgentEvent(
                 timestamp=timestamp,
                 source="warp",
                 session_id=f"warp_{conversation_id}",
                 project_path=most_recent.get("working_directory"),
-                model=most_recent.get("model_id"),
+                model=model_id,
                 raw_log_path=str(self.db_path),
             )
 

Sensor/tests/test_parsers.py

@@ -435,11 +435,11 @@ def test_parse_no_directory(self):
         assert entries == []
 
 
-def _build_warp_db(db_path: Path, conversations: list) -> None:
+def _build_warp_db(db_path: Path, conversations: list, *, include_ai_blocks: bool = True) -> None:
     """Create a synthetic warp.sqlite matching the schema WarpParser queries.
 
     Each item in `conversations` is a dict with keys:
-        conversation_id, last_modified_at, exchanges
+        conversation_id, last_modified_at, exchanges, and optional model_id
     where `exchanges` is a list of (exchange_id, start_ts, input_json, llm_output_json).
     """
     conn = sqlite3.connect(str(db_path))
@@ -466,7 +466,8 @@ def _build_warp_db(db_path: Path, conversations: list) -> None:
         )
         """
     )
-    cursor.execute("CREATE TABLE ai_blocks (exchange_id TEXT, output TEXT)")
+    if include_ai_blocks:
+        cursor.execute("CREATE TABLE ai_blocks (exchange_id TEXT, output TEXT)")
 
     for conv in conversations:
         cursor.execute(
@@ -476,9 +477,17 @@ def _build_warp_db(db_path: Path, conversations: list) -> None:
         for exchange_id, start_ts, input_json, llm_output_json in conv.get("exchanges", []):
             cursor.execute(
                 "INSERT INTO ai_queries VALUES (?, ?, ?, ?, ?, ?, ?)",
-                (exchange_id, conv["conversation_id"], start_ts, input_json, "/tmp/project", "success", "claude"),
+                (
+                    exchange_id,
+                    conv["conversation_id"],
+                    start_ts,
+                    input_json,
+                    "/tmp/project",
+                    "success",
+                    conv.get("model_id", "model-default"),
+                ),
             )
-            if llm_output_json is not None:
+            if include_ai_blocks and llm_output_json is not None:
                 cursor.execute("INSERT INTO ai_blocks VALUES (?, ?)", (exchange_id, llm_output_json))
 
     conn.commit()
@@ -598,6 +607,93 @@ def test_parses_conversation_content(self, tmp_path):
         assert assistant_msg.tools[0].tool_name == "execute_command"
         assert assistant_msg.tools[0].status == "success"
 
+    def test_parses_current_schema_without_ai_blocks(self, tmp_path):
+        """Current databases without ai_blocks should still produce conversation events."""
+        now = datetime.now(timezone.utc).isoformat()
+        conversations = [
+            {
+                "conversation_id": "current-conv",
+                "last_modified_at": now,
+                "exchanges": [self._query_exchange("current-exchange", now)],
+            }
+        ]
+        db_path = tmp_path / "warp.sqlite"
+        _build_warp_db(db_path, conversations, include_ai_blocks=False)
+
+        entries = self._make_parser(tmp_path).parse_all()
+
+        assert len(entries) == 1
+        assert entries[0].session_id == "warp_current-conv"
+        assert entries[0].chat_history[0].content == "hello from current-exchange"
+
+    def test_legacy_schema_retains_llm_output(self, tmp_path):
+        """Legacy databases with ai_blocks should retain their assistant output."""
+        now = datetime.now(timezone.utc).isoformat()
+        action_result_input = json.dumps([{"ActionResult": {"id": "synthetic-tool", "result": {}}}])
+        llm_output = json.dumps({"Received": {"output": [{"Text": {"text": "synthetic response"}}]}})
+        conversations = [
+            {
+                "conversation_id": "legacy-conv",
+                "last_modified_at": now,
+                "exchanges": [("legacy-exchange", now, action_result_input, llm_output)],
+            }
+        ]
+        db_path = tmp_path / "warp.sqlite"
+        _build_warp_db(db_path, conversations)
+
+        entries = self._make_parser(tmp_path).parse_all()
+
+        assert len(entries) == 1
+        assert entries[0].chat_history[0].content == "synthetic response"
+
+    def test_checks_legacy_table_once_per_database(self, tmp_path):
+        """Schema detection should not add one sqlite_master query per conversation."""
+        now = datetime.now(timezone.utc).isoformat()
+        conversations = [
+            {
+                "conversation_id": f"conversation-{index}",
+                "last_modified_at": now,
+                "exchanges": [self._query_exchange(f"exchange-{index}", now)],
+            }
+            for index in range(2)
+        ]
+        db_path = tmp_path / "warp.sqlite"
+        _build_warp_db(db_path, conversations, include_ai_blocks=False)
+        parser = self._make_parser(tmp_path)
+
+        with patch.object(parser, "_has_table", wraps=parser._has_table) as has_table:
+            entries = parser.parse_all()
+
+        assert len(entries) == 2
+        has_table.assert_called_once()
+
+    @pytest.mark.parametrize(
+        ("stored_model_id", "expected_model_id"),
+        [
+            ('"model-quoted"', "model-quoted"),
+            ("model-unquoted", "model-unquoted"),
+            ('"model-malformed', '"model-malformed'),
+        ],
+    )
+    def test_normalizes_model_id(self, tmp_path, stored_model_id, expected_model_id):
+        """Only valid JSON-quoted model IDs should be decoded."""
+        now = datetime.now(timezone.utc).isoformat()
+        conversations = [
+            {
+                "conversation_id": "model-conv",
+                "last_modified_at": now,
+                "model_id": stored_model_id,
+                "exchanges": [self._query_exchange("model-exchange", now)],
+            }
+        ]
+        db_path = tmp_path / "warp.sqlite"
+        _build_warp_db(db_path, conversations, include_ai_blocks=False)
+
+        entries = self._make_parser(tmp_path).parse_all()
+
+        assert len(entries) == 1
+        assert entries[0].model == expected_model_id
+
     def test_parse_all_no_database(self, tmp_path):
         """Test parse_all when the database file doesn't exist."""
         parser = self._make_parser(tmp_path)