Arsham/AI_Agro_Support
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Refactor scheduling logic and update documentation for readiness target

Browse Source 
- Changed terminology in `config.yaml` from "SLA deadline" to "readiness target" to clarify that jobs still run after the specified time.
- Updated `README.md` to reflect the new readiness target concept, emphasizing that jobs started after the target are still executed and reported.
- Modified `pipeline` logic to incorporate the new readiness target, ensuring accurate tracking of jobs that begin after the target time.
- Enhanced reporting in the run report to include counts of jobs that started after the readiness target, improving transparency in job execution outcomes.
This commit is contained in:
Arsham Mirehvandi 2026-10-14 13:55:44 +02:00
parent f77c0abebd
commit b8f74bd027
8 changed files with 331 additions and 53 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
README.md
View File

@ -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/<YYYY-MM-DD>/`:
- `field<field_id>__<crop>__<disease slug>.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.

10
config.yaml
View File

@ -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"

5
pipeline/__main__.py
View File

@ -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",

93
pipeline/batch.py
View File

@ -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:

2
pipeline/config.py
View File

@ -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)

4
pipeline/job.py
View File

@ -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

15
pipeline/output.py
View File

@ -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/<as_of>/_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],

235
tests/test_batch.py Normal file
View File

@ -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"])