Scheduler Context Chaining

Chain scheduled jobs so each downstream tick receives its upstreams' last successful output as bounded context

Turn isolated scheduled jobs into a composable pipeline — a downstream job names its upstreams, and the runner prepends each upstream’s most-recent successful output to the downstream message as bounded, clearly-delimited context.

Quick Start

1. One upstream feeds one downstream

   Point a downstream job at an upstream by name. At fire time the runner resolves the upstream’s last successful output and prepends it as context.

   from praisonaiagents import Agent
   from praisonaiagents.scheduler import ScheduleJob, Schedule
   
   agent = Agent(name="Briefer", instructions="Write a short morning briefing")
   
   fetch = ScheduleJob(
       name="fetch_inputs",
       schedule=Schedule(kind="cron", cron_expr="0 6 * * *"),
       message="Fetch today's calendar and unread email summary.",
   )
   
   brief = ScheduleJob(
       name="briefing",
       schedule=Schedule(kind="cron", cron_expr="5 6 * * *"),
       message="Draft a briefing from the fetched inputs.",
       agent_id=agent.id,
       context_from=["fetch_inputs"],
   )

2. Bare-string upstream in YAML

   For a single upstream, write context_from as a plain string — from_dict coerces it to a one-element list.

   jobs:
     - name: briefing
       message: "Draft a briefing from the fetched inputs."
       schedule:
         kind: cron
         cron_expr: "5 6 * * *"
       context_from: fetch_inputs

3. Multiple upstreams + skip on empty

   Name several upstreams, tighten the per-stage budget, and refuse to run when an input is missing.

   from praisonaiagents.scheduler import ScheduleJob, Schedule
   
   brief = ScheduleJob(
       name="briefing",
       schedule=Schedule(kind="cron", cron_expr="5 6 * * *"),
       message="Merge calendar and email into one briefing.",
       context_from=["calendar", "email"],
       context_max_chars=1500,
       on_missing_context="skip",
   )

How It Works

Each downstream tick resolves its upstreams from execution history, bounds each output, and formats one section per upstream.

Each resolved upstream becomes a section formatted exactly as ### Context from '{ref}' followed by its output, and sections join with a blank line:

### Context from 'fetch_inputs'
calendar + unread email summary

Two upstreams preserve declared order:

### Context from 'a'
AAA

### Context from 'b'
BBB

Configuration Options

Three declarable fields on ScheduleJob control chaining.

ScheduleRunner.resolve_context(job) returns a (context, missing) tuple.

from praisonaiagents.scheduler import ScheduleJob, Schedule, ScheduleRunner, FileScheduleStore

store = FileScheduleStore(store_dir="/tmp/schedules")
runner = ScheduleRunner(store)

brief = ScheduleJob(name="briefing", context_from=["fetch_inputs"])
context, missing = runner.resolve_context(brief)

Common Patterns

Morning briefing pipeline

Three chained jobs — fetch, summarise, deliver — each small and independently observable in run history.

from praisonaiagents import Agent
from praisonaiagents.scheduler import ScheduleJob, Schedule, DeliveryTarget

writer = Agent(name="Briefer", instructions="Write a concise morning briefing")

fetch = ScheduleJob(
    name="fetch_inputs",
    schedule=Schedule(kind="cron", cron_expr="0 6 * * *"),
    message="Fetch today's calendar and unread email summary.",
)

merge = ScheduleJob(
    name="briefing",
    schedule=Schedule(kind="cron", cron_expr="5 6 * * *"),
    message="Merge the fetched inputs into one briefing.",
    agent_id=writer.id,
    context_from=["fetch_inputs"],
)

deliver = ScheduleJob(
    name="deliver_briefing",
    schedule=Schedule(kind="cron", cron_expr="10 6 * * *"),
    message="Send the briefing to the team.",
    agent_id=writer.id,
    context_from=["briefing"],
    delivery=DeliveryTarget(channel="telegram", channel_id="123456"),
)

The final stage typically delivers — see Scheduler Delivery.

Fail-closed downstream

Refuse to draft a briefing when the inbox fetch produced nothing.

from praisonaiagents.scheduler import ScheduleJob, Schedule

brief = ScheduleJob(
    name="briefing",
    schedule=Schedule(kind="cron", cron_expr="5 6 * * *"),
    message="Draft a briefing.",
    context_from=["fetch_inputs"],
    on_missing_context="skip",
)

on_missing_context="skip" records the tick as the existing skipped outcome — no tokens, no delivery — so the downstream never runs on empty inputs.

Fan-in from two watchers

One downstream consumes two independent monitor jobs’ outputs.

from praisonaiagents.scheduler import ScheduleJob, Schedule

releases = ScheduleJob(
    name="release-watch",
    schedule=Schedule(kind="every", every_seconds=3600),
    message="Summarise what changed.",
    monitor={"url": "https://example.com/releases"},
)

status = ScheduleJob(
    name="status-watch",
    schedule=Schedule(kind="every", every_seconds=3600),
    message="Summarise service status.",
    monitor={"command": "curl -sf --max-time 5 https://status.example.com/health"},
)

digest = ScheduleJob(
    name="ops-digest",
    schedule=Schedule(kind="cron", cron_expr="0 * * * *"),
    message="Combine release and status notes into one digest.",
    context_from=["release-watch", "status-watch"],
)

Best Practices

ScheduleJob(name="brief", context_from=["fetch"], context_max_chars=1500)

Related