gentletask 0.1.0

Stoppable task hierarchies and semantic context (the throughline) for Python concurrency.

gentletask

Stoppable task hierarchies and a semantic-context "throughline" for Python concurrency.

Python's concurrency primitives are capable but cumbersome. concurrent.futures gives you threads and results; it does not give you a way to stop a hierarchy of running tasks, propagate that stop signal inward, or know — when something goes wrong three threads deep — what your program was actually trying to do when it failed. gentletask fills those gaps without adding complexity that isn't earned. A gentle task doesn't spin up a thread it doesn't need; when you ask it to stop, it stops, asks its children to stop too, and waits for them. And it carries a narrative of meaning across every boundary it crosses, so your logs tell a story instead of a list of disconnected events.

Two problems, two primitives:

• Stopability. A Task can be stopped cooperatively. Stop signals propagate down through hierarchies. Replacement blocking primitives (sleep, Queue, Event, poll) respect the stop signal, so you don't have to instrument every wait site by hand.
• Semantic context. A SemanticStack carries labeled context — what your program is doing, not just where it is — across thread (and, when values are pickleable, process) boundaries. The module singleton is called throughline: a continuous thread of meaning running through all your concurrent execution.

---

Installation

pip install gentletask

gentletask is pure Python, requires Python >= 3.10, and has no runtime dependencies.

---

Core concepts

Task hierarchies and cooperative stopping

A Task is anything that satisfies the Task protocol — a unit of work that can report whether it is is_done / is_stopped, be wait()-ed on, and be asked to stop(). The two built-in implementations are ThreadTask (runs a callable in its own daemon thread) and WorkTask (one job queued to a long-lived WorkerThread).

Tasks form a hierarchy automatically: when a task is created or waited on from inside another running task, it registers as a child of that task. Calling stop() on a parent cascades the stop to every child, and when a task finishes (normally or by exception) it stops any still-running children — unless they were explicitly detach()-ed.

Stopping is cooperative: a task is never interrupted mid-statement. Instead, the stop-aware blocking primitives (sleep, check_stop, Queue.get, Event.wait, poll, and Task.wait) notice the stop at their wait site and raise Stopped, which unwinds normally so your finally blocks run.

Stop propagation is poll-free: stop() pushes a notification rather than relying on a polling loop. Before a primitive parks, it registers a zero-arg callback via add_stop_callback; stop() fires those callbacks (exactly once, on the first stop) to wake the waiter the instant it is requested. An idle wait consumes no CPU, and stop latency is independent of any polling interval.

The throughline / semantic stack

A SemanticStack is a context-local stack of frames, where each frame is an ordered collection of key/value pairs. It is backed by a single ContextVar, and each frame is stored as an immutable tuple[tuple[str, Any], ...], so no defensive copying is needed and returned dicts are always fresh.

The library keeps two SemanticStack singletons, each with one declared job:

• throughline — the public, human-readable narrative. It requires a name on every frame and, by convention, carries nothing else. task_chain() and ThroughlineNameFilter read this, and it is what shows up in your logs.
• _task_stack — private to the module. It requires a task on every frame and exists only to track which Task is running; current_task() reads it. Keeping the task object off the throughline keeps the narrative clean and name-only.

Task code enters both stacks at once through the task_context(task, name) helper. Context travels across thread boundaries automatically: ThreadTask copies the calling context with contextvars.copy_context(), while WorkTask snapshots both stacks at submit() time and restores them when the worker picks up the job — so a job inherits the context of the code that caused it to be submitted, not the worker's.

asynch and synch

These are inverses that let a single call site choose how a function runs.

• asynch(fn, ...) returns a launcher: calling the launcher starts fn in a new ThreadTask and returns the task immediately.
• synch(fn) returns a synchronous version that always yields a concrete value. It flattens two layers of asynchrony: if fn was produced by asynch(), it is de-wrapped to the original callable so the work runs inline (no extra thread); and if the call returns something implementing the Task protocol, synch waits for that task and returns its result instead of the task. A plain function returning a plain value is simply called — so synch is safe to apply either way.

---

Quick start

import time
from gentletask import ThreadTask, sleep, Stopped

def counter(progress):
    n = 0
    while True:
        n += 1
        progress.append(n)
        sleep(0.02)  # stop-aware: raises Stopped when the task is stopped

progress = []
task = ThreadTask(counter, args=(progress,), name="counter")

time.sleep(0.1)   # let it tick a few times
task.stop()       # request a cooperative stop

try:
    task.wait()
except Stopped:
    print("counter stopped after", len(progress), "ticks")

print("is_stopped:", task.is_stopped)

---

Guide

SemanticStack basics

Each with block pushes a frame; leaving the block pops it. get(key) returns the innermost value, collect(key) returns all values outermost-first (skipping frames without the key), and frames() hands back fresh dicts.

from gentletask import SemanticStack

ctx = SemanticStack()
with ctx(operation="abc123", user="alice"):
    with ctx(operation="resize"):
        print(ctx.get("operation"))      # "resize" (innermost)
        print(ctx.get("user"))           # "alice" (from the outer frame)
        print(ctx.collect("operation"))  # ('abc123', 'resize')
        for f in ctx.frames():
            print(f)
        # {'operation': 'abc123', 'user': 'alice'}
        # {'operation': 'resize'}

A stack can require certain keys on every frame. required is a floor, not a schema — extra keys are always welcome.

labeled = SemanticStack(required=("name",))

with labeled(name="ok", extra="also fine"):   # extra keys welcome
    ...

labeled(extra="but no name")                  # raises ValueError: missing "name"

Snapshots

snapshot() captures the current frames; the returned SemanticSnapshot's restore() is a context manager that reinstalls them for a block. This is the machinery that carries context onto worker threads.

from gentletask import throughline

with throughline(name="request_handler", request_id="r-42"):
    snap = throughline.snapshot()

print(throughline.collect("name"))   # () — back outside, the stack is empty

with snap.restore():
    print(throughline.collect("name"), throughline.get("request_id"))
    # ('request_handler',) r-42

ThreadTask

ThreadTask runs a callable in a new daemon thread. wait() blocks and returns the result (or re-raises the worker's exception); result is shorthand for wait().

from gentletask import ThreadTask

t = ThreadTask(lambda: 6 * 7)
print(t.wait())     # 42
print(t.is_done)    # True

# Exceptions surface through wait()
def boom():
    raise ValueError("kaboom")

failed = ThreadTask(boom)
try:
    failed.wait()
except ValueError as e:
    print("caught:", e)

It accepts positional/keyword args, an explicit name, and an on_finish callback invoked with (result, exception) when the task finishes.

log_lines = []
t = ThreadTask(
    lambda x, y: x + y,
    args=(3,),
    kwargs={"y": 4},
    name="adder",
    on_finish=lambda result, exc: log_lines.append(f"adder finished -> {result}"),
)
print(t.wait())     # 7
print(log_lines)    # ['adder finished -> 7']

asynch and synch

from gentletask import asynch, synch, ThreadTask

add = asynch(lambda x, y: x + y, name="async-add")
task = add(10, 20)            # starts a ThreadTask
print(task.wait())           # 30

def process(x):
    return x * 10

job = asynch(process)                 # job(...) -> ThreadTask
print(synch(job)(5))                  # 50 — de-wrapped, runs inline, no thread
print(synch(process)(5))              # 50 — plain function passes through

def schedule(x):                      # a function that hands back a task
    return ThreadTask(lambda: process(x), name="scheduled")

print(synch(schedule)(5))             # 50 — synch waits for the returned task

Cooperative stop and stop-aware primitives

sleep, check_stop, poll, Queue.get, and Event.wait all check current_task() at their wait site and raise Stopped when a stop has been requested. Called outside any task, they behave like their stdlib equivalents.

from gentletask import sleep, check_stop, Queue, Event, poll

def worker(q: Queue, done: Event):
    while True:
        check_stop()          # surrender at a known-safe point (like sleep(0))
        item = q.get()        # raises Stopped if stopped while waiting
        process(item)
        if done.wait(0.01):   # raises Stopped if stopped while waiting
            break

# poll() samples a predicate on `interval`, but the inter-sample wait is
# poll-free with respect to stop:
poll(lambda: connection.ready, interval=0.05, timeout=5.0)

Stop cascades to children

Stopping a parent propagates to every child it started; when a task finishes it also stops any still-running children.

import time
from gentletask import ThreadTask, sleep, Stopped

child_ticks = []

def child():
    while True:
        child_ticks.append(1)
        sleep(0.02)

def parent():
    c = ThreadTask(child, name="child")
    c.wait()                  # parent blocks on the child

p = ThreadTask(parent, name="parent")
time.sleep(0.1)
p.stop()                      # cascades into the child
try:
    p.wait()
except Stopped:
    pass
print("child ticks frozen at", len(child_ticks))

detach() opts a child out of the cascade, so it keeps running after the parent stops:

def parent_detaching():
    c = ThreadTask(long_child, name="detached-child")
    c.detach()                # parent.stop() will NOT reach this child
    sleep(10)

WorkerThread and WorkTask — serialized jobs with inherited context

A WorkerThread is a long-lived thread that runs submitted jobs one at a time. submit() returns a WorkTask immediately and snapshots both stacks at submit time, so the job inherits the context of the code that caused it — not the worker's.

from gentletask import WorkerThread, throughline, task_chain

worker = WorkerThread(name="io-worker")
captured = {}

def job(label):
    captured[label] = task_chain()
    return label.upper()

# Submitted inside a frame -> the job inherits it
with throughline(name="request_handler"):
    a = worker.submit(job, ("a",), name="job-a")
    a.wait()

# Submitted outside any frame -> the job sees only its own name
b = worker.submit(job, ("b",), name="job-b")
b.wait()

print(captured["a"])   # ('request_handler', 'job-a')
print(captured["b"])   # ('job-b',)
print(a.result)        # 'A'
worker.stop()          # already-queued jobs drain, then the thread shuts down

After stop(), the worker drains any jobs already queued and then exits; further submit() calls raise RuntimeError.

Logging integration

ThroughlineNameFilter injects throughline.collect("name") onto every log record as record.throughline, giving each line its full task ancestry — across thread boundaries, with no manual plumbing.

import logging, sys
from gentletask import ThreadTask, ThroughlineNameFilter

logger = logging.getLogger("gentletask.demo")
logger.setLevel(logging.INFO)
handler = logging.StreamHandler(sys.stdout)
handler.setFormatter(logging.Formatter("%(throughline)s | %(message)s"))
handler.addFilter(ThroughlineNameFilter())
logger.addHandler(handler)
logger.propagate = False

def calibrate():
    logger.info("measuring offset")

def acquire():
    logger.info("starting acquisition")
    ThreadTask(calibrate, name="calibrate").wait()
    logger.info("acquisition complete")

ThreadTask(acquire, name="acquisition").wait()
# ('acquisition',)               | starting acquisition
# ('acquisition', 'calibrate')   | measuring offset
# ('acquisition',)               | acquisition complete

A custom Task

Task is a runtime_checkable structural Protocol — any object with the right shape qualifies, no base class required. Wrap your work in task_context(self, name) so current_task() and the log chain line up without touching either stack directly. For poll-free stop, implement add_stop_callback / remove_stop_callback and fire the registered callbacks exactly once from stop(): those are the hooks the blocking primitives use to wake without polling.

import threading
from gentletask import task_context, sleep, Task

class CountdownTask:
    """A minimal hand-rolled Task that counts down on a background thread."""

    def __init__(self, n):
        self._n = n
        self.is_stopped = False
        self.is_done = False
        self._result = None
        # Internal plumbing uses plain threading primitives, NOT gentletask's.
        # These coordinate the task's own lifecycle and are read from outside
        # the task (e.g. wait() is usually called by a parent), so they must
        # not raise Stopped based on the caller's current_task().
        self._done = threading.Event()  # completion signal
        self._lock = threading.Lock()  # guards the stop-callback list
        self._stop_callbacks = []
        self._thread = threading.Thread(target=self._run, daemon=True)
        self._thread.start()

    def _run(self):
        with task_context(self, "countdown"):
            try:
                while self._n > 0 and not self.is_stopped:
                    self._n -= 1
                    # The actual work waits with gentletask.sleep so a stop
                    # aborts it cooperatively (poll-free, raises Stopped).
                    sleep(0.01)
                self._result = "liftoff" if not self.is_stopped else "aborted"
            finally:
                self.is_done = True
                self._done.set()

    def wait(self, timeout=None):
        self._done.wait(timeout)
        return self._result

    @property
    def result(self):
        return self.wait()

    def stop(self):
        with self._lock:
            if self.is_stopped:
                return
            self.is_stopped = True
            callbacks, self._stop_callbacks = list(self._stop_callbacks), []
        for cb in callbacks:
            cb()

    def add_finish_callback(self, fn):
        self.wait()
        fn(self._result, None)

    def add_stop_callback(self, fn):
        with self._lock:
            if not self.is_stopped:
                self._stop_callbacks.append(fn)
                return
        fn()

    def remove_stop_callback(self, fn):
        with self._lock:
            if fn in self._stop_callbacks:
                self._stop_callbacks.remove(fn)

    def detach(self):
        pass

cd = CountdownTask(5)
print(isinstance(cd, Task))   # True
print(cd.wait())              # 'liftoff'

Note. runtime_checkable only verifies attribute presence, not full signatures — so an object can pass isinstance(obj, Task) while still missing the behavior the primitives rely on. Implement all of the protocol members, especially the stop-callback hooks, for a task that participates fully in poll-free stopping.

threading or gentletask? When to use which

The example above deliberately mixes the two, and the choice is not arbitrary. The rule of thumb:

• Use the gentletask primitives (sleep, Event, Queue, poll, check_stop) for the work a task performs — the wait points that should abort with Stopped when that task is stopped. These are stop-aware: they consult current_task() and raise Stopped so a cooperative stop unwinds the work cleanly. Reach for these whenever you are blocking inside a task and want the block to be interruptible.
• Use the plain threading primitives (threading.Event, Lock, Thread, queue.Queue) for a task's own lifecycle machinery — completion signals, internal locks, the worker thread itself. This plumbing is read and driven from outside the running task (a parent calls wait(), stop() may fire from any thread), so it must not be stop-aware: a gentletask.Event used for self._done would raise Stopped if wait() happened to be called from some unrelated stopped task, corrupting the task's own bookkeeping.

In short: gentletask primitives for interruptible work; threading primitives for the scaffolding that makes a task a task. The built-in ThreadTask and WorkTask follow exactly this split internally — their done/stop signaling is threading-based, while the work you hand them is free to use the stop-aware primitives.

---

API reference

Exceptions

Name	Description
Stopped	Raised inside a task when stop() has been requested. Unwinds normally; finally blocks run.

Tasks

Name	Description
Task	runtime_checkable structural Protocol for a stoppable, waitable unit of work.
ThreadTask(fn, args=(), kwargs=None, *, name=None, detach=False, on_finish=None)	Runs fn in a new daemon thread; implements Task.
WorkTask(fn, args, kwargs, name=None)	One job queued to a WorkerThread; implements Task. Usually created via WorkerThread.submit.
WorkerThread(name=None)	Long-lived worker thread that serializes submitted jobs. submit(...) returns a WorkTask; stop() drains queued jobs and shuts down.
asynch(fn, name=None, detach=False, on_finish=None)	Returns a launcher that starts fn in a new ThreadTask when called.
synch(fn)	Returns a synchronous version of fn that de-wraps asynch and awaits returned tasks, yielding a concrete value.

Task protocol members

Member	Description
is_done: bool	Whether the task has finished.
is_stopped: bool	Whether a stop has been requested.
result	Property; shorthand for wait().
wait(timeout=None)	Block until done, re-raising any worker exception. A stop on the calling parent propagates here and raises Stopped.
stop()	Request cooperative stop; fire stop callbacks once; cascade to children.
add_finish_callback(fn)	Call fn(result, exception) when the task finishes (immediately if already finished).
add_stop_callback(fn)	Call zero-arg fn() once when the task is stopped (immediately if already stopped).
remove_stop_callback(fn)	Unregister a stop callback; no-op if absent.
detach()	Remove this task from its parent's stop propagation. Parent-only: the caller must be the task whose children include this one (a task cannot detach itself); otherwise raises RuntimeError.

Context / throughline

Name	Description
SemanticStack(name="semantic_stack", *, required=())	Context-local stack of labeled frames backed by a ContextVar.
SemanticStack.__call__(**kwargs)	Context manager that pushes a frame for the block (raises ValueError if a required key is missing).
SemanticStack.get(key, default=None)	Value of key from the innermost frame that has it.
SemanticStack.collect(key)	Tuple of all values for key, outermost-first.
SemanticStack.walk(fn)	Apply fn to each frame dict, outermost-first; return a tuple of results.
SemanticStack.frames()	Full stack as fresh dicts, outermost-first.
SemanticStack.snapshot()	Capture current state as a SemanticSnapshot.
SemanticSnapshot.restore()	Context manager that installs the captured frames for the block.
throughline	Module singleton SemanticStack("throughline", required=("name",)) — the human-readable narrative.
task_context(task, name)	Context manager that enters name on the throughline and task on the private task stack.
current_task()	The innermost running Task, or None outside any task.
task_chain()	Tuple of task names, outermost-first (throughline.collect("name")).
ThroughlineNameFilter	logging.Filter that injects task_chain() onto each record as record.throughline.

Stop-aware blocking primitives

Name	Description
sleep(seconds)	Drop-in for time.sleep; raises Stopped if the current task is stopped (poll-free).
check_stop()	Raise Stopped if the current task is stopped; like sleep(0).
poll(fn, *, interval=0.05, timeout=None)	Sample fn until truthy; stop-aware inter-sample wait. Returns fn's truthy value, or the last falsy value on timeout.
Queue(maxsize=0)	Drop-in for queue.Queue; get() raises Stopped if the current task is stopped.
Event()	Drop-in for threading.Event; wait() raises Stopped if the current task is stopped.

---

Testing

The test suite runs under pytest:

pip install -e ".[dev]"
pytest

---

Contributing

Contributions are welcome. Please open an issue to discuss substantial changes first. Keep changes small and focused, match the surrounding code style, and add tests covering new behavior. The library is intentionally dependency-free and pure Python — please keep it that way.

---

License

MIT License. See the LICENSE file for details.

---

Authors

• Martin Chase
• Luke Campagnola

Copyright © Martin Chase and Luke Campagnola.