AutoMiningScript

"""Install built AMS exports into an unloaded save through AgentBridge's generic CLI."""

import argparse
from datetime import datetime, timezone
import hashlib
import json
from pathlib import Path
import subprocess
import sys
import uuid

ROOT = Path(__file__).resolve().parents[2]
ROLES = {
    "矿机控制器": "miner", "Miner controller": "miner",
    "舰队主控（母舰）": "fleet", "Fleet controller (mother ship)": "fleet",
    "标点终端": "marker", "Marker terminal": "marker",
    "安全看门狗": "watchdog", "Safety watchdog": "watchdog",
}


def sha(text):
    return hashlib.sha256(text.encode("utf-8")).hexdigest()


def role_of(source):
    header = source.splitlines()[0] if source else ""
    parts = [part.strip() for part in header.split("|")]
    if len(parts) != 3 or parts[0] != "// XFE AMS":
        return None
    if parts[1] not in ROLES:
        raise ValueError("Unrecognized AMS role: " + header)
    return ROLES[parts[1]]


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--cli", type=Path, required=True)
    parser.add_argument("--world", type=Path, required=True)
    parser.add_argument("--language", choices=("zh-CN", "en-US"), default="zh-CN")
    parser.add_argument("--apply", action="store_true", help="Write the original save; without this flag only preview")
    args = parser.parse_args()
    stamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
    evidence = ROOT / "artifacts" / "agent-debug" / ("world-install-" + stamp + "-" + uuid.uuid4().hex[:8])
    evidence.mkdir(parents=True)

    def write(name, result):
        (evidence / name).write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding="utf-8")

    def call(*arguments):
        process = subprocess.run([str(args.cli.resolve()), *map(str, arguments)], capture_output=True, encoding="utf-8", timeout=180)
        result = json.loads(process.stdout)
        if process.returncode:
            write("failure.json", result)
            raise RuntimeError(json.dumps(result, ensure_ascii=False))
        return result

    package = json.loads((ROOT / "dist" / "manifest.json").read_text(encoding="utf-8-sig"))
    if not package.get("logicTestsRun"):
        raise ValueError("Run the full release build before installation")
    exports = {entry["role"]: entry for entry in package["exports"] if entry["language"] == args.language}
    for entry in exports.values():
        path = (ROOT / "dist" / entry["path"]).resolve()
        source = path.read_bytes().decode("utf-8-sig")
        if not entry.get("roleVerified") or sha(source) != entry["sha256"] or role_of(source) != entry["role"]:
            raise ValueError("Build manifest/source mismatch: " + str(path))
        if len(source.encode("utf-16-le")) // 2 != entry["characters"]:
            raise ValueError("Build manifest character count mismatch")
        entry["absolutePath"] = str(path)
    world = str(args.world.resolve())
    write("host.json", call("host", "status"))
    inspection = call("world", "inspect", "--path", world)
    write("before.json", inspection)
    roles_by_hash = {}
    updates = []
    selected = []
    for program in inspection["programs"]:
        current_hash = program["sha256"]
        if current_hash not in roles_by_hash:
            existing = call("world", "read", "--path", world, "--block", program["entityId"])
            if sha(existing["source"]) != current_hash:
                raise ValueError("Save changed while identifying script roles; retry inspection")
            roles_by_hash[current_hash] = role_of(existing["source"])
        role = roles_by_hash[current_hash]
        if role is None:
            continue
        export = exports[role]
        updates.append({"entityId": program["entityId"], "expectedSha256": current_hash,
                        "sourcePath": export["absolutePath"], "sourceSha256": export["sha256"]})
        selected.append({"entityId": program["entityId"], "name": program["name"], "role": role})
    if not updates:
        raise ValueError("No existing AMS programmable blocks identified by source header")
    manifest = {"version": 1, "worldPath": world, "expectedSectorSha256": inspection["sectorSha256"], "scripts": updates}
    write("deployment.json", manifest)
    write("selected.json", selected)
    preview = call("world", "deploy", "--manifest", evidence / "deployment.json")
    write("preview.json", preview)
    result = preview
    if args.apply:
        result = call("world", "deploy", "--manifest", evidence / "deployment.json", "--apply", "true")
        write("applied.json", result)
        after = call("world", "inspect", "--path", world)
        write("after.json", after)
        expected = {item["entityId"]: item["sourceSha256"] for item in updates}
        actual = {item["entityId"]: item for item in after["programs"]}
        if set(actual) != {item["entityId"] for item in inspection["programs"]}:
            raise ValueError("Programmable block set changed")
        for before in inspection["programs"]:
            item = actual[before["entityId"]]
            if item["sha256"] != expected.get(before["entityId"], before["sha256"]) or item["storageSha256"] != before["storageSha256"] or item["enabled"] != before["enabled"]:
                raise ValueError("Post-deployment verification failed: " + before["entityId"])
        if after["sectorSha256"] != result["afterSectorSha256"] or after["caches"]:
            raise ValueError("Sector or binary-cache verification failed")
        write("verification.json", {"allSourceHashesMatch": True, "storageAndEnablementPreserved": True,
                                   "totalProgrammableBlocks": len(actual), "updated": len(expected), "unchanged": len(actual) - len(expected)})
    print(json.dumps({"applied": result["applied"], "worldPath": world, "selected": selected,
                      "backupPath": result["backupPath"], "evidence": str(evidence), "compilation": result["compilation"]}, ensure_ascii=False, indent=2))


if __name__ == "__main__":
    sys.stdout.reconfigure(encoding="utf-8")
    main()