script-build-and-framework-tmp/script_build_host/tests/test_mission_factory.py

import json
from datetime import UTC, datetime

from script_build_host.agents.prompts import PARAGRAPH_WORKER_PROMPT, PLANNER_PROMPT
from script_build_host.application.mission_factory import ScriptMissionFactory
from script_build_host.domain.input_snapshot import ScriptBuildInputSnapshotV1
from script_build_host.domain.records import MissionBinding


def _snapshot() -> ScriptBuildInputSnapshotV1:
    return ScriptBuildInputSnapshotV1(
        snapshot_id="11",
        script_build_id=4,
        execution_id=1,
        topic_build_id=2,
        topic_id=3,
        topic={
            "topic": {
                "id": 3,
                "result": "真实嵌套选题结果",
                "topic_direction": "fallback",
            }
        },
        account={"account_name": "每天心理学"},
        persona_points=(),
        section_patterns=(),
        strategies=(),
        prompt_manifest=(),
        datasource_manifest={},
        model_manifest={
            "presets": {
                "script_planner": {
                    "model": "planner-model",
                    "temperature": 0.1,
                    "max_iterations": 42,
                }
            }
        },
        canonical_sha256="sha256:" + "b" * 64,
        created_at=datetime.now(UTC),
    )


def _binding() -> MissionBinding:
    now = datetime.now(UTC)
    return MissionBinding(
        binding_id=1,
        script_build_id=4,
        root_trace_id="root",
        input_snapshot_id=11,
        active_direction_artifact_version_id=None,
        accepted_root_artifact_version_id=None,
        engine_version="test",
        schema_version="v1",
        created_at=now,
        updated_at=now,
    )


def test_factory_reads_nested_real_topic_and_does_not_double_prefix_digest() -> None:
    factory = ScriptMissionFactory()
    spec = factory.build_root_task_spec(_snapshot())
    assert "真实嵌套选题结果" in str(spec["objective"])
    assert "title" not in str(spec)
    assert "workflow" not in str(spec).lower()
    message = json.loads(factory.build_planner_message(_binding(), _snapshot()))
    assert message["input_sha256"] == "sha256:" + "b" * 64
    assert "Retrieval children under that Direction" in message["instruction"]
    assert "exact reason PHASE_ONE_CAPABILITY_BOUNDARY" in message["instruction"]
    config = factory.build_run_config(_binding(), _snapshot())
    assert config.agent_type == "script_planner"
    assert config.model == "planner-model"
    assert config.temperature == 0.1
    assert config.max_iterations == 42
    assert config.new_trace_id == "root"
    assert config.context["phase"] == 1
    assert config.enable_memory is False
    assert config.enable_research_flow is False


def test_phase_prompts_isolate_boundaries_and_expose_dynamic_phase_two_invariants() -> None:
    factory = ScriptMissionFactory()
    planner_prompt = " ".join(PLANNER_PROMPT.split())
    assert "PHASE_TWO_CANDIDATE_PORTFOLIO_READY" not in PLANNER_PROMPT
    assert "Direction 下创建并 ACCEPT 所需 Retrieval" in planner_prompt
    assert "阶段策略给出的精确 reason" in planner_prompt
    phase_two = factory.build_phase_two_policy(_snapshot())
    assert "PHASE_TWO_CANDIDATE_PORTFOLIO_READY" in phase_two
    assert "Create exactly one CandidatePortfolio as Root's Phase2 child" in phase_two
    assert "Plan any number and order" in phase_two
    assert "Every Phase2 contract has nonempty goal_ids" in phase_two
    assert "automatically pinned to the active accepted Direction" in phase_two
    assert "carry all Direction Goal IDs" in phase_two
    assert "Paragraph-first exploration may use a null base" in phase_two
    assert "only with a Structure covering its scope" in phase_two
    assert "selects accepted child Decision IDs" in phase_two
    assert "Host partitions the complete candidate closure" in phase_two
    assert "write scopes" in phase_two
    assert "closure fields are Host-owned" in phase_two
    assert "不能从名称或 objective 猜测" in PARAGRAPH_WORKER_PROMPT