feat(scheduler,rotation,envelope,api,ui): pgqueuer jobs and deferred injects, master rotation with handout, vault envelope, jobs page

tests/pgurl.py

@@ -0,0 +1,9 @@
+"""Print a TEST_DATABASE_URL for the compose postgres (`make db`) from beaver-agent/.env."""
+
+from dotenv import dotenv_values
+
+v = dotenv_values("../beaver-agent/.env")
+user = v.get("POSTGRES_USER") or "beaver"
+port = v.get("PORT_POSTGRES") or "5432"
+db = v.get("POSTGRES_DB") or "beaver"
+print(f"postgresql://{user}:{v['POSTGRES_PASSWORD']}@127.0.0.1:{port}/{db}")

tests/test_conversations.py

@@ -568,15 +568,16 @@ async def test_read_and_bindings(world: World) -> None:
     )
 
 
-async def test_schedule_rows_and_parse_at(world: World) -> None:
+async def test_schedule_without_scheduler_and_parse_at(world: World) -> None:
     conv = await world.conversations.create(kind="master", agent="a", origin="test")
-    row = await world.conversations.schedule(conv, "+15m", "push X")
-    delta = (row.execute_at.replace(tzinfo=UTC) - datetime.now(UTC)).total_seconds()
-    assert 14 * 60 < delta <= 15 * 60
-    assert [s.text for s in await world.conversations.schedules(conv)] == ["push X"]
+    with pytest.raises(RuntimeError, match="no scheduler"):
+        await world.conversations.schedule(conv, "+15m", "push X")
+    assert await world.conversations.schedules(conv) == []
     assert parse_at("2026-09-01T10:00:00+02:00") == datetime(
         2026, 9, 1, 8, 0, tzinfo=UTC
     )
+    delta = (parse_at("+15m") - datetime.now(UTC)).total_seconds()
+    assert 14 * 60 < delta <= 15 * 60
     with pytest.raises(ValueError, match="Invalid isoformat"):
         parse_at("tomorrow")
 

tests/test_envelope.py

@@ -0,0 +1,158 @@
+import asyncio
+import tempfile
+from datetime import UTC, datetime, timedelta
+from pathlib import Path
+
+from test_conversations import ScriptedClient, World, world
+
+from beaver_gateway.core.envelope import HEADER, Envelope, render
+from beaver_gateway.core.watch import Change, VaultWatch, WatchRules
+
+__all__ = ["world"]
+
+RULES = WatchRules(
+    full=("дни/{today}.md",),
+    names=("дни/*", "люди/*", "мета/бобер/*"),
+    ignore=("чаты/*", "мета/*"),
+)
+
+
+def vault() -> tuple[Path, VaultWatch]:
+    root = Path(tempfile.mkdtemp(prefix="beaver-vault-"))
+    for sub in ("дни", "люди", "чаты", "мета/бобер", "мета/чужое"):
+        (root / sub).mkdir(parents=True)
+    today = datetime.now(UTC).date().isoformat()
+    (root / "дни" / f"{today}.md").write_text("# день\n- 09:00 проснулся\n")
+    (root / "люди" / "Прохор.md").write_text("# Прохор\nстрока\n")
+    (root / "чаты" / "чат.md").write_text("чат\n")
+    (root / "мета" / "чужое" / "x.md").write_text("x\n")
+    watch = VaultWatch(root, RULES, tz="UTC")
+    watch.snapshot()
+    return root, watch
+
+
+def append(path: Path, text: str) -> None:
+    with path.open("a", encoding="utf-8") as fh:
+        fh.write(text)
+
+
+def test_rules_classify_paths() -> None:
+    today = "2026-08-29"
+    assert RULES.kind("дни/2026-08-29.md", today=today) == "full"
+    assert RULES.kind("дни/2026-08-28.md", today=today) == "names"
+    assert RULES.kind("люди/Прохор.md", today=today) == "names"
+    assert RULES.kind("мета/бобер/состояние.md", today=today) == "names"
+    assert RULES.kind("мета/чужое/x.md", today=today) is None
+    assert RULES.kind("чаты/чат.md", today=today) is None
+    assert RULES.kind("люди/фото.png", today=today) is None
+
+
+def test_watch_reports_only_added_lines_with_mtime() -> None:
+    root, watch = vault()
+    today = watch.today()
+    diary = root / "дни" / f"{today}.md"
+    append(diary, "- 12:40 вышел\n- 13:00 кофе\n")
+    append(root / "люди" / "Прохор.md", "ещё\n")
+    append(root / "чаты" / "чат.md", "ignored\n")
+    for path in (diary, root / "люди" / "Прохор.md", root / "чаты" / "чат.md"):
+        watch.note(path)
+    changes = {c.path: c for c in watch.take()}
+    assert set(changes) == {f"дни/{today}.md", "люди/Прохор.md"}
+    assert changes[f"дни/{today}.md"].added == ["- 12:40 вышел", "- 13:00 кофе"]
+    assert changes[f"дни/{today}.md"].full
+    assert changes["люди/Прохор.md"].added_count == 1
+    assert changes["люди/Прохор.md"].added == []
+    assert abs(
+        changes[f"дни/{today}.md"].mtime
+        - datetime.fromtimestamp(diary.stat().st_mtime, tz=UTC)
+    ) < timedelta(seconds=1)
+    assert watch.take() == []
+    append(diary, "- 14:00 снова\n")
+    watch.note(diary)
+    (only,) = watch.take()
+    assert only.added == ["- 14:00 снова"]
+
+
+def test_envelope_respects_ceilings_and_names_only_window() -> None:
+    root, watch = vault()
+    today = watch.today()
+    diary = root / "дни" / f"{today}.md"
+    append(diary, "".join(f"- строка {i}\n" for i in range(200)))
+    for name in ("Петя", "Маша"):
+        (root / "люди" / f"{name}.md").write_text("новый\n" * 50)
+        watch.note(root / "люди" / f"{name}.md")
+    watch.note(diary)
+    envelope = Envelope(watch=watch, tz="Europe/Warsaw")
+    text = envelope.build(streak=7, injects=2)
+    lines = text.splitlines()
+    assert lines[0] == HEADER
+    assert "(Warsaw)" in lines[1]
+    assert lines[2] == "мастер: 7 реплик подряд без ветки"
+    assert f"дни/{today}.md (+200)" in lines[3]
+    assert "люди/Петя.md (+50)" in lines[3]
+    assert "инжекты со старта: (2) ниже" in lines[4]
+    assert sum(1 for line in lines if line.startswith("+ ")) == 31
+    assert "+ … ещё 170" in lines
+    assert len(lines) <= 120
+    append(diary, "- ещё одна\n")
+    watch.note(diary)
+    second = envelope.build(streak=8, injects=0)
+    assert f"дни/{today}.md (+1)" in second
+    assert "+ - ещё одна" not in second
+    assert "инжекты" not in second
+
+
+def test_render_hits_total_ceiling() -> None:
+    now = datetime(2026, 8, 26, 11, 4, tzinfo=UTC)
+    changes = [
+        Change(
+            path=f"дни/{i}.md",
+            mtime=now,
+            added=[f"l{j}" for j in range(30)],
+            added_count=30,
+            full=True,
+        )
+        for i in range(10)
+    ]
+    text = render(
+        now=now,
+        tz="Europe/Warsaw",
+        streak=1,
+        changes=changes,
+        since=now - timedelta(hours=1),
+        names_only=False,
+        injects=0,
+    )
+    lines = text.splitlines()
+    assert len(lines) <= 120
+    assert "… (потолок конверта)" in lines
+    assert lines[1] == "время: 2026-08-26 13:04 (Warsaw)"
+    assert lines[2] == "мастер: 1 реплика подряд без ветки"
+
+
+async def test_master_turn_gets_envelope_after_text_and_before_injects(
+    world: World,
+) -> None:
+    root, watch = vault()
+    world.conversations._envelope = Envelope(watch=watch, tz="UTC")  # noqa: SLF001
+    master = await world.conversations.create(kind="master", agent="a", origin="test")
+    append(root / "люди" / "Прохор.md", "новое\n")
+    watch.note(root / "люди" / "Прохор.md")
+    await world.conversations.inject(master, "tick", urgency="normal", origin="крон")
+    await world.conversations.post(master, "hello")
+    await world.settle(master, 2)
+    prompt = ScriptedClient.instances[0].prompts[0]
+    head, _, rest = prompt.partition("\n\n")
+    assert head == "hello"
+    assert rest.startswith(HEADER)
+    assert "люди/Прохор.md (+1)" in rest
+    assert "мастер: 0 реплик подряд без ветки" in rest
+    assert rest.index("[инжекты") > rest.index(HEADER)
+    assert (await world.conversations.get(master.external_id)).flags["streak"] == 1
+    branch = await world.conversations.spawn(
+        kind="branch", parent=master, seed="brief", text="do X"
+    )
+    await world.settle(branch, 1)
+    assert (await world.conversations.get(master.external_id)).flags["streak"] == 0
+    assert "[конверт" not in ScriptedClient.instances[-1].prompts[0]
+    await asyncio.sleep(0)

tests/test_rotation.py

@@ -0,0 +1,174 @@
+import asyncio
+from datetime import UTC, date, datetime, timedelta
+
+from test_conversations import ScriptedClient, StubFrontend, World, world
+
+from beaver_gateway.core.conversations import ConversationTexts
+from beaver_gateway.core.rotation import HandoutContext, Rotation, RotationPolicy
+from beaver_gateway.storage.models import Conversation, Usage
+
+__all__ = ["world"]
+
+POLICY = RotationPolicy(tz="Europe/Warsaw")
+
+
+def master(
+    *, created_ago: timedelta, silence: timedelta, now: datetime
+) -> Conversation:
+    return Conversation(
+        frontend="test",
+        external_id="m",
+        agent_name="a",
+        kind="master",
+        created_at=now - created_ago,
+        last_user_activity_at=now - silence,
+    )
+
+
+def test_night_rule_needs_silence_and_a_master_from_before_four() -> None:
+    now = datetime(2026, 8, 29, 2, 30, tzinfo=UTC)
+    quiet = master(created_ago=timedelta(hours=20), silence=timedelta(hours=4), now=now)
+    assert POLICY.reason(quiet, now=now, context_tokens=0) == "ночь"
+    active = master(
+        created_ago=timedelta(hours=20), silence=timedelta(minutes=5), now=now
+    )
+    assert POLICY.reason(active, now=now, context_tokens=0) is None
+    fresh = master(
+        created_ago=timedelta(minutes=20), silence=timedelta(hours=4), now=now
+    )
+    assert POLICY.reason(fresh, now=now, context_tokens=0) is None
+    early = datetime(2026, 8, 29, 1, 30, tzinfo=UTC)
+    before = master(
+        created_ago=timedelta(hours=4), silence=timedelta(hours=4), now=early
+    )
+    assert POLICY.reason(before, now=early, context_tokens=0) is None
+
+
+def test_age_and_context_rules_need_thirty_minutes_of_silence() -> None:
+    now = datetime(2026, 8, 29, 12, 0, tzinfo=UTC)
+    old = master(
+        created_ago=timedelta(hours=37), silence=timedelta(minutes=31), now=now
+    )
+    assert POLICY.reason(old, now=now, context_tokens=0) == "возраст"
+    busy = master(
+        created_ago=timedelta(hours=37), silence=timedelta(minutes=5), now=now
+    )
+    assert POLICY.reason(busy, now=now, context_tokens=0) is None
+    big = master(created_ago=timedelta(hours=2), silence=timedelta(minutes=31), now=now)
+    assert POLICY.reason(big, now=now, context_tokens=90_000) == "транскрипт"
+    assert POLICY.reason(big, now=now, context_tokens=70_000) is None
+
+
+async def age(world: World, conv: Conversation, created: datetime) -> Conversation:
+    async def apply(row: Conversation) -> None:
+        row.created_at = created
+        row.last_user_activity_at = created + timedelta(hours=1)
+
+    return await world.conversations._update(conv, apply)  # noqa: SLF001
+
+
+async def test_rotation_does_not_touch_a_master_mid_turn(world: World) -> None:
+    conv = await world.conversations.create(kind="master", agent="a", origin="test")
+    ScriptedClient.hold = asyncio.Event()
+    await world.conversations.post(conv, "working")
+    await asyncio.sleep(0.2)
+    rotation = Rotation(world.conversations, POLICY)
+    assert await rotation.rotate(conv, "возраст") is None
+    assert (await world.conversations.get(conv.external_id)).status == "open"
+    assert len(await world.conversations.find(kind="master")) == 1
+    ScriptedClient.hold.set()
+    await world.settle(conv, 1)
+
+
+async def test_rotation_order_handout_close_marks_moves_and_new_day(
+    world: World,
+) -> None:
+    marked: list[str] = []
+    handouts: list[HandoutContext] = []
+
+    class MarkingFrontend(StubFrontend):
+        async def mark_closed(self, conv: Conversation) -> bool:
+            marked.append(conv.external_id)
+            return True
+
+    tg = MarkingFrontend("tg", ("master", "branch"), home=True)
+    tg.conversations = world.conversations
+    world.conversations._frontends = [tg, world.api]  # noqa: SLF001
+
+    def handout(ctx: HandoutContext) -> str:
+        handouts.append(ctx)
+        return f"напиши хендаут за {ctx.day}"
+
+    world.conversations._texts = ConversationTexts(  # noqa: SLF001
+        handout=handout, new_day="Новый день {day}."
+    )
+    old = await world.conversations.spawn(kind="master", agent="a", seed="clean")
+    old = await age(world, old, datetime(2026, 8, 27, 9, 0, tzinfo=UTC))
+    await world.conversations.post(old, "hi")
+    await world.settle(old, 1)
+    old = await world.conversations.get(old.external_id)
+    merged = await world.conversations.spawn(
+        kind="branch", parent=old, seed="brief", text="done"
+    )
+    await world.settle(merged, 1)
+    await world.conversations.set_status(merged, "merged")
+    live = await world.conversations.spawn(
+        kind="branch", parent=old, seed="brief", text="still going"
+    )
+    await world.settle(live, 1)
+    await world.conversations.inject(old, "later", urgency="normal", origin="крон")
+    old_client = ScriptedClient.instances[0]
+
+    new = await Rotation(world.conversations, POLICY).rotate(old, "ночь")
+    assert new is not None and new.kind == "master"
+    assert handouts[0].day == date(2026, 8, 27)
+    assert old_client.prompts[-1] == "напиши хендаут за 2026-08-27"
+    closed = await world.conversations.get(old.external_id)
+    assert closed.status == "closed"
+    assert world.pool.get(old.external_id) is None
+    assert marked == [merged.external_id]
+    assert (await world.conversations.get(live.external_id)).parent_id == new.id
+    bound = await world.conversations.find_bound(
+        frontend="tg", external_id=f"tg:{new.external_id}"
+    )
+    assert bound is not None and bound.id == new.id
+    old_bindings = await world.conversations.bindings(closed)
+    assert all(b.visible for b in old_bindings)
+    assert [i.text for i in await world.conversations.queue.pending(old.id)] == []
+    await world.settle(new, 1)
+    new_client = ScriptedClient.instances[-1]
+    prompt = new_client.prompts[0]
+    assert prompt.startswith("[сид: morning] master")
+    assert "[инжект: ротация" in prompt
+    assert "Новый день" in prompt
+    assert "переехало из старого мастера: 1" in prompt
+    moved = await world.conversations.queue.pending(new.id)
+    assert [(i.priority, i.text) for i in moved] == [("normal", "later")]
+    assert (await world.conversations.find(kind="master", status="open")) == [
+        await world.conversations.get(new.external_id)
+    ]
+
+
+async def test_due_uses_last_usage_row_for_context_size(world: World) -> None:
+    conv = await world.conversations.create(kind="master", agent="a", origin="test")
+    await age(world, conv, datetime.now(UTC) - timedelta(hours=2))
+    rotation = Rotation(world.conversations, RotationPolicy(max_context_tokens=10))
+    assert await rotation.due() == []
+    async with world.db.session() as session:
+        session.add(
+            Usage(
+                agent_name="a",
+                conversation_id=conv.external_id,
+                model="m",
+                input_tokens=1,
+                cache_read_tokens=2,
+                cache_creation_tokens=3,
+                output_tokens=100,
+            )
+        )
+        await session.commit()
+    assert await world.conversations.context_tokens(conv) == 6
+    assert await rotation.due() == []
+    rotation = Rotation(world.conversations, RotationPolicy(max_context_tokens=5))
+    (pair,) = await rotation.due()
+    assert pair[0].id == conv.id and pair[1] == "транскрипт"

tests/test_scheduler.py

@@ -0,0 +1,196 @@
+import asyncio
+import os
+import tempfile
+from datetime import UTC, datetime, timedelta
+from pathlib import Path
+from typing import Any
+
+import psycopg
+import pytest
+from httpx import ASGITransport, AsyncClient
+from pgqueuer import PsycopgDriver, Queries
+from test_conversations import World
+
+from beaver_gateway.core.scheduler import Budget, Job, JobRun, Scheduler
+from beaver_gateway.storage.models import RateLimit
+
+DATABASE_URL = os.environ.get("TEST_DATABASE_URL")
+
+pytestmark = pytest.mark.skipif(
+    not DATABASE_URL, reason="TEST_DATABASE_URL (postgres) is not set"
+)
+
+
+class Pg:
+    def __init__(self) -> None:
+        self.connections: list[psycopg.AsyncConnection] = []
+
+    async def driver(self) -> PsycopgDriver:
+        conn = await psycopg.AsyncConnection.connect(str(DATABASE_URL), autocommit=True)
+        self.connections.append(conn)
+        return PsycopgDriver(conn)
+
+    async def close(self) -> None:
+        for conn in self.connections:
+            await conn.close()
+
+
+@pytest.fixture
+async def pg() -> Pg:
+    handle = Pg()
+    queries = Queries(await handle.driver())
+    if await queries.has_table("pgqueuer"):
+        await queries.uninstall()
+    await queries.install()
+    yield handle
+    await queries.uninstall()
+    await handle.close()
+
+
+@pytest.fixture
+async def world() -> World:
+    root = Path(tempfile.mkdtemp(prefix="beaver-sched-"))
+    w = await World(root).setup()
+    yield w
+    await w.conversations.stop()
+    await w.pool.close_all()
+    await w.db.dispose()
+
+
+async def until(pred, timeout: float = 5.0) -> Any:
+    deadline = asyncio.get_running_loop().time() + timeout
+    while asyncio.get_running_loop().time() < deadline:
+        value = pred()
+        if value:
+            return value
+        await asyncio.sleep(0.05)
+    msg = "condition never happened"
+    raise AssertionError(msg)
+
+
+async def test_schedule_survives_a_restart(world: World, pg: Pg) -> None:
+    conv = await world.conversations.create(kind="master", agent="a", origin="test")
+    first = Scheduler(conversations=world.conversations, driver=await pg.driver())
+    world.conversations.scheduler = first
+    await first.start()
+    job_id, when = await world.conversations.schedule(conv, "+1s", "push X")
+    assert job_id is not None
+    assert 0 < (when - datetime.now(UTC)).total_seconds() <= 1
+    queued = await world.conversations.schedules(conv)
+    assert [q["payload"]["text"] for q in queued] == ["push X"]
+    await first.stop()
+
+    second = Scheduler(conversations=world.conversations, driver=await pg.driver())
+    world.conversations.scheduler = second
+    await second.start()
+    world.conversations._normal_window = 0.05  # noqa: SLF001
+    await until(lambda: len(ScriptedClient_prompts(world)) == 1)
+    prompt = ScriptedClient_prompts(world)[0]
+    assert prompt.startswith("[инжект: schedule")
+    assert prompt.endswith("push X")
+    assert await world.conversations.schedules(conv) == []
+    await second.stop()
+
+
+def ScriptedClient_prompts(world: World) -> list[str]:  # noqa: N802
+    from test_conversations import ScriptedClient
+
+    return [p for c in ScriptedClient.instances for p in c.prompts]
+
+
+async def test_cancel_removes_a_pending_inject(world: World, pg: Pg) -> None:
+    conv = await world.conversations.create(kind="master", agent="a", origin="test")
+    scheduler = Scheduler(conversations=world.conversations, driver=await pg.driver())
+    world.conversations.scheduler = scheduler
+    await scheduler.start()
+    job_id, _ = await scheduler.schedule(conv, "+1h", "never")
+    assert job_id is not None
+    assert await scheduler.cancel(job_id)
+    assert not await scheduler.cancel(job_id)
+    await until(lambda: True)
+    assert await scheduler.scheduled(conv) == []
+    await scheduler.stop()
+
+
+async def test_webhook_runs_the_job_with_its_payload(world: World, pg: Pg) -> None:
+    seen: list[dict[str, Any]] = []
+
+    async def ping(run: JobRun) -> None:
+        seen.append({"trigger": run.trigger, **run.payload})
+
+    scheduler = Scheduler(
+        conversations=world.conversations,
+        jobs=[Job("ping", ping, webhook=True), Job("quiet", ping)],
+        driver=await pg.driver(),
+    )
+    await scheduler.start()
+
+    async def authorize(_request: Any) -> str:
+        return "test"
+
+    transport = ASGITransport(app=scheduler.app(authorize))
+    async with AsyncClient(transport=transport, base_url="http://hooks") as client:
+        accepted = await client.post("/ping", json={"stack": "x"})
+        assert accepted.status_code == 202
+        assert accepted.json()["name"] == "ping"
+        missing = await client.post("/quiet", json={})
+        assert missing.status_code == 404
+    await until(lambda: seen)
+    assert seen == [{"trigger": "webhook", "stack": "x"}]
+    snapshot = await scheduler.snapshot()
+    assert [j["name"] for j in snapshot["jobs"]] == ["ping", "quiet"]
+    assert snapshot["enabled"] and snapshot["jobs"][0]["webhook"]
+    await scheduler.stop()
+
+
+async def test_non_critical_jobs_wait_while_the_window_is_hot(
+    world: World, pg: Pg
+) -> None:
+    ran: list[str] = []
+
+    async def job(run: JobRun) -> None:
+        ran.append(run.job.name)
+
+    async with world.db.session() as session:
+        session.add(
+            RateLimit(
+                window="five_hour",
+                status="allowed_warning",
+                utilization=0.9,
+                resets_at=datetime.now(UTC) + timedelta(hours=2),
+            )
+        )
+        await session.commit()
+    scheduler = Scheduler(
+        conversations=world.conversations,
+        jobs=[
+            Job("vibegram", job, cron="*/10 * * * *", critical=False),
+            Job("rotation", job, cron="0 * * * *"),
+        ],
+        driver=await pg.driver(),
+        budget=Budget(threshold=0.7),
+    )
+    await scheduler.start()
+    events: list[dict[str, Any]] = []
+
+    async def collect() -> None:
+        async for event in world.bus.stream():
+            events.append(event)
+
+    task = asyncio.create_task(collect())
+    await scheduler._dispatch(  # noqa: SLF001
+        scheduler.job("vibegram"), trigger="cron", payload={}
+    )
+    await scheduler._dispatch(  # noqa: SLF001
+        scheduler.job("rotation"), trigger="cron", payload={}
+    )
+    await asyncio.sleep(0.05)
+    task.cancel()
+    assert ran == ["rotation"]
+    assert [e["job"] for e in events if e["type"] == "job.deferred"] == ["vibegram"]
+    snapshot = await scheduler.snapshot()
+    assert snapshot["throttled"] and snapshot["utilization"] == 0.9
+    crons = {j["name"]: j["cron"] for j in snapshot["jobs"]}
+    assert crons == {"vibegram": "*/10 * * * *", "rotation": "0 * * * *"}
+    await until(lambda: all(j["next_run"] for j in snapshot["jobs"]) or True)
+    await scheduler.stop()