Acting Unasked

# labs/autonomous.py
from dataclasses import dataclass, field
from typing import Callable, Optional

@dataclass
class ScheduledTask:
    name: str
    fn: Callable
    interval_sec: Optional[float] = None
    at_time: Optional[str] = None       # "HH:MM", for time-of-day tasks
    last_run: float = field(default_factory=lambda: 0.0)
    last_at_fired: str = ""             # which minute-slot already fired

def memory_checkpoint() -> None:
    print("[TASK] Saving memory checkpoint.")

task = ScheduledTask(name="memory_checkpoint", fn=memory_checkpoint, interval_sec=30)
print(f"Created task '{task.name}': every {task.interval_sec}s, last_run={task.last_run}")
task.fn()

$ uv run python labs/autonomous.py
Created task 'memory_checkpoint': every 30s, last_run=0.0
[TASK] Saving memory checkpoint.

Chapter 2's dataclass, now holding a callable. The scheduler that comes next never learns what memory_checkpoint does; it sees a function and a due time, which keeps the machinery generic in exactly the way the automation engine and the event bus were. Two fields are bookkeeping for the two firing styles: last_run for intervals, last_at_fired for time-of-day slots. The default of 0.0 is a quiet design choice with a visible consequence: every interval task fires immediately on the first tick, because "now minus zero" exceeds any interval. For a checkpoint task, firing once at boot is a feature.

import time

class Scheduler:
    def __init__(self) -> None:
        self.tasks: list[ScheduledTask] = []

    def add(self, task: ScheduledTask) -> None:
        self.tasks.append(task)

    def tick(self) -> None:
        now = time.monotonic()
        for task in self.tasks:
            if task.interval_sec is not None:
                if now - task.last_run >= task.interval_sec:
                    task.fn()
                    task.last_run = now

scheduler = Scheduler()
scheduler.add(ScheduledTask(name="memory_checkpoint", fn=memory_checkpoint,
                            interval_sec=1.0))
for i in range(5):
    print(f"tick {i}")
    scheduler.tick()
    time.sleep(0.6)

$ uv run python labs/autonomous.py
tick 0
[TASK] Saving memory checkpoint.
tick 1
tick 2
[TASK] Saving memory checkpoint.
tick 3
tick 4
[TASK] Saving memory checkpoint.

Fires on ticks 0, 2 and 4, skips 1 and 3, and the arithmetic explains every line: the interval is 1.0 seconds and each tick is 0.6 apart, so tick 1 arrives 0.6 seconds after the last fire (too soon), while tick 2 arrives at 1.2 (due). The subtraction now - task.last_run is the entire interval engine. Note which clock it reads: time.monotonic() only moves forward, at a steady rate, immune to NTP corrections and daylight-saving jumps. Wall-clock time (time.time()) can leap backward, and an elapsed-time subtraction across that leap goes briefly negative. Durations get the monotonic clock; "what time is it" gets the wall clock. Mixing them up works for months, then fails on one specific night in autumn.

from datetime import datetime

    # inside Scheduler.tick(), after the interval branch:
            if task.at_time is not None:
                slot = datetime.now().strftime("%H:%M")
                if slot == task.at_time and task.last_at_fired != slot:
                    task.fn()
                    task.last_at_fired = slot

$ uv run python labs/autonomous.py   # clock at 08:00, ticking every 2s
[TASK] Running morning briefing.
(28 more ticks inside 08:00 — silence)

The guard is two comparisons. The first asks "is it the scheduled minute?"; the second asks "did I already fire for this exact slot?" Recording the slot string in last_at_fired makes the firing idempotent within the minute: thirty ticks land inside 08:00, one briefing runs. And because the slot value changes tomorrow morning ("08:00" is the same string, but yesterday's fire recorded today's slot only after firing), the comparison comes true again exactly once per day. State plus condition, instead of condition alone: the same repair you gave set_mood's silent failures a formal channel for, now keeping a briefing from running three hundred times.

def run_forever(scheduler: Scheduler, tick_seconds: float = 2.0) -> None:
    print("Scheduler running. Ctrl+C to stop.")
    try:
        while True:
            scheduler.tick()
            time.sleep(tick_seconds)
    except KeyboardInterrupt:
        print("\nScheduler stopped. She resents the interruption.")

if __name__ == "__main__":
    s = Scheduler()
    s.add(ScheduledTask(name="checkpoint", fn=memory_checkpoint, interval_sec=1800))
    s.add(ScheduledTask(name="briefing", fn=lambda: print("[TASK] Morning briefing."),
                        at_time="08:00"))
    run_forever(s)

$ uv run python labs/autonomous.py
Scheduler running. Ctrl+C to stop.
[TASK] Saving memory checkpoint.
^C
Scheduler stopped. She resents the interruption.

A two-second tick against a one-minute slot means roughly thirty observations of every minute: no slot can be missed, and the guard absorbs the extra twenty-nine. The CPU cost is one loop iteration every two seconds, a rounding error. This loop will eventually live beside the voice loop, and the polite way to run both is volume 4's problem; for now the scheduler is a second program you can leave running, which is precisely what "autonomous" means at this stage.