diff --git a/Sensor/adr_sensor/parsers/warp_parser.py b/Sensor/adr_sensor/parsers/warp_parser.py index 1cc3f7f..f853d69 100644 --- a/Sensor/adr_sensor/parsers/warp_parser.py +++ b/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), ) diff --git a/Sensor/tests/test_parsers.py b/Sensor/tests/test_parsers.py index 4112a14..fa65c35 100644 --- a/Sensor/tests/test_parsers.py +++ b/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)