From bac40ff26898df2f22988762b2b53c06744c119c Mon Sep 17 00:00:00 2001 From: h Date: Tue, 1 Sep 2026 23:13:17 +0200 Subject: [PATCH] feat(scheduler,conversations,api): cron and schedule in the gateway tz, job runs persisted --- src/beaver_gateway/cli.py | 1 + src/beaver_gateway/core/conversations.py | 11 +- src/beaver_gateway/core/scheduler.py | 145 +++++++++++++++---- src/beaver_gateway/core/transcript.py | 4 +- src/beaver_gateway/frontends/api/frontend.py | 20 ++- src/beaver_gateway/storage/models.py | 27 ++++ tests/test_api.py | 37 ++++- tests/test_conversations.py | 4 + tests/test_scheduler.py | 107 +++++++++++++- ui/src/lib/api/types.ts | 13 +- 10 files changed, 326 insertions(+), 43 deletions(-) diff --git a/src/beaver_gateway/cli.py b/src/beaver_gateway/cli.py index 467181d..5fecf99 100644 --- a/src/beaver_gateway/cli.py +++ b/src/beaver_gateway/cli.py @@ -210,6 +210,7 @@ async def _async_main() -> None: rotation=Rotation( conversations, gateway.rotation or RotationPolicy(tz=gateway.tz) ), + tz=gateway.tz, ) conversations.scheduler = scheduler diff --git a/src/beaver_gateway/core/conversations.py b/src/beaver_gateway/core/conversations.py index 4115d0f..32100ed 100644 --- a/src/beaver_gateway/core/conversations.py +++ b/src/beaver_gateway/core/conversations.py @@ -23,7 +23,7 @@ import logging import re import uuid from dataclasses import dataclass, field -from datetime import UTC, date, datetime, timedelta +from datetime import UTC, date, datetime, timedelta, tzinfo from typing import TYPE_CHECKING, Any, cast from claude_agent_sdk import ( @@ -298,6 +298,10 @@ class Conversations: self._tasks: set[asyncio.Task[None]] = set() self._idle_task: asyncio.Task[None] | None = None + @property + def db(self) -> Database: + return self._db + @property def queue(self) -> InjectQueue: return self._queue @@ -964,6 +968,7 @@ class Conversations: if result.text.strip(): await self.inject(parent, result.text, urgency="normal", origin="слив") await self.set_status(conv, "merged") + await self.mark_closed(conv) self._bus.publish( "conversation.merged", conversation_id=conv.external_id, @@ -1946,7 +1951,7 @@ class Conversations: ) -def parse_at(at: str) -> datetime: +def parse_at(at: str, tz: tzinfo = UTC) -> datetime: raw = at.strip() match = _RELATIVE.match(raw.replace(" ", "")) if match: @@ -1954,7 +1959,7 @@ def parse_at(at: str) -> datetime: return datetime.now(UTC) + timedelta(seconds=int(amount) * _UNITS[unit]) parsed = datetime.fromisoformat(raw) if parsed.tzinfo is None: - parsed = parsed.astimezone() + parsed = parsed.replace(tzinfo=tz) return parsed.astimezone(UTC) diff --git a/src/beaver_gateway/core/scheduler.py b/src/beaver_gateway/core/scheduler.py index 95f126c..984331d 100644 --- a/src/beaver_gateway/core/scheduler.py +++ b/src/beaver_gateway/core/scheduler.py @@ -14,19 +14,27 @@ from __future__ import annotations import asyncio import contextlib +import functools import json import logging +import traceback from dataclasses import dataclass -from datetime import UTC, datetime, timedelta +from datetime import UTC, datetime, timedelta, tzinfo from typing import TYPE_CHECKING, Any, cast +from zoneinfo import ZoneInfo +from croniter import croniter from pgqueuer import PgQueuer, Queries +from pgqueuer.core.executors import ScheduleExecutor from pgqueuer.models import JobId +from sqlalchemy import func +from sqlmodel import col, select from starlette.applications import Starlette from starlette.responses import JSONResponse from starlette.routing import Route from beaver_gateway.core.conversations import parse_at +from beaver_gateway.storage.models import JobRunRecord if TYPE_CHECKING: from collections.abc import Awaitable, Callable, Coroutine, Sequence @@ -42,12 +50,27 @@ if TYPE_CHECKING: from beaver_gateway.core.rotation import Rotation from beaver_gateway.storage.models import Conversation -__all__ = ["INJECT", "Budget", "Job", "JobRun", "Scheduler"] +__all__ = ["INJECT", "Budget", "Job", "JobRun", "LocalCron", "Scheduler", "next_run"] _log = logging.getLogger("beaver_gateway.core.scheduler") INJECT = "inject" RETRY = timedelta(minutes=15) +ERROR_MAX = 2000 + + +def next_run(expression: str, tz: tzinfo, now: datetime | None = None) -> datetime: + """The cron's next fire time, read in ``tz``, returned in UTC.""" + start = (now or datetime.now(UTC)).astimezone(tz) + return croniter(expression, start_time=start).get_next(datetime).astimezone(UTC) + + +@dataclass +class LocalCron(ScheduleExecutor): + tz: tzinfo = UTC + + def get_next(self) -> datetime: + return next_run(self.parameters.expression, self.tz) @dataclass(frozen=True, slots=True) @@ -165,9 +188,11 @@ class Scheduler: budget: Budget | None = None, rotation: Rotation | None = None, heartbeat: timedelta = timedelta(seconds=30), + tz: str = "UTC", ) -> None: self.conversations = conversations self.rotation = rotation + self.tz = ZoneInfo(tz) self.budget = budget or Budget() self._jobs = {job.name: job for job in jobs} self._driver = driver @@ -229,7 +254,12 @@ class Scheduler: if job.cron: - @pgq.schedule(job.name, job.cron, clean_old=True) + @pgq.schedule( + job.name, + job.cron, + clean_old=True, + executor_factory=functools.partial(LocalCron, tz=self.tz), + ) async def cron(_schedule: PgSchedule) -> None: await self._dispatch(job, trigger="cron", payload={}) @@ -257,19 +287,33 @@ class Scheduler: return started = datetime.now(UTC) bus.publish("job.start", job=job.name, trigger=trigger) - status = "done" + status, error = "done", None try: await job.run(JobRun(job, trigger, payload, self)) - except Exception: # noqa: BLE001 - status = "failed" + except Exception as exc: # noqa: BLE001 + status, error = "failed", _error_text(exc) _log.exception("job %s (%s) failed", job.name, trigger) - self._runs[job.name] = { - "trigger": trigger, - "started_at": started.isoformat(timespec="seconds"), - "status": status, - } + record = JobRunRecord( + job=job.name, + trigger=trigger, + started_at=started, + finished_at=datetime.now(UTC), + status=status, + error=error, + payload=payload, + ) + await self._record(record) + self._runs[job.name] = self._run_public(record) bus.publish("job.end", job=job.name, trigger=trigger, status=status) + async def _record(self, record: JobRunRecord) -> None: + try: + async with self.conversations.db.session() as session: + session.add(record) + await session.commit() + except Exception: # noqa: BLE001 + _log.exception("job %s: run not recorded", record.job) + async def trigger( self, job: Job, @@ -325,14 +369,14 @@ class Scheduler: if self._queries is None: msg = "scheduler needs postgres; `schedule` is unavailable" raise RuntimeError(msg) - when = parse_at(at) + when = parse_at(at, self.tz).astimezone(self.tz) delay = max(when - datetime.now(UTC), timedelta(0)) payload = _encode( { "conversation": conv.external_id, "text": text, "urgency": urgency, - "at": when.isoformat(timespec="seconds"), + "at": when.isoformat(timespec="minutes"), } ) ids = await self._queries.enqueue( @@ -347,7 +391,7 @@ class Scheduler: "schedule.created", conversation_id=conv.external_id, job=job_id, - execute_at=when.isoformat(timespec="seconds"), + execute_at=when.isoformat(timespec="minutes"), ) return job_id, when @@ -411,16 +455,22 @@ class Scheduler: for row in await self._queries.peek_schedule(): crons[str(row.entrypoint)] = row queue = [ - _job_public(row) for row in await self._queries.browse_queue(limit=200) + _job_public(row, self.tz) + for row in await self._queries.browse_queue(limit=200) ] utilization = await self.utilization() + last = await self._last_runs() return { "enabled": self.enabled, + "tz": str(self.tz), "utilization": utilization, "throttled": utilization is not None and utilization > self.budget.threshold, "threshold": self.budget.threshold, - "jobs": [self._job_public(job, crons.get(job.name)) for job in self.jobs], + "jobs": [ + self._job_public(job, crons.get(job.name), last.get(job.name)) + for job in self.jobs + ], "queue": queue, } @@ -428,24 +478,60 @@ class Scheduler: if self._queries is None: return [] rows = await self._queries.browse_queue(limit=500, entrypoints=[INJECT]) - out = [_job_public(row) for row in rows] + out = [_job_public(row, self.tz) for row in rows] if conv is not None: out = [ j for j in out if j["payload"].get("conversation") == conv.external_id ] return out - def _job_public(self, job: Job, cron: PgSchedule | None) -> dict[str, Any]: + async def runs(self, name: str, limit: int = 50) -> list[dict[str, Any]]: + stmt = ( + select(JobRunRecord) + .where(JobRunRecord.job == name) + .order_by(col(JobRunRecord.id).desc()) + .limit(limit) + ) + async with self.conversations.db.session() as session: + rows = (await session.exec(stmt)).all() + return [self._run_public(row) for row in rows] + + async def _last_runs(self) -> dict[str, dict[str, Any]]: + newest = select(func.max(JobRunRecord.id)).group_by(JobRunRecord.job) + stmt = select(JobRunRecord).where(col(JobRunRecord.id).in_(newest)) + try: + async with self.conversations.db.session() as session: + rows = (await session.exec(stmt)).all() + except Exception: # noqa: BLE001 + _log.exception("job runs unreadable, using memory") + return dict(self._runs) + return {row.job: self._run_public(row) for row in rows} + + def _job_public( + self, job: Job, cron: PgSchedule | None, run: dict[str, Any] | None + ) -> dict[str, Any]: return { "name": job.name, "cron": job.cron, "webhook": job.webhook, "events": list(job.events), "critical": job.critical, - "next_run": _iso(cron.next_run) if cron is not None else None, - "last_run": _iso(cron.last_run) if cron is not None else None, + "next_run": _iso(cron.next_run, self.tz) if cron is not None else None, + "last_run": _iso(cron.last_run, self.tz) if cron is not None else None, "status": str(cron.status) if cron is not None else None, - "run": self._runs.get(job.name), + "run": run or self._runs.get(job.name), + } + + def _run_public(self, row: JobRunRecord) -> dict[str, Any]: + return { + "id": row.id, + "job": row.job, + "trigger": row.trigger, + "started_at": _iso(row.started_at, self.tz), + "finished_at": _iso(row.finished_at, self.tz), + "status": row.status, + "error": row.error, + "payload": row.payload, } # ---- http ---------------------------------------------------------- @@ -506,21 +592,28 @@ def _decode(payload: bytes | None) -> dict[str, Any]: return data if isinstance(data, dict) else {"raw": data} -def _job_public(row: PgJob) -> dict[str, Any]: +def _job_public(row: PgJob, tz: tzinfo) -> dict[str, Any]: return { "id": int(row.id), "entrypoint": row.entrypoint, "status": str(row.status), - "execute_after": _iso(row.execute_after), - "created": _iso(row.created), + "execute_after": _iso(row.execute_after, tz), + "created": _iso(row.created, tz), "attempts": row.attempts, "payload": _decode(row.payload), } +def _error_text(exc: BaseException) -> str: + head = f"{type(exc).__name__}: {exc}" + return f"{head}\n{''.join(traceback.format_exception(exc))}"[:ERROR_MAX] + + def _aware(value: datetime) -> datetime: return value if value.tzinfo is not None else value.replace(tzinfo=UTC) -def _iso(value: datetime | None) -> str | None: - return _aware(value).isoformat(timespec="seconds") if value is not None else None +def _iso(value: datetime | None, tz: tzinfo = UTC) -> str | None: + if value is None: + return None + return _aware(value).astimezone(tz).isoformat(timespec="seconds") diff --git a/src/beaver_gateway/core/transcript.py b/src/beaver_gateway/core/transcript.py index 609842f..7340431 100644 --- a/src/beaver_gateway/core/transcript.py +++ b/src/beaver_gateway/core/transcript.py @@ -22,7 +22,9 @@ from datetime import UTC, datetime from typing import TYPE_CHECKING, Any try: - from claude_agent_sdk._cli_version import __cli_version__ as _cli_version + from claude_agent_sdk._cli_version import __cli_version__ + + _cli_version = str(__cli_version__) except ImportError: # pragma: no cover _cli_version = "2.1.248" diff --git a/src/beaver_gateway/frontends/api/frontend.py b/src/beaver_gateway/frontends/api/frontend.py index cc46705..b3d1354 100644 --- a/src/beaver_gateway/frontends/api/frontend.py +++ b/src/beaver_gateway/frontends/api/frontend.py @@ -62,7 +62,7 @@ if TYPE_CHECKING: from pathlib import Path from beaver_gateway.core.conversations import Conversations - from beaver_gateway.core.scheduler import Scheduler + from beaver_gateway.core.scheduler import Job, Scheduler from beaver_gateway.frontends.base import GatewayRuntime _log = logging.getLogger("beaver_gateway.frontends.api") @@ -580,15 +580,27 @@ def build_app(runtime: GatewayRuntime, *, memory_root: Path | None = None) -> Fa await require_token(request, runtime, scope=SCOPE) return await scheduler_of().snapshot() + def job_of(scheduler: Scheduler, name: str) -> Job: + job = scheduler.job(name) + if job is None: + raise HTTPException(status.HTTP_404_NOT_FOUND, f"no job {name!r}") + return job + @app.post("/jobs/{name}/run", status_code=status.HTTP_202_ACCEPTED) async def run_job(name: str, request: Request) -> dict[str, Any]: await require_token(request, runtime, scope=SCOPE) scheduler = scheduler_of() - job = scheduler.job(name) - if job is None: - raise HTTPException(status.HTTP_404_NOT_FOUND, f"no job {name!r}") + job = job_of(scheduler, name) return {"job": await scheduler.trigger(job, await body_of(request))} + @app.get("/jobs/{name}/runs") + async def job_runs(name: str, request: Request) -> dict[str, Any]: + await require_token(request, runtime, scope=SCOPE) + scheduler = scheduler_of() + job = job_of(scheduler, name) + limit = min(int(request.query_params.get("limit", 50)), 500) + return {"job": job.name, "runs": await scheduler.runs(job.name, limit)} + @app.delete("/jobs/queue/{job_id}") async def cancel_job(job_id: int, request: Request) -> dict[str, Any]: await require_token(request, runtime, scope=SCOPE) diff --git a/src/beaver_gateway/storage/models.py b/src/beaver_gateway/storage/models.py index c9ce2bb..d5446d9 100644 --- a/src/beaver_gateway/storage/models.py +++ b/src/beaver_gateway/storage/models.py @@ -360,6 +360,32 @@ class RateLimit(SQLModel, table=True): ) +class JobRunRecord(SQLModel, table=True): + """One finished run of a scheduler job (§3.6): what fired it, how it ended. + + ``error`` keeps the head of the traceback of a failed run; ``payload`` + is the trigger's data (``{}`` for cron) and should stay small. + """ + + __tablename__ = "job_runs" + + id: int | None = Field(default=None, primary_key=True) + job: str = Field(index=True) + trigger: str + started_at: datetime + finished_at: datetime + status: str + error: str | None = Field(default=None) + payload: dict[str, Any] = Field( + default_factory=dict, + sa_column=Column( + JSON().with_variant(JSONB(), "postgresql"), + nullable=False, + server_default=text("'{}'"), + ), + ) + + __all__ = [ "AuditLog", "Conversation", @@ -367,6 +393,7 @@ __all__ = [ "ConversationMessage", "Delivery", "InjectQueueItem", + "JobRunRecord", "RateLimit", "TelegramUpdate", "Token", diff --git a/tests/test_api.py b/tests/test_api.py index eb645d2..d005f12 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -21,6 +21,7 @@ from test_conversations import ScriptedClient, World from beaver_gateway.core.conversation_store import rewrite_messages from beaver_gateway.core.auth import TokenStore from beaver_gateway.core.registry import McpRegistry +from beaver_gateway.core.scheduler import Job, JobRun, Scheduler from beaver_gateway.core.transcript import build_entries from beaver_gateway.frontends.admin import AdminFrontend from beaver_gateway.frontends.admin.frontend import build_app as build_admin @@ -96,7 +97,12 @@ class ToolClient(ScriptedClient): class Api: - def __init__(self, world: World, memory_root: Path | None = None) -> None: + def __init__( + self, + world: World, + memory_root: Path | None = None, + scheduler: Scheduler | None = None, + ) -> None: self.world = world self.store = TokenStore(world.db, bootstrap={"t": TOKEN}) self.runtime = GatewayRuntime( @@ -112,6 +118,7 @@ class Api: conversations=world.conversations, bus=world.bus, pool=world.pool, + scheduler=scheduler, ) self.app = build_api(self.runtime, memory_root=memory_root) self.http = AsyncClient( @@ -146,6 +153,34 @@ async def seed_usage(world: World, rows: list[dict[str, Any]]) -> None: await session.commit() +async def test_job_runs_come_from_the_db(world: World) -> None: + async def ping(_run: JobRun) -> None: + pass + + scheduler = Scheduler( + conversations=world.conversations, jobs=[Job("ping", ping)], tz="Europe/Warsaw" + ) + api = Api(world, scheduler=scheduler) + await scheduler.start() + res = await api.http.post("/jobs/ping/run", headers=HEADERS) + assert res.status_code == 202, res.text + for _ in range(100): + runs = (await api.get("/jobs/ping/runs", {"limit": 10}))["runs"] + if runs: + break + await asyncio.sleep(0.02) + assert [(r["trigger"], r["status"], r["payload"]) for r in runs] == [ + ("manual", "done", {}) + ] + jobs = await api.get("/jobs") + assert jobs["tz"] == "Europe/Warsaw" + assert jobs["jobs"][0]["run"]["id"] == runs[0]["id"] + assert jobs["jobs"][0]["run"]["finished_at"].endswith("+02:00") + missing = await api.http.get("/jobs/nope/runs", headers=HEADERS) + assert missing.status_code == 404 + await scheduler.stop() + + async def test_usage_groups_by_agent_day_and_model(world: World) -> None: api = Api(world) now = datetime.now(UTC).replace(tzinfo=None) diff --git a/tests/test_conversations.py b/tests/test_conversations.py index 09d63ee..1ba506b 100644 --- a/tests/test_conversations.py +++ b/tests/test_conversations.py @@ -4,6 +4,7 @@ import uuid from datetime import UTC, datetime from pathlib import Path from typing import Any +from zoneinfo import ZoneInfo import pytest from claude_agent_sdk import ( @@ -629,6 +630,9 @@ async def test_schedule_without_scheduler_and_parse_at(world: World) -> None: assert parse_at("2026-09-01T10:00:00+02:00") == datetime( 2026, 9, 1, 8, 0, tzinfo=UTC ) + assert parse_at("2026-09-01T10:00", ZoneInfo("Europe/Warsaw")) == datetime( + 2026, 9, 1, 8, 0, tzinfo=UTC + ) delta = (parse_at("+15m") - datetime.now(UTC)).total_seconds() assert 14 * 60 < delta <= 15 * 60 with pytest.raises(ValueError, match="Invalid isoformat"): diff --git a/tests/test_scheduler.py b/tests/test_scheduler.py index f7a0a55..e4392dc 100644 --- a/tests/test_scheduler.py +++ b/tests/test_scheduler.py @@ -1,9 +1,11 @@ import asyncio +import inspect import os import tempfile from datetime import UTC, datetime, timedelta from pathlib import Path from typing import Any +from zoneinfo import ZoneInfo import psycopg import pytest @@ -11,14 +13,12 @@ from httpx import ASGITransport, AsyncClient from pgqueuer import PsycopgDriver, Queries from test_conversations import World -from beaver_gateway.core.scheduler import Budget, Job, JobRun, Scheduler +from beaver_gateway.core.conversations import parse_at +from beaver_gateway.core.scheduler import Budget, Job, JobRun, Scheduler, next_run from beaver_gateway.storage.models import RateLimit DATABASE_URL = os.environ.get("TEST_DATABASE_URL") - -pytestmark = pytest.mark.skipif( - not DATABASE_URL, reason="TEST_DATABASE_URL (postgres) is not set" -) +WARSAW = ZoneInfo("Europe/Warsaw") class Pg: @@ -37,6 +37,8 @@ class Pg: @pytest.fixture async def pg() -> Pg: + if not DATABASE_URL: + pytest.skip("TEST_DATABASE_URL (postgres) is not set") handle = Pg() queries = Queries(await handle.driver()) if await queries.has_table("pgqueuer"): @@ -61,6 +63,8 @@ async def until(pred, timeout: float = 5.0) -> Any: deadline = asyncio.get_running_loop().time() + timeout while asyncio.get_running_loop().time() < deadline: value = pred() + if inspect.isawaitable(value): + value = await value if value: return value await asyncio.sleep(0.05) @@ -68,16 +72,91 @@ async def until(pred, timeout: float = 5.0) -> Any: raise AssertionError(msg) +def test_cron_is_read_in_the_gateway_tz() -> None: + summer = datetime(2026, 7, 1, 12, tzinfo=UTC) + assert next_run("0 4 * * *", WARSAW, summer) == datetime(2026, 7, 2, 2, tzinfo=UTC) + assert next_run("0 4 * * *", UTC, summer) == datetime(2026, 7, 2, 4, tzinfo=UTC) + before_fallback = datetime(2026, 10, 24, 12, tzinfo=UTC) + assert next_run("0 4 * * *", WARSAW, before_fallback) == datetime( + 2026, 10, 25, 3, tzinfo=UTC + ) + during_fallback = datetime(2026, 10, 25, 1, 30, tzinfo=UTC) + assert next_run("0 4 * * *", WARSAW, during_fallback) == datetime( + 2026, 10, 25, 3, tzinfo=UTC + ) + assert next_run("*/15 * * * *", WARSAW, during_fallback) == datetime( + 2026, 10, 25, 1, 45, tzinfo=UTC + ) + soon = next_run("* * * * *", WARSAW) + assert soon.tzinfo is UTC + assert timedelta(0) < soon - datetime.now(UTC) <= timedelta(minutes=1) + + +def test_parse_at_reads_naive_iso_in_tz() -> None: + assert parse_at("2026-09-01T10:00", WARSAW) == datetime( + 2026, 9, 1, 8, 0, tzinfo=UTC + ) + assert parse_at("2026-12-01T10:00", WARSAW) == datetime( + 2026, 12, 1, 9, 0, tzinfo=UTC + ) + assert parse_at("2026-09-01T10:00:00+02:00", WARSAW) == datetime( + 2026, 9, 1, 8, 0, tzinfo=UTC + ) + assert parse_at("2026-09-01T10:00") == datetime(2026, 9, 1, 10, 0, tzinfo=UTC) + + +async def test_runs_are_recorded_without_postgres(world: World) -> None: + async def ok(run: JobRun) -> None: + run.payload["seen"] = True + + async def boom(_run: JobRun) -> None: + msg = "no network" + raise ConnectionError(msg) + + scheduler = Scheduler( + conversations=world.conversations, + jobs=[Job("ok", ok, webhook=True), Job("boom", boom)], + tz="Europe/Warsaw", + ) + await scheduler.start() + await scheduler.hook("ok", {"stack": "x"}) + await scheduler.trigger(scheduler.job("boom")) + await until(lambda: len(scheduler._runs) == 2) # noqa: SLF001 + runs = await scheduler.runs("boom") + assert [r["status"] for r in runs] == ["failed"] + assert runs[0]["error"].startswith("ConnectionError: no network\n") + assert runs[0]["trigger"] == "manual" + assert runs[0]["started_at"] <= runs[0]["finished_at"] + assert runs[0]["started_at"].endswith("+02:00") + ok_runs = await scheduler.runs("ok") + assert ok_runs[0]["payload"] == {"stack": "x", "seen": True} + assert ok_runs[0]["trigger"] == "webhook" + await scheduler.hook("ok", {"stack": "y"}) + await until(lambda: scheduler._runs["ok"]["payload"].get("stack") == "y") # noqa: SLF001 + assert len(await scheduler.runs("ok")) == 2 + assert len(await scheduler.runs("ok", limit=1)) == 1 + snapshot = await scheduler.snapshot() + assert snapshot["tz"] == "Europe/Warsaw" + runs_by_name = {j["name"]: j["run"] for j in snapshot["jobs"]} + assert runs_by_name["ok"]["payload"]["stack"] == "y" + assert runs_by_name["boom"]["status"] == "failed" + await scheduler.stop() + + async def test_schedule_survives_a_restart(world: World, pg: Pg) -> None: conv = await world.conversations.create(kind="master", agent="a", origin="test") - first = Scheduler(conversations=world.conversations, driver=await pg.driver()) + first = Scheduler( + conversations=world.conversations, driver=await pg.driver(), tz="Europe/Warsaw" + ) world.conversations.scheduler = first await first.start() job_id, when = await world.conversations.schedule(conv, "+1s", "push X") assert job_id is not None assert 0 < (when - datetime.now(UTC)).total_seconds() <= 1 + assert when.utcoffset() == datetime.now(WARSAW).utcoffset() queued = await world.conversations.schedules(conv) assert [q["payload"]["text"] for q in queued] == ["push X"] + assert queued[0]["payload"]["at"] == when.isoformat(timespec="minutes") await first.stop() second = Scheduler(conversations=world.conversations, driver=await pg.driver()) @@ -138,9 +217,14 @@ async def test_webhook_runs_the_job_with_its_payload(world: World, pg: Pg) -> No assert missing.status_code == 404 await until(lambda: seen) assert seen == [{"trigger": "webhook", "stack": "x"}] + await until(lambda: scheduler._runs.get("ping")) # noqa: SLF001 snapshot = await scheduler.snapshot() assert [j["name"] for j in snapshot["jobs"]] == ["ping", "quiet"] assert snapshot["enabled"] and snapshot["jobs"][0]["webhook"] + assert snapshot["jobs"][0]["run"]["status"] == "done" + assert snapshot["jobs"][0]["run"]["trigger"] == "webhook" + runs = await scheduler.runs("ping") + assert [r["payload"] for r in runs] == [{"stack": "x"}] await scheduler.stop() @@ -170,6 +254,7 @@ async def test_non_critical_jobs_wait_while_the_window_is_hot( ], driver=await pg.driver(), budget=Budget(threshold=0.7), + tz="Europe/Warsaw", ) await scheduler.start() events: list[dict[str, Any]] = [] @@ -193,5 +278,13 @@ async def test_non_critical_jobs_wait_while_the_window_is_hot( assert snapshot["throttled"] and snapshot["utilization"] == 0.9 crons = {j["name"]: j["cron"] for j in snapshot["jobs"]} assert crons == {"vibegram": "*/10 * * * *", "rotation": "0 * * * *"} - await until(lambda: all(j["next_run"] for j in snapshot["jobs"]) or True) + + async def crons_ready() -> dict[str, Any] | None: + snap = await scheduler.snapshot() + return snap if all(j["next_run"] for j in snap["jobs"]) else None + + snapshot = await until(crons_ready) + expected = datetime.now(WARSAW).utcoffset() + for job_info in snapshot["jobs"]: + assert datetime.fromisoformat(job_info["next_run"]).utcoffset() == expected await scheduler.stop() diff --git a/ui/src/lib/api/types.ts b/ui/src/lib/api/types.ts index 67d344e..77dc380 100644 --- a/ui/src/lib/api/types.ts +++ b/ui/src/lib/api/types.ts @@ -270,6 +270,17 @@ export interface QueuedJob { status: string; } +export interface JobRunInfo { + error: string | null; + finished_at: string | null; + id: number | null; + job: string; + payload: Record; + started_at: string; + status: string; + trigger: string; +} + export interface JobInfo { critical: boolean; cron: string | null; @@ -277,7 +288,7 @@ export interface JobInfo { last_run: string | null; name: string; next_run: string | null; - run: { trigger: string; started_at: string; status: string } | null; + run: JobRunInfo | null; status: string | null; webhook: boolean; }