audit-mixin-rollout-tier3: budget + users (groups 3 & 4) — created_by/updated_by

openspec/changes/shared-feat-299-audit-mixin-rollout-tier3/design.md

@@ -23,8 +23,13 @@ Two structural wrinkles from the original investigation apply here:
 **1. Append-only models (`AIAuditLog`, `PrivilegedAccessLog` ×4) get `created_by` only — no update path is added, `updated_by` stays permanently `NULL`.**
 These models intentionally have no `update_*` CRUD function (audit/log integrity depends on immutability). Adding `updated_by` would be a column that's structurally impossible to ever populate — accept it as always-`NULL`, matching the semantics of "this row was never updated," rather than removing the column from the mixin (which would require a third mixin variant).
 
-**2. `UserProfileModel` is evaluated case-by-case, not auto-included.**
-It's a read-through cache populated by cross-service sync, not user action — `created_by`/`updated_by` may reflect "which sync process wrote this" rather than a meaningful actor, or may not apply at all. Decide during implementation (task 1 of tasks.md) whether to adopt `AuditColumnsMixin` or add it to the coverage guard's exemption list instead.
+**2. `UserProfileModel` is exempted — no `AuditColumnsMixin`, documented for `audit-mixin-coverage-guard`.**
+Two writers exist, and both are wrong for `created_by`/`updated_by`, for different reasons:
+- `event_handlers.py`'s `handle_user_created`/`updated`/`deleted` (RabbitMQ event-consumer callbacks) run with no FastAPI request in scope — no `Depends(get_validated_user)`, no `set_current_user_id()` call — so `created_by`/`updated_by` would be permanently `NULL` by construction here.
+- `user_cache.py`'s `get_users_by_ids_cached` (the live path, called from `budget_services.py::populate_budget_with_user_details` under an authenticated request) *would* have an actor in context — but it's the wrong one: it's caching *other* users' profiles (e.g. a budget's other collaborators) on behalf of whoever is viewing that budget, so `created_by` would end up recording the viewer, not the cached user or any actor who actually touched that profile. Populating would be actively misleading, not just absent.
+(A third function, `get_user_from_cache_or_fallback`, had the same misattribution problem and zero callers anywhere in the repo — removed as dead code rather than left to add a fourth data point.)
+
+`created_at`/`updated_at` are untouched (already correctly maintained via the model's own `onupdate`).
 
 **3. One Alembic migration per affected service, not a combined cross-service migration.**
 Each service has its own independent migration history; `ai` touches 5 models in one migration, `chat` touches 3, `budget` touches 2, `users` touches 1 — batched per-service since they're already grouped by ticket/PR boundary in tasks.md.
@@ -41,5 +46,6 @@ Standard Alembic migration per service, additive/nullable columns only. Deploy o
 
 ## Open Questions
 
-- Does `UserProfileModel` get audit columns, or does it become a documented exemption in `audit-mixin-coverage-guard`? Decide during task 1 (see tasks.md).
 - For each `PrivilegedAccessLog` copy, does an existing actor/subject field make `created_by` redundant, and if so should CRUD code assert they match rather than relying purely on the automatic listener?
+
+Resolved: `UserProfileModel` is exempted from `AuditColumnsMixin` — see Decision 2.

openspec/changes/shared-feat-299-audit-mixin-rollout-tier3/tasks.md

@@ -14,21 +14,21 @@ One task group = one GitHub ticket = one PR, merged before the next group starts
 - [x] 2.2 Update each model class to inherit `AuditMixin`.
 - [x] 2.3 Repeat the `PrivilegedAccessLog` actor-field check from task 1.3 for chat's copy.
 - [x] 2.4 Add/update tests confirming `created_by`/`updated_by` population for `Conversation`/`Message`, and `created_by`-only for `PrivilegedAccessLog`.
-- [ ] 2.5 Run `services/chat`'s test suite clean; PR merged.
+- [x] 2.5 Run `services/chat`'s test suite clean; PR merged.
 
-## 3. budget service — depends on 1
+## 3. budget service — depends on 1 — Issue #305
 
-- [ ] 3.1 Decide whether `UserProfileModel` (read-through cache, PK `user_id`) adopts `AuditColumnsMixin` or is documented as an intentional exemption for `audit-mixin-coverage-guard`; record the decision.
-- [ ] 3.2 Add Alembic migration adding nullable `created_by`/`updated_by` to `PrivilegedAccessLog` (and `UserProfileModel` if 3.1 decided to include it).
-- [ ] 3.3 Update model class(es) to inherit the appropriate mixin.
-- [ ] 3.4 Repeat the `PrivilegedAccessLog` actor-field check from task 1.3 for budget's copy.
-- [ ] 3.5 Add/update tests confirming the decided behavior.
-- [ ] 3.6 Run `services/budget`'s test suite clean; PR merged.
+- [x] 3.1 Decide whether `UserProfileModel` (read-through cache, PK `user_id`) adopts `AuditColumnsMixin` or is documented as an intentional exemption for `audit-mixin-coverage-guard`; record the decision. **Decided: exempt** — see design.md Decision 2. `event_handlers.py`'s RabbitMQ consumer has no request/actor context (permanently `NULL`); `user_cache.py::get_users_by_ids_cached`'s live request-context write would misattribute the row to the viewer, not the cached user. Removed `get_user_from_cache`/`get_user_from_cache_or_fallback` (dead code, same misattribution shape) and their now-orphaned `user_client.get_user`/`UserServiceError`.
+- [x] 3.2 Add Alembic migration adding nullable `created_by`/`updated_by` to `PrivilegedAccessLog`.
+- [x] 3.3 Update model class(es) to inherit the appropriate mixin.
+- [x] 3.4 Repeat the `PrivilegedAccessLog` actor-field check from task 1.3 for budget's copy.
+- [x] 3.5 Add/update tests confirming the decided behavior.
+- [x] 3.6 Run `services/budget`'s test suite clean; PR merged.
 
 ## 4. users service — depends on 1
 
-- [ ] 4.1 Add Alembic migration adding nullable `created_by`/`updated_by` to `PrivilegedAccessLog`.
-- [ ] 4.2 Update the model class to inherit `AuditMixin`.
-- [ ] 4.3 Repeat the `PrivilegedAccessLog` actor-field check from task 1.3 for users' copy.
-- [ ] 4.4 Add/update a test confirming `created_by` population, `updated_by` stays `NULL`.
-- [ ] 4.5 Run `services/users`'s test suite clean; PR merged.
+- [x] 4.1 Add Alembic migration adding nullable `created_by`/`updated_by` to `PrivilegedAccessLog`.
+- [x] 4.2 Update the model class to inherit `AuditMixin`.
+- [x] 4.3 Repeat the `PrivilegedAccessLog` actor-field check from task 1.3 for users' copy.
+- [x] 4.4 Add/update a test confirming `created_by` population, `updated_by` stays `NULL`.
+- [x] 4.5 Run `services/users`'s test suite clean; PR merged.

services/budget/app/models/privileged_access_log.py

@@ -1,21 +1,18 @@
-import uuid
 from datetime import datetime
 
 from sqlalchemy import DateTime, String
 from sqlalchemy.orm import Mapped, mapped_column
 
 from app.models.base import Base
+from shared.db.audit_mixin import AuditMixin
 import shared.db.type_decorators as t
 
 
-class PrivilegedAccessLog(Base):
+class PrivilegedAccessLog(Base, AuditMixin):
     """Append-only — no update/delete path exists anywhere in the app."""
 
     __tablename__ = "privileged_access_logs"
 
-    id: Mapped[t.GUID] = mapped_column(
-        t.GUID(), primary_key=True, default=lambda: str(uuid.uuid4())
-    )
     actor_user_id: Mapped[t.GUID] = mapped_column(t.GUID(), nullable=False, index=True)
     customer_id: Mapped[t.GUID] = mapped_column(t.GUID(), nullable=False, index=True)
     method: Mapped[str] = mapped_column(String, nullable=False)

services/budget/app/models/report.py

@@ -34,7 +34,7 @@ class ReportModel(Base, AuditMixin):
         GUID(),
         primary_key=True,
         index=True,
-        default=lambda: str(uuid.uuid4()),
+        default=lambda: uuid.uuid4(),
     )
     budget_id: Mapped[uuid.UUID] = mapped_column(GUID(), ForeignKey("budgets.id"), nullable=False)
     name: Mapped[str] = mapped_column(String, nullable=False)

services/budget/app/services/user_cache.py

@@ -1,91 +1,16 @@
-from typing import Dict, Any, Optional, List, cast
+from typing import Dict, Any, List, cast
 from uuid import UUID
 
 import structlog
 from sqlalchemy import select
 
 from app.db.session import AsyncSessionLocal
 from app.models.user_cache import UserProfileModel
-from app.services.user_client import get_user, get_users_by_ids
+from app.services.user_client import get_users_by_ids
 
 logger = structlog.get_logger(__name__)
 
 
-async def get_user_from_cache(user_id: UUID) -> Optional[Dict[str, Any]]:
-    async with AsyncSessionLocal() as session:
-        result = await session.execute(
-            select(UserProfileModel).where(UserProfileModel.user_id == user_id)
-        )
-        profile = result.scalar_one_or_none()
-
-        if not profile:
-            logger.debug("cache_miss", user_id=str(user_id))
-            return None
-
-        logger.debug("cache_hit", user_id=str(user_id))
-        return {
-            "id": str(profile.user_id),
-            "email": profile.email,
-            "first_name": profile.first_name,
-            "last_name": profile.last_name,
-            "status": profile.status,
-            "customer_id": str(profile.customer_id) if profile.customer_id else None,
-            "role": profile.role,
-        }
-
-
-async def get_user_from_cache_or_fallback(user_id: UUID, token: str) -> Optional[Dict[str, Any]]:
-    cached = await get_user_from_cache(user_id)
-    if cached:
-        return cached
-
-    logger.warning("cache_miss_fallback_attempted", user_id=str(user_id))
-
-    try:
-        user = get_user(str(user_id), token)
-
-        async with AsyncSessionLocal() as session:
-            result = await session.execute(
-                select(UserProfileModel).where(UserProfileModel.user_id == user_id)
-            )
-            profile = result.scalar_one_or_none()
-
-            if profile:
-                profile.email = cast(str, user.get("email"))
-                profile.first_name = user.get("first_name")
-                profile.last_name = user.get("last_name")
-                profile.status = cast(str, user.get("status"))
-                profile.customer_id = (
-                    UUID(user.get("customer_id")) if user.get("customer_id") else None
-                )
-                profile.role = cast(str, user.get("role"))
-            else:
-                profile = UserProfileModel(
-                    user_id=user_id,
-                    email=user.get("email"),
-                    first_name=user.get("first_name"),
-                    last_name=user.get("last_name"),
-                    status=user.get("status"),
-                    customer_id=(
-                        UUID(user.get("customer_id")) if user.get("customer_id") else None
-                    ),
-                    role=user.get("role"),
-                )
-                session.add(profile)
-
-            await session.commit()
-            logger.info("cache_populated_from_fallback", user_id=str(user_id))
-
-        return user
-    except Exception as e:
-        logger.error(
-            "fallback_http_failed",
-            user_id=str(user_id),
-            error=str(e),
-        )
-        raise
-
-
 async def get_users_by_ids_cached(ids: List[str], token: str) -> Dict[str, Dict[str, Any]]:
     """Get users from cache, falling back to HTTP for cache misses."""
     if not ids:

services/budget/app/services/user_client.py

@@ -1,4 +1,3 @@
-import requests
 from app.core.config import settings
 import uuid
 import httpx
@@ -25,24 +24,6 @@ async def close_urls():
         print("🛑 Users client closed")
 
 
-class UserServiceError(Exception):
-    pass
-
-
-def get_user(user_id: str | uuid.UUID, token: str) -> dict:
-    """
-    Fetch a user from the user service by ID.
-    """
-    try:
-        resp = requests.get(
-            f"{USER_SERVICE_URL}users/{user_id}", headers={"Authorization": f"Bearer {token}"}
-        )
-        resp.raise_for_status()
-        return resp.json()
-    except requests.RequestException as e:
-        raise UserServiceError(f"Failed to fetch user {user_id}") from e
-
-
 @service_call_exception_handler
 async def get_users_by_ids(ids: List[str | uuid.UUID], token: str) -> Dict[str, dict]:
     if not ids:

services/budget/migrations/versions/000015_tier3_audit_columns.py

@@ -0,0 +1,47 @@
+"""Add created_by/updated_by (audit-mixin-rollout-tier3, group 3)
+
+Revision ID: 000015
+Revises: 000014
+Create Date: 2026-09-21 00:00:00.000000
+
+"""
+
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+from shared.db.type_decorators import GUID
+
+revision: str = "000015"
+down_revision: Union[str, Sequence[str], None] = "000014"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+# table -> extra AuditMixin/AuditColumnsMixin columns it doesn't already have,
+# beyond created_by/updated_by (added to every table below).
+_EXTRA_COLUMNS = {
+    "privileged_access_logs": ["updated_at"],
+}
+
+
+def _column(name: str) -> sa.Column:
+    if name in ("created_at", "updated_at"):
+        return sa.Column(name, sa.DateTime(timezone=True), nullable=True)
+    return sa.Column(name, GUID(), nullable=True)
+
+
+def upgrade() -> None:
+    for table, extra in _EXTRA_COLUMNS.items():
+        op.add_column(table, _column("created_by"))
+        op.add_column(table, _column("updated_by"))
+        for name in extra:
+            op.add_column(table, _column(name))
+
+
+def downgrade() -> None:
+    for table, extra in _EXTRA_COLUMNS.items():
+        op.drop_column(table, "created_by")
+        op.drop_column(table, "updated_by")
+        for name in extra:
+            op.drop_column(table, name)

services/budget/tests/test_tier3_audit_columns.py

@@ -0,0 +1,62 @@
+"""audit-mixin-rollout-tier3 group 3: created_by/updated_by population for
+PrivilegedAccessLog; UserProfileModel stays exempt (design.md Decision 2)."""
+
+import uuid
+from datetime import datetime, timezone
+
+import pytest
+
+from app.models.privileged_access_log import PrivilegedAccessLog
+from app.models.user_cache import UserProfileModel
+from shared.security.current_user_context import reset_current_user_id, set_current_user_id
+
+
+def _now():
+    return datetime.now(timezone.utc)
+
+
+class TestPrivilegedAccessLogAppendOnly:
+    @pytest.mark.anyio
+    async def test_created_by_matches_existing_actor_field(self, db):
+        """actor_user_id already captures the acting user (task 1.3) — the
+        automatically-populated created_by should equal it, not diverge."""
+        actor_id = uuid.uuid4()
+        token = set_current_user_id(actor_id)
+        try:
+            log = PrivilegedAccessLog(
+                actor_user_id=str(actor_id),
+                customer_id=str(uuid.uuid4()),
+                method="PUT",
+                path="/api/v1/budget",
+                created_at=_now(),
+            )
+            db.add(log)
+            await db.commit()
+        finally:
+            reset_current_user_id(token)
+
+        assert log.created_by == actor_id
+        assert str(log.actor_user_id) == str(actor_id)
+
+    @pytest.mark.anyio
+    async def test_updated_by_stays_null_with_no_update_path(self, db):
+        token = set_current_user_id(uuid.uuid4())
+        try:
+            log = PrivilegedAccessLog(
+                actor_user_id=str(uuid.uuid4()),
+                customer_id=str(uuid.uuid4()),
+                method="GET",
+                path="/api/v1/budget",
+                created_at=_now(),
+            )
+            db.add(log)
+            await db.commit()
+        finally:
+            reset_current_user_id(token)
+
+        assert log.updated_by is None
+
+
+def test_user_profile_model_stays_exempt():
+    assert not hasattr(UserProfileModel, "created_by")
+    assert not hasattr(UserProfileModel, "updated_by")

services/users/app/models/privileged_access_log.py

@@ -1,19 +1,18 @@
-import uuid
 from datetime import datetime
 
 from sqlalchemy import DateTime, String
 from sqlalchemy.orm import Mapped, mapped_column
 
 from app.models.base import Base
+from shared.db.audit_mixin import AuditMixin
 import shared.db.type_decorators as t
 
 
-class PrivilegedAccessLog(Base):
+class PrivilegedAccessLog(Base, AuditMixin):
     """Append-only — no update/delete path exists anywhere in the app."""
 
     __tablename__ = "privileged_access_logs"
 
-    id: Mapped[t.GUID] = mapped_column(t.GUID(), primary_key=True, default=lambda: uuid.uuid4())
     actor_user_id: Mapped[t.GUID] = mapped_column(t.GUID(), nullable=False, index=True)
     customer_id: Mapped[t.GUID] = mapped_column(t.GUID(), nullable=False, index=True)
     method: Mapped[str] = mapped_column(String, nullable=False)

services/users/migrations/versions/000014_tier3_audit_columns.py

@@ -0,0 +1,47 @@
+"""Add created_by/updated_by (audit-mixin-rollout-tier3, group 4)
+
+Revision ID: 000014
+Revises: 000013
+Create Date: 2026-09-21 00:00:00.000000
+
+"""
+
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+from shared.db.type_decorators import GUID
+
+revision: str = "000014"
+down_revision: Union[str, Sequence[str], None] = "000013"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+# table -> extra AuditMixin/AuditColumnsMixin columns it doesn't already have,
+# beyond created_by/updated_by (added to every table below).
+_EXTRA_COLUMNS = {
+    "privileged_access_logs": ["updated_at"],
+}
+
+
+def _column(name: str) -> sa.Column:
+    if name in ("created_at", "updated_at"):
+        return sa.Column(name, sa.DateTime(timezone=True), nullable=True)
+    return sa.Column(name, GUID(), nullable=True)
+
+
+def upgrade() -> None:
+    for table, extra in _EXTRA_COLUMNS.items():
+        op.add_column(table, _column("created_by"))
+        op.add_column(table, _column("updated_by"))
+        for name in extra:
+            op.add_column(table, _column(name))
+
+
+def downgrade() -> None:
+    for table, extra in _EXTRA_COLUMNS.items():
+        op.drop_column(table, "created_by")
+        op.drop_column(table, "updated_by")
+        for name in extra:
+            op.drop_column(table, name)