mirror of
https://github.com/QwenLM/qwen-code.git
synced 2026-11-07 06:19:43 -10:00
fix(channels): watch every open bound PR and keep polling during public turns (#16502)
* fix(channels): watch every open bound PR and keep polling during public turns The PR state watch selected bound PRs purely by binding recency under a hard cap of 30, so older open bindings were never polled once newer bindings (including already-merged ones) filled the budget, and merged bindings outside the cap never received the terminal observation that would retire them. Selection now watches open PRs first and drains terminal-pending bindings with the leftover budget; the cap becomes the configurable prWatchLimit (default 30, range 1-500, fractional values round up), exposed in the channel descriptor and management API. Public-path tasks (review requests and assignments on unbound PRs) ran their full agent turn inside the poll loop, freezing watch detection, bound delivery, and every other notification for the turn's duration (observed: 64 minutes on a production daemon). They now execute in a background FIFO runner (concurrency 1); claims stay durable and synchronous, the cursor gate and crash recovery keep their semantics, and bound/watch deliveries stay inline. The first-contact body feed gains a thread-level gate so an async body turn cannot overtake and steer-cancel a still-retrying comment turn. * fix(channels): close review gaps in watch coverage and the async task runner Review round on the bound-delivery reliability fix. Three correctness findings, each reproduced against the base: - The first-contact body deferral gate could lose the body permanently: nothing reserves a deferred feed, and the end-of-poll cursor commit can mark the thread read once the blocking task settles. Remove the gate; the runner FIFO already serializes the body behind the comment turn. The absence gate now runs synchronously at dispatch so backpressure deferrals keep the false verdict the lane's body suppression needs. - disconnect retired the runner generation but the winding-down runner's restart hook drained the queued tasks anyway. Guard the restart by generation and resolve idle waiters at disconnect. - Recovery could re-resolve a backpressure-parked task whose public turn was running (the running transition merges patches) and deliver it bound a second time. Skip in-flight ids before any recovery branch mutates a task; crashed tasks without an in-flight entry still recover. Also: gone-session public fallbacks requeue through the serial runner instead of awaiting the turn in the poll lane (head-of-line block and steer-cancel on shared sessions); a same-instance reconnect keeps the orphaned runner's executing id in flight so recovery cannot double-run it; held threads skip redundant comment re-fetches via a poll-start snapshot; escaped runner failures are journaled once per task; the watch budget reserves a terminal slice (20%, at least one) so the drain progresses even when open PRs fill the limit; stale-open bindings tier as terminal once the watch observed closed; the limit constants are exported for the descriptor, the field is registered and translated in the Web Shell editor, and validation enforces the bound-delivery prerequisite; the rate arithmetic is documented at every surface. * fix(channels): drain terminal bindings outside the open watch budget The in-budget terminal reservation excluded up to ceil(limit*0.2) open bindings while a merged/closed backlog drained, permanently swallowing edge events on those bindings — the defect this branch fixes. Open PRs now take the full prWatchLimit; terminal-pending bindings drain in an additional slice outside it (20%, at least one, plus leftover budget), bounded at limit + ceil(limit*0.2) pulls.get per poll during the drain. Pins the verification fixture (30 open + 5 newer terminal at limit 30: oldest open observed, its head change delivered) and keeps the removed in-budget variant as a mutation witness. Rate arithmetic updated on the descriptor, i18n (EN/ZH), user guide, and bilingual design docs. Co-authored-by: Qwen-Coder <[email protected]> --------- Co-authored-by: Qwen-Coder <[email protected]>
This commit is contained in:
Shaojin WenandQwen-Coder authored and GitHub committed
1 parent
c2e77d1473
commit
5b4dcbad19
8 files changed
+2178
-87
No files matched your search
@@ -0,0 +1,303 @@
|
||||
# GitHub Channel: Watch Coverage and Non-Blocking Task Execution
|
||||
|
||||
[English](github-channel-watch-coverage-async-tasks.md) | [简体中文](github-channel-watch-coverage-async-tasks.zh-CN.md)
|
||||
|
||||
Status: implemented and verified (2026-11-06); see
|
||||
`.qwen/e2e-tests/github-channel-watch-coverage-async-tasks.md` for the
|
||||
baseline dry-run, the local-build E2E results, and the mutation witnesses.
|
||||
|
||||
## Problem statement
|
||||
|
||||
The GitHub channel's bound-session delivery (`deliverToBoundSessions` +
|
||||
`prWatchEvents`) is unreliable on a daemon with many PR bindings. Two
|
||||
independent defects, both observed live on a production daemon running
|
||||
0.26.8 with 315 session bindings (242 active watch candidates, ~94 open-PR
|
||||
bindings):
|
||||
|
||||
1. **Watch starvation.** The PR state watch polls at most
|
||||
`MAX_WATCHED_PRS = 30` bound PRs per pass, selected purely by binding
|
||||
`updatedAt` recency (`GithubAdapter.ts`, `runPrStateWatch`). Every other
|
||||
bound PR is never polled, so `head_change` / `ci_failure` /
|
||||
`merge_conflict` / `state_change` events for it are never detected. The
|
||||
daemon logs `PR watch cap exceeded: watching 30/242 bound PRs` on every
|
||||
poll. Two effects compound:
|
||||
- Open PRs with older bindings are crowded out by newer bindings,
|
||||
including bindings of PRs that are already merged.
|
||||
- A PR whose binding records `merged`/`closed` stays in the active set
|
||||
until the watch itself observes the terminal transition. If the cap
|
||||
excludes it, that observation never happens, so it occupies an active
|
||||
slot forever. The active set grows monotonically with dead PRs.
|
||||
2. **Head-of-line blocking.** The poll loop is serial
|
||||
(`PollingChannelBase.runLoop` awaits `pollOnce`). Inside `pollOnce`,
|
||||
both fresh dispatch (`dispatchEnvelope` public branch) and recovery
|
||||
(`recoverInboundTasks`) await `runInboundTask`, whose public path
|
||||
(`handleInbound`) awaits the **entire agent turn** before the reply is
|
||||
published. A single `review_requested` on an unbound PR therefore
|
||||
freezes the whole channel — watch pass, bound deliveries, and every
|
||||
other notification — for the full turn duration (observed: 64 minutes).
|
||||
Bound delivery itself is fast (a prompt POST that does not wait for the
|
||||
turn); only the public path blocks.
|
||||
|
||||
## Goals
|
||||
|
||||
- Every open bound PR is watched whenever the configured watch budget
|
||||
allows it; terminal-pending PRs drain out of the active set instead of
|
||||
accumulating.
|
||||
- The watch budget is operator-configurable within documented rate-limit
|
||||
trade-offs.
|
||||
- A long public-path agent turn never blocks the poll loop: watch
|
||||
detection, bound delivery, and intake of new notifications continue
|
||||
while the turn runs.
|
||||
- No change to delivery guarantees: at-least-once, durable claims before
|
||||
cursor advance, edge-triggered watch semantics, crash recovery, and
|
||||
backpressure budgets all behave exactly as today.
|
||||
|
||||
## Non-goals
|
||||
|
||||
- Parallel public turns (concurrency stays 1; ordering is preserved).
|
||||
- Changing the notification-cursor contract: recoverable non-backpressure
|
||||
tasks still hold mark-as-read (`hasCursorBlockingTasks`), as today.
|
||||
- Raising GitHub rate limits or adding conditional-request caching.
|
||||
- Any change to bound-session delivery itself (fan-out, roles, templates).
|
||||
|
||||
## Proposed changes
|
||||
|
||||
### 1. Watch selection and budget (`packages/channels/github`)
|
||||
|
||||
- New channel config field `prWatchLimit`:
|
||||
- Descriptor: `kind: 'number'`, `exclusiveMinimum: 0`, `maximum` and
|
||||
`default` built from the adapter's exported `MAX_PR_WATCH_LIMIT` (500)
|
||||
and `DEFAULT_PR_WATCH_LIMIT` (30) so the form bound and the effective
|
||||
bound cannot drift; the description states the rate arithmetic (~60
|
||||
requests/hour per watched PR at the 60 s default; a 5000/hour PAT
|
||||
budget admits ~80 watched PRs, ~69 open while a terminal backlog
|
||||
drains at the 1.2x fan-out; over budget, every watch pass aborts).
|
||||
- Adapter: a `prWatchLimit` getter clamps like the existing
|
||||
`pollInterval` getter (invalid/unset → 30; fractional → rounded up;
|
||||
above 500 → clamped to 500). Hand-edited settings reach the getter
|
||||
unchecked by the descriptor, so the clamp is load-bearing and tested.
|
||||
- Validation: `validateBoundSessionConfig` rejects `prWatchLimit`
|
||||
without `deliverToBoundSessions: true`, with the sibling wording.
|
||||
- Web Shell: the field is registered in the editor's GitHub label map
|
||||
and bound-session panel, with both locale catalogues translated.
|
||||
- `runPrStateWatch` partitions the active set:
|
||||
- **open**: binding state is neither `merged` nor `closed`, and the
|
||||
watch entry has not observed `prState === 'closed'` (a stale-open
|
||||
registry must not let an observed-closed PR hold an open-tier slot);
|
||||
- **terminal-pending**: everything else that still needs its terminal
|
||||
observation.
|
||||
Open PRs take the FULL budget (by `updatedAt` desc). Terminal-pending
|
||||
PRs drain in an ADDITIONAL slice outside it —
|
||||
`min(terminalCount, max(1, ceil(limit * 0.2)))`, plus any leftover
|
||||
open budget. Without the slice, `limit == openCount` would starve the
|
||||
drain forever and the active set would grow with every merge. An
|
||||
IN-budget reservation was tried and removed: it excludes up to
|
||||
`ceil(limit * 0.2)` open bindings while the backlog drains, and an
|
||||
edge event on an excluded binding inside that window is swallowed
|
||||
permanently — the very defect this design fixes. The additional
|
||||
slice costs at most `ceil(limit * 0.2)` extra `pulls.get` per poll
|
||||
during the drain; each drained binding leaves the active set for
|
||||
good, so the fan-out returns to the open budget once the backlog
|
||||
clears.
|
||||
- The cap warning keeps the `PR watch cap exceeded` prefix (pinned by
|
||||
tests) and fires whenever the selection cannot cover the whole active
|
||||
set: it reports `watched/active` counts (watched may exceed the open
|
||||
limit while the terminal slice drains), the configured open limit, and
|
||||
the excluded keys (open and terminal-pending alike).
|
||||
- Terminal observation stays API-authoritative: no `state_change` is ever
|
||||
synthesized from binding registry data. The registry and the watch
|
||||
entry only decide _priority_, never event content. A genuinely reopened
|
||||
PR gets one leftover-budget observation and returns to the open tier.
|
||||
|
||||
### 2. Background public-task runner (`packages/channels/github`)
|
||||
|
||||
- A per-channel FIFO runner (concurrency 1) owns public-path
|
||||
`runInboundTask` execution. It follows the existing
|
||||
`retryPendingFinalDeliveries` background-loop pattern: started on
|
||||
demand, retired by generation on `disconnect`/`connect`.
|
||||
- `dispatchEnvelope` public branch: claim the task durably (unchanged),
|
||||
run the absence gate synchronously, then enqueue and return `true`.
|
||||
The absence gate (incomplete binding snapshot cannot prove a PR is
|
||||
unbound) must run at dispatch so a deferred event keeps the `false`
|
||||
verdict the comment lane's first-contact body suppression depends on;
|
||||
`runInboundTask` re-checks it for tasks recovery enqueues.
|
||||
- Gone-session public fallbacks (bound target vanished, absence
|
||||
authoritative) clear the delivery durably and requeue through the same
|
||||
runner (`requeued` verdict) instead of awaiting `handleInbound` in the
|
||||
poll lane: they are full public turns, and running them inline both
|
||||
head-of-line-blocks the poll and can steer-cancel another public turn
|
||||
on a shared channel session.
|
||||
- Bound fan-out branch stays inline: delivery is a fast prompt POST whose
|
||||
verdict (delivered/deferred/retired) the dispatch return value and the
|
||||
cursor gate already depend on.
|
||||
- `recoverInboundTasks` skips tasks in the runner's in-flight set BEFORE
|
||||
any branch mutates them (the retarget branch would otherwise re-resolve
|
||||
a backpressure-parked task whose public turn is running — the running
|
||||
transition merges patches, so `backpressure: true` survives — and
|
||||
deliver it bound a second time). A running task with no in-flight entry
|
||||
is a genuine crash and is still recovered and charged. Bound tasks run
|
||||
inline as today; other recoverable tasks are enqueued.
|
||||
- An in-flight set (queued + running task ids) prevents double execution
|
||||
when a recovery pass re-scans a queued task. On a same-instance
|
||||
reconnect, the id the orphaned runner is still executing stays in the
|
||||
set (the orphan's `finally` removes it); queued-but-unstarted ids are
|
||||
dropped and re-enqueued by recovery from the durable store.
|
||||
- The runner isolates failures: `runInboundTask` already transitions
|
||||
errors to `failed`/backpressure/retirement; an unexpected throw is
|
||||
logged AND journaled as an inbound `failed` row once per task id (the
|
||||
poll loop's per-thread catch no longer sees it), then the runner
|
||||
continues with the next task.
|
||||
- `disconnect` retires the runner generation and resolves idle waiters.
|
||||
The winding-down runner's restart hook only fires for the CURRENT
|
||||
generation, so a retired queue is never drained after disconnect; the
|
||||
tasks stay durable for the next connect's recovery. An in-flight turn
|
||||
keeps today's semantics: the worker exits, the daemon session survives.
|
||||
- While a thread's in-flight public work held the cursor at poll start,
|
||||
the comment/aggregate lanes skip re-fetching its comments and the
|
||||
direct lane skips re-processing (snapshot, not live: a task claimed
|
||||
during this poll must not suppress the same thread's later lanes, and
|
||||
a crashed task that is not in flight must stay re-dispatchable). The
|
||||
first-contact body feed is NOT gated: it claims once and the runner
|
||||
serializes it behind any running comment turn. A deferral gate was
|
||||
tried and removed — nothing reserves a deferred body, and the
|
||||
end-of-poll cursor commit can mark the thread read once the blocking
|
||||
task settles, after which first contact (`!lastReadAt`) is unreachable
|
||||
and the body is lost permanently.
|
||||
|
||||
### Invariants preserved (why this is safe)
|
||||
|
||||
- Claims are durable before the dispatch return, so cursor advance never
|
||||
outruns consumption — unchanged.
|
||||
- `hasCursorBlockingTasks` still freezes mark-as-read while a public task
|
||||
is recoverable; polls during a long turn re-list the same window and
|
||||
dedupe skips already-claimed events, exactly the state crash recovery
|
||||
already produces today.
|
||||
- Public tasks execute serially in claim order (FIFO, concurrency 1), so
|
||||
per-thread reply ordering is unchanged.
|
||||
- Watch pending/fired reconciliation, attempt budgets, backpressure
|
||||
horizons, and the publication audit are untouched.
|
||||
|
||||
## Files affected
|
||||
|
||||
- `packages/channels/github/src/GithubAdapter.ts` — selection, limit
|
||||
getter, runner, dispatch/recovery hand-off, absence gate extraction,
|
||||
gone-session requeue, held-thread fetch skip, disconnect/connect.
|
||||
- `packages/channels/github/src/index.ts` — `prWatchLimit` descriptor
|
||||
field built from the exported constants.
|
||||
- `packages/channels/github/src/GithubAdapter.test.ts` — new coverage;
|
||||
the `pollOnce` test helpers additionally await runner idle so existing
|
||||
public-path assertions keep their meaning.
|
||||
- `packages/web-shell/client/components/channels/ChannelEditorDialog.tsx`
|
||||
and `packages/web-shell/client/i18n.tsx` — field registration and both
|
||||
locale catalogues.
|
||||
- Docs: the GitHub channel guide's bound-session section gains the
|
||||
`prWatchLimit` knob and the non-blocking execution note.
|
||||
|
||||
## Design decisions and rationale
|
||||
|
||||
1. **Open-first priority with an additional terminal slice.** A larger
|
||||
hard cap only postpones starvation and spends rate budget on dead
|
||||
PRs. Priority fixes the ordering defect; the additional slice (20% of
|
||||
the limit, at least one, OUTSIDE the open budget) guarantees the drain
|
||||
progresses even when open PRs fill the limit without excluding a
|
||||
single open binding, and disappears once the backlog clears. The
|
||||
in-budget variant was measured to re-starve the oldest open bindings
|
||||
for the whole drain window (verification fixture: 30 open + 5
|
||||
terminal at limit 30 must still deliver the oldest open PR's head
|
||||
change), so the slice is additive and the worst-case fan-out is
|
||||
bounded at the limit plus `ceil(limit * 0.2)` (1.2x at the default
|
||||
30).
|
||||
2. **Configurable limit, default unchanged (30).** Each watched PR costs
|
||||
roughly one `pulls.get` per poll plus conditional Checks fetches —
|
||||
~60 requests/hour at the 60 s default, so a 5000 req/h PAT budget
|
||||
admits ~80 watched PRs (~69 open while a terminal backlog drains at
|
||||
the 1.2x fan-out). Operators with more open bindings raise
|
||||
`prWatchLimit` (and `pollInterval`) deliberately. The watch pass
|
||||
already fails fast on rate limits, so an over-budget configuration
|
||||
degrades instead of hammering the API; the arithmetic is stated in
|
||||
the descriptor and the guide.
|
||||
3. **Concurrency 1 runner, not per-session parallelism.** The defect is
|
||||
liveness, not throughput. Serial execution preserves reply ordering
|
||||
and avoids bridge-level concurrent-prompt hazards documented in
|
||||
`ChannelBase` (steer/chain-and-wait). Every transition into public
|
||||
execution goes through the runner, including gone-session fallbacks.
|
||||
4. **No binding-state shortcut for terminal observation.** Synthesizing
|
||||
`state_change` from registry data would break the edge-triggered
|
||||
contract (the registry can lag or lead the API). One real observation
|
||||
per terminal PR is cheap and drains the set permanently. The watch
|
||||
entry may tier a stale-open binding as terminal, never fabricate an
|
||||
event.
|
||||
5. **No first-contact body deferral gate.** A deferred body is not
|
||||
reserved by anything: once the blocking task settles, the end-of-poll
|
||||
cursor commit can mark the thread read, and first contact
|
||||
(`!lastReadAt`) becomes unreachable — permanent loss. The runner's
|
||||
FIFO already serializes the body turn behind any running comment
|
||||
turn, which is the property the gate was meant to protect. The
|
||||
synchronous absence gate at dispatch preserves the lane's
|
||||
false-verdict body suppression for backpressure-deferred comments.
|
||||
|
||||
## Constraints and risks
|
||||
|
||||
- **Test migration surface.** Existing tests assert public outcomes
|
||||
immediately after `await pollOnce()`. Centralizing the runner-idle wait
|
||||
in the test helpers keeps most assertions intact; tests that
|
||||
intentionally observe mid-flight state must opt out explicitly.
|
||||
- **Frozen-cursor polling cost.** While a long public turn runs, the
|
||||
window re-lists every poll; the held-thread snapshot skips the
|
||||
redundant per-thread comment fetches and direct-lane reprocessing, so
|
||||
the request volume no longer scales with turn duration. Dispatch
|
||||
dedupe is unchanged.
|
||||
- **Runner shutdown.** A turn in flight at `disconnect` is not cancelled
|
||||
by this change (same as today); recovery replays it at next start with
|
||||
the existing attempt budget. The retired generation never restarts
|
||||
the queue, and idle waiters resolve at disconnect.
|
||||
|
||||
## Validation plan
|
||||
|
||||
- Unit tests (collocated): open-first selection with the incident shape;
|
||||
the additional terminal slice at `limit == openCount` (every open
|
||||
binding keeps its observation while the backlog drains, and the oldest
|
||||
open PR's head change inside the drain window is delivered);
|
||||
stale-open tiering; limit clamping (over-ceiling, zero, non-finite) and
|
||||
descriptor acceptance; validation prerequisite; warning shape; runner
|
||||
hand-off,
|
||||
FIFO order, in-flight dedupe via a mid-turn recovery pass, recovery
|
||||
retarget skip, gone-session requeue, same-instance reconnect,
|
||||
disconnect stop + idle waiters, held-thread fetch skip, failure
|
||||
journaling; cursor gate unchanged while a public task runs.
|
||||
- E2E (`.qwen/e2e-tests/github-channel-watch-coverage-async-tasks.md`):
|
||||
real daemon + local GitHub-compatible stub + local model endpoint, per
|
||||
the channel real-scenario harness. Baseline dry-run on the installed
|
||||
release must show both defects; the local build must pass all groups.
|
||||
- Mutation witnesses: revert open-first ordering, revert the enqueue
|
||||
hand-off, remove the recovery in-flight skip, remove the additional
|
||||
terminal slice, shrink the open budget by the slice (the removed
|
||||
in-budget reservation), remove the disconnect generation guard — each
|
||||
must turn a dedicated test red.
|
||||
|
||||
## Acceptance criteria
|
||||
|
||||
- With >30 open bound PRs and default limit, the most recent open PRs
|
||||
are watched within the FULL open budget and terminal-pending PRs drain
|
||||
through the additional slice outside it; with `prWatchLimit` raised to
|
||||
the open count, every open bound PR receives `head_change` delivery.
|
||||
- Terminal-pending PRs leave the active set after one observed poll,
|
||||
even when open PRs fill the limit, without excluding any open binding
|
||||
(fixture: 30 open + 5 newer terminal at limit 30 — the oldest open PR
|
||||
is observed and its head change is delivered); the active count in the
|
||||
cap warning shrinks accordingly.
|
||||
- While a public review turn runs for minutes, a bound PR's head change
|
||||
is still delivered within one poll interval, new notifications are
|
||||
still claimed, and the held thread's comments are not re-fetched every
|
||||
poll.
|
||||
- A first-contact body is never lost to a cursor commit; a deferred
|
||||
(backpressure) comment still suppresses the body for that poll.
|
||||
- All existing delivery/recovery/backpressure tests pass (helpers await
|
||||
runner idle); lint, typecheck, and focused unit suites green.
|
||||
|
||||
## Open questions
|
||||
|
||||
- None blocking. A future enhancement could batch the PR state queries
|
||||
to remove the per-PR fan-out entirely (raised in triage as the
|
||||
preferred long-term shape); it is separate work with its own
|
||||
rate-limit questions and is deliberately deferred.
|
||||
@@ -0,0 +1,239 @@
|
||||
# GitHub 渠道：watch 覆盖与非阻塞任务执行
|
||||
|
||||
[English](github-channel-watch-coverage-async-tasks.md) | [简体中文](github-channel-watch-coverage-async-tasks.zh-CN.md)
|
||||
|
||||
状态：已实现并验证（2026-11-06 提案，2026-11-07 评审修复轮）；基线干跑、
|
||||
本地构建 E2E 结果与变异见证见
|
||||
`.qwen/e2e-tests/github-channel-watch-coverage-async-tasks.md`。
|
||||
|
||||
## 问题陈述
|
||||
|
||||
在绑定数量较多的 daemon 上，GitHub 渠道的绑定会话投递
|
||||
（`deliverToBoundSessions` + `prWatchEvents`）不可靠。两个相互独立的缺陷，
|
||||
均在一台运行 0.26.8 的生产 daemon 上实盘观测到（315 条会话绑定、242 个
|
||||
活跃 watch 候选、约 94 条 open PR 绑定）：
|
||||
|
||||
1. **watch 饥饿。** PR 状态 watch 每轮至多轮询
|
||||
`MAX_WATCHED_PRS = 30` 个绑定 PR，且纯按绑定 `updatedAt` 新旧选择
|
||||
（`GithubAdapter.ts` 的 `runPrStateWatch`）。其余绑定 PR 永不被轮询，
|
||||
它们的 `head_change` / `ci_failure` / `merge_conflict` / `state_change`
|
||||
事件永不被检测。daemon 每轮日志都警告
|
||||
`PR watch cap exceeded: watching 30/242 bound PRs`。两个效应叠加：
|
||||
- 绑定较早的 open PR 被更新的绑定挤出，其中不少是早已合入的 PR 的绑定。
|
||||
- 绑定里记录为 `merged`/`closed` 的 PR 会一直留在活跃集里，直到 watch
|
||||
亲自观测到终态迁移。若上限把它排除在外，这次观测永远不会发生，
|
||||
于是它永久占据一个活跃名额。活跃集随死 PR 单调增长。
|
||||
2. **队头阻塞。** 轮询循环是串行的
|
||||
（`PollingChannelBase.runLoop` await `pollOnce`）。在 `pollOnce` 内部，
|
||||
新派发（`dispatchEnvelope` 公共分支）与恢复（`recoverInboundTasks`）
|
||||
都 await `runInboundTask`，其公共路径（`handleInbound`）会等待
|
||||
**整个 agent turn** 跑完才发布回复。于是一个落在未绑定 PR 上的
|
||||
`review_requested` 就把整个渠道——watch、绑定投递、所有其他通知——
|
||||
冻结整个 turn 时长（实测 64 分钟）。绑定投递本身很快（一次不等 turn
|
||||
的 prompt POST）；只有公共路径阻塞。
|
||||
|
||||
## 目标
|
||||
|
||||
- 只要配置的 watch 预算允许，每个 open 绑定 PR 都被 watch；终态待观测
|
||||
PR 从活跃集中排出，而不是不断累积。
|
||||
- watch 预算可由运维配置，并写明限流权衡。
|
||||
- 冗长的公共路径 agent turn 永不阻塞轮询循环：turn 运行期间，watch
|
||||
检测、绑定投递、新通知接收照常进行。
|
||||
- 不改变投递保证：至少一次、游标推进前先持久化认领、边沿触发的 watch
|
||||
语义、崩溃恢复、背压预算，全部与今天一致。
|
||||
|
||||
## 非目标
|
||||
|
||||
- 公共 turn 并行化（并发度保持 1，顺序不变）。
|
||||
- 改变通知游标契约：可恢复的非背压任务仍然持有 mark-as-read
|
||||
（`hasCursorBlockingTasks`），与今天一致。
|
||||
- 提升 GitHub 限流额度或引入条件请求缓存。
|
||||
- 改动绑定会话投递本身（fan-out、角色、模板）。
|
||||
|
||||
## 提议的改动
|
||||
|
||||
### 1. watch 选择与预算（`packages/channels/github`）
|
||||
|
||||
- 新渠道配置字段 `prWatchLimit`：
|
||||
- 描述符：`kind: 'number'`、`exclusiveMinimum: 0`，`maximum` 与
|
||||
`default` 取自适配器导出的 `MAX_PR_WATCH_LIMIT`（500）与
|
||||
`DEFAULT_PR_WATCH_LIMIT`（30），表单边界与生效边界不会漂移；
|
||||
描述写明限流算术（默认 60 秒间隔下每个被 watch 的 PR 约 60 次
|
||||
请求/小时；5000 次/小时的 PAT 预算约容纳 80 个被 watch 的 PR，终态
|
||||
积压按 1.2 倍扇出排出时约 69 个 open；超预算会让每轮 watch 都中止）。
|
||||
- 适配器：`prWatchLimit` getter 沿用现有 `pollInterval` getter 的钳制
|
||||
方式（非法/缺省 → 30；小数 → 向上取整；超过 500 → 钳到 500）。
|
||||
手工编辑的设置文件不经过描述符校验直达 getter，因此钳制是承重
|
||||
设计并有测试。
|
||||
- 校验：`validateBoundSessionConfig` 在缺少
|
||||
`deliverToBoundSessions: true` 时拒绝 `prWatchLimit`，措辞与兄弟
|
||||
字段一致。
|
||||
- Web Shell：该字段登记进编辑器的 GitHub 标签映射与绑定会话面板，
|
||||
两份语言词条均已翻译。
|
||||
- `runPrStateWatch` 对活跃集分区：
|
||||
- **open**：绑定状态既非 `merged` 也非 `closed`，且 watch 条目尚未
|
||||
观测到 `prState === 'closed'`（陈旧的 open 注册表不能让已观测关闭
|
||||
的 PR 占据 open 档名额）；
|
||||
- **终态待观测**：其余仍需终态观测的条目。
|
||||
open PR 独占**全部**预算（按 `updatedAt` 降序）。终态待观测 PR 在
|
||||
预算之外的一块**额外**名额中排出——
|
||||
`min(终态数, max(1, ceil(limit * 0.2)))`，外加 open 预算的任何剩余。
|
||||
没有这块名额时，`limit == open 数` 会让排出永远饿死，活跃集随每次
|
||||
合入增长。曾尝试**预算内**预留并移除：它会在积压排出期间排除至多
|
||||
`ceil(limit * 0.2)` 个 open 绑定，窗口内被排除绑定上的边沿事件会被
|
||||
永久吞掉——正是本设计要修的缺陷。额外名额在排出期间每轮至多多花
|
||||
`ceil(limit * 0.2)` 次 `pulls.get`；每个排出的绑定永久离开活跃集，
|
||||
积压清空后扇出即回到 open 预算。
|
||||
- 上限警告保留 `PR watch cap exceeded` 前缀（有测试 pin），只要选择覆盖
|
||||
不了整个活跃集就触发：报告 `已watch/活跃` 计数（终态名额排出期间
|
||||
已 watch 数可能超过 open 上限）、配置的 open 上限、以及被排除的键
|
||||
（open 与终态待观测一视同仁）。
|
||||
- 终态观测仍以 API 为权威：绝不从绑定注册表数据合成 `state_change`。
|
||||
注册表与 watch 条目只决定**优先级**，从不决定事件内容。真正被重新
|
||||
打开的 PR 会得到一次剩余预算观测并回到 open 档。
|
||||
|
||||
### 2. 后台公共任务运行器（`packages/channels/github`）
|
||||
|
||||
- 每渠道一个 FIFO 运行器（并发度 1）承接公共路径的 `runInboundTask`
|
||||
执行。沿用现有 `retryPendingFinalDeliveries` 后台循环模式：按需启动，
|
||||
`disconnect`/`connect` 以代次退役。
|
||||
- `dispatchEnvelope` 公共分支：持久化认领任务（不变），同步执行缺席
|
||||
闸门，然后入队并返回 `true`。缺席闸门（不完整的绑定快照无法证明 PR
|
||||
无绑定）必须在派发时执行，被延迟的事件才能保持评论车道 first-contact
|
||||
body 抑制所依赖的 `false` 裁决；`runInboundTask` 对恢复入队的任务
|
||||
再检查一次。
|
||||
- 绑定会话消失的公共回退（目标消失且缺席已确定为权威）持久化清空
|
||||
delivery 后经同一运行器重新入队（`requeued` 裁决），而不是在轮询车道
|
||||
内 await `handleInbound`：它们是完整的公共 turn，内联执行既会队头
|
||||
阻塞轮询，也可能在共享渠道会话上 steer-cancel 另一个公共 turn。
|
||||
- 绑定 fan-out 分支保持内联：投递是一次快速 prompt POST，其裁决
|
||||
（delivered/deferred/retired）是派发返回值与游标闸门已经依赖的输入。
|
||||
- `recoverInboundTasks` 在任何分支改动任务**之前**跳过运行器在途集合
|
||||
中的任务（否则重定向分支会重新解析一个公共 turn 正在运行的背压挂起
|
||||
任务——running 转换是合并语义，`backpressure: true` 会保留——并把
|
||||
它第二次绑定投递）。没有在途条目的 running 任务是真正的崩溃，仍然
|
||||
恢复并计次。绑定任务照旧内联执行；其余可恢复任务入队。
|
||||
- 在途集合（已入队 + 运行中的任务 id）防止恢复 pass 重扫已入队任务时
|
||||
重复执行。同实例重连时，孤立运行器正在执行的任务 id 保留在集合中
|
||||
（孤立运行器的 `finally` 会移除它）；已入队但未启动的 id 被丢弃，
|
||||
由恢复从持久存储重新入队。
|
||||
- 运行器隔离失败：`runInboundTask` 已把错误迁移为
|
||||
`failed`/背压/退役；意外抛出会被记日志**并**按任务 id 各记一次
|
||||
inbound `failed` 事件行（轮询循环的按线程 catch 不再看到它），然后
|
||||
运行器继续下一个任务。
|
||||
- `disconnect` 退役运行器代次并释放空闲等待者。收尾运行器的重启钩子
|
||||
只对**当前**代次生效，因此 disconnect 之后绝不会继续排空退役队列；
|
||||
任务保持持久化，等下次 connect 的恢复。在途 turn 保持今天的语义：
|
||||
worker 退出、daemon 会话存活。
|
||||
- 当某线程的在途公共工作在轮询开始时已持有游标，评论/聚合车道跳过对
|
||||
它的评论重抓，直接车道跳过重复处理（快照而非实时：本轮认领的任务
|
||||
不能抑制同线程后续车道；不在途的崩溃任务必须仍可重新派发）。
|
||||
first-contact body 投递**不加**闸门：它只认领一次，运行器会把它串行
|
||||
排在任何运行中的评论 turn 之后。曾尝试延迟闸门并移除——没有任何
|
||||
东西为被延迟的 body 占位，一旦阻塞任务落定，轮询末尾的游标提交可能
|
||||
把线程标记已读，此后 first contact（`!lastReadAt`）不可达，body
|
||||
永久丢失。
|
||||
|
||||
### 保留的不变量（为什么安全）
|
||||
|
||||
- 认领在派发返回前已持久化，游标推进永不超越消费——不变。
|
||||
- 公共任务可恢复期间，`hasCursorBlockingTasks` 仍冻结 mark-as-read；
|
||||
长 turn 期间的每轮轮询重列同一窗口，去重跳过已认领事件——这正是今天
|
||||
崩溃恢复已经产生的状态。
|
||||
- 公共任务按认领顺序串行执行（FIFO、并发 1），每线程回复顺序不变。
|
||||
- watch pending/fired 对账、尝试预算、背压地平线、发布审计均不改动。
|
||||
|
||||
## 受影响文件
|
||||
|
||||
- `packages/channels/github/src/GithubAdapter.ts` —— 选择逻辑、上限
|
||||
getter、运行器、派发/恢复交接、缺席闸门提取、gone-session 重新入队、
|
||||
持有线程免重抓、disconnect/connect。
|
||||
- `packages/channels/github/src/index.ts` —— 由导出常量构建的
|
||||
`prWatchLimit` 描述符字段。
|
||||
- `packages/channels/github/src/GithubAdapter.test.ts` —— 新增覆盖；
|
||||
`pollOnce` 测试助手额外等待运行器空闲，使既有公共路径断言含义不变。
|
||||
- `packages/web-shell/client/components/channels/ChannelEditorDialog.tsx`
|
||||
与 `packages/web-shell/client/i18n.tsx` —— 字段登记与两份语言词条。
|
||||
- 文档：GitHub 渠道指南的绑定会话小节补充 `prWatchLimit` 旋钮与
|
||||
非阻塞执行说明。
|
||||
|
||||
## 设计决策与理由
|
||||
|
||||
1. **open 优先 + 额外终态名额。** 更大的硬上限只是推迟饥饿，还把限流
|
||||
预算花在死 PR 上。优先级修复排序缺陷；额外名额（上限的 20%，至少
|
||||
一个，在 open 预算**之外**）保证 open 占满上限时排出仍能推进，且不
|
||||
排除任何一个 open 绑定，积压清空后即消失。预算内变体经实测会在整个
|
||||
排出窗口内重新饿死最旧的 open 绑定（验证夹具：limit 30 下 30 open +
|
||||
5 终态，最旧 open PR 的 head 变化仍必须投递），因此名额是加法的，
|
||||
最坏扇出界定为上限加 `ceil(limit * 0.2)`（默认 30 时为 1.2 倍）。
|
||||
2. **上限可配置，默认不变（30）。** 每个被 watch 的 PR 每轮约花费一次
|
||||
`pulls.get` 加条件性 Checks 拉取——默认 60 秒间隔下约 60 次请求/小时，
|
||||
5000 次/小时的 PAT 预算约容纳 80 个被 watch 的 PR（终态积压按 1.2 倍
|
||||
扇出排出时约 69 个 open）。绑定更多的运维有意识地调高
|
||||
`prWatchLimit`（与 `pollInterval`）。watch pass 本就在限流时快速失败，
|
||||
超预算配置会降级而不是锤爆 API；算术写明在描述符与指南中。
|
||||
3. **并发 1 运行器，而非按会话并行。** 缺陷是活性，不是吞吐。串行执行
|
||||
保住回复顺序，避开 `ChannelBase` 里记载的桥级并发 prompt 风险
|
||||
（steer/chain-and-wait）。所有进入公共执行的转换都经过运行器，包括
|
||||
gone-session 回退。
|
||||
4. **终态观测不走绑定状态捷径。** 从注册表数据合成 `state_change` 会
|
||||
破坏边沿触发契约（注册表可能滞后或超前于 API）。每个终态 PR 一次
|
||||
真实观测，成本低且永久排出活跃集。watch 条目可以把陈旧 open 绑定
|
||||
分为终态档，但绝不凭空生成事件。
|
||||
5. **不加 first-contact body 延迟闸门。** 被延迟的 body 没有任何占位：
|
||||
阻塞任务一旦落定，轮询末尾的游标提交可能把线程标记已读，first
|
||||
contact（`!lastReadAt`）从此不可达——永久丢失。运行器 FIFO 本就把
|
||||
body turn 串行排在运行中的评论 turn 之后，这正是闸门想保护的性质。
|
||||
派发时的同步缺席闸门保留了车道对背压延迟评论的 false 裁决 body
|
||||
抑制。
|
||||
|
||||
## 约束与风险
|
||||
|
||||
- **测试迁移面。** 既有测试在 `await pollOnce()` 后立即断言公共结果。
|
||||
把运行器空闲等待集中到测试助手可保住大多数断言；刻意观察在途状态的
|
||||
测试必须显式退出该等待。
|
||||
- **冻结游标期的轮询成本。** 长公共 turn 运行期间窗口每轮重列；持有
|
||||
线程快照跳过冗余的按线程评论重抓与直接车道重复处理，请求量不再随
|
||||
turn 时长线性增长。派发去重不变。
|
||||
- **运行器关停。** `disconnect` 时不取消在途 turn（与今天一致），恢复
|
||||
机制在下次启动按既有尝试预算重放。退役代次绝不重启队列，空闲等待者
|
||||
在 disconnect 时释放。
|
||||
|
||||
## 验证计划
|
||||
|
||||
- 单测（同目录）：事故形态的 open 优先选择；`limit == open 数` 时的额外
|
||||
终态名额（积压排出期间每个 open 绑定保住观测，窗口内最旧 open PR 的
|
||||
head 变化仍投递）；陈旧 open 分档；上限钳制（超上限、零、非有限）与
|
||||
描述符接受；校验前置依赖；警告形态；运行器交接、FIFO 顺序、turn 中途
|
||||
恢复 pass 钉住在途去重、恢复重定向跳过、gone-session 重新入队、同实例
|
||||
重连、disconnect 停止 + 空闲等待者、持有线程免重抓、失败记账；公共
|
||||
任务运行期间游标闸门不变。
|
||||
- E2E（`.qwen/e2e-tests/github-channel-watch-coverage-async-tasks.md`）：
|
||||
按渠道真场景 harness，用真实 daemon + 本地 GitHub 兼容桩 + 本地模型
|
||||
端点。基线干跑（已安装发行版）必须复现两个缺陷；本地构建必须通过
|
||||
全部组。
|
||||
- 变异见证：回退 open 优先排序、回退入队交接、移除恢复在途跳过、移除
|
||||
额外终态名额、把 open 预算缩减该名额（已移除的预算内预留）、移除
|
||||
disconnect 代次守卫——每一项都必须让对应专属测试变红。
|
||||
|
||||
## 验收标准
|
||||
|
||||
- open 绑定 PR 超过 30 且用默认上限时，最新的 open PR 在**完整** open
|
||||
预算内被 watch，终态待观测 PR 经预算之外的额外名额排出；把
|
||||
`prWatchLimit` 抬到 open 数量后，每个 open 绑定 PR 都能收到
|
||||
`head_change` 投递。
|
||||
- 终态待观测 PR 在一次观测轮询后离开活跃集，即使 open PR 占满上限，且
|
||||
不排除任何 open 绑定（夹具：limit 30 下 30 open + 5 个更新的终态——
|
||||
最旧 open PR 被观测、其 head 变化被投递）；上限警告里的活跃数相应
|
||||
下降。
|
||||
- 公共评审 turn 运行数分钟期间，绑定 PR 的 head 变化仍在一个轮询间隔
|
||||
内投递，新通知仍被认领，持有线程的评论不再每轮重抓。
|
||||
- first-contact body 绝不因游标提交而丢失；被延迟（背压）的评论仍在
|
||||
该轮抑制 body。
|
||||
- 既有投递/恢复/背压测试全部通过（助手等待运行器空闲）；lint、
|
||||
typecheck、聚焦单测全绿。
|
||||
|
||||
## 开放问题
|
||||
|
||||
- 无阻塞项。未来可用批量 PR 状态查询彻底去掉按 PR 扇出（triage 提出的
|
||||
长期优选形态）；那是独立工作，有自己的限流问题，刻意推迟。
|
||||
@@ -80,20 +80,21 @@ Local `gh` authentication requires an HTTPS `baseUrl` so the daemon host credent
|
||||
|
||||
## Configuration Options
|
||||
|
||||
| Option | Default | Description |
|
||||
| ------------------------- | ------------------------ | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ |
|
||||
| `token` | unset | Optional classic PAT with `notifications` scope; overrides local `gh` authentication |
|
||||
| `useLocalGh` | `false` | Explicitly reuse the daemon host's account-wide GitHub CLI authentication |
|
||||
| `pollInterval` | `60000` | Poll interval in ms |
|
||||
| `baseUrl` | `https://api.github.com` | API base URL (for GHE) |
|
||||
| `groupPolicy` | `"disabled"` | Must be `"open"`, `"allowlist"` with the repo (`owner/repo`) listed in `groups`, or `"pairing"` with the repo approved for notifications to flow |
|
||||
| `groups.*.requireMention` | `true` | Require @mentions for ordinary comments; directed notification reasons still run |
|
||||
| `reasonFilter` | unset | Optional allowlist of GitHub notification reasons to process |
|
||||
| `repoFilterMode` | `"off"` | Repository event filter: `"allowlist"` processes only the repositories listed in `repoFilter`, `"blocklist"` processes all except them, `"off"` processes everything. Applies to notification intake only |
|
||||
| `repoFilter` | unset | `owner/repo` repository names used by `repoFilterMode`; required while the mode is `"allowlist"` or `"blocklist"` |
|
||||
| `deliverToBoundSessions` | `false` | Deliver events for PRs bound to a live session into that session as a prompt instead of a channel-owned session (daemon-managed channels only) |
|
||||
| `prWatchEvents` | unset | Optional state transitions to watch for bound PRs: `ci_failure`, `merge_conflict`, `head_change`, `state_change`. Requires `deliverToBoundSessions` |
|
||||
| `boundSessionActions` | unset | Optional operator instruction templates for bound-session delivery, keyed by event class with optional role prefixes (`review:`, `verify:`, `fix:`). Enables role-aware fan-out. Requires `deliverToBoundSessions` |
|
||||
| Option | Default | Description |
|
||||
| ------------------------- | ------------------------ | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
|
||||
| `token` | unset | Optional classic PAT with `notifications` scope; overrides local `gh` authentication |
|
||||
| `useLocalGh` | `false` | Explicitly reuse the daemon host's account-wide GitHub CLI authentication |
|
||||
| `pollInterval` | `60000` | Poll interval in ms |
|
||||
| `baseUrl` | `https://api.github.com` | API base URL (for GHE) |
|
||||
| `groupPolicy` | `"disabled"` | Must be `"open"`, `"allowlist"` with the repo (`owner/repo`) listed in `groups`, or `"pairing"` with the repo approved for notifications to flow |
|
||||
| `groups.*.requireMention` | `true` | Require @mentions for ordinary comments; directed notification reasons still run |
|
||||
| `reasonFilter` | unset | Optional allowlist of GitHub notification reasons to process |
|
||||
| `repoFilterMode` | `"off"` | Repository event filter: `"allowlist"` processes only the repositories listed in `repoFilter`, `"blocklist"` processes all except them, `"off"` processes everything. Applies to notification intake only |
|
||||
| `repoFilter` | unset | `owner/repo` repository names used by `repoFilterMode`; required while the mode is `"allowlist"` or `"blocklist"` |
|
||||
| `deliverToBoundSessions` | `false` | Deliver events for PRs bound to a live session into that session as a prompt instead of a channel-owned session (daemon-managed channels only) |
|
||||
| `prWatchEvents` | unset | Optional state transitions to watch for bound PRs: `ci_failure`, `merge_conflict`, `head_change`, `state_change`. Requires `deliverToBoundSessions` |
|
||||
| `prWatchLimit` | `30` | Maximum open bound PRs watched per poll, kept between 1 and 500. Open PRs keep the full budget; merged/closed bindings drain in an additional slice (20% of the limit, at least one) outside it, so they never crowd out open PRs. Each watched PR costs about one `pulls.get` per poll — ~60 requests/hour at the 60s default, so a 5000/hour PAT budget admits ~80 watched PRs (~69 open while a terminal backlog drains at the 1.2x fan-out); over budget, every watch pass aborts on rate limits. Raise `pollInterval` when raising this. Requires `deliverToBoundSessions` and `prWatchEvents` |
|
||||
| `boundSessionActions` | unset | Optional operator instruction templates for bound-session delivery, keyed by event class with optional role prefixes (`review:`, `verify:`, `fix:`). Enables role-aware fan-out. Requires `deliverToBoundSessions` |
|
||||
|
||||
Use `reasonFilter` to drop noisy notification classes such as `ci_activity` or `state_change`. Do not use `reasonFilter: ["mention"]` as a replacement for `groups.*.requireMention`: GitHub's `mention` reason is sticky at the thread level, so real new @mentions can arrive later under `comment`, `subscribed`, `author`, or other reasons and would be skipped.
|
||||
|
||||
@@ -215,6 +216,17 @@ By default every event is handled by a channel-owned session. With
|
||||
creates or reviews a PR) are delivered into that session as a prompt instead,
|
||||
so the session that follows the PR handles the event with its existing context.
|
||||
|
||||
Public-path replies (events with no bound session, including the fallback
|
||||
when a bound session has gone and no replacement exists) run in a background
|
||||
task runner: the poll loop claims the event durably and moves on, so a long
|
||||
agent turn on one PR never blocks watch detection, bound delivery, or intake
|
||||
of new notifications. Public tasks execute one at a time in claim order —
|
||||
which also serializes them against each other on a shared channel session —
|
||||
and a turn interrupted by a restart is recovered from the durable task store.
|
||||
While a thread's public turn runs, the channel stops re-fetching that
|
||||
thread's comments on every poll of the frozen notification window; the
|
||||
thread is re-read once the turn settles.
|
||||
|
||||
- The injected prompt marks all GitHub content as untrusted event data. The
|
||||
session's reply stays in the session; it is **not** published to GitHub
|
||||
automatically, and the session may act on GitHub through its own tools.
|
||||
@@ -317,12 +329,26 @@ settled CI costs one API call per poll; a PR whose CI is still pending is
|
||||
re-checked every 5 minutes until it settles, and settled heads are re-checked
|
||||
every 10 minutes (GitHub does not advance a PR's `updated_at` on check
|
||||
re-runs, so the periodic re-check is what catches a re-run turning red). At
|
||||
most 30 bound PRs are actively polled. A credential failure or rate limit
|
||||
(401/403/429) aborts the whole watch pass for that poll — one shared failure
|
||||
must not issue one doomed request per watched PR — while a per-PR permanent
|
||||
status (404/410/451) only skips that PR. A failing Checks endpoint never
|
||||
suppresses the other three detections: their data comes from `pulls.get`,
|
||||
and the last known CI aggregate stands until a fetch succeeds again.
|
||||
most `prWatchLimit` open bound PRs are actively polled (default 30), most
|
||||
recently bound first. Bindings already recorded as merged/closed drain in an
|
||||
additional slice outside that budget (20% of the limit, at least one, plus
|
||||
any leftover open budget) — each needs one final observation before it
|
||||
leaves the watch set, so dead bindings never crowd out live PRs and the
|
||||
drain still progresses when open PRs fill the limit. A binding the watch
|
||||
already observed closed is tiered as terminal even if the registry still
|
||||
says open. When the open budget cannot cover every open binding, the
|
||||
channel logs `PR watch cap exceeded` with the excluded list; raise
|
||||
`prWatchLimit` (and `pollInterval`) to cover more open PRs. Each watched PR
|
||||
costs about one `pulls.get` per poll — ~60 requests/hour at the 60s default,
|
||||
so a 5000/hour PAT budget admits ~80 watched PRs (~69 open while a terminal
|
||||
backlog drains at the 1.2x fan-out); an over-budget configuration makes
|
||||
every watch pass abort on rate limits. A credential
|
||||
failure or rate limit (401/403/429) aborts the whole watch pass for that poll
|
||||
— one shared failure must not issue one doomed request per watched PR —
|
||||
while a per-PR permanent status (404/410/451) only skips that PR. A failing
|
||||
Checks endpoint never suppresses the other three detections: their data
|
||||
comes from `pulls.get`, and the last known CI aggregate stands until a fetch
|
||||
succeeds again.
|
||||
Transitions are persisted before delivery, so restarts do not re-fire them.
|
||||
|
||||
### Role-Aware Action Templates
|
||||
|
||||
File diff suppressed because it is too large.
Load diff
@@ -51,6 +51,7 @@ interface GithubConfig extends ChannelConfig, RepoFilterConfigFields {
|
||||
useLocalGh?: boolean;
|
||||
deliverToBoundSessions?: boolean;
|
||||
prWatchEvents?: unknown;
|
||||
prWatchLimit?: unknown;
|
||||
boundSessionActions?: unknown;
|
||||
}
|
||||
|
||||
@@ -475,8 +476,19 @@ function isBoundActionKey(key: string): boolean {
|
||||
return (PR_NOTIFICATION_REASONS as readonly string[]).includes(event);
|
||||
}
|
||||
|
||||
/** Bounds one poll's watch fan-out (2-3 GitHub calls per watched PR). */
|
||||
const MAX_WATCHED_PRS = 30;
|
||||
/**
|
||||
* Default open-PR watch fan-out per poll (~1 pulls.get per watched PR plus
|
||||
* conditional Checks fetches; terminal-pending bindings drain in an
|
||||
* additional slice of up to 20% outside this budget). Operators with more
|
||||
* open bindings raise `prWatchLimit` (and `pollInterval`) deliberately;
|
||||
* the watch pass already fails fast on rate limits, so an over-budget
|
||||
* configuration degrades instead of hammering the API. Exported for the
|
||||
* descriptor so the form bound and the effective bound cannot drift.
|
||||
*/
|
||||
export const DEFAULT_PR_WATCH_LIMIT = 30;
|
||||
|
||||
/** Hard ceiling for the configurable watch fan-out. */
|
||||
export const MAX_PR_WATCH_LIMIT = 500;
|
||||
|
||||
/**
|
||||
* After a head change, checks register asynchronously (queued or self-hosted
|
||||
@@ -688,6 +700,7 @@ function isPrWatchEntryShape(value: unknown): value is PrWatchEntry {
|
||||
interface BoundSessionConfigFields {
|
||||
deliverToBoundSessions?: unknown;
|
||||
prWatchEvents?: unknown;
|
||||
prWatchLimit?: unknown;
|
||||
boundSessionActions?: unknown;
|
||||
}
|
||||
|
||||
@@ -787,6 +800,12 @@ export function validateBoundSessionConfig(
|
||||
) {
|
||||
return `[Channel:${channelName}] deliverToBoundSessions must be a boolean.`;
|
||||
}
|
||||
if (
|
||||
config.prWatchLimit !== undefined &&
|
||||
config.deliverToBoundSessions !== true
|
||||
) {
|
||||
return `prWatchLimit for channel ${channelName} requires deliverToBoundSessions: true.`;
|
||||
}
|
||||
try {
|
||||
normalizePrWatchEvents(config, channelName);
|
||||
normalizeBoundSessionActions(config, channelName);
|
||||
@@ -980,6 +999,7 @@ type BoundDeliveryResult =
|
||||
| { kind: 'delivered' }
|
||||
| { kind: 'retired'; reason: string; sessionId?: string }
|
||||
| { kind: 'public' }
|
||||
| { kind: 'requeued' }
|
||||
| { kind: 'deferred' };
|
||||
|
||||
function isOptionalStringArray(value: unknown): value is string[] | undefined {
|
||||
@@ -1430,6 +1450,26 @@ export class GithubChannel extends PollingChannelBase<GithubCursor> {
|
||||
private inboundTaskSnapshot: InboundTaskRecord[] = [];
|
||||
private activeInboundTaskIdsByMessage = new Map<string, string>();
|
||||
private cancelledInboundTaskIds = new Set<string>();
|
||||
/**
|
||||
* Public-path tasks execute in a background FIFO runner (concurrency 1)
|
||||
* instead of inline in the poll loop: a public turn awaits the full agent
|
||||
* reply, and running it inside pollOnce froze watch detection, bound
|
||||
* delivery, and every other notification for the turn's duration. Claims
|
||||
* stay durable and synchronous, so recovery owns execution exactly as it
|
||||
* does for a task interrupted by a crash. Bound and watch deliveries stay
|
||||
* inline: they are fast prompt POSTs whose verdict the dispatch return
|
||||
* value and the cursor gate depend on.
|
||||
*/
|
||||
private publicTaskQueue: string[] = [];
|
||||
private publicTasksInFlight = new Set<string>();
|
||||
private publicTaskRunner: Promise<void> | undefined;
|
||||
private publicTaskRunnerGeneration = 0;
|
||||
private publicTaskWake: (() => void) | undefined;
|
||||
private publicTaskIdleWaiters: Array<() => void> = [];
|
||||
/** The task the runner is executing right now (undefined when parked). */
|
||||
private publicTaskRunningId: string | undefined;
|
||||
/** Escaped runner failures already journaled, once per task id. */
|
||||
private publicTaskFailureJournaled = new Set<string>();
|
||||
private pendingCursorUpdatedAt: string | undefined;
|
||||
private inboundPersistenceBlocked = false;
|
||||
/**
|
||||
@@ -1630,6 +1670,22 @@ export class GithubChannel extends PollingChannelBase<GithubCursor> {
|
||||
this.migrateLegacyPublicationState();
|
||||
this.inboundPersistenceBlocked = false;
|
||||
this.inboundRecoveryPending = true;
|
||||
// A reconnect starts a fresh runner generation; any runner left over
|
||||
// from the previous connection exits after its current task. Keep the
|
||||
// id that orphaned runner is STILL executing in the in-flight set so
|
||||
// recovery cannot run the same task a second time; the orphan's finally
|
||||
// removes it. Queued-but-unstarted ids are dropped here and re-enqueued
|
||||
// by recovery from the durable store.
|
||||
this.publicTaskRunnerGeneration += 1;
|
||||
this.publicTaskRunner = undefined;
|
||||
this.publicTaskQueue = [];
|
||||
this.publicTasksInFlight = new Set(
|
||||
this.publicTaskRunningId ? [this.publicTaskRunningId] : [],
|
||||
);
|
||||
this.publicTaskWake = undefined;
|
||||
const idleWaiters = this.publicTaskIdleWaiters;
|
||||
this.publicTaskIdleWaiters = [];
|
||||
for (const resolve of idleWaiters) resolve();
|
||||
try {
|
||||
this.recoverableInboundTasks = this.readInboundTasks().filter((record) =>
|
||||
this.isRecoverableInboundTask(record),
|
||||
@@ -1666,6 +1722,15 @@ export class GithubChannel extends PollingChannelBase<GithubCursor> {
|
||||
|
||||
disconnect(): void {
|
||||
this.stopPollLoop();
|
||||
// Retire the runner generation: an idle runner wakes and exits, and a
|
||||
// runner mid-turn exits after that turn. The durable task stays
|
||||
// recoverable for the next connection. Release idle waiters: with the
|
||||
// queue retired, "idle" will never come from task completion.
|
||||
this.publicTaskRunnerGeneration += 1;
|
||||
this.publicTaskWake?.();
|
||||
const idleWaiters = this.publicTaskIdleWaiters;
|
||||
this.publicTaskIdleWaiters = [];
|
||||
for (const resolve of idleWaiters) resolve();
|
||||
this.pendingFinalDeliveryRetryStopRequested = true;
|
||||
if (this.pendingFinalDeliveryRequestsActive === 0) {
|
||||
this.pendingFinalDeliveryRetryAbort?.abort();
|
||||
@@ -2476,6 +2541,24 @@ export class GithubChannel extends PollingChannelBase<GithubCursor> {
|
||||
? maxUpdatedAt
|
||||
: this.pendingCursorUpdatedAt;
|
||||
}
|
||||
// Snapshot the threads whose in-flight work holds the cursor BEFORE the
|
||||
// lanes run; see threadHeldAtPollStart.
|
||||
this.pollStartHeldThreads = new Set();
|
||||
try {
|
||||
for (const task of this.readInboundTasks()) {
|
||||
if (
|
||||
this.publicTasksInFlight.has(task.id) &&
|
||||
this.isCursorBlockingTask(task)
|
||||
) {
|
||||
this.pollStartHeldThreads.add(
|
||||
`${task.source.chatId}|${task.source.threadId}`,
|
||||
);
|
||||
}
|
||||
}
|
||||
} catch {
|
||||
// Store unreadable: fail open (lanes fetch normally); the claim path
|
||||
// surfaces the persistence failure itself.
|
||||
}
|
||||
let reasonFilterJournaled = 0;
|
||||
let reasonFilterCollapsed = 0;
|
||||
let repoFilterJournaled = 0;
|
||||
@@ -2944,6 +3027,10 @@ export class GithubChannel extends PollingChannelBase<GithubCursor> {
|
||||
ctx: NotificationContext,
|
||||
reason: 'review_requested' | 'assign',
|
||||
): Promise<void> {
|
||||
// Held thread: the trigger stays unclaimed and the frozen window
|
||||
// re-lists it until the running task settles (no first-contact
|
||||
// semantics here, so skipping the whole lane loses nothing).
|
||||
if (this.threadHeldAtPollStart(ctx.chatId, ctx.threadId)) return;
|
||||
const trigger = await this.findDirectTrigger(ctx, reason);
|
||||
if (!trigger) return;
|
||||
|
||||
@@ -3143,6 +3230,9 @@ export class GithubChannel extends PollingChannelBase<GithubCursor> {
|
||||
private async fetchNewComments(
|
||||
ctx: NotificationContext,
|
||||
): Promise<GithubComment[]> {
|
||||
// The thread's running task already claimed its comments; re-fetching
|
||||
// every poll of the frozen window only spends the shared PAT budget.
|
||||
if (this.threadHeldAtPollStart(ctx.chatId, ctx.threadId)) return [];
|
||||
const [owner, repo] = ctx.chatId.split('/');
|
||||
const comments = (await this.githubApi(
|
||||
() =>
|
||||
@@ -3215,6 +3305,13 @@ export class GithubChannel extends PollingChannelBase<GithubCursor> {
|
||||
const { chatId, threadId, issueNumber } = ctx;
|
||||
const bodyKey = `${chatId}|${threadId}`;
|
||||
if (this.cursor.dispatchedBodies?.includes(bodyKey)) return;
|
||||
// No deferral gate here: the body dispatch claims a public task that the
|
||||
// background runner executes in FIFO order behind any comment turn already
|
||||
// claimed for this thread, so the two can never overlap (and steer-cancel
|
||||
// each other) on the channel session. Deferring the feed instead would
|
||||
// risk losing the body permanently: nothing reserves it, and the
|
||||
// end-of-poll cursor commit can mark the thread read once the blocking
|
||||
// task settles, after which first contact is unreachable (!lastReadAt).
|
||||
// Fetch and dispatch failures reach pollOnce, which keeps the
|
||||
// notification unread when GitHub may answer differently next poll.
|
||||
const issue = await this.fetchIssueMeta(ctx);
|
||||
@@ -3476,7 +3573,15 @@ export class GithubChannel extends PollingChannelBase<GithubCursor> {
|
||||
undefined,
|
||||
privateOnly,
|
||||
);
|
||||
return this.runInboundTask(task);
|
||||
// The claim is durable; execution belongs to the background runner.
|
||||
// Awaiting the agent turn here would freeze the whole poll loop
|
||||
// (watch, bound delivery, other notifications) for the turn's duration.
|
||||
// A private-only task is retired inside runInboundTask, never published.
|
||||
// The absence gate runs synchronously so a deferred event keeps the
|
||||
// false verdict the lane's body-suppression semantics depend on.
|
||||
if (this.applyPublicAbsenceGate(task)) return false;
|
||||
this.enqueuePublicTask(task);
|
||||
return true;
|
||||
}
|
||||
|
||||
private claimInboundTask(
|
||||
@@ -3556,6 +3661,155 @@ export class GithubChannel extends PollingChannelBase<GithubCursor> {
|
||||
return task;
|
||||
}
|
||||
|
||||
/**
|
||||
* Hand a claimed public task to the background runner. Idempotent: a
|
||||
* re-listed event re-claiming a queued/running task, or a recovery pass
|
||||
* re-scanning it, must not execute it twice.
|
||||
*/
|
||||
private enqueuePublicTask(task: InboundTaskRecord): void {
|
||||
if (this.publicTasksInFlight.has(task.id)) return;
|
||||
if (!this.isRecoverableInboundTask(task)) return;
|
||||
this.publicTasksInFlight.add(task.id);
|
||||
this.publicTaskQueue.push(task.id);
|
||||
this.startPublicTaskRunner();
|
||||
this.publicTaskWake?.();
|
||||
}
|
||||
|
||||
private startPublicTaskRunner(): void {
|
||||
if (this.publicTaskRunner) return;
|
||||
const generation = this.publicTaskRunnerGeneration;
|
||||
const runner = (async () => {
|
||||
while (generation === this.publicTaskRunnerGeneration) {
|
||||
const id = this.publicTaskQueue.shift();
|
||||
if (id === undefined) {
|
||||
await new Promise<void>((resolve) => {
|
||||
this.publicTaskWake = resolve;
|
||||
});
|
||||
this.publicTaskWake = undefined;
|
||||
continue;
|
||||
}
|
||||
let task: InboundTaskRecord | undefined;
|
||||
this.publicTaskRunningId = id;
|
||||
try {
|
||||
task = this.readInboundTasks().find((t) => t.id === id);
|
||||
if (task) await this.runInboundTask(task);
|
||||
this.publicTaskFailureJournaled.delete(id);
|
||||
} catch (err) {
|
||||
// runInboundTask owns failure transitions; an unexpected throw
|
||||
// here must not kill the runner or strand the remaining queue.
|
||||
// Journal it once per task so the operator-visible timeline shows
|
||||
// the failure (the poll loop's per-thread catch no longer sees it).
|
||||
const message = sanitizeLogText(
|
||||
err instanceof Error ? err.message : String(err),
|
||||
200,
|
||||
);
|
||||
process.stderr.write(
|
||||
`[Channel:${this.name}] public task runner error for ${sanitizeLogText(id, 128)}: ${message}\n`,
|
||||
);
|
||||
if (!this.publicTaskFailureJournaled.has(id)) {
|
||||
this.publicTaskFailureJournaled.add(id);
|
||||
this.recordChannelEvent({
|
||||
type: 'inbound',
|
||||
outcome: 'failed',
|
||||
repository: task?.source.chatId,
|
||||
thread: task?.source.threadId,
|
||||
number: task?.issueNumber,
|
||||
messageId: task?.source.messageId,
|
||||
error: message,
|
||||
});
|
||||
}
|
||||
} finally {
|
||||
this.publicTaskRunningId = undefined;
|
||||
this.publicTasksInFlight.delete(id);
|
||||
if (
|
||||
this.publicTaskQueue.length === 0 &&
|
||||
this.publicTasksInFlight.size === 0
|
||||
) {
|
||||
const waiters = this.publicTaskIdleWaiters;
|
||||
this.publicTaskIdleWaiters = [];
|
||||
for (const resolve of waiters) resolve();
|
||||
}
|
||||
}
|
||||
}
|
||||
})().finally(() => {
|
||||
if (this.publicTaskRunner === runner) {
|
||||
this.publicTaskRunner = undefined;
|
||||
// Work enqueued while this runner was winding down must not strand,
|
||||
// but a retired generation (disconnect/reconnect) must NOT start a
|
||||
// new runner: the queued tasks stay durable for the next connect's
|
||||
// recovery. Only the current generation may continue.
|
||||
if (
|
||||
generation === this.publicTaskRunnerGeneration &&
|
||||
this.publicTaskQueue.length > 0
|
||||
) {
|
||||
this.startPublicTaskRunner();
|
||||
}
|
||||
}
|
||||
});
|
||||
this.publicTaskRunner = runner;
|
||||
}
|
||||
|
||||
/**
|
||||
* Resolves once no public task is queued or running (tests/diagnostics).
|
||||
* Signal-based, not timer-based: callers may run under fake timers.
|
||||
*/
|
||||
async whenPublicTasksIdle(): Promise<void> {
|
||||
if (
|
||||
this.publicTaskQueue.length === 0 &&
|
||||
this.publicTasksInFlight.size === 0
|
||||
) {
|
||||
return;
|
||||
}
|
||||
await new Promise<void>((resolve) => {
|
||||
this.publicTaskIdleWaiters.push(resolve);
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* The public absence gate: defer (or at the horizon retire) an unmatched
|
||||
* PR event while the binding snapshot cannot prove absence. Returns true
|
||||
* when the task must not execute now. Called synchronously at dispatch so
|
||||
* the lane keeps its false-verdict semantics (a deferred comment suppresses
|
||||
* the first-contact body this poll), and again in runInboundTask for tasks
|
||||
* recovery enqueues after a re-resolve.
|
||||
*/
|
||||
private applyPublicAbsenceGate(task: InboundTaskRecord): boolean {
|
||||
const envelope = task.envelope;
|
||||
if (!envelope) return false;
|
||||
const bound = task.delivery?.kind === 'bound-session';
|
||||
const isWatch = envelope.messageId?.startsWith('watch:') === true;
|
||||
if (
|
||||
bound ||
|
||||
isWatch ||
|
||||
!this.deliverToBoundSessions ||
|
||||
envelope.threadId?.startsWith('pr:') !== true
|
||||
) {
|
||||
return false;
|
||||
}
|
||||
if (
|
||||
this.boundSnapshotTrusted &&
|
||||
!this.boundRefreshFailed &&
|
||||
!this.lastBindingsPartial
|
||||
) {
|
||||
return false;
|
||||
}
|
||||
const deferCount = (task.deferCount ?? 0) + 1;
|
||||
if (deferCount >= MAX_BOUND_DEFERRALS) {
|
||||
this.retireBoundTask(
|
||||
task,
|
||||
`binding snapshot never completed after ${deferCount} deferrals; absence never established`,
|
||||
);
|
||||
return true;
|
||||
}
|
||||
this.transitionInboundTask(task.id, 'failed', {
|
||||
error: 'binding snapshot incomplete; absence not established',
|
||||
attempts: task.attempts ?? 0,
|
||||
deferCount,
|
||||
backpressure: true,
|
||||
});
|
||||
return true;
|
||||
}
|
||||
|
||||
private async runInboundTask(task: InboundTaskRecord): Promise<boolean> {
|
||||
if (!this.isRecoverableInboundTask(task)) return true;
|
||||
const envelope = task.envelope;
|
||||
@@ -3577,31 +3831,7 @@ export class GithubChannel extends PollingChannelBase<GithubCursor> {
|
||||
// not stall (or dead-letter) behind a degraded binding scan, including
|
||||
// a permanent partial where a complete snapshot never lands. Known
|
||||
// targets (bound tasks) are unaffected and proceed immediately.
|
||||
if (
|
||||
!bound &&
|
||||
!isWatch &&
|
||||
this.deliverToBoundSessions &&
|
||||
envelope.threadId?.startsWith('pr:') === true &&
|
||||
(!this.boundSnapshotTrusted ||
|
||||
this.boundRefreshFailed ||
|
||||
this.lastBindingsPartial)
|
||||
) {
|
||||
const deferCount = (task.deferCount ?? 0) + 1;
|
||||
if (deferCount >= MAX_BOUND_DEFERRALS) {
|
||||
this.retireBoundTask(
|
||||
task,
|
||||
`binding snapshot never completed after ${deferCount} deferrals; absence never established`,
|
||||
);
|
||||
return false;
|
||||
}
|
||||
this.transitionInboundTask(task.id, 'failed', {
|
||||
error: 'binding snapshot incomplete; absence not established',
|
||||
attempts: task.attempts ?? 0,
|
||||
deferCount,
|
||||
backpressure: true,
|
||||
});
|
||||
return false;
|
||||
}
|
||||
if (this.applyPublicAbsenceGate(task)) return false;
|
||||
const attempts = (task.attempts ?? 0) + 1;
|
||||
this.activeInboundTaskIdsByMessage.set(
|
||||
this.inboundMessageKey(envelope.chatId, envelope.messageId),
|
||||
@@ -3631,6 +3861,10 @@ export class GithubChannel extends PollingChannelBase<GithubCursor> {
|
||||
if (bound || isWatch) {
|
||||
boundResult = await this.deliverToBoundSession(task, envelope);
|
||||
if (boundResult.kind === 'deferred') return false;
|
||||
// The gone-session fallback requeued the task as public work: the
|
||||
// runner owns execution and bookkeeping from here; this inline frame
|
||||
// must not run the public post-run path against the still-queued task.
|
||||
if (boundResult.kind === 'requeued') return true;
|
||||
} else if (
|
||||
this.boundRoleAware &&
|
||||
this.boundEventAlreadyDelivered(envelope)
|
||||
@@ -3893,6 +4127,13 @@ export class GithubChannel extends PollingChannelBase<GithubCursor> {
|
||||
|
||||
for (const task of tasks) {
|
||||
if (!task.envelope) continue;
|
||||
// Skip tasks this process already owns BEFORE any branch mutates them:
|
||||
// the retarget branch below would otherwise re-resolve a backpressure
|
||||
// parked task whose public turn is running in the background runner
|
||||
// (the running transition merges patches, so backpressure: true
|
||||
// survives) and deliver it bound a second time. A running task with no
|
||||
// in-flight entry is a genuine crash and must still be recovered.
|
||||
if (this.publicTasksInFlight.has(task.id)) continue;
|
||||
this.applyTaskDedupe(task.dedupe);
|
||||
if (
|
||||
this.watchReconcileBlocked &&
|
||||
@@ -4032,7 +4273,17 @@ export class GithubChannel extends PollingChannelBase<GithubCursor> {
|
||||
continue;
|
||||
}
|
||||
}
|
||||
await this.runInboundTask(task);
|
||||
if (
|
||||
task.delivery?.kind === 'bound-session' ||
|
||||
task.envelope.messageId?.startsWith('watch:') === true
|
||||
) {
|
||||
// Bound and watch deliveries are fast prompt POSTs whose verdict
|
||||
// recovery applies inline; public turns go to the background runner
|
||||
// so a long reply cannot stall the poll loop.
|
||||
await this.runInboundTask(task);
|
||||
} else {
|
||||
this.enqueuePublicTask(task);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -4069,6 +4320,28 @@ export class GithubChannel extends PollingChannelBase<GithubCursor> {
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* Threads whose in-flight public work already held the cursor when this
|
||||
* poll's notification loop started. Snapshot, not live: a task claimed
|
||||
* during this poll must not suppress the same thread's later lanes, and a
|
||||
* crashed task that is not in flight must stay re-dispatchable (recovery
|
||||
* may have failed this poll). While a thread is held, the frozen window
|
||||
* re-lists it every poll; re-fetching its comments each time spends the
|
||||
* shared PAT budget for nothing (dedupe already claimed them). The thread
|
||||
* is re-read when its task settles. The first-contact body feed is NOT
|
||||
* skipped: it claims once and the runner serializes it behind the running
|
||||
* turn, so deferring it would only risk the cursor committing past an
|
||||
* unclaimed body.
|
||||
*/
|
||||
private pollStartHeldThreads = new Set<string>();
|
||||
|
||||
private threadHeldAtPollStart(
|
||||
chatId: string,
|
||||
threadId: string | undefined,
|
||||
): boolean {
|
||||
return this.pollStartHeldThreads.has(`${chatId}|${threadId}`);
|
||||
}
|
||||
|
||||
private isCursorBlockingTask(task: InboundTaskRecord): boolean {
|
||||
return (
|
||||
this.isRecoverableInboundTask(task) &&
|
||||
@@ -5716,8 +5989,11 @@ export class GithubChannel extends PollingChannelBase<GithubCursor> {
|
||||
if (task.privateOnly === true) {
|
||||
return this.retirePrivateOnly(task, envelope);
|
||||
}
|
||||
await this.handleInbound(envelope);
|
||||
return { kind: 'public' };
|
||||
// The fallback is a full public turn: serialize it through the same
|
||||
// runner so it neither blocks the poll lane nor overlaps (and
|
||||
// steer-cancels) another public turn on a shared channel session.
|
||||
this.enqueuePublicTask(task);
|
||||
return { kind: 'requeued' };
|
||||
}
|
||||
// The target was frozen at claim/detection time. Re-validate it against
|
||||
// the latest COMPLETE snapshot: a live rebind must not keep delivering
|
||||
@@ -5962,8 +6238,11 @@ export class GithubChannel extends PollingChannelBase<GithubCursor> {
|
||||
}
|
||||
this.transitionInboundTask(task.id, 'running', { delivery: undefined });
|
||||
task.delivery = undefined;
|
||||
await this.handleInbound(envelope);
|
||||
return { kind: 'public' };
|
||||
// Gone-session fallback is a full public turn: hand it to the serial
|
||||
// runner instead of awaiting it in the poll lane (head-of-line block)
|
||||
// or overlapping another public turn on a shared session (steer-cancel).
|
||||
this.enqueuePublicTask(task);
|
||||
return { kind: 'requeued' };
|
||||
}
|
||||
return { kind: 'delivered' };
|
||||
}
|
||||
@@ -6128,6 +6407,24 @@ export class GithubChannel extends PollingChannelBase<GithubCursor> {
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Open-PR watch fan-out per poll (terminal-pending bindings drain in an
|
||||
* additional slice outside it). Clamped like pollInterval: unset or
|
||||
* invalid keeps the default; a fractional count rounds up so the
|
||||
* configured budget is never silently smaller than requested.
|
||||
*/
|
||||
private get prWatchLimit(): number {
|
||||
const configured: unknown = (this.config as GithubConfig).prWatchLimit;
|
||||
if (
|
||||
typeof configured === 'number' &&
|
||||
Number.isFinite(configured) &&
|
||||
configured > 0
|
||||
) {
|
||||
return Math.min(Math.max(Math.ceil(configured), 1), MAX_PR_WATCH_LIMIT);
|
||||
}
|
||||
return DEFAULT_PR_WATCH_LIMIT;
|
||||
}
|
||||
|
||||
/**
|
||||
* Edge-triggered state watch over the PRs bound to live sessions. Runs
|
||||
* inside the existing poll loop; detection is API-level and costs no model
|
||||
@@ -6142,7 +6439,7 @@ export class GithubChannel extends PollingChannelBase<GithubCursor> {
|
||||
>();
|
||||
const active = new Map<
|
||||
string,
|
||||
{ chatId: string; number: number; updatedAt: string }
|
||||
{ chatId: string; number: number; updatedAt: string; terminal: boolean }
|
||||
>();
|
||||
for (const summary of this.lastBindings) {
|
||||
if (summary.sourceType === 'channel') continue;
|
||||
@@ -6180,6 +6477,15 @@ export class GithubChannel extends PollingChannelBase<GithubCursor> {
|
||||
chatId: identity.chatId,
|
||||
number: identity.number,
|
||||
updatedAt,
|
||||
// Tiering only — observation stays API-authoritative. A binding
|
||||
// still saying open while the watch already observed closed must
|
||||
// not hold an open-tier slot (stale registry, e.g. refresh
|
||||
// disabled); a genuinely reopened PR gets one leftover-budget
|
||||
// observation and returns to the open tier, self-correcting.
|
||||
terminal:
|
||||
pr.state === 'merged' ||
|
||||
pr.state === 'closed' ||
|
||||
this.prWatchState.entries[key]?.prState === 'closed',
|
||||
});
|
||||
}
|
||||
}
|
||||
@@ -6244,19 +6550,51 @@ export class GithubChannel extends PollingChannelBase<GithubCursor> {
|
||||
}
|
||||
}
|
||||
}
|
||||
const watched = [...active.entries()]
|
||||
.sort((a, b) => b[1].updatedAt.localeCompare(a[1].updatedAt))
|
||||
.slice(0, MAX_WATCHED_PRS);
|
||||
if (active.size > MAX_WATCHED_PRS) {
|
||||
// Open bindings take the FULL watch budget: a merged or closed PR that
|
||||
// the watch has not observed yet must never crowd a live PR out of it,
|
||||
// even when open bindings fill the limit. Terminal-pending bindings
|
||||
// drain in an ADDITIONAL slice outside the open budget (20% of the
|
||||
// limit, at least one, plus any leftover open budget): without it they
|
||||
// would never receive their one terminal observation when open PRs
|
||||
// fill the limit, and the active set would grow with every merge. The
|
||||
// extra slice costs at most ceil(limit*0.2) additional pulls.get per
|
||||
// poll while a backlog drains; each drained binding leaves the active
|
||||
// set for good, so the fan-out returns to the open budget once the
|
||||
// backlog clears.
|
||||
const limit = this.prWatchLimit;
|
||||
const byRecency = (
|
||||
a: [string, { updatedAt: string }],
|
||||
b: [string, { updatedAt: string }],
|
||||
) => b[1].updatedAt.localeCompare(a[1].updatedAt);
|
||||
const openEntries = [...active.entries()]
|
||||
.filter(([, candidate]) => !candidate.terminal)
|
||||
.sort(byRecency);
|
||||
const terminalEntries = [...active.entries()]
|
||||
.filter(([, candidate]) => candidate.terminal)
|
||||
.sort(byRecency);
|
||||
const terminalReserved =
|
||||
terminalEntries.length > 0
|
||||
? Math.min(terminalEntries.length, Math.max(1, Math.ceil(limit * 0.2)))
|
||||
: 0;
|
||||
const watched = openEntries.slice(0, limit);
|
||||
const terminalSlots = Math.max(terminalReserved, limit - watched.length);
|
||||
watched.push(
|
||||
...terminalEntries.slice(
|
||||
0,
|
||||
Math.min(terminalEntries.length, terminalSlots),
|
||||
),
|
||||
);
|
||||
if (watched.length < active.size) {
|
||||
// The cap deterministically excludes the oldest bindings every poll.
|
||||
// Surface it (dead-letter principle: surfaced, never silently
|
||||
// consumed) so an operator knows some bound PRs are never watched.
|
||||
// consumed) so an operator knows some bound PRs are not watched this
|
||||
// pass.
|
||||
const watchedKeys = new Set(watched.map(([key]) => key));
|
||||
const excluded = [...active.keys()].filter(
|
||||
(key) => !watchedKeys.has(key),
|
||||
);
|
||||
process.stderr.write(
|
||||
`[Channel:${this.name}] PR watch cap exceeded: watching ${MAX_WATCHED_PRS}/${active.size} bound PRs (most recent); not watched: ${excluded.join(', ')}\n`,
|
||||
`[Channel:${this.name}] PR watch cap exceeded: watching ${watched.length}/${active.size} bound PRs (open first, most recent; open limit ${limit}); not watched: ${excluded.join(', ')}\n`,
|
||||
);
|
||||
}
|
||||
for (const [key, candidate] of watched) {
|
||||
|
||||
@@ -6,6 +6,8 @@ import type { ChannelPlugin } from '@qwen-code/channel-base';
|
||||
import {
|
||||
GithubChannel,
|
||||
validateBoundSessionConfig,
|
||||
DEFAULT_PR_WATCH_LIMIT,
|
||||
MAX_PR_WATCH_LIMIT,
|
||||
validateRepoFilterConfig,
|
||||
PR_NOTIFICATION_REASONS,
|
||||
PR_WATCH_EVENTS,
|
||||
@@ -177,6 +179,16 @@ export const plugin: ChannelPlugin = {
|
||||
// never drift from the validator/emit path (R2-18).
|
||||
options: PR_WATCH_EVENTS.map((value) => ({ value, label: value })),
|
||||
},
|
||||
{
|
||||
key: 'prWatchLimit',
|
||||
label: 'PR Watch Limit',
|
||||
kind: 'number',
|
||||
exclusiveMinimum: 0,
|
||||
maximum: MAX_PR_WATCH_LIMIT,
|
||||
default: String(DEFAULT_PR_WATCH_LIMIT),
|
||||
description:
|
||||
'Maximum open bound PRs watched per poll (requires Deliver To PR-Bound Sessions and PR Watch Events). Open PRs keep the full budget; merged/closed bindings drain in an additional slice (20% of the limit, at least one) outside it, so they never crowd out open PRs. Each watched PR costs about one pulls.get per poll — at the 60s default that is ~60 requests/hour per PR, so a 5000/hour PAT budget admits ~80 watched PRs (~69 open while a terminal backlog drains at the 1.2x fan-out); an over-budget configuration makes every watch pass abort on rate limits. Raise Poll Interval when raising this. Leave empty for the default.',
|
||||
},
|
||||
{
|
||||
key: 'boundSessionActions',
|
||||
label: 'Bound Session Actions',
|
||||
|
||||
@@ -123,6 +123,7 @@ const FIELD_LABEL_KEYS: Record<string, Record<string, string>> = {
|
||||
deliverToBoundSessions:
|
||||
'channels.editor.field.github.deliverToBoundSessions',
|
||||
prWatchEvents: 'channels.editor.field.github.prWatchEvents',
|
||||
prWatchLimit: 'channels.editor.field.github.prWatchLimit',
|
||||
boundSessionActions: 'channels.editor.field.github.boundSessionActions',
|
||||
},
|
||||
gitlab: {
|
||||
@@ -147,11 +148,12 @@ const SHARED_SESSION_FIELD_KEYS = new Set([
|
||||
'multiSession',
|
||||
'instructions',
|
||||
]);
|
||||
// One panel: watch events and action templates only mean something with
|
||||
// bound delivery enabled.
|
||||
// One panel: watch events, the watch limit, and action templates only mean
|
||||
// something with bound delivery enabled.
|
||||
const BOUND_SESSION_FIELD_KEYS = new Set([
|
||||
'deliverToBoundSessions',
|
||||
'prWatchEvents',
|
||||
'prWatchLimit',
|
||||
'boundSessionActions',
|
||||
]);
|
||||
|
||||
|
||||
@@ -4799,6 +4799,11 @@ const EN: Messages = {
|
||||
'head_change — the PR’s head commit changes',
|
||||
'channels.editor.field.github.prWatchEvents.option.state_change':
|
||||
'state_change — the PR is merged or closed',
|
||||
'channels.editor.field.github.prWatchLimit': 'PR Watch Limit',
|
||||
'channels.editor.field.github.prWatchLimit.description':
|
||||
'Maximum open bound PRs watched per poll. Open PRs keep the full budget; merged/closed bindings drain in an additional slice (20% of the limit, at least one) outside it, so they never crowd out open PRs. Each watched PR costs about one API request per poll — at the 60s default, a 5000/hour token budget admits ~80 watched PRs (~69 open while a terminal backlog drains at the 1.2x fan-out); over budget, every watch pass aborts on rate limits. Raise the poll interval when raising this. Requires Deliver To PR-Bound Sessions and PR Watch Events. Leave empty for the default (30).',
|
||||
'channels.editor.field.github.prWatchLimit.disabledHint':
|
||||
'Enable "Deliver To PR-Bound Sessions" above to set the watch limit.',
|
||||
'channels.editor.field.github.boundSessionActions': 'Bound Session Actions',
|
||||
'channels.editor.field.github.boundSessionActions.description':
|
||||
'Optional. Instruction templates delivered to PR-bound sessions. Pick an event key below; a role prefix (review:, verify:, fix:) targets that binding role only. Requires Deliver To PR-Bound Sessions. Each role with a matching key gets its own delivery; roles without one skip the event. Variables: %repo%, %pr%, %url%, %title%, %event%, %sender%, %role%; %% for a literal %. Leave empty for the default generic prompt.',
|
||||
@@ -9513,6 +9518,11 @@ const ZH: Messages = {
|
||||
'head_change — PR 的 head 提交变更',
|
||||
'channels.editor.field.github.prWatchEvents.option.state_change':
|
||||
'state_change — PR 被合并或关闭',
|
||||
'channels.editor.field.github.prWatchLimit': 'PR 监视上限',
|
||||
'channels.editor.field.github.prWatchLimit.description':
|
||||
'每轮轮询最多监视的 open 绑定 PR 数。open PR 独占全部预算；已合并/关闭的绑定在预算之外的一块额外名额（上限的 20%，至少一个）中排出，因此绝不会挤占 open PR。每个被监视的 PR 每轮约消耗一次 API 请求——按默认 60 秒间隔，5000 次/小时的 token 预算约容纳 80 个被监视 PR（终态积压按 1.2 倍扇出排出时约 69 个 open）；超出预算会让每轮监视都因限流中止。提高本值时请同时提高轮询间隔。需开启"投递到 PR 绑定会话"并配置监视事件。留空则使用默认值（30）。',
|
||||
'channels.editor.field.github.prWatchLimit.disabledHint':
|
||||
'开启上方「投递到 PR 绑定会话」后即可设置监视上限。',
|
||||
'channels.editor.field.github.boundSessionActions': '绑定会话动作模板',
|
||||
'channels.editor.field.github.boundSessionActions.description':
|
||||
'可选。投递给 PR 绑定会话的 operator 指令模板。在下方选择事件键；加角色前缀（review:、verify:、fix:）可只定向该绑定角色。需开启"投递到 PR 绑定会话"。有匹配键的角色各自投递，没有匹配键的角色跳过该事件。变量：%repo%、%pr%、%url%、%title%、%event%、%sender%、%role%；%% 表示字面 %。留空则使用默认通用 prompt。',
|
||||
|
||||
Reference in new issue
Block a user