Compare commits

..

3 Commits

Author SHA1 Message Date
Carlos Escalante
3d0547743f Log client details for API and agent requests
All checks were successful
Deploy to VPS / test (push) Successful in 1m42s
Deploy to VPS / deploy (push) Successful in 14s
2026-07-14 22:55:50 -06:00
Carlos Escalante
8ceed0ab2e Add individual salary deposit agent tool
All checks were successful
Deploy to VPS / test (push) Successful in 1m43s
Deploy to VPS / deploy (push) Successful in 27s
2026-07-14 22:41:53 -06:00
Carlos Escalante
ca3115e99c Add retained chat and API request diagnostics 2026-07-14 22:34:33 -06:00
14 changed files with 801 additions and 5 deletions

View File

@@ -0,0 +1,35 @@
"""track agui run id in chat logs
Revision ID: 10f4a4f71964
Revises: ecb99a158fbf
Create Date: 2026-07-14 22:32:16.771655
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
import sqlmodel
# revision identifiers, used by Alembic.
revision: str = '10f4a4f71964'
down_revision: Union[str, Sequence[str], None] = 'ecb99a158fbf'
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
"""Upgrade schema."""
# ### commands auto generated by Alembic - please adjust! ###
op.add_column('chatrunlog', sa.Column('agui_run_id', sqlmodel.sql.sqltypes.AutoString(), nullable=True))
op.create_index(op.f('ix_chatrunlog_agui_run_id'), 'chatrunlog', ['agui_run_id'], unique=False)
# ### end Alembic commands ###
def downgrade() -> None:
"""Downgrade schema."""
# ### commands auto generated by Alembic - please adjust! ###
op.drop_index(op.f('ix_chatrunlog_agui_run_id'), table_name='chatrunlog')
op.drop_column('chatrunlog', 'agui_run_id')
# ### end Alembic commands ###

View File

@@ -0,0 +1,37 @@
"""record chat client details
Revision ID: 86104b4ff6ae
Revises: 10f4a4f71964
Create Date: 2026-07-14 22:54:55.519380
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
import sqlmodel
# revision identifiers, used by Alembic.
revision: str = '86104b4ff6ae'
down_revision: Union[str, Sequence[str], None] = '10f4a4f71964'
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
"""Upgrade schema."""
# ### commands auto generated by Alembic - please adjust! ###
op.add_column('chatrunlog', sa.Column('client_ip', sa.Text(), nullable=True))
op.add_column('chatrunlog', sa.Column('user_agent', sa.Text(), nullable=True))
op.add_column('chatrunlog', sa.Column('browser', sa.Text(), nullable=True))
# ### end Alembic commands ###
def downgrade() -> None:
"""Downgrade schema."""
# ### commands auto generated by Alembic - please adjust! ###
op.drop_column('chatrunlog', 'browser')
op.drop_column('chatrunlog', 'user_agent')
op.drop_column('chatrunlog', 'client_ip')
# ### end Alembic commands ###

View File

@@ -0,0 +1,58 @@
"""add chat run audit logs
Revision ID: ecb99a158fbf
Revises: 794baf50635b
Create Date: 2026-07-14 22:22:06.035643
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
import sqlmodel
from sqlalchemy.dialects import postgresql
# revision identifiers, used by Alembic.
revision: str = 'ecb99a158fbf'
down_revision: Union[str, Sequence[str], None] = '794baf50635b'
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
"""Upgrade schema."""
# ### commands auto generated by Alembic - please adjust! ###
op.create_table('chatrunlog',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('request_id', sqlmodel.sql.sqltypes.AutoString(), nullable=False),
sa.Column('requested_thread_id', sqlmodel.sql.sqltypes.AutoString(), nullable=True),
sa.Column('response_thread_id', sqlmodel.sql.sqltypes.AutoString(), nullable=True),
sa.Column('model', sqlmodel.sql.sqltypes.AutoString(), nullable=False),
sa.Column('user_prompt', sa.Text(), nullable=True),
sa.Column('tool_calls', sa.JSON().with_variant(postgresql.JSONB(astext_type=sa.Text()), 'postgresql'), server_default='[]', nullable=False),
sa.Column('assistant_response', sa.Text(), nullable=True),
sa.Column('status', sqlmodel.sql.sqltypes.AutoString(), nullable=False),
sa.Column('error', sa.Text(), nullable=True),
sa.Column('started_at', sa.DateTime(), nullable=False),
sa.Column('completed_at', sa.DateTime(), nullable=True),
sa.Column('expires_at', sa.DateTime(), nullable=False),
sa.PrimaryKeyConstraint('id')
)
op.create_index(op.f('ix_chatrunlog_expires_at'), 'chatrunlog', ['expires_at'], unique=False)
op.create_index(op.f('ix_chatrunlog_request_id'), 'chatrunlog', ['request_id'], unique=True)
op.create_index(op.f('ix_chatrunlog_requested_thread_id'), 'chatrunlog', ['requested_thread_id'], unique=False)
op.create_index(op.f('ix_chatrunlog_response_thread_id'), 'chatrunlog', ['response_thread_id'], unique=False)
op.create_index(op.f('ix_chatrunlog_status'), 'chatrunlog', ['status'], unique=False)
# ### end Alembic commands ###
def downgrade() -> None:
"""Downgrade schema."""
# ### commands auto generated by Alembic - please adjust! ###
op.drop_index(op.f('ix_chatrunlog_status'), table_name='chatrunlog')
op.drop_index(op.f('ix_chatrunlog_response_thread_id'), table_name='chatrunlog')
op.drop_index(op.f('ix_chatrunlog_requested_thread_id'), table_name='chatrunlog')
op.drop_index(op.f('ix_chatrunlog_request_id'), table_name='chatrunlog')
op.drop_index(op.f('ix_chatrunlog_expires_at'), table_name='chatrunlog')
op.drop_table('chatrunlog')
# ### end Alembic commands ###

View File

@@ -35,6 +35,9 @@ How to answer:
"ayer", "el martes", a specific date) need get_daily_spending with that "ayer", "el martes", a specific date) need get_daily_spending with that
day's bounds — cycle or category summaries have NO per-day data, so never day's bounds — cycle or category summaries have NO per-day data, so never
answer a day question from them. answer a day question from them.
- For a question about a specific salary deposit or its amount (including
"cuánto recibí hoy"), call get_salary_deposits with the relevant date range;
get_salary_summary is aggregate-only and cannot answer it.
- Answer like an analyst, not a calculator. An aggregate number on its own - Answer like an analyst, not a calculator. An aggregate number on its own
is not an answer — show what it is made of, right away, without waiting is not an answer — show what it is made of, right away, without waiting
to be asked: to be asked:

View File

@@ -0,0 +1,153 @@
"""Durable diagnostics for individual AG-UI agent runs."""
from __future__ import annotations
import asyncio
import contextvars
from datetime import timedelta
from typing import Any
from sqlmodel import Session, delete, select
from app.config import settings
from app.db import engine
from app.models.models import ChatRunLog
from app.timeutil import utcnow
CHAT_LOG_RETENTION_DAYS = 90
_chat_run_id: contextvars.ContextVar[int | None] = contextvars.ContextVar(
"chat_run_id", default=None
)
def set_current_chat_run(run_id: int) -> contextvars.Token:
return _chat_run_id.set(run_id)
def reset_current_chat_run(token: contextvars.Token) -> None:
_chat_run_id.reset(token)
def current_chat_run_id() -> int | None:
return _chat_run_id.get()
def _last_user_prompt(messages: list[dict[str, Any]]) -> str | None:
for message in reversed(messages):
if message.get("role") == "user" and isinstance(message.get("content"), str):
return message["content"]
return None
def create_chat_run(
*,
request_id: str,
agui_run_id: str | None,
thread_id: str | None,
messages: list[dict[str, Any]],
client_ip: str | None = None,
user_agent: str | None = None,
browser: str | None = None,
) -> int:
now = utcnow()
with Session(engine) as session:
row = ChatRunLog(
request_id=request_id,
agui_run_id=agui_run_id,
requested_thread_id=thread_id,
model=settings.AGENT_MODEL,
client_ip=client_ip,
user_agent=user_agent,
browser=browser,
user_prompt=_last_user_prompt(messages),
expires_at=now + timedelta(days=CHAT_LOG_RETENTION_DAYS),
)
session.add(row)
session.commit()
session.refresh(row)
assert row.id is not None
return row.id
def _tool_activity(messages: list[dict[str, Any]]) -> list[dict[str, Any]]:
calls: dict[str, dict[str, Any]] = {}
order: list[str] = []
for message in messages:
if message.get("role") == "assistant":
for call in message.get("toolCalls") or message.get("tool_calls") or []:
call_id = str(call.get("id") or "")
if not call_id or call_id in calls:
continue
function = call.get("function") or {}
calls[call_id] = {
"id": call_id,
"name": function.get("name"),
"arguments": function.get("arguments"),
}
order.append(call_id)
elif message.get("role") == "tool":
call_id = str(message.get("toolCallId") or message.get("tool_call_id") or "")
if call_id in calls:
calls[call_id]["result"] = message.get("content")
if message.get("error") is not None:
calls[call_id]["error"] = message["error"]
return [calls[call_id] for call_id in order]
def _last_assistant_response(messages: list[dict[str, Any]]) -> str | None:
for message in reversed(messages):
if message.get("role") == "assistant" and isinstance(message.get("content"), str):
return message["content"]
return None
def finalize_chat_run_in_session(
session: Session, *, run_id: int, response_thread_id: str, messages: list[dict[str, Any]]
) -> None:
row = session.get(ChatRunLog, run_id)
if row is None:
return
row.response_thread_id = response_thread_id
row.tool_calls = _tool_activity(messages)
row.assistant_response = _last_assistant_response(messages)
row.status = "completed"
row.completed_at = utcnow()
session.add(row)
def mark_chat_run_failed(run_id: int, error: str) -> None:
with Session(engine) as session:
row = session.get(ChatRunLog, run_id)
if row is None or row.status == "completed":
return
row.status = "failed"
row.error = error[:4000]
row.completed_at = utcnow()
session.add(row)
session.commit()
def finish_pending_chat_run(run_id: int) -> None:
with Session(engine) as session:
row = session.get(ChatRunLog, run_id)
if row is None or row.status != "started":
return
row.status = "completed"
row.completed_at = utcnow()
session.add(row)
session.commit()
def purge_expired_chat_runs() -> int:
with Session(engine) as session:
result = session.exec(delete(ChatRunLog).where(ChatRunLog.expires_at < utcnow()))
session.commit()
return result.rowcount or 0
async def purge_expired_chat_runs_periodically() -> None:
"""Purge at startup and then once per day for a bounded audit footprint."""
while True:
await asyncio.to_thread(purge_expired_chat_runs)
await asyncio.sleep(24 * 60 * 60)

View File

@@ -25,6 +25,7 @@ from agent_framework_ag_ui import AGUIThreadSnapshot
from sqlmodel import Session, select from sqlmodel import Session, select
from app.db import engine from app.db import engine
from app.agent.chat_audit import current_chat_run_id, finalize_chat_run_in_session
from app.models.models import ChatThreadSnapshot from app.models.models import ChatThreadSnapshot
from app.timeutil import utcnow from app.timeutil import utcnow
@@ -130,6 +131,14 @@ class PostgresSnapshotStore:
row.interrupt = snapshot.interrupt row.interrupt = snapshot.interrupt
row.updated_at = utcnow() row.updated_at = utcnow()
session.add(row) session.add(row)
run_id = current_chat_run_id()
if run_id is not None:
finalize_chat_run_in_session(
session,
run_id=run_id,
response_thread_id=thread_id,
messages=row.messages,
)
session.commit() session.commit()
await asyncio.to_thread(_save) await asyncio.to_thread(_save)

View File

@@ -48,6 +48,7 @@ from app.services.exchange_rate import (
) )
_session_ctx: contextvars.ContextVar[Session] = contextvars.ContextVar("agent_session") _session_ctx: contextvars.ContextVar[Session] = contextvars.ContextVar("agent_session")
SALARY_TRANSACTION_TYPES = (TransactionType.SALARY, TransactionType.DEPOSITO)
def set_session(session: Session) -> contextvars.Token: def set_session(session: Session) -> contextvars.Token:
@@ -339,7 +340,7 @@ def get_salary_summary() -> dict:
func.count(), func.count(),
func.coalesce(func.sum(amount_crc), 0), func.coalesce(func.sum(amount_crc), 0),
func.max(Transaction.date), func.max(Transaction.date),
).where(Transaction.transaction_type == TransactionType.SALARY) ).where(col(Transaction.transaction_type).in_(SALARY_TRANSACTION_TYPES))
).first() ).first()
count = row[0] if row else 0 count = row[0] if row else 0
total = float(row[1]) if row else 0.0 total = float(row[1]) if row else 0.0
@@ -347,6 +348,44 @@ def get_salary_summary() -> dict:
return {"count": count, "total_crc": total, "latest_date": latest} return {"count": count, "total_crc": total, "latest_date": latest}
def get_salary_deposits(
limit: Annotated[int, Field(ge=1, le=100, description="How many deposits to return")] = 20,
start_date: Annotated[
Optional[str], Field(description="ISO date lower bound, inclusive")
] = None,
end_date: Annotated[
Optional[str], Field(description="ISO date upper bound, exclusive")
] = None,
) -> list[dict]:
"""Individual salary deposits, newest first. Use this for questions about
a specific salary or a day/range (for example, "cuánto recibí hoy").
Amounts remain in their recorded currency; use the currency field when
presenting them."""
q = select(Transaction).where(
col(Transaction.transaction_type).in_(SALARY_TRANSACTION_TYPES)
)
if start_date:
q = q.where(Transaction.date >= datetime.fromisoformat(start_date))
if end_date:
q = q.where(Transaction.date < datetime.fromisoformat(end_date))
q = q.order_by(col(Transaction.date).desc(), col(Transaction.id).desc()).limit(limit)
return [
{
"id": t.id,
"date": t.date.isoformat(),
"amount": float(t.amount),
"currency": t.currency.value,
"merchant": t.merchant,
"source": t.source.value,
"bank": t.bank.value,
"transaction_type": t.transaction_type.value,
"reference": t.reference,
"notes": t.notes,
}
for t in _s().exec(q).all()
]
def get_municipal_receipts( def get_municipal_receipts(
limit: Annotated[int, Field(ge=1, le=50)] = 12, limit: Annotated[int, Field(ge=1, le=50)] = 12,
offset: Annotated[int, Field(ge=0, description="Skip the N most recent")] = 0, offset: Annotated[int, Field(ge=0, description="Skip the N most recent")] = 0,
@@ -567,6 +606,7 @@ TOOLS = [
list_recurring_items, list_recurring_items,
get_pension_snapshots, get_pension_snapshots,
get_salary_summary, get_salary_summary,
get_salary_deposits,
get_municipal_receipts, get_municipal_receipts,
get_analytics_by_category, get_analytics_by_category,
get_monthly_trend, get_monthly_trend,

31
backend/app/logging.py Normal file
View File

@@ -0,0 +1,31 @@
"""Structured, redacted application logging for container stdout."""
from __future__ import annotations
import json
from datetime import date, datetime
from enum import Enum
from typing import Any
class _JSONEncoder(json.JSONEncoder):
def default(self, value: Any) -> Any:
if isinstance(value, (datetime, date)):
return value.isoformat()
if isinstance(value, Enum):
return value.value
return super().default(value)
def configure_structured_logging() -> None:
"""Reserved lifecycle hook for the structured stdout audit channel."""
def log_event(event: str, **fields: Any) -> None:
"""Write a single JSON line; callers must not pass credentials or bodies."""
# stdout is Docker's durable capture point. `flush=True` also makes this
# visible immediately in `docker compose logs` during an active SSE run.
print(
json.dumps({"event": event, **fields}, cls=_JSONEncoder, separators=(",", ":")),
flush=True,
)

View File

@@ -1,6 +1,7 @@
import asyncio import asyncio
import json import json
import uuid import uuid
from time import perf_counter
from http.cookies import SimpleCookie from http.cookies import SimpleCookie
from contextlib import asynccontextmanager from contextlib import asynccontextmanager
@@ -11,6 +12,14 @@ from jose import JWTError, jwt
from pydantic import BaseModel from pydantic import BaseModel
from app.agent.agent import build_agent 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.snapshot_store import PostgresSnapshotStore
from app.agent.tools import reset_session, set_session from app.agent.tools import reset_session, set_session
from app.api.v1.router import api_router from app.api.v1.router import api_router
@@ -25,9 +34,45 @@ from app.db import get_session, run_alembic_upgrade
from app.seed import seed_db from app.seed import seed_db
from app.timeutil import reset_client_timezone, set_client_timezone from app.timeutil import reset_client_timezone, set_client_timezone
from app.services.exchange_rate import refresh_rates_periodically from app.services.exchange_rate import refresh_rates_periodically
from app.logging import configure_structured_logging, log_event
AGENT_PATH = "/api/v1/agent/agui" AGENT_PATH = "/api/v1/agent/agui"
MAX_USER_AGENT_LENGTH = 512
def _request_client_details(request: Request) -> tuple[str, str | None, str | None]:
"""Return the browser client details propagated through the frontend BFF.
Production nginx-proxy supplies X-Real-IP and the frontend forwards it to
FastAPI. For direct local development, fall back to X-Forwarded-For and
then the socket peer. The BFF is the only production path to the backend,
which is not publicly exposed.
"""
client_ip = request.headers.get("x-real-ip", "").strip()
if not client_ip:
forwarded = request.headers.get("x-forwarded-for", "")
client_ip = forwarded.split(",", 1)[0].strip()
if not client_ip:
client_ip = request.client.host if request.client else "unknown"
user_agent = request.headers.get("user-agent", "").strip()[:MAX_USER_AGENT_LENGTH] or None
if not user_agent:
return client_ip, None, None
ua = user_agent.lower()
if "edg/" in ua or "edgios/" in ua:
browser = "Edge"
elif "opr/" in ua or "opera" in ua:
browser = "Opera"
elif "firefox/" in ua or "fxios/" in ua:
browser = "Firefox"
elif "chrome/" in ua or "crios/" in ua:
browser = "Chrome"
elif "safari/" in ua:
browser = "Safari"
else:
browser = "Other"
return client_ip, user_agent, browser
def _pair_orphan_tool_calls(messages: list) -> list: def _pair_orphan_tool_calls(messages: list) -> list:
@@ -101,17 +146,24 @@ def _drop_stale_client_history(session, thread_id, messages: list) -> list:
@asynccontextmanager @asynccontextmanager
async def lifespan(app: FastAPI): async def lifespan(app: FastAPI):
configure_structured_logging()
run_alembic_upgrade() run_alembic_upgrade()
seed_db() seed_db()
rate_refresh_task = asyncio.create_task(refresh_rates_periodically()) rate_refresh_task = asyncio.create_task(refresh_rates_periodically())
chat_audit_cleanup_task = asyncio.create_task(purge_expired_chat_runs_periodically())
try: try:
yield yield
finally: finally:
rate_refresh_task.cancel() rate_refresh_task.cancel()
chat_audit_cleanup_task.cancel()
try: try:
await rate_refresh_task await rate_refresh_task
except asyncio.CancelledError: except asyncio.CancelledError:
pass pass
try:
await chat_audit_cleanup_task
except asyncio.CancelledError:
pass
app = FastAPI(title="WealthySmart API", version="0.1.0", lifespan=lifespan) app = FastAPI(title="WealthySmart API", version="0.1.0", lifespan=lifespan)
@@ -125,6 +177,67 @@ 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
client_ip, user_agent, browser = _request_client_details(request)
request.state.client_ip = client_ip
request.state.user_agent = user_agent
request.state.browser = browser
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,
client_ip=client_ip,
user_agent=user_agent,
browser=browser,
)
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") @app.middleware("http")
async def agent_auth_and_session(request: Request, call_next): async def agent_auth_and_session(request: Request, call_next):
"""For the AG-UI route, validate the JWT, repair message history, and """For the AG-UI route, validate the JWT, repair message history, and
@@ -159,6 +272,7 @@ async def agent_auth_and_session(request: Request, call_next):
session = next(session_gen) session = next(session_gen)
token_var = set_session(session) token_var = set_session(session)
chat_run_id: int | None = None
# Repair orphan tool_calls before the MAF agent sees the message history. # 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", ""): if request.method == "POST" and "application/json" in request.headers.get("content-type", ""):
raw = await request.body() raw = await request.body()
@@ -172,16 +286,75 @@ async def agent_auth_and_session(request: Request, call_next):
) )
body["messages"] = _pair_orphan_tool_calls(body["messages"]) body["messages"] = _pair_orphan_tool_calls(body["messages"])
raw = json.dumps(body).encode() raw = json.dumps(body).encode()
except Exception: log_event(
pass "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 []
),
client_ip=request.state.client_ip,
user_agent=request.state.user_agent,
browser=request.state.browser,
)
# 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 [],
client_ip=request.state.client_ip,
user_agent=request.state.user_agent,
browser=request.state.browser,
)
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. # Starlette caches the body; replace it so call_next sees the fixed bytes.
request._body = raw # type: ignore[attr-defined] request._body = raw # type: ignore[attr-defined]
# Browser's IANA timezone, forwarded by the BFF — lets tools resolve # Browser's IANA timezone, forwarded by the BFF — lets tools resolve
# "today" wherever the user is (falls back to Costa Rica). # "today" wherever the user is (falls back to Costa Rica).
tz_token = set_client_timezone(request.headers.get("x-client-timezone")) 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: 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: finally:
if audit_token is not None:
reset_current_chat_run(audit_token)
reset_client_timezone(tz_token) reset_client_timezone(tz_token)
reset_session(token_var) reset_session(token_var)
try: try:
@@ -190,6 +363,12 @@ async def agent_auth_and_session(request: Request, call_next):
pass 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 # Register app routes
app.include_router(api_router) app.include_router(api_router)

View File

@@ -4,7 +4,7 @@ from decimal import Decimal
from typing import Annotated, Optional from typing import Annotated, Optional
from pydantic import PlainSerializer from pydantic import PlainSerializer
from sqlalchemy import JSON, Column, UniqueConstraint from sqlalchemy import JSON, Column, Text, UniqueConstraint
from sqlalchemy.dialects.postgresql import JSONB from sqlalchemy.dialects.postgresql import JSONB
from sqlmodel import Field, Relationship, SQLModel from sqlmodel import Field, Relationship, SQLModel
@@ -566,3 +566,38 @@ class ChatThreadSnapshot(SQLModel, table=True):
default=None, sa_column=Column(JSON_OR_JSONB, nullable=True) default=None, sa_column=Column(JSON_OR_JSONB, nullable=True)
) )
updated_at: datetime = Field(default_factory=utcnow) updated_at: datetime = Field(default_factory=utcnow)
# --- Assistant Chat Run Audit ---
class ChatRunLog(SQLModel, table=True):
"""Append-only diagnostic record for an AG-UI agent/run request.
Thread snapshots are mutable conversation state. This table instead keeps
the request, model, tool activity, and final emitted answer for a bounded
retention period so individual assistant runs can be investigated later.
"""
id: Optional[int] = Field(default=None, primary_key=True)
request_id: str = Field(index=True, unique=True)
agui_run_id: Optional[str] = Field(default=None, index=True)
requested_thread_id: Optional[str] = Field(default=None, index=True)
response_thread_id: Optional[str] = Field(default=None, index=True)
model: str
client_ip: Optional[str] = Field(default=None, sa_column=Column(Text, nullable=True))
user_agent: Optional[str] = Field(default=None, sa_column=Column(Text, nullable=True))
browser: Optional[str] = Field(default=None, sa_column=Column(Text, nullable=True))
user_prompt: Optional[str] = Field(default=None, sa_column=Column(Text, nullable=True))
tool_calls: list = Field(
default_factory=list,
sa_column=Column(JSON_OR_JSONB, nullable=False, server_default="[]"),
)
assistant_response: Optional[str] = Field(
default=None, sa_column=Column(Text, nullable=True)
)
status: str = Field(default="started", index=True)
error: Optional[str] = Field(default=None, sa_column=Column(Text, nullable=True))
started_at: datetime = Field(default_factory=utcnow)
completed_at: Optional[datetime] = Field(default=None)
expires_at: datetime = Field(index=True)

View File

@@ -17,6 +17,7 @@ from app.models.models import (
MunicipalReceipt, MunicipalReceipt,
PensionSnapshot, PensionSnapshot,
Transaction, Transaction,
TransactionType,
WaterMeterReading, WaterMeterReading,
) )
@@ -173,6 +174,52 @@ def test_recent_transactions_exclude_future_cuotas(bound_session):
assert "FUTURE CUOTA:03/06" not in merchants assert "FUTURE CUOTA:03/06" not in merchants
def test_salary_deposits_returns_individual_filtered_rows(bound_session):
bound_session.add_all(
[
Transaction(
amount=Decimal("1500"),
merchant="SALARY JULY",
date=datetime(2026, 7, 14, 9),
transaction_type=TransactionType.SALARY,
reference="PAY-2026-07",
),
Transaction(
amount=Decimal("1400"),
merchant="SALARY JUNE",
date=datetime(2026, 6, 14, 9),
transaction_type=TransactionType.SALARY,
),
Transaction(
amount=Decimal("999"),
merchant="PURCHASE",
date=datetime(2026, 7, 14, 10),
),
]
)
bound_session.commit()
out = tools.get_salary_deposits(
start_date="2026-07-14", end_date="2026-07-15"
)
json.dumps(out)
assert out == [
{
"id": out[0]["id"],
"date": "2026-07-14T09:00:00",
"amount": 1500.0,
"currency": "CRC",
"merchant": "SALARY JULY",
"source": "CREDIT_CARD",
"bank": "BAC",
"transaction_type": "SALARY",
"reference": "PAY-2026-07",
"notes": None,
}
]
def test_get_current_date_is_costa_rica_and_cycle_consistent(): def test_get_current_date_is_costa_rica_and_cycle_consistent():
from app.timeutil import today_cr from app.timeutil import today_cr

View File

@@ -0,0 +1,139 @@
"""Chat-run audit persistence and redacted HTTP access logs."""
import asyncio
from datetime import timedelta
from agent_framework_ag_ui import AGUIThreadSnapshot
from starlette.requests import Request
from starlette.responses import Response
from sqlmodel import select
import app.agent.chat_audit as chat_audit
import app.agent.snapshot_store as snapshot_store_module
from app.agent.snapshot_store import PostgresSnapshotStore
from app.models.models import ChatRunLog
from app.timeutil import utcnow
def _audit_store(monkeypatch, engine):
monkeypatch.setattr(chat_audit, "engine", engine)
monkeypatch.setattr(snapshot_store_module, "engine", engine)
return PostgresSnapshotStore()
def test_snapshot_completes_chat_run_with_tools_and_response(monkeypatch, engine, session):
store = _audit_store(monkeypatch, engine)
run_id = chat_audit.create_chat_run(
request_id="req-1",
agui_run_id="run-1",
thread_id="main",
messages=[{"role": "user", "content": "¿Cuál es mi saldo?"}],
client_ip="203.0.113.7",
user_agent="Mozilla/5.0 TestBrowser/1.0",
browser="TestBrowser",
)
token = chat_audit.set_current_chat_run(run_id)
try:
asyncio.run(
store.save(
scope="default",
thread_id="resp_123",
snapshot=AGUIThreadSnapshot(
messages=[
{"id": "u1", "role": "user", "content": "¿Cuál es mi saldo?"},
{
"id": "a1",
"role": "assistant",
"content": None,
"tool_calls": [
{
"id": "call-1",
"type": "function",
"function": {"name": "get_net_worth", "arguments": "{}"},
}
],
},
{"id": "t1", "role": "tool", "tool_call_id": "call-1", "content": "{\"net_crc\": 1}"},
{"id": "a2", "role": "assistant", "content": "Tu saldo es ₡1."},
]
),
)
)
finally:
chat_audit.reset_current_chat_run(token)
row = session.get(ChatRunLog, run_id)
assert row is not None
assert row.status == "completed"
assert row.agui_run_id == "run-1"
assert row.response_thread_id == "resp_123"
assert row.user_prompt == "¿Cuál es mi saldo?"
assert row.client_ip == "203.0.113.7"
assert row.user_agent == "Mozilla/5.0 TestBrowser/1.0"
assert row.browser == "TestBrowser"
assert row.assistant_response == "Tu saldo es ₡1."
assert row.tool_calls == [
{
"id": "call-1",
"name": "get_net_worth",
"arguments": "{}",
"result": '{"net_crc": 1}',
}
]
def test_failure_and_retention_cleanup(monkeypatch, engine, session):
_audit_store(monkeypatch, engine)
failed_id = chat_audit.create_chat_run(
request_id="req-failed", agui_run_id="run-failed", thread_id="main", messages=[]
)
chat_audit.mark_chat_run_failed(failed_id, "RuntimeError: provider unavailable")
assert session.get(ChatRunLog, failed_id).status == "failed"
expired = ChatRunLog(
request_id="req-expired",
model="gpt-5.6-luna",
expires_at=utcnow() - timedelta(seconds=1),
)
session.add(expired)
session.commit()
assert chat_audit.purge_expired_chat_runs() == 1
assert session.exec(select(ChatRunLog).where(ChatRunLog.request_id == "req-expired")).first() is None
def test_api_access_log_never_contains_request_body(monkeypatch):
from app.main import log_api_request
logged = []
monkeypatch.setattr("app.main.log_event", lambda event, **fields: logged.append((event, fields)))
scope = {
"type": "http",
"method": "POST",
"path": "/api/auth/login",
"raw_path": b"/api/auth/login",
"query_string": b"",
"headers": [
(b"x-real-ip", b"203.0.113.9"),
(b"user-agent", b"Mozilla/5.0 Chrome/138.0.0.0 Safari/537.36"),
],
"scheme": "http",
"server": ("testserver", 80),
"client": ("testclient", 1234),
}
request = Request(scope)
async def call_next(_request):
return Response(status_code=201)
response = asyncio.run(log_api_request(request, call_next))
assert response.headers["X-Request-ID"]
assert logged[0][0] == "api_request"
fields = logged[0][1]
assert fields["path"] == "/api/auth/login"
assert fields["status_code"] == 201
assert fields["client_ip"] == "203.0.113.9"
assert fields["browser"] == "Chrome"
assert fields["user_agent"] == "Mozilla/5.0 Chrome/138.0.0.0 Safari/537.36"
assert "password" not in fields
assert "body" not in fields

View File

@@ -16,6 +16,11 @@ services:
interval: 10s interval: 10s
timeout: 5s timeout: 5s
retries: 5 retries: 5
logging:
driver: json-file
options:
max-size: "10m"
max-file: "5"
backend: backend:
build: build:
@@ -51,6 +56,11 @@ services:
timeout: 5s timeout: 5s
retries: 3 retries: 3
start_period: 10s start_period: 10s
logging:
driver: json-file
options:
max-size: "10m"
max-file: "5"
frontend: frontend:
build: build:
@@ -84,6 +94,11 @@ services:
interval: 30s interval: 30s
timeout: 10s timeout: 10s
retries: 3 retries: 3
logging:
driver: json-file
options:
max-size: "10m"
max-file: "5"
networks: networks:
wealthysmart-network: wealthysmart-network:

View File

@@ -15,6 +15,11 @@ services:
interval: 5s interval: 5s
timeout: 3s timeout: 3s
retries: 5 retries: 5
logging:
driver: json-file
options:
max-size: "10m"
max-file: "5"
backend: backend:
build: build:
@@ -40,6 +45,11 @@ services:
depends_on: depends_on:
db: db:
condition: service_healthy condition: service_healthy
logging:
driver: json-file
options:
max-size: "10m"
max-file: "5"
develop: develop:
watch: watch:
- path: ./backend/app - path: ./backend/app
@@ -66,6 +76,11 @@ services:
AGENT_URL: http://backend:8000/api/v1/agent/agui AGENT_URL: http://backend:8000/api/v1/agent/agui
depends_on: depends_on:
- backend - backend
logging:
driver: json-file
options:
max-size: "10m"
max-file: "5"
develop: develop:
watch: watch:
- path: ./frontend/src - path: ./frontend/src