"""Record a real AMS fleet cycle through AgentBridge; never invent game state."""
import argparse
from collections import defaultdict
from datetime import datetime, timezone
import json
from pathlib import Path
import subprocess
import sys
import tempfile
import time
from trace_ams import MAPPINGS, STATES
class Client:
def __init__(self, cli, pid=None):
self.cli, self.pid = str(cli), pid
def call(self, method, parameters=None):
payload = json.dumps(parameters or {})
request_file = None
try:
# PB sources exceed the Windows command-line limit. The CLI accepts
# UTF-8 request files and the plugin still enforces its normal limits.
if len(payload) > 4096:
with tempfile.NamedTemporaryFile(mode="w", suffix=".json", encoding="utf-8", delete=False) as stream:
request_file = Path(stream.name)
stream.write(payload)
payload = "@" + str(request_file)
args = [self.cli, "call", method, "--params", payload]
if self.pid:
args += ["--pid", str(self.pid)]
run = subprocess.run(args, capture_output=True, encoding="utf-8", timeout=35)
finally:
if request_file is not None:
request_file.unlink()
value = json.loads(run.stdout)
if run.returncode or "error" in value:
raise RuntimeError(json.dumps(value, ensure_ascii=False))
return value["result"]
def write(path, value):
temporary = path.with_name(path.name + ".tmp")
temporary.write_text(json.dumps(value, ensure_ascii=False, indent=2), encoding="utf-8")
temporary.replace(path)
def setup(client, preparation, directory):
status = client.call("world.status")
if not status.get("canMutate") or Path(status["path"]) != Path(preparation["debugWorld"]):
raise RuntimeError("Expected the authorized debug copy to be ready")
if (directory / "fixture.json").exists():
raise FileExistsError("Baseline already exists; use a new directory")
directory.mkdir(parents=True, exist_ok=True)
fixture = {"world": status, "programs": [], "observedIds": []}
for old in preparation["programmableBlocks"]:
program = client.call("pb.read", {"entityId": old["entityId"]})
write(directory / ("program-" + old["entityId"] + ".json"), program)
role = "fleet" if "舰队主控" in program["source"].splitlines()[0] else "miner"
if program["sha256"] != old["sourceSha256"] or not program["hasInstance"] or program["hasCompileErrors"]:
raise RuntimeError("Unexpected source or compilation failure: " + old["name"])
if role == "miner" and program["sha256"] not in MAPPINGS:
raise RuntimeError("Derive the exact packed field mapping before recording this version")
fixture["programs"].append({"id": old["entityId"], "gridId": old["gridId"], "name": old["name"], "role": role, "sha256": program["sha256"]})
blocks, offset = [], 0
while True:
page = client.call("blocks.list", {"gridId": old["gridId"], "limit": 256, "offset": offset})
blocks += page["blocks"]
if not page["truncated"]:
break
offset += len(page["blocks"])
write(directory / ("blocks-" + old["gridId"] + ".json"), blocks)
fixture["observedIds"] += [b["entityId"] for b in blocks if b.get("inventoryCount", 0) > 0 or b["entityId"] == old["entityId"]]
full = client.call("telemetry.snapshot", {"gridId": old["gridId"], "limit": 1024, "includeScreens": False})
write(directory / ("hardware-" + old["gridId"] + ".json"), full)
fixture["observedIds"] = list(dict.fromkeys(fixture["observedIds"]))
write(directory / "fixture.json", fixture)
return fixture
def sample(client, fixture):
record = {"utc": datetime.now(timezone.utc).isoformat(), "snapshots": [], "miners": []}
cargo = defaultdict(lambda: defaultdict(float))
connectors, grids = defaultdict(list), {}
ids = fixture["observedIds"]
for index in range(0, len(ids), 256):
shot = client.call("telemetry.snapshot", {"entityIds": ids[index:index+256], "limit": 256, "includeScreens": False})
record["snapshots"].append(shot)
grids.update((g["entityId"], g) for g in shot["grids"])
for block in shot["blocks"]:
if "telemetryError" in block:
raise RuntimeError("Incomplete telemetry: " + str(block))
if "connector" in block:
connectors[block["gridId"]].append({"id": block["entityId"], **block["connector"]})
if block.get("cargoInventory"):
for inventory in block.get("inventories", []):
if inventory.get("truncated"):
raise RuntimeError("Truncated cargo inventory")
for item in inventory.get("items", []):
if item["type"] == "MyObjectBuilder_Ore":
cargo[block["gridId"]][item["subtype"]] += item["amount"]
for program in fixture["programs"]:
if program["role"] != "miner":
continue
# Depth 3 expands cached rays/resources and can exhaust the plugin's
# 600-node budget before reaching the state fields on a busy miner.
fields = client.call("pb.inspect", {"entityId": program["id"], "fields": ["E", "H"], "depth": 2})["fields"]
mapping = MAPPINGS[program["sha256"]]
role = fields["H"]
flight, job = role[mapping["flight"]], role[mapping["job"]]
grid = grids[program["gridId"]]
record["miners"].append({"id": program["id"], "name": program["name"], "gridId": program["gridId"],
"sourceSha256": program["sha256"], "time": fields["E"], "state": STATES.get(role[mapping["state"]], role[mapping["state"]]),
"reason": role["l"], "speed": grid["speed"], "position": grid["pose"]["position"], "integrity": grid.get("integrity"),
"jobId": job.get("g") if isinstance(job, dict) else None, "job": job, "report": role[mapping["report"]],
"cargoOres": dict(cargo[program["gridId"]]), "connectors": connectors[program["gridId"]],
"flight": {key: flight[mapping[key]] for key in ("command", "attitude", "clearance", "scanIndex", "blocked", "problem", "cameraDetail", "obstacle")},
"fields": fields})
record["cargoByGrid"] = dict(cargo)
return record
def summary(record):
return {"utc": record["utc"], "miners": [{key: miner[key] for key in ("name", "state", "reason", "jobId", "speed", "cargoOres")} for miner in record["miners"]]}
def report(directory, after):
"""Summarize recorded observations, never treat an assignment as mining."""
miners, samples = {}, 0
for path in sorted(directory.glob("trace-*.jsonl")):
with path.open(encoding="utf-8") as stream:
for line in stream:
# A currently running recorder may not have finished its last line.
if not line.endswith("\n"):
continue
record = json.loads(line)
if record["utc"] < after:
continue
samples += 1
for m in record["miners"]:
entry = miners.setdefault(m["id"], {"name": m["name"], "states": {}, "peakCargoKg": 0,
"firstOreUtc": None, "dockedEmptyAfterOreUtc": None, "maximumDamage": 0,
"minimumBlockCount": None, "minimumIntegrity": None, "sourceHashes": []})
entry["states"].setdefault(m["state"], record["utc"])
if m["sourceSha256"] not in entry["sourceHashes"]:
entry["sourceHashes"].append(m["sourceSha256"])
ore = sum(m["cargoOres"].values())
entry["peakCargoKg"] = max(entry["peakCargoKg"], ore)
if ore > .01 and entry["firstOreUtc"] is None:
entry["firstOreUtc"] = record["utc"]
if entry["firstOreUtc"] and ore <= .01 and any(c["status"] == "Connected" for c in m["connectors"]):
entry["dockedEmptyAfterOreUtc"] = entry["dockedEmptyAfterOreUtc"] or record["utc"]
if m.get("integrity"):
entry["maximumDamage"] = max(entry["maximumDamage"], m["integrity"]["currentDamage"])
for target, field in [("minimumBlockCount", "inspectedBlockCount"), ("minimumIntegrity", "current")]:
value = m["integrity"][field]
entry[target] = value if entry[target] is None else min(entry[target], value)
entry["lastState"] = m["state"]
result = {"afterUtc": after, "samples": samples, "miners": miners,
"observedMiners": len(miners), "observedOrePickup": sum(bool(m["firstOreUtc"]) for m in miners.values()),
"observedDockedEmptyAfterPickup": sum(bool(m["dockedEmptyAfterOreUtc"]) for m in miners.values())}
write(directory / "cycle-report.json", result)
return result
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("mode", choices=("setup", "watch", "sample", "report"))
parser.add_argument("--cli", type=Path)
parser.add_argument("--pid", type=int)
parser.add_argument("--directory", type=Path, required=True)
parser.add_argument("--preparation", type=Path)
parser.add_argument("--seconds", type=float, default=60)
parser.add_argument("--interval", type=float, default=4)
parser.add_argument("--after", help="Report only observations after this ISO UTC timestamp")
args = parser.parse_args()
if args.mode == "report":
if not args.after:
parser.error("report requires --after to exclude earlier cargo")
print(json.dumps(report(args.directory, args.after), ensure_ascii=False, indent=2))
return
if args.cli is None:
parser.error("--cli is required for game requests")
client = Client(args.cli, args.pid)
if args.mode == "setup":
fixture = setup(client, json.loads(args.preparation.read_text(encoding="utf-8")), args.directory)
print(json.dumps({"programs": len(fixture["programs"]), "observedBlocks": len(fixture["observedIds"])}))
return
fixture = json.loads((args.directory / "fixture.json").read_text(encoding="utf-8"))
if args.mode == "sample":
result = sample(client, fixture)
write(args.directory / "latest.json", result)
print(json.dumps(summary(result), ensure_ascii=False))
return
stamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%S")
started = time.monotonic()
with (args.directory / ("trace-" + stamp + ".jsonl")).open("x", encoding="utf-8") as output:
while time.monotonic() - started < args.seconds:
result = sample(client, fixture)
output.write(json.dumps(result, ensure_ascii=False, separators=(",", ":")) + "\n")
output.flush()
write(args.directory / "latest.json", result)
print(json.dumps(summary(result), ensure_ascii=False), flush=True)
time.sleep(args.interval)
if __name__ == "__main__":
sys.stdout.reconfigure(encoding="utf-8")
main()
"""Record a real AMS fleet cycle through AgentBridge; never invent game state."""
import argparse
from collections import defaultdict
from datetime import datetime, timezone
import json
from pathlib import Path
import subprocess
import sys
import tempfile
import time
from trace_ams import MAPPINGS, STATES
class Client:
def __init__(self, cli, pid=None):
self.cli, self.pid = str(cli), pid
def call(self, method, parameters=None):
payload = json.dumps(parameters or {})
request_file = None
try:
# PB sources exceed the Windows command-line limit. The CLI accepts
# UTF-8 request files and the plugin still enforces its normal limits.
if len(payload) > 4096:
with tempfile.NamedTemporaryFile(mode="w", suffix=".json", encoding="utf-8", delete=False) as stream:
request_file = Path(stream.name)
stream.write(payload)
payload = "@" + str(request_file)
args = [self.cli, "call", method, "--params", payload]
if self.pid:
args += ["--pid", str(self.pid)]
run = subprocess.run(args, capture_output=True, encoding="utf-8", timeout=35)
finally:
if request_file is not None:
request_file.unlink()
value = json.loads(run.stdout)
if run.returncode or "error" in value:
raise RuntimeError(json.dumps(value, ensure_ascii=False))
return value["result"]
def write(path, value):
temporary = path.with_name(path.name + ".tmp")
temporary.write_text(json.dumps(value, ensure_ascii=False, indent=2), encoding="utf-8")
temporary.replace(path)
def setup(client, preparation, directory):
status = client.call("world.status")
if not status.get("canMutate") or Path(status["path"]) != Path(preparation["debugWorld"]):
raise RuntimeError("Expected the authorized debug copy to be ready")
if (directory / "fixture.json").exists():
raise FileExistsError("Baseline already exists; use a new directory")
directory.mkdir(parents=True, exist_ok=True)
fixture = {"world": status, "programs": [], "observedIds": []}
for old in preparation["programmableBlocks"]:
program = client.call("pb.read", {"entityId": old["entityId"]})
write(directory / ("program-" + old["entityId"] + ".json"), program)
role = "fleet" if "舰队主控" in program["source"].splitlines()[0] else "miner"
if program["sha256"] != old["sourceSha256"] or not program["hasInstance"] or program["hasCompileErrors"]:
raise RuntimeError("Unexpected source or compilation failure: " + old["name"])
if role == "miner" and program["sha256"] not in MAPPINGS:
raise RuntimeError("Derive the exact packed field mapping before recording this version")
fixture["programs"].append({"id": old["entityId"], "gridId": old["gridId"], "name": old["name"], "role": role, "sha256": program["sha256"]})
blocks, offset = [], 0
while True:
page = client.call("blocks.list", {"gridId": old["gridId"], "limit": 256, "offset": offset})
blocks += page["blocks"]
if not page["truncated"]:
break
offset += len(page["blocks"])
write(directory / ("blocks-" + old["gridId"] + ".json"), blocks)
fixture["observedIds"] += [b["entityId"] for b in blocks if b.get("inventoryCount", 0) > 0 or b["entityId"] == old["entityId"]]
full = client.call("telemetry.snapshot", {"gridId": old["gridId"], "limit": 1024, "includeScreens": False})
write(directory / ("hardware-" + old["gridId"] + ".json"), full)
fixture["observedIds"] = list(dict.fromkeys(fixture["observedIds"]))
write(directory / "fixture.json", fixture)
return fixture
def sample(client, fixture):
record = {"utc": datetime.now(timezone.utc).isoformat(), "snapshots": [], "miners": []}
cargo = defaultdict(lambda: defaultdict(float))
connectors, grids = defaultdict(list), {}
ids = fixture["observedIds"]
for index in range(0, len(ids), 256):
shot = client.call("telemetry.snapshot", {"entityIds": ids[index:index+256], "limit": 256, "includeScreens": False})
record["snapshots"].append(shot)
grids.update((g["entityId"], g) for g in shot["grids"])
for block in shot["blocks"]:
if "telemetryError" in block:
raise RuntimeError("Incomplete telemetry: " + str(block))
if "connector" in block:
connectors[block["gridId"]].append({"id": block["entityId"], **block["connector"]})
if block.get("cargoInventory"):
for inventory in block.get("inventories", []):
if inventory.get("truncated"):
raise RuntimeError("Truncated cargo inventory")
for item in inventory.get("items", []):
if item["type"] == "MyObjectBuilder_Ore":
cargo[block["gridId"]][item["subtype"]] += item["amount"]
for program in fixture["programs"]:
if program["role"] != "miner":
continue
# Depth 3 expands cached rays/resources and can exhaust the plugin's
# 600-node budget before reaching the state fields on a busy miner.
fields = client.call("pb.inspect", {"entityId": program["id"], "fields": ["E", "H"], "depth": 2})["fields"]
mapping = MAPPINGS[program["sha256"]]
role = fields["H"]
flight, job = role[mapping["flight"]], role[mapping["job"]]
grid = grids[program["gridId"]]
record["miners"].append({"id": program["id"], "name": program["name"], "gridId": program["gridId"],
"sourceSha256": program["sha256"], "time": fields["E"], "state": STATES.get(role[mapping["state"]], role[mapping["state"]]),
"reason": role["l"], "speed": grid["speed"], "position": grid["pose"]["position"], "integrity": grid.get("integrity"),
"jobId": job.get("g") if isinstance(job, dict) else None, "job": job, "report": role[mapping["report"]],
"cargoOres": dict(cargo[program["gridId"]]), "connectors": connectors[program["gridId"]],
"flight": {key: flight[mapping[key]] for key in ("command", "attitude", "clearance", "scanIndex", "blocked", "problem", "cameraDetail", "obstacle")},
"fields": fields})
record["cargoByGrid"] = dict(cargo)
return record
def summary(record):
return {"utc": record["utc"], "miners": [{key: miner[key] for key in ("name", "state", "reason", "jobId", "speed", "cargoOres")} for miner in record["miners"]]}
def report(directory, after):
"""Summarize recorded observations, never treat an assignment as mining."""
miners, samples = {}, 0
for path in sorted(directory.glob("trace-*.jsonl")):
with path.open(encoding="utf-8") as stream:
for line in stream:
# A currently running recorder may not have finished its last line.
if not line.endswith("\n"):
continue
record = json.loads(line)
if record["utc"] < after:
continue
samples += 1
for m in record["miners"]:
entry = miners.setdefault(m["id"], {"name": m["name"], "states": {}, "peakCargoKg": 0,
"firstOreUtc": None, "dockedEmptyAfterOreUtc": None, "maximumDamage": 0,
"minimumBlockCount": None, "minimumIntegrity": None, "sourceHashes": []})
entry["states"].setdefault(m["state"], record["utc"])
if m["sourceSha256"] not in entry["sourceHashes"]:
entry["sourceHashes"].append(m["sourceSha256"])
ore = sum(m["cargoOres"].values())
entry["peakCargoKg"] = max(entry["peakCargoKg"], ore)
if ore > .01 and entry["firstOreUtc"] is None:
entry["firstOreUtc"] = record["utc"]
if entry["firstOreUtc"] and ore <= .01 and any(c["status"] == "Connected" for c in m["connectors"]):
entry["dockedEmptyAfterOreUtc"] = entry["dockedEmptyAfterOreUtc"] or record["utc"]
if m.get("integrity"):
entry["maximumDamage"] = max(entry["maximumDamage"], m["integrity"]["currentDamage"])
for target, field in [("minimumBlockCount", "inspectedBlockCount"), ("minimumIntegrity", "current")]:
value = m["integrity"][field]
entry[target] = value if entry[target] is None else min(entry[target], value)
entry["lastState"] = m["state"]
result = {"afterUtc": after, "samples": samples, "miners": miners,
"observedMiners": len(miners), "observedOrePickup": sum(bool(m["firstOreUtc"]) for m in miners.values()),
"observedDockedEmptyAfterPickup": sum(bool(m["dockedEmptyAfterOreUtc"]) for m in miners.values())}
write(directory / "cycle-report.json", result)
return result
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("mode", choices=("setup", "watch", "sample", "report"))
parser.add_argument("--cli", type=Path)
parser.add_argument("--pid", type=int)
parser.add_argument("--directory", type=Path, required=True)
parser.add_argument("--preparation", type=Path)
parser.add_argument("--seconds", type=float, default=60)
parser.add_argument("--interval", type=float, default=4)
parser.add_argument("--after", help="Report only observations after this ISO UTC timestamp")
args = parser.parse_args()
if args.mode == "report":
if not args.after:
parser.error("report requires --after to exclude earlier cargo")
print(json.dumps(report(args.directory, args.after), ensure_ascii=False, indent=2))
return
if args.cli is None:
parser.error("--cli is required for game requests")
client = Client(args.cli, args.pid)
if args.mode == "setup":
fixture = setup(client, json.loads(args.preparation.read_text(encoding="utf-8")), args.directory)
print(json.dumps({"programs": len(fixture["programs"]), "observedBlocks": len(fixture["observedIds"])}))
return
fixture = json.loads((args.directory / "fixture.json").read_text(encoding="utf-8"))
if args.mode == "sample":
result = sample(client, fixture)
write(args.directory / "latest.json", result)
print(json.dumps(summary(result), ensure_ascii=False))
return
stamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%S")
started = time.monotonic()
with (args.directory / ("trace-" + stamp + ".jsonl")).open("x", encoding="utf-8") as output:
while time.monotonic() - started < args.seconds:
result = sample(client, fixture)
output.write(json.dumps(result, ensure_ascii=False, separators=(",", ":")) + "\n")
output.flush()
write(args.directory / "latest.json", result)
print(json.dumps(summary(result), ensure_ascii=False), flush=True)
time.sleep(args.interval)
if __name__ == "__main__":
sys.stdout.reconfigure(encoding="utf-8")
main()
