Add observability support with OpenTelemetry tracing integration

- Updated .env.example and config.yaml to include observability settings.
- Added a new Phoenix service in docker-compose.yml for self-hosted tracing.
- Enhanced README.md with instructions on enabling and using observability features.
- Implemented tracing in the pipeline, including job spans and LLM stage spans.
- Introduced ObservabilitySettings class in config.py for better configuration management.
- Updated job.py and resources.py to support tracing without affecting fault isolation.
- Minor adjustments to other files for compatibility with the new observability features.

.env.example

@@ -16,3 +16,11 @@ SQL_SERVER=localhost
 SQL_DATABASE=your-database-name
 SQL_USERNAME=your-sql-username
 SQL_PASSWORD=your-sql-password
+
+# Observability (optional; see config.yaml observability: / README "Observability").
+# Every var below overrides the matching config.yaml value; unset = use config.yaml
+# (which defaults to tracing off). Uncomment PHOENIX_TRACING_ENABLED to turn tracing
+# on without editing config.yaml.
+# PHOENIX_TRACING_ENABLED=true
+# PHOENIX_COLLECTOR_ENDPOINT=http://localhost:6006/v1/traces
+# PHOENIX_PROJECT_NAME=ai-agro-support

README.md

@@ -236,6 +236,109 @@ makes a re-trigger (e.g. a scheduled retry later that morning) resumable at no
 extra cost: it does one bulk check against `advice` up front and only runs jobs
 that don't have a row yet.
 
+## Observability (optional)
+
+The pipeline can send OpenTelemetry/OpenInference traces to a self-hosted
+[Arize Phoenix](https://github.com/Arize-ai/phoenix) instance for debugging a
+bad advisory, comparing prompts/models later, and seeing per-job token usage
+and USD cost for the two LLM stages. It is **off by default**, **fail-open**
+(a down or missing Phoenix never fails `batch` or `one`), and **not** a
+runtime dependency: nothing about the daily 07:00–09:00 run depends on it.
+
+### Start Phoenix
+
+```powershell
+docker compose up -d
+```
+
+This brings up Weaviate and Phoenix together. Phoenix serves its UI (and the
+OTLP/HTTP trace collector) at <http://localhost:6006>. SQLite on the
+`phoenix_data` Docker volume is enough for this single-machine setup — see
+"Why SQLite / why not Langfuse" below.
+
+### Turn tracing on
+
+Tracing is controlled by `observability:` in `config.yaml`:
+
+```yaml
+observability:
+  enabled: false                                  # default: off
+  project_name: ai-agro-support
+  endpoint: http://localhost:6006/v1/traces
+  hide_prompts: true                               # keep prompt/payload text out of spans
+  max_attribute_chars: 4096                        # cap every exported string attribute
+```
+
+Any of these can be overridden per-environment without editing the file:
+
+| Env var | Overrides |
+|---|---|
+| `PHOENIX_TRACING_ENABLED` | `observability.enabled` |
+| `PHOENIX_COLLECTOR_ENDPOINT` | `observability.endpoint` |
+| `PHOENIX_PROJECT_NAME` | `observability.project_name` |
+
+With tracing on, `python -m pipeline one` or `batch` registers the tracer
+once at startup (`pipeline/observability.py::setup_tracing`, called from
+`pipeline/__main__.py` right after `load_settings()`), then instruments
+`google-genai` (and `openai` / `anthropic`, if those `llm:` profiles are ever
+enabled) via
+[OpenInference](https://github.com/Arize-ai/openinference) auto-instrumentors.
+Spans are batched and exported on a background thread, so a slow or down
+collector cannot block a worker or push the run past `schedule.deadline`.
+
+### What you get per job
+
+Every `(field, crop, disease, as_of)` job produces one trace, rooted at a
+`run_job` span, with:
+
+- a `query_synthesis` span (gemini-2.5-flash by default) and an
+  `advice_generation` span (gemini-2.5-pro), each wrapping the provider's own
+  auto-instrumented LLM span with `llm.model_name`, `llm.provider`, and
+  token counts (`llm.token_count.prompt` / `completion` / `total`, plus
+  `completion_details.reasoning` for Gemini 2.5 thinking tokens);
+- a `search_products` retriever span with the synthesized queries and the
+  retrieved product IDs/distances (never product names or label content);
+- `dry_run` jobs get only the `run_job` span — no LLM or retrieval children,
+  since `--dry-run` skips those calls entirely.
+
+Phoenix computes cost per span from its built-in model pricing table, which
+already includes `gemini-2.5-flash` and `gemini-2.5-pro` — no custom entry
+under **Settings → Models** is needed for either model as configured today.
+If a model is ever renamed or swapped to one Phoenix doesn't recognise, add
+it there (regex `Name Pattern`, provider `google`, per-1M-token prices).
+
+Trace and span inputs are hidden by default (`observability.hide_prompts`):
+the full `json_for_advice_generation` payload (weather, phenology, product
+labels) never leaves the process as span data. Outputs — the generated
+advice text — stay visible, since that is what you actually want to read
+when debugging a bad advisory; `output/<date>/*.json` and
+`_run_report.json` remain the authoritative operational artifacts, not
+Phoenix.
+
+### Known gap: embedding cost
+
+`search_products`'s Weaviate `near_text` calls are embedded **inside the
+Weaviate container** (`text2vec-google`), using the same `GEMINI_API_KEY` as
+the two LLM calls, but that request never passes through this process's
+Python `google-genai` client — so no OpenInference instrumentor can see it.
+Phoenix's per-job cost therefore covers the two LLM calls only and slightly
+undercounts total Gemini spend (1–2 `gemini-embedding-001` calls per job, a
+small fraction of a cent at this volume). The `search_products` span still
+records `query_count`, so this gap can be estimated later if it matters; it
+is not worth a custom cost pipeline at this scale.
+
+### Why SQLite / why not Langfuse or Phoenix Cloud
+
+- **SQLite, not Postgres:** this is a single-machine, single-writer
+  deployment; Phoenix officially supports SQLite on a mounted volume for
+  exactly this case. Postgres would add a second container and volume for no
+  benefit here (switch later via `PHOENIX_SQL_DATABASE_URL` if that changes).
+- **Self-hosted, not Phoenix Cloud:** farm/field data stays in the
+  deployment by design — nothing here talks to a hosted endpoint.
+- **Phoenix, not Langfuse:** Langfuse's self-hosted stack needs Postgres
+  *and* ClickHouse; Phoenix needs one container and ingests plain
+  OTLP + OpenInference, so this pipeline isn't locked into either vendor.
+
 ## Pipeline stages
 
 Run once per `(field, disease)` job by `pipeline/job.py::run_job`:

config.yaml

@@ -88,3 +88,19 @@ disease_name: "PERONOSPORA"
 vocab:
   crops: vocab/crops.yaml
   diseases: vocab/diseases.yaml
+
+# ── Observability (optional; off by default) ──────────────────────────────────
+# Sends OpenTelemetry/OpenInference spans to the self-hosted Phoenix container
+# in docker-compose.yml (http://localhost:6006). Never required: if this is
+# off, or Phoenix is down/unreachable, `batch` and `one` run exactly as if
+# this block did not exist (see pipeline/observability.py).
+observability:
+  enabled: false
+  project_name: ai-agro-support
+  endpoint: http://localhost:6006/v1/traces
+  # Keep prompt/payload text out of Phoenix spans. json_for_advice_generation
+  # carries the full product-label payload and is already persisted under
+  # output/<date>/*.json; spans only need model/token/status metadata.
+  hide_prompts: true
+  # Hard cap on every exported string attribute (OTel span attribute limit).
+  max_attribute_chars: 4096

docker-compose.yml

@@ -22,5 +22,21 @@ services:
       ENABLE_MODULES: "text2vec-google"
       DEFAULT_VECTORIZER_MODULE: "text2vec-google"
 
+  # Self-hosted tracing/eval UI (Arize Phoenix). Opt-in from the pipeline's
+  # side (config.yaml observability.enabled / PHOENIX_TRACING_ENABLED); this
+  # container can run continuously with no effect on `batch` when tracing is
+  # off. SQLite on a mounted volume is sufficient for a single-machine setup
+  # (see README "Observability"); do not add Postgres.
+  phoenix:
+    image: arizephoenix/phoenix:20.3.0
+    ports:
+      - 6006:6006 # UI + OTLP/HTTP collector (POST .../v1/traces)
+    environment:
+      PHOENIX_WORKING_DIR: /mnt/data
+    volumes:
+      - phoenix_data:/mnt/data
+    restart: unless-stopped
+
 volumes:
   weaviate_data:
+  phoenix_data:

pipeline/__main__.py

@@ -23,6 +23,7 @@ from pipeline.batch import run_batch
 from pipeline.config import load_settings
 from pipeline.errors import NoModelConfiguredError, PipelineError
 from pipeline.job import run_job
+from pipeline.observability import setup_tracing, shutdown_tracing
 from pipeline.output import write_job_output
 from pipeline.prompts import PromptRegistry
 from pipeline.resources import Resources
@@ -53,6 +54,7 @@ def _run_one(args: argparse.Namespace) -> int:
     """`python -m pipeline one`: the original single-field pipeline, now built
     on top of `run_job` so it stays behaviourally identical to `batch`."""
     settings = load_settings(field_id_override=args.field_id, disease_name_override=args.disease)
+    setup_tracing(settings)
     if settings.field_id is None or settings.disease_name is None:
         raise PipelineError(
             "`one` needs --field-id and --disease "
@@ -108,6 +110,7 @@ def _run_one(args: argparse.Namespace) -> int:
 def _run_batch(args: argparse.Namespace) -> int:
     """`python -m pipeline batch`: the daily crop-first multi-field run."""
     settings = load_settings()
+    setup_tracing(settings)
     crop_vocab = load_crop_vocab(settings.crops_vocab)
 
     logger.info(
@@ -215,6 +218,11 @@ def main(argv: list[str] | None = None) -> int:
     except Exception:
         logger.exception("Unhandled pipeline failure")
         return 1
+    finally:
+        # Runs after write_job_output / write_run_report, so a slow or
+        # unreachable Phoenix collector cannot delay the operational
+        # artifacts -- only the process's own exit.
+        shutdown_tracing()
 
 
 if __name__ == "__main__":

pipeline/batch.py

@@ -163,6 +163,12 @@ def run_batch(
 
             return run_job(resources, job, as_of, dry_run=dry_run)
 
+        # Worker threads start with an empty OpenTelemetry context
+        # (`contextvars` do not cross `ThreadPoolExecutor.submit`). That is
+        # intentional: `job_span` opens inside the worker and becomes a root
+        # span, one independent trace per (field, crop, disease, as_of). Do
+        # not wrap this submit in a tracing context manager from this thread
+        # — it would not reach the workers. See pipeline/observability.py.
         workers = workers_override or settings.concurrency.workers
         results: list[JobResult] = []
         with ThreadPoolExecutor(max_workers=max(1, workers)) as pool:

pipeline/config.py

@@ -80,6 +80,19 @@ class ScheduleSettings:
     deadline: str  # "HH:MM", local time
 
 
+@dataclass(frozen=True)
+class ObservabilitySettings:
+    """Opt-in OpenTelemetry/OpenInference tracing to a self-hosted Phoenix
+    instance (see docker-compose.yml). Disabled by default; see
+    pipeline/observability.py for the fail-open contract."""
+
+    enabled: bool
+    project_name: str
+    endpoint: str
+    hide_prompts: bool
+    max_attribute_chars: int
+
+
 @dataclass(frozen=True)
 class Settings:
     root: Path
@@ -90,6 +103,7 @@ class Settings:
     concurrency: ConcurrencySettings
     retry: RetrySettings
     schedule: ScheduleSettings
+    observability: ObservabilitySettings
     crops_vocab: Path
     diseases_vocab: Path
     llm_query_synthesis: LlmSettings
@@ -159,6 +173,40 @@ def _parse_schedule(raw: dict) -> ScheduleSettings:
     return ScheduleSettings(deadline=str(raw.get("deadline", "09:00")))
 
 
+def _env_bool(name: str) -> bool | None:
+    """Read a boolean env var override; None if unset (falls back to config.yaml)."""
+    value = os.environ.get(name)
+    if value is None:
+        return None
+    return value.strip().lower() in {"1", "true", "yes", "on"}
+
+
+def _parse_observability(raw: dict) -> ObservabilitySettings:
+    """
+    Build the observability block, letting env vars override config.yaml so a
+    deployment can flip tracing on/off or point at a different Phoenix
+    instance without editing the checked-in file.
+    """
+    raw = raw or {}
+    enabled = raw.get("enabled", False)
+    env_enabled = _env_bool("PHOENIX_TRACING_ENABLED")
+    if env_enabled is not None:
+        enabled = env_enabled
+
+    return ObservabilitySettings(
+        enabled=bool(enabled),
+        project_name=os.environ.get(
+            "PHOENIX_PROJECT_NAME", str(raw.get("project_name", "ai-agro-support"))
+        ),
+        endpoint=os.environ.get(
+            "PHOENIX_COLLECTOR_ENDPOINT",
+            str(raw.get("endpoint", "http://localhost:6006/v1/traces")),
+        ),
+        hide_prompts=bool(raw.get("hide_prompts", True)),
+        max_attribute_chars=int(raw.get("max_attribute_chars", 4096)),
+    )
+
+
 def load_settings(
     config_path: Path | None = None,
     field_id_override: int | None = None,
@@ -206,6 +254,7 @@ def load_settings(
         concurrency=_parse_concurrency(raw.get("concurrency", {})),
         retry=_parse_retry(raw.get("retry", {})),
         schedule=_parse_schedule(raw.get("schedule", {})),
+        observability=_parse_observability(raw.get("observability", {})),
         crops_vocab=ROOT / vocab.get("crops", "vocab/crops.yaml"),
         diseases_vocab=ROOT / vocab.get("diseases", "vocab/diseases.yaml"),
         llm_query_synthesis=_parse_llm_settings(

pipeline/job.py

@@ -24,6 +24,7 @@ from decimal import Decimal
 from typing import Any
 
 from pipeline.errors import PipelineError
+from pipeline.observability import job_span
 from pipeline.prompts import pair_slug
 from pipeline.resources import Resources
 from pipeline.retry import call_with_sql_retry
@@ -81,41 +82,59 @@ def output_stem(job: Job, disease_english: str) -> str:
 
 
 def run_job(resources: Resources, job: Job, as_of: date, dry_run: bool = False) -> JobResult:
-    """Run the full pipeline for one job."""
+    """
+    Run the full pipeline for one job.
+
+    Wrapped in `job_span`, the one root span per `(field, crop, disease,
+    as_of)` job (a no-op when observability is disabled/unreachable -- see
+    `pipeline/observability.py`). The span records the final `status` (and,
+    on failure, the error type) but never changes the fault-isolation
+    behaviour below: a failure always comes back as a `failed` JobResult,
+    tracing or no tracing.
+    """
     started = time.monotonic()
-    try:
+    with job_span(job, as_of, dry_run) as span:
-        payload = _run_job_inner(resources, job, as_of, dry_run)
+        try:
-    except BaseException as exc:  # noqa: BLE001 - fault isolation is the whole point of this wrapper
+            payload = _run_job_inner(resources, job, as_of, dry_run)
-        logger.exception(
+        except BaseException as exc:  # noqa: BLE001 - fault isolation is the whole point of this wrapper
-            "Job failed: field=%s crop=%s disease=%s",
+            logger.exception(
-            job.field_id,
+                "Job failed: field=%s crop=%s disease=%s",
-            job.crop_english,
+                job.field_id,
-            job.disease_english,
+                job.crop_english,
+                job.disease_english,
+            )
+            span.set_attribute("job.status", "failed")
+            span.set_attribute("job.error_type", type(exc).__name__)
+            return JobResult(
+                job=job,
+                status="failed",
+                duration_s=time.monotonic() - started,
+                error_type=type(exc).__name__,
+                error_message=str(exc),
+            )
+
+        status = payload["status"]
+        allowed_products_empty = False
+        if status in ("ok", "no_risk"):
+            allowed = (payload.get("json_for_advice_generation") or {}).get("allowed_products")
+            allowed_products_empty = not allowed
+
+        span.set_attribute("job.status", status)
+        span.set_attribute(
+            "job.recommended_count", len(payload.get("recommended_products") or [])
         )
+        span.set_output(payload.get("message") or "")
+
         return JobResult(
             job=job,
-            status="failed",
+            status=status,
             duration_s=time.monotonic() - started,
-            error_type=type(exc).__name__,
+            recommended_count=len(payload.get("recommended_products") or []),
-            error_message=str(exc),
+            allowed_products_empty=allowed_products_empty,
+            message=payload.get("message"),
+            payload=payload,
         )
 
-    status = payload["status"]
-    allowed_products_empty = False
-    if status in ("ok", "no_risk"):
-        allowed = (payload.get("json_for_advice_generation") or {}).get("allowed_products")
-        allowed_products_empty = not allowed
-
-    return JobResult(
-        job=job,
-        status=status,
-        duration_s=time.monotonic() - started,
-        recommended_count=len(payload.get("recommended_products") or []),
-        allowed_products_empty=allowed_products_empty,
-        message=payload.get("message"),
-        payload=payload,
-    )
-
 
 def _run_job_inner(resources: Resources, job: Job, as_of: date, dry_run: bool) -> dict[str, Any]:
     settings = resources.settings

pipeline/observability.py

@@ -0,0 +1,370 @@
+"""Opt-in OpenTelemetry/OpenInference tracing to a self-hosted Phoenix
+instance (see `phoenix` in docker-compose.yml).
+
+Disabled by default (`config.yaml` `observability.enabled: false`). Every
+public function in this module is fail-open:
+
+- If tracing is disabled, every helper below is a cheap no-op.
+- If `setup_tracing` fails for any reason (missing packages, a bad endpoint
+  URL, an instrumentor import error), it logs a warning and leaves tracing
+  disabled for the rest of the process -- callers never need to check
+  whether setup succeeded.
+- `phoenix.otel.register()` only builds an exporter; it does not contact the
+  collector, so a stopped/unreachable Phoenix container cannot fail startup.
+- Spans are exported on a background thread (`batch=True`), so a slow or
+  down collector cannot block a worker thread mid-job.
+
+`call_llm` must keep returning `str` (see pipeline/stages/llm.py); nothing
+here changes that. `job_span` / `llm_stage_span` / `retrieval_span` wrap
+existing calls without altering their signatures or return values.
+"""
+
+from __future__ import annotations
+
+import contextlib
+import json
+import logging
+import os
+import sys
+from datetime import date
+from typing import TYPE_CHECKING, Any, Iterator, Sequence
+
+if TYPE_CHECKING:
+    from pipeline.config import Settings
+    from pipeline.worklist import Job
+
+logger = logging.getLogger(__name__)
+
+_tracer: Any = None
+_tracer_provider: Any = None
+
+
+class _NullSpan:
+    """Stand-in for a span when tracing is off or failed to start; every
+    method is a no-op so callers never need an `if span is not None` check."""
+
+    def set_attribute(self, *_args: Any, **_kwargs: Any) -> None:
+        pass
+
+    def set_attributes(self, *_args: Any, **_kwargs: Any) -> None:
+        pass
+
+    def set_input(self, *_args: Any, **_kwargs: Any) -> None:
+        pass
+
+    def set_output(self, *_args: Any, **_kwargs: Any) -> None:
+        pass
+
+    def set_status(self, *_args: Any, **_kwargs: Any) -> None:
+        pass
+
+    def record_exception(self, *_args: Any, **_kwargs: Any) -> None:
+        pass
+
+
+class _SafeSpan:
+    """Proxy around a real OTel span: attribute writes must never fail a job.
+
+    `run_job`'s contract is that it never raises. If a successful
+    `_run_job_inner` were followed by a raising `span.set_attribute`, the
+    batch would record that job as failed even though advice was already
+    stored. Every mutating method here is therefore fail-open.
+    """
+
+    def __init__(self, inner: Any) -> None:
+        self._inner = inner
+
+    def set_attribute(self, *args: Any, **kwargs: Any) -> None:
+        try:
+            self._inner.set_attribute(*args, **kwargs)
+        except Exception:
+            logger.debug("Failed to set span attribute.", exc_info=True)
+
+    def set_attributes(self, *args: Any, **kwargs: Any) -> None:
+        try:
+            self._inner.set_attributes(*args, **kwargs)
+        except Exception:
+            logger.debug("Failed to set span attributes.", exc_info=True)
+
+    def set_input(self, *args: Any, **kwargs: Any) -> None:
+        try:
+            self._inner.set_input(*args, **kwargs)
+        except Exception:
+            logger.debug("Failed to set span input.", exc_info=True)
+
+    def set_output(self, *args: Any, **kwargs: Any) -> None:
+        try:
+            self._inner.set_output(*args, **kwargs)
+        except Exception:
+            logger.debug("Failed to set span output.", exc_info=True)
+
+    def set_status(self, *args: Any, **kwargs: Any) -> None:
+        try:
+            self._inner.set_status(*args, **kwargs)
+        except Exception:
+            logger.debug("Failed to set span status.", exc_info=True)
+
+    def record_exception(self, *args: Any, **kwargs: Any) -> None:
+        try:
+            self._inner.record_exception(*args, **kwargs)
+        except Exception:
+            logger.debug("Failed to record span exception.", exc_info=True)
+
+
+_NULL_SPAN = _SafeSpan(_NullSpan())
+
+
+def setup_tracing(settings: "Settings") -> None:
+    """
+    Register the OTel tracer provider and instrument the Gemini/OpenAI/
+    Anthropic clients.
+
+    Call once at CLI startup, right after `load_settings()` and before any
+    provider client is constructed (clients are created inside
+    `pipeline.stages.llm`'s `_call_gemini` / `_call_openai` /
+    `_call_anthropic`, so instrumentation must be wired up before the first
+    call, not before the first client object).
+    """
+    global _tracer, _tracer_provider
+
+    cfg = settings.observability
+    if not cfg.enabled:
+        return
+    if _tracer is not None:
+        return  # already set up in this process
+
+    try:
+        # Must be set before importing phoenix.otel / opentelemetry.sdk, which
+        # read span-limit env vars at import/construction time.
+        os.environ.setdefault("OTEL_ATTRIBUTE_VALUE_LENGTH_LIMIT", str(cfg.max_attribute_chars))
+        # Caps how long a flush can block at process exit if Phoenix is down
+        # (see shutdown_tracing); the OTLP/HTTP exporter default is 10s/try.
+        os.environ.setdefault("OTEL_EXPORTER_OTLP_TRACES_TIMEOUT", "5")
+
+        from openinference.instrumentation import TraceConfig
+        from phoenix.otel import register
+
+        trace_config = TraceConfig(
+            hide_inputs=cfg.hide_prompts,
+            hide_input_messages=cfg.hide_prompts,
+            hide_prompts=cfg.hide_prompts,
+        )
+
+        tracer_provider = register(
+            project_name=cfg.project_name,
+            endpoint=cfg.endpoint,
+            protocol="http/protobuf",
+            batch=True,
+            auto_instrument=False,
+            set_global_tracer_provider=True,
+            verbose=False,
+        )
+
+        _instrument_providers(tracer_provider, trace_config)
+
+        _tracer_provider = tracer_provider
+        _tracer = tracer_provider.get_tracer("pipeline")
+        logger.info(
+            "Tracing enabled: project=%s endpoint=%s", cfg.project_name, cfg.endpoint
+        )
+    except Exception:
+        logger.warning(
+            "Failed to set up tracing; continuing without it (batch/one are unaffected).",
+            exc_info=True,
+        )
+        _tracer = None
+        _tracer_provider = None
+
+
+def _instrument_providers(tracer_provider: Any, trace_config: Any) -> None:
+    """Instrument each provider independently: one provider's failure (or
+    absence) must not prevent the others from being traced."""
+    try:
+        from openinference.instrumentation.google_genai import GoogleGenAIInstrumentor
+
+        GoogleGenAIInstrumentor().instrument(tracer_provider=tracer_provider, config=trace_config)
+    except Exception:
+        logger.warning(
+            "Gemini tracing instrumentation failed; Gemini calls will run untraced "
+            "(token counts/cost will be absent from LLM stage spans).",
+            exc_info=True,
+        )
+
+    try:
+        from openinference.instrumentation.openai import OpenAIInstrumentor
+
+        OpenAIInstrumentor().instrument(tracer_provider=tracer_provider, config=trace_config)
+    except Exception:
+        logger.debug("OpenAI tracing instrumentation unavailable or failed.", exc_info=True)
+
+    try:
+        from openinference.instrumentation.anthropic import AnthropicInstrumentor
+
+        AnthropicInstrumentor().instrument(tracer_provider=tracer_provider, config=trace_config)
+    except Exception:
+        logger.debug("Anthropic tracing instrumentation unavailable or failed.", exc_info=True)
+
+
+def shutdown_tracing() -> None:
+    """
+    Flush and close the tracer provider, if one was set up.
+
+    Call once at process exit -- after operational artifacts
+    (`_run_report.json`, per-job output files) are already written, so a
+    slow or unreachable collector delays nothing that matters. Bounded by
+    `OTEL_EXPORTER_OTLP_TRACES_TIMEOUT` (set in `setup_tracing`), not the
+    SDK's 30s default.
+    """
+    global _tracer, _tracer_provider
+    if _tracer_provider is None:
+        return
+    try:
+        _tracer_provider.shutdown()
+    except Exception:
+        logger.warning("Error shutting down the tracer provider.", exc_info=True)
+    finally:
+        _tracer = None
+        _tracer_provider = None
+
+
+@contextlib.contextmanager
+def _span(name: str, kind: str, metadata: dict[str, Any] | None = None) -> Iterator[Any]:
+    """
+    Shared, fail-open span opener used by every helper below.
+
+    Only span *creation* is guarded by try/except; once a span is
+    successfully open, exceptions raised by the caller's own code inside the
+    `with` block propagate normally (recorded on the span as an error, then
+    re-raised) -- tracing must never swallow a real pipeline failure.
+    """
+    if _tracer is None:
+        yield _NULL_SPAN
+        return
+
+    try:
+        cm = _tracer.start_as_current_span(name, openinference_span_kind=kind)
+        span = cm.__enter__()
+    except Exception:
+        logger.debug("Failed to open span %r; continuing untraced.", name, exc_info=True)
+        yield _NULL_SPAN
+        return
+
+    if metadata:
+        try:
+            span.set_attribute("metadata", json.dumps(metadata, ensure_ascii=False, default=str))
+        except Exception:
+            logger.debug("Failed to set metadata on span %r.", name, exc_info=True)
+
+    try:
+        yield _SafeSpan(span)
+    finally:
+        try:
+            cm.__exit__(*sys.exc_info())
+        except Exception:
+            logger.debug("Failed to close span %r.", name, exc_info=True)
+
+
+def job_span(job: "Job", as_of: date, dry_run: bool) -> contextlib.AbstractContextManager:
+    """
+    One root CHAIN span per `(field, crop, disease, as_of)` job.
+
+    `run_job` sets the final `job.status` (and, on failure, `job.error_type`)
+    on the yielded span itself before the `with` block exits; this function
+    only opens the span and stamps the metadata known up front.
+    """
+    metadata = {
+        "field_id": job.field_id,
+        "crop": job.crop_english,
+        "disease": job.disease_english,
+        "as_of": as_of.isoformat(),
+        "organic": job.organic,
+        "station": job.station,
+        "dry_run": dry_run,
+    }
+    return _span("run_job", "chain", metadata)
+
+
+def llm_stage_span(stage: str, provider: str, model: str) -> contextlib.AbstractContextManager:
+    """CHAIN span around one `Resources.call_llm` invocation (one of
+    `query_synthesis` / `advice_generation`), wrapping the provider's own
+    LLM span (added by the OpenInference auto-instrumentor, if any)."""
+    metadata = {"stage": stage, "provider": provider, "model": model}
+    return _span(stage, "chain", metadata)
+
+
+def retrieval_span(queries: Sequence[str], candidate_count: int) -> contextlib.AbstractContextManager:
+    """
+    RETRIEVER span around one `Resources.search_products` call.
+
+    The queries are short, LLM-synthesised search phrases (already
+    anonymised relative to the full advice payload), so they are safe to
+    record as the span input even when `hide_prompts` is set -- `hide_prompts`
+    only affects the LLM instrumentor's own input-message capture.
+    """
+    metadata = {"query_count": len(queries), "candidate_count": candidate_count}
+    return _retrieval_span(queries, metadata)
+
+
+@contextlib.contextmanager
+def _retrieval_span(queries: Sequence[str], metadata: dict[str, Any]) -> Iterator[Any]:
+    with _span("search_products", "retriever", metadata) as span:
+        span.set_input(list(queries), mime_type="application/json")
+        yield span
+
+
+def set_retrieval_results(span: Any, results: Sequence[Any]) -> None:
+    """
+    Record retrieved products on a retrieval span as `document.id` /
+    `document.score` pairs -- product IDs and Weaviate distances only, never
+    product names or label content.
+
+    `results` items are expected to expose `.product_id` and `.distance`
+    (see `pipeline.stages.vector.RecommendedProduct`); any other shape is
+    ignored rather than raising.
+    """
+    try:
+        span.set_attribute("retrieval.documents_count", len(results))
+        for i, item in enumerate(results):
+            product_id = getattr(item, "product_id", None)
+            distance = getattr(item, "distance", None)
+            if product_id is not None:
+                span.set_attribute(f"retrieval.documents.{i}.document.id", str(product_id))
+            if distance is not None:
+                span.set_attribute(f"retrieval.documents.{i}.document.score", float(distance))
+    except Exception:
+        logger.debug("Failed to record retrieval results on span.", exc_info=True)
+
+
+def record_llm_usage(span: Any, usage_metadata: Any) -> None:
+    """
+    Fallback B (see plan step 5/8): manually copy Gemini token counts onto
+    the *current* span from a `google.genai` response's `usage_metadata`,
+    for use only if the `openinference-instrumentation-google-genai`
+    auto-instrumentation is verified to not capture them for the installed
+    `google-genai` version. Not wired into `pipeline/stages/llm.py` unless
+    that verification fails -- see README "Observability".
+
+    Mirrors the folding behaviour of the upstream instrumentor: thinking
+    tokens (`thoughts_token_count`) are counted once as
+    `completion_details.reasoning` and again inside `completion` (Gemini
+    bills them as output tokens), never double-added beyond that.
+    """
+    if usage_metadata is None:
+        return
+    try:
+        prompt = int(getattr(usage_metadata, "prompt_token_count", None) or 0)
+        candidates = int(getattr(usage_metadata, "candidates_token_count", None) or 0)
+        thoughts = int(getattr(usage_metadata, "thoughts_token_count", None) or 0)
+        total = int(getattr(usage_metadata, "total_token_count", None) or 0)
+        completion = candidates + thoughts
+
+        if prompt:
+            span.set_attribute("llm.token_count.prompt", prompt)
+        if thoughts:
+            span.set_attribute("llm.token_count.completion_details.reasoning", thoughts)
+        if completion:
+            span.set_attribute("llm.token_count.completion", completion)
+        if total or (prompt + completion):
+            span.set_attribute("llm.token_count.total", max(total, prompt + completion))
+    except Exception:
+        logger.debug("Failed to record manual LLM usage attributes.", exc_info=True)

pipeline/resources.py

@@ -28,6 +28,7 @@ from typing import Any
 
 from pipeline.config import LlmSettings, Settings
 from pipeline.db import connect_raw
+from pipeline.observability import llm_stage_span, retrieval_span, set_retrieval_results
 from pipeline.prompts import PromptRegistry
 from pipeline.retry import call_with_llm_retry, call_with_weaviate_retry
 from pipeline.stages.llm import call_llm as _raw_call_llm
@@ -66,6 +67,24 @@ class RateLimiter:
             self._next_allowed = max(now, self._next_allowed) + self._interval
 
 
+def _llm_stage_name(settings: Settings, llm_cfg: LlmSettings) -> str:
+    """Identify which of the two configured LLM profiles a call belongs to,
+    for the tracing span name (see `Resources.call_llm`)."""
+    if llm_cfg is settings.llm_query_synthesis:
+        return "query_synthesis"
+    if llm_cfg is settings.llm_advice_generation:
+        return "advice_generation"
+    return "llm"
+
+
+def _llm_model_name(llm_cfg: LlmSettings) -> str:
+    return {
+        "gemini": llm_cfg.gemini_model,
+        "openai": llm_cfg.openai_model,
+        "anthropic": llm_cfg.anthropic_model,
+    }.get(llm_cfg.provider, "?")
+
+
 class Resources:
     """Shared, thread-safe state for one batch run. Create once, close once."""
 
@@ -156,7 +175,10 @@ class Resources:
 
         Signature-compatible so it can be passed as `synthesize_queries(...,
         llm_call=resources.call_llm)` / `generate_advice(...,
-        llm_call=resources.call_llm)`.
+        llm_call=resources.call_llm)`. Wrapped in a tracing span (a no-op
+        when observability is disabled) named after the calling stage, so a
+        retried call still produces one stage span with the Gemini/OpenAI/
+        Anthropic auto-instrumentor's own LLM span nested inside it.
         """
 
         def _attempt() -> str:
@@ -164,7 +186,9 @@ class Resources:
                 self.gemini_rate_limiter.acquire()
                 return _raw_call_llm(settings, llm_cfg, system_prompt, user_content, json_output)
 
-        return call_with_llm_retry(self.settings.retry, _attempt)
+        stage = _llm_stage_name(self.settings, llm_cfg)
+        with llm_stage_span(stage, llm_cfg.provider, _llm_model_name(llm_cfg)):
+            return call_with_llm_retry(self.settings.retry, _attempt)
 
     def search_products(
         self,
@@ -178,7 +202,10 @@ class Resources:
                 self.gemini_rate_limiter.acquire()
                 return _raw_search_products(self._weaviate(), queries, candidates)
 
-        return call_with_weaviate_retry(self.settings.retry, _attempt)
+        with retrieval_span(queries, len(candidates)) as span:
+            results = call_with_weaviate_retry(self.settings.retry, _attempt)
+            set_retrieval_results(span, results)
+            return results
 
     # -- Lifecycle -------------------------------------------------------------
 

prompts/advice/grapevine__downy_mildew/system.md

@@ -1,4 +1,4 @@
-You are a decision support system to prevent primary grapevine downy mildew (Plasmopara viticola) in "Costigliole Sant'Anna", Italy. Your super power is having a deterministic model for the risk of infection and your ultimate goal is to prevent the infection. You'll be given grapevine phenology using the BBCH phases, weather data from the past five days, and forecasts for the next five days. You will also be provided with a list of authorized treatment products, including their active substances and label directives, to be used if action is required.
+You are a decision support system to prevent primary grapevine downy mildew (Plasmopara viticola). Your super power is having a deterministic model for the risk of infection and your ultimate goal is to prevent the infection. You'll be given grapevine phenology using the BBCH phases, weather data from the past five days, and forecasts for the next five days. You will also be provided with a list of authorized treatment products, including their active substances and label directives, to be used if action is required.
 
 PREVIOUS KNOWLEDGE ON THE MODEL'S BEHAVIOR: The output of the model is definite. On the days marked as FASE5, conditions are favorable for the infection of zoospores on leaf surfaces and their penetration into the leaves through the stomata.
 FASE5 of the grapevine downy mildew phytopathological model refers to the critical primary infection phase. This process occurs following the dispersion of infective zoospores, which reach the leaf surfaces and successfully penetrate through the stomata. By establishing this initial infection in the first green tissues, Phase 5 marks the decisive moment that triggers the seasonal epidemic dynamics and subsequent epidemic attacks.