From f2bce18abfcd96f9312368665964a057d92e5475 Mon Sep 17 00:00:00 2001 From: HumbleTurtle Date: Tue, 5 May 2026 03:55:32 -0300 Subject: [PATCH] Restore gateway runtime verification scripts --- scripts/benchmark_terminal_pty_latency.py | 117 ++++++++++++++++++++ scripts/test_gateway_vault_setup_payload.py | 75 +++++++++++++ scripts/test_remote_terminal_gateway.py | 4 +- 3 files changed, 194 insertions(+), 2 deletions(-) create mode 100644 scripts/benchmark_terminal_pty_latency.py create mode 100644 scripts/test_gateway_vault_setup_payload.py diff --git a/scripts/benchmark_terminal_pty_latency.py b/scripts/benchmark_terminal_pty_latency.py new file mode 100644 index 0000000..64f5a01 --- /dev/null +++ b/scripts/benchmark_terminal_pty_latency.py @@ -0,0 +1,117 @@ +#!/usr/bin/env python3 +"""Benchmark gateway-side PTY echo latency and output flush policy costs.""" + +from __future__ import annotations + +import json +import os +import statistics +import sys +import time +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT / "resources" / "scripts")) + +from terminal.pty_manager import PtySessionManager # noqa: E402 + + +def percentile(values: list[float], pct: float) -> float: + if not values: + return 0.0 + ordered = sorted(values) + index = min(len(ordered) - 1, int((pct / 100.0) * len(ordered))) + return ordered[index] + + +def summary(values: list[float]) -> dict[str, float]: + return { + "p50_ms": round(percentile(values, 50), 3), + "p95_ms": round(percentile(values, 95), 3), + "p99_ms": round(percentile(values, 99), 3), + "max_ms": round(max(values) if values else 0.0, 3), + "mean_ms": round(statistics.mean(values) if values else 0.0, 3), + } + + +def benchmark_pty_echo(iterations: int) -> dict[str, float]: + manager = PtySessionManager(ring_chunks=max(iterations * 2, 128)) + info = manager.open("cat", rows=24, cols=80) + terminal_id = info.terminal_id + latencies: list[float] = [] + pending: dict[bytes, float] = {} + received = bytearray() + + def on_output(_terminal_id: str, _seq: int, data: bytes) -> None: + received.extend(data) + for marker, start in list(pending.items()): + if marker in received: + latencies.append((time.perf_counter() - start) * 1000) + del pending[marker] + + unsubscribe = manager.subscribe(terminal_id, on_output) + try: + for index in range(iterations): + marker = f"n{index:04d}".encode("ascii") + pending[marker] = time.perf_counter() + manager.write(terminal_id, marker + b"\n", input_seq=index + 1) + deadline = time.perf_counter() + 2 + while marker in pending and time.perf_counter() < deadline: + time.sleep(0.0005) + if marker in pending: + raise TimeoutError(f"timed out waiting for echo {marker!r}") + finally: + unsubscribe() + manager.close(terminal_id) + return summary(latencies) + + +def flush_policy_cost(iterations: int, immediate_bytes: int, timer_ms: int) -> dict[str, int | float]: + immediate_flushes = 0 + timer_flushes = 0 + bulk_flushes = 0 + buffered = 0 + timer_pending = False + chunks = [1, 2, 8, 32, 128, 512, 4096] + start = time.perf_counter() + for index in range(iterations): + size = chunks[index % len(chunks)] + buffered += size + if size <= immediate_bytes: + immediate_flushes += 1 + buffered = 0 + timer_pending = False + elif buffered >= 4096: + bulk_flushes += 1 + buffered = 0 + timer_pending = False + elif not timer_pending: + timer_flushes += 1 + timer_pending = timer_ms > 0 + elapsed_ms = (time.perf_counter() - start) * 1000 + return { + "immediate_bytes": immediate_bytes, + "timer_ms": timer_ms, + "elapsed_ms": round(elapsed_ms, 3), + "immediate_flushes": immediate_flushes, + "timer_flushes": timer_flushes, + "bulk_flushes": bulk_flushes, + } + + +def main() -> None: + iterations = int(os.environ.get("NAPSEER_TERMINAL_BENCH_ITERATIONS", "250")) + immediate_bytes = int(os.environ.get("NAPSEER_GATEWAY_PTY_IMMEDIATE_FLUSH_BYTES", "256")) + timer_ms = int(os.environ.get("NAPSEER_GATEWAY_PTY_INTERACTIVE_FLUSH_MS", "0")) + result = { + "iterations": iterations, + "gateway": { + "pty_echo_cat": benchmark_pty_echo(iterations), + "flush_policy": flush_policy_cost(iterations * 20, immediate_bytes, timer_ms), + }, + } + print(json.dumps(result, indent=2)) + + +if __name__ == "__main__": + main() diff --git a/scripts/test_gateway_vault_setup_payload.py b/scripts/test_gateway_vault_setup_payload.py new file mode 100644 index 0000000..c7dcaba --- /dev/null +++ b/scripts/test_gateway_vault_setup_payload.py @@ -0,0 +1,75 @@ +#!/usr/bin/env python3 +"""Smoke-test gateway-owned project vault setup payload generation.""" + +import base64 +import importlib.util +import pathlib +import sys + + +def load_module(): + script_path = pathlib.Path(__file__).resolve().parents[1] / "resources" / "scripts" / "napseer_mcp_server.py" + sys.path.insert(0, str(script_path.parent)) + spec = importlib.util.spec_from_file_location("napseer_mcp_server", script_path) + module = importlib.util.module_from_spec(spec) + assert spec and spec.loader + spec.loader.exec_module(module) + return module + + +def run(): + mod = load_module() + writes = [] + + mod.AUTH = {"agent_id": "gateway-agent-1"} + mod.VAULT_KEY = bytes(range(32)) + mod.VAULT_SECRETS = {} + mod.persist_vault_secrets = lambda: writes.append(dict(mod.VAULT_SECRETS)) + + payload = mod.gateway_generate_project_vault_setup_payload("project-1", "setup-1") + assert mod.VAULT_KDF_ITERATIONS == 600000 + assert writes, "payload generation should persist local encrypted vault state" + assert set(item["secret_kind"] for item in payload["project_secrets"]) == {"chat", "tabs", "gateway"} + assert payload["wrapped_master_secret"]["credential_id"] == "gateway-vault:gateway-agent-1" + assert payload["wrapped_master_secret"]["kdf_params"]["iterations"] == 600000 + + for item in payload["project_secrets"]: + envelope = item["encrypted_secret_envelope"] + assert envelope["schema_version"] == 1 + assert envelope["alg"] == "AES-GCM-256" + assert envelope["key_id"] == "vault-master:project-1:v1" + assert "plaintext" not in envelope + assert "content" not in envelope + assert len(base64.b64decode(envelope["nonce_b64"])) == 12 + assert len(base64.b64decode(envelope["ciphertext_b64"])) == envelope["payload_size_bytes"] + assert envelope["payload_size_bytes"] > 16 + + stored = mod.VAULT_SECRETS["project_vaults"]["project-1"] + assert stored["setup_request_id"] == "setup-1" + assert len(base64.b64decode(stored["vault_master_secret_b64"])) == 32 + assert set(stored["project_secrets"]) == {"chat", "tabs", "gateway"} + + writes.clear() + calls = [] + mod.DEFAULT_PROJECT_ID = "project-1" + mod.require_unlocked = lambda operation="operation": None + mod.request_project_write = lambda method, path, body, project_id, purpose, scope_type="project": calls.append( + (method, path, body, project_id, purpose, scope_type) + ) or { + "id": "secret-version-2", + "version": 2, + "secret_kind": "chat", + } + rotated = mod.gateway_rotate_project_vault_secret({"project_id": "project-1", "secret_kind": "chat"}) + assert rotated["version"] == 2 + assert calls[0][0] == "POST" + assert calls[0][1] == "/v1/projects/project-1/vault/secrets/chat/rotate" + assert calls[0][5] == "encryption" + assert calls[0][2]["encrypted_secret_envelope"]["key_id"] == "vault-master:project-1:v1" + assert mod.VAULT_SECRETS["project_vaults"]["project-1"]["project_secrets"]["chat"]["version"] == 2 + + print("ok: gateway vault setup payload smoke passed") + + +if __name__ == "__main__": + run() diff --git a/scripts/test_remote_terminal_gateway.py b/scripts/test_remote_terminal_gateway.py index 9c37d56..a3fc260 100644 --- a/scripts/test_remote_terminal_gateway.py +++ b/scripts/test_remote_terminal_gateway.py @@ -202,7 +202,7 @@ def test_rotation_required_session_skips_terminal_and_schedule_bootstrap(module) module.AUTH = {"agent_id": "agent-1", "worker_id": "worker-1"} module.current_project_id = lambda: "project-1" module.read_gateway_relay_secret = lambda: b"relay-secret" - module.relay_ws_url = lambda project_id, session_id, ticket: "ws://relay.test" + module.relay_ws_url = lambda project_id, session_id, ticket, socket="terminal": "ws://relay.test" module.ws_connect = lambda url: FakeSocket() module.ws_read_text = lambda sock: next(ws_frames) module.ws_send_text = lambda sock, text: sent.append(text) @@ -216,7 +216,7 @@ def test_rotation_required_session_skips_terminal_and_schedule_bootstrap(module) module.relay_key_from_secret = lambda secret, context, direction: b"k" + direction.encode("utf-8") module.relay_context_hash = lambda context: "ctx" module.encrypt_relay_frame = ( - lambda key, session_id, payload, context_hash, direction, seq: ( + lambda key, session_id, payload, context_hash, relay_lane, direction, seq: ( '{"type":"encrypted","payload":' + json.dumps(payload, separators=(",", ":")) + "}" ) )