feat: integra segnali cross-repo (SO catalog_signals + DI pipeline_signals) (#5)

* feat: fetch source-observatory signals from GitHub

Adds GitHubCollector.get_raw_file() for fetching raw file content from
raw.githubusercontent.com (no token required on public repos). The renderer
now reads source-observatory/data/catalog/catalog_signals.json at build time
and injects a Source Health section into session_bootstrap.md and a
source_health key into workspace_triage.json.

- signals.py: dataclasses + parser (parse_source_observatory_signals) — pure,
  no I/O, 100% covered
- Renderer caches the fetch result so bootstrap + triage share one HTTP call
- Graceful degradation: unavailable file → labelled error, no crash

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* docs: define repo-signals standard for cross-repo ACB integration

Shared spec for JSON signal files that Lab repos publish for ACB to consume.
Covers schema v1 (envelope + signals array with ok/warn/error status),
topic conventions, legacy SO format note, and onboarding steps for new repos.

Enables dataset-incubator to produce pipeline_signals.json following
this standard as the next integration step.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* feat: integrate dataset-incubator pipeline_signals into ACB context

Adds RepoSignal/RepoSignals dataclasses and parse_repo_signals() in signals.py
for consuming the repo-signals standard v1. Renderer now fetches
dataset-incubator/registry/pipeline_signals.json and injects:
- Pipeline State section in session_bootstrap.md (warn/error candidates only)
- pipeline_state key in workspace_triage.json

Cache ensures both SO and DI signals are fetched exactly once per build.
DI signals degrade gracefully until feat/pipeline-signals is merged to main.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* docs: chiarisce semantica di status:ok in repo-signals standard

Il significato di ok è topic-dipendente. Per pipeline_state, ok indica
struttura candidate valida + mart SQL presente, non readiness completa.
Aggiunta nota esplicita per i consumer.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* fix: risolvi 3 MEDIUM da review Matteo su feat/github-signals

- sposta _UNSET dopo gli import (PEP 8)
- rendi regressions e alerts mutualmente esclusivi in SourceObservatorySignals
- aggiunge test diretti per parse_repo_signals e RepoSignals
- cache test usa side_effect con JSON realistici distinti per SO e DI

---------

Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>
Co-authored-by: Matteo <matteo.cavo.data@gmail.com>

Author: 3 people

schemas/repo-signals.md

@@ -0,0 +1,144 @@
+# Repo Signals — Standard di interoperabilità per ACB
+
+Spec per i file JSON che i repo del Lab possono pubblicare affinché
+`agent-context-builder` li consumi durante il build del contesto.
+
+---
+
+## Scopo
+
+Ogni repo ha conoscenza locale che ACB non può derivare da GitHub:
+- **source-observatory**: salute delle fonti dati (endpoint up/down, regressioni)
+- **dataset-incubator**: stato dei candidati nel pipeline (stage, blocchi, pronti per promozione)
+- (futuri repo): qualunque segnale operativo rilevante per un agente
+
+Anziché costruire parser ad-hoc per ogni repo, ACB legge file JSON che seguono
+questo standard e li aggrega in `workspace_triage.json` e `session_bootstrap.md`.
+
+---
+
+## Convenzione di posizionamento
+
+Ogni repo pubblica il proprio file in una path nota e stabile, committata nel
+repo stesso e aggiornata da CI (o manualmente). ACB la legge via
+`raw.githubusercontent.com` — nessun clone locale necessario.
+
+Paths per repo attivi:
+
+| Repo | Path nel repo | Nota |
+|------|--------------|------|
+| `source-observatory` | `data/catalog/catalog_signals.json` | formato legacy (vedi §Legacy) |
+| `dataset-incubator` | `registry/pipeline_signals.json` | adotta questo standard |
+
+---
+
+## Schema (v1)
+
+```json
+{
+  "schema_version": "1",
+  "generated_at": "2026-04-15",
+  "repo": "dataset-incubator",
+  "topic": "pipeline_state",
+  "summary": {
+    "total": 20,
+    "by_status": { "ok": 16, "warn": 3, "error": 1 }
+  },
+  "signals": [
+    {
+      "id": "irpef-comunale",
+      "status": "ok",
+      "label": "irpef-comunale",
+      "detail": "stage: incubating — anni 2019-2023, fonte: mef",
+      "action": ""
+    },
+    {
+      "id": "ispra-balneazione",
+      "status": "warn",
+      "label": "ispra-balneazione",
+      "detail": "stage: incubating — nessun mart SQL, issue aperta da 60+ giorni",
+      "action": "valutare se bloccare o archiviare"
+    }
+  ]
+}
+```
+
+### Campi obbligatori
+
+| Campo | Tipo | Descrizione |
+|-------|------|-------------|
+| `schema_version` | `"1"` | Versione dello schema (stringa, non numero) |
+| `generated_at` | `YYYY-MM-DD` | Data di generazione |
+| `repo` | string | Nome del repo GitHub (es. `"dataset-incubator"`) |
+| `topic` | string | Dominio del segnale — vedi §Topic |
+| `signals` | array | Lista di segnali — vedi §Segnale |
+
+### Campi opzionali
+
+| Campo | Tipo | Descrizione |
+|-------|------|-------------|
+| `summary` | object | Conteggi aggregati per status |
+| `summary.total` | int | Totale elementi osservati |
+| `summary.by_status` | object | Conteggi per valore di status |
+
+---
+
+## Valori di `topic`
+
+| Valore | Repo | Significato |
+|--------|------|-------------|
+| `catalog_health` | source-observatory | Salute degli endpoint delle fonti |
+| `pipeline_state` | dataset-incubator | Stato dei candidati nel pipeline |
+
+Nuovi topic: aprire issue in `agent-context-builder` prima di adottarli,
+per coordinare il parser ACB.
+
+---
+
+## Schema del segnale (`signals[]`)
+
+| Campo | Tipo | Obbligatorio | Descrizione |
+|-------|------|-------------|-------------|
+| `id` | string | sì | Identificatore stabile (slug del dataset, nome fonte, ecc.) |
+| `status` | `"ok"` \| `"warn"` \| `"error"` | sì | Stato sintetico |
+| `label` | string | sì | Nome leggibile da mostrare in UI/bootstrap |
+| `detail` | string | sì | Descrizione human-readable dello stato corrente |
+| `action` | string | sì | Azione suggerita — stringa vuota `""` se nessuna |
+
+### Semantica di `status`
+
+| Valore | Significato |
+|--------|-------------|
+| `ok` | Nessun problema noto, nessuna azione richiesta |
+| `warn` | Attenzione: qualcosa è anomalo ma non bloccante |
+| `error` | Bloccante o regressione: richiede azione |
+
+ACB mostra nel `session_bootstrap.md` solo i segnali `warn` e `error`.
+I segnali `ok` appaiono solo nei conteggi di `summary`.
+
+**Nota per i consumer**: il significato di `ok` dipende dal `topic` del file.
+Per `pipeline_state` (dataset-incubator), `ok` significa **struttura candidate
+coerente e layer mart presente** — non implica che la pipeline abbia girato,
+che i dati siano freschi o che il candidate sia pronto per la promozione.
+Leggere sempre il campo `detail` per il contesto specifico del segnale.
+
+---
+
+## Legacy: source-observatory
+
+`source-observatory` usa un formato precedente allo standard (`catalog_signals.json`).
+ACB mantiene un parser dedicato (`signals.py::parse_source_observatory_signals`)
+che mappa il vecchio formato sul modello interno.
+
+Migrazione pianificata: quando SO adotterà questo schema, il parser legacy
+verrà rimosso. Non c'è urgenza — il formato SO è stabile e ben definito.
+
+---
+
+## Come aggiungere un nuovo repo
+
+1. Definire il `topic` (aprire issue in ACB se non esiste)
+2. Scrivere lo script che produce il JSON (es. `scripts/build_pipeline_signals.py`)
+3. Aggiungere il file alla CI del repo (workflow che lo aggiorna e committa)
+4. Registrare la path in ACB (`dataciviclab.config.yml` o nel codice di `GitHubCollector`)
+5. Aggiungere il parser in `render.py` se il topic non è già supportato

src/agent_context_builder/github.py

@@ -104,6 +104,32 @@ def _get_repo_prs(self, repo: str, state: str = "open") -> list[PR]:
             )
         return prs
 
+    def get_raw_file(self, repo: str, path: str, ref: str = "main") -> str | None:
+        """Fetch raw file content from GitHub.
+
+        Uses raw.githubusercontent.com — works without token on public repos.
+        On failure, records the error in self.fetch_errors and returns None.
+
+        Args:
+            repo: Repository name (under self.org)
+            path: File path within the repo
+            ref: Branch or tag (default: main)
+
+        Returns:
+            Raw file content as string, or None on failure.
+        """
+        url = f"https://raw.githubusercontent.com/{self.org}/{repo}/{ref}/{path}"
+        headers = {}
+        if self.token:
+            headers["Authorization"] = f"token {self.token}"
+        try:
+            response = requests.get(url, headers=headers, timeout=10)
+            response.raise_for_status()
+            return response.text
+        except Exception as exc:
+            self.fetch_errors[f"{repo}:{path}"] = str(exc)
+            return None
+
     def _get_repo_issues(self, repo: str, state: str = "open") -> list[Issue]:
         """Get issues for a specific repo (excluding pull requests).
 

src/agent_context_builder/render.py

@@ -7,6 +7,16 @@
 from .discussions import Discussion, DiscussionCollector
 from .github import GitHubCollector, PR
 from .git_local import GitLocalCollector, GitState
+from .signals import (
+    RepoSignals,
+    SourceObservatorySignals,
+    parse_repo_signals,
+    parse_source_observatory_signals,
+)
+
+
+class _UNSET:
+    """Sentinel for uninitialized cache values."""
 
 
 class Renderer:
@@ -34,6 +44,9 @@ def __init__(
         self.git_collector = git_collector
         self.discussion_collector = discussion_collector
         self.fixed_timestamp = fixed_timestamp or datetime.now().isoformat()
+        # Cache for remote signal fetches — avoids double requests within one build
+        self._so_signals_cache: SourceObservatorySignals | None | type[_UNSET] = _UNSET
+        self._di_signals_cache: RepoSignals | None | type[_UNSET] = _UNSET
 
     def render_session_bootstrap(self) -> str:
         """Render session_bootstrap.md.
@@ -113,8 +126,65 @@ def render_session_bootstrap(self) -> str:
                 lines.append(f"- {topic_name}")
             lines.append("")
 
+        # Source health (only issues — skip stable sources)
+        so = self._fetch_source_observatory_signals()
+        lines += self._render_source_health_section(so)
+
+        # Pipeline state (only warn/error candidates)
+        di = self._fetch_di_pipeline_signals()
+        lines += self._render_pipeline_state_section(di)
+
         return "\n".join(lines)
 
+    def _fetch_source_observatory_signals(self) -> SourceObservatorySignals | None:
+        if self._so_signals_cache is not _UNSET:
+            return self._so_signals_cache  # type: ignore[return-value]
+        raw = self.github_collector.get_raw_file(
+            "source-observatory", "data/catalog/catalog_signals.json"
+        )
+        if raw is None:
+            self._so_signals_cache = None
+            return None
+        try:
+            result = parse_source_observatory_signals(raw)
+        except ValueError as exc:
+            self.github_collector.fetch_errors["source-observatory:catalog_signals"] = str(exc)
+            result = None
+        self._so_signals_cache = result
+        return result
+
+    def _render_source_health_section(
+        self, so: SourceObservatorySignals | None
+    ) -> list[str]:
+        lines = []
+        lines.append("## Source Health")
+        lines.append("")
+        if so is None:
+            err = self.github_collector.fetch_errors.get(
+                "source-observatory:data/catalog/catalog_signals.json"
+            ) or self.github_collector.fetch_errors.get(
+                "source-observatory:catalog_signals"
+            )
+            if err:
+                lines.append(f"> *catalog_signals unavailable — {err}*")
+            else:
+                lines.append("> *catalog_signals unavailable*")
+            lines.append("")
+            return lines
+
+        # Show regressions first, then other alerts (mutually exclusive sets)
+        issues = so.regressions + so.alerts
+        if issues:
+            for s in issues:
+                lines.append(f"- **{s.source}** ({s.protocol}): {s.result} — {s.detail}")
+                if s.suggested_action and s.suggested_action != "nessuna":
+                    lines.append(f"  - azione: {s.suggested_action}")
+        else:
+            lines.append(f"*All {so.sources_checked} sources stable* (as of {so.captured_at})")
+        lines.append(f"  *(captured {so.captured_at}, {so.sources_checked} sources checked)*")
+        lines.append("")
+        return lines
+
     def render_workspace_triage(self) -> dict[str, Any]:
         """Render workspace_triage.json.
 
@@ -186,9 +256,120 @@ def render_workspace_triage(self) -> dict[str, Any]:
                 for repo, state in repos_state.items()
             },
             "warnings": self._collect_warnings(prs, repos_state),
+            "source_health": self._build_source_health_dict(),
+            "pipeline_state": self._build_pipeline_state_dict(),
         }
         return triage
 
+    def _build_source_health_dict(self) -> dict[str, Any]:
+        so = self._fetch_source_observatory_signals()
+        if so is None:
+            return {
+                "available": False,
+                "errors": {
+                    k: v for k, v in self.github_collector.fetch_errors.items()
+                    if "source-observatory" in k
+                },
+            }
+        return {
+            "available": True,
+            "captured_at": so.captured_at,
+            "sources_checked": so.sources_checked,
+            "regressions": [
+                {
+                    "source": s.source,
+                    "protocol": s.protocol,
+                    "detail": s.detail,
+                    "suggested_action": s.suggested_action,
+                }
+                for s in so.regressions
+            ],
+            "alerts": [
+                {
+                    "source": s.source,
+                    "protocol": s.protocol,
+                    "signal_type": s.signal_type,
+                    "result": s.result,
+                    "detail": s.detail,
+                    "suggested_action": s.suggested_action,
+                }
+                for s in so.alerts
+            ],
+        }
+
+    def _fetch_di_pipeline_signals(self) -> RepoSignals | None:
+        if self._di_signals_cache is not _UNSET:
+            return self._di_signals_cache  # type: ignore[return-value]
+        raw = self.github_collector.get_raw_file(
+            "dataset-incubator", "registry/pipeline_signals.json"
+        )
+        if raw is None:
+            self._di_signals_cache = None
+            return None
+        try:
+            result = parse_repo_signals(raw)
+        except ValueError as exc:
+            self.github_collector.fetch_errors["dataset-incubator:pipeline_signals"] = str(exc)
+            result = None
+        self._di_signals_cache = result
+        return result
+
+    def _render_pipeline_state_section(self, di: RepoSignals | None) -> list[str]:
+        lines = []
+        lines.append("## Pipeline State")
+        lines.append("")
+        if di is None:
+            err = self.github_collector.fetch_errors.get(
+                "dataset-incubator:registry/pipeline_signals.json"
+            ) or self.github_collector.fetch_errors.get(
+                "dataset-incubator:pipeline_signals"
+            )
+            if err:
+                lines.append(f"> *pipeline_signals unavailable — {err}*")
+            else:
+                lines.append("> *pipeline_signals unavailable*")
+            lines.append("")
+            return lines
+
+        summary = di.summary
+        total = summary.get("total", len(di.signals))
+        by_status = summary.get("by_status", {})
+        actionable = di.actionable
+        if actionable:
+            for s in actionable:
+                lines.append(f"- **{s.label}** [{s.status}]: {s.detail}")
+                if s.action:
+                    lines.append(f"  - azione: {s.action}")
+        else:
+            lines.append(f"*{total} candidates, tutti ok*")
+        lines.append(
+            f"  *(as of {di.generated_at} — "
+            + ", ".join(f"{v} {k}" for k, v in sorted(by_status.items()) if v)
+            + ")*"
+        )
+        lines.append("")
+        return lines
+
+    def _build_pipeline_state_dict(self) -> dict[str, Any]:
+        di = self._fetch_di_pipeline_signals()
+        if di is None:
+            return {
+                "available": False,
+                "errors": {
+                    k: v for k, v in self.github_collector.fetch_errors.items()
+                    if "dataset-incubator" in k
+                },
+            }
+        return {
+            "available": True,
+            "generated_at": di.generated_at,
+            "summary": di.summary,
+            "actionable": [
+                {"id": s.id, "status": s.status, "detail": s.detail, "action": s.action}
+                for s in di.actionable
+            ],
+        }
+
     def _collect_warnings(
         self, prs: list[PR], repos_state: dict[str, GitState]
     ) -> list[str]: