diff --git a/cron/scheduler.py b/cron/scheduler.py index 9eb8f28da8..5985a88002 100644 --- a/cron/scheduler.py +++ b/cron/scheduler.py @@ -3550,6 +3550,16 @@ def _build_job_prompt( if isinstance(context_from, str): context_from = [context_from] for source_job_id in context_from: + # "self" resolves to the job's own id: the job wakes up with its + # most recent output injected, giving recurring jobs continuity + # across runs (dedupe against what was already reported, continue + # where the last run left off) without touching session history. + is_self = False + if isinstance(source_job_id, str) and source_job_id.strip().lower() == "self": + source_job_id = str(job.get("id") or "") + is_self = True + elif source_job_id == job.get("id"): + is_self = True # Guard against path traversal — valid job IDs are 12-char hex strings if not source_job_id or not all(c in "0123456789abcdef" for c in source_job_id): logger.warning( @@ -3577,13 +3587,24 @@ def _build_job_prompt( if len(latest_output) > _MAX_CONTEXT_CHARS: latest_output = latest_output[:_MAX_CONTEXT_CHARS] + "\n\n[... output truncated ...]" if latest_output: - prompt = ( - f"## Output from job '{source_job_id}'\n" - "The following is the most recent output from a preceding " - "cron job. Use it as context for your analysis.\n\n" - f"```\n{latest_output}\n```\n\n" - f"{prompt}" - ) + if is_self: + prompt = ( + "## Your previous run's output\n" + "The following is this job's most recent output from its " + "previous run. Use it for continuity: avoid repeating what " + "was already reported, and continue where the last run " + "left off.\n\n" + f"```\n{latest_output}\n```\n\n" + f"{prompt}" + ) + else: + prompt = ( + f"## Output from job '{source_job_id}'\n" + "The following is the most recent output from a preceding " + "cron job. Use it as context for your analysis.\n\n" + f"```\n{latest_output}\n```\n\n" + f"{prompt}" + ) has_injected_data = True else: continue # silent skip — empty output diff --git a/tests/cron/test_cron_context_from.py b/tests/cron/test_cron_context_from.py index 99acaff35e..fcc5046ab0 100644 --- a/tests/cron/test_cron_context_from.py +++ b/tests/cron/test_cron_context_from.py @@ -217,3 +217,103 @@ class TestUpdateContextFrom: assert reloaded["context_from"] == [job_a["id"]] +class TestSelfContext: + """The special 'self' value injects the job's OWN previous output. + + Inspired by Amp's "Right on Schedule" (agents wake up with their saved + context and continue where they left off): recurring jobs get run-to-run + continuity without touching session history. + """ + + def test_self_injects_own_previous_output(self, cron_env): + from cron.jobs import create_job, OUTPUT_DIR + from cron.scheduler import _build_job_prompt + + job = create_job( + prompt="Scan for news", schedule="every 1h", context_from="self" + ) + out_dir = OUTPUT_DIR / job["id"] + out_dir.mkdir(parents=True, exist_ok=True) + (out_dir / "2026-08-01_10-00-00.md").write_text( + "Reported: story A, story B", encoding="utf-8" + ) + + prompt = _build_job_prompt(job) + assert "Reported: story A, story B" in prompt + assert "previous run" in prompt.lower() + # Self-context uses continuity framing, not the upstream-job framing. + assert f"Output from job '{job['id']}'" not in prompt + + def test_self_case_insensitive(self, cron_env): + from cron.jobs import create_job, OUTPUT_DIR + from cron.scheduler import _build_job_prompt + + job = create_job( + prompt="Scan", schedule="every 1h", context_from="SELF" + ) + out_dir = OUTPUT_DIR / job["id"] + out_dir.mkdir(parents=True, exist_ok=True) + (out_dir / "2026-08-01_10-00-00.md").write_text("prev", encoding="utf-8") + prompt = _build_job_prompt(job) + assert "prev" in prompt + + def test_self_silent_skip_on_first_run(self, cron_env): + from cron.jobs import create_job + from cron.scheduler import _build_job_prompt + + job = create_job( + prompt="Scan for news", schedule="every 1h", context_from="self" + ) + # No output yet (first run) — base prompt intact, no placeholder. + prompt = _build_job_prompt(job) + assert "Scan for news" in prompt + assert "previous run" not in prompt.lower() + + def test_own_id_treated_as_self(self, cron_env): + """Passing the job's literal id gets the continuity framing too.""" + from cron.jobs import create_job, update_job, get_job, OUTPUT_DIR + from cron.scheduler import _build_job_prompt + + job = create_job(prompt="Scan", schedule="every 1h") + update_job(job["id"], {"context_from": [job["id"]]}) + job = get_job(job["id"]) + assert job is not None + + out_dir = OUTPUT_DIR / job["id"] + out_dir.mkdir(parents=True, exist_ok=True) + (out_dir / "2026-08-01_10-00-00.md").write_text("prev output", encoding="utf-8") + + prompt = _build_job_prompt(job) + assert "prev output" in prompt + assert "previous run" in prompt.lower() + + def test_tool_create_accepts_self(self, cron_env): + from tools.cronjob_tools import cronjob + from cron.jobs import get_job + import json + + result = json.loads(cronjob( + action="create", + prompt="Scan for news", + schedule="every 1h", + context_from="self", + )) + assert result["success"] is True + job_id = result["job_id"] + assert get_job(job_id)["context_from"] == ["self"] + + def test_tool_update_accepts_self(self, cron_env): + from cron.jobs import create_job, get_job + from tools.cronjob_tools import cronjob + import json + + job = create_job(prompt="Scan", schedule="every 1h") + result = json.loads(cronjob( + action="update", + job_id=job["id"], + context_from="self", + )) + assert result["success"] is True + assert get_job(job["id"])["context_from"] == ["self"] + + diff --git a/tools/cronjob_tools.py b/tools/cronjob_tools.py index 11c94b4772..cf8a834030 100644 --- a/tools/cronjob_tools.py +++ b/tools/cronjob_tools.py @@ -1220,6 +1220,11 @@ def cronjob( from cron.jobs import get_job as _get_job refs = [context_from] if isinstance(context_from, str) else context_from for ref_id in refs: + # "self" is resolved to the job's own id at run time — + # it can't be validated against the store (the job does + # not exist yet at create time). + if isinstance(ref_id, str) and ref_id.strip().lower() == "self": + continue if not _get_job(ref_id): return tool_error( f"context_from job '{ref_id}' not found. " @@ -1500,6 +1505,9 @@ def cronjob( if refs: from cron.jobs import get_job as _get_job for ref_id in refs: + # "self" resolves to the job's own id at run time. + if ref_id.lower() == "self": + continue if not _get_job(ref_id): return tool_error( f"context_from job '{ref_id}' not found. " @@ -1649,7 +1657,10 @@ Scheduling from cron-run sessions is disabled by default and enabled via cron.al "Optional job ID or list of job IDs whose most recent completed output is " "injected into the prompt as context before each run. " "Use this to chain cron jobs: job A collects data, job B processes it. " - "Each entry must be a valid job ID (from cronjob action='list'). " + "The special value 'self' injects this job's OWN previous output, giving " + "recurring jobs continuity across runs (dedupe against what was already " + "reported, continue where the last run left off). " + "Each other entry must be a valid job ID (from cronjob action='list'). " "Note: injects the most recent completed output — does not wait for " "upstream jobs running in the same tick. " "On update, pass an empty array to clear." diff --git a/website/docs/user-guide/features/cron.md b/website/docs/user-guide/features/cron.md index 7b3c7e1a6f..30efca02dd 100644 --- a/website/docs/user-guide/features/cron.md +++ b/website/docs/user-guide/features/cron.md @@ -629,14 +629,32 @@ cronjob( |--------|---------| | Single job ID (string) | `context_from="a1b2c3d4"` | | Multiple job IDs (list) | `context_from=["job_a", "job_b"]` | +| The special value `self` | `context_from="self"` | Outputs are concatenated in the order listed. +**Self-context: continuity across runs** + +`context_from="self"` injects the job's *own* most recent output into each run. Recurring jobs normally start every run with amnesia — a news scout re-reports the same stories, a monitor re-alerts on the same condition. With self-context, the job wakes up seeing what it reported last time and can dedupe and continue where it left off: + +```python +cronjob( + action="create", + prompt="Scan HN and arXiv for new agent-tooling papers. Report only items NOT already covered in your previous run's output.", + schedule="every 6h", + context_from="self", + name="Agent Tooling Scout", +) +``` + +The first run has no previous output, so the prompt runs as-is. On later runs the previous output is prepended with continuity framing ("avoid repeating what was already reported"). You can combine it with upstream jobs: `context_from=["self", ""]`. + **When to use it:** - Multi-stage pipelines (collect → filter → format → deliver) - Dependent tasks where step N's work depends on step N−1's output - Fan-out/fan-in patterns where one job aggregates results from several others +- Recurring scouts/monitors that should dedupe against their own previous report (`context_from="self"`) ## Provider recovery