動かした文書に、どの回が動かしたかを脇書きとして残す

変更履歴(app/collect_log.py)は回ごとに 1 行で、動いた見出しは頭の 20 件まで。
しかも回ごとの間隔は桁違いなので、短い回の行が長い回の行を押し流す ——
「その回が何を直したのか」は、たいてい流れた後にしか引きに来られない。
文書の側に印を押しておけば、いま手元にあるものについては回の間隔と関係なく読める。

置くのは脇書き。全文検索が索引するのは見出しと本文だけ、AI へ差し込む一覧に
載るのは配信日と出典だけなので、ここへ入れたものはどちらにも出てこない。
本文やタグへ書くと、読む人にも AI にも文書の中身として見える。

残すのは 1 回分だけで、後の回が同じ 1 件に触れば印はそちらへ移る。溜めると
1 件ごとに際限なく伸び、焼き直すたびに全件がそのぶん重くなる。いつ動いたかは
updated_at が既に持っているので二重に置かない。

鍵の数の天井より後に押す。あの天井は AI が運ぶ事実が際限なく増えないための
もので、こちらの印まで落とすと、脇書きの多い 1 件だけ印が付かない。

引くための索引は集める層のアダプタにだけ足す。式索引なので列を足さずに済み、
SCHEMA_VERSION は動かない —— 動かすとダンプ由来のソースの焼き直しまで要求する。

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

Author: claude

CLAUDE.md

@@ -669,6 +669,21 @@ GeoNames 全世界地名辞典 = `geonames`(いずれも 348 言語版・195 か
       —— 1 時間ごとに見出しを溜める回が 24 行を占めれば、4 時間ごとに整理する回の
       差分は 1 日ぶんも残らない。**選択肢は控えの側から引く**(`collect_log.sweeps`)
       —— 定義の巡回から作ると、名前を変えた回や消した回の行が目の前に出ているのに選べない
+    - **動かした 1 件には、どの回が動かしたかを脇書きに残す**(`collect._stamped` /
+      `extra.changed_by` / `extra.change`)。**変更履歴だけでは読めない問いがある** ——
+      控えは回ごとに 1 行で、動いた見出しは頭の 20 件までしか残らず、しかも短い回が
+      長い回を押し流す。文書の側に押しておけば、**いま手元にあるものについては
+      回の間隔と関係なく読める**(画面の「その回が動かしたもの」/ `collect.changed_here`)。
+      **残すのは 1 回分だけ**(上書きする)—— 溜めると 1 件ごとに際限なく伸び、
+      焼き直すたびに全件がそのぶん重くなる。「いつ」は `updated_at` が既に持っている。
+      **置くのは脇書き** —— 全文検索が索引するのは見出しと本文だけ、AI へ差し込む
+      一覧に載るのは配信日と出典だけなので、ここへ入れたものはどちらにも出てこない
+      (本文やタグへ書くと、読む人にも AI にも「文書の中身」として見える)。
+      **鍵の数の天井(`MAX_EXTRA_KEYS`)より後に押す** —— あの天井は AI が運ぶ事実が
+      際限なく増えないためのもので、こちらの印まで落とすと、脇書きの多い 1 件だけ
+      印が付かない(いちばん読みたい 1 件がそれになる)。**引くための索引は集める層だけ**
+      (`ingest/core.py` の `CHANGED_BY_INDEX_DDL`)—— 式索引なので列を足さずに済み、
+      `SCHEMA_VERSION` は動かない(動かすと jawiki の焼き直しまで要求することになる)
     - **直した件数は足した件数と別に控える**(`last_updated`)。整理の回は足すものが
       無くても大量に直っていることがあり、追加だけ見ていると「何もしなかった」ように読める
       (`material` は数えていたのに、定義へ書き戻すところで落としていた)

app/collect.py

@@ -2110,6 +2110,40 @@ def recent(name: str, sources: dict, limit: int = 5) -> list[dict]:
     ]
 
 
+def changed_here(name: str, sources: dict, sweep: str, limit: int = 100) -> list[dict]:
+    """その回が最後に動かしたものを、新しい順に(`_stamped` の印で引く)。
+
+    **変更履歴(`app/collect_log.py`)では答えられない問いのほう。** あちらは回ごとに
+    1 行で、動いた見出しは頭の 20 件までしか残らず、しかも回ごとの間隔は桁違いなので
+    短い回の行が長い回の行を押し流す。ここは**いま手元にあるものを文書の側から**引くので、
+    その回が何回前に走っていようが読める。
+
+    **読めるのは最後に動かした回だけ。** 後の回が同じ 1 件に触れば印はそちらに移る
+    (印は 1 回分しか持たない)—— 「整理が直したあと見出しが触った」ものは
+    見出しのほうに出る。
+    """
+    src = sources.get(name)
+    if src is None or not sweep or limit <= 0:
+        return []
+    rows = db.query(
+        src.path,
+        "SELECT title, opening, tags, updated_at, extra FROM docs"
+        f" WHERE json_extract(extra, '$.{CHANGED_BY_KEY}') = ?"
+        " ORDER BY updated_at DESC LIMIT ?",
+        (sweep, limit),
+    )
+    return [
+        {
+            "title": row["title"],
+            "opening": row["opening"],
+            "tags": load_tags(row["tags"]),
+            "updated_at": row["updated_at"],
+            "extra": load_json(row["extra"]),
+        }
+        for row in rows
+    ]
+
+
 def doc_versions(name: str, sources: dict, title: str) -> dict:
     """1 件の見出しについて、**いまの世代と 1 つ前の世代**の中身を返す。
 
@@ -2375,6 +2409,7 @@ def stream_docs(
     only_new: bool = False,
     edits: bool = False,
     diff: dict | None = None,
+    sweep: str = "",
 ):
     """焼く素材を 1 件ずつ返す。`material` の中身で、**丸ごとは持たない**。
 
@@ -2387,6 +2422,8 @@ def stream_docs(
     同じ並びになる(新しいぶんは採番の順で後ろに付く)。
 
     `diff` を渡すと、数えた結果をそこへ書く(戻り値にできないため)。
+
+    **動かした 1 件には、どの回が動かしたかを脇書きに残す**(`_stamped`)。
     """
     counts = diff if diff is not None else {}
     now = _iso(_now())
@@ -2419,7 +2456,7 @@ def stream_docs(
             # 墓標。**消さずに印を付けて残す**
             removed_titles.append(title)
             updated += 1
-            yield _buried(before, raw, now)
+            yield _stamped(_buried(before, raw, now), sweep, "removed")
             continue
         doc = _to_doc(raw, now, item.web)
         if doc is None:
@@ -2436,7 +2473,9 @@ def stream_docs(
         else:
             # 集めるほうで同じ見出しが来るのは「もう持っている」の意味
             skipped += 1
-        yield {**doc, "doc_id": before["doc_id"]}
+            yield {**doc, "doc_id": before["doc_id"]}
+            continue
+        yield _stamped({**doc, "doc_id": before["doc_id"]}, sweep, "updated")
 
     next_id += 1
     for raw in edits_of.rest():
@@ -2455,7 +2494,7 @@ def stream_docs(
         doc["extra"] = _merge_extra({}, doc["extra"])
         added += 1
         added_titles.append(doc["title"])
-        yield {**doc, "doc_id": next_id}
+        yield _stamped({**doc, "doc_id": next_id}, sweep, "added")
         next_id += 1
 
     counts.update({
@@ -2479,6 +2518,7 @@ def material(
     collected: list[dict],
     only_new: bool = False,
     edits: bool = False,
+    sweep: str = "",
 ) -> tuple[list[dict], dict]:
     """焼く素材と、前世代との差分を組み立てる。
 
@@ -2514,7 +2554,7 @@ def material(
     """
     counts: dict = {}
     rows = sorted(previous.values(), key=lambda d: d["doc_id"])
-    docs = list(stream_docs(item, rows, collected, only_new, edits, counts))
+    docs = list(stream_docs(item, rows, collected, only_new, edits, counts, sweep))
     return docs, counts
 
 
@@ -2618,6 +2658,50 @@ def _buried(doc: dict, raw: dict, now: str) -> dict:
     }
 
 
+CHANGE_ADDED = "added"
+CHANGE_UPDATED = "updated"
+CHANGE_REMOVED = "removed"
+
+# 脇書きに残す「最後に動かした回」の鍵。
+CHANGED_BY_KEY = "changed_by"
+CHANGE_KEY = "change"
+
+
+def _stamped(doc: dict, sweep: str, change: str) -> dict:
+    """動かした 1 件に、**どの回が・どう動かしたか**を脇書きとして押す。
+
+    **変更履歴を別に持つだけでは、読みたいほうが読めない。** 控え
+    (`app/collect_log.py`)は回ごとに 1 行で、動いた見出しは頭の 20 件しか
+    残らない。しかも回ごとの間隔は桁違いなので、短い回の行が長い回の行を押し流す
+    —— 「整理が何を直したのか」を引きに来ると、たいてい流れた後になる。
+    文書の側に押しておけば、**いま手元にあるものについては回の間隔と関係なく読める**。
+
+    **残すのは 1 回分だけ**(上書きする)。履歴を溜める場所ではない ——
+    溜めると 1 件ごとに際限なく伸び、焼き直すたびに全件がその分だけ重くなる。
+    「いつ」は既にある `updated_at` 列が持っているので、ここには持たない。
+
+    **脇書きに置くのは、検索と本文精査の邪魔をしないため。** 全文検索が索引するのは
+    見出しと本文だけ(`ingest/core.py` の `docs_fts`)、AI へ差し込む一覧に載るのも
+    配信日と出典だけ(`_current_line`)なので、ここへ入れたものはどちらにも出てこない。
+    本文やタグへ書くと、読む人にも AI にも「文書の中身」として見える。
+
+    **鍵の数の天井(`MAX_EXTRA_KEYS`)より後に押す。** あの天井は AI が運んでくる
+    事実が際限なく増えないためのもので、こちらが押す印まで落とすと、
+    脇書きの多い 1 件だけ印が付かないことになる(いちばん読みたい 1 件がそれになる)。
+    """
+    extra = doc.get("extra") if isinstance(doc.get("extra"), dict) else {}
+    stamp = {CHANGE_KEY: change}
+    if sweep:
+        stamp[CHANGED_BY_KEY] = sweep
+    return {**doc, "extra": {**extra, **stamp}}
+
+
+def changed_by(doc: dict) -> str:
+    """その 1 件を最後に動かした回。持っていなければ空。"""
+    extra = doc.get("extra")
+    return str((extra or {}).get(CHANGED_BY_KEY) or "") if isinstance(extra, dict) else ""
+
+
 def removed_reason(doc: dict) -> str:
     """その 1 件を消した理由。持っていなければ空。"""
     extra = doc.get("extra")
@@ -2912,8 +2996,12 @@ def _count_dropped(item, plan, previous, collected, only_new, edits) -> int:
 
 
 def bake_lines(item, sources: dict, previous, collected, only_new=False, edits=False,
-               survey: dict | None = None):
-    """焼く素材を 1 行ずつ返す。**2 周目**(数えるのは `bake_survey`)。"""
+               survey: dict | None = None, sweep: str = ""):
+    """焼く素材を 1 行ずつ返す。**2 周目**(数えるのは `bake_survey`)。
+
+    **どの回が焼いたかはここで渡す。** 押すのは実際に焼かれる素材のほうで、
+    数える 1 周目ではない(あちらは件数しか使わない)。
+    """
     plan = survey or bake_survey(item, sources, previous, collected, only_new, edits)
     rows = _rows_of(previous)
     limit = _expiry_limit(item)
@@ -2926,7 +3014,7 @@ def bake_lines(item, sources: dict, previous, collected, only_new=False, edits=F
         }
     }, ensure_ascii=False)
 
-    for doc in stream_docs(item, rows(), collected, only_new, edits):
+    for doc in stream_docs(item, rows(), collected, only_new, edits, None, sweep):
         if limit is not None and _doc_time(doc) < limit:
             continue
         for n, rule in enumerate(plan["rules"]):

app/main.py

@@ -650,7 +650,7 @@ def flow():
         try:
             yield from collect.bake_lines(
                 baked_as, sources, previous, items,
-                focus is None and sweep.only_new, edits, plan,
+                focus is None and sweep.only_new, edits, plan, label,
             )
         finally:
             if hasattr(items, "close"):

app/views/admin.py

@@ -672,6 +672,10 @@ def _sweep_fields(sweep, removable: bool, shared_prompt: str = "") -> str:
 # 設定の行はこれを全部つないで 1 つのセルにする
 SWEEP_COLUMNS = 7
 
+# 「その回が動かしたもの」に出す件数。**全部は出さない** —— 1 回で数千件動く回が
+# あるので、天井を置かないと画面が開かなくなる。読みたいのは新しいほうから。
+CHANGED_HERE_LIMIT = 200
+
 
 def _sweep_backend_fields(sweep, backend, mechanical: bool) -> str:
     """頼む相手の欄。**機械で引く巡回には出さない**。
@@ -1020,6 +1024,63 @@ def _collect_running_html(name: str | None = None) -> str:
     )
 
 
+CHANGE_LABELS = {
+    collect.CHANGE_ADDED: "足した",
+    collect.CHANGE_UPDATED: "直した",
+    collect.CHANGE_REMOVED: "消した",
+}
+
+
+def _changed_here_html(name: str, sources: dict, sweep: str | None) -> str:
+    """その回が動かしたもの(`collect.changed_here`)。**回を選んだときだけ出す**。
+
+    **「直近の変更」では届かないところを埋める。** あちらは回ごとに 1 行で、
+    動いた見出しは頭の 20 件までしか残らない —— 1 回で数千件動く回では、
+    何が動いたのかがほとんど読めない。ここは焼いたものを文書の側から引くので、
+    その回が何回前に走っていようが、動かした全部が出る。
+
+    **出るのは「最後に動かした回」だけ。** 後の回が同じ 1 件に触れば、その 1 件は
+    そちらの一覧へ移る —— 印を 1 回分しか持たないため。書いておかないと、
+    整理が直したはずのものが見当たらない理由が読めない。
+    """
+    if not sweep:
+        return ""
+    docs = collect.changed_here(name, sources, sweep, limit=CHANGED_HERE_LIMIT)
+    head = f"<summary>「{esc(sweep)}」が動かしたもの</summary>"
+    if not docs:
+        return (
+            f'<details id="changed">{head}'
+            '<p class="muted">いま手元にあるもののうち、この回が最後に動かしたものは'
+            "ありません。</p></details>"
+        )
+    rows = []
+    for doc in docs:
+        at = jst.parse(doc["updated_at"] or "")
+        change = (doc.get("extra") or {}).get(collect.CHANGE_KEY) or ""
+        mark = CHANGE_LABELS.get(str(change), "")
+        rows.append(
+            f"<tr><td>{esc(jst.format(at)) if at else ''}</td>"
+            f'<td>{esc(mark) if mark else "<span class=\"muted\">—</span>"}</td>'
+            f"<td>{_doc_link(name, doc['title'])}"
+            f'<br><span class="muted">{esc((doc["opening"] or "")[:120])}</span></td></tr>'
+        )
+    more = (
+        f'<p class="muted">新しい順に {CHANGED_HERE_LIMIT} 件まで。</p>'
+        if len(docs) >= CHANGED_HERE_LIMIT else ""
+    )
+    return f"""
+<details id="changed" open>{head}
+<table>
+<thead><tr><th>いつ</th><th>何を</th><th>見出し</th></tr></thead>
+<tbody>{"".join(rows)}</tbody>
+</table>
+{more}
+<p class="muted">印は 1 件につき 1 回分だけなので、<strong>後の回が同じ見出しに触れば
+そちらへ移る</strong>。消えたものもここに出る(読み口からは返らないが、手元には残っている)。</p>
+</details>
+"""
+
+
 def _changes_filter_html(name: str | None, sweep: str | None) -> str:
     """「どの回を見るか」の切り替え。**回が 1 本しかなければ出さない**(選ぶ先が無い)。
 
@@ -2924,6 +2985,7 @@ def admin_collect_detail(
 {_collect_detail_html(item, disabled, request.app.state.sources)}
 {_collect_running_html(name)}
 {_collect_changes_html(name=name, sweep=sweep)}
+{_changed_here_html(name, request.app.state.sources, sweep)}
 <p class="muted"><a href="/admin/memory#collect">集める の一覧へ戻る</a></p>
 """
     return HTMLResponse(content=page_shell(name, body))

ingest/core.py

@@ -135,6 +135,18 @@
 CREATE INDEX idx_docs_updated ON docs(updated_at DESC, doc_id DESC);
 """
 
+# 「どの回が最後に動かしたか」で引くための索引(`app/collect.py` の `_stamped`)。
+# **これもコアには入れない** —— 押しているのは集める層だけで、ダンプ由来のソースの
+# 脇書きにこの鍵は無い。式索引なので列を足さずに済み、`SCHEMA_VERSION` は動かない。
+#
+# updated_at まで入れているのは、並び順まで索引で満たすため。無いと「その回が
+# 動かしたもの」を全部読んでから並べ替えることになり、**回の間隔が短い収集ほど
+# 重くなる**(1 回で数千件動く網羅の収集がそれに当たる)。
+CHANGED_BY_INDEX_DDL = """
+CREATE INDEX idx_docs_changed_by
+    ON docs(json_extract(extra, '$.changed_by'), updated_at DESC);
+"""
+
 # docs 投入後に doc_tags を組み立てる SQL(索引を張る前に流す)。
 # docs から作り直す方式にしているのは、docs 側が INSERT OR REPLACE を使う(同じ doc_id が
 # 二度来ても最後の 1 件が残る)ため。行ごとに append すると置き換えられた古いタグが残る。

ingest/sources/collect.py

@@ -24,7 +24,7 @@
 import os
 from pathlib import Path
 
-from core import RECENCY_INDEX_DDL, SourceAdapter
+from core import CHANGED_BY_INDEX_DDL, RECENCY_INDEX_DDL, SourceAdapter
 from sources.remote import PluginError, RemotePluginAdapter, RemoteSource, _get_json
 
 log = logging.getLogger("chiezo.ingest")
@@ -54,8 +54,11 @@ class CollectAdapter(RemotePluginAdapter):
     """
 
     # 集めたものは**溜まっていく**ので、新しい順に引けるようにする。
-    # 「この 1 日で何が入ったか」は、この層でいちばん訊かれる問い
-    extra_index_ddl = RECENCY_INDEX_DDL
+    # 「この 1 日で何が入ったか」は、この層でいちばん訊かれる問い。
+    # **どの回が動かしたか**でも引く —— 変更履歴(`app/collect_log.py`)は回ごとに
+    # 1 行で、動いた見出しは頭の 20 件しか残らないうえ、短い回の行が長い回の行を
+    # 押し流す。文書の側の印なら、回の間隔と関係なく引ける
+    extra_index_ddl = RECENCY_INDEX_DDL + CHANGED_BY_INDEX_DDL
 
     def __init__(self, src: RemoteSource) -> None:
         super().__init__(src)