From 71a77fe287ec83f9a272d052e51af3fbc6f77deb Mon Sep 17 00:00:00 2001 From: pufit Date: Wed, 11 Nov 2026 11:51:27 -0500 Subject: [PATCH] Cron: memory-maintenance and skill-extractor run only after the agent has worked Both built-in jobs ran on every schedule, memory-maintenance once a day and skill-extractor twice a day, also on an agent that had done nothing since their last run. Each run is a full turn that finds nothing new, and on a platform that parks an idle machine each run wakes the machine. A new run gate, `agent_activity`, is satisfied when the agent has worked since the job's last successful run started: a turn ended, or a message was stored in a session, so a turn that somebody asked for counts also when it is stopped or fails. Not counted: the job's own runs (its cron sessions and the workflow runs and review loops it started), the runs of the jobs in `ignore_jobs`, and the runs of every other job with this gate, so two such jobs never start each other. A job whose gate forecasts a skip is not the next scheduled task, so a parked machine is not woken for it. The gate estimates whether a run is worth it, so a run that a person starts is not checked by it. A catch-up asks it about an earlier moment (after a pause, the time the run came due; at startup, the start of the catch-up), so the message that wakes a parked machine does not start the job beside the turn it began. The two jobs have the gate by default. It is not written into system.yaml: the scheduler gives it to an entry that has neither `run_if` nor `skip_when_idle`. Nothing writes that file again on an update, so this is how an install from before the gate has it and how a later change reaches every install, and a file never names a gate type that an older Nerve would refuse. `run_if: []` keeps a job on every schedule. memU's memory-maintenance reviews the records of the day before the run, so its gate has `overlap_hours: 24` and the job runs at its next two scheduled times after the last work. Loom's checks the namespace as it is and has no overlap. Gates now get the scheduler's job list and, for a catch-up, the time the run came due (`GateContext.jobs`, `GateContext.as_of`). --- docs/cron.md | 136 ++- docs/personal-best-practices.md | 2 +- docs/worker-best-practices.md | 7 +- docs/worker-guide.md | 2 +- nerve/cron/gate_plugins.py | 9 +- nerve/cron/gates.py | 148 +++- nerve/cron/jobs.py | 60 +- nerve/cron/service.py | 92 +- nerve/db/cron.py | 162 ++++ nerve/setup_writer.py | 54 ++ .../templates/guides/worker-best-practices.md | 7 +- nerve/templates/personal/TOOLS.md | 2 +- nerve/templates/worker/TOOLS.md | 2 +- tests/test_config_validate.py | 14 + tests/test_cron_gates.py | 837 ++++++++++++++++++ tests/test_cron_resume.py | 71 ++ tests/test_cron_schedule.py | 38 + tests/test_cron_source_delivery.py | 4 +- tests/test_setup_writer.py | 84 ++ 19 files changed, 1676 insertions(+), 55 deletions(-) diff --git a/docs/cron.md b/docs/cron.md index 6566c20c7..a8a6f11a4 100644 --- a/docs/cron.md +++ b/docs/cron.md @@ -206,6 +206,10 @@ Gates are **fail-open while checking**: if a gate raises during evaluation (e.g. a transient DB issue), the run proceeds rather than being skipped — an occasional wasted run beats a cron that silently never fires. +A run that a person starts (**Run now**, `nerve cron `, the trigger API) is +checked like a scheduled one, with one exception: an +[`agent_activity`](#agent_activity) gate is not checked for it. + A gate that can't be *built* is a different matter and is **refused**. If a `run_if` entry names a type nothing provides — a typo, or a gate plugin that was deleted or fails to import — the job does not load: reload returns `400` naming @@ -264,6 +268,73 @@ already expired still starts again at the newest message, as a new one does. > `idle_consumer: ` fields still work — they are translated into an > equivalent `messages` gate at load time. Prefer `run_if` for new jobs. +#### `agent_activity` + +Satisfied when the agent has worked since the job last ran. It is for a job +that looks after the agent's own work, such as a memory check or a search for +repeated workflows: while the agent has done nothing, the job has nothing to do. + +| Field | Type | Default | Description | +|-------|------|---------|-------------| +| `ignore_jobs` | list | — | Ids of cron jobs whose runs do not count as work. A single id may be written as a string. A bare `ignore_jobs:` means unset; any other value is refused | +| `overlap_hours` | number | 0 | Also count the work of this many hours before the job's last run (see below). From 0 to 8784 | + +```yaml +run_if: + - type: agent_activity + ignore_jobs: [weekly-cleanup] +``` + +The agent has worked when one of these happened since the job's last +successful run started: + +- A turn ended and reported its usage, whatever started it: a chat, a message + from a channel, a wakeup, a hook, another cron job, a workflow run. +- A message was stored in a session. A turn that somebody asked for stores + what it was asked when it starts, so it counts from then on, also when it is + stopped or fails. So does a conversation that Nerve takes in from an external + agent. A turn that the agent starts for itself (a wakeup, a continuation + after background work) stores its messages without moving the session's + time, so it counts when it ends and reports its usage. If it fails before + that, it can leave no record. + +A job that has not succeeded yet runs as soon as there is any work at all. A +run that failed does not move the start of the window, so the job runs again +at its next time. + +Not counted: + +- The job's own runs, so a run is never the reason for the next one. The + sessions of a job are `cron:`, `cron::`, and the sessions + of the workflow runs that the job or one of those sessions started, the legs + of a review loop included. What a person writes in the job's own chat is in + such a session too. +- The runs of the jobs in `ignore_jobs`. +- The runs of every other job that has this gate. Such a job runs only because + the agent worked, so its run is not work of its own. Two jobs with the gate + therefore never start each other, and need not name each other. +- Work that leaves neither a turn nor a message: a memory written through the + MCP endpoint or on the Memory page, for example. +- Work in a session that was deleted since: its records go with it. +- For a catch-up, what was recorded after the moment it asks about: + [after a pause](#after-a-pause-sleep-suspend-snapshot-restore), the time the + run came due; at startup, the start of the catch-up. The message that wakes a + parked machine is work for the job's next scheduled time. It does not start a + run that came due while nothing had happened. + +`overlap_hours` is for a job that looks at a fixed period before its run, and +not at everything since its last run. `memory-maintenance` under memU reviews +the records of the day before the run, so what the agent did in the hours +before a run is not in that run's day yet. With `overlap_hours: 24` that work +counts again for the next run: a daily job runs at its next two scheduled times +after the agent's last work, and then waits. + +The gate estimates whether a run is worth it. It does not say whether the job +has anything to act on, as `tasks` and `messages` do. So a run that a person +starts (**Run now**, `nerve cron `, `POST /api/cron/jobs/{job_id}/trigger`) +is not checked by this gate: the person may know of work that it cannot see. +The other gates of the job are checked as usual. + #### `github_pr_activity` Satisfied when something actionable changed on an author's open PRs: a merge or @@ -325,7 +396,11 @@ that check. instance schedules itself, after which the forecast is read afresh, but also a task created or moved from a chat, the web UI or the API, which is only seen the next time the schedule is read (see [What runs next](#what-runs-next)). - `tasks` forecasts this way. `messages` reads the consumer's cursors + `tasks` and `agent_activity` forecast this way: what `agent_activity` + reads changes when the agent works or the job runs, and Nerve reads the + forecast afresh after a turn and after a run. Other work, such as a + conversation taken in from an external agent, is seen at the next read. + `messages` reads the consumer's cursors without starting one, and takes a cursor that will have expired by `at` as expired, since the check will find it so then. - `github_pr_activity` always forecasts a run: whether a PR moved is only known @@ -372,7 +447,7 @@ class StaleTasksGate(CronGate): self.min_age_minutes = min_age_minutes async def is_satisfied(self, ctx: GateContext) -> bool: - # ctx exposes {job_id, db} — DB-only (see note below). + # ctx exposes {job_id, db, jobs, as_of} (see note below). ... async def forecast(self, ctx: GateContext, at: datetime) -> bool: @@ -427,11 +502,21 @@ that name its gate, see below): remove the `run_if` entry) and reload again. Deleting a plugin is not a way to switch a gate off; it stops those jobs instead of ungating them. -> **Context is DB-only.** A gate receives `GateContext{job_id, db}`, which is -> enough for DB-driven conditions (task counts, source cursors, age filters). A -> gate that needs live runtime state — e.g. which sessions are currently -> running — is **not** supported by this loader; that would require widening -> the gate context, a separate change. +> **Context is the database and the job list.** A gate receives +> `GateContext{job_id, db, jobs, as_of}`, which is enough for DB-driven +> conditions (task counts, source cursors, age filters) and for a condition +> that depends on how the other jobs are set up: `jobs` maps the id of each +> loaded job, disabled ones too, to its definition. `as_of` is set for a +> catch-up: the time the run it stands in for came due, for a gate that reads +> a record of what happened and must leave out what was recorded since. A gate +> that needs live runtime state — e.g. which sessions are currently running — +> is **not** supported by this loader; that would require widening the gate +> context, a separate change. +> +> A plugin gate that calls a built-in gate passes on the context it was given. +> `agent_activity` finds the jobs that carry it among each job's own gates, so +> a job whose plugin gate only calls it is not found: name that job in the +> `ignore_jobs` of the jobs that must not count its runs. > **Trust note.** Files in the gate-plugins directory are imported (executed) > at daemon startup. This is the same trust model as `config.yaml`, configured @@ -517,10 +602,10 @@ These ship in `/config/cron/system.yaml`, written when the mode is ch | Job | Schedule | Session Mode | Description | Personal | Worker | |-----|----------|-------------|-------------|:--------:|:------:| -| `memory-maintenance` | Daily 5 AM | isolated | Dedup, prune stale entries, improve memory wording. Runs silently. On an install set up with `memory.backend: loom`: runs Loom's integrity report (`loom_namespace_validate`) on the home namespace and fixes what is its own, leaving deduplication to Loom; a `system.yaml` written for memU runs in this wording while its prompt is the shipped one. | ✅ always | ✅ always | +| `memory-maintenance` | Daily 5 AM, when the agent has worked (see below) | isolated | Dedup, prune stale entries, improve memory wording. Runs silently. On an install set up with `memory.backend: loom`: runs Loom's integrity report (`loom_namespace_validate`) on the home namespace and fixes what is its own, leaving deduplication to Loom; a `system.yaml` written for memU runs in this wording while its prompt is the shipped one. | ✅ always | ✅ always | | `inbox-processor` | Every 30 min | persistent (24h rotation, reminder mode) | Polls all sync sources (email, GitHub, Telegram). Triages, creates tasks, memorizes facts, sends notifications for urgent items. | ✅ default | — | | `task-planner` | Every 4 hours | persistent (168h rotation) | Reviews open tasks, explores codebases, proposes implementation plans via plan-approve workflow. Gated on `tasks` (status `pending`) — stays idle when there's nothing to plan. | ✅ default | off; onboarding enables it | -| `skill-extractor` | Every 12 hours | persistent | Identifies repeated workflows from recent conversations, memory, and completed tasks. Proposes new skills via task+plan system. | ✅ optional | ✅ default | +| `skill-extractor` | Every 12 hours, when the agent has worked (see below) | persistent | Identifies repeated workflows from recent conversations, memory, and completed tasks. Proposes new skills via task+plan system. | ✅ optional | ✅ default | | `skill-reviser` | Weekly (Sun 3 AM) | persistent | Reviews existing skills for accuracy (outdated paths, credentials), completeness (missing steps), and quality (trigger phrases, examples). Proposes revisions via task+plan. | ✅ optional | ✅ default | | `machine-maintenance` | Weekly (Sun 4 AM) | isolated | Cleans up the machine Nerve runs on. The agent first finds out what this machine has (disk use, repositories and worktrees, build and package caches, Docker, services that listen on a port), then looks only at items that are old and that nothing uses. It removes items that hold no data without asking: caches, dangling Docker images and build cache, a worktree whose pull request is merged and that has no local changes. It sends one `ask_user` for anything that can hold data (Docker volumes, stopped containers, worktrees with local work, leftover dev servers or test containers on open ports) and acts only on what the answer approves. It never touches Nerve itself, system services or anything in use. Sends a low-priority `notify` with the space freed when it changed something. | ✅ default | ✅ default | @@ -528,6 +613,25 @@ These ship in `/config/cron/system.yaml`, written when the mode is ch - **Personal**: `memory-maintenance` (always on) + `inbox-processor` + `task-planner` + `machine-maintenance` enabled by default. `skill-extractor` and `skill-reviser` are offered as toggles on the Automation step of the setup page (and in `nerve init --terminal`). - **Worker**: `memory-maintenance` (always on) + `skill-extractor` + `skill-reviser` + `machine-maintenance` enabled by default when the mode is chosen on the setup page or by `NERVE_MODE=worker nerve init --non-interactive`. `task-planner` is written with `enabled: false` on every path: whether the worker plans through the approval loop is decided during onboarding (the onboarding conversation, or the session in which a worker set up without the browser sets itself up), which enables it by overriding the job by id in `jobs.yaml` with a prompt and a `tasks` gate scoped to the worker's tag (the Automation step of the setup page can also switch it on). `nerve init --terminal` has no cron step for workers and writes the two skill jobs and `machine-maintenance` with `enabled: false` as well; set `enabled: true` in `system.yaml` (or override the job by id in `jobs.yaml`), then `nerve reload`. `inbox-processor` is not included (workers don't have sync sources). +`memory-maintenance` and `skill-extractor` look after the agent's own work, so both wait for it with an [`agent_activity`](#agent_activity) gate: + +- Each runs at its next scheduled time after the agent has worked, and not again until the agent works again. A catch-up can start it at another time: when Nerve starts, or the machine resumes, after a scheduled time of the job has passed without a run (see [Startup catch-up](#startup-catch-up) and [After a pause](#after-a-pause-sleep-suspend-snapshot-restore)). Only work from before that moment counts for it, so the message that wakes a parked machine does not start either job. +- The runs of the four maintenance jobs (these two, `skill-reviser` and `machine-maintenance`) do not count as work for either, so one of them never starts another. +- An agent that does nothing runs neither: nobody uses it, and no other job of it runs. On a platform that parks an idle machine, the machine is not woken for them (see [What runs next](#what-runs-next)). An agent with a scheduled job of its own, a scan every hour for example, works at each run of it, so both jobs go on running. `inbox-processor` and `task-planner` count too when they run. +- Under memU, the gate of `memory-maintenance` has `overlap_hours: 24`: the job reviews the records of the day before the run, so it runs at its next two scheduled times after the agent's last work. Under Loom it checks the namespace as it is and has no overlap; what reaches Loom after a check waits for the job's next run. +- **Run now** starts either job whatever the gate says. + +`system.yaml` does not state these two gates. The scheduler gives them to an entry of the two jobs that has neither `run_if` nor `skip_when_idle`, in the form for the configured memory backend, and logs one line for each when it loads the file. Nothing writes `system.yaml` again when Nerve is updated, so this is how an install from before the gates has them, and how a later change to them reaches every install. + +| The entry in `system.yaml` | What runs | +|---|---| +| No `run_if` key | The job with its default gate. The prompt and the schedule of the entry do not matter: an edited entry gets the gate too | +| `run_if: []`, or a bare `run_if:` | The job on every schedule | +| A `run_if` of its own | The job with those gates | +| The job is replaced in `jobs.yaml` | The job as `jobs.yaml` has it | + +An old install that keeps every job in `jobs.yaml`, with no `system.yaml`, is not changed. `inbox-processor` and `task-planner` have their gates in the file, and an entry of theirs without one runs on every schedule. Only the scheduler gives the default: `nerve config validate` and `nerve doctor` read the file as it is. + `machine-maintenance` is on by default and optional in both modes: switch it off on the Automation step of the setup page, or with a job of the same id in `jobs.yaml`. Its prompt is general on purpose: the agent learns what each machine has on each run. To keep an item out of its reach, write it in `TOOLS.md` or ask the agent to remember it; the job reads both before it acts. Both skill jobs use `source="skill-extractor"` or `source="skill-reviser"` on created tasks. When their plans are approved, an `impl-` session is spawned like for any plan; its prompt tells it to apply the plan (a full SKILL.md file) with `skill_create` or `skill_update` and then close the task with `task_done`. @@ -547,6 +651,7 @@ If a job should have fired while the server was down, it fires **once** on start - **First-ever run**: No catch-up (no history to compare against). - **Multiple missed fires**: Coalesced into a single catch-up run. - **Catch-up runs concurrently**: All overdue jobs fire in parallel, in the background (doesn't block startup). +- **Run gates apply**: a job whose gates are not satisfied is not caught up. A run that a gate skipped leaves no row, so a gated job can count as overdue at every start; its gates decide. They are asked about the moment the catch-up began, so `agent_activity` counts only work recorded before it. ### Opting out @@ -586,7 +691,9 @@ them start late or be dropped as [late runs](#late-runs): start 15 seconds after the resume is noticed, because the network is often not back yet the moment a laptop wakes, and a run that starts without it fails. Run gates and `lock` apply as for a scheduled run, and the run is logged as - usual. + usual. The gates are asked about the time the run came due: `agent_activity` + leaves out what was recorded since, such as the message that woke the + machine. - Jobs with `catchup: false` only move on; the run each one skipped is recorded in `cron_logs` with status `missed`. A run that the job's gates would have skipped, going by their [forecast](#forecasts) for the time it came due, is @@ -688,10 +795,11 @@ synced by hand (`POST /api/sources/{name}/sync`, `POST /api/sources/sync-all`). Not every change is announced. A wakeup that a session records during its turn, tasks created, moved or edited from a chat turn, the web UI or the API (which -can change what a `tasks` gate forecasts), and a message a channel buffers for -its `:observed` drain (which makes that drain count) count only from -the next time the schedule is read. Code that keeps the answer should therefore -also refresh it periodically, and when a turn ends. +can change what a `tasks` gate forecasts), a turn or a stored message (which +can change what an `agent_activity` gate forecasts), and a message a channel +buffers for its `:observed` drain (which makes that drain count) count +only from the next time the schedule is read. Code that keeps the answer should +therefore also refresh it periodically, and when a turn ends. ## Source Runners diff --git a/docs/personal-best-practices.md b/docs/personal-best-practices.md index 2bf5ebb7c..cbb0150de 100644 --- a/docs/personal-best-practices.md +++ b/docs/personal-best-practices.md @@ -90,7 +90,7 @@ On an instance whose memory is Loom (`memory.backend: loom`), this tier is Loom ### Housekeeping -The built-in `memory-maintenance` cron ([cron.md](cron.md#built-in-system-crons)) runs daily and silently over long-term memory: it deduplicates, prunes and rewords with `memory_records_by_date`, `memory_update` and `memory_delete`. On a Loom instance it runs Loom's integrity report on the agent's home namespace instead and leaves deduplication to Loom. It does not touch `MEMORY.md` or the context files. Their housekeeping (stubs over the cap, tag review, eviction, archiving dormant files) is yours: do it by hand in a quiet turn, or add a daily isolated job to `config/cron/jobs.yaml` that notifies only when it needs a ruling. A maintenance job that reports routine work teaches the human to ignore notifications. +The built-in `memory-maintenance` cron ([cron.md](cron.md#built-in-system-crons)) runs silently over long-term memory, daily while the agent is in use (it waits while the agent has done nothing since its last run): it deduplicates, prunes and rewords with `memory_records_by_date`, `memory_update` and `memory_delete`. On a Loom instance it runs Loom's integrity report on the agent's home namespace instead and leaves deduplication to Loom. It does not touch `MEMORY.md` or the context files. Their housekeeping (stubs over the cap, tag review, eviction, archiving dormant files) is yours: do it by hand in a quiet turn, or add a daily isolated job to `config/cron/jobs.yaml` that notifies only when it needs a ruling. A maintenance job that reports routine work teaches the human to ignore notifications. ## Working Discipline diff --git a/docs/worker-best-practices.md b/docs/worker-best-practices.md index 373ed49a2..6214214bf 100644 --- a/docs/worker-best-practices.md +++ b/docs/worker-best-practices.md @@ -30,8 +30,8 @@ Each job plays one role, and each role moves tasks in one direction: | Work | Claims one actionable task, moves it to `in_progress`, and ends the run with it in a truthful status | `tasks` on `pending` and the tag it owns | `isolated` | | Monitor | Watches a waiting status and returns the task when the external event happens | `tasks` on that status, plus `github_pr_activity` for PRs | `persistent` or `isolated` | | Planner (`task-planner`) | Proposes a plan for an open task that has none. Ships disabled for workers; onboarding enables it when the worker plans through the approval loop | `tasks` on `pending`, and the worker's tag once overridden | `persistent` | -| Skills (`skill-extractor`, `skill-reviser`) | Propose new or revised skills, each as a task with a plan | None | `persistent` | -| Maintenance (`memory-maintenance`, `machine-maintenance`) | Tidies long-term memory, and once a week the machine (old caches, Docker leftovers, worktrees of merged pull requests, leftover services); asks before it removes or stops anything that can hold data; moves no tasks | None | `isolated` | +| Skills (`skill-extractor`, `skill-reviser`) | Propose new or revised skills, each as a task with a plan | `agent_activity` for `skill-extractor`; none for `skill-reviser` | `persistent` | +| Maintenance (`memory-maintenance`, `machine-maintenance`) | Tidies long-term memory, and once a week the machine (old caches, Docker leftovers, worktrees of merged pull requests, leftover services); asks before it removes or stops anything that can hold data; moves no tasks | `agent_activity` for `memory-maintenance`; none for `machine-maintenance` | `isolated` | The last three ship in `system.yaml` ([cron.md](cron.md#built-in-system-crons)); scan, work and monitor jobs are yours, in `config/cron/jobs.yaml`. `task-planner` ships disabled for workers. When the worker should plan through the approval loop (`task-planner` proposes, the operator approves, and an implementation session does the work), onboarding enables it by overriding the job by id in `jobs.yaml`, with a prompt and a `tasks` gate that name the worker's tag; a job in `jobs.yaml` replaces the `system.yaml` job with the same id. Add a work job when `TASK.md` § Approval Required grants autonomy for routine items. When both run, give each its own tasks by tag. @@ -72,10 +72,11 @@ The same machine as a transition table, in the form `TASK.md` § Actions uses (` ### Gates keep jobs idle until there is work -A run gate is checked right before a job fires; when it is not satisfied, no session starts ([cron.md](cron.md#run-gates)). Three gate types ship with Nerve: +A run gate is checked right before a job fires; when it is not satisfied, no session starts ([cron.md](cron.md#run-gates)). Four gate types ship with Nerve: - `tasks`: enough tasks match a status, or a list of statuses, and an optional tag. A work job gates on the queue it owns, a monitor on the waiting status it watches, the planner on `pending`. No built-in job gates on `backlog`. - `messages`: a sync source has unread messages for a consumer. It drives inbox work, if the worker has sources. +- `agent_activity`: the agent has worked since the job's last run, not counting the job's own runs, the runs of the jobs in `ignore_jobs`, or the runs of other jobs with this gate. The built-in `memory-maintenance` and `skill-extractor` have it by default, so neither runs while the agent has done nothing since its last run. A worker with a scan job works at every scan, so both go on running there. - `github_pr_activity`: something actionable changed on an author's open PRs (a merge or close, a new commit, a review, a CI transition, a human comment); `force_run_after_hours` fires it anyway after a quiet spell. Gates in one list must all pass and are checked in order, so put the cheap `tasks` gate before the network-backed `github_pr_activity`. A gate that errors while checking lets the run proceed; a gate type that does not exist keeps the job from loading. Drop-in plugins in `config/cron/gates/` add conditions the built-in types do not cover. Scan jobs stay ungated: their job is to find work nobody has recorded yet. diff --git a/docs/worker-guide.md b/docs/worker-guide.md index 62b15a411..01c8e6aab 100644 --- a/docs/worker-guide.md +++ b/docs/worker-guide.md @@ -247,7 +247,7 @@ Workers create their own skills during onboarding and refine them over time. - Skills can include reference docs (`references/`), scripts (`scripts/`), and assets (`assets/`) **Automated skill lifecycle:** -- `skill-extractor` (every 12h) — watches for repeated workflows in conversations and completed tasks, proposes new skills via task+plan +- `skill-extractor` (every 12h, when the agent has worked since its last run) — watches for repeated workflows in conversations and completed tasks, proposes new skills via task+plan - `skill-reviser` (weekly) — reviews existing skills for accuracy, completeness, and quality, proposes revisions When a skill-related plan is approved, an `impl-` session is spawned like for any plan; its prompt tells it to apply the plan (a full SKILL.md) with `skill_create` or `skill_update` and close the task with `task_done`. diff --git a/nerve/cron/gate_plugins.py b/nerve/cron/gate_plugins.py index eb172b4ac..f4a9162e3 100644 --- a/nerve/cron/gate_plugins.py +++ b/nerve/cron/gate_plugins.py @@ -27,7 +27,7 @@ def __init__(self, min_age_minutes: int = 30): self.min_age_minutes = min_age_minutes async def is_satisfied(self, ctx: GateContext) -> bool: - ... # ctx gives {job_id, db} — DB-only + ... # ctx gives {job_id, db, jobs, as_of} async def forecast(self, ctx: GateContext, at: datetime) -> bool: ... # a task too young now may be old enough at `at` @@ -47,9 +47,10 @@ def from_config(cls, spec: dict) -> "StaleTasksGate": **first loaded (filename-sorted) wins**. * Any import/exec error in a plugin file is logged (naming the file) and the file is skipped; the remaining files still load. -* A gate gets only ``GateContext{job_id, db}`` (DB-only). A liveness/registry - based gate is out of scope for this loader — it would need the context - widened, a separate change. +* A gate gets only ``GateContext{job_id, db, jobs, as_of}``: the database, + the loaded jobs by id, and for a catch-up the time its run came due. A + liveness/registry based gate is out of scope for this loader — it would + need the context widened, a separate change. * **A gate that does not define ``forecast`` has its check run on every schedule read, so that check must be side-effect-free.** A gate also answers ``forecast(ctx, at)``: what it would decide for a run at ``at``, judged from diff --git a/nerve/cron/gates.py b/nerve/cron/gates.py index 327fae5c7..697a08ee7 100644 --- a/nerve/cron/gates.py +++ b/nerve/cron/gates.py @@ -40,7 +40,8 @@ import logging import time from abc import ABC, abstractmethod -from dataclasses import dataclass +from collections.abc import Mapping +from dataclasses import dataclass, field from datetime import datetime, timedelta, timezone from pathlib import Path from typing import TYPE_CHECKING, ClassVar @@ -48,6 +49,7 @@ from nerve import paths if TYPE_CHECKING: + from nerve.cron.jobs import CronJob from nerve.db import Database logger = logging.getLogger(__name__) @@ -68,6 +70,16 @@ class GateContext: job_id: str db: "Database" + #: The jobs the scheduler has loaded, by id, disabled ones too, for a + #: gate whose answer depends on how the other jobs are set up. Empty when + #: the caller has none to give. + jobs: "Mapping[str, CronJob]" = field(default_factory=dict) + #: The time the check is made for, when that is not now (timezone-aware). + #: A catch-up after a pause asks about the time the run it stands in for + #: came due, and the catch-up at startup about its own start, so that a + #: gate which reads a record of what happened can leave out what was + #: recorded since: the message that woke the machine, above all. + as_of: datetime | None = None @dataclass @@ -96,6 +108,13 @@ class CronGate(ABC): #: out-of-tree gate nothing has read. spec_keys: ClassVar[frozenset[str]] = frozenset() + #: Whether a run that a person starts by hand ("Run now", ``nerve cron + #: ``, the trigger API) is checked too. True for a gate that says + #: whether the job has anything to act on: with no pending task there is + #: nothing to plan, whoever asks. False for a gate that only estimates + #: whether a run is worth it, where the person who asks may know better. + checked_on_manual_runs: ClassVar[bool] = True + @abstractmethod async def is_satisfied(self, ctx: GateContext) -> bool: """Return True if the job is allowed to run under this condition.""" @@ -339,6 +358,132 @@ def from_config(cls, spec: dict) -> "TasksGate": return cls(targets=targets, tag=tag, min_count=min_count) +#: The most hours an ``agent_activity`` gate looks back before the job's +#: last run. A year is far more than any use has, and keeps the number one +#: that a time can be moved by. +_MAX_OVERLAP_HOURS = 24 * 366 + + +class AgentActivityGate(CronGate): + """Satisfied when the agent has worked since the job last ran. + + The motivating case: a job that looks after the agent's own work — a + memory check, a search for repeated workflows — has nothing to do while + the agent has done nothing, so it runs only after there was work. + + Config:: + + type: agent_activity + ignore_jobs: [weekly-cleanup] # jobs whose runs are not work + overlap_hours: 24 # also count this long before the last run + + The agent has worked when, since the job's last successful run started, a + turn ended or a message was stored in a session + (:meth:`nerve.db.cron.CronStore.has_activity_since_cron_run` says exactly + what counts). A job that has never succeeded runs as soon as there is any + work at all, and a run that failed does not move the start. + + Not counted: + + * the job's own runs, so a run never causes the next one; + * the runs of the jobs in ``ignore_jobs``; + * the runs of every other job that has this gate. Such a job runs only + because the agent worked, so its run is not work of its own: if two of + them counted each other's runs, each run of one would start the other, + forever. The gate knows the other jobs from ``GateContext.jobs``, and + finds this gate only among a job's own gates: a plugin gate that calls + this one must pass the context it was given, and its job must be named + in ``ignore_jobs`` by the jobs that should not count it; + * work that leaves no message and no turn, such as a memory written + through the MCP endpoint; + * for a catch-up, what was recorded after the run it stands in for came + due (``GateContext.as_of``): the message that woke the machine is work + for the job's next scheduled time, not for a run that came due while + nothing had happened. + + ``overlap_hours`` is for a job that looks at a fixed period before its + run rather than at everything since its last run (the records of the day + before, say): what the agent did shortly before a run is not in that + run's period yet, so the job must run once more for it. With an overlap, + the work of that many hours before the last run counts again. + + The gate estimates whether a run is worth it; it does not say whether + the job can run. So a run that a person starts by hand is not checked: + the person may know of work the gate cannot see. + + The check reads the database and nothing else, so the default forecast — + the check, made now — holds until the agent works or the job runs. A + turn and a run are followed by a new forecast; other work (a + conversation taken in, say) is seen at the next read of the schedule. + """ + + type = "agent_activity" + spec_keys = frozenset({"ignore_jobs", "overlap_hours"}) + checked_on_manual_runs = False + + def __init__( + self, ignore_jobs: list[str] | None = None, overlap_hours: float = 0.0, + ) -> None: + self.ignore_jobs = list(dict.fromkeys(ignore_jobs or [])) + self.overlap_hours = overlap_hours + + async def is_satisfied(self, ctx: GateContext) -> bool: + gated_alike = [ + job_id for job_id, job in ctx.jobs.items() + if any(isinstance(gate, AgentActivityGate) for gate in job.gates) + ] + return await ctx.db.has_activity_since_cron_run( + ctx.job_id, + ignore_jobs=[*self.ignore_jobs, *gated_alike], + overlap_seconds=round(self.overlap_hours * 3600), + until=ctx.as_of, + ) + + def describe(self) -> str: + since = "this job's last run" + if self.overlap_hours: + since = f"{self.overlap_hours:g}h before {since}" + ignored = ( + f", not counting the runs of {', '.join(self.ignore_jobs)}" + if self.ignore_jobs else "" + ) + return f"agent activity since {since}{ignored}" + + @classmethod + def from_config(cls, spec: dict) -> "AgentActivityGate": + # As for github_pr_activity's ignore_actors: only a bare key means + # "not set". A value of the wrong shape is refused, not read as an + # empty list — that would count the runs the spec meant to leave out. + ignore = spec.get("ignore_jobs") + if ignore is None: + ignore = [] + elif isinstance(ignore, str): + ignore = [ignore] + if not isinstance(ignore, list) or not all( + isinstance(job, str) and job.strip() for job in ignore + ): + raise GateConfigError( + "'agent_activity' gate 'ignore_jobs' must be a job id or a " + f"list of job ids, got {ignore!r}" + ) + hours = spec.get("overlap_hours") + if hours is None: + hours = 0 + # The range is checked on the number as written: a float made of an + # integer too large for one raises instead of being refused. NaN + # fails both comparisons. + if ( + isinstance(hours, bool) + or not isinstance(hours, (int, float)) + or not 0 <= hours <= _MAX_OVERLAP_HOURS + ): + raise GateConfigError( + "'agent_activity' gate 'overlap_hours' must be a number of " + f"hours from 0 to {_MAX_OVERLAP_HOURS}, got {hours!r}" + ) + return cls(ignore_jobs=[job.strip() for job in ignore], overlap_hours=float(hours)) + + class GitHubPrActivityGate(CronGate): """Satisfied only when an author's open PRs show *new* activity. @@ -685,6 +830,7 @@ def github_is_hosted() -> bool: GATE_REGISTRY: dict[str, type[CronGate]] = { MessagesGate.type: MessagesGate, TasksGate.type: TasksGate, + AgentActivityGate.type: AgentActivityGate, GitHubPrActivityGate.type: GitHubPrActivityGate, } diff --git a/nerve/cron/jobs.py b/nerve/cron/jobs.py index ba696f107..7b96208bb 100644 --- a/nerve/cron/jobs.py +++ b/nerve/cron/jobs.py @@ -7,6 +7,7 @@ import logging import math +from collections.abc import Callable from dataclasses import dataclass, field from pathlib import Path from typing import TYPE_CHECKING, Any @@ -286,12 +287,31 @@ def merge_jobs(system_jobs: list[CronJob], user_jobs: list[CronJob]) -> list[Cro return list(jobs_by_id.values()) +def _with_default_run_if( + item: Any, default_run_if: Callable[[str], list[dict]] | None, +) -> Any: + """*item* with the default gates of its job as ``run_if``, when it states + none; otherwise *item* itself. + + A ``run_if`` or ``skip_when_idle`` key that is there, even bare or empty, + is the file's answer: only an entry with neither gets a default. + """ + if default_run_if is None or not isinstance(item, dict): + return item + if "run_if" in item or "skip_when_idle" in item: + return item + job_id = item.get("id") + specs = default_run_if(job_id) if isinstance(job_id, str) else None + return {**item, "run_if": specs} if specs else item + + def load_jobs( jobs_file: Path, strict: bool = False, errors: list[str] | None = None, *, build_gates: bool = True, + default_run_if: Callable[[str], list[dict]] | None = None, ) -> list[CronJob]: """Load cron jobs from a YAML file. @@ -311,6 +331,14 @@ def load_jobs( ``build_gates=False`` is passed straight to :meth:`CronJob.from_dict`; see there for who wants it and why. + + ``default_run_if`` gives the run gates of a job whose entry states none. + It is called with the id of each entry that has neither a ``run_if`` nor + a ``skip_when_idle`` key, and a non-empty answer becomes the entry's + ``run_if``; each job that gets one is logged. A default that does not + build is logged as an error and left out: the entry loads as the file has + it. The scheduler passes it for ``system.yaml``, so that a built-in job + has the gates it ships with without the file stating them. """ if not jobs_file.exists(): logger.info("No cron jobs file at %s", jobs_file) @@ -339,9 +367,35 @@ def load_jobs( jobs = [] for item in jobs_data: try: - jobs.append(CronJob.from_dict( - item, base_dir=jobs_file.parent, gates=build_gates, - )) + entry = _with_default_run_if(item, default_run_if) + try: + job = CronJob.from_dict( + entry, base_dir=jobs_file.parent, gates=build_gates, + ) + except (KeyError, TypeError, ValueError) as e: + if entry is item: + raise + # The default may be what does not build, and the file never + # stated it: the entry loads as written. If the entry itself + # is at fault, this raises what it would have raised. + job = CronJob.from_dict( + item, base_dir=jobs_file.parent, gates=build_gates, + ) + logger.error( + "Cron job %s: its default gates do not build (%s). It " + "runs as %s has it, on every schedule.", + job.id, e, jobs_file.name, + ) + entry = item + jobs.append(job) + if entry is not item: + logger.info( + "Cron job %s: %s states no run_if for it, so it runs " + "under its default gates (%s). Write `run_if: []` to run " + "it on every schedule.", + job.id, jobs_file.name, + "; ".join(gate.describe() for gate in job.gates) or "not built", + ) except (KeyError, TypeError, ValueError) as e: message = f"Invalid cron job in {jobs_file}: {item!r} — {e}" if errors is not None: diff --git a/nerve/cron/service.py b/nerve/cron/service.py index 38caf273d..46e179a7b 100644 --- a/nerve/cron/service.py +++ b/nerve/cron/service.py @@ -45,6 +45,7 @@ CRON_FILE_HAND_EDITED, CRON_FILE_READ_ONLY, CRON_JOB_MISSING, + default_run_if, plan_cron_job_enabled, publish_text, ) @@ -53,7 +54,7 @@ if TYPE_CHECKING: from collections.abc import Awaitable, Callable, Iterator - from nerve.cron.gates import CronGate + from nerve.cron.gates import CronGate, GateContext from nerve.sources.runner import SourceRunner logger = logging.getLogger(__name__) @@ -534,6 +535,9 @@ def __init__(self, config: NerveConfig, engine: AgentEngine, db: Database): self._clock_sample: tuple[float, float] | None = None self._clock_watch_task: asyncio.Task | None = None self._catchup_due: dict[str, int] = {} + # When each of those jobs came due during the pause: the time its + # catch-up asks the gates about. + self._catchup_due_at: dict[str, datetime] = {} self._catchup_lock = asyncio.Lock() self._catchup_tasks: set[asyncio.Task] = set() # Called whenever the schedule may have changed (see @@ -1268,6 +1272,7 @@ def _rearm_after_pause( if catchup: if self._may_start(sched_job.id, "catchup"): due.append(sched_job.id) + self._catchup_due_at[sched_job.id] = missed elif func == self._run_job_wrapper: asyncio.create_task( self._record_paused_run(sched_job.args[0], missed), @@ -1309,10 +1314,10 @@ async def _record_paused_run(self, job: CronJob, due: datetime) -> None: forecast that raises lets the run through, as a check that raises does. """ if job.gates: - from nerve.cron.gates import GateContext, forecast_gates + from nerve.cron.gates import forecast_gates decision = await forecast_gates( - job.gates, GateContext(job_id=job.id, db=self.db), due, + job.gates, self._gate_context(job, as_of=due), due, ) if not decision.should_run: logger.debug( @@ -1334,7 +1339,9 @@ async def _catch_up_after_pause(self) -> None: then looks each job up as the scheduler holds it now: a job removed meanwhile does not run, one redefined runs as it is now, and one that has started a run since it was queued is not caught up on top of it. - Gates, ``lock`` and the run log apply as for a scheduled run. + Gates, ``lock`` and the run log apply as for a scheduled run; the + gates are asked about the time the job came due (see + :attr:`nerve.cron.gates.GateContext.as_of`). """ await asyncio.sleep(_CATCHUP_SETTLE_SECONDS) async with self._catchup_lock: @@ -1354,6 +1361,7 @@ async def _catch_up_after_pause(self) -> None: def _catch_up_run(self, job_id: str, started: int) -> Awaitable[None] | None: """The catch-up run of *job_id*, which had started *started* runs when it was queued — or ``None`` when it is gone or has run since.""" + came_due = self._catchup_due_at.pop(job_id, None) sched_job = self.scheduler.get_job(job_id) if sched_job is None: logger.info("Not catching up %s: it is no longer scheduled", job_id) @@ -1362,7 +1370,9 @@ def _catch_up_run(self, job_id: str, started: int) -> Awaitable[None] | None: logger.info("Not catching up %s: it has run since the pause", job_id) return None if sched_job.func == self._run_job_wrapper: - return self._run_job_wrapper(sched_job.args[0], origin="catchup") + return self._run_job_wrapper( + sched_job.args[0], origin="catchup", as_of=came_due, + ) if sched_job.func == self._run_source_wrapper: return self._run_source_wrapper(sched_job.args[0], origin="catchup") if sched_job.func == self._cleanup_expired: @@ -1438,7 +1448,10 @@ async def _catchup_missed_jobs(self) -> None: ) async with self._catchup_lock: await asyncio.gather( - *(self._run_job_wrapper(job, origin="catchup") for job in overdue), + *( + self._run_job_wrapper(job, origin="catchup", as_of=now) + for job in overdue + ), ) @staticmethod @@ -1701,8 +1714,18 @@ def _load_merged_jobs(self, strict: bool = False) -> list[CronJob]: system_file = self.config.cron.system_file jobs_file = self.config.cron.jobs_file - system_jobs = load_jobs(system_file, strict=strict) user_jobs = load_jobs(jobs_file, strict=strict) + # A built-in job has the gates it ships with unless its entry states + # some. A job that jobs.yaml replaces is not loaded for running, so + # it is given none (and no line in the log says it runs under them). + replaced = {j.id for j in user_jobs} + backend = self.config.memory.backend + system_jobs = load_jobs( + system_file, strict=strict, + default_run_if=lambda job_id: ( + [] if job_id in replaced else default_run_if(job_id, backend) + ), + ) if not system_jobs and user_jobs: # Backward compat: old install with everything in jobs.yaml @@ -1741,13 +1764,15 @@ def _load_merged_jobs(self, strict: bool = False) -> list[CronJob]: async def _run_job_wrapper( self, job: CronJob, attribution: Attribution | None = None, - *, origin: RunOrigin = "schedule", + *, origin: RunOrigin = "schedule", as_of: datetime | None = None, ) -> None: """Wrapper to run a cron job with logging and optional lock. ``attribution`` names who started the run and its cause. ``None`` (the scheduler's call) is the instance's own run. ``origin`` says how - the run was started; the scheduler's call is ``"schedule"``. A run that + the run was started; the scheduler's call is ``"schedule"``. ``as_of`` + is the time a catch-up asks the gates about (see + :meth:`_run_job_inner`). A run that would overlap another run of the job is skipped as :meth:`_may_start` says; otherwise it counts as in progress until it ends. @@ -1762,9 +1787,13 @@ async def _run_job_wrapper( if job.lock: lock = self._job_locks.setdefault(job.id, asyncio.Lock()) async with lock: - await self._run_job_inner(job, attribution) + await self._run_job_inner( + job, attribution, origin=origin, as_of=as_of, + ) else: - await self._run_job_inner(job, attribution) + await self._run_job_inner( + job, attribution, origin=origin, as_of=as_of, + ) finally: self._schedule_changed() @@ -1801,23 +1830,42 @@ def _may_start(self, job_id: str, origin: RunOrigin, *, lock: bool = False) -> b return False return True + def _gate_context(self, job: CronJob, as_of: datetime | None = None) -> GateContext: + """What the gates of *job* are checked and forecast with: the loaded + jobs, and the time the check is made for when that is not now.""" + from nerve.cron.gates import GateContext + + return GateContext( + job_id=job.id, db=self.db, jobs={j.id: j for j in self._jobs}, + as_of=as_of, + ) + async def _run_job_inner( self, job: CronJob, attribution: Attribution | None = None, + *, origin: RunOrigin = "schedule", as_of: datetime | None = None, ) -> None: - """Inner implementation of job execution.""" + """Inner implementation of job execution. + + ``origin`` says how the run was started: a manual run is checked only + by the gates that hold for one. ``as_of`` is the time a catch-up asks + the gates about: when the run it stands in for came due. + """ # The engine is draining before a shutdown. Write no log row, so the # catch-up after the restart sees the job as overdue. if not self.engine.accepting_turns: logger.info("Skipping cron job %s: Nerve is draining", job.id) return - # Pre-check: skip if any configured run gate is unsatisfied. - if job.gates: - from nerve.cron.gates import GateContext, evaluate_gates + # Pre-check: skip if any configured run gate is unsatisfied. A run + # that a person starts is checked only by the gates that say whether + # the job has anything to act on (CronGate.checked_on_manual_runs). + gates = job.gates + if origin == "manual": + gates = [gate for gate in gates if gate.checked_on_manual_runs] + if gates: + from nerve.cron.gates import evaluate_gates - decision = await evaluate_gates( - job.gates, GateContext(job_id=job.id, db=self.db), - ) + decision = await evaluate_gates(gates, self._gate_context(job, as_of)) if not decision.should_run: logger.info("Skipping cron job %s: %s", job.id, decision.reason) return @@ -2564,7 +2612,9 @@ def add_schedule_listener(self, callback: Callable[[], object]) -> None: (``ScheduleWakeup``) does so inside its turn, which has no reference to this service; tasks created, moved or edited from a chat turn, the web UI or the API change what a ``tasks`` gate forecasts without any - call here; and a message a channel buffers for its + call here; a turn, or a message stored in a session, changes what an + ``agent_activity`` gate forecasts without one; and a message a channel + buffers for its ``:observed`` drain makes that drain count (see :meth:`_undrained`) without one. Those count from the next :meth:`next_scheduled` on, so a caller that keeps the answer should @@ -2726,13 +2776,13 @@ async def _forecast_job(self, sched_job, limit: datetime) -> _Forecast: are tried, asking only the gates with a forecast of their own: a gate's real check never runs more than once per forecast. """ - from nerve.cron.gates import GateContext, forecast_gates, has_own_forecast + from nerve.cron.gates import forecast_gates, has_own_forecast job = sched_job.args[0] # the job the scheduler will run, gates included next_run = getattr(sched_job, "next_run_time", None) if next_run is None or not job.gates: return _Forecast("cron", next_run, runs=True, runs_at=next_run) - ctx = GateContext(job_id=job.id, db=self.db) + ctx = self._gate_context(job) steady = [gate for gate in job.gates if not has_own_forecast(gate)] timed = [gate for gate in job.gates if has_own_forecast(gate)] decision = await forecast_gates(steady, ctx, next_run) diff --git a/nerve/db/cron.py b/nerve/db/cron.py index 9232b72c1..9f9ac6fab 100644 --- a/nerve/db/cron.py +++ b/nerve/db/cron.py @@ -2,6 +2,10 @@ from __future__ import annotations +from collections.abc import Iterable +from datetime import UTC, datetime + +from nerve.db.usage import _CRON_RUN_STAMP_GLOB from nerve.identity import UNATTRIBUTED, Attribution from nerve.utils.time import utc_now_iso @@ -16,6 +20,70 @@ _COMPLETED = "status IN ('success', 'error')" +#: What follows ``cron::`` in the id of a session of one run or one +#: generation of the job: the run id that ``new_cron_run_id`` makes, and the +#: older one without the random tail. +_RUN_ID_GLOBS = ( + _CRON_RUN_STAMP_GLOB + "-" + "[0-9a-f]" * 8, + _CRON_RUN_STAMP_GLOB, +) + + +def _cron_session_of(expr: str, job_id: str) -> tuple[str, list]: + """SQL, with its parameters, that is true when *expr* is the id of a cron + session of *job_id*: ``cron:`` or ``cron::``. + + The job id is compared as text, never as a pattern, and what follows it + must be a run id: ``cron:a:b`` is the session of the job ``a:b``, not a + run of ``a``. + """ + prefix = f"cron:{job_id}:" + rest = f"substr({expr}, ?)" + return ( + f"({expr} = ? OR (substr({expr}, 1, ?) = ? " + f"AND ({rest} GLOB ? OR {rest} GLOB ?)))", + [ + f"cron:{job_id}", len(prefix), prefix, + len(prefix) + 1, _RUN_ID_GLOBS[0], len(prefix) + 1, _RUN_ID_GLOBS[1], + ], + ) + + +def _sessions_of_jobs(column: str, job_ids: list[str]) -> tuple[str, list]: + """SQL, with its parameters, that is true when *column* holds the id of a + session of one of *job_ids*. *column* names its table, since the SQL + refers to it from a subquery. + + A session of a job is a cron session of it, or the session of a workflow + run that the job started (``created_by`` is ``cron:``), that one of + its cron sessions started (``session:``), or that is a leg of + a review loop which one of its cron sessions started + (``review-loop:``, the loop's ``session_id``). A workflow run + can start neither, so there is nothing further down. + """ + direct, started, params, started_params = [], [], [], [] + for job_id in job_ids: + sql, sql_params = _cron_session_of(column, job_id) + direct.append(sql) + params += sql_params + by_session, by_session_params = _cron_session_of("substr(w.created_by, 9)", job_id) + by_loop, by_loop_params = _cron_session_of("l.session_id", job_id) + started.append( + "w.created_by = ? " + f"OR (substr(w.created_by, 1, 8) = 'session:' AND {by_session}) " + "OR (substr(w.created_by, 1, 12) = 'review-loop:' AND EXISTS (" + "SELECT 1 FROM review_loops l " + f"WHERE l.id = substr(w.created_by, 13) AND {by_loop}))" + ) + started_params += [f"cron:{job_id}", *by_session_params, *by_loop_params] + return ( + f"{' OR '.join(direct)} OR EXISTS (" + f"SELECT 1 FROM workflow_runs w WHERE w.session_id = {column} " + f"AND ({' OR '.join(started)}))", + params + started_params, + ) + + def _ranked(where: str = "") -> str: """The rows of ``cron_logs`` (with ``where``), each with ``row_rank``: 1 for the latest row of its job, by start and then id, as the history lists them.""" @@ -212,6 +280,100 @@ async def get_last_successful_cron_run(self, job_id: str) -> dict | None: row = await cursor.fetchone() return dict(row) if row else None + async def has_activity_since_cron_run( + self, + job_id: str, + *, + ignore_jobs: Iterable[str] = (), + overlap_seconds: int = 0, + until: datetime | None = None, + ) -> bool: + """Whether the agent has worked since the job's last successful run + started, not counting the work of the job itself or of a job of + ``ignore_jobs``. + + Two records count as work, each in a session that belongs to none of + those jobs: + + * a row of ``session_usage``: a turn ended and reported its usage, + whatever started it (a chat, a channel message, a wakeup, a hook, + another cron job, a workflow run); + * a session whose ``updated_at`` moved: a message was stored in it. + A turn that somebody asked for stores what it was asked when it + starts, so such a turn counts from then on, and also when it is + stopped or fails before it reports any usage; so does a + conversation taken in from an external agent. A turn that the + agent starts for itself (a wakeup, a continuation after background + work) stores its messages without moving ``updated_at``: it counts + by its usage row, when it ends with one. + + ``overlap_seconds`` moves the start of the window back by that much, + so what happened in that time before the last run counts again. A job + with no successful run in ``cron_logs`` has no window start: any work + counts. The cleanup keeps the latest finished run of each job, so the + start outlives the 14 days unless a later run failed. + + ``until`` ends the window: only work recorded before it counts. A + catch-up asks about the time the run it stands in for came due. It + must be timezone-aware: a naive time raises ``ValueError``. A session + has one ``updated_at``, the time of its latest message, so a message + stored after ``until`` hides the session's earlier ones from this + record; a turn that reported its usage before ``until`` still counts. + + The sessions of a job are its cron sessions, ``cron:`` and + ``cron::``, and the sessions of the workflow runs that + the job or one of those sessions started, a review loop's legs + included (:func:`_sessions_of_jobs`). + """ + async with self.db.execute( + "SELECT datetime(MAX(datetime(started_at)), ?) FROM cron_logs " + "WHERE job_id = ? AND status = 'success'", + (f"-{max(0, int(overlap_seconds))} seconds", job_id), + ) as cursor: + row = await cursor.fetchone() + since = row[0] if row else None + if until is not None and until.utcoffset() is None: + raise ValueError(f"has_activity_since_cron_run() needs a timezone-aware until, not {until!r}") + before = until.astimezone(UTC).strftime("%Y-%m-%d %H:%M:%S") if until else None + + jobs = list(dict.fromkeys((job_id, *ignore_jobs))) + # Both columns hold UTC. session_usage.created_at is SQLite's + # CURRENT_TIMESTAMP text, the form datetime() returns, so it compares + # as text. sessions.updated_at is ISO 8601 where a message moved it + # and CURRENT_TIMESTAMP text where none has: the day picks the rows + # through the index, and datetime() compares what either form says. + checks = ( + ("session_usage.session_id", "created_at >= ?", [since], "created_at < ?", ""), + ( + "sessions.id", + "updated_at >= ? AND datetime(updated_at) >= ?", + [since[:10], since] if since else [], + "datetime(updated_at) < ?", + "COALESCE(message_count, 0) > 0", + ), + ) + for column, window, window_params, end, extra in checks: + clauses, params = [], [] + if since is not None: + clauses.append(window) + params += window_params + if before is not None: + clauses.append(end) + params.append(before) + if extra: + clauses.append(extra) + mine, mine_params = _sessions_of_jobs(column, jobs) + clauses.append(f"NOT ({mine})") + params += mine_params + table = column.partition(".")[0] + async with self.db.execute( + f"SELECT 1 FROM {table} WHERE {' AND '.join(clauses)} LIMIT 1", + tuple(params), + ) as cursor: + if await cursor.fetchone() is not None: + return True + return False + async def get_latest_cron_session_id(self, job_id: str) -> str | None: """Return the most recently active session id for a cron job. diff --git a/nerve/setup_writer.py b/nerve/setup_writer.py index c5b7991a5..dfc01f4e4 100644 --- a/nerve/setup_writer.py +++ b/nerve/setup_writer.py @@ -1326,6 +1326,60 @@ def _cron_description(cron: dict[str, Any], memory_backend: str) -> str: return cron["description"] +# The built-in jobs that look after the agent's own work. A run of one is no +# reason for another to run, so the default gates below leave them all out. +MAINTENANCE_CRON_IDS = ( + "memory-maintenance", "skill-extractor", "skill-reviser", "machine-maintenance", +) + + +def _other_maintenance_crons(job_id: str) -> list[str]: + """The maintenance jobs besides *job_id*, whose own runs never count.""" + return [cron_id for cron_id in MAINTENANCE_CRON_IDS if cron_id != job_id] + + +# Run gates that a built-in job has without ``system.yaml`` stating them: the +# scheduler gives them to an entry of the job that has no ``run_if`` +# (:func:`default_run_if`). Unlike the gates of inbox-processor and +# task-planner, they are not written into the file: +# +# * nothing writes ``system.yaml`` again when Nerve is updated, so a gate +# written there never changes, and an install from before it never has it; +# * a Nerve that predates a gate type refuses a job that names it, so a file +# with the gate could not go back to the release before. +_DEFAULT_RUN_IF: dict[str, list[dict[str, Any]]] = { + # Nothing to clean up while the agent has done nothing since the last + # run. The prompt reviews the records of the day before the run, so what + # the agent did shortly before a run is left to the next one: with the + # overlap, the job runs at its next two times after the last activity. + "memory-maintenance": [{ + "type": "agent_activity", + "ignore_jobs": _other_maintenance_crons("memory-maintenance"), + "overlap_hours": 24, + }], + # Recent activity is all the job looks at: no run while there was none. + "skill-extractor": [{ + "type": "agent_activity", + "ignore_jobs": _other_maintenance_crons("skill-extractor"), + }], +} + + +def default_run_if(job_id: str, memory_backend: str) -> list[dict[str, Any]]: + """The run gates of a built-in job whose ``system.yaml`` entry states + none, as the caller's own copy; empty for a job that has no default. + + Under Loom, ``memory-maintenance`` checks the namespace as it is, not the + records of one day, so its gate has no overlap. What reaches Loom after a + check waits for the job's next run, after the agent has worked again. + """ + specs = copy.deepcopy(_DEFAULT_RUN_IF.get(job_id, [])) + if memory_backend == "loom" and job_id == "memory-maintenance": + for spec in specs: + spec.pop("overlap_hours", None) + return specs + + def loom_wording_of_shipped_job( job_id: str, prompt: str, description: str, ) -> tuple[str, str] | None: diff --git a/nerve/templates/guides/worker-best-practices.md b/nerve/templates/guides/worker-best-practices.md index 373ed49a2..6214214bf 100644 --- a/nerve/templates/guides/worker-best-practices.md +++ b/nerve/templates/guides/worker-best-practices.md @@ -30,8 +30,8 @@ Each job plays one role, and each role moves tasks in one direction: | Work | Claims one actionable task, moves it to `in_progress`, and ends the run with it in a truthful status | `tasks` on `pending` and the tag it owns | `isolated` | | Monitor | Watches a waiting status and returns the task when the external event happens | `tasks` on that status, plus `github_pr_activity` for PRs | `persistent` or `isolated` | | Planner (`task-planner`) | Proposes a plan for an open task that has none. Ships disabled for workers; onboarding enables it when the worker plans through the approval loop | `tasks` on `pending`, and the worker's tag once overridden | `persistent` | -| Skills (`skill-extractor`, `skill-reviser`) | Propose new or revised skills, each as a task with a plan | None | `persistent` | -| Maintenance (`memory-maintenance`, `machine-maintenance`) | Tidies long-term memory, and once a week the machine (old caches, Docker leftovers, worktrees of merged pull requests, leftover services); asks before it removes or stops anything that can hold data; moves no tasks | None | `isolated` | +| Skills (`skill-extractor`, `skill-reviser`) | Propose new or revised skills, each as a task with a plan | `agent_activity` for `skill-extractor`; none for `skill-reviser` | `persistent` | +| Maintenance (`memory-maintenance`, `machine-maintenance`) | Tidies long-term memory, and once a week the machine (old caches, Docker leftovers, worktrees of merged pull requests, leftover services); asks before it removes or stops anything that can hold data; moves no tasks | `agent_activity` for `memory-maintenance`; none for `machine-maintenance` | `isolated` | The last three ship in `system.yaml` ([cron.md](cron.md#built-in-system-crons)); scan, work and monitor jobs are yours, in `config/cron/jobs.yaml`. `task-planner` ships disabled for workers. When the worker should plan through the approval loop (`task-planner` proposes, the operator approves, and an implementation session does the work), onboarding enables it by overriding the job by id in `jobs.yaml`, with a prompt and a `tasks` gate that name the worker's tag; a job in `jobs.yaml` replaces the `system.yaml` job with the same id. Add a work job when `TASK.md` § Approval Required grants autonomy for routine items. When both run, give each its own tasks by tag. @@ -72,10 +72,11 @@ The same machine as a transition table, in the form `TASK.md` § Actions uses (` ### Gates keep jobs idle until there is work -A run gate is checked right before a job fires; when it is not satisfied, no session starts ([cron.md](cron.md#run-gates)). Three gate types ship with Nerve: +A run gate is checked right before a job fires; when it is not satisfied, no session starts ([cron.md](cron.md#run-gates)). Four gate types ship with Nerve: - `tasks`: enough tasks match a status, or a list of statuses, and an optional tag. A work job gates on the queue it owns, a monitor on the waiting status it watches, the planner on `pending`. No built-in job gates on `backlog`. - `messages`: a sync source has unread messages for a consumer. It drives inbox work, if the worker has sources. +- `agent_activity`: the agent has worked since the job's last run, not counting the job's own runs, the runs of the jobs in `ignore_jobs`, or the runs of other jobs with this gate. The built-in `memory-maintenance` and `skill-extractor` have it by default, so neither runs while the agent has done nothing since its last run. A worker with a scan job works at every scan, so both go on running there. - `github_pr_activity`: something actionable changed on an author's open PRs (a merge or close, a new commit, a review, a CI transition, a human comment); `force_run_after_hours` fires it anyway after a quiet spell. Gates in one list must all pass and are checked in order, so put the cheap `tasks` gate before the network-backed `github_pr_activity`. A gate that errors while checking lets the run proceed; a gate type that does not exist keeps the job from loading. Drop-in plugins in `config/cron/gates/` add conditions the built-in types do not cover. Scan jobs stay ungated: their job is to find work nobody has recorded yet. diff --git a/nerve/templates/personal/TOOLS.md b/nerve/templates/personal/TOOLS.md index 06305e83f..75efd96a6 100644 --- a/nerve/templates/personal/TOOLS.md +++ b/nerve/templates/personal/TOOLS.md @@ -72,7 +72,7 @@ Call tools as `mcp__nerve__`. The system prompt lists what is live; this i ### Features - **Channels**: web UI; Telegram bot (pair with `nerve pair` and `/pair `); Slack via Socket Mode (`/nerve help` lists subcommands). -- **Crons**: `config/cron/jobs.yaml` (yours) and `system.yaml` (built-ins `memory-maintenance`, `inbox-processor`, `task-planner`, `skill-extractor`, `skill-reviser`, `machine-maintenance`); `schedule`, `prompt`/`prompt_file`, `session_mode` (`isolated`/`persistent`/`main`), `run_if` gates (`tasks`, `messages`, `github_pr_activity`, plugins in `config/cron/gates/`) or a `workflow:` block; `nerve cron ` runs one now. +- **Crons**: `config/cron/jobs.yaml` (yours) and `system.yaml` (built-ins `memory-maintenance`, `inbox-processor`, `task-planner`, `skill-extractor`, `skill-reviser`, `machine-maintenance`); `schedule`, `prompt`/`prompt_file`, `session_mode` (`isolated`/`persistent`/`main`), `run_if` gates (`tasks`, `messages`, `agent_activity`, `github_pr_activity`, plugins in `config/cron/gates/`) or a `workflow:` block; `nerve cron ` runs one now. - **External agents**: Codex as session backend (`agent.backend`; `agent.cron_backend` for crons and hooks; sticky per session). The external MCP endpoint (`mcp_endpoint`, `/mcp/v1`, token via `nerve codex token`) serves these tools to other agents as satellite sessions. Codex thread sync ingests local transcripts; `external_agents` renders memory bundles into other agents' home dirs; Claude Code plugins add their MCP servers (Claude backend only). - **Config, workspace sync, lockdown**: settings layers `workspace/config/settings.yaml` < `config.yaml` < `config.local.yaml`. `nerve config init-repo` makes the workspace a config repo; `workspace_sync` pulls, validates and reloads it. The reviewed surface is `config/`, `skills/` and the root instruction files; under lockdown it is read-only locally and `propose_config_change` is the only route (see the `nerve-workspace` skill). `nerve reload` applies most edits live. - **Backups**: `nerve backup` (`--state-only`, `--no-secrets`) bundles databases, secrets, cron and workspace; `nerve restore ` verifies first; scheduled `backup.*` is opt-in; `nerve db prune|vacuum` for retention. diff --git a/nerve/templates/worker/TOOLS.md b/nerve/templates/worker/TOOLS.md index da99cd828..bad5de8a1 100644 --- a/nerve/templates/worker/TOOLS.md +++ b/nerve/templates/worker/TOOLS.md @@ -72,7 +72,7 @@ Call tools as `mcp__nerve__`. The system prompt lists what is live; this i ### Features - **Channels**: web UI; Telegram bot (pair with `nerve pair` and `/pair `); Slack via Socket Mode (`/nerve help` lists subcommands). -- **Crons**: `config/cron/jobs.yaml` (yours) and `system.yaml` (built-ins `memory-maintenance`, `task-planner`, `skill-extractor`, `skill-reviser`, `machine-maintenance`); `schedule`, `prompt`/`prompt_file`, `session_mode` (`isolated`/`persistent`/`main`), `run_if` gates (`tasks`, `messages`, `github_pr_activity`, plugins in `config/cron/gates/`) or a `workflow:` block; `nerve cron ` runs one now. +- **Crons**: `config/cron/jobs.yaml` (yours) and `system.yaml` (built-ins `memory-maintenance`, `task-planner`, `skill-extractor`, `skill-reviser`, `machine-maintenance`); `schedule`, `prompt`/`prompt_file`, `session_mode` (`isolated`/`persistent`/`main`), `run_if` gates (`tasks`, `messages`, `agent_activity`, `github_pr_activity`, plugins in `config/cron/gates/`) or a `workflow:` block; `nerve cron ` runs one now. - **External agents**: Codex as session backend (`agent.backend`; `agent.cron_backend` for crons and hooks; sticky per session). The external MCP endpoint (`mcp_endpoint`, `/mcp/v1`, token via `nerve codex token`) serves these tools to other agents as satellite sessions. Codex thread sync ingests local transcripts; `external_agents` renders memory bundles into other agents' home dirs; Claude Code plugins add their MCP servers (Claude backend only). - **Config, workspace sync, lockdown**: settings layers `workspace/config/settings.yaml` < `config.yaml` < `config.local.yaml`. `nerve config init-repo` makes the workspace a config repo; `workspace_sync` pulls, validates and reloads it. The reviewed surface is `config/`, `skills/` and the root instruction files; under lockdown it is read-only locally and `propose_config_change` is the only route (see the `nerve-workspace` skill). `nerve reload` applies most edits live. - **Backups**: `nerve backup` (`--state-only`, `--no-secrets`) bundles databases, secrets, cron and workspace; `nerve restore ` verifies first; scheduled `backup.*` is opt-in; `nerve db prune|vacuum` for retention. diff --git a/tests/test_config_validate.py b/tests/test_config_validate.py index 1a0c9ba44..015136332 100644 --- a/tests/test_config_validate.py +++ b/tests/test_config_validate.py @@ -798,6 +798,20 @@ def test_valid_builtin_gate_passes(self, tmp_path): assert result.ok, result.errors assert result.warnings == [] + def test_the_activity_gate_is_checked_in_full(self, tmp_path): + spec = {"type": "agent_activity", "ignore_jobs": ["weekly-cleanup"], "overlap_hours": 24} + result = self._run(tmp_path, [spec]) + assert result.ok, result.errors + assert result.warnings == [] + + result = self._run(tmp_path, [{**spec, "overlap_hours": -1}]) + assert not result.ok + assert any("overlap_hours" in e for e in result.errors) + + # A misspelled field would take the default and count the job's runs. + result = self._run(tmp_path, [{"type": "agent_activity", "ignore_job": ["weekly-cleanup"]}]) + assert any("ignore_job" in line and "ignore_jobs" in line for line in result.warnings) + def test_spec_that_is_not_a_mapping_is_an_error(self, tmp_path): result = self._run(tmp_path, ["tasks"]) diff --git a/tests/test_cron_gates.py b/tests/test_cron_gates.py index e79343ae1..9b3a509d3 100644 --- a/tests/test_cron_gates.py +++ b/tests/test_cron_gates.py @@ -2,14 +2,21 @@ from __future__ import annotations +import asyncio import json from datetime import UTC, datetime, timedelta, timezone +from types import SimpleNamespace from unittest.mock import AsyncMock import pytest +import pytest_asyncio +import nerve.workflows +from nerve.agent.sessions import new_cron_run_id +from nerve.agent.streaming import broadcaster from nerve.cron.gates import ( GATE_REGISTRY, + AgentActivityGate, CronGate, GateConfigError, GateContext, @@ -21,7 +28,13 @@ evaluate_gates, ) from nerve.cron.jobs import CronJob +from nerve.cron.service import CronService from nerve.sources.models import SourceRecord +from nerve.workflows.service import ENGINE_CLAUDE, WorkflowRunService +from tests.actor_rows import ensure_system_principal +from tests.test_turn_outcomes import _engine, _session, _token_seen +from tests.test_turn_outcomes import _turn as _real_turn +from tests.test_workflow_runs import _drain, _make_config # --------------------------------------------------------------------------- @@ -940,3 +953,827 @@ async def test_empty_login_cannot_mute_a_ghost_author(self): def test_describe_mentions_comments(self): assert "comment" in GitHubPrActivityGate(author="my-bot").describe() + + +# --------------------------------------------------------------------------- +# AgentActivityGate +# --------------------------------------------------------------------------- + +def _ago(minutes: float) -> datetime: + return datetime.now(UTC) - timedelta(minutes=minutes) + + +def _run_session(job_id: str) -> str: + """The id of a session of one run, or one generation, of ``job_id``.""" + return f"cron:{job_id}:{new_cron_run_id()}" + + +async def _turn(db, session_id: str, *, minutes_ago: float = 0) -> None: + """A turn that ended that long ago in ``session_id`` (made on demand) and + reported its usage. Nothing else is stored: only the usage row says so.""" + if await db.get_session(session_id) is None: + await db.create_session(session_id, source="test", actor=None) + await db.record_turn_usage( + session_id=session_id, input_tokens=10, output_tokens=5, + cache_creation=0, cache_read=0, max_context=200_000, + ) + await db.db.execute( + "UPDATE session_usage SET created_at = ? WHERE id = (SELECT MAX(id) FROM session_usage)", + (_ago(minutes_ago).strftime("%Y-%m-%d %H:%M:%S"),), + ) + await db.db.commit() + + +async def _stored_message(db, session_id: str, *, minutes_ago: float = 0, **kwargs) -> None: + """A message stored that long ago in ``session_id`` (made on demand), as a + turn stores what it was asked. No usage row.""" + if await db.get_session(session_id) is None: + await db.create_session(session_id, source="test", actor=None) + await db.db.execute( # the session is older than anything a test asks about + "UPDATE sessions SET updated_at = datetime('now', '-400 days') WHERE id = ?", + (session_id,), + ) + await db.db.commit() + await db.add_message(session_id, "user", "hello", actor=None, **kwargs) + if kwargs.get("bump_updated_at", True): + await db.db.execute( # the form add_message writes + "UPDATE sessions SET updated_at = ? WHERE id = ?", + (_ago(minutes_ago).isoformat(), session_id), + ) + await db.db.commit() + + +async def _ran(db, job_id: str, *, minutes_ago: float = 0, status: str = "success") -> None: + """A finished run of ``job_id`` that started that long ago.""" + log_id = await db.log_cron_start(job_id) + await db.log_cron_finish(log_id, status) + await db.db.execute( + "UPDATE cron_logs SET started_at = ? WHERE id = ?", + (_ago(minutes_ago).strftime("%Y-%m-%d %H:%M:%S"), log_id), + ) + await db.db.commit() + + +async def _workflow_run(db, run_id: str, created_by: str) -> str: + """A dispatched workflow run, linked to its ``workflow:`` session.""" + session_id = f"workflow:{run_id}" + await db.create_session(session_id, source="workflow", actor=None) + await db.create_workflow_run(run_id, "claude-workflow", {"prompt": "x"}, 5.0, created_by=created_by) + assert await db.transition_workflow_run(run_id, "running", expect=("pending",)) + await db.update_workflow_run(run_id, {"session_id": session_id}) + return session_id + + +# How the agent's work reaches the database: a turn that ended and reported +# its usage, or a message that a turn stored. +WORK = pytest.mark.parametrize("work", [_turn, _stored_message], ids=["a turn that ended", "a stored message"]) + + +@pytest.mark.asyncio +class TestAgentActivityGate: + async def _open(self, db, gate: AgentActivityGate | None = None, job_id: str = "tidy", jobs=()) -> bool: + ctx = GateContext(job_id=job_id, db=db, jobs={job.id: job for job in jobs}) + return await (gate or AgentActivityGate()).is_satisfied(ctx) + + async def test_an_agent_that_never_worked_has_no_activity(self, db): + assert await self._open(db) is False + + @WORK + async def test_a_job_that_never_ran_runs_after_any_work(self, db, work): + await work(db, "chat-1", minutes_ago=60 * 24 * 90) + + assert await self._open(db) is True + + @WORK + async def test_work_before_the_last_run_is_not_activity(self, db, work): + await work(db, "chat-1", minutes_ago=60) + await _ran(db, "tidy", minutes_ago=30) + + assert await self._open(db) is False + + @WORK + async def test_work_since_the_last_run_is_activity(self, db, work): + await _ran(db, "tidy", minutes_ago=60) + await work(db, "chat-1", minutes_ago=30) + + assert await self._open(db) is True + + @pytest.mark.parametrize("stored", ["2026-01-02T03:04:05.678901+00:00", "2026-01-02 03:04:05"]) + async def test_a_session_is_compared_by_its_time_in_either_stored_form(self, db, stored): + """``updated_at`` is ISO 8601 where a message moved it and SQLite's + own text where a status change did. As text, the first is later than + any time of its day in the second form, and the second earlier.""" + await _stored_message(db, "chat-1") + await _ran(db, "tidy") + await db.db.execute("UPDATE sessions SET updated_at = ?", (stored,)) + + for run_started, open_ in (("2026-01-02 03:04:06", False), ("2026-01-02 03:04:05", True), + ("2026-01-01 23:59:59", True), ("2026-01-03 00:00:00", False)): + await db.db.execute("UPDATE cron_logs SET started_at = ?", (run_started,)) + await db.db.commit() + assert await self._open(db) is open_, run_started + + async def test_a_turn_in_the_second_the_run_started_counts(self, db): + """Seconds are all the two clocks have: a turn that ended in the + second the run started may have ended after it began.""" + await _ran(db, "tidy") + await _turn(db, "chat-1") + await db.db.execute("UPDATE cron_logs SET started_at = '2026-01-02 03:04:05'") + await db.db.execute("UPDATE session_usage SET created_at = '2026-01-02 03:04:04'") + await db.db.commit() + assert await self._open(db) is False + + await db.db.execute("UPDATE session_usage SET created_at = '2026-01-02 03:04:05'") + await db.db.commit() + assert await self._open(db) is True + + async def test_a_session_nothing_was_stored_in_is_not_activity(self, db): + await _ran(db, "tidy", minutes_ago=60) + await db.create_session("empty", source="test", actor=None) + + assert await self._open(db) is False + + async def test_a_wakeups_message_is_seen_through_its_turn(self, db): + """A turn nobody asked for stores its messages without moving the + session's ``updated_at``. Its usage row is what says it ran.""" + await _ran(db, "tidy", minutes_ago=60) + await _stored_message(db, "chat-1", bump_updated_at=False) + assert await self._open(db) is False + + await _turn(db, "chat-1") + assert await self._open(db) is True + + async def test_a_conversation_taken_in_from_an_external_agent_is_activity(self, db): + """Its messages keep their own times; the session moves when they are + stored.""" + await _ran(db, "tidy", minutes_ago=60) + await db.create_session("thread-1", source="codex", actor=None) + await db.db.execute("UPDATE sessions SET updated_at = datetime('now', '-400 days')") + await db.db.commit() + assert await self._open(db) is False + + await db.add_message_idempotent( + "thread-1", "assistant", "done", external_id="ext-1", + created_at=_ago(60 * 24 * 3).isoformat(), actor=None, + ) + assert await self._open(db) is True + + @WORK + @pytest.mark.parametrize("session_id", [ + _run_session("tidy"), "cron:tidy:20260102-030405", "cron:tidy", + ], ids=["a run", "a run of an older release", "the stable session of an old persistent job"]) + async def test_the_jobs_own_runs_are_not_activity(self, db, work, session_id): + """A run of the job must not be the reason for its next run.""" + await _ran(db, "tidy", minutes_ago=60) + await work(db, session_id, minutes_ago=59) + + assert await self._open(db) is False + + @WORK + async def test_the_runs_of_another_job_are_activity(self, db, work): + await _ran(db, "tidy", minutes_ago=60) + await work(db, _run_session("report"), minutes_ago=30) + + assert await self._open(db) is True + + @WORK + async def test_the_runs_of_an_ignored_job_are_not_activity(self, db, work): + await _ran(db, "tidy", minutes_ago=60) + await work(db, _run_session("sweep"), minutes_ago=30) + await work(db, "cron:sweep", minutes_ago=30) + + assert await self._open(db, AgentActivityGate(ignore_jobs=["sweep"])) is False + assert await self._open(db, AgentActivityGate(ignore_jobs=["other"])) is True + + @WORK + @pytest.mark.parametrize( + "session_id", [_run_session("tidy:deep"), "cron:tidy:deep"], + ids=["a run of it", "its stable session"], + ) + async def test_a_job_whose_id_extends_this_one_is_another_job(self, db, work, session_id): + """What follows ``cron:tidy:`` in a session of ``tidy`` is a run id: + these are sessions of the job ``tidy:deep``.""" + await _ran(db, "tidy", minutes_ago=60) + await work(db, session_id, minutes_ago=30) + + assert await self._open(db) is True + assert await self._open(db, AgentActivityGate(ignore_jobs=["tidy:deep"])) is False + + @pytest.mark.parametrize("job_id", ["50%_off", "a*b[c]?", "it's \"quoted\"", "задача"]) + async def test_a_job_id_is_text_not_a_pattern(self, db, job_id): + await _ran(db, job_id, minutes_ago=60) + await _turn(db, _run_session(job_id), minutes_ago=30) + assert await self._open(db, job_id=job_id) is False + + await _turn(db, _run_session("50x_off"), minutes_ago=30) + await _turn(db, _run_session("axb[c]?"), minutes_ago=30) + assert await self._open(db, job_id=job_id) is True + + @pytest.mark.parametrize("started_by", ["cron:tidy", "session:cron:tidy", "session:{run}"]) + async def test_a_workflow_run_the_job_started_is_its_own_run(self, db, started_by): + """A workflow-run job does its work in ``workflow:``, and a + prompt job can start a workflow run from its session. Counted as + activity, either would start the job again at every scheduled time.""" + await _ran(db, "tidy", minutes_ago=60) + mine = await _workflow_run(db, "run-1", started_by.format(run=_run_session("tidy"))) + await _turn(db, mine, minutes_ago=30) + await _stored_message(db, mine, minutes_ago=30) + assert await self._open(db) is False + assert await self._open(db, AgentActivityGate(ignore_jobs=["tidy"]), "other") is False + + theirs = await _workflow_run(db, "run-2", "cron:report") + await _turn(db, theirs, minutes_ago=30) + assert await self._open(db) is True + assert await self._open(db, AgentActivityGate(ignore_jobs=["report"])) is False + + async def test_a_workflow_run_a_person_started_is_activity(self, db): + await _ran(db, "tidy", minutes_ago=60) + await _turn(db, await _workflow_run(db, "run-1", "session:chat-1"), minutes_ago=30) + + assert await self._open(db) is True + + async def test_a_failed_run_does_not_move_the_window(self, db): + """What the agent did before a run that failed is still to be looked + at: the next run is not skipped for it.""" + await _ran(db, "tidy", minutes_ago=120) + await _turn(db, "chat-1", minutes_ago=90) + await _ran(db, "tidy", minutes_ago=60, status="error") + + assert await self._open(db) is True + + @WORK + async def test_the_overlap_counts_the_time_before_the_last_run_again(self, db, work): + await work(db, "chat-1", minutes_ago=120) + await _ran(db, "tidy", minutes_ago=60) + + assert await self._open(db, AgentActivityGate()) is False + assert await self._open(db, AgentActivityGate(overlap_hours=2)) is True + assert await self._open(db, AgentActivityGate(overlap_hours=0.5)) is False + + @pytest.mark.parametrize("hours_before_first_run", [15, 3], ids=["an afternoon turn", "a turn in the small hours"]) + async def test_a_daily_job_with_a_days_overlap_runs_twice_after_the_last_work( + self, db, hours_before_first_run, + ): + """The job reviews the day before each run. Whenever the turn was, + the two runs between them review the day it fell on, and then the job + waits: no third run.""" + gate = AgentActivityGate(overlap_hours=24) + await _turn(db, "chat-1", minutes_ago=60 * (72 + hours_before_first_run)) + assert await self._open(db, gate) is True # the first scheduled time after it + + await _ran(db, "tidy", minutes_ago=60 * 72) + assert await self._open(db, gate) is True # a day later: once more + + await _ran(db, "tidy", minutes_ago=60 * 48) + assert await self._open(db, gate) is False # and then it waits + + async def test_jobs_with_this_gate_are_not_activity_for_each_other(self, db): + """One piece of work, then nothing: each job runs once for it. Neither + names the other. A run of one that counted for the other would start + it, whose run would start the first again, for as long as the + instance lives.""" + gated = [{"type": "agent_activity"}] + memory = CronJob(id="memory", schedule="1h", prompt="p", run_if=gated) + skills = CronJob(id="skills", schedule="1h", prompt="p", run_if=gated) + digest = CronJob(id="digest", schedule="1h", prompt="p") + jobs = [memory, skills, digest] + await _turn(db, "chat-1", minutes_ago=50) + assert await self._open(db, memory.gates[0], "memory", jobs) is True + assert await self._open(db, skills.gates[0], "skills", jobs) is True + + await _ran(db, "memory", minutes_ago=40) + await _turn(db, _run_session("memory"), minutes_ago=39) + assert await self._open(db, memory.gates[0], "memory", jobs) is False + assert await self._open(db, skills.gates[0], "skills", jobs) is True + + await _ran(db, "skills", minutes_ago=30) + await _turn(db, _run_session("skills"), minutes_ago=29) + assert await self._open(db, memory.gates[0], "memory", jobs) is False + assert await self._open(db, skills.gates[0], "skills", jobs) is False + # Without the other jobs to look at, the run of ``skills`` is work. + assert await self._open(db, memory.gates[0], "memory") is True + + # A job without the gate is work for both. + await _turn(db, _run_session("digest"), minutes_ago=20) + assert await self._open(db, memory.gates[0], "memory", jobs) is True + assert await self._open(db, skills.gates[0], "skills", jobs) is True + + @WORK + async def test_a_catch_up_counts_only_what_was_recorded_before_the_run_came_due(self, db, work): + """A catch-up stands in for a run that came due while the machine was + paused. What was recorded since, the message that woke the machine + above all, is work for the job's next scheduled time.""" + gate = AgentActivityGate() + await _ran(db, "tidy", minutes_ago=600) + await work(db, "chat-1", minutes_ago=1) + + async def open_as_of(minutes_ago: float | None) -> bool: + as_of = _ago(minutes_ago) if minutes_ago is not None else None + return await gate.is_satisfied(GateContext(job_id="tidy", db=db, as_of=as_of)) + + assert await open_as_of(60) is False + assert await open_as_of(None) is True + + await work(db, "chat-2", minutes_ago=300) + assert await open_as_of(60) is True + assert await open_as_of(301) is False + + async def test_the_time_a_catch_up_asks_about_names_its_zone(self, db): + """A time without a zone would be read as the machine's local time.""" + naive = datetime.now(UTC).replace(tzinfo=None) + + with pytest.raises(ValueError, match="timezone-aware"): + await db.has_activity_since_cron_run("tidy", until=naive) + + async def test_the_legs_of_a_review_loop_the_job_started_are_its_own_run(self, db): + """The legs run in workflow sessions that name the loop, and the loop + names the session that started it.""" + async def loop(loop_id: str, session_id: str) -> str: + if await db.get_session(session_id) is None: + await db.create_session(session_id, source="test", actor=None) + await db.create_review_loop( + loop_id, title="t", session_id=session_id, goal_prompt="g", verifier_prompt="v", + criteria_adoption="no", criteria=[], implementer={}, verifier={}, cwd=None, + max_iterations=1, budget_usd=1.0, + ) + return await _workflow_run(db, f"leg-{loop_id}", f"review-loop:{loop_id}") + + await _ran(db, "tidy", minutes_ago=60) + await _turn(db, await loop("mine", _run_session("tidy")), minutes_ago=30) + assert await self._open(db) is False + assert await self._open(db, AgentActivityGate(ignore_jobs=["tidy"]), "other") is False + + await _turn(db, await loop("theirs", "chat-1"), minutes_ago=30) + assert await self._open(db) is True + + +class TestAgentActivityGateConfig: + def test_from_config(self): + gate = build_gate({"type": "agent_activity"}) + assert isinstance(gate, AgentActivityGate) + assert (gate.ignore_jobs, gate.overlap_hours) == ([], 0.0) + + gate = build_gate({ + "type": "agent_activity", "ignore_jobs": ["a", " b ", "a"], "overlap_hours": 24, + }) + assert (gate.ignore_jobs, gate.overlap_hours) == (["a", "b"], 24.0) + assert build_gate({"type": "agent_activity", "ignore_jobs": "a"}).ignore_jobs == ["a"] + # A bare key is the one way to say "not set". + bare = build_gate({"type": "agent_activity", "ignore_jobs": None, "overlap_hours": None}) + assert (bare.ignore_jobs, bare.overlap_hours) == ([], 0.0) + + @pytest.mark.parametrize("bad", [0, False, "", {}, {"a": 1}, [""], ["a", 3], [None]]) + def test_a_malformed_ignore_list_is_refused(self, bad): + """Read as an empty list, it would count the runs the spec left out.""" + with pytest.raises(GateConfigError, match="ignore_jobs"): + build_gate({"type": "agent_activity", "ignore_jobs": bad}) + + @pytest.mark.parametrize("bad", [ + -1, True, "24", [24], float("nan"), float("inf"), 24 * 366 + 1, 10 ** 400, + ]) + def test_a_malformed_overlap_is_refused(self, bad): + """Refused as a gate that does not build, which costs the one job. An + integer too large for a float must not escape as another error: that + would cost every job of the file.""" + with pytest.raises(GateConfigError, match="overlap_hours"): + build_gate({"type": "agent_activity", "overlap_hours": bad}) + + def test_describe(self): + assert AgentActivityGate().describe() == "agent activity since this job's last run" + described = AgentActivityGate(ignore_jobs=["a", "b"], overlap_hours=24).describe() + assert described == ( + "agent activity since 24h before this job's last run, " + "not counting the runs of a, b" + ) + + def test_only_this_gate_lets_a_manual_run_through(self): + """It estimates whether a run is worth it. The other gates say whether + the job has anything to act on.""" + unchecked = [key for key, cls in GATE_REGISTRY.items() if not cls.checked_on_manual_runs] + assert unchecked == ["agent_activity"] + + +# --------------------------------------------------------------------------- +# AgentActivityGate on real turns and real runs +# --------------------------------------------------------------------------- + +@pytest_asyncio.fixture +async def agent_db(db): + """The database with the agent's identity, as a real engine needs it.""" + await ensure_system_principal(db) + return db + + +async def _age(db, minutes: int) -> None: + """Move everything recorded so far that long into the past.""" + step = f"-{minutes} minutes" + await db.db.execute("UPDATE session_usage SET created_at = datetime(created_at, ?)", (step,)) + await db.db.execute("UPDATE sessions SET updated_at = datetime(updated_at, ?)", (step,)) + await db.db.execute("UPDATE cron_logs SET started_at = datetime(started_at, ?)", (step,)) + await db.db.commit() + + +OK = [("text", "done"), ("result", "sdk-ok")] + + +@pytest.mark.asyncio +class TestAgentActivityOfRealTurns: + """Every way a turn can end leaves a record the gate reads, and leaves it + before the session stops counting as running. That is the moment the next + scheduled task is forecast again: a platform that parks an idle machine + must hear then that the job will run.""" + + @staticmethod + def _asked_when_a_session_stops(engine, db) -> tuple[list[bool], list[asyncio.Future]]: + answers: list[bool] = [] + asked: list[asyncio.Future] = [] + + async def ask() -> None: + answers.append(await AgentActivityGate().is_satisfied(GateContext(job_id="tidy", db=db))) + + def stopped(_session_id: str, running: bool) -> None: + if not running: + asked.append(asyncio.ensure_future(ask())) + + engine.sessions.add_running_listener(stopped) + return answers, asked + + @pytest.mark.parametrize("script, status", [ + (OK, "completed"), + ([("text", "half an answer"), ("raise",)], "failed"), + ([("text", "done"), ("result", "sdk-1", {"usage": None})], "completed"), + ], ids=["completed", "the runtime died", "a result without usage"]) + async def test_a_turn_is_activity_however_it_ends(self, agent_db, tmp_path, script, status): + engine, _ = await _engine(agent_db, tmp_path, [script]) + await _session(engine, "chat") + await _ran(agent_db, "tidy") + await _age(agent_db, 10) + gate = AgentActivityGate() + assert await gate.is_satisfied(_ctx(agent_db, "tidy")) is False + answers, asked = self._asked_when_a_session_stops(engine, agent_db) + + result = await _real_turn(engine, "chat") + await asyncio.gather(*asked) + + assert result.outcome.status == status + assert answers == [True] + + async def test_a_turn_that_was_stopped_is_activity(self, agent_db, tmp_path): + engine, _ = await _engine(agent_db, tmp_path, [[("text", "x"), ("hang",)]]) + await _session(engine, "chat") + await _ran(agent_db, "tidy") + await _age(agent_db, 10) + answers, asked = self._asked_when_a_session_stops(engine, agent_db) + token = await _token_seen("chat") + try: + turn = asyncio.create_task(_real_turn(engine, "chat")) + await asyncio.wait_for(token.wait(), 5) + turn.cancel() + result = await turn + finally: + await broadcaster.unregister("chat", "turn-outcome-test") + await asyncio.gather(*asked) + + assert result.outcome.status == "interrupted" + assert answers == [True] + + +@pytest.mark.asyncio +class TestAgentActivityOfRealRuns: + """The gate on runs of the real scheduler, with the session ids the code + itself makes: one piece of work, then several scheduled times with the + agent otherwise idle.""" + + GATED = [{"type": "agent_activity"}] + + async def _scheduled_times(self, svc: CronService, job: CronJob, times: int, settle=None) -> list[dict]: + for _ in range(times): + await svc._run_job_inner(job) + if settle is not None: + await settle() + await _age(svc.db, 10) + return await svc.db.get_cron_logs(job.id) + + async def _one_chat_turn(self, engine, db) -> None: + await _session(engine, "chat") + await _real_turn(engine, "chat") + await _age(db, 10) + + @pytest.mark.parametrize("mode", ["isolated", "persistent"]) + async def test_a_job_runs_once_for_one_piece_of_work(self, agent_db, tmp_path, mode): + engine, _ = await _engine(agent_db, tmp_path, [OK]) + svc = CronService(engine.config, engine, agent_db) + job = CronJob(id="tidy", schedule="1h", prompt="p", session_mode=mode, + context_rotate_hours=0, run_if=self.GATED) + assert await self._scheduled_times(svc, job, 1) == [] # nothing has happened yet + + await self._one_chat_turn(engine, agent_db) + logs = await self._scheduled_times(svc, job, 3) + + assert [log["status"] for log in logs] == ["success"] + + async def test_a_workflow_run_job_runs_once_for_one_piece_of_work( + self, agent_db, tmp_path, monkeypatch, + ): + """Its work is done in ``workflow:``, not in a cron session.""" + engine, _ = await _engine(agent_db, tmp_path, [OK]) + service = WorkflowRunService(_make_config(tmp_path), agent_db, engine) + monkeypatch.setattr(nerve.workflows, "_service", service) + svc = CronService(engine.config, engine, agent_db) + job = CronJob(id="nightly-audit", schedule="1h", run_if=self.GATED, + workflow={"engine": ENGINE_CLAUDE, "prompt": "audit", "budget_usd": 1.0}) + await self._one_chat_turn(engine, agent_db) + + logs = await self._scheduled_times(svc, job, 4, settle=lambda: _drain(service)) + + assert len(logs) == 1 + assert logs[0]["session_id"].startswith("workflow:") + + async def test_jobs_with_the_gate_do_not_start_each_other(self, agent_db, tmp_path): + """Neither names the other: the scheduler tells each gate which jobs + carry it. One piece of work, then six scheduled times.""" + engine, _ = await _engine(agent_db, tmp_path, [OK]) + svc = CronService(engine.config, engine, agent_db) + memory = CronJob(id="memory", schedule="1h", prompt="p", run_if=self.GATED) + skills = CronJob(id="skills", schedule="1h", prompt="p", run_if=self.GATED) + svc._jobs = [memory, skills] + await self._one_chat_turn(engine, agent_db) + + for _ in range(3): + await self._scheduled_times(svc, memory, 1) + await self._scheduled_times(svc, skills, 1) + + assert len(await agent_db.get_cron_logs("memory")) == 1 + assert len(await agent_db.get_cron_logs("skills")) == 1 + + @pytest.mark.parametrize("origin", ["schedule", "catchup", "delivery"]) + async def test_a_run_that_nobody_started_is_checked(self, agent_db, tmp_path, origin): + engine, _ = await _engine(agent_db, tmp_path, [OK]) + svc = CronService(engine.config, engine, agent_db) + job = CronJob(id="tidy", schedule="1h", prompt="p", run_if=self.GATED) + svc._jobs = [job] + + await svc._run_job_wrapper(job, origin=origin) + + assert await agent_db.get_cron_logs("tidy") == [] + + @pytest.mark.parametrize("worked_before", [False, True]) + async def test_the_startup_catch_up_asks_about_its_own_start( + self, agent_db, tmp_path, monkeypatch, worked_before, + ): + """A run that the gate skipped leaves no row, so the job counts as + overdue at every start. A message that arrives while the catch-up is + under way is work for the job's next scheduled time; work from before + the start is what the catch-up is for.""" + engine, _ = await _engine(agent_db, tmp_path, [OK]) + svc = CronService(engine.config, engine, agent_db) + job = CronJob(id="tidy", schedule="1h", prompt="p", run_if=self.GATED) + svc._jobs = [job] + await _ran(agent_db, "tidy", minutes_ago=180) + await agent_db.db.execute("UPDATE cron_logs SET finished_at = ?", (_ago(179).isoformat(),)) + await agent_db.db.commit() + if worked_before: + await _turn(agent_db, "chat-0", minutes_ago=30) + last_run = agent_db.get_last_successful_cron_run + + async def then_a_message_arrives(job_id: str): + row = await last_run(job_id) + await _stored_message(agent_db, "chat-1") + return row + + monkeypatch.setattr(agent_db, "get_last_successful_cron_run", then_a_message_arrives) + + await svc._catchup_missed_jobs() + + runs = [log["status"] for log in await agent_db.get_cron_logs("tidy")] + assert len(runs) == (2 if worked_before else 1) + + async def test_a_run_that_a_person_starts_is_not_checked(self, agent_db, tmp_path): + """The gate estimates whether a run is worth it. A person who starts + the job may know of work it cannot see, such as a memory written + through the MCP endpoint. A gate that says whether the job has + anything to act on still holds.""" + engine, _ = await _engine(agent_db, tmp_path, [OK]) + svc = CronService(engine.config, engine, agent_db) + tidy = CronJob(id="tidy", schedule="1h", prompt="p", run_if=self.GATED) + planner = CronJob(id="planner", schedule="1h", prompt="p", + run_if=[*self.GATED, {"type": "tasks", "status": "pending"}]) + svc._jobs = [tidy, planner] + + await svc._run_job_wrapper(tidy) # as the scheduler starts it + assert await agent_db.get_cron_logs("tidy") == [] + + await svc.run_job("tidy") + await svc.run_job("planner") + + assert [log["status"] for log in await agent_db.get_cron_logs("tidy")] == ["success"] + assert await agent_db.get_cron_logs("planner") == [] + + +# --------------------------------------------------------------------------- +# The default gates of built-in jobs +# --------------------------------------------------------------------------- + +def _load(tmp_path, system: list[dict] | None, user: list[dict] | None = None, + backend: str = "memu", strict: bool = False) -> dict[str, CronJob]: + """The jobs the scheduler loads from the two files, by id.""" + import yaml + + for name, jobs in (("system.yaml", system), ("jobs.yaml", user)): + if jobs is not None: + (tmp_path / name).write_text(yaml.safe_dump({"jobs": jobs}), encoding="utf-8") + service = SimpleNamespace( + config=SimpleNamespace( + cron=SimpleNamespace( + system_file=tmp_path / "system.yaml", jobs_file=tmp_path / "jobs.yaml", + ), + memory=SimpleNamespace(backend=backend), + ), + _rejected_job_ids=[], + ) + return {job.id: job for job in CronService._load_merged_jobs(service, strict=strict)} + + +def _system_yaml(backend: str = "memu", mode: str = "worker") -> list[dict]: + """The system.yaml job list the installer writes.""" + from nerve.setup_writer import DEFAULT_CRONS, SetupChoices, build_cron_jobs + + return build_cron_jobs(SetupChoices( + mode=mode, enabled_crons=list(DEFAULT_CRONS[mode]), memory_backend=backend, + )) + + +def _gates(job: CronJob) -> list[tuple]: + return [(type(g), g.ignore_jobs, g.overlap_hours) for g in job.gates] + + +def _said_about_defaults(caplog) -> list[str]: + return [r.getMessage() for r in caplog.records if "default gates" in r.getMessage()] + + +class TestDefaultGatesOfBuiltInJobs: + """memory-maintenance and skill-extractor wait for the agent to work. The + gates are not in system.yaml, which nothing writes again when Nerve is + updated: the scheduler gives them to an entry that states none, so an + install from before them has them too, and a later change reaches all.""" + + MEMORY = ["skill-extractor", "skill-reviser", "machine-maintenance"] + SKILLS = ["memory-maintenance", "skill-reviser", "machine-maintenance"] + + @pytest.mark.parametrize("strict", [False, True], ids=["at startup", "on a reload"]) + def test_the_two_jobs_have_their_gates_without_the_file_stating_them(self, tmp_path, strict): + written = {job["id"]: job for job in _system_yaml()} + assert "run_if" not in written["memory-maintenance"] + assert "run_if" not in written["skill-extractor"] + + loaded = _load(tmp_path, list(written.values()), strict=strict) + + assert _gates(loaded["memory-maintenance"]) == [(AgentActivityGate, self.MEMORY, 24.0)] + assert _gates(loaded["skill-extractor"]) == [(AgentActivityGate, self.SKILLS, 0.0)] + + def test_the_memory_backend_decides_the_overlap_not_the_file(self, tmp_path): + """memU's job reviews the records of the day before the run, so it + looks back a day before its last run; Loom's checks the namespace as + it is. The file here was written for memU, and the backend changed.""" + loaded = _load(tmp_path, _system_yaml("memu"), backend="loom") + + assert _gates(loaded["memory-maintenance"]) == [(AgentActivityGate, self.MEMORY, 0.0)] + assert _gates(loaded["skill-extractor"]) == [(AgentActivityGate, self.SKILLS, 0.0)] + + @pytest.mark.parametrize("said", [ + {"run_if": []}, + {"run_if": None}, + {"skip_when_idle": []}, + {"run_if": [{"type": "tasks", "status": "pending"}]}, + ]) + def test_an_entry_that_says_anything_about_gates_is_left_alone(self, tmp_path, said): + """``run_if: []`` is how a file says "on every schedule".""" + jobs = _system_yaml() + for job in jobs: + if job["id"] == "memory-maintenance": + job.update(said) + loaded = _load(tmp_path, jobs)["memory-maintenance"] + + assert not any(isinstance(gate, AgentActivityGate) for gate in loaded.gates) + assert len(loaded.gates) == len(said.get("run_if") or []) + + def test_an_edited_entry_without_run_if_gets_them_too(self, tmp_path): + """The gates go by the job's id and the missing key, nothing else: the + wording of the prompt has changed between releases, so it cannot say + whether an entry is the shipped one.""" + jobs = _system_yaml() + for job in jobs: + if job["id"] == "skill-extractor": + job.update(prompt="Look for what I do twice.", schedule="0 9 * * 1") + loaded = _load(tmp_path, jobs)["skill-extractor"] + + assert loaded.prompt == "Look for what I do twice." + assert _gates(loaded) == [(AgentActivityGate, self.SKILLS, 0.0)] + + def test_no_other_built_in_job_gets_a_gate_it_does_not_state(self, tmp_path): + """inbox-processor and task-planner have their gates in the file. An + entry of theirs without one goes on running on every schedule.""" + jobs = _system_yaml(mode="personal") + for job in jobs: + job.pop("run_if", None) + loaded = _load(tmp_path, jobs) + + gated = {job_id for job_id, job in loaded.items() if job.gates} + assert gated == {"memory-maintenance", "skill-extractor"} + assert {"inbox-processor", "task-planner", "skill-reviser", "machine-maintenance"} <= set(loaded) + + def test_a_job_that_jobs_yaml_replaces_keeps_what_that_file_says(self, tmp_path, caplog): + mine = {"id": "memory-maintenance", "schedule": "0 6 * * *", "prompt": "My own tidy-up."} + with caplog.at_level("INFO", logger="nerve.cron.jobs"): + loaded = _load(tmp_path, _system_yaml(), [mine]) + + assert loaded["memory-maintenance"].prompt == "My own tidy-up." + assert loaded["memory-maintenance"].gates == [] + assert loaded["skill-extractor"].gates + said = _said_about_defaults(caplog) + assert len(said) == 1 and "skill-extractor" in said[0] + + def test_built_in_jobs_kept_in_jobs_yaml_are_left_alone(self, tmp_path): + """An old install with every job in jobs.yaml and no system.yaml: that + file is the user's.""" + loaded = _load(tmp_path, None, _system_yaml()) + + assert loaded["memory-maintenance"].gates == [] + assert loaded["skill-extractor"].gates == [] + + def test_the_load_says_which_gates_it_gave(self, tmp_path, caplog): + with caplog.at_level("INFO", logger="nerve.cron.jobs"): + _load(tmp_path, _system_yaml()) + + said = _said_about_defaults(caplog) + assert len(said) == 2 + assert all( + "system.yaml" in line and "agent activity since" in line and "run_if: []" in line + for line in said + ) + + def test_only_the_scheduler_gives_them(self, tmp_path): + """A plain load of the file returns what the file says: the default + is the scheduler's to give.""" + import yaml + + from nerve.cron.jobs import load_jobs + + path = tmp_path / "system.yaml" + path.write_text(yaml.safe_dump({"jobs": _system_yaml()}), encoding="utf-8") + + assert all(job.gates == [] for job in load_jobs(path) if job.id != "task-planner") + + def test_a_default_that_does_not_build_leaves_the_job_as_the_file_has_it(self, tmp_path, caplog): + """The file never stated the gate: a default that fails must not cost + the job, nor blame the entry.""" + import yaml + + from nerve.cron.jobs import load_jobs + + path = tmp_path / "system.yaml" + path.write_text(yaml.safe_dump({"jobs": [ + {"id": "tidy", "schedule": "1h", "prompt": "p"}, + "not a mapping", + {"id": "broken", "schedule": "1h"}, + ]}), encoding="utf-8") + + with caplog.at_level("INFO", logger="nerve.cron.jobs"): + jobs = load_jobs(path, default_run_if=lambda job_id: [{"type": "no-such-gate"}]) + + assert [(job.id, job.gates) for job in jobs] == [("tidy", [])] + errors = [r.getMessage() for r in caplog.records if r.levelname == "ERROR"] + assert len(errors) == 1 and "tidy" in errors[0] and "no-such-gate" in errors[0] + assert _said_about_defaults(caplog) == errors + # An entry that is itself at fault is refused for what it lacks, with + # or without a default that builds. + path.write_text(yaml.safe_dump({"jobs": [{"id": "broken", "schedule": "1h"}]}), encoding="utf-8") + for default in ([{"type": "tasks"}], [{"type": "no-such-gate"}]): + with pytest.raises(Exception, match="needs a 'prompt'"): + load_jobs(path, strict=True, default_run_if=lambda job_id, d=default: d) + + def test_the_default_is_asked_only_for_an_entry_without_the_keys(self, tmp_path): + import yaml + + from nerve.cron.jobs import load_jobs + + entry = {"schedule": "1h", "prompt": "p"} + path = tmp_path / "jobs.yaml" + path.write_text(yaml.safe_dump({"jobs": [ + {"id": "plain", **entry}, + {"id": "stated", **entry, "run_if": []}, + {"id": "legacy", **entry, "skip_when_idle": ["mail"]}, + {"id": "none-for-it", **entry}, + ]}), encoding="utf-8") + asked: list[str] = [] + + def default(job_id: str) -> list[dict]: + asked.append(job_id) + return [] if job_id == "none-for-it" else [{"type": "tasks", "status": "pending"}] + + jobs = {job.id: job for job in load_jobs(path, default_run_if=default)} + + assert asked == ["plain", "none-for-it"] + assert [type(gate) for gate in jobs["plain"].gates] == [TasksGate] + assert jobs["stated"].gates == [] and jobs["none-for-it"].gates == [] + assert [type(gate) for gate in jobs["legacy"].gates] == [MessagesGate] diff --git a/tests/test_cron_resume.py b/tests/test_cron_resume.py index dfc17973e..bfb7da38b 100644 --- a/tests/test_cron_resume.py +++ b/tests/test_cron_resume.py @@ -437,6 +437,77 @@ async def test_only_catch_up_jobs_run_but_every_trigger_moves_on(held_service): assert svc.scheduler.get_job("foreign").next_run_time == came_due +@pytest.mark.asyncio +@pytest.mark.parametrize("worked_before_the_pause", [False, True]) +async def test_the_message_that_woke_the_machine_starts_no_catch_up_of_a_job_that_waits_for_work( + held_service, db, worked_before_the_pause, +): + """A machine parked for hours is woken by a message. A job gated on the + agent's activity came due while it was parked, when nothing had happened. + Its catch-up asks about that time, so the message, stored since, does not + start the job beside the turn it began; work from before the pause does. + The message counts for the job's next scheduled time.""" + from tests.actor_rows import ensure_system_principal + + await ensure_system_principal(db) + job = CronJob( + id="tidy", schedule=EVERY_TEN_MINUTES, prompt="p", run_if=[{"type": "agent_activity"}], + ) + svc = await held_service([job], db=db) + paused, resumed = _pause_window() + svc.scheduler.modify_job("tidy", next_run_time=resumed - timedelta(hours=1)) + for session_id in ("before", "woke-it"): + await db.create_session(session_id, source="test", actor=None) + if worked_before_the_pause: + await db.add_message("before", "user", "earlier", actor=None) + await db.db.execute( + "UPDATE sessions SET updated_at = ? WHERE id = 'before'", + ((paused - timedelta(minutes=10)).isoformat(),), + ) + await db.db.commit() + await db.add_message("woke-it", "user", "hello", actor=None) + + await _resume(svc, paused, resumed) + + assert svc.engine.run_cron.await_count == (1 if worked_before_the_pause else 0) + forecast = {j["id"]: j["forecast"] for j in (await svc.schedule_report())["jobs"]} + assert forecast["tidy"] == "run" + + +@pytest.mark.asyncio +@pytest.mark.parametrize("worked_before_the_pause", [False, True]) +async def test_a_job_that_waits_for_work_misses_only_a_run_it_would_have_made( + held_service, db, worked_before_the_pause, +): + """With ``catchup: false`` the run that came due during the pause is + recorded as missed unless its gates would have skipped it then. The + message that woke the machine came later: it makes no run missed.""" + job = CronJob( + id="tidy", schedule=EVERY_TEN_MINUTES, prompt="p", catchup=False, + run_if=[{"type": "agent_activity"}], + ) + svc = await held_service([job], db=db) + paused, resumed = _pause_window() + svc.scheduler.modify_job("tidy", next_run_time=resumed - timedelta(hours=1)) + for session_id in ("before", "woke-it"): + await db.create_session(session_id, source="test", actor=None) + if worked_before_the_pause: + await db.add_message("before", "user", "earlier", actor=None) + await db.db.execute( + "UPDATE sessions SET updated_at = ? WHERE id = 'before'", + ((paused - timedelta(minutes=10)).isoformat(),), + ) + await db.db.commit() + await db.add_message("woke-it", "user", "hello", actor=None) + + await _resume(svc, paused, resumed) + await asyncio.sleep(0.3) # the record is written by a task of its own + + statuses = [log["status"] for log in await db.get_cron_logs("tidy")] + assert statuses == (["missed"] if worked_before_the_pause else []) + svc.engine.run_cron.assert_not_awaited() + + @pytest.mark.asyncio async def test_the_catch_up_gives_the_network_time_to_come_back(held_service, monkeypatch): monkeypatch.setattr(cron_service, "_CATCHUP_SETTLE_SECONDS", 0.3) diff --git a/tests/test_cron_schedule.py b/tests/test_cron_schedule.py index 5bf61006c..71571a021 100644 --- a/tests/test_cron_schedule.py +++ b/tests/test_cron_schedule.py @@ -363,6 +363,44 @@ async def test_a_job_whose_gate_forecasts_a_skip_is_passed_over(held): assert (tick.name, tick.at) == ("planner", now + timedelta(minutes=10)) +@pytest.mark.asyncio +async def test_a_job_gated_on_activity_is_the_next_tick_only_after_the_agent_worked(held, db): + """A maintenance job with nothing to look after is not the next scheduled + task, so a platform that parks the machine does not wake it for the job. + Once the agent has worked it is, and after its run it is passed over again.""" + now = datetime.now(UTC) + svc = await held( + [_job("tidy", run_if=[{"type": "agent_activity", "ignore_jobs": ["sweep"]}])], db=db, + ) + _due(svc, "tidy", now + timedelta(minutes=10)) + + async def turn(session_id: str) -> None: + await db.create_session(session_id, source="test", actor=None) + await db.record_turn_usage( + session_id=session_id, input_tokens=1, output_tokens=1, + cache_creation=0, cache_read=0, max_context=200_000, + ) + + assert await svc.next_scheduled(now=now) is None + report = {j["id"]: j for j in (await svc.schedule_report())["jobs"]} + assert report["tidy"]["forecast"] == "skip" + assert report["tidy"]["reason"].startswith("gate 'agent_activity' not satisfied") + + await turn("cron:sweep:20260102-030405-0a1b2c3d") # another maintenance job ran + assert await svc.next_scheduled(now=now) is None + + await turn("chat-1") + tick = await svc.next_scheduled(now=now) + assert (tick.name, tick.at) == ("tidy", now + timedelta(minutes=10)) + + # The job runs: its own turn, and a run on record that started after the chat. + await db.db.execute("UPDATE session_usage SET created_at = datetime('now', '-1 minute')") + await db.db.commit() + await db.log_cron_finish(await db.log_cron_start("tidy"), "success") + await turn("cron:tidy:20260102-040506-1b2c3d4e") + assert await svc.next_scheduled(now=now) is None + + @pytest.mark.asyncio async def test_a_source_runner_counts(held): now = datetime.now(UTC) diff --git a/tests/test_cron_source_delivery.py b/tests/test_cron_source_delivery.py index ea45ffb07..fbfad9240 100644 --- a/tests/test_cron_source_delivery.py +++ b/tests/test_cron_source_delivery.py @@ -67,7 +67,7 @@ def __init__(self) -> None: self.hold: dict[str, asyncio.Event] = {} self.changed = asyncio.Event() - async def __call__(self, job: CronJob, attribution: Any = None) -> None: + async def __call__(self, job: CronJob, attribution: Any = None, **_origin: Any) -> None: self.started.append((job.id, attribution)) self.changed.set() gate = self.hold.get(job.id) @@ -403,7 +403,7 @@ async def test_a_drain_that_ends_while_stop_winds_down_starts_no_run(self, svc, install(svc, job("pr-monitor", sources=["github"]), runners=[runner]) started: list[str] = [] - async def winding_down(job: CronJob, attribution: Any = None) -> None: + async def winding_down(job: CronJob, attribution: Any = None, **_origin: Any) -> None: started.append(job.id) try: await asyncio.Event().wait() diff --git a/tests/test_setup_writer.py b/tests/test_setup_writer.py index c2f53f4df..32befe59c 100644 --- a/tests/test_setup_writer.py +++ b/tests/test_setup_writer.py @@ -1177,6 +1177,90 @@ def test_names_no_memory_tool(self) -> None: assert setup_writer._cron_prompt(job, "loom") == job["prompt"] +class TestMaintenanceJobsWaitForActivity: + """memory-maintenance and skill-extractor look after the agent's own work, + so each has a default gate on the agent's activity: an agent that has done + nothing since the job's last run is not started, or woken, for it. The + scheduler gives the gate to an entry that states none + (``default_run_if``).""" + + OTHERS = { + "memory-maintenance": ["skill-extractor", "skill-reviser", "machine-maintenance"], + "skill-extractor": ["memory-maintenance", "skill-reviser", "machine-maintenance"], + } + + @staticmethod + def _jobs(mode: str, backend: str) -> dict[str, dict]: + choices = SetupChoices( + mode=mode, enabled_crons=list(setup_writer.DEFAULT_CRONS[mode]), memory_backend=backend, + ) + return {job["id"]: job for job in setup_writer.build_cron_jobs(choices)} + + @pytest.mark.parametrize("backend", ["memu", "loom"]) + def test_both_jobs_have_a_default_gate_on_activity(self, backend: str) -> None: + for job_id, others in self.OTHERS.items(): + (spec,) = setup_writer.default_run_if(job_id, backend) + assert (spec["type"], spec["ignore_jobs"]) == ("agent_activity", others) + + def test_no_maintenance_job_counts_as_work_for_another(self) -> None: + """The weekly ones are left out too: their run is no reason for a + memory check or a search for new skills.""" + assert set(setup_writer.MAINTENANCE_CRON_IDS) == { + "memory-maintenance", "skill-extractor", "skill-reviser", "machine-maintenance", + } + known = {c["id"] for c in (*setup_writer.CORE_CRONS, *setup_writer.PRODUCTIVITY_CRONS)} + assert set(setup_writer.MAINTENANCE_CRON_IDS) <= known + for job_id, others in self.OTHERS.items(): + assert set(others) == set(setup_writer.MAINTENANCE_CRON_IDS) - {job_id} + + def test_only_memus_memory_job_looks_back_before_its_last_run(self) -> None: + """memU's prompt reviews the records of the day before the run, so + what the agent did shortly before a run is reviewed by the next one. + Loom's checks the namespace as it is.""" + (memu,) = setup_writer.default_run_if("memory-maintenance", "memu") + (loom,) = setup_writer.default_run_if("memory-maintenance", "loom") + assert memu["overlap_hours"] == 24 + assert "overlap_hours" not in loom + prompt = self._jobs("worker", "memu")["memory-maintenance"]["prompt"] + assert "memory_records_by_date(date=yesterday" in prompt + for backend in ("memu", "loom"): + assert "overlap_hours" not in setup_writer.default_run_if("skill-extractor", backend)[0] + # Giving Loom its gate leaves memU's as it is. + assert setup_writer.default_run_if("memory-maintenance", "memu") == [memu] + + def test_the_gates_build_and_validate(self) -> None: + from nerve.cron.gates import AgentActivityGate, build_gates + + for backend in ("memu", "loom"): + for job_id in self.OTHERS: + specs = setup_writer.default_run_if(job_id, backend) + (gate,) = build_gates(specs) + assert isinstance(gate, AgentActivityGate) + assert set(specs[0]) - {"type"} <= AgentActivityGate.spec_keys + + @pytest.mark.parametrize("mode", ["personal", "worker"]) + @pytest.mark.parametrize("backend", ["memu", "loom"]) + def test_system_yaml_does_not_state_them(self, mode: str, backend: str) -> None: + """Nothing writes the file again when Nerve is updated, so a gate + written there would never change again; and a Nerve from before the + gate type would refuse the two jobs of a file that names it.""" + jobs = self._jobs(mode, backend) + for job_id in self.OTHERS: + assert "run_if" not in jobs[job_id] + assert "agent_activity" not in yaml.safe_dump(list(jobs.values())) + # The gates that are older than this rule stay in the file. + assert jobs["task-planner"]["run_if"] == [{"type": "tasks", "status": "pending"}] + + @pytest.mark.parametrize("backend", ["memu", "loom"]) + def test_no_other_job_has_a_default_gate(self, backend: str) -> None: + for other in ("inbox-processor", "task-planner", "skill-reviser", "machine-maintenance", "mine"): + assert setup_writer.default_run_if(other, backend) == [] + + def test_a_default_gate_is_the_callers_copy(self) -> None: + setup_writer.default_run_if("skill-extractor", "memu")[0]["ignore_jobs"].append("x") + assert "x" not in setup_writer.default_run_if("skill-extractor", "memu")[0]["ignore_jobs"] + + def test_task_planner_asks_for_a_plan_summary() -> None: """The plan page shows the summary under the plan title. The planner prompt asks for it with the parameter names of the tool schema."""