fix(l1): write audit_logs rows at resolve/escalate with acting_as

Per spec §5.6.1, audit rows are written at session terminal events
(resolve, escalate, escalate_without_walk). log_audit gains an optional
acting_as parameter that propagates the session's acting_as tag
('l1_coverage' for engineer coverers, null for native L1 users).
Final code review flagged this as Important — column existed but was
never populated. Four new integration tests cover all three paths.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
2026-05-28 16:19:26 -04:00
parent 10b5d4e9b0
commit 7882b4723b
3 changed files with 198 additions and 3 deletions

View File

@@ -3,7 +3,10 @@ import uuid
import pytest
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from app.models.account import Account
from app.models.audit_log import AuditLog
from app.models.user import User
from app.models.tree import Tree
from app.models.ai_session import AISession
@@ -773,3 +776,142 @@ async def test_escalate_without_walk_reason_is_optional(test_db: AsyncSession):
)
assert session.escalation_reason is None
assert session.escalation_reason_category == "no_kb_content"
# ---------------------------------------------------------------------------
# T14 audit log tests (spec §5.6.1)
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_resolve_writes_audit_log_with_acting_as(test_db: AsyncSession):
"""resolve() writes an audit_logs row with acting_as='l1_coverage' for engineers."""
account = await _make_account(test_db)
eng = await _make_user(
test_db,
account_id=account.id,
account_role="engineer",
can_cover_l1=True,
)
ticket = await _make_internal_ticket(
test_db, account_id=account.id, user_id=eng.id
)
session = await start_adhoc_session(
test_db,
account_id=account.id,
user=eng,
ticket_id=str(ticket.id),
ticket_kind="internal",
)
await resolve(
test_db,
session_id=session.id,
helpful=True,
resolution_notes="Coverage engineer resolved",
)
result = await test_db.execute(
select(AuditLog).where(
AuditLog.action == "l1.session.resolve",
AuditLog.resource_id == session.id,
)
)
row = result.scalar_one()
assert row.acting_as == "l1_coverage"
assert row.user_id == eng.id
assert row.account_id == account.id
assert row.details["helpful"] is True
@pytest.mark.asyncio
async def test_resolve_writes_audit_log_native_l1_acting_as_null(
test_db: AsyncSession,
):
"""resolve() writes an audit_logs row with acting_as=None for native l1_tech."""
account = await _make_account(test_db)
l1 = await _make_user(test_db, account_id=account.id, account_role="l1_tech")
ticket = await _make_internal_ticket(
test_db, account_id=account.id, user_id=l1.id
)
session = await start_adhoc_session(
test_db,
account_id=account.id,
user=l1,
ticket_id=str(ticket.id),
ticket_kind="internal",
)
await resolve(
test_db,
session_id=session.id,
helpful=False,
resolution_notes="Native L1 resolved",
)
result = await test_db.execute(
select(AuditLog).where(
AuditLog.action == "l1.session.resolve",
AuditLog.resource_id == session.id,
)
)
row = result.scalar_one()
assert row.acting_as is None
@pytest.mark.asyncio
async def test_escalate_writes_audit_log(test_db: AsyncSession):
"""escalate() writes an audit_logs row with action='l1.session.escalate'."""
account = await _make_account(test_db)
l1 = await _make_user(test_db, account_id=account.id)
ticket = await _make_internal_ticket(
test_db, account_id=account.id, user_id=l1.id
)
session = await start_adhoc_session(
test_db,
account_id=account.id,
user=l1,
ticket_id=str(ticket.id),
ticket_kind="internal",
)
await escalate(
test_db,
session_id=session.id,
reason="Beyond scope",
reason_category="out_of_scope",
)
result = await test_db.execute(
select(AuditLog).where(
AuditLog.action == "l1.session.escalate",
AuditLog.resource_id == session.id,
)
)
row = result.scalar_one()
assert row.details["escalation_reason_category"] == "out_of_scope"
assert row.account_id == account.id
@pytest.mark.asyncio
async def test_escalate_without_walk_writes_audit_log(test_db: AsyncSession):
"""escalate_without_walk() writes an audit_logs row."""
account = await _make_account(test_db)
l1 = await _make_user(test_db, account_id=account.id)
ticket = await _make_internal_ticket(
test_db, account_id=account.id, user_id=l1.id
)
session = await escalate_without_walk(
test_db,
account_id=account.id,
user=l1,
ticket_id=str(ticket.id),
ticket_kind="internal",
reason_category="no_kb_content",
)
result = await test_db.execute(
select(AuditLog).where(
AuditLog.action == "l1.session.escalate_no_walk",
AuditLog.resource_id == session.id,
)
)
row = result.scalar_one()
assert row.account_id == account.id
assert row.details["escalation_reason_category"] == "no_kb_content"