治理任务契约、写隔离与失败恢复

script_build_host/src/script_build_host/agents/prompts/script_planner.md

@@ -14,6 +14,10 @@
   `description`。Retrieval 只允许 `evidence_nonempty`、`source_attributable`、
   `semantic_relevance`、`concept_coverage`、`freshness`，不能要求映射、人设分析、写作或
   结构化合成文档。通用硬标准由 Host 注入。
+- Retrieval `params` 必须严格按类型填写：`evidence_nonempty` 只可选
+  `minimum_count`；`source_attributable` 使用空对象；`semantic_relevance` 和
+  `concept_coverage` 必须提供 1–12 个非空字符串的 `concepts`，可选
+  `minimum_matches`；`freshness` 只可选 `max_age_days`。不存在 `query` 参数。
 - 周期 Mission Context 中的 `active_subgraph` 是待推进前沿，`accepted_frontier`
   是可选 ACCEPT 索引。Decision ID 只能来自该索引。不要提交物理引用、
   内容摘要值、producer/write scope、schema、budget、candidate closure 或 compose order；
@@ -29,6 +33,12 @@
   当前 Compose/Portfolio 的采用闭包。
 - `scope_selector` 只写 `{anchor: mission|parent, path: [语义段...]}`；Host 负责拼接、
   包含关系和写范围，不得构造 URI。
+- 显式消费已 ACCEPT 的创作产物时，消费者必须留在同一个 Compose 创作分支。
+  Structure、Paragraph、ElementSet 通常是同一 Compose Task 下的兄弟 Task：用 Compose
+  Task ID 作 `parent_task_id`，再把生产者 Decision 放入 `input_decision_ids`。已 completed
+  的生产者 Task 不能再接收子 Task。不要改用 `base_decision_id` 绕过 scope。发生
+  `INPUT_SCOPE_MISMATCH` 时优先按 failure details 中的 `producer_parent_task_id` 和 `recovery`
+  重建正确层级。
 - Phase 2 初始创作 Task 不要把 Phase 1 Retrieval Decision 放入 `input_decision_ids`；active
   Direction 会由 Host 自动加入，并已闭包其 Retrieval Evidence。只有同一创作分支中确需采用的
   已 ACCEPT 产物才显式列入输入。
@@ -41,7 +51,8 @@ REPAIR、BLOCK、CANCEL 不提交 Task input。Compose 和 CandidatePortfolio
 `selected_decision_ids`，由 Host 自动补齐 closure、未采用项和 compose order。Compose 必须选择
 已 ACCEPT 的 Structure、Paragraph、ElementSet；Portfolio 只选择一个已 ACCEPT Compose。
 
-Paragraph 可以无 base 探索；若修订已接受 Structure，必须使用其 `base_decision_id`。替换已
+Paragraph 可以无 base 探索；`base_decision_id` 只表示修订同类已接受产物，不能用来跨
+类型消费 Structure。替换已
 ACCEPT 产物时才用 `supersedes_decision_ids`。等价 Task 会被拒绝：应继续、修订或处置已有 Task，
 不要用新 ID 制造随机重复候选。
 

script_build_host/src/script_build_host/application/failure_policy.py

@@ -0,0 +1,134 @@
+"""Classify Agent-visible script-build failures in one application policy."""
+
+from __future__ import annotations
+
+from collections.abc import Mapping
+from typing import Any
+
+from agent import FailureDetail, FailureDisposition
+
+from script_build_host.domain.errors import ScriptBuildError
+
+_REPLAN_CODES = frozenset(
+    {
+        "INPUT_SCOPE_MISMATCH",
+        "CHILD_DECISION_INVALID",
+        "PHASE_POLICY_VIOLATION",
+        "PHASE_TWO_BOUNDARY_NOT_READY",
+        "PHASE_THREE_BOUNDARY_NOT_READY",
+        "PROTOCOL_VIOLATION",
+        "TASK_CHILD_BLOCKED",
+        "TASK_BUDGET_EXCEEDED",
+        "TASK_KIND_PRESET_MISMATCH",
+        "TASK_STATE_CONFLICT",
+        "ATTEMPT_WORKSPACE_FROZEN",
+        "CONTEXT_HANDLE_UNKNOWN",
+        "WRITE_SCOPE_VIOLATION",
+        "WRITE_SCOPE_CONFLICT",
+        "SUPERSESSION_CYCLE",
+    }
+)
+_REPAIR_CODES = frozenset(
+    {
+        "LEGACY_WRITE_INVALID",
+        "LEGACY_REFERENCE_INVALID",
+        "ARTIFACT_NOT_FOUND",
+    }
+)
+_RETRY_CODES = frozenset(
+    {
+        "PUBLICATION_LOCK_TIMEOUT",
+        "PUBLICATION_DEADLOCK_RETRYABLE",
+        "STALE_BASE_REVISION",
+        "STALE_WORKBENCH_STATE",
+        "VALIDATION_EVIDENCE_INVALID",
+        "CONTEXT_NOT_EXHAUSTED",
+        "CONTEXT_HANDLE_INVALID",
+        "CONTEXT_HANDLE_FORMAT_INVALID",
+        "CONTEXT_HANDLE_UNAUTHORIZED",
+        "CONTEXT_CURSOR_INVALID",
+        "TRANSIENT_TIMEOUT",
+        "TRANSIENT_CONNECTION_LOST",
+        "RATE_LIMITED",
+        "UPSTREAM_UNAVAILABLE",
+    }
+)
+_ABORT_MARKERS = (
+    "STOP",
+    "FENCING",
+    "OWNERSHIP",
+    "ALREADY_OWNED",
+    "DIGEST",
+    "RECOVERY",
+    "MIGRATION",
+    "RECONCILIATION",
+)
+
+
+def classify_script_failure(
+    error: ScriptBuildError | ValueError,
+    *,
+    source_tool: str,
+    context: Mapping[str, Any] | None = None,
+) -> FailureDetail:
+    """Return the sole Host classification for an Agent-visible failure."""
+
+    if isinstance(error, ValueError) and not isinstance(error, ScriptBuildError):
+        return FailureDetail(
+            code="INVALID_TOOL_ARGUMENT",
+            message=str(error),
+            disposition=FailureDisposition.RETRY_CALL,
+            source_tool=source_tool,
+            details=_failure_details(error, context),
+        )
+
+    code = error.code
+    if code == "SCRIPT_CONTENT_INCOMPLETE":
+        disposition = (
+            FailureDisposition.REPAIR_ATTEMPT
+            if source_tool in {"submit_attempt", "save_paragraph_candidate"}
+            else FailureDisposition.REPLAN_TASK
+        )
+    elif (
+        code in _REPLAN_CODES
+        or code.startswith("GOAL_")
+        or code.startswith("TASK_CAPABILITY_")
+        or code.startswith("TASK_CONTRACT_")
+        or code.startswith("ADOPTION_")
+    ):
+        disposition = FailureDisposition.REPLAN_TASK
+    elif code in _REPAIR_CODES:
+        disposition = FailureDisposition.REPAIR_ATTEMPT
+    elif code in _RETRY_CODES or code.startswith("CONTEXT_CURSOR_"):
+        disposition = FailureDisposition.RETRY_CALL
+    elif any(marker in code for marker in _ABORT_MARKERS):
+        disposition = FailureDisposition.ABORT_RUN
+    else:
+        disposition = FailureDisposition.ABORT_RUN
+    return FailureDetail(
+        code=code,
+        message=error.summary,
+        disposition=disposition,
+        source_tool=source_tool,
+        details=_failure_details(error, context),
+    )
+
+
+def _failure_details(
+    error: ScriptBuildError | ValueError,
+    context: Mapping[str, Any] | None,
+) -> dict[str, Any]:
+    raw = getattr(error, "details", None)
+    details = dict(raw) if isinstance(raw, Mapping) else {}
+    if isinstance(context, Mapping):
+        details.update(
+            {
+                key: str(context[key])
+                for key in ("task_id", "attempt_id", "validation_id", "operation_id")
+                if context.get(key) not in (None, "")
+            }
+        )
+    return details
+
+
+__all__ = ["classify_script_failure"]

script_build_host/src/script_build_host/application/mission_workbench.py

@@ -12,10 +12,11 @@ from dataclasses import asdict, is_dataclass, replace
 from hashlib import sha256
 from typing import Any, cast
 
-from agent.failures import FailureDetail, FailureDisposition, ToolExecutionError
+from agent.failures import ToolExecutionError
 from agent.orchestration import AgentRole, DecisionAction, RoleContextRequest, TaskStatus
 
 from script_build_host.application.context_semantics import AdaptiveSemanticSelector
+from script_build_host.application.failure_policy import classify_script_failure
 from script_build_host.domain.context_broker import (
     ContextBundle,
     ContextCard,
@@ -132,12 +133,10 @@ class ContextBroker:
             return {}
         except WorkbenchError as exc:
             raise ToolExecutionError(
-                FailureDetail(
-                    code=exc.code,
-                    message=exc.message,
-                    disposition=FailureDisposition.ABORT_RUN,
+                classify_script_failure(
+                    exc,
                     source_tool="mission_workbench_role_context",
-                    details={"task_id": request.task_id, "role": request.role.value},
+                    context={"task_id": request.task_id, "role": request.role.value},
                 )
             ) from exc
 
@@ -297,10 +296,7 @@ class ContextBroker:
             },
             phase_protocol=(await self._phase_protocol(ledger, phase)).to_payload(),
         )
-        _bounded_context(
-            state.to_payload(), char_limit=24_000, token_limit=6_000, label="PlannerState"
-        )
-        return state
+        return _fit_planner_state(state)
 
     async def _phase_protocol(self, ledger: Any, phase: int) -> PhaseProtocolSnapshot:
         root = ledger.tasks[ledger.root_task_id]
@@ -2295,6 +2291,42 @@ def _bounded_context(value: Any, *, char_limit: int, token_limit: int, label: st
     _bounded_tokens(value, token_limit, label)
 
 
+def _fit_planner_state(state: PlannerState) -> PlannerState:
+    candidates = (
+        state,
+        replace(
+            state,
+            active_subgraph=state.active_subgraph[:8],
+            accepted_frontier=state.accepted_frontier[:4],
+            active_operations=state.active_operations[:4],
+            latest_failures=state.latest_failures[:4],
+            changes=state.changes[:4],
+            active_has_more=state.active_has_more or len(state.active_subgraph) > 8,
+            accepted_has_more=state.accepted_has_more or len(state.accepted_frontier) > 4,
+        ),
+        replace(
+            state,
+            active_subgraph=state.active_subgraph[:3],
+            accepted_frontier=state.accepted_frontier[:2],
+            active_operations=state.active_operations[:2],
+            latest_failures=state.latest_failures[:2],
+            changes=(),
+            active_has_more=state.active_has_more or len(state.active_subgraph) > 3,
+            accepted_has_more=state.accepted_has_more or len(state.accepted_frontier) > 2,
+            context_bundle={
+                **dict(state.context_bundle),
+                "cards": [],
+                "compacted": True,
+            },
+        ),
+    )
+    for candidate in candidates:
+        payload = candidate.to_payload()
+        if len(_json(payload)) <= 24_000 and estimate_context_tokens(payload) <= 6_000:
+            return candidate
+    raise WorkbenchError("CONTEXT_BUDGET_EXCEEDED", "minimal PlannerState exceeds budget")
+
+
 def _context_fragment(value: str, *, start: int, token_budget: int) -> str:
     """Return the largest bounded fragment without splitting context by byte size."""
 

script_build_host/src/script_build_host/application/phase_two_inputs.py

@@ -170,9 +170,24 @@ class AcceptedInputResolver:
                         "CHILD_DECISION_INVALID",
                         "frozen direct-child decision is not owned by a direct child",
                     )
-                _validate_consumer_input(contract, item.contract, direct_child=True)
+                _validate_consumer_input(
+                    contract,
+                    item.contract,
+                    direct_child=True,
+                    same_parent=False,
+                )
             else:
-                _validate_consumer_input(contract, item.contract, direct_child=False)
+                consumer_task = ledger.tasks[task_id]
+                producer_task = ledger.tasks[item.artifact_version.task_id]
+                _validate_consumer_input(
+                    contract,
+                    item.contract,
+                    direct_child=False,
+                    same_parent=(
+                        consumer_task.parent_task_id is not None
+                        and consumer_task.parent_task_id == producer_task.parent_task_id
+                    ),
+                )
             resolved.append(item)
 
         input_values = tuple(item.accepted_input for item in resolved)
@@ -552,6 +567,7 @@ def _validate_consumer_input(
     produced: ScriptTaskContractV1,
     *,
     direct_child: bool,
+    same_parent: bool,
 ) -> None:
     try:
         validate_accepted_input(
@@ -562,6 +578,7 @@ def _validate_consumer_input(
             source=(
                 AcceptedInputSource.DIRECT_CHILD if direct_child else AcceptedInputSource.EXPLICIT
             ),
+            same_parent=same_parent,
         )
     except AcceptedInputCompatibilityError as exc:
         raise PhaseTwoInputError(

script_build_host/src/script_build_host/application/phase_two_planning.py

@@ -84,6 +84,9 @@ _CANDIDATE_KINDS = frozenset(
         ScriptTaskKind.COMPOSE,
     }
 )
+_WORKSPACE_WRITERS = frozenset(
+    {ScriptTaskKind.STRUCTURE, ScriptTaskKind.PARAGRAPH, ScriptTaskKind.ELEMENT_SET}
+)
 _DIRECTION_INPUT_KINDS = frozenset(
     {
         ScriptTaskKind.STRUCTURE,
@@ -503,7 +506,9 @@ class PhaseTwoPlanningService:
         await self._guard_task_budget(root_trace_id, ledger, parsed_contracts, parent)
         await self._guard_hierarchy(root_trace_id, ledger, parent, parsed_contracts)
         await self._guard_supersessions(root_trace_id, ledger, parsed_contracts)
-        await self._guard_accepted_input_scopes(root_trace_id, ledger, parsed_contracts)
+        await self._guard_accepted_input_scopes(
+            root_trace_id, ledger, parsed_contracts, parent=parent
+        )
         await self._guard_duplicate_contracts(
             root_trace_id=root_trace_id,
             ledger=ledger,
@@ -515,13 +520,16 @@ class PhaseTwoPlanningService:
         )
         for item, planned_task_id in zip(planned, planned_task_ids, strict=True):
             item.draft["_task_id"] = planned_task_id
-        result = await self.coordinator.create_tasks(
-            root_trace_id,
-            [item.draft for item in planned],
-            parent_task_id=parent_id,
-            placement={"after_task_id": after_task_id, "focus": focus},
-            idempotency_key=_optional(context, "tool_call_id"),
-        )
+        try:
+            result = await self.coordinator.create_tasks(
+                root_trace_id,
+                [item.draft for item in planned],
+                parent_task_id=parent_id,
+                placement={"after_task_id": after_task_id, "focus": focus},
+                idempotency_key=_optional(context, "tool_call_id"),
+            )
+        except TaskConflict as exc:
+            raise TaskContractError("TASK_STATE_CONFLICT", str(exc)) from exc
         created_task_ids = [str(item["task_id"]) for item in result.get("tasks", [])]
         return {
             **result,
@@ -685,7 +693,9 @@ class PhaseTwoPlanningService:
                 parent=parent,
             )
             await self._guard_supersessions(root_trace_id, ledger, (contract,))
-            await self._guard_accepted_input_scopes(root_trace_id, ledger, (contract,))
+            await self._guard_accepted_input_scopes(
+                root_trace_id, ledger, (contract,), parent=parent
+            )
             planned_replacement = (
                 await self._freeze_all(root_trace_id, str(binding.input_snapshot_id), (contract,))
             )[0]
@@ -876,11 +886,14 @@ class PhaseTwoPlanningService:
                 raise TaskContractError("TASK_BUDGET_EXCEEDED", "Task attempt budget is exhausted")
             self._guard_task_usage(ledger, task, contract)
             await self._guard_supersessions(root_trace_id, ledger, (contract,))
-            await self._guard_accepted_input_scopes(root_trace_id, ledger, (contract,))
+            await self._guard_accepted_input_scopes(
+                root_trace_id, ledger, (contract,), parent=parent
+            )
             contract.require_execution_ready()
             self._guard_retrieval_parent(ledger, task, contract.task_kind)
             preset = TASK_PRESET_BY_KIND[contract.task_kind]
             presets.append(preset)
+        await self._guard_concurrent_write_scopes(root_trace_id, ledger, task_ids)
         operations = await self.coordinator.dispatch_tasks(
             root_trace_id,
             list(task_ids),
@@ -892,6 +905,52 @@ class PhaseTwoPlanningService:
         )
         return [item.to_dict() for item in operations]
 
+    async def _guard_concurrent_write_scopes(
+        self, root_trace_id: str, ledger: Any, task_ids: Sequence[str]
+    ) -> None:
+        requested = set(task_ids)
+        active = {
+            task_id
+            for operation in ledger.operations.values()
+            if operation.status in _ACTIVE_OPERATION
+            for task_id in operation.task_ids
+        }
+        relevant = requested | active
+        contracts = {
+            task_id: (await self.contract_for_task(root_trace_id, ledger.tasks[task_id])).contract
+            for task_id in relevant
+            if task_id in ledger.tasks
+        }
+        pairs = {
+            tuple(sorted((left, right)))
+            for left in requested
+            for right in relevant
+            if left != right
+        }
+        for left, right in sorted(pairs):
+            left_scope = contracts.get(left)
+            right_scope = contracts.get(right)
+            if left_scope is None or right_scope is None:
+                continue
+            same_workspace = (
+                ledger.tasks[left].parent_task_id is not None
+                and ledger.tasks[left].parent_task_id == ledger.tasks[right].parent_task_id
+                and left_scope.task_kind in _WORKSPACE_WRITERS
+                and right_scope.task_kind in _WORKSPACE_WRITERS
+            )
+            overlapping_writes = any(
+                _scope_contains(one, two) or _scope_contains(two, one)
+                for one in left_scope.write_scope
+                for two in right_scope.write_scope
+            )
+            if same_workspace or overlapping_writes:
+                raise TaskContractError(
+                    "TASK_STATE_CONFLICT",
+                    "Tasks sharing a candidate workspace or overlapping write scopes must be "
+                    "dispatched serially",
+                    details={"task_ids": [left, right]},
+                )
+
     async def ensure_tool_budget(self, *, root_trace_id: str, task_id: str, entry: str) -> None:
         """Enforce Phase2 budgets at Retrieval and Attempt submission boundaries."""
 
@@ -1164,7 +1223,7 @@ class PhaseTwoPlanningService:
             objective=value.objective,
             input_decision_refs=tuple(refs),
             base_artifact_ref=base_ref,
-            write_scope=_write_scope_for(value.task_kind),
+            write_scope=_write_scope_for(value.task_kind, scope_ref=scope_ref),
             gap_ref=(f"script-build://gaps/{value.gap_key}" if value.gap_key else None),
             output_schema=output_schema_for(
                 value.task_kind, patched=value.base_decision_id is not None
@@ -1685,6 +1744,8 @@ class PhaseTwoPlanningService:
         root_trace_id: str,
         ledger: Any,
         contracts: Sequence[ScriptTaskContractV1],
+        *,
+        parent: Any | None = None,
     ) -> None:
         """Reject accepted-input scopes that can never form a valid frozen closure.
 
@@ -1735,6 +1796,10 @@ class PhaseTwoPlanningService:
                         producer_kind=producer.task_kind,
                         producer_scope=producer.scope_ref,
                         source=AcceptedInputSource.EXPLICIT,
+                        same_parent=(
+                            parent is not None
+                            and producer_task.parent_task_id == parent.task_id
+                        ),
                     )
                 except AcceptedInputCompatibilityError as exc:
                     raise TaskContractError(
@@ -1742,11 +1807,18 @@ class PhaseTwoPlanningService:
                         str(exc),
                         details={
                             "decision_id": imported_ref.decision_id,
+                            "producer_task_id": producer_task.task_id,
+                            "producer_parent_task_id": producer_task.parent_task_id,
                             "consumer_kind": consumer.task_kind.value,
                             "consumer_scope": consumer.scope_ref,
                             "producer_kind": producer.task_kind.value,
                             "producer_scope": producer.scope_ref,
                             "source": AcceptedInputSource.EXPLICIT.value,
+                            "recovery": (
+                                "create the consumer in the producer's creative branch, normally "
+                                "as a sibling under producer_parent_task_id; changing "
+                                "input_decision_ids to base_decision_id does not change scope"
+                            ),
                         },
                     ) from exc
 
@@ -2105,8 +2177,12 @@ def _scope_contains(parent: str, child: str) -> bool:
     return child == parent or child.startswith(f"{parent.rstrip('/')}/")
 
 
-def _write_scope_for(kind: ScriptTaskKind) -> tuple[str, ...]:
-    return task_capability(kind).write_scope
+def _write_scope_for(kind: ScriptTaskKind, *, scope_ref: str) -> tuple[str, ...]:
+    write_scope = task_capability(kind).write_scope
+    if kind not in {ScriptTaskKind.PARAGRAPH, ScriptTaskKind.ELEMENT_SET} or not write_scope:
+        return write_scope
+    scope_key = hashlib.sha256(scope_ref.encode("utf-8")).hexdigest()[:24]
+    return (f"{write_scope[0].rstrip('/')}/scopes/{scope_key}",)
 
 
 def _contract_signature(contract: ScriptTaskContractV1) -> str:

script_build_host/src/script_build_host/domain/input_compatibility.py

@@ -68,11 +68,13 @@ def validate_accepted_input(
     producer_kind: ScriptTaskKind,
     producer_scope: str,
     source: AcceptedInputSource,
+    same_parent: bool = False,
 ) -> None:
     placement = (
         source is AcceptedInputSource.EXPLICIT
-        and consumer_kind is ScriptTaskKind.ELEMENT_SET
-        and producer_kind is ScriptTaskKind.PARAGRAPH
+        and same_parent
+        and consumer_kind in _CREATIVE
+        and producer_kind in _CREATIVE
     )
     if not scopes_compatible(consumer_scope, producer_scope) and not placement:
         raise AcceptedInputCompatibilityError("accepted input is outside the consumer scope")

script_build_host/src/script_build_host/tools/failures.py

@@ -1,130 +1,8 @@
-"""Map script-build domain errors to the framework's business-neutral protocol."""
+"""Compatibility export for the centralized application failure policy."""
 
-from __future__ import annotations
+from script_build_host.application.failure_policy import classify_script_failure
 
-from collections.abc import Mapping
-from typing import Any
-
-from agent import FailureDetail, FailureDisposition
-
-from script_build_host.domain.errors import ScriptBuildError
-
-_REPLAN_CODES = frozenset(
-    {
-        "INPUT_SCOPE_MISMATCH",
-        "CHILD_DECISION_INVALID",
-        "PHASE_POLICY_VIOLATION",
-        "PHASE_TWO_BOUNDARY_NOT_READY",
-        "PHASE_THREE_BOUNDARY_NOT_READY",
-        "PROTOCOL_VIOLATION",
-        "TASK_CHILD_BLOCKED",
-        "TASK_BUDGET_EXCEEDED",
-        "TASK_KIND_PRESET_MISMATCH",
-        "TASK_STATE_CONFLICT",
-        "ATTEMPT_WORKSPACE_FROZEN",
-        "CONTEXT_HANDLE_UNKNOWN",
-        "WRITE_SCOPE_VIOLATION",
-        "SUPERSESSION_CYCLE",
-    }
-)
-_REPAIR_CODES = frozenset(
-    {
-        "LEGACY_WRITE_INVALID",
-        "LEGACY_REFERENCE_INVALID",
-        "ARTIFACT_NOT_FOUND",
-    }
-)
-_RETRY_CODES = frozenset(
-    {
-        "PUBLICATION_LOCK_TIMEOUT",
-        "PUBLICATION_DEADLOCK_RETRYABLE",
-        "STALE_BASE_REVISION",
-        "STALE_WORKBENCH_STATE",
-        "VALIDATION_EVIDENCE_INVALID",
-        "CONTEXT_NOT_EXHAUSTED",
-        "CONTEXT_HANDLE_INVALID",
-        "CONTEXT_HANDLE_FORMAT_INVALID",
-        "TRANSIENT_TIMEOUT",
-        "TRANSIENT_CONNECTION_LOST",
-        "RATE_LIMITED",
-        "UPSTREAM_UNAVAILABLE",
-    }
-)
-_ABORT_MARKERS = (
-    "STOP",
-    "FENCING",
-    "OWNERSHIP",
-    "ALREADY_OWNED",
-    "DIGEST",
-    "RECOVERY",
-    "MIGRATION",
-    "RECONCILIATION",
-)
-
-
-def classify_script_tool_failure(
-    error: ScriptBuildError | ValueError,
-    *,
-    source_tool: str,
-    context: Mapping[str, Any] | None = None,
-) -> FailureDetail:
-    """Return the sole Host classification for an Agent-visible tool error."""
-
-    if isinstance(error, ValueError) and not isinstance(error, ScriptBuildError):
-        return FailureDetail(
-            code="INVALID_TOOL_ARGUMENT",
-            message=str(error),
-            disposition=FailureDisposition.RETRY_CALL,
-            source_tool=source_tool,
-            details=_failure_details(error, context),
-        )
-
-    code = error.code
-    if code == "SCRIPT_CONTENT_INCOMPLETE":
-        disposition = (
-            FailureDisposition.REPAIR_ATTEMPT
-            if source_tool in {"submit_attempt", "save_paragraph_candidate"}
-            else FailureDisposition.REPLAN_TASK
-        )
-    elif (
-        code in _REPLAN_CODES
-        or code.startswith("GOAL_")
-        or code.startswith("TASK_CONTRACT_")
-        or code.startswith("ADOPTION_")
-    ):
-        disposition = FailureDisposition.REPLAN_TASK
-    elif code in _REPAIR_CODES:
-        disposition = FailureDisposition.REPAIR_ATTEMPT
-    elif code in _RETRY_CODES:
-        disposition = FailureDisposition.RETRY_CALL
-    elif any(marker in code for marker in _ABORT_MARKERS):
-        disposition = FailureDisposition.ABORT_RUN
-    else:
-        disposition = FailureDisposition.ABORT_RUN
-    return FailureDetail(
-        code=code,
-        message=error.summary,
-        disposition=disposition,
-        source_tool=source_tool,
-        details=_failure_details(error, context),
-    )
-
-
-def _failure_details(
-    error: ScriptBuildError | ValueError,
-    context: Mapping[str, Any] | None,
-) -> dict[str, Any]:
-    raw = getattr(error, "details", None)
-    details = dict(raw) if isinstance(raw, Mapping) else {}
-    if isinstance(context, Mapping):
-        details.update(
-            {
-                key: str(context[key])
-                for key in ("task_id", "attempt_id", "validation_id", "operation_id")
-                if context.get(key) not in (None, "")
-            }
-        )
-    return details
+classify_script_tool_failure = classify_script_failure
 
 
 __all__ = ["classify_script_tool_failure"]

script_build_host/src/script_build_host/tools/registry.py

@@ -800,7 +800,16 @@ def _plan_script_tasks_schema() -> dict[str, Any]:
                         "minItems": 1,
                         "maxItems": 50,
                     },
-                    "parent_task_id": {"type": ["string", "null"], "default": None},
+                    "parent_task_id": {
+                        "type": ["string", "null"],
+                        "default": None,
+                        "description": (
+                            "Parent Task that owns every new semantic scope. Structure, Paragraph "
+                            "and ElementSet inputs stay in one creative branch and are normally "
+                            "siblings under the same Compose parent; completed producer Tasks "
+                            "cannot receive new children."
+                        ),
+                    },
                 },
                 "required": ["contracts"],
                 "additionalProperties": False,
@@ -949,7 +958,17 @@ def _planner_task_input_schema() -> dict[str, Any]:
                             ],
                         },
                         "description": {"type": "string", "default": ""},
-                        "params": {"type": "object", "default": {}},
+                        "params": {
+                            "type": "object",
+                            "default": {},
+                            "description": (
+                                "Shapes by criterion_type: evidence_nonempty accepts optional "
+                                "integer minimum_count; source_attributable uses {}; "
+                                "semantic_relevance and concept_coverage require string array "
+                                "concepts and accept optional integer minimum_matches; freshness "
+                                "accepts optional integer max_age_days. Never use query."
+                            ),
+                        },
                         "hard": {"type": "boolean"},
                     },
                     "required": ["client_key", "criterion_type", "hard"],

script_build_host/tests/test_mission_workbench.py

@@ -8,7 +8,11 @@ from typing import Any
 import pytest
 from agent.orchestration import AgentRole, ArtifactRef, RoleContextRequest, TaskStatus
 
-from script_build_host.application.mission_workbench import MissionWorkbench, WorkbenchError
+from script_build_host.application.mission_workbench import (
+    MissionWorkbench,
+    WorkbenchError,
+    _fit_planner_state,
+)
 from script_build_host.domain.task_contracts import (
     ScriptCriterion,
     ScriptIntentClass,
@@ -16,7 +20,7 @@ from script_build_host.domain.task_contracts import (
     ScriptTaskContractV1,
     ScriptTaskKind,
 )
-from script_build_host.domain.workbench import NotModified, semantic_handle
+from script_build_host.domain.workbench import NotModified, PlannerState, semantic_handle
 
 
 class _TaskStore:
@@ -121,6 +125,37 @@ class _Candidates:
         }
 
 
+def test_planner_state_compacts_optional_frontiers_instead_of_failing() -> None:
+    state = PlannerState(
+        state_revision="ledger:1",
+        phase=2,
+        root={"task_id": "root", "status": "waiting_children", "objective": "compose"},
+        focused_task_id="root",
+        active_subgraph=tuple(
+            {"task_id": f"task-{index}", "objective": "x" * 500} for index in range(12)
+        ),
+        accepted_frontier=tuple(
+            {"decision_id": f"decision-{index}", "summary": "y" * 1_200}
+            for index in range(10)
+        ),
+        goal_progress={"total": 1, "covered": 0, "uncovered_goal_ids": ["goal"]},
+        counts_by_status={"completed": 10},
+        active_operations=(),
+        latest_failures=(),
+        retired_task_count=20,
+        active_total=12,
+        accepted_total=10,
+        context_bundle={"cards": [{"summary": "z" * 1_000}]},
+        phase_protocol={"phase": 2},
+    )
+
+    compacted = _fit_planner_state(state)
+
+    assert compacted.accepted_has_more is True
+    assert len(compacted.accepted_frontier) < len(state.accepted_frontier)
+    assert len(json.dumps(compacted.to_payload(), ensure_ascii=False)) <= 24_000
+
+
 def _root_task(*, children: list[str] | None = None) -> Any:
     return SimpleNamespace(
         task_id="root-task",

script_build_host/tests/test_phase_two_contracts.py

@@ -45,7 +45,18 @@ from script_build_host.infrastructure.task_contract_store import FileScriptTaskC
 
 
 def test_element_set_owns_only_elements_even_when_host_materializes_paragraph_endpoints() -> None:
-    assert _write_scope_for(ScriptTaskKind.ELEMENT_SET) == ("script-build://writes/elements",)
+    first = _write_scope_for(
+        ScriptTaskKind.ELEMENT_SET, scope_ref="script-build://scopes/compose/elements"
+    )
+    repeated = _write_scope_for(
+        ScriptTaskKind.ELEMENT_SET, scope_ref="script-build://scopes/compose/elements"
+    )
+    other = _write_scope_for(
+        ScriptTaskKind.ELEMENT_SET, scope_ref="script-build://scopes/other/elements"
+    )
+    assert first == repeated
+    assert first[0].startswith("script-build://writes/elements/scopes/")
+    assert first != other
 
 
 def test_execution_budget_counts_stage_intervals_not_planner_idle_time() -> None:
@@ -636,7 +647,7 @@ async def test_accepted_input_scope_is_rejected_before_worker_dispatch(tmp_path:
     structure = ScriptTaskContractV1.from_payload(_contract("structure", scope=structure_scope))
     paragraph_payload = _contract(
         "paragraph",
-        scope="script-build://direction/main/compose/candidate-1/paragraph/p1",
+        scope="script-build://direction/main/compose/candidate-2/paragraph/p1",
     )
     accepted_ref = {
         "decision_id": "accepted-structure",
@@ -652,7 +663,7 @@ async def test_accepted_input_scope_is_rejected_before_worker_dispatch(tmp_path:
     paragraph_payload["input_decision_refs"] = [accepted_ref]
     paragraph_payload["base_artifact_ref"] = accepted_ref["artifact_ref"]
     paragraph = ScriptTaskContractV1.from_payload(paragraph_payload)
-    producer_task = SimpleNamespace(task_id="structure-task")
+    producer_task = SimpleNamespace(task_id="structure-task", parent_task_id="compose-task")
     ledger = SimpleNamespace(
         decisions={
             "accepted-structure": SimpleNamespace(
@@ -674,14 +685,90 @@ async def test_accepted_input_scope_is_rejected_before_worker_dispatch(tmp_path:
         await service._guard_accepted_input_scopes(root, ledger, (paragraph,))
     assert raised.value.details == {
         "decision_id": "accepted-structure",
+        "producer_task_id": "structure-task",
+        "producer_parent_task_id": "compose-task",
         "consumer_kind": "paragraph",
         "consumer_scope": paragraph.scope_ref,
         "producer_kind": "structure",
         "producer_scope": structure.scope_ref,
         "source": "explicit",
+        "recovery": (
+            "create the consumer in the producer's creative branch, normally as a sibling under "
+            "producer_parent_task_id; changing input_decision_ids to base_decision_id does not "
+            "change scope"
+        ),
     }
 
 
+@pytest.mark.asyncio
+async def test_overlapping_write_scopes_are_dispatched_serially(tmp_path: Path) -> None:
+    service, _coordinator, root = await _service(tmp_path)
+    active = SimpleNamespace(task_id="active", parent_task_id="compose-a")
+    requested = SimpleNamespace(task_id="requested", parent_task_id="compose-b")
+    ledger = SimpleNamespace(
+        tasks={active.task_id: active, requested.task_id: requested},
+        operations={
+            "running": SimpleNamespace(
+                status=OperationStatus.RUNNING,
+                task_ids=[active.task_id],
+            )
+        },
+    )
+    active_payload = _contract("structure", scope="script-build://scopes/compose/structure")
+    active_payload["write_scope"] = ["script-build://writes/paragraphs"]
+    requested_payload = _contract(
+        "paragraph", scope="script-build://scopes/compose/paragraph"
+    )
+    requested_payload["write_scope"] = ["script-build://writes/paragraphs/scopes/p1"]
+    contracts = {
+        active.task_id: ScriptTaskContractV1.from_payload(active_payload),
+        requested.task_id: ScriptTaskContractV1.from_payload(requested_payload),
+    }
+
+    async def frozen_contract(_root_trace_id: str, task: Any) -> Any:
+        return SimpleNamespace(contract=contracts[task.task_id])
+
+    service.contract_for_task = frozen_contract  # type: ignore[method-assign]
+
+    with pytest.raises(TaskContractError, match="dispatched serially") as raised:
+        await service._guard_concurrent_write_scopes(root, ledger, [requested.task_id])
+    assert raised.value.code == "TASK_STATE_CONFLICT"
+
+
+@pytest.mark.asyncio
+async def test_candidate_workspace_writers_are_serial_even_with_distinct_scopes(
+    tmp_path: Path,
+) -> None:
+    service, _coordinator, root = await _service(tmp_path)
+    active = SimpleNamespace(task_id="active", parent_task_id="compose")
+    requested = SimpleNamespace(task_id="requested", parent_task_id="compose")
+    ledger = SimpleNamespace(
+        tasks={active.task_id: active, requested.task_id: requested},
+        operations={
+            "running": SimpleNamespace(
+                status=OperationStatus.RUNNING,
+                task_ids=[active.task_id],
+            )
+        },
+    )
+    active_payload = _contract("paragraph", scope="script-build://scopes/compose/p1")
+    active_payload["write_scope"] = ["script-build://writes/paragraphs/scopes/p1"]
+    requested_payload = _contract("paragraph", scope="script-build://scopes/compose/p2")
+    requested_payload["write_scope"] = ["script-build://writes/paragraphs/scopes/p2"]
+    contracts = {
+        active.task_id: ScriptTaskContractV1.from_payload(active_payload),
+        requested.task_id: ScriptTaskContractV1.from_payload(requested_payload),
+    }
+
+    async def frozen_contract(_root_trace_id: str, task: Any) -> Any:
+        return SimpleNamespace(contract=contracts[task.task_id])
+
+    service.contract_for_task = frozen_contract  # type: ignore[method-assign]
+
+    with pytest.raises(TaskContractError, match="candidate workspace"):
+        await service._guard_concurrent_write_scopes(root, ledger, [requested.task_id])
+
+
 class _BlockingExecutor:
     def __init__(self) -> None:
         self.started = asyncio.Event()

script_build_host/tests/test_phase_two_inputs_validation.py

@@ -42,6 +42,7 @@ from script_build_host.domain.artifacts import (
     EvidenceRecordV1,
 )
 from script_build_host.domain.input_compatibility import (
+    AcceptedInputCompatibilityError,
     AcceptedInputSource,
     validate_accepted_input,
 )
@@ -78,6 +79,29 @@ def test_candidate_portfolio_can_pin_the_active_direction_explicitly() -> None:
     )
 
 
+def test_explicit_creative_inputs_can_cross_sibling_scopes_in_one_branch() -> None:
+    validate_accepted_input(
+        consumer_kind=ScriptTaskKind.PARAGRAPH,
+        consumer_scope="script-build://scopes/direction/portfolio/compose/paragraph",
+        producer_kind=ScriptTaskKind.STRUCTURE,
+        producer_scope="script-build://scopes/direction/portfolio/compose/structure",
+        source=AcceptedInputSource.EXPLICIT,
+        same_parent=True,
+    )
+
+
+def test_explicit_creative_inputs_cannot_cross_candidate_branches() -> None:
+    with pytest.raises(AcceptedInputCompatibilityError, match="outside the consumer scope"):
+        validate_accepted_input(
+            consumer_kind=ScriptTaskKind.PARAGRAPH,
+            consumer_scope="script-build://scopes/direction/portfolio/compose-a/paragraph",
+            producer_kind=ScriptTaskKind.STRUCTURE,
+            producer_scope="script-build://scopes/direction/portfolio/compose-b/structure",
+            source=AcceptedInputSource.EXPLICIT,
+            same_parent=False,
+        )
+
+
 def _contract(
     *,
     kind: ScriptTaskKind = ScriptTaskKind.PARAGRAPH,
@@ -349,7 +373,7 @@ async def test_mid_retrieval_replacement_freezes_new_evidence_and_old_candidate(
     # The old accepted Paragraph remains a completed sibling.  The replacement
     # consumer gets new Evidence from its own direct Retrieval child and imports
     # the immutable old Decision explicitly; it never reads a mutable "latest".
-    ledger.tasks["child"].parent_task_id = "root-task"
+    ledger.tasks["child"].parent_task_id = "other-branch"
     ledger.tasks["root-task"].child_task_ids = ["child", "parent"]
     replacement_task = _task(
         "parent", parent="root-task", kind="paragraph", status=TaskStatus.RUNNING
@@ -563,7 +587,7 @@ async def test_direct_child_is_revalidated_against_consumer_scope_and_kind() ->
 async def test_explicit_input_is_revalidated_against_consumer_scope() -> None:
     resolver, ledger, attempt = _closed_fixture()
     child_ref = ledger.attempts["attempt-child"].submission.artifact_refs[0]
-    ledger.tasks["child"].parent_task_id = "root-task"
+    ledger.tasks["child"].parent_task_id = "other-branch"
     ledger.tasks["parent"].child_task_ids = []
     ledger.attempts[attempt.attempt_id] = replace(
         attempt,

script_build_host/tests/test_tool_failure_bridge.py

@@ -58,6 +58,26 @@ def _knowledge_off():
             ScriptBuildError("TASK_KIND_PRESET_MISMATCH", "Task kind cannot change"),
             FailureDisposition.REPLAN_TASK,
         ),
+        (
+            ScriptBuildError("TASK_CAPABILITY_MISMATCH", "criterion params are invalid"),
+            FailureDisposition.REPLAN_TASK,
+        ),
+        (
+            ScriptBuildError("CONTEXT_HANDLE_UNAUTHORIZED", "outside task"),
+            FailureDisposition.RETRY_CALL,
+        ),
+        (
+            ScriptBuildError("CONTEXT_CURSOR_INVALID", "cursor is stale"),
+            FailureDisposition.RETRY_CALL,
+        ),
+        (
+            ScriptBuildError("CONTEXT_CURSOR_SCOPE_MISMATCH", "cursor belongs to another query"),
+            FailureDisposition.RETRY_CALL,
+        ),
+        (
+            ScriptBuildError("WRITE_SCOPE_CONFLICT", "choose one candidate"),
+            FailureDisposition.REPLAN_TASK,
+        ),
         (
             PhaseTwoCandidateError("LEGACY_WRITE_INVALID", "workspace is invalid"),
             FailureDisposition.REPAIR_ATTEMPT,

思路.md

@@ -0,0 +1,2664 @@
+结论先说：
+
+1. 你们的系统确实比旧版“更像强化学习的外形”，但严格来说不是强化学习。
+2. 对没有绝对标准的创作任务，最优解不是二选一：应当用你们当前架构负责收敛、追踪和验收，再吸收旧版“多路径探索、动态改目标”的能力。
+3. 每个 Task 不需要一律再制定详细 Steps。你们已经有 Task 级计划；复杂 Task 才需要一份可动态修改的“微计划”。
+
+以下判断基于当前本地 `main@d854ce9` 工作区；仓库存在较多未提交改动，所以说的是你现在磁盘上的代码。
+
+## 一、为什么说像强化学习，但不是强化学习
+
+你们现在可以这样类比：
+
+| 强化学习概念 | 你们当前实现 |
+|---|---|
+| 状态 State | `TaskLedger`、已接受 Artifact、当前 Validation |
+| 行动 Action | 新建 Task、dispatch、accept、repair、retry、revise、split |
+| 环境 Environment | Worker、工具、Attempt workspace |
+| 反馈 Reward | Validator 的 passed/failed、criteria 和 defects |
+| 状态转移 | Coordinator 更新 Task 状态和任务树 |
+| 策略 Policy | Planner 根据当前状态选择下一动作 |
+
+真实循环就是：
+
+`Task → Worker Attempt → Artifact Snapshot → Validator → Planner Decision → 下一状态`
+
+代码文档也明确如此：[orchestration-v1.md](/Users/samlee/Documents/works/脚本构建第三版-agent老框架/agent/agent/docs/orchestration-v1.md:3)。
+
+但它缺少强化学习最关键的一步：
+
+> Validator 的反馈没有更新 Planner 模型参数，也没有形成一个跨任务持续学习的 policy。
+
+它只是这一次运行里“看到反馈后重新推理”。所以更准确的名字是：
+
+> 基于评价器反馈的在线搜索 / 反馈控制 / evaluator-guided planning。
+
+如果未来你们把大量 Build 的 Task、Attempt、Validation、人工最终选择积累下来，用于更新 Prompt、偏好模型、奖励模型或微调 Planner，那么才会真正向 RL、offline RL 或 preference learning 靠近。
+
+旧版则更像：
+
+> 目标树搜索 + 多分支候选搜索 + 选择/合并。
+
+所以二者的区别大致是：
+
+- 旧版更像树搜索或进化式探索。
+- 你们现在更像带 Validator 的状态机控制与规划。
+
+## 二、发散型创作，到底哪种好
+
+客观说：
+
+| 维度 | 旧版：动态目标树、纵向深挖 | 你们：冻结目标、动态 Task 图 |
+|---|---|---|
+| 创意发散 | 更强，容易发现原本没想到的方向 | 较弱，容易被最初 Direction 锁死 |
+| 动态适应 | 可以直接改目标、加子目标 | 主要修改实现 Task，Goal 原则上冻结 |
+| 稳定收敛 | 较弱，可能越做越偏 | 较强，始终围绕同一组 Goal |
+| 可验收性 | 容易目标和评价一起漂移 | 很强，criteria、Snapshot、Decision 都可追踪 |
+| 局部与整体 | 更容易保持围绕一个业务目标纵向推进 | 容易每个 Task 都合格，但拼起来不够好 |
+| 工程治理 | 状态复杂，容易失控 | 状态边界和责任非常清楚 |
+
+你们当前 Direction 会生成通常 3–6 个顶层 Goal、最多两层，并要求每个 Goal 有成功标准：[direction_worker.md](/Users/samlee/Documents/works/脚本构建第三版-agent老框架/script_build_host/src/script_build_host/agents/prompts/direction_worker.md:5)。之后执行阶段遵循“Goal 冻结、Task 图动态调整”：[script_planner.md](/Users/samlee/Documents/works/脚本构建第三版-agent老框架/script_build_host/src/script_build_host/agents/prompts/script_planner.md:27)。
+
+这非常适合生产系统，但有两个创作风险：
+
+- 一开始 Direction 定错了，后面会非常认真地把错误方向做完。
+- 每个 Paragraph、Structure、ElementSet 都通过局部 Validator，最终脚本仍可能缺少整体惊喜、节奏或感染力。
+
+因此我的建议不是退回旧版，而是做“双层 Loop”：
+
+```mermaid
+flowchart TD
+    A["冻结输入"] --> B["生成 2-4 个实质不同的 Direction 候选"]
+    B --> C["整体比较：目标、受众价值、独特性、可实现性"]
+    C --> D["选择并冻结 Direction v1"]
+
+    D --> E["动态规划最小 Task"]
+    E --> F["可选微计划"]
+    F --> G["Worker Attempt"]
+    G --> H["局部 Validator"]
+    H -->|"局部缺陷"| E
+    H -->|"通过"| I["ACCEPT / 组合"]
+
+    I --> J["完整脚本全局评价"]
+    J -->|"实现问题"| E
+    J -->|"方向根本错误"| K["受控生成 Direction v2"]
+    K --> E
+    J -->|"通过或预算耗尽"| L["发布 / 人工选择"]
+```
+
+也就是：
+
+- 前期宽：多方向、多候选、允许发散。
+- 中期窄：Direction 一旦选定，使用你们当前严格 Task Loop。
+- 后期整体看：不能只做局部 Task 验收，还要重新评价完整脚本。
+- 只有全局评价证明“方向本身有问题”，才允许生成 `Direction v2`，不能因为某段不好就随便移动目标。
+
+这也符合各家公开经验：
+
+- Anthropic 认为 evaluator–optimizer 特别适合“评价标准清楚、反馈确实能让内容变好”的任务，并直接把它类比为写作修改过程；无法预先知道子任务时，则适合 orchestrator–workers。[Anthropic：Building effective agents](https://www.anthropic.com/engineering/building-effective-agents)
+- Google 推荐创作类复杂生成使用 iterative refinement / generator–critic，但强调必须有质量阈值、最大轮数等退出条件，否则会无限循环并持续增加成本。[Google Cloud：Agentic AI design patterns](https://docs.cloud.google.com/architecture/choose-design-pattern-agentic-ai-system)
+- Anthropic 在处理“审美没有绝对答案”的设计任务时，采用 planner–generator–evaluator，关键不是假装审美客观，而是把主观判断拆成具体、可评分的维度。[Anthropic：Harness design for long-running application development](https://www.anthropic.com/engineering/harness-design-long-running-apps)
+- AlphaEvolve 的“广泛生成—自动评价—保留优秀候选”很成功，但 Google 特别指出它适合能够自动、量化验证的问题。脚本创作没有这么强的客观评价器，因此不能简单复制成单一 reward 分数。[Google DeepMind：AlphaEvolve](https://deepmind.google/blog/alphaevolve-a-gemini-powered-coding-agent-for-designing-advanced-algorithms/)
+
+## 三、每个 Task 需不需要 Plan
+
+你们已经有第一层 Plan。
+
+`TaskSpec` 已经包含 objective、acceptance criteria、context refs：[models.py](/Users/samlee/Documents/works/脚本构建第三版-agent老框架/agent/agent/orchestration/models.py:281)。业务层 `PlannerTaskInput` 还有 task_kind、goal_ids、输入 Decision 和 criteria：[task_contracts.py](/Users/samlee/Documents/works/脚本构建第三版-agent老框架/script_build_host/src/script_build_host/domain/task_contracts.py:202)。
+
+这其实就是：
+
+> 做什么、为什么做、使用什么输入、做到什么算完成。
+
+当前缺少的是 Worker 内部的“怎么做”。数据库兼容层虽然写了 `script_build_task_plan_step`，但实际上每个 branch 只生成一个 `step_key="1"`，标题就是 Task objective，不是真正的多步骤计划：[workspace.py](/Users/samlee/Documents/works/脚本构建第三版-agent老框架/script_build_host/src/script_build_host/repositories/workspace.py:1183)。
+
+我的建议是：不要每个 Task 强制 Plan，而是按复杂度触发。
+
+| Task 类型 | 是否需要额外微计划 |
+|---|---|
+| 单次 Retrieval | 不需要，步骤已经确定 |
+| Structure | 通常不需要额外 Plan，因为 Structure 本身就是创作计划 |
+| 单段 Paragraph | 通常不需要 |
+| 多段 Paragraph / 大范围改写 | 需要 |
+| ElementSet | 简单场景不需要；涉及多个 Paragraph 映射时需要 |
+| Compare | 需要先冻结比较维度和取舍顺序 |
+| Compose | 需要候选采用与组合计划 |
+| Root Delivery | 需要全局检查清单，但不需要详细创作 Steps |
+
+触发微计划的条件可以很简单，满足任一条件才生成：
+
+- 预计需要三个以上动作。
+- 有多个输入候选需要取舍。
+- 要同时覆盖多个 Goal。
+- 中间结果会改变后续路径。
+- Task 跨较长上下文或可能换 Worker。
+- 失败代价较高。
+
+微计划不要写成几十步，只保存四类信息：
+
+```text
+1. 本 Attempt 准备采用什么方法
+2. 准备覆盖哪些 Goal / criteria
+3. 需要读取或生成哪些中间结果
+4. 什么观察结果会触发调整或停止
+```
+
+每次工具调用后也不必重写 Plan。只有下面三种情况才调整：
+
+- 原假设被新证据推翻。
+- Validator 指出的是路径性缺陷，而不是局部措辞问题。
+- 当前路径连续两次没有质量提升。
+
+OpenAI 的公开建议也是两面性的：复杂任务应拆成更小、更明确的步骤，但同时建议从简单架构开始；较新的模型提示词还应保持精简，只在代表性 eval 证明有效时增加规划脚手架。[OpenAI：A practical guide to building agents](https://openai.com/business/guides-and-resources/a-practical-guide-to-building-ai-agents/)、[OpenAI API：Model guidance](https://developers.openai.com/api/docs/guides/latest-model)。Anthropic 也建议不要 rigid 地考核固定执行路径，而应主要评价最终产物，否则会惩罚模型发现的有效创意路径。[Anthropic：Demystifying evals for AI agents](https://www.anthropic.com/engineering/demystifying-evals-for-ai-agents)
+
+最终架构上，我会选择：
+
+> 多 Direction 候选的发散外环 + 冻结版本化目标 + 当前动态 Task/Validator 内环 + 按需微计划 + 最终全局脚本评价。
+
+这样既保留旧版的创造性，又保留你们现在最有价值的可追溯、可验收和可收敛能力。真正优先级最高的改造不是“给所有 Task 加 Steps”，而是“增加方向候选和全局评价，并允许有证据的受控 Direction 升版”。
+
+
+## ************************************************************************
+
+我们的路线，有没有可能最终创作效果更好，更符合比如，人设，更有创意。 （假设工程底子一样）只思考系统的路线和流程
+
+真要往强化学习应该很难做吧。是不是就像sft和rl，让系统自己学会效果更好啊？
+
+还是我们更多给他目标，你搜索各种顶级模型或者agent怎么做的， 顶会论文， openai， claude，deepmind等
+
+帮我想想本项目的路线出发怎么做才效果好
+
+
+我的判断是：
+
+> 你们这条路线的“创作上限”确实可能比旧版更高，但当前流程还没有自动获得这个优势。
+> 它现在拥有一套很好的治理骨架，却还需要补上真正的“创意搜索、人设控制和跨稿学习”。
+
+如果补对了，它不仅会更稳定，还可能做到：
+
+- 更符合账号人设；
+- 更少模板化；
+- 有更丰富的创作方向；
+- 整稿比单次大模型直出更好；
+- 随着真实反馈积累，逐渐更懂某个账号。
+
+但重点不是马上做 RL，而是先把“搜索—比较—修订—学习”这套系统做完整。
+
+## 一、你们目前的优势和缺口
+
+当前系统已经有：
+
+- 冻结输入；
+- `persona_points`；
+- Direction Goal；
+- 动态 Task；
+- 独立 Attempt；
+- Validator；
+- Planner 根据反馈调整；
+- 最终 Root 全文验收。
+
+代码上，人设已经进入冻结输入：[input_snapshot.py](/Users/samlee/Documents/works/脚本构建第三版-agent老框架/script_build_host/src/script_build_host/domain/input_snapshot.py:8)，Root 也把“人设一致性”列为验收标准：[mission_factory.py](/Users/samlee/Documents/works/脚本构建第三版-agent老框架/script_build_host/src/script_build_host/application/mission_factory.py:70)。
+
+但目前人设主要还是一个“验收维度”：
+
+- Candidate Validator 的 D5 把 topic、persona、受众放在一起检查：[candidate_validator.md](/Users/samlee/Documents/works/脚本构建第三版-agent老框架/script_build_host/src/script_build_host/agents/prompts/candidate_validator.md:10)。
+- Root Validator 也只是把 persona 和 Paragraph、tone、strategy 一起检查：[root_validator.md](/Users/samlee/Documents/works/脚本构建第三版-agent老框架/script_build_host/src/script_build_host/agents/prompts/root_validator.md:7)。
+
+这可以阻止明显跑偏，但还不足以让作品“像这个人写的”。
+
+你们现在最大的三个缺口是：
+
+1. 人设只有检查，没有变成真正的创作控制器。
+2. Direction 内有 3–6 个 Goal，但这些是一个方向的组成部分，不是多个不同创意方向。
+3. Validator 主要负责淘汰错误，还没有承担“从多个好稿中选择更有趣的一稿”的职责。
+
+所以当前路线会提高：
+
+- 完整性；
+- 一致性；
+- 可追溯性；
+- 减少低级错误。
+
+但不一定自然提高：
+
+- 独特性；
+- 惊喜感；
+- 人格魅力；
+- 表达的新鲜程度。
+
+## 二、你们是不是应该走强化学习
+
+先把几个概念分开：
+
+| 方法 | 本质 | 适合解决 |
+|---|---|---|
+| Prompt / Agent Loop | 本次运行中调整上下文和行动 | 快速改善流程 |
+| Verbal Learning / Memory | 保存过去的错误和经验，下次注入 | 跨运行逐渐变好，但不改模型权重 |
+| SFT | 给模型大量好示范，让它模仿 | 格式、话术、人设基础风格 |
+| DPO/偏好优化 | 给两个结果，告诉模型更喜欢哪个 | 细微风格、用户偏好、稿件选择 |
+| RL/RLHF | 用奖励反复采样并更新策略 | 长程决策、可稳定评价的复杂行为 |
+
+你们目前更接近 NeurIPS 2023 的 Reflexion：
+
+> 不更新模型权重，而是把失败反馈转成语言经验，保存在记忆里，影响后续尝试。
+
+这篇论文甚至称之为“Verbal Reinforcement Learning”，但明确不修改模型参数。[Reflexion，NeurIPS 2023](https://papers.nips.cc/paper_files/paper/2023/hash/1b44b878bb782e6954cd888628510e90-Abstract-Conference.html)
+
+所以可以把你们的近期目标理解成：
+
+> 先做“系统层面的强化学习”，以后再考虑“模型权重层面的强化学习”。
+
+### 真正做 RL 为什么难
+
+创作任务的奖励很难定义。
+
+数学、代码可以判断：
+
+- 答案对不对；
+- 测试过不过；
+- 编译成不成功。
+
+但脚本的“更好”包含：
+
+- 是不是像这个账号；
+- 是否有新意；
+- 是否自然；
+- 是否有传播力；
+- 是不是高级而非故作高级；
+- 某个受众是否喜欢。
+
+这些维度不仅主观，还可能互相冲突。
+
+OpenAI 的研究已经证明：Reward Model 只是人类偏好的代理，过度优化这个代理，真实质量反而会下降，也就是 Goodhart’s Law。[OpenAI：Reward model overoptimization](https://openai.com/index/scaling-laws-for-reward-model-overoptimization/)
+
+更危险的是，普通 RLHF 往往优化“平均人的偏好”，而平均偏好通常偏爱：
+
+- 流畅；
+- 安全；
+- 熟悉；
+- 表面完整；
+- 容易理解。
+
+这可能把创意磨平。ICML 2024 的 Quality Diversity through Human Feedback 就指出，平均偏好优化不适合需要多样化输出的生成任务。[QDHF，ICML 2024](https://proceedings.mlr.press/v235/ding24h.html)
+
+OpenAI 的 o1 路线也说明了这个边界：RL 对数学、代码、科学推理提升明显，但官方的人类评测中，o1 在某些自然语言任务上并不比 GPT‑4o 更受偏好。[OpenAI：Learning to reason with LLMs](https://openai.com/index/learning-to-reason-with-llms/)
+
+因此：
+
+> RL 能让模型更会解题，不等于更会创作；更长的推理也不等于更符合人设。
+
+ACL 2025 甚至发现，普通 CoT 和 reasoning-optimized 模型可能降低角色扮演表现。[Reasoning Does Not Necessarily Improve Role-Playing Ability](https://aclanthology.org/2025.findings-acl.537/)
+
+## 三、最适合本项目的路线：三个嵌套 Loop
+
+我建议把项目设计成三个 Loop，而不是一个巨大 Loop。
+
+### Loop A：创意方向搜索
+
+这是当前最需要加强的部分。
+
+```text
+冻结输入
+  ↓
+编译 Persona Contract
+  ↓
+生成多个实质不同的创意假设
+  ↓
+聚类、去重、保留差异
+  ↓
+事实与约束过滤
+  ↓
+Persona 适配评价
+  ↓
+候选两两比较
+  ↓
+保留主方向 + 冒险方向
+```
+
+注意：不是一个 Direction 里面多放几个 Goal。
+
+应该是：
+
+```text
+Direction Portfolio
+├── Direction A：从人物冲突出发
+├── Direction B：从反常识事实出发
+├── Direction C：从具体生活场景出发
+└── Direction D：更冒险的叙事形式
+```
+
+然后每个 Direction 内部再有自己的 3–6 个 Goal。
+
+这和 DeepMind Co-Scientist 的路线非常相似：
+
+- Generation：生成大量假设；
+- Proximity：聚类，保证探索空间有差异；
+- Reflection：检查正确性、质量和新颖性；
+- Ranking：做 pairwise tournament；
+- Evolution：组合和发展优秀方向；
+- Supervisor：动态安排探索资源。
+
+这不是依赖一次模型“想出最佳答案”，而是在系统层面建立创意进化机制。[Google DeepMind：Co-Scientist](https://deepmind.google/blog/co-scientist-a-multi-agent-ai-partner-to-accelerate-research/)
+
+对本项目，可以先控制在：
+
+- 生成 6–10 个粗方向；
+- 聚类后保留 3–4 个实质不同方向；
+- 最终选择一个主方向；
+- 同时保留一个 wildcard 冒险方向进入成稿候选。
+
+这样可以防止 Validator 永远选择最稳妥、最模板化的结果。
+
+### Loop B：单个方向的脚本构建
+
+这个 Loop 主要沿用你们现在的架构。
+
+```text
+Direction
+  ↓
+详细结构规划
+  ↓
+动态 Task 图
+  ↓
+Worker 生成候选
+  ↓
+局部 Validator
+  ↓
+定点修改
+  ↓
+Compose
+  ↓
+整稿 Validator
+```
+
+研究证据很强：
+
+- Re3 使用“总体计划＋当前故事状态＋多个续写候选重排＋一致性编辑”，长故事的情节连贯性和主题相关性明显提升。[Re3，EMNLP 2022](https://aclanthology.org/2022.emnlp-main.296/)
+- DOC 使用分层详细大纲和生成过程控制，人类评价中的 plot coherence、outline relevance、interestingness 都显著超过 Re3。[DOC，ACL 2023](https://aclanthology.org/2023.acl-long.190/)
+- Self-Refine 不训练模型，仅靠生成—反馈—修订，在多个任务上平均获得约 20% 的绝对提升。[Self-Refine，NeurIPS 2023](https://papers.neurips.cc/paper_files/paper/2023/hash/91edff07232fb1b55a505a9e9f6c0ff3-Abstract-Conference.html)
+
+但需要做一个重要的角色分工：
+
+| 角色 | 应该怎么工作 |
+|---|---|
+| Planner | 强推理，分析目标、缺口、依赖和资源 |
+| Structure Worker | 制定分层结构和叙事推进 |
+| Paragraph Writer | 少做显式规划，直接在紧凑人设和创作约束下写 |
+| Validator | 分维度评价，不参与创作 |
+| Reviser | 只改明确缺陷，保护已有亮点 |
+| Root Validator | 按真实阅读顺序检查整稿 |
+
+尤其不要让 Paragraph Worker 写很长的计划和 CoT。
+
+> 规划能力应主要放在 Planner/Structure；写作者应保留语言生成自由度。
+
+否则很容易出现“逻辑解释得很清楚，但作品没有生命力”。
+
+## 四、把人设从一个字段升级成 Persona Contract
+
+`persona_points` 不应该原样扔给所有 Worker。
+
+先增加一个“Persona Compiler”，把原始人设加工成稳定合同：
+
+```text
+Persona Contract
+├── 核心身份：他是谁、服务谁
+├── 价值排序：什么最重要，冲突时如何选择
+├── 观点边界：哪些观点会说，哪些不会说
+├── 叙事习惯：从事实、经历、冲突还是观点切入
+├── 语言指纹：句式、词汇、节奏、幽默方式
+├── 情感表达：克制、热烈、讽刺、共情
+├── 正样例：哪些历史内容最像他
+├── 反样例：哪些表达绝对不像他
+└── 当前选题下的立场推演
+```
+
+RoleLLM 的路线也是：
+
+1. Role Profile Construction；
+2. 根据角色上下文生成知识和指令；
+3. Role Prompting；
+4. 数据足够后再进行 Role-Conditioned Instruction Tuning。
+
+而不是只在 Prompt 里放一句“你是某某人”。[RoleLLM，ACL Findings 2024](https://aclanthology.org/2024.findings-acl.878/)
+
+还有一个关键问题：人设会随着长上下文逐渐淡化。EACL 2026 的研究发现，在长对话和目标导向任务中，Persona fidelity 会持续下降，并逐渐回到普通模型的默认表达。[Persistent Personas，EACL 2026](https://aclanthology.org/2026.eacl-long.246/)
+
+所以每个创作 Attempt 不应读取全部历史，而应重新注入一个紧凑 Persona Capsule：
+
+```text
+本段必须体现的 2 个 persona 特征
+本段不能出现的 2 个反人设表达
+当前叙事位置下该账号的自然反应
+需要保留的语言指纹
+```
+
+## 五、Validator 要拆成硬标准、人格和创意三种判断
+
+不要让一个总分决定所有东西。
+
+### 硬标准
+
+适合 PASS/FAIL：
+
+- 事实是否有依据；
+- 是否满足格式；
+- Goal 是否覆盖；
+- 是否有 placeholder；
+- 是否违反 constraints；
+- Paragraph/Element/Link 是否完整。
+
+### 人设标准
+
+适合 pairwise 比较：
+
+- 哪稿更像这个账号；
+- 观点是否符合价值排序；
+- 语言习惯是否自然；
+- 是否只是表面模仿口头禅；
+- 换成另一个账号后是否仍然完全成立。
+
+一个很有效的测试是：
+
+> 隐去账号名字，给评审多个账号 Persona，让它判断这篇稿最可能属于谁。
+
+如果经常无法识别，说明人设只存在于标签里，没有进入文本行为。
+
+### 创意标准
+
+不要问一个模糊的“是否有创意”，拆成：
+
+- Novelty：与账号历史内容是否不同；
+- Surprise：是否产生合理但非预期的变化；
+- Diversity：候选之间是否真的不同；
+- Relevance：创意是否服务选题；
+- Coherence：惊喜是否破坏整体逻辑；
+- Memorability：是否存在具体可记忆点。
+
+创意优先用候选两两比较，不要急着算统一总分：
+
+```text
+A 和 B 哪个更新鲜？
+A 和 B 哪个更像账号？
+A 和 B 哪个更适合这次选题？
+两者各自最值得保留的东西是什么？
+```
+
+最后采用 Pareto 选择，而不是一个总分：
+
+- 人设强、创意中；
+- 创意强、人设中；
+- 稳定性强；
+- 综合最优。
+
+## 六、修改时必须“保护亮点”
+
+普通 evaluator–optimizer 容易越改越平。
+
+因此 Validator 反馈需要同时返回：
+
+```text
+defects_to_fix
+strengths_to_preserve
+creative_commitments
+persona_invariants
+```
+
+例如：
+
+```text
+要修：
+- 中段观点重复
+- 转折缺少因果
+
+必须保护：
+- 开场的电梯场景
+- “看起来在省钱，实际在买焦虑”的反转
+- 账号克制、不说教的语气
+```
+
+Reviser 只能修指定缺陷，不允许默认重写全文。
+
+只有以下情况才允许整稿重构：
+
+- Direction 根本错误；
+- 结构无法承载全部 Goal；
+- 人设与选题视角根本冲突；
+- 局部修订连续两次没有提升。
+
+## 七、真正的跨运行“学习路线”
+
+建议按这个顺序走：
+
+### 第一阶段：Inference-time Search
+
+先不训练：
+
+- 多 Direction；
+- 多稿候选；
+- pairwise comparison；
+- Persona Contract；
+- 局部修订；
+- Root 整稿评价。
+
+这是当前最容易获得质量提升的阶段。
+
+### 第二阶段：Verbal Learning
+
+每次 Build 结束记录：
+
+- 哪个方向被选；
+- 哪些候选被淘汰；
+- 人工为什么改；
+- 哪些 Validator 判断后来被人否定；
+- 哪些创意受到用户喜欢；
+- 哪些表达被认为“不像账号”。
+
+把稳定经验加工成账号级 Memory：
+
+```text
+这个账号偏爱什么
+经常拒绝什么
+哪些开头形式已经用烂
+什么样的反转对他有效
+哪些 Validator 规则经常误判
+```
+
+这就是最适合你们当前架构的“轻量强化学习”。
+
+### 第三阶段：SFT
+
+当积累了足够多高质量“输入—成稿—人工修改”数据后：
+
+- 用 SFT 教模型稳定输出结构；
+- 学账号常用叙事方式；
+- 学人设语言习惯；
+- 学怎么把结构稿写成成稿。
+
+SFT 更像教它“怎么写”。
+
+### 第四阶段：偏好优化
+
+积累同一输入下的候选比较：
+
+```text
+A vs B
+用户选择 A
+原因：更像账号，但保留 B 的开场
+```
+
+再做 DPO、IPO 或专门的 preference model。
+
+偏好优化更像教它“两个都会写时，应该选哪种”。
+
+重点是：
+
+- 按账号或账号簇训练；
+- 不要把所有人的偏好平均在一起；
+- 保留少数派和 wildcard 创意；
+- 同时优化 quality 和 diversity。
+
+### 第五阶段：有限 RL
+
+如果以后真做 RL，我建议只先训练相对可验证的过程能力：
+
+- 是否完整读取输入；
+- 是否正确覆盖 Goal；
+- 是否遵守 Persona Contract；
+- 是否发现前后矛盾；
+- 是否选择了有效工具；
+- 是否能根据缺陷做定点修订；
+- 是否在质量不再提升时停止。
+
+不要一开始就用一个“创作质量 0–100 分”训练整个模型。
+
+## 最终建议
+
+从本项目路线出发，我会定成：
+
+> 先让系统学会“搜索好创意、识别好创意、保护好创意”，然后才让模型权重“更容易直接写出好创意”。
+
+对应架构就是：
+
+```text
+创意方向 Portfolio
+        ↓
+Persona Contract
+        ↓
+方向聚类与两两竞赛
+        ↓
+主方向 + wildcard
+        ↓
+详细结构规划
+        ↓
+动态 Task / Attempt / Validation
+        ↓
+缺陷修订 + 亮点保护
+        ↓
+整稿人设、创意、叙事复核
+        ↓
+人工选择和修改
+        ↓
+账号级经验记忆
+        ↓
+未来 SFT / 偏好优化 / 有限 RL
+```
+
+一句话评价：
+
+> 你们当前路线的骨架是对的，而且上限可能高于单纯动态目标树；真正决定最终效果的，不是 Task 状态机本身，而是能不能在“冻结目标之前充分发散”，在“每次修订时保护差异”，并把真实的人类选择沉淀成下一次可用的经验。
+
+
+
+## *************************************************************************
+
+并不是要百分百按照老版来，我是在探索和老版不同的实现创作构建的方法，想让他能有创造力，能很好完成最初的创作目标（比如要做反差感），能真正创作，能符合人设和事实，能有有趣的内容。
+
+你能搜到这块的一些，顶尖ai公司或者论文的实践吗，再来评估
+
+你问得对：**如果只说“总目标冻结、实现策略动态调整”，那建议版和老版确实基本一样。**上一张图没有把真正差异画清楚。
+
+真正区别不是“调不调整策略”，而是：
+
+> **老版把策略当成 GoalStep 对应的局部 Patch 分支；建议版把策略当成完整创作假设，先整体竞争，选中后再动态执行。**
+
+## 先回答两个问题
+
+### 1. 老版整体目标生成后会冻结吗？
+
+设计上：**会冻结。**
+
+老版提示词明确要求：
+
+> 总目标一次写入、全程冻结，不得修改、放松、重解释。
+
+见 [script_build_system_v2.md](</Users/samlee/Documents/works/脚本构建第二版/aiddit/pattern/patter_from_global_and_build/prompts/script/script_build_system_v2.md:28>)。
+
+但实现上只是**软冻结**，不是代码硬冻结。
+
+[save_script_direction](</Users/samlee/Documents/works/脚本构建第二版/aiddit/pattern/patter_from_global_and_build/script_build_agent_tools.py:4469>) 没有检查是否已经存在，直接：
+
+```python
+record.script_direction = direction_str
+```
+
+也就是说：
+
+- Prompt 规定不能改。
+- 数据库工具实际上允许覆盖。
+- 所以老版是“行为约束冻结”，不是“系统硬冻结”。
+
+当前新版则是真正硬冻结：一旦绑定了一个 Direction，再绑定另一个 Artifact 会直接抛出协议错误。见 [sqlalchemy.py](</Users/samlee/Documents/works/脚本构建第三版-agent老框架/script_build_host/src/script_build_host/repositories/sqlalchemy.py:366>)。
+
+---
+
+### 2. 建议版还动态调整实现策略吗？
+
+**当然调整，而且有两层调整。**
+
+```text
+第一层：宏观创作策略
+例如：
+A. 价格预期反转
+B. 博主自我认知反转
+C. 观众投票反转
+
+第二层：选中策略后的执行办法
+例如：
+- 先补事实材料
+- 调整反转出现的位置
+- 重写前15秒
+- 增加铺垫
+- 删除解释过度的段落
+```
+
+所以建议版不是：
+
+```text
+选择一个策略
+    ↓
+定死
+    ↓
+一路执行到底
+```
+
+而是：
+
+```text
+固定创作目标
+    ↓
+动态产生多个创作策略
+    ↓
+选择一个策略写整篇
+    ↓
+评估
+   ↙  ↓  ↘
+改执行  补材料  换整个策略
+```
+
+## 老版和建议版真正的区别
+
+### 老版：逐块优化，再合并进 Base
+
+```text
+冻结创作总目标
+        │
+        ▼
+动态 GoalStep 树
+        │
+        ▼
+选择一个局部目标
+“把开头反差做出来”
+        │
+   ┌────┴────┐
+   ▼         ▼
+路径A       路径B
+高价开头    专家判断开头
+   │         │
+ Patch A    Patch B
+   └────┬────┘
+        ▼
+选择/组合/合并进 Base
+        │
+        ▼
+下一轮再解决中段或结尾
+        │
+        ▼
+整篇评估后修改 GoalStep
+```
+
+老版的动态性非常强：
+
+- `GoalStep` 可以增加、改写、放弃，见 [plan_goal_steps](</Users/samlee/Documents/works/脚本构建第二版/aiddit/pattern/patter_from_global_and_build/script_build_agent_tools.py:2768>)。
+- 每轮可以重新制定多路径。
+- 竞争路径必须绑定不同的兄弟 GoalStep。
+- 每个路径产生独立 Patch。
+- 主 Agent 决定选择、组合或合并。
+
+见 [record_multipath_plan](</Users/samlee/Documents/works/脚本构建第二版/aiddit/pattern/patter_from_global_and_build/script_build_agent_tools.py:2201>)。
+
+它的思路是：
+
+> **一边写 Base，一边围绕局部问题开分支，然后把好的局部增量合进去。**
+
+---
+
+### 建议版：完整创作方案先竞争，再进入执行
+
+```text
+冻结 CreativeContract
+“必须有反差、符合人设、不编事实”
+        │
+        ▼
+产生完整创作方案
+   ┌────┼────┐
+   ▼    ▼    ▼
+方案A  方案B  方案C
+价格   自我   观众
+反转   反转   参与
+   │    │    │
+   ▼    ▼    ▼
+各自完整的：
+铺垫→反转→解释→结尾
+   └────┼────┘
+        ▼
+比较整套创作机制
+        ▼
+选择方案B
+        │
+        ▼
+一个主笔写完整草稿
+        │
+        ▼
+动态 TaskLedger
+   ┌────┼────────┐
+   ▼    ▼        ▼
+补事实  改开头   调节奏
+   └────┼────────┘
+        ▼
+整篇再次评估
+   ↙          ↘
+局部不好      方案本身不好
+继续改Task    切换A/生成D
+```
+
+它的思路是：
+
+> **先比较“整篇准备怎么创造反差”，选中后再开始深入写；不要一开始就把局部 Patch 合进 Base。**
+
+## 用同一个“反差感”例子对比
+
+冻结的总目标完全相同：
+
+> 让观众先相信“贵的一定更适合”，随后用真实材料完成反转；不能编数据，符合严谨克制的人设。
+
+### 老版怎么做
+
+第一轮目标：
+
+```text
+GoalStep 1：建立“贵的一定更好”的预期
+```
+
+开两个竞争分支：
+
+```text
+1.1 用高价产品包装建立预期
+1.2 用专业人士判断建立预期
+```
+
+选中 1.1，合并进 Base。
+
+第二轮：
+
+```text
+GoalStep 2：制造反转
+```
+
+再开分支：
+
+```text
+2.1 用盲听结果反转
+2.2 用真实使用场景反转
+```
+
+再选一个合并。
+
+所以最后可能是：
+
+```text
+方案A的开头
+＋方案B的反转
+＋方案C的解释
+＋方案A的结尾
+```
+
+每个局部都可能不错，但整篇创作机制可能不够统一。
+
+---
+
+### 建议版怎么做
+
+先生成三个完整方案：
+
+```text
+方案A：
+高价产品登场
+→观众形成期待
+→盲听结果反转
+→解释价格和适配不是一回事
+→回扣开头
+
+方案B：
+博主承认自己也默认贵的好
+→实际体验产生疑问
+→查证材料
+→推翻自己的初始判断
+→克制总结
+
+方案C：
+先让观众投票
+→隐藏产品价格进行体验
+→公布价格
+→反转观众和博主的共同预期
+→解释原因
+```
+
+先比较三套完整机制，选择 B。
+
+然后动态 Task：
+
+```text
+Task 1：补齐博主过去表达样例
+Task 2：确认体验和事实依据
+Task 3：生成完整节拍
+Task 4：写完整草稿
+Task 5：评估反转强度
+Task 6：重写前15秒
+```
+
+如果只是开头弱：
+
+```text
+不换方案B
+只动态调整 Task
+```
+
+如果发现“博主自我反转”不符合人设：
+
+```text
+放弃方案B
+切换方案A，或者生成方案D
+```
+
+如果事实不足：
+
+```text
+增加检索 Task
+不修改创作目标
+```
+
+## 最准确的总结
+
+```text
+共同点：
+老版和建议版都会冻结创作总目标；
+老版和建议版都会动态调整实现策略。
+
+真正区别：
+
+老版
+= 动态目标树
++ 局部实现分支
++ Patch 持续合并进 Base
++ 边写边决定整篇怎么长
+
+建议版
+= 固定创作契约
++ 完整创作方案先竞争
++ 选中后由一个主笔保持整篇统一
++ Task 动态调整具体执行
++ 必要时可以退回并更换整套方案
+```
+
+所以，如果建议版最后仍然是：
+
+```text
+把方案A/B/C写成GoalStep
+→各生成局部Patch
+→选一点A、合一点B
+→逐轮写入Base
+```
+
+那它确实只是**老版换了名字**。
+
+要真正不同，必须守住三条：
+
+1. 竞争方案是完整创作机制，不是局部 Patch。
+2. 方案比较完成前，不写入唯一 Base。
+3. 选中后由一个主笔负责整篇；只有证明创作机制不成立时，才退回切换方案。
+
+##   *********************************************************************
+这个是不是和建议版本的思路很像啊
+是，**核心上就是我们刚才“建议版”的完整版**，不是另一套思路。
+
+简易版是最小可运行 loop；这份文字在外面又加了人设编译、创意防收敛和跨运行学习。
+
+## 两者对应关系
+
+| 简易版 | 这份完整版 |
+|---|---|
+| `CreativeContract` 冻结 | 冻结输入＋创作目标/约束 |
+| `ApproachPortfolio` | 多 Direction＋聚类去重 |
+| 方案 A/B/C | 6–10 个粗方向，保留 3–4 个 |
+| 硬门槛 | 事实、格式、Goal、constraint 检查 |
+| Pairwise 比较 | 人设、创意两两竞赛＋Pareto 选择 |
+| 选择一个 Approach | 主方向＋wildcard 冒险方向 |
+| 动态 TaskLedger | 单方向的动态 Task/Attempt/Validation |
+| Critic Panel | 硬标准、人设、创意三类 Validator |
+| RevisionBrief | 缺陷修订＋亮点保护 |
+| 失败后换 Approach | Direction 错误时回到方向搜索 |
+| 暂时没有跨稿学习 | Memory → SFT → DPO → 有限 RL |
+
+简易版：
+
+```text
+固定创作契约
+    ↓
+生成方案 A / B / C
+    ↓
+比较并选择
+    ↓
+动态 Task 写整篇
+    ↓
+批评、改稿或换方案
+    ↓
+完成
+```
+
+完整版：
+
+```text
+固定创作契约
+    ↓
+编译 Persona Contract
+    ↓
+生成更多创作方案
+    ↓
+聚类去重、事实过滤、两两竞赛
+    ↓
+主方向 + wildcard
+    ↓
+动态 Task 构建两份整稿
+    ↓
+人设 / 创意 / 硬标准分开评估
+    ↓
+保护亮点地修改
+    ↓
+人工反馈
+    ↓
+形成账号 Memory
+    ↓
+未来再做 SFT / DPO / RL
+```
+
+## 但有一个重要术语需要纠正
+
+原文最后说：
+
+> 在“冻结目标之前充分发散”。
+
+这句话容易把系统带偏。准确说法应该是：
+
+> **在冻结具体创作方案之前充分发散，而不是在冻结原始创作目标之前发散。**
+
+需要分成三层：
+
+### 1. CreativeContract：固定
+
+用户最初提出的创作要求：
+
+```text
+要有反差感
+符合账号人设
+不能编事实
+内容要有趣
+```
+
+这些必须先固定，不能让搜索过程修改。
+
+### 2. CreativeApproach：动态搜索
+
+实现“反差感”的办法：
+
+```text
+A：价格预期反转
+B：博主自我认知反转
+C：观众参与反转
+D：时间顺序倒置
+```
+
+这里才需要充分发散、比较、淘汰和重新生成。
+
+### 3. Task：动态执行
+
+选中 B 以后，怎么把它写好：
+
+```text
+补人设样例
+找事实依据
+调整反转位置
+重写前15秒
+压缩中段解释
+强化结尾回扣
+```
+
+Task 同样动态调整。
+
+所以不是：
+
+```text
+先发散多个“要不要反差”
+```
+
+而是：
+
+```text
+“必须有反差”固定
+        ↓
+发散多个“怎么制造反差”
+```
+
+## 和老版仍然有什么不同
+
+这份完整版和老版确实也有重叠：
+
+- 总目标冻结。
+- 有多路径。
+- 有评估。
+- 失败后动态调整。
+- 最后整篇验收。
+
+真正区别依然在“多路径的单位”。
+
+### 老版
+
+```text
+本轮局部 GoalStep
+    ↓
+多个 Patch 实现分支
+    ↓
+选 / 组 / 合并到 Base
+    ↓
+继续下一块
+```
+
+优化单位是局部 Patch。可能出现：
+
+```text
+A 的开头
+＋B 的中段
+＋C 的反转
+＋A 的结尾
+```
+
+### 这份完整版
+
+```text
+同一个 CreativeContract
+    ↓
+多个完整 CreativeApproach
+    ↓
+每个方案有完整的铺垫—反转—结尾
+    ↓
+按整套创作机制比较
+    ↓
+选主方向和 wildcard
+    ↓
+再进入动态 Task 构建
+```
+
+优化单位首先是**完整创作假设**，不是局部 Patch。
+
+如果最终实现成：
+
+```text
+把 Direction A/B/C 变成 GoalStep
+→分别生成局部 Patch
+→不断合并到 Base
+```
+
+那确实又变回老版。
+
+## 对当前项目最实际的落法
+
+不需要马上实现文档里所有能力。第一阶段可以只做简易版：
+
+```text
+CreativeContractArtifact：1份，硬冻结
+        ↓
+CreativeApproachArtifact：生成3–4份
+        ↓
+ApproachPortfolio：保留主方案＋wildcard
+        ↓
+现有 TaskLedger：分别构建完整节拍或草稿
+        ↓
+硬标准过滤
+        ↓
+人设＋创意 Pairwise 比较
+        ↓
+RevisionBrief：
+- defects_to_fix
+- strengths_to_preserve
+        ↓
+改稿或切换 Approach
+```
+
+暂时不做：
+
+- 长期 Memory；
+- SFT；
+- DPO；
+- RL；
+- 6–10 个高成本完整成稿；
+- 复杂的账号级训练。
+
+一句话总结：
+
+> **这份文档是简易建议版的“生产完整版＋长期学习路线”。主干完全一致；最需要修正的是，不要把用户创作目标和具体创作 Direction 混在一起——目标固定，Direction/Approach 动态搜索，Task 动态执行。**
+
+
+##  *******************************************************************
+我在想怎么能做好一个智能创作系统，后面再延伸到全模块。 第二版的创作系统是个高自由度规划的系统了，我们的可能更无限发散的task。 不知能能不能像rl一样学习起来，学习经验知识，但是又想不到奖励函数
+
+可以像 RL 一样“越跑越会”，但我不建议现在就训练一个端到端 RL Planner。对于智能创作，真正困难的不是算法，而是：
+
+> 创作没有唯一正确答案，奖励天然是多目标、分层、延迟、带主观性的。
+
+因此应该先建设“经验学习系统”，等反馈数据稳定后，再决定哪些部分值得做 RL。
+
+## 一、先区分三种“学习”
+
+### 1. 单次 Run 内学习
+
+```text
+Plan → Task → Worker → Validator → 返工/换路 → 完成
+```
+
+这是当前已经具备的动态循环，但严格说不叫学习，因为下一次 Run 不会自动继承这次经验。
+
+### 2. 跨 Run 经验学习
+
+```text
+这次为什么失败
+→ 最后怎么修好的
+→ 什么条件下这个办法有效
+→ 下次遇到相似任务自动召回
+```
+
+模型参数不变，但系统确实会越跑越好。
+
+Reflexion、Contextual Experience Replay、Voyager 的共同思想，都是把成功技能和失败反思保存起来，在后续任务中检索复用；不一定先训练模型权重。[Contextual Experience Replay](https://openreview.net/forum?id=RXvFK5dnpz)、[Voyager](https://voyager.minedojo.org/assets/documents/voyager.pdf)
+
+这是我们最应该先做的。
+
+### 3. 真正的策略训练
+
+当积累了足够多稳定数据后，训练：
+
+- Planner 在当前状态下应该创建什么 Task。
+- 应该检索哪类数据。
+- 应该并行探索几路。
+- 哪个候选值得继续投入。
+- 什么时候停止发散。
+- 某类错误应该 retry、repair 还是 replan。
+
+这才更接近 RL、偏好优化或者 contextual bandit。
+
+---
+
+# 二、为什么创作不适合一个总奖励分数
+
+假设我们给一篇脚本定义：
+
+```text
+总奖励 =
+内容质量 × 0.4
++ 用户喜欢 × 0.3
++ 阅读数据 × 0.2
+- Token 成本 × 0.1
+```
+
+看起来合理，实际上很危险。
+
+模型可能学会：
+
+- 为了“用户喜欢”，不断制造夸张标题。
+- 为了“阅读数据”，牺牲账号长期人设。
+- 为了“低成本”，不再探索候选。
+- 为了“Validator 分高”，专门写 Validator 喜欢的标签。
+- 为了“Goal 覆盖率”，让每个 Paragraph 声称覆盖所有 Goal。
+
+Build 900036 的 Structure 声称覆盖全部 Goal、实际正文为空，就是典型的“满足字面指标、没有完成真实意图”。DeepMind 将这种现象称为 specification gaming：Agent 找到奖励规则的漏洞，而不是完成设计者真正想要的目标。[DeepMind：Specification gaming](https://deepmind.google/blog/specification-gaming-the-flip-side-of-ai-ingenuity/)
+
+所以我们不应该先做一个加权总分，而应该做分层、带门禁的奖励协议。
+
+---
+
+# 三、建议使用“奖励向量 + 排序规则”
+
+每个 Artifact、Task、Mission 保存一组独立反馈：
+
+```text
+Outcome
+├── hard_constraints_passed
+├── goal_satisfaction
+├── content_quality
+├── evidence_quality
+├── human_preference
+├── downstream_utility
+├── novelty
+├── diversity
+├── cost
+└── latency
+```
+
+不急着把它们压成一个分数。
+
+## 第一层：硬门禁
+
+以下任意一项失败，候选直接淘汰：
+
+- 违反 Constraint。
+- 正文不完整。
+- Evidence 伪造或缺失。
+- Artifact 闭包错误。
+- Goal 没有真实实现。
+- 制作表无法还原脚本。
+- 合规、安全、品牌规范不通过。
+
+这部分适合确定性代码和强 Validator。
+
+## 第二层：目标完成度
+
+逐个 Goal 和 success criterion 判断：
+
+```text
+g1：0.9
+g2：0.7
+g3：0.8
+```
+
+父 Goal、子 Goal分别评分，不用子 Goal 自动推导父 Goal。
+
+OpenAI 在过程监督研究中发现，只奖励最终结果会丢失大量中间信息；对中间步骤提供反馈，更有利于识别到底是哪一步出错。[OpenAI：Process supervision](https://openai.com/index/improving-mathematical-reasoning-with-process-supervision/)
+
+对应到我们的系统就是：
+
+- Task Validation：局部过程奖励。
+- Root Validation：最终结果奖励。
+- Decision lineage：信用分配路径。
+
+## 第三层：候选之间做偏好比较
+
+创作质量很难绝对打 83 分，但人通常比较容易回答：
+
+> A 和 B 哪个更适合这个账号和选题？为什么？
+
+因此创作候选优先采用：
+
+```text
+A 胜 B
+B 与 C 平手
+C 在结构上更好
+A 在语言上更好
+```
+
+而不是强迫 Validator 输出一个虚假的精确总分。
+
+Anthropic 的 Constitutional AI 也使用原则和候选间偏好来构造反馈，而不是要求人给每个结果设计精确奖励函数。[Anthropic：Constitutional AI](https://www.anthropic.com/research/constitutional-ai-harmlessness-from-ai-feedback)
+
+## 第四层：成本只用于同质量候选之间排序
+
+例如：
+
+```text
+候选 A：质量通过，耗时 8 分钟
+候选 B：质量通过，耗时 35 分钟
+```
+
+这时选择 A。
+
+但不能让“低 Token”抵消“正文不合格”。
+
+因此是词典序：
+
+```text
+硬约束
+  > Goal 完成
+    > 人类/语义偏好
+      > 下游效果
+        > 成本与耗时
+```
+
+不是把所有指标混成一个加权平均数。
+
+---
+
+# 四、当前系统其实已经有了 RL 最缺的东西
+
+第二版有高自由度探索，但轨迹很难可靠归因：
+
+```text
+Agent读取大量状态
+→ 写很多数据库记录
+→ 多轮分支合并
+→ 最后成功/失败
+```
+
+很难判断最终成功究竟是哪一步造成的。
+
+当前系统已经有：
+
+- Direction Goal。
+- TaskSpec。
+- Attempt。
+- Artifact。
+- Validation。
+- Decision。
+- accepted closure。
+- FailureDetail。
+- 模型调用与工具调用成本。
+- Workbench 状态 revision。
+
+这些天然构成了经验轨迹：
+
+```text
+状态 S
+→ Planner 动作 A
+→ Worker 产物
+→ Validator 局部反馈
+→ Planner下一动作
+→ Root 最终结果
+```
+
+这正是后续策略学习需要的数据。
+
+当前真正缺的是一个跨 Run 的“经验层”，而不是重写 Agent 框架。
+
+---
+
+# 五、建议增加一个 Experience Repository
+
+它不属于单次 Mission Workbench。
+
+- Workbench：维护当前 Run 正在发生什么。
+- Experience Repository：总结过去 Run 学到了什么。
+
+建议的经验结构：
+
+```text
+Experience
+├── applicability
+│   ├── module：topic/script/production
+│   ├── content_type
+│   ├── account/persona
+│   ├── goal_pattern
+│   └── preconditions
+├── state_signature
+├── decision
+│   ├── selected_action
+│   ├── rejected_alternatives
+│   └── reason
+├── outcome
+│   ├── validation_results
+│   ├── human_feedback
+│   ├── downstream_metrics
+│   ├── cost
+│   └── latency
+├── lesson
+│   ├── what_worked
+│   ├── what_failed
+│   ├── why
+│   └── next_time
+├── source_trace_refs
+├── confidence
+├── counterexamples
+└── expires_at
+```
+
+例如：
+
+```text
+适用条件：
+知识型小红书脚本，Goal 包含“纠正常见误区”
+
+失败动作：
+先写完整正文，最后才检索领域证据
+
+失败原因：
+正文中的三条判断无法找到 accepted evidence，
+导致 Paragraph 全部返工
+
+有效策略：
+先创建 Knowledge Retrieval，
+冻结关键事实后再生成 Paragraph
+
+效果：
+返工次数 4 → 1
+Token 下降 38%
+```
+
+下次遇到相似 Goal，Planner Bootstrap 只召回这一条经验，而不是重新读取完整历史 Run。
+
+---
+
+# 六、经验不能让模型自己随便总结后直接生效
+
+否则系统会出现“自我强化错误”：
+
+```text
+某次碰巧成功
+→ 模型总结错误因果
+→ 后续一直复用
+→ 错误经验越来越强
+```
+
+经验进入正式库需要门槛：
+
+1. 对应 Run 有完整 Trace 和 Artifact。
+2. 最终确实通过 Root Validation。
+3. 局部经验能找到对应 Decision 和结果。
+4. 至少有一次复现，或标记为低置信度。
+5. 有反例时降低置信度。
+6. 人工修改过的内容，以人类最终版本为准。
+7. Experience 只能被检索使用，不能直接绕过 Validator。
+
+经验应该有三种状态：
+
+```text
+candidate → validated → deprecated
+```
+
+---
+
+# 七、怎样控制“无限发散的 Task”
+
+高自由度不是让 Planner 无限创建 Task。
+
+应该给 Planner 一个“探索预算”：
+
+```text
+ExplorationBudget
+├── max_active_tasks
+├── max_candidate_width
+├── max_retrieval_calls
+├── max_validator_calls
+├── token_budget
+├── time_budget
+└── minimum_expected_information_gain
+```
+
+每次创建新 Task 前问：
+
+> 这个 Task 会带来新的信息、真实不同的候选，还是只是在重复已有工作？
+
+## 可以代码判断的部分
+
+- 相同语义合同去重。
+- 相同 Query 去重。
+- Goal Coverage 是否已有候选覆盖。
+- 新候选与已有候选的语义相似度。
+- 上一次相同动作是否已经失败。
+- 当前剩余预算。
+- Task 是否会解锁依赖。
+- 是否有尚未处理的 Validation。
+
+## Planner真正判断的部分
+
+- 当前最有价值的创作缺口是什么。
+- 值得探索竞争候选还是直接执行。
+- 需要外部证据还是可以原创。
+- 现有候选差异是否足够。
+- 当前质量是否值得继续投入。
+
+最后可以学习的是：
+
+```text
+在这种状态下，
+创建哪一种 Task 最可能提升最终结果？
+```
+
+这比直接训练“模型如何写整篇脚本”容易归因得多。
+
+---
+
+# 八、全模块怎样共用学习能力
+
+不同模块不能共用同一奖励定义，但可以共用同一经验基础设施。
+
+| 模块 | 主要局部奖励 |
+|---|---|
+| 选题 | 新颖性、需求相关性、数据支持、可创作空间、账号适配 |
+| 脚本 | Goal 实现、正文质量、叙事推进、证据、人格一致 |
+| 制作 | 可还原性、制作完整度、视觉一致性、执行成本、素材可得性 |
+| 发布后 | 用户选择、人工修改量、制作返工量、真实业务表现 |
+
+同时保留端到端结果：
+
+```text
+选题被脚本采用
+→ 脚本是否容易制作
+→ 制作表是否需要大量人工返工
+→ 最终内容是否被发布
+→ 发布后是否达到业务目标
+```
+
+这样可以发现跨模块问题：
+
+- 选题分数很高，但脚本一直写不出来：选题缺乏可展开性。
+- 脚本 Validator 很高，但制作表返工巨大：脚本缺少可视化信息。
+- 制作表完整，但人工大量改风格：Persona/制作规律没有真正落实。
+
+不过最终流量不能直接倒灌成唯一奖励，因为流量同时受发布时间、账号体量、平台推荐等外部因素影响。它只能作为同账号、相近题材、相近时间窗口下的弱反馈。
+
+---
+
+# 九、落地顺序
+
+## 第一阶段：经验学习，不训练模型
+
+先增加：
+
+- Experience Repository。
+- Run 后经验提炼器。
+- 经验审核和置信度。
+- Planner 按状态检索经验。
+- 成功/失败策略对照。
+- 人工修改差异采集。
+- 跨模块 outcome event。
+
+这是投入产出最高的阶段。
+
+## 第二阶段：学习选择，不学习生成
+
+用 contextual bandit 或简单统计策略学习：
+
+- 哪个 Retrieval 最有效。
+- 某类 Goal 应创建几个候选。
+- 何时 Compare。
+- 何时停止探索。
+- 哪类失败应该 repair/replan。
+- 哪些经验值得召回。
+
+模型仍负责创作，系统学习“怎么组织创作”。
+
+## 第三阶段：偏好模型
+
+积累同一目标下的 A/B 候选和真实选择后，训练或配置一个 Candidate Ranker：
+
+```text
+输入：Direction + 候选A + 候选B
+输出：
+- winner
+- 各维度差异
+- 置信度
+```
+
+## 第四阶段：Planner 策略优化
+
+最后才用离线轨迹训练 Planner：
+
+```text
+state → next task/action
+```
+
+而不是端到端训练整条内容生成链。
+
+---
+
+## 我的总体建议
+
+我们可以把第二版的高自由度探索和当前版的可靠任务审计结合起来：
+
+```text
+第二版提供：
+动态目标树、多路竞争、创作自由度
+
+当前版提供：
+Task/Attempt/Validation/Decision、不可变Artifact、Workbench
+
+新增经验层提供：
+跨Run复用、失败反思、策略选择、全模块反馈
+```
+
+最终系统会形成三个循环：
+
+```mermaid
+flowchart TD
+    A["Mission内循环<br/>Plan → Execute → Validate → Replan"] --> B["Run后学习<br/>提炼成功与失败经验"]
+    B --> C["跨Run复用<br/>召回经验与策略"]
+    C --> A
+    B --> D["积累偏好和结果数据"]
+    D --> E["训练Ranker/Planner策略"]
+    E --> A
+```
+
+所以答案是：能学习，而且当前架构比第二版更适合学习。但第一步不应该是寻找一个完美奖励函数，而是建立“可追踪的多维反馈 + 跨 Run 经验库 + 经验生效治理”。
+
+奖励不是一个数字，而是一套分层判定协议。等这套协议在线运行稳定后，RL 才会有可信的训练数据。
+
+
+## ****************************************************************
+我不知道这样的经验学习效果会不会好，也许不如第二版发散地动态规划路径
+
+你的担心是对的。
+
+如果把“经验学习”做成：
+
+> 过去成功过，所以以后优先照过去的方式做。
+
+那很可能不如第二版。它会越来越保守，最后所有选题、脚本、制作路径趋同，形成创作系统最危险的“经验收敛”：
+
+```text
+以前反差叙事成功
+→ 下次继续反差叙事
+→ 反差案例越来越多
+→ 系统更确信反差最好
+→ 逐渐不再尝试别的路径
+```
+
+所以经验学习不能替代动态规划。更正确的组合是：
+
+> 动态规划负责探索，经验负责提供先验，Validator 和真实结果负责淘汰。
+
+## 一、第二版真正有价值的是“动态搜索”
+
+第二版的核心不是完全自由，而是一种搜索：
+
+```text
+当前目标
+→ 提出多条不同路径
+→ 每条路径独立实现
+→ 比较结果
+→ 选择、组合、挂起或放弃
+→ 根据整体缺口规划下一轮
+```
+
+用 RL 术语说，它更像：
+
+- Planner：策略 rollout。
+- 多路分支：并行探索。
+- Validator：局部奖励。
+- 合并/放弃：剪枝。
+- Goal Tree：搜索树。
+- 下一轮规划：根据结果扩展搜索树。
+
+这部分应该保留，而且可以比第二版做得更自由。
+
+## 二、经验应该是什么角色
+
+经验不能对 Planner 说：
+
+> 你必须走这条路。
+
+它只能说：
+
+> 过去在类似条件下，这条路成功率较高；但它在这些情况下失败过。
+
+也就是“参考地图”，不是“导航命令”。
+
+```text
+动态 Planner
+├── 自由提出新路径
+├── 从经验库召回旧路径
+├── 主动提出与旧经验相反的路径
+└── 对已有路径进行组合和变形
+```
+
+经验只影响每条路径的初始优先级，不能取消其他路径。
+
+## 三、我建议把核心组件定义成 Mission Search Engine
+
+Experience Repository 只是其中一个部件，系统主体仍然是搜索。
+
+```mermaid
+flowchart TD
+    S["当前 Mission 状态"] --> P["Search Planner"]
+    E["历史经验"] --> P
+    P --> H["生成多条路径假设"]
+    H --> T["编译成 Task 子图"]
+    T --> W["Worker并行实现"]
+    W --> V["Validator独立验收"]
+    V --> C["比较/剪枝/组合"]
+    C -->|仍有缺口| P
+    C -->|完成| R["Root交付"]
+    R --> L["提炼新经验"]
+    L --> E
+```
+
+这个结构里：
+
+- Workbench 管当前状态。
+- Search Planner 负责动态发散。
+- Task Compiler 把语义路径变成机械合同。
+- Worker 负责实施。
+- Validator 提供过程反馈。
+- Experience Repository 提供历史先验。
+- Search Controller 控制预算、探索和剪枝。
+
+这样经验不会代替创造力。
+
+---
+
+# 四、Planner 每轮不是直接创建 Task，而是先提出“路径假设”
+
+第二版直接规划实现任务，但我建议当前系统先增加一层轻量的 `PathHypothesis`：
+
+```text
+PathHypothesis
+├── hypothesis_id
+├── goal_ids
+├── statement
+├── rationale
+├── approach
+├── expected_gain
+├── evidence_need
+├── differentiation
+├── risks
+├── estimated_cost
+└── origin
+    ├── novel
+    ├── experience
+    ├── counter
+    └── repair
+```
+
+例如一个 Goal 是：
+
+> 让职场新人意识到“努力但无反馈”不一定是能力问题。
+
+Planner 可以先提出：
+
+```text
+路径 A｜经验复用
+用一个“连续加班却从未确认老板期望”的真实案例展开。
+
+路径 B｜全新探索
+从管理学反馈回路切入，用具体机制解释问题。
+
+路径 C｜反向路径
+先站在老板视角解释为什么员工越努力越可能偏航。
+
+路径 D｜融合路径
+用A的故事进入，用B解释原因，用C制造认知反转。
+```
+
+然后 Host 再把被选中的路径编译成：
+
+- Retrieval Task
+- Paragraph Task
+- ElementSet Task
+- Compare Task
+
+这样 Planner 发散的是创作假设，不是数据库合同。
+
+---
+
+# 五、强制保留探索配额
+
+不能让经验路径占满全部预算。
+
+每轮候选可以保持一个可配置组合，例如：
+
+```text
+候选路径
+├── 经验证路径：利用已有经验
+├── 新颖路径：禁止直接复用已有经验
+├── 反向路径：挑战当前最强经验
+└── 融合路径：组合两个不相同的方向
+```
+
+不一定每轮固定四条，但 Search Controller 必须检查：
+
+- 是否至少存在一条真正的新路径。
+- 候选之间是否语义重复。
+- 是否都只是换说法，没有方法差异。
+- 是否存在对当前经验的反例探索。
+- 新路径是否带来真实信息增量。
+
+如果四条都是“故事化表达，只是换了人物”，应判定为重复，而不是四路探索。
+
+## 经验丰富以后也不能取消探索
+
+可以采用类似 explore/exploit 的预算：
+
+```text
+大部分预算：走当前看起来最有价值的路径
+一部分预算：尝试未经验证的新路径
+少量预算：专门挑战当前最强经验
+```
+
+比例不应该全局写死，可以随风险变化：
+
+- 熟悉、低风险任务：更多利用经验。
+- 新账号、新题材：增加探索。
+- 连续多次产物相似：强制增加探索。
+- 当前最优候选质量已很高：减少无意义探索。
+- Validator 对候选分歧大：增加对比探索。
+
+---
+
+# 六、经验库必须同时保存成功和失败
+
+只保存成功经验一定会导致偏差。
+
+例如：
+
+```text
+成功经验：
+反差开场在知识型脚本中提高了首段吸引力。
+
+失败经验：
+当选题本身需要建立可信度时，
+过早制造反差导致信息基础不足，Root Validator拒绝。
+
+适用条件：
+受众已经理解基本背景，且反差有事实证据。
+
+反例条件：
+陌生领域、事实密集、需要先建立概念。
+```
+
+Planner看到的应该是：
+
+```text
+经验 + 适用条件 + 失败边界 + 反例
+```
+
+而不是一句“反差叙事效果好”。
+
+## 经验召回也要保持异质性
+
+不能只返回相似度最高的五条经验，因为它们可能全是同一种路线。
+
+召回应包含：
+
+- 最相似成功经验。
+- 最相似失败经验。
+- 相邻领域成功经验。
+- 与主流经验冲突的反例。
+- 一条未验证的新假设。
+
+---
+
+# 七、奖励不负责生成创意，只负责更新路径价值
+
+我们不需要设计一个奖励函数告诉模型：
+
+> 怎样写出伟大的脚本。
+
+只需要相对可靠地回答：
+
+> 在这个状态下，这条路径值不值得继续？
+
+路径价值可以来自：
+
+```text
+PathOutcome
+├── hard_pass
+├── goal_gain
+├── quality_gain
+├── information_gain
+├── novelty_contribution
+├── downstream_edit_cost
+├── token_cost
+└── latency
+```
+
+例如：
+
+| 路径 | Goal提升 | 内容质量 | 新信息 | 成本 | 结论 |
+|---|---:|---:|---:|---:|---|
+| A 复用经验 | 高 | 高 | 低 | 低 | 可以采用 |
+| B 新探索 | 中 | 高 | 高 | 中 | 值得保留 |
+| C 反向挑战 | 低 | 中 | 高 | 高 | 本次不采用，但形成反例经验 |
+| D 融合 | 高 | 最高 | 中 | 中 | 最终采用 |
+
+路径 C 没被采用，也不是“毫无价值的失败”。它可能证明：
+
+> 老板视角不适合这次受众，但对管理者账号可能有效。
+
+这是一条带适用范围的经验。
+
+---
+
+# 八、当前系统怎样恢复第二版的自由度
+
+当前的 Task Graph 完全可以表达第二版的动态目标树，只是现在表达得偏合同化。
+
+建议加两层：
+
+```text
+Direction Goal
+└── Search Episode
+    ├── PathHypothesis A
+    │   ├── Retrieval
+    │   ├── Paragraph
+    │   └── ElementSet
+    ├── PathHypothesis B
+    │   ├── Retrieval
+    │   └── Paragraph
+    ├── Compare
+    └── Adoption Decision
+```
+
+一次 Search Episode 结束后：
+
+- Goal 达成：关闭这个 Goal。
+- 局部达成：增加补充路径。
+- 路径相互冲突：创建新的 Compare。
+- Goal 定义过大：拆分子 Goal。
+- 新证据推翻前提：回到上层重新规划。
+- 多条路径各有价值：Compose 组合。
+- 没有有效路径：扩大检索或调整假设。
+
+这比第二版数据库 Branch 更自由，同时仍有 Artifact 和 Validation 审计。
+
+---
+
+# 九、不要一上来决定经验学习是否有效，先做影子验证
+
+你的怀疑应该通过实验回答，而不是靠架构想象。
+
+## 第一步：Shadow Mode
+
+系统正常使用动态规划，经验库只在后台工作，不暴露给 Planner。
+
+每次 Run 后检查：
+
+- 经验召回是否包含最后胜出的路径。
+- 经验预测的优胜路径是否真的胜出。
+- 经验是否会错误压制本次新路径。
+- 成功经验在新题材上是否失效。
+
+这一阶段不影响创作质量。
+
+## 第二步：辅助模式
+
+把经验展示给 Planner，但明确：
+
+```text
+这些是有边界的历史参考，不是必须采用的方案。
+至少提出一条不依赖历史经验的新路径。
+```
+
+与无经验组做同输入、同模型、同预算比较：
+
+- 人工 A/B 偏好。
+- Goal 完成率。
+- 候选语义多样性。
+- 最终人工修改量。
+- Root 通过率。
+- Task 数量。
+- Token 和耗时。
+
+## 第三步：策略模式
+
+只有当数据证明经验确实提高质量或效率，才允许经验影响：
+
+- 候选数量。
+- 路径优先级。
+- 检索选择。
+- 剪枝时机。
+- 停止条件。
+
+---
+
+## 最后的判断
+
+经验学习本身确实可能不如第二版自由发散；如果把经验当成答案，它甚至会让系统退化。
+
+更好的智能创作系统应该是：
+
+```text
+动态规划是发动机
+经验是先验
+Validator是反馈
+Workbench是当前世界状态
+Artifact/Decision是可归因轨迹
+Search Controller负责探索与利用平衡
+```
+
+不是“经验系统取代第二版”，而是：
+
+> 保留并强化第二版的动态多路搜索，再用当前框架记录每条路径为什么产生、做出了什么、为什么胜出，最后让经验逐渐提高搜索效率，但永远保留新路径和反例探索。
+
+如果只能先做一个组件，我会优先做 `Search Episode + PathHypothesis`，而不是 Experience Repository。因为先有高质量、多样化、可归因的搜索轨迹，后面才有值得学习的经验。
+
+## *************************************************************
+
+我的情况顶级的论文和ai公司有遇到过吗，还有评价一个创作脚本的好坏，是不是只能human自己来打分，这样数据量大了确实成本很高。 然后我在想如果把选题脚本制作，和后面的真实生成的agent连起来，用一套统一框架，最后还是调用seeddance api，生成的视频可以拿出去测用户数据反馈，时长，点击量之类。是不是可以有学习算法让他迭代。 但是seeddance本来也只是黑盒，我们也就是做到提示词，
+
+
+有，而且你的问题几乎正好落在几条顶级研究路线的交叉点上：
+
+1. 开放式内容没有客观答案，怎样定义奖励。
+2. 人类评价很准，但太贵。
+3. 生成模型是黑盒，只能控制 Prompt、参考素材和参数。
+4. 最终用户反馈很真实，但噪声大、延迟长、难归因。
+5. 系统既要利用经验，又不能失去创作发散性。
+
+没有哪家公司已经公开了一套和你完全一样的“选题 → 脚本 → 制作 → 视频生成 → 发布反馈”系统，但每个子问题都有成熟研究。
+
+## 一、顶级公司确实遇到了相同问题
+
+| 你的问题 | 对应研究 |
+|---|---|
+| 创作质量无法写成规则 | OpenAI 用人类两两比较训练 Reward Model |
+| 人工标注太贵 | Anthropic 用 AI Feedback 和 Constitution 扩大评价规模 |
+| 长流程最后才知道好坏 | OpenAI 研究 Process Supervision |
+| 黑盒模型只能改 Prompt | Google DeepMind 用 OPRO 自动优化 Prompt |
+| 用户反馈有短期和长期差异 | Google/YouTube 用 RL 优化长期用户价值 |
+| 内容生成与消费反馈联动 | Google 研究生成与消费联合优化 |
+| 视频生成质量是多维目标 | Seedance 自己也使用视频专用 RLHF 和多维奖励 |
+
+OpenAI 在摘要这种同样主观的任务上，不要求人类给每个摘要打绝对分，而是让人回答“A 和 B 哪个更好”，然后训练 Reward Model 预测人类偏好。[OpenAI：Learning to summarize with human feedback](https://openai.com/index/learning-to-summarize-with-human-feedback/)
+
+Anthropic进一步用明确原则让 AI 比较候选，只保留较少的人类监督，这就是 RLAIF；目的正是降低全量人工评价的成本。[Anthropic：Constitutional AI](https://www.anthropic.com/research/constitutional-ai-harmlessness-from-ai-feedback)
+
+Google DeepMind 的 OPRO 则证明了：即使被调用模型是黑盒，也可以根据“Prompt → 输出 → 分数”的历史，让另一个模型不断提出更优 Prompt。[Large Language Models as Optimizers](https://arxiv.org/abs/2309.03409)
+
+最接近你后半段设想的是 Google 的推荐系统研究：它们会利用点击、观看和长期回访等反馈优化推荐策略，并已在 YouTube 实验长期价值学习。[Google：SlateQ](https://research.google/pubs/reinforcement-learning-for-slate-based-recommender-systems-a-tractable-decomposition-and-practical-methodology/)
+
+甚至已经有“联合优化内容生成和内容消费”的研究，不再把生成系统和推荐系统完全割裂。[Google：Co-optimize Content Generation and Consumption](https://research.google/pubs/co-optimize-content-generation-and-consumption-in-a-large-scale-video-recommendation-system/)
+
+Seedance 的技术报告也说明，它自身训练阶段就用了视频专用 RLHF 和多维奖励。但那是字节训练 Seedance 内部权重；我们通过 API 无法继续训练它。[Seedance 1.0 技术报告](https://arxiv.org/abs/2506.09113)
+
+---
+
+# 二、脚本好坏不是只能靠 Human，但“好”的定义最终来自人
+
+需要区分两件事：
+
+- 人类是否要逐条评价每个脚本：不需要。
+- 系统最终追求什么价值：仍然需要人类定义。
+
+合理的是五层评价漏斗。
+
+## 第一层：确定性代码检查
+
+不用人，也不用模型：
+
+- 是否有完整正文。
+- 是否有空字段或 placeholder。
+- Goal 是否全部覆盖。
+- Constraint 是否违反。
+- Evidence 是否真实存在。
+- Paragraph、Element、Link 是否闭合。
+- 时长、字数、格式是否满足要求。
+
+这部分当前系统已经在做。
+
+## 第二层：AI 语义 Validator
+
+AI 判断：
+
+- 叙事有没有推进。
+- 内容是不是空话。
+- 人设是否一致。
+- 开头是否建立预期。
+- 转折是否真的改变理解。
+- 信息是否具体。
+- 情绪是否自然。
+- Goal 是否在真实文稿中实现。
+
+AI Judge 可以大规模工作，但不能当绝对真理。研究表明，它可能有：
+
+- 位置偏差。
+- 偏爱长答案。
+- 偏爱与自己风格相似的内容。
+- 被流畅但空洞的文字欺骗。
+
+因此需要交换 A/B 顺序、多 Judge、证据引用和人类校准。[MT-Bench / LLM-as-a-Judge](https://arxiv.org/abs/2306.05685)
+
+## 第三层：人类只做两两比较
+
+不让编辑回答：
+
+> 这个脚本是 7.3 分还是 7.8 分？
+
+只问：
+
+> 同一个选题，A 和 B 哪个更值得制作？为什么？
+
+两两比较：
+
+- 更容易判断。
+- 一致性更高。
+- 可以直接训练 Candidate Ranker。
+- 不需要精确发明一个总分。
+
+人类还可以提交：
+
+- A 胜 B。
+- 平手。
+- A 开头好、B 中段好。
+- 两个都不值得制作。
+- 选 A，但需要修改这些部分。
+
+## 第四层：把人工编辑行为变成免费标签
+
+编辑正常工作时产生的行为，就是训练数据：
+
+- 选择哪个候选。
+- 删除了哪些段落。
+- 重写了哪些句子。
+- 哪些 Goal 被重新定义。
+- 哪个制作方案被采用。
+- Seedance 生成结果是否直接采用。
+- 第几次生成才通过。
+- 哪些 Prompt 修改带来了改善。
+
+这比让编辑额外打分更自然。
+
+例如：
+
+```text
+原脚本 → 人工最终稿
+```
+
+可以自动提取：
+
+- 被保留的结构。
+- 被删除的套话。
+- 新增的信息点。
+- 调整的叙事顺序。
+- 语气变化。
+- 哪种错误最常返工。
+
+## 第五层：只把“系统不确定的样本”交给人
+
+不需要随机评价大量内容。
+
+优先让人评价：
+
+- 多个 AI Judge 意见不一致。
+- 新账号、新题材、新内容形式。
+- 经验库没有相似案例。
+- 两个候选非常接近。
+- AI 高分但用户反馈很差。
+- AI 低分但用户反馈很好。
+- 可能进入重要投放的内容。
+
+这叫 Active Preference Learning：把有限人工预算花在最有信息量的样本上，而不是平均标注所有样本。[ICML 2024：Active Preference Learning](https://proceedings.mlr.press/v235/muldrew24a.html)
+
+因此最合理的比例不是“所有脚本都让人打分”，而是：
+
+```text
+100% 确定性检查
+100% AI Validator
+少量高价值样本由 Human 比较
+所有已发布内容收集真实行为反馈
+定期用 Human 校准 AI Judge
+```
+
+---
+
+# 三、把选题、脚本、制作、Seedance 和用户反馈连起来是可行的
+
+但我不建议做成一个巨型 Agent 从头跑到底。
+
+应该共用统一框架和审计协议，同时保留不同 Mission：
+
+```mermaid
+flowchart LR
+    T["Topic Mission"] --> TA["Accepted Topic"]
+    TA --> S["Script Mission"]
+    S --> SA["Accepted Script"]
+    SA --> P["Production Mission"]
+    P --> PA["Accepted Production Plan"]
+    PA --> G["Generation Mission"]
+    G --> V["Seedance Video"]
+    V --> Q["Video QA与人工选择"]
+    Q --> E["发布实验"]
+    E --> O["用户Outcome"]
+    O --> L["Learning Service"]
+    L --> T
+    L --> S
+    L --> P
+    L --> G
+```
+
+统一的是：
+
+- Task
+- Attempt
+- Artifact
+- Validation
+- Decision
+- Goal
+- Failure
+- Cost
+- Trace
+- Experiment
+- Outcome
+
+不统一的是：
+
+- 选题的业务 Artifact。
+- 脚本的评价标准。
+- 制作表的数据结构。
+- 视频生成的输入协议。
+- 每个模块的局部奖励。
+
+这叫统一运行框架，不是统一成一个巨大 Prompt。
+
+---
+
+# 四、Seedance 是黑盒，仍然有很多东西可以学习
+
+我们不能训练 Seedance 内部权重，但可以学习“如何使用 Seedance”。
+
+把 Seedance 看成环境：
+
+```text
+输入：
+Prompt + 参考素材 + 制作规划 + API参数
+
+黑盒：
+Seedance
+
+输出：
+视频
+
+反馈：
+质量检测 + 人工选择 + 用户数据
+```
+
+## 我们可以优化的 Action
+
+即使只能调用 API，外层仍可学习：
+
+- 选择哪个选题。
+- 选择哪个脚本候选。
+- 如何拆镜头。
+- 一次生成单镜头还是多镜头。
+- 每个镜头提供哪些文字描述。
+- 哪些内容用参考图。
+- 哪些内容用参考视频、音频或角色素材。
+- Prompt 字段顺序和表达方式。
+- 镜头运动、主体、动作、环境如何描述。
+- API 暴露的尺寸、时长及其他参数。
+- 一次生成几个候选。
+- 哪些失败值得重试。
+- 重试时应该改 Prompt、换参考素材还是改制作表。
+- 哪个生成结果值得发布。
+
+Seedance 官方页面说明它是文本/图像驱动的多镜头生成模型；具体能控制哪些字段，要以你最终签约 API 暴露的参数为准。[ByteDance Seedance](https://seed.bytedance.com/en/seedance)
+
+## 可以训练或学习的外层模型
+
+我们可以自己训练：
+
+1. `Prompt Compiler`
+   - 把制作表翻译成 Seedance Prompt。
+
+2. `Prompt Ranker`
+   - 在调用 Seedance 前预测哪个 Prompt 更可能成功。
+
+3. `Video Quality Ranker`
+   - 在多个生成结果中选择最佳视频。
+
+4. `Retry Policy`
+   - 判断失败后改哪一层。
+
+5. `Budget Allocator`
+   - 判断一个内容值得生成 1 次、3 次还是停止。
+
+6. `Mission Planner Policy`
+   - 根据历史决定下一步创建什么 Task。
+
+7. `Outcome Predictor`
+   - 预测哪个视频更可能满足业务目标。
+
+这些都不需要 Seedance 开放内部梯度。
+
+---
+
+# 五、用户数据确实可以形成学习算法，但不能直接拿点击量当奖励
+
+假设视频 A：
+
+- 标题非常夸张。
+- 点击率很高。
+- 用户三秒后退出。
+- 负反馈很多。
+
+视频 B：
+
+- 点击率略低。
+- 完播率高。
+- 收藏和关注转化高。
+
+如果只奖励点击，系统很快会学成标题党。
+
+应该保存多维结果：
+
+```text
+UserOutcome
+├── impressions
+├── click_through_rate
+├── 3s_retention
+├── normalized_watch_time
+├── completion_rate
+├── rewatch_rate
+├── like_rate
+├── comment_rate
+├── save_rate
+├── share_rate
+├── follow_conversion
+├── hide_or_not_interested
+├── unfollow
+└── later_return
+```
+
+注意：
+
+- 点击必须除以曝光量。
+- 观看时长必须按视频长度归一化。
+- 完播率会偏爱短视频。
+- 点赞比较轻，收藏/分享通常代表不同价值。
+- 评论可能是喜欢，也可能是争议。
+- 平台分发流量会产生巨大干扰。
+- 不同账号、题材、发布时间不能直接横向比较。
+
+Google也在研究用用户再次回访等行为作为长期体验代理，而不仅是短期点击。[Google：Surrogate for Long-Term User Experience](https://research.google/pubs/surrogate-for-long-term-user-experience-in-recommender-systems/)
+
+---
+
+# 六、最大难题不是奖励，而是信用分配
+
+最终视频数据好，究竟是因为：
+
+- 选题好？
+- 标题好？
+- 脚本开头好？
+- 制作设计好？
+- Seedance 恰好生成得好？
+- 封面好？
+- 发布时间好？
+- 平台给了更好的流量？
+
+如果全部一起变化，算法学不到可靠结论。
+
+## 必须保留全链路 Lineage
+
+```text
+用户曝光
+→ video_id
+→ generation_attempt_id
+→ production_artifact_id
+→ script_artifact_id
+→ topic_artifact_id
+→ 每层 Decision ID
+```
+
+同时记录：
+
+- 使用的模型和模型版本。
+- 完整请求参数。
+- Prompt digest。
+- 参考素材 digest。
+- 生成时间。
+- 候选数量。
+- 为什么选中这个结果。
+- 发布平台和账号。
+- 发布时间。
+- 初始流量来源。
+- 实验组。
+- 当时采用的 Policy 版本。
+
+否则模型或 API 一升级，过去经验就会被错误复用。
+
+## 早期实验一次只变一层
+
+例如：
+
+### 测脚本
+
+```text
+同一选题
+同一制作规范
+同一视频风格
+只改变脚本结构
+```
+
+### 测制作表
+
+```text
+同一选题
+同一脚本文案
+只改变镜头和视觉呈现
+```
+
+### 测 Seedance Prompt
+
+```text
+同一制作表
+同一参考素材
+只改变 Prompt 编译方式
+```
+
+这样才能知道提升来自哪一层。
+
+---
+
+# 七、最适合你的学习算法不是一上来做完整 RL
+
+建议按四步走。
+
+## 第一步：A/B 实验和完整日志
+
+先建立：
+
+- Variant。
+- Exposure。
+- Outcome。
+- Propensity/实验分流概率。
+- 全链路 Artifact lineage。
+- 人工采用和修改记录。
+
+没有这层数据，后续所有“学习”都可能学错因果。
+
+## 第二步：Contextual Bandit
+
+Contextual Bandit 非常适合：
+
+> 面对这个账号、题材、受众和历史表现，应该选择哪个候选方案？
+
+它在每次决策中：
+
+- 大部分选择当前预计最好的方案。
+- 少量探索不确定的新方案。
+- 根据实际用户反馈更新选择策略。
+
+可用于：
+
+- 选题候选选择。
+- 脚本路径选择。
+- 制作风格选择。
+- Prompt 模板选择。
+- 候选生成数量选择。
+
+这比端到端 RL 简单，也更容易解释。
+
+## 第三步：黑盒 Prompt 优化
+
+对 Seedance 建立：
+
+```text
+Prompt候选
+→ 调用Seedance
+→ 视频质量评价
+→ 新Prompt候选
+```
+
+可以使用：
+
+- OPRO 类语言优化。
+- 进化搜索。
+- Bayesian Optimization。
+- Best-of-N + Ranker。
+- Prompt 模板 Bandit。
+
+## 第四步：Offline RL 或 Planner Policy Learning
+
+当积累了大量可靠轨迹后，再学习：
+
+```text
+Mission State
+→ 下一步创建什么Task
+→ 哪条路径继续
+→ 哪条路径停止
+→ 预算怎样分配
+```
+
+不要直接让 RL 学“整篇脚本怎么写”；让它先学习“怎么组织搜索和资源”。
+
+---
+
+# 八、建议的奖励结构
+
+仍然不要一个简单总分，而是约束式多目标：
+
+```text
+第一层：硬性门禁
+安全、版权、品牌、完整性、事实、技术质量
+
+第二层：内容质量
+Goal实现、叙事、具体性、人设、信息增量
+
+第三层：生成质量
+Prompt遵循、人物一致、动作合理、镜头连续、音画质量
+
+第四层：用户价值
+留存、完播、收藏、分享、关注、长期回访
+
+第五层：成本
+Token、Seedance调用次数、人工修改时间、总时延
+```
+
+选择顺序是：
+
+```text
+先满足硬门禁
+→ 再比较内容与生成质量
+→ 再比较真实用户价值
+→ 同等价值下优化成本
+```
+
+如果算法最终必须接收一个标量，也应该由上述层级经过受控转换产生，而不是直接：
+
+```text
+reward = 点击量
+```
+
+---
+
+# 九、Human 最终应该负责什么
+
+Human 不应该每天给几千个脚本逐条打分。
+
+Human 应该负责：
+
+- 定义什么是好内容。
+- 定义 Constraint 和长期价值。
+- 对少量关键候选做 A/B。
+- 处理 AI Judge 分歧。
+- 审核新题材和高风险内容。
+- 标记真实喜欢与不喜欢。
+- 定期检查系统是否学歪。
+- 修改 Constitution、Rubric 和奖励边界。
+
+AI负责：
+
+- 大规模初筛。
+- 确定性检查。
+- 语义比较。
+- 从人工修改中提取偏好。
+- 发现异常样本。
+- 预测 Human 可能选择什么。
+
+用户行为负责：
+
+- 验证内容在真实世界是否有效。
+- 提供延迟、带噪声的业务反馈。
+
+三者不是互相替代，而是分工。
+
+---
+
+## 最终判断
+
+你的设想可以做成真正会迭代的智能创作系统，而且 Seedance 是黑盒并不构成根本障碍。
+
+我们无法训练 Seedance 本体，但可以逐渐学会：
+
+> 什么选题、什么脚本路径、什么制作结构、什么参考素材和什么 Prompt，在什么账号与受众条件下，更容易得到合格视频和真实用户价值。
+
+最关键的架构不是先引入 PPO 或某个 RL 库，而是先打通：
+
+```text
+Topic Decision
+→ Script Decision
+→ Production Decision
+→ Seedance Request
+→ Generated Video
+→ Publication Exposure
+→ User Outcome
+```
+
+并保证每个 Outcome 能追溯到完整 Decision lineage。
+
+一旦这条链可靠，先做 A/B 和 Contextual Bandit，随后做 Prompt 黑盒优化，最后再考虑 Offline RL。这样既保留第二版的动态发散，又能让系统从真实视频结果中逐渐学会“哪些发散值得继续”。