fix(poller): cap diff-path files per run against the subrequest budget [poller, pipeline, docs, tests]

docs/0-requirements.ja.md

@@ -188,11 +188,17 @@ commit diff poller は 2-phase 構成:
 - **forward phase** — `(lastPolledAt, pollStartTime]` の window を列挙し、その中の**古い側から**取り込む（webhook 取りこぼし時の redundancy）。watermark namespace は `diffs:${repo}`。
 - **backward phase** — `until=oldestUnprocessedDate` で履歴を徐々に遡行する（新規 deployment や webhook 起動前の commit を backfill する経路）。watermark namespace は `diffs_backfill:${repo}`。
 
-1 run あたりの取り込み上限は forward / backward それぞれ 5 commits。`processAndUpsertCommitDiff` の upsert は `(repo, commit_sha, file_path)` で idempotent なので、webhook / 両 phase 間で overlap しても副作用はない。
+1 repo 1 run あたりの上限は、各 phase が 2 軸で持つ: commit 数 5 件と、file 数 `diffFileBudgetPerPhase`。`processAndUpsertCommitDiff` の upsert は `(repo, commit_sha, file_path)` で idempotent なので、webhook / 両 phase 間で overlap しても副作用はない。
 
-両 phase は同一の watermark 不変条件に従う: **取り込みに成功していない commit を watermark が追い越さない。** commit が「取り込み済み」とみなされるのは、detail 取得が成功し、かつ `processAndUpsertCommitDiff` の failed が 0 の場合のみ（embedding / Vectorize の失敗は throw ではなく戻り値で報告されるため、戻り値も判定に含める）。判定は dense 側のみを見る: D1 FTS mirror / store row の失敗は pipeline 側で log のみ・failed に計上しない設計で、vector は既に landing 済み・sparse index は reindex で reconcile されるため。mirror 側まで watermark の条件にすると、FTS 側の障害が diff surface 全体を止めてしまう。帰結:
+commit 数は phase の消費量を縛れない。1 file の index は worst case で 3 subrequest（D1 FTS mirror 書き込み、store row、および embed batch の按分——1 batch は 2 subrequest を使い、`MAX_EMBEDDING_INPUT_CHARS` が 1 input を 8000 文字に切る一方 batch 予算は 30000 token なので、1 batch は最低 3 file を載せる）であり、1 commit は最大 300 file を運ぶため、5 commits のコストは 2 桁の幅を持つ。実測された帰結: `POLL_REPOS` の末尾 repo にある 44 file の commit で、embed batch 3 本すべてが `Too many subrequests by single Worker invocation` で拒否された。同じ loop の手前にある repo は正常に index されていた。そして後述の不変条件がその commit で watermark を止める——token 軸が起こしたのとまったく同じ、毎 cron 決定論的に再現する停止が、file 軸で起きた（issue #238）。
 
-- 失敗した commit と、1 run 上限で持ち越された commit は、次回 run の window に残る
+そこで diff surface は invocation 予算のうち自分の取り分を明示する: `DIFF_SUBREQUEST_BUDGET_PER_RUN` = Cloudflare が許す 1000 のうち 900。diffs は専用 cron を持つので、この天井を docs / wiki / issue / release と共有はしていない。しかし `POLL_REPOS` の全 repo と両 phase では共有しており、上記の失敗を生んだのはそちらの共有である。`diffFileBudgetPerPhase` は宣言した取り分を `repoCount × 2` で割り、残りを file 数に換算する。literal を固定せず repo list から導出するのは意図的である: `POLL_REPOS` は通常の config commit で増える（issue #233 が 6 番目の repo を追加した）ため、その日の list に合わせた literal は次の追加で天井を超え、しかもその超過は原因となった変更ではなく loop が最後に到達した repo の失敗として現れる。repo list が長く予算を下回るところまで割った場合は 5 file の下限が効く——超過は次の cron が再試行するが、予算 0 は全 diff watermark を恒久的に止める。それはいま取り除こうとしている失敗そのものである。
+
+file 予算は 1 commit を複数 run に分割しうる。間引きはしない: 1 run は残っている file の先頭側を index し、次の run は structured store（`GET /diffs`）から index 済み path を読み戻してそこから続ける。この読み戻しが分割を収束させる——これが無いと、上限付きの run が毎 cron 同じ先頭側を再 index し、その commit は永遠に完了しない。読み戻しを行うのは file 数が予算を超える commit だけなので、通常の commit はその判定に subrequest を払わない。またこの読み戻しが store 側エラーで失敗した場合は「まだ何も index されていない」と読むのではなく commit の失敗として扱う。前者は分割を黙って先頭からやり直させてしまう。
+
+両 phase は同一の watermark 不変条件に従う: **取り込みに成功していない commit を watermark が追い越さない。** commit が「取り込み済み」とみなされるのは、detail 取得が成功し、かつ `processAndUpsertCommitDiff` の failed が 0 で、**さらに未処理の file が残っていない**場合のみ（embedding / Vectorize の失敗は throw ではなく戻り値で報告されるため、戻り値も判定に含める）。判定は dense 側のみを見る: D1 FTS mirror / store row の失敗は pipeline 側で log のみ・failed に計上しない設計で、vector は既に landing 済み・sparse index は reindex で reconcile されるため。mirror 側まで watermark の条件にすると、FTS 側の障害が diff surface 全体を止めてしまう。帰結:
+
+- 失敗した commit、file 予算で一部しか index されなかった commit、そして 2 つの 1 run 上限のいずれかで持ち越された commit は、いずれも次回 run の window に残る。一部だけ index された commit はエラーではなく分割の途中であり、watermark を止めることが残りの file への到達性を保つ
 - commit list 取得が失敗した run は watermark を動かさないので、その期間全体が再試行対象のまま残る
 - forward phase は window の古い端を確定してからでないと watermark を進められないため、window は 100 件/page で列挙する。page が満杯なら「列挙不能」とみなして window 終端を半分に縮める（最大 8 回）。それでも収まらない場合は watermark を保持して log に出す — silent な欠損より、観測できる stall を選ぶ。
 

docs/0-requirements.md

@@ -193,11 +193,17 @@ The commit-diff poller runs in two phases:
 - **forward phase** — enumerates the window `(lastPolledAt, pollStartTime]` and ingests its **oldest** commits first, acting as redundancy when webhook delivery has stalled. Watermark namespace: `diffs:${repo}`.
 - **backward phase** — walks backward through history using `until=oldestUnprocessedDate`, backfilling commits that predate the webhook or a fresh deployment. Watermark namespace: `diffs_backfill:${repo}`.
 
-Each phase ingests at most 5 commits per repo per run. Upserts through `processAndUpsertCommitDiff` are idempotent on `(repo, commit_sha, file_path)`, so overlap between webhook and either phase is safe.
+Each phase is bounded on two axes per repo per run: at most 5 commits, and at most `diffFileBudgetPerPhase` files. Upserts through `processAndUpsertCommitDiff` are idempotent on `(repo, commit_sha, file_path)`, so overlap between webhook and either phase is safe.
 
-Both phases obey one watermark invariant: **a watermark never moves past a commit that has not been successfully ingested.** A commit counts as ingested only when its detail fetch succeeded and `processAndUpsertCommitDiff` reported zero failed files — embedding and Vectorize failures are reported by return value rather than by throwing, so the return value is part of the check. The criterion is the dense side only: a D1 FTS mirror or store-row failure is logged by the pipeline and deliberately not counted, since the vector has already landed and the sparse index reconciles on reindex; gating the watermark on the mirror would let an FTS-side outage stall the whole diff surface. Consequences:
+A commit count does not bound what a phase spends. Each indexed file costs 3 subrequests worst case — the D1 FTS mirror write, the store row, and the amortised share of its embed batch (a batch spends 2 and holds at least 3 files, since `MAX_EMBEDDING_INPUT_CHARS` caps an input at 8000 characters against a 30000-token batch budget) — and one commit carries up to 300 files, so 5 commits span two orders of magnitude of cost. The measured consequence: a 44-file commit in the last repo of `POLL_REPOS` had all three of its embed batches rejected with `Too many subrequests by single Worker invocation` while the repos ahead of it in the loop indexed normally, and the invariant below then held its watermark on that commit — the same deterministic, cron-after-cron stall the token axis produced, on the file axis (issue #238).
 
-- a failed commit, and every commit the per-run cap deferred, stay inside the next run's window
+The diff surface therefore declares its share of the invocation ceiling explicitly: `DIFF_SUBREQUEST_BUDGET_PER_RUN` = 900 of the 1000 Cloudflare allows. The diffs cron is the surface's own invocation, so that ceiling is not shared with docs / wiki / issues / releases — but it *is* shared across every repo in `POLL_REPOS` and both phases, which is the sharing that produced the failure above. `diffFileBudgetPerPhase` divides the declared share by `repoCount × 2` and converts the remainder to files. Deriving it from the repo list rather than fixing a literal is deliberate: `POLL_REPOS` grows by ordinary config commits (issue #233 appended the sixth repo), and a literal sized against the list of the day overruns the ceiling on the next append, surfacing as failures on whichever repo the loop reaches last rather than on the change that caused them. A floor of 5 files holds under a repo list long enough to divide the budget below it — an overrun is retried by the next cron, whereas a budget of zero stalls every diff watermark permanently, which is the failure being removed.
+
+The file budget can split one commit across runs. Nothing is thinned: a run indexes a prefix of the files still outstanding, and the next run reads the already-indexed paths back from the structured store (`GET /diffs`) and continues from there. That read is what makes the split converge — without it a bounded run re-indexes the same leading files every cron tick and the commit never finishes. It is spent only on a commit whose file count exceeds the budget, so an ordinary commit costs nothing to find out, and a store error on it counts the commit as failed rather than as "nothing indexed yet", which would silently restart it.
+
+Both phases obey one watermark invariant: **a watermark never moves past a commit that has not been successfully ingested.** A commit counts as ingested only when its detail fetch succeeded and `processAndUpsertCommitDiff` reported zero failed files *and* no files left outstanding — embedding and Vectorize failures are reported by return value rather than by throwing, so the return value is part of the check. The criterion is the dense side only: a D1 FTS mirror or store-row failure is logged by the pipeline and deliberately not counted, since the vector has already landed and the sparse index reconciles on reindex; gating the watermark on the mirror would let an FTS-side outage stall the whole diff surface. Consequences:
+
+- a failed commit, a commit the file budget only partly indexed, and every commit either per-run cap deferred, all stay inside the next run's window. A partly-indexed commit is not an error — it is mid-split, and holding the watermark is what keeps the rest of its files reachable
 - a failed commit list leaves the watermark untouched, so the whole period stays retryable
 - the forward phase must know the oldest end of its window before advancing, so the window is enumerated with a 100-entry page; a full page means "not enumerable" and the window end is halved (at most 8 times) until one page covers it. If even then the window overflows, the run holds the watermark and logs — a visible stall is preferred over a silent gap.
 

docs/installation.ja.md

@@ -212,7 +212,8 @@ POST /admin/diff-watermark?repo=owner/repo&since=2026-07-06T00:00:00Z
 
 運用上の注意:
 
-- 追いつき速度は 1 repo あたり 1 run 5 commits（毎時 `:30` なので約 120 commits/日）。数週間分の欠損は数日かかる
+- 追いつき速度は 1 repo 1 run あたり 2 軸で決まる: commit 数 5 件と、`POLL_REPOS` の repo 数から導出される phase ごとの file 予算（6 repo なら各 18 file 前後）。毎時 `:30` なので、commit が小さい範囲では約 120 commits/日、file 数が多い commit では遅くなる——予算を超える commit は複数 run に分割され、最後の file が index されるまでその watermark は止まる。数週間分の欠損は数日かかる
+- worker log の `{repo} diffs: ... [processed=.., partial=.., failed=.., deferred=..]` 行は 3 者を区別する: `partial` は分割途中の commit、`deferred` はその run が到達しなかった commit で、いずれもエラーではない
 - 再走査中も新規 commit は影響を受けない（webhook 経路が即時 index する）
 - upsert は `(repo, commit_sha, file_path)` で idempotent なので、index 済み期間を再走査しても安全
 - 進捗は worker log の `{repo} diffs: forward [...]` 行、または該当期間を `type: "diff"` で検索して確認する

docs/installation.md

@@ -212,7 +212,8 @@ Authentication:
 
 Operational notes:
 
-- catch-up rate is 5 commits per repo per cron run (~120 commits/day at the hourly `:30` trigger), so a multi-week gap takes several days to drain
+- catch-up rate is bounded on two axes per repo per cron run: 5 commits, and a per-phase file budget derived from how many repos `POLL_REPOS` holds (~18 files each at six repos). At the hourly `:30` trigger that is ~120 commits/day where commits are small, and slower where they are file-heavy — a commit larger than the budget is split across runs and its watermark holds until the last of its files is indexed. A multi-week gap takes several days to drain
+- the `{repo} diffs: ... [processed=.., partial=.., failed=.., deferred=..]` log line separates the three: `partial` is a commit mid-split, `deferred` is one this run never reached, and neither is an error
 - new commits are unaffected while a replay is in flight: the webhook path indexes them in real time
 - upserts are idempotent on `(repo, commit_sha, file_path)`, so re-covering an already-indexed period is safe
 - verify progress with the `{repo} diffs: forward [...]` line in the worker logs, or by searching `type: "diff"` for the period

src/pipeline/embed-diff.test.ts

@@ -74,7 +74,14 @@ describe("embed-diff: the batch axis is the token budget, not the file count", (
 
     const result = await processAndUpsertCommitDiff(env, mkStore(), REPO, commit);
 
-    expect(result).toEqual({ embedded: 30, skipped: 0, failed: 0, batches: 1 });
+    expect(result).toEqual({
+      embedded: 30,
+      skipped: 0,
+      failed: 0,
+      batches: 1,
+      alreadyIndexed: 0,
+      deferred: 0,
+    });
     expect(aiCalls).toHaveLength(1);
     expect(aiCalls[0]).toHaveLength(30);
   });
@@ -146,7 +153,123 @@ describe("embed-diff: the batch axis is the token budget, not the file count", (
 
     const result = await processAndUpsertCommitDiff(env, mkStore(), REPO, commit);
 
-    expect(result).toEqual({ embedded: 1, skipped: 1, failed: 0, batches: 1 });
+    expect(result).toEqual({
+      embedded: 1,
+      skipped: 1,
+      failed: 0,
+      batches: 1,
+      alreadyIndexed: 0,
+      deferred: 0,
+    });
     expect(aiCalls[0]).toHaveLength(1);
   });
 });
+
+describe("embed-diff: a file-heavy commit is split across runs, not thinned", () => {
+  it("indexes up to maxFiles and reports the rest as deferred", async () => {
+    const { env, aiCalls, upsertedIds } = mkEnv();
+    const commit = mkCommit(
+      Array.from({ length: 44 }, (_, i) => `@@ -1 +1 @@\n+line ${i}`),
+    );
+
+    const result = await processAndUpsertCommitDiff(env, mkStore(), REPO, commit, {
+      maxFiles: 18,
+    });
+
+    expect(result.embedded).toBe(18);
+    expect(result.deferred).toBe(26);
+    expect(result.failed).toBe(0);
+    // The bound is on files indexed, not on files seen: nothing past it is touched.
+    expect(aiCalls.flat()).toHaveLength(18);
+    expect(upsertedIds.flat()).toHaveLength(18);
+  });
+
+  it("takes the leading files first so the split has a stable order", async () => {
+    const { env, aiCalls } = mkEnv();
+    const commit = mkCommit(Array.from({ length: 5 }, (_, i) => `patch ${i}`));
+
+    await processAndUpsertCommitDiff(env, mkStore(), REPO, commit, { maxFiles: 2 });
+
+    // Input format is "{message}\n\n{path}\n\n{patch}".
+    const paths = aiCalls.flat().map((text) => text.split("\n\n")[1]);
+    expect(paths).toEqual(["src/file-0.ts", "src/file-1.ts"]);
+  });
+
+  it("resumes past the files a previous run already indexed", async () => {
+    const { env, aiCalls } = mkEnv();
+    const commit = mkCommit(Array.from({ length: 5 }, (_, i) => `patch ${i}`));
+
+    const result = await processAndUpsertCommitDiff(env, mkStore(), REPO, commit, {
+      maxFiles: 2,
+      indexedFilePaths: new Set(["src/file-0.ts", "src/file-1.ts"]),
+    });
+
+    const paths = aiCalls.flat().map((text) => text.split("\n\n")[1]);
+    expect(paths).toEqual(["src/file-2.ts", "src/file-3.ts"]);
+    expect(result.alreadyIndexed).toBe(2);
+    expect(result.embedded).toBe(2);
+    expect(result.deferred).toBe(1);
+  });
+
+  it("closes the split — the last run reports nothing deferred", async () => {
+    // Repeated bounded calls, each fed the paths the previous ones landed, must
+    // reach every file exactly once. A resume set that failed to shrink the pending
+    // list would loop on the same prefix and never report deferred=0.
+    const { env } = mkEnv();
+    const commit = mkCommit(Array.from({ length: 7 }, (_, i) => `patch ${i}`));
+    const indexed = new Set<string>();
+    let runs = 0;
+    let last!: Awaited<ReturnType<typeof processAndUpsertCommitDiff>>;
+
+    do {
+      if (runs++ > 10) throw new Error("split did not converge");
+      last = await processAndUpsertCommitDiff(env, mkStore(), REPO, commit, {
+        maxFiles: 3,
+        indexedFilePaths: new Set(indexed),
+      });
+      for (let i = 0; i < last.embedded; i++) {
+        indexed.add(`src/file-${indexed.size}.ts`);
+      }
+    } while (last.deferred > 0);
+
+    expect(runs).toBe(3);
+    expect(indexed.size).toBe(7);
+    expect(last.deferred).toBe(0);
+  });
+
+  it("leaves the unbounded call untouched (the webhook path)", async () => {
+    const { env, aiCalls } = mkEnv();
+    const commit = mkCommit(Array.from({ length: 12 }, (_, i) => `patch ${i}`));
+
+    const result = await processAndUpsertCommitDiff(env, mkStore(), REPO, commit);
+
+    expect(result.embedded).toBe(12);
+    expect(result.deferred).toBe(0);
+    expect(result.alreadyIndexed).toBe(0);
+    expect(aiCalls.flat()).toHaveLength(12);
+  });
+
+  it("reports a fully-indexed commit as done rather than re-embedding it", async () => {
+    const { env, aiCalls } = mkEnv();
+    const commit = mkCommit(Array.from({ length: 3 }, (_, i) => `patch ${i}`));
+
+    const result = await processAndUpsertCommitDiff(env, mkStore(), REPO, commit, {
+      maxFiles: 2,
+      indexedFilePaths: new Set([
+        "src/file-0.ts",
+        "src/file-1.ts",
+        "src/file-2.ts",
+      ]),
+    });
+
+    expect(result).toEqual({
+      embedded: 0,
+      skipped: 0,
+      failed: 0,
+      batches: 0,
+      alreadyIndexed: 3,
+      deferred: 0,
+    });
+    expect(aiCalls).toHaveLength(0);
+  });
+});