Prove Leo KB composition and approved apply lifecycle
This commit is contained in:
twentyOne2x
parent
bd5f1e5bc4
commit
3d5b0b75ef
9 changed files with 1752 additions and 455 deletions
File diff suppressed because it is too large
Load diff
39
fixtures/working-leo/document-ingestion-v1.json
Normal file
39
fixtures/working-leo/document-ingestion-v1.json
Normal file
|
|
@ -0,0 +1,39 @@
|
|||
{
|
||||
"schema": "livingip.workingLeoDocumentIngestionFixture.v1",
|
||||
"source": {
|
||||
"identity": "document:working-leo-review-gated-composition-v1",
|
||||
"source_key": "operating_document",
|
||||
"source_type": "article",
|
||||
"title": "Working Leo review-gated composition note",
|
||||
"locator": "fixture://working-leo/document-ingestion-v1",
|
||||
"body": "Working Leo composition note. A newly captured document may produce claim and evidence candidates, but those candidates are not canonical knowledge.\n\nA staged proposal remains non-canonical until an operator reviews it and a separate guarded apply succeeds. The proposal must retain the source content hash, exact evidence excerpt, and source identity so reviewers can trace every planned row back to the captured artifact.\n\nFor a staging-only rehearsal, pending_review is the terminal state. No public knowledge-base row should be written.",
|
||||
"metadata": {
|
||||
"author": "Working Leo canary fixture",
|
||||
"captured_at": "2026-07-12T00:00:00Z",
|
||||
"document_kind": "operating_note",
|
||||
"language": "en"
|
||||
}
|
||||
},
|
||||
"extraction": {
|
||||
"claim": {
|
||||
"claim_key": "staged_proposal_remains_noncanonical",
|
||||
"type": "structural",
|
||||
"confidence": 0.99,
|
||||
"text": "A staged knowledge proposal remains non-canonical until review and guarded apply succeed.",
|
||||
"body": "A staged proposal remains non-canonical until an operator reviews it and a separate guarded apply succeeds.",
|
||||
"metadata": {
|
||||
"extractor": "deterministic_fixture_replay",
|
||||
"extraction_version": 1,
|
||||
"needs_research": false
|
||||
}
|
||||
},
|
||||
"evidence": {
|
||||
"role": "grounds",
|
||||
"body": "A staged proposal remains non-canonical until an operator reviews it and a separate guarded apply succeeds.",
|
||||
"metadata": {
|
||||
"evidence_kind": "exact_quote",
|
||||
"section": "paragraph_2"
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -85,6 +85,7 @@ from __future__ import annotations
|
|||
|
||||
import argparse
|
||||
import json
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
import uuid
|
||||
|
|
@ -857,8 +858,9 @@ def load_password(secrets_file: str) -> str:
|
|||
|
||||
|
||||
def run_psql(args: argparse.Namespace, sql: str, password: str) -> str:
|
||||
docker_binary = shutil.which("docker") or "docker"
|
||||
command = [
|
||||
"docker", "exec", "-e", "PGPASSWORD", "-i", args.container,
|
||||
docker_binary, "exec", "-e", "PGPASSWORD", "-i", args.container,
|
||||
"psql", "-U", args.role, "-h", args.host, "-d", args.db,
|
||||
"-v", "ON_ERROR_STOP=1", "-At", "-q",
|
||||
]
|
||||
|
|
|
|||
|
|
@ -15,6 +15,7 @@ import argparse
|
|||
import hashlib
|
||||
import json
|
||||
import re
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
import uuid
|
||||
|
|
@ -135,7 +136,7 @@ def _role_psql(
|
|||
password: str,
|
||||
) -> subprocess.CompletedProcess[str]:
|
||||
command = [
|
||||
"docker",
|
||||
shutil.which("docker") or "docker",
|
||||
"exec",
|
||||
"-e",
|
||||
"PGPASSWORD",
|
||||
|
|
@ -213,26 +214,34 @@ def _schema_copy(container: str, source_db: str, clone_db: str) -> None:
|
|||
|
||||
|
||||
def _service_state(service: str) -> dict[str, Any]:
|
||||
result = _run(
|
||||
[
|
||||
"systemctl",
|
||||
"show",
|
||||
service,
|
||||
"--no-pager",
|
||||
"-p",
|
||||
"ActiveState",
|
||||
"-p",
|
||||
"SubState",
|
||||
"-p",
|
||||
"MainPID",
|
||||
"-p",
|
||||
"NRestarts",
|
||||
"-p",
|
||||
"ExecMainStartTimestamp",
|
||||
],
|
||||
check=False,
|
||||
)
|
||||
state: dict[str, Any] = {"returncode": result.returncode}
|
||||
command = [
|
||||
"systemctl",
|
||||
"show",
|
||||
service,
|
||||
"--no-pager",
|
||||
"-p",
|
||||
"ActiveState",
|
||||
"-p",
|
||||
"SubState",
|
||||
"-p",
|
||||
"MainPID",
|
||||
"-p",
|
||||
"NRestarts",
|
||||
"-p",
|
||||
"ExecMainStartTimestamp",
|
||||
]
|
||||
try:
|
||||
result = _run(command, check=False)
|
||||
except FileNotFoundError:
|
||||
return {
|
||||
"available": False,
|
||||
"returncode": 127,
|
||||
"reason": "systemctl_unavailable",
|
||||
}
|
||||
state: dict[str, Any] = {
|
||||
"available": result.returncode == 0,
|
||||
"returncode": result.returncode,
|
||||
}
|
||||
for line in result.stdout.splitlines():
|
||||
key, separator, value = line.partition("=")
|
||||
if separator:
|
||||
|
|
@ -826,6 +835,108 @@ def _table_deltas(before: dict[str, int], after: dict[str, int]) -> dict[str, in
|
|||
return {key: after[key] - before[key] for key in sorted(before)}
|
||||
|
||||
|
||||
def _canonical_row_keys(rows: dict[str, list[dict[str, Any]]]) -> dict[str, Any]:
|
||||
"""Return stable primary/composite keys for every canonical row in a bundle."""
|
||||
return {
|
||||
"public.claims": sorted(row["id"] for row in rows.get("claims", [])),
|
||||
"public.sources": sorted(row["id"] for row in rows.get("sources", [])),
|
||||
"public.claim_evidence": sorted(
|
||||
[
|
||||
{
|
||||
"claim_id": row["claim_id"],
|
||||
"source_id": row["source_id"],
|
||||
"role": row["role"],
|
||||
}
|
||||
for row in rows.get("claim_evidence", [])
|
||||
],
|
||||
key=lambda row: (row["claim_id"], row["source_id"], row["role"]),
|
||||
),
|
||||
"public.claim_edges": sorted(
|
||||
[
|
||||
{
|
||||
"from_claim": row["from_claim"],
|
||||
"to_claim": row["to_claim"],
|
||||
"edge_type": row["edge_type"],
|
||||
}
|
||||
for row in rows.get("claim_edges", [])
|
||||
],
|
||||
key=lambda row: (row["from_claim"], row["to_claim"], row["edge_type"]),
|
||||
),
|
||||
"public.reasoning_tools": sorted(
|
||||
row["id"] for row in rows.get("reasoning_tools", [])
|
||||
),
|
||||
}
|
||||
|
||||
|
||||
def _strict_receipt(
|
||||
result: dict[str, Any],
|
||||
expected_rows: dict[str, list[dict[str, Any]]],
|
||||
expected_table_deltas: dict[str, int],
|
||||
) -> dict[str, Any]:
|
||||
"""Build the compact pass/fail surface from actual database readbacks."""
|
||||
applied = result.get("apply_transition", {}).get("applied_readback") or {}
|
||||
actual_rows = result.get("row_readback") or {}
|
||||
expected_keys = _canonical_row_keys(expected_rows)
|
||||
actual_keys = _canonical_row_keys(actual_rows)
|
||||
checks = {
|
||||
"proposal_id_exact": applied.get("id")
|
||||
== result.get("fixture_ids", {}).get("proposal_id"),
|
||||
"proposal_status_applied": applied.get("status") == "applied",
|
||||
"proposal_applied_at_present": bool(applied.get("applied_at")),
|
||||
"canonical_rows_exact": actual_rows == expected_rows,
|
||||
"canonical_row_keys_exact": actual_keys == expected_keys,
|
||||
"before_after_counts_present": (
|
||||
isinstance(result.get("clone_counts_before_apply"), dict)
|
||||
and isinstance(result.get("clone_counts_after_apply"), dict)
|
||||
),
|
||||
"table_deltas_exact": result.get("clone_apply_table_deltas")
|
||||
== expected_table_deltas,
|
||||
"conflict_transaction_rolled_back": result.get("checks", {}).get(
|
||||
"conflict_transaction_rolled_back_exactly"
|
||||
)
|
||||
is True,
|
||||
"clone_database_removed": (
|
||||
result.get("clone_dropped") is True
|
||||
and result.get("cleanup_readback", {}).get(
|
||||
"leftover_clone_database_count"
|
||||
)
|
||||
== 0
|
||||
),
|
||||
"source_database_unchanged": result.get("source_database_mutated") is False,
|
||||
}
|
||||
return {
|
||||
"schema": "working-leo.approved-change-strict-receipt.v1",
|
||||
"pass": result.get("status") == "pass" and all(checks.values()),
|
||||
"checks": checks,
|
||||
"proposal": {
|
||||
"id": applied.get("id"),
|
||||
"status": applied.get("status"),
|
||||
"applied_at": applied.get("applied_at"),
|
||||
"applied_by_handle": applied.get("applied_by_handle"),
|
||||
},
|
||||
"canonical_row_keys": {
|
||||
"expected": expected_keys,
|
||||
"actual": actual_keys,
|
||||
},
|
||||
"row_counts": {
|
||||
"before_apply": result.get("clone_counts_before_apply"),
|
||||
"after_apply": result.get("clone_counts_after_apply"),
|
||||
"expected_deltas": expected_table_deltas,
|
||||
"actual_deltas": result.get("clone_apply_table_deltas"),
|
||||
},
|
||||
"rollback_cleanup": {
|
||||
"conflict_transaction_rolled_back": checks[
|
||||
"conflict_transaction_rolled_back"
|
||||
],
|
||||
"clone_database_removed": checks["clone_database_removed"],
|
||||
"leftover_clone_database_count": result.get("cleanup_readback", {}).get(
|
||||
"leftover_clone_database_count"
|
||||
),
|
||||
"source_database_unchanged": checks["source_database_unchanged"],
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def _exact_bundle_readback(
|
||||
container: str, database: str, payload: dict[str, Any]
|
||||
) -> dict[str, list[dict[str, Any]]]:
|
||||
|
|
@ -1905,6 +2016,18 @@ drop database if exists {clone_db};
|
|||
f"{type(exc).__name__}: {exc}"
|
||||
)
|
||||
|
||||
gateway_process_observable = (
|
||||
result.get("service_before", {}).get("available") is True
|
||||
and result.get("service_after", {}).get("available") is True
|
||||
)
|
||||
gateway_process_unchanged = (
|
||||
result.get("service_before", {}).get("MainPID")
|
||||
== result.get("service_after", {}).get("MainPID")
|
||||
and result.get("service_before", {}).get("NRestarts")
|
||||
== result.get("service_after", {}).get("NRestarts")
|
||||
if gateway_process_observable
|
||||
else None
|
||||
)
|
||||
result["checks"] = {
|
||||
"security_definer_dependencies_ready": _security_dependencies_ready(
|
||||
result.get("security_definer_readback") or {}
|
||||
|
|
@ -2058,12 +2181,7 @@ drop database if exists {clone_db};
|
|||
"source_files_unchanged_during_run": (
|
||||
result.get("source_binding", {}).get("unchanged_during_run") is True
|
||||
),
|
||||
"gateway_process_unchanged": (
|
||||
result.get("service_before", {}).get("MainPID")
|
||||
== result.get("service_after", {}).get("MainPID")
|
||||
and result.get("service_before", {}).get("NRestarts")
|
||||
== result.get("service_after", {}).get("NRestarts")
|
||||
),
|
||||
"gateway_process_unchanged": gateway_process_unchanged,
|
||||
"clone_cleanup_complete": (
|
||||
result.get("clone_dropped") is True
|
||||
and result.get("leftover_clone_count") == 0
|
||||
|
|
@ -2072,9 +2190,11 @@ drop database if exists {clone_db};
|
|||
result["source_counts_observed_unchanged"] = (
|
||||
result.get("source_counts_before") == result.get("source_counts_after")
|
||||
)
|
||||
result["gateway_process_observable"] = gateway_process_observable
|
||||
result["status"] = (
|
||||
"pass"
|
||||
if all(result["checks"].values()) and "error" not in result
|
||||
if all(value is True for value in result["checks"].values() if value is not None)
|
||||
and "error" not in result
|
||||
else "fail"
|
||||
)
|
||||
result["source_database_mutated"] = not all(
|
||||
|
|
@ -2085,16 +2205,24 @@ drop database if exists {clone_db};
|
|||
"source_table_deltas_exactly_zero",
|
||||
)
|
||||
)
|
||||
result["service_restarted"] = not result["checks"][
|
||||
"gateway_process_unchanged"
|
||||
]
|
||||
result["service_restarted"] = (
|
||||
None
|
||||
if gateway_process_unchanged is None
|
||||
else not gateway_process_unchanged
|
||||
)
|
||||
result["post_run_cleanup"] = {
|
||||
"leftover_clone_database_count": result.get("leftover_clone_count"),
|
||||
"gateway_process_observable": gateway_process_observable,
|
||||
"gateway_active_state": result.get("service_after", {}).get("ActiveState"),
|
||||
"gateway_sub_state": result.get("service_after", {}).get("SubState"),
|
||||
"gateway_main_pid": result.get("service_after", {}).get("MainPID"),
|
||||
"gateway_restart_count": result.get("service_after", {}).get("NRestarts"),
|
||||
}
|
||||
result["strict_receipt"] = _strict_receipt(
|
||||
result, expected_rows, expected_table_deltas
|
||||
)
|
||||
if not result["strict_receipt"]["pass"]:
|
||||
result["status"] = "fail"
|
||||
return result
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -6,6 +6,7 @@ source_container="${SOURCE_CONTAINER:-teleo-pg}"
|
|||
source_db="${SOURCE_DB:-teleo}"
|
||||
service="${LEO_SERVICE:-leoclean-gateway.service}"
|
||||
image="${POSTGRES_CANARY_IMAGE:-postgres:16}"
|
||||
source_tier="${SOURCE_TIER:-live-readonly}"
|
||||
normalization_json=""
|
||||
output="${repo_root}/docs/reports/leo-working-state-20260709/approve-claim-clone-canary-current.json"
|
||||
|
||||
|
|
@ -31,6 +32,10 @@ while [[ $# -gt 0 ]]; do
|
|||
service="$2"
|
||||
shift 2
|
||||
;;
|
||||
--source-tier)
|
||||
source_tier="$2"
|
||||
shift 2
|
||||
;;
|
||||
*)
|
||||
echo "unknown argument: $1" >&2
|
||||
exit 2
|
||||
|
|
@ -38,6 +43,22 @@ while [[ $# -gt 0 ]]; do
|
|||
esac
|
||||
done
|
||||
|
||||
if [[ "$source_tier" != "isolated-local" && "$source_tier" != "live-readonly" ]]; then
|
||||
echo "--source-tier must be isolated-local or live-readonly" >&2
|
||||
exit 2
|
||||
fi
|
||||
|
||||
source_network_mode="$(docker inspect -f '{{.HostConfig.NetworkMode}}' "$source_container")"
|
||||
source_canary_label="$(
|
||||
docker inspect -f '{{index .Config.Labels "livingip.canary"}}' "$source_container"
|
||||
)"
|
||||
if [[ "$source_tier" == "isolated-local" ]]; then
|
||||
if [[ "$source_network_mode" != "none" || "$source_canary_label" != "working-leo-approved-source" ]]; then
|
||||
echo "isolated-local source must be networkless and labeled livingip.canary=working-leo-approved-source" >&2
|
||||
exit 2
|
||||
fi
|
||||
fi
|
||||
|
||||
container="working-leo-approve-claim-$RANDOM-$$"
|
||||
workdir="/tmp/${container}"
|
||||
apply_password="isolated-apply-${RANDOM}-${RANDOM}"
|
||||
|
|
@ -64,8 +85,17 @@ source_counts() {
|
|||
}
|
||||
|
||||
service_state() {
|
||||
systemctl show "$service" --no-pager \
|
||||
-p ActiveState -p SubState -p MainPID -p NRestarts -p ExecMainStartTimestamp
|
||||
if ! command -v systemctl >/dev/null 2>&1; then
|
||||
printf 'Available=false\nReason=systemctl_unavailable\n'
|
||||
return 0
|
||||
fi
|
||||
local output
|
||||
if output="$(systemctl show "$service" --no-pager \
|
||||
-p ActiveState -p SubState -p MainPID -p NRestarts -p ExecMainStartTimestamp 2>&1)"; then
|
||||
printf 'Available=true\n%s\n' "$output"
|
||||
else
|
||||
printf 'Available=false\nReason=service_readback_failed\n'
|
||||
fi
|
||||
}
|
||||
|
||||
mkdir -p "$workdir" "$(dirname "$output")"
|
||||
|
|
@ -73,17 +103,30 @@ source_counts_before="$(source_counts)"
|
|||
service_before="$(service_state)"
|
||||
|
||||
docker run -d --name "$container" \
|
||||
--network none \
|
||||
--tmpfs /var/lib/postgresql/data:rw,nosuid,size=512m \
|
||||
-e POSTGRES_PASSWORD="isolated-postgres-${RANDOM}-${RANDOM}" \
|
||||
-e POSTGRES_DB=teleo \
|
||||
"$image" >/dev/null
|
||||
|
||||
for _ in $(seq 1 40); do
|
||||
ready_streak=0
|
||||
for _ in $(seq 1 80); do
|
||||
if docker exec "$container" pg_isready -U postgres -d teleo >/dev/null 2>&1; then
|
||||
break
|
||||
ready_streak=$((ready_streak + 1))
|
||||
if [[ "$ready_streak" -ge 2 ]]; then
|
||||
break
|
||||
fi
|
||||
else
|
||||
ready_streak=0
|
||||
fi
|
||||
sleep 0.25
|
||||
done
|
||||
docker exec "$container" pg_isready -U postgres -d teleo >/dev/null
|
||||
if [[ "$ready_streak" -lt 2 ]]; then
|
||||
echo "disposable Postgres did not become stably ready" >&2
|
||||
exit 1
|
||||
fi
|
||||
container_network_mode="$(docker inspect -f '{{.HostConfig.NetworkMode}}' "$container")"
|
||||
container_mounts_json="$(docker inspect -f '{{json .Mounts}}' "$container")"
|
||||
|
||||
docker exec "$source_container" pg_dump -U postgres -d "$source_db" \
|
||||
--schema-only --no-owner --no-privileges >"$workdir/schema.sql"
|
||||
|
|
@ -173,6 +216,11 @@ CONTAINER_NAME_READBACK="$container_name_readback" \
|
|||
CONTAINER_NAME_READBACK_RC="$container_name_readback_rc" \
|
||||
OUTER_WORKDIR_LEXISTS="$outer_workdir_lexists" \
|
||||
CANARY_RC="$canary_rc" \
|
||||
SOURCE_TIER="$source_tier" \
|
||||
SOURCE_NETWORK_MODE="$source_network_mode" \
|
||||
SOURCE_CANARY_LABEL="$source_canary_label" \
|
||||
CANARY_NETWORK_MODE="$container_network_mode" \
|
||||
CANARY_MOUNTS_JSON="$container_mounts_json" \
|
||||
python3 - "$output" <<'PY'
|
||||
import hashlib
|
||||
import json
|
||||
|
|
@ -254,9 +302,40 @@ after_counts = json.loads(os.environ["SOURCE_COUNTS_AFTER"])
|
|||
live_table_deltas = table_deltas(before_counts, after_counts)
|
||||
before_service = service(os.environ["SERVICE_BEFORE"])
|
||||
after_service = service(os.environ["SERVICE_AFTER"])
|
||||
service_stable = all(
|
||||
before_service.get(key) == after_service.get(key)
|
||||
for key in ("ActiveState", "SubState", "MainPID", "NRestarts")
|
||||
source_tier = os.environ["SOURCE_TIER"]
|
||||
source_network_mode = os.environ["SOURCE_NETWORK_MODE"]
|
||||
source_canary_label = os.environ["SOURCE_CANARY_LABEL"]
|
||||
container_network_mode = os.environ["CANARY_NETWORK_MODE"]
|
||||
container_mounts = json.loads(os.environ["CANARY_MOUNTS_JSON"])
|
||||
container_volume_mounts = [
|
||||
mount for mount in container_mounts if mount.get("Type") == "volume"
|
||||
]
|
||||
container_isolated = (
|
||||
container_network_mode == "none" and not container_volume_mounts
|
||||
)
|
||||
source_tier_enforced = (
|
||||
source_tier != "isolated-local"
|
||||
or (
|
||||
source_network_mode == "none"
|
||||
and source_canary_label == "working-leo-approved-source"
|
||||
)
|
||||
)
|
||||
service_observable = (
|
||||
before_service.get("Available") == "true"
|
||||
and after_service.get("Available") == "true"
|
||||
)
|
||||
service_stable = (
|
||||
all(
|
||||
before_service.get(key) == after_service.get(key)
|
||||
for key in ("ActiveState", "SubState", "MainPID", "NRestarts")
|
||||
)
|
||||
if service_observable
|
||||
else None
|
||||
)
|
||||
service_gate_passes = (
|
||||
service_stable is True
|
||||
if source_tier == "live-readonly"
|
||||
else service_stable is not False
|
||||
)
|
||||
container_absent = (
|
||||
int(os.environ["CONTAINER_NAME_READBACK_RC"]) == 0
|
||||
|
|
@ -277,15 +356,22 @@ inner_runtime_pass = data.get("status") == "pass" and int(os.environ["CANARY_RC"
|
|||
outer_ok = (
|
||||
before_counts == after_counts
|
||||
and all(delta == 0 for delta in live_table_deltas.values())
|
||||
and service_stable
|
||||
and service_gate_passes
|
||||
and container_absent
|
||||
and workdir_absent
|
||||
and source_binding_ok
|
||||
and stale_ephemeral_paths_absent
|
||||
and container_isolated
|
||||
and source_tier_enforced
|
||||
)
|
||||
data["required_tier"] = "T3_live_readonly"
|
||||
data["required_tiers"] = ["T2_runtime", "T3_live_readonly"]
|
||||
data["runtime_scope"] = "isolated_postgres_container_from_readonly_live_schema"
|
||||
if source_tier == "isolated-local":
|
||||
data["required_tier"] = "T2_runtime"
|
||||
data["required_tiers"] = ["T2_runtime"]
|
||||
data["runtime_scope"] = "isolated_postgres_container_from_readonly_local_fixture"
|
||||
else:
|
||||
data["required_tier"] = "T3_live_readonly"
|
||||
data["required_tiers"] = ["T2_runtime", "T3_live_readonly"]
|
||||
data["runtime_scope"] = "isolated_postgres_container_from_readonly_live_schema"
|
||||
data["outer_isolation"] = {
|
||||
"source_container": os.environ.get("SOURCE_CONTAINER", "teleo-pg"),
|
||||
"source_database": os.environ.get("SOURCE_DB", "teleo"),
|
||||
|
|
@ -295,7 +381,22 @@ data["outer_isolation"] = {
|
|||
"source_counts_unchanged": before_counts == after_counts,
|
||||
"service_before": before_service,
|
||||
"service_after": after_service,
|
||||
"service_required": source_tier == "live-readonly",
|
||||
"service_observable": service_observable,
|
||||
"service_stable": service_stable,
|
||||
"service_gate_passes": service_gate_passes,
|
||||
"source_boundary": {
|
||||
"source_tier": source_tier,
|
||||
"network_mode": source_network_mode,
|
||||
"canary_label": source_canary_label,
|
||||
"tier_enforced": source_tier_enforced,
|
||||
},
|
||||
"container_isolation": {
|
||||
"network_mode": container_network_mode,
|
||||
"volume_mounts": container_volume_mounts,
|
||||
"network_disabled": container_network_mode == "none",
|
||||
"no_docker_volumes": not container_volume_mounts,
|
||||
},
|
||||
"cleanup_readback": {
|
||||
"container_remove_returncode": int(os.environ["OUTER_CONTAINER_REMOVE_RC"]),
|
||||
"container_name_query_returncode": int(
|
||||
|
|
@ -312,14 +413,24 @@ data["checks"]["outer_live_counts_exactly_unchanged"] = before_counts == after_c
|
|||
data["checks"]["outer_live_table_deltas_exactly_zero"] = all(
|
||||
delta == 0 for delta in live_table_deltas.values()
|
||||
)
|
||||
data["checks"]["outer_service_stable"] = service_stable
|
||||
data["checks"]["outer_service_observed_if_required"] = (
|
||||
source_tier != "live-readonly" or service_observable
|
||||
)
|
||||
data["checks"]["outer_service_stable_if_observed"] = service_gate_passes
|
||||
data["checks"]["outer_container_absent_independent_readback"] = container_absent
|
||||
data["checks"]["outer_workdir_absent_independent_readback"] = workdir_absent
|
||||
data["checks"]["source_binding_independently_verified"] = source_binding_ok
|
||||
data["checks"]["stale_ephemeral_paths_absent"] = stale_ephemeral_paths_absent
|
||||
data["checks"]["outer_source_tier_enforced"] = source_tier_enforced
|
||||
data["checks"]["outer_container_network_disabled"] = (
|
||||
container_network_mode == "none"
|
||||
)
|
||||
data["checks"]["outer_container_has_no_docker_volumes"] = not container_volume_mounts
|
||||
data["checks"]["outer_isolation_complete"] = outer_ok
|
||||
if outer_ok and inner_runtime_pass:
|
||||
data["current_tier"] = "T3_live_readonly"
|
||||
data["current_tier"] = (
|
||||
"T2_runtime" if source_tier == "isolated-local" else "T3_live_readonly"
|
||||
)
|
||||
else:
|
||||
data["status"] = "fail"
|
||||
data["current_tier"] = "T2_runtime" if inner_runtime_pass else "T1_model"
|
||||
|
|
|
|||
519
scripts/run_leo_local_ingestion_proposal_canary.py
Normal file
519
scripts/run_leo_local_ingestion_proposal_canary.py
Normal file
|
|
@ -0,0 +1,519 @@
|
|||
#!/usr/bin/env python3
|
||||
"""Run a document-ingestion-to-staged-proposal canary in disposable local Postgres."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import hashlib
|
||||
import json
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
import time
|
||||
import uuid
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
HERE = Path(__file__).resolve().parent
|
||||
REPO_ROOT = HERE.parent
|
||||
sys.path.insert(0, str(HERE))
|
||||
|
||||
import apply_proposal as ap # noqa: E402
|
||||
import stage_normalized_proposal as normalized_stage # noqa: E402
|
||||
|
||||
SCHEMA = "livingip.workingLeoLocalIngestionProposalCanary.v1"
|
||||
FIXTURE_SCHEMA = "livingip.workingLeoDocumentIngestionFixture.v1"
|
||||
DEFAULT_FIXTURE = REPO_ROOT / "fixtures" / "working-leo" / "document-ingestion-v1.json"
|
||||
CONTAINER_LABEL = "livingip.canary=working-leo-local-ingestion"
|
||||
|
||||
|
||||
class CanaryError(RuntimeError):
|
||||
pass
|
||||
|
||||
|
||||
def sha256_bytes(value: bytes) -> str:
|
||||
return hashlib.sha256(value).hexdigest()
|
||||
|
||||
|
||||
def canonical_sha256(value: Any) -> str:
|
||||
return sha256_bytes(json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=True).encode())
|
||||
|
||||
|
||||
def stable_uuid(artifact_sha256: str, label: str) -> str:
|
||||
return str(uuid.uuid5(uuid.NAMESPACE_URL, f"livingip:working-leo-ingestion:{artifact_sha256}:{label}"))
|
||||
|
||||
|
||||
def load_fixture(path: Path) -> tuple[dict[str, Any], dict[str, Any]]:
|
||||
raw = path.read_bytes()
|
||||
try:
|
||||
fixture = json.loads(raw)
|
||||
except json.JSONDecodeError as exc:
|
||||
raise CanaryError(f"fixture is not valid JSON: {exc}") from exc
|
||||
if not isinstance(fixture, dict) or fixture.get("schema") != FIXTURE_SCHEMA:
|
||||
raise CanaryError(f"fixture schema must be {FIXTURE_SCHEMA}")
|
||||
source = fixture.get("source")
|
||||
extraction = fixture.get("extraction")
|
||||
claim = extraction.get("claim") if isinstance(extraction, dict) else None
|
||||
evidence = extraction.get("evidence") if isinstance(extraction, dict) else None
|
||||
if not all(isinstance(item, dict) for item in (source, claim, evidence)):
|
||||
raise CanaryError("fixture requires source, extraction.claim, and extraction.evidence objects")
|
||||
required_source = ("identity", "source_key", "source_type", "title", "locator", "body", "metadata")
|
||||
required_claim = ("claim_key", "type", "confidence", "text", "body", "metadata")
|
||||
required_evidence = ("role", "body", "metadata")
|
||||
for label, value, required in (
|
||||
("source", source, required_source),
|
||||
("claim", claim, required_claim),
|
||||
("evidence", evidence, required_evidence),
|
||||
):
|
||||
missing = [key for key in required if value.get(key) in (None, "")]
|
||||
if missing:
|
||||
raise CanaryError(f"fixture {label} is missing: {', '.join(missing)}")
|
||||
if not isinstance(value.get("metadata"), dict):
|
||||
raise CanaryError(f"fixture {label}.metadata must be an object")
|
||||
if claim["body"] not in source["body"]:
|
||||
raise CanaryError("extracted claim body must be an exact substring of source.body")
|
||||
if evidence["body"] not in source["body"]:
|
||||
raise CanaryError("extracted evidence body must be an exact substring of source.body")
|
||||
artifact_sha256 = sha256_bytes(raw)
|
||||
source_content_sha256 = sha256_bytes(source["body"].encode())
|
||||
return fixture, {
|
||||
"path": str(path.resolve()),
|
||||
"artifact_sha256": artifact_sha256,
|
||||
"source_content_sha256": source_content_sha256,
|
||||
"source_identity": source["identity"],
|
||||
}
|
||||
|
||||
|
||||
def build_row_ids(artifact_sha256: str) -> dict[str, str]:
|
||||
return {
|
||||
"source_capture_id": stable_uuid(artifact_sha256, "source-capture"),
|
||||
"claim_extraction_id": stable_uuid(artifact_sha256, "claim-extraction"),
|
||||
"evidence_extraction_id": stable_uuid(artifact_sha256, "evidence-extraction"),
|
||||
"parent_proposal_id": stable_uuid(artifact_sha256, "rich-parent-proposal"),
|
||||
}
|
||||
|
||||
|
||||
def build_parent_packet(
|
||||
fixture: dict[str, Any], input_receipt: dict[str, Any], row_ids: dict[str, str]
|
||||
) -> dict[str, Any]:
|
||||
source = fixture["source"]
|
||||
claim = fixture["extraction"]["claim"]
|
||||
evidence = fixture["extraction"]["evidence"]
|
||||
parent_source_ref = (
|
||||
f"local-ingestion:{input_receipt['artifact_sha256']}:"
|
||||
f"source={row_ids['source_capture_id']}:claim={row_ids['claim_extraction_id']}:"
|
||||
f"evidence={row_ids['evidence_extraction_id']}"
|
||||
)
|
||||
return {
|
||||
"id": row_ids["parent_proposal_id"],
|
||||
"proposal_type": "attach_evidence",
|
||||
"status": "pending_review",
|
||||
"proposed_by_handle": "leo",
|
||||
"proposed_by_agent_id": None,
|
||||
"channel": "local_ingestion_canary",
|
||||
"source_ref": parent_source_ref,
|
||||
"rationale": "Stage one hash-bound claim/evidence extraction from a captured local document fixture.",
|
||||
"payload": {
|
||||
"claim_candidates": [
|
||||
{
|
||||
"claim_key": claim["claim_key"],
|
||||
"type": claim["type"],
|
||||
"confidence": claim["confidence"],
|
||||
"text": claim["text"],
|
||||
"body": claim["body"],
|
||||
"evidence_tier": "isolated_fixture",
|
||||
"needs_research": claim["metadata"].get("needs_research", False),
|
||||
"evidence": [{"source_key": source["source_key"], "role": evidence["role"]}],
|
||||
"edges": [],
|
||||
}
|
||||
],
|
||||
"source_candidates": [
|
||||
{
|
||||
"source_key": source["source_key"],
|
||||
"source_type": source["source_type"],
|
||||
"title": source["title"],
|
||||
"storage_path": input_receipt["path"],
|
||||
"url": None,
|
||||
"source_quality": "local realistic ingestion fixture",
|
||||
"key_excerpt": evidence["body"],
|
||||
"content_sha256": input_receipt["source_content_sha256"],
|
||||
}
|
||||
],
|
||||
"dedupe_conflict_assessment": {
|
||||
"database_search_query": claim["claim_key"].replace("_", " "),
|
||||
"potential_existing_claim_ids": [],
|
||||
"conflicts": [],
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def build_schema_sql() -> str:
|
||||
return r"""\set ON_ERROR_STOP on
|
||||
create schema kb_canary;
|
||||
create schema kb_stage;
|
||||
create table kb_canary.source_captures (
|
||||
id uuid primary key,
|
||||
source_identity text not null unique,
|
||||
source_type text not null,
|
||||
title text not null,
|
||||
locator text not null,
|
||||
artifact_path text not null,
|
||||
artifact_sha256 text not null check (artifact_sha256 ~ '^[0-9a-f]{64}$'),
|
||||
content_sha256 text not null check (content_sha256 ~ '^[0-9a-f]{64}$'),
|
||||
body text not null,
|
||||
metadata jsonb not null
|
||||
);
|
||||
create table kb_canary.claim_extractions (
|
||||
id uuid primary key,
|
||||
source_capture_id uuid not null references kb_canary.source_captures(id),
|
||||
claim_key text not null,
|
||||
body text not null,
|
||||
metadata jsonb not null
|
||||
);
|
||||
create table kb_canary.evidence_extractions (
|
||||
id uuid primary key,
|
||||
claim_extraction_id uuid not null references kb_canary.claim_extractions(id),
|
||||
source_capture_id uuid not null references kb_canary.source_captures(id),
|
||||
role text not null,
|
||||
body text not null,
|
||||
metadata jsonb not null
|
||||
);
|
||||
create table kb_stage.kb_proposals (
|
||||
id uuid primary key,
|
||||
proposal_type text not null,
|
||||
status text not null,
|
||||
proposed_by_handle text,
|
||||
proposed_by_agent_id uuid,
|
||||
channel text not null,
|
||||
source_ref text not null,
|
||||
rationale text not null,
|
||||
payload jsonb not null,
|
||||
reviewed_by_handle text,
|
||||
reviewed_by_agent_id uuid,
|
||||
reviewed_at timestamptz,
|
||||
review_note text,
|
||||
applied_by_handle text,
|
||||
applied_by_agent_id uuid,
|
||||
applied_at timestamptz
|
||||
);
|
||||
"""
|
||||
|
||||
|
||||
def build_capture_sql(fixture: dict[str, Any], input_receipt: dict[str, Any], row_ids: dict[str, str]) -> str:
|
||||
source = fixture["source"]
|
||||
claim = fixture["extraction"]["claim"]
|
||||
evidence = fixture["extraction"]["evidence"]
|
||||
q = ap.sql_literal
|
||||
claim_metadata = {
|
||||
**claim["metadata"],
|
||||
"claim_type": claim["type"],
|
||||
"confidence": claim["confidence"],
|
||||
"text": claim["text"],
|
||||
}
|
||||
return rf"""\set ON_ERROR_STOP on
|
||||
begin;
|
||||
insert into kb_canary.source_captures
|
||||
(id, source_identity, source_type, title, locator, artifact_path,
|
||||
artifact_sha256, content_sha256, body, metadata)
|
||||
values
|
||||
({q(row_ids["source_capture_id"])}::uuid, {q(source["identity"])}, {q(source["source_type"])},
|
||||
{q(source["title"])}, {q(source["locator"])}, {q(input_receipt["path"])},
|
||||
{q(input_receipt["artifact_sha256"])}, {q(input_receipt["source_content_sha256"])},
|
||||
{q(source["body"])}, {q(json.dumps(source["metadata"], sort_keys=True))}::jsonb);
|
||||
insert into kb_canary.claim_extractions
|
||||
(id, source_capture_id, claim_key, body, metadata)
|
||||
values
|
||||
({q(row_ids["claim_extraction_id"])}::uuid, {q(row_ids["source_capture_id"])}::uuid,
|
||||
{q(claim["claim_key"])}, {q(claim["body"])}, {q(json.dumps(claim_metadata, sort_keys=True))}::jsonb);
|
||||
insert into kb_canary.evidence_extractions
|
||||
(id, claim_extraction_id, source_capture_id, role, body, metadata)
|
||||
values
|
||||
({q(row_ids["evidence_extraction_id"])}::uuid, {q(row_ids["claim_extraction_id"])}::uuid,
|
||||
{q(row_ids["source_capture_id"])}::uuid, {q(evidence["role"])}, {q(evidence["body"])},
|
||||
{q(json.dumps(evidence["metadata"], sort_keys=True))}::jsonb);
|
||||
commit;
|
||||
"""
|
||||
|
||||
|
||||
def build_readback_sql(row_ids: dict[str, str], proposal_id: str) -> str:
|
||||
q = ap.sql_literal
|
||||
return rf"""\set ON_ERROR_STOP on
|
||||
begin transaction read only;
|
||||
select jsonb_build_object(
|
||||
'source_capture', (select to_jsonb(s) from kb_canary.source_captures s
|
||||
where id = {q(row_ids["source_capture_id"])}::uuid),
|
||||
'claim_extraction', (select to_jsonb(c) from kb_canary.claim_extractions c
|
||||
where id = {q(row_ids["claim_extraction_id"])}::uuid),
|
||||
'evidence_extraction', (select to_jsonb(e) from kb_canary.evidence_extractions e
|
||||
where id = {q(row_ids["evidence_extraction_id"])}::uuid),
|
||||
'staged_proposal', (select to_jsonb(p) from kb_stage.kb_proposals p
|
||||
where id = {q(proposal_id)}::uuid),
|
||||
'counts', jsonb_build_object(
|
||||
'source_captures', (select count(*) from kb_canary.source_captures),
|
||||
'claim_extractions', (select count(*) from kb_canary.claim_extractions),
|
||||
'evidence_extractions', (select count(*) from kb_canary.evidence_extractions),
|
||||
'staged_proposals', (select count(*) from kb_stage.kb_proposals)
|
||||
)
|
||||
)::text;
|
||||
rollback;
|
||||
"""
|
||||
|
||||
|
||||
class DockerPostgres:
|
||||
def __init__(self, name: str) -> None:
|
||||
self.name = name
|
||||
self.docker = shutil.which("docker")
|
||||
if not self.docker:
|
||||
raise CanaryError("docker executable is unavailable")
|
||||
self.started = False
|
||||
self.volume_mounts: list[dict[str, Any]] = []
|
||||
|
||||
def _run(
|
||||
self, args: list[str], *, input_text: str | None = None, check: bool = True
|
||||
) -> subprocess.CompletedProcess[str]:
|
||||
result = subprocess.run([self.docker, *args], input=input_text, text=True, capture_output=True, check=False)
|
||||
if check and result.returncode != 0:
|
||||
detail = (result.stderr or result.stdout).strip()
|
||||
raise CanaryError(f"docker {' '.join(args[:3])} failed ({result.returncode}): {detail}")
|
||||
return result
|
||||
|
||||
def start(self) -> dict[str, Any]:
|
||||
result = self._run(
|
||||
[
|
||||
"run",
|
||||
"--rm",
|
||||
"-d",
|
||||
"--name",
|
||||
self.name,
|
||||
"--network",
|
||||
"none",
|
||||
"--label",
|
||||
CONTAINER_LABEL,
|
||||
"--tmpfs",
|
||||
"/var/lib/postgresql/data:rw,nosuid,size=256m",
|
||||
"-e",
|
||||
"POSTGRES_HOST_AUTH_METHOD=trust",
|
||||
"-e",
|
||||
"POSTGRES_DB=teleo",
|
||||
"postgres:16-alpine",
|
||||
]
|
||||
)
|
||||
self.started = True
|
||||
container_id = result.stdout.strip()
|
||||
deadline = time.monotonic() + 30
|
||||
ready_streak = 0
|
||||
while time.monotonic() < deadline:
|
||||
ready = self._run(["exec", self.name, "pg_isready", "-U", "postgres", "-d", "teleo"], check=False)
|
||||
if ready.returncode == 0:
|
||||
ready_streak += 1
|
||||
if ready_streak >= 2:
|
||||
break
|
||||
else:
|
||||
ready_streak = 0
|
||||
time.sleep(0.25)
|
||||
else:
|
||||
raise CanaryError("disposable Postgres did not become ready within 30 seconds")
|
||||
inspect = json.loads(self._run(["inspect", self.name]).stdout)[0]
|
||||
if inspect["HostConfig"]["NetworkMode"] != "none":
|
||||
raise CanaryError("disposable Postgres must use network mode none")
|
||||
if inspect["Config"]["Labels"].get("livingip.canary") != "working-leo-local-ingestion":
|
||||
raise CanaryError("disposable Postgres is missing its canary label")
|
||||
self.volume_mounts = [
|
||||
{
|
||||
"type": mount.get("Type"),
|
||||
"name": mount.get("Name"),
|
||||
"destination": mount.get("Destination"),
|
||||
}
|
||||
for mount in inspect.get("Mounts", [])
|
||||
if mount.get("Type") == "volume"
|
||||
]
|
||||
if self.volume_mounts:
|
||||
raise CanaryError("disposable Postgres unexpectedly created a Docker volume")
|
||||
return {
|
||||
"container_id": container_id,
|
||||
"name": self.name,
|
||||
"image": inspect["Config"]["Image"],
|
||||
"network_mode": inspect["HostConfig"]["NetworkMode"],
|
||||
"tmpfs_data_dir": "/var/lib/postgresql/data" in inspect["HostConfig"].get("Tmpfs", {}),
|
||||
"docker_volume_mounts": self.volume_mounts,
|
||||
"canary_label": inspect["Config"]["Labels"].get("livingip.canary"),
|
||||
}
|
||||
|
||||
def psql(self, sql: str) -> str:
|
||||
result = self._run(
|
||||
["exec", "-i", self.name, "psql", "-U", "postgres", "-d", "teleo", "-Atq"],
|
||||
input_text=sql,
|
||||
)
|
||||
return result.stdout.strip()
|
||||
|
||||
def psql_json(self, sql: str) -> dict[str, Any]:
|
||||
output = self.psql(sql)
|
||||
lines = [line for line in output.splitlines() if line.strip()]
|
||||
if not lines:
|
||||
raise CanaryError("Postgres readback returned no JSON")
|
||||
try:
|
||||
value = json.loads(lines[-1])
|
||||
except json.JSONDecodeError as exc:
|
||||
raise CanaryError(f"Postgres readback was not JSON: {lines[-1]}") from exc
|
||||
if not isinstance(value, dict):
|
||||
raise CanaryError("Postgres readback must be a JSON object")
|
||||
return value
|
||||
|
||||
def cleanup(self) -> dict[str, Any]:
|
||||
remove = self._run(["rm", "-f", self.name], check=False) if self.started else None
|
||||
inspect = self._run(["inspect", self.name], check=False)
|
||||
return {
|
||||
"remove_attempted": self.started,
|
||||
"remove_returncode": remove.returncode if remove else None,
|
||||
"container_absent": inspect.returncode != 0,
|
||||
"docker_volume_mounts_observed": self.volume_mounts,
|
||||
"anonymous_or_named_volume_created": bool(self.volume_mounts),
|
||||
}
|
||||
|
||||
|
||||
def proposal_links(readback: dict[str, Any], input_receipt: dict[str, Any]) -> dict[str, Any]:
|
||||
proposal = readback["staged_proposal"]
|
||||
apply_payload = proposal["payload"]["apply_payload"]
|
||||
source_rows = apply_payload["sources"]
|
||||
evidence_rows = apply_payload["evidence"]
|
||||
return {
|
||||
"proposal_id": proposal["id"],
|
||||
"proposal_status": proposal["status"],
|
||||
"proposal_source_ref": proposal["source_ref"],
|
||||
"planned_claim_ids": [row["id"] for row in apply_payload["claims"]],
|
||||
"planned_source_ids": [row["id"] for row in source_rows],
|
||||
"planned_evidence_keys": [
|
||||
{"claim_id": row["claim_id"], "source_id": row["source_id"], "role": row["role"]} for row in evidence_rows
|
||||
],
|
||||
"source_content_hash_matches": any(
|
||||
row.get("hash") == input_receipt["source_content_sha256"] for row in source_rows
|
||||
),
|
||||
"parent_source_ref_embedded": any(
|
||||
readback["source_capture"]["artifact_sha256"] in str(row.get("excerpt"))
|
||||
and readback["source_capture"]["id"] in str(row.get("excerpt"))
|
||||
and readback["claim_extraction"]["id"] in str(row.get("excerpt"))
|
||||
and readback["evidence_extraction"]["id"] in str(row.get("excerpt"))
|
||||
for row in source_rows
|
||||
),
|
||||
}
|
||||
|
||||
|
||||
def evaluate_checks(
|
||||
fixture: dict[str, Any], input_receipt: dict[str, Any], row_ids: dict[str, str], readback: dict[str, Any]
|
||||
) -> dict[str, bool]:
|
||||
source = readback.get("source_capture") or {}
|
||||
claim = readback.get("claim_extraction") or {}
|
||||
evidence = readback.get("evidence_extraction") or {}
|
||||
proposal = readback.get("staged_proposal") or {}
|
||||
links = proposal_links(readback, input_receipt)
|
||||
return {
|
||||
"input_artifact_hash_persisted": source.get("artifact_sha256") == input_receipt["artifact_sha256"],
|
||||
"source_content_hash_persisted": source.get("content_sha256") == input_receipt["source_content_sha256"],
|
||||
"source_identity_persisted": source.get("source_identity") == input_receipt["source_identity"],
|
||||
"claim_links_to_source_capture": claim.get("source_capture_id") == row_ids["source_capture_id"],
|
||||
"claim_body_and_metadata_persisted": (
|
||||
claim.get("body") == fixture["extraction"]["claim"]["body"]
|
||||
and claim.get("metadata", {}).get("text") == fixture["extraction"]["claim"]["text"]
|
||||
),
|
||||
"evidence_links_to_claim_and_source": (
|
||||
evidence.get("claim_extraction_id") == row_ids["claim_extraction_id"]
|
||||
and evidence.get("source_capture_id") == row_ids["source_capture_id"]
|
||||
),
|
||||
"evidence_body_and_metadata_persisted": (
|
||||
evidence.get("body") == fixture["extraction"]["evidence"]["body"]
|
||||
and evidence.get("metadata") == fixture["extraction"]["evidence"]["metadata"]
|
||||
),
|
||||
"one_row_per_ingestion_stage": readback.get("counts")
|
||||
== {"source_captures": 1, "claim_extractions": 1, "evidence_extractions": 1, "staged_proposals": 1},
|
||||
"proposal_is_pending_review": proposal.get("status") == "pending_review",
|
||||
"proposal_has_source_ref": bool(proposal.get("source_ref")),
|
||||
"proposal_payload_contains_input_hash": links["source_content_hash_matches"],
|
||||
"proposal_payload_contains_capture_row_linkage": links["parent_source_ref_embedded"],
|
||||
"no_review_or_apply_metadata": all(
|
||||
proposal.get(key) is None
|
||||
for key in ("reviewed_by_handle", "reviewed_at", "applied_by_handle", "applied_at")
|
||||
),
|
||||
}
|
||||
|
||||
|
||||
def run_canary(fixture_path: Path, output: Path | None = None) -> dict[str, Any]:
|
||||
fixture, input_receipt = load_fixture(fixture_path)
|
||||
row_ids = build_row_ids(input_receipt["artifact_sha256"])
|
||||
parent = build_parent_packet(fixture, input_receipt, row_ids)
|
||||
child = normalized_stage.prepare_normalized_child(parent)
|
||||
container_name = f"working-leo-ingest-{uuid.uuid4().hex[:12]}"
|
||||
postgres = DockerPostgres(container_name)
|
||||
receipt: dict[str, Any] = {
|
||||
"schema": SCHEMA,
|
||||
"status": "fail",
|
||||
"required_tier": "isolated_realistic_ingestion_to_proposal",
|
||||
"input": input_receipt,
|
||||
"row_ids": row_ids,
|
||||
"production_mutation_attempted": False,
|
||||
"canonical_apply_attempted": False,
|
||||
}
|
||||
try:
|
||||
receipt["environment"] = postgres.start()
|
||||
postgres.psql(build_schema_sql())
|
||||
postgres.psql(build_capture_sql(fixture, input_receipt, row_ids))
|
||||
stage_result = postgres.psql_json(normalized_stage.build_stage_sql(child))
|
||||
readback = postgres.psql_json(build_readback_sql(row_ids, child["id"]))
|
||||
checks = evaluate_checks(fixture, input_receipt, row_ids, readback)
|
||||
receipt.update(
|
||||
{
|
||||
"stage_command_result": stage_result,
|
||||
"rows": readback,
|
||||
"row_link_fields_proven": proposal_links(readback, input_receipt),
|
||||
"checks": checks,
|
||||
"status": "pass" if all(checks.values()) else "fail",
|
||||
}
|
||||
)
|
||||
except (CanaryError, OSError, ValueError, KeyError) as exc:
|
||||
receipt["error"] = {"type": type(exc).__name__, "message": str(exc)}
|
||||
finally:
|
||||
receipt["cleanup"] = postgres.cleanup()
|
||||
receipt["cleanup"]["network_mode_none"] = (
|
||||
receipt.get("environment", {}).get("network_mode") == "none"
|
||||
)
|
||||
receipt["cleanup"]["no_production_target_referenced"] = receipt[
|
||||
"cleanup"
|
||||
]["network_mode_none"]
|
||||
if not all(
|
||||
(
|
||||
receipt["cleanup"]["container_absent"],
|
||||
receipt["cleanup"]["network_mode_none"],
|
||||
)
|
||||
):
|
||||
receipt["status"] = "fail"
|
||||
receipt["strongest_claim"] = (
|
||||
"A realistic local document fixture was hash-captured, replayed through deterministic claim/evidence "
|
||||
"extraction, normalized by the existing rich-proposal normalizer, and staged/read back as one "
|
||||
"pending_review proposal in disposable networkless Postgres."
|
||||
)
|
||||
receipt["residual_gap"] = (
|
||||
"This does not prove live generative extraction, arbitrary document coverage, canonical apply, or any "
|
||||
"hosted/production runtime."
|
||||
)
|
||||
if output:
|
||||
output.parent.mkdir(parents=True, exist_ok=True)
|
||||
output.write_text(json.dumps(receipt, indent=2, sort_keys=True) + "\n", encoding="utf-8")
|
||||
return receipt
|
||||
|
||||
|
||||
def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
parser.add_argument("--fixture", type=Path, default=DEFAULT_FIXTURE)
|
||||
parser.add_argument("--output", type=Path)
|
||||
return parser.parse_args(argv)
|
||||
|
||||
|
||||
def main(argv: list[str] | None = None) -> int:
|
||||
args = parse_args(argv)
|
||||
receipt = run_canary(args.fixture.resolve(), args.output.resolve() if args.output else None)
|
||||
print(json.dumps(receipt, sort_keys=True, separators=(",", ":")))
|
||||
return 0 if receipt["status"] == "pass" else 1
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
79
tests/test_run_approve_claim_clone_canary.py
Normal file
79
tests/test_run_approve_claim_clone_canary.py
Normal file
|
|
@ -0,0 +1,79 @@
|
|||
"""Focused tests for the strict approved-change clone receipt."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
sys.path.insert(0, str(ROOT / "scripts"))
|
||||
|
||||
import run_approve_claim_clone_canary as canary # noqa: E402
|
||||
|
||||
|
||||
def _passing_result(expected_rows: dict, expected_deltas: dict) -> dict:
|
||||
proposal_id = "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa"
|
||||
return {
|
||||
"status": "pass",
|
||||
"fixture_ids": {"proposal_id": proposal_id},
|
||||
"apply_transition": {
|
||||
"applied_readback": {
|
||||
"id": proposal_id,
|
||||
"status": "applied",
|
||||
"applied_at": "2026-07-12 12:00:00+00",
|
||||
"applied_by_handle": "kb-apply",
|
||||
}
|
||||
},
|
||||
"row_readback": expected_rows,
|
||||
"clone_counts_before_apply": {key: 0 for key in expected_deltas},
|
||||
"clone_counts_after_apply": expected_deltas,
|
||||
"clone_apply_table_deltas": expected_deltas,
|
||||
"clone_dropped": True,
|
||||
"cleanup_readback": {"leftover_clone_database_count": 0},
|
||||
"source_database_mutated": False,
|
||||
"checks": {"conflict_transaction_rolled_back_exactly": True},
|
||||
}
|
||||
|
||||
|
||||
def test_service_state_records_unavailable_systemctl(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.setattr(
|
||||
canary,
|
||||
"_run",
|
||||
lambda *_args, **_kwargs: (_ for _ in ()).throw(FileNotFoundError()),
|
||||
)
|
||||
|
||||
assert canary._service_state("unused.service") == {
|
||||
"available": False,
|
||||
"returncode": 127,
|
||||
"reason": "systemctl_unavailable",
|
||||
}
|
||||
|
||||
|
||||
def test_strict_receipt_requires_exact_ids_counts_applied_at_and_cleanup() -> None:
|
||||
expected_rows = canary._expected_bundle_rows(canary.build_fixture("unit")["apply_payload"])
|
||||
expected_deltas = canary._expected_table_deltas(expected_rows)
|
||||
result = _passing_result(expected_rows, expected_deltas)
|
||||
|
||||
receipt = canary._strict_receipt(result, expected_rows, expected_deltas)
|
||||
|
||||
assert receipt["pass"] is True
|
||||
assert receipt["proposal"]["status"] == "applied"
|
||||
assert receipt["proposal"]["applied_at"]
|
||||
assert receipt["canonical_row_keys"]["actual"] == receipt["canonical_row_keys"]["expected"]
|
||||
assert receipt["row_counts"]["actual_deltas"] == expected_deltas
|
||||
assert receipt["rollback_cleanup"]["clone_database_removed"] is True
|
||||
|
||||
result["apply_transition"]["applied_readback"]["applied_at"] = None
|
||||
result["row_readback"] = {**expected_rows, "claims": []}
|
||||
result["clone_apply_table_deltas"] = {**expected_deltas, "public.claims": 0}
|
||||
result["clone_dropped"] = False
|
||||
|
||||
failed = canary._strict_receipt(result, expected_rows, expected_deltas)
|
||||
|
||||
assert failed["pass"] is False
|
||||
assert failed["checks"]["proposal_applied_at_present"] is False
|
||||
assert failed["checks"]["canonical_row_keys_exact"] is False
|
||||
assert failed["checks"]["table_deltas_exact"] is False
|
||||
assert failed["checks"]["clone_database_removed"] is False
|
||||
102
tests/test_run_approve_claim_clone_canary_macos_path.py
Normal file
102
tests/test_run_approve_claim_clone_canary_macos_path.py
Normal file
|
|
@ -0,0 +1,102 @@
|
|||
"""Platform-path regression tests for the approved-change clone canary."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import subprocess
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
sys.path.insert(0, str(ROOT / "scripts"))
|
||||
WRAPPER = ROOT / "scripts" / "run_approve_claim_isolated_container_canary.sh"
|
||||
|
||||
import apply_proposal as apply # noqa: E402
|
||||
import run_approve_claim_clone_canary as canary # noqa: E402
|
||||
|
||||
|
||||
def test_role_psql_resolves_docker_before_scrubbing_path(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
captured: dict[str, object] = {}
|
||||
|
||||
monkeypatch.setattr(
|
||||
canary.shutil,
|
||||
"which",
|
||||
lambda executable: "/opt/homebrew/bin/docker"
|
||||
if executable == "docker"
|
||||
else None,
|
||||
)
|
||||
|
||||
def fake_run(command: list[str], **kwargs: object) -> subprocess.CompletedProcess[str]:
|
||||
captured["command"] = command
|
||||
captured["env"] = kwargs["env"]
|
||||
return subprocess.CompletedProcess(command, 0, "", "")
|
||||
|
||||
monkeypatch.setattr(canary, "_run", fake_run)
|
||||
|
||||
canary._role_psql(
|
||||
"local-postgres",
|
||||
"teleo",
|
||||
"select 1;",
|
||||
role="kb_apply",
|
||||
password="test-only-password",
|
||||
)
|
||||
|
||||
assert captured["command"][0] == "/opt/homebrew/bin/docker"
|
||||
assert captured["env"] == {
|
||||
"PGPASSWORD": "test-only-password",
|
||||
"PATH": "/usr/bin:/bin:/usr/local/bin",
|
||||
}
|
||||
|
||||
|
||||
def test_apply_tool_resolves_docker_before_scrubbing_path(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
captured: dict[str, object] = {}
|
||||
monkeypatch.setattr(
|
||||
apply.shutil,
|
||||
"which",
|
||||
lambda executable: "/opt/homebrew/bin/docker"
|
||||
if executable == "docker"
|
||||
else None,
|
||||
)
|
||||
|
||||
def fake_run(command: list[str], **kwargs: object) -> subprocess.CompletedProcess[str]:
|
||||
captured["command"] = command
|
||||
captured["env"] = kwargs["env"]
|
||||
return subprocess.CompletedProcess(command, 0, "ok", "")
|
||||
|
||||
monkeypatch.setattr(apply.subprocess, "run", fake_run)
|
||||
args = type(
|
||||
"Args",
|
||||
(),
|
||||
{
|
||||
"container": "local-postgres",
|
||||
"role": "kb_apply",
|
||||
"host": "127.0.0.1",
|
||||
"db": "teleo",
|
||||
},
|
||||
)()
|
||||
|
||||
assert apply.run_psql(args, "select 1;", "test-only-password") == "ok"
|
||||
assert captured["command"][0] == "/opt/homebrew/bin/docker"
|
||||
assert captured["env"] == {
|
||||
"PGPASSWORD": "test-only-password",
|
||||
"PATH": "/usr/bin:/bin:/usr/local/bin",
|
||||
}
|
||||
|
||||
|
||||
def test_outer_wrapper_enforces_local_source_and_networkless_tmpfs() -> None:
|
||||
text = WRAPPER.read_text(encoding="utf-8")
|
||||
|
||||
assert 'source_canary_label" != "working-leo-approved-source"' in text
|
||||
assert "--network none" in text
|
||||
assert "--tmpfs /var/lib/postgresql/data:rw,nosuid,size=512m" in text
|
||||
assert 'CANARY_NETWORK_MODE="$container_network_mode"' in text
|
||||
assert 'CANARY_MOUNTS_JSON="$container_mounts_json"' in text
|
||||
assert 'data["checks"]["outer_source_tier_enforced"]' in text
|
||||
assert 'data["checks"]["outer_container_network_disabled"]' in text
|
||||
assert 'data["checks"]["outer_container_has_no_docker_volumes"]' in text
|
||||
assert 'source_tier != "live-readonly" or service_observable' in text
|
||||
164
tests/test_run_leo_local_ingestion_proposal_canary.py
Normal file
164
tests/test_run_leo_local_ingestion_proposal_canary.py
Normal file
|
|
@ -0,0 +1,164 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import copy
|
||||
import json
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parents[1]
|
||||
sys.path.insert(0, str(REPO_ROOT / "scripts"))
|
||||
|
||||
import run_leo_local_ingestion_proposal_canary as canary # noqa: E402
|
||||
|
||||
|
||||
def _loaded() -> tuple[dict, dict, dict]:
|
||||
fixture, input_receipt = canary.load_fixture(canary.DEFAULT_FIXTURE)
|
||||
row_ids = canary.build_row_ids(input_receipt["artifact_sha256"])
|
||||
return fixture, input_receipt, row_ids
|
||||
|
||||
|
||||
def test_fixture_is_realistic_hash_bound_and_exactly_evidenced() -> None:
|
||||
fixture, input_receipt, row_ids = _loaded()
|
||||
|
||||
assert fixture["extraction"]["evidence"]["body"] in fixture["source"]["body"]
|
||||
assert len(input_receipt["artifact_sha256"]) == 64
|
||||
assert len(input_receipt["source_content_sha256"]) == 64
|
||||
assert input_receipt["artifact_sha256"] != input_receipt["source_content_sha256"]
|
||||
assert len(set(row_ids.values())) == 4
|
||||
|
||||
|
||||
def test_fixture_validation_fails_closed_when_evidence_is_not_in_source(tmp_path: Path) -> None:
|
||||
fixture = json.loads(canary.DEFAULT_FIXTURE.read_text(encoding="utf-8"))
|
||||
fixture["extraction"]["evidence"]["body"] = "invented evidence"
|
||||
path = tmp_path / "bad-fixture.json"
|
||||
path.write_text(json.dumps(fixture), encoding="utf-8")
|
||||
|
||||
with pytest.raises(canary.CanaryError, match="exact substring"):
|
||||
canary.load_fixture(path)
|
||||
|
||||
|
||||
def test_fixture_validation_fails_closed_when_claim_body_is_not_in_source(
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
fixture = json.loads(canary.DEFAULT_FIXTURE.read_text(encoding="utf-8"))
|
||||
fixture["extraction"]["claim"]["body"] = "invented claim body"
|
||||
path = tmp_path / "bad-claim-fixture.json"
|
||||
path.write_text(json.dumps(fixture), encoding="utf-8")
|
||||
|
||||
with pytest.raises(canary.CanaryError, match="claim body must be an exact substring"):
|
||||
canary.load_fixture(path)
|
||||
|
||||
|
||||
def test_capture_sql_escapes_quotes_and_backslashes_from_fixture(
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
fixture = json.loads(canary.DEFAULT_FIXTURE.read_text(encoding="utf-8"))
|
||||
fixture["source"]["title"] = "O'Brien\\review"
|
||||
fixture["source"]["metadata"]["note"] = "quoted ' value \\ path"
|
||||
path = tmp_path / "quoted-fixture.json"
|
||||
path.write_text(json.dumps(fixture), encoding="utf-8")
|
||||
loaded, input_receipt = canary.load_fixture(path)
|
||||
row_ids = canary.build_row_ids(input_receipt["artifact_sha256"])
|
||||
|
||||
sql = canary.build_capture_sql(loaded, input_receipt, row_ids)
|
||||
|
||||
assert canary.ap.sql_literal(fixture["source"]["title"]) in sql
|
||||
assert canary.ap.sql_literal(
|
||||
json.dumps(fixture["source"]["metadata"], sort_keys=True)
|
||||
) in sql
|
||||
|
||||
|
||||
def test_parent_normalizes_to_hash_bound_pending_proposal_with_embedded_row_links() -> None:
|
||||
fixture, input_receipt, row_ids = _loaded()
|
||||
parent = canary.build_parent_packet(fixture, input_receipt, row_ids)
|
||||
child = canary.normalized_stage.prepare_normalized_child(parent)
|
||||
payload = child["payload"]["apply_payload"]
|
||||
|
||||
assert child["status"] == "pending_review"
|
||||
assert child["proposal_type"] == "approve_claim"
|
||||
assert len(payload["claims"]) == 1
|
||||
assert len(payload["sources"]) == 2
|
||||
assert len(payload["evidence"]) == 2
|
||||
assert input_receipt["source_content_sha256"] in {row["hash"] for row in payload["sources"]}
|
||||
excerpts = "\n".join(row["excerpt"] for row in payload["sources"])
|
||||
assert input_receipt["artifact_sha256"] in excerpts
|
||||
assert row_ids["source_capture_id"] in excerpts
|
||||
assert row_ids["claim_extraction_id"] in excerpts
|
||||
assert row_ids["evidence_extraction_id"] in excerpts
|
||||
|
||||
|
||||
def test_capture_and_stage_sql_never_mutate_canonical_public_tables() -> None:
|
||||
fixture, input_receipt, row_ids = _loaded()
|
||||
child = canary.normalized_stage.prepare_normalized_child(
|
||||
canary.build_parent_packet(fixture, input_receipt, row_ids)
|
||||
)
|
||||
sql = "\n".join(
|
||||
(
|
||||
canary.build_schema_sql(),
|
||||
canary.build_capture_sql(fixture, input_receipt, row_ids),
|
||||
canary.normalized_stage.build_stage_sql(child),
|
||||
)
|
||||
).lower()
|
||||
|
||||
assert "insert into public." not in sql
|
||||
assert "update public." not in sql
|
||||
assert "delete from public." not in sql
|
||||
assert "insert into kb_canary.source_captures" in sql
|
||||
assert "insert into kb_canary.claim_extractions" in sql
|
||||
assert "insert into kb_canary.evidence_extractions" in sql
|
||||
assert "insert into kb_stage.kb_proposals" in sql
|
||||
|
||||
|
||||
def test_check_evaluation_requires_every_row_link_and_staging_boundary() -> None:
|
||||
fixture, input_receipt, row_ids = _loaded()
|
||||
child = canary.normalized_stage.prepare_normalized_child(
|
||||
canary.build_parent_packet(fixture, input_receipt, row_ids)
|
||||
)
|
||||
apply_payload = child["payload"]["apply_payload"]
|
||||
readback = {
|
||||
"source_capture": {
|
||||
"id": row_ids["source_capture_id"],
|
||||
"artifact_sha256": input_receipt["artifact_sha256"],
|
||||
"content_sha256": input_receipt["source_content_sha256"],
|
||||
"source_identity": input_receipt["source_identity"],
|
||||
},
|
||||
"claim_extraction": {
|
||||
"id": row_ids["claim_extraction_id"],
|
||||
"source_capture_id": row_ids["source_capture_id"],
|
||||
"body": fixture["extraction"]["claim"]["body"],
|
||||
"metadata": {"text": fixture["extraction"]["claim"]["text"]},
|
||||
},
|
||||
"evidence_extraction": {
|
||||
"id": row_ids["evidence_extraction_id"],
|
||||
"claim_extraction_id": row_ids["claim_extraction_id"],
|
||||
"source_capture_id": row_ids["source_capture_id"],
|
||||
"body": fixture["extraction"]["evidence"]["body"],
|
||||
"metadata": fixture["extraction"]["evidence"]["metadata"],
|
||||
},
|
||||
"staged_proposal": {
|
||||
"id": child["id"],
|
||||
"status": "pending_review",
|
||||
"source_ref": child["source_ref"],
|
||||
"payload": {"apply_payload": apply_payload},
|
||||
"reviewed_by_handle": None,
|
||||
"reviewed_at": None,
|
||||
"applied_by_handle": None,
|
||||
"applied_at": None,
|
||||
},
|
||||
"counts": {
|
||||
"source_captures": 1,
|
||||
"claim_extractions": 1,
|
||||
"evidence_extractions": 1,
|
||||
"staged_proposals": 1,
|
||||
},
|
||||
}
|
||||
|
||||
checks = canary.evaluate_checks(fixture, input_receipt, row_ids, readback)
|
||||
assert all(checks.values())
|
||||
broken = copy.deepcopy(readback)
|
||||
broken["evidence_extraction"]["claim_extraction_id"] = canary.stable_uuid("0" * 64, "wrong")
|
||||
assert (
|
||||
canary.evaluate_checks(fixture, input_receipt, row_ids, broken)["evidence_links_to_claim_and_source"] is False
|
||||
)
|
||||
Loading…
Reference in a new issue