fix(LongTermMemory): enforce stored context budgets before publication

plugins/KnowledgeEngine/LongTermMemory/README.md

@@ -652,4 +652,4 @@ Existing historical summary records without lineage cannot be retroactively trac
 Document deletion invalidates dependent summaries and pending drafts for every affected memory scope before removing imported records. Deleting a knowledge base clears only that collection's pending drafts. Preview generation and collection deletion share the same collection lock, so an in-flight model preview cannot save source text after deletion completes; a preview for a removed KB is rejected before calling the model. Each model call holding that lock is capped at 30 seconds; queued callers and storage operations still follow their normal API lifecycle.
 
 
-A reviewed summary must also fit the bounded recall record budget before publication; oversized summaries are rejected without writing an episode or hiding source facts. Reduce the consolidation candidate count and preview again. This is a character/serialization bound, not an exact token estimate.
+A reviewed summary must also fit the bounded recall record budget before publication; oversized summaries are rejected before activation, with source facts retained. If Host-added metadata takes a staged summary over budget, the staged record is removed. Reduce the consolidation candidate count and preview again. This is a character/serialization bound, not an exact token estimate.

plugins/KnowledgeEngine/LongTermMemory/store/consolidation.py

@@ -220,6 +220,10 @@ async def apply_summary(store, collection_id: str, embedding_model_uuid: str, us
     written = await store.get_episode_by_id(collection_id, episode_id, user_key)
     if written is None or any(written["metadata"].get(key) != value for key, value in meta.items()):
         raise RuntimeError("Consolidation draft could not be verified; source memories were retained")
+    stored_record = {"id": episode_id, "metadata": written["metadata"], "document": written["content"], "score": None, "distance": 0.0}
+    if not select_memories([stored_record], "", top_k=1):
+        await store.delete_episode_by_id(collection_id, episode_id, user_key)
+        raise ValueError("Stored summary exceeds the recall budget; source memories were retained")
     await store._write_json(proposal_key(collection_id, user_key), {}, collection_id)
     summary = await store.update_episode_status(
         collection_id, embedding_model_uuid, episode_id, user_key, store.EPISODE_STATUS_ACTIVE

plugins/KnowledgeEngine/LongTermMemory/tests/store/test_semantic_consolidation.py

@@ -362,3 +362,20 @@ async def test_summary_too_large_for_recall_is_not_published():
         await store.apply_consolidation("kb", "emb", "scope", proposal_id=draft["proposal_id"])
     assert set(plugin.records) == {"one", "two"}
     assert all(item["metadata"]["status"] == "active" for item in plugin.records.values())
+
+
+@pytest.mark.asyncio
+async def test_host_metadata_cannot_make_published_summary_exceed_budget():
+    class VerboseHostPlugin(SemanticPlugin):
+        async def vector_upsert(self, *args, **kwargs):
+            kwargs["metadata"] = [{**metadata, "_backend_annotation": "x" * 8000} for metadata in kwargs["metadata"]]
+            await super().vector_upsert(*args, **kwargs)
+
+    plugin = VerboseHostPlugin()
+    add_source(plugin, "source", "A fact that must remain searchable")
+    store = MemoryStore(plugin)
+    draft = await preview(store)
+    with pytest.raises(ValueError, match="Stored summary exceeds"):
+        await store.apply_consolidation("kb", "emb", "scope", proposal_id=draft["proposal_id"])
+    assert set(plugin.records) == {"source"}
+    assert plugin.records["source"]["metadata"]["status"] == "active"

tests/e2e/test_longterm_memory_processor.py

@@ -390,4 +390,31 @@ def test_local_agent_memory_tools_without_pipeline(
         )
         assert replay.get('error'), replay
 
+        remaining = next(item for item in source_memories if item['id'] != original_id)
+        model.queue_llm_responses(
+            json.dumps({'facts': [{'text': remaining['content'], 'source_ids': [remaining['id']]}]})
+        )
+        pending = await page_api('/consolidation/preview', body)
+        assert pending['proposal_id']
+        from langbot.pkg.entity.persistence.bstorage import BinaryStorage
+        import sqlalchemy
+
+        async def proposal_values():
+            result = await ap.persistence_mgr.execute_async(
+                sqlalchemy.select(BinaryStorage.value).where(
+                    BinaryStorage.workspace_uuid == context.workspace_uuid,
+                    BinaryStorage.owner == 'langbot-team/LongTermMemory',
+                    BinaryStorage.key.like('ltm.consolidation.proposal.%'),
+                )
+            )
+            return [json.loads(row[0]) for row in result.all()]
+
+        assert any(value.get('id') == pending['proposal_id'] for value in await proposal_values())
+        await ap.knowledge_service.delete_knowledge_base(context, memory_kb_id)
+        assert all(value == {} for value in await proposal_values())
+        removed = await ap.plugin_connector.handle_page_api(
+            'langbot-team', 'LongTermMemory', 'memory_console', '/consolidation/preview', 'POST', body
+        )
+        assert removed.get('error'), removed
+
     local._run_local_agent_probe(local_agent_e2e_tmpdir, read_after_restart)