import base64
import importlib.util
import io
import json
from pathlib import Path
import tempfile
import threading
import unittest
from unittest.mock import MagicMock, call, mock_open, patch

spec = importlib.util.spec_from_file_location("ec_agent", Path(__file__).resolve().parents[1] / "ec_agent.py")
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
Agent, AgentError, Held, Journal = module.Agent, module.AgentError, module.Held, module.Journal
INSTANCE = "a53d99da-a1eb-4f01-9b10-a6268b556542"


class FakeDocker:
    def __init__(self):
        self.containers = {}
        self.calls = []
        self.fail_restart = False
        self.lose_reply_after = None
        self.started = 0

    def inspect(self, name):
        return self.containers.get(name)

    def call(self, *args, **kwargs):
        self.calls.append(args)
        if args[0] == "info":
            return json.dumps({"ServerVersion": "27.5.1"})
        if args[:2] == ("image", "ls"):
            return "ghcr.io/ayymoss/plutainer:latest\n"
        if args[:2] == ("container", "create"):
            labels = {args[index + 1].split("=", 1)[0]: args[index + 1].split("=", 1)[1] for index, arg in enumerate(args) if arg == "--label"}
            name = args[args.index("--name") + 1]
            self.containers[name] = {"Id": "container-123", "Config": {"Labels": labels}, "State": {"Status": "created", "Running": False}}
        if args[:2] in (("container", "start"), ("container", "restart")):
            if self.fail_restart and args[1] == "restart":
                raise AgentError("Docker command unavailable or timed out")
            self.started += 1
            self.containers[args[-1]]["State"] = {"Status": "running", "Running": True, "StartedAt": f"started-{self.started}"}
        if args[:2] == ("container", "stop"):
            self.containers[args[-1]]["State"] = {"Status": "exited", "Running": False}
        if args[:2] == ("container", "rm"):
            self.containers.pop(args[-1])
        if args[0] == "container" and args[1] == self.lose_reply_after:
            raise AgentError("Docker command unavailable or timed out")
        return ""


class Response(io.BytesIO):
    def __init__(self, content, status=200, headers=None):
        super().__init__(content)
        self.status = status
        self.headers = headers or {}


class JournalRetentionTests(unittest.TestCase):
    def setUp(self):
        self.temp = tempfile.TemporaryDirectory()
        self.path = Path(self.temp.name) / "operations.sqlite3"
        self.journal = Journal(self.path)
        self.now = 2000000
        self.sequence = 0

    def tearDown(self):
        self.journal.close()
        self.temp.cleanup()

    def record(self, kind="files.stat", payload=None, result=None, state="completed", error=None, at=None, confirmed=True):
        self.sequence += 1
        operation = {"id": f"operation-{self.sequence}", "kind": kind, "payload": payload or {}}
        self.journal.accept(operation)
        with patch.object(module.time, "time", return_value=self.now - 172800 if at is None else at):
            row = self.journal.update(operation["id"], state, result, error)
        if confirmed:
            self.journal.confirm_completed(row)
        return row

    def prune(self, days=1):
        with patch.object(module.time, "time", return_value=self.now):
            return self.journal.prune_observations(days)

    def test_cutoff_requires_an_older_confirmed_success(self):
        old = self.record(at=self.now - 86401)
        boundary = self.record(at=self.now - 86400)
        recent = self.record(at=self.now - 10)
        unconfirmed = self.record(confirmed=False)
        self.assertEqual(self.prune(), 1)
        with self.assertRaisesRegex(AgentError, "Unknown operation"):
            self.journal.get(old["id"])
        for row in (boundary, recent, unconfirmed):
            self.assertEqual(self.journal.get(row["id"])["state"], "completed")

    def test_zero_and_invalid_retention_preserve_every_record(self):
        row = self.record()
        for days in (0, -1, 36501, True, False, None, "1", 1.0, 1.5):
            with self.subTest(days=days):
                self.assertEqual(self.prune(days), 0)
                self.assertEqual(self.journal.get(row["id"])["state"], "completed")

    def test_all_read_kinds_are_eligible_but_mutations_keep_their_replay_journal(self):
        for kind in ("host.inspect", "instance.inspect", "files.read", "files.list", "files.stat", "files.manifest", "logs.read"):
            self.record(kind, result={"healthy": True})
        for command in ("status", "getstatus", "version"):
            self.record("rcon.execute", {"command": command}, {"ok": True})
        mutations = [self.record(kind) for kind in ("instance.restart", "instance.stop", "instance.create", "files.stage", "files.activate", "credentials.rotate")]
        mutations.append(self.record("rcon.execute", {"command": "map_restart"}, {"ok": True}))
        self.assertEqual(self.prune(), 10)
        for row in mutations:
            saved, accepted = self.journal.accept(json.loads(row["request"]))
            self.assertFalse(accepted)
            self.assertEqual(saved["id"], row["id"])

    def test_errors_failures_and_unfinished_operations_are_retained(self):
        retained = [self.record(result=result) for result in ({"ok": False}, {"healthy": False}, {"error": "partial result"}, {"error": False}, {"error": []})]
        retained.append(self.record(error="read did not finish"))
        retained.extend(self.record(state=state) for state in ("accepted", "running", "unknown", "failed"))
        self.assertEqual(self.prune(), 0)
        for row in retained:
            self.assertEqual(self.journal.get(row["id"])["state"], row["state"])

    def test_old_receipt_cannot_confirm_a_changed_or_reexecuted_result(self):
        old = self.record(confirmed=False)
        with patch.object(module.time, "time", return_value=self.now - 90000):
            current = self.journal.update(old["id"], "completed", {"new": "result"})
        self.journal.confirm_completed(old)
        self.assertIsNone(self.journal.get(old["id"])["reported_at"])
        self.assertEqual(self.prune(), 0)
        self.journal.confirm_completed(current)
        self.assertEqual(self.prune(), 1)

    def test_pruning_uses_bounded_indexed_batches(self):
        template = self.record()
        self.journal.db.executemany("""INSERT INTO operations(id,fingerprint,request,state,result,updated,reported_at,retention_eligible)
            VALUES(?,?,?,'completed',NULL,?,?,1)""", [
                (f"batch-{index}", template["fingerprint"], template["request"], self.now - 172800, self.now - 172800)
                for index in range(504)
            ])
        self.journal.db.commit()
        self.assertEqual(self.prune(), 500)
        self.assertEqual(self.journal.db.execute("SELECT COUNT(*) FROM operations").fetchone()[0], 5)
        self.assertEqual(self.prune(), 5)
        plan = self.journal.db.execute("""EXPLAIN QUERY PLAN SELECT id FROM operations
            WHERE state='completed' AND reported_at IS NOT NULL AND retention_eligible=1 AND updated<? ORDER BY updated,id LIMIT 500""", (self.now,)).fetchall()
        self.assertIn("operations_observation_retention", " ".join(str(tuple(row)) for row in plan))

    def test_migration_keeps_legacy_completions_until_their_receipt_is_reacknowledged(self):
        self.journal.close()
        self.path.unlink()
        legacy = module.sqlite3.connect(self.path)
        legacy.execute("CREATE TABLE operations(id TEXT PRIMARY KEY, fingerprint TEXT NOT NULL, request TEXT NOT NULL, state TEXT NOT NULL, result TEXT, error TEXT, updated REAL NOT NULL)")
        request = {"id": "legacy-read", "kind": "files.list", "instanceId": None, "generation": None, "payload": {}}
        canonical = json.dumps(request, sort_keys=True, separators=(",", ":"))
        legacy.execute("INSERT INTO operations VALUES(?,?,?,'completed',?,NULL,?)", (request["id"], module.digest(canonical.encode()), canonical, '{"entries":[]}', self.now - 172800))
        legacy.commit()
        legacy.close()
        self.journal = Journal(self.path)
        row, accepted = self.journal.accept(request)
        self.assertFalse(accepted)
        self.assertIsNone(row["reported_at"])
        self.assertEqual(self.prune(), 0)
        self.journal.confirm_completed(row)
        self.assertEqual(self.prune(), 1)


class ManifestTests(unittest.TestCase):
    def setUp(self):
        self.temp = tempfile.TemporaryDirectory()
        self.root = Path(self.temp.name) / "app"
        self.root.mkdir()
        self.stopping = threading.Event()

    def tearDown(self):
        self.temp.cleanup()

    def write(self, path, content=b"content"):
        target = self.root / path
        target.parent.mkdir(parents=True, exist_ok=True)
        target.write_bytes(content)
        return target

    def manifest(self, roots=None, ignore_names=None, paths=None):
        return module.managed_manifest(self.root, {
            "roots": ["scripts", "configs", "plugins"] if roots is None else roots,
            "ignoreNames": [] if ignore_names is None else ignore_names,
            "paths": [] if paths is None else paths,
        }, self.stopping)

    def test_hashes_sorted_unique_files_across_overlapping_roots_and_skips_ignored_names(self):
        self.write("scripts/nested/z.gsc", b"script")
        self.write("configs/server.cfg", b"config")
        self.write("scripts/cache/ignored.gsc")
        self.write("scripts/nested/ignore.gsc")
        expected = [{"path": path, "size": len(content), "sha256": module.digest(content)} for path, content in (
            ("configs/server.cfg", b"config"), ("scripts/nested/z.gsc", b"script"),
        )]
        self.assertEqual(self.manifest(["scripts/nested", "configs", "scripts", "scripts", "plugins"], ["cache", "ignore.gsc"]), {"files": expected})

    def test_nested_symlinks_are_skipped_but_requested_symlink_roots_fail(self):
        self.write("scripts/good.gsc")
        outside = Path(self.temp.name) / "outside"
        outside.mkdir()
        (outside / "secret").write_bytes(b"secret")
        (self.root / "scripts/file-link").symlink_to(outside / "secret")
        (self.root / "scripts/dir-link").symlink_to(outside, target_is_directory=True)
        (self.root / "scripts/dangling").symlink_to(outside / "missing")
        self.assertEqual([item["path"] for item in self.manifest()["files"]], ["scripts/good.gsc"])
        for roots in (["scripts/dir-link"], ["scripts", "scripts/dir-link"]):
            with self.subTest(roots=roots), self.assertRaisesRegex(AgentError, "symlink|Symlink"):
                self.manifest(roots)

    def test_validates_roots_and_ignored_names(self):
        for roots in ([], "scripts", [""], ["."], [".."], ["../scripts"], ["/scripts"], ["scripts//nested"], ["scripts/../configs"], [None], ["scripts"] * 65):
            with self.subTest(roots=roots), self.assertRaises(AgentError):
                self.manifest(roots)
        for ignored in ("cache", [None], [""], ["."], [".."], ["scripts/cache"], ["a\\b"], ["a\x00b"], ["cache"] * 257):
            with self.subTest(ignored=ignored), self.assertRaises(AgentError):
                self.manifest(ignore_names=ignored)
        for paths in ("scripts/a", [None], ["../outside"], ["."], ["scripts/a"] * 10001):
            with self.subTest(paths=paths), self.assertRaises(AgentError):
                self.manifest(paths=paths)

    def test_explicit_managed_paths_cannot_be_symlinks_or_directories(self):
        self.write("scripts/good.gsc")
        (self.root / "scripts/link.gsc").symlink_to(self.root / "scripts/good.gsc")
        (self.root / "scripts/dangling.gsc").symlink_to(self.root / "missing")
        for path in ("scripts", "scripts/link.gsc", "scripts/dangling.gsc"):
            with self.subTest(path=path), self.assertRaises(AgentError):
                self.manifest(paths=[path])
        self.assertEqual([item["path"] for item in self.manifest(paths=["scripts/good.gsc", "scripts/missing.gsc", "missing/future.gsc"])["files"]], ["scripts/good.gsc"])

    def test_explicit_managed_paths_are_validated_again_after_the_scan(self):
        self.write("scripts/good.gsc")
        original = module.managed_open
        opened = 0

        @module.contextlib.contextmanager
        def swap(root, value):
            nonlocal opened
            opened += 1
            if opened == 2:
                (self.root / "scripts/good.gsc").unlink()
                (self.root / "scripts/good.gsc").symlink_to(self.root / "missing")
            with original(root, value) as stream:
                yield stream

        with patch.object(module, "managed_open", side_effect=swap), self.assertRaisesRegex(AgentError, "Symlinks"):
            self.manifest(paths=["scripts/good.gsc"])

    def test_missing_destinations_are_empty_but_unavailable_app_and_regular_roots_fail(self):
        self.assertEqual(self.manifest(), {"files": []})
        self.write("scripts", b"not a directory")
        with self.assertRaisesRegex(AgentError, "complete managed tree"):
            self.manifest()
        (self.root / "scripts").unlink()
        self.root.rmdir()
        with self.assertRaisesRegex(AgentError, "complete managed tree"):
            self.manifest()
        replacement = Path(self.temp.name) / "replacement"
        replacement.mkdir()
        self.root.symlink_to(replacement, target_is_directory=True)
        with self.assertRaisesRegex(AgentError, "symlink"):
            self.manifest()

    def test_errors_and_special_files_never_return_a_partial_manifest(self):
        self.write("scripts/good.gsc")
        self.write("scripts/unreadable.gsc")
        original = module.os.stat

        def denied(path, *args, **kwargs):
            if path == "unreadable.gsc":
                raise PermissionError("unreadable")
            return original(path, *args, **kwargs)

        with patch.object(module.os, "stat", side_effect=denied), self.assertRaisesRegex(AgentError, "complete managed tree"):
            self.manifest()
        module.os.mkfifo(self.root / "scripts/fifo")
        with self.assertRaisesRegex(AgentError, "non-regular"):
            self.manifest()

    def test_scan_limits_fail_instead_of_truncating_results(self):
        self.write("scripts/a.gsc")
        self.write("scripts/b.gsc")
        for setting, limit, message in (
            ("MANIFEST_MAX_ENTRIES", 1, "entry limit"),
            ("MANIFEST_MAX_BYTES", 1, "result size limit"),
            ("MANIFEST_MAX_SECONDS", 0, "time limit"),
            ("MANIFEST_MAX_DEPTH", 0, "depth limit"),
        ):
            with self.subTest(setting=setting), patch.object(module, setting, limit), self.assertRaisesRegex(AgentError, message):
                self.manifest()
        self.stopping.set()
        with self.assertRaisesRegex(AgentError, "interrupted"):
            self.manifest()

    def test_files_are_streamed_and_not_limited_by_the_content_read_api(self):
        content = b"x" * (2 * 1024 * 1024 + 13)
        self.write("plugins/large.dll", content)
        original = module.os.fdopen
        reads = []

        @module.contextlib.contextmanager
        def tracked(*args, **kwargs):
            with original(*args, **kwargs) as stream:
                proxy = MagicMock(wraps=stream)
                def read(size):
                    reads.append(size)
                    return stream.read(size)
                proxy.read.side_effect = read
                yield proxy

        with patch.object(module, "MAX_FILE", 1), patch.object(module.os, "fdopen", side_effect=tracked):
            result = self.manifest()
        self.assertEqual(result, {"files": [{"path": "plugins/large.dll", "size": len(content), "sha256": module.digest(content)}]})
        self.assertGreater(len(reads), 2)
        self.assertEqual(set(reads), {1024 * 1024})

    def test_file_changes_while_hashing_invalidate_the_scan(self):
        target = self.write("scripts/changing.gsc", b"before")
        original = module.hashlib.sha256

        def changing_hash():
            digest = original()
            proxy = MagicMock(wraps=digest)
            def update(block):
                digest.update(block)
                prior = target.stat()
                target.write_bytes(b"after!")
                module.os.utime(target, ns=(prior.st_atime_ns, prior.st_mtime_ns + 1))
            proxy.update.side_effect = update
            return proxy

        with patch.object(module.hashlib, "sha256", side_effect=changing_hash), self.assertRaisesRegex(AgentError, "file changed"):
            self.manifest()

    def test_symlink_swap_after_validation_cannot_hash_outside_the_app(self):
        self.write("scripts/good.gsc")
        outside = Path(self.temp.name) / "outside"
        outside.mkdir()
        (outside / "secret").write_bytes(b"secret")
        original = module.scoped

        def swap(root, relative, **kwargs):
            result = original(root, relative, **kwargs)
            (self.root / "scripts").rename(self.root / "saved")
            (self.root / "scripts").symlink_to(outside, target_is_directory=True)
            return result

        with patch.object(module, "scoped", side_effect=swap), self.assertRaisesRegex(AgentError, "complete managed tree"):
            self.manifest(["scripts"])

    def test_disappearing_app_after_hashing_is_not_successful_evidence(self):
        self.write("scripts/good.gsc")
        original = module.hashlib.sha256

        def moving_hash():
            digest = original()
            proxy = MagicMock(wraps=digest)
            def update(block):
                digest.update(block)
                self.root.rename(self.root.with_name("moved-app"))
            proxy.update.side_effect = update
            return proxy

        with patch.object(module.hashlib, "sha256", side_effect=moving_hash), self.assertRaisesRegex(AgentError, "complete managed tree"):
            self.manifest()


class AgentTests(unittest.TestCase):
    def setUp(self):
        self.temp = tempfile.TemporaryDirectory()
        self.config = {"root": self.temp.name, "controller": "https://controller.example", "credential": "secret", "portRange": {"min": 28960, "max": 28980}}
        self.docker = FakeDocker()
        self.agent = Agent(self.config, self.docker)
        self.library_id = "library-one"
        self.library = self.agent.library_dir("t5", self.library_id)
        (self.library / "content").mkdir(parents=True)
        (self.library / "content" / "game.ff").write_bytes(b"game-data")
        files = [{"path": "game.ff", "size": 9, "sha256": module.digest(b"game-data")}]
        module.atomic_json(self.library / "manifest.json", {"libraryId": self.library_id, "game": "t5", "size": 9, "files": files, "manifestSha256": module.digest(json.dumps(files, separators=(",", ":")).encode())})

    def tearDown(self):
        self.agent.journal.close()
        self.temp.cleanup()

    def test_library_remove_preserves_other_versions_and_is_idempotent(self):
        other = self.agent.library_dir("t5", "other-library")
        other.mkdir()
        staging = self.library.with_name("." + self.library_id + ".partial")
        staging.mkdir()
        (staging / "partial").write_bytes(b"partial")
        payload = {"libraryVersionId": self.library_id, "game": "t5"}
        self.assertTrue(self.agent.library_remove(payload)["removed"])
        self.assertFalse(self.library.exists())
        self.assertFalse(staging.exists())
        self.assertTrue(other.exists())
        self.assertTrue(self.agent.library_remove(payload)["removed"])

    def test_library_remove_refuses_stopped_container_mounts_and_docker_errors(self):
        payload = {"libraryVersionId": self.library_id, "game": "t5"}
        with patch.object(self.docker, "call", return_value="stopped-container"), patch.object(self.docker, "inspect", return_value={
            "State": {"Running": False}, "Mounts": [{"Source": str(self.library / "content")}]
        }):
            with self.assertRaisesRegex(AgentError, "still mounted"):
                self.agent.library_remove(payload)
        with patch.object(self.docker, "call", side_effect=AgentError("Docker unavailable")):
            with self.assertRaises(AgentError):
                self.agent.library_remove(payload)
        self.assertTrue((self.library / "content" / "game.ff").exists())

    def test_library_remove_recovers_after_interrupted_deletion(self):
        payload = {"libraryVersionId": self.library_id, "game": "t5"}
        operation = {"id": "library-cleanup", "kind": "library.remove", "payload": payload}
        self.agent.journal.accept(operation)
        self.agent.journal.update(operation["id"], "running")
        self.agent.journal.checkpoint(operation["id"], {"phase": "removing-library"})
        (self.library / "manifest.json").unlink()
        row = self.agent.reconcile(operation, self.agent.journal.get(operation["id"]))
        self.assertEqual(row["state"], "completed")
        self.assertFalse(self.library.exists())

    def test_library_remove_ignores_unrelated_and_tmpfs_mounts(self):
        with patch.object(self.docker, "call", return_value="other-container"), patch.object(self.docker, "inspect", return_value={
            "Mounts": [{"Type": "tmpfs", "Source": ""}, {"Source": str(self.agent.root / "instances")}]
        }):
            self.agent.library_remove({"libraryVersionId": self.library_id, "game": "t5"})
        self.assertFalse(self.library.exists())

    def test_library_remove_refuses_symlink_storage(self):
        linked = self.agent.library_dir("t5", "linked-library")
        linked.symlink_to(self.library, target_is_directory=True)
        with self.assertRaisesRegex(AgentError, "symlink"):
            self.agent.library_remove({"libraryVersionId": "linked-library", "game": "t5"})
        self.assertTrue(self.library.exists())

    def cpu_sample(self, line):
        with patch.object(Path, "open", mock_open(read_data=line)):
            return self.agent.cpu_usage_percent()

    def test_cpu_usage_uses_interval_and_excludes_guest_double_counting(self):
        self.assertIsNone(self.cpu_sample("cpu 100 0 50 800 50 0 0 0 20 0\n"))
        # 30 busy ticks (including steal), 60 idle, 10 I/O wait = 30%.
        self.assertEqual(self.cpu_sample("cpu 120 0 55 860 60 0 0 5 30 0\n"), 30.0)
        self.assertEqual(self.cpu_sample("cpu 120 0 55 960 60 0 0 5 30 0\n"), 0.0)
        self.assertEqual(self.cpu_sample("cpu 220 0 55 960 60 0 0 5 30 0\n"), 100.0)

    def test_cpu_usage_rebases_after_reset_or_invalid_sample(self):
        baseline = "cpu 100 0 50 800 50 0 0 0\n"
        self.assertIsNone(self.cpu_sample(baseline))
        self.assertIsNone(self.cpu_sample(baseline))
        self.assertIsNone(self.cpu_sample("cpu 10 0 5 80 5 0 0 0\n"))
        self.assertIsNone(self.cpu_sample("cpu 20 0 5 90 4 0 0 0\n"))
        for invalid in ("", "cpu0 1 2 3 4 5 6 7 8", "cpu 1 2", "cpu x 0 0 0 0 0 0 0", "cpu -1 0 0 0 0 0 0 0"):
            self.assertIsNone(self.cpu_sample(invalid))
            self.assertIsNone(self.cpu_sample(baseline))
        with patch.object(Path, "open", side_effect=OSError("unavailable")):
            self.assertIsNone(self.agent.cpu_usage_percent())
        self.assertIsNone(self.cpu_sample(baseline))

    def test_inventory_reports_cpu_only_when_available(self):
        for usage in (None, 0.0, 42.3, 100.0):
            with patch.object(self.agent, "cpu_usage_percent", return_value=usage):
                inventory = self.agent.inventory()
            if usage is None:
                self.assertNotIn("cpuUsagePercent", inventory)
            else:
                self.assertEqual(inventory["cpuUsagePercent"], usage)

    def resource_container(self):
        return {"Name": self.agent.container_name(INSTANCE, 1),
                "Config": {"Labels": {module.LABEL + ".managed": "true", module.LABEL + ".instance": INSTANCE, module.LABEL + ".generation": "1"}},
                "State": {"Running": True, "StartedAt": "2000-01-01T00:00:00.000000000Z"},
                "Mounts": [{"Type": "bind", "Source": str(self.agent.instance_dir(INSTANCE, 1) / "app"), "Destination": "/home/plutainer/app"}]}

    def test_instance_resources_batches_owned_containers_and_normalizes_core_cpu_and_memory(self):
        name = self.agent.container_name(INSTANCE, 1)
        rows = json.dumps({"Name": name, "CPUPerc": "150.00%", "MemUsage": "384.5MiB / 8GiB", "PIDs": "12"})
        with patch.object(self.docker, "call", side_effect=["container-id\n", json.dumps([self.resource_container()]), rows]) as call:
            samples = self.agent.instance_resources()
        self.assertEqual(len(samples), 1)
        self.assertEqual(samples[0]["instanceId"], INSTANCE)
        self.assertEqual(samples[0]["generation"], 1)
        self.assertEqual(samples[0]["cpuCores"], 1.5)
        self.assertEqual(samples[0]["memoryBytes"], int(384.5 * 1024 ** 2))
        self.assertGreater(samples[0]["uptimeSeconds"], 300)
        self.assertEqual(call.call_args_list[0].args, ("container", "ls", "--filter", "label=io.electric-cherry.managed=true", "--no-trunc", "--format", "{{.ID}}"))
        self.assertEqual(call.call_args_list[-1].args, ("stats", "--no-stream", "--no-trunc", "--format", "{{json .}}", name))

    def test_instance_resources_ignores_stopped_foreign_and_mismatched_containers(self):
        for change in ("stopped", "foreign", "generation", "name"):
            info = self.resource_container()
            if change == "stopped":
                info["State"]["Running"] = False
            elif change == "foreign":
                info["Mounts"][0]["Source"] = "/another-installation/app"
            elif change == "generation":
                info["Config"]["Labels"][module.LABEL + ".generation"] = "2"
            else:
                info["Name"] = "unrelated"
            with patch.object(self.docker, "call", side_effect=["container-id\n", json.dumps([info])]) as call:
                self.assertEqual(self.agent.instance_resources(), [])
                self.assertEqual(call.call_count, 2)

    def test_instance_resources_skips_unavailable_stats_instead_of_reporting_zero_usage(self):
        good = {"Name": self.agent.container_name(INSTANCE, 1), "CPUPerc": "0.00%", "MemUsage": "200MiB / 8GiB", "PIDs": "10"}
        for override in ({"CPUPerc": "NaN%"}, {"CPUPerc": "-1%"}, {"CPUPerc": "--"}, {"MemUsage": "bad"}, {"MemUsage": "0B / 0B"}, {"PIDs": "0"}, {"Name": "foreign"}):
            with patch.object(self.docker, "call", side_effect=["container-id\n", json.dumps([self.resource_container()]), json.dumps({**good, **override})]):
                self.assertEqual(self.agent.instance_resources(), [])
        with patch.object(self.docker, "call", side_effect=["container-id\n", json.dumps([self.resource_container()]), json.dumps(good)]):
            self.assertEqual(self.agent.instance_resources()[0]["cpuCores"], 0)
        with patch.object(self.agent, "instance_resources", side_effect=AgentError("timed out")):
            inventory = self.agent.inventory()
            self.assertNotIn("instanceResources", inventory)
            self.assertIn("sampling failed", inventory["instanceResourcesError"])
        with patch.object(self.agent, "instance_resources", return_value=[]):
            inventory = self.agent.inventory()
            self.assertEqual(inventory["instanceResources"], [])
            self.assertNotIn("instanceResourcesError", inventory)

    def operation(self, kind, payload=None, op_id="operation-1", generation=1):
        return {"id": op_id, "kind": kind, "instanceId": INSTANCE, "generation": generation, "payload": payload or {}}

    def create(self):
        payload = {"name": "t5_zombie_test_1", "game": "t5", "libraryId": self.library_id, "image": "ghcr.io/ayymoss/plutainer:latest", "port": 28960, "env": {"PLUTAINER_PORT": "28960"}, "rconPassword": "rcon-secret", "files": [{"path": "configs/server.cfg", "contentBase64": base64.b64encode(b"original").decode()}]}
        result = self.agent.execute(self.operation("instance.create", payload, "create"))
        self.assertEqual(result["state"], "completed", result)
        return payload

    def test_same_id_reports_saved_result_without_executing_again(self):
        self.create()
        operation = self.operation("instance.restart")
        self.agent.rcon = lambda *_: {"ok": True, "players": 0}
        first = self.agent.execute(operation)
        self.assertEqual(first["state"], "completed")
        second = self.agent.execute({**operation, "state": "running", "updatedAt": "later", "result": None})
        self.assertEqual(second["result"], first["result"])
        self.assertEqual(sum(call[:2] == ("container", "restart") for call in self.docker.calls), 1)

    def test_manifest_is_advertised_and_reconciles_as_a_read_operation(self):
        self.create()
        operation = self.operation("files.manifest", {"roots": ["configs", "scripts"], "ignoreNames": [], "paths": ["configs/server.cfg"]})
        self.assertIn("files.manifest", self.agent.inventory()["capabilities"])
        self.assertTrue(self.agent.read_operation(operation))
        self.assertFalse(self.agent.health_operation(operation))
        self.agent.journal.accept(operation)
        self.agent.journal.update(operation["id"], "unknown")
        calls = list(self.docker.calls)
        result = self.agent.execute(operation)
        self.assertEqual(result["state"], "completed")
        self.assertEqual(json.loads(result["result"]), {"files": [{"path": "configs/server.cfg", "size": 8, "sha256": module.digest(b"original")}]})
        self.assertEqual(self.docker.calls, calls)

    def test_reusing_id_with_different_payload_is_rejected(self):
        self.agent.execute(self.operation("host.inspect"))
        with self.assertRaisesRegex(AgentError, "different content"):
            self.agent.execute(self.operation("host.inspect", {"changed": True}))

    def test_agent_restart_marks_running_unknown_and_never_replays(self):
        operation = self.operation("instance.restart")
        self.agent.journal.accept(operation)
        self.agent.journal.update(operation["id"], "running")
        self.agent.journal.close()
        self.agent = Agent(self.config, self.docker)
        result = self.agent.execute(operation)
        self.assertEqual(result["state"], "unknown")
        self.assertEqual(self.docker.calls, [])

    def test_agent_restart_resumes_runtime_pull_even_when_an_older_tag_is_cached(self):
        image = "ghcr.io/ayymoss/plutainer:latest"
        operation = self.operation("runtime.pull", {"image": image})
        self.agent.journal.accept(operation)
        self.agent.journal.update(operation["id"], "running")
        aliases = self.agent.root / "agent/runtime-images.json"
        module.atomic_json(aliases, {image: "sha256:older-image"})
        self.agent.journal.close()
        self.agent = Agent(self.config, self.docker)
        original = self.docker.call

        def call(*args, **kwargs):
            result = original(*args, **kwargs)
            return "sha256:new-image" if args[:2] == ("image", "inspect") else result

        self.docker.call = call
        result = self.agent.execute(operation)
        self.assertEqual(result["state"], "completed")
        self.assertEqual(json.loads(result["result"]), {"image": image, "ready": True, "reconciled": True})
        self.assertEqual(json.loads(aliases.read_text())[image], "sha256:new-image")
        self.agent.execute(operation)
        self.assertEqual(sum(args[:2] == ("image", "pull") for args in self.docker.calls), 1)

    def test_interrupted_runtime_pull_failure_is_reported_for_automatic_retry(self):
        operation = self.operation("runtime.pull", {"image": "ghcr.io/ayymoss/plutainer:latest"})
        self.agent.journal.accept(operation)
        self.agent.journal.update(operation["id"], "running")
        self.agent.journal.close()
        self.agent = Agent(self.config, self.docker)
        with patch.object(self.docker, "call", side_effect=AgentError("Registry unavailable")):
            result = self.agent.execute(operation)
        self.assertEqual(result["state"], "failed")
        self.assertIn("runtime download", result["error"])

    def test_occupied_and_unknown_wait_then_revalidate_same_operation(self):
        self.create()
        operation = self.operation("instance.restart")
        for players in (2, None, 7):
            self.agent.rcon = lambda *_, count=players: {"ok": True, "players": count}
            result = self.agent.execute(operation)
            self.assertEqual(result["state"], "accepted")
            self.assertFalse(any(call[:2] == ("container", "restart") for call in self.docker.calls))
        self.agent.rcon = lambda *_: {"ok": True, "players": 0}
        self.assertEqual(self.agent.execute(operation)["state"], "completed")

    def test_occupancy_hold_explains_local_rcon_failure_without_exposing_reply(self):
        self.create()
        operation = self.operation("instance.restart")
        for reason, detail in (
            ("unconfigured", "password is not configured"),
            ("unreachable", "endpoint is unreachable"),
            ("invalid-reply", "invalid packet prefix"),
            ("timeout", "query timed out"),
            ("authentication", "authentication failed"),
            (None, "no recognized player count"),
        ):
            with self.subTest(reason=reason):
                self.agent.rcon = lambda *_: {"ok": reason is None, "reason": reason, "players": None, "text": "sensitive raw reply"}
                result = self.agent.execute(operation)
                self.assertEqual(result["state"], "accepted")
                message = json.loads(result["result"])["message"]
                self.assertIn(detail, message)
                self.assertNotIn("sensitive raw reply", message)
                self.assertFalse(any(call[:2] == ("container", "restart") for call in self.docker.calls))
        self.agent.rcon = lambda *_: {"ok": True, "players": 0}
        self.assertEqual(self.agent.execute(operation)["state"], "completed")

    def test_unknown_container_health_cannot_disrupt(self):
        self.create()
        self.docker.containers[self.agent.container_name(INSTANCE, 1)]["State"] = {"Status": "unknown", "Running": False}
        self.agent.rcon = lambda *_: {"ok": True, "players": 0}
        self.assertEqual(self.agent.execute(self.operation("instance.stop"))["state"], "accepted")
        self.assertFalse(any(call[:2] == ("container", "stop") for call in self.docker.calls))

    def test_files_stage_is_nondisruptive_activation_snapshots_and_rollback(self):
        self.create()
        target = self.agent.instance_dir(INSTANCE, 1) / "app/configs/server.cfg"
        staged = {"stageId": "campaign", "files": [{"id": 9, "path": "configs/server.cfg", "contentBase64": base64.b64encode(b"new").decode(), "sha256": module.digest(b"new"), "checkBefore": True, "expectedBeforeSha": module.digest(b"original")}]}
        self.agent.execute(self.operation("files.stage", staged, "stage"))
        self.assertEqual(target.read_bytes(), b"original")
        self.agent.rcon = lambda *_: {"ok": True, "players": 3}
        operation = self.operation("instance.restart", {"stageId": "campaign"})
        self.assertEqual(self.agent.execute(operation)["state"], "accepted")
        self.assertEqual(target.read_bytes(), b"original")
        self.agent.rcon = lambda *_: {"ok": True, "players": 0}
        result = json.loads(self.agent.execute(operation)["result"])
        self.assertEqual(base64.b64decode(result["before"][0]["contentBase64"]), b"original")
        self.assertEqual(target.read_bytes(), b"new")
        rollback = self.agent.execute(self.operation("files.rollback", {"stageId": "campaign", "restart": False}, "rollback"))
        self.assertEqual(rollback["state"], "completed")
        self.assertEqual(target.read_bytes(), b"original")

    def test_changed_target_fails_before_activation_or_restart(self):
        self.create()
        target = self.agent.instance_dir(INSTANCE, 1) / "app/configs/server.cfg"
        self.agent.execute(self.operation("files.stage", {"stageId": "campaign", "files": [{"path": "configs/server.cfg", "contentBase64": base64.b64encode(b"new").decode(), "checkBefore": True, "expectedBeforeSha": "0" * 64}]}, "stage"))
        self.agent.rcon = lambda *_: {"ok": True, "players": 0}
        result = self.agent.execute(self.operation("instance.restart", {"stageId": "campaign"}))
        self.assertEqual(result["state"], "failed")
        self.assertEqual(target.read_bytes(), b"original")
        self.assertFalse(any(call[:2] == ("container", "restart") for call in self.docker.calls))

    def test_ambiguous_restart_keeps_before_images_on_disk_and_is_not_replayed(self):
        self.create()
        self.agent.rcon = lambda *_: {"ok": True, "players": 0}
        self.agent.execute(self.operation("files.stage", {"stageId": "campaign", "files": [{"path": "configs/server.cfg", "contentBase64": base64.b64encode(b"new").decode()}]}, "stage"))
        self.docker.fail_restart = True
        operation = self.operation("instance.restart", {"stageId": "campaign"})
        self.assertEqual(self.agent.execute(operation)["state"], "unknown")
        self.agent.execute(operation)
        self.assertEqual(sum(call[:2] == ("container", "restart") for call in self.docker.calls), 1)
        saved = json.loads((self.agent.instance_dir(INSTANCE, 1) / "stages/campaign/manifest.json").read_text())
        self.assertEqual(saved["state"], "activated")
        self.assertEqual(base64.b64decode(saved["before"][0]["contentBase64"]), b"original")

    def test_remove_only_own_instance_and_preserves_shared_library(self):
        self.create()
        self.agent.rcon = lambda *_: {"ok": True, "players": 0}
        result = self.agent.execute(self.operation("instance.remove"))
        self.assertEqual(result["state"], "completed")
        self.assertTrue((self.library / "content/game.ff").exists())
        self.assertFalse(self.agent.instance_dir(INSTANCE, 1).exists())

    def test_cancel_creation_bypasses_occupancy_and_preserves_library(self):
        self.create()
        self.agent.rcon = lambda *_: self.fail("Cancellation must not wait for player evidence")
        operation = self.operation("instance.cancel-create", {"createOperationId": "create"}, "cancel")
        self.assertEqual(self.agent.execute(operation)["state"], "completed")
        self.assertFalse(self.agent.inspect(INSTANCE, 1)["exists"])
        self.assertFalse(self.agent.instance_dir(INSTANCE, 1).exists())
        self.assertTrue((self.library / "content/game.ff").exists())
        self.assertEqual(self.agent.execute(operation)["state"], "completed")
        self.assertEqual(sum(call[:2] == ("container", "rm") for call in self.docker.calls), 1)

    def test_heartbeat_cancellation_stops_create_before_starting_container(self):
        original_call = self.docker.call
        def cancel_during_create(*args, **kwargs):
            result = original_call(*args, **kwargs)
            if args[:2] == ("container", "create"):
                self.agent.cancelled_creations = frozenset(["create"])
            return result
        self.docker.call = cancel_during_create
        payload = {"name": "test", "game": "t5", "libraryId": self.library_id, "image": "ghcr.io/ayymoss/plutainer:latest", "port": 28960, "env": {}, "rconPassword": "secret", "files": []}
        self.assertEqual(self.agent.execute(self.operation("instance.create", payload, "create"))["state"], "unknown")
        self.assertFalse(any(call[:2] == ("container", "start") for call in self.docker.calls))
        result = self.agent.execute(self.operation("instance.cancel-create", {"createOperationId": "create"}))
        self.assertEqual(result["state"], "completed")
        self.assertFalse(self.agent.inspect(INSTANCE, 1)["exists"])

    def test_cancellation_cleanup_resumes_after_agent_restart(self):
        self.create()
        self.docker.lose_reply_after = "rm"
        cancel = self.operation("instance.cancel-create", {"createOperationId": "create"}, "cancel")
        self.assertEqual(self.agent.execute(cancel)["state"], "unknown")
        self.agent.journal.db.close()
        self.agent = Agent(self.config, self.docker)
        self.assertEqual(self.agent.execute(cancel)["state"], "completed")
        self.assertFalse(self.agent.instance_dir(INSTANCE, 1).exists())

    def test_cancel_before_creation_delivery_fences_delayed_create(self):
        result = self.agent.execute(self.operation("instance.cancel-create", {"createOperationId": "late-create"}))
        self.assertEqual(result["state"], "completed", result)
        result = self.agent.execute(self.operation("instance.create", {}, "late-create"))
        self.assertEqual(result["state"], "failed")
        self.assertFalse(any(call[:2] == ("container", "create") for call in self.docker.calls))

    def test_cancel_removes_partial_files_without_a_container(self):
        directory = self.agent.instance_dir(INSTANCE, 1)
        (directory / "app").mkdir(parents=True)
        (directory / "app/partial.cfg").write_text("partial")
        self.agent.journal.accept(self.operation("instance.create", {}, "create"))
        result = self.agent.execute(self.operation("instance.cancel-create", {"createOperationId": "create"}))
        self.assertEqual(result["state"], "completed", result)
        self.assertFalse(directory.exists())
        self.assertTrue((self.library / "content/game.ff").exists())

    def test_cancel_rejects_a_creation_id_from_another_instance(self):
        other = self.operation("instance.create", {}, "other-create")
        other["instanceId"] = "another-instance"
        self.agent.journal.accept(other)
        result = self.agent.execute(self.operation("instance.cancel-create", {"createOperationId": "other-create"}))
        self.assertEqual(result["state"], "failed", result)
        self.assertFalse((self.agent.instance_dir(INSTANCE, 1).parent / "current.json").exists())

    def test_cancel_creation_rejects_other_creation_and_newer_generation(self):
        self.create()
        result = self.agent.execute(self.operation("instance.cancel-create", {"createOperationId": "wrong"}))
        self.assertEqual(result["state"], "failed")
        module.atomic_json(self.agent.instance_dir(INSTANCE, 1).parent / "current.json", {"generation": 2})
        result = self.agent.execute(self.operation("instance.cancel-create", {"createOperationId": "create"}, "old-generation"))
        self.assertEqual(result["state"], "failed")
        self.assertTrue(self.agent.inspect(INSTANCE, 1)["running"])
        self.assertFalse(any(call[:2] == ("container", "rm") for call in self.docker.calls))

    def test_cancel_creation_reconciles_lost_removal_reply_and_fences_unknown_create(self):
        self.docker.lose_reply_after = "start"
        payload = {"name": "test", "game": "t5", "libraryId": self.library_id, "image": "ghcr.io/ayymoss/plutainer:latest", "port": 28960, "env": {}, "rconPassword": "secret", "files": []}
        create = self.operation("instance.create", payload, "create")
        self.assertEqual(self.agent.execute(create)["state"], "unknown")
        self.docker.lose_reply_after = "rm"
        cancel = self.operation("instance.cancel-create", {"createOperationId": "create"}, "cancel")
        self.assertEqual(self.agent.execute(cancel)["state"], "unknown")
        self.assertEqual(self.agent.execute(cancel)["state"], "completed")
        self.assertEqual(self.agent.execute(create)["state"], "failed")
        self.assertFalse(self.agent.instance_dir(INSTANCE, 1).exists())
        self.assertTrue(json.loads((self.agent.instance_dir(INSTANCE, 1).parent / "current.json").read_text())["removed"])

    def test_cancel_creation_rejects_unowned_container(self):
        self.create()
        self.docker.containers[self.agent.container_name(INSTANCE, 1)]["Config"]["Labels"] = {}
        result = self.agent.execute(self.operation("instance.cancel-create", {"createOperationId": "create"}))
        self.assertEqual(result["state"], "failed")
        self.assertFalse(any(call[:2] == ("container", "rm") for call in self.docker.calls))

    def test_unowned_container_never_mutated(self):
        self.create()
        self.docker.containers[self.agent.container_name(INSTANCE, 1)]["Config"]["Labels"] = {}
        self.assertEqual(self.agent.execute(self.operation("instance.remove"))["state"], "failed")
        self.assertFalse(any(call[:2] == ("container", "stop") for call in self.docker.calls))

    def test_rebuild_moves_generation_and_rejects_old_commands(self):
        payload = self.create()
        self.agent.rcon = lambda *_: {"ok": True, "players": 0}
        result = self.agent.execute(self.operation("instance.rebuild", {**payload, "nextGeneration": 2}))
        self.assertEqual(result["state"], "completed", result)
        self.assertEqual(json.loads(result["result"])["generation"], 2)
        self.assertEqual(self.agent.execute(self.operation("instance.stop", op_id="old-stop"))["state"], "failed")
        self.assertTrue(self.agent.inspect(INSTANCE, 2)["running"])

    def test_traversal_and_symlinks_are_rejected(self):
        self.create()
        app = self.agent.instance_dir(INSTANCE, 1) / "app"
        (app / "escape").symlink_to(self.temp.name)
        for path in ("../spec.json", "/etc/passwd", "escape/secret", "configs/../../spec.json"):
            with self.assertRaises(AgentError):
                module.scoped(app, path)

    def manifest(self, content=b"complete-game-file"):
        return self.manifest_files({"zone/test.ff": content})

    def manifest_files(self, contents):
        files = [{"path": path, "size": len(content), "sha256": module.digest(content)} for path, content in contents.items()]
        return {"id": "downloaded", "game": "t6", "files": files, "sizeBytes": sum(item["size"] for item in files), "manifestSha256": module.digest(json.dumps(files, separators=(",", ":")).encode())}

    def test_downloads_use_bounded_parallel_transfers_and_aggregate_progress(self):
        contents = {f"zone/file-{index}.ff": bytes([index]) * 31 for index in range(8)}
        manifest = self.manifest_files(contents)
        barrier = threading.Barrier(4)
        lock = threading.Lock()
        active = maximum = 0
        snapshots = []
        agent = self.agent

        class ConcurrentResponse(Response):
            def __exit__(self, *args):
                nonlocal active
                with lock:
                    active -= 1
                    snapshots.append(agent.library_progress["downloaded"]["bytesDownloaded"])
                return super().__exit__(*args)

        def fetch(method, path, data=None, extra_headers=None):
            nonlocal active, maximum
            if path == "/manifest":
                return Response(json.dumps(manifest).encode())
            with lock:
                active += 1
                maximum = max(maximum, active)
            # A serial downloader cannot reach this barrier. Two batches also
            # prove that an unlimited worker pool cannot open all eight files.
            barrier.wait(timeout=5)
            return ConcurrentResponse(contents[path.removeprefix("/files/")])

        self.agent.artifact_open = fetch
        result = self.agent.library_download({"libraryVersionId": "downloaded", "manifestUrl": "/manifest", "filesBaseUrl": "/files"})
        self.assertEqual(maximum, 4)
        self.assertEqual(snapshots, sorted(snapshots))
        self.assertEqual(max(snapshots), manifest["sizeBytes"])
        self.assertEqual(result["size"], manifest["sizeBytes"])
        for path, content in contents.items():
            self.assertEqual((self.agent.library_dir("t6", "downloaded") / "content" / path).read_bytes(), content)

    def test_all_downloads_finish_before_any_file_is_hashed_and_verified_from_disk(self):
        contents = {"zone/first.ff": b"game-file" * 200000, "zone/second.ff": b"second-file"}
        manifest = self.manifest_files(contents)
        self.agent.artifact_open = lambda method, path, **kwargs: Response(json.dumps(manifest).encode() if path == "/manifest" else contents[path.removeprefix("/files/")])
        original_hash, original_open = module.hashlib.sha256, Path.open
        reads = []

        def hashing(*args, **kwargs):
            progress = self.agent.library_progress.get("downloaded")
            if progress:
                self.assertEqual(progress["status"], "verifying")
                self.assertEqual(progress["bytesDownloaded"], manifest["sizeBytes"])
            return original_hash(*args, **kwargs)

        def opening(path, mode="r", *args, **kwargs):
            if mode == "rb" and path.name.endswith(".ff"):
                self.assertFalse(self.agent.library_dir("t6", "downloaded").exists())
                reads.append(path.name)
            return original_open(path, mode, *args, **kwargs)

        with patch.object(module.hashlib, "sha256", side_effect=hashing), patch.object(Path, "open", opening):
            result = self.agent.library_download({"libraryVersionId": "downloaded", "manifestUrl": "/manifest", "filesBaseUrl": "/files"})
        self.assertEqual(result["state"], "ready")
        self.assertCountEqual(reads, ["first.ff", "second.ff"])
        cache = next(item for item in self.agent.inventory()["libraryCaches"] if item["libraryVersionId"] == "downloaded")
        self.assertEqual(cache["bytesVerified"], manifest["sizeBytes"])
        self.assertEqual(cache["transfer"]["phase"], "verifying")

    def test_phase_boundaries_are_reported_even_between_heartbeats_and_status_errors_are_nonfatal(self):
        content = b"game-file"
        manifest = self.manifest(content)
        reports = []

        def fetch(method, path, data=None, extra_headers=None, timeout=60):
            if path == "/status":
                self.assertEqual(timeout, 5)
                reports.append(json.loads(data))
                raise AgentError("Controller temporarily unavailable")
            self.assertEqual(timeout, 60)
            return Response(json.dumps(manifest).encode() if path == "/manifest" else content)

        self.agent.artifact_open = fetch
        result = self.agent.library_download({"libraryVersionId": "downloaded", "manifestUrl": "/manifest", "filesBaseUrl": "/files", "statusUrl": "/status"})
        self.assertEqual(result["state"], "ready")
        self.assertEqual([item["status"] for item in reports], ["downloading", "verifying", "ready"])
        self.assertEqual([item["bytesVerified"] for item in reports], [0, 0, len(content)])
        self.assertEqual(reports[1]["bytesDownloaded"], len(content))

    def test_status_socket_timeout_is_short_without_reducing_download_timeout(self):
        with patch.object(module.urllib.request, "build_opener") as build_opener:
            self.agent.artifact_open("GET", "/file")
            self.assertEqual(build_opener.return_value.open.call_args.kwargs["timeout"], 60)
            self.agent.artifact_open("POST", "/status", b"{}", timeout=5)
            self.assertEqual(build_opener.return_value.open.call_args.kwargs["timeout"], 5)

    def test_ignored_range_restarts_file_and_checksum(self):
        content = b"complete-game-file"
        manifest = self.manifest(content)
        staging = self.agent.root / "libraries/t6/.downloaded.partial/content/zone"
        staging.mkdir(parents=True)
        (staging / "test.ff").write_bytes(b"bad-prefix")

        def fetch(method, path, data=None, extra_headers=None):
            if path == "/manifest":
                return Response(json.dumps(manifest).encode())
            self.assertEqual(extra_headers, {"Range": "bytes=10-"})
            return Response(content)

        self.agent.artifact_open = fetch
        result = self.agent.library_download({"libraryVersionId": "downloaded", "manifestUrl": "/manifest", "filesBaseUrl": "/files"})
        self.assertEqual(result["state"], "ready")
        self.assertEqual((self.agent.library_dir("t6", "downloaded") / "content/zone/test.ff").read_bytes(), content)

    def test_resumed_download_verifies_saved_prefix(self):
        content = b"complete-game-file"
        manifest = self.manifest(content)
        staging = self.agent.root / "libraries/t6/.downloaded.partial/content/zone"
        staging.mkdir(parents=True)
        (staging / "test.ff").write_bytes(b"corrupt!")
        self.agent.artifact_open = lambda method, path, **kwargs: Response(json.dumps(manifest).encode()) if path == "/manifest" else Response(content[8:], 206, {"Content-Range": f"bytes 8-{len(content)-1}/{len(content)}"})
        with self.assertRaisesRegex(AgentError, "integrity"):
            self.agent.library_download({"libraryVersionId": "downloaded", "manifestUrl": "/manifest", "filesBaseUrl": "/files"})
        self.assertFalse(self.agent.library_dir("t6", "downloaded").exists())
        self.assertFalse((staging / "test.ff").exists())
        self.assertEqual(self.agent.library_progress["downloaded"]["bytesDownloaded"], 0)
        requests = []

        def fetch(method, path, data=None, extra_headers=None):
            if path != "/manifest":
                requests.append(extra_headers)
            return Response(json.dumps(manifest).encode() if path == "/manifest" else content)

        self.agent.artifact_open = fetch
        self.assertEqual(self.agent.library_download({"libraryVersionId": "downloaded", "manifestUrl": "/manifest", "filesBaseUrl": "/files"})["state"], "ready")
        self.assertEqual(requests, [{}])

    def test_all_corrupt_complete_files_are_removed_so_retry_downloads_them_again(self):
        contents = {"zone/first.ff": b"first", "zone/second.ff": b"second", "zone/good.ff": b"good"}
        manifest = self.manifest_files(contents)
        staging = self.agent.root / "libraries/t6/.downloaded.partial/content/zone"
        staging.mkdir(parents=True)
        for name, content in contents.items():
            (staging / Path(name).name).write_bytes(content if name.endswith("good.ff") else b"x" * len(content))
        requests = []

        def fetch(method, path, data=None, extra_headers=None):
            requests.append(path)
            return Response(json.dumps(manifest).encode() if path == "/manifest" else contents[path.removeprefix("/files/")])

        self.agent.artifact_open = fetch
        payload = {"libraryVersionId": "downloaded", "manifestUrl": "/manifest", "filesBaseUrl": "/files"}
        with self.assertRaisesRegex(AgentError, "integrity"):
            self.agent.library_download(payload)
        self.assertEqual(requests, ["/manifest"])
        self.assertEqual([path.name for path in staging.iterdir()], ["good.ff"])
        self.assertEqual(self.agent.library_download(payload)["state"], "ready")
        self.assertCountEqual(requests, ["/manifest", "/manifest", "/files/zone/first.ff", "/files/zone/second.ff"])

    def test_verification_keeps_download_phase_complete_until_failure_reports_retained_bytes(self):
        contents = {"zone/first.ff": b"first", "zone/second.ff": b"second"}
        for mode in ("staged-corrupt", "existing-missing", "existing-short"):
            with self.subTest(mode=mode):
                manifest = {**self.manifest_files(contents), "id": mode}
                directory = self.agent.library_dir("t6", mode)
                if mode.startswith("existing"):
                    (directory / "content/zone").mkdir(parents=True)
                    (directory / "content/zone/second.ff").write_bytes(contents["zone/second.ff"])
                    if mode == "existing-short":
                        (directory / "content/zone/first.ff").write_bytes(b"f")
                    module.atomic_json(directory / "manifest.json", {"libraryId": mode, "game": "t6", "size": manifest["sizeBytes"], "files": manifest["files"], "manifestSha256": manifest["manifestSha256"]})

                def fetch(method, path, **kwargs):
                    if path == "/manifest":
                        return Response(json.dumps(manifest).encode())
                    content = contents[path.removeprefix("/files/")]
                    return Response(b"x" * len(content) if path.endswith("first.ff") else content)

                self.agent.artifact_open = fetch
                original_open = Path.open
                snapshots = []

                def opening(path, open_mode="r", *args, **kwargs):
                    open_mode = kwargs.pop("mode", open_mode)
                    if open_mode == "rb" and path.name == "second.ff":
                        cache = next(item for item in self.agent.inventory()["libraryCaches"] if item["libraryVersionId"] == mode)
                        snapshots.append(cache)
                        self.assertEqual(cache["status"], "verifying")
                        self.assertEqual(cache["bytesDownloaded"], cache["bytesTotal"])
                        self.assertEqual(cache["transfer"]["phase"], "verifying")
                    return original_open(path, open_mode, *args, **kwargs)

                with patch.object(Path, "open", opening), self.assertRaisesRegex(AgentError, "integrity"):
                    self.agent.library_download({"libraryVersionId": mode, "manifestUrl": "/manifest", "filesBaseUrl": "/files"})
                self.assertEqual(len(snapshots), 1)
                failed = next(item for item in self.agent.inventory()["libraryCaches"] if item["libraryVersionId"] == mode)
                self.assertEqual(failed["status"], "failed")
                self.assertEqual(failed["bytesDownloaded"], len(contents["zone/second.ff"]) + (1 if mode == "existing-short" else 0))

    def test_parallel_failure_retains_partial_files_for_verified_resume(self):
        contents = {"zone/first.ff": b"first-game-file", "zone/second.ff": b"second-game-file"}
        manifest = self.manifest_files(contents)
        payload = {"libraryVersionId": "downloaded", "manifestUrl": "/manifest", "filesBaseUrl": "/files"}

        def failed_fetch(method, path, data=None, extra_headers=None):
            if path == "/manifest":
                return Response(json.dumps(manifest).encode())
            return Response(contents[path.removeprefix("/files/")][:5])

        self.agent.artifact_open = failed_fetch
        with self.assertRaises(AgentError):
            self.agent.library_download(payload)
        self.assertFalse(self.agent.library_dir("t6", "downloaded").exists())
        self.assertEqual(self.agent.library_progress["downloaded"]["status"], "failed")
        resumed = []

        def resume_fetch(method, path, data=None, extra_headers=None):
            if path == "/manifest":
                return Response(json.dumps(manifest).encode())
            content = contents[path.removeprefix("/files/")]
            if extra_headers:
                self.assertEqual(extra_headers, {"Range": "bytes=5-"})
                resumed.append(path)
                return Response(content[5:], 206, {"Content-Range": f"bytes 5-{len(content)-1}/{len(content)}"})
            return Response(content)

        self.agent.artifact_open = resume_fetch
        self.assertEqual(self.agent.library_download(payload)["state"], "ready")
        self.assertTrue(resumed)
        self.assertEqual(self.agent.verify_library(self.agent.library_dir("t6", "downloaded"))["size"], manifest["sizeBytes"])

    def test_overlapping_paths_rejected_before_concurrent_downloads(self):
        manifest = self.manifest_files({"zone": b"file", "zone/child.ff": b"child"})
        calls = []

        def fetch(method, path, data=None, extra_headers=None):
            calls.append(path)
            return Response(json.dumps(manifest).encode())

        self.agent.artifact_open = fetch
        with self.assertRaisesRegex(AgentError, "overlap"):
            self.agent.library_download({"libraryVersionId": "downloaded", "manifestUrl": "/manifest", "filesBaseUrl": "/files"})
        self.assertEqual(calls, ["/manifest"])

    def test_failed_transfer_restart_reports_retained_bytes_and_resumes_with_only_remaining_space(self):
        content = b"complete-game-file"
        manifest = self.manifest(content)
        payload = {"libraryVersionId": "downloaded", "manifestUrl": "/manifest", "filesBaseUrl": "/files"}
        self.agent.artifact_open = lambda method, path, **kwargs: Response(json.dumps(manifest).encode() if path == "/manifest" else content[:8])
        with self.assertRaises(AgentError):
            self.agent.library_download(payload)
        self.agent.journal.close()
        self.agent = Agent(self.config, self.docker)
        cache = next(item for item in self.agent.inventory()["libraryCaches"] if item["libraryVersionId"] == "downloaded")
        self.assertEqual(cache["status"], "failed")
        self.assertEqual(cache["bytesDownloaded"], 8)
        self.assertEqual(cache["bytesVerified"], 0)
        self.assertEqual(cache["bytesTotal"], len(content))
        self.assertEqual(cache["manifestSha256"], manifest["manifestSha256"])
        self.assertEqual(cache["transfer"]["phase"], "downloading")
        self.assertIsNone(cache["transfer"]["etaSeconds"])

        def fetch(method, path, data=None, extra_headers=None):
            if path == "/manifest":
                return Response(json.dumps(manifest).encode())
            self.assertEqual(extra_headers, {"Range": "bytes=8-"})
            return Response(content[8:], 206, {"Content-Range": f"bytes 8-{len(content)-1}/{len(content)}"})

        self.agent.artifact_open = fetch
        usage = module.shutil.disk_usage(self.agent.root)
        with patch.object(module.shutil, "disk_usage", return_value=type(usage)(usage.total, usage.used, len(content) - 8)):
            self.assertEqual(self.agent.library_download(payload)["state"], "ready")

    def test_unpromoted_manifest_is_never_reported_as_ready_and_progress_is_not_duplicated(self):
        content = b"complete-game-file"
        manifest = self.manifest(content)
        staging = self.agent.root / "libraries/t6/.downloaded.partial"
        (staging / "content/zone").mkdir(parents=True)
        # Oversized retained files can be replaced during resume. Their progress
        # must still stay within the published file size.
        (staging / "content/zone/test.ff").write_bytes(content + b"extra")
        saved = {"libraryId": "downloaded", "game": "t6", "size": len(content), "files": manifest["files"], "manifestSha256": manifest["manifestSha256"]}
        module.atomic_json(staging / "source.json", saved)
        module.atomic_json(staging / "manifest.json", saved)
        inventory = self.agent.inventory()
        caches = [item for item in inventory["libraryCaches"] if item["libraryVersionId"] == "downloaded"]
        self.assertEqual(len(caches), 1)
        self.assertEqual(caches[0]["status"], "missing")
        self.assertEqual(caches[0]["bytesDownloaded"], len(content))
        self.assertNotIn({"id": "downloaded", "state": "ready"}, inventory["libraries"])
        self.agent.library_progress["downloaded"] = {**caches[0], "status": "downloading", "bytesDownloaded": 3}
        caches = [item for item in self.agent.inventory()["libraryCaches"] if item["libraryVersionId"] == "downloaded"]
        self.assertEqual(len(caches), 1)
        self.assertEqual(caches[0]["status"], "downloading")
        self.assertEqual(caches[0]["bytesDownloaded"], 3)

    def test_interrupted_verification_restarts_from_disk_without_redownloading_and_keeps_elapsed(self):
        content = b"complete-game-file"
        manifest = self.manifest(content)
        payload = {"libraryVersionId": "downloaded", "manifestUrl": "/manifest", "filesBaseUrl": "/files"}
        self.agent.artifact_open = lambda method, path, **kwargs: Response(json.dumps(manifest).encode() if path == "/manifest" else content)

        def interrupted(files, targets, transfer, repair=False):
            transfer.advance("bytesVerified", 5, 5)
            transfer.persist(force=True)
            raise KeyboardInterrupt("Simulated process termination")

        with patch.object(self.agent, "verify_downloaded_files", side_effect=interrupted), self.assertRaises(KeyboardInterrupt):
            self.agent.library_download(payload)
        previous = self.agent.library_transfers["downloaded"].snapshot()
        self.assertFalse(self.agent.library_dir("t6", "downloaded").exists())
        self.agent.journal.close()
        self.agent = Agent(self.config, self.docker)
        interrupted_cache = next(item for item in self.agent.inventory()["libraryCaches"] if item["libraryVersionId"] == "downloaded")
        self.assertEqual(interrupted_cache["status"], "missing")
        self.assertEqual(interrupted_cache["bytesDownloaded"], len(content))
        self.assertEqual(interrupted_cache["bytesVerified"], 5)
        requests = []

        def fetch(method, path, **kwargs):
            requests.append(path)
            return Response(json.dumps(manifest).encode())

        original_verify = self.agent.verify_downloaded_files

        def verifying(files, targets, transfer, repair=False):
            self.assertEqual(transfer.snapshot()["bytesVerified"], 0)
            self.assertIsNone(transfer.snapshot()["transfer"]["etaSeconds"])
            return original_verify(files, targets, transfer, repair)

        self.agent.artifact_open = fetch
        with patch.object(self.agent, "verify_downloaded_files", side_effect=verifying):
            self.assertEqual(self.agent.library_download(payload)["state"], "ready")
        self.assertEqual(requests, ["/manifest"])
        cache = next(item for item in self.agent.inventory()["libraryCaches"] if item["libraryVersionId"] == "downloaded")
        self.assertEqual(cache["bytesVerified"], len(content))
        self.assertEqual(cache["transfer"]["startedAt"], previous["transfer"]["startedAt"])
        self.assertGreaterEqual(cache["transfer"]["elapsedSeconds"], interrupted_cache["transfer"]["elapsedSeconds"])
        self.assertGreaterEqual(cache["transfer"]["phaseElapsedSeconds"], interrupted_cache["transfer"]["phaseElapsedSeconds"])
        self.agent.journal.close()
        self.agent = Agent(self.config, self.docker)
        self.assertEqual(next(item for item in self.agent.inventory()["libraryCaches"] if item["libraryVersionId"] == "downloaded"), cache)

    def test_crash_before_atomic_promotion_reverifies_read_only_staging_without_network_transfers(self):
        content = b"complete-game-file"
        manifest = self.manifest(content)
        payload = {"libraryVersionId": "downloaded", "manifestUrl": "/manifest", "filesBaseUrl": "/files"}
        self.agent.artifact_open = lambda method, path, **kwargs: Response(json.dumps(manifest).encode() if path == "/manifest" else content)
        original_replace = module.os.replace

        def replace(source, destination):
            if Path(destination) == self.agent.library_dir("t6", "downloaded"):
                raise KeyboardInterrupt("Simulated process termination before promotion")
            return original_replace(source, destination)

        with patch.object(module.os, "replace", side_effect=replace), self.assertRaises(KeyboardInterrupt):
            self.agent.library_download(payload)
        staging = self.agent.root / "libraries/t6/.downloaded.partial/content"
        self.assertEqual(staging.stat().st_mode & 0o777, 0o555)
        self.agent.journal.close()
        self.agent = Agent(self.config, self.docker)
        requests = []

        def fetch(method, path, **kwargs):
            requests.append(path)
            return Response(json.dumps(manifest).encode())

        self.agent.artifact_open = fetch
        self.assertEqual(self.agent.library_download(payload)["state"], "ready")
        self.assertEqual(requests, ["/manifest"])
        self.assertEqual((self.agent.library_dir("t6", "downloaded") / "content/zone/test.ff").read_bytes(), content)

    def test_existing_cache_reverification_is_visible_and_failure_revokes_readiness_after_restart(self):
        saved = json.loads((self.library / "manifest.json").read_text())
        manifest = {**saved, "id": self.library_id}
        self.agent.artifact_open = lambda *args, **kwargs: Response(json.dumps(manifest).encode())
        original_verify = self.agent.verify_downloaded_files

        def verifying(files, targets, transfer, repair=False):
            inventory = self.agent.inventory()
            cache = next(item for item in inventory["libraryCaches"] if item["libraryVersionId"] == self.library_id)
            self.assertEqual(cache["status"], "verifying")
            self.assertEqual(cache["bytesVerified"], 0)
            self.assertNotIn({"id": self.library_id, "state": "ready"}, inventory["libraries"])
            return original_verify(files, targets, transfer, repair)

        (self.library / "content/game.ff").write_bytes(b"corrupt!!")
        with patch.object(self.agent, "verify_downloaded_files", side_effect=verifying), self.assertRaisesRegex(AgentError, "integrity"):
            self.agent.library_download({"libraryVersionId": self.library_id, "manifestUrl": "/manifest", "filesBaseUrl": "/files"})
        self.agent.journal.close()
        self.agent = Agent(self.config, self.docker)
        inventory = self.agent.inventory()
        self.assertNotIn({"id": self.library_id, "state": "ready"}, inventory["libraries"])
        self.assertEqual(next(item for item in inventory["libraryCaches"] if item["libraryVersionId"] == self.library_id)["status"], "failed")

    def test_elapsed_and_phase_eta_use_new_work_only_and_freeze_at_completion_or_failure(self):
        for terminal_status in ("ready", "failed"):
            with self.subTest(terminal_status=terminal_status):
                now = [100.0]
                progress = {"libraryVersionId": "downloaded", "status": "downloading", "bytesDownloaded": 80, "bytesVerified": 0, "bytesTotal": 100}
                previous = {"transfer": {"phase": "downloading", "startedAt": "2026-10-04T00:00:00Z", "elapsedSeconds": 20, "phaseElapsedSeconds": 20}}
                with patch.object(module.time, "monotonic", side_effect=lambda: now[0]):
                    transfer = module.LibraryTransfer(progress, self.agent.root / "agent/timing.json", previous)
                    self.assertIsNone(transfer.snapshot()["transfer"]["etaSeconds"])
                    now[0] += 0.5
                    transfer.advance("bytesDownloaded", 5, 5)
                    self.assertIsNone(transfer.snapshot()["transfer"]["etaSeconds"])
                    now[0] += 1.5
                    snapshot = transfer.snapshot()["transfer"]
                    self.assertEqual(snapshot["bytesPerSecond"], 2.5)
                    self.assertEqual(snapshot["etaSeconds"], 6)
                    self.assertEqual(snapshot["elapsedSeconds"], 22)
                    now[0] += 1
                    transfer.advance("bytesDownloaded", 15, 15)
                    transfer.begin_verification()
                    self.assertEqual(transfer.snapshot()["transfer"]["phaseElapsedSeconds"], 0)
                    self.assertIsNone(transfer.snapshot()["transfer"]["etaSeconds"])
                    now[0] += 2
                    transfer.advance("bytesVerified", 40, 40)
                    snapshot = transfer.snapshot()["transfer"]
                    self.assertEqual(snapshot["bytesPerSecond"], 20)
                    self.assertEqual(snapshot["etaSeconds"], 3)
                    self.assertEqual(snapshot["elapsedSeconds"], 25)
                    transfer.finish(terminal_status)
                    final = transfer.snapshot()
                    now[0] += 100
                    self.assertEqual(transfer.snapshot(), final)
                    self.assertEqual(final["transfer"]["etaSeconds"], 0 if terminal_status == "ready" else None)

    def test_partial_inventory_rejects_changed_manifest_and_symlinked_content(self):
        content = b"complete-game-file"
        manifest = self.manifest(content)
        staging = self.agent.root / "libraries/t6/.downloaded.partial"
        (staging / "content/zone").mkdir(parents=True)
        source = {"libraryId": "downloaded", "game": "t6", "files": manifest["files"], "manifestSha256": "0" * 64}
        module.atomic_json(staging / "source.json", source)
        (staging / "content/zone/test.ff").write_bytes(content)
        self.assertFalse(any(item["libraryVersionId"] == "downloaded" for item in self.agent.inventory()["libraryCaches"]))
        source["manifestSha256"] = manifest["manifestSha256"]
        module.atomic_json(staging / "source.json", source)
        (staging / "content/zone/test.ff").unlink()
        (staging / "content/zone/test.ff").symlink_to(self.library / "content/game.ff")
        self.assertFalse(any(item["libraryVersionId"] == "downloaded" for item in self.agent.inventory()["libraryCaches"]))

    def test_resume_verifies_and_atomically_publishes_library(self):
        content = b"complete-game-file"
        manifest = self.manifest(content)
        staging = self.agent.root / "libraries/t6/.downloaded.partial/content/zone"
        staging.mkdir(parents=True)
        (staging / "test.ff").write_bytes(content[:8])
        calls = []
        def fetch(method, path, data=None, extra_headers=None):
            calls.append((path, extra_headers))
            if path == "/manifest":
                return Response(json.dumps(manifest).encode())
            return Response(content[8:], 206, {"Content-Range": f"bytes 8-{len(content)-1}/{len(content)}"})
        self.agent.artifact_open = fetch
        result = self.agent.library_download({"libraryVersionId": "downloaded", "manifestUrl": "/manifest", "filesBaseUrl": "/files"})
        self.assertEqual(result["state"], "ready")
        self.assertEqual(calls[1][1], {"Range": "bytes=8-"})
        self.assertEqual((self.agent.library_dir("t6", "downloaded") / "content/zone/test.ff").read_bytes(), content)
        self.assertEqual(self.agent.inventory()["libraryCaches"][-1]["status"], "ready")

    def test_corrupt_or_incomplete_library_never_promoted(self):
        manifest = self.manifest()
        def fetch(method, path, data=None, extra_headers=None):
            return Response(json.dumps(manifest).encode() if path == "/manifest" else b"corrupt")
        self.agent.artifact_open = fetch
        with self.assertRaisesRegex(AgentError, "integrity"):
            self.agent.library_download({"libraryVersionId": "downloaded", "manifestUrl": "/manifest", "filesBaseUrl": "/files"})
        self.assertFalse(self.agent.library_dir("t6", "downloaded").exists())
        self.assertEqual(self.agent.library_progress["downloaded"]["status"], "failed")

    def test_manifest_tampering_rejected_before_download(self):
        manifest = self.manifest()
        manifest["files"][0]["size"] = 999
        self.agent.artifact_open = lambda *_: Response(json.dumps(manifest).encode())
        with self.assertRaisesRegex(AgentError, "manifest checksum"):
            self.agent.library_download({"libraryVersionId": "downloaded", "manifestUrl": "/manifest", "filesBaseUrl": "/files"})

    def test_artifact_urls_never_leak_credentials_to_other_origins(self):
        for value in ("https://attacker.example/files", "//attacker.example/files", "http://controller.example/files", "https://user:pass@controller.example/files"):
            with self.assertRaises(AgentError):
                self.agent.artifact_url(value)

    def test_unsupported_operation_does_not_spawn_shell(self):
        result = self.agent.execute(self.operation("shell.execute", {"command": "touch /tmp/unsafe"}))
        self.assertEqual(result["state"], "failed")
        self.assertEqual(self.docker.calls, [])

    def test_restart_lost_acknowledgement_recovers_from_changed_start_timestamp(self):
        self.create()
        self.agent.rcon = lambda *_: {"ok": True, "players": 0}
        self.docker.lose_reply_after = "restart"
        operation = self.operation("instance.restart")
        self.assertEqual(self.agent.execute(operation)["state"], "unknown")
        self.agent.journal.close()
        self.agent = Agent(self.config, self.docker)
        recovered = self.agent.execute(operation)
        self.assertEqual(recovered["state"], "completed")
        self.assertTrue(json.loads(recovered["result"])["reconciled"])
        self.assertEqual(sum(call[:2] == ("container", "restart") for call in self.docker.calls), 1)

    def test_created_container_lost_start_reply_reconciles_immutable_operation_label(self):
        payload = {"name": "t5_zombie_test_1", "game": "t5", "libraryId": self.library_id, "image": "ghcr.io/ayymoss/plutainer:latest", "port": 28960, "env": {}, "rconPassword": "secret", "files": []}
        self.docker.lose_reply_after = "start"
        operation = self.operation("instance.create", payload)
        self.assertEqual(self.agent.execute(operation)["state"], "unknown")
        result = self.agent.execute(operation)
        self.assertEqual(result["state"], "completed")
        self.assertEqual(sum(call[:2] == ("container", "create") for call in self.docker.calls), 1)

    def test_remove_lost_reply_reconciles_absence_without_duplicate_delete(self):
        self.create()
        self.agent.rcon = lambda *_: {"ok": True, "players": 0}
        self.docker.lose_reply_after = "rm"
        operation = self.operation("instance.remove")
        self.assertEqual(self.agent.execute(operation)["state"], "unknown")
        self.assertEqual(self.agent.execute(operation)["state"], "completed")
        self.assertFalse(self.agent.instance_dir(INSTANCE, 1).exists())
        self.assertEqual(sum(call[:2] == ("container", "rm") for call in self.docker.calls), 1)

    def test_rotation_recovers_lost_reply_using_durably_saved_pending_credential(self):
        self.agent.config_path = str(Path(self.temp.name) / "agent.json")
        accepted = {}
        def request(method, path, payload):
            if not accepted:
                accepted["credential"] = payload["credential"]
                persisted = json.loads(Path(self.agent.config_path).read_text())
                self.assertEqual(persisted["pendingCredential"], payload["credential"])
                raise AgentError("Controller request failed")
            self.assertEqual(self.agent.config["credential"], accepted["credential"])
            return {"ok": True}
        self.agent.request = request
        result = self.agent.execute(self.operation("credentials.rotate"))
        self.assertEqual(result["state"], "completed")
        saved = json.loads(Path(self.agent.config_path).read_text())
        self.assertNotIn("pendingCredential", saved)
        self.assertEqual(saved["credential"], accepted["credential"])

    def test_status_parser_never_treats_partial_or_unfamiliar_rows_as_empty(self):
        header = "map: zombie_cosmodrome\nnum score ping guid name\n"
        self.assertIsNone(module.status_players(header))
        self.assertEqual(module.status_players(header + "--- ----- ---- ---- ----\n"), 0)
        for row in ("0 0 CNCT abc player", "1 90 ZMBI abc player", "2 4 ???? abc player"):
            self.assertEqual(module.status_players(header + "--- ----- ---- ---- ----\n" + row), 1)
        self.assertIsNone(module.status_players(header + "--- ----- ---- ---- ----\nconnection in progress"))
        self.assertEqual(module.status_players('statusResponse\n\\mapname\\mp_kino\\sv_maxclients\\4\n10 45 "joe"'), 1)

    def test_udp_rcon_uses_host_local_endpoint_and_returns_structured_occupancy(self):
        self.create()
        class FakeSocket:
            def __enter__(self): return self
            def __exit__(self, *_): pass
            def settimeout(self, value): pass
            def connect(self, endpoint): self.endpoint = endpoint
            def send(self, packet): self.packet = packet
            def recv(self, _size):
                if getattr(self, "done", False): raise module.socket.timeout()
                self.done = True
                return b"\xff\xff\xff\xffprint\nmap: mp_kino\nnum score ping guid name\n--- ----- ---- ---- ----\n0 42 CNCT abc player\n"
        sock = FakeSocket()
        with patch.object(module.socket, "socket", return_value=sock):
            result = self.agent.rcon(INSTANCE, 1, "status")
        self.assertEqual(sock.endpoint, ("127.0.0.1", 28960))
        self.assertTrue(sock.packet.startswith(b"\xff\xff\xff\xffrcon rcon-secret status"))
        self.assertEqual(result["players"], 1)
        self.assertTrue(result["ok"])

    @staticmethod
    def rcon_socket(*replies):
        sock = MagicMock()
        sock.__enter__.return_value = sock
        sock.recv.side_effect = replies
        return sock

    def test_udp_status_recovers_after_transient_failures(self):
        self.create()
        for error in (module.socket.timeout(), OSError("unreachable")):
            with self.subTest(error=type(error).__name__):
                failed = self.rcon_socket(error)
                # Keep all response packets, including occupied player rows.
                recovered = self.rcon_socket(
                    b"\xff\xff\xff\xffprint\nmap: mp_kino\nnum score ping guid name\n--- ----- ---- ---- ----\n",
                    b"\xff\xff\xff\xffprint\n0 42 CNCT abc player\n", module.socket.timeout())
                with patch.object(module.socket, "socket", side_effect=[failed, recovered]) as sockets, patch.object(module.time, "sleep") as sleep:
                    result = self.agent.rcon(INSTANCE, 1, "status")
                self.assertTrue(result["ok"])
                self.assertEqual(result["players"], 1)
                self.assertEqual(sockets.call_count, 2)
                sleep.assert_called_once_with(0.25)
                failed.__exit__.assert_called_once()
                recovered.__exit__.assert_called_once()

    def test_udp_status_retries_are_bounded(self):
        self.create()
        attempts = [self.rcon_socket(module.socket.timeout()) for _ in range(3)]
        with patch.object(module.socket, "socket", side_effect=attempts) as sockets, patch.object(module.time, "sleep") as sleep:
            result = self.agent.rcon(INSTANCE, 1, "status")
        self.assertFalse(result["ok"])
        self.assertEqual(result["reason"], "timeout")
        self.assertEqual(sockets.call_count, 3)
        self.assertEqual(sleep.call_args_list, [call(0.25), call(0.5)])

    def test_udp_status_does_not_retry_bad_replies_or_partial_tables(self):
        self.create()
        for packet in (b"wrong listener", b"\xff\xff\xff\xffprint\nBad rcon_password",
                       b"\xff\xff\xff\xffprint\nmap: mp_kino\nnum score ping guid name\n"):
            with self.subTest(packet=packet):
                sock = self.rcon_socket(packet, module.socket.timeout())
                with patch.object(module.socket, "socket", return_value=sock) as sockets, patch.object(module.time, "sleep") as sleep:
                    result = self.agent.rcon(INSTANCE, 1, "status")
                self.assertIsNone(result["players"])
                sockets.assert_called_once()
                sleep.assert_not_called()

    def test_udp_commands_with_effects_are_never_replayed(self):
        self.create()
        for command in ("map_restart", "kick 0", "version"):
            with self.subTest(command=command):
                sock = self.rcon_socket(module.socket.timeout())
                with patch.object(module.socket, "socket", return_value=sock) as sockets, patch.object(module.time, "sleep") as sleep:
                    result = self.agent.rcon(INSTANCE, 1, command)
                self.assertEqual(result["reason"], "timeout")
                sockets.assert_called_once()
                sleep.assert_not_called()

    def test_t6_bot_column_allows_empty_restart_but_holds_players_and_partial_replies(self):
        self.create()
        header = "map: zm_nuked\nnum score bot ping guid                             name             lastmsg address               qport  rate\n"
        table = header + "--- ----- --- ---- -------------------------------- ---------------- ------- --------------------- ------ -----\n"
        operation = self.operation("instance.restart")
        for body, players in (
            (header, None),
            (table + "0 0 0 CNCT abc player 0 127.0.0.1:28960 123 25000\n", 1),
            (table + "0 0 1 0 abc bot 0 loopback 0 25000\n", 1),
            (table + "unrecognized trailing output", None),
            (table, 0),
        ):
            with self.subTest(body=body):
                self.assertEqual(module.status_players(body), players)
                self.agent.rcon = lambda *_: {"ok": True, "players": module.status_players(body)}
                result = self.agent.execute(operation)
                self.assertEqual(result["state"], "completed" if players == 0 else "accepted")
                if players != 0:
                    self.assertFalse(any(call[:2] == ("container", "restart") for call in self.docker.calls))

    def test_invalid_rebuild_preflight_preserves_running_original(self):
        payload = self.create()
        result = self.agent.execute(self.operation("instance.rebuild", {**payload, "image": "--privileged", "nextGeneration": 2}))
        self.assertEqual(result["state"], "failed")
        self.assertTrue(self.agent.inspect(INSTANCE, 1)["running"])
        self.assertFalse(any(call[:2] == ("container", "stop") for call in self.docker.calls))

    def test_file_operation_resists_parent_symlink_swap_after_validation(self):
        self.create()
        app = self.agent.instance_dir(INSTANCE, 1) / "app"
        outside = Path(self.temp.name) / "outside"
        outside.mkdir()
        (outside / "server.cfg").write_bytes(b"host-data")
        original = module.scoped
        def swap(root, path, **kwargs):
            result = original(root, path, **kwargs)
            (app / "configs").rename(app / "saved-configs")
            (app / "configs").symlink_to(outside)
            return result
        with patch.object(module, "scoped", side_effect=swap):
            with self.assertRaises(OSError):
                module.managed_write(app, "configs/server.cfg", b"escape-attempt")
        self.assertEqual((outside / "server.cfg").read_bytes(), b"host-data")

    def test_deleting_absent_file_with_absent_parent_is_idempotent(self):
        self.create()
        app = self.agent.instance_dir(INSTANCE, 1) / "app"
        module.managed_write(app, "missing/subdirectory/file.gsc", None)
        self.assertFalse((app / "missing").exists())

    def test_controller_contact_failure_holds_before_running_any_docker_command(self):
        self.create()
        count = len(self.docker.calls)
        self.agent.report = lambda *_: (_ for _ in ()).throw(AgentError("Controller request failed"))
        result = self.agent.execute(self.operation("instance.stop"), report_running=True)
        self.assertEqual(result["state"], "accepted")
        self.assertEqual(len(self.docker.calls), count)

    def test_async_import_publication_resumes_after_controller_restart(self):
        responses = iter([{"status": "verifying"}, {"status": "pending"}, {"status": "verifying"}, {"status": "published", "version": {"id": "library-one"}}])
        methods = []
        def request(method, path, *args):
            methods.append(method)
            return Response(json.dumps(next(responses)).encode())
        self.agent.artifact_open = request
        with patch.object(self.agent.stopping, "wait", return_value=False):
            result = self.agent.complete_import({"completeUrl": "/complete"}, start=True)
        self.assertEqual(result["status"], "published")
        self.assertEqual(methods, ["POST", "GET", "POST", "GET"])

    def test_digest_pinned_runtime_inventory_and_exact_aliases(self):
        image_digest = "sha256:" + "1" * 64
        image_id = "sha256:" + "2" * 64
        original = self.docker.call
        def call(*args, **kwargs):
            if args[:2] == ("image", "ls"):
                return f"ghcr.io/ayymoss/plutainer|latest|{image_digest}|{image_id}\n"
            return original(*args, **kwargs)
        self.docker.call = call
        module.atomic_json(self.agent.root / "agent/runtime-images.json", {"ghcr.io/ayymoss/plutainer": image_id, "removed-image": "missing-id"})
        images = self.agent.inventory()["runtimeImages"]
        self.assertIn("ghcr.io/ayymoss/plutainer@" + image_digest, images)
        self.assertIn("ghcr.io/ayymoss/plutainer:latest", images)
        self.assertIn("ghcr.io/ayymoss/plutainer", images)
        self.assertNotIn("removed-image", images)

    def test_heartbeat_applies_configured_host_port_range_before_deployment(self):
        self.agent.request = lambda *_: {"host": {"portMin": 30000, "portMax": 30100}}
        self.agent.heartbeat()
        self.assertEqual(self.agent.config["portRange"], {"min": 30000, "max": 30100})

    def test_heartbeat_requires_valid_retention_and_never_changes_the_journal(self):
        self.assertEqual(self.agent.observation_keep_days, 0)
        with patch.object(self.agent.journal, "prune_observations", side_effect=AssertionError("heartbeat must not use the journal")), patch.object(self.agent.journal, "confirm_completed", side_effect=AssertionError("heartbeat must not use the journal")):
            for value, expected in ((1, 1), (36500, 36500), (0, 0), (True, 0), (False, 0), (1.0, 0), ("1", 0), (-1, 0), (36501, 0), (None, 0)):
                with self.subTest(value=value):
                    self.agent.observation_keep_days = 3
                    self.agent.request = lambda *_, value=value: {"observationKeepDays": value}
                    self.agent.heartbeat()
                    self.assertEqual(self.agent.observation_keep_days, expected)
            self.agent.observation_keep_days = 3
            self.agent.request = lambda *_: {}
            self.agent.heartbeat()
            self.assertEqual(self.agent.observation_keep_days, 0)

    def test_lost_completed_receipt_is_retained_and_replay_reports_the_saved_result(self):
        operation = self.operation("files.stat", {"path": "configs/server.cfg"})
        self.agent.journal.accept(operation)
        with patch.object(module.time, "time", return_value=module.time.time() - 172800):
            row = self.agent.journal.update(operation["id"], "completed", {"size": 8})
        self.agent.request = MagicMock(side_effect=AgentError("receipt lost"))
        with self.assertRaisesRegex(AgentError, "receipt lost"):
            self.agent.report(row)
        self.assertIsNone(self.agent.journal.get(operation["id"])["reported_at"])
        self.assertEqual(self.agent.journal.prune_observations(1), 0)
        with patch.object(self.agent, "dispatch", side_effect=AssertionError("completed read must not be executed again")):
            replay = self.agent.execute(operation)
        self.assertEqual(replay["result"], row["result"])
        for response in ({}, None, {"operation": {"id": "different", "state": "completed"}}, {"operation": {"id": operation["id"], "state": "running"}}):
            self.agent.request = lambda *_, response=response: response
            self.agent.report(replay)
            self.assertIsNone(self.agent.journal.get(operation["id"])["reported_at"])
        self.agent.request = lambda *_: {"operation": {"id": operation["id"], "state": "completed"}}
        self.agent.report(replay)
        self.assertIsNotNone(self.agent.journal.get(operation["id"])["reported_at"])
        self.assertEqual(self.agent.journal.prune_observations(1), 1)

    def test_threaded_read_lane_confirms_and_prunes_through_its_own_connection(self):
        self.create()
        old_operation = self.operation("files.list", op_id="old-observation")
        self.agent.journal.accept(old_operation)
        with patch.object(module.time, "time", return_value=module.time.time() - 172800):
            old = self.agent.journal.update(old_operation["id"], "completed", {"entries": []})
        self.agent.journal.confirm_completed(old)
        self.agent.observation_keep_days = 1
        operation = self.operation("files.stat", {"path": "configs/server.cfg"}, "thread-observation")
        polls = 0
        failures = []

        def request(method, path, payload=None):
            nonlocal polls
            if method == "POST":
                return {"operation": {"id": operation["id"], "state": payload["state"]}}
            polls += 1
            if polls == 1:
                return {"operations": [operation]}
            self.agent.stopping.set()
            return {"operations": []}

        def run():
            try:
                self.agent.read_loop()
            except BaseException as error:
                failures.append(error)

        self.agent.request = request
        with patch.object(self.agent.journal, "confirm_completed", side_effect=AssertionError("main journal used by reader")), patch.object(self.agent.journal, "prune_observations", side_effect=AssertionError("main journal used by reader")):
            thread = threading.Thread(target=run)
            thread.start()
            thread.join(timeout=5)
            self.agent.stopping.set()
            thread.join(timeout=5)
        self.assertFalse(thread.is_alive())
        self.assertEqual(failures, [])
        self.assertEqual(polls, 2)
        self.assertIsNotNone(self.agent.journal.get(operation["id"])["reported_at"])
        with self.assertRaisesRegex(AgentError, "Unknown operation"):
            self.agent.journal.get(old_operation["id"])
        self.assertEqual(self.agent.journal.get("create")["state"], "completed")

    def test_download_content_permissions_allow_container_read_access_with_private_umask(self):
        content = b"complete-game-file"
        manifest = self.manifest(content)
        self.agent.artifact_open = lambda method, path, **kwargs: Response(json.dumps(manifest).encode() if path == "/manifest" else content)
        previous = module.os.umask(0o077)
        try:
            self.agent.library_download({"libraryVersionId": "downloaded", "manifestUrl": "/manifest", "filesBaseUrl": "/files"})
        finally:
            module.os.umask(previous)
        content_root = self.agent.library_dir("t6", "downloaded") / "content"
        self.assertEqual(content_root.stat().st_mode & 0o777, 0o555)
        self.assertEqual((content_root / "zone").stat().st_mode & 0o777, 0o555)
        self.assertEqual((content_root / "zone/test.ff").stat().st_mode & 0o777, 0o444)

    def test_agent_restart_resumes_same_immutable_library_operation(self):
        content = b"complete-game-file"
        manifest = self.manifest(content)
        payload = {"libraryVersionId": "downloaded", "manifestUrl": "/manifest", "filesBaseUrl": "/files", "manifestSha256": manifest["manifestSha256"]}
        operation = self.operation("library.download", payload)
        self.agent.journal.accept(operation)
        self.agent.journal.update(operation["id"], "running")
        staging = self.agent.root / "libraries/t6/.downloaded.partial/content/zone"
        staging.mkdir(parents=True)
        (staging / "test.ff").write_bytes(content[:8])
        self.agent.journal.close()
        self.agent = Agent(self.config, self.docker)
        def fetch(method, path, data=None, extra_headers=None):
            if path == "/manifest":
                return Response(json.dumps(manifest).encode())
            self.assertEqual(extra_headers, {"Range": "bytes=8-"})
            return Response(content[8:], 206, {"Content-Range": f"bytes 8-{len(content)-1}/{len(content)}"})
        self.agent.artifact_open = fetch
        result = self.agent.execute(operation)
        self.assertEqual(result["state"], "completed")
        self.assertEqual((self.agent.library_dir("t6", "downloaded") / "content/zone/test.ff").read_bytes(), content)

    def test_player_join_during_activation_holds_restart_without_reapplying_files(self):
        self.create()
        target = self.agent.instance_dir(INSTANCE, 1) / "app/configs/server.cfg"
        original_write = module.managed_write
        for index, kind in enumerate(("instance.restart", "files.activate", "instance.start")):
            with self.subTest(kind=kind):
                before = target.read_bytes()
                stage_id = "late-join-" + str(index)
                content = ("new-" + str(index)).encode()
                self.agent.execute(self.operation("files.stage", {"stageId": stage_id, "files": [{"id": index, "path": "configs/server.cfg", "contentBase64": base64.b64encode(content).decode()}]}, "stage-late-" + str(index)))
                joined, writes = [False], []
                self.agent.rcon = lambda *_: {"ok": True, "players": 1 if joined[0] else 0}
                def write_then_join(*args, **kwargs):
                    original_write(*args, **kwargs)
                    writes.append(args)
                    joined[0] = True
                operation = self.operation(kind, {"stageId": stage_id}, "action-late-" + str(index))
                restarts_before = sum(call[:2] == ("container", "restart") for call in self.docker.calls)
                with patch.object(module, "managed_write", side_effect=write_then_join):
                    held = self.agent.execute(operation)
                    self.assertEqual(held["state"], "accepted")
                    self.assertEqual(json.loads(held["result"])["blocked"], "occupied")
                    self.assertEqual(target.read_bytes(), content)
                    self.assertEqual(sum(call[:2] == ("container", "restart") for call in self.docker.calls), restarts_before)
                    joined[0] = False
                    completed = self.agent.execute(operation)
                self.assertEqual(completed["state"], "completed")
                self.assertEqual(len(writes), 1)
                self.assertEqual(base64.b64decode(json.loads(completed["result"])["before"][0]["contentBase64"]), before)
                self.assertEqual(sum(call[:2] == ("container", "restart") for call in self.docker.calls), restarts_before + 1)

    def test_player_join_during_rollback_holds_restart_without_repeating_restore(self):
        self.create()
        self.agent.rcon = lambda *_: {"ok": True, "players": 0}
        self.agent.execute(self.operation("files.stage", {"stageId": "rollback-join", "files": [{"path": "configs/server.cfg", "contentBase64": base64.b64encode(b"new").decode()}]}, "stage-rollback-join"))
        self.agent.execute(self.operation("files.activate", {"stageId": "rollback-join", "restart": False}, "activate-rollback-join"))
        joined, writes = [False], []
        original_write = module.managed_write
        self.agent.rcon = lambda *_: {"ok": True, "players": 1 if joined[0] else 0}
        def restore_then_join(*args, **kwargs):
            original_write(*args, **kwargs)
            writes.append(args)
            joined[0] = True
        operation = self.operation("files.rollback", {"stageId": "rollback-join"}, "rollback-join")
        with patch.object(module, "managed_write", side_effect=restore_then_join):
            self.assertEqual(self.agent.execute(operation)["state"], "accepted")
            self.assertFalse(any(call[:2] == ("container", "restart") for call in self.docker.calls))
            joined[0] = False
            self.assertEqual(self.agent.execute(operation)["state"], "completed")
        self.assertEqual(len(writes), 1)
        self.assertEqual((self.agent.instance_dir(INSTANCE, 1) / "app/configs/server.cfg").read_bytes(), b"original")

    def test_start_with_stage_does_not_reconcile_success_before_required_restart(self):
        self.create()
        self.agent.rcon = lambda *_: {"ok": True, "players": 0}
        self.agent.execute(self.operation("files.stage", {"stageId": "start-stage", "files": [{"path": "configs/server.cfg", "contentBase64": base64.b64encode(b"new").decode()}]}, "stage-start"))
        self.docker.fail_restart = True
        operation = self.operation("instance.start", {"stageId": "start-stage"})
        self.assertEqual(self.agent.execute(operation)["state"], "unknown")
        self.assertEqual(self.agent.execute(operation)["state"], "unknown")
        self.assertEqual(sum(call[:2] == ("container", "restart") for call in self.docker.calls), 1)

    def test_running_unhealthy_container_uses_fresh_occupancy_before_recovery(self):
        self.create()
        self.docker.containers[self.agent.container_name(INSTANCE, 1)]["State"]["Health"] = {"Status": "unhealthy"}
        operation = self.operation("instance.restart")
        self.agent.rcon = lambda *_: {"ok": True, "players": 2}
        self.assertEqual(self.agent.execute(operation)["state"], "accepted")
        self.agent.rcon = lambda *_: {"ok": True, "players": 0}
        self.assertEqual(self.agent.execute(operation)["state"], "completed")

    def test_game_log_can_follow_internal_relative_plutainer_symlink(self):
        self.create()
        app = self.agent.instance_dir(INSTANCE, 1) / "app"
        (app / "logs").mkdir()
        (app / "runtime").mkdir()
        (app / "runtime/game.log").write_text("first\nsecond\nthird\n")
        (app / "logs/games_sp.log").symlink_to("../runtime/game.log")
        result = self.agent.execute(self.operation("logs.read", {"path": "logs/games_sp.log", "lines": 2}))
        self.assertEqual(result["state"], "completed")
        self.assertEqual(json.loads(result["result"])["lines"], ["second", "third"])
        with self.assertRaises(AgentError):
            module.managed_write(app, "logs/games_sp.log", b"must-not-write-through-link")

    def test_game_log_rejects_symlink_escaping_instance(self):
        self.create()
        app = self.agent.instance_dir(INSTANCE, 1) / "app"
        (app / "logs").mkdir()
        outside = Path(self.temp.name) / "host-secret"
        outside.write_text("secret")
        (app / "logs/games_sp.log").symlink_to(outside)
        result = self.agent.execute(self.operation("logs.read", {"path": "logs/games_sp.log"}))
        self.assertEqual(result["state"], "failed")
        self.assertIn("outside", result["error"])

    def test_console_log_falls_back_to_owned_docker_output_including_stderr(self):
        self.create()
        original = self.docker.call
        def call(*args, **kwargs):
            if args[:2] == ("container", "logs"):
                self.assertTrue(kwargs["merge_stderr"])
                self.assertEqual(args[-1], self.agent.container_name(INSTANCE, 1))
                return "stdout-line\nstderr-line\n"
            return original(*args, **kwargs)
        self.docker.call = call
        result = self.agent.execute(self.operation("logs.read", {"path": "logs/console.log"}))
        self.assertEqual(result["state"], "completed")
        self.assertEqual(json.loads(result["result"])["lines"], ["stdout-line", "stderr-line"])


if __name__ == "__main__":
    unittest.main()