AI 编程 Agent 工具输出遭截断：应记录 stdout_bytes 和截断标志

一个编码 Agent 关闭了那个不稳定的测试工单。JSONL span 显示了 pytest 和一个简短的预览。下一次人工运行时在同一个模块失败了。

wrapper 把 stdout 限制在了 8192 字节。断言文本出现在那个上限之后。模型从未读到那些失败的行。

这不是首先要解决的 prompt 问题。而是工具 span 上缺少 I/O 元数据。

大多数 agent 追踪只存一个简短的文本预览。它们也存一个模糊的成功标志。这两个字段无法描述一个被截断的管道。

三种 wrapper 隐藏了真实的工具结果：

Head-cap 只保留前 N 字节。

Tail-cap 只保留后 N 字节。

Pipe-stall 丢弃流并存储空文本。

然后模型基于一份不完整的文档做计划。后面的调试会话信任那份不完整的文档。团队调优 temperature 而那些字节早已不存在了。

典型 span 已经包含什么

一个典型的工具 span 已经有 name。它经常包含 arguments 和 duration。有些收集器也保留 exit code。

这些字段仍然没有覆盖被截断的管道。Exit code 0 可能伴随着被截断的 stdout。Exit code 1 可能伴随着被截断的 stderr。两条流都可能在完整管道之后为空。

每个工具 span 需要添加的字段

在每个工具 span 上记录以下键：

stdout_bytes: 任何 wrapper 限制之前的字节长度

stderr_bytes: 任何 wrapper 限制之前的字节长度

stdout_cap: wrapper 限制的字节数，或 null

stderr_cap: wrapper 限制的字节数，或 null

stdout_truncated: 限制检查后的布尔值

stderr_truncated: 限制检查后的布尔值

stdout_sha256: 未截断时的完整流哈希

stderr_sha256: 未截断时的完整流哈希

stdout_head_sha256 / stdout_tail_sha256: 截断情况

encoding: utf-8 或 binary

有意让存储的 preview 保持简短。把你拒绝保留的字节存成哈希。不要把 preview 当作完整流。

提议的 JSONL schema

下一条记录只是一个提议的 schema。不是从生产环境采样的。把它当作本地收集器的契约来使用。

{
  "span_type": "tool",
  "tool_name": "pytest",
  "cwd": "/work/repo",
  "preview_stdout": "===== test session starts =====\n",
  "preview_stderr": "",
  "stdout_bytes": 24118,
  "stderr_bytes": 902,
  "stdout_cap": 8192,
  "stderr_cap": 8192,
  "stdout_truncated": true,
  "stderr_truncated": false,
  "stdout_sha256": null,
  "stderr_sha256": "b3a1...",
  "stdout_head_sha256": "9c21...",
  "stdout_tail_sha256": "44de...",
  "encoding": "utf-8",
  "ok_preview": true
}

ok_preview 是 agent 看到的内容。它不是对完整过程的声明。在调试策略之前先 lint 这个区别。

一个本地子进程 wrapper

下面的 wrapper 是一个带标签的本地示例。在你的机器上针对一条命令运行它。它不是生产级沙箱或 jail。

#!/usr/bin/env python3
"""Record truncation metadata around one tool command."""
from __future__ import annotations

import hashlib
import json
import subprocess
import sys
from pathlib import Path

CAP = 8192  # example cap; set from your collector
PREVIEW = 512

def sha256_bytes(data: bytes) -> str:
    return hashlib.sha256(data).hexdigest()

def clip_meta(data: bytes, cap: int) -> dict:
    truncated = len(data) > cap
    kept = data[:cap] if truncated else data
    meta = {
        "bytes": len(data),
        "cap": cap,
        "truncated": truncated,
        "sha256": None if truncated else sha256_bytes(data),
        "head_sha256": sha256_bytes(data[: min(len(data), 256)]),
        "tail_sha256": sha256_bytes(data[-min(len(data), 256) :]) if data else None,
        "preview": kept[:PREVIEW].decode("utf-8", "replace"),
        "encoding": "utf-8",
    }
    return meta

def run_tool(argv: list[str], cwd: str) -> dict:
    proc = subprocess.run(
        argv,
        cwd=cwd,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        check=False,
    )
    out = clip_meta(proc.stdout, CAP)
    err = clip_meta(proc.stderr, CAP)
    return {
        "span_type": "tool",
        "argv": argv,
        "cwd": str(Path(cwd).resolve()),
        "returncode": proc.returncode,
        "stdout_bytes": out["bytes"],
        "stderr_bytes": err["bytes"],
        "stdout_cap": out["cap"],
        "stderr_cap": err["cap"],
        "stdout_truncated": out["truncated"],
        "stderr_truncated": err["truncated"],
        "stdout_sha256": out["sha256"],
        "stderr_sha256": err["sha256"],
        "stdout_head_sha256": out["head_sha256"],
        "stdout_tail_sha256": out["tail_sha256"],
        "preview_stdout": out["preview"],
        "preview_stderr": err["preview"],
        "encoding": "utf-8",
        "ok_preview": proc.returncode == 0 and not out["truncated"] and not err["truncated"],
    }

if __name__ == "__main__":
    if len(sys.argv) < 3:
        print("usage: wrap_tool.py CWD CMD [ARGS...]", file=sys.stderr)
        sys.exit(2)
    record = run_tool(sys.argv[2:], sys.argv[1])
    print(json.dumps(record, ensure_ascii=False))

python3 wrap_tool.py "$PWD" pytest -q tests/test_billing.py

把那行 JSON 重定向到 run.jsonl。每个 agent 运行保持一个文件，不要按周合并。不要把不相关的任务追加到同一个文件。

把追踪喂给这个检查器。它在标志与字节数不一致时失败。它也在 preview 在 clip 之后声称成功时失败。

#!/usr/bin/env python3
"""Lint tool spans for truncation contract breaks."""
from __future__ import annotations

import json
import sys
from pathlib import Path

REQUIRED = (
    "stdout_bytes",
    "stderr_bytes",
    "stdout_cap",
    "stderr_cap",
    "stdout_truncated",
    "stderr_truncated",
)

def problems(span: dict, line_no: int) -> list[str]:
    issues: list[str] = []
    if span.get("span_type") != "tool":
        return issues
    missing = [key for key in REQUIRED if key not in span]
    if missing:
        issues.append(f"L{line_no}: missing {missing}")
        return issues
    for stream in ("stdout", "stderr"):
        n = span[f"{stream}_bytes"]
        cap = span[f"{stream}_cap"]
        flag = span[f"{stream}_truncated"]
        if cap is not None and n > cap and flag is not True:
            issues.append(f"L{line_no}: {stream} clipped without flag")
        if cap is not None and n <= cap and flag is True:
            issues.append(f"L{line_no}: {stream} flag set under cap")
        if flag and span.get(f"{stream}_sha256"):
            issues.append(f"L{line_no}: full {stream} hash on clipped stream")
    if span.get("ok_preview") is True and (
        span["stdout_truncated"] or span["stderr_truncated"]
    ):
        issues.append(f"L{line_no}: ok_preview true while truncated")
    return issues

def lint_path(path: Path) -> int:
    found = 0
    with path.open() as handle:
        for line_no, raw in enumerate(handle, start=1):
            raw = raw.strip()
            if not raw:
                continue
            span = json.loads(raw)
            for item in problems(span, line_no):
                print(item)
                found += 1
    return found

if __name__ == "__main__":
    if len(sys.argv) != 2:
        print("usage: lint_truncation.py run.jsonl", file=sys.stderr)
        sys.exit(2)
    count = lint_path(Path(sys.argv[1]))
    print(f"issues={count}")
    sys.exit(1 if count else 0)

python3 lint_truncation.py run.jsonl

非零退出意味着停止模型调试。先在没有限制的情况下重新运行被截断的工具。把完整的 stderr 与存储的 preview 做 diff。

在事件复盘时使用这张表。不要把它变成模型评分标准。artifact 是 span 契约，不是分数。

一个可以复用的调试循环

在下一个不透明的 agent 失误上使用这个循环：

只为那一次运行导出 JSONL。

运行 lint_truncation.py run.jsonl。

如果任何 span 设置了截断标志就停止。

用移除 CAP 的方式重新运行那一条 argv。

把新的 stderr 与 preview_stderr 做 diff。

只有在管道诚实之后才检查 prompt。

不要从新的 system prompt 开始。不要换更大的模型。先修好流，再重新评估策略。

scratch model 和 server 适合放在哪里

批量 linting JSONL 不需要 GPU。一台小 VM 就可以解析追踪并重新运行工具。那是这种方法需要的唯一容量。

披露：本文是 MonkeyCode 产品推广的一部分。

MonkeyCode 提供免费模型访问和免费 server 选项。这两个就足够托管 linter 和廉价重跑了。检查器不依赖那个产品才能工作。移除产品，span 契约依然成立。

警告

• 哈希无法重建你已经丢弃的字节。
• Stdout 可能包含 token、路径和 secrets。
• 嵌套 shell 可以施加第二个静默限制。
• 二进制编译器破坏朴素 UTF-8 preview。
• duration_ms 仍然不说明截断点。
• 当日志包含时间戳时 preview diff 会失败。
• 把脱敏规则放在收集器旁边。如果策略禁止原始命令输出就先哈希。不要把完整 pytest 日志发送到共享 trace bucket。
• 如果 agent 从不调用工具，跳过这个工作流。
• 如果你的收集器已经存储完整 blob，跳过这个工作流。
• 如果策略禁止保留命令输出，跳过这个工作流。
• 交互式 TTY 工具带光标码的，跳过这个工作流。

下一步

模型无法修复它从未收到的字节。在归咎于策略之前先记录截断标志。这周在一份失败的 JSONL 文件上运行 lint_truncation.py。