diff --git a/scripts/apply_proposal.py b/scripts/apply_proposal.py index f357591..90aaaf2 100644 --- a/scripts/apply_proposal.py +++ b/scripts/apply_proposal.py @@ -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}""" ) diff --git a/scripts/proposal_apply_lifecycle.py b/scripts/proposal_apply_lifecycle.py index 03ab6d0..cc0e79b 100644 --- a/scripts/proposal_apply_lifecycle.py +++ b/scripts/proposal_apply_lifecycle.py @@ -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, diff --git a/tests/test_apply_proposal.py b/tests/test_apply_proposal.py index 96bf7e1..65dad6f 100644 --- a/tests/test_apply_proposal.py +++ b/tests/test_apply_proposal.py @@ -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 diff --git a/tests/test_proposal_apply_lifecycle.py b/tests/test_proposal_apply_lifecycle.py index d05fd00..a5cad00 100644 --- a/tests/test_proposal_apply_lifecycle.py +++ b/tests/test_proposal_apply_lifecycle.py @@ -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 = {