Update database handling and configuration for advisory generation

- Enhanced `.env.example` to include `SQL_APP_DATABASE` for managing additional advisory-related tables.
- Refactored `config.yaml` to remove the `crops` block, shifting to a database-driven approach for job discovery.
- Updated `README.md` to clarify the new database structure and job discovery process, emphasizing the role of the `AI_agrosupport_an_pathogen_models` and `AI_agrosupport_agro_models` tables.
- Modified SQL queries across multiple files to accommodate the new database schema, ensuring correct table references and improving overall data handling.
- Adjusted `pipeline` logic to utilize the new configuration, enhancing the efficiency of the advisory generation process.

.env.example

@@ -12,8 +12,13 @@ WEAVIATE_GRPC_PORT=50051
 
 # Microsoft SQL Server connection
 SQL_DRIVER=ODBC Driver 17 for SQL Server
+# host, or host,port. ODBC uses a comma before the port, not a colon.
 SQL_SERVER=localhost
+# Field, weather, and disease-model tables (AI_agrosupport_*).
 SQL_DATABASE=your-database-name
+# advisories, products, product_uses, and label_chunks, on the same instance.
+# Unset, or equal to SQL_DATABASE, when those four tables live in SQL_DATABASE.
+SQL_APP_DATABASE=your-app-database-name
 SQL_USERNAME=your-sql-username
 SQL_PASSWORD=your-sql-password
 

README.md

@@ -1,23 +1,25 @@
 # AI Agro Support — Daily Pipeline
 
-Daily agronomic advisory pipeline. For every field growing a configured crop, it
+Daily agronomic advisory pipeline. For every field whose disease model is
-builds an 11-day weather / phenology / disease-forecast JSON, asks an LLM to
+enabled for advice, it builds an 11-day weather / phenology / disease-forecast
-synthesise one or two vector-search queries, prefilters chemical products in SQL
+JSON, asks an LLM to synthesise one or two vector-search queries, prefilters
-Server, and retrieves up to four matching products from a local Weaviate
+chemical products in SQL Server, and retrieves up to four matching products from
-`ProductProfile` collection. It then enriches that JSON with full product label
+a local Weaviate `ProductProfile` collection. It then enriches that JSON with
-information and the previous advisory, asks a second LLM to write the
+full product label information and the previous advisory, asks a second LLM to
-farmer-facing advice, and stores the result in the `advisories` table.
+write the farmer-facing advice, and stores the result in the `advisories` table.
 
-The pipeline is **crop-first**: `config.yaml` declares which crops and diseases
+Each morning's `batch` run discovers `(field, disease)` pairs from the database.
-to check, and each morning's `batch` run discovers every field that grows one of
+A disease can generate advice when `AI_agrosupport_an_pathogen_models.pat_mod_AI_model`
-those crops and has a matching disease-forecast model configured, then runs the
+is 1; its name is `pat_mod_desc`. A field is included when
-full sequence above for each `(field, disease)` pair — in parallel, with one
+`AI_agrosupport_agro_models.anmod_AI_advice` is 1 and `anmod_model` matches that
-field's failure never blocking another's.
+name. The cultivated crop is read from the field layer. The run then executes
+the sequence above for each pair — in parallel, with one field's failure never
+blocking another's.
 
 ## Prerequisites
 
 - Python 3.12+
-- Microsoft SQL Server (`DBMeteo`) reachable with ODBC Driver 17 (or 18)
+- Microsoft SQL Server (`DBMeteo`, and `SQL_APP_DATABASE` on the same instance) reachable with ODBC Driver 17 (or 18)
 - Weaviate running locally (see `docker-compose.yml`) with a populated `ProductProfile` collection
 - API key for the chosen LLM provider (Gemini by default)
 
@@ -40,35 +42,40 @@ docker compose up -d
 
 ## Configuration
 
-- [`config.yaml`](config.yaml) — the `crops:` worklist, concurrency/rate limits, retry policy,
+- [`config.yaml`](config.yaml) — concurrency/rate limits, retry policy, the 09:00 SLA
-  the 09:00 SLA deadline, LLM provider/model, and vocab paths
+  deadline, LLM provider/model, vocab paths, and an optional field allowlist.
+  Advice targets come from the database (see below)
 - [`.env`](.env) — SQL Server, Weaviate, and API keys
 - [`vocab/crops.yaml`](vocab/crops.yaml) / [`vocab/diseases.yaml`](vocab/diseases.yaml) — Italian → English maps
 - [`prompts/`](prompts) — the prompt registry (see below)
 
-### `crops:` worklist
+### SQL databases
 
-```yaml
+The pipeline opens one connection to `SQL_DATABASE`. Field, weather, and disease-model tables (`AI_agrosupport_*`) stay in that database (`DBMeteo`). `advisories`, `products`, `product_uses`, and `label_chunks` are read through the same connection from `SQL_APP_DATABASE` when that names a different database on the same instance. Leave `SQL_APP_DATABASE` unset, or set it equal to `SQL_DATABASE`, when those four tables live in the connected database. `SQL_SERVER` is `host` or `host,port`; the ODBC driver uses a comma before the port.
-crops:
-  - crop: grapevine              # canonical English, matched via vocab/crops.yaml
-    diseases:
-      - model_name: PERONOSPORA  # matched against AI_agrosupport_agro_models.anmod_model
-        disease: downy mildew    # canonical English, used for products/advice/prompts
-```
 
-Every morning, `batch` maps each `crop` to the fields growing it
+### Advice targets
-(`AI_agrosupport_an_colture` → `AI_agrosupport_cmp_layers`), joins to
+
-`AI_agrosupport_agro_models` for each listed `model_name`, and runs one job per
+`batch` does not read a crop or disease list from `config.yaml`. It loads
-matching `(field, disease)` pair. To add a crop or disease: add an entry here, add
+diseases with `pat_mod_AI_model = 1` from `AI_agrosupport_an_pathogen_models`
-the Italian/English mapping to `vocab/crops.yaml` / `vocab/diseases.yaml` if it
+(`pat_mod_desc`), then fields from `AI_agrosupport_agro_models` where
-doesn't already exist, and add a prompt directory (see below) — otherwise the
+`anmod_AI_advice = 1` and `anmod_model` matches one of those names. The field
-pair silently uses the generic fallback prompt.
+id is `anmod_cmplay`. The crop is `cmplay_colture` on `AI_agrosupport_cmp_layers`,
+joined to `AI_agrosupport_an_colture` and mapped through
+[`vocab/crops.yaml`](vocab/crops.yaml) (`Vite` → `grapevine`). The English
+disease name is mapped through [`vocab/diseases.yaml`](vocab/diseases.yaml)
+(`PERONOSPORA` → `downy mildew`). Those two English names select the prompt
+and the product filter. One agro-model row is one job.
+
+To enable another disease, set both flags in SQL. Add a `pat_mod_desc` mapping
+to `vocab/diseases.yaml` if it is not already there (`peronospora` already is),
+and optionally add a prompt directory (see below). An unmapped disease or crop
+name is skipped and logged. A pair without a dedicated prompt uses the generic
+fallback and is logged at startup.
 
 ### `worklist:`, `concurrency:`, `retry:`, `schedule:`
 
 ```yaml
 worklist:
-  require_enabled_model: true   # AI_agrosupport_agro_models.anmod_enabled must be 1
   field_allowlist: []           # restrict to specific field IDs while testing
 
 concurrency:
@@ -90,7 +97,7 @@ schedule:
 Each crop-disease pair needs its own advice prompt (its phytopathological
 model's phase semantics, phenology notes, etc.). Prompts are plain Markdown
 files resolved by `pipeline/prompts.py::PromptRegistry`, with a shared fallback
-so a newly configured pair never fails outright — it just logs a warning and
+so a newly enabled pair never fails outright — it just logs a warning and
 uses a generic prompt until a dedicated one is written.
 
 ```
@@ -102,18 +109,18 @@ prompts/
   advice/<crop>__<disease>/{system.md,user.md} per-pair advice prompts
 ```
 
-The `<crop>__<disease>` slug is built from the same canonical English names used
+The `<crop>__<disease>` slug is built from the canonical English crop and disease
-in `config.yaml`'s `crops:` block (lower-cased, non-alphanumeric runs collapsed
+names (lower-cased, non-alphanumeric runs collapsed to `_`) — e.g. `grapevine` +
-to `_`) — e.g. `grapevine` + `downy mildew` → `grapevine__downy_mildew`. To add a
+`downy mildew` → `grapevine__downy_mildew`. To add a prompt for a new pair, create
-prompt for a new pair, create `prompts/advice/<slug>/system.md` and `user.md`;
+`prompts/advice/<slug>/system.md` and `user.md`; `batch` logs every discovered
-`batch` logs every configured pair still falling back to `_default` at startup.
+pair still falling back to `_default` at startup.
 
 ## Run
 
 ```powershell
 .\.venv\Scripts\Activate.ps1
 
-# Daily entry point: every crop/disease pair in config.yaml, across all matching fields
+# Daily entry point: every field/disease pair enabled for advice in the database
 python -m pipeline batch
 
 # Historical / test dates (useful when the live window has no FASE2/FASE5/INCUBAZPRIMARIA/observation)
@@ -372,7 +379,7 @@ is not worth a custom cost pipeline at this scale.
 
 Run once per `(field, disease)` job by `pipeline/job.py::run_job`:
 
-1. **Worklist** — crop → fields (`AI_agrosupport_an_colture` + `AI_agrosupport_cmp_layers`) → disease model (`AI_agrosupport_agro_models`); organic flag is `cmplay_imp == 3` (`pipeline/worklist.py`, run once per batch, not per job)
+1. **Worklist** — supported diseases (`AI_agrosupport_an_pathogen_models.pat_mod_AI_model = 1`) → advice-enabled field models (`AI_agrosupport_agro_models.anmod_AI_advice = 1`) → crop from `AI_agrosupport_cmp_layers` + `AI_agrosupport_an_colture`; organic flag is `cmplay_imp == 3` (`pipeline/worklist.py`, run once per batch, not per job)
 2. **Disease model** — per day over as_of±5, from the worklist's known `anmod_id`. `disease_forecast` is a list or null. Each `INCUBAZPRIMARIA` index is kept (`INCUBAZPRIMARIA2_80.7`, or `INCUBAZPRIMARIA_40.3` when the description has no trailing index), using that index's highest `model_value`. `FASE5` is appended when present; `FASE2` is appended only when `FASE5` is absent. `predicted_incubations_management` (loaded from the previous `advisories.advice_input_json`, then written back with this run) records each index's `start_date` and `infection_covered`. Today counts as provisional: a start locks only once the index appears on a day before `as_of`. A covered incubation alone is not risk.
 3. **Observation** — `AI_agrosupport_agro_ril_pathogen` (`rilpato_layer`/`rilpato_date`/`rilpato_diffusion`); `observation` is true for `as_of` or any of the past 5 days if a row exists there with `rilpato_diffusion` other than `32` (including `NULL`)
 4. **Weather** — `TDatiMeteo_D` for past days, `TDatiMeteo_D_FRC` for today/future

config.yaml

@@ -29,31 +29,20 @@ llm:
     thinking_level: medium
     temperature: 0.0
 
-# ── Crop-first worklist ────────────────────────────────────────────────────────
+# ── Worklist filters ───────────────────────────────────────────────────────────
-# `python -m pipeline batch` discovers jobs from this list: for every crop, it
+# `python -m pipeline batch` discovers jobs from the database, not from this
-# finds all fields growing it (AI_agrosupport_cmp_layers ⨝ an_colture) that also
+# file. A disease can generate advice when
-# have a matching row in AI_agrosupport_agro_models for each listed disease's
+# AI_agrosupport_an_pathogen_models.pat_mod_AI_model = 1; its name is
-# model_name, and runs the full single-field pipeline for every resulting
+# pat_mod_desc. A field is included when AI_agrosupport_agro_models.anmod_AI_advice
-# (field, disease) pair. `disease` is the canonical English name used for the
+# = 1 and anmod_model matches that name. The cultivated crop comes from
-# product prefilter, the vector search, and the advice/prompt lookup; it must
+# AI_agrosupport_cmp_layers / AI_agrosupport_an_colture and vocab/crops.yaml.
-# exist in vocab/diseases.yaml. `model_name` is matched (case-insensitively)
+# The English disease name comes from vocab/diseases.yaml.
-# against AI_agrosupport_agro_models.anmod_model.
 #
 # Each crop-disease pair needs a matching prompt directory under
-# prompts/advice/<crop>__<disease slug>/ (see prompts/README or the pipeline
+# prompts/advice/<crop>__<disease slug>/ (see the pipeline README for the slug
-# README for the exact slug rules); pairs without one fall back to
+# rules); pairs without one fall back to prompts/advice/_default/ and are
-# prompts/advice/_default/ and are logged at startup.
+# logged at startup.
-crops:
-  - crop: grapevine
-    diseases:
-      - model_name: PERONOSPORA
-        disease: downy mildew
-
-# ── Worklist filters ───────────────────────────────────────────────────────────
 worklist:
-  # AI_agrosupport_agro_models.anmod_enabled is True for every model row in the
-  # current database; keep this on so a disabled model never generates advice.
-  require_enabled_model: true
   # Restrict the batch to specific field IDs while testing; empty = no filter.
   field_allowlist: []
 

pipeline/__main__.py

@@ -7,9 +7,10 @@
     python -m pipeline one --field-id N --disease NAME [--as-of DATE] [--dry-run]
 
 `batch` is the daily entry point: it discovers every (field, disease) job
-implied by config.yaml's `crops:` list and the database, and runs them
+from AI_agrosupport_an_pathogen_models (pat_mod_AI_model = 1) and
-through a bounded thread pool (see pipeline/batch.py). `one` keeps the
+AI_agrosupport_agro_models (anmod_AI_advice = 1), and runs them through a
-original single-field behaviour for ad-hoc debugging.
+bounded thread pool (see pipeline/batch.py). `one` keeps the original
+single-field behaviour for ad-hoc debugging.
 """
 
 from __future__ import annotations
@@ -108,10 +109,11 @@ 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."""
+    """`python -m pipeline batch`: every database-enabled field/disease pair."""
     settings = load_settings()
     setup_tracing(settings)
     crop_vocab = load_crop_vocab(settings.crops_vocab)
+    disease_vocab = load_disease_vocab(settings.diseases_vocab)
 
     logger.info(
         "Starting batch as_of=%s crop_filter=%s disease_filter=%s field_filter=%s "
@@ -128,6 +130,7 @@ def _run_batch(args: argparse.Namespace) -> int:
     return run_batch(
         settings,
         crop_vocab,
+        disease_vocab,
         args.as_of,
         crop=args.crop,
         disease=args.disease,
@@ -145,7 +148,7 @@ def main(argv: list[str] | None = None) -> int:
 
     batch_parser = subparsers.add_parser(
         "batch",
-        help="Run every configured crop/disease pair across all matching fields.",
+        help="Run every field/disease pair enabled for advice in the database.",
     )
     batch_parser.add_argument(
         "--as-of", type=_parse_as_of, default=None,

pipeline/batch.py

@@ -17,7 +17,7 @@ from datetime import date, datetime
 from datetime import time as dtime
 
 from pipeline.config import Settings
-from pipeline.db import fetch_all
+from pipeline.db import fetch_all, qualify_app_sql
 from pipeline.job import JobResult, run_job
 from pipeline.output import write_job_output, write_run_report
 from pipeline.prompts import PromptRegistry
@@ -27,7 +27,7 @@ from pipeline.worklist import Job, build_jobs
 
 logger = logging.getLogger(__name__)
 
-_EXISTING_ADVICE_SQL = "SELECT field_id, disease FROM advisories WHERE issuance_date = ?"
+_EXISTING_ADVICE_SQL = "SELECT field_id, disease FROM {advisories} WHERE issuance_date = ?"
 
 
 def _parse_deadline(deadline_str: str) -> datetime:
@@ -69,13 +69,18 @@ def filter_jobs(
 def _existing_advice_keys(resources: Resources, as_of: date) -> set[tuple[int, str]]:
     """One bulk SELECT instead of one per job, for --skip-existing-advice."""
     conn = resources.sql_connection()
-    rows = fetch_all(conn, _EXISTING_ADVICE_SQL, (as_of,))
+    rows = fetch_all(
+        conn,
+        qualify_app_sql(resources.settings.sql, _EXISTING_ADVICE_SQL),
+        (as_of,),
+    )
     return {(int(row.field_id), str(row.disease)) for row in rows}
 
 
 def run_batch(
     settings: Settings,
     crop_vocab: Vocabulary,
+    disease_vocab: Vocabulary,
     as_of: date,
     *,
     crop: str | None = None,
@@ -88,8 +93,21 @@ def run_batch(
 ) -> int:
     """Run every discovered job and write the run report. Returns the process exit code."""
     prompts = PromptRegistry(settings.prompts_dir)
-    configured_pairs = [(cp.crop, dp.disease) for cp in settings.crops for dp in cp.diseases]
+    resources = Resources(settings, prompts)
-    fallbacks = prompts.validate(configured_pairs)
+    try:
+        jobs = build_jobs(
+            resources.sql_connection(),
+            settings.worklist,
+            crop_vocab,
+            disease_vocab,
+        )
+        jobs = filter_jobs(jobs, crop=crop, disease=disease, field_ids=field_ids)
+        logger.info("Worklist discovered %d job(s) for as_of=%s", len(jobs), as_of.isoformat())
+
+        # Unique pairs only: one warning per crop/disease, not one per field.
+        # An empty job list validates nothing.
+        prompt_pairs = list(dict.fromkeys((job.crop_english, job.disease_english) for job in jobs))
+        fallbacks = prompts.validate(prompt_pairs)
         if fallbacks:
             logger.warning(
                 "%d crop/disease pair(s) have no dedicated advice prompt and will fall back "
@@ -98,16 +116,10 @@ def run_batch(
                 fallbacks,
             )
 
-    resources = Resources(settings, prompts)
-    try:
-        jobs = build_jobs(resources.sql_connection(), settings.crops, settings.worklist, crop_vocab)
-        jobs = filter_jobs(jobs, crop=crop, disease=disease, field_ids=field_ids)
-        logger.info("Worklist discovered %d job(s) for as_of=%s", len(jobs), as_of.isoformat())
-
         if not jobs:
             logger.warning(
-                "No jobs to run for as_of=%s; nothing matched the configured "
+                "No jobs to run for as_of=%s; nothing matched pat_mod_AI_model = 1 "
-                "crops/diseases (or the CLI filters).",
+                "and anmod_AI_advice = 1 (or the CLI filters).",
                 as_of.isoformat(),
             )
             write_run_report(settings.output_dir, as_of, [])

pipeline/config.py

@@ -28,6 +28,7 @@ class SqlSettings:
     driver: str
     server: str
     database: str
+    app_database: str
     username: str
     password: str
 
@@ -39,25 +40,8 @@ class WeaviateSettings:
     grpc_port: int
 
 
-@dataclass(frozen=True)
-class DiseasePlan:
-    """One crop-disease pair to run the batch for."""
-
-    model_name: str  # matched against AI_agrosupport_agro_models.anmod_model
-    disease: str  # canonical English disease name (vocab/diseases.yaml)
-
-
-@dataclass(frozen=True)
-class CropPlan:
-    """A crop and the diseases to check for every field growing it."""
-
-    crop: str  # canonical English crop name (vocab/crops.yaml)
-    diseases: tuple[DiseasePlan, ...]
-
-
 @dataclass(frozen=True)
 class WorklistSettings:
-    require_enabled_model: bool
     field_allowlist: tuple[int, ...]
 
 
@@ -99,7 +83,6 @@ class Settings:
     root: Path
     field_id: int | None
     disease_name: str | None
-    crops: tuple[CropPlan, ...]
     worklist: WorklistSettings
     concurrency: ConcurrencySettings
     retry: RetrySettings
@@ -142,24 +125,8 @@ def _parse_llm_settings(raw: dict, defaults: dict) -> LlmSettings:
     )
 
 
-def _parse_crops(raw: list[dict] | None) -> tuple[CropPlan, ...]:
-    plans: list[CropPlan] = []
-    for entry in raw or []:
-        crop = str(entry["crop"]).strip()
-        diseases = tuple(
-            DiseasePlan(
-                model_name=str(d["model_name"]).strip(),
-                disease=str(d["disease"]).strip(),
-            )
-            for d in entry.get("diseases", [])
-        )
-        plans.append(CropPlan(crop=crop, diseases=diseases))
-    return tuple(plans)
-
-
 def _parse_worklist(raw: dict) -> WorklistSettings:
     return WorklistSettings(
-        require_enabled_model=bool(raw.get("require_enabled_model", True)),
         field_allowlist=tuple(int(v) for v in raw.get("field_allowlist", []) or []),
     )
 
@@ -234,8 +201,8 @@ def load_settings(
     llm_raw = raw.get("llm", {})
     vocab = raw.get("vocab", {})
 
-    # field_id / disease_name only matter for `python -m pipeline one`; the
+    # field_id / disease_name only matter for `python -m pipeline one`. The
-    # batch worklist is driven entirely by the crops: block below.
+    # batch worklist is discovered from the pathogen and agro-model tables.
     field_id = field_id_override if field_id_override is not None else raw.get("field_id")
     field_id = int(field_id) if field_id is not None else None
     disease_name = disease_name_override or raw.get("disease_name")
@@ -264,7 +231,6 @@ def load_settings(
         root=ROOT,
         field_id=field_id,
         disease_name=disease_name,
-        crops=_parse_crops(raw.get("crops")),
         worklist=_parse_worklist(raw.get("worklist", {})),
         concurrency=_parse_concurrency(raw.get("concurrency", {})),
         retry=_parse_retry(raw.get("retry", {})),
@@ -282,6 +248,8 @@ def load_settings(
             driver=os.environ.get("SQL_DRIVER", "ODBC Driver 17 for SQL Server"),
             server=os.environ["SQL_SERVER"],
             database=os.environ["SQL_DATABASE"],
+            app_database=os.environ.get("SQL_APP_DATABASE", "").strip()
+            or os.environ["SQL_DATABASE"],
             username=os.environ["SQL_USERNAME"],
             password=os.environ["SQL_PASSWORD"],
         ),

pipeline/db.py

@@ -10,10 +10,64 @@ import pyodbc
 from pipeline.config import SqlSettings
 
 
+# Tables that live in SQL_APP_DATABASE. Every other name stays in SQL_DATABASE.
+_APP_TABLES = ("advisories", "products", "product_uses", "label_chunks")
+
+
+def _bracket(identifier: str) -> str:
+    """Quote a SQL Server identifier. A closing bracket inside the name is doubled."""
+    return "[" + identifier.replace("]", "]]") + "]"
+
+
+def _odbc_server(server: str) -> str:
+    """
+    ODBC Driver expects host,port. Accept host:port and leave named instances alone.
+
+    `192.168.220.50:1433` becomes `192.168.220.50,1433`. `localhost\\SQLEXPRESS`
+    has no colon, so it is returned unchanged.
+    """
+    host, colon, port = server.rpartition(":")
+    if colon and port.isdigit() and ":" not in host:
+        return f"{host},{port}"
+    return server
+
+
+def app_table(settings: SqlSettings, table: str) -> str:
+    """
+    Name of an application table on the open connection.
+
+    Bare when `app_database` is the connected database, so a single-database
+    setup keeps unqualified SQL. Otherwise a three-part name,
+    `[Agrosupport].dbo.[advisories]`, so one DBMeteo connection can read the
+    other database on the same instance.
+    """
+    if table not in _APP_TABLES:
+        allowed = ", ".join(_APP_TABLES)
+        raise ValueError(f"Unknown application table {table!r}. Expected one of {allowed}.")
+    if (
+        not settings.app_database
+        or settings.app_database.casefold() == settings.database.casefold()
+    ):
+        return table
+    return f"{_bracket(settings.app_database)}.dbo.{_bracket(table)}"
+
+
+def qualify_app_sql(settings: SqlSettings, sql: str) -> str:
+    """
+    Replace `{advisories}`, `{products}`, `{product_uses}`, and `{label_chunks}`.
+
+    Call this before `.format(placeholders=...)`. Only those four tokens are
+    substituted, so a `{placeholders}` brace list is left for the later format.
+    """
+    for table in _APP_TABLES:
+        sql = sql.replace("{" + table + "}", app_table(settings, table))
+    return sql
+
+
 def _connection_string(settings: SqlSettings) -> str:
     return (
         f"DRIVER={{{settings.driver}}};"
-        f"SERVER={settings.server};"
+        f"SERVER={_odbc_server(settings.server)};"
         f"DATABASE={settings.database};"
         f"UID={settings.username};"
         f"PWD={settings.password};"

pipeline/job.py

@@ -176,7 +176,7 @@ def _run_job_inner(resources: Resources, job: Job, as_of: date, dry_run: bool) -
             raise RuntimeError("load_product_index() must run before run_job().")
         previous_register = parse_register(
             load_previous_incubation_register(
-                conn, job.field_id, disease.disease_english, as_of
+                conn, job.field_id, disease.disease_english, as_of, settings.sql
             )
         )
         interval_open = last_treatment_interval_open(
@@ -186,6 +186,7 @@ def _run_job_inner(resources: Resources, job: Job, as_of: date, dry_run: bool) -
             disease_english=disease.disease_english,
             latest_spray=treatment_history.latest(),
             as_of=as_of,
+            sql=settings.sql,
         )
 
         def _covers(start: date):
@@ -196,6 +197,7 @@ def _run_job_inner(resources: Resources, job: Job, as_of: date, dry_run: bool) -
                 disease_english=disease.disease_english,
                 start_date=start,
                 treatment_history=treatment_history,
+                sql=settings.sql,
             )
 
         incubation_register = reconcile_register(
@@ -343,7 +345,7 @@ def _run_job_inner(resources: Resources, job: Job, as_of: date, dry_run: bool) -
     builder = resources.product_json_builder(job.crop_english, disease.disease_english)
     with resources.sql_sem:
         last_summary, last_treatments, last_date = load_last_advice(
-            conn, job.field_id, disease.disease_english, as_of
+            conn, job.field_id, disease.disease_english, as_of, settings.sql
         )
         enrich_advice_json(
             json_for_advice_generation,
@@ -379,6 +381,7 @@ def _run_job_inner(resources: Resources, job: Job, as_of: date, dry_run: bool) -
                 disease_english=disease.disease_english,
                 advice_input_json=advice_input_json,
                 result=result,
+                sql=settings.sql,
             )
 
     call_with_sql_retry(settings.retry, _insert)

pipeline/resources.py

@@ -134,7 +134,7 @@ class Resources:
 
     def load_product_index(self) -> ProductIndex:
         """Load the whole `products` table once; call before starting the pool."""
-        self.product_index = ProductIndex.load(self.sql_connection())
+        self.product_index = ProductIndex.load(self.sql_connection(), self.settings.sql)
         return self.product_index
 
     def product_json_builder(self, crop_english: str, disease_english: str) -> ProductJsonBuilder:
@@ -148,7 +148,12 @@ class Resources:
         with self._builder_lock:
             builder = self._builder_cache.get(key)
             if builder is None:
-                builder = ProductJsonBuilder(self.product_index, crop_english, disease_english)
+                builder = ProductJsonBuilder(
+                    self.product_index,
+                    crop_english,
+                    disease_english,
+                    self.settings.sql,
+                )
                 self._builder_cache[key] = builder
             return builder
 

pipeline/stages/advice.py

@@ -11,8 +11,8 @@ from typing import Any, Callable
 
 import pyodbc
 
-from pipeline.config import Settings
+from pipeline.config import Settings, SqlSettings
-from pipeline.db import execute, transaction
+from pipeline.db import execute, qualify_app_sql, transaction
 from pipeline.errors import PipelineError
 from pipeline.stages.advice_guardrails import AdviceRejected, validate_advice
 from pipeline.stages.advice_summary import sanitize_advice_summary
@@ -26,14 +26,14 @@ _RECOMMENDED_PRODUCT_NAMES_MAX_CHARS = 200
 _RECOMMENDED_DOSES_MAX_CHARS = 100
 
 _DELETE_ADVICE_SQL = """
-DELETE FROM advisories
+DELETE FROM {advisories}
 WHERE issuance_date = ?
   AND field_id = ?
   AND disease = ?
 """
 
 _INSERT_ADVICE_SQL = """
-INSERT INTO advisories (
+INSERT INTO {advisories} (
     issuance_date,
     field_id,
     disease,
@@ -176,6 +176,7 @@ def insert_advice(
     disease_english: str,
     advice_input_json: dict[str, Any],
     result: AdviceResult,
+    sql: SqlSettings,
 ) -> None:
     """
     Store the advisory, replacing any earlier row for the same day/field/disease.
@@ -202,10 +203,14 @@ def insert_advice(
     sent = json.dumps(advice_input_json, ensure_ascii=False, default=_json_default)
 
     with transaction(conn):
-        execute(conn, _DELETE_ADVICE_SQL, (issuance_date, field_id, disease_english))
         execute(
             conn,
-            _INSERT_ADVICE_SQL,
+            qualify_app_sql(sql, _DELETE_ADVICE_SQL),
+            (issuance_date, field_id, disease_english),
+        )
+        execute(
+            conn,
+            qualify_app_sql(sql, _INSERT_ADVICE_SQL),
             (
                 issuance_date,
                 field_id,

pipeline/stages/advice_context.py

@@ -10,7 +10,8 @@ from typing import Any, Iterable
 
 import pyodbc
 
-from pipeline.db import fetch_one
+from pipeline.config import SqlSettings
+from pipeline.db import fetch_one, qualify_app_sql
 from pipeline.jsonutils import parse_json_list
 from pipeline.stages.product_json import ProductJsonBuilder, normalize_name
 from pipeline.stages.treatments import TreatmentHistory
@@ -26,7 +27,7 @@ SELECT TOP 1
     issuance_date,
     advice_summary,
     recommended_product_names
-FROM advisories
+FROM {advisories}
 WHERE field_id = ?
   AND disease = ?
   AND issuance_date < ?
@@ -46,7 +47,7 @@ def _row_date(value: Any) -> date | None:
 
 _LAST_SENT_SQL = """
 SELECT TOP 1 advice_input_json
-FROM advisories
+FROM {advisories}
 WHERE field_id = ?
   AND disease = ?
   AND issuance_date < ?
@@ -59,9 +60,14 @@ def load_previous_incubation_register(
     field_id: int,
     disease_english: str,
     as_of: date,
+    sql: SqlSettings,
 ) -> dict[str, Any]:
     """Return predicted_incubations_management from the latest advice before as_of."""
-    row = fetch_one(conn, _LAST_SENT_SQL, (field_id, disease_english, as_of))
+    row = fetch_one(
+        conn,
+        qualify_app_sql(sql, _LAST_SENT_SQL),
+        (field_id, disease_english, as_of),
+    )
     if row is None or not row.advice_input_json:
         return {}
     try:
@@ -86,9 +92,14 @@ def load_last_advice(
     field_id: int,
     disease_english: str,
     as_of: date,
+    sql: SqlSettings,
 ) -> tuple[str, list[str], date | None]:
     """Return (advice_summary, treatment names, issuance date) of the most recent advice before as_of."""
-    row = fetch_one(conn, _LAST_ADVICE_SQL, (field_id, disease_english, as_of))
+    row = fetch_one(
+        conn,
+        qualify_app_sql(sql, _LAST_ADVICE_SQL),
+        (field_id, disease_english, as_of),
+    )
     if row is None:
         logger.info(
             "No previous advice for field %s / %s before %s.",

pipeline/stages/disease.py

@@ -158,9 +158,8 @@ def find_anmod_id(conn: pyodbc.Connection, field_id: int, disease_name: str) ->
     Look up the anmod_id for a field's disease model by name.
 
     Used by the single-field `one` CLI path, where `anmod_id` is not known
-    ahead of time. The batch worklist resolves it directly with one join per
+    ahead of time. The batch worklist resolves it in one database join
-    crop-disease pair instead (see `pipeline.worklist.build_jobs`), so it
+    (see `pipeline.worklist.build_jobs`), so it never calls this per field.
-    never calls this per field.
     """
     models = fetch_all(conn, _MODEL_SQL, (field_id,))
     target = disease_name.strip().upper()

pipeline/stages/incubations.py

@@ -9,7 +9,8 @@ from typing import Any, Callable
 
 import pyodbc
 
-from pipeline.db import fetch_all
+from pipeline.config import SqlSettings
+from pipeline.db import fetch_all, qualify_app_sql
 from pipeline.stages.disease import incub_register_key
 from pipeline.stages.products import ProductIndex
 from pipeline.stages.treatments import TreatmentHistory
@@ -19,7 +20,7 @@ logger = logging.getLogger(__name__)
 
 _USES_INTERVAL_SQL = """
 SELECT product_id, treatment_interval_min_days
-FROM product_uses
+FROM {product_uses}
 WHERE crop = ?
   AND disease = ?
   AND product_id IN ({placeholders})
@@ -225,6 +226,7 @@ def incubation_start_covered(
     disease_english: str,
     start_date: date,
     treatment_history: TreatmentHistory,
+    sql: SqlSettings,
 ) -> CoveringTreatment | None:
     """
     Latest spray strictly before start_date whose minimum interval still reaches it.
@@ -266,7 +268,7 @@ def incubation_start_covered(
     placeholders = ",".join("?" * len(all_ids))
     use_rows = fetch_all(
         conn,
-        _USES_INTERVAL_SQL.format(placeholders=placeholders),
+        qualify_app_sql(sql, _USES_INTERVAL_SQL).format(placeholders=placeholders),
         (crop_english, disease_english, *all_ids),
     )
     min_days_by_id: dict[str, int] = {}

pipeline/stages/product_json.py

@@ -9,7 +9,8 @@ from typing import Any
 
 import pyodbc
 
-from pipeline.db import fetch_all
+from pipeline.config import SqlSettings
+from pipeline.db import fetch_all, qualify_app_sql
 from pipeline.jsonutils import normalize_name, parse_json_list
 from pipeline.stages.products import ProductIndex
 
@@ -30,7 +31,7 @@ SELECT
     pre_harvest_interval_days,
     growth_stage_start,
     growth_stage_end
-FROM product_uses
+FROM {product_uses}
 WHERE product_id = ?
   AND crop = ?
   AND disease = ?
@@ -43,7 +44,7 @@ SELECT
     chunk_text,
     target_crops,
     target_diseases
-FROM label_chunks
+FROM {label_chunks}
 WHERE product_id = ?
 ORDER BY label_chunk_id ASC
 """
@@ -91,10 +92,12 @@ class ProductJsonBuilder:
         product_index: ProductIndex,
         crop_english: str,
         disease_english: str,
+        sql: SqlSettings,
     ) -> None:
         self._products = product_index
         self._crop = crop_english
         self._disease = disease_english
+        self._sql = sql
         self._cache: dict[str, dict[str, Any] | str] = {}
         self._cache_lock = threading.Lock()
 
@@ -138,14 +141,18 @@ class ProductJsonBuilder:
         }
 
     def _load_uses(self, conn: pyodbc.Connection, product_id: str) -> list[dict[str, Any]]:
-        rows = fetch_all(conn, _USES_SQL, (product_id, self._crop, self._disease))
+        rows = fetch_all(
+            conn,
+            qualify_app_sql(self._sql, _USES_SQL),
+            (product_id, self._crop, self._disease),
+        )
         return [
             {key: _scalar(getattr(row, key)) for key in _USE_KEYS}
             for row in rows
         ]
 
     def _load_chunks(self, conn: pyodbc.Connection, product_id: str) -> list[dict[str, Any]]:
-        rows = fetch_all(conn, _CHUNKS_SQL, (product_id,))
+        rows = fetch_all(conn, qualify_app_sql(self._sql, _CHUNKS_SQL), (product_id,))
         chunks: list[dict[str, Any]] = []
         for row in rows:
             if self._crop not in parse_json_list(row.target_crops):

pipeline/stages/products.py

@@ -9,7 +9,8 @@ from typing import Any
 
 import pyodbc
 
-from pipeline.db import fetch_all
+from pipeline.config import SqlSettings
+from pipeline.db import fetch_all, qualify_app_sql
 from pipeline.jsonutils import normalize_name, parse_json_list
 from pipeline.stages.treatments import LAST_TREATMENT_MAX_AGE_DAYS
 
@@ -63,12 +64,12 @@ SELECT
     target_diseases,
     active_ingredients,
     frac_codes
-FROM products
+FROM {products}
 """
 
 _USES_INTERVAL_SQL = """
 SELECT product_id, treatment_interval_min_days
-FROM product_uses
+FROM {product_uses}
 WHERE crop = ?
   AND disease = ?
   AND product_id IN ({placeholders})
@@ -175,8 +176,8 @@ class ProductIndex:
         )
 
     @classmethod
-    def load(cls, conn: pyodbc.Connection) -> "ProductIndex":
+    def load(cls, conn: pyodbc.Connection, sql: SqlSettings) -> "ProductIndex":
-        product_rows = fetch_all(conn, _PRODUCTS_SQL)
+        product_rows = fetch_all(conn, qualify_app_sql(sql, _PRODUCTS_SQL))
         catalog_rows = fetch_all(conn, _CATALOG_STATE_SQL)
         permitted_by_registration = _permitted_by_registration(catalog_rows)
 
@@ -235,6 +236,7 @@ def last_treatment_interval_open(
     disease_english: str,
     latest_spray: tuple[date, list[str]] | None,
     as_of: date,
+    sql: SqlSettings,
 ) -> bool:
     """
     True if the latest spray is still inside its treatment interval.
@@ -276,7 +278,7 @@ def last_treatment_interval_open(
     placeholders = ",".join("?" * len(product_ids))
     use_rows = fetch_all(
         conn,
-        _USES_INTERVAL_SQL.format(placeholders=placeholders),
+        qualify_app_sql(sql, _USES_INTERVAL_SQL).format(placeholders=placeholders),
         (crop_english, disease_english, *product_ids),
     )
 

pipeline/worklist.py

@@ -1,20 +1,23 @@
-"""Crop-first job discovery: find every (field, disease) pair to run today.
+"""Database-driven job discovery: find every (field, disease) pair to run today.
 
-Unlike the original single-field pipeline, the batch does not start from a
+The batch does not start from a field ID or from a crop list in config.yaml.
-field ID. It starts from `config.yaml`'s `crops:` list, resolves which fields
+It reads the diseases that can generate advice
-grow each crop, and for each of that crop's configured diseases finds the
+(`AI_agrosupport_an_pathogen_models.pat_mod_AI_model = 1`, named by
-fields that also have a matching disease-forecast model configured
+`pat_mod_desc`) and the fields that have advice enabled for one of those
-(`AI_agrosupport_agro_models`). Each match becomes one `Job`.
+diseases (`AI_agrosupport_agro_models.anmod_AI_advice = 1`). The cultivated
+crop comes from the field layer. Each matching agro-model row becomes one
+`Job`.
 """
 
 from __future__ import annotations
 
 import logging
+from collections import Counter
 from dataclasses import dataclass
 
 import pyodbc
 
-from pipeline.config import CropPlan, WorklistSettings
+from pipeline.config import WorklistSettings
 from pipeline.db import fetch_all
 from pipeline.errors import PipelineError
 from pipeline.vocab import Vocabulary
@@ -36,128 +39,104 @@ class Job:
     anmod_id: int
 
 
-_COLTURE_SQL = "SELECT colture_id, colture_name FROM AI_agrosupport_an_colture"
+_JOBS_SQL = """
-
+SELECT DISTINCT
-_JOBS_SQL_TEMPLATE = """
-SELECT
     l.cmplay_id,
     l.cmplay_colture,
     l.cmplay_imp,
     l.cmplay_station,
+    c.colture_name,
     m.anmod_id,
-    m.anmod_enabled
+    m.anmod_model
-FROM AI_agrosupport_cmp_layers l
+FROM AI_agrosupport_agro_models m
-JOIN AI_agrosupport_agro_models m ON m.anmod_cmplay = l.cmplay_id
+INNER JOIN (
-WHERE l.cmplay_colture IN ({placeholders})
+    SELECT DISTINCT LTRIM(RTRIM(pat_mod_desc)) AS model_name
-  AND UPPER(LTRIM(RTRIM(m.anmod_model))) = ?
+    FROM AI_agrosupport_an_pathogen_models
+    WHERE pat_mod_AI_model = 1
+      AND pat_mod_desc IS NOT NULL
+      AND LTRIM(RTRIM(pat_mod_desc)) <> ''
+) p
+    ON UPPER(LTRIM(RTRIM(m.anmod_model))) = UPPER(p.model_name)
+INNER JOIN AI_agrosupport_cmp_layers l
+    ON l.cmplay_id = m.anmod_cmplay
+LEFT JOIN AI_agrosupport_an_colture c
+    ON c.colture_id = l.cmplay_colture
+WHERE m.anmod_AI_advice = 1
 """
 
 
-def _load_crop_to_colture_ids(
-    conn: pyodbc.Connection, crop_vocab: Vocabulary
-) -> dict[str, list[int]]:
-    """
-    Map every canonical English crop name to the colture_ids that resolve to it.
-
-    `Vocabulary.to_english` already strips whitespace and lower-cases before
-    matching, so `AI_agrosupport_an_colture` rows like `"Vite "` (trailing
-    space) resolve correctly without extra cleanup here. Colture names with no
-    vocabulary entry are logged once and skipped, so a single unmapped crop
-    cannot break discovery for every other crop.
-    """
-    mapping: dict[str, list[int]] = {}
-    unmapped: list[str] = []
-    for row in fetch_all(conn, _COLTURE_SQL):
-        italian = (row.colture_name or "").strip()
-        if not italian:
-            continue
-        try:
-            english = crop_vocab.to_english(italian, kind="crop")
-        except PipelineError:
-            unmapped.append(italian)
-            continue
-        mapping.setdefault(english, []).append(int(row.colture_id))
-
-    if unmapped:
-        logger.warning(
-            "%d colture name(s) have no crop vocabulary mapping and will be "
-            "ignored by the worklist: %s",
-            len(unmapped),
-            sorted(set(unmapped)),
-        )
-    return mapping
-
-
-def _colture_names_by_id(conn: pyodbc.Connection) -> dict[int, str]:
-    return {
-        int(row.colture_id): (row.colture_name or "").strip()
-        for row in fetch_all(conn, _COLTURE_SQL)
-    }
-
-
 def build_jobs(
     conn: pyodbc.Connection,
-    crops: tuple[CropPlan, ...],
     worklist: WorklistSettings,
     crop_vocab: Vocabulary,
+    disease_vocab: Vocabulary,
 ) -> list[Job]:
-    """Discover every (field, disease) job implied by `crops` and `worklist`."""
+    """Discover every (field, disease) job enabled in the database."""
-    crop_to_colture = _load_crop_to_colture_ids(conn, crop_vocab)
-    colture_names = _colture_names_by_id(conn)
     allowlist = set(worklist.field_allowlist)
+    rows = fetch_all(conn, _JOBS_SQL)
 
     jobs: list[Job] = []
-    for crop_plan in crops:
+    unmapped_diseases: list[str] = []
-        colture_ids = crop_to_colture.get(crop_plan.crop, [])
+    unmapped_colture: list[str] = []
-        if not colture_ids:
-            logger.warning(
-                "No field layer maps to crop '%s' (check vocab/crops.yaml against "
-                "AI_agrosupport_an_colture); skipping its %d configured disease(s).",
-                crop_plan.crop,
-                len(crop_plan.diseases),
-            )
-            continue
 
-        placeholders = ",".join("?" for _ in colture_ids)
-        sql = _JOBS_SQL_TEMPLATE.format(placeholders=placeholders)
-
-        for disease_plan in crop_plan.diseases:
-            params = (*colture_ids, disease_plan.model_name.strip().upper())
-            rows = fetch_all(conn, sql, params)
-
-            kept = 0
     for row in rows:
         if allowlist and int(row.cmplay_id) not in allowlist:
             continue
-                if worklist.require_enabled_model and not bool(row.anmod_enabled):
+
+        model_name = (row.anmod_model or "").strip()
+        try:
+            disease_english = disease_vocab.to_english(model_name, kind="disease")
+        except PipelineError:
+            unmapped_diseases.append(model_name or "(blank)")
+            continue
+
+        colture_name = (row.colture_name or "").strip()
+        try:
+            crop_english = crop_vocab.to_english(colture_name, kind="crop")
+        except PipelineError:
+            unmapped_colture.append(colture_name or "(blank)")
             continue
 
         jobs.append(
             Job(
                 field_id=int(row.cmplay_id),
-                        crop_english=crop_plan.crop,
+                crop_english=crop_english,
-                        crop_italian=colture_names.get(int(row.cmplay_colture), ""),
+                crop_italian=colture_name,
-                        model_name=disease_plan.model_name,
+                model_name=model_name,
-                        disease_english=disease_plan.disease,
+                disease_english=disease_english,
                 organic=row.cmplay_imp == 3,
                 station=(
-                            int(row.cmplay_station)
+                    int(row.cmplay_station) if row.cmplay_station is not None else None
-                            if row.cmplay_station is not None
-                            else None
                 ),
                 anmod_id=int(row.anmod_id),
             )
         )
-                kept += 1
 
+    jobs.sort(key=lambda job: (job.field_id, job.model_name, job.anmod_id))
+
+    if unmapped_diseases:
+        logger.warning(
+            "%d disease model name(s) have no disease vocabulary mapping and will be "
+            "ignored by the worklist: %s",
+            len(set(unmapped_diseases)),
+            sorted(set(unmapped_diseases)),
+        )
+    if unmapped_colture:
+        logger.warning(
+            "%d colture name(s) have no crop vocabulary mapping and will be "
+            "ignored by the worklist: %s",
+            len(set(unmapped_colture)),
+            sorted(set(unmapped_colture)),
+        )
+
+    counts = Counter((job.crop_english, job.disease_english, job.model_name) for job in jobs)
+    for crop, disease, model_name in sorted(counts):
         logger.info(
-                "Worklist: crop=%s disease=%s (model=%s) -> %d field(s) "
+            "Worklist: crop=%s disease=%s (model=%s) -> %d job(s)",
-                "(%d row(s) matched before allowlist/enabled filters)",
+            crop,
-                crop_plan.crop,
+            disease,
-                disease_plan.disease,
+            model_name,
-                disease_plan.model_name,
+            counts[(crop, disease, model_name)],
-                kept,
-                len(rows),
         )
 
     return jobs

tests/test_db.py

@@ -0,0 +1,235 @@
+"""Unit tests for the application-database qualifier and ODBC server form."""
+
+from __future__ import annotations
+
+import ast
+import unittest
+from datetime import date, timedelta
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+from pipeline.config import SqlSettings
+from pipeline.db import _connection_string, app_table, qualify_app_sql
+from pipeline.stages.advice import AdviceResult, insert_advice
+from pipeline.stages.advice_context import load_last_advice, load_previous_incubation_register
+from pipeline.stages.incubations import incubation_start_covered
+from pipeline.stages.product_json import ProductJsonBuilder
+from pipeline.stages.products import ProductIndex, ProductRow, last_treatment_interval_open
+from pipeline.stages.treatments import TreatmentHistory
+
+_ROOT = Path(__file__).resolve().parent.parent
+
+
+def _sql(**overrides: str) -> SqlSettings:
+    values = {
+        "driver": "ODBC Driver 17 for SQL Server",
+        "server": "localhost",
+        "database": "DBMeteo",
+        "app_database": "DBMeteo",
+        "username": "user",
+        "password": "secret",
+    }
+    values.update(overrides)
+    return SqlSettings(**values)
+
+
+class AppTableTests(unittest.TestCase):
+    def test_same_database_stays_bare(self) -> None:
+        self.assertEqual(app_table(_sql(), "advisories"), "advisories")
+        self.assertEqual(app_table(_sql(app_database="dbmeteo"), "products"), "products")
+
+    def test_different_database_is_three_part(self) -> None:
+        settings = _sql(app_database="Agrosupport")
+        self.assertEqual(app_table(settings, "advisories"), "[Agrosupport].dbo.[advisories]")
+
+    def test_qualify_leaves_placeholders(self) -> None:
+        sql = "SELECT product_id FROM {product_uses} WHERE product_id IN ({placeholders})"
+        qualified = qualify_app_sql(_sql(app_database="Agrosupport"), sql)
+        self.assertEqual(
+            qualified,
+            "SELECT product_id FROM [Agrosupport].dbo.[product_uses] "
+            "WHERE product_id IN ({placeholders})",
+        )
+
+    def test_unknown_table_rejected(self) -> None:
+        with self.assertRaises(ValueError):
+            app_table(_sql(), "AI_agrosupport_cmp_layers")
+
+
+class ConnectionStringTests(unittest.TestCase):
+    def test_colon_port_becomes_comma(self) -> None:
+        text = _connection_string(_sql(server="192.168.220.50:1433"))
+        self.assertIn("SERVER=192.168.220.50,1433;", text)
+
+    def test_named_instance_unchanged(self) -> None:
+        text = _connection_string(_sql(server=r"localhost\SQLEXPRESS"))
+        self.assertIn(r"SERVER=localhost\SQLEXPRESS;", text)
+
+    def test_comma_port_unchanged(self) -> None:
+        text = _connection_string(_sql(server="192.168.220.50,1433"))
+        self.assertIn("SERVER=192.168.220.50,1433;", text)
+
+
+def _product() -> ProductRow:
+    return ProductRow(
+        product_id="p1",
+        product_name="LIETO",
+        registration_number=None,
+        is_organic_certified=False,
+        systemicity="contact",
+        has_preventive_action=True,
+        has_curative_action=False,
+        has_eradicant_action=False,
+        target_crops=["grapevine"],
+        target_diseases=["downy mildew"],
+        active_ingredients=[],
+        frac_codes=[],
+    )
+
+
+class QualifiedSqlTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.split = _sql(app_database="Agrosupport")
+
+    def test_product_index_qualifies_products_and_leaves_the_catalog(self) -> None:
+        with patch("pipeline.stages.products.fetch_all", side_effect=[[], []]) as fetch:
+            ProductIndex.load(MagicMock(), self.split)
+        products_sql, catalog_sql = (call.args[1] for call in fetch.call_args_list)
+        self.assertIn("[Agrosupport].dbo.[products]", products_sql)
+        self.assertIn("AI_agrosupport_products_catalog", catalog_sql)
+        self.assertNotIn("Agrosupport", catalog_sql)
+
+    def test_treatment_interval_qualifies_product_uses(self) -> None:
+        with patch("pipeline.stages.products.fetch_all", return_value=[]) as fetch:
+            last_treatment_interval_open(
+                MagicMock(),
+                ProductIndex([_product()]),
+                crop_english="grapevine",
+                disease_english="downy mildew",
+                latest_spray=(date(2026, 4, 6), ["LIETO"]),
+                as_of=date(2026, 4, 16),
+                sql=self.split,
+            )
+        self.assertIn("[Agrosupport].dbo.[product_uses]", fetch.call_args.args[1])
+
+    def test_incubation_coverage_qualifies_product_uses(self) -> None:
+        as_of = date(2026, 4, 16)
+        start = as_of - timedelta(days=2)
+        with patch("pipeline.stages.incubations.fetch_all", return_value=[]) as fetch:
+            incubation_start_covered(
+                MagicMock(),
+                ProductIndex([_product()]),
+                crop_english="grapevine",
+                disease_english="downy mildew",
+                start_date=start,
+                treatment_history=TreatmentHistory(as_of=as_of, by_day={start - timedelta(days=4): ["LIETO"]}),
+                sql=self.split,
+            )
+        self.assertIn("[Agrosupport].dbo.[product_uses]", fetch.call_args.args[1])
+
+    def test_product_json_qualifies_uses_and_chunks(self) -> None:
+        builder = ProductJsonBuilder(ProductIndex([_product()]), "grapevine", "downy mildew", self.split)
+        with patch("pipeline.stages.product_json.fetch_all", side_effect=[[], []]) as fetch:
+            builder.build(MagicMock(), "LIETO")
+        uses_sql, chunks_sql = (call.args[1] for call in fetch.call_args_list)
+        self.assertIn("[Agrosupport].dbo.[product_uses]", uses_sql)
+        self.assertIn("[Agrosupport].dbo.[label_chunks]", chunks_sql)
+
+    def test_advice_reads_qualify_advisories(self) -> None:
+        as_of = date(2026, 4, 16)
+        with patch("pipeline.stages.advice_context.fetch_one", return_value=None) as fetch:
+            load_last_advice(MagicMock(), 4012, "downy mildew", as_of, self.split)
+            load_previous_incubation_register(MagicMock(), 4012, "downy mildew", as_of, self.split)
+        for call in fetch.call_args_list:
+            self.assertIn("[Agrosupport].dbo.[advisories]", call.args[1])
+
+    def test_advice_write_qualifies_advisories(self) -> None:
+        result = AdviceResult(
+            full_advice="text",
+            advice_summary="summary",
+            is_treatment_recommended=False,
+            recommended_product_names=[],
+            recommended_doses=[],
+            recommended_application_date=None,
+        )
+        with (
+            patch("pipeline.stages.advice.transaction", MagicMock()),
+            patch("pipeline.stages.advice.execute") as execute,
+        ):
+            insert_advice(
+                MagicMock(),
+                issuance_date=date(2026, 4, 16),
+                field_id=4012,
+                disease_english="downy mildew",
+                advice_input_json={},
+                result=result,
+                sql=self.split,
+            )
+        delete_sql, insert_sql = (call.args[1] for call in execute.call_args_list)
+        self.assertIn("DELETE FROM [Agrosupport].dbo.[advisories]", delete_sql)
+        self.assertIn("INSERT INTO [Agrosupport].dbo.[advisories]", insert_sql)
+
+
+def _called_name(node: ast.Call) -> str | None:
+    func = node.func
+    if isinstance(func, ast.Name):
+        return func.id
+    if isinstance(func, ast.Attribute):
+        return func.attr
+    return None
+
+
+def _passes_sql(node: ast.Call) -> bool:
+    values = list(node.args) + [keyword.value for keyword in node.keywords]
+    return any(isinstance(value, ast.Attribute) and value.attr == "sql" for value in values)
+
+
+class CallSiteTests(unittest.TestCase):
+    def _calls(self, relative: str) -> list[ast.Call]:
+        tree = ast.parse((_ROOT / relative).read_text(encoding="utf-8"))
+        return [node for node in ast.walk(tree) if isinstance(node, ast.Call)]
+
+    def test_job_passes_sql_settings(self) -> None:
+        calls = self._calls("pipeline/job.py")
+        expected = {
+            "load_previous_incubation_register",
+            "last_treatment_interval_open",
+            "incubation_start_covered",
+            "load_last_advice",
+            "insert_advice",
+        }
+        found = {name for node in calls if (name := _called_name(node)) in expected}
+        self.assertEqual(found, expected)
+        for node in calls:
+            if _called_name(node) in expected:
+                self.assertTrue(_passes_sql(node), _called_name(node))
+
+    def test_resources_pass_sql_settings(self) -> None:
+        calls = self._calls("pipeline/resources.py")
+        load = [node for node in calls if _called_name(node) == "load"]
+        builder = [node for node in calls if _called_name(node) == "ProductJsonBuilder"]
+        self.assertEqual(len(load), 1)
+        self.assertEqual(len(builder), 1)
+        self.assertTrue(_passes_sql(load[0]))
+        self.assertTrue(_passes_sql(builder[0]))
+
+    def test_batch_existing_advice_uses_app_database(self) -> None:
+        calls = self._calls("pipeline/batch.py")
+        qualify = [node for node in calls if _called_name(node) == "qualify_app_sql"]
+        self.assertEqual(len(qualify), 1)
+        self.assertTrue(_passes_sql(qualify[0]))
+        module = ast.parse((_ROOT / "pipeline/batch.py").read_text(encoding="utf-8"))
+        constant = next(
+            node.value.value
+            for node in module.body
+            if isinstance(node, ast.Assign)
+            and isinstance(node.targets[0], ast.Name)
+            and node.targets[0].id == "_EXISTING_ADVICE_SQL"
+            and isinstance(node.value, ast.Constant)
+        )
+        qualified = qualify_app_sql(_sql(app_database="Agrosupport"), constant)
+        self.assertIn("FROM [Agrosupport].dbo.[advisories]", qualified)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_incubations.py

@@ -7,6 +7,7 @@ from datetime import date, timedelta
 from types import SimpleNamespace
 from unittest.mock import MagicMock, patch
 
+from pipeline.config import SqlSettings
 from pipeline.stages.disease import select_prefilter_rule
 from pipeline.stages.incubations import (
     CoveringTreatment,
@@ -20,6 +21,14 @@ from pipeline.stages.incubations import (
 from pipeline.stages.products import ProductIndex, ProductRow
 from pipeline.stages.treatments import TreatmentHistory
 
+_SQL = SqlSettings(
+    driver="ODBC Driver 17 for SQL Server",
+    server="localhost",
+    database="DBMeteo",
+    app_database="DBMeteo",
+    username="user",
+    password="secret",
+)
 _AS_OF = date(2026, 4, 16)
 _PAST = _AS_OF - timedelta(days=2)
 _TODAY = _AS_OF
@@ -194,6 +203,7 @@ class CoverageTests(unittest.TestCase):
                 disease_english="downy mildew",
                 start_date=_PAST,
                 treatment_history=self._history(_PAST),
+                sql=_SQL,
             )
         self.assertIsNone(covered)
         fetch.assert_not_called()
@@ -209,6 +219,7 @@ class CoverageTests(unittest.TestCase):
                 disease_english="downy mildew",
                 start_date=_PAST,
                 treatment_history=self._history(spray),
+                sql=_SQL,
             )
         self.assertIsNotNone(covered)
         assert covered is not None

tests/test_products.py

@@ -7,6 +7,7 @@ from datetime import date
 from types import SimpleNamespace
 from unittest.mock import MagicMock, patch
 
+from pipeline.config import SqlSettings
 from pipeline.stages.products import (
     ProductIndex,
     ProductRow,
@@ -21,6 +22,14 @@ _AS_OF = date(2026, 4, 16)
 _SPRAY_DATE = date(2026, 4, 6)  # 10 days before as_of
 _CROP = "grapevine"
 _DISEASE = "downy mildew"
+_SQL = SqlSettings(
+    driver="ODBC Driver 17 for SQL Server",
+    server="localhost",
+    database="DBMeteo",
+    app_database="DBMeteo",
+    username="user",
+    password="secret",
+)
 
 
 def _product(
@@ -66,6 +75,7 @@ def _interval_open(
             disease_english=_DISEASE,
             latest_spray=(spray_date, names),
             as_of=as_of,
+            sql=_SQL,
         )
     fetch.assert_called_once()
     return result
@@ -82,6 +92,7 @@ class LastTreatmentIntervalOpenTests(unittest.TestCase):
                 disease_english=_DISEASE,
                 latest_spray=None,
                 as_of=_AS_OF,
+                sql=_SQL,
             )
         fetch.assert_not_called()
         self.assertFalse(result)
@@ -172,6 +183,7 @@ class LastTreatmentIntervalOpenTests(unittest.TestCase):
                 disease_english=_DISEASE,
                 latest_spray=(_SPRAY_DATE, ["UNKNOWN"]),
                 as_of=_AS_OF,
+                sql=_SQL,
             )
         fetch.assert_not_called()
         self.assertFalse(result)
@@ -361,7 +373,7 @@ class ProductIndexLoadTests(unittest.TestCase):
             "pipeline.stages.products.fetch_all",
             side_effect=[products, catalog],
         ) as fetch:
-            index = ProductIndex.load(MagicMock())
+            index = ProductIndex.load(MagicMock(), _SQL)
 
         self.assertEqual(fetch.call_count, 2)
         self.assertIn("FROM products", fetch.call_args_list[0].args[1])

tests/test_worklist.py

@@ -0,0 +1,107 @@
+"""Unit tests for database-driven worklist discovery."""
+
+from __future__ import annotations
+
+import unittest
+from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import patch
+
+from pipeline.config import WorklistSettings
+from pipeline.vocab import load_crop_vocab, load_disease_vocab
+from pipeline.worklist import build_jobs
+
+_ROOT = Path(__file__).resolve().parent.parent
+_CROP_VOCAB = load_crop_vocab(_ROOT / "vocab" / "crops.yaml")
+_DISEASE_VOCAB = load_disease_vocab(_ROOT / "vocab" / "diseases.yaml")
+
+
+def _row(**overrides: object) -> SimpleNamespace:
+    fields = {
+        "cmplay_id": 4012,
+        "cmplay_colture": 1,
+        "cmplay_imp": 3,
+        "cmplay_station": 7,
+        "colture_name": "Vite",
+        "anmod_id": 10,
+        "anmod_model": "PERONOSPORA",
+    }
+    fields.update(overrides)
+    return SimpleNamespace(**fields)
+
+
+def _build(rows: list[SimpleNamespace], allowlist: tuple[int, ...] = ()) -> tuple[list, str]:
+    worklist = WorklistSettings(field_allowlist=allowlist)
+    with patch("pipeline.worklist.fetch_all", return_value=rows) as fetch:
+        jobs = build_jobs(object(), worklist, _CROP_VOCAB, _DISEASE_VOCAB)
+    sql = fetch.call_args.args[1]
+    return jobs, sql
+
+
+class BuildJobsTests(unittest.TestCase):
+    def test_peronospora_on_grapevine_becomes_one_job(self) -> None:
+        jobs, sql = _build([_row()])
+
+        self.assertEqual(len(jobs), 1)
+        job = jobs[0]
+        self.assertEqual(job.field_id, 4012)
+        self.assertEqual(job.model_name, "PERONOSPORA")
+        self.assertEqual(job.disease_english, "downy mildew")
+        self.assertEqual(job.crop_english, "grapevine")
+        self.assertEqual(job.crop_italian, "Vite")
+        self.assertTrue(job.organic)
+        self.assertEqual(job.station, 7)
+        self.assertEqual(job.anmod_id, 10)
+        self.assertIn("pat_mod_AI_model = 1", sql)
+        self.assertIn("anmod_AI_advice = 1", sql)
+        self.assertIn("anmod_cmplay", sql)
+
+    def test_allowlist_drops_other_field_ids(self) -> None:
+        jobs, _sql = _build(
+            [_row(cmplay_id=4012), _row(cmplay_id=4085, anmod_id=11)],
+            allowlist=(4085,),
+        )
+
+        self.assertEqual([job.field_id for job in jobs], [4085])
+
+    def test_unmapped_disease_and_colture_are_skipped(self) -> None:
+        jobs, _sql = _build([
+            _row(),
+            _row(anmod_id=11, anmod_model="NOT_A_DISEASE"),
+            _row(cmplay_id=4085, anmod_id=12, colture_name="Not A Crop"),
+        ])
+
+        self.assertEqual([(job.field_id, job.anmod_id) for job in jobs], [(4012, 10)])
+
+    def test_blank_and_null_colture_are_skipped(self) -> None:
+        jobs, _sql = _build([
+            _row(cmplay_id=1, anmod_id=1, colture_name=""),
+            _row(cmplay_id=2, anmod_id=2, colture_name=None),
+            _row(cmplay_id=3, anmod_id=3),
+        ])
+
+        self.assertEqual([job.field_id for job in jobs], [3])
+
+    def test_same_english_disease_keeps_distinct_anmod_rows(self) -> None:
+        jobs, _sql = _build([
+            _row(anmod_id=20, anmod_model="PERONOSPORA"),
+            _row(anmod_id=11, anmod_model="Peronospora"),
+        ])
+
+        self.assertEqual(
+            [(job.anmod_id, job.model_name, job.disease_english) for job in jobs],
+            [
+                (20, "PERONOSPORA", "downy mildew"),
+                (11, "Peronospora", "downy mildew"),
+            ],
+        )
+
+    def test_imp_other_than_3_is_not_organic(self) -> None:
+        jobs, _sql = _build([_row(cmplay_imp=1, cmplay_station=None)])
+
+        self.assertFalse(jobs[0].organic)
+        self.assertIsNone(jobs[0].station)
+
+
+if __name__ == "__main__":
+    unittest.main()