-
-
CPU
-
{latestTelemetry.cpu.toFixed(1)}%
+ {/* Triggers Summary */}
+
+ {triggers.length === 0 ? (
+ No Triggers
+ ) : (
+ triggers.map(t => (
+
+ {t.trigger_type === 'cron' && `β° ${t.cron_expression}`}
+ {t.trigger_type === 'interval' && `π ${t.interval_seconds}s`}
+ {t.trigger_type === 'webhook' && `π Webhook`}
+ {t.trigger_type === 'manual' && `ποΈ Manual`}
+
+ ))
+ )}
+
+
+
+
+ Runs
+ {agent?.total_runs || 0}
-
-
RAM
-
{latestTelemetry.ram}MB
+
+ Success
+ {agent?.total_runs ? Math.round(((agent?.successful_runs || 0) / agent.total_runs) * 100) : 0}%
+
+
+ Last Run
+ {formatTimeLocal(agent?.last_heartbeat)}
-
-
-
-
-
-
-
-
-
-
- null} />
-
-
-
+
@@ -524,28 +530,20 @@
-
-
diff --git a/frontend/src/services/api/agentService.js b/frontend/src/services/api/agentService.js
index 42ac720..db58304 100644
--- a/frontend/src/services/api/agentService.js
+++ b/frontend/src/services/api/agentService.js
@@ -1,7 +1,7 @@
import { fetchWithAuth } from './apiClient';
export const getAgents = async () => {
- return await fetchWithAuth(`/agents`);
+ return await fetchWithAuth(`/agents?_t=${Date.now()}`);
};
export const getAgentTelemetry = async (id) => {
@@ -55,3 +55,9 @@
method: 'DELETE'
});
};
+
+export const resetAgentMetrics = async (id) => {
+ return await fetchWithAuth(`/agents/${id}/metrics/reset`, {
+ method: 'POST'
+ });
+};
diff --git a/test_run.log b/test_run.log
new file mode 100644
index 0000000..57a30ec
--- /dev/null
+++ b/test_run.log
@@ -0,0 +1,31 @@
+==========================================
+ CORTEX HUB INTEGRATION TESTS SETUP
+==========================================
+Docker daemon not reachable (likely running in a Dev Container). Switching to Native Python mode...
+Starting AI Hub natively in the background...
+Waiting for AI Hub to be ready...
+Waiting for AI Hub Backend natively...
+AI Hub Backend is online.
+==========================================
+ EXECUTING E2E INTEGRATION SUITE
+==========================================
+/app/run_integration_tests.sh: line 100: /tmp/venv/bin/activate: No such file or directory
+No venv found, hoping pytest is in global PATH.
+============================= test session starts ==============================
+platform linux -- Python 3.11.13, pytest-9.0.2, pluggy-1.6.0 -- /usr/local/bin/python3.11
+cachedir: .pytest_cache
+rootdir: /
+plugins: mock-3.15.1, anyio-4.12.1, trio-0.8.0, tornasync-0.6.0.post2, asyncio-1.3.0
+asyncio: mode=Mode.STRICT, debug=False, asyncio_default_fixture_loop_scope=None, asyncio_default_test_loop_scope=function
+collecting ... ERROR: file or directory not found: ai-hub/integration_tests/
+
+collected 0 items
+
+=============================== warnings summary ===============================
+home/vscode/.local/lib/python3.11/site-packages/_pytest/cacheprovider.py:475
+ /home/vscode/.local/lib/python3.11/site-packages/_pytest/cacheprovider.py:475: PytestCacheWarning: could not create cache path /.pytest_cache/v/cache/nodeids: [Errno 13] Permission denied: '/pytest-cache-files-z77lkhsd'
+ config.cache.set("cache/nodeids", sorted(self.cached_nodeids))
+
+-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html
+============================== 1 warning in 0.00s ==============================
+Terminated
diff --git a/test_sandbox/test_local.py b/test_sandbox/test_local.py
new file mode 100644
index 0000000..b00fa69
--- /dev/null
+++ b/test_sandbox/test_local.py
@@ -0,0 +1,943 @@
+"""
+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")
+