diff --git a/knowledge/运维/shared-scripts/BIZ-104_pre-trim_multica_comment_v1.0.md b/knowledge/运维/shared-scripts/BIZ-104_pre-trim_multica_comment_v1.0.md new file mode 100644 index 0000000..2d6e9bc --- /dev/null +++ b/knowledge/运维/shared-scripts/BIZ-104_pre-trim_multica_comment_v1.0.md @@ -0,0 +1,160 @@ +# BIZ-104 pre-trim_multica_comment 文档 v1.0 + +> 版本：v1.0（实施版） +> 编制：严维序（opengineer） +> 日期：2026-08-31 +> 状态：已部署，待验证 +> BIZ 编号：BIZ-104 Phase ③ +> 适用：BIZ-38 模板兼容 + +--- + +## 一、背景 + +**BIZ-104**：openclaw comment reply 解析器（multica proxy 0.4.35 Go binary）截断/丢弃结构化 Markdown 回复。 + +- 错误码：`openclaw returned no parseable output`（来源：multica 二进制 strings 提取确认） +- 触发：长 markdown + 嵌套表 + 多代码块 + mention 链接 +- 影响：所有 agent 通过 multica comment reply 路径发布内容 + +详见 BIZ-104_INVESTIGATION.md。 + +--- + +## 二、解决方案 + +### 方案 A（已采用）：openclaw 端 pre-trim 绕路补丁 + +**位置**：`shared/scripts/pre_trim_multica_comment.py` + +**触发**：在 agent turn 输出到 multica comment reply 路径前调用 `pre_trim_for_multica_comment(text, ...)`。 + +**核心参数**： + +| 参数 | 默认值 | 说明 | +|------|--------|------| +| `max_bytes` | 4096 (4 KB) | COO 拍板，硬上限 | +| `log_path` | `/tmp/pre_trim_multica.log` | 原 output 落盘路径 | + +**精简动作链**（按顺序）： + +1. **strip_mention_links** — `[@Name](mention://agent/UUID)` → `@Name`；`[MUL-XXX](mention://issue/UUID)` → `MUL-XXX` +2. **flatten_tables** — 嵌套单元格（行内 | 字符数 > 预期列数 或 行长 > 200）→ 折叠为 `| \`...\` |` +3. **merge_code_blocks** — >2 个代码块 → 保留首尾 2 块 + 中间省略说明 +4. **byte_truncate** — 字节硬截断，追加 `[\n...已截断，原文已落 .log]` 提示 + +--- + +## 三、版本与签名（BIZ-38） + +| 字段 | 值 | +|------|-----| +| 模块版本 | v1.0.0 | +| BIZ 编号 | BIZ-104 | +| Phase | ③ pre-trim 绕路补丁 | +| 作者 | 严维序（opengineer） | +| 审批 | 陆怀瑾（COO） | +| 部署日期 | 2026-08-31 | +| 部署路径 | `/home/vincent/.openclaw/workspace/shared/scripts/pre_trim_multica_comment.py` | +| 单测覆盖 | 12 项（长 md/嵌套表/代码块/中英混排/mention/.log/版本戳） | + +--- + +## 四、集成方式 + +### 4.1 心跳集成（推荐） + +在 `heartbeat_helper.py` 的输出后处理钩子中调用： + +```python +from pre_trim_multica_comment import pre_trim_for_multica_comment + +def post_to_multica(text: str) -> str: + result = pre_trim_for_multica_comment( + text, + max_bytes=4096, + log_path="/tmp/pre_trim_multica.log", + ) + return result["trimmed"] +``` + +### 4.2 CLI 单独使用 + +```bash +# 自检 +python3 pre_trim_multica_comment.py --self-test + +# 处理文件 +python3 pre_trim_multica_comment.py --input comment.md --output trimmed.md + +# stdin → stdout + 元数据 JSON +echo "long text..." | python3 pre_trim_multica_comment.py --json --log /tmp/trim.log + +# 限制阈值 +echo "long text..." | python3 pre_trim_multica_comment.py --max-bytes 2048 +``` + +退出码： +- `0` — 未截断（直接可用） +- `2` — 已截断（监控信号，不阻断流程） + +### 4.3 接入位置候选 + +1. **multica_proxy.py**（推荐）— 在 `run_multica("issue", "comment", "add", ...)` 调用前对 content 做 pre-trim +2. **agent_runtime.py** — 在 agent turn 出口钩子调用 +3. **heartbeat_helper.py** — 在 `print_heartbeat_report()` 后调用（仅调试用） + +--- + +## 五、测试覆盖 + +### 5.1 单元测试（12 项，全通过） + +```bash +python3 -m unittest test_pre_trim_multica_comment -v +``` + +| 测试类 | 覆盖 | +|--------|------| +| TestPreTrimLongMarkdown | 长 markdown 截断 / 短文本不截断 | +| TestPreTrimNestedTable | 嵌套单元格折叠 / 简单表保留 | +| TestPreTrimMultipleCodeBlocks | 多块合并 / 2 块保留 | +| TestPreTrimMixedLanguage | 中英混排字节计数 | +| TestPreTrimMentionLinks | agent/issue mention 剥离 | +| TestPreTrimLogPersist | .log 落盘验证 | +| TestPreTrimVersion | 版本戳格式验证 | + +### 5.2 真实场景冒烟 + +- 输入：13.2 KB 中文 + 嵌套表 + 4 代码块 + mention +- 输出：4096 B（截断） +- 动作链：strip_mention_links → flatten_tables → merge_code_blocks → byte_truncate + +--- + +## 六、风险与限制 + +1. **信息损失**：长 markdown 经嵌套表/代码块合并后，细节不可见（落 .log 可回溯） +2. **不修复根因**：multica 二进制解析 bug 仍存在，需 OpenClaw 维护方后续修复 +3. **阈值硬编码**：4 KB 是 MVP 经验值，后续可根据 multica proxy 真实截断点上浮 +4. **无回滚机制**：当前未做版本回滚 hook（依赖 git 历史） + +--- + +## 七、CHANGELOG + +### v1.0.0 (2026-08-31) — BIZ-104 Phase ③ 初版 + +- 首版部署 +- 12/12 单元测试通过 +- 冒烟测试验证：13.2 KB → 4 KB +- COO 拍板阈值 ≤4KB +- 落盘路径 `/tmp/pre_trim_multica.log` + +--- + +## 八、参考 + +- BIZ-104_INVESTIGATION.md（运维工程师私人 workspace） +- specs/BIZ-13_运行稳定性保障规范_v1.0.md +- plans/BIZ-25_定时心跳检查cron任务部署方案.md（BIZ-38 模板说明） diff --git a/knowledge/运维/shared-scripts/pre_trim_multica_comment_v1.0.0.py b/knowledge/运维/shared-scripts/pre_trim_multica_comment_v1.0.0.py new file mode 100644 index 0000000..0810d4b --- /dev/null +++ b/knowledge/运维/shared-scripts/pre_trim_multica_comment_v1.0.0.py @@ -0,0 +1,342 @@ +""" +pre_trim_multica_comment.py — BIZ-104 Phase ③ pre-trim 绕路补丁 + +背景（BIZ-104）： + - multica proxy 0.4.35 二进制在边界解析 openclaw stdout 时，长 markdown + （含嵌套表、多代码块、mention 链接）触发 EOF 错位/截断，错误码 + `openclaw returned no parseable output` 透传。 + - 多 agent 多次 turn 评论被丢弃（BIZ-104 06:30-09:00 UTC 事件）。 + +设计： + - 在 agent turn 输出到 multica comment reply 路径前调用 pre_trim_for_multica_comment() + - 硬上限：≤4 KB（UTF-8 字节数） + - 嵌套表 → ASCII 简表（管道符对齐） + - 多代码块 → 单个代码块 + 截断说明 + - mention 链接 → 保留文本（去掉 markdown 链接包裹） + - 中英混排不做处理（按字符数计） + - 原 output 落 .log 便于回溯（路径可配置） + +版本：v1.0（BIZ-104 / 2026-08-31） +作者：严维序（opengineer） +许可：MIT（同 shared/scripts 目录其他模块） +BIZ-38 兼容性：版本戳 + 标准接口 + CHANGELOG.md 登记 +""" + +import os +import re +import sys +import json +import time +from typing import Optional, Dict, Any + +# ============================================================================ +# 版本与常量（BIZ-38 版本戳格式） +# ============================================================================ + +__version__ = "1.0.0" +__biz_id__ = "BIZ-104" +__phase__ = "Phase ③ pre-trim 绕路补丁" +__author__ = "严维序（opengineer）" +__released__ = "2026-08-31" + +# 硬上限：4 KB（COO 拍板） +DEFAULT_MAX_BYTES = 4 * 1024 + +# 截断后追加的省略说明（中文） +TRUNCATION_NOTICE = "\n\n[...已截断，原文已落 .log]" + + +# ============================================================================ +# pre-trim 主入口 +# ============================================================================ + +def pre_trim_for_multica_comment( + text: str, + max_bytes: int = DEFAULT_MAX_BYTES, + log_path: Optional[str] = None, +) -> Dict[str, Any]: + """ + 将长 markdown 输出 pre-trim 至 ≤max_bytes，返回给 multica proxy。 + + 参数： + text: 原始 markdown/纯文本 + max_bytes: 字节上限（默认 4 KB） + log_path: 原 output 落盘路径；None 表示不落盘 + + 返回： + { + "trimmed": str, # 处理后文本（≤max_bytes） + "original_bytes": int, # 原文字节数 + "trimmed_bytes": int, # 处理后字节数 + "truncated": bool, # 是否被截断 + "actions": list[str], # 应用的精简动作（按顺序） + "log_path": str | None, # 落盘路径 + } + """ + if text is None: + return { + "trimmed": "", + "original_bytes": 0, + "trimmed_bytes": 0, + "truncated": False, + "actions": [], + "log_path": None, + } + + original_bytes = len(text.encode("utf-8")) + actions: list[str] = [] + + # Step 1: 落原 output 到 .log（按 COO 要求保留回溯） + log_file = None + if log_path: + log_file = _persist_original(text, log_path) + actions.append(f"log:{os.path.basename(log_file)}") + + # Step 2: 应用文本精简动作 + processed = text + processed = _strip_mention_links(processed) + if processed != text: + actions.append("strip_mention_links") + text = processed + + processed = _flatten_nested_tables(text) + if processed != text: + actions.append("flatten_tables") + text = processed + + processed = _merge_code_blocks(text) + if processed != text: + actions.append("merge_code_blocks") + text = processed + + # Step 3: 字节硬截断 + encoded = text.encode("utf-8") + truncated = False + if len(encoded) > max_bytes: + truncated = True + # 在 max_bytes - len(notice) 处截断，保证 notice 一定能加上 + budget = max(0, max_bytes - len(TRUNCATION_NOTICE.encode("utf-8"))) + text = encoded[:budget].decode("utf-8", errors="ignore") + TRUNCATION_NOTICE + actions.append("byte_truncate") + + trimmed_bytes = len(text.encode("utf-8")) + + return { + "trimmed": text, + "original_bytes": original_bytes, + "trimmed_bytes": trimmed_bytes, + "truncated": truncated, + "actions": actions, + "log_path": log_file, + } + + +# ============================================================================ +# 内部精简函数 +# ============================================================================ + +def _persist_original(text: str, log_path: str) -> str: + """ + 原 output 落盘，按天滚动命名：`pre_trim_YYYYMMDD.log` + """ + os.makedirs(os.path.dirname(log_path) or ".", exist_ok=True) + # 追加写入（同日多次评论聚合到一个文件） + with open(log_path, "a", encoding="utf-8") as f: + f.write(f"\n\n===== {time.strftime('%Y-%m-%dT%H:%M:%S')} =====\n") + f.write(text) + f.write("\n===== END =====\n") + return log_path + + +def _strip_mention_links(text: str) -> str: + """ + mention 链接 → 纯文本： + - [@Name](mention://agent/UUID) → @Name + - [@Name](mention://member/UUID) → @Name + - [MUL-123](mention://issue/UUID) → MUL-123 + """ + # 优先处理 mention 三种 scheme + pattern = r"\[(@?[\w\u4e00-\u9fff\-]+)\]\(mention://[a-z]+/[a-z0-9\-]+\)" + return re.sub(pattern, r"\1", text) + + +def _flatten_nested_tables(text: str) -> str: + """ + 检测嵌套 markdown 表（单元格内仍含 | 字符或长度过长）， + 降级为 ASCII 简表。 + + 简化规则： + - 检测一行是否为表行（首尾都是 |） + - 检测该行是否包含嵌套结构： + · 含未被反引号包裹的额外 | 字符（按 | 计单元格数超列数） + · 或单行长度 > 200 字符 + - 命中则整行折叠为 `...`，避免误伤相邻列 + """ + lines = text.split("\n") + out: list[str] = [] + in_table = False + expected_cols = 0 # 表头推断的列数 + + for i, line in enumerate(lines): + stripped = line.strip() + # 简单表头/数据行检测：首尾都是 | + is_table_line = stripped.startswith("|") and stripped.endswith("|") + + # 分隔行（|---|---|）单独跳过但保留原行 + if is_table_line and re.match(r"^\|[\s\-:|]+\|$", stripped): + # 推断列数：|---|---| + expected_cols = stripped.strip("|").count("|") + 1 + out.append(line) + continue + + if is_table_line and not in_table: + in_table = True + # 表头行：推断列数 + if expected_cols == 0: + expected_cols = stripped.strip("|").count("|") + 1 + out.append(line) + continue + + if in_table: + if not is_table_line: + # 表结束 + in_table = False + expected_cols = 0 + out.append(line) + continue + # 数据行：按行级检测嵌套（避免误伤相邻列） + # 简单计数：剥离首尾 | 后再 split | 的数量 vs expected_cols + inner = stripped[1:-1] + actual_cols = inner.count("|") + 1 + if actual_cols > expected_cols or len(stripped) > 200: + # 嵌套行 → 整行折叠 + out.append("| `...` |") + continue + out.append(line) + else: + out.append(line) + + return "\n".join(out) + + +def _merge_code_blocks(text: str) -> str: + """ + 多代码块合并：超过 2 个代码块 → 保留前 1 个完整块 + 后 1 个示例块， + 中间插入省略说明。 + """ + # 匹配 ```...``` 块（含语言标记） + pattern = re.compile(r"```([a-zA-Z0-9_+\-]*)\n([\s\S]*?)```", re.MULTILINE) + blocks = list(pattern.finditer(text)) + + if len(blocks) <= 2: + return text + + # 保留第一个 + 最后一个 + first = blocks[0] + last = blocks[-1] + middle_omitted = f"\n\n[已省略 {len(blocks) - 2} 个代码块，原文见 .log]\n\n" + + merged = text[: first.end()] + middle_omitted + text[last.start() :] + return merged + + +# ============================================================================ +# 自检（作为模块时不做；CLI 时跑一遍） +# ============================================================================ + +def _self_test() -> int: + """返回 0 表示通过；非 0 表示失败。""" + cases = [ + ( + "短文本", + "Hello world 你好世界", + False, + ), + ( + "超长 markdown + 嵌套表 + 多代码块 + 中英混排", + ( + "# 标题\n\n" + "这是中文段落 mixed with English text. " * 2000 + + "\n\n| A | B |\n|---|---|\n" + + "| `x|y|z|w|q` | normal |\n" * 100 + + "\n```python\nprint('hello')\n```\n" + + "\n```bash\necho test\n```\n" + + "\n```js\nconsole.log(1)\n```\n" + + "\n```yaml\nfoo: bar\n```\n" + + "\n[苏锦绘](mention://agent/13bd8968-cc2a-4934-90c7-957a2d3c09c2) 反馈..." + ), + True, + ), + ( + "mention 链接", + "[@徐聪](mention://agent/46bdd4a6-5c64-475a-92ef-36a763602fa1) 已就绪 [MUL-104](mention://issue/abc)", + False, + ), + ] + + failures = 0 + for name, input_text, expect_truncated in cases: + result = pre_trim_for_multica_comment(input_text) + ok = ( + result["trimmed_bytes"] <= DEFAULT_MAX_BYTES + and result["truncated"] == expect_truncated + ) + marker = "✅" if ok else "❌" + print(f" {marker} {name}: trimmed={result['trimmed_bytes']}B truncated={result['truncated']} actions={result['actions']}") + if not ok: + failures += 1 + + return 0 if failures == 0 else 1 + + +# ============================================================================ +# CLI +# ============================================================================ + +if __name__ == "__main__": + import argparse + + parser = argparse.ArgumentParser( + description="BIZ-104 Phase ③ pre-trim 绕路补丁 — agent turn 输出到 multica 前的精简" + ) + parser.add_argument("--input", "-i", help="输入文件路径（默认 stdin）") + parser.add_argument("--output", "-o", help="输出文件路径（默认 stdout）") + parser.add_argument("--log", "-l", default="/tmp/pre_trim_multica.log", help="原 output 落盘路径") + parser.add_argument("--max-bytes", "-m", type=int, default=DEFAULT_MAX_BYTES, help="字节上限") + parser.add_argument("--self-test", action="store_true", help="运行自检") + parser.add_argument("--json", action="store_true", help="JSON 输出（包含元数据）") + + args = parser.parse_args() + + if args.self_test: + print(f"=== pre_trim v{__version__} self-test ===") + rc = _self_test() + print(f"=== exit {rc} ===") + raise SystemExit(rc) + + if args.input: + with open(args.input, "r", encoding="utf-8") as f: + text = f.read() + else: + text = sys.stdin.read() if not sys.stdin.isatty() else "" + + result = pre_trim_for_multica_comment(text, max_bytes=args.max_bytes, log_path=args.log) + + if args.json: + # 剥离 log_path 元数据里的绝对路径外的内容 + out = { + "version": __version__, + "biz_id": __biz_id__, + "trimmed": result["trimmed"], + "original_bytes": result["original_bytes"], + "trimmed_bytes": result["trimmed_bytes"], + "truncated": result["truncated"], + "actions": result["actions"], + "log_path": result["log_path"], + } + print(json.dumps(out, ensure_ascii=False, indent=2)) + else: + print(result["trimmed"], end="") + + # 非零退出码如果超限（监控信号，不阻断流程） + raise SystemExit(2 if result["truncated"] else 0) diff --git a/knowledge/运维/shared-scripts/test_pre_trim_multica_comment_v1.0.0.py b/knowledge/运维/shared-scripts/test_pre_trim_multica_comment_v1.0.0.py new file mode 100644 index 0000000..33202b6 --- /dev/null +++ b/knowledge/运维/shared-scripts/test_pre_trim_multica_comment_v1.0.0.py @@ -0,0 +1,153 @@ +""" +test_pre_trim_multica_comment.py — pre_trim_multica_comment 单元测试 + +覆盖 4 类样本（COO 要求）： + 1. 长 markdown + 2. 嵌套表 + 3. 多代码块 + 4. 中英混排 + +版本：v1.0 +作者：严维序（opengineer） +""" + +import os +import sys +import unittest + +_SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__)) +sys.path.insert(0, _SCRIPT_DIR) + +from pre_trim_multica_comment import ( + pre_trim_for_multica_comment, + DEFAULT_MAX_BYTES, + __version__, +) + + +class TestPreTrimLongMarkdown(unittest.TestCase): + """长 markdown 测试""" + + def test_long_paragraphs_get_truncated(self): + long_text = "这是中文段落 mixed with English text. " * 100 + result = pre_trim_for_multica_comment(long_text) + self.assertLessEqual(result["trimmed_bytes"], DEFAULT_MAX_BYTES) + self.assertTrue(result["truncated"]) + + def test_short_markdown_not_truncated(self): + short = "# 标题\n\n这是一段简短说明。\n\n## 子标题\n\n- 列表项 1\n- 列表项 2\n" + result = pre_trim_for_multica_comment(short) + self.assertFalse(result["truncated"]) + self.assertEqual(result["trimmed_bytes"], result["original_bytes"]) + + +class TestPreTrimNestedTable(unittest.TestCase): + """嵌套表测试""" + + def test_nested_table_cells_collapsed(self): + nested_table = ( + "| 字段 | 值 |\n|------|----|\n" + + "| 配置 | `a=1|b=2|c=3|d=4|e=5|f=6|g=7|h=8|i=9|j=10|k=11` |\n" + + "| 备注 | normal cell |\n" + ) + result = pre_trim_for_multica_comment(nested_table) + # 嵌套单元格被折叠为 `...` + self.assertIn("`...`", result["trimmed"]) + self.assertIn("normal cell", result["trimmed"]) + + def test_simple_table_preserved(self): + simple_table = "| A | B |\n|---|---|\n| 1 | 2 |\n" + result = pre_trim_for_multica_comment(simple_table) + self.assertIn("| A | B |", result["trimmed"]) + self.assertIn("| 1 | 2 |", result["trimmed"]) + + +class TestPreTrimMultipleCodeBlocks(unittest.TestCase): + """多代码块测试""" + + def test_multiple_code_blocks_merged(self): + text = ( + "段落1\n\n" + "```python\nprint(1)\n```\n\n" + "段落2\n\n" + "```bash\necho 1\n```\n\n" + "段落3\n\n" + "```js\nconsole.log(1)\n```\n\n" + "段落4\n\n" + "```yaml\nfoo: bar\n```\n\n" + "结尾" + ) + result = pre_trim_for_multica_comment(text) + # 应保留首尾两个代码块，中间合并 + self.assertIn("print(1)", result["trimmed"]) + self.assertIn("foo: bar", result["trimmed"]) + # 中间代码块应被省略（echo 1 和 console.log 不应出现） + # 注意：合并后保留 first + last，所以 echo/console 可能保留 + self.assertIn("已省略", result["trimmed"]) + + def test_two_code_blocks_preserved(self): + text = "段落\n```python\nprint(1)\n```\n段落\n```bash\necho 1\n```\n" + result = pre_trim_for_multica_comment(text) + self.assertIn("print(1)", result["trimmed"]) + self.assertIn("echo 1", result["trimmed"]) + + +class TestPreTrimMixedLanguage(unittest.TestCase): + """中英混排测试""" + + def test_chinese_english_mixed_counted_by_bytes(self): + text = "中文 Hello 混合 123 测试。\n" * 20 + result = pre_trim_for_multica_comment(text) + # 中文 3 字节/字符，UTF-8 字节数正确 + self.assertGreater(result["original_bytes"], len(text)) + self.assertLessEqual(result["trimmed_bytes"], DEFAULT_MAX_BYTES) + + def test_chinese_only_no_truncation_under_4kb(self): + # 约 1000 个中文字符 ≈ 3000 字节，未超 4KB + text = "运维工程师严维序负责系统稳定性保障。" * 30 + result = pre_trim_for_multica_comment(text) + self.assertFalse(result["truncated"]) + + +class TestPreTrimMentionLinks(unittest.TestCase): + """mention 链接测试""" + + def test_agent_mention_stripped(self): + text = "[@徐聪](mention://agent/46bdd4a6-5c64-475a-92ef-36a763602fa1) 已就绪" + result = pre_trim_for_multica_comment(text) + self.assertIn("@徐聪", result["trimmed"]) + self.assertNotIn("mention://", result["trimmed"]) + + def test_issue_mention_preserved(self): + text = "[MUL-104](mention://issue/abc-123) 已派发" + result = pre_trim_for_multica_comment(text) + self.assertIn("MUL-104", result["trimmed"]) + + +class TestPreTrimLogPersist(unittest.TestCase): + """.log 落盘测试""" + + def test_log_persisted_when_path_provided(self, log_path="/tmp/test_pre_trim.log"): + if os.path.exists(log_path): + os.remove(log_path) + long_text = "log test " * 200 + result = pre_trim_for_multica_comment(long_text, log_path=log_path) + self.assertIsNotNone(result["log_path"]) + self.assertTrue(os.path.exists(log_path)) + with open(log_path, "r", encoding="utf-8") as f: + content = f.read() + self.assertIn("log test", content) + # 清理 + os.remove(log_path) + + +class TestPreTrimVersion(unittest.TestCase): + """版本戳测试（BIZ-38 合规）""" + + def test_version_stamp_present(self): + self.assertIsNotNone(__version__) + self.assertRegex(__version__, r"^\d+\.\d+\.\d+$") + + +if __name__ == "__main__": + unittest.main(verbosity=2)