feat(l1): flywheel capture on resolve + engineer notification on escalate

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

backend/tests/test_l1_session_service.py

@@ -869,6 +869,74 @@ async def test_advance_ai_build_wrong_session_kind_raises(test_db: AsyncSession,
             test_db, session_id=s.id, problem_text="printer", category="printer")
 
 
+# ---------------------------------------------------------------------------
+# 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 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",
+    )
+    # Populate walked_path with at least one node (needed for normalize_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"
+    assert props[0].proposal_type == "new_flow"
+    assert props[0].proposed_flow_data["match_keywords"] == []
+
+
+@pytest.mark.asyncio
+async def test_escalate_notifies_engineers(test_db: AsyncSession, monkeypatch):
+    """escalate() calls notify with event='l1.session.escalated' and explicit engineer recipients."""
+    from app.services import l1_session_service as svc
+    calls = {}
+
+    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 is the session owner (account_role="l1_tech" by default — NOT in the recipient query)
+    l1_user = await _make_user(test_db, account_id=account.id)
+    # Seed an eligible recipient: account_role="engineer" matches the production query
+    # (owner/admin/engineer). Without this user, target_ids would be [] and the
+    # eng.id assertion below would fail, proving the assertion is non-vacuous.
+    eng = await _make_user(test_db, account_id=account.id, account_role="engineer")
+    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 isinstance(calls["target_user_ids"], list) and len(calls["target_user_ids"]) >= 1
+    assert eng.id in calls["target_user_ids"]  # the eligible engineer is a recipient
+
+
 # ---------------------------------------------------------------------------
 # T14 audit log tests (spec §5.6.1)
 # ---------------------------------------------------------------------------