292 lines
15 KiB
Python
292 lines
15 KiB
Python
# #region Tooling.LabReset.Files [C:5] [TYPE Module] [SEMANTICS maintenance,storage,holds,archive]
|
|
# @BRIEF Exact regular-file preservation, selected-byte archive and SSOT hold reset under the native root lock.
|
|
# @PRE All DB locks/reflection/owner rows already acquired; no function in this module queries DB.
|
|
# @INVARIANT Activation marker, root lock, directories and unmatched files survive; no alternate evidence I/O.
|
|
# @RELATION DEPENDS_ON -> [Storage.NativeResultHold]
|
|
from hashlib import sha256
|
|
import os
|
|
from pathlib import Path
|
|
import stat
|
|
|
|
from dashboard_testing_lab_reset_archive import (
|
|
decode, digest, encode, json_bytes, read_bytes, read_json, require, safe_path, sync_directory, write_bytes,
|
|
)
|
|
from dashboard_testing_lab_reset_db import node
|
|
|
|
ROOTS = ("/app/backend/data/dashboard_testing", "/app/backend/git_repos", "/app/storage")
|
|
MARKER = "/app/storage/.native-result-holds/activated"
|
|
INDEX = "/app/storage/.native-result-holds/index.json"
|
|
LOCK = "/app/storage/.native-result-mutation.catalog.lock"
|
|
PACK = "selected-files.json"
|
|
|
|
|
|
# #region Tooling.LabReset.Files.Confined [C:2] [TYPE Function]
|
|
def confined(root, path):
|
|
path = Path(path)
|
|
require(path.is_absolute() and path != Path(root), "LAB_RESET_PATH_INVALID")
|
|
try:
|
|
relative = path.relative_to(root)
|
|
except ValueError:
|
|
require(False, "LAB_RESET_PATH_INVALID")
|
|
require(str(Path(root) / relative) == str(path), "LAB_RESET_PATH_INVALID")
|
|
return safe_path(root, str(relative))
|
|
# #endregion Tooling.LabReset.Files.Confined
|
|
|
|
|
|
# #region Tooling.LabReset.Files.Member [C:3] [TYPE Function]
|
|
def member(root, expected):
|
|
path = confined(root, expected["path"])
|
|
info = path.lstat()
|
|
require(stat.S_ISREG(info.st_mode) and info.st_size == expected["bytes"],
|
|
"LAB_RESET_FILE_CHANGED")
|
|
raw = read_bytes(path)
|
|
require(len(raw) == expected["bytes"] and sha256(raw).hexdigest() == expected["sha256"],
|
|
"LAB_RESET_FILE_CHANGED")
|
|
return raw
|
|
# #endregion Tooling.LabReset.Files.Member
|
|
|
|
|
|
# #region Tooling.LabReset.Files.Scan [C:3] [TYPE Function]
|
|
def scan(root):
|
|
require(Path(root).is_dir() and not Path(root).is_symlink(), "LAB_RESET_ROOT_INVALID")
|
|
regular, directories = [], []
|
|
for home, folders, names in os.walk(root, followlinks=False):
|
|
for name in folders + names:
|
|
path = Path(home) / name
|
|
mode = path.lstat().st_mode
|
|
require(not stat.S_ISLNK(mode), "LAB_RESET_SYMLINK_REFUSED")
|
|
require(stat.S_ISREG(mode) or stat.S_ISDIR(mode), "LAB_RESET_SPECIAL_FILE_REFUSED")
|
|
(directories if stat.S_ISDIR(mode) else regular).append(str(path))
|
|
return sorted(regular), sorted(directories)
|
|
# #endregion Tooling.LabReset.Files.Scan
|
|
|
|
|
|
# #region Tooling.LabReset.Files.Checkpoint [C:3] [TYPE Function]
|
|
def checkpoint(scope):
|
|
require(set(scope["files"]) == set(ROOTS), "LAB_RESET_ROOT_INVALID")
|
|
state = {}
|
|
for root, manifest in sorted(scope["files"].items()):
|
|
members = manifest["selected"] + manifest["preserve"]
|
|
regular, directories = scan(root)
|
|
require(regular == sorted(item["path"] for item in members), "LAB_RESET_FILE_SET_CHANGED")
|
|
for item in members:
|
|
member(root, item)
|
|
state[root] = {"selected": manifest["selected"], "preserve": manifest["preserve"],
|
|
"directories_sha256": digest(directories), "regular_count": len(regular),
|
|
"file_metadata_sha256": digest([[path, file_metadata(Path(path))] for path in regular]),
|
|
"preserved_metadata": {item["path"]: file_metadata(Path(item["path"])) for item in members
|
|
if item in manifest["preserve"] or item["path"] == MARKER}}
|
|
return state
|
|
# #endregion Tooling.LabReset.Files.Checkpoint
|
|
|
|
|
|
# #region Tooling.LabReset.Files.Metadata [C:1] [TYPE Function]
|
|
def file_metadata(path):
|
|
info = path.lstat()
|
|
return {"mode": stat.S_IMODE(info.st_mode), "uid": info.st_uid,
|
|
"gid": info.st_gid, "mtime_ns": info.st_mtime_ns}
|
|
# #endregion Tooling.LabReset.Files.Metadata
|
|
|
|
|
|
# #region Tooling.LabReset.Files.RootLock [C:3] [TYPE Function]
|
|
# @POST Refuse missing coordination files before the existing lock owner could create them.
|
|
def root_lock(scope):
|
|
from src.services.storage_holds import mutation_lock
|
|
keep = scope["files"]["/app/storage"]["preserve"]
|
|
matches = [item for item in keep if item["path"] == LOCK]
|
|
require(len(matches) == 1, "LAB_RESET_ROOT_LOCK_REQUIRED")
|
|
member("/app/storage", matches[0])
|
|
require(Path(LOCK).stat().st_size == 0, "LAB_RESET_ROOT_LOCK_REQUIRED")
|
|
for path in (MARKER, INDEX):
|
|
require(Path(path).is_file() and not Path(path).is_symlink(), "LAB_RESET_HOLD_STATE_REQUIRED")
|
|
return mutation_lock("/app/storage")
|
|
# #endregion Tooling.LabReset.Files.RootLock
|
|
|
|
|
|
# #region Tooling.LabReset.Files.SelectedOwner [C:3] [TYPE Function]
|
|
def selected_owner(state, table, identity):
|
|
_, keys, rows, selected, _ = state
|
|
matches = [row for row in rows[table] if row["id"] == identity]
|
|
require(len(matches) == 1 and node(table, matches[0], keys[table]) in selected,
|
|
"LAB_RESET_HOLD_OWNER_CONFLICT")
|
|
return matches[0]
|
|
# #endregion Tooling.LabReset.Files.SelectedOwner
|
|
|
|
|
|
# #region Tooling.LabReset.Files.HoldOwner [C:4] [TYPE Function]
|
|
# @POST Hold identities name exact selected same-run DB rows and an exact issued root record; no new authority is manufactured.
|
|
def hold_owner(hold, state):
|
|
run = selected_owner(state, "agent_runs", hold.agent_run_id)
|
|
capture = selected_owner(state, "draft_artifacts", hold.capture_artifact_id)
|
|
candidate = selected_owner(state, "draft_artifacts", hold.candidate_id)
|
|
gate = selected_owner(state, "approval_gates", hold.gate_id)
|
|
require(all(row["run_id"] == hold.agent_run_id for row in (capture, candidate, gate)),
|
|
"LAB_RESET_HOLD_OWNER_CONFLICT")
|
|
context = run["context_snapshot"]
|
|
require(isinstance(context, dict) and isinstance(context.get("result_authority"), dict),
|
|
"LAB_RESET_HOLD_OWNER_CONFLICT")
|
|
records = context["result_authority"].get("records")
|
|
require(isinstance(records, list), "LAB_RESET_HOLD_OWNER_CONFLICT")
|
|
matches = [record for record in records if isinstance(record, dict)
|
|
and record.get("capture_artifact_id") == hold.capture_artifact_id]
|
|
require(len(matches) == 1 and digest(matches[0]) == hold.authority_record_sha256,
|
|
"LAB_RESET_HOLD_OWNER_CONFLICT")
|
|
record = matches[0]
|
|
require(capture["sha256"] == record["source_response_hash"]
|
|
and capture["content_ref"] == record["capture_content_ref"], "LAB_RESET_HOLD_OWNER_CONFLICT")
|
|
fields = hold.model_dump(mode="json", exclude={"hold_id"})
|
|
require(digest(fields) == hold.hold_id, "LAB_RESET_HOLD_OWNER_CONFLICT")
|
|
return record, context, candidate
|
|
# #endregion Tooling.LabReset.Files.HoldOwner
|
|
|
|
|
|
# #region Tooling.LabReset.Files.HoldRefs [C:4] [TYPE Function]
|
|
# @RATIONALE The producer appends fresh refs without deduplication; repeated exact refs retain ordered hold identity.
|
|
# @REJECTED Uniqueness refusal invents an invariant absent from NativeHold and rejects valid visual holds.
|
|
def hold_refs(hold, record, context, candidate, state, journal):
|
|
request = record["descriptor"]["native_request"]
|
|
source = context["reference_source"]
|
|
needed = {(request["content_ref"], request["sha256"], "request"),
|
|
(record["capture_content_ref"], record["source_response_hash"], "response"),
|
|
(source["source_url_ref"], source["source_url_sha256"], "source")}
|
|
meta = candidate["capture_meta"]
|
|
require(isinstance(meta, dict), "LAB_RESET_HOLD_OWNER_CONFLICT")
|
|
if meta.get("kind") == "visual":
|
|
review = selected_owner(state, "draft_artifacts", meta["review_disposition_id"])
|
|
require(review["run_id"] == hold.agent_run_id and review["sha256"] == meta["review_disposition_sha256"],
|
|
"LAB_RESET_HOLD_OWNER_CONFLICT")
|
|
needed |= {(review["content_ref"], review["sha256"], "review"),
|
|
(meta["expected_image_content_ref"], meta["expected_image_sha256"], "image")}
|
|
actual = {(ref.content_ref, ref.sha256, ref.purpose) for ref in hold.refs}
|
|
needed_subset = needed.issubset(actual)
|
|
missing = needed - actual
|
|
journal.event("REFLECT" if needed_subset else "EXPLORE", "Producer-owned hold reference coverage checked",
|
|
code=None if needed_subset else "LAB_RESET_HOLD_OWNER_CONFLICT",
|
|
needed_subset=needed_subset, ref_count=len(hold.refs), distinct_count=len(actual),
|
|
duplicate_count=len(hold.refs) - len(actual),
|
|
missing_purpose_counts={purpose: sum(item[2] == purpose for item in missing)
|
|
for purpose in ("request", "response", "source", "image", "review")})
|
|
require(needed_subset, "LAB_RESET_HOLD_OWNER_CONFLICT")
|
|
# Extra visual wire refs must also be registered selected same-run artifacts.
|
|
for ref, hashed, purpose in actual - needed:
|
|
require(purpose in ("request", "response"), "LAB_RESET_HOLD_OWNER_CONFLICT")
|
|
_, keys, rows, selected, _ = state
|
|
matches = [row for row in rows["draft_artifacts"] if row["run_id"] == hold.agent_run_id
|
|
and row["content_ref"] == ref and row["sha256"] == hashed
|
|
and node("draft_artifacts", row, keys["draft_artifacts"]) in selected]
|
|
require(len(matches) == 1, "LAB_RESET_HOLD_OWNER_CONFLICT")
|
|
# #endregion Tooling.LabReset.Files.HoldRefs
|
|
|
|
|
|
# #region Tooling.LabReset.Files.HeldPath [C:1] [TYPE Function]
|
|
def held_path(ref):
|
|
from src.models.storage import FileCategory
|
|
from src.services.storage_holds import relative_ref
|
|
return Path("/app/storage") / FileCategory.DRAFT.value / relative_ref(ref)
|
|
# #endregion Tooling.LabReset.Files.HeldPath
|
|
|
|
|
|
# #region Tooling.LabReset.Files.Holds [C:3] [TYPE Function]
|
|
def validate_holds(scope, state, journal):
|
|
from src.services.storage_holds import load_index, verify_ref
|
|
index = load_index("/app/storage")
|
|
selected = {item["path"]: item for item in scope["files"]["/app/storage"]["selected"]}
|
|
for hold in index.holds:
|
|
record, context, candidate = hold_owner(hold, state)
|
|
hold_refs(hold, record, context, candidate, state, journal)
|
|
for ref in hold.refs:
|
|
path = str(held_path(ref))
|
|
require(path in selected and selected[path]["sha256"] == ref.sha256,
|
|
"LAB_RESET_UNSELECTED_HOLD_BYTES")
|
|
member("/app/storage", selected[path])
|
|
verify_ref("/app/storage", ref)
|
|
return digest(index.model_dump(mode="json"))
|
|
# #endregion Tooling.LabReset.Files.Holds
|
|
|
|
|
|
# #region Tooling.LabReset.Files.Archive [C:3] [TYPE Function]
|
|
# @POST One exclusive private canonical pack deduplicates exact selected bytes; every member binds the pack digest.
|
|
def archive_files(scope, evidence):
|
|
result = []
|
|
blobs = {}
|
|
for root, manifest in sorted(scope["files"].items()):
|
|
for item in manifest["selected"]:
|
|
raw = member(root, item)
|
|
hashed = item["sha256"]
|
|
require(hashed not in blobs or blobs[hashed] == raw, "LAB_RESET_ARCHIVE_READBACK_FAILED")
|
|
blobs[hashed] = raw
|
|
result.append({**item, "member": hashed, "file_metadata": file_metadata(Path(item["path"]))})
|
|
packed_sha = write_bytes(evidence, PACK, json_bytes(encode(blobs)))
|
|
return [{**item, "pack_sha256": packed_sha} for item in result]
|
|
# #endregion Tooling.LabReset.Files.Archive
|
|
|
|
|
|
# #region Tooling.LabReset.Files.VerifyArchive [C:3] [TYPE Function]
|
|
# @PRE Pack keys, canonical encoding, digest and each original selected member match exactly before any deletion.
|
|
def verify_archive_files(scope, evidence, archived):
|
|
expected = sorted([item for manifest in scope["files"].values() for item in manifest["selected"]],
|
|
key=lambda item: item["path"])
|
|
require([{key: value for key, value in item.items() if key not in ("member", "file_metadata", "pack_sha256")}
|
|
for item in sorted(archived, key=lambda item: item["path"])] == expected,
|
|
"LAB_RESET_ARCHIVE_SCOPE_CONFLICT")
|
|
pack_path = safe_path(evidence, PACK)
|
|
packed_raw = read_bytes(pack_path, private=True)
|
|
packed_sha = sha256(packed_raw).hexdigest()
|
|
blobs = decode(read_json(pack_path, packed_sha, private=True))
|
|
require(type(blobs) is dict and set(blobs) == {item["sha256"] for item in expected}
|
|
and json_bytes(encode(blobs)) == packed_raw, "LAB_RESET_ARCHIVE_SCOPE_CONFLICT")
|
|
roots = {item["path"]: root for root, manifest in scope["files"].items() for item in manifest["selected"]}
|
|
for item in archived:
|
|
require(item["member"] == item["sha256"] and item["pack_sha256"] == packed_sha,
|
|
"LAB_RESET_ARCHIVE_SCOPE_CONFLICT")
|
|
require(item["file_metadata"] == file_metadata(Path(item["path"])), "LAB_RESET_ARCHIVE_SCOPE_CONFLICT")
|
|
raw = blobs[item["member"]]
|
|
require(type(raw) is bytes and len(raw) == item["bytes"] and sha256(raw).hexdigest() == item["sha256"]
|
|
and raw == member(roots[item["path"]], item),
|
|
"LAB_RESET_ARCHIVE_READBACK_FAILED")
|
|
# #endregion Tooling.LabReset.Files.VerifyArchive
|
|
|
|
|
|
# #region Tooling.LabReset.Files.Reset [C:4] [TYPE Function]
|
|
# @PRE DB committed; root lock remains held; every selected byte and hold already archived/read back.
|
|
# @POST Index replaced only via SSOT; exact unlinks after ordinary guard; partial failure remains STOP, never autohealed.
|
|
def reset_files(scope, journal):
|
|
from src.services.storage_holds import HoldIndex, guard_delete, load_index, write_index
|
|
write_index("/app/storage", HoldIndex(schema_version=1, holds=()))
|
|
require(load_index("/app/storage").holds == (), "LAB_RESET_HOLD_RESET_FAILED")
|
|
journal.event("REFLECT", "Native hold index reset through current storage owner", holds=0)
|
|
for root, manifest in sorted(scope["files"].items()):
|
|
for item in manifest["selected"]:
|
|
if item["path"] in (MARKER, INDEX):
|
|
continue
|
|
member(root, item)
|
|
path = confined(root, item["path"])
|
|
if root == "/app/storage":
|
|
guard_delete(root, path)
|
|
path.unlink()
|
|
sync_directory(path.parent)
|
|
journal.event("REFLECT", "Exact selected file removed", member_sha256=item["sha256"])
|
|
# #endregion Tooling.LabReset.Files.Reset
|
|
|
|
|
|
# #region Tooling.LabReset.Files.AfterReset [C:3] [TYPE Function]
|
|
def verify_reset(scope, before):
|
|
from src.services.storage_holds import load_index
|
|
require(load_index("/app/storage").holds == (), "LAB_RESET_HOLD_RESET_FAILED")
|
|
for root, manifest in scope["files"].items():
|
|
expected = [item["path"] for item in manifest["preserve"]]
|
|
if root == "/app/storage":
|
|
expected.extend([MARKER, INDEX])
|
|
actual, directories = scan(root)
|
|
require(actual == sorted(expected) and digest(directories) == before[root]["directories_sha256"],
|
|
"LAB_RESET_POST_FILE_STATE_CONFLICT")
|
|
for item in manifest["preserve"]:
|
|
member(root, item)
|
|
for path, metadata in before[root]["preserved_metadata"].items():
|
|
require(file_metadata(Path(path)) == metadata, "LAB_RESET_POST_FILE_STATE_CONFLICT")
|
|
if root == "/app/storage":
|
|
marker = next(item for item in manifest["selected"] if item["path"] == MARKER)
|
|
member(root, marker)
|
|
# #endregion Tooling.LabReset.Files.AfterReset
|
|
# #endregion Tooling.LabReset.Files
|