diff --git a/hermes-agent/leoclean-bin/cloudsql_memory_tool.py b/hermes-agent/leoclean-bin/cloudsql_memory_tool.py index c49017a..3b20567 100755 --- a/hermes-agent/leoclean-bin/cloudsql_memory_tool.py +++ b/hermes-agent/leoclean-bin/cloudsql_memory_tool.py @@ -109,6 +109,16 @@ def parse_args() -> argparse.Namespace: propose.add_argument("--rationale", required=True) propose.add_argument("--format", choices=["markdown", "json"], default=argparse.SUPPRESS) + propose_edge = sub.add_parser("propose-edge", help="Stage a strict add_edge proposal for reviewer approval.") + propose_edge.add_argument("from_claim") + propose_edge.add_argument("edge_type") + propose_edge.add_argument("to_claim") + propose_edge.add_argument("--proposed-by", default="leo") + propose_edge.add_argument("--channel", default="telegram") + propose_edge.add_argument("--source-ref", required=True) + propose_edge.add_argument("--rationale", required=True) + propose_edge.add_argument("--format", choices=["markdown", "json"], default=argparse.SUPPRESS) + list_proposals = sub.add_parser("list-proposals", help="List staged KB proposals from Cloud SQL.") list_proposals.add_argument("--status", default="pending_review") list_proposals.add_argument("--limit", type=int, default=10) @@ -145,7 +155,7 @@ def claim_url(claim_id: str | None) -> str | None: def markdown_claim_link(claim_id: str | None, label: str | None = None) -> str: if not claim_id: return "`unknown`" - text = label or claim_id + text = " ".join(str(label or claim_id).replace("`", "'").split()) return f"[`{text}`]({claim_url(claim_id)})" @@ -832,6 +842,99 @@ from inserted; return rows[0] +def propose_edge(args: argparse.Namespace) -> dict[str, Any]: + sql = f""" +with params as ( + select + {sql_literal(args.from_claim)}::uuid as from_claim, + {sql_literal(args.to_claim)}::uuid as to_claim, + {sql_literal(args.edge_type)}::edge_type as edge_type, + nullif({sql_literal(args.proposed_by)}, '') as proposed_by_handle, + nullif({sql_literal(args.channel)}, '') as channel, + nullif({sql_literal(args.source_ref)}, '') as source_ref, + {sql_literal(args.rationale)} as rationale +), from_row as ( + select c.id, c.text + from public.claims c, params p + where c.id = p.from_claim +), to_row as ( + select c.id, c.text + from public.claims c, params p + where c.id = p.to_claim +), proposer as ( + select a.id, a.handle + from public.agents a, params p + where a.handle = p.proposed_by_handle + limit 1 +), existing_edge as ( + select e.id + from public.claim_edges e, params p + where e.from_claim = p.from_claim + and e.to_claim = p.to_claim + and e.edge_type = p.edge_type + limit 1 +), inserted as ( + insert into kb_stage.kb_proposals ( + proposal_type, + status, + proposed_by_handle, + proposed_by_agent_id, + channel, + source_ref, + rationale, + payload + ) + select + 'add_edge', + 'pending_review', + p.proposed_by_handle, + proposer.id, + coalesce(p.channel, 'telegram'), + p.source_ref, + p.rationale, + jsonb_build_object( + 'from_claim', from_row.id::text, + 'from_text', from_row.text, + 'edge_type', p.edge_type::text, + 'to_claim', to_row.id::text, + 'to_text', to_row.text, + 'existing_edge_id', (select id::text from existing_edge), + 'apply_payload', jsonb_build_object( + 'from_claim', from_row.id::text, + 'to_claim', to_row.id::text, + 'edge_type', p.edge_type::text + ) + ) + from params p + join from_row on true + join to_row on true + left join proposer on true + returning * +) +select jsonb_build_object( + 'artifact', 'teleo_cloudsql_kb_edge_proposal', + 'backend', {sql_literal(canonical_backend(args))}, + 'id', id::text, + 'proposal_type', proposal_type, + 'status', status, + 'proposed_by_handle', proposed_by_handle, + 'proposed_by_agent_id', proposed_by_agent_id::text, + 'channel', channel, + 'source_ref', source_ref, + 'rationale', rationale, + 'payload', payload, + 'created_at', created_at::text, + 'canonical_apply_done', false, + 'runtime_memory_write_done', false +)::text +from inserted; +""" + rows = psql_json_lines(args, sql, db=args.canonical_db) + if not rows: + raise SystemExit("No edge proposal inserted. Check that both claim ids exist and edge_type is valid.") + return rows[0] + + def list_proposals(args: argparse.Namespace) -> dict[str, Any]: where = "" if args.status.lower() != "all": @@ -992,9 +1095,10 @@ def emit_markdown(value: dict[str, Any]) -> None: print("No matching canonical claims found.") return for i, claim in enumerate(value["claims"], 1): - print(f"### {i}. {claim.get('text', '[redacted claim text]')}\n") - print(f"- id: `{claim['id']}`") - print(f"- claim page: {claim_url(claim['id'])}") + claim_label = claim.get("text") or "[redacted claim text]" + print(f"### {i}. {markdown_claim_link(claim['id'], claim_label[:140])}\n") + print(f"- claim id: {markdown_claim_link(claim['id'], claim['id'])}") + print(f"- open full claim/body/edges: {markdown_claim_link(claim['id'], 'claim page')}") print(f"- type: `{claim.get('type')}`; confidence: `{claim.get('confidence')}`; score: `{claim.get('score')}`") print(f"- tags: `{', '.join(claim.get('tags') or [])}`") print(f"- evidence rows: `{claim.get('evidence_count', len(claim.get('evidence', [])))}`") @@ -1008,11 +1112,11 @@ def emit_markdown(value: dict[str, Any]) -> None: if claim.get("edges"): print("\nEdges:") for edge in claim["edges"]: - text = f": {edge['connected_text']}" if edge.get("connected_text") else "" connected = edge.get("connected_id") + connected_label = (edge.get("connected_text") or str(connected or ""))[:120] print( f"- `{edge.get('direction')}` `{edge.get('edge_type')}` " - f"({markdown_claim_link(connected, str(connected or '')[:8])}){text}" + f"{markdown_claim_link(connected, connected_label)}" ) print() return @@ -1021,12 +1125,12 @@ def emit_markdown(value: dict[str, Any]) -> None: claim = value["claim"] print("# Teleo KB Claim\n") print(f"- Backend: `{value['backend']}`") - print(f"- id: `{claim['id']}`") - print(f"- claim page: {claim_url(claim['id'])}") + print(f"- claim id: {markdown_claim_link(claim['id'], claim['id'])}") + print(f"- open full claim/body/edges: {markdown_claim_link(claim['id'], 'claim page')}") print(f"- type: `{claim.get('type')}`; status: `{claim.get('status')}`; confidence: `{claim.get('confidence')}`") print(f"- tags: `{', '.join(claim.get('tags') or [])}`") if claim.get("text"): - print(f"\n{claim['text']}\n") + print(f"\n{markdown_claim_link(claim['id'], claim['text'][:180])}\n") print("## Evidence\n") for ev in value.get("evidence", []): source = ev.get("storage_path") or ev.get("url") or "(no source pointer)" @@ -1034,26 +1138,27 @@ def emit_markdown(value: dict[str, Any]) -> None: print(f"- `{ev.get('role')}` / `{ev.get('source_type')}` / `{source}`{excerpt}") print("\n## Edges\n") for edge in value.get("edges", []): - text = f": {edge['connected_text']}" if edge.get("connected_text") else "" connected = edge.get("connected_id") + connected_label = (edge.get("connected_text") or str(connected or ""))[:120] print( f"- `{edge.get('direction')}` `{edge.get('edge_type')}` " - f"({markdown_claim_link(connected, str(connected or '')[:8])}){text}" + f"{markdown_claim_link(connected, connected_label)}" ) return if value["artifact"] in {"teleo_cloudsql_canonical_kb_evidence", "teleo_cloudsql_canonical_kb_edges"}: print(f"# {value['artifact']}\n") print(f"- Backend: `{value['backend']}`") - print(f"- Claim: `{value['claim_id']}`\n") - print(f"- Claim page: {claim_url(value['claim_id'])}\n") + print(f"- Claim: {markdown_claim_link(value['claim_id'], value['claim_id'])}\n") + print(f"- Open full claim/body/edges: {markdown_claim_link(value['claim_id'], 'claim page')}\n") rows = value.get("evidence") or value.get("edges") or [] for row in rows: print(f"- `{json.dumps(row, sort_keys=True)}`") return - if value["artifact"] == "teleo_cloudsql_kb_core_change_proposal": - print("# Staged KB Core-Change Proposal\n") + if value["artifact"] in {"teleo_cloudsql_kb_core_change_proposal", "teleo_cloudsql_kb_edge_proposal"}: + title = "Staged KB Edge Proposal" if value["artifact"] == "teleo_cloudsql_kb_edge_proposal" else "Staged KB Core-Change Proposal" + print(f"# {title}\n") print(f"- Proposal id: `{value['id']}`") print(f"- Type: `{value['proposal_type']}`") print(f"- Status: `{value['status']}`") @@ -1125,6 +1230,8 @@ def main() -> None: result = edges_canonical_claim(args) elif args.command == "propose-core-change": result = propose_core_change(args) + elif args.command == "propose-edge": + result = propose_edge(args) elif args.command == "list-proposals": result = list_proposals(args) elif args.command == "show-proposal": diff --git a/hermes-agent/leoclean-bin/kb_tool.py b/hermes-agent/leoclean-bin/kb_tool.py index 23a9522..0cee32d 100755 --- a/hermes-agent/leoclean-bin/kb_tool.py +++ b/hermes-agent/leoclean-bin/kb_tool.py @@ -233,7 +233,7 @@ def claim_url(claim_id: str | None) -> str | None: def markdown_claim_link(claim_id: str | None, label: str | None = None) -> str: if not claim_id: return "`unknown`" - text = label or claim_id + text = " ".join(str(label or claim_id).replace("`", "'").split()) return f"[`{text}`]({claim_url(claim_id)})" @@ -470,7 +470,12 @@ def propose_edge(args: argparse.Namespace) -> dict[str, Any]: 'edge_type', p.edge_type::text, 'to_claim', to_row.id::text, 'to_text', to_row.text, - 'existing_edge_id', (select id::text from existing_edge) + 'existing_edge_id', (select id::text from existing_edge), + 'apply_payload', jsonb_build_object( + 'from_claim', from_row.id::text, + 'to_claim', to_row.id::text, + 'edge_type', p.edge_type::text + ) ) from params p join from_row on true @@ -804,9 +809,9 @@ def print_claim_bundle(data: dict[str, Any]) -> None: return print("## Claims\n") for idx, claim in enumerate(data["claims"], start=1): - print(f"### {idx}. {claim['text']}\n") - print(f"- id: `{claim['id']}`") - print(f"- claim page: {claim_url(claim['id'])}") + print(f"### {idx}. {markdown_claim_link(claim['id'], truncate(claim.get('text'), 140))}\n") + print(f"- claim id: {markdown_claim_link(claim['id'], claim['id'])}") + print(f"- open full claim/body/edges: {markdown_claim_link(claim['id'], 'claim page')}") print(f"- type: `{claim['type']}`; confidence: `{claim['confidence']}`; score: `{claim.get('score', '-')}`") print(f"- tags: `{', '.join(claim.get('tags') or [])}`") print(f"- evidence rows: `{claim.get('evidence_count', len(claim.get('evidence', [])))}`") @@ -821,10 +826,10 @@ def print_claim_bundle(data: dict[str, Any]) -> None: print("\nEdges:") for edge in claim["edges"]: connected = edge["connected_id"] + connected_label = truncate(edge.get("connected_text") or connected, 120) print( f"- `{edge['direction']}` `{edge['edge_type']}` " - f"({markdown_claim_link(connected, connected[:8])}): " - f"{truncate(edge['connected_text'], 260)}" + f"{markdown_claim_link(connected, connected_label)}" ) print() @@ -843,9 +848,11 @@ def print_proposal(proposal: dict[str, Any]) -> None: payload = proposal.get("payload") or {} if proposal["proposal_type"] == "add_edge": print("\n## Proposed Edge\n") - print(f"- from: `{payload.get('from_claim')}` - {truncate(payload.get('from_text'), 260)}") + print( + f"- from: {markdown_claim_link(payload.get('from_claim'), truncate(payload.get('from_text'), 120))}" + ) print(f"- edge_type: `{payload.get('edge_type')}`") - print(f"- to: `{payload.get('to_claim')}` - {truncate(payload.get('to_text'), 260)}") + print(f"- to: {markdown_claim_link(payload.get('to_claim'), truncate(payload.get('to_text'), 120))}") if payload.get("existing_edge_id"): print(f"- existing_edge_id: `{payload['existing_edge_id']}`") else: @@ -868,8 +875,8 @@ def print_proposal_list(rows: list[dict[str, Any]]) -> None: print(f"- created_at: `{row.get('created_at')}`") if row["proposal_type"] == "add_edge": print( - f"- edge: `{payload.get('from_claim')}` " - f"`{payload.get('edge_type')}` `{payload.get('to_claim')}`" + f"- edge: {markdown_claim_link(payload.get('from_claim'), truncate(payload.get('from_text'), 80))} " + f"`{payload.get('edge_type')}` {markdown_claim_link(payload.get('to_claim'), truncate(payload.get('to_text'), 80))}" ) print(f"- rationale: {truncate(row.get('rationale'), 320)}") print() diff --git a/hermes-agent/leoclean-bin/teleo-kb b/hermes-agent/leoclean-bin/teleo-kb index 1c0b1d1..2548c3a 100755 --- a/hermes-agent/leoclean-bin/teleo-kb +++ b/hermes-agent/leoclean-bin/teleo-kb @@ -10,7 +10,7 @@ CLOUDSQL_STATUS_TIMEOUT=${TELEO_CLOUDSQL_STATUS_TIMEOUT_SECONDS:-30} cloudsql_supported() { case "$COMMAND" in - status|search|context|show|evidence|edges|propose-core-change|list-proposals|show-proposal) return 0 ;; + status|search|context|show|evidence|edges|propose-core-change|propose-edge|list-proposals|show-proposal) return 0 ;; *) return 1 ;; esac } diff --git a/tests/test_apply_worker.py b/tests/test_apply_worker.py index 2c855e0..0ac3bea 100644 --- a/tests/test_apply_worker.py +++ b/tests/test_apply_worker.py @@ -6,8 +6,8 @@ standalone (`python3 tests/test_apply_worker.py`). """ import argparse -from pathlib import Path import sys +from pathlib import Path try: import pytest @@ -33,6 +33,19 @@ sys.path.insert(0, str(REPO_ROOT / "scripts")) import apply_worker as w # noqa: E402 +@pytest.fixture(autouse=True) +def restore_apply_worker_dependencies(): + original_load_password = w.ap.load_password + original_fetch_candidates = w.fetch_candidates + original_apply_one = w.apply_one + original_render_one = w.render_one + yield + w.ap.load_password = original_load_password + w.fetch_candidates = original_fetch_candidates + w.apply_one = original_apply_one + w.render_one = original_render_one + + # --- candidate query ------------------------------------------------------ # def test_candidate_query_filters_on_approved(): sql = w.build_candidate_query() @@ -192,7 +205,7 @@ if __name__ == "__main__": try: fn() print(f"PASS {name}") - except Exception: # noqa: BLE001 + except Exception: failures += 1 print(f"FAIL {name}") traceback.print_exc() diff --git a/tests/test_hermes_leoclean_kb_bridge_source.py b/tests/test_hermes_leoclean_kb_bridge_source.py index 835ab18..7826ae3 100644 --- a/tests/test_hermes_leoclean_kb_bridge_source.py +++ b/tests/test_hermes_leoclean_kb_bridge_source.py @@ -3,7 +3,9 @@ from __future__ import annotations import ast +import contextlib import importlib.util +import io import re import subprocess from pathlib import Path @@ -38,17 +40,29 @@ def test_cloudsql_wrapper_supports_expected_review_gated_commands() -> None: "evidence", "edges", "propose-core-change", + "propose-edge", "list-proposals", "show-proposal", ]: assert command in wrapper_text assert command in cloudsql_text - assert "propose-edge" not in wrapper_text assert "kb_stage.kb_proposals" in cloudsql_text assert "canonical apply" in cloudsql_text.lower() +def test_bridge_edge_proposals_stage_strict_apply_payloads() -> None: + cloudsql_text = (BRIDGE_DIR / "cloudsql_memory_tool.py").read_text() + vps_text = (BRIDGE_DIR / "kb_tool.py").read_text() + + for text in (cloudsql_text, vps_text): + assert "'proposal_type', proposal_type" in text or "select 'add_edge'" in text + assert "'apply_payload', jsonb_build_object(" in text + assert "'from_claim', from_row.id::text" in text + assert "'to_claim', to_row.id::text" in text + assert "'edge_type', p.edge_type::text" in text + + def test_bridge_source_does_not_commit_raw_secret_values() -> None: combined = "\n".join(path.read_text(errors="replace") for path in BRIDGE_DIR.iterdir() if path.is_file()) forbidden_patterns = [ @@ -82,3 +96,55 @@ def test_kb_bridges_emit_public_claim_links_for_telegram_rendering() -> None: assert module.markdown_claim_link(claim_id, "d3fb892b") == ( f"[`d3fb892b`]({expected})" ) + assert module.markdown_claim_link(claim_id, "claim `with ticks`") == ( + f"[`claim 'with ticks'`]({expected})" + ) + + +def test_vps_bridge_markdown_links_claim_text_and_edges() -> None: + module = _load_module(BRIDGE_DIR / "kb_tool.py") + claim_id = "d3fb892b-3c5a-4700-9512-55e5c680eec1" + connected_id = "9d5281fe-7ee4-4fe1-b1bf-c54c4d7fb6a5" + data = { + "query": "strategy kernel", + "context_rows": [], + "claims": [ + { + "id": claim_id, + "text": "Claims should be easy to scan in Telegram.", + "type": "belief", + "confidence": 0.8, + "tags": ["telegram"], + "score": 4, + "evidence_count": 2, + "edge_count": 1, + "evidence": [], + "edges": [ + { + "direction": "outgoing", + "edge_type": "supports", + "connected_id": connected_id, + "connected_text": "Full claim pages expose body, evidence, and graph edges.", + } + ], + } + ], + } + + buffer = io.StringIO() + with contextlib.redirect_stdout(buffer): + module.print_claim_bundle(data) + markdown = buffer.getvalue() + + assert ( + f"[`Claims should be easy to scan in Telegram.`](https://leo.livingip.xyz/kb/claims/{claim_id})" + in markdown + ) + assert ( + f"- open full claim/body/edges: [`claim page`](https://leo.livingip.xyz/kb/claims/{claim_id})" + in markdown + ) + assert ( + f"[`Full claim pages expose body, evidence, and graph edges.`](https://leo.livingip.xyz/kb/claims/{connected_id})" + in markdown + )