# Author: Simon-Pierre Boucher — simon-pierre.boucher@uqo.ca
"""Evidence packets: construction from the analytic row, least-privilege role views, user-prompt rendering
and the deterministic evidence-verification tool (arm H, faithfulness audit).
Invariants
----------
* A packet never contains the subject's sale price, street address or any identifier other than ``tx_id``.
* Coordinates are rounded to 3 decimals (~100 m).
* ``missing_fields`` lists the subject attributes that are null so that agents keep them missing.
* The user prompt embeds ``{json}`` (parsed by the mock provider) and,
in deliberation rounds, ``{json}`` with ``committee_median``.
"""
from __future__ import annotations
import hashlib
import json
import math
from datetime import date, timedelta
from typing import Any
from ..schemas import AgentEstimate, Comparable, EvidencePacket, GeoContext, MarketContext, SubjectProperty
from .roles import SECTIONS, Role
FORBIDDEN_KEYS = {"price", "log_price", "street", "address", "civic_number", "civic", "owner", "buyer", "seller", "remarks", "listing_remarks", "parcel_id"}
_SUBJECT_FIELDS = (
"property_class", "subtype", "borough", "floor_area_m2", "land_area_m2", "year_built", "storeys", "n_units",
"roll_land_value", "roll_building_value", "roll_total_value", "roll_reference_date", "prior_sale_price", "prior_sale_date", "new_construction",
)
_MISSING_CHECK = ("floor_area_m2", "land_area_m2", "year_built", "storeys", "n_units", "roll_land_value", "roll_building_value", "roll_total_value", "subtype")
# ---------------------------------------------------------------------------------------------------
# task instructions (stable strings; part of the freeze through the code hash)
# ---------------------------------------------------------------------------------------------------
ESTIMATE_INSTRUCTION = "TASK: Return an AgentEstimate JSON object with your independent estimate of the market value in CAD as of the valuation date, using only the sections visible to you."
CRITIQUE_INSTRUCTION = "TASK: Return a CritiqueSet JSON object. For each committee member in committee_state.estimates (other than yourself), state whether you agree with its estimate, the evidence-based points that support or contradict it (cite ids), and a suggested value only when you can justify it. Do not critique yourself."
REVISION_INSTRUCTION = "TASK: Return a revised AgentEstimate JSON object. Consider the committee_state (estimates of the other members, the committee median, the critiques addressed to you and any verification flags). Change your estimate only for evidential reasons; otherwise keep it and explain why in confidence_rationale."
SKEPTIC_VERDICT_INSTRUCTION = "TASK: You are a blinded reviewer: you see the packet and the committee's final estimates only, not the discussion. Return an AgentEstimate JSON object with your own independent verdict on the market value, and use counterarguments to state the strongest reasons the committee could be wrong."
SELF_CRITIQUE_INSTRUCTION = "TASK: Return a CritiqueSet JSON object with exactly one Critique whose target_role is your own role: review your own previous estimate shown in committee_state for arithmetic errors, unsupported assumptions, missing evidence and over-confidence."
SELF_REFINE_INSTRUCTION = "TASK: Return a refined AgentEstimate JSON object that addresses your own critique shown in committee_state. Keep the estimate if the critique found nothing substantive."
_INSTRUCTIONS = {
"estimate": ESTIMATE_INSTRUCTION,
"critique": CRITIQUE_INSTRUCTION,
"revision": REVISION_INSTRUCTION,
"skeptic_verdict": SKEPTIC_VERDICT_INSTRUCTION,
"self_critique": SELF_CRITIQUE_INSTRUCTION,
"self_refine": SELF_REFINE_INSTRUCTION,
}
def _f(v: Any) -> float | None:
if v is None:
return None
try:
x = float(v)
except (TypeError, ValueError):
return None
return None if math.isnan(x) else x
def _i(v: Any) -> int | None:
x = _f(v)
return None if x is None else int(round(x))
def _d(v: Any) -> date | None:
if v is None:
return None
if isinstance(v, date):
return v
try:
return date.fromisoformat(str(v)[:10])
except ValueError:
return None
def build_packet(analytic_row: dict[str, Any], comps: list[Comparable], market: MarketContext, geo: GeoContext | None, cfg: dict[str, Any] | None = None) -> EvidencePacket:
"""Build the full (unfiltered) evidence packet for one subject. Price and address never enter."""
row = analytic_row
sale_date = _d(row.get("date") or row.get("valuation_date"))
if sale_date is None:
raise ValueError("analytic row needs a 'date' (valuation date)")
cutoff = _d(row.get("information_cutoff")) or (sale_date - timedelta(days=1))
sub = row.get("subtype") or "unknown"
if sub not in ("detached", "semi", "row", "integrated", "unknown"):
sub = "unknown"
fields: dict[str, Any] = {
"tx_id": str(row["tx_id"]),
"valuation_date": sale_date,
"information_cutoff": cutoff,
"property_class": row["property_class"],
"subtype": sub,
"region": str(row.get("region_name") or row.get("region") or "unknown"),
"municipality": str(row.get("mun_name") or row.get("municipality") or "unknown"),
"borough": row.get("borough_name") or row.get("borough"),
"lat": round(float(row["lat"]), 3),
"lng": round(float(row["lng"]), 3),
"floor_area_m2": _f(row.get("floor_area_m2")),
"land_area_m2": _f(row.get("land_area_m2")),
"year_built": _i(row.get("year_built")),
"storeys": _f(row.get("storeys")),
"n_units": _i(row.get("n_units")),
"roll_land_value": _f(row.get("roll_land_value")),
"roll_building_value": _f(row.get("roll_building_value")),
"roll_total_value": _f(row.get("roll_total_value")),
"roll_reference_date": (str(row["roll_reference_date"]) if row.get("roll_reference_date") else None),
"prior_sale_price": _f(row.get("prior_sale_price")),
"prior_sale_date": _d(row.get("prior_sale_date")),
"new_construction": (None if row.get("new_construction") is None else bool(row.get("new_construction"))),
}
if fields["prior_sale_date"] is not None and fields["prior_sale_date"] >= cutoff: # schema: strictly before the cutoff
fields["prior_sale_price"], fields["prior_sale_date"] = None, None # leakage guard
missing = [k for k in _MISSING_CHECK if fields.get(k) in (None, "unknown")]
if fields["roll_reference_date"] is None:
missing.append("roll_reference_date")
fields["missing_fields"] = missing
subject = SubjectProperty(**fields)
max_comps = int((cfg or {}).get("comparables", {}).get("max_comps", 12)) if cfg else 12
clean: list[Comparable] = []
for c in comps[:max_comps]:
if c.sale_date > cutoff:
continue # strict as-of rule, defensive
clean.append(c)
clean = [c.model_copy(update={"evidence_id": f"C{i + 1:02d}"}) for i, c in enumerate(clean)]
notes = [
f"Valuation date {sale_date.isoformat()}; information cutoff {cutoff.isoformat()}: evidence dated after the cutoff is excluded.",
"No inspection has been performed; condition, renovations and defects are unobserved for every role.",
"The subject's sale price is not in the packet and is never disclosed.",
]
if subject.roll_reference_date:
notes.append(f"Roll values are those in force at the sale date; their market reference date is {subject.roll_reference_date}.")
if not clean:
notes.append("No comparable sale satisfied the retrieval rule before the cutoff.")
if geo is None:
notes.append("No geospatial context is available for this property.")
pid = hashlib.sha256(f"{subject.tx_id}|{cutoff.isoformat()}|{len(clean)}".encode()).hexdigest()[:12]
packet = EvidencePacket(packet_id=f"pkt_{subject.tx_id}_{pid}", subject=subject, comparables=clean, market=market, geo=geo, notes=notes)
_assert_no_leak(packet)
return packet
def _assert_no_leak(packet: EvidencePacket) -> None:
"""Defensive invariant: the subject carries no sale price and no address-like field."""
subj = packet.subject.model_dump()
for k in subj:
if k in FORBIDDEN_KEYS:
raise AssertionError(f"subject contains forbidden field {k!r}")
txt = packet.model_dump_json().lower()
for tok in ("\"street\"", "\"address\"", "\"civic"):
if tok in txt:
raise AssertionError(f"packet contains forbidden token {tok!r}")
def role_view(packet: EvidencePacket, role: Role) -> dict[str, Any]:
"""Least-privilege view: only the sections the role is entitled to, evidence ids preserved."""
full = packet.model_dump(mode="json")
view: dict[str, Any] = {
"packet_id": packet.packet_id,
"valuation_date": full["subject"]["valuation_date"],
"information_cutoff": full["subject"]["information_cutoff"],
"visible_sections": list(role.visible_sections),
}
ids: list[str] = []
if role.sees("subject"):
view["subject"] = full["subject"]
ids.append(packet.subject.evidence_id)
if role.sees("comparables"):
view["comparables"] = full["comparables"]
ids += [c.evidence_id for c in packet.comparables]
if role.sees("market"):
view["market"] = full["market"]
ids.append(packet.market.evidence_id)
if role.sees("geo"):
view["geo"] = full["geo"] # may be None: absent is absent
if packet.geo is not None:
ids.append(packet.geo.evidence_id)
view["evidence_ids_visible"] = ids
view["notes"] = list(packet.notes)
return view
def visible_ids(packet: EvidencePacket, role: Role) -> set[str]:
return set(role_view(packet, role)["evidence_ids_visible"])
def render_user_prompt(view: dict[str, Any], extra: dict[str, Any] | None = None, *, kind: str = "estimate", verification_flags: list[str] | None = None) -> str:
"""User message = header + + optional + optional flags + task instruction."""
if kind not in _INSTRUCTIONS:
raise ValueError(f"unknown prompt kind {kind!r}")
packet_json = json.dumps(view, ensure_ascii=False, separators=(",", ":"), default=str)
parts = [
f"Evidence packet {view.get('packet_id', '?')} — valuation date {view.get('valuation_date')}, information cutoff {view.get('information_cutoff')}.",
f"Sections visible to your role: {', '.join(view.get('visible_sections', []))}.",
f"{packet_json}",
]
if extra is not None:
parts.append(f"{json.dumps(extra, ensure_ascii=False, separators=(',', ':'), default=str)}")
if verification_flags:
parts.append(
"" + json.dumps(verification_flags, ensure_ascii=False) + "\n"
"A deterministic verification tool flagged the items above in your previous answer (unknown evidence id, adjustment outside ±40 % of the raw comparable price, or point estimate outside the adjusted-comparable hull ±25 % without justification). Correct them or justify them explicitly in key_assumptions."
)
parts.append(_INSTRUCTIONS[kind])
return "\n".join(parts)
def committee_state(round_no: int, estimates: dict[str, AgentEstimate], *, for_role: str | None = None, critiques: list[dict[str, Any]] | None = None, visible: set[str] | None = None) -> dict[str, Any]:
"""Anonymised committee table (no model/provider identity) + committee median + critiques addressed to ``for_role``."""
pts = sorted(e.point_estimate for e in estimates.values())
med = None
if pts:
n = len(pts)
med = pts[n // 2] if n % 2 else 0.5 * (pts[n // 2 - 1] + pts[n // 2])
table = [
{"role": rid, "point": e.point_estimate, "q05": e.q05, "q95": e.q95, "reasoning_summary": e.reasoning_summary[:600]}
for rid, e in estimates.items()
]
state: dict[str, Any] = {"round": round_no, "n_estimates": len(table), "committee_median": med, "estimates": table}
if for_role is not None:
state["your_role"] = for_role
others = sorted({i for rid, e in estimates.items() if rid != for_role for i in e.evidence_ids_used})
if visible is not None:
state["evidence_ids_cited_by_others_outside_your_view"] = [i for i in others if i not in visible]
state["critiques_addressed_to_you"] = critiques or []
return state
# ---------------------------------------------------------------------------------------------------
# deterministic evidence verification (arm H and faithfulness audit)
# ---------------------------------------------------------------------------------------------------
ADJ_BAND = 0.40
HULL_BAND = 0.25
def verify_evidence(estimate: AgentEstimate, packet: EvidencePacket, allowed_ids: set[str] | None = None) -> list[str]:
"""Flags: ``unknown_evidence_id:``, ``unknown_comparable:``, ``adjustment_out_of_band:``,
``point_outside_comparable_hull``. ``allowed_ids`` restricts citations to the role's view when given."""
flags: list[str] = []
ids = packet.evidence_ids() if allowed_ids is None else set(allowed_ids)
for i in estimate.evidence_ids_used:
if i not in ids:
flags.append(f"unknown_evidence_id:{i}")
raw = {c.evidence_id: c.price for c in packet.comparables}
adjusted: list[float] = []
for a in estimate.adjustments:
if a.comparable_id not in raw:
flags.append(f"unknown_comparable:{a.comparable_id}")
continue
p = raw[a.comparable_id]
if p > 0 and abs(a.adjusted_price / p - 1.0) > ADJ_BAND:
flags.append(f"adjustment_out_of_band:{a.comparable_id}")
else:
adjusted.append(a.adjusted_price)
hull = adjusted or [c.price for c in packet.comparables]
if hull:
lo, hi = min(hull) * (1 - HULL_BAND), max(hull) * (1 + HULL_BAND)
if not (lo <= estimate.point_estimate <= hi):
flags.append("point_outside_comparable_hull")
return flags
def faithfulness_summary(estimate: AgentEstimate, packet: EvidencePacket) -> dict[str, Any]:
"""Counts used by the evaluation layer (precision of cited ids, adjustment consistency, hull)."""
ids = packet.evidence_ids()
cited = estimate.evidence_ids_used
ok = sum(1 for i in cited if i in ids)
flags = verify_evidence(estimate, packet)
return {
"n_cited": len(cited),
"n_cited_valid": ok,
"citation_precision": (ok / len(cited)) if cited else None,
"n_adjustments": len(estimate.adjustments),
"n_adjustments_out_of_band": sum(1 for f in flags if f.startswith("adjustment_out_of_band")),
"point_in_hull": "point_outside_comparable_hull" not in flags,
"flags": flags,
}
def sections_valid(sections: tuple[str, ...] | list[str]) -> bool:
return all(s in SECTIONS for s in sections)