"""Todo8 r3 trust-boundary and uncertain-provider probes."""

from __future__ import annotations

import os
import sys
from copy import copy
from collections.abc import Callable, Mapping
from concurrent.futures import ThreadPoolExecutor
from dataclasses import replace
from pathlib import Path
from typing import TypeAlias

import pytest

PROFILE_PACKAGE = Path(__file__).resolve().parents[2] / "dualcoach" / "profile"
if str(PROFILE_PACKAGE) not in sys.path:
    sys.path.insert(0, str(PROFILE_PACKAGE))

from checkin_cli.weekly_operations_grounding import WeeklyGroundingError
from checkin_cli.weekly_operations_knowledge import (
    WeeklyKnowledgeError,
    load_shipped_weekly_public_knowledge,
)
from gateway.platforms.nutrition_weekly_owner_draft import (
    WeeklyOwnerDraftService,
    WeeklyProviderOutcomeUnknown,
    weekly_generation_base,
)
from gateway.platforms.nutrition_weekly_owner_ledger import (
    WeeklyGenerationLedger,
    WeeklyGenerationLedgerError,
    WeeklyGenerationOutcome,
    WeeklyGenerationRow,
    WeeklyGenerationState,
)
from gateway.platforms.nutrition_weekly_owner_model import JsonValue, weekly_model_request
from gateway.platforms.nutrition_weekly_owner_storage import (
    WeeklyOwnerStorageError,
)
from tests.gateway.test_nutrition_weekly_owner_draft_model import (
    FakeLifecycle,
    ModelGeneratedResponse,
    ModelRaised,
    ModelResponse,
    SequenceModel,

    owner_draft_request,
)

RowForge: TypeAlias = Callable[[WeeklyGenerationRow], WeeklyGenerationRow]
_SENTINELS = (
    "RAW-MEAL", "RAW-SYMPTOM", "RAW-FREE-TEXT", "RAW-HEALTH",
    "CUSTOMER-NAME", "CUSTOMER-KEY", "TELEGRAM-USER", "TELEGRAM-CHAT",
    "WEIGHT-RAW", "CALORIE-RAW", "DIGESTION-RAW",
)


def _valid_selection(request: Mapping[str, JsonValue]) -> Mapping[str, JsonValue]:
    return {
        "grounding_digest": request["grounding_digest"],
        "principle_ids": ["recorded_trend"], "emphasis": "summary_first",
    }


def _forge_customer(row: WeeklyGenerationRow) -> WeeklyGenerationRow:
    return replace(row, customer_digest="f" * 64)


def _forge_week(row: WeeklyGenerationRow) -> WeeklyGenerationRow:
    return replace(row, week_start="2026-08-10")


def _forge_route(row: WeeklyGenerationRow) -> WeeklyGenerationRow:
    return replace(row, owner_route_digest="e" * 64)


def _forge_aggregate(row: WeeklyGenerationRow) -> WeeklyGenerationRow:
    return replace(row, aggregate_digest="d" * 64)


def _forge_source(row: WeeklyGenerationRow) -> WeeklyGenerationRow:
    return replace(row, source_authority_digest="c" * 64)


def _forge_request(row: WeeklyGenerationRow) -> WeeklyGenerationRow:
    return replace(row, request_base_digest="b" * 64)


@pytest.mark.parametrize("sentinel", _SENTINELS)
def test_fact_model_copy_rejects_arbitrary_finite_field_even_with_digest(
    tmp_path: Path, sentinel: str,
) -> None:
    lifecycle = FakeLifecycle(tmp_path)
    facts = owner_draft_request(lifecycle).facts
    with pytest.raises(WeeklyGroundingError):
        _ = facts.model_copy(update={
            "week_status": sentinel,
            "grounding_digest": facts.grounding_digest,
        })


@pytest.mark.parametrize("sentinel", _SENTINELS)
def test_unverified_direct_knowledge_never_reaches_provider_request(
    tmp_path: Path, sentinel: str,
) -> None:
    lifecycle = FakeLifecycle(tmp_path)
    facts = owner_draft_request(lifecycle).facts
    knowledge = load_shipped_weekly_public_knowledge()
    poisoned = replace(
        knowledge,
        principles=(
            replace(knowledge.principles[0], approved_text=sentinel),
            *knowledge.principles[1:],
        ),
    )
    model = SequenceModel(())
    with pytest.raises(WeeklyKnowledgeError):
        _ = weekly_model_request(facts, poisoned, correction=False)
    assert model.calls == 0


def test_unknown_initial_is_terminal_across_restart_and_concurrency(tmp_path: Path) -> None:
    lifecycle = FakeLifecycle(tmp_path)
    request = owner_draft_request(lifecycle)
    model = SequenceModel((ModelRaised(WeeklyProviderOutcomeUnknown("after call")),))
    with ThreadPoolExecutor(max_workers=10) as pool:
        def create_one(_index: int):
            return WeeklyOwnerDraftService(lifecycle, model).create(request)

        results = tuple(pool.map(
            create_one,
            range(10),
        ))
    replay = WeeklyOwnerDraftService(lifecycle, model).create(request)
    states = tuple(row.state for row in WeeklyGenerationLedger(lifecycle.storage).read_rows())
    assert model.calls == 1
    assert all(result.accepted and result.fallback_used for result in results)
    assert (replay.provider_calls, replay.fallback_used) == (0, True)
    assert WeeklyGenerationState.UNKNOWN in states


def test_unknown_correction_is_terminal_without_third_call(tmp_path: Path) -> None:
    lifecycle = FakeLifecycle(tmp_path)
    request = owner_draft_request(lifecycle)
    model = SequenceModel((
        ModelResponse({"malformed": True}),
        ModelRaised(WeeklyProviderOutcomeUnknown("correction after call")),
    ))
    result = WeeklyOwnerDraftService(lifecycle, model).create(request)
    replay = WeeklyOwnerDraftService(lifecycle, model).create(request)
    assert (result.provider_calls, model.calls, replay.provider_calls) == (2, 2, 0)
    assert result.fallback_used and replay.fallback_used


def test_current_todo7_authority_is_final_fenced_before_draft_persistence(tmp_path: Path) -> None:
    lifecycle = FakeLifecycle(tmp_path)
    request = owner_draft_request(lifecycle)
    stale = copy(request.bound_summary)
    object.__setattr__(stale, "authority_digest", "f" * 64)
    calls = 0

    def current():
        nonlocal calls
        calls += 1
        return request.bound_summary if calls < 3 else stale

    fenced = replace(request, current_bound_summary=current)
    model = SequenceModel((ModelGeneratedResponse(_valid_selection),))
    result = WeeklyOwnerDraftService(lifecycle, model).create(fenced)
    assert (result.accepted, result.error, result.provider_calls) == (
        False, "grounding_stale", 1,
    )
    assert not lifecycle.sources


@pytest.mark.parametrize(
    "forge",
    (
        _forge_customer, _forge_week, _forge_route,
        _forge_aggregate, _forge_source, _forge_request,
    ),
)
def test_self_hashed_forged_bound_row_is_typed_request_conflict(
    tmp_path: Path, forge: RowForge,
) -> None:
    lifecycle = FakeLifecycle(tmp_path)
    request = owner_draft_request(lifecycle)
    model = SequenceModel((ModelGeneratedResponse(_valid_selection),))
    result = WeeklyOwnerDraftService(lifecycle, model).create(request)
    assert result.accepted
    ledger = WeeklyGenerationLedger(lifecycle.storage)
    with ledger.locked() as transaction:
        latest = transaction.latest(weekly_generation_base(request).logical_key)
        assert latest is not None
        _ = transaction.append(
            forge(weekly_generation_base(request)),
            state=WeeklyGenerationState.BOUND,
            outcome=WeeklyGenerationOutcome.DRAFT_BOUND,
            ordinal=latest.ordinal, request_digest=latest.request_digest,
            principle_ids=latest.principle_ids, emphasis=latest.emphasis,
            fallback=latest.fallback, source_digest=latest.source_digest,
            draft_id=latest.draft_id,
        )
    with pytest.raises(WeeklyGenerationLedgerError, match="request authority conflict"):
        _ = WeeklyOwnerDraftService(lifecycle, model).create(request)
    assert model.calls == 1


def test_cross_root_request_and_inode_swap_fail_before_provider(tmp_path: Path) -> None:
    original = FakeLifecycle(tmp_path / "original")
    request = owner_draft_request(original)
    other = FakeLifecycle(tmp_path / "other")
    model = SequenceModel(())
    denied = WeeklyOwnerDraftService(other, model).create(request)
    assert (denied.accepted, denied.provider_calls, model.calls) == (False, 0, 0)

    moved = tmp_path / "moved"
    os.rename(original.root, moved)
    original.root.mkdir(mode=0o700)
    with pytest.raises(WeeklyOwnerStorageError, match="inode drift"):
        _ = WeeklyOwnerDraftService(original, model)
    assert model.calls == 0