编码 Agent 的 wrapper 常截断 stdout 到 8192 字节,导致模型基于不完整输出做决策。建议在每个 tool span 中记录 stdout_bytes、stdout_cap、stdout_truncated、stdout_sha256 等字段,以便回溯真实输出。
一个编码 Agent 关闭了那个不稳定的测试工单。JSONL span 显示了 pytest 和一个简短的预览。下一次人工运行时在同一个模块失败了。
wrapper 把 stdout 限制在了 8192 字节。断言文本出现在那个上限之后。模型从未读到那些失败的行。
这不是首先要解决的 prompt 问题。而是工具 span 上缺少 I/O 元数据。
大多数 agent 追踪只存一个简短的文本预览。它们也存一个模糊的成功标志。这两个字段无法描述一个被截断的管道。
三种 wrapper 隐藏了真实的工具结果:
Head-cap 只保留前 N 字节。
Tail-cap 只保留后 N 字节。
Pipe-stall 丢弃流并存储空文本。
然后模型基于一份不完整的文档做计划。后面的调试会话信任那份不完整的文档。团队调优 temperature 而那些字节早已不存在了。
一个典型的工具 span 已经有 name。它经常包含 arguments 和 duration。有些收集器也保留 exit code。
这些字段仍然没有覆盖被截断的管道。Exit code 0 可能伴随着被截断的 stdout。Exit code 1 可能伴随着被截断的 stderr。两条流都可能在完整管道之后为空。
在每个工具 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 当作完整流。
下一条记录只是一个提议的 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 是一个带标签的本地示例。在你的机器上针对一条命令运行它。它不是生产级沙箱或 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 开始。不要换更大的模型。先修好流,再重新评估策略。
批量 linting JSONL 不需要 GPU。一台小 VM 就可以解析追踪并重新运行工具。那是这种方法需要的唯一容量。
披露:本文是 MonkeyCode 产品推广的一部分。
MonkeyCode 提供免费模型访问和免费 server 选项。这两个就足够托管 linter 和廉价重跑了。检查器不依赖那个产品才能工作。移除产品,span 契约依然成立。
模型无法修复它从未收到的字节。在归咎于策略之前先记录截断标志。这周在一份失败的 JSONL 文件上运行 lint_truncation.py。