diff --git a/hermes-agent/leoclean-bin/kb_tool.py b/hermes-agent/leoclean-bin/kb_tool.py index ea9b349..32492aa 100755 --- a/hermes-agent/leoclean-bin/kb_tool.py +++ b/hermes-agent/leoclean-bin/kb_tool.py @@ -52,6 +52,53 @@ DECISION_MATRIX_TABLES = tuple( for schema in ("public", "kb_stage") for table in ("matrix_voters", "proposal_votes", "proposal_decisions") ) +PROPOSAL_QUERY_STOPWORDS = frozenset( + { + "about", + "actually", + "after", + "already", + "anything", + "applied", + "approved", + "before", + "because", + "canonical", + "change", + "changed", + "claim", + "claims", + "database", + "document", + "does", + "exists", + "happen", + "inspect", + "knowledge", + "landed", + "leo", + "live", + "mutate", + "must", + "part", + "pending", + "proposal", + "proposed", + "report", + "review", + "right", + "says", + "someone", + "source", + "status", + "those", + "true", + "update", + "updated", + "what", + "would", + } +) CURRENT_PUBLIC_SCHEMA = { "behavioral_rules": ("id", "agent_id", "category", "rank", "rule", "rationale", "created_at"), "beliefs": ( @@ -395,6 +442,65 @@ def _has_unnegated_action(query: str, actions: tuple[str, ...]) -> bool: return False +def proposal_query_terms(query: str) -> list[str]: + """Return artifact discriminators while dropping lifecycle boilerplate.""" + + terms: list[str] = [] + for term in re.findall(r"[a-z0-9]+", query.lower()): + if term in STOPWORDS or term in PROPOSAL_QUERY_STOPWORDS: + continue + if len(term) < 4 and not re.fullmatch(r"v\d+", term): + continue + if term not in terms: + terms.append(term) + return terms + + +def rank_query_matching_proposals( + query: str, + proposals: list[dict[str, Any]], + *, + limit: int = 3, +) -> tuple[list[dict[str, Any]], list[str]]: + """Rank proposal payload/source matches without treating generic lifecycle words as identity.""" + + terms = proposal_query_terms(query) + if not terms: + return [], [] + + version_terms = {term for term in terms if re.fullmatch(r"v\d+", term)} + minimum_term_matches = min(2, len(terms)) + status_priority = {"applied": 4, "approved": 3, "pending_review": 2, "canceled": 0} + ranked: list[tuple[int, int, str, int, dict[str, Any], list[str]]] = [] + for index, proposal in enumerate(proposals): + haystack = json.dumps(proposal, sort_keys=True, default=str).lower() + matched_terms = [ + term + for term in terms + if re.search(rf"(?= 8 or re.fullmatch(r"v\d+", term) else 1 for term in matched_terms) + if score < 2: + continue + ranked.append( + ( + score, + status_priority.get(str(proposal.get("status") or ""), 1), + str(proposal.get("created_at") or ""), + -index, + proposal, + matched_terms, + ) + ) + + ranked.sort(key=lambda item: item[:4], reverse=True) + selected = ranked[: max(1, limit)] + observed_terms = [term for term in terms if any(term in item[5] for item in selected)] + return [item[4] for item in selected], observed_terms + + def operational_contracts( query: str, *, @@ -437,7 +543,7 @@ def operational_contracts( ) and any( term in lowered - for term in ("because", "wrong", "missing", "stuck", "link", "point", "cause", "canonical evidence", "audit") + for term in ("wrong", "missing", "stuck", "link", "point", "canonical evidence", "audit") ) ) kb_scope = ( @@ -445,6 +551,11 @@ def operational_contracts( or "database" in lowered or bool(re.search(r"\bkb\b", lowered)) or "canonical knowledge" in lowered + or "live knowledge" in lowered + or ( + "canonical" in lowered + and any(term in lowered for term in ("proposal", "proposed", "pending", "approved", "applied")) + ) ) kb_implementation_question = bool( re.search( @@ -870,9 +981,10 @@ def format_database_count_readback(status_data: dict[str, Any] | None) -> str: def format_proposal_readback(proposal: dict[str, Any]) -> str: applied_at = proposal.get("applied_at") or "none" readiness = (proposal.get("readiness") or {}).get("review_state") or "unknown" + source_ref = proposal.get("source_ref") or "none" return ( f"DB readback: proposal: {proposal.get('id')}; status: {proposal.get('status')}; " - f"applied_at: {applied_at}; readiness: {readiness}." + f"applied_at: {applied_at}; source_ref: {source_ref}; readiness: {readiness}." ) @@ -1095,6 +1207,8 @@ def compile_operational_response(contracts: list[dict[str, Any]]) -> str | None: if proposal_state: status_data = proposal_state.get("database_status") or {} states = status_data.get("proposal_status_counts") or {} + matches = list(proposal_state.get("matching_proposals") or []) + primary = matches[0] if matches else None approved_rows = [ item for item in proposal_state.get("applied_and_approved_proposals") or [] @@ -1104,6 +1218,39 @@ def compile_operational_response(contracts: list[dict[str, Any]]) -> str | None: (item.get("readiness") or {}).get("review_state") == "approved_needs_apply_payload" for item in approved_rows ) + if primary: + status = str(primary.get("status") or "unknown") + applied_at = primary.get("applied_at") + if status == "applied" and applied_at: + lead = "Partly: the matching proposal has an apply receipt, but its canonical rows still need postflight proof." + state_sentence = ( + "The proposal ledger marks it applied; that receipt alone does not identify the exact public.* " + "rows written." + ) + elif status == "approved": + lead = "No." + state_sentence = "The matching proposal is reviewer-approved, but applied_at is empty; approved is not applied." + elif status == "pending_review": + lead = "No." + state_sentence = ( + "The matching proposal remains pending_review and applied_at is empty, so it has not reached " + "canonical apply." + ) + else: + lead = "No." + state_sentence = ( + f"The matching proposal is {status} and has no canonical apply receipt; it is not live knowledge." + ) + return ( + f"{lead}\nFresh live readback.\n{format_database_count_readback(status_data)}\n" + f"{format_proposal_readback(primary)}\nProposal states are applied: {states.get('applied', 0)}, " + f"approved: {states.get('approved', 0)}, pending_review: {states.get('pending_review', 0)}, canceled: " + f"{states.get('canceled', 0)}. {state_sentence} A proposal row is not canonical knowledge without " + "the guarded apply receipt and row-level public.* postflight.\n\n" + "Next proof-changing follow-up: inspect the matching proposal payload and exact source binding, " + "complete human review, obtain explicit guarded-apply authorization, then return non-null applied_at " + "plus the created public.* row IDs and hashes." + ) lead = "No." if proposal_state.get("approval_only_claim") else "Partly." return ( f"{lead}\nFresh live readback.\n{format_database_count_readback(status_data)}\nProposal states are applied: " @@ -1460,7 +1607,9 @@ def context_operational_contracts(args: argparse.Namespace, query: str) -> list[ contract_ids = {contract["id"] for contract in contracts} direct_snapshot = load_direct_readback_snapshot(args) if contract_ids & DIRECT_READBACK_CONTRACTS else {} status_data = direct_snapshot.get("database_status") - proposal_rows = [compact_proposal(with_proposal_readiness(item)) for item in direct_snapshot.get("proposals") or []] + raw_proposal_rows = list(direct_snapshot.get("proposals") or []) + proposal_rows = [compact_proposal(with_proposal_readiness(item)) for item in raw_proposal_rows] + matching_proposal_rows, matching_terms = rank_query_matching_proposals(query, raw_proposal_rows) for contract in contracts: contract_id = contract["id"] @@ -1472,6 +1621,10 @@ def context_operational_contracts(args: argparse.Namespace, query: str) -> list[ contract["applied_and_approved_proposals"] = [ item for item in proposal_rows if item.get("status") in {"applied", "approved"} ] + contract["matching_proposals"] = [ + compact_proposal(with_proposal_readiness(item)) for item in matching_proposal_rows + ] + contract["proposal_query_terms"] = matching_terms elif contract_id == "named_packet_readback": contract["matching_proposals"] = [ compact_proposal(with_proposal_readiness(item)) diff --git a/hermes-agent/leoclean-plugins/vps/leo-db-context/__init__.py b/hermes-agent/leoclean-plugins/vps/leo-db-context/__init__.py index 86238e5..a85f55a 100644 --- a/hermes-agent/leoclean-plugins/vps/leo-db-context/__init__.py +++ b/hermes-agent/leoclean-plugins/vps/leo-db-context/__init__.py @@ -462,6 +462,9 @@ def _live_readback_issues(response: str, contracts: list[dict[str, Any]]) -> lis issues.append("missing_proposal_apply_boundary") if re.search(r"all approved (?:rows|proposals).{0,40}(?:already )?(?:canonical|applied|live)", response, re.I): issues.append("contradictory_approved_state_claim") + matching_proposals = list(proposal_state.get("matching_proposals") or []) + if matching_proposals and not _has_proposal_readback(response, matching_proposals[0]): + issues.append("missing_matching_proposal_readback") named_packet = by_id.get("named_packet_readback") if named_packet: diff --git a/tests/test_hermes_leoclean_db_context_plugin.py b/tests/test_hermes_leoclean_db_context_plugin.py index 84b3ea1..cd1ddfa 100644 --- a/tests/test_hermes_leoclean_db_context_plugin.py +++ b/tests/test_hermes_leoclean_db_context_plugin.py @@ -242,6 +242,51 @@ def test_plugin_replaces_memory_only_status_answer_with_exact_live_db_receipt(tm ) == {"assistant_response": compiled} +def test_plugin_requires_named_proposal_receipt_when_live_match_exists() -> None: + module = load_plugin() + database_status = { + "high_signal_rows": { + "claims": 1837, + "sources": 4145, + "claim_edges": 4916, + "claim_evidence": 4670, + "kb_proposals": 29, + }, + "proposal_status_counts": {"applied": 2, "approved": 3, "pending_review": 16, "canceled": 8}, + } + contracts = [ + {"id": "reply_budget", "hard_max_words": 200}, + { + "id": "proposal_state_readback", + "database_status": database_status, + "matching_proposals": [ + { + "id": "10bc0719-1c2b-5f42-a41e-86e6478692cb", + "status": "pending_review", + "applied_at": None, + } + ], + }, + ] + common = ( + "DB readback: claims: 1837; sources: 4145; claim_edges: 4916; claim_evidence: 4670; kb_proposals: 29.\n" + "Proposal states are applied: 2, approved: 3, pending_review: 16, canceled: 8. Approved is not applied; " + "applied_at and row-level postflight are required.\n\n" + "Next proof-changing follow-up: inspect the named proposal, then return its apply receipt and public rows." + ) + valid = ( + "No.\n" + "DB readback: proposal: 10bc0719-1c2b-5f42-a41e-86e6478692cb; status: pending_review; " + "applied_at: none; readiness: needs_human_review.\n" + + common + ) + + missing_receipt_issues = module.response_contract_issues("No.\n" + common, contracts) + + assert "missing_matching_proposal_readback" in missing_receipt_issues + assert module.response_contract_issues(valid, contracts) == [] + + def test_plugin_fails_closed_when_database_contract_lookup_fails(tmp_path: Path) -> None: module = load_plugin() home = tmp_path / "profile" diff --git a/tests/test_hermes_leoclean_kb_bridge_source.py b/tests/test_hermes_leoclean_kb_bridge_source.py index faaec24..84bf1c8 100644 --- a/tests/test_hermes_leoclean_kb_bridge_source.py +++ b/tests/test_hermes_leoclean_kb_bridge_source.py @@ -357,6 +357,11 @@ 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", + ( + "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 " + "claims are canonical? Do not mutate anything." + ): "proposal_state_readback", "Can we demo a real database write from Leo?": "demo_capability_readback", "Does editing the SOUL file alter the source of truth for identity?": "identity_canonicality_readback", ( @@ -404,6 +409,93 @@ def test_vps_bridge_direct_intents_reject_unrelated_and_resolve_overlaps() -> No } assert substring_ids & direct_ids == {"proposal_state_readback"} + v3_ids = { + item["id"] + for item in module.operational_contracts( + "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 " + "claims are canonical? Do not mutate anything." + ) + } + assert v3_ids & direct_ids == {"proposal_state_readback"} + assert "source_link_audit" not in v3_ids + + +def test_vps_bridge_named_document_question_matches_exact_proposal_and_compiles_receipt(monkeypatch) -> None: + module = _load_module(BRIDGE_DIR / "kb_tool.py") + database_status, approved, matrix, source_audit = _direct_contract_fixtures() + database_status = { + **database_status, + "high_signal_rows": {**database_status["high_signal_rows"], "kb_proposals": 29}, + "proposal_status_counts": {"applied": 2, "approved": 3, "pending_review": 16, "canceled": 8}, + } + source_metadata = { + "apply_payload": { + "sources": [ + { + "excerpt": json.dumps( + { + "metadata": { + "title": "LivingIP KB Agent Identity Architecture v3", + "source_key": "livingip_kb_agent_identity_architecture_v3", + } + } + ) + } + ] + } + } + corrected = { + **approved, + "id": "10bc0719-1c2b-5f42-a41e-86e6478692cb", + "status": "pending_review", + "source_ref": "normalized:12e38cb0-b72b-557c-95e5-ffc789a29f62:72895c9b58b79d41", + "payload": source_metadata, + "created_at": "2026-07-13T04:00:00+00:00", + "reviewed_at": None, + "applied_at": None, + } + canceled = { + **corrected, + "id": "a9eb4158-5aa2-5484-b5a7-3d249588b3ec", + "status": "canceled", + "created_at": "2026-07-13T03:00:00+00:00", + } + monkeypatch.setattr( + module, + "load_direct_readback_snapshot", + lambda _args: { + "snapshot_id": "300:300:", + "statement_timestamp": "2026-07-13T05:00:00+00:00", + "database_status": database_status, + "proposals": [approved, canceled, corrected], + "named_packet_proposals": [approved], + "named_packet_canonical_matches": {"reasoning_tools": [], "claims": [], "sources": []}, + "decision_matrix_status": matrix, + "source_link_audit": source_audit, + }, + ) + prompt = ( + "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 claims " + "are canonical? Do not mutate anything." + ) + + contracts = module.context_operational_contracts(SimpleNamespace(), prompt) + proposal_state = next(item for item in contracts if item["id"] == "proposal_state_readback") + response = module.compile_operational_response(contracts) + + assert proposal_state["proposal_query_terms"] == ["v3", "architecture"] + assert [item["id"] for item in proposal_state["matching_proposals"]] == [corrected["id"], canceled["id"]] + assert response is not None + assert response.startswith("No.\nFresh live readback.") + assert f"proposal: {corrected['id']}; status: pending_review; applied_at: none" in response + assert corrected["source_ref"] in response + assert "proposal states are applied: 2, approved: 3, pending_review: 16, canceled: 8" in response.lower() + assert "has not reached canonical apply" in response + assert "telegram_file_refs" not in response + assert len(re.findall(r"\b\w+(?:[-']\w+)*\b", response)) <= 200 + def test_vps_bridge_oos_intents_preserve_specific_contracts_and_negated_actions() -> None: module = _load_module(BRIDGE_DIR / "kb_tool.py")