symfonic.capabilities.tools.execution.progress

progress

Progress — intermediate values from a still-running tool (T3.1.3).

A tool authored as an async generator yields several times and only the last yield is the result; the rest are progress. The drain rule is one-step lookahead: hold the newest value as the candidate result and emit the previous one as progress when a newer arrives. That is the only shape where "which yield was the answer?" needs no sentinel from the tool author.

Every emit is awaited before the next value is pulled, so a slow consumer applies backpressure instead of silently dropping frames.

NullProgressSink

The default sink: progress is built by nobody and goes nowhere.

ProgressSink

Bases: Protocol

Where a :class:ToolProgress goes once the executor builds one.

drain_with_progress async

drain_with_progress(values: AsyncIterator[Any], emit: Callable[[int, Any], Awaitable[None]]) -> Any

Drain values, emitting all but the last, and return the last.

• empty iterator -> None (what a plain tool returning None would surface, so the result round-trip stays consistent);
• one value -> returned directly, no progress emitted;
• N values -> N - 1 progress emissions numbered 0..N-2.

Source code in src/symfonic/capabilities/tools/execution/progress.py

async def drain_with_progress(
    values: AsyncIterator[Any],
    emit: Callable[[int, Any], Awaitable[None]],
) -> Any:
    """Drain ``values``, emitting all but the last, and return the last.

    * empty iterator -> ``None`` (what a plain tool returning ``None``
      would surface, so the result round-trip stays consistent);
    * one value -> returned directly, no progress emitted;
    * N values -> ``N - 1`` progress emissions numbered ``0..N-2``.
    """
    last: Any = _EMPTY
    sequence = 0
    async for value in values:
        if last is not _EMPTY:
            await emit(sequence, last)
            sequence += 1
        last = value
    if last is _EMPTY:
        return None
    return last