fix(memory/graph): skip a corrupt row instead of aborting a whole graph scan (#3239)

## Description

`SQLiteGraphStore._row_to_entity` and `_row_to_relationship` parse
stored text back into objects with no error handling:

```python
properties=json.loads(row["properties"]),
created_at=datetime.fromisoformat(row["created_at"]),
metadata=json.loads(row["metadata"]),
```

These run inside row loops in the multi-row scans — `get_relationships`
and `query_subgraph` (both the relationship loop and neighbour-entity
expansion). A single unparseable row — from a partial write, a manual
edit, or a bad migration — raises `ValueError` (`JSONDecodeError`/bad
ISO timestamp) *inside the loop*, aborting the **entire** query and
taking unrelated, perfectly good edges/nodes down with it.

Reproduction (A→B and A→C both valid; corrupt only A→B's `properties`):

```python
# corrupt one row out-of-band
con.execute("UPDATE relationships SET properties='{oops' WHERE target_id=?", (b.id,))
# BEFORE: both of these raise JSONDecodeError, even though A->C is fine:
await store.get_relationships(a.id)
await store.query_subgraph([a.id], max_hops=1, direction=OUTGOING)
```

This is the same "one bad row breaks the whole scan" robustness gap
already fixed for the CCR store (`cache/backends/sqlite.py`) and the
vector adapter (`memory/adapters/sqlite_vector.py`); the graph adapter
was the remaining store with unguarded row parsing.

## Type of Change

- [x] Bug fix (non-breaking change that fixes an issue)
- [ ] New feature (non-breaking change that adds functionality)
- [ ] Breaking change (fix or feature that would cause existing
functionality to change)
- [ ] Documentation update
- [ ] Performance improvement
- [ ] Code refactoring (no functional changes)

## Changes Made

- `headroom/memory/adapters/sqlite_graph.py`:
- `_row_to_entity` / `_row_to_relationship` now return `... | None`,
wrapping construction in `except (ValueError, TypeError, KeyError)` and
returning `None` (with a `logger.warning`) on a corrupt row.
- Multi-row call sites skip `None`: `get_relationships`,
`query_subgraph` (initial entities, relationship loop, neighbour
expansion), and the per-user entity listing. The single-row `get_entity`
/ `get_entity_by_name` already return `Entity | None`, so a corrupt row
now reads as "not found" rather than raising.
  - Added a module `logger`.
- `tests/test_sqlite_graph_store.py`: added
`test_one_corrupt_row_does_not_abort_a_multi_row_scan` — corrupts one
relationship row out-of-band and asserts `get_relationships` returns the
one good edge and `query_subgraph` completes with `{A, C}`.

## Testing

- [x] Unit tests pass (`pytest`)
- [x] Linting passes (`ruff check`)
- [x] Type checking passes (`mypy headroom`)
- [x] New tests added for new functionality

### Test Output

```text
tests/test_sqlite_graph_store.py::...one_corrupt_row_does_not_abort_a_multi_row_scan  ->  passes with fix, FAILS without it (verified via git stash)
uvx ruff@0.16.2 check headroom/memory/adapters/sqlite_graph.py tests/test_sqlite_graph_store.py  ->  All checks passed!
uvx mypy@1.20.2 headroom/memory/adapters/sqlite_graph.py  ->  Success: no issues found in 1 source file
```

(Note: this test file has pre-existing, unrelated failures/errors on
`main` on Windows — `TestSQLiteGraphStoreMemoryTrackerIntegration` plus
temp-file teardown `WinError 32` in the `NamedTemporaryFile`-based
fixtures. Verified identical counts before and after this change; my new
test uses `tmp_path` and is unaffected.)

## Real Behavior Proof

- Environment: Windows 11, Python 3.12.11, project venv, pytest 9.1.1,
ruff 0.16.2 and mypy 1.20.2 via uvx.
- Exact command / steps: built A→B and A→C edges, corrupted A→B's
`properties` to invalid JSON via a direct sqlite connection, then called
`get_relationships(A)` and `query_subgraph([A], OUTGOING)`. Before the
fix both raised `JSONDecodeError`; after the fix `get_relationships`
returns just the A→C edge and `query_subgraph` returns entities `{A, C}`
with one relationship, skipping the corrupt row.
- Observed result: corrupt rows are skipped (with a warning log); valid
rows in the same scan are returned normally.
- Not tested: no corruption occurs in normal operation; the corrupt row
is produced out-of-band to exercise the guard (matching the real
triggers: partial write, manual edit, migration).

## Runtime Rollout Safety

- Rollout-managed feature(s): none. This is a SQLite graph-store read
path in the memory subsystem, not a rollout-channel-gated runtime
feature.
- Minimum rollout channel: N/A.
- Stable/default behavior changed: no for well-formed data — every valid
row parses and is returned exactly as before. Only the
previously-crashing corrupt-row case changes, from an aborted query to a
skipped row.
- Kill switch / disable path: N/A.
- Unsafe override required: no.
- Qualification impact: none.
- Rollback path: revert this PR.

## Review Readiness

- [x] I have performed a self-review
- [x] This PR is ready for human review

## Checklist

- [x] My code follows the project's style guidelines
- [x] I have performed a self-review of my code
- [x] I have commented my code, particularly in hard-to-understand areas
- [ ] I have made corresponding changes to the documentation (N/A:
internal behavior)
- [x] My changes generate no new warnings
- [x] I have added tests that prove my fix is effective or that my
feature works
- [x] New and existing unit tests pass locally with my changes
- [x] I did **not** edit `CHANGELOG.md`

headroom/memory/adapters/sqlite_graph.py

@@ -14,6 +14,7 @@ This is a drop-in replacement for InMemoryGraphStore that:
 from __future__ import annotations
 
 import json
+import logging
 import sqlite3
 from collections import deque
 from datetime import datetime
@@ -26,6 +27,8 @@ from .graph_models import Entity, Relationship, RelationshipDirection, Subgraph
 if TYPE_CHECKING:
     from ..tracker import ComponentStats
 
+logger = logging.getLogger(__name__)
+
 
 class SQLiteGraphStore:
     """SQLite-based graph store implementing the GraphStore protocol.
@@ -165,19 +168,31 @@ class SQLiteGraphStore:
             "metadata": json.dumps(entity.metadata),
         }
 
-    def _row_to_entity(self, row: sqlite3.Row) -> Entity:
+    def _row_to_entity(self, row: sqlite3.Row) -> Entity | None:
-        """Convert database row to Entity object."""
+        """Convert a database row to an Entity, or None if the row is corrupt.
-        return Entity(
+
-            id=row["id"],
+        ``properties``/``metadata`` (JSON) and ``created_at``/``updated_at``
-            user_id=row["user_id"],
+        (ISO timestamps) are parsed from stored text. A single unparseable row —
-            name=row["name"],
+        from a partial write, a manual edit, or a bad migration — must not abort
-            entity_type=row["entity_type"],
+        an entire multi-row scan (``query_subgraph``, neighbour expansion): one
-            description=row["description"],
+        corrupt edge would otherwise make an unrelated part of the graph
-            properties=json.loads(row["properties"]),
+        unqueryable. Skip the bad row instead.
-            created_at=datetime.fromisoformat(row["created_at"]),
+        """
-            updated_at=datetime.fromisoformat(row["updated_at"]),
+        try:
-            metadata=json.loads(row["metadata"]),
+            return Entity(
-        )
+                id=row["id"],
+                user_id=row["user_id"],
+                name=row["name"],
+                entity_type=row["entity_type"],
+                description=row["description"],
+                properties=json.loads(row["properties"]),
+                created_at=datetime.fromisoformat(row["created_at"]),
+                updated_at=datetime.fromisoformat(row["updated_at"]),
+                metadata=json.loads(row["metadata"]),
+            )
+        except (ValueError, TypeError, KeyError) as exc:
+            logger.warning("skipping corrupt entity row %r: %s", row["id"], exc)
+            return None
 
     def _relationship_to_row(self, relationship: Relationship) -> dict[str, Any]:
         """Convert Relationship object to row dict for insertion."""
@@ -193,19 +208,29 @@ class SQLiteGraphStore:
             "metadata": json.dumps(relationship.metadata),
         }
 
-    def _row_to_relationship(self, row: sqlite3.Row) -> Relationship:
+    def _row_to_relationship(self, row: sqlite3.Row) -> Relationship | None:
-        """Convert database row to Relationship object."""
+        """Convert a database row to a Relationship, or None if the row is corrupt.
-        return Relationship(
+
-            id=row["id"],
+        Same contract as :meth:`_row_to_entity`: a single unparseable relationship
-            user_id=row["user_id"],
+        row (bad ``properties``/``metadata`` JSON or ``created_at`` timestamp) must
-            source_id=row["source_id"],
+        not abort a whole ``get_relationships`` / ``query_subgraph`` scan and take
-            target_id=row["target_id"],
+        unrelated edges down with it. Skip the bad row instead.
-            relation_type=row["relation_type"],
+        """
-            weight=row["weight"],
+        try:
-            properties=json.loads(row["properties"]),
+            return Relationship(
-            created_at=datetime.fromisoformat(row["created_at"]),
+                id=row["id"],
-            metadata=json.loads(row["metadata"]),
+                user_id=row["user_id"],
-        )
+                source_id=row["source_id"],
+                target_id=row["target_id"],
+                relation_type=row["relation_type"],
+                weight=row["weight"],
+                properties=json.loads(row["properties"]),
+                created_at=datetime.fromisoformat(row["created_at"]),
+                metadata=json.loads(row["metadata"]),
+            )
+        except (ValueError, TypeError, KeyError) as exc:
+            logger.warning("skipping corrupt relationship row %r: %s", row["id"], exc)
+            return None
 
     # =========================================================================
     # Entity Operations
@@ -373,7 +398,9 @@ class SQLiteGraphStore:
                     params,
                 )
 
-                return [self._row_to_relationship(row) for row in cursor]
+                return [
+                    rel for row in cursor if (rel := self._row_to_relationship(row)) is not None
+                ]
 
     async def delete_relationship(self, relationship_id: str) -> bool:
         """Delete a single relationship.
@@ -436,9 +463,12 @@ class SQLiteGraphStore:
                     )
                     row = cursor.fetchone()
                     if row is not None:
+                        entity = self._row_to_entity(row)
+                        if entity is None:
+                            continue
                         queue.append((entity_id, 0))
                         visited.add(entity_id)
-                        collected_entities[entity_id] = self._row_to_entity(row)
+                        collected_entities[entity_id] = entity
 
                 # BFS traversal
                 while queue:
@@ -470,6 +500,8 @@ class SQLiteGraphStore:
 
                     for rel_row in cursor:
                         rel = self._row_to_relationship(rel_row)
+                        if rel is None:
+                            continue
 
                         # Add relationship
                         collected_relationships[rel.id] = rel
@@ -496,8 +528,11 @@ class SQLiteGraphStore:
                             )
                             neighbor_row = neighbor_cursor.fetchone()
                             if neighbor_row is not None:
+                                neighbor = self._row_to_entity(neighbor_row)
+                                if neighbor is None:
+                                    continue
                                 visited.add(neighbor_id)
-                                collected_entities[neighbor_id] = self._row_to_entity(neighbor_row)
+                                collected_entities[neighbor_id] = neighbor
                                 queue.append((neighbor_id, depth + 1))
 
                 return Subgraph(
@@ -651,7 +686,9 @@ class SQLiteGraphStore:
                     "SELECT * FROM entities WHERE user_id = ?",
                     (user_id,),
                 )
-                return [self._row_to_entity(row) for row in cursor]
+                return [
+                    entity for row in cursor if (entity := self._row_to_entity(row)) is not None
+                ]
 
     async def clear(self) -> None:
         """Clear all data from the store."""

tests/test_sqlite_graph_store.py

@@ -689,6 +689,49 @@ class TestSQLiteGraphStoreEdgeCases:
         assert len(subgraph.entities) == 0
         assert len(subgraph.relationships) == 0
 
+    @pytest.mark.asyncio
+    async def test_one_corrupt_row_does_not_abort_a_multi_row_scan(self, tmp_path):
+        """A single unparseable row must not take down an entire query.
+
+        Regression: ``_row_to_relationship`` / ``_row_to_entity`` parsed stored
+        JSON/timestamps with no guard, so one corrupt row (partial write, manual
+        edit, bad migration) raised inside the row loop and aborted the whole
+        ``query_subgraph`` / ``get_relationships`` scan — taking unrelated,
+        perfectly good edges down with it. The corrupt row is now skipped.
+        """
+        import sqlite3
+
+        store = SQLiteGraphStore(db_path=str(tmp_path / "graph.db"))
+        a = Entity(user_id="u", name="A", entity_type="n")
+        b = Entity(user_id="u", name="B", entity_type="n")
+        c = Entity(user_id="u", name="C", entity_type="n")
+        for entity in (a, b, c):
+            await store.add_entity(entity)
+        await store.add_relationship(
+            Relationship(user_id="u", source_id=a.id, target_id=b.id, relation_type="e")
+        )
+        await store.add_relationship(
+            Relationship(user_id="u", source_id=a.id, target_id=c.id, relation_type="e")
+        )
+
+        # Corrupt the A->B relationship row's properties JSON out-of-band.
+        con = sqlite3.connect(str(store.db_path))
+        con.execute("UPDATE relationships SET properties = '{oops' WHERE target_id = ?", (b.id,))
+        con.commit()
+        con.close()
+
+        # get_relationships returns the one good edge instead of raising.
+        rels = await store.get_relationships(a.id)
+        assert len(rels) == 1
+        assert rels[0].target_id == c.id
+
+        # query_subgraph completes, skipping the corrupt edge and its node.
+        subgraph = await store.query_subgraph(
+            [a.id], max_hops=1, direction=RelationshipDirection.OUTGOING
+        )
+        assert {e.name for e in subgraph.entities} == {"A", "C"}
+        assert len(subgraph.relationships) == 1
+
     @pytest.mark.asyncio
     async def test_entity_with_special_characters(self, store):
         """Test entity names with special characters."""