848 lines
46 KiB
Python
848 lines
46 KiB
Python
# #region Tooling.LabReset.Chart8Scope [C:4] [TYPE Module] [SEMANTICS lab,ownership,selection,reset]
|
|
# @BRIEF Select only the reviewed chart-8 catalog/scenario closure for an offline row reset.
|
|
# @PRE Scope IDs are explicit and rows are reflected from the live PostgreSQL transaction.
|
|
# @INVARIANT Selection never discovers a first/latest scenario, pin, run or release.
|
|
import argparse
|
|
from contextlib import contextmanager
|
|
from datetime import UTC, datetime
|
|
from hashlib import sha256
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import stat
|
|
import sys
|
|
|
|
from dashboard_testing_lab_reset_archive import require
|
|
from dashboard_testing_lab_reset_archive import (
|
|
Stop, digest, encode, json_bytes, private_directory, read_bytes, write_bytes,
|
|
)
|
|
import dashboard_testing_lab_reset_db as database
|
|
from dashboard_testing_lab_reset_db import node
|
|
|
|
|
|
# #region Tooling.LabReset.Chart8Scope.Schema [C:2] [TYPE Function]
|
|
def validate_scope(scope):
|
|
require(type(scope) is dict and scope.get("schema_version") == 1,
|
|
"LAB_RESET_SCOPE_SCHEMA_INVALID")
|
|
required = {"schema_version", "database_name", "candidate_manifest_path", "candidate_manifest_sha256", "repository_id",
|
|
"dashboard_id", "dashboard_key", "catalog_id",
|
|
"catalog_current_revision_id", "preserve_release_id", "scenario_id",
|
|
"scenario_revision_id", "agent_run_ids", "scenario_entry_deletion_authorized"}
|
|
require(set(scope) == required, "LAB_RESET_SCOPE_SCHEMA_INVALID")
|
|
candidate_value = scope["candidate_manifest_path"]
|
|
require(type(candidate_value) is str, "LAB_RESET_SCOPE_IDENTITY_CONFLICT")
|
|
candidate_path = Path(candidate_value)
|
|
archive_root = Path.home() / "ss-tools-evidence-archive" / "lab-resets"
|
|
require(scope["database_name"] == "ss_tools"
|
|
and scope["repository_id"] == "3acdeba7-6e1c-4f60-ae9c-1b5b10567418"
|
|
and scope["dashboard_id"] == 2 and scope["dashboard_key"] == "dashboard-2"
|
|
and scope["preserve_release_id"] == "23917070-17f8-4cec-a973-05ac3c1f0cd5"
|
|
and candidate_path.is_absolute() and candidate_path.is_relative_to(archive_root)
|
|
and type(scope["candidate_manifest_sha256"]) is str
|
|
and re.fullmatch(r"[0-9a-f]{64}", scope["candidate_manifest_sha256"]) is not None,
|
|
"LAB_RESET_SCOPE_IDENTITY_CONFLICT")
|
|
identifiers = (scope["catalog_id"], scope["catalog_current_revision_id"], scope["scenario_id"],
|
|
scope["scenario_revision_id"])
|
|
uuid_pattern = r"[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}"
|
|
require(all(type(value) is str and re.fullmatch(uuid_pattern, value) for value in identifiers),
|
|
"LAB_RESET_SCOPE_IDENTITY_CONFLICT")
|
|
require(type(scope["agent_run_ids"]) is list and scope["agent_run_ids"]
|
|
and all(type(value) is str and re.fullmatch(uuid_pattern, value) for value in scope["agent_run_ids"])
|
|
and len(set(scope["agent_run_ids"])) == len(scope["agent_run_ids"]),
|
|
"LAB_RESET_SCOPE_IDENTITY_CONFLICT")
|
|
require(type(scope["scenario_entry_deletion_authorized"]) is bool,
|
|
"LAB_RESET_SCOPE_AUTHORIZATION_INVALID")
|
|
# #endregion Tooling.LabReset.Chart8Scope.Schema
|
|
|
|
|
|
# #region Tooling.LabReset.Chart8Scope.Contains [C:2] [TYPE Function]
|
|
def contains(value, expected):
|
|
if type(value) is str:
|
|
return value == expected
|
|
if type(value) is dict:
|
|
return any(contains(key, expected) or contains(item, expected) for key, item in value.items())
|
|
if type(value) in (list, tuple):
|
|
return any(contains(item, expected) for item in value)
|
|
return False
|
|
# #endregion Tooling.LabReset.Chart8Scope.Contains
|
|
|
|
|
|
# #region Tooling.LabReset.Chart8Scope.AgentRun [C:2] [TYPE Function]
|
|
def agent_run_matches_scope(row, scope):
|
|
return (row.get("intent") == "manual_reference_capture"
|
|
and row.get("dashboard_id") == str(scope["dashboard_id"])
|
|
and row.get("environment_id") == "full-flow-preprod"
|
|
and contains(row.get("context_snapshot"), scope["catalog_id"]))
|
|
# #endregion Tooling.LabReset.Chart8Scope.AgentRun
|
|
|
|
|
|
# #region Tooling.LabReset.Chart8Scope.Select [C:4] [TYPE Function]
|
|
# @POST Selected nodes are exact rows in the authored repo/dashboard/catalog/scenario/run scope.
|
|
def select_chart8_nodes(scope, rows, keys, *, for_apply=False):
|
|
validate_scope(scope)
|
|
selected = set()
|
|
|
|
def add(table, predicate):
|
|
require(table in rows and table in keys and keys[table], "LAB_RESET_SCOPE_TABLE_MISSING")
|
|
for row in rows[table]:
|
|
if predicate(row):
|
|
selected.add(node(table, row, keys[table]))
|
|
|
|
catalog_id = scope["catalog_id"]
|
|
scenario_id = scope["scenario_id"]
|
|
revision_id = scope["scenario_revision_id"]
|
|
add("baseline_catalog_heads", lambda row: row.get("id") == catalog_id
|
|
and row.get("repository") == scope["repository_id"]
|
|
and row.get("dashboard_key") == scope["dashboard_key"]
|
|
and row.get("current_revision_id") == scope["catalog_current_revision_id"])
|
|
heads = [row for row in rows["baseline_catalog_heads"]
|
|
if node("baseline_catalog_heads", row, keys["baseline_catalog_heads"]) in selected]
|
|
require(len(heads) == 1, "LAB_RESET_SCOPE_ANCHOR_CONFLICT")
|
|
add("baseline_catalog_revisions", lambda row: row.get("catalog_id") == catalog_id)
|
|
catalog_revisions = [row for row in rows["baseline_catalog_revisions"]
|
|
if node("baseline_catalog_revisions", row, keys["baseline_catalog_revisions"]) in selected]
|
|
revision_ids = {row["id"] for row in catalog_revisions}
|
|
require(catalog_revisions and scope["catalog_current_revision_id"] in revision_ids,
|
|
"LAB_RESET_SCOPE_CATALOG_REVISION_CONFLICT")
|
|
add("baseline_catalog_memberships", lambda row: row.get("catalog_revision_id") in revision_ids)
|
|
add("publication_operations", lambda row: row.get("catalog_revision_id") in revision_ids)
|
|
|
|
add("scenario_registry_entries", lambda row: row.get("scenario_id") == scenario_id)
|
|
entries = [row for row in rows["scenario_registry_entries"]
|
|
if node("scenario_registry_entries", row, keys["scenario_registry_entries"]) in selected]
|
|
require(len(entries) == 1, "LAB_RESET_SCOPE_SCENARIO_CONFLICT")
|
|
require(entries[0].get("dashboard_id") == scope["dashboard_id"],
|
|
"LAB_RESET_SCOPE_SCENARIO_DASHBOARD_CONFLICT")
|
|
require(entries[0].get("current_revision_id") == revision_id,
|
|
"LAB_RESET_SCOPE_SCENARIO_REVISION_CONFLICT")
|
|
if for_apply:
|
|
require(scope["scenario_entry_deletion_authorized"], "LAB_RESET_SCENARIO_ENTRY_AUTHORIZATION_REQUIRED")
|
|
add("scenario_revisions", lambda row: row.get("scenario_id") == scenario_id
|
|
and row.get("revision_id") == revision_id)
|
|
scenario_revisions = [row for row in rows["scenario_revisions"]
|
|
if node("scenario_revisions", row, keys["scenario_revisions"]) in selected]
|
|
require(len(scenario_revisions) == 1, "LAB_RESET_SCOPE_SCENARIO_REVISION_CONFLICT")
|
|
graph_snapshot = scenario_revisions[0].get("graph_snapshot")
|
|
require(contains(graph_snapshot, catalog_id)
|
|
and contains(graph_snapshot, scope["catalog_current_revision_id"]),
|
|
"LAB_RESET_SCOPE_SCENARIO_GRAPH_CONFLICT")
|
|
|
|
run_tables = ("scenario_runs", "agent_evaluations", "scenario_step_runs", "scenario_step_leases",
|
|
"provider_capacity_leases", "provider_operation_receipts", "scenario_investigation_queue",
|
|
"scenario_investigation_signals", "scenario_recurring_failure_episodes",
|
|
"scenario_notification_events", "scenario_outbox_events", "agent_runs",
|
|
"agent_run_events", "approval_gates", "draft_artifacts",
|
|
"visual_capture_authority_bindings", "visual_capture_operations",
|
|
"agent_authoring_workspaces", "scenario_revision_materializations",
|
|
"scenario_compiled_handles", "scenario_draft_packs", "scenario_test_pack_profiles",
|
|
"scenario_test_pack_profile_receipts")
|
|
for table in run_tables:
|
|
require(table in rows and table in keys, "LAB_RESET_SCOPE_TABLE_MISSING")
|
|
|
|
add("agent_authoring_workspaces", lambda row: row.get("scenario_id") == scenario_id
|
|
and row.get("base_revision_id") == revision_id)
|
|
add("scenario_revision_materializations", lambda row: row.get("revision_id") == revision_id)
|
|
add("scenario_compiled_handles", lambda row: row.get("consumed_by_revision_id") == revision_id)
|
|
add("scenario_draft_packs", lambda row: row.get("consumed_by_revision_id") == revision_id)
|
|
draft_rows = [row for row in rows["scenario_draft_packs"]
|
|
if node("scenario_draft_packs", row, keys["scenario_draft_packs"]) in selected]
|
|
profile_ids = {row.get("profile_receipt", {}).get("profile_handle_id")
|
|
for row in draft_rows if type(row.get("profile_receipt")) is dict}
|
|
profile_ids.discard(None)
|
|
add("scenario_test_pack_profiles", lambda row: row.get("profile_handle_id") in profile_ids)
|
|
add("scenario_test_pack_profile_receipts", lambda row: row.get("profile_handle_id") in profile_ids)
|
|
|
|
add("scenario_runs", lambda row: row.get("scenario_id") == scenario_id)
|
|
runs = [row for row in rows["scenario_runs"]
|
|
if node("scenario_runs", row, keys["scenario_runs"]) in selected]
|
|
run_ids = {row["id"] for row in runs}
|
|
require(runs and all(contains(row.get("runner_plan"), catalog_id)
|
|
and contains(row.get("runner_plan"), scope["catalog_current_revision_id"])
|
|
and contains(row.get("target_snapshot"), catalog_id)
|
|
and contains(row.get("target_snapshot"), scope["catalog_current_revision_id"]) for row in runs),
|
|
"LAB_RESET_SCOPE_RUN_PIN_CONFLICT")
|
|
add("agent_evaluations", lambda row: row.get("scenario_run_id") in run_ids)
|
|
add("scenario_step_runs", lambda row: row.get("run_id") in run_ids)
|
|
add("scenario_step_leases", lambda row: row.get("run_id") in run_ids)
|
|
add("provider_capacity_leases", lambda row: row.get("run_id") in run_ids)
|
|
add("provider_operation_receipts", lambda row: row.get("run_id") in run_ids)
|
|
add("scenario_recurring_failure_episodes", lambda row: row.get("scenario_id") == scenario_id)
|
|
episodes = [row for row in rows["scenario_recurring_failure_episodes"]
|
|
if node("scenario_recurring_failure_episodes", row,
|
|
keys["scenario_recurring_failure_episodes"]) in selected]
|
|
episode_ids = {row["id"] for row in episodes}
|
|
add("scenario_investigation_queue", lambda row: row.get("scenario_id") == scenario_id
|
|
or row.get("run_id") in run_ids or row.get("episode_id") in episode_ids)
|
|
add("scenario_investigation_signals", lambda row: row.get("run_id") in run_ids
|
|
or row.get("episode_id") in episode_ids)
|
|
add("scenario_notification_events", lambda row: row.get("scenario_id") == scenario_id
|
|
or row.get("run_id") in run_ids)
|
|
add("scenario_outbox_events", lambda row: row.get("aggregate_type") == "scenario_revision"
|
|
and row.get("aggregate_id") == revision_id)
|
|
|
|
agent_ids = set(scope["agent_run_ids"])
|
|
agent_rows = [row for row in rows["agent_runs"] if row.get("id") in agent_ids]
|
|
require(len(agent_rows) == len(agent_ids)
|
|
and all(agent_run_matches_scope(row, scope) for row in agent_rows),
|
|
"LAB_RESET_SCOPE_AGENT_RUN_CONFLICT")
|
|
add("agent_runs", lambda row: row.get("id") in agent_ids)
|
|
add("agent_run_events", lambda row: row.get("run_id") in agent_ids)
|
|
add("approval_gates", lambda row: row.get("run_id") in agent_ids)
|
|
add("draft_artifacts", lambda row: row.get("run_id") in agent_ids)
|
|
add("visual_capture_authority_bindings", lambda row: row.get("agent_run_id") in agent_ids
|
|
and row.get("dashboard_id") == scope["dashboard_id"]
|
|
and row.get("release_id") == scope["preserve_release_id"])
|
|
add("visual_capture_operations", lambda row: row.get("agent_run_id") in agent_ids)
|
|
return selected
|
|
# #endregion Tooling.LabReset.Chart8Scope.Select
|
|
|
|
|
|
# #region Tooling.LabReset.Chart8Scope.Snapshot [C:3] [TYPE Function]
|
|
def snapshot_chart8(connection, table_scope, scope, identity, *, for_apply=False):
|
|
tables, keys, fks, rows = database.reflect(connection, table_scope)
|
|
if already_empty(scope, rows, keys):
|
|
return {"schema_version": 1, "status": "already_empty", "scope": scope_ids(scope),
|
|
"database_identity_sha256": digest(identity), "selected_rows": 0, "selected_tables": 0}, None
|
|
|
|
selected = select_chart8_nodes(scope, rows, keys, for_apply=for_apply)
|
|
preserved_rows = {name: [row for row in table_rows
|
|
if database.node(name, row, keys[name]) not in selected]
|
|
for name, table_rows in rows.items()}
|
|
residual = residual_reference_counts(scope, preserved_rows)
|
|
if residual:
|
|
_emit("EXPLORE", "preserved_scope_check", "Preserved rows still reference selected scope",
|
|
error_code="LAB_RESET_PRESERVED_REFERENCE_CONFLICT", residual_rows_by_table=residual)
|
|
require(False, "LAB_RESET_PRESERVED_REFERENCE_CONFLICT")
|
|
edges = database.foreign_edges(rows, keys, selected, fks)
|
|
order = database.catalog_child_first(rows, keys, selected, fks)
|
|
summary = {}
|
|
for name, table in sorted(tables.items()):
|
|
selected_rows = [row for row in rows[name]
|
|
if database.node(name, row, keys[name]) in selected]
|
|
preserved_rows = [row for row in rows[name]
|
|
if database.node(name, row, keys[name]) not in selected]
|
|
primary_keys = sorted([database.primary_key(row, keys[name]) for row in selected_rows],
|
|
key=lambda value: json_bytes(value))
|
|
summary[name] = {
|
|
"columns": [[str(column.name), str(column.type)] for column in table.columns],
|
|
"primary_key_columns": keys[name], "foreign_keys": fks[name],
|
|
"selected_count": len(selected_rows), "selected_primary_keys": primary_keys,
|
|
"selected_content_sha256": digest(sorted([encode(row) for row in selected_rows], key=digest)),
|
|
"preserved_count": len(preserved_rows),
|
|
}
|
|
manifest = {"schema_version": 1, "status": "prepared", "scope": scope_ids(scope),
|
|
"database_identity_sha256": digest(identity), "tables": summary,
|
|
"total_rows": len(selected), "total_tables": sum(item["selected_count"] > 0
|
|
for item in summary.values()),
|
|
"foreign_key_edges": len(edges), "delete_order_sha256": digest(order)}
|
|
return manifest, (tables, keys, rows, selected, order)
|
|
# #endregion Tooling.LabReset.Chart8Scope.Snapshot
|
|
|
|
|
|
# #region Tooling.LabReset.Chart8Scope.Empty [C:2] [TYPE Function]
|
|
def scope_ids(scope):
|
|
return {key: scope[key] for key in (
|
|
"database_name", "repository_id", "dashboard_id", "dashboard_key", "catalog_id",
|
|
"catalog_current_revision_id", "preserve_release_id", "scenario_id", "scenario_revision_id",
|
|
)}
|
|
|
|
|
|
def already_empty(scope, rows, keys):
|
|
repositories = [row for row in rows.get("git_repositories", [])
|
|
if row.get("id") == scope["repository_id"]
|
|
and row.get("dashboard_id") == scope["dashboard_id"]]
|
|
releases = [row for row in rows.get("dashboard_releases", [])
|
|
if row.get("id") == scope["preserve_release_id"]
|
|
and row.get("repository_id") == scope["repository_id"]]
|
|
require(len(repositories) == 1 and len(releases) == 1,
|
|
"LAB_RESET_PRESERVED_ANCHOR_CONFLICT")
|
|
targets = historical_targets(scope)
|
|
residual = [(name, row) for name, table_rows in rows.items()
|
|
for row in table_rows
|
|
if name not in ("git_repositories", "dashboard_releases")
|
|
and any(contains(row, target) for target in targets)
|
|
and not historical_delivery_reference(name, row, rows, scope, targets)]
|
|
return not residual
|
|
|
|
|
|
def historical_delivery_reference(table_name, row, rows, scope, targets):
|
|
if table_name != "investigation_delivery_events":
|
|
return False
|
|
references = row.get("target_references")
|
|
if (type(references) is not dict or references.get("scenario") != scope["scenario_id"]
|
|
or row.get("kind") not in {"episode_opened", "restored", "check_failure", "warning"}):
|
|
return False
|
|
event_id = row.get("id")
|
|
live_episodes = {item.get("id") for item in rows.get("scenario_recurring_failure_episodes", [])}
|
|
if row.get("episode_id") in live_episodes:
|
|
return False
|
|
if any(item.get("target_type") == "scenario" and item.get("target_id") == scope["scenario_id"]
|
|
for item in rows.get("investigation_subscriptions", [])):
|
|
return False
|
|
if any(item.get("event_id") == event_id for item in rows.get("investigation_recipient_receipts", [])):
|
|
return False
|
|
if any(event_id in (item.get("event_ids") or [])
|
|
for item in rows.get("investigation_recipient_notifications", [])):
|
|
return False
|
|
for field, value in row.items():
|
|
if field == "target_references":
|
|
if any(ref_key != "scenario" and any(contains(ref_value, target) for target in targets)
|
|
for ref_key, ref_value in references.items()):
|
|
return False
|
|
elif any(contains(value, target) for target in targets):
|
|
return False
|
|
return True
|
|
|
|
|
|
def historical_delivery_reference_count(rows, scope, targets):
|
|
return sum(historical_delivery_reference("investigation_delivery_events", row, rows, scope, targets)
|
|
for row in rows.get("investigation_delivery_events", []))
|
|
|
|
|
|
def historical_targets(scope):
|
|
return {scope["catalog_id"], scope["catalog_current_revision_id"], scope["scenario_id"],
|
|
scope["scenario_revision_id"], *scope["agent_run_ids"]}
|
|
|
|
|
|
def residual_reference_counts(scope, rows):
|
|
targets = historical_targets(scope)
|
|
return {name: count for name, table_rows in rows.items()
|
|
if name not in ("git_repositories", "dashboard_releases")
|
|
and (count := sum(any(contains(row, target) for target in targets)
|
|
and not historical_delivery_reference(name, row, rows, scope, targets)
|
|
for row in table_rows))}
|
|
|
|
|
|
def require_preserved_rows_clean(scope, rows):
|
|
residual = residual_reference_counts(scope, rows)
|
|
require(not residual, "LAB_RESET_PRESERVED_REFERENCE_CONFLICT")
|
|
# #endregion Tooling.LabReset.Chart8Scope.Empty
|
|
|
|
|
|
# #region Tooling.LabReset.Chart8Scope.Archive [C:3] [TYPE Function]
|
|
def archive_parent():
|
|
path = Path.home() / "ss-tools-evidence-archive" / "lab-resets"
|
|
path.mkdir(parents=True, exist_ok=True, mode=0o700)
|
|
require(not path.is_symlink() and stat.S_IMODE(path.stat().st_mode) == 0o700
|
|
and path.stat().st_uid == os.geteuid(), "LAB_RESET_PRIVATE_MODE_INVALID")
|
|
return private_directory(path, os.geteuid())
|
|
|
|
|
|
def new_archive_dir():
|
|
parent = archive_parent()
|
|
stem = datetime.now(UTC).strftime("%Y%m%dT%H%M%SZ")
|
|
for suffix in range(100):
|
|
name = stem if suffix == 0 else f"{stem}-{suffix:02d}"
|
|
path = parent / name
|
|
try:
|
|
os.mkdir(path, 0o700)
|
|
except FileExistsError:
|
|
continue
|
|
return private_directory(path, os.geteuid())
|
|
raise Stop("LAB_RESET_ARCHIVE_PATH_CONFLICT")
|
|
|
|
|
|
def write_archive(path, manifest, state):
|
|
raw_manifest = json_bytes(manifest)
|
|
rows_raw = json_bytes(database.archive_rows(state))
|
|
manifest_sha = write_bytes(path, "manifest.json", raw_manifest)
|
|
rows_sha = write_bytes(path, "selected-rows.json", rows_raw)
|
|
write_bytes(path, "manifest.sha256", f"{manifest_sha} manifest.json\n".encode())
|
|
write_bytes(path, "selected-rows.sha256", f"{rows_sha} selected-rows.json\n".encode())
|
|
require(read_bytes(path / "manifest.json", private=True) == raw_manifest
|
|
and read_bytes(path / "selected-rows.json", private=True) == rows_raw,
|
|
"LAB_RESET_ARCHIVE_READBACK_FAILED")
|
|
return manifest_sha, rows_sha
|
|
# #endregion Tooling.LabReset.Chart8Scope.Archive
|
|
|
|
|
|
# #region Tooling.LabReset.Chart8Scope.PreviousManifest [C:2] [TYPE Function]
|
|
def manifest_primary_keys(primary_keys, columns):
|
|
values = []
|
|
for primary_key in primary_keys:
|
|
if type(primary_key) is dict:
|
|
require(set(primary_key) == set(columns), "LAB_RESET_PRIOR_MANIFEST_PK_CONFLICT")
|
|
values.append([primary_key[column] for column in columns])
|
|
else:
|
|
require(type(primary_key) is list and len(primary_key) == len(columns),
|
|
"LAB_RESET_PRIOR_MANIFEST_PK_CONFLICT")
|
|
values.append(primary_key)
|
|
return sorted(values, key=json_bytes)
|
|
|
|
|
|
def compare_previous_manifest(path, scope, manifest):
|
|
previous = Path(path)
|
|
private_directory(previous.parent, os.geteuid())
|
|
raw = read_bytes(previous, private=True)
|
|
require(sha256(raw).hexdigest() == scope["candidate_manifest_sha256"],
|
|
"LAB_RESET_PRIOR_MANIFEST_DIGEST_CONFLICT")
|
|
prior = json.loads(raw)
|
|
require(prior.get("total_rows") == manifest["total_rows"]
|
|
and prior.get("total_tables") == manifest["total_tables"]
|
|
and prior.get("scope") == {key: scope[key] for key in (
|
|
"repository_id", "dashboard_id", "dashboard_key", "catalog_id",
|
|
"catalog_current_revision_id", "scenario_id", "scenario_revision_id")},
|
|
"LAB_RESET_PRIOR_MANIFEST_SCOPE_CONFLICT")
|
|
current = {name: item for name, item in manifest["tables"].items() if item["selected_count"]}
|
|
require(set(prior.get("tables", {})) == set(current), "LAB_RESET_PRIOR_MANIFEST_TABLE_CONFLICT")
|
|
for name, entry in current.items():
|
|
expected = manifest_primary_keys(entry["selected_primary_keys"], entry["primary_key_columns"])
|
|
prior_pks = manifest_primary_keys(prior["tables"][name].get("pks", []),
|
|
entry["primary_key_columns"])
|
|
require(prior["tables"][name].get("count") == entry["selected_count"]
|
|
and prior["tables"][name].get("pk_columns") == entry["primary_key_columns"]
|
|
and prior_pks == expected,
|
|
"LAB_RESET_PRIOR_MANIFEST_PK_CONFLICT")
|
|
return raw
|
|
# #endregion Tooling.LabReset.Chart8Scope.PreviousManifest
|
|
|
|
|
|
# #region Tooling.LabReset.Chart8Scope.IO [C:3] [TYPE Function]
|
|
def load_scope(path):
|
|
source = Path(path)
|
|
require(source.is_absolute() and not source.is_symlink(), "LAB_RESET_SCOPE_PATH_INVALID")
|
|
raw = read_bytes(source, private=True)
|
|
scope = json.loads(raw)
|
|
validate_scope(scope)
|
|
return scope
|
|
|
|
|
|
def load_database_url(path):
|
|
source = Path(path)
|
|
require(source.is_absolute() and not source.is_symlink(), "LAB_RESET_DATABASE_CONFIG_INVALID")
|
|
raw = read_bytes(source, private=True)
|
|
try:
|
|
text = raw.decode("utf-8")
|
|
except UnicodeDecodeError as exc:
|
|
raise Stop("LAB_RESET_DATABASE_CONFIG_INVALID") from exc
|
|
lines = text.splitlines()
|
|
require(len(lines) == 1 and lines[0].startswith("DATABASE_URL=")
|
|
and len(lines[0]) > len("DATABASE_URL="), "LAB_RESET_DATABASE_CONFIG_INVALID")
|
|
return lines[0][len("DATABASE_URL="):]
|
|
|
|
|
|
def archive_member(directory, name):
|
|
path = Path(directory)
|
|
private_directory(path, os.geteuid())
|
|
return read_bytes(path / name, private=True)
|
|
|
|
|
|
def verified_archive(directory, expected_manifest_sha, expected_rows_sha, expected_candidate_sha):
|
|
path = private_directory(Path(directory), os.geteuid())
|
|
require(path.parent == archive_parent(), "LAB_RESET_ARCHIVE_PATH_CONFLICT")
|
|
require(re.fullmatch(r"[0-9a-f]{64}", expected_manifest_sha or "") is not None
|
|
and re.fullmatch(r"[0-9a-f]{64}", expected_rows_sha or "") is not None,
|
|
"LAB_RESET_ARCHIVE_DIGEST_REQUIRED")
|
|
manifest_raw = archive_member(path, "manifest.json")
|
|
rows_raw = archive_member(path, "selected-rows.json")
|
|
require(sha256(manifest_raw).hexdigest() == expected_manifest_sha
|
|
and sha256(rows_raw).hexdigest() == expected_rows_sha,
|
|
"LAB_RESET_ARCHIVE_DIGEST_CONFLICT")
|
|
require(archive_member(path, "manifest.sha256") == f"{expected_manifest_sha} manifest.json\n".encode()
|
|
and archive_member(path, "selected-rows.sha256") ==
|
|
f"{expected_rows_sha} selected-rows.json\n".encode(),
|
|
"LAB_RESET_ARCHIVE_DIGEST_CONFLICT")
|
|
candidate_raw = archive_member(path, "previous-candidate-manifest.json")
|
|
require(sha256(candidate_raw).hexdigest() == expected_candidate_sha,
|
|
"LAB_RESET_PRIOR_MANIFEST_DIGEST_CONFLICT")
|
|
candidate_sha = f"{sha256(candidate_raw).hexdigest()} previous-candidate-manifest.json\n".encode()
|
|
require(archive_member(path, "previous-candidate-manifest.sha256") == candidate_sha,
|
|
"LAB_RESET_ARCHIVE_DIGEST_CONFLICT")
|
|
return path, json.loads(manifest_raw), json.loads(rows_raw), manifest_raw
|
|
# #endregion Tooling.LabReset.Chart8Scope.IO
|
|
|
|
|
|
# #region Tooling.LabReset.Chart8Scope.Manifest [C:3] [TYPE Function]
|
|
def _emit(marker, phase, intent, *, error_code=None, **payload):
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "backend"))
|
|
from src.core.cot_logger import build_cot_event
|
|
|
|
event = build_cot_event(src="Tooling.LabReset", marker=marker, intent=intent,
|
|
error_code=error_code, payload={"phase": phase, **payload})
|
|
print(json_bytes(event).decode("utf-8"), flush=True)
|
|
|
|
|
|
def _archive_matches_current(archived, state):
|
|
current = database.archive_rows(state)
|
|
return json_bytes(current) == json_bytes(archived)
|
|
|
|
|
|
def verify_deleted_counts(connection, table_scope, manifest, state):
|
|
from sqlalchemy import func, select
|
|
|
|
tables, keys, rows, selected, _ = state
|
|
database.table_names(connection, table_scope)
|
|
database.no_foreign_sessions(connection)
|
|
for name, table in tables.items():
|
|
count = connection.execute(select(func.count()).select_from(table)).scalar_one()
|
|
require(count == manifest["tables"][name]["preserved_count"],
|
|
"LAB_RESET_POST_DELETE_COUNT_CONFLICT")
|
|
for row in rows[name]:
|
|
if database.node(name, row, keys[name]) in selected:
|
|
predicate = [table.c[column] == row[column] for column in keys[name]]
|
|
require(connection.execute(select(table).where(*predicate)).first() is None,
|
|
"LAB_RESET_POST_DELETE_ROW_PRESENT")
|
|
|
|
|
|
# #region Tooling.LabReset.Chart8Scope.AppendOnly [C:3] [TYPE Function]
|
|
_APPEND_ONLY_TRIGGERS = {
|
|
("baseline_catalog_revisions", "baseline_catalog_append_only", "baseline_catalog_append_only"),
|
|
("baseline_catalog_memberships", "baseline_catalog_append_only", "baseline_catalog_append_only"),
|
|
}
|
|
|
|
|
|
def append_only_trigger_inventory(connection):
|
|
from sqlalchemy import text
|
|
|
|
rows = connection.execute(text("""
|
|
SELECT c.relname AS table_name, t.tgname AS trigger_name,
|
|
p.proname AS function_name, fn.nspname AS function_schema, t.tgenabled AS enabled,
|
|
t.tgisinternal AS internal, t.tgtype AS trigger_type, t.tgqual IS NULL AS no_when,
|
|
octet_length(t.tgargs) AS argument_bytes, l.lanname AS function_language,
|
|
pg_get_function_result(p.oid) AS function_result, p.prosrc AS function_source
|
|
FROM pg_trigger AS t
|
|
JOIN pg_class AS c ON c.oid = t.tgrelid
|
|
JOIN pg_namespace AS n ON n.oid = c.relnamespace
|
|
JOIN pg_proc AS p ON p.oid = t.tgfoid
|
|
JOIN pg_namespace AS fn ON fn.oid = p.pronamespace
|
|
JOIN pg_language AS l ON l.oid = p.prolang
|
|
WHERE n.nspname = 'public'
|
|
AND c.relname IN ('baseline_catalog_revisions', 'baseline_catalog_memberships')
|
|
AND NOT t.tgisinternal
|
|
ORDER BY c.relname, t.tgname
|
|
""")).mappings().all()
|
|
return [dict(row) for row in rows]
|
|
|
|
|
|
def validate_append_only_trigger_inventory(rows, expected_enabled):
|
|
require(type(rows) is list and len(rows) == len(_APPEND_ONLY_TRIGGERS),
|
|
"LAB_RESET_IMMUTABLE_TRIGGER_SCHEMA_CONFLICT")
|
|
observed = set()
|
|
for row in rows:
|
|
require(type(row) is dict, "LAB_RESET_IMMUTABLE_TRIGGER_SCHEMA_CONFLICT")
|
|
identity = (row.get("table_name"), row.get("trigger_name"), row.get("function_name"))
|
|
require(identity in _APPEND_ONLY_TRIGGERS and identity not in observed
|
|
and row.get("function_schema") == "public"
|
|
and row.get("internal") is False and row.get("enabled") == expected_enabled
|
|
and row.get("trigger_type") == 27 and row.get("no_when") is True
|
|
and row.get("argument_bytes") == 0,
|
|
"LAB_RESET_IMMUTABLE_TRIGGER_STATE_CONFLICT")
|
|
function_source = row.get("function_source")
|
|
normalized_source = re.sub(r"\s+", "", function_source).lower() if type(function_source) is str else None
|
|
expected_source = ("beginraiseexceptionusingmessage='catalog_revision_immutable',"
|
|
"errcode='23514',constraint='ck_baseline_catalog_append_only';end")
|
|
require(row.get("function_language") == "plpgsql" and row.get("function_result") == "trigger"
|
|
and normalized_source == expected_source,
|
|
"LAB_RESET_IMMUTABLE_TRIGGER_FUNCTION_CONFLICT")
|
|
observed.add(identity)
|
|
require(observed == _APPEND_ONLY_TRIGGERS,
|
|
"LAB_RESET_IMMUTABLE_TRIGGER_SCHEMA_CONFLICT")
|
|
|
|
|
|
@contextmanager
|
|
def temporarily_disable_append_only_triggers(connection, args):
|
|
from sqlalchemy import text
|
|
|
|
args.operation_stage = "inspect_append_only_triggers"
|
|
inventory = append_only_trigger_inventory(connection)
|
|
validate_append_only_trigger_inventory(inventory, "O")
|
|
quote = connection.dialect.identifier_preparer.quote
|
|
|
|
def restore_and_verify():
|
|
args.operation_stage = "restore_append_only_triggers"
|
|
for table_name, trigger_name, _ in sorted(_APPEND_ONLY_TRIGGERS):
|
|
connection.execute(text(
|
|
f"ALTER TABLE public.{quote(table_name)} ENABLE TRIGGER {quote(trigger_name)}"))
|
|
args.operation_stage = "verify_append_only_triggers_restored"
|
|
validate_append_only_trigger_inventory(append_only_trigger_inventory(connection), "O")
|
|
|
|
try:
|
|
args.operation_stage = "disable_append_only_triggers"
|
|
for table_name, trigger_name, _ in sorted(_APPEND_ONLY_TRIGGERS):
|
|
connection.execute(text(
|
|
f"ALTER TABLE public.{quote(table_name)} DISABLE TRIGGER {quote(trigger_name)}"))
|
|
args.operation_stage = "verify_append_only_triggers_disabled"
|
|
validate_append_only_trigger_inventory(append_only_trigger_inventory(connection), "D")
|
|
yield
|
|
except Exception as exc:
|
|
# PostgreSQL marks the transaction failed for DBAPI errors; the enclosing transaction rollback
|
|
# restores the trigger catalog state atomically with the selected rows.
|
|
if getattr(exc, "orig", None) is None:
|
|
restore_and_verify()
|
|
raise
|
|
else:
|
|
restore_and_verify()
|
|
# #endregion Tooling.LabReset.Chart8Scope.AppendOnly
|
|
|
|
|
|
def _safe_identifier(value):
|
|
return value if type(value) is str and re.fullmatch(r"[a-z_][a-z0-9_]{0,62}", value) else None
|
|
|
|
|
|
def safe_database_error_facts(exc):
|
|
original = getattr(exc, "orig", None)
|
|
diagnostic = getattr(original, "diag", None)
|
|
sqlstate = getattr(original, "sqlstate", None) or getattr(original, "pgcode", None)
|
|
if type(sqlstate) is not str or re.fullmatch(r"[0-9A-Z]{5}", sqlstate) is None:
|
|
sqlstate = None
|
|
table = _safe_identifier(getattr(diagnostic, "table_name", None))
|
|
column = _safe_identifier(getattr(diagnostic, "column_name", None))
|
|
constraint = _safe_identifier(getattr(diagnostic, "constraint_name", None))
|
|
return {"sqlstate": sqlstate, "table": table, "column": column, "constraint": constraint}
|
|
|
|
|
|
def _retain_safe_failure_facts(archive_dir, args, facts):
|
|
if args.phase != "apply" or archive_dir is None:
|
|
return None
|
|
trace = os.urandom(12).hex()
|
|
name = f"failure-facts-{trace}.json"
|
|
payload = {"schema_version": 1, "phase": "apply", "operation_stage": args.operation_stage,
|
|
"failure_type": args.failure_type, **facts}
|
|
write_bytes(archive_dir, name, json_bytes(payload))
|
|
return name
|
|
|
|
|
|
def _prepare_archive(manifest, state, candidate_path, previous_raw):
|
|
archive_dir = new_archive_dir()
|
|
manifest_sha, rows_sha = write_archive(archive_dir, manifest, state)
|
|
require(read_bytes(candidate_path, private=True) == previous_raw,
|
|
"LAB_RESET_PRIOR_MANIFEST_CHANGED")
|
|
write_bytes(archive_dir, "previous-candidate-manifest.json", previous_raw)
|
|
require(read_bytes(archive_dir / "previous-candidate-manifest.json", private=True) == previous_raw,
|
|
"LAB_RESET_ARCHIVE_READBACK_FAILED")
|
|
directory_fd = os.open(archive_dir, os.O_RDONLY | os.O_DIRECTORY)
|
|
try:
|
|
os.fsync(directory_fd)
|
|
finally:
|
|
os.close(directory_fd)
|
|
candidate_sha = sha256(previous_raw).hexdigest()
|
|
write_bytes(archive_dir, "previous-candidate-manifest.sha256",
|
|
f"{candidate_sha} previous-candidate-manifest.json\n".encode())
|
|
return archive_dir, manifest_sha, rows_sha
|
|
# #endregion Tooling.LabReset.Chart8Scope.Manifest
|
|
|
|
|
|
# #region Tooling.LabReset.Chart8Scope.Run [C:4] [TYPE Function]
|
|
def run_phase(args):
|
|
args.operation_stage = "load_scope_and_database"
|
|
scope = load_scope(args.scope)
|
|
database_url = load_database_url(args.database_url_file)
|
|
require(database_url.startswith("postgresql"), "LAB_RESET_POSTGRESQL_REQUIRED")
|
|
if args.phase == "apply":
|
|
archive_dir, reviewed, archived_rows, reviewed_raw = verified_archive(
|
|
args.archive_dir, args.manifest_sha256, args.rows_sha256, scope["candidate_manifest_sha256"])
|
|
require(reviewed.get("status") == "prepared" and reviewed.get("scope") == scope_ids(scope),
|
|
"LAB_RESET_ARCHIVE_SCOPE_CONFLICT")
|
|
else:
|
|
archive_dir = reviewed = archived_rows = reviewed_raw = None
|
|
|
|
from sqlalchemy import create_engine, inspect, text
|
|
from sqlalchemy.pool import NullPool
|
|
|
|
engine = create_engine(database_url, poolclass=NullPool, echo=False, hide_parameters=True)
|
|
try:
|
|
require(engine.dialect.name == "postgresql", "LAB_RESET_POSTGRESQL_REQUIRED")
|
|
if args.phase == "verify":
|
|
with engine.connect() as connection:
|
|
identity = dict(connection.execute(text(
|
|
"SELECT current_database() AS database, inet_server_addr() AS address, "
|
|
"inet_server_port() AS port, "
|
|
"(SELECT oid FROM pg_database WHERE datname=current_database()) AS oid"
|
|
)).mappings().one())
|
|
require(identity["database"] == scope["database_name"], "LAB_RESET_DATABASE_IDENTITY_CONFLICT")
|
|
names = inspect(connection).get_table_names(schema="public")
|
|
table_scope = {"database_tables": [{"table": name} for name in names],
|
|
"protected_tables": {}}
|
|
tables, keys, _, rows = database.reflect(connection, table_scope)
|
|
require(already_empty(scope, rows, keys), "LAB_RESET_NOOP_VERIFY_CONFLICT")
|
|
_emit("REFLECT", "verify", "Exact reset scope is empty; repository and release anchors remain",
|
|
selected_rows=0, preserved_anchors=2,
|
|
historical_reference_count=historical_delivery_reference_count(rows, scope,
|
|
historical_targets(scope)))
|
|
return 0
|
|
|
|
if args.phase == "apply":
|
|
with engine.connect().execution_options(isolation_level="REPEATABLE READ") as connection:
|
|
with connection.begin():
|
|
connection.execute(text("SET TRANSACTION READ ONLY"))
|
|
connection.execute(text("SET LOCAL statement_timeout = '120s'"))
|
|
identity = dict(connection.execute(text(
|
|
"SELECT current_database() AS database, inet_server_addr() AS address, "
|
|
"inet_server_port() AS port, "
|
|
"(SELECT oid FROM pg_database WHERE datname=current_database()) AS oid"
|
|
)).mappings().one())
|
|
require(identity["database"] == scope["database_name"],
|
|
"LAB_RESET_DATABASE_IDENTITY_CONFLICT")
|
|
names = inspect(connection).get_table_names(schema="public")
|
|
table_scope = {"database_tables": [{"table": name} for name in names],
|
|
"protected_tables": {}}
|
|
_, keys, _, rows = database.reflect(connection, table_scope)
|
|
if already_empty(scope, rows, keys):
|
|
count = historical_delivery_reference_count(rows, scope, historical_targets(scope))
|
|
_emit("REFLECT", "apply", "Exact reset scope is empty; read-only no-op verified",
|
|
selected_rows=0, preserved_anchors=2, historical_reference_count=count)
|
|
return 0
|
|
_emit("REFLECT", "apply", "Read-only preflight found scoped rows; locked apply remains gated",
|
|
classification="scoped_references_remain",
|
|
residual_rows_by_table=residual_reference_counts(scope, rows))
|
|
|
|
if args.phase == "prepare":
|
|
with engine.connect() as connection:
|
|
identity = dict(connection.execute(text(
|
|
"SELECT current_database() AS database, inet_server_addr() AS address, "
|
|
"inet_server_port() AS port, "
|
|
"(SELECT oid FROM pg_database WHERE datname=current_database()) AS oid"
|
|
)).mappings().one())
|
|
require(identity["database"] == scope["database_name"], "LAB_RESET_DATABASE_IDENTITY_CONFLICT")
|
|
names = inspect(connection).get_table_names(schema="public")
|
|
table_scope = {"database_tables": [{"table": name} for name in names],
|
|
"protected_tables": {}}
|
|
tables, keys, _, rows = database.reflect(connection, table_scope)
|
|
if already_empty(scope, rows, keys):
|
|
_emit("REFLECT", "prepare", "Exact reset scope is already empty; preserved anchors remain",
|
|
selected_rows=0, preserved_anchors=2,
|
|
historical_reference_count=historical_delivery_reference_count(
|
|
rows, scope, historical_targets(scope)))
|
|
return 0
|
|
|
|
with engine.connect().execution_options(isolation_level="READ COMMITTED") as connection:
|
|
with connection.begin():
|
|
args.operation_stage = "set_transaction_deadlines"
|
|
connection.execute(text("SET LOCAL lock_timeout = '5s'"))
|
|
connection.execute(text("SET LOCAL statement_timeout = '120s'"))
|
|
args.operation_stage = "verify_database_identity"
|
|
identity = dict(connection.execute(text(
|
|
"SELECT current_database() AS database, inet_server_addr() AS address, "
|
|
"inet_server_port() AS port, "
|
|
"(SELECT oid FROM pg_database WHERE datname=current_database()) AS oid"
|
|
)).mappings().one())
|
|
require(identity["database"] == scope["database_name"], "LAB_RESET_DATABASE_IDENTITY_CONFLICT")
|
|
names = sorted(inspect(connection).get_table_names(schema="public"))
|
|
table_scope = {"database_tables": [{"table": name} for name in names],
|
|
"protected_tables": {}}
|
|
args.operation_stage = "lock_database_tables"
|
|
database.lock_tables(connection, table_scope)
|
|
identity["address"] = str(identity["address"])
|
|
args.operation_stage = "rebuild_locked_manifest"
|
|
manifest, state = snapshot_chart8(connection, table_scope, scope, identity,
|
|
for_apply=args.phase == "apply")
|
|
if state is None:
|
|
_emit("REFLECT", args.phase, "Exact reset scope is already empty; preserved anchors remain",
|
|
selected_rows=0, preserved_anchors=2,
|
|
historical_reference_count=historical_delivery_reference_count(
|
|
rows, scope, historical_targets(scope)))
|
|
return 0
|
|
|
|
manifest_raw = json_bytes(manifest)
|
|
if args.phase == "prepare":
|
|
args.operation_stage = "archive_locked_manifest"
|
|
previous_path = Path(scope["candidate_manifest_path"])
|
|
previous_raw = compare_previous_manifest(previous_path, scope, manifest)
|
|
archive_dir, manifest_sha, rows_sha = _prepare_archive(
|
|
manifest, state, previous_path, previous_raw)
|
|
_emit("REFLECT", "prepare", "Exact locked row closure archived and read back",
|
|
selected_rows=manifest["total_rows"], selected_tables=manifest["total_tables"],
|
|
foreign_key_edges=manifest["foreign_key_edges"], archive=str(archive_dir),
|
|
manifest_sha256=manifest_sha, selected_rows_sha256=rows_sha)
|
|
return 0
|
|
|
|
args.operation_stage = "compare_locked_manifest"
|
|
require(manifest_raw == reviewed_raw
|
|
and manifest["database_identity_sha256"] == reviewed["database_identity_sha256"],
|
|
"LAB_RESET_LOCKED_MANIFEST_CHANGED")
|
|
args.operation_stage = "compare_archived_rows"
|
|
require(_archive_matches_current(archived_rows, state), "LAB_RESET_ARCHIVE_ROW_CONTENT_CONFLICT")
|
|
args.operation_stage = "lock_selected_rows"
|
|
database.lock_selected(connection, state)
|
|
args.operation_stage = "temporarily_disable_append_only_triggers"
|
|
with temporarily_disable_append_only_triggers(connection, args):
|
|
args.operation_stage = "delete_exact_reviewed_rows"
|
|
database.delete_rows(connection, state)
|
|
args.operation_stage = "flush_deferred_foreign_key_events"
|
|
connection.execute(text("SET CONSTRAINTS ALL IMMEDIATE"))
|
|
args.operation_stage = "verify_deleted_rows_before_commit"
|
|
verify_deleted_counts(connection, table_scope, manifest, state)
|
|
args.operation_stage = "commit_exact_scoped_reset"
|
|
_emit("REFLECT", "apply", "Exact scoped database transaction committed; post-commit verification begins",
|
|
selected_rows=reviewed["total_rows"], selected_tables=reviewed["total_tables"],
|
|
manifest_sha256=args.manifest_sha256)
|
|
return verify_after_apply(database_url, scope)
|
|
finally:
|
|
engine.dispose()
|
|
# #endregion Tooling.LabReset.Chart8Scope.Run
|
|
|
|
|
|
# #region Tooling.LabReset.Chart8Scope.PostApply [C:2] [TYPE Function]
|
|
def verify_after_apply(database_url, scope):
|
|
from sqlalchemy import create_engine, inspect, text
|
|
from sqlalchemy.pool import NullPool
|
|
|
|
engine = create_engine(database_url, poolclass=NullPool, echo=False, hide_parameters=True)
|
|
try:
|
|
require(engine.dialect.name == "postgresql", "LAB_RESET_POSTGRESQL_REQUIRED")
|
|
with engine.connect() as connection:
|
|
identity = connection.execute(text("SELECT current_database() AS database")).mappings().one()
|
|
require(identity["database"] == scope["database_name"], "LAB_RESET_DATABASE_IDENTITY_CONFLICT")
|
|
names = inspect(connection).get_table_names(schema="public")
|
|
table_scope = {"database_tables": [{"table": name} for name in names],
|
|
"protected_tables": {}}
|
|
_, keys, _, rows = database.reflect(connection, table_scope)
|
|
require(already_empty(scope, rows, keys), "LAB_RESET_POST_COMMIT_VERIFY_FAILED")
|
|
_emit("REFLECT", "post_apply", "Selected rows are absent and preserved anchors remain",
|
|
selected_rows=0, preserved_anchors=2,
|
|
historical_reference_count=historical_delivery_reference_count(
|
|
rows, scope, historical_targets(scope)))
|
|
return 0
|
|
finally:
|
|
engine.dispose()
|
|
# #endregion Tooling.LabReset.Chart8Scope.PostApply
|
|
|
|
|
|
# #region Tooling.LabReset.Chart8Scope.Main [C:3] [TYPE Function]
|
|
def main(argv=None):
|
|
parser = argparse.ArgumentParser(description="Prepare, apply or verify the exact chart-8 laboratory row reset.")
|
|
parser.add_argument("--scope", required=True, type=Path)
|
|
parser.add_argument("--database-url-file", required=True, type=Path)
|
|
parser.add_argument("--phase", choices=("prepare", "apply", "verify"), default="prepare")
|
|
parser.add_argument("--archive-dir", type=Path)
|
|
parser.add_argument("--manifest-sha256")
|
|
parser.add_argument("--rows-sha256")
|
|
args = parser.parse_args(argv)
|
|
try:
|
|
if args.phase == "apply":
|
|
require(args.archive_dir is not None and args.manifest_sha256 and args.rows_sha256,
|
|
"LAB_RESET_REVIEWED_ARCHIVE_REQUIRED")
|
|
else:
|
|
require(args.archive_dir is None and args.manifest_sha256 is None and args.rows_sha256 is None,
|
|
"LAB_RESET_APPLY_ARGUMENTS_UNEXPECTED")
|
|
return run_phase(args)
|
|
except Stop as exc:
|
|
_emit("EXPLORE", args.phase, "Laboratory row reset refused; no fallback is attempted",
|
|
error_code=str(exc))
|
|
return 2
|
|
except Exception as exc:
|
|
args.failure_type = type(exc).__name__
|
|
facts = safe_database_error_facts(exc)
|
|
code = "LAB_RESET_OPERATION_FAILED"
|
|
if (args.operation_stage == "delete_exact_reviewed_rows"
|
|
and facts["sqlstate"] == "23514"
|
|
and facts["table"] in {"baseline_catalog_revisions", "baseline_catalog_memberships"}):
|
|
code = "CATALOG_REVISION_IMMUTABLE"
|
|
retained = None
|
|
try:
|
|
retained = _retain_safe_failure_facts(args.archive_dir, args, facts)
|
|
except Exception:
|
|
retained = None
|
|
_emit("EXPLORE", args.phase, "Laboratory row reset stopped with bounded database diagnostics",
|
|
error_code=code, failure_type=type(exc).__name__,
|
|
operation_stage=args.operation_stage, **facts,
|
|
private_facts_file=retained)
|
|
return 2
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|
|
# #endregion Tooling.LabReset.Chart8Scope.Main
|
|
# #endregion Tooling.LabReset.Chart8Scope
|