Add Pipeline.resize_concurrency runtime API + enable_adaptive_concurrency flag

Summary:
This is Diff 3b in the T262755626 stack. Adds the user-facing API on top of Diff 3a's foundation:

1. `Pipeline.resize_concurrency(qualified_name, new_value)` — synchronous foreground-thread API. Detects in-loop callers via `asyncio.get_running_loop()` and raises RuntimeError to prevent self-deadlock. Lifecycle gated against NOT_STARTED and STOPPED states. Forwards to `_resize_concurrency_async` via `_EventLoop.run_coroutine_threadsafe`.
2. `Pipeline._resize_concurrency_async(qualified_name, new_value)` — async source-of-truth. Safe to call from a `BackgroundTask` or any in-loop coroutine. Updates the registered `ResizableSemaphore` AND the `_dynamic_concurrency` sibling dict atomically (single asyncio turn). Track B's `AdaptiveTuner` (RFC) will use this directly.
3. `Pipeline.list_stages()` — returns sorted qualified names registered with semaphores. Useful for notebook discovery during A/B sweeps.
4. `build_pipeline(..., enable_adaptive_concurrency=False)` and `PipelineBuilder.build(..., enable_adaptive_concurrency=False)` — public flag that wires every Pipe stage with a `ResizableSemaphore` (matching its static `concurrency`) so `Pipeline.resize_concurrency` can adjust admission caps at runtime. Default False keeps the static admission gate (zero per-task overhead) on the no-opt path. Production callers (e.g. LCA's `SPDLConfig.enable_adaptive_concurrency` in Diff 4) use this flag; the existing `_install_semaphores_for_test` knob remains for SPDL's own tests.

V5.7 continuous-mode regression test included: builds a continuous-mode pipeline, runs across 3 epochs, calls `resize_concurrency` mid-run, and verifies (a) no crash on epoch sentinel propagation, (b) resized cap takes effect for subsequent items, (c) in-flight items at resize time complete normally.

Error semantics:
- `KeyError` on unknown qualified_name; the message lists valid names from the registry.
- `ValueError` on `new_value < 1`.
- `RuntimeError` on lifecycle violations or in-loop sync calls.

Documentation note included in `resize_concurrency` docstring: Meta-internal dashboards (e.g., `_logging.py`) currently report build-time `StageInfo.concurrency` and do not reflect resizes performed via this API. Operators reading live values from a notebook should use `list_stages()` and `pipeline._impl._dynamic_concurrency` directly. Future Track B work may extend Meta logging to read the dict.

Differential Revision: D102929252

Author: justinvjoseph

src/spdl/pipeline/_build.py

@@ -130,6 +130,7 @@ def _build_pipeline(
     stage_id: int = 0,
     background_tasks: list[BackgroundTaskFactory] | None = None,
     use_priority_scheduler: bool = False,
+    enable_adaptive_concurrency: bool = False,
     _install_semaphores_for_test: bool = False,
 ) -> Pipeline[U]:
     if _DEFAULT_BUILD_CALLBACK is not None:
@@ -178,9 +179,12 @@ def _build_pipeline(
         )
 
     # V5.1+V5.5 Diff 3a: pre-allocate the per-pipeline registries that
-    # `_components/_node.py` populates when ``_install_semaphores_for_test``
-    # is True. These are then handed off to ``_PipelineImpl.__init__`` so
-    # ``Pipeline.resize_concurrency`` (Diff 3b) can find them.
+    # `_components/_node.py` populates when semaphore installation is
+    # enabled (either via the public ``enable_adaptive_concurrency`` flag
+    # or the test-only ``_install_semaphores_for_test`` knob). These are
+    # then handed off to ``_PipelineImpl.__init__`` so
+    # ``Pipeline._resize_concurrency_async`` (Diff 3b internal) can find
+    # them.
     # ``output_queue_by_name`` (Phase D) caches each registered stage's
     # output ``AsyncQueue`` so the Diff 6 controller can read its lap
     # stats — the same key set as ``semaphore_registry``.
@@ -189,6 +193,8 @@ def _build_pipeline(
     stage_info_by_name: dict[str, StageInfo] = {}
     output_queue_by_name: dict[str, AsyncQueue] = {}
 
+    install_semaphores = enable_adaptive_concurrency or _install_semaphores_for_test
+
     coro, queue = _build_pipeline_coro(
         pipeline_cfg,
         max_failures=max_failures,
@@ -198,7 +204,7 @@ def _build_pipeline(
         stage_id=stage_id,
         background_tasks=all_bg_tasks or None,
         scheduler=scheduler,
-        install_semaphores_for_test=_install_semaphores_for_test,
+        install_semaphores_for_test=install_semaphores,
         semaphore_registry=semaphore_registry,
         dynamic_concurrency=dynamic_concurrency,
         stage_info_by_name=stage_info_by_name,
@@ -229,6 +235,7 @@ def build_pipeline(
     stage_id: int = 0,
     background_tasks: list[BackgroundTaskFactory] | None = None,
     use_priority_scheduler: bool = False,
+    enable_adaptive_concurrency: bool = False,
     _install_semaphores_for_test: bool = False,
 ) -> Pipeline[U]:
     """Build a pipeline from the config.
@@ -301,15 +308,25 @@ def build_pipeline(
             Deeper stages (closer to sink) are given higher priority,
             reducing pipeline bubble time.
 
-        _install_semaphores_for_test: **Test-only.** When ``True``, every
-            ``Pipe`` stage is built with a :py:class:`ResizableSemaphore`
-            whose initial value matches its static ``concurrency``, and
-            the semaphore is wired to the V5.6 REPLACE admission gate in
-            :py:func:`~spdl.pipeline._components._pipe._pipe`. Used by
-            the V5.5 throughput regression test and the
-            :py:meth:`Pipeline.resize_concurrency` continuous-mode test.
-            Production code MUST NOT pass this — it bypasses the
-            selective opt-in semantics of ``Pipeline.resize_concurrency``.
+        enable_adaptive_concurrency: If ``True``, every ``Pipe`` stage is
+            built with a :py:class:`ResizableSemaphore` whose initial value
+            matches its static ``concurrency``, and the semaphore is wired
+            to the V5.6 REPLACE admission gate in
+            :py:func:`~spdl.pipeline._components._pipe._pipe`. This enables
+            runtime adjustment of per-stage concurrency via the internal
+            :py:meth:`Pipeline._resize_concurrency_async` (intended to be
+            driven by an in-loop adaptive-concurrency controller running
+            as a :py:class:`BackgroundTask`). Default: ``False``
+            (per-stage concurrency is fixed at build time, with zero
+            per-task overhead in the admission gate).
+
+        _install_semaphores_for_test: **Test-only.** Same mechanical effect
+            as ``enable_adaptive_concurrency`` (both flip the same internal
+            switch), kept as a separate flag so tests can opt in without
+            implying the production-facing semantic. Used by the V5.5
+            throughput regression test and the in-loop async-resize
+            regression test. Production code MUST use
+            ``enable_adaptive_concurrency``.
     """
     from . import _profile
 
@@ -326,6 +343,7 @@ def build_pipeline(
         stage_id=stage_id,
         background_tasks=background_tasks,
         use_priority_scheduler=use_priority_scheduler,
+        enable_adaptive_concurrency=enable_adaptive_concurrency,
         _install_semaphores_for_test=_install_semaphores_for_test,
     )
 

src/spdl/pipeline/_builder.py

@@ -292,6 +292,7 @@ def build(
         task_hook_factory: Callable[[StageInfo], list[TaskHook]] | None = None,
         stage_id: int = 0,
         use_priority_scheduler: bool = False,
+        enable_adaptive_concurrency: bool = False,
     ) -> Pipeline[U]:
         """Build the pipeline.
 
@@ -334,6 +335,14 @@ def build(
                 dispatch for sync stages via :py:class:`PriorityScheduler`.
                 Deeper stages (closer to sink) are given higher priority,
                 reducing pipeline bubble time.
+
+            enable_adaptive_concurrency: If ``True``, every ``Pipe`` stage
+                is built with a :py:class:`ResizableSemaphore` so per-stage
+                concurrency can be adjusted at runtime via the internal
+                :py:meth:`Pipeline._resize_concurrency_async` (intended to
+                be driven by an in-loop adaptive-concurrency controller
+                running as a :py:class:`BackgroundTask`). Default:
+                ``False`` (per-stage concurrency is fixed at build time).
         """
         return build_pipeline(
             self.get_config(),
@@ -344,4 +353,5 @@ def build(
             task_hook_factory=task_hook_factory,
             stage_id=stage_id,
             use_priority_scheduler=use_priority_scheduler,
+            enable_adaptive_concurrency=enable_adaptive_concurrency,
         )

src/spdl/pipeline/_components/_node.py

@@ -482,7 +482,7 @@ def _convert_config(
 
 
 def _qualified_name(info: StageInfo, branch_label: str | None = None) -> str:
-    """Build the qualified stage name used by ``Pipeline.resize_concurrency``.
+    """Build the qualified stage name used by ``Pipeline._resize_concurrency_async``.
 
     For non-MultiPipe stages, ``qualified_name = info.stage_name``.
     For MultiPipe sub-pipelines, the branch label is prefixed with ``/``
@@ -825,8 +825,11 @@ def _build_node_recursive(
             every Pipe stage gets a :py:class:`ResizableSemaphore` whose
             initial value matches its static ``concurrency``, and the
             semaphore is passed to ``_pipe()`` so its V5.6 REPLACE branch
-            governs admission. Production code should use
-            :py:meth:`Pipeline.resize_concurrency` (Diff 3b) instead.
+            governs admission. Production code should pass
+            ``enable_adaptive_concurrency=True`` to
+            :py:func:`build_pipeline` instead so that an in-loop
+            controller can call
+            :py:meth:`Pipeline._resize_concurrency_async`.
         semaphore_registry: Per-pipeline ``dict[qualified_name, ResizableSemaphore]``
             populated when ``install_semaphores_for_test=True``.
         dynamic_concurrency: Per-pipeline ``dict[qualified_name, int]``

src/spdl/pipeline/_components/_pipe.py

@@ -381,8 +381,8 @@ def _pipe_with_semaphore(
     """V5.6 REPLACE branch: ``semaphore.acquire()`` IS the admission gate.
 
     ``args.concurrency`` is ignored — the registered semaphore (whose value
-    is mutable via :py:meth:`Pipeline.resize_concurrency`) governs the
-    in-flight cap. Task completion calls ``semaphore.release()`` via a
+    is mutable via :py:meth:`Pipeline._resize_concurrency_async`) governs
+    the in-flight cap. Task completion calls ``semaphore.release()`` via a
     done-callback so the admit cycle is symmetric with the gate.
     """
 
@@ -421,7 +421,7 @@ async def pipe() -> None:
             # V5.6 admission gate: REPLACES `len(tasks) >= args.concurrency`.
             # `acquire()` blocks here when the in-flight count reaches the
             # semaphore's current value (which may have been resized via
-            # Pipeline.resize_concurrency).
+            # Pipeline._resize_concurrency_async).
             await semaphore.acquire()
 
             task = create_task(

src/spdl/pipeline/_pipeline.py

@@ -600,6 +600,66 @@ def __iter__(self) -> Iterator[T]:
         """Call :py:meth:`~spdl.pipeline.Pipeline.get_iterator` without arguments."""
         return self.get_iterator()
 
+    # --------------------------------------------------------------
+    # Diff 3b — runtime concurrency adjustment (INTERNAL)
+    # --------------------------------------------------------------
+
+    async def _resize_concurrency_async(
+        self,
+        qualified_name: str,
+        new_value: int,
+    ) -> None:
+        """Resize the in-flight admission cap for a registered stage.
+
+        INTERNAL — must be awaited from a coroutine running on the
+        pipeline's own event loop (e.g., a
+        :py:class:`~spdl.pipeline.BackgroundTask`). There is no
+        public, foreground-thread wrapper: cross-thread resize is
+        intentionally not exposed because the only intended caller is
+        an in-loop adaptive-concurrency controller.
+
+        Atomicity: the body has zero ``await`` statements between
+        :py:meth:`ResizableSemaphore.resize` and the
+        ``_dynamic_concurrency`` dict assignment. asyncio is
+        single-threaded, so the entire method runs in one event-loop
+        turn and is therefore cancel-safe by structural invariant
+        (``CancelledError`` can only fire BEFORE the call begins or
+        AFTER it completes, never between the two writes).
+
+        Args:
+            qualified_name: A fully-qualified stage name. For
+                non-MultiPipe stages this equals
+                :py:attr:`StageInfo.stage_name` (e.g.,
+                ``"decode_single_frame"``). For MultiPipe sub-pipelines
+                it is ``"<branch_label>/<stage_name>"``. The set of
+                valid names is ``pipeline._impl._semaphore_registry``.
+            new_value: New admission cap; must be >= 1. Resize-up is
+                immediate; resize-down is graceful — in-flight tasks
+                finish, but no new tasks admit until in-flight drops
+                below ``new_value``.
+
+        Raises:
+            ValueError: ``new_value < 1``.
+            KeyError: ``qualified_name`` is not registered. The error
+                message includes the list of valid names.
+        """
+        if new_value < 1:
+            raise ValueError(f"new_value must be >= 1, got {new_value}")
+
+        registry = self._impl._semaphore_registry
+        sem = registry.get(qualified_name)
+        if sem is None:
+            valid = sorted(registry.keys())
+            raise KeyError(
+                f"qualified_name {qualified_name!r} not found. "
+                f"Valid stage names: {valid}"
+            )
+        sem.resize(new_value)
+        # Keep the sibling registry in sync for observability.
+        # asyncio is single-threaded so this update is atomic with the
+        # semaphore.resize() above (no concurrent writer).
+        self._impl._dynamic_concurrency[qualified_name] = new_value
+
 
 class PipelineIterator(Generic[T]):
     """PipelineIterator()"""