Protect pending procedure writes and version segmentation checkpoints

plugins/procedure-memory/README.md

@@ -81,7 +81,7 @@ The operation journal records the destination, target content and stable scoped
 
 Failed files remain intact and are temporarily excluded from **both** search views. Startup and unrelated trajectories continue. Search returns the available experiences with `index_complete=false` and a `pending_indexes` list containing the excluded paths, versions and errors, rather than silently returning stale records. A corrected file re-enters both views after synchronization. A segment with its own pending writes cannot advance; resume lets the agent run again to correct them. Structural index failures that prevent safe exclusion still fail explicitly.
 
-Checkpoints save segment position, bounded summary, processing fingerprint, recoverable input and final reply. A resumed segment receives its prior operation ledger so it can read completed experiences and continue; it does not assume the model will repeat identical decisions. A saved final reply is reused if the segment checkpoint failed to advance. Restart reconciles pending writes and rebuilds both views from experience JSON. Rebuilds never delete or rewrite experience or trajectory files.
+Checkpoints save segment position, bounded summary, processing fingerprint, recoverable input and final reply. The fingerprint includes the segmentation algorithm version; finish runs created by the previous segmentation algorithm with that plugin version before upgrading. Incompatible checkpoints fail explicitly without changing source files or progress. Long lines use bounded, progressively larger tokenization windows instead of repeatedly encoding the remaining line. A resumed segment receives its prior operation ledger so it can read completed experiences and continue; it does not assume the model will repeat identical decisions. A saved final reply is reused if the segment checkpoint failed to advance. Restart reconciles pending writes and rebuilds both views from experience JSON. Rebuilds never delete or rewrite experience or trajectory files.
 
 For explicit maintenance, with other workspace writers stopped:
 

plugins/procedure-memory/README_ZH.md

@@ -83,7 +83,7 @@ auto-memory 获得 `procedure_search`、`procedure_read`、`procedure_create`、
 
 失败经验的 JSON 保留，并暂时从**两个**搜索视图排除；启动及其他轨迹仍可继续。检索返回可用经验，同时明确设置 `index_complete=false`，通过 `pending_indexes` 列出排除的路径、版本与错误，避免静默返回过期记录。修正并完成同步后重新进入两个视图。当前段若仍有待同步写入，不推进检查点；续跑会再次调用 agent 进行修正。若底层索引故障导致无法安全排除记录，仍明确报错。
 
-检查点保存段位置、摘要、配置指纹、提炼输入及最终响应。续跑将已有操作记录交给 agent，以读取已写经验后继续；若最终响应已保存但段位置未推进，复用响应。启动时恢复未完成操作，并从经验 JSON 重建索引。维护操作不改写或删除原始轨迹和经验。
+检查点保存段位置、摘要、配置指纹、提炼输入及最终响应。指纹包含分段算法版本；旧分段算法创建的未完成任务须先使用原插件版本完成，再升级。不兼容检查点明确报错，不改动原文件或进度。超长单行通过逐步扩大的局部窗口计算 token，不再反复编码剩余全文。续跑将已有操作记录交给 agent，以读取已写经验后继续；若最终响应已保存但段位置未推进，复用响应。启动时恢复未完成操作，并从经验 JSON 重建索引。维护操作不改写或删除原始轨迹和经验。
 
 停止同 workspace 的其他写入进程后，可显式重建：
 

plugins/procedure-memory/src/reme_procedure_memory/auto_memory.py

@@ -10,7 +10,7 @@ from .checkpoint import atomic_json, digest, json_text, load_json
 from .schema import ProcessingCheckpoint, SegmentResult
 from .search import ProcedureStep
 from .store import ProcedureIndexError
-from .trajectory import prepare_text, segments
+from .trajectory import SEGMENTATION_VERSION, prepare_text, segments
 
 TOOLS = ["procedure_search", "procedure_read", "procedure_create", "procedure_update"]
 
@@ -50,6 +50,7 @@ class ProcedureAutoMemoryStep(ProcedureStep):
             json_text(
                 {
                     "version": 1,
+                    "segmentation": SEGMENTATION_VERSION,
                     "tokenizer": store.budget.source,
                     # Raising an execution context budget must remain a valid
                     # way to resume; it does not alter the saved segmentation.

plugins/procedure-memory/src/reme_procedure_memory/store.py

@@ -173,6 +173,14 @@ class ProcedureMemoryStore(BaseComponent):
                 record = load_json(operation_path)
                 if record["signature"] != signature:
                     raise ValueError("operation_key already belongs to different arguments; read completed operations")
+                if record["status"] not in {"indexed", "superseded"}:
+                    raise ProcedureIndexError(
+                        {
+                            **self._operation_result(record),
+                            "status": "index_pending",
+                            "error": record.get("index_error", "Procedure indexes are still pending recovery"),
+                        },
+                    )
                 return self._operation_result(record)
             record = self._prepare_mutation(action, arguments, scope, signature)
             # Persist the replacement's dependency list before touching either

plugins/procedure-memory/src/reme_procedure_memory/token_budget.py

@@ -45,19 +45,31 @@ class TokenBudget:
         if exceeds:
             raise ValueError(f"{field} has {count} tokens; maximum is {self.limits[field]}. Rewrite it more concisely.")
 
-    def prefix(self, text: str, limit: int) -> int:
-        """Find a nonempty original-text prefix within a token budget."""
-        if self.count(text) <= limit:
-            return len(text)
-        low, high = 0, len(text)
+    def prefix(self, text: str, limit: int, *, start: int = 0) -> int:
+        """Find a bounded prefix at start, probing progressively larger windows."""
+        available = len(text) - start
+        if not available:
+            return 0
+        low, high = 0, min(available, limit)
+        while True:
+            count = self.count(text[start : start + high])
+            if count > limit:
+                high -= 1
+                break
+            low = high
+            if high == available:
+                return high
+            # An exact fit commonly occurs in logs. One extra character can
+            # establish the boundary without another full-sized search window.
+            if count == limit and self.count(text[start : start + high + 1]) > limit:
+                return high
+            high = min(available, high * 2)
         while low < high:
             middle = (low + high + 1) // 2
-            if self.count(text[:middle]) <= limit:
+            if self.count(text[start : start + middle]) <= limit:
                 low = middle
             else:
                 high = middle - 1
-        while low and self.count(text[:low]) > limit:
-            low -= 1
         if not low:
             raise ValueError("Token budget cannot hold one character")
         return low

plugins/procedure-memory/src/reme_procedure_memory/trajectory.py

@@ -13,6 +13,7 @@ from reme.steps.file_io._file_io import detect_file_encoding
 from .checkpoint import atomic_json, checked_path, digest, load_json
 
 SUPPORTED = {"", ".txt", ".md", ".markdown", ".json", ".jsonl", ".log"}
+SEGMENTATION_VERSION = 2
 
 
 def file_hash(path: Path, max_bytes: int) -> str:
@@ -129,15 +130,22 @@ def segments(path: Path, encoding: str, budget):
     limit = budget.limits["segment"]
     with path.open(encoding=encoding, errors="strict", newline="") as stream:
         for line in stream:
-            if pending and budget.count(pending + line) > limit:
+            if pending:
+                combined = pending + line
+                if budget.prefix(combined, limit) == len(combined):
+                    pending = combined
+                    continue
                 yield {"offset": offset, "text": pending}
                 offset += len(pending)
                 pending = ""
-            while budget.count(line) > limit:
-                end = budget.prefix(line, limit)
-                yield {"offset": offset, "text": line[:end]}
+            start = 0
+            while start < len(line):
+                end = budget.prefix(line, limit, start=start)
+                if start + end == len(line):
+                    pending = line[start:]
+                    break
+                yield {"offset": offset, "text": line[start : start + end]}
                 offset += end
-                line = line[end:]
-            pending += line
+                start += end
         if pending:
             yield {"offset": offset, "text": pending}

plugins/procedure-memory/tests/test_recovery.py

@@ -151,6 +151,34 @@ async def test_reply_saved_before_checkpoint_advances(app_factory, tmp_path, mon
     assert calls == 1
 
 
+@pytest.mark.asyncio
+async def test_segmentation_version_change_rejects_resume(app_factory, tmp_path, monkeypatch):
+    app, store = await app_factory()
+    source = tmp_path / "segmentation.log"
+    source.write_text("evidence")
+    calls = 0
+
+    async def reply(_inputs, _options):
+        nonlocal calls
+        calls += 1
+        return {"result": '{"summary":"unfinished","complete":false}'}
+
+    app.context.metadata["fake_reply"] = reply
+    job = app.context.jobs["procedure_auto_memory"]
+    arguments = {"file": str(source), "filename": "segmentation.log", "request_id": "segmentation"}
+    with monkeypatch.context() as patch:
+        patch.setattr(auto_memory, "SEGMENTATION_VERSION", auto_memory.SEGMENTATION_VERSION - 1)
+        assert not (await job(**arguments)).success
+    checkpoint_path = store.checkpoint_path("session/trajectories/segmentation.log")
+    checkpoint = checkpoint_path.read_bytes()
+    result = await job(**arguments)
+    assert not result.success
+    assert "configuration changed" in result.metadata["error"]
+    assert calls == 1
+    assert checkpoint_path.read_bytes() == checkpoint
+    assert store.source_path(result.metadata["source"]).read_text() == "evidence"
+
+
 @pytest.mark.asyncio
 async def test_incomplete_is_resumable_and_config_change_rejected(app_factory, tmp_path):
     app, store = await app_factory()
@@ -314,6 +342,63 @@ def reject_payload(monkeypatch):
     return install
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize("action", ["create", "update"])
+@pytest.mark.parametrize("restart", [False, True])
+async def test_repeated_pending_write_stays_failed_until_indexes_recover(
+    runtime,
+    experience,
+    reject_payload,
+    monkeypatch,
+    app_factory,
+    action,
+    restart,
+):
+    app, store, scope = runtime
+    if action == "create":
+        arguments = {**experience, "content": "unembeddable-payload"}
+    else:
+        original = await store.apply_mutation("create", experience, scope)
+        arguments = {
+            "path": original["path"],
+            "expected_version": original["version"],
+            "operation_key": "pending-update",
+            "changes": {"content": "unembeddable-payload"},
+        }
+    backend = type(store.views["when_to_use"].embedding_store)
+    get_embeddings = backend.get_embeddings
+    reject_payload(store.views["when_to_use"].embedding_store)
+    job = app.context.jobs[f"procedure_{action}"]
+    failed = await job(**arguments, _procedure_scope=scope)
+    assert not failed.success
+    if restart:
+        workspace = store.workspace_path
+        source = failed.metadata["path"]
+        saved_source = (await store.read_experience(source))["experience"]["sources"][0]
+        await app.close()
+        app, store = await app_factory(workspace=workspace)
+        scope = store.open_scope(saved_source, 0)
+        job = app.context.jobs[f"procedure_{action}"]
+    for _ in range(2):
+        repeated = await job(**arguments, _procedure_scope=scope)
+        assert not repeated.success
+        assert repeated.metadata["status"] == "index_pending"
+        assert repeated.metadata["path"] == failed.metadata["path"]
+        assert repeated.metadata["version"] == failed.metadata["version"]
+        assert repeated.metadata["error"]
+        assert "new operation_key" in repeated.metadata["repair"]
+        assert not (await store.search("repair", "when_to_use"))["index_complete"]
+    monkeypatch.setattr(backend, "get_embeddings", get_embeddings)
+    recovered = await job(**arguments, _procedure_scope=scope)
+    assert recovered.success, recovered.answer
+    assert recovered.metadata["status"] == "indexed"
+    assert (await job(**arguments, _procedure_scope=scope)).metadata == recovered.metadata
+    for view in store.views.values():
+        chunks = list(view.file_chunks.values())
+        assert len(chunks) == 1
+        assert chunks[0].metadata["version"] == recovered.metadata["version"]
+
+
 @pytest.mark.asyncio
 async def test_permanent_failure_can_be_corrected_without_blocking_other_files(runtime, experience, reject_payload):
     app, store, scope = runtime

plugins/procedure-memory/tests/test_token_budget.py

@@ -13,6 +13,10 @@ def test_exact_counter_and_boundaries():
     with pytest.raises(ValueError, match="maximum is 6"):
         budget.check("content", "你好!")
     assert budget.prefix("你好!", 6) == 2
+    assert budget.prefix("skip你好!", 6, start=4) == 2
+    assert budget.prefix("skip", 6, start=4) == 0
+    with pytest.raises(ValueError, match="cannot hold one character"):
+        budget.prefix("你", 1)
 
 
 def test_no_estimate_fallback():

plugins/procedure-memory/tests/test_trajectory.py

@@ -9,6 +9,42 @@ from reme_procedure_memory.token_budget import TokenBudget
 from reme_procedure_memory.trajectory import prepare_text, segments
 
 
+@pytest.mark.parametrize("size", [1024 * 1024, 4 * 1024 * 1024])
+@pytest.mark.parametrize("chars_per_token", [1, 4])
+def test_long_line_tokenization_is_bounded_by_segment_size(tmp_path, size, chars_per_token):
+    counted = 0
+    largest = 0
+
+    def count(text):
+        nonlocal counted, largest
+        counted += len(text)
+        largest = max(largest, len(text))
+        return (len(text) + chars_per_token - 1) // chars_per_token
+
+    text = "x" * size + "\r\nlast line\r\n"
+    path = tmp_path / "long.log"
+    path.write_bytes(text.encode())
+    budget = TokenBudget(counter=count)
+    chunks = list(segments(path, "utf-8", budget))
+    assert "".join(chunk["text"] for chunk in chunks) == text
+    assert all(0 < len(chunk["text"]) <= budget.limits["segment"] * chars_per_token for chunk in chunks)
+    assert [chunk["offset"] for chunk in chunks] == [
+        sum(len(previous["text"]) for previous in chunks[:n]) for n in range(len(chunks))
+    ]
+    assert largest <= budget.limits["segment"] * chars_per_token + 1
+    assert counted < len(text) * 8
+
+
+def test_multibyte_long_lines_remain_lossless(tmp_path):
+    text = "leading\r\n" + "你好🙂" * 1000 + "\r\ntrailing\n"
+    path = tmp_path / "unicode.log"
+    path.write_bytes(text.encode())
+    budget = TokenBudget(counter=lambda value: len(value.encode()), segment=63)
+    chunks = list(segments(path, "utf-8", budget))
+    assert "".join(chunk["text"] for chunk in chunks) == text
+    assert all(len(chunk["text"].encode()) <= 63 for chunk in chunks)
+
+
 @pytest.mark.asyncio
 @pytest.mark.parametrize(
     "filename,text",