aa82445ec7
- example_candidate(七态状态机，DB 级评测候选禁接受硬拒 05§8.4) + example_user_decision(append-only+理据) - write_canonical.py：accept/discard 单事务(写正文块CAS revision+1→来源归因落块→命令幂等→决策归档→翻候选态)， 任一失败整体回滚；dry-run + eval/draft 守卫已验证；复用 db skill DSN - confirm/SKILL.md 从 git commit 语义改为写库语义(正文走 write_canonical；知识卡走双轨；规划暂 git)
206 lines
12 KiB
Python
206 lines
12 KiB
Python
#!/usr/bin/env python3
|
||
"""正文 Canonical 写入层 —— 候选接受/丢弃的唯一 DB 写路径（Shadow→Canonical）。
|
||
|
||
落库设计 §2.9 的单事务序列（任一失败整体回滚，绝不留"无来源指针的正式正文"）：
|
||
|
||
accept preflight（主会话先调 check_writer_acceptance 纯函数做生产模式全检）
|
||
-> 读候选 + 硬校验（run_type=production、state=passed，DB 级兜底）
|
||
-> 写 muse_content_block（content_text，revision+1，CAS 乐观锁） [拿到 block_id]
|
||
-> 写 muse_content_block_source_attribution（block_id, revision, lineage_payload=候选身份；来源权威落块，架构-02 §3）
|
||
-> 写 muse_content_command_log（command_id 幂等审计）
|
||
-> 写 example_user_decision（canonical_block_id, command_id, 理据）
|
||
-> UPDATE example_candidate SET state='accepted'
|
||
|
||
只由 confirm 入口调用；主会话与智能体不得自行拼接这些写。
|
||
--dry-run 试跑：完整走一遍事务再回滚，校验通过但不落库。
|
||
"""
|
||
import json
|
||
import sys
|
||
from pathlib import Path
|
||
|
||
# 复用 db skill 锁死的 DSN（.claude/skills/db/scripts），不另硬编码连接串
|
||
DB_SCRIPTS = Path(__file__).resolve().parents[2] / "db" / "scripts"
|
||
sys.path.insert(0, str(DB_SCRIPTS))
|
||
from db import connect # noqa: E402
|
||
|
||
CREATOR = "confirm"
|
||
|
||
|
||
class ConflictError(Exception):
|
||
"""接受/丢弃被拒（状态不对、revision 冲突、候选不存在等）——失败关闭，不静默。"""
|
||
|
||
|
||
def accept(candidate_id, decided_by="1", rationale=None, basis_ref=None,
|
||
expected_revision=None, command_id=None, source_type="ai_candidate",
|
||
dry_run=False):
|
||
"""接受候选为正式正文。单事务；返回 {status, block_id, revision, decision_id, ...}。"""
|
||
with connect() as conn:
|
||
try:
|
||
# 0) command_id 幂等：重放同一命令直接返回，不重复写
|
||
if command_id and conn.execute(
|
||
"SELECT 1 FROM muse_content_command_log WHERE command_id=%s AND deleted=false",
|
||
(command_id,)).fetchone():
|
||
return {"status": "already_applied", "command_id": command_id}
|
||
# 1) 读候选 + DB 级硬校验（preflight 的兜底，不靠调用方自觉）
|
||
row = conn.execute(
|
||
"SELECT id, work_id, target_chapter, run_id, attempt, run_type, candidate_version, "
|
||
"candidate_sha256, candidate_body, mode, source_role, state FROM example_candidate "
|
||
"WHERE id=%s AND deleted=false", (candidate_id,)).fetchone()
|
||
if not row:
|
||
raise ConflictError(f"候选 {candidate_id} 不存在")
|
||
cid, work_id, chap, run_id, attempt, run_type, cver, csha, body, mode, source_role, state = row
|
||
if run_type != "production":
|
||
raise ConflictError(f"run_type={run_type} 为评测/诊断候选，不得接受（05 §8.4）")
|
||
if state != "passed":
|
||
raise ConflictError(f"候选 state={state}，只有 passed 可接受")
|
||
if not body:
|
||
raise ConflictError("候选正文为空，不能接受")
|
||
# 2) 定位章 → 现有正文块（一章一块）；revision CAS 乐观锁
|
||
ch = conn.execute(
|
||
"SELECT id, title FROM muse_content_chapter WHERE work_id=%s AND order_no=%s AND deleted=false",
|
||
(work_id, chap)).fetchone()
|
||
if not ch:
|
||
raise ConflictError(f"作品 {work_id} 无第 {chap} 章，无法写正文")
|
||
chapter_id, chapter_title = ch
|
||
blk = conn.execute(
|
||
"SELECT id, revision FROM muse_content_block WHERE chapter_id=%s AND deleted=false "
|
||
"ORDER BY revision DESC LIMIT 1", (chapter_id,)).fetchone()
|
||
word_count = len(body)
|
||
if blk:
|
||
block_id, cur_rev = blk
|
||
if expected_revision is not None and int(expected_revision) != cur_rev:
|
||
raise ConflictError(f"REVISION_CONFLICT：期望 {expected_revision}，正文块当前 {cur_rev}")
|
||
new_rev = cur_rev + 1
|
||
n = conn.execute(
|
||
"UPDATE muse_content_block SET content_text=%s, revision=%s, word_count=%s, "
|
||
"command_id=%s, updater=%s WHERE id=%s AND revision=%s",
|
||
(body, new_rev, word_count, command_id, decided_by, block_id, cur_rev)).rowcount
|
||
if n != 1:
|
||
raise ConflictError("REVISION_CONFLICT：正文块被并发修改")
|
||
else:
|
||
if expected_revision not in (None, 0):
|
||
raise ConflictError(f"REVISION_CONFLICT：正文块不存在但 expected_revision={expected_revision}")
|
||
new_rev = 1
|
||
block_id = conn.execute(
|
||
"INSERT INTO muse_content_block(work_id, chapter_id, order_no, block_type, title, "
|
||
"content_text, revision, word_count, command_id, creator, updater) "
|
||
"VALUES (%s,%s,%s,'scene',%s,%s,1,%s,%s,%s,%s) RETURNING id",
|
||
(work_id, chapter_id, chap, chapter_title, body, word_count, command_id,
|
||
decided_by, decided_by)).fetchone()[0]
|
||
# 3) 来源归因落块（来源权威；block_id+revision 唯一，按版本归因）
|
||
lineage = {"candidate_id": cid, "run_id": run_id, "candidate_version": cver,
|
||
"candidate_sha256": csha, "mode": mode, "source_role": source_role,
|
||
"source_type": source_type}
|
||
conn.execute(
|
||
"INSERT INTO muse_content_block_source_attribution(work_id, block_id, revision, source_type, "
|
||
"source_object_id, source_version, lineage_payload, creator) "
|
||
"VALUES (%s,%s,%s,%s,%s,%s,%s::jsonb,%s)",
|
||
(work_id, block_id, new_rev, source_type, str(cid), int(attempt or 1),
|
||
json.dumps(lineage, ensure_ascii=False), CREATOR))
|
||
# 4) 命令幂等审计
|
||
if command_id:
|
||
conn.execute(
|
||
"INSERT INTO muse_content_command_log(command_id, command_type, owner_user_id, target_type, "
|
||
"target_id, request_hash, result_snapshot, creator) "
|
||
"VALUES (%s,'accept_candidate',1,'content_block',%s,%s,%s::jsonb,%s)",
|
||
(command_id, block_id, csha,
|
||
json.dumps({"canonical_block_id": block_id, "revision": new_rev}, ensure_ascii=False),
|
||
CREATOR))
|
||
# 5) 决策归档（append-only）
|
||
dec_id = conn.execute(
|
||
"INSERT INTO example_user_decision(candidate_id, work_id, target_chapter, decision, rationale, "
|
||
"basis_ref, decided_by, expected_revision, command_id, canonical_block_id, creator) "
|
||
"VALUES (%s,%s,%s,'accept',%s,%s,%s,%s,%s,%s,%s) RETURNING id",
|
||
(cid, work_id, chap, rationale, basis_ref, decided_by, expected_revision, command_id,
|
||
block_id, CREATOR)).fetchone()[0]
|
||
# 6) 翻候选状态
|
||
n = conn.execute(
|
||
"UPDATE example_candidate SET state='accepted', updater=%s WHERE id=%s AND state='passed'",
|
||
(decided_by, cid)).rowcount
|
||
if n != 1:
|
||
raise ConflictError("候选状态被并发修改")
|
||
if dry_run:
|
||
conn.rollback()
|
||
return {"status": "dry_run_ok", "block_id": block_id, "revision": new_rev,
|
||
"decision_id": dec_id, "word_count": word_count, "note": "试跑已回滚，未落库"}
|
||
conn.commit()
|
||
return {"status": "accepted", "block_id": block_id, "revision": new_rev,
|
||
"decision_id": dec_id, "word_count": word_count}
|
||
except Exception:
|
||
conn.rollback()
|
||
raise
|
||
|
||
|
||
def discard(candidate_id, decided_by="1", rationale=None, basis_ref=None,
|
||
command_id=None, dry_run=False):
|
||
"""丢弃候选：正式正文不变，记一条 discard 决策 + 候选翻 discarded。"""
|
||
with connect() as conn:
|
||
try:
|
||
if command_id and conn.execute(
|
||
"SELECT 1 FROM muse_content_command_log WHERE command_id=%s AND deleted=false",
|
||
(command_id,)).fetchone():
|
||
return {"status": "already_applied", "command_id": command_id}
|
||
row = conn.execute(
|
||
"SELECT id, work_id, target_chapter, state FROM example_candidate WHERE id=%s AND deleted=false",
|
||
(candidate_id,)).fetchone()
|
||
if not row:
|
||
raise ConflictError(f"候选 {candidate_id} 不存在")
|
||
cid, work_id, chap, state = row
|
||
if state in ("accepted", "discarded", "archived"):
|
||
raise ConflictError(f"候选 state={state}，不可再丢弃")
|
||
if command_id:
|
||
conn.execute(
|
||
"INSERT INTO muse_content_command_log(command_id, command_type, owner_user_id, target_type, "
|
||
"target_id, request_hash, creator) VALUES (%s,'discard_candidate',1,'candidate',%s,%s,%s)",
|
||
(command_id, cid, str(cid), CREATOR))
|
||
dec_id = conn.execute(
|
||
"INSERT INTO example_user_decision(candidate_id, work_id, target_chapter, decision, rationale, "
|
||
"basis_ref, decided_by, command_id, creator) VALUES (%s,%s,%s,'discard',%s,%s,%s,%s,%s) RETURNING id",
|
||
(cid, work_id, chap, rationale, basis_ref, decided_by, command_id, CREATOR)).fetchone()[0]
|
||
conn.execute("UPDATE example_candidate SET state='discarded', updater=%s WHERE id=%s",
|
||
(decided_by, cid))
|
||
if dry_run:
|
||
conn.rollback()
|
||
return {"status": "dry_run_ok", "decision_id": dec_id, "note": "试跑已回滚，未落库"}
|
||
conn.commit()
|
||
return {"status": "discarded", "decision_id": dec_id}
|
||
except Exception:
|
||
conn.rollback()
|
||
raise
|
||
|
||
|
||
if __name__ == "__main__":
|
||
import argparse
|
||
ap = argparse.ArgumentParser(description="正文 Canonical 写入层（候选接受/丢弃）")
|
||
sub = ap.add_subparsers(dest="cmd", required=True)
|
||
pa = sub.add_parser("accept", help="接受候选为正式正文")
|
||
pa.add_argument("candidate_id", type=int)
|
||
pa.add_argument("--decided-by", default="1")
|
||
pa.add_argument("--rationale", default=None)
|
||
pa.add_argument("--basis-ref", default=None)
|
||
pa.add_argument("--expected-revision", type=int, default=None)
|
||
pa.add_argument("--command-id", default=None)
|
||
pa.add_argument("--source-type", default="ai_candidate",
|
||
choices=["ai_candidate", "user_merge"])
|
||
pa.add_argument("--dry-run", action="store_true")
|
||
pd = sub.add_parser("discard", help="丢弃候选")
|
||
pd.add_argument("candidate_id", type=int)
|
||
pd.add_argument("--decided-by", default="1")
|
||
pd.add_argument("--rationale", default=None)
|
||
pd.add_argument("--basis-ref", default=None)
|
||
pd.add_argument("--command-id", default=None)
|
||
pd.add_argument("--dry-run", action="store_true")
|
||
args = ap.parse_args()
|
||
try:
|
||
if args.cmd == "accept":
|
||
result = accept(args.candidate_id, decided_by=args.decided_by, rationale=args.rationale,
|
||
basis_ref=args.basis_ref, expected_revision=args.expected_revision,
|
||
command_id=args.command_id, source_type=args.source_type, dry_run=args.dry_run)
|
||
else:
|
||
result = discard(args.candidate_id, decided_by=args.decided_by, rationale=args.rationale,
|
||
basis_ref=args.basis_ref, command_id=args.command_id, dry_run=args.dry_run)
|
||
print(json.dumps(result, ensure_ascii=False))
|
||
except ConflictError as e:
|
||
print(f"[拒绝] {e}", file=sys.stderr)
|
||
sys.exit(1)
|