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_common_planner_prompt_does_not_leak_phase_two_boundary() -> None:
    factory = ScriptMissionFactory()
    assert "PHASE_TWO_CANDIDATE_PORTFOLIO_READY" not in PLANNER_PROMPT
    assert "using the returned Direction Task ID as `parent_task_id`" in PLANNER_PROMPT
    assert "`PHASE_ONE_CAPABILITY_BOUNDARY` with no additional text" in PLANNER_PROMPT
    phase_two = factory.build_phase_two_policy(_snapshot())
    assert "PHASE_TWO_CANDIDATE_PORTFOLIO_READY" in phase_two
    assert "CandidatePortfolio and Compose must use the exact parent" in phase_two
    assert "script-build://writes/paragraphs/<stable-selector>" in phase_two
    assert "script-build://writes/elements/<stable-selector>" in phase_two
    assert "Never copy scope_ref into write_scope" in phase_two
    assert "whole Structure accepted_decision_ref" in phase_two
    assert "whole artifact_ref unchanged into base_artifact_ref" in phase_two
    assert "must also include and adopt its covering" in phase_two
    assert "must partition that complete candidate list" in phase_two
    assert "candidate_decision_ids, adopted_decision_ids, and compose_order" in phase_two
    assert "derive a strategy reference" in PARAGRAPH_WORKER_PROMPT