"""HTTP daemon that collects Codex conversations and writes daily JSON logs.""" from __future__ import annotations import json import signal import threading from dataclasses import asdict from http import HTTPStatus from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from typing import Any, Dict, Optional from .config import LoggerConfig, load_config from .storage import ConversationStorage, LogEntry class _RequestHandler(BaseHTTPRequestHandler): server_version = "CodexLogger/1.0" protocol_version = "HTTP/1.1" def do_GET(self) -> None: if self.path == self.server.config.health_path: # type: ignore[attr-defined] self._respond_json({"status": "ok"}) elif self.path == self.server.config.metrics_path: # type: ignore[attr-defined] metrics = self.server.metrics_snapshot() # type: ignore[attr-defined] body = "\n".join(f"{key} {value}" for key, value in metrics.items()) self._respond_text(body, content_type="text/plain; charset=utf-8") else: self._respond_text("Not Found", status=HTTPStatus.NOT_FOUND) def do_POST(self) -> None: if self.path != self.server.config.ingest_path: # type: ignore[attr-defined] self._respond_text("Not Found", status=HTTPStatus.NOT_FOUND) return length = int(self.headers.get("Content-Length", "0")) if length <= 0: self._respond_text("Empty payload", status=HTTPStatus.BAD_REQUEST) return try: raw = self.rfile.read(length) payload = json.loads(raw) entry = self.server.build_entry(payload) # type: ignore[attr-defined] self.server.storage.add_entry(entry) # type: ignore[attr-defined] self.server.increment_messages() # type: ignore[attr-defined] except json.JSONDecodeError: self._respond_text("Invalid JSON payload", status=HTTPStatus.BAD_REQUEST) return except ValueError as err: self._respond_text(str(err), status=HTTPStatus.BAD_REQUEST) return except Exception: self._respond_text("Internal server error", status=HTTPStatus.INTERNAL_SERVER_ERROR) return self._respond_json({"status": "ok"}) def log_message(self, format: str, *args: Any) -> None: # noqa: A003 # Suppress default stdout logging; journald/systemd captures logs instead. return def _respond_json(self, payload: Dict[str, Any], status: HTTPStatus = HTTPStatus.OK) -> None: body = json.dumps(payload).encode("utf-8") self._respond(body, status=status, content_type="application/json; charset=utf-8") def _respond_text( self, body: str, status: HTTPStatus = HTTPStatus.OK, content_type: str = "text/plain; charset=utf-8", ) -> None: self._respond(body.encode("utf-8"), status=status, content_type=content_type) def _respond(self, body: bytes, status: HTTPStatus, content_type: str) -> None: self.send_response(status) self.send_header("Content-Type", content_type) self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) class CodexLoggerServer(ThreadingHTTPServer): daemon_threads = True def __init__(self, config: LoggerConfig): self.config = config self.storage = ConversationStorage(config) self._shutdown_event = threading.Event() self._message_counter = 0 server_address = (config.server.host, config.server.port) super().__init__(server_address, _RequestHandler, False) try: self.server_bind() self.server_activate() except Exception: # Ensure socket is released if activation fails self.server_close() raise self.timeout = 1 def process_request_thread(self, request, client_address): # type: ignore[override] try: super().process_request_thread(request, client_address) finally: self.shutdown_check() def build_entry(self, payload: Dict[str, Any]) -> LogEntry: required_fields = {"text", "role"} missing = required_fields - payload.keys() if missing: raise ValueError(f"Missing required fields: {', '.join(sorted(missing))}") timestamp = payload.get("timestamp") if not timestamp: timestamp = self._utc_timestamp() entry = LogEntry( turn_index=0, timestamp=timestamp, role=payload["role"], text=payload["text"], session_id=payload.get("session_id"), conversation_id=payload.get("conversation_id"), channel=payload.get("channel"), metadata=payload.get("metadata"), token_usage=self._parse_token_usage(payload.get("token_usage")), ) return entry def metrics_snapshot(self) -> Dict[str, int]: return {"codex_logger_messages_total": self._message_counter} def increment_messages(self) -> None: self._message_counter += 1 def shutdown_check(self) -> None: if self._shutdown_event.is_set(): self.shutdown() def set_shutdown(self) -> None: self._shutdown_event.set() def stop(self) -> None: self.set_shutdown() self.storage.flush() @staticmethod def _utc_timestamp() -> str: from datetime import datetime, timezone return datetime.now(timezone.utc).isoformat() @staticmethod def _parse_token_usage(raw: Optional[Any]) -> Optional[Dict[str, int]]: if raw is None: return None if not isinstance(raw, dict): raise ValueError("token_usage must be an object with integer counts") parsed: Dict[str, int] = {} for key, value in raw.items(): if not isinstance(value, int): raise ValueError("token_usage values must be integers") parsed[key] = value return parsed def install_signal_handlers(server: CodexLoggerServer) -> None: def handler(signum, _frame): server.stop() signal.signal(signal.SIGTERM, handler) signal.signal(signal.SIGINT, handler) def run_server(config_path: Optional[Path]) -> None: config = load_config(config_path) server = CodexLoggerServer(config) install_signal_handlers(server) with server: try: server.serve_forever() finally: server.stop() def main(argv: Optional[list[str]] = None) -> None: import argparse parser = argparse.ArgumentParser(description="Codex conversation logging daemon") parser.add_argument("--config", type=Path, help="Path to YAML configuration file") args = parser.parse_args(argv) run_server(args.config) if __name__ == "__main__": main()