Add KB proposal apply preview

diagnostics/kb_proposal_routes.py

@@ -89,6 +89,7 @@ def _load_packets(request: web.Request, filters: dict[str, Any]) -> list[dict[st
     for proposal in proposals:
         packet = proposal_review.classify_proposal(proposal)
         packet["normalization_preview"] = proposal_normalize.normalize_proposal(proposal)
+        packet["apply_preview"] = build_apply_preview(proposal, packet)
         packets.append(packet)
     return packets
 
@@ -135,11 +136,243 @@ def _code(value: Any) -> str:
     return f"<code>{escape(str(value))}</code>"
 
 
+def _payload_dict(value: Any) -> dict[str, Any]:
+    return value if isinstance(value, dict) else {}
+
+
+def _payload_list(value: Any) -> list[Any]:
+    return value if isinstance(value, list) else []
+
+
+def _apply_preview_row(
+    *,
+    action: str,
+    table: str,
+    target: str,
+    claim_ids: list[str] | None = None,
+    details: dict[str, Any] | None = None,
+    source: str = "proposal",
+) -> dict[str, Any]:
+    return {
+        "action": action,
+        "table": table,
+        "target": target,
+        "claim_ids": claim_ids or [],
+        "details": details or {},
+        "source": source,
+    }
+
+
+def _strict_apply_preview_rows(
+    proposal_type: str | None,
+    apply_payload: dict[str, Any],
+    *,
+    source: str = "proposal.apply_payload",
+) -> list[dict[str, Any]]:
+    if proposal_type == "add_edge":
+        from_claim = str(apply_payload.get("from_claim") or "")
+        to_claim = str(apply_payload.get("to_claim") or "")
+        edge_type = str(apply_payload.get("edge_type") or "")
+        return [
+            _apply_preview_row(
+                action="insert_if_missing",
+                table="public.claim_edges",
+                target=f"{from_claim} -> {to_claim} ({edge_type or 'edge_type missing'})",
+                claim_ids=[claim_id for claim_id in [from_claim, to_claim] if claim_id],
+                details={
+                    "from_claim": from_claim,
+                    "to_claim": to_claim,
+                    "edge_type": edge_type,
+                    "weight": apply_payload.get("weight"),
+                },
+                source=source,
+            )
+        ]
+
+    if proposal_type == "attach_evidence":
+        rows = []
+        for evidence in _payload_list(apply_payload.get("evidence")):
+            ev = _payload_dict(evidence)
+            claim_id = str(ev.get("claim_id") or "")
+            source_id = str(ev.get("source_id") or "")
+            role = str(ev.get("role") or "grounds")
+            rows.append(
+                _apply_preview_row(
+                    action="insert_if_missing",
+                    table="public.claim_evidence",
+                    target=f"{claim_id} <= {source_id} ({role})",
+                    claim_ids=[claim_id] if claim_id else [],
+                    details={
+                        "claim_id": claim_id,
+                        "source_id": source_id,
+                        "role": role,
+                        "weight": ev.get("weight"),
+                    },
+                    source=source,
+                )
+            )
+        return rows
+
+    if proposal_type == "revise_strategy":
+        agent_id = str(apply_payload.get("agent_id") or "")
+        strategy_nodes = _payload_list(apply_payload.get("strategy_nodes"))
+        return [
+            _apply_preview_row(
+                action="update_then_insert",
+                table="public.strategies",
+                target=f"new active strategy for agent {agent_id or 'unknown'}",
+                details={
+                    "agent_id": agent_id,
+                    "strategy_keys": sorted(_payload_dict(apply_payload.get("strategy")).keys()),
+                },
+                source=source,
+            ),
+            _apply_preview_row(
+                action="retire_then_insert",
+                table="public.strategy_nodes",
+                target=f"{len(strategy_nodes)} replacement strategy node(s)",
+                details={"agent_id": agent_id, "strategy_node_count": len(strategy_nodes)},
+                source=source,
+            ),
+        ]
+
+    return [
+        _apply_preview_row(
+            action="blocked",
+            table="none",
+            target=f"unsupported proposal_type {proposal_type or 'unknown'}",
+            details={"proposal_type": proposal_type},
+            source=source,
+        )
+    ]
+
+
+def build_apply_preview(proposal: dict[str, Any], packet: dict[str, Any]) -> dict[str, Any]:
+    """Build a read-only, row-level preview of what an apply step would touch."""
+    payload = _payload_dict(proposal.get("payload"))
+    apply_payload = _payload_dict(payload.get("apply_payload"))
+    proposal_type = proposal.get("proposal_type")
+    normalization = _payload_dict(packet.get("normalization_preview"))
+
+    if proposal.get("status") == "applied":
+        return {
+            "state": "applied",
+            "rows": [],
+            "blocked_fragments": [],
+            "note": "Proposal is already applied; use the canonical claim/edge/evidence readback.",
+        }
+
+    if apply_payload and proposal_type in proposal_review.ap.APPLYABLE_TYPES:
+        return {
+            "state": "strict_apply_payload_ready",
+            "rows": _strict_apply_preview_rows(proposal_type, apply_payload),
+            "blocked_fragments": [],
+            "note": "Preview only; the page does not execute canonical writes.",
+        }
+
+    child_rows = []
+    for index, child in enumerate(_payload_list(normalization.get("strict_child_proposals"))):
+        child_payload = _payload_dict(_payload_dict(child).get("payload"))
+        child_apply_payload = _payload_dict(child_payload.get("apply_payload"))
+        child_type = str(_payload_dict(child).get("proposal_type") or "")
+        child_rows.extend(
+            _strict_apply_preview_rows(
+                child_type,
+                child_apply_payload,
+                source=f"normalization.strict_child_proposals[{index}]",
+            )
+        )
+
+    blocked = _payload_list(normalization.get("blocked_fragments"))
+    if child_rows and not blocked:
+        state = "strict_child_proposals_ready"
+        note = "Preview of strict child proposals; reviewer still needs to stage those children before apply."
+    elif child_rows:
+        state = "partial_preview_blocked_fragments"
+        note = "Some strict child rows can be previewed; blocked fragments still need canonical IDs or schema decisions."
+    else:
+        state = "not_applyable_yet"
+        note = "No canonical write preview is safe until the proposal has a strict apply_payload or strict child proposals."
+
+    return {
+        "state": state,
+        "rows": child_rows,
+        "blocked_fragments": blocked,
+        "note": note,
+    }
+
+
 def _status_link(label: str, status: str, filters: dict[str, Any]) -> str:
     query = urlencode({"status": status, "limit": filters["limit"]})
     return f'<a class="filter-link" href="/kb-proposals?{query}">{escape(label)}</a>'
 
 
+def _apply_preview_html(preview: dict[str, Any]) -> str:
+    state = _badge(preview.get("state"))
+    rows = _payload_list(preview.get("rows"))
+    blocked = _payload_list(preview.get("blocked_fragments"))
+    note = escape(str(preview.get("note") or ""))
+    row_html = ""
+    if rows:
+        rendered_rows = []
+        for row in rows:
+            details = _payload_dict(_payload_dict(row).get("details"))
+            detail_text = ", ".join(
+                f"{escape(str(key))}={escape(str(value))}"
+                for key, value in details.items()
+                if value not in (None, "")
+            )
+            claim_links = " ".join(
+                claim_link_html(claim_id, label=str(claim_id)[:8])
+                for claim_id in _payload_list(_payload_dict(row).get("claim_ids"))
+            ) or '<span class="muted">none</span>'
+            detail_html = escape(detail_text) if detail_text else '<span class="muted">none</span>'
+            rendered_rows.append(
+                "<tr>"
+                f"<td>{_code(_payload_dict(row).get('action') or 'unknown')}</td>"
+                f"<td>{_code(_payload_dict(row).get('table') or 'unknown')}</td>"
+                f"<td>{escape(str(_payload_dict(row).get('target') or ''))}</td>"
+                f"<td>{claim_links}</td>"
+                f"<td>{detail_html}</td>"
+                "</tr>"
+            )
+        row_html = f"""
+    <table class="proposal-detail apply-preview-table">
+      <thead><tr><th>Action</th><th>Table</th><th>Target</th><th>Claims</th><th>Details</th></tr></thead>
+      <tbody>{''.join(rendered_rows)}</tbody>
+    </table>"""
+    else:
+        row_html = '<p class="muted">No canonical row preview is executable yet.</p>'
+
+    blocked_html = ""
+    if blocked:
+        blocked_rows = []
+        for fragment in blocked:
+            frag = _payload_dict(fragment)
+            missing = ", ".join(escape(str(value)) for value in _payload_list(frag.get("missing"))) or "unknown"
+            blocked_rows.append(
+                "<tr>"
+                f"<td>{_code(frag.get('kind') or 'fragment')}</td>"
+                f"<td>{escape(str(frag.get('reason') or 'blocked'))}</td>"
+                f"<td>{missing}</td>"
+                "</tr>"
+            )
+        blocked_html = f"""
+    <div class="label preview-label">Blocked fragments</div>
+    <table class="proposal-detail apply-preview-table">
+      <thead><tr><th>Kind</th><th>Reason</th><th>Missing</th></tr></thead>
+      <tbody>{''.join(blocked_rows)}</tbody>
+    </table>"""
+
+    return f"""
+  <div class="next-action apply-preview">
+    <div class="label">Apply preview</div>
+    <p>{state} <span class="muted">{note}</span></p>
+    {row_html}
+    {blocked_html}
+  </div>"""
+
+
 def _packet_card(packet: dict[str, Any]) -> str:
     auth = packet.get("identity_and_authorization") or {}
     summary = packet.get("payload_summary") or {}
@@ -186,6 +419,7 @@ def _packet_card(packet: dict[str, Any]) -> str:
     <div class="label">Normalization action</div>
     <p>{escape(str(normalization.get("next_normalization_action") or "No normalization action recorded."))}</p>
   </div>"""
+    apply_preview_html = _apply_preview_html(packet.get("apply_preview") or {})
     return f"""
 <article class="proposal-card">
   <div class="proposal-card-header">
@@ -199,6 +433,7 @@ def _packet_card(packet: dict[str, Any]) -> str:
     <p>{next_action}</p>
   </div>
   {normalization_html}
+  {apply_preview_html}
 </article>"""
 
 

tests/test_kb_proposal_routes.py

@@ -84,11 +84,37 @@ def test_api_kb_proposals_returns_review_packets_and_counts():
     assert data["packets"][0]["identity_and_authorization"]["source_ref"] == "telegram:leo:m3taversal:claim-review"
     assert data["packets"][0]["normalization_preview"]["normalization_state"] == "blocked_missing_canonical_ids"
     assert data["packets"][1]["normalization_preview"]["normalization_state"] == "already_strict_apply_payload"
+    assert data["packets"][0]["apply_preview"]["state"] == "not_applyable_yet"
+    assert data["packets"][0]["apply_preview"]["blocked_fragments"][0]["missing"] == [
+        "canonical from_claim",
+        "canonical to_claim",
+    ]
+    assert data["packets"][1]["apply_preview"]["state"] == "strict_apply_payload_ready"
+    assert data["packets"][1]["apply_preview"]["rows"] == [
+        {
+            "action": "insert_if_missing",
+            "table": "public.claim_edges",
+            "target": "bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb -> cccccccc-cccc-cccc-cccc-cccccccccccc (supports)",
+            "claim_ids": [
+                "bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb",
+                "cccccccc-cccc-cccc-cccc-cccccccccccc",
+            ],
+            "details": {
+                "from_claim": "bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb",
+                "to_claim": "cccccccc-cccc-cccc-cccc-cccccccccccc",
+                "edge_type": "supports",
+                "weight": None,
+            },
+            "source": "proposal.apply_payload",
+        }
+    ]
 
 
 def test_page_renders_applyability_and_next_admin_action():
+    proposal = _approved_freeform_proposal()
     packet = routes.proposal_review.classify_proposal(_approved_freeform_proposal())
-    packet["normalization_preview"] = routes.proposal_normalize.normalize_proposal(_approved_freeform_proposal())
+    packet["normalization_preview"] = routes.proposal_normalize.normalize_proposal(proposal)
+    packet["apply_preview"] = routes.build_apply_preview(proposal, packet)
     data = routes.build_kb_proposal_response(
         [packet],
         {"status": "approved", "proposal_id": "", "limit": 20},
@@ -105,6 +131,29 @@ def test_page_renders_applyability_and_next_admin_action():
     assert "blocked_missing_canonical_ids" in html
     assert 'href="/kb/claims/d0000000-0000-0000-0000-000000000005"' in html
     assert "<code>d0000000-0000-0000-0000-000000000005</code>" in html
+    assert "Apply preview" in html
+    assert "not_applyable_yet" in html
+    assert "No canonical row preview is executable yet." in html
+    assert "canonical from_claim" in html
+
+
+def test_page_renders_strict_apply_payload_claim_links():
+    proposal = _approved_applyable_proposal()
+    packet = routes.proposal_review.classify_proposal(proposal)
+    packet["normalization_preview"] = routes.proposal_normalize.normalize_proposal(proposal)
+    packet["apply_preview"] = routes.build_apply_preview(proposal, packet)
+    data = routes.build_kb_proposal_response(
+        [packet],
+        {"status": "approved", "proposal_id": "", "limit": 20},
+    )
+
+    html = routes.render_kb_proposals_page(data)
+
+    assert "strict_apply_payload_ready" in html
+    assert "public.claim_edges" in html
+    assert "insert_if_missing" in html
+    assert 'href="/kb/claims/bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb"' in html
+    assert 'href="/kb/claims/cccccccc-cccc-cccc-cccc-cccccccccccc"' in html
 
 
 def test_api_kb_proposals_supports_proposal_id_filter():