#!/usr/bin/env bash set -Eeuo pipefail umask 077 [ "$(id -un)" = "ops" ] || { echo "STOP: запускать на router-ops пользователем ops" exit 1 } STEP="STEP_050M07R04_LOCK_EXACT_EDIT_MAP_AND_REPLACEMENT_BLUEPRINTS" PASS_DECISION="PASS_${STEP}" TOKEN="e94a0859747d7b96f29c7fdafc2d0351ba603bb0a7e9e5a4" PUBLIC_BASE="https://helena-background-beam-harry.trycloudflare.com/r/${TOKEN}" ROOT="/opt/router-ops" STATE_ROOT="${ROOT}/state" PUBROOT="${ROOT}/public/r/${TOKEN}" CANONICAL_STATE="${STATE_ROOT}/current-vm101-canonical.env" REFERENCE_STATE="${STATE_ROOT}/current-vm101-reference.env" CANONICAL_PUBLIC_ROOT="${PUBROOT}/vm101-canonical" SOURCE_PUBLIC_CURRENT="${CANONICAL_PUBLIC_ROOT}/current" BUILD_TS="$(date -u +%Y%m%d-%H%M%S)" CANONICAL_ID="${BUILD_TS}_vm101_m07_canonical_v1a_editmap" PRIVATE_ROOT="${STATE_ROOT}/canonical/vm101" WORKSPACE="${PRIVATE_ROOT}/snapshots/${CANONICAL_ID}" PUBLIC_SNAPSHOT="${CANONICAL_PUBLIC_ROOT}/snapshots/${CANONICAL_ID}" PUBLIC_CURRENT="${CANONICAL_PUBLIC_ROOT}/current" REPORT_SLUG="${BUILD_TS}_step050m07r04_lock_exact_edit_map_and_replacement_blueprints" REPORT_DIR="${PUBROOT}/${REPORT_SLUG}" TRYCF_REPORT="${PUBLIC_BASE}/${REPORT_SLUG}/" REPORT_TXT="${TRYCF_REPORT}report.txt" FACTS_JSON="${TRYCF_REPORT}facts.json" ARCHITECTURE_PLAN="${PUBLIC_BASE}/20260711-181158_local_architecture_plan_vm101_autonomous_hmn_recovery/" XS_MAP="${PUBLIC_BASE}/20260711-120734_xs_map_audit_repair_publish/" GLOBAL_PROJECT_PLAN="${PUBLIC_BASE}/20260711-123348_global_project_plan_wg_paid/" VM101_REFERENCE="${PUBLIC_BASE}/vm101-reference/current/" VM101_CANONICAL="${PUBLIC_BASE}/vm101-canonical/current/" VM101_CANONICAL_SNAPSHOT="${PUBLIC_BASE}/vm101-canonical/snapshots/${CANONICAL_ID}/" CURRENT_STAGE="initialization" LAST_SUCCESS="step_saved" SOURCE_CANONICAL_ID="UNRESOLVED" SOURCE_WORKSPACE="UNRESOLVED" REFERENCE_ID="UNRESOLVED" SOURCE_REFERENCE_ID="UNRESOLVED" TARGET_COUNT=0 LOCKED_TARGETS=0 EDIT_READY_TARGETS=0 EDIT_REGIONS=0 PUBLIC_EXCERPTS=0 SECRET_FINDINGS=0 SYNTAX_FAILURES=0 VM101_CONTACTED=false VM101_MODIFIED=false PUBLIC_SNAPSHOT_PUBLISHED=false PUBLIC_CURRENT_UPDATED=false mkdir -p "$REPORT_DIR" cp -a "$0" "$REPORT_DIR/step.sh" chmod 600 "$REPORT_DIR/step.sh" PROGRESS_LOG="$REPORT_DIR/progress.log" : > "$PROGRESS_LOG" stage() { CURRENT_STAGE="$1" echo echo ">>> [$1] $2" | tee -a "$PROGRESS_LOG" date -u '+ utc=%Y-%m-%dT%H:%M:%SZ' | tee -a "$PROGRESS_LOG" } mark_success() { LAST_SUCCESS="$1" echo "last_success=$LAST_SUCCESS" >> "$PROGRESS_LOG" } print_links() { echo echo "TRYCF_REPORT=$TRYCF_REPORT" echo "REPORT_TXT=$REPORT_TXT" echo "FACTS_JSON=$FACTS_JSON" echo "ARCHITECTURE_PLAN=$ARCHITECTURE_PLAN" echo "XS_MAP=$XS_MAP" echo "GLOBAL_PROJECT_PLAN=$GLOBAL_PROJECT_PLAN" echo "VM101_REFERENCE=$VM101_REFERENCE" echo "VM101_CANONICAL=$VM101_CANONICAL" } create_index() { cat > "$REPORT_DIR/index.html" < ${STEP}

${STEP}

Canonical

EOF } write_stop() { local reason="$1" local rc="$2" local line="$3" cat > "$REPORT_DIR/report.txt" < "$REPORT_DIR/facts.json" <<'PY' import json import sys ( step, reason, rc, line, stage, last_success, source_canonical_id, source_workspace, reference_id, source_reference_id, target_count, locked_targets, edit_ready_targets, edit_regions, public_excerpts, secret_findings, syntax_failures, snapshot_published, current_updated, report, report_txt, facts_json, architecture, xs_map, global_plan, vm101_reference, vm101_canonical, ) = sys.argv[1:] print(json.dumps({ "schema": "router-step-facts-v1", "step": step, "assessment": { "decision": f"STOP_{step}_{reason}", "step_execution": "STOP", "operation_result": "EDIT_MAP_NOT_PUBLISHED", "production_health": "UNCHANGED", "milestone_status": "M07_IN_PROGRESS", "all_ok": False, "failure": { "reason": reason, "rc": int(rc), "line": int(line), "stage": stage, "last_success": last_success, }, }, "source": { "source_canonical_id": source_canonical_id, "source_workspace": source_workspace, "reference_id": reference_id, "source_reference_id": source_reference_id, }, "counts": { "target_count": int(target_count), "locked_targets": int(locked_targets), "edit_ready_targets": int(edit_ready_targets), "edit_regions": int(edit_regions), "public_excerpts": int(public_excerpts), "secret_findings": int(secret_findings), "syntax_failures": int(syntax_failures), }, "safety": { "local_only": True, "vm101_contacted": False, "vm101_modified": False, "public_snapshot_published": snapshot_published == "true", "public_current_updated": current_updated == "true", "network_changed": False, "services_changed": False, "state_changed": False, "refresh_ran": False, "rebalance_ran": False, }, "publish": { "trycf_report": report, "report_txt": report_txt, "facts_json": facts_json, "architecture_plan": architecture, "xs_map": xs_map, "global_project_plan": global_plan, "vm101_reference": vm101_reference, "vm101_canonical": vm101_canonical, }, }, ensure_ascii=False, indent=2)) PY create_index find "$REPORT_DIR" \ -type f \ ! -name SHA256SUMS \ -print0 | sort -z | xargs -0 sha256sum \ > "$REPORT_DIR/SHA256SUMS" print_links exit "$rc" } fatal() { local reason="$1" local rc="${2:-1}" local line="${3:-$LINENO}" trap - ERR write_stop "$reason" "$rc" "$line" } trap 'rc=$?; fatal "UNEXPECTED_ERROR" "$rc" "$LINENO"' ERR stage "01/08" "Проверяю current canonical и private workspace" for required in \ "$CANONICAL_STATE" \ "$REFERENCE_STATE" \ "$SOURCE_PUBLIC_CURRENT/canonical-manifest.json" \ "$SOURCE_PUBLIC_CURRENT/contracts/source-analysis.json" \ "$SOURCE_PUBLIC_CURRENT/contracts/requirements.json" \ "$SOURCE_PUBLIC_CURRENT/SHA256SUMS" do [ -s "$required" ] || { echo "MISSING_REQUIRED=$required" fatal "CURRENT_CANONICAL_INCOMPLETE" 2 "$LINENO" } done # shellcheck disable=SC1090 . "$CANONICAL_STATE" SOURCE_CANONICAL_ID="$VM101_CANONICAL_ID" SOURCE_WORKSPACE="$VM101_CANONICAL_PRIVATE_WORKSPACE" REFERENCE_ID="$VM101_CANONICAL_REFERENCE_ID" SOURCE_REFERENCE_ID="$VM101_CANONICAL_SOURCE_REFERENCE_ID" for required in \ "$SOURCE_WORKSPACE/targets.tsv" \ "$SOURCE_WORKSPACE/contracts/source-locks.tsv" \ "$SOURCE_WORKSPACE/contracts/source-analysis.json" \ "$SOURCE_WORKSPACE/contracts/requirements.json" \ "$SOURCE_WORKSPACE/canonical-manifest-private.json" \ "$SOURCE_WORKSPACE/SHA256SUMS" do [ -s "$required" ] || { echo "MISSING_PRIVATE_COMPONENT=$required" fatal "PRIVATE_CANONICAL_INCOMPLETE" 3 "$LINENO" } done ( cd "$SOURCE_PUBLIC_CURRENT" sha256sum -c SHA256SUMS ) > "$REPORT_DIR/source-public-sha256.txt" 2>&1 || fatal "SOURCE_PUBLIC_SHA256_FAILED" 4 "$LINENO" ( cd "$SOURCE_WORKSPACE" sha256sum -c SHA256SUMS ) > "$REPORT_DIR/source-private-sha256.txt" 2>&1 || fatal "SOURCE_PRIVATE_SHA256_FAILED" 5 "$LINENO" mark_success "source_canonical_verified" stage "02/08" "Клонирую immutable private workspace" [ ! -e "$WORKSPACE" ] || fatal "PRIVATE_WORKSPACE_ALREADY_EXISTS" 6 "$LINENO" mkdir -p "$(dirname "$WORKSPACE")" cp -a "$SOURCE_WORKSPACE" "$WORKSPACE" rm -f "$WORKSPACE/SHA256SUMS" mkdir -p \ "$WORKSPACE/contracts" \ "$WORKSPACE/docs/source-excerpts" \ "$WORKSPACE/tests" mark_success "private_workspace_cloned" stage "03/08" "Проверяю source locks и неизменённый candidate baseline" python3 - \ "$WORKSPACE" \ "$WORKSPACE/contracts/source-lock-validation.json" <<'PY' import hashlib import json import sys from pathlib import Path workspace = Path(sys.argv[1]) output = Path(sys.argv[2]) def sha256(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as source: for chunk in iter( lambda: source.read(1024 * 1024), b"", ): digest.update(chunk) return digest.hexdigest() locks_path = workspace / "contracts/source-locks.tsv" analysis_path = workspace / "contracts/source-analysis.json" analysis = json.loads( analysis_path.read_text(encoding="utf-8") ) analysis_by_label = { item["label"]: item for item in analysis["targets"] } rows = [] for number, line in enumerate( locks_path.read_text( encoding="utf-8" ).splitlines() ): if number == 0: continue ( label, source_path, source_sha256, size_bytes, sensitive, ) = line.split("\t") relative = source_path.lstrip("/") source_file = workspace / "source" / relative candidate_file = workspace / "candidate" / relative source_present = source_file.is_file() candidate_present = candidate_file.is_file() source_actual = ( sha256(source_file) if source_present else None ) candidate_actual = ( sha256(candidate_file) if candidate_present else None ) analysis_hash = ( analysis_by_label.get(label, {}).get("sha256") ) checks = { "source_present": source_present, "candidate_present": candidate_present, "source_matches_lock": source_actual == source_sha256, "candidate_matches_source": candidate_actual == source_sha256, "analysis_matches_source": analysis_hash == source_sha256, "size_matches": source_present and source_file.stat().st_size == int(size_bytes), } rows.append({ "label": label, "source_path": source_path, "source_sha256": source_sha256, "sensitive": sensitive == "true", "checks": checks, "passed": all(checks.values()), }) result = { "schema": "vm101-source-lock-validation-v1", "passed": all(item["passed"] for item in rows), "target_count": len(rows), "locked_target_count": sum(1 for item in rows if item["passed"]), "targets": rows, } output.write_text( json.dumps( result, ensure_ascii=False, indent=2, ) + "\n", encoding="utf-8", ) if not result["passed"]: raise SystemExit(20) PY TARGET_COUNT="$( python3 - "$WORKSPACE/contracts/source-lock-validation.json" <<'PY' import json import sys data = json.load(open(sys.argv[1], encoding="utf-8")) print(data["target_count"]) PY )" LOCKED_TARGETS="$( python3 - "$WORKSPACE/contracts/source-lock-validation.json" <<'PY' import json import sys data = json.load(open(sys.argv[1], encoding="utf-8")) print(data["locked_target_count"]) PY )" [ "$TARGET_COUNT" -eq 9 ] || fatal "TARGET_COUNT_INVALID" 7 "$LINENO" [ "$LOCKED_TARGETS" -eq "$TARGET_COUNT" ] || fatal "SOURCE_LOCK_VALIDATION_FAILED" 8 "$LINENO" mark_success "source_locks_validated" stage "04/08" "Создаю exact private edit map и sanitized excerpts" python3 - \ "$WORKSPACE" \ "$CANONICAL_ID" \ "$SOURCE_CANONICAL_ID" \ "$REFERENCE_ID" \ "$SOURCE_REFERENCE_ID" <<'PY' import hashlib import json import re import sys from pathlib import Path ( workspace_arg, canonical_id, source_canonical_id, reference_id, source_reference_id, ) = sys.argv[1:] workspace = Path(workspace_arg) analysis_path = workspace / "contracts/source-analysis.json" analysis = json.loads( analysis_path.read_text(encoding="utf-8") ) excerpt_root = workspace / "docs/source-excerpts" excerpt_root.mkdir(parents=True, exist_ok=True) def sha256_text(value: str) -> str: return hashlib.sha256( value.encode("utf-8") ).hexdigest() def merge_ranges(ranges): cleaned = sorted( (max(1, start), max(1, end)) for start, end in ranges ) merged = [] for start, end in cleaned: if not merged or start > merged[-1][1] + 1: merged.append([start, end]) else: merged[-1][1] = max( merged[-1][1], end, ) return merged def sanitize_line(line: str) -> tuple[str, int]: original = line redactions = 0 shell = re.match( r"^(\s*(?:export\s+)?)" r"([A-Za-z_][A-Za-z0-9_]*)" r"(\s*=\s*)" r"(.*)$", line, ) if shell: name = shell.group(2).lower() sensitive = ( name in { "password", "passwd", "token", "secret", "private_key", "preshared_key", "access_code", "hmn_code", "api_key", "authorization", } or name.endswith(( "_password", "_passwd", "_token", "_secret", "_private_key", "_preshared_key", "_access_code", "_api_key", "_client_secret", )) ) if sensitive: line = ( f"{shell.group(1)}" f"{shell.group(2)}" f"{shell.group(3)}" "'REDACTED'" ) redactions += 1 uci = re.match( r"^(\s*option\s+)" r"([A-Za-z0-9_.-]+)" r"(\s+)(.*)$", line, re.IGNORECASE, ) if uci and uci.group(2).lower() in { "private_key", "preshared_key", "password", "token", "secret", "access_code", }: line = ( f"{uci.group(1)}" f"{uci.group(2)}" f"{uci.group(3)}" "'REDACTED'" ) redactions += 1 line, count = re.subn( r"(Authorization\s*:\s*)" r"(?:Bearer|Basic)\s+\S+", r"\1REDACTED", line, flags=re.IGNORECASE, ) redactions += count line, count = re.subn( r"([A-Za-z0-9+/]{40,}={0,2})", "REDACTED_BLOB", line, ) redactions += count line, count = re.subn( r"\b[0-9a-fA-F]{48,}\b", "REDACTED_HEX", line, ) redactions += count if line != original and redactions == 0: redactions = 1 return line, redactions rules = { "refresh_pool_safe": { "purpose": "storage preflight, backup manifest and rollback gate", "tokens": [ "table_200", "tmp_backup_path", ], "functions": [ "restore_backup", "run_validate_current_pool", "run_body", ], "assignments": [ "BACKUPDIR", "BACK", "LOCK", ], "required_transformations": [ "calculate required backup bytes before mutation", "refuse operation when temporary storage is insufficient", "create surgical backup manifest", "validate every rollback artifact before download", "refuse rollback from incomplete backup", ], }, "code_test": { "purpose": "replace table-200-only active interface detection", "tokens": [ "table_200", "active_interface_error", ], "functions": [], "assignments": ["ACTIVE"], "required_transformations": [ "source canonical runtime library", "select first strict healthy vpn1-vpn5", "treat table 200 only as optional preferred candidate", ], }, "download_all_awg": { "purpose": "replace table-200-only bootstrap detection", "tokens": [ "table_200", "active_interface_error", ], "functions": [], "assignments": ["ACTIVE"], "required_transformations": [ "source canonical runtime library", "select first strict healthy vpn1-vpn5", "retain failure when no healthy slot exists", ], }, "refresh_awg": { "purpose": "replace active interface and table-200 contract", "tokens": [ "table_200", "active_interface_error", "hmn_endpoint", "amneziawg", ], "functions": [ "get_table_default_dev", ], "assignments": [ "ACTIVE", "TABLE_DEV", ], "required_transformations": [ "source canonical runtime library", "use vm101_healthy_bootstrap_iface", "keep table 200 informational only", ], }, "validate_current_pool": { "purpose": "healthy bootstrap selection and runtime endpoint lookup", "tokens": [ "table_200", "hmn_endpoint", ], "functions": [], "assignments": ["ACTIVE"], "required_transformations": [ "source canonical runtime library", "select a strict healthy bootstrap slot", "use runtime endpoint before UCI fallback", ], }, "planner": { "purpose": "runtime endpoint as current source of truth", "tokens": ["hmn_endpoint"], "functions": [], "assignments": ["CURRENT"], "required_transformations": [ "source canonical runtime library", "read current endpoint from amneziawg runtime", "use UCI hmn_endpoint only as explicit fallback", ], }, "rebalance_apply": { "purpose": "runtime current, strict retry and shell/JSON exit agreement", "tokens": [ "hmn_endpoint", "commit_failed", "apply_ok", "strict_ping", ], "functions": ["strict_ping"], "assignments": [], "required_transformations": [ "source canonical runtime library", "read current endpoint from runtime", "retry strict health after ifup", "return nonzero for commit_failed", "return nonzero for commit refusal", "keep JSON decision consistent with shell status", ], }, "emergency_runner": { "purpose": "validate rebalance shell status and JSON contract", "tokens": ["rebalance_rc"], "functions": [ "emit_json", ], "assignments": [], "required_transformations": [ "capture complete rebalance output", "require shell rc zero", "require JSON decision commit_ok or noop", "require JSON apply_ok true", "convert false-success JSON to nonzero result", ], }, "slots_apply": { "purpose": "make table 200 optional", "tokens": ["table_200"], "functions": [], "assignments": [], "required_transformations": [ "remove legacy table-200 hard gate", "retain authoritative validation for tables 201-205", ], }, } private_targets = [] public_targets = [] total_regions = 0 ready_count = 0 public_excerpt_count = 0 total_redactions = 0 for item in analysis["targets"]: label = item["label"] if label not in rules: continue rule = rules[label] source_path = item["source_path"] source_file = ( workspace / "source" / source_path.lstrip("/") ) lines = source_file.read_text( encoding="utf-8", errors="strict", ).splitlines() anchors = [] for token in rule["tokens"]: for line_number in item["token_lines"].get( token, [], ): anchors.append({ "type": "token", "name": token, "line": line_number, }) for function_name in rule["functions"]: for function in item["functions"]: if function["name"] == function_name: anchors.append({ "type": "function", "name": function_name, "line": function["line"], }) for assignment_name in rule["assignments"]: for assignment in item["uppercase_assignments"]: if assignment["name"] == assignment_name: anchors.append({ "type": "assignment", "name": assignment_name, "line": assignment["line"], }) anchor_lines = sorted({ anchor["line"] for anchor in anchors }) ranges = merge_ranges([ ( line_number - 8, line_number + 10, ) for line_number in anchor_lines ]) regions = [] public_regions = [] excerpt_lines = [] public_excerpt_lines = [] for index, (start, end) in enumerate( ranges, start=1, ): end = min(end, len(lines)) exact_lines = [ { "line": number, "text": lines[number - 1], } for number in range(start, end + 1) ] exact_text = "\n".join( f"{entry['line']:04d}: {entry['text']}" for entry in exact_lines ) + "\n" regions.append({ "region_id": f"{label}-r{index}", "start_line": start, "end_line": end, "exact_sha256": sha256_text(exact_text), "lines": exact_lines, }) sanitized_lines = [] region_redactions = 0 for entry in exact_lines: sanitized, redactions = sanitize_line( entry["text"] ) region_redactions += redactions sanitized_lines.append({ "line": entry["line"], "text": sanitized, }) sanitized_text = "\n".join( f"{entry['line']:04d}: {entry['text']}" for entry in sanitized_lines ) + "\n" public_regions.append({ "region_id": f"{label}-r{index}", "start_line": start, "end_line": end, "sanitized_sha256": sha256_text(sanitized_text), "redactions": region_redactions, }) excerpt_lines.extend([ f"=== {label}-r{index} " f"lines {start}-{end} ===", exact_text.rstrip(), "", ]) public_excerpt_lines.extend([ f"=== {label}-r{index} " f"lines {start}-{end} ===", sanitized_text.rstrip(), "", ]) total_redactions += region_redactions exact_excerpt = "\n".join( excerpt_lines ).rstrip() + "\n" public_excerpt = "\n".join( public_excerpt_lines ).rstrip() + "\n" private_excerpt_path = ( workspace / "docs/source-excerpts" / f"{label}.private.txt" ) public_excerpt_path = ( workspace / "docs/source-excerpts" / f"{label}.public.txt" ) private_excerpt_path.write_text( exact_excerpt, encoding="utf-8", ) public_excerpt_path.write_text( public_excerpt, encoding="utf-8", ) ready = ( bool(anchors) and bool(regions) and len( rule["required_transformations"] ) > 0 ) if ready: ready_count += 1 total_regions += len(regions) public_excerpt_count += 1 private_targets.append({ "label": label, "source_path": source_path, "source_sha256": item["sha256"], "source_line_count": item["line_count"], "purpose": rule["purpose"], "anchors": anchors, "regions": regions, "required_transformations": rule["required_transformations"], "edit_ready": ready, "private_excerpt": str( private_excerpt_path.relative_to( workspace ) ), "public_excerpt": str( public_excerpt_path.relative_to( workspace ) ), }) public_targets.append({ "label": label, "source_path": source_path, "source_sha256": ( "PRIVATE_ONLY" if item.get("sensitive") else item["sha256"] ), "source_line_count": item["line_count"], "purpose": rule["purpose"], "anchors": anchors, "regions": public_regions, "required_transformations": rule["required_transformations"], "edit_ready": ready, "excerpt": f"docs/source-excerpts/{label}.txt", }) private_map = { "schema": "vm101-canonical-private-edit-map-v1", "canonical_id": canonical_id, "source_canonical_id": source_canonical_id, "reference_id": reference_id, "source_reference_id": source_reference_id, "target_count": len(private_targets), "edit_ready_target_count": ready_count, "region_count": total_regions, "targets": private_targets, } public_map = { "schema": "vm101-canonical-public-edit-map-v1", "canonical_id": canonical_id, "source_canonical_id": source_canonical_id, "reference_id": reference_id, "source_reference_id": source_reference_id, "target_count": len(public_targets), "edit_ready_target_count": ready_count, "region_count": total_regions, "public_excerpt_count": public_excerpt_count, "sanitization": { "performed": True, "total_redactions": total_redactions, "private_exact_lines_published": False, }, "targets": public_targets, } blueprints = { "schema": "vm101-complete-replacement-blueprints-v1", "canonical_id": canonical_id, "source_canonical_id": source_canonical_id, "status": "EDIT_MAP_LOCKED_REPLACEMENTS_NOT_YET_WRITTEN", "generation_policy": { "source_hash_must_match": True, "source_line_count_must_match": True, "all_anchor_regions_must_match": True, "whole_file_output_required": True, "regex_patch_of_live_file": False, "candidate_syntax_check_required": True, "unified_diff_required": True, "fixture_tests_required": True, "vm101_install_allowed": False, }, "targets": [ { "label": target["label"], "source_path": target["source_path"], "source_sha256": target["source_sha256"], "source_line_count": target["source_line_count"], "required_transformations": target[ "required_transformations" ], "anchor_region_hashes": [ { "region_id": region["region_id"], "start_line": region["start_line"], "end_line": region["end_line"], "exact_sha256": region["exact_sha256"], } for region in target["regions"] ], "output_contract": { "complete_file": True, "destination": target["source_path"], "installation_in_this_step": False, }, } for target in private_targets ], } (workspace / "contracts/edit-map-private.json").write_text( json.dumps( private_map, ensure_ascii=False, indent=2, ) + "\n", encoding="utf-8", ) (workspace / "contracts/edit-map-public.json").write_text( json.dumps( public_map, ensure_ascii=False, indent=2, ) + "\n", encoding="utf-8", ) ( workspace / "contracts/replacement-blueprints.json" ).write_text( json.dumps( blueprints, ensure_ascii=False, indent=2, ) + "\n", encoding="utf-8", ) if ready_count != len(rules): raise SystemExit( f"edit map incomplete: " f"{ready_count}/{len(rules)}" ) PY EDIT_READY_TARGETS="$( python3 - "$WORKSPACE/contracts/edit-map-public.json" <<'PY' import json import sys data = json.load(open(sys.argv[1], encoding="utf-8")) print(data["edit_ready_target_count"]) PY )" EDIT_REGIONS="$( python3 - "$WORKSPACE/contracts/edit-map-public.json" <<'PY' import json import sys data = json.load(open(sys.argv[1], encoding="utf-8")) print(data["region_count"]) PY )" PUBLIC_EXCERPTS="$( python3 - "$WORKSPACE/contracts/edit-map-public.json" <<'PY' import json import sys data = json.load(open(sys.argv[1], encoding="utf-8")) print(data["public_excerpt_count"]) PY )" [ "$EDIT_READY_TARGETS" -eq "$TARGET_COUNT" ] || fatal "EDIT_MAP_INCOMPLETE" 9 "$LINENO" [ "$EDIT_REGIONS" -gt 0 ] || fatal "NO_EDIT_REGIONS" 10 "$LINENO" mark_success "exact_edit_map_created" stage "05/08" "Проверяю candidate syntax и private integrity" : > "$WORKSPACE/tests/syntax-checks-r04.txt" while IFS= read -r file; do relative="${file#${WORKSPACE}/candidate/}" if sh -n "$file" \ >> "$WORKSPACE/tests/syntax-checks-r04.txt" \ 2>&1 then echo "PASS $relative" \ >> "$WORKSPACE/tests/syntax-checks-r04.txt" else echo "FAIL $relative" \ >> "$WORKSPACE/tests/syntax-checks-r04.txt" SYNTAX_FAILURES=$((SYNTAX_FAILURES + 1)) fi done < <( find "$WORKSPACE/candidate" \ -type f | sort ) cp -a \ "$WORKSPACE/tests/syntax-checks-r04.txt" \ "$REPORT_DIR/syntax-checks.txt" [ "$SYNTAX_FAILURES" -eq 0 ] || fatal "CANDIDATE_SYNTAX_FAILED" 11 "$LINENO" python3 - \ "$WORKSPACE/canonical-manifest-private.json" \ "$WORKSPACE/contracts/edit-map-private.json" \ "$WORKSPACE/contracts/replacement-blueprints.json" \ "$CANONICAL_ID" \ "$SOURCE_CANONICAL_ID" <<'PY' import json import sys from pathlib import Path ( manifest_path, edit_map_path, blueprints_path, canonical_id, source_canonical_id, ) = sys.argv[1:] manifest_file = Path(manifest_path) manifest = json.loads( manifest_file.read_text(encoding="utf-8") ) edit_map = json.loads( Path(edit_map_path).read_text( encoding="utf-8" ) ) blueprints = json.loads( Path(blueprints_path).read_text( encoding="utf-8" ) ) manifest["canonical_id"] = canonical_id manifest["source_canonical_id"] = ( source_canonical_id ) manifest["candidate_status"] = ( "EDIT_MAP_LOCKED_REPLACEMENTS_NOT_YET_WRITTEN" ) manifest["edit_map"] = { "target_count": edit_map["target_count"], "edit_ready_target_count": edit_map["edit_ready_target_count"], "region_count": edit_map["region_count"], } manifest["replacement_blueprints"] = { "status": blueprints["status"], "target_count": len(blueprints["targets"]), } manifest_file.write_text( json.dumps( manifest, ensure_ascii=False, indent=2, ) + "\n", encoding="utf-8", ) PY ( cd "$WORKSPACE" find . \ -type f \ ! -name SHA256SUMS \ -print0 | sort -z | xargs -0 sha256sum \ > SHA256SUMS sha256sum -c SHA256SUMS ) > "$REPORT_DIR/private-sha256-check.txt" 2>&1 || fatal "PRIVATE_WORKSPACE_SHA256_FAILED" 12 "$LINENO" mark_success "private_workspace_validated" stage "06/08" "Строю public canonical snapshot с excerpts" PUBLIC_STAGE="${WORKSPACE}/public-summary" rm -rf "$PUBLIC_STAGE" mkdir -p "$PUBLIC_STAGE" cp -a "$SOURCE_PUBLIC_CURRENT/." "$PUBLIC_STAGE/" rm -f "$PUBLIC_STAGE/SHA256SUMS" mkdir -p \ "$PUBLIC_STAGE/contracts" \ "$PUBLIC_STAGE/docs/source-excerpts" \ "$PUBLIC_STAGE/tests" cp -a \ "$WORKSPACE/contracts/edit-map-public.json" \ "$PUBLIC_STAGE/contracts/edit-map.json" cp -a \ "$WORKSPACE/contracts/replacement-blueprints.json" \ "$PUBLIC_STAGE/contracts/replacement-blueprints.json" cp -a \ "$WORKSPACE/tests/syntax-checks-r04.txt" \ "$PUBLIC_STAGE/tests/syntax-checks-r04.txt" for source in \ "$WORKSPACE"/docs/source-excerpts/*.public.txt do label="$( basename "$source" .public.txt )" cp -a \ "$source" \ "$PUBLIC_STAGE/docs/source-excerpts/${label}.txt" done python3 - \ "$PUBLIC_STAGE/canonical-manifest.json" \ "$PUBLIC_STAGE/contracts/edit-map.json" \ "$CANONICAL_ID" \ "$SOURCE_CANONICAL_ID" <<'PY' import json import sys from pathlib import Path ( manifest_path, edit_map_path, canonical_id, source_canonical_id, ) = sys.argv[1:] manifest_file = Path(manifest_path) manifest = json.loads( manifest_file.read_text(encoding="utf-8") ) edit_map = json.loads( Path(edit_map_path).read_text( encoding="utf-8" ) ) manifest["canonical_id"] = canonical_id manifest["source_canonical_id"] = ( source_canonical_id ) manifest["candidate_status"] = ( "EDIT_MAP_LOCKED_REPLACEMENTS_NOT_YET_WRITTEN" ) manifest["edit_map"] = { "target_count": edit_map["target_count"], "edit_ready_target_count": edit_map["edit_ready_target_count"], "region_count": edit_map["region_count"], "public_excerpt_count": edit_map["public_excerpt_count"], "private_exact_lines_published": False, } manifest.setdefault("safety", {}) manifest["safety"][ "private_source_contents_published" ] = False manifest["safety"][ "vm101_contacted" ] = False manifest["safety"][ "vm101_modified" ] = False manifest_file.write_text( json.dumps( manifest, ensure_ascii=False, indent=2, ) + "\n", encoding="utf-8", ) PY cat > "$PUBLIC_STAGE/docs/edit-map-status.md" < "$PUBLIC_STAGE/index.html" < VM101 Canonical ${CANONICAL_ID}

VM101 Canonical

Canonical ID: ${CANONICAL_ID}

Source canonical ID: ${SOURCE_CANONICAL_ID}

VM101 was not contacted or modified.

EOF ( cd "$PUBLIC_STAGE" find . \ -type f \ ! -name SHA256SUMS \ -print0 | sort -z | xargs -0 sha256sum \ > SHA256SUMS sha256sum -c SHA256SUMS ) > "$REPORT_DIR/public-sha256-check.txt" 2>&1 || fatal "PUBLIC_STAGE_SHA256_FAILED" 13 "$LINENO" cp -a \ "$PUBLIC_STAGE/contracts/edit-map.json" \ "$REPORT_DIR/edit-map-public.json" cp -a \ "$PUBLIC_STAGE/contracts/replacement-blueprints.json" \ "$REPORT_DIR/replacement-blueprints.json" mark_success "public_snapshot_staged" stage "07/08" "Публикую immutable canonical snapshot и current" [ ! -e "$PUBLIC_SNAPSHOT" ] || fatal "PUBLIC_SNAPSHOT_ALREADY_EXISTS" 14 "$LINENO" PUBLIC_TMP="${PUBLIC_SNAPSHOT}.new.$$" CURRENT_NEW="${CANONICAL_PUBLIC_ROOT}/current.new.$$" CURRENT_OLD="${CANONICAL_PUBLIC_ROOT}/current.old.$$" mkdir -p "$CANONICAL_PUBLIC_ROOT/snapshots" rm -rf \ "$PUBLIC_TMP" \ "$CURRENT_NEW" \ "$CURRENT_OLD" cp -a "$PUBLIC_STAGE" "$PUBLIC_TMP" chmod -R a+rX "$PUBLIC_TMP" mv "$PUBLIC_TMP" "$PUBLIC_SNAPSHOT" PUBLIC_SNAPSHOT_PUBLISHED=true cp -a "$PUBLIC_SNAPSHOT" "$CURRENT_NEW" if [ -e "$PUBLIC_CURRENT" ]; then mv "$PUBLIC_CURRENT" "$CURRENT_OLD" fi mv "$CURRENT_NEW" "$PUBLIC_CURRENT" rm -rf "$CURRENT_OLD" PUBLIC_CURRENT_UPDATED=true rm -f "${PRIVATE_ROOT}/current" ln -s \ "snapshots/${CANONICAL_ID}" \ "${PRIVATE_ROOT}/current" cat > "$CANONICAL_STATE" < "$REPORT_DIR/current-sha256-check.txt" 2>&1 || fatal "PUBLISHED_CURRENT_SHA256_FAILED" 15 "$LINENO" PRIVATE_MAP_SHA="$( sha256sum "$WORKSPACE/contracts/edit-map-private.json" | cut -d' ' -f1 )" PUBLIC_MAP_SHA="$( sha256sum "$PUBLIC_CURRENT/contracts/edit-map.json" | cut -d' ' -f1 )" BLUEPRINT_SHA="$( sha256sum "$PUBLIC_CURRENT/contracts/replacement-blueprints.json" | cut -d' ' -f1 )" python3 - \ "$STEP" \ "$CANONICAL_ID" \ "$SOURCE_CANONICAL_ID" \ "$REFERENCE_ID" \ "$SOURCE_REFERENCE_ID" \ "$WORKSPACE" \ "$TARGET_COUNT" \ "$LOCKED_TARGETS" \ "$EDIT_READY_TARGETS" \ "$EDIT_REGIONS" \ "$PUBLIC_EXCERPTS" \ "$SECRET_FINDINGS" \ "$SYNTAX_FAILURES" \ "$PRIVATE_MAP_SHA" \ "$PUBLIC_MAP_SHA" \ "$BLUEPRINT_SHA" \ "$VM101_CANONICAL" \ "$VM101_CANONICAL_SNAPSHOT" \ > "$REPORT_DIR/assessment.json" <<'PY' import json import sys ( step, canonical_id, source_canonical_id, reference_id, source_reference_id, workspace, target_count, locked_targets, edit_ready_targets, edit_regions, public_excerpts, secret_findings, syntax_failures, private_map_sha, public_map_sha, blueprint_sha, current_url, snapshot_url, ) = sys.argv[1:] print(json.dumps({ "schema": "vm101-canonical-edit-map-assessment-v1", "decision": f"PASS_{step}", "step_execution": "PASS", "operation_result": "EXACT_EDIT_MAP_AND_REPLACEMENT_BLUEPRINTS_LOCKED", "production_health": "UNCHANGED", "milestone_status": "M07_IN_PROGRESS", "all_ok": True, "canonical": { "canonical_id": canonical_id, "source_canonical_id": source_canonical_id, "reference_id": reference_id, "source_reference_id": source_reference_id, "private_workspace": workspace, "current_url": current_url, "snapshot_url": snapshot_url, }, "counts": { "target_count": int(target_count), "locked_targets": int(locked_targets), "edit_ready_targets": int(edit_ready_targets), "edit_regions": int(edit_regions), "public_excerpts": int(public_excerpts), "secret_findings": int(secret_findings), "syntax_failures": int(syntax_failures), }, "integrity": { "private_edit_map_sha256": private_map_sha, "public_edit_map_sha256": public_map_sha, "replacement_blueprint_sha256": blueprint_sha, "source_hashes_locked": True, "exact_region_hashes_locked": True, }, "candidate_state": { "replacement_files_written": False, "installation_allowed": False, "vm101_installed": False, }, "safety": { "local_only": True, "vm101_contacted": False, "vm101_modified": False, "network_changed": False, "services_changed": False, "state_changed": False, "refresh_ran": False, "rebalance_ran": False, }, "next_step": "GENERATE_COMPLETE_CANONICAL_REPLACEMENT_FILES", }, ensure_ascii=False, indent=2)) PY cat > "$REPORT_DIR/report.txt" < "$REPORT_DIR/facts.json" <<'PY' import json import sys ( assessment_path, step, generated_at, report, report_txt, facts_json, architecture, xs_map, global_plan, vm101_reference, vm101_canonical, ) = sys.argv[1:] assessment = json.load( open(assessment_path, encoding="utf-8") ) print(json.dumps({ "schema": "router-step-facts-v1", "step": step, "generated_at_utc": generated_at, "assessment": assessment, "safety": assessment["safety"], "plan": { "current_milestone": "M07", "milestone_completed": False, }, "next_step": "GENERATE_COMPLETE_CANONICAL_REPLACEMENT_FILES", "publish": { "trycf_report": report, "report_txt": report_txt, "facts_json": facts_json, "architecture_plan": architecture, "xs_map": xs_map, "global_project_plan": global_plan, "vm101_reference": vm101_reference, "vm101_canonical": vm101_canonical, }, }, ensure_ascii=False, indent=2)) PY create_index find "$REPORT_DIR" \ -type f \ ! -name SHA256SUMS \ -print0 | sort -z | xargs -0 sha256sum \ > "$REPORT_DIR/SHA256SUMS" trap - ERR echo "decision=$PASS_DECISION" echo "step_execution=PASS" echo "operation_result=EXACT_EDIT_MAP_AND_REPLACEMENT_BLUEPRINTS_LOCKED" echo "production_health=UNCHANGED" echo "canonical_id=$CANONICAL_ID" echo "source_canonical_id=$SOURCE_CANONICAL_ID" echo "target_count=$TARGET_COUNT" echo "locked_targets=$LOCKED_TARGETS" echo "edit_ready_targets=$EDIT_READY_TARGETS" echo "edit_regions=$EDIT_REGIONS" echo "public_excerpts=$PUBLIC_EXCERPTS" echo "secret_findings=$SECRET_FINDINGS" echo "syntax_failures=$SYNTAX_FAILURES" echo "vm101_contacted=false" echo "vm101_modified=false" print_links