Preserve natural claim follow-up reasoning

hermes-agent/leoclean-bin/kb_tool.py

@@ -581,10 +581,25 @@ def operational_contracts(
         term in lowered
         for term in ("proposal", "proposed", "pending", "approved", "applied", "staged", "reviewer approval")
     )
+    claim_reasoning_question = bool(
+        any(term in lowered for term in ("claim", "belief", "exact body"))
+        and any(
+            term in lowered
+            for term in (
+                "assumption",
+                "evidence",
+                "falsif",
+                "observed support",
+                "replacement",
+                "narrower",
+                "what supports",
+            )
+        )
+    )
     broad_kb_change_question = _has_unnegated_action(
         query, ("change", "changed", "update", "updated", "landed")
     )
-    proposal_state_question = kb_scope and not forecast_resolution_question and (
+    proposal_state_question = kb_scope and not forecast_resolution_question and not claim_reasoning_question and (
         proposal_lifecycle_question
         or ("demo" not in lowered and broad_kb_change_question and not kb_implementation_question)
     )

scripts/run_leo_direct_claim_handler_suite.py

@@ -58,11 +58,13 @@ ASSIGNMENT_SECRET_PATTERN = re.compile(
 DB_CONTEXT_PLUGIN = (
     ROOT / "hermes-agent" / "leoclean-plugins" / "vps" / "leo-db-context" / "__init__.py"
 )
+KB_TOOL = ROOT / "hermes-agent" / "leoclean-bin" / "kb_tool.py"
 BEHAVIOR_MANIFEST = ROOT / "scripts" / "leo_behavior_manifest.py"
 
 
 REMOTE_SCRIPT = r'''
 import asyncio
+import copy
 import hashlib
 import json
 import os
@@ -86,6 +88,7 @@ PROMPTS = __PROMPTS_JSON__
 REPORT_PREFIX = __REPORT_PREFIX_JSON__
 REPORT_PATH = Path(f"/tmp/{REPORT_PREFIX}-{RUN_ID}.json")
 DB_CONTEXT_PLUGIN_SOURCE = __DB_CONTEXT_PLUGIN_SOURCE_JSON__
+KB_TOOL_SOURCE = __KB_TOOL_SOURCE_JSON__
 BEHAVIOR_MANIFEST_SOURCE = __BEHAVIOR_MANIFEST_SOURCE_JSON__
 TURN_TRACE_PLUGIN_SOURCE = """\
 import hashlib
@@ -410,6 +413,11 @@ def install_harness_plugins(profile):
     db_context.parent.mkdir(parents=True, exist_ok=True)
     db_context.write_text(DB_CONTEXT_PLUGIN_SOURCE, encoding="utf-8")
 
+    kb_tool = profile / "bin" / "kb_tool.py"
+    kb_tool.parent.mkdir(parents=True, exist_ok=True)
+    kb_tool.write_text(KB_TOOL_SOURCE, encoding="utf-8")
+    kb_tool.chmod(0o700)
+
     turn_trace = profile / "plugins" / "leo-turn-execution-trace"
     turn_trace.mkdir(parents=True, exist_ok=True)
     (turn_trace / "__init__.py").write_text(TURN_TRACE_PLUGIN_SOURCE, encoding="utf-8")
@@ -483,7 +491,7 @@ def current_agent_messages(runner, session_key):
         return []
     agent = cached[0]
     messages = getattr(agent, "_session_messages", None) or []
-    return [dict(message) for message in messages if isinstance(message, dict)]
+    return [copy.deepcopy(message) for message in messages if isinstance(message, dict)]
 
 
 def conversation_trace(messages):
@@ -531,6 +539,7 @@ def conversation_trace_rows(messages):
         content_text = content if isinstance(content, str) else json.dumps(content, sort_keys=True, default=str)
         tool_calls = message.get("tool_calls") or []
         rows.append({
+            "role": message.get("role"),
             "content_sha256": hashlib.sha256(content_text.encode("utf-8")).hexdigest(),
             "content_bytes": len(content_text.encode("utf-8")),
             "tool_calls_sha256": hashlib.sha256(
@@ -654,7 +663,9 @@ async def run_suite():
             reply = await asyncio.wait_for(runner._handle_message(event), timeout=300)
             ended = datetime.now(timezone.utc)
             messages_after = current_agent_messages(runner, session_key)
-            history_prefix_preserved = messages_after[: len(messages_before)] == messages_before
+            before_rows = conversation_trace_rows(messages_before)
+            after_rows = conversation_trace_rows(messages_after)
+            history_prefix_preserved = after_rows[: len(before_rows)] == before_rows
             new_messages = messages_after[len(messages_before) :] if history_prefix_preserved else messages_after
             database_tool_trace = extract_kb_tool_trace(new_messages)
             results.append(
@@ -806,6 +817,7 @@ def build_remote_script(
         .replace("__REPORT_PREFIX_JSON__", json.dumps(report_prefix))
         .replace("__PROMPT_NOTE_JSON__", json.dumps(prompt_note))
         .replace("__DB_CONTEXT_PLUGIN_SOURCE_JSON__", json.dumps(DB_CONTEXT_PLUGIN.read_text(encoding="utf-8")))
+        .replace("__KB_TOOL_SOURCE_JSON__", json.dumps(KB_TOOL.read_text(encoding="utf-8")))
         .replace("__BEHAVIOR_MANIFEST_SOURCE_JSON__", json.dumps(BEHAVIOR_MANIFEST.read_text(encoding="utf-8")))
     )
 

tests/test_hermes_leoclean_db_context_plugin.py

@@ -486,6 +486,7 @@ def test_plugin_fails_closed_on_missing_snapshot_or_invalid_compiler() -> None:
 def test_handler_harness_retains_database_context_proof_fields() -> None:
     source = (ROOT / "scripts" / "run_leo_direct_claim_handler_suite.py").read_text(encoding="utf-8")
     assert "LEO_DB_CONTEXT_TRACE_PATH" in source
+    assert "KB_TOOL_SOURCE" in source
     assert '"database_context_trace"' in source
     assert '"database_context_injection_count"' in source
     assert '"database_context_all_ok"' in source
@@ -503,6 +504,8 @@ def test_handler_harness_retains_database_context_proof_fields() -> None:
     assert '"model_response_trace"' in source
     assert '"conversation_before"' in source
     assert '"conversation_after"' in source
+    assert "copy.deepcopy(message)" in source
+    assert "after_rows[: len(before_rows)] == before_rows" in source
     assert "attach_execution_manifests" in source
     assert "report_passes_safety" in source
     assert "json.loads(redact(candidate))" in source

tests/test_hermes_leoclean_kb_bridge_source.py

@@ -846,6 +846,16 @@ def test_vps_bridge_direct_intents_reject_unrelated_and_resolve_overlaps() -> No
     assert v3_ids & direct_ids == {"proposal_state_readback"}
     assert "source_link_audit" not in v3_ids
 
+    reasoning_followup_ids = {
+        item["id"]
+        for item in module.operational_contracts(
+            "Inspect the exact claim body and evidence: what is assumption versus observed support, what would "
+            "falsify it, and which replacement claims would you put into review? Keep them as proposals, not live "
+            "knowledge."
+        )
+    }
+    assert reasoning_followup_ids & direct_ids == set()
+
 
 def test_vps_bridge_named_document_question_matches_exact_proposal_and_compiles_receipt(monkeypatch) -> None:
     module = _load_module(BRIDGE_DIR / "kb_tool.py")