teleo/teleo-infrastructure
8
0
Fork 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions 1

Apply evidence against the live composite schema

Browse source
This commit is contained in:
twentyOne2x 2026-07-15 05:20:34 +02:00
parent ed827626db
commit 000bda06c2
4 changed files with 80 additions and 42 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

16
scripts/apply_proposal.py
View file

@ -439,10 +439,9 @@ def build_attach_evidence_sql(
if role not in EVIDENCE_ROLES:
raise ValueError(f"attach_evidence.evidence[{i}].role must be one of {sorted(EVIDENCE_ROLES)}")
weight = _weight(ev.get("weight"), f"attach_evidence.evidence[{i}].weight")
row_id = deterministic_row_uuid(proposal_id, "claim_evidence", claim_id, source_id, role)
statements.append(
f"""insert into public.claim_evidence (id, claim_id, source_id, role, weight)
select {sql_literal(row_id)}::uuid, {sql_literal(claim_id)}::uuid, {sql_literal(source_id)}::uuid,
f"""insert into public.claim_evidence (claim_id, source_id, role, weight)
select {sql_literal(claim_id)}::uuid, {sql_literal(source_id)}::uuid,
{sql_literal(role)}::evidence_role, {sql_literal(weight)}
on conflict (claim_id, source_id, role) do nothing;"""
)
@ -594,13 +593,6 @@ def build_approve_claim_sql(
raise ValueError(f"{path}.role must be one of {sorted(EVIDENCE_ROLES)}")
evidence.append(
{
"id": deterministic_row_uuid(
proposal_id,
"claim_evidence",
_uuid(row.get("claim_id"), f"{path}.claim_id"),
_uuid(row.get("source_id"), f"{path}.source_id"),
role,
),
"claim_id": _uuid(row.get("claim_id"), f"{path}.claim_id"),
"source_id": _uuid(row.get("source_id"), f"{path}.source_id"),
"role": role,
@ -781,8 +773,8 @@ select {sql_literal(row["id"])}::uuid, {sql_literal(row["agent_id"])}::uuid,
for row in evidence:
conflict_clause = ";" if fresh_owned else "\non conflict (claim_id, source_id, role) do nothing;"
statements.append(
f"""insert into public.claim_evidence (id, claim_id, source_id, role, weight, created_by)
select {sql_literal(row["id"])}::uuid, {sql_literal(row["claim_id"])}::uuid, {sql_literal(row["source_id"])}::uuid,
f"""insert into public.claim_evidence (claim_id, source_id, role, weight, created_by)
select {sql_literal(row["claim_id"])}::uuid, {sql_literal(row["source_id"])}::uuid,
{sql_literal(row["role"])}::evidence_role, {sql_literal(row["weight"])},
{sql_literal(row["created_by"])}::uuid{conflict_clause}"""
)

76
scripts/proposal_apply_lifecycle.py
View file

@ -69,6 +69,14 @@ DELETE_ORDER = (
)
def canonical_row_key(family: str, row: dict[str, Any]) -> str | None:
if family == "claim_evidence":
values = (row.get("claim_id"), row.get("source_id"), row.get("role"))
return "|".join(str(value) for value in values) if all(values) else None
identifier = row.get("id")
return str(identifier) if identifier else None
def canonical_json(value: Any) -> str:
return json.dumps(value, ensure_ascii=False, separators=(",", ":"), sort_keys=True)
@ -288,9 +296,9 @@ def _require_apply_receipt(
for name, table_rows in rows.items():
if not isinstance(table_rows, list) or any(not isinstance(row, dict) for row in table_rows):
raise ValueError(f"apply receipt {name} rows are malformed")
identifiers = [row.get("id") for row in table_rows]
if any(not identifier for identifier in identifiers) or len(set(identifiers)) != len(identifiers):
raise ValueError(f"apply receipt {name} rows require unique persisted ids")
row_keys = [canonical_row_key(name, row) for row in table_rows]
if any(not row_key for row_key in row_keys) or len(set(row_keys)) != len(row_keys):
raise ValueError(f"apply receipt {name} rows require unique canonical keys")
try:
replay_receipt.validate_replay_receipt(
apply_receipt,
@ -299,16 +307,6 @@ def _require_apply_receipt(
except ValueError as exc:
raise ValueError("apply receipt failed full replay-contract validation") from exc
proposal_id = str(proposal_before.get("id"))
for row in rows["claim_evidence"]:
expected_id = ap.deterministic_row_uuid(
proposal_id,
"claim_evidence",
row.get("claim_id"),
row.get("source_id"),
row.get("role"),
)
if row.get("id") != expected_id:
raise ValueError("fresh apply receipt evidence id is not deterministic")
for row in rows["claim_edges"]:
expected_id = ap.deterministic_row_uuid(
proposal_id,
@ -322,8 +320,29 @@ def _require_apply_receipt(
return rows
def _row_assertion(table: str, row: dict[str, Any], label: str) -> str:
def _evidence_predicate(alias: str, rows: list[dict[str, Any]]) -> str:
if not rows:
return "false"
return (
"("
+ " or ".join(
f"({alias}.claim_id = {ap.sql_literal(row['claim_id'])}::uuid "
f"and {alias}.source_id = {ap.sql_literal(row['source_id'])}::uuid "
f"and {alias}.role = {ap.sql_literal(row['role'])}::evidence_role)"
for row in rows
)
+ ")"
)
def _row_assertion(family: str, table: str, row: dict[str, Any], label: str) -> str:
row_json = ap.sql_literal(canonical_json(row))
if family == "claim_evidence":
predicate = _evidence_predicate("t", [row])
return f""" if (select count(*) from {table} t
where {predicate} and to_jsonb(t) = {row_json}::jsonb) <> 1 then
raise exception 'proposal rollback drift: {label} row does not match apply receipt';
end if;"""
row_id = ap.sql_literal(row["id"])
return f""" if (select count(*) from {table} t
where t.id = {row_id}::uuid and to_jsonb(t) = {row_json}::jsonb) <> 1 then
@ -331,9 +350,16 @@ def _row_assertion(table: str, row: dict[str, Any], label: str) -> str:
end if;"""
def _delete_statement(table: str, rows: list[dict[str, Any]], label: str) -> str:
def _delete_statement(family: str, table: str, rows: list[dict[str, Any]], label: str) -> str:
if not rows:
return f" -- no {label} rows owned by this proposal"
if family == "claim_evidence":
predicate = _evidence_predicate("target", rows)
return f""" delete from {table} target where {predicate};
get diagnostics affected = row_count;
if affected <> {len(rows)} then
raise exception 'proposal rollback count mismatch for {label}: expected {len(rows)}, got %', affected;
end if;"""
identifiers = ", ".join(f"{ap.sql_literal(row['id'])}::uuid" for row in rows)
return f""" delete from {table} where id in ({identifiers});
get diagnostics affected = row_count;
@ -403,16 +429,16 @@ def build_compensating_rollback_sql(
approval_json = ap.sql_literal(canonical_json(approval_snapshot))
claim_ids = _uuid_array([str(row["id"]) for row in rows["claims"]])
source_ids = _uuid_array([str(row["id"]) for row in rows["sources"]])
evidence_ids = _uuid_array([str(row["id"]) for row in rows["claim_evidence"]])
owned_evidence_predicate = _evidence_predicate("ce", rows["claim_evidence"])
edge_ids = _uuid_array([str(row["id"]) for row in rows["claim_edges"]])
row_assertions = []
for family in DELETE_ORDER:
table = ROW_TABLES[family]
row_assertions.extend(
_row_assertion(table, row, f"{family}[{index}]") for index, row in enumerate(rows[family])
_row_assertion(family, table, row, f"{family}[{index}]") for index, row in enumerate(rows[family])
)
deletions = [_delete_statement(ROW_TABLES[family], rows[family], family) for family in DELETE_ORDER]
deletions = [_delete_statement(family, ROW_TABLES[family], rows[family], family) for family in DELETE_ORDER]
return f"""begin;
set local standard_conforming_strings = on;
@ -466,7 +492,7 @@ begin
select ce.id
from public.claim_evidence ce
where (ce.claim_id = any({claim_ids}) or ce.source_id = any({source_ids}))
and ce.id <> all({evidence_ids})
and not ({owned_evidence_predicate})
union all
select edge.id
from public.claim_edges edge
@ -616,8 +642,8 @@ def validate_production_rollback_artifact(
return artifact
def _row_ids(rows: dict[str, list[dict[str, Any]]]) -> dict[str, list[str]]:
return {name: sorted(str(row["id"]) for row in table_rows) for name, table_rows in rows.items()}
def _row_keys(rows: dict[str, list[dict[str, Any]]]) -> dict[str, list[str]]:
return {name: sorted(str(canonical_row_key(name, row)) for row in table_rows) for name, table_rows in rows.items()}
def build_lifecycle_receipt(
@ -709,7 +735,7 @@ def build_lifecycle_receipt(
"apply_receipt_sha256": sha256_json(apply_receipt),
"apply_sql_sha256": apply_receipt["apply_engine"]["apply_sql_sha256"],
"fresh_owned_apply_sql_sha256": expected_fresh_apply_sha256,
"applied_row_ids": _row_ids(rows),
"applied_row_keys": _row_keys(rows),
"applied_row_sha256": {name: [sha256_json(row) for row in table_rows] for name, table_rows in rows.items()},
"rollback": {
"schema": ROLLBACK_SCHEMA,
@ -764,7 +790,7 @@ def _require_lifecycle_receipt(lifecycle_receipt: dict[str, Any]) -> None:
"apply_receipt_sha256",
"apply_sql_sha256",
"fresh_owned_apply_sql_sha256",
"applied_row_ids",
"applied_row_keys",
"applied_row_sha256",
"rollback",
"checks",
@ -816,7 +842,7 @@ def _require_lifecycle_receipt(lifecycle_receipt: dict[str, Any]) -> None:
raise ValueError("lifecycle receipt is missing an exact SHA-256 binding")
if lifecycle_receipt.get("apply_sql_sha256") != lifecycle_receipt.get("fresh_owned_apply_sql_sha256"):
raise ValueError("lifecycle receipt apply SQL did not consume the fresh-owned preflight")
row_ids = lifecycle_receipt.get("applied_row_ids")
row_ids = lifecycle_receipt.get("applied_row_keys")
row_hashes = lifecycle_receipt.get("applied_row_sha256")
if (
not isinstance(row_ids, dict)
@ -1097,7 +1123,7 @@ def build_authorization_packet(
"conditional_rollback_requires_explicit_authorization": True,
},
"expected_rows": {
"row_ids": lifecycle_receipt["applied_row_ids"],
"row_keys": lifecycle_receipt["applied_row_keys"],
"row_sha256": lifecycle_receipt["applied_row_sha256"],
},
"production_preconditions": production_preconditions,

4
tests/test_apply_proposal.py
View file

@ -149,6 +149,8 @@ def test_attach_evidence_sql_with_source_id():
payload = {"evidence": [{"claim_id": CLAIM_A, "source_id": SOURCE_A, "role": "grounds", "weight": 0.78}]}
sql = ap.build_attach_evidence_sql(payload, "pid-3", None, _approval_meta("attach_evidence", payload))
assert "insert into public.claim_evidence" in sql
assert "public.claim_evidence (id," not in sql
assert "public.claim_evidence (claim_id, source_id, role, weight)" in sql
assert "::evidence_role" in sql
assert "not exists" in sql
@ -251,6 +253,8 @@ def test_approve_claim_bundle_sql_is_atomic_and_row_verified():
assert "insert into public.claims" in sql
assert "insert into public.sources" in sql
assert "insert into public.claim_evidence" in sql
assert "public.claim_evidence (id," not in sql
assert "public.claim_evidence (claim_id, source_id, role, weight, created_by)" in sql
assert "insert into public.claim_edges" in sql
assert "insert into public.reasoning_tools" in sql
assert "source hash collision" in sql

26
tests/test_proposal_apply_lifecycle.py
View file

@ -139,7 +139,6 @@ def _approval_snapshot(proposal: dict) -> dict:
def _canonical_rows() -> dict[str, list[dict]]:
evidence_id = apply.deterministic_row_uuid(PROPOSAL_ID, "claim_evidence", CLAIM_A, SOURCE_ID, "grounds")
edge_id = apply.deterministic_row_uuid(PROPOSAL_ID, "claim_edge", CLAIM_A, CLAIM_B, "supports")
return {
"claims": [
@ -182,7 +181,6 @@ def _canonical_rows() -> dict[str, list[dict]]:
],
"claim_evidence": [
{
"id": evidence_id,
"claim_id": CLAIM_A,
"source_id": SOURCE_ID,
"role": "grounds",
@ -565,11 +563,10 @@ def test_rollback_sql_hash_binds_every_receipt_observed_field():
assert "proposal rollback drift: claims[0] row does not match apply receipt" in changed_sql
@pytest.mark.parametrize("family", ("claim_evidence", "claim_edges"))
def test_fresh_apply_receipt_requires_deterministic_generated_row_ids(family: str):
def test_fresh_apply_receipt_requires_deterministic_generated_edge_ids():
proposal, approval, preflight, receipt, _rollback_sql = _materials()
changed_rows = copy.deepcopy(receipt["canonical_rows"])
changed_rows[family][0]["id"] = "eeeeeeee-eeee-4eee-8eee-eeeeeeeeeeee"
changed_rows["claim_edges"][0]["id"] = "eeeeeeee-eeee-4eee-8eee-eeeeeeeeeeee"
changed_receipt = _apply_receipt(proposal, changed_rows, preflight)
with pytest.raises(ValueError, match="is not deterministic"):
@ -581,6 +578,25 @@ def test_fresh_apply_receipt_requires_deterministic_generated_row_ids(family: st
)
def test_fresh_apply_receipt_uses_composite_claim_evidence_key_without_id():
proposal, approval, preflight, receipt, _rollback_sql = _materials()
evidence = receipt["canonical_rows"]["claim_evidence"][0]
assert "id" not in evidence
assert lifecycle.canonical_row_key("claim_evidence", evidence) == f"{CLAIM_A}|{SOURCE_ID}|grounds"
changed_rows = copy.deepcopy(receipt["canonical_rows"])
changed_rows["claim_evidence"].append(copy.deepcopy(evidence))
changed_receipt = _apply_receipt(proposal, changed_rows, preflight)
with pytest.raises(ValueError, match=r"unique canonical keys|replay-contract validation"):
lifecycle.build_compensating_rollback_sql(
proposal,
approval,
preflight,
changed_receipt,
)
def test_production_rollback_artifact_reconstructs_exact_sql_and_rejects_noop_substitution():
proposal, approval, preflight, apply_receipt, rollback_sql = _materials()
destination = {