Recheck runs still recorded as running before the completed backlog

docs/incremental-history.md

@@ -45,7 +45,7 @@ Actions 历史按固定 created 时间窗遍历，遇到 GitHub 每次筛选查
 
 报告只在内存下载、解析。公开历史不保存原始 ZIP、日志、截图、trace 或逐用例执行内容（保留期内 E2E 运行的步骤与 HTML 报告由独立的[执行记录](e2e-run-records.md)按产物有效期展示，不进入历史）；仅保留各层 total / passed / failed / skipped / flaky、可读覆盖率与 GitHub 链接。已有数值不因产物被删除／过期清零。接入前已过期且未采集到的数据保持未知，不用 job success 或测试文件数推断用例通过率。
 
-运行中的 run 每 2 分钟复查，先完成的层的报告不代表整次运行；运行完成时立即读取其余产物。完成的 attempt 的 job 与产物不再变化：报告全部读到即移出复查队列；没有可读报告的，完成后再看一次（防止产物列表延迟）后移出；只有详情读取失败的继续按退避重试，最长一周；长期记录由对账再次检查。解析版本改变且报告仍可读时重算，过期报告保留旧版本指标。当前产物预算为单 ZIP 80 MiB、展开 160 MiB／3,000 条目；超预算明确为不可读取，不清除先前指标。
+运行中的 run 每 2 分钟复查，先完成的层的报告不代表整次运行；运行完成时立即读取其余产物。run 列表按创建时间增量读取，已经读过的 run 状态变化只能靠复查队列更新，所以每轮先复查 Bot 通知的重跑，再复查队列里记录仍为排队或运行中的 run，之后才按「编号最新」和「到期最早」各占一半处理其余积压；这样已完成 run 的积压或下载预算用尽，都不会让已经结束的工作流一直显示为运行中。完成的 attempt 的 job 与产物不再变化：报告全部读到即移出复查队列；没有可读报告的，完成后再看一次（防止产物列表延迟）后移出；只有详情读取失败的继续按退避重试，最长一周；长期记录由对账再次检查。解析版本改变且报告仍可读时重算，过期报告保留旧版本指标。当前产物预算为单 ZIP 80 MiB、展开 160 MiB／3,000 条目；超预算明确为不可读取，不清除先前指标。
 
 ## 只读取没见过的产物
 
@@ -63,4 +63,4 @@ Actions 产物上传后不再变化，看板已经读过的内容不再下载：
 
 Pages 每次部署仍整站上传 site，并非文件级增量部署。GitHub Pages 有站点容量边界；本实现不承诺无限字节容量。数据增长后需观测仓库与站点体积，再决定外置历史存储，不能悄悄改回业务条数截断。
 
-验证入口：`tests/test_incremental.py`、`tests/test_run_changes.py` 与 `tests/test_oidc_collection.py`。覆盖超旧上限回填、预算重启、权限失败、重开撤销计数、缓存和过期、attempt 隔离、按 ID 补采、详情失败保留待办及发布后确认。`PYTHONDONTWRITEBYTECODE=1 python3 test/journey-rerun-collection.py` 通过本地 HTTP 替身运行真实采集 CLI，验证窗口外的失败 run 重跑变绿、源端失败不覆盖旧结论、原子发布冲突不确认；不访问真实 GitHub，也不派发采集。
+验证入口：`tests/test_incremental.py`、`tests/test_run_changes.py` 与 `tests/test_oidc_collection.py`。覆盖超旧上限回填、预算重启、权限失败、重开撤销计数、缓存和过期、attempt 隔离、按 ID 补采、详情失败保留待办、记录为运行中的 run 先于已完成积压复查，以及发布后确认。`PYTHONDONTWRITEBYTECODE=1 python3 test/journey-rerun-collection.py` 通过本地 HTTP 替身运行真实采集 CLI，验证窗口外的失败 run 重跑变绿、源端失败不覆盖旧结论、原子发布冲突不确认；不访问真实 GitHub，也不派发采集。

gsb/sync.py

@@ -296,9 +296,21 @@ def graphql(self, *_):
         _enrich_prs_graphql(Context(OnePR(), self.cfg, self.now, self.meta), [row])
         self.history.put("prs", row)
 
+    def in_flight(self, key):
+        """A queued run still recorded as queued or in progress."""
+        if not key.startswith("run:"):
+            return False
+        row = self.history.get("runs", key[4:])
+        return bool(row) and row.get("status") != "completed"
+
     def pending(self, count=30, priority=(), only_priority=False):
         ready = [(key, value) for key, value in self.state["pending"].items() if value["due"] <= stamp(self.now) and (not only_priority or key in priority)]
-        urgent = [item for item in ready if item[0] in priority][:count]
+        # A run's status changes only through this queue: the run lists are read by
+        # creation time. Runs still recorded as running therefore go first, right after
+        # notified ones, so a backlog of completed runs (or a budget that runs out
+        # downloading their reports) cannot keep a finished workflow showing as running.
+        running = [] if only_priority else sorted((item for item in ready if item[0] not in priority and self.in_flight(item[0])), key=lambda item: (item[1]["due"], item[0]))
+        urgent = ([item for item in ready if item[0] in priority] + running)[:count]
         urgent_keys = {key for key, _ in urgent}
         remaining = count - len(urgent)
         rest = [item for item in ready if item[0] not in urgent_keys]

tests/test_run_changes.py

@@ -105,3 +105,23 @@ def test_receipts_and_input_are_nonsecret_bounded_and_source_scoped(self):
         write_receipt(receipt, REPO, [], 'a' * 40); self.assertFalse(receipt.exists())
         write_receipt(receipt, REPO, [hint()], 'b' * 40)
         self.assertEqual(json.loads(receipt.read_text()), {'source': REPO, 'runs': [hint()], 'commit': 'b' * 40})
+    def test_a_run_recorded_as_running_is_rechecked_before_the_completed_backlog(self):
+        # 39 completed runs wait for inspection. Run 20 was saved while running and has
+        # finished since; it is neither among the newest IDs nor among the oldest due.
+        finished = dict(run(20), status='completed', conclusion='success', updated_at=stamp(NOW + timedelta(hours=3)))
+        gh = RerunSource([finished])
+        sync = Sync(gh, self.root, REPO, now=NOW)
+        for number in range(1, 41):
+            sync.save_run(dict(run(number), status='in_progress', conclusion=None) if number == 20 else run(number))
+        inspected = []
+        def inspect(_gh, _cfg, record, **_kw):
+            inspected.append(record['id']); return self.report(_gh, _cfg, record)
+        sync.now = NOW + timedelta(hours=3)
+        with patch('gsb.sync.run_details', side_effect=inspect): sync.pending(count=4)
+        self.assertEqual(inspected[0], 20)
+        self.assertEqual(len(inspected), 4)
+        row = sync.history.get('runs', '20-1')
+        self.assertEqual((row['status'], row['conclusion']), ('completed', 'success'))
+        # Read in full, it leaves the queue; the rest of the backlog keeps its turn order.
+        self.assertNotIn('run:20-1', sync.state['pending'])
+