Observal · SrihariLegend · Aug 3, 2026 · Aug 3, 2026 · coderabbitai · Aug 3, 2026
@@ -0,0 +1,40 @@
+# SPDX-FileCopyrightText: 2026 The Observal Authors
+# SPDX-License-Identifier: Apache-2.0
+
+"""Persist insights analysis previously discarded each run.
+
+Adds ``version_impact`` and ``registry_offer`` (JSON, nullable) to
+``insight_reports``. The pipeline already computes both every run — the
+cross-user layer/config correlation analysis and the deterministic registry
+component shortlist shown to the model — but historically folded them into the
+LLM prompt and then discarded them. These columns persist that analysis so
+downstream consumers (duplicate detection, pull-time recommendations,
+governance drift signals) can read it without re-running the pipeline.
+
+Both columns are nullable: pre-existing reports predate the columns (and never
+had the data), and a run may legitimately produce no version impact or an
+empty offer. No backfill is needed — null is the correct value for old rows.
+
+Revision ID: 023_insight_analysis_payload
+Revises: 022_user_recommendations
+Create Date: 2026-08-03
+"""
+
+import sqlalchemy as sa
+
+from alembic import op
+
+revision = "023_insight_analysis_payload"
+down_revision = "022_user_recommendations"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.add_column("insight_reports", sa.Column("version_impact", sa.JSON(), nullable=True))
+    op.add_column("insight_reports", sa.Column("registry_offer", sa.JSON(), nullable=True))
+
+
+def downgrade() -> None:
+    op.drop_column("insight_reports", "registry_offer")
+    op.drop_column("insight_reports", "version_impact")
@@ -5,15 +5,19 @@
 
 from fastapi import Depends, HTTPException, Query
 from fastapi.responses import HTMLResponse
+from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from api.deps import get_db, require_role
+from api.deps import check_listing_visibility_async, get_db, optional_current_user, require_role
+from models.insight_report import InsightReport, InsightReportStatus
 from models.user import User, UserRole
 from schemas.insights import (
     ApplySuggestionsRequest,
     GenerateInsightRequest,
     InsightReportListItem,
     InsightReportResponse,
+    RecommendedAddition,
+    RecommendedAdditionsResponse,
 )
 
 from ._router import router
@@ -113,3 +117,89 @@ async def delete_agent_insight_reports(
     from api.routes.insights import clear_agent_reports
 
     return await clear_agent_reports(agent_id, db, current_user)
+
+
+@router.get("/{agent_id}/insights/recommended-additions", response_model=RecommendedAdditionsResponse)
+async def agent_recommended_additions(
+    agent_id: str,
+    db: AsyncSession = Depends(get_db),
+    current_user: User | None = Depends(optional_current_user),
+):
+    """Public, evidence-backed add-on recommendations for an agent.
+
+    Returns the latest completed insight report's deterministic component
+    shortlist (``registry_offer``) — public registry components the agent does
+    not yet use but might benefit from based on observed usage. This is a
+    *public* surface: unlike the full insight report, it exposes only public
+    component references and never session telemetry, so it is available to
+    anyone who can see the agent (including anonymous browsing).
+
+    Empty ``items`` means no report exists, the offer was empty, or the feature
+    was disabled at generation time. Callers hide the rail rather than erroring.
+    """
+    from api.routes.agent.helpers import _load_agent
+
+    agent = await _load_agent(
+        db,
+        agent_id,
+        prefer_user_id=current_user.id if current_user else None,
+        current_user=current_user,
+        include_all_statuses=False,
+    )
+    if not agent or not await check_listing_visibility_async(agent, current_user, db):
+        raise HTTPException(status_code=404, detail="Agent not found")
+
+    # Latest completed report for this agent. Only completed reports carry a
+    # registry_offer; pending/running/failed rows are skipped.
+    stmt = (
+        select(InsightReport)
+        .where(
+            InsightReport.agent_id == agent.id,
+            InsightReport.status == InsightReportStatus.completed,
+        )
+        .order_by(InsightReport.completed_at.desc().nulls_last())
+        .limit(1)
+    )
+    report = (await db.execute(stmt)).scalar_one_or_none()
+
+    empty = RecommendedAdditionsResponse(agent_id=agent.id)
+    if not report:
+        return empty
+
+    offer = report.registry_offer
+    if not isinstance(offer, dict) or not offer.get("enabled", True):
+        return empty
+
+    entries_by_type = offer.get("entries_by_type") or {}
+    if not isinstance(entries_by_type, dict):
+        return empty
+
+    items: list[RecommendedAddition] = []
+    for _type_plural, entries in entries_by_type.items():
+        if not isinstance(entries, list):
+            continue
+        for entry in entries:
+            if not isinstance(entry, dict):
+                continue
+            # Each entry is a CatalogOffer.to_catalog_entry() dict: type, id,
+            # qualified_name, name, description, category. Skip anything that
+            # lacks the minimum fields needed to render a link.
+            if not entry.get("id") or not entry.get("type"):
+                continue
+            items.append(
+                RecommendedAddition(
+                    type=str(entry["type"]),
+                    id=str(entry["id"]),
+                    qualified_name=str(entry.get("qualified_name") or entry.get("name") or entry["id"]),
+                    name=str(entry.get("name") or entry.get("qualified_name") or entry["id"]),
+                    description=entry.get("description"),
+                    category=entry.get("category"),
+                )
+            )
+
+    return RecommendedAdditionsResponse(
+        agent_id=agent.id,
+        items=items,
+        source_report_id=report.id,
+        generated_at=report.completed_at,
+    )
@@ -72,6 +72,16 @@ class InsightReport(Base):
     aggregated_data: Mapped[dict | None] = mapped_column(JSON, nullable=True)
     report_version: Mapped[int] = mapped_column(Integer, default=1)
 
+    # Analysis previously computed and discarded each run. Persisted now so
+    # downstream consumers (duplicate detection, pull-time recommendations,
+    # governance drift signals) can read them without re-running the pipeline.
+    # ``version_impact`` is the cross-user layer/config correlation analysis;
+    # ``registry_offer`` is the deterministic component shortlist the model was
+    # shown. Both are nullable: old reports predate the columns, and a run may
+    # legitimately produce no version impact or an empty offer.
+    version_impact: Mapped[dict | None] = mapped_column(JSON, nullable=True)
+    registry_offer: Mapped[dict | None] = mapped_column(JSON, nullable=True)
+
     # Self-learn fields
     applied_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
     applied_items: Mapped[dict | None] = mapped_column(JSON, nullable=True)

@@ -0,0 +1,88 @@
+# SPDX-FileCopyrightText: 2026 The Observal Authors
+# SPDX-License-Identifier: Apache-2.0
+
+"""Typed schema for the insights pipeline's analytical payload.
+
+The pipeline computes a rich set of analysis *before* it writes any narrative
+prose: deterministic metrics, LLM-extracted facets, cross-user version-impact
+analysis, and a deterministic registry shortlist. Historically most of this
+was stored as untyped JSON blobs on the ``InsightReport`` (``metrics``,
+``narrative``, ``aggregated_data``), while ``version_impact`` and
+``registry_offer`` were computed and then discarded every run.
+
+This module defines the canonical top-level shape of that payload so there is
+a single source of truth for what one pipeline run produces, and so structural
+drift (a whole section going missing, a dict arriving as a list) is detectable
+at write time instead of silently persisted.
+
+The schema is intentionally permissive on nested fields: ``metrics``,
+``narrative`` and ``aggregated_data`` carry LLM-generated content whose exact
+shape varies, so they are typed as ``dict`` rather than fully modelled.
+Validation catches gross structural errors, not minor LLM output variance — a
+report must never fail to persist because the model added an unexpected field.
+"""
+
+from __future__ import annotations
+
+import structlog
+from pydantic import BaseModel, ConfigDict, Field, ValidationError
+
+logger = structlog.get_logger(__name__)
+
+
+class InsightAnalysisPayload(BaseModel):
+    """The analytical output of one insights pipeline run.
+
+    Fields are grouped by how they were produced so a reader can tell
+    deterministic data apart from LLM-inferred content:
+
+    * Deterministic: ``metrics``, ``aggregated_data``, ``version_impact``,
+      ``registry_offer``, ``facets_summary`` (the last is an aggregate of
+      per-session LLM facets, but the aggregation itself is deterministic).
+    * Generated: ``narrative`` (the LLM-written report sections, including
+      ``suggestions``).
+    """
+
+    model_config = ConfigDict(extra="allow")
+
+    # Deterministic metrics from ClickHouse (has ``rich`` / ``overview``).
+    metrics: dict = Field(default_factory=dict)
+    # LLM-generated narrative sections (including ``suggestions``).
+    narrative: dict = Field(default_factory=dict)
+    # Pre-existing roll-up: metrics + facets_summary + regressions +
+    # cross_user_patterns. Kept as a dict for back-compat with readers.
+    aggregated_data: dict = Field(default_factory=dict)
+    # Cross-user layer/config correlation analysis. Previously discarded
+    # after being folded into the LLM prompt; now persisted.
+    version_impact: dict | None = None
+    # Deterministic shortlist of registry components the agent does not yet
+    # use. Previously discarded after being shown to the model; now persisted
+    # so future consumers (duplicate detection, pull-time recs) can read it
+    # without re-running the pipeline.
+    registry_offer: dict | None = None
+    # Aggregate of per-session facets.
+    facets_summary: dict = Field(default_factory=dict)
+    # Number of sessions the run analysed.
+    sessions_analyzed: int = 0
+
+
+def validate_payload(content: dict) -> dict | None:
+    """Structurally validate a pipeline run's output.
+
+    Returns a validated ``InsightAnalysisPayload`` dict on success, or ``None``
+    on structural mismatch. Failures are *never* fatal: callers persist the
+    raw ``content`` dict as a fallback so a schema mismatch cannot break
+    report generation. The mismatch is logged so drift is observable.
+    """
+    try:
+        payload = InsightAnalysisPayload.model_validate(content)
+    except ValidationError as e:
+        # Log only loc/type/msg — never the raw input, which can carry
+        # LLM-generated narrative or session-derived text from ``content``.
+        safe = [{"loc": err["loc"], "type": err["type"], "msg": err["msg"]} for err in e.errors(include_input=False)]
+        logger.warning("insight_payload_validation_failed", errors=safe)
+        return None
+    return payload.model_dump()
+
+
+__all__ = ["InsightAnalysisPayload", "validate_payload"]
@@ -75,7 +75,42 @@ class InsightReportResponse(BaseModel):
     previous_report_id: uuid.UUID | None = None
     aggregated_data: dict | None = None
     report_version: int = 3
+    # Persisted analysis (previously discarded each run). Nullable: old
+    # reports predate the columns, and a run may produce neither.
+    version_impact: dict | None = None
+    registry_offer: dict | None = None
     # Self-learn fields
     applied_at: datetime | None = None
     applied_items: dict | None = None
     model_config = {"from_attributes": True}
+
+
+class RecommendedAddition(BaseModel):
+    """A single public registry component recommended for an agent.
+
+    Drawn from an insight report's ``registry_offer`` — the deterministic
+    shortlist of approved, visible components the agent does not yet use but
+    might benefit from based on observed usage. Contains only public component
+    references; no session telemetry is exposed.
+    """
+
+    type: str
+    id: str
+    qualified_name: str
+    name: str
+    description: str | None = None
+    category: str | None = None
+
+
+class RecommendedAdditionsResponse(BaseModel):
+    """Evidence-backed add-on recommendations for an agent.
+
+    Empty ``items`` means either no insight report exists for the agent, the
+    report's registry offer is empty, or the feature was disabled at generation
+    time. Callers should simply hide the rail rather than treat it as an error.
+    """
+
+    agent_id: uuid.UUID
+    items: list[RecommendedAddition] = []
+    source_report_id: uuid.UUID | None = None
+    generated_at: datetime | None = None
@@ -19,6 +19,7 @@
 from database import async_session
 from models.agent import Agent, AgentStatus, AgentVersion
 from models.insight_report import InsightReport, InsightReportStatus
+from schemas.insight_analysis import validate_payload
 from services.clickhouse import _query
 from services.insight_version_filters import agent_version_filter
 from services.redis import _get_arq_pool
@@ -245,7 +246,10 @@ async def progress_callback(phase: str, current: int, total: int, message: str)
 
             await _update_report_progress(db, report, "saving", 9, 9, "Saving report")
 
-            # Persist results
+            # Persist results. The pipeline now returns the analysis it
+            # previously discarded (version_impact, registry_offer); persist
+            # those alongside the existing fields so downstream consumers can
+            # read them without a rerun.
             report.metrics = content.get("metrics")
             report.narrative = content.get("narrative")
             report.sessions_analyzed = content.get("sessions_analyzed", 0)
@@ -255,8 +259,21 @@ async def progress_callback(phase: str, current: int, total: int, message: str)
                 "regressions": content.get("regressions"),
                 "cross_user_patterns": content.get("cross_user_patterns"),
             }
+            report.version_impact = content.get("version_impact")
+            report.registry_offer = content.get("registry_offer")
             report.report_version = 3
 
+            # Structural validation of the payload. Non-destructive: a
+            # mismatch is logged but never blocks the write, so a schema drift
+            # cannot break report generation.
+            validated = validate_payload(content)
+            if validated is not None:
+                logger.info(
+                    "insight_payload_validated",
+                    has_version_impact=validated.get("version_impact") is not None,
+                    has_registry_offer=validated.get("registry_offer") is not None,
+                )
+
             models_used = content.get("models_used", [])
             report.llm_model_used = ", ".join(models_used) if models_used else None
 

@@ -485,6 +485,14 @@ async def _run_pipeline(
         "regressions": [],
         "facets_summary": facets_summary,
         "cross_user_patterns": {},
+        # Analysis previously discarded after being folded into the prompt.
+        # Persisted now so downstream consumers can read them without a rerun.
+        "version_impact": version_impact,
+        # Use an explicit ``is not None`` check: CatalogOffer defines __bool__ as
+        # bool(entries_by_type), so an empty/disabled/failed offer would evaluate
+        # falsy and we'd persist None — losing the enabled / registry_has_components
+        # metadata that matters most in exactly those cases.
+        "registry_offer": registry_offer.to_dict() if registry_offer is not None else None,
     }
 
 
@@ -756,4 +764,6 @@ def _empty_report() -> dict:
         "regressions": [],
         "facets_summary": {},
         "cross_user_patterns": {},
+        "version_impact": None,
+        "registry_offer": None,
     }
@@ -88,6 +88,22 @@ def to_summary(self, reused: int = 0) -> dict:
             "registry_has_components": self.registry_has_components,
         }
 
+    def to_dict(self) -> dict:
+        """Serialize the full offer for persistence as analysis payload.
+
+        Unlike :meth:`to_summary` (which records *that* a search happened),
+        this captures *what* was offered so downstream consumers can read the
+        shortlist without re-running the recommender. ``offered_ids`` is a
+        set of UUIDs and is serialized as sorted strings for stable storage.
+        """
+        return {
+            "enabled": self.enabled,
+            "registry_has_components": self.registry_has_components,
+            "item_count": self.item_count,
+            "offered_ids": sorted(str(cid) for cid in self.offered_ids),
+            "entries_by_type": self.entries_by_type,
+        }
+
 
 def build_signals(
     agg: dict | None = None,