From a7dd07a6201d7cb17c22b182fb3c2dbc4bb1e80a Mon Sep 17 00:00:00 2001 From: itswl Date: Mon, 19 Oct 2026 10:09:31 +0800 Subject: [PATCH] watch: the cursor moves when the round is known to have landed MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The scanner wrote its new cursors before the model ran, so a round whose run failed was never looked at again and nothing said so; a second deployment running a copy of this scanner lost two rounds that way on 2026-10-12. A round that fires now stages its cursors (and Jira's state, on a staged copy) and the next tick settles them first, before the window check, against the node's own run record for the event the pipe numbered — the timer writes the pipe's answer to round.json after each POST. completed commits; failed, a POST that failed, an answer other than routed, or no finished record within WATCH_RUN_TIMEOUT puts the cursors back and the next scan re-offers from them with a line saying so; a run still going holds the tick rather than offering the same messages twice. After two replays and an hour since the first failure only the first lost round's content is skipped and the skipped span is named in the digest until a round lands. Jira is never skipped: its lookback re-reports the span, and a repeat is a nuisance where a miss is a failure. feeds in scan.json now always means committed; offered and previous keep their meaning for the signer and the poster. Twelve tests; the work compose names WATCH_ROUND_FILE and WATCH_NODE_RESULTS; decision note beside it. --- ...-when-the-round-is-known-to-have-landed.md | 55 +++++ deploy/docker-compose.work.yml | 7 + deploy/watch/tests/test_scan.py | 164 +++++++++++++ deploy/watch/watch_scan.py | 232 +++++++++++++++--- docs/deployments.md | 13 + hookprobe/examples/patrols/patrol-timer.sh | 23 ++ 6 files changed, 457 insertions(+), 37 deletions(-) create mode 100644 .agents/notes/implemented/2026-10-19-the-cursor-moves-when-the-round-is-known-to-have-landed.md diff --git a/.agents/notes/implemented/2026-10-19-the-cursor-moves-when-the-round-is-known-to-have-landed.md b/.agents/notes/implemented/2026-10-19-the-cursor-moves-when-the-round-is-known-to-have-landed.md new file mode 100644 index 00000000..946a1029 --- /dev/null +++ b/.agents/notes/implemented/2026-10-19-the-cursor-moves-when-the-round-is-known-to-have-landed.md @@ -0,0 +1,55 @@ +--- +title: The cursor moves when the round is known to have landed +status: implemented +date: 2026-10-19 +scope: stack +--- + +## Decision + +The watch scanner (`deploy/watch/watch_scan.py`) no longer commits a round's +cursors when it hands the digest over. A round that fires stages them under +`staged` in `scan.json`, and Jira runs against a staged copy of its state. The +next tick settles the previous round first, before the window check: + +- the timer writes what the pipe answered (`round.json`: posted, event id, + outcome, fired-at) after each POST (`patrol-timer.sh`); +- the scanner reads the node's run record for `probe:watch-due:` + under the timer's read-only `/node-state/results` mount; +- `completed` commits the staged cursors and the Jira copy; `failed`, a POST + that failed, an answer other than `routed`, a record that never appears + within `WATCH_RUN_TIMEOUT` put the cursors back and the next scan re-offers + from them with a line saying so; a run still going holds the tick; +- after two replays and an hour since the first failure, only the first lost + round's content is skipped (its cursors are committed) and the skipped span + is named in the digest until a round lands. Jira is never skipped. + +`feeds` in `scan.json` therefore always means "committed"; `offered` and +`previous` keep their meaning for the signer and the poster. + +## Why + +The cursor was the agent's promise written before the agent ran. A model error, +a dead gateway, a refused delivery or a timer restart between scan and POST all +lost that round's messages with no trace — the next tick scanned from the new +cursor as if the round had been judged. A second deployment running a copy of +this scanner measured it on 2026-10-12: two rounds lost to one configuration +fault, and the fix there (stage, then reconcile against the job's run record) +is the shape taken here with the pieces this family has — the pipe's event id +and the node's run record, both already on the timer's mounts. + +Holding the tick while a run is in flight is the price of not offering the same +messages twice; the node's runs finish in minutes and `WATCH_RUN_TIMEOUT` +bounds the wait. Giving up after two replays and an hour is what keeps a poison +round from stalling the feed forever, while three failures in fifteen minutes +still read as an outage to retry. + +## Consequences + +- `scan.json` gains `staged`, `replay` and `skipped`; `watch_feed_stats.py` + and the signer read `offered`/`previous` and are unaffected. +- A deployment without `WATCH_ROUND_FILE` or without the node's results mount + settles every fired round as lost next tick and replays it: loud, and the + compose names both. +- A storm-suppressed or unrouted round is replayed rather than counted as + judged, which is the correct reading of "the pipe did not run it". diff --git a/deploy/docker-compose.work.yml b/deploy/docker-compose.work.yml index c931a365..0cb311cc 100644 --- a/deploy/docker-compose.work.yml +++ b/deploy/docker-compose.work.yml @@ -979,6 +979,13 @@ services: # `origin` prefix the contract checker splits on. CHAT_TOOL_PREFIX: ${WATCH_CHAT_TOOL:?set WATCH_CHAT_TOOL in .env} WATCH_SCAN_FILE: /timer-state/scan.json + # The round's cursors are STAGED until the node's run for it has finished + # (watch_scan.py reconcile, 2026-10-19). The timer writes what the pipe + # answered here after each POST; the scanner reads the node's run record + # for that event under the read-only /node-state mount below. A round whose + # run failed is re-scanned from the old cursors next tick instead of lost. + WATCH_ROUND_FILE: /timer-state/round.json + WATCH_NODE_RESULTS: /node-state/results # 降频名单：这些 feed 的新消息攒 WATCH_BATCH_MINUTES 再一次性端给模型，游标 # 不动所以一条不丢。名字是内部标识符，只活在 .env。实测依据（2026-09-20， # 最近 90 轮）：47 轮的全部内容只来自三个高音量群 feed，其中 71% 整轮 diff --git a/deploy/watch/tests/test_scan.py b/deploy/watch/tests/test_scan.py index aca3049a..86cd6e2a 100644 --- a/deploy/watch/tests/test_scan.py +++ b/deploy/watch/tests/test_scan.py @@ -2,9 +2,12 @@ from __future__ import annotations +import json import sys from pathlib import Path +import pytest + sys.path.insert(0, str(Path(__file__).resolve().parents[1])) import watch_scan # noqa: E402 @@ -18,3 +21,164 @@ def test_fences_and_invisible_characters_do_not_reach_the_body(): def test_ordinary_text_is_untouched(): line = "- 10-12 09:00 someone：请看一下 `foo` 的状态，`` 也算" assert watch_scan.clean(line) == line + + +# ── the cursor moves when the round is known to have landed ────────────────── +# +# The scanner used to write the new cursors before the model ran; a round whose +# run failed was never looked at again. Now the cursors are staged and settled +# next tick against the node's run record for the event the pipe numbered. + + +@pytest.fixture +def state(tmp_path, monkeypatch): + results = tmp_path / "results" + results.mkdir() + monkeypatch.setattr(watch_scan, "SCAN", tmp_path / "scan.json") + monkeypatch.setattr(watch_scan, "ROUND", tmp_path / "round.json") + monkeypatch.setattr(watch_scan, "NODE_RESULTS", results) + monkeypatch.setattr(watch_scan, "JIRA_STATE", tmp_path / "jira.json") + monkeypatch.setattr(watch_scan, "JIRA_STAGED", tmp_path / "jira.staged.json") + return tmp_path + + +def _staged(feeds, staged_at=1000.0, replays=0, first_at=None, first_feeds=None): + return { + "round_at": staged_at, + "feeds": {"ops": 10.0}, + "staged": { + "feeds": feeds, + "staged_at": staged_at, + "replays": replays, + "first_at": first_at or staged_at, + "first_feeds": first_feeds or feeds, + }, + } + + +def _posted(state, event_id=77, fired_at=1001.0, posted=True, outcome="routed"): + (state / "round.json").write_text( + json.dumps({"posted": posted, "event_id": event_id, "outcome": outcome, "fired_at": fired_at}) + ) + + +def _ran(state, event_id, status): + (state / "results" / f"probe:watch-due:{event_id}.json").write_text( + json.dumps({"session_key": f"probe:watch-due:{event_id}", "status": status}) + ) + + +def test_a_finished_run_commits_the_staged_cursors_and_the_jira_copy(state): + _posted(state) + _ran(state, 77, "completed") + (state / "jira.staged.json").write_text('{"jira_last_check": 2}') + scan, verdict = watch_scan.reconcile(_staged({"ops": 20.0}), now=1100.0) + assert verdict == "done" and scan["feeds"] == {"ops": 20.0} and "staged" not in scan + assert json.loads((state / "jira.json").read_text()) == {"jira_last_check": 2} + assert not (state / "jira.staged.json").exists() + + +def test_a_failed_run_puts_the_cursors_back_and_asks_for_a_replay(state): + _posted(state) + _ran(state, 77, "failed") + (state / "jira.staged.json").write_text("{}") + scan, verdict = watch_scan.reconcile(_staged({"ops": 20.0}), now=1100.0) + assert verdict == "replay" and scan["feeds"] == {"ops": 10.0} + assert scan["replay"]["replays"] == 1 and scan["replay"]["first_feeds"] == {"ops": 20.0} + assert not (state / "jira.staged.json").exists(), "the Jira copy of a lost round is not kept" + + +@pytest.mark.parametrize( + "why", + ["not posted", "posted before the scan", "not routed", "no event id"], +) +def test_a_round_that_never_reached_a_run_is_lost(state, why): + if why == "not posted": + _posted(state, posted=False) + elif why == "posted before the scan": + _posted(state, fired_at=999.0) # the timer died between the scan and this round's POST + elif why == "not routed": + _posted(state, outcome="skipped") + else: + _posted(state, event_id=None) + _, verdict = watch_scan.reconcile(_staged({"ops": 20.0}), now=1100.0) + assert verdict == "replay", why + + +def test_a_run_still_going_holds_the_tick_and_a_run_that_vanished_does_not(state): + _posted(state, fired_at=1001.0) + _, young = watch_scan.reconcile(_staged({"ops": 20.0}), now=1001.0 + 60) + assert young == "pending", "no record yet and a minute old: the run may still be starting" + _ran(state, 77, "running") + _, running = watch_scan.reconcile(_staged({"ops": 20.0}), now=1001.0 + 600) + assert running == "pending" + _, stale = watch_scan.reconcile(_staged({"ops": 20.0}), now=1001.0 + watch_scan.RUN_TIMEOUT + 1) + assert stale == "replay", "past the run timeout, a run with no end is a lost round" + + +def test_after_two_replays_and_an_hour_only_the_first_lost_round_is_skipped(state): + _posted(state) + _ran(state, 77, "failed") + staged = _staged({"ops": 30.0}, staged_at=5000.0, replays=2, first_at=1000.0, first_feeds={"ops": 20.0}) + _posted(state, fired_at=5001.0) + scan, verdict = watch_scan.reconcile(staged, now=5100.0) + assert verdict == "gave-up" + assert scan["feeds"] == {"ops": 20.0}, "the first failed round's cursors: its content is what is skipped" + assert scan["skipped"]["to"] == 1000.0 and scan["skipped"]["attempts"] == 3 + assert "replay" not in scan + + +def test_two_replays_inside_the_hour_keep_trying(state): + _posted(state, fired_at=1801.0) + _ran(state, 77, "failed") + staged = _staged({"ops": 30.0}, staged_at=1800.0, replays=2, first_at=1000.0, first_feeds={"ops": 20.0}) + _, verdict = watch_scan.reconcile(staged, now=1900.0) + assert verdict == "replay", "three failures in fifteen minutes is an outage, not a poison round" + + +def test_nothing_staged_settles_nothing_and_drops_a_stray_jira_copy(state): + (state / "jira.staged.json").write_text("{}") + scan, verdict = watch_scan.reconcile({"round_at": 1.0, "feeds": {"ops": 10.0}}, now=2.0) + assert verdict == "none" and scan["feeds"] == {"ops": 10.0} + assert not (state / "jira.staged.json").exists() + + +def test_a_delivered_round_clears_the_skipped_note(state): + _posted(state) + _ran(state, 77, "completed") + scan = _staged({"ops": 20.0}) + scan["skipped"] = {"from": 1.0, "to": 2.0, "attempts": 3} + settled, verdict = watch_scan.reconcile(scan, now=1100.0) + assert verdict == "done" and "skipped" not in settled + + +def test_the_run_record_is_found_by_its_key_even_under_another_file_name(state): + (state / "results" / "something-else.json").write_text( + json.dumps({"session_key": "probe:watch-due:5", "status": "completed"}) + ) + (state / "results" / "probe:watch-due:6.json").write_text( + json.dumps({"session_key": "probe:watch-due:6", "status": "failed"}) + ) + assert watch_scan.run_record(6)["status"] == "failed" + assert watch_scan.run_record(5) is None, "a record is found by its name, not by a scan of everything" + assert watch_scan.run_record(9) is None + + +def test_jira_runs_against_a_staged_copy_of_its_state(state, monkeypatch): + script = state / "fake_jira.py" + script.write_text( + "import json, os, pathlib\n" + "p = pathlib.Path(os.environ['ATL_STATE_FILE'])\n" + "before = json.loads(p.read_text()).get('jira_last_check') if p.exists() else None\n" + "p.write_text(json.dumps({'from': before, 'jira_last_check': 9}))\n" + "print('PROJ-1 [Open] a ticket')\n" + ) + monkeypatch.setattr(watch_scan, "JIRA", str(script)) + (state / "jira.json").write_text(json.dumps({"jira_last_check": 4})) + notes: list[str] = [] + out = watch_scan.run_jira(notes) + assert out.startswith("PROJ-1") and notes == [] + committed = json.loads((state / "jira.json").read_text()) + assert committed == {"jira_last_check": 4}, "the committed state is untouched" + staged = json.loads((state / "jira.staged.json").read_text()) + assert staged["from"] == 4, "the copy started from the committed state" diff --git a/deploy/watch/watch_scan.py b/deploy/watch/watch_scan.py index 4fd98a08..ae0c9d4b 100644 --- a/deploy/watch/watch_scan.py +++ b/deploy/watch/watch_scan.py @@ -21,6 +21,18 @@ `feeds`（读到哪了）归这个脚本，写在自己的 cursors.json；`reported`（真的 告诉过他什么）归节点，由 watch_report.py 在发信号的同一次调用里写。 +游标什么时候算数（2026-10-19） + 交出摘要的那一轮，新游标不直接落进 `feeds`，而是放在 `staged` 里；Jira 的 + 状态同样写在一份暂存副本上。下一轮开头先结算上一轮：timer 在 POST 之后写下 + 管道给的事件号（round.json），节点那边 `probe:watch-due:<事件号>` 的运行记录 + 说它 completed 就转正，failed、没投出去、管道没路由、超时没结果就丢掉暂存， + 从旧游标重扫，并在摘要里注明。重交两次、离第一次失败满一小时还不行，就只跳过 + 第一次失败那一轮的内容，之后的照常交出，跳过的时段写进摘要直到真正送达。 + 运行还在跑时这一轮不扫（扫了会把同一批消息交两次）。以前新游标在模型跑之前 + 就写死了，模型那轮一失败，这一轮的消息就再也没人看到——另一处跑着这份脚本的 + 部署 2026-10-12 就这样丢了两轮。Jira 不跳过：它的回看窗会把那段再报一遍， + 重复一条是麻烦，漏一条不是。 + 分成两个文件不是洁癖。一个文件两个写者，agent 写 reported 的同时这里写 feeds，读-改-写一重叠就丢数据。而且 timer 把节点状态挂成只读是有理由的 （判官不改证据，见 docker-compose.work.yml），这个脚本跑在 timer 里，本来 @@ -37,6 +49,7 @@ import json import os import re +import shutil import subprocess import sys import time @@ -53,6 +66,18 @@ # 只在 scan.json 还不存在时读一次，把老状态文件里的 feeds 接过来。没有它，头一轮 # 会把 47 个会话的当前时间戳全当成基线重新播种——正确，但白丢一轮的历史。 SEED = os.environ.get("WATCH_SEED_STATE") or "" +# timer 在每轮 POST 之后写的记录（patrol-timer.sh）：{"posted": bool, "event_id", "outcome", "fired_at"}。 +# 下一轮据此判断暂存的游标能不能转正。 +ROUND = Path(os.environ.get("WATCH_ROUND_FILE") or str(SCAN.parent / "round.json")) +# 节点的运行记录目录（只读挂载）：唯一能说「那一轮的运行跑完了、失败了、还是根本没发生」的地方。 +NODE_RESULTS = Path(os.environ.get("WATCH_NODE_RESULTS") or "/node-state/results") +# 投出去这么久还没有结果就当丢了；节点的运行上限远在这之下。 +RUN_TIMEOUT = float(os.environ.get("WATCH_RUN_TIMEOUT") or 1800) +MAX_REPLAYS = 2 # 同一批内容最多再交两次 +GIVE_UP_AFTER = 3600.0 # 而且离第一次失败至少满一小时才放弃 +# jira_watch.py 的状态文件（它自己也按这个变量找）；交出摘要的那一轮它写的是暂存副本。 +JIRA_STATE = Path(os.environ.get("ATL_STATE_FILE") or os.path.expanduser("~/.atlassian_watch_state.json")) +JIRA_STAGED = JIRA_STATE.with_name(JIRA_STATE.stem + ".staged" + JIRA_STATE.suffix) WINDOW = os.environ.get("WATCH_WINDOW") or "09:30-19:30" DOW = os.environ.get("WATCH_DOW") or "1-5" EXCLUDE = [s.strip() for s in (os.environ.get("WATCH_EXCLUDE") or "").split(",") if s.strip()] @@ -214,8 +239,18 @@ def run_jira(notes: list[str]) -> str: """ if not JIRA: return "" + # 这一轮读写的是暂存副本：和聊天游标一起转正或丢弃（见 reconcile）。 + try: + if JIRA_STATE.exists(): + shutil.copyfile(JIRA_STATE, JIRA_STAGED) + else: + JIRA_STAGED.unlink(missing_ok=True) + except OSError as exc: + notes.append(f"Jira 取不到：暂存状态写不了 {exc}") + return "" + env = {**os.environ, "ATL_STATE_FILE": str(JIRA_STAGED)} try: - p = subprocess.run([sys.executable, JIRA], capture_output=True, text=True, timeout=180) + p = subprocess.run([sys.executable, JIRA], capture_output=True, text=True, timeout=180, env=env) except (OSError, subprocess.SubprocessError) as exc: notes.append(f"Jira 取不到：{type(exc).__name__} {str(exc)[:200]}") return "" @@ -226,6 +261,89 @@ def run_jira(notes: list[str]) -> str: return (p.stdout or "").strip() +def run_record(event_id: object) -> dict | None: + """节点为管道编号 event_id 的那一轮留下的运行记录，没有就是 None。""" + key = f"probe:watch-due:{event_id}" + direct = _read(NODE_RESULTS / f"{key}.json") + if direct and str(direct.get("session_key") or "") == key: + return direct + try: + for path in NODE_RESULTS.glob("probe:watch-due:*.json"): # 文件名规则变了也找得到 + data = _read(path) + if data and str(data.get("session_key") or "") == key: + return data + except OSError: + pass + return None + + +def round_outcome(staged: dict, now: float) -> str: + """上一轮交出去的内容有没有被处理完：done、lost 或 pending（运行还在跑）。""" + posted = _read(ROUND) or {} + if not posted.get("posted") or float(posted.get("fired_at") or 0) < float(staged.get("staged_at") or 0): + return "lost" # 没投出去，或者 timer 在扫完和投出之间死了 + if posted.get("outcome") != "routed" or posted.get("event_id") in (None, ""): + return "lost" # 管道收了但没有路由（风暴熔断、门没开）：没有运行会发生 + status = str((run_record(posted.get("event_id")) or {}).get("status") or "") + if status == "completed": + return "done" + if status == "failed": + return "lost" + return "pending" if now - float(posted.get("fired_at") or 0) < RUN_TIMEOUT else "lost" + + +def reconcile(scan: dict, now: float) -> tuple[dict, str]: + """结算上一轮：运行跑完了就把暂存的游标转正，没跑完就丢掉暂存、从旧游标重来。 + + 返回 (新的 scan.json 内容, 结论)。结论是 none（没有暂存）、pending（还在跑， + 这一轮别扫）、done、replay（丢掉暂存，下一轮重扫并注明）或 gave-up（只跳过第 + 一次失败那一轮的内容，记下跳过的时段）。Jira 的暂存副本跟着转正或丢弃。 + """ + staged = scan.get("staged") + if not isinstance(staged, dict) or not staged: + JIRA_STAGED.unlink(missing_ok=True) # 没有暂存的摘要却有 Jira 暂存：那一轮扫到一半就死了 + return scan, "none" + verdict = round_outcome(staged, now) + if verdict == "pending": + return scan, "pending" + scan = {k: v for k, v in scan.items() if k != "staged"} + if verdict == "done": + scan["feeds"] = dict(staged.get("feeds") or {}) + if JIRA_STAGED.exists(): + JIRA_STAGED.replace(JIRA_STATE) + scan.pop("replay", None) + scan.pop("skipped", None) # 真正送达了一轮，之前跳过的时段已经说过 + return scan, "done" + JIRA_STAGED.unlink(missing_ok=True) + replays = int(staged.get("replays") or 0) + 1 + first_at = float(staged.get("first_at") or staged.get("staged_at") or now) + first_feeds = dict(staged.get("first_feeds") or staged.get("feeds") or {}) + if replays > MAX_REPLAYS and now - first_at >= GIVE_UP_AFTER: + scan["feeds"] = first_feeds # 只跳过第一次失败那一轮的内容；之后的消息下一轮照常交出去 + previous = scan.get("skipped") if isinstance(scan.get("skipped"), dict) else {} + scan["skipped"] = { + "from": previous.get("from") or scan.get("round_at") or first_at, + "to": first_at, + "attempts": int(previous.get("attempts") or 0) + replays, + } + scan.pop("replay", None) + return scan, "gave-up" + scan["replay"] = { + "replays": replays, + "failed_at": staged.get("staged_at"), + "first_at": first_at, + "first_feeds": first_feeds, + } + return scan, "replay" + + +def _hhmm(ts: object) -> str: + try: + return time.strftime("%m-%d %H:%M", time.localtime(float(ts))) # type: ignore[arg-type] + except (TypeError, ValueError, OverflowError, OSError): + return "?" + + def scan_chat( chat: Chat, cursors: dict[str, float], @@ -386,14 +504,24 @@ def trim( def main() -> int: now = time.time() + # 先结算上一轮，窗口外也结算：一天最后一轮的失败不该等到第二天早上才处理。 + scan_file = _read(SCAN) + verdict = "none" + if scan_file is not None: + scan_file, verdict = reconcile(scan_file, now) + if verdict != "none": + _write(SCAN, scan_file) if not in_window(now): return 0 # 一个字节都不输出 = 这一轮不烧钱 + if verdict == "pending": + return 0 # 上一轮的运行还在跑：现在扫会把同一批消息交两次 notes: list[str] = [] jira_out = run_jira(notes) - scan_file = _read(SCAN) cursors: dict[str, float] = dict((scan_file or {}).get("feeds") or {}) + replay: dict = scan_file.get("replay") if scan_file and isinstance(scan_file.get("replay"), dict) else {} + skipped: dict = scan_file.get("skipped") if scan_file and isinstance(scan_file.get("skipped"), dict) else {} unreachable: dict[str, dict] = dict((scan_file or {}).get("unreachable") or {}) pending: dict[str, float] = dict((scan_file or {}).get("pending") or {}) stats: dict[str, dict] = dict((scan_file or {}).get("stats") or {}) @@ -404,6 +532,7 @@ def main() -> int: chat = Chat() findings: list[tuple[str, list[tuple[float, str, str]]]] = [] handed: set[str] = set() + committed: dict[str, float] = dict(cursors) # 转正前 feeds 里仍然是这份 if cursors: findings = scan_chat(chat, cursors, now, notes, handed, unreachable, pending) else: @@ -430,43 +559,72 @@ def main() -> int: row["offered"] = int(row.get("offered") or 0) + 1 row["last"] = now - _write( - SCAN, - { - "round_at": now, - # 上一轮的 offer 和它的时钟，原样留一轮：一次跑过了下一跳的运行（排队、 - # 慢模型）投递时读到的是这一轮的 offer，而它拿到的是上一轮的。签名器和 - # watch_report.py 两边都认，各按自己那轮计数。再往前不留：运行上限 30 - # 分钟，跨不过两跳。 - "previous": { - "round_at": float((scan_file or {}).get("round_at") or 0), - "offered": dict((scan_file or {}).get("offered") or {}), - }, - "feeds": cursors, - # 够不着的源和上次提醒的时间——让「说一次、6 小时后再提醒」跨轮成立。 - "unreachable": unreachable, - # 这一轮真的端给模型看的会话，以及各自读到哪。契约检查器用它判断 - # 「报了一个没人给过它的会话」，watch_report.py 用它决定 reported - # 写什么值——两者读同一份，所以不会各自算出不同的答案。 - # offered 是「这一轮交给模型的全部主语」，不只是读成功的群：⚠️ 里点名的 - # 读不到的会话（模型会原样转述），和 Jira 段里的工单号（模型用 - # --origin "Jira / KEY" 上报，检查器取 " / " 后半段当主语）。少任何一类， - # 检查器就会把一次正确的转述判成「报了没人给过它的会话」——2026-09-07 - # 上午它就这样每个 tick 都在喊，六条假违约进了盯守群。 - "offered": offered_now, - # 降频攒着还没端出去的内容（feed → 内容最早出现时刻）。放行或清空后 - # 消失。写在这里是因为它是 scan.json 的形状之一，不是给模型看的。 - "pending": pending, - # 每 feed 的累计出场次数。和 offered 同一轮写、同一个键空间，供 - # watch_feed_stats.py 和 reported 侧的 counts 拼出「谁在灌水」—— - # 最近 90 轮里 47 轮的全部内容只来自三个群，这个数该是常设的， - # 不该每次都从留存的 brief 里反推。 - "stats": stats, + quiet = not (findings or notes or jira_out) + if replay and not quiet: + notes.append( + f"上一轮（{_hhmm(replay.get('failed_at'))}）交出的内容没有被处理完，这一轮从那之前的位置重新扫了一遍，" + f"所以下面可能有之前出现过的消息" + ) + if skipped and not quiet: + notes.append( + f"{_hhmm(skipped.get('from'))} 到 {_hhmm(skipped.get('to'))} 之间的消息连续 {skipped.get('attempts')} 次" + f"都没有处理完，已经跳过，不会再出现；要看就去聊天里翻" + ) + record: dict = { + "round_at": now, + # 上一轮的 offer 和它的时钟，原样留一轮：一次跑过了下一跳的运行（排队、 + # 慢模型）投递时读到的是这一轮的 offer，而它拿到的是上一轮的。签名器和 + # watch_report.py 两边都认，各按自己那轮计数。再往前不留：运行上限 30 + # 分钟，跨不过两跳。 + "previous": { + "round_at": float((scan_file or {}).get("round_at") or 0), + "offered": dict((scan_file or {}).get("offered") or {}), }, - ) - - if not (findings or notes or jira_out): + # feeds 在下面按这一轮有没有叫模型决定：转正过的，还是直接算数的。 + # 够不着的源和上次提醒的时间——让「说一次、6 小时后再提醒」跨轮成立。 + "unreachable": unreachable, + # 这一轮真的端给模型看的会话，以及各自读到哪。契约检查器用它判断 + # 「报了一个没人给过它的会话」，watch_report.py 用它决定 reported + # 写什么值——两者读同一份，所以不会各自算出不同的答案。 + # offered 是「这一轮交给模型的全部主语」，不只是读成功的群：⚠️ 里点名的 + # 读不到的会话（模型会原样转述），和 Jira 段里的工单号（模型用 + # --origin "Jira / KEY" 上报，检查器取 " / " 后半段当主语）。少任何一类， + # 检查器就会把一次正确的转述判成「报了没人给过它的会话」——2026-09-07 + # 上午它就这样每个 tick 都在喊，六条假违约进了盯守群。 + "offered": offered_now, + # 降频攒着还没端出去的内容（feed → 内容最早出现时刻）。放行或清空后 + # 消失。写在这里是因为它是 scan.json 的形状之一，不是给模型看的。 + "pending": pending, + # 每 feed 的累计出场次数。和 offered 同一轮写、同一个键空间，供 + # watch_feed_stats.py 和 reported 侧的 counts 拼出「谁在灌水」—— + # 最近 90 轮里 47 轮的全部内容只来自三个群，这个数该是常设的， + # 不该每次都从留存的 brief 里反推。 + "stats": stats, + } + if quiet: + # 没叫模型，没有什么会丢：新游标直接算数，Jira 的暂存副本也直接转正。 + record["feeds"] = cursors + if skipped: + record["skipped"] = skipped + try: + if JIRA_STAGED.exists(): + JIRA_STAGED.replace(JIRA_STATE) + except OSError: + pass + _write(SCAN, record) return 0 + # 叫了模型：feeds 里仍是转正过的游标，新游标暂存，下一轮结算（见 reconcile）。 + record["feeds"] = committed + if skipped: + record["skipped"] = skipped + record["staged"] = { + "feeds": cursors, + "staged_at": now, + "replays": int(replay.get("replays") or 0), + "first_at": replay.get("first_at") or now, + "first_feeds": replay.get("first_feeds") or cursors, + } + _write(SCAN, record) # 预算要连 notes 和 Jira 一起算：它们都进同一个正文，而 notes 在坏日子里最长 # （每个够不着的会话一条），正好是最需要预算的那天。 diff --git a/docs/deployments.md b/docs/deployments.md index fe91c16b..a930412a 100644 --- a/docs/deployments.md +++ b/docs/deployments.md @@ -86,6 +86,19 @@ generic — a prescan that exits quiet skips the round, and one that FAILS fires anyway, carrying the failure. A prescan that broke and a prescan that found nothing are the same silence from outside, and only one of them is a quiet day. +**The cursor moves when the round is known to have landed (2026-10-19).** The +scan used to write its new cursors before the model ran, so a round whose run +failed was never looked at again; a second deployment running a copy of the +scanner lost two rounds that way in one day. Now a round that fires stages its +cursors (and Jira's state) and the next tick settles them against the node's own +run record for the event the pipe numbered, which the timer writes down after +the POST: `completed` commits, `failed`, never-posted, never-routed or timed +out puts the cursors back and re-scans with a line saying so, a run still going +holds the tick rather than offering the same messages twice. After two replays +and an hour the first lost round's content is skipped, named in the digest +until a round lands. Jira is not skipped: its lookback re-reports the span, and +a repeat is a nuisance where a miss is a failure. + Two chats rather than one, which is the whole reason this is not the alert shape: a signal ("someone assigned you this") and a plan ("here is how it would be done") are read at different moments. One bot would interleave a five-second diff --git a/hookprobe/examples/patrols/patrol-timer.sh b/hookprobe/examples/patrols/patrol-timer.sh index 3b0d3d4e..b1beebd7 100644 --- a/hookprobe/examples/patrols/patrol-timer.sh +++ b/hookprobe/examples/patrols/patrol-timer.sh @@ -213,11 +213,34 @@ open(os.environ["BRIEF_OUT"], "w", encoding="utf-8").write( fi fi + posted=0 if out=$(bash "$HERE/patrol.sh" "$BODY" "$TITLE" 2>&1); then + posted=1 log "fired: $out" else log "FAILED (rc=$?): $out" fi + # What the pipe answered, for the prescan's next tick: the scanner stages the + # round's cursors and commits them only once the node's run for THIS event has + # finished (watch_scan.py reconcile). Without the event id it cannot ask, and + # a round that never posted must read as lost, not as landed. + if [ -n "${PATROL_PRESCAN:-}" ] && [ -n "${WATCH_ROUND_FILE:-}" ]; then + ROUND_OUT="$out" ROUND_POSTED="$posted" python3 - <<'PY' || log "round record NOT written to $WATCH_ROUND_FILE" +import json, os, pathlib, time +out = os.environ.get("ROUND_OUT", "") +event = outcome = None +try: + answer = json.loads([ln for ln in out.splitlines() if ln.strip()][-1]) if out.strip() else {} + event, outcome = answer.get("event_id"), answer.get("outcome") +except (ValueError, IndexError, AttributeError): + pass +path = pathlib.Path(os.environ["WATCH_ROUND_FILE"]) +tmp = path.with_suffix(".tmp") +tmp.write_text(json.dumps({"posted": os.environ.get("ROUND_POSTED") == "1", "event_id": event, + "outcome": outcome, "fired_at": time.time()}), encoding="utf-8") +tmp.replace(path) +PY + fi [ "$BODY" = "$BRIEF_NOW" ] || rm -f "$BODY" [ -z "$BRIEF_RENDERED" ] || rm -f "$BRIEF_RENDERED" done