fix(cron): reject id mutation + validate output paths under OUTPUT_DIR

Two defense-in-depth fixes on cron output path handling:

1. cron/jobs.py:update_job() rejects mutation of the immutable 'id' field
   (raises ValueError). Dashboard PUT /api/cron/jobs/{id} converts this to
   HTTP 400. Without this, an attacker who can reach the update endpoint
   could rename a job's id to '../escape' and move its output directory
   outside OUTPUT_DIR.

2. cron/jobs.py:_job_output_dir() validates job IDs before composing
   paths: rejects '.', '..', '/', '\\', absolute paths, and Windows drive
   prefixes. Used by save_job_output() and remove_job() so legacy unsafe
   IDs (from before this guard) fail closed rather than half-applying a
   shutil.rmtree or output write outside the sandbox.

Tests:
  - update_job rejects {'id': '../escape'} without renaming
  - remove_job(legacy '../escape' id) raises ValueError without deleting
    files outside OUTPUT_DIR or removing the job from the store
  - save_job_output rejects '..', './escape', 'nested/escape',
    absolute paths
  - dashboard PUT /api/cron/jobs/{id} with {'id': '../escape'} returns
    400, job list unchanged

Salvaged from PR #29826 by @zapabob. Simplified implementation:
- Dropped a 23-line _validate_job_output_id() helper using Path.parts
  semantics. The inline check (path separators + dot-components +
  is_absolute) is shorter and behaviorally identical.
- Dropped the secondary OUTPUT_DIR.resolve()/relative_to() check —
  redundant once we reject any path separator at the input boundary.
- Dropped the _docs/2026-05-21_cron-output-path-hardening_codex.md
  planning artifact (we don't check planning docs into the repo).

Co-authored-by: teknium1 <127238744+teknium1@users.noreply.github.com>

cron/jobs.py

@@ -45,6 +45,28 @@ _jobs_file_lock = threading.Lock()
 OUTPUT_DIR = CRON_DIR / "output"
 ONESHOT_GRACE_SECONDS = 120
 
+# Fields on a cron job that must never change after creation. ``id`` is used
+# as a filesystem path component under ``OUTPUT_DIR``; allowing it to be
+# updated lets an unsafe value (``../escape``, absolute path, nested) leak
+# into output writes/deletes.
+_IMMUTABLE_JOB_FIELDS = frozenset({"id"})
+
+
+def _job_output_dir(job_id: str) -> Path:
+    """Resolve a job's output directory, rejecting any path-escape attempt.
+
+    Job IDs are filesystem path components under ``OUTPUT_DIR``. A legacy or
+    crafted ID containing ``..``, absolute paths, or nested separators would
+    allow output writes/deletes to escape the cron output sandbox. Reject
+    anything that isn't a single safe path component.
+    """
+    text = str(job_id or "").strip()
+    if not text or text in {".", ".."} or "/" in text or "\\" in text:
+        raise ValueError(f"Invalid cron job id for output path: {job_id!r}")
+    if Path(text).is_absolute() or Path(text).drive:
+        raise ValueError(f"Invalid cron job id for output path: {job_id!r}")
+    return OUTPUT_DIR / text
+
 
 def _normalize_skill_list(skill: Optional[str] = None, skills: Optional[Any] = None) -> List[str]:
     """Normalize legacy/single-skill and multi-skill inputs into a unique ordered list."""
@@ -728,6 +750,15 @@ def list_jobs(include_disabled: bool = False) -> List[Dict[str, Any]]:
 
 def update_job(job_id: str, updates: Dict[str, Any]) -> Optional[Dict[str, Any]]:
     """Update a job by ID, refreshing derived schedule fields when needed."""
+    # Block mutation of immutable fields. ``id`` in particular is a filesystem
+    # path component under OUTPUT_DIR — letting an update change it leaks
+    # path-escape values into output writes/deletes.
+    bad_fields = _IMMUTABLE_JOB_FIELDS.intersection(updates or {})
+    if bad_fields:
+        raise ValueError(
+            f"Cron job field(s) cannot be updated: {', '.join(sorted(bad_fields))}"
+        )
+
     jobs = load_jobs()
     for i, job in enumerate(jobs):
         if job["id"] != job_id:
@@ -845,9 +876,12 @@ def remove_job(job_id: str) -> bool:
     original_len = len(jobs)
     jobs = [j for j in jobs if j["id"] != canonical_id]
     if len(jobs) < original_len:
+        # Resolve the output dir BEFORE saving so a legacy unsafe ID (e.g.
+        # left over from before the create-time guard) fails closed without
+        # half-applying the removal.
+        job_output_dir = _job_output_dir(canonical_id)
         save_jobs(jobs)
         # Clean up output directory to prevent orphaned dirs accumulating
-        job_output_dir = OUTPUT_DIR / canonical_id
         if job_output_dir.exists():
             shutil.rmtree(job_output_dir)
         return True
@@ -1061,7 +1095,7 @@ def _get_due_jobs_locked() -> List[Dict[str, Any]]:
 def save_job_output(job_id: str, output: str):
     """Save job output to file."""
     ensure_dirs()
-    job_output_dir = OUTPUT_DIR / job_id
+    job_output_dir = _job_output_dir(job_id)
     job_output_dir.mkdir(parents=True, exist_ok=True)
     _secure_dir(job_output_dir)
     

hermes_cli/web_server.py

@@ -2711,7 +2711,10 @@ async def update_cron_job(job_id: str, body: CronJobUpdate, profile: Optional[st
     selected = profile or _find_cron_job_profile(job_id)
     if not selected:
         raise HTTPException(status_code=404, detail="Job not found")
-    job = _call_cron_for_profile(selected, "update_job", job_id, body.updates)
+    try:
+        job = _call_cron_for_profile(selected, "update_job", job_id, body.updates)
+    except ValueError as exc:
+        raise HTTPException(status_code=400, detail=str(exc)) from exc
     if not job:
         raise HTTPException(status_code=404, detail="Job not found")
     return job
@@ -2755,7 +2758,11 @@ async def delete_cron_job(job_id: str, profile: Optional[str] = None):
     selected = profile or _find_cron_job_profile(job_id)
     if not selected:
         raise HTTPException(status_code=404, detail="Job not found")
-    if not _call_cron_for_profile(selected, "remove_job", job_id):
+    try:
+        removed = _call_cron_for_profile(selected, "remove_job", job_id)
+    except ValueError as exc:
+        raise HTTPException(status_code=400, detail=str(exc)) from exc
+    if not removed:
         raise HTTPException(status_code=404, detail="Job not found")
     return {"ok": True}
 

tests/cron/test_jobs.py

@@ -232,6 +232,23 @@ class TestJobCRUD:
         assert remove_job(job["id"]) is True
         assert get_job(job["id"]) is None
 
+    def test_remove_job_rejects_unsafe_legacy_id_before_output_cleanup(self, tmp_cron_dir):
+        """Legacy unsafe IDs left over from before the create-time guard
+        must fail closed without half-applying the removal."""
+        job = create_job(prompt="Legacy unsafe", schedule="every 1h")
+        job["id"] = "../escape"
+        save_jobs([job])
+        outside = tmp_cron_dir / "escape"
+        outside.mkdir()
+        (outside / "keep.txt").write_text("keep", encoding="utf-8")
+
+        with pytest.raises(ValueError, match="output path"):
+            remove_job("../escape")
+
+        # Job should still be in the store and the escape dir untouched.
+        assert load_jobs()[0]["id"] == "../escape"
+        assert (outside / "keep.txt").exists()
+
     def test_remove_nonexistent_returns_false(self, tmp_cron_dir):
         assert remove_job("nonexistent") is False
 
@@ -300,6 +317,17 @@ class TestUpdateJob:
         result = update_job("nonexistent_id", {"name": "X"})
         assert result is None
 
+    def test_update_rejects_id_change(self, tmp_cron_dir):
+        """Job IDs are filesystem path components — must be immutable."""
+        job = create_job(prompt="Original", schedule="every 1h")
+
+        with pytest.raises(ValueError, match="id"):
+            update_job(job["id"], {"id": "../escape"})
+
+        # Original job still resolvable, no rename happened.
+        assert get_job(job["id"]) is not None
+        assert get_job("../escape") is None
+
 
 class TestPauseResumeJob:
     def test_pause_sets_state(self, tmp_cron_dir):
@@ -953,3 +981,16 @@ class TestSaveJobOutput:
         assert output_file.exists()
         assert output_file.read_text() == "# Results\nEverything ok."
         assert "test123" in str(output_file)
+
+    @pytest.mark.parametrize("bad_job_id", ["../escape", "nested/escape", ".", "..", ""])
+    def test_rejects_unsafe_job_id(self, tmp_cron_dir, bad_job_id):
+        """Path-escape attempts must fail closed and never create dirs."""
+        with pytest.raises(ValueError, match="output path"):
+            save_job_output(bad_job_id, "# Results")
+        assert not (tmp_cron_dir / "escape").exists()
+
+    def test_rejects_absolute_job_id(self, tmp_cron_dir):
+        """Absolute paths as job IDs must fail closed."""
+        with pytest.raises(ValueError, match="output path"):
+            save_job_output(str(tmp_cron_dir / "outside"), "# Results")
+        assert not (tmp_cron_dir / "outside").exists()

tests/hermes_cli/test_web_server_cron_profiles.py

@@ -131,6 +131,33 @@ async def test_cron_mutation_without_profile_finds_named_profile_job(isolated_pr
     assert worker_jobs[0]["enabled"] is False
 
 
+@pytest.mark.asyncio
+async def test_update_cron_job_rejects_id_mutation(isolated_profiles):
+    """Dashboard surfaces a 400 (not a 500 or silent rename) when an
+    id-mutation attempt is rejected by cron/jobs.update_job."""
+    from hermes_cli import web_server
+
+    worker_job = web_server._call_cron_for_profile(
+        "worker_alpha",
+        "create_job",
+        prompt="managed by named profile",
+        schedule="every 1h",
+        name="immutable-id-job",
+    )
+
+    with pytest.raises(HTTPException) as exc:
+        await web_server.update_cron_job(
+            worker_job["id"],
+            web_server.CronJobUpdate(updates={"id": "../escape"}),
+            profile="worker_alpha",
+        )
+
+    assert exc.value.status_code == 400
+    assert "id" in exc.value.detail
+    worker_jobs = await web_server.list_cron_jobs(profile="worker_alpha")
+    assert [job["id"] for job in worker_jobs] == [worker_job["id"]]
+
+
 @pytest.mark.asyncio
 async def test_cron_delete_with_profile_deletes_only_target_profile(isolated_profiles):
     from hermes_cli import web_server