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