feat: customer UI rebuild -- 5-destination IA, real-time push, tax-lot designation, pagination

backend/.importlinter

@@ -42,6 +42,7 @@ forbidden_modules =
     app.integrations.azure
     app.integrations.marketdata
     app.integrations.fake
+    app.integrations.redis
 
 [importlinter:contract:models-know-no-business-rules]
 name = models never import services or controllers (S0 §3)

backend/Dockerfile

@@ -28,8 +28,13 @@ RUN uv sync --frozen --no-dev
 FROM python:3.12-slim AS runtime
 
 # libpq is needed at runtime by psycopg[binary]'s C extension load path on slim images.
+# `apt-get upgrade` pulls the base image's own OS packages up to their latest patched version at
+# build time -- without it, a build can silently ship whatever CVEs were open in the base image's
+# last-published layer, unrelated to anything this Dockerfile installs itself (verified 2026-09-16:
+# 3 CRITICAL + 9 HIGH CVEs in gzip/perl-base/libsqlite3-0/libpcre2-8-0, all fixed by this one line).
 RUN apt-get update \
     && apt-get install -y --no-install-recommends libpq5 \
+    && apt-get upgrade -y \
     && rm -rf /var/lib/apt/lists/*
 
 RUN groupadd --system trueup && useradd --system --gid trueup --create-home trueup
@@ -47,4 +52,15 @@ EXPOSE 8000
 
 # Migrations run as a separate CD step (never at container start -- a crash-looping web replica
 # must never race a migration against itself). Gunicorn, not Flask's dev server, serves requests.
-CMD ["gunicorn", "--bind", "0.0.0.0:8000", "--workers", "4", "--timeout", "30", "wsgi:app"]
+#
+# gthread, not the default sync worker (S12 S6, NFR-18): a sync worker's arbiter heartbeat is
+# only refreshed between requests, never while a response body is being written, so a long-lived
+# SSE stream (GET /api/v1/events/stream, up to 30 minutes) would be SIGKILLed by murder_workers()
+# at --timeout regardless of its value -- taking any other in-flight request on that worker with
+# it. gthread's own run loop calls notify() independently of request threads, so it survives long
+# streams; 2 workers x 32 threads gives 64 concurrent slots on 0.5 vCPU / 1Gi, each SSE connection
+# pinning one thread blocked on Redis I/O (never CPU, never the DB pool -- the SSE view opens no
+# UnitOfWork).
+CMD ["gunicorn", "--bind", "0.0.0.0:8000", \
+     "--worker-class", "gthread", "--workers", "2", "--threads", "32", \
+     "--timeout", "60", "wsgi:app"]

backend/alembic/versions/20260908_2034-cd483b78ace2_add_order_designated_lot_ids.py

@@ -0,0 +1,33 @@
+"""add order.designated_lot_ids
+
+Revision ID: cd483b78ace2
+Revises: 24a90b294386
+Create Date: 2026-09-08 20:34:28.910779
+
+"""
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from sqlalchemy.dialects import postgresql
+
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = 'cd483b78ace2'
+down_revision: str | Sequence[str] | None = '24a90b294386'
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    """Upgrade schema."""
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.add_column('order', sa.Column('designated_lot_ids', postgresql.ARRAY(sa.UUID()), nullable=True))
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    """Downgrade schema."""
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_column('order', 'designated_lot_ids')
+    # ### end Alembic commands ###

backend/app/__init__.py

@@ -11,19 +11,22 @@
 from app.config import Settings, get_settings
 from app.controllers.admin.agent_requests import agent_requests_bp
 from app.controllers.admin.customers import admin_customers_bp
+from app.controllers.admin.events import admin_events_bp
 from app.controllers.admin.kyc_overrides import admin_kyc_overrides_bp
 from app.controllers.admin.rebalance import admin_rebalance_bp
 from app.controllers.admin.staff_api_tokens import staff_api_tokens_bp
 from app.controllers.api.auth import auth_bp, init_auth
 from app.controllers.api.breaks import breaks_bp
 from app.controllers.api.chat import chat_bp
+from app.controllers.api.events import events_bp
 from app.controllers.api.fees import fees_bp
 from app.controllers.api.funding import funding_bp
 from app.controllers.api.identity import identity_bp
 from app.controllers.api.lots import lots_bp
 from app.controllers.api.orders import orders_bp
 from app.controllers.api.portfolios import portfolios_bp
 from app.controllers.api.profile import profile_bp
+from app.controllers.api.securities import securities_bp
 from app.controllers.api.statements import statements_bp
 from app.controllers.api.valuation import valuation_bp
 from app.controllers.health import health_bp
@@ -103,8 +106,11 @@ def create_app(settings: Settings | None = None) -> Flask:
     app.register_blueprint(funding_bp)
     app.register_blueprint(breaks_bp)
     app.register_blueprint(chat_bp)
+    app.register_blueprint(events_bp)
+    app.register_blueprint(admin_events_bp)
     app.register_blueprint(portfolios_bp)
     app.register_blueprint(profile_bp)
+    app.register_blueprint(securities_bp)
     app.register_blueprint(admin_rebalance_bp)
     app.register_blueprint(admin_customers_bp)
     app.register_blueprint(admin_kyc_overrides_bp)

backend/app/controllers/admin/events.py

@@ -0,0 +1,56 @@
+"""Adviser/admin real-time push (S12 §6, NFR-18): a single SSE stream over `adviser:events`,
+carrying newly-opened reconciliation breaks. Staff-only (`@requires_role`); a dropped/never-open
+connection is not data loss -- see `app/controllers/api/events.py`'s module docstring.
+"""
+
+from __future__ import annotations
+
+import json
+import time
+from typing import TYPE_CHECKING, Any
+
+from flask import Blueprint, Response, stream_with_context
+
+from app.config import get_settings
+from app.core.security import requires_role
+from app.extensions import limiter
+from app.integrations.redis.event_bus import RedisEventBus
+from app.services.ops.event_publisher import ADVISER_CHANNEL
+from app.views.events import EventMessage
+
+if TYPE_CHECKING:
+    from collections.abc import Iterator
+
+admin_events_bp = Blueprint("admin_events", __name__, url_prefix="/api/v1/admin/events")
+
+_HEARTBEAT_EVENT_TYPE = "heartbeat"
+_MAX_STREAM_LIFETIME_SECONDS = 30 * 60
+"""Hard cap so a single SSE connection never holds a worker forever -- `EventSource` reconnects."""
+
+_STAFF_ROLES = ("adviser", "admin")
+
+
+def _generate() -> Iterator[str]:
+    subscription = RedisEventBus(get_settings().redis_url).subscribe(ADVISER_CHANNEL)
+    started = time.monotonic()
+    try:
+        for message in subscription:
+            if message.get("event_type") == _HEARTBEAT_EVENT_TYPE:
+                yield ": ping\n\n"
+            else:
+                event = EventMessage.model_validate(dict(message))
+                yield f"data: {json.dumps(event.model_dump(mode='json'))}\n\n"
+            if time.monotonic() - started >= _MAX_STREAM_LIFETIME_SECONDS:
+                return
+    finally:
+        subscription.close()
+
+
+@admin_events_bp.route("/stream", methods=["GET"])
+@limiter.limit("10 per minute")
+@requires_role(*_STAFF_ROLES)
+def stream_admin_events() -> Any:
+    return Response(stream_with_context(_generate()), mimetype="text/event-stream")
+
+
+__all__ = ["admin_events_bp"]

backend/app/controllers/api/events.py

@@ -0,0 +1,74 @@
+"""Customer real-time push (S12 §6, NFR-18): a single self-scoped SSE stream over
+`customer:<id>:events`. A dropped/never-open connection is not data loss -- nothing published
+over Pub/Sub is ever the sole record of a fact; the next plain `GET` still returns truth.
+
+`_current_customer_id` reads `flask.session["_user_id"]`, not `current_user.id`, to avoid a
+`DetachedInstanceError` (`chat.py`/`profile.py`'s precedent).
+"""
+
+from __future__ import annotations
+
+import json
+import time
+import uuid
+from typing import TYPE_CHECKING, Any
+
+from flask import Blueprint, Response, stream_with_context
+from flask import session as flask_session
+from flask_login import current_user
+
+from app.config import get_settings
+from app.core.errors import ForbiddenError, UnauthenticatedError
+from app.extensions import limiter
+from app.integrations.redis.event_bus import RedisEventBus
+from app.services.ops.event_publisher import customer_channel
+from app.views.events import EventMessage
+
+if TYPE_CHECKING:
+    from collections.abc import Iterator
+
+events_bp = Blueprint("events", __name__, url_prefix="/api/v1/events")
+
+_HEARTBEAT_EVENT_TYPE = "heartbeat"
+_MAX_STREAM_LIFETIME_SECONDS = 30 * 60
+"""Hard cap so a single SSE connection never holds a worker forever -- `EventSource` reconnects."""
+
+
+def _current_customer_id() -> uuid.UUID:
+    if not current_user.is_authenticated:
+        raise UnauthenticatedError("Authentication required")
+    if current_user.role != "customer":
+        raise ForbiddenError("Real-time events are available to customer accounts only")
+    raw_user_id = flask_session.get("_user_id")
+    if not raw_user_id:
+        raise UnauthenticatedError("No authenticated session")
+    return uuid.UUID(raw_user_id)
+
+
+def _generate(channel: str) -> Iterator[str]:
+    subscription = RedisEventBus(get_settings().redis_url).subscribe(channel)
+    started = time.monotonic()
+    try:
+        for message in subscription:
+            if message.get("event_type") == _HEARTBEAT_EVENT_TYPE:
+                yield ": ping\n\n"
+            else:
+                event = EventMessage.model_validate(dict(message))
+                yield f"data: {json.dumps(event.model_dump(mode='json'))}\n\n"
+            if time.monotonic() - started >= _MAX_STREAM_LIFETIME_SECONDS:
+                return
+    finally:
+        subscription.close()
+
+
+@events_bp.route("/stream", methods=["GET"])
+@limiter.limit("10 per minute")
+def stream_events() -> Any:
+    customer_id = _current_customer_id()
+    return Response(
+        stream_with_context(_generate(customer_channel(str(customer_id)))),
+        mimetype="text/event-stream",
+    )
+
+
+__all__ = ["events_bp"]

backend/app/controllers/api/funding.py

@@ -7,7 +7,7 @@
 from __future__ import annotations
 
 import uuid
-from datetime import UTC, datetime
+from datetime import UTC, date, datetime
 from typing import Any
 
 from flask import Blueprint, jsonify, request
@@ -25,6 +25,7 @@
     resolve_replay,
 )
 from app.core.money import Money  # noqa: TC001 -- Pydantic needs the real type at class-build time.
+from app.core.pagination import decode_cursor, normalize_limit, paginate
 from app.core.uow import SessionRole
 from app.core.watermark import Watermark
 from app.extensions import DbRole, limiter
@@ -109,6 +110,34 @@ def _session_role_and_customer_id(
     return SessionRole.ADMIN, None
 
 
+def _parse_optional_limit(raw: str | None) -> int | None:
+    if raw is None:
+        return None
+    try:
+        return int(raw)
+    except ValueError as exc:
+        raise ValidationError("limit must be an integer") from exc
+
+
+def _decode_funding_history_cursor(raw: str) -> tuple[date, datetime, uuid.UUID]:
+    decoded = decode_cursor(raw)
+    if (
+        len(decoded) != 3
+        or not isinstance(decoded[0], str)
+        or not isinstance(decoded[1], str)
+        or not isinstance(decoded[2], str)
+    ):
+        raise ValidationError("invalid pagination cursor")
+    try:
+        return (
+            date.fromisoformat(decoded[0]),
+            datetime.fromisoformat(decoded[1]),
+            uuid.UUID(decoded[2]),
+        )
+    except ValueError as exc:
+        raise ValidationError("invalid pagination cursor") from exc
+
+
 def _plaid_adapter() -> PlaidBankAdapter:
     settings = get_settings()
     if settings.plaid_client_id is None or settings.plaid_secret is None:
@@ -191,9 +220,25 @@ def get_funding_history() -> Any:
     """Deposits and withdrawals, most-recent-first, with per-deposit settlement status (S2 §6)."""
     customer_id = _resolve_customer_id_for_get(request.args.get("customer_id"))
     role, uow_customer_id = _session_role_and_customer_id(customer_id)
+    limit = normalize_limit(_parse_optional_limit(request.args.get("limit")))
+    after: tuple[date, datetime, uuid.UUID] | None = None
+    raw_cursor = request.args.get("cursor")
+    if raw_cursor:
+        after = _decode_funding_history_cursor(raw_cursor)
 
     with FundingUnitOfWork(customer_id=uow_customer_id, role=role, db_role=DbRole.APP) as uow:
-        entries = FundingHistoryService(uow).history(customer_id, as_of=Watermark.live())
+        entries = FundingHistoryService(uow).history(
+            customer_id, as_of=Watermark.live(), limit=limit, after=after
+        )
+        page = paginate(
+            entries,
+            limit=limit,
+            cursor_key=lambda e: (
+                e.effective_date.isoformat(),
+                e.recorded_at.isoformat(),
+                str(e.journal_entry_id),
+            ),
+        )
         view = FundingHistoryResponse(
             entries=[
                 FundingHistoryEntryResponse(
@@ -208,8 +253,9 @@ def get_funding_history() -> Any:
                     expected_settlement_date=e.expected_settlement_date,
                     failure_reason=e.failure_reason,
                 )
-                for e in entries
-            ]
+                for e in page.items
+            ],
+            next_cursor=page.next_cursor,
         )
 
     return jsonify(view.model_dump(mode="json")), 200