fix(mcp): recover box stdio sessions after disconnect

This commit is contained in:
huanghuoguoguo
2026-07-19 22:34:09 +08:00
parent 3359951837
commit 8afab5fa49
17 changed files with 1147 additions and 51 deletions
+123 -4
View File
@@ -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")
+11 -3
View File
@@ -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) {
+32
View File
@@ -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);
@@ -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);
@@ -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);
+34 -1
View File
@@ -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"
]
},
{
@@ -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
@@ -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"
@@ -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"]
},
{
@@ -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"));
}
+97
View File
@@ -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 {
+3 -9
View File
@@ -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
+85 -22
View File
@@ -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.
@@ -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()
@@ -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...')
@@ -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']