fix(chat): restore tool execution details

Merge PR #1 after CLI, contract, test, and container gates passed.

.dockerignore

@@ -1,7 +1,11 @@
 .git
 node_modules
 .opencode/node_modules
+.env
+.env.*
 .local.env
+data/
+logs/
 dist
 .vscode
 *.log

.gitea/workflows/deploy-transport-canary.yml

@@ -1,25 +0,0 @@
-name: Dev deployment transport canary
-
-on:
-  workflow_dispatch:
-
-jobs:
-  verify-runner-to-dev:
-    runs-on: ubuntu-22.04
-    steps:
-      - name: Configure restricted deployment key
-        env:
-          DEV_DEPLOY_SSH_KEY: ${{ secrets.DEV_DEPLOY_SSH_KEY }}
-        run: |
-          set -eu
-          test -n "$DEV_DEPLOY_SSH_KEY"
-          install -d -m 700 ~/.ssh
-          printf '%s\n' "$DEV_DEPLOY_SSH_KEY" > ~/.ssh/id_ed25519
-          chmod 600 ~/.ssh/id_ed25519
-          ssh-keyscan -H 192.168.1.114 >> ~/.ssh/known_hosts
-      - name: Verify deployment endpoint and candidate registry pull
-        run: |
-          set -eu
-          ssh -i ~/.ssh/id_ed25519 cicd-deploy@192.168.1.114 verify
-          ssh -i ~/.ssh/id_ed25519 cicd-deploy@192.168.1.114 \
-            'candidate agent gitea.waternetwork.cn/orgtjwater/tjwateragent:latest'

.gitea/workflows/package-v2.yml

@@ -1,23 +0,0 @@
-name: Agent CI/CD v2
-
-on:
-  push:
-    tags:
-      - "v*"
-  workflow_dispatch: {}
-
-jobs:
-  build-test-publish-and-deploy:
-    uses: OrgTJWater/ci-templates/.gitea/workflows/container-cd.yml@main
-    with:
-      image_name: gitea.waternetwork.cn/orgtjwater/tjwateragent
-      dockerfile: Dockerfile
-      build_context: .
-      cache_image: gitea.waternetwork.cn/orgtjwater/tjwateragent:ci-cache
-      test_target: build
-      deploy_service: agent
-      deploy_host: 192.168.1.114
-    secrets:
-      REGISTRY_USERNAME: ${{ secrets.REGISTRY_USERNAME }}
-      REGISTRY_PASSWORD: ${{ secrets.REGISTRY_PASSWORD }}
-      DEV_DEPLOY_SSH_KEY: ${{ secrets.DEV_DEPLOY_SSH_KEY }}

.gitea/workflows/package.yml

@@ -1,221 +1,23 @@
-name: Agent CI/CD
+name: Agent CI/CD v2
 
 on:
   push:
     tags:
       - "v*"
-      - "latest"
   workflow_dispatch: {}
 
 jobs:
-  docker-image:
+  build-test-publish-and-deploy:
-    runs-on: ubuntu-22.04
+    uses: OrgTJWater/ci-templates/.gitea/workflows/container-cd.yml@main
-    if: startsWith(github.ref, 'refs/tags/')
+    with:
-    permissions:
+      image_name: gitea.waternetwork.cn/orgtjwater/tjwateragent
-      contents: read
+      dockerfile: Dockerfile
-    defaults:
+      build_context: .
-      run:
+      cache_image: gitea.waternetwork.cn/orgtjwater/tjwateragent:ci-cache
-        shell: bash
+      test_target: test
-
+      deploy_service: agent
-    steps:
+      deploy_host: 192.168.1.114
-      - name: Setup tools
+    secrets:
-        run: |
-          sudo apt-get update -qq && sudo apt-get install -y -qq jq
-          jq --version
-
-      - name: Checkout code
-        env:
-          SERVER_URL: ${{ github.server_url }}
-          REPOSITORY: ${{ github.repository }}
-          COMMIT_SHA: ${{ github.sha }}
-          GIT_USERNAME: ${{ github.actor }}
-          GIT_TOKEN: ${{ github.token }}
-        run: |
-          case "$SERVER_URL" in
-            http://*)
-              AUTH_SERVER_URL="http://${GIT_USERNAME}:${GIT_TOKEN}@${SERVER_URL#http://}"
-              ;;
-            https://*)
-              AUTH_SERVER_URL="https://${GIT_USERNAME}:${GIT_TOKEN}@${SERVER_URL#https://}"
-              ;;
-            *)
-              AUTH_SERVER_URL="$SERVER_URL"
-              ;;
-          esac
-
-          if [ ! -d .git ]; then
-            git init .
-          fi
-
-          if git remote get-url origin >/dev/null 2>&1; then
-            git remote set-url origin "${AUTH_SERVER_URL}/${REPOSITORY}.git"
-          else
-            git remote add origin "${AUTH_SERVER_URL}/${REPOSITORY}.git"
-          fi
-
-          git fetch --depth=1 origin "$COMMIT_SHA"
-          git checkout --force --detach FETCH_HEAD
-          git clean -ffdx
-
-      - name: Normalize image metadata
-        env:
-          RAW_REGISTRY_HOST: ${{ vars.REGISTRY_HOST }}
-          RAW_REPOSITORY: ${{ github.repository }}
-          RAW_REF: ${{ github.ref }}
-          RAW_REF_NAME: ${{ github.ref_name }}
-        run: |
-          RAW_REGISTRY_HOST="$(printf '%s' "${RAW_REGISTRY_HOST}" | sed 's/^[[:space:]]*//;s/[[:space:]]*$//')"
-
-          if [ -z "${RAW_REGISTRY_HOST}" ]; then
-            echo "Missing required repository variable: REGISTRY_HOST"
-            exit 1
-          fi
-
-          REGISTRY_HOST="${RAW_REGISTRY_HOST#http://}"
-          REGISTRY_HOST="${REGISTRY_HOST#https://}"
-          REGISTRY_HOST="${REGISTRY_HOST%/}"
-
-          if [ -z "${REGISTRY_HOST}" ]; then
-            echo "Repository variable REGISTRY_HOST resolves to an empty host"
-            exit 1
-          fi
-
-          REPOSITORY_PATH="${RAW_REPOSITORY#/}"
-          IMAGE_REPOSITORY_PATH="$(printf '%s' "$REPOSITORY_PATH" | tr '[:upper:]' '[:lower:]')"
-          IMAGE_NAME="${REGISTRY_HOST}/${IMAGE_REPOSITORY_PATH}"
-          IMAGE_TAG="${RAW_REF_NAME}"
-          {
-            echo "REGISTRY_HOST=${REGISTRY_HOST}"
-            echo "REPOSITORY_PATH=${REPOSITORY_PATH}"
-            echo "IMAGE_REPOSITORY_PATH=${IMAGE_REPOSITORY_PATH}"
-            echo "IMAGE_NAME=${IMAGE_NAME}"
-            echo "IMAGE_TAG=${IMAGE_TAG}"
-            echo "IMAGE_REF=${IMAGE_NAME}:${IMAGE_TAG}"
-          } >> "$GITHUB_ENV"
-
-      - name: Login to Gitea Container Registry
-        env:
       REGISTRY_USERNAME: ${{ secrets.REGISTRY_USERNAME }}
       REGISTRY_PASSWORD: ${{ secrets.REGISTRY_PASSWORD }}
-        run: |
+      DEV_DEPLOY_SSH_KEY: ${{ secrets.DEV_DEPLOY_SSH_KEY }}
-          if [ -z "${REGISTRY_HOST:-}" ]; then
-            echo "Missing resolved environment value: REGISTRY_HOST"
-            echo "The previous step should write REGISTRY_HOST into GITHUB_ENV."
-            exit 1
-          fi
-
-          if [ -z "${REGISTRY_USERNAME}" ]; then
-            echo "Missing required repository secret: REGISTRY_USERNAME"
-            exit 1
-          fi
-
-          if [ -z "${REGISTRY_PASSWORD}" ]; then
-            echo "Missing required repository secret: REGISTRY_PASSWORD"
-            exit 1
-          fi
-
-          echo "Logging into registry host: ${REGISTRY_HOST}"
-          echo "${REGISTRY_PASSWORD}" | docker login "$REGISTRY_HOST" \
-            --username "${REGISTRY_USERNAME}" \
-            --password-stdin
-
-      - name: Build and Push Image
-        run: |
-          if [ -z "${IMAGE_NAME:-}" ] || [ -z "${IMAGE_TAG:-}" ]; then
-            echo "Missing resolved image metadata: IMAGE_NAME or IMAGE_TAG"
-            exit 1
-          fi
-
-          push_with_retry() {
-            image_ref="$1"
-            attempt=1
-            max_attempts=3
-
-            while [ "$attempt" -le "$max_attempts" ]; do
-              if docker push "$image_ref"; then
-                return 0
-              fi
-
-              if [ "$attempt" -eq "$max_attempts" ]; then
-                return 1
-              fi
-
-              echo "Push failed for $image_ref (attempt $attempt/$max_attempts); retrying in 10s..."
-              attempt=$((attempt + 1))
-              sleep 10
-            done
-          }
-
-          if [ "${IMAGE_TAG}" = "latest" ]; then
-            docker build \
-              --network=host \
-              -f ./Dockerfile \
-              -t "${IMAGE_NAME}:latest" \
-              .
-            push_with_retry "${IMAGE_NAME}:latest"
-          else
-            docker build \
-              --network=host \
-              -f ./Dockerfile \
-              -t "${IMAGE_NAME}:${IMAGE_TAG}" \
-              -t "${IMAGE_NAME}:latest" \
-              .
-            push_with_retry "${IMAGE_NAME}:${IMAGE_TAG}"
-            push_with_retry "${IMAGE_NAME}:latest"
-          fi
-
-      - name: Notify Deploy Server
-        run: |
-          post_deploy_webhook() {
-            label="$1"
-            payload="$2"
-            webhook_url="${{ vars.DEPLOY_WEBHOOK_URL }}"
-            token="${{ secrets.DEPLOY_WEBHOOK_TOKEN }}"
-
-            # Trim whitespace
-            webhook_url=$(echo "$webhook_url" | xargs)
-
-            echo "[$label] Calling webhook: $webhook_url"
-
-            http_code=$(curl -sS -D /tmp/deploy_headers.txt -o /tmp/deploy_response.txt -w "%{http_code}" -X POST "$webhook_url" \
-              -H "Content-Type: application/json" \
-              -H "Authorization: Bearer $token" \
-              -d "$payload")
-
-            echo "[$label] webhook HTTP status: ${http_code}"
-            if [ "$http_code" -ge 200 ] && [ "$http_code" -lt 300 ]; then
-              return 0
-            fi
-
-            echo "[$label] response headers:"
-            cat /tmp/deploy_headers.txt
-            echo "[$label] response body:"
-            cat /tmp/deploy_response.txt
-            return 1
-          }
-
-          PRIMARY_PAYLOAD="{\"image\":\"${IMAGE_REF}\",\"tag\":\"${IMAGE_TAG}\",\"repo\":\"${REPOSITORY_PATH}\"}"
-          FALLBACK_PAYLOAD="{\"image\":\"${IMAGE_REF}\",\"tag\":\"${IMAGE_TAG}\",\"repo\":\"${IMAGE_REPOSITORY_PATH}\"}"
-
-          echo "Deploy webhook target: ${{ vars.DEPLOY_WEBHOOK_URL }}"
-          echo "Deploy payload(primary): image=${IMAGE_REF}, tag=${IMAGE_TAG}, repo=${REPOSITORY_PATH}"
-          if post_deploy_webhook "primary" "$PRIMARY_PAYLOAD"; then
-            exit 0
-          fi
-
-          echo "Primary webhook request failed, retrying with lowercase repo path..."
-          echo "Deploy payload(fallback): image=${IMAGE_REF}, tag=${IMAGE_TAG}, repo=${IMAGE_REPOSITORY_PATH}"
-          if post_deploy_webhook "fallback" "$FALLBACK_PAYLOAD"; then
-            exit 0
-          fi
-
-          echo "Deploy webhook failed after primary and fallback attempts."
-          exit 1
-
-  deploy-fallback-log:
-    runs-on: ubuntu-22.04
-    needs: docker-image
-    if: failure()
-    steps:
-      - name: Deployment not triggered
-        run: echo "Image build/push failed, deployment webhook was not called."

.opencode/agents/instruction.md

@@ -2,10 +2,14 @@
 description: TJWater Agent，用于供水网络分析和操作员工作流
 mode: primary
 model: deepseek/deepseek-v4-flash
-temperature: 0.2
 ---
 你是 TJWater 供水管网分析 Agent，运用水力专业知识，回复用户时使用简体中文，内容要求简洁准确。
 
+## 回复要求
+
+- 工具执行期间不输出过程说明，全部完成后只回复最终结果
+- 直接给出结论、关键数据和可执行建议，默认仅展示最重要的 Top 5；数据不足或任务失败时简要说明影响和下一步
+
 ## 工作流生命周期
 
 Skills 树是**动态生长的**——工作流不是预置的，而是从实际任务中沉淀出来的：

Dockerfile

@@ -65,6 +65,15 @@ COPY cli ./cli
 COPY .opencode ./.opencode
 RUN bun run check
 
+FROM build AS test
+RUN apt-get update && apt-get install -y --no-install-recommends nodejs && \
+    rm -rf /var/lib/apt/lists/*
+COPY contracts ./contracts
+COPY node-tests ./node-tests
+COPY scripts ./scripts
+COPY tests ./tests
+RUN bun run test:ci
+
 FROM build AS runner
 WORKDIR /app
 

package.json

@@ -8,7 +8,9 @@
     "install:opencode": "bun install --cwd .opencode",
     "typecheck": "tsc --noEmit -p tsconfig.json",
     "typecheck:opencode": "bun run --cwd .opencode typecheck",
+    "test": "bun test tests",
     "test:cli": "node --test node-tests/cli/*.node.mjs",
+    "test:ci": "bun run contract:check && /usr/bin/node --test node-tests/cli/*.node.mjs && bun test tests",
     "dev": "bun --watch src/server.ts",
     "build": "bun run check",
     "check": "bun run typecheck && bun run typecheck:opencode",

src/cli/executeCliCommand.ts

@@ -0,0 +1,196 @@
+import { spawn } from "node:child_process";
+
+import { type RuntimeSessionContext } from "../runtime/sessionContext.js";
+
+type OutputStream = "stdout" | "stderr";
+
+export type CliExecutionResult = {
+  outcome: "completed" | "timeout" | "output_limit";
+  exitCode: number | null;
+  signal: NodeJS.Signals | null;
+  status: number;
+  stderr: string;
+  stdout: string;
+  exceededStream?: OutputStream;
+};
+
+type ExecuteCliCommandOptions = {
+  apiBaseUrl: string;
+  cliPath: string;
+  maxOutputBytes: number;
+  terminationGraceMs?: number;
+};
+
+const getCompletedStatus = (exitCode: number | null, stdout: string) => {
+  let errorCode = "";
+  try {
+    const payload = JSON.parse(stdout) as { error?: { code?: unknown } };
+    errorCode =
+      typeof payload.error?.code === "string" ? payload.error.code : "";
+  } catch {
+    errorCode = "";
+  }
+
+  if (errorCode === "HTTP_401" || errorCode === "UNAUTHENTICATED") {
+    return 401;
+  }
+  if (errorCode === "HTTP_403") {
+    return 403;
+  }
+  return exitCode === 0 ? 200 : 502;
+};
+
+export const executeCliCommand = async (
+  context: RuntimeSessionContext,
+  command: string,
+  timeoutSec: number,
+  options: ExecuteCliCommandOptions,
+): Promise<CliExecutionResult> => {
+  const maxOutputBytes = options.maxOutputBytes;
+  if (!Number.isSafeInteger(maxOutputBytes) || maxOutputBytes <= 0) {
+    throw new Error("maxOutputBytes must be a positive safe integer");
+  }
+
+  const child = spawn(
+    options.cliPath,
+    ["--auth-stdin", ...command.split(/\s+/).filter(Boolean)],
+    { stdio: ["pipe", "pipe", "pipe"] },
+  );
+  const stdoutChunks: Buffer[] = [];
+  const stderrChunks: Buffer[] = [];
+  let stdoutBytes = 0;
+  let stderrBytes = 0;
+  let terminationReason:
+    | "timeout"
+    | "output_limit"
+    | "execution_error"
+    | null = null;
+  let exceededStream: OutputStream | undefined;
+  let terminationStarted = false;
+  let settled = false;
+  let executionError: Error | null = null;
+  let forceKillTimer: ReturnType<typeof setTimeout> | undefined;
+
+  const result = await new Promise<CliExecutionResult>((resolve, reject) => {
+    const cleanup = () => {
+      clearTimeout(timeoutTimer);
+      if (forceKillTimer) {
+        clearTimeout(forceKillTimer);
+      }
+    };
+
+    const terminate = (
+      reason: "timeout" | "output_limit" | "execution_error",
+    ) => {
+      if (terminationStarted) {
+        return;
+      }
+      terminationStarted = true;
+      terminationReason = reason;
+
+      if (child.exitCode === null && child.signalCode === null) {
+        child.kill("SIGTERM");
+      }
+      forceKillTimer = setTimeout(() => {
+        if (child.exitCode === null && child.signalCode === null) {
+          child.kill("SIGKILL");
+        }
+      }, options.terminationGraceMs ?? 1500);
+    };
+
+    const capture = (stream: OutputStream, data: Buffer) => {
+      if (terminationReason) {
+        return;
+      }
+      const chunks = stream === "stdout" ? stdoutChunks : stderrChunks;
+      const bytes = stream === "stdout" ? stdoutBytes : stderrBytes;
+      if (bytes + data.length > maxOutputBytes) {
+        exceededStream = stream;
+        terminate("output_limit");
+        return;
+      }
+      chunks.push(data);
+      if (stream === "stdout") {
+        stdoutBytes += data.length;
+      } else {
+        stderrBytes += data.length;
+      }
+    };
+
+    const timeoutTimer = setTimeout(() => {
+      if (child.exitCode === null && child.signalCode === null) {
+        terminate("timeout");
+      }
+    }, timeoutSec * 1000);
+
+    child.stdout.on("data", (data: Buffer) => capture("stdout", data));
+    child.stderr.on("data", (data: Buffer) => capture("stderr", data));
+    child.stdin.on("error", (error) => {
+      if (terminationReason === null) {
+        executionError = error;
+        terminate("execution_error");
+      }
+    });
+    child.on("error", (error) => {
+      if (terminationReason === null) {
+        executionError = error;
+        terminate("execution_error");
+      }
+    });
+    child.on("close", (exitCode, signal) => {
+      if (settled) {
+        return;
+      }
+      settled = true;
+      cleanup();
+      if (terminationReason === "timeout") {
+        resolve({
+          outcome: "timeout",
+          exitCode,
+          signal,
+          status: 504,
+          stderr: "",
+          stdout: "",
+        });
+        return;
+      }
+      if (executionError) {
+        reject(executionError);
+        return;
+      }
+      if (terminationReason === "output_limit") {
+        resolve({
+          outcome: "output_limit",
+          exceededStream,
+          exitCode,
+          signal,
+          status: 502,
+          stderr: "",
+          stdout: "",
+        });
+        return;
+      }
+
+      const stdout = Buffer.concat(stdoutChunks, stdoutBytes).toString("utf-8");
+      const stderr = Buffer.concat(stderrChunks, stderrBytes).toString("utf-8");
+      resolve({
+        outcome: "completed",
+        exitCode,
+        signal,
+        status: getCompletedStatus(exitCode, stdout),
+        stderr,
+        stdout,
+      });
+    });
+
+    child.stdin.end(
+      JSON.stringify({
+        server: options.apiBaseUrl,
+        access_token: context.accessToken,
+        project_id: context.projectId,
+      }),
+    );
+  });
+
+  return result;
+};

src/routes/chatStream.ts

@@ -111,24 +111,40 @@ const toRuntimeModel = (model?: SupportedModel) => {
   };
 };
 
-const emitFallbackMessage = async (
+const emitFinalMessage = async (
   runtime: OpencodeRuntimeAdapter,
   sessionId: string,
   clientSessionId: string,
+  currentAssistantMessageIds: Set<string>,
+  assistantTextParts: Map<string, Map<string, string>>,
   write: (event: string, data: Record<string, unknown>) => void,
 ) => {
+  let text = [...currentAssistantMessageIds]
+    .reverse()
+    .map((messageId) => [...(assistantTextParts.get(messageId)?.values() ?? [])].join(""))
+    .find((content) => content.length > 0) ?? "";
+
+  if (!text) {
     const messages = await runtime.messages(sessionId);
     const assistantMessage = [...messages]
       .reverse()
-    .find((message) => message.info.role === "assistant");
+      .find(
-  const parts = assistantMessage?.parts ?? [];
+        (message) =>
-  const text = collectTextContent(parts);
+          message.info.role === "assistant" &&
+          (currentAssistantMessageIds.size === 0 ||
+            currentAssistantMessageIds.has(message.info.id)),
+      );
+    text = collectTextContent(assistantMessage?.parts ?? []);
+  }
+
   if (text) {
     write("token", {
       session_id: clientSessionId,
       content: text,
     });
+    return true;
   }
+  return false;
 };
 
 export const streamPromptResponse = async ({
@@ -156,15 +172,14 @@ export const streamPromptResponse = async ({
   const emittedToolParts = new Set<string>();
   const emittedQuestionToolParts = new Set<string>();
   const emittedQuestionRequestIds = new Set<string>();
+  const currentAssistantMessageIds = new Set<string>();
+  const assistantTextParts = new Map<string, Map<string, string>>();
   const partTypes = new Map<string, Part["type"]>();
-  const pendingPartTextDeltas = new Map<string, string[]>();
+  const pendingTextDeltas = new Map<string, string[]>();
-  const reasoningDeltas = new Map<string, string[]>();
   const reasoningStatuses = new Map<string, "running" | "completed">();
   const toolStatuses = new Map<string, string>();
   let firstSessionEventLogged = false;
   let firstNonStatusEventLogged = false;
-  let firstTokenLogged = false;
-  let firstReasoningLogged = false;
   let firstToolEventLogged = false;
   let lastSessionStatus: string | null = null;
   let lastSessionStatusMessage: string | null = null;
@@ -626,45 +641,26 @@ export const streamPromptResponse = async ({
       if (event.type === "message.updated") {
         if (event.properties.info.role === "assistant") {
           sawResponseActivity = true;
+          currentAssistantMessageIds.add(event.properties.info.id);
         }
         continue;
       }
 
       if (event.type === "message.part.delta" && event.properties.field === "text") {
         sawResponseActivity = true;
+        currentAssistantMessageIds.add(event.properties.messageID);
         const partType = partTypes.get(event.properties.partID);
         if (partType === "text") {
-          if (!firstTokenLogged) {
+          const messageParts = assistantTextParts.get(event.properties.messageID) ?? new Map();
-            firstTokenLogged = true;
+          messageParts.set(
-            logDevelopmentDebug("first response token emitted", {
+            event.properties.partID,
-              ...debugContext,
+            `${messageParts.get(event.properties.partID) ?? ""}${event.properties.delta}`,
-              partId: event.properties.partID,
+          );
-              elapsedMs: Math.max(0, Date.now() - requestStartedAt),
+          assistantTextParts.set(event.properties.messageID, messageParts);
-              sincePromptDispatchMs: Math.max(0, Date.now() - promptStartedAt),
-            });
-          }
-          emittedText = true;
-          write("token", {
-            session_id: clientSessionId,
-            content: event.properties.delta,
-          });
-        } else if (partType === "reasoning") {
-          if (!firstReasoningLogged) {
-            firstReasoningLogged = true;
-            logDevelopmentDebug("first reasoning delta received", {
-              ...debugContext,
-              partId: event.properties.partID,
-              elapsedMs: Math.max(0, Date.now() - requestStartedAt),
-              sincePromptDispatchMs: Math.max(0, Date.now() - promptStartedAt),
-            });
-          }
-          const pending = reasoningDeltas.get(event.properties.partID) ?? [];
-          pending.push(event.properties.delta);
-          reasoningDeltas.set(event.properties.partID, pending);
         } else if (!partType) {
-          const pending = pendingPartTextDeltas.get(event.properties.partID) ?? [];
+          const pending = pendingTextDeltas.get(event.properties.partID) ?? [];
           pending.push(event.properties.delta);
-          pendingPartTextDeltas.set(event.properties.partID, pending);
+          pendingTextDeltas.set(event.properties.partID, pending);
         }
         continue;
       }
@@ -673,23 +669,19 @@ export const streamPromptResponse = async ({
         sawResponseActivity = true;
         const part = event.properties.part;
         partTypes.set(part.id, part.type);
+        if (part.type === "text" || part.type === "reasoning" || part.type === "tool") {
+          currentAssistantMessageIds.add(part.messageID);
+        }
         if (part.type === "text") {
-          const pending = pendingPartTextDeltas.get(part.id) ?? [];
+          const pendingText = (pendingTextDeltas.get(part.id) ?? []).join("");
-          pendingPartTextDeltas.delete(part.id);
+          pendingTextDeltas.delete(part.id);
-          for (const content of pending) {
+          const messageParts = assistantTextParts.get(part.messageID) ?? new Map();
-            emittedText = true;
+          messageParts.set(part.id, part.text || pendingText);
-            write("token", {
+          assistantTextParts.set(part.messageID, messageParts);
-              session_id: clientSessionId,
+        } else {
-              content,
+          pendingTextDeltas.delete(part.id);
-            });
         }
-        } else if (part.type === "reasoning") {
+        if (part.type === "reasoning") {
-          const pending = pendingPartTextDeltas.get(part.id) ?? [];
-          if (pending.length > 0) {
-            const existing = reasoningDeltas.get(part.id) ?? [];
-            reasoningDeltas.set(part.id, existing.concat(pending));
-          }
-          pendingPartTextDeltas.delete(part.id);
           const reasoningStatus = part.time.end ? "completed" : "running";
           if (reasoningStatuses.get(part.id) !== reasoningStatus) {
             reasoningStatuses.set(part.id, reasoningStatus);
@@ -697,14 +689,10 @@ export const streamPromptResponse = async ({
               ...debugContext,
               partId: part.id,
               status: reasoningStatus,
-              chunkCount: (reasoningDeltas.get(part.id) ?? []).length,
               elapsedMs: Math.max(0, Date.now() - requestStartedAt),
             });
           }
-          const reasoningDetail = buildReasoningProgressDetail(
+          const reasoningDetail = buildReasoningProgressDetail(part.time.end);
-            reasoningDeltas.get(part.id) ?? [],
-            part.time.end,
-          );
           emitProgress({
             id: part.id,
             phase: "planning",
@@ -932,13 +920,14 @@ export const streamPromptResponse = async ({
     }
 
     await promptPromise;
-    if (!emittedText) {
+    emittedText = await emitFinalMessage(
-      logDevelopmentDebug("no streamed text emitted, falling back to messages()", {
+      runtime,
-        ...debugContext,
+      sessionId,
-        elapsedMs: Math.max(0, Date.now() - requestStartedAt),
+      clientSessionId,
-      });
+      currentAssistantMessageIds,
-      await emitFallbackMessage(runtime, sessionId, clientSessionId, write);
+      assistantTextParts,
-    }
+      write,
+    );
     emitProgress({
       id: "request-received",
       phase: "start",

src/routes/chatStreamEvents.ts

@@ -376,12 +376,6 @@ const formatProgressValue = (value: unknown): string => {
   }
 };
 
-const normalizeProgressText = (chunks: string[]) =>
-  chunks.join("").replace(/\s+/g, " ").trim();
-
-const truncateProgressText = (text: string, maxLength: number) =>
-  text.length > maxLength ? `${text.slice(0, maxLength - 3)}...` : text;
-
 const summarizeToolParams = (params: Record<string, unknown>) => {
   const ignoredKeys = new Set(["reason", "request_reason", "why", "purpose", "rationale"]);
   const summary = Object.entries(params)
@@ -413,19 +407,8 @@ export const buildSessionStatusDetail = (status: { type: string; message?: strin
 };
 
 export const buildReasoningProgressDetail = (
-  chunks: string[],
   ended?: string | number | Date | null,
-) => {
+) => ended ? "分析步骤已整理完成。" : "Agent 正在分析问题。";
-  const reasoningText = truncateProgressText(normalizeProgressText(chunks), 800);
-  if (ended) {
-    return reasoningText
-      ? `推理过程：${reasoningText}`
-      : "当前推理阶段已完成，Agent 将继续输出答案或进入工具执行。";
-  }
-  return reasoningText
-    ? `正在推理：${reasoningText}`
-    : "Agent 正在拆解问题、梳理执行步骤并判断是否需要调用工具。";
-};
 
 export const buildToolProgressDetail = (
   tool: string,

src/server.ts

@@ -1,5 +1,4 @@
 import { randomUUID } from "node:crypto";
-import { spawn } from "node:child_process";
 import cors from "cors";
 import express from "express";
 
@@ -11,6 +10,7 @@ import {
   runWithCredentialRefresh,
 } from "./auth/credentialRefresh.js";
 import { SessionTranscriptStore } from "./sessions/transcriptStore.js";
+import { executeCliCommand } from "./cli/executeCliCommand.js";
 import { ChatSessionBridge } from "./chat/sessionBridge.js";
 import { config } from "./config.js";
 import { SessionUiStateStore } from "./sessions/uiStateStore.js";
@@ -229,7 +229,12 @@ app.post("/internal/tools/tjwater-cli-call", async (req, res) => {
     result = await runWithCredentialRefresh(
       credentialRefreshCoordinator,
       context,
-      (activeContext) => executeCliCommand(activeContext, command, timeoutSec),
+      (activeContext) =>
+        executeCliCommand(activeContext, command, timeoutSec, {
+          apiBaseUrl: config.TJWATER_API_BASE_URL,
+          cliPath: config.TJWATER_CLI_PATH,
+          maxOutputBytes: config.MAX_INLINE_RESULT_BYTES,
+        }),
     );
   } catch (error) {
     if (!(error instanceof CredentialRefreshError)) {
@@ -266,6 +271,20 @@ app.post("/internal/tools/tjwater-cli-call", async (req, res) => {
     return;
   }
 
+  if (result.outcome === "output_limit") {
+    res.status(502).json({
+      ok: false,
+      schema_version: "tjwater-cli/v1",
+      summary: "CLI 输出超过安全限制",
+      error: {
+        code: "OUTPUT_LIMIT_EXCEEDED",
+        message: `${result.exceededStream ?? "output"} exceeded ${config.MAX_INLINE_RESULT_BYTES} bytes`,
+        retryable: false,
+      },
+    });
+    return;
+  }
+
   if (result.status === 401) {
     markAuthExpired(
       getRuntimeSessionContext(sessionId) ?? context,
@@ -300,68 +319,6 @@ app.post("/internal/tools/tjwater-cli-call", async (req, res) => {
   }
 });
 
-const executeCliCommand = async (
-  context: RuntimeSessionContext,
-  command: string,
-  timeoutSec: number,
-) => {
-  const child = spawn(
-    config.TJWATER_CLI_PATH,
-    ["--auth-stdin", ...command.split(/\s+/).filter(Boolean)],
-    { stdio: ["pipe", "pipe", "pipe"] },
-  );
-  let stdout = "";
-  let stderr = "";
-  child.stdout.on("data", (data: Buffer) => {
-    stdout += data.toString("utf-8");
-  });
-  child.stderr.on("data", (data: Buffer) => {
-    stderr += data.toString("utf-8");
-  });
-  child.stdin.write(
-    JSON.stringify({
-      server: config.TJWATER_API_BASE_URL,
-      access_token: context.accessToken,
-      project_id: context.projectId,
-    }),
-  );
-  child.stdin.end();
-
-  const exitCode = await new Promise<number | null>((resolve, reject) => {
-    const timer = setTimeout(() => {
-      child.kill("SIGTERM");
-      resolve(-1);
-    }, timeoutSec * 1000);
-    child.on("close", (code) => {
-      clearTimeout(timer);
-      resolve(code);
-    });
-    child.on("error", (error) => {
-      clearTimeout(timer);
-      reject(error);
-    });
-  });
-  let errorCode = "";
-  try {
-    const payload = JSON.parse(stdout) as { error?: { code?: unknown } };
-    errorCode =
-      typeof payload.error?.code === "string" ? payload.error.code : "";
-  } catch {
-    errorCode = "";
-  }
-  const status =
-    exitCode === -1
-      ? 504
-      : errorCode === "HTTP_401" || errorCode === "UNAUTHENTICATED"
-        ? 401
-        : errorCode === "HTTP_403"
-          ? 403
-          : exitCode === 0
-            ? 200
-            : 502;
-  return { exitCode, status, stderr, stdout };
-};
-
 app.post("/internal/tools/store-render-ref", async (req, res) => {
   if (req.header("x-agent-internal-token") !== internalToken) {
     res.status(403).json({ message: "forbidden" });

tests/cli/executeCliCommand.test.ts

@@ -0,0 +1,122 @@
+import { describe, expect, test } from "bun:test";
+import { fileURLToPath } from "node:url";
+
+import { executeCliCommand } from "../../src/cli/executeCliCommand.js";
+import { type RuntimeSessionContext } from "../../src/runtime/sessionContext.js";
+
+const cliPath = fileURLToPath(
+  new URL("../fixtures/fakeCli.mjs", import.meta.url),
+);
+
+const context: RuntimeSessionContext = {
+  accessToken: "test-token",
+  actorKey: "actor-1",
+  clientSessionId: "client-1",
+  projectId: "project-1",
+  projectKey: "project-1",
+  sessionId: "session-1",
+  traceId: "trace-1",
+};
+
+const run = (
+  command: string,
+  options: {
+    maxOutputBytes?: number;
+    terminationGraceMs?: number;
+    timeoutSec?: number;
+  } = {},
+) =>
+  executeCliCommand(context, command, options.timeoutSec ?? 1, {
+    apiBaseUrl: "http://127.0.0.1:8000",
+    cliPath,
+    maxOutputBytes: options.maxOutputBytes ?? 64,
+    terminationGraceMs: options.terminationGraceMs ?? 20,
+  });
+
+describe("executeCliCommand", () => {
+  test("accepts output at the byte limit", async () => {
+    await expect(run("stdout 123456", { maxOutputBytes: 6 })).resolves.toMatchObject({
+      outcome: "completed",
+      exitCode: 0,
+      status: 200,
+      stdout: "123456",
+    });
+  });
+
+  test("rejects multibyte output above the byte limit without returning a partial body", async () => {
+    await expect(run("stdout 水水", { maxOutputBytes: 5 })).resolves.toMatchObject({
+      outcome: "output_limit",
+      exceededStream: "stdout",
+      status: 502,
+      stderr: "",
+      stdout: "",
+    });
+  });
+
+  test("limits stderr independently", async () => {
+    await expect(run("stderr 1234567", { maxOutputBytes: 6 })).resolves.toMatchObject({
+      outcome: "output_limit",
+      exceededStream: "stderr",
+      status: 502,
+      stderr: "",
+      stdout: "",
+    });
+  });
+
+  test("waits for a SIGTERM-aware process to close after timeout", async () => {
+    const startedAt = Date.now();
+    const result = await run("term", {
+      terminationGraceMs: 100,
+      timeoutSec: 0.25,
+    });
+
+    expect(result).toMatchObject({ outcome: "timeout", status: 504 });
+    expect(Date.now() - startedAt).toBeGreaterThanOrEqual(270);
+  });
+
+  test("uses SIGKILL when a timed-out process ignores SIGTERM", async () => {
+    const result = await run("ignore-term", { timeoutSec: 0.25 });
+
+    expect(result).toMatchObject({
+      outcome: "timeout",
+      signal: "SIGKILL",
+      status: 504,
+    });
+  });
+
+  test("keeps the timeout outcome when closing stdin also errors", async () => {
+    const largeContext = {
+      ...context,
+      accessToken: "x".repeat(1024 * 1024),
+    };
+
+    await expect(
+      executeCliCommand(largeContext, "ignore-term", 0.25, {
+        apiBaseUrl: "http://127.0.0.1:8000",
+        cliPath,
+        maxOutputBytes: 64,
+        terminationGraceMs: 20,
+      }),
+    ).resolves.toMatchObject({
+      outcome: "timeout",
+      signal: "SIGKILL",
+      status: 504,
+    });
+  });
+
+  test("rejects a deterministic stdin pipe error without crashing", async () => {
+    const largeContext = {
+      ...context,
+      accessToken: "x".repeat(1024 * 1024),
+    };
+
+    await expect(
+      executeCliCommand(largeContext, "closed-stdin", 1, {
+        apiBaseUrl: "http://127.0.0.1:8000",
+        cliPath,
+        maxOutputBytes: 64,
+        terminationGraceMs: 20,
+      }),
+    ).rejects.toBeInstanceOf(Error);
+  });
+});

tests/fixtures/fakeCli.mjs

@@ -0,0 +1,33 @@
+#!/usr/bin/env node
+
+import { closeSync } from "node:fs";
+
+const command = process.argv[3];
+const value = process.argv[4] ?? "";
+
+if (command === "stdout") {
+  process.stdout.write(value);
+  process.exit(0);
+}
+
+if (command === "stderr") {
+  process.stderr.write(value);
+  process.exit(1);
+}
+
+if (command === "term") {
+  process.on("SIGTERM", () => {
+    setTimeout(() => process.exit(0), 30);
+  });
+  setInterval(() => undefined, 1000);
+}
+
+if (command === "ignore-term") {
+  process.on("SIGTERM", () => undefined);
+  setInterval(() => undefined, 1000);
+}
+
+if (command === "closed-stdin") {
+  closeSync(0);
+  setInterval(() => undefined, 1000);
+}

tests/routes/chatStream.test.ts

@@ -15,6 +15,242 @@ const createEventStream = (events: unknown[]) => ({
 });
 
 describe("streamPromptResponse", () => {
+  it("emits only the final assistant text after tool-driven intermediate messages", async () => {
+    const runtime = {
+      subscribeEvents: async () =>
+        createEventStream([
+          {
+            type: "message.part.delta",
+            properties: {
+              sessionID: "runtime-session-1",
+              messageID: "assistant-intermediate",
+              partID: "text-part-intermediate",
+              field: "text",
+              delta: "正在加载工作流并尝试分页参数。",
+            },
+          },
+          {
+            type: "message.part.updated",
+            properties: {
+              sessionID: "runtime-session-1",
+              part: {
+                id: "text-part-intermediate",
+                sessionID: "runtime-session-1",
+                messageID: "assistant-intermediate",
+                type: "text",
+                text: "正在加载工作流并尝试分页参数。",
+                time: { start: 1, end: 2 },
+              },
+              time: 2,
+            },
+          },
+          {
+            type: "message.part.delta",
+            properties: {
+              sessionID: "runtime-session-1",
+              messageID: "assistant-final",
+              partID: "text-part-final",
+              field: "text",
+              delta: "共识别 56 条瓶颈管段，建议优先改造 Top 5。",
+            },
+          },
+          {
+            type: "message.part.updated",
+            properties: {
+              sessionID: "runtime-session-1",
+              part: {
+                id: "text-part-final",
+                sessionID: "runtime-session-1",
+                messageID: "assistant-final",
+                type: "text",
+                text: "共识别 56 条瓶颈管段，建议优先改造 Top 5。",
+                time: { start: 3, end: 4 },
+              },
+              time: 4,
+            },
+          },
+          {
+            type: "session.idle",
+            properties: { sessionID: "runtime-session-1" },
+          },
+        ]),
+      prompt: async () => undefined,
+      messages: async () => [
+        {
+          info: { id: "assistant-intermediate", role: "assistant" },
+          parts: [
+            {
+              id: "text-part-intermediate",
+              sessionID: "runtime-session-1",
+              messageID: "assistant-intermediate",
+              type: "text",
+              text: "正在加载工作流并尝试分页参数。",
+            },
+          ],
+        },
+        {
+          info: { id: "assistant-final", role: "assistant" },
+          parts: [
+            {
+              id: "text-part-final",
+              sessionID: "runtime-session-1",
+              messageID: "assistant-final",
+              type: "text",
+              text: "共识别 56 条瓶颈管段，建议优先改造 Top 5。",
+            },
+          ],
+        },
+      ],
+    } as unknown as OpencodeRuntimeAdapter;
+    const events: Array<{ event: string; data: Record<string, unknown> }> = [];
+
+    await streamPromptResponse({
+      runtime,
+      sessionId: "runtime-session-1",
+      clientSessionId: "client-session-1",
+      message: "分析管网瓶颈",
+      write: (event, data) => events.push({ event, data }),
+    });
+
+    expect(
+      events
+        .filter((item) => item.event === "token")
+        .map((item) => item.data.content)
+        .join(""),
+    ).toBe("共识别 56 条瓶颈管段，建议优先改造 Top 5。");
+  });
+
+  it("uses the final text event cache when the messages lookup fails", async () => {
+    const runtime = {
+      subscribeEvents: async () =>
+        createEventStream([
+          {
+            type: "message.part.updated",
+            properties: {
+              sessionID: "runtime-session-1",
+              part: {
+                id: "text-part-final",
+                sessionID: "runtime-session-1",
+                messageID: "assistant-final",
+                type: "text",
+                text: "最终分析结果。",
+                time: { start: 1, end: 2 },
+              },
+              time: 2,
+            },
+          },
+          {
+            type: "session.idle",
+            properties: { sessionID: "runtime-session-1" },
+          },
+        ]),
+      prompt: async () => undefined,
+      messages: async () => {
+        throw new Error("transient messages lookup failure");
+      },
+    } as unknown as OpencodeRuntimeAdapter;
+    const events: Array<{ event: string; data: Record<string, unknown> }> = [];
+
+    const result = await streamPromptResponse({
+      runtime,
+      sessionId: "runtime-session-1",
+      clientSessionId: "client-session-1",
+      message: "分析管网瓶颈",
+      write: (event, data) => events.push({ event, data }),
+    });
+
+    expect(result.failed).toBe(false);
+    expect(
+      events
+        .filter((item) => item.event === "token")
+        .map((item) => item.data.content)
+        .join(""),
+    ).toBe("最终分析结果。");
+  });
+
+  it("keeps reasoning generic while preserving tool execution details", async () => {
+    const runtime = {
+      subscribeEvents: async () =>
+        createEventStream([
+          {
+            type: "message.part.updated",
+            properties: {
+              sessionID: "runtime-session-1",
+              part: {
+                id: "reasoning-part-1",
+                sessionID: "runtime-session-1",
+                messageID: "assistant-1",
+                type: "reasoning",
+                text: "内部推理：尝试 limit=5000 并读取临时路径。",
+                time: { start: 1, end: 2 },
+              },
+              time: 2,
+            },
+          },
+          {
+            type: "message.part.delta",
+            properties: {
+              sessionID: "runtime-session-1",
+              messageID: "assistant-1",
+              partID: "reasoning-part-1",
+              field: "text",
+              delta: "内部推理：尝试 limit=5000 并读取临时路径。",
+            },
+          },
+          {
+            type: "message.part.updated",
+            properties: {
+              sessionID: "runtime-session-1",
+              part: {
+                id: "tool-part-1",
+                sessionID: "runtime-session-1",
+                messageID: "assistant-1",
+                type: "tool",
+                callID: "call-1",
+                tool: "tjwater_cli",
+                state: {
+                  status: "error",
+                  input: {
+                    command: "network get-all-pipes-properties --limit 5000",
+                    reason: "尝试突破分页限制",
+                  },
+                  error: "HTTP_422 raw backend payload with trace_id=secret-trace",
+                  time: { start: 1, end: 2 },
+                },
+              },
+              time: 2,
+            },
+          },
+          {
+            type: "session.idle",
+            properties: { sessionID: "runtime-session-1" },
+          },
+        ]),
+      prompt: async () => undefined,
+      messages: async () => [],
+    } as unknown as OpencodeRuntimeAdapter;
+    const events: Array<{ event: string; data: Record<string, unknown> }> = [];
+
+    await streamPromptResponse({
+      runtime,
+      sessionId: "runtime-session-1",
+      clientSessionId: "client-session-1",
+      message: "分析管网瓶颈",
+      write: (event, data) => events.push({ event, data }),
+    });
+
+    const reasoningProgress = events.find(
+      (item) => item.event === "progress" && item.data.id === "reasoning-part-1",
+    );
+    const toolProgress = events.find(
+      (item) => item.event === "progress" && item.data.id === "tool-part-1",
+    );
+    expect(reasoningProgress?.data.detail).toBe("分析步骤已整理完成。");
+    expect(toolProgress?.data.detail).toBe(
+      "tjwater_cli 调用失败；调用原因：尝试突破分页限制；关键参数：command=network get-all-pipes-properties --limit 5000；错误：HTTP_422 raw backend payload with trace_id=secret-trace",
+    );
+  });
+
   it("forwards opencode permission requests as SSE payloads", async () => {
     const runtime = {
       subscribeEvents: async () =>