Add retained chat and API request diagnostics

backend/app/main.py

@@ -1,6 +1,7 @@
 import asyncio
 import json
 import uuid
+from time import perf_counter
 from http.cookies import SimpleCookie
 from contextlib import asynccontextmanager
 
@@ -11,6 +12,14 @@ from jose import JWTError, jwt
 from pydantic import BaseModel
 
 from app.agent.agent import build_agent
+from app.agent.chat_audit import (
+    create_chat_run,
+    finish_pending_chat_run,
+    mark_chat_run_failed,
+    purge_expired_chat_runs_periodically,
+    reset_current_chat_run,
+    set_current_chat_run,
+)
 from app.agent.snapshot_store import PostgresSnapshotStore
 from app.agent.tools import reset_session, set_session
 from app.api.v1.router import api_router
@@ -25,6 +34,7 @@ from app.db import get_session, run_alembic_upgrade
 from app.seed import seed_db
 from app.timeutil import reset_client_timezone, set_client_timezone
 from app.services.exchange_rate import refresh_rates_periodically
+from app.logging import configure_structured_logging, log_event
 
 
 AGENT_PATH = "/api/v1/agent/agui"
@@ -101,17 +111,24 @@ def _drop_stale_client_history(session, thread_id, messages: list) -> list:
 
 @asynccontextmanager
 async def lifespan(app: FastAPI):
+    configure_structured_logging()
     run_alembic_upgrade()
     seed_db()
     rate_refresh_task = asyncio.create_task(refresh_rates_periodically())
+    chat_audit_cleanup_task = asyncio.create_task(purge_expired_chat_runs_periodically())
     try:
         yield
     finally:
         rate_refresh_task.cancel()
+        chat_audit_cleanup_task.cancel()
         try:
             await rate_refresh_task
         except asyncio.CancelledError:
             pass
+        try:
+            await chat_audit_cleanup_task
+        except asyncio.CancelledError:
+            pass
 
 
 app = FastAPI(title="WealthySmart API", version="0.1.0", lifespan=lifespan)
@@ -125,6 +142,60 @@ app.add_middleware(
 )
 
 
+async def log_api_request(request: Request, call_next):
+    """Emit redacted, one-line diagnostics for API requests only.
+
+    This deliberately records no request/response bodies, headers, tokens,
+    passwords, or financial payloads. Full assistant content is retained only
+    in the bounded chat-run audit table.
+    """
+    request_id = uuid.uuid4().hex
+    request.state.request_id = request_id
+    started = perf_counter()
+    status_code = 500
+    is_logged_api = request.url.path.startswith("/api/") and request.url.path != "/api/health"
+
+    def emit(error_type: str | None = None) -> None:
+        if not is_logged_api:
+            return
+        route = request.scope.get("route")
+        log_event(
+            "api_request",
+            request_id=request_id,
+            method=request.method,
+            path=getattr(route, "path", request.url.path),
+            status_code=status_code,
+            duration_ms=round((perf_counter() - started) * 1000, 1),
+            error_type=error_type,
+        )
+
+    try:
+        response = await call_next(request)
+        status_code = response.status_code
+        response.headers["X-Request-ID"] = request_id
+        original_iterator = getattr(response, "body_iterator", None)
+        if original_iterator is None:
+            emit()
+            return response
+
+        async def logged_stream():
+            error_type: str | None = None
+            try:
+                async for chunk in original_iterator:
+                    yield chunk
+            except Exception as exc:
+                error_type = type(exc).__name__
+                raise
+            finally:
+                emit(error_type)
+
+        response.body_iterator = logged_stream()
+        return response
+    except Exception as exc:
+        emit(type(exc).__name__)
+        raise
+
+
 @app.middleware("http")
 async def agent_auth_and_session(request: Request, call_next):
     """For the AG-UI route, validate the JWT, repair message history, and
@@ -159,6 +230,7 @@ async def agent_auth_and_session(request: Request, call_next):
     session = next(session_gen)
     token_var = set_session(session)
 
+    chat_run_id: int | None = None
     # Repair orphan tool_calls before the MAF agent sees the message history.
     if request.method == "POST" and "application/json" in request.headers.get("content-type", ""):
         raw = await request.body()
@@ -172,16 +244,69 @@ async def agent_auth_and_session(request: Request, call_next):
                 )
                 body["messages"] = _pair_orphan_tool_calls(body["messages"])
                 raw = json.dumps(body).encode()
-        except Exception:
-            pass
+            log_event(
+                "agent_request",
+                request_id=request.state.request_id,
+                agui_run_id=body.get("runId"),
+                thread_id=body.get("threadId") or body.get("thread_id"),
+                message_count=len(body.get("messages") or []),
+                has_user_message=any(
+                    message.get("role") == "user" for message in body.get("messages") or []
+                ),
+            )
+            # The BFF removes its `method: agent/run` envelope before
+            # forwarding to MAF. A non-empty user message is the reliable
+            # backend-side distinction between an agent run and hydration.
+            if any(message.get("role") == "user" for message in body.get("messages") or []):
+                chat_run_id = create_chat_run(
+                    request_id=request.state.request_id,
+                    agui_run_id=body.get("runId"),
+                    thread_id=body.get("threadId") or body.get("thread_id"),
+                    messages=body.get("messages") or [],
+                )
+        except Exception as exc:
+            # Parsing/repair must not prevent the agent endpoint from serving;
+            # the access log will still capture a resulting error response.
+            log_event(
+                "chat_audit_start_failed",
+                request_id=request.state.request_id,
+                error_type=type(exc).__name__,
+            )
+            body = None
         # Starlette caches the body; replace it so call_next sees the fixed bytes.
         request._body = raw  # type: ignore[attr-defined]
     # Browser's IANA timezone, forwarded by the BFF — lets tools resolve
     # "today" wherever the user is (falls back to Costa Rica).
     tz_token = set_client_timezone(request.headers.get("x-client-timezone"))
+    audit_token = set_current_chat_run(chat_run_id) if chat_run_id is not None else None
     try:
-        return await call_next(request)
+        response = await call_next(request)
+        if chat_run_id is None:
+            return response
+
+        original_iterator = response.body_iterator
+
+        async def audit_stream():
+            try:
+                async for chunk in original_iterator:
+                    yield chunk
+            except Exception as exc:
+                mark_chat_run_failed(chat_run_id, f"{type(exc).__name__}: {exc}")
+                raise
+            else:
+                # MAF normally finalizes via PostgresSnapshotStore.save(). If
+                # it emitted no snapshot, do not leave a permanently-open row.
+                finish_pending_chat_run(chat_run_id)
+
+        response.body_iterator = audit_stream()
+        return response
+    except Exception as exc:
+        if chat_run_id is not None:
+            mark_chat_run_failed(chat_run_id, f"{type(exc).__name__}: {exc}")
+        raise
     finally:
+        if audit_token is not None:
+            reset_current_chat_run(audit_token)
         reset_client_timezone(tz_token)
         reset_session(token_var)
         try:
@@ -190,6 +315,12 @@ async def agent_auth_and_session(request: Request, call_next):
             pass
 
 
+# FastAPI wraps function middlewares in reverse registration order. Register
+# this after the AG-UI auth middleware so it observes every API response,
+# including auth failures that return before calling the inner application.
+app.middleware("http")(log_api_request)
+
+
 # Register app routes
 app.include_router(api_router)