# 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.

A group is **open** to new members until you **close** it (`group.close()`), until `expectedMembers` members have joined, or until it finalizes. Closed groups show state `SEALED`. A result that depends on the full member list, such as "the threshold can no longer be reached" or "every member has finished", is only decided once the group is closed. While it is open, a group can still succeed, and it can fail only when its deadline passes or its run is cancelled.

| `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`), open or closed | the group is closed and accepted members plus members that could still be accepted can't reach `threshold` (`threshold_unreachable`), or the deadline passes |
| `ALL_TERMINAL` | the group is closed, every member has finished, at least one was accepted and none rejected (`all_terminal`) | the group is closed 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}` }));
}
// No more members: from now on the group can also fail as unreachable.
await group.close();

// 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}"))
# No more members: from now on the group can also fail as unreachable.
await group.close()

# 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()`.

### Close the group

`group.close()` stops new members from joining and evaluates the group right away, in the same step. It returns `{ state, outcome, outcomeReason }`: `state` is `SEALED` while the group is still undecided, or `FINALIZED` if closing decided it (for example an `N_ACCEPTED` group with too few members left that could be accepted). Calling `close()` again is safe; on a finalized group it returns the recorded outcome.

If you know the member count up front, pass `expectedMembers` / `expected_members` (1 to 100) to `create()` instead. The group closes itself when that many members have joined, so the submit that adds the last member also closes it. For `N_ACCEPTED`, `expectedMembers` can't be lower than `threshold`. After the group is closed, submitting a new member fails with `ConflictError`; resubmitting a member that already joined (same key and input) returns that member.

Until a group is closed, members that end `FAILED`, `CANCELLED` or `EXPIRED` don't fail it: you may still add members that succeed. After the group is closed they stop counting toward the threshold.

> **Note: Groups created before close() existed**
>
> An `N_ACCEPTED` group used to fail with `threshold_unreachable` as soon as its current members couldn't reach the threshold, even while members were still being added. Now it waits for `close()`, `expectedMembers` or its deadline. If you create `N_ACCEPTED` groups without a deadline, call `close()` after adding members, or the group can stay open indefinitely when too few members succeed. `seal()` still works and now behaves like `close()`, except that it throws `ConflictError` on a group that has already finalized.

### 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. Pass `name` to `create()` or `child()` to label a run; `status()` returns it, but runs are never looked up by name.

## 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. Creating a barrier is configuration and needs a key with `resource_config`; arriving and waiting need `coordination_write` and `coordination_read`.

## 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/).