Lesson 04 — 副作用与幂等：能重放的才敢自动化

# publish.py
def idempotency_key(thread_id: str, content: str) -> str:
    """thread_id + 内容指纹 → 幂等键。"""
    content_hash = hashlib.sha256(content.encode("utf-8")).hexdigest()[:12]
    thread_hash = hashlib.sha256(thread_id.encode("utf-8")).hexdigest()[:8]
    return f"{thread_hash}:{content_hash}"

CREATE TABLE publishes (
    idempotency_key TEXT PRIMARY KEY,   -- ← 去重键
    thread_id, content_hash, output_path, published_at, result_json
)

def publish_report(thread_id, content, dry_run=None) -> dict:
    key = idempotency_key(thread_id, content)
    if dry_run:
        return {"published": False, "dry_run": True, ...}   # 只打印不执行

    row = 查注册表(key)
    if row 已存在:
        return row.result + {"idempotent_replay": True}     # 幂等 no-op

    # 新发布：写 outputs/ + 注册表
    ...
    return {"published": True, "idempotent_replay": False, ...}

# graph.py（enable_publish 时）
if settings.enable_publish:
    builder.add_node("publish", make_publish_node())
    # reviewer PASS → publish → END（而非直接 END）
    def review_route_with_publish(state):
        target = review_route(state)
        return "publish" if target == END else target
    builder.add_conditional_edges("reviewer", review_route_with_publish)
    builder.add_edge("publish", END)
else:
    builder.add_conditional_edges("reviewer", review_route)  # 现状：pass→END

【before · 裸奔】（无幂等键）
  第 1 次 publish：seq=1  → 真的发布了 ⚠️
  第 2 次 publish：seq=2  → 又发布了一次 ☠️（重复发布事故）
  实际发布次数：2（应该 1 次）

【after · 开幂等键】
  第 1 次 publish：replay=False, seq=1  → 真发布
  第 2 次 publish：replay=True           → 幂等 no-op ✅
  第 3 次 publish（内容变了）：replay=False, seq=2  → 新发布
  发布历史记录数：2（去重后）

cd portfolio-projects/research-assistant
python ../../agent-ops-lessons/04_sideeffect_idempotent/code.py

cd portfolio-projects/research-assistant

# 1. 全部测试绿（171 + 14 = 185）
python -m pytest -q
# 预期：185 passed

# 2. enable_publish=off 时图与现状等价（节点集不含 publish）
python -m pytest tests/test_publish.py::test_graph_no_publish_node_when_disabled -q
# 预期：passed

# 3. 幂等：同 thread+同内容第二次 no-op
python -c "
from research_assistant import publish
publish._DB_PATH = ':memory:'  # 测试用（实际 sqlite 不支持 memory 建表，用临时文件）
import tempfile, os
publish._DB_PATH = os.path.join(tempfile.mkdtemp(), 't.db')
r1 = publish.publish_report('t1', '内容')
r2 = publish.publish_report('t1', '内容')
print('第1次 replay:', r1['idempotent_replay'], '| 第2次 replay:', r2['idempotent_replay'])
"
# 预期：第1次 replay: False | 第2次 replay: True