feat(l1): proposal L1-source block + engineer L1-escalations section

- flow-proposal.ts: source_session_id nullable + add l1_session_id (matches backend
  FlowProposalSummary).
- ProposalDetail.tsx: render an 'AI L1 walk (outcome-validated)' note when
  l1_session_id is set instead of the /pilot/{source_session_id} link; fall back to
  the link for ai_session-sourced proposals.
- New L1EscalationsSection.tsx (GET /l1/escalations) — expandable rows with walked-path
  summary; renders nothing if empty. Mounted below the FlowPilot queue on
  EscalationQueuePage. tsc -b + eslint clean.

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

.ai/HANDOFF.md

@@ -2,67 +2,105 @@
 
 # HANDOFF.md
 
-**Last updated:** 2026-05-14
+**Last updated:** 2026-05-30
 
-**Active task:** Phase O cutover for self-serve signup. All code blockers remain closed on `main`. **Still blocked on Stripe live-mode activation — root cause is EIN, not code.** User does not yet have an EIN for ResolutionFlow, LLC; Stripe requires a tax ID for live-mode activation. EIN application via IRS.gov was scheduled for 2026-05-13 — confirm status at next session start. Mailing-address decision (carried forward from 2026-05-12): user enters home address into Stripe's **private** business profile temporarily so live-mode isn't blocked on the P.O. Box; public `ContactPage`/`PoliciesPage` mailing-address TODOs stay "available on request" until the P.O. Box is purchased. Stripe accepts an address update later without re-verification. Apex DNS at Namecheap is still missing (separate user-side issue, only matters once Stripe runs site-verification). Nothing on the code side blocks live-mode flip.
+**Active task:** Executing the **L1 AI Tree Builder Phase 2A** plan
+(`docs/superpowers/plans/2026-05-29-l1-ai-tree-builder-phase-2a.md`, 19 tasks) via
+subagent-driven-development on branch `feat/l1-ai-tree-builder-phase-2a`
+(branched from `main` @ `87236b5`; **not pushed**, `main` untouched).
 
-**Bug-pending-capture item (2026-05-12) — likely resolved:** Prior session noted "user reported finding a bug, will send screenshot next session." This session surfaced two concrete UX bugs that were fixed and merged (PR #168): the dashboard "Start a session" CTA was a dead link, and welcome step-2's PSA setup had a near-invisible "Connect now →" link that didn't even persist `primary_psa`. **Confirm with user at next session start whether the screenshot bug was one of these or something else still pending.**
+## ⚠️ Tooling note (read first — why this session stopped at Task 16)
+The harness's **Bash output channel became intermittently unreliable** — returning
+stale/cached output (a Bash command that wrote `/tmp/perm.txt` instead returned a
+PRIOR command's `/tmp/vc.txt` content; a `cat` returned the wrong commit SHA). The
+Write/Edit channel stayed reliable; Read mostly reliable but occasionally served a
+stale temp file. Work stopped at Task 16 because wiring a new route/nav requires
+accurately reading `router.tsx` + `AccountSettingsPage.tsx` then editing them, and
+read-then-edit against stale reads is exactly what produced the broken Tasks 14–15
+earlier this session. **On resume: confirm the shell is reliable first** — write a
+unique sentinel to a file and read it back; cross-check any Read against a fresh
+`grep`; never commit without a sentinel-wrapped `tsc -b`/pytest verification whose
+unique sentinel you can see in the same output.
 
-## Where this session ended
+Earlier-this-session gotcha that cost ~an hour: pytest `-p no:cov` conflicts with the
+`--cov` baked into `pytest.ini` addopts → pytest exits before running → `&& echo PASS`
+chains mislabel. Always use `--override-ini="addopts="`, never `-p no:cov`.
 
-Two PRs merged into main:
+Backend test invocation that works:
+`docker exec resolutionflow_backend pytest <path> --override-ini="addopts=" -q`
+Do **NOT** use `-p no:cov` — `pytest.ini` bakes `--cov` into `addopts`; disabling the
+cov plugin makes `--cov` unrecognized so pytest exits before running, silently turning
+`&& echo PASS || echo FAIL` chains into false FAILs (this cost ~an hour of confusion).
+Frontend gate via file-redirect:
+`docker exec -w /app resolutionflow_frontend sh -c 'npx tsc -b > /app/_o.txt 2>&1; echo EXIT=$? >> /app/_o.txt'`
+then Read `frontend/_o.txt` (frontend is bind-mounted at /app).
 
-- **PR #166** (`fe0e692`) — docs/handoff doc updates from prior session. Squash-merged 2026-05-14.
-- **PR #168** (`3a35121`) — session expiration policy + dashboard NextStep CTA fix + welcome step-2 PSA CTA reshape. Merge-committed 2026-05-14. Three notable additions:
-  - `feat(dashboard)` `8d79dd9` — The "Start a session" CTAs on NextStepCard and SetupChecklist used to `Link`-navigate to `/`, leaving the user on the same page (the StartSessionInput lives on the dashboard) with no visible response. Replaced with a `FOCUS_START_SESSION_EVENT` window event the StartSessionInput listens for: scrolls input to viewport top (`scrollIntoView({block:'start'})`), focuses the textarea (with `preventScroll:true` so it doesn't fight the smooth scroll), pulses a `rgba(96,165,250,…)` ring for 900ms. NextStepCard hides itself via local `locallyHidden` state on click so the user isn't double-prompted while typing. SetupChecklist gets the same event-dispatch treatment for its `ran_session` row.
-  - `feat(welcome)` `dc88797` — Welcome step-2 PSA CTA reshaped. Selecting a real PSA now swaps the single Continue + tiny "Connect now →" link for an explicit two-button choice: `Connect <PSA> now` (primary, blue — saves `primary_psa` then routes to `/account/integrations`) and `Connect later` (secondary outlined — saves `primary_psa` then continues to step 3). **Important pre-existing bug fixed**: the old subtle Link never actually persisted `primary_psa` before navigating away. Both new buttons do. "No PSA yet" and no-selection states still show the original single Continue. Skip-this-step and Skip-the-rest unchanged. Existing tests pass without edits (testids `welcome-step-2-connect-now` and `welcome-step-2-continue` reused).
-  - `docs:` `e5b2624` — added `docs/plans/2026-05-13-public-landing-routing-refactor.md`, `docs/architecture/` reports (god-node map + report 2026-05-06, workflows.json/html, workflows-analysis.html), `docs/tutorials/build-a-page.md`, and `abc-feat-self-serve-signup-phase-2-design-20260507-112020.md` at repo root.
+## Status: Tasks 1–15 DONE & committed. Tasks 16–19 remain (all frontend + final).
 
-`tsc --project tsconfig.app.json --noEmit` clean across all changes. Local vitest blocked by root-owned `node_modules/.vite-temp` (same env issue noted in prior handoffs); CI ran the suite green.
+**Backend (Tasks 1–12)** — 17 commits `16b9abf`…`04b5511` + handoff `fdac72e`.
+Last full run: **114 passed** across all 11 Phase 2A backend test files. 3 alembic
+migrations applied; head `1fd88a68b145`. Shipped: `ai_build` session kind;
+`accounts.enabled_l1_categories`; `FlowProposal.l1_session_id` (+ nullable
+source_session_id + exactly-one CHECK + schema made optional); `l1_category_service`;
+`ai_tree_builder` (constrained gen, validate, depth cap, `normalize_walked_path`,
+**skips `meta` entries**); `match_or_build` (bands; flow_id→str); session-service
+`start_ai_build_session`/`advance_ai_build` (stores `node_text`)/flywheel capture in
+`resolve`/engineer notify in `escalate`; `l1.session.escalated` notification (+ link
+`/escalations` + `_resolve_recipients` honors explicit empty list); API
+`/l1/intake` (dispatch; build seeds hidden `{"node_type":"meta","category":...}`
+walked_path entry), `POST /l1/sessions/{id}/next-node`, `GET /l1/escalations`,
+`GET|PATCH /accounts/me/l1-categories`, `require_account_owner_or_admin` dep.
 
-**Two issues filed for session leftovers:**
+**Frontend (Tasks 13–15) — committed; whole-project `tsc -b` + eslint clean. VERIFIED HEAD `076a9ec`, tree clean.**
+- `03e8748` Task 13 — `types/l1.ts` (+ai_build, IntakeOutcome/Result, NearMiss, TreeNode,
+  NextNodeRequest/Result, L1Categories) + `api/l1.ts` (intake→IntakeResult; nextNode,
+  escalations, getCategories, setCategories). nextNode body carries `node_text`.
+- Tasks 14/15 took THREE commits because the flaky shell caused two broken commits
+  (`df7150f`, `f483196` had missing-export/props errors; `ad9c4c8` was committed with
+  TSC_EXIT=2 because I batched the commit with its own failing verification). The REAL
+  working fix is **`076a9ec`** — confirmed via single-value commands: committed
+  `L1WalkTreeVariant.tsx` has `advanceNode` (grep -c = 3), committed `L1Dashboard.tsx`
+  has `useSuggestedFlow` (= 2); and a sentinel-wrapped `npx tsc -b` returned TSC=0,
+  eslint=0 on the on-disk files before commit. What landed:
+  - `L1Dashboard.tsx`: outcome dispatch on the REAL page (matched/build→walker;
+    suggest→use-flow/build-new; out_of_scope→escalate-without-walk). Original
+    PageMeta/greeting/inputs/open-tickets layout preserved.
+  - `L1WalkTreeVariant.tsx`: real props `{session,onSessionUpdate,onDone}` +
+    ResolveModal/EscalateModal + header + transcript sidebar kept; added ai_build branch
+    that walks nodes via /next-node (passes node_text), disclaimer banner (`bg-warning/10`
+    — NOTE: `*-dim` tokens are NOT `--color-*-dim`; use `/10` opacity), terminal→modals.
+    flow/proposal keep the Phase-1 synthetic path.
+  - `L1WalkPage.tsx` unchanged (already routes ai_build → tree variant).
+  NOT browser-verified (chromium can't launch here).
+- **SHELL DISCIPLINE for resume:** single-value Bash commands (`grep -c`, `wc -l`,
+  `git rev-parse --short`, `git log -1 --format=%s`) are RELIABLE; multi-line
+  `{ echo; … } > file` blocks get GARBLED/interleaved. NEVER batch a commit with its
+  own verification — verify in a separate step and READ the result before committing.
 
-- **Issue #171** — Test coverage for the new welcome step-2 "Connect now" path (existing tests still pass but don't exercise the new button's save + redirect-to-integrations behavior).
-- **Issue #172** — Repo hygiene: gitignore `core.[0-9]*` + `**/.remember/`, and delete the existing 20MB core dumps (`core.144926`, `core.145678`, `docs/architecture/core.1392564`) and `docs/architecture/.remember/`. Carried forward across multiple sessions.
+## Resume point — Tasks 16–19
 
-Working tree clean except those persistent untracked items (intentionally left for issue #172).
+16. **`pages/account/L1CategoriesPage.tsx`** (does NOT exist yet) — checkbox list of
+    `available` toggling `enabled` via `l1Api.getCategories/setCategories`; read-only
+    hard-floor list. Register lazy route under the `account` children in `router.tsx`
+    (the L1CategoriesPage import is NOT yet there — verify) and add a link card in
+    `AccountSettingsPage.tsx` (AccountLayout has no sidebar nav — see CLAUDE.md
+    "Account sub-page"). Gate visibility to owner/admin via `usePermissions`.
+17. **`ProposalDetail.tsx`** — branch on `l1_session_id` to show an L1-source block
+    instead of the `/pilot/{source_session_id}` link (add `l1_session_id?: string|null`
+    to its proposal type). **`EscalationQueuePage.tsx`** — add an "L1 escalations"
+    section via `l1Api.escalations()`.
+18. **`frontend/e2e/l1-workspace.spec.ts`** — network-stubbed AI-build flow; rely on CI
+    to run it (chromium can't launch here).
+19. **Final:** full backend suite + `tsc -b`/`npm run lint`/`npm run build`; migration
+    downgrade/upgrade roundtrip (head `1fd88a68b145`, down 3); push branch + open PR to
+    `main` listing deferred items (KB grounding/connectors, PSA reassign, escalation
+    package, AI chat handoff, proposal-matching). Then run requesting-code-review +
+    finishing-a-development-branch per the subagent-driven-development skill.
 
-Single alembic head: `4ce3e594cb87` (no schema changes this session).
+**Working tree:** clean except this HANDOFF.md edit (committing now). Temp `_*.txt`
+files under `frontend/` were scratch — delete any that remain.
 
-## Resume point
-
-**First thing next session:**
-
-1. Confirm with user whether the "bug-pending-capture" screenshot bug from 2026-05-12 was one of the two PR #168 fixes or something else still pending.
-2. Check EIN application status (filed 2026-05-13 via IRS.gov). If granted, unblocks the Phase O Stripe live-mode setup chain.
-
-After that — **Phase O manual ops, all user-side, all gated on EIN landing first:**
-
-1. **EIN application status check** (user, applied 2026-05-13).
-2. **Stripe Dashboard live-mode** (once EIN is in hand):
-   - 3 Products (Starter, Pro, Enterprise). Monthly Prices for Starter ($19.99) + Pro ($29.99). No Prices on Enterprise (sales-led).
-   - Customer Portal with plan-switching disabled.
-   - Webhook at `https://api.resolutionflow.com/api/v1/webhooks/stripe` with 5 events. Save live signing secret.
-   - **Business profile fields**: Customer service URL `https://resolutionflow.com/contact`. Refund/cancellation policy URL `https://resolutionflow.com/policies`. Terms `https://resolutionflow.com/terms`. Privacy `https://resolutionflow.com/privacy`. Phone `(470) 949-4131`. Mailing address = user's home address temporarily (private Stripe field; swap to P.O. Box later without re-verification). EIN = the newly-issued tax ID.
-3. **Apex DNS fix at Namecheap** (re-add `@` ALIAS → `c9g7uku8.up.railway.app`, or re-add apex as a Railway custom domain). Becomes the next blocker once Stripe runs site-verification.
-4. **Railway prod env**: `STRIPE_SECRET_KEY=sk_live_...`, `STRIPE_WEBHOOK_SECRET`, `STRIPE_PUBLISHABLE_KEY` + `VITE_STRIPE_PUBLISHABLE_KEY` (frontend redeploy required — Vite bake-at-build, Lesson 60), `OAUTH_REDIRECT_BASE=https://resolutionflow.com`, `SELF_SERVE_ENABLED=false` (still false at this point), `INTERNAL_TESTER_EMAILS=<allowlist>`, prod Google + Microsoft OAuth credentials.
-5. **Bootstrap prod super-admin** via `create_site_admin.py` (PR #167) — already done end-to-end on prod per 2026-05-12 user confirmation. Re-runnable if needed.
-6. **Sync Stripe → DB**: `railway run python -m scripts.sync_stripe_plan_ids` (or via `railway ssh`). Verify `plan_billing` rows have `sk_live_*` price IDs.
-7. **Internal validation (Phase O Task 46)**: 9 scenarios with internal testers whose emails match `INTERNAL_TESTER_EMAILS`.
-8. **Flag flip (Task 47)**: email pilots, set `SELF_SERVE_ENABLED=true` + `VITE_SELF_SERVE_ENABLED=true` (frontend redeploy). PostHog signup-funnel dashboard + Sentry alert at >1/hour Stripe webhook errors.
-
-## Open issues from prior session (non-code, user-side)
-
-- **Apex DNS missing.** `resolutionflow.com` (apex) returns no A/CNAME at the authoritative DNS (Namecheap). When `www` was reconfigured in Railway, the apex record got dropped from the zone. `www` works (cert provisioned 2026-05-08 01:40 UTC). User to re-add apex record at Namecheap (ALIAS `@` → `c9g7uku8.up.railway.app`) or re-add the apex as a Railway custom domain. Railway path is more durable.
-- **Edge HSTS sticky state on user's machine.** Browser remembers the earlier broken-cert visit. Fix: `edge://net-internals/#hsts` (delete `resolutionflow.com` and `www.resolutionflow.com`) + `#dns` clear host cache + `#sockets` flush.
-
-## Carry-forward
-
-- Annual pricing intentionally NOT implemented — user wants exit flexibility. Schema columns preserved as nullable. `sync_stripe_plan_ids.py` leaves annual fields NULL.
-- `INTERNAL_TESTER_EMAILS` parsed comma-separated → normalized lowercase list. Anonymous callers always see the global flag — allowlist never leaks via unauthenticated request content (regression test enforces).
-- Office-hours design doc now at `docs/` root (`abc-feat-self-serve-signup-phase-2-design-20260507-112020.md`) as of this session. NOT yet adopted as roadmap — gated on 3 cold calls with external Directors of Onboarding.
-- Mailing address fill-in: search for `TODO: replace with full mailing address` in `frontend/src/pages/ContactPage.tsx` and `frontend/src/pages/PoliciesPage.tsx` (one each) once P.O. Box is purchased.
-- `backend/scripts/create_site_admin.py` is the durable site-admin bootstrap tool — idempotent. Three modes: `--send-reset`, `--print-reset`, `--promote-only`. Run from inside the deployed backend container via `railway ssh`.
-- Bot-crawlability of legal pages: still SPA-rendered. Stripe didn't enforce content scraping last time (issue was DNS). If a future vendor review flags it, pre-render with `vite-plugin-prerender-spa` (~half day).
-- Frontend env additions for cutover: `VITE_SELF_SERVE_ENABLED`, `VITE_GOOGLE_CLIENT_ID`, `VITE_MS_CLIENT_ID`, `VITE_OAUTH_REDIRECT_BASE`, `VITE_CALENDLY_URL`, `VITE_STRIPE_PUBLISHABLE_KEY`.
-- **Branch hygiene note (process learning):** PR #168 ended up bundling unrelated work — session expiration policy (the original scope of `feat/session-expiration-policy`) plus dashboard CTA fixes plus welcome step-2 reshape. The mixed scope was deliberate (user wanted it on the same PR), but worth flagging for future PRs: if onboarding-UX work continues, branch it separately from auth/session work.
+## Carry-forward (Phase O — separate, user-side, gated on EIN)
+Phase O self-serve cutover (Stripe live-mode, apex DNS, Railway prod env, flag flip)
+remains the prior active task; all code blockers closed, blocked on user's EIN. Not
+touched this session.

.gitea/workflows/mirror-to-github.yml

@@ -15,5 +15,8 @@ jobs:
           git clone --mirror https://gitea.resolutionflow.com/chihlasm/resolutionflow.git repo
           cd repo
           git remote add github https://x-access-token:${{ secrets.GH_MIRROR_TOKEN }}@github.com/${{ secrets.GH_MIRROR_REPO }}
-          git push github --all --force
-          git push github --tags --force
+          # --all + --tags scopes the push to refs/heads/* and refs/tags/*,
+          # avoiding refs/pull/* (which GitHub refuses with "deny updating a
+          # hidden ref"). --prune makes deletions on the Gitea side propagate.
+          git push github --all --prune --force
+          git push github --tags --prune --force

.gitignore

@@ -249,3 +249,6 @@ graphify-out/
 
 # remember skill runtime state (hook logs, PIDs)
 .remember/
+
+# MCP server config (per-machine, references local env vars for auth)
+.mcp.json

backend/alembic/versions/1fd88a68b145_flow_proposal_l1_source_linkage.py

@@ -0,0 +1,61 @@
+"""flow_proposal l1 source linkage
+
+Revision ID: 1fd88a68b145
+Revises: cb9e282267d2
+Create Date: 2026-05-29 19:33:09.188681
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.dialects import postgresql
+
+
+# revision identifiers, used by Alembic.
+revision: str = '1fd88a68b145'
+down_revision: Union[str, None] = 'cb9e282267d2'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        "flow_proposals",
+        sa.Column("l1_session_id", postgresql.UUID(as_uuid=True), nullable=True),
+    )
+    op.create_index(
+        "ix_flow_proposals_l1_session_id",
+        "flow_proposals",
+        ["l1_session_id"],
+    )
+    op.create_foreign_key(
+        "fk_flow_proposals_l1_session_id",
+        "flow_proposals",
+        "l1_walk_sessions",
+        ["l1_session_id"],
+        ["id"],
+        ondelete="SET NULL",
+    )
+    op.alter_column("flow_proposals", "source_session_id", nullable=True)
+    op.create_check_constraint(
+        "ck_flow_proposals_exactly_one_source",
+        "flow_proposals",
+        "(source_session_id IS NOT NULL) <> (l1_session_id IS NOT NULL)",
+    )
+
+
+def downgrade() -> None:
+    op.drop_constraint(
+        "ck_flow_proposals_exactly_one_source",
+        "flow_proposals",
+        type_="check",
+    )
+    op.alter_column("flow_proposals", "source_session_id", nullable=False)
+    op.drop_constraint(
+        "fk_flow_proposals_l1_session_id",
+        "flow_proposals",
+        type_="foreignkey",
+    )
+    op.drop_index("ix_flow_proposals_l1_session_id", "flow_proposals")
+    op.drop_column("flow_proposals", "l1_session_id")

backend/alembic/versions/beca7464b6b4_add_ai_build_session_kind.py

@@ -0,0 +1,48 @@
+"""add ai_build session kind
+
+Revision ID: beca7464b6b4
+Revises: b3358ba0e48c
+Create Date: 2026-05-29 18:41:38.601537
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = 'beca7464b6b4'
+down_revision: Union[str, None] = 'b3358ba0e48c'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.drop_constraint("ck_l1_walk_sessions_session_kind", "l1_walk_sessions", type_="check")
+    op.create_check_constraint(
+        "ck_l1_walk_sessions_session_kind", "l1_walk_sessions",
+        "session_kind IN ('flow', 'proposal', 'adhoc', 'ai_build')",
+    )
+    op.drop_constraint("ck_l1_walk_sessions_target_consistency", "l1_walk_sessions", type_="check")
+    op.create_check_constraint(
+        "ck_l1_walk_sessions_target_consistency", "l1_walk_sessions",
+        "(session_kind = 'flow' AND flow_id IS NOT NULL AND flow_proposal_id IS NULL) "
+        "OR (session_kind = 'proposal' AND flow_proposal_id IS NOT NULL AND flow_id IS NULL) "
+        "OR (session_kind IN ('adhoc', 'ai_build') AND flow_id IS NULL AND flow_proposal_id IS NULL)",
+    )
+
+
+def downgrade() -> None:
+    op.drop_constraint("ck_l1_walk_sessions_target_consistency", "l1_walk_sessions", type_="check")
+    op.create_check_constraint(
+        "ck_l1_walk_sessions_target_consistency", "l1_walk_sessions",
+        "(session_kind = 'flow' AND flow_id IS NOT NULL AND flow_proposal_id IS NULL) "
+        "OR (session_kind = 'proposal' AND flow_proposal_id IS NOT NULL AND flow_id IS NULL) "
+        "OR (session_kind = 'adhoc' AND flow_id IS NULL AND flow_proposal_id IS NULL)",
+    )
+    op.drop_constraint("ck_l1_walk_sessions_session_kind", "l1_walk_sessions", type_="check")
+    op.create_check_constraint(
+        "ck_l1_walk_sessions_session_kind", "l1_walk_sessions",
+        "session_kind IN ('flow', 'proposal', 'adhoc')",
+    )

backend/alembic/versions/cb9e282267d2_add_enabled_l1_categories_to_accounts.py

@@ -0,0 +1,35 @@
+"""add enabled_l1_categories to accounts
+
+Revision ID: cb9e282267d2
+Revises: beca7464b6b4
+Create Date: 2026-05-29 18:48:27.155183
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.dialects import postgresql
+
+
+# revision identifiers, used by Alembic.
+revision: str = 'cb9e282267d2'
+down_revision: Union[str, None] = 'beca7464b6b4'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+_DEFAULT = ('["password_reset","account_lockout","printer","email_outlook_client",'
+            '"wifi_network_basics","vpn_connect","teams_zoom_av","browser_cache_cookies",'
+            '"peripheral_reconnect","os_restart_update"]')
+
+
+def upgrade() -> None:
+    op.add_column("accounts", sa.Column(
+        "enabled_l1_categories", postgresql.JSONB(), nullable=False,
+        server_default=sa.text(f"'{_DEFAULT}'::jsonb"),
+    ))
+
+
+def downgrade() -> None:
+    op.drop_column("accounts", "enabled_l1_categories")

backend/app/api/deps.py

@@ -276,6 +276,20 @@ async def require_account_owner(
     )
 
 
+async def require_account_owner_or_admin(
+    current_user: Annotated[User, Depends(get_current_active_user)]
+) -> User:
+    """Require account owner or account-admin (blocks engineers); super_admin bypass."""
+    if current_user.is_super_admin:
+        return current_user
+    if current_user.account_role in ("owner", "admin"):
+        return current_user
+    raise HTTPException(
+        status_code=status.HTTP_403_FORBIDDEN,
+        detail="Account owner or admin access required",
+    )
+
+
 def get_service_account_id(request: Request) -> Optional[UUID]:
     """Return the cached ResolutionFlow service account UUID from app.state.
 

backend/app/api/endpoints/accounts.py

@@ -23,9 +23,17 @@ from app.schemas.account import AccountResponse, AccountUpdate, AccountInviteCre
 from app.schemas.subscription import SubscriptionResponse, PlanLimitsResponse, UsageResponse, SubscriptionDetails
 from app.schemas.user import UserResponse, AccountRoleUpdate, CoverageUpdate
 from app.core.security import verify_password
-from app.api.deps import get_current_active_user, require_account_owner, require_engineer_or_admin
+from app.api.deps import (
+    get_current_active_user,
+    require_account_owner,
+    require_account_owner_or_admin,
+    require_engineer_or_admin,
+    require_l1_or_above,
+)
+from app.services import l1_category_service
 from app.services.seat_enforcement import check_seat_available, get_seat_usage
 from app.schemas.seat_enforcement import SeatUsage
+from app.schemas.l1_categories import L1CategoriesResponse, L1CategoriesUpdate
 
 _SEAT_CHECKED_ROLES = frozenset({"engineer", "l1_tech"})
 
@@ -164,6 +172,45 @@ async def get_my_account_seat_usage(
     return SeatUsage(engineer=engineer, l1_tech=l1_tech)
 
 
+@router.get("/me/l1-categories", response_model=L1CategoriesResponse)
+async def get_l1_categories(
+    db: Annotated[AsyncSession, Depends(get_db)],
+    current_user: Annotated[User, Depends(require_l1_or_above)],
+):
+    """The account's enabled L1 AI-build categories + the available + hard-floor lists.
+
+    Readable by any L1-or-above user (the walker needs to know what's buildable);
+    only owners/admins may change it (PATCH below).
+    """
+    enabled = await l1_category_service.get_enabled_categories(current_user.account_id, db)
+    return L1CategoriesResponse(
+        enabled=enabled,
+        available=l1_category_service.DEFAULT_L1_CATEGORIES,
+        hard_floor=l1_category_service.HARD_FLOOR_FORBIDDEN,
+    )
+
+
+@router.patch("/me/l1-categories", response_model=L1CategoriesResponse)
+async def set_l1_categories(
+    payload: L1CategoriesUpdate,
+    db: Annotated[AsyncSession, Depends(get_db)],
+    current_user: Annotated[User, Depends(require_account_owner_or_admin)],
+):
+    """Set the account's enabled L1 categories (owner/admin only).
+
+    Unknown and hard-floored keys are dropped by the service before persisting.
+    """
+    enabled = await l1_category_service.set_enabled_categories(
+        current_user.account_id, payload.enabled, db
+    )
+    await db.commit()
+    return L1CategoriesResponse(
+        enabled=enabled,
+        available=l1_category_service.DEFAULT_L1_CATEGORIES,
+        hard_floor=l1_category_service.HARD_FLOOR_FORBIDDEN,
+    )
+
+
 @router.patch("/me", response_model=AccountResponse)
 async def update_my_account(
     data: AccountUpdate,

backend/app/api/endpoints/l1.py

@@ -9,7 +9,7 @@ from fastapi import APIRouter, Depends, HTTPException, status as http_status
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from app.api.deps import get_db, require_l1_or_coverage
+from app.api.deps import get_db, require_engineer_or_admin, require_l1_or_coverage
 from app.models.l1_walk_session import L1WalkSession
 from app.models.user import User
 from app.schemas.l1 import (
@@ -17,13 +17,15 @@ from app.schemas.l1 import (
     EscalateWithoutWalkRequest,
     IntakeRequest,
     IntakeResponse,
+    NextNodeRequest,
+    NextNodeResponse,
     NotesRequest,
     QueueRow,
     ResolveRequest,
     StepRequest,
     WalkSessionResponse,
 )
-from app.services import internal_ticket_service, l1_session_service
+from app.services import internal_ticket_service, l1_session_service, match_or_build
 
 
 router = APIRouter(prefix="/l1", tags=["l1"])
@@ -72,11 +74,34 @@ async def intake(
     db: Annotated[AsyncSession, Depends(get_db)],
     user: Annotated[User, Depends(require_l1_or_coverage)],
 ):
-    """L1 intake: creates an internal ticket and starts a walk session.
+    """L1 intake (Phase 2A): match a published flow, else gate + build.
 
-    Phase 1: internal-ticket only (PSA support follows in Phase 2 escalation polish).
-    If `flow_id` is provided, starts a flow session; otherwise an adhoc session.
+    Runs the match_or_build orchestrator. Outcomes:
+    - matched  → create ticket + flow session, walk the published flow.
+    - build    → create ticket + ai_build session (category persisted as a hidden
+                 meta entry on walked_path for /next-node), walk an AI-built tree.
+    - suggest  → near-miss prompt; no session created.
+    - out_of_scope → category disabled/unknown; no session created.
     """
+    result = await match_or_build.match_or_build(
+        user.account_id,
+        payload.problem_statement,
+        None,
+        ticket_ref="",
+        db=db,
+        force_build=payload.force_build,
+    )
+    outcome = result["outcome"]
+
+    if outcome in ("suggest", "out_of_scope"):
+        await db.commit()
+        return IntakeResponse(
+            outcome=outcome,
+            near_miss=result.get("near_miss"),
+            category=result.get("category"),
+        )
+
+    # matched OR build → create a ticket and a session
     ticket = await internal_ticket_service.create_ticket(
         db,
         account_id=user.account_id,
@@ -85,29 +110,38 @@ async def intake(
         customer_name=payload.customer_name,
         customer_contact=payload.customer_contact,
     )
-    if payload.flow_id is not None:
+    if outcome == "matched":
         session = await l1_session_service.start_flow_session(
             db,
             account_id=user.account_id,
             user=user,
-            flow_id=payload.flow_id,
+            flow_id=UUID(result["flow_id"]),
             ticket_id=str(ticket.id),
             ticket_kind="internal",
         )
-    else:
-        session = await l1_session_service.start_adhoc_session(
+    else:  # build
+        session = await l1_session_service.start_ai_build_session(
             db,
             account_id=user.account_id,
             user=user,
             ticket_id=str(ticket.id),
             ticket_kind="internal",
         )
+        # Persist the classified category as a hidden meta entry so /next-node
+        # can recover it (no dedicated column; ai_tree_builder skips meta entries).
+        session.walked_path = [
+            {"node_type": "meta", "category": result.get("category", "unknown")}
+        ]
+        await db.flush()
+
     await db.commit()
     return IntakeResponse(
+        outcome=outcome,
         session_id=session.id,
         session_kind=session.session_kind,
         ticket_id=str(ticket.id),
         ticket_kind="internal",
+        flow_id=UUID(result["flow_id"]) if outcome == "matched" else None,
     )
 
 
@@ -250,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/app/core/config.py

@@ -211,6 +211,10 @@ class Settings(BaseSettings):
         # concrete rendered script so a draft_template can be proposed.
         # Creates a persistent library artifact on accept, so Sonnet.
         "template_extraction": "standard",
+        # L1 AI tree builder (Phase 2A): per-node generation is latency-sensitive
+        # on a live call → Sonnet; classification is a short label task → Haiku.
+        "l1_realtime_build": "standard",
+        "l1_classify": "fast",
     }
 
     def get_model_for_action(self, action_type: str) -> str:

backend/app/models/account.py

@@ -1,7 +1,7 @@
 import uuid
 from datetime import datetime, timezone
 from typing import Optional, TYPE_CHECKING
-from sqlalchemy import String, DateTime, ForeignKey, Boolean, Integer
+from sqlalchemy import String, DateTime, ForeignKey, Boolean, Integer, text as sa_text
 from sqlalchemy.orm import Mapped, mapped_column, relationship
 from sqlalchemy.dialects.postgresql import UUID, JSONB
 from app.core.database import Base
@@ -67,6 +67,19 @@ class Account(Base):
     sso_provider: Mapped[Optional[str]] = mapped_column(String(20), nullable=True)  # "saml" | "oidc"
     sso_config: Mapped[Optional[dict]] = mapped_column(JSONB, nullable=True)
 
+    # L1 AI tree builder — per-account allowlist of problem categories.
+    # Keep this server_default in sync with DEFAULT_L1_CATEGORIES in
+    # app/services/l1_category_service.py when adding/removing categories.
+    enabled_l1_categories: Mapped[list[str]] = mapped_column(
+        JSONB(), nullable=False,
+        server_default=sa_text(
+            "'[\"password_reset\",\"account_lockout\",\"printer\","
+            "\"email_outlook_client\",\"wifi_network_basics\",\"vpn_connect\","
+            "\"teams_zoom_av\",\"browser_cache_cookies\",\"peripheral_reconnect\","
+            "\"os_restart_update\"]'::jsonb"
+        ),
+    )
+
     # Relationships
     owner: Mapped["User"] = relationship("User", foreign_keys=[owner_id], back_populates="owned_account")
     users: Mapped[list["User"]] = relationship("User", foreign_keys="[User.account_id]", back_populates="account")

backend/app/models/flow_proposal.py

@@ -19,6 +19,7 @@ if TYPE_CHECKING:
     from app.models.account import Account
     from app.models.tree import Tree
     from app.models.ai_session import AISession
+    from app.models.l1_walk_session import L1WalkSession
 
 
 class FlowProposal(Base):
@@ -56,6 +57,10 @@ class FlowProposal(Base):
             "linked_ticket_kind IS NULL OR linked_ticket_kind IN ('psa', 'internal')",
             name="ck_flow_proposals_linked_ticket_kind",
         ),
+        CheckConstraint(
+            "(source_session_id IS NOT NULL) <> (l1_session_id IS NOT NULL)",
+            name="ck_flow_proposals_exactly_one_source",
+        ),
     )
 
     id: Mapped[uuid.UUID] = mapped_column(
@@ -73,10 +78,16 @@ class FlowProposal(Base):
         nullable=True,
         index=True,
     )
-    source_session_id: Mapped[uuid.UUID] = mapped_column(
+    source_session_id: Mapped[Optional[uuid.UUID]] = mapped_column(
         UUID(as_uuid=True),
         ForeignKey("ai_sessions.id", ondelete="CASCADE"),
-        nullable=False,
+        nullable=True,
+        index=True,
+    )
+    l1_session_id: Mapped[Optional[uuid.UUID]] = mapped_column(
+        UUID(as_uuid=True),
+        ForeignKey("l1_walk_sessions.id", ondelete="SET NULL"),
+        nullable=True,
         index=True,
     )
 
@@ -164,7 +175,17 @@ class FlowProposal(Base):
     # ── Relationships ──
     account: Mapped["Account"] = relationship("Account")
     team: Mapped[Optional["Team"]] = relationship("Team")
-    source_session: Mapped["AISession"] = relationship("AISession")
-    target_flow: Mapped[Optional["Tree"]] = relationship("Tree", foreign_keys=[target_flow_id])
-    published_flow: Mapped[Optional["Tree"]] = relationship("Tree", foreign_keys=[published_flow_id])
+    source_session: Mapped[Optional["AISession"]] = relationship("AISession")
+    # Two FK paths exist between FlowProposal and L1WalkSession
+    # (FlowProposal.l1_session_id here, L1WalkSession.flow_proposal_id there),
+    # so each relationship must name its foreign_keys explicitly.
+    l1_session: Mapped[Optional["L1WalkSession"]] = relationship(
+        "L1WalkSession", foreign_keys="[FlowProposal.l1_session_id]"
+    )
+    target_flow: Mapped[Optional["Tree"]] = relationship(
+        "Tree", foreign_keys=[target_flow_id]
+    )
+    published_flow: Mapped[Optional["Tree"]] = relationship(
+        "Tree", foreign_keys=[published_flow_id]
+    )
     reviewer: Mapped[Optional["User"]] = relationship("User")

backend/app/models/l1_walk_session.py

@@ -30,6 +30,7 @@ class L1WalkSession(Base):
     - flow: Walking a published flow (flow_id required, flow_proposal_id null).
     - proposal: Walking a draft flow proposal (flow_proposal_id required, flow_id null).
     - adhoc: Free-form investigation (both flow_id and flow_proposal_id null).
+    - ai_build: AI-generated decision-tree walk (both flow_id and flow_proposal_id null).
 
     status lifecycle:
     - active: Session is in progress.
@@ -45,7 +46,7 @@ class L1WalkSession(Base):
             name="ck_l1_walk_sessions_ticket_kind",
         ),
         CheckConstraint(
-            "session_kind IN ('flow', 'proposal', 'adhoc')",
+            "session_kind IN ('flow', 'proposal', 'adhoc', 'ai_build')",
             name="ck_l1_walk_sessions_session_kind",
         ),
         CheckConstraint(
@@ -55,7 +56,7 @@ class L1WalkSession(Base):
         CheckConstraint(
             "(session_kind = 'flow' AND flow_id IS NOT NULL AND flow_proposal_id IS NULL) "
             "OR (session_kind = 'proposal' AND flow_proposal_id IS NOT NULL AND flow_id IS NULL) "
-            "OR (session_kind = 'adhoc' AND flow_id IS NULL AND flow_proposal_id IS NULL)",
+            "OR (session_kind IN ('adhoc', 'ai_build') AND flow_id IS NULL AND flow_proposal_id IS NULL)",
             name="ck_l1_walk_sessions_target_consistency",
         ),
     )
@@ -138,4 +139,9 @@ class L1WalkSession(Base):
     account: Mapped["Account"] = relationship("Account")
     created_by: Mapped["User"] = relationship("User", foreign_keys=[created_by_user_id])
     flow: Mapped[Optional["Tree"]] = relationship("Tree")
-    flow_proposal: Mapped[Optional["FlowProposal"]] = relationship("FlowProposal")
+    # Two FK paths exist between L1WalkSession and FlowProposal
+    # (L1WalkSession.flow_proposal_id here, FlowProposal.l1_session_id there),
+    # so each relationship must name its foreign_keys explicitly.
+    flow_proposal: Mapped[Optional["FlowProposal"]] = relationship(
+        "FlowProposal", foreign_keys="[L1WalkSession.flow_proposal_id]"
+    )

backend/app/schemas/flow_proposal.py

@@ -19,7 +19,10 @@ class FlowProposalSummary(BaseModel):
     supporting_session_count: int
     status: str
     target_flow_id: UUID | None = None
-    source_session_id: UUID
+    # Exactly one source is set: source_session_id (FlowPilot ai_session) XOR
+    # l1_session_id (L1 ai_build walk). Both are nullable on the model.
+    source_session_id: UUID | None = None
+    l1_session_id: UUID | None = None
     created_at: datetime
 
     model_config = {"from_attributes": True}

backend/app/schemas/l1.py

@@ -11,13 +11,31 @@ class IntakeRequest(BaseModel):
     customer_name: Optional[str] = None
     customer_contact: Optional[str] = None
     flow_id: Optional[UUID] = None
+    force_build: bool = False
 
 
 class IntakeResponse(BaseModel):
-    session_id: UUID
-    session_kind: Literal["flow", "proposal", "adhoc"]
-    ticket_id: str
-    ticket_kind: Literal["psa", "internal"]
+    outcome: Literal["matched", "suggest", "out_of_scope", "build"]
+    session_id: Optional[UUID] = None
+    session_kind: Optional[Literal["flow", "proposal", "adhoc", "ai_build"]] = None
+    ticket_id: Optional[str] = None
+    ticket_kind: Optional[str] = None
+    flow_id: Optional[UUID] = None   # for 'matched'
+    near_miss: Optional[dict] = None  # for 'suggest'
+    category: Optional[str] = None   # for 'out_of_scope'
+
+
+class NextNodeRequest(BaseModel):
+    node_id: Optional[str] = None
+    node_text: Optional[str] = None  # rendered text of the node being answered (carry-forward Task 8)
+    answer: Optional[str] = None     # 'yes' | 'no' for questions
+    acknowledged: Optional[bool] = None
+    note: Optional[str] = None
+
+
+class NextNodeResponse(BaseModel):
+    node: dict
+    session_status: str
 
 
 class StepRequest(BaseModel):

backend/app/schemas/l1_categories.py

@@ -0,0 +1,14 @@
+"""Schemas for the account L1 AI-build category settings surface (Phase 2A)."""
+from pydantic import BaseModel
+
+
+class L1CategoriesResponse(BaseModel):
+    """Current enabled set + the full available list + the read-only hard floor."""
+    enabled: list[str]
+    available: list[str]
+    hard_floor: list[str]
+
+
+class L1CategoriesUpdate(BaseModel):
+    """Owner/admin write: the new enabled set (unknown/hard-floored keys dropped)."""
+    enabled: list[str]

backend/app/schemas/notification.py

@@ -11,6 +11,7 @@ VALID_EVENTS = {
     "proposal.pending",
     "proposal.approved",
     "knowledge_gap.detected",
+    "l1.session.escalated",
 }
 
 

backend/app/services/ai_tree_builder.py

@@ -0,0 +1,167 @@
+"""Constrained, node-by-node L1 decision-tree generation (spec §4/§5/§6.1).
+
+Each call produces ONE node given the problem, category, and full walked path.
+Generation is constrained to safe/reversible L1 steps and biased to escalate
+early. normalize_walked_path() turns a resolved walk into a valid tree object
+for flywheel capture.
+"""
+import logging
+from typing import Any, Optional
+
+from app.core.ai_provider import get_ai_provider
+from app.core.config import settings
+from app.services.l1_category_service import HARD_FLOOR_TEXT_PATTERNS
+from app.services.llm_utils import parse_llm_json
+
+logger = logging.getLogger(__name__)
+
+MAX_DEPTH = 12
+VALID_NODE_TYPES = {"question", "instruction", "resolved", "escalate"}
+
+
+class UnsafeNodeError(ValueError):
+    """Raised when a generated node violates the hard floor or is malformed."""
+
+
+SYSTEM_PROMPT = """\
+You are an L1 helpdesk troubleshooting guide builder. Given a problem and the
+steps already tried, produce the SINGLE next node of a yes/no decision tree.
+
+HARD RULES:
+- Only safe, reversible, observe-or-restart-class steps: checking status, toggling,
+  restarting, reconnecting, re-entering credentials the USER already knows.
+- NEVER produce steps that: edit the registry/system files/boot config; delete or
+  format data/disks; change credentials/MFA/security/firewall/AV; run elevated or
+  admin scripts; touch domain controllers/DNS/DHCP or production servers; or have
+  billing/license impact. These are out of L1 scope.
+- When you run out of safe in-scope steps, DO NOT GUESS. Emit an "escalate" node.
+
+Return ONLY a JSON object for ONE node, one of:
+{"node_type":"question","text":"<yes/no question>"}
+{"node_type":"instruction","text":"<one safe reversible action>"}
+{"node_type":"resolved","text":"<confirmation the issue is fixed>"}
+{"node_type":"escalate","reason_category":"exhausted_safe_steps","text":"<why>"}
+No prose, no markdown fences.
+"""
+
+
+def _strip_meta(walked_path: list[dict]) -> list[dict]:
+    """Drop the hidden ``meta`` entry (category carrier) the intake endpoint seeds.
+
+    The first walked_path entry on an ai_build session may be a
+    ``{"node_type": "meta", "category": ...}`` marker used to persist the
+    classified category; it is not a real walk step and must be excluded from
+    both model context and tree normalization.
+    """
+    return [s for s in walked_path if s.get("node_type") != "meta"]
+
+
+def _build_context(problem_text: str, category: str, walked_path: list[dict]) -> str:
+    walked_path = _strip_meta(walked_path)
+    lines = [f"PROBLEM: {problem_text}", f"CATEGORY: {category}", "STEPS SO FAR:"]
+    if not walked_path:
+        lines.append("(none yet — produce the first diagnostic question)")
+    for i, step in enumerate(walked_path, 1):
+        ans = step.get("answer")
+        suffix = f" -> {ans}" if ans else ""
+        lines.append(f"{i}. [{step.get('node_type','?')}] {step.get('text','')}{suffix}")
+    return "\n".join(lines)
+
+
+def validate_node(node: dict[str, Any]) -> dict[str, Any]:
+    """Shape + hard-floor validation. Raises UnsafeNodeError on violation."""
+    if not isinstance(node, dict) or node.get("node_type") not in VALID_NODE_TYPES:
+        raise UnsafeNodeError(f"invalid node_type: {node!r}")
+    text = (node.get("text") or "").lower()
+    for pat in HARD_FLOOR_TEXT_PATTERNS:
+        if pat in text:
+            raise UnsafeNodeError(f"hard-floor pattern '{pat}' in node text")
+    return node
+
+
+def escalate_if_depth_exceeded(walked_path: list[dict]) -> Optional[dict[str, Any]]:
+    if len(walked_path) >= MAX_DEPTH:
+        return {
+            "node_type": "escalate",
+            "reason_category": "depth_cap",
+            "text": "Reached the L1 troubleshooting depth limit — escalating to engineering.",
+        }
+    return None
+
+
+async def generate_next_node(
+    problem_text: str, category: str, walked_path: list[dict]
+) -> dict[str, Any]:
+    """Generate + validate the next node. Regenerate once on failure, then escalate."""
+    capped = escalate_if_depth_exceeded(walked_path)
+    if capped:
+        return capped
+
+    provider = get_ai_provider(settings.get_model_for_action("l1_realtime_build"))
+    context = _build_context(problem_text, category, walked_path)
+
+    for attempt in range(2):
+        try:
+            raw, _, _ = await provider.generate_json(
+                system_prompt=SYSTEM_PROMPT,
+                messages=[{"role": "user", "content": context}],
+                max_tokens=1024,
+            )
+            node = parse_llm_json(raw)
+            return validate_node(node)
+        except Exception as e:
+            logger.warning("ai_tree_builder node attempt %d failed: %s", attempt + 1, e)
+            continue
+
+    return {
+        "node_type": "escalate",
+        "reason_category": "generation_failed",
+        "text": "Could not generate a safe next step — escalating to engineering.",
+    }
+
+
+def normalize_walked_path(walked_path: list[dict]) -> dict[str, Any]:
+    """Turn a resolved walk into a valid troubleshooting tree (spec §6.1).
+
+    Root = first node's id; question nodes' traversed branch points to the next
+    node, the untraversed branch to a needs_review stub; terminal node ends it.
+    Returns {id, nodes: {id: node}} — a dict with an id (passes the proposal
+    approval guard).
+    """
+    walked_path = _strip_meta(walked_path)
+    nodes: dict[str, Any] = {}
+    if not walked_path:
+        root_id = "root"
+        nodes[root_id] = {"id": root_id, "node_type": "needs_review",
+                          "text": "Empty walk — needs authoring."}
+        return {"id": root_id, "nodes": nodes}
+
+    stub_seq = 0
+    for i, step in enumerate(walked_path):
+        nid = step.get("id") or f"n{i+1}"
+        ntype = step.get("node_type", "question")
+        nxt = walked_path[i + 1].get("id", f"n{i+2}") if i + 1 < len(walked_path) else None
+        node: dict[str, Any] = {"id": nid, "node_type": ntype, "text": step.get("text", "")}
+        if step.get("reason_category"):
+            node["reason_category"] = step["reason_category"]
+        if ntype == "question":
+            answer = (step.get("answer") or "").lower()
+            stub_seq += 1
+            stub_id = f"review-{stub_seq}"
+            nodes[stub_id] = {"id": stub_id, "node_type": "needs_review",
+                              "text": "Branch not explored during the originating call."}
+            traversed_next = nxt
+            if traversed_next is None:
+                # Walk ended on this question (no terminal recorded) — stub the
+                # branch the tech actually took so the tree has no dangling edge.
+                stub_seq += 1
+                traversed_next = f"review-{stub_seq}"
+                nodes[traversed_next] = {"id": traversed_next, "node_type": "needs_review",
+                                         "text": "Walk ended here before a terminal step was reached."}
+            node["yes_next"] = traversed_next if answer == "yes" else stub_id
+            node["no_next"] = traversed_next if answer == "no" else stub_id
+        elif ntype == "instruction":
+            node["next"] = nxt
+        nodes[nid] = node
+
+    return {"id": walked_path[0].get("id", "n1"), "nodes": nodes}

backend/app/services/l1_category_service.py

@@ -0,0 +1,69 @@
+"""L1 category allowlist + the always-forbidden hard floor.
+
+DEFAULT_L1_CATEGORIES seeds an account's enabled set. HARD_FLOOR_FORBIDDEN is a
+category-independent safety floor the AI tree builder must never emit and admins
+cannot enable. See spec §5.1/§5.2.
+"""
+from uuid import UUID
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.models.account import Account
+
+# WARNING: keep in sync with Account.enabled_l1_categories server_default in
+# app/models/account.py. The migration default (cb9e282267d2) is intentionally
+# a frozen copy and is NOT updated when this list changes.
+DEFAULT_L1_CATEGORIES: list[str] = [
+    "password_reset", "account_lockout", "printer", "email_outlook_client",
+    "wifi_network_basics", "vpn_connect", "teams_zoom_av",
+    "browser_cache_cookies", "peripheral_reconnect", "os_restart_update",
+]
+
+# Always-forbidden action classes (keys are stable identifiers; the human-readable
+# phrasing lives in the builder system prompt). Admins cannot enable these.
+HARD_FLOOR_FORBIDDEN: list[str] = [
+    "registry_edit", "system_file_or_boot_edit", "data_or_disk_deletion",
+    "credential_or_mfa_change", "security_or_av_or_firewall_change",
+    "elevated_or_admin_script", "domain_dns_dhcp_change",
+    "server_or_production_config", "billing_or_license_change",
+]
+
+# Substrings that, if present in a generated node's text, indicate a hard-floor
+# violation. Used by ai_tree_builder per-node validation (defense in depth).
+HARD_FLOOR_TEXT_PATTERNS: list[str] = [
+    "regedit", "registry", "format ", "delete partition", "diskpart",
+    "reset password for", "disable firewall", "disable antivirus", "disable defender",
+    "run as administrator", "sudo ", "domain controller", "dns record", "dhcp scope",
+    "uninstall security", "bitlocker",
+]
+
+
+def is_category_enabled(category: str, enabled: list[str]) -> bool:
+    """A category is buildable only if explicitly enabled and not hard-floored."""
+    if category in HARD_FLOOR_FORBIDDEN:
+        return False
+    return category in enabled
+
+
+async def get_enabled_categories(account_id: UUID, db: AsyncSession) -> list[str]:
+    """Return the account's enabled L1 categories (``or []`` guards pre-default rows)."""
+    acct = (await db.execute(select(Account).where(Account.id == account_id))).scalar_one()
+    return list(acct.enabled_l1_categories or [])
+
+
+async def set_enabled_categories(
+    account_id: UUID, categories: list[str], db: AsyncSession
+) -> list[str]:
+    """Persist the enabled set, dropping anything unknown or hard-floored.
+
+    Hard-floored keys (HARD_FLOOR_FORBIDDEN) are by design never present in
+    DEFAULT_L1_CATEGORIES, so the DEFAULT membership filter already excludes them.
+    If you ever add a key to DEFAULT_L1_CATEGORIES, verify it is not also in
+    HARD_FLOOR_FORBIDDEN. dict.fromkeys dedupes while preserving first-seen order.
+    """
+    cleaned = list(dict.fromkeys(c for c in categories if c in DEFAULT_L1_CATEGORIES))
+    acct = (await db.execute(select(Account).where(Account.id == account_id))).scalar_one()
+    acct.enabled_l1_categories = cleaned
+    await db.flush()
+    return cleaned

backend/app/services/l1_session_service.py

@@ -7,13 +7,16 @@ from datetime import datetime, timezone
 from typing import Optional
 from uuid import UUID
 
+from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.core.audit import log_audit
 from app.models.flow_proposal import FlowProposal
 from app.models.l1_walk_session import L1WalkSession
 from app.models.user import User
+from app.services import ai_tree_builder
 from app.services import internal_ticket_service
+from app.services.notification_service import notify
 
 
 def _resolve_acting_as(user: User) -> Optional[str]:
@@ -98,6 +101,82 @@ async def start_adhoc_session(
     return session
 
 
+async def start_ai_build_session(
+    db: AsyncSession,
+    *,
+    account_id: UUID,
+    user: User,
+    ticket_id: str,
+    ticket_kind: str,
+) -> L1WalkSession:
+    """Start an AI-built tree session (nodes generated on demand via next-node)."""
+    session = L1WalkSession(
+        account_id=account_id,
+        created_by_user_id=user.id,
+        acting_as=_resolve_acting_as(user),
+        ticket_id=ticket_id,
+        ticket_kind=ticket_kind,
+        session_kind="ai_build",
+    )
+    db.add(session)
+    await db.flush()
+    return session
+
+
+async def advance_ai_build(
+    db: AsyncSession,
+    *,
+    session_id: UUID,
+    problem_text: str,
+    category: str,
+    node_id: Optional[str] = None,
+    node_text: Optional[str] = None,
+    answer: Optional[str] = None,
+    note: Optional[str] = None,
+) -> dict:
+    """Append the answered/acked node to walked_path, then generate the next node.
+
+    On the first call (node_id is None) nothing is appended — we just generate the
+    first node. Returns the next node dict (caller persists current_node_id).
+    Raises ValueError on missing/inactive/non-ai_build session.
+
+    ``node_text`` is the display text of the node being answered. It is supplied by
+    the caller/endpoint, which holds the served node. Storing it here ensures that
+    later nodes receive full prior-step context via ``ai_tree_builder._build_context``
+    and that captured flywheel trees (``normalize_walked_path``) have meaningful text.
+    """
+    session = await db.get(L1WalkSession, session_id)
+    if not session:
+        raise ValueError(f"L1WalkSession {session_id} not found")
+    if session.session_kind != "ai_build":
+        raise ValueError("advance_ai_build requires an ai_build session")
+    if session.status != "active":
+        raise ValueError(f"Session {session_id} is not active (status={session.status})")
+
+    if node_id is not None:
+        # node_type inferred from the answer: questions are answered yes/no;
+        # instructions are acknowledged (answer is None) per the next-node endpoint contract.
+        # Note: entry uses key "id" (not "node_id" as record_step uses) because
+        # ai_tree_builder.normalize_walked_path reads step.get("id"); the two coexist
+        # safely because they are segregated by session_kind.
+        entry = {
+            "node_type": "question" if answer in ("yes", "no") else "instruction",
+            "id": node_id,
+            "text": node_text or "",
+            "answer": answer,
+            "l1_note": note,
+        }
+        # JSONB requires assigning a new list — in-place mutation isn't tracked
+        session.walked_path = [*session.walked_path, entry]
+
+    next_node = await ai_tree_builder.generate_next_node(
+        problem_text, category, session.walked_path)
+    session.current_node_id = next_node.get("id")
+    session.last_step_at = datetime.now(timezone.utc)
+    await db.flush()
+    return next_node
+
+
 async def record_step(
     db: AsyncSession,
     *,
@@ -186,6 +265,24 @@ async def resolve(
         if proposal:
             proposal.validated_by_outcome = True
 
+    # Flywheel capture: persist a validated FlowProposal for ai_build sessions
+    # resolved as helpful. Captures the AI-generated path as training signal.
+    if helpful and session.session_kind == "ai_build" and session.walked_path:
+        tree_structure = ai_tree_builder.normalize_walked_path(session.walked_path)
+        db.add(FlowProposal(
+            account_id=session.account_id,
+            l1_session_id=session.id,
+            source_session_id=None,
+            proposal_type="new_flow",
+            title=(session.resolution_notes or "AI L1 resolution")[:255],
+            proposed_flow_data={"tree_structure": tree_structure, "match_keywords": []},
+            source="ai_realtime_l1",
+            validated_by_outcome=True,
+            linked_ticket_id=session.ticket_id,
+            linked_ticket_kind=session.ticket_kind,
+            status="pending",
+        ))
+
     if session.ticket_kind == "internal":
         await internal_ticket_service.update_status(
             db,
@@ -262,6 +359,28 @@ async def escalate(
         account_id=session.account_id,
         acting_as=session.acting_as,
     )
+
+    # Notify engineers (owner/admin/engineer roles) about the escalation.
+    eng_rows = await db.execute(
+        select(User.id).where(
+            User.account_id == session.account_id,
+            User.is_active.is_(True),
+            User.account_role.in_(("owner", "admin", "engineer")),
+        )
+    )
+    target_ids = [r[0] for r in eng_rows.all()]
+    await notify(
+        "l1.session.escalated",
+        session.account_id,
+        {
+            "problem_summary": session.ticket_id,
+            "session_id": str(session.id),
+            "reason_category": reason_category,
+        },
+        db,
+        target_user_ids=target_ids,
+    )
+
     await db.flush()
     return session
 

backend/app/services/match_or_build.py

@@ -0,0 +1,78 @@
+"""Intake orchestrator: match published flows first, gate generic build behind
+the account's enabled categories (spec §3). Match runs BEFORE the category gate
+so an authored flow is never blocked by category settings (Finding 4)."""
+import logging
+import re
+from typing import Any, Optional
+from uuid import UUID
+
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.core.ai_provider import get_ai_provider
+from app.core.config import settings
+from app.services import flow_matching_engine
+from app.services.l1_category_service import (
+    DEFAULT_L1_CATEGORIES, get_enabled_categories, is_category_enabled,
+)
+from app.services.llm_utils import parse_llm_json
+
+logger = logging.getLogger(__name__)
+
+MATCH_THRESHOLD = 0.75   # spec §5.3
+SUGGEST_THRESHOLD = 0.60  # spec §5.3
+
+_CLASSIFY_PROMPT = (
+    "Classify the IT support problem into exactly one of these category keys, "
+    "or 'unknown'. Return JSON {\"category\":\"<key>\"} only.\nKEYS: "
+    + ", ".join(DEFAULT_L1_CATEGORIES)
+)
+
+
+async def classify(problem_text: str) -> str:
+    """Map a problem to a category key via a short model call; keyword fallback."""
+    try:
+        provider = get_ai_provider(settings.get_model_for_action("l1_classify"))
+        raw, _, _ = await provider.generate_json(
+            system_prompt=_CLASSIFY_PROMPT,
+            messages=[{"role": "user", "content": problem_text}],
+            max_tokens=64,
+        )
+        cat = parse_llm_json(raw).get("category", "unknown")
+        return cat if cat in DEFAULT_L1_CATEGORIES else "unknown"
+    except Exception as e:  # noqa: BLE001 — fall back, never hard-fail intake
+        logger.warning("classify model call failed (%s); keyword fallback", e)
+        text = problem_text.lower()
+        for cat in DEFAULT_L1_CATEGORIES:
+            if any(re.search(rf"\b{re.escape(tok)}\b", text) for tok in cat.split("_")):
+                return cat
+        return "unknown"
+
+
+async def match_or_build(
+    account_id: UUID,
+    problem_text: str,
+    problem_domain: Optional[str],
+    ticket_ref: str,  # passed through for caller/session use; not consumed here (Task 10)
+    *,
+    db: AsyncSession,
+    force_build: bool = False,
+) -> dict[str, Any]:
+    if not force_build:
+        hits = await flow_matching_engine.find_matches(
+            problem_text, problem_domain, account_id, db)
+        best = max(hits, key=lambda h: h["score"], default=None) if hits else None
+        # find_matches returns tree_id as a UUID object; normalize the public
+        # contract to str so callers can re-parse with UUID(...) without TypeError.
+        if best and best["score"] >= MATCH_THRESHOLD:
+            return {"outcome": "matched", "flow_id": str(best["tree_id"]), "session_kind": "flow"}
+        if best and best["score"] >= SUGGEST_THRESHOLD:
+            return {"outcome": "suggest",
+                    "near_miss": {"flow_id": str(best["tree_id"]), "flow_name": best["tree_name"],
+                                  "score": best["score"]},
+                    "can_build": True}
+
+    category = await classify(problem_text)
+    enabled = await get_enabled_categories(account_id, db)
+    if not is_category_enabled(category, enabled):
+        return {"outcome": "out_of_scope", "category": category}
+    return {"outcome": "build", "session_kind": "ai_build", "category": category}

backend/app/services/notification_service.py

@@ -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))
@@ -381,6 +386,7 @@ def _build_notification_title(event: str, payload: dict[str, Any]) -> str:
         "proposal.pending": "New flow proposal: {title}",
         "proposal.approved": "Flow proposal approved: {title}",
         "knowledge_gap.detected": "Knowledge gap detected: {gap_type}",
+        "l1.session.escalated": "L1 session escalated: {problem_summary}",
         "test": "Test Notification from ResolutionFlow",
     }
 
@@ -415,6 +421,7 @@ def _build_notification_body(event: str, payload: dict[str, Any]) -> str:
         "proposal.pending": "A new flow proposal \"{title}\" is awaiting review in the review queue.",
         "proposal.approved": "The flow proposal \"{title}\" has been approved and is ready for use.",
         "knowledge_gap.detected": "A {gap_type} knowledge gap has been identified. Review recommended.",
+        "l1.session.escalated": "L1 escalated a ticket: {problem_summary}",
         "test": "This is a test notification to verify your notification channel is working correctly.",
     }
     template = bodies.get(event, f"Event: {event}")
@@ -437,6 +444,9 @@ def _build_notification_link(event: str, payload: dict[str, Any]) -> Optional[st
         "proposal.pending": "/review-queue",
         "proposal.approved": "/review-queue",
         "knowledge_gap.detected": "/analytics/flowpilot",
+        # L1 AI-build escalations go to the escalations dashboard — not to
+        # a specific pilot session, which may not have a pickup flow.
+        "l1.session.escalated": "/escalations",
     }
     template = links.get(event)
     if template is None:

backend/tests/test_account_l1_categories_column.py

@@ -0,0 +1,7 @@
+from app.models.account import Account
+
+
+def test_account_has_enabled_l1_categories_default():
+    a = Account(name="Acme", display_code="ABC12345")
+    # Column default is applied at flush; attribute may be None pre-flush.
+    assert hasattr(a, "enabled_l1_categories")

backend/tests/test_ai_tree_builder.py

@@ -0,0 +1,58 @@
+import pytest
+from app.services import ai_tree_builder as atb
+
+
+def test_validate_node_rejects_hard_floor_text():
+    node = {"node_type": "instruction", "id": "n1", "text": "Open regedit and change the key", "next": "generate"}
+    with pytest.raises(atb.UnsafeNodeError):
+        atb.validate_node(node)
+
+
+def test_validate_node_accepts_safe_instruction():
+    node = {"node_type": "instruction", "id": "n1", "text": "Restart the printer.", "next": "generate"}
+    assert atb.validate_node(node)["node_type"] == "instruction"
+
+
+def test_depth_cap_forces_escalate():
+    walked = [{"node_type": "question", "id": f"n{i}", "text": "?", "answer": "no"} for i in range(atb.MAX_DEPTH)]
+    node = atb.escalate_if_depth_exceeded(walked)
+    assert node is not None and node["node_type"] == "escalate"
+
+
+def test_normalize_walked_path_builds_valid_tree():
+    walked = [
+        {"node_type": "question", "id": "n1", "text": "Powered on?", "answer": "no"},
+        {"node_type": "instruction", "id": "n2", "text": "Power it on.", "answer": "ack"},
+        {"node_type": "resolved", "id": "n3", "text": "Fixed."},
+    ]
+    tree = atb.normalize_walked_path(walked)
+    assert isinstance(tree, dict) and tree.get("id") == "n1"
+    # untraversed 'yes' branch of n1 became a needs_review stub
+    assert any(n["node_type"] == "needs_review" for n in tree["nodes"].values())
+
+
+def test_normalize_walk_ending_on_question_has_no_none_branches():
+    walked = [
+        {"node_type": "question", "id": "n1", "text": "Powered on?", "answer": "no"},
+    ]
+    tree = atb.normalize_walked_path(walked)
+    n1 = tree["nodes"]["n1"]
+    assert n1["yes_next"] is not None and n1["no_next"] is not None
+    # both branches must reference real nodes present in the tree
+    assert n1["yes_next"] in tree["nodes"] and n1["no_next"] in tree["nodes"]
+
+
+def test_normalize_preserves_escalate_reason_category():
+    walked = [
+        {"node_type": "question", "id": "n1", "text": "On?", "answer": "no"},
+        {"node_type": "escalate", "id": "n2", "text": "Beyond L1.",
+         "reason_category": "exhausted_safe_steps"},
+    ]
+    tree = atb.normalize_walked_path(walked)
+    assert tree["nodes"]["n2"]["reason_category"] == "exhausted_safe_steps"
+
+
+def test_normalize_empty_walk_returns_needs_review_root():
+    tree = atb.normalize_walked_path([])
+    assert tree["id"] in tree["nodes"]
+    assert tree["nodes"][tree["id"]]["node_type"] == "needs_review"

backend/tests/test_flow_proposal_l1_source.py

@@ -0,0 +1,16 @@
+import uuid
+from app.models.flow_proposal import FlowProposal
+
+
+def test_flow_proposal_accepts_l1_session_id_without_source_session():
+    p = FlowProposal(
+        account_id=uuid.uuid4(),
+        l1_session_id=uuid.uuid4(),
+        source_session_id=None,
+        proposal_type="new_flow",
+        title="AI L1 draft",
+        proposed_flow_data={"tree_structure": {"id": "root"}},
+        source="ai_realtime_l1",
+        status="pending",
+    )
+    assert p.l1_session_id is not None and p.source_session_id is None

backend/tests/test_l1_ai_build_flow.py

@@ -0,0 +1,161 @@
+"""End-to-end backend integration test for the L1 AI-build flow (Phase 2A).
+
+Drives the real endpoint + service path — intake (build) → next-node walk →
+resolve — and asserts an outcome-validated FlowProposal is captured. Only the AI
+boundary is mocked: match_or_build's outcome and ai_tree_builder.generate_next_node.
+A second test drives intake → escalate and asserts the engineer notification fires
+and the session surfaces in GET /l1/escalations.
+"""
+import uuid
+from unittest.mock import AsyncMock, patch
+
+import pytest
+from httpx import AsyncClient
+from sqlalchemy import delete, select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.models.flow_proposal import FlowProposal
+from app.models.subscription import Subscription
+from app.models.user import User
+
+
+async def _register(client: AsyncClient, *, email: str) -> dict:
+    resp = await client.post(
+        "/api/v1/auth/register",
+        json={"email": email, "password": "TestPassword123!", "name": "Test User"},
+    )
+    assert resp.status_code in (200, 201), resp.text
+    return resp.json()
+
+
+async def _login(client: AsyncClient, *, email: str) -> dict:
+    resp = await client.post(
+        "/api/v1/auth/login/json",
+        json={"email": email, "password": "TestPassword123!"},
+    )
+    assert resp.status_code == 200, resp.text
+    return {"Authorization": f"Bearer {resp.json()['access_token']}"}
+
+
+async def _ensure_subscription(db: AsyncSession, account_id: uuid.UUID) -> None:
+    await db.execute(delete(Subscription).where(Subscription.account_id == account_id))
+    db.add(Subscription(account_id=account_id, plan="pro", status="active"))
+    await db.commit()
+
+
+async def _make_user(client: AsyncClient, db: AsyncSession, *, email: str, account_role: str) -> dict:
+    data = await _register(client, email=email)
+    uid = uuid.UUID(data["id"])
+    acct_id = uuid.UUID(data["account_id"])
+    user = (await db.execute(select(User).where(User.id == uid))).scalar_one()
+    user.account_role = account_role
+    await db.commit()
+    await _ensure_subscription(db, acct_id)
+    headers = await _login(client, email=email)
+    return {"headers": headers, "account_id": acct_id, "user_id": uid}
+
+
+@pytest.mark.asyncio
+async def test_intake_build_walk_resolve_creates_proposal(client: AsyncClient, test_db: AsyncSession):
+    """intake(build) → answer a question node → reach resolved → resolve → proposal."""
+    info = await _make_user(client, test_db, email="flow_resolve@example.com", account_role="l1_tech")
+    headers = info["headers"]
+
+    # 1. force a build outcome at intake (real ticket + ai_build session created)
+    with patch(
+        "app.api.endpoints.l1.match_or_build.match_or_build",
+        new=AsyncMock(return_value={"outcome": "build", "session_kind": "ai_build",
+                                    "category": "printer"}),
+    ):
+        r = await client.post("/api/v1/l1/intake",
+                              json={"problem_statement": "printer jam"}, headers=headers)
+    assert r.status_code == 200, r.text
+    sid = r.json()["session_id"]
+
+    # 2. drive next-node deterministically: first a question, then a resolved terminal
+    seq = iter([
+        {"node_type": "question", "id": "n1", "text": "Is the printer powered on?"},
+        {"node_type": "resolved", "id": "n2", "text": "Printer prints a test page."},
+    ])
+
+    async def fake_next(problem_text, category, walked_path):
+        return next(seq)
+
+    with patch("app.services.ai_tree_builder.generate_next_node", new=fake_next):
+        r1 = await client.post(f"/api/v1/l1/sessions/{sid}/next-node",
+                               json={}, headers=headers)
+        assert r1.status_code == 200, r1.text
+        assert r1.json()["node"]["node_type"] == "question"
+
+        r2 = await client.post(
+            f"/api/v1/l1/sessions/{sid}/next-node",
+            json={"node_id": "n1", "node_text": "Is the printer powered on?", "answer": "no"},
+            headers=headers,
+        )
+        assert r2.status_code == 200, r2.text
+        assert r2.json()["node"]["node_type"] == "resolved"
+
+    # 3. resolve helpful → outcome-validated proposal captured
+    rr = await client.post(f"/api/v1/l1/sessions/{sid}/resolve",
+                           json={"helpful": True, "resolution_notes": "Powered it on."},
+                           headers=headers)
+    assert rr.status_code == 200, rr.text
+    assert rr.json()["status"] == "resolved"
+
+    props = (await test_db.execute(
+        select(FlowProposal).where(FlowProposal.source == "ai_realtime_l1")
+    )).scalars().all()
+    assert len(props) == 1
+    p = props[0]
+    assert p.validated_by_outcome is True
+    assert p.source_session_id is None
+    assert str(p.l1_session_id) == sid
+    # the walked question 'n1' becomes the captured tree root (meta entry skipped)
+    assert p.proposed_flow_data["tree_structure"]["id"] == "n1"
+
+
+@pytest.mark.asyncio
+async def test_intake_build_escalate_notifies_and_lists(client: AsyncClient, test_db: AsyncSession):
+    """intake(build) → escalate → notify fires for engineers → appears in GET /escalations."""
+    # an engineer in the same account is the escalation recipient + the queue viewer
+    l1 = await _make_user(client, test_db, email="flow_esc_l1@example.com", account_role="l1_tech")
+    eng_data = await _register(client, email="flow_esc_eng@example.com")
+    eng_uid = uuid.UUID(eng_data["id"])
+    # put the engineer in the L1 tech's account
+    eng = (await test_db.execute(select(User).where(User.id == eng_uid))).scalar_one()
+    eng.account_id = l1["account_id"]
+    eng.account_role = "engineer"
+    await test_db.commit()
+    eng_headers = await _login(client, email="flow_esc_eng@example.com")
+
+    with patch(
+        "app.api.endpoints.l1.match_or_build.match_or_build",
+        new=AsyncMock(return_value={"outcome": "build", "session_kind": "ai_build",
+                                    "category": "printer"}),
+    ):
+        r = await client.post("/api/v1/l1/intake",
+                              json={"problem_statement": "weird driver fault"},
+                              headers=l1["headers"])
+    assert r.status_code == 200, r.text
+    sid = r.json()["session_id"]
+
+    captured = {}
+
+    async def fake_notify(event, account_id, payload, db, target_user_ids=None):
+        captured["event"] = event
+        captured["target_user_ids"] = target_user_ids
+
+    with patch("app.services.l1_session_service.notify", new=fake_notify):
+        re_ = await client.post(f"/api/v1/l1/sessions/{sid}/escalate",
+                                json={"reason_category": "exhausted_safe_steps",
+                                      "reason": "Beyond L1 scope"},
+                                headers=l1["headers"])
+    assert re_.status_code == 200, re_.text
+    assert re_.json()["status"] == "escalated"
+    assert captured["event"] == "l1.session.escalated"
+    assert eng_uid in (captured["target_user_ids"] or [])
+
+    # engineer sees it in the escalations queue
+    q = await client.get("/api/v1/l1/escalations", headers=eng_headers)
+    assert q.status_code == 200, q.text
+    assert any(row["id"] == sid for row in q.json())

backend/tests/test_l1_ai_build_model.py

@@ -0,0 +1,16 @@
+import uuid
+
+from app.models.l1_walk_session import L1WalkSession
+
+
+def test_ai_build_session_kind_allowed_by_model_constraint():
+    """ai_build is a valid session_kind with both target FKs null (like adhoc)."""
+    s = L1WalkSession(
+        account_id=uuid.uuid4(),
+        created_by_user_id=uuid.uuid4(),
+        ticket_id="t1",
+        ticket_kind="internal",
+        session_kind="ai_build",
+    )
+    assert s.session_kind == "ai_build"
+    assert s.flow_id is None and s.flow_proposal_id is None

backend/tests/test_l1_api_ai_build.py

@@ -0,0 +1,157 @@
+"""Integration tests for the Phase 2A L1 AI-build API surface.
+
+Covers intake dispatch (match_or_build outcomes), the next-node endpoint, and the
+engineer escalations list. The orchestrator and node generator are mocked — this
+exercises the endpoint wiring, not the AI. Auth/subscription follow the same
+register → promote-role → ensure-subscription → login pattern as test_l1_endpoints.
+"""
+import uuid
+from datetime import datetime, timezone
+from unittest.mock import AsyncMock, patch
+
+import pytest
+from httpx import AsyncClient
+from sqlalchemy import delete, select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.models.l1_walk_session import L1WalkSession
+from app.models.subscription import Subscription
+from app.models.user import User
+
+
+async def _register(client: AsyncClient, *, email: str) -> dict:
+    resp = await client.post(
+        "/api/v1/auth/register",
+        json={"email": email, "password": "TestPassword123!", "name": "Test User"},
+    )
+    assert resp.status_code in (200, 201), resp.text
+    return resp.json()
+
+
+async def _login(client: AsyncClient, *, email: str) -> dict:
+    resp = await client.post(
+        "/api/v1/auth/login/json",
+        json={"email": email, "password": "TestPassword123!"},
+    )
+    assert resp.status_code == 200, resp.text
+    return {"Authorization": f"Bearer {resp.json()['access_token']}"}
+
+
+async def _ensure_subscription(db: AsyncSession, account_id: uuid.UUID) -> None:
+    await db.execute(delete(Subscription).where(Subscription.account_id == account_id))
+    db.add(Subscription(account_id=account_id, plan="pro", status="active"))
+    await db.commit()
+
+
+async def _make_user(client: AsyncClient, db: AsyncSession, *, email: str, account_role: str) -> dict:
+    """Register a user, promote to a role, ensure an active subscription, return headers + ids."""
+    data = await _register(client, email=email)
+    uid = uuid.UUID(data["id"])
+    acct_id = uuid.UUID(data["account_id"])
+    user = (await db.execute(select(User).where(User.id == uid))).scalar_one()
+    user.account_role = account_role
+    await db.commit()
+    await _ensure_subscription(db, acct_id)
+    headers = await _login(client, email=email)  # login AFTER role change
+    return {"headers": headers, "account_id": acct_id, "user_id": uid}
+
+
+@pytest.mark.asyncio
+async def test_intake_build_outcome_creates_ai_build_session(client: AsyncClient, test_db: AsyncSession):
+    """intake → match_or_build returns 'build' → an ai_build session is created."""
+    info = await _make_user(client, test_db, email="aib_build@example.com", account_role="l1_tech")
+    with patch(
+        "app.api.endpoints.l1.match_or_build.match_or_build",
+        new=AsyncMock(return_value={"outcome": "build", "session_kind": "ai_build",
+                                    "category": "printer"}),
+    ):
+        r = await client.post("/api/v1/l1/intake",
+                              json={"problem_statement": "printer jam"}, headers=info["headers"])
+    assert r.status_code == 200, r.text
+    body = r.json()
+    assert body["outcome"] == "build"
+    assert body["session_kind"] == "ai_build"
+    assert body["session_id"]
+
+
+@pytest.mark.asyncio
+async def test_intake_out_of_scope(client: AsyncClient, test_db: AsyncSession):
+    """intake → 'out_of_scope' → no session, surfaced to the caller."""
+    info = await _make_user(client, test_db, email="aib_oos@example.com", account_role="l1_tech")
+    with patch(
+        "app.api.endpoints.l1.match_or_build.match_or_build",
+        new=AsyncMock(return_value={"outcome": "out_of_scope", "category": "unknown"}),
+    ):
+        r = await client.post("/api/v1/l1/intake",
+                              json={"problem_statement": "weird"}, headers=info["headers"])
+    assert r.status_code == 200, r.text
+    body = r.json()
+    assert body["outcome"] == "out_of_scope"
+    assert body.get("session_id") is None
+
+
+@pytest.mark.asyncio
+async def test_intake_suggest_returns_near_miss(client: AsyncClient, test_db: AsyncSession):
+    """intake → 'suggest' → near_miss prompt, no session."""
+    info = await _make_user(client, test_db, email="aib_sugg@example.com", account_role="l1_tech")
+    near = {"flow_id": str(uuid.uuid4()), "flow_name": "VPN", "score": 0.66}
+    with patch(
+        "app.api.endpoints.l1.match_or_build.match_or_build",
+        new=AsyncMock(return_value={"outcome": "suggest", "near_miss": near, "can_build": True}),
+    ):
+        r = await client.post("/api/v1/l1/intake",
+                              json={"problem_statement": "vpn"}, headers=info["headers"])
+    assert r.status_code == 200, r.text
+    assert r.json()["near_miss"]["flow_name"] == "VPN"
+
+
+@pytest.mark.asyncio
+async def test_next_node_returns_generated_node(client: AsyncClient, test_db: AsyncSession):
+    """After a build intake, /next-node returns the node from advance_ai_build."""
+    info = await _make_user(client, test_db, email="aib_next@example.com", account_role="l1_tech")
+    with patch(
+        "app.api.endpoints.l1.match_or_build.match_or_build",
+        new=AsyncMock(return_value={"outcome": "build", "session_kind": "ai_build",
+                                    "category": "printer"}),
+    ):
+        r = await client.post("/api/v1/l1/intake",
+                              json={"problem_statement": "printer jam"}, headers=info["headers"])
+    sid = r.json()["session_id"]
+    with patch(
+        "app.api.endpoints.l1.l1_session_service.advance_ai_build",
+        new=AsyncMock(return_value={"node_type": "question", "id": "n1", "text": "Powered on?"}),
+    ):
+        r2 = await client.post(f"/api/v1/l1/sessions/{sid}/next-node",
+                               json={}, headers=info["headers"])
+    assert r2.status_code == 200, r2.text
+    assert r2.json()["node"]["node_type"] == "question"
+
+
+@pytest.mark.asyncio
+async def test_escalations_lists_escalated_sessions_for_engineer(client: AsyncClient, test_db: AsyncSession):
+    """GET /l1/escalations returns escalated L1 sessions for an engineer-or-above user."""
+    info = await _make_user(client, test_db, email="aib_eng@example.com", account_role="engineer")
+    now = datetime.now(timezone.utc)
+    sess = L1WalkSession(
+        account_id=info["account_id"],
+        created_by_user_id=info["user_id"],
+        ticket_id="t-esc",
+        ticket_kind="internal",
+        session_kind="ai_build",
+        status="escalated",
+        started_at=now,
+        last_step_at=now,
+    )
+    test_db.add(sess)
+    await test_db.commit()
+    r = await client.get("/api/v1/l1/escalations", headers=info["headers"])
+    assert r.status_code == 200, r.text
+    assert any(row["id"] == str(sess.id) for row in r.json())
+
+
+@pytest.mark.asyncio
+async def test_escalations_forbidden_for_l1_tech(client: AsyncClient, test_db: AsyncSession):
+    """An l1_tech (not engineer-or-above) is rejected from the escalations queue."""
+    info = await _make_user(client, test_db, email="aib_l1@example.com", account_role="l1_tech")
+    r = await client.get("/api/v1/l1/escalations", headers=info["headers"])
+    assert r.status_code == 403, r.text

backend/tests/test_l1_categories_api.py

@@ -0,0 +1,119 @@
+"""Tests for the account L1 AI-build category settings API (Phase 2A).
+
+GET /accounts/me/l1-categories — readable by L1-or-above.
+PATCH /accounts/me/l1-categories — owner/admin only; drops unknown/hard-floored keys.
+"""
+import uuid
+
+import pytest
+from httpx import AsyncClient
+from sqlalchemy import delete, select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.models.subscription import Subscription
+from app.models.user import User
+
+
+async def _register(client: AsyncClient, *, email: str) -> dict:
+    resp = await client.post(
+        "/api/v1/auth/register",
+        json={"email": email, "password": "TestPassword123!", "name": "Test User"},
+    )
+    assert resp.status_code in (200, 201), resp.text
+    return resp.json()
+
+
+async def _login(client: AsyncClient, *, email: str) -> dict:
+    resp = await client.post(
+        "/api/v1/auth/login/json",
+        json={"email": email, "password": "TestPassword123!"},
+    )
+    assert resp.status_code == 200, resp.text
+    return {"Authorization": f"Bearer {resp.json()['access_token']}"}
+
+
+async def _ensure_subscription(db: AsyncSession, account_id: uuid.UUID) -> None:
+    await db.execute(delete(Subscription).where(Subscription.account_id == account_id))
+    db.add(Subscription(account_id=account_id, plan="pro", status="active"))
+    await db.commit()
+
+
+async def _make_user(client: AsyncClient, db: AsyncSession, *, email: str, account_role: str) -> dict:
+    """Register → promote role → ensure subscription → login (after the role change)."""
+    data = await _register(client, email=email)
+    uid = uuid.UUID(data["id"])
+    acct_id = uuid.UUID(data["account_id"])
+    user = (await db.execute(select(User).where(User.id == uid))).scalar_one()
+    user.account_role = account_role
+    await db.commit()
+    await _ensure_subscription(db, acct_id)
+    headers = await _login(client, email=email)
+    return {"headers": headers, "account_id": acct_id, "user_id": uid}
+
+
+@pytest.mark.asyncio
+async def test_get_categories_returns_enabled_available_hard_floor(client: AsyncClient, test_db: AsyncSession):
+    info = await _make_user(client, test_db, email="cat_owner_get@example.com", account_role="owner")
+    r = await client.get("/api/v1/accounts/me/l1-categories", headers=info["headers"])
+    assert r.status_code == 200, r.text
+    body = r.json()
+    assert "enabled" in body and "available" in body and "hard_floor" in body
+    # New account defaults to the full available allowlist (10 keys).
+    assert len(body["available"]) == 10
+    assert "password_reset" in body["available"]
+    assert "registry_edit" in body["hard_floor"]
+
+
+@pytest.mark.asyncio
+async def test_get_categories_readable_by_l1_tech(client: AsyncClient, test_db: AsyncSession):
+    info = await _make_user(client, test_db, email="cat_l1_get@example.com", account_role="l1_tech")
+    r = await client.get("/api/v1/accounts/me/l1-categories", headers=info["headers"])
+    assert r.status_code == 200, r.text
+
+
+@pytest.mark.asyncio
+async def test_patch_categories_owner_can_set(client: AsyncClient, test_db: AsyncSession):
+    info = await _make_user(client, test_db, email="cat_owner_patch@example.com", account_role="owner")
+    r = await client.patch(
+        "/api/v1/accounts/me/l1-categories",
+        json={"enabled": ["printer", "vpn_connect"]},
+        headers=info["headers"],
+    )
+    assert r.status_code == 200, r.text
+    assert set(r.json()["enabled"]) == {"printer", "vpn_connect"}
+
+
+@pytest.mark.asyncio
+async def test_patch_categories_drops_unknown_and_hard_floored(client: AsyncClient, test_db: AsyncSession):
+    info = await _make_user(client, test_db, email="cat_owner_drop@example.com", account_role="owner")
+    r = await client.patch(
+        "/api/v1/accounts/me/l1-categories",
+        json={"enabled": ["printer", "registry_edit", "bogus_key"]},
+        headers=info["headers"],
+    )
+    assert r.status_code == 200, r.text
+    # registry_edit (hard floor) and bogus_key (unknown) are dropped.
+    assert r.json()["enabled"] == ["printer"]
+
+
+@pytest.mark.asyncio
+async def test_patch_categories_forbidden_for_l1_tech(client: AsyncClient, test_db: AsyncSession):
+    info = await _make_user(client, test_db, email="cat_l1_patch@example.com", account_role="l1_tech")
+    r = await client.patch(
+        "/api/v1/accounts/me/l1-categories",
+        json={"enabled": ["printer"]},
+        headers=info["headers"],
+    )
+    assert r.status_code == 403, r.text
+
+
+@pytest.mark.asyncio
+async def test_patch_categories_forbidden_for_engineer(client: AsyncClient, test_db: AsyncSession):
+    """Write is owner/admin only — engineers (who pass require_engineer_or_admin) are blocked."""
+    info = await _make_user(client, test_db, email="cat_eng_patch@example.com", account_role="engineer")
+    r = await client.patch(
+        "/api/v1/accounts/me/l1-categories",
+        json={"enabled": ["printer"]},
+        headers=info["headers"],
+    )
+    assert r.status_code == 403, r.text

backend/tests/test_l1_category_service.py

@@ -0,0 +1,16 @@
+from app.services.l1_category_service import (
+    DEFAULT_L1_CATEGORIES, HARD_FLOOR_FORBIDDEN, is_category_enabled,
+)
+
+
+def test_defaults_and_hard_floor_present():
+    assert "password_reset" in DEFAULT_L1_CATEGORIES
+    assert "registry_edit" in HARD_FLOOR_FORBIDDEN  # representative forbidden action key
+    assert len(DEFAULT_L1_CATEGORIES) == 10
+
+
+def test_is_category_enabled():
+    enabled = ["printer", "vpn_connect"]
+    assert is_category_enabled("printer", enabled) is True
+    assert is_category_enabled("registry_edit", enabled) is False
+    assert is_category_enabled("unknown", enabled) is False

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):
-    """POST /l1/intake without flow_id creates adhoc session."""
+async def test_intake_build_creates_ai_build_session(client: AsyncClient, test_db: AsyncSession):
+    """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(
-        "/api/v1/l1/intake",
-        json={"problem_statement": "Printer won't turn on", "customer_name": "Alice"},
-        headers=headers,
-    )
+    with patch(
+        "app.api.endpoints.l1.match_or_build.match_or_build",
+        new=AsyncMock(return_value={"outcome": "build", "session_kind": "ai_build",
+                                    "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 "ticket_id" in body
+    assert body["session_id"]
+    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
-    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"]
+    session_id = await _create_adhoc_session(test_db, info, problem="Adhoc issue")
 
     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(
-        "/api/v1/l1/intake",
-        json={"problem_statement": "Notes test"},
-        headers=headers,
-    )
-    assert resp.status_code == 200, resp.text
-    session_id = resp.json()["session_id"]
+    session_id = await _create_adhoc_session(test_db, info, problem="Notes test")
 
     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(
-        "/api/v1/l1/intake",
-        json={"problem_statement": "Resolve test"},
-        headers=headers,
-    )
-    assert resp.status_code == 200, resp.text
-    session_id = resp.json()["session_id"]
+    session_id = await _create_adhoc_session(test_db, info, problem="Resolve test")
 
     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(
-        "/api/v1/l1/intake",
-        json={"problem_statement": "Escalation test"},
-        headers=headers,
-    )
-    assert resp.status_code == 200, resp.text
-    session_id = resp.json()["session_id"]
+    session_id = await _create_adhoc_session(test_db, info, problem="Escalation test")
 
     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(
-        "/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"]
+    session_id = await _create_adhoc_session(test_db, info_a, problem="Account A issue")
 
     # Account B: different user in a different account
     info_b = await _make_l1_user(client, test_db, email="l1tenantb@example.com")

backend/tests/test_l1_session_service.py

@@ -778,6 +778,165 @@ async def test_escalate_without_walk_reason_is_optional(test_db: AsyncSession):
     assert session.escalation_reason_category == "no_kb_content"
 
 
+# ---------------------------------------------------------------------------
+# T7: start_ai_build_session
+# ---------------------------------------------------------------------------
+
+@pytest.mark.asyncio
+async def test_start_ai_build_session(test_db: AsyncSession):
+    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)
+    s = await svc.start_ai_build_session(
+        test_db, account_id=account.id, user=l1_user,
+        ticket_id="t-ai", ticket_kind="internal",
+    )
+    assert s.session_kind == "ai_build"
+    assert s.flow_id is None and s.flow_proposal_id is None
+    assert s.status == "active"
+
+
+# ---------------------------------------------------------------------------
+# T8: advance_ai_build
+# ---------------------------------------------------------------------------
+
+@pytest.mark.asyncio
+async def test_advance_ai_build_appends_and_returns_next(test_db: AsyncSession, monkeypatch):
+    from app.services import l1_session_service as svc
+    from app.services import ai_tree_builder
+    account = await _make_account(test_db)
+    l1_user = await _make_user(test_db, account_id=account.id)
+    s = await svc.start_ai_build_session(
+        test_db, account_id=account.id, user=l1_user,
+        ticket_id="t-ai", ticket_kind="internal")
+
+    async def fake_next(problem, category, walked):
+        return {"node_type": "resolved", "id": "done", "text": "Fixed."}
+    monkeypatch.setattr(ai_tree_builder, "generate_next_node", fake_next)
+
+    next_node = await svc.advance_ai_build(
+        test_db, session_id=s.id, problem_text="printer", category="printer",
+        node_id="n1", node_text="Powered on?", answer="no", note=None)
+    assert next_node["node_type"] == "resolved"
+    refreshed = await test_db.get(type(s), s.id)
+    assert len(refreshed.walked_path) == 1
+    assert refreshed.walked_path[0]["answer"] == "no"
+    assert refreshed.walked_path[0]["text"] == "Powered on?"
+
+
+@pytest.mark.asyncio
+async def test_advance_ai_build_first_call_does_not_append(test_db: AsyncSession, monkeypatch):
+    from app.services import l1_session_service as svc
+    from app.services import ai_tree_builder
+    account = await _make_account(test_db)
+    l1_user = await _make_user(test_db, account_id=account.id)
+    s = await svc.start_ai_build_session(
+        test_db, account_id=account.id, user=l1_user,
+        ticket_id="t-ai-first", ticket_kind="internal")
+
+    async def fake_next(problem, category, walked):
+        return {"node_type": "question", "id": "q1", "text": "Is it plugged in?"}
+    monkeypatch.setattr(ai_tree_builder, "generate_next_node", fake_next)
+
+    # First call: node_id=None — nothing should be appended
+    next_node = await svc.advance_ai_build(
+        test_db, session_id=s.id, problem_text="printer", category="printer",
+        node_id=None)
+    assert next_node["node_type"] == "question"
+    assert next_node["id"] == "q1"
+    refreshed = await test_db.get(type(s), s.id)
+    assert len(refreshed.walked_path) == 0
+    assert refreshed.current_node_id == "q1"
+
+
+@pytest.mark.asyncio
+async def test_advance_ai_build_wrong_session_kind_raises(test_db: AsyncSession, monkeypatch):
+    from app.services import l1_session_service as svc
+    from app.services import ai_tree_builder
+    account = await _make_account(test_db)
+    l1_user = await _make_user(test_db, account_id=account.id)
+    # start an adhoc session (not ai_build)
+    s = await svc.start_adhoc_session(
+        test_db, account_id=account.id, user=l1_user,
+        ticket_id="t-adhoc-guard", ticket_kind="internal")
+
+    async def fake_next(problem, category, walked):  # pragma: no cover
+        return {"node_type": "question", "id": "q1", "text": "?"}
+    monkeypatch.setattr(ai_tree_builder, "generate_next_node", fake_next)
+
+    with pytest.raises(ValueError, match="ai_build"):
+        await svc.advance_ai_build(
+            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)
 # ---------------------------------------------------------------------------
@@ -914,4 +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"

backend/tests/test_match_or_build.py

@@ -0,0 +1,98 @@
+import uuid
+import pytest
+from unittest.mock import AsyncMock, patch
+from app.services import match_or_build as mob
+
+
+@pytest.mark.asyncio
+async def test_match_wins_before_category_gate():
+    """A strong published-flow match returns 'matched' even if category disabled."""
+    with patch.object(mob.flow_matching_engine, "find_matches", new=AsyncMock(
+            return_value=[{"tree_id": str(uuid.uuid4()), "tree_name": "VPN", "score": 0.9}])), \
+         patch.object(mob, "get_enabled_categories", new=AsyncMock(return_value=[])):
+        res = await mob.match_or_build(uuid.uuid4(), "vpn down", None, "t1", db=AsyncMock(), force_build=False)
+    assert res["outcome"] == "matched"
+    assert res["session_kind"] == "flow"
+
+
+@pytest.mark.asyncio
+async def test_suggest_band():
+    with patch.object(mob.flow_matching_engine, "find_matches", new=AsyncMock(
+            return_value=[{"tree_id": str(uuid.uuid4()), "tree_name": "X", "score": 0.66}])):
+        res = await mob.match_or_build(uuid.uuid4(), "p", None, "t1", db=AsyncMock(), force_build=False)
+    assert res["outcome"] == "suggest"
+    assert res["near_miss"]["flow_name"] == "X"
+    assert "flow_id" in res["near_miss"] and isinstance(res["near_miss"]["flow_id"], str)
+    assert res["near_miss"]["score"] == 0.66
+    assert res["can_build"] is True
+
+
+@pytest.mark.asyncio
+async def test_out_of_scope_when_category_disabled_on_build_path():
+    with patch.object(mob.flow_matching_engine, "find_matches", new=AsyncMock(return_value=[])), \
+         patch.object(mob, "classify", new=AsyncMock(return_value="printer")), \
+         patch.object(mob, "get_enabled_categories", new=AsyncMock(return_value=["vpn_connect"])):
+        res = await mob.match_or_build(uuid.uuid4(), "printer jam", None, "t1", db=AsyncMock(), force_build=False)
+    assert res["outcome"] == "out_of_scope"
+
+
+@pytest.mark.asyncio
+async def test_build_when_enabled_and_no_match():
+    with patch.object(mob.flow_matching_engine, "find_matches", new=AsyncMock(return_value=[])), \
+         patch.object(mob, "classify", new=AsyncMock(return_value="printer")), \
+         patch.object(mob, "get_enabled_categories", new=AsyncMock(return_value=["printer"])):
+        res = await mob.match_or_build(uuid.uuid4(), "printer jam", None, "t1", db=AsyncMock(), force_build=False)
+    assert res["outcome"] == "build"
+    assert res["session_kind"] == "ai_build"
+
+
+@pytest.mark.asyncio
+async def test_force_build_skips_match_but_still_gates():
+    fm = AsyncMock(return_value=[{"tree_id": str(uuid.uuid4()), "tree_name": "X", "score": 0.99}])
+    with patch.object(mob.flow_matching_engine, "find_matches", new=fm), \
+         patch.object(mob, "classify", new=AsyncMock(return_value="printer")), \
+         patch.object(mob, "get_enabled_categories", new=AsyncMock(return_value=["printer"])):
+        res = await mob.match_or_build(uuid.uuid4(), "p", None, "t1", db=AsyncMock(), force_build=True)
+    fm.assert_not_called()
+    assert res["outcome"] == "build"
+
+
+@pytest.mark.asyncio
+async def test_score_exactly_match_threshold_is_matched():
+    with patch.object(mob.flow_matching_engine, "find_matches", new=AsyncMock(
+            return_value=[{"tree_id": str(uuid.uuid4()), "tree_name": "X", "score": 0.75}])):
+        res = await mob.match_or_build(uuid.uuid4(), "p", None, "t1", db=AsyncMock(), force_build=False)
+    assert res["outcome"] == "matched"
+
+
+@pytest.mark.asyncio
+async def test_score_exactly_suggest_threshold_is_suggest():
+    with patch.object(mob.flow_matching_engine, "find_matches", new=AsyncMock(
+            return_value=[{"tree_id": str(uuid.uuid4()), "tree_name": "X", "score": 0.60}])):
+        res = await mob.match_or_build(uuid.uuid4(), "p", None, "t1", db=AsyncMock(), force_build=False)
+    assert res["outcome"] == "suggest"
+
+
+@pytest.mark.asyncio
+async def test_score_below_suggest_falls_through_to_build_path():
+    with patch.object(mob.flow_matching_engine, "find_matches", new=AsyncMock(
+            return_value=[{"tree_id": str(uuid.uuid4()), "tree_name": "X", "score": 0.4}])), \
+         patch.object(mob, "classify", new=AsyncMock(return_value="printer")), \
+         patch.object(mob, "get_enabled_categories", new=AsyncMock(return_value=["printer"])):
+        res = await mob.match_or_build(uuid.uuid4(), "printer", None, "t1", db=AsyncMock(), force_build=False)
+    assert res["outcome"] == "build"
+
+
+@pytest.mark.asyncio
+async def test_classify_keyword_fallback_matches_word():
+    with patch.object(mob, "get_ai_provider", side_effect=RuntimeError("model down")):
+        cat = await mob.classify("the printer is jammed")
+    assert cat == "printer"
+
+
+@pytest.mark.asyncio
+async def test_classify_keyword_fallback_no_substring_false_match():
+    # "have" must NOT match teams_zoom_av via the 'av' token; no real category word present
+    with patch.object(mob, "get_ai_provider", side_effect=RuntimeError("model down")):
+        cat = await mob.classify("i have a general question")
+    assert cat == "unknown"

frontend/src/api/l1.ts

@@ -1,7 +1,10 @@
 import { apiClient } from './client'
 import type {
   IntakeRequest,
-  IntakeResponse,
+  IntakeResult,
+  L1Categories,
+  NextNodeRequest,
+  NextNodeResult,
   QueueRow,
   WalkSession,
   AdhocNote,
@@ -9,7 +12,23 @@ import type {
 
 export const l1Api = {
   intake: (body: IntakeRequest) =>
-    apiClient.post<IntakeResponse>('/l1/intake', body).then(r => r.data),
+    apiClient.post<IntakeResult>('/l1/intake', body).then(r => r.data),
+
+  nextNode: (sessionId: string, body: NextNodeRequest) =>
+    apiClient
+      .post<NextNodeResult>(`/l1/sessions/${sessionId}/next-node`, body)
+      .then(r => r.data),
+
+  escalations: () =>
+    apiClient.get<WalkSession[]>('/l1/escalations').then(r => r.data),
+
+  getCategories: () =>
+    apiClient.get<L1Categories>('/accounts/me/l1-categories').then(r => r.data),
+
+  setCategories: (enabled: string[]) =>
+    apiClient
+      .patch<L1Categories>('/accounts/me/l1-categories', { enabled })
+      .then(r => r.data),
 
   queue: (statusFilter?: string) =>
     apiClient.get<QueueRow[]>('/l1/queue', {

frontend/src/components/l1/L1EscalationsSection.tsx

@@ -0,0 +1,77 @@
+import { useEffect, useState } from 'react'
+import { l1Api } from '@/api/l1'
+import type { WalkSession } from '@/types/l1'
+
+/**
+ * Engineer-visible list of escalated L1 sessions (the Phase 2A handoff queue).
+ * Backed by GET /l1/escalations (engineer-or-above). Pollable, dependency-free —
+ * each row expands to show the walked path summary. Renders nothing if empty.
+ */
+export function L1EscalationsSection() {
+  const [rows, setRows] = useState<WalkSession[]>([])
+  const [loaded, setLoaded] = useState(false)
+  const [expanded, setExpanded] = useState<string | null>(null)
+
+  useEffect(() => {
+    l1Api
+      .escalations()
+      .then(setRows)
+      .catch(() => setRows([]))
+      .finally(() => setLoaded(true))
+  }, [])
+
+  if (!loaded || rows.length === 0) return null
+
+  return (
+    <section className="space-y-3">
+      <div>
+        <h2 className="font-heading text-lg font-bold text-heading">L1 escalations</h2>
+        <p className="text-sm text-text-muted">
+          Tickets an L1 tech escalated mid-walk — pick one up to continue.
+        </p>
+      </div>
+      <div className="rounded-lg border border-default bg-card overflow-hidden">
+        {rows.map((s) => {
+          const isOpen = expanded === s.id
+          return (
+            <div key={s.id} className="border-b border-default last:border-b-0">
+              <button
+                type="button"
+                onClick={() => setExpanded(isOpen ? null : s.id)}
+                className="w-full px-4 py-3 flex items-center justify-between text-left hover:bg-elevated transition-colors"
+              >
+                <div className="flex items-center gap-3 min-w-0">
+                  <span className="font-mono text-xs text-text-muted">#{s.id.slice(0, 8)}</span>
+                  <span className="text-sm text-text-primary truncate">
+                    {s.walked_path.length} step{s.walked_path.length === 1 ? '' : 's'} walked
+                  </span>
+                </div>
+                <span className="text-xs text-text-muted whitespace-nowrap">
+                  {new Date(s.last_step_at).toLocaleString()}
+                </span>
+              </button>
+              {isOpen && (
+                <div className="px-4 pb-3 space-y-1.5">
+                  {s.walked_path.length === 0 ? (
+                    <p className="text-xs text-text-muted">No steps recorded.</p>
+                  ) : (
+                    <ol className="space-y-1.5 text-sm">
+                      {s.walked_path.map((step, i) => (
+                        <li key={i} className="flex flex-col">
+                          <span className="text-text-muted text-xs">{step.question}</span>
+                          {step.answer && (
+                            <span className="font-medium text-text-primary">→ {step.answer}</span>
+                          )}
+                        </li>
+                      ))}
+                    </ol>
+                  )}
+                </div>
+              )}
+            </div>
+          )
+        })}
+      </div>
+    </section>
+  )
+}

frontend/src/components/l1/L1WalkTreeVariant.tsx

@@ -1,8 +1,8 @@
-import { useState } from 'react'
+import { useState, useEffect, useCallback } from 'react'
 import { ChevronLeft } from 'lucide-react'
 import { Link } from 'react-router-dom'
 import { l1Api } from '@/api/l1'
-import type { WalkSession } from '@/types/l1'
+import type { TreeNode, WalkSession } from '@/types/l1'
 import { EscalateModal, ResolveModal } from '@/components/l1/WalkModals'
 
 interface Props {
@@ -16,6 +16,59 @@ export function L1WalkTreeVariant({ session, onSessionUpdate, onDone }: Props) {
   const [showEscalate, setShowEscalate] = useState(false)
   const [note, setNote] = useState('')
 
+  // Phase 2A: ai_build sessions are walked node-by-node against /next-node
+  // (real AI-generated decision tree), not the synthetic stepping below.
+  const isAiBuild = session.session_kind === 'ai_build'
+  const [node, setNode] = useState<TreeNode | null>(null)
+  const [nodeLoading, setNodeLoading] = useState(false)
+  const [nodeError, setNodeError] = useState<string | null>(null)
+
+  useEffect(() => {
+    if (!isAiBuild || session.status !== 'active') return
+    let cancelled = false
+    setNodeLoading(true)
+    l1Api
+      .nextNode(session.id, {})
+      .then((r) => {
+        if (!cancelled) setNode(r.node)
+      })
+      .catch(() => {
+        if (!cancelled) setNodeError('Could not generate the next step.')
+      })
+      .finally(() => {
+        if (!cancelled) setNodeLoading(false)
+      })
+    return () => {
+      cancelled = true
+    }
+  }, [isAiBuild, session.id, session.status])
+
+  const advanceNode = useCallback(
+    async (body: { answer?: 'yes' | 'no'; acknowledged?: boolean }) => {
+      if (!node) return
+      setNodeLoading(true)
+      setNodeError(null)
+      try {
+        const r = await l1Api.nextNode(session.id, {
+          node_id: node.id,
+          node_text: node.text,
+          ...body,
+        })
+        setNode(r.node)
+      } catch {
+        setNodeError('Could not generate the next step.')
+      } finally {
+        setNodeLoading(false)
+      }
+    },
+    [node, session.id],
+  )
+
+  const isTerminalNode =
+    node?.node_type === 'resolved' ||
+    node?.node_type === 'escalate' ||
+    node?.node_type === 'needs_review'
+
   // Phase 1: we don't have the live flow-tree fetch wired up here yet
   // (the tree-navigation pages have their own loader). The walker shows the
   // walked-path side panel, advance buttons stubbed for now — Phase 2 wires
@@ -55,7 +108,7 @@ export function L1WalkTreeVariant({ session, onSessionUpdate, onDone }: Props) {
         <Link to="/l1" className="flex items-center gap-2 text-muted-foreground hover:text-heading transition-colors">
           <ChevronLeft className="w-4 h-4" />
           <span className="font-mono text-xs">#{session.id.slice(0, 8)}</span>
-          {session.session_kind === 'proposal' && (
+          {(session.session_kind === 'proposal' || session.session_kind === 'ai_build') && (
             <span className="ml-2 text-xs bg-accent/10 text-accent px-2 py-0.5 rounded">AI-built</span>
           )}
         </Link>
@@ -80,6 +133,13 @@ export function L1WalkTreeVariant({ session, onSessionUpdate, onDone }: Props) {
       {/* Two-pane body */}
       <div className="flex-1 flex min-h-0">
         <main className="flex-1 p-6 overflow-y-auto min-h-0">
+          {isAiBuild && (
+            <div className="mb-4 max-w-2xl rounded-md border border-warning/30 bg-warning/10 px-4 py-2 text-xs text-warning">
+              These are high-confidence troubleshooting steps, but they come from
+              outside your organization’s knowledge base — review them before acting.
+              When in doubt, escalate early.
+            </div>
+          )}
           <p className="font-sans text-[0.625rem] uppercase tracking-[0.12em] font-semibold text-muted-foreground mb-2">
             Step {session.walked_path.length + 1}
           </p>
@@ -92,6 +152,66 @@ export function L1WalkTreeVariant({ session, onSessionUpdate, onDone }: Props) {
                 Back to workspace
               </button>
             </div>
+          ) : isAiBuild ? (
+            <div className="rounded-lg border border-default bg-card p-6 max-w-2xl space-y-4">
+              {nodeLoading && (
+                <p className="text-sm text-muted-foreground">Thinking through the next step…</p>
+              )}
+              {nodeError && <p className="text-sm text-danger">{nodeError}</p>}
+
+              {!nodeLoading && node?.node_type === 'question' && (
+                <>
+                  <p className="text-lg">{node.text}</p>
+                  <div className="flex gap-3">
+                    <button
+                      onClick={() => advanceNode({ answer: 'yes' })}
+                      className="flex-1 rounded-md bg-accent text-white py-3 text-base font-medium hover:bg-accent/90 min-h-[44px] transition-colors"
+                    >
+                      Yes
+                    </button>
+                    <button
+                      onClick={() => advanceNode({ answer: 'no' })}
+                      className="flex-1 rounded-md border border-default py-3 text-base font-medium hover:bg-elevated min-h-[44px] transition-colors"
+                    >
+                      No
+                    </button>
+                  </div>
+                </>
+              )}
+
+              {!nodeLoading && node?.node_type === 'instruction' && (
+                <>
+                  <p className="text-lg">{node.text}</p>
+                  <button
+                    onClick={() => advanceNode({ acknowledged: true })}
+                    className="rounded-md bg-accent text-white px-5 py-3 text-base font-medium hover:bg-accent/90 min-h-[44px] transition-colors"
+                  >
+                    Done — next step
+                  </button>
+                </>
+              )}
+
+              {!nodeLoading && isTerminalNode && node && (
+                <>
+                  <p className="text-lg">{node.text}</p>
+                  {node.node_type === 'resolved' ? (
+                    <button
+                      onClick={() => setShowResolve(true)}
+                      className="rounded-md bg-accent text-white px-5 py-3 text-base font-medium hover:bg-accent/90 min-h-[44px] transition-colors"
+                    >
+                      Mark resolved ✓
+                    </button>
+                  ) : (
+                    <button
+                      onClick={() => setShowEscalate(true)}
+                      className="rounded-md bg-warning text-white px-5 py-3 text-base font-medium hover:bg-warning/90 min-h-[44px] transition-colors"
+                    >
+                      Escalate to engineering
+                    </button>
+                  )}
+                </>
+              )}
+            </div>
           ) : (
             <div className="rounded-lg border border-default bg-card p-6 max-w-2xl">
               <p className="text-lg mb-6">Continue the walk:</p>

frontend/src/pages/AccountSettingsPage.tsx

@@ -18,6 +18,7 @@ import {
   RefreshCw,
   Server,
   Shield,
+  Wand2,
   UserCog,
   X,
 } from 'lucide-react'
@@ -662,6 +663,12 @@ export function AccountSettingsPage() {
                 title="Team categories"
                 description="Shared flow categories for your workspace"
               />
+              <SettingsRow
+                to="/account/l1-categories"
+                icon={<Wand2 className="h-4 w-4" />}
+                title="L1 AI build categories"
+                description="Which problem types the L1 assistant may build trees for"
+              />
               <SettingsRow
                 to="/account/target-lists"
                 icon={<Server className="h-4 w-4" />}

frontend/src/pages/account/L1CategoriesPage.tsx

@@ -0,0 +1,98 @@
+import { useEffect, useState } from 'react'
+import { PageMeta } from '@/components/common/PageMeta'
+import { l1Api } from '@/api/l1'
+import { toast } from '@/lib/toast'
+import type { L1Categories } from '@/types/l1'
+
+const prettify = (key: string) =>
+  key.replace(/_/g, ' ').replace(/\b\w/g, (c) => c.toUpperCase())
+
+export default function L1CategoriesPage() {
+  const [data, setData] = useState<L1Categories | null>(null)
+  const [saving, setSaving] = useState<string | null>(null)
+
+  useEffect(() => {
+    l1Api
+      .getCategories()
+      .then(setData)
+      .catch(() => toast.error('Failed to load L1 categories.'))
+  }, [])
+
+  const toggle = async (cat: string) => {
+    if (!data) return
+    const next = data.enabled.includes(cat)
+      ? data.enabled.filter((c) => c !== cat)
+      : [...data.enabled, cat]
+    setSaving(cat)
+    try {
+      const updated = await l1Api.setCategories(next)
+      setData({ ...data, enabled: updated.enabled })
+      toast.success('L1 categories updated.')
+    } catch {
+      toast.error('Could not update categories.')
+    } finally {
+      setSaving(null)
+    }
+  }
+
+  if (!data) {
+    return (
+      <div className="max-w-2xl">
+        <PageMeta title="L1 AI Build Categories" />
+        <p className="text-sm text-muted-foreground">Loading…</p>
+      </div>
+    )
+  }
+
+  return (
+    <div className="max-w-2xl space-y-6">
+      <PageMeta title="L1 AI Build Categories" />
+      <div>
+        <h1 className="font-heading text-2xl font-bold text-heading">
+          L1 AI build categories
+        </h1>
+        <p className="mt-2 text-sm text-muted-foreground">
+          When an L1 tech describes a problem with no matching published flow, the
+          assistant can build a troubleshooting tree on the fly — but only for the
+          categories you enable here. Disabled categories fall back to an ad-hoc walk
+          or escalation.
+        </p>
+      </div>
+
+      <div className="space-y-2">
+        {data.available.map((cat) => {
+          const checked = data.enabled.includes(cat)
+          return (
+            <label
+              key={cat}
+              className="flex items-center gap-3 rounded-md border border-default bg-card px-4 py-3 cursor-pointer hover:bg-elevated transition-colors"
+            >
+              <input
+                type="checkbox"
+                checked={checked}
+                disabled={saving === cat}
+                onChange={() => toggle(cat)}
+                className="h-4 w-4 accent-accent"
+              />
+              <span className="text-sm text-primary">{prettify(cat)}</span>
+            </label>
+          )
+        })}
+      </div>
+
+      <div>
+        <h2 className="font-heading text-sm font-semibold text-heading mb-2">
+          Always excluded (safety)
+        </h2>
+        <p className="text-xs text-muted-foreground mb-2">
+          These action classes are never built automatically and cannot be enabled.
+        </p>
+        <ul className="list-disc pl-5 text-xs text-muted-foreground space-y-1">
+          {data.hard_floor.map((h) => (
+            <li key={h}>{prettify(h)}</li>
+          ))}
+        </ul>
+      </div>
+    </div>
+  )
+}

frontend/src/pages/l1/L1Dashboard.tsx

@@ -6,7 +6,7 @@ import { l1Api } from '@/api/l1'
 import { toast } from '@/lib/toast'
 import { EmptyStateCard } from '@/components/l1/EmptyStateCard'
 import { ResumeInProgress } from '@/components/l1/ResumeInProgress'
-import type { QueueRow } from '@/types/l1'
+import type { NearMiss, QueueRow } from '@/types/l1'
 
 export default function L1Dashboard() {
   const user = useAuthStore((s) => s.user)
@@ -17,6 +17,8 @@ export default function L1Dashboard() {
   const [submitting, setSubmitting] = useState(false)
   const [queue, setQueue] = useState<QueueRow[]>([])
   const [isEmpty, setIsEmpty] = useState(false)
+  const [suggestion, setSuggestion] = useState<NearMiss | null>(null)
+  const [outOfScope, setOutOfScope] = useState<string | null>(null)
 
   useEffect(() => {
     l1Api.queue('open').then(setQueue).catch(() => setQueue([]))
@@ -37,8 +39,42 @@ export default function L1Dashboard() {
     }
   }, [queue])
 
+  const resetPrompts = () => {
+    setSuggestion(null)
+    setOutOfScope(null)
+  }
+
   const handleStart = async () => {
     if (!problem.trim()) return
+    setSubmitting(true)
+    resetPrompts()
+    try {
+      // Phase 2A: intake dispatches via match_or_build and returns an `outcome`.
+      const response = await l1Api.intake({
+        problem_statement: problem.trim(),
+        customer_name: customerName.trim() || undefined,
+        customer_contact: customerContact.trim() || undefined,
+      })
+      if (response.outcome === 'matched' || response.outcome === 'build') {
+        navigate(`/l1/walk/${response.session_id}`)
+      } else if (response.outcome === 'suggest') {
+        setSuggestion(response.near_miss ?? null)
+      } else if (response.outcome === 'out_of_scope') {
+        setOutOfScope(response.category ?? 'unknown')
+      }
+    } catch (err) {
+      const detail = (err as { response?: { data?: { detail?: string } } }).response?.data?.detail
+      const msg =
+        typeof detail === 'string' ? detail : 'Failed to start walk. Try again.'
+      toast.error(msg)
+    } finally {
+      setSubmitting(false)
+    }
+  }
+
+  // "Use this flow" — re-run intake with the same text; it matches again and
+  // returns a `matched` outcome with a started flow session (acceptable Phase 2A).
+  const useSuggestedFlow = async () => {
     setSubmitting(true)
     try {
       const response = await l1Api.intake({
@@ -46,12 +82,54 @@ export default function L1Dashboard() {
         customer_name: customerName.trim() || undefined,
         customer_contact: customerContact.trim() || undefined,
       })
-      navigate(`/l1/walk/${response.session_id}`)
-    } catch (err) {
-      const detail = (err as { response?: { data?: { detail?: string } } }).response?.data?.detail
-      const msg =
-        typeof detail === 'string' ? detail : 'Failed to start walk. Try again.'
-      toast.error(msg)
+      if (response.session_id) navigate(`/l1/walk/${response.session_id}`)
+      else resetPrompts()
+    } catch {
+      toast.error('Could not start the matched flow. Try again.')
+    } finally {
+      setSubmitting(false)
+    }
+  }
+
+  // "Build new" — skip the match pass (force_build); still gated by enabled categories.
+  const buildNew = async () => {
+    setSubmitting(true)
+    resetPrompts()
+    try {
+      const response = await l1Api.intake({
+        problem_statement: problem.trim(),
+        customer_name: customerName.trim() || undefined,
+        customer_contact: customerContact.trim() || undefined,
+        force_build: true,
+      })
+      if (response.outcome === 'build' && response.session_id) {
+        navigate(`/l1/walk/${response.session_id}`)
+      } else if (response.outcome === 'out_of_scope') {
+        setOutOfScope(response.category ?? 'unknown')
+      }
+    } catch {
+      toast.error('Failed to start walk. Try again.')
+    } finally {
+      setSubmitting(false)
+    }
+  }
+
+  // out-of-scope fallback: escalate straight to engineering (no walk).
+  const escalateOutOfScope = async () => {
+    if (!problem.trim()) return
+    setSubmitting(true)
+    try {
+      const session = await l1Api.escalateWithoutWalk({
+        problem_statement: problem.trim(),
+        customer_name: customerName.trim() || undefined,
+        customer_contact: customerContact.trim() || undefined,
+        reason_category: 'out_of_scope',
+        reason: 'Problem is outside the enabled L1 AI-build categories.',
+      })
+      toast.success('Escalated to engineering.')
+      navigate(`/l1/walk/${session.id}`)
+    } catch {
+      toast.error('Could not escalate. Try again.')
     } finally {
       setSubmitting(false)
     }
@@ -160,6 +238,63 @@ export default function L1Dashboard() {
           </section>
         )}
 
+        {/* Suggest: near-miss flow found */}
+        {suggestion && (
+          <div className="space-y-3 rounded-lg border border-default bg-card p-4">
+            <p className="text-sm text-primary">
+              Found a similar flow: <strong>{suggestion.flow_name}</strong>. Use it, or
+              build a new troubleshooting tree for this problem?
+            </p>
+            <div className="flex gap-2">
+              <button
+                type="button"
+                onClick={useSuggestedFlow}
+                disabled={submitting}
+                className="rounded-md bg-accent text-white px-4 py-2 text-sm font-medium hover:bg-accent/90 transition-colors disabled:opacity-50"
+              >
+                Use this flow
+              </button>
+              <button
+                type="button"
+                onClick={buildNew}
+                disabled={submitting}
+                className="rounded-md border border-default px-4 py-2 text-sm hover:bg-elevated transition-colors disabled:opacity-50"
+              >
+                Build new
+              </button>
+            </div>
+          </div>
+        )}
+
+        {/* Out of scope: category disabled/unknown */}
+        {outOfScope && (
+          <div className="space-y-3 rounded-lg border border-default bg-card p-4">
+            <p className="text-sm text-primary">
+              This problem isn’t in your account’s enabled L1 categories
+              {outOfScope !== 'unknown' ? ` (${outOfScope.replace(/_/g, ' ')})` : ''}, so
+              there’s no AI-built walk for it. You can escalate it to engineering.
+            </p>
+            <div className="flex gap-2">
+              <button
+                type="button"
+                onClick={escalateOutOfScope}
+                disabled={submitting}
+                className="rounded-md bg-accent text-white px-4 py-2 text-sm font-medium hover:bg-accent/90 transition-colors disabled:opacity-50"
+              >
+                Escalate to engineering
+              </button>
+              <button
+                type="button"
+                onClick={resetPrompts}
+                disabled={submitting}
+                className="rounded-md border border-default px-4 py-2 text-sm hover:bg-elevated transition-colors disabled:opacity-50"
+              >
+                Cancel
+              </button>
+            </div>
+          </div>
+        )}
+
         {/* Resume in progress */}
         <ResumeInProgress />
       </div>

frontend/src/router.tsx

@@ -114,6 +114,7 @@ const IntegrationsPage = lazyWithRetry(() => import('@/pages/account/Integration
 const BrandingSettingsPage = lazyWithRetry(() => import('@/pages/account/BrandingSettingsPage'))
 const BillingPage = lazyWithRetry(() => import('@/pages/account/BillingPage'))
 const SelectPlanPage = lazyWithRetry(() => import('@/pages/account/SelectPlanPage'))
+const L1CategoriesPage = lazyWithRetry(() => import('@/pages/account/L1CategoriesPage'))
 
 /** Wraps a lazy-loaded page with Suspense + ErrorBoundary */
 function page(Component: React.LazyExoticComponent<React.ComponentType>) {
@@ -363,6 +364,14 @@ export const router = sentryCreateBrowserRouter([
               </ProtectedRoute>
             ),
           },
+          {
+            path: 'l1-categories',
+            element: (
+              <ProtectedRoute requiredRole="owner">
+                {page(L1CategoriesPage)}
+              </ProtectedRoute>
+            ),
+          },
           {
             path: 'chat-retention',
             element: (

frontend/src/types/flow-proposal.ts

@@ -10,7 +10,10 @@ export interface FlowProposalSummary {
   supporting_session_count: number
   status: 'pending' | 'approved' | 'modified' | 'rejected' | 'dismissed' | 'auto_reinforced'
   target_flow_id: string | null
-  source_session_id: string
+  // Exactly one source is set: source_session_id (FlowPilot ai_session) XOR
+  // l1_session_id (L1 ai_build walk). Both nullable on the backend (Phase 2A).
+  source_session_id: string | null
+  l1_session_id: string | null
   created_at: string
 }
 

frontend/src/types/l1.ts

@@ -1,4 +1,4 @@
-export type SessionKind = 'flow' | 'proposal' | 'adhoc'
+export type SessionKind = 'flow' | 'proposal' | 'adhoc' | 'ai_build'
 export type SessionStatus = 'active' | 'resolved' | 'escalated' | 'abandoned'
 export type TicketKind = 'psa' | 'internal'
 
@@ -42,11 +42,53 @@ export interface IntakeRequest {
   customer_name?: string
   customer_contact?: string
   flow_id?: string
+  force_build?: boolean
 }
 
-export interface IntakeResponse {
-  session_id: string
-  session_kind: SessionKind
-  ticket_id: string
-  ticket_kind: TicketKind
+export type IntakeOutcome = 'matched' | 'suggest' | 'out_of_scope' | 'build'
+
+export interface NearMiss {
+  flow_id: string
+  flow_name: string
+  score: number
+}
+
+/** Phase 2A intake response — `outcome` drives the frontend dispatch.
+ *  Session fields are present only for `matched` / `build`. */
+export interface IntakeResult {
+  outcome: IntakeOutcome
+  session_id?: string
+  session_kind?: SessionKind
+  ticket_id?: string
+  ticket_kind?: TicketKind
+  flow_id?: string        // for 'matched'
+  near_miss?: NearMiss     // for 'suggest'
+  category?: string        // for 'out_of_scope'
+}
+
+/** A single node of an AI-built decision tree, returned by /next-node. */
+export type TreeNode =
+  | { node_type: 'question'; id: string; text: string }
+  | { node_type: 'instruction'; id: string; text: string }
+  | { node_type: 'resolved'; id: string; text: string }
+  | { node_type: 'escalate'; id: string; reason_category?: string; text: string }
+  | { node_type: 'needs_review'; id: string; text: string }
+
+export interface NextNodeRequest {
+  node_id?: string
+  node_text?: string       // rendered text of the node being answered
+  answer?: 'yes' | 'no'
+  acknowledged?: boolean
+  note?: string
+}
+
+export interface NextNodeResult {
+  node: TreeNode
+  session_status: string
+}
+
+export interface L1Categories {
+  enabled: string[]
+  available: string[]
+  hard_floor: string[]
 }