From ceb90af9af6b54a24f7c6af8b889c5c4a75ae83c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=85=B1=E7=89=9B=E8=82=89?= Date: Mon, 26 Oct 2026 20:26:55 +0800 Subject: [PATCH] Protect pending procedure writes and version segmentation checkpoints --- plugins/procedure-memory/README.md | 2 +- plugins/procedure-memory/README_ZH.md | 2 +- .../src/reme_procedure_memory/auto_memory.py | 3 +- .../src/reme_procedure_memory/store.py | 8 ++ .../src/reme_procedure_memory/token_budget.py | 28 ++++-- .../src/reme_procedure_memory/trajectory.py | 20 +++-- .../procedure-memory/tests/test_recovery.py | 85 +++++++++++++++++++ .../tests/test_token_budget.py | 4 + .../procedure-memory/tests/test_trajectory.py | 36 ++++++++ 9 files changed, 171 insertions(+), 17 deletions(-) diff --git a/plugins/procedure-memory/README.md b/plugins/procedure-memory/README.md index 9183760a..4463cfb3 100644 --- a/plugins/procedure-memory/README.md +++ b/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: diff --git a/plugins/procedure-memory/README_ZH.md b/plugins/procedure-memory/README_ZH.md index 8dd33166..b82cf9f2 100644 --- a/plugins/procedure-memory/README_ZH.md +++ b/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 的其他写入进程后，可显式重建： diff --git a/plugins/procedure-memory/src/reme_procedure_memory/auto_memory.py b/plugins/procedure-memory/src/reme_procedure_memory/auto_memory.py index 635dfac7..22e8712e 100644 --- a/plugins/procedure-memory/src/reme_procedure_memory/auto_memory.py +++ b/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. diff --git a/plugins/procedure-memory/src/reme_procedure_memory/store.py b/plugins/procedure-memory/src/reme_procedure_memory/store.py index 571663d8..310715ee 100644 --- a/plugins/procedure-memory/src/reme_procedure_memory/store.py +++ b/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 diff --git a/plugins/procedure-memory/src/reme_procedure_memory/token_budget.py b/plugins/procedure-memory/src/reme_procedure_memory/token_budget.py index e79c1c71..356e0948 100644 --- a/plugins/procedure-memory/src/reme_procedure_memory/token_budget.py +++ b/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 diff --git a/plugins/procedure-memory/src/reme_procedure_memory/trajectory.py b/plugins/procedure-memory/src/reme_procedure_memory/trajectory.py index 7dc9ae9d..4d31c2d4 100644 --- a/plugins/procedure-memory/src/reme_procedure_memory/trajectory.py +++ b/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} diff --git a/plugins/procedure-memory/tests/test_recovery.py b/plugins/procedure-memory/tests/test_recovery.py index 31ad701f..509a736a 100644 --- a/plugins/procedure-memory/tests/test_recovery.py +++ b/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 diff --git a/plugins/procedure-memory/tests/test_token_budget.py b/plugins/procedure-memory/tests/test_token_budget.py index 9e57d89e..9879d22c 100644 --- a/plugins/procedure-memory/tests/test_token_budget.py +++ b/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(): diff --git a/plugins/procedure-memory/tests/test_trajectory.py b/plugins/procedure-memory/tests/test_trajectory.py index 09dc3227..3a7e2680 100644 --- a/plugins/procedure-memory/tests/test_trajectory.py +++ b/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",