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["context_entrypoint"] == "get_current_context"
    assert "input_snapshot_ref" not in message
    assert "input_sha256" not in message
    assert "logical evidence work" 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.no_progress.planner_stall_limit == 3
    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 "自动创建 一个 logical evidence-retrieval Task" 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 "Submit business intents for Structure, Paragraph, ElementSet" in phase_two
    assert "Host determines containers, branch identity, semantic targets" in phase_two
    assert "exactly one accepted Structure, Paragraph and ElementSet" in phase_two
    assert "CandidatePortfolio intent selects exactly one accepted Compose" in phase_two
    assert "Do not create physical Retrieval Tasks" in phase_two
    assert "scopes, artifact refs, presets, budgets or closure fields" in phase_two
    assert "不能从名称或 objective 猜测" in PARAGRAPH_WORKER_PROMPT