fix(audit): verify filtered logs against workspace predecessors

Co-authored-by: multica-agent <[email protected]>

src/langbot/pkg/operation_trace/service.py

@@ -1219,7 +1219,7 @@ class WorkspaceSettingsService:
         # later read only has to verify the rows appended since. Without this the
         # panel re-hashed up to MAX_INTEGRITY_SCAN_ROWS rows on every open,
         # refresh and page turn, which is what made reading the log feel slow.
-        self._integrity_cache: dict[str, dict[str, typing.Any]] = {}
+        self._integrity_cache: dict[tuple, dict[str, typing.Any]] = {}
         self._dropped_count = 0
         # Failed persistence attempts, so a silently broken writer (a missing
         # tenant scope, an RLS rejection) is reported instead of only counted.
@@ -2040,25 +2040,13 @@ class WorkspaceSettingsService:
             )
             rows = list(rows_result.all())
 
-            # Fetch one extra older row so the oldest visible record can still
-            # have its chain link verified instead of being reported as broken.
-            chain_rows: list[typing.Any] = list(rows)
-            if rows:
-                oldest_id = rows[-1].id
-                older_result = await self.ap.persistence_mgr.execute_async(
-                    sqlalchemy.select(model)
-                    .where(model.workspace_uuid == workspace_uuid, model.id < oldest_id)
-                    .order_by(model.id.desc())
-                    .limit(1)
-                )
-                chain_rows = list(rows) + list(older_result.all())
-
+            predecessors = await self._log_predecessors(model, rows)
             records = [
                 self._serialize_log(
                     row,
-                    previous_row=chain_rows[index + 1] if index + 1 < len(chain_rows) else None,
+                    previous_row=predecessors.get(row.id),
                 )
-                for index, row in enumerate(rows)
+                for row in rows
             ]
             integrity_summary = summary['summary']
         except Exception as exc:  # pragma: no cover - defensive
@@ -2094,6 +2082,27 @@ class WorkspaceSettingsService:
             'truncated': False,
         }
 
+    async def _log_predecessors(self, model: typing.Any, rows: list[typing.Any]) -> dict[int, typing.Any]:
+        """Fetch each row's immediate predecessor in its unfiltered Workspace history."""
+
+        if not rows:
+            return {}
+        previous = model.__table__.alias('previous_log')
+        previous_id = (
+            sqlalchemy.select(previous.c.id)
+            .where(previous.c.workspace_uuid == model.workspace_uuid, previous.c.id < model.id)
+            .order_by(previous.c.id.desc())
+            .limit(1)
+            .correlate(model)
+            .scalar_subquery()
+        )
+        result = await self.ap.persistence_mgr.execute_async(
+            sqlalchemy.select(model.id, previous.c.record_hash)
+            .join(previous, previous.c.id == previous_id)
+            .where(model.id.in_([row.id for row in rows]))
+        )
+        return {row.id: row for row in result.all()}
+
     def _invalidate_integrity_cache(self, workspace_uuid: str) -> None:
         """Drop every cached verification for one Workspace.
 
@@ -2144,13 +2153,14 @@ class WorkspaceSettingsService:
         A result computed within :data:`INTEGRITY_CACHE_TTL_SECONDS` is served
         from the cache, which is what keeps the panel responsive: opening,
         refreshing and paging all land inside that window and pay nothing. The
-        cache key is ``(Workspace, listing filters)`` -- an integrity drill-down
-        reuses the very result it is drilling into, and both "everything" and
+        cache key includes the Workspace, listing filters and bound values. An
+        integrity drill-down reuses its source result, and both "everything" and
         "mutations only" are cached independently -- so a cache miss is rare.
         A miss re-verifies the *whole* window from scratch: the scan projects
         only the hash columns and the verifier only computes two booleans, so
         the cost is bounded, and an edit to a row verified on a previous pass is
-        still caught.
+        still caught. Each link is checked against its immediate predecessor in
+        the unfiltered Workspace history, independently of the scan window.
         """
 
         workspace_uuid = None
@@ -2160,7 +2170,11 @@ class WorkspaceSettingsService:
             except AttributeError:
                 continue
             break
-        cache_key = (workspace_uuid, tuple(str(condition) for condition in filters))
+        compiled_filters = [condition.compile() for condition in filters]
+        cache_key = (
+            workspace_uuid,
+            tuple((str(condition), tuple(sorted(condition.params.items()))) for condition in compiled_filters),
+        )
         cacheable = workspace_uuid is not None
         now = time.monotonic()
 
@@ -2176,27 +2190,13 @@ class WorkspaceSettingsService:
             )
             rows = list(rows_result.all())
 
-            # Baseline for the oldest scanned row: its link points at a row that
-            # sits just outside the window. Reads are bounded by
-            # ``MAX_INTEGRITY_SCAN_ROWS``, so only a history larger than that
-            # window ever needs this extra row.
-            previous_row = None
-            oldest_id = rows[-1].id if rows else None
-            if oldest_id is not None:
-                older_result = await self.ap.persistence_mgr.execute_async(
-                    _integrity_scan_select(model, model.workspace_uuid == rows[0].workspace_uuid, model.id < oldest_id)
-                    .order_by(model.id.desc())
-                    .limit(1)
-                )
-                previous_row = older_result.first()
-
+            predecessors = await self._log_predecessors(model, rows)
             # Every pass re-verifies every scanned row. The map is deliberately
             # not merged with the previous cache: a row verified once is exactly
             # what an attacker would edit afterwards.
             new_verified: dict[int, tuple[bool, bool]] = {}
-            for index, row in enumerate(rows):
-                predecessor = rows[index + 1] if index + 1 < len(rows) else previous_row
-                new_verified[row.id] = self._verify_hash_and_chain(row, predecessor)
+            for row in rows:
+                new_verified[row.id] = self._verify_hash_and_chain(row, predecessors.get(row.id))
 
             tampered_ids, integrity_failed_ids, chain_failed_ids = self._verified_lists(new_verified)
             scanned = len(new_verified)

tests/unit_tests/test_operation_trace_incremental.py

@@ -275,3 +275,103 @@ async def test_invalidation_clears_every_cached_view(trace_env):
 
     # A deletion invalidates *all* views for the Workspace, not just one.
     assert _cached_views(service) == []
+
+
+@pytest.fixture
+async def mixed_history(trace_env):
+    service, engine = trace_env
+    previous_hash = None
+    ids = []
+    for index, resource_type in enumerate(['bot', 'pipeline', 'bot', 'model', 'bot', 'pipeline']):
+        row = _build_row(index, previous_hash, created_at=datetime.datetime(2026, 1, index + 1))
+        row.update(
+            resource_type=resource_type,
+            action='update' if resource_type == 'bot' else 'view',
+            level=1 if resource_type == 'bot' else 2,
+            actor_account_uuid='bot-operator' if resource_type == 'bot' else 'reader',
+        )
+        row['record_hash'] = service_module.compute_record_hash(row)
+        previous_hash = row['record_hash']
+        foreign_row = _build_row(index, None)
+        foreign_row['workspace_uuid'] = 'other-workspace'
+        foreign_row['record_hash'] = service_module.compute_record_hash(foreign_row)
+        async with engine.begin() as connection:
+            result = await connection.execute(sqlalchemy.insert(MODEL).values(**row))
+            ids.append(result.inserted_primary_key[0])
+            await connection.execute(sqlalchemy.insert(MODEL).values(**foreign_row))
+    return service, engine, ids
+
+
+@pytest.mark.parametrize(
+    'filters, expected_count',
+    [
+        ({}, 6),
+        ({'resource_type': 'bot'}, 3),
+        ({'action': 'update'}, 3),
+        ({'level': 1}, 3),
+        ({'actor_account_uuid': 'bot-operator'}, 3),
+        ({'search': 'bot'}, 3),
+        ({'since': datetime.datetime(2026, 1, 2), 'until': datetime.datetime(2026, 1, 5)}, 4),
+        ({'resource_type': 'missing'}, 0),
+    ],
+)
+async def test_filtered_history_uses_workspace_predecessors(mixed_history, filters, expected_count):
+    service, _engine, _ids = mixed_history
+    for offset in (0, 1):
+        result = await service.query_logs(WORKSPACE, limit=2, offset=offset, **filters)
+        assert result['total'] == expected_count
+        assert result['scanned_count'] == expected_count
+        assert result['tampered_count'] == result['chain_failed_count'] == result['integrity_failed_count'] == 0
+        assert all(row['chain_ok'] and row['integrity_ok'] for row in result['records'])
+
+
+@pytest.mark.parametrize('field', ['summary', 'prev_hash'])
+async def test_filtered_history_detects_tampering(mixed_history, monkeypatch, field):
+    service, engine, ids = mixed_history
+    monkeypatch.setattr(service_module, 'INTEGRITY_CACHE_TTL_SECONDS', 0.0)
+    await service.query_logs(WORKSPACE, resource_type='bot')
+    async with engine.begin() as connection:
+        await connection.execute(sqlalchemy.update(MODEL).where(MODEL.id == ids[2]).values(**{field: 'tampered'}))
+    for filters in ({}, {'resource_type': 'bot'}, {'resource_type': 'bot', 'integrity': 'issues'}):
+        result = await service.query_logs(WORKSPACE, **filters)
+        assert result['tampered_count'] == result['integrity_failed_count'] == 1
+        assert result['chain_failed_count'] == (1 if field == 'prev_hash' else 0)
+        damaged = next(row for row in result['records'] if row['id'] == ids[2])
+        assert damaged['integrity_ok'] is False
+        assert damaged['chain_ok'] is (field != 'prev_hash')
+        assert all(row['chain_ok'] for row in result['records'] if row['id'] != ids[2])
+
+
+async def test_filter_values_have_distinct_cache_entries(mixed_history):
+    service, engine, ids = mixed_history
+    async with engine.begin() as connection:
+        await connection.execute(sqlalchemy.update(MODEL).where(MODEL.id == ids[2]).values(summary='tampered'))
+    for resource_type, scanned, tampered in [('bot', 3, 1), ('pipeline', 2, 0), ('model', 1, 0), ('bot', 3, 1)]:
+        result = await service.query_logs(WORKSPACE, resource_type=resource_type)
+        assert result['scanned_count'] == scanned
+        assert result['tampered_count'] == tampered
+        assert result['chain_failed_count'] == 0
+    assert len(_cached_views(service)) == 3
+
+
+async def test_integrity_drilldown_keeps_real_predecessors(mixed_history):
+    service, engine, ids = mixed_history
+    damaged_ids = {ids[0], ids[4]}
+    async with engine.begin() as connection:
+        await connection.execute(sqlalchemy.update(MODEL).where(MODEL.id.in_(damaged_ids)).values(summary='tampered'))
+    result = await service.query_logs(WORKSPACE, resource_type='bot', integrity='hash_mismatch')
+    assert {row['id'] for row in result['records']} == damaged_ids
+    assert result['tampered_count'] == result['integrity_failed_count'] == 2
+    assert result['chain_failed_count'] == 0
+    assert all(row['chain_ok'] and not row['integrity_ok'] for row in result['records'])
+
+
+async def test_filtered_scan_limit_keeps_real_predecessors(mixed_history, monkeypatch):
+    service, _engine, _ids = mixed_history
+    monkeypatch.setattr(service_module, 'MAX_INTEGRITY_SCAN_ROWS', 2)
+    result = await service.query_logs(WORKSPACE, resource_type='bot')
+    assert result['total'] == 3
+    assert result['scanned_count'] == 2
+    assert result['scan_truncated'] is True
+    assert result['tampered_count'] == 0
+    assert all(row['chain_ok'] for row in result['records'])