@@ -199,6 +433,7 @@ def _packet_card(packet: dict[str, Any]) -> str:
{normalization_html}
+ {apply_preview_html}
{next_action}
{escape(str(value))}"
+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'{escape(label)}'
+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 'none'
+ detail_html = escape(detail_text) if detail_text else 'none'
+ rendered_rows.append(
+ "| Action | Table | Target | Claims | Details |
|---|
No canonical row preview is executable yet.
' + + 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( + "| Kind | Reason | Missing |
|---|
{escape(str(normalization.get("next_normalization_action") or "No normalization action recorded."))}
""" + apply_preview_html = _apply_preview_html(packet.get("apply_preview") or {}) return f"""{next_action}
d0000000-0000-0000-0000-000000000005" 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():