diff --git a/backend/tests/test_l1_session_service.py b/backend/tests/test_l1_session_service.py
index 217b30cc..12ca5e83 100644
--- a/backend/tests/test_l1_session_service.py
+++ b/backend/tests/test_l1_session_service.py
@@ -1074,3 +1074,67 @@ 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