框架(规划管线): 建规划落库机制 persist_planning,planner 产结构化草稿主会话落库

- persist_planning.py:write(shadow)/confirm(confirmed) 落 example_planning_section;
  section_type=setting/outline/state/assembly/fine_outline;fine_outline 章级强制 target_chapter;全流程已验证
- planning SKILL.md:落库合同(section_type/payload/确认/已确认细纲取用SQL),规划正式权威在库、works/文件退为留痕
- planner.md 禁区:不写库/git,落库由主会话经 persist_planning 做
This commit is contained in:
zizi 2026-07-31 22:43:39 +08:00
parent 96aa1e61a3
commit e7b5100ba5
3 changed files with 111 additions and 2 deletions

View File

@ -27,4 +27,4 @@ model: opus
## 禁区 ## 禁区
不写正文;不动 `meta/` 与框架文件;不执行 git 写操作。 不写正文;不动 `meta/` 与框架文件;不执行 git 写操作、不写数据库(规划落库由主会话经 `planning` skill 的 `persist_planning.py` 做)。

View File

@ -34,7 +34,19 @@ disable-model-invocation: true
## 输出合同 ## 输出合同
`works/<书名>/` 全套文件,不提交;完成后报:字段存疑清单/咬合自查结论/伏笔台账条数。 产出按 section_type 分类:设定(`setting`)/大纲(`outline`)/状态(`state`)/装配(`assembly`)/细纲(`fine_outline`,章级)。planner 产出 schema 字段结构化内容(payload JSON),由主会话经 `scripts/persist_planning.py` 落 `example_planning_section`(先 shadow);`works/<书名>/` 文件是人类可读留痕,**规划正式权威在库**。完成后报:字段存疑清单/咬合自查结论/伏笔台账条数。
## 规划落库(shadow→confirmed)
1. 落库:`.venv/bin/python .claude/skills/planning/scripts/persist_planning.py write <payload.json> --work-id N --section-type <type> [--target-chapter N] [--schema-type X] [--version N]`(planner 不碰库/git,主会话落库;先 `--dry-run` 试跑)。
2. 确认:`persist_planning.py confirm <section_id>`(用户确认后 shadow→confirmed;**确认后的规划才进生成上下文**)。
3. 已确认细纲取用(read-context 改道读库,不再当内存参数):
```sql
SELECT payload FROM example_planning_section
WHERE work_id=? AND section_type='fine_outline' AND target_chapter=? AND state='confirmed' AND deleted=false
ORDER BY version DESC LIMIT 1
```
4. `fine_outline` 是章级 section,必须带 `target_chapter`;书级(设定/大纲/状态/装配)`target_chapter` 为空。
## 红线 ## 红线

View File

@ -0,0 +1,97 @@
#!/usr/bin/env python3
"""规划落库 —— 把 planner 产出的 schema 驱动结构化内容落 example_planning_section(shadow→confirmed)。
落库设计 §2.6:规划/细纲/设定的正式权威在库(planning_section),works/<书名>/ 文件退为人类可读留痕。
- write:落一条 shadow 规划(planner 产出结构化 payload,主会话调本脚本落库;planner 不碰库/git)。
- confirm:shadow→confirmed(用户确认后翻态,confirmed_at 落时间)——确认后的规划才进生成上下文。
- 已确认细纲取用见 planning SKILL.md 的 SQL。
payload = 对应 section_type 的 schema 字段结构化内容(JSON),由 planner/主会话按 meta/schemas 产出。
--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 = "planning"
SECTION_TYPES = ("setting", "outline", "state", "assembly", "fine_outline")
def write_section(work_id, *, section_type, payload, target_chapter=None, schema_type=None,
version=1, state="shadow", creator=CREATOR, dry_run=False):
"""落一条规划(默认 shadow)。返回 {status, id, state}。"""
if section_type not in SECTION_TYPES:
raise ValueError(f"section_type 非法:{section_type}(应为 {SECTION_TYPES})")
if section_type == "fine_outline" and target_chapter is None:
raise ValueError("fine_outline 是章级 section,必须给 target_chapter")
if not isinstance(payload, dict) or not payload:
raise ValueError("payload 必须是非空 JSON 对象(schema 字段结构化内容)")
with connect() as conn:
try:
row = conn.execute(
"INSERT INTO example_planning_section(work_id, target_chapter, section_type, schema_type, "
"payload, version, state, creator) VALUES (%s,%s,%s,%s,%s::jsonb,%s,%s,%s) RETURNING id, state",
(work_id, target_chapter, section_type, schema_type,
json.dumps(payload, ensure_ascii=False), version, state, creator)).fetchone()
if dry_run:
conn.rollback()
return {"status": "dry_run_ok", "id": row[0], "state": row[1], "note": "试跑已回滚,未落库"}
conn.commit()
return {"status": "written", "id": row[0], "state": row[1]}
except Exception:
conn.rollback()
raise
def confirm_section(section_id, *, updater=CREATOR, dry_run=False):
"""确认规划:shadow→confirmed。返回 {status, id, state}。"""
with connect() as conn:
try:
row = conn.execute(
"UPDATE example_planning_section SET state='confirmed', confirmed_at=CURRENT_TIMESTAMP, "
"updater=%s WHERE id=%s AND state='shadow' AND deleted=false RETURNING id, state",
(updater, section_id)).fetchone()
if not row:
cur = conn.execute("SELECT state FROM example_planning_section WHERE id=%s", (section_id,)).fetchone()
raise ValueError(f"规划 {section_id} 不可确认(不存在或 state={cur[0] if cur else '无'},只有 shadow 可确认)")
if dry_run:
conn.rollback()
return {"status": "dry_run_ok", "id": row[0], "state": row[1], "note": "试跑已回滚,未落库"}
conn.commit()
return {"status": "confirmed", "id": row[0], "state": row[1]}
except Exception:
conn.rollback()
raise
if __name__ == "__main__":
import argparse
ap = argparse.ArgumentParser(description="规划落库(shadow→confirmed)")
sub = ap.add_subparsers(dest="cmd", required=True)
pw = sub.add_parser("write", help="落一条规划(默认 shadow)")
pw.add_argument("payload_json", help="payload JSON 文件路径(schema 字段结构化内容)")
pw.add_argument("--work-id", type=int, required=True)
pw.add_argument("--section-type", required=True, choices=SECTION_TYPES)
pw.add_argument("--target-chapter", type=int, default=None)
pw.add_argument("--schema-type", default=None)
pw.add_argument("--version", type=int, default=1)
pw.add_argument("--creator", default=CREATOR)
pw.add_argument("--dry-run", action="store_true")
pc = sub.add_parser("confirm", help="确认规划(shadow→confirmed)")
pc.add_argument("section_id", type=int)
pc.add_argument("--updater", default=CREATOR)
pc.add_argument("--dry-run", action="store_true")
args = ap.parse_args()
if args.cmd == "write":
payload = json.loads(Path(args.payload_json).read_text(encoding="utf-8"))
out = write_section(args.work_id, section_type=args.section_type, payload=payload,
target_chapter=args.target_chapter, schema_type=args.schema_type,
version=args.version, creator=args.creator, dry_run=args.dry_run)
else:
out = confirm_section(args.section_id, updater=args.updater, dry_run=args.dry_run)
print(json.dumps(out, ensure_ascii=False))