Houmeres/LLeMbas
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 4 Wiki Activity

Something can happen because time passed, and land somewhere worth reading

Browse Source 
Nothing in LLeMbas ever happened on its own. Every reply was downstream of
somebody pressing Send, and the one exception -- jobs.wake, waking a chat when a
background job finishes -- was downstream of a command they had run. PLAN.md
never listed scheduling as unbuilt because services/chat.py:618 had recorded it
as a decision: "a scheduler is a whole new concern for a single-worker
application". This is that concern, taken on deliberately, plus the two places
its output goes.

Reports first, because it is useful with no scheduling at all. A report is not a
Chat with one Message in it: it has no turns and no reply, it is read top to
bottom, and it must be writable with no chat behind it -- being the fallback for
a run whose own chat has gone. As a Chat it would need a sidebar row per daily
report, a title that regenerates itself, a composer to suppress and a bubble with
a rewind button around something that is not a turn. The section's character is
enforced by absence: nothing under reports/ includes the composer or renders
chat/_message.html, so there is no sse-connect anywhere and nothing on those
pages *can* start a generation. The test reads that off the OpenAPI schema, not
by walking app.routes -- this FastAPI keeps an included router wrapped rather
than flattening it, so the walk finds nothing and the assertion passes for the
wrong reason.

rule.py is pure, total, and was finished before anything called it. No session,
no wall clock, nothing that raises: validate clamps what it recognises, drops
what it does not, and answers {} for prose -- at which point the caller shows the
manual form. It had to be that way because the compile step's output is model
output that becomes a *timer*, which is the sharpest case of hard rule 6 here.
The invariant, pinned: anything validate accepts has a computable next
occurrence. A schedule that can never fire looks exactly like a working one on
every screen it appears on.

Wall-clock and elapsed time are kept apart because they mean different things.
at.times are wall-clock in the owner's zone, so 15:00 stays 15:00 across a
daylight-saving change -- that is what "every Monday at 3PM" means. every is
elapsed real time, so six hours stays six hours across a 23- or 25-hour day --
that is what a timer means. Conflating them gets one of the two wrong twice a
year. A time inside the spring-forward gap fires at the first minute that exists;
left to zoneinfo's own resolution it lands an hour away wearing a wall-clock time
that did not happen, and a daily 02:30 report vanishing once a year on a machine
nobody watches is the failure this file is arranged around.

The ticker claims and commits *before* it fires. The other order is a hot loop: a
firing that raises is retried every tick for ever against whatever it was that
failed, and the only symptom is load. Its blanket except is copied from the
terminal reaper for a sharper reason -- a ticker that dies on one bad row stops
every schedule on the instance and says nothing at all. No request fails, no
reply errors, no dot appears. The reports simply stop.

Three rules that look like bugs from outside: a firing arriving while the chat is
still answering queues rather than starting a second reply, and past max_queued
is skipped with the reason on the row; Run now does not advance next_fire_at, or
testing a schedule silently consumes the run it was testing; resuming recomputes
from now, or a schedule paused for a month fires the instant it comes back, once
per occurrence it missed. Catching up lives in the sweep and not in a startup
hook, because a suspended host and a long stall reproduce "its time passed while
nothing was running" with no restart to hang one on.

services/wake.py is the lock discipline extracted rather than copied. A finished
job and a due schedule are the same problem, and both depend on there being no
await between the running_for check and the writes; two lock dictionaries for one
invariant is how one of them drifts. jobs.wake is now a caller that supplies
wording, and _completion_text stayed exactly where it was because tool.background
quotes its opening sentence.

A scheduled run has no reader, so ask_user is withdrawn from resolve_tools rather
than merely discouraged in core.unattended -- a rule living only in a system
message is one a page the model just read can argue with, and a parked question
holds the reply for the whole approval_timeout with nobody to answer it. For the
same reason a task chat may not be an agent chat in v1: Manual, Edit and Plan all
stop to ask on RISK_EXECUTE, so the only two outcomes would be unattended
execution and a reply that stalls. That deserves its own pass.

Messages is bounded in the request and unbounded on disk. Only the latest chunk
is sent; everything else stays exactly where it was written. Nothing is folded
into text and nothing is deleted -- the visible conversation is identical either
way, so destroying the older rows would buy only disk, against being irreversible
and losing every attachment and tool call in the range, and it would contradict
the rule compaction already holds. should_compact refuses this kind for the
matching reason: two mechanisms narrowing one transcript is how a summary ends up
summarising a summary. The history route is the mirror of thread_tail and keeps
its four properties; the fifth is its own, that prepending moves the scroll
position, so app.js records scrollHeight before the swap and adds the difference
back after.

An empty Chat.kind meant "both sides of the switch" and had been read as "no
filter" since there were only two of them. The sidebar passes "" precisely when
agent chats are switched off -- so the moment a third kind existed, every task
chat and every Messages conversation appeared in somebody's ordinary chat list,
on exactly the instances whose owners would never think to look. KINDS stays the
two-sided fork, because set_sidebar_kind validates against it and a third entry
there makes the tree filterable to a side with no button to leave it; ALL_KINDS
is what a row may be. Both narrowings are pinned, because they are two
implementations of one rule and only one of them is SQL.

Per-user timezone had to exist for any of this: harness.py:179 was telling every
reader the *server's* idea of the date, which is survivable while the answer is
prose and stops being survivable the moment somebody says "every Monday at 3" and
something has to work out when that is.

Three things were caught by a test being wrong rather than by the code being
wrong. The task-chat "no composer" assertions were passing against a page
rendering its no-models-configured branch. A permission test asserted the same
thing twice because the administrator bypasses every permission. And every
Messages test passed with default_model never called, because none of them
configured a model -- so the pair it returns was being assigned straight to
model_id, and SQLite refuses a tuple in a String column. The fixtures now say why
they exist.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Jaroslav Beneš
2026-08-05 21:31:36 +02:00
parent 178742501d
commit 9ddc0a2103
63 changed files with 6773 additions and 67 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/lembas/services/schedule/__init__.py
+17
View File
@@ -0,0 +1,17 @@
"""Scheduling: what should happen later, and what makes it happen.
Four modules, split by what each of them is allowed to touch:
- `clock.py` -- whose idea of "now" is in force. No session, no rows.
- `rule.py` -- the recurrence spec, and when it next comes due. Pure and
total: it never raises, never opens a session and never reads
the wall clock, which is what lets it be tested exhaustively
before anything calls it.
- `ticker.py` -- the loop that notices a schedule is due, and claims it.
- `runner.py` -- what actually happens when one fires.
The order matters and is the phasing: everything upstream of `ticker.py` can be
got wrong quietly, so it is settled first.
"""
from __future__ import annotations
src/lembas/services/schedule/clock.py
+111
View File
@@ -0,0 +1,111 @@
"""Whose idea of "now" is in force.
Until schedules existed, nothing here needed a timezone: `harness.py` stamped
`datetime.now().astimezone()` and every reader was told the *server's* idea of
the date. That is harmless when the answer is prose and wrong the moment a
person says "every Monday at 3" and something has to work out when that is.
One resolver, because the model compiling a schedule, the screen echoing it back
and the ticker firing it must agree about what Monday means. A disagreement here
does not raise -- it fires at the wrong time, which is the kind of wrong nobody
can debug from the outside.
Deliberately no new column. The zone lives in `user.settings_json["timezone"]`
beside the theme, empty meaning "whatever the server is set to" -- which is the
honest default for the single-user instance this mostly runs on, and is a real
answer rather than a prompt to go and choose one.
"""
from __future__ import annotations
import logging
from datetime import UTC, datetime, tzinfo
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError, available_timezones
from lembas.db.models import User
log = logging.getLogger(__name__)
SETTING_KEY = "timezone"
def server_zone() -> tzinfo:
"""What the machine is set to, as a real tzinfo.
`astimezone()` on a naive stamp attaches the system zone, which is what the
harness has always used. Read once per call rather than cached: a host whose
zone changes under a long-running process is rare, and a cache that gets it
wrong is worse than the lookup.
"""
return datetime.now().astimezone().tzinfo or UTC
def known(name: str) -> bool:
"""Whether this is a zone name Python can actually resolve.
`available_timezones()` reads the system database and is not cheap, so it is
only consulted for a value that is about to be stored. Everything on the
read path goes through `zone_for`, which simply falls back.
"""
return bool(name) and name in available_timezones()
def resolve(name: str) -> tzinfo:
"""A zone by name, falling back to the server's rather than raising.
A stored name can stop resolving -- the tz database is a system package and
a zone can be renamed out from under a row. Falling back means a schedule
fires an hour out at worst; raising means it does not fire at all and the
ticker logs an exception nobody reads.
"""
if not name:
return server_zone()
try:
return ZoneInfo(name)
except (ZoneInfoNotFoundError, ValueError, OSError):
log.warning("unknown timezone %r, falling back to the server's", name)
return server_zone()
def name_for(user: User | None) -> str:
"""The stored name, or "" meaning the server's. Never resolved here --
the settings form wants the raw value so an unset zone shows as unset."""
if user is None:
return ""
return str((user.settings_json or {}).get(SETTING_KEY) or "")
def zone_for(user: User | None) -> tzinfo:
"""The zone a schedule of this person's fires in, and the one the harness
should tell them the time in."""
return resolve(name_for(user))
def now_for(user: User | None) -> datetime:
"""Aware, in the reader's zone."""
return datetime.now(tz=zone_for(user))
def to_utc(moment: datetime, *, zone: tzinfo) -> datetime:
"""A wall-clock stamp in `zone`, as an instant.
Naive input is *interpreted* in `zone`; aware input is converted, so a
caller that already knows the offset cannot have it silently reassigned.
"""
if moment.tzinfo is None:
moment = moment.replace(tzinfo=zone)
return moment.astimezone(UTC)
def as_utc(moment: datetime) -> datetime:
"""An instant, whatever it arrived as.
SQLite does not store the offset, so a row read back from disk is naive
while one still in the session's identity map keeps its tzinfo, and
comparing the two raises -- the same trap `compaction.moment` exists for.
A naive stamp from the database is UTC by construction, because that is what
every column here is written with.
"""
if moment.tzinfo is None:
return moment.replace(tzinfo=UTC)
return moment.astimezone(UTC)
src/lembas/services/schedule/compile.py
+218
View File
@@ -0,0 +1,218 @@
"""Turning "remind me every Monday to check the build" into a schedule.
One request, once, when a schedule is created. It does two things a person
should not have to do by hand: work out the recurrence, and rewrite the
description into something that reads sensibly with **no conversation around
it** — because that is how it will be read, days later, by a model that was not
present when it was typed.
Three rules hold this up:
- **The rule goes through `rule.validate` and nothing else.** That function is
total and clamping, and this is the reason it had to be: what arrives here is
model output that becomes a *timer*. There is one normaliser, shared with the
manual form, so there cannot be two ideas of what a legal schedule is.
- **A compile that fails is not an error.** It hands back what it could work out
and the caller shows the manual form with the reader's own words in it. A
model that answers in prose must never quietly produce a schedule that never
fires.
- **Clearing `task.schedule_compile` switches off the compiling, not the
feature.** That is what makes "an empty override means off" safe here, and it
is only safe because the manual form exists. `task.compact` set the precedent
that clearing a fragment kills a feature, so this one says otherwise in its
own hint.
Deliberately no `response_format`. Several local endpoints reject unknown
parameters outright, and this is exactly the `apply_effort` lesson: a request
that 400s here would be the compile silently switching itself off.
"""
from __future__ import annotations
import json
import logging
import re
from dataclasses import dataclass, field
from datetime import UTC, datetime
from lembas.db.models import TARGET_CHAT, TARGETS, Chat, User
from lembas.services.llm.openai_client import Endpoint, LLMError, complete
from lembas.services.reasoning import strip_reasoning
from lembas.services.schedule import clock
from lembas.services.schedule import rule as rule_service
log = logging.getLogger(__name__)
# Enough for a small model that thinks before answering. The title lesson
# applies: too small is not a shorter answer, it is no answer, because the
# thinking consumes the budget and content comes back empty.
MAX_TOKENS = 900
MAX_REQUEST_CHARS = 2000
_FENCE = re.compile(r"```(?:json)?\s*(.*?)```", re.DOTALL)
@dataclass(frozen=True)
class Compiled:
"""What the compile worked out. `ok` is False when the reader must finish
the job by hand -- the fields are still filled in as far as they went."""
ok: bool = False
title: str = ""
instruction: str = ""
target: str = TARGET_CHAT
rule: dict = field(default_factory=dict)
reason: str = ""
def _payload(raw: str) -> dict:
"""The first JSON object in a reply, however it was wrapped.
Lenient for the reason `tools.parse_arguments` is: a small model sends
something close to the shape rather than the shape, and refusing it costs a
whole round trip to end up showing the manual form anyway.
"""
text = (raw or "").strip()
fenced = _FENCE.search(text)
if fenced:
text = fenced.group(1).strip()
if not text.startswith("{"):
start, end = text.find("{"), text.rfind("}")
if start == -1 or end <= start:
return {}
text = text[start : end + 1]
try:
parsed = json.loads(text)
except (ValueError, TypeError):
return {}
return parsed if isinstance(parsed, dict) else {}
def render_prompt(template: str, *, request: str, user: User | None) -> str:
"""Fill the fragment in. Separate so a test can read what was asked."""
from lembas.services import prompts as prompts_service
zone = clock.zone_for(user)
now = datetime.now(tz=UTC).astimezone(zone)
return prompts_service.substitute(
template,
{
"request": request[:MAX_REQUEST_CHARS],
"now": now.strftime("%A %-d %B %Y, %H:%M"),
"timezone": clock.name_for(user) or str(clock.server_zone()),
"targets": ", ".join(TARGETS),
},
)
async def compile_request(
endpoint: Endpoint,
model_id: str,
request: str,
*,
template: str,
user: User | None = None,
) -> Compiled:
"""Work a plain-language request into a schedule.
Never raises. Every failure -- a cleared fragment, an endpoint that is down,
prose instead of JSON, a rule that normalises to nothing -- comes back as
`ok=False` with whatever was salvageable, and the route shows the manual form.
"""
plain = (request or "").strip()
if not plain:
return Compiled(reason="Say what you want to happen.")
if not template.strip():
# An administrator cleared the fragment. That switches off the
# *compiling*: the reader fills the form in themselves, with their own
# words already in it.
return Compiled(instruction=plain, title=plain[:80], reason="")
prompt = render_prompt(template, request=plain, user=user)
body = {
"model": model_id,
"messages": [{"role": "user", "content": prompt}],
"max_tokens": MAX_TOKENS,
"temperature": 0.2,
}
try:
raw = await complete(endpoint, body)
except LLMError as exc:
log.info("schedule compile failed: %s", exc)
return Compiled(
instruction=plain,
title=plain[:80],
reason="The model could not be reached, so fill this in yourself.",
)
# A model that thinks inline puts its reasoning in `content`, which is the
# field `complete` hands back verbatim -- the same trap auto-titling hit.
answered, _ = strip_reasoning(raw)
payload = _payload(answered)
if not payload:
return Compiled(
instruction=plain,
title=plain[:80],
reason="The model did not answer with a schedule, so fill this in yourself.",
)
raw_rule = payload.get("schedule") or payload.get("rule") or {}
if isinstance(raw_rule, dict):
# A model asked for "every six hours" writes `{"every": {"hours": 6}}`
# and nothing else, which is the natural reading and cannot fire: a
# timer measures from a start, and `rule.py` has no clock to invent one.
# Filled in here, exactly as the manual form's `_rule_from_form` does,
# so the two paths agree about what a startless timer means. A model
# that puts `start` at the top level instead is read the same way rather
# than being told its schedule means nothing.
raw_rule = dict(raw_rule)
if raw_rule.get("every") and not raw_rule.get("start"):
raw_rule["start"] = payload.get("start") or datetime.now(tz=UTC).isoformat()
clean = rule_service.validate(raw_rule)
title = str(payload.get("title") or "").strip() or plain[:80]
instruction = str(payload.get("instruction") or "").strip() or plain
target = str(payload.get("target") or TARGET_CHAT)
if target not in TARGETS:
target = TARGET_CHAT
if not clean:
return Compiled(
title=title,
instruction=instruction,
target=target,
reason="The model could not work out when this should run — say when below.",
)
if rule_service.next_after(clean, datetime.now(tz=UTC), zone=clock.zone_for(user)) is None:
# Normalised, but with nothing left to fire. Refused for the same reason
# `schedules.create` refuses it: a schedule that can never run looks
# exactly like a working one on every screen it appears on.
return Compiled(
title=title,
instruction=instruction,
target=target,
rule=clean,
reason="That time has already passed — say when it should run.",
)
return Compiled(ok=True, title=title, instruction=instruction, target=target, rule=clean)
def endpoint_for(db, user: User) -> tuple[Endpoint, str] | None:
"""A connection and model to compile with, or None if there is none.
Built on a throwaway `Chat` that is never added to a session, exactly as
`agent/draft.py` does: `resolve_endpoint` reads `model_id` and
`connection_id` and nothing else, so it works unchanged and did not have to
learn what a compile is.
"""
from lembas.services import chat as chat_service
models = chat_service.available_models(db, user)
if not models:
return None
chosen = next((m for m in models if m.pinned), models[0])
stand_in = Chat(user_id=user.id, model_id=chosen.model_id, connection_id=chosen.connection_id)
try:
return chat_service.resolve_endpoint(db, stand_in)
except LLMError:
return None
src/lembas/services/schedule/rule.py
+506
View File
@@ -0,0 +1,506 @@
"""When a schedule next comes due.
Pure and total. Nothing here opens a session, reads the wall clock or raises:
every function takes what it needs and answers, so the whole of this module can
be tested exhaustively before anything calls it. That is deliberate, because
everything downstream fails *quietly* -- a schedule that never fires looks
exactly like a working one on the list page, and a schedule that fires an hour
out looks like nothing at all until somebody notices the report is late.
## The shape
Plain cron cannot say "ten minutes from now, five times", so the rule is a dict
with two independent generators and a bound:
{
"start": "2026-08-05T14:30:00Z", # first candidate instant, UTC
"every": {"minutes": 10}, # a stride
"at": {"weekdays": [0], # 0 = Monday
"days": [1, 15], # day of the month
"months": [1, 4, 7, 10],
"times": ["15:00"]}, # wall-clock, in the owner's zone
"count": 5, # total firings, 0 = unbounded
"until": "2026-12-31T00:00:00Z" # last instant, "" = unbounded
}
`every` and `at` compose, and the four combinations are the whole vocabulary:
every at meaning
----- ---- ------------------------------------------------------------
- - fire once, at `start`
x - a timer: start, start + every, start + 2*every, ...
- x a calendar: every matching wall-clock moment after `start`
x x a calendar with a stride: matching moments, every Nth kept
## Timezone, and why the two halves differ
`at.times` are **wall-clock** in the owner's zone: 15:00 stays 15:00 across a
DST change, because that is what "every Monday at 3PM" means to the person who
said it. `every` durations are **elapsed real time**: ten minutes is ten
minutes, and a six-hourly timer must not skip or double on a 23- or 25-hour day.
Those are different meanings, not an inconsistency, and conflating them is how
one of the two comes out wrong twice a year.
A wall-clock time that does not exist (the hour skipped on a spring-forward day)
fires at the first instant that does, rather than being skipped -- a daily report
vanishing once a year is precisely the silent failure this file exists to avoid.
One that occurs twice on a fall-back day fires on the first, once.
## Not expressible
Said plainly, because the gap is the point: "the last Friday of the month", "the
third Monday", "weekdays except holidays", "the Nth business day", sub-minute
intervals, sunrise-relative times, and any conditional firing ("only if the
build is red"). The first two are what people will actually ask for; the rule is
JSON, so an `nth` key inside `at` adds them later with no migration.
"""
from __future__ import annotations
import logging
from datetime import UTC, datetime, timedelta, tzinfo
log = logging.getLogger(__name__)
# The stride units, and how many seconds each is worth. Months are absent on
# purpose: a month is not a duration, and "every month" is `at: {days: [n]}`,
# which is what somebody means by it.
UNITS: dict[str, int] = {
"minutes": 60,
"hours": 3600,
"days": 86400,
"weeks": 604800,
}
# Bounds. Every one of these is a clamp rather than a rejection, because the
# rule can arrive from a *model* -- the compile step's output is model output
# that becomes a timer, and `validate` is this feature's `nh3.clean`.
MIN_INTERVAL_SECONDS = 60
MAX_INTERVAL_SECONDS = 366 * 86400
MAX_COUNT = 10_000
MAX_TIMES = 24
MAX_HORIZON_DAYS = 366 * 5
# How far ahead a calendar search will walk before giving up. A rule asking for
# 31 February matches nothing, and a search with no bound would spin for ever
# inside the ticker. Days rather than iterations, so the limit is a statement
# about the schedule rather than about the loop.
SEARCH_DAYS = 366 * 4
WEEKDAYS = (0, 1, 2, 3, 4, 5, 6)
# --- Reading a rule ------------------------------------------------------------
def _int(value: object, *, low: int, high: int, default: int = 0) -> int:
try:
number = int(value) # type: ignore[arg-type]
except (TypeError, ValueError):
return default
return max(low, min(number, high))
def _stamp(value: object) -> datetime | None:
"""An ISO instant, or None. Naive input is read as UTC.
`fromisoformat` handles a trailing Z from Python 3.11, but a model writes
all sorts of things, so anything unparseable is simply absent.
"""
if isinstance(value, datetime):
return value if value.tzinfo else value.replace(tzinfo=UTC)
if not isinstance(value, str) or not value.strip():
return None
try:
parsed = datetime.fromisoformat(value.strip().replace("Z", "+00:00"))
except ValueError:
return None
return parsed if parsed.tzinfo else parsed.replace(tzinfo=UTC)
def _times(value: object) -> list[tuple[int, int]]:
"""Wall-clock times as (hour, minute), sorted and deduplicated.
Accepts "15:00", "15:00:30" and "9:5", because a model writes all three, and
a rule refused for its punctuation is a round trip spent on nothing.
"""
if isinstance(value, str):
value = [value]
if not isinstance(value, (list, tuple)):
return []
found: set[tuple[int, int]] = set()
for item in list(value)[:MAX_TIMES]:
if not isinstance(item, str) or ":" not in item:
continue
hour, _, rest = item.strip().partition(":")
minute = rest.partition(":")[0]
try:
pair = (int(hour), int(minute))
except ValueError:
continue
if 0 <= pair[0] <= 23 and 0 <= pair[1] <= 59:
found.add(pair)
return sorted(found)
def _numbers(value: object, *, low: int, high: int) -> list[int]:
if isinstance(value, int) and not isinstance(value, bool):
value = [value]
if not isinstance(value, (list, tuple)):
return []
found: set[int] = set()
for item in value:
if isinstance(item, bool):
continue
try:
number = int(item) # type: ignore[arg-type]
except (TypeError, ValueError):
continue
if low <= number <= high:
found.add(number)
return sorted(found)
def _every(value: object) -> dict[str, int]:
"""A stride, clamped to something that can actually be run.
An interval under a minute is refused rather than clamped to a minute: the
ticker's own granularity is coarser than that, so honouring it is impossible
and pretending to would be a schedule that silently runs late for ever.
Clamped up, because "every 10 seconds" from a model means "often", and often
is a minute.
"""
if not isinstance(value, dict):
return {}
seconds = 0
for unit, size in UNITS.items():
seconds += _int(value.get(unit), low=0, high=MAX_INTERVAL_SECONDS) * size
if seconds <= 0:
return {}
seconds = max(MIN_INTERVAL_SECONDS, min(seconds, MAX_INTERVAL_SECONDS))
return {"minutes": seconds // 60}
def validate(rule: object) -> dict:
"""Normalise a rule, or return {} for one that cannot be made sense of.
**Total on purpose.** The compile step hands this whatever a model wrote, so
it drops what it does not recognise and clamps what it does, and never
raises. `{}` is the honest answer for prose, for a cron string, for an empty
object -- and the caller's job is then to show the manual form rather than
write a schedule that never fires. A schedule that can never fire is
indistinguishable from a working one on every screen it appears on, which is
this feature's flagship silent failure.
The invariant worth holding on to, and pinned in the tests: **anything this
returns non-empty has a computable next occurrence.**
"""
if not isinstance(rule, dict):
return {}
every = _every(rule.get("every"))
raw_at = rule.get("at") if isinstance(rule.get("at"), dict) else {}
at = {
"weekdays": _numbers(raw_at.get("weekdays"), low=0, high=6),
"days": _numbers(raw_at.get("days"), low=1, high=31),
"months": _numbers(raw_at.get("months"), low=1, high=12),
"times": [f"{hour:02d}:{minute:02d}" for hour, minute in _times(raw_at.get("times"))],
}
# A calendar with no time of day has no time of day. Midnight is the only
# defensible reading and it is what every cron-like thing does, so it is
# filled in rather than making the whole `at` block meaningless.
if any(at[key] for key in ("weekdays", "days", "months")) and not at["times"]:
at["times"] = ["00:00"]
if not at["times"]:
at = {}
start = _stamp(rule.get("start"))
until = _stamp(rule.get("until"))
count = _int(rule.get("count"), low=0, high=MAX_COUNT)
# A one-shot is `start` and nothing else, so without a start there is
# nothing to fire and nothing to infer -- unlike a calendar, which is
# perfectly meaningful from now onwards.
if not every and not at and start is None:
return {}
# A window that closes before it opens produces nothing, which is a rule
# that cannot fire rather than one that fires oddly.
if start is not None and until is not None and until < start:
return {}
normalised: dict = {}
if start is not None:
normalised["start"] = start.astimezone(UTC).isoformat()
if every:
normalised["every"] = every
if at:
normalised["at"] = {key: value for key, value in at.items() if value}
normalised["at"]["times"] = at["times"]
if count:
normalised["count"] = count
if until is not None:
normalised["until"] = until.astimezone(UTC).isoformat()
return normalised
# --- When it next comes due -----------------------------------------------------
def _interval(rule: dict) -> timedelta:
return timedelta(minutes=int((rule.get("every") or {}).get("minutes") or 0))
def _matches(moment: datetime, at: dict) -> bool:
"""Whether a local date satisfies the calendar constraints.
Empty means "every", per field, which is what makes `{"times": ["09:00"]}`
read as "daily at nine" without having to enumerate seven weekdays.
"""
weekdays = at.get("weekdays") or []
days = at.get("days") or []
months = at.get("months") or []
if weekdays and moment.weekday() not in weekdays:
return False
if days and moment.day not in days:
return False
return not (months and moment.month not in months)
def _wall(day: datetime, hour: int, minute: int, zone: tzinfo) -> datetime:
"""A wall-clock time on a given local day, as an instant.
Two DST cases, both handled here rather than left to `zoneinfo`'s defaults:
- **The hour that does not exist.** On a spring-forward day, 02:30 is not a
time. Constructing it anyway yields something that does not round-trip, so
the gap is detected by comparing and the result is pushed to the first
instant that does exist. Skipping the day instead is how a daily report
disappears once a year.
- **The hour that happens twice.** `fold=0` picks the first, and the
advance-past-the-last-fire rule upstream is what stops the second being
taken as a separate occurrence.
"""
naive = day.replace(hour=hour, minute=minute, second=0, microsecond=0, tzinfo=None)
local = naive.replace(tzinfo=zone, fold=0)
# A time inside the spring-forward gap does not survive the round trip.
if local.astimezone(UTC).astimezone(zone).replace(tzinfo=None) != naive:
# Walk forward a minute at a time to the far side of the gap. Gaps are
# an hour at most in every zone the database has ever carried, so this
# is bounded and cheap; adding the offset difference directly would
# assume the size of a gap this code has no business knowing.
for extra in range(1, 181):
candidate = (naive + timedelta(minutes=extra)).replace(tzinfo=zone, fold=0)
round_trip = candidate.astimezone(UTC).astimezone(zone).replace(tzinfo=None)
if round_trip == naive + timedelta(minutes=extra):
return candidate.astimezone(UTC)
return local.astimezone(UTC)
def _calendar_after(rule: dict, after: datetime, *, zone: tzinfo) -> datetime | None:
"""The first calendar occurrence strictly after `after`."""
at = rule.get("at") or {}
times = [tuple(int(part) for part in value.split(":")) for value in at.get("times") or []]
if not times:
return None
local = after.astimezone(zone)
day = local.replace(hour=0, minute=0, second=0, microsecond=0)
for _ in range(SEARCH_DAYS):
if _matches(day, at):
for hour, minute in times:
moment = _wall(day, hour, minute, zone)
if moment > after:
return moment
day += timedelta(days=1)
# Re-anchor to local midnight: adding a day across a DST boundary
# otherwise leaves the cursor an hour either side of it, and the day
# after a fall-back would be searched from 23:00 the previous evening.
day = day.astimezone(zone).replace(hour=0, minute=0, second=0, microsecond=0)
return None
def _exhausted(rule: dict, moment: datetime, fired: int) -> bool:
count = int(rule.get("count") or 0)
if count and fired >= count:
return True
until = _stamp(rule.get("until"))
return bool(until and moment > until)
def next_after(
rule: dict, after: datetime, *, zone: tzinfo, fired: int = 0
) -> datetime | None:
"""The next instant this rule comes due, strictly after `after`.
`None` means never again: the count is spent, the window has closed, or the
calendar matches nothing inside the search horizon. A caller seeing `None`
disables the schedule -- exhaustion switches off, it does not loop.
`fired` is how many times it has already run, and is what makes `count`
work without the rule having to carry mutable state.
"""
if not isinstance(rule, dict) or not rule:
return None
count = int(rule.get("count") or 0)
if count and fired >= count:
return None
after = after.astimezone(UTC)
start = _stamp(rule.get("start"))
every = _interval(rule)
at = rule.get("at") or {}
moment: datetime | None
if at:
# A calendar never fires before its start, so the search begins at
# whichever of the two is later.
floor = max(after, start - timedelta(microseconds=1)) if start else after
moment = _calendar_after(rule, floor, zone=zone)
if moment is not None and every:
# A stride over a calendar keeps every Nth match. Counted from the
# start rather than from `after`, so "every other Monday" means the
# same two Mondays whenever it is asked.
stride = max(1, int(round(every.total_seconds() / 86400)) or 1)
if stride > 1 and start is not None:
elapsed = (moment.astimezone(zone).date() - start.astimezone(zone).date()).days
skipped = 0
while elapsed % stride and skipped < SEARCH_DAYS:
moment = _calendar_after(rule, moment, zone=zone)
if moment is None:
break
elapsed = (
moment.astimezone(zone).date() - start.astimezone(zone).date()
).days
skipped += 1
elif every:
if start is None:
return None
if after < start:
moment = start
else:
# Absolute arithmetic, deliberately: a timer measures elapsed time,
# so it must not shift when the offset does. Computed rather than
# stepped, so a schedule idle for a year costs one division.
elapsed = (after - start).total_seconds()
steps = int(elapsed // every.total_seconds()) + 1
moment = start + every * steps
else:
# A one-shot. Due exactly once, and only if it has not already run --
# `fired` is what stops it being re-offered for ever once its moment has
# passed, since `start > after` is false from then on.
if start is None or fired:
return None
moment = start if start > after else None
if moment is None or _exhausted(rule, moment, fired):
return None
return moment
def advance(
rule: dict, *, after: datetime, now: datetime, zone: tzinfo, fired: int = 0
) -> tuple[bool, datetime | None]:
"""Catch up on a schedule whose time passed while nothing was running.
Answers two things at once: whether it is owed a firing *now*, and when it
should next come due. The pair is one function because the second depends on
the first -- a caller that asked separately would have to decide what
"next" means for a schedule it has just decided to fire.
**A missed run collapses to one.** The next occurrence returned is the first
one strictly after `now`, not the one after the slot that was missed -- so a
host switched off for a week comes back owing one report rather than a
hundred and sixty-eight. That is the whole reason this is not just
`next_after`.
It is called from the *sweep* rather than only at startup, because a
suspended laptop, a paused container and a long stall all reproduce the
same situation with no restart to hang a startup hook on.
"""
due = next_after(rule, after, zone=zone, fired=fired)
if due is None:
return False, None
if due > now:
return False, due
# Overdue. Fire once, and resume from wherever the rule is now -- counting
# this firing, so `count` is spent by what actually ran.
return True, next_after(rule, now, zone=zone, fired=fired + 1)
# --- Saying it back -------------------------------------------------------------
_DAY_NAMES = ("Monday", "Tuesday", "Wednesday", "Thursday", "Friday", "Saturday", "Sunday")
_MONTH_NAMES = (
"January", "February", "March", "April", "May", "June",
"July", "August", "September", "October", "November", "December",
)
def _join(words: list[str]) -> str:
if len(words) <= 1:
return "".join(words)
return f"{', '.join(words[:-1])} and {words[-1]}"
def _ordinal(number: int) -> str:
if 10 <= number % 100 <= 20:
return f"{number}th"
return f"{number}{ {1: 'st', 2: 'nd', 3: 'rd'}.get(number % 10, 'th') }"
def _duration(delta: timedelta) -> str:
minutes = int(delta.total_seconds() // 60)
for size, unit in ((10080, "week"), (1440, "day"), (60, "hour"), (1, "minute")):
if minutes >= size and not minutes % size:
amount = minutes // size
return f"{amount} {unit}{'s' if amount != 1 else ''}"
return f"{minutes} minute{'s' if minutes != 1 else ''}"
def describe(rule: dict, *, zone: tzinfo) -> str:
"""One line saying what this rule does, in the reader's own zone.
Not decoration. It is what the setup screen echoes back before anything is
saved, what the list page shows beside each schedule, and what the harness
tells a model about its own chat. A row reading "Every Monday at 3PM" over a
rule that fires daily is the same class of failure as three places
disagreeing about a tool's name -- and this one is the reader's only view of
a decision that happens while they are not looking.
"""
rule = rule or {}
if not rule:
return "Never"
at = rule.get("at") or {}
every = _interval(rule)
parts: list[str] = []
if at:
times = _join(list(at.get("times") or []))
when = []
if at.get("weekdays"):
when.append(_join([_DAY_NAMES[day] for day in at["weekdays"]]))
if at.get("days"):
when.append(f"the {_join([_ordinal(day) for day in at['days']])}")
if at.get("months"):
when.append(f"of {_join([_MONTH_NAMES[month - 1] for month in at['months']])}")
parts.append(
f"Every {' '.join(when)} at {times}" if when else f"Every day at {times}"
)
# A stride over a calendar is a qualifier rather than a rewording:
# "Every Monday at 15:00, skipping to every 14 days" is clumsy but true,
# and inventing "every other Monday" for it would be a phrase that stops
# being true the moment the stride is not two.
stride_days = int(every.total_seconds() // 86400) if every else 0
if stride_days > 1:
parts.append(f"but only every {stride_days} days")
elif every:
parts.append(f"Every {_duration(every)}")
else:
start = _stamp(rule.get("start"))
local = start.astimezone(zone) if start else None
return f"Once, on {local.strftime('%-d %B %Y at %H:%M')}" if local else "Once"
count = int(rule.get("count") or 0)
if count:
parts.append(f"{count} time{'s' if count != 1 else ''}")
until = _stamp(rule.get("until"))
if until:
parts.append(f"until {until.astimezone(zone).strftime('%-d %B %Y')}")
return ", ".join(parts)
src/lembas/services/schedule/runner.py
+301
View File
@@ -0,0 +1,301 @@
"""What happens when a schedule fires.
Every schedule fires the same way — a turn into a chat, answered by the ordinary
generation loop — and the *target* decides only what becomes of the finished
reply. One mechanism, three deliveries:
- `chat` leave it there. The reply is the point, and it is already in the
task chat where somebody will read it.
- `report` copy it into a `Report` and keep the chat out of the way.
- `messages` copy it into the reader's Messages conversation, as an assistant
turn marked `machine`. Copied rather than moved: the task chat is
the working area and keeps the tool calls, the steps and the
metrics; Messages gets the answer.
The alternative — a one-shot `complete()` in the shape of `generate_title` — was
rejected because it has no tools and no rounds, which is useless for the case
this feature exists for. "Give me a daily news report" needs to search the web.
**Nothing in `services/generation.py` changes.** The waiting happens here, in a
task per firing, which is the shape `jobs._watch` already established. Making
generation aware of schedules would mean a branch inside `_persist`, and that is
the single writer with one rule.
"""
from __future__ import annotations
import asyncio
import contextlib
import logging
from datetime import UTC, datetime
from lembas.db.models import (
ROLE_ASSISTANT,
TARGET_CHAT,
TARGET_MESSAGES,
TARGET_REPORT,
Chat,
Message,
Schedule,
User,
)
from lembas.db.session import session_scope
from lembas.services import reports as reports_service
from lembas.services import wake as wake_service
from lembas.services.schedule import clock
from lembas.services.schedule import rule as rule_service
log = logging.getLogger(__name__)
# How long to wait for a firing's reply before giving up on delivering it. The
# reply itself is not cancelled -- it goes on and lands in its chat, which is
# where a task chat's output belongs anyway. What times out is only *this*
# task's interest in copying the result somewhere.
DELIVERY_TIMEOUT = 3600.0
# How often the waiter looks. Coarse on purpose: nothing is watching this, and a
# report arriving three seconds late costs nobody anything.
POLL_SECONDS = 3.0
def _preamble(schedule: Schedule, *, zone, due_at: datetime | None) -> str:
"""The turn a firing puts into the chat.
Names itself a scheduled event in *words*, because the role stays `user` --
`_inject` sends a queued turn verbatim and `build_messages` must keep seeing
a user turn. The framing therefore cannot live in the role, exactly as it
cannot for a finished background job.
The scheduled time is stated as well as the actual one, so a run caught up
after an outage can say so rather than reporting stale news as current.
"""
now = datetime.now(tz=UTC).astimezone(zone)
lines = [
"This turn was started by a schedule, not by the person — "
"they are not necessarily at the keyboard.",
"",
f"[schedule: {schedule.title or 'untitled'}] "
f"{rule_service.describe(schedule.rule_json or {}, zone=zone)}",
f"It is now {now.strftime('%A %-d %B %Y, %H:%M')}.",
]
if due_at is not None:
late = (datetime.now(tz=UTC) - clock.as_utc(due_at)).total_seconds()
if late > 600:
local = clock.as_utc(due_at).astimezone(zone)
lines.append(
f"This run was due at {local.strftime('%A %-d %B, %H:%M')} and is late — "
"say so if it makes any of what follows out of date."
)
lines += ["", schedule.instruction or schedule.request or ""]
return "\n".join(lines)
async def _await_reply(chat_id: str, message_id: str) -> None:
"""Wait for one generation to finish.
Polled rather than awaited on the task itself: `generation` owns its
registry and its tasks, and reaching into either from here would couple this
to internals whose whole job is to be replaceable. A poll costs nothing at
this interval and cannot deadlock.
"""
from lembas.services import generation as generation_service
waited = 0.0
while waited < DELIVERY_TIMEOUT:
running = generation_service.running_for(chat_id)
# `running_for` already excludes a finished generation, so `None` is the
# ordinary end of this loop. The id check is what stops us waiting on
# somebody's *next* reply in the same chat, which would otherwise happen
# whenever a queued turn is drained straight after ours.
if running is None or running.message_id != message_id:
return
await asyncio.sleep(POLL_SECONDS)
waited += POLL_SECONDS
log.warning("gave up waiting for the reply to schedule message %s", message_id)
def _finished_reply(db, chat_id: str, message_id: str) -> Message | None:
message = db.get(Message, message_id)
if message is None or message.chat_id != chat_id:
return None
if not message.complete or message.error:
return None
return message
async def deliver(schedule_id: str, message_id: str, *, since: datetime) -> None:
"""Put a finished reply where the schedule said it should go.
`since` is the moment the firing began, and it is what tells a report the
model filed itself apart from one filed on a previous run.
"""
with session_scope() as db:
schedule = db.get(Schedule, schedule_id)
if schedule is None:
return
target = schedule.target
chat_id = schedule.chat_id
if target == TARGET_CHAT:
# Already where it belongs. Stated rather than left to fall through, so
# a reader of this function does not have to infer the common case.
return
await _await_reply(chat_id, message_id)
with session_scope() as db:
schedule = db.get(Schedule, schedule_id)
if schedule is None:
return
owner = db.get(User, schedule.user_id)
if owner is None:
return
message = _finished_reply(db, chat_id, message_id)
if target == TARGET_REPORT:
# If the model filed one itself with `report_write`, that is the
# report and this must not file a second. The tool stamps
# `source_id` with the chat, which is what makes them the same run;
# `since` is what makes it *this* run. Both stamps go through
# `as_utc` because one comes from a row read back from SQLite (which
# loses the offset) and the other is still in memory -- comparing
# the two raises, the trap `compaction.moment` exists for.
already = reports_service.recent(db, owner, limit=5)
if any(
r.source_id == chat_id and clock.as_utc(r.created_at) >= clock.as_utc(since)
for r in already
):
return
if message is None:
reports_service.create(
db,
owner=owner,
title=schedule.title or "Scheduled run",
body="",
source="schedule",
source_id=chat_id,
schedule_id=schedule.id,
error="The run did not produce a reply.",
)
return
reports_service.create(
db,
owner=owner,
title=schedule.title or "Scheduled run",
body=message.content or "",
source="schedule",
source_id=chat_id,
schedule_id=schedule.id,
model_id=message.model_id or "",
)
return
if target == TARGET_MESSAGES:
if message is None:
schedule.last_error = "The run did not produce anything to post."
db.commit()
return
# Copied in as an assistant turn rather than moved, because the task
# chat is the working area and holds the tool calls, the steps and
# the metrics -- the Messages conversation gets the answer. Marked
# `machine` for the same reason a job completion is: the reader did
# not write it, and the bubble should not imply they did.
from lembas.services import chat as chat_service
from lembas.services import messages as messages_service
conversation = messages_service.for_user(db, owner)
chat_service.create_message(
db,
conversation,
ROLE_ASSISTANT,
message.content or "",
model_id=message.model_id or "",
machine=True,
)
conversation.unread = True
conversation.unread_notified = False
db.commit()
async def fire(schedule_id: str, *, due_at: datetime | None = None) -> None:
"""Run one schedule now.
Never raises: the ticker calls this and one bad schedule must not stop the
others. Anything that goes wrong is written to `last_error`, where the
schedule's own page shows it — a run that failed silently is
indistinguishable from one that was never due.
"""
from lembas.services import settings_store
try:
with session_scope() as db:
schedule = db.get(Schedule, schedule_id)
if schedule is None:
return
owner = db.get(User, schedule.user_id)
chat = db.get(Chat, schedule.chat_id) if schedule.chat_id else None
if owner is None:
return
if chat is None or chat.user_id != owner.id:
# The chat was deleted, or never belonged to this owner. Stop
# rather than fire into nothing on every tick from now on.
schedule.enabled = False
schedule.last_error = "Its chat no longer exists, so it has been switched off."
db.commit()
return
limit = int(settings_store.schedules(db).get("max_queued") or 3)
zone = clock.zone_for(owner)
content = _preamble(schedule, zone=zone, due_at=due_at)
chat_id = chat.id
model_id = schedule.model_id or chat.model_id
began = datetime.now(tz=UTC)
schedule.claimed_at = began
schedule.last_error = ""
db.commit()
# Outside the session: a chat already carrying a backlog is one whose
# replies are slower than its schedule, and adding to it makes that
# permanently worse. `_drain` takes one queued turn per reply.
if wake_service.queued_count(chat_id) >= limit:
with session_scope() as db:
schedule = db.get(Schedule, schedule_id)
if schedule is not None:
schedule.last_error = (
"Skipped: the previous run was still going, and turns are "
"already waiting in its chat."
)
schedule.claimed_at = None
db.commit()
return
message_id = await wake_service.wake_chat(chat_id, content, model_id=model_id)
with session_scope() as db:
schedule = db.get(Schedule, schedule_id)
if schedule is not None:
schedule.claimed_at = None
db.commit()
if message_id:
await deliver(schedule_id, message_id, since=began)
except asyncio.CancelledError:
raise
except Exception: # noqa: BLE001 - one bad schedule must not stop the rest
log.exception("schedule %s failed to fire", schedule_id)
with contextlib.suppress(Exception), session_scope() as db:
schedule = db.get(Schedule, schedule_id)
if schedule is not None:
schedule.last_error = "Something went wrong running this. See the log."
schedule.claimed_at = None
db.commit()
async def run_now(schedule_id: str) -> None:
"""Fire a schedule because somebody pressed the button.
**Deliberately does not advance `next_fire_at`.** Testing a schedule must
not consume the run it was testing -- somebody who presses this at 14:00 to
check a 15:00 report still expects the 15:00 one. The ticker owns advancing,
and it is the only thing that does.
"""
await fire(schedule_id)
src/lembas/services/schedule/ticker.py
+210
View File
@@ -0,0 +1,210 @@
"""The loop that notices a schedule is due, and claims it.
Modelled on `agent/terminal.py:_reaper_loop`, which is the only periodic task
this codebase had before now — including the blanket `except` around the sweep,
for a reason that is sharper here: **a ticker that dies on one bad row stops
every schedule on the instance, and says nothing.** Nothing else would notice.
There is no request failing, no reply erroring, no dot appearing. The reports
simply stop, and the first person to find out is whoever eventually wonders why.
Started from the lifespan rather than lazily like the reaper. Lazy is right for
terminals — a shell only exists once somebody opened one — and wrong here: a
schedule can be due at startup with nobody logged in, which is most of the point.
## Claiming, and why the order is the whole design
One worker and one loop, so the risk is not two processes racing; it is two
*overlapping sweeps*, and a firing that raises being retried every tick for ever.
Three things answer that:
1. A lock around the sweep, so a slow one (a firing awaits a model, which can
take minutes) cannot overlap the next tick.
2. **Advance, then fire.** The row is moved on and committed *before* anything
is awaited. A firing that dies has still consumed its slot, so the schedule
resumes at its next occurrence with the reason on the row — rather than
becoming a hot loop against an endpoint that is down.
3. `claimed_at` outliving a firing is what lets a run that never finished say so
instead of looking like one that never started.
Exhaustion **disables**: a rule with nothing left returns `None`, and the row is
switched off rather than being re-examined for ever.
"""
from __future__ import annotations
import asyncio
import contextlib
import logging
from datetime import UTC, datetime, timedelta
from sqlalchemy import select
from lembas.db.models import Schedule, User
from lembas.db.session import session_scope
from lembas.services import settings_store
from lembas.services.schedule import clock, runner
from lembas.services.schedule import rule as rule_service
log = logging.getLogger(__name__)
_TICKER: asyncio.Task | None = None
_SWEEPING = asyncio.Lock()
# Live firings, so shutdown can wait for them rather than leaving a half-written
# reply and a `claimed_at` that never clears.
_FIRING: set[asyncio.Task] = set()
# Fallback when nothing has been configured. `settings_store.schedules` clamps
# the stored value; this is only for a sweep that runs before anything is read.
TICK_SECONDS = 30.0
def _due(now: datetime):
return (
select(Schedule)
.where(
Schedule.enabled.is_(True),
Schedule.next_fire_at.is_not(None),
Schedule.next_fire_at <= now,
)
.order_by(Schedule.next_fire_at)
)
def claim(schedule: Schedule, *, now: datetime, zone) -> tuple[bool, datetime | None]:
"""Move one schedule on, and say whether it is owed a firing.
Pure bookkeeping on the row: it does not fire anything and does not commit,
so the caller decides the transaction boundary. The caller must commit
before awaiting.
"""
rule = schedule.rule_json or {}
after = clock.as_utc(schedule.next_fire_at) if schedule.next_fire_at else now
fire_now, following = rule_service.advance(
rule,
# A microsecond earlier, because `next_after` answers *strictly* after
# what it is given -- so handing it the stored due moment would return
# the one following and skip the firing that is actually owed. The
# alternative, making `next_after` inclusive, would break the far more
# common "give me the one after this one" call it exists for.
after=after - timedelta(microseconds=1),
now=now,
zone=zone,
fired=schedule.fired_count or 0,
)
if fire_now:
schedule.fired_count = (schedule.fired_count or 0) + 1
schedule.last_fire_at = now
schedule.next_fire_at = following
if following is None:
# Nothing left to do: a spent count, a closed window, a calendar that
# matches nothing inside the horizon. Switched off rather than left
# enabled with a null next time, which would read as "waiting" for ever.
schedule.enabled = False
return fire_now, following
async def sweep(*, now: datetime | None = None) -> int:
"""One pass. Returns how many schedules were fired.
Claims every due row and commits, then starts the firings — in that order,
and with the commit in between, which is the property `test_schedule_ticker`
checks by making a firing raise.
"""
now = now or datetime.now(tz=UTC)
to_fire: list[tuple[str, datetime]] = []
async with _SWEEPING:
with session_scope() as db:
if not settings_store.schedules(db).get("enabled"):
return 0
limit = int(settings_store.schedules(db).get("max_concurrent") or 3)
for schedule in db.scalars(_due(now)):
try:
owner = db.get(User, schedule.user_id)
if owner is None:
# The account is gone; the CASCADE will take the row.
schedule.enabled = False
continue
due_at = clock.as_utc(schedule.next_fire_at) if schedule.next_fire_at else now
fire_now, _ = claim(schedule, now=now, zone=clock.zone_for(owner))
if fire_now:
to_fire.append((schedule.id, due_at))
except Exception: # noqa: BLE001 - one bad row must not stop the sweep
log.exception("could not claim schedule %s", schedule.id)
with contextlib.suppress(Exception):
schedule.enabled = False
schedule.last_error = "This schedule could not be read, so it was stopped."
# Committed before a single firing starts. This is the claim.
db.commit()
if not to_fire:
return 0
semaphore = asyncio.Semaphore(max(1, limit))
async def _guarded(schedule_id: str, due_at: datetime) -> None:
async with semaphore:
await runner.fire(schedule_id, due_at=due_at)
for schedule_id, due_at in to_fire:
task = asyncio.create_task(_guarded(schedule_id, due_at))
_FIRING.add(task)
task.add_done_callback(_FIRING.discard)
return len(to_fire)
def _interval() -> float:
with contextlib.suppress(Exception), session_scope() as db:
return float(settings_store.schedules(db).get("tick_seconds") or TICK_SECONDS)
return TICK_SECONDS
async def _loop() -> None:
while True:
try:
await asyncio.sleep(_interval())
await sweep()
except asyncio.CancelledError:
raise
except Exception: # noqa: BLE001 - the ticker must outlive one bad sweep
log.exception("the schedule ticker raised")
def start() -> None:
"""Begin ticking, once. Idempotent, so a second call in one process is not a
second ticker firing everything twice."""
global _TICKER
if _TICKER is None or _TICKER.done():
_TICKER = asyncio.create_task(_loop())
async def shutdown() -> None:
global _TICKER
if _TICKER is not None:
_TICKER.cancel()
with contextlib.suppress(asyncio.CancelledError, Exception):
await _TICKER
_TICKER = None
for task in list(_FIRING):
task.cancel()
for task in list(_FIRING):
with contextlib.suppress(asyncio.CancelledError, Exception):
await task
_FIRING.clear()
def release_claims() -> int:
"""Clear `claimed_at` on rows whose firing did not survive the last run.
A restart abandons a reply in flight -- that is already true of every
generation here -- so a schedule whose firing was interrupted would
otherwise carry a claim stamp for ever and read as permanently running.
"""
with session_scope() as db:
stuck = list(db.scalars(select(Schedule).where(Schedule.claimed_at.is_not(None))))
for schedule in stuck:
schedule.claimed_at = None
schedule.last_error = "This run was interrupted by a restart."
if stuck:
db.commit()
return len(stuck)