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

Prevent GCP proof harness heredoc deadlocks

Browse source
This commit is contained in:
twentyOne2x 2026-07-13 07:13:18 +02:00
parent be23b5e95b
commit 200749ca5b
5 changed files with 348 additions and 243 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

100
ops/build_gcp_cloudsql_restore_proof.py Executable file
View file

@ -0,0 +1,100 @@
#!/usr/bin/env python3
"""Build the prepared proof manifest for a Cloud SQL restore drill."""
from __future__ import annotations
import argparse
import hashlib
import json
from datetime import datetime, timezone
from pathlib import Path
def sha256(path: Path) -> str:
digest = hashlib.sha256()
with path.open("rb") as handle:
for chunk in iter(lambda: handle.read(1024 * 1024), b""):
digest.update(chunk)
return digest.hexdigest()
def build_proof(args: argparse.Namespace) -> dict[str, object]:
summary = json.loads(args.source_summary.read_text(encoding="utf-8"))
return {
"artifact": "gcp_cloudsql_restore_drill",
"generated_at_utc": datetime.now(timezone.utc).isoformat(),
"mode": "execute" if args.execute else "dry_run",
"status": "prepared_for_execute" if args.execute else "prepared",
"project_id": args.project_id,
"region": args.region,
"cloudsql_instance": args.instance,
"database": args.database,
"postgres_schema": summary["schema"],
"source_sqlite_backup": str(args.sqlite_backup),
"source_sqlite_backup_sha256": sha256(args.sqlite_backup),
"source_integrity_check": summary["source_integrity_check"],
"source_table_count": summary["table_count"],
"source_total_rows": summary["source_total_rows"],
"source_table_counts": {table["name"]: table["row_count"] for table in summary["tables"]},
"gcs_import_uri": args.gcs_import_uri,
"local_private_artifacts": {
"postgres_import_sql": str(args.import_sql),
"source_summary_json": str(args.source_summary),
"target_counts_sql": str(args.target_counts_sql),
},
"planned_commands": [
f"gcloud storage cp {args.import_sql} {args.gcs_import_uri}",
f"gcloud sql import sql {args.instance} {args.gcs_import_uri} --project={args.project_id} "
f"--database={args.database} --quiet --format=json",
"run target-counts.sql from a trusted VPC/Cloud SQL connector path and compare "
"source_total_rows/source_table_count",
],
"not_proven_by_this_artifact": [
"GCS object generation until EXECUTE=1 succeeds",
"Cloud SQL import operation until EXECUTE=1 succeeds",
"Cloud SQL query readback until target-counts.sql is run from a trusted VPC/connector path",
],
}
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--source-summary", type=Path, required=True)
parser.add_argument("--sqlite-backup", type=Path, required=True)
parser.add_argument("--import-sql", type=Path, required=True)
parser.add_argument("--target-counts-sql", type=Path, required=True)
parser.add_argument("--proof-json", type=Path, required=True)
parser.add_argument("--project-id", required=True)
parser.add_argument("--instance", required=True)
parser.add_argument("--database", required=True)
parser.add_argument("--region", required=True)
parser.add_argument("--gcs-import-uri", required=True)
parser.add_argument("--execute", action="store_true")
return parser.parse_args()
def main() -> int:
args = parse_args()
proof = build_proof(args)
args.proof_json.write_text(json.dumps(proof, indent=2, sort_keys=True) + "\n", encoding="utf-8")
print(
json.dumps(
{
"artifact": proof["artifact"],
"mode": proof["mode"],
"status": proof["status"],
"source_integrity_check": proof["source_integrity_check"],
"source_table_count": proof["source_table_count"],
"source_total_rows": proof["source_total_rows"],
"gcs_import_uri": proof["gcs_import_uri"],
"proof": str(args.proof_json),
},
indent=2,
sort_keys=True,
)
)
return 0
if __name__ == "__main__":
raise SystemExit(main())

84
ops/run_gcp_cloudsql_restore_drill.sh
View file

@ -54,73 +54,25 @@ python3 ops/sqlite_to_postgres_dump.py \
--target-counts-sql "$TARGET_COUNTS_SQL" \
> "$CONVERT_LOG"
python3 - "$SOURCE_SUMMARY" "$SQLITE_BACKUP" "$IMPORT_SQL" "$TARGET_COUNTS_SQL" "$PROOF_JSON" "$PROJECT_ID" "$INSTANCE" "$DATABASE" "$REGION" "$GCS_IMPORT_URI" "$EXECUTE" <<'PY'
import hashlib
import json
import sys
from datetime import UTC, datetime
from pathlib import Path
execute_arg=()
if [[ "$EXECUTE" == "1" ]]; then
execute_arg=(--execute)
fi
summary_path = Path(sys.argv[1])
sqlite_backup = Path(sys.argv[2])
import_sql = Path(sys.argv[3])
target_counts_sql = Path(sys.argv[4])
proof_path = Path(sys.argv[5])
project_id, instance, database, region, gcs_import_uri, execute = sys.argv[6:12]
summary = json.loads(summary_path.read_text())
def sha256(path: Path) -> str:
digest = hashlib.sha256()
with path.open("rb") as handle:
for chunk in iter(lambda: handle.read(1024 * 1024), b""):
digest.update(chunk)
return digest.hexdigest()
proof = {
"artifact": "gcp_cloudsql_restore_drill",
"generated_at_utc": datetime.now(UTC).isoformat(),
"mode": "execute" if execute == "1" else "dry_run",
"status": "prepared" if execute != "1" else "prepared_for_execute",
"project_id": project_id,
"region": region,
"cloudsql_instance": instance,
"database": database,
"postgres_schema": summary["schema"],
"source_sqlite_backup": str(sqlite_backup),
"source_sqlite_backup_sha256": sha256(sqlite_backup),
"source_integrity_check": summary["source_integrity_check"],
"source_table_count": summary["table_count"],
"source_total_rows": summary["source_total_rows"],
"source_table_counts": {table["name"]: table["row_count"] for table in summary["tables"]},
"gcs_import_uri": gcs_import_uri,
"local_private_artifacts": {
"postgres_import_sql": str(import_sql),
"source_summary_json": str(summary_path),
"target_counts_sql": str(target_counts_sql),
},
"planned_commands": [
f"gcloud storage cp {import_sql} {gcs_import_uri}",
f"gcloud sql import sql {instance} {gcs_import_uri} --project={project_id} --database={database} --quiet --format=json",
"run target-counts.sql from a trusted VPC/Cloud SQL connector path and compare source_total_rows/source_table_count",
],
"not_proven_by_this_artifact": [
"GCS object generation until EXECUTE=1 succeeds",
"Cloud SQL import operation until EXECUTE=1 succeeds",
"Cloud SQL query readback until target-counts.sql is run from a trusted VPC/connector path",
],
}
proof_path.write_text(json.dumps(proof, indent=2, sort_keys=True) + "\n")
print(json.dumps({
"artifact": proof["artifact"],
"mode": proof["mode"],
"status": proof["status"],
"source_integrity_check": proof["source_integrity_check"],
"source_table_count": proof["source_table_count"],
"source_total_rows": proof["source_total_rows"],
"gcs_import_uri": proof["gcs_import_uri"],
"proof": str(proof_path),
}, indent=2, sort_keys=True))
PY
# The helper records source_table_counts in the proof manifest without a large
# stdin heredoc, which can deadlock on constrained macOS Bash pipe buffers.
python3 ops/build_gcp_cloudsql_restore_proof.py \
--source-summary "$SOURCE_SUMMARY" \
--sqlite-backup "$SQLITE_BACKUP" \
--import-sql "$IMPORT_SQL" \
--target-counts-sql "$TARGET_COUNTS_SQL" \
--proof-json "$PROOF_JSON" \
--project-id "$PROJECT_ID" \
--instance "$INSTANCE" \
--database "$DATABASE" \
--region "$REGION" \
--gcs-import-uri "$GCS_IMPORT_URI" \
"${execute_arg[@]}"
if [[ "$EXECUTE" != "1" ]]; then
exit 0

164
scripts/gcp_iap_operator.sh
View file

@ -21,6 +21,8 @@ readonly CORY_REPLAY_LEGACY_DIR="/home/teleo/gcp-cory-model-replay-20260712t1940
readonly DISABLED_ROLLBACK_DB="teleo_canonical_pre_20260712t1905z"
readonly REQUEST_ID_RE='^iap-[a-z0-9]{12,32}$'
readonly TARGET_DB_RE='^teleo_clone_[a-z0-9][a-z0-9_]{0,50}$'
readonly SCRIPT_DIR="$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd)"
readonly RESULT_HELPER="$SCRIPT_DIR/gcp_iap_operator_result.py"
RUNNER_TEMP_DIR=""
VALIDATED_BUNDLE_DIR=""
@ -600,69 +602,13 @@ write_sanitized_result() {
local request_id="$4"
local target_db="$5"
local exit_code="$6"
RESULT_PATH="$output_path" RAW_STDOUT="$raw_stdout" OPERATION="$operation" \
REQUEST_ID="$request_id" TARGET_DB="$target_db" EXIT_CODE="$exit_code" python3 - <<'PY'
import hashlib
import json
import os
from pathlib import Path
raw = Path(os.environ["RAW_STDOUT"]).read_bytes()
exit_code = int(os.environ["EXIT_CODE"])
remote = {}
if exit_code == 0:
try:
candidate = json.loads(raw.decode("utf-8").splitlines()[-1])
allowed = {
"operation",
"request_id",
"target_db",
"status",
"instance",
"expected_internal_ip",
"active_state",
"sub_state",
"nrestarts",
"dispatcher_version",
"dispatcher_sha256",
"bundle_commit",
"bundle_sha256",
"result_sha256",
"result_bytes",
"full_result_export_ready",
"fixed_baseline_validation",
"no_message_send",
"live_service_unchanged",
"live_profile_unchanged",
"clone_database_remaining",
"run_directory_remaining",
"legacy_directory_remaining",
"export_file_remaining",
"leftover_temp_directory_count",
"rollback_database_datallowconn",
"rollback_database_connections",
}
remote = {key: value for key, value in candidate.items() if key in allowed}
except (UnicodeDecodeError, json.JSONDecodeError, IndexError):
exit_code = 90
payload = {
"schema": "livingip.gcpIapOperatorSanitizedResult.v1",
"operation": os.environ["OPERATION"],
"request_id": os.environ["REQUEST_ID"],
"target_db": os.environ["TARGET_DB"],
"status": "pass" if exit_code == 0 and remote.get("status") == "pass" else "fail",
"ssh_exit_code": exit_code,
"remote_result": remote,
"remote_stdout_sha256": hashlib.sha256(raw).hexdigest(),
"remote_stdout_bytes": len(raw),
"raw_stdout_retained": False,
"raw_stderr_retained": False,
"credential_values_logged": False,
"ssh_debug_enabled": False,
}
Path(os.environ["RESULT_PATH"]).write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8")
PY
python3 "$RESULT_HELPER" sanitize \
--result-path "$output_path" \
--raw-stdout "$raw_stdout" \
--operation "$operation" \
--request-id "$request_id" \
--target-db "$target_db" \
--exit-code "$exit_code"
chmod 0600 "$output_path"
}
@ -685,83 +631,22 @@ validate_downloaded_replay() {
local result_path="$1"
local full_result="$2"
local target_db="$3"
RESULT_PATH="$result_path" FULL_RESULT="$full_result" TARGET_DB="$target_db" \
EXPECTED_CORY_REPLAY_DB="$CORY_REPLAY_DB" \
EXPECTED_CORY_REPLAY_FINGERPRINT="$CORY_REPLAY_FINGERPRINT" \
python3 - <<'PY'
import hashlib
import json
import os
from pathlib import Path
result_path = Path(os.environ["RESULT_PATH"])
full_path = Path(os.environ["FULL_RESULT"])
result = json.loads(result_path.read_text(encoding="utf-8"))
full_bytes = full_path.read_bytes()
full = json.loads(full_bytes)
remote = result.get("remote_result") or {}
checks = full.get("checks") or {}
required = {
"result_hash": hashlib.sha256(full_bytes).hexdigest() == remote.get("result_sha256"),
"result_size": len(full_bytes) == remote.get("result_bytes"),
"status": full.get("status") == "pass",
"target": full.get("target_database") == os.environ["TARGET_DB"],
"strict_six_of_six": (full.get("score") or {}).get("passes") == 6
and (full.get("score") or {}).get("expected_prompt_count") == 6,
"all_checks": bool(checks) and all(checks.values()),
}
if os.environ["TARGET_DB"] == os.environ["EXPECTED_CORY_REPLAY_DB"]:
expected_counts = {
"claims": 1837,
"sources": 4145,
"claim_edges": 4916,
"claim_evidence": 4670,
"kb_proposals": 26,
}
before_counts = ((full.get("canonical_status_before") or {}).get("high_signal_rows") or {})
after_counts = ((full.get("canonical_status_after") or {}).get("high_signal_rows") or {})
required.update(
{
"fixed_fingerprint_before": (full.get("database_fingerprint_before") or {}).get("sha256")
== os.environ["EXPECTED_CORY_REPLAY_FINGERPRINT"],
"fixed_fingerprint_after": (full.get("database_fingerprint_after") or {}).get("sha256")
== os.environ["EXPECTED_CORY_REPLAY_FINGERPRINT"],
"fixed_counts_before": all(before_counts.get(key) == value for key, value in expected_counts.items()),
"fixed_counts_after": all(after_counts.get(key) == value for key, value in expected_counts.items()),
}
)
failed = sorted(name for name, passed in required.items() if not passed)
if failed:
raise SystemExit("downloaded replay receipt failed: " + ", ".join(failed))
result["full_result_exported_and_validated"] = True
result["full_result_retained"] = False
result["full_result_sha256"] = hashlib.sha256(full_bytes).hexdigest()
result["full_result_bytes"] = len(full_bytes)
result["full_result_fixed_checks"] = required
result_path.write_text(json.dumps(result, indent=2, sort_keys=True) + "\n", encoding="utf-8")
full_path.unlink()
PY
python3 "$RESULT_HELPER" validate-replay \
--result-path "$result_path" \
--full-result "$full_result" \
--target-db "$target_db" \
--expected-cory-replay-db "$CORY_REPLAY_DB" \
--expected-cory-replay-fingerprint "$CORY_REPLAY_FINGERPRINT"
}
mark_result_export_failure() {
local result_path="$1"
local exit_code="$2"
local cleanup_exit_code="$3"
RESULT_PATH="$result_path" EXIT_CODE="$exit_code" CLEANUP_EXIT_CODE="$cleanup_exit_code" python3 - <<'PY'
import json
import os
from pathlib import Path
path = Path(os.environ["RESULT_PATH"])
payload = json.loads(path.read_text(encoding="utf-8"))
payload["status"] = "fail"
payload["full_result_exported_and_validated"] = False
payload["full_result_retained"] = False
payload["full_result_export_failure"] = "receipt_download_validation_or_remote_export_cleanup_failed"
payload["ssh_exit_code"] = int(os.environ["EXIT_CODE"])
payload["remote_full_result_export_removed"] = int(os.environ["CLEANUP_EXIT_CODE"]) == 0
path.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8")
PY
python3 "$RESULT_HELPER" mark-export-failure \
--result-path "$result_path" \
--exit-code "$exit_code" \
--cleanup-exit-code "$cleanup_exit_code"
}
runner_main() {
@ -892,16 +777,7 @@ runner_main() {
if [[ "$exit_code" -ne 0 ]]; then
mark_result_export_failure "$result_path" "$exit_code" "$export_cleanup_exit_code"
else
RESULT_PATH="$result_path" python3 - <<'PY'
import json
import os
from pathlib import Path
path = Path(os.environ["RESULT_PATH"])
payload = json.loads(path.read_text(encoding="utf-8"))
payload["remote_full_result_export_removed"] = True
path.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8")
PY
python3 "$RESULT_HELPER" mark-export-removed --result-path "$result_path"
fi
fi
printf 'sanitized_result=%s\n' "$result_path"

179
scripts/gcp_iap_operator_result.py Executable file
View file

@ -0,0 +1,179 @@
#!/usr/bin/env python3
"""Sanitize and validate local GCP IAP operator result artifacts."""
from __future__ import annotations
import argparse
import hashlib
import json
from pathlib import Path
from typing import Any
ALLOWED_REMOTE_FIELDS = {
"operation",
"request_id",
"target_db",
"status",
"instance",
"expected_internal_ip",
"active_state",
"sub_state",
"nrestarts",
"dispatcher_version",
"dispatcher_sha256",
"bundle_commit",
"bundle_sha256",
"result_sha256",
"result_bytes",
"full_result_export_ready",
"fixed_baseline_validation",
"no_message_send",
"live_service_unchanged",
"live_profile_unchanged",
"clone_database_remaining",
"run_directory_remaining",
"legacy_directory_remaining",
"export_file_remaining",
"leftover_temp_directory_count",
"rollback_database_datallowconn",
"rollback_database_connections",
}
def write_sanitized_result(args: argparse.Namespace) -> int:
raw = args.raw_stdout.read_bytes()
exit_code = args.exit_code
remote: dict[str, Any] = {}
if exit_code == 0:
try:
candidate = json.loads(raw.decode("utf-8").splitlines()[-1])
remote = {key: value for key, value in candidate.items() if key in ALLOWED_REMOTE_FIELDS}
except (AttributeError, UnicodeDecodeError, json.JSONDecodeError, IndexError):
exit_code = 90
payload = {
"schema": "livingip.gcpIapOperatorSanitizedResult.v1",
"operation": args.operation,
"request_id": args.request_id,
"target_db": args.target_db,
"status": "pass" if exit_code == 0 and remote.get("status") == "pass" else "fail",
"ssh_exit_code": exit_code,
"remote_result": remote,
"remote_stdout_sha256": hashlib.sha256(raw).hexdigest(),
"remote_stdout_bytes": len(raw),
"raw_stdout_retained": False,
"raw_stderr_retained": False,
"credential_values_logged": False,
"ssh_debug_enabled": False,
}
args.result_path.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8")
return 0
def validate_downloaded_replay(args: argparse.Namespace) -> int:
result = json.loads(args.result_path.read_text(encoding="utf-8"))
full_bytes = args.full_result.read_bytes()
full = json.loads(full_bytes)
remote = result.get("remote_result") or {}
checks = full.get("checks") or {}
required = {
"result_hash": hashlib.sha256(full_bytes).hexdigest() == remote.get("result_sha256"),
"result_size": len(full_bytes) == remote.get("result_bytes"),
"status": full.get("status") == "pass",
"target": full.get("target_database") == args.target_db,
"strict_six_of_six": (full.get("score") or {}).get("passes") == 6
and (full.get("score") or {}).get("expected_prompt_count") == 6,
"all_checks": bool(checks) and all(checks.values()),
}
if args.target_db == args.expected_cory_replay_db:
expected_counts = {
"claims": 1837,
"sources": 4145,
"claim_edges": 4916,
"claim_evidence": 4670,
"kb_proposals": 26,
}
before_counts = (full.get("canonical_status_before") or {}).get("high_signal_rows") or {}
after_counts = (full.get("canonical_status_after") or {}).get("high_signal_rows") or {}
required.update(
{
"fixed_fingerprint_before": (full.get("database_fingerprint_before") or {}).get("sha256")
== args.expected_cory_replay_fingerprint,
"fixed_fingerprint_after": (full.get("database_fingerprint_after") or {}).get("sha256")
== args.expected_cory_replay_fingerprint,
"fixed_counts_before": all(before_counts.get(key) == value for key, value in expected_counts.items()),
"fixed_counts_after": all(after_counts.get(key) == value for key, value in expected_counts.items()),
}
)
failed = sorted(name for name, passed in required.items() if not passed)
if failed:
raise SystemExit("downloaded replay receipt failed: " + ", ".join(failed))
result["full_result_exported_and_validated"] = True
result["full_result_retained"] = False
result["full_result_sha256"] = hashlib.sha256(full_bytes).hexdigest()
result["full_result_bytes"] = len(full_bytes)
result["full_result_fixed_checks"] = required
args.result_path.write_text(json.dumps(result, indent=2, sort_keys=True) + "\n", encoding="utf-8")
args.full_result.unlink()
return 0
def mark_export_failure(args: argparse.Namespace) -> int:
payload = json.loads(args.result_path.read_text(encoding="utf-8"))
payload["status"] = "fail"
payload["full_result_exported_and_validated"] = False
payload["full_result_retained"] = False
payload["full_result_export_failure"] = "receipt_download_validation_or_remote_export_cleanup_failed"
payload["ssh_exit_code"] = args.exit_code
payload["remote_full_result_export_removed"] = args.cleanup_exit_code == 0
args.result_path.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8")
return 0
def mark_export_removed(args: argparse.Namespace) -> int:
payload = json.loads(args.result_path.read_text(encoding="utf-8"))
payload["remote_full_result_export_removed"] = True
args.result_path.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8")
return 0
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(description=__doc__)
subparsers = parser.add_subparsers(dest="command", required=True)
sanitize = subparsers.add_parser("sanitize")
sanitize.add_argument("--result-path", type=Path, required=True)
sanitize.add_argument("--raw-stdout", type=Path, required=True)
sanitize.add_argument("--operation", required=True)
sanitize.add_argument("--request-id", required=True)
sanitize.add_argument("--target-db", required=True)
sanitize.add_argument("--exit-code", type=int, required=True)
sanitize.set_defaults(handler=write_sanitized_result)
replay = subparsers.add_parser("validate-replay")
replay.add_argument("--result-path", type=Path, required=True)
replay.add_argument("--full-result", type=Path, required=True)
replay.add_argument("--target-db", required=True)
replay.add_argument("--expected-cory-replay-db", required=True)
replay.add_argument("--expected-cory-replay-fingerprint", required=True)
replay.set_defaults(handler=validate_downloaded_replay)
export_failure = subparsers.add_parser("mark-export-failure")
export_failure.add_argument("--result-path", type=Path, required=True)
export_failure.add_argument("--exit-code", type=int, required=True)
export_failure.add_argument("--cleanup-exit-code", type=int, required=True)
export_failure.set_defaults(handler=mark_export_failure)
export_removed = subparsers.add_parser("mark-export-removed")
export_removed.add_argument("--result-path", type=Path, required=True)
export_removed.set_defaults(handler=mark_export_removed)
return parser.parse_args()
def main() -> int:
args = parse_args()
return args.handler(args)
if __name__ == "__main__":
raise SystemExit(main())

64
tests/test_gcp_iap_operator_access.py
View file

@ -10,6 +10,7 @@ REPO_ROOT = Path(__file__).resolve().parents[1]
PLAN = REPO_ROOT / "ops" / "plan_gcp_iap_operator_access.py"
WORKFLOW = REPO_ROOT / ".github" / "workflows" / "gcp-iap-operator.yml"
OPERATOR = REPO_ROOT / "scripts" / "gcp_iap_operator.sh"
RESULT_HELPER = REPO_ROOT / "scripts" / "gcp_iap_operator_result.py"
def run_plan(*args: str) -> str:
@ -245,6 +246,7 @@ def test_operator_cleanup_is_bound_to_marker_clone_prefix_and_exact_run_dir() ->
def test_operator_exports_and_fixed_validates_full_replay_before_cleanup() -> None:
source = OPERATOR.read_text(encoding="utf-8")
result_helper = RESULT_HELPER.read_text(encoding="utf-8")
assert 'CORY_REPLAY_FINGERPRINT="48ea5332c4f303dea02d503447f1ac8e1d8aa00e60d198a566d628b37dbdac1c"' in source
assert 'export_path="$caller_home/.teleo-iap-result-$request_id.json"' in source
@ -255,10 +257,10 @@ def test_operator_exports_and_fixed_validates_full_replay_before_cleanup() -> No
assert source.index('validate_downloaded_replay "$result_path" "$full_result" "$target_db"') < source.index(
'rm -f -- "$full_result"'
)
assert '"strict_six_of_six"' in source
assert '"fixed_fingerprint_before"' in source
assert '"fixed_counts_after"' in source
assert 'payload["remote_full_result_export_removed"] = True' in source
assert '"strict_six_of_six"' in result_helper
assert '"fixed_fingerprint_before"' in result_helper
assert '"fixed_counts_after"' in result_helper
assert 'payload["remote_full_result_export_removed"] = True' in result_helper
def test_operator_uses_five_minute_key_private_modes_and_cleanup_trap() -> None:
@ -390,35 +392,31 @@ def test_clone_runner_scps_only_request_derived_bundle_over_iap(tmp_path: Path,
observed_fingerprint = expected_fingerprint if valid_receipt else "0" * 64
full_receipt.write_text(
json.dumps(
{
"status": "pass",
"target_database": "teleo_clone_cory_20260712t1940z",
"score": {"passes": 6, "expected_prompt_count": 6},
"checks": {"all_runtime_checks": True},
"database_fingerprint_before": {
"sha256": observed_fingerprint
},
"database_fingerprint_after": {
"sha256": observed_fingerprint
},
"canonical_status_before": {
"high_signal_rows": {
"claims": 1837,
"sources": 4145,
"claim_edges": 4916,
"claim_evidence": 4670,
"kb_proposals": 26,
}
},
"canonical_status_after": {
"high_signal_rows": {
"claims": 1837,
"sources": 4145,
"claim_edges": 4916,
"claim_evidence": 4670,
"kb_proposals": 26,
}
},
{
"status": "pass",
"target_database": "teleo_clone_cory_20260712t1940z",
"score": {"passes": 6, "expected_prompt_count": 6},
"checks": {"all_runtime_checks": True},
"database_fingerprint_before": {"sha256": observed_fingerprint},
"database_fingerprint_after": {"sha256": observed_fingerprint},
"canonical_status_before": {
"high_signal_rows": {
"claims": 1837,
"sources": 4145,
"claim_edges": 4916,
"claim_evidence": 4670,
"kb_proposals": 26,
}
},
"canonical_status_after": {
"high_signal_rows": {
"claims": 1837,
"sources": 4145,
"claim_edges": 4916,
"claim_evidence": 4670,
"kb_proposals": 26,
}
},
},
sort_keys=True,
)