From eebe8ffca8956010bc3e31aef56c1992c891f704 Mon Sep 17 00:00:00 2001 From: senna-lang Date: Mon, 7 Sep 2026 08:59:23 +0900 Subject: [PATCH 1/4] fix(indexer): harden OpenCode ingestion against corrupt rows and cursor drift (#21) Four independent robustness bugs in the OpenCode (SQLite) ingestion path, all in indexer.py: - _load_opencode_raw_entries: json.loads(row["data"]) and _epoch_ms_to_iso(row["time_created"]) ran unguarded per row. One corrupt row (bad JSON, NULL time_created) raised and aborted ingestion of the entire DB (all projects/sessions). Extracted _parse_opencode_row(), which try/excepts (TypeError, ValueError, OverflowError, json.JSONDecodeError) per row and returns None for a malformed row; the row is skipped instead of poisoning the batch. - index_opencode_db: os.path.realpath(row["worktree"]) crashed with TypeError when a project's worktree column was NULL. Now skips rows with worktree is None before calling realpath. - index_opencode_db: the "already ingested" filter compared ex.ply_start (an index into the (time_created, id)-sorted message/part list, rebuilt fresh every run) against a stored last_ply_end. A new message arriving with an earlier time_created than already-processed messages shifts every later index, which (a) changed source_turn_id for old exchanges, re-emitting them under a new exchange_id, while (b) genuinely new exchanges could collide with a stale exchange_id and get silently dropped. Replaced the position-based source_turn_id (str(ply_start)) with the exchange's own stable id (sha256 of conversation_id + the OpenCode user message id, already computed by parse_opencode_exchanges and independent of list position), and replaced the last_ply_end lookup with a direct query of already-persisted source_turn_ids for the session. Position drift can no longer duplicate or drop turns. - index_opencode_db: sqlite3.connect(f"file:{path}?mode=ro", uri=True) misparses a path containing '?', '#', or spaces (SQLite URI syntax treats them as query/fragment delimiters), silently opening the wrong file or failing to open it. Percent-encode the path (urllib.parse.quote, safe="/") before building the URI. Verification (RED -> GREEN): added tests/test_opencode_ingest_robustness.py with one test per bug. Confirmed each fails against the pre-fix code (git stash of indexer.py) with the exact reported failure mode -- JSONDecodeError aborting the whole DB, TypeError on NULL worktree, dropped/duplicated exchange content after an out-of-order message insert (content-level assertion, not just row count, since the row count coincidentally matches under the bug), and OperationalError: no such table for a '#' in the DB path -- then confirmed all four pass after restoring the fix. make check (run directly, not via the pre-commit hook): ruff 0 errors, pyright 0 errors, 665 pytest tests pass (4 new). Note: --no-verify used deliberately, per issue #46 (this worktree's git hooks run tests that spawn nested git subprocesses which can corrupt the outer repo's .git/config and index when run from inside a hook). make check was run directly beforehand and passes clean. Closes #21 --- src/codeatrium/indexer.py | 87 ++++---- tests/test_opencode_ingest_robustness.py | 260 +++++++++++++++++++++++ 2 files changed, 309 insertions(+), 38 deletions(-) create mode 100644 tests/test_opencode_ingest_robustness.py diff --git a/src/codeatrium/indexer.py b/src/codeatrium/indexer.py index c025539..9b57759 100644 --- a/src/codeatrium/indexer.py +++ b/src/codeatrium/indexer.py @@ -23,6 +23,7 @@ from datetime import UTC, datetime from pathlib import Path from typing import TYPE_CHECKING, Any +from urllib.parse import quote from codeatrium.adapters.harness import claude as claude_adapter from codeatrium.adapters.harness import codex as codex_adapter @@ -718,6 +719,29 @@ def _epoch_ms_to_iso(epoch_ms: int) -> str: return datetime.fromtimestamp(epoch_ms / 1000, tz=UTC).isoformat() +def _parse_opencode_row( + kind: str, row: sqlite3.Row, session_id: str +) -> tuple[int, str, dict] | None: + """message/part の1行を envelope に変換する。 + + 不正 JSON・NULL の time_created/data など破損行は None を返す + （1行の破損が DB 全体の取り込みを中断させないよう、呼び出し側でスキップする）。 + """ + try: + envelope: dict = { + "kind": kind, + "id": row["id"], + "session_id": session_id, + "timestamp": _epoch_ms_to_iso(row["time_created"]), + "data": json.loads(row["data"]), + } + if kind == "part": + envelope["message_id"] = row["message_id"] + return row["time_created"], row["id"], envelope + except (TypeError, ValueError, OverflowError, json.JSONDecodeError): + return None + + def _load_opencode_raw_entries( src: sqlite3.Connection, session_id: str ) -> tuple[list[dict | None], str]: @@ -738,34 +762,13 @@ def _load_opencode_raw_entries( ordered: list[tuple[int, str, dict]] = [] for row in messages: - ordered.append( - ( - row["time_created"], - row["id"], - { - "kind": "message", - "id": row["id"], - "session_id": session_id, - "timestamp": _epoch_ms_to_iso(row["time_created"]), - "data": json.loads(row["data"]), - }, - ) - ) + parsed = _parse_opencode_row("message", row, session_id) + if parsed is not None: + ordered.append(parsed) for row in parts: - ordered.append( - ( - row["time_created"], - row["id"], - { - "kind": "part", - "id": row["id"], - "message_id": row["message_id"], - "session_id": session_id, - "timestamp": _epoch_ms_to_iso(row["time_created"]), - "data": json.loads(row["data"]), - }, - ) - ) + parsed = _parse_opencode_row("part", row, session_id) + if parsed is not None: + ordered.append(parsed) ordered.sort(key=lambda item: (item[0], item[1])) raw_entries: list[dict | None] = [entry for _, _, entry in ordered] @@ -794,14 +797,16 @@ def index_opencode_db( if project_root is None: return 0 - src = sqlite3.connect(f"file:{opencode_db_path}?mode=ro", uri=True) + db_uri = f"file:{quote(str(opencode_db_path), safe='/')}?mode=ro" + src = sqlite3.connect(db_uri, uri=True) src.row_factory = sqlite3.Row try: project_root_real = os.path.realpath(str(project_root)) project_ids = [ row["id"] for row in src.execute("SELECT id, worktree FROM project") - if os.path.realpath(row["worktree"]) == project_root_real + if row["worktree"] is not None + and os.path.realpath(row["worktree"]) == project_root_real ] if not project_ids: return 0 @@ -818,20 +823,26 @@ def index_opencode_db( for session_row in session_rows: session_id = session_row["id"] source_path = f"{opencode_db_path}#{session_id}" - conversation_id = sha256(source_path) - row = con.execute( - "SELECT last_ply_end FROM conversations WHERE id = ?", - (conversation_id,), - ).fetchone() - last_ply_end = row["last_ply_end"] if row is not None else -1 + # ply_start はセッション内での位置添字で、新規メッセージが既存より + # 古い time_created で到着すると添字が全体シフトする。位置ではなく + # exchange.id（user message id 由来で位置非依存）で既取り込み分を + # 判定し、旧ターンの再emit/新規ターンの取りこぼしを防ぐ。 + known_exchange_ids = { + row["source_turn_id"] + for row in con.execute( + "SELECT source_turn_id FROM exchanges " + "WHERE harness = 'opencode' AND source_session_id = ?", + (session_id,), + ) + } raw_entries, started_at = _load_opencode_raw_entries(src, session_id) exchanges = parse_opencode_exchanges( source_path, raw_entries, min_chars=min_chars ) new_exchanges = [ - ex for ex in exchanges if ex.ply_start > last_ply_end + ex for ex in exchanges if ex.id not in known_exchange_ids ] if not new_exchanges: continue @@ -869,7 +880,7 @@ def index_opencode_db( if touches or renames: artifacts.append( ExchangeArtifacts( - source_turn_id=str(exchange.ply_start), + source_turn_id=exchange.id, code_touches=touches, file_renames=renames, ) @@ -883,7 +894,7 @@ def index_opencode_db( f"{exchange.ply_start}-{exchange.ply_end}" ), source_session_id=session_id, - source_turn_id=str(exchange.ply_start), + source_turn_id=exchange.id, ply_start=exchange.ply_start, ply_end=exchange.ply_end, user_content=exchange.user_content, diff --git a/tests/test_opencode_ingest_robustness.py b/tests/test_opencode_ingest_robustness.py new file mode 100644 index 0000000..de28494 --- /dev/null +++ b/tests/test_opencode_ingest_robustness.py @@ -0,0 +1,260 @@ +"""OpenCode 取り込みのロバスト性（issue #21）を検証する。 + +_load_opencode_raw_entries / index_opencode_db の4つの既知バグを対象にする: + 1. 1行の破損（不正 JSON・NULL）が DB 全体の取り込みを中断してはならない + 2. project.worktree が NULL のとき os.path.realpath で例外を起こしてはならない + 3. ply_start（位置添字）ベースのカーソルは、time_created が既存行より古い新規 + メッセージの到着で添字が全体シフトし、既取り込みターンを再emit（重複登録）する + 4. sqlite3 の file: URI は DB パスに ?/#/空白 を含むと誤解釈されるため、 + percent-encode してから接続しなければならない +""" + +from __future__ import annotations + +import json +import sqlite3 +from pathlib import Path + +from codeatrium.db import get_connection, init_db +from codeatrium.indexer import index_opencode_db + +_SCHEMA = """ +CREATE TABLE project (id TEXT PRIMARY KEY, worktree TEXT, vcs TEXT, name TEXT); +CREATE TABLE session (id TEXT PRIMARY KEY, project_id TEXT NOT NULL, directory TEXT NOT NULL); +CREATE TABLE message ( + id TEXT PRIMARY KEY, session_id TEXT NOT NULL, + time_created INTEGER, time_updated INTEGER, data TEXT +); +CREATE TABLE part ( + id TEXT PRIMARY KEY, message_id TEXT NOT NULL, session_id TEXT NOT NULL, + time_created INTEGER, time_updated INTEGER, data TEXT +); +""" + + +def _user_message(msg_id: str, session_id: str, time_created: int) -> tuple: + return ( + msg_id, + session_id, + time_created, + time_created, + json.dumps({"role": "user"}), + ) + + +def _text_part(part_id: str, message_id: str, session_id: str, time_created: int, text: str) -> tuple: + return ( + part_id, + message_id, + session_id, + time_created, + time_created, + json.dumps({"type": "text", "text": text}), + ) + + +def _connect(db_file: Path) -> sqlite3.Connection: + con = sqlite3.connect(db_file) + con.executescript(_SCHEMA) + return con + + +def test_corrupt_row_does_not_abort_whole_db_ingestion(tmp_path: Path) -> None: + """破損した1行（不正 JSON）があっても、他セッションの取り込みは継続する。""" + project_root = tmp_path / "project" + project_root.mkdir() + opencode_db = tmp_path / "opencode.db" + + con = _connect(opencode_db) + con.execute( + "INSERT INTO project (id, worktree, vcs, name) VALUES (?, ?, ?, ?)", + ("proj1", str(project_root), "git", "repo"), + ) + con.execute( + "INSERT INTO session (id, project_id, directory) VALUES (?, ?, ?)", + ("ses_ok", "proj1", str(project_root)), + ) + con.execute( + "INSERT INTO message (id, session_id, time_created, time_updated, data) " + "VALUES (?, ?, ?, ?, ?)", + _user_message("msg_ok", "ses_ok", 1000), + ) + con.execute( + "INSERT INTO part (id, message_id, session_id, time_created, time_updated, data) " + "VALUES (?, ?, ?, ?, ?, ?)", + _text_part("prt_ok", "msg_ok", "ses_ok", 1001, "a" * 60), + ) + # 破損行: data が不正 JSON。同一セッション内に混在させる。 + con.execute( + "INSERT INTO message (id, session_id, time_created, time_updated, data) " + "VALUES (?, ?, ?, ?, ?)", + ("msg_corrupt", "ses_ok", 1002, 1002, "{not-valid-json"), + ) + # 破損行: time_created が NULL。 + con.execute( + "INSERT INTO message (id, session_id, time_created, time_updated, data) " + "VALUES (?, ?, ?, ?, ?)", + ("msg_null_time", "ses_ok", None, None, json.dumps({"role": "user"})), + ) + con.commit() + con.close() + + db_path = project_root / ".codeatrium" / "memory.db" + init_db(db_path) + + # 破損行があっても例外を送出せず、正常な exchange は取り込まれる。 + indexed = index_opencode_db( + opencode_db, db_path, min_chars=1, project_root=project_root + ) + assert indexed == 1 + + +def test_worktree_none_is_skipped_not_raised(tmp_path: Path) -> None: + """project.worktree が NULL の行は os.path.realpath に渡さずスキップする。""" + project_root = tmp_path / "project" + project_root.mkdir() + opencode_db = tmp_path / "opencode.db" + + con = _connect(opencode_db) + con.execute( + "INSERT INTO project (id, worktree, vcs, name) VALUES (?, ?, ?, ?)", + ("proj_null", None, "git", "orphan"), + ) + con.execute( + "INSERT INTO project (id, worktree, vcs, name) VALUES (?, ?, ?, ?)", + ("proj_ok", str(project_root), "git", "repo"), + ) + con.execute( + "INSERT INTO session (id, project_id, directory) VALUES (?, ?, ?)", + ("ses_ok", "proj_ok", str(project_root)), + ) + con.execute( + "INSERT INTO message (id, session_id, time_created, time_updated, data) " + "VALUES (?, ?, ?, ?, ?)", + _user_message("msg_ok", "ses_ok", 1000), + ) + con.execute( + "INSERT INTO part (id, message_id, session_id, time_created, time_updated, data) " + "VALUES (?, ?, ?, ?, ?, ?)", + _text_part("prt_ok", "msg_ok", "ses_ok", 1001, "b" * 60), + ) + con.commit() + con.close() + + db_path = project_root / ".codeatrium" / "memory.db" + init_db(db_path) + + indexed = index_opencode_db( + opencode_db, db_path, min_chars=1, project_root=project_root + ) + assert indexed == 1 + + +def test_out_of_order_message_does_not_reemit_existing_exchange(tmp_path: Path) -> None: + """time_created が既存より古い新規メッセージの到着で ply 添字が全体シフトしても、 + 既に取り込み済みのターンを重複登録しない（安定 message-id ベースのカーソル）。""" + project_root = tmp_path / "project" + project_root.mkdir() + opencode_db = tmp_path / "opencode.db" + + con = _connect(opencode_db) + con.execute( + "INSERT INTO project (id, worktree, vcs, name) VALUES (?, ?, ?, ?)", + ("proj1", str(project_root), "git", "repo"), + ) + con.execute( + "INSERT INTO session (id, project_id, directory) VALUES (?, ?, ?)", + ("ses1", "proj1", str(project_root)), + ) + con.execute( + "INSERT INTO message (id, session_id, time_created, time_updated, data) " + "VALUES (?, ?, ?, ?, ?)", + _user_message("msg1", "ses1", 1000), + ) + con.execute( + "INSERT INTO part (id, message_id, session_id, time_created, time_updated, data) " + "VALUES (?, ?, ?, ?, ?, ?)", + _text_part("prt1", "msg1", "ses1", 1001, "c" * 60), + ) + con.commit() + con.close() + + db_path = project_root / ".codeatrium" / "memory.db" + init_db(db_path) + + first = index_opencode_db( + opencode_db, db_path, min_chars=1, project_root=project_root + ) + assert first == 1 + + # msg0 が msg1 より「古い」time_created で後から到着する + # （バックフィル・クロックスキュー等の実運用シナリオ）。 + # (time_created, id) 順ソートで msg0/prt0 が msg1/prt1 の手前に入り、 + # msg1 の raw_entries 内の位置添字（ply_start）が変わる。 + con = sqlite3.connect(opencode_db) + con.execute( + "INSERT INTO message (id, session_id, time_created, time_updated, data) " + "VALUES (?, ?, ?, ?, ?)", + _user_message("msg0", "ses1", 500), + ) + con.execute( + "INSERT INTO part (id, message_id, session_id, time_created, time_updated, data) " + "VALUES (?, ?, ?, ?, ?, ?)", + _text_part("prt0", "msg0", "ses1", 501, "d" * 60), + ) + con.commit() + con.close() + + second = index_opencode_db( + opencode_db, db_path, min_chars=1, project_root=project_root + ) + # msg0 の1件だけが新規に登録され、msg1 は重複登録されない。 + assert second == 1 + + con = get_connection(db_path) + contents = sorted( + row[0] for row in con.execute("SELECT user_content FROM exchanges") + ) + con.close() + # 位置添字ベースのカーソルだと、msg1 は ply_start シフトで再emit（重複）される + # 一方、真に新規な msg0 は旧 msg1 と偶然ハッシュ衝突して黙殺され得る + # （id ベースのカーソルなら両方が正確に1件ずつ残る）。 + assert contents == sorted(["c" * 60, "d" * 60]) + + +def test_db_path_with_special_uri_characters_is_opened_correctly(tmp_path: Path) -> None: + """DB パスに '#' を含む場合でも file: URI が誤解釈されず正しいファイルを開く。""" + project_root = tmp_path / "project" + project_root.mkdir() + # '#' は file: URI のフラグメント区切りとして誤解釈され得る文字。 + opencode_db = tmp_path / "op#session.db" + + con = _connect(opencode_db) + con.execute( + "INSERT INTO project (id, worktree, vcs, name) VALUES (?, ?, ?, ?)", + ("proj1", str(project_root), "git", "repo"), + ) + con.execute( + "INSERT INTO session (id, project_id, directory) VALUES (?, ?, ?)", + ("ses1", "proj1", str(project_root)), + ) + con.execute( + "INSERT INTO message (id, session_id, time_created, time_updated, data) " + "VALUES (?, ?, ?, ?, ?)", + _user_message("msg1", "ses1", 1000), + ) + con.execute( + "INSERT INTO part (id, message_id, session_id, time_created, time_updated, data) " + "VALUES (?, ?, ?, ?, ?, ?)", + _text_part("prt1", "msg1", "ses1", 1001, "e" * 60), + ) + con.commit() + con.close() + + db_path = project_root / ".codeatrium" / "memory.db" + init_db(db_path) + + indexed = index_opencode_db( + opencode_db, db_path, min_chars=1, project_root=project_root + ) + assert indexed == 1 From 828564c21e73ed25960c212652711b4a84abf43a Mon Sep 17 00:00:00 2001 From: senna-lang Date: Mon, 7 Sep 2026 09:19:54 +0900 Subject: [PATCH 2/4] =?UTF-8?q?fix(indexer):=20address=20PR=20#48=20review?= =?UTF-8?q?=20=E2=80=94=20non-dict=20JSON=20+=20upgrade-path=20duplication?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two priority-1 issues raised on PR #48 for #21: - _parse_opencode_row only caught JSON decode/conversion errors. A row whose data is syntactically valid JSON but not an object (null, [], "text") passed the guard, entered raw_entries, and crashed parse_opencode_exchanges's `entry["data"].get(...)` calls with AttributeError -- the same "one bad row aborts the whole DB" failure #21 asked to eliminate, just via a different input shape. Now checks `isinstance(data, dict)` after json.loads and skips the row (returns None) if it isn't. - index_opencode_db's new id-based dedup broke upgrading from pre-patch data: exchanges indexed before this fix have source_turn_id = str(ply_start) (a numeric string); after the cursor change every freshly-parsed exchange computes a hashed exchange.id instead, so known_exchange_ids (built from the stored source_turn_id column) never matched a prior row and the first post-upgrade run would insert a second copy of every previously-indexed OpenCode exchange. Added a fallback check: an exchange is also treated as already-known if str(ex.ply_start) is present in known_exchange_ids. New-scheme ids are always 64-char sha256 hex, so this can never false-positive against a genuinely new exchange -- it only matches legacy numeric turn ids left over from before the upgrade. Tests added to tests/test_opencode_ingest_robustness.py: - test_non_dict_json_row_is_skipped_not_crashed - test_upgrade_from_legacy_position_based_cursor_does_not_duplicate (seeds the target DB via ingest_parse_result with a source_turn_id=str(ply_start) exchange -- i.e. exactly what the pre-patch index_opencode_db would have persisted -- then re-runs the current index_opencode_db and asserts the count stays at 1) Verification (RED -> GREEN): stashed the two indexer.py fixes (keeping the new tests), confirmed both fail with the exact reported failure modes (AttributeError: 'list' object has no attribute 'get'; assert 1 == 0 on the duplicate-count check), then restored the fixes and confirmed both pass. All 6 tests in the file pass. make check (run directly, not via the pre-commit hook, per #46): ruff 0 errors, pyright 0 errors, 667 pytest tests pass (2 new). Addresses review comments on #48. --- src/codeatrium/indexer.py | 19 ++- tests/test_opencode_ingest_robustness.py | 166 ++++++++++++++++++++++- 2 files changed, 181 insertions(+), 4 deletions(-) diff --git a/src/codeatrium/indexer.py b/src/codeatrium/indexer.py index 9b57759..e5c21b1 100644 --- a/src/codeatrium/indexer.py +++ b/src/codeatrium/indexer.py @@ -724,16 +724,20 @@ def _parse_opencode_row( ) -> tuple[int, str, dict] | None: """message/part の1行を envelope に変換する。 - 不正 JSON・NULL の time_created/data など破損行は None を返す + 不正 JSON・NULL の time_created/data、および dict でない JSON 値 + （null・配列・文字列など）を持つ破損行は None を返す （1行の破損が DB 全体の取り込みを中断させないよう、呼び出し側でスキップする）。 """ try: + data = json.loads(row["data"]) + if not isinstance(data, dict): + return None envelope: dict = { "kind": kind, "id": row["id"], "session_id": session_id, "timestamp": _epoch_ms_to_iso(row["time_created"]), - "data": json.loads(row["data"]), + "data": data, } if kind == "part": envelope["message_id"] = row["message_id"] @@ -828,6 +832,12 @@ def index_opencode_db( # 古い time_created で到着すると添字が全体シフトする。位置ではなく # exchange.id（user message id 由来で位置非依存）で既取り込み分を # 判定し、旧ターンの再emit/新規ターンの取りこぼしを防ぐ。 + # + # 後方互換: この修正より前に取り込んだ行は source_turn_id に + # str(ply_start)（数値文字列）を格納している。新スキームの + # exchange.id（sha256 ハッシュ）とは一致しないため、旧スキームの + # 数値 id も known set に含め、二重登録を防ぐ（新スキームの id は + # 常にハッシュ文字列なので、数値文字列との衝突は起きない）。 known_exchange_ids = { row["source_turn_id"] for row in con.execute( @@ -842,7 +852,10 @@ def index_opencode_db( source_path, raw_entries, min_chars=min_chars ) new_exchanges = [ - ex for ex in exchanges if ex.id not in known_exchange_ids + ex + for ex in exchanges + if ex.id not in known_exchange_ids + and str(ex.ply_start) not in known_exchange_ids ] if not new_exchanges: continue diff --git a/tests/test_opencode_ingest_robustness.py b/tests/test_opencode_ingest_robustness.py index de28494..1d7c160 100644 --- a/tests/test_opencode_ingest_robustness.py +++ b/tests/test_opencode_ingest_robustness.py @@ -15,8 +15,14 @@ import sqlite3 from pathlib import Path +from codeatrium.core.ingest import ingest_parse_result +from codeatrium.core.models import CanonicalExchange, CanonicalSession, ParseResult from codeatrium.db import get_connection, init_db -from codeatrium.indexer import index_opencode_db +from codeatrium.indexer import ( + _load_opencode_raw_entries, + index_opencode_db, + parse_opencode_exchanges, +) _SCHEMA = """ CREATE TABLE project (id TEXT PRIMARY KEY, worktree TEXT, vcs TEXT, name TEXT); @@ -109,6 +115,65 @@ def test_corrupt_row_does_not_abort_whole_db_ingestion(tmp_path: Path) -> None: assert indexed == 1 +def test_non_dict_json_row_is_skipped_not_crashed(tmp_path: Path) -> None: + """data が構文的に正しい JSON でも dict でない（null/配列/文字列）行はスキップする。 + + json.loads はこれらの値に対して例外を出さずに成功するため、JSONDecodeError だけを + 捕捉するガードはこの形の破損行を素通りさせてしまい、parse_opencode_exchanges の + entry["data"].get(...) 呼び出しで AttributeError を起こして DB 全体の取り込みを + 中断させる（#21 が排除しようとした失敗モードそのものが別の入力形で再発する）。 + """ + project_root = tmp_path / "project" + project_root.mkdir() + opencode_db = tmp_path / "opencode.db" + + con = _connect(opencode_db) + con.execute( + "INSERT INTO project (id, worktree, vcs, name) VALUES (?, ?, ?, ?)", + ("proj1", str(project_root), "git", "repo"), + ) + con.execute( + "INSERT INTO session (id, project_id, directory) VALUES (?, ?, ?)", + ("ses_ok", "proj1", str(project_root)), + ) + con.execute( + "INSERT INTO message (id, session_id, time_created, time_updated, data) " + "VALUES (?, ?, ?, ?, ?)", + _user_message("msg_ok", "ses_ok", 1000), + ) + con.execute( + "INSERT INTO part (id, message_id, session_id, time_created, time_updated, data) " + "VALUES (?, ?, ?, ?, ?, ?)", + _text_part("prt_ok", "msg_ok", "ses_ok", 1001, "z" * 60), + ) + # 破損行: data が構文的に正しい JSON だが dict ではない（null / 配列 / 文字列）。 + for bad_id, bad_json in ( + ("msg_null", "null"), + ("msg_list", "[]"), + ("msg_str", '"just text"'), + ): + con.execute( + "INSERT INTO message (id, session_id, time_created, time_updated, data) " + "VALUES (?, ?, ?, ?, ?)", + (bad_id, "ses_ok", 1003, 1003, bad_json), + ) + con.execute( + "INSERT INTO part (id, message_id, session_id, time_created, time_updated, data) " + "VALUES (?, ?, ?, ?, ?, ?)", + ("prt_null", "msg_ok", "ses_ok", 1004, 1004, "null"), + ) + con.commit() + con.close() + + db_path = project_root / ".codeatrium" / "memory.db" + init_db(db_path) + + indexed = index_opencode_db( + opencode_db, db_path, min_chars=1, project_root=project_root + ) + assert indexed == 1 + + def test_worktree_none_is_skipped_not_raised(tmp_path: Path) -> None: """project.worktree が NULL の行は os.path.realpath に渡さずスキップする。""" project_root = tmp_path / "project" @@ -258,3 +323,102 @@ def test_db_path_with_special_uri_characters_is_opened_correctly(tmp_path: Path) opencode_db, db_path, min_chars=1, project_root=project_root ) assert indexed == 1 + + + +def test_upgrade_from_legacy_position_based_cursor_does_not_duplicate( + tmp_path: Path, +) -> None: + """パッチ適用前に position ベースの source_turn_id (str(ply_start)) で取り込み + 済みの exchange は、id ベースのカーソルへ移行した後の再取り込みで重複登録 + されない（旧スキームの数値 id を known set のフォールバックとして扱う）。""" + project_root = tmp_path / "project" + project_root.mkdir() + opencode_db = tmp_path / "opencode.db" + + con = _connect(opencode_db) + con.execute( + "INSERT INTO project (id, worktree, vcs, name) VALUES (?, ?, ?, ?)", + ("proj1", str(project_root), "git", "repo"), + ) + con.execute( + "INSERT INTO session (id, project_id, directory) VALUES (?, ?, ?)", + ("ses1", "proj1", str(project_root)), + ) + con.execute( + "INSERT INTO message (id, session_id, time_created, time_updated, data) " + "VALUES (?, ?, ?, ?, ?)", + _user_message("msg1", "ses1", 1000), + ) + con.execute( + "INSERT INTO part (id, message_id, session_id, time_created, time_updated, data) " + "VALUES (?, ?, ?, ?, ?, ?)", + _text_part("prt1", "msg1", "ses1", 1001, "legacy " + "x" * 60), + ) + con.commit() + con.close() + + db_path = project_root / ".codeatrium" / "memory.db" + init_db(db_path) + + # 旧スキーム（source_turn_id=str(ply_start)）で取り込み済みの状態を直接構築する + # （このパッチ以前の index_opencode_db が生成していた永続化結果を模する）。 + src = sqlite3.connect(f"file:{opencode_db}?mode=ro", uri=True) + src.row_factory = sqlite3.Row + raw_entries, started_at = _load_opencode_raw_entries(src, "ses1") + src.close() + source_path = f"{opencode_db}#ses1" + exchanges = parse_opencode_exchanges(source_path, raw_entries, min_chars=1) + assert len(exchanges) == 1 + legacy_exchange = exchanges[0] + legacy_turn_id = str(legacy_exchange.ply_start) + + con = get_connection(db_path) + ingest_parse_result( + con, + CanonicalSession( + harness="opencode", + source_session_id="ses1", + primary_ref=source_path, + project_key=str(project_root), + started_at=started_at, + ), + ParseResult( + exchanges=( + CanonicalExchange( + harness="opencode", + session_ref=( + f"{source_path}#ply=" + f"{legacy_exchange.ply_start}-{legacy_exchange.ply_end}" + ), + source_session_id="ses1", + source_turn_id=legacy_turn_id, + ply_start=legacy_exchange.ply_start, + ply_end=legacy_exchange.ply_end, + user_content=legacy_exchange.user_content, + agent_content=legacy_exchange.agent_content, + files_touched=tuple(legacy_exchange.files), + git_branch=legacy_exchange.git_branch, + ), + ), + next_cursor=f"v1:ply:{legacy_exchange.ply_end}", + ), + ) + con.commit() + con.close() + + con = get_connection(db_path) + pre_upgrade_count = con.execute("SELECT COUNT(*) FROM exchanges").fetchone()[0] + con.close() + assert pre_upgrade_count == 1 + + # パッチ適用後の index_opencode_db を同じ opencode DB に対して再実行する。 + reindexed = index_opencode_db( + opencode_db, db_path, min_chars=1, project_root=project_root + ) + assert reindexed == 0 + + con = get_connection(db_path) + post_upgrade_count = con.execute("SELECT COUNT(*) FROM exchanges").fetchone()[0] + con.close() + assert post_upgrade_count == 1 \ No newline at end of file From e0dc3134fcb87f5d3504625897c1dd8fe4698317 Mon Sep 17 00:00:00 2001 From: senna-lang Date: Mon, 7 Sep 2026 09:33:44 +0900 Subject: [PATCH 3/4] fix(indexer): replace position-based legacy fallback with content-based migration MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Follow-up to PR #48 review: the previous fix (str(ex.ply_start) fallback in known_exchange_ids) was still broken, because a legacy source_turn_id identifies a *position*, not a stable identity. Exact failure the reviewer traced: a legacy exchange at ply_start==0 plus a newly-arrived out-of-order message that now sorts ahead of it shifts the new message into position 0 (wrongly matched as "known" against the legacy id) while the old exchange shifts to position 2 (re-inserted as a duplicate under its new hash id) -- the exact upgrade scenario #21 needs to survive, still failing. Fix: drop the position-based fallback entirely. Legacy rows (source_turn_id is a short all-digit string, distinguishable from the new scheme's 64-char sha256 hex by _is_legacy_opencode_turn_id) are now matched to the current parse by *content* -- the (user_content, agent_content) pair, which is invariant to how much the underlying (time_created, id)-sorted list has shifted, since it's derived from the message's actual text, not its list position. On a match, the existing DB row's source_turn_id is rewritten in place to the new exchange.id (a stable hash of conversation_id + the OpenCode message id); intentionally left untouched: `id`/`canonical_exchange_id` (the FK target for code_touches/exchange_files/palace_objects/ vec_exchanges) — rewriting those would cascade into palace_objects.id (itself sha256(f"palace:{exchange_id}")) and its own downstream rooms/symbols/vec_palace, for no functional gain: index_opencode_db never relies on ingest_parse_result's id-based dedup for OpenCode once this content-based pre-filter runs, so leaving the primary key stable is strictly safer with an identical deduplication guarantee. Tests (tests/test_opencode_ingest_robustness.py): - extracted _seed_legacy_opencode_exchange() to build a pre-patch persisted state without duplicating ~50 lines per test - test_upgrade_with_out_of_order_message_migrates_legacy_and_captures_new reproduces the reviewer's exact scenario: seed one legacy exchange at ply_start==0, then insert a new message with an earlier time_created, then reindex; asserts exactly one new row is inserted and both the legacy and new exchange content are present (not duplicated/dropped) Verification (RED -> GREEN): stashed the indexer.py fix (keeping the new test), confirmed it fails exactly as predicted -- the new message's content ("new ...") is dropped and the legacy content ("legacy ...") appears as if unique when it should coexist with the new one, i.e. the new exchange is suppressed and nothing gets duplicated in this particular assertion shape, matching the reviewer's "loses the new exchange" half of the report. Restored the fix: all 7 tests in the file pass, including the pre-existing test_upgrade_from_legacy_ position_based_cursor_does_not_duplicate. make check (run directly, not via the pre-commit hook, per #46): ruff 0 errors, pyright 0 errors, 668 pytest tests pass (1 new). Addresses further review comment on #48. --- src/codeatrium/indexer.py | 69 +++++++--- tests/test_opencode_ingest_robustness.py | 164 +++++++++++++++++------ 2 files changed, 174 insertions(+), 59 deletions(-) diff --git a/src/codeatrium/indexer.py b/src/codeatrium/indexer.py index e5c21b1..278b820 100644 --- a/src/codeatrium/indexer.py +++ b/src/codeatrium/indexer.py @@ -782,6 +782,17 @@ def _load_opencode_raw_entries( return raw_entries, started_at +def _is_legacy_opencode_turn_id(source_turn_id: str) -> bool: + """このパッチより前の index_opencode_db が書き込んだ position ベースの + source_turn_id（str(ply_start)、短い数値文字列）かどうかを判定する。 + + 新スキームの source_turn_id は常に sha256 hexdigest（64桁の16進文字列）で、 + 数字だけになる確率は無視できるほど低いため、「全桁が数字」かつ「64桁未満」を + 旧スキームの判定に使う。 + """ + return source_turn_id.isdigit() and len(source_turn_id) < 64 + + def index_opencode_db( opencode_db_path: Path, db_path: Path, @@ -833,30 +844,52 @@ def index_opencode_db( # exchange.id（user message id 由来で位置非依存）で既取り込み分を # 判定し、旧ターンの再emit/新規ターンの取りこぼしを防ぐ。 # - # 後方互換: この修正より前に取り込んだ行は source_turn_id に - # str(ply_start)（数値文字列）を格納している。新スキームの - # exchange.id（sha256 ハッシュ）とは一致しないため、旧スキームの - # 数値 id も known set に含め、二重登録を防ぐ（新スキームの id は - # 常にハッシュ文字列なので、数値文字列との衝突は起きない）。 - known_exchange_ids = { - row["source_turn_id"] - for row in con.execute( - "SELECT source_turn_id FROM exchanges " - "WHERE harness = 'opencode' AND source_session_id = ?", - (session_id,), - ) + # 後方互換: このパッチより前に取り込んだ行は source_turn_id に + # str(ply_start)（位置そのもの）を格納しており、位置は安定した + # identity ではない。アップグレード後に新規メッセージが + # out-of-order で到着すると位置が全体シフトするため、position での + # 突き合わせでは新規メッセージを誤って「既知」扱いし、旧 exchange を + # 新ハッシュ id で二重登録してしまう（#48 レビュー指摘）。 + # 位置ではなく実際の内容（user_content・agent_content の組）で + # 旧行と現在のパース結果を突き合わせ、一致した旧行の + # source_turn_id を新スキームへその場で書き換える。 + # id/canonical_exchange_id は変更しない — code_touches / + # exchange_files / palace_objects / vec_exchanges からの + # exchange_id 参照はそのまま有効であり、以後の重複判定は本関数の + # 事前フィルタのみで行われるため書き換え不要。 + existing_rows = con.execute( + "SELECT source_turn_id, user_content, agent_content " + "FROM exchanges WHERE harness = 'opencode' " + "AND source_session_id = ?", + (session_id,), + ).fetchall() + known_exchange_ids = {row["source_turn_id"] for row in existing_rows} + legacy_by_content = { + (row["user_content"], row["agent_content"]): row["source_turn_id"] + for row in existing_rows + if _is_legacy_opencode_turn_id(row["source_turn_id"]) } raw_entries, started_at = _load_opencode_raw_entries(src, session_id) exchanges = parse_opencode_exchanges( source_path, raw_entries, min_chars=min_chars ) - new_exchanges = [ - ex - for ex in exchanges - if ex.id not in known_exchange_ids - and str(ex.ply_start) not in known_exchange_ids - ] + new_exchanges = [] + for exchange in exchanges: + if exchange.id in known_exchange_ids: + continue + legacy_turn_id = legacy_by_content.pop( + (exchange.user_content, exchange.agent_content), None + ) + if legacy_turn_id is not None: + con.execute( + "UPDATE exchanges SET source_turn_id = ? " + "WHERE harness = 'opencode' AND source_session_id = ? " + "AND source_turn_id = ?", + (exchange.id, session_id, legacy_turn_id), + ) + continue + new_exchanges.append(exchange) if not new_exchanges: continue diff --git a/tests/test_opencode_ingest_robustness.py b/tests/test_opencode_ingest_robustness.py index 1d7c160..199a051 100644 --- a/tests/test_opencode_ingest_robustness.py +++ b/tests/test_opencode_ingest_robustness.py @@ -326,48 +326,18 @@ def test_db_path_with_special_uri_characters_is_opened_correctly(tmp_path: Path) -def test_upgrade_from_legacy_position_based_cursor_does_not_duplicate( - tmp_path: Path, +def _seed_legacy_opencode_exchange( + opencode_db: Path, db_path: Path, project_root: Path, session_id: str ) -> None: - """パッチ適用前に position ベースの source_turn_id (str(ply_start)) で取り込み - 済みの exchange は、id ベースのカーソルへ移行した後の再取り込みで重複登録 - されない（旧スキームの数値 id を known set のフォールバックとして扱う）。""" - project_root = tmp_path / "project" - project_root.mkdir() - opencode_db = tmp_path / "opencode.db" - - con = _connect(opencode_db) - con.execute( - "INSERT INTO project (id, worktree, vcs, name) VALUES (?, ?, ?, ?)", - ("proj1", str(project_root), "git", "repo"), - ) - con.execute( - "INSERT INTO session (id, project_id, directory) VALUES (?, ?, ?)", - ("ses1", "proj1", str(project_root)), - ) - con.execute( - "INSERT INTO message (id, session_id, time_created, time_updated, data) " - "VALUES (?, ?, ?, ?, ?)", - _user_message("msg1", "ses1", 1000), - ) - con.execute( - "INSERT INTO part (id, message_id, session_id, time_created, time_updated, data) " - "VALUES (?, ?, ?, ?, ?, ?)", - _text_part("prt1", "msg1", "ses1", 1001, "legacy " + "x" * 60), - ) - con.commit() - con.close() - - db_path = project_root / ".codeatrium" / "memory.db" - init_db(db_path) - - # 旧スキーム（source_turn_id=str(ply_start)）で取り込み済みの状態を直接構築する - # （このパッチ以前の index_opencode_db が生成していた永続化結果を模する）。 + """このパッチ以前の index_opencode_db が生成していた永続化結果 + （source_turn_id=str(ply_start)、数値の位置カーソル）を、実際に opencode_db を + 1回パースした結果から直接構築する。session_id には唯一の user メッセージが + 含まれる前提（他のテストの前提と合わせるため）。""" src = sqlite3.connect(f"file:{opencode_db}?mode=ro", uri=True) src.row_factory = sqlite3.Row - raw_entries, started_at = _load_opencode_raw_entries(src, "ses1") + raw_entries, started_at = _load_opencode_raw_entries(src, session_id) src.close() - source_path = f"{opencode_db}#ses1" + source_path = f"{opencode_db}#{session_id}" exchanges = parse_opencode_exchanges(source_path, raw_entries, min_chars=1) assert len(exchanges) == 1 legacy_exchange = exchanges[0] @@ -378,7 +348,7 @@ def test_upgrade_from_legacy_position_based_cursor_does_not_duplicate( con, CanonicalSession( harness="opencode", - source_session_id="ses1", + source_session_id=session_id, primary_ref=source_path, project_key=str(project_root), started_at=started_at, @@ -391,7 +361,7 @@ def test_upgrade_from_legacy_position_based_cursor_does_not_duplicate( f"{source_path}#ply=" f"{legacy_exchange.ply_start}-{legacy_exchange.ply_end}" ), - source_session_id="ses1", + source_session_id=session_id, source_turn_id=legacy_turn_id, ply_start=legacy_exchange.ply_start, ply_end=legacy_exchange.ply_end, @@ -407,6 +377,44 @@ def test_upgrade_from_legacy_position_based_cursor_does_not_duplicate( con.commit() con.close() + +def test_upgrade_from_legacy_position_based_cursor_does_not_duplicate( + tmp_path: Path, +) -> None: + """パッチ適用前に position ベースの source_turn_id (str(ply_start)) で取り込み + 済みの exchange は、id ベースのカーソルへ移行した後の再取り込みで重複登録 + されない（内容一致で旧行を新スキームへ移行する）。""" + project_root = tmp_path / "project" + project_root.mkdir() + opencode_db = tmp_path / "opencode.db" + + con = _connect(opencode_db) + con.execute( + "INSERT INTO project (id, worktree, vcs, name) VALUES (?, ?, ?, ?)", + ("proj1", str(project_root), "git", "repo"), + ) + con.execute( + "INSERT INTO session (id, project_id, directory) VALUES (?, ?, ?)", + ("ses1", "proj1", str(project_root)), + ) + con.execute( + "INSERT INTO message (id, session_id, time_created, time_updated, data) " + "VALUES (?, ?, ?, ?, ?)", + _user_message("msg1", "ses1", 1000), + ) + con.execute( + "INSERT INTO part (id, message_id, session_id, time_created, time_updated, data) " + "VALUES (?, ?, ?, ?, ?, ?)", + _text_part("prt1", "msg1", "ses1", 1001, "legacy " + "x" * 60), + ) + con.commit() + con.close() + + db_path = project_root / ".codeatrium" / "memory.db" + init_db(db_path) + + _seed_legacy_opencode_exchange(opencode_db, db_path, project_root, "ses1") + con = get_connection(db_path) pre_upgrade_count = con.execute("SELECT COUNT(*) FROM exchanges").fetchone()[0] con.close() @@ -421,4 +429,78 @@ def test_upgrade_from_legacy_position_based_cursor_does_not_duplicate( con = get_connection(db_path) post_upgrade_count = con.execute("SELECT COUNT(*) FROM exchanges").fetchone()[0] con.close() - assert post_upgrade_count == 1 \ No newline at end of file + assert post_upgrade_count == 1 + + +def test_upgrade_with_out_of_order_message_migrates_legacy_and_captures_new( + tmp_path: Path, +) -> None: + """アップグレード後、既存の legacy exchange（ply_start==0）より time_created が + 古い新規メッセージが到着しても、新規メッセージは正しく取り込まれ、位置が + シフトした旧 exchange は内容一致で新スキームへ移行され重複登録されない + （#48 レビュー指摘の再現シナリオ: 位置ベースの fallback では新規メッセージが + 位置0に来て誤って「既知」扱いされ、旧 exchange が新ハッシュ id で二重登録 + されてしまう）。""" + project_root = tmp_path / "project" + project_root.mkdir() + opencode_db = tmp_path / "opencode.db" + + con = _connect(opencode_db) + con.execute( + "INSERT INTO project (id, worktree, vcs, name) VALUES (?, ?, ?, ?)", + ("proj1", str(project_root), "git", "repo"), + ) + con.execute( + "INSERT INTO session (id, project_id, directory) VALUES (?, ?, ?)", + ("ses1", "proj1", str(project_root)), + ) + con.execute( + "INSERT INTO message (id, session_id, time_created, time_updated, data) " + "VALUES (?, ?, ?, ?, ?)", + _user_message("msg1", "ses1", 1000), + ) + con.execute( + "INSERT INTO part (id, message_id, session_id, time_created, time_updated, data) " + "VALUES (?, ?, ?, ?, ?, ?)", + _text_part("prt1", "msg1", "ses1", 1001, "legacy " + "x" * 60), + ) + con.commit() + con.close() + + db_path = project_root / ".codeatrium" / "memory.db" + init_db(db_path) + + # msg1 を旧スキーム（source_turn_id=str(ply_start)=="0"）で取り込み済みにする。 + _seed_legacy_opencode_exchange(opencode_db, db_path, project_root, "ses1") + + # msg1 より古い time_created を持つ新規メッセージが到着する + # （バックフィル・クロックスキュー等の実運用シナリオ）。(time_created, id) 順 + # ソートで msg0/prt0 が msg1/prt1 の手前に入り、msg1 の再パース時の ply_start + # は 0 -> 2 へシフトする。 + con = sqlite3.connect(opencode_db) + con.execute( + "INSERT INTO message (id, session_id, time_created, time_updated, data) " + "VALUES (?, ?, ?, ?, ?)", + _user_message("msg0", "ses1", 500), + ) + con.execute( + "INSERT INTO part (id, message_id, session_id, time_created, time_updated, data) " + "VALUES (?, ?, ?, ?, ?, ?)", + _text_part("prt0", "msg0", "ses1", 501, "new " + "y" * 60), + ) + con.commit() + con.close() + + reindexed = index_opencode_db( + opencode_db, db_path, min_chars=1, project_root=project_root + ) + # msg0（真に新規）の1件だけが新規登録される。位置がシフトした msg1 は + # 内容一致で旧行が新スキームへ移行されるだけで、重複登録されない。 + assert reindexed == 1 + + con = get_connection(db_path) + contents = sorted( + row[0] for row in con.execute("SELECT user_content FROM exchanges") + ) + con.close() + assert contents == sorted(["legacy " + "x" * 60, "new " + "y" * 60]) \ No newline at end of file From 62a87e35347a1aed904461291bf23855c29d5c14 Mon Sep 17 00:00:00 2001 From: senna-lang Date: Mon, 7 Sep 2026 09:44:10 +0900 Subject: [PATCH 4/4] fix(indexer): fix duplicate-content collisions and stale ply/session_ref on legacy migration Follow-up to PR #48 review round 3, two priority-2 gaps in the content-based legacy migration: 1. legacy_by_content was a plain dict keyed by (user_content, agent_content). Two legacy exchanges sharing that exact pair (a repeated user message in the same session) collided: the second overwrote the first before matching ran, so after a position shift only one got migrated and the other stayed behind as a permanent duplicate on the next re-index. Fixed by keying to a list of candidate rows per content pair, sorted by their original ply_start and consumed FIFO. Old messages' relative order is preserved across any number of new out-of-order insertions (time_created never changes once a message exists), so N legacy rows with identical content correctly line up 1:1 with the N corresponding entries encountered while walking the current parse in ply order. 2. Migrating a legacy row only rewrote source_turn_id, leaving ply_start/ply_end/session_ref at their pre-upgrade values. After a 0->2 shift this left the migrated row pointing at ply 0 -- now occupied by a different, newly-arrived exchange -- corrupting ply-adjacent context ordering and verbatim_ref resolution. The UPDATE now also sets ply_start, ply_end, and session_ref to the exchange's actual current position. Tests added to tests/test_opencode_ingest_robustness.py: - generalized _seed_legacy_opencode_exchange -> _seed_legacy_opencode_exchanges to seed N pre-patch exchanges in one call (was hardcoded to exactly 1) - test_upgrade_with_duplicate_content_legacy_exchanges_does_not_leave_duplicate: two legacy exchanges with byte-identical (user_content, agent_content), a third genuinely-new message shifts both; asserts exactly 3 rows survive with no duplicate - test_migrated_legacy_exchange_reflects_new_position_not_stale_one: asserts a migrated exchange's ply_start/ply_end/session_ref move from the stale 0-1 to the actual current 2-3 after a shift Verification (RED -> GREEN): stashed the two fixes (keeping the new tests) -- confirmed both fail exactly as predicted (3-row assertion saw 2 due to the dict-collision drop; ply_start/ply_end assertion saw the stale (0, 1) instead of (2, 3)). Restored the fixes: all 9 tests in the file pass. make check (run directly, not via the pre-commit hook, per #46): ruff 0 errors, pyright 0 errors, 670 pytest tests pass (2 new). Addresses round-3 review comment on #48. Disclosed in the PR reply: a residual limitation remains for byte-identical-content exchanges that collide with a *newly arriving* same-content message in the same reindex run -- ply_start/session_ref among that specific identical- content group can be assigned to a different sibling than its original one. This never causes duplication or data loss (the count and content set are always correct), only a traceability nit confined to rows whose displayed content is indistinguishable from each other anyway. A fully exact fix would require having persisted each legacy row's real OpenCode message id at original ingestion time, which pre-patch code never recorded -- content is the only durable signal available for reconciling rows created before this migration existed. --- src/codeatrium/indexer.py | 53 +++++-- tests/test_opencode_ingest_robustness.py | 192 +++++++++++++++++++++-- 2 files changed, 216 insertions(+), 29 deletions(-) diff --git a/src/codeatrium/indexer.py b/src/codeatrium/indexer.py index 278b820..b53b711 100644 --- a/src/codeatrium/indexer.py +++ b/src/codeatrium/indexer.py @@ -852,23 +852,34 @@ def index_opencode_db( # 新ハッシュ id で二重登録してしまう（#48 レビュー指摘）。 # 位置ではなく実際の内容（user_content・agent_content の組）で # 旧行と現在のパース結果を突き合わせ、一致した旧行の - # source_turn_id を新スキームへその場で書き換える。 - # id/canonical_exchange_id は変更しない — code_touches / - # exchange_files / palace_objects / vec_exchanges からの - # exchange_id 参照はそのまま有効であり、以後の重複判定は本関数の - # 事前フィルタのみで行われるため書き換え不要。 + # source_turn_id / ply_start / ply_end / session_ref を新しい + # 位置・スキームへその場で書き換える。id/canonical_exchange_id は + # 変更しない — code_touches / exchange_files / palace_objects / + # vec_exchanges からの exchange_id 参照はそのまま有効であり、以後の + # 重複判定は本関数の事前フィルタのみで行われるため書き換え不要。 + # + # 同一セッション内で (user_content, agent_content) が完全一致する + # legacy exchange が複数存在し得る（同じ発話の繰り返し等）ため、 + # content キーごとに候補を list で保持し、元の ply_start 昇順で + # FIFO 消費する。old メッセージ同士の相対順序は新規メッセージの + # 挿入位置に関わらず保存される（time_created が不変なため）ので、 + # 現在のパース結果を ply_start 昇順で辿る順序と一致し、1対1で + # 正しく対応付けられる。単一 dict スロットだと2件目が1件目を + # 上書きし、シフト後に片方が永久に重複したまま残ってしまう。 existing_rows = con.execute( - "SELECT source_turn_id, user_content, agent_content " + "SELECT source_turn_id, ply_start, user_content, agent_content " "FROM exchanges WHERE harness = 'opencode' " "AND source_session_id = ?", (session_id,), ).fetchall() known_exchange_ids = {row["source_turn_id"] for row in existing_rows} - legacy_by_content = { - (row["user_content"], row["agent_content"]): row["source_turn_id"] - for row in existing_rows - if _is_legacy_opencode_turn_id(row["source_turn_id"]) - } + legacy_by_content: dict[tuple[str, str], list[sqlite3.Row]] = {} + for row in existing_rows: + if _is_legacy_opencode_turn_id(row["source_turn_id"]): + key = (row["user_content"], row["agent_content"]) + legacy_by_content.setdefault(key, []).append(row) + for candidates in legacy_by_content.values(): + candidates.sort(key=lambda row: row["ply_start"]) raw_entries, started_at = _load_opencode_raw_entries(src, session_id) exchanges = parse_opencode_exchanges( @@ -878,15 +889,25 @@ def index_opencode_db( for exchange in exchanges: if exchange.id in known_exchange_ids: continue - legacy_turn_id = legacy_by_content.pop( - (exchange.user_content, exchange.agent_content), None + candidates = legacy_by_content.get( + (exchange.user_content, exchange.agent_content) ) - if legacy_turn_id is not None: + if candidates: + legacy_row = candidates.pop(0) con.execute( - "UPDATE exchanges SET source_turn_id = ? " + "UPDATE exchanges SET source_turn_id = ?, " + "ply_start = ?, ply_end = ?, session_ref = ? " "WHERE harness = 'opencode' AND source_session_id = ? " "AND source_turn_id = ?", - (exchange.id, session_id, legacy_turn_id), + ( + exchange.id, + exchange.ply_start, + exchange.ply_end, + f"{source_path}#ply=" + f"{exchange.ply_start}-{exchange.ply_end}", + session_id, + legacy_row["source_turn_id"], + ), ) continue new_exchanges.append(exchange) diff --git a/tests/test_opencode_ingest_robustness.py b/tests/test_opencode_ingest_robustness.py index 199a051..3bccc2e 100644 --- a/tests/test_opencode_ingest_robustness.py +++ b/tests/test_opencode_ingest_robustness.py @@ -326,22 +326,19 @@ def test_db_path_with_special_uri_characters_is_opened_correctly(tmp_path: Path) -def _seed_legacy_opencode_exchange( +def _seed_legacy_opencode_exchanges( opencode_db: Path, db_path: Path, project_root: Path, session_id: str ) -> None: """このパッチ以前の index_opencode_db が生成していた永続化結果 （source_turn_id=str(ply_start)、数値の位置カーソル）を、実際に opencode_db を - 1回パースした結果から直接構築する。session_id には唯一の user メッセージが - 含まれる前提（他のテストの前提と合わせるため）。""" + 1回パースした結果から直接構築する。session_id 内の全 exchange を対象にする。""" src = sqlite3.connect(f"file:{opencode_db}?mode=ro", uri=True) src.row_factory = sqlite3.Row raw_entries, started_at = _load_opencode_raw_entries(src, session_id) src.close() source_path = f"{opencode_db}#{session_id}" exchanges = parse_opencode_exchanges(source_path, raw_entries, min_chars=1) - assert len(exchanges) == 1 - legacy_exchange = exchanges[0] - legacy_turn_id = str(legacy_exchange.ply_start) + assert len(exchanges) >= 1 con = get_connection(db_path) ingest_parse_result( @@ -354,7 +351,7 @@ def _seed_legacy_opencode_exchange( started_at=started_at, ), ParseResult( - exchanges=( + exchanges=tuple( CanonicalExchange( harness="opencode", session_ref=( @@ -362,16 +359,17 @@ def _seed_legacy_opencode_exchange( f"{legacy_exchange.ply_start}-{legacy_exchange.ply_end}" ), source_session_id=session_id, - source_turn_id=legacy_turn_id, + source_turn_id=str(legacy_exchange.ply_start), ply_start=legacy_exchange.ply_start, ply_end=legacy_exchange.ply_end, user_content=legacy_exchange.user_content, agent_content=legacy_exchange.agent_content, files_touched=tuple(legacy_exchange.files), git_branch=legacy_exchange.git_branch, - ), + ) + for legacy_exchange in exchanges ), - next_cursor=f"v1:ply:{legacy_exchange.ply_end}", + next_cursor=f"v1:ply:{exchanges[-1].ply_end}", ), ) con.commit() @@ -413,7 +411,7 @@ def test_upgrade_from_legacy_position_based_cursor_does_not_duplicate( db_path = project_root / ".codeatrium" / "memory.db" init_db(db_path) - _seed_legacy_opencode_exchange(opencode_db, db_path, project_root, "ses1") + _seed_legacy_opencode_exchanges(opencode_db, db_path, project_root, "ses1") con = get_connection(db_path) pre_upgrade_count = con.execute("SELECT COUNT(*) FROM exchanges").fetchone()[0] @@ -471,7 +469,7 @@ def test_upgrade_with_out_of_order_message_migrates_legacy_and_captures_new( init_db(db_path) # msg1 を旧スキーム（source_turn_id=str(ply_start)=="0"）で取り込み済みにする。 - _seed_legacy_opencode_exchange(opencode_db, db_path, project_root, "ses1") + _seed_legacy_opencode_exchanges(opencode_db, db_path, project_root, "ses1") # msg1 より古い time_created を持つ新規メッセージが到着する # （バックフィル・クロックスキュー等の実運用シナリオ）。(time_created, id) 順 @@ -503,4 +501,172 @@ def test_upgrade_with_out_of_order_message_migrates_legacy_and_captures_new( row[0] for row in con.execute("SELECT user_content FROM exchanges") ) con.close() - assert contents == sorted(["legacy " + "x" * 60, "new " + "y" * 60]) \ No newline at end of file + assert contents == sorted(["legacy " + "x" * 60, "new " + "y" * 60]) + + +def test_upgrade_with_duplicate_content_legacy_exchanges_does_not_leave_duplicate( + tmp_path: Path, +) -> None: + """同一セッション内に (user_content, agent_content) が完全一致する legacy + exchange が複数存在する場合でも、位置シフト後にどちらも重複登録されずに + それぞれ新スキームへ移行される（#48 レビュー round3 指摘: dict の単一スロットで + 2件目が1件目を上書きすると、片方が永久に重複したまま残ってしまう）。""" + project_root = tmp_path / "project" + project_root.mkdir() + opencode_db = tmp_path / "opencode.db" + + dup_text = "dup " + "x" * 60 + con = _connect(opencode_db) + con.execute( + "INSERT INTO project (id, worktree, vcs, name) VALUES (?, ?, ?, ?)", + ("proj1", str(project_root), "git", "repo"), + ) + con.execute( + "INSERT INTO session (id, project_id, directory) VALUES (?, ?, ?)", + ("ses1", "proj1", str(project_root)), + ) + # 2つの独立した user メッセージが、たまたま同じ本文を持つ + # （同じ発話の繰り返し等）。 + con.execute( + "INSERT INTO message (id, session_id, time_created, time_updated, data) " + "VALUES (?, ?, ?, ?, ?)", + _user_message("msg_a", "ses1", 1000), + ) + con.execute( + "INSERT INTO part (id, message_id, session_id, time_created, time_updated, data) " + "VALUES (?, ?, ?, ?, ?, ?)", + _text_part("prt_a", "msg_a", "ses1", 1001, dup_text), + ) + con.execute( + "INSERT INTO message (id, session_id, time_created, time_updated, data) " + "VALUES (?, ?, ?, ?, ?)", + _user_message("msg_b", "ses1", 1002), + ) + con.execute( + "INSERT INTO part (id, message_id, session_id, time_created, time_updated, data) " + "VALUES (?, ?, ?, ?, ?, ?)", + _text_part("prt_b", "msg_b", "ses1", 1003, dup_text), + ) + con.commit() + con.close() + + db_path = project_root / ".codeatrium" / "memory.db" + init_db(db_path) + + # 両方とも旧スキーム（source_turn_id=str(ply_start)）で取り込み済みにする。 + _seed_legacy_opencode_exchanges(opencode_db, db_path, project_root, "ses1") + + con = get_connection(db_path) + pre_upgrade_count = con.execute("SELECT COUNT(*) FROM exchanges").fetchone()[0] + con.close() + assert pre_upgrade_count == 2 + + # msg_a より古い time_created を持つ、内容の異なる新規メッセージが到着する。 + # 両方の legacy exchange の位置がシフトする。 + con = sqlite3.connect(opencode_db) + con.execute( + "INSERT INTO message (id, session_id, time_created, time_updated, data) " + "VALUES (?, ?, ?, ?, ?)", + _user_message("msg_new", "ses1", 500), + ) + con.execute( + "INSERT INTO part (id, message_id, session_id, time_created, time_updated, data) " + "VALUES (?, ?, ?, ?, ?, ?)", + _text_part("prt_new", "msg_new", "ses1", 501, "new " + "z" * 60), + ) + con.commit() + con.close() + + reindexed = index_opencode_db( + opencode_db, db_path, min_chars=1, project_root=project_root + ) + # 新規に登録されるのは msg_new の1件だけ。同一内容の2件の legacy exchange は + # どちらも重複登録されず、それぞれ正しく新スキームへ移行される。 + assert reindexed == 1 + + con = get_connection(db_path) + contents = sorted( + row[0] for row in con.execute("SELECT user_content FROM exchanges") + ) + total_rows = con.execute("SELECT COUNT(*) FROM exchanges").fetchone()[0] + con.close() + assert total_rows == 3 + assert contents == sorted([dup_text, dup_text, "new " + "z" * 60]) + + +def test_migrated_legacy_exchange_reflects_new_position_not_stale_one( + tmp_path: Path, +) -> None: + """legacy exchange が新スキームへ移行される際、ply_start/ply_end/session_ref も + その exchange の現在の実際の位置に更新される（アップグレード前の位置に + 取り残されない）。#48 レビュー round3 指摘: これを放置すると context の順序や + verbatim_ref の解決が誤った位置を指してしまう。""" + project_root = tmp_path / "project" + project_root.mkdir() + opencode_db = tmp_path / "opencode.db" + + con = _connect(opencode_db) + con.execute( + "INSERT INTO project (id, worktree, vcs, name) VALUES (?, ?, ?, ?)", + ("proj1", str(project_root), "git", "repo"), + ) + con.execute( + "INSERT INTO session (id, project_id, directory) VALUES (?, ?, ?)", + ("ses1", "proj1", str(project_root)), + ) + con.execute( + "INSERT INTO message (id, session_id, time_created, time_updated, data) " + "VALUES (?, ?, ?, ?, ?)", + _user_message("msg1", "ses1", 1000), + ) + con.execute( + "INSERT INTO part (id, message_id, session_id, time_created, time_updated, data) " + "VALUES (?, ?, ?, ?, ?, ?)", + _text_part("prt1", "msg1", "ses1", 1001, "legacy " + "x" * 60), + ) + con.commit() + con.close() + + db_path = project_root / ".codeatrium" / "memory.db" + init_db(db_path) + + # msg1 は ply_start=0, ply_end=1 の legacy exchange として取り込み済み。 + _seed_legacy_opencode_exchanges(opencode_db, db_path, project_root, "ses1") + con = get_connection(db_path) + before = con.execute( + "SELECT ply_start, ply_end, session_ref FROM exchanges " + "WHERE user_content = ?", + ("legacy " + "x" * 60,), + ).fetchone() + con.close() + assert (before["ply_start"], before["ply_end"]) == (0, 1) + assert before["session_ref"] == f"{opencode_db}#ses1#ply=0-1" + + # msg1 より古い time_created を持つ新規メッセージが到着し、msg1 の実際の位置は + # 0-1 から 2-3 へシフトする。 + con = sqlite3.connect(opencode_db) + con.execute( + "INSERT INTO message (id, session_id, time_created, time_updated, data) " + "VALUES (?, ?, ?, ?, ?)", + _user_message("msg0", "ses1", 500), + ) + con.execute( + "INSERT INTO part (id, message_id, session_id, time_created, time_updated, data) " + "VALUES (?, ?, ?, ?, ?, ?)", + _text_part("prt0", "msg0", "ses1", 501, "new " + "y" * 60), + ) + con.commit() + con.close() + + index_opencode_db(opencode_db, db_path, min_chars=1, project_root=project_root) + + con = get_connection(db_path) + after = con.execute( + "SELECT ply_start, ply_end, session_ref FROM exchanges " + "WHERE user_content = ?", + ("legacy " + "x" * 60,), + ).fetchone() + con.close() + # 移行後は stale な 0-1 ではなく、実際の新しい位置 2-3 を指す。 + assert (after["ply_start"], after["ply_end"]) == (2, 3) + assert after["session_ref"] == f"{opencode_db}#ses1#ply=2-3" \ No newline at end of file