Agent 写入后的回滚与审计

Agent 写入后的回滚与审计

文章目录

• Agent 写入后的回滚与审计
• • 1. 只拦写入仍会留下不可恢复的错误
  • 2. 先说结论
  • 3. 审计策略
  • 4. 审计日志与快照
  • 5. 写入工具接入审计
  • 6. 按操作号回滚
  • 7. 评测任务与验收
  • 8. 一键演示
  • 9. 与路径守卫、双通道、CI 的衔接
  • 10. 接到真实 Agent 时的映射
  • 11. 常见故障与处理
  • 12. 核对清单
  • 13. 术语对照
  • 14. 小结
  • 15. 相关阅读
  • 16. 审计字段在工单里的最小模板
  • 17. 与 CI 的最小衔接

摘要：《按 schema 约束的 Agent 写入 Wiki》解决智能体（Agent）能不能写、写到哪里。真正上线后还有第二问：写错了怎么定点恢复、如何留证对账。本文在写入成功时同步落操作号、前后快照与追加式审计日志；按操作号回滚时校验当前哈希，拒绝重复回滚与幽灵操作号。标记包括 AUDIT_OK、ROLLBACK_OK、ROLLBACK_BLOCKED、AUDIT_EVAL_OK。标准库即可跑通。

说明：承接 按 schema 约束的 Agent 写入 Wiki。上一篇给写入面；本篇给写入后的审计链与回滚面。二者叠加，Agent 维护 wiki 才可追责、可恢复。

承接前文：

• 按 schema 约束的 Agent 写入 Wiki
• LLM Wiki 的三层结构与落地流程
• Wiki 优先与缺页回退的双通道问答
• RAG 发布验收接入 CI
• RAG 索引指针灰度切换与回滚

建议目录：

mkdir -p ~/agent-wiki-audit/{notes,scripts,configs,fixtures/{raw,wiki/{sources,entities,concepts},eval,audit/revisions,agent_jobs},images,logs}
cd ~/agent-wiki-audit

| configs/agent_audit_policy.example.json | 写入边界 + 审计路径 + 标记 |

| fixtures/audit/ops.jsonl | 追加式操作审计日志 |

| fixtures/audit/revisions/ | 每次写入的 before/after 快照 |

| scripts/audit_log.py | 记审计、校验审计链 |

| scripts/agent_write_tool.py | 写入时强制落审计 |

| scripts/rollback_op.py | 按操作号回滚 |

| scripts/eval_audit.py | 审计与回滚验收 |

| scripts/run_audit_demo.py | 一键演示 |

| notes/agent_audit_checklist.md | 核对清单 |

文中策略与核心脚本全文给出。演示仍用本地函数模拟 Agent 工具；接到真实产品时，把同一套 op_id 与哈希校验放进工具实现即可。

1. 只拦写入仍会留下不可恢复的错误

路径白名单与 schema 检查能挡住明显越权，但挡不住“合法却写错”：摘要写反、互链指错、把临时结论写成概念页。若写入时不留快照，事后只能凭聊天记录猜改前内容；若没有操作号，就无法定点回滚，只能整库还原。

图1. 无快照、无操作号、无审计链时，写错后难以定点恢复。

本篇硬规则：

每次成功写入必须生成操作号与前后快照；回滚前校验当前哈希；已回滚与不存在的操作号一律阻断。

2. 先说结论

| AGENT_WRITE_OK | 写入成功，并已记入审计（若策略要求） |

| AUDIT_OK | 审计链完整：快照文件齐全、回滚指向有效 |

| AUDIT_BLOCKED | 审计链破损，不得开放回滚入口 |

| ROLLBACK_OK | 按操作号恢复（或删除新建页）成功 |

| ROLLBACK_BLOCKED | 缺号、已回滚、哈希不符、路径越权 |

| AUDIT_EVAL_OK | 评测任务全部通过 |

| AUDIT_EVAL_BLOCKED | 评测失败，不得上线 |

| 写入 | op_id、路径、前后哈希、before/after 文件 |

| 回滚 | 目标 op_id、恢复结果、mark_rolled_back |

| 验收 | 任务集覆盖写入审计、链校验、成功回滚、重复回滚拒绝 |

四条落地判断：

3. 审计策略

保存为 configs/agent_audit_policy.example.json：

{
"raw_dir": "fixtures/raw",
"wiki_dir": "fixtures/wiki",
"index_path": "fixtures/wiki/index.md",
"log_path": "fixtures/wiki/log.md",
"schema_path": "configs/WIKI_SCHEMA.md",
"audit_ops_path": "fixtures/audit/ops.jsonl",
"audit_revisions_dir": "fixtures/audit/revisions",
"min_inbound_links": 1,
"require_source_backlink": true,
"allowed_write_prefixes": [
"fixtures/wiki/sources/",
"fixtures/wiki/entities/",
"fixtures/wiki/concepts/",
"fixtures/wiki/index.md",
"fixtures/wiki/log.md"
],
"denied_write_prefixes": [
"fixtures/raw/",
"configs/",
"scripts/",
"notes/",
"fixtures/audit/"
],
"allowed_tools": [
"read_wiki",
"write_wiki_page",
"append_wiki_log",
"lint_wiki",
"rollback_op",
"audit_query"
],
"denied_tools": [
"write_raw",
"delete_raw",
"shell_exec",
"edit_policy",
"purge_audit"
],
"min_summary_chars": 8,
"require_audit_on_write": true,
"success_tokens": {
"write_allowed": "WRITE_ALLOWED",
"write_denied": "WRITE_DENIED",
"schema_ok": "SCHEMA_OK",
"schema_blocked": "SCHEMA_BLOCKED",
"write_ok": "AGENT_WRITE_OK",
"write_blocked": "AGENT_WRITE_BLOCKED",
"audit_ok": "AUDIT_OK",
"audit_blocked": "AUDIT_BLOCKED",
"rollback_ok": "ROLLBACK_OK",
"rollback_blocked": "ROLLBACK_BLOCKED",
"eval_ok": "AUDIT_EVAL_OK",
"eval_blocked": "AUDIT_EVAL_BLOCKED",
"demo": "AGENT_AUDIT_DEMO_OK",
"lint_ok": "WIKI_LINT_OK",
"lint_blocked": "WIKI_LINT_BLOCKED",
"ingest": "WIKI_INGEST_OK"
}
}

require_audit_on_write: true 表示写入工具在落盘后必须调用审计模块。fixtures/audit/ 本身列入禁止写入前缀，防止 Agent 篡改证据。回滚与查询通过专用工具完成，不开放通用文件写。

4. 审计日志与快照

每次成功写入追加一条 JSON 行，并在 revisions/ 下落盘快照。新建页的 before_hash 为空；覆盖写则同时存 before 与 after。

保存为 scripts/audit_log.py：

#!/usr/bin/env python3
"""Append-only audit journal for Agent wiki writes."""

from __future__ import annotations

import hashlib
import json
import uuid
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

from wiki_common import ROOT

def content_hash(text: str) –> str:
return hashlib.sha256(text.encode("utf-8")).hexdigest()[:16]

def ops_path(policy: dict[str, Any]) –> Path:
return ROOT / policy["audit_ops_path"]

def revisions_dir(policy: dict[str, Any]) –> Path:
d = ROOT / policy["audit_revisions_dir"]
d.mkdir(parents=True, exist_ok=True)
return d

def new_op_id() –> str:
return datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") + "-" + uuid.uuid4().hex[:8]

def append_op(policy: dict[str, Any], record: dict[str, Any]) –> dict[str, Any]:
path = ops_path(policy)
path.parent.mkdir(parents=True, exist_ok=True)
with path.open("a", encoding="utf-8") as f:
f.write(json.dumps(record, ensure_ascii=False) + "\\n")
return record

def load_ops(policy: dict[str, Any]) –> list[dict[str, Any]]:
path = ops_path(policy)
if not path.exists():
return []
rows = []
for line in path.read_text(encoding="utf-8").splitlines():
if line.strip():
rows.append(json.loads(line))
return rows

def get_op(policy: dict[str, Any], op_id: str) –> dict[str, Any] | None:
for row in load_ops(policy):
if row.get("op_id") == op_id:
return row
return None

def save_revision(policy: dict[str, Any], op_id: str, before_text: str | None, after_text: str) –> dict[str, str]:
d = revisions_dir(policy)
paths = {}
if before_text is not None:
bp = d / f"{op_id}.before.md"
bp.write_text(before_text, encoding="utf-8")
paths["before"] = str(bp.relative_to(ROOT))
ap = d / f"{op_id}.after.md"
ap.write_text(after_text, encoding="utf-8")
paths["after"] = str(ap.relative_to(ROOT))
return paths

def record_write(
policy: dict[str, Any],
*,
rel: str,
before_text: str | None,
after_text: str,
actor: str = "agent",
tool: str = "write_wiki_page",
status: str = "written",
) –> dict[str, Any]:
op_id = new_op_id()
before_h = content_hash(before_text) if before_text is not None else None
after_h = content_hash(after_text)
rev = save_revision(policy, op_id, before_text, after_text)
record = {
"op_id": op_id,
"ts": datetime.now(timezone.utc).isoformat(),
"actor": actor,
"tool": tool,
"path": rel,
"before_hash": before_h,
"after_hash": after_h,
"status": status,
"rolled_back": False,
"revisions": rev,
}
append_op(policy, record)
return record

def validate_audit_chain(policy: dict[str, Any]) –> dict[str, Any]:
"""Basic integrity: each written op has after revision; rollback ops reference target."""
tokens = policy["success_tokens"]
ops = load_ops(policy)
issues: list[str] = []
for op in ops:
if op.get("status") == "written":
after = op.get("revisions", {}).get("after")
if not after or not (ROOT / after).exists():
issues.append(f"MISSING_AFTER {op.get('op_id')}")
if op.get("before_hash") is not None:
before = op.get("revisions", {}).get("before")
if not before or not (ROOT / before).exists():
issues.append(f"MISSING_BEFORE {op.get('op_id')}")
if op.get("status") == "rollback":
target = op.get("target_op_id")
if not target or get_op(policy, target) is None:
issues.append(f"BAD_ROLLBACK_TARGET {op.get('op_id')}")
ok = not issues
return {
"ok": ok,
"token": tokens["audit_ok"] if ok else tokens["audit_blocked"],
"op_count": len(ops),
"issues": issues,
}

图2. 路径与 schema 通过后，写页、存快照、追加操作日志。

字段说明：

• 操作号：op_id，回滚与对账的主键；
• 前后哈希：before_hash / after_hash，用于完整性检查；
• 状态：written / rollback / mark_rolled_back；
• 快照路径：revisions 指向 before/after 文件。

5. 写入工具接入审计

在上一篇的路径与 schema 守卫之后，落盘成功就调用 record_write。这样 Agent 无法“只写页不记账”，除非关掉策略开关（生产环境不要关）。

保存为 scripts/agent_write_tool.py：

#!/usr/bin/env python3
"""Agent wiki write tools with path/schema guards and audit snapshots."""

from __future__ import annotations

import argparse
import json
import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent))

from audit_log import record_write # noqa: E402
from path_guard import check_tool, check_write_path, normalize_rel # noqa: E402
from schema_check import check_page_schema # noqa: E402
from wiki_common import ROOT, append_log, load_policy # noqa: E402

def tool_read_wiki(policy: dict, path: str) –> dict:
ok, token = check_tool(policy, "read_wiki")
if not ok:
return {"ok": False, "token": token, "reason": "tool_denied"}
try:
rel = normalize_rel(path)
except ValueError as exc:
return {"ok": False, "token": policy["success_tokens"]["write_denied"], "reason": str(exc)}
if not rel.startswith("fixtures/wiki/"):
return {
"ok": False,
"token": policy["success_tokens"]["write_denied"],
"reason": "read_outside_wiki",
"rel": rel,
}
fp = ROOT / rel
if not fp.exists():
return {"ok": False, "token": policy["success_tokens"]["write_denied"], "reason": "missing", "rel": rel}
return {
"ok": True,
"token": policy["success_tokens"]["write_allowed"],
"rel": rel,
"text": fp.read_text(encoding="utf-8"),
}

def tool_write_wiki_page(
policy: dict,
path: str,
content: str,
dry_run: bool = False,
actor: str = "agent",
) –> dict:
ok, token = check_tool(policy, "write_wiki_page")
if not ok:
return {"ok": False, "token": token, "stage": "tool", "reason": "tool_denied"}

path_result = check_write_path(policy, path)
if not path_result["ok"]:
return {
"ok": False,
"token": path_result["token"],
"stage": "path",
"reason": path_result["reason"],
"rel": path_result["rel"],
}

schema = check_page_schema(policy, path_result["rel"], content)
if not schema["ok"]:
return {
"ok": False,
"token": policy["success_tokens"]["write_blocked"],
"stage": "schema",
"path_token": path_result["token"],
"schema_token": schema["token"],
"issues": schema["issues"],
"rel": path_result["rel"],
}

rel = path_result["rel"]
fp = ROOT / rel
before_text = fp.read_text(encoding="utf-8") if fp.exists() else None

if dry_run:
return {
"ok": True,
"token": policy["success_tokens"]["write_ok"],
"stage": "dry_run",
"rel": rel,
"schema_token": schema["token"],
"would_audit": bool(policy.get("require_audit_on_write", True)),
}

fp.parent.mkdir(parents=True, exist_ok=True)
fp.write_text(content, encoding="utf-8")

audit = None
if policy.get("require_audit_on_write", True):
audit = record_write(
policy,
rel=rel,
before_text=before_text,
after_text=content,
actor=actor,
tool="write_wiki_page",
status="written",
)

append_log(
policy,
"agent_write",
f"{rel} | {policy['success_tokens']['write_ok']} | op={audit['op_id'] if audit else '-'}",
)
return {
"ok": True,
"token": policy["success_tokens"]["write_ok"],
"stage": "written",
"rel": rel,
"schema_token": schema["token"],
"op_id": audit["op_id"] if audit else None,
"audit": audit,
}

def main() –> int:
ap = argparse.ArgumentParser(description="Agent wiki write with audit")
ap.add_argument("tool", choices=["read_wiki", "write_wiki_page", "append_wiki_log", "write_raw"])
ap.add_argument("–path", default="")
ap.add_argument("–content-file", default="")
ap.add_argument("–content", default="")
ap.add_argument("–detail", default="")
ap.add_argument("–actor", default="agent")
ap.add_argument("–dry-run", action="store_true")
args = ap.parse_args()
policy = load_policy()

if args.tool == "write_raw":
ok, token = check_tool(policy, "write_raw")
print(token)
print(json.dumps({"ok": ok, "token": token, "reason": "tool_denied"}, ensure_ascii=False))
return 1

if args.tool == "read_wiki":
result = tool_read_wiki(policy, args.path)
elif args.tool == "append_wiki_log":
ok, token = check_tool(policy, "append_wiki_log")
if not ok:
result = {"ok": False, "token": token}
else:
append_log(policy, "agent_note", args.detail or "agent note")
result = {"ok": True, "token": policy["success_tokens"]["write_ok"]}
else:
content = args.content
if args.content_file:
content = Path(args.content_file).read_text(encoding="utf-8")
result = tool_write_wiki_page(
policy, args.path, content, dry_run=args.dry_run, actor=args.actor
)

print(result.get("token", policy["success_tokens"]["write_blocked"]))
printable = {k: v for k, v in result.items() if k not in ("text", "audit")}
if "audit" in result and result["audit"]:
printable["op_id"] = result["audit"]["op_id"]
printable["before_hash"] = result["audit"]["before_hash"]
printable["after_hash"] = result["audit"]["after_hash"]
print(json.dumps(printable, ensure_ascii=False, indent=2))
return 0 if result.get("ok") else 1

if __name__ == "__main__":
raise SystemExit(main())

命令行示例：

python3 scripts/agent_write_tool.py write_wiki_page \\
–path fixtures/wiki/concepts/audit_demo.md \\
–content-file fixtures/agent_jobs/good_concept.md

成功时应同时看到 AGENT_WRITE_OK 与返回里的 op_id。

6. 按操作号回滚

回滚不是“再让模型改一版”，而是机械恢复快照：

保存为 scripts/rollback_op.py：

#!/usr/bin/env python3
"""Rollback a prior Agent write by op_id."""

from __future__ import annotations

import argparse
import json
import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent))

from audit_log import ( # noqa: E402
append_op,
content_hash,
get_op,
new_op_id,
save_revision,
)
from path_guard import check_tool, check_write_path # noqa: E402
from wiki_common import ROOT, append_log, load_policy # noqa: E402
from datetime import datetime, timezone

def rollback_op(policy: dict, op_id: str, dry_run: bool = False) –> dict:
tokens = policy["success_tokens"]
ok, token = check_tool(policy, "rollback_op")
if not ok:
return {"ok": False, "token": token, "stage": "tool", "reason": "tool_denied"}

target = get_op(policy, op_id)
if target is None:
return {
"ok": False,
"token": tokens["rollback_blocked"],
"stage": "lookup",
"reason": "op_not_found",
"op_id": op_id,
}
if target.get("status") != "written":
return {
"ok": False,
"token": tokens["rollback_blocked"],
"stage": "lookup",
"reason": "not_a_write_op",
"op_id": op_id,
"status": target.get("status"),
}
if target.get("rolled_back"):
return {
"ok": False,
"token": tokens["rollback_blocked"],
"stage": "lookup",
"reason": "already_rolled_back",
"op_id": op_id,
}

# append-only journal: rolled_back flag lives on mark ops
from audit_log import load_ops

for op in load_ops(policy):
if op.get("status") == "mark_rolled_back" and op.get("target_op_id") == op_id:
return {
"ok": False,
"token": tokens["rollback_blocked"],
"stage": "lookup",
"reason": "already_rolled_back",
"op_id": op_id,
}

rel = target["path"]
path_result = check_write_path(policy, rel)
if not path_result["ok"]:
return {
"ok": False,
"token": tokens["rollback_blocked"],
"stage": "path",
"reason": path_result["reason"],
"rel": rel,
}

before_rel = target.get("revisions", {}).get("before")
if before_rel is None:
# original write created a new file — rollback deletes it
restore_text = None
else:
restore_text = (ROOT / before_rel).read_text(encoding="utf-8")

fp = ROOT / rel
current = fp.read_text(encoding="utf-8") if fp.exists() else None
if current is not None and content_hash(current) != target.get("after_hash"):
return {
"ok": False,
"token": tokens["rollback_blocked"],
"stage": "hash",
"reason": "current_mismatch",
"expected_after": target.get("after_hash"),
"current_hash": content_hash(current) if current is not None else None,
}

if dry_run:
return {
"ok": True,
"token": tokens["rollback_ok"],
"stage": "dry_run",
"op_id": op_id,
"rel": rel,
"restore_mode": "delete" if restore_text is None else "restore",
}

rb_id = new_op_id()
after_snapshot = current or ""
if restore_text is None:
if fp.exists():
fp.unlink()
restored = ""
else:
fp.parent.mkdir(parents=True, exist_ok=True)
fp.write_text(restore_text, encoding="utf-8")
restored = restore_text

rev = save_revision(policy, rb_id, after_snapshot, restored)
# mark original as rolled back by appending a marker op + rewriting not possible on append-only;
# we append rollback op and a side marker file list via status field; also append rolled_back note
record = {
"op_id": rb_id,
"ts": datetime.now(timezone.utc).isoformat(),
"actor": "operator",
"tool": "rollback_op",
"path": rel,
"target_op_id": op_id,
"before_hash": content_hash(after_snapshot) if after_snapshot else None,
"after_hash": content_hash(restored) if restore_text is not None else None,
"status": "rollback",
"rolled_back": False,
"revisions": rev,
}
append_op(policy, record)
# append a lightweight flag op so queries can see target rolled back
append_op(
policy,
{
"op_id": new_op_id(),
"ts": datetime.now(timezone.utc).isoformat(),
"actor": "system",
"tool": "audit_mark",
"path": rel,
"target_op_id": op_id,
"status": "mark_rolled_back",
"rolled_back": True,
},
)
append_log(policy, "rollback", f"{op_id} -> {rel} | {tokens['rollback_ok']}")
return {
"ok": True,
"token": tokens["rollback_ok"],
"stage": "restored",
"op_id": op_id,
"rollback_op_id": rb_id,
"rel": rel,
"restore_mode": "delete" if restore_text is None else "restore",
}

def is_rolled_back(policy: dict, op_id: str) –> bool:
for op in __import__("audit_log", fromlist=["load_ops"]).load_ops(policy):
if op.get("status") == "mark_rolled_back" and op.get("target_op_id") == op_id:
return True
if op.get("op_id") == op_id and op.get("rolled_back"):
return True
return False

def main() –> int:
ap = argparse.ArgumentParser(description="Rollback Agent write by op_id")
ap.add_argument("op_id")
ap.add_argument("–dry-run", action="store_true")
args = ap.parse_args()
policy = load_policy()
result = rollback_op(policy, args.op_id, dry_run=args.dry_run)
print(result["token"])
print(json.dumps(result, ensure_ascii=False, indent=2))
return 0 if result.get("ok") else 1

if __name__ == "__main__":
raise SystemExit(main())

图3. 查号 → 核哈希 → 恢复/删除 → 记回滚标记。

图4. 审计通过、回滚成功、回滚阻断三类结局。

python3 scripts/rollback_op.py <op_id>
python3 scripts/rollback_op.py <op_id> # 第二次应 ROLLBACK_BLOCKED

7. 评测任务与验收

fixtures/eval/audit_jobs.jsonl：

{"id": "a1_write_audit", "kind": "write_and_audit", "path": "fixtures/wiki/concepts/audit_eval_page.md", "content_file": "fixtures/agent_jobs/good_concept.md", "expect_token": "AGENT_WRITE_OK"}
{"id": "a2_chain_ok", "kind": "audit_chain", "expect_token": "AUDIT_OK"}
{"id": "a3_rollback_ok", "kind": "rollback", "dry_run": false, "expect_token": "ROLLBACK_OK"}
{"id": "a4_second_rollback_blocked", "kind": "second_rollback_blocked", "expect_token": "ROLLBACK_BLOCKED"}
{"id": "a5_missing_op", "kind": "rollback_missing", "op_id": "no-such-op", "expect_token": "ROLLBACK_BLOCKED"}

覆盖五类断言：

验收脚本 scripts/eval_audit.py：

#!/usr/bin/env python3
"""Evaluate audit + rollback jobs."""

from __future__ import annotations

import argparse
import json
import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent))

from agent_write_tool import tool_write_wiki_page # noqa: E402
from audit_log import get_op, load_ops, validate_audit_chain # noqa: E402
from rollback_op import is_rolled_back, rollback_op # noqa: E402
from wiki_common import ROOT, load_policy # noqa: E402

def main() –> int:
ap = argparse.ArgumentParser(description="Evaluate audit/rollback jobs")
ap.add_argument("–jobs", default=str(ROOT / "fixtures/eval/audit_jobs.jsonl"))
args = ap.parse_args()
policy = load_policy()
jobs = [
json.loads(line)
for line in Path(args.jobs).read_text(encoding="utf-8").splitlines()
if line.strip()
]

# shared state across jobs in one eval run
last_op_id = None
results = []

for job in jobs:
jid = job["id"]
kind = job["kind"]
expect = job["expect_token"]
got = None
passed = False
detail = {}

if kind == "write_and_audit":
content = (ROOT / job["content_file"]).read_text(encoding="utf-8")
r = tool_write_wiki_page(policy, job["path"], content, dry_run=False, actor="eval")
got = r["token"]
last_op_id = r.get("op_id")
op = get_op(policy, last_op_id) if last_op_id else None
chain = validate_audit_chain(policy)
passed = (
r.get("ok")
and got == expect
and op is not None
and op.get("after_hash")
and chain["ok"]
)
detail = {"op_id": last_op_id, "chain": chain["token"]}

elif kind == "audit_chain":
chain = validate_audit_chain(policy)
got = chain["token"]
passed = got == expect
detail = {"issues": chain["issues"], "op_count": chain["op_count"]}

elif kind == "rollback":
target = job.get("op_id") or last_op_id
r = rollback_op(policy, target, dry_run=bool(job.get("dry_run", False)))
got = r["token"]
passed = got == expect and (r.get("ok") if expect == "ROLLBACK_OK" else not r.get("ok"))
if expect == "ROLLBACK_OK" and r.get("ok") and not job.get("dry_run"):
passed = passed and is_rolled_back(policy, target)
detail = {"target": target, "stage": r.get("stage"), "reason": r.get("reason")}

elif kind == "rollback_missing":
r = rollback_op(policy, job["op_id"], dry_run=True)
got = r["token"]
passed = got == expect
detail = {"reason": r.get("reason")}

elif kind == "second_rollback_blocked":
target = job.get("op_id") or last_op_id
r = rollback_op(policy, target, dry_run=False)
got = r["token"]
passed = got == expect
detail = {"reason": r.get("reason")}

else:
got = "UNKNOWN"
passed = False

results.append(
{
"id": jid,
"passed": passed,
"expect": expect,
"got": got,
"detail": detail,
}
)

failed = [r for r in results if not r["passed"]]
token = (
policy["success_tokens"]["eval_ok"]
if not failed
else policy["success_tokens"]["eval_blocked"]
)
print(token)
print(f"jobs={len(results)} failed={len(failed)} ops={len(load_ops(policy))}")
for r in results:
mark = "PASS" if r["passed"] else "FAIL"
print(f"- {mark} {r['id']} expect={r['expect']} got={r['got']} detail={r['detail']}")
return 0 if not failed else 1

if __name__ == "__main__":
raise SystemExit(main())

图5. 写入审计、链校验、回滚与拒绝路径都要进任务集。

python3 scripts/eval_audit.py

通过时应看到 AUDIT_EVAL_OK。

8. 一键演示

scripts/run_audit_demo.py：

#!/usr/bin/env python3
"""Demo: write with audit → query → rollback → lint → eval."""

from __future__ import annotations

import json
import subprocess
import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent))

from agent_write_tool import tool_write_wiki_page # noqa: E402
from audit_log import load_ops, validate_audit_chain # noqa: E402
from rollback_op import rollback_op # noqa: E402
from wiki_common import ROOT, load_policy # noqa: E402

PAGE_V1 = """# 审计演示页

审计演示页用于验证写入快照与回滚。

来源：[[sources/rag_first]]

相关：[[entities/rag]] [[concepts/hit_rate]]
"""

PAGE_V2 = """# 审计演示页

审计演示页第二版：故意写入可回滚的错误摘要。

来源：[[sources/rag_first]]

相关：[[entities/rag]] [[concepts/hit_rate]]

## 错误说明

本段用于演示回滚，正式环境不应保留。
"""

def run(cmd: list[str]) –> str:
p = subprocess.run(cmd, cwd=ROOT, capture_output=True, text=True, check=False)
out = (p.stdout or "") + (p.stderr or "")
print(out.rstrip())
return out

def main() –> int:
policy = load_policy()
# clean audit journal for deterministic demo
ops = ROOT / policy["audit_ops_path"]
rev = ROOT / policy["audit_revisions_dir"]
if ops.exists():
ops.unlink()
if rev.exists():
for f in rev.glob("*"):
f.unlink()

print("=== 1) ingest ===")
run([sys.executable, "scripts/ingest_source.py", "–all"])

print("\\n=== 2) first write (create page) ===")
r1 = tool_write_wiki_page(
policy,
"fixtures/wiki/concepts/audit_demo.md",
PAGE_V1,
dry_run=False,
actor="agent",
)
print(r1["token"], "op_id=", r1.get("op_id"))
print(json.dumps({k: v for k, v in r1.items() if k != "audit"}, ensure_ascii=False))

print("\\n=== 3) second write (overwrite) ===")
r2 = tool_write_wiki_page(
policy,
"fixtures/wiki/concepts/audit_demo.md",
PAGE_V2,
dry_run=False,
actor="agent",
)
print(r2["token"], "op_id=", r2.get("op_id"))
op2 = r2.get("op_id")

print("\\n=== 4) audit chain ===")
chain = validate_audit_chain(policy)
print(chain["token"], f"ops={chain['op_count']}")
for op in load_ops(policy):
print(
f"- {op['op_id']} status={op['status']} path={op.get('path')} "
f"before={op.get('before_hash')} after={op.get('after_hash')}"
)

print("\\n=== 5) rollback second write ===")
rb = rollback_op(policy, op2, dry_run=False)
print(rb["token"])
print(json.dumps(rb, ensure_ascii=False, indent=2))
text = (ROOT / "fixtures/wiki/concepts/audit_demo.md").read_text(encoding="utf-8")
print("restored_has_v1_marker=", "第二版" not in text and "用于验证写入快照" in text)

print("\\n=== 6) backlink + index + lint ===")
rag = (ROOT / "fixtures/wiki/entities/rag.md").read_text(encoding="utf-8")
if "[[concepts/audit_demo]]" not in rag:
rag = rag.rstrip() + "\\n\\n相关概念：[[concepts/audit_demo]]\\n"
tool_write_wiki_page(policy, "fixtures/wiki/entities/rag.md", rag, dry_run=False)
run([sys.executable, "scripts/update_index.py"])
run([sys.executable, "scripts/lint_wiki.py"])

print("\\n=== 7) eval ===")
# reset journal so eval jobs are deterministic and independent of demo ops
if ops.exists():
ops.unlink()
if rev.exists():
for f in rev.glob("*"):
f.unlink()
rc = subprocess.run(
[sys.executable, "scripts/eval_audit.py"],
cwd=ROOT,
capture_output=True,
text=True,
check=False,
)
print((rc.stdout or "") + (rc.stderr or ""))
if rc.returncode != 0:
print("EVAL_FAILED")
return 1

print("\\n" + policy["success_tokens"]["demo"])
return 0

if __name__ == "__main__":
raise SystemExit(main())

python3 scripts/run_audit_demo.py

完整输出：

=== 1) ingest ===
WIKI_INGEST_OK
– fixtures/raw/ci_release.md -> sources/ci_release.md related=4
– fixtures/raw/pointer_canary.md -> sources/pointer_canary.md related=5
– fixtures/raw/rag_first.md -> sources/rag_first.md related=4

=== 2) first write (create page) ===
AGENT_WRITE_OK op_id= 20260916T012614Z-d1bfe79e
{"ok": true, "token": "AGENT_WRITE_OK", "stage": "written", "rel": "fixtures/wiki/concepts/audit_demo.md", "schema_token": "SCHEMA_OK", "op_id": "20260916T012614Z-d1bfe79e"}

=== 3) second write (overwrite) ===
AGENT_WRITE_OK op_id= 20260916T012614Z-0a73fe57

=== 4) audit chain ===
AUDIT_OK ops=2
– 20260916T012614Z-d1bfe79e status=written path=fixtures/wiki/concepts/audit_demo.md before=09438d6cec1eaa2b after=09438d6cec1eaa2b
– 20260916T012614Z-0a73fe57 status=written path=fixtures/wiki/concepts/audit_demo.md before=09438d6cec1eaa2b after=d55ceb86e0db22f5

=== 5) rollback second write ===
ROLLBACK_OK
{
"ok": true,
"token": "ROLLBACK_OK",
"stage": "restored",
"op_id": "20260916T012614Z-0a73fe57",
"rollback_op_id": "20260916T012614Z-381d96fb",
"rel": "fixtures/wiki/concepts/audit_demo.md",
"restore_mode": "restore"
}
restored_has_v1_marker= True

=== 6) backlink + index + lint ===
INDEX_OK fixtures/wiki/index.md
WIKI_LINT_OK
pages=17 issues=0

=== 7) eval ===
AUDIT_EVAL_OK
jobs=5 failed=0 ops=3
– PASS a1_write_audit expect=AGENT_WRITE_OK got=AGENT_WRITE_OK detail={'op_id': '20260916T012614Z-106e6acd', 'chain': 'AUDIT_OK'}
– PASS a2_chain_ok expect=AUDIT_OK got=AUDIT_OK detail={'issues': [], 'op_count': 1}
– PASS a3_rollback_ok expect=ROLLBACK_OK got=ROLLBACK_OK detail={'target': '20260916T012614Z-106e6acd', 'stage': 'restored', 'reason': None}
– PASS a4_second_rollback_blocked expect=ROLLBACK_BLOCKED got=ROLLBACK_BLOCKED detail={'reason': 'already_rolled_back'}
– PASS a5_missing_op expect=ROLLBACK_BLOCKED got=ROLLBACK_BLOCKED detail={'reason': 'op_not_found'}

AGENT_AUDIT_DEMO_OK

演示顺序：摄入 → 首次写入 → 覆盖写入 → 审计链 → 回滚第二版 → 巡检 → 独立评测。回滚后页面应回到第一版摘要，且评测输出 AUDIT_EVAL_OK。

9. 与路径守卫、双通道、CI 的衔接

| 写入边界 | 回滚前仍走同一允许前缀 | 《按 schema 约束的 Agent 写入 Wiki》 |

| 问答路由 | 回滚不替代拒答；错页先回滚再答 | 《Wiki 优先与缺页回退的双通道问答》 |

| 发布流水线 | AUDIT_EVAL_OK 可进 CI | 《RAG 发布验收接入 CI》 |

| 索引回滚 | 概念不同：这里回滚 wiki 页，不是索引指针 | 《RAG 索引指针灰度切换与回滚》 |

不要把本篇回滚理解成索引包回滚。索引指针灰度切换处理的是检索侧包版本；本篇处理的是 wiki 页面内容版本。两套操作号与审计日志建议分仓存放，避免混查。

10. 接到真实 Agent 时的映射

| 写入返回 op_id | 工具结果里强制回传，并写入会话日志 |

| rollback_op | 仅运维角色或需二次确认的工具 |

| audit_query | 只读查询最近 N 条操作 |

| 禁止 purge_audit | 删除审计走人工变更单，不进 Agent 工具表 |

若模型请求“清掉审计重来”，直接 WRITE_DENIED / 工具不存在。需要压缩历史时，导出冷存储后由人工归档，而不是在线覆写 ops.jsonl。

11. 常见故障与处理

| 有写入无 op_id | 未开 require_audit_on_write | 打开开关；写入与审计同路径发布 |

| AUDIT_BLOCKED | 快照文件丢失 | 禁止手工删 revisions/；从备份恢复 |

| current_mismatch | 回滚前又有人改过该页 | 先查后续 op_id，按最新错误操作回滚，或人工合并 |

| 重复回滚仍成功 | 未写 mark_rolled_back | 升级回滚脚本；补评测任务 a4 |

| 回滚后 ORPHAN | 回链页被一并删掉 | 回滚范围只含目标页；回链变更单独记操作号 |

| 与摄入冲突 | ingest 与 Agent 同页并发 | 加页级锁或分时段；冲突记矛盾记录 |

| 审计被 Agent 改写 | fixtures/audit 未禁写 | 加入 denied_write_prefixes |

12. 核对清单

• 每次成功写入都生成 op_id，并落 before/after 快照
• 审计日志只追加，不提供 purge_audit 给 Agent
• 回滚前校验当前内容哈希等于写入时的 after_hash
• 已回滚的 op_id 再次回滚必须 ROLLBACK_BLOCKED
• 不存在的 op_id 必须 ROLLBACK_BLOCKED
• 审计链校验通过才输出 AUDIT_OK
• 回滚后刷新索引并跑巡检
• 评测任务覆盖：写入审计、链完整、回滚成功、重复回滚拒绝
• 看到 AUDIT_EVAL_OK 才允许开放线上回滚入口
• 原始资料层变更仍不走 Agent 回滚通道

13. 术语对照

| 操作号 | op_id | 单次写入主键 |

| 改前哈希 | before_hash | 覆盖写前内容指纹；新建为空 |

| 改后哈希 | after_hash | 写入后内容指纹 |

| 审计通过 | AUDIT_OK | 快照与引用完整 |

| 回滚成功 | ROLLBACK_OK | 已恢复或删除新建页 |

| 回滚阻断 | ROLLBACK_BLOCKED | 缺号/已回滚/哈希不符等 |

| 已回滚标记 | mark_rolled_back | 追加式标记，不改历史行 |

| 任务集通过 | AUDIT_EVAL_OK | 评测全过 |

14. 小结

Agent 写入 wiki 要同时具备三层能力：

上文已给出策略、审计模块、写入接入、回滚、任务集、演示与清单。与上一篇写入面合并后，再把 AUDIT_EVAL_OK 接进 CI，Agent 维护链路才算可上线。

15. 相关阅读

• 按 schema 约束的 Agent 写入 Wiki
• LLM Wiki 的三层结构与落地流程
• Wiki 优先与缺页回退的双通道问答
• RAG 发布验收接入 CI
• RAG 索引指针灰度切换与回滚

《按 schema 约束的 Agent 写入 Wiki》解决谁可以改；《Agent 写入后的回滚与审计》解决改错之后如何留证与恢复。两边对齐后，本地 Wiki 维护才既受控又可追责。

如果本篇对你有帮助，欢迎点赞、收藏，也欢迎关注后续更新。

16. 审计字段在工单里的最小模板

线上发现错页时，工单至少粘贴这些字段，避免口头描述无法回滚：

path: fixtures/wiki/concepts/xxx.md
bad_op_id: 20260916T012614Z-0a73fe57
after_hash: d55ceb86e0db22f5
action: rollback_op
expect: ROLLBACK_OK

回滚完成后，把 rollback_op_id 与巡检结果 WIKI_LINT_OK 回写工单。若出现 current_mismatch，说明回滚窗口内又有新写入，应先列出该 path 的后续操作号再决定顺序。

17. 与 CI 的最小衔接

写入或回滚脚本变更时，CI 建议固定：

不要把“模型说已经回滚”当作合并条件。合并条件只认标记与任务集。