From fd73de357303bd96816af5a4453f18d9f833c2fe Mon Sep 17 00:00:00 2001 From: boy Date: Mon, 13 Jul 2026 21:27:33 +0200 Subject: [PATCH 1/5] fix(test): eliminate false positive in Langfuse session leak test MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The previous leak check flagged legitimate step-1 traces (litellm-proxy, agent-completion) as 'leaks' because they appeared in the 20 most recent traces alongside step-2 traces and shared the same session_id — even though they were correctly attributed to step 1, not step 2. This was NOT an OTel contextvar race condition; it was a test methodology bug. A single chat request produces multiple Langfuse traces (parent triage, child litellm-proxy, downstream agent-completion), all correctly carrying session_id. The test found ONE as session_trace_id and flagged the others. Changes: - Collect ALL step-1 trace IDs into a set (not just one). - Record step-2 request timestamp for time-based filtering. - Only examine traces with timestamps AFTER step-2 cutoff, excluding all step-1 IDs. - Expand fetch limit from 20 to 50 for better coverage. - Add diagnostic info (trace counts, excluded IDs) to check messages. --- .../verify_canonical_endpoints.py | 87 ++++++++++++------- 1 file changed, 57 insertions(+), 30 deletions(-) diff --git a/scripts/verification/verify_canonical_endpoints.py b/scripts/verification/verify_canonical_endpoints.py index a0a17045..19c68797 100644 --- a/scripts/verification/verify_canonical_endpoints.py +++ b/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,14 +427,18 @@ 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: @@ -441,13 +446,14 @@ def test_langfuse_session_propagation(cfg: dict) -> tuple[int, int]: 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 + step1_trace_ids.add(t["id"]) + if session_trace_id is None: + session_trace_id = t["id"] if session_trace_id: break except Exception as e: if _attempt == 9: - print(f" ⚠ trace poll error: {str(e)[:100]}") + print(f" warning trace poll error: {str(e)[:100]}") continue # Report session trace result @@ -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,36 +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) + # Wait for Langfuse SDK auto-flush (poll up to 10s for leak check). + # We poll until step 2 traces appear (session_trace_id is no longer at + # index 0, meaning newer traces from step 2 have been flushed), then + # check only traces with timestamps AFTER step 2 for session leaks. session_trace_visible = False - traces2 = [] + step2_cutoff_iso = None + step2_traces = [] 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] + all_traces = resp2.json().get("data", []) + recent_ids = [t["id"] for t in all_traces] if session_trace_id and session_trace_id in recent_ids: if recent_ids.index(session_trace_id) > 0: session_trace_visible = True + # Filter to only traces after step 2, excluding step 1 IDs. + # The step 2 request timestamp is a few ms before the router + # creates traces, so use a small safety window. + step2_cutoff_iso = ( + datetime.datetime.fromtimestamp( + step2_request_time - 2, tz=datetime.timezone.utc + ).isoformat() + ) + step2_traces = [ + t for t in all_traces + if ( + t["id"] not in step1_trace_ids + and t.get("timestamp", "") > step2_cutoff_iso + ) + ] break # 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: - print(f" ⚠ trace poll error: {str(e)[:100]}") + print(f" warning trace poll error: {str(e)[:100]}") continue # Report leak test result @@ -525,36 +549,39 @@ def test_langfuse_session_propagation(cfg: dict) -> tuple[int, int]: # We cannot distinguish between "no leak" and "session trace still unflushed." # Don't count this as a failure — just warn and move on. if session_trace_id is None: - print(" ⚠ No session leak — INCONCLUSIVE (session trace not found in step 1 polling)") + print(" warning 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 session_trace_visible: print( - " ⚠ No session leak — INCONCLUSIVE " + " warning 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).""" @@ -677,4 +704,4 @@ def main(): if __name__ == "__main__": - main() + main() \ No newline at end of file From 9f6e1242f366a708411df9cfbe1f68c57668ffe4 Mon Sep 17 00:00:00 2001 From: boy Date: Mon, 13 Jul 2026 22:01:56 +0200 Subject: [PATCH 2/5] fix(test): robust timestamp parsing and continuous trace polling MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Addresses CodeRabbit and Gemini Code Assist review findings on PR#326: 1. Parse timestamps as timezone-aware datetime objects instead of lexicographic string comparison — avoids Z/+00:00 suffix mismatch and sub-second precision differences. 2. Remove early break in step-1 trace polling loop to ensure all sibling traces (litellm-proxy, agent-completion, triage-parent) are collected in step1_trace_ids. 3. Replace step-2 early break with continuous polling across the full 10s window. Accumulate step2_traces with dedup via seen_step2_ids to prevent missing late-flushing leaked traces. All 193 unit tests and 21/21 E2E tests pass. --- .../verify_canonical_endpoints.py | 47 ++++++++++++------- 1 file changed, 29 insertions(+), 18 deletions(-) diff --git a/scripts/verification/verify_canonical_endpoints.py b/scripts/verification/verify_canonical_endpoints.py index 19c68797..4f96e552 100644 --- a/scripts/verification/verify_canonical_endpoints.py +++ b/scripts/verification/verify_canonical_endpoints.py @@ -449,8 +449,8 @@ def test_langfuse_session_propagation(cfg: dict) -> tuple[int, int]: step1_trace_ids.add(t["id"]) if session_trace_id is None: session_trace_id = t["id"] - if session_trace_id: - break + # Continue polling for remaining seconds to collect + # all sibling traces (they may flush asynchronously) except Exception as e: if _attempt == 9: print(f" warning trace poll error: {str(e)[:100]}") @@ -501,8 +501,14 @@ def test_langfuse_session_propagation(cfg: dict) -> tuple[int, int]: # index 0, meaning newer traces from step 2 have been flushed), then # check only traces with timestamps AFTER step 2 for session leaks. session_trace_visible = False - step2_cutoff_iso = None + # 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: @@ -517,22 +523,27 @@ def test_langfuse_session_propagation(cfg: dict) -> tuple[int, int]: if session_trace_id and session_trace_id in recent_ids: if recent_ids.index(session_trace_id) > 0: session_trace_visible = True - # Filter to only traces after step 2, excluding step 1 IDs. - # The step 2 request timestamp is a few ms before the router - # creates traces, so use a small safety window. - step2_cutoff_iso = ( - datetime.datetime.fromtimestamp( - step2_request_time - 2, tz=datetime.timezone.utc - ).isoformat() + # 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") ) - step2_traces = [ - t for t in all_traces - if ( - t["id"] not in step1_trace_ids - and t.get("timestamp", "") > step2_cutoff_iso - ) - ] - break + 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. From 561119c4d569ee421adf067d5420b259431784a7 Mon Sep 17 00:00:00 2001 From: boy Date: Mon, 13 Jul 2026 22:09:07 +0200 Subject: [PATCH 3/5] chore: remove redundant local import time, add trailing newline Addressing minor review suggestions on PR#326: - Remove redundant local 'import time' in native_agy_stream_generator() (already available at module level, line 8) - Add missing trailing newline to verify_canonical_endpoints.py --- router/main.py | 1 - scripts/verification/verify_canonical_endpoints.py | 2 +- 2 files changed, 1 insertion(+), 2 deletions(-) diff --git a/router/main.py b/router/main.py index d2d937b0..567f6f09 100644 --- a/router/main.py +++ b/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 diff --git a/scripts/verification/verify_canonical_endpoints.py b/scripts/verification/verify_canonical_endpoints.py index 4f96e552..4f14a33c 100644 --- a/scripts/verification/verify_canonical_endpoints.py +++ b/scripts/verification/verify_canonical_endpoints.py @@ -715,4 +715,4 @@ def main(): if __name__ == "__main__": - main() \ No newline at end of file + main() From 2eadca62ea55dfa0fd982b42fa26f35a58e0cf8f Mon Sep 17 00:00:00 2001 From: boy Date: Wed, 22 Jul 2026 15:58:19 +0200 Subject: [PATCH 4/5] fix(test): replace index-based trace visibility with step2_traces check & standardize warning markers --- .../verify_canonical_endpoints.py | 19 ++++++------------- 1 file changed, 6 insertions(+), 13 deletions(-) diff --git a/scripts/verification/verify_canonical_endpoints.py b/scripts/verification/verify_canonical_endpoints.py index 4f14a33c..9d20b9e2 100644 --- a/scripts/verification/verify_canonical_endpoints.py +++ b/scripts/verification/verify_canonical_endpoints.py @@ -453,7 +453,7 @@ def test_langfuse_session_propagation(cfg: dict) -> tuple[int, int]: # all sibling traces (they may flush asynchronously) except Exception as e: if _attempt == 9: - print(f" warning trace poll error: {str(e)[:100]}") + print(f" ⚠ trace poll error: {str(e)[:100]}") continue # Report session trace result @@ -497,10 +497,7 @@ def test_langfuse_session_propagation(cfg: dict) -> tuple[int, int]: return passed, total # Wait for Langfuse SDK auto-flush (poll up to 10s for leak check). - # We poll until step 2 traces appear (session_trace_id is no longer at - # index 0, meaning newer traces from step 2 have been flushed), then - # check only traces with timestamps AFTER step 2 for session leaks. - session_trace_visible = False + # 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. @@ -519,10 +516,6 @@ def test_langfuse_session_propagation(cfg: dict) -> tuple[int, int]: if resp2.status_code != 200: continue all_traces = resp2.json().get("data", []) - recent_ids = [t["id"] for t in all_traces] - if session_trace_id and session_trace_id in recent_ids: - if recent_ids.index(session_trace_id) > 0: - session_trace_visible = True # 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. @@ -552,7 +545,7 @@ def test_langfuse_session_propagation(cfg: dict) -> tuple[int, int]: break except Exception as e: if _attempt2 == 9: - print(f" warning trace poll error: {str(e)[:100]}") + print(f" ⚠ trace poll error: {str(e)[:100]}") continue # Report leak test result @@ -560,13 +553,13 @@ def test_langfuse_session_propagation(cfg: dict) -> tuple[int, int]: # We cannot distinguish between "no leak" and "session trace still unflushed." # Don't count this as a failure — just warn and move on. if session_trace_id is None: - print(" warning No session leak — INCONCLUSIVE (session trace not found in step 1 polling)") + print(" ⚠ No session leak — INCONCLUSIVE (session trace not found in step 1 polling)") return passed, total total += 1 - if not session_trace_visible: + if not step2_traces: print( - " warning No session leak — INCONCLUSIVE " + " ⚠ No session leak — INCONCLUSIVE " "(second request trace not flushed within 10s)" ) return passed, total From e7cd3deaafbc83ad93f4629e5c3e6105cd89d8a0 Mon Sep 17 00:00:00 2001 From: boy Date: Wed, 22 Jul 2026 15:58:52 +0200 Subject: [PATCH 5/5] chore: standardize production pod naming fallback to prod-router-pod (Closes #341) --- README.md | 8 ++++---- scripts/backup.sh | 2 +- scripts/verification/verify_canonical_endpoints.py | 2 +- start-stack.sh | 2 +- 4 files changed, 7 insertions(+), 7 deletions(-) diff --git a/README.md b/README.md index 5aaf9309..061ec42b 100644 --- a/README.md +++ b/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 ``` --- diff --git a/scripts/backup.sh b/scripts/backup.sh index 5f873cb8..247e9765 100755 --- a/scripts/backup.sh +++ b/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" diff --git a/scripts/verification/verify_canonical_endpoints.py b/scripts/verification/verify_canonical_endpoints.py index 9d20b9e2..ff2c2497 100644 --- a/scripts/verification/verify_canonical_endpoints.py +++ b/scripts/verification/verify_canonical_endpoints.py @@ -657,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() diff --git a/start-stack.sh b/start-stack.sh index 657095be..6087a442 100755 --- a/start-stack.sh +++ b/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}"