muse-agent-example/src/muse/正文写作/写作任务.py

"""多粒度写作任务合同：形成目标、范围、前置和预算；不调用模型、不采纳候选。"""

from __future__ import annotations

from dataclasses import dataclass, field
from typing import Any
from uuid import NAMESPACE_URL, uuid5

from muse.作品规划.接口 import 读取章节归属
from muse.共享.错误 import Muse错误
from muse.正文写作.存储 import 正文存储
from muse.正文写作.模型 import 正文错误
from muse.正文写作.正文格式 import 可见文本, 文稿身份


class 写作任务错误(Muse错误):
    错误码 = "WRITING_TASK_INVALID"


def 写作粒度集() -> tuple[str, str, str]:
    """多粒度写作的稳定取值：新章、扩写、选段修改。"""

    return ("new_chapter", "expand_scene", "edit_selection")


@dataclass(frozen=True, slots=True)
class 字符选区:
    """候选可见文本上的字符区间；半开区间 [起点, 终点)。"""

    起点: int
    终点: int

    def __post_init__(self) -> None:
        if (
            isinstance(self.起点, bool)
            or isinstance(self.终点, bool)
            or not (0 <= self.起点 < self.终点)
        ):
            raise 写作任务错误("字符选区必须是 0 ≤ 起点 < 终点的整数区间")


@dataclass(frozen=True, slots=True)
class 写作任务合同:
    """一次多粒度写作的业务目标；身份由作者与命令确定性派生，重放同身份。"""

    任务ID: str
    命令ID: str
    作者: str
    粒度: str
    用途说明: str
    work_id: str
    chapter_id: str
    branch_id: str
    基线版本: int
    选区: tuple[字符选区, ...] = ()
    保护区: tuple[字符选区, ...] = ()
    预算: dict[str, Any] = field(default_factory=dict)
    停止条件: tuple[str, ...] = ()

    def __post_init__(self) -> None:
        if self.粒度 not in 写作粒度集():
            raise 写作任务错误(f"未知写作粒度：{self.粒度}")
        if self.粒度 == "edit_selection" and not self.选区:
            raise 写作任务错误("选段修改必须携带明确选区")
        if self.基线版本 < 0 or not self.任务ID or not self.命令ID or not self.作者:
            raise 写作任务错误("写作任务缺少身份或基线版本不合法")
        for 甲 in self.选区:
            for 乙 in self.保护区:
                if 甲.起点 < 乙.终点 and 乙.起点 < 甲.终点:
                    raise 写作任务错误("选区与保护区不能重叠")


def 创建写作任务(
    连,
    身份,
    命令ID: str,
    粒度: str,
    *,
    work_id: str,
    chapter_id: str,
    分支: str = "main",
    用途说明: str = "",
    选区: tuple[字符选区, ...] = (),
    保护区: tuple[字符选区, ...] = (),
    预算: dict[str, Any] | None = None,
    停止条件: tuple[str, ...] = (),
) -> 写作任务合同:
    """按当前章位冻结写作任务；章位归属与选区边界在创建时机械校验。"""

    if not 命令ID:
        raise 写作任务错误("写作任务需要明确命令身份")
    章节 = 读取章节归属(连, 身份.作者, chapter_id)
    if 章节["work_id"] != work_id:
        raise 写作任务错误("章节不属于目标作品")
    存储 = 正文存储(连)
    文档ID = 文稿身份(chapter_id, 分支)
    正文 = 存储.读取(文档ID)
    基线版本 = 正文["revision"] if 正文 else 0
    任务ID = str(uuid5(NAMESPACE_URL, "muse:B05:writing_task:" + _身份哈希(身份.作者, 命令ID)))
    任务 = 写作任务合同(
        任务ID,
        命令ID,
        身份.作者,
        粒度,
        用途说明,
        work_id,
        chapter_id,
        分支,
        基线版本,
        选区,
        保护区,
        预算 or {},
        停止条件,
    )
    if 基线版本 == 0:
        if 任务.粒度 != "new_chapter":
            raise 写作任务错误("没有既有正文，只允许新章粒度")
        return 任务
    可见长度 = len(可见文本(存储.读取草稿(文档ID, 基线版本)))
    for 区 in 任务.选区 + 任务.保护区:
        if 区.终点 > 可见长度:
            raise 写作任务错误(f"选区越界：终点 {区.终点} 超出当前正文长度 {可见长度}")
    return 任务


def _身份哈希(作者: str, 命令ID: str) -> str:
    from muse.正式变更.接口 import 固定哈希

    return 固定哈希([作者, 命令ID])


__all__ = [
    "写作任务合同",
    "写作任务错误",
    "写作粒度集",
    "字符选区",
    "创建写作任务",
    "正文错误",
]


_写作输出合同 = {
    "type": "object",
    "properties": {
        "paragraphs": {
            "description": (
                "正文段落数组；每项 text 是完整段落文本，不能包含 LF 或 CR 换行符，"
                "需要另起一段时另建一个段落对象；系统不改写或规范化段内换行。"
            ),
            "type": "array",
            "minItems": 1,
            "items": {
                "type": "object",
                "properties": {
                    "text": {"type": "string", "minLength": 1, "not": {"pattern": "[\\r\\n]"}}
                },
                "required": ["text"],
                "additionalProperties": False,
            },
        },
        "blocked": {
            "description": (
                "写作死结出口：仅当细纲与前置事实存在不可化解冲突"
                "（任何写法都必然虚构冲突事实）时输出，reason 一句话说明。"
            ),
            "type": "object",
            "properties": {"reason": {"type": "string", "minLength": 1}},
            "required": ["reason"],
            "additionalProperties": False,
        },
    },
    "oneOf": [{"required": ["paragraphs"]}, {"required": ["blocked"]}],
    "additionalProperties": False,
}


def 写作输出合同() -> dict:
    """生产与评测共享同一无工具写作输出外壳；返回独立副本。"""
    import copy

    return copy.deepcopy(_写作输出合同)


def 生成正文模板() -> tuple[str, str]:
    """读取发布包登记的写作模板及实际字节哈希。

    角色身份、权限、预算及输出合同由调用方装配，不作为写作指令发给模型。
    """
    import hashlib

    from muse.资源加载 import 打开资源

    data = 打开资源("正文/生成正文.md")
    return data.decode("utf-8"), hashlib.sha256(data).hexdigest()