config validate: check that a prompt_file exists

nerve/config_validate.py

@@ -304,7 +304,14 @@ def validate_config_bundle(
     if portable_only:
         cron_files = _tracked_cron_only(cron_files, portable_root, result)
 
-    _validate_cron(cron_files, result, strict_keys=strict_keys)
+    _validate_cron(
+        cron_files,
+        result,
+        strict_keys=strict_keys,
+        # Out-of-tree prompt files are only judgeable when the filesystem being
+        # validated is the one that will run the jobs — see _prompt_file_problem.
+        prompt_root=portable_root if portable_only else None,
+    )
     if config is not None:
         # Source runners are scheduled from the same parser as cron jobs, so a
         # typo'd sync schedule fails identically — and needs the same gate.
@@ -621,10 +628,18 @@ def _note_layers(
 
 
 def _validate_cron(
-    cron_files, result: ValidationResult, *, strict_keys: bool = False,
+    cron_files,
+    result: ValidationResult,
+    *,
+    strict_keys: bool = False,
+    prompt_root: Path | None = None,
 ) -> None:
     """Validate cron files strictly, including gate specs (which build_gates
-    otherwise swallows)."""
+    otherwise swallows).
+
+    ``prompt_root`` bounds where ``prompt_file`` existence is judged: set to the
+    tracked config root under ``portable_only``, ``None`` to judge every path.
+    """
     from nerve.cron.gates import GATE_REGISTRY, GateConfigError, build_gate
     from nerve.cron.jobs import load_jobs
 
@@ -657,6 +672,11 @@ def _validate_cron(
             problem = _schedule_problem(job.schedule)
             if problem:
                 result.errors.append(f"{where}: {problem}")
+            problem = _prompt_file_problem(job, prompt_root)
+            if problem:
+                # Fatal only without an inline prompt to fall back to.
+                bucket = result.warnings if job.prompt else result.errors
+                bucket.append(f"{where}: {problem}")
             for spec in _job_gate_specs(job, where, result):
                 problem = _gate_spec_problem(spec)
                 if problem:
@@ -699,6 +719,58 @@ def _validate_cron(
                     )
 
 
+def _prompt_file_problem(job: Any, root: Path | None = None) -> str | None:
+    """Why *job*'s ``prompt_file`` would not be readable when the job fires.
+
+    ``resolve_prompt`` re-reads the file on **every run** — that is what makes a
+    prompt editable without a restart — so a path that never resolves is not
+    caught at load. The job loads, validates, schedules, and then fails only when
+    it fires, which for a nightly job can be a day later.
+
+    A missing file is fatal exactly when there is no inline ``prompt`` to fall
+    back to. With one it degrades quietly to that instead, which is survivable
+    but almost never what the author meant: a fallback is typically a short
+    summary of a long prompt, so the job silently runs a lesser version of
+    itself. That earns a warning, not an error.
+
+    Three cases are deliberately not flagged:
+
+    * a ``workflow`` job never calls ``resolve_prompt`` at all, so its prompt
+      settings are inert;
+    * a path still carrying a literal ``${VAR}`` is unset-by-definition here and
+      gets the same leniency as every other unresolved reference in the bundle —
+      the environment that runs the daemon is not the one validating it;
+    * a path outside *root*, when a *root* is given. Under ``portable_only`` the
+      question is whether the shared bundle is sound, and a bundle cannot be
+      expected to carry a file the author deliberately kept on the machine. Same
+      substitution ``_tracked_cron_only`` refuses one directory over: judging a
+      config repo by what the reviewing filesystem happens to have. With no
+      *root* the filesystem being validated is the one that will run the jobs,
+      so every path is fair game.
+    """
+    if getattr(job, "workflow", None) is not None or not job.prompt_file:
+        return None
+    if _is_unresolved(job.prompt_file):
+        return None
+    path = job.prompt_path
+    if path is None or path.is_file():
+        return None
+    if root is not None and not path.is_relative_to(root):
+        return None
+    # A directory is as unreadable as an absent file (read_text raises), and it
+    # is the likelier typo of the two — `prompts/` instead of `prompts/x.md`.
+    what = "is a directory, not a file" if path.is_dir() else "does not exist"
+    if job.prompt:
+        return (
+            f"prompt_file {path} {what}, so every run falls back to the inline "
+            f"prompt instead — which is rarely the same instruction"
+        )
+    return (
+        f"prompt_file {path} {what} and the job has no inline prompt to fall "
+        f"back to, so every run of it fails"
+    )
+
+
 def _schedule_problem(schedule: Any) -> str | None:
     """Why the daemon would not run *schedule* as its author wrote it.
 

tests/test_config_validate.py

@@ -628,6 +628,155 @@ def test_unset_env_ref_in_a_schedule_is_not_flagged(self, tmp_path, monkeypatch)
         assert any("SYNC_SCHED" in i for i in result.info)
 
 
+class TestPromptFileChecking:
+    """A prompt_file that will not resolve must fail here, not when the job fires.
+
+    resolve_prompt re-reads the file on every run, so nothing about a bad path is
+    known at load: the job validates, schedules, and then fails on its own
+    cadence — a day later for a nightly job.
+    """
+
+    def _run(self, tmp_path, job_extra, *, make=None):
+        ws = tmp_path / "ws"
+        _jobs(ws, [{"id": "j", "schedule": "1h", **job_extra}])
+        if make:
+            target = ws / "config" / "cron" / make
+            target.parent.mkdir(parents=True, exist_ok=True)
+            target.write_text("do the thing\n", encoding="utf-8")
+        return validate_config_bundle(
+            _cfg(tmp_path, workspace=ws), workspace_override=ws,
+        )
+
+    def test_existing_prompt_file_passes(self, tmp_path):
+        result = self._run(
+            tmp_path, {"prompt_file": "prompts/j.md"}, make="prompts/j.md",
+        )
+
+        assert result.ok, result.errors
+        assert result.warnings == []
+
+    def test_missing_prompt_file_without_fallback_is_an_error(self, tmp_path):
+        result = self._run(tmp_path, {"prompt_file": "prompts/typo.md"})
+
+        assert not result.ok
+        assert any("prompts/typo.md" in e for e in result.errors)
+
+    def test_the_error_names_the_job_and_the_consequence(self, tmp_path):
+        result = self._run(tmp_path, {"prompt_file": "prompts/typo.md"})
+
+        assert any(
+            "job 'j'" in e and "every run of it fails" in e for e in result.errors
+        )
+
+    def test_missing_prompt_file_with_inline_fallback_is_only_a_warning(
+        self, tmp_path,
+    ):
+        result = self._run(
+            tmp_path, {"prompt_file": "prompts/typo.md", "prompt": "fallback"},
+        )
+
+        assert result.ok, result.errors
+        assert any("prompts/typo.md" in w for w in result.warnings)
+
+    def test_a_directory_is_reported_as_such(self, tmp_path):
+        result = self._run(
+            tmp_path, {"prompt_file": "prompts"}, make="prompts/j.md",
+        )
+
+        assert not result.ok
+        assert any("is a directory" in e for e in result.errors)
+
+    def test_relative_path_resolves_against_the_jobs_file(self, tmp_path):
+        """Not the process's working directory — that is where cron resolves it."""
+        result = self._run(
+            tmp_path, {"prompt_file": "prompts/j.md"}, make="prompts/j.md",
+        )
+
+        assert result.ok, result.errors
+
+    def test_inline_prompt_only_is_not_flagged(self, tmp_path):
+        result = self._run(tmp_path, {"prompt": "hi"})
+
+        assert result.ok, result.errors
+        assert result.warnings == []
+
+    def test_unset_env_ref_in_a_prompt_file_is_not_flagged(self, tmp_path):
+        result = self._run(tmp_path, {"prompt_file": "${PROMPTS_DIR}/j.md"})
+
+        assert result.ok, result.errors
+        assert not any("prompt_file" in w for w in result.warnings)
+
+    def test_a_workflow_job_is_not_flagged(self, tmp_path):
+        """A workflow job never calls resolve_prompt, so its path is inert."""
+        result = self._run(tmp_path, {
+            "prompt_file": "prompts/typo.md",
+            "workflow": {
+                "engine": "claude-workflow", "prompt": "go", "budget_usd": 1,
+            },
+        })
+
+        assert result.ok, result.errors
+
+    def test_every_bad_job_is_reported_not_just_the_first(self, tmp_path):
+        ws = tmp_path / "ws"
+        _jobs(ws, [
+            {"id": "a", "schedule": "1h", "prompt_file": "prompts/a.md"},
+            {"id": "b", "schedule": "1h", "prompt_file": "prompts/b.md"},
+        ])
+        result = validate_config_bundle(
+            _cfg(tmp_path, workspace=ws), workspace_override=ws,
+        )
+
+        assert not result.ok
+        assert sum("job 'a'" in e or "job 'b'" in e for e in result.errors) == 2
+
+    def test_an_out_of_tree_path_is_judged_when_validating_this_machine(
+        self, tmp_path,
+    ):
+        """No portable_only: the filesystem checked is the one that will run it."""
+        ws = tmp_path / "ws"
+        _jobs(ws, [
+            {"id": "j", "schedule": "1h",
+             "prompt_file": str(tmp_path / "elsewhere" / "j.md")},
+        ])
+        result = validate_config_bundle(
+            _cfg(tmp_path, workspace=ws), workspace_override=ws,
+        )
+
+        assert not result.ok
+
+    def test_portable_only_does_not_judge_an_out_of_tree_path(self, tmp_path):
+        """A shared bundle cannot be expected to carry a machine-local prompt,
+        and failing a config repo over one is the same substitution
+        _tracked_cron_only refuses one directory over."""
+        ws = tmp_path / "ws"
+        _settings(ws, "timezone: UTC\n")
+        _jobs(ws, [
+            {"id": "j", "schedule": "1h",
+             "prompt_file": str(tmp_path / "elsewhere" / "j.md")},
+        ])
+        result = validate_config_bundle(
+            _cfg(tmp_path, workspace=ws), workspace_override=ws,
+            portable_only=True,
+        )
+
+        assert result.ok, result.errors
+
+    def test_portable_only_still_judges_a_path_inside_the_bundle(self, tmp_path):
+        """The footgun this check exists for: the in-repo prompts/<id>.md
+        convention, which the bundle does carry and CI can therefore verify."""
+        ws = tmp_path / "ws"
+        _settings(ws, "timezone: UTC\n")
+        _jobs(ws, [{"id": "j", "schedule": "1h", "prompt_file": "prompts/typo.md"}])
+        result = validate_config_bundle(
+            _cfg(tmp_path, workspace=ws), workspace_override=ws,
+            portable_only=True,
+        )
+
+        assert not result.ok
+        assert any("prompts/typo.md" in e for e in result.errors)
+
+
 class TestGateSpecChecking:
     """Three cases, and only three: structure is always enforced, a built-in
     type is checked in full, anything else is reported as unverified."""