643 lines
25 KiB
Python
643 lines
25 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import hmac
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
import json
|
|
import logging
|
|
import os
|
|
from pathlib import Path
|
|
import socket
|
|
import subprocess
|
|
import threading
|
|
import time
|
|
from typing import Any
|
|
from urllib.parse import urlparse
|
|
|
|
from .clients import ResponsesClient, WorkspaceAgentClient
|
|
from .config import Config
|
|
from .dispatch import DispatchSink
|
|
from .errors import (
|
|
AuthenticationError,
|
|
ConflictError,
|
|
DispatchError,
|
|
OrchestratorError,
|
|
TriggerError,
|
|
ValidationError,
|
|
)
|
|
from .models import Decision, Event
|
|
from .securefs import ensure_secure_directory, ensure_shared_directory, read_secure_json
|
|
from .store import Store
|
|
from .sync import OpenClawSynchronizer, lifecycle_record
|
|
|
|
|
|
LOG = logging.getLogger("aios-orchestrator-gateway")
|
|
|
|
|
|
class Runtime:
|
|
def __init__(self, config: Config):
|
|
self.config = config
|
|
self.store = Store(config.database_path)
|
|
self.dispatch = DispatchSink(
|
|
config.dispatch_outbox_dir,
|
|
config.command_queue_dir,
|
|
config.contract_commit,
|
|
)
|
|
self.workspace = None
|
|
workspace_token = config.workspace_token()
|
|
if config.workspace_trigger_id and workspace_token:
|
|
self.workspace = WorkspaceAgentClient(
|
|
config.workspace_trigger_id,
|
|
workspace_token,
|
|
)
|
|
self.responses = None
|
|
responses_key = config.responses_api_key()
|
|
if responses_key and config.responses_model:
|
|
self.responses = ResponsesClient(
|
|
responses_key,
|
|
config.responses_model,
|
|
)
|
|
self.synchronizer = OpenClawSynchronizer(
|
|
config.openclaw_sync_dir,
|
|
endpoint=config.openclaw_sync_endpoint,
|
|
bearer_token=config.openclaw_sync_token(),
|
|
)
|
|
self.worker_id = f"{socket.gethostname()}-{os.getpid()}-{threading.get_native_id()}"
|
|
self.stop_event = threading.Event()
|
|
self.worker = threading.Thread(target=self._loop, daemon=True)
|
|
self._next_git_poll = 0.0
|
|
for directory in (
|
|
config.event_inbox_dir,
|
|
config.decision_inbox_dir,
|
|
config.dispatch_outbox_dir,
|
|
config.openclaw_sync_dir,
|
|
):
|
|
ensure_secure_directory(directory)
|
|
ensure_shared_directory(config.command_queue_dir)
|
|
|
|
def start(self) -> None:
|
|
self.worker.start()
|
|
|
|
def stop(self) -> None:
|
|
self.stop_event.set()
|
|
if self.worker.is_alive():
|
|
self.worker.join(timeout=15)
|
|
self.store.close()
|
|
|
|
def submit_event(self, payload: dict[str, Any]) -> tuple[bool, dict[str, Any]]:
|
|
event = Event.from_dict(payload)
|
|
if event.contract_commit != self.config.contract_commit:
|
|
raise ConflictError("contract_version_mismatch")
|
|
created, stored = self.store.register_event(event)
|
|
if created:
|
|
self._enqueue_event_lifecycle(event)
|
|
return created, stored
|
|
|
|
def submit_decision(self, payload: dict[str, Any]) -> int:
|
|
decision = Decision.from_dict(payload)
|
|
if decision.contract_commit != self.config.contract_commit:
|
|
raise ConflictError("contract_version_mismatch")
|
|
version = self.store.apply_decision(
|
|
decision,
|
|
max_hops_per_gate=self.config.max_hops_per_gate,
|
|
max_same_blocker_retries=self.config.max_same_blocker_retries,
|
|
)
|
|
self._enqueue_decision_lifecycle(decision, version)
|
|
return version
|
|
|
|
def _loop(self) -> None:
|
|
next_summary = 0.0
|
|
while not self.stop_event.wait(self.config.poll_interval_seconds):
|
|
try:
|
|
self.poll_once()
|
|
if time.monotonic() >= next_summary:
|
|
self.enqueue_periodic_summaries()
|
|
next_summary = time.monotonic() + self.config.summary_interval_seconds
|
|
except Exception:
|
|
LOG.exception("poll_cycle_failed")
|
|
|
|
def poll_once(self) -> None:
|
|
"""One deterministic reliability cycle.
|
|
|
|
This is deliberately callable from a systemd timer or health supervisor
|
|
in addition to the internal loop.
|
|
"""
|
|
self.ingest_file_inboxes()
|
|
self.poll_git_refs()
|
|
self.process_new_events()
|
|
self.process_workspace_timeouts()
|
|
self.poll_background_responses()
|
|
self.dispatch_work_units()
|
|
self.synchronize_openclaw()
|
|
|
|
def ingest_file_inboxes(self) -> None:
|
|
self._ingest_directory(self.config.event_inbox_dir, kind="event")
|
|
self._ingest_directory(self.config.decision_inbox_dir, kind="decision")
|
|
|
|
def _ingest_directory(self, directory: Path, *, kind: str) -> None:
|
|
pending = directory / "pending"
|
|
processed = directory / "processed"
|
|
rejected = directory / "rejected"
|
|
ensure_shared_directory(pending, group_writable=True)
|
|
ensure_secure_directory(processed)
|
|
ensure_secure_directory(rejected)
|
|
for path in sorted(pending.glob("*.json"))[:128]:
|
|
try:
|
|
payload = read_secure_json(path, expected_uid=None)
|
|
if kind == "event":
|
|
self.submit_event(payload)
|
|
else:
|
|
self.submit_decision(payload)
|
|
target = processed / path.name
|
|
except Exception as exc:
|
|
LOG.warning("file_inbox_rejected path=%s error=%s", path, type(exc).__name__)
|
|
target = rejected / path.name
|
|
if target.exists():
|
|
# Same file name is an idempotency replay. Keep original audit copy.
|
|
path.unlink()
|
|
else:
|
|
path.replace(target)
|
|
|
|
def poll_git_refs(self) -> None:
|
|
"""Poll authoritative local bare refs so missed webhooks cannot stall the loop."""
|
|
if not self.config.git_poll_enabled or self.config.git_poll_bare_repo is None:
|
|
return
|
|
now = time.monotonic()
|
|
if now < self._next_git_poll:
|
|
return
|
|
self._next_git_poll = now + self.config.git_poll_interval_seconds
|
|
command = [
|
|
"git",
|
|
f"--git-dir={self.config.git_poll_bare_repo}",
|
|
"for-each-ref",
|
|
"--format=%(refname) %(objectname)",
|
|
*self.config.git_poll_ref_prefixes,
|
|
]
|
|
try:
|
|
result = subprocess.run(
|
|
command,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
text=True,
|
|
check=False,
|
|
timeout=30,
|
|
env={"PATH": "/usr/bin:/bin", "LANG": "C", "LC_ALL": "C"},
|
|
)
|
|
except (OSError, subprocess.SubprocessError):
|
|
LOG.exception("git_ref_poll_failed")
|
|
return
|
|
if result.returncode != 0:
|
|
LOG.warning("git_ref_poll_rc=%s", result.returncode)
|
|
return
|
|
refs: dict[str, str] = {}
|
|
for line in result.stdout.splitlines():
|
|
refname, sep, oid = line.partition(" ")
|
|
if not sep or not refname.startswith("refs/") or len(oid) != 40:
|
|
continue
|
|
if any(ch not in "0123456789abcdef" for ch in oid):
|
|
continue
|
|
refs[refname] = oid
|
|
initialized, changes = self.store.observe_refs(refs)
|
|
if not initialized or not changes:
|
|
return
|
|
payload_changes = [
|
|
{"ref": refname, "old_oid": old_oid, "new_oid": new_oid}
|
|
for refname, old_oid, new_oid in sorted(changes)
|
|
]
|
|
canonical = json.dumps(payload_changes, separators=(",", ":"), sort_keys=True)
|
|
event_id = "gitpoll-" + hashlib.sha256(canonical.encode()).hexdigest()[:40]
|
|
state = self.store.project_state(self.config.project_key)
|
|
state_version = 0 if state is None else int(state["state_version"])
|
|
event = Event.from_dict(
|
|
{
|
|
"schema_version": 1,
|
|
"project_key": self.config.project_key,
|
|
"contract_commit": self.config.contract_commit,
|
|
"event_id": event_id,
|
|
"source_agent": "gitea",
|
|
"event_type": "git_refs_changed",
|
|
"work_unit_id": "GIT-REF-POLL",
|
|
"status": "READY",
|
|
"state_version": state_version,
|
|
"implementation_commit": None,
|
|
"result_commit": None,
|
|
"blocker": None,
|
|
"evidence_refs": [],
|
|
"payload": {"changes": payload_changes},
|
|
}
|
|
)
|
|
try:
|
|
self.submit_event(event.to_dict())
|
|
except ConflictError:
|
|
LOG.info("git_ref_event_already_registered event_id=%s", event_id)
|
|
|
|
def process_new_events(self) -> None:
|
|
rows = self.store.claim_events(
|
|
owner=self.worker_id,
|
|
statuses=("RECEIVED", "TRIGGER_FAILED"),
|
|
limit=32,
|
|
lease_seconds=self.config.lease_seconds,
|
|
)
|
|
for row in rows:
|
|
event = Event.from_dict(json.loads(row["event_json"]))
|
|
attempts = int(row["attempts"])
|
|
try:
|
|
if self.workspace is not None:
|
|
self.workspace.trigger(event)
|
|
self.store.update_event_trigger(
|
|
event.event_id,
|
|
status="AWAITING_DECISION",
|
|
channel="workspace_agent",
|
|
remote_id=None,
|
|
)
|
|
elif self.responses is not None:
|
|
response_id = self.responses.start(event)
|
|
self.store.update_event_trigger(
|
|
event.event_id,
|
|
status="POLLING_RESPONSE",
|
|
channel="responses_api",
|
|
remote_id=response_id,
|
|
retry_after=self.config.poll_interval_seconds,
|
|
)
|
|
else:
|
|
raise TriggerError("no_model_channel")
|
|
except TriggerError as exc:
|
|
self._retry_or_dead_letter(event, attempts, str(exc))
|
|
|
|
def process_workspace_timeouts(self) -> None:
|
|
cutoff = int(time.time()) - self.config.callback_timeout_seconds
|
|
for row in self.store.claim_workspace_timeouts(
|
|
owner=self.worker_id,
|
|
older_than=cutoff,
|
|
limit=32,
|
|
lease_seconds=self.config.lease_seconds,
|
|
):
|
|
event = Event.from_dict(json.loads(row["event_json"]))
|
|
if self.responses is None:
|
|
self._retry_or_dead_letter(
|
|
event,
|
|
int(row["attempts"]),
|
|
"workspace_callback_timeout",
|
|
)
|
|
continue
|
|
try:
|
|
response_id = self.responses.start(event)
|
|
self.store.update_event_trigger(
|
|
event.event_id,
|
|
status="POLLING_RESPONSE",
|
|
channel="responses_api",
|
|
remote_id=response_id,
|
|
retry_after=self.config.poll_interval_seconds,
|
|
increment_attempt=True,
|
|
)
|
|
except TriggerError as exc:
|
|
self._retry_or_dead_letter(event, int(row["attempts"]), str(exc))
|
|
|
|
def poll_background_responses(self) -> None:
|
|
if self.responses is None:
|
|
return
|
|
for row in self.store.claim_response_polls(
|
|
owner=self.worker_id,
|
|
limit=32,
|
|
lease_seconds=self.config.lease_seconds,
|
|
):
|
|
event = Event.from_dict(json.loads(row["event_json"]))
|
|
response_id = row.get("remote_id")
|
|
if not isinstance(response_id, str):
|
|
self._retry_or_dead_letter(event, int(row["attempts"]), "missing_response_id")
|
|
continue
|
|
try:
|
|
status, decision = self.responses.poll(response_id)
|
|
if status in {"queued", "in_progress"}:
|
|
attempts = int(row["attempts"])
|
|
if attempts + 1 > self.config.max_retries:
|
|
self._retry_or_dead_letter(
|
|
event, attempts, f"response_{status}_retry_limit"
|
|
)
|
|
continue
|
|
self.store.update_event_trigger(
|
|
event.event_id,
|
|
status="POLLING_RESPONSE",
|
|
channel="responses_api",
|
|
remote_id=response_id,
|
|
retry_after=self.config.poll_interval_seconds,
|
|
increment_attempt=True,
|
|
)
|
|
continue
|
|
if decision is None:
|
|
raise TriggerError("completed_without_decision")
|
|
self.submit_decision(decision.to_dict())
|
|
except OrchestratorError as exc:
|
|
self._retry_or_dead_letter(event, int(row["attempts"]), str(exc))
|
|
|
|
def dispatch_work_units(self) -> None:
|
|
rows = self.store.claim_outbox(
|
|
owner=self.worker_id,
|
|
limit=64,
|
|
lease_seconds=self.config.lease_seconds,
|
|
)
|
|
for row in rows:
|
|
try:
|
|
outbox_path, queue_path = self.dispatch.dispatch(row)
|
|
self.store.mark_outbox(row["work_unit_id"], status="DISPATCHED")
|
|
payload = {
|
|
"assigned_agent": row["assigned_agent"],
|
|
"event_id": row["event_id"],
|
|
"dispatch_outbox": str(outbox_path),
|
|
"command_queue": str(queue_path),
|
|
}
|
|
record = lifecycle_record(
|
|
contract_commit=self.config.contract_commit,
|
|
project_key=self._project_for_event(row["event_id"]),
|
|
kind="TASK_DISPATCHED",
|
|
work_unit_id=row["work_unit_id"],
|
|
payload=payload,
|
|
)
|
|
self.store.enqueue_lifecycle(
|
|
sync_id=record["sync_id"],
|
|
project_key=record["project_key"],
|
|
work_unit_id=row["work_unit_id"],
|
|
kind=record["kind"],
|
|
payload=record,
|
|
)
|
|
except Exception:
|
|
attempts = int(row["attempts"])
|
|
status = "DEAD_LETTER" if attempts >= self.config.max_retries else "RETRY"
|
|
self.store.mark_outbox(
|
|
row["work_unit_id"],
|
|
status=status,
|
|
retry_after=self._backoff(attempts),
|
|
increment_attempt=True,
|
|
)
|
|
LOG.exception("dispatch_failed work_unit=%s", row["work_unit_id"])
|
|
|
|
def synchronize_openclaw(self) -> None:
|
|
rows = self.store.claim_lifecycle(
|
|
owner=self.worker_id,
|
|
limit=64,
|
|
lease_seconds=self.config.lease_seconds,
|
|
)
|
|
for row in rows:
|
|
try:
|
|
payload = json.loads(row["payload_json"])
|
|
self.synchronizer.emit(payload)
|
|
self.store.mark_lifecycle(row["sync_id"], status="SENT")
|
|
except Exception:
|
|
attempts = int(row["attempts"])
|
|
status = "DEAD_LETTER" if attempts >= self.config.max_retries else "RETRY"
|
|
self.store.mark_lifecycle(
|
|
row["sync_id"],
|
|
status=status,
|
|
retry_after=self._backoff(attempts),
|
|
increment_attempt=True,
|
|
)
|
|
LOG.exception("openclaw_sync_failed sync_id=%s", row["sync_id"])
|
|
|
|
def enqueue_periodic_summaries(self) -> None:
|
|
for state in self.store.all_project_states():
|
|
project_key = state["project_key"]
|
|
if not self.store.summary_due(project_key, self.config.summary_interval_seconds):
|
|
continue
|
|
summary = self.store.project_summary(project_key)
|
|
if summary is None:
|
|
continue
|
|
digest = hashlib.sha256(
|
|
json.dumps(summary, separators=(",", ":"), sort_keys=True).encode()
|
|
).hexdigest()
|
|
record = lifecycle_record(
|
|
contract_commit=self.config.contract_commit,
|
|
project_key=project_key,
|
|
kind="PROJECT_SUMMARY",
|
|
work_unit_id=None,
|
|
payload=summary,
|
|
)
|
|
self.store.enqueue_lifecycle(
|
|
sync_id=record["sync_id"],
|
|
project_key=project_key,
|
|
work_unit_id=None,
|
|
kind=record["kind"],
|
|
payload=record,
|
|
)
|
|
self.store.mark_summary(project_key, digest)
|
|
|
|
def _enqueue_event_lifecycle(self, event: Event) -> None:
|
|
if event.event_type == "work_unit_started":
|
|
kind = "TASK_STARTED"
|
|
elif event.event_type in {"work_unit_completed", "work_unit_failed", "work_unit_blocked"}:
|
|
kind = "TASK_ENDED"
|
|
else:
|
|
kind = "EVENT_RECEIVED"
|
|
record = lifecycle_record(
|
|
contract_commit=self.config.contract_commit,
|
|
project_key=event.project_key,
|
|
kind=kind,
|
|
work_unit_id=event.work_unit_id,
|
|
payload=event.to_dict(),
|
|
)
|
|
self.store.enqueue_lifecycle(
|
|
sync_id=record["sync_id"],
|
|
project_key=event.project_key,
|
|
work_unit_id=event.work_unit_id,
|
|
kind=kind,
|
|
payload=record,
|
|
)
|
|
|
|
def _enqueue_decision_lifecycle(self, decision: Decision, version: int) -> None:
|
|
kind = "PROJECT_COMPLETED" if decision.verdict == "COMPLETE" else "DECISION_APPLIED"
|
|
record = lifecycle_record(
|
|
contract_commit=self.config.contract_commit,
|
|
project_key=decision.project_key,
|
|
kind=kind,
|
|
work_unit_id=None,
|
|
payload={"decision": decision.to_dict(), "state_version": version},
|
|
)
|
|
self.store.enqueue_lifecycle(
|
|
sync_id=record["sync_id"],
|
|
project_key=decision.project_key,
|
|
work_unit_id=None,
|
|
kind=kind,
|
|
payload=record,
|
|
)
|
|
summary = self.store.project_summary(decision.project_key)
|
|
if summary:
|
|
summary_record = lifecycle_record(
|
|
contract_commit=self.config.contract_commit,
|
|
project_key=decision.project_key,
|
|
kind="PROJECT_SUMMARY",
|
|
work_unit_id=None,
|
|
payload=summary,
|
|
)
|
|
self.store.enqueue_lifecycle(
|
|
sync_id=summary_record["sync_id"],
|
|
project_key=decision.project_key,
|
|
work_unit_id=None,
|
|
kind=summary_record["kind"],
|
|
payload=summary_record,
|
|
)
|
|
|
|
def _project_for_event(self, event_id: str) -> str:
|
|
rows = self.store._conn().execute(
|
|
"SELECT project_key FROM events WHERE event_id=?",
|
|
(event_id,),
|
|
).fetchone()
|
|
if rows is None:
|
|
raise ConflictError("event_not_found")
|
|
return str(rows["project_key"])
|
|
|
|
def _retry_or_dead_letter(self, event: Event, attempts: int, reason: str) -> None:
|
|
next_attempt = attempts + 1
|
|
if next_attempt > self.config.max_retries:
|
|
self.store.update_event_trigger(
|
|
event.event_id,
|
|
status="DEAD_LETTER",
|
|
channel=None,
|
|
remote_id=None,
|
|
increment_attempt=True,
|
|
)
|
|
record = lifecycle_record(
|
|
contract_commit=self.config.contract_commit,
|
|
project_key=event.project_key,
|
|
kind="TASK_ENDED",
|
|
work_unit_id=event.work_unit_id,
|
|
payload={
|
|
"status": "FAIL",
|
|
"reason": reason,
|
|
"event_id": event.event_id,
|
|
"attempts": next_attempt,
|
|
},
|
|
)
|
|
self.store.enqueue_lifecycle(
|
|
sync_id=record["sync_id"],
|
|
project_key=event.project_key,
|
|
work_unit_id=event.work_unit_id,
|
|
kind=record["kind"],
|
|
payload=record,
|
|
)
|
|
return
|
|
self.store.update_event_trigger(
|
|
event.event_id,
|
|
status="TRIGGER_FAILED",
|
|
channel=None,
|
|
remote_id=None,
|
|
retry_after=self._backoff(attempts),
|
|
increment_attempt=True,
|
|
)
|
|
|
|
def _backoff(self, attempts: int) -> int:
|
|
return min(3600, self.config.retry_base_seconds * (2 ** min(attempts, 10)))
|
|
|
|
|
|
class Handler(BaseHTTPRequestHandler):
|
|
runtime: Runtime
|
|
|
|
def log_message(self, fmt: str, *args: object) -> None:
|
|
LOG.info("%s %s", self.client_address[0], fmt % args)
|
|
|
|
def _reply(self, status: int, payload: dict[str, Any]) -> None:
|
|
body = json.dumps(payload, separators=(",", ":"), sort_keys=True).encode()
|
|
self.send_response(status)
|
|
self.send_header("Content-Type", "application/json")
|
|
self.send_header("Content-Length", str(len(body)))
|
|
self.send_header("Cache-Control", "no-store")
|
|
self.end_headers()
|
|
self.wfile.write(body)
|
|
|
|
def _body(self) -> dict[str, Any]:
|
|
try:
|
|
length = int(self.headers.get("Content-Length", "0"))
|
|
except ValueError as exc:
|
|
raise ValidationError("invalid_content_length") from exc
|
|
if length <= 0 or length > 1_048_576:
|
|
raise ValidationError("invalid_body_size")
|
|
try:
|
|
value = json.loads(self.rfile.read(length))
|
|
except json.JSONDecodeError as exc:
|
|
raise ValidationError("invalid_json") from exc
|
|
if not isinstance(value, dict):
|
|
raise ValidationError("json_root_not_object")
|
|
return value
|
|
|
|
def _authenticate(self, expected: str) -> None:
|
|
header = self.headers.get("Authorization", "")
|
|
if not header.startswith("Bearer "):
|
|
raise AuthenticationError("missing_bearer")
|
|
if not hmac.compare_digest(header[7:], expected):
|
|
raise AuthenticationError("invalid_bearer")
|
|
|
|
def do_GET(self) -> None:
|
|
path = urlparse(self.path).path.rstrip("/") or "/"
|
|
try:
|
|
if path == "/healthz":
|
|
self._reply(
|
|
200,
|
|
{
|
|
"service": "aios-orchestrator-gateway",
|
|
"status": "active",
|
|
"contract_commit": self.runtime.config.contract_commit,
|
|
"polling": True,
|
|
},
|
|
)
|
|
return
|
|
if path.startswith("/v1/orchestrator/state/"):
|
|
self._authenticate(self.runtime.config.inbound_token())
|
|
key = path.rsplit("/", 1)[-1]
|
|
state = self.runtime.store.project_summary(key)
|
|
self._reply(200 if state else 404, state or {"error": "project_not_found"})
|
|
return
|
|
self._reply(404, {"error": "not_found"})
|
|
except AuthenticationError as exc:
|
|
self._reply(401, {"error": str(exc)})
|
|
except OrchestratorError as exc:
|
|
self._reply(400, {"error": str(exc)})
|
|
|
|
def do_POST(self) -> None:
|
|
path = urlparse(self.path).path.rstrip("/")
|
|
try:
|
|
if path == "/v1/orchestrator/events":
|
|
self._authenticate(self.runtime.config.inbound_token())
|
|
created, event = self.runtime.submit_event(self._body())
|
|
self._reply(202 if created else 200, {"created": created, "event": event})
|
|
return
|
|
if path == "/v1/orchestrator/decisions":
|
|
self._authenticate(self.runtime.config.callback_token())
|
|
version = self.runtime.submit_decision(self._body())
|
|
self._reply(202, {"accepted": True, "state_version": version})
|
|
return
|
|
if path == "/v1/orchestrator/poll":
|
|
self._authenticate(self.runtime.config.inbound_token())
|
|
self.runtime.poll_once()
|
|
self._reply(200, {"polled": True})
|
|
return
|
|
self._reply(404, {"error": "not_found"})
|
|
except AuthenticationError as exc:
|
|
self._reply(401, {"error": str(exc)})
|
|
except ConflictError as exc:
|
|
self._reply(409, {"error": str(exc)})
|
|
except ValidationError as exc:
|
|
self._reply(422, {"error": str(exc)})
|
|
except OrchestratorError as exc:
|
|
self._reply(400, {"error": str(exc)})
|
|
except Exception:
|
|
LOG.exception("request_failed path=%s", path)
|
|
self._reply(500, {"error": "internal_error"})
|
|
|
|
|
|
def serve(config_path: Path) -> None:
|
|
config = Config.from_file(config_path)
|
|
runtime = Runtime(config)
|
|
Handler.runtime = runtime
|
|
server = ThreadingHTTPServer((config.host, config.port), Handler)
|
|
runtime.start()
|
|
LOG.info(
|
|
"gateway_started host=%s port=%s contract=%s polling=%ss",
|
|
config.host,
|
|
config.port,
|
|
config.contract_commit,
|
|
config.poll_interval_seconds,
|
|
)
|
|
try:
|
|
server.serve_forever(poll_interval=0.5)
|
|
finally:
|
|
server.server_close()
|
|
runtime.stop()
|