Merge pull request #132 from living-ip/codex/leo-v3-proposal-state-routing-20260713

Fix Leo named proposal readback routing

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"(?<![a-z0-9]){re.escape(term)}(?![a-z0-9])", haystack)
+        ]
+        if len(matched_terms) < minimum_term_matches or not version_terms.issubset(matched_terms):
+            continue
+        score = sum(2 if len(term) >= 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))

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:

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"

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")