"""Hosted entrypoint (Fase 4d): the Foundry hosted-agent runtime contract implemented
DIRECTLY around ``run_project`` — a wrapper, never ``Workflow.as_agent()``.

The form was decided by TWO measurements (13.08), not preference:

* ``agent-framework-foundry-hosting``'s ``InvocationsHostServer`` exists only in builds
  requiring ``agent-framework-core>=1.13.0`` (this tree locks 1.9.0); the sole
  1.9-compatible build (``1.0.0a260618``) ships broken metadata — it imports ``mcp``
  without declaring it — and is superseded. With no usable protocol library, the runtime
  contract (port 8088/``PORT``, ``GET /readiness``, ``POST /invocations``, SIGTERM
  shutdown — spike §1.1) is served HERE, including ``/readiness``, which a protocol
  library would otherwise have provided.
* A BUILT workflow is single-use on core 1.9.0 (measured: client-call series [2, 0, 0]
  across three ``.run()`` calls on ONE object — the round cap persists in the object, so
  reuse yields EMPTY runs, not just contaminated ones). A long-lived hosted process must
  therefore never hold a workflow; every invocation goes through ``run_project``, which
  builds a fresh one per call (the B7 factory).

``as_agent()`` alone would also serve UNGATED proposals: the deterministic validator,
baseline anchoring, checker gate, ledger and learning loop all live OUTSIDE the Workflow
graph (spike §5) — wrapping the graph wraps the wrong boundary.

**The server is asyncio on the ONE loop — no threads, by NG1.** ``http.server``'s
threading variant would put concurrent ``run_project`` calls on OS threads, where none of
S3.3's determinism reasoning holds and MAF's thread-safety is undocumented — exactly what
``test_no_thread_or_process_path_exists_under_src`` ratchets against. ``asyncio.start_server``
plus ~40 lines of HTTP/1.1 parsing keeps liveness (``/readiness`` answers while an
invocation awaits model I/O) and stays inside the sanctioned concurrency model: concurrent
invocations interleave as coroutines, the same way ``run_portfolio``'s waves do. Honest
limit: during a CPU-bound stretch (the CBC solve) the loop — and thus readiness — stalls
for that stretch; chunked request bodies are not supported (``Content-Length`` only).

Surface: ``POST /invocations`` takes a JSON object whitelisted onto ``run_project``'s
signature. ``profile`` defaults to ``"azure"`` on THIS surface only (``run_project``'s own
default stays LOCAL): a hosted container has no local OpenAI-compatible endpoint, and the
AZURE profile reads its environment at call time (Fase 4b). Validation, never repair: an
unknown field is a 400 naming the field — the permissive-schema trap (valg-doc §0) applied
to our own surface. Error mapping is honest: ``ValueError`` (pydantic contract violations
subclass it) → 400; any other failure → 500 ``{error_type, error}`` (mirrors
``RunFailure``); a ``Rejection`` is a SUCCESSFUL run → 200 with ``outcome_type:
"rejected"`` — the negative outcome belongs to the payload, never to the transport.

``BudgetExceeded`` gets its OWN arm → 429, for the same reason ``BudgetStop`` is kept out of
``stop_reason`` (S3.4): a cap that fires is the feature working (``Budget`` exists so a run can
never hang unbounded), and answering it on the crash channel makes "it did not work"
unreadable — the first live run died exactly here and the surface said 500, the same thing it
says when the endpoint falls over. It is NOT 200 either: unlike a ``Rejection``, which is a run
that CONCLUDED, an exhausted budget produced no proposal, and a 2xx would let an automated
caller record "analysed" for a run that analysed nothing. 429 because the condition arises from
an ALLOWANCE — ``max_rounds``/``max_tokens`` are whitelisted request fields and raising them is
the caller's own remedy — never from a server fault. The ``kind``/``limit``/``observed`` triple
is carried as STRUCTURE, not flattened into ``str(exc)`` (kø-(y): it describes one ledger and
answering "which cap bound, and by how much" is the operational question), and ``error_type``
is deliberately absent — that key belongs to the failure channel. Honesty limit, stated: no
``Retry-After``. Retrying an unchanged body hits the same cap; the remedy is a larger allowance
or accepting the stop, and a header promising time would be a lie. The
platform's injected headers (``x-agent-user-id``/``x-agent-foundry-call-id``) are absent
locally by contract and unused here; forwarding the call-id on outgoing Foundry calls has
no seam in ``backends.py`` today and is deliberately not built (90 %-prinsippet).
"""

from __future__ import annotations

import asyncio
import json
import os
import signal
import sys
from collections.abc import Mapping
from typing import Any

from portfolio_optimiser.budget import BudgetExceeded
from portfolio_optimiser.explore import ExplorationContract, explore
from portfolio_optimiser.outbox import outcome_payload
from portfolio_optimiser.run import RunResult, run_project
from portfolio_optimiser.tracing import configure_tracing, tracing_notice

DEFAULT_PORT = 8088
_HOSTED_DEFAULT_PROFILE = "azure"
_REQUIRED_FIELDS = ("project_id", "docs_dir")
#: ``verdict_input`` is OPTIONAL since F2 (non-goal 3). It used to be required, which forced an
#: external caller to invent an expert verdict just to get a run at all — a field that could not be
#: filled honestly, on the surface handed over 2026-08-14. The move is a pure WIDENING: a caller
#: that still sends it is unaffected, and one that omits it now gets a run whose verdict is
#: honestly absent.
_OPTIONAL_FIELDS = (
    "bundle_dir",
    "profile",
    "max_rounds",
    "max_tokens",
    "top_k",
    "verdict_input",
)
#: Fields this surface CONSUMES rather than forwards (U4). They are not ``run_project``
#: parameters — the exploration runs first and hands ``run_project`` a ``Mandate`` — so passing one
#: through would be a ``TypeError`` answered as a 500. The whitelist is therefore a THREE-way
#: partition, and the Fase 4e proof gained a negative half to match: every forwarded field must be
#: a real ``run_project`` parameter, and every consumed field must not be.
_CONSUMED_FIELDS = ("explore_prompt", "explore_contract")
_ALLOWED_FIELDS = frozenset(_REQUIRED_FIELDS + _OPTIONAL_FIELDS + _CONSUMED_FIELDS)
_REASONS = {
    200: "OK",
    400: "Bad Request",
    404: "Not Found",
    429: "Too Many Requests",
    500: "Internal Server Error",
}


class InvocationRefused(ValueError):
    """A request the invocations contract refuses — unknown, missing or non-object input.
    Validation, never repair (``write_concept_file`` precedent): nothing is dropped,
    defaulted or corrected on the caller's behalf."""


def resolve_port() -> int:
    """PORT on truthiness, not presence (the 4b rule): an exported-empty PORT is a shell
    accident, not a bind instruction. 8088 is the hosted-agent contract's port."""
    return int(os.environ.get("PORT") or DEFAULT_PORT)


def _run_kwargs(payload: Any) -> tuple[str, dict[str, Any], dict[str, Any]]:
    """Whitelist the JSON payload onto ``run_project``'s signature. Everything not named in
    the whitelist — including server-side seams like ``outbox_dir``, ``client_factory`` or
    ``verdict_dir`` — is refused by name, never silently dropped.

    Returns ``(project_id, forwarded_kwargs, consumed)``. The consumed half is split out HERE
    rather than filtered at the call site so there is one place that decides which fields reach
    ``run_project``: a consumed field left in ``kwargs`` is an argument the signature does not have,
    which the container answers as a 500 for what is really a wiring mistake."""
    if not isinstance(payload, dict):
        raise InvocationRefused("body must be a JSON object")
    unknown = sorted(set(payload) - _ALLOWED_FIELDS)
    if unknown:
        raise InvocationRefused(f"unknown field(s): {', '.join(unknown)}")
    missing = [field for field in _REQUIRED_FIELDS if field not in payload]
    if missing:
        raise InvocationRefused(f"missing required field(s): {', '.join(missing)}")
    consumed = {k: payload[k] for k in _CONSUMED_FIELDS if k in payload}
    kwargs: dict[str, Any] = {
        k: payload[k] for k in payload if k != "project_id" and k not in _CONSUMED_FIELDS
    }
    kwargs.setdefault("profile", _HOSTED_DEFAULT_PROFILE)
    return payload["project_id"], kwargs, consumed


async def _shaped_mandate(consumed: Mapping[str, Any], kwargs: Mapping[str, Any]) -> Any:
    """Run the U4 exploration this invocation asked for and return the mandate it shaped.

    Every refusal here is the CALLER's error and therefore a ``ValueError`` (the 400 arm), by name.
    That placement is deliberate rather than incidental: ``explore()`` refuses two of these itself,
    but ``ExplorationError`` is a ``RuntimeError``, so leaving them to the loop would answer a
    caller's configuration mistake on the crash channel — the same conflation ``BudgetExceeded``
    was given its own 429 to end.

    ``enable_plan_review`` is refused outright. The U13 door is SYNCHRONOUS: it blocks the loop on
    a human or persona, and an HTTP request has neither — the invocation would hang rather than
    answer. The library API is where that door opens."""
    prompt = consumed.get("explore_prompt")
    raw_contract = consumed.get("explore_contract")
    if prompt is None:
        raise InvocationRefused(
            "explore_contract without explore_prompt: the bounds describe an exploration that "
            "would never run"
        )
    if raw_contract is None:
        raise InvocationRefused(
            "explore_prompt without explore_contract: an exploration's bounds are never defaulted "
            "(an omitted cap falls back to an unbounded loop)"
        )
    if not kwargs.get("bundle_dir"):
        raise InvocationRefused(
            "explore_prompt without bundle_dir: the exploration navigates knowledge bases, and "
            "with none configured it would spend its budget reading nothing"
        )
    if not isinstance(raw_contract, dict):
        raise InvocationRefused("explore_contract must be a JSON object")
    contract = ExplorationContract(**raw_contract)  # ValidationError subclasses ValueError -> 400
    if contract.enable_plan_review:
        raise InvocationRefused(
            "explore_contract sets enable_plan_review, but this surface has no reviewer to answer "
            "it: the synchronous plan review would block the request on nobody, and would block "
            "the event loop that answers /readiness while doing it. The operator door is the CLI's "
            "--plan-review (or explore(..., plan_reviewer=...) in-process)"
        )
    result = await explore(
        str(prompt),
        contract=contract,
        bundle_dirs=(kwargs["bundle_dir"],),
        profile=kwargs["profile"],
    )
    return result.mandate


def _response_payload(result: RunResult) -> dict[str, Any]:
    return {
        **outcome_payload(
            result.outcome,
            checker_verdict=result.checker_verdict,
            # The candidate's KEY, not evidence that anybody decided (F2): identical to the
            # captured verdict's id whenever one was given, and still the id under which a later
            # expert verdict on this candidate will arrive when none was.
            verdict_id=result.verdict_key,
        ),
        "proposal": result.outcome.proposal.model_dump(),
        "provenance": result.provenance.model_dump(),
        "refinements": [rejection.reason for rejection in result.refinements],
    }


async def invoke(payload: Any) -> dict[str, Any]:
    """One invocation: validate → ``run_project`` → outbox-shaped response payload.
    ``run_project`` is resolved through this module's namespace at call time (the test
    seam). ``live_dry_run`` is not on the whitelist, so the union narrows to RunResult.

    With ``explore_prompt`` the exploration runs FIRST and its mandate is what the pipeline then
    evaluates — level 2 and 3 of the guarantee table are unchanged, and the exploration itself
    still writes nothing."""
    project_id, kwargs, consumed = _run_kwargs(payload)
    if consumed:
        kwargs["mandate"] = await _shaped_mandate(consumed, kwargs)
    result = await run_project(project_id, **kwargs)
    assert isinstance(result, RunResult)
    return _response_payload(result)


def _budget_payload(exc: BudgetExceeded) -> dict[str, Any]:
    """The exhausted-budget body: the ledger's own triple, plus the human line for the log.

    The ``budget_exhausted`` key's PRESENCE is the discriminator — it is not folded into
    ``outcome_type`` (whose values, ``validated``/``rejected``, mean "the run concluded and
    here is the verdict") for the same reason ``BudgetStop`` was given its own field instead of
    widening ``stop_reason``. Nor could it be: ``outcome_payload`` is the ONE copy of that fork
    and takes a ``ValidatedProposal | Rejection``, neither of which an exhausted run has."""
    return {
        "budget_exhausted": {"kind": exc.kind, "limit": exc.limit, "observed": exc.observed},
        "error": str(exc),
    }


def _http_response(status: int, content_type: str, body: bytes) -> bytes:
    head = (
        f"HTTP/1.1 {status} {_REASONS[status]}\r\n"
        f"Content-Type: {content_type}\r\n"
        f"Content-Length: {len(body)}\r\n"
        "Connection: close\r\n\r\n"
    )
    return head.encode("latin-1") + body


def _json_response(status: int, payload: dict[str, Any]) -> bytes:
    body = (json.dumps(payload, sort_keys=True) + "\n").encode("utf-8")
    return _http_response(status, "application/json", body)


async def _read_request(
    reader: asyncio.StreamReader,
) -> tuple[str, str, bytes] | None:
    """Parse one HTTP/1.1 request: (method, path, body), or None when unparseable. The
    platform terminates TLS and speaks plain HTTP/1.1 (spike §1.1); ``Content-Length`` is
    required for a body — chunked transfer is not supported (an honest MVP limit)."""
    request_line = await reader.readline()
    parts = request_line.decode("latin-1", errors="replace").split()
    if len(parts) != 3:
        return None
    method, path = parts[0], parts[1]
    content_length = 0
    while True:
        line = await reader.readline()
        if line in (b"\r\n", b"\n", b""):
            break
        name, _, value = line.decode("latin-1", errors="replace").partition(":")
        if name.strip().lower() == "content-length":
            try:
                content_length = int(value.strip())
            except ValueError:
                return None
    body = b""
    if content_length > 0:
        try:
            body = await reader.readexactly(content_length)
        except asyncio.IncompleteReadError:
            return None
    return method, path, body


async def _respond(method: str, path: str, body: bytes) -> bytes:
    if method == "GET" and path == "/readiness":
        return _http_response(200, "text/plain", b"ok\n")
    if method == "POST" and path == "/invocations":
        try:
            payload = json.loads(body.decode("utf-8"))
        except (UnicodeDecodeError, json.JSONDecodeError):
            return _json_response(400, {"error": "body is not valid JSON"})
        try:
            return _json_response(200, await invoke(payload))
        except BudgetExceeded as exc:
            # A cap that fired, not a failure — its own channel, and the triple kept as
            # structure rather than re-parsed out of the message by whoever reads this.
            return _json_response(429, _budget_payload(exc))
        except ValueError as exc:
            # The caller's error: InvocationRefused + run_project's fail-fast contract
            # violations (pydantic ValidationError subclasses ValueError).
            return _json_response(400, {"error": str(exc)})
        except Exception as exc:
            # The run's failure, answered rather than dropped — RunFailure's honest shape
            # (error_type + text), so BudgetExceeded reads as what it is, not as a 400.
            return _json_response(500, {"error_type": type(exc).__name__, "error": str(exc)})
    return _json_response(404, {"error": f"no such path: {path}"})


async def _handle(reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None:
    try:
        request = await _read_request(reader)
        if request is None:
            writer.write(_json_response(400, {"error": "malformed HTTP request"}))
        else:
            writer.write(await _respond(*request))
        await writer.drain()
    except (ConnectionError, asyncio.CancelledError):
        pass  # client went away / server shutting down — nothing to answer
    finally:
        writer.close()
        try:
            await writer.wait_closed()
        except ConnectionError:
            pass


async def start_server(host: str, port: int) -> asyncio.AbstractServer:
    """The contract server on the CURRENT loop, bindable to port 0 for tests."""
    return await asyncio.start_server(_handle, host, port)


async def _serve_until_sigterm() -> None:
    stop = asyncio.Event()
    asyncio.get_running_loop().add_signal_handler(signal.SIGTERM, stop.set)
    server = await start_server("0.0.0.0", resolve_port())
    async with server:
        await stop.wait()
    # Leaving the context closes the listening socket; in-flight handlers already hold
    # their connections and finish on the loop before asyncio.run() tears it down.


def main() -> None:
    """Serve the hosted-agent contract until SIGTERM (bind 0.0.0.0 — the platform
    terminates TLS in front of us), then exit 0.

    U14: the tracing seam is installed before the loop starts and announced on stderr, which in a
    container IS the log. This is the entry where "an organisation must be able to see what the run
    did" is actually cashed — the demo is a scripted proof, not the product. A malformed
    ``PORTFOLIO_OTEL`` propagates: a server whose telemetry cannot be configured as asked must not
    start and then look healthy on ``/readiness``."""
    setup = configure_tracing()
    notice = tracing_notice(setup)
    if notice is not None:
        print(notice, file=sys.stderr)
    asyncio.run(_serve_until_sigterm())