fix(l1): drop duplicate T9 tests + honor explicit empty notify recipients

- Remove the weaker shadowing copies of the two T9 tests so the stronger
  originals (which seed an engineer and assert eng.id in target_user_ids,
  plus proposal_type/match_keywords) actually run.
- _resolve_recipients: treat an explicit empty target_user_ids as 'no
  recipients' instead of falling back to the default owner/admin set.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
2026-05-30 03:45:13 -04:00
parent e3da5b7502
commit cc41f20668
2 changed files with 7 additions and 67 deletions

View File

@@ -171,8 +171,13 @@ async def _resolve_recipients(
target_user_ids: Optional[list[uuid.UUID]],
db: AsyncSession,
) -> list[User]:
"""Resolve notification recipients. Defaults to team admins + account owners + admins."""
if target_user_ids:
"""Resolve notification recipients. Defaults to team admins + account owners + admins.
An explicit ``target_user_ids`` (even an empty list) means the caller has already
computed the recipient set — honor it exactly. Only ``None`` falls back to the
default owner/admin/team-admin set.
"""
if target_user_ids is not None:
result = await db.execute(
select(User)
.where(User.id.in_(target_user_ids))

View File

@@ -1073,68 +1073,3 @@ async def test_escalate_without_walk_writes_audit_log(test_db: AsyncSession):
)
row = result.scalar_one()
assert row.account_id == account.id
assert row.details["escalation_reason_category"] == "no_kb_content"
# ---------------------------------------------------------------------------
# T9: flywheel capture on resolve + engineer notification on escalate
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_resolve_ai_build_creates_outcome_validated_proposal(test_db: AsyncSession, monkeypatch):
"""resolve(helpful=True) on an ai_build session creates a FlowProposal with
source='ai_realtime_l1' and validated_by_outcome=True."""
from app.services import l1_session_service as svc
account = await _make_account(test_db)
l1_user = await _make_user(test_db, account_id=account.id)
ticket = await _make_internal_ticket(test_db, account_id=account.id, user_id=l1_user.id)
s = await svc.start_ai_build_session(
test_db, account_id=account.id, user=l1_user,
ticket_id=str(ticket.id), ticket_kind="internal")
# Simulate a walked path
s.walked_path = [
{"node_type": "question", "id": "n1", "text": "On?", "answer": "no"},
{"node_type": "resolved", "id": "n2", "text": "Fixed."},
]
await test_db.flush()
await svc.resolve(test_db, session_id=s.id, helpful=True, resolution_notes="ok")
props = (await test_db.execute(
select(FlowProposal).where(FlowProposal.l1_session_id == s.id))).scalars().all()
assert len(props) == 1
assert props[0].source == "ai_realtime_l1"
assert props[0].validated_by_outcome is True
assert props[0].source_session_id is None
assert props[0].proposed_flow_data["tree_structure"]["id"] == "n1"
@pytest.mark.asyncio
async def test_escalate_notifies_engineers(test_db: AsyncSession, monkeypatch):
"""escalate() sends a notification to engineer-or-above users in the account."""
from app.services import l1_session_service as svc
calls: dict = {}
async def fake_notify(event, account_id, payload, db, target_user_ids=None):
calls["event"] = event
calls["target_user_ids"] = target_user_ids
monkeypatch.setattr(svc, "notify", fake_notify)
account = await _make_account(test_db)
l1_user = await _make_user(test_db, account_id=account.id)
ticket = await _make_internal_ticket(test_db, account_id=account.id, user_id=l1_user.id)
s = await svc.start_ai_build_session(
test_db, account_id=account.id, user=l1_user,
ticket_id=str(ticket.id), ticket_kind="internal")
await svc.escalate(test_db, session_id=s.id, reason="stuck",
reason_category="exhausted_safe_steps")
assert calls["event"] == "l1.session.escalated"
assert calls["target_user_ids"] is not None # explicit engineer recipients