上限に当たったら断る。黙って粗くも、黙って切りもしない

区画の天井に当たったときの振る舞いを、切り落としからも「目安を勝手に上げる」からも
やめて、理由と必要な値を添えて断る形にした。

数で打ち切ると、打ち切った先の母集団がどの区画にも入らず、どの巡回にも回ってこない
（68 万件を目安 150 で割ろうとして 2,000 区画で止まり、40 万件ぶんの範囲が台帳から
消えた）。目安を勝手に上げるのも同じ筋で、頼んだ細かさと違うもので回り続ける。
どちらを選ぶかは人が決めること。

タグで割る道も多い順に切り落としていた。消えるのはいつも細かい分類のほうで、
そこにこそ漏れが溜まる。

そのうえで天井を 2,000 → 20,000 に上げる。目安 150 なら 300 万件まで割れるので、
普通の使い方では当たらない。実測: 68 万件・目安 150 → 8,192 区画・1 区画 83〜84・
合計は母集団と一致・台帳 1.24 MB。

天井を上げられるのは、定義を 1 収集 1 件に分けたから。まとめて 1 つの JSON に
入れていると、どれか 1 つを直すだけで全部を読み書きすることになり、1 つが壊れれば
全部が読めなくなり、1 つが太れば全部が重くなる。名指しの取得は 1 件しか読まない。
まとめて入れていた頃のものは、読むときに分ける（分け終えてから古い行を消すので、
途中で落ちても次に読むときにやり直せる）。

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

Author: claude

CLAUDE.md

@@ -613,13 +613,17 @@ GeoNames 全世界地名辞典 = `geonames`(いずれも 348 言語版・195 か
         動かすのは区画の印だけで、**進み具合も中身も触らない**(やり直したいのは
         見る仕事であって、集めたものではない)。**走らせない** —— 印を外すだけなので、
         直したい設定を入れる前に 1 回消費しない。**区画を持つ収集にだけ出す**
-      - **区画の数が天井(`MAX_PARTITIONS`)に当たったら、削るのは細かさであって
-        範囲ではない**(`partition._fits`)。打ち切ると、**打ち切った先の母集団が
-        どの区画にも入らず、どの巡回にも回ってこない** —— 本番で 68 万件を目安 150 で
-        割ろうとして 2,000 区画(1 区画 200 未満)で止まり、40 万件ぶんの範囲が
-        台帳から消えた。目安を上げれば 1 回に見る量が増えるだけで済む。
-        **余裕は 2 倍で見る** —— 二分は割り切れないので、できる区画は目安の半分から
-        等倍に散らばる。実測: 68 万件・目安 150 → 1,024 区画・1 区画 670・合計は一致
+      - **区画の数が天井(`MAX_PARTITIONS`)に当たったら断る**(`partition._must_fit`)。
+        **黙って減らさない** —— 数で打ち切ると、打ち切った先の母集団がどの区画にも
+        入らず、どの巡回にも回ってこない。本番で 68 万件を目安 150 で割ろうとして
+        2,000 区画(1 区画 200 未満)で止まり、40 万件ぶんの範囲が台帳から消えた。
+        **目安を勝手に上げるのも同じ筋で駄目**で、頼んだ細かさと違うもので回り続ける
+        —— 目安を上げるか母集団を絞るかは人が決めること(理由と必要な値を添えて断る)。
+        **余裕は 2 倍で見る**(二分は割り切れず、区画は目安の半分から等倍に散らばる)。
+        タグで割る道も同じで、**多い順に切り落とさない** —— 消えるのはいつも
+        細かい分類のほうで、そこにこそ漏れが溜まる。
+        実測: 68 万件・目安 150 → 8,192 区画・1 区画 83〜84・合計は母集団と一致・
+        台帳 1.24 MB
       - **区画の母集団を足しても長期記憶の数には届かない**(画面に差を出す)。
         消えたものは数えないし、**座標を持たない 1 件はどの区画にも入らない**
         (地図で割る収集)。どちらも仕様だが、出ていないと「区画分けが壊れている」
@@ -1159,6 +1163,14 @@ GeoNames 全世界地名辞典 = `geonames`(いずれも 348 言語版・195 か
       黙って飛ばされる
     - **失敗しても次回の予定を入れる**(`record_result`)。入れないと、一度こけた収集が
       二度と走らなくなる。理由は定義側に控えるので、画面の「前回」の欄から読める
+    - **定義は 1 収集 = 1 件**(`collect/<名前>`。`app/machine_store.py`)。
+      **まとめて 1 つの JSON に入れない** —— 区画の台帳は収集 1 つで MB 単位になるので、
+      どれか 1 つを直すだけで全部を読み書きすることになる。**1 つが壊れれば全部が
+      読めなくなり、1 つが太れば全部が重くなる**。`get` は名指しの 1 件しか読まない。
+      **並びは作った順**(1 件ずつ置くと配列の順を持てないので `created_at` で並べ直す)。
+      **まとめて 1 件に入れていた頃のものは、読むときに分ける**
+      (`_split_out_the_old_row`。分け終えてから古い行を消すので、途中で落ちても
+      次に読むときにやり直せる)
     - **画面の「削除」は設定と焼いたものをまとめて消す**(`admin_collect_delete`)。
       設定だけ消して中身が残ると、一覧から消えたのに検索には出続けるものができ、
       **どこから来たのかも読めなくなる**(収集の定義が消えているため)。

app/collect.py

@@ -941,7 +941,10 @@ def normalize_focus(raw) -> Focus | None:
 
 
 def _stored() -> str | None:
-    """いまの定義(JSON の文字列)。まだ 1 度も置いていなければ None。"""
+    """**まとめて 1 件に入れていた頃**の定義(JSON の文字列)。無ければ None。
+
+    いまは 1 収集 = 1 件(`collect/<名前>`)。ここを読むのは移行のためだけ。
+    """
     return machine_store.get(DEFS_KIND, DEFS_KEY)
 
 
@@ -1115,47 +1118,101 @@ def edits_what_is_there(prompt: str, only_new: bool = False) -> bool:
     return any(p in (prompt or "") for p in (MATERIAL_PLACEHOLDER, RECENT_PLACEHOLDER))
 
 
-def _to_json(items: list[Collection]) -> str:
-    return json.dumps(
-        {"collections": [c.__dict__ for c in items]},
-        ensure_ascii=False,
-        indent=2,
-    )
-
-
 def load() -> list[Collection]:
-    """定義の一覧(並びは配列の順)。
+    """定義の一覧(並びは作った順)。**1 収集 = 1 件**(`collect/<名前>`)。
+
+    **まとめて 1 件に入れない。** 区画の台帳は収集 1 つで MB 単位になりうるので、
+    全部を 1 つの JSON に入れると、**どれか 1 つを直すだけで全部を読み書きする**
+    ことになる —— 1 つが壊れれば全部が読めなくなり、1 つが太れば全部が重くなる。
+    (**上限を置いて逃げない**ための前提でもある。台帳の天井は、この形にして
+    初めて「1 つの収集の都合」に閉じられる。)
 
     **本文が壊れていたら黙って作り直さない** —— 中身ごと消えるので、読めないことを
-    見せて人に直させる(プロジェクトと同じ判断)。
+    見せて人に直させる(プロジェクトと同じ判断)。**壊れているのはその 1 件だけ**
+    だと分かるように、名前を添えて上げる。
+    """
+    _split_out_the_old_row()
+    out = []
+    for key in machine_store.keys(DEFS_KIND):
+        if key == DEFS_KEY:
+            continue
+        body = machine_store.get(DEFS_KIND, key)
+        if body is None:
+            continue
+        try:
+            raw = json.loads(body)
+            if not isinstance(raw, dict):
+                raise TypeError("object を入れてください")
+        except (ValueError, TypeError) as e:
+            raise HTTPException(400, {"error": f"{DEFS_BROKEN}(「{key}」): {e}"}) from None
+        out.append(_from_json(raw))
+    # **並びは作った順**。1 件ずつ置くと配列の順を持てないので、作った時刻で並べ直す
+    return sorted(out, key=lambda c: (c.created_at, c.name))
+
+
+def _split_out_the_old_row() -> None:
+    """まとめて 1 件に入れていた頃のものを、1 収集 1 件へ分ける。**1 度だけ**。
+
+    **古い行が残っているあいだは、そちらが正**(分け終えてから消す)——
+    途中で落ちても、次に読むときにもう一度分け直せる。
     """
     body = _stored()
     if body is None:
-        return []
+        return
     try:
-        payload = json.loads(body or "{}")
-        raw = payload["collections"]
+        raw = json.loads(body or "{}")["collections"]
         if not isinstance(raw, list):
-            raise ValueError("collections must be a list")
+            raise TypeError("collections must be a list")
     except (ValueError, KeyError, TypeError) as e:
         raise HTTPException(400, {"error": f"{DEFS_BROKEN}: {e}"}) from None
-    return [_from_json(i) for i in raw if isinstance(i, dict)]
+    for item in raw:
+        if isinstance(item, dict) and (name := str(item.get("name") or "")):
+            machine_store.put(DEFS_KIND, name, json.dumps(item, ensure_ascii=False, indent=2))
+    machine_store.drop(DEFS_KIND, DEFS_KEY)
+    log.info("split %d collection definitions into one record each", len(raw))
 
 
 def save(items: list[Collection]) -> None:
-    machine_store.put(DEFS_KIND, DEFS_KEY, _to_json(items))
+    """一覧をそのまま書き込む。**消えたものはこの場で落とす**。
+
+    1 件だけ直すなら `_replace_one` のほうが安い(そちらは 1 件しか書かない)。
+    """
+    _split_out_the_old_row()
+    keep = {c.name for c in items}
+    for item in items:
+        _put_one(item)
+    for key in machine_store.keys(DEFS_KIND):
+        if key != DEFS_KEY and key not in keep:
+            machine_store.drop(DEFS_KIND, key)
+
+
+def _put_one(item: Collection) -> None:
+    machine_store.put(
+        DEFS_KIND, item.name, json.dumps(item.__dict__, ensure_ascii=False, indent=2),
+    )
 
 
 def get(name: str) -> Collection:
-    for item in load():
-        if item.name == name:
-            return item
-    raise HTTPException(404, {"error": f"収集「{name}」がありません"})
+    """名前で 1 つ。**一覧を読まない** —— 台帳を抱えた他の収集まで読む理由が無い。"""
+    _split_out_the_old_row()
+    body = machine_store.get(DEFS_KIND, name) if name and name != DEFS_KEY else None
+    if body is None:
+        raise HTTPException(404, {"error": f"収集「{name}」がありません"})
+    try:
+        raw = json.loads(body)
+        if not isinstance(raw, dict):
+            raise TypeError("object を入れてください")
+    except (ValueError, TypeError) as e:
+        raise HTTPException(400, {"error": f"{DEFS_BROKEN}(「{name}」): {e}"}) from None
+    return _from_json(raw)
 
 
 def _replace_one(name: str, updated: Collection) -> None:
-    items = load()
-    save([updated if c.name == name else c for c in items])
+    """その 1 件だけを書き換える。**他の収集は読みも書きもしない**。"""
+    get(name)
+    if updated.name != name:
+        machine_store.drop(DEFS_KIND, name)
+    _put_one(updated)
 
 
 def create(
@@ -1317,10 +1374,8 @@ def remove(name: str) -> None:
     同じ名前で作り直すのは普通に起きる —— 残しておくと、前の収集が足した・消した
     ものが新しい収集の履歴に混ざって見える。
     """
-    items = load()
-    if not any(c.name == name for c in items):
-        raise HTTPException(404, {"error": f"収集「{name}」がありません"})
-    save([c for c in items if c.name != name])
+    get(name)
+    machine_store.drop(DEFS_KIND, name)
     collect_log.forget(name)
     # **待ち行列からも外す** —— 消えた収集を抱えたままだと、そのワーカーは
     # 起こそうとして 404 を踏み続ける

app/machine_store.py

@@ -214,6 +214,24 @@ def drop(kind: str, key: str) -> bool:
     return True
 
 
+def keys(kind: str) -> list[str]:
+    """その種類に置いてあるものの名前(見出しの `<種類>/` の後ろ)。
+
+    **1 件ずつ置くものを数え上げるのに要る。** まとめて 1 件に入れていたものを
+    種類の下へ分けると、一覧は「何が置いてあるか」を先に引いてから中身を取る形になる
+    —— `records()` は全部の種類を並べるので、種類で絞れるほうが読む側も安い。
+    """
+    if not is_enabled():
+        return []
+    ensure_db()
+    head = kind + KEY_SEP
+    with _connect() as conn:
+        rows = conn.execute(
+            "SELECT title FROM docs WHERE title LIKE ? ORDER BY title", (head + "%",)
+        ).fetchall()
+    return [r["title"][len(head):] for r in rows if r["title"].startswith(head)]
+
+
 def records() -> list[dict]:
     """置いてあるもの全部の**目録**(中身は運ばない)。