feat(jobs): per-filter event cursors, multi-replica tests, lease docs (closes #1252)

docs/concepts/jobs.md

@@ -485,9 +485,42 @@ such jobs run unbounded, subject only to lease recovery.
 
 For a job that runs for hours: set `max_duration` to its real
 per-attempt runtime (this sizes the lease AND floors the stale
-ceiling) and/or emit periodic progress to keep the lease renewed. Set
-`total_deadline` if you want a hard total-runtime bound across all
-attempts, or to opt out of the registry-wide stale ceiling entirely.
+ceiling). A handler sitting in `recv_event` gates renews the lease on
+every poll; one that computes silently for long stretches should emit
+periodic progress to keep the lease renewed. Set `total_deadline` if you
+want a hard total-runtime bound across all attempts, or to opt out of the
+registry-wide stale ceiling entirely.
+
+`MCP_MESH_JOB_STALE_TIMEOUT` is a **reap ceiling, not the lease window** —
+it caps total runtime from submission but does not size or extend the
+per-attempt lease. The lease window is derived per-job from `max_duration`
+(300s default) and is renewed by progress deltas and executor `recv_event`
+polls; tuning the stale timeout does not change how quickly a quiet
+handler's lease expires.
+
+### Multi-replica fencing and poll-liveness
+
+Several replicas can declare the same `task=true` capability. Each job is
+claimed by exactly one replica per attempt (the guarded atomic `UPDATE`
+above, so concurrent claimers race and one wins). The owner's lease is
+renewed by **any accepted non-terminal delta and any executor `recv_event`
+poll from the current claim** — a handler parked in a legitimate
+`recv_event` gate is provably alive, so no artificial `update_progress`
+keepalives are needed. Poll-liveness is capped so it never pushes the lease
+past `total_deadline` or the stale ceiling.
+
+If a lease genuinely expires (a handler wedged, neither progressing nor
+polling), the sweep returns the job to the pool and a peer re-claims it.
+Every claim — including a same-replica re-claim — carries a monotonically
+increasing **claim epoch**. The registry fences the superseded execution:
+its writes are rejected `claim_superseded` and its cancel token fires, so a
+parked `recv_event` breaks out with the cancelled error and the handler
+observes cancellation exactly as it would a user cancel. Handlers should
+honor cancellation promptly; the read-only `claimEpoch` accessor on the job
+context (`job.claim_epoch` in Python, `currentJob()?.claimEpoch` in
+TypeScript, `JobContext.current().claimEpoch` in Java) can be stamped onto
+external side effects so downstream can dedupe a fenced re-execution's
+duplicate write.
 
 The **agent runtime** is a thin client. On the producer side, it
 maintains a claim queue per `task=true` capability (`UPDATE jobs SET
@@ -564,7 +597,7 @@ sequenceDiagram
     Note over R: Append-only log<br/>(per-job seq)
     H->>R: GET /jobs/{id}/events?after=N  (long-poll)
     R-->>H: event dict (or empty after timeout)
-    Note over H: Cursor advances<br/>per-controller
+    Note over H: Cursor advances<br/>per-controller, per-filter
 ```
 
 ### Receiving events inside a handler
@@ -640,9 +673,15 @@ The producer-side recv loop. Filter by `types` to drop noise, set a
     ```
 <!-- markdownlint-enable MD046 -->
 
-The handler's `recv_event` cursor is per-`JobController` instance — a
-fresh controller for the same `job_id` replays from `seq=0`. The
-running handler always sees events in monotonic seq order.
+The handler's `recv_event` cursor is per-`JobController` instance **and
+per-`types` filter**: each distinct filter is an independent stream with
+its own cursor, so interleaving `recv_event(types=["A"])` and
+`recv_event(types=["B"])` never lets one filter's consumption skip the
+other's earlier events. Delivery is exactly-once within a filter stream
+and at-least-once across different filters. A fresh controller for the
+same `job_id` (e.g. a re-claim) replays every filter from `seq=0`; the
+running handler always sees events in monotonic seq order within a
+stream.
 
 ### Posting events from outside the handler
 

src/core/cli/man/content/environment.md

@@ -531,6 +531,12 @@ export MCP_MESH_SWEEP_INTERVAL=5m
 # set their own total_deadline are fully exempt. Default: off (unset /
 # "0") — jobs without an explicit total_deadline run unbounded, subject
 # only to lease recovery.
+#
+# NOT the lease window. This is a total-runtime REAP CEILING (measured
+# from submission); it does not size or extend the per-attempt lease. The
+# lease window is derived per-job from max_duration (300s default) and is
+# renewed by progress deltas / recvEvent polls — tuning this variable does
+# NOT change how quickly a quiet handler's lease expires.
 export MCP_MESH_JOB_STALE_TIMEOUT=2h
 
 # CORS

src/core/cli/man/content/jobs.md

@@ -237,6 +237,17 @@ elif event["type"] == "cancelled":
     return {"status": "cancelled", "reason": event["payload"].get("reason")}
 ```
 
+Each distinct `types` filter is an **independent event stream** with its
+own cursor, so interleaving `recv_event(types=["A"])` and
+`recv_event(types=["B"])` never lets one filter's consumption skip the
+other's earlier events. Delivery is **exactly-once within a filter stream**
+and **at-least-once across different filters** (an event matching two
+filters can surface once per stream). The cursor is per-handler and starts
+at the beginning of the log on every (re-)claim — a re-claimed handler
+replays from the start. Handlers doing **non-idempotent** work per event
+should checkpoint (e.g. persist their own event cursor) or design
+idempotent phases.
+
 **Outside the handler, with a `job_id` in scope — fire an event:**
 
 ```python
@@ -415,13 +426,22 @@ intervention:
   the matching capability picks it up. Jobs parked in `input_required`
   are covered too — a job waiting on a consumer answer whose owner then
   dies is reclaimed, not stranded.
-- **Lease recovery.** Every accepted progress/heartbeat delta extends
-  the owner's lease. A job whose lease expires with no further deltas —
-  a wedged or silently-crashed handler — is reset to claimable while
-  retries remain, or marked `failed` once the retry budget is spent.
-  This includes jobs parked in `input_required`: if the producer stops
-  extending the lease while waiting for an answer that never comes, the
-  job is reclaimed rather than held forever.
+- **Lease recovery.** The lease window is derived from `max_duration`
+  (the 300s claim default when undeclared) — declare `max_duration` ≈
+  the job's real per-attempt ceiling for long-running work. The lease
+  renews on **any accepted non-terminal delta AND any `recv_event` poll
+  from the current claim**: a handler parked in a legitimate `recv_event`
+  gate is provably alive, so no artificial `update_progress` keepalives
+  are needed. Poll-liveness never pushes the lease past the job's
+  `total_deadline` or stale ceiling, so an actively-polling handler
+  cannot outlive the point where the sweep would legitimately reap it. A
+  job whose lease expires with no renewal — a genuinely wedged handler,
+  neither progressing nor polling — is reset to claimable while retries
+  remain, or marked `failed` once the retry budget is spent. This
+  includes jobs parked in `input_required`: a handler that crashes while
+  awaiting an answer stops renewing and is reclaimed rather than held
+  forever (a live handler polling `recv_event` for the answer keeps its
+  lease renewed).
 - **Total-deadline ceiling.** A job that set `total_deadline` is failed
   with `deadline_exceeded` once that wall-clock deadline passes.
 - **Default stale ceiling (opt-in).** Set `MCP_MESH_JOB_STALE_TIMEOUT`
@@ -441,10 +461,45 @@ Reaping is observable in-handler via a synthetic `stale` event — see
 
 **Long-running (hours) jobs.** Set `max_duration` to the job's real
 per-attempt runtime — this sizes the lease window AND floors the stale
-ceiling — and/or emit periodic progress to keep the lease renewed. Set
+ceiling. A handler sitting in `recv_event` gates renews the lease on
+every poll; one that computes silently for long stretches should emit
+periodic `update_progress` to keep the lease renewed. Set
 `total_deadline` if you want a hard total-runtime bound across all
 attempts, or to opt out of the registry-wide stale ceiling entirely.
 
+## Multi-replica execution and fencing
+
+Several replicas may declare the same `task=True` capability. Each job is
+claimed by **exactly one** replica per attempt — the claim is a guarded
+atomic update, so concurrent claimers race and only one wins. If a lease
+genuinely expires (a handler wedged, neither progressing nor polling), the
+sweep returns the job to the pool and a peer re-claims it.
+
+Every claim — including a re-claim of the same job, even by the same
+replica — carries a **monotonically increasing claim epoch**. The registry
+fences the superseded execution: its writes are rejected
+(`claim_superseded`) and it observes cancellation through the **same
+surface as a user cancel** — the handler's cancel token fires, so a parked
+`recv_event` raises the cancelled error and the next `await` point raises
+`asyncio.CancelledError`. Honor cancellation promptly (return / stop side
+effects) instead of running to completion under a lost claim.
+
+For a side effect a fenced re-execution might repeat, stamp it with the
+claim epoch so downstream can dedupe:
+
+```python
+@app.tool()
+@mesh.tool(capability="charge_card", task=True)
+async def charge_card(order_id: str, amount: float, job: mesh.MeshJob = None) -> dict:
+    # job.claim_epoch is the generation this attempt runs under (int, or
+    # None for a push-mode inbound job / an old registry). Read-only and
+    # additive — supersession works without it (it rides the cancellation
+    # path); the epoch exists only so downstream can distinguish a fenced
+    # re-execution's duplicate write.
+    await ledger.upsert(key=order_id, claim_epoch=job.claim_epoch, amount=amount)
+    ...
+```
+
 ## Out-of-band inspection
 
 Three SDK-managed helper tools are auto-registered on every mesh

src/core/cli/man/content/jobs_java.md

@@ -345,6 +345,17 @@ public Map<String, Object> runWorkflow(
 }
 ```
 
+Each distinct `types` filter is an **independent event stream** with its
+own cursor, so interleaving `recvEvent(List.of("A"), ...)` and
+`recvEvent(List.of("B"), ...)` never lets one filter's consumption skip
+the other's earlier events. Delivery is **exactly-once within a filter
+stream** and **at-least-once across different filters** (an event matching
+two filters can surface once per stream). The cursor is per-handler and
+starts at the beginning of the log on every (re-)claim — a re-claimed
+handler replays from the start. Handlers doing **non-idempotent** work per
+event should checkpoint (e.g. persist their own event cursor) or design
+idempotent phases.
+
 **Outside the handler, with a `jobId` in scope — fire an event:**
 
 ```java
@@ -601,13 +612,22 @@ intervention:
   the matching capability picks it up. Jobs parked in `input_required`
   are covered too — a job waiting on a consumer answer whose owner then
   dies is reclaimed, not stranded.
-- **Lease recovery.** Every accepted progress/heartbeat delta extends
-  the owner's lease. A job whose lease expires with no further deltas —
-  a wedged or silently-crashed handler — is reset to claimable while
-  retries remain, or marked `failed` once the retry budget is spent.
-  This includes jobs parked in `input_required`: if the producer stops
-  extending the lease while waiting for an answer that never comes, the
-  job is reclaimed rather than held forever.
+- **Lease recovery.** The lease window is derived from `maxDuration`
+  (the 300s claim default when undeclared) — declare `maxDuration` ≈
+  the job's real per-attempt ceiling for long-running work. The lease
+  renews on **any accepted non-terminal delta AND any `recvEvent` poll
+  from the current claim**: a handler parked in a legitimate `recvEvent`
+  gate is provably alive, so no artificial `updateProgress` keepalives
+  are needed. Poll-liveness never pushes the lease past the job's
+  `totalDeadline` or stale ceiling, so an actively-polling handler
+  cannot outlive the point where the sweep would legitimately reap it. A
+  job whose lease expires with no renewal — a genuinely wedged handler,
+  neither progressing nor polling — is reset to claimable while retries
+  remain, or marked `failed` once the retry budget is spent. This
+  includes jobs parked in `input_required`: a handler that crashes while
+  awaiting an answer stops renewing and is reclaimed rather than held
+  forever (a live handler polling `recvEvent` for the answer keeps its
+  lease renewed).
 - **Total-deadline ceiling.** A job that set `totalDeadline` is failed
   with `deadline_exceeded` once that wall-clock deadline passes.
 - **Default stale ceiling (opt-in).** Set `MCP_MESH_JOB_STALE_TIMEOUT`
@@ -627,9 +647,44 @@ Reaping is observable in-handler via a synthetic `stale` event — see
 
 **Long-running (hours) jobs.** Set `maxDuration` to the job's real
 per-attempt runtime — this sizes the lease window AND floors the stale
-ceiling — and/or emit periodic progress to keep the lease renewed. Set
-`totalDeadline` if you want a hard total-runtime bound across all
-attempts, or to opt out of the registry-wide stale ceiling entirely.
+ceiling. A handler sitting in `recvEvent` gates renews the lease on
+every poll; one that computes silently for long stretches should emit
+periodic `updateProgress` to keep the lease renewed. Set `totalDeadline`
+if you want a hard total-runtime bound across all attempts, or to opt out
+of the registry-wide stale ceiling entirely.
+
+## Multi-replica execution and fencing
+
+Several replicas may declare the same `task = true` capability. Each job is
+claimed by **exactly one** replica per attempt — the claim is a guarded
+atomic update, so concurrent claimers race and only one wins. If a lease
+genuinely expires (a handler wedged, neither progressing nor polling), the
+sweep returns the job to the pool and a peer re-claims it.
+
+Every claim — including a re-claim of the same job, even by the same
+replica — carries a **monotonically increasing claim epoch**. The registry
+fences the superseded execution: its writes are rejected
+(`claim_superseded`) and it observes cancellation the same way a user
+cancel surfaces — `controller.isCancelled()` turns `true` (poll it between
+work units) and a parked `recvEvent` breaks out with the cancelled error.
+Honor cancellation promptly (return / stop side effects) instead of running
+to completion under a lost claim.
+
+For a side effect a fenced re-execution might repeat, stamp it with the
+claim epoch so downstream can dedupe. Read the epoch from the active job
+context:
+
+```java
+import io.mcpmesh.JobContext;
+
+// JobContext.current().claimEpoch is the generation this attempt runs
+// under (Long, or null for a push-mode inbound job / an old registry).
+// Read-only and additive — supersession works without it (it rides the
+// cancellation path); the epoch exists only so downstream can distinguish
+// a fenced re-execution's duplicate write.
+Long claimEpoch = JobContext.current().claimEpoch;
+ledger.upsert(orderId, claimEpoch, amount);
+```
 
 ## Out-of-band inspection