feat(scheduler): gate DRM jobs and emit liveness

.env.example

@@ -154,6 +154,17 @@ SEARXNG_INSTANCE=http://localhost:8080
 # if you intentionally want scheduled scripts to run remotely.
 # ODYSSEUS_SCRIPT_HOST=localhost
 
+# Vault-backed DRM action guard. DRM run_local/run_script/ssh_command tasks
+# fail closed unless this checkout contains a current heartbeat whose enabled
+# actions and SHA-256 match automation/config/odysseus_actions.yaml.
+# ODYSSEUS_DRM_VAULT_ROOT=/home/agent/projects/vault-remote-upkeep
+# ODYSSEUS_DRM_HEARTBEAT_MAX_AGE_MINUTES=1800
+
+# Scheduled LLM/research jobs emit Activity liveness every five minutes and
+# time out after one hour by default. Both values remain bounded in code.
+# ODYSSEUS_TASK_LIVENESS_SECONDS=300
+# ODYSSEUS_MODEL_TASK_TIMEOUT_SECONDS=3600
+
 # Chat / agent attachment size cap in bytes (default: 10 MB).
 # Raise this for local installs that need larger PDFs or text documents.
 # Example: 52428800 = 50 MB.

src/drm_runtime_guard.py

@@ -0,0 +1,90 @@
+"""Fail-closed heartbeat guard for vault-backed DRM action tasks."""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import os
+import re
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any
+
+_ACTION_RE = re.compile(r"^  ([A-Za-z0-9][A-Za-z0-9_-]*):\s*(?:#.*)?$")
+_ENABLED_RE = re.compile(r"^    enabled:\s*(true|false)\b")
+
+
+def _parse_time(value: Any) -> datetime | None:
+    try:
+        parsed = datetime.fromisoformat(str(value or "").replace("Z", "+00:00"))
+    except ValueError:
+        return None
+    if parsed.tzinfo is None:
+        parsed = parsed.replace(tzinfo=timezone.utc)
+    return parsed.astimezone(timezone.utc)
+
+
+def _allowlist(path: Path) -> dict[str, bool]:
+    actions: dict[str, bool] = {}
+    in_actions = False
+    current = ""
+    for raw in path.read_text(encoding="utf-8").splitlines():
+        line = raw.rstrip()
+        if line.startswith("actions:"):
+            in_actions = True
+            current = ""
+            continue
+        if in_actions and line and not line.startswith(" "):
+            break
+        if not in_actions:
+            continue
+        match = _ACTION_RE.match(line)
+        if match:
+            current = match.group(1)
+            actions[current] = False
+            continue
+        enabled = _ENABLED_RE.match(line)
+        if current and enabled:
+            actions[current] = enabled.group(1) == "true"
+    return actions
+
+
+def _allowlist_hash(path: Path) -> str:
+    text = path.read_text(encoding="utf-8").replace("\r\n", "\n").replace("\r", "\n")
+    return hashlib.sha256(text.encode("utf-8")).hexdigest()
+
+
+def check_drm_runtime_guard(task_name: str, *, now: datetime | None = None) -> tuple[bool, str]:
+    """Return whether a DRM action may run under the configured vault heartbeat."""
+    if not str(task_name or "").strip().lower().startswith("drm "):
+        return True, "not-a-drm-task"
+    vault_root = Path(os.environ.get("ODYSSEUS_DRM_VAULT_ROOT", "")).expanduser()
+    if not str(vault_root) or str(vault_root) == ".":
+        return False, "ODYSSEUS_DRM_VAULT_ROOT is not configured"
+    allowlist_path = vault_root / "automation/config/odysseus_actions.yaml"
+    heartbeat_dir = vault_root / "automation/review/routine-reports/odysseus-heartbeat"
+    if not allowlist_path.is_file():
+        return False, f"DRM allowlist missing: {allowlist_path}"
+    candidates = sorted(heartbeat_dir.glob("*.heartbeat.json")) if heartbeat_dir.is_dir() else []
+    if not candidates:
+        return False, f"DRM heartbeat missing: {heartbeat_dir}"
+    try:
+        payload = json.loads(candidates[-1].read_text(encoding="utf-8"))
+    except (OSError, json.JSONDecodeError) as exc:
+        return False, f"DRM heartbeat unreadable: {exc}"
+    if payload.get("report_kind") != "odysseus-heartbeat":
+        return False, "DRM heartbeat report_kind is invalid"
+    generated = _parse_time(payload.get("generated_at"))
+    if generated is None:
+        return False, "DRM heartbeat generated_at is invalid"
+    current = now or datetime.now(timezone.utc)
+    max_age_minutes = max(1, int(os.environ.get("ODYSSEUS_DRM_HEARTBEAT_MAX_AGE_MINUTES", "1800")))
+    if (current.astimezone(timezone.utc) - generated).total_seconds() / 60 > max_age_minutes:
+        return False, "DRM heartbeat is stale"
+    if payload.get("actions_yaml_sha256") != _allowlist_hash(allowlist_path):
+        return False, "DRM heartbeat allowlist hash mismatch"
+    actions = _allowlist(allowlist_path)
+    enabled = payload.get("enabled_actions")
+    if not isinstance(enabled, list) or any(not actions.get(str(name), False) for name in enabled):
+        return False, "DRM heartbeat enabled_actions exceed the committed allowlist"
+    return True, "current heartbeat and allowlist verified"

src/task_scheduler.py

@@ -1,8 +1,10 @@
 """Background scheduler for ScheduledTask execution."""
 
 import asyncio
+import contextlib
 import json
 import logging
+import os
 import re
 import time
 import uuid
@@ -12,6 +14,14 @@
 logger = logging.getLogger(__name__)
 
 
+def _model_task_timeout_seconds() -> int:
+    try:
+        value = int(os.environ.get("ODYSSEUS_MODEL_TASK_TIMEOUT_SECONDS", "3600"))
+    except ValueError:
+        value = 3600
+    return max(300, value)
+
+
 def _utcnow() -> datetime:
     """Return naive UTC for task DB fields without using deprecated APIs."""
     return datetime.now(timezone.utc).replace(tzinfo=None)
@@ -274,6 +284,19 @@ def _set_run_progress(self, run_id: str, message: str):
         except Exception:
             logger.debug("Task progress update failed", exc_info=True)
 
+    async def _emit_run_liveness(self, run_id: str, task_name: str, interval_seconds: float | None = None):
+        """Persist a bounded progress heartbeat until the owning task cancels this coroutine."""
+        if interval_seconds is None:
+            try:
+                interval_seconds = float(os.environ.get("ODYSSEUS_TASK_LIVENESS_SECONDS", "300"))
+            except ValueError:
+                interval_seconds = 300.0
+        interval_seconds = max(0.01, interval_seconds)
+        while True:
+            await asyncio.sleep(interval_seconds)
+            stamp = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
+            self._set_run_progress(run_id, f"Running {task_name}; liveness {stamp}")
+
     def _mark_run_aborted(self, task_id: str, run_id: str | None = None, message: str = "Stopped by user") -> bool:
         """Mark an active run as aborted. Used by stop/cancel paths."""
         try:
@@ -667,6 +690,7 @@ async def _execute_task_locked(self, task_id: str, run_id: str, *, release_execu
         from core.database import SessionLocal, ScheduledTask, TaskRun
 
         db = SessionLocal()
+        liveness_task = None
         try:
             task = db.query(ScheduledTask).filter(ScheduledTask.id == task_id).first()
             if not task or task.status != "active":
@@ -702,6 +726,8 @@ async def _execute_task_locked(self, task_id: str, run_id: str, *, release_execu
                 db.add(run)
                 db.commit()
 
+            liveness_task = asyncio.create_task(self._emit_run_liveness(run_id, task.name or task.id))
+
             task_type = task.task_type or "llm"
 
             from src.builtin_actions import TaskDeferred, TaskNoop
@@ -718,12 +744,18 @@ async def _execute_task_locked(self, task_id: str, run_id: str, *, release_execu
                     if not success:
                         run.error = result
                 elif task_type == "research":
-                    result = await self._execute_research_task(task, db)
+                    result = await asyncio.wait_for(
+                        self._execute_research_task(task, db),
+                        timeout=_model_task_timeout_seconds(),
+                    )
                     run.status = "success"
                     run.result = result
                 else:
                     # LLM task — use agent loop for tool access
-                    result = await self._execute_llm_task(task, db)
+                    result = await asyncio.wait_for(
+                        self._execute_llm_task(task, db),
+                        timeout=_model_task_timeout_seconds(),
+                    )
                     run.status = "success"
                     run.result = result
                 # Record which model actually ran (resolved inside the executor).
@@ -935,6 +967,10 @@ async def _execute_task_locked(self, task_id: str, run_id: str, *, release_execu
             except Exception:
                 logger.exception("Task %s error-path failed unexpectedly", task_id)
         finally:
+            if liveness_task is not None:
+                liveness_task.cancel()
+                with contextlib.suppress(asyncio.CancelledError):
+                    await liveness_task
             db.close()
             handle = self._task_handles.get(task_id)
             if handle is asyncio.current_task():
@@ -1015,6 +1051,12 @@ async def _execute_action(self, task, run_id: str | None = None) -> tuple:
         if not action_fn:
             return f"Unknown action: {task.action}", False
 
+        if task.action in ("run_local", "run_script", "ssh_command"):
+            from src.drm_runtime_guard import check_drm_runtime_guard
+            allowed, reason = check_drm_runtime_guard(task.name)
+            if not allowed:
+                return f"DRM runtime guard refused actuation: {reason}", False
+
         from src.builtin_actions import TaskNoop
         try:
             # Pass task prompt as script/command for ssh_command/run_script actions.

tests/test_drm_runtime_guard.py

@@ -0,0 +1,82 @@
+from __future__ import annotations
+
+import asyncio
+import hashlib
+import json
+from datetime import datetime, timezone
+
+import pytest
+
+from src.drm_runtime_guard import check_drm_runtime_guard
+from src.task_scheduler import TaskScheduler, _model_task_timeout_seconds
+
+
+ALLOWLIST = "version: 1\nactions:\n  validate:\n    enabled: true\n  task-transition:\n    enabled: false\n"
+NOW = datetime(2026, 6, 22, 18, 0, tzinfo=timezone.utc)
+
+
+def _vault(tmp_path, *, enabled=None, generated_at="2026-06-22T17:00:00Z"):
+    config = tmp_path / "automation/config"
+    heartbeats = tmp_path / "automation/review/routine-reports/odysseus-heartbeat"
+    config.mkdir(parents=True)
+    heartbeats.mkdir(parents=True)
+    (config / "odysseus_actions.yaml").write_text(ALLOWLIST, encoding="utf-8")
+    digest = hashlib.sha256(ALLOWLIST.encode("utf-8")).hexdigest()
+    payload = {
+        "report_kind": "odysseus-heartbeat",
+        "generated_at": generated_at,
+        "actions_yaml_sha256": digest,
+        "enabled_actions": ["validate"] if enabled is None else enabled,
+    }
+    (heartbeats / "2026-06-22.heartbeat.json").write_text(json.dumps(payload), encoding="utf-8")
+    return tmp_path
+
+
+def test_guard_accepts_current_hash_matching_heartbeat(tmp_path, monkeypatch):
+    monkeypatch.setenv("ODYSSEUS_DRM_VAULT_ROOT", str(_vault(tmp_path)))
+    assert check_drm_runtime_guard("DRM Validate", now=NOW) == (
+        True,
+        "current heartbeat and allowlist verified",
+    )
+
+
+@pytest.mark.parametrize(
+    "enabled,generated_at,reason",
+    [
+        (["task-transition"], "2026-06-22T17:00:00Z", "enabled_actions exceed"),
+        (["validate"], "2026-06-20T00:00:00Z", "heartbeat is stale"),
+    ],
+)
+def test_guard_fails_closed_on_drift_or_staleness(tmp_path, monkeypatch, enabled, generated_at, reason):
+    monkeypatch.setenv("ODYSSEUS_DRM_VAULT_ROOT", str(_vault(tmp_path, enabled=enabled, generated_at=generated_at)))
+    allowed, message = check_drm_runtime_guard("DRM Validate", now=NOW)
+    assert not allowed
+    assert reason in message
+
+
+def test_guard_requires_configuration_for_drm_only(monkeypatch):
+    monkeypatch.delenv("ODYSSEUS_DRM_VAULT_ROOT", raising=False)
+    assert check_drm_runtime_guard("Calendar Classify Events", now=NOW)[0]
+    assert not check_drm_runtime_guard("DRM Validate", now=NOW)[0]
+
+
+def test_model_timeout_defaults_to_one_hour(monkeypatch):
+    monkeypatch.delenv("ODYSSEUS_MODEL_TASK_TIMEOUT_SECONDS", raising=False)
+    assert _model_task_timeout_seconds() == 3600
+    monkeypatch.setenv("ODYSSEUS_MODEL_TASK_TIMEOUT_SECONDS", "900")
+    assert _model_task_timeout_seconds() == 900
+
+
+@pytest.mark.asyncio
+async def test_liveness_emits_periodic_progress():
+    scheduler = TaskScheduler.__new__(TaskScheduler)
+    messages = []
+    scheduler._set_run_progress = lambda run_id, message: messages.append((run_id, message))
+    task = asyncio.create_task(scheduler._emit_run_liveness("run-1", "Long task", interval_seconds=0.01))
+    await asyncio.sleep(0.025)
+    task.cancel()
+    with pytest.raises(asyncio.CancelledError):
+        await task
+    assert messages
+    assert messages[0][0] == "run-1"
+    assert "liveness" in messages[0][1]