# Task groups and cancellation

> Decide exactly once when a set of tasks is done (first accepted, N accepted, or all finished), stop handing out work in the same step, and cancel a whole run from one call.

Source: https://docs.getrunstate.com/guides/completion-and-cancellation/

Sometimes you need every answer, sometimes five good ones, sometimes the first acceptable result is enough. Once you have it, the rest of the swarm should stop picking up work. runstate gives you two tools:

- **Task groups** decide once, under races, whether a set of tasks is done, and close the group to new work in the same step.
- **Run cancellation** refuses new work for a run and everything under it, immediately, on the server.

Barriers and timers, lower-level building blocks for "wait until everyone arrives" and "wake me later", are at the end.

## Task groups

A task group owns its own run (the *gate*) and delivers its member tasks through a work queue. You submit members, workers complete them, and something in your system gives each successful member a verdict: **accept** or **reject**. The group finalizes as soon as its condition is decided.

| `condition` | Finalizes with `SUCCESS` when | Finalizes with `FAILURE` when |
| --- | --- | --- |
| `FIRST_ACCEPTED` | the first member is accepted (`first_accepted`) | its deadline passes (`deadline`) |
| `N_ACCEPTED` (with `threshold`) | `threshold` members are accepted (`threshold_reached`) | accepted members plus members that could still be accepted can no longer reach `threshold` (`threshold_unreachable`), or the deadline passes |
| `ALL_TERMINAL` | the group is sealed, every member has finished, at least one was accepted and none rejected (`all_terminal`) | the group is sealed and every member has finished otherwise (`all_terminal`), or the deadline passes |

The value in parentheses is the `outcomeReason`.

### Create a group and submit members

**TypeScript**

```ts
await rs.mailboxes.ensure('reviews', { mode: 'WORK' });
const run = await rs.scopes.create();

// Done when five drafts have been accepted.
const group = await run.groups.create({
  mailbox: 'reviews',
  condition: 'N_ACCEPTED',
  threshold: 5,
  parentScopeId: run.id, // cancelling the run also closes the group
});

const tickets = [];
for (let i = 0; i < 12; i++) {
  tickets.push(await group.submit({ draft: i }, { key: `draft-${i}` }));
}

// Workers must attach to the group's own run to receive its members.
const { scopeId: groupRunId } = await group.status();
console.log(`start workers with RUNSTATE_SCOPE_ID=${groupRunId}`);
```

**Python**

```python
await rs.mailboxes.ensure("reviews", mode="WORK")
run = await rs.scopes.create()

# Done when five drafts have been accepted.
group = await run.groups.create(
    mailbox="reviews",
    condition="N_ACCEPTED",
    threshold=5,
    parent_scope_id=run.id,  # cancelling the run also closes the group
)

tickets = []
for i in range(12):
    tickets.append(await group.submit({"draft": i}, key=f"draft-{i}"))

# Workers must attach to the group's own run to receive its members.
group_run_id = (await group.status())["scopeId"]
print(f"start workers with RUNSTATE_SCOPE_ID={group_run_id}")
```

- The queue named in `mailbox` must exist before you create the group (`NotFoundError` otherwise).
- A group holds at most 100 members. `group.joinTask(taskId)` / `join_task(task_id)` adds an existing task while the group is open.
- `deadline` (ISO 8601) finalizes the group with `FAILURE` if it hasn't decided by then.
- Workers are ordinary queue consumers (see [Claims, shared tasks and takeover](https://docs.getrunstate.com/guides/work-once/#workers-and-crash-takeover)) attached to the group's run id from `group.status()`.

> **Caution: Submit members right after creating an `N_ACCEPTED` group**
>
> Groups are re-evaluated in the background every few seconds, not only when you record a verdict. An `N_ACCEPTED` group whose accepted members plus not-yet-judged pending, running or succeeded members are fewer than `threshold` finalizes with `threshold_unreachable`, and that includes a group that doesn't have enough members yet. Submit at least `threshold` members immediately after `create()`. Members that end `FAILED`, `CANCELLED` or `EXPIRED` also stop counting toward the threshold.

### Record verdicts and wait for the decision

Accepting requires the member task to have `SUCCEEDED`; rejecting requires it to have finished. Both are idempotent.

**TypeScript**

```ts
// A reviewer process: judge each finished member.
for (const ticket of tickets) {
  const result = await ticket.result({ timeoutMs: 600_000 });
  if (result.state !== 'SUCCEEDED') continue;
  const verdict = isGoodDraft(result.outcome) ? await group.accept(ticket.id) : await group.reject(ticket.id);
  if (verdict.groupState === 'FINALIZED') break;
}

const decision = await group.wait({ timeoutMs: 600_000 });
console.log(decision.outcome, decision.outcomeReason); // e.g. SUCCESS threshold_reached
```

**Python**

```python
# A reviewer process: judge each finished member.
for ticket in tickets:
    result = await ticket.result(timeout_ms=600_000)
    if result["state"] != "SUCCEEDED":
        continue
    if is_good_draft(result["outcome"]):
        verdict = await group.accept(ticket.id)
    else:
        verdict = await group.reject(ticket.id)
    if verdict["groupState"] == "FINALIZED":
        break

decision = await group.wait(timeout_ms=600_000)
print(decision["outcome"], decision["outcomeReason"])  # e.g. SUCCESS threshold_reached
```

A process that restarts can re-attach with `run.groups.byId(groupId)` (Python `by_id`) and call `status()` or `wait()` again.

### What finalizing does

In the same transaction that records the decision, runstate **cancels the group's run**. From that moment:

- workers asking for more of the group's work are refused, and `consume()` loops on the group's run stop;
- members that haven't finished are cancelled;
- verdicts that arrive afterwards are recorded with `late: true` and never change the outcome.

Group `status()` shows every member with its task state, verdict and `late` flag.

## Cancel a run

Cancelling a run is how you hit stop on a swarm.

**TypeScript**

```ts
const run = await rs.scopes.create({
  deadline: new Date(Date.now() + 60 * 60 * 1000).toISOString(), // optional: stop accepting work after an hour
});
const shard = await run.child(); // a sub-run, cancelled along with its parent

// ... later, from any process that knows the run id:
await rs.scope(run.id).cancel();

const status = await shard.status();
console.log(status.storedState, status.effective); // ACTIVE INACTIVE
```

**Python**

```python
from datetime import datetime, timedelta, timezone

deadline = (datetime.now(timezone.utc) + timedelta(hours=1)).isoformat()
run = await rs.scopes.create(deadline=deadline)  # optional: stop accepting work after an hour
shard = await run.child()  # a sub-run, cancelled along with its parent

# ... later, from any process that knows the run id:
await rs.scope(run.id).cancel()

status = await shard.status()
print(status["storedState"], status["effective"])  # ACTIVE INACTIVE
```

After `cancel()`, for the run and every run below it:

- **New work is refused** with `SCOPE_CANCELLED` (`ScopeCancelledError`): claims, sends, submits, receives, admissions, pool acquisitions, quota takes and budget reservations.
- **Queued work is cancelled.** Tasks that haven't finished become `CANCELLED`; queued messages and armed timers are cancelled.
- **Running agents are asked to stop, not killed.** Lease renewals are refused too, so the SDK marks held claims, pool leases and deliveries as lost: `lease.signal` aborts in TypeScript, and in Python `run()` cancels its handler. `consume()` stops taking work.
- **Agents can still hand back what they hold.** Releasing claims and pool leases and retrying or rejecting deliveries remain allowed. Completing a delivery is refused.
- **Recorded results stay readable.** Cancelling never hides a task result that was already recorded.

`status()` returns `storedState` (the run's own state: `ACTIVE`, `CANCELLED` or `COMPLETED`) and `effective`, which is `INACTIVE` if the run or any ancestor is no longer active or past its deadline.

`complete()` closes a run the same way but records that it finished normally. Either is final: a run can't be reopened, and cancelling a completed run returns a conflict.

A run with a `deadline` refuses new work with `DEADLINE_EXCEEDED` once the deadline passes. `childLimit` caps how many child runs a run can have (default 1000). Two calls to `rs.scopes.create()` always create two runs.

## Barriers

A barrier releases everyone waiting on it once `target` distinct participants have arrived: a wait-for-all rendezvous, for example "start the merge step when all three shards are done".

**TypeScript**

```ts
// Coordinator:
const barrier = await run.barriers.create({ target: 3 });

// Each shard, given barrier.id:
const handle = rs.scope(run.id).barriers.get(barrier.id);
await handle.arrive({ key: 'shard-2' }); // idempotent per key
const view = await handle.wait({ timeoutMs: 300_000 });
console.log(view.state, `${view.arrivals}/${view.target}`);
```

**Python**

```python
# Coordinator:
barrier = await run.barriers.create(target=3)

# Each shard, given barrier.id:
handle = rs.scope(run.id).barriers.get(barrier.id)
await handle.arrive(key="shard-2")  # idempotent per key
view = await handle.wait(timeout_ms=300_000)
print(view["state"], f"{view['arrivals']}/{view['target']}")
```

`wait()` polls until the barrier is `RELEASED` (default timeout 60 s) and throws `ConflictError` with `BARRIER_CANCELLED` if the barrier is cancelled, for example by cancelling its run. Arriving after release returns `BARRIER_RELEASED`. The HTTP API can advance a released barrier to a new epoch for reuse (`POST /barriers/{id}/nextEpoch`); neither SDK wraps that call yet.

## Timers

A timer is a durable wakeup at a time (`at`, ISO 8601) or after a delay (`afterSeconds`). It fires on the server whether or not anyone is waiting.

**TypeScript**

```ts
const timer = await run.timers.create({ afterSeconds: 300 });
console.log(`fires at ${timer.fireAt.toISOString()}`);

const { firedAt } = await timer.wait({ timeoutMs: 360_000 });
console.log(`follow-up check at ${firedAt.toISOString()}`);
```

**Python**

```python
timer = await run.timers.create(after_seconds=300)
print(f"fires at {timer.fire_at.isoformat()}")

fired = await timer.wait(timeout_ms=360_000)
print(f"follow-up check at {fired['fired_at'].isoformat()}")
```

- `wait()` polls for the timer to fire and acknowledges it in the same call. Have only one process wait on a given timer.
- `timer.cancel()` disarms a timer that hasn't fired. Cancelling the run cancels its armed timers.
- `run.timers.get(timerId)` re-attaches to a timer by id after a restart.

## Related

- [Events and monitoring](https://docs.getrunstate.com/guides/observability/): watch `group.finalized` and `task.cancelled` events as they happen.
- API reference: [Task groups](https://docs.getrunstate.com/api/task-groups/), [Runs](https://docs.getrunstate.com/api/runs/), [Barriers](https://docs.getrunstate.com/api/barriers/), [Timers](https://docs.getrunstate.com/api/timers/).