[autoresearch] add pipeline authoring guide + async/auto_stop doc fixes (#1578)

Add a from-scratch SPDL pipeline authoring guide
(`tools/skills/authoring/building_pipelines.md`) covering stage
decomposition by operation nature, concurrency budgeting, `spdl.io` for
media, and multi-threading-in-subprocess for production. Its minimal
example shows the build-once / re-iterate-per-epoch pattern.

Clarify async stage handling in the pipeline reference docs: async
functions can be passed to `.pipe()` directly and must not be wrapped in
`asyncio.run()`, which only adds per-item event-loop init/finalize
overhead and forces the work onto worker threads instead of running as
cheap cooperative coroutines on the shared event loop.

Mark `auto_stop()` as obsolete in the optimization reference docs — a
`Pipeline` is built once and re-iterated each epoch without it.

Author: mthrok

src/spdl/autoresearch/pipeline_optimization/prompts/apply_changes.md

@@ -76,7 +76,7 @@ __PIPELINE_CODE__
 13. **TorchTNT scripts**: If the code uses TorchTNT (`torchtnt.framework.fit`, `train`, `AutoUnit`), the SPDL `Pipeline` is passed directly to TorchTNT as the `train_dataloader` (Pipeline is iterable — no wrapper class is required unless instrumentation was added). When applying changes:
    - **Pipeline construction changes** (concurrency, MTP, batch size): Modify the function that builds the `PipelineBuilder`, same as non-TorchTNT code. The `Pipeline` abstracts MTP vs pure multithreading, so the code passing Pipeline to TorchTNT does not change.
    - **Do NOT modify TorchTNT internals** (`fit()`, `train()`, `AutoUnit.train_step`). Only modify the pipeline construction.
-   - The pipeline is built once and iterated many times. There is no need for `auto_stop()` or rebuilding per epoch.
+   - The pipeline is built once and iterated many times. `auto_stop()` is obsolete — do not call it, and do not rebuild per epoch.
 
 Output the modified file:
 

src/spdl/autoresearch/pipeline_optimization/prompts/knowledge/knowledge.md

@@ -268,7 +268,7 @@ This wrapper is passed to TorchTNT's `fit()` or `train()` as the `train_dataload
 **Key facts about Pipeline:**
 - `Pipeline` is both iterable and iterator. It can technically be directly iterated, but the wrapper pattern above is recommended because it provides `get_iterator(timeout=...)` for timeout handling and `__len__` for epoch length.
 - **Preferred iteration**: Always use `pipeline.get_iterator(timeout=<seconds>)` to obtain an iterator with a timeout, so that jobs do not get stuck. Directly iterating with `for batch in pipeline:` or `iter(pipeline)` is discouraged because it lacks timeout handling.
-- The pipeline is built once and iterated many times. There is no need for `auto_stop()` or rebuilding per epoch.
+- The pipeline is built once and iterated many times. `auto_stop()` is obsolete — do not call it, and do not rebuild per epoch.
 - The `Pipeline` abstracts away whether it uses MTP (subprocess) or pure multithreading internally, so switching between them does not affect how the Pipeline is consumed.
 - TorchTNT calls `iter(dataloader)` at the start of each epoch, then `next(data_iter)` per step until `StopIteration`.
 - The pipeline's source controls how many items are produced per epoch. When the source is exhausted, the iterator raises `StopIteration`, ending the epoch in TorchTNT.

src/spdl/autoresearch/pipeline_optimization/prompts/knowledge/optimization_strategies.md

@@ -33,6 +33,37 @@ pipeline = (
 | `buffer_size` | `.add_sink(N)` | 2-10; affects memory, NOT throughput |
 | `output_order` | `.pipe(fn, output_order=...)` | "completion" for I/O (avoids head-of-line blocking), "input" for determinism |
 
+### Async Stage Functions: Pass Them As-Is
+
+`.pipe()` accepts both regular (`def`) and asynchronous (`async def`) callables. **Pass an async function directly — do not wrap it in `asyncio.run()`:**
+
+```python
+async def fetch(key):
+    async with session.get(url(key)) as resp:
+        return await resp.read()
+
+pipeline = (
+    PipelineBuilder()
+    .add_source(source)
+    .pipe(fetch, concurrency=32)   # CORRECT: coroutine function passed as-is
+    ...
+)
+```
+
+SPDL runs stages on its own event loop, so it awaits async stage functions natively and drives up to `concurrency` of them at once as cheap cooperative coroutines — ideal for I/O-bound work.
+
+**Anti-pattern:** wrapping the coroutine so the stage becomes synchronous.
+
+```python
+# WRONG — asyncio.run() spins up and tears down a brand-new event loop on
+# every single item. That per-item loop init/finalize is pure overhead, and
+# it forces the work onto pipeline worker threads instead of running as cheap
+# cooperative coroutines on SPDL's shared event loop.
+.pipe(lambda key: asyncio.run(fetch(key)), concurrency=32)
+```
+
+If a stage is already synchronous, just pass it as a normal function — no event loop is involved either way.
+
 ## Recommended Architecture: Multi-Threading in Subprocess (MTP)
 
 The production pattern runs the CPU-heavy pipeline in a **subprocess** and only does GPU transfer in the main process.

src/spdl/autoresearch/pipeline_optimization/prompts/plan_next.md

@@ -77,7 +77,7 @@ Before you can consider stopping, ALL of the following must have been attempted
    - **Critical — understand the existing code first**: Inspect the pipeline builder function's return type and structure. If it calls `.build()` or `.get_iterator()` (returning a `Pipeline` or iterator), the MTP refactor must change it to return a `PipelineConfig` (via `.get_config()`) or an unbuilt `PipelineBuilder` instead. `run_pipeline_in_subprocess()` accepts ONLY `PipelineConfig` — NOT `Pipeline` objects or iterators. Check type annotations and return statements.
    - **Separate CPU and GPU stages**: The backend pipeline (subprocess) must contain ONLY CPU-bound stages (fetch, decode, aggregate, collate). GPU stages like `transfer_tensor` require a CUDA context and MUST be in the frontend pipeline (main process). If the existing code includes `transfer_tensor` or similar GPU ops in the pipeline, exclude them from the backend config.
    - **Write the `description` field to explicitly instruct the code modifier** about these structural requirements, e.g.: "Refactor `build_pipeline()` to NOT call `.build()`. Instead, build a `PipelineBuilder` with only CPU stages (no `transfer_tensor`), call `.get_config()`, pass the config to `run_pipeline_in_subprocess(config, num_threads=N)`, then create a frontend `PipelineBuilder` that takes the subprocess iterable as source and applies `transfer_tensor`."
-   - **TorchTNT scripts**: If the code uses TorchTNT (`fit()`, `train()`, `AutoUnit`), the MTP refactor targets only the pipeline builder function — NOT TorchTNT internals. The `Pipeline` object abstracts away MTP vs pure multithreading, so the wrapper class that calls `get_iterator(timeout=...)` does not need to change. Do NOT use `auto_stop()` — the pipeline is built once and iterated many times.
+   - **TorchTNT scripts**: If the code uses TorchTNT (`fit()`, `train()`, `AutoUnit`), the MTP refactor targets only the pipeline builder function — NOT TorchTNT internals. The `Pipeline` object abstracts away MTP vs pure multithreading, so the wrapper class that calls `get_iterator(timeout=...)` does not need to change. Do NOT use `auto_stop()` (obsolete) — the pipeline is built once and iterated many times.
    - Use the **two-tier approach** for pickling: first try module-level functions with `functools.partial` (Tier 1). If the subprocess crashes silently (0 batches), retry with picklable callable classes that pickle objects directly from the main process (Tier 2). See "Pickling Constraints" in the knowledge base.
    - **HuggingFace tokenizers are NOT thread-safe** — use thread-local storage (TLS) when tokenizing with concurrency > 1.
    - Wrap the sampler with **`spdl.source.utils.embed_shuffle()`** for correct sampling behavior.
@@ -204,6 +204,6 @@ Rules:
 - torchx entrypoint args use **underscores**, not dashes (e.g. `--num_threads`)
 - The orchestrator will call a separate Claude session to apply the code changes described in `description`, commit them, rebuild the image, and then launch. **Write the `description` field as precise instructions for what code to modify** — specify function names, SPDL API calls to add/change, and the exact transformation. Do not write vague descriptions like "enable MTP" — instead write "Refactor `build_pipeline()` to return a `PipelineBuilder` config (without `.build()`), wrap it with `spdl.pipeline.run_pipeline_in_subprocess(config, num_threads=16, mp_context='forkserver')`, and create an outer pipeline that takes the subprocess source and applies GPU transfer via `pipe(transfer_tensor, executor=ThreadPoolExecutor(1))`."
 - Each experiment should differ from the baseline in exactly one dimension (or a small, justified set of changes)
-- **TorchTNT scripts**: When writing `description` for rebuild experiments in TorchTNT code, specify changes to the pipeline builder function only. The `Pipeline` abstracts away MTP vs pure multithreading, so the wrapper class (which calls `get_iterator(timeout=...)`) and TorchTNT internals (`fit()`, `train()`, `AutoUnit`) do not change. Do NOT use `auto_stop()` — Pipeline supports multiple iterations without it.
+- **TorchTNT scripts**: When writing `description` for rebuild experiments in TorchTNT code, specify changes to the pipeline builder function only. The `Pipeline` abstracts away MTP vs pure multithreading, so the wrapper class (which calls `get_iterator(timeout=...)`) and TorchTNT internals (`fit()`, `train()`, `AutoUnit`) do not change. Do NOT use `auto_stop()` (obsolete) — Pipeline supports multiple iterations natively.
 - **`best_practices_tags`**: Tag each experiment with which best practices it covers from the valid tags list. This is how the orchestrator tracks progress. If an experiment covers multiple practices, include all relevant tags.
 - **`goto`** (per-experiment): The orchestrator checks out the instrumentation (anchor) commit before applying each experiment's code changes by default. This ensures every experiment starts from a clean slate. Set `goto` to `null` in most cases. Only set it to a specific commit hash if you want to stack changes on top of a previous successful experiment (e.g., adding batch size tuning on top of an MTP experiment that already improved metrics). Never stack incompatible changes (e.g., GPU decode on top of MTP — they are mutually exclusive). **The `goto` field also determines the experiment's parent in the hypothesis tree**: `null` means the experiment branches from baseline; a specific commit means it branches from the experiment that produced that commit. This allows a single planning round to propose experiments with different parents (e.g., NVDEC from baseline + batch_size from a successful MTP experiment).

tools/skills/migration/migrating_to_spdl_pipeline.md

@@ -59,6 +59,8 @@ Or any iterable that yields keys/paths/metadata for downstream stages.
 
 Map each classified operation to a `.pipe()` stage with appropriate concurrency. Separate operations of different natures into distinct stages — never bundle network I/O and CPU decode in one function.
 
+`.pipe()` accepts both sync (`def`) and async (`async def`) callables. If the original loader used `async` I/O (e.g. `aiohttp`), **pass the coroutine function to `.pipe()` as-is** — SPDL awaits it on its own event loop. Do **not** wrap it as `asyncio.run(coro(x))`: that spins up and tears down a brand-new event loop on every item — pure per-item overhead — and forces the work onto worker threads instead of running as cheap cooperative coroutines on SPDL's shared event loop.
+
 Use `PriorityThreadPoolExecutor` from `spdl.pipeline` as the shared executor. It prioritizes downstream stages over upstream ones — when the thread pool is contended, items closer to the pipeline output are processed first, reducing end-to-end latency and ensuring data flows through the pipeline faster.
 
 ```python