"""Close the loop on what the assistant submitted, without waiting to be asked.

Everything else in this feature answers on demand. That leaves one gap: a
request whose approval landed last Tuesday stays ``active`` forever unless the
employee happens to come back and tap something. The funnel then reads as
"nobody's requests ever complete", which is false, and the abandonment sweep
eventually files it as ``abandoned``, which is worse — it is the platform
recording its own success as a disappearance.

This sweep refreshes the live phase of every non-terminal submission and closes
the service requests behind the ones that landed.

**It sends nothing.** No new push: the domains already notify on their own
terminal states (an approval that passes tells the applicant), and adding a
second "by the way, that thing you asked for worked" would be the platform
talking about itself. Results reach people when they ask, when they reply to a
card, or through the notification the owning domain already sends.

**It never closes on a failure.** ``rejected`` and ``failed`` mark the request
``action_needs_attention`` and leave it active. A closed request nobody looked
at is exactly the outcome this whole line of work exists to stop producing.
"""
from __future__ import annotations

import logging
from datetime import UTC, datetime
from typing import Any

from sqlalchemy import or_, select
from sqlalchemy.ext.asyncio import AsyncSession

from app.assistant.models import AssistantObjectLink
from app.core.coordinators.assistant_outcome_coordinator import (
    PHASE_EFFECTIVE,
    PHASE_FAILED,
    PHASE_GONE,
    PHASE_REJECTED,
    PHASE_UNREADABLE,
    TERMINAL_PHASES,
    Outcome,
    resolve_and_cache,
)
from app.core.coordinators.assistant_support_coordinator import (
    OUTCOME_ACTION_NEEDS_ATTENTION,
    OUTCOME_SYSTEM_VERIFIED,
)
from app.core.txn import best_effort
from app.domains.support.models import (
    REQUEST_ACTIVE,
    REQUEST_SELF_RESOLVED,
    SupportServiceRequest,
)
from app.identity.person_models import IdentityPerson
from app.scheduled_jobs.registry import register_scheduled_job

logger = logging.getLogger(__name__)

#: How many links one tick will re-read. Each is a handful of indexed lookups,
#: but the bound exists so a backlog cannot turn one tick into an hour-long
#: transaction holding locks against the assistant's own write path.
MAX_LINKS_PER_RUN = 500

#: What the sweep considers finished with. Wider than ``TERMINAL_PHASES``,
#: deliberately, and only here: ``unreadable`` is not an end state of the
#: *request* (which is what that constant means everywhere else) but it IS the
#: end of anything this sweep can do — the object left the person it was linked
#: to. Folding it into the shared constant would quietly tell the employee
#: surface that a refused read is a finished request.
_SWEEP_SETTLED = frozenset(TERMINAL_PHASES) | {PHASE_UNREADABLE}


@register_scheduled_job(
    key="assistant.submission_reconcile",
    display_name="助手 · 提交结果对账",
    description=(
        "重读助手提交的业务对象的实时状态;全部生效则把对应的服务请求"
        "标记为 system_verified 关闭,出现 rejected/failed 则标记为待处理"
        "并保持 active。不发送任何通知 —— 终态通知由各领域自己发。"
    ),
)
async def run_assistant_submission_reconcile(
    db: AsyncSession, config: dict[str, Any],
) -> dict[str, Any]:
    summary: dict[str, Any] = {
        "links_checked": 0,
        "links_settled": 0,
        "requests_closed": 0,
        "requests_flagged": 0,
        "unreadable": 0,
        "errors": 0,
        "orphaned": 0,
    }

    links = list(
        (
            await db.execute(
                select(AssistantObjectLink)
                .where(
                    or_(
                        AssistantObjectLink.phase.is_(None),
                        AssistantObjectLink.phase.notin_(tuple(_SWEEP_SETTLED)),
                    )
                )
                .order_by(AssistantObjectLink.created_at.asc())
                .limit(MAX_LINKS_PER_RUN)
            )
        ).scalars().all()
    )
    summary["links_checked"] = len(links)
    if not links:
        return summary

    touched_requests: set[Any] = set()
    for link in links:
        person = await db.get(IdentityPerson, link.person_id)
        if person is None:
            # The FK to ``identity_persons`` makes this all but unreachable;
            # counted rather than passed over silently so that if it ever does
            # happen it is a number somebody can see.
            summary["orphaned"] += 1
            continue

        outcome: Outcome | None = None
        # Per-link isolation. One resolver blowing up — a domain model renamed
        # out from under a spec, a row in a shape nothing expects — must not
        # stop the other 499 from settling, which is what an unguarded loop
        # would do: the tick fails and NOTHING gets closed. Counted rather than
        # swallowed, so a resolver that is broken for everyone shows up as a
        # number climbing instead of as silence.
        async with best_effort(
            db, label=f"resolve assistant link {link.id}", log=logger,
        ) as attempt:
            outcome = await resolve_and_cache(db, link=link, person=person)
        if attempt.error is not None or outcome is None:
            # ``outcome`` is bound inside the block above, so it is None exactly
            # when that block failed. Checked explicitly rather than left to the
            # ordering of this ``continue`` — an edit that moved the branch
            # would otherwise read the PREVIOUS link's result for this one.
            summary["errors"] += 1
            continue
        # ``unreadable`` here means the object changed hands. Counted rather
        # than acted on: the sweep must not close somebody's request on the
        # strength of a row that is no longer theirs.
        #
        # And recorded, which ``resolve_and_cache`` deliberately does not do.
        # Its reason for not caching is that ``unreadable`` describes the
        # person asking rather than the object — but the person asking HERE is
        # the link's own person, so for the sweep it is a durable fact about
        # the link. Without writing it the link stays non-terminal and is
        # re-read every tick forever; the queue is oldest-first, so enough of
        # them would eventually crowd fresh links out of the per-run cap.
        if outcome.phase == PHASE_UNREADABLE:
            link.phase = PHASE_UNREADABLE
            link.phase_at = datetime.now(UTC)
            summary["unreadable"] += 1
            continue
        if outcome.is_terminal:
            summary["links_settled"] += 1
        if link.service_request_id is not None:
            touched_requests.add(link.service_request_id)

    if len(links) >= MAX_LINKS_PER_RUN:
        # Never a silent truncation — a capped run looks identical to a caught-up
        # one on every metric this returns except this flag.
        logger.warning(
            "assistant.submission_reconcile: hit the %d-link cap; the rest run "
            "on the next tick", MAX_LINKS_PER_RUN,
        )
        summary["capped"] = True

    for request_id in touched_requests:
        request = await db.get(SupportServiceRequest, request_id)
        if request is None or request.status not in (REQUEST_ACTIVE,):
            continue
        phases = list(
            (
                await db.execute(
                    select(AssistantObjectLink.phase).where(
                        AssistantObjectLink.service_request_id == request.id,
                    )
                )
            ).scalars().all()
        )
        if not phases:
            continue
        if any(p in (PHASE_REJECTED, PHASE_FAILED) for p in phases):
            request.outcome = OUTCOME_ACTION_NEEDS_ATTENTION
            summary["requests_flagged"] += 1
            continue
        if all(p in (PHASE_EFFECTIVE, PHASE_GONE) for p in phases):
            request.status = REQUEST_SELF_RESOLVED
            request.outcome = OUTCOME_SYSTEM_VERIFIED
            request.resolved_at = datetime.now(UTC)
            summary["requests_closed"] += 1

    return summary