Skip to content

stream_events

Internal stream-control events shared by orchestration strategies and interfaces.

These sentinels are empty strings so they can pass through existing text stream paths without emitting user-visible content. Metadata attributes carry control instructions to consumers that understand MADA's internal stream protocol.

InternalError

Bases: str

Empty stream chunk that signals an internal terminal error condition.

Source code in src/mada/core/orchestration/stream_events.py 
42
43
44
45
46
47
48
49
50
class InternalError(str):
    """
    Empty stream chunk that signals an internal terminal error condition.
    """

    def __new__(cls, error_message: str):
        value = str.__new__(cls, "")
        setattr(value, ERROR_MESSAGE_ATTR, error_message)
        return value

InternalResponseReplacement

Bases: str

Empty stream chunk that replaces previously collected response text.

Source code in src/mada/core/orchestration/stream_events.py 
31
32
33
34
35
36
37
38
39
class InternalResponseReplacement(str):
    """
    Empty stream chunk that replaces previously collected response text.
    """

    def __new__(cls, replacement: str):
        value = str.__new__(cls, "")
        setattr(value, RESPONSE_REPLACEMENT_ATTR, replacement)
        return value

InternalToolCallSignal

Bases: str

Empty stream chunk that carries background-detach metadata.

Source code in src/mada/core/orchestration/stream_events.py 
20
21
22
23
24
25
26
27
28
class InternalToolCallSignal(str):
    """
    Empty stream chunk that carries background-detach metadata.
    """

    def __new__(cls, tool_call_name: str):
        value = str.__new__(cls, "")
        setattr(value, TOOL_CALL_NAME_ATTR, tool_call_name)
        return value

apply_text_control(chunks, chunk)

Apply replacement/error controls to a text accumulator.

Returns:

Type Description
bool

A pair of booleans: whether the chunk was a control chunk, and whether
bool

collection should stop because the control chunk was terminal.
Source code in src/mada/core/orchestration/stream_events.py 
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
def apply_text_control(chunks: list[str], chunk: Any) -> tuple[bool, bool]:
    """
    Apply replacement/error controls to a text accumulator.

    Returns:
        A pair of booleans: whether the chunk was a control chunk, and whether
        collection should stop because the control chunk was terminal.
    """
    replacement = response_replacement(chunk)
    if replacement is not None:
        chunks[:] = [replacement]
        return True, False

    message = error_message(chunk)
    if message is not None:
        chunks[:] = [message]
        return True, True

    return False, False

error_message(chunk)

Return terminal error text carried by a stream chunk, if present.

Source code in src/mada/core/orchestration/stream_events.py 
69
70
71
72
73
74
def error_message(chunk: Any) -> str | None:
    """
    Return terminal error text carried by a stream chunk, if present.
    """
    value = getattr(chunk, ERROR_MESSAGE_ATTR, None)
    return None if value is None else str(value)

response_replacement(chunk)

Return replacement text carried by a stream chunk, if present.

Source code in src/mada/core/orchestration/stream_events.py 
61
62
63
64
65
66
def response_replacement(chunk: Any) -> str | None:
    """
    Return replacement text carried by a stream chunk, if present.
    """
    value = getattr(chunk, RESPONSE_REPLACEMENT_ATTR, None)
    return None if value is None else str(value)

tool_call_name(chunk)

Return the internal tool-call name carried by a stream chunk, if present.

Source code in src/mada/core/orchestration/stream_events.py 
53
54
55
56
57
58
def tool_call_name(chunk: Any) -> str | None:
    """
    Return the internal tool-call name carried by a stream chunk, if present.
    """
    value = getattr(chunk, TOOL_CALL_NAME_ATTR, None)
    return None if value is None else str(value)