Expand description
agents tasks run — fire every pending schedule in the caller’s
own subtree, recording a run row and a full output log per firing.
Scope is fixed to the caller’s own AIH
(ctx.config.agent_instance_hierarchy) plus every descendant.
db::tasks::claim_pending atomically selects the pending rows AND
inserts their tasks_runs rows (advisory-locked, so concurrent
tasks run callers never double-fire a schedule); each claimed
row’s stored argv is then dispatched through the root crate::run
— the same entry the binary and plugins run’s nested-command path
use — in parallel. Per-task streams are merged via
futures::stream::SelectAll; each item is wrapped with the
source schedule’s identity (name / aih / version / plugin).
A log-writer listener is spawned at the top of execute with an
unbounded receiver; the log wrapper (layer 1, identical in both
modes) sends every envelope (with its claim’s run_id, which rides
next to the item internally and never hits the wire) into the
channel, and the listener appends each to tasks_logs as it
arrives. After the merge is exhausted the wrapper drops the sender
and awaits the listener so every log line is durably written before
the stream ends.
Output has two modes (layer 2), selected by request.stream_all:
--stream-all passes every envelope through verbatim
(ResponseItem::Value); the default engages the success layer
instead — items are swallowed and each task yields exactly one
ResponseItem::Success { .., success } when its stream completes,
success being false iff the task’s final item was an error.
The mode never affects what is logged.
Each task runs with the schedule’s captured identity
(apply_agent_arguments) and the plugin that registered it
(apply_plugin) re-installed on the run ctx — both
config.plugin_* and ctx.plugin.
Pre-stream Errs (e.g. the scheduled command failed to parse) are
re-emitted as a single-item error stream in the merged output.