diff --git a/docs/concepts/jobs.md b/docs/concepts/jobs.md
index d36603f3c..7ec572776 100644
--- a/docs/concepts/jobs.md
+++ b/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
(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
per-controller
+ Note over H: Cursor advances
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
```
-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
diff --git a/src/core/cli/man/content/environment.md b/src/core/cli/man/content/environment.md
index 93e047fbc..d3b6dec7d 100644
--- a/src/core/cli/man/content/environment.md
+++ b/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
diff --git a/src/core/cli/man/content/jobs.md b/src/core/cli/man/content/jobs.md
index b1761ce2d..6d5976fd5 100644
--- a/src/core/cli/man/content/jobs.md
+++ b/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
diff --git a/src/core/cli/man/content/jobs_java.md b/src/core/cli/man/content/jobs_java.md
index 71fee49f7..5936e1c63 100644
--- a/src/core/cli/man/content/jobs_java.md
+++ b/src/core/cli/man/content/jobs_java.md
@@ -345,6 +345,17 @@ public Map 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
diff --git a/src/core/cli/man/content/jobs_typescript.md b/src/core/cli/man/content/jobs_typescript.md
index ccc4b46ba..40df43000 100644
--- a/src/core/cli/man/content/jobs_typescript.md
+++ b/src/core/cli/man/content/jobs_typescript.md
@@ -272,6 +272,16 @@ if (event === null) {
}
```
+Each distinct `types` filter is an **independent event stream** with its
+own cursor, so interleaving `recvEvent(["A"])` and `recvEvent(["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:**
```typescript
@@ -463,13 +473,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`
@@ -489,9 +508,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 through the **same
+surface as a user cancel** — the handler's `AbortSignal` (`job?.signal`)
+aborts and a parked `recvEvent` rejects 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:
+
+```typescript
+import { currentJob } from "@mcpmesh/sdk";
+
+// currentJob()?.claimEpoch is the generation this attempt runs under
+// (number, 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.
+const claimEpoch = currentJob()?.claimEpoch ?? null;
+await ledger.upsert({ key: orderId, claimEpoch, amount });
+```
## Out-of-band inspection
diff --git a/src/runtime/core/src/jobs.rs b/src/runtime/core/src/jobs.rs
index 104bec551..fad651d99 100644
--- a/src/runtime/core/src/jobs.rs
+++ b/src/runtime/core/src/jobs.rs
@@ -10,7 +10,7 @@
use std::collections::HashMap;
use std::future::Future;
-use std::sync::atomic::{AtomicI64, Ordering};
+use std::sync::atomic::Ordering;
use std::sync::Arc;
use std::time::Duration;
@@ -296,10 +296,12 @@ async fn flush_once(
///
/// Also exposes [`Self::recv_event`] for handlers running inside a
/// `task=True` job to drain events posted via [`JobProxy::send_event`].
-/// The per-controller `last_seen_seq` cursor is shared across `Clone`s of
-/// the same instance (via `Arc`), so passing the controller
-/// into helper tasks doesn't reset the cursor. A NEW controller created
-/// for the same `job_id` starts from seq=0 — replay is per-instance.
+/// Event delivery uses PER-FILTER cursors (issue #1252 Phase 3): each
+/// distinct type filter is an independent stream with its own cursor,
+/// shared across `Clone`s of the same instance, so consuming a type-A
+/// match at seq N no longer skips a type-B event at seq < N. A NEW
+/// controller for the same `job_id` starts every cursor at seq=0 — replay
+/// is per-instance, per filter.
#[derive(Clone)]
pub struct JobController {
job_id: String,
@@ -324,20 +326,31 @@ pub struct JobController {
/// queue-resident so the shared claim-worker queue doesn't accumulate
/// one entry per completed job forever (issue #1166 LOW).
terminal: Arc,
- /// Last event seq returned by `recv_event` on this controller (or any
- /// of its `Clone`s). `0` means no events consumed yet — the next
- /// `recv_event` will return the first event in the job's event log.
- last_seen_seq: Arc,
- /// Serialises the entire `recv_event` load→list→store window across
- /// `Clone`s of this controller. Without this, two concurrent callers
- /// on clones can both load the same `last_seen_seq`, both call
- /// `list_job_events`, and both return the SAME head event — the
- /// AtomicI64 prevents memory tearing but not the read-fetch-write
- /// race across the `.await`. The `Arc` keeps clones cheap while
- /// pinning every clone to the same lock instance; lock contention
- /// is naturally bounded to the small number of helper tasks a
- /// single handler typically spawns.
- recv_event_lock: Arc>,
+ /// Per-filter event cursors (issue #1252 Phase 3). Keyed by canonical
+ /// filter identity ([`Self::filter_key`]): the sorted+deduped type list,
+ /// with `None`/empty (the unfiltered stream) collapsing to one distinct
+ /// key. Each `recv_event` reads/advances ONLY its own filter's cursor,
+ /// so consuming a type-A match at seq N no longer permanently skips a
+ /// type-B event at seq < N (the shared-cursor defect).
+ ///
+ /// Stream semantics: delivery is exactly-once WITHIN a filter stream; an
+ /// event matching two DIFFERENT filter streams may be observed once per
+ /// stream (intentional at-least-once ACROSS streams — consistent with
+ /// the at-least-once job model). Shared across `Clone`s; a NEW controller
+ /// for the same `job_id` starts every cursor at 0 (replay-from-0 per
+ /// filter on re-claim). Advancement is monotonic per key (never retreats).
+ cursors: Arc>>,
+ /// Per-filter serialization locks. `recv_event` holds ONLY its own
+ /// filter's async lock across the load→list→store window, so concurrent
+ /// calls on DIFFERENT filters make progress independently (a 60s type-A
+ /// long-poll must NOT block a type-B call — the whole point of per-filter
+ /// cursors), while two calls on the SAME filter serialize so they can't
+ /// both load the same cursor, both `list_job_events` with the same
+ /// `after`, and both return the SAME head event (the read-fetch-write
+ /// race across the `.await`). Shared across `Clone`s; the map grows one
+ /// small entry per distinct filter the handler uses (bounded), never per
+ /// event.
+ recv_locks: Arc>>>>,
}
impl JobController {
@@ -374,8 +387,45 @@ impl JobController {
backend,
queue,
terminal: Arc::new(std::sync::atomic::AtomicBool::new(false)),
- last_seen_seq: Arc::new(AtomicI64::new(0)),
- recv_event_lock: Arc::new(Mutex::new(())),
+ cursors: Arc::new(std::sync::Mutex::new(HashMap::new())),
+ recv_locks: Arc::new(std::sync::Mutex::new(HashMap::new())),
+ }
+ }
+
+ /// Canonical identity of a `recv_event` type filter, used as the
+ /// per-filter cursor key (issue #1252 Phase 3). `None`, an empty list, and
+ /// a list of only empty/whitespace strings all mean "all events" and
+ /// collapse to ONE unfiltered key (the empty string): each entry is
+ /// trimmed and empty entries are dropped, mirroring the registry's own
+ /// `types` parsing (`strings.TrimSpace` + drop-empty in
+ /// `ent_handlers_jobs.go`) so `Some(vec!["".into()])` canonicalizes — and
+ /// filters — identically on both backends. The surviving entries are then
+ /// sorted + deduped and joined with a non-printable unit separator, so
+ /// `["B","A"]`, `["A","B"]`, and `["A","A","B"]` share one cursor while
+ /// `["AB"]` and `["A","B"]` do NOT collide.
+ ///
+ /// Unsupported edge inputs (documented, not defended — no known caller
+ /// passes them): a type name that itself contains the U+001F unit
+ /// separator collides with the joined form of a multi-type filter (e.g.
+ /// `["A\u{1f}B"]` shares a key with `["A","B"]`), and a type name
+ /// containing a comma (`["a,b"]`) is a single filter on the client but,
+ /// once serialized as the comma-joined `types` query param and re-split by
+ /// the registry, becomes the server-identical filter `["a","b"]` — so two
+ /// distinct client cursors map to one server-side filter. Callers pass
+ /// plain event-type identifiers, which never contain these bytes.
+ fn filter_key(types: &Option>) -> String {
+ match types {
+ None => String::new(),
+ Some(ts) => {
+ let mut v: Vec<&str> = ts
+ .iter()
+ .map(|s| s.trim())
+ .filter(|s| !s.is_empty())
+ .collect();
+ v.sort_unstable();
+ v.dedup();
+ v.join("\u{1f}")
+ }
}
}
@@ -651,14 +701,19 @@ impl JobController {
///
/// On first call (or for a freshly-constructed controller) starts from
/// `seq=0` and returns the first matching event in the log. Subsequent
- /// calls return events strictly newer than the last consumed — the
- /// internal cursor is incremented on every successful return. The
- /// cursor is per-`JobController`-instance (shared across `Clone`s of
- /// the same instance via `Arc`); a NEW controller for the
- /// same `job_id` starts replay from `seq=0`.
+ /// calls return events strictly newer than the last consumed FOR THE
+ /// SAME FILTER — each distinct `types` filter has its OWN cursor (issue
+ /// #1252 Phase 3), so interleaving `recv_event(A)` and `recv_event(B)`
+ /// never lets one filter's consumption skip the other's earlier events.
+ /// Cursors are per-`JobController`-instance (shared across `Clone`s); a
+ /// NEW controller for the same `job_id` replays every filter from
+ /// `seq=0`.
///
/// `types`: if `Some`, only events whose `event_type` matches one of
/// the provided strings are returned. If `None`, all event types match.
+ /// The filter identity is canonical (order- and duplicate-insensitive;
+ /// `None` and `Some([])` are the same unfiltered stream) — see
+ /// [`Self::filter_key`].
///
/// `timeout` bounds the long-poll. `None` waits indefinitely (loops
/// the registry's 60s cap until an event arrives). `Some(d)` returns
@@ -689,16 +744,65 @@ impl JobController {
// Server-side limit (matches OpenAPI `limit` maximum).
const FETCH_LIMIT: usize = 100;
- // Serialise the load→list→store window across `Clone`s of this
- // controller. Without this, two concurrent callers can both load
- // the same `last_seen_seq` cursor, both issue `list_job_events`
- // with the same `after`, and both return the SAME head event.
- // The AtomicI64 prevents tearing but not the read-fetch-write
- // race across the `.await`. Acquire BEFORE the deadline check
- // so a fast clone-vs-clone race ordering is deterministic.
- let _guard = self.recv_event_lock.lock().await;
+ // Per-filter cursor + serialization (issue #1252 Phase 3). This
+ // filter's identity keys BOTH its cursor and its serialization lock.
+ // Hold ONLY this filter's lock across the load→list→store window:
+ // two calls on the SAME filter serialize (so they can't both load the
+ // same cursor, both `list_job_events` with the same `after`, and both
+ // return the SAME head event across the `.await`), while calls on
+ // DIFFERENT filters run concurrently — a 60s type-A long-poll must not
+ // block a type-B call.
+ let key = Self::filter_key(&types);
+ let filter_lock = {
+ let mut locks = self.recv_locks.lock().expect("recv_locks poisoned");
+ locks
+ .entry(key.clone())
+ .or_insert_with(|| Arc::new(Mutex::new(())))
+ .clone()
+ };
+ // Compute the deadline at ENTRY — BEFORE awaiting the per-filter lock
+ // — so time spent QUEUED behind a same-filter long-poll counts against
+ // the caller's OWN budget (issue #1252 review). Acquiring the lock
+ // first and only THEN starting the clock let a queued same-filter
+ // caller run for lock-wait + full-budget, blowing past its `timeout`.
let deadline = timeout.map(|t| Instant::now() + t);
+
+ // Snapshot this execution's cancel/supersession token so the lock
+ // acquisition below can be raced against it: a cancelled execution
+ // must NOT sit parked on the per-filter lock waiting out a sibling's
+ // 60s long-poll. Watch THIS execution's OWN frame (keyed by claim
+ // epoch); a legacy controller (no epoch) watches the top-of-stack.
+ let queue_cancel_token = match self.claim_epoch {
+ Some(epoch) => cancel_registry::get_state_for_epoch(&self.job_id, epoch),
+ None => cancel_registry::get_state(&self.job_id),
+ }
+ .map(|(cancel, _ended)| cancel);
+
+ // Acquire the per-filter serialization lock, racing it against BOTH
+ // the caller's remaining budget (return `Ok(None)` on exhaustion — the
+ // same shape a normal long-poll timeout takes) AND the cancel token
+ // (return `Cancelled`). `wait_on_optional_cancel` never completes when
+ // there is no token; the deadline branch never completes when
+ // `timeout` is `None` — so an untimed, uncancellable call simply awaits
+ // the lock. Once acquired, the loop below derives every poll's budget
+ // from `deadline`, so wait-on-lock time is already subtracted.
+ let _guard = {
+ let deadline_sleep = async {
+ match deadline {
+ Some(d) => tokio::time::sleep_until(d).await,
+ None => std::future::pending::<()>().await,
+ }
+ };
+ tokio::select! {
+ biased;
+ _ = wait_on_optional_cancel(&queue_cancel_token) => {
+ return Err(JobError::Cancelled)
+ }
+ _ = deadline_sleep => return Ok(None),
+ g = filter_lock.lock() => g,
+ }
+ };
// Tracks the start of the current contiguous transient-failure
// window for the same recovery semantics as `JobProxy::wait`.
let mut transient_failure_started: Option = None;
@@ -758,7 +862,15 @@ impl JobController {
}
}
- let after = self.last_seen_seq.load(Ordering::SeqCst);
+ // Read THIS filter's cursor (default 0 for a never-consumed
+ // filter). The per-filter lock above serializes same-filter
+ // callers, so the load→list→store window is race-free.
+ let after = *self
+ .cursors
+ .lock()
+ .expect("cursors poisoned")
+ .get(&key)
+ .unwrap_or(&0);
let types_ref = types.as_deref();
// Executor identity: the OWNER's controller supplies
// `(instance_id, claim_epoch)` so the registry fences this read
@@ -793,19 +905,40 @@ impl JobController {
backoff_ms = POLL_INITIAL_MS;
// First matching event wins. The registry already
// filters by `types` and orders ascending; we return
- // the head and advance the cursor to its seq so the
- // next call picks up strictly after it.
+ // the head and advance THIS filter's cursor to its seq so
+ // the next same-filter call picks up strictly after it.
+ // Advance monotonically (never retreat).
if let Some(ev) = resp.events.into_iter().next() {
- self.last_seen_seq.store(ev.seq, Ordering::SeqCst);
+ let mut cursors = self.cursors.lock().expect("cursors poisoned");
+ let cur = cursors.entry(key.clone()).or_insert(0);
+ if ev.seq > *cur {
+ *cur = ev.seq;
+ }
return Ok(Some(ev));
}
- // Empty page: keep next_after as the cursor (matches
- // the registry's contract — the watermark may have
- // advanced even though no matching event arrived).
- // We only ever advance, never retreat.
- let cur = self.last_seen_seq.load(Ordering::SeqCst);
- if resp.next_after > cur {
- self.last_seen_seq.store(resp.next_after, Ordering::SeqCst);
+ // Empty (no-match) page. Under the registry's contract a
+ // no-match page returns `next_after == after` — the
+ // watermark does NOT leap past events this filter didn't
+ // return. The `list_events_mock_matches_registry_contract`
+ // test pins that the in-process mock reproduces this
+ // contract AND that RegistryHttpBackend sends the right
+ // request (`after` + `types`) and parses the reply the same
+ // way; the registry's own adherence to the contract is
+ // pinned Go-side (see
+ // `src/core/registry/ent_handlers_job_events_test.go`). So
+ // this branch is a DEFENSIVE no-op under that
+ // contract: it advances this filter's cursor only if a
+ // backend ever reported `next_after` strictly ahead of the
+ // cursor, and it never retreats. It must NOT advance past
+ // unreturned events for OTHER filters — the cursor is
+ // per-filter, so an unfiltered `next_after` can never
+ // clobber a type filter's position.
+ if resp.next_after > after {
+ let mut cursors = self.cursors.lock().expect("cursors poisoned");
+ let cur = cursors.entry(key.clone()).or_insert(0);
+ if resp.next_after > *cur {
+ *cur = resp.next_after;
+ }
}
// No event yet — loop. The deadline check at the top
// of the next iteration returns Ok(None) if expired.
@@ -1324,10 +1457,22 @@ mod tests {
use crate::task_backend::{
CancelJobResponse, ClaimedJob, JobBatchResponse, JobEventListResponse,
+ RegistryHttpBackend,
};
// ---- mock backend -------------------------------------------------------
+ /// Test gate for [`MockBackend::list_job_events`]. When installed, an
+ /// EMPTY result page fires `entered` and then parks on `release` — letting
+ /// a test PROVE a `recv_event` long-poll is in-flight (and thus holding
+ /// its per-filter lock) before a second call starts, closing the "did the
+ /// two calls actually overlap?" gap. `release` is level-triggered: once
+ /// cancelled, subsequent empty pages return without blocking.
+ struct ListGate {
+ entered: tokio::sync::Notify,
+ release: CancellationToken,
+ }
+
/// What `get_job` should return for the next call.
enum GetJobInjection {
/// Return the in-memory job state normally.
@@ -1381,6 +1526,10 @@ mod tests {
always_transient: AtomicUsize, // 0 = false, 1 = true
/// If set, every `get_job` returns NotFound (non-transient).
always_not_found: AtomicUsize,
+ /// Optional test gate: when installed, `list_job_events` parks on an
+ /// EMPTY result page so a test can hold a `recv_event` long-poll open
+ /// deterministically. See [`ListGate`].
+ list_gate: StdMutex>>,
}
impl MockBackend {
@@ -1402,8 +1551,20 @@ mod tests {
transient_remaining: AtomicUsize::new(0),
always_transient: AtomicUsize::new(0),
always_not_found: AtomicUsize::new(0),
+ list_gate: StdMutex::new(None),
})
}
+ /// Install a gate so the next `list_job_events` calls that hit an EMPTY
+ /// page fire `entered` then park on `release`. Returns the gate so the
+ /// test can await `entered` and later `release.cancel()` to unblock.
+ fn install_list_gate(&self) -> Arc {
+ let gate = Arc::new(ListGate {
+ entered: tokio::sync::Notify::new(),
+ release: CancellationToken::new(),
+ });
+ *self.list_gate.lock().unwrap() = Some(gate.clone());
+ gate
+ }
fn push_event(&self, job_id: &str, event_type: &str, payload: serde_json::Value) {
let mut all = self.events.lock().unwrap();
let list = all.entry(job_id.to_string()).or_default();
@@ -1713,35 +1874,62 @@ mod tests {
}
}
}
- let all = self.events.lock().unwrap();
- let list = match all.get(job_id) {
- Some(l) => l,
- None => {
+ // Canonicalize the `types` filter exactly as the registry does
+ // (`ent_handlers_jobs.go`): trim each entry and drop empties. An
+ // absent filter, an empty list, or a list of only empty/whitespace
+ // strings all match everything — so `types=[""]` behaves as
+ // unfiltered on the mock just as it does on the real registry.
+ let active_types: Vec<&str> = types
+ .map(|ts| {
+ ts.iter()
+ .map(|t| t.trim())
+ .filter(|t| !t.is_empty())
+ .collect()
+ })
+ .unwrap_or_default();
+ let (mut out, next_after) = {
+ let all = self.events.lock().unwrap();
+ match all.get(job_id) {
// No events yet for this job — registry returns empty
// (NOT NotFound — `not found` is the job-row state).
- return Ok(JobEventListResponse {
- events: vec![],
- next_after: after,
- });
+ None => (Vec::::new(), after),
+ Some(list) => {
+ let out: Vec = list
+ .iter()
+ .filter(|e| e.seq > after)
+ .filter(|e| {
+ active_types.is_empty()
+ || active_types
+ .iter()
+ .any(|t| *t == e.event_type.as_str())
+ })
+ .take(limit)
+ .cloned()
+ .collect();
+ // Default `next_after` to the caller's watermark when
+ // nothing matched; otherwise advance to the last seq.
+ let next_after = out.last().map(|e| e.seq).unwrap_or(after);
+ (out, next_after)
+ }
}
};
- let mut out: Vec = list
- .iter()
- .filter(|e| e.seq > after)
- .filter(|e| match types {
- Some(ts) if !ts.is_empty() => ts.iter().any(|t| t == &e.event_type),
- _ => true,
- })
- .take(limit)
- .cloned()
- .collect();
- // Default `next_after` to the caller's watermark when nothing
- // matched; otherwise advance to the last returned seq.
- let next_after = out.last().map(|e| e.seq).unwrap_or(after);
- // Mock semantics: no actual long-poll. Tests that need wait
- // behaviour drive the clock manually.
- // (We don't drain — list_job_events is read-only.)
out.shrink_to_fit();
+ // Test gate: on an EMPTY page, signal that this long-poll is
+ // in-flight (the caller's `recv_event` is holding its per-filter
+ // lock) and park until the test releases. Level-triggered, so once
+ // released later empty pages fall straight through. The events
+ // guard is dropped above before this await (no std mutex across
+ // await). Mock semantics otherwise: no real long-poll — tests that
+ // need wait behaviour drive the clock manually.
+ if out.is_empty() {
+ let gate = self.list_gate.lock().unwrap().clone();
+ if let Some(g) = gate {
+ if !g.release.is_cancelled() {
+ g.entered.notify_one();
+ g.release.cancelled().await;
+ }
+ }
+ }
Ok(JobEventListResponse {
events: out,
next_after,
@@ -3139,6 +3327,44 @@ mod tests {
assert_eq!(ev.seq, 2);
}
+ #[tokio::test]
+ async fn recv_event_empty_type_string_is_unfiltered() {
+ // `types=[""]` must canonicalize to the unfiltered stream — the
+ // registry trims + drops empty type strings and serves unfiltered, so
+ // the client mirrors that: it shares the "" cursor key with `None` and
+ // delivers events of ANY type (not a no-match timeout).
+ let (backend, ctrl) = make_controller("j-recv-empty-type").await;
+ backend.push_event("j-recv-empty-type", "A", serde_json::json!({"n": 1}));
+ backend.push_event("j-recv-empty-type", "B", serde_json::json!({"n": 2}));
+
+ // filter_key parity: the empty-string filter collapses to the
+ // unfiltered key, so both share one cursor.
+ assert_eq!(
+ JobController::filter_key(&Some(vec!["".into()])),
+ JobController::filter_key(&None),
+ "types=[\"\"] must canonicalize to the unfiltered key"
+ );
+
+ // recv_event with types=[""] returns the FIRST event regardless of
+ // type (unfiltered), not a no-match timeout.
+ let first = ctrl
+ .recv_event(Some(vec!["".into()]), Some(Duration::from_secs(2)))
+ .await
+ .unwrap()
+ .expect("types=[\"\"] must behave as unfiltered and deliver seq=1");
+ assert_eq!(first.seq, 1);
+ assert_eq!(first.event_type, "A");
+
+ // The cursor advanced on the shared "" key, so a follow-up unfiltered
+ // (None) call resumes strictly AFTER it — proving they share a cursor.
+ let second = ctrl
+ .recv_event(None, Some(Duration::from_secs(2)))
+ .await
+ .unwrap()
+ .expect("shared unfiltered cursor must resume at seq=2");
+ assert_eq!(second.seq, 2);
+ }
+
#[tokio::test]
async fn recv_event_increments_cursor_between_calls() {
// Two successive calls return seq=1 then seq=2 — the cursor must
@@ -3450,6 +3676,456 @@ mod tests {
let _ = a_handle.await.unwrap();
}
+ // ---- Phase 3: per-filter event cursors (issue #1252) --------------------
+
+ #[tokio::test]
+ async fn recv_event_per_filter_cursors_do_not_skip_earlier_events() {
+ // THE Phase-3 defect pin: with a single shared cursor, consuming a
+ // type-A match at seq 5 sets the cursor to 5, so a later
+ // recv_event(type=B) reads after=5 and PERMANENTLY skips the type-B
+ // event at seq 3. With per-filter cursors, the B stream has its own
+ // cursor (still 0) and delivers seq 3.
+ let (backend, ctrl) = make_controller("j-perfilter").await;
+ backend.push_event("j-perfilter", "C", serde_json::json!({})); // seq 1
+ backend.push_event("j-perfilter", "C", serde_json::json!({})); // seq 2
+ backend.push_event("j-perfilter", "B", serde_json::json!({})); // seq 3
+ backend.push_event("j-perfilter", "C", serde_json::json!({})); // seq 4
+ backend.push_event("j-perfilter", "A", serde_json::json!({})); // seq 5
+
+ // Consume type-A at seq 5 (advances ONLY the A-filter cursor).
+ let a = ctrl
+ .recv_event(Some(vec!["A".into()]), Some(Duration::from_secs(2)))
+ .await
+ .unwrap()
+ .expect("type-A event");
+ assert_eq!(a.seq, 5);
+ assert_eq!(a.event_type, "A");
+
+ // type-B must still be delivered at seq 3 — NOT skipped.
+ let b = ctrl
+ .recv_event(Some(vec!["B".into()]), Some(Duration::from_secs(2)))
+ .await
+ .unwrap()
+ .expect("type-B event must NOT be skipped by the A cursor");
+ assert_eq!(b.seq, 3, "per-filter cursor must deliver the earlier B");
+ assert_eq!(b.event_type, "B");
+ }
+
+ #[tokio::test]
+ async fn recv_event_unfiltered_and_filtered_are_independent_streams() {
+ // The unfiltered stream (None) and a type filter each keep their own
+ // cursor. An event matching both streams is observed once PER stream
+ // (documented at-least-once across streams).
+ let (backend, ctrl) = make_controller("j-streams").await;
+ backend.push_event("j-streams", "A", serde_json::json!({})); // seq 1
+ backend.push_event("j-streams", "B", serde_json::json!({})); // seq 2
+ backend.push_event("j-streams", "A", serde_json::json!({})); // seq 3
+
+ // Unfiltered head = seq 1.
+ let u1 = ctrl.recv_event(None, Some(Duration::from_secs(2))).await.unwrap().unwrap();
+ assert_eq!(u1.seq, 1);
+ // A-filter is independent: re-observes seq 1 (the first A).
+ let a1 = ctrl
+ .recv_event(Some(vec!["A".into()]), Some(Duration::from_secs(2)))
+ .await
+ .unwrap()
+ .unwrap();
+ assert_eq!(a1.seq, 1, "A-stream cursor is independent of the unfiltered one");
+ // Unfiltered advances to seq 2.
+ let u2 = ctrl.recv_event(None, Some(Duration::from_secs(2))).await.unwrap().unwrap();
+ assert_eq!(u2.seq, 2);
+ // A-filter advances to seq 3 (next A after its own cursor=1).
+ let a2 = ctrl
+ .recv_event(Some(vec!["A".into()]), Some(Duration::from_secs(2)))
+ .await
+ .unwrap()
+ .unwrap();
+ assert_eq!(a2.seq, 3);
+ }
+
+ #[tokio::test]
+ async fn recv_event_filter_key_is_canonical() {
+ // Same filter expressed with different order / duplicates hits the
+ // SAME cursor; and None == Some([]) (both unfiltered).
+ let (backend, ctrl) = make_controller("j-canon").await;
+ backend.push_event("j-canon", "A", serde_json::json!({})); // seq 1
+ backend.push_event("j-canon", "B", serde_json::json!({})); // seq 2
+ backend.push_event("j-canon", "A", serde_json::json!({})); // seq 3
+
+ // ["A","B"] then ["B","A"] must share one cursor: second call does NOT
+ // re-deliver seq 1 — it advances to seq 2.
+ let first = ctrl
+ .recv_event(Some(vec!["A".into(), "B".into()]), Some(Duration::from_secs(2)))
+ .await
+ .unwrap()
+ .unwrap();
+ assert_eq!(first.seq, 1);
+ let second = ctrl
+ .recv_event(Some(vec!["B".into(), "A".into()]), Some(Duration::from_secs(2)))
+ .await
+ .unwrap()
+ .unwrap();
+ assert_eq!(second.seq, 2, "reordered filter must share the cursor, not replay");
+ // Duplicates canonicalize identically → cursor now past seq 2 → next A.
+ let third = ctrl
+ .recv_event(
+ Some(vec!["A".into(), "A".into(), "B".into()]),
+ Some(Duration::from_secs(2)),
+ )
+ .await
+ .unwrap()
+ .unwrap();
+ assert_eq!(third.seq, 3);
+
+ // None and Some([]) are the SAME unfiltered stream.
+ let (backend2, ctrl2) = make_controller("j-canon2").await;
+ backend2.push_event("j-canon2", "X", serde_json::json!({})); // seq 1
+ backend2.push_event("j-canon2", "Y", serde_json::json!({})); // seq 2
+ let n1 = ctrl2.recv_event(None, Some(Duration::from_secs(2))).await.unwrap().unwrap();
+ assert_eq!(n1.seq, 1);
+ let e2 = ctrl2
+ .recv_event(Some(vec![]), Some(Duration::from_secs(2)))
+ .await
+ .unwrap()
+ .unwrap();
+ assert_eq!(e2.seq, 2, "Some([]) shares the unfiltered cursor with None");
+ }
+
+ /// Grab a clone of the per-filter serialization lock for `types` off a
+ /// controller, mirroring `recv_event`'s own lookup. Holding the returned
+ /// mutex simulates a same-filter long-poll in flight (that filter's slot
+ /// is occupied) WITHOUT racing a spawned task to the lock, so the tests
+ /// below have a deterministic "call 1 is provably pending" precondition.
+ fn filter_lock_of(ctrl: &JobController, types: &Option>) -> Arc> {
+ let key = JobController::filter_key(types);
+ let mut locks = ctrl.recv_locks.lock().unwrap();
+ locks
+ .entry(key)
+ .or_insert_with(|| Arc::new(Mutex::new(())))
+ .clone()
+ }
+
+ #[tokio::test]
+ async fn recv_event_different_filters_do_not_block_each_other() {
+ // Per-filter locks (not one global lock): a filter-A long-poll must
+ // NOT block a filter-B call that has an event ready. We PROVABLY
+ // occupy filter A's slot by holding its lock for the whole test (the
+ // deterministic stand-in for an in-flight A long-poll), then assert a
+ // filter-B call still completes immediately.
+ let (backend, ctrl) = make_controller("j-nofilterblock").await;
+ backend.push_event("j-nofilterblock", "B", serde_json::json!({})); // seq 1
+
+ let filter_a = Some(vec!["A".into()]);
+ let a_lock = filter_lock_of(&ctrl, &filter_a);
+ let _a_held = a_lock.lock().await; // filter A provably occupied
+
+ let start = Instant::now();
+ let b = ctrl
+ .recv_event(Some(vec!["B".into()]), Some(Duration::from_secs(2)))
+ .await
+ .unwrap()
+ .expect("B should be observed immediately");
+ let elapsed = start.elapsed();
+ assert_eq!(b.seq, 1);
+ assert!(
+ elapsed < Duration::from_millis(400),
+ "filter B must not serialise behind filter A's held lock (elapsed={:?})",
+ elapsed,
+ );
+ }
+
+ #[tokio::test]
+ async fn recv_event_same_filter_serializes_under_proven_overlap() {
+ // Two same-filter recvs must observe DISTINCT, ordered seqs — never a
+ // duplicate. A mock gate makes call 1 PROVABLY parked in its long-poll
+ // (holding the per-filter lock) before call 2 starts, so the calls
+ // genuinely overlap: without the lock both would read cursor 0 and
+ // return seq 1.
+ let (backend, ctrl) = make_controller("j-concfilter").await;
+ let gate = backend.install_list_gate();
+
+ let ctrl1 = ctrl.clone();
+ let call1 = tokio::spawn(async move {
+ ctrl1
+ .recv_event(Some(vec!["A".into()]), Some(Duration::from_secs(5)))
+ .await
+ .unwrap()
+ .unwrap()
+ });
+ // Call 1 is now provably parked on an empty page, holding filter A's
+ // lock. Only NOW publish the events and start the queued call 2.
+ gate.entered.notified().await;
+ backend.push_event("j-concfilter", "A", serde_json::json!({})); // seq 1
+ backend.push_event("j-concfilter", "A", serde_json::json!({})); // seq 2
+
+ let ctrl2 = ctrl.clone();
+ let call2 = tokio::spawn(async move {
+ ctrl2
+ .recv_event(Some(vec!["A".into()]), Some(Duration::from_secs(5)))
+ .await
+ .unwrap()
+ .unwrap()
+ });
+
+ // Release call 1's gated poll: it advances the cursor to seq 1, then
+ // call 2 serializes behind it and must see seq 2 (never a duplicate 1).
+ gate.release.cancel();
+
+ let ev1 = call1.await.unwrap();
+ let ev2 = call2.await.unwrap();
+ assert_eq!(ev1.seq, 1, "the in-flight caller consumes seq 1");
+ assert_eq!(
+ ev2.seq, 2,
+ "the queued caller must serialize behind it and see seq 2, not a duplicate"
+ );
+ }
+
+ #[tokio::test]
+ async fn recv_event_queued_same_filter_times_out_at_own_budget() {
+ // COMMENT-1 fix: a same-filter recv queued behind an in-flight
+ // long-poll must time out at ~ITS OWN budget, not lock-wait + budget.
+ // We hold filter A's lock for the whole test (an in-flight A long-poll
+ // that never releases); a second A recv with a 300ms budget must
+ // return Ok(None) at ~300ms rather than blocking forever on the lock.
+ let (_backend, ctrl) = make_controller("j-lockbudget").await;
+ let filter_a = Some(vec!["A".into()]);
+ let a_lock = filter_lock_of(&ctrl, &filter_a);
+ let _a_held = a_lock.lock().await;
+
+ let start = Instant::now();
+ let r = ctrl
+ .recv_event(filter_a.clone(), Some(Duration::from_millis(300)))
+ .await;
+ let elapsed = start.elapsed();
+ assert!(
+ matches!(r, Ok(None)),
+ "a queued caller must time out as Ok(None), got {:?}",
+ r
+ );
+ assert!(
+ elapsed >= Duration::from_millis(250) && elapsed < Duration::from_millis(900),
+ "must time out at ~its own 300ms budget, not lock-wait + budget (elapsed={:?})",
+ elapsed
+ );
+ }
+
+ #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+ async fn recv_event_queued_on_lock_returns_cancelled_promptly() {
+ // COMMENT-1 fix: an execution cancelled WHILE queued on the per-filter
+ // lock must return Cancelled promptly — not sit parked until the
+ // holder's (up to 60s) long-poll releases. We hold filter A's lock for
+ // the whole test; a second A recv queues behind it, then the job's
+ // cancel token fires and the queued recv must surface Cancelled.
+ let (_backend, ctrl) =
+ make_controller_with_epoch("j-lockcancel", Some(1), 1).await;
+ let token = CancellationToken::new();
+ let generation = cancel_registry::register_active_job_with_epoch(
+ "j-lockcancel",
+ token.clone(),
+ Some(1),
+ );
+
+ let filter_a = Some(vec!["A".into()]);
+ let a_lock = filter_lock_of(&ctrl, &filter_a);
+ let _a_held = a_lock.lock().await;
+
+ let ctrl_clone = ctrl.clone();
+ let recv = tokio::spawn(async move {
+ ctrl_clone
+ .recv_event(Some(vec!["A".into()]), Some(Duration::from_secs(10)))
+ .await
+ });
+ // Let the recv park on the held lock, then fire the cancel.
+ sleep(Duration::from_millis(80)).await;
+ token.cancel();
+
+ let started = Instant::now();
+ let result = tokio::time::timeout(Duration::from_secs(3), recv)
+ .await
+ .expect("recv queued on the lock must return promptly on cancel")
+ .unwrap();
+ assert!(
+ matches!(result, Err(JobError::Cancelled)),
+ "a cancel while queued on the lock must surface Cancelled, got {:?}",
+ result
+ );
+ assert!(
+ started.elapsed() < Duration::from_secs(2),
+ "must return promptly, took {:?}",
+ started.elapsed()
+ );
+ cancel_registry::unregister_active_job("j-lockcancel", generation);
+ }
+
+ // ---- Mock-vs-registry contract test (refuted-D3 divergence guard) -------
+ //
+ // A shared scenario table for `list_job_events` executed against BOTH the
+ // in-process `MockBackend` AND a mockito-simulated `RegistryHttpBackend`,
+ // asserting identical `(returned seqs, next_after)`. Two distinct
+ // properties are pinned:
+ //
+ // 1. The `MockBackend` reproduces the registry's page contract — empty
+ // filtered page returns `next_after == after` (no leap past
+ // unreturned events), filtered delivery is ascending and
+ // type-restricted — so the mock the rest of the unit tests run
+ // against can't silently drift from the real registry's behaviour.
+ //
+ // 2. `RegistryHttpBackend` sends the CORRECT request for each scenario:
+ // the mockito expectation matches the request path AND the `after`
+ // and `types` query params (not `Matcher::Any`), so a regression that
+ // sent the wrong `after` or dropped the `types` filter would miss the
+ // mock and fail the test rather than pass on a canned body. It also
+ // pins that the backend parses the reply into the same
+ // `(seqs, next_after)` the mock produces.
+ //
+ // What this test does NOT do is pin the registry's OWN adherence to the
+ // page contract (the mock body here is hand-built to match it) — that is
+ // an HTTP-server behaviour, pinned Go-side in
+ // `src/core/registry/ent_handlers_job_events_test.go`. This test pins the
+ // Rust CLIENT's request/parse contract against a faithful mock.
+
+ /// (log events as (seq,type), after, types, expected returned seqs,
+ /// expected next_after).
+ fn list_events_contract_scenarios() -> Vec<(
+ Vec<(i64, &'static str)>,
+ i64,
+ Option>,
+ Vec,
+ i64,
+ )> {
+ let log = vec![(1_i64, "A"), (2, "B"), (3, "A")];
+ vec![
+ // Unfiltered from 0: all three, next_after = last returned.
+ (log.clone(), 0, None, vec![1, 2, 3], 3),
+ // Type A from 0: 1 and 3 only, next_after = 3.
+ (log.clone(), 0, Some(vec!["A".into()]), vec![1, 3], 3),
+ // Type B after 1: just 2, next_after = 2.
+ (log.clone(), 1, Some(vec!["B".into()]), vec![2], 2),
+ // Type A after 3 (tip): EMPTY page — next_after == after (3).
+ (log.clone(), 3, Some(vec!["A".into()]), vec![], 3),
+ // No-match type from 0: EMPTY page — next_after == after (0).
+ (log.clone(), 0, Some(vec!["Z".into()]), vec![], 0),
+ // Empty-string type from 0: the registry trims + drops empty type
+ // strings and serves UNFILTERED, so this must match `None` above —
+ // all three, next_after = 3. Pins the `types=[""]` ⇒ unfiltered
+ // equivalence on BOTH backends.
+ (log.clone(), 0, Some(vec!["".into()]), vec![1, 2, 3], 3),
+ ]
+ }
+
+ async fn run_scenario_on_backend(
+ backend: &dyn TaskBackend,
+ after: i64,
+ types: &Option>,
+ ) -> (Vec, i64) {
+ let resp = backend
+ .list_job_events(
+ "j",
+ after,
+ types.as_deref(),
+ Duration::from_secs(0),
+ 100,
+ None,
+ )
+ .await
+ .unwrap();
+ (resp.events.iter().map(|e| e.seq).collect(), resp.next_after)
+ }
+
+ #[tokio::test]
+ async fn list_events_mock_matches_registry_contract() {
+ for (log, after, types, expected_seqs, expected_next_after) in
+ list_events_contract_scenarios()
+ {
+ // --- in-process MockBackend ---
+ let mock = MockBackend::new();
+ for (_seq, ty) in &log {
+ mock.push_event("j", ty, serde_json::json!({}));
+ }
+ let (mock_seqs, mock_next) =
+ run_scenario_on_backend(mock.as_ref(), after, &types).await;
+
+ // --- mockito-simulated registry: canned response encoding the
+ // documented contract (built from the log via the SAME
+ // filter/next_after rules the registry implements). ---
+ let mut server = mockito::Server::new_async().await;
+ let events_json: Vec = expected_seqs
+ .iter()
+ .map(|&seq| {
+ let ty = log.iter().find(|(s, _)| *s == seq).unwrap().1;
+ serde_json::json!({
+ "job_id": "j",
+ "seq": seq,
+ "type": ty,
+ "payload": null,
+ "trace_context": null,
+ "posted_by": null,
+ "created_at": 1_700_000_000_i64 + seq,
+ })
+ })
+ .collect();
+ let body = serde_json::json!({
+ "events": events_json,
+ "next_after": expected_next_after,
+ })
+ .to_string();
+ // Match the exact request the client MUST send for this scenario:
+ // the events path plus the `after` watermark, and — for filtered
+ // scenarios — the comma-joined `types` param. If a regression sent
+ // the wrong `after` or dropped `types`, the request won't match
+ // this mock and mockito serves a 501, failing the `.unwrap()`
+ // below instead of silently passing on the canned body.
+ let mut query_matchers = vec![mockito::Matcher::UrlEncoded(
+ "after".into(),
+ after.to_string(),
+ )];
+ if let Some(ts) = &types {
+ let joined = ts
+ .iter()
+ .map(|t| t.trim())
+ .filter(|t| !t.is_empty())
+ .collect::>()
+ .join(",");
+ if !joined.is_empty() {
+ query_matchers.push(mockito::Matcher::UrlEncoded(
+ "types".into(),
+ joined,
+ ));
+ }
+ }
+ let _m = server
+ .mock("GET", "/jobs/j/events")
+ .match_query(mockito::Matcher::AllOf(query_matchers))
+ .with_status(200)
+ .with_header("content-type", "application/json")
+ .with_body(body)
+ .create_async()
+ .await;
+ let registry = RegistryHttpBackend::new(&server.url()).unwrap();
+ let (reg_seqs, reg_next) =
+ run_scenario_on_backend(®istry, after, &types).await;
+
+ // Both backends must agree with each other AND the contract.
+ assert_eq!(
+ mock_seqs, expected_seqs,
+ "mock diverged from contract (after={after}, types={types:?})"
+ );
+ assert_eq!(
+ mock_next, expected_next_after,
+ "mock next_after diverged (after={after}, types={types:?})"
+ );
+ assert_eq!(
+ reg_seqs, expected_seqs,
+ "registry response diverged (after={after}, types={types:?})"
+ );
+ assert_eq!(
+ reg_next, expected_next_after,
+ "registry next_after diverged (after={after}, types={types:?})"
+ );
+ assert_eq!((mock_seqs, mock_next), (reg_seqs, reg_next));
+ }
+ }
+
// ---- JobProxy::list_events ---------------------------------------------
//
// `list_events` is the low-level batch primitive that the language
diff --git a/tests/integration/global/routines.yaml b/tests/integration/global/routines.yaml
index e22bf6c66..a79bbf8e1 100644
--- a/tests/integration/global/routines.yaml
+++ b/tests/integration/global/routines.yaml
@@ -272,6 +272,54 @@ routines:
# UTILITIES
#############################################################################
+ # LIVENESS-ONLY registration wait (settle-grace design, issue #1193):
+ # polls `meshctl list` until every named agent APPEARS. Deliberately does
+ # NOT gate on dependency resolution ("1/1") or health status — dependency
+ # resolution is covered at call time by the runtime's settle window (the
+ # consumer's first mesh call blocks until deps settle). Do NOT add
+ # dep-aware conditions here.
+ #
+ # Matching is a case-insensitive substring grep per name against the
+ # meshctl list output (case-insensitive so display-cased ids like the
+ # api-gateway's "API ..." row match too).
+ wait_for_agents_registered:
+ description: "Poll meshctl list until all named agents appear (liveness-only, 240s deadline)"
+ params:
+ agents:
+ type: string
+ required: true
+ description: "Space-separated agent names to await in meshctl list"
+ steps:
+ - handler: shell
+ workdir: /workspace
+ command: |
+ EXPECTED="${params.agents}"
+ TOTAL=$(echo "$EXPECTED" | wc -w | tr -d ' ')
+ echo "Waiting for $TOTAL agent(s) to register: $EXPECTED"
+ for i in $(seq 1 120); do
+ LIST=$(meshctl list 2>/dev/null || true)
+ COUNT=0
+ for agent in $EXPECTED; do
+ if echo "$LIST" | grep -qi "$agent"; then
+ COUNT=$((COUNT + 1))
+ fi
+ done
+ if [ "$COUNT" -ge "$TOTAL" ]; then
+ echo "All $TOTAL agent(s) registered after ~$((i*2))s"
+ exit 0
+ fi
+ sleep 2
+ done
+ echo "ERROR: only $COUNT/$TOTAL agent(s) registered within 240s (expected: $EXPECTED)"
+ meshctl list 2>/dev/null || true
+ for agent in $EXPECTED; do
+ echo "=== logs: $agent ==="
+ meshctl logs "$agent" 2>/dev/null | tail -30 || true
+ done
+ exit 1
+ capture: wait_agents_registered
+ timeout: 260
+
# Cleanup workspace
cleanup_workspace:
description: "Remove all files from workspace directory"
diff --git a/tests/integration/suites/uc06_observability/tc03_java_llm_tracing/test.yaml b/tests/integration/suites/uc06_observability/tc03_java_llm_tracing/test.yaml
index 0d39e8ad3..d5e399aa2 100644
--- a/tests/integration/suites/uc06_observability/tc03_java_llm_tracing/test.yaml
+++ b/tests/integration/suites/uc06_observability/tc03_java_llm_tracing/test.yaml
@@ -18,7 +18,10 @@ tags:
- java
- llm
- mesh-delegation
-timeout: 360
+# Total budget must cover the two 260s registration-wait steps STACKED on
+# top of the Java build/start and LLM-call steps — a slow-but-successful
+# registration must not exhaust the test budget after every step succeeded.
+timeout: 1200
pre_run:
# Start observability infrastructure (Redis + Tempo)
@@ -65,25 +68,10 @@ test:
workdir: /workspace
capture: start_provider
- # Wait for provider to register
- - name: "Wait for provider registration"
- handler: shell
- workdir: /workspace
- command: |
- echo "Waiting for claude-provider to start and register..."
- for i in $(seq 1 30); do
- if meshctl list 2>/dev/null | grep -q "claude-provider"; then
- echo "claude-provider registered after ${i}s"
- exit 0
- fi
- sleep 2
- done
- echo "ERROR: claude-provider did not register within 60s"
- meshctl list 2>/dev/null || true
- meshctl logs claude-provider 2>/dev/null | tail -30 || true
- exit 1
- capture: wait_provider
- timeout: 90
+ # Wait for provider to register (liveness-only shared routine)
+ - routine: global.wait_for_agents_registered
+ params:
+ agents: claude-provider
# Start analyst agent with tracing enabled (depends on provider)
- name: "Start analyst agent"
@@ -92,33 +80,13 @@ test:
workdir: /workspace
capture: start_analyst
- # Wait for analyst to register and resolve LLM provider dependency
- - name: "Wait for analyst registration"
- handler: shell
- workdir: /workspace
- command: |
- echo "Waiting for analyst agent to start and resolve LLM dependency..."
- for i in $(seq 1 45); do
- # Check if analyst is registered and has LLM provider resolved (1/1 deps)
- if meshctl list 2>/dev/null | grep -q "analyst.*1/1"; then
- echo "analyst registered and LLM provider resolved after ${i}s"
- exit 0
- fi
- # Also check if it shows as healthy without deps indicator
- if meshctl list 2>/dev/null | grep -q "analyst.*healthy"; then
- echo "analyst registered as healthy after ${i}s"
- exit 0
- fi
- sleep 2
- done
- echo "ERROR: analyst agent did not register properly within 90s"
- echo "=== Agent list ==="
- meshctl list 2>/dev/null || true
- echo "=== analyst logs ==="
- meshctl logs analyst 2>/dev/null | tail -30 || true
- exit 1
- capture: wait_analyst
- timeout: 120
+ # Wait for the analyst to register — LIVENESS-ONLY (settle-grace design,
+ # issue #1193): the old loop's "1/1" dep condition was against convention;
+ # dependency resolution is covered at call time by the runtime's settle
+ # window (the chat step's first call blocks until deps settle).
+ - routine: global.wait_for_agents_registered
+ params:
+ agents: analyst
# Verify all agents are running
- name: "List agents"
diff --git a/tests/integration/suites/uc10_toolcalls/tc19_consumer_java_provider_claude_py_tool_py/test.yaml b/tests/integration/suites/uc10_toolcalls/tc19_consumer_java_provider_claude_py_tool_py/test.yaml
index cbcc41455..1d69550c3 100644
--- a/tests/integration/suites/uc10_toolcalls/tc19_consumer_java_provider_claude_py_tool_py/test.yaml
+++ b/tests/integration/suites/uc10_toolcalls/tc19_consumer_java_provider_claude_py_tool_py/test.yaml
@@ -15,7 +15,7 @@ tags:
- claude
- python
- phase1
-timeout: 300
+timeout: 600
pre_run:
- routine: global.setup_for_java_agent
@@ -91,9 +91,14 @@ test:
echo "Consumer agent starting..."
capture: start_consumer
- - name: "Wait for consumer and resolution"
- handler: wait
- seconds: 25
+ # Liveness only: poll until the Java consumer REGISTERS instead of a fixed
+ # sleep — cold-JVM starts (mvn compile + Spring Boot startup) routinely
+ # outran the old 25s wait. Dependency resolution stays covered by the
+ # runtime's settle window (no dep-resolution wait) — the shared routine is
+ # liveness-only by design (issue #1193).
+ - routine: global.wait_for_agents_registered
+ params:
+ agents: analyst-java
# 6. Verify all agents registered
- name: "Verify agents registered"
diff --git a/tests/integration/suites/uc10_toolcalls/tc20_consumer_java_provider_claude_py_tool_ts/test.yaml b/tests/integration/suites/uc10_toolcalls/tc20_consumer_java_provider_claude_py_tool_ts/test.yaml
index c813e65bd..3b1391d18 100644
--- a/tests/integration/suites/uc10_toolcalls/tc20_consumer_java_provider_claude_py_tool_ts/test.yaml
+++ b/tests/integration/suites/uc10_toolcalls/tc20_consumer_java_provider_claude_py_tool_ts/test.yaml
@@ -16,7 +16,7 @@ tags:
- typescript
- phase1
- python
-timeout: 300
+timeout: 600
pre_run:
- routine: global.setup_for_java_agent
@@ -93,9 +93,14 @@ test:
echo "Consumer agent starting..."
capture: start_consumer
- - name: "Wait for consumer and resolution"
- handler: wait
- seconds: 25
+ # Liveness only: poll until the Java consumer REGISTERS instead of a fixed
+ # sleep — cold-JVM starts (mvn compile + Spring Boot startup) routinely
+ # outran the old 25s wait. Dependency resolution stays covered by the
+ # runtime's settle window (no dep-resolution wait) — the shared routine is
+ # liveness-only by design (issue #1193).
+ - routine: global.wait_for_agents_registered
+ params:
+ agents: analyst-java
# 6. Verify all agents registered
- name: "Verify agents registered"
diff --git a/tests/integration/suites/uc10_toolcalls/tc22_consumer_java_provider_openai_py_tool_py/test.yaml b/tests/integration/suites/uc10_toolcalls/tc22_consumer_java_provider_openai_py_tool_py/test.yaml
index a290c214d..fd1fae8c0 100644
--- a/tests/integration/suites/uc10_toolcalls/tc22_consumer_java_provider_openai_py_tool_py/test.yaml
+++ b/tests/integration/suites/uc10_toolcalls/tc22_consumer_java_provider_openai_py_tool_py/test.yaml
@@ -15,7 +15,7 @@ tags:
- openai
- python
- phase1
-timeout: 300
+timeout: 600
pre_run:
- routine: global.setup_for_java_agent
@@ -91,9 +91,14 @@ test:
echo "Consumer agent starting..."
capture: start_consumer
- - name: "Wait for consumer and resolution"
- handler: wait
- seconds: 25
+ # Liveness only: poll until the Java consumer REGISTERS instead of a fixed
+ # sleep — cold-JVM starts (mvn compile + Spring Boot startup) routinely
+ # outran the old 25s wait. Dependency resolution stays covered by the
+ # runtime's settle window (no dep-resolution wait) — the shared routine is
+ # liveness-only by design (issue #1193).
+ - routine: global.wait_for_agents_registered
+ params:
+ agents: analyst-java
# 6. Verify all agents registered
- name: "Verify agents registered"
diff --git a/tests/integration/suites/uc10_toolcalls/tc23_consumer_java_provider_openai_py_tool_ts/test.yaml b/tests/integration/suites/uc10_toolcalls/tc23_consumer_java_provider_openai_py_tool_ts/test.yaml
index 7a66f11c4..2f399eba9 100644
--- a/tests/integration/suites/uc10_toolcalls/tc23_consumer_java_provider_openai_py_tool_ts/test.yaml
+++ b/tests/integration/suites/uc10_toolcalls/tc23_consumer_java_provider_openai_py_tool_ts/test.yaml
@@ -16,7 +16,7 @@ tags:
- typescript
- phase1
- python
-timeout: 300
+timeout: 600
pre_run:
- routine: global.setup_for_java_agent
@@ -93,9 +93,14 @@ test:
echo "Consumer agent starting..."
capture: start_consumer
- - name: "Wait for consumer and resolution"
- handler: wait
- seconds: 25
+ # Liveness only: poll until the Java consumer REGISTERS instead of a fixed
+ # sleep — cold-JVM starts (mvn compile + Spring Boot startup) routinely
+ # outran the old 25s wait. Dependency resolution stays covered by the
+ # runtime's settle window (no dep-resolution wait) — the shared routine is
+ # liveness-only by design (issue #1193).
+ - routine: global.wait_for_agents_registered
+ params:
+ agents: analyst-java
# 6. Verify all agents registered
- name: "Verify agents registered"
diff --git a/tests/integration/suites/uc10_toolcalls/tc25_consumer_java_provider_gemini_py_tool_py/test.yaml b/tests/integration/suites/uc10_toolcalls/tc25_consumer_java_provider_gemini_py_tool_py/test.yaml
index 996131aaa..2185fae33 100644
--- a/tests/integration/suites/uc10_toolcalls/tc25_consumer_java_provider_gemini_py_tool_py/test.yaml
+++ b/tests/integration/suites/uc10_toolcalls/tc25_consumer_java_provider_gemini_py_tool_py/test.yaml
@@ -15,7 +15,7 @@ tags:
- gemini
- python
- phase1
-timeout: 300
+timeout: 600
pre_run:
- routine: global.setup_for_java_agent
params:
@@ -81,9 +81,14 @@ test:
meshctl start analyst-java --env MCP_MESH_HTTP_PORT=9003 -d
echo "Consumer agent starting..."
capture: start_consumer
- - name: "Wait for consumer and resolution"
- handler: wait
- seconds: 25
+ # Liveness only: poll until the Java consumer REGISTERS instead of a fixed
+ # sleep — cold-JVM starts (mvn compile + Spring Boot startup) routinely
+ # outran the old 25s wait. Dependency resolution stays covered by the
+ # runtime's settle window (no dep-resolution wait) — the shared routine is
+ # liveness-only by design (issue #1193).
+ - routine: global.wait_for_agents_registered
+ params:
+ agents: analyst-java
# 6. Verify all agents registered
- name: "Verify agents registered"
handler: shell
diff --git a/tests/integration/suites/uc10_toolcalls/tc26_consumer_java_provider_gemini_py_tool_ts/test.yaml b/tests/integration/suites/uc10_toolcalls/tc26_consumer_java_provider_gemini_py_tool_ts/test.yaml
index 92c64cfc7..be482d577 100644
--- a/tests/integration/suites/uc10_toolcalls/tc26_consumer_java_provider_gemini_py_tool_ts/test.yaml
+++ b/tests/integration/suites/uc10_toolcalls/tc26_consumer_java_provider_gemini_py_tool_ts/test.yaml
@@ -16,7 +16,7 @@ tags:
- typescript
- phase1
- python
-timeout: 300
+timeout: 600
pre_run:
- routine: global.setup_for_java_agent
params:
@@ -83,9 +83,14 @@ test:
meshctl start analyst-java --env MCP_MESH_HTTP_PORT=9003 -d
echo "Consumer agent starting..."
capture: start_consumer
- - name: "Wait for consumer and resolution"
- handler: wait
- seconds: 25
+ # Liveness only: poll until the Java consumer REGISTERS instead of a fixed
+ # sleep — cold-JVM starts (mvn compile + Spring Boot startup) routinely
+ # outran the old 25s wait. Dependency resolution stays covered by the
+ # runtime's settle window (no dep-resolution wait) — the shared routine is
+ # liveness-only by design (issue #1193).
+ - routine: global.wait_for_agents_registered
+ params:
+ agents: analyst-java
# 6. Verify all agents registered
- name: "Verify agents registered"
handler: shell
diff --git a/tests/integration/suites/uc10_toolcalls/tc46_consumer_java_provider_claude_ts_tool_py/test.yaml b/tests/integration/suites/uc10_toolcalls/tc46_consumer_java_provider_claude_ts_tool_py/test.yaml
index ab89d4073..e6ba83d87 100644
--- a/tests/integration/suites/uc10_toolcalls/tc46_consumer_java_provider_claude_ts_tool_py/test.yaml
+++ b/tests/integration/suites/uc10_toolcalls/tc46_consumer_java_provider_claude_ts_tool_py/test.yaml
@@ -16,7 +16,7 @@ tags:
- typescript
- phase2
- python
-timeout: 300
+timeout: 600
pre_run:
- routine: global.setup_for_java_agent
@@ -93,9 +93,14 @@ test:
echo "Consumer agent starting..."
capture: start_consumer
- - name: "Wait for consumer and resolution"
- handler: wait
- seconds: 25
+ # Liveness only: poll until the Java consumer REGISTERS instead of a fixed
+ # sleep — cold-JVM starts (mvn compile + Spring Boot startup) routinely
+ # outran the old 25s wait. Dependency resolution stays covered by the
+ # runtime's settle window (no dep-resolution wait) — the shared routine is
+ # liveness-only by design (issue #1193).
+ - routine: global.wait_for_agents_registered
+ params:
+ agents: analyst-java
# 6. Verify all agents registered
- name: "Verify agents registered"
diff --git a/tests/integration/suites/uc20_tutorial/tc07_day07_committee/test.yaml b/tests/integration/suites/uc20_tutorial/tc07_day07_committee/test.yaml
index 19c2ef8a2..98d86a336 100644
--- a/tests/integration/suites/uc20_tutorial/tc07_day07_committee/test.yaml
+++ b/tests/integration/suites/uc20_tutorial/tc07_day07_committee/test.yaml
@@ -18,7 +18,11 @@ tags:
- meshctl
- gateway
- fastapi
-timeout: 600
+# Total budget must cover the two 260s registration-wait steps STACKED on
+# top of 13 agent starts + the 360s plan_trip LLM call — a slow-but-
+# successful registration must not exhaust the test budget after every
+# step succeeded.
+timeout: 1200
pre_run:
- routine: global.start_observability
@@ -62,35 +66,11 @@ test:
workdir: /workspace
capture: start_agents_output
- # Wait for registration and DI resolution
- - name: "Wait for all agents to register"
- handler: shell
- workdir: /workspace
- command: |
- EXPECTED="flight-agent hotel-agent weather-agent poi-agent user-prefs-agent claude-provider openai-provider planner-agent chat-history-agent budget-analyst adventure-advisor logistics-planner"
- EXPECTED_COUNT=12
- echo "Waiting for $EXPECTED_COUNT agents to register..."
- for i in $(seq 1 30); do
- AGENTS=$(meshctl list 2>/dev/null || true)
- COUNT=0
- for agent in $EXPECTED; do
- if echo "$AGENTS" | grep -q "$agent"; then
- COUNT=$((COUNT + 1))
- fi
- done
- if [ "$COUNT" -ge "$EXPECTED_COUNT" ]; then
- echo "All $EXPECTED_COUNT agents registered after $((i * 2))s"
- meshctl list
- exit 0
- fi
- echo " $COUNT/$EXPECTED_COUNT agents registered (attempt $i/30)..."
- sleep 2
- done
- echo "ERROR: Only $COUNT/$EXPECTED_COUNT agents registered within 60s"
- meshctl list 2>/dev/null || true
- exit 1
- capture: wait_agents_output
- timeout: 90
+ # Wait for all 12 agents to register (liveness-only shared routine;
+ # DI resolution is covered at call time by the settle window)
+ - routine: global.wait_for_agents_registered
+ params:
+ agents: "flight-agent hotel-agent weather-agent poi-agent user-prefs-agent claude-provider openai-provider planner-agent chat-history-agent budget-analyst adventure-advisor logistics-planner"
# Start the gateway
- name: "Start gateway"
@@ -99,26 +79,12 @@ test:
workdir: /workspace
capture: start_gateway_output
- # Wait for gateway to register and resolve dependencies
- - name: "Wait for gateway to register"
- handler: shell
- workdir: /workspace
- command: |
- echo "Waiting for API gateway to register..."
- for i in $(seq 1 30); do
- if meshctl list 2>/dev/null | grep -qi "API"; then
- echo "API gateway registered after $((i * 2))s"
- meshctl list
- exit 0
- fi
- echo " API gateway not yet registered (attempt $i/30)..."
- sleep 2
- done
- echo "ERROR: API gateway not registered within 60s"
- meshctl list 2>/dev/null || true
- exit 1
- capture: wait_gateway_output
- timeout: 90
+ # Wait for the gateway to register (liveness-only shared routine; the
+ # gateway registers under an api-* service id — the routine's grep is
+ # case-insensitive, matching the original `grep -qi "API"`)
+ - routine: global.wait_for_agents_registered
+ params:
+ agents: api
# Verify all 13 agents are registered
- name: "Verify all agents are registered"
diff --git a/tests/integration/suites/uc33_meshjob_replicas/artifacts/gate-driver b/tests/integration/suites/uc33_meshjob_replicas/artifacts/gate-driver
new file mode 120000
index 000000000..4c490c090
--- /dev/null
+++ b/tests/integration/suites/uc33_meshjob_replicas/artifacts/gate-driver
@@ -0,0 +1 @@
+../fixtures/gate-driver
\ No newline at end of file
diff --git a/tests/integration/suites/uc33_meshjob_replicas/artifacts/gated-worker-a b/tests/integration/suites/uc33_meshjob_replicas/artifacts/gated-worker-a
new file mode 120000
index 000000000..87275a51f
--- /dev/null
+++ b/tests/integration/suites/uc33_meshjob_replicas/artifacts/gated-worker-a
@@ -0,0 +1 @@
+../fixtures/gated-worker-a
\ No newline at end of file
diff --git a/tests/integration/suites/uc33_meshjob_replicas/artifacts/gated-worker-b b/tests/integration/suites/uc33_meshjob_replicas/artifacts/gated-worker-b
new file mode 120000
index 000000000..c9eefd667
--- /dev/null
+++ b/tests/integration/suites/uc33_meshjob_replicas/artifacts/gated-worker-b
@@ -0,0 +1 @@
+../fixtures/gated-worker-b
\ No newline at end of file
diff --git a/tests/integration/suites/uc33_meshjob_replicas/fixtures/gate-driver/main.py b/tests/integration/suites/uc33_meshjob_replicas/fixtures/gate-driver/main.py
new file mode 100644
index 000000000..3d011cf2b
--- /dev/null
+++ b/tests/integration/suites/uc33_meshjob_replicas/fixtures/gate-driver/main.py
@@ -0,0 +1,84 @@
+#!/usr/bin/env python3
+"""MeshJob submitter for uc33 (issue #1252 Phase 4).
+
+Submit-only driver: each tool submits one of the gated-worker capabilities
+and returns ``{job_id}`` immediately. The test driver then sequences the
+scenario itself — posting ``go`` / ``finish`` events and reading job status
+and the event log directly via the registry HTTP API — because blocking on
+``proxy.wait()`` inside a tool would hide exactly the mid-flight signal
+(claims, reclaims, fenced writes) these tests exist to observe. Mirrors
+uc21's ``commission_submit_only`` shape.
+"""
+
+import os
+from typing import Any
+
+import mesh
+from fastmcp import FastMCP
+from mesh import MeshJob
+
+app = FastMCP("Gate Driver (uc33)")
+
+
+@app.tool()
+@mesh.tool(
+ capability="submit_gated",
+ dependencies=["gated_phases"],
+ description="Submit gated_phases (sequential recv_event gates) and return the job_id.",
+)
+async def submit_gated(
+ phases: int = 3,
+ max_duration: int = 15,
+ max_retries: int = 2,
+ gated_phases: MeshJob = None,
+) -> dict[str, Any]:
+ if gated_phases is None:
+ return {"error": "gated_phases submitter not injected"}
+ # Small max_duration sizes the LEASE window (leaseWindowFor derives the
+ # lease from it) so the quiet-gate silence in tc01 comfortably exceeds
+ # it. max_retries > 0 so a pre-fix lease lapse would RECLAIM (the field
+ # corruption shape) rather than terminally fail — either divergence
+ # from attempt_count == 1 fails the test.
+ proxy = await gated_phases.submit(
+ phases=phases,
+ max_duration=max_duration,
+ max_retries=max_retries,
+ )
+ return {"job_id": getattr(proxy, "job_id", None)}
+
+
+@app.tool()
+@mesh.tool(
+ capability="submit_sleepy",
+ dependencies=["sleepy_phases"],
+ description="Submit sleepy_phases (attempt 1 wedges past the lease) and return the job_id.",
+)
+async def submit_sleepy(
+ sleep_secs: int = 35,
+ max_duration: int = 6,
+ max_retries: int = 1,
+ sleepy_phases: MeshJob = None,
+) -> dict[str, Any]:
+ if sleepy_phases is None:
+ return {"error": "sleepy_phases submitter not injected"}
+ # max_duration=6 makes the lease lapse ~6s into attempt 1's wedged
+ # sleep; max_retries=1 budgets exactly the one reclaim tc02 needs (a
+ # second lapse would mark the job failed and the test would catch it).
+ proxy = await sleepy_phases.submit(
+ sleep_secs=sleep_secs,
+ max_duration=max_duration,
+ max_retries=max_retries,
+ )
+ return {"job_id": getattr(proxy, "job_id", None)}
+
+
+@mesh.agent(
+ name="gate-driver",
+ version="1.0.0",
+ description="Submit-only MeshJob driver (uc33) — multi-replica execution integrity fixture.",
+ http_port=int(os.environ.get("MCP_MESH_HTTP_PORT", "9110")),
+ enable_http=True,
+ auto_run=True,
+)
+class GateDriver:
+ pass
diff --git a/tests/integration/suites/uc33_meshjob_replicas/fixtures/gated-worker-a/main.py b/tests/integration/suites/uc33_meshjob_replicas/fixtures/gated-worker-a/main.py
new file mode 100644
index 000000000..f468efc7d
--- /dev/null
+++ b/tests/integration/suites/uc33_meshjob_replicas/fixtures/gated-worker-a/main.py
@@ -0,0 +1,231 @@
+#!/usr/bin/env python3
+"""Gated MeshJob worker — replica A (uc33, issue #1252 Phase 4).
+
+Replica pair: ``gated-worker-a`` / ``gated-worker-b`` are byte-identical
+except for the declared agent name, the ``REPLICA`` stamp and the default
+port. They are distinct fixture files because meshctl's "is this agent
+already running" check keys off the ``@mesh.agent(name=...)`` decorator
+literal (same pattern as uc21's bystander-x / bystander-y). The tests start
+BOTH with ``--env MCP_MESH_AGENT_NAME=gated-worker`` so both INSTANCE IDS
+share the ``gated-worker-`` prefix (the env var only seeds the instance-id
+prefix — the registered ``/agents`` ``.name`` stays this decorator literal).
+What mirrors production replicas (``replicas: 2``) is the part #1252 cares
+about: two claim workers providing the SAME task capability on ONE shared
+job queue — while meshctl manages them as two distinct local processes.
+
+Both capabilities are ``task=True`` and claimed via the shared job queue:
+whichever replica's claim worker wins executes the attempt. Every app-level
+"transition" event the handlers post is stamped with
+``{marker, phase/attempt, epoch: job.claim_epoch, replica: REPLICA}`` so the
+test can detect duplicated execution and attribute every side effect to a
+specific (replica, claim-epoch) pair.
+
+Capabilities:
+
+- ``gated_phases`` — the #1252 field shape: sequential ``recv_event`` gates.
+ Each gate polls in SHORT rounds (``GATE_ROUND_SECS`` ≤ 5s), so every round
+ is one executor read = one poll-liveness lease extension, keeping the
+ extension cadence well inside even a small lease window. A handler blocked
+ on a legitimately quiet gate is therefore provably alive — the registry
+ must NOT reclaim it (tc01).
+
+- ``sleepy_phases`` — the wedged-owner shape: attempt 1 blocks in a PURE
+ ``asyncio.sleep`` (no recv_event polls → no liveness credit, no progress
+ deltas → no lease renewal) past the lease window, so the registry
+ legitimately reclaims and the next claim (epoch 2) supersedes attempt 1.
+ When attempt 1 wakes, its first fenced write (a progress delta carrying
+ the stale epoch) must be rejected as ``claim_superseded`` and fire this
+ attempt's cancel token — aborting it BEFORE the duplicate side effects
+ further down the handler body can land (tc02).
+"""
+
+import asyncio
+import os
+from typing import Any
+
+import mesh
+from fastmcp import FastMCP
+from mesh import MeshJob
+
+REPLICA = "a"
+
+app = FastMCP(f"Gated Worker {REPLICA.upper()} (uc33)")
+
+# Short recv_event rounds: one executor read (= one poll-liveness lease
+# extension) every <= GATE_ROUND_SECS. Sized for gated_phases (tc01,
+# max_duration=15): 4s rounds land several renewals per 15s lease window.
+GATE_ROUND_SECS = 4.0
+# Per-phase gate budget: GATE_ROUNDS * GATE_ROUND_SECS = 120s. Deliberately
+# LONGER than any lease window in the tests — the whole point is that a
+# quiet gate outlasting the lease must survive on poll-liveness alone.
+GATE_ROUNDS = 30
+
+# sleepy_phases (tc02) submits with max_duration=6 — a 6s lease. A 4s poll
+# round would leave only ~2s of renewal margin per round against that lease
+# (and with max_retries already spent on the deliberate reclaim, a >2s CI
+# hiccup coinciding with the 10s sweep tick would terminally fail the job).
+# The finish gate on the re-claimed attempt therefore polls in 2s rounds:
+# >=2 liveness renewals per 6s lease window. Same 120s total budget.
+SLEEPY_GATE_ROUND_SECS = 2.0
+SLEEPY_GATE_ROUNDS = 60
+
+
+async def _gate(
+ job: MeshJob,
+ types: list[str],
+ round_secs: float = GATE_ROUND_SECS,
+ rounds: int = GATE_ROUNDS,
+):
+ """Block on the next event of ``types``, polling in short rounds.
+
+ Each ``recv_event`` round is an identity-bearing executor read, so the
+ registry extends the lease per round (issue #1252 Phase 2). Returns the
+ event, or ``None`` when the whole gate budget elapses.
+ """
+ for _ in range(rounds):
+ event = await job.recv_event(types=types, timeout_secs=round_secs)
+ if event is not None:
+ return event
+ return None
+
+
+async def _post_transition(job: MeshJob, **payload: Any) -> None:
+ """Stamp an app-level transition event with (replica, epoch)."""
+ await mesh.jobs.post_event(
+ job_id=job.job_id,
+ event_type="transition",
+ payload={"replica": REPLICA, "epoch": job.claim_epoch, **payload},
+ )
+
+
+# ---------------------------------------------------------------------------
+# tc01 — quiet gate must survive on poll-liveness (single owner, epoch 1)
+# ---------------------------------------------------------------------------
+
+
+@app.tool()
+@mesh.tool(
+ capability="gated_phases",
+ task=True,
+ description="Sequential recv_event('go') gates; posts a stamped transition per phase.",
+)
+async def gated_phases(
+ phases: int = 3,
+ job: MeshJob = None,
+) -> dict[str, Any]:
+ if job is None:
+ return {"status": "no_job_ctx"}
+ epoch = job.claim_epoch
+ for phase in range(1, phases + 1):
+ event = await _gate(job, ["go"])
+ if event is None:
+ # Loud terminal failure — a silent return would let the test
+ # misread a swallowed event as a pass.
+ await job.fail(f"gate timeout at phase {phase} (replica={REPLICA})")
+ return {"status": "gate_timeout", "phase": phase, "replica": REPLICA}
+ await _post_transition(job, marker="phase_done", phase=phase)
+ await job.update_progress(
+ phase / phases, f"phase {phase}/{phases} (replica={REPLICA}, epoch={epoch})"
+ )
+ payload = {"status": "done", "phases": phases, "epoch": epoch, "replica": REPLICA}
+ await job.complete(payload)
+ return payload
+
+
+# ---------------------------------------------------------------------------
+# tc02 — wedged attempt 1 is superseded; its post-wake writes are fenced
+# ---------------------------------------------------------------------------
+#
+# Attempt counter shared across replicas via a job-id-keyed file (both
+# replicas run in the same test container). Same pattern as uc21's
+# report_with_transient_failures counter — the reclaim dispatches the retry
+# on whichever replica's claim worker wins, so process-local state can't
+# distinguish attempts.
+
+
+def _bump_attempt(job_id: str) -> int:
+ path = f"/tmp/uc33-sleepy-attempt-{job_id}"
+ try:
+ with open(path) as f:
+ n = int((f.read() or "0").strip())
+ except FileNotFoundError:
+ n = 0
+ n += 1
+ with open(path, "w") as f:
+ f.write(str(n))
+ return n
+
+
+@app.tool()
+@mesh.tool(
+ capability="sleepy_phases",
+ task=True,
+ description="Attempt 1 wedges in a pure sleep past the lease; re-claimed attempt gates on 'finish'.",
+)
+async def sleepy_phases(
+ sleep_secs: int = 35,
+ job: MeshJob = None,
+) -> dict[str, Any]:
+ if job is None:
+ return {"status": "no_job_ctx"}
+ epoch = job.claim_epoch
+ attempt = _bump_attempt(job.job_id)
+ await _post_transition(job, marker="claimed", attempt=attempt)
+
+ if attempt == 1:
+ # WEDGED-OWNER SIMULATION: pure sleep — no recv_event polls (no
+ # poll-liveness credit) and no progress deltas (no lease renewal).
+ # The lease (sized by max_duration) lapses mid-sleep; the registry
+ # reclaims and a peer (or this same instance) claims epoch 2.
+ await asyncio.sleep(float(sleep_secs))
+
+ # Post-wake: this attempt has been superseded. The progress delta
+ # below carries the STALE epoch — the registry must reject it as
+ # claim_superseded, which fires this attempt's cancel token. The
+ # chunked grace loop below gives the async cancel a window to land
+ # (batch flush + token propagation) BEFORE the would-be duplicate
+ # side effects further down.
+ await job.update_progress(
+ 0.9, f"post-sleep write from superseded attempt (replica={REPLICA})"
+ )
+ for _ in range(20):
+ await asyncio.sleep(0.5)
+
+ # DUPLICATE side effects — with working supersession fencing the
+ # CancelledError above means execution NEVER reaches this point.
+ # If it does, the post_sleep transition lands in the event log and
+ # the test fails on it.
+ await _post_transition(job, marker="post_sleep", attempt=attempt)
+ payload = {"status": "done", "attempt": attempt, "epoch": epoch, "replica": REPLICA}
+ await job.complete(payload)
+ return payload
+
+ # attempt >= 2: the healthy re-claimed run. Gate on 'finish' with SHORT
+ # 2s polling rounds (see SLEEPY_GATE_ROUND_SECS — the 6s lease leaves no
+ # margin for 4s rounds) — poll-liveness keeps THIS attempt's lease alive
+ # while the test window proves the stale attempt was fenced on a LIVE row.
+ event = await _gate(
+ job,
+ ["finish"],
+ round_secs=SLEEPY_GATE_ROUND_SECS,
+ rounds=SLEEPY_GATE_ROUNDS,
+ )
+ if event is None:
+ await job.fail(f"finish gate timeout on re-claimed attempt (replica={REPLICA})")
+ return {"status": "gate_timeout", "attempt": attempt, "replica": REPLICA}
+ await _post_transition(job, marker="finish_seen", attempt=attempt)
+ payload = {"status": "done", "attempt": attempt, "epoch": epoch, "replica": REPLICA}
+ await job.complete(payload)
+ return payload
+
+
+@mesh.agent(
+ name="gated-worker-a",
+ version="1.0.0",
+ description="Gated MeshJob worker replica A (uc33) — multi-replica execution integrity fixture.",
+ http_port=int(os.environ.get("MCP_MESH_HTTP_PORT", "9111")),
+ enable_http=True,
+ auto_run=True,
+)
+class GatedWorkerA:
+ pass
diff --git a/tests/integration/suites/uc33_meshjob_replicas/fixtures/gated-worker-b/main.py b/tests/integration/suites/uc33_meshjob_replicas/fixtures/gated-worker-b/main.py
new file mode 100644
index 000000000..e5195f96a
--- /dev/null
+++ b/tests/integration/suites/uc33_meshjob_replicas/fixtures/gated-worker-b/main.py
@@ -0,0 +1,231 @@
+#!/usr/bin/env python3
+"""Gated MeshJob worker — replica B (uc33, issue #1252 Phase 4).
+
+Replica pair: ``gated-worker-a`` / ``gated-worker-b`` are byte-identical
+except for the declared agent name, the ``REPLICA`` stamp and the default
+port. They are distinct fixture files because meshctl's "is this agent
+already running" check keys off the ``@mesh.agent(name=...)`` decorator
+literal (same pattern as uc21's bystander-x / bystander-y). The tests start
+BOTH with ``--env MCP_MESH_AGENT_NAME=gated-worker`` so both INSTANCE IDS
+share the ``gated-worker-`` prefix (the env var only seeds the instance-id
+prefix — the registered ``/agents`` ``.name`` stays this decorator literal).
+What mirrors production replicas (``replicas: 2``) is the part #1252 cares
+about: two claim workers providing the SAME task capability on ONE shared
+job queue — while meshctl manages them as two distinct local processes.
+
+Both capabilities are ``task=True`` and claimed via the shared job queue:
+whichever replica's claim worker wins executes the attempt. Every app-level
+"transition" event the handlers post is stamped with
+``{marker, phase/attempt, epoch: job.claim_epoch, replica: REPLICA}`` so the
+test can detect duplicated execution and attribute every side effect to a
+specific (replica, claim-epoch) pair.
+
+Capabilities:
+
+- ``gated_phases`` — the #1252 field shape: sequential ``recv_event`` gates.
+ Each gate polls in SHORT rounds (``GATE_ROUND_SECS`` ≤ 5s), so every round
+ is one executor read = one poll-liveness lease extension, keeping the
+ extension cadence well inside even a small lease window. A handler blocked
+ on a legitimately quiet gate is therefore provably alive — the registry
+ must NOT reclaim it (tc01).
+
+- ``sleepy_phases`` — the wedged-owner shape: attempt 1 blocks in a PURE
+ ``asyncio.sleep`` (no recv_event polls → no liveness credit, no progress
+ deltas → no lease renewal) past the lease window, so the registry
+ legitimately reclaims and the next claim (epoch 2) supersedes attempt 1.
+ When attempt 1 wakes, its first fenced write (a progress delta carrying
+ the stale epoch) must be rejected as ``claim_superseded`` and fire this
+ attempt's cancel token — aborting it BEFORE the duplicate side effects
+ further down the handler body can land (tc02).
+"""
+
+import asyncio
+import os
+from typing import Any
+
+import mesh
+from fastmcp import FastMCP
+from mesh import MeshJob
+
+REPLICA = "b"
+
+app = FastMCP(f"Gated Worker {REPLICA.upper()} (uc33)")
+
+# Short recv_event rounds: one executor read (= one poll-liveness lease
+# extension) every <= GATE_ROUND_SECS. Sized for gated_phases (tc01,
+# max_duration=15): 4s rounds land several renewals per 15s lease window.
+GATE_ROUND_SECS = 4.0
+# Per-phase gate budget: GATE_ROUNDS * GATE_ROUND_SECS = 120s. Deliberately
+# LONGER than any lease window in the tests — the whole point is that a
+# quiet gate outlasting the lease must survive on poll-liveness alone.
+GATE_ROUNDS = 30
+
+# sleepy_phases (tc02) submits with max_duration=6 — a 6s lease. A 4s poll
+# round would leave only ~2s of renewal margin per round against that lease
+# (and with max_retries already spent on the deliberate reclaim, a >2s CI
+# hiccup coinciding with the 10s sweep tick would terminally fail the job).
+# The finish gate on the re-claimed attempt therefore polls in 2s rounds:
+# >=2 liveness renewals per 6s lease window. Same 120s total budget.
+SLEEPY_GATE_ROUND_SECS = 2.0
+SLEEPY_GATE_ROUNDS = 60
+
+
+async def _gate(
+ job: MeshJob,
+ types: list[str],
+ round_secs: float = GATE_ROUND_SECS,
+ rounds: int = GATE_ROUNDS,
+):
+ """Block on the next event of ``types``, polling in short rounds.
+
+ Each ``recv_event`` round is an identity-bearing executor read, so the
+ registry extends the lease per round (issue #1252 Phase 2). Returns the
+ event, or ``None`` when the whole gate budget elapses.
+ """
+ for _ in range(rounds):
+ event = await job.recv_event(types=types, timeout_secs=round_secs)
+ if event is not None:
+ return event
+ return None
+
+
+async def _post_transition(job: MeshJob, **payload: Any) -> None:
+ """Stamp an app-level transition event with (replica, epoch)."""
+ await mesh.jobs.post_event(
+ job_id=job.job_id,
+ event_type="transition",
+ payload={"replica": REPLICA, "epoch": job.claim_epoch, **payload},
+ )
+
+
+# ---------------------------------------------------------------------------
+# tc01 — quiet gate must survive on poll-liveness (single owner, epoch 1)
+# ---------------------------------------------------------------------------
+
+
+@app.tool()
+@mesh.tool(
+ capability="gated_phases",
+ task=True,
+ description="Sequential recv_event('go') gates; posts a stamped transition per phase.",
+)
+async def gated_phases(
+ phases: int = 3,
+ job: MeshJob = None,
+) -> dict[str, Any]:
+ if job is None:
+ return {"status": "no_job_ctx"}
+ epoch = job.claim_epoch
+ for phase in range(1, phases + 1):
+ event = await _gate(job, ["go"])
+ if event is None:
+ # Loud terminal failure — a silent return would let the test
+ # misread a swallowed event as a pass.
+ await job.fail(f"gate timeout at phase {phase} (replica={REPLICA})")
+ return {"status": "gate_timeout", "phase": phase, "replica": REPLICA}
+ await _post_transition(job, marker="phase_done", phase=phase)
+ await job.update_progress(
+ phase / phases, f"phase {phase}/{phases} (replica={REPLICA}, epoch={epoch})"
+ )
+ payload = {"status": "done", "phases": phases, "epoch": epoch, "replica": REPLICA}
+ await job.complete(payload)
+ return payload
+
+
+# ---------------------------------------------------------------------------
+# tc02 — wedged attempt 1 is superseded; its post-wake writes are fenced
+# ---------------------------------------------------------------------------
+#
+# Attempt counter shared across replicas via a job-id-keyed file (both
+# replicas run in the same test container). Same pattern as uc21's
+# report_with_transient_failures counter — the reclaim dispatches the retry
+# on whichever replica's claim worker wins, so process-local state can't
+# distinguish attempts.
+
+
+def _bump_attempt(job_id: str) -> int:
+ path = f"/tmp/uc33-sleepy-attempt-{job_id}"
+ try:
+ with open(path) as f:
+ n = int((f.read() or "0").strip())
+ except FileNotFoundError:
+ n = 0
+ n += 1
+ with open(path, "w") as f:
+ f.write(str(n))
+ return n
+
+
+@app.tool()
+@mesh.tool(
+ capability="sleepy_phases",
+ task=True,
+ description="Attempt 1 wedges in a pure sleep past the lease; re-claimed attempt gates on 'finish'.",
+)
+async def sleepy_phases(
+ sleep_secs: int = 35,
+ job: MeshJob = None,
+) -> dict[str, Any]:
+ if job is None:
+ return {"status": "no_job_ctx"}
+ epoch = job.claim_epoch
+ attempt = _bump_attempt(job.job_id)
+ await _post_transition(job, marker="claimed", attempt=attempt)
+
+ if attempt == 1:
+ # WEDGED-OWNER SIMULATION: pure sleep — no recv_event polls (no
+ # poll-liveness credit) and no progress deltas (no lease renewal).
+ # The lease (sized by max_duration) lapses mid-sleep; the registry
+ # reclaims and a peer (or this same instance) claims epoch 2.
+ await asyncio.sleep(float(sleep_secs))
+
+ # Post-wake: this attempt has been superseded. The progress delta
+ # below carries the STALE epoch — the registry must reject it as
+ # claim_superseded, which fires this attempt's cancel token. The
+ # chunked grace loop below gives the async cancel a window to land
+ # (batch flush + token propagation) BEFORE the would-be duplicate
+ # side effects further down.
+ await job.update_progress(
+ 0.9, f"post-sleep write from superseded attempt (replica={REPLICA})"
+ )
+ for _ in range(20):
+ await asyncio.sleep(0.5)
+
+ # DUPLICATE side effects — with working supersession fencing the
+ # CancelledError above means execution NEVER reaches this point.
+ # If it does, the post_sleep transition lands in the event log and
+ # the test fails on it.
+ await _post_transition(job, marker="post_sleep", attempt=attempt)
+ payload = {"status": "done", "attempt": attempt, "epoch": epoch, "replica": REPLICA}
+ await job.complete(payload)
+ return payload
+
+ # attempt >= 2: the healthy re-claimed run. Gate on 'finish' with SHORT
+ # 2s polling rounds (see SLEEPY_GATE_ROUND_SECS — the 6s lease leaves no
+ # margin for 4s rounds) — poll-liveness keeps THIS attempt's lease alive
+ # while the test window proves the stale attempt was fenced on a LIVE row.
+ event = await _gate(
+ job,
+ ["finish"],
+ round_secs=SLEEPY_GATE_ROUND_SECS,
+ rounds=SLEEPY_GATE_ROUNDS,
+ )
+ if event is None:
+ await job.fail(f"finish gate timeout on re-claimed attempt (replica={REPLICA})")
+ return {"status": "gate_timeout", "attempt": attempt, "replica": REPLICA}
+ await _post_transition(job, marker="finish_seen", attempt=attempt)
+ payload = {"status": "done", "attempt": attempt, "epoch": epoch, "replica": REPLICA}
+ await job.complete(payload)
+ return payload
+
+
+@mesh.agent(
+ name="gated-worker-b",
+ version="1.0.0",
+ description="Gated MeshJob worker replica B (uc33) — multi-replica execution integrity fixture.",
+ http_port=int(os.environ.get("MCP_MESH_HTTP_PORT", "9112")),
+ enable_http=True,
+ auto_run=True,
+)
+class GatedWorkerB:
+ pass
diff --git a/tests/integration/suites/uc33_meshjob_replicas/routines.yaml b/tests/integration/suites/uc33_meshjob_replicas/routines.yaml
new file mode 100644
index 000000000..ab7da2619
--- /dev/null
+++ b/tests/integration/suites/uc33_meshjob_replicas/routines.yaml
@@ -0,0 +1,60 @@
+# UC-level routines for uc33_meshjob_replicas (issue #1252 Phase 4).
+#
+# Clone of uc21_meshjob's fast-sweep registry routine: the lease-reclaim
+# path under test is driven by the registry's sweep loop, whose default
+# 5-minute tick is far too slow for a test that must observe (or prove
+# the absence of) a re-claim within seconds. See uc21/routines.yaml for
+# the full knob-chain documentation.
+#
+# NOTE: MCP_MESH_JOB_STALE_TIMEOUT is deliberately NOT set. With no stale
+# ceiling and no total_deadline, cappedPollLease has no cap — an executor
+# recv_event poll extends the lease freely (the poll-liveness behavior
+# tc01 asserts). A stale ceiling would reap the quiet-gate job outright
+# and mask the lease-reclaim-vs-poll-liveness distinction.
+
+routines:
+ start_registry_with_fast_sweep:
+ description: "Start the mesh registry with sweep/health knobs cranked for fast lease-reclaim tests"
+ steps:
+ - handler: shell
+ workdir: /workspace
+ command: |
+ # Reclaim timing is driven by the SWEEP TICK (10s) + the job's own
+ # lease — MCP_MESH_RETENTION is a separate knob that only controls
+ # unhealthy-agent purge and event-log GC. Keep it at 60s: at 10s,
+ # two missed 5s heartbeats would PURGE the agent (an orphan-reset
+ # masquerading as a fencing failure) and terminal-event GC at
+ # now-10s could wipe the event log before the final captures.
+ MCP_MESH_SWEEP_INTERVAL=10s \
+ MCP_MESH_RETENTION=60s \
+ MCP_MESH_HEALTH_CHECK_INTERVAL=5 \
+ DEFAULT_TIMEOUT_THRESHOLD=15 \
+ meshctl start --registry-only -d
+ for i in $(seq 1 20); do
+ if curl -sf http://localhost:8000/health > /dev/null 2>&1; then
+ echo "registry ready after ${i}s"
+ # Verify the sweep override actually stamped its log line.
+ # If MCP_MESH_SWEEP_INTERVAL didn't take effect the suite
+ # must fail loudly here, not via slow per-test timeouts.
+ tail -n 200 ~/.mcp-mesh/logs/registry.log \
+ | grep -qE '\[sweep\] using MCP_MESH_SWEEP_INTERVAL' \
+ || { echo "FAIL: MCP_MESH_SWEEP_INTERVAL override did not take effect"; exit 1; }
+ echo "[setup] confirmed sweep override active"
+ exit 0
+ fi
+ sleep 1
+ done
+ echo "ERROR: registry did not become ready in 20s"
+ tail -50 ~/.mcp-mesh/logs/registry.log 2>/dev/null || true
+ exit 1
+ capture: registry_start
+ timeout: 30
+
+ # Stop everything started by this test (agents + registry).
+ stop_all:
+ description: "Stop all mesh processes started by this test"
+ steps:
+ - handler: shell
+ workdir: /workspace
+ command: meshctl stop 2>/dev/null || true
+ ignore_errors: true
diff --git a/tests/integration/suites/uc33_meshjob_replicas/tc01_quiet_gate_single_owner/test.yaml b/tests/integration/suites/uc33_meshjob_replicas/tc01_quiet_gate_single_owner/test.yaml
new file mode 100644
index 000000000..2ba9713ff
--- /dev/null
+++ b/tests/integration/suites/uc33_meshjob_replicas/tc01_quiet_gate_single_owner/test.yaml
@@ -0,0 +1,327 @@
+# tc01_quiet_gate_single_owner — a quietly-waiting gate earns poll-liveness;
+# the job is NEVER re-claimed (issue #1252 Phases 1+2, PR #1253).
+#
+# Field shape (from the #1252 repro): an interactive job runs sequential
+# recv_event(type="go") gates with the executing capability at 2 replicas.
+# One gate legitimately blocks in silence LONGER than the lease window.
+# Pre-fix, recv_event was an anonymous read that never touched the lease:
+# the lease lapsed mid-gate, the sweep re-dispatched the job to the peer
+# replica, and both owners executed — duplicate transitions, phase slip.
+# Post-fix, every executor recv_event poll carries (instance_id, claim_epoch)
+# and extends the lease — an actively-polling handler is provably alive.
+#
+# Topology: gated-worker-a + gated-worker-b (both providing task=true
+# capability gated_phases; MCP_MESH_AGENT_NAME=gated-worker gives both a
+# shared "gated-worker-" instance-id PREFIX — the production replica shape
+# that matters here is two claim workers on ONE shared job queue. NOTE:
+# MCP_MESH_AGENT_NAME only seeds the instance id — the registered /agents
+# .name stays each fixture's decorator literal, gated-worker-a/-b) +
+# gate-driver (submit-only). The worker polls its gates in <=4s rounds, so
+# poll-liveness credit lands several times per lease window (max_duration=15).
+#
+# Flow:
+# 1. Submit gated_phases (3 phases, max_duration=15 -> lease window 15s,
+# max_retries=2 so a pre-fix lapse would RECLAIM, not fail).
+# 2. Feed "go" #1 normally; observe the phase_done(1) transition.
+# 3. GO SILENT for 30s — twice the lease window, and past the fast-sweep
+# tick (10s). Only executor poll credit can keep the claim alive.
+# 4. Verify mid-silence: still working, attempt_count == 1, SAME owner.
+# 5. Feed "go" #2 and #3; job completes.
+#
+# Asserts (single claim, no duplicate execution):
+# - attempt_count == 1 (claim_epoch is minted per claim; a re-claim would
+# bump both — attempt_count is the row-level observable, and every
+# handler-stamped transition carries the epoch itself)
+# - owner_instance_id unchanged across the silent window
+# - exactly 3 phase_done transitions: phases {1,2,3}, ALL epoch 1, ALL
+# from ONE replica — no duplicates, no second owner
+# - result payload stamped {status: done, epoch: 1}
+
+name: "MeshJob replicas: quiet recv_event gate outlasting the lease is NOT re-claimed"
+description: "2-replica gated_phases job survives a 30s silent gate (lease 15s) on poll-liveness alone — single claim, epoch 1, no duplicate transitions"
+tags:
+ - meshjob
+ - python
+ - issue-1252
+ - replicas
+ - lease
+ - slow
+timeout: 300
+
+pre_run:
+ - routine: global.setup_for_python_agent
+
+test:
+ - name: "Copy artifacts"
+ handler: shell
+ workdir: /workspace
+ command: |
+ cp -rL /uc-artifacts/gated-worker-a /workspace/
+ cp -rL /uc-artifacts/gated-worker-b /workspace/
+ cp -rL /uc-artifacts/gate-driver /workspace/
+
+ - name: "Install worker A deps"
+ handler: pip-install
+ path: /workspace/gated-worker-a
+
+ - name: "Install worker B deps"
+ handler: pip-install
+ path: /workspace/gated-worker-b
+
+ - name: "Install driver deps"
+ handler: pip-install
+ path: /workspace/gate-driver
+
+ # Fast sweep so a lease lapse (the pre-fix failure mode) would surface as
+ # a reclaim within ~10s instead of ~5min. NO stale timeout — poll-liveness
+ # must be able to extend the lease freely.
+ - routine: start_registry_with_fast_sweep
+
+ # Both replicas register under the SAME agent name (production shape);
+ # meshctl manages them as distinct processes via their declared names.
+ - name: "Start worker replica A (port 9111)"
+ handler: shell
+ workdir: /workspace
+ command: meshctl start gated-worker-a/main.py --env MCP_MESH_HTTP_PORT=9111 --env MCP_MESH_AGENT_NAME=gated-worker -d
+
+ - name: "Start worker replica B (port 9112)"
+ handler: shell
+ workdir: /workspace
+ command: meshctl start gated-worker-b/main.py --env MCP_MESH_HTTP_PORT=9112 --env MCP_MESH_AGENT_NAME=gated-worker -d
+
+ - name: "Start driver (port 9110)"
+ handler: shell
+ workdir: /workspace
+ command: meshctl start gate-driver/main.py --env MCP_MESH_HTTP_PORT=9110 --env MCP_MESH_SETTLE_TIMEOUT=60 -d
+
+ # Liveness only: BOTH gated-worker instances + the driver must be
+ # registered. Matching is PREFIX-based on /agents .name: the replicas
+ # register as gated-worker-a / gated-worker-b (MCP_MESH_AGENT_NAME only
+ # prefixes the instance id, it does not rename the agent). The trailing
+ # dash in "gated-worker-" keeps gate-driver from ever cross-matching.
+ # Dependency resolution is covered by the settle window — no
+ # dep-resolution wait.
+ - name: "Wait for both replicas and the driver to register"
+ handler: shell
+ workdir: /workspace
+ command: |
+ echo "Waiting for 2 gated-worker replicas + gate-driver..."
+ for i in $(seq 1 60); do
+ REPLICAS=$(curl -s http://localhost:8000/agents | jq '[.agents[] | select(.name | startswith("gated-worker-"))] | length' 2>/dev/null || echo 0)
+ DRIVER=$(curl -s http://localhost:8000/agents | jq '[.agents[] | select(.name | startswith("gate-driver"))] | length' 2>/dev/null || echo 0)
+ if [ "$REPLICAS" -ge 2 ] && [ "$DRIVER" -ge 1 ]; then
+ echo "REPLICAS_READY=$REPLICAS after ~${i}s"
+ exit 0
+ fi
+ sleep 1
+ done
+ echo "ERROR: replicas did not register within 60s (replicas=$REPLICAS driver=$DRIVER)"
+ meshctl list 2>/dev/null || true
+ exit 1
+ capture: wait_registration
+ timeout: 90
+
+ - name: "Submit gated_phases (3 phases, max_duration=15, max_retries=2)"
+ handler: shell
+ workdir: /workspace
+ command: |
+ meshctl call submit_gated '{"phases":3,"max_duration":15,"max_retries":2}'
+ capture: submit_resp
+ timeout: 30
+
+ - name: "Capture job_id"
+ handler: shell
+ workdir: /workspace
+ command: |
+ echo '${captured.submit_resp}' | jq -r '.. | objects | select(has("job_id")) | .job_id' | head -1 | xargs -I{} echo "JOB_ID={}"
+ capture: job_id_extract
+
+ # Deterministic claim gate: wait until a replica has CLAIMED the job and
+ # record WHICH one, so the silent-window check below can prove the owner
+ # never changed.
+ - name: "Wait for the claim and record the owner"
+ handler: shell
+ workdir: /workspace
+ command: |
+ JOB_ID=$(echo '${captured.job_id_extract}' | grep '^JOB_ID=' | cut -d= -f2)
+ echo "JOB_ID=${JOB_ID}"
+ for i in $(seq 1 30); do
+ OWNER=$(curl -s "http://localhost:8000/jobs/${JOB_ID}" | jq -r '.owner_instance_id // empty')
+ if [ -n "$OWNER" ]; then
+ echo "CLAIMED=1"
+ echo "OWNER=${OWNER}"
+ exit 0
+ fi
+ sleep 1
+ done
+ echo "ERROR: job never claimed within 30s"
+ curl -s "http://localhost:8000/jobs/${JOB_ID}" | jq '.'
+ exit 1
+ capture: claim_wait
+ timeout: 45
+
+ - name: "Post go #1 and wait for the phase 1 transition"
+ handler: shell
+ workdir: /workspace
+ command: |
+ JOB_ID=$(echo '${captured.job_id_extract}' | grep '^JOB_ID=' | cut -d= -f2)
+ curl -s -X POST "http://localhost:8000/jobs/${JOB_ID}/events" \
+ -H 'Content-Type: application/json' \
+ -d '{"type":"go","payload":{"n":1}}' | jq -c '.'
+ for i in $(seq 1 20); do
+ SEEN=$(curl -s "http://localhost:8000/jobs/${JOB_ID}/events?after=0&limit=100" \
+ | jq '[.events[] | select(.type == "transition") | select(.payload.phase == 1)] | length')
+ if [ "$SEEN" -ge 1 ]; then
+ echo "PHASE1_SEEN=1 after ~${i}s"
+ exit 0
+ fi
+ sleep 1
+ done
+ echo "ERROR: phase 1 transition never appeared within 20s"
+ curl -s "http://localhost:8000/jobs/${JOB_ID}/events?after=0&limit=100" | jq '.'
+ exit 1
+ capture: phase1_wait
+ timeout: 40
+
+ # THE QUIET GATE: 30 seconds of silence — twice the 15s lease window and
+ # well past the 10s sweep tick. The handler sits parked on recv_event
+ # rounds; only executor poll credit (Phase 2) keeps the claim alive. This
+ # is the scenario wait itself, NOT a registration/dep wait.
+ - name: "Go silent for a full gate window longer than the lease"
+ handler: wait
+ seconds: 30
+
+ # Mid-silence integrity: still the FIRST claim, SAME owner, still working.
+ - name: "Verify the job was not re-claimed during the silence"
+ handler: shell
+ workdir: /workspace
+ command: |
+ JOB_ID=$(echo '${captured.job_id_extract}' | grep '^JOB_ID=' | cut -d= -f2)
+ OWNER_BEFORE=$(echo '${captured.claim_wait}' | grep '^OWNER=' | cut -d= -f2)
+ ROW=$(curl -s "http://localhost:8000/jobs/${JOB_ID}")
+ echo "$ROW" | jq '{status, attempt_count, owner_instance_id}'
+ STATUS=$(echo "$ROW" | jq -r '.status')
+ ATTEMPTS=$(echo "$ROW" | jq -r '.attempt_count')
+ OWNER_NOW=$(echo "$ROW" | jq -r '.owner_instance_id // empty')
+ if [ "$STATUS" != "working" ]; then
+ echo "FAIL: expected status=working mid-silence, got ${STATUS}"
+ exit 1
+ fi
+ if [ "$ATTEMPTS" != "1" ]; then
+ echo "FAIL: expected attempt_count=1 mid-silence, got ${ATTEMPTS} — the quiet gate was re-claimed"
+ exit 1
+ fi
+ if [ "$OWNER_NOW" != "$OWNER_BEFORE" ]; then
+ echo "FAIL: owner changed mid-silence (${OWNER_BEFORE} -> ${OWNER_NOW})"
+ exit 1
+ fi
+ echo "OWNER_UNCHANGED=1"
+ echo "ATTEMPT_STILL_1=1"
+ capture: mid_silence_check
+ timeout: 30
+
+ - name: "Post go #2 and wait for the phase 2 transition"
+ handler: shell
+ workdir: /workspace
+ command: |
+ JOB_ID=$(echo '${captured.job_id_extract}' | grep '^JOB_ID=' | cut -d= -f2)
+ curl -s -X POST "http://localhost:8000/jobs/${JOB_ID}/events" \
+ -H 'Content-Type: application/json' \
+ -d '{"type":"go","payload":{"n":2}}' | jq -c '.'
+ for i in $(seq 1 20); do
+ SEEN=$(curl -s "http://localhost:8000/jobs/${JOB_ID}/events?after=0&limit=100" \
+ | jq '[.events[] | select(.type == "transition") | select(.payload.phase == 2)] | length')
+ if [ "$SEEN" -ge 1 ]; then
+ echo "PHASE2_SEEN=1 after ~${i}s"
+ exit 0
+ fi
+ sleep 1
+ done
+ echo "ERROR: phase 2 transition never appeared within 20s — the silent gate lost the claim or the event"
+ curl -s "http://localhost:8000/jobs/${JOB_ID}/events?after=0&limit=100" | jq '.'
+ curl -s "http://localhost:8000/jobs/${JOB_ID}" | jq '{status, attempt_count, owner_instance_id, error}'
+ exit 1
+ capture: phase2_wait
+ timeout: 40
+
+ - name: "Post go #3 and wait for completion"
+ handler: shell
+ workdir: /workspace
+ command: |
+ JOB_ID=$(echo '${captured.job_id_extract}' | grep '^JOB_ID=' | cut -d= -f2)
+ curl -s -X POST "http://localhost:8000/jobs/${JOB_ID}/events" \
+ -H 'Content-Type: application/json' \
+ -d '{"type":"go","payload":{"n":3}}' | jq -c '.'
+ for i in $(seq 1 30); do
+ ST=$(curl -s "http://localhost:8000/jobs/${JOB_ID}" | jq -r '.status')
+ if [ "$ST" = "completed" ]; then
+ echo "COMPLETED=1 after ~${i}s"
+ exit 0
+ fi
+ sleep 1
+ done
+ echo "ERROR: job did not complete within 30s of go #3"
+ curl -s "http://localhost:8000/jobs/${JOB_ID}" | jq '.'
+ exit 1
+ capture: completion_wait
+ timeout: 45
+
+ - name: "Capture final job row"
+ handler: shell
+ workdir: /workspace
+ command: |
+ JOB_ID=$(echo '${captured.job_id_extract}' | grep '^JOB_ID=' | cut -d= -f2)
+ curl -s "http://localhost:8000/jobs/${JOB_ID}"
+ capture: final_job
+
+ - name: "Capture final event log"
+ handler: shell
+ workdir: /workspace
+ command: |
+ JOB_ID=$(echo '${captured.job_id_extract}' | grep '^JOB_ID=' | cut -d= -f2)
+ curl -s "http://localhost:8000/jobs/${JOB_ID}/events?after=0&limit=100"
+ capture: final_events
+
+assertions:
+ # IDIOM: equality lives INSIDE the jq filter so the interpolation yields a
+ # bare true/false (tsuite Go-formats extracted values, and the
+ # interpolation body cannot contain '}').
+
+ # Scenario preconditions surfaced by the driver steps
+ - expr: "${captured.wait_registration} contains 'REPLICAS_READY=2'"
+ message: "both gated-worker-* replicas must register before the job is submitted"
+ - expr: "${captured.claim_wait} contains 'CLAIMED=1'"
+ message: "a replica must claim the job before the gates are fed"
+ - expr: "${captured.phase1_wait} contains 'PHASE1_SEEN=1'"
+ message: "phase 1 must transition before the silent window starts"
+ - expr: "${captured.mid_silence_check} contains 'OWNER_UNCHANGED=1'"
+ message: "the owner must not change across a silent gate longer than the lease"
+ - expr: "${captured.mid_silence_check} contains 'ATTEMPT_STILL_1=1'"
+ message: "the quiet gate must NOT be re-claimed — poll-liveness keeps the claim alive"
+ - expr: "${captured.completion_wait} contains 'COMPLETED=1'"
+ message: "the job must complete once all three gates are fed"
+
+ # Single claim end-to-end: one attempt, first epoch.
+ - expr: "${jq:captured.final_job:(.status == \"completed\") and (.attempt_count == 1)} == 'true'"
+ message: "job must complete on its FIRST and ONLY claim (attempt_count == 1)"
+ - expr: "${jq:captured.final_job:(.result.status == \"done\") and (.result.epoch == 1)} == 'true'"
+ message: "the completing handler must have executed under claim epoch 1"
+
+ # No duplicate transitions: exactly one per phase, all epoch 1, one replica.
+ - expr: "${jq:captured.final_events:([.events[] | select(.type == \"transition\")] | length) == 3} == 'true'"
+ message: "exactly 3 phase transitions — a re-claimed duplicate execution would post more"
+ - expr: "${jq:captured.final_events:([.events[] | select(.type == \"transition\") | .payload.phase] | sort) == [1,2,3]} == 'true'"
+ message: "phases 1,2,3 must each transition exactly once (sorted set-equality — ordering is not asserted)"
+ - expr: "${jq:captured.final_events:([.events[] | select(.type == \"transition\")] | all(.payload.epoch == 1))} == 'true'"
+ message: "every transition must be stamped with claim epoch 1 — no second claim ever executed"
+ - expr: "${jq:captured.final_events:([.events[] | select(.type == \"transition\") | .payload.replica] | unique | length) == 1} == 'true'"
+ message: "all transitions must come from ONE replica — dual ownership would interleave both"
+
+ # No undelivered events: all 3 posted 'go' events are in the log and the
+ # job consumed them all (it completed after exactly 3 gates).
+ - expr: "${jq:captured.final_events:([.events[] | select(.type == \"go\")] | length) == 3} == 'true'"
+ message: "all three posted go events must be in the job's event log"
+
+post_run:
+ - routine: stop_all
+ - routine: global.cleanup_workspace
diff --git a/tests/integration/suites/uc33_meshjob_replicas/tc02_supersession_fences_stale_owner/test.yaml b/tests/integration/suites/uc33_meshjob_replicas/tc02_supersession_fences_stale_owner/test.yaml
new file mode 100644
index 000000000..b81aada4e
--- /dev/null
+++ b/tests/integration/suites/uc33_meshjob_replicas/tc02_supersession_fences_stale_owner/test.yaml
@@ -0,0 +1,305 @@
+# tc02_supersession_fences_stale_owner — a genuinely wedged attempt is
+# reclaimed (epoch 2) and the stale owner's post-wake writes are fenced
+# (issue #1252 Phase 1, PR #1253).
+#
+# Scenario: sleepy_phases attempt 1 blocks in a PURE sleep (35s) — no
+# recv_event polls (no poll-liveness credit), no progress (no lease
+# renewal) — past its lease window (max_duration=6). The registry's
+# lease-reclaim sweep resets the row; a replica claims it again, minting
+# claim epoch 2 (reclaim itself does NOT bump the epoch — the next claim
+# does). The re-claimed attempt gates on a 'finish' event, holding the row
+# LIVE so the stale owner's post-wake write is fenced on a working row (not
+# trivially rejected as terminal).
+#
+# When attempt 1 wakes at ~35s the row belongs to epoch 2 (possibly on the
+# SAME instance — the same-instance re-claim gap is exactly what epochs
+# close). Its first write is a progress delta carrying the stale epoch:
+# the registry must reject it as claim_superseded, which fires attempt 1's
+# per-epoch cancel token and aborts the handler BEFORE its duplicate side
+# effects (a 'post_sleep' transition + a complete()) can land. The handler
+# gives the async cancel a 10s chunked grace window — if the fence or the
+# cancel translation is broken, 'post_sleep' lands in the event log and the
+# assertions fail.
+#
+# Asserts:
+# - a second claim happened: 2 'claimed' transitions, epochs {1, 2},
+# attempt_count == 2
+# - the row is STILL WORKING (attempt 2 gated) after the stale owner woke
+# and was fenced — its complete() did not land
+# - NO 'post_sleep' transition ever appears (stale owner aborted mid-grace)
+# - the stale owner's progress message never lands on the row
+# - exactly ONE owner's transitions appear after the re-claim
+# ('finish_seen' once, stamped epoch 2)
+# - the job completes exactly once, from attempt 2 / epoch 2
+
+name: "MeshJob replicas: wedged owner is superseded (epoch 2); stale writes are fenced"
+description: "sleepy_phases attempt 1 wedges past the lease; reclaim mints epoch 2, stale owner's post-wake writes are rejected and its handler aborted — job completes exactly once"
+tags:
+ - meshjob
+ - python
+ - issue-1252
+ - replicas
+ - lease
+ - fencing
+ - slow
+timeout: 300
+
+pre_run:
+ - routine: global.setup_for_python_agent
+
+test:
+ - name: "Copy artifacts"
+ handler: shell
+ workdir: /workspace
+ command: |
+ cp -rL /uc-artifacts/gated-worker-a /workspace/
+ cp -rL /uc-artifacts/gated-worker-b /workspace/
+ cp -rL /uc-artifacts/gate-driver /workspace/
+
+ - name: "Install worker A deps"
+ handler: pip-install
+ path: /workspace/gated-worker-a
+
+ - name: "Install worker B deps"
+ handler: pip-install
+ path: /workspace/gated-worker-b
+
+ - name: "Install driver deps"
+ handler: pip-install
+ path: /workspace/gate-driver
+
+ # Fast sweep so the lease-reclaim lands within ~10s of the lapse.
+ - routine: start_registry_with_fast_sweep
+
+ - name: "Start worker replica A (port 9111)"
+ handler: shell
+ workdir: /workspace
+ command: meshctl start gated-worker-a/main.py --env MCP_MESH_HTTP_PORT=9111 --env MCP_MESH_AGENT_NAME=gated-worker -d
+
+ - name: "Start worker replica B (port 9112)"
+ handler: shell
+ workdir: /workspace
+ command: meshctl start gated-worker-b/main.py --env MCP_MESH_HTTP_PORT=9112 --env MCP_MESH_AGENT_NAME=gated-worker -d
+
+ - name: "Start driver (port 9110)"
+ handler: shell
+ workdir: /workspace
+ command: meshctl start gate-driver/main.py --env MCP_MESH_HTTP_PORT=9110 --env MCP_MESH_SETTLE_TIMEOUT=60 -d
+
+ # PREFIX-matched on /agents .name: the replicas register as
+ # gated-worker-a / gated-worker-b (MCP_MESH_AGENT_NAME only prefixes the
+ # instance id, it does not rename the agent). The trailing dash in
+ # "gated-worker-" keeps gate-driver from ever cross-matching.
+ - name: "Wait for both replicas and the driver to register"
+ handler: shell
+ workdir: /workspace
+ command: |
+ echo "Waiting for 2 gated-worker replicas + gate-driver..."
+ for i in $(seq 1 60); do
+ REPLICAS=$(curl -s http://localhost:8000/agents | jq '[.agents[] | select(.name | startswith("gated-worker-"))] | length' 2>/dev/null || echo 0)
+ DRIVER=$(curl -s http://localhost:8000/agents | jq '[.agents[] | select(.name | startswith("gate-driver"))] | length' 2>/dev/null || echo 0)
+ if [ "$REPLICAS" -ge 2 ] && [ "$DRIVER" -ge 1 ]; then
+ echo "REPLICAS_READY=$REPLICAS after ~${i}s"
+ exit 0
+ fi
+ sleep 1
+ done
+ echo "ERROR: replicas did not register within 60s (replicas=$REPLICAS driver=$DRIVER)"
+ meshctl list 2>/dev/null || true
+ exit 1
+ capture: wait_registration
+ timeout: 90
+
+ # sleep_secs=35 comfortably outlasts the worst-case reclaim+re-claim
+ # (~6s lease + 10s sweep tick + ~5s claim-poll backoff ≈ 21s), so the
+ # stale owner always wakes AFTER epoch 2 exists.
+ - name: "Submit sleepy_phases (sleep=35, max_duration=6, max_retries=1)"
+ handler: shell
+ workdir: /workspace
+ command: |
+ meshctl call submit_sleepy '{"sleep_secs":35,"max_duration":6,"max_retries":1}'
+ capture: submit_resp
+ timeout: 30
+
+ - name: "Capture job_id"
+ handler: shell
+ workdir: /workspace
+ command: |
+ echo '${captured.submit_resp}' | jq -r '.. | objects | select(has("job_id")) | .job_id' | head -1 | xargs -I{} echo "JOB_ID={}"
+ capture: job_id_extract
+
+ # First claim: attempt 1's 'claimed' transition, stamped epoch 1.
+ - name: "Wait for attempt 1 to claim (epoch 1)"
+ handler: shell
+ workdir: /workspace
+ command: |
+ JOB_ID=$(echo '${captured.job_id_extract}' | grep '^JOB_ID=' | cut -d= -f2)
+ echo "JOB_ID=${JOB_ID}"
+ for i in $(seq 1 30); do
+ SEEN=$(curl -s "http://localhost:8000/jobs/${JOB_ID}/events?after=0&limit=100" \
+ | jq '[.events[] | select(.type == "transition") | select(.payload.marker == "claimed") | select(.payload.attempt == 1)] | length')
+ if [ "$SEEN" -ge 1 ]; then
+ echo "ATTEMPT1_CLAIMED=1 after ~${i}s"
+ exit 0
+ fi
+ sleep 1
+ done
+ echo "ERROR: attempt 1 never claimed within 30s"
+ curl -s "http://localhost:8000/jobs/${JOB_ID}" | jq '.'
+ exit 1
+ capture: attempt1_wait
+ timeout: 45
+
+ # The reclaim + second claim: lease lapses ~6s in, sweep (10s tick)
+ # resets the row, a replica claims it again minting epoch 2.
+ #
+ # FAIL-FAST at 30s: the stale owner wakes at ~t0+35. If attempt 2 has not
+ # claimed by then, the scenario SILENTLY CHANGES — the stale owner's
+ # post-wake write would land on an UNCLAIMED row instead of an
+ # epoch-mismatched live one, and a pass would no longer prove supersession
+ # fencing. Worst-case intended latency is ~21s (6s lease + 10s sweep tick
+ # + ~5s claim-poll backoff), so a 30s trip means the mechanism under test
+ # genuinely failed to engage — never wait past it.
+ - name: "Wait for the re-claim (attempt 2, epoch 2)"
+ handler: shell
+ workdir: /workspace
+ command: |
+ JOB_ID=$(echo '${captured.job_id_extract}' | grep '^JOB_ID=' | cut -d= -f2)
+ for i in $(seq 1 30); do
+ SEEN=$(curl -s "http://localhost:8000/jobs/${JOB_ID}/events?after=0&limit=100" \
+ | jq '[.events[] | select(.type == "transition") | select(.payload.marker == "claimed") | select(.payload.attempt == 2)] | length')
+ if [ "$SEEN" -ge 1 ]; then
+ echo "ATTEMPT2_CLAIMED=1 after ~${i}s"
+ curl -s "http://localhost:8000/jobs/${JOB_ID}" | jq '{status, attempt_count, owner_instance_id}'
+ exit 0
+ fi
+ sleep 1
+ done
+ echo "ERROR: no re-claim within 30s — the stale owner (waking at ~35s) would hit an UNCLAIMED row, not an epoch-mismatched one, so this run can no longer prove supersession fencing"
+ curl -s "http://localhost:8000/jobs/${JOB_ID}" | jq '.'
+ curl -s "http://localhost:8000/jobs/${JOB_ID}/events?after=0&limit=100" | jq '.'
+ exit 1
+ capture: attempt2_wait
+ timeout: 45
+
+ # Let the stale owner wake (t0+35) and slam into the fence: its stale-
+ # epoch progress delta is rejected, the cancel token fires, and its grace
+ # window (10s) elapses. 30s from here covers the worst case with margin.
+ # Scenario wait — not a registration/dep wait.
+ - name: "Wait out the stale owner's wake-up and fencing window"
+ handler: wait
+ seconds: 30
+
+ # The row must STILL be working (attempt 2 is gated on 'finish'): the
+ # stale owner's complete() must NOT have landed, its progress message
+ # must not be on the row, and no 'post_sleep' transition may exist.
+ - name: "Verify the stale owner was fenced on a live row"
+ handler: shell
+ workdir: /workspace
+ command: |
+ JOB_ID=$(echo '${captured.job_id_extract}' | grep '^JOB_ID=' | cut -d= -f2)
+ ROW=$(curl -s "http://localhost:8000/jobs/${JOB_ID}")
+ echo "$ROW" | jq '{status, attempt_count, owner_instance_id, progress_message}'
+ STATUS=$(echo "$ROW" | jq -r '.status')
+ if [ "$STATUS" != "working" ]; then
+ echo "FAIL: expected status=working while attempt 2 is gated, got ${STATUS} — a stale write landed"
+ exit 1
+ fi
+ if echo "$ROW" | jq -r '.progress_message // ""' | grep -q "post-sleep write"; then
+ echo "FAIL: the stale owner's progress message landed on the row"
+ exit 1
+ fi
+ POST_SLEEP=$(curl -s "http://localhost:8000/jobs/${JOB_ID}/events?after=0&limit=100" \
+ | jq '[.events[] | select(.type == "transition") | select(.payload.marker == "post_sleep")] | length')
+ if [ "$POST_SLEEP" != "0" ]; then
+ echo "FAIL: the stale owner posted its post_sleep transition — it was not aborted"
+ exit 1
+ fi
+ echo "STALE_OWNER_FENCED=1"
+ capture: fence_check
+ timeout: 30
+
+ # Release attempt 2's gate; the job must complete exactly once, from
+ # the epoch-2 execution.
+ - name: "Post finish and wait for completion"
+ handler: shell
+ workdir: /workspace
+ command: |
+ JOB_ID=$(echo '${captured.job_id_extract}' | grep '^JOB_ID=' | cut -d= -f2)
+ curl -s -X POST "http://localhost:8000/jobs/${JOB_ID}/events" \
+ -H 'Content-Type: application/json' \
+ -d '{"type":"finish","payload":{}}' | jq -c '.'
+ for i in $(seq 1 30); do
+ ST=$(curl -s "http://localhost:8000/jobs/${JOB_ID}" | jq -r '.status')
+ if [ "$ST" = "completed" ]; then
+ echo "COMPLETED=1 after ~${i}s"
+ exit 0
+ fi
+ sleep 1
+ done
+ echo "ERROR: job did not complete within 30s of finish"
+ curl -s "http://localhost:8000/jobs/${JOB_ID}" | jq '.'
+ exit 1
+ capture: completion_wait
+ timeout: 45
+
+ - name: "Capture final job row"
+ handler: shell
+ workdir: /workspace
+ command: |
+ JOB_ID=$(echo '${captured.job_id_extract}' | grep '^JOB_ID=' | cut -d= -f2)
+ curl -s "http://localhost:8000/jobs/${JOB_ID}"
+ capture: final_job
+
+ - name: "Capture final event log"
+ handler: shell
+ workdir: /workspace
+ command: |
+ JOB_ID=$(echo '${captured.job_id_extract}' | grep '^JOB_ID=' | cut -d= -f2)
+ curl -s "http://localhost:8000/jobs/${JOB_ID}/events?after=0&limit=100"
+ capture: final_events
+
+assertions:
+ # IDIOM: equality lives INSIDE the jq filter so the interpolation yields a
+ # bare true/false (tsuite Go-formats extracted values, and the
+ # interpolation body cannot contain '}').
+
+ # Scenario markers from the driver steps
+ - expr: "${captured.wait_registration} contains 'REPLICAS_READY=2'"
+ message: "both gated-worker-* replicas must register before the job is submitted"
+ - expr: "${captured.attempt1_wait} contains 'ATTEMPT1_CLAIMED=1'"
+ message: "attempt 1 must claim first"
+ - expr: "${captured.attempt2_wait} contains 'ATTEMPT2_CLAIMED=1'"
+ message: "the wedged job must be re-claimed — a second claim (epoch 2) must happen"
+ - expr: "${captured.fence_check} contains 'STALE_OWNER_FENCED=1'"
+ message: "the stale owner's post-wake writes must be rejected while the row is live"
+ - expr: "${captured.completion_wait} contains 'COMPLETED=1'"
+ message: "the re-claimed attempt must complete the job after finish"
+
+ # Two claims, epochs minted 1 then 2.
+ - expr: "${jq:captured.final_events:([.events[] | select(.type == \"transition\") | select(.payload.marker == \"claimed\")] | length) == 2} == 'true'"
+ message: "exactly two claims must execute the handler (attempts 1 and 2)"
+ - expr: "${jq:captured.final_events:([.events[] | select(.type == \"transition\") | select(.payload.marker == \"claimed\") | .payload.epoch] | sort) == [1,2]} == 'true'"
+ message: "the claims must carry epochs 1 then 2 — the re-claim mints a fresh epoch"
+
+ # Stale owner fenced: its duplicate side effects never landed.
+ - expr: "${jq:captured.final_events:([.events[] | select(.type == \"transition\") | select(.payload.marker == \"post_sleep\")] | length) == 0} == 'true'"
+ message: "the superseded attempt must be aborted BEFORE its post-wake side effects — no post_sleep transition may exist"
+ - expr: "${captured.final_job} not contains 'post-sleep write from superseded attempt'"
+ message: "the stale owner's progress message must never land on the job row"
+
+ # Exactly one owner after the re-claim: finish_seen once, from epoch 2.
+ - expr: "${jq:captured.final_events:([.events[] | select(.type == \"transition\") | select(.payload.marker == \"finish_seen\")] | length) == 1} == 'true'"
+ message: "exactly ONE owner's transitions may appear after the re-claim"
+ - expr: "${jq:captured.final_events:([.events[] | select(.type == \"transition\") | select(.payload.marker == \"finish_seen\")] | all(.payload.epoch == 2))} == 'true'"
+ message: "the post-re-claim transitions must be stamped with claim epoch 2"
+
+ # Completed exactly once, by attempt 2 under epoch 2.
+ - expr: "${jq:captured.final_job:(.status == \"completed\") and (.attempt_count == 2)} == 'true'"
+ message: "the job must complete after exactly two claims"
+ - expr: "${jq:captured.final_job:(.result.attempt == 2) and (.result.epoch == 2)} == 'true'"
+ message: "the terminal result must come from attempt 2 / epoch 2 — not the superseded owner"
+
+post_run:
+ - routine: stop_all
+ - routine: global.cleanup_workspace