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

from __future__ import annotations

from types import SimpleNamespace
from typing import Any

import pytest

from script_build_host.application.input_snapshot_service import (
    AssembleInputRequest,
    ScriptInputSnapshotService,
)
from script_build_host.domain.input_snapshot import ScriptBuildInput, ScriptBuildInputSnapshotV1
from script_build_host.domain.ports import PersonaInput, PromptRequest
from script_build_host.domain.records import Principal
from script_build_host.infrastructure.canonical_json import canonical_sha256
from script_build_host.repositories.sqlalchemy import SqlAlchemyInputSnapshotRepository


class FakeLegacyInput:
    def __init__(self) -> None:
        self.calls: list[tuple[int, int, int]] = []

    async def read_topic_graph(
        self, *, execution_id: int, topic_build_id: int, topic_id: int
    ) -> dict[str, Any]:
        self.calls.append((execution_id, topic_build_id, topic_id))
        return {
            "build_record": {
                "personal_config": {"account_name": "account"},
                "strategies_config": {"always_on": [7], "on_demand": [8]},
                "agent_config": {"api_key": "hidden", "model": "fake"},
            },
            "topic": {"id": topic_id},
            "points": [],
            "composition_items": [],
            "relations": [],
            "sources": [],
        }


class FakePersona:
    async def load(self, account_name: str) -> PersonaInput:
        return PersonaInput(account_name, account_name, ({"point": "one"},), (), {"p": "sha256:x"})


class FakeStrategies:
    async def load(
        self, *, always_on: tuple[Any, ...], on_demand: tuple[Any, ...]
    ) -> tuple[dict[str, Any], ...]:
        return ({"always_on": list(always_on), "on_demand": list(on_demand)},)


class FakePrompts:
    async def load(self, requests: tuple[Any, ...]) -> tuple[dict[str, Any], ...]:
        return ({"source": "fixture", "count": len(requests)},)


class FakeSnapshots:
    def __init__(self) -> None:
        self.frozen: ScriptBuildInput | None = None

    async def freeze(
        self, assembled: ScriptBuildInput, *, version: int = 1
    ) -> ScriptBuildInputSnapshotV1:
        self.frozen = assembled
        raise NotImplementedError

    async def get(self, snapshot_id: str, *, script_build_id: int) -> ScriptBuildInputSnapshotV1:
        raise NotImplementedError


@pytest.mark.asyncio
async def test_validate_source_is_read_only_and_assemble_redacts_secrets() -> None:
    legacy = FakeLegacyInput()
    service = ScriptInputSnapshotService(
        legacy_input=legacy,
        persona_source=FakePersona(),
        strategy_source=FakeStrategies(),
        prompt_source=FakePrompts(),
        snapshots=FakeSnapshots(),
    )
    await service.validate_source(execution_id=10, topic_build_id=20, topic_id=30)
    assembled = await service.assemble(
        AssembleInputRequest(
            script_build_id=1,
            execution_id=10,
            topic_build_id=20,
            topic_id=30,
            principal=Principal("user-1"),
            strategies_always_on=(7,),
            strategies_on_demand=(8,),
            datasource_manifest={"endpoint": "https://api.example?q=ok&token=hidden"},
            model_manifest={"planner": {"model": "fake", "api_key": "hidden"}},
        )
    )
    assert legacy.calls == [(10, 20, 30), (10, 20, 30)]
    assert "hidden" not in str(assembled.canonical_payload())
    assert assembled.strategies[0] == {"always_on": [7], "on_demand": [8]}


@pytest.mark.asyncio
async def test_extend_prompt_lineage_is_append_only_and_preserves_business_digest(
    database,
) -> None:
    _, sessions = database
    repository = SqlAlchemyInputSnapshotRepository(sessions)
    parent = await repository.freeze(
        ScriptBuildInput(
            script_build_id=1,
            execution_id=10,
            topic_build_id=20,
            topic_id=30,
            topic={"topic": {"id": 30}},
            account={"account_name": "acct"},
            prompt_manifest=(
                {
                    "preset": "script_planner",
                    "role": "planner",
                    "content": "phase one",
                    "content_sha256": canonical_sha256("phase one").wire,
                },
            ),
        )
    )

    class Prompts:
        async def load(self, _requests):
            return (
                {
                    "preset": "script_compose_worker",
                    "role": "worker",
                    "content": "phase two",
                    "content_sha256": canonical_sha256("phase two").wire,
                },
            )

    service = ScriptInputSnapshotService(
        legacy_input=SimpleNamespace(),
        persona_source=SimpleNamespace(),
        strategy_source=SimpleNamespace(),
        prompt_source=Prompts(),
        snapshots=repository,
    )
    child = await service.extend_prompt_lineage(
        parent,
        prompt_requests=(
            PromptRequest("compose", "compose.md", "script_compose_worker", "worker"),
        ),
        model_manifest={"presets": {"script_compose_worker": {"model": "fake"}}},
    )
    assert child.snapshot_id != parent.snapshot_id
    assert child.parent_snapshot_id == parent.snapshot_id
    assert child.parent_snapshot_sha256 == parent.canonical_sha256
    assert (
        child.business_input_sha256 == canonical_sha256(parent.to_input().business_payload()).wire
    )
    assert {item["preset"] for item in child.prompt_manifest} == {
        "script_planner",
        "script_compose_worker",
    }