fix(agent): harden runner delivery validation

2026-08-09 04:40:57 +00:00 · 2026-07-27 16:01:57 +08:00
parent 7d3b53bbce
commit 9c32ea4063
11 changed files with 403 additions and 47 deletions
@@ -57,6 +57,9 @@ try {
  const python = join(repo, ".venv", "bin", "python");
  const args = ["scripts/e2e/agent-run-ledger-audit.py", "--repo", repo, "--output", auditPath];
  if (env.LANGBOT_AGENT_RUN_ID) args.push("--run-id", env.LANGBOT_AGENT_RUN_ID);
+  if (env.LANGBOT_AGENT_TOOL_AUTHORIZATION_MODE) {
+    args.push("--tool-authorization-mode", env.LANGBOT_AGENT_TOOL_AUTHORIZATION_MODE);
+  }
  const execution = await run(python, args, process.cwd());
  const report = JSON.parse(await readFile(auditPath, "utf8"));
  result.status = report.status;
@@ -16,7 +16,12 @@ import sqlalchemy
 import yaml
 from sqlalchemy.ext.asyncio import create_async_engine

-from agent_run_ledger_policy import classify_invalid_tool_argument_errors, load_ledger_json
+from agent_run_ledger_policy import (
+    classify_invalid_tool_argument_errors,
+    classify_tool_authorization,
+    invalid_tool_argument_error_signal,
+    load_ledger_json,
+)


 def database_url(repo: pathlib.Path) -> str:
@@ -80,6 +85,7 @@ async def audit(
    expected_tool_name: str | None = None,
    expected_parameters: dict | None = None,
    expected_result_text: str | None = None,
+    tool_authorization_mode: str = "strict",
 ) -> dict:
    engine = create_async_engine(database_url(repo))
    failures: list[dict] = []
@@ -175,7 +181,6 @@ async def audit(
        r"invalid json(?! arguments)|unauthori[sz]ed|permission denied|forbidden|timed?\s*out|timeout",
        re.I,
    )
-    invalid_tool_arguments_pattern = re.compile(r"invalid json arguments", re.I)

    def error_surface(value: object) -> list[str]:
        """Collect diagnostic fields without treating normal tool parameters as errors."""
@@ -221,10 +226,10 @@ async def audit(
        if match:
            suspicious_errors.append({"sequence": row["sequence"], "type": event_type, "signal": match.group(0)})
        elif event_type == "tool.call.completed":
-            match = invalid_tool_arguments_pattern.search(diagnostic_text)
-            if match:
+            signal = invalid_tool_argument_error_signal(diagnostic_text)
+            if signal:
                invalid_tool_argument_errors.append(
-                    {"sequence": row["sequence"], "type": event_type, "signal": match.group(0)}
+                    {"sequence": row["sequence"], "type": event_type, "signal": signal}
                )

    if run_row["status"] != "completed":
@@ -248,8 +253,12 @@ async def audit(
            failures.append({"kind": "tool_call_order", "tool_call_id": call_id})
        if started[0]["tool_name"] not in allowed_tools:
            unauthorized_calls.append({"tool_call_id": call_id, "tool_name": started[0]["tool_name"]})
-    if unauthorized_calls:
-        failures.append({"kind": "unauthorized_tool_calls", "calls": unauthorized_calls})
+    authorization_failures, authorization_warnings = classify_tool_authorization(
+        unauthorized_calls,
+        authorization_mode=tool_authorization_mode,
+    )
+    failures.extend(authorization_failures)
+    warnings.extend(authorization_warnings)
    unrecovered_argument_errors, recovered_argument_warnings = classify_invalid_tool_argument_errors(
        invalid_tool_argument_errors,
        successful_tool_completion_sequences=successful_tool_completion_sequences,
@@ -305,6 +314,8 @@ async def audit(
        "tool_call_completed": sum(len(items) for items in completions.values()),
        "tool_call_ids": len(all_call_ids),
        "authorized_tool_count": len(allowed_tools),
+        "tool_authorization_mode": tool_authorization_mode,
+        "runner_native_tool_call_count": len(unauthorized_calls) if tool_authorization_mode == "runner-native" else 0,
        "invalid_event_json": invalid_event_json,
        "suspicious_error_count": len(suspicious_errors),
        "recovered_tool_argument_error_count": len(recovered_argument_warnings),
@@ -334,6 +345,11 @@ def main() -> int:
    parser.add_argument("--expected-tool-name")
    parser.add_argument("--expected-parameters-json")
    parser.add_argument("--expected-result-text")
+    parser.add_argument(
+        "--tool-authorization-mode",
+        choices=("strict", "runner-native"),
+        default="strict",
+    )
    parser.add_argument("--output", required=True)
    args = parser.parse_args()
    try:
@@ -351,6 +367,7 @@ def main() -> int:
            expected_tool_name=args.expected_tool_name,
            expected_parameters=expected_parameters,
            expected_result_text=args.expected_result_text,
+            tool_authorization_mode=args.tool_authorization_mode,
        ))
    except Exception as exc:  # noqa: BLE001 - probe must classify environment failures
        report = {"status": "env_issue", "reason": str(exc), "failures": [], "warnings": []}
@@ -3,6 +3,13 @@
 from __future__ import annotations

 import json
+import re
+
+
+_INVALID_TOOL_ARGUMENT_PATTERN = re.compile(
+    r"invalid json arguments|\b\d+\s+validation errors?\s+for\s+[A-Za-z_][A-Za-z0-9_]*Args\b",
+    re.IGNORECASE,
+)


 def load_ledger_json(value: str | None, *, field: str, failures: list[dict]) -> object:
@@ -16,6 +23,12 @@ def load_ledger_json(value: str | None, *, field: str, failures: list[dict]) ->
        return {}


+def invalid_tool_argument_error_signal(value: str) -> str:
+    """Return the persisted signal for malformed model-supplied tool arguments."""
+    match = _INVALID_TOOL_ARGUMENT_PATTERN.search(value)
+    return match.group(0) if match else ""
+
+
 def classify_invalid_tool_argument_errors(
    events: list[dict],
    *,
@@ -41,3 +54,25 @@ def classify_invalid_tool_argument_errors(
        else:
            failures.append(event)
    return failures, warnings
+
+
+def classify_tool_authorization(
+    calls: list[dict],
+    *,
+    authorization_mode: str,
+) -> tuple[list[dict], list[dict]]:
+    """Classify tool names absent from the Host authorization snapshot."""
+    if not calls:
+        return [], []
+    if authorization_mode == "runner-native":
+        return [], [
+            {
+                "kind": "runner_native_tool_calls",
+                "calls": calls,
+                "reason": (
+                    "External runner tool telemetry is not a LangBot Host tool call; "
+                    "the runner's own permission system governs it."
+                ),
+            }
+        ]
+    return [{"kind": "unauthorized_tool_calls", "calls": calls}], []
@@ -1,6 +1,7 @@
 import {
  bodyText,
  clickFirstVisible,
+  clickFirstVisibleLocator,
  countOccurrences,
  gotoFrontend,
  isLoginUrl,
@@ -50,19 +51,21 @@ function debugChatInput(page) {
 }

 async function clickDebugChatTab(page) {
-  const tabByRole = page.getByRole("tab", { name: /Debug Chat|调试聊天|调试对话|Debug|调试/i }).first();
-  if (await tabByRole.isVisible({ timeout: 3_000 }).catch(() => false)) {
-    await tabByRole.click();
-    return true;
-  }
-
-  const tabBySelector = page.locator('[role="tab"]').filter({ hasText: /Debug Chat|调试聊天|调试对话|Debug|调试/i }).first();
-  if (await tabBySelector.isVisible({ timeout: 2_000 }).catch(() => false)) {
-    await tabBySelector.click();
-    return true;
-  }
-
-  return Boolean(await clickFirstVisible(page, ["Debug Chat", "调试聊天", "调试对话"], 2_000));
+  const label = /^(?:Debug Chat|调试聊天|调试对话|对话调试)$/i;
+  const configuredTimeout = Number.parseInt(
+    process.env.LANGBOT_E2E_UI_READY_TIMEOUT_MS
+      || process.env.LANGBOT_E2E_NAVIGATION_TIMEOUT_MS
+      || "30000",
+    10,
+  );
+  const timeout = Number.isFinite(configuredTimeout) && configuredTimeout > 0
+    ? configuredTimeout
+    : 30_000;
+  return await clickFirstVisibleLocator(page, [
+    page.getByRole("tab", { name: label }),
+    page.locator('[data-slot="tabs-trigger"]').filter({ hasText: label }),
+    page.getByText(label, { exact: true }),
+  ], timeout);
 }

 export async function waitForDebugChatReady(page, timeout = 20_000) {
@@ -103,6 +106,7 @@ export function classifyDebugChatResult({
  beforeMessages = null,
  afterMessages = null,
  latestAssistantText = "",
+  latestAssistantIsFinal = null,
  maxNewAssistantMessages = null,
  failureSignals = DEBUG_CHAT_FAILURE_SIGNALS,
 }) {
@@ -154,6 +158,18 @@ export function classifyDebugChatResult({
        ...assistantMessageEvidence,
      };
    }
+    if (latestAssistantIsFinal === false) {
+      return {
+        status: "fail",
+        reason: "The latest assistant message contained the expected text but was not final.",
+        min_expected_count: minExpectedCount,
+        final_count: finalCount,
+        before_assistant_expected_count: beforeAssistantExpectedCount,
+        after_assistant_expected_count: afterAssistantExpectedCount,
+        ...assistantMessageEvidence,
+        latest_assistant_is_final: false,
+      };
+    }
    if (maxNewAssistantMessages !== null && newAssistantMessageCount > maxNewAssistantMessages) {
      return {
        status: "fail",
@@ -241,8 +257,16 @@ export function classifyDebugChatResult({

 export async function openPipelineDebugChat(page, { pipelineUrl, pipelineName, envHint = "LANGBOT_PIPELINE_URL or LANGBOT_PIPELINE_NAME" }) {
  if (pipelineUrl) {
-    await page.goto(pipelineUrl, { waitUntil: "domcontentloaded" });
-    await page.waitForLoadState("networkidle", { timeout: 10_000 }).catch(() => {});
+    let alreadyAtPipeline = false;
+    try {
+      alreadyAtPipeline = new URL(page.url()).href === new URL(pipelineUrl).href;
+    } catch {
+      // Invalid URLs are handled by page.goto below.
+    }
+    if (!alreadyAtPipeline) {
+      await page.goto(pipelineUrl, { waitUntil: "commit" });
+      await page.waitForLoadState("networkidle", { timeout: 10_000 }).catch(() => {});
+    }
  } else {
    if (!pipelineName) {
      return {
@@ -394,6 +418,62 @@ export async function waitForDebugChatTextStable(page, { timeoutMs = 5_000, quie
  }
 }

+async function fetchDebugChatHistory(page, { backendUrl, pipelineId, sessionType }) {
+  if (!backendUrl || !pipelineId || !sessionType) {
+    return { status: "not_required", messages: [] };
+  }
+  return await page.evaluate(async ({ backendUrl, pipelineId, sessionType }) => {
+    const token = localStorage.getItem("token") || "";
+    const response = await fetch(
+      `${backendUrl.replace(/\/$/, "")}/api/v1/pipelines/${encodeURIComponent(pipelineId)}/ws/messages/${encodeURIComponent(sessionType)}`,
+      { headers: token ? { Authorization: `Bearer ${token}` } : {} },
+    );
+    const json = await response.json().catch(() => ({}));
+    return {
+      status: response.ok && json.code === 0 ? "ready" : "fail",
+      http_status: response.status,
+      code: json.code ?? null,
+      messages: json.data?.messages || [],
+      reason: response.ok && json.code === 0 ? "" : json.msg || `Debug Chat history returned HTTP ${response.status}.`,
+    };
+  }, { backendUrl, pipelineId, sessionType });
+}
+
+async function waitForFinalDebugChatAssistant(page, {
+  backendUrl,
+  pipelineId,
+  sessionType,
+  beforeAssistantCount,
+  timeoutMs,
+}) {
+  if (!backendUrl || !pipelineId || !sessionType) {
+    return { status: "not_required", latest_assistant_is_final: null };
+  }
+  const deadline = Date.now() + Math.max(1, timeoutMs);
+  let lastHistory = null;
+  while (Date.now() < deadline) {
+    lastHistory = await fetchDebugChatHistory(page, { backendUrl, pipelineId, sessionType });
+    if (lastHistory.status === "fail") return lastHistory;
+    const assistants = lastHistory.messages.filter((message) => message.role === "assistant");
+    const latest = assistants.at(-1);
+    if (assistants.length > beforeAssistantCount && latest?.is_final === true) {
+      return {
+        status: "pass",
+        latest_assistant_is_final: true,
+        assistant_message_count: assistants.length,
+      };
+    }
+    await page.waitForTimeout(Math.min(250, Math.max(1, deadline - Date.now())));
+  }
+  const assistants = (lastHistory?.messages || []).filter((message) => message.role === "assistant");
+  return {
+    status: "fail",
+    reason: "Timed out waiting for the new assistant message to become final.",
+    latest_assistant_is_final: assistants.at(-1)?.is_final === true,
+    assistant_message_count: assistants.length,
+  };
+}
+
 export async function attachDebugChatImage(page, imagePath) {
  if (!imagePath) return { status: "not_required", reason: "" };
  const input = page.locator('input[type="file"][accept*="image"], input[type="file"]').first();
@@ -429,11 +509,16 @@ export async function runDebugChatPrompt(page, {
  expectedTexts = null,
  responseTimeoutMs,
  imagePath = "",
+  backendUrl = "",
+  pipelineId = "",
+  sessionType = "person",
  maxNewAssistantMessages = null,
  failureSignals = DEBUG_CHAT_FAILURE_SIGNALS,
 }) {
  const beforeText = await bodyText(page);
  const beforeMessages = await visibleDebugChatMessages(page);
+  const beforeHistory = await fetchDebugChatHistory(page, { backendUrl, pipelineId, sessionType });
+  const beforeHistoryAssistantCount = beforeHistory.messages.filter((message) => message.role === "assistant").length;
  const requiredExpectedTexts = [...new Set(
    (Array.isArray(expectedTexts) && expectedTexts.length > 0 ? expectedTexts : [expectedText])
      .map(String)
@@ -449,6 +534,7 @@ export async function runDebugChatPrompt(page, {
    return { status: "fail", reason: "Could not find a Debug Chat text input." };
  }

+  const responseStartedAt = Date.now();
  await waitForExpectedDebugChatText(page, {
    expectedText,
    expectedTexts: requiredExpectedTexts,
@@ -459,6 +545,13 @@ export async function runDebugChatPrompt(page, {
    beforeText,
    failureSignals,
  });
+  const finalAssistant = await waitForFinalDebugChatAssistant(page, {
+    backendUrl,
+    pipelineId,
+    sessionType,
+    beforeAssistantCount: beforeHistoryAssistantCount,
+    timeoutMs: Math.max(1, responseTimeoutMs - (Date.now() - responseStartedAt)),
+  });
  await waitForDebugChatTextStable(page);

  const afterText = await bodyText(page);
@@ -468,7 +561,7 @@ export async function runDebugChatPrompt(page, {
  const failureText = findNewFailureSignal(beforeText, afterText, failureSignals);
  const latestFailureLeaf = failureText ? await latestVisibleLeafText(page, [failureText]) : "";

-  return classifyDebugChatResult({
+  const classified = classifyDebugChatResult({
    beforeText,
    afterText,
    expectedText,
@@ -479,9 +572,16 @@ export async function runDebugChatPrompt(page, {
    beforeMessages,
    afterMessages,
    latestAssistantText,
+    latestAssistantIsFinal: finalAssistant.latest_assistant_is_final,
    maxNewAssistantMessages,
    failureSignals,
  });
+  return {
+    ...classified,
+    latest_assistant_is_final: finalAssistant.latest_assistant_is_final,
+    final_assistant_wait_status: finalAssistant.status,
+    final_assistant_wait_reason: finalAssistant.reason || "",
+  };
 }

 export async function setDebugChatStreamOutput(page, desired) {
@@ -428,6 +428,10 @@ export async function createBrowser(paths) {
    context = await browser.newContext({ viewport: { width: 1440, height: 960 } });
  }
  const page = context.pages()[0] || await context.newPage();
+  const navigationTimeoutMs = Number.parseInt(env.LANGBOT_E2E_NAVIGATION_TIMEOUT_MS || "30000", 10);
+  if (Number.isFinite(navigationTimeoutMs) && navigationTimeoutMs > 0) {
+    page.setDefaultNavigationTimeout(navigationTimeoutMs);
+  }

  page.on("console", (message) => {
    appendLine(paths.consoleLog, `[${message.type()}] ${message.text()}`).catch(() => {});
@@ -483,29 +487,42 @@ export function countOccurrences(haystack, needle) {
  return String(haystack).split(needle).length - 1;
 }

-export async function clickFirstVisible(page, labels, timeout = 2_000) {
-  for (const label of labels) {
-    const roleButton = page.getByRole("button", { name: label }).first();
-    if (await roleButton.isVisible({ timeout }).catch(() => false)) {
-      await roleButton.click();
-      return label;
+async function clickVisibleCandidate(page, candidates, timeout) {
+  const deadline = Date.now() + Math.max(1, timeout);
+  do {
+    for (const candidate of candidates) {
+      const count = await candidate.locator.count().catch(() => 0);
+      for (let index = 0; index < count; index += 1) {
+        const element = candidate.locator.nth(index);
+        if (!await element.isVisible().catch(() => false)) continue;
+        const remaining = Math.max(1, deadline - Date.now());
+        const clicked = await element.click({ timeout: Math.min(1_000, remaining) })
+          .then(() => true)
+          .catch(() => false);
+        if (clicked) return candidate.value;
+      }
    }
-
-    const roleLink = page.getByRole("link", { name: label }).first();
-    if (await roleLink.isVisible({ timeout }).catch(() => false)) {
-      await roleLink.click();
-      return label;
-    }
-
-    const text = page.getByText(label, { exact: false }).first();
-    if (await text.isVisible({ timeout }).catch(() => false)) {
-      await text.click();
-      return label;
-    }
-  }
+    const remaining = deadline - Date.now();
+    if (remaining <= 0) break;
+    await page.waitForTimeout(Math.min(100, remaining));
+  } while (Date.now() < deadline);
  return null;
 }

+export async function clickFirstVisibleLocator(page, locators, timeout = 2_000) {
+  const candidates = locators.map((locator) => ({ locator, value: true }));
+  return Boolean(await clickVisibleCandidate(page, candidates, timeout));
+}
+
+export async function clickFirstVisible(page, labels, timeout = 2_000) {
+  const candidates = labels.flatMap((label) => [
+    { locator: page.getByRole("button", { name: label }), value: label },
+    { locator: page.getByRole("link", { name: label }), value: label },
+    { locator: page.getByText(label, { exact: false }), value: label },
+  ]);
+  return await clickVisibleCandidate(page, candidates, timeout);
+}
+
 export async function fillFirstTextInput(page, value) {
  const candidates = [
    page.getByRole("textbox").last(),
@@ -57,6 +57,7 @@ const resetDebugChat = boolFromEnv(env.LANGBOT_E2E_RESET_DEBUG_CHAT, false);
 const restoreRunnerConfig = boolFromEnv(env.LANGBOT_E2E_RESTORE_RUNNER_CONFIG, true);
 const restoreExtensions = boolFromEnv(env.LANGBOT_E2E_RESTORE_EXTENSIONS, true);
 const debugChatSessionType = env.LANGBOT_E2E_DEBUG_CHAT_SESSION_TYPE || "person";
+const maxNewAssistantMessages = Number.parseInt(env.LANGBOT_E2E_MAX_NEW_ASSISTANT_MESSAGES || "1", 10);
 const pipelineConfigDiagnosticPath = resolve(paths.evidenceDir, "pipeline-config-diagnostic.json");
 const pipelineExtensionsDiagnosticPath = resolve(paths.evidenceDir, "pipeline-extensions-diagnostic.json");
 const debugChatResetDiagnosticPath = resolve(paths.evidenceDir, "debug-chat-reset-diagnostic.json");
@@ -873,7 +874,7 @@ try {
  result.expected_text = promptSteps.at(-1)?.expectedText || expectedText;

  const authDiagnostic = await ensureAuthenticatedBrowser(page, {
-    frontendUrl: env.LANGBOT_FRONTEND_URL || "",
+    frontendUrl: pipelineUrl || env.LANGBOT_FRONTEND_URL || "",
    backendUrl,
  });
  result.browser_auth = authDiagnostic;
@@ -999,7 +1000,12 @@ try {
            expectedTexts: step.expectedTexts,
            responseTimeoutMs: step.responseTimeoutMs,
            imagePath: index === 0 ? imagePath : "",
-            maxNewAssistantMessages: streamOutput === false ? 1 : null,
+            backendUrl,
+            pipelineId: result.pipeline_config?.pipeline_id || pipelineIdFromUrl(pipelineUrl),
+            sessionType: debugChatSessionType,
+            maxNewAssistantMessages: Number.isFinite(maxNewAssistantMessages) && maxNewAssistantMessages >= 0
+              ? maxNewAssistantMessages
+              : 1,
            failureSignals: failureSignals.length > 0 ? failureSignals : undefined,
          });
          const promptDurationMs = Date.now() - promptStartedAt;
@@ -1017,6 +1023,9 @@ try {
            before_assistant_message_count: chatResult.before_assistant_message_count,
            after_assistant_message_count: chatResult.after_assistant_message_count,
            new_assistant_message_count: chatResult.new_assistant_message_count,
+            latest_assistant_is_final: chatResult.latest_assistant_is_final,
+            final_assistant_wait_status: chatResult.final_assistant_wait_status,
+            final_assistant_wait_reason: chatResult.final_assistant_wait_reason,
            failure_signal: chatResult.failure_signal || "",
            missing_expected_texts: chatResult.missing_expected_texts || [],
          });
@@ -16,6 +16,7 @@ skills:
 automation: scripts/e2e/agent-run-ledger-audit.mjs
 steps:
  - "Set LANGBOT_AGENT_RUN_ID to audit a specific run, or leave it unset to audit the latest persisted AgentRunner run."
+  - "For an external runner's own CLI tools, set LANGBOT_AGENT_TOOL_AUTHORIZATION_MODE=runner-native; keep the default strict mode for Host tool calls."
  - "Read the active LangBot database configuration and inspect the selected run and its ordered events."
  - "Verify completed terminal state, run.completed, paired tool.call.started/completed events, stable tool names, and monotonic ordering."
  - "Compare called tools with the authorization snapshot and validate each advertised tool has owner/source, description, and parameter schema."
@@ -32,6 +33,7 @@ evidence_required:
 diagnostics:
  - "Run this immediately after a complex UI Agent task so latest-run selection cannot drift to unrelated traffic."
  - "Use LANGBOT_AGENT_RUN_ID when multiple operators share the test instance."
+  - "runner-native mode still checks tool pairing, stable names, ordering, terminal state, and error signals; the external runner's permission system owns its local tools."
 success_patterns:
  - "Agent run ledger audit passed"
 failure_patterns:
@@ -60,6 +60,7 @@ import {
 } from "../scripts/e2e/lib/debug-chat.mjs";
 import {
  beginBackendLogCapture,
+  clickFirstVisible,
  ensureAuthenticatedBrowser,
  finishBackendLogCapture,
  resolveLangBotRepo,
@@ -68,6 +69,32 @@ import {

 const root = process.cwd();

+test("clickFirstVisible waits for a later visible DOM match", async () => {
+  let pollCount = 0;
+  let clickedIndex = -1;
+  const emptyLocator = {
+    count: async () => 0,
+    nth: () => { throw new Error("empty locator has no children"); },
+  };
+  const textLocator = {
+    count: async () => 2,
+    nth: (index: number) => ({
+      isVisible: async () => index === 1 && pollCount >= 1,
+      click: async () => { clickedIndex = index; },
+    }),
+  };
+  const page = {
+    getByRole: () => emptyLocator,
+    getByText: () => textLocator,
+    waitForTimeout: async () => { pollCount += 1; },
+  };
+
+  const clicked = await clickFirstVisible(page, ["Debug Chat"], 1_000);
+
+  assert.equal(clicked, "Debug Chat");
+  assert.equal(clickedIndex, 1);
+});
+
 test("repo root detects the skills tree before generated bin exists", () => {
  const tmp = mkdtempSync(join(tmpdir(), "lbs-root-no-bin-"));
  try {
@@ -2064,6 +2091,25 @@ test("debug chat classifier passes when expected text appears in a new assistant
  assert.equal(result.new_assistant_message_count, 1);
 });

+test("debug chat classifier rejects a matching assistant message that is not final", () => {
+  const result = classifyDebugChatResult({
+    beforeText: "",
+    afterText: "Bot: E2E_OK:skill",
+    expectedText: "E2E_OK:skill",
+    prompt: "Run the task",
+    latestExpectedLeaf: "E2E_OK:skill",
+    latestFailureLeaf: "",
+    beforeMessages: [],
+    afterMessages: [{ role: "assistant", text: "E2E_OK:skill" }],
+    latestAssistantText: "E2E_OK:skill",
+    latestAssistantIsFinal: false,
+  });
+
+  assert.equal(result.status, "fail");
+  assert.match(result.reason, /was not final/);
+  assert.equal(result.latest_assistant_is_final, false);
+});
+
 test("debug chat classifier accepts formatted responses containing every required fragment", () => {
  const expectedTexts = ["MULTITOOL_COMBO_FINAL", "passcode-6718", "rag-7421", "tool-a", "tool-b"];
  const result = classifyDebugChatResult({
@@ -9,6 +9,8 @@ sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts" / "e2e"))

 from agent_run_ledger_policy import (  # noqa: E402
    classify_invalid_tool_argument_errors,
+    classify_tool_authorization,
+    invalid_tool_argument_error_signal,
    load_ledger_json,
 )

@@ -64,6 +66,34 @@ class AgentRunLedgerPolicyTests(unittest.TestCase):
        self.assertEqual(failures[0]["kind"], "invalid_json")
        self.assertEqual(failures[0]["field"], "agent_run_event[10].data_json")

+    def test_detects_pydantic_tool_argument_validation_error(self) -> None:
+        signal = invalid_tool_argument_error_signal(
+            "1 validation error for RetrieveKnowledgeArgs\n"
+            "kb_id\n  Field required [type=missing, input_value={'top_k': 3}, input_type=dict]"
+        )
+
+        self.assertEqual(signal, "1 validation error for RetrieveKnowledgeArgs")
+
+    def test_ignores_unrelated_tool_execution_error(self) -> None:
+        self.assertEqual(invalid_tool_argument_error_signal("Exit code 1"), "")
+
+    def test_strict_mode_rejects_tool_absent_from_host_authorization(self) -> None:
+        call = {"tool_call_id": "call-1", "tool_name": "Read"}
+
+        failures, warnings = classify_tool_authorization([call], authorization_mode="strict")
+
+        self.assertEqual(failures, [{"kind": "unauthorized_tool_calls", "calls": [call]}])
+        self.assertEqual(warnings, [])
+
+    def test_runner_native_mode_records_non_host_tool_as_warning(self) -> None:
+        call = {"tool_call_id": "call-1", "tool_name": "Read"}
+
+        failures, warnings = classify_tool_authorization([call], authorization_mode="runner-native")
+
+        self.assertEqual(failures, [])
+        self.assertEqual(warnings[0]["kind"], "runner_native_tool_calls")
+        self.assertEqual(warnings[0]["calls"], [call])
+

 if __name__ == "__main__":
    unittest.main()
@@ -142,7 +142,18 @@ class WebSocketAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter)
        embed_target = self._parse_embed_target(sender_id)
        if embed_target is not None:
            return embed_target
-        return typing.cast(str, self.ap.platform_mgr.websocket_proxy_bot.bot_entity.use_pipeline_uuid), None
+        pipeline_uuid = getattr(message_source, '_langbot_pipeline_uuid', None)
+        if isinstance(pipeline_uuid, str) and pipeline_uuid:
+            return pipeline_uuid, None
+
+        legacy_pipeline_uuid = getattr(
+            self.ap.platform_mgr.websocket_proxy_bot.bot_entity,
+            'use_pipeline_uuid',
+            '',
+        )
+        if legacy_pipeline_uuid:
+            return typing.cast(str, legacy_pipeline_uuid), None
+        raise RuntimeError(f'Could not resolve WebSocket reply context for sender {sender_id!r}')

    async def send_message(
        self,
@@ -306,8 +317,9 @@ class WebSocketAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter)
            # 更新历史记录中的对应消息
            message_list[existing_index] = message_data

-        if message_is_final and resp_message_id:
-            stream_message_indexes.pop(resp_message_id, None)
+        # Keep the index for the lifetime of the history entry. Some runners
+        # emit a final delta followed by message.completed/run.completed. They
+        # all share one Host response id and must update one UI message.

        await ws_connection_manager.broadcast_to_pipeline(
            pipeline_uuid,
@@ -7,6 +7,9 @@ from unittest.mock import AsyncMock, Mock
 import pytest

 import langbot_plugin.api.entities.builtin.platform.events as platform_events
+import langbot_plugin.api.entities.builtin.platform.entities as platform_entities
+import langbot_plugin.api.entities.builtin.platform.message as platform_message
+import langbot_plugin.api.entities.builtin.provider.message as provider_message
 from langbot.pkg.platform.sources import websocket_adapter as websocket_adapter_module
 from langbot.pkg.platform.sources.websocket_adapter import WebSocketAdapter, WebSocketMessage, WebSocketSession
 from langbot.pkg.platform.sources.websocket_manager import WebSocketConnectionManager, is_valid_session_id
@@ -164,6 +167,88 @@ async def test_stable_session_launcher_resolves_to_active_connection(monkeypatch
    assert await manager.get_connection_by_session_id(session_id, 'pipeline-1') is None


+@pytest.mark.asyncio
+async def test_dashboard_reply_uses_event_pipeline_after_connection_closes(monkeypatch):
+    manager = WebSocketConnectionManager()
+    connection = await manager.add_connection(
+        websocket=Mock(),
+        pipeline_uuid='pipeline-1',
+        session_type='person',
+    )
+    monkeypatch.setattr(websocket_adapter_module, 'ws_connection_manager', manager)
+
+    app = Mock()
+    app.platform_mgr.websocket_proxy_bot.bot_entity = Mock(spec=[])
+    adapter = WebSocketAdapter.model_construct(ap=app, logger=AsyncMock())
+    message_source = platform_events.FriendMessage(
+        sender=platform_entities.Friend(
+            id=f'websocket_{connection.connection_id}',
+            nickname='User',
+            remark='User',
+        ),
+        message_chain=platform_message.MessageChain([platform_message.Plain(text='hello')]),
+        time=1,
+    )
+    object.__setattr__(message_source, '_langbot_pipeline_uuid', 'pipeline-1')
+    await manager.remove_connection(connection.connection_id)
+
+    assert await adapter._get_message_context(message_source) == ('pipeline-1', None)
+
+
+@pytest.mark.asyncio
+async def test_late_final_events_update_one_stream_message(monkeypatch):
+    manager = WebSocketConnectionManager()
+    connection = await manager.add_connection(
+        websocket=Mock(),
+        pipeline_uuid='pipeline-1',
+        session_type='person',
+    )
+    monkeypatch.setattr(websocket_adapter_module, 'ws_connection_manager', manager)
+
+    adapter = WebSocketAdapter.model_construct(ap=Mock(), logger=AsyncMock())
+    adapter.websocket_person_session = WebSocketSession(id='person')
+    adapter.websocket_group_session = WebSocketSession(id='group')
+    message_source = platform_events.FriendMessage(
+        sender=platform_entities.Friend(
+            id=f'websocket_{connection.connection_id}',
+            nickname='User',
+            remark='User',
+        ),
+        message_chain=platform_message.MessageChain([platform_message.Plain(text='hello')]),
+        time=1,
+    )
+    first = provider_message.MessageChunk(
+        role='assistant',
+        content='first final',
+        is_final=True,
+        resp_message_id='response-1',
+    )
+    second = provider_message.MessageChunk(
+        role='assistant',
+        content='corrected final',
+        is_final=True,
+        resp_message_id='response-1',
+    )
+
+    await adapter.reply_message_chunk(
+        message_source,
+        first,
+        platform_message.MessageChain([platform_message.Plain(text='first final')]),
+        is_final=True,
+    )
+    await adapter.reply_message_chunk(
+        message_source,
+        second,
+        platform_message.MessageChain([platform_message.Plain(text='corrected final')]),
+        is_final=True,
+    )
+
+    messages = adapter.get_websocket_messages('pipeline-1', 'person')
+    assert len(messages) == 1
+    assert messages[0]['content'] == 'corrected final'
+    assert messages[0]['is_final'] is True
+
+
 def test_session_ids_must_be_canonical_random_uuids():
    assert is_valid_session_id('31c0f2e9-b115-4ee6-8f15-3e624d6456b1')
    assert not is_valid_session_id('session-a')