From e0171314030fa5fad2e7e7e96c116c98a0178e33 Mon Sep 17 00:00:00 2001 From: Teknium Date: Sat, 18 Apr 2026 19:30:07 -0700 Subject: [PATCH] =?UTF-8?q?feat(cron):=20add=20wakeAgent=20gate=20?= =?UTF-8?q?=E2=80=94=20scripts=20can=20skip=20the=20agent=20entirely?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Extends the existing cron script hook with a wake gate ported from nanoclaw #1232. When a cron job's pre-check Python script (already sandboxed to HERMES_HOME/scripts/) writes a JSON line like ```json {"wakeAgent": false} ``` on its last stdout line, `run_job()` returns the SILENT marker and skips the agent entirely — no LLM call, no delivery, no tokens spent. Useful for frequent polls (every 1-5 min) that only need to wake the agent when something has genuinely changed. Any other script output (non-JSON, missing key, non-dict, `wakeAgent: true`, truthy/falsy non-False values) behaves as before: stdout is injected as context and the agent runs normally. Strict `False` is required to skip — avoids accidental gating from arbitrary JSON. Refactor: - New pure helper `_parse_wake_gate(script_output)` in cron/scheduler.py - `_build_job_prompt` accepts optional `prerun_script` tuple so the script runs exactly once per job (run_job runs it for the gate check, reuses the output for prompt injection) - `run_job` short-circuits with SILENT_MARKER when gate fires Script failures (success=False) still cannot trigger the gate — the failure is reported as context to the agent as before. This replaces the approach in closed PR #3837, which inlined bash scripts via tempfile and lost the path-traversal/scripts-dir sandbox that main's impl has. The wake-gate idea (the one net-new capability) is ported on top of the existing sandboxed Python-script model. Tests: - 11 pure unit tests for _parse_wake_gate (empty, whitespace, non-JSON, non-dict JSON, missing key, truthy/falsy non-False, multi-line, trailing blanks, non-last-line JSON) - 5 integration tests for run_job wake-gate (skip returns SILENT, wake-true passes through, script-runs-only-once, script failure doesn't gate, no-script regression) - Full tests/cron/ suite: 194/194 pass --- cron/scheduler.py | 69 +++++++++++++- tests/cron/test_scheduler.py | 174 +++++++++++++++++++++++++++++++++++ 2 files changed, 239 insertions(+), 4 deletions(-) diff --git a/cron/scheduler.py b/cron/scheduler.py index 8938063c7f..6e93fc02fe 100644 --- a/cron/scheduler.py +++ b/cron/scheduler.py @@ -564,15 +564,53 @@ def _run_job_script(script_path: str) -> tuple[bool, str]: return False, f"Script execution failed: {exc}" -def _build_job_prompt(job: dict) -> str: - """Build the effective prompt for a cron job, optionally loading one or more skills first.""" +def _parse_wake_gate(script_output: str) -> bool: + """Parse the last non-empty stdout line of a cron job's pre-check script + as a wake gate. + + The convention (ported from nanoclaw #1232): if the last stdout line is + JSON like ``{"wakeAgent": false}``, the agent is skipped entirely — no + LLM run, no delivery. Any other output (non-JSON, missing flag, gate + absent, or ``wakeAgent: true``) means wake the agent normally. + + Returns True if the agent should wake, False to skip. + """ + if not script_output: + return True + stripped_lines = [line for line in script_output.splitlines() if line.strip()] + if not stripped_lines: + return True + last_line = stripped_lines[-1].strip() + try: + gate = json.loads(last_line) + except (json.JSONDecodeError, ValueError): + return True + if not isinstance(gate, dict): + return True + return gate.get("wakeAgent", True) is not False + + +def _build_job_prompt(job: dict, prerun_script: Optional[tuple] = None) -> str: + """Build the effective prompt for a cron job, optionally loading one or more skills first. + + Args: + job: The cron job dict. + prerun_script: Optional ``(success, stdout)`` from a script that has + already been executed by the caller (e.g. for a wake-gate check). + When provided, the script is not re-executed and the cached + result is used for prompt injection. When omitted, the script + (if any) runs inline as before. + """ prompt = job.get("prompt", "") skills = job.get("skills") # Run data-collection script if configured, inject output as context. script_path = job.get("script") if script_path: - success, script_output = _run_job_script(script_path) + if prerun_script is not None: + success, script_output = prerun_script + else: + success, script_output = _run_job_script(script_path) if success: if script_output: prompt = ( @@ -674,7 +712,30 @@ def run_job(job: dict) -> tuple[bool, str, str, Optional[str]]: job_id = job["id"] job_name = job["name"] - prompt = _build_job_prompt(job) + + # Wake-gate: if this job has a pre-check script, run it BEFORE building + # the prompt so a ``{"wakeAgent": false}`` response can short-circuit + # the whole agent run. We pass the result into _build_job_prompt so + # the script is only executed once. + prerun_script = None + script_path = job.get("script") + if script_path: + prerun_script = _run_job_script(script_path) + _ran_ok, _script_output = prerun_script + if _ran_ok and not _parse_wake_gate(_script_output): + logger.info( + "Job '%s' (ID: %s): wakeAgent=false, skipping agent run", + job_name, job_id, + ) + silent_doc = ( + f"# Cron Job: {job_name}\n\n" + f"**Job ID:** {job_id}\n" + f"**Run Time:** {_hermes_now().strftime('%Y-%m-%d %H:%M:%S')}\n\n" + "Script gate returned `wakeAgent=false` — agent skipped.\n" + ) + return True, silent_doc, SILENT_MARKER, None + + prompt = _build_job_prompt(job, prerun_script=prerun_script) origin = _resolve_origin(job) _cron_session_id = f"cron_{job_id}_{_hermes_now().strftime('%Y%m%d_%H%M%S')}" diff --git a/tests/cron/test_scheduler.py b/tests/cron/test_scheduler.py index 2717584e46..b889ede372 100644 --- a/tests/cron/test_scheduler.py +++ b/tests/cron/test_scheduler.py @@ -1175,6 +1175,180 @@ class TestBuildJobPromptSilentHint: assert system_pos < prompt_pos +class TestParseWakeGate: + """Unit tests for _parse_wake_gate — pure function, no side effects.""" + + def test_empty_output_wakes(self): + from cron.scheduler import _parse_wake_gate + assert _parse_wake_gate("") is True + assert _parse_wake_gate(None) is True + + def test_whitespace_only_wakes(self): + from cron.scheduler import _parse_wake_gate + assert _parse_wake_gate(" \n\n \t\n") is True + + def test_non_json_last_line_wakes(self): + from cron.scheduler import _parse_wake_gate + assert _parse_wake_gate("hello world") is True + assert _parse_wake_gate("line 1\nline 2\nplain text") is True + + def test_json_non_dict_wakes(self): + """Bare arrays, numbers, strings must not be interpreted as a gate.""" + from cron.scheduler import _parse_wake_gate + assert _parse_wake_gate("[1, 2, 3]") is True + assert _parse_wake_gate("42") is True + assert _parse_wake_gate('"wakeAgent"') is True + + def test_wake_gate_false_skips(self): + from cron.scheduler import _parse_wake_gate + assert _parse_wake_gate('{"wakeAgent": false}') is False + + def test_wake_gate_true_wakes(self): + from cron.scheduler import _parse_wake_gate + assert _parse_wake_gate('{"wakeAgent": true}') is True + + def test_wake_gate_missing_wakes(self): + """A JSON dict without a wakeAgent key defaults to waking.""" + from cron.scheduler import _parse_wake_gate + assert _parse_wake_gate('{"data": {"foo": "bar"}}') is True + + def test_non_boolean_false_still_wakes(self): + """Only strict ``False`` skips — truthy/falsy shortcuts are too risky.""" + from cron.scheduler import _parse_wake_gate + assert _parse_wake_gate('{"wakeAgent": 0}') is True + assert _parse_wake_gate('{"wakeAgent": null}') is True + assert _parse_wake_gate('{"wakeAgent": ""}') is True + + def test_only_last_non_empty_line_parsed(self): + from cron.scheduler import _parse_wake_gate + multi = 'some log output\nmore output\n{"wakeAgent": false}' + assert _parse_wake_gate(multi) is False + + def test_trailing_blank_lines_ignored(self): + from cron.scheduler import _parse_wake_gate + multi = '{"wakeAgent": false}\n\n\n' + assert _parse_wake_gate(multi) is False + + def test_non_last_json_line_does_not_gate(self): + """A JSON gate on an earlier line with plain text after it does NOT trigger.""" + from cron.scheduler import _parse_wake_gate + multi = '{"wakeAgent": false}\nactually this is the real output' + assert _parse_wake_gate(multi) is True + + +class TestRunJobWakeGate: + """Integration tests for run_job wake-gate short-circuit.""" + + def _make_job(self, name="wake-gate-test", script="check.py"): + """Minimal valid cron job dict for run_job.""" + return { + "id": f"job_{name}", + "name": name, + "prompt": "Do a thing", + "schedule": "*/5 * * * *", + "script": script, + } + + def test_wake_false_skips_agent_and_returns_silent(self, caplog): + """When _run_job_script output ends with {wakeAgent: false}, the agent + is not invoked and run_job returns the SILENT marker so delivery is + suppressed.""" + from cron.scheduler import SILENT_MARKER + import cron.scheduler as scheduler + + with patch.object(scheduler, "_run_job_script", + return_value=(True, '{"wakeAgent": false}')), \ + patch("run_agent.AIAgent") as agent_cls: + success, doc, final, err = scheduler.run_job(self._make_job()) + + assert success is True + assert err is None + assert final == SILENT_MARKER + assert "Script gate returned `wakeAgent=false`" in doc + agent_cls.assert_not_called() + + def test_wake_true_runs_agent_with_injected_output(self): + """When the script returns {wakeAgent: true, data: ...}, the agent is + invoked and the data line still shows up in the prompt.""" + import cron.scheduler as scheduler + + script_output = '{"wakeAgent": true, "data": {"new": 3}}' + agent = MagicMock() + agent.run_conversation = MagicMock(return_value={ + "final_response": "ok", "messages": [] + }) + with patch.object(scheduler, "_run_job_script", + return_value=(True, script_output)), \ + patch("run_agent.AIAgent", return_value=agent) as agent_cls: + success, doc, final, err = scheduler.run_job(self._make_job()) + + agent_cls.assert_called_once() + # The script output should be visible in the prompt passed to + # run_conversation. + call_kwargs = agent.run_conversation.call_args + prompt_arg = call_kwargs.args[0] if call_kwargs.args else call_kwargs.kwargs.get("user_message", "") + assert script_output in prompt_arg + assert success is True + assert err is None + + def test_script_runs_only_once_on_wake(self): + """Wake-true path must not re-run the script inside _build_job_prompt + (script would execute twice otherwise, wasting work and risking + double-side-effects).""" + import cron.scheduler as scheduler + + call_count = 0 + def _script_stub(path): + nonlocal call_count + call_count += 1 + return (True, "regular output") + + agent = MagicMock() + agent.run_conversation = MagicMock(return_value={ + "final_response": "ok", "messages": [] + }) + with patch.object(scheduler, "_run_job_script", side_effect=_script_stub), \ + patch("run_agent.AIAgent", return_value=agent): + scheduler.run_job(self._make_job()) + + assert call_count == 1, f"script ran {call_count}x, expected exactly 1" + + def test_script_failure_does_not_trigger_gate(self): + """If _run_job_script returns success=False, the gate is NOT evaluated + and the agent still runs (the failure is reported as context).""" + import cron.scheduler as scheduler + + # Malicious or broken script whose stderr happens to contain the + # gate JSON — we must NOT honor it because ran_ok is False. + agent = MagicMock() + agent.run_conversation = MagicMock(return_value={ + "final_response": "ok", "messages": [] + }) + with patch.object(scheduler, "_run_job_script", + return_value=(False, '{"wakeAgent": false}')), \ + patch("run_agent.AIAgent", return_value=agent) as agent_cls: + success, doc, final, err = scheduler.run_job(self._make_job()) + + agent_cls.assert_called_once() # Agent DID wake despite the gate-like text + + def test_no_script_path_runs_agent_normally(self): + """Regression: jobs without a script still work.""" + import cron.scheduler as scheduler + + agent = MagicMock() + agent.run_conversation = MagicMock(return_value={ + "final_response": "ok", "messages": [] + }) + job = self._make_job(script=None) + job.pop("script", None) + with patch.object(scheduler, "_run_job_script") as script_fn, \ + patch("run_agent.AIAgent", return_value=agent) as agent_cls: + scheduler.run_job(job) + + script_fn.assert_not_called() + agent_cls.assert_called_once() + + class TestBuildJobPromptMissingSkill: """Verify that a missing skill logs a warning and does not crash the job."""