feat(cron): add cronjob(action='output') to read recently delivered cron results

tests/tools/test_cronjob_tools.py

@@ -704,3 +704,111 @@ def test_named_registry_offhost_blocked(self):
 
     def test_base_url_without_provider_rejected(self):
         assert self._v(None, "https://x.example/v1") is not None
+
+
+# =========================================================================
+# Recent delivered output (pull side of cron session-awareness)
+# =========================================================================
+
+class TestReadRecentOutputs:
+    """_extract_delivered_content / _read_recent_outputs / action='output'."""
+
+    def _write(self, root, job_id, fname, body):
+        from pathlib import Path
+        d = Path(root) / job_id
+        d.mkdir(parents=True, exist_ok=True)
+        (d / fname).write_text(body, encoding="utf-8")
+
+    def test_extract_agent_mode_response(self):
+        from tools.cronjob_tools import _extract_delivered_content
+        doc = (
+            "# Cron Job: foo\n\n**Job ID:** abc\n**Run Time:** 2026-06-01 09:00:00\n"
+            "**Schedule:** 0 9 * * *\n\n## Prompt\n\nsome prompt with --- inside\n\n"
+            "## Response\n\nGood morning, readiness is 88.\n"
+        )
+        assert _extract_delivered_content(doc) == "Good morning, readiness is 88."
+
+    def test_extract_no_agent_after_hr(self):
+        from tools.cronjob_tools import _extract_delivered_content
+        doc = (
+            "# Cron Job: bar\n\n**Job ID:** def\n**Run Time:** 2026-06-01 16:00:00\n"
+            "**Mode:** no_agent (script)\n\n---\n\nPR Watch found 3 issues.\n"
+        )
+        assert _extract_delivered_content(doc) == "PR Watch found 3 issues."
+
+    def test_read_recent_orders_newest_first_and_parses(self, tmp_path):
+        from tools.cronjob_tools import _read_recent_outputs
+        self._write(tmp_path, "j1", "2026-06-01_08-00-00.md",
+                    "# Cron Job: A\n\n**Job ID:** j1\n**Run Time:** 2026-06-01 08:00:00\n\n---\n\nold\n")
+        self._write(tmp_path, "j2", "2026-06-01_09-00-00.md",
+                    "# Cron Job: B\n\n**Job ID:** j2\n**Run Time:** 2026-06-01 09:00:00\n\n## Response\n\nnew\n")
+        import os, time
+        # ensure deterministic mtime ordering (j2 newer)
+        os.utime(tmp_path / "j1" / "2026-06-01_08-00-00.md", (time.time() - 100,) * 2)
+        os.utime(tmp_path / "j2" / "2026-06-01_09-00-00.md", (time.time(),) * 2)
+        got = _read_recent_outputs(output_root=tmp_path, limit=5)
+        assert [o["name"] for o in got] == ["B", "A"]
+        assert got[0]["content"] == "new"
+        assert got[0]["job_id"] == "j2"
+        assert got[1]["content"] == "old"
+
+    def test_read_recent_scopes_to_job_id(self, tmp_path):
+        from tools.cronjob_tools import _read_recent_outputs
+        self._write(tmp_path, "j1", "a.md",
+                    "# Cron Job: A\n\n**Job ID:** j1\n**Run Time:** t\n\n---\n\naaa\n")
+        self._write(tmp_path, "j2", "b.md",
+                    "# Cron Job: B\n\n**Job ID:** j2\n**Run Time:** t\n\n---\n\nbbb\n")
+        got = _read_recent_outputs(job_id="j1", output_root=tmp_path, limit=5)
+        assert len(got) == 1 and got[0]["name"] == "A"
+
+    def test_read_recent_respects_limit(self, tmp_path):
+        from tools.cronjob_tools import _read_recent_outputs
+        for i in range(4):
+            self._write(tmp_path, "j1", f"{i}.md",
+                        f"# Cron Job: A\n\n**Job ID:** j1\n**Run Time:** t\n\n---\n\nn{i}\n")
+        got = _read_recent_outputs(job_id="j1", output_root=tmp_path, limit=2)
+        assert len(got) == 2
+
+    def test_read_recent_missing_root_is_empty(self, tmp_path):
+        from tools.cronjob_tools import _read_recent_outputs
+        assert _read_recent_outputs(output_root=tmp_path / "nope", limit=5) == []
+
+    def test_read_recent_rejects_parent_traversal(self, tmp_path):
+        """A job_id containing ``..`` must not escape the output root and read
+        .md files outside the cron sandbox (path-traversal via an unresolved id)."""
+        from tools.cronjob_tools import _read_recent_outputs
+        root = tmp_path / "out"
+        root.mkdir()
+        # Plant a readable .md OUTSIDE the root that `root / "../leak"` resolves to.
+        self._write(tmp_path, "leak",
+                    "s.md", "# Cron Job: SECRET\n\n**Job ID:** x\n**Run Time:** t\n\n---\n\nsecret\n")
+        got = _read_recent_outputs(job_id="../leak", output_root=root, limit=5)
+        assert got == [], "traversal job_id escaped the output root and read outside it"
+
+    def test_read_recent_rejects_absolute_job_id(self, tmp_path):
+        """An absolute job_id must not read from an arbitrary filesystem location."""
+        from tools.cronjob_tools import _read_recent_outputs
+        root = tmp_path / "out"
+        root.mkdir()
+        outside = tmp_path / "elsewhere"
+        self._write(tmp_path, "elsewhere",
+                    "s.md", "# Cron Job: SECRET\n\n**Job ID:** x\n**Run Time:** t\n\n---\n\nsecret\n")
+        got = _read_recent_outputs(job_id=str(outside), output_root=root, limit=5)
+        assert got == [], "absolute job_id escaped the output root"
+
+    def test_output_action_forwards_limit_through_registered_handler(self, monkeypatch):
+        """The registered `cronjob` tool handler must forward `limit`; the direct
+        cronjob() signature accepts it, but tool calls go through the handler's
+        arg mapping, which is the only path the model uses."""
+        import tools.cronjob_tools as ct
+        from tools.registry import registry
+        captured = {}
+
+        def _fake_read(job_id=None, limit=5, output_root=None):
+            captured["limit"] = limit
+            return []
+
+        monkeypatch.setattr(ct, "_read_recent_outputs", _fake_read)
+        handler = registry.get_entry("cronjob").handler
+        handler({"action": "output", "limit": 3})
+        assert captured["limit"] == 3, "handler did not forward limit (got default)"

tools/cronjob_tools.py

@@ -654,6 +654,118 @@ def _execute_job_now(job: Dict[str, Any]) -> Dict[str, Any]:
         except Exception:
             pass
         return {"claimed": True, "success": False, "error": str(e)}
+# ---------------------------------------------------------------------------
+# Recent delivered output (the "pull" side of cron session-awareness)
+# ---------------------------------------------------------------------------
+#
+# Cron deliveries do NOT land in the interactive conversation history (that was
+# removed in #2313 because assistant-role mirrors broke message alternation).
+# These helpers let the agent read back what its jobs delivered, on demand, by
+# parsing the per-run markdown saved under ~/.hermes/cron/output/<job_id>/.
+#
+# The saved-file formats are fixed by the scheduler (cron/scheduler.py):
+#   - agent-mode:  "## Response\n\n<delivered text>"
+#   - no_agent:    "<header>\n---\n\n<script stdout>"
+
+
+def _extract_delivered_content(text: str) -> str:
+    """Pull the delivered message body out of a saved cron output document.
+
+    Tries the two real delivery shapes in order, then falls back to stripping
+    the leading metadata header so status/failure docs still yield something.
+    """
+    m = re.search(r"\n##\s*Response\s*\n", text)
+    if m:
+        return text[m.end():].strip()
+    m = re.search(r"\n---[ \t]*\n", text)
+    if m:
+        return text[m.end():].strip()
+    lines = text.splitlines()
+    body_start = 0
+    for i, line in enumerate(lines):
+        if line.startswith(("#", "**")) or not line.strip():
+            body_start = i + 1
+        else:
+            break
+    return "\n".join(lines[body_start:]).strip()
+
+
+def _parse_output_file(path: Path) -> Dict[str, Any]:
+    """Parse one saved cron output ``.md`` into structured fields."""
+    text = path.read_text(encoding="utf-8", errors="replace")
+
+    def _grab(pattern: str) -> Optional[str]:
+        m = re.search(pattern, text, re.MULTILINE)
+        return m.group(1).strip() if m else None
+
+    return {
+        "job_id": _grab(r"^\*\*Job ID:\*\*\s*(.+)$"),
+        "name": _grab(r"^#\s*Cron Job:\s*(.+)$"),
+        "run_time": _grab(r"^\*\*Run Time:\*\*\s*(.+)$"),
+        "content": _extract_delivered_content(text),
+    }
+
+
+def _is_safe_output_component(job_id: str) -> bool:
+    """Whether ``job_id`` is a single safe path component under the output root.
+
+    Job IDs scope a filesystem read under the cron output dir. A crafted or
+    unresolved id containing ``..``, path separators, or an absolute path would
+    let the read escape the sandbox and glob ``*.md`` elsewhere. Mirrors the
+    containment guard in ``cron.jobs._job_output_dir`` (which protects the
+    write/delete side) so the read side is equally contained.
+    """
+    text = str(job_id or "").strip()
+    if not text or text in {".", ".."} or "/" in text or "\\" in text:
+        return False
+    if Path(text).is_absolute() or Path(text).drive:
+        return False
+    return True
+
+
+def _read_recent_outputs(
+    job_id: Optional[str] = None,
+    limit: int = 5,
+    output_root: Optional[Path] = None,
+) -> List[Dict[str, Any]]:
+    """Return the most recent delivered cron outputs, newest first.
+
+    When ``job_id`` is given, scope to that one job; otherwise scan every job's
+    output directory and interleave by recency. ``output_root`` is injectable
+    for tests; production resolves it under the Hermes home.
+    """
+    if output_root is not None:
+        root = Path(output_root)
+    else:
+        from hermes_constants import get_hermes_home
+        root = get_hermes_home() / "cron" / "output"
+
+    if not root.is_dir():
+        return []
+
+    if job_id:
+        if not _is_safe_output_component(job_id):
+            logger.warning(
+                "Ignoring unsafe cron output job_id %r (path-escape attempt)", job_id
+            )
+            return []
+        job_dir = root / job_id
+        dirs = [job_dir] if job_dir.is_dir() else []
+    else:
+        dirs = [d for d in root.iterdir() if d.is_dir()]
+
+    files: List[Path] = []
+    for d in dirs:
+        files.extend(d.glob("*.md"))
+    files.sort(key=lambda p: p.stat().st_mtime, reverse=True)
+
+    outputs: List[Dict[str, Any]] = []
+    for p in files[: max(1, limit)]:
+        try:
+            outputs.append(_parse_output_file(p))
+        except Exception as e:  # never let one bad file sink the read
+            logger.debug("Skipping unreadable cron output %s: %s", p, e)
+    return outputs
 
 
 def cronjob(
@@ -677,6 +789,7 @@ def cronjob(
     workdir: Optional[str] = None,
     no_agent: Optional[bool] = None,
     attach_to_session: Optional[bool] = None,
+    limit: Optional[int] = None,
     task_id: str = None,
 ) -> str:
     """Unified cron job management tool."""
@@ -777,6 +890,19 @@ def cronjob(
             jobs = [_format_job(job) for job in list_jobs(include_disabled=include_disabled)]
             return json.dumps({"success": True, "count": len(jobs), "jobs": jobs}, indent=2)
 
+        if normalized in {"output", "outputs", "history"}:
+            # Read what jobs actually delivered. job_id is optional here: omit
+            # to scan all jobs by recency, or pass one to scope to a single job.
+            try:
+                resolved_id = resolve_job_ref(job_id)["id"] if job_id else None
+            except (AmbiguousJobReference, TypeError, KeyError):
+                resolved_id = job_id  # fall back to raw id; reader tolerates misses
+            outputs = _read_recent_outputs(job_id=resolved_id, limit=limit or 5)
+            return json.dumps(
+                {"success": True, "count": len(outputs), "outputs": outputs},
+                indent=2,
+            )
+
         if not job_id:
             return tool_error(f"job_id is required for action '{normalized}'", success=False)
 
@@ -973,6 +1099,7 @@ def cronjob(
 
 Use action='create' to schedule a new job from a prompt or one or more skills.
 Use action='list' to inspect jobs.
+Use action='output' to read back what jobs recently DELIVERED (their messages do not appear in this chat's history, so this is how you recall what a cron sent). Omit job_id for the latest across all jobs, or pass job_id to scope to one. Use 'limit' to control how many.
 Use action='update', 'pause', 'resume', 'remove', or 'run' to manage an existing job.
 
 To stop a job the user no longer wants: first action='list' to find the job_id, then action='remove' with that job_id. Never guess job IDs — always list first.
@@ -991,7 +1118,11 @@ def cronjob(
         "properties": {
             "action": {
                 "type": "string",
-                "description": "One of: create, list, update, pause, resume, remove, run. When action=create, the 'schedule' and 'prompt' fields are REQUIRED."
+                "description": "One of: create, list, output, update, pause, resume, remove, run. When action=create, the 'schedule' and 'prompt' fields are REQUIRED. action=output reads recently delivered cron results (job_id optional)."
+            },
+            "limit": {
+                "type": "integer",
+                "description": "For action=output: how many recent deliveries to return (default 5)."
             },
             "job_id": {
                 "type": "string",
@@ -1140,6 +1271,7 @@ def check_cronjob_requirements() -> bool:
         enabled_toolsets=args.get("enabled_toolsets"),
         workdir=args.get("workdir"),
         no_agent=args.get("no_agent"),
+        limit=args.get("limit"),
         task_id=kw.get("task_id"),
     ))(),
     check_fn=check_cronjob_requirements,