+ Finished work, filed to be read later. A model writes one when you ask for it
+ or when it finishes something worth keeping, and anything running on a
+ schedule leaves its result here. Nothing on this page can be replied to.
+
+
+
+ Work that runs on its own, whether or not you are here. Each one has its own
+ chat, and replies into it every time it comes round.
+
+
+{% if not schedules %}
+
Install as an app
diff --git a/tests/conftest.py b/tests/conftest.py
index 96b4d77..7570d91 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -102,6 +102,36 @@ def fresh_terminal_registry() -> Iterator[None]:
_clear()
+@pytest.fixture(autouse=True)
+def fresh_schedule_ticker() -> Iterator[None]:
+ """Stop the schedule ticker and forget any firings, for the same reason.
+
+ The ticker is a module-level task like the terminal reaper, and a firing is
+ a task holding a chat id. One left running would wake up inside the next
+ test's event loop, against the next test's database, and fire something
+ nobody in that test has ever heard of.
+
+ The wake locks go too: they are keyed on chat id, and `make_chat` recycles
+ ids freely across a session.
+ """
+ from lembas.services import wake as wake_service
+ from lembas.services.schedule import ticker as ticker_service
+
+ def _clear() -> None:
+ running = ticker_service._TICKER
+ if running is not None:
+ running.cancel()
+ ticker_service._TICKER = None
+ for task in list(ticker_service._FIRING):
+ task.cancel()
+ ticker_service._FIRING.clear()
+ wake_service._LOCKS.clear()
+
+ _clear()
+ yield
+ _clear()
+
+
@pytest.fixture(autouse=True)
def fresh_project_index() -> Iterator[None]:
"""Empty the directory-listing cache between tests, for the third time.
diff --git a/tests/test_agent_policy.py b/tests/test_agent_policy.py
index 77f0d08..09da0ea 100644
--- a/tests/test_agent_policy.py
+++ b/tests/test_agent_policy.py
@@ -266,6 +266,10 @@ def test_the_builtins_that_change_things_say_so():
"memory_forget",
"skill_create",
"skill_edit",
+ # Filing a report writes a durable artefact of the reader's, the same
+ # class as a note. Plan mode meaning "look but do not touch" has to mean
+ # this too, even though what it touches is a page rather than a machine.
+ "report_write",
}
diff --git a/tests/test_chat.py b/tests/test_chat.py
index b994f57..0310951 100644
--- a/tests/test_chat.py
+++ b/tests/test_chat.py
@@ -836,7 +836,12 @@ def test_the_unread_poll_reports_dots(client: TestClient, db, registered, make_c
response = client.get("/api/chats/unread")
assert f'id="unread-{chat_id}"' in response.text
- assert "hidden" not in response.text
+ # This chat's own span, not the whole body: the response also carries the
+ # section dots, and one of those being hidden is right rather than wrong.
+ chat_dot = next(
+ span for span in response.text.split(" User:
+ return db.scalars(select(User).order_by(User.created_at)).first()
+
+
+def _fill(db, chat: Chat, count: int, *, day: int = 1) -> list[Message]:
+ """`count` alternating turns, a minute apart so the order is unambiguous."""
+ start = datetime(2026, 1, day, tzinfo=UTC)
+ rows = []
+ for index in range(count):
+ rows.append(
+ Message(
+ chat_id=chat.id,
+ role="user" if index % 2 == 0 else "assistant",
+ content=f"turn {index}",
+ created_at=start + timedelta(minutes=index),
+ complete=True,
+ )
+ )
+ db.add_all(rows)
+ db.commit()
+ return rows
+
+
+# --- The conversation itself ------------------------------------------------------
+def test_there_is_exactly_one_per_person(client: TestClient, db, registered):
+ """Get-or-create, so a schedule can post here before anybody has opened the
+ page — the second deliberate exception to "chats are created lazily"."""
+ first = messages_service.for_user(db, _user(db))
+ second = messages_service.for_user(db, _user(db))
+
+ assert first.id == second.id
+ assert first.kind == KIND_MESSAGES
+ assert len(db.scalars(select(Chat).where(Chat.kind == KIND_MESSAGES)).all()) == 1
+
+
+def test_it_is_not_in_the_chat_tree(client: TestClient, db, registered):
+ """It has a section of its own. This is the kind-leakage trap again, from
+ the other side."""
+ from lembas.api.pages import sidebar_context
+
+ conversation = messages_service.for_user(db, _user(db))
+ listed = {c.id for c in sidebar_context(db, _user(db))["unfiled_chats"]}
+
+ assert conversation.id not in listed
+
+
+# --- What reaches the model -------------------------------------------------------
+def test_the_request_does_not_grow_with_the_conversation(
+ client: TestClient, db, registered
+):
+ """The whole point. A conversation meant to run for years cannot all be
+ sent, and a request that grows until the endpoint refuses it is the failure
+ nobody sees coming — there is nothing wrong on screen right up until it
+ stops working.
+ """
+ from lembas.services import chat as chat_service
+
+ conversation = messages_service.for_user(db, _user(db))
+ _fill(db, conversation, 10)
+ short = chat_service.build_messages(db, conversation)
+
+ _fill(db, conversation, 300)
+ long = chat_service.build_messages(db, conversation)
+
+ assert len(short) == 10
+ assert len(long) == messages_service.LIVE_CHUNK
+ assert len(long) < len(short) + 300
+
+
+def test_it_is_the_most_recent_turns_that_are_sent(client: TestClient, db, registered):
+ from lembas.services import chat as chat_service
+
+ conversation = messages_service.for_user(db, _user(db))
+ _fill(db, conversation, messages_service.LIVE_CHUNK + 20)
+
+ payload = chat_service.build_messages(db, conversation)
+ bodies = [entry["content"] for entry in payload]
+
+ assert bodies[-1] == f"turn {messages_service.LIVE_CHUNK + 19}"
+ assert "turn 0" not in bodies
+
+
+def test_an_ordinary_chat_still_sends_everything(client: TestClient, db, registered):
+ """The bound is one branch on one kind. A chat is not silently truncated."""
+ from lembas.services import chat as chat_service
+
+ ordinary = Chat(user_id=_user(db).id, model_id="m")
+ db.add(ordinary)
+ db.commit()
+ _fill(db, ordinary, messages_service.LIVE_CHUNK + 20)
+
+ assert len(chat_service.build_messages(db, ordinary)) == messages_service.LIVE_CHUNK + 20
+
+
+def test_compaction_never_fires_on_it(client: TestClient, db, registered):
+ """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."""
+ from lembas.services import compaction as compaction_service
+
+ conversation = messages_service.for_user(db, _user(db))
+ _fill(db, conversation, 200)
+
+ assert compaction_service.should_compact(db, conversation) is False
+
+
+# --- Nothing is lost ---------------------------------------------------------------
+def test_every_turn_is_kept_however_old(client: TestClient, db, registered):
+ """Bounded in the request, unbounded on disk. Deliberately not folded into
+ text: the visible conversation would be identical either way, so the only
+ thing destroying the rows would buy is disk — against irreversibility."""
+ conversation = messages_service.for_user(db, _user(db))
+ _fill(db, conversation, 250)
+
+ assert messages_service.count(db, conversation) == 250
+
+
+# --- Reading backwards --------------------------------------------------------------
+def test_the_page_opens_on_the_latest_chunk(client: TestClient, db, registered):
+ conversation = messages_service.for_user(db, _user(db))
+ _fill(db, conversation, 120)
+
+ body = client.get("/messages").text
+
+ assert "turn 119" in body
+ assert "turn 0" not in body
+ assert "history-sentinel" in body
+
+
+def test_a_short_conversation_has_no_sentinel(client: TestClient, db, registered):
+ conversation = messages_service.for_user(db, _user(db))
+ _fill(db, conversation, 5)
+
+ assert "history-sentinel" not in client.get("/messages").text
+
+
+def test_scrolling_up_returns_the_page_before(client: TestClient, db, registered):
+ conversation = messages_service.for_user(db, _user(db))
+ rows = _fill(db, conversation, 200)
+ oldest_shown = rows[-messages_service.LIVE_CHUNK]
+
+ response = client.get(f"/api/messages/history?before={oldest_shown.id}")
+
+ assert response.status_code == 200
+ assert f"turn {200 - messages_service.LIVE_CHUNK - 1}" in response.text
+ # The turn it was asked to go before is not repeated.
+ assert f">turn {200 - messages_service.LIVE_CHUNK}<" not in response.text
+
+
+def test_a_cursor_it_cannot_place_is_answered_with_204(
+ client: TestClient, db, registered
+):
+ """Never with "the oldest page": that would prepend a block the reader is
+ already looking at, and a duplicated transcript is something only a reload
+ can reconcile."""
+ conversation = messages_service.for_user(db, _user(db))
+ _fill(db, conversation, 100)
+
+ other = Chat(user_id=_user(db).id, model_id="m")
+ db.add(other)
+ db.commit()
+ stray = Message(chat_id=other.id, role="user", content="elsewhere")
+ db.add(stray)
+ db.commit()
+
+ assert client.get("/api/messages/history").status_code == 204
+ assert client.get("/api/messages/history?before=nope").status_code == 204
+ assert client.get(f"/api/messages/history?before={stray.id}").status_code == 204
+
+
+def test_the_oldest_page_stops_rather_than_looping(client: TestClient, db, registered):
+ conversation = messages_service.for_user(db, _user(db))
+ rows = _fill(db, conversation, 3)
+
+ response = client.get(f"/api/messages/history?before={rows[0].id}")
+ assert response.status_code == 204
+
+
+def test_two_turns_sharing_a_timestamp_are_each_returned_once(
+ client: TestClient, db, registered
+):
+ """The `id` tie-breaker. 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."""
+ conversation = messages_service.for_user(db, _user(db))
+ stamp = datetime(2026, 1, 1, tzinfo=UTC)
+ twins = [
+ Message(chat_id=conversation.id, role="user", content=f"same {i}", created_at=stamp)
+ for i in range(2)
+ ]
+ db.add_all(twins)
+ db.commit()
+ # A later day, so the cursor is unambiguously after both twins -- otherwise
+ # the cursor shares their stamp and the test is about id ordering, which is
+ # random, rather than about the tie-breaker.
+ later = _fill(db, conversation, 2, day=2)
+
+ page = messages_service.older_than(db, conversation, later[0])
+ assert {m.content for m in page} == {"same 0", "same 1"}
+
+
+def test_the_sentinel_names_its_own_target(client: TestClient, db, registered):
+ """It sits on a page whose composer form carries `hx-target="#thread"`, and
+ htmx resolves that by walking up the DOM. The jobs chip demonstrated once
+ what an unstated target does to a transcript."""
+ conversation = messages_service.for_user(db, _user(db))
+ _fill(db, conversation, 120)
+ body = client.get("/messages").text
+
+ sentinel = next(chunk for chunk in body.split(" User:
+ return db.scalars(select(User).order_by(User.created_at)).first()
+
+
+def _file(db, **kwargs) -> Report:
+ from lembas.services import reports as reports_service
+
+ fields = {"title": "A report", "body": "What I found.", **kwargs}
+ return reports_service.create(db, owner=_user(db), **fields)
+
+
+# --- The store ----------------------------------------------------------------
+def test_a_report_is_filed_and_read_back(client: TestClient, db, registered):
+ from lembas.services import reports as reports_service
+
+ report = _file(db, title="Build failures", body="# Heading\n\nThree tests fail.")
+ assert reports_service.get(db, report.id, _user(db)) is report
+
+
+def test_a_missing_summary_falls_back_to_the_first_real_line(client: TestClient, db, registered):
+ """A list of forty reports all reading '# ' is a bug somebody has to explain."""
+ report = _file(db, body="# Weekly build report\n\nThree tests fail on ARM.", summary="")
+ assert report.summary == "Weekly build report"
+
+
+def test_a_report_belongs_to_its_owner_alone(client: TestClient, db, registered):
+ """There is no sharing here, which is a decision rather than an omission --
+ so a second account must not reach the first one's reports."""
+ from lembas.security.passwords import hash_password
+ from lembas.services import reports as reports_service
+
+ report = _file(db)
+ stranger = User(
+ email="sam@shire.test", name="Sam", password_hash=hash_password("gardening-is-hard")
+ )
+ db.add(stranger)
+ db.commit()
+
+ assert reports_service.get(db, report.id, stranger) is None
+ assert reports_service.recent(db, stranger) == []
+
+
+def test_an_over_long_body_is_trimmed_rather_than_refused(client: TestClient, db, registered):
+ """The rule `memories` already follows: file it short with everything else
+ intact, instead of failing the turn that wrote it."""
+ from lembas.services import reports as reports_service
+
+ report = _file(db, body="x" * (reports_service.MAX_BODY_CHARS + 500))
+ assert len(report.body) == reports_service.MAX_BODY_CHARS
+
+
+# --- The tool -----------------------------------------------------------------
+@pytest.mark.anyio
+async def test_report_write_actually_runs(client: TestClient, db, registered):
+ """Not that it is declared -- that it *runs*.
+
+ `_run_scratch_write` read a field its own dataclass did not have and raised
+ AttributeError for the whole life of the feature, swallowed by `run_tool`'s
+ blanket except into a message that reads exactly like a model calling it
+ wrongly. The test that existed asserted the family and the risk, which are
+ properties of the declaration.
+ """
+ from lembas.services import tools as tools_service
+
+ context = tools_service.ToolContext(owner_id=_user(db).id, chat_id="", model_id="test-model")
+ outcome = await tools_service.run_tool(
+ context, "report_write", '{"title": "Findings", "body": "Three tests fail."}'
+ )
+
+ assert outcome.event["status"] == "ok", outcome.content
+ filed = db.scalars(select(Report)).all()
+ assert [r.title for r in filed] == ["Findings"]
+ assert filed[0].model_id == "test-model"
+
+
+@pytest.mark.anyio
+async def test_report_write_refuses_an_empty_body(client: TestClient, db, registered):
+ from lembas.services import tools as tools_service
+
+ context = tools_service.ToolContext(owner_id=_user(db).id, chat_id="")
+ outcome = await tools_service.run_tool(context, "report_write", '{"title": "Nothing"}')
+
+ assert outcome.event["status"] == "error"
+ assert db.scalars(select(Report)).all() == []
+
+
+@pytest.mark.anyio
+async def test_report_search_finds_by_body_word(client: TestClient, db, registered):
+ """FTS tables are outside the model-driven schema sync, so this is also the
+ check that `reports_fts` was actually created and its triggers fire."""
+ from lembas.services import tools as tools_service
+
+ _file(db, title="Unrelated", body="Nothing about the thing.")
+ _file(db, title="The one", body="A palantir was involved.")
+
+ context = tools_service.ToolContext(owner_id=_user(db).id, chat_id="")
+ outcome = await tools_service.run_tool(context, "report_search", '{"query": "palantir"}')
+
+ assert [r["title"] for r in outcome.event["results"]] == ["The one"]
+
+
+def test_the_index_is_backfilled_on_an_upgrade(client: TestClient, db, registered):
+ """The upgrade path, which the fresh-database case does not exercise.
+
+ FTS tables are outside the model-driven schema sync -- they are not
+ SQLAlchemy models, so `sync_schema` cannot diff them. On an instance that
+ already has reports, `ensure_fts` has to create the index *and* backfill
+ what is already in the table; without the backfill every report filed before
+ the upgrade is invisible to search forever, and nothing says so.
+ """
+ from sqlalchemy import text
+
+ from lembas.db.migrations import ensure_fts
+ from lembas.db.session import get_engine
+ from lembas.services import reports as reports_service
+
+ _file(db, title="Before the upgrade", body="A palantir was involved.")
+
+ # Drop the index and its triggers, leaving the rows: an instance upgrading
+ # into this release looks exactly like this.
+ with get_engine().begin() as connection:
+ for suffix in ("_ai", "_ad", "_au"):
+ connection.execute(text(f"DROP TRIGGER IF EXISTS reports_fts{suffix}"))
+ connection.execute(text("DROP TABLE IF EXISTS reports_fts"))
+
+ assert "reports_fts" in ensure_fts(get_engine())
+ assert [r.title for r in reports_service.search(db, _user(db), "palantir")] == [
+ "Before the upgrade"
+ ]
+
+
+# --- The section's character --------------------------------------------------
+def test_the_reports_pages_carry_no_composer(client: TestClient, db, registered):
+ """Reports are read, never answered, and the way to be sure of that is for
+ the machinery that would answer to be absent.
+
+ `chat/_message.html` is the state machine: an `sse-connect` anywhere on
+ these pages is a generation this section has no business starting.
+ """
+ report = _file(db)
+
+ for url in ("/reports", f"/reports/{report.id}"):
+ body = client.get(url).text
+ assert "sse-connect" not in body, url
+ assert "composer__form" not in body, url
+ # Nothing on the page can send anything anywhere. Checked as "no field
+ # named content" rather than by looking for a URL, because the sidebar
+ # legitimately links to sections that do have composers.
+ assert 'name="content"' not in body, url
+
+
+def test_no_route_accepts_a_message_for_a_report(client: TestClient, registered):
+ """Asserted on the resolved routes rather than on the templates, because the
+ failure this guards against is somebody adding the endpoint first."""
+ from lembas.main import app
+
+ # Read from the OpenAPI schema rather than by walking `app.routes`: this
+ # FastAPI keeps an included router wrapped rather than flattening it, so the
+ # walk finds nothing at all and the assertion passes for the wrong reason.
+ writable = {
+ f"{method.upper()} {path}"
+ for path, methods in app.openapi()["paths"].items()
+ if path.startswith(("/reports", "/api/reports"))
+ for method in methods
+ if method.upper() in {"POST", "PATCH", "PUT"}
+ }
+ assert writable == {"POST /api/reports/{report_id}/delete"}
+
+
+def test_a_report_body_cannot_smuggle_html(client: TestClient, db, registered):
+ """Model output, hard rule 6. Rendered through services/markdown.py, which
+ is the one path allowed to emit HTML here."""
+ report = _file(
+ db,
+ body="\n\n[click](javascript:alert(1))",
+ )
+ body = client.get(f"/reports/{report.id}").text
+
+ assert "" not in body
+ # The link is left as inert text rather than becoming an anchor, which is
+ # the property that matters: what must never appear is the href.
+ assert 'href="javascript:' not in body
+
+
+def test_opening_a_report_clears_its_dot(client: TestClient, db, registered):
+ report = _file(db)
+ assert report.unread is True
+
+ client.get(f"/reports/{report.id}")
+ db.refresh(report)
+ assert report.unread is False
+
+
+def test_the_unread_poll_carries_the_section_dot(client: TestClient, db, registered):
+ """Sent on every tick including empty, because it has to be able to clear.
+ A dot that survived reading the last report would be news nobody can dismiss.
+ """
+ report = _file(db)
+
+ showing = client.get("/api/chats/unread").text
+ dot = next(s for s in showing.split(" User:
+ return db.scalars(select(User).order_by(User.created_at)).first()
+
+
+def _endpoint() -> Endpoint:
+ return Endpoint(base_url="http://127.0.0.1:1/v1", api_key="", extra_headers={})
+
+
+def _template(db) -> str:
+ return prompts_service.resolve(db, "task.schedule_compile")
+
+
+async def _compile(db, reply: str, monkeypatch, request: str = "every monday at 3"):
+ async def answer(endpoint, payload):
+ return reply
+
+ monkeypatch.setattr(compile_service, "complete", answer)
+ return await compile_service.compile_request(
+ _endpoint(), "test-model", request, template=_template(db), user=_user(db)
+ )
+
+
+# --- What it can read -------------------------------------------------------------
+@pytest.mark.anyio
+async def test_a_plain_json_answer_compiles(client: TestClient, db, registered, monkeypatch):
+ compiled = await _compile(
+ db,
+ json.dumps(
+ {
+ "title": "Build check",
+ "instruction": "Check the build and say what broke.",
+ "target": "report",
+ "schedule": {"at": {"weekdays": [0], "times": ["15:00"]}},
+ }
+ ),
+ monkeypatch,
+ )
+
+ assert compiled.ok is True
+ assert compiled.title == "Build check"
+ assert compiled.target == "report"
+ assert compiled.rule["at"]["times"] == ["15:00"]
+
+
+@pytest.mark.anyio
+async def test_a_fenced_answer_compiles(client: TestClient, db, registered, monkeypatch):
+ """Small models fence their JSON however they were trained to. Refusing it
+ costs a whole round trip to end up showing the manual form anyway — the
+ same reasoning `tools.parse_arguments` already follows."""
+ body = json.dumps(
+ {"title": "T", "instruction": "I", "schedule": {"at": {"times": ["09:00"]}}}
+ )
+ compiled = await _compile(
+ db, f"Here you go:\n```json\n{body}\n```\nHope that helps!", monkeypatch
+ )
+
+ assert compiled.ok is True
+ assert compiled.rule["at"]["times"] == ["09:00"]
+
+
+@pytest.mark.anyio
+async def test_a_timer_with_no_start_begins_now(
+ client: TestClient, db, registered, monkeypatch
+):
+ """"Every six hours" is written `{"every": {"hours": 6}}` and nothing else,
+ which is the natural reading and cannot fire on its own — a timer measures
+ from a start, and `rule.py` has no clock to invent one. Filled in here
+ exactly as the manual form does, or the commonest request of all compiles to
+ a schedule that never runs."""
+ compiled = await _compile(
+ db,
+ json.dumps({"title": "T", "instruction": "I", "schedule": {"every": {"hours": 6}}}),
+ monkeypatch,
+ request="every six hours",
+ )
+
+ assert compiled.ok is True
+ assert compiled.rule["every"] == {"minutes": 360}
+ assert compiled.rule["start"]
+
+
+@pytest.mark.anyio
+async def test_thinking_is_stripped_before_parsing(
+ client: TestClient, db, registered, monkeypatch
+):
+ """A model that thinks inline puts its reasoning in `content`, which is the
+ field `complete` hands back verbatim — the trap auto-titling hit."""
+ body = json.dumps({"title": "T", "instruction": "I", "schedule": {"at": {"times": ["09:00"]}}})
+ compiled = await _compile(db, f"Let me work this out…{body}", monkeypatch)
+
+ assert compiled.ok is True
+ assert compiled.title == "T"
+
+
+# --- How it fails ------------------------------------------------------------------
+@pytest.mark.anyio
+async def test_prose_falls_back_to_the_readers_own_words(
+ client: TestClient, db, registered, monkeypatch
+):
+ """Never a schedule nobody asked for. The reader's words survive so the form
+ is filled in rather than blank."""
+ compiled = await _compile(db, "Sure! I'd suggest running that weekly.", monkeypatch)
+
+ assert compiled.ok is False
+ assert compiled.instruction == "every monday at 3"
+ assert compiled.reason
+
+
+@pytest.mark.anyio
+async def test_a_rule_that_normalises_to_nothing_is_not_ok(
+ client: TestClient, db, registered, monkeypatch
+):
+ """The compile's output is model output that becomes a *timer*, and this is
+ the reason `rule.validate` had to be total."""
+ compiled = await _compile(
+ db, json.dumps({"title": "T", "instruction": "I", "schedule": "0 3 * * 1"}), monkeypatch
+ )
+
+ assert compiled.ok is False
+ assert compiled.rule == {}
+ assert "when" in compiled.reason
+
+
+@pytest.mark.anyio
+async def test_a_time_already_past_is_not_ok(client: TestClient, db, registered, monkeypatch):
+ compiled = await _compile(
+ db,
+ json.dumps(
+ {"title": "T", "instruction": "I", "schedule": {"start": "2020-01-01T09:00:00Z"}}
+ ),
+ monkeypatch,
+ )
+
+ assert compiled.ok is False
+ assert "already passed" in compiled.reason
+
+
+@pytest.mark.anyio
+async def test_an_endpoint_that_is_down_is_not_an_error(
+ client: TestClient, db, registered, monkeypatch
+):
+ async def refuse(endpoint, payload):
+ raise LLMError("connection refused")
+
+ monkeypatch.setattr(compile_service, "complete", refuse)
+ compiled = await compile_service.compile_request(
+ _endpoint(), "test-model", "daily at nine", template=_template(db), user=_user(db)
+ )
+
+ assert compiled.ok is False
+ assert compiled.instruction == "daily at nine"
+
+
+@pytest.mark.anyio
+async def test_clearing_the_fragment_switches_off_the_compiling_not_the_feature(
+ client: TestClient, db, registered, monkeypatch
+):
+ """`task.compact` set the precedent that clearing a fragment kills a
+ feature. Here it must not: the manual form is what makes "an empty override
+ means off" safe, and no request is made at all."""
+ called = False
+
+ async def answer(endpoint, payload):
+ nonlocal called
+ called = True
+ return "{}"
+
+ monkeypatch.setattr(compile_service, "complete", answer)
+ compiled = await compile_service.compile_request(
+ _endpoint(), "test-model", "daily at nine", template="", user=_user(db)
+ )
+
+ assert called is False
+ assert compiled.ok is False
+ assert compiled.instruction == "daily at nine"
+ assert compiled.reason == ""
+
+
+def test_the_prompt_carries_the_readers_zone(client: TestClient, db, registered):
+ """The model works out "Monday at 3" and the ticker fires it. If they
+ disagree about the zone, nothing errors — it simply runs at the wrong time."""
+ user = _user(db)
+ user.settings_json = {**(user.settings_json or {}), "timezone": "Asia/Tokyo"}
+ db.commit()
+
+ prompt = compile_service.render_prompt(
+ _template(db), request="every monday at 3", user=user
+ )
+ assert "Asia/Tokyo" in prompt
+ assert "every monday at 3" in prompt
+
+
+# --- The review step ----------------------------------------------------------------
+def test_describing_shows_it_back_rather_than_creating_it(
+ client: TestClient, db, registered, monkeypatch
+):
+ """A timing a model chose and nobody looked at is exactly the standing
+ instruction this codebase refuses to create silently elsewhere."""
+
+ async def answer(endpoint, payload):
+ return json.dumps(
+ {
+ "title": "Build check",
+ "instruction": "Check the build.",
+ "schedule": {"at": {"weekdays": [0], "times": ["15:00"]}},
+ }
+ )
+
+ monkeypatch.setattr(compile_service, "complete", answer)
+
+ response = client.post("/api/schedules/describe", data={"request": "mondays at 3"})
+
+ assert response.status_code == 200
+ assert "Every Monday at 15:00" in response.text
+ assert "Build check" in response.text
+ # Shown, not saved.
+ assert db.scalars(select(Schedule)).all() == []
+
+
+def test_describing_with_no_model_configured_still_answers(
+ client: TestClient, db, registered
+):
+ from lembas.db.models import Connection
+
+ for connection in db.scalars(select(Connection)):
+ db.delete(connection)
+ db.commit()
+
+ response = client.post("/api/schedules/describe", data={"request": "mondays at 3"})
+
+ assert response.status_code == 200
+ assert "fill it in yourself" in response.text
+
+
+# --- Unattended ---------------------------------------------------------------------
+def test_ask_user_is_not_offered_in_a_task_chat(client: TestClient, db, registered):
+ """Enforced in `resolve_tools`, not merely discouraged in the prompt.
+
+ A parked `ask_user` holds the reply for the whole `approval_timeout` with
+ nobody there to answer — a run that silently does nothing for fifteen
+ minutes and then gives up. A rule living only in a system message is one a
+ page the model just read can argue with.
+ """
+ from lembas.db.models import Model
+ from lembas.services import tools as tools_service
+
+ model = db.scalars(select(Model)).one()
+ model.capabilities_json = {"tools": True, "tool_ask": True}
+ db.commit()
+ settings_store.update(db, {"default_permissions": {"tools.ask": True}})
+
+ ordinary = Chat(user_id=_user(db).id, model_id="test-model")
+ task = Chat(user_id=_user(db).id, model_id="test-model", kind=KIND_TASK)
+ db.add_all([ordinary, task])
+ db.commit()
+
+ offered = {t.name for t in tools_service.resolve_tools(db, ordinary, _user(db)).defs}
+ assert "ask_user" in offered
+
+ withdrawn = {t.name for t in tools_service.resolve_tools(db, task, _user(db)).defs}
+ assert "ask_user" not in withdrawn
+
+
+def test_a_task_chat_is_told_what_it_is_for(client: TestClient, db, registered, monkeypatch):
+ """A task chat accumulates every run, so by the tenth the instruction is far
+ out of sight up the transcript."""
+ from lembas.services import harness
+ from lembas.services import schedules as schedules_service
+
+ schedule = schedules_service.create(
+ db,
+ owner=_user(db),
+ title="Build check",
+ instruction="Check the build and say what broke.",
+ rule={"at": {"weekdays": [0], "times": ["15:00"]}},
+ )
+ chat = db.get(Chat, schedule.chat_id)
+
+ values = harness.context_variables(db, _user(db), [], chat)
+ assert values["schedule_instruction"] == "Check the build and say what broke."
+ assert values["schedule_summary"] == "Every Monday at 15:00"
+
+ block = harness.compose(db, _user(db), [], chat)
+ assert "nobody is necessarily reading it" in block
+ assert "Check the build and say what broke." in block
+
+
+def test_an_ordinary_chat_is_told_none_of_it(client: TestClient, db, registered):
+ """`core.unattended` and `context.schedule` are gated on the same variable,
+ so the warning cannot appear without the thing it warns about."""
+ from lembas.services import harness
+
+ chat = Chat(user_id=_user(db).id, model_id="test-model")
+ db.add(chat)
+ db.commit()
+
+ block = harness.compose(db, _user(db), [], chat)
+ assert "nobody is necessarily reading" not in block
+ assert "This scheduled task" not in block
diff --git a/tests/test_schedule_rule.py b/tests/test_schedule_rule.py
new file mode 100644
index 0000000..94d74d1
--- /dev/null
+++ b/tests/test_schedule_rule.py
@@ -0,0 +1,373 @@
+"""The recurrence rule, on its own.
+
+`rule.py` is pure and total: no session, no wall clock, nothing that raises. So
+it is tested exhaustively here, before anything calls it — which is the whole
+reason it was built first. Everything downstream of it fails quietly. A schedule
+that never fires looks exactly like a working one on every screen it appears on,
+and one that fires an hour out looks like nothing at all until the report is
+late.
+
+The DST cases are the ones worth reading. They are not hypotheticals: each of
+them happens twice a year, on a machine nobody is watching.
+"""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime, timedelta
+from zoneinfo import ZoneInfo
+
+from lembas.services.schedule import rule as rule_service
+
+# A zone with an interesting spring and autumn, and one without.
+PRAGUE = ZoneInfo("Europe/Prague")
+UTC_ZONE = UTC
+
+
+def at(text: str) -> datetime:
+ """A UTC instant from an ISO string, for readable expectations."""
+ return datetime.fromisoformat(text).replace(tzinfo=UTC)
+
+
+def local(text: str, zone=PRAGUE) -> datetime:
+ """A wall-clock stamp in a zone, as an instant."""
+ return datetime.fromisoformat(text).replace(tzinfo=zone).astimezone(UTC)
+
+
+# --- validate: total, clamping, never raising ---------------------------------
+def test_validate_never_raises_on_anything():
+ """The compile step hands this whatever a model wrote. Model output that
+ becomes a *timer* is the sharpest case of hard rule 6 in the codebase."""
+ for junk in (
+ None, 0, "", "every monday", [], {}, {"every": "often"},
+ {"at": "3pm"}, {"count": "lots"}, {"start": "not a date"},
+ {"every": {"minutes": -5}}, {"at": {"weekdays": [9, "x", None]}},
+ {"until": "2020-01-01T00:00:00Z", "start": "2026-01-01T00:00:00Z"},
+ {"at": {"times": ["25:99", "nope", ""]}},
+ ):
+ assert isinstance(rule_service.validate(junk), dict)
+
+
+def test_prose_and_cron_normalise_to_nothing():
+ """`{}` is the honest answer, and the caller's cue to show the manual form
+ rather than write a schedule that can never fire."""
+ assert rule_service.validate("0 3 * * 1") == {}
+ assert rule_service.validate({"cron": "0 3 * * 1"}) == {}
+ assert rule_service.validate({"every": {"seconds": 5}}) == {}
+
+
+def test_anything_validate_accepts_can_actually_fire():
+ """The flagship invariant. A rule that normalises to something non-empty but
+ has no next occurrence is a schedule indistinguishable from a working one on
+ the list page — which is this feature's worst silent failure."""
+ now = at("2026-08-05T12:00:00")
+ candidates = [
+ {"start": "2026-08-05T14:30:00Z"},
+ {"start": "2026-08-05T14:30:00Z", "every": {"minutes": 10}},
+ {"at": {"times": ["15:00"]}},
+ {"at": {"weekdays": [0], "times": ["15:00"]}},
+ {"at": {"days": [1, 15], "times": ["09:00"]}},
+ {"at": {"months": [1, 7], "days": [1], "times": ["00:00"]}},
+ {"at": {"weekdays": [0], "times": ["15:00"]}, "count": 5},
+ {"start": "2026-08-05T00:00:00Z", "at": {"weekdays": [0], "times": ["15:00"]},
+ "every": {"weeks": 2}},
+ ]
+ for raw in candidates:
+ clean = rule_service.validate(raw)
+ assert clean, raw
+ assert rule_service.next_after(clean, now, zone=PRAGUE) is not None, clean
+
+
+def test_an_interval_below_a_minute_is_raised_not_honoured():
+ """The ticker's granularity is coarser than a second, so honouring it is
+ impossible and pretending to would run silently late for ever."""
+ clean = rule_service.validate({"start": "2026-08-05T00:00:00Z", "every": {"minutes": 0}})
+ assert "every" not in clean
+
+ clean = rule_service.validate(
+ {"start": "2026-08-05T00:00:00Z", "every": {"minutes": 1}}
+ )
+ assert clean["every"] == {"minutes": 1}
+
+
+def test_a_calendar_with_no_time_gets_midnight():
+ """Otherwise "every Monday" means nothing at all, and the whole `at` block
+ would have to be discarded."""
+ clean = rule_service.validate({"at": {"weekdays": [0]}})
+ assert clean["at"]["times"] == ["00:00"]
+
+
+def test_a_window_that_closes_before_it_opens_is_refused():
+ assert rule_service.validate(
+ {"start": "2026-08-05T00:00:00Z", "until": "2026-08-01T00:00:00Z",
+ "every": {"hours": 1}}
+ ) == {}
+
+
+# --- The four combinations ------------------------------------------------------
+def test_a_one_shot_fires_once_and_then_never():
+ clean = rule_service.validate({"start": "2026-08-05T14:30:00Z"})
+ assert rule_service.next_after(clean, at("2026-08-05T12:00:00"), zone=PRAGUE) == at(
+ "2026-08-05T14:30:00"
+ )
+ # Once it has run, `fired` is what stops it being offered again — its moment
+ # is in the past, so nothing else would.
+ assert rule_service.next_after(
+ clean, at("2026-08-05T12:00:00"), zone=PRAGUE, fired=1
+ ) is None
+ assert rule_service.next_after(clean, at("2026-08-05T15:00:00"), zone=PRAGUE) is None
+
+
+def test_a_timer_steps_from_its_start():
+ clean = rule_service.validate(
+ {"start": "2026-08-05T12:00:00Z", "every": {"minutes": 10}}
+ )
+ assert rule_service.next_after(clean, at("2026-08-05T11:00:00"), zone=PRAGUE) == at(
+ "2026-08-05T12:00:00"
+ )
+ assert rule_service.next_after(clean, at("2026-08-05T12:00:00"), zone=PRAGUE) == at(
+ "2026-08-05T12:10:00"
+ )
+ assert rule_service.next_after(clean, at("2026-08-05T12:05:00"), zone=PRAGUE) == at(
+ "2026-08-05T12:10:00"
+ )
+
+
+def test_a_timer_idle_for_a_year_costs_one_division():
+ """Computed rather than stepped. A loop here would spin a hundred thousand
+ times inside the ticker for a schedule nobody touched."""
+ clean = rule_service.validate(
+ {"start": "2020-01-01T00:00:00Z", "every": {"minutes": 1}}
+ )
+ assert rule_service.next_after(clean, at("2026-08-05T12:00:30"), zone=PRAGUE) == at(
+ "2026-08-05T12:01:00"
+ )
+
+
+def test_every_monday_at_three_means_local_three():
+ clean = rule_service.validate({"at": {"weekdays": [0], "times": ["15:00"]}})
+ # 5 August 2026 is a Wednesday; the next Monday is the 10th.
+ found = rule_service.next_after(clean, at("2026-08-05T12:00:00"), zone=PRAGUE)
+ assert found == local("2026-08-10T15:00:00")
+ assert found.astimezone(PRAGUE).strftime("%A %H:%M") == "Monday 15:00"
+
+
+def test_several_times_a_day_are_all_taken_in_order():
+ clean = rule_service.validate({"at": {"times": ["09:00", "17:00"]}})
+ first = rule_service.next_after(clean, local("2026-08-05T08:00:00"), zone=PRAGUE)
+ assert first == local("2026-08-05T09:00:00")
+ second = rule_service.next_after(clean, first, zone=PRAGUE)
+ assert second == local("2026-08-05T17:00:00")
+ third = rule_service.next_after(clean, second, zone=PRAGUE)
+ assert third == local("2026-08-06T09:00:00")
+
+
+def test_a_stride_over_a_calendar_keeps_every_nth():
+ """"Every other Monday" — and anchored on `start`, so it names the same two
+ Mondays whenever it is asked rather than depending on when you looked."""
+ clean = rule_service.validate(
+ {
+ "start": "2026-08-10T00:00:00Z",
+ "at": {"weekdays": [0], "times": ["15:00"]},
+ "every": {"weeks": 2},
+ }
+ )
+ first = rule_service.next_after(clean, at("2026-08-05T00:00:00"), zone=PRAGUE)
+ assert first == local("2026-08-10T15:00:00")
+ second = rule_service.next_after(clean, first, zone=PRAGUE)
+ assert second == local("2026-08-24T15:00:00") # not the 17th
+
+ later = rule_service.next_after(clean, local("2026-08-20T00:00:00"), zone=PRAGUE)
+ assert later == local("2026-08-24T15:00:00")
+
+
+def test_a_day_of_month_that_does_not_exist_every_month_still_finds_one():
+ """31 is a real answer in January and no answer in February. The search must
+ walk on rather than concluding the rule is dead."""
+ clean = rule_service.validate({"at": {"days": [31], "times": ["09:00"]}})
+ found = rule_service.next_after(clean, local("2026-02-01T00:00:00"), zone=PRAGUE)
+ assert found == local("2026-03-31T09:00:00")
+
+
+def test_an_impossible_calendar_answers_never_rather_than_spinning():
+ """31 February matches nothing. Bounded by the search horizon, so the ticker
+ cannot be hung by one bad row."""
+ clean = rule_service.validate(
+ {"at": {"months": [2], "days": [31], "times": ["09:00"]}}
+ )
+ assert rule_service.next_after(clean, at("2026-08-05T00:00:00"), zone=PRAGUE) is None
+
+
+# --- Daylight saving ------------------------------------------------------------
+def test_a_wall_clock_time_survives_spring_forward():
+ """29 March 2026, Prague: 02:00 becomes 03:00 and 02:30 does not exist.
+
+ A daily 02:30 report vanishing once a year, on a schedule nobody is
+ watching, is exactly the failure this whole module is arranged around. It
+ fires at the first instant that does exist instead.
+ """
+ clean = rule_service.validate({"at": {"times": ["02:30"]}})
+ found = rule_service.next_after(clean, local("2026-03-29T00:00:00"), zone=PRAGUE)
+
+ assert found is not None
+ assert found.astimezone(PRAGUE).date().isoformat() == "2026-03-29"
+ # Exactly the first minute that exists, not "somewhere after". Left to
+ # zoneinfo's own resolution this reads 02:30+01:00 — an hour later as an
+ # instant, and a wall-clock time that did not happen.
+ assert found.astimezone(PRAGUE).strftime("%H:%M") == "03:00"
+
+
+def test_a_wall_clock_time_fires_once_across_fall_back():
+ """25 October 2026, Prague: 02:00–03:00 happens twice. Once, not twice."""
+ clean = rule_service.validate({"at": {"times": ["02:30"]}})
+ first = rule_service.next_after(clean, local("2026-10-24T12:00:00"), zone=PRAGUE)
+ assert first.astimezone(PRAGUE).date().isoformat() == "2026-10-25"
+
+ # The next occurrence is the following day, not the repeat of the same hour.
+ second = rule_service.next_after(clean, first, zone=PRAGUE)
+ assert second.astimezone(PRAGUE).date().isoformat() == "2026-10-26"
+ # 25 real hours between them, because the clock went back in between. Taking
+ # the second 02:30 instead would make this one hour.
+ assert second - first == timedelta(hours=25)
+
+
+def test_a_daily_calendar_holds_its_wall_clock_across_a_boundary():
+ """The half that people mean by "every day at 9": 09:00 stays 09:00, and the
+ real interval between two firings is 23 or 25 hours."""
+ clean = rule_service.validate({"at": {"times": ["09:00"]}})
+ before = rule_service.next_after(clean, local("2026-03-28T00:00:00"), zone=PRAGUE)
+ after = rule_service.next_after(clean, before, zone=PRAGUE)
+
+ assert before.astimezone(PRAGUE).hour == 9
+ assert after.astimezone(PRAGUE).hour == 9
+ assert after - before == timedelta(hours=23)
+
+
+def test_a_timer_holds_its_interval_across_a_boundary():
+ """The other half, and the opposite behaviour on purpose: six hours is six
+ hours, so a 23-hour day must not shift or double it."""
+ clean = rule_service.validate(
+ {"start": "2026-03-28T00:00:00Z", "every": {"hours": 6}}
+ )
+ cursor = local("2026-03-28T12:00:00")
+ steps = []
+ for _ in range(6):
+ cursor = rule_service.next_after(clean, cursor, zone=PRAGUE)
+ steps.append(cursor)
+
+ assert all(b - a == timedelta(hours=6) for a, b in zip(steps, steps[1:], strict=False))
+
+
+# --- Exhaustion -----------------------------------------------------------------
+def test_a_count_is_spent_and_then_it_is_over():
+ """"Five times" meaning "for ever" is the failure. Exhaustion returns None,
+ which is the caller's cue to disable rather than to loop."""
+ clean = rule_service.validate(
+ {"start": "2026-08-05T12:00:00Z", "every": {"minutes": 10}, "count": 3}
+ )
+ now = at("2026-08-05T11:00:00")
+ assert rule_service.next_after(clean, now, zone=PRAGUE, fired=2) is not None
+ assert rule_service.next_after(clean, now, zone=PRAGUE, fired=3) is None
+ assert rule_service.next_after(clean, now, zone=PRAGUE, fired=99) is None
+
+
+def test_an_until_closes_the_window():
+ clean = rule_service.validate(
+ {
+ "start": "2026-08-05T12:00:00Z",
+ "every": {"days": 1},
+ "until": "2026-08-08T00:00:00Z",
+ }
+ )
+ assert rule_service.next_after(clean, at("2026-08-06T00:00:00"), zone=PRAGUE) == at(
+ "2026-08-06T12:00:00"
+ )
+ assert rule_service.next_after(clean, at("2026-08-08T00:00:00"), zone=PRAGUE) is None
+
+
+# --- Catching up ----------------------------------------------------------------
+def test_a_missed_run_collapses_to_one():
+ """A host off for a week comes back owing one report, not a hundred and
+ sixty-eight. This is the whole reason `advance` is not just `next_after`."""
+ clean = rule_service.validate(
+ {"start": "2026-08-01T00:00:00Z", "every": {"hours": 1}}
+ )
+ due, following = rule_service.advance(
+ clean,
+ after=at("2026-08-01T00:00:00"),
+ now=at("2026-08-08T00:30:00"),
+ zone=PRAGUE,
+ )
+ assert due is True
+ # The next one is measured from now, not from the slot that was missed.
+ assert following == at("2026-08-08T01:00:00")
+
+
+def test_a_schedule_not_yet_due_is_left_alone():
+ clean = rule_service.validate(
+ {"start": "2026-08-05T12:00:00Z", "every": {"hours": 1}}
+ )
+ due, following = rule_service.advance(
+ clean,
+ after=at("2026-08-05T12:00:00"),
+ now=at("2026-08-05T12:30:00"),
+ zone=PRAGUE,
+ )
+ assert due is False
+ assert following == at("2026-08-05T13:00:00")
+
+
+def test_catching_up_the_last_of_a_count_leaves_nothing_behind():
+ """The firing being caught up counts, so `count` is spent by what actually
+ ran rather than by what was scheduled."""
+ clean = rule_service.validate(
+ {"start": "2026-08-01T00:00:00Z", "every": {"hours": 1}, "count": 3}
+ )
+ due, following = rule_service.advance(
+ clean, after=at("2026-08-01T02:00:00"), now=at("2026-08-08T00:00:00"),
+ zone=PRAGUE, fired=2,
+ )
+ assert due is True
+ assert following is None
+
+
+def test_a_missed_one_shot_still_fires():
+ """A reminder is not less wanted for being late."""
+ clean = rule_service.validate({"start": "2026-08-01T09:00:00Z"})
+ due, following = rule_service.advance(
+ clean, after=at("2026-07-31T00:00:00"), now=at("2026-08-08T00:00:00"), zone=PRAGUE
+ )
+ assert due is True
+ assert following is None
+
+
+# --- Saying it back --------------------------------------------------------------
+def test_describe_says_what_the_rule_actually_does():
+ """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 is the reader's only view of something that happens while they are not
+ looking."""
+ cases = [
+ ({"at": {"weekdays": [0], "times": ["15:00"]}}, "Every Monday at 15:00"),
+ ({"at": {"times": ["09:00"]}}, "Every day at 09:00"),
+ ({"at": {"times": ["09:00", "17:00"]}}, "Every day at 09:00 and 17:00"),
+ ({"start": "2026-08-05T12:00:00Z", "every": {"minutes": 10}}, "Every 10 minutes"),
+ ({"start": "2026-08-05T12:00:00Z", "every": {"hours": 6}}, "Every 6 hours"),
+ (
+ {"start": "2026-08-05T12:00:00Z", "every": {"minutes": 10}, "count": 5},
+ "Every 10 minutes, 5 times",
+ ),
+ ({"at": {"days": [1], "times": ["09:00"]}}, "Every the 1st at 09:00"),
+ ]
+ for raw, expected in cases:
+ assert rule_service.describe(rule_service.validate(raw), zone=PRAGUE) == expected
+
+
+def test_describe_survives_an_empty_rule():
+ assert rule_service.describe({}, zone=PRAGUE) == "Never"
+ assert rule_service.describe(rule_service.validate("nonsense"), zone=PRAGUE) == "Never"
+
+
+def test_describe_names_the_one_shot_moment_in_the_readers_zone():
+ clean = rule_service.validate({"start": "2026-08-05T13:00:00Z"})
+ assert "15:00" in rule_service.describe(clean, zone=PRAGUE)
+ assert "13:00" in rule_service.describe(clean, zone=UTC_ZONE)
diff --git a/tests/test_schedule_ticker.py b/tests/test_schedule_ticker.py
new file mode 100644
index 0000000..c63fab5
--- /dev/null
+++ b/tests/test_schedule_ticker.py
@@ -0,0 +1,450 @@
+"""The ticker and the runner: claiming a firing, and not doing it twice.
+
+`rule.py` is tested on its own in `test_schedule_rule.py`. What is tested here
+is everything that goes wrong *around* a correct rule — which is the half that
+fails silently. Nothing in this file needs an endpoint: the firing itself is
+stubbed, because what is being checked is the bookkeeping, not the reply.
+"""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime, timedelta
+
+import pytest
+from fastapi.testclient import TestClient
+from sqlalchemy import select
+
+from lembas.db.models import (
+ ROLE_ASSISTANT,
+ TARGET_CHAT,
+ TARGET_REPORT,
+ Chat,
+ Message,
+ Report,
+ Schedule,
+ User,
+)
+from lembas.services import settings_store
+from lembas.services.schedule import clock, runner, ticker
+from lembas.services.schedule import rule as rule_service
+
+
+@pytest.fixture(autouse=True)
+def scheduling_on(db):
+ settings_store.update(db, {"enabled": True}, key=settings_store.SCHEDULES)
+ return None
+
+
+def _user(db) -> User:
+ return db.scalars(select(User).order_by(User.created_at)).first()
+
+
+def _schedule(db, *, chat_id: str = "", **kwargs) -> Schedule:
+ fields = {
+ "user_id": _user(db).id,
+ "title": "A schedule",
+ "instruction": "Do the thing.",
+ "rule_json": rule_service.validate(
+ {"start": "2026-01-01T00:00:00Z", "every": {"hours": 1}}
+ ),
+ "target": TARGET_CHAT,
+ "chat_id": chat_id,
+ "enabled": True,
+ "next_fire_at": datetime(2026, 1, 1, tzinfo=UTC),
+ **kwargs,
+ }
+ schedule = Schedule(**fields)
+ db.add(schedule)
+ db.commit()
+ return schedule
+
+
+# --- Claiming -------------------------------------------------------------------
+@pytest.mark.anyio
+async def test_a_firing_that_raises_still_moves_the_schedule_on(
+ client: TestClient, db, registered, make_chat, monkeypatch
+):
+ """The single most important property here.
+
+ Claim, commit, *then* fire. The other order is a hot loop: a schedule whose
+ firing fails is retried every tick for ever, against whatever it was that
+ failed — and the only symptom is load.
+ """
+ schedule = _schedule(db, chat_id=make_chat())
+
+ async def explode(schedule_id, **kwargs):
+ raise RuntimeError("the endpoint is down")
+
+ monkeypatch.setattr(runner, "fire", explode)
+
+ fired = await ticker.sweep(now=datetime(2026, 1, 1, 0, 30, tzinfo=UTC))
+ assert fired == 1
+ for task in list(ticker._FIRING):
+ with pytest.raises(RuntimeError):
+ await task
+
+ db.refresh(schedule)
+ assert schedule.next_fire_at is not None
+ assert schedule.fired_count == 1
+
+
+@pytest.mark.anyio
+async def test_two_overlapping_sweeps_fire_once(
+ client: TestClient, db, registered, make_chat, monkeypatch
+):
+ """A sweep can take minutes — a firing awaits a model. The lock is what
+ stops the next tick claiming the same row again."""
+ schedule = _schedule(db, chat_id=make_chat())
+ calls: list[str] = []
+
+ async def record(schedule_id, **kwargs):
+ calls.append(schedule_id)
+
+ monkeypatch.setattr(runner, "fire", record)
+
+ now = datetime(2026, 1, 1, 0, 30, tzinfo=UTC)
+ await ticker.sweep(now=now)
+ await ticker.sweep(now=now)
+ for task in list(ticker._FIRING):
+ await task
+
+ assert calls == [schedule.id]
+
+
+@pytest.mark.anyio
+async def test_exhaustion_disables_rather_than_looping(
+ client: TestClient, db, registered, make_chat, monkeypatch
+):
+ """"Five times" meaning "for ever" is the failure. A rule with nothing left
+ switches the row off, so it stops being examined at all."""
+ monkeypatch.setattr(runner, "fire", lambda *a, **k: _noop())
+ schedule = _schedule(
+ db,
+ chat_id=make_chat(),
+ rule_json=rule_service.validate(
+ {"start": "2026-01-01T00:00:00Z", "every": {"hours": 1}, "count": 1}
+ ),
+ )
+
+ await ticker.sweep(now=datetime(2026, 1, 1, 0, 30, tzinfo=UTC))
+ for task in list(ticker._FIRING):
+ await task
+
+ db.refresh(schedule)
+ assert schedule.fired_count == 1
+ assert schedule.enabled is False
+ assert schedule.next_fire_at is None
+
+
+@pytest.mark.anyio
+async def test_a_missed_week_fires_once(
+ client: TestClient, db, registered, make_chat, monkeypatch
+):
+ """The host was off. It comes back owing one run, not a hundred and
+ sixty-eight — and the next one is measured from now."""
+ calls: list[str] = []
+
+ async def record(schedule_id, **kwargs):
+ calls.append(schedule_id)
+
+ monkeypatch.setattr(runner, "fire", record)
+ schedule = _schedule(db, chat_id=make_chat())
+
+ now = datetime(2026, 1, 8, 0, 30, tzinfo=UTC)
+ await ticker.sweep(now=now)
+ for task in list(ticker._FIRING):
+ await task
+
+ assert len(calls) == 1
+ db.refresh(schedule)
+ assert schedule.fired_count == 1
+ assert schedule.next_fire_at.replace(tzinfo=UTC) > now
+
+
+@pytest.mark.anyio
+async def test_one_unreadable_row_does_not_stop_the_sweep(
+ client: TestClient, db, registered, make_chat, monkeypatch
+):
+ """A ticker that dies on one bad row stops every schedule on the instance,
+ and nothing anywhere says so."""
+ calls: list[str] = []
+
+ async def record(schedule_id, **kwargs):
+ calls.append(schedule_id)
+
+ monkeypatch.setattr(runner, "fire", record)
+
+ broken = _schedule(db, chat_id=make_chat(), title="Broken")
+ healthy = _schedule(db, chat_id=make_chat(), title="Healthy")
+
+ real_advance = rule_service.advance
+
+ def selective(rule, **kwargs):
+ if rule.get("_broken"):
+ raise ValueError("unreadable")
+ return real_advance(rule, **kwargs)
+
+ broken.rule_json = {**broken.rule_json, "_broken": True}
+ db.commit()
+ monkeypatch.setattr(rule_service, "advance", selective)
+
+ await ticker.sweep(now=datetime(2026, 1, 1, 0, 30, tzinfo=UTC))
+ for task in list(ticker._FIRING):
+ await task
+
+ assert calls == [healthy.id]
+ db.refresh(broken)
+ assert broken.enabled is False
+ assert broken.last_error
+
+
+def test_a_deleted_account_takes_its_schedules_with_it(
+ client: TestClient, db, registered, make_chat
+):
+ """`user_id` is a real ForeignKey with CASCADE — unlike `chat_id`, which is
+ a plain id because `migrations.py` cannot add a REFERENCES clause to a table
+ that already exists. So an orphaned schedule cannot be reached at all, and
+ the owner check in the sweep is a guard rather than a path.
+
+ Worth pinning as the *reason* that guard looks unreachable: somebody
+ removing it would be right about today and wrong the moment the column
+ convention changes.
+ """
+ _schedule(db, chat_id=make_chat())
+ owner = _user(db)
+
+ db.delete(owner)
+ db.commit()
+ # The cascade is enforced by SQLite, not by an ORM relationship, so the
+ # session's identity map still holds the row it was told about. Ask the
+ # database rather than the cache.
+ db.expunge_all()
+
+ assert db.scalars(select(Schedule)).all() == []
+
+
+@pytest.mark.anyio
+async def test_nothing_fires_while_scheduling_is_switched_off(
+ client: TestClient, db, registered, make_chat, monkeypatch
+):
+ """The instance switch is a switch, not a suggestion. Checked in the sweep
+ rather than at the routes, so a row created while it was on does not go on
+ firing after it is turned off."""
+ calls: list[str] = []
+ monkeypatch.setattr(runner, "fire", lambda sid, **k: calls.append(sid) or _noop())
+ _schedule(db, chat_id=make_chat())
+ settings_store.update(db, {"enabled": False}, key=settings_store.SCHEDULES)
+
+ assert await ticker.sweep(now=datetime(2026, 1, 1, 0, 30, tzinfo=UTC)) == 0
+ assert calls == []
+
+
+# --- The runner -----------------------------------------------------------------
+async def _noop() -> None:
+ return None
+
+
+@pytest.mark.anyio
+async def test_a_firing_writes_a_machine_turn_and_starts_a_reply(
+ client: TestClient, db, registered, make_chat
+):
+ """The role stays `user` — `build_messages` needs one there and `_inject`
+ sends a queued turn verbatim. `machine` is what stops the transcript
+ claiming the reader typed it."""
+ chat_id = make_chat()
+ schedule = _schedule(db, chat_id=chat_id)
+
+ await runner.fire(schedule.id)
+
+ turns = list(
+ db.scalars(select(Message).where(Message.chat_id == chat_id).order_by(Message.created_at))
+ )
+ assert turns[0].role == "user"
+ assert turns[0].machine is True
+ assert "started by a schedule" in turns[0].content
+ assert "Do the thing." in turns[0].content
+ # And a reply was opened for it.
+ assert turns[-1].role == ROLE_ASSISTANT
+ assert turns[-1].complete is False
+
+
+@pytest.mark.anyio
+async def test_a_deleted_chat_switches_the_schedule_off(
+ client: TestClient, db, registered, make_chat
+):
+ """Rather than firing into nothing on every tick from now on — which is a
+ schedule that looks alive and produces nothing."""
+ schedule = _schedule(db, chat_id="nosuchchat")
+
+ await runner.fire(schedule.id)
+
+ db.refresh(schedule)
+ assert schedule.enabled is False
+ assert "no longer exists" in schedule.last_error
+
+
+@pytest.mark.anyio
+async def test_a_backlog_skips_rather_than_queues_for_ever(
+ client: TestClient, db, registered, make_chat
+):
+ """`_drain` takes one queued turn per reply, so a schedule firing faster
+ than its chat can answer would build a backlog that outlives the day that
+ caused it."""
+ chat_id = make_chat()
+ chat = db.get(Chat, chat_id)
+ for index in range(5):
+ db.add(
+ Message(chat_id=chat.id, role="user", content=f"waiting {index}", queued=True)
+ )
+ db.commit()
+ schedule = _schedule(db, chat_id=chat_id)
+
+ await runner.fire(schedule.id)
+
+ db.refresh(schedule)
+ assert "previous run was still going" in schedule.last_error
+ assert schedule.claimed_at is None
+
+
+@pytest.mark.anyio
+async def test_run_now_does_not_consume_the_scheduled_run(
+ client: TestClient, db, registered, make_chat
+):
+ """Testing a schedule must not skip the run it was testing. Advancing is the
+ ticker's job and nothing else's."""
+ schedule = _schedule(db, chat_id=make_chat())
+ before = clock.as_utc(schedule.next_fire_at)
+ fired_before = schedule.fired_count
+
+ await runner.run_now(schedule.id)
+
+ db.refresh(schedule)
+ # Through `as_utc` on both sides: a row read back from SQLite is naive while
+ # one still in the session keeps its tzinfo, and comparing the two raises.
+ assert clock.as_utc(schedule.next_fire_at) == before
+ assert schedule.fired_count == fired_before
+
+
+def _finished(db, chat_id: str, text: str) -> Message:
+ """A reply that has already been written.
+
+ `deliver` is tested against this rather than against the output of
+ `runner.fire`, because `fire` starts a real generation — which, with no
+ endpoint configured, races the test to write the same row. The delivery's
+ job begins at a finished message, so that is what it is handed.
+ """
+ message = Message(
+ chat_id=chat_id, role=ROLE_ASSISTANT, content=text, complete=True, model_id="m"
+ )
+ db.add(message)
+ db.commit()
+ return message
+
+
+@pytest.mark.anyio
+async def test_a_report_target_files_the_reply(
+ client: TestClient, db, registered, make_chat
+):
+ chat_id = make_chat()
+ schedule = _schedule(db, chat_id=chat_id, target=TARGET_REPORT, title="Daily news")
+ assistant = _finished(db, chat_id, "Three things happened.")
+
+ await runner.deliver(
+ schedule.id, assistant.id, since=datetime(2020, 1, 1, tzinfo=UTC)
+ )
+
+ filed = db.scalars(select(Report)).all()
+ assert [r.title for r in filed] == ["Daily news"]
+ assert filed[0].body == "Three things happened."
+ assert filed[0].schedule_id == schedule.id
+ assert filed[0].unread is True
+
+
+@pytest.mark.anyio
+async def test_a_run_that_produced_nothing_still_files_something(
+ client: TestClient, db, registered, make_chat
+):
+ """A scheduled report that silently did not appear is indistinguishable
+ from a schedule that never fired. So the failure is filed as a report."""
+ chat_id = make_chat()
+ schedule = _schedule(db, chat_id=chat_id, target=TARGET_REPORT, title="Daily news")
+ broken = Message(chat_id=chat_id, role=ROLE_ASSISTANT, content="", complete=True,
+ error="the endpoint refused")
+ db.add(broken)
+ db.commit()
+
+ await runner.deliver(schedule.id, broken.id, since=datetime(2020, 1, 1, tzinfo=UTC))
+
+ filed = db.scalars(select(Report)).all()
+ assert len(filed) == 1
+ assert filed[0].error
+
+
+@pytest.mark.anyio
+async def test_a_report_the_model_filed_itself_is_not_duplicated(
+ client: TestClient, db, registered, make_chat
+):
+ """`report_write` during the run *is* the report. Filing the reply beside it
+ would put two of everything in the feed."""
+ from lembas.services import reports as reports_service
+
+ chat_id = make_chat()
+ schedule = _schedule(db, chat_id=chat_id, target=TARGET_REPORT)
+ began = datetime.now(tz=UTC) - timedelta(seconds=5)
+ reports_service.create(
+ db, owner=_user(db), title="Filed by the model", body="...",
+ source="chat", source_id=chat_id,
+ )
+ assistant = _finished(db, chat_id, "Also this.")
+
+ await runner.deliver(schedule.id, assistant.id, since=began)
+
+ assert [r.title for r in db.scalars(select(Report))] == ["Filed by the model"]
+
+
+@pytest.mark.anyio
+async def test_a_report_from_an_earlier_run_does_not_suppress_this_one(
+ client: TestClient, db, registered, make_chat
+):
+ """The other half of the dedup, and the one that would fail silently: a
+ daily report would be filed once and then never again, because last week's
+ is still sitting there with the same `source_id`."""
+ from lembas.services import reports as reports_service
+
+ chat_id = make_chat()
+ schedule = _schedule(db, chat_id=chat_id, target=TARGET_REPORT, title="Daily news")
+ reports_service.create(
+ db, owner=_user(db), title="Yesterday's", body="...",
+ source="schedule", source_id=chat_id,
+ )
+ assistant = _finished(db, chat_id, "Today's news.")
+
+ # This run began *after* yesterday's report was filed.
+ await runner.deliver(schedule.id, assistant.id, since=datetime.now(tz=UTC))
+
+ assert sorted(r.title for r in db.scalars(select(Report))) == [
+ "Daily news",
+ "Yesterday's",
+ ]
+
+
+# --- Lifecycle -------------------------------------------------------------------
+def test_starting_twice_makes_one_ticker(client: TestClient, registered):
+ """Two tickers in one process fires everything twice, which is the failure
+ two workers would cause and the reason this is idempotent."""
+ ticker.start()
+ first = ticker._TICKER
+ ticker.start()
+ assert ticker._TICKER is first
+
+
+def test_release_claims_clears_an_interrupted_run(client: TestClient, db, registered):
+ """A restart abandons a firing in flight. Without this the row keeps its
+ claim stamp for ever and reads as permanently running."""
+ schedule = _schedule(db, claimed_at=datetime(2026, 1, 1, tzinfo=UTC))
+
+ assert ticker.release_claims() == 1
+
+ db.refresh(schedule)
+ assert schedule.claimed_at is None
+ assert "interrupted by a restart" in schedule.last_error
diff --git a/tests/test_schedules_ui.py b/tests/test_schedules_ui.py
new file mode 100644
index 0000000..b9732bc
--- /dev/null
+++ b/tests/test_schedules_ui.py
@@ -0,0 +1,373 @@
+"""The Scheduled section: making one, changing it, and the strip on its chat.
+
+The failures worth pinning here are the ones that look like working software:
+a control wired to a method its route does not serve, a form that quietly
+creates something which can never fire, and a task chat that still has a way to
+send a message into it.
+"""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+
+import pytest
+from fastapi.testclient import TestClient
+from sqlalchemy import select
+
+from lembas.db.models import KIND_TASK, TARGET_REPORT, Chat, Schedule, User
+from lembas.security import permissions
+from lembas.services import schedules as schedules_service
+from lembas.services import settings_store
+from lembas.services.schedule import clock
+
+
+@pytest.fixture(autouse=True)
+def scheduling_allowed(db, registered):
+ """`schedule.use` is off by default, deliberately. Granted here so the tests
+ are about the feature rather than about the gate — which has its own test.
+
+ A model is configured too, and that is not scaffolding: with none, the chat
+ page renders its "no models yet" branch instead of the conversation, and
+ every assertion about what the composer area does or does not contain passes
+ for the wrong reason. This file caught exactly that.
+ """
+ from lembas.db.models import Connection, Model
+ from lembas.services.crypto import encrypt
+
+ settings_store.update(db, {"enabled": True}, key=settings_store.SCHEDULES)
+ settings_store.update(
+ db, {"default_permissions": {"schedule.use": True, "reports.use": True}}
+ )
+ connection = Connection(
+ name="Test", base_url="http://127.0.0.1:1", api_key_encrypted=encrypt("")
+ )
+ db.add(connection)
+ db.commit()
+ db.add(Model(connection_id=connection.id, model_id="test-model"))
+ db.commit()
+ return None
+
+
+def _user(db) -> User:
+ return db.scalars(select(User).order_by(User.created_at)).first()
+
+
+def _make(client: TestClient, **overrides) -> None:
+ data = {
+ "title": "Monday build check",
+ "instruction": "Check the build and say what broke.",
+ "target": "chat",
+ "repeat": "calendar",
+ "weekdays": "0",
+ "times": "15:00",
+ "count": "0",
+ **overrides,
+ }
+ return client.post("/api/schedules", data=data, follow_redirects=False)
+
+
+# --- Creating -------------------------------------------------------------------
+def test_creating_a_schedule_makes_its_chat_too(client: TestClient, db, registered):
+ """The one place "chats are created lazily" is bent, and on purpose: the
+ first firing may be days away with nobody present to make one."""
+ response = _make(client)
+ assert response.status_code == 303
+
+ schedule = db.scalars(select(Schedule)).one()
+ chat = db.get(Chat, schedule.chat_id)
+ assert chat is not None
+ assert chat.kind == KIND_TASK
+ assert response.headers["location"] == f"/chat/{chat.id}"
+
+
+def test_a_new_schedule_has_a_next_run(client: TestClient, db, registered):
+ """A schedule that can never fire looks exactly like a working one on the
+ list page. This is the invariant that stops one being written at all."""
+ _make(client)
+ schedule = db.scalars(select(Schedule)).one()
+
+ assert schedule.next_fire_at is not None
+ assert clock.as_utc(schedule.next_fire_at) > datetime.now(tz=UTC)
+ assert schedule.enabled is True
+
+
+def test_a_rule_that_means_nothing_is_refused_with_a_reason(
+ client: TestClient, db, registered
+):
+ """Not a 400 nobody can act on: back to the form, with the reason. A silent
+ refusal here would be a "Schedule it" button that appears to do nothing."""
+ response = _make(client, repeat="once", start_date="", start_time="")
+
+ assert response.status_code == 303
+ assert "/scheduled/new?error=" in response.headers["location"]
+ assert db.scalars(select(Schedule)).all() == []
+
+
+def test_a_one_shot_in_the_past_is_refused(client: TestClient, db, registered):
+ response = _make(client, repeat="once", start_date="2020-01-01", start_time="09:00")
+
+ assert "error=" in response.headers["location"]
+ assert db.scalars(select(Schedule)).all() == []
+
+
+def test_the_form_and_the_engine_agree_about_what_was_stored(
+ client: TestClient, db, registered
+):
+ """The edit screen is derived from the *normalised* rule, so a form showing
+ something other than what runs is impossible rather than merely unlikely."""
+ _make(client, repeat="every", every_amount="6", every_unit="hours")
+ schedule = db.scalars(select(Schedule)).one()
+
+ page = client.get(f"/scheduled/{schedule.id}/edit").text
+ assert 'value="every"\n checked' in page or 'value="every" checked' in page
+ assert 'value="6"' in page
+
+
+def test_a_per_user_ceiling_is_enforced(client: TestClient, db, registered):
+ settings_store.update(db, {"max_per_user": 1}, key=settings_store.SCHEDULES)
+ _make(client)
+
+ response = _make(client, title="A second one")
+
+ assert "error=" in response.headers["location"]
+ assert len(db.scalars(select(Schedule)).all()) == 1
+
+
+# --- The gate --------------------------------------------------------------------
+def test_scheduling_is_off_unless_granted(client: TestClient, db, registered):
+ """`schedule.use` defaults to False: this spends model time with nobody at
+ the keyboard, which is a capability chosen on purpose."""
+ assert permissions.DEFAULT_PERMISSIONS["schedule.use"] is False
+
+
+def _as_stranger(client: TestClient, db, *, role: str = "user") -> User:
+ """Sign in as somebody who is not the administrator.
+
+ Necessary for any test about a permission: `permissions.resolve` gives an
+ admin everything, so asking Frodo whether a gate works answers a different
+ question and answers it yes.
+ """
+ from lembas.security.passwords import hash_password
+
+ stranger = User(
+ email="sam@shire.test",
+ name="Sam",
+ password_hash=hash_password("gardening-is-hard"),
+ role=role,
+ )
+ db.add(stranger)
+ db.commit()
+ client.post("/auth/logout")
+ client.post(
+ "/auth/login", data={"email": "sam@shire.test", "password": "gardening-is-hard"}
+ )
+ return stranger
+
+
+def test_the_sidebar_entry_follows_the_permission(client: TestClient, db, registered):
+ _as_stranger(client, db)
+ assert 'href="/scheduled"' in client.get("/chat").text
+
+ settings_store.update(db, {"default_permissions": {"schedule.use": False}})
+ assert 'href="/scheduled"' not in client.get("/chat").text
+
+
+def test_a_non_admin_without_the_permission_cannot_reach_it(
+ client: TestClient, db, registered
+):
+ settings_store.update(db, {"default_permissions": {"schedule.use": False}})
+ _as_stranger(client, db)
+
+ assert client.get("/scheduled", follow_redirects=False).status_code in (302, 303, 403)
+ assert client.post("/api/schedules", data={}, follow_redirects=False).status_code in (
+ 302,
+ 303,
+ 403,
+ )
+
+
+# --- The task chat ----------------------------------------------------------------
+def test_a_task_chat_has_no_composer(client: TestClient, db, registered):
+ """Suppressed by absence, not by hiding: `chat/_composer.html` is the only
+ thing that posts a message, so its absence is the guarantee. A hidden one
+ would still be a form anybody could post to."""
+ _make(client)
+ schedule = db.scalars(select(Schedule)).one()
+
+ body = client.get(f"/chat/{schedule.chat_id}").text
+ assert "composer__form" not in body
+ assert 'name="content"' not in body
+ # And the controls that do apply are there instead.
+ assert f"/api/schedules/{schedule.id}/run" in body
+ assert f"/api/schedules/{schedule.id}/toggle" in body
+
+
+def test_the_strip_survives_its_schedule_being_removed(
+ client: TestClient, db, registered
+):
+ """Removing a schedule keeps its chat by default. The chat becomes an
+ ordinary one, so it is reachable — a KIND_TASK chat with no schedule behind
+ it would be in no list at all."""
+ _make(client)
+ schedule = db.scalars(select(Schedule)).one()
+ chat_id = schedule.chat_id
+
+ client.post(f"/api/schedules/{schedule.id}/delete", data={"keep_chat": "1"})
+
+ chat = db.get(Chat, chat_id)
+ db.refresh(chat)
+ assert chat is not None
+ assert chat.kind == "chat"
+ assert client.get(f"/chat/{chat_id}").status_code == 200
+
+
+def test_removing_a_schedule_can_take_its_chat(client: TestClient, db, registered):
+ _make(client)
+ schedule = db.scalars(select(Schedule)).one()
+ chat_id = schedule.chat_id
+
+ client.post(f"/api/schedules/{schedule.id}/delete", data={"keep_chat": "0"})
+
+ db.expunge_all()
+ assert db.get(Chat, chat_id) is None
+
+
+# --- Controls that write -----------------------------------------------------------
+def test_pausing_and_resuming_move_the_row(client: TestClient, db, registered):
+ """Asserted on the row rather than on the response: a control wired to a
+ method its route does not serve returns 405 and looks exactly like working
+ software, which cost the agent-mode select an entire release."""
+ _make(client)
+ schedule = db.scalars(select(Schedule)).one()
+
+ client.post(f"/api/schedules/{schedule.id}/toggle", data={"enabled": "0"})
+ db.refresh(schedule)
+ assert schedule.enabled is False
+
+ client.post(f"/api/schedules/{schedule.id}/toggle", data={"enabled": "1"})
+ db.refresh(schedule)
+ assert schedule.enabled is True
+
+
+def test_resuming_recomputes_from_now(client: TestClient, db, registered):
+ """A schedule paused for a month must not come back owing a month of runs.
+ Without this it fires the instant it is switched on."""
+ _make(client, repeat="every", every_amount="1", every_unit="hours")
+ schedule = db.scalars(select(Schedule)).one()
+
+ schedule.enabled = False
+ schedule.next_fire_at = datetime(2020, 1, 1, tzinfo=UTC)
+ db.commit()
+
+ schedules_service.set_enabled(db, schedule, owner=_user(db), enabled=True)
+
+ assert clock.as_utc(schedule.next_fire_at) > datetime.now(tz=UTC)
+
+
+def test_editing_the_rule_restarts_the_count(client: TestClient, db, registered):
+ """An edited schedule is a new intention. Carrying the old `fired_count`
+ into a new `count` would spend most of it before the first run."""
+ _make(client)
+ schedule = db.scalars(select(Schedule)).one()
+ schedule.fired_count = 7
+ db.commit()
+
+ client.post(
+ f"/api/schedules/{schedule.id}",
+ data={
+ "title": "Changed",
+ "instruction": "Something else.",
+ "target": "report",
+ "repeat": "every",
+ "every_amount": "2",
+ "every_unit": "hours",
+ "count": "3",
+ },
+ )
+
+ db.refresh(schedule)
+ assert schedule.fired_count == 0
+ assert schedule.title == "Changed"
+ assert schedule.target == TARGET_REPORT
+
+
+def test_the_routes_refuse_the_wrong_verb(client: TestClient, db, registered):
+ """The other half of the agent-mode lesson: assert the wrong method is
+ *refused*, because only that half would have failed throughout."""
+ _make(client)
+ schedule = db.scalars(select(Schedule)).one()
+
+ assert client.get(f"/api/schedules/{schedule.id}/toggle").status_code == 405
+ assert client.get(f"/api/schedules/{schedule.id}/run").status_code == 405
+ assert client.patch(f"/api/schedules/{schedule.id}/delete").status_code == 405
+
+
+def test_one_persons_schedule_is_not_anothers(client: TestClient, db, registered):
+ from lembas.security.passwords import hash_password
+
+ _make(client)
+ schedule = db.scalars(select(Schedule)).one()
+
+ stranger = User(
+ email="sam@shire.test",
+ name="Sam",
+ password_hash=hash_password("gardening-is-hard"),
+ role="admin",
+ )
+ db.add(stranger)
+ db.commit()
+
+ assert schedules_service.get(db, schedule.id, stranger) is None
+
+
+# --- The admin page ----------------------------------------------------------------
+def test_the_admin_page_saves_and_clamps(client: TestClient, db, registered):
+ """Clamped on read as well as here, for the reason `agents` and `images`
+ give: a value stored by an earlier release, or edited into the database by
+ hand, has to be survivable too. What this asserts is that neither half has
+ been quietly dropped."""
+ response = client.post(
+ "/admin/schedules",
+ data={
+ "enabled": "true",
+ "tick_seconds": "1",
+ "max_per_user": "9999",
+ "max_concurrent": "0",
+ "min_interval_seconds": "1",
+ "max_queued": "500",
+ },
+ follow_redirects=False,
+ )
+ assert response.status_code == 303
+
+ values = settings_store.schedules(db)
+ assert values["enabled"] is True
+ assert values["tick_seconds"] == 5 # floor: a busy loop otherwise
+ assert values["max_per_user"] == 200 # ceiling
+ # 0 falls back to the default rather than clamping to 1, because the
+ # accessor reads `int(stored or default)` -- the same shape `agents` and
+ # `images` use. There is no reading of "no runs at once" that anybody wants:
+ # it would be a ticker that claims work and never does it.
+ assert values["max_concurrent"] == 3
+ assert values["min_interval_seconds"] == 60
+ assert values["max_queued"] == 50
+
+
+def test_the_switch_actually_stops_firing(client: TestClient, db, registered):
+ """A switch that only greys something out is the failure. The sweep reads
+ it, so a schedule created while it was on stops when it is turned off."""
+ import anyio
+
+ from lembas.services.schedule import ticker
+
+ _make(client)
+ client.post("/admin/schedules", data={"enabled": ""}, follow_redirects=False)
+
+ assert anyio.run(ticker.sweep) == 0
+
+
+def test_the_admin_page_is_admin_only(client: TestClient, db, registered):
+ _as_stranger(client, db)
+ assert client.get("/admin/schedules", follow_redirects=False).status_code in (
+ 302, 303, 403, 404,
+ )
diff --git a/tests/test_sidebar_sections.py b/tests/test_sidebar_sections.py
new file mode 100644
index 0000000..015f0e2
--- /dev/null
+++ b/tests/test_sidebar_sections.py
@@ -0,0 +1,116 @@
+"""What may appear in the sidebar's chat tree, and what may not.
+
+The sidebar narrows on `Chat.kind`, and passes `""` whenever the Chat/Agent
+switch is absent -- which is every instance with agent chats turned off. For as
+long as there were exactly two kinds, "" meaning "no filter" and "" meaning
+"both sides of the switch" were the same thing. They stopped being the same
+thing the moment a third kind existed, and the difference is invisible until
+somebody has a conversation that belongs to a section instead of to the tree.
+
+That is the shape this file exists for: correct code whose meaning changed
+underneath it. It is pinned in both places that do the narrowing, because they
+are two implementations of one rule and only one of them is SQL.
+"""
+
+from __future__ import annotations
+
+from fastapi.testclient import TestClient
+from sqlalchemy import select
+
+from lembas.db.models import ALL_KINDS, KIND_MESSAGES, KIND_TASK, KINDS, Chat, Folder, User
+
+
+def _user(db) -> User:
+ return db.scalars(select(User).order_by(User.created_at)).first()
+
+
+def _chat(db, *, kind: str, folder: Folder | None = None) -> Chat:
+ chat = Chat(user_id=_user(db).id, kind=kind, folder_id=folder.id if folder else None)
+ db.add(chat)
+ db.commit()
+ return chat
+
+
+# --- The vocabulary -----------------------------------------------------------
+def test_kinds_stays_the_two_sided_switch():
+ """`api/preferences.py:set_sidebar_kind` validates against KINDS, so a third
+ entry makes the tree filterable to a side with no button to leave it -- the
+ "one side of a fork nobody can move" failure `sidebar_split` already guards.
+ New kinds go in ALL_KINDS.
+ """
+ assert KINDS == ("chat", "agent")
+ assert set(ALL_KINDS) > set(KINDS)
+
+
+def test_the_sidebar_switch_refuses_a_kind_that_is_not_a_side(
+ client: TestClient, db, registered
+):
+ client.post("/api/preferences/sidebar-kind", data={"kind": KIND_TASK})
+ stored = (_user(db).settings_json or {}).get("sidebar_kind")
+ assert stored != KIND_TASK
+
+
+# --- The two narrowings -------------------------------------------------------
+def test_unfiled_sections_chats_stay_out_of_the_tree(client: TestClient, db, registered):
+ """With no switch on screen the sidebar asks for "" -- and "" must not mean
+ "everything". This is the SQL half, in `sidebar_context`."""
+ from lembas.api.pages import sidebar_context
+
+ ordinary = _chat(db, kind="chat")
+ task = _chat(db, kind=KIND_TASK)
+ conversation = _chat(db, kind=KIND_MESSAGES)
+
+ listed = {c.id for c in sidebar_context(db, _user(db))["unfiled_chats"]}
+ assert ordinary.id in listed
+ assert task.id not in listed
+ assert conversation.id not in listed
+
+
+def test_foldered_sections_chats_stay_out_of_the_tree(client: TestClient, db, registered):
+ """And this is the Python half, in `Folder.visible_chats`. Two
+ implementations of one rule, so both are pinned: fixing only the query would
+ leave a task chat filed in a folder showing up anyway.
+ """
+ folder = Folder(user_id=_user(db).id, name="Work")
+ db.add(folder)
+ db.commit()
+
+ ordinary = _chat(db, kind="chat", folder=folder)
+ _chat(db, kind=KIND_TASK, folder=folder)
+
+ listed = {c.id for c in folder.visible_chats()}
+ assert listed == {ordinary.id}
+ # And with the switch present it is still only the ordinary one.
+ assert {c.id for c in folder.visible_chats("chat")} == {ordinary.id}
+
+
+def test_a_folder_holding_only_a_task_chat_reads_as_empty(client: TestClient, db, registered):
+ """`shown_in` keeps a folder that is empty of everything, because hiding a
+ container somebody just made means it can never be filed into. A folder
+ holding only a task chat has to count as that empty one -- otherwise it
+ shows on both sides claiming contents nobody can see.
+ """
+ folder = Folder(user_id=_user(db).id, name="Scheduled work")
+ db.add(folder)
+ db.commit()
+ _chat(db, kind=KIND_TASK, folder=folder)
+
+ assert folder.holds() is False
+ assert folder.shown_in("chat") is True
+
+
+def test_the_composer_cannot_manufacture_a_section_chat(client: TestClient, db, registered):
+ """`_new_chat` collapses kind to agent-or-chat, so this is already true by
+ construction. Pinned so it stays true: the ordinary composer is a form
+ anybody can post to."""
+ from lembas.db.models import Connection, Model
+
+ connection = Connection(name="local", base_url="http://x.test/v1", enabled=True)
+ db.add(connection)
+ db.commit()
+ db.add(Model(connection_id=connection.id, model_id="m", display_name="M", enabled=True))
+ db.commit()
+
+ client.post("/api/chats/start", data={"content": "hello", "kind": KIND_TASK})
+ kinds = {c.kind for c in db.scalars(select(Chat))}
+ assert KIND_TASK not in kinds