Run heavy CLI dumps on a dedicated long-timeout SSH lane.

Split interface detail/route-class collects from the light lane so they no longer block other checks, and scope Collect Now busy state per task.

Co-authored-by: Cursor <cursoragent@cursor.com>

netx_api/biz_state/collect_runner.py

@@ -3,6 +3,8 @@
 from __future__ import annotations
 
 import logging
+import threading
+from concurrent.futures import ThreadPoolExecutor
 from datetime import datetime
 from typing import Any
 from uuid import uuid4
@@ -39,6 +41,42 @@ from .profiles import get_profile
 
 _log = logging.getLogger("netx.biz_state.runner")
 
+# concrete, params, profile_id, item_id, mode
+WorkItem = tuple[str, dict[str, str], str, str, str]
+
+_heavy_pool_lock = threading.Lock()
+_heavy_pool: ThreadPoolExecutor | None = None
+
+
+def _heavy_cli_pool() -> ThreadPoolExecutor:
+    global _heavy_pool
+    with _heavy_pool_lock:
+        if _heavy_pool is None:
+            n = max(1, int(getattr(settings, "biz_state_heavy_workers", 4) or 4))
+            _heavy_pool = ThreadPoolExecutor(max_workers=n, thread_name_prefix="biz-heavy")
+        return _heavy_pool
+
+
+def work_item_lane(profile_id: str, mode: str) -> str:
+    """Return collect lane for a work item (custom_raw always light)."""
+    if str(mode or "").strip() == "custom":
+        return "light"
+    profile = get_profile(profile_id) or _resolve_collect_profile(profile_id)
+    lane = str(getattr(profile, "collect_lane", "light") or "light").strip().lower() if profile else "light"
+    return "heavy" if lane == "heavy" else "light"
+
+
+def partition_work(work: list[WorkItem]) -> tuple[list[WorkItem], list[WorkItem]]:
+    """Split work into (light, heavy) lanes."""
+    light: list[WorkItem] = []
+    heavy: list[WorkItem] = []
+    for item in work:
+        if work_item_lane(item[2], item[4]) == "heavy":
+            heavy.append(item)
+        else:
+            light.append(item)
+    return light, heavy
+
 
 def _utcnow() -> datetime:
     return datetime.utcnow()
@@ -311,6 +349,307 @@ def dispatch_collect(task_id: str, *, manual: bool = False) -> None:
         _finish_task(task_id, error=error)
 
 
+def _run_collect_lane(
+    *,
+    work: list[WorkItem],
+    batch_id: str,
+    creds: dict[str, Any],
+    vendor_eff: str,
+    device_type_eff: str,
+    vendor_key: str,
+    per_cmd: int,
+    cap: int,
+    label: str,
+) -> tuple[int, int, bool, bool]:
+    """Run one SSH lane (own connection + CollectSession + timeout budget)."""
+    if not work:
+        return 0, 0, False, False
+
+    budget = min(int(cap), int(per_cmd) * max(1, len(work)) + 90)
+    holder: dict[str, Any] = {}
+
+    def _session() -> tuple[int, int, bool, bool]:
+        from ..ne_netmiko import drain_read_channel
+
+        conn = open_netmiko_connection(creds, session_timeout=budget)
+        holder["conn"] = conn
+        total_rows = 0
+        cmd_count = 0
+        any_fail = False
+        any_ok = False
+        try:
+            try:
+                disable_target_paging(
+                    conn,
+                    vendor=str(creds.get("vendor") or vendor_eff or ""),
+                    device_type=str(creds.get("device_type") or device_type_eff or ""),
+                )
+            except Exception:
+                pass
+            try:
+                drain_read_channel(conn)
+            except Exception:
+                pass
+
+            sdb = SessionLocal()
+            session = CollectSession(
+                conn,
+                vendor=vendor_eff,
+                device_type=device_type_eff,
+                vendor_key=vendor_key,
+                read_timeout=per_cmd,
+            )
+            try:
+                batch_row = sdb.get(BizStateBatch, batch_id)
+                if not batch_row:
+                    return 0, 0, True, False
+                for concrete, params, profile_id, item_id, mode in work:
+                    if holder.get("timed_out"):
+                        raise TimeoutError(f"{label}_aborted")
+                    cmd_count += 1
+                    cmd_row = BizStateBatchCommand(
+                        id=uuid4().hex,
+                        batch_id=batch_id,
+                        task_item_id=item_id,
+                        profile_id=profile_id,
+                        raw_command=concrete[:512],
+                        params_json=dict(params or {}),
+                        created_at=_utcnow(),
+                    )
+                    cache_hit_primary = False
+                    try:
+                        cached = session.get_cached(concrete)
+                        if cached is not None and str(cached.raw or "").strip():
+                            cmd_row.raw_text = str(cached.raw or "")
+                            cache_hit_primary = True
+                        else:
+                            raw = send_show_command(conn, concrete, read_timeout=per_cmd)
+                            cmd_row.raw_text = str(raw or "")
+                    except Exception as exc:
+                        any_fail = True
+                        cmd_row.parse_status = "failed"
+                        cmd_row.message = _format_error(exc)
+                        sdb.add(cmd_row)
+                        sdb.commit()
+                        continue
+
+                    if mode == "custom":
+                        cmd_row.parse_status = "skipped_custom"
+                        cmd_row.message = "custom_raw"
+                        sdb.add(cmd_row)
+                        sdb.commit()
+                        any_ok = True
+                        continue
+
+                    hit = match_command(vendor_key=vendor_key, command=concrete)
+                    if not hit:
+                        any_fail = True
+                        cmd_row.parse_status = "unmatched"
+                        cmd_row.message = "no profile matched concrete command"
+                        sdb.add(cmd_row)
+                        sdb.commit()
+                        continue
+
+                    cmd_row.profile_id = hit.profile.profile_id
+                    cmd_row.parser_id = hit.profile.parser_id
+                    cmd_row.metric_id = hit.profile.metric_id
+                    merged = {**params, **hit.params}
+                    cmd_row.params_json = merged
+
+                    if not get_parser(hit.profile.parser_id):
+                        any_fail = True
+                        cmd_row.parse_status = "failed"
+                        cmd_row.message = f"unknown parser {hit.profile.parser_id}"
+                        sdb.add(cmd_row)
+                        sdb.commit()
+                        continue
+
+                    if cache_hit_primary:
+                        pass
+                    else:
+                        session.remember(
+                            concrete,
+                            raw=cmd_row.raw_text or "",
+                            ok=True,
+                            cmd_row_id=cmd_row.id,
+                        )
+
+                    resolved_aux = []
+                    aux_results: dict[str, Any] = {}
+                    for aux in list(hit.profile.aux_commands or []):
+                        try:
+                            ra = resolve_aux_command(aux, params=merged)
+                        except ValueError as exc:
+                            aux_row = BizStateBatchCommand(
+                                id=uuid4().hex,
+                                batch_id=batch_id,
+                                task_item_id=item_id,
+                                profile_id=str(aux.profile_id or "")[:128],
+                                raw_command=str(aux.key or "")[:512],
+                                params_json={},
+                                parse_status="aux_failed",
+                                message=f"aux_for={cmd_row.id};resolve:{exc}"[:1020],
+                                created_at=_utcnow(),
+                            )
+                            cmd_count += 1
+                            sdb.add(aux_row)
+                            sdb.commit()
+                            continue
+                        resolved_aux.append(ra)
+                        aux_row = BizStateBatchCommand(
+                            id=uuid4().hex,
+                            batch_id=batch_id,
+                            task_item_id=item_id,
+                            profile_id=ra.profile_id,
+                            parser_id=ra.parser_id,
+                            metric_id="",
+                            raw_command=ra.command[:512],
+                            params_json={},
+                            created_at=_utcnow(),
+                        )
+                        cmd_count += 1
+                        entry, cache_hit = session.fetch_and_parse(
+                            ra.command,
+                            parser_id=ra.parser_id,
+                            textfsm_command=ra.textfsm_command,
+                            params=merged,
+                            cmd_row_id=aux_row.id,
+                        )
+                        aux_results[ra.key] = entry
+                        if cache_hit:
+                            aux_row.parse_status = "aux_cached"
+                            aux_row.message = (
+                                f"aux_for={cmd_row.id};cache_hit;src={entry.cmd_row_id}"
+                            )[:1020]
+                            aux_row.raw_text = ""
+                            aux_row.row_count = len(entry.records or [])
+                        elif not entry.ok:
+                            aux_row.parse_status = "aux_failed"
+                            aux_row.message = (
+                                f"aux_for={cmd_row.id};{entry.error}"
+                            )[:1020]
+                            aux_row.raw_text = entry.raw
+                        else:
+                            aux_row.parse_status = "aux"
+                            aux_row.message = f"aux_for={cmd_row.id}"[:1020]
+                            aux_row.raw_text = entry.raw
+                            aux_row.row_count = len(entry.records or [])
+                            entry.cmd_row_id = aux_row.id
+                        sdb.add(aux_row)
+                        sdb.commit()
+
+                    bundle = build_parse_bundle(
+                        primary_raw=cmd_row.raw_text or "",
+                        primary_parser_id=hit.profile.parser_id,
+                        aux_results=aux_results,
+                        resolved_aux=resolved_aux,
+                    )
+                    try:
+                        records, fsm_tables, rule_keys = run_primary_with_bundle(
+                            hit.profile.parser_id,
+                            bundle=bundle,
+                            vendor=vendor_eff,
+                            device_type=device_type_eff,
+                            command=hit.profile.textfsm_command or concrete,
+                            textfsm_command=hit.profile.textfsm_command or "",
+                            params=merged,
+                            enrich_joins=list(hit.profile.enrich_joins or []),
+                        )
+                        session.remember(
+                            concrete,
+                            raw=cmd_row.raw_text or "",
+                            fsm_tables=fsm_tables,
+                            records=records,
+                            ok=True,
+                            cmd_row_id=cmd_row.id,
+                        )
+                        hints = []
+                        if rule_keys:
+                            nonempty = [k for k in rule_keys if fsm_tables.get(k)]
+                            hints.append(
+                                f"fsm_keys={','.join(rule_keys)};hit={','.join(nonempty)}"
+                            )
+                        if bundle.aux_records:
+                            hints.append(
+                                "aux="
+                                + ",".join(
+                                    f"{k}:{len(v)}" for k, v in bundle.aux_records.items()
+                                )
+                            )
+                        if hit.profile.enrich_joins:
+                            hints.append(
+                                "enrich="
+                                + ",".join(j.from_aux for j in hit.profile.enrich_joins)
+                            )
+                        if hints:
+                            cmd_row.message = ";".join(hints)[:1020]
+                    except Exception as exc:
+                        any_fail = True
+                        cmd_row.parse_status = "failed"
+                        cmd_row.message = f"parse: {_format_error(exc)}"
+                        sdb.add(cmd_row)
+                        sdb.commit()
+                        continue
+
+                    n = 0
+                    if hit.profile.metric_id == "lldp_neighbor":
+                        n = _persist_lldp_rows(
+                            sdb, batch=batch_row, cmd_row=cmd_row, records=records
+                        )
+                    elif hit.profile.metric_id in _GENERIC_METRICS:
+                        n = _persist_metric_rows(
+                            sdb,
+                            batch=batch_row,
+                            cmd_row=cmd_row,
+                            metric_id=hit.profile.metric_id,
+                            records=records,
+                        )
+                    cmd_row.row_count = n
+                    cmd_row.parse_status = "ok"
+                    total_rows += n
+                    any_ok = True
+                    sdb.add(cmd_row)
+                    sdb.commit()
+            finally:
+                sdb.close()
+            return total_rows, cmd_count, any_fail, any_ok
+        finally:
+            holder.pop("conn", None)
+            close_netmiko_connection(conn)
+
+    try:
+        return run_cli_with_timeout(
+            _session,
+            timeout_sec=budget,
+            conn_holder=holder,
+            label=label,
+            acquire_budget=True,
+        )
+    except TimeoutError as exc:
+        raise RuntimeError(str(exc)[:1020]) from exc
+
+
+def _absorb_lane_result(
+    result: tuple[int, int, bool, bool] | BaseException,
+    *,
+    total_rows: int,
+    cmd_count: int,
+    any_fail: bool,
+    any_ok: bool,
+    lane_errors: list[str],
+) -> tuple[int, int, bool, bool]:
+    if isinstance(result, BaseException):
+        lane_errors.append(_format_error(result))
+        return total_rows, cmd_count, True, any_ok
+    rows, cmds, fail, ok = result
+    return (
+        total_rows + int(rows or 0),
+        cmd_count + int(cmds or 0),
+        any_fail or bool(fail),
+        any_ok or bool(ok),
+    )
+
+
 def _run_collect_session(
     *,
     task_id: str,
@@ -320,8 +659,10 @@ def _run_collect_session(
     vendor: str,
     device_type: str,
 ) -> None:
-    per_cmd = int(settings.ne_collect_read_timeout_sec or 120)
+    light_per = int(settings.ne_collect_read_timeout_sec or 120)
-    cap = int(settings.ne_collect_run_timeout_cap_sec or 600)
+    light_cap = int(settings.ne_collect_run_timeout_cap_sec or 600)
+    heavy_per = int(getattr(settings, "biz_state_heavy_read_timeout_sec", 300) or 300)
+    heavy_cap = int(getattr(settings, "biz_state_heavy_run_timeout_cap_sec", 900) or 900)
 
     db = SessionLocal()
     try:
@@ -365,8 +706,7 @@ def _run_collect_session(
         )
 
         # Build work list before opening session
-        work: list[tuple[str, dict[str, str], str, str, str]] = []
+        work: list[WorkItem] = []
-        # concrete, params, profile_id, item_id, mode
         # Dedupe same CLI → same metric (e.g. legacy if_intf + config_interface).
         seen_work: set[tuple[str, str]] = set()
         for item in items:
@@ -417,270 +757,94 @@ def _run_collect_session(
             db.commit()
             raise RuntimeError("no commands to run")
 
-        budget = min(cap, per_cmd * max(1, len(work)) + 90)
+        light_work, heavy_work = partition_work(work)
-        holder: dict[str, Any] = {}
+        lane_kwargs = dict(
+            batch_id=batch_id,
+            creds=creds,
+            vendor_eff=vendor_eff,
+            device_type_eff=device_type_eff,
+            vendor_key=vendor_key,
+        )
 
-        def _session() -> tuple[int, int, bool, bool]:
+        def _run_light() -> tuple[int, int, bool, bool]:
-            from ..ne_netmiko import drain_read_channel
+            return _run_collect_lane(
-
+                work=light_work,
-            conn = open_netmiko_connection(creds, session_timeout=budget)
+                per_cmd=light_per,
-            holder["conn"] = conn
+                cap=light_cap,
-            total_rows = 0
+                label="biz_state_light",
-            cmd_count = 0
+                **lane_kwargs,
-            any_fail = False
-            any_ok = False
-            try:
-                try:
-                    disable_target_paging(
-                        conn,
-                        vendor=str(creds.get("vendor") or vendor_eff or ""),
-                        device_type=str(creds.get("device_type") or device_type_eff or ""),
-                    )
-                except Exception:
-                    pass
-                try:
-                    drain_read_channel(conn)
-                except Exception:
-                    pass
-
-                sdb = SessionLocal()
-                session = CollectSession(
-                    conn,
-                    vendor=vendor_eff,
-                    device_type=device_type_eff,
-                    vendor_key=vendor_key,
-                    read_timeout=per_cmd,
-                )
-                try:
-                    batch_row = sdb.get(BizStateBatch, batch_id)
-                    if not batch_row:
-                        return 0, 0, True, False
-                    for concrete, params, profile_id, item_id, mode in work:
-                        if holder.get("timed_out"):
-                            raise TimeoutError("biz_state_aborted")
-                        cmd_count += 1
-                        cmd_row = BizStateBatchCommand(
-                            id=uuid4().hex,
-                            batch_id=batch_id,
-                            task_item_id=item_id,
-                            profile_id=profile_id,
-                            raw_command=concrete[:512],
-                            params_json=dict(params or {}),
-                            created_at=_utcnow(),
-                        )
-                        cache_hit_primary = False
-                        try:
-                            cached = session.get_cached(concrete)
-                            if cached is not None and str(cached.raw or "").strip():
-                                cmd_row.raw_text = str(cached.raw or "")
-                                cache_hit_primary = True
-                            else:
-                                raw = send_show_command(conn, concrete, read_timeout=per_cmd)
-                                cmd_row.raw_text = str(raw or "")
-                        except Exception as exc:
-                            any_fail = True
-                            cmd_row.parse_status = "failed"
-                            cmd_row.message = _format_error(exc)
-                            sdb.add(cmd_row)
-                            sdb.commit()
-                            continue
-
-                        if mode == "custom":
-                            cmd_row.parse_status = "skipped_custom"
-                            cmd_row.message = "custom_raw"
-                            sdb.add(cmd_row)
-                            sdb.commit()
-                            any_ok = True
-                            continue
-
-                        hit = match_command(vendor_key=vendor_key, command=concrete)
-                        if not hit:
-                            any_fail = True
-                            cmd_row.parse_status = "unmatched"
-                            cmd_row.message = "no profile matched concrete command"
-                            sdb.add(cmd_row)
-                            sdb.commit()
-                            continue
-
-                        cmd_row.profile_id = hit.profile.profile_id
-                        cmd_row.parser_id = hit.profile.parser_id
-                        cmd_row.metric_id = hit.profile.metric_id
-                        merged = {**params, **hit.params}
-                        cmd_row.params_json = merged
-
-                        if not get_parser(hit.profile.parser_id):
-                            any_fail = True
-                            cmd_row.parse_status = "failed"
-                            cmd_row.message = f"unknown parser {hit.profile.parser_id}"
-                            sdb.add(cmd_row)
-                            sdb.commit()
-                            continue
-
-                        if cache_hit_primary:
-                            # Keep existing cache entry; primary will re-parse from shared raw.
-                            pass
-                        else:
-                            session.remember(
-                                concrete,
-                                raw=cmd_row.raw_text or "",
-                                ok=True,
-                                cmd_row_id=cmd_row.id,
-                            )
-
-                        resolved_aux = []
-                        aux_results: dict[str, Any] = {}
-                        for aux in list(hit.profile.aux_commands or []):
-                            try:
-                                ra = resolve_aux_command(aux, params=merged)
-                            except ValueError as exc:
-                                aux_row = BizStateBatchCommand(
-                                    id=uuid4().hex,
-                                    batch_id=batch_id,
-                                    task_item_id=item_id,
-                                    profile_id=str(aux.profile_id or "")[:128],
-                                    raw_command=str(aux.key or "")[:512],
-                                    params_json={},
-                                    parse_status="aux_failed",
-                                    message=f"aux_for={cmd_row.id};resolve:{exc}"[:1020],
-                                    created_at=_utcnow(),
-                                )
-                                cmd_count += 1
-                                sdb.add(aux_row)
-                                sdb.commit()
-                                continue
-                            resolved_aux.append(ra)
-                            aux_row = BizStateBatchCommand(
-                                id=uuid4().hex,
-                                batch_id=batch_id,
-                                task_item_id=item_id,
-                                profile_id=ra.profile_id,
-                                parser_id=ra.parser_id,
-                                metric_id="",
-                                raw_command=ra.command[:512],
-                                params_json={},
-                                created_at=_utcnow(),
-                            )
-                            cmd_count += 1
-                            entry, cache_hit = session.fetch_and_parse(
-                                ra.command,
-                                parser_id=ra.parser_id,
-                                textfsm_command=ra.textfsm_command,
-                                params=merged,
-                                cmd_row_id=aux_row.id,
-                            )
-                            aux_results[ra.key] = entry
-                            if cache_hit:
-                                aux_row.parse_status = "aux_cached"
-                                aux_row.message = (
-                                    f"aux_for={cmd_row.id};cache_hit;src={entry.cmd_row_id}"
-                                )[:1020]
-                                aux_row.raw_text = ""
-                                aux_row.row_count = len(entry.records or [])
-                            elif not entry.ok:
-                                aux_row.parse_status = "aux_failed"
-                                aux_row.message = (
-                                    f"aux_for={cmd_row.id};{entry.error}"
-                                )[:1020]
-                                aux_row.raw_text = entry.raw
-                            else:
-                                aux_row.parse_status = "aux"
-                                aux_row.message = f"aux_for={cmd_row.id}"[:1020]
-                                aux_row.raw_text = entry.raw
-                                aux_row.row_count = len(entry.records or [])
-                                # refresh cache row id to this aux row on first fetch
-                                entry.cmd_row_id = aux_row.id
-                            sdb.add(aux_row)
-                            sdb.commit()
-
-                        bundle = build_parse_bundle(
-                            primary_raw=cmd_row.raw_text or "",
-                            primary_parser_id=hit.profile.parser_id,
-                            aux_results=aux_results,
-                            resolved_aux=resolved_aux,
-                        )
-                        try:
-                            records, fsm_tables, rule_keys = run_primary_with_bundle(
-                                hit.profile.parser_id,
-                                bundle=bundle,
-                                vendor=vendor_eff,
-                                device_type=device_type_eff,
-                                command=hit.profile.textfsm_command or concrete,
-                                textfsm_command=hit.profile.textfsm_command or "",
-                                params=merged,
-                                enrich_joins=list(hit.profile.enrich_joins or []),
-                            )
-                            session.remember(
-                                concrete,
-                                raw=cmd_row.raw_text or "",
-                                fsm_tables=fsm_tables,
-                                records=records,
-                                ok=True,
-                                cmd_row_id=cmd_row.id,
-                            )
-                            hints = []
-                            if rule_keys:
-                                nonempty = [k for k in rule_keys if fsm_tables.get(k)]
-                                hints.append(
-                                    f"fsm_keys={','.join(rule_keys)};hit={','.join(nonempty)}"
-                                )
-                            if bundle.aux_records:
-                                hints.append(
-                                    "aux="
-                                    + ",".join(
-                                        f"{k}:{len(v)}" for k, v in bundle.aux_records.items()
-                                    )
-                                )
-                            if hit.profile.enrich_joins:
-                                hints.append(
-                                    "enrich="
-                                    + ",".join(j.from_aux for j in hit.profile.enrich_joins)
-                                )
-                            if hints:
-                                cmd_row.message = ";".join(hints)[:1020]
-                        except Exception as exc:
-                            any_fail = True
-                            cmd_row.parse_status = "failed"
-                            cmd_row.message = f"parse: {_format_error(exc)}"
-                            sdb.add(cmd_row)
-                            sdb.commit()
-                            continue
-
-                        n = 0
-                        if hit.profile.metric_id == "lldp_neighbor":
-                            n = _persist_lldp_rows(
-                                sdb, batch=batch_row, cmd_row=cmd_row, records=records
-                            )
-                        elif hit.profile.metric_id in _GENERIC_METRICS:
-                            n = _persist_metric_rows(
-                                sdb,
-                                batch=batch_row,
-                                cmd_row=cmd_row,
-                                metric_id=hit.profile.metric_id,
-                                records=records,
-                            )
-                        cmd_row.row_count = n
-                        cmd_row.parse_status = "ok"
-                        total_rows += n
-                        any_ok = True
-                        sdb.add(cmd_row)
-                        sdb.commit()
-                finally:
-                    sdb.close()
-                return total_rows, cmd_count, any_fail, any_ok
-            finally:
-                holder.pop("conn", None)
-                close_netmiko_connection(conn)
-
-        try:
-            total_rows, cmd_count, any_fail, any_ok = run_cli_with_timeout(
-                _session,
-                timeout_sec=budget,
-                conn_holder=holder,
-                label="biz_state",
-                acquire_budget=True,
             )
-        except TimeoutError as exc:
+
-            raise RuntimeError(str(exc)[:1020]) from exc
+        def _run_heavy() -> tuple[int, int, bool, bool]:
+            return _run_collect_lane(
+                work=heavy_work,
+                per_cmd=heavy_per,
+                cap=heavy_cap,
+                label="biz_state_heavy",
+                **lane_kwargs,
+            )
+
+        total_rows = 0
+        cmd_count = 0
+        any_fail = False
+        any_ok = False
+        lane_errors: list[str] = []
+
+        if light_work and heavy_work:
+            heavy_fut = _heavy_cli_pool().submit(_run_heavy)
+            light_res: tuple[int, int, bool, bool] | BaseException
+            try:
+                light_res = _run_light()
+            except Exception as exc:
+                light_res = exc
+            heavy_res: tuple[int, int, bool, bool] | BaseException
+            try:
+                heavy_res = heavy_fut.result()
+            except Exception as exc:
+                heavy_res = exc
+            total_rows, cmd_count, any_fail, any_ok = _absorb_lane_result(
+                light_res,
+                total_rows=total_rows,
+                cmd_count=cmd_count,
+                any_fail=any_fail,
+                any_ok=any_ok,
+                lane_errors=lane_errors,
+            )
+            total_rows, cmd_count, any_fail, any_ok = _absorb_lane_result(
+                heavy_res,
+                total_rows=total_rows,
+                cmd_count=cmd_count,
+                any_fail=any_fail,
+                any_ok=any_ok,
+                lane_errors=lane_errors,
+            )
+        elif heavy_work:
+            try:
+                total_rows, cmd_count, any_fail, any_ok = _run_heavy()
+            except Exception as exc:
+                total_rows, cmd_count, any_fail, any_ok = _absorb_lane_result(
+                    exc,
+                    total_rows=0,
+                    cmd_count=0,
+                    any_fail=False,
+                    any_ok=False,
+                    lane_errors=lane_errors,
+                )
+        else:
+            try:
+                total_rows, cmd_count, any_fail, any_ok = _run_light()
+            except Exception as exc:
+                total_rows, cmd_count, any_fail, any_ok = _absorb_lane_result(
+                    exc,
+                    total_rows=0,
+                    cmd_count=0,
+                    any_fail=False,
+                    any_ok=False,
+                    lane_errors=lane_errors,
+                )
+
+        if lane_errors and not any_ok and cmd_count == 0:
+            raise RuntimeError("; ".join(lane_errors)[:1020])
 
         batch = db.get(BizStateBatch, batch_id)
         if batch:
@@ -689,11 +853,16 @@ def _run_collect_session(
             batch.ended_at = _utcnow()
             if any_fail and any_ok:
                 batch.status = "partial"
+                if lane_errors:
+                    batch.message = "; ".join(lane_errors)[:1020]
             elif any_fail and not any_ok:
                 batch.status = "failed"
-                batch.message = "all commands failed"
+                batch.message = (
+                    "; ".join(lane_errors)[:1020] if lane_errors else "all commands failed"
+                )
             else:
                 batch.status = "success"
+                batch.message = ""
             db.commit()
             if batch.status in ("success", "partial"):
                 try:

netx_api/biz_state/profiles.py

@@ -64,6 +64,8 @@ class ParseProfile:
     kind: str = "collect"  # collect | discover
     aux_commands: list[AuxCommand] = field(default_factory=list)
     enrich_joins: list[EnrichJoin] = field(default_factory=list)
+    # light: default shared SSH lane; heavy: dedicated long-timeout connection
+    collect_lane: str = "light"
 
 
 _LLDP_FIELDS: list[FieldDef] = [
@@ -1270,11 +1272,32 @@ def _zte_status_profiles() -> list[ParseProfile]:
 
 _PROFILES: list[ParseProfile] | None = None
 
+# Huge CLI dumps: dedicated heavy lane (longer read_timeout / own SSH).
+_HEAVY_LANE_METRICS = frozenset(
+    {
+        "interface_detail",
+        "ip_route",
+        "ipv6_route",
+        "bgp_route",
+        "l2vpn_mac",
+        "evpn_mac",
+    }
+)
+
+
+def _apply_collect_lanes(profiles: list[ParseProfile]) -> list[ParseProfile]:
+    for p in profiles:
+        if str(p.metric_id or "").strip() in _HEAVY_LANE_METRICS:
+            p.collect_lane = "heavy"
+    return profiles
+
 
 def all_profiles() -> list[ParseProfile]:
     global _PROFILES
     if _PROFILES is None:
-        _PROFILES = _lldp_profiles() + _vrf_profiles() + _zte_status_profiles()
+        _PROFILES = _apply_collect_lanes(
+            _lldp_profiles() + _vrf_profiles() + _zte_status_profiles()
+        )
     return list(_PROFILES)
 
 
@@ -1357,6 +1380,7 @@ def profile_to_public_dict(p: ParseProfile, *, overrides: dict[str, Any] | None
         "tags": list(p.tags),
         "sort_order": p.sort_order,
         "enabled": bool(ov.get("enabled")) if "enabled" in ov else p.enabled,
+        "collect_lane": str(p.collect_lane or "light"),
         "kind": p.kind,
         "match": p.match,
         "textfsm_command": p.textfsm_command or p.command_template,

netx_api/config.py

@@ -107,6 +107,10 @@ class Settings(BaseSettings):
     biz_state_scheduler_enabled: bool = True
     biz_state_scheduler_tick_sec: int = 15
     biz_state_dispatch_workers: int = 2
+    # Heavy CLI lane (interface detail / routes / MAC): own SSH + longer timeouts.
+    biz_state_heavy_read_timeout_sec: int = 300
+    biz_state_heavy_run_timeout_cap_sec: int = 900
+    biz_state_heavy_workers: int = 4
     # Managed NE exec: max CLI commands per request (lab can raise; hard-capped in ne_exec).
     ne_exec_max_commands: int = 5
     # WebCRT interactive terminal sessions (multi-operator concurrent terminals).

tests/test_collect_lanes.py

@@ -0,0 +1,162 @@
+"""Light/heavy collect lane partition and dual-lane scheduling."""
+
+from __future__ import annotations
+
+import time
+import unittest
+from unittest.mock import patch
+
+from netx_api.biz_state.collect_runner import (
+    partition_work,
+    work_item_lane,
+)
+from netx_api.biz_state.profiles import get_profile, reload_profiles
+
+
+class CollectLaneProfileTests(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls) -> None:
+        reload_profiles()
+
+    def test_interface_detail_is_heavy(self) -> None:
+        p = get_profile("zte.interface_detail")
+        self.assertIsNotNone(p)
+        assert p is not None
+        self.assertEqual(p.collect_lane, "heavy")
+
+    def test_route_profiles_are_heavy(self) -> None:
+        for pid in (
+            "zte.ip_route_vrf",
+            "zte.ip_route",
+            "zte.ipv6_route_vrf",
+            "zte.ipv6_route",
+        ):
+            p = get_profile(pid)
+            self.assertIsNotNone(p, pid)
+            assert p is not None
+            self.assertEqual(p.collect_lane, "heavy", pid)
+
+    def test_arp_is_light(self) -> None:
+        p = get_profile("zte.arp")
+        self.assertIsNotNone(p)
+        assert p is not None
+        self.assertEqual(p.collect_lane, "light")
+
+
+class PartitionWorkTests(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls) -> None:
+        reload_profiles()
+
+    def test_custom_raw_always_light(self) -> None:
+        self.assertEqual(work_item_lane("", "custom"), "light")
+        self.assertEqual(work_item_lane("zte.interface_detail", "custom"), "light")
+
+    def test_partition_mixed(self) -> None:
+        work = [
+            ("show arp", {}, "zte.arp", "i1", "normal"),
+            (
+                "show interface | include ifindex",
+                {},
+                "zte.interface_detail",
+                "i2",
+                "normal",
+            ),
+            ("show ip forwarding route vrf a", {}, "zte.ip_route_vrf", "i3", "normal"),
+            ("show version", {}, "", "i4", "custom"),
+        ]
+        light, heavy = partition_work(work)
+        self.assertEqual(len(light), 2)
+        self.assertEqual({x[2] for x in light}, {"zte.arp", ""})
+        self.assertEqual(len(heavy), 2)
+        self.assertEqual(
+            {x[2] for x in heavy},
+            {"zte.interface_detail", "zte.ip_route_vrf"},
+        )
+
+    def test_partition_light_only(self) -> None:
+        work = [("show arp", {}, "zte.arp", "i1", "normal")]
+        light, heavy = partition_work(work)
+        self.assertEqual(len(light), 1)
+        self.assertEqual(heavy, [])
+
+    def test_partition_heavy_only(self) -> None:
+        work = [
+            (
+                "show interface | include ifindex",
+                {},
+                "zte.interface_detail",
+                "i1",
+                "normal",
+            )
+        ]
+        light, heavy = partition_work(work)
+        self.assertEqual(light, [])
+        self.assertEqual(len(heavy), 1)
+
+
+class DualLaneParallelTests(unittest.TestCase):
+    def test_light_not_blocked_by_heavy_sleep(self) -> None:
+        """When both lanes run, light completes without waiting for heavy."""
+        from netx_api.biz_state import collect_runner as cr
+
+        order: list[str] = []
+
+        def fake_lane(*, work, label, **_kwargs):
+            if "heavy" in label:
+                time.sleep(0.35)
+                order.append("heavy_done")
+                return (0, len(work), False, True)
+            order.append("light_done")
+            return (0, len(work), False, True)
+
+        light_work = [("show arp", {}, "zte.arp", "i1", "normal")]
+        heavy_work = [
+            (
+                "show interface | include ifindex",
+                {},
+                "zte.interface_detail",
+                "i2",
+                "normal",
+            )
+        ]
+
+        with patch.object(cr, "_run_collect_lane", side_effect=fake_lane):
+            t0 = time.perf_counter()
+            heavy_fut = cr._heavy_cli_pool().submit(
+                lambda: cr._run_collect_lane(
+                    work=heavy_work,
+                    batch_id="b",
+                    creds={},
+                    vendor_eff="zte",
+                    device_type_eff="",
+                    vendor_key="zte",
+                    per_cmd=1,
+                    cap=10,
+                    label="biz_state_heavy",
+                )
+            )
+            light_res = cr._run_collect_lane(
+                work=light_work,
+                batch_id="b",
+                creds={},
+                vendor_eff="zte",
+                device_type_eff="",
+                vendor_key="zte",
+                per_cmd=1,
+                cap=10,
+                label="biz_state_light",
+            )
+            light_elapsed = time.perf_counter() - t0
+            heavy_res = heavy_fut.result()
+            total_elapsed = time.perf_counter() - t0
+
+        self.assertEqual(light_res[1], 1)
+        self.assertEqual(heavy_res[1], 1)
+        self.assertEqual(order[0], "light_done")
+        self.assertLess(light_elapsed, 0.25)
+        self.assertGreaterEqual(total_elapsed, 0.3)
+
+
+if __name__ == "__main__":
+    unittest.main()

web/src/pages/network/BizStatePage.tsx

@@ -197,6 +197,7 @@ export function BizStatePage() {
 
   const [tasks, setTasks] = useState<TaskRow[]>([]);
   const [busy, setBusy] = useState(false);
+  const [collectingIds, setCollectingIds] = useState<Record<string, true>>({});
   const [listKeyword, setListKeyword] = useState("");
   const debouncedListKw = useDebouncedValue(listKeyword, 250);
   const [purposeFilter, setPurposeFilter] = useState<"all" | "portrait" | "cutover_hf">("all");
@@ -253,7 +254,9 @@ export function BizStatePage() {
     const purpose =
       purposeFilter === "all" ? "" : purposeFilter === "portrait" ? "portrait" : "cutover_hf";
     const res = await bizStateListTasks(purpose);
-    setTasks((res.items || []) as TaskRow[]);
+    const items = (res.items || []) as TaskRow[];
+    setTasks(items);
+    return items;
   }, [purposeFilter]);
 
   useEffect(() => {
@@ -619,27 +622,58 @@ export function BizStatePage() {
     }
   };
 
-  const collectNow = async () => {
+  const setTaskCollecting = (id: string, on: boolean) => {
-    if (!taskId) return;
+    setCollectingIds((prev) => {
-    setBusy(true);
+      if (on) {
+        if (prev[id]) return prev;
+        return { ...prev, [id]: true };
+      }
+      if (!prev[id]) return prev;
+      const next = { ...prev };
+      delete next[id];
+      return next;
+    });
+  };
+
+  const collectNowForTask = async (id: string, fromModal = false) => {
+    setTaskCollecting(id, true);
     try {
-      await bizStateCollectNow(taskId);
+      await bizStateCollectNow(id);
+      showOk(t("bizState.collecting"));
       for (let i = 0; i < 20; i++) {
         await new Promise((r) => setTimeout(r, 1500));
-        const task = await bizStateGetTask(taskId);
+        if (fromModal && taskId === id) {
-        setDetail(task);
+          try {
-        if (!task.collect_running) break;
+            const task = await bizStateGetTask(id);
+            setDetail(task);
+            await refreshTasks();
+            if (!task.collect_running) break;
+          } catch {
+            break;
+          }
+          continue;
+        }
+        const items = await refreshTasks();
+        const latest = items.find((x) => x.id === id);
+        if (!latest?.collect_running) break;
       }
-      await loadTask(taskId);
       await refreshTasks();
-      setTaskTab("batches");
+      if (fromModal && taskId === id) {
+        await loadTask(id);
+        setTaskTab("batches");
+      }
     } catch (e) {
       showError(formatErr(e));
     } finally {
-      setBusy(false);
+      setTaskCollecting(id, false);
     }
   };
 
+  const collectNow = async () => {
+    if (!taskId) return;
+    await collectNowForTask(taskId, true);
+  };
+
   const removeTask = async (id: string) => {
     if (!window.confirm(t("bizState.confirmDelete"))) return;
     setBusy(true);
@@ -939,27 +973,8 @@ export function BizStatePage() {
                       <Button
                         size="sm"
                         variant="secondary"
-                        isDisabled={busy || row.collect_running}
+                        isDisabled={Boolean(row.collect_running || collectingIds[row.id])}
-                        onPress={async () => {
+                        onPress={() => void collectNowForTask(row.id, false)}
-                          setBusy(true);
-                          try {
-                            await bizStateCollectNow(row.id);
-                            showOk(t("bizState.collecting"));
-                            for (let i = 0; i < 20; i++) {
-                              await new Promise((r) => setTimeout(r, 1500));
-                              await refreshTasks();
-                              const cur = (await bizStateListTasks(purposeFilter === "all" ? "" : purposeFilter)).items?.find(
-                                (x: any) => x.id === row.id,
-                              );
-                              if (!cur?.collect_running) break;
-                            }
-                            await refreshTasks();
-                          } catch (e) {
-                            showError(formatErr(e));
-                          } finally {
-                            setBusy(false);
-                          }
-                        }}
                       >
                         {t("bizState.collectNow")}
                       </Button>
@@ -1439,7 +1454,7 @@ export function BizStatePage() {
           <Button
             size="sm"
             variant="primary"
-            isDisabled={busy || Boolean(detail?.collect_running)}
+            isDisabled={Boolean(detail?.collect_running || (taskId && collectingIds[taskId]))}
             onPress={() => void collectNow()}
           >
             {t("bizState.collectNow")}