From 3a918d9e114c8085847ba0f08d779d17509bc694 Mon Sep 17 00:00:00 2001 From: Fiber Date: Wed, 11 Nov 2026 01:22:28 +0800 Subject: [PATCH] fix(learning): keep runtime vectors out of dropped batch items A memory dropped because its derived source changed during the model call still carried the numpy embedding, so writing the batch result raised TypeError and failed the whole batch. Strip the vector in one helper used by both _apply paths that can see it. The other dropped paths run before embeddings are attached or hold no memory item. ARCHITECTURE: maintenance_batch_retries now retries on days 1, 3 and 7. --- ARCHITECTURE.md | 2 +- src/iris/learning.py | 11 +++- tests/test_learning_dropped_serialization.py | 62 ++++++++++++++++++++ 3 files changed, 71 insertions(+), 4 deletions(-) create mode 100644 tests/test_learning_dropped_serialization.py diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 8e4485d5..dc0897d4 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -180,7 +180,7 @@ memory_ops 的人工正文编辑与删除都核对预期修订号，并在同一 ## M2 生命周期与确定性维护 -迁移 008 添加 maintenance_runs（阶段游标与固定大小的汇总计数）／maintenance_items（仅实际变更与失败）、memory_dependency_losses（每对来源的一次性扣减）、maintenance_batch_retries（每批一次）、batch_message_refs（冻结 JSON 段的引用索引），扩展 admin_operations 的对象字段，并在 memories 保存中等重要度的衰减余数和 purged_at。升级回填批次引用、当前失效依据及人工修订操作标识；在事务内延迟外键检查并原样重建 messages 为 AUTOINCREMENT，防止已清理消息的 ID 被复用。原始记忆与检索排序保持原样。 +迁移 008 添加 maintenance_runs（阶段游标与固定大小的汇总计数）／maintenance_items（仅实际变更与失败）、memory_dependency_losses（每对来源的一次性扣减）、maintenance_batch_retries（迁移 036 起每批在第 1、3、7 天各一次）、batch_message_refs（冻结 JSON 段的引用索引），扩展 admin_operations 的对象字段，并在 memories 保存中等重要度的衰减余数和 purged_at。升级回填批次引用、当前失效依据及人工修订操作标识；在事务内延迟外键检查并原样重建 messages 为 AUTOINCREMENT，防止已清理消息的 ID 被复用。原始记忆与检索排序保持原样。 `adjust_retention(store, id, delta=0, *, value=None, floor=None, expected_revision=None, current=None, settings=None, _conn=None)` 接受原子增量、绝对值或下限；有 _conn 时加入调用者的短事务，否则自行开启 Store.write。缺失／已删除／修订不符返回 None；成功返回最新行。截断、双阈值与 forgotten_at 同事务，不加修订号，不写日常修订。`confirm_retention(..., _conn=conn)` 读取 confirmation_increment 并更新 last_confirmed_at；学习已在 confirm_once 中调用，confirmed_once 保证每批一次，默认 +5；显式确认与写入去重共用同一事务和集合。在途记忆仅转为遗忘且修订号不变时仍可确认，修正／反驳继续要求有效状态。反馈、管理员强度、维护衰减及 M11 已共用此接口。 diff --git a/src/iris/learning.py b/src/iris/learning.py index e5eed21b..9797bbee 100644 --- a/src/iris/learning.py +++ b/src/iris/learning.py @@ -58,6 +58,12 @@ def _validate_output_shape(output: Any, raw: str) -> None: raise ValueError("invalid learning output shape (" + "; ".join(errors) + ")") +def _dropped_memory(item: dict[str, Any], reason: str) -> dict[str, Any]: + # The batch result is JSON; the runtime vector attached before _apply is not. + return {"section": "memories", "reason": reason, + "item": {key: value for key, value in item.items() if key != "_embedding"}} + + def _score(value: Any, default: int) -> int: try: return max(0, min(100, round(float(value)))) @@ -781,7 +787,7 @@ def self_output_only(memory: Any, evidence: list[int]) -> bool: "SELECT 1 FROM memories WHERE id=? AND revision=? AND lifecycle='active'", (m['id'],m['revision'] + (m['id'] in updated))).fetchone() for m in derived) if stale: - dropped.append({'section':'memories','item':item,'reason':'derived memory changed or invisible'}) + dropped.append(_dropped_memory(item, 'derived memory changed or invisible')) continue incoming_scope = visibility.evidence(item['evidence'], [m['id'] for m in derived]) speaker_id = canonical_subject(conn, item["speaker_id"]) @@ -818,8 +824,7 @@ def same_meaning(memory: Any) -> bool: if duplicate: if duplicate["id"] not in created: if self_output_only(duplicate, item["evidence"]): - dropped.append({"section": "memories", "reason": SELF_ONLY_CONFIRMATION, - "item": {k: v for k, v in item.items() if k != "_embedding"}}) + dropped.append(_dropped_memory(item, SELF_ONLY_CONFIRMATION)) else: confirm_once(duplicate["id"]) for message_id in item["evidence"]: diff --git a/tests/test_learning_dropped_serialization.py b/tests/test_learning_dropped_serialization.py new file mode 100644 index 00000000..2b0258c4 --- /dev/null +++ b/tests/test_learning_dropped_serialization.py @@ -0,0 +1,62 @@ +"""Dropped items are saved in the batch result, so runtime vectors must stay out of them.""" +import json + +from conftest import FakeGateway, msg +from iris.learning import LearningEngine, PROMPT_VERSION +from iris.models import ModelConfig +from iris.queue import form_batch +from test_batches import memory + + +def run(store, output, hook=None): + formed = form_batch(store, "A", PROMPT_VERSION, target_count=1, future_count=2, history_count=2) + gateway = FakeGateway(output, hook=hook) + gateway.configs["embedding"] = ModelConfig("fake", "", "fake-embedding") + return formed, LearningEngine(store, gateway).run_batch(formed.id, force=True) + + +def stored_result(store, batch_id): + with store.read() as conn: + row = conn.execute("SELECT state,result_json FROM batches WHERE id=?", (batch_id,)).fetchone() + return row["state"], json.loads(row["result_json"]) + + +def assert_no_runtime_fields(result): + assert all("_embedding" not in item["item"] for item in result["dropped"] if isinstance(item["item"], dict)) + + +def test_stale_derived_memory_drop_does_not_fail_the_batch(store): + msg(store, 1, "小林养了一只猫") + _, seeded = run(store, {"memories": [memory()]}) + source = seeded["created"][0] + msg(store, 2, "小林最近常去宠物店") + + def forget_source(_): + with store.write() as conn: + conn.execute("UPDATE memories SET lifecycle='forgotten' WHERE id=?", (source,)) + + output = {"memories": [{**memory("小林常去宠物店", [2]), "derived_from": ["M1"]}]} + # The material must offer M1 for the derived reference to pass validation. + formed, result = run(store, output, hook=forget_source) + + assert result["created"] == [] + assert [item["reason"] for item in result["dropped"]] == ["derived memory changed or invisible"] + assert_no_runtime_fields(result) + state, stored = stored_result(store, formed.id) + assert state == "succeeded" + assert stored["dropped"][0]["item"]["content"] == "小林常去宠物店" + assert_no_runtime_fields(stored) + + +def test_batch_limit_drops_stay_serializable_with_embeddings(store): + msg(store, 1, "小林列了很多爱好") + output = {"memories": [memory(f"小林喜欢第{i}种爱好") for i in range(22)]} + + formed, result = run(store, output) + + assert len(result["created"]) == 20 + assert [item["reason"] for item in result["dropped"]] == ["batch memory limit 20"] * 2 + assert_no_runtime_fields(result) + state, stored = stored_result(store, formed.id) + assert state == "succeeded" + assert_no_runtime_fields(stored)