feat(cron): replay check-in reads Discord before repeating (RIG-8084)

cron/scheduler.py

@@ -2389,6 +2389,7 @@ def __init__(self, job: dict, job_id: str, execution_id: Optional[str]):
         self._cron_session_token = None
         self._non_dispatcher_token = None
         self._spend_token = None
+        self._replay_read_token = None
 
     def enter(self) -> None:
         # Scope cron approval policy; exit() RESETS via token (pinning "" would suppress the legacy
@@ -2416,9 +2417,25 @@ def exit(self) -> None:
         if self._spend_token is not None:
             from tools.mcp_tool_costs import unbind_run
             unbind_run(self._spend_token)
+        if self._replay_read_token is not None:
+            from tools.discord_tool import reset_cron_replay_read
+            reset_cron_replay_read(self._replay_read_token)
         for name in _CRON_DELIVERY_VARS:
             self._var_map[name].set("")
 
+    def bind_replay_read(self, job: dict) -> None:
+        """A replay may read its job's Discord deliver channels inside its window (DL-1691)."""
+        from cron.occurrences import claim_replay
+        from tools.discord_tool import bind_cron_replay_read
+        replay = claim_replay(job)
+        if replay is None:
+            return
+        channel_ids = frozenset(
+            str(target[key]) for target in _resolve_delivery_targets(job)
+            if str(target.get("platform") or "").lower() == "discord"
+            for key in ("chat_id", "thread_id") if target.get(key) not in (None, ""))
+        self._replay_read_token = bind_cron_replay_read(channel_ids, replay.get("started_at"))
+
 
 def _reload_dotenv_and_publish_delivery_target(job: dict) -> None:
     """Re-read .env for this run and publish the auto-deliver target into the session ContextVars."""
@@ -2607,6 +2624,7 @@ def run_job(
         if scope.workdir:
             logger.info("Job '%s': using task-scoped workdir %s", job_id, scope.workdir)
         _reload_dotenv_and_publish_delivery_target(job)
+        scope.bind_replay_read(job)
 
         jc = _load_cron_job_config(job, job_id, job_name)
         _cfg = jc.cfg

cron/scheduler_gateway_gate.py

@@ -49,6 +49,7 @@ def _note_fire_refused(job_ids, detail: str) -> None:
     """Stamp ``last_fire_error`` (shown by ``hermes cron list``) and drop the dispatch marks the
     due scan just wrote: the slot was never handed over, so nothing must "restore" or "reclaim" it."""
     from cron.jobs import _hermes_now, _jobs_lock, load_jobs, save_jobs
+    from cron.occurrences import slot_replay
 
     ids = set(job_ids)
     with _jobs_lock():
@@ -58,8 +59,7 @@ def _note_fire_refused(job_ids, detail: str) -> None:
                 continue
             job["last_fire_error"] = {"at": _hermes_now().isoformat(), "detail": detail[:500]}
             # A replay's slot was never handed over: keep it so the next tick still replays it.
-            pending = job.get("pending_slot")
-            if not (isinstance(pending, dict) and isinstance(pending.get("replay"), dict)):
+            if slot_replay(job) is None:
                 job.pop("pending_slot", None)
             if job.get("run_claim") is not None:
                 job["run_claim"] = None

cron/scheduler_prompt.py

@@ -301,6 +301,21 @@ def _run_time_line() -> str:
     return f"[CURRENT RUN TIME: {safe_strftime(now, '%A, %Y-%m-%d %H:%M %Z').strip()}{zone}]\n\n"
 
 
+def _replay_hint(replay: dict) -> str:
+    started_at = replay.get("started_at") or "unknown"
+    interrupted_at = replay.get("interrupted_at") or "unknown"
+    return (
+        f"REPLAY: A run of this same scheduled occurrence started at {started_at} "
+        f"and was stopped by a gateway restart at {interrupted_at}. It may already "
+        "have posted messages or delivered part of its result. Before you post "
+        "anything, read what is there: use the discord tool (list_forum_posts, "
+        "fetch_messages) on the channels and threads this job posts to. Look at "
+        f"every message sent at or after {started_at}. Do not post again anything "
+        "that is already there. Do only the steps that are missing. If nothing is "
+        "missing, respond with exactly [SILENT].\n\n"
+    )
+
+
 def _build_job_prompt(
     job: dict, prerun_script: Optional[tuple] = None, extra_prompt: Optional[str] = None,
     runtime_data_prompt: Optional[str] = None,
@@ -360,7 +375,9 @@ def _build_job_prompt(
         "use terminal and file tools instead. "
         if approval_context._get_cron_approval_mode() == "deny" else ""
     )
-    prompt = _CRON_HINT + _run_time_line() + execute_code_hint + prompt
+    from cron.occurrences import claim_replay
+    replay = claim_replay(job)
+    prompt = _CRON_HINT + (_replay_hint(replay) if replay is not None else "") + _run_time_line() + execute_code_hint + prompt
     skill_names = _job_skill_names(job)
     if not skill_names:
         return _scan_assembled_cron_prompt(

tests/cron/test_replay_deliver_read.py

@@ -0,0 +1,174 @@
+"""DL-1691: a replay may read its job's ``deliver`` channel inside its window, and nothing else.
+
+An ordinary cron run keeps the ``discord.post_channels`` boundary. A replay also reads its job's
+deliver channels (and threads under them), sees only this bot's messages at or after the replay's
+``started_at``, and never posts there.
+"""
+import json
+from datetime import UTC, datetime, timedelta
+from unittest.mock import patch
+
+import pytest
+
+from gateway.session_context import _VAR_MAP
+from tools import discord_tool
+from tools.discord_tool import bind_cron_replay_read, discord_core, reset_cron_replay_read
+
+BOT = "9999"
+USER = "42"
+POST = "500"      # discord.post_channels
+DELIVER = "700"   # the job's deliver channel, not in post_channels
+OTHER = "900"     # neither allowed nor the deliver channel
+GUILD = "1"
+EPOCH_MS = 1420070400000
+T0 = datetime(2026, 11, 10, 10, 0, tzinfo=UTC)
+STARTED = T0.isoformat()
+
+
+def ts(delta: timedelta = timedelta()) -> str:
+    return (T0 + delta).isoformat()
+
+
+def snow(delta: timedelta = timedelta()) -> str:
+    return str(int((T0 + delta).timestamp() * 1000 - EPOCH_MS) << 22)
+
+
+def message(msg_id: str, author: str, when: str) -> dict:
+    return {"id": msg_id, "content": "x", "author": {"id": author}, "timestamp": when}
+
+
+class FakeDiscord:
+    """Records REST calls and answers the few paths the replay read touches."""
+
+    def __init__(self, messages=(), threads=()):
+        self.messages = list(messages)
+        self.threads = list(threads)
+        self.calls: list[tuple[str, str]] = []
+
+    def __call__(self, method, path, token, params=None, body=None, timeout=15):
+        self.calls.append((method, path))
+        if path == "/users/@me":
+            return {"id": BOT}
+        if path == f"/channels/{DELIVER}":
+            return {"id": DELIVER, "type": 0, "guild_id": GUILD}
+        if path == f"/channels/{OTHER}":
+            return {"id": OTHER, "type": 0, "guild_id": GUILD}
+        if path == f"/channels/{DELIVER}/messages":
+            return self.messages
+        if path == f"/guilds/{GUILD}/threads/active":
+            return {"threads": self.threads}
+        if path == f"/channels/{DELIVER}/threads/archived/public":
+            return {"threads": [], "has_more": False}
+        raise AssertionError(f"unexpected Discord call {method} {path}")
+
+    def posted(self) -> bool:
+        return any(method == "POST" for method, _ in self.calls)
+
+
+@pytest.fixture
+def cron(monkeypatch):
+    monkeypatch.setenv("DISCORD_BOT_TOKEN", "test-token")
+    token = _VAR_MAP["HERMES_CRON_SESSION"].set("1")
+    yield
+    _VAR_MAP["HERMES_CRON_SESSION"].reset(token)
+
+
+@pytest.fixture
+def allowed():
+    with patch("tools.discord_tool._load_post_channels_config", return_value={POST}):
+        yield
+
+
+@pytest.fixture
+def replay(cron, allowed):
+    """A live replay window on ``DELIVER`` that started at ``STARTED``; closed after the test."""
+    token = bind_cron_replay_read(frozenset({DELIVER}), STARTED)
+    yield
+    reset_cron_replay_read(token)
+
+
+def test_ordinary_cron_read_outside_post_channels_is_refused(cron, allowed):
+    fake = FakeDiscord(messages=[message("1", BOT, ts(timedelta(minutes=1)))])
+    with patch("tools.discord_tool._discord_request", side_effect=fake):
+        result = json.loads(discord_core(action="fetch_messages", channel_id=DELIVER))
+    assert "outside discord.post_channels" in result["error"]
+    assert not any(path.endswith("/messages") for _, path in fake.calls)
+
+
+def test_replay_fetch_on_deliver_channel_returns_only_bot_messages_after_started(replay):
+    fake = FakeDiscord(messages=[
+        message("after-bot", BOT, ts(timedelta(minutes=1))),
+        message("after-user", USER, ts(timedelta(minutes=2))),
+        message("before-bot", BOT, ts(timedelta(minutes=-1))),
+    ])
+    with patch("tools.discord_tool._discord_request", side_effect=fake):
+        result = json.loads(discord_core(action="fetch_messages", channel_id=DELIVER))
+    assert [m["id"] for m in result["messages"]] == ["after-bot"]
+
+
+def test_replay_fetch_on_channel_outside_deliver_and_post_channels_is_refused(replay):
+    fake = FakeDiscord()
+    with patch("tools.discord_tool._discord_request", side_effect=fake):
+        result = json.loads(discord_core(action="fetch_messages", channel_id=OTHER))
+    assert "outside discord.post_channels" in result["error"]
+    assert not any(path.endswith("/messages") for _, path in fake.calls)
+
+
+def test_replay_post_to_deliver_channel_is_refused(replay):
+    fake = FakeDiscord()
+    with patch("tools.discord_tool._discord_request", side_effect=fake):
+        result = json.loads(discord_core(action="post_message", channel_id=DELIVER, content="x"))
+    assert "outside discord.post_channels" in result["error"]
+    assert not fake.posted()
+
+
+def test_replay_forum_list_keeps_only_bot_posts_started_or_active_after_started(replay):
+    threads = [
+        {"id": "11", "name": "checkin", "owner_id": BOT, "parent_id": DELIVER,
+         "thread_metadata": {"create_timestamp": ts(timedelta(minutes=1))},
+         "last_message_id": snow(timedelta(minutes=1))},
+        {"id": "12", "name": "checkin", "owner_id": USER, "parent_id": DELIVER,
+         "thread_metadata": {"create_timestamp": ts(timedelta(minutes=1))},
+         "last_message_id": snow(timedelta(minutes=1))},
+        {"id": "13", "name": "old", "owner_id": BOT, "parent_id": DELIVER,
+         "thread_metadata": {"create_timestamp": ts(timedelta(hours=-1))},
+         "last_message_id": snow(timedelta(hours=-1))},
+        {"id": "14", "name": "old-but-replied", "owner_id": BOT, "parent_id": DELIVER,
+         "thread_metadata": {"create_timestamp": ts(timedelta(hours=-1))},
+         "last_message_id": snow(timedelta(minutes=2))},
+    ]
+    fake = FakeDiscord(threads=threads)
+    with patch("tools.discord_tool._discord_request", side_effect=fake):
+        result = json.loads(discord_core(action="list_forum_posts", channel_id=DELIVER))
+    assert {post["id"] for post in result["posts"]} == {"11", "14"}
+
+
+def test_replay_with_unreadable_started_at_refuses_every_replay_read(cron, allowed):
+    fake = FakeDiscord(messages=[message("1", BOT, ts(timedelta(minutes=1)))])
+    token = bind_cron_replay_read(frozenset({DELIVER}), "not-a-time")
+    try:
+        with patch("tools.discord_tool._discord_request", side_effect=fake):
+            result = json.loads(discord_core(action="fetch_messages", channel_id=DELIVER))
+    finally:
+        reset_cron_replay_read(token)
+    assert "started_at is unreadable" in result["error"]
+
+
+def test_scheduler_binds_replay_window_only_for_a_replay_with_a_discord_deliver(monkeypatch):
+    from cron import scheduler
+
+    monkeypatch.setattr(
+        scheduler, "_resolve_delivery_targets",
+        lambda job: [{"platform": "discord", "chat_id": DELIVER, "thread_id": None}])
+    scope = object.__new__(scheduler._CronRunScope)
+    scope._replay_read_token = None
+
+    scope.bind_replay_read({"id": "job"})
+    assert discord_tool._CRON_REPLAY_READ.get() is None
+
+    scope.bind_replay_read({"id": "job", "fire_claim": {"replay": {"started_at": STARTED}}})
+    window = discord_tool._CRON_REPLAY_READ.get()
+    assert window.deliver_channel_ids == frozenset({DELIVER})
+    assert window.started_at == T0
+    reset_cron_replay_read(scope._replay_read_token)
+    assert discord_tool._CRON_REPLAY_READ.get() is None

tests/cron/test_replay_notice.py

@@ -0,0 +1,99 @@
+"""Replay prompts preserve the interrupted occurrence's delivery context."""
+import asyncio
+import json
+from contextlib import nullcontext
+from dataclasses import dataclass
+from types import SimpleNamespace
+
+import pytest
+
+from cron import scheduler_prompt
+
+START = "2026-11-07T10:00:00+00:00"
+STOP = "2026-11-07T10:05:00+00:00"
+
+
+def notice(start, stop):
+    return (
+        f"REPLAY: A run of this same scheduled occurrence started at {start} "
+        f"and was stopped by a gateway restart at {stop}. It may already "
+        "have posted messages or delivered part of its result. Before you post "
+        "anything, read what is there: use the discord tool (list_forum_posts, "
+        "fetch_messages) on the channels and threads this job posts to. Look at "
+        f"every message sent at or after {start}. Do not post again anything "
+        "that is already there. Do only the steps that are missing. If nothing is "
+        "missing, respond with exactly [SILENT].\n\n"
+    )
+
+
+@pytest.fixture
+def prompt_env(tmp_path, monkeypatch):
+    from cron import notepad
+    from tools import approval_context
+
+    monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+    monkeypatch.setattr(notepad, "render_notepad_section", lambda job_id: "")
+    monkeypatch.setattr(approval_context, "_get_cron_approval_mode", lambda: "approve")
+
+
+def test_replay_notice_follows_cron_hint_and_preserves_normal_prompt(prompt_env):
+    job = {"id": "job", "prompt": "Do the task."}
+    normal = scheduler_prompt._build_job_prompt(job, extra_prompt="Run detail")
+    assert normal == scheduler_prompt._CRON_HINT + "Do the task.\n\n## Run Context\nRun detail"
+    replay = dict(job, fire_claim={"replay": {"started_at": START, "interrupted_at": STOP}})
+    actual = scheduler_prompt._build_job_prompt(replay, extra_prompt="Run detail")
+    assert actual == scheduler_prompt._CRON_HINT + notice(START, STOP) + normal[len(scheduler_prompt._CRON_HINT):]
+
+
+@pytest.mark.parametrize("marker", [None, "broken", 42, {}])
+def test_malformed_replay_marker_uses_unknown_times(prompt_env, marker):
+    job = {"id": "job", "prompt": "Do the task.", "fire_claim": {"replay": marker}}
+    assert scheduler_prompt._build_job_prompt(job) == scheduler_prompt._CRON_HINT + notice("unknown", "unknown") + "Do the task."
+
+
+@pytest.mark.parametrize("with_skill", [False, True])
+def test_replay_notice_passes_assembled_prompt_scanner(prompt_env, monkeypatch, with_skill):
+    job = {"id": "job", "prompt": "Do the task.", "fire_claim": {"replay": {"started_at": START, "interrupted_at": STOP}}}
+    if with_skill:
+        job["skills"] = ["example"]
+        monkeypatch.setattr(scheduler_prompt, "_load_cron_skill_parts", lambda job, names: ["Use this skill."])
+    prompt = scheduler_prompt._build_job_prompt(job)
+    assert notice(START, STOP) in prompt
+    assert scheduler_prompt._scan_assembled_cron_prompt(prompt, job, has_skills=with_skill) == prompt
+
+
+def test_canonical_creation_and_execution_use_stored_replay(prompt_env, monkeypatch, tmp_path):
+    from cron import jobs, scheduler
+    from gateway import run, session_cron, session_local_recovery, session_policy
+
+    stored_job = {"id": "job", "prompt": "Do the task.", "fire_claim": {"replay": {"started_at": START, "interrupted_at": STOP}}}
+    monkeypatch.setattr(jobs, "get_job", lambda jid: stored_job if jid == "job" else None)
+    monkeypatch.setattr(run, "_load_gateway_config", dict)
+    monkeypatch.setattr(run, "_profile_runtime_scope", lambda home: nullcontext())
+    monkeypatch.setattr(scheduler, "_load_cron_job_config", lambda *args: SimpleNamespace(model="test"))
+
+    @dataclass
+    class Policy:
+        source: str = "cli"
+        platform: str = "cli"
+        model: str = "test"
+        request_json: str = ""
+    monkeypatch.setattr(session_policy, "build_policy", lambda *args, **kwargs: Policy())
+    monkeypatch.setattr(session_policy, "bind_launch_key", lambda authority, sid, policy, *args, **kwargs: policy)
+    monkeypatch.setattr(session_local_recovery, "local_identity", lambda *args: "session")
+    monkeypatch.setattr(session_local_recovery, "local_source", lambda *args: SimpleNamespace())
+    from gateway import session
+    monkeypatch.setattr(session.SessionEntry, "to_dict", lambda self: {})
+    authority = SimpleNamespace(
+        profile_id="profile", db=SimpleNamespace(db_path=str(tmp_path / "state.db"), get_session=lambda sid: None),
+        runner=SimpleNamespace(session_store=SimpleNamespace(_generate_session_key=lambda source: "route")),
+        sessions={"session": SimpleNamespace(source=SimpleNamespace(user_id="actor"))}, pending_results={},
+    )
+    actor = SimpleNamespace(subject="actor")
+    receipt, request_id = session_cron._create(authority, actor, {"job_id": "job", "request_id": "fire", "extra_prompt": None})
+    policy = Policy(**receipt["policy"])
+    assert json.loads(policy.request_json)["cron_job"] == stored_job
+    monkeypatch.setattr(session_cron, "_run_identified", lambda run_job, job, execution_id, **kwargs: (True, "", scheduler_prompt._build_job_prompt(job), None))
+    row = {"payload": {"text": ""}, "request_id": request_id, "principal_id": "actor", "admission_id": "admission"}
+    prompt = asyncio.run(session_cron.execute(authority, SimpleNamespace(session_id="session"), row, policy))
+    assert prompt == scheduler_prompt._CRON_HINT + notice(START, STOP) + "Do the task."

tests/cron/test_shutdown_replay.py

@@ -283,6 +283,19 @@ def test_refused_dispatch_keeps_replay_stamp(clock, monkeypatch):
     assert stored["last_fire_error"]["detail"] == "no gateway"
 
 
+def test_gateway_down_preserves_malformed_replay_slot(clock, monkeypatch):
+    from cron import scheduler_gateway_gate
+
+    job = jobs.create_job("report", "every 24h")
+    records = jobs.load_jobs()
+    records[0]["pending_slot"] = {"scheduled_at": job["next_run_at"], "replay": "invalid"}
+    jobs.save_jobs(records)
+    monkeypatch.setattr(scheduler_gateway_gate, "_gateway_unavailable", lambda _home: "down")
+    assert scheduler_gateway_gate.refuse_agent_jobs_without_gateway([job]) == []
+    stored = jobs.get_job(job["id"])
+    assert stored["pending_slot"]["replay"] == "invalid"
+
+
 def test_second_shutdown_does_not_stamp_again(clock):
     job = jobs.create_job("report", "every 24h", repeat=3)
     clock[0] = datetime.fromisoformat(job["next_run_at"])