"""Write path for the assistant: propose, show, confirm, execute.

The model never executes a mutation. It may only *propose* one, from a fixed
whitelist, and the proposal is inert until the employee has looked at the exact
parameters and pressed a button. Four properties make that more than a slogan:

1. **Whitelist, server-side.** Only actions in :mod:`app.assistant.actions` can
   ever be proposed; an action name the model invents resolves to nothing.
2. **The server owns the parameters.** They are validated and normalised at
   propose time and stored. The confirm step re-reads the stored row — the model
   is not consulted again and cannot influence what runs.
3. **The snapshot is bound to the click.** The confirmation view renders a hash
   of the stored parameters, and the confirm call must echo that hash back. If
   anything about the proposal changed between looking and pressing, the confirm
   fails instead of executing something the employee did not read.
4. **One press, by that person, for a short while.** A single-use action token
   scoped to the proposal, plus a row-level expiry and a status machine that only
   moves forward.

Pressing confirm is therefore not "trusting the AI" — it is signing a form the
employee has read, and execution goes through the domain's own service with its
existing RBAC, idempotency, audit and approval behaviour.
"""
from __future__ import annotations

import hashlib
import json
import logging
import uuid
from dataclasses import dataclass
from datetime import UTC, datetime, timedelta
from typing import Any, cast

from sqlalchemy import update
from sqlalchemy.ext.asyncio import AsyncSession

from app.assistant.models import (
    STATUS_CONFIRMED,
    STATUS_EXECUTED,
    STATUS_EXPIRED,
    STATUS_PENDING,
    AssistantActionProposal,
)
from app.assistant.outcome_contract import MISS_LINK_ERROR
from app.core.exceptions import (
    AppException,
    is_connection_lost_db_error,
    is_transaction_aborting_db_error,
)
from app.core.security import (
    create_action_token,
    decode_action_token,
    release_action_jti,
)
from app.identity.person_models import IdentityPerson

logger = logging.getLogger(__name__)

ACTION_TOKEN_PURPOSE = "assistant_action"
#: A proposal is a live offer, not a standing authorisation. Short enough that a
#: forgotten card in a chat history cannot be pressed days later into a state
#: that no longer matches; long enough to survive a meeting.
PROPOSAL_TTL_MINUTES = 60


#: Named once because a caller has to branch on it: the card path treats this
#: code as "your earlier tap already did it" rather than as a failure.
CODE_ALREADY_HANDLED = "ASSISTANT_PROPOSAL_ALREADY_HANDLED"


class ProposalError(AppException):
    def __init__(self, code: str, message: str, status_code: int = 400) -> None:
        super().__init__(code=code, message=message, status_code=status_code)


def params_hash(params: dict[str, Any]) -> str:
    """Stable digest of the exact parameters an employee is being shown."""
    canonical = json.dumps(params, sort_keys=True, ensure_ascii=False, default=str)
    return hashlib.sha256(canonical.encode("utf-8")).hexdigest()


@dataclass(frozen=True)
class IssuedProposal:
    proposal: AssistantActionProposal
    token: str


async def create_proposal(
    db: AsyncSession,
    *,
    person: IdentityPerson,
    action_key: str,
    params: dict[str, Any],
    conversation_id: uuid.UUID | None,
    trace_id: str,
) -> IssuedProposal:
    """Store a validated proposal and mint its single-use confirmation token."""
    now = datetime.now(UTC)
    proposal = AssistantActionProposal(
        person_id=person.id,
        conversation_id=conversation_id,
        action_key=action_key,
        params=params,
        params_hash=params_hash(params),
        status=STATUS_PENDING,
        expires_at=now + timedelta(minutes=PROPOSAL_TTL_MINUTES),
        trace_id=trace_id,
    )
    db.add(proposal)
    await db.flush()
    token = create_action_token(
        purpose=ACTION_TOKEN_PURPOSE,
        subject_id=str(person.id),
        resource_id=str(proposal.id),
        # Matched to the offer, not to a day. A 24h token on a 60-minute
        # proposal meant a forwarded card kept reading the confirmation page —
        # including the post-execution result — for 23 hours after the offer
        # itself was dead.
        ttl_minutes=PROPOSAL_TTL_MINUTES,
    )
    return IssuedProposal(proposal=proposal, token=token)


async def load_for_token(
    db: AsyncSession, token: str,
) -> tuple[AssistantActionProposal, dict[str, Any]]:
    """Resolve a confirmation token to its proposal, or explain the refusal.

    Returns the proposal and the decoded token payload. Does NOT consume the
    single-use marker: the preview must be re-openable, only the confirm burns.
    """
    try:
        payload = decode_action_token(token, expected_purpose=ACTION_TOKEN_PURPOSE)
    except Exception as exc:  # noqa: BLE001 — expired / forged / wrong purpose
        raise ProposalError(
            "ASSISTANT_PROPOSAL_TOKEN_INVALID",
            "This confirmation link is no longer valid.",
            status_code=401,
        ) from exc

    # ``decode_action_token`` requires exp/iat/sub/type/iss but NOT ``res``,
    # so a token carrying the right purpose and no resource id reaches here and
    # ``uuid.UUID(str(None))`` raises out of the handler as a 500 error envelope
    # rather than as the 401 every other bad-token path returns. Only Helix
    # mints these today; treating a shape we cannot read as "invalid link" is
    # the same refusal, one layer earlier.
    try:
        resource_id = uuid.UUID(str(payload.get("res")))
    except (ValueError, AttributeError, TypeError) as exc:
        raise ProposalError(
            "ASSISTANT_PROPOSAL_TOKEN_INVALID",
            "This confirmation link is no longer valid.",
            status_code=401,
        ) from exc

    proposal = await db.get(AssistantActionProposal, resource_id)
    if proposal is None:
        raise ProposalError(
            "ASSISTANT_PROPOSAL_NOT_FOUND", "This request no longer exists.",
            status_code=404,
        )
    # NOT the thing that stops a forwarded link. ``payload["sub"]`` and
    # ``proposal.person_id`` are minted together, so this compares a value with
    # itself — it catches a token pointed at the WRONG PROPOSAL and nothing
    # else. The sentence that used to sit here claimed the opposite and was
    # believed through six review rounds. What actually stops a forwarded link
    # is the signed-in-person check in ``router.confirm_proposal``.
    if str(proposal.person_id) != str(payload.get("sub")):
        raise ProposalError(
            "ASSISTANT_PROPOSAL_FORBIDDEN", "This request belongs to someone else.",
            status_code=403,
        )
    return proposal, payload


def assert_actionable(proposal: AssistantActionProposal) -> None:
    """Refuse a proposal that is already used, or past its own expiry."""
    if proposal.status != STATUS_PENDING:
        raise ProposalError(
            CODE_ALREADY_HANDLED,
            "This request has already been handled.",
            status_code=409,
        )
    expires_at = proposal.expires_at
    if expires_at is not None:
        if expires_at.tzinfo is None:
            expires_at = expires_at.replace(tzinfo=UTC)
        if expires_at < datetime.now(UTC):
            raise ProposalError(
                "ASSISTANT_PROPOSAL_EXPIRED",
                "This request expired. Ask the assistant again.",
                status_code=410,
            )


async def confirm_and_execute_for_person(
    db: AsyncSession,
    *,
    proposal_id: uuid.UUID,
    person: IdentityPerson,
    trace_id: str,
) -> AssistantActionProposal:
    """Execute a proposal the employee confirmed by tapping the card itself.

    No token, and none is carried: the tap names the pressing account in the
    callback envelope, and the button's payload carries only an id. What
    authenticates that envelope is the shared verification token — Feishu sends
    NO ``X-Lark-Signature`` on a card callback (see ``approval/router.py``), so
    this is the same trust level every approval callback runs on, not a
    per-request signature. An earlier version of this paragraph claimed a
    "signature-checked envelope" and was simply wrong; the same false claim was
    corrected in ``card_actions`` and missed here.

    What that means in practice: the operator ids are body content, so anyone
    holding the verification token could name any account. The proposal id being
    a uuid4 is what makes that unusable rather than the identity check — worth
    knowing before this path is widened.

    "Exactly once" rests on the same conditional claim the link path relies on —
    which that path's own comment already names as the authority, since the
    Redis single-use marker fails open.
    """
    proposal = await db.get(AssistantActionProposal, proposal_id)
    if proposal is None:
        raise ProposalError(
            "ASSISTANT_PROPOSAL_NOT_FOUND", "This request no longer exists.",
            status_code=404,
        )
    if str(proposal.person_id) != str(person.id):
        # A card forwarded into a group would otherwise let a colleague submit
        # in the requester's name.
        raise ProposalError(
            "ASSISTANT_PROPOSAL_FORBIDDEN", "This request belongs to someone else.",
            status_code=403,
        )
    assert_actionable(proposal)

    if not await _claim(db, proposal):
        raise ProposalError(
            CODE_ALREADY_HANDLED,
            "This request has already been handled.",
            status_code=409,
        )
    return await _run_claimed(
        db, proposal=proposal, person=person, trace_id=trace_id, payload=None,
    )


async def _claim(db: AsyncSession, proposal: AssistantActionProposal) -> bool:
    """Conditional pending → confirmed. The row, not Redis, decides the race."""
    claimed = await db.execute(
        update(AssistantActionProposal)
        .where(
            AssistantActionProposal.id == proposal.id,
            AssistantActionProposal.status == STATUS_PENDING,
        )
        .values(status=STATUS_CONFIRMED, confirmed_at=datetime.now(UTC)),
    )
    await db.commit()
    if cast("Any", claimed).rowcount != 1:
        return False
    await db.refresh(proposal)
    return True


async def _run_claimed(
    db: AsyncSession,
    *,
    proposal: AssistantActionProposal,
    person: IdentityPerson,
    trace_id: str,
    payload: dict[str, Any] | None,
) -> AssistantActionProposal:
    """Run a proposal already claimed by exactly one confirmer."""
    from app.assistant.actions import get_action  # noqa: PLC0415 — cycle

    action = get_action(proposal.action_key)
    if action is None:
        raise ProposalError(
            "ASSISTANT_ACTION_UNKNOWN", "This request can no longer be submitted.",
            status_code=410,
        )

    # Captured BEFORE the attempt: the failure branch rolls back, and a
    # rollback expires every instance in the session — reading ``proposal.id``
    # afterwards would trigger a lazy refresh and blow up with MissingGreenlet
    # inside the error handler itself.
    proposal_id = proposal.id
    params_snapshot = dict(proposal.params or {})

    try:
        result = await action.execute(
            db, person=person, params=params_snapshot,
            proposal_id=proposal_id, trace_id=trace_id,
        )
    except Exception as exc:  # noqa: BLE001 — record, surface, allow a retry
        # Back to ``pending``, not to a terminal ``failed``: ``assert_actionable``
        # refuses anything that is not pending, so parking the row in ``failed``
        # would make the released token useless and the card permanently dead —
        # the employee would be told "already handled" for something that never
        # happened. The attempt is still visible in ``result``.
        #
        # ``attempt_count`` / ``last_attempted_at`` / ``last_error_code`` are
        # bumped in this same statement because the rollback above discards
        # anything set before the call. Without them, resetting to ``pending``
        # made "pressed and it failed" byte-identical to "never pressed" in
        # every queryable column — which is how production came to hold 14
        # proposals that all read as unexecuted while two of them had in fact
        # been pressed, and failed on a parameter contract.
        await db.rollback()
        await db.execute(
            update(AssistantActionProposal)
            .where(AssistantActionProposal.id == proposal_id)
            .values(
                status=STATUS_PENDING,
                confirmed_at=None,
                attempt_count=AssistantActionProposal.attempt_count + 1,
                last_attempted_at=datetime.now(UTC),
                last_error_code=type(exc).__name__[:64],
                result={
                    "last_error": type(exc).__name__,
                    "message": str(exc)[:500],
                },
            ),
        )
        await db.commit()
        if payload is not None:
            await release_action_jti(payload)
        logger.exception("assistant: proposal %s failed", proposal_id)
        raise ProposalError(
            "ASSISTANT_ACTION_FAILED",
            "Submitting this request failed. Please try again or contact IT.",
            status_code=500,
        ) from exc

    proposal.status = STATUS_EXECUTED
    proposal.result = result
    proposal.attempt_count = (proposal.attempt_count or 0) + 1
    proposal.last_attempted_at = datetime.now(UTC)
    proposal.last_error_code = None

    # ``executed`` means the call returned — nothing more. Which row it created,
    # and therefore whether anything actually happened for the employee, is
    # recorded here and read later from the domain that owns it.
    #
    # Linking is inside this transaction so a link can never outlive its
    # execution nor go missing after one, but a link failure must not fail the
    # submission: for a curated action the employee's row is written through
    # this same session and committed just below, and reporting a genuine
    # success as a failure would both mislead them and — because the failure
    # path resets to ``pending`` — offer the action again.
    from app.assistant.object_links import link_executed_proposal  # noqa: PLC0415

    try:
        _link, miss = await link_executed_proposal(
            db, proposal=proposal, result=result, trace_id=trace_id,
        )
    except Exception as exc:  # noqa: BLE001 — see above
        # Two families a SAVEPOINT cannot rescue: the transaction is already
        # gone server-side (1205/1213/1305), or the connection is (2006/2013).
        # Swallowing those would claim a session that is not usable, and the
        # ``commit`` below would fail anyway — with the real cause lost. The
        # same predicate ``core.txn.best_effort`` uses, rather than a second
        # opinion about which errors are survivable.
        if is_transaction_aborting_db_error(exc) or is_connection_lost_db_error(exc):
            raise
        logger.exception(
            "assistant: linking proposal %s to its business object failed",
            proposal_id,
        )
        miss = MISS_LINK_ERROR
    proposal.link_miss_reason = miss

    await db.commit()
    return proposal


async def expire_if_lapsed(db: AsyncSession, proposal: AssistantActionProposal) -> bool:
    """Move one lapsed proposal to ``expired`` so the record reads truthfully.

    Called on the read path rather than from a sweeper: the only moment anyone
    cares is when someone looks at it, and a cron that exists to fix a display
    string is a cron to maintain forever. Returns whether it lapsed.
    """
    expires_at = proposal.expires_at
    if proposal.status != STATUS_PENDING or expires_at is None:
        return False
    if expires_at.tzinfo is None:
        expires_at = expires_at.replace(tzinfo=UTC)
    if expires_at >= datetime.now(UTC):
        return False
    await db.execute(
        update(AssistantActionProposal)
        .where(
            AssistantActionProposal.id == proposal.id,
            AssistantActionProposal.status == STATUS_PENDING,
        )
        .values(status=STATUS_EXPIRED),
    )
    await db.commit()
    await db.refresh(proposal)
    return True