teleo/teleo-infrastructure
8
0
Fork 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions 1

Harden broad Leo database reasoning contracts

Browse source
This commit is contained in:
twentyOne2x 2026-07-16 05:44:21 +02:00
parent 813d4cd91d
commit 3cbbbfaa18
6 changed files with 370 additions and 23 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

139
hermes-agent/leoclean-bin/kb_tool.py
View file

@ -484,7 +484,7 @@ def _has_unnegated_database_state_request(query: str) -> bool:
"?" in segment
or re.search(
r"\b(?:what|which|whether|status|state|readback|audit|inspect|check|show|list|changed|"
r"updated|approved|applied|pending|canonical|exists?|landed|live)\b",
r"updated|approved|applied|pending|canonical|exists?|landed|live|reviewers?|sign(?:ed)? off)\b",
segment,
)
)
@ -560,11 +560,19 @@ def operational_contracts(
normalized = re.sub(r"[-_/]+", " ", lowered)
schema = current_schema if current_schema is not None else CURRENT_PUBLIC_SCHEMA
constraints = current_constraints or {}
requested_word_limit_match = re.search(
r"\b(?:under|within|at most|no more than|maximum(?: of)?|max(?:imum)?(?: of)?)\s+"
r"(?P<words>\d{2,4})\s+words?\b",
normalized,
)
requested_word_limit = (
max(20, min(220, int(requested_word_limit_match.group("words")))) if requested_word_limit_match else 220
)
contracts: list[dict[str, Any]] = [
{
"id": "reply_budget",
"target_words": 170,
"hard_max_words": 220,
"target_words": min(170, requested_word_limit),
"hard_max_words": requested_word_limit,
"shape": (
"no intro; at most three compact bullets covering mapping, review/apply, and receipt/limitation; "
"omission is better than exceeding the budget"
@ -580,7 +588,10 @@ def operational_contracts(
for term in (
"document",
"attachment",
"attached",
"extracted text",
"provenance",
"source packet",
"source row",
"source_ref",
"source ref",
@ -616,7 +627,18 @@ def operational_contracts(
)
proposal_lifecycle_question = any(
term in lowered
for term in ("proposal", "proposed", "pending", "approved", "applied", "staged", "reviewer approval")
for term in (
"proposal",
"proposed",
"pending",
"approved",
"applied",
"staged",
"reviewer approval",
"reviewers signed off",
"reviewer signed off",
"sign-off",
)
)
claim_reasoning_question = bool(
any(term in lowered for term in ("claim", "belief", "exact body"))
@ -633,6 +655,31 @@ def operational_contracts(
)
)
)
claim_evidence_challenge_question = bool(
any(term in lowered for term in ("claim", "exact body", "claim body"))
and any(term in lowered for term in ("evidence", "source"))
and any(
term in lowered
for term in ("challenge", "unsupported", "narrower", "assumption", "what supports", "falsif")
)
)
if claim_evidence_challenge_question:
contracts.append(
{
"id": "claim_evidence_challenge",
"required_sections": [
"one retrieved canonical claim ID and exact claim body",
"one linked source or evidence ID and exact evidence excerpt",
"one inference not established by that evidence",
"one narrower evidence-bounded revision",
],
"boundary": (
"the claim body and evidence excerpt are distinct database values; do not paraphrase either "
"as if it were the other"
),
"mutation_policy": "read-only; proposed wording remains a candidate until separately staged and reviewed",
}
)
broad_kb_change_question = _has_unnegated_action(query, ("change", "changed", "update", "updated", "landed"))
proposal_state_question = (
kb_scope
@ -979,7 +1026,21 @@ def operational_contracts(
}
)
if any(term in lowered for term in ("restart", "prior session", "erased", "answer behavior", "database totals")):
if any(
term in lowered
for term in (
"restart",
"prior session",
"erased",
"answer behavior",
"database totals",
"fresh process launch",
"process launch",
"behavioral parity",
"blank session",
"persisted conversation state",
)
):
contracts.append(
{
"id": "runtime_persistence",
@ -1066,6 +1127,47 @@ def format_proposal_readback(proposal: dict[str, Any]) -> str:
)
def _bounded_exact_excerpt(value: Any, *, max_words: int = 45) -> str:
"""Return an exact prefix suitable for a compact, attributable response."""
text = str(value or "").strip()
words = list(re.finditer(r"\S+", text))
if len(words) <= max_words:
return text
return text[: words[max_words - 1].end()].rstrip()
def bind_claim_evidence_challenge(
contracts: list[dict[str, Any]],
claims: list[dict[str, Any]],
evidence: dict[str, list[dict[str, Any]]],
) -> None:
"""Bind the challenge contract to one exact retrieved claim/source pair."""
contract = next((item for item in contracts if item.get("id") == "claim_evidence_challenge"), None)
if contract is None:
return
for claim in claims:
claim_id = str(claim.get("id") or "")
claim_text = str(claim.get("text") or "").strip()
if not claim_id or not claim_text:
continue
for row in evidence.get(claim_id, []):
source_id = str(row.get("source_id") or "")
excerpt = _bounded_exact_excerpt(row.get("excerpt"))
if not source_id or not excerpt:
continue
contract["binding_status"] = "bound"
contract["selected_claim"] = {"id": claim_id, "text": claim_text}
contract["selected_evidence"] = {
"source_id": source_id,
"excerpt": excerpt,
"role": row.get("role"),
}
return
contract["binding_status"] = "unavailable"
def compile_operational_response(contracts: list[dict[str, Any]]) -> str | None:
"""Compile a safe response when a model draft violates live DB contracts."""
@ -1083,6 +1185,32 @@ def compile_operational_response(contracts: list[dict[str, Any]]) -> str | None:
"sender handle."
)
challenge = by_id.get("claim_evidence_challenge")
if challenge:
claim = challenge.get("selected_claim") or {}
evidence = challenge.get("selected_evidence") or {}
claim_id = str(claim.get("id") or "")
claim_text = str(claim.get("text") or "")
source_id = str(evidence.get("source_id") or "")
excerpt = str(evidence.get("excerpt") or "")
if challenge.get("binding_status") != "bound" or not all((claim_id, claim_text, source_id, excerpt)):
return (
"Live database readback returned no claim/source pair with a non-empty evidence excerpt, so an exact "
"claim-versus-evidence challenge cannot be completed from this result. No identifiers or support were "
"inferred, and no database write was made. Next proof-changing follow-up: broaden the read-only query "
"until one canonical claim and linked source excerpt are returned, then rerun the challenge."
)
return (
f"Claim ID: {claim_id}\n"
f'Exact claim body: "{claim_text}"\n\n'
f"Source ID: {source_id}\n"
f'Exact evidence excerpt: "{excerpt}"\n\n'
"Unsupported leap: this evidence excerpt does not establish every broader causal, trend, or generality "
"assertion in the claim body.\n\n"
"Narrower revision: the linked source establishes only the excerpted observation; any broader inference "
"remains unproven by this evidence row alone. No database write was made."
)
mixed = by_id.get("mixed_packet_composition")
if mixed:
mapping = mixed.get("map") or {}
@ -3134,6 +3262,7 @@ def _bundle_once(args: argparse.Namespace, query: str, claim_limit: int, context
claim_ids = [claim["id"] for claim in claims]
evidence = load_evidence(args, claim_ids, 4)
edges = load_edges(args, claim_ids, 6)
bind_claim_evidence_challenge(contracts, claims, evidence)
return {
"query": query,
"terms": query_terms(query),

48
hermes-agent/leoclean-plugins/vps/leo-db-context/__init__.py
View file

@ -43,6 +43,7 @@ COMPILED_CONTRACT_IDS = frozenset(
"forecast_resolution_schema",
"telegram_delivery_proof",
"claim_supersession_schema",
"claim_evidence_challenge",
}
)
DECISION_MATRIX_TABLES = tuple(
@ -638,6 +639,42 @@ def _source_intake_issues(response: str) -> list[str]:
return sorted(set(issues))
def _claim_evidence_challenge_issues(response: str, contract: dict[str, Any]) -> list[str]:
"""Require exact retrieved values before accepting a claim challenge."""
claim = contract.get("selected_claim") or {}
evidence = contract.get("selected_evidence") or {}
if contract.get("binding_status") != "bound":
if re.search(
r"no claim/source pair.{0,120}exact claim-versus-evidence challenge cannot be completed",
response,
re.I | re.S,
) and re.search(r"no (?:database )?write was made", response, re.I):
return []
return ["claim_challenge_binding_unavailable"]
normalized_response = " ".join(response.split()).lower()
def contains_exact(value: Any) -> bool:
normalized = " ".join(str(value or "").split()).lower()
return bool(normalized and normalized in normalized_response)
issues: list[str] = []
if not contains_exact(claim.get("id")):
issues.append("missing_challenge_claim_id")
if not contains_exact(claim.get("text")):
issues.append("missing_exact_claim_body")
if not contains_exact(evidence.get("source_id")):
issues.append("missing_challenge_source_id")
if not contains_exact(evidence.get("excerpt")):
issues.append("missing_exact_evidence_excerpt")
if not re.search(r"unsupported leap|does not establish|doesn't establish|not supported by", response, re.I):
issues.append("missing_evidence_bounded_challenge")
if not re.search(r"narrower (?:revision|claim|candidate)|revise(?:d)?\b.{0,40}\bto\b", response, re.I | re.S):
issues.append("missing_narrower_revision")
return issues
def _contract_map(contracts: list[dict[str, Any]]) -> dict[str, dict[str, Any]]:
return {str(contract.get("id") or ""): contract for contract in contracts}
@ -814,7 +851,7 @@ def _schema_contract_issues(response: str, contracts: list[dict[str, Any]]) -> l
if "runtime_persistence" in contract_ids and affirmative(
r"(?:state\.db|session\s+jsonl).{0,100}"
r"(?:in[- ]memory|ephemeral|disappears?|vanishes?|erased|discarded|gone|lost|deleted)"
r"(?:in[- ]memory|ephemeral|disappears?|vanishes?|erased|discarded|gone|lost|deleted|absent|zeroed|empty)"
):
issues.append("runtime_persistence_contradiction")
@ -851,7 +888,8 @@ def _schema_contract_issues(response: str, contracts: list[dict[str, Any]]) -> l
def response_contract_issues(response: str, contracts: list[dict[str, Any]]) -> list[str]:
contract_ids = {str(contract.get("id") or "") for contract in contracts}
by_id = _contract_map(contracts)
contract_ids = set(by_id)
issues: list[str] = []
hard_max = _reply_hard_max(contracts)
if hard_max is not None and _word_count(response) > hard_max:
@ -860,15 +898,17 @@ def response_contract_issues(response: str, contracts: list[dict[str, Any]]) ->
issues.extend(_mixed_packet_issues(response))
elif "source_intake" in contract_ids:
issues.extend(_source_intake_issues(response))
if "claim_evidence_challenge" in contract_ids:
issues.extend(_claim_evidence_challenge_issues(response, by_id["claim_evidence_challenge"]))
issues.extend(_live_readback_issues(response, contracts))
issues.extend(_schema_contract_issues(response, contracts))
return sorted(set(issues))
def _requires_compiled_fallback(issues: list[str]) -> bool:
"""Replace only concrete unsafe contradictions, not harmless omissions."""
"""Use the database-compiled answer whenever a declared contract is incomplete."""
return any(issue != "reply_budget_exceeded" and not issue.startswith("missing_") for issue in issues)
return bool(issues)
def _pre_llm_call(**kwargs: Any) -> dict[str, str]:

38
scripts/working_leo_m3taversal_oos_benchmark.py
View file

@ -335,7 +335,9 @@ CONCEPT_PATTERNS: dict[str, tuple[re.Pattern[str], ...]] = {
r"(?:canonical evidence|canonical link).{0,100}(?:requires|exists only when|is complete only when).{0,100}"
r"(?:public\.sources|source row|claim_evidence)|"
r"(?:not yet canonical|staging-only).{0,300}(?:public\.sources|source row).{0,200}claim_evidence|"
r"(?:public\.sources|source row).{0,120}(?:absent|missing|requires?).{0,200}claim_evidence",
r"(?:public\.sources|source row).{0,120}(?:absent|missing|requires?).{0,200}claim_evidence|"
r"public\.sources.{0,180}(?:no (?:matching )?row|none|absent|zero).{0,220}"
r"claim_evidence.{0,120}(?:no (?:matching )?(?:row|link)|none|absent|zero)",
re.I | re.S,
),
),
@ -435,21 +437,32 @@ CONCEPT_PATTERNS: dict[str, tuple[re.Pattern[str], ...]] = {
),
"agent_specific_positions": (
re.compile(r"public\.beliefs", re.I),
re.compile(r"agent_id|agent attribution|attributed to (?:that|each|an?) agent", re.I),
re.compile(
r"agent_id|agent attribution|attributed to (?:that|each|an?) agent|"
r"one (?:public\.beliefs |beliefs )?row per agent(?: position)?|"
r"each agent(?:'s)? (?:belief|position|stance)",
re.I,
),
re.compile(r"belief|position|stance|confidence", re.I),
re.compile(r"no.{0,40}claim(?:-ID|_id).{0,30}(?:foreign key|link)|schema gap", re.I | re.S),
),
"forecast_history": (
re.compile(r"original (?:probability|confidence)|original.{0,120}(?:probability|confidence)|60%|history", re.I),
re.compile(
r"original (?:probability|confidence)|original.{0,120}(?:probability|confidence)|"
r"60%|0[.]60|history",
re.I,
),
re.compile(
r"preserve|retain|do not overwrite|don'?t overwrite|keep.{0,60}unmodified|"
r"leave.{0,60}(?:untouched|unmodified)|original.{0,60}survives",
r"leave.{0,60}(?:untouched|unmodified)|original.{0,80}(?:survives|unchanged|unmodified)|"
r"historical record",
re.I | re.S,
),
re.compile(
r"ambiguous|missing.{0,30}criteria|no.{0,30}criteria|"
r"lack(?:ed|s|ing).{0,30}(?:success|resolution) criteria|"
r"criteria.{0,40}(?:never existed|were never defined|did not exist)|without.{0,30}criteria",
r"criteria.{0,40}(?:never existed|were never defined|did not exist)|"
r"without.{0,60}(?:criteria|resolution rule)",
re.I | re.S,
),
),
@ -580,7 +593,9 @@ CONCEPT_PATTERNS: dict[str, tuple[re.Pattern[str], ...]] = {
),
re.compile(
r"\b(?:narrower|only|limited to|at most|supports? that|evidence shows?|bounded|unestablished|"
r"undemonstrated|unproven|not established|not demonstrated|remains unknown|open constraint)\b",
r"undemonstrated|unproven|not established|not demonstrated|does not establish|do not establish|"
r"doesn't establish|don't establish|remains unknown|open constraint|design aspiration|"
r"not a runtime property)\b",
re.I,
),
),
@ -689,8 +704,10 @@ RESTART_ERASES_ALL_RE = re.compile(
r"restart.{0,100}(?:erases?|forgets?|loses?).{0,40}(?:all|every|prior[- ]session)", re.I | re.S
)
DURABLE_SESSION_EPHEMERAL_RE = re.compile(
r"session JSONL.{0,100}(?:is|are|gets?|becomes?).{0,30}(?:gone|lost|deleted|discarded)|"
r"session JSONL.{0,100}(?:gone|lost|deleted|discarded).{0,50}(?:session closes?|restart)",
r"session JSONL.{0,100}(?:is|are|gets?|becomes?|confirm).{0,30}"
r"(?:gone|lost|deleted|discarded|absent|zeroed|empty)|"
r"session JSONL.{0,100}(?:gone|lost|deleted|discarded|absent|zeroed|empty).{0,50}"
r"(?:session closes?|restart|process launch)",
re.I | re.S,
)
REASONING_TOOL_CLAIM_EDGE_RE = re.compile(
@ -741,7 +758,7 @@ CANONICAL_SOURCE_BEFORE_REVIEW_RE = re.compile(
)
APPLYABILITY_GAP_RE = re.compile(
r"approved_needs_apply_payload|worker_(?:contract_)?applyable(?:_count)?\s*[:=]?\s*(?:false|0)|"
r"(?:no|missing|without).{0,50}(?:strict )?apply_payload|"
r"(?:no|missing|without).{0,50}(?:strict )?apply[_ ]payloads?|"
r"normaliz(?:e|ation).{0,100}(?:before|then).{0,60}(?:review|apply)|"
r"not (?:directly )?applyable|(?:after|once|must|requires?).{0,40}strict apply payload|"
r"strict apply payload.{0,50}(?:built|build|reviewed|review)",
@ -755,7 +772,8 @@ CLAIM_EVIDENCE_CONFLATION_RE = re.compile(
REVISION_LABEL_RE = re.compile(r"\brevision\s*:\s*(?P<value>[^\n]+)", re.I)
REVISION_BOUNDARY_RE = re.compile(
r"\b(?:narrower|only|limited to|at most|supports? that|evidence shows?|bounded|unestablished|unproven|"
r"undemonstrated|not established|not demonstrated|remains unknown|open constraint)\b",
r"undemonstrated|not established|not demonstrated|does not establish|do not establish|doesn't establish|"
r"don't establish|remains unknown|open constraint|design aspiration|not a runtime property)\b",
re.I,
)
REVISION_REPETITION_RE = re.compile(

59
tests/test_hermes_leoclean_db_context_plugin.py
View file

@ -48,12 +48,17 @@ def test_runtime_persistence_contract_rejects_lost_session_jsonl() -> None:
"state.db persists, but the session JSONL is lost when the session closes.",
contracts,
)
absent = module.response_contract_issues(
"Compare the row hashes and confirm session JSONL is absent or zeroed after the process launch.",
contracts,
)
good = module.response_contract_issues(
"state.db and the session JSONL persist on disk and survive a service restart.",
contracts,
)
assert "runtime_persistence_contradiction" in bad
assert "runtime_persistence_contradiction" in absent
assert "runtime_persistence_contradiction" not in good
@ -75,6 +80,49 @@ def test_forecast_contract_requires_reviewed_apply_boundary() -> None:
assert "forecast_review_apply_incomplete" not in complete
def test_claim_challenge_contract_requires_exact_bound_claim_and_evidence() -> None:
module = load_plugin()
contract = {
"id": "claim_evidence_challenge",
"binding_status": "bound",
"selected_claim": {
"id": "11111111-1111-4111-8111-111111111111",
"text": "The association proves a universal mechanism.",
},
"selected_evidence": {
"source_id": "22222222-2222-4222-8222-222222222222",
"excerpt": "The study observed one bounded association.",
},
}
incomplete = (
"Claim 11111111-1111-4111-8111-111111111111 is too broad. Source "
"22222222-2222-4222-8222-222222222222 does not establish causality. Narrower revision: association only."
)
complete = (
"Claim ID: 11111111-1111-4111-8111-111111111111\n"
'Exact claim body: "The association proves a universal mechanism."\n'
"Source ID: 22222222-2222-4222-8222-222222222222\n"
'Exact evidence excerpt: "The study observed one bounded association."\n'
"Unsupported leap: the evidence does not establish a universal mechanism.\n"
"Narrower revision: the evidence establishes only one bounded association."
)
issues = module.response_contract_issues(incomplete, [contract])
assert "missing_exact_claim_body" in issues
assert "missing_exact_evidence_excerpt" in issues
assert module.response_contract_issues(complete, [contract]) == []
unavailable = {
"id": "claim_evidence_challenge",
"binding_status": "unavailable",
}
unavailable_response = (
"Live database readback returned no claim/source pair with a non-empty evidence excerpt, so an exact "
"claim-versus-evidence challenge cannot be completed. No database write was made."
)
assert module.response_contract_issues(unavailable_response, [unavailable]) == []
def test_plugin_injects_bounded_retrieval_rows_and_writes_body_redacted_trace(tmp_path: Path, monkeypatch) -> None:
module = load_plugin()
home = tmp_path / "profile"
@ -420,7 +468,7 @@ def test_plugin_preserves_same_session_chat_only_memory_set_and_recall_responses
assert "OXBOW" not in trace.read_text(encoding="utf-8")
def test_plugin_preserves_noncontradictory_status_draft_but_replaces_contradiction(tmp_path: Path, monkeypatch) -> None:
def test_plugin_replaces_incomplete_status_draft_and_contradiction(tmp_path: Path, monkeypatch) -> None:
module = load_plugin()
home = tmp_path / "profile"
kb_tool = home / "bin" / "kb_tool.py"
@ -465,7 +513,9 @@ def test_plugin_preserves_noncontradictory_status_draft_but_replaces_contradicti
query = "What changed in the KB rather than staying proposed?"
snapshot = module._load_database_snapshot(query, hermes_home=home, runner=fake_runner)
module._store_snapshot("session-direct", snapshot)
assert module._post_llm_call(session_id="session-direct", user_message=query, assistant_response=invalid) is None
assert module._post_llm_call(session_id="session-direct", user_message=query, assistant_response=invalid) == {
"assistant_response": compiled
}
alternative_valid = compiled.replace("Partly.", "Current state.")
assert module.response_contract_issues(alternative_valid, contracts) == []
@ -490,8 +540,9 @@ def test_plugin_preserves_noncontradictory_status_draft_but_replaces_contradicti
if json.loads(line).get("event") == "post_llm_call"
]
assert [record["status"] for record in post_records] == ["ok", "ok", "ok"]
assert [record["contract_satisfied"] for record in post_records] == [False, True, True]
assert post_records[0]["delivered_response_sha256"] == module.hashlib.sha256(invalid.encode()).hexdigest()
assert [record["contract_satisfied"] for record in post_records] == [True, True, True]
assert post_records[0]["delivered_response_sha256"] == module.hashlib.sha256(compiled.encode()).hexdigest()
assert post_records[0]["compiled_response_enforced"] is True
def test_plugin_requires_named_proposal_receipt_when_live_match_exists() -> None:

56
tests/test_hermes_leoclean_kb_bridge_source.py
View file

@ -656,6 +656,9 @@ def test_vps_bridge_operational_contracts_are_query_specific_and_schema_exact()
assert reply_budget["target_words"] == 170
assert reply_budget["hard_max_words"] == 220
assert "omission is better" in reply_budget["shape"]
constrained_budget = module.operational_contracts("Audit the restart claim under 180 words.")[0]
assert constrained_budget["target_words"] == 170
assert constrained_budget["hard_max_words"] == 180
source_contracts = {
item["id"]: item
@ -711,6 +714,43 @@ def test_vps_bridge_operational_contracts_are_query_specific_and_schema_exact()
assert {"reply_budget", "runtime_persistence"} <= runtime_ids
def test_vps_bridge_binds_claim_challenge_to_exact_retrieved_rows() -> None:
module = _load_module(BRIDGE_DIR / "kb_tool.py")
contracts = module.operational_contracts(
"Inspect one exact claim body and its evidence, challenge an unsupported assumption, and give a narrower "
"replacement."
)
claim_id = "11111111-1111-4111-8111-111111111111"
source_id = "22222222-2222-4222-8222-222222222222"
claim_text = "The observed association proves a universal causal mechanism."
excerpt = "The retained study observed an association in one bounded cohort."
module.bind_claim_evidence_challenge(
contracts,
[{"id": claim_id, "text": claim_text}],
{claim_id: [{"source_id": source_id, "excerpt": excerpt, "role": "grounds"}]},
)
response = module.compile_operational_response(contracts)
assert response is not None
assert claim_id in response
assert source_id in response
assert f'Exact claim body: "{claim_text}"' in response
assert f'Exact evidence excerpt: "{excerpt}"' in response
assert "Unsupported leap:" in response
assert "Narrower revision:" in response
unbound_contracts = module.operational_contracts(
"Inspect one exact claim body and its evidence, challenge an unsupported assumption, and give a narrower "
"replacement."
)
module.bind_claim_evidence_challenge(unbound_contracts, [], {})
unbound_response = module.compile_operational_response(unbound_contracts)
assert unbound_response is not None
assert "no claim/source pair" in unbound_response
assert "no database write was made" in unbound_response
def test_vps_prepare_source_retrieves_canonical_candidates_before_model_extraction(tmp_path: Path, monkeypatch) -> None:
module = _load_module(BRIDGE_DIR / "kb_tool.py")
preparer = tmp_path / "prepare_kb_source_manifest.py"
@ -804,6 +844,14 @@ def test_vps_bridge_broad_db_questions_select_query_specific_live_contracts() ->
"Has Helmer's Seven Powers framework landed as live knowledge?": "named_packet_readback",
"Was that accepted through the decision-matrix?": "decision_matrix_readback",
"Is the proposal backlog stuck on missing document-to-source links?": "source_link_audit",
(
"Reviewers signed off on three database proposals. Challenge that conclusion from current state "
"semantics and name the closing receipt."
): "proposal_state_readback",
(
"A source packet is attached to a pending proposal, so provenance is supposedly finished. Audit the "
"canonical evidence link and distinguish it from a weak locator."
): "source_link_audit",
(
"Someone says the V3 architecture document is already part of Leo's live knowledge because a proposal "
"exists. What is actually true right now, what would you inspect, and what must happen before those "
@ -821,6 +869,14 @@ def test_vps_bridge_broad_db_questions_select_query_specific_live_contracts() ->
contract_ids = {item["id"] for item in module.operational_contracts(query)}
assert expected_id in contract_ids, query
runtime_ids = {
item["id"]
for item in module.operational_contracts(
"After a fresh process launch, do unchanged totals prove behavioral parity and a blank session?"
)
}
assert "runtime_persistence" in runtime_ids
def test_vps_bridge_direct_intents_reject_unrelated_and_resolve_overlaps() -> None:
module = _load_module(BRIDGE_DIR / "kb_tool.py")

53
tests/test_working_leo_m3taversal_oos_benchmark.py
View file

@ -741,6 +741,33 @@ def test_db_compiled_responses_pass_all_database_backed_oos_cases() -> None:
assert score["pass"] is True, score
def test_db_compiled_claim_challenge_passes_autonomous_reasoning_benchmark() -> None:
kb_tool = load_kb_tool()
token = "demo-ledger-deadbeef"
prompt = next(item for item in benchmark.prompt_catalog(token) if item["id"] == "OOS-16")
contracts = kb_tool.operational_contracts(prompt["message"])
claim_id = "11111111-1111-4111-8111-111111111111"
source_id = "22222222-2222-4222-8222-222222222222"
kb_tool.bind_claim_evidence_challenge(
contracts,
[{"id": claim_id, "text": "A bounded association proves a universal causal mechanism."}],
{
claim_id: [
{
"source_id": source_id,
"excerpt": "The study observed an association in one bounded cohort.",
"role": "grounds",
}
]
},
)
response = kb_tool.compile_operational_response(contracts)
assert response is not None
assert benchmark.score_reply(prompt, response, memory_token=token)["pass"] is True
def test_oos_state_boundary_accepts_approved_is_not_applied() -> None:
assert benchmark.matched_concept("Approved is not applied.", "state_boundary") is True
@ -782,6 +809,9 @@ def test_oos_runtime_reasoning_rejects_durable_session_as_ephemeral() -> None:
reply = "state.db survives, but the session JSONL is lost when the session closes."
assert benchmark.broad_semantic_issues(reply) == ["durable_session_called_ephemeral"]
assert benchmark.broad_semantic_issues(
"Compare row hashes, then confirm session JSONL is absent or zeroed after process launch."
) == ["durable_session_called_ephemeral"]
def test_oos_agent_positions_accept_attribution_without_literal_column_name() -> None:
@ -845,6 +875,29 @@ def test_oos_semantic_scorer_accepts_natural_receipt_and_policy_paraphrases() ->
"Revision: adoption is documented, while lock-in remains undemonstrated.",
"narrower_claim_revision",
)
assert benchmark.matched_concept(
"public.sources contains no matching row and claim_evidence contains no link, so canonical evidence is absent.",
"canonical_evidence_boundary",
)
assert benchmark.matched_concept(
"Agent divergence lives in public.beliefs, one row per agent position; no claim-ID foreign key exists.",
"agent_specific_positions",
)
assert benchmark.matched_concept(
"The original forecast remains a historical record at confidence 0.60; without a resolution rule the "
"outcome is ambiguous.",
"forecast_history",
)
assert benchmark.matched_concept(
"Narrower revision: the snapshots document recurrence but do not establish an increasing rate.",
"narrower_claim_revision",
)
def test_oos_applyability_gap_accepts_human_readable_payload_wording() -> None:
reply = "The approved proposal has no apply payload, so normalization and renewed review must precede apply."
assert benchmark.proposal_readiness_issues("OOS-07", reply) == []
def test_oos_schema_gap_accepts_explicitly_absent_edge_type() -> None: