fix: isolate trust resolution and recall depth

Author: Coding-Dev-Tools

engraphis/core/engine.py

@@ -518,7 +518,7 @@ def _resolve_and_store(self, content: str, *, text: str, vec: Optional[np.ndarra
                 record for record in claim_history
                 if record.valid_from is not None and record.valid_from <= valid_from
                 and (record.valid_to is None or valid_from < record.valid_to)
-                and (trusted_write or not provenance_is_trusted(record.provenance))
+                and provenance_is_trusted(record.provenance) == trusted_write
             ]
             if predecessors:
                 predecessor = max(
@@ -914,7 +914,7 @@ def _resolve_against_neighbors(self, text: str, vec: np.ndarray, *, workspace_id
             if (nrec and nrec.workspace_id == workspace_id and nrec.repo_id == repo_id
                     and nrec.scope == scope and nrec.mtype == mtype
                     and (scope != Scope.SESSION or nrec.session_id == session_id)
-                    and (trusted_write or not provenance_is_trusted(nrec.provenance))
+                    and provenance_is_trusted(nrec.provenance) == trusted_write
                     and (memory_matches_filter(nrec, flt)
                          or (current_fallback and nrec.expired_at is None
                              and nrec.valid_to is None))):
@@ -945,7 +945,7 @@ def _resolve_against_neighbors(self, text: str, vec: np.ndarray, *, workspace_id
             authoritative = [
                 record for record in claim_history
                 if memory_matches_filter(record, flt, at=valid_at)
-                and (trusted_write or not provenance_is_trusted(record.provenance))
+                and provenance_is_trusted(record.provenance) == trusted_write
             ]
             if not authoritative and valid_at is not None:
                 # A backfill before the first recorded version has no visible
@@ -957,7 +957,7 @@ def _resolve_against_neighbors(self, text: str, vec: np.ndarray, *, workspace_id
                     if record.expired_at is None
                     and record.valid_from is not None
                     and record.valid_from > valid_at
-                    and (trusted_write or not provenance_is_trusted(record.provenance))
+                    and provenance_is_trusted(record.provenance) == trusted_write
                 ]
                 if later:
                     authoritative = [min(

engraphis/core/recall.py

@@ -145,8 +145,8 @@ def recall(self, query: str, flt: Optional[SearchFilter] = None, *, k: int = 8,
         # vector indexes do not carry provenance. Over-fetch so external records cannot
         # crowd trusted evidence out of a grounded/adaptive context candidate set.
         prompt_only = bool(prompt_only or not include_untrusted)
-        arm_candidate_k = candidate_k if not prompt_only else min(
-            250, max(candidate_k, candidate_k * 4)
+        arm_candidate_k = candidate_k if not prompt_only else (
+            candidate_k + min(250, candidate_k * 3)
         )
         if config.vector:
             qvec = self.embedder.embed([query])[0]

tests/test_poisoning.py

@@ -338,3 +338,31 @@ def test_untrusted_write_cannot_resolve_or_link_to_trusted_memory():
     assert trusted["id"] in ordinary_ids
     assert external["id"] not in ordinary_ids
     assert {trusted["id"], external["id"]} <= inspection_ids
+
+
+def test_trusted_write_creates_an_approved_record_for_an_untrusted_duplicate():
+    eng, wid, rid = _engine()
+    external = eng.remember_with_resolution(
+        "Production releases deploy to the blue environment.",
+        workspace_id=wid,
+        repo_id=rid,
+        metadata={"provenance": {"source": "web", "trusted": False}},
+    )
+
+    approved = eng.remember_with_resolution(
+        "Production releases deploy to the blue environment.",
+        workspace_id=wid,
+        repo_id=rid,
+        metadata={"provenance": {"source": "human", "trusted": True}},
+    )
+
+    assert approved["op"] == "add"
+    assert approved["id"] != external["id"]
+    assert eng.store.get_memory(approved["id"]).provenance["trusted"] is True
+    ordinary_ids = {
+        chunk["id"] for chunk in eng.recall(
+            "Where do production releases deploy?", workspace_id=wid, repo_id=rid, k=10,
+        ).chunks
+    }
+    assert approved["id"] in ordinary_ids
+    assert external["id"] not in ordinary_ids

tests/test_recall.py

@@ -314,6 +314,31 @@ def test_lexical_recall_is_filtered_before_candidate_limit():
     assert [c["id"] for c in res.chunks] == [wanted]
 
 
+def test_prompt_overfetch_never_reduces_the_requested_candidate_depth():
+    store = Store(":memory:")
+    emb = DeterministicEmbedder(256)
+    index = NumpyVectorIndex(store)
+    requested: list[int] = []
+    original_search = index.search
+
+    def recording_search(query, k, filter=None):
+        requested.append(k)
+        return original_search(query, k, filter=filter)
+
+    index.search = recording_search
+    eng = RecallEngine(store, emb, index, IdentityReranker())
+    wid = store.get_or_create_workspace("w")
+    _add(store, emb, wid, None, "A sufficiently deep candidate set remains available.")
+
+    result = eng.recall(
+        "candidate depth", SearchFilter(workspace_id=wid), k=1, candidate_k=500,
+    )
+
+    assert result.candidate_k_requested == 500
+    assert result.candidate_k_used == 500
+    assert requested[0] == 750
+
+
 def test_graph_arm_does_not_match_entity_names_inside_other_words():
     from engraphis.core.interfaces import Edge, Node