Scheduled and webhook-triggered runs: the logic behind SuperPi.Plugins.Scheduler,
SuperPi.Workers.ScheduleTickWorker, SuperPi.Workers.ScheduleRunWorker and SuperPi.Triggers.
A schedule is a row of schedules (SuperPi.Durable.Schemas.Schedule). One static
Oban.Plugins.Cron entry runs the tick worker every minute. Nothing holds a durable task open
between fires: a daily schedule is a row with a next_fire_at, not a sleeping task.
Exactly once, in three layers
Oban.Plugins.Cronelects a leader, so one tick job is inserted per minute for the cluster.- The tick inserts one run job per due row with the unique key
[:schedule_id, :fire_at]andperiod: :infinity.fire_atis the scheduled minute, never "now", so a duplicate or retried tick inserts nothing. It also movesnext_fire_atforward in the same transaction, guarded by the value it read. - The run claims its fire with one conditional update,
WHERE id = ? AND status = 'active' AND (last_fire_at IS NULL OR last_fire_at < ?fire_at). Zero rows updated means another run or an earlier attempt has it. The claim, the target conversation and the run task commit in one transaction, so a crash leaves either all of them or none, and an Oban retry or aLifelinerescue claims again only when nothing was stored. The claim time is the database clock (CURRENT_TIMESTAMP), not a node's clock.
Oban's unique check only sees jobs that the Pruner has not deleted (24 h by default), which
bounds how long a webhook Idempotency-Key is remembered.
Run modes
| Mode | Target |
|---|---|
"fork" (default) | A fork of the owning conversation at its latest entry. The parent transcript stays clean and the run inherits pi.agent. |
"append" | The owning conversation, for work that should accumulate. |
"fresh" | A new conversation from the agent snapshot stored with the schedule (cheapest in tokens). |
The run is a pending pi.generation task (background: true, restart_limit: 2, and
abandon_on_restart: true unless catch_up is on). The conversation machine picks it up as it
would after a restart (SuperPi.Conversation.Machine.resume_pending/1). If the node dies, Horde
restarts the machine elsewhere and recovery resumes the task. With catch_up off a resumed run
is cancelled instead, so a 07:00 briefing is never delivered at 11:00.
Failures
The tick accounts for settled runs. A run that fails adds one to consecutive_failures, a run
that completes resets it. At three the schedule becomes "paused" and a note goes to the
pi.inbox document of the owning conversation ("notes" list; "queued" is untouched). A
paused schedule stays paused until resume/1.
Catch-up
A tick fires a due schedule when it is at most two minutes late, or with catch_up: true
however late it is. Missed minutes collapse into one run. Otherwise the fire is skipped and
next_fire_at moves on.
Forks do not copy schedules: rows belong to a conversation id, not to a conversation document.
Summary
Functions
Cancels a schedule of the conversation. A run that was claimed already still finishes.
Creates a schedule. Returns {:ok, schedule, token}; token is the webhook secret for
kind: "webhook" (shown once, only its hash is stored) and nil for a cron schedule.
The active webhook schedule for a presented token, compared in constant time, or nil.
One schedule, or nil.
SHA-256 of a webhook token, hex. Only this is stored.
Schedules of a conversation that are not cancelled, oldest first.
Appends a note to the notes list of the conversation's pi.inbox document (the newest
50 are kept). "queued" is left alone. Returns :ok; a conversation that is gone is ignored.
Adds one failure to the schedule (a failed run, or a run job that gave up) and pauses it at
Resumes a paused schedule: clears the failure count and computes the next fire.
Executes one fire. args are the run job's arguments ("schedule_id", "fire_at", and for a
webhook "payload", "received_at", "idempotency_key"). Returns {:ok, task_id} when this call
started the run and {:ok, :skipped} when the fire was claimed already or the schedule is gone,
paused or cancelled.
True when task_id is a scheduled run, or descends from one (an owner task, or the owner task
of its conversation, as a subagent child does). schedule_create refuses to run inside one:
a scheduled run that schedules more runs multiplies without a human to stop it.
Accounts for the runs that settled since the last tick. Failed adds one to
consecutive_failures, completed resets it, any other end (cancelled, abandoned) changes
nothing. At 3 failures in a row the schedule pauses and writes an inbox note.
One tick: accounts for settled runs, then fires every due cron schedule. tick_minute is the
minute the tick job was scheduled for. Returns the number of due schedules seen.
Functions
Cancels a schedule of the conversation. A run that was claimed already still finishes.
@spec create(map(), keyword()) :: {:ok, SuperPi.Durable.Schemas.Schedule.t(), String.t() | nil} | {:error, String.t()}
Creates a schedule. Returns {:ok, schedule, token}; token is the webhook secret for
kind: "webhook" (shown once, only its hash is stored) and nil for a cron schedule.
Attributes: :conversation_id and :prompt (required), :kind, :cron, :timezone,
:mode, :catch_up, :name, :harness, :agent, :principal, :now (test hook).
Options: :max_per_conversation (default 10) and
:min_interval_minutes (default 5), which stop a model from
scheduling itself into a spending loop.
@spec find_webhook(String.t()) :: SuperPi.Durable.Schemas.Schedule.t() | nil
The active webhook schedule for a presented token, compared in constant time, or nil.
One schedule, or nil.
SHA-256 of a webhook token, hex. Only this is stored.
@spec list(integer()) :: [SuperPi.Durable.Schemas.Schedule.t()]
Schedules of a conversation that are not cancelled, oldest first.
Appends a note to the notes list of the conversation's pi.inbox document (the newest
50 are kept). "queued" is left alone. Returns :ok; a conversation that is gone is ignored.
Adds one failure to the schedule (a failed run, or a run job that gave up) and pauses it at
s.last_task_idis cleared when the failure came from that task.
@spec resume(integer(), DateTime.t()) :: :ok | {:error, term()}
Resumes a paused schedule: clears the failure count and computes the next fire.
Executes one fire. args are the run job's arguments ("schedule_id", "fire_at", and for a
webhook "payload", "received_at", "idempotency_key"). Returns {:ok, task_id} when this call
started the run and {:ok, :skipped} when the fire was claimed already or the schedule is gone,
paused or cancelled.
With start: false the run is stored but no machine starts (tests).
True when task_id is a scheduled run, or descends from one (an owner task, or the owner task
of its conversation, as a subagent child does). schedule_create refuses to run inside one:
a scheduled run that schedules more runs multiplies without a human to stop it.
@spec settle_runs() :: :ok
Accounts for the runs that settled since the last tick. Failed adds one to
consecutive_failures, completed resets it, any other end (cancelled, abandoned) changes
nothing. At 3 failures in a row the schedule pauses and writes an inbox note.
@spec tick(DateTime.t()) :: non_neg_integer()
One tick: accounts for settled runs, then fires every due cron schedule. tick_minute is the
minute the tick job was scheduled for. Returns the number of due schedules seen.