"""
Human.work Agent Service — FastAPI app.

This file wires the ACP runtime and runtime services into the public agent
endpoints.

Module map:
  hermes_acp_client.py — resident hermes-run supervisor client (socket per turn)
"""
from __future__ import annotations

import logging
import os
import time
from typing import Any

from fastapi import FastAPI, HTTPException
from fastapi.responses import Response
from sentry_init import init_sentry

# #828 / P0-1: init Sentry before constructing FastAPI() so the framework
# integrations register against the running app and unhandled exceptions
# from request handlers are captured. Safe no-op when SENTRY_DSN unset.
init_sentry()

# #1230: init OpenTelemetry BEFORE constructing FastAPI() so the FastAPI
# instrumentation can patch the app and the inbound `traceparent` (sent by the
# API's undici instrumentation) is extracted into a CHILD span — giving an
# end-to-end api -> agent trace. Off by default (no-op unless OTEL_ENABLED=true
# + SIGNOZ_INGESTION_KEY set); fail-open so a telemetry error never blocks the
# agent. `instrument_app(app)` is called right after FastAPI() is constructed
# below; `shutdown_tracing()` is wired into the shutdown event.
from instrumentation.tracing import init_tracing, instrument_app, shutdown_tracing

init_tracing()

from cost_cap import COST_CAP_REASON, CostTracker
from env_utils import clean_env, has_surrounding_whitespace
from hermes_acp_client import (
    HermesAcpClient,
    HermesACPTimeoutError,
    OrgBootstrap,
    configure_stderr_relay_logging,
)
import metrics as agent_metrics
from middleware.rate_limit import RateLimiter, get_rate_limiter
import structured_logging
from models.chat import ChatRequest, ChatResponse, SessionNewRequest, SessionNewResponse


logger = logging.getLogger("humanwork.agent")


# ── Module-level singletons (constructed at import) ──────────────────────────
hermes_client = HermesAcpClient()
# US-017: tracks per-Org daily cost_usd and blocks /chat once the cap is hit.
# Tests patch `main.cost_tracker.<method>` to inject below/above-cap states.
cost_tracker = CostTracker()
# Issue #511: HTTP-level rate limiting for defense-in-depth (aligns with NestJS ThrottlerModule)
rate_limiter = get_rate_limiter()

# US-016: install the JSON formatter on the chat logger so every /chat
# completion emits one structured line CloudWatch / SigNoz can parse.
structured_logging.configure_json_logging()
# hermes-run's own stderr diagnostics (Landlock guard/exec/config lines,
# gated behind HERMES_RUN_VERBOSE) are streamed to this logger on every spawn
# path, success included — install its handler for the same reason as the
# line above: uvicorn installs no root handler at INFO by default, so
# without this the lines are computed and immediately dropped.
configure_stderr_relay_logging()


# ── App ──────────────────────────────────────────────────────────────────────
app = FastAPI(
    title="Human.work Agent Service",
    version="0.4.0",
    description="Humanwork managed Hermes ACP runtime",
)

# #1230: attach FastAPI server-span instrumentation. This reads the inbound
# `traceparent` and makes the /chat span a CHILD of the API span. No-op when
# tracing is disabled.
instrument_app(app)


# #1230: flush + shut down OTel on FastAPI shutdown (uvicorn handles SIGTERM ->
# lifespan shutdown) so batched spans are exported before the process exits.
# No-op when tracing was never started.
@app.on_event("shutdown")
async def _otel_shutdown() -> None:
    shutdown_tracing()


# ── Startup guard ────────────────────────────────────────────────────────────

# Secret/config vars whose stray surrounding whitespace used to silently break
# agent calls (#4035: a trailing space on the gateway token → illegal
# Authorization header → opaque APIConnectionError). Every read boundary now
# normalizes via clean_env, so a dirtied value is TOLERATED at runtime — this
# scan is a boot-time hygiene signal that names the exact var to clean (e.g. to
# rotate a leaked secret), turning a multi-hour "connection error" hunt into a
# one-line startup log. It is deliberately WARN-only in every environment:
# hard-failing would needlessly refuse startup on an otherwise-healthy container
# over a now-harmless value — especially a lower-priority fallback that runtime
# resolution never even consumes (e.g. a dirty OPENAI_API_KEY while the active
# HUMANWORK_MODEL_GATEWAY_TOKEN is clean). The one genuinely-fatal case — a
# required secret that is absent or effectively-empty — is caught by the
# clean_env-normalized presence check below, not here.
_WHITESPACE_SENSITIVE_ENV = (
    "AGENT_SERVICE_SECRET",
    "HUMANWORK_MODEL_GATEWAY_TOKEN",
    "HUMANWORK_MODEL_GATEWAY_URL",
    "PLATFORM_API_TOKEN",
    "PLATFORM_API_URL",
)


def _env_whitespace_problems() -> list[str]:
    """Critical env vars carrying stray surrounding whitespace (#4035).

    Detects on the RAW value (independent of clean_env), so the boot log names
    exactly which var is dirtied even though the runtime read paths now tolerate
    it. Absent/whitespace-only vars are not reported here (has_surrounding_
    whitespace is False for them) — an effectively-empty REQUIRED secret is
    handled by the presence check in the startup guard, not by this scan.
    """
    return [
        name
        for name in _WHITESPACE_SENSITIVE_ENV
        if has_surrounding_whitespace(os.getenv(name))
    ]


@app.on_event("startup")
async def _check_agent_service_secret() -> None:
    import logging as _logging
    _log = _logging.getLogger("humanwork.agent")

    is_deployed = os.getenv("APP_ENV") in ("production", "staging")

    # clean_env so a whitespace-only value ("   ") is treated as UNSET rather
    # than a truthy-but-unusable secret that boots the agent into a state where
    # every authenticated platform request 401s (the raw presence check let a
    # whitespace-only secret through). See #4035 review.
    if not clean_env(os.getenv("AGENT_SERVICE_SECRET")):
        if is_deployed:
            raise RuntimeError("AGENT_SERVICE_SECRET must be set in production/staging")
        _log.warning("AGENT_SERVICE_SECRET not set — agent auth is DISABLED (dev mode)")

    # #4035: hygiene signal only — read boundaries already tolerate whitespace,
    # so warn (never raise) and name the vars to clean. WARN in all envs.
    dirty = _env_whitespace_problems()
    if dirty:
        _log.warning(
            "Env vars have stray surrounding whitespace (tolerated at read time "
            "via clean_env, but clean the value — e.g. rotate if a secret): %s",
            ", ".join(dirty),
        )

# ── Endpoints ────────────────────────────────────────────────────────────────
# Railway platform health checks do not send X-Agent-Secret. Keep this route
# public so deploy liveness probes do not fail when AGENT_SERVICE_SECRET is set.
@app.get("/health")
async def health():
    # Public (no auth dependency) so Railway / uptime probes do not fail when
    # AGENT_SERVICE_SECRET is set. Never include secret values in health output.
    return {
        "status": "ok",
        "service": "humanwork-agent",
        "version": "0.4.0",
    }


# US-015: Prometheus scrape target. Exposes the four metrics defined in
# `metrics.py` in standard exposition format. Operators point Prometheus at
# `/metrics`; no auth is enforced because the route lives behind the
# platform's internal network boundary (same as /health).
@app.get("/metrics")
async def metrics():
    body, content_type = agent_metrics.render_metrics()
    return Response(content=body, media_type=content_type)


async def _reserve_turn_cost(req: ChatRequest):
    """Reserve this turn's budget or reject it before Hermes/session work.

    A resource limit is an HTTP failure, never an assistant message. When the
    deployment is configured alert-only, the event is logged and no synthetic
    reservation or response is introduced.
    """
    role = "unclassified"
    over_cap, current_cost, daily_cap, reservation = await cost_tracker.reserve(req.org_id)
    if not over_cap:
        return reservation

    structured_logging.emit_cost_cap_exceeded(
        org_id=req.org_id,
        current_cost_usd=current_cost,
        daily_cap_usd=daily_cap,
        agent_role=role,
        customer_id=None,
        conversation_id=req.session_id,
    )
    if cost_tracker.hard_stop_enabled():
        raise HTTPException(
            status_code=429,
            detail={"code": "COST_CAP_EXCEEDED", "message": COST_CAP_REASON},
        )
    logger.warning(
        "Cost cap reached for org %s (%.4f >= %.2f) but "
        "HUMANWORK_HARD_STOP_ENABLED=false — alert-only, proceeding.",
        req.org_id,
        current_cost,
        daily_cap,
    )
    return None


async def chat(req: ChatRequest, on_text=None, on_session=None, on_tool=None) -> ChatResponse:
    role = "unclassified"

    # Issue #511: HTTP-level rate limiting (defense-in-depth, aligns with NestJS ThrottlerModule)
    # Run before cost cap so rate-limited requests cost nothing.
    if RateLimiter.is_enabled():
        guard = rate_limiter.guard(req.org_id)
        guard.check()

    # Resource enforcement happens before session admission and
    # returns an explicit HTTP failure; it never fabricates assistant content.
    cost_reservation = await _reserve_turn_cost(req)

    # Locals initialized before the try so the except/finally handlers below
    # can reference them even if setup fails. Kept request-local (never module
    # scope) so concurrent requests never share state (P0.6).
    response_session_id = req.session_id
    # #4285: the turn's real cost, set on the success path and read by the
    # finally to settle the reservation (None on any failure → full refund).
    cost_usd: float | None = None
    chat_start = time.perf_counter()

    try:
        # ACP is the sole reasoning loop. The agent service does not classify,
        # wrap, ground, summarize, or otherwise transform the user's text.
        request_hermes_client = hermes_client
        hermes_start = time.perf_counter()
        run = await request_hermes_client.chat(
            req.prompt,
            bootstrap=OrgBootstrap.from_request(
                req.org_id,
                agentfs_url=req.agentfs_url,
                agentfs_token=req.agentfs_token,
                sessiondb_sync_url=req.sessiondb_sync_url,
                sessiondb_sync_token=req.sessiondb_sync_token,
            ),
            specialist_id=req.specialist_id,
            session_id=req.session_id,
            on_text=on_text,
            on_session=on_session,
            on_tool=on_tool,
        )
        agent_metrics.observe_hermes_invocation(time.perf_counter() - hermes_start)
        # Exact assistant output. Python does not decode, classify, gate, or
        # rewrite it, and does not fabricate confidence/review metadata.
        agentReply = run.response
        response_session_id = run.session_id
        token_input = run.usage.input_tokens if run.usage else None
        token_output = run.usage.output_tokens if run.usage else None
        # Cost has no channel from Hermes: the ACP `Usage` schema (see
        # hermes_acp_client.HermesUsage) carries only token counts, never a
        # dollar figure. Real per-turn cost is computed by the platform's
        # model-gateway (api/), which is not reachable from this process.
        # Left None here (never fabricated) so the cost-cap reservation
        # settle()s a full refund and humanwork_hermes_cost_usd_total stays
        # honest about what could actually be measured.
        cost_usd = None
    except Exception as exc:
        # Always log the real error for ops/alerting — never surface provider
        # error details, billing messages, or model error strings to end users.
        logger.error(
            "Inference failure for session_id=%s org=%s: %s",
            req.session_id,
            req.org_id,
            exc,
            exc_info=True,
        )
        if isinstance(exc, HermesACPTimeoutError):
            raise HTTPException(status_code=504, detail=str(exc)) from exc
        if isinstance(exc, HTTPException) and exc.status_code == 504:
            raise
        # Non-timeout inference/gateway errors fail the request without leaking
        # internal details.
        agent_metrics.record_inference_error(org_id=req.org_id)
        _exc_detail = str(getattr(exc, "detail", "")) or str(exc)
        structured_logging.emit_inference_error(
            org_id=req.org_id,
            conversation_id=req.session_id,
            error_type=type(exc).__name__,
            error_detail=_exc_detail,
            latency_ms=round((time.perf_counter() - chat_start) * 1000, 1),
        )
        # A failed inference is an HTTP error with no assistant reply or
        # session identity. Detail stays generic: provider/
        # billing/model error strings never reach end users (they are in the
        # ops log above).
        raise HTTPException(
            status_code=502,
            detail="Inference failed",
        ) from exc
    finally:
        # #4285: reconcile the up-front budget reservation with this turn's real
        # cost — refunding the unspent slice or debiting any overage. Runs on
        # EVERY exit path (success or raised timeout / file / inference error),
        # so an in-flight reservation is never leaked into the counter.
        # cost_usd is None on any failure path → full refund (no cost incurred),
        # matching the pre-#4285 "debit only on success" behavior.
        if cost_reservation is not None:
            await cost_reservation.settle(cost_usd)

    # Native Hermes and AgentFS remain the audit authority. This boundary emits
    # operational counters and direct measured completion scalars only; it
    # creates no response-side audit object.
    chat_latency_seconds = time.perf_counter() - chat_start
    latency_ms = round(chat_latency_seconds * 1000.0, 3)
    # US-015: emit Prometheus metrics on the success path. Failures (HTTP
    # 4xx/5xx) raise above and are not counted; native Hermes telemetry owns
    # the runtime error record.
    agent_metrics.record_chat_request(
        org_id=req.org_id,
        role=role,
    )
    agent_metrics.observe_chat_latency(chat_latency_seconds)
    agent_metrics.add_hermes_cost(cost_usd, org_id=req.org_id)

    # US-017 / #4285: the per-Org daily cost counter is reconciled in the
    # finally above via cost_reservation.settle(cost_usd) — it pairs with the
    # reserve() gate to enforce the cap atomically across concurrent calls.

    # US-016: one structured JSON log line per /chat for CloudWatch /
    # SigNoz ingestion. Emitted only on the success path; failure paths
    # raise HTTPException above and surface via uvicorn access logs.
    structured_logging.emit_chat_completion(
        org_id=req.org_id,
        conversation_id=response_session_id,
        customer_id=None,
        agent_role=role,
        tokens_in=token_input,
        tokens_out=token_output,
        cost_usd=cost_usd,
        latency_ms=latency_ms,
    )

    return ChatResponse(
        reply=agentReply,
        session_id=response_session_id,
    )


async def create_session(req: SessionNewRequest) -> SessionNewResponse:
    """Mint a new Hermes ACP session for a specialist.

    Unlike `chat`, this carries no model inference: no cost reservation, no
    per-turn token accounting. It is the resident machine's session/new
    admission (op:"session_new"), wrapped with the same OrgBootstrap
    creds pattern as /v1/chat.
    """
    try:
        session_id = await hermes_client.session_new(
            bootstrap=OrgBootstrap.from_request(
                req.org_id,
                agentfs_url=req.agentfs_url,
                agentfs_token=req.agentfs_token,
                sessiondb_sync_url=req.sessiondb_sync_url,
                sessiondb_sync_token=req.sessiondb_sync_token,
            ),
            specialist_id=req.specialist_id,
        )
    except HermesACPTimeoutError as exc:
        raise HTTPException(status_code=504, detail=str(exc)) from exc
    except Exception as exc:
        logger.error(
            "Session mint failure for specialist_id=%s org=%s: %s",
            req.specialist_id,
            req.org_id,
            exc,
            exc_info=True,
        )
        # Propagate the REAL failure text: the api's mint recovery path
        # discriminates on it (e.g. "AgentFS invariant is missing" -> lazy
        # home establishment + one retry). A generic "Session creation
        # failed" makes every cold org an opaque 502.
        raise HTTPException(status_code=502, detail=f"Session creation failed: {exc}") from exc
    return SessionNewResponse(session_id=session_id)


# ── Canonical v1 API surface ─────────────────────────────────────────────────
#
# Chat is exposed once through /v1/chat (+ /v1/sessions for session/new).
# Portal SSE is a separate API transport over the canonical session
# pipeline, not a second Hermes execution path. The per-session WebSocket
# channel (/v1/session/ws) is the same pipeline again over the persistent
# api<->agent connection — session_new + turns on one socket, 15-min idle
# TTL (see routers/session_ws.py).
from routers import health as _v1_health
from routers import chat as _v1_chat
from routers import corrections as _v1_corrections
from routers import session_ws as _v1_session_ws

app.include_router(_v1_health.router, prefix="/v1")
app.include_router(_v1_chat.router, prefix="/v1")
app.include_router(_v1_corrections.router, prefix="/v1")
app.include_router(_v1_session_ws.router, prefix="/v1")