"""Accepted inbound turns have one durable owner, including exception fallback."""

import json
import subprocess
import sys
from pathlib import Path

from agent.turn_failure_copy import FAILED_TURN_DISPLAY_KIND, PARTIAL_FAILED_TURN_NOTICE


def test_gateway_failure_writer_preserves_accepted_turn_identity(tmp_path):
    root = Path(__file__).resolve().parents[2]
    receipt = tmp_path / "receipt.json"
    result = subprocess.run(
        [
            sys.executable,
            str(root / "evals/gateway_failure_ownership/probe.py"),
            str(root),
            str(receipt),
        ],
        capture_output=True,
        text=True,
        stdin=subprocess.DEVNULL,
        timeout=90,
    )
    assert receipt.exists(), result.stdout + result.stderr
    data = json.loads(receipt.read_text())
    assert all(row["reached"] for row in data["observations"]), data
    assert data["passed"] == data["total"], data["observations"]
    assert result.returncode == 0, result.stdout + result.stderr


def test_failure_owner_follows_only_live_lineage_markers(tmp_path):
    import asyncio
    import sqlite3
    from gateway.config import GatewayConfig, Platform
    from gateway.platforms.event import MessageEvent
    from gateway.run import GatewayRunner
    from gateway.session import SessionSource, SessionStore

    async def check():
        store = SessionStore(tmp_path / "sessions", GatewayConfig())
        runner = object.__new__(GatewayRunner)
        runner.session_store = store

        async def stop_typing(event, source):
            return None

        runner._hmwa_stop_typing_for_turn = stop_typing
        # An archived ancestor or the published live child can own a turn. Neither
        # a reaped sibling nor undone/observed/foreign-marker rows can own it.
        cases = ("ancestor", "middle-ancestor", "live-child", "reaped", "undone", "observed", "foreign", "unmarked")
        for index, (location, pid) in enumerate(
            (location, pid) for location in cases for pid in (None, "100")
        ):
            source = SessionSource(
                platform=Platform.TELEGRAM, chat_id=f"fixture-{index}", user_id="fixture"
            )
            entry = store.get_or_create_session(source)
            sid = entry.session_id
            db = store._db_for_session_id(sid)
            owner = f"accepted-owner-{index}"
            metadata = {"gateway_input_owner": owner}
            prepared = runner._PreparedTurn(
                [], "", "same", [{"type": "text", "text": "same"}],
                1700000000, None, sid, owner,
            )
            db.append_message(sid, "user", "same [screenshot]")
            middle = sid + "-middle"
            db.create_session(middle, source="telegram", parent_session_id=sid)
            orphan = sid + "-orphan"
            child = sid + "-live"
            db.create_session(orphan, source="telegram", parent_session_id=middle)
            db.end_session(orphan, "ws_orphan_reap")
            db.create_session(child, source="telegram", parent_session_id=middle)
            target = {"ancestor": sid, "middle-ancestor": middle, "reaped": orphan}.get(location, child)
            marker = {"gateway_input_owner": "foreign-writer"} if location == "foreign" else metadata
            current_id = db.append_message(
                target, "user", "same [screenshot]", platform_message_id=pid,
                observed=location == "observed",
                display_metadata=None if location == "unmarked" else marker,
            )
            db.end_session(sid, "compression")
            db.end_session(middle, "compression")
            if location in ("ancestor", "middle-ancestor", "undone"):
                with sqlite3.connect(db.db_path) as conn:
                    conn.execute(
                        "UPDATE messages SET active=0, compacted=? WHERE id=?",
                        (int(location != "undone"), current_id),
                    )
            store._publish_transcript_reroute(sid, child)
            owned = location in ("ancestor", "middle-ancestor", "live-child")
            assert store.has_input_owner(sid, owner) is owned, location
            # A restarted store has no published map and must pick the same live child.
            store._transcript_reroutes.clear()
            assert store.has_input_owner(sid, owner) is owned, location
            before = db.message_count()
            open_tail = store.transcript_tail_role(sid) == "user"
            reply = await runner._hmwa_agent_error_reply(
                RuntimeError("controlled post-compaction failure"),
                MessageEvent(text="same", source=source, message_id=pid),
                source, entry, entry.session_key, prepared,
            )
            # Exception fallback adds the missing user only when unowned, and a boundary iff that
            # leaves an open user tail on the live route — never anything else.
            assert db.message_count() == before + (not owned) + ((not owned) or open_tail), location
            assert store.has_input_owner(sid, owner), location
            assert PARTIAL_FAILED_TURN_NOTICE in reply
            live_messages = db.get_messages(child)
            assert not live_messages or live_messages[-1]["role"] != "user", location
            assert sum(m["role"] == "assistant" for m in live_messages) <= 1, location
            if not owned:
                assert live_messages[-1]["content"] == PARTIAL_FAILED_TURN_NOTICE
                persisted_user = live_messages[-2]
                assert persisted_user["content"] == prepared.persist_user_message
                assert persisted_user["display_metadata"]["gateway_input_owner"] == owner
            # A repeated delivery of the same closed turn adds no second boundary.
            closed = db.message_count()
            await runner._hmwa_agent_error_reply(
                RuntimeError("redelivered"), MessageEvent(text="same", source=source, message_id=pid),
                source, entry, entry.session_key, prepared,
            )
            assert db.message_count() == closed, location
        db.close()

    asyncio.run(check())


def test_context_overflow_exception_persists_nothing(tmp_path):
    """Exception-path overflow (400/500 on a long session) must not write the user row or a
    boundary — the same no-grow rule as the persist path (#1630)."""
    import asyncio
    from gateway.config import GatewayConfig, Platform
    from gateway.platforms.event import MessageEvent
    from gateway.run import GatewayRunner
    from gateway.session import SessionSource, SessionStore

    async def check():
        store = SessionStore(tmp_path / "sessions", GatewayConfig())
        runner = object.__new__(GatewayRunner)
        runner.session_store = store

        async def stop_typing(event, source):
            return None

        runner._hmwa_stop_typing_for_turn = stop_typing
        source = SessionSource(platform=Platform.TELEGRAM, chat_id="overflow", user_id="u")
        entry = store.get_or_create_session(source)
        db = store._db_for_session_id(entry.session_id)
        prepared = runner._PreparedTurn(
            [{"role": "user", "content": "x"}] * 51, "", "x", "x", None, None, entry.session_id, "owner-overflow",
        )
        err = RuntimeError("payload too large")
        err.status_code = 400
        before = db.message_count()
        reply = await runner._hmwa_agent_error_reply(
            err, MessageEvent(text="x", source=source, message_id="m-overflow"), source, entry, entry.session_key, prepared,
        )
        assert db.message_count() == before
        assert "/compress" in reply and "/new" in reply
        db.close()

    asyncio.run(check())


def test_context_overflow_error_reply_carries_no_partial_effect_notice():
    """Overflow is a deterministic rejection (#107567); the reply must stay the /compress
    guidance alone rather than inherit the indeterminate "actions may have run" warning."""
    import asyncio
    from gateway.config import Platform
    from gateway.platforms.event import MessageEvent
    from gateway.run import GatewayRunner
    from gateway.session import SessionSource

    runner = object.__new__(GatewayRunner)

    async def stop_typing(event, source):
        return None

    runner._hmwa_stop_typing_for_turn = stop_typing
    source = SessionSource(platform=Platform.TELEGRAM, chat_id="c", user_id="u")
    prepared = runner._PreparedTurn([{"role": "user", "content": "x"}] * 51, "", None, None, None, None)
    err = RuntimeError("payload too large")
    err.status_code = 400

    reply = asyncio.run(runner._hmwa_agent_error_reply(
        err, MessageEvent(text="x", source=source), source, None, "k", prepared,
    ))

    from gateway.run import _context_overflow_reply
    assert reply == _context_overflow_reply()
    assert PARTIAL_FAILED_TURN_NOTICE not in reply


def test_fresh_session_agent_flushed_failed_turn_is_closed(tmp_path):
    """First turn of a new session, agent-persisted runtime: the agent's turn-start flush wrote the
    user row (platform id stamped) before the gateway appends ``session_meta``. The boundary must
    still land — ``session_meta`` is stripped before the model sees history, so an open user row
    behind it is exactly the #107070 replay shape."""
    import asyncio
    from gateway.config import GatewayConfig, Platform
    from gateway.platforms.event import MessageEvent
    from gateway.run import GatewayRunner
    from gateway.session import SessionSource, SessionStore

    async def check():
        store = SessionStore(tmp_path / "sessions", GatewayConfig())
        runner = object.__new__(GatewayRunner)
        runner.session_store = store
        runner._session_db = object()  # agent_persisted defaults True

        async def noop(*args, **kwargs):
            return None

        runner._refresh_agent_cache_message_count = noop
        source = SessionSource(platform=Platform.TELEGRAM, chat_id="fresh", user_id="u")
        entry = store.get_or_create_session(source)
        sid = entry.session_id
        db = store._db_for_session_id(sid)
        db.append_message(sid, "user", "reset the password", platform_message_id="m-1")
        failed = {"failed": True, "final_response": "429", "error": "429", "messages": [],
                  "history_offset": 0, "last_prompt_tokens": 0}
        for _ in range(2):  # second pass = platform redelivery of the same failed message
            await runner._hmwa_persist_turn_transcript(
                event=MessageEvent(text="reset the password", source=source, message_id="m-1"),
                source=source, session_entry=entry, session_key=entry.session_key, agent_result=failed,
                agent_messages=[], prepared=runner._PreparedTurn([], "", "reset the password", None, None, None, sid, "o"),
                response="x", agent_failed_early=True, hidden_reasoning_incomplete=False,
                is_context_overflow_failure=False,
            )
        rows = [m for m in db.get_messages(sid) if m["role"] != "session_meta"]
        assert [m["role"] for m in rows] == ["user", "assistant"]
        # Typed like the core closer's row, or Desktop reads the boundary as the model's reply.
        assert rows[-1]["display_kind"] == FAILED_TURN_DISPLAY_KIND
        assert store.transcript_tail_role(sid) == "assistant"
        db.close()

    asyncio.run(check())