import json from datetime import UTC, datetime 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 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.enable_memory is False assert config.enable_research_flow is False