"""
File Sync Integration Tests
============================
Verifies the end-to-end mesh file synchronisation behaviour across nodes and
the Hub server mirror.  These tests run against a *live* deployment that has
at least two test nodes connected:

  •  test-node-1  (NODE_1 constant)
  •  test-node-2  (NODE_2 constant)

The Hub exposes REST endpoints used to:
  - read  files: GET  /api/v1/nodes/{node_id}/fs/cat
  - list  dirs:  GET  /api/v1/nodes/{node_id}/fs/ls
  - write files: POST /api/v1/nodes/{node_id}/fs/touch
  - delete:      POST /api/v1/nodes/{node_id}/fs/rm

A shared swarm-control session is created once per test module so that all
nodes are in the same mesh workspace, and file operations propagate correctly.

Environment assumptions
-----------------------
  BASE_URL   http://127.0.0.1:8000   (inside container) or http://192.168.68.113 (from outside)
  NODE_1     test-node-1
  NODE_2     test-node-2
  USER_ID    SYNC_TEST_USER_ID env var (required for node access checks)
  TIMEOUT    10 s for small files, 60 s for 20 MB files
"""

import os
import time
import uuid
import hashlib
import pytest
import httpx

# ── Configuration ──────────────────────────────────────────────────────────────
BASE_URL = os.getenv("SYNC_TEST_BASE_URL", "http://127.0.0.1:8002/api/v1")
USER_ID  = os.getenv("SYNC_TEST_USER_ID", "c4401d34-8784-4d6e-93a0-c702bd202b66")
NODE_1   = os.getenv("SYNC_TEST_NODE1", "test-node-1")
NODE_2   = os.getenv("SYNC_TEST_NODE2", "test-node-2")

SMALL_FILE_TIMEOUT   = 10   # seconds
LARGE_FILE_TIMEOUT   = 60   # seconds  (20 MB)
LARGE_FILE_SIZE_MB   = 20
POLL_INTERVAL        = 0.5  # seconds

# Paths — relative to BASE_URL
SESSIONS_PATH = "/sessions"
NODES_PATH    = "/nodes"


# ── Module-level: skip the whole file if nodes are not online ──────────────────
def pytest_configure(config):
    config.addinivalue_line(
        "markers",
        "requires_nodes: mark test as requiring live agent nodes to be connected",
    )


pytestmark = pytest.mark.requires_nodes


# ── Helpers ─────────────────────────────────────────────────────────────────────

def _get_user_id() -> str:
    return os.getenv("SYNC_TEST_USER_ID", "integration_tester_sync")

def _headers():
    return {"X-User-ID": _get_user_id()}


def _unique(prefix="synctest"):
    return f"{prefix}_{uuid.uuid4().hex[:8]}.txt"


def _large_content(mb: int = LARGE_FILE_SIZE_MB) -> str:
    """Return a UTF-8 string of approximately `mb` megabytes."""
    line = "A" * 1023 + "\n"          # 1 KB per line
    return line * (mb * 1024)          # mb * 1024 lines ≈ mb MB


def _poll_until(fn, timeout: float, interval: float = POLL_INTERVAL):
    """
    Repeatedly call fn() until it returns a truthy value or timeout expires.
    Returns the last return value of fn().
    """
    deadline = time.time() + timeout
    last = None
    while time.time() < deadline:
        try:
            last = fn()
            if last:
                return last
        except Exception:
            pass
        time.sleep(interval)
    return last


def _cat(client: httpx.Client, node_id: str, path: str, session_id: str) -> str | None:
    """Read a file from a node; return its text content or None on error."""
    r = client.get(
        f"{NODES_PATH}/{node_id}/fs/cat",
        params={"path": path, "session_id": session_id},
        headers=_headers(),
    )
    if r.status_code == 200:
        return r.json().get("content", "")
    return None


def _mirror_cat(client: httpx.Client, path: str, session_id: str) -> str | None:
    """
    Read a file from the Hub server mirror directly by asking node-1 for it
    using the session_id workspace (the Hub mirror is queried when the node
    reflects a workspace file).

    For the server-side write tests we can call the same endpoint but the
    source is the Hub mirror, not the live node FS.
    """
    # The Hub's /cat endpoint fetches from node, then caches in mirror.
    # For "server wrote it → does node have it?" we ask the node directly.
    return _cat(client, NODE_1, path, session_id)


def _touch(
    client: httpx.Client,
    node_id: str,
    path: str,
    content: str,
    session_id: str,
    is_dir: bool = False,
) -> dict:
    """Write a file to a node via the REST API."""
    r = client.post(
        f"{NODES_PATH}/{node_id}/fs/touch",
        json={"path": path, "content": content, "is_dir": is_dir, "session_id": session_id},
        headers=_headers(),
        timeout=120.0,
    )
    r.raise_for_status()
    return r.json()


def _rm(client: httpx.Client, node_id: str, path: str, session_id: str) -> dict:
    """Delete a file from a node via the REST API."""
    r = client.post(
        f"{NODES_PATH}/{node_id}/fs/rm",
        json={"path": path, "session_id": session_id},
        headers=_headers(),
        timeout=30.0,
    )
    r.raise_for_status()
    return r.json()


def _file_missing(client: httpx.Client, node_id: str, path: str, session_id: str) -> bool:
    """Return True if the file does NOT exist on the node."""
    r = client.get(
        f"{NODES_PATH}/{node_id}/fs/cat",
        params={"path": path, "session_id": session_id},
        headers=_headers(),
    )
    return r.status_code != 200


# ── Session fixture ─────────────────────────────────────────────────────────────

@pytest.fixture(scope="module")
def sync_client():
    """Synchronous httpx client for the whole module (avoids asyncio overhead)."""
    with httpx.Client(base_url=BASE_URL, timeout=60.0) as c:
        # Quick connectivity + node-online check
        try:
            r = c.get(f"{NODES_PATH}/{NODE_1}/status", headers=_headers())
            if r.status_code not in (200, 404):
                pytest.skip(f"{NODE_1} unreachable — hub returned {r.status_code}")
        except Exception as exc:
            pytest.skip(f"Hub unreachable at {BASE_URL}: {exc}")
        yield c


@pytest.fixture(scope="module")
def swarm_session(sync_client: httpx.Client) -> str:
    """
    Create (or reuse) one swarm_control session that has both test nodes
    attached.  Returned value is the workspace_id string used by all sync ops.
    """
    # Create the session
    r = sync_client.post(
        f"{SESSIONS_PATH}/",
        json={"user_id": _get_user_id(), "provider_name": "gemini", "feature_name": "swarm_control"},
        headers=_headers(),
    )
    assert r.status_code == 200, f"Create session failed: {r.text}"
    session_id = r.json()["id"]

    # Attach both nodes with source="empty" so they both watch the workspace
    r2 = sync_client.post(
        f"{SESSIONS_PATH}/{session_id}/nodes",
        json={"node_ids": [NODE_1, NODE_2], "config": {"source": "empty"}},
        headers=_headers(),
    )
    assert r2.status_code == 200, f"Attach nodes failed: {r2.text}"
    workspace_id = r2.json().get("sync_workspace_id")
    assert workspace_id, "Expected sync_workspace_id in response"

    # Give nodes a moment to ACK the workspace and start watching
    time.sleep(2.0)

    yield workspace_id

    # Teardown: archive the session
    sync_client.delete(f"{SESSIONS_PATH}/{session_id}", headers=_headers())


# ══════════════════════════════════════════════════════════════════════════════
#  SMALL FILE TESTS (< 1 chunk)
# ══════════════════════════════════════════════════════════════════════════════

class TestSmallFileSync:
    """Cases 1–4: single small-file create + delete in both directions."""

    # ── Case 1: node-1 → node-2 + server ───────────────────────────────────
    def test_case1_write_from_node1_visible_on_node2_and_server(
        self, sync_client, swarm_session
    ):
        """
        Write a file from test-node-1; verify test-node-2 AND the server
        mirror receive it within SMALL_FILE_TIMEOUT seconds.
        """
        filename = _unique("case1")
        content  = f"Case 1 payload – node-1 → mesh – {uuid.uuid4()}"
        workspace = swarm_session

        print(f"\n[Case 1] Writing {filename!r} to {NODE_1} in workspace {workspace}")
        result = _touch(sync_client, NODE_1, filename, content, workspace)
        assert result.get("success"), f"Write failed: {result}"

        # Verify on node-2
        node2_content = _poll_until(
            lambda: _cat(sync_client, NODE_2, filename, workspace),
            timeout=SMALL_FILE_TIMEOUT,
        )
        assert node2_content is not None, (
            f"[Case 1] File '{filename}' did NOT appear on {NODE_2} within "
            f"{SMALL_FILE_TIMEOUT}s"
        )
        assert content in node2_content, (
            f"[Case 1] Content mismatch on {NODE_2}. Got: {node2_content!r}"
        )
        print(f"[Case 1] ✅ {NODE_2} received the file.")

        # Verify on Hub server mirror (query node-1 with session scope uses mirror)
        server_content = _poll_until(
            lambda: _cat(sync_client, NODE_1, filename, workspace),
            timeout=SMALL_FILE_TIMEOUT,
        )
        assert server_content is not None and content in server_content, (
            f"[Case 1] File '{filename}' not found on Hub mirror."
        )
        print(f"[Case 1] ✅ Hub mirror has the file.")

    # ── Case 2: server → node-1 + node-2 ───────────────────────────────────
    def test_case2_write_from_server_visible_on_all_nodes(
        self, sync_client, swarm_session
    ):
        """
        Write a file via the server (the touch endpoint dispatches to all nodes
        in the session + writes to Hub mirror).  Verify both client nodes and
        the mirror reflect it.
        """
        filename = _unique("case2")
        content  = f"Case 2 payload – server → mesh – {uuid.uuid4()}"
        workspace = swarm_session

        print(f"\n[Case 2] Writing {filename!r} via server to workspace {workspace}")
        # Intentionally write via node-1 (server-dispatched; Hub mirror updated first)
        result = _touch(sync_client, NODE_1, filename, content, workspace)
        assert result.get("success"), f"Write failed: {result}"

        # Node-2 should receive via broadcast
        node2_content = _poll_until(
            lambda: _cat(sync_client, NODE_2, filename, workspace),
            timeout=SMALL_FILE_TIMEOUT,
        )
        assert node2_content is not None and content in node2_content, (
            f"[Case 2] File '{filename}' did NOT appear on {NODE_2}."
        )
        print(f"[Case 2] ✅ {NODE_2} received the file.")

        # Node-1 should also have it (it was written directly to it and mirrored)
        node1_content = _cat(sync_client, NODE_1, filename, workspace)
        assert node1_content is not None and content in node1_content, (
            f"[Case 2] File '{filename}' not found on {NODE_1}."
        )
        print(f"[Case 2] ✅ {NODE_1} has the file.")

    # ── Case 3: delete from server → nodes purged ──────────────────────────
    def test_case3_delete_from_server_purges_client_nodes(
        self, sync_client, swarm_session
    ):
        """
        Create a file via server, then delete it via the server endpoint.
        Verify both client nodes no longer have the file.
        """
        filename  = _unique("case3")
        content   = f"Case 3 – to be deleted by server – {uuid.uuid4()}"
        workspace = swarm_session

        # Setup: write the file from node-1 (server-side orchestrated)
        r = _touch(sync_client, NODE_1, filename, content, workspace)
        assert r.get("success"), f"Setup write failed: {r}"

        # Make sure node-2 got it before we delete
        got = _poll_until(
            lambda: _cat(sync_client, NODE_2, filename, workspace),
            timeout=SMALL_FILE_TIMEOUT,
        )
        assert got is not None, f"[Case 3] Setup: file not on {NODE_2}."

        print(f"\n[Case 3] Deleting {filename!r} from server (via {NODE_1} endpoint)")
        _rm(sync_client, NODE_1, filename, workspace)

        # node-2 should no longer have the file
        gone_node2 = _poll_until(
            lambda: _file_missing(sync_client, NODE_2, filename, workspace),
            timeout=SMALL_FILE_TIMEOUT,
        )
        assert gone_node2, (
            f"[Case 3] File '{filename}' still present on {NODE_2} after server delete."
        )
        print(f"[Case 3] ✅ {NODE_2} no longer has the file.")

        # node-1 should also have it gone
        gone_node1 = _file_missing(sync_client, NODE_1, filename, workspace)
        assert gone_node1, (
            f"[Case 3] File '{filename}' still present on {NODE_1} after server delete."
        )
        print(f"[Case 3] ✅ {NODE_1} no longer has the file.")

    # ── Case 4: delete from node-2 → server + node-1 purged ───────────────
    def test_case4_delete_from_node2_purges_server_and_node1(
        self, sync_client, swarm_session
    ):
        """
        Create a file, let it propagate, then delete it FROM node-2.
        Verify the Hub mirror and node-1 no longer have the file.
        """
        filename  = _unique("case4")
        content   = f"Case 4 – to be deleted from node-2 – {uuid.uuid4()}"
        workspace = swarm_session

        # Setup: write from node-1 so both nodes and mirror have it
        r = _touch(sync_client, NODE_1, filename, content, workspace)
        assert r.get("success"), f"Setup write failed: {r}"

        got_node2 = _poll_until(
            lambda: _cat(sync_client, NODE_2, filename, workspace),
            timeout=SMALL_FILE_TIMEOUT,
        )
        assert got_node2 is not None, f"[Case 4] Setup: file did not reach {NODE_2}."

        print(f"\n[Case 4] Deleting {filename!r} from {NODE_2}")
        _rm(sync_client, NODE_2, filename, workspace)

        # Hub mirror (observed via node-1's workspace view) should purge
        gone_server = _poll_until(
            lambda: _file_missing(sync_client, NODE_1, filename, workspace),
            timeout=SMALL_FILE_TIMEOUT,
        )
        assert gone_server, (
            f"[Case 4] File '{filename}' still present on Hub mirror after node-2 delete."
        )
        print(f"[Case 4] ✅ Hub mirror no longer has the file.")

        # node-1 should also be purged
        gone_node1 = _file_missing(sync_client, NODE_1, filename, workspace)
        assert gone_node1, (
            f"[Case 4] File '{filename}' still present on {NODE_1} after node-2 delete."
        )
        print(f"[Case 4] ✅ {NODE_1} no longer has the file.")

    # ── Case 9: cat on deleted file returns quickly, not after timeout ──────
    def test_case9_cat_deleted_file_returns_quickly_not_timeout(
        self, sync_client, swarm_session
    ):
        """
        Regression test for the silent-return bug in _push_file (node side)
        and the missing mirror short-circuit in cat() (hub side).

        Before the fix, reading a deleted file would stall for the full 15s
        journal timeout because the node returned nothing and the hub just sat
        waiting.  After the fix:
          - hub:  cat() checks the mirror first; file absent → instant "File not found"
          - node: _push_file sends an ERROR SyncStatus immediately when file missing

        This test enforces that a cat call on a deleted file resolves in under
        MAX_LATENCY_S seconds on BOTH nodes.
        """
        MAX_LATENCY_S = 3.0   # well below the 15s journal timeout
        filename  = _unique("case9_latency")
        content   = f"Case 9 — delete latency probe — {uuid.uuid4()}"
        workspace = swarm_session

        # Setup: write the file and wait for full propagation
        r = _touch(sync_client, NODE_1, filename, content, workspace)
        assert r.get("success"), f"[Case 9] Setup write failed: {r}"
        synced = _poll_until(
            lambda: _cat(sync_client, NODE_2, filename, workspace),
            timeout=SMALL_FILE_TIMEOUT,
        )
        assert synced is not None, f"[Case 9] Setup: file did not propagate to {NODE_2}."

        # Delete from server
        print(f"\n[Case 9] Deleting {filename!r}, then timing cat() on both nodes")
        _rm(sync_client, NODE_1, filename, workspace)

        # Give delete broadcast a moment to reach nodes (but not the full poll timeout)
        time.sleep(1.5)

        # Measure cat latency on node-1 (hub mirror path — should be instant)
        t0 = time.time()
        res1 = _cat(sync_client, NODE_1, filename, workspace)
        latency_node1 = time.time() - t0
        assert res1 is None, (
            f"[Case 9] {NODE_1} still returned content after delete: {res1!r}"
        )
        assert latency_node1 < MAX_LATENCY_S, (
            f"[Case 9] cat() on {NODE_1} took {latency_node1:.1f}s — expected < {MAX_LATENCY_S}s. "
            f"Hub mirror short-circuit may be broken."
        )
        print(f"[Case 9] ✅ {NODE_1} cat returned in {latency_node1:.2f}s (file absent, fast-fail).")

        # Measure cat latency on node-2 (hub mirror path — should also be instant)
        t0 = time.time()
        res2 = _cat(sync_client, NODE_2, filename, workspace)
        latency_node2 = time.time() - t0
        assert res2 is None, (
            f"[Case 9] {NODE_2} still returned content after delete: {res2!r}"
        )
        assert latency_node2 < MAX_LATENCY_S, (
            f"[Case 9] cat() on {NODE_2} took {latency_node2:.1f}s — expected < {MAX_LATENCY_S}s. "
            f"Node _push_file may not be sending error status on missing file."
        )
        print(f"[Case 9] ✅ {NODE_2} cat returned in {latency_node2:.2f}s (file absent, fast-fail).")

    # ── Case 11: Agent "hub" pseudo-node write visibility ───────────────────
    def test_case11_hub_pseudo_node_write_visibility(
        self, sync_client, swarm_session
    ):
        """
        Regression test for AI agents writing to node='hub'.
        When an AI uses mesh_file_explorer with node='hub', it directly modifies
        the local mirror without broadcasting to an agent node immediately if it's
        just "hub". Wait, if session_id is provided, it DOES broadcast!
        Let's ensure that writing to 'hub' with a valid session scope returns success,
        is immediately visible in the mirror, and is retrievable via fs/cat!
        """
        filename = _unique("case11_hub")
        content = f"Case 11 — Dummy file from hub — {uuid.uuid4()}"
        workspace = swarm_session

        print(f"\n[Case 11] Writing {filename!r} to pseudonode 'hub' in workspace {workspace}")
        result = _touch(sync_client, "hub", filename, content, workspace)
        assert result.get("success"), f"Write to 'hub' failed: {result}"

        # Mirror cat (since we read from node-1, the Hub mirror resolves it instantly if it exists)
        # Or better yet, read from "hub" pseudonode!
        def _cat_hub():
            r = sync_client.get(
                f"{NODES_PATH}/hub/fs/cat",
                params={"path": filename, "session_id": workspace},
                headers=_headers(),
            )
            if r.status_code == 200:
                return r.json().get("content", "")
            return None

        # Verify on Hub mirror
        hub_content = _poll_until(_cat_hub, timeout=SMALL_FILE_TIMEOUT)
        assert hub_content is not None, f"[Case 11] File '{filename}' not found on 'hub' pseudonode."
        assert content in hub_content, f"[Case 11] Content mismatch on Hub loopback."
        print(f"[Case 11] ✅ 'hub' pseudonode successfully returned the file.")



# ══════════════════════════════════════════════════════════════════════════════
#  NODE RECONNECT / RESYNC TESTS
# ══════════════════════════════════════════════════════════════════════════════

# Docker container names for the test nodes on the production server
_NODE_CONTAINER = {
    "test-node-1": "cortex-test-1",
    "test-node-2": "cortex-test-2",
}

import subprocess
import os

def _get_remote_env():
    try:
        script_path = os.path.abspath(os.path.join(os.path.dirname(__file__), "../../.agent/utils/env_loader.sh"))
        if os.path.exists(script_path):
            cmd = f"source {script_path} >/dev/null 2>&1 && echo \"${{REMOTE_PASSWORD}}|${{REMOTE_USER}}|${{REMOTE_HOST}}\""
            res = subprocess.run(["bash", "-c", cmd], capture_output=True, text=True, check=True)
            parts = res.stdout.strip().split("|")
            if len(parts) == 3 and parts[0]:
                return parts[0], parts[1], parts[2]
    except Exception:
        pass
    return os.environ.get("REMOTE_PASSWORD", ""), os.environ.get("REMOTE_USER", "axieyangb"), os.environ.get("REMOTE_HOST", "192.168.68.113")

_REMOTE_PASSWORD, _REMOTE_USER, _REMOTE_HOST = _get_remote_env()
_SSH_CMD = f"sshpass -p '{_REMOTE_PASSWORD}' ssh -o StrictHostKeyChecking=no {_REMOTE_USER}@{_REMOTE_HOST}"


def _restart_test_node(node_id: str):
    """
    Restart the named test-node Docker container on the production server.
    This wipes /tmp/cortex-sync on the node, simulating a real reboot.
    """
    import subprocess
    container = _NODE_CONTAINER.get(node_id)
    if not container:
        pytest.skip(f"No container mapping for {node_id}")
    cmd = (
        f"sshpass -p '{_REMOTE_PASSWORD}' ssh -o StrictHostKeyChecking=no {_REMOTE_USER}@{_REMOTE_HOST} "
        f"\"echo '{_REMOTE_PASSWORD}' | sudo -S docker restart {container}\""
    )
    result = subprocess.run(cmd, shell=True, capture_output=True, text=True, timeout=30)
    if result.returncode != 0:
        pytest.skip(f"Could not restart {container}: {result.stderr}")


class TestNodeResync:
    """
    Case 10: node reconnect / workspace resync after container restart.

    Real-world scenario: a test node restarts (deploy, crash, reboot) and
    /tmp/cortex-sync is wiped.  The Hub must re-push the workspace to the
    reconnected node via manifest-driven reconciliation.
    """

    # ── Case 10: node-2 restart → hub re-delivers workspace ────────────────
    def test_case10_node_resync_after_restart(
        self, sync_client, swarm_session
    ):
        """
        1. Write a file to node-1 and confirm node-2 received it.
        2. Restart the node-2 container (wipes /tmp/cortex-sync).
        3. Wait for node-2 to reconnect and receive the manifest from Hub.
        4. Assert that the file re-appears on node-2 within RESYNC_TIMEOUT.

        This guards against regressions in the push_workspace / manifest-driven
        reconciliation loop that re-delivers Hub mirror contents to a freshly
        reconnected node.
        """
        RESYNC_TIMEOUT   = 30    # seconds for node to reconnect + resync
        RESTART_WAIT     = 8     # seconds to allow container to come back up

        filename  = _unique("case10_resync")
        content   = f"Case 10 — node resync after restart — {uuid.uuid4()}"
        workspace = swarm_session

        # Setup: write from node-1, wait for node-2 to receive
        r = _touch(sync_client, NODE_1, filename, content, workspace)
        assert r.get("success"), f"[Case 10] Setup write failed: {r}"

        synced = _poll_until(
            lambda: _cat(sync_client, NODE_2, filename, workspace),
            timeout=SMALL_FILE_TIMEOUT,
        )
        assert synced is not None, f"[Case 10] Setup: file did not reach {NODE_2} before restart."
        print(f"\n[Case 10] File confirmed on {NODE_2}. Restarting container…")

        # Restart node-2 container — wipes /tmp/cortex-sync
        _restart_test_node(NODE_2)

        # Brief pause to let the container fully stop, then wait for reconnect
        time.sleep(RESTART_WAIT)
        print(f"[Case 10] Container restarted. Waiting for {NODE_2} to reconnect and resync…")

        # After reconnect, node sends its (now-empty) manifest → Hub sends back
        # all missing files.  Poll until the file reappears.
        resynced = _poll_until(
            lambda: _cat(sync_client, NODE_2, filename, workspace),
            timeout=RESYNC_TIMEOUT,
        )
        assert resynced is not None, (
            f"[Case 10] File '{filename}' did NOT re-appear on {NODE_2} within "
            f"{RESYNC_TIMEOUT}s after container restart. "
            f"Manifest-driven resync may be broken."
        )
        assert content in resynced, (
            f"[Case 10] Content mismatch on {NODE_2} after resync. Got: {resynced!r}"
        )
        print(f"[Case 10] ✅ {NODE_2} resynced the file after container restart.")


# ══════════════════════════════════════════════════════════════════════════════
#  LARGE FILE TESTS (20 MB, multi-chunk)
# ══════════════════════════════════════════════════════════════════════════════

class TestLargeFileSync:
    """Cases 5–8: 20 MB file create + delete in both directions."""

    @pytest.fixture(scope="class", autouse=True)
    def _large_content(self):
        """Pre-build the 20 MB string once per class to save CPU time."""
        self.__class__._content = _large_content(LARGE_FILE_SIZE_MB)
        self.__class__._expected_hash = hashlib.sha256(
            self._content.encode()
        ).hexdigest()

    # ── Case 5: 20 MB from node-1 → server + node-2 ────────────────────────
    def test_case5_large_file_from_node1_to_server_and_node2(
        self, sync_client, swarm_session
    ):
        """
        Create a 20 MB file from test-node-1.
        Both server mirror and test-node-2 should receive it within
        LARGE_FILE_TIMEOUT seconds.
        """
        filename  = _unique("case5_large")
        workspace = swarm_session

        print(f"\n[Case 5] Writing {LARGE_FILE_SIZE_MB} MB file {filename!r} from {NODE_1}")
        result = _touch(sync_client, NODE_1, filename, self._content, workspace)
        assert result.get("success"), f"Write failed: {result}"

        # Verify node-2 received the file
        node2_content = _poll_until(
            lambda: _cat(sync_client, NODE_2, filename, workspace),
            timeout=LARGE_FILE_TIMEOUT,
        )
        assert node2_content is not None, (
            f"[Case 5] Large file did NOT appear on {NODE_2} within {LARGE_FILE_TIMEOUT}s."
        )
        got_hash = hashlib.sha256(node2_content.encode()).hexdigest()
        assert got_hash == self._expected_hash, (
            f"[Case 5] Hash mismatch on {NODE_2}. Expected {self._expected_hash}, got {got_hash}"
        )
        print(f"[Case 5] ✅ {NODE_2} received and verified 20 MB large file.")

        # Verify server mirror
        mirror_content = _cat(sync_client, NODE_1, filename, workspace)
        assert mirror_content is not None, (
            f"[Case 5] Large file not on Hub mirror."
        )
        print(f"[Case 5] ✅ Hub mirror has the large file.")

    # ── Case 6: 20 MB from server → node-1 + node-2 ────────────────────────
    def test_case6_large_file_from_server_to_all_nodes(
        self, sync_client, swarm_session
    ):
        """
        Write a 20 MB file via the server (touch endpoint with session scope).
        Both client nodes should receive it within LARGE_FILE_TIMEOUT seconds.
        """
        filename  = _unique("case6_large")
        workspace = swarm_session

        print(f"\n[Case 6] Writing {LARGE_FILE_SIZE_MB} MB file {filename!r} via server")
        result = _touch(sync_client, NODE_1, filename, self._content, workspace)
        assert result.get("success"), f"Write failed: {result}"

        # node-2 receives via mesh broadcast
        node2_ok = _poll_until(
            lambda: _cat(sync_client, NODE_2, filename, workspace),
            timeout=LARGE_FILE_TIMEOUT,
        )
        assert node2_ok is not None, (
            f"[Case 6] Large file did NOT appear on {NODE_2} within {LARGE_FILE_TIMEOUT}s."
        )
        print(f"[Case 6] ✅ {NODE_2} received the 20 MB file.")

        node1_ok = _cat(sync_client, NODE_1, filename, workspace)
        assert node1_ok is not None, f"[Case 6] File not on {NODE_1}."
        print(f"[Case 6] ✅ {NODE_1} has the 20 MB file.")

    # ── Case 7: delete large file from server → nodes purged ───────────────
    def test_case7_delete_large_file_from_server_purges_nodes(
        self, sync_client, swarm_session
    ):
        """
        Write and fully sync a large file, then delete via server.
        Verify all client nodes are purged.
        """
        filename  = _unique("case7_large")
        workspace = swarm_session

        # Setup
        r = _touch(sync_client, NODE_1, filename, self._content, workspace)
        assert r.get("success"), f"Setup write failed: {r}"

        synced = _poll_until(
            lambda: _cat(sync_client, NODE_2, filename, workspace),
            timeout=LARGE_FILE_TIMEOUT,
        )
        assert synced is not None, f"[Case 7] Setup: large file did not reach {NODE_2}."

        print(f"\n[Case 7] Deleting large file {filename!r} from server")
        _rm(sync_client, NODE_1, filename, workspace)

        gone_node2 = _poll_until(
            lambda: _file_missing(sync_client, NODE_2, filename, workspace),
            timeout=SMALL_FILE_TIMEOUT,
        )
        assert gone_node2, (
            f"[Case 7] Large file still present on {NODE_2} after server delete."
        )
        print(f"[Case 7] ✅ {NODE_2} purged the large file.")

        gone_node1 = _file_missing(sync_client, NODE_1, filename, workspace)
        assert gone_node1, (
            f"[Case 7] Large file still present on {NODE_1} after server delete."
        )
        print(f"[Case 7] ✅ {NODE_1} purged the large file.")

    # ── Case 8: delete large file from node-2 → server + node-1 ───────────
    def test_case8_delete_large_file_from_node2_purges_server_and_node1(
        self, sync_client, swarm_session
    ):
        """
        Write and sync a large file, then delete it FROM node-2.
        Verify Hub mirror and node-1 are both purged.
        """
        filename  = _unique("case8_large")
        workspace = swarm_session

        # Setup
        r = _touch(sync_client, NODE_1, filename, self._content, workspace)
        assert r.get("success"), f"Setup write failed: {r}"

        synced = _poll_until(
            lambda: _cat(sync_client, NODE_2, filename, workspace),
            timeout=LARGE_FILE_TIMEOUT,
        )
        assert synced is not None, f"[Case 8] Setup: large file did not reach {NODE_2}."

        print(f"\n[Case 8] Deleting large file {filename!r} from {NODE_2}")
        _rm(sync_client, NODE_2, filename, workspace)

        gone_server = _poll_until(
            lambda: _file_missing(sync_client, NODE_1, filename, workspace),
            timeout=SMALL_FILE_TIMEOUT,
        )
        assert gone_server, (
            f"[Case 8] Large file still on Hub mirror after {NODE_2} delete."
        )
        print(f"[Case 8] ✅ Hub mirror purged the large file.")

        gone_node1 = _file_missing(sync_client, NODE_1, filename, workspace)
        assert gone_node1, (
            f"[Case 8] Large file still on {NODE_1} after {NODE_2} delete."
        )
        print(f"[Case 8] ✅ {NODE_1} purged the large file.")

# ══════════════════════════════════════════════════════════════════════════════
#  GIGABYTE FILE TEST (1000 MB)
# ══════════════════════════════════════════════════════════════════════════════

class TestGigabyteFileSync:
    """Tests synchronizing a 1GB file across the mesh via DD CLI tool."""

    def test_case_1gb_sync_from_client_to_server_and_node(
        self, sync_client, swarm_session
    ):
        """
        Creates a 1 GB file on test-node-1 using the shell command `dd`.
        Verifies that it syncs to both the server mirror and test-node-2.
        """
        filename = _unique("gigabyte")
        workspace = swarm_session

        print(f"\\n[Case 1GB] Disabling memory limit checks and triggering 1GB creation on {NODE_1}...")

        # Create a 1GB file consisting of zeros (highly compressible over the network) on NODE_1 directly.
        # This will trigger the Inotify watcher to push chunks back up to the Hub.
        # We output to the active session workspace path on the node.
        is_native = os.environ.get("SKIP_DOCKER_NODES") == "true"
        sync_dir = f"/tmp/cortex-sync-{NODE_1}" if is_native else "/tmp/cortex-sync"
        dd_command = f"dd if=/dev/zero of={sync_dir}/{workspace}/{filename} bs=1M count=1000"
        
        r_disp = sync_client.post(
            f"{NODES_PATH}/{NODE_1}/dispatch",
            params={"user_id": _get_user_id()},
            json={"command": dd_command},
            headers=_headers(),
            timeout=180.0
        )
        assert r_disp.status_code == 200, f"Failed to dispatch 1GB write to {NODE_1}"
        
        # Give the agent node ample time to write to disk and push chunks over gRPC.
        # Wait up to 180 seconds.
        def _check_node2_ls():
            r = sync_client.get(
                f"{NODES_PATH}/{NODE_2}/fs/ls",
                params={"path": ".", "session_id": workspace},
                headers=_headers(),
                timeout=30.0
            )
            if r.status_code != 200:
                return False
            for f in r.json().get("files", []):
                # Only return true when size is fully 1 GB (1000 * 1024 * 1024 = 1048576000)
                if f.get("name") == filename and f.get("size", 0) >= 1048576000:
                    return f
            return False

        print(f"[Case 1GB] Polling {NODE_2} for the file...")
        node2_file = _poll_until(_check_node2_ls, timeout=180)
        assert node2_file, f"1GB Large file {filename} did not reach {NODE_2} within 180s in full 1GB size."
        print(f"[Case 1GB] ✅ {NODE_2} verified 1GB file sync with correct size.")

        # Verify Server Mirror also saw it and recorded 1GB size
        def _check_server_ls():
            r = sync_client.get(
                f"{NODES_PATH}/{NODE_1}/fs/ls",
                params={"path": ".", "session_id": workspace},
                headers=_headers(),
                timeout=30.0
            )
            if r.status_code != 200:
                return False
            for f in r.json().get("files", []):
                if f.get("name") == filename and f.get("size", 0) >= 1048576000:
                    return f
            return False

        server_file = _check_server_ls()
        assert server_file, f"1GB Large file {filename} did not appear with 1GB size on Server Mirror."
        print(f"[Case 1GB] ✅ Hub mirror successfully verified 1GB file sync with correct size.")
        
        # Cleanup
        _rm(sync_client, NODE_1, filename, workspace)


# ══════════════════════════════════════════════════════════════════════════════
#  SESSION AUTO-PURGE TEST
# ══════════════════════════════════════════════════════════════════════════════

class TestSessionAutoPurge:
    """Verifies that deleting a session purges the physical file system mirrors completely."""

    def test_session_lifecycle_cleanup(self, sync_client):
        """
        Creates a session, touches a file inside it, then deletes the session via API.
        Verifies that both the server-side mirror folder and client-side tmp folders
        are definitively purged and removed from the physical disk logic.
        """
        import subprocess

        print("\n[Case Purge] Starting session cleanup lifecycle test...")
        # 1. Create a throwaway session
        r = sync_client.post(
            f"{SESSIONS_PATH}/",
            json={"user_id": _get_user_id(), "provider_name": "gemini", "feature_name": "auto-purge-test"},
            headers=_headers(),
        )
        assert r.status_code == 200
        session_id = r.json()["id"]

        # Attach nodes
        r2 = sync_client.post(
            f"{SESSIONS_PATH}/{session_id}/nodes",
            json={"node_ids": [NODE_1, NODE_2], "config": {"source": "empty"}},
            headers=_headers(),
        )
        assert r2.status_code == 200
        workspace_id = r2.json().get("sync_workspace_id")
        
        # Give nodes a moment to ACK the workspace and create folders
        time.sleep(2.0)
        
        # 2. Write a file
        filename = _unique("autopurge")
        res = _touch(sync_client, NODE_1, filename, "garbage payload", workspace_id)
        assert res.get("success"), "Failed to write setup file for auto-purge test"

        # 3. Verify it reached Node 2 (assumes the filesystem structures were physically booted)
        node2_ok = _poll_until(
            lambda: _cat(sync_client, NODE_2, filename, workspace_id),
            timeout=SMALL_FILE_TIMEOUT,
        )
        assert node2_ok is not None, "Auto-purge setup file did not sync correctly to node 2"
        print("[Case Purge] ✅ Session folders dynamically booted across the mesh")

        # 4. DELETE the Session
        # Wait for the watcher to debounce (1s) and push the chunks
        print("[Case Purge] Waiting 2 seconds to let the dog flush the chunks...")
        time.sleep(2.0)
        
        print("[Case Purge] Calling API DELETE on the session...")
        r_del = sync_client.delete(f"{SESSIONS_PATH}/{session_id}", headers=_headers())
        assert r_del.status_code == 200

        # Wait a bit for PURGE propagation
        print("[Case Purge] Waiting 3 seconds for propagation...")
        time.sleep(3.0)

        # 5. Check client-side folders are purged using DISPATCH to run "ls"
        is_native = os.environ.get("SKIP_DOCKER_NODES") == "true"
        n1_dir = f"/tmp/cortex-sync-{NODE_1}" if is_native else "/tmp/cortex-sync"
        # Node 1
        r_d1 = sync_client.post(
            f"{NODES_PATH}/{NODE_1}/dispatch", 
            params={"user_id": _get_user_id()},
            json={"command": f"stat {n1_dir}/{workspace_id}"}, 
            headers=_headers()
        )
        assert "No such file or directory" in r_d1.json().get("stderr", "") or r_d1.json().get("status") != "successful", (
            f"Node 1 failed to purge its physical tmp folder: {r_d1.text}"
        )

        n2_dir = f"/tmp/cortex-sync-{NODE_2}" if is_native else "/tmp/cortex-sync"
        # Node 2
        r_d2 = sync_client.post(
            f"{NODES_PATH}/{NODE_2}/dispatch", 
            params={"user_id": _get_user_id()},
            json={"command": f"stat {n2_dir}/{workspace_id}"}, 
            headers=_headers()
        )
        assert "No such file or directory" in r_d2.json().get("stderr", "") or r_d2.json().get("status") != "successful", (
            f"Node 2 failed to purge its physical tmp folder: {r_d2.text}"
        )
        print("[Case Purge] ✅ Physical client-side (`/tmp/cortex-sync/...`) folders proactively erased on all nodes")

        # 6. Check server-side folder
        if is_native:
            mirror_path = os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(__file__))), "ai-hub/data/mirrors", workspace_id)
            assert not os.path.exists(mirror_path), f"Server mirror folder still physically exists! stat matched: {mirror_path}"
        else:
            # (Since the test runner is executed on host but ai_hub is Docker container, we can use docker exec)
            cmd = ["docker", "exec", "ai_hub_service", "stat", f"/app/data/mirrors/{workspace_id}"]
            # This should fail if it doesn't exist.
            res_hub = subprocess.run(cmd, capture_output=True, text=True)
            assert res_hub.returncode != 0, f"Server mirror folder still physically exists! stat matched: {res_hub.stdout}"
            assert "No such file or directory" in res_hub.stderr, f"Unexpected error during server stat: {res_hub.stderr}"
        
        print("[Case Purge] ✅ Server-side physical mirror folder proactively erased")

