Add failure-isolated loop iterations (#377)

* Add opt-in loop failure continuation

* Isolate opted-in loop iterations

* Test resilient scheduled loop iterations

* Harden failure-isolated loop execution

Distinguish body activity failures from child runtime and protocol faults so continue-on-failure loops only consume expected application failures.

Reject conditional loops combined with failure continuation, document the new loop policy and lifetime, and pin deterministic child input serialization.

* Unify the loop SQL API

* Support resilient conditional loops

* Unify conditional loop failure handling

* Correct the pg_textsearch compactor example

* Clarify the loop API test name

* Prioritize fatal JOIN branch failures

* Harden failure-isolated loop conditions

* Mark continue_on_failure syntax experimental

---------

Co-authored-by: tjgreen42 <1738591+tjgreen42@users.noreply.github.com>

Author: tjgreen42

.agents/skills/pg-durable-sql/SKILL.md

@@ -93,9 +93,18 @@ df.if(condition TEXT, then_branch TEXT, else_branch TEXT) → TEXT
 -- result_name is a capture from |=> earlier in the graph.
 df.if_rows(result_name TEXT, then_branch TEXT, else_branch TEXT) → TEXT
 
--- Loop — infinite or while-condition
-df.loop(body TEXT) → TEXT                            -- Infinite loop
-df.loop(body TEXT, condition TEXT) → TEXT             -- While-loop: repeats while condition is truthy
+-- Loop — one unified signature
+df.loop(
+    body TEXT,
+    condition TEXT DEFAULT NULL,
+    continue_on_failure BOOLEAN DEFAULT false
+) → TEXT
+-- Experimental: continue_on_failure syntax may change in future releases.
+-- NULL condition: infinite loop.
+-- Non-NULL condition: do-while semantics; evaluate it after each successful body.
+-- With continue_on_failure => true, a consumed body activity failure skips the
+-- condition and starts the next iteration. The loop's condition and
+-- orchestration/runtime failures remain fatal.
 
 -- Break from enclosing loop
 df.break() → TEXT                                    -- Exit with NULL
@@ -227,7 +236,7 @@ Automatically available during execution:
 
 ## Condition Evaluation (Truthiness)
 
-Used by: `?>`, `!>`, `df.if()`, `df.loop(body, condition)`
+Used by: `?>`, `!>`, `df.if()`, and the optional condition in `df.loop()`
 
 The first column of the first row is evaluated:
 
@@ -345,7 +354,7 @@ SELECT df.start(
 -- Cancel with: SELECT df.cancel('instance_id', 'Stopping heartbeat');
 ```
 
-### While-Loop with Break
+### Loop with Break
 
 ```sql
 SELECT df.start(
@@ -361,6 +370,27 @@ SELECT df.start(
 );
 ```
 
+### Conditional Loop with Failure Continuation
+
+```sql
+SELECT df.start(
+    df.loop(
+        'SELECT process_next_item()',
+        'SELECT EXISTS (
+            SELECT 1 FROM work_queue WHERE status = ''pending''
+        )',
+        continue_on_failure => true
+    ),
+    'resilient-worker'
+);
+```
+
+This is a do-while loop: after a successful body execution, the condition is
+evaluated and the loop continues while it is truthy. After a consumed body
+activity failure, the condition is skipped and the body starts again.
+Errors returned by body SQL, HTTP, and multipart activities are consumable.
+Condition and orchestration/runtime failures remain fatal.
+
 ### Cron Scheduled Job
 
 ```sql

CHANGELOG.md

@@ -6,6 +6,25 @@ Pre-1.0 note: while `pg_durable` is in major version `0`, minor releases may inc
 
 ## [0.2.8] - Unreleased
 
+### Added
+
+- **Failure-isolated loops:** the unified
+  `df.loop(body, condition DEFAULT NULL, continue_on_failure DEFAULT false)`
+  signature supports resilient infinite and conditional loops. With
+  `continue_on_failure => true`, a consumed typed body activity failure skips
+  the condition and starts the next iteration; after a successful body, the
+  condition is evaluated normally. All errors returned by body SQL, HTTP, and
+  multipart activities are consumable, including query, authorization,
+  connection, and network errors. Condition, graph, protocol, unknown child,
+  and orchestration/runtime failures remain fatal. The
+  `continue_on_failure` syntax is experimental and may change in future
+  releases.
+
+### Changed
+
+- **Loop lifetime:** raises the loop-iteration backstop from 100,000 to
+  8,388,608 (`2^23`), approximately 80 years at five-minute ticks.
+
 ### Fixed
 
 - **Caller-transaction handoff:** `df.start()` now tracks the originating transaction until it commits or aborts, so legal caller transactions lasting more than five seconds no longer leave a `pending` `df.instances` row paired with a failed engine execution. Graph admission uses durable backoff and bounded-history compaction rather than holding a worker connection while it waits.

USER_GUIDE.md

@@ -264,8 +264,7 @@ df.sql('SELECT 1') ~> df.sql('SELECT 2')
 | `df.join3(a, b, c)` | Three in parallel | `df.join3(a, b, c)` |
 | `df.race(a, b)` | Execute in parallel, first wins | `df.race(fast_query, slow_query)` |
 | `df.if(cond, then, else)` | Conditional branch | `df.if('SELECT true', a, b)` |
-| `df.loop(body)` | Repeat forever | `df.loop(body)` |
-| `df.loop(body, cond)` | Repeat while condition is true | `df.loop(body, 'SELECT count(*) > 0 FROM q')` |
+| `df.loop(body, condition DEFAULT NULL, continue_on_failure DEFAULT false)` | Repeat forever or while a condition is true, optionally continuing after body activity failures | `df.loop(body, 'SELECT count(*) > 0 FROM q', continue_on_failure => true)` |
 | `df.break()` | Exit enclosing loop | `df.break()` |
 | `df.break(value)` | Exit loop with **literal** return value (not auto-wrapped as SQL) | `df.break('{"done": true}')` |
 | `df.start(func, label, database)` | Start function (optionally in another database) | `df.start('SELECT 1', 'job')` |
@@ -531,7 +530,8 @@ SELECT df.start(
 
 ### Loop Condition Example
 
-For `df.loop(body, condition)`, the condition is evaluated after each iteration:
+For `df.loop(body, condition)`, the condition is evaluated after each
+successful body execution:
 
 ```sql
 -- Loop while there are pending items
@@ -1214,6 +1214,75 @@ SELECT df.start(
 );
 ```
 
+The unified signature is
+`df.loop(body, condition DEFAULT NULL, continue_on_failure DEFAULT false)`.
+
+> **Experimental:** The `continue_on_failure` syntax is subject to change in
+> future releases.
+
+It supports four call shapes:
+
+```sql
+df.loop(body)
+df.loop(body, condition)
+df.loop(body, continue_on_failure => true)
+df.loop(body, condition, continue_on_failure => true)
+```
+
+By default, it and `@>` are fail-fast: an iteration failure fails the loop.
+Passing `continue_on_failure => false` is equivalent to the default and does
+not change where the loop is hosted. When it is `true`, each body iteration
+runs as a child orchestration. After a successful body iteration, the child's
+results are merged into the parent result map before the optional condition is
+evaluated. After a consumed typed application failure from a body activity,
+the condition is skipped and the parent starts the next iteration. This
+includes any error returned by a body SQL, HTTP, or multipart activity, such as
+a query, authorization, connection, or network error. Condition failures,
+malformed graph or child data, unrecognized child errors, child-ID collisions,
+and orchestration/runtime failures remain fatal.
+
+This is useful for scheduled maintenance such as a pg_textsearch-style
+per-index compactor. A simplified periodic backstop repeatedly runs one
+compaction step until the index has no more reducible debt:
+
+```sql
+SELECT df.start(
+    df.loop(
+        df.wait_for_schedule('*/5 * * * *')
+        ~> df.loop(
+            $$SELECT public.bm25_compact_step(
+                'documents_idx'::regclass
+            ) AS ran$$
+            |=> 'step'
+            ~> df.if(
+                'SELECT $step.ran',
+                'SELECT true',
+                df.break()
+            )
+        ),
+        continue_on_failure => true
+    ),
+    'documents-index-compaction'
+);
+```
+
+The managed pg_textsearch workflow also validates the physical index identity
+before each cascade, so replacing or dropping the index stops stale work.
+
+The condition and continuation policy can be combined:
+
+```sql
+df.loop(
+    'SELECT process_next_item()',
+    'SELECT count(*) > 0 FROM task_queue',
+    continue_on_failure => true
+)
+```
+
+The next schedule is computed only after the prior iteration finishes. Slow or
+failed iterations therefore do not overlap, queue missed ticks, or backfill
+earlier schedule times.
+
 ### Cron-Style Scheduling
 
 Use `df.wait_for_schedule()` with a cron expression:
@@ -1319,10 +1388,21 @@ SELECT df.start(
 
 Each loop iteration advances via *continue-as-new*, which restarts the loop with fresh state while preserving durability. Where that restart happens depends on whether the loop is the **root** of the function:
 
-- A **root loop** (the function graph's outermost node, e.g. `df.start(df.loop(...))` or `df.start(@> 'SELECT work()')`) runs inline on the function's own orchestration. There is no surrounding work to preserve, so each iteration simply restarts the function. The `@>` operator does not make a loop root by itself: `df.seq('SELECT setup()', @> 'SELECT work()')` is non-root because the sequence is the graph root.
-- A **non-root loop** (a loop with prefix/suffix nodes, or one nested inside a `df.if()`, JOIN (`&`), or RACE (`|`) branch) runs as its own **child sub-orchestration**. Only the loop body restarts on each iteration — any work *before* the loop runs exactly once and is never re-executed, and a loop nested in a parallel branch gets its own durable instance.
-
-This is transparent to your workflow; it only affects observability. The child sub-orchestration is an internal durable instance: it does **not** appear in `df.list_instances()` (which lists only the instances you started with `df.start()`). Instead, the loop node's status in `df.instance_nodes()` / `df.explain()` reflects the child's progress, so you observe the loop through its parent instance as usual.
+- A fail-fast **root loop** (the function graph's outermost node, e.g. `df.start(df.loop(...))` or `df.start(@> 'SELECT work()')`) runs inline on the function's own orchestration. There is no surrounding work to preserve, so each iteration simply restarts the function. The `@>` operator does not make a loop root by itself: `df.seq('SELECT setup()', @> 'SELECT work()')` is non-root because the sequence is the graph root.
+- A fail-fast **non-root loop** (a loop with prefix/suffix nodes, or one nested inside a `df.if()`, JOIN (`&`), or RACE (`|`) branch) runs as its own **child sub-orchestration**. Only the loop body restarts on each iteration — any work *before* the loop runs exactly once and is never re-executed, and a loop nested in a parallel branch gets its own durable instance.
+- Fail-fast iterations execute inline within whichever orchestration hosts the loop. Explicit `continue_on_failure => false` keeps this placement; it does not force a non-root loop into the function's root orchestration.
+- A loop with `continue_on_failure => true` runs every body iteration in a child orchestration. After a successful body iteration, the child's results are merged into the parent result map before the optional condition is evaluated. After any error returned by a body SQL, HTTP, or multipart activity, the condition is skipped and the loop parent proceeds to the next iteration. Condition failures and unrecognized child, graph, protocol, orchestration, or runtime failures remain fatal.
+
+These child sub-orchestrations are internal durable instances and do **not**
+appear in `df.list_instances()`, which lists only instances started with
+`df.start()`. `df.instance_nodes()` / `df.explain()` show the loop parent as
+running and may show a failed iteration on its body-root or descendant nodes.
+Those node statuses are not permanent per-iteration history: a later iteration
+can supersede them with newer execution status.
+
+All loops have a finite backstop of 8,388,608 (`2^23`) iterations: about 80
+years at five-minute intervals, or a little over 97 days at the enforced
+one-second minimum.
 
 ### Stopping a Loop Externally
 

docs/ARCHITECTURE.md

@@ -477,31 +477,36 @@ of every handler having to recognise an in-band JSON break sentinel:
 pub enum NodeError {
     /// df.break() fired. Carries the break value. Caught only by execute_loop_node.
     Break(String),
-    /// A genuine failure. Surfaces as a failed instance.
+    /// An expected workflow activity failure.
+    Application(String),
+    /// A graph, protocol, configuration, or runtime failure.
     Failure(String),
 }
 
 pub type NodeResult = Result<String, NodeError>;
 
-// Any `?` on an existing Result<_, String> auto-converts the error to Failure, so
-// activity calls and helpers need no per-call changes.
+// Structural/configuration helpers still convert String errors to Failure.
 impl From<String> for NodeError {
     fn from(e: String) -> Self {
         NodeError::Failure(e)
     }
 }
 ```
 
-`execute_loop_node` is the only handler that catches `NodeError::Break` (turning it into the
-loop's `Ok` result); `NodeError::Failure` keeps propagating. The orchestration boundary
-functions (`execute` / `execute_subtree`) still return `Result<String, String>` because they
-are registered with duroxide:
+SQL, HTTP, and multipart activity scheduling boundaries explicitly map activity
+errors to `NodeError::Application`; structural/configuration helper errors use
+`NodeError::Failure`. `execute_loop_node` is the only handler that catches
+`NodeError::Break` (turning it into the loop's `Ok` result). The orchestration
+boundary functions (`execute` / `execute_subtree`) still return
+`Result<String, String>` because they are registered with duroxide:
 
 - `execute`: an uncaught top-level `Break` becomes a clear `Err` ("df.break() was called
   outside of a loop"), so the instance fails instead of completing with a sentinel value.
-- `execute_subtree` (used by JOIN/RACE branches): a `Break` is carried out-of-band in the
-  subtree envelope's `control` field and re-raised as `NodeError::Break` by
-  `parse_subtree_envelope` in the parent orchestration.
+- `execute_subtree`: a `Break` is carried out-of-band in the subtree envelope's
+  `control` field. `NodeError::Application` is encoded as a namespaced,
+  serde-tagged subtree failure so application classification survives nested
+  JOIN, RACE, and LOOP boundaries. `NodeError::Failure` remains an unrecognized
+  child error and propagates fatally.
 
 The orchestration walks the graph recursively:
 
@@ -696,34 +701,25 @@ pub fn is_truthy(value: &serde_json::Value) -> bool {
 
 ### Parallel Execution (JOIN/RACE)
 
-JOIN and RACE use duroxide's sub-orchestration support:
-
-```rust
-async fn execute_join_node(...) -> Result<String, String> {
-    let left_id = node.left_node.as_ref().ok_or("JOIN missing left")?;
-    let right_id = node.right_node.as_ref().ok_or("JOIN missing right")?;
-    
-    // Create sub-orchestration inputs
-    let left_input = create_subtree_input(graph, left_id, results);
-    let right_input = create_subtree_input(graph, right_id, results);
-    
-    // Schedule parallel sub-orchestrations
-    let left_handle = ctx.schedule_orchestration(SUBTREE_NAME, &left_id, left_input);
-    let right_handle = ctx.schedule_orchestration(SUBTREE_NAME, &right_id, right_input);
-    
-    // Wait for all to complete (duroxide handles parallelism)
-    let (left_result, right_result) = tokio::join!(
-        left_handle.into_orchestration(),
-        right_handle.into_orchestration()
-    );
-    
-    // Combine results
-    let results = vec![left_result?, right_result?];
-    Ok(serde_json::to_string(&results)?)
-}
-```
-
-For RACE, duroxide's `select` is used to return the first completed result.
+Each JOIN or RACE branch runs as an explicitly named `execute_subtree` child
+whose input contains the validated graph snapshot, variables, label, and
+canonically serialized named results. The child returns a `SubtreeEnvelope`
+containing its result, named-result updates, and optional `df.break()` control
+flow.
+
+JOIN schedules its two branches, plus any ordered `join3` extras, then waits
+with `ctx.join()`. Successful envelopes are processed in branch order and
+their named results are merged into the parent. Historically, JOIN returned
+the first branch error in that same order. That behavior remains unchanged
+outside failure-isolated loop bodies for replay compatibility. Inside a
+failure-isolated body, every settled outcome is inspected so a fatal sibling
+cannot be hidden by a recoverable activity failure or `df.break()`. The
+deterministic priority is fatal failure, then break, then recoverable activity
+failure; equal-priority outcomes keep the first branch.
+
+RACE uses `ctx.select2()` and returns the first completed branch. The losing
+branch is cancelled, and a losing loop branch receives a terminal fallback
+node-status stamp because cancellation may stop it before it can stamp itself.
 
 ### Loops and Continue-As-New
 
@@ -734,7 +730,27 @@ Loops use duroxide's `continue_as_new` to avoid unbounded history growth. Their
 
 `execute_subtree` is therefore structurally identical to `execute_function_graph`: both root an execution context at their own node and host an inline root loop. They differ only in the input envelope they re-enter with on `continue_as_new` (`FunctionInput` vs `SubtreeInput`), in the fact that only the root orchestration touches instance-level status, and in where their graph comes from — `execute_function_graph` loads it from `df.nodes` on its first generation, while `execute_subtree` receives it inline from its parent. The graph is loaded exactly once per instance and then carried inline through every child input and every `continue_as_new` generation, so `submitted_by` is fixed for the instance's lifetime and a post-start `df.nodes` tamper is never read. Role deletion and privilege revocation are still enforced per node execution, by connecting *as* `submitted_by` for SQL and by re-checking `EXECUTE` privilege per HTTP request.
 
-Both paths call `run_loop_iteration`, which executes the body, catches `NodeError::Break`, evaluates the optional post-body condition, and propagates `NodeError::Failure`. A child stamps its LOOP node `running` on each generation and `completed` or `failed` on exit; because `continue_as_new` returns a future that never resolves, a continuing generation never stamps a terminal status. If a live loop loses a RACE, the parent records the loop node as terminal `failed` with a cancellation reason because duroxide cancellation stops the child before it can run its own terminal stamp.
+Fail-fast loops call `run_loop_iteration`, which executes the body inline,
+catches `NodeError::Break`, evaluates the optional post-body condition, and
+propagates both application and non-application failures. A loop configured
+with `continue_on_failure => true` instead calls
+`run_failure_isolated_body`: each body iteration runs as a fresh
+`execute_subtree` child. On success, the subtree envelope is parsed and its
+named results are merged into the parent result map before the parent evaluates
+the optional condition. On a structurally encoded body application failure,
+the parent consumes the failure, skips the condition because body results may
+be absent, and advances to the next generation. Every error returned by a body
+SQL, HTTP, or multipart activity uses this application-failure path, including
+query, authorization, connection, and network errors. Condition failures,
+malformed envelopes or graph data, child-ID collisions, and unrecognized
+orchestration/runtime failures remain fatal.
+
+A child stamps its LOOP node `running` on each generation and `completed` or
+`failed` on exit; because `continue_as_new` returns a future that never
+resolves, a continuing generation never stamps a terminal status. If a live
+loop loses a RACE, the parent records the loop node as terminal `failed` with a
+cancellation reason because duroxide cancellation stops the child before it
+can run its own terminal stamp.
 
 Node status stamps contain the full composed orchestration lineage: `{root_instance}::{generation}::{child_node}::{generation}...`. Read-time inference and the write fence walk that lineage so stale writes and superseded nested branches are evaluated at every ancestor generation.