392 lines
15 KiB
Python
392 lines
15 KiB
Python
"""Shared fact ledger for MAF roundtable runs (Phase 3).
|
|
|
|
The ledger is the authoritative record of confirmed facts, tool outcomes, and
|
|
external-action status. Chat text must never be treated as proof that a write
|
|
succeeded — hosts synthesize conclusions from this structure.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
from uuid import uuid4
|
|
|
|
_TOOL_STATUSES = frozenset({"completed", "waiting_approval", "failed", "unknown"})
|
|
_ACTION_STATUSES = frozenset({"pending", "executed", "failed", "unknown", "rejected"})
|
|
|
|
|
|
def empty_ledger() -> dict[str, Any]:
|
|
"""Return a fresh ledger with all Phase-3 partitions.
|
|
|
|
Phase-3 separation of evidence trust levels:
|
|
- ``candidate_facts`` — unverified seat claims; the model said it, that
|
|
does not make it true. Each carries a server-generated ``candidate_id``.
|
|
- ``confirmed_facts`` — only facts a cross-review confirmed (with evidence
|
|
and a confirmer) ever land here.
|
|
- ``tool_results`` — outcomes of tools that actually ran.
|
|
- ``pending_actions`` / ``executed_actions`` / ``failed_or_unknown_actions``
|
|
— external writes partitioned by lifecycle state.
|
|
"""
|
|
return {
|
|
"candidate_facts": [],
|
|
"confirmed_facts": [],
|
|
"evidence": [],
|
|
"tool_results": [],
|
|
"executed_actions": [],
|
|
"pending_actions": [],
|
|
"failed_or_unknown_actions": [],
|
|
"conflicts": [],
|
|
# Legacy alias kept for old checkpoints / prompts.
|
|
"proposed_actions": [],
|
|
}
|
|
|
|
|
|
def normalize_ledger(raw: dict[str, Any] | None) -> dict[str, Any]:
|
|
"""Coerce a possibly-partial / legacy ledger into the full shape (in-place when possible)."""
|
|
if not isinstance(raw, dict):
|
|
return empty_ledger()
|
|
base = empty_ledger()
|
|
for key in base:
|
|
if key not in raw or not isinstance(raw.get(key), list):
|
|
raw[key] = []
|
|
return raw
|
|
|
|
|
|
def candidate_id_map(ledger: dict[str, Any]) -> dict[str, dict[str, Any]]:
|
|
"""Return ``{candidate_id: candidate_entry}`` for O(1) confirmation lookup."""
|
|
ledger = normalize_ledger(ledger)
|
|
return {
|
|
str(c["candidate_id"]): c
|
|
for c in ledger["candidate_facts"]
|
|
if isinstance(c, dict) and c.get("candidate_id")
|
|
}
|
|
|
|
|
|
def known_evidence_refs(ledger: dict[str, Any]) -> set[str]:
|
|
"""The set of evidence references the ledger actually has backing for.
|
|
|
|
A cross-review may only confirm a candidate using evidence that the ledger
|
|
already recorded (either in ``evidence`` or as a successful
|
|
``tool_results`` entry). This is the trust anchor that stops a weak model
|
|
from inventing evidence references to promote a seat claim.
|
|
"""
|
|
ledger = normalize_ledger(ledger)
|
|
refs: set[str] = set()
|
|
for e in ledger["evidence"]:
|
|
if isinstance(e, dict):
|
|
r = e.get("ref") or e.get("text")
|
|
if r:
|
|
refs.add(str(r))
|
|
elif e:
|
|
refs.add(str(e))
|
|
for tr in ledger["tool_results"]:
|
|
# Only successful tool runs count as evidence (a failed tool proves
|
|
# nothing). Each tool result is addressable as tool:<name>:<turn>.
|
|
if isinstance(tr, dict) and tr.get("status") == "completed":
|
|
name = tr.get("tool_name", "")
|
|
turn = tr.get("turn_index")
|
|
if name:
|
|
refs.add(f"tool:{name}:{turn}" if turn is not None else f"tool:{name}")
|
|
return refs
|
|
|
|
|
|
def merge_seat_output(
|
|
ledger: dict[str, Any],
|
|
seat_output: dict[str, Any],
|
|
*,
|
|
agent_id: str,
|
|
) -> dict[str, Any]:
|
|
"""Merge a validated seat structured output into the ledger.
|
|
|
|
Phase-3: seat ``findings`` are written to ``candidate_facts`` (unverified
|
|
claims), NOT to ``confirmed_facts``. Promotion to ``confirmed_facts``
|
|
happens only via :func:`confirm_candidate_facts` after cross-review.
|
|
Proposed actions stay in ``proposed_actions`` — they are never promoted to
|
|
``executed_actions`` here.
|
|
"""
|
|
ledger = normalize_ledger(ledger)
|
|
for finding in seat_output.get("findings") or []:
|
|
text = str(finding).strip()
|
|
if text:
|
|
ledger["candidate_facts"].append({
|
|
"candidate_id": uuid4().hex,
|
|
"text": text,
|
|
"agent_id": agent_id,
|
|
"source": "seat_finding",
|
|
"confirmed": False,
|
|
})
|
|
for ref in seat_output.get("evidence_refs") or []:
|
|
if isinstance(ref, dict):
|
|
entry = dict(ref)
|
|
entry.setdefault("agent_id", agent_id)
|
|
ledger["evidence"].append(entry)
|
|
elif ref:
|
|
ledger["evidence"].append({"ref": str(ref), "agent_id": agent_id})
|
|
for conflict in seat_output.get("conflicts") or []:
|
|
if conflict:
|
|
ledger["conflicts"].append({"text": str(conflict), "agent_id": agent_id})
|
|
for action in seat_output.get("proposed_actions") or []:
|
|
if isinstance(action, dict):
|
|
entry = dict(action)
|
|
entry.setdefault("agent_id", agent_id)
|
|
entry.setdefault("status", "proposed")
|
|
ledger["proposed_actions"].append(entry)
|
|
elif action:
|
|
ledger["proposed_actions"].append({"text": str(action), "agent_id": agent_id, "status": "proposed"})
|
|
# Executed actions from model text are NOT trusted — only ActionCoordinator
|
|
# / resume may write executed_actions. Ignore seat_output.executed_actions.
|
|
return ledger
|
|
|
|
|
|
def confirm_candidates(
|
|
ledger: dict[str, Any],
|
|
*,
|
|
confirmed_candidates: list[dict[str, Any]],
|
|
confirmed_by: str,
|
|
rejections: list[dict[str, Any]] | None = None,
|
|
accepted: list[dict[str, Any]] | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Promote candidate facts to ``confirmed_facts`` after cross-review.
|
|
|
|
Phase-3 P1-1 — server-side evidence binding. A candidate is promoted ONLY
|
|
when **all** of the following hold:
|
|
|
|
1. ``candidate_id`` exists in ``candidate_facts`` (the model cannot invent a
|
|
confirmation for a fact a seat never asserted);
|
|
2. the candidate carries **at least one** evidence reference;
|
|
3. every evidence reference resolves to an entry the ledger actually
|
|
recorded — either in ``evidence`` or as a **successful** ``tool_results``
|
|
entry. An evidence ref the model invented (not in the ledger) is a
|
|
hallucination and blocks promotion.
|
|
|
|
Candidates that fail any check are left as candidates (``confirmed=False``)
|
|
and a rejection reason is appended to ``rejections`` so the cross-review
|
|
summary can explain why a claim was not trusted. When supplied, ``accepted``
|
|
receives only the server-validated confirmations, so callers never need to
|
|
infer success from the model's requested confirmation list. Idempotent: a
|
|
candidate already present in ``confirmed_facts`` is not re-confirmed.
|
|
"""
|
|
ledger = normalize_ledger(ledger)
|
|
if rejections is None:
|
|
rejections = []
|
|
by_id = candidate_id_map(ledger)
|
|
valid_refs = known_evidence_refs(ledger)
|
|
existing_ids = {
|
|
str(f.get("candidate_id"))
|
|
for f in ledger["confirmed_facts"]
|
|
if isinstance(f, dict) and f.get("candidate_id")
|
|
}
|
|
for item in confirmed_candidates or []:
|
|
if not isinstance(item, dict):
|
|
continue
|
|
cid = str(item.get("candidate_id", "")).strip()
|
|
refs = list(item.get("evidence_refs") or [])
|
|
if not cid:
|
|
rejections.append({"reason": "missing_candidate_id"})
|
|
continue
|
|
cand = by_id.get(cid)
|
|
if cand is None:
|
|
rejections.append({"candidate_id": cid, "reason": "unknown_candidate_id"})
|
|
continue
|
|
if cid in existing_ids:
|
|
# Already confirmed — idempotent no-op.
|
|
continue
|
|
if not refs:
|
|
rejections.append({"candidate_id": cid, "reason": "no_evidence"})
|
|
continue
|
|
bad_refs = [r for r in refs if str(r) not in valid_refs]
|
|
if bad_refs:
|
|
rejections.append({
|
|
"candidate_id": cid,
|
|
"reason": "evidence_not_in_ledger",
|
|
"evidence_refs": bad_refs,
|
|
})
|
|
continue
|
|
cand["confirmed"] = True
|
|
ledger["confirmed_facts"].append({
|
|
"candidate_id": cid,
|
|
"text": cand.get("text", ""),
|
|
"agent_id": cand.get("agent_id"),
|
|
"source": "cross_review_confirmed",
|
|
"confirmed_by": confirmed_by,
|
|
"evidence_refs": list(refs),
|
|
})
|
|
if accepted is not None:
|
|
accepted.append({
|
|
"candidate_id": cid,
|
|
"evidence_refs": list(refs),
|
|
})
|
|
existing_ids.add(cid)
|
|
return ledger
|
|
|
|
|
|
def confirm_candidate_facts(
|
|
ledger: dict[str, Any],
|
|
*,
|
|
confirmed_candidate_ids: list[str],
|
|
confirmed_by: str,
|
|
evidence_refs: list[str] | None = None,
|
|
accepted: list[dict[str, Any]] | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Back-compat wrapper: confirm a flat id list against one shared evidence set.
|
|
|
|
Phase-3 P1-1 tightened confirmation so each candidate must carry at least
|
|
one evidence reference that the ledger has backing for. The structured
|
|
path (:func:`confirm_candidates`) is the primary API; this wrapper exists
|
|
for callers that still pass a flat ``confirmed_candidate_ids`` list — it
|
|
promotes a candidate ONLY when ``evidence_refs`` is non-empty AND every
|
|
ref resolves to ledger evidence / successful tool results.
|
|
"""
|
|
shared_refs = list(evidence_refs or [])
|
|
candidates = [
|
|
{"candidate_id": cid, "evidence_refs": shared_refs}
|
|
for cid in confirmed_candidate_ids or []
|
|
]
|
|
return confirm_candidates(
|
|
ledger,
|
|
confirmed_candidates=candidates,
|
|
confirmed_by=confirmed_by,
|
|
accepted=accepted,
|
|
)
|
|
|
|
|
|
def record_tool_result(
|
|
ledger: dict[str, Any],
|
|
*,
|
|
tool_name: str,
|
|
status: str,
|
|
agent_id: str,
|
|
turn_index: int | None = None,
|
|
detail: dict[str, Any] | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Append a tool result. ``status`` must be one of the allowed tool statuses."""
|
|
ledger = normalize_ledger(ledger)
|
|
st = status if status in _TOOL_STATUSES else "unknown"
|
|
entry: dict[str, Any] = {
|
|
"tool_name": tool_name,
|
|
"status": st,
|
|
"agent_id": agent_id,
|
|
}
|
|
if turn_index is not None:
|
|
entry["turn_index"] = turn_index
|
|
if detail:
|
|
entry["detail"] = detail
|
|
ledger["tool_results"].append(entry)
|
|
return ledger
|
|
|
|
|
|
def record_action_transition(
|
|
ledger: dict[str, Any],
|
|
*,
|
|
action_id: str,
|
|
tool_name: str,
|
|
status: str,
|
|
agent_id: str | None = None,
|
|
external_object_id: str | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Move / record an external action under the correct ledger partition.
|
|
|
|
Rules:
|
|
- ``proposed`` never becomes ``executed`` via this helper unless status is
|
|
explicitly ``executed`` (called only after ToolExecutor success).
|
|
- ``pending`` entries are removed when transitioning to a terminal status.
|
|
"""
|
|
ledger = normalize_ledger(ledger)
|
|
st = status if status in _ACTION_STATUSES else "unknown"
|
|
entry: dict[str, Any] = {
|
|
"action_id": action_id,
|
|
"tool_name": tool_name,
|
|
"status": st,
|
|
}
|
|
if agent_id:
|
|
entry["agent_id"] = agent_id
|
|
if external_object_id:
|
|
entry["external_object_id"] = external_object_id
|
|
|
|
# Drop matching pending rows when leaving pending.
|
|
if st != "pending":
|
|
ledger["pending_actions"] = [
|
|
p
|
|
for p in ledger["pending_actions"]
|
|
if not (isinstance(p, dict) and p.get("action_id") == action_id)
|
|
]
|
|
|
|
if st == "pending":
|
|
# Replace existing pending with same action_id.
|
|
ledger["pending_actions"] = [
|
|
p
|
|
for p in ledger["pending_actions"]
|
|
if not (isinstance(p, dict) and p.get("action_id") == action_id)
|
|
]
|
|
ledger["pending_actions"].append(entry)
|
|
elif st == "executed":
|
|
ledger["executed_actions"] = [
|
|
e
|
|
for e in ledger["executed_actions"]
|
|
if not (isinstance(e, dict) and e.get("action_id") == action_id)
|
|
]
|
|
ledger["executed_actions"].append(entry)
|
|
# Strip from failed list if previously unknown/failed.
|
|
ledger["failed_or_unknown_actions"] = [
|
|
e
|
|
for e in ledger["failed_or_unknown_actions"]
|
|
if not (isinstance(e, dict) and e.get("action_id") == action_id)
|
|
]
|
|
elif st in ("failed", "unknown", "rejected"):
|
|
ledger["failed_or_unknown_actions"] = [
|
|
e
|
|
for e in ledger["failed_or_unknown_actions"]
|
|
if not (isinstance(e, dict) and e.get("action_id") == action_id)
|
|
]
|
|
ledger["failed_or_unknown_actions"].append(entry)
|
|
# Never keep executed copy of a failed/unknown action.
|
|
ledger["executed_actions"] = [
|
|
e
|
|
for e in ledger["executed_actions"]
|
|
if not (isinstance(e, dict) and e.get("action_id") == action_id)
|
|
]
|
|
return ledger
|
|
|
|
|
|
def assert_ledger_invariants(ledger: dict[str, Any]) -> None:
|
|
"""Raise AssertionError if the ledger violates evidence/action constraints."""
|
|
ledger = normalize_ledger(ledger)
|
|
executed_ids = {
|
|
e.get("action_id")
|
|
for e in ledger["executed_actions"]
|
|
if isinstance(e, dict) and e.get("action_id")
|
|
}
|
|
for prop in ledger["proposed_actions"]:
|
|
if not isinstance(prop, dict):
|
|
continue
|
|
# A proposed-only entry must not also appear as executed with status proposed.
|
|
if prop.get("status") == "proposed" and prop.get("action_id") in executed_ids:
|
|
# Same action_id may be proposed then later executed — that is OK if
|
|
# the executed entry has status executed. Only flag if proposed row
|
|
# itself claims executed.
|
|
if prop.get("status") == "executed":
|
|
raise AssertionError("proposed_actions entry has status=executed")
|
|
for e in ledger["executed_actions"]:
|
|
if isinstance(e, dict) and e.get("status") not in (None, "executed"):
|
|
raise AssertionError(f"executed_actions entry has status={e.get('status')!r}")
|
|
for tr in ledger["tool_results"]:
|
|
if isinstance(tr, dict) and tr.get("status") not in _TOOL_STATUSES:
|
|
raise AssertionError(f"invalid tool_results status={tr.get('status')!r}")
|
|
|
|
|
|
def proposed_action_ids(ledger: dict[str, Any]) -> set[str]:
|
|
ledger = normalize_ledger(ledger)
|
|
return {
|
|
str(p["action_id"])
|
|
for p in ledger["proposed_actions"]
|
|
if isinstance(p, dict) and p.get("action_id")
|
|
}
|
|
|
|
|
|
def executed_action_ids(ledger: dict[str, Any]) -> set[str]:
|
|
ledger = normalize_ledger(ledger)
|
|
return {
|
|
str(e["action_id"])
|
|
for e in ledger["executed_actions"]
|
|
if isinstance(e, dict) and e.get("action_id")
|
|
}
|