style: ruff format the CDC + MCP work

CI lint job has been red since 918aa23 because the CDC commits ran
ruff check locally but never `ruff format --check`. This is a
purely cosmetic pass — `ruff format` over the 13 files the CI run
flagged. Tests still 761 passing.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: SonAIengine

src/synaptic/extensions/cdc/state.py

@@ -418,9 +418,7 @@ async def find_deleted_pks(
         # force schema-fresh state — `IF NOT EXISTS` would let stale
         # rows from a prior run leak through.
         await self._conn.execute("DROP TABLE IF EXISTS cdc_current_pks")
-        await self._conn.execute(
-            "CREATE TEMP TABLE cdc_current_pks (pk TEXT PRIMARY KEY)"
-        )
+        await self._conn.execute("CREATE TEMP TABLE cdc_current_pks (pk TEXT PRIMARY KEY)")
 
         payload = [(canonical_pk(p),) for p in live_pks]
         if payload:

src/synaptic/extensions/cdc/sync.py

@@ -162,9 +162,7 @@ async def detect_deletes(
         else:
             live_pks = live_result  # type: ignore[assignment]
 
-        deleted_rows = await self._store.find_deleted_pks(
-            self._source_url, schema.name, live_pks
-        )
+        deleted_rows = await self._store.find_deleted_pks(self._source_url, schema.name, live_pks)
         if not deleted_rows:
             return 0
 
@@ -274,14 +272,10 @@ async def sync_table(
                 row_is_new.append(False)
                 continue
             pk_str = canonical_pk(pk_val)
-            existing = await self._store.get_node_id(
-                self._source_url, schema.name, pk_str
-            )
+            existing = await self._store.get_node_id(self._source_url, schema.name, pk_str)
             row_is_new.append(existing is None)
             if existing is not None and fk_map:
-                snapshot = await self._store.get_fk_edges(
-                    self._source_url, schema.name, pk_str
-                )
+                snapshot = await self._store.get_fk_edges(self._source_url, schema.name, pk_str)
                 if snapshot:
                     prior_fks[pk_str] = snapshot
 
@@ -335,11 +329,7 @@ async def sync_table(
             node_id = deterministic_row_id(self._source_url, schema.name, pk_val)
             fk_snapshot: dict[str, str] | None = None
             if fk_map:
-                fk_snapshot = {
-                    col: str(row[col])
-                    for col in fk_map
-                    if row.get(col) is not None
-                }
+                fk_snapshot = {col: str(row[col]) for col in fk_map if row.get(col) is not None}
             pk_batch.append((canonical_pk(pk_val), node_id, None, fk_snapshot))
 
             change_val = row.get(change_col)
@@ -384,9 +374,7 @@ async def _prune_stale_fk_edges(
         rows: list[dict[str, Any]],
         prior_fks: dict[str, dict[str, str]],
     ) -> int:
-        return await _prune_stale_fk_edges(
-            graph, schema, fk_map, rows, prior_fks, self._source_url
-        )
+        return await _prune_stale_fk_edges(graph, schema, fk_map, rows, prior_fks, self._source_url)
 
 
 async def _prune_stale_fk_edges(
@@ -429,19 +417,14 @@ async def _prune_stale_fk_edges(
             if target_table is None:
                 continue
 
-            old_target_node = deterministic_row_id(
-                source_url, target_table, old_target_pk
-            )
+            old_target_node = deterministic_row_id(source_url, target_table, old_target_pk)
 
             if source_node not in edge_cache:
                 edge_cache[source_node] = await graph.backend.get_edges(
                     source_node, direction="outgoing"
                 )
             for edge in edge_cache[source_node]:
-                if (
-                    edge.target_id == old_target_node
-                    and edge.kind == EdgeKind.RELATED
-                ):
+                if edge.target_id == old_target_node and edge.kind == EdgeKind.RELATED:
                     await graph.backend.delete_edge(edge.id)
                     removed += 1
                     edge_cache[source_node] = [
@@ -506,9 +489,7 @@ async def sync_table(
         # Hash mode always does a full read — there is no watermark
         # to filter on. Pass `where_clause=None` so the SQLite
         # reader still applies the LIMIT but skips the WHERE.
-        reader_result = row_reader(
-            schema.name, where_clause=None, where_params=()
-        )
+        reader_result = row_reader(schema.name, where_clause=None, where_params=())
         if hasattr(reader_result, "__await__"):
             rows = await reader_result  # type: ignore[misc]
         else:
@@ -535,12 +516,8 @@ async def sync_table(
             new_hash = row_hash(row)
             new_hashes[pk_str] = new_hash
 
-            prior_hash = await self._store.get_row_hash(
-                self._source_url, schema.name, pk_str
-            )
-            existing_node = await self._store.get_node_id(
-                self._source_url, schema.name, pk_str
-            )
+            prior_hash = await self._store.get_row_hash(self._source_url, schema.name, pk_str)
+            existing_node = await self._store.get_node_id(self._source_url, schema.name, pk_str)
 
             if existing_node is None:
                 stats.added += 1
@@ -549,19 +526,15 @@ async def sync_table(
                 stats.updated += 1
                 to_ingest.append(row)
                 if fk_map:
-                    snapshot = await self._store.get_fk_edges(
-                        self._source_url, schema.name, pk_str
-                    )
+                    snapshot = await self._store.get_fk_edges(self._source_url, schema.name, pk_str)
                     if snapshot:
                         prior_fks[pk_str] = snapshot
             # else: unchanged — skip
 
         if not to_ingest:
             # Nothing changed; no state advance needed beyond the
             # last_sync_at heartbeat for monitoring.
-            prior_state = await self._store.load_state(
-                self._source_url, schema.name
-            )
+            prior_state = await self._store.load_state(self._source_url, schema.name)
             new_state = TableSyncState(
                 source_url=self._source_url,
                 table_name=schema.name,
@@ -611,27 +584,15 @@ async def sync_table(
             if pk_val is None:
                 continue
             pk_str = canonical_pk(pk_val)
-            node_id = deterministic_row_id(
-                self._source_url, schema.name, pk_val
-            )
+            node_id = deterministic_row_id(self._source_url, schema.name, pk_val)
             fk_snapshot: dict[str, str] | None = None
             if fk_map:
-                fk_snapshot = {
-                    col: str(row[col])
-                    for col in fk_map
-                    if row.get(col) is not None
-                }
-            pk_batch.append(
-                (pk_str, node_id, new_hashes.get(pk_str), fk_snapshot)
-            )
+                fk_snapshot = {col: str(row[col]) for col in fk_map if row.get(col) is not None}
+            pk_batch.append((pk_str, node_id, new_hashes.get(pk_str), fk_snapshot))
 
-        await self._store.upsert_pk_batch(
-            self._source_url, schema.name, pk_batch
-        )
+        await self._store.upsert_pk_batch(self._source_url, schema.name, pk_batch)
 
-        prior_state = await self._store.load_state(
-            self._source_url, schema.name
-        )
+        prior_state = await self._store.load_state(self._source_url, schema.name)
         row_count = (prior_state.row_count if prior_state else 0) + stats.added
         new_state = TableSyncState(
             source_url=self._source_url,

src/synaptic/extensions/db_ingester.py

@@ -364,9 +364,7 @@ async def _read_pg_pks(dsn: str, table: str, pk_col: str) -> list[str]:
 def _pg_row_reader(dsn: str, row_limit: int):
     """Bind a PG row reader closure for the CDC orchestrator."""
 
-    async def reader(
-        table: str, *, where_clause: str | None = None, where_params: tuple = ()
-    ):
+    async def reader(table: str, *, where_clause: str | None = None, where_params: tuple = ()):
         return await _read_pg_rows(
             dsn,
             table,
@@ -592,9 +590,7 @@ async def _read_mysql_pks(dsn: str, table: str, pk_col: str) -> list[str]:
 def _mysql_row_reader(dsn: str, row_limit: int):
     """Bind a MySQL row reader closure for the CDC orchestrator."""
 
-    async def reader(
-        table: str, *, where_clause: str | None = None, where_params: tuple = ()
-    ):
+    async def reader(table: str, *, where_clause: str | None = None, where_params: tuple = ()):
         return await _read_mysql_rows(
             dsn,
             table,

src/synaptic/graph.py

@@ -736,9 +736,7 @@ async def from_database(
                     row_limit=row_limit,
                 )
                 return graph
-            if connection_string.startswith("mysql") or connection_string.startswith(
-                "mariadb"
-            ):
+            if connection_string.startswith("mysql") or connection_string.startswith("mariadb"):
                 await ingester.sync_from_mysql(
                     connection_string,
                     graph,
@@ -846,9 +844,7 @@ async def sync_from_database(
                 tables=tables,
                 row_limit=row_limit,
             )
-        if connection_string.startswith("mysql") or connection_string.startswith(
-            "mariadb"
-        ):
+        if connection_string.startswith("mysql") or connection_string.startswith("mariadb"):
             return await ingester.sync_from_mysql(
                 connection_string,
                 self,

tests/test_cdc_dialects.py

@@ -16,10 +16,7 @@ def test_pg_numbers_placeholders(self):
         assert _translate_placeholders('"updated_at" >= ?', "pg") == '"updated_at" >= $1'
 
     def test_pg_numbers_multiple(self):
-        assert (
-            _translate_placeholders('a = ? AND b = ?', "pg")
-            == 'a = $1 AND b = $2'
-        )
+        assert _translate_placeholders("a = ? AND b = ?", "pg") == "a = $1 AND b = $2"
 
     def test_mysql_uses_percent_s(self):
         assert _translate_placeholders('"col" >= ?', "mysql") == '"col" >= %s'
@@ -28,7 +25,7 @@ def test_sqlite_passthrough(self):
         assert _translate_placeholders('"col" >= ?', "sqlite") == '"col" >= ?'
 
     def test_unknown_dialect_passthrough(self):
-        assert _translate_placeholders('? ? ?', "made-up") == '? ? ?'
+        assert _translate_placeholders("? ? ?", "made-up") == "? ? ?"
 
     def test_question_mark_only_in_placeholder_position(self):
         # `?` inside a literal would be a footgun in real SQL but

tests/test_cdc_ids.py

@@ -23,8 +23,9 @@
 
 class TestNormalizeSourceURL:
     def test_lowercases_scheme_and_host(self):
-        assert normalize_source_url("Postgres://user@HOST:5432/db") == \
-            normalize_source_url("postgresql://user@host:5432/db".replace("postgresql", "postgres"))
+        assert normalize_source_url("Postgres://user@HOST:5432/db") == normalize_source_url(
+            "postgresql://user@host:5432/db".replace("postgresql", "postgres")
+        )
 
     def test_strips_trailing_slash(self):
         a = normalize_source_url("postgres://user@host/db/")
@@ -155,8 +156,5 @@ def test_no_collision_synthetic_50k(self):
         Not a formal collision proof — just guards against regressions
         like hashing only the table name.
         """
-        ids = {
-            deterministic_row_id("postgres://h/d", "t", f"row_{i}")
-            for i in range(50_000)
-        }
+        ids = {deterministic_row_id("postgres://h/d", "t", f"row_{i}") for i in range(50_000)}
         assert len(ids) == 50_000

tests/test_cdc_search_regression.py

@@ -73,12 +73,8 @@ async def two_graphs(self):
             _seed(src_path)
             conn_str = f"sqlite:////{str(src_path).lstrip('/')}"
 
-            full_graph = await SynapticGraph.from_database(
-                conn_str, db=str(full_db), mode="full"
-            )
-            cdc_graph = await SynapticGraph.from_database(
-                conn_str, db=str(cdc_db), mode="cdc"
-            )
+            full_graph = await SynapticGraph.from_database(conn_str, db=str(full_db), mode="full")
+            cdc_graph = await SynapticGraph.from_database(conn_str, db=str(cdc_db), mode="cdc")
             yield full_graph, cdc_graph
             await full_graph.backend.close()
             await cdc_graph.backend.close()
@@ -98,13 +94,12 @@ async def test_korean_query_top_result_matches(self, two_graphs):
             assert cdc_top, f"cdc graph returned nothing for {query!r}"
             # Top-1 must agree on the same data point.
             assert full_top[0] == cdc_top[0], (
-                f"top result diverged for {query!r}: "
-                f"full={full_top[0]} cdc={cdc_top[0]}"
+                f"top result diverged for {query!r}: full={full_top[0]} cdc={cdc_top[0]}"
             )
 
     async def test_topk_set_matches(self, two_graphs):
         full_graph, cdc_graph = two_graphs
-        for query in ("운동",  "겨울", "정장"):
+        for query in ("운동", "겨울", "정장"):
             full_top = set(await _topk_titles(full_graph, query, k=5))
             cdc_top = set(await _topk_titles(cdc_graph, query, k=5))
             # The top-k *set* should match — internal ordering can
@@ -126,9 +121,7 @@ async def graph_and_src(self):
             _seed(src_path)
             conn_str = f"sqlite:////{str(src_path).lstrip('/')}"
 
-            graph = await SynapticGraph.from_database(
-                conn_str, db=str(graph_db), mode="cdc"
-            )
+            graph = await SynapticGraph.from_database(conn_str, db=str(graph_db), mode="cdc")
             yield graph, conn_str
             await graph.backend.close()
 

tests/test_cdc_sync_delete.py

@@ -89,9 +89,7 @@ async def test_delete_row_removes_node_and_edges(self, graph_and_src):
 
         # PK index forgot the row.
         store = graph.backend.cdc_state_store()
-        assert (
-            await store.get_node_id(conn_str, "products", "1")
-        ) is None
+        assert (await store.get_node_id(conn_str, "products", "1")) is None
 
     async def test_other_rows_untouched(self, graph_and_src):
         graph, conn_str, src_path = graph_and_src

tests/test_cdc_sync_fk.py

@@ -89,9 +89,7 @@ async def test_fk_repoint_removes_old_edge_and_creates_new(self, graph_and_src):
 
         # Repoint product 1 from category 1 → category 3.
         con = sqlite3.connect(src_path)
-        con.execute(
-            "UPDATE products SET category_id = 3, updated_at = 5000 WHERE id = 1"
-        )
+        con.execute("UPDATE products SET category_id = 3, updated_at = 5000 WHERE id = 1")
         con.commit()
         con.close()
 

tests/test_cdc_sync_hash.py

@@ -142,9 +142,7 @@ async def test_value_change_propagates(self, graph_and_src):
         # bind us to the digest function. Just make sure it changed.)
         from synaptic.extensions.cdc.hashing import row_hash as rh
 
-        assert new_hash == rh(
-            {"id": 1, "key": "theme", "value": "light"}
-        )
+        assert new_hash == rh({"id": 1, "key": "theme", "value": "light"})
 
     async def test_insert_propagates(self, graph_and_src):
         graph, conn_str, src_path = graph_and_src