Something can happen because time passed, and land somewhere worth reading
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š
@@ -73,9 +73,6 @@ LAUNCH_GRACE = 10.0
|
||||
_JOBS: dict[str, JobState] = {}
|
||||
# One watcher task per job being polled to completion.
|
||||
_WATCHERS: dict[str, asyncio.Task] = {}
|
||||
# One lock per chat, so two jobs finishing at once cannot each start a reply --
|
||||
# see `wake`.
|
||||
_WAKE_LOCKS: dict[str, asyncio.Lock] = {}
|
||||
|
||||
# Stop watching a job after this. The remote process may keep running; we simply
|
||||
# stop holding a watcher for it and mark it lost. A job that runs longer than
|
||||
@@ -615,13 +612,6 @@ async def _watch(
|
||||
|
||||
|
||||
# --- Waking the model ----------------------------------------------------------
|
||||
def _lock(chat_id: str) -> asyncio.Lock:
|
||||
lock = _WAKE_LOCKS.get(chat_id)
|
||||
if lock is None:
|
||||
lock = _WAKE_LOCKS[chat_id] = asyncio.Lock()
|
||||
return lock
|
||||
|
||||
|
||||
def _completion_text(
|
||||
job_id: str, command: str, status: str, exit_status: int | None, output: str
|
||||
) -> str:
|
||||
@@ -651,48 +641,18 @@ async def wake(
|
||||
`queued` for that reply's `_inject`/`_drain` to deliver; if the chat is idle,
|
||||
a fresh reply is started to answer it, the `send_queued_now` move.
|
||||
|
||||
The whole thing is under a per-chat lock, and there is no `await` between the
|
||||
running-check and starting the reply, so two jobs finishing at once cannot
|
||||
each spin up a generation -- the second sees the first's reply already live
|
||||
and leaves its completion for it. That is the invariant the queue exists to
|
||||
hold, reached here from outside a request.
|
||||
|
||||
The completion is a user-role turn whose *content* names itself a machine
|
||||
event -- `_inject` sends a queued turn verbatim, so the framing cannot live
|
||||
there; it lives in the words, the way `execute_plan` quotes the plan.
|
||||
The lock discipline that makes that safe lives in `services/wake.py`, which
|
||||
is the one copy of it -- schedules need the identical rule, and two lock
|
||||
dictionaries for one invariant is how one of them drifts. What stays here is
|
||||
the *wording*, because `tool.background` quotes `_completion_text`'s opening
|
||||
sentence to the model and rewording it would break that instruction with
|
||||
nothing anywhere to notice.
|
||||
"""
|
||||
from lembas.db.models import ROLE_ASSISTANT, ROLE_USER, Chat
|
||||
from lembas.db.session import session_scope
|
||||
from lembas.services import chat as chat_service
|
||||
from lembas.services import generation as generation_service
|
||||
from lembas.services import wake as wake_service
|
||||
|
||||
content = _completion_text(job_id, command, status, exit_status, output)
|
||||
async with _lock(chat_id):
|
||||
running = generation_service.running_for(chat_id) is not None
|
||||
assistant_id = ""
|
||||
try:
|
||||
with session_scope() as db:
|
||||
chat = db.get(Chat, chat_id)
|
||||
if chat is None:
|
||||
return
|
||||
# `machine` changes the bubble and nothing else: the role stays
|
||||
# `user` because `_inject` sends a queued turn verbatim and the
|
||||
# request must carry a user turn, and the framing the *model*
|
||||
# reads is in the words `_completion_text` wrote. What it stops
|
||||
# is the transcript claiming the reader typed this.
|
||||
chat_service.create_message(
|
||||
db, chat, ROLE_USER, content, queued=running, machine=True
|
||||
)
|
||||
if not running:
|
||||
assistant = chat_service.create_message(
|
||||
db, chat, ROLE_ASSISTANT, "", complete_=False, model_id=chat.model_id
|
||||
)
|
||||
assistant_id = assistant.id
|
||||
except Exception: # noqa: BLE001 - a failed wake must not crash the watcher
|
||||
log.exception("could not wake chat %s for job %s", chat_id, job_id)
|
||||
return
|
||||
if assistant_id:
|
||||
generation_service.ensure(chat_id, assistant_id)
|
||||
await wake_service.wake_chat(
|
||||
chat_id, _completion_text(job_id, command, status, exit_status, output)
|
||||
)
|
||||
|
||||
|
||||
# --- Rehydration and shutdown --------------------------------------------------
|
||||
|
||||
@@ -10,6 +10,7 @@ from sqlalchemy import func, select
|
||||
from sqlalchemy.orm import Session as DBSession
|
||||
|
||||
from lembas.db.models import (
|
||||
KIND_MESSAGES,
|
||||
ROLE_ASSISTANT,
|
||||
ROLE_SYSTEM,
|
||||
ROLE_USER,
|
||||
@@ -265,6 +266,21 @@ def build_messages(
|
||||
select(Message).where(Message.chat_id == chat.id).order_by(Message.created_at)
|
||||
).all()
|
||||
|
||||
# The Messages conversation never ends, so it cannot all be sent. Only the
|
||||
# most recent turns go; everything before them stays on screen and out of
|
||||
# the request. One branch, and the bound is applied before the loop rather
|
||||
# than inside it so the filters below still see a contiguous tail.
|
||||
#
|
||||
# Not compaction: that summarises with a model call and a threshold, on a
|
||||
# conversation somebody decided to shorten. This is mechanical, lossless and
|
||||
# permanent, which is why `compaction.should_compact` refuses this kind --
|
||||
# two mechanisms fighting over one transcript is how you get a summary of a
|
||||
# summary.
|
||||
if chat.kind == KIND_MESSAGES:
|
||||
from lembas.services import messages as messages_service
|
||||
|
||||
history = history[-messages_service.LIVE_CHUNK :]
|
||||
|
||||
for message in history:
|
||||
if upto is not None and message.id == upto.id:
|
||||
break
|
||||
|
||||
@@ -25,7 +25,7 @@ from datetime import UTC, datetime
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.orm import Session as DBSession
|
||||
|
||||
from lembas.db.models import ROLE_ASSISTANT, Chat, Message
|
||||
from lembas.db.models import KIND_MESSAGES, ROLE_ASSISTANT, Chat, Message
|
||||
from lembas.services import metrics as metrics_service
|
||||
from lembas.services import settings_store, tokens
|
||||
|
||||
@@ -187,6 +187,14 @@ def should_compact(db: DBSession, chat: Chat, *, pending: str = "") -> bool:
|
||||
if limit <= 0:
|
||||
return False
|
||||
|
||||
# The Messages conversation bounds its own request mechanically, in
|
||||
# `build_messages`. Two mechanisms narrowing one transcript is how a summary
|
||||
# ends up summarising a summary -- and this one would be summarising turns
|
||||
# that are already outside the request, which achieves nothing at the cost
|
||||
# of a model call and a divider on a page that has no divider.
|
||||
if chat.kind == KIND_MESSAGES:
|
||||
return False
|
||||
|
||||
last = last_complete(db, chat)
|
||||
if last is None:
|
||||
return False
|
||||
|
||||
@@ -33,16 +33,16 @@ clearing those fragments in the admin page restores it exactly.
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from datetime import datetime
|
||||
from typing import Any
|
||||
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.orm import Session as DBSession
|
||||
|
||||
from lembas.db.models import User
|
||||
from lembas.db.models import KIND_TASK, User
|
||||
from lembas.services import prompts, settings_store
|
||||
from lembas.services.library import memories as memories_service
|
||||
from lembas.services.library import skills as skills_service
|
||||
from lembas.services.schedule import clock
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
@@ -176,11 +176,22 @@ def context_variables(
|
||||
|
||||
offered = tools or []
|
||||
families = _families(db, offered)
|
||||
stamp = datetime.now().astimezone()
|
||||
# The reader's zone, not the server's. Telling somebody in another country
|
||||
# that it is Tuesday when it is Wednesday where they are was survivable
|
||||
# while the answer was only ever prose; it stops being survivable the moment
|
||||
# they can say "every Monday at 3" and something has to work out when that
|
||||
# is. `zone_for` falls back to the server's, so an instance where nobody has
|
||||
# set one behaves exactly as it always did.
|
||||
stamp = clock.now_for(user)
|
||||
|
||||
values: dict[str, str] = {
|
||||
"today": stamp.strftime("%A %-d %B %Y"),
|
||||
"now": stamp.strftime("%A %-d %B %Y, %H:%M (UTC%z)"),
|
||||
# Named so a model working out a schedule can say which zone it meant,
|
||||
# and so `core.today` can carry it without a second fragment. Empty when
|
||||
# nobody has chosen one, which drops the line rather than printing the
|
||||
# server's zone as though it were a decision.
|
||||
"timezone": clock.name_for(user),
|
||||
"instance_name": str(settings_store.get(db, "instance_name") or "LLeMbas"),
|
||||
"user_name": (user.name or "") if user is not None else "",
|
||||
"model_name": "",
|
||||
@@ -230,6 +241,12 @@ def context_variables(
|
||||
"agent_instructions": "",
|
||||
"agent_instructions_file": "",
|
||||
"plan": "",
|
||||
# Empty everywhere but a scheduled task's own chat, which is what makes
|
||||
# it the gate on `core.unattended` as well as the content of
|
||||
# `context.schedule`. Two fragments, one variable, and no way for the
|
||||
# warning to appear without the thing it warns about.
|
||||
"schedule_instruction": "",
|
||||
"schedule_summary": "",
|
||||
}
|
||||
|
||||
if chat is not None:
|
||||
@@ -251,9 +268,37 @@ def context_variables(
|
||||
if "agent" in families:
|
||||
values.update(_agent_values(db, chat, user))
|
||||
|
||||
# Not gated on a family: a scheduled task has no tools of its own, and
|
||||
# the thing that must reach the model is precisely that nobody is
|
||||
# reading. One primary-key lookup, the same deal `plan` gets.
|
||||
if chat.kind == KIND_TASK:
|
||||
values.update(_schedule_values(db, chat, user))
|
||||
|
||||
return values
|
||||
|
||||
|
||||
def _schedule_values(db: DBSession, chat, user) -> dict[str, str]:
|
||||
"""What a scheduled task's chat is for, and how often it comes round.
|
||||
|
||||
A task chat accumulates every run, so by the tenth the original instruction
|
||||
is far out of sight up the transcript. Put back in front of the model each
|
||||
turn rather than left to be inferred -- exactly what `Chat.plan_message_id`
|
||||
exists to do for a plan.
|
||||
"""
|
||||
from lembas.services import schedules as schedules_service
|
||||
|
||||
schedule = schedules_service.for_chat(db, chat)
|
||||
if schedule is None:
|
||||
# The schedule was removed and its chat kept. There is nothing standing
|
||||
# to say, so the fragments vanish rather than describing a timer that no
|
||||
# longer exists.
|
||||
return {}
|
||||
return {
|
||||
"schedule_instruction": schedule.instruction or schedule.request or "",
|
||||
"schedule_summary": schedules_service.describe(schedule, owner=user),
|
||||
}
|
||||
|
||||
|
||||
def _agent_values(db: DBSession, chat, user) -> dict[str, str]:
|
||||
"""What an agent chat's harness needs to say about where it is."""
|
||||
from lembas.services import plans as plans_service
|
||||
|
||||
@@ -0,0 +1,156 @@
|
||||
"""Messages: one long-running conversation per person.
|
||||
|
||||
Signal-shaped rather than chat-shaped. There is exactly one of these per
|
||||
account, it is never titled, never filed and never deleted, and it is meant to
|
||||
run for years — which is the whole difficulty, because a conversation that never
|
||||
ends cannot all be sent to a model.
|
||||
|
||||
**What is stored and what is used are different things, and only the second is
|
||||
bounded.** Every turn is kept, for ever, and scrolling up shows all of them
|
||||
exactly as they were written. What reaches the model is the most recent
|
||||
`LIVE_CHUNK` turns and nothing before them.
|
||||
|
||||
**Nothing is folded into text and nothing is deleted**, and that is a
|
||||
deliberate reading of "compressed and history only". The visible conversation
|
||||
would be identical either way, so the only thing destroying the older turns
|
||||
would buy is disk — against which it is irreversible, it loses every attachment
|
||||
and tool call in the folded range, and it contradicts the rule this codebase
|
||||
already holds for compaction: *hiding turns is not deleting them*. Bounding the
|
||||
request achieves the whole of what the feature needs. If the rows ever do need
|
||||
folding, it is one function against this same boundary and the pages above it do
|
||||
not change.
|
||||
|
||||
The consequence is worth stating plainly rather than discovering: **a Messages
|
||||
conversation is infinite on screen and finite in the request.** Past the live
|
||||
chunk the model genuinely does not see what was said, and it is told so.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
|
||||
from sqlalchemy import func, select
|
||||
from sqlalchemy.orm import Session as DBSession
|
||||
|
||||
from lembas.db.models import KIND_MESSAGES, Chat, Message, User
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
# How many turns reach the model. The "latest chunk", and deliberately larger
|
||||
# than a page of history: it is the part that has to be enough to hold a
|
||||
# conversation in, while the rest only has to be readable.
|
||||
LIVE_CHUNK = 40
|
||||
|
||||
# How many older turns one scroll-up fetches. Bigger than the live chunk because
|
||||
# reading back is cheap -- no tokens, no request, just rows.
|
||||
HISTORY_PAGE = 100
|
||||
|
||||
|
||||
def for_user(db: DBSession, user: User) -> Chat:
|
||||
"""This person's Messages conversation, made if it is not there yet.
|
||||
|
||||
The second deliberate exception to "chats are created lazily", and for a
|
||||
different reason than a task chat's: a schedule can post in here before
|
||||
anybody has ever opened the page, and `wake_chat` needs a row to write to.
|
||||
Get-or-create rather than a startup sweep, so an account that never opens
|
||||
Messages never grows one.
|
||||
"""
|
||||
from lembas.services import chat as chat_service
|
||||
|
||||
existing = db.scalars(
|
||||
select(Chat)
|
||||
.where(Chat.user_id == user.id, Chat.kind == KIND_MESSAGES)
|
||||
.order_by(Chat.created_at)
|
||||
).first()
|
||||
if existing is not None:
|
||||
return existing
|
||||
|
||||
# `default_model` answers with the *pair* -- the model id and the connection
|
||||
# it was reached through -- because a chat stores both and resolving the
|
||||
# second later would pick whichever connection happens to offer the id.
|
||||
# Unpacked rather than assigned, which is the mistake this comment exists to
|
||||
# stop being made again: assigning the tuple straight to `model_id` writes a
|
||||
# tuple into a String column and SQLite refuses the insert.
|
||||
chosen = chat_service.default_model(db, user)
|
||||
model_id, connection_id = chosen if chosen else ("", None)
|
||||
|
||||
conversation = Chat(
|
||||
user_id=user.id,
|
||||
kind=KIND_MESSAGES,
|
||||
title="Messages",
|
||||
# Titling never runs on this one: there is no first exchange to name and
|
||||
# the name is fixed. Set so nothing downstream has to special-case it.
|
||||
title_generated=True,
|
||||
model_id=model_id,
|
||||
connection_id=connection_id,
|
||||
)
|
||||
db.add(conversation)
|
||||
db.commit()
|
||||
return conversation
|
||||
|
||||
|
||||
def count(db: DBSession, chat: Chat) -> int:
|
||||
return int(
|
||||
db.scalar(select(func.count()).select_from(Message).where(Message.chat_id == chat.id))
|
||||
or 0
|
||||
)
|
||||
|
||||
|
||||
def live_messages(db: DBSession, chat: Chat, *, limit: int = LIVE_CHUNK) -> list[Message]:
|
||||
"""The most recent turns, oldest first.
|
||||
|
||||
Fetched newest-first and reversed rather than offset from the start: an
|
||||
offset would have to be recomputed from a count on every request, and would
|
||||
be wrong the moment a turn arrived between the two queries.
|
||||
"""
|
||||
newest = db.scalars(
|
||||
select(Message)
|
||||
.where(Message.chat_id == chat.id)
|
||||
.order_by(Message.created_at.desc(), Message.id.desc())
|
||||
.limit(limit)
|
||||
).all()
|
||||
return list(reversed(newest))
|
||||
|
||||
|
||||
def older_than(
|
||||
db: DBSession, chat: Chat, cursor: Message, *, limit: int = HISTORY_PAGE
|
||||
) -> list[Message]:
|
||||
"""The page of turns immediately before `cursor`, oldest first.
|
||||
|
||||
The comparison is done in SQL with an `id` tie-breaker, exactly as
|
||||
`thread_tail` does going the other way. That is not decoration: under a bare
|
||||
`<`, a row sharing the cursor's microsecond can never be reached, and a
|
||||
message that cannot be scrolled back to is a message that is gone.
|
||||
"""
|
||||
rows = db.scalars(
|
||||
select(Message)
|
||||
.where(
|
||||
Message.chat_id == chat.id,
|
||||
(Message.created_at < cursor.created_at)
|
||||
| ((Message.created_at == cursor.created_at) & (Message.id < cursor.id)),
|
||||
)
|
||||
.order_by(Message.created_at.desc(), Message.id.desc())
|
||||
.limit(limit)
|
||||
).all()
|
||||
return list(reversed(rows))
|
||||
|
||||
|
||||
def has_more_before(db: DBSession, chat: Chat, cursor: Message) -> bool:
|
||||
"""Whether the sentinel should be rendered again above a page.
|
||||
|
||||
Asked separately rather than by fetching one extra row, because the answer
|
||||
is needed *after* the page has been reversed and the extra row would have to
|
||||
be trimmed off the wrong end.
|
||||
"""
|
||||
return (
|
||||
db.scalar(
|
||||
select(func.count())
|
||||
.select_from(Message)
|
||||
.where(
|
||||
Message.chat_id == chat.id,
|
||||
(Message.created_at < cursor.created_at)
|
||||
| ((Message.created_at == cursor.created_at) & (Message.id < cursor.id)),
|
||||
)
|
||||
)
|
||||
or 0
|
||||
) > 0
|
||||
@@ -124,6 +124,24 @@ class Variable:
|
||||
VARIABLES: tuple[Variable, ...] = (
|
||||
Variable("today", "Today's date", "The current date, written out in full."),
|
||||
Variable("now", "Date and time", "The current date and time, with the offset from UTC."),
|
||||
Variable(
|
||||
"schedule_instruction",
|
||||
"Scheduled instruction",
|
||||
"In a scheduled task's chat: what it is to do each time it runs. Empty "
|
||||
"everywhere else, which is what makes it the gate on the unattended "
|
||||
"guidance as well as its content.",
|
||||
),
|
||||
Variable(
|
||||
"schedule_summary",
|
||||
"Schedule",
|
||||
"In a scheduled task's chat: how often it runs, in words.",
|
||||
),
|
||||
Variable(
|
||||
"timezone",
|
||||
"Timezone",
|
||||
"The reader's timezone, as an IANA name. Empty when they have not chosen "
|
||||
"one, in which case the times above are the server's.",
|
||||
),
|
||||
Variable("instance_name", "Instance name", "What this installation is called."),
|
||||
Variable("user_name", "User's name", "The name of the person in the conversation."),
|
||||
Variable("model_name", "Model", "The display name of the model answering."),
|
||||
@@ -248,6 +266,18 @@ VARIABLES: tuple[Variable, ...] = (
|
||||
),
|
||||
Variable("question", "Question", "The first message. Chat title task only."),
|
||||
Variable("answer", "Answer", "The first reply. Chat title task only."),
|
||||
Variable(
|
||||
"request",
|
||||
"The request",
|
||||
"What somebody said they wanted to happen, in their own words. "
|
||||
"Working out a schedule only.",
|
||||
),
|
||||
Variable(
|
||||
"targets",
|
||||
"Destinations",
|
||||
"Where a scheduled run's result may be sent, as a list of the values "
|
||||
"that are accepted. Working out a schedule only.",
|
||||
),
|
||||
Variable(
|
||||
"transcript",
|
||||
"Transcript",
|
||||
@@ -573,13 +603,16 @@ BUILTIN: tuple[Fragment, ...] = (
|
||||
label="Today's date",
|
||||
group=GROUP_CORE,
|
||||
order=20,
|
||||
variables=("today",),
|
||||
variables=("today", "timezone"),
|
||||
hint="A model has no clock. Without this it cannot tell whether what it "
|
||||
"recalls is current, and will not think to check.",
|
||||
"recalls is current, and will not think to check. The timezone line "
|
||||
"carries its own variable, so it disappears on an instance where nobody "
|
||||
"has chosen one rather than announcing the server's as a decision.",
|
||||
default=(
|
||||
"Today is {{today}}. Your training data stops well before this, so treat "
|
||||
"anything time-sensitive as something to check rather than something you "
|
||||
"already know."
|
||||
"already know.\n"
|
||||
"- Times the person gives you are in {{timezone}} unless they say otherwise."
|
||||
),
|
||||
),
|
||||
Fragment(
|
||||
@@ -1117,6 +1150,72 @@ BUILTIN: tuple[Fragment, ...] = (
|
||||
),
|
||||
),
|
||||
# --- Context -------------------------------------------------------------
|
||||
Fragment(
|
||||
key="core.unattended",
|
||||
label="Nobody is watching",
|
||||
group=GROUP_CORE,
|
||||
order=35,
|
||||
requires=("schedule_instruction",),
|
||||
hint="Only in a scheduled task's chat. The point a model cannot work "
|
||||
"out for itself is that there is no reader — so the usual moves of "
|
||||
"asking what was meant, or stopping to check, end the run having done "
|
||||
"nothing. This is the prompt half; the enforcement is that `ask_user` "
|
||||
"is not offered here at all, because a rule living only in a system "
|
||||
"message is one a page the model just read can argue with.",
|
||||
default=(
|
||||
"- This chat runs on a schedule and nobody is necessarily reading it. You "
|
||||
"cannot ask a question and wait for an answer: there is no one to answer, "
|
||||
"and the run would simply end. Where something is ambiguous, choose the "
|
||||
"most reasonable reading, do the work, and say plainly in your reply what "
|
||||
"you assumed and what you would want confirmed. Finish what you were asked "
|
||||
"to do in this one reply."
|
||||
),
|
||||
),
|
||||
Fragment(
|
||||
key="context.schedule",
|
||||
label="What this task is for",
|
||||
group=GROUP_CONTEXT,
|
||||
order=310,
|
||||
variables=("schedule_instruction", "schedule_summary"),
|
||||
requires=("schedule_instruction",),
|
||||
hint="A task chat accumulates every run, so by the tenth the original "
|
||||
"instruction is far out of sight. This puts it back in front of the "
|
||||
"model each turn, the same way the current plan is — one lookup, and no "
|
||||
"guessing from the transcript.",
|
||||
default=(
|
||||
"## This scheduled task\n"
|
||||
"It runs: {{schedule_summary}}\n"
|
||||
"Each time, you are to: {{schedule_instruction}}\n"
|
||||
"Earlier runs are above. Say what has changed since the last one rather "
|
||||
"than repeating it, unless there is nothing above to compare with."
|
||||
),
|
||||
),
|
||||
Fragment(
|
||||
key="tool.report",
|
||||
label="Reports",
|
||||
group=GROUP_TOOLS,
|
||||
order=246,
|
||||
families=("report",),
|
||||
hint="Appears when the report tools are offered. The whole of what a "
|
||||
"model cannot infer from the schema is the audience: a report is read "
|
||||
"somewhere else, later, by somebody who cannot answer it. Everything "
|
||||
"else here follows from that — write it whole, do not end on a "
|
||||
"question, and do not file one for a two-line answer that has already "
|
||||
"been given in the conversation.",
|
||||
default=(
|
||||
"- You can file a report with report_write: a finished piece of work, kept "
|
||||
"where the person will find it later. Write one when you are asked for one, "
|
||||
"and when you finish something long enough that its result is worth keeping — "
|
||||
"an investigation, an account of what you changed, a summary of what you "
|
||||
"found. Do not file one for an answer you have just given in two lines; the "
|
||||
"conversation already holds that. A report is read on its own, away from this "
|
||||
"chat and possibly long afterwards, and the person cannot reply to it — so "
|
||||
"say what you were asked, what you found and what you conclude, refer to "
|
||||
"nothing above, and end on a finding rather than a question. report_search "
|
||||
"and report_get read back ones filed earlier, which is worth doing before a "
|
||||
"recurring report so this one can say what changed."
|
||||
),
|
||||
),
|
||||
Fragment(
|
||||
key="context.knowledge_scope",
|
||||
label="Which knowledge bases",
|
||||
@@ -1518,6 +1617,57 @@ BUILTIN: tuple[Fragment, ...] = (
|
||||
"from there."
|
||||
),
|
||||
),
|
||||
Fragment(
|
||||
key="task.schedule_compile",
|
||||
label="Working out a schedule",
|
||||
group=GROUP_TASKS,
|
||||
order=440,
|
||||
variables=("request", "now", "timezone", "targets"),
|
||||
hint="One request, made once, when somebody describes something they "
|
||||
"want to happen later. It turns their words into a recurrence and into "
|
||||
"an instruction that reads sensibly with no conversation around it — "
|
||||
"which is how it will be read, days later, by a model that was not "
|
||||
"there when it was typed. Clearing this switches off the *working out*, "
|
||||
"not scheduling: the setup screen then asks for the time in its own "
|
||||
"fields, with the reader's words already filled in. The reply is parsed "
|
||||
"leniently and anything unusable falls back to that same form, so a "
|
||||
"model that answers in prose costs a moment rather than a broken "
|
||||
"schedule.",
|
||||
default=(
|
||||
"Turn the request below into a schedule. Reply with one JSON object and "
|
||||
"nothing else — no commentary, no code fence.\n"
|
||||
"\n"
|
||||
"It is currently {{now}} ({{timezone}}). Times you write are in that zone.\n"
|
||||
"\n"
|
||||
"The object has these keys:\n"
|
||||
'- "title": a short name for this, five words or fewer.\n'
|
||||
'- "instruction": what should be done each time it runs, written out in '
|
||||
"full. It will be read on its own, with none of this conversation around "
|
||||
"it and nobody available to answer a question about it, so say everything "
|
||||
"it needs. Write it as an instruction, not as a description.\n"
|
||||
'- "target": where the result goes — one of: {{targets}}. Use "report" '
|
||||
"when the point is something to read later, and \"chat\" otherwise.\n"
|
||||
'- "schedule": an object saying when, with these optional keys:\n'
|
||||
' "start": an ISO timestamp for the first (or only) run.\n'
|
||||
' "every": one of {"minutes": n}, {"hours": n}, {"days": n}, '
|
||||
'{"weeks": n} — a plain timer.\n'
|
||||
' "at": {"weekdays": [0-6, Monday is 0], "days": [1-31], '
|
||||
'"months": [1-12], "times": ["HH:MM"]} — a calendar. Leave a list out '
|
||||
"to mean every one of them.\n"
|
||||
' "count": how many times in total, if they said a number.\n'
|
||||
' "until": an ISO timestamp to stop after, if they gave one.\n'
|
||||
"\n"
|
||||
'Use "every" for "in ten minutes" or "every six hours". Use "at" for '
|
||||
'"every Monday at 3" or "daily at nine". Use both only for something '
|
||||
'like "every other Tuesday". For a one-off, give "start" alone.\n'
|
||||
"\n"
|
||||
"If they did not say when, guess the most ordinary reading rather than "
|
||||
"leaving it out — daily at 09:00 for something described as daily.\n"
|
||||
"\n"
|
||||
"The request:\n"
|
||||
"{{request}}"
|
||||
),
|
||||
),
|
||||
)
|
||||
|
||||
register_source(_builtin_source)
|
||||
|
||||
@@ -0,0 +1,159 @@
|
||||
"""Reports: filing a finished piece of work, and finding it again.
|
||||
|
||||
A report is written and read; it is never answered. That is the whole shape of
|
||||
the thing, and it is why this store is deliberately thinner than
|
||||
`services/library/`: there is no sharing, because a report is a record of what
|
||||
somebody's own model did on their behalf, and no revisions, because a report
|
||||
describes a moment rather than a document being worked on.
|
||||
|
||||
`sharing.visible_to` is therefore absent on purpose rather than forgotten. If
|
||||
reports ever become shareable, `RESOURCE_TYPES` is where that starts, and every
|
||||
listing here has to go through the helper -- six independently written
|
||||
permission checks is how one of them ends up written slightly differently.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
|
||||
from sqlalchemy import func, select
|
||||
from sqlalchemy.orm import Session as DBSession
|
||||
|
||||
from lembas.db.models import SOURCE_MANUAL, SOURCES, Report, User
|
||||
from lembas.services.library.fts import search_ids
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
INDEX = "reports_fts"
|
||||
|
||||
MAX_TITLE_CHARS = 300
|
||||
MAX_SUMMARY_CHARS = 500
|
||||
MAX_BODY_CHARS = 60_000
|
||||
SNIPPET_CHARS = 400
|
||||
|
||||
|
||||
def visible(user: User | None):
|
||||
"""Every report this person owns.
|
||||
|
||||
Takes no session because it builds a query rather than running one, and
|
||||
takes `None` to mean nobody so an unauthenticated caller gets an empty
|
||||
result instead of an exception -- the same shape `sharing.visible_to` has,
|
||||
so a later move to shared reports is a change of one line here.
|
||||
"""
|
||||
if user is None:
|
||||
return select(Report).where(Report.id.is_(None))
|
||||
return select(Report).where(Report.owner_id == user.id)
|
||||
|
||||
|
||||
def get(db: DBSession, report_id: str, user: User | None) -> Report | None:
|
||||
report = db.get(Report, report_id)
|
||||
if report is None or user is None or report.owner_id != user.id:
|
||||
return None
|
||||
return report
|
||||
|
||||
|
||||
def recent(db: DBSession, user: User | None, *, limit: int = 20) -> list[Report]:
|
||||
return list(db.scalars(visible(user).order_by(Report.created_at.desc()).limit(limit)))
|
||||
|
||||
|
||||
def search(db: DBSession, user: User | None, needle: str, *, limit: int = 20) -> list[Report]:
|
||||
"""Reports matching `needle`, best match first.
|
||||
|
||||
Ids come back from FTS and the rows are re-ordered by hit position, exactly
|
||||
as the library stores do -- the index knows about ranking and the ORM query
|
||||
knows about ownership, and neither is asked to do the other's job.
|
||||
"""
|
||||
hits = search_ids(db, INDEX, needle, limit=limit * 4)
|
||||
if not hits:
|
||||
return []
|
||||
order = {hit.id: position for position, hit in enumerate(hits)}
|
||||
rows = list(db.scalars(visible(user).where(Report.id.in_(list(order)))))
|
||||
rows.sort(key=lambda report: order.get(report.id, len(order)))
|
||||
return rows[:limit]
|
||||
|
||||
|
||||
def unread_count(db: DBSession, user: User | None) -> int:
|
||||
if user is None:
|
||||
return 0
|
||||
return int(
|
||||
db.scalar(
|
||||
select(func.count()).select_from(Report).where(
|
||||
Report.owner_id == user.id, Report.unread.is_(True)
|
||||
)
|
||||
)
|
||||
or 0
|
||||
)
|
||||
|
||||
|
||||
def _first_line(body: str) -> str:
|
||||
"""A summary for a model that did not write one.
|
||||
|
||||
Markdown headings are stripped rather than shown: a list of reports all
|
||||
beginning "# " reads as a bug, and the heading is nearly always the title
|
||||
again.
|
||||
"""
|
||||
for line in (body or "").splitlines():
|
||||
stripped = line.strip().lstrip("#").strip()
|
||||
if stripped:
|
||||
return stripped[:MAX_SUMMARY_CHARS]
|
||||
return ""
|
||||
|
||||
|
||||
def create(
|
||||
db: DBSession,
|
||||
*,
|
||||
owner: User,
|
||||
title: str,
|
||||
body: str,
|
||||
summary: str = "",
|
||||
source: str = SOURCE_MANUAL,
|
||||
source_id: str = "",
|
||||
schedule_id: str = "",
|
||||
model_id: str = "",
|
||||
error: str = "",
|
||||
unread: bool = True,
|
||||
) -> Report:
|
||||
"""File a report.
|
||||
|
||||
Trimming happens here rather than at the column so an over-long write from
|
||||
a tool is filed short with everything else intact, instead of failing the
|
||||
turn -- the rule `memories` already follows.
|
||||
|
||||
`unread` defaults to True because every caller that matters is something
|
||||
that happened without the reader present. A report somebody typed themselves
|
||||
passes False.
|
||||
"""
|
||||
report = Report(
|
||||
owner_id=owner.id,
|
||||
title=(title.strip() or "Untitled report")[:MAX_TITLE_CHARS],
|
||||
summary=(summary.strip() or _first_line(body))[:MAX_SUMMARY_CHARS],
|
||||
body=(body or "").strip()[:MAX_BODY_CHARS],
|
||||
source=source if source in SOURCES else SOURCE_MANUAL,
|
||||
source_id=source_id or "",
|
||||
schedule_id=schedule_id or "",
|
||||
model_id=model_id or "",
|
||||
error=error or "",
|
||||
unread=unread,
|
||||
)
|
||||
db.add(report)
|
||||
db.commit()
|
||||
return report
|
||||
|
||||
|
||||
def mark_read(db: DBSession, report: Report) -> Report:
|
||||
if report.unread:
|
||||
report.unread = False
|
||||
db.commit()
|
||||
return report
|
||||
|
||||
|
||||
def delete(db: DBSession, report: Report) -> None:
|
||||
db.delete(report)
|
||||
db.commit()
|
||||
|
||||
|
||||
def snippet(report: Report) -> str:
|
||||
text = (report.summary or report.body or "").strip()
|
||||
if len(text) <= SNIPPET_CHARS:
|
||||
return text
|
||||
return text[:SNIPPET_CHARS].rstrip() + "…"
|
||||
@@ -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
|
||||
@@ -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)
|
||||
@@ -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
|
||||
@@ -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)
|
||||
@@ -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)
|
||||
@@ -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)
|
||||
@@ -0,0 +1,224 @@
|
||||
"""Making, changing and stopping a schedule.
|
||||
|
||||
The row-level half: what the routes and the tools both need, so neither has its
|
||||
own idea of what creating a schedule involves. `services/schedule/` holds the
|
||||
machinery — when it next comes due, and what happens when it does.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from datetime import UTC, datetime
|
||||
|
||||
from sqlalchemy import func, select
|
||||
from sqlalchemy.orm import Session as DBSession
|
||||
|
||||
from lembas.db.models import (
|
||||
KIND_TASK,
|
||||
ORIGIN_USER,
|
||||
ORIGINS,
|
||||
TARGET_CHAT,
|
||||
TARGETS,
|
||||
Chat,
|
||||
Schedule,
|
||||
User,
|
||||
)
|
||||
from lembas.services import settings_store
|
||||
from lembas.services.schedule import clock
|
||||
from lembas.services.schedule import rule as rule_service
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
MAX_TITLE_CHARS = 200
|
||||
MAX_INSTRUCTION_CHARS = 8000
|
||||
|
||||
|
||||
class ScheduleError(Exception):
|
||||
"""Something a person needs told, in words they can act on."""
|
||||
|
||||
|
||||
def visible(user: User | None):
|
||||
if user is None:
|
||||
return select(Schedule).where(Schedule.id.is_(None))
|
||||
return select(Schedule).where(Schedule.user_id == user.id)
|
||||
|
||||
|
||||
def get(db: DBSession, schedule_id: str, user: User | None) -> Schedule | None:
|
||||
schedule = db.get(Schedule, schedule_id)
|
||||
if schedule is None or user is None or schedule.user_id != user.id:
|
||||
return None
|
||||
return schedule
|
||||
|
||||
|
||||
def for_chat(db: DBSession, chat: Chat) -> Schedule | None:
|
||||
"""The schedule a task chat belongs to, if any."""
|
||||
return db.scalars(select(Schedule).where(Schedule.chat_id == chat.id)).first()
|
||||
|
||||
|
||||
def count_for(db: DBSession, user: User) -> int:
|
||||
return int(
|
||||
db.scalar(
|
||||
select(func.count()).select_from(Schedule).where(Schedule.user_id == user.id)
|
||||
)
|
||||
or 0
|
||||
)
|
||||
|
||||
|
||||
def create(
|
||||
db: DBSession,
|
||||
*,
|
||||
owner: User,
|
||||
title: str,
|
||||
instruction: str,
|
||||
rule: dict,
|
||||
request: str = "",
|
||||
target: str = TARGET_CHAT,
|
||||
model_id: str = "",
|
||||
origin: str = ORIGIN_USER,
|
||||
) -> Schedule:
|
||||
"""Write a schedule and the chat it fires into.
|
||||
|
||||
**The chat is created here, with the schedule**, and this is the one place
|
||||
"chats are created lazily" is deliberately bent. That rule exists so an
|
||||
opened-and-abandoned chat never appears in the sidebar; a task chat is not
|
||||
opened and abandoned, because creating it *is* the act. It also has to exist
|
||||
before the first firing, which may be days away and will have nobody
|
||||
present to make one.
|
||||
|
||||
A task chat is never `KIND_AGENT`. Scheduling one would mean running
|
||||
commands on a timer with nobody watching — and since every mode except Auto
|
||||
stalls waiting for an approval that will not come, the only two outcomes are
|
||||
"unattended execution" and "does nothing". That deserves its own pass with a
|
||||
mode built for it, not a flag here.
|
||||
"""
|
||||
clean = rule_service.validate(rule)
|
||||
if not clean:
|
||||
raise ScheduleError(
|
||||
"That does not describe a time anything could run at. "
|
||||
"Say when it should happen — a date and time, or how often."
|
||||
)
|
||||
if rule_service.next_after(clean, datetime.now(tz=UTC), zone=clock.zone_for(owner)) is None:
|
||||
# Belt and braces over `validate`'s own invariant. A schedule that can
|
||||
# never fire looks exactly like a working one on every screen it appears
|
||||
# on, so it is refused at the only moment somebody is present to be told.
|
||||
raise ScheduleError("That schedule has no next run — its time has already passed.")
|
||||
|
||||
limit = int(settings_store.schedules(db).get("max_per_user") or 20)
|
||||
if count_for(db, owner) >= limit:
|
||||
raise ScheduleError(
|
||||
f"You already have {limit} schedules, which is the most this instance allows. "
|
||||
"Remove one before adding another."
|
||||
)
|
||||
|
||||
chat = Chat(
|
||||
user_id=owner.id,
|
||||
kind=KIND_TASK,
|
||||
title=(title.strip() or "Scheduled task")[:MAX_TITLE_CHARS],
|
||||
model_id=model_id or "",
|
||||
)
|
||||
db.add(chat)
|
||||
db.flush()
|
||||
|
||||
schedule = Schedule(
|
||||
user_id=owner.id,
|
||||
title=(title.strip() or "Scheduled task")[:MAX_TITLE_CHARS],
|
||||
request=(request or "").strip()[:MAX_INSTRUCTION_CHARS],
|
||||
instruction=(instruction or "").strip()[:MAX_INSTRUCTION_CHARS],
|
||||
rule_json=clean,
|
||||
target=target if target in TARGETS else TARGET_CHAT,
|
||||
chat_id=chat.id,
|
||||
model_id=model_id or "",
|
||||
origin=origin if origin in ORIGINS else ORIGIN_USER,
|
||||
enabled=True,
|
||||
next_fire_at=rule_service.next_after(
|
||||
clean, datetime.now(tz=UTC), zone=clock.zone_for(owner)
|
||||
),
|
||||
compiled_at=datetime.now(tz=UTC),
|
||||
)
|
||||
db.add(schedule)
|
||||
db.commit()
|
||||
return schedule
|
||||
|
||||
|
||||
def update(
|
||||
db: DBSession,
|
||||
schedule: Schedule,
|
||||
*,
|
||||
owner: User,
|
||||
title: str | None = None,
|
||||
instruction: str | None = None,
|
||||
rule: dict | None = None,
|
||||
target: str | None = None,
|
||||
) -> Schedule:
|
||||
"""Change a schedule. Absent arguments are left alone."""
|
||||
if title is not None and title.strip():
|
||||
schedule.title = title.strip()[:MAX_TITLE_CHARS]
|
||||
if instruction is not None:
|
||||
schedule.instruction = instruction.strip()[:MAX_INSTRUCTION_CHARS]
|
||||
if target is not None and target in TARGETS:
|
||||
schedule.target = target
|
||||
if rule is not None:
|
||||
clean = rule_service.validate(rule)
|
||||
if not clean:
|
||||
raise ScheduleError(
|
||||
"That does not describe a time anything could run at. "
|
||||
"Say when it should happen — a date and time, or how often."
|
||||
)
|
||||
schedule.rule_json = clean
|
||||
# Recomputed from now, and the count restarted: an edited schedule is a
|
||||
# new intention, and carrying the old `fired_count` into a new `count`
|
||||
# would silently spend most of it before the first run.
|
||||
schedule.fired_count = 0
|
||||
schedule.next_fire_at = rule_service.next_after(
|
||||
clean, datetime.now(tz=UTC), zone=clock.zone_for(owner)
|
||||
)
|
||||
if schedule.next_fire_at is None:
|
||||
raise ScheduleError("That schedule has no next run — its time has already passed.")
|
||||
db.commit()
|
||||
return schedule
|
||||
|
||||
|
||||
def set_enabled(db: DBSession, schedule: Schedule, *, owner: User, enabled: bool) -> Schedule:
|
||||
"""Pause or resume.
|
||||
|
||||
**Resuming recomputes from now**, never from the stored value. A schedule
|
||||
paused for a month would otherwise come back due — and with catching-up in
|
||||
the sweep, it would fire the moment it was switched on, having decided it
|
||||
was owed a run from four weeks ago.
|
||||
"""
|
||||
schedule.enabled = bool(enabled)
|
||||
if enabled:
|
||||
schedule.last_error = ""
|
||||
schedule.next_fire_at = rule_service.next_after(
|
||||
schedule.rule_json or {},
|
||||
datetime.now(tz=UTC),
|
||||
zone=clock.zone_for(owner),
|
||||
fired=schedule.fired_count or 0,
|
||||
)
|
||||
if schedule.next_fire_at is None:
|
||||
schedule.enabled = False
|
||||
schedule.last_error = "There are no runs left in this schedule."
|
||||
db.commit()
|
||||
return schedule
|
||||
|
||||
|
||||
def delete(db: DBSession, schedule: Schedule, *, keep_chat: bool = True) -> None:
|
||||
"""Remove a schedule, and by default keep its transcript.
|
||||
|
||||
Keeping is the default because deleting a conversation as a side effect of
|
||||
removing a timer is exactly the destructive default this codebase avoids
|
||||
elsewhere. The chat becomes an ordinary one so it is reachable again — a
|
||||
`KIND_TASK` chat with no schedule behind it would be in no list at all.
|
||||
"""
|
||||
chat = db.get(Chat, schedule.chat_id) if schedule.chat_id else None
|
||||
if chat is not None:
|
||||
if keep_chat:
|
||||
chat.kind = "chat"
|
||||
else:
|
||||
db.delete(chat)
|
||||
db.delete(schedule)
|
||||
db.commit()
|
||||
|
||||
|
||||
def describe(schedule: Schedule, *, owner: User | None) -> str:
|
||||
return rule_service.describe(schedule.rule_json or {}, zone=clock.zone_for(owner))
|
||||
@@ -30,6 +30,7 @@ SEARCH = "search"
|
||||
PROMPTS = "prompts"
|
||||
AGENTS = "agents"
|
||||
IMAGES = "images"
|
||||
SCHEDULES = "schedules"
|
||||
|
||||
|
||||
def _general_defaults() -> dict[str, Any]:
|
||||
@@ -272,6 +273,36 @@ def _images_defaults() -> dict[str, Any]:
|
||||
}
|
||||
|
||||
|
||||
def _schedules_defaults() -> dict[str, Any]:
|
||||
"""Scheduling: work that happens because time passed rather than because
|
||||
somebody asked just now.
|
||||
|
||||
Off until an administrator turns it on, for the reason agent execution is:
|
||||
this spends model time — and, in an agent chat, runs commands — with nobody
|
||||
at the keyboard, which is a capability somebody chooses on purpose rather
|
||||
than one that arrives with an upgrade.
|
||||
"""
|
||||
return {
|
||||
"enabled": False,
|
||||
# How often the ticker looks. The rule's own granularity is a minute, so
|
||||
# this bounds how late a firing can be; 30s costs one indexed SELECT.
|
||||
"tick_seconds": 30,
|
||||
# A ceiling per person, so one account cannot fill the ticker's sweep.
|
||||
"max_per_user": 20,
|
||||
# Firings running at once. Fifty schedules due at 09:00 must not open
|
||||
# fifty generations against one endpoint.
|
||||
"max_concurrent": 3,
|
||||
# The floor `rule.validate` clamps an interval up to. Separate from the
|
||||
# rule module's own hard minimum: an administrator may want a coarser
|
||||
# floor than "a minute" without editing code.
|
||||
"min_interval_seconds": 60,
|
||||
# How many turns may pile up unanswered in one chat before a firing is
|
||||
# skipped instead of queued. `_drain` takes one per reply, so an
|
||||
# unbounded queue is a backlog that outlives the day that caused it.
|
||||
"max_queued": 3,
|
||||
}
|
||||
|
||||
|
||||
_DEFAULTS: dict[str, Any] = {
|
||||
GENERAL: _general_defaults,
|
||||
AUDIO: _audio_defaults,
|
||||
@@ -279,6 +310,7 @@ _DEFAULTS: dict[str, Any] = {
|
||||
PROMPTS: _prompts_defaults,
|
||||
AGENTS: _agents_defaults,
|
||||
IMAGES: _images_defaults,
|
||||
SCHEDULES: _schedules_defaults,
|
||||
}
|
||||
|
||||
|
||||
@@ -411,6 +443,26 @@ def images(db: DBSession) -> dict[str, Any]:
|
||||
return values
|
||||
|
||||
|
||||
def schedules(db: DBSession) -> dict[str, Any]:
|
||||
"""Scheduling settings, with the numbers clamped on read.
|
||||
|
||||
Clamped here rather than at the save, for the reason `agents` gives: a value
|
||||
stored by an earlier version cannot bite either. Every floor below is a
|
||||
number that means something bad at zero -- a tick of 0 is a busy loop, a
|
||||
concurrency of 0 is a ticker that claims firings and never runs them, and
|
||||
both would look from the outside like scheduling simply not working.
|
||||
"""
|
||||
values = get_group(db, SCHEDULES)
|
||||
values["tick_seconds"] = min(max(int(values.get("tick_seconds") or 30), 5), 300)
|
||||
values["max_per_user"] = min(max(int(values.get("max_per_user") or 20), 1), 200)
|
||||
values["max_concurrent"] = min(max(int(values.get("max_concurrent") or 3), 1), 20)
|
||||
values["min_interval_seconds"] = min(
|
||||
max(int(values.get("min_interval_seconds") or 60), 60), 86400
|
||||
)
|
||||
values["max_queued"] = min(max(int(values.get("max_queued") or 3), 1), 50)
|
||||
return values
|
||||
|
||||
|
||||
def images_ready(db: DBSession) -> bool:
|
||||
"""Whether image generation can actually happen.
|
||||
|
||||
|
||||
@@ -62,6 +62,10 @@ LABELS: dict[str, str] = {
|
||||
"notes_edit": "Note updated",
|
||||
"notes_delete": "Note deleted",
|
||||
"scratch_write": "Canvas written",
|
||||
# Reports.
|
||||
"report_write": "Report filed",
|
||||
"report_search": "Reports searched",
|
||||
"report_get": "Report read",
|
||||
"image_generate": "Image",
|
||||
"memory_add": "Memory saved",
|
||||
"memory_forget": "Memory removed",
|
||||
@@ -95,6 +99,9 @@ ICONS: dict[str, str] = {
|
||||
"notes_edit": "pencil",
|
||||
"notes_delete": "trash",
|
||||
"scratch_write": "file-text",
|
||||
"report_write": "pencil",
|
||||
"report_search": "search",
|
||||
"report_get": "file-text",
|
||||
"image_generate": "image",
|
||||
"memory_add": "star",
|
||||
"memory_forget": "trash",
|
||||
@@ -133,6 +140,9 @@ ACTIONS: dict[str, str] = {
|
||||
"notes_edit": "Change a note",
|
||||
"notes_delete": "Delete a note",
|
||||
"scratch_write": "Write in the canvas",
|
||||
"report_write": "File a report",
|
||||
"report_search": "Search reports",
|
||||
"report_get": "Read a report",
|
||||
"image_generate": "Generate an image",
|
||||
"memory_add": "Remember something",
|
||||
"memory_forget": "Forget something",
|
||||
@@ -155,6 +165,12 @@ DETAIL_KEYS: dict[str, str] = {
|
||||
"web_search": "query",
|
||||
"knowledge_search": "query",
|
||||
"notes_search": "query",
|
||||
"report_search": "query",
|
||||
# The title, not the body: a card has room for a line and the body is the
|
||||
# report. Editable for the same reason the image prompt is -- correcting
|
||||
# what a report will be called before it is filed is cheap, and renaming
|
||||
# one afterwards means finding it first.
|
||||
"report_write": "title",
|
||||
"job_stop": "id",
|
||||
# The thing being agreed to is what will be drawn, not which sampler draws
|
||||
# it. Also what makes the box on the card editable: a prompt corrected
|
||||
|
||||
@@ -32,9 +32,10 @@ from typing import Any
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.orm import Session as DBSession
|
||||
|
||||
from lembas.db.models import AUTHOR_MODEL, Chat, User
|
||||
from lembas.db.models import AUTHOR_MODEL, KIND_TASK, SOURCE_CHAT, Chat, User
|
||||
from lembas.db.session import session_scope
|
||||
from lembas.services import prompts as prompts_service
|
||||
from lembas.services import reports as reports_service
|
||||
from lembas.services import scratch as scratch_service
|
||||
from lembas.services import search as search_service
|
||||
from lembas.services import settings_store
|
||||
@@ -119,6 +120,16 @@ FAMILY_AGENT = "agent"
|
||||
# the whole cost of this one is somewhere else.
|
||||
FAMILY_IMAGE = "image"
|
||||
|
||||
# Filing a finished piece of work where the reader will find it later.
|
||||
# Deliberately not part of `notes`, and the line is the one a note already
|
||||
# draws from the other side: a note is something to be found again *by the
|
||||
# model*, searched for mid-conversation and edited when it turns out to be
|
||||
# wrong. A report is addressed to a person, read once, and never answered --
|
||||
# so it is the destination for work nobody was watching, which is exactly what
|
||||
# a note is not. Narrowing notes off must not take it away, and turning it on
|
||||
# must not hand out the notebook.
|
||||
FAMILY_REPORT = "report"
|
||||
|
||||
# The built-in families, in the order they are offered.
|
||||
FAMILIES = (
|
||||
FAMILY_SEARCH,
|
||||
@@ -130,6 +141,7 @@ FAMILIES = (
|
||||
FAMILY_SCRATCH,
|
||||
FAMILY_ASK,
|
||||
FAMILY_IMAGE,
|
||||
FAMILY_REPORT,
|
||||
FAMILY_AGENT,
|
||||
)
|
||||
|
||||
@@ -670,6 +682,93 @@ async def _run_memory_forget(context: ToolContext, args: dict[str, Any]) -> Tool
|
||||
)
|
||||
|
||||
|
||||
# --- Reports -----------------------------------------------------------------
|
||||
async def _run_report_write(context: ToolContext, args: dict[str, Any]) -> ToolOutcome:
|
||||
title = str(args.get("title") or "").strip()
|
||||
body = str(args.get("body") or "").strip()
|
||||
summary = str(args.get("summary") or "").strip()
|
||||
if not body:
|
||||
return ToolOutcome(
|
||||
"A report needs a body. Write what you found, not a note saying you found it.",
|
||||
{"name": "report_write", "status": "error", "error": "Empty body."},
|
||||
)
|
||||
with session_scope() as db:
|
||||
user = db.get(User, context.owner_id)
|
||||
if user is None:
|
||||
return ToolOutcome(
|
||||
"That report could not be filed.",
|
||||
{"name": "report_write", "status": "error", "error": "No such owner."},
|
||||
)
|
||||
report = reports_service.create(
|
||||
db,
|
||||
owner=user,
|
||||
title=title,
|
||||
body=body,
|
||||
summary=summary,
|
||||
source=SOURCE_CHAT,
|
||||
source_id=context.chat_id or "",
|
||||
model_id=context.model_id or "",
|
||||
)
|
||||
return ToolOutcome(
|
||||
f"Filed report {report.id} — {report.title!r}. "
|
||||
"The reader will find it under Reports; they cannot reply to it there.",
|
||||
{
|
||||
"name": "report_write",
|
||||
"query": report.title,
|
||||
"status": "ok",
|
||||
"results": [{"title": report.title, "id": report.id}],
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
async def _run_report_search(context: ToolContext, args: dict[str, Any]) -> ToolOutcome:
|
||||
query = str(args.get("query") or "").strip()
|
||||
with session_scope() as db:
|
||||
user = db.get(User, context.owner_id)
|
||||
found = (
|
||||
reports_service.search(db, user, query, limit=8)
|
||||
if query
|
||||
else reports_service.recent(db, user, limit=8)
|
||||
)
|
||||
event = {
|
||||
"name": "report_search",
|
||||
"query": query,
|
||||
"status": "ok",
|
||||
"results": [{"title": r.title, "id": r.id} for r in found],
|
||||
}
|
||||
if not found:
|
||||
return ToolOutcome("There are no reports matching that.", event)
|
||||
lines = ["Reports:"]
|
||||
for report in found:
|
||||
when = report.created_at.strftime("%Y-%m-%d %H:%M")
|
||||
lines.append(
|
||||
f"\n[{report.id}] {when} — {report.title}\n{reports_service.snippet(report)}"
|
||||
)
|
||||
lines.append("\nUse report_get with an id to read one in full.")
|
||||
return ToolOutcome("\n".join(lines), event)
|
||||
|
||||
|
||||
async def _run_report_get(context: ToolContext, args: dict[str, Any]) -> ToolOutcome:
|
||||
with session_scope() as db:
|
||||
user = db.get(User, context.owner_id)
|
||||
report = reports_service.get(db, str(args.get("id") or ""), user)
|
||||
if report is None:
|
||||
return ToolOutcome(
|
||||
"There is no such report.",
|
||||
{"name": "report_get", "status": "error", "error": "Not found."},
|
||||
)
|
||||
when = report.created_at.strftime("%Y-%m-%d %H:%M")
|
||||
return ToolOutcome(
|
||||
f"{report.title}\nFiled {when}\n\n{report.body}",
|
||||
{
|
||||
"name": "report_get",
|
||||
"query": report.title,
|
||||
"status": "ok",
|
||||
"results": [{"title": report.title, "id": report.id}],
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
# --- Skills ------------------------------------------------------------------
|
||||
async def _run_skill_get(context: ToolContext, args: dict[str, Any]) -> ToolOutcome:
|
||||
name = str(args.get("name") or "").strip()
|
||||
@@ -971,6 +1070,73 @@ REGISTRY: dict[str, ToolDef] = {
|
||||
run=_run_memory_forget,
|
||||
risk=RISK_WRITE,
|
||||
),
|
||||
ToolDef(
|
||||
name="report_write",
|
||||
family=FAMILY_REPORT,
|
||||
description=(
|
||||
"File a report: a finished piece of work, written for the person "
|
||||
"to read later. Use this when you have been asked for one, and "
|
||||
"when you finish a long piece of work whose result is worth "
|
||||
"keeping — an investigation, a summary of what you found, an "
|
||||
"account of what you changed. A report is read on its own, away "
|
||||
"from this conversation and possibly long after it, and THE "
|
||||
"READER CANNOT REPLY TO IT. So write it whole: say what you were "
|
||||
"asked, what you found and what you conclude, and do not refer "
|
||||
"to 'the above' or ask a question at the end."
|
||||
),
|
||||
parameters=_object(
|
||||
{
|
||||
"title": {
|
||||
**_STRING,
|
||||
"description": (
|
||||
"One line naming what this is about, as it will appear "
|
||||
"in a list of dozens. 'Build failures this week', not "
|
||||
"'Report' or 'Results'."
|
||||
),
|
||||
},
|
||||
"body": {
|
||||
**_STRING,
|
||||
"description": (
|
||||
"The report itself, in Markdown. Headings and lists are "
|
||||
"rendered. This is the whole of what the reader gets, so "
|
||||
"it should stand on its own with no further context."
|
||||
),
|
||||
},
|
||||
"summary": {
|
||||
**_STRING,
|
||||
"description": (
|
||||
"One sentence for the list page, so the report can be "
|
||||
"triaged without opening it. Say the finding, not the "
|
||||
"subject: 'Three tests fail on ARM only', not 'About the "
|
||||
"test failures'. Omit it and the first line of the body "
|
||||
"is used instead."
|
||||
),
|
||||
},
|
||||
},
|
||||
["title", "body"],
|
||||
),
|
||||
run=_run_report_write,
|
||||
risk=RISK_WRITE,
|
||||
),
|
||||
ToolDef(
|
||||
name="report_search",
|
||||
family=FAMILY_REPORT,
|
||||
description=(
|
||||
"Search reports filed earlier, yours and the reader's. With no "
|
||||
"query, returns the most recent. Worth doing before writing a "
|
||||
"recurring report, so this week's can say what changed since last "
|
||||
"week's rather than repeating it."
|
||||
),
|
||||
parameters=_object({"query": _STRING}, []),
|
||||
run=_run_report_search,
|
||||
),
|
||||
ToolDef(
|
||||
name="report_get",
|
||||
family=FAMILY_REPORT,
|
||||
description="Read one report in full, by the id a search returned.",
|
||||
parameters=_object({"id": _STRING}, ["id"]),
|
||||
run=_run_report_get,
|
||||
),
|
||||
ToolDef(
|
||||
name="skill_get",
|
||||
family=FAMILY_SKILLS,
|
||||
@@ -1153,12 +1319,23 @@ def _family_allowed(
|
||||
# the shape `resolve_tools` already refuses for `skill_get` with an
|
||||
# empty library. `settings_store.images_ready` answers all three.
|
||||
return bool(allowed.get("tools.image") and images)
|
||||
if gate in (FAMILY_CUSTOM, FAMILY_MCP, FAMILY_ASK, FAMILY_AGENT, FAMILY_SCRATCH):
|
||||
if gate in (
|
||||
FAMILY_CUSTOM,
|
||||
FAMILY_MCP,
|
||||
FAMILY_ASK,
|
||||
FAMILY_AGENT,
|
||||
FAMILY_SCRATCH,
|
||||
FAMILY_REPORT,
|
||||
):
|
||||
# Deliberately without `library.use`: an HTTP endpoint an administrator
|
||||
# wrote has nothing to do with this person's own documents and notes,
|
||||
# and requiring the library permission for it would be a coincidence of
|
||||
# naming rather than a rule. The same goes for being asked a question,
|
||||
# and for a pad that belongs to this chat and goes nowhere else.
|
||||
# for a pad that belongs to this chat and goes nowhere else, and for
|
||||
# filing a report -- which is addressed to the reader rather than kept
|
||||
# for the model, and is the fallback destination for scheduled work, so
|
||||
# gating it behind the library would switch that off for anyone whose
|
||||
# instance does not use one.
|
||||
return bool(allowed.get(f"tools.{gate}"))
|
||||
return bool(allowed.get(f"tools.{gate}") and allowed.get("library.use"))
|
||||
|
||||
@@ -1293,6 +1470,16 @@ def resolve_tools(db: DBSession, chat: Chat, user: User | None) -> ToolSet:
|
||||
off = scoped_off(chat)
|
||||
empty_library = not skills_service.count_enabled(db, user, exclude=scoped_skills_off(chat))
|
||||
|
||||
# A scheduled task runs with nobody present, so `ask_user` cannot work here:
|
||||
# it pauses the reply and waits for a POST that will never come, until
|
||||
# `approval_timeout` expires -- a run that silently does nothing for fifteen
|
||||
# minutes and then gives up. Withdrawn from the offered set rather than
|
||||
# merely discouraged in `core.unattended`, because a rule living only in a
|
||||
# system message is one a page the model just read can argue with. The
|
||||
# fragment is the half that stops it *planning* around a tool it has not got.
|
||||
if chat is not None and chat.kind == KIND_TASK:
|
||||
off = off | {FAMILY_ASK}
|
||||
|
||||
return ToolSet(
|
||||
tuple(
|
||||
tool
|
||||
|
||||
@@ -0,0 +1,120 @@
|
||||
"""Starting a reply when nobody asked for one just now.
|
||||
|
||||
Two things need this: a background job finishing on the far side, and a schedule
|
||||
coming due. Both are the same problem — put a turn into a chat from outside any
|
||||
request, and get it answered — and both depend on the same invariant, which is
|
||||
subtle enough that having two copies of it is how one of them drifts.
|
||||
|
||||
**The invariant.** A chat may have one generation running at a time. So the
|
||||
check and the writes happen under a per-chat lock with **no `await` between
|
||||
them**: two things waking one chat at the same moment cannot each spin up a
|
||||
reply, because the second sees the first's already live and leaves its turn
|
||||
`queued` for that reply's `_inject`/`_drain` to deliver.
|
||||
|
||||
**The role is load-bearing.** The turn is written as `user`, because `_inject`
|
||||
sends a queued turn verbatim and `build_messages` must keep seeing a user turn
|
||||
where one belongs. What stops the transcript claiming the reader typed it is
|
||||
`Message.machine`, which changes the bubble and nothing about the request. The
|
||||
framing the *model* reads therefore has to live in the words -- which is why
|
||||
every caller passes prose that names itself, the way `execute_plan` quotes a
|
||||
plan rather than asserting it.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
_LOCKS: dict[str, asyncio.Lock] = {}
|
||||
|
||||
|
||||
def lock_for(chat_id: str) -> asyncio.Lock:
|
||||
"""The lock for one chat, made on first use.
|
||||
|
||||
Not cleaned up: a lock is two pointers, chats are finite, and a reaper for
|
||||
these would be able to delete one while a waiter held it.
|
||||
"""
|
||||
existing = _LOCKS.get(chat_id)
|
||||
if existing is None:
|
||||
existing = _LOCKS[chat_id] = asyncio.Lock()
|
||||
return existing
|
||||
|
||||
|
||||
async def wake_chat(chat_id: str, content: str, *, model_id: str = "") -> str:
|
||||
"""Put a turn into a chat and get it answered. Returns the assistant message
|
||||
id if a reply was started, or "" if the turn was queued for one already
|
||||
running.
|
||||
|
||||
Never raises: a caller here is a watcher or a ticker, and one chat that
|
||||
cannot be woken must not stop the others.
|
||||
|
||||
`model_id` overrides the chat's own, which is what lets a schedule name the
|
||||
model it wants without editing the chat. Empty means the chat decides.
|
||||
"""
|
||||
from lembas.db.models import ROLE_ASSISTANT, ROLE_USER, Chat
|
||||
from lembas.db.session import session_scope
|
||||
from lembas.services import chat as chat_service
|
||||
from lembas.services import generation as generation_service
|
||||
|
||||
assistant_id = ""
|
||||
async with lock_for(chat_id):
|
||||
# Read before the writes and with nothing awaited in between. This is
|
||||
# the whole invariant; moving either half out of the lock, or awaiting
|
||||
# anything between them, reintroduces two concurrent replies to one
|
||||
# chat -- which is a Stop button pointing at whichever bubble happens to
|
||||
# come first in the document.
|
||||
running = generation_service.running_for(chat_id) is not None
|
||||
try:
|
||||
with session_scope() as db:
|
||||
chat = db.get(Chat, chat_id)
|
||||
if chat is None:
|
||||
return ""
|
||||
chat_service.create_message(
|
||||
db, chat, ROLE_USER, content, queued=running, machine=True
|
||||
)
|
||||
if not running:
|
||||
assistant = chat_service.create_message(
|
||||
db,
|
||||
chat,
|
||||
ROLE_ASSISTANT,
|
||||
"",
|
||||
complete_=False,
|
||||
model_id=model_id or chat.model_id,
|
||||
)
|
||||
assistant_id = assistant.id
|
||||
except Exception: # noqa: BLE001 - a failed wake must not kill its caller
|
||||
log.exception("could not wake chat %s", chat_id)
|
||||
return ""
|
||||
|
||||
# Outside the lock: `ensure` spawns a task that will want a session of its
|
||||
# own, and holding this while it starts would serialise every wake on the
|
||||
# instance behind the slowest one.
|
||||
if assistant_id:
|
||||
generation_service.ensure(chat_id, assistant_id)
|
||||
return assistant_id
|
||||
|
||||
|
||||
def queued_count(chat_id: str) -> int:
|
||||
"""How many turns are waiting to be delivered into this chat.
|
||||
|
||||
`_drain` takes **one** per reply, deliberately -- two consecutive user turns
|
||||
in a request is worse than a delay. So something firing repeatedly into a
|
||||
chat that is answering slowly can build a backlog nobody asked for, and the
|
||||
callers here use this to stop rather than to queue for ever.
|
||||
"""
|
||||
from sqlalchemy import func, select
|
||||
|
||||
from lembas.db.models import Message
|
||||
from lembas.db.session import session_scope
|
||||
|
||||
with session_scope() as db:
|
||||
return int(
|
||||
db.scalar(
|
||||
select(func.count())
|
||||
.select_from(Message)
|
||||
.where(Message.chat_id == chat_id, Message.queued.is_(True))
|
||||
)
|
||||
or 0
|
||||
)
|
||||
Reference in New Issue
Block a user