fix(test): eliminate false positive in Langfuse session leak test

README.md

@@ -8,7 +8,7 @@ The gateway exposes a unified OpenAI-compatible endpoint that dynamically assess
 
 ## 1. System Architecture
 
-The gateway runs as a rootless Podman pod (`agent-router-pod`) utilizing **Host Networking** (`hostNetwork: true`). This design eliminates complex container network bridges, allowing microservices to communicate with extremely low latency and bind directly to localhost ports, matching the behavior of your native services (such as your local GPU-accelerated `llama-server`).
+The gateway runs as a rootless Podman pod (`prod-router-pod`) utilizing **Host Networking** (`hostNetwork: true`). This design eliminates complex container network bridges, allowing microservices to communicate with extremely low latency and bind directly to localhost ports, matching the behavior of your native services (such as your local GPU-accelerated `llama-server`).
 
 ### High-Level Topology
 
@@ -397,10 +397,10 @@ Run the startup script from the root of the repository:
 *Note: If running for the first time, the script will prompt you for your `OpenRouter API Key`, securely saving it inside `.env` with restrictive permissions (`chmod 600`). The script also automatically generates and persists secure random secrets (`LITELLM_MASTER_KEY`, `POSTGRES_PASSWORD`, `NEXTAUTH_SECRET`, `SALT`, `ENCRYPTION_KEY`, and `ROUTER_API_KEY`) to this file on startup if they are missing.*
 
 ### 2. Verify Container Status
-Check that all **10 containers** inside `agent-router-pod` are up and running:
+Check that all **10 containers** inside `prod-router-pod` are up and running:
 ```bash
 podman pod ps
-podman ps --pod --filter pod=agent-router-pod
+podman ps --pod --filter pod=prod-router-pod
 ```
 Your output should display:
 * `valkey-cache` (Redis-compatible cache)
@@ -491,7 +491,7 @@ curl -s http://127.0.0.1:5000/v1/chat/completions \
 
 Check the triage classification and model cascades by viewing the router container's standard output logs:
 ```bash
-podman logs agent-router-pod-llm-triage-router
+podman logs prod-router-pod-llm-triage-router
 ```
 
 ---

router/main.py

@@ -2225,7 +2225,6 @@ async def chat_completions(request: Request):
                             # Real native stream generator
                             async def native_agy_stream_generator(stream_gen, model_name):
                                 """Asynchronous generator yielding native OpenAI-compatible streaming chunks from the real agy daemon."""
-                                import time
                                 created_time = int(time.time())
                                 chunk_id = f"chatcmpl-{uuid.uuid4().hex[:12]}"
                                 token_count = 0

scripts/backup.sh

@@ -20,7 +20,7 @@ fi
 if [ -n "${DEV_ENV_FILE:-}" ] && [ -f "$DEV_ENV_FILE" ]; then
     set -a; source "$DEV_ENV_FILE"; set +a
 fi
-POD_NAME="${POD_NAME:-agent-router-pod}"
+POD_NAME="${POD_NAME:-prod-router-pod}"
 POSTGRES_PORT="${POSTGRES_PORT:-5432}"
 
 mkdir -p "$BACKUP_DIR"

scripts/verification/verify_canonical_endpoints.py

@@ -12,6 +12,7 @@
 import sys
 import json
 import time
+import datetime
 import argparse
 import httpx
 from pathlib import Path
@@ -426,25 +427,30 @@ def test_langfuse_session_propagation(cfg: dict) -> tuple[int, int]:
         passed += check("Session request (200)", False, str(e)[:100])
         return passed, total
 
-    # Wait for Langfuse SDK auto-flush (poll up to 10s)
+    # Wait for Langfuse SDK auto-flush (poll up to 10s).
+    # Collect ALL trace IDs created by step 1 (a single request produces
+    # multiple traces: litellm-proxy, agent-completion, triage-* parent).
+    # We need the full set to exclude them from the leak check below.
     session_trace_id = None
+    step1_trace_ids = set()
     _attempt = 0
     for _attempt in range(10):
         time.sleep(1)
         try:
             resp = httpx.get(
-                f"{lf_base}/api/public/traces?page=1&limit=20&orderBy=timestamp.desc",
+                f"{lf_base}/api/public/traces?page=1&limit=50&orderBy=timestamp.desc",
                 auth=auth, timeout=10,
             )
             if resp.status_code != 200:
                 continue
             traces = resp.json().get("data", [])
             for t in traces:
                 if t.get("sessionId") == session_id and t.get("userId") == user_id:
-                    session_trace_id = t["id"]
-                    break
-            if session_trace_id:
-                break
+                    step1_trace_ids.add(t["id"])
+                    if session_trace_id is None:
+                        session_trace_id = t["id"]
+            # Continue polling for remaining seconds to collect
+            # all sibling traces (they may flush asynchronously)
         except Exception as e:
             if _attempt == 9:
                 print(f"  ⚠ trace poll error: {str(e)[:100]}")
@@ -455,7 +461,8 @@ def test_langfuse_session_propagation(cfg: dict) -> tuple[int, int]:
     if session_trace_id:
         passed += check(
             "Trace has session+user", True,
-            f"session={session_id[:12]} user={user_id} (found after {_attempt+1}s)",
+            f"session={session_id[:12]} user={user_id} "
+            f"(found after {_attempt+1}s, {len(step1_trace_ids)} matching traces)",
         )
     else:
         passed += check(
@@ -470,6 +477,7 @@ def test_langfuse_session_propagation(cfg: dict) -> tuple[int, int]:
         "messages": [{"role": "user", "content": "Say 'leaktest' and nothing else."}],
         "max_tokens": 5,
     }
+    step2_request_time = time.time()
     try:
         r2 = httpx.post(
             f"{router_base}/v1/chat/completions",
@@ -488,32 +496,52 @@ def test_langfuse_session_propagation(cfg: dict) -> tuple[int, int]:
         passed += check("No-session request (200)", False, str(e)[:100])
         return passed, total
 
-    # Wait for Langfuse SDK auto-flush (poll up to 10s for leak check)
-    session_trace_visible = False
-    traces2 = []
+    # Wait for Langfuse SDK auto-flush (poll up to 10s for leak check).
+    # Collect traces with timestamps AFTER step 2 for session leak verification.
+    # Compute cutoff as timezone-aware datetime for robust comparison.
+    # Parsing timestamps avoids lexicographic pitfalls between Z/+00:00
+    # suffixes and differing sub-second precision.
+    step2_cutoff = datetime.datetime.fromtimestamp(
+        step2_request_time - 2, tz=datetime.timezone.utc
+    )
+    step2_traces = []
+    seen_step2_ids = set()
     for _attempt2 in range(10):
         time.sleep(1)
         try:
             resp2 = httpx.get(
-                f"{lf_base}/api/public/traces?page=1&limit=20&orderBy=timestamp.desc",
+                f"{lf_base}/api/public/traces?page=1&limit=50&orderBy=timestamp.desc",
                 auth=auth, timeout=10,
             )
             if resp2.status_code != 200:
                 continue
-            traces2 = resp2.json().get("data", [])
-            # Ensure the second request's trace has been flushed before checking for leaks.
-            # The session trace from step 1 is already in Langfuse, so we wait until
-            # a NEWER trace appears (session_trace_id is NOT at index 0).
-            recent_ids = [t["id"] for t in traces2]
-            if session_trace_id and session_trace_id in recent_ids:
-                if recent_ids.index(session_trace_id) > 0:
-                    session_trace_visible = True
-                    break
+            all_traces = resp2.json().get("data", [])
+            # Accumulate step-2 traces across all polls — do NOT break early.
+            # Sibling traces may flush at different times; collecting across
+            # the full 10s window ensures we don't miss a late-flushing leak.
+            for t in all_traces:
+                tid = t["id"]
+                if tid in seen_step2_ids or tid in step1_trace_ids:
+                    continue
+                t_ts = t.get("timestamp", "")
+                if not t_ts:
+                    continue
+                try:
+                    # Normalize 'Z' suffix for fromisoformat (Python 3.11+ handles
+                    # Z natively, but we support older Pythons).
+                    t_dt = datetime.datetime.fromisoformat(
+                        t_ts.replace("Z", "+00:00")
+                    )
+                    if t_dt > step2_cutoff:
+                        step2_traces.append(t)
+                        seen_step2_ids.add(tid)
+                except ValueError:
+                    pass
             # If session trace hasn't appeared yet, continue waiting
             if not session_trace_id:
                 # Session trace was never found in step 1; leak check is inconclusive.
                 # Poll for remaining time, then break without a definitive answer.
-                if _attempt2 >= 5 and len(traces2) > 0:
+                if _attempt2 >= 5 and len(all_traces) > 0:
                     break
         except Exception as e:
             if _attempt2 == 9:
@@ -527,34 +555,37 @@ def test_langfuse_session_propagation(cfg: dict) -> tuple[int, int]:
     if session_trace_id is None:
         print("  ⚠ No session leak — INCONCLUSIVE (session trace not found in step 1 polling)")
         return passed, total
-    leaked = False
-    if session_trace_visible and traces2:
-        leaked = any(
-            t.get("sessionId") == session_id and t.get("id") != session_trace_id
-            for t in traces2
-        )
-    elif _attempt2 == 9:
+
+    total += 1
+    if not step2_traces:
         print(
             "  ⚠ No session leak — INCONCLUSIVE "
             "(second request trace not flushed within 10s)"
         )
         return passed, total
-    total += 1
+
+    # Only check step 2 traces (filtered by timestamp and step1 exclusion)
+    leaked = any(
+        t.get("sessionId") == session_id
+        for t in step2_traces
+    )
 
     if leaked:
         passed += check(
             "No session leak", False,
-            "Previous session leaked into no-session request!"
+            f"Previous session leaked into no-session request! "
+            f"({len(step2_traces)} step-2 traces checked, "
+            f"{len(step1_trace_ids)} step-1 IDs excluded)"
         )
     else:
         passed += check(
             "No session leak", True,
-            f"{len(traces2)} recent traces clean (excluded known session trace)"
+            f"{len(step2_traces)} step-2 traces clean "
+            f"({len(step1_trace_ids)} step-1 IDs excluded)"
         )
 
     return passed, total
 
-
 def test_canonical_urls(cfg: dict) -> tuple[int, int, int]:
     """Verify canonical HTTPS URLs are reachable (if PUBLIC_BASE_URL is set).
     Returns (passed, total, skipped)."""
@@ -626,7 +657,7 @@ def main():
         "--dev", action="store_true", help="Test dev environment (dev-router-pod)"
     )
     parser.add_argument(
-        "--prod", action="store_true", help="Test prod environment (agent-router-pod, default)"
+        "--prod", action="store_true", help="Test prod environment (prod-router-pod, default)"
     )
     args = parser.parse_args()
 

start-stack.sh

@@ -78,7 +78,7 @@ if [ -n "${DEV_ENV_FILE:-}" ] && [ -f "$DEV_ENV_FILE" ]; then
 fi
 
 # Port assignments — read from env (set by .env or .env.dev) with prod defaults
-POD_NAME="${POD_NAME:-agent-router-pod}"
+POD_NAME="${POD_NAME:-prod-router-pod}"
 ROUTER_PORT="${ROUTER_PORT:-5000}"
 LITELLM_PORT="${LITELLM_PORT:-4000}"
 LANGFUSE_WEB_PORT="${LANGFUSE_WEB_PORT:-3001}"