diff --git a/README.md b/README.md index 5359913..d4629a1 100644 --- a/README.md +++ b/README.md @@ -42,8 +42,8 @@ docker compose up -d ## Configuration -- [`config.yaml`](config.yaml) — concurrency/rate limits, retry policy, the 09:00 SLA - deadline, LLM provider/model, vocab paths, and an optional field allowlist. +- [`config.yaml`](config.yaml) — concurrency/rate limits, retry policy, the 09:00 + readiness target, 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 @@ -89,7 +89,7 @@ retry: max_backoff_seconds: 30 schedule: - deadline: "09:00" # jobs not started by this local time are skipped + deadline: "09:00" # readiness target; jobs still run after this time ``` ### Prompt registry (`prompts/`) @@ -141,6 +141,7 @@ python -m pipeline one --field-id 4085 --disease PERONOSPORA --dry-run `batch` accepts `--deadline HH:MM` and `--workers N` to override `schedule.deadline` / `concurrency.workers` from `config.yaml` for one run. +Jobs still run after `--deadline`; it is the time advice should be ready, not a cutoff. ## Concurrency model @@ -180,9 +181,10 @@ Each run writes to `output//`: - `field____.json` — one file per job (see fields below) - `_run_report.json` — the operational artifact for the morning: totals by - status plus a per-job breakdown (`status`, `duration_s`, - `recommended_count`, `allowed_products_empty`, `message`, and, on failure, - `error_type` / `error_message`) + status, `jobs_started_after_deadline` (jobs that began after the 09:00 + target; they are still written), plus a per-job breakdown (`status`, + `duration_s`, `recommended_count`, `allowed_products_empty`, `message`, + and, on failure, `error_type` / `error_message`) Per-job payload fields: @@ -260,7 +262,7 @@ table, so concurrent workers don't block each other on unrelated rows. ## Daily scheduling Provide a CLI entry point only. Schedule it yourself on the deployment machine. -Advice must be ready by the `schedule.deadline` (09:00 by default), and weather +Advice should be ready by the `schedule.deadline` (09:00 by default), and weather data should stay fresh, so start the run no earlier than ~07:00 — for example via Windows Task Scheduler, daily at 07:00: @@ -268,7 +270,9 @@ via Windows Task Scheduler, daily at 07:00: d:\Uni\Thesis\AI-Agro-Support\.venv\Scripts\python.exe -m pipeline batch --skip-existing-advice ``` -with working directory `d:\Uni\Thesis\AI-Agro-Support`. `--skip-existing-advice` +with working directory `d:\Uni\Thesis\AI-Agro-Support`. Jobs still waiting at +09:00 are run and written. `_run_report.json` records how many of them started +late in `jobs_started_after_deadline`. `--skip-existing-advice` makes a re-trigger (e.g. a scheduled retry later that morning) resumable at no extra cost: it does one bulk check against `advisories` up front and only runs jobs that don't have a row yet. diff --git a/config.yaml b/config.yaml index 4433d82..575f568 100644 --- a/config.yaml +++ b/config.yaml @@ -64,11 +64,11 @@ retry: initial_backoff_seconds: 2 max_backoff_seconds: 30 -# ── Daily SLA ─────────────────────────────────────────────────────────────────── -# The batch must not still be running after this local time; jobs not yet -# started by then are skipped (status skipped_deadline) rather than risking a -# late advisory. Schedule the run itself no earlier than ~07:00 so weather is -# fresh (see README). +# ── Daily readiness target ────────────────────────────────────────────────────── +# Advice should be ready by this local time. Start the run no earlier than +# ~07:00 so weather is fresh (see README). Jobs not yet started by then are +# still run and written; the run report counts them in +# jobs_started_after_deadline. schedule: deadline: "09:00" diff --git a/pipeline/__main__.py b/pipeline/__main__.py index ce79ae1..8f5dcfb 100644 --- a/pipeline/__main__.py +++ b/pipeline/__main__.py @@ -166,7 +166,10 @@ def main(argv: list[str] | None = None) -> int: ) batch_parser.add_argument( "--deadline", default=None, - help="Override schedule.deadline from config.yaml (HH:MM, local time).", + help=( + "Override schedule.deadline from config.yaml (HH:MM, local time). " + "Advice should be ready by then; jobs still waiting are still run." + ), ) batch_parser.add_argument( "--skip-existing-advice", action="store_true", diff --git a/pipeline/batch.py b/pipeline/batch.py index b41f673..43ba514 100644 --- a/pipeline/batch.py +++ b/pipeline/batch.py @@ -5,7 +5,7 @@ One `ThreadPoolExecutor` runs every discovered (field, disease) job so a slow crop cannot serialise the others; `pipeline.job.run_job` isolates each job's failures from the rest. This module owns everything above that: worklist discovery + CLI filters, the `--skip-existing-advice` resume check, the -09:00-style deadline, and writing the run report. +09:00 readiness target, and writing the run report. """ from __future__ import annotations @@ -32,13 +32,14 @@ _EXISTING_ADVICE_SQL = "SELECT field_id, disease FROM {advisories} WHERE issuanc def _parse_deadline(deadline_str: str) -> datetime: """ - Build today's deadline datetime from an "HH:MM" string. + Build today's readiness-target datetime from an "HH:MM" string. Deliberately anchored to the wall-clock date the batch is actually - running on, not `as_of`: the SLA ("advice ready by 09:00") is about when - the run happens, and in production `as_of` is always today anyway. + running on, not `as_of`: the target ("advice ready by 09:00") is about + when the run happens, and in production `as_of` is always today anyway. Anchoring to `as_of` instead would make every historical/backfill run - (`--as-of` in the past) look like its deadline had already passed. + (`--as-of` in the past) look like its target had already passed. + Jobs that start after the target still run. """ try: hour_str, minute_str = deadline_str.split(":", 1) @@ -47,6 +48,29 @@ def _parse_deadline(deadline_str: str) -> datetime: raise ValueError(f"Invalid deadline '{deadline_str}'. Expected HH:MM.") from exc +def _deadline_notice( + now: datetime, + deadline: datetime, + batch_start: datetime, +) -> str | None: + """Classify `now` against the readiness target. Never means "skip". + + "late" — the target has passed; the job should still run. + "budget_80" — at least 80% of the time before the target has been used. + None — neither. A batch that has already started past the target is + "late"; the 80% branch does not run in that case. + """ + if now >= deadline: + return "late" + total_budget_s = (deadline - batch_start).total_seconds() + if total_budget_s <= 0: + return None + elapsed_s = (now - batch_start).total_seconds() + if elapsed_s >= 0.8 * total_budget_s: + return "budget_80" + return None + + def filter_jobs( jobs: list[Job], *, @@ -139,40 +163,44 @@ def run_batch( deadline_str = deadline_override or settings.schedule.deadline deadline = _parse_deadline(deadline_str) batch_start = datetime.now() - total_budget_s = (deadline - batch_start).total_seconds() warned_80pct = False + warned_late = False + late_starts = 0 warn_lock = threading.Lock() def _run_one(job: Job) -> JobResult: - nonlocal warned_80pct + nonlocal warned_80pct, warned_late, late_starts now = datetime.now() - if now >= deadline: - logger.error( - "Deadline %s reached; skipping field=%s crop=%s disease=%s", - deadline.isoformat(timespec="minutes"), - job.field_id, - job.crop_english, - job.disease_english, - ) - return JobResult(job=job, status="skipped_deadline") - - if total_budget_s > 0: - elapsed_s = (now - batch_start).total_seconds() - if elapsed_s >= 0.8 * total_budget_s: - with warn_lock: - already_warned, warned_80pct = warned_80pct, True - if not already_warned: - logger.warning( - "80%% of the time budget before the %s deadline has been " - "used; some of the remaining %d job(s) may be skipped.", - deadline_str, - len(jobs), - ) + notice = _deadline_notice(now, deadline, batch_start) + started_late = notice == "late" + if notice == "late": + with warn_lock: + first_late = not warned_late + warned_late = True + if first_late: + logger.warning( + "Target %s has passed; continuing remaining jobs so every " + "field still receives advice.", + deadline_str, + ) + elif notice == "budget_80": + with warn_lock: + first_80, warned_80pct = not warned_80pct, True + if first_80: + logger.warning( + "80%% of the time before the %s target has been used; " + "the run may finish after the target, and remaining jobs " + "will still run.", + deadline_str, + ) if (job.field_id, job.disease_english) in existing: return JobResult(job=job, status="skipped_existing") + if started_late: + with warn_lock: + late_starts += 1 return run_job(resources, job, as_of, dry_run=dry_run) # Worker threads start with an empty OpenTelemetry context @@ -207,7 +235,12 @@ def run_batch( if result.payload is not None: write_job_output(settings.output_dir, as_of, result.payload) - report = write_run_report(settings.output_dir, as_of, results) + report = write_run_report( + settings.output_dir, + as_of, + results, + jobs_started_after_deadline=late_starts, + ) logger.info("Batch complete: %s", {k: v for k, v in report["totals"].items() if v}) return 0 if report["totals"].get("failed", 0) == 0 else 1 finally: diff --git a/pipeline/config.py b/pipeline/config.py index aac3c0a..9fa8482 100644 --- a/pipeline/config.py +++ b/pipeline/config.py @@ -62,7 +62,7 @@ class RetrySettings: @dataclass(frozen=True) class ScheduleSettings: - deadline: str # "HH:MM", local time + deadline: str # "HH:MM", local readiness target; jobs still run after it @dataclass(frozen=True) diff --git a/pipeline/job.py b/pipeline/job.py index 446d520..7417b24 100644 --- a/pipeline/job.py +++ b/pipeline/job.py @@ -65,10 +65,10 @@ logger = logging.getLogger(__name__) class JobResult: """Outcome of one job. Constructed either by `run_job` or, for jobs the batch orchestrator decides not to run at all, directly by `pipeline.batch` - (`skipped_existing`, `skipped_deadline`).""" + (`skipped_existing`).""" job: Job - status: str # "ok" | "no_risk" | "dry_run" | "failed" | "skipped_existing" | "skipped_deadline" + status: str # "ok" | "no_risk" | "dry_run" | "failed" | "skipped_existing" duration_s: float = 0.0 error_type: str | None = None error_message: str | None = None diff --git a/pipeline/output.py b/pipeline/output.py index 20af094..943e0bc 100644 --- a/pipeline/output.py +++ b/pipeline/output.py @@ -27,7 +27,6 @@ _STATUSES = ( "dry_run", "failed", "skipped_existing", - "skipped_deadline", ) @@ -67,6 +66,8 @@ def write_run_report( output_dir: Path, as_of: date, results: list[JobResult], + *, + jobs_started_after_deadline: int = 0, ) -> dict[str, Any]: """ Write output//_run_report.json: totals by status plus per-job @@ -75,11 +76,12 @@ def write_run_report( This is the operational artifact for the daily run: it names exactly which jobs succeeded, which were skipped and why, and which failed and with what error, so the exit code alone never has to carry that - information. `ok_with_empty_allowed_products` is tracked separately from - a bare count of `ok` jobs because, once the `products` table has organic - entries for every crop, a sudden jump in that number is the fastest - signal that the prefilter or the Weaviate allowlist filter has regressed - (see pipeline/stages/vector.py). + information. `jobs_started_after_deadline` counts jobs that began after + the readiness target (they are still run). `ok_with_empty_allowed_products` + is tracked separately from a bare count of `ok` jobs because, once the + `products` table has organic entries for every crop, a sudden jump in + that number is the fastest signal that the prefilter or the Weaviate + allowlist filter has regressed (see pipeline/stages/vector.py). """ totals = {status: 0 for status in _STATUSES} for result in results: @@ -90,6 +92,7 @@ def write_run_report( report: dict[str, Any] = { "as_of": as_of.isoformat(), "job_count": len(results), + "jobs_started_after_deadline": jobs_started_after_deadline, "totals": totals, "ok_with_empty_allowed_products": empty_allowed_products, "jobs": [_job_report_entry(r) for r in results], diff --git a/tests/test_batch.py b/tests/test_batch.py new file mode 100644 index 0000000..ba5161e --- /dev/null +++ b/tests/test_batch.py @@ -0,0 +1,235 @@ +"""Unit tests for the 09:00 readiness target: late jobs still run.""" + +from __future__ import annotations + +import tempfile +import unittest +from datetime import date, datetime, timedelta +from datetime import time as dtime +from pathlib import Path +from unittest.mock import MagicMock, patch + +from pipeline.batch import _deadline_notice, run_batch +from pipeline.config import ( + ConcurrencySettings, + LlmSettings, + ObservabilitySettings, + RetrySettings, + ScheduleSettings, + Settings, + SqlSettings, + WeaviateSettings, + WorklistSettings, +) +from pipeline.job import JobResult +from pipeline.output import write_run_report +from pipeline.worklist import Job + + +def _at(hour: int, minute: int = 0, second: int = 0) -> datetime: + return datetime.combine(date.today(), dtime(hour, minute, second)) + + +def _job(field_id: int = 4012) -> Job: + return Job( + field_id=field_id, + crop_english="grapevine", + crop_italian="Vite", + model_name="PERONOSPORA", + disease_english="downy mildew", + organic=False, + station=7, + anmod_id=10, + ) + + +def _settings() -> Settings: + llm = LlmSettings( + provider="gemini", + openai_model="m", + anthropic_model="m", + gemini_model="m", + max_tokens=8, + temperature=0.0, + thinking_level=None, + ) + return Settings( + root=Path("."), + field_id=None, + disease_name=None, + worklist=WorklistSettings(field_allowlist=()), + concurrency=ConcurrencySettings( + workers=1, + sql_limit=1, + gemini_limit=1, + gemini_requests_per_minute=60, + ), + retry=RetrySettings(attempts=1, initial_backoff_seconds=0.0, max_backoff_seconds=0.0), + schedule=ScheduleSettings(deadline="09:00"), + observability=ObservabilitySettings( + enabled=False, + project_name="p", + endpoint="http://localhost", + hide_prompts=True, + max_attribute_chars=100, + ), + crops_vocab=Path("crops.yaml"), + diseases_vocab=Path("diseases.yaml"), + llm_query_synthesis=llm, + llm_advice_generation=llm, + sql=SqlSettings( + driver="d", + server="localhost", + database="db", + app_database="db", + username="u", + password="p", + ), + weaviate=WeaviateSettings(host="localhost", http_port=8080, grpc_port=50051), + gemini_api_key="", + openai_api_key="", + anthropic_api_key="", + prompts_dir=Path("prompts"), + output_dir=Path("output"), + ) + + +class DeadlineNoticeTests(unittest.TestCase): + def test_before_80_percent_is_quiet(self) -> None: + start = datetime(2026, 10, 14, 7, 0) + deadline = datetime(2026, 10, 14, 9, 0) + now = start + timedelta(seconds=0.8 * 7200 - 1) + self.assertIsNone(_deadline_notice(now, deadline, start)) + + def test_at_80_percent_and_still_before_deadline(self) -> None: + start = datetime(2026, 10, 14, 7, 0) + deadline = datetime(2026, 10, 14, 9, 0) + now = start + timedelta(seconds=0.8 * 7200) + self.assertLess(now, deadline) + self.assertEqual(_deadline_notice(now, deadline, start), "budget_80") + + def test_exactly_at_deadline_is_late(self) -> None: + start = datetime(2026, 10, 14, 7, 0) + deadline = datetime(2026, 10, 14, 9, 0) + self.assertEqual(_deadline_notice(deadline, deadline, start), "late") + + def test_after_deadline_is_late(self) -> None: + start = datetime(2026, 10, 14, 7, 0) + deadline = datetime(2026, 10, 14, 9, 0) + now = datetime(2026, 10, 14, 9, 5) + self.assertEqual(_deadline_notice(now, deadline, start), "late") + + def test_batch_already_started_past_deadline_is_late(self) -> None: + deadline = datetime(2026, 10, 14, 9, 0) + start = datetime(2026, 10, 14, 10, 0) + now = datetime(2026, 10, 14, 10, 1) + self.assertEqual(_deadline_notice(now, deadline, start), "late") + + +class RunBatchDeadlineTests(unittest.TestCase): + def _run( + self, + jobs: list[Job], + *, + now, + skip_existing: bool = False, + existing: set[tuple[int, str]] | None = None, + ) -> dict: + captured: dict = {} + prompts = MagicMock() + prompts.validate.return_value = [] + + def _write_report(output_dir, as_of, results, *, jobs_started_after_deadline=0): + captured["results"] = list(results) + captured["late"] = jobs_started_after_deadline + return {"totals": {}} + + def _run_job(resources, job, as_of, dry_run=False): + return JobResult(job=job, status="ok", payload={"output_stem": str(job.field_id)}) + + now_fn = now if callable(now) else (lambda tz=None: now) + with ( + patch("pipeline.batch.PromptRegistry", return_value=prompts), + patch("pipeline.batch.Resources", return_value=MagicMock()), + patch("pipeline.batch.build_jobs", return_value=jobs), + patch("pipeline.batch.run_job", side_effect=_run_job) as run_job, + patch("pipeline.batch.write_job_output"), + patch("pipeline.batch.write_run_report", side_effect=_write_report), + patch("pipeline.batch._existing_advice_keys", return_value=existing or set()), + patch("pipeline.batch.datetime") as frozen, + self.assertLogs("pipeline.batch", level="WARNING") as logs, + ): + frozen.now.side_effect = now_fn + frozen.combine = datetime.combine + captured["code"] = run_batch( + _settings(), + MagicMock(), + MagicMock(), + date(2026, 10, 14), + skip_existing_advice=skip_existing, + ) + captured["run_job"] = run_job + captured["logs"] = [record.getMessage() for record in logs.records] + return captured + + def test_job_after_deadline_still_runs(self) -> None: + captured = self._run([_job()], now=_at(10, 0)) + + self.assertEqual(captured["run_job"].call_count, 1) + self.assertEqual(captured["late"], 1) + self.assertEqual(captured["code"], 0) + self.assertEqual([result.status for result in captured["results"]], ["ok"]) + self.assertTrue(all(result.status != "skipped_deadline" for result in captured["results"])) + self.assertEqual(sum("has passed" in message for message in captured["logs"]), 1) + + def test_two_late_jobs_warn_once(self) -> None: + jobs = [_job(4012), _job(4013)] + captured = self._run(jobs, now=_at(10, 0)) + + self.assertEqual(captured["run_job"].call_count, 2) + self.assertEqual(captured["late"], 2) + self.assertEqual(sum("has passed" in message for message in captured["logs"]), 1) + + def test_existing_advice_after_deadline_is_not_a_late_start(self) -> None: + job = _job() + captured = self._run( + [job], + now=_at(10, 0), + skip_existing=True, + existing={(job.field_id, job.disease_english)}, + ) + + self.assertEqual(captured["run_job"].call_count, 0) + self.assertEqual(captured["late"], 0) + self.assertEqual([result.status for result in captured["results"]], ["skipped_existing"]) + + def test_budget_warning_is_logged_once(self) -> None: + start = _at(7, 0) + at_80 = start + timedelta(seconds=0.8 * 2 * 3600) + later = at_80 + timedelta(minutes=5) + times = iter([start, at_80, later]) + captured = self._run([_job(4012), _job(4013)], now=lambda tz=None: next(times)) + + self.assertEqual(captured["run_job"].call_count, 2) + self.assertEqual(captured["late"], 0) + self.assertEqual(sum("80%" in message for message in captured["logs"]), 1) + self.assertFalse(any("has passed" in message for message in captured["logs"])) + + +class RunReportTests(unittest.TestCase): + def test_report_records_late_starts_without_skipped_deadline(self) -> None: + result = JobResult(job=_job(), status="ok") + with tempfile.TemporaryDirectory() as tmp: + report = write_run_report( + Path(tmp), + date(2026, 10, 14), + [result], + jobs_started_after_deadline=2, + ) + empty = write_run_report(Path(tmp), date(2026, 10, 14), []) + + self.assertEqual(report["jobs_started_after_deadline"], 2) + self.assertNotIn("skipped_deadline", report["totals"]) + self.assertIn("skipped_existing", report["totals"]) + self.assertEqual(empty["jobs_started_after_deadline"], 0) + self.assertNotIn("skipped_deadline", empty["totals"])