fix(l1): add missing next-node + escalations routes; reconcile Phase-1 intake tests

An earlier anchor-edit silently failed, so POST /sessions/{id}/next-node and
GET /escalations were never added (they 404'd). Add both, anchored on the real
/escalate-without-walk route.

Phase-1 test_l1_endpoints tests used POST /intake to create adhoc setup sessions,
but Phase 2A intake now dispatches via match_or_build (build/matched/suggest/
out_of_scope — never adhoc). Add a _create_adhoc_session service helper and route
the step/notes/resolve/escalate/cross-account setup through it; rewrite
test_intake_adhoc as test_intake_build_creates_ai_build_session (mocked outcome).

All green: test_l1_endpoints + test_l1_api_ai_build = 25 passed; full Phase 2A
backend service/unit/model suite = 56 passed; notification suite = 18 passed.

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

backend/app/api/endpoints/l1.py

@@ -284,6 +284,68 @@ async def post_escalate(
     return _to_response(updated)
 
 
+@router.post("/sessions/{session_id}/next-node", response_model=NextNodeResponse)
+async def next_node(
+    session_id: UUID,
+    payload: NextNodeRequest,
+    db: Annotated[AsyncSession, Depends(get_db)],
+    user: Annotated[User, Depends(require_l1_or_coverage)],
+):
+    """Record the answer/ack on the current node, then generate the next node.
+
+    problem_text comes from the linked internal ticket; category from the hidden
+    meta entry seeded at intake (ai_tree_builder skips meta entries). node_text is
+    the rendered text of the node being answered (the client holds it) so the
+    walked path and the captured tree stay legible.
+    """
+    session = await _get_session_or_404(db, session_id, user)
+    ticket = await internal_ticket_service.get_ticket(
+        db, ticket_id=UUID(session.ticket_id)
+    )
+    problem_text = ticket.problem_statement if ticket else ""
+    category = next(
+        (s.get("category") for s in (session.walked_path or [])
+         if s.get("node_type") == "meta"),
+        "unknown",
+    )
+    try:
+        node = await l1_session_service.advance_ai_build(
+            db,
+            session_id=session_id,
+            problem_text=problem_text,
+            category=category or "unknown",
+            node_id=payload.node_id,
+            node_text=payload.node_text,
+            answer=payload.answer,
+            note=payload.note,
+        )
+    except ValueError as exc:
+        raise HTTPException(
+            status_code=http_status.HTTP_409_CONFLICT, detail=str(exc)
+        )
+    await db.commit()
+    return NextNodeResponse(node=node, session_status=session.status)
+
+
+@router.get("/escalations", response_model=list[WalkSessionResponse])
+async def l1_escalations(
+    db: Annotated[AsyncSession, Depends(get_db)],
+    user: Annotated[User, Depends(require_engineer_or_admin)],
+    limit: int = 50,
+):
+    """Engineer-visible list of escalated L1 sessions (the handoff queue)."""
+    rows = await db.execute(
+        select(L1WalkSession)
+        .where(
+            L1WalkSession.account_id == user.account_id,
+            L1WalkSession.status == "escalated",
+        )
+        .order_by(L1WalkSession.last_step_at.desc())
+        .limit(limit)
+    )
+    return [_to_response(s) for s in rows.scalars()]
+
+
 @router.post("/escalate-without-walk", response_model=WalkSessionResponse)
 async def post_escalate_without_walk(
     payload: EscalateWithoutWalkRequest,

backend/tests/test_l1_endpoints.py

@@ -82,27 +82,72 @@ async def _make_l1_user(
         return {"user_data": {"id": str(user.id), "account_id": str(account_id)}, "headers": None}
 
 
+async def _create_adhoc_session(db: AsyncSession, info: dict, *, problem: str = "setup") -> str:
+    """Create an adhoc walk session (backed by a real internal ticket) via the service.
+
+    Phase 2A: POST /l1/intake dispatches through match_or_build and no longer
+    yields an adhoc session directly, so step/notes/resolve/escalate/cross-account
+    tests build their setup session here instead of through intake. The test
+    client shares this same DB session (conftest override_get_db), so the
+    committed session is visible to the API immediately.
+    """
+    from sqlalchemy import select as sa_select
+    from app.services import internal_ticket_service, l1_session_service
+
+    account_id = info["account_id"]
+    user_id = uuid.UUID(info["user_data"]["id"])
+    user = (await db.execute(sa_select(User).where(User.id == user_id))).scalar_one()
+    ticket = await internal_ticket_service.create_ticket(
+        db,
+        account_id=account_id,
+        created_by_user_id=user_id,
+        problem_statement=problem,
+        customer_name=None,
+        customer_contact=None,
+    )
+    session = await l1_session_service.start_adhoc_session(
+        db,
+        account_id=account_id,
+        user=user,
+        ticket_id=str(ticket.id),
+        ticket_kind="internal",
+    )
+    await db.commit()
+    return str(session.id)
+
+
 # ---------------------------------------------------------------------------
-# 1. Intake without flow_id → 200 + session_kind='adhoc'
+# 1. Intake (Phase 2A): build outcome → 200 + session_kind='ai_build'
 # ---------------------------------------------------------------------------
 
 @pytest.mark.asyncio
-async def test_intake_adhoc(client: AsyncClient, test_db: AsyncSession):
+async def test_intake_build_creates_ai_build_session(client: AsyncClient, test_db: AsyncSession):
-    """POST /l1/intake without flow_id creates adhoc session."""
+    """POST /l1/intake with a 'build' outcome creates an ai_build session.
+
+    Phase 2A: intake dispatches via match_or_build; 'adhoc' is no longer a direct
+    intake outcome (it is offered from the out_of_scope prompt on the frontend).
+    """
+    from unittest.mock import AsyncMock, patch
     info = await _make_l1_user(client, test_db, email="l1intake@example.com")
     headers = info["headers"]
 
-    resp = await client.post(
+    with patch(
-        "/api/v1/l1/intake",
+        "app.api.endpoints.l1.match_or_build.match_or_build",
-        json={"problem_statement": "Printer won't turn on", "customer_name": "Alice"},
+        new=AsyncMock(return_value={"outcome": "build", "session_kind": "ai_build",
-        headers=headers,
+                                    "category": "printer"}),
-    )
+    ):
+        resp = await client.post(
+            "/api/v1/l1/intake",
+            json={"problem_statement": "Printer won't turn on", "customer_name": "Alice"},
+            headers=headers,
+        )
     assert resp.status_code == 200, resp.text
     body = resp.json()
-    assert body["session_kind"] == "adhoc"
+    assert body["outcome"] == "build"
+    assert body["session_kind"] == "ai_build"
     assert body["ticket_kind"] == "internal"
-    assert "session_id" in body
+    assert body["session_id"]
-    assert "ticket_id" in body
+    assert body["ticket_id"]
 
 
 # ---------------------------------------------------------------------------
@@ -156,14 +201,7 @@ async def test_step_on_adhoc_returns_400(client: AsyncClient, test_db: AsyncSess
     info = await _make_l1_user(client, test_db, email="l1step@example.com")
     headers = info["headers"]
 
-    # Create adhoc session via intake
+    session_id = await _create_adhoc_session(test_db, info, problem="Adhoc issue")
-    resp = await client.post(
-        "/api/v1/l1/intake",
-        json={"problem_statement": "Adhoc issue"},
-        headers=headers,
-    )
-    assert resp.status_code == 200, resp.text
-    session_id = resp.json()["session_id"]
 
     resp = await client.post(
         f"/api/v1/l1/sessions/{session_id}/step",
@@ -184,13 +222,7 @@ async def test_notes_on_adhoc_session(client: AsyncClient, test_db: AsyncSession
     info = await _make_l1_user(client, test_db, email="l1notes@example.com")
     headers = info["headers"]
 
-    resp = await client.post(
+    session_id = await _create_adhoc_session(test_db, info, problem="Notes test")
-        "/api/v1/l1/intake",
-        json={"problem_statement": "Notes test"},
-        headers=headers,
-    )
-    assert resp.status_code == 200, resp.text
-    session_id = resp.json()["session_id"]
 
     notes_payload = [{"text": "Customer called about printer", "ts": "2026-05-28T10:00:00Z"}]
     resp = await client.post(
@@ -213,13 +245,7 @@ async def test_resolve_session(client: AsyncClient, test_db: AsyncSession):
     info = await _make_l1_user(client, test_db, email="l1resolve@example.com")
     headers = info["headers"]
 
-    resp = await client.post(
+    session_id = await _create_adhoc_session(test_db, info, problem="Resolve test")
-        "/api/v1/l1/intake",
-        json={"problem_statement": "Resolve test"},
-        headers=headers,
-    )
-    assert resp.status_code == 200, resp.text
-    session_id = resp.json()["session_id"]
 
     resp = await client.post(
         f"/api/v1/l1/sessions/{session_id}/resolve",
@@ -245,13 +271,7 @@ async def test_escalate_session(client: AsyncClient, test_db: AsyncSession):
     info = await _make_l1_user(client, test_db, email="l1escalate@example.com")
     headers = info["headers"]
 
-    resp = await client.post(
+    session_id = await _create_adhoc_session(test_db, info, problem="Escalation test")
-        "/api/v1/l1/intake",
-        json={"problem_statement": "Escalation test"},
-        headers=headers,
-    )
-    assert resp.status_code == 200, resp.text
-    session_id = resp.json()["session_id"]
 
     resp = await client.post(
         f"/api/v1/l1/sessions/{session_id}/escalate",
@@ -344,15 +364,8 @@ async def test_get_session_cross_account_returns_404(client: AsyncClient, test_d
     """GET /l1/sessions/{id} from a different account → 404."""
     # Account A: creates a session
     info_a = await _make_l1_user(client, test_db, email="l1tenanta@example.com")
-    headers_a = info_a["headers"]
 
-    resp = await client.post(
+    session_id = await _create_adhoc_session(test_db, info_a, problem="Account A issue")
-        "/api/v1/l1/intake",
-        json={"problem_statement": "Account A issue"},
-        headers=headers_a,
-    )
-    assert resp.status_code == 200, resp.text
-    session_id = resp.json()["session_id"]
 
     # Account B: different user in a different account
     info_b = await _make_l1_user(client, test_db, email="l1tenantb@example.com")