From 8afab5fa49a9201ffe5786612d199c119b3e8ba6 Mon Sep 17 00:00:00 2001 From: huanghuoguoguo <60681390+huanghuoguoguo@users.noreply.github.com> Date: Sun, 19 Jul 2026 22:34:09 +0800 Subject: [PATCH] fix(mcp): recover box stdio sessions after disconnect --- skills/scripts/e2e/agent-run-ledger-audit.py | 127 +++++- skills/scripts/e2e/fake-openai-provider.mjs | 14 +- skills/scripts/e2e/lib/langbot-e2e.mjs | 32 ++ .../e2e/local-agent-steering-debug-chat.mjs | 9 + skills/scripts/e2e/pipeline-debug-chat.mjs | 9 + skills/skills.index.json | 35 +- .../cases/box-mcp-heartbeat-recovery.yaml | 85 ++++ ...cal-agent-effective-prompt-debug-chat.yaml | 5 + .../langbot-testing/fixtures/fixtures.json | 2 +- .../dist/qa-plugin-smoke-0.1.0.lbpkg | Bin 5160 -> 14159 bytes .../probes/box-mcp-heartbeat-recovery.mjs | 396 ++++++++++++++++++ skills/test/lbs-cli.test.ts | 97 +++++ src/langbot/pkg/box/connector.py | 12 +- src/langbot/pkg/provider/tools/loaders/mcp.py | 107 ++++- .../pkg/provider/tools/loaders/mcp_stdio.py | 9 +- tests/unit_tests/box/test_box_connector.py | 21 + .../provider/test_mcp_box_integration.py | 238 ++++++++++- 17 files changed, 1147 insertions(+), 51 deletions(-) create mode 100644 skills/skills/langbot-testing/cases/box-mcp-heartbeat-recovery.yaml create mode 100644 skills/skills/langbot-testing/probes/box-mcp-heartbeat-recovery.mjs diff --git a/skills/scripts/e2e/agent-run-ledger-audit.py b/skills/scripts/e2e/agent-run-ledger-audit.py index a001e03d0..ec969b8aa 100644 --- a/skills/scripts/e2e/agent-run-ledger-audit.py +++ b/skills/scripts/e2e/agent-run-ledger-audit.py @@ -5,8 +5,8 @@ from __future__ import annotations import argparse import asyncio +import datetime import json -import os import pathlib import re import sys @@ -47,7 +47,48 @@ def load_json(value: str | None, *, field: str, failures: list[dict]) -> object: return {} -async def audit(repo: pathlib.Path, run_id: str | None) -> dict: +def parse_created_after(value: str | None) -> datetime.datetime | None: + if not value: + return None + parsed = datetime.datetime.fromisoformat(value.replace("Z", "+00:00")) + if parsed.tzinfo is not None: + parsed = parsed.astimezone(datetime.timezone.utc).replace(tzinfo=None) + return parsed + + +def event_matches_tool_call(data_json: str | None, tool_name: str, parameters: dict | None) -> bool: + try: + data = json.loads(data_json or "{}") + except (TypeError, ValueError): + return False + if not isinstance(data, dict) or data.get("tool_name") != tool_name: + return False + return parameters is None or data.get("parameters") == parameters + + +def collect_result_texts(value: object) -> list[str]: + texts: list[str] = [] + if isinstance(value, dict): + for key, item in value.items(): + if key == "text" and isinstance(item, str): + texts.append(item) + else: + texts.extend(collect_result_texts(item)) + elif isinstance(value, list): + for item in value: + texts.extend(collect_result_texts(item)) + return texts + + +async def audit( + repo: pathlib.Path, + run_id: str | None, + *, + created_after: datetime.datetime | None = None, + expected_tool_name: str | None = None, + expected_parameters: dict | None = None, + expected_result_text: str | None = None, +) -> dict: engine = create_async_engine(database_url(repo)) failures: list[dict] = [] warnings: list[dict] = [] @@ -58,12 +99,41 @@ async def audit(repo: pathlib.Path, run_id: str | None) -> dict: sqlalchemy.text("SELECT * FROM agent_run WHERE run_id = :run_id"), {"run_id": run_id}, )).mappings().first() + elif expected_tool_name: + query = "SELECT * FROM agent_run" + params = {} + if created_after is not None: + query += " WHERE created_at >= :created_after" + params["created_after"] = created_after + query += " ORDER BY id DESC LIMIT 100" + candidates = (await connection.execute(sqlalchemy.text(query), params)).mappings().all() + run_row = None + for candidate in candidates: + started_rows = (await connection.execute( + sqlalchemy.text( + "SELECT data_json FROM agent_run_event " + "WHERE run_id = :run_id AND type = 'tool.call.started' ORDER BY sequence" + ), + {"run_id": str(candidate["run_id"])}, + )).mappings().all() + if any( + event_matches_tool_call(row.get("data_json"), expected_tool_name, expected_parameters) + for row in started_rows + ): + run_row = candidate + break else: run_row = (await connection.execute( sqlalchemy.text("SELECT * FROM agent_run ORDER BY id DESC LIMIT 1") )).mappings().first() if run_row is None: - return {"status": "env_issue", "reason": "No matching AgentRunner run exists.", "failures": [], "warnings": []} + status = "fail" if expected_tool_name else "env_issue" + return { + "status": status, + "reason": "No AgentRunner run contains the expected tool call." if expected_tool_name else "No matching AgentRunner run exists.", + "failures": [{"kind": "expected_tool_call_missing"}] if expected_tool_name else [], + "warnings": [], + } selected_run_id = str(run_row["run_id"]) event_rows = (await connection.execute( @@ -173,6 +243,36 @@ async def audit(repo: pathlib.Path, run_id: str | None) -> dict: if not tools: warnings.append({"kind": "no_authorized_tools", "reason": "The run authorization snapshot exposes no tools."}) + expected_call_summary = None + if expected_tool_name: + matching_starts = [ + item + for items in starts.values() + for item in items + if item["tool_name"] == expected_tool_name + and (expected_parameters is None or item["data"].get("parameters") == expected_parameters) + ] + if len(matching_starts) != 1: + failures.append({"kind": "expected_tool_call_count", "actual": len(matching_starts), "expected": 1}) + matching_completions = [] + for started in matching_starts: + call_id = str(started["data"].get("tool_call_id", "")) + matching_completions.extend(completions.get(call_id, [])) + result_text_match = expected_result_text is None or any( + expected_result_text in collect_result_texts(completed["data"].get("result")) + for completed in matching_completions + ) + if expected_result_text is not None and not result_text_match: + failures.append({"kind": "expected_tool_result_text_missing"}) + expected_call_summary = { + "tool_name": expected_tool_name, + "parameters_match_required": expected_parameters is not None, + "matched_started_count": len(matching_starts), + "matched_completed_count": len(matching_completions), + "result_text_match_required": expected_result_text is not None, + "result_text_match": result_text_match, + } + metrics = { "event_count": len(event_rows), "tool_call_started": sum(len(items) for items in starts.values()), @@ -193,6 +293,7 @@ async def audit(repo: pathlib.Path, run_id: str | None) -> dict: "finished_at": str(run_row["finished_at"]), }, "metrics": metrics, + "expected_tool_call": expected_call_summary, "failures": failures, "warnings": warnings, } @@ -202,10 +303,28 @@ def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("--repo", required=True) parser.add_argument("--run-id") + parser.add_argument("--created-after") + parser.add_argument("--expected-tool-name") + parser.add_argument("--expected-parameters-json") + parser.add_argument("--expected-result-text") parser.add_argument("--output", required=True) args = parser.parse_args() try: - report = asyncio.run(audit(pathlib.Path(args.repo).resolve(), args.run_id)) + expected_parameters = None + if args.expected_parameters_json: + expected_parameters = json.loads(args.expected_parameters_json) + if not isinstance(expected_parameters, dict): + raise ValueError("--expected-parameters-json must decode to an object") + if (expected_parameters is not None or args.expected_result_text) and not args.expected_tool_name: + raise ValueError("--expected-tool-name is required with expected parameters or result text") + report = asyncio.run(audit( + pathlib.Path(args.repo).resolve(), + args.run_id, + created_after=parse_created_after(args.created_after), + expected_tool_name=args.expected_tool_name, + expected_parameters=expected_parameters, + expected_result_text=args.expected_result_text, + )) except Exception as exc: # noqa: BLE001 - probe must classify environment failures report = {"status": "env_issue", "reason": str(exc), "failures": [], "warnings": []} pathlib.Path(args.output).write_text(json.dumps(report, indent=2) + "\n", encoding="utf-8") diff --git a/skills/scripts/e2e/fake-openai-provider.mjs b/skills/scripts/e2e/fake-openai-provider.mjs index da0d88c9c..673f7ddd3 100755 --- a/skills/scripts/e2e/fake-openai-provider.mjs +++ b/skills/scripts/e2e/fake-openai-provider.mjs @@ -585,10 +585,18 @@ function buildResponse(payload) { return buildSummaryResponse(text); } + if (/qa-effective-prompt/i.test(current) && /PROMPT_PREPROCESS_OK/.test(text)) { + return { role: "assistant", content: "PROMPT_PREPROCESS_OK" }; + } + const pluginTool = firstToolName(tools, ["qa_plugin_echo"]); const pluginFailTool = firstToolName(tools, ["qa_plugin_fail"]); const pluginSleepTool = firstToolName(tools, ["qa_plugin_sleep"]); const mcpTool = firstToolName(tools, ["qa_mcp_echo"]); + const requestedMcpEchoText = current.match( + /qa_mcp_echo[\s\S]*?exactly this text:\s*([A-Za-z0-9_:-]+)/i, + )?.[1] || "mcp-ok-local-agent"; + const expectedMcpEchoResult = `qa_mcp_echo:${requestedMcpEchoText}`; if (/STEERING_NO_FOLLOWUP|qa_plugin_sleep|steering-e2e-anchor|qa_steering_sentinel_6194/i.test(current || text)) { if (text.includes(STEERING_FOLLOWUP_SENTINEL) && text.includes(STEERING_SLEEP_RESULT)) { @@ -620,9 +628,9 @@ function buildResponse(payload) { if (/COMBO_CONTEXT_PRESSURE_READY/.test(current)) return { role: "assistant", content: "COMBO_CONTEXT_PRESSURE_READY" }; if (/CONTEXT_PRESSURE_READY/.test(current)) return { role: "assistant", content: "CONTEXT_PRESSURE_READY" }; - if (/qa_mcp_echo:mcp-ok-local-agent/.test(text)) return { role: "assistant", content: "qa_mcp_echo:mcp-ok-local-agent" }; - if (/qa_mcp_echo|mcp-ok-local-agent/i.test(current || text) && mcpTool && !/qa_mcp_echo:mcp-ok-local-agent/.test(text)) { - return toolCall("call_qa_mcp_echo", mcpTool, { text: "mcp-ok-local-agent" }); + if (text.includes(expectedMcpEchoResult)) return { role: "assistant", content: expectedMcpEchoResult }; + if (/qa_mcp_echo|mcp-ok-local-agent/i.test(current || text) && mcpTool && !text.includes(expectedMcpEchoResult)) { + return toolCall("call_qa_mcp_echo", mcpTool, { text: requestedMcpEchoText }); } if (/LOOP_LIMIT|loop-limit-repeat-local-agent|iteration limit/i.test(current || text) && pluginTool) { diff --git a/skills/scripts/e2e/lib/langbot-e2e.mjs b/skills/scripts/e2e/lib/langbot-e2e.mjs index e733a58d6..a566ca529 100755 --- a/skills/scripts/e2e/lib/langbot-e2e.mjs +++ b/skills/scripts/e2e/lib/langbot-e2e.mjs @@ -50,6 +50,38 @@ export async function ensureEvidence(paths) { await appendFile(paths.networkLog, "", "utf8"); } +export async function beginBackendLogCapture(evidenceDir, sourcePath = env.LANGBOT_BACKEND_LOG || "") { + if (!sourcePath) return null; + const source = resolve(sourcePath); + try { + const info = await stat(source); + return { + source, + start_offset: info.size, + target: resolve(evidenceDir, "backend.log"), + }; + } catch { + return null; + } +} + +export async function finishBackendLogCapture(capture) { + if (!capture) return null; + try { + const content = await readFile(capture.source); + const start = content.length >= capture.start_offset ? capture.start_offset : 0; + const window = content.subarray(start); + if (window.length === 0) return null; + await writeFile(capture.target, window); + return { + path: capture.target, + bytes: window.length, + }; + } catch { + return null; + } +} + export async function pathExists(path) { try { await stat(path); diff --git a/skills/scripts/e2e/local-agent-steering-debug-chat.mjs b/skills/scripts/e2e/local-agent-steering-debug-chat.mjs index 5a73f17bd..fb8225d51 100755 --- a/skills/scripts/e2e/local-agent-steering-debug-chat.mjs +++ b/skills/scripts/e2e/local-agent-steering-debug-chat.mjs @@ -10,11 +10,13 @@ import { waitForDebugChatTextStable, } from "./lib/debug-chat.mjs"; import { + beginBackendLogCapture, createBrowser, ensureAuthenticatedBrowser, ensureEvidence, evidencePaths, exitCode, + finishBackendLogCapture, localIsoWithOffset, loadEnvFiles, pathExists, @@ -27,6 +29,7 @@ await loadEnvFiles(); const caseId = env.LBS_CASE_ID || "local-agent-steering-debug-chat"; const paths = evidencePaths(caseId); await ensureEvidence(paths); +const backendLogCapture = await beginBackendLogCapture(paths.evidenceDir); const backendUrl = (env.LANGBOT_BACKEND_URL || "").replace(/\/$/, ""); const pipelineUrl = env.LANGBOT_E2E_PIPELINE_URL || env.LANGBOT_LOCAL_AGENT_PIPELINE_URL || env.LANGBOT_PIPELINE_URL || ""; @@ -190,6 +193,12 @@ try { } finally { if (browser?.page) await safeScreenshot(browser.page, paths.screenshot); if (browser) await browser.close().catch(() => {}); + const backendLog = await finishBackendLogCapture(backendLogCapture); + if (backendLog) { + result.evidence.backend_log = backendLog.path; + result.backend_log = backendLog; + if (!result.evidence_collected.includes("backend_log")) result.evidence_collected.push("backend_log"); + } const finishedAt = new Date(); result.finished_at = finishedAt.toISOString(); result.finished_at_local = localIsoWithOffset(finishedAt); diff --git a/skills/scripts/e2e/pipeline-debug-chat.mjs b/skills/scripts/e2e/pipeline-debug-chat.mjs index a66590b70..1128b2210 100755 --- a/skills/scripts/e2e/pipeline-debug-chat.mjs +++ b/skills/scripts/e2e/pipeline-debug-chat.mjs @@ -10,11 +10,13 @@ import { setDebugChatStreamOutput, } from "./lib/debug-chat.mjs"; import { + beginBackendLogCapture, createBrowser, ensureAuthenticatedBrowser, ensureEvidence, evidencePaths, exitCode, + finishBackendLogCapture, localIsoWithOffset, pathExists, safeScreenshot, @@ -25,6 +27,7 @@ import { const caseId = env.LBS_CASE_ID || "pipeline-debug-chat"; const paths = evidencePaths(caseId); await ensureEvidence(paths); +const backendLogCapture = await beginBackendLogCapture(paths.evidenceDir); const expectedText = env.LANGBOT_E2E_EXPECTED_TEXT || "OK"; const prompt = env.LANGBOT_E2E_PROMPT || `请只回复 ${expectedText}，用于前端调试测试。`; @@ -1062,6 +1065,12 @@ try { result.pipeline_config_restore = restoreDiagnostic; } if (browser) await browser.close().catch(() => {}); + const backendLog = await finishBackendLogCapture(backendLogCapture); + if (backendLog) { + result.evidence.backend_log = backendLog.path; + result.backend_log = backendLog; + if (!result.evidence_collected.includes("backend_log")) result.evidence_collected.push("backend_log"); + } const finishedAt = new Date(); result.finished_at = finishedAt.toISOString(); result.finished_at_local = localIsoWithOffset(finishedAt); diff --git a/skills/skills.index.json b/skills/skills.index.json index 48298764c..8f4b89a27 100644 --- a/skills/skills.index.json +++ b/skills/skills.index.json @@ -155,6 +155,7 @@ "agent-runner-release-preflight", "agent-runner-runtime-chaos", "bot-event-routing-product-flow", + "box-mcp-heartbeat-recovery", "dify-agent-debug-chat", "langbot-fake-provider-debug-chat-cross-pipeline-isolation", "langbot-fake-provider-debug-chat-fault-recovery", @@ -567,6 +568,37 @@ "api_diagnostic" ] }, + { + "id": "box-mcp-heartbeat-recovery", + "title": "Box heartbeat restores an existing stdio MCP tool session", + "mode": "probe", + "area": "reliability", + "type": "chaos", + "priority": "p1", + "risk": "high", + "ci_eligible": false, + "tags": [ + "box", + "mcp", + "stdio", + "reliability", + "chaos", + "fault-injection" + ], + "automation": "skills/langbot-testing/probes/box-mcp-heartbeat-recovery.mjs", + "setup_automation": [], + "setup_provides_env": [], + "evidence_required": [ + "ui", + "screenshot", + "console", + "network", + "metrics", + "api_diagnostic", + "resource_log", + "filesystem" + ] + }, { "id": "dify-agent-debug-chat", "title": "Dify AgentRunner returns a response through Pipeline Debug Chat", @@ -2231,7 +2263,8 @@ "kind": "python", "path": "fixtures/mcp/qa_mcp_echo_server.py", "related_cases": [ - "mcp-stdio-tool-call" + "mcp-stdio-tool-call", + "box-mcp-heartbeat-recovery" ] }, { diff --git a/skills/skills/langbot-testing/cases/box-mcp-heartbeat-recovery.yaml b/skills/skills/langbot-testing/cases/box-mcp-heartbeat-recovery.yaml new file mode 100644 index 000000000..556d24f32 --- /dev/null +++ b/skills/skills/langbot-testing/cases/box-mcp-heartbeat-recovery.yaml @@ -0,0 +1,85 @@ +id: box-mcp-heartbeat-recovery +title: "Box heartbeat restores an existing stdio MCP tool session" +mode: probe +area: reliability +type: chaos +priority: p1 +risk: high +ci_eligible: false +tags: + - box + - mcp + - stdio + - reliability + - chaos + - fault-injection +skills: + - langbot-env-setup + - langbot-testing +env: + - LANGBOT_BACKEND_URL + - LANGBOT_FRONTEND_URL + - LANGBOT_E2E_LOGIN_USER + - LANGBOT_REPO + - LANGBOT_LOCAL_AGENT_PIPELINE_URL + - LANGBOT_MCP_QA_STDIO_SERVER_UUID +automation: skills/langbot-testing/probes/box-mcp-heartbeat-recovery.mjs +automation_env: + - LANGBOT_BACKEND_URL + - LANGBOT_FRONTEND_URL + - LANGBOT_E2E_LOGIN_USER + - LANGBOT_REPO + - LANGBOT_BROWSER_PROFILE + - LANGBOT_CHROMIUM_EXECUTABLE + - LANGBOT_LOCAL_AGENT_PIPELINE_URL + - LANGBOT_MCP_QA_STDIO_SERVER_UUID +metrics_thresholds_json: '{"recovery_duration_ms":{"max":30000},"mcp_retry_count_increased":{"min":1},"debug_chat_status":{"equals":"pass"},"ledger_tool_call_status":{"equals":"pass"}}' +fault_model_json: '{"target":"local Box runtime child owned by the active LangBot checkout","fault":"SIGTERM","blast_radius":"one local development instance","expected_recovery":"heartbeat replaces Box, existing MCP session reconnects, and a fresh browser Debug Chat tool call passes without MCP registration"}' +load_profile_json: '{"requests":1,"concurrency":1,"classification":"recovery-not-throughput-benchmark"}' +preconditions: + - "Run only against a disposable local LangBot development instance; never use a shared or production-like backend." + - "The selected Local Agent pipeline uses a model route that supports qa_mcp_echo tool calls." + - "qa-local-stdio is already connected and LANGBOT_MCP_QA_STDIO_SERVER_UUID points to it." +steps: + - "Verify Box is available, the saved MCP server is connected, and qa_mcp_echo is globally visible." + - "Select exactly one Box child whose parent is main.py running from LANGBOT_REPO; abort on zero or multiple matches." + - "Send SIGTERM to that Box child and poll process, Box status, MCP runtime info, and global tools for up to 30 seconds." + - "Without running MCP setup or registration, reset Debug Chat and call qa_mcp_echo with a unique per-run value through the browser." + - "Audit the matching AgentRunner ledger run and require the exact qa_mcp_echo arguments plus the complete tool result text." +checks: + - "The old Box PID exits and a new Box PID appears under the same LangBot parent." + - "Box returns available=true with at least one active session and managed process." + - "The MCP server returns connected, tool_count>=1, and a retry_count greater than preflight." + - "A new browser assistant message contains the unique recovery value." + - "The run ledger proves qa_mcp_echo was invoked with the exact unique value and completed with the full qa_mcp_echo result." +cleanup: + - "The probe waits for heartbeat recovery and does not spawn Box or re-register MCP. If recovery times out, restart only the disposable local LangBot instance before further tests." +evidence_required: + - ui + - screenshot + - console + - network + - metrics + - api_diagnostic + - resource_log + - filesystem +diagnostics: + - "Set LANGBOT_BOX_RUNTIME_PID only after manually verifying the exact Box process when more than one local instance is running." + - "resource-log.json records each recovery poll; api-diagnostic.json retains preflight, recovered MCP state, browser, and ledger findings even when a later assertion fails." + - "ledger-audit.json distinguishes a real tool invocation from a model merely repeating the prompt token." + - "This case intentionally has no MCP registration setup step because registration during recovery would mask the product behavior under test." +success_patterns: + - "Connected to Box runtime via stdio" + - "Box runtime reconnected, sandbox features restored" +failure_patterns: + - "Box/MCP did not recover" + - "Expected exactly one local Box child" + - "Recovered Debug Chat tool call failed" + - "Recovered Debug Chat ledger audit failed" +expected_failures: + - "Disconnected from Box runtime, trying to reconnect" + - "Box MCP transport task was cancelled unexpectedly" +troubleshooting: + - backend-not-listening + - debug-chat-history-contaminates-automation + - local-agent-model-route-unavailable diff --git a/skills/skills/langbot-testing/cases/local-agent-effective-prompt-debug-chat.yaml b/skills/skills/langbot-testing/cases/local-agent-effective-prompt-debug-chat.yaml index a5cadcc05..3c28167a9 100644 --- a/skills/skills/langbot-testing/cases/local-agent-effective-prompt-debug-chat.yaml +++ b/skills/skills/langbot-testing/cases/local-agent-effective-prompt-debug-chat.yaml @@ -28,6 +28,11 @@ automation_env: - LANGBOT_LOCAL_AGENT_PIPELINE_NAME automation_pipeline_url_env: LANGBOT_LOCAL_AGENT_PIPELINE_URL automation_pipeline_name_env: LANGBOT_LOCAL_AGENT_PIPELINE_NAME +automation_expected_runner_id: "plugin:langbot-team/LocalAgent/default" +automation_extensions_patch_json: '{"enable_all_plugins":false,"bound_plugins":[{"author":"langbot-team","name":"LocalAgent"},{"author":"qa","name":"plugin-smoke"}],"enable_all_mcp_servers":false,"bound_mcp_servers":[],"enable_all_skills":false,"bound_skills":[]}' +automation_restore_extensions: "1" +automation_reset_debug_chat: "1" +automation_debug_chat_session_type: "person" automation_prompt: "qa-effective-prompt" automation_expected_text: "PROMPT_PREPROCESS_OK" automation_response_timeout_ms: "180000" diff --git a/skills/skills/langbot-testing/fixtures/fixtures.json b/skills/skills/langbot-testing/fixtures/fixtures.json index 24d4437fd..1ffe1fc8c 100644 --- a/skills/skills/langbot-testing/fixtures/fixtures.json +++ b/skills/skills/langbot-testing/fixtures/fixtures.json @@ -41,7 +41,7 @@ "title": "MCP stdio qa_mcp_echo server", "kind": "python", "path": "fixtures/mcp/qa_mcp_echo_server.py", - "related_cases": ["mcp-stdio-tool-call"], + "related_cases": ["mcp-stdio-tool-call", "box-mcp-heartbeat-recovery"], "checks": ["exists", "direct_mcp_script", "langbot_registration_script"] }, { diff --git a/skills/skills/langbot-testing/fixtures/plugins/qa-plugin-smoke/dist/qa-plugin-smoke-0.1.0.lbpkg b/skills/skills/langbot-testing/fixtures/plugins/qa-plugin-smoke/dist/qa-plugin-smoke-0.1.0.lbpkg index a4a50f803e0af7218ec8800d3cadff9e64d9b60b..6e7cc28bb8124ed937907a91a1ef5ec0e574baf1 100644 GIT binary patch literal 14159 zcmeHOYm6k-bsj^E<(5!l1*FIiJN)jtoL`Wc3|uU_T)J!1mmR*shmO+itrd)}3M3c1wL3T5T(|8YLmDVc7HhhIrV* zjcfH~Lkt+I8uY!>5)XYj@B%yZ{82;n!f+5YYBjlQ^#_hLZFk#p>~_tu-0qea)*iO1 zxzXDZ5esJBTmZVZ3|hWD2#GS>3L)L*XEqz+Uaw_2qGRudLtkndM91^Ry}*6U3!_JJ zQ3@tagjj1}GLA=j%}-fw04y5#;4A0feC?&nfBeG!7ha^lkDb5#o8LP4{ui&j@YLlC ze{k^4KiU80i~G<0VE?yXR*?UA?l*wQv9_d>q3Y8|HPghVW`_{*WO%7Yr_lqA-6F1A zK_J7RMmJ`#-7N(J*&^Jy@7Udj`0Rsx(;s-Qbi+VRyS9zr_u4@yUFoYwLUequf2Zwh z5Fx5!V1>N~FhzrFlEwu!U{s&JwR@)=5Kx_;o$x%pu&Y8B0(=U#b*n4i25|#`7h+ok zZTymuRfP=|jkGQ!j95un2!f{9D~)B57QtH=zt6A7xTFZ-DQYkXxoh^qzEhIBvNenp zaxj8HXw_5*`j+jQgAt>&!Vt!y>_}><^;(DP)dKZ*R45Qs>Q_04J4>#MggOw1Fsh&ook6D3KdRQuz zS`NHTS}n`T>v?6VRxol~U{^;lKDg~iB!uU5=0$Z?+y^HVS+s!AoQ~;Bt37J=E&sIi z#fmWUY~c4sjsvw1e&z4pfa2423pg3*GwdxeKRc9tC|NXBqE@sFfYN)jE5ouu z4~%({rJr6`MB8qK^Flg-6aXc+TUIaBpbBg3JX1rf0}4Wx%ta3e7RZDZ4W!#HgHGi< z)F90c!!RH6Y`4sy^AX>*R21L!O)=kN`2{%>it#gv`YtQbXk8jdMHpJdhF;wmJ*^(x zT8647KblIL0)dpLQWzqZ81!i-gAAea-$VpAd?2UF`MkTKS$i6c>rk(C}%NL3nO zGErktil#_QPrBOpnZraW5yLKwf53&6VJoY(uZZwwXAcNLh zHNRb_O^}s73i->KyKC6Gl_>+!ke!Zfh4!|rstF8MbZ}QK zX_gmB?N)@8K1$GuBGR)WQDLvOh7QLhSXi`S#9Bea=yP0c&gHQe^wRenxncw(WNv+< zQV62G`${naLSi6CP>ZLpE#%^~ul1zR9+XW^ z-?;zy`ibWH#u|Qoa&2?7dH+3TF~0fnRV4~Slg;O1!6&R6zxua`0qMGRb}TrtYLPLC zOz}e3DzV`m zK%J6gYcianl0w9DOsU2oSAXmGpS}FrqbS!<5hkjVsMP!FlXS(dOqR|o=2}5k0FQm( zPmR9;^XdAm_ocX8-t{B6x zQ(ZDdE&Y^jws)wdnggYQP% z?pMUnt_CQ-RuSLZ9VDK-3s#4AMO2Y}IkKvrRKzjIc26I-TAK`U7f>Mx-IQHViqG7Q zNH4*XOt@vWPIrB8NCj-f{N-v@8(MX%TM@T*_%oltw>*am+F4Kv!2n#Z*OwOWh=C)G zpgK*48M}(r4RQELYVi?x%WKONc*3Z`t~iqB)t2WY$*=lW+a4lczCH6UwJg88Wt9&f zsfha0$17s-_N9tIk?Hmvoy)aYg3Gl?yOv3iktSgok$jhX^^79mR!eqyfI=%mmB@5g z^MxHU+~wLJ2M^a$^<{uK$ylXy3#t(VE!T*HnhI6k7lGeeF=`l9+pgEtJ`vQKLHl%# z@K^D}d?+whu_$f_WCll5YeAcyyFtaMvI;{xLl<$Phv^iRO+DV&4|9t+k8{3=w1d@y;hXGHTbM}xVk=u1Gk_RyqW4}t*oI9ux>Vh&c_7d?<%{3N z!-I3rAN=m)=)Gt2Jlb;+Me6Q01=S7VZ9Rlq5U&mn1+^^dx*9@1!)ECE=ybc`y?A0B zXr|fFaN^gLQyh0GkW_6VF`DPTcjeM|_Fw)E4ut^AB-7cL)_JD(mqQ0xs)%tB0fw~A ztERLM6~tkfQN(i6(b#g|eC5^5I7P)<7ruVwg%=L?p51@$OX}>(!EZmk|M;u1IVpg5 zXij9xkq$<-G*i*FmbZ)5TyLyVrE-$K)66;jKI-gPBRD3AN$l8RZ>+`~-VP9erh`xY6glHU5^rW2lVsD$*pA~_VL62_8Kt|9LLGTu}KPQfInaq5T7g;ZEfrZy#>PKQM*SLRgP_Dt5= z2yL~OD!qY(xLXDb)OW+5roh+BM@QLxb{GVhR+B@NFV@7#Q;el z^7JzYzxQWTLJ%5@LrB@sk}X_snJNvbQZL2s;8+}QjZ|XJh3s$s=yf^`%V%ZZeC2oc zzxRv^Jocab5~6kx5l0&TC4u`jMH&dHSsKi&@tC{_6D$@0l37(3PUV|RexJe2~?|ogH)Mkm>o|+Ty~(LJbgxppMfqZ+FF55DVhS?fBo&M;YusUR%Cx zq1}R0@hf~bf&XQI7Bd}eSDh}THX)*QYE&77peOBa4~;HH2Hf>U0|)yZ;Lr^igL_}g z)Z-OsMI5g$i9>hvKJFYAcZkDP{Cc3i^r`yNBk{09OIz*Ek{;Hw{g#7HG%zn7L91+p zZ}ijr72{AmKs9>C2OI+6i~*>6Z+BNw{CUrHH(fvZUt56Z{5gj}BlmAF(nE_5g{y&c zbLaLhAZ>_`h^L1Yol=g5UD|7)Pnn!qPLC@(g`A8#k8%1;a(+<$;BGSL!VOn(j5m)i zfA%9m_wApXiH;mz9%25}M1t_;4`dKhcPXfGomc(kI$R!I{tQ5ZZt3S|pc{ t#VcZtl955=h^#t&a+?j2ROJCAQ0#($hmmR;5?VCIuQ#5+Q9~a zAi%E$y~;?vXZtzDBz={5HXt7%@i zNhDen^GS0(*JMUNCz%eL;C^yVh?uHvb98K?#dV<=DQM}Lgx~(`W1Ep)oOi0WcNv`U ziZtEv+&-D!)`=lAr@jsY1M&qb4?2lq8uytI_6f(8CYsmkBQQDE zoz>i8AE?O8r#kZ?#q(25ArW`<@xn=Kec)GO;yR~S&&M(yi?vk?>eSPvkBM_Lo6|&k zYwZs2WV6m%v@u}^b9M{FYFgdyuhL|sAn|M8Qx5vf(2p(RyFnN2b)N`H5ET!Pp^CLy z-J(z6GP}W(w9IZ4lq^>Rx)*|(`jsbk?7BJ$Z=>Urvdw)2L?Vc z987H-q0U~gsC&dn9x&`E18HXaav7w^;|vPv#JW8uk~rSUrX-RuHpVY=I(Q)&Fk3*= z_M|>ZAhx#6H0%v1yfYY)Z&6@iDfaP4co)<^Qx6Qamv`YbU>uuUJcRIx6DzO@cn(~j z`3Z&>ZqXlE;WhI(@wOS^E6{tNt0(t0-}*thrBMvm85az5p0jV`*9s=eSSX&w7nvkF z?o~r$Y=5JzE(Zl{TLz%vDlpraJ2^sKJ**Hh?W!F-;1}CIg#8)ycOd#Ww#-#yP`U=^ z?N4P@is|f-{?Fg~Kq{gPjE~>lJvs638}vng5&H?BSl3Lub-iD>;++9vsyf_t=p!C7 z%zgn;El*4*cvL`Yw)2PyVmU8wo7^g{SFyW7zI^z&@Q}9dQ&6#RI(P;qY6_IN*pAu9FMPV{HEsHp5L`7V_~V=C`Ez%8 z(+bFQbzkVhgXttnTZa#=A)`aK0C|rJAsDqQ^y1b|61b9I6ck}TvKZReug@HHVx%i8 zdaIFjx~(6F-@v)LvCOBFi1bP7BJo{kq|UfYfpj3)w_g6Zi)it#Jor;EVKdsu+H>rN znbLTd--`cox%?M~6LQHwPEqynn~bCjjh0E4G&pZKCB5xlQTSI0RB5>!1hI0}UvGaY z6v6>bc$)VjaF#~g9kj351}0jnHKl(l+&H;x^jeWsao8A(W!&azy#G z3TGQ=Nj)bU5u{#k6IFaV?W&>23h_G9fqNZAxGk(5vu&uen_HDOuImhT|a4Od{C#>KDJMKkvh#&Vz}4yWH%l5n!4cTGMej1Jfvd~^@U3Dcg$txh6>z+wu%4;9*~TpLay z@r3k0onk*&8@f9$vKafr7MAK}i};u~dckir6qv(QAM%-eEj<`=Cmm z!gApun?b%>-6PuLHcUiH-~4^OaTiRlntuD&Nk~|WBM-N%sOjC^TAzjwvwjsGB=tX! zz8&sXHC-4K{#qAdPwOg`{4;xAnL=WdfUKDS1{r4d{Jw&9qV5r5-y?;cWSFTkOchdd za^j8(4r~$MX-I|iHEeoga}Y&;{(|dddQD>r9kA-R8UWssyVn>b@XF5JW6oykRp&cb zrBF6@wcwkCws>vpO0uEPNzC5yCq#=2qNoX_?4$4@C$y}ZI3QiSy54VUMn`KWa4k0V zQc4w+(d&o3(>xJpO`95DI5MrCc9%Eu6Re~!MDD{xD_+bv=9sd{1D}0sYnu)Q45kd6bB zHp^==x^Fm16*)~iksvAFAV%=x5O~R)xViX%7*-(f%shdq502ms;?Jy+btyFhdJ7A> zli9?oBi%3T_LNJX`xBPu6GVQF(_gwr8g=zB5i4`W586&BAoWs2^H4MZwpx9JHsihU zOz`x|}P)YMz#V*`Lpxelp?<=pc%4|cfOmj%dv%q`f zfr85Uj8Pv*asjlo`g{B`;>&)Svv!{pjywM8rD+bpd70*KQJ zVmi}|`Yv?vr#iDiR>QaD`4O}ha@`f(8>onLj7!O(<&_l?5a3y0{7x%CkZPbb0unx@>&>pSmonxzJ^*y@MyaC#K&+03Q7?yU_VQ9jxI3QmITvQIgBJsgk#M zeDl|7*)a0+;P|&RX(>g%8#I~;_cr88s7fn{YY@Hv{xkLd#IIHE1MWlg)Oo73gyROu z0dIN(>ukMAtWZPnDpj1VPt24r&zwcsT-l0s11GNmN}WQ-T|HmP_yrFI^}IdFaM2`JY=b#X<`p#CZvq+jd`g3ak4}P2J4P?G&V!(nmMiEpW>4v4|Xi@M(@paZxdG z>_lCxztX)7A(`EwH4v1QgV!AQKIBhUQg!Q0;sI9Ffkx}vAFR0ATUt8*uC!VzPIC+t zE&Cig3l#<`w1YMiEq^ z4(=1zBAHrY>3NDnsa`YuSvF^Ht|Z1_%fqkND&Zb~K1Wk^Ji2*L9xcUeIvdB)O=I^e zoXlV1JcTkFoHzX4GwWj-R;2LF7+vdxd%oR!q1H{d0sHL5j1wbh=Cg1%ZQbX%7G z#53m3vAq@zV+C+L$^kM{hzI@h1EN^Fpd|f1R_3eyjV~KBhCfm~tMXuYYc54aJoC2H zlKKu-8OoMmV(SeuJdm@GbPwg||Oyd-_zN9kV1g$X{TW02CeoiS^rlUzVi5 zDEysS4w~X>boWTVAL$N|^k(KLRvR`YNaNe+wjMaIeXfnPqJzItr%k(h8Xg{505ML%I!hg-IirV30z5%7LM*JQf$JxYOHyBV7GetA;#TDJ{*B=c$Xg zP|cY~Ws5{Wi-@C|Q_>XSE?pt>}dFg6h<*{ z8yFxm{gG*|ZC)EA)yP)R6_1W*>)+NT@pTJ2UQ>5c?Kv(=mBYT~tzea~Bc5k2+lfIubn34Yk_U>o=3YHaWp^%`7=HQI%Smo`NFf*i4oJ+BQX ziu`)+T(QOcoI0%IV`E;jd}gvdYU<{uwpH8Q!&fPEQnYf5t&S`o8%xI|6>&Fy8{{;S2sTp05Ly@Znj ztsVyaf31&M>n|Tr2)v-Xe8At8A9Dkkh2|1(1Nd`H{9il3yYl;AVBk9lhzIt&Wdi*J D9m=s2 diff --git a/skills/skills/langbot-testing/probes/box-mcp-heartbeat-recovery.mjs b/skills/skills/langbot-testing/probes/box-mcp-heartbeat-recovery.mjs new file mode 100644 index 000000000..da3755994 --- /dev/null +++ b/skills/skills/langbot-testing/probes/box-mcp-heartbeat-recovery.mjs @@ -0,0 +1,396 @@ +#!/usr/bin/env node + +import { execFile } from "node:child_process"; +import { readFile, readlink, writeFile } from "node:fs/promises"; +import { resolve } from "node:path"; +import { env, exit } from "node:process"; +import { promisify } from "node:util"; +import { + apiJson, + ensureEvidence, + evidencePaths, + exitCode, + loadEnvFiles, + localIsoWithOffset, + resetAndAuthLocalUser, + writeResult, +} from "../../../scripts/e2e/lib/langbot-e2e.mjs"; + +const execFileAsync = promisify(execFile); +const DEFAULT_LOCAL_PASSWORD = "LangBotE2ELocalPass!2026"; +const BOX_COMMAND_RE = /(?:^|\s)-m\s+langbot_plugin\.cli\.__init__\s+box(?:\s|$)/; + +await loadEnvFiles(); +const caseId = env.LBS_CASE_ID || "box-mcp-heartbeat-recovery"; +const paths = evidencePaths(caseId); +await ensureEvidence(paths); + +const startedAt = new Date(); +const backendUrl = env.LANGBOT_BACKEND_URL || ""; +const user = env.LANGBOT_E2E_LOGIN_USER || ""; +const password = env.LANGBOT_E2E_LOGIN_PASSWORD || DEFAULT_LOCAL_PASSWORD; +const serverUuid = env.LANGBOT_MCP_QA_STDIO_SERVER_UUID || ""; +const expectedTool = env.LANGBOT_E2E_EXPECTED_TOOL || "qa_mcp_echo"; +const recoveryTimeoutMs = positiveInteger(env.LANGBOT_BOX_RECOVERY_TIMEOUT_MS, 30_000); +const pollIntervalMs = positiveInteger(env.LANGBOT_BOX_RECOVERY_POLL_INTERVAL_MS, 250); +const faultModelPath = resolve(paths.evidenceDir, "fault-model.json"); +const apiDiagnosticPath = resolve(paths.evidenceDir, "api-diagnostic.json"); +const resourceLogPath = resolve(paths.evidenceDir, "resource-log.json"); +const metricsPath = resolve(paths.evidenceDir, "metrics.json"); +const ledgerAuditPath = resolve(paths.evidenceDir, "ledger-audit.json"); +const debugChatEvidenceDir = resolve(paths.evidenceDir, "debug-chat"); +const probeValue = `box-recovery-${Date.now().toString(36)}`; + +const result = { + source: "automation", + case_id: caseId, + run_id: paths.runId, + status: "fail", + reason: "", + started_at: startedAt.toISOString(), + started_at_local: localIsoWithOffset(startedAt), + finished_at: "", + finished_at_local: "", + duration_ms: 0, + evidence: { + fault_model_json: faultModelPath, + api_diagnostic_json: apiDiagnosticPath, + resource_log_json: resourceLogPath, + metrics_json: metricsPath, + ledger_audit_json: ledgerAuditPath, + debug_chat_evidence_dir: debugChatEvidenceDir, + automation_result_json: paths.automationResultJson, + result_json: paths.resultJson, + }, + evidence_collected: [ + "metrics", + "api_diagnostic", + "resource_log", + "filesystem", + ], +}; + +let resourceSamples = []; +let apiDiagnostic = { + expected_tool_call: { + tool_name: expectedTool, + parameters: { text: probeValue }, + result_text: `${expectedTool}:${probeValue}`, + }, +}; + +try { + requireEnv("LANGBOT_BACKEND_URL", backendUrl); + requireEnv("LANGBOT_E2E_LOGIN_USER", user); + requireEnv("LANGBOT_MCP_QA_STDIO_SERVER_UUID", serverUuid); + requireEnv("LANGBOT_LOCAL_AGENT_PIPELINE_URL", env.LANGBOT_LOCAL_AGENT_PIPELINE_URL || ""); + requireEnv("LANGBOT_FRONTEND_URL", env.LANGBOT_FRONTEND_URL || ""); + + const auth = await resetAndAuthLocalUser({ backendUrl, user, password }); + const preflight = await runtimeSnapshot(auth.token); + apiDiagnostic.preflight = preflight; + assertReadySnapshot(preflight, "preflight"); + + const oldBox = await selectLocalBoxProcess(); + const faultModel = { + target: "local Box runtime child process owned by the active LangBot checkout", + injected_fault: "SIGTERM to the selected Box child process", + destructive: true, + blast_radius: "single local LangBot development instance", + old_box_pid: oldBox.pid, + parent_pid: oldBox.ppid, + recovery_timeout_ms: recoveryTimeoutMs, + abort_conditions: [ + "no exact Box command match", + "multiple eligible Box processes", + "Box parent is not main.py from LANGBOT_REPO", + ], + cleanup: "wait for the LangBot heartbeat to replace Box and restore the MCP session; do not spawn or register anything", + }; + await writeFile(faultModelPath, `${JSON.stringify(faultModel, null, 2)}\n`, "utf8"); + + process.kill(oldBox.pid, "SIGTERM"); + const faultStartedAt = performance.now(); + let observedOldProcessExit = false; + let observedBoxUnavailable = false; + let recovered = null; + const deadline = Date.now() + recoveryTimeoutMs; + + while (Date.now() < deadline) { + const processes = await boxProcesses(); + const oldAlive = processes.some((item) => item.pid === oldBox.pid); + observedOldProcessExit ||= !oldAlive; + const replacement = processes.find((item) => item.pid !== oldBox.pid && item.ppid === oldBox.ppid) || null; + const snapshot = await runtimeSnapshot(auth.token).catch((error) => ({ error: error.message })); + observedBoxUnavailable ||= snapshot.box?.available === false; + const sample = { + elapsed_ms: Math.round(performance.now() - faultStartedAt), + old_process_alive: oldAlive, + replacement_pid: replacement?.pid || null, + box_available: snapshot.box?.available ?? null, + box_active_sessions: snapshot.box?.active_sessions ?? null, + box_managed_processes: snapshot.box?.managed_processes ?? null, + mcp_status: snapshot.mcp?.status || null, + mcp_tool_count: snapshot.mcp?.tool_count ?? null, + mcp_retry_count: snapshot.mcp?.retry_count ?? null, + tools_has_expected: snapshot.tools?.includes(expectedTool) ?? false, + error: snapshot.error || "", + }; + resourceSamples.push(sample); + + if (observedOldProcessExit + && replacement + && snapshotReady(snapshot) + && (snapshot.mcp.retry_count ?? 0) > (preflight.mcp.retry_count ?? 0)) { + recovered = { replacement, snapshot, elapsedMs: sample.elapsed_ms }; + break; + } + await delay(pollIntervalMs); + } + + if (!recovered) { + throw new Error(`Box/MCP did not recover within ${recoveryTimeoutMs} ms.`); + } + apiDiagnostic.recovered = recovered.snapshot; + + const chat = await runRecoveredDebugChat(probeValue); + apiDiagnostic.debug_chat = { + status: chat.status, + reason: chat.reason, + expected_assistant_text: probeValue, + evidence_dir: debugChatEvidenceDir, + }; + result.evidence_collected.push("ui", "screenshot", "console", "network"); + const ledger = await auditRecoveredToolCall(probeValue, startedAt); + apiDiagnostic.ledger = { + status: ledger.status, + reason: ledger.reason, + run: ledger.run || null, + expected_tool_call: ledger.expected_tool_call || null, + evidence_path: ledgerAuditPath, + }; + if (chat.status !== "pass") { + throw new Error(`Recovered Debug Chat tool call failed: ${chat.reason || chat.status}`); + } + if (ledger.status !== "pass") { + throw new Error(`Recovered Debug Chat ledger audit failed: ${ledger.reason || ledger.status}`); + } + + result.status = "pass"; + result.reason = `Box was replaced in ${recovered.elapsedMs} ms, MCP reconnected without registration, and the browser plus ledger proved ${expectedTool} returned the unique recovery value.`; + result.metrics_summary = { + recovery_duration_ms: recovered.elapsedMs, + old_box_pid: oldBox.pid, + new_box_pid: recovered.replacement.pid, + observed_old_process_exit: observedOldProcessExit, + observed_box_unavailable: observedBoxUnavailable, + mcp_retry_count_before: preflight.mcp.retry_count ?? 0, + mcp_retry_count_after: recovered.snapshot.mcp.retry_count ?? 0, + debug_chat_response_ms: chat.metrics_summary?.response_p95_ms ?? null, + ledger_run_id: ledger.run?.run_id || null, + }; + result.thresholds_summary = { + recovery_duration_ms: { + actual: recovered.elapsedMs, + max: recoveryTimeoutMs, + pass: recovered.elapsedMs <= recoveryTimeoutMs, + }, + mcp_retry_count_increased: { + actual: recovered.snapshot.mcp.retry_count ?? 0, + min: (preflight.mcp.retry_count ?? 0) + 1, + pass: (recovered.snapshot.mcp.retry_count ?? 0) > (preflight.mcp.retry_count ?? 0), + }, + debug_chat_status: { actual: chat.status, expected: "pass", pass: chat.status === "pass" }, + ledger_tool_call_status: { actual: ledger.status, expected: "pass", pass: ledger.status === "pass" }, + }; +} catch (error) { + if (result.status === "fail" && /is required/.test(error.message)) result.status = "env_issue"; + result.reason = error.message; +} finally { + const finishedAt = new Date(); + result.finished_at = finishedAt.toISOString(); + result.finished_at_local = localIsoWithOffset(finishedAt); + result.duration_ms = finishedAt.getTime() - startedAt.getTime(); + await writeFile(resourceLogPath, `${JSON.stringify(resourceSamples, null, 2)}\n`, "utf8"); + await writeFile(apiDiagnosticPath, `${JSON.stringify(apiDiagnostic, null, 2)}\n`, "utf8"); + await writeFile(metricsPath, `${JSON.stringify({ + probe: caseId, + status: result.status, + duration_ms: result.duration_ms, + metrics_summary: result.metrics_summary || {}, + thresholds_summary: result.thresholds_summary || {}, + }, null, 2)}\n`, "utf8"); + await writeResult(paths, result); + console.log(JSON.stringify(result, null, 2)); + exit(exitCode(result.status)); +} + +function requireEnv(name, value) { + if (!value) throw new Error(`${name} is required.`); +} + +function positiveInteger(value, fallback) { + const parsed = Number.parseInt(String(value || ""), 10); + return Number.isFinite(parsed) && parsed > 0 ? parsed : fallback; +} + +function delay(milliseconds) { + return new Promise((resolveDelay) => setTimeout(resolveDelay, milliseconds)); +} + +function isApiFailure(response) { + return response.status >= 400 || response.json.code !== 0; +} + +async function runtimeSnapshot(token) { + const [boxResponse, serverListResponse, toolsResponse] = await Promise.all([ + apiJson(backendUrl, "/api/v1/box/status", { token }), + apiJson(backendUrl, "/api/v1/mcp/servers", { token }), + apiJson(backendUrl, "/api/v1/tools", { token }), + ]); + for (const response of [boxResponse, serverListResponse, toolsResponse]) { + if (isApiFailure(response)) throw new Error(response.json.msg || `API diagnostic failed with HTTP ${response.status}.`); + } + const servers = serverListResponse.json.data?.servers || []; + const listed = servers.find((server) => server.uuid === serverUuid); + if (!listed) throw new Error(`MCP server UUID ${serverUuid} is not registered.`); + const detailResponse = await apiJson( + backendUrl, + `/api/v1/mcp/servers/${encodeURIComponent(listed.name)}`, + { token }, + ); + if (isApiFailure(detailResponse)) { + throw new Error(detailResponse.json.msg || `MCP detail failed with HTTP ${detailResponse.status}.`); + } + const server = detailResponse.json.data?.server || listed; + const runtime = server.runtime_info || {}; + return { + box: boxResponse.json.data || {}, + mcp: { + uuid: server.uuid || listed.uuid, + name: server.name || listed.name, + status: runtime.status || null, + tool_count: runtime.tool_count ?? 0, + retry_count: runtime.retry_count ?? 0, + error_message: runtime.error_message || "", + }, + tools: (toolsResponse.json.data?.tools || []) + .map((tool) => tool.name || tool.tool_name || tool.function?.name || "") + .filter(Boolean), + }; +} + +function snapshotReady(snapshot) { + return snapshot.box?.available === true + && (snapshot.box?.active_sessions ?? 0) >= 1 + && (snapshot.box?.managed_processes ?? 0) >= 1 + && snapshot.mcp?.status === "connected" + && (snapshot.mcp?.tool_count ?? 0) >= 1 + && snapshot.tools?.includes(expectedTool); +} + +function assertReadySnapshot(snapshot, phase) { + if (!snapshotReady(snapshot)) { + throw new Error(`${phase} Box/MCP state is not ready: ${JSON.stringify(snapshot)}`); + } +} + +async function boxProcesses() { + const { stdout } = await execFileAsync("ps", ["-ww", "-eo", "pid=,ppid=,args="]); + return stdout.split(/\r?\n/).map((line) => { + const match = line.match(/^\s*(\d+)\s+(\d+)\s+(.*)$/); + return match ? { pid: Number(match[1]), ppid: Number(match[2]), args: match[3] } : null; + }).filter((item) => item && BOX_COMMAND_RE.test(item.args)); +} + +async function selectLocalBoxProcess() { + const processes = await boxProcesses(); + const configuredPid = Number.parseInt(env.LANGBOT_BOX_RUNTIME_PID || "", 10); + let candidates = Number.isFinite(configuredPid) + ? processes.filter((item) => item.pid === configuredPid) + : processes; + const repo = resolve(env.LANGBOT_REPO || ".."); + const eligible = []; + for (const candidate of candidates) { + const parentArgs = await readFile(`/proc/${candidate.ppid}/cmdline`, "utf8").catch(() => ""); + const parentCwd = await readlink(`/proc/${candidate.ppid}/cwd`).catch(() => ""); + if (/main\.py(?:\0|$)/.test(parentArgs) && resolve(parentCwd) === repo) eligible.push(candidate); + } + candidates = eligible; + if (candidates.length !== 1) { + throw new Error(`Expected exactly one local Box child for ${repo}, found ${candidates.length}. Set LANGBOT_BOX_RUNTIME_PID only after verifying the process.`); + } + return candidates[0]; +} + +async function runRecoveredDebugChat(value) { + const script = resolve("scripts/e2e/pipeline-debug-chat.mjs"); + const childEnv = { + ...env, + LBS_CASE_ID: `${caseId}-debug-chat`, + LBS_RUN_ID: paths.runId, + LBS_EVIDENCE_DIR: debugChatEvidenceDir, + LANGBOT_E2E_PIPELINE_REQUIRED: "1", + LANGBOT_E2E_PIPELINE_URL: env.LANGBOT_LOCAL_AGENT_PIPELINE_URL, + LANGBOT_E2E_PIPELINE_NAME: env.LANGBOT_LOCAL_AGENT_PIPELINE_NAME || "", + LANGBOT_E2E_RESET_DEBUG_CHAT: "1", + LANGBOT_E2E_RESPONSE_TIMEOUT_MS: env.LANGBOT_E2E_RESPONSE_TIMEOUT_MS || "60000", + LANGBOT_E2E_EXPECTED_TOOL: expectedTool, + LANGBOT_E2E_PROMPT: `Call the ${expectedTool} MCP tool with exactly this text: ${value}. Return only the tool result.`, + LANGBOT_E2E_EXPECTED_TEXT: value, + LANGBOT_E2E_EXTENSIONS_PATCH_JSON: JSON.stringify({ + enable_all_plugins: false, + bound_plugins: [{ author: "langbot-team", name: "LocalAgent" }], + enable_all_mcp_servers: false, + bound_mcp_servers: [serverUuid], + enable_all_skills: false, + bound_skills: [], + }), + LANGBOT_E2E_RESTORE_EXTENSIONS: "1", + }; + for (const key of ["ALL_PROXY", "all_proxy", "HTTP_PROXY", "HTTPS_PROXY", "http_proxy", "https_proxy"]) { + delete childEnv[key]; + } + try { + const { stdout } = await execFileAsync(process.execPath, [script], { + cwd: resolve("."), + env: childEnv, + maxBuffer: 4 * 1024 * 1024, + timeout: positiveInteger(env.LANGBOT_E2E_RESPONSE_TIMEOUT_MS, 60_000) + 30_000, + }); + return JSON.parse(stdout.trim()); + } catch (error) { + const childResultPath = resolve(debugChatEvidenceDir, "automation-result.json"); + const childResult = await readFile(childResultPath, "utf8").then(JSON.parse).catch(() => null); + if (childResult) return childResult; + throw error; + } +} + +async function auditRecoveredToolCall(value, createdAfter) { + const repo = resolve(env.LANGBOT_REPO || ".."); + const python = resolve(repo, ".venv/bin/python"); + const script = resolve(repo, "skills/scripts/e2e/agent-run-ledger-audit.py"); + const args = [ + script, + "--repo", repo, + "--output", ledgerAuditPath, + "--created-after", createdAfter.toISOString(), + "--expected-tool-name", expectedTool, + "--expected-parameters-json", JSON.stringify({ text: value }), + "--expected-result-text", `${expectedTool}:${value}`, + ]; + try { + await execFileAsync(python, args, { + cwd: repo, + env, + maxBuffer: 4 * 1024 * 1024, + timeout: 30_000, + }); + } catch (error) { + const report = await readFile(ledgerAuditPath, "utf8").then(JSON.parse).catch(() => null); + if (report) return report; + throw error; + } + return JSON.parse(await readFile(ledgerAuditPath, "utf8")); +} diff --git a/skills/test/lbs-cli.test.ts b/skills/test/lbs-cli.test.ts index 4d20faf62..a818a709b 100644 --- a/skills/test/lbs-cli.test.ts +++ b/skills/test/lbs-cli.test.ts @@ -58,7 +58,9 @@ import { minExpectedOccurrences, } from "../scripts/e2e/lib/debug-chat.mjs"; import { + beginBackendLogCapture, ensureAuthenticatedBrowser, + finishBackendLogCapture, resolveLangBotRepo, scanBrowserDiagnostics, } from "../scripts/e2e/lib/langbot-e2e.mjs"; @@ -96,6 +98,26 @@ test("e2e helpers resolve embedded LangBot repo from skills cwd", async () => { } }); +test("e2e helpers capture only the current backend log window", async () => { + const tmp = mkdtempSync(join(tmpdir(), "lbs-backend-log-window-")); + try { + const source = join(tmp, "backend-source.log"); + writeFileSync(source, "old startup line\n"); + const capture = await beginBackendLogCapture(tmp, source); + appendFileSync(source, "current request line\ncurrent completion line\n"); + + const completed = await finishBackendLogCapture(capture); + + assert.ok(completed); + assert.equal( + readFileSync(completed.path, "utf8"), + "current request line\ncurrent completion line\n", + ); + } finally { + rmSync(tmp, { recursive: true, force: true }); + } +}); + test("e2e browser diagnostics fail on console server errors", async () => { const tmp = mkdtempSync(join(tmpdir(), "lbs-browser-diagnostics-")); try { @@ -3499,6 +3521,81 @@ test("fake provider returns IMAGE_OK only when image metadata is present", async } }); +test("fake provider requires the effective system prompt before returning its sentinel", async () => { + const provider = await startFakeProviderForTest(); + try { + const withEffectivePrompt = fakeProviderMessage( + await requestFakeProvider(provider, { + messages: [ + { + role: "system", + content: + "When the current user message contains qa-effective-prompt, reply only PROMPT_PREPROCESS_OK.", + }, + { role: "user", content: "qa-effective-prompt" }, + ], + }), + ); + assert.equal(withEffectivePrompt.content, "PROMPT_PREPROCESS_OK"); + + const withoutEffectivePrompt = fakeProviderMessage( + await requestFakeProvider(provider, { + messages: [{ role: "user", content: "qa-effective-prompt" }], + }), + ); + assert.equal(withoutEffectivePrompt.content, "OK"); + } finally { + await provider.stop(); + } +}); + +test("fake provider preserves a unique qa_mcp_echo probe value across the tool loop", async () => { + const provider = await startFakeProviderForTest(); + const tool = { + type: "function", + function: { + name: "qa_mcp_echo", + parameters: { + type: "object", + properties: { text: { type: "string" } }, + }, + }, + }; + try { + const initial = fakeProviderMessage( + await requestFakeProvider(provider, { + tools: [tool], + messages: [{ + role: "user", + content: "Call qa_mcp_echo with exactly this text: box-recovery-unique-42. Return only the tool result.", + }], + }), + ); + assert.equal(initial.tool_calls?.[0]?.function?.name, "qa_mcp_echo"); + assert.deepEqual( + JSON.parse(initial.tool_calls?.[0]?.function?.arguments || "{}"), + { text: "box-recovery-unique-42" }, + ); + + const completed = fakeProviderMessage( + await requestFakeProvider(provider, { + tools: [tool], + messages: [ + { + role: "user", + content: "Call qa_mcp_echo with exactly this text: box-recovery-unique-42. Return only the tool result.", + }, + initial, + { role: "tool", tool_call_id: initial.tool_calls?.[0]?.id, content: "qa_mcp_echo:box-recovery-unique-42" }, + ], + }), + ); + assert.equal(completed.content, "qa_mcp_echo:box-recovery-unique-42"); + } finally { + await provider.stop(); + } +}); + test("fake provider drives steering through qa_plugin_sleep and follow-up context", async () => { const provider = await startFakeProviderForTest(); try { diff --git a/src/langbot/pkg/box/connector.py b/src/langbot/pkg/box/connector.py index adb1e2244..992517d04 100644 --- a/src/langbot/pkg/box/connector.py +++ b/src/langbot/pkg/box/connector.py @@ -328,15 +328,9 @@ class BoxRuntimeConnector(ManagedRuntimeConnector): # closed) or raised after the initial handshake succeeded. # Either way, treat it as a disconnect. if connected.is_set(): - if self._uses_websocket(): - self.ap.logger.error('Disconnected from Box runtime, trying to reconnect...') - if self.runtime_disconnect_callback is not None: - await self.runtime_disconnect_callback(self) - else: - self.ap.logger.error( - 'Disconnected from Box runtime via stdio. ' - 'Cannot automatically reconnect — please restart LangBot.' - ) + self.ap.logger.error('Disconnected from Box runtime, trying to reconnect...') + if self.runtime_disconnect_callback is not None: + await self.runtime_disconnect_callback(self) return new_connection_callback diff --git a/src/langbot/pkg/provider/tools/loaders/mcp.py b/src/langbot/pkg/provider/tools/loaders/mcp.py index fe9b9d515..94a886a80 100644 --- a/src/langbot/pkg/provider/tools/loaders/mcp.py +++ b/src/langbot/pkg/provider/tools/loaders/mcp.py @@ -45,6 +45,7 @@ MCP_RESOURCE_CONTEXT_MAX_BYTES = 96 * 1024 MCP_RESOURCE_TRACE_QUERY_KEY = '_mcp_resource_reads' MCP_RESOURCE_LINKS_QUERY_KEY = '_mcp_resource_links' MCP_RESOURCE_CONTEXT_QUERY_KEY = '_mcp_resource_context' +MCP_TRANSPORT_CLEANUP_TIMEOUT_SECONDS = 5.0 TEXT_LIKE_MIME_TYPES = { 'application/json', @@ -282,6 +283,7 @@ class RuntimeMCPSession: # together, instead of each racing to rebuild the session itself. self._reconnect_event = asyncio.Event() self._reconnected_event: asyncio.Event | None = None + self._connection_generation = 0 # Set transiently when a WS transport drop should NOT stop the managed # process (it will be re-attached on the next initialize()). self._preserve_managed_process = False @@ -423,6 +425,7 @@ class RuntimeMCPSession: async def _lifecycle_loop(self): """Manage the full MCP session lifecycle in a background task.""" + wait_tasks: list[asyncio.Task] = [] try: if self.server_config['mode'] == 'stdio': await self._init_stdio_python_server() @@ -438,6 +441,7 @@ class RuntimeMCPSession: await self.refresh() self.status = MCPSessionStatus.CONNECTED + self._connection_generation += 1 # Notify start() that connection is established self._ready_event.set() @@ -447,8 +451,9 @@ class RuntimeMCPSession: monitor_task = asyncio.create_task(self._box_stdio_runtime.monitor_process_health()) shutdown_task = asyncio.create_task(self._shutdown_event.wait()) reconnect_task = asyncio.create_task(self._reconnect_event.wait()) + wait_tasks = [shutdown_task, monitor_task, reconnect_task] done, pending = await asyncio.wait( - [shutdown_task, monitor_task, reconnect_task], + wait_tasks, return_when=asyncio.FIRST_COMPLETED, ) for task in pending: @@ -516,31 +521,31 @@ class RuntimeMCPSession: # are exhausted or on success. raise # Re-raise so _lifecycle_loop_with_retry can catch it finally: - # Clean up all resources in the same task - try: - if self.exit_stack: - await self.exit_stack.aclose() - self.exit_stack = AsyncExitStack() - self.functions.clear() - self.resources.clear() - self.session = None - except Exception as e: - self.ap.logger.error(f'Error cleaning up MCP session {self.server_name}: {e}\n{traceback.format_exc()}') - finally: - # On a transport-only reconnect the managed process is healthy - # and will be re-attached on the next initialize(); do NOT stop - # it. Any other exit path fully tears the session down. - if getattr(self, '_preserve_managed_process', False): - self._preserve_managed_process = False - else: - await self._cleanup_box_stdio_session() + for task in wait_tasks: + if not task.done(): + task.cancel() + # AsyncExitStack contains AnyIO cancel scopes that must exit in + # the same task where their contexts were entered. + await self._cleanup_lifecycle_attempt() async def _lifecycle_loop_with_retry(self): """Wrap _lifecycle_loop with retry and exponential backoff.""" attempt = 0 while attempt <= self._MAX_RETRIES: + connection_generation_before = self._connection_generation try: - await self._lifecycle_loop() + lifecycle_task = asyncio.create_task(self._lifecycle_loop()) + try: + await lifecycle_task + except asyncio.CancelledError as exc: + if self._shutdown_event.is_set(): + return + if not self._uses_box_stdio(): + raise + error = self._prepare_box_transport_retry('Box MCP transport task was cancelled unexpectedly') + raise error from exc + if self._uses_box_stdio() and not self._shutdown_event.is_set(): + raise self._prepare_box_transport_retry('Box MCP lifecycle ended unexpectedly') return # Normal shutdown, don't retry except _TransportReconnect as e: # Transient WS transport drop while the managed process is still @@ -614,9 +619,15 @@ class RuntimeMCPSession: await asyncio.sleep(2) continue except Exception as e: - self.retry_count = attempt + 1 + self.retry_count += 1 if self._shutdown_event.is_set(): return # Shutdown requested, don't retry + # A lifecycle that reached CONNECTED proved its preceding + # startup attempt was healthy. Give a later runtime failure a + # fresh consecutive-startup retry budget instead of exhausting + # the budget across the entire process lifetime. + if self._connection_generation > connection_generation_before: + attempt = 0 # BOX_UNAVAILABLE is a deliberate refusal, not a transient # failure — retrying produces log spam and a misleading # "Failed after N attempts" message. Surface it immediately. @@ -635,7 +646,6 @@ class RuntimeMCPSession: f'MCP session {self.server_name} failed (attempt {attempt + 1}), ' f'retrying in {delay}s: {self._describe_exception(e)}' ) - await self._cleanup_box_stdio_session() # Reset status for retry self.status = MCPSessionStatus.CONNECTING self.error_message = None @@ -643,6 +653,59 @@ class RuntimeMCPSession: await asyncio.sleep(delay) attempt += 1 + def _prepare_box_transport_retry(self, message: str) -> RuntimeError: + error = RuntimeError(message) + self.status = MCPSessionStatus.ERROR + self.error_message = str(error) + self.error_phase = MCPSessionErrorPhase.RELAY_CONNECT + self.ap.logger.error(f'Error in MCP session lifecycle {self.server_name}: {error}') + self.functions.clear() + self.resources.clear() + self.resource_templates.clear() + self.resource_capabilities = {} + self._resource_cache.clear() + self.session = None + return error + + async def _cleanup_lifecycle_attempt(self) -> None: + stale_exit_stack = self.exit_stack + self.exit_stack = AsyncExitStack() + try: + async with asyncio.timeout(MCP_TRANSPORT_CLEANUP_TIMEOUT_SECONDS): + await stale_exit_stack.aclose() + except TimeoutError: + self.ap.logger.warning( + f'Timed out cleaning up MCP transport for {self.server_name}; continuing lifecycle recovery' + ) + except asyncio.CancelledError: + self.ap.logger.warning( + f'MCP transport cleanup was cancelled for {self.server_name}; continuing lifecycle recovery' + ) + except Exception as e: + self.ap.logger.warning( + f'Error cleaning up MCP transport for {self.server_name}; ' + f'continuing lifecycle recovery: {self._describe_exception(e)}' + ) + + self.functions.clear() + self.resources.clear() + self.resource_templates.clear() + self.resource_capabilities = {} + self._resource_cache.clear() + self.session = None + if self._preserve_managed_process: + self._preserve_managed_process = False + return + try: + await asyncio.wait_for( + self._cleanup_box_stdio_session(), + timeout=MCP_TRANSPORT_CLEANUP_TIMEOUT_SECONDS, + ) + except asyncio.TimeoutError: + self.ap.logger.warning( + f'Timed out cleaning up MCP managed process for {self.server_name}; continuing lifecycle recovery' + ) + @staticmethod def _describe_exception(exc: BaseException) -> str: """Flatten an exception into its underlying leaf messages. diff --git a/src/langbot/pkg/provider/tools/loaders/mcp_stdio.py b/src/langbot/pkg/provider/tools/loaders/mcp_stdio.py index 134c60a88..eca87fd62 100644 --- a/src/langbot/pkg/provider/tools/loaders/mcp_stdio.py +++ b/src/langbot/pkg/provider/tools/loaders/mcp_stdio.py @@ -185,11 +185,10 @@ class BoxStdioSessionRuntime: box_service = getattr(self.ap, 'box_service', None) if box_service is None: return False - # When Box is configured but currently unavailable (disabled or - # connection failed), do NOT silently fall through to host-stdio — - # that would bypass the sandbox the operator asked for. The caller - # is expected to refuse the stdio MCP server with a clear error. - return bool(getattr(box_service, 'available', False)) + # Transport selection follows operator intent, not momentary health. + # An enabled Box may be reconnecting; initialize() waits for it instead + # of falling through to unsandboxed host stdio or exhausting retries. + return bool(getattr(box_service, 'enabled', True)) async def initialize(self) -> None: await self._wait_for_box_runtime() diff --git a/tests/unit_tests/box/test_box_connector.py b/tests/unit_tests/box/test_box_connector.py index e160b0cab..1ee19019c 100644 --- a/tests/unit_tests/box/test_box_connector.py +++ b/tests/unit_tests/box/test_box_connector.py @@ -125,3 +125,24 @@ async def test_box_runtime_connector_heartbeat_failure_requests_reconnect(monkey await heartbeat assert callbacks == [connector] + + +@pytest.mark.asyncio +async def test_box_runtime_stdio_disconnect_requests_reconnect(monkeypatch: pytest.MonkeyPatch): + monkeypatch.setattr('langbot.pkg.utils.platform.standalone_box', False) + logger = Mock() + on_disconnect = AsyncMock() + handler = SimpleNamespace( + call_action=AsyncMock(return_value={}), + run=AsyncMock(return_value=None), + ) + monkeypatch.setattr('langbot.pkg.box.connector.Handler', Mock(return_value=handler)) + connector = BoxRuntimeConnector(make_app(logger), runtime_disconnect_callback=on_disconnect) + connected = asyncio.Event() + + callback = connector._make_connection_callback('stdio', connected, []) + await callback(Mock()) + + assert connected.is_set() + on_disconnect.assert_awaited_once_with(connector) + logger.error.assert_called_once_with('Disconnected from Box runtime, trying to reconnect...') diff --git a/tests/unit_tests/provider/test_mcp_box_integration.py b/tests/unit_tests/provider/test_mcp_box_integration.py index 6b0c141ed..de7cdc11b 100644 --- a/tests/unit_tests/provider/test_mcp_box_integration.py +++ b/tests/unit_tests/provider/test_mcp_box_integration.py @@ -6,6 +6,7 @@ triggering the circular import chain through the app module. from __future__ import annotations +import asyncio import importlib import importlib.util import os @@ -680,14 +681,29 @@ class TestGetRuntimeInfoDict: # ... but are isolated by distinct process_ids within that session. assert transient._box_stdio_runtime.process_id != live._box_stdio_runtime.process_id - def test_stdio_session_refuses_when_box_unavailable(self, mcp_module): - """Policy: when Box is configured but unavailable (disabled in config - OR connection failed), stdio MCP servers are NOT treated as box-stdio. - ``_init_stdio_python_server`` will raise a clear refusal at start - time; until then, the runtime info simply omits box_session_id so the - UI can render the disabled state cleanly.""" + def test_stdio_session_keeps_box_transport_while_runtime_reconnects(self, mcp_module): ap = _make_ap() ap.box_service.available = False + ap.box_service.enabled = True + s = _make_session( + mcp_module, + { + 'name': 'test', + 'uuid': 'test-uuid', + 'mode': 'stdio', + 'command': 'python', + 'args': [], + }, + ap=ap, + ) + info = s.get_runtime_info_dict() + assert info['box_session_id'] == 'mcp-shared' + assert info['box_enabled'] is True + + def test_stdio_session_refuses_when_box_is_disabled(self, mcp_module): + ap = _make_ap() + ap.box_service.available = False + ap.box_service.enabled = False s = _make_session( mcp_module, { @@ -757,6 +773,216 @@ class TestBoxConfigParsing: assert s.box_config.host_path_mode == 'ro' +@pytest.mark.asyncio +async def test_lifecycle_cleanup_timeout_does_not_block_box_stdio_retry(mcp_module, monkeypatch): + class HangingExitStack: + async def aclose(self): + await asyncio.Event().wait() + + session = _make_session( + mcp_module, + { + 'name': 'cleanup-timeout', + 'uuid': 'cleanup-timeout-uuid', + 'mode': 'stdio', + 'command': 'python', + 'args': [], + }, + ) + session.exit_stack = HangingExitStack() + session.functions.append(Mock()) + session.resources.append({'uri': 'test://stale'}) + session.session = Mock() + session._cleanup_box_stdio_session = AsyncMock() + monkeypatch.setattr(mcp_module, 'MCP_TRANSPORT_CLEANUP_TIMEOUT_SECONDS', 0.01) + + await asyncio.wait_for(session._cleanup_lifecycle_attempt(), timeout=1) + + assert isinstance(session.exit_stack, mcp_module.AsyncExitStack) + assert session.functions == [] + assert session.resources == [] + assert session.session is None + session._cleanup_box_stdio_session.assert_awaited_once() + session.ap.logger.warning.assert_called_once_with( + 'Timed out cleaning up MCP transport for cleanup-timeout; continuing lifecycle recovery' + ) + + +@pytest.mark.asyncio +async def test_lifecycle_transport_cleanup_error_is_recovery_warning(mcp_module): + class FailingExitStack: + async def aclose(self): + raise RuntimeError('transport already closed') + + session = _make_session( + mcp_module, + { + 'name': 'cleanup-error', + 'uuid': 'cleanup-error-uuid', + 'mode': 'stdio', + 'command': 'python', + 'args': [], + }, + ) + session.exit_stack = FailingExitStack() + session._cleanup_box_stdio_session = AsyncMock() + + await session._cleanup_lifecycle_attempt() + + session.ap.logger.warning.assert_called_once_with( + 'Error cleaning up MCP transport for cleanup-error; ' + 'continuing lifecycle recovery: RuntimeError: transport already closed' + ) + session.ap.logger.error.assert_not_called() + + +@pytest.mark.asyncio +async def test_unexpected_box_stdio_transport_cancellation_becomes_retryable_error(mcp_module, monkeypatch): + session = _make_session( + mcp_module, + { + 'name': 'cancelled-transport', + 'uuid': 'cancelled-transport-uuid', + 'mode': 'stdio', + 'command': 'python', + 'args': [], + }, + ) + session._uses_box_stdio = Mock(return_value=True) + lifecycle_calls = 0 + + async def cancelled_then_shutdown(): + nonlocal lifecycle_calls + lifecycle_calls += 1 + if lifecycle_calls == 1: + raise asyncio.CancelledError() + session._shutdown_event.set() + + session._lifecycle_loop = AsyncMock(side_effect=cancelled_then_shutdown) + session._cleanup_box_stdio_session = AsyncMock() + session.functions.append(Mock()) + session.resources.append({'uri': 'test://stale'}) + session.resource_templates.append({'uri_template': 'test://{id}'}) + session.resource_capabilities = {'subscribe': True} + session._resource_cache[('test://stale', 1, None, False)] = {'content': 'stale'} + session.session = Mock() + monkeypatch.setattr(session, '_RETRY_DELAYS', [0, 0, 0]) + + await session._lifecycle_loop_with_retry() + + assert session._lifecycle_loop.await_count == 2 + assert session.functions == [] + assert session.resources == [] + assert session.resource_templates == [] + assert session.resource_capabilities == {} + assert session._resource_cache == {} + assert session.session is None + session.ap.logger.error.assert_called_once_with( + 'Error in MCP session lifecycle cancelled-transport: ' + 'Box MCP transport task was cancelled unexpectedly' + ) + + +@pytest.mark.asyncio +async def test_cancelled_lifecycle_closes_transport_in_its_own_task(mcp_module): + class TrackingExitStack: + def __init__(self): + self.close_task = None + + async def aclose(self): + self.close_task = asyncio.current_task() + + session = _make_session( + mcp_module, + { + 'name': 'same-task-cleanup', + 'uuid': 'same-task-cleanup-uuid', + 'mode': 'stdio', + 'command': 'python', + 'args': [], + }, + ) + transport_stack = TrackingExitStack() + session.exit_stack = transport_stack + session._init_stdio_python_server = AsyncMock(side_effect=asyncio.CancelledError()) + session._cleanup_box_stdio_session = AsyncMock() + + lifecycle_task = asyncio.create_task(session._lifecycle_loop()) + with pytest.raises(asyncio.CancelledError): + await lifecycle_task + + assert transport_stack.close_task is lifecycle_task + session._cleanup_box_stdio_session.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_connected_lifecycle_failures_receive_fresh_retry_budgets(mcp_module, monkeypatch): + session = _make_session( + mcp_module, + { + 'name': 'repeated-runtime-recovery', + 'uuid': 'repeated-runtime-recovery-uuid', + 'mode': 'stdio', + 'command': 'python', + 'args': [], + }, + ) + session._uses_box_stdio = Mock(return_value=True) + session._cleanup_box_stdio_session = AsyncMock() + lifecycle_calls = 0 + + async def connected_then_failed_repeatedly(): + nonlocal lifecycle_calls + lifecycle_calls += 1 + if lifecycle_calls <= 4: + session._connection_generation += 1 + raise RuntimeError(f'runtime failure {lifecycle_calls}') + session._shutdown_event.set() + + session._lifecycle_loop = AsyncMock(side_effect=connected_then_failed_repeatedly) + monkeypatch.setattr(session, '_MAX_RETRIES', 1) + monkeypatch.setattr(session, '_RETRY_DELAYS', [0]) + + await session._lifecycle_loop_with_retry() + + assert session._lifecycle_loop.await_count == 5 + assert session.retry_count == 4 + assert session.status != mcp_module.MCPSessionStatus.ERROR + + +@pytest.mark.asyncio +async def test_unexpected_box_stdio_lifecycle_return_is_retried(mcp_module, monkeypatch): + session = _make_session( + mcp_module, + { + 'name': 'ended-transport', + 'uuid': 'ended-transport-uuid', + 'mode': 'stdio', + 'command': 'python', + 'args': [], + }, + ) + session._uses_box_stdio = Mock(return_value=True) + session._cleanup_box_stdio_session = AsyncMock() + lifecycle_calls = 0 + + async def ended_then_shutdown(): + nonlocal lifecycle_calls + lifecycle_calls += 1 + if lifecycle_calls == 2: + session._shutdown_event.set() + + session._lifecycle_loop = AsyncMock(side_effect=ended_then_shutdown) + monkeypatch.setattr(session, '_RETRY_DELAYS', [0, 0, 0]) + + await session._lifecycle_loop_with_retry() + + assert session._lifecycle_loop.await_count == 2 + session.ap.logger.error.assert_called_once_with( + 'Error in MCP session lifecycle ended-transport: Box MCP lifecycle ended unexpectedly' + ) + + @pytest.mark.asyncio async def test_init_box_stdio_server_stages_host_path_in_shared_workspace(mcp_module, tmp_path): mcp_stdio_module = sys.modules['langbot.pkg.provider.tools.loaders.mcp_stdio']