Cron: memory-maintenance and skill-extractor run only after the agent has worked

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
 

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.

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-<id>` 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`.

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

nerve/cron/gates.py

@@ -40,14 +40,16 @@
 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
 
 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
+    #: <id>``, 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,
 }
 

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: