open/portfolio-optimiser
1
0
Fork 0
Code Issues Releases 3 Activity

feat(4d): hostet inngang — main.py wrapper rundt run_project på én asyncio-løkke

Browse source 
To målinger avgjorde formen FØR koden: (1) hosting-pakkas InvocationsHostServer
finnes kun i bygg som krever agent-framework-core>=1.13.0 (treet låser 1.9.0;
eneste 1.9-kompatible bygg er en forlatt alfa som importerer mcp udeklarert),
(2) et gjenbrukt bygget workflow er single-use på 1.9.0 (kall-serie [2,0,0] —
rundetaket persisterer; ferskt objekt per kall er ren kontroll). Derfor spikens
§5-fallback: hosting.py serverer kontrakten (8088/PORT, /readiness,
/invocations, SIGTERM→0) selv, stdlib asyncio på ÉN løkke — aldri as_agent()
(gatene ligger utenfor grafen), aldri tråder (NG1-guarden fanget første utkast
med ThreadingHTTPServer; asyncio-formen består den by construction).

Payload whitelistes på run_projects signatur — ukjente felt nektes ved navn
(400), aldri stille droppet; profile defaulter til azure kun her. ValueError →
400, alt annet → 500 {error_type, error}; Rejection er vellykket kjøring → 200.
outbox.outcome_payload ekstrahert som den ENE kopien av validated/rejected-
forgreningen (kø-(p)-regelen). azure.yaml validert GRØNN mot begge autoritative
skjemaer (jsonschema, hentet ferskt); ingen env:, ingen startupCommand (imagets
CMD er den ene kopien). Dockerfile: 3.12-slim-bookworm + git + uv==0.9.8 +
uv sync --frozen --no-dev; git archive <indeks-tre> | docker build
--platform linux/amd64 grønn på nøyaktig de stagede bytene.

Iron Law fulgt: testfila rød ved collection FØR modulen fantes. 835 passed /
4 skipped (fra 821), ruff+format+mypy rene. Seks mutasjoner mot HELE suiten,
alle røde på riktig test: detach felt-mappingen · dropp ukjente felt stille ·
flipp 400/500 · detach azure-defaulten · detach SIGTERM-handleren · detach
main.py-shimen (de to siste kun fanget av subprosess-testen, P4-presedensen).
Deploy IKKE utført — azd-steget er operatørens.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01PEiiSGRShizKc771ZBa1iq
This commit is contained in:
Kjell Tore Guttormsen 2026-08-13 23:31:02 +02:00
commit 426ccb0ad6
8 changed files with 766 additions and 14 deletions
Download patch file Download diff file Expand all files Collapse all files

221
src/portfolio_optimiser/hosting.py Normal file
View file

@ -0,0 +1,221 @@
"""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. 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
from typing import Any
from portfolio_optimiser.outbox import outcome_payload
from portfolio_optimiser.run import RunResult, run_project
DEFAULT_PORT = 8088
_HOSTED_DEFAULT_PROFILE = "azure"
_REQUIRED_FIELDS = ("project_id", "docs_dir", "verdict_input")
_OPTIONAL_FIELDS = ("bundle_dir", "profile", "max_rounds", "max_tokens", "top_k")
_ALLOWED_FIELDS = frozenset(_REQUIRED_FIELDS + _OPTIONAL_FIELDS)
_REASONS = {200: "OK", 400: "Bad Request", 404: "Not Found", 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]]:
"""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."""
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)}")
kwargs: dict[str, Any] = {k: payload[k] for k in payload if k != "project_id"}
kwargs.setdefault("profile", _HOSTED_DEFAULT_PROFILE)
return payload["project_id"], kwargs
def _response_payload(result: RunResult) -> dict[str, Any]:
return {
**outcome_payload(
result.outcome,
checker_verdict=result.checker_verdict,
verdict_id=result.verdict.id,
),
"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."""
project_id, kwargs = _run_kwargs(payload)
result = await run_project(project_id, **kwargs)
assert isinstance(result, RunResult)
return _response_payload(result)
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 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."""
asyncio.run(_serve_until_sigterm())

46
src/portfolio_optimiser/outbox.py
View file

@ -88,9 +88,33 @@ def write_outbox(
encoding="utf-8",
)
outcome_path = directory / f"{stem}-outcome.json"
outcome_path.write_text(
_dump(
{
**keys,
**outcome_payload(outcome, checker_verdict=checker_verdict, verdict_id=verdict_id),
}
),
encoding="utf-8",
)
return proposal_path, outcome_path
def outcome_payload(
outcome: ValidatedProposal | Rejection,
*,
checker_verdict: str | None,
verdict_id: str,
) -> dict[str, Any]:
"""The outcome artefact's payload minus the file keys — the ONE copy of the
validated/rejected branching, shared by ``write_outbox`` and the hosted invocations
response (``hosting._response_payload``). Two copies of the branch would drift, and a
drifted copy would let the HTTP surface describe an outcome the outbox never wrote —
the ``to_ore`` single-source rule (kø-(p)) applied to a payload shape."""
if isinstance(outcome, ValidatedProposal):
outcome_payload: dict[str, Any] = {
**keys,
return {
"outcome_type": "validated",
"p10": outcome.p10,
"p50": outcome.p50,
@ -99,18 +123,12 @@ def write_outbox(
"checker_verdict": checker_verdict,
"verdict_id": verdict_id,
}
else:
outcome_payload = {
**keys,
"outcome_type": "rejected",
"reason": outcome.reason,
"checker_verdict": checker_verdict,
"verdict_id": verdict_id,
}
outcome_path = directory / f"{stem}-outcome.json"
outcome_path.write_text(_dump(outcome_payload), encoding="utf-8")
return proposal_path, outcome_path
return {
"outcome_type": "rejected",
"reason": outcome.reason,
"checker_verdict": checker_verdict,
"verdict_id": verdict_id,
}
def write_run_config(