fix(sync): R5 — code↔DB sync reliability & correctness (closes 22 audit findings)

CHANGELOG.md

@@ -6,6 +6,45 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
 
 ## [Unreleased]
 
+### Fixed — R5: code↔DB sync reliability & correctness (2026-06-25 sync audit)
+
+Closes all 22 findings of the five-specialist code↔DB synchronization audit (9 High, 9 Medium,
+4 Low). Branch `fix/sync-remediation-2026-06-25`; 18 TDD tasks; combined suite 4560 passing, 75%
+coverage; ruff/mypy clean. Spec `docs/superpowers/specs/2026-06-25-sync-remediation-design.md`,
+plan `docs/superpowers/plans/2026-06-25-sync-remediation.md`.
+
+- **Reliability (High):** the daily-sync parent `IndexingRun` now emits a continuous heartbeat
+  (targeted `UPDATE`, no `version` lost-update) so the stale-run reaper no longer kills a healthy
+  multi-minute sync (**H1**); the daily cron sub-steps **adopt-or-skip** instead of launching an
+  untracked concurrent pipeline on an active-run conflict (**H9**); `RunCoordinator.start` translates
+  the single-active `IntegrityError` into a clean `RunAlreadyActiveError`/409 with session rollback,
+  and the partial-unique active index is mirrored onto the model for `create_all` test parity (**H8**);
+  `is_indexed` now only counts `completed`/`completed_partial` (a failed-only index is no longer
+  reported as indexed) (**H7**).
+- **Data correctness (High):** batch table analyses are reconciled by the LLM-echoed `table_name`
+  instead of tool-call position, ending silent cross-table misattribution (**H2**); a malformed
+  `confidence_score` degrades only its own table instead of aborting the batch (**H3**); a degraded
+  LLM run (mostly fallback) no longer overwrites previously-good sync rows, and low-confidence rows
+  no longer enforce/surface required-filter guidance (**H4**).
+- **Cost & privacy (High):** sync LLM calls are now metered + budget-gated against the project
+  owner (manual triggers 429 on exhaustion; cron degrades gracefully; ownerless projects run
+  unenforced) (**H5**); DB sample data + distinct values are scrubbed (column denylist + value
+  redaction) before egress to the LLM at both the sync and db-index analyzers, with a per-connection
+  `send_sample_data_to_llm` opt-out (default on) (**H6**).
+- **Medium:** freshness reconciler now covers all connections, not just the first (**M1**);
+  schema-qualified table identity prevents same-named cross-schema tables from collapsing (**M2**);
+  the daily-sync parent run advances through manifest steps instead of 0%→100% (**M3**); the cron
+  wave honors the per-project schedule hour (**M4**); daily sync regenerates the project overview
+  and the worker logs the correct matched count (**M5**); investigation enrichment is routed to a
+  non-enforced field and `required_filters` payloads are validated + value mappings deep-merged
+  (**M6**); graph-derived `op_kind` heuristics are labelled non-authoritative (over-broad write verbs
+  reclassified) (**M7**); freshness `warnings` uses a proper default + a `sync_failed` flag (**M8**);
+  `get_index_age` guards a NULL `indexed_at` (**M9**).
+- **Low:** the reaper logs a sweep even when the driver returns an unknown rowcount (**L1**); the
+  prompt header no longer fabricates an "analyzed" date for a never-completed sync (**L2**); daily
+  child-run orphaning is covered by H1+H9 (**L3**); context truncation is marked and relevance
+  matching tightened (**L4**).
+
 ### Added
 
 - **MCP protocol-polish (F5/F6/F9).** Shipped in three batched releases on top

backend/.env.example

@@ -177,6 +177,12 @@ CORS_ORIGINS=["http://localhost:3000","http://localhost:3100","https://checkmyda
 # AUTO_INDEX_DB_ON_TEST=false
 # DB_INDEX_INCREMENTAL_ENABLED=true   # diff-only schema reindex on refresh (R2-3)
 
+# ----- R5 sync remediation ---------------------------------------------------
+# SYNC_PII_SCRUBBING_ENABLED=true
+# SYNC_MIN_CONFIDENCE_TO_ENFORCE_FILTERS=2
+# SYNC_MIN_SUCCESS_RATIO_TO_PERSIST=0.5
+# SYNC_BUDGET_ENFORCEMENT_ENABLED=true
+
 # ----- Knowledge lifecycle maintenance ----------------------------------------
 # How often (hours) the background loop runs learning/session-note confidence
 # decay and insight TTL/decay. Independent of backups so decay always runs.

backend/alembic/versions/2317bf9d9126_sync_remediation_schema_qualified_.py

@@ -0,0 +1,34 @@
+"""sync_remediation_schema_qualified_uniqueness
+
+Revision ID: 2317bf9d9126
+Revises: a7c8d9e0f1a2
+Create Date: 2026-06-26 15:52:54.795013
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+revision: str = '2317bf9d9126'
+down_revision: Union[str, None] = 'a7c8d9e0f1a2'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    with op.batch_alter_table("db_index") as b:
+        b.drop_constraint("uq_db_index_conn_table", type_="unique")
+        b.create_unique_constraint(
+            "uq_db_index_conn_schema_table",
+            ["connection_id", "table_schema", "table_name"],
+        )
+
+
+def downgrade() -> None:
+    with op.batch_alter_table("db_index") as b:
+        b.drop_constraint("uq_db_index_conn_schema_table", type_="unique")
+        b.create_unique_constraint(
+            "uq_db_index_conn_table",
+            ["connection_id", "table_name"],
+        )

backend/alembic/versions/e909ec65d857_sync_remediation_connection_flag.py

@@ -0,0 +1,33 @@
+"""sync_remediation_connection_flag
+
+Revision ID: e909ec65d857
+Revises: f37386df158c
+Create Date: 2026-06-26 16:34:53.464295
+"""
+
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "e909ec65d857"
+down_revision: Union[str, None] = "f37386df158c"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        "connections",
+        sa.Column(
+            "send_sample_data_to_llm",
+            sa.Boolean(),
+            nullable=False,
+            server_default=sa.text("1"),
+        ),
+    )
+
+
+def downgrade() -> None:
+    with op.batch_alter_table("connections") as batch_op:
+        batch_op.drop_column("send_sample_data_to_llm")

backend/alembic/versions/f37386df158c_sync_remediation_indexing_run_active_.py

@@ -0,0 +1,35 @@
+"""sync_remediation_indexing_run_active_index
+
+Idempotent parity migration: ensures `uq_indexing_runs_active_one` exists on
+every environment.  Production already has this index from migration
+`a1f2b3c4d5e6`; environments that skipped that hotfix (or fresh dev/staging
+setups) get it here.
+
+Revision ID: f37386df158c
+Revises: 2317bf9d9126
+Create Date: 2026-06-26 16:06:16.161424
+"""
+
+from typing import Sequence, Union
+
+from alembic import op
+
+revision: str = "f37386df158c"
+down_revision: Union[str, None] = "2317bf9d9126"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    # NO-OP: the index `uq_indexing_runs_active_one` is authoritatively created
+    # by migration `a1f2b3c4d5e6_add_indexing_runs_and_error_log.py`.
+    # The `__table_args__` entry in `app/models/indexing_run.py` provides
+    # `create_all` parity for tests and fresh dev setups.
+    # Performing create_index here would cause a double-drop on `downgrade base`.
+    pass
+
+
+def downgrade() -> None:
+    # NO-OP: index owned by a1f2b3c4d5e6; dropping it here would cause
+    # a double-drop when a1f2b3c4d5e6 also runs its downgrade.
+    pass

backend/app/agents/sql_agent.py

@@ -1504,17 +1504,22 @@ async def _load_sync_filters_and_mappings(self, connection_id: str) -> tuple[str
         try:
             import json as json_mod
 
+            from app.config import settings
             from app.models.base import async_session_factory
             from app.services.code_db_sync_service import CodeDbSyncService
 
             svc = CodeDbSyncService()
             async with async_session_factory() as session:
                 entries = await svc.get_sync(session, connection_id)
 
+            min_conf: int = settings.sync_min_confidence_to_enforce_filters
             filters_lines: list[str] = []
             mappings_lines: list[str] = []
 
             for e in entries:
+                # H4: skip low-confidence entries — omit from prompt guidance too
+                if (getattr(e, "confidence_score", 0) or 0) < min_conf:
+                    continue
                 rf = getattr(e, "required_filters_json", "{}") or "{}"
                 try:
                     filters = json_mod.loads(rf)
@@ -1547,23 +1552,33 @@ async def _load_required_filters_by_table(self, cfg: ConnectionConfig) -> dict[s
         try:
             import json as json_mod
 
+            from app.config import settings
             from app.core.required_filter_guard import merge_required_filters
             from app.models.base import async_session_factory
             from app.services.code_db_sync_service import CodeDbSyncService
             from app.services.db_index_service import DbIndexService
 
+            min_conf: int = settings.sync_min_confidence_to_enforce_filters
             sync_filters: dict[str, dict[str, str]] = {}
             sync_svc = CodeDbSyncService()
             async with async_session_factory() as session:
                 entries = await sync_svc.get_sync(session, cfg.connection_id)
             for sync_entry in entries:
+                # H4: skip low-confidence / fallback rows — do not enforce their filters
+                if (getattr(sync_entry, "confidence_score", 0) or 0) < min_conf:
+                    continue
                 raw = getattr(sync_entry, "required_filters_json", "{}") or "{}"
                 try:
                     parsed = json_mod.loads(raw)
                 except (json_mod.JSONDecodeError, TypeError):
                     parsed = {}
                 if parsed and isinstance(parsed, dict):
                     sync_filters[sync_entry.table_name] = parsed
+                    # C6: also index under bare suffix when table_name is schema-qualified
+                    # (e.g. "analytics.orders" → also register under "orders")
+                    bare = sync_entry.table_name.split(".")[-1]
+                    if bare != sync_entry.table_name:
+                        sync_filters.setdefault(bare, {}).update(parsed)
 
             index_hints: dict[str, str] = {}
             idx_svc = DbIndexService()

backend/app/api/routes/connections.py

@@ -24,6 +24,7 @@
 from app.services.connection_service import ConnectionService
 from app.services.db_index_service import DbIndexService
 from app.services.membership_service import MembershipService
+from app.services.sync_budget import preflight_owner_budget
 
 logger = logging.getLogger(__name__)
 
@@ -318,6 +319,8 @@ class ConnectionCreate(BaseModel):
     mcp_server_url: str | None = Field(None, max_length=1024)
     mcp_transport_type: Literal["stdio", "sse"] | None = None
     mcp_env: dict[str, str] | None = None
+    # H6: opt-out of sending DB sample data to the LLM (default True = send)
+    send_sample_data_to_llm: bool = True
 
     @field_validator("mcp_env", mode="before")
     @classmethod
@@ -381,6 +384,8 @@ class ConnectionUpdate(BaseModel):
     mcp_server_url: str | None = Field(None, max_length=2000)
     mcp_transport_type: str | None = Field(None, max_length=50)
     mcp_env: dict[str, str] | None = None
+    # H6: opt-out of sending DB sample data to the LLM
+    send_sample_data_to_llm: bool | None = None
 
     @field_validator("ssh_pre_commands")
     @classmethod
@@ -409,6 +414,7 @@ class ConnectionResponse(BaseModel):
     db_user: str | None
     is_read_only: bool
     is_active: bool
+    send_sample_data_to_llm: bool
     ssh_exec_mode: bool
     ssh_command_template: str | None
     ssh_pre_commands: str | None
@@ -1019,6 +1025,11 @@ async def trigger_sync(
         raise HTTPException(status_code=404, detail="Connection not found")
     await _membership_svc.require_role(db, conn.project_id, user["user_id"], "editor")
 
+    # H5: budget pre-flight — block over-budget owners before we even acquire the lock.
+    ok, reason, _ = await preflight_owner_budget(db, conn.project_id)
+    if not ok:
+        raise HTTPException(status_code=429, detail=reason)
+
     sync_start_lock = _sync_start_locks.setdefault(connection_id, asyncio.Lock())
     async with sync_start_lock:
         existing = _sync_tasks.get(connection_id)

backend/app/api/routes/data_investigations.py

@@ -334,8 +334,8 @@ async def _enrich_sync_from_investigation(
                 db,
                 connection_id=inv.connection_id,
                 table_name=table,
-                field="required_filters_json",
-                value=json.dumps({"source": "investigation", "filter": inv.root_cause}),
+                field="query_recommendations",
+                value=f"[from investigation] {inv.root_cause}",
             )
         elif inv.root_cause_category == "column_format":
             await sync_svc.add_runtime_enrichment(

backend/app/api/routes/projects.py

@@ -19,6 +19,7 @@
 from app.services.membership_service import MembershipService
 from app.services.project_service import ProjectService
 from app.services.rule_service import RuleService
+from app.services.sync_budget import preflight_owner_budget
 
 logger = logging.getLogger(__name__)
 
@@ -554,6 +555,12 @@ async def sync_now(
     import uuid
 
     await _membership_svc.require_role(db, project_id, user["user_id"], "editor")
+
+    # C3: budget pre-flight — block over-budget owners before we even acquire the lock.
+    ok, reason, _ = await preflight_owner_budget(db, project_id)
+    if not ok:
+        raise HTTPException(status_code=429, detail=reason)
+
     from app.core import task_queue
     from app.services.run_coordinator import RunAlreadyActiveError, RunCoordinator
 

backend/app/config.py

@@ -168,6 +168,18 @@ def _fix_database_url(self) -> "Settings":
     # Database index settings
     db_index_ttl_hours: int = 24
     db_index_batch_size: int = 5
+
+    # --- R5 sync remediation -------------------------------------------------
+    # H6: scrub PII / secrets from DB samples + distinct values before LLM egress.
+    sync_pii_scrubbing_enabled: bool = True
+    # H4: per-table analyses below this confidence never enforce hard SQL filters.
+    sync_min_confidence_to_enforce_filters: int = 2
+    # H4: if the fraction of non-fallback analyses is below this, keep prior rows
+    # instead of overwriting with a degraded run. 0.0 disables the guard.
+    sync_min_success_ratio_to_persist: float = 0.5
+    # H5: gate sync LLM spend on the project owner's token budget.
+    sync_budget_enforcement_enabled: bool = True
+
     auto_index_db_on_test: bool = False
     # R2-3: reuse prior LLM table analysis for tables whose schema signature
     # is unchanged since the last successful index, instead of re-LLM-ing every

backend/app/connectors/base.py

@@ -25,6 +25,7 @@ class ConnectionConfig:
     ssh_pre_commands: list[str] | None = None
 
     is_read_only: bool = True
+    send_sample_data_to_llm: bool = True
     extra: dict[str, Any] = field(default_factory=dict)
 
     connection_id: str | None = None