feat: stopWhen / stopWhenSync source adapters (0.9.2)

README.md

@@ -31,6 +31,7 @@ array.filter(predicate).map(transform)
 Pipelean gives you:
 
 - `series` for sequential work over arrays **and async iterables** — live `onProgress`, `pause` rate limits, `take`, first-class error strategies
+- `stopWhen` to cancel or predicate-stop any source — composes with every consumer, no option changes
 - `scan` / `reduce` for stateful accumulation across many items
 - `flow` for stateful accumulation across one input — each operation enriches the same state
 - `pipe` for vertical composition

docs/architecture.md

@@ -21,7 +21,7 @@ Our Approach: Eager Execution.
 
 When you run series or scan, the work happens immediately and you get a structured report `{ results, errors, failure }` back. No surprises.
 
-This is a focus, not a rejection of lazy iterators. Pipelean *consumes* async iterables (arrays, generators, anything `for await` can consume) through `series`/`scan`/`reduce`/`filter`. When you need to *build* a lazy producer — for example an async generator that yields pages — pipelean steps aside: write a standard JavaScript generator and let it yield. The `no-loop-without-yield` lint rule encodes exactly this boundary: the only loops pipelean allows are the ones inside a generator that yields.
+This is a focus, not a rejection of lazy iterators. Pipelean *consumes* async iterables (arrays, generators, anything `for await` can consume) through `series`/`scan`/`reduce`/`filter`. When you need to *build* a lazy producer — for example an async generator that yields pages — pipelean steps aside: write a standard JavaScript generator and let it yield. The `no-loop-without-yield` lint rule encodes exactly this boundary: the only loops pipelean allows are the ones inside a generator that yields. `stopWhen` follows the same rule: it is a tiny yielding adapter (a producer, not an iterator), shipped because every consumer needed the same five lines of stop logic.
 
 ## Terminology
 

docs/examples.md

@@ -73,6 +73,30 @@ const {value: totalDuration} = await reduce(
 // totalDuration = 22 — no .at(-1), no fallback
 ```
 
+#### Example: stopWhen (Cancellable Pipeline)
+
+Scenario: You are enriching albums against the MusicBrainz API. The job takes minutes, users hit Cancel, and every request must respect a rate limit. Cancellation, pacing, error strategy and progress each live in exactly one place:
+
+```js
+import { series, stopWhen } from 'pipelean'
+
+const enrichLibrary = async (albums, { shouldStop = () => false, onProgress }) => {
+  await series(enrichOne, {
+    total: albums.length,
+    pause: ENRICH_DELAY_MS,     // rate limit lives in series
+    pauseOnErrors: true,
+    onProgress: onItem,         // progress lives in series
+  })(stopWhen(albums, shouldStop)) // cancellation lives in the source
+}
+
+const { results } = await enrichLibrary(albums, {
+  shouldStop: () => cancelRequested,
+  onProgress: ({ index, total }) => updateBar(index + 1, total),
+})
+```
+
+Before `stopWhen`, this needed an async generator with inline `shouldStop()` checks and the rate-limit delay tangled into its `finally` block — which is how double-yield bugs are born. Now each concern has one home.
+
 #### Example: tryCatch as an App-Layer Primitive
 
 Scenario: You want a reusable "Error Boundary" for your application that automatically logs errors to a monitoring service (like Sentry) and pushes a notification to your UI state (e.g., a Svelte store), ensuring the app never crashes silently. 

docs/guide.md

@@ -16,6 +16,7 @@ Pipelean provides core tools grouped by **data flow direction** (horizontal vs v
 6. pipe (Vertical / Composition)
 7. flow (Vertical / Stateful accumulation — one input, many enrichments, final accumulated value)
 8. assign (Utility for creating conditional property assignments for flow)
+9. stopWhen (Source adapter — predicate-based early exit for any iterable)
 
 > **Sync variants** — The iteration functions above also have synchronous
 > counterparts: `seriesSync`, `filterSync`, `findSync`, `scanSync`, and
@@ -95,6 +96,11 @@ Errors thrown by the **iteration itself** (e.g. an async generator dying mid-str
   - `total` for progress math; omitted when the size is unknown.
   - `scan` and `reduce` do **not** have `onProgress`, `pause`, or `take`.
 
+* **Source adapters**
+  - `stopWhen(items, predicate)` stops pulling as soon as the predicate is truthy — checked *before* the item is offered downstream.
+  - Composes with every consumer (`series`, `scan`, `reduce`, `filter`, raw `for await`) with no option changes: `series(fn, opts)(stopWhen(items, shouldStop))`.
+  - Cancel is a clean shorter run (`failure: false`), never a source error.
+
 * **Order Guarantee**
   - Because execution is sequential, output order strictly matches input order (no race conditions).
 

docs/migration.md

@@ -2,6 +2,28 @@
 
 How to replace common imperative and error-prone patterns with pipelean equivalents.
 
+## 0.9.2: `stopWhen` source adapter
+
+`stopWhen(items, predicate)` stops pulling from any iterable once the predicate is truthy, checked before the item is offered downstream. Additive — no migration needed, but delete your hand-rolled stop logic when you see it:
+
+```js
+// Before: a generator with inline checks (and cleanup tangled into finally)
+async function * events () {
+  for (const album of albums) {
+    if (shouldStop())
+      return
+    yield await enrichOne(album)
+  }
+}
+
+// After: cancellation lives in the source, work lives in the operation
+await series(enrichOne, {pause: DELAY})(stopWhen(albums, shouldStop))
+```
+
+The predicate receives `(item, index)` and may close over cancel flags or counters (`() => count >= limit`). Stopping is a clean completion (`failure: false`), not an error; the abandoned source's cleanup still runs. See [stopWhen](reference.md#stopwhen).
+
+Also in 0.9.2: iterable objects (generators, custom adapters) passed to `filter` / `filterSync` / `findSync` are now correctly treated as sources instead of `where()` patterns — `filter(pages(), pred)` used to silently misbehave.
+
 ## 0.8.4: Node engine requirement
 
 Pipelean now requires Node `>22.7` (package.json `engines`).

docs/patterns.md

@@ -156,3 +156,33 @@ for (const rawAlbum of rawAlbums) {
 ```
 
 Operations are reusable building blocks. The same `processAlbum` can be called from a CLI import script, a REST endpoint, or a background job — and the error shapes are normalized (`operation` is the function's `name` or `operation-${index}`), so error reports are consistent across entry points.
+
+### Pattern 9: Cancel / limit a run with `stopWhen()`
+
+Use `stopWhen()` when a run should stop pulling items once a condition is met: a user-requested cancel, a batch limit, or a content-based stop. The predicate is checked **before** each item is offered downstream, so the triggering item is never processed. It composes with every pipelean consumer — wrap the source, keep your options untouched:
+
+```javascript
+import { series, stopWhen } from 'pipelean'
+
+const { results } = await series(enrichOne, {
+  total: albums.length,        // forwarded from the array — optional with stopWhen
+  pause: ENRICH_DELAY_MS,
+  pauseOnErrors: true,
+  onProgress: onItem,
+})(stopWhen(albums, shouldStop))
+```
+
+The predicate receives `(item, index)` and may close over external state instead of looking at the item at all — that covers cancel flags and limits with one combinator:
+
+```javascript
+stopWhen(albums, shouldStop)                    // cancel flag
+stopWhen(folders, () => count >= limit)         // batch limit
+stopWhen(pages, page => page.isLast)            // content-based stop
+```
+
+Stopping is a clean completion for the consumer: `failure` stays `false`, `sourceErrors` stays empty — cancel is a shorter run, not an error. Once stopped, the underlying source is abandoned mid-stream and its cleanup (`finally`, `iterator.return()`) still runs.
+
+Two things `stopWhen` does **not** do:
+
+- it does not abort work already started on an item — intra-item cancellation stays in your operation (e.g. `AbortController`);
+- it is not a `series` option on purpose: wrapping the source also serves raw `for await` loops and `reduce`/`scan` pipelines, where a consumer option would not reach.

docs/reference.md

@@ -30,6 +30,7 @@
 - [scan](#scan) - Stateful Sequential Transformation
 - [reduce](#reduce) - Pure Reduction
 - [filter](#filter) - Stateless Selection
+- [stopWhen](#stopwhen) - Source Adapter for Predicate-Based Early Exit (alias of sorts: the `while` you don't need)
 
 ### Composition
 
@@ -453,7 +454,7 @@ const {value: totalDuration} = await reduce(
 
 **Key Characteristics**:
 - The predicate's return value is never placed into `results` — only truthiness is checked, and the original `item` is what gets kept or dropped.
-- Pattern objects are supported: `filter(users, {active: true})` works via `where()`.
+- Pattern objects are supported: `filter(users, {active: true})` works via `where()`. Anything implementing the iteration protocols (arrays, generators, source adapters like `stopWhen`) is always treated as a source, never as a pattern.
 
 **Usage Example**:
 ```javascript
@@ -468,6 +469,42 @@ const adults = await filter(
 
 ---
 
+### stopWhen
+
+**Purpose**: Source adapter that stops pulling from an iterable as soon as a predicate is truthy. The standard way to cancel or predicate-stop any source consumed by `series`, `scan`, `reduce`, `filter`, or a raw `for await`.
+
+**Type**: `(items, predicate?) => asyncIterable`
+Sync: `(items, predicate?) => iterable`
+
+**Parameters**:
+- `items`: An array or (async) iterable — generators and paging sources included
+- `predicate(item, index)`: Checked **before** each item is yielded. Truthy → stop. Defaults to `() => false` (never stops). Sync only.
+
+**Key Characteristics**:
+- **Check before yield**: the triggering item is pulled from the source but never offered downstream — "cancel before work"
+- **Pull-lazy**: once stopped, the underlying source is abandoned mid-stream; native cleanup (`iterator.return()`, generator `finally`) still runs
+- **Clean completion, not an error**: consumers see a finished source — `failure: false`, empty `sourceErrors`. Cancel is a shorter run, not a source death
+- **Composable**: zero changes in `series` / `scan` / `reduce` / `filter`; they just see an iterable that ends early. Works with raw `for await` too
+- **Length forwarding**: when the source has a numeric `length` (arrays), it is forwarded on the wrapper so `series` keeps progress totals without an explicit `total`. Generators have no cheap size — `total` stays omitted
+- **Predicate throws are source errors**: reported through `onSourceError` / `sourceErrors`, never `onError`
+- This is **not** intra-item abort: work already started on an item is not interrupted (use `AbortController` in your operation for that)
+
+**Usage Example**:
+```javascript
+import { series } from 'pipelean'
+
+const {results} = await series(enrichOne, {
+  total: albums.length,
+  pause: ENRICH_DELAY_MS,
+  onProgress: onItem,
+})(stopWhen(albums, shouldStop))
+
+// limit-style: close over counters instead of the item
+stopWhen(folders, () => count >= limit)
+```
+
+---
+
 ## Composition
 
 ### pipe
@@ -765,6 +802,7 @@ want Pipelean's structured error collection.
 - `pipeSync` composes synchronous functions left-to-right
 - `flowSync` returns `{value, errors, failure}` directly and runs a state-enrichment pipeline synchronously
 - `tryCatchSync` wraps a synchronous function with lifecycle hooks
+- `stopWhenSync` is the sync source adapter — same contract as `stopWhen`, sync iterables only
 
 The sync variants handle **source errors** with the same semantics as their async twins: `seriesSync`, `filterSync`, `scanSync`, and `reduceSync` accept `onSourceError({error, index})` and report errors thrown by the iteration itself in the `sourceErrors` field. See [Source errors](#source-errors) under `series`.
 

package.json

@@ -1,6 +1,6 @@
 {
   "name": "pipelean",
-  "version": "0.9.1",
+  "version": "0.9.2",
   "description": "A pragmatic library for sequential async operations with first-class error handling.",
   "type": "module",
   "license": "MIT",

skills/core/SKILL.md

@@ -67,7 +67,8 @@ const {results, errors, sourceErrors, failure} = await series(
 8. **`tryCatch(fn, {onStart, onSuccess, onError, onFinally})`**: Single-function lifecycle. Returns `null` on error.
 9. **`where(pattern)`**: Strict-equality object predicate. Used with `filter` / `findSync`.
 10. **`assign(property, parse)`**: `flow` step. Sets `{[property]: value}` unless `parse(state)` is `undefined` (returns `{}`).
-11. **`*Sync`**: `seriesSync`, `filterSync`, `findSync`, `scanSync`, `reduceSync`, `pipeSync`, `flowSync`, `tryCatchSync`. Same strategies and shapes, no Promises. No `pause` (needs async delay). No async iterables. `findSync` is sync-only early-exit: `{result, errors, failure}`.
+11. **`stopWhen(items, predicate?)`** / **`stopWhenSync`**: Source adapter — stops pulling once the predicate `(item, index)` is truthy, checked before the item is yielded. Cancel flags, limits (`() => count >= limit`), content stops. Clean completion (`failure: false`, no sourceErrors), source cleanup still runs, predicate throws are source errors. Composes with every consumer: `series(fn, opts)(stopWhen(items, shouldStop))`. Not intra-item abort.
+12. **`*Sync`**: `seriesSync`, `filterSync`, `findSync`, `scanSync`, `reduceSync`, `pipeSync`, `flowSync`, `tryCatchSync`. Same strategies and shapes, no Promises. No `pause` (needs async delay). No async iterables. `findSync` is sync-only early-exit: `{result, errors, failure}`.
 
 ## Error strategies
 

src/functional-sync.js

@@ -1,6 +1,6 @@
 /* eslint-disable max-lines */
 /* eslint-disable max-lines-per-function */
-import {getPlannedTotal, withTotal} from './shared.js'
+import {getPlannedTotal, isPatternObject, withTotal} from './shared.js'
 import {
   collect, failFast, normalizeOperationError, where,
 } from './functional.js'
@@ -138,11 +138,8 @@ export const seriesSync = (...args) => {
 }
 
 export const filterSync = (...args) => {
-  const isPattern = x => x !== null &&
-    typeof x === 'object' &&
-    !Array.isArray(x)
-  const toPredicate = x => isPattern(x) ? where(x) : x
-  const immediate = typeof args[0] !== 'function' && !isPattern(args[0])
+  const toPredicate = x => isPatternObject(x) ? where(x) : x
+  const immediate = typeof args[0] !== 'function' && !isPatternObject(args[0])
   const [items, rawPredicate, opts] = immediate
     ? args
     : [null, args[0], args[1]]
@@ -158,11 +155,8 @@ export const filterSync = (...args) => {
 }
 
 export const findSync = (...args) => {
-  const isPattern = x => x !== null &&
-    typeof x === 'object' &&
-    !Array.isArray(x)
-  const toPredicate = x => isPattern(x) ? where(x) : x
-  const immediate = typeof args[0] !== 'function' && !isPattern(args[0])
+  const toPredicate = x => isPatternObject(x) ? where(x) : x
+  const immediate = typeof args[0] !== 'function' && !isPatternObject(args[0])
   const [items, rawPredicate, opts = {}] = immediate
     ? args
     : [null, args[0], args[1]]

src/functional.js

@@ -1,6 +1,6 @@
 /* eslint-disable max-lines */
 /* eslint-disable max-lines-per-function */
-import {getPlannedTotal, withTotal} from './shared.js'
+import {getPlannedTotal, isPatternObject, withTotal} from './shared.js'
 
 export const failFast = Object.freeze({name: 'failFast'})
 export const collect = Object.freeze({name: 'collect'})
@@ -189,11 +189,8 @@ export const series = (...args) => {
 }
 
 export const filter = (...args) => {
-  const isPattern = x => x !== null &&
-    typeof x === 'object' &&
-    !Array.isArray(x)
-  const toPredicate = x => isPattern(x) ? where(x) : x
-  const immediate = typeof args[0] !== 'function' && !isPattern(args[0])
+  const toPredicate = x => isPatternObject(x) ? where(x) : x
+  const immediate = typeof args[0] !== 'function' && !isPatternObject(args[0])
   const [items, rawPredicate, opts] = immediate
     ? args
     : [null, args[0], args[1]]

src/index.js

@@ -1,2 +1,3 @@
 export * from './functional.js'
 export * from './functional-sync.js'
+export * from './iterables.js'

src/iterables.js

@@ -0,0 +1,36 @@
+// Source adapters: lazy iterable producers, consumed by series/reduce/scan,
+// filter, findSync, or any raw for await / for of.
+//
+// stopWhen stops pulling from the source as soon as the predicate is truthy.
+// The predicate is checked BEFORE yielding: the triggering item is pulled but
+// not offered downstream ("cancel before work"). Stopping is a clean
+// completion for every consumer — cancel is a shorter run, not a source error.
+
+const forwardedLength = items =>
+  typeof items?.length === 'number' ? {length: items.length} : {}
+
+export const stopWhen = (items, predicate = () => false) => ({
+  ...forwardedLength(items),
+  async * [Symbol.asyncIterator] () {
+    let index = 0
+    for await (const item of items) {
+      if (predicate(item, index))
+        return
+      yield item
+      index++
+    }
+  },
+})
+
+export const stopWhenSync = (items, predicate = () => false) => ({
+  ...forwardedLength(items),
+  * [Symbol.iterator] () {
+    let index = 0
+    for (const item of items) {
+      if (predicate(item, index))
+        return
+      yield item
+      index++
+    }
+  },
+})

src/shared.js

@@ -1,6 +1,15 @@
 export const getKnownTotal = (items, total) =>
   total !== undefined ? total : items.length
 
+// A where() pattern is a plain data object. Anything that can be iterated
+// (arrays, generators, source adapters) is an input, not a pattern.
+export const isPatternObject = x =>
+  x !== null &&
+  typeof x === 'object' &&
+  !Array.isArray(x) &&
+  !(Symbol.asyncIterator in x) &&
+  !(Symbol.iterator in x)
+
 export const getPlannedTotal = ({items, take, total}) => {
   const knownTotal = getKnownTotal(items, total)