diff --git a/cron/scheduler.py b/cron/scheduler.py index 6aae7a77a614..a429c9f343a4 100644 --- a/cron/scheduler.py +++ b/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 diff --git a/cron/scheduler_gateway_gate.py b/cron/scheduler_gateway_gate.py index 15cc6d9e7818..1e14fafa326e 100644 --- a/cron/scheduler_gateway_gate.py +++ b/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 diff --git a/cron/scheduler_prompt.py b/cron/scheduler_prompt.py index 6b7dca5ad11c..efba58951ca1 100644 --- a/cron/scheduler_prompt.py +++ b/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( diff --git a/tests/cron/test_replay_deliver_read.py b/tests/cron/test_replay_deliver_read.py new file mode 100644 index 000000000000..5943e75bc15a --- /dev/null +++ b/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 diff --git a/tests/cron/test_replay_notice.py b/tests/cron/test_replay_notice.py new file mode 100644 index 000000000000..8a6cad905b03 --- /dev/null +++ b/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." diff --git a/tests/cron/test_shutdown_replay.py b/tests/cron/test_shutdown_replay.py index d0bb2305ab6b..c01159a68cb2 100644 --- a/tests/cron/test_shutdown_replay.py +++ b/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"]) diff --git a/tools/discord_tool.py b/tools/discord_tool.py index 3116fa14f67e..7561f2fcad3b 100644 --- a/tools/discord_tool.py +++ b/tools/discord_tool.py @@ -13,6 +13,8 @@ import math import threading import time +from contextvars import ContextVar, Token +from dataclasses import dataclass from datetime import UTC, datetime import urllib.error import urllib.parse @@ -461,7 +463,8 @@ def _thread_created_at(thread: dict[str, Any]) -> tuple[int, str]: def _list_forum_posts( - token: str, channel_id: str, query: str = "", limit: int = 50, **_kwargs: Any, + token: str, channel_id: str, query: str = "", limit: int = 50, *, + replay_scope: tuple[str, float] | None = None, **_kwargs: Any, ) -> str: """List active and archived forum threads, newest first.""" channel = _discord_request("GET", f"/channels/{channel_id}", token) @@ -508,6 +511,7 @@ def _list_forum_posts( filtered = [ thread for thread in by_id.values() if query.casefold() in str(thread.get("name", "")).casefold() + and (replay_scope is None or _replay_thread_kept(thread, *replay_scope)) ] filtered.sort(key=lambda thread: _thread_created_at(thread)[0], reverse=True) posts = [] @@ -1059,7 +1063,8 @@ def check_discord_tool_requirements() -> bool: # ── handlers ───────────────────────────────────────────────────────────────── # Cron jobs run unattended over untrusted input, so there the tool only reads and posts inside -# ``discord.post_channels`` (a listed channel or a thread under one) for the forum workflow. +# ``discord.post_channels`` (a listed channel or a thread under one) for the forum workflow. A replay +# also reads its job's deliver channel inside its window (DL-1691); see ``_run_replay_read``. _CRON_ACTIONS = frozenset({"list_forum_posts", "create_forum_post", "post_message", "fetch_messages"}) @@ -1069,18 +1074,115 @@ def _in_cron_session() -> bool: return is_truthy_value(get_session_env("HERMES_CRON_SESSION", "")) -def _require_cron_scope(token: str, action: str, channel_id: str) -> None: +_REPLAY_READ_ACTIONS = frozenset({"list_forum_posts", "fetch_messages"}) +_REPLAY_PAGE_LIMIT = 100 +_REPLAY_MAX_PAGES = 10 # up to 1,000 messages after started_at; beyond that, has_more is set +_DISCORD_EPOCH_MS = 1420070400000 + + +@dataclass(frozen=True) +class CronReplayRead: + """A replay's read window (DL-1691): its job's Discord deliver channel IDs and its ``started_at``. + ``started_at`` is None when it cannot be parsed; every replay read is then refused.""" + + deliver_channel_ids: frozenset[str] + started_at: datetime | None + + +# Set for each cron run by ``cron.scheduler``; None for every other run. +_CRON_REPLAY_READ = ContextVar("discord_cron_replay_read", default=None) + + +def bind_cron_replay_read(deliver_channel_ids: frozenset[str], started_at: object) -> Token: + """Open this run's replay read window; pass the returned token to ``reset_cron_replay_read``.""" + return _CRON_REPLAY_READ.set(CronReplayRead(frozenset(deliver_channel_ids), _parse_started_at(started_at))) + + +def reset_cron_replay_read(token: Token) -> None: + _CRON_REPLAY_READ.reset(token) + + +def _parse_started_at(value: object) -> datetime | None: + try: + parsed = datetime.fromisoformat(str(value)) + except ValueError: + return None + return parsed if parsed.tzinfo is not None else None + + +def _require_cron_scope(token: str, action: str, channel_id: str) -> CronReplayRead | None: """In a cron session, refuse actions outside the forum workflow and any channel outside - ``discord.post_channels``; an unset allowlist refuses everything (fail closed).""" + ``discord.post_channels``; an unset allowlist refuses everything (fail closed). A replay's read + actions also reach its deliver channels, and the replay window is returned for those reads.""" if not _in_cron_session(): - return + return None if action not in _CRON_ACTIONS: raise ValueError( f"Action '{action}' is not available in cron jobs; allowed: {', '.join(sorted(_CRON_ACTIONS))}") allowed = _load_post_channels_config() if not allowed: raise ValueError("discord.post_channels is not set; cron jobs may not use the discord tool") - _require_post_scope(token, channel_id, allowed=allowed) + replay = _CRON_REPLAY_READ.get() + if replay is None or action not in _REPLAY_READ_ACTIONS: + _require_post_scope(token, channel_id, allowed=allowed) + return None + if replay.started_at is None: + raise ValueError("The replay's started_at is unreadable; refusing the replay read") + _require_post_scope(token, channel_id, allowed=allowed | replay.deliver_channel_ids) + return replay + + +def _snowflake_before(ms: int) -> str: + """A snowflake just below epoch milliseconds ``ms``: ``after=`` it returns messages sent at or after ``ms``.""" + return str(((ms - _DISCORD_EPOCH_MS) << 22) - 1) + + +def _replay_message_kept(msg: dict[str, Any], bot_id: str, started_ms: float) -> bool: + """This bot's message sent at or after ``started_ms``. A message with an unreadable time is dropped.""" + try: + sent_ms = datetime.fromisoformat(str(msg.get("timestamp"))).timestamp() * 1000 + except ValueError: + return False + return str((msg.get("author") or {}).get("id")) == bot_id and sent_ms >= started_ms + + +def _replay_thread_kept(thread: dict[str, Any], bot_id: str, started_ms: float) -> bool: + """A forum post this bot started at or after ``started_ms``, or one with activity since then + (the bot may have replied in it). Posts started by anyone else are never listed.""" + if str(thread.get("owner_id")) != bot_id: + return False + if _thread_created_at(thread)[0] >= started_ms: + return True + return int(thread.get("last_message_id") or 0) > int(_snowflake_before(int(started_ms))) + + +def _replay_fetch_messages(token: str, channel_id: str, bot_id: str, started_ms: float) -> str: + """This bot's messages in ``channel_id`` sent at or after ``started_ms``, read forward page by page. + ``before``, ``after`` and ``limit`` from the caller are ignored on a replay read.""" + after = _snowflake_before(int(started_ms)) + kept: list[dict[str, Any]] = [] + has_more = True + for _ in range(_REPLAY_MAX_PAGES): + page = json.loads(_fetch_messages(token, channel_id, limit=_REPLAY_PAGE_LIMIT, after=after))["messages"] + kept.extend(m for m in page if _replay_message_kept(m, bot_id, started_ms)) + if len(page) < _REPLAY_PAGE_LIMIT: + has_more = False + break + after = str(max(int(m["id"]) for m in page)) + return json.dumps({"messages": kept, "count": len(kept), "has_more": has_more}) + + +def _run_replay_read( + action: str, action_fn: Any, token: str, kwargs: dict[str, Any], replay: CronReplayRead, +) -> str: + """Run a replay's read: only this bot's posts and messages at or after the replay's ``started_at``.""" + started_ms = replay.started_at.timestamp() * 1000 + bot_id = str(_discord_request("GET", "/users/@me", token).get("id") or "") + if not bot_id: + raise ValueError("Could not resolve the bot user; refusing the replay read") + if action == "list_forum_posts": + return action_fn(token=token, replay_scope=(bot_id, started_ms), **kwargs) + return _replay_fetch_messages(token, str(kwargs["channel_id"]), bot_id, started_ms) _HANDLER_DEFAULTS = { @@ -1108,9 +1210,11 @@ def _run_discord_action(action: str, valid_actions: dict[str, Any], tool_label: if missing: return tool_error(f"Missing required parameters for '{action}': {', '.join(missing)}") try: - _require_cron_scope(token, action, kwargs.get("channel_id") or "") + replay = _require_cron_scope(token, action, kwargs.get("channel_id") or "") _require_mesh_scope( token, action, channel_id=kwargs.get("channel_id") or "", guild_id=kwargs.get("guild_id") or "") + if replay is not None: + return _run_replay_read(action, action_fn, token, kwargs, replay) return action_fn(token=token, **kwargs) except ValueError as e: return tool_error(str(e)) diff --git a/website/docs/developer-guide/cron-internals.md b/website/docs/developer-guide/cron-internals.md index acc764a56ae7..a4a9c25e28fe 100644 --- a/website/docs/developer-guide/cron-internals.md +++ b/website/docs/developer-guide/cron-internals.md @@ -249,6 +249,8 @@ unknown attempt remains in run history with the skipped detail. This does not co can still run. One-shot jobs and other unknown outcomes stay paused; recovery delivers their failure notice through the job's configured failure target. +A replay may also read its job's `deliver` Discord channel, and threads under it, for the duration of its window. The read returns only this bot's messages at or after the replay's `started_at`, and it never grants ordinary channel history. Posts there are still refused, and every other channel keeps the `discord.post_channels` boundary. + **Fire-claim lease during a run.** A firing run holds `fire_claim = {at, by}` and a heartbeat thread refreshes `at` every 60 s (the lease is 300 s). A heartbeat sample that reads the claim as someone else's is re-sampled once before it