feat(observer): wire CoverageTrendManager + CoverageAlertManager into the service

.console/log.md

@@ -6984,3 +6984,20 @@ report + persists; failure is swallowed). Pruned format_flaky_tests_markdown +
 save_test_results from audit.d12_baseline (now wired → D12 gate confirms 0
 findings). Follow-up: cross-session trend load (needs FlakyTestResult.from_dict +
 a history loader) to light up query_trend_analysis.
+
+## 2026-06-18 — COMPLETE coverage trend/alert engines: wire into observer service
+
+Observer-plane #313 remediation. CoverageTrendManager + CoverageAlertManager
+(#279) were built+tested but never driven; the #279 PR claimed "Integration into
+generate_snapshot()" which never existed. Wired them into RepoObserverService:
+default-construct a CoverageTrendManager rooted under the observer artifact dir;
+after coverage is collected, _record_coverage_trend bridges the live
+CoverageSignal → CoverageSnapshot, records it (building trend history), computes
+the trend + a regression check, runs CoverageAlertManager, persists trend+alerts,
+and logs regressions/alerts. Best-effort (try/except) so it never breaks an
+observation; skips cleanly when coverage is unavailable or storage can't build.
+2 tests (records on live coverage; skips when unavailable). Pruned the now-wired
+detect_regression/generate_alerts/save_snapshot/save_alert from d12_baseline —
+D12 gate confirms 0. (calculate_trend_slope/volatility/get_historical_data and
+categorize_alert/get_routes_for_alert remain genuinely unwired public API — stay
+baselined.) Observer suite 1389 green; ruff+ty+audit(B2)+doctor clean.

.custodian/config.yaml

@@ -48,7 +48,6 @@ audit:
     - create
     - create_and_merge
     - detect_pending_advances
-    - detect_regression
     - do_POST
     - evaluate_alerts_dry_run
     - evaluate_per_collector_thresholds
@@ -61,7 +60,6 @@ audit:
     - find_violations
     - format_flaky_tests_table
     - format_report_text
-    - generate_alerts
     - get_assertion_messages
     - get_audit_type_spec
     - get_channels_for_alert
@@ -139,11 +137,9 @@ audit:
     - revert_commit
     - route_alert
     - save_aggregation
-    - save_alert
     - save_report_json
     - save_session_report
     - save_session_results
-    - save_snapshot
     - scrub_sample
     - should_alert_on_failure_rate
     - should_alert_on_flaky_count

src/operations_center/observer/service.py

@@ -200,6 +200,11 @@ def __init__(
         )
         self.artifact_writer = artifact_writer or ObserverArtifactWriter()
         logger.debug("  Infrastructure: artifact_writer (%s)", type(self.artifact_writer).__name__)
+        # Coverage trend + alert engines (CoverageTrendManager / CoverageAlertManager
+        # from #279) live behind this. They were built + tested but never driven;
+        # default-construct one rooted under the observer artifact dir so each
+        # observation records coverage history and computes trends/regressions/alerts.
+        self._coverage_trend_manager = self._build_coverage_trend_manager()
         self.metrics_exporter = metrics_exporter
         if metrics_exporter is not None:
             logger.debug("  Infrastructure: metrics_exporter (%s)", type(metrics_exporter).__name__)
@@ -417,6 +422,9 @@ def observe(self, context: ObserverContext) -> tuple[RepoStateSnapshot, list[str
             )[1]
         )
 
+        # Drive the coverage trend + alert engines from this run's live coverage.
+        self._record_coverage_trend(coverage_signal, context)
+
         signals = RepoSignalsSnapshot(
             recent_commits=recent_commits,
             file_hotspots=file_hotspots,
@@ -453,6 +461,74 @@ def observe(self, context: ObserverContext) -> tuple[RepoStateSnapshot, list[str
         )
         return snapshot, artifacts
 
+    def _build_coverage_trend_manager(self) -> Any:
+        """Default-construct a CoverageTrendManager rooted under the observer
+        artifact dir. Best-effort: returns None if the engine can't be built."""
+        try:
+            from operations_center.observer.coverage_trend_manager import (
+                CoverageTrendManager,
+            )
+
+            return CoverageTrendManager.create_local(
+                root=self.artifact_writer.root / "coverage-trends"
+            )
+        except Exception as exc:  # noqa: BLE001 — feature is best-effort
+            logger.debug("coverage trend manager unavailable: %s", exc)
+            return None
+
+    def _record_coverage_trend(
+        self, coverage_signal: CoverageSignal, context: ObserverContext
+    ) -> None:
+        """Drive the coverage trend + alert engines from this run's coverage.
+
+        Bridges the live CoverageSignal into a CoverageSnapshot, records it to
+        CoverageTrendManager (building the history its trend analysis needs),
+        computes the trend + a regression check, runs CoverageAlertManager, and
+        persists the trend + alerts — logging any regression/alert. This is the
+        live integration for the CoverageTrendManager / CoverageAlertManager
+        engines (#279), which were built and tested but never driven.
+
+        Best-effort: coverage trend/alerting must never break an observation."""
+        manager = self._coverage_trend_manager
+        if manager is None or getattr(coverage_signal, "status", "") == "unavailable":
+            return
+        pct = getattr(coverage_signal, "total_coverage_pct", None)
+        if pct is None:
+            return
+        try:
+            from operations_center.observer.coverage_alerting import CoverageAlertManager
+            from operations_center.observer.coverage_models import CoverageSnapshot
+
+            snapshot = CoverageSnapshot(
+                timestamp=context.observed_at,
+                run_id=context.run_id,
+                source="coverage.py",
+                # The CoverageSignal exposes a single overall (line-based) figure;
+                # use it for all three metrics as the best available approximation.
+                overall_statement_coverage_pct=float(pct),
+                overall_branch_coverage_pct=float(pct),
+                overall_line_coverage_pct=float(pct),
+            )
+            manager.save_snapshot(snapshot)
+            trend = manager.compute_trend_analysis(
+                metric_type="line", granularity="repository", window_days=7
+            )
+            manager.save_trend_analysis(trend)
+            regressed = manager.detect_regression(snapshot, metric_type="line")
+            alerts = CoverageAlertManager().generate_alerts(snapshot, trend_analysis=trend)
+            for alert in alerts:
+                manager.save_alert(alert)
+            if regressed or alerts:
+                logger.warning(
+                    "Coverage trend: run=%s coverage=%.1f%% regressed=%s alerts=%d",
+                    context.run_id,
+                    float(pct),
+                    regressed,
+                    len(alerts),
+                )
+        except Exception as exc:  # noqa: BLE001 — trend/alerting is best-effort
+            logger.debug("coverage trend recording skipped: %s", exc)
+
     def query(self, root: Path | None = None) -> TestSignalQuery:
         """Create a query API for test signal visibility.
 

tests/unit/observer/test_service_cov.py

@@ -137,6 +137,48 @@ def test_observe_happy_path_passes_signals_and_returns(tmp_path: Path) -> None:
     assert signals.file_hotspots[0].path == "a.py"
 
 
+def test_observe_drives_coverage_trend_recording(tmp_path: Path) -> None:
+    # An observation with live coverage now drives CoverageTrendManager: the
+    # bridged snapshot is recorded (building the trend history) — the live wire
+    # for the previously-unwired trend/alert engines.
+    builder = MagicMock()
+    builder.build.return_value = "BUILT"
+    writer = MagicMock()
+    writer.root = tmp_path / "obs"
+    writer.write.return_value = ["x.json"]
+    svc = _make_service(
+        snapshot_builder=builder,
+        artifact_writer=writer,
+        coverage_signal_collector=_collector(
+            CoverageSignal(status="ok", total_coverage_pct=91.5)
+        ),
+    )
+
+    svc.observe(_make_context(tmp_path))
+
+    manager = svc._coverage_trend_manager
+    assert manager is not None
+    assert len(manager.list_snapshots(limit=5)) == 1
+
+
+def test_observe_no_coverage_does_not_record_trend(tmp_path: Path) -> None:
+    builder = MagicMock()
+    builder.build.return_value = "BUILT"
+    writer = MagicMock()
+    writer.root = tmp_path / "obs"
+    writer.write.return_value = ["x.json"]
+    svc = _make_service(
+        snapshot_builder=builder,
+        artifact_writer=writer,
+        coverage_signal_collector=_collector(CoverageSignal(status="unavailable")),
+    )
+
+    svc.observe(_make_context(tmp_path))
+
+    assert svc._coverage_trend_manager is not None
+    assert svc._coverage_trend_manager.list_snapshots(limit=5) == []
+
+
 def test_observe_optional_collectors_present(tmp_path: Path) -> None:
     builder = MagicMock()
     builder.build.return_value = "BUILT"