Shield semantics: aiotools (deferred-edge) vs. asyncbis (level cancellation) -- companion experiments for the Async-SIG post

00_README.md

Shield semantics: aiotools (deferred-edge) vs. asyncbis (level cancellation)

Companion experiments for the Async-SIG post comparing nested block-scoped shielding between aiotools' TaskScope and asyncbis' CancelScope/TaskGroup.

Scenario ported from aiotools' test_taskscope_shielded_nested_4 (timings scaled 1/10, wall-clock instead of VirtualClock).

Setup

Each script carries a PEP 723 inline script metadata block, so uv resolves the pinned dependencies (and CPython 3.12 itself) into a throwaway environment -- no venv setup needed:

$ uv run 01_nested_shield_equivalence.py
$ uv run 02_task_cancel_divergence.py
$ uv run 03_repeated_cancel.py

Add --no-project if you run these from inside another project's directory. Or set it up manually:

$ pip install "aiotools==2.2.4" typing-extensions \
      "git+https://github.com/agronholm/asyncbis@592cc0b"
$ python 01_nested_shield_equivalence.py

Tested with CPython 3.12 on Linux. Each script asserts its claims and prints OK: ... on success. Note that asyncbis is a single-commit research sketch; 01 and 03 reach into task._cancel_scope (captured before the task's first step) because there is no public API for a task's root scope yet. typing-extensions is listed explicitly because aiotools 2.2.4 imports it without declaring it as a runtime dependency.

What each script shows

#	Claim	Key observation
01	Block-level shield-nesting semantics are observably equivalent across the two models -- if the cancel is addressed to the task's root scope on asyncbis	Identical trace / completed set / cancelled(). Divergence: the unshielded sibling L1 dies at t=1.42s (aiotools: deferred delivery replayed at the outermost shield's exit) vs. t=0.03s (asyncbis: immediate tree-wide level propagation)
02	Task.cancel() -- the entry point existing asyncio code uses -- inverts the outcome on asyncbis	It targets the innermost active scope; the shield is pierced (shields only block parent-to-child propagation), the CancelledError is absorbed at that block's boundary, and the task completes successfully (cancelled() == False). Under aiotools the identical call is deferred and the task ends cancelled
03	Repeated cancels: counted & preserved vs. idempotent / scope-peeling; neither is a force-cancel	aiotools: two latched requests replay on exit, cancelling() == 2 (attribution works). asyncbis: same-tick repeat is a no-op; spaced repeats peel one scope per call, each absorbed. Bonus (part D): after a genuine root-scope cancellation, task.cancelling() == 0 -- the 3.11 counting protocol is inert under the level model

Verified output

===== 01_nested_shield_equivalence.py =====
--- aiotools (task.cancel via cancel_and_wait) ---
  trace        = ['level0-begin', 'level1-begin', 'level2-begin', 'level3-begin', 'level4-begin', 'level5-begin', 'level5-end', 'level4-end', 'level3-end', 'level2-end']
  completed    = ['L2', 'L3', 'L4', 'L5']
  cancel_times = {'L1': 1.422}
  cancelled()  = True
--- asyncbis (root cancel scope) ---
  trace        = ['level0-begin', 'level1-begin', 'level2-begin', 'level3-begin', 'level4-begin', 'level5-begin', 'level5-end', 'level4-end', 'level3-end', 'level2-end']
  completed    = ['L2', 'L3', 'L4', 'L5']
  cancel_times = {'L1': 0.026}
  cancelled()  = True

OK: identical trace/results/cancelled; L1 cancelled at 1.422s (aiotools) vs 0.026s (asyncbis)

===== 02_task_cancel_divergence.py =====
trace        = ['level0-begin', 'level1-begin', 'level2-begin', 'level3-begin', 'level4-begin', 'level5-begin', 'level4-end', 'level3-end', 'level2-end', 'level1-end', 'level0-end']
completed    = ['L1', 'L2', 'L3', 'L4']
cancel_times = {'L5': 0.026}
cancelled()  = False

OK: task.cancel() pierced the innermost shield, was absorbed at its boundary, and the task completed successfully (cancelled() == False)

===== 03_repeated_cancel.py =====
[A: aiotools, double cancel while shielded]
   trace=['outer-begin', 'inner-begin', 'inner-end']  cancelled=True  cancelling()=2
[B: asyncbis, same-tick double cancel]
   trace=['outer-begin', 'inner-begin', 'outer-end', 'body-end']  cancelled=False
[C: asyncbis, spaced double cancel]
   trace=['outer-begin', 'inner-begin', 'body-end']  cancelled=False
[D: asyncbis, root-scope cancel]
   cancelled=True  cancelling()=0

OK: aiotools counts & defers repeats (never force); asyncbis is idempotent per scope / peels one scope per call, and cancelling() stays 0 even for a genuinely cancelled task

Timing-based assertions use generous margins (> 1.0s / < 0.2s) and should be robust on any non-pathological machine, but they are wall-clock based.

01_nested_shield_equivalence.py

# /// script
# requires-python = "==3.12.*"
# dependencies = [
#     "aiotools==2.2.4",
#     "asyncbis @ git+https://github.com/agronholm/asyncbis@592cc0b",
#     "typing-extensions",  # aiotools 2.2.4 imports it but does not declare it
# ]
# ///
"""Experiment 1: nested block-shield semantics are observably equivalent
between aiotools (deferred-edge) and asyncbis (level) -- when the cancellation
is addressed to the task's ROOT cancel scope on asyncbis.

Port of aiotools' test_taskscope_shielded_nested_4 (timings scaled 1/10):
https://github.com/achimnol/aiotools/blob/c05cebfe84404d4cdab3ed551da39f838f54158c/tests/test_taskscope.py#L548-L622

Nesting within a single task, shield flags [off, ON, off, off, ON]:

    ts1(shield=False)  spawns L1 (1.5s)
      ts2(shield=True)   spawns L2 (1.4s)
        ts3(shield=False)  spawns L3 (1.3s)
          ts4(shield=False)  spawns L4 (1.2s)
            ts5(shield=True)   spawns L5 (1.1s)
              <- external cancel arrives here (t ~= 0.025s)

Expected in BOTH frameworks: every block body and child task up to the
OUTERMOST shield (ts2) completes; CancelledError fires at ts2's boundary;
ts1's body and L1 are cancelled; the task ends cancelled.

Observed difference (asserted below): the unshielded sibling L1 is cancelled
  - aiotools: at t ~= 1.42s (deferred delivery replays at ts2's exit,
              then ts1 aborts its children)
  - asyncbis: at t ~= 0.03s (level propagation walks the scope tree
              immediately, skipping only the shielded ts2 subtree)

Tested with: Python 3.12, aiotools 2.2.4, asyncbis @ 592cc0b.
"""

import asyncio
import time

import asyncbis

from aiotools import TaskScope
from aiotools.cancel import cancel_and_wait

EXPECTED_TRACE = [
    "level0-begin",
    "level1-begin",
    "level2-begin",
    "level3-begin",
    "level4-begin",
    "level5-begin",
    "level5-end",
    "level4-end",
    "level3-end",
    "level2-end",
    # level1-end / level0-end absent: cancelled right after ts2's exit
]
EXPECTED_RESULTS = {"L2", "L3", "L4", "L5"}  # L1 must be cancelled


# --------------------------------------------------------------------------
# aiotools reference: cancellation is TASK-addressed (task.cancel() latched
# at the innermost shield, replayed upward at each shield's exit)
# --------------------------------------------------------------------------
async def aiotools_main() -> tuple[list[str], set[str], dict[str, float], bool]:
    t0 = time.perf_counter()
    trace: list[str] = []
    results: set[str] = set()
    cancel_times: dict[str, float] = {}

    async def work(delay: float, name: str) -> None:
        try:
            await asyncio.sleep(delay)
            results.add(name)
        except asyncio.CancelledError:
            cancel_times[name] = round(time.perf_counter() - t0, 3)
            raise

    async def nested_task() -> None:
        trace.append("level0-begin")
        await asyncio.sleep(0.01)
        async with TaskScope(shield=False) as ts1:
            trace.append("level1-begin")
            ts1.create_task(work(1.5, "L1"))
            await asyncio.sleep(0.01)
            async with TaskScope(shield=True) as ts2:
                trace.append("level2-begin")
                ts2.create_task(work(1.4, "L2"))
                async with TaskScope(shield=False) as ts3:
                    trace.append("level3-begin")
                    ts3.create_task(work(1.3, "L3"))
                    async with TaskScope(shield=False) as ts4:
                        trace.append("level4-begin")
                        ts4.create_task(work(1.2, "L4"))
                        async with TaskScope(shield=True) as ts5:
                            trace.append("level5-begin")
                            ts5.create_task(work(1.1, "L5"))
                            await asyncio.sleep(0.01)  # <- cancel arrives here
                            trace.append("level5-end")
                        await asyncio.sleep(0.01)
                        trace.append("level4-end")
                    await asyncio.sleep(0.01)
                    trace.append("level3-end")
                await asyncio.sleep(0.01)
                trace.append("level2-end")
            await asyncio.sleep(0.01)
            trace.append("level1-end")
        await asyncio.sleep(0.01)
        trace.append("level0-end")

    task = asyncio.create_task(nested_task())
    await asyncio.sleep(0.025)
    await cancel_and_wait(task)  # task-addressed: plain task.cancel() inside
    return trace, results, cancel_times, task.cancelled()


# --------------------------------------------------------------------------
# asyncbis port: cancellation must be SCOPE-addressed (the task's root scope).
# TaskScope(shield=X) maps to TaskGroup() + cancel_scope.shield = X.
# NOTE: asyncbis (single-commit sketch) has no public API for a task's root
# scope; we capture task._cancel_scope right after create_task(), before the
# task's first step, while it still points at the root scope.
# --------------------------------------------------------------------------
async def asyncbis_main() -> tuple[list[str], set[str], dict[str, float], bool]:
    t0 = time.perf_counter()
    trace: list[str] = []
    results: set[str] = set()
    cancel_times: dict[str, float] = {}

    async def work(delay: float, name: str) -> None:
        try:
            await asyncbis.sleep(delay)
            results.add(name)
        except asyncbis.CancelledError:
            cancel_times[name] = round(time.perf_counter() - t0, 3)
            raise

    async def nested_task() -> None:
        trace.append("level0-begin")
        await asyncbis.sleep(0.01)
        async with asyncbis.TaskGroup() as ts1:  # shield=False
            trace.append("level1-begin")
            ts1.create_task(work(1.5, "L1"))
            await asyncbis.sleep(0.01)
            async with asyncbis.TaskGroup() as ts2:
                ts2.cancel_scope.shield = True  # shield=True
                trace.append("level2-begin")
                ts2.create_task(work(1.4, "L2"))
                async with asyncbis.TaskGroup() as ts3:  # shield=False
                    trace.append("level3-begin")
                    ts3.create_task(work(1.3, "L3"))
                    async with asyncbis.TaskGroup() as ts4:  # shield=False
                        trace.append("level4-begin")
                        ts4.create_task(work(1.2, "L4"))
                        async with asyncbis.TaskGroup() as ts5:
                            ts5.cancel_scope.shield = True  # shield=True
                            trace.append("level5-begin")
                            ts5.create_task(work(1.1, "L5"))
                            await asyncbis.sleep(0.01)  # <- cancel arrives here
                            trace.append("level5-end")
                        await asyncbis.sleep(0.01)
                        trace.append("level4-end")
                    await asyncbis.sleep(0.01)
                    trace.append("level3-end")
                await asyncbis.sleep(0.01)
                trace.append("level2-end")
            await asyncbis.sleep(0.01)
            trace.append("level1-end")
        await asyncbis.sleep(0.01)
        trace.append("level0-end")

    async with asyncbis.TaskGroup() as tg:
        task = tg.create_task(nested_task())
        root_scope = task._cancel_scope  # root scope: captured pre-first-step
        await asyncbis.sleep(0.025)
        root_scope.cancel("external")  # scope-addressed cancellation
        # group __aexit__ waits for the task to finish unwinding
    return trace, results, cancel_times, task.cancelled()


def report(label: str, trace, results, cancel_times, cancelled) -> None:
    print(f"--- {label} ---")
    print(f"  trace        = {trace}")
    print(f"  completed    = {sorted(results)}")
    print(f"  cancel_times = {cancel_times}")
    print(f"  cancelled()  = {cancelled}")


if __name__ == "__main__":
    a_trace, a_results, a_ct, a_cancelled = asyncio.run(aiotools_main())
    report(
        "aiotools (task.cancel via cancel_and_wait)",
        a_trace,
        a_results,
        a_ct,
        a_cancelled,
    )

    b_trace, b_results, b_ct, b_cancelled = asyncbis.run(asyncbis_main())
    report("asyncbis (root cancel scope)", b_trace, b_results, b_ct, b_cancelled)

    # Equivalence of the block-level protection semantics
    assert a_trace == EXPECTED_TRACE, a_trace
    assert b_trace == EXPECTED_TRACE, b_trace
    assert a_results == EXPECTED_RESULTS, a_results
    assert b_results == EXPECTED_RESULTS, b_results
    assert a_cancelled and b_cancelled

    # The observable divergence: WHEN the unshielded sibling L1 dies
    assert a_ct["L1"] > 1.0, f"aiotools L1 died early: {a_ct}"  # ~1.42s
    assert b_ct["L1"] < 0.2, f"asyncbis L1 died late: {b_ct}"  # ~0.03s

    print()
    print(
        "OK: identical trace/results/cancelled;"
        f" L1 cancelled at {a_ct['L1']}s (aiotools) vs {b_ct['L1']}s (asyncbis)"
    )

02_task_cancel_divergence.py

# /// script
# requires-python = "==3.12.*"
# dependencies = [
#     "asyncbis @ git+https://github.com/agronholm/asyncbis@592cc0b",
# ]
# ///
"""Experiment 2: the same scenario as 01, but using Task.cancel() -- the
entry point existing asyncio code actually uses -- as the cancellation
trigger on asyncbis.

On asyncbis, AsyncbisTask.cancel() delegates to the task's CURRENT
(innermost active) cancel scope, and a shield only blocks parent->child
propagation, not a direct cancel() on the scope itself (Trio semantics).
Consequently, with the task sleeping inside the innermost SHIELDED scope
(ts5), an external task.cancel():

  1. pierces ts5's shield and kills its body immediately
     ("level5-end" is never appended; only L5 is cancelled, at t ~= 0.03s),
  2. is then ABSORBED at ts5's boundary (parent scope not cancelled
     -> the CancelledError is swallowed on scope exit),
  3. after which the task resumes and RUNS TO COMPLETION:
     levels 4..0 all finish, L1..L4 all complete, task.cancelled() is False.

Under aiotools the identical call defers to the outermost shield's exit and
the task ends cancelled (see 01). Same API call, inverted outcome -- this is
the task-addressed vs. scope-addressed divergence.

Tested with: Python 3.12, asyncbis @ 592cc0b.
"""

import time

import asyncbis

EXPECTED_TRACE = [
    "level0-begin",
    "level1-begin",
    "level2-begin",
    "level3-begin",
    "level4-begin",
    "level5-begin",
    # "level5-end" MISSING: ts5 was killed despite shield=True
    "level4-end",
    "level3-end",
    "level2-end",
    "level1-end",
    "level0-end",
]
EXPECTED_RESULTS = {"L1", "L2", "L3", "L4"}  # everything except L5 completes


async def main() -> None:
    t0 = time.perf_counter()
    trace: list[str] = []
    results: set[str] = set()
    cancel_times: dict[str, float] = {}

    async def work(delay: float, name: str) -> None:
        try:
            await asyncbis.sleep(delay)
            results.add(name)
        except asyncbis.CancelledError:
            cancel_times[name] = round(time.perf_counter() - t0, 3)
            raise

    async def nested_task() -> None:
        trace.append("level0-begin")
        await asyncbis.sleep(0.01)
        async with asyncbis.TaskGroup() as ts1:
            trace.append("level1-begin")
            ts1.create_task(work(1.5, "L1"))
            await asyncbis.sleep(0.01)
            async with asyncbis.TaskGroup() as ts2:
                ts2.cancel_scope.shield = True
                trace.append("level2-begin")
                ts2.create_task(work(1.4, "L2"))
                async with asyncbis.TaskGroup() as ts3:
                    trace.append("level3-begin")
                    ts3.create_task(work(1.3, "L3"))
                    async with asyncbis.TaskGroup() as ts4:
                        trace.append("level4-begin")
                        ts4.create_task(work(1.2, "L4"))
                        async with asyncbis.TaskGroup() as ts5:
                            ts5.cancel_scope.shield = True
                            trace.append("level5-begin")
                            ts5.create_task(work(1.1, "L5"))
                            await asyncbis.sleep(0.01)  # <- cancel arrives here
                            trace.append("level5-end")
                        await asyncbis.sleep(0.01)
                        trace.append("level4-end")
                    await asyncbis.sleep(0.01)
                    trace.append("level3-end")
                await asyncbis.sleep(0.01)
                trace.append("level2-end")
            await asyncbis.sleep(0.01)
            trace.append("level1-end")
        await asyncbis.sleep(0.01)
        trace.append("level0-end")

    async with asyncbis.TaskGroup() as tg:
        task = tg.create_task(nested_task())
        await asyncbis.sleep(0.025)
        task.cancel("external")  # TASK-addressed, as existing asyncio code does

    print(f"trace        = {trace}")
    print(f"completed    = {sorted(results)}")
    print(f"cancel_times = {cancel_times}")
    print(f"cancelled()  = {task.cancelled()}")

    assert trace == EXPECTED_TRACE, trace
    assert results == EXPECTED_RESULTS, results
    assert set(cancel_times) == {"L5"} and cancel_times["L5"] < 0.2, cancel_times
    assert task.cancelled() is False

    print()
    print(
        "OK: task.cancel() pierced the innermost shield, was absorbed at its"
        " boundary, and the task completed successfully (cancelled() == False)"
    )


if __name__ == "__main__":
    asyncbis.run(main())

03_repeated_cancel.py

# /// script
# requires-python = "==3.12.*"
# dependencies = [
#     "aiotools==2.2.4",
#     "asyncbis @ git+https://github.com/agronholm/asyncbis@592cc0b",
#     "typing-extensions",  # aiotools 2.2.4 imports it but does not declare it
# ]
# ///
"""Experiment 3: repeated Task.cancel() semantics, and the fate of the
3.11 cancellation-counting protocol.

Body used throughout (single task, sync-block nesting):

    outer scope (shield=False)
      "outer-begin"
      inner scope (shield=True)
        "inner-begin"; sleep; "inner-end"
      sleep
      "outer-end"
    "body-end"

Part A -- aiotools, task.cancel() twice while inside the SHIELDED inner scope:
  both requests are LATCHED and replayed at the shield's exit;
  the shield holds ("inner-end" appended), the task ends cancelled, and
  task.cancelling() == 2, which is what lets cancel_and_wait() attribute a
  concurrent external cancel. A second cancel is NOT a force-cancel.

Part B -- asyncbis, task.cancel() twice in the same tick:
  the first kills the inner scope directly (shield pierced, "inner-end"
  missing) and is absorbed at its boundary; the second is an idempotent
  no-op on the already-cancelled scope. Task completes (cancelled() False).

Part C -- asyncbis, task.cancel() twice, spaced so the second lands in the
  outer body: each call peels exactly ONE scope level (whatever is innermost
  at that moment) and is absorbed at that scope's boundary. Both "inner-end"
  and "outer-end" are missing, yet "body-end" runs and the task completes.
  Not a force-cancel either -- an emergent peeling semantic.

Part D -- asyncbis, actual cancellation via the root scope:
  even after the task genuinely ends cancelled, task.cancelling() == 0.
  The 3.11 counting protocol (cancelling()/uncancel()) is inert; cancellation
  attribution moves entirely to scope identity.

Tested with: Python 3.12, aiotools 2.2.4, asyncbis @ 592cc0b.
"""

import asyncio

import asyncbis

from aiotools import TaskScope


# --------------------------------------------------------------------------
# Part A: aiotools -- repeated cancels are counted and preserved
# --------------------------------------------------------------------------
async def part_a() -> None:
    trace: list[str] = []

    async def body() -> None:
        async with TaskScope(shield=False):
            trace.append("outer-begin")
            async with TaskScope(shield=True):
                trace.append("inner-begin")
                await asyncio.sleep(0.1)
                trace.append("inner-end")
            await asyncio.sleep(0.1)
            trace.append("outer-end")
        trace.append("body-end")

    task = asyncio.create_task(body())
    await asyncio.sleep(0.05)  # task is inside the SHIELDED inner scope
    task.cancel("first")
    task.cancel("second")
    try:
        await task
    except asyncio.CancelledError:
        pass

    print("[A: aiotools, double cancel while shielded]")
    print(
        f"   trace={trace}  cancelled={task.cancelled()}  cancelling()={task.cancelling()}"
    )
    assert trace == ["outer-begin", "inner-begin", "inner-end"], trace
    assert task.cancelled() is True
    assert task.cancelling() == 2  # both requests preserved -> attribution works


# --------------------------------------------------------------------------
# asyncbis body shared by parts B and C
# --------------------------------------------------------------------------
def make_asyncbis_body(trace: list[str], outer_sleep: float):
    async def body() -> None:
        async with asyncbis.TaskGroup() as s1:  # outer, unshielded
            trace.append("outer-begin")
            async with asyncbis.TaskGroup() as s2:
                s2.cancel_scope.shield = True  # inner, SHIELDED
                trace.append("inner-begin")
                await asyncbis.sleep(0.1)
                trace.append("inner-end")
            await asyncbis.sleep(outer_sleep)
            trace.append("outer-end")
        trace.append("body-end")

    return body


# --------------------------------------------------------------------------
# Part B: asyncbis -- same-tick double cancel: pierce + absorb, then no-op
# --------------------------------------------------------------------------
async def part_b() -> None:
    trace: list[str] = []
    async with asyncbis.TaskGroup() as tg:
        task = tg.create_task(make_asyncbis_body(trace, outer_sleep=0.1)())
        await asyncbis.sleep(0.05)  # inside the SHIELDED inner scope
        task.cancel("first")  # kills inner scope (shield pierced), absorbed
        task.cancel("second")  # idempotent no-op on the same cancelled scope

    print("[B: asyncbis, same-tick double cancel]")
    print(f"   trace={trace}  cancelled={task.cancelled()}")
    assert trace == ["outer-begin", "inner-begin", "outer-end", "body-end"], trace
    assert task.cancelled() is False


# --------------------------------------------------------------------------
# Part C: asyncbis -- spaced double cancel: one scope peeled per call
# --------------------------------------------------------------------------
async def part_c() -> None:
    trace: list[str] = []
    async with asyncbis.TaskGroup() as tg:
        task = tg.create_task(make_asyncbis_body(trace, outer_sleep=0.3)())
        await asyncbis.sleep(0.05)
        task.cancel("first")  # peels the inner (shielded!) scope
        await asyncbis.sleep(0.15)  # task is now in the outer body sleep
        task.cancel("second")  # peels the outer scope

    print("[C: asyncbis, spaced double cancel]")
    print(f"   trace={trace}  cancelled={task.cancelled()}")
    assert trace == ["outer-begin", "inner-begin", "body-end"], trace
    assert task.cancelled() is False


# --------------------------------------------------------------------------
# Part D: asyncbis -- counting protocol is inert even for real cancellation
# --------------------------------------------------------------------------
async def part_d() -> None:
    async def body() -> None:
        await asyncbis.sleep(0.2)

    async with asyncbis.TaskGroup() as tg:
        task = tg.create_task(body())
        root_scope = task._cancel_scope  # pre-first-step: the root scope
        await asyncbis.sleep(0.05)
        root_scope.cancel("external")

    print("[D: asyncbis, root-scope cancel]")
    print(f"   cancelled={task.cancelled()}  cancelling()={task.cancelling()}")
    assert task.cancelled() is True
    assert task.cancelling() == 0  # 3.11 counting protocol never fed


if __name__ == "__main__":
    asyncio.run(part_a())
    asyncbis.run(part_b())
    asyncbis.run(part_c())
    asyncbis.run(part_d())
    print()
    print(
        "OK: aiotools counts & defers repeats (never force);"
        " asyncbis is idempotent per scope / peels one scope per call,"
        " and cancelling() stays 0 even for a genuinely cancelled task"
    )