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

from __future__ import annotations

from types import SimpleNamespace

import pytest
from agent.orchestration import TaskLedger, TaskRecord, TaskSpec
from agent.orchestration.models import AcceptanceCriterion

from script_build_host.application.task_planning_v2 import TaskPlanningServiceV2
from script_build_host.tools.registry import _planner_task_input_schema


class Store:
    def __init__(self, ledger: TaskLedger) -> None:
        self.ledger = ledger

    async def load(self, root_trace_id: str) -> TaskLedger:
        assert root_trace_id == self.ledger.root_trace_id
        return self.ledger


class Coordinator:
    def __init__(self, ledger: TaskLedger) -> None:
        self.task_store = Store(ledger)
        self.call = None

    async def replay_command(self, *_args, **_kwargs):
        return None

    async def create_task_graph(self, root_trace_id: str, drafts, **kwargs):
        self.call = (root_trace_id, drafts, kwargs)
        for document in kwargs["context_documents"]:
            self.task_store.ledger.context_documents[document.uri] = document
        for draft in drafts:
            task_id = draft["_task_id"]
            self.task_store.ledger.tasks[task_id] = TaskRecord(
                task_id,
                draft["_parent_task_id"],
                "1.1",
                [
                    TaskSpec(
                        1,
                        draft["objective"],
                        tuple(
                            AcceptanceCriterion(
                                item["criterion_id"],
                                item["description"],
                                item["hard"],
                            )
                            for item in draft["acceptance_criteria"]
                        ),
                        tuple(draft["context_refs"]),
                    )
                ],
            )
        return {"tasks": [{"task_id": drafts[0]["_task_id"]}]}

    async def get_tasks(self, _root_trace_id: str, task_ids):
        return {"tasks": [{"task_id": item} for item in task_ids]}


@pytest.mark.asyncio
async def test_v2_planning_passes_contract_documents_in_same_task_mutation() -> None:
    root = TaskRecord(
        "root-task",
        None,
        "1",
        [
            TaskSpec(
                1,
                "mission",
                (AcceptanceCriterion("done", "done"),),
            )
        ],
    )
    ledger = TaskLedger("root", "mission", "root-task", tasks={"root-task": root})
    coordinator = Coordinator(ledger)
    service = TaskPlanningServiceV2(
        coordinator=coordinator,
        bindings=SimpleNamespace(
            get_by_root=lambda _: None,
        ),
        artifacts=SimpleNamespace(),
        input_snapshots=SimpleNamespace(),
    )

    async def binding(_: str):
        return SimpleNamespace(
            script_build_id=1,
            input_snapshot_id="snapshot",
            active_direction_artifact_version_id=None,
        )

    service.bindings.get_by_root = binding

    async def input_snapshot(*_args, **_kwargs):
        return SimpleNamespace(
            snapshot_id="snapshot",
            account={},
            persona_points=(),
            section_patterns=({"pattern": "opening"},),
        )

    service.input_snapshots.get = input_snapshot
    result = await service.plan(
        intents=[{"task_kind": "direction", "objective": "choose direction"}],
        context={"root_trace_id": "root", "tool_call_id": "call-1"},
    )
    assert result["task_ids"]
    assert coordinator.call is not None
    _, drafts, kwargs = coordinator.call
    documents = kwargs["context_documents"]
    assert len(documents) == len(drafts) == 1
    assert documents[0].uri in drafts[0]["context_refs"]
    assert documents[0].payload["schema_version"] == "script-task-contract/v2"


def test_planner_schema_exposes_business_intent_only() -> None:
    properties = _planner_task_input_schema()["properties"]
    assert "decision_ids" in properties
    assert "scope_selector" not in properties
    assert "criteria" not in properties
    assert "parent_task_id" not in properties