view asyncio_threads/stop_token/main.py @ 279:b3b547563ec7

Add Google connector service and agent wiki Implement the C/Seobeo Google Drive and Gmail connector with encrypted OAuth storage, Zenbu authentication, browser testing, AI tool discovery, chunked HTTP decoding, and Bazel coverage. Consolidate repository guidance into progressive wiki documentation and enforce arena-first allocation for new first-party C code. Co-authored-by: Copilot <[email protected]> Copilot-Session: 84c338fd-0939-4bb3-b7f3-1062eb213e5d
author MrJuneJune <me@mrjunejune.com>
date Mon, 17 Aug 2026 22:22:36 -0700
parents 46daba6e3cf4
children
line wrap: on
line source

from typing import Generator
from typing import List

# [Hello There, "I am <EN", "D> Something"]
#     |

def truncate_stream(text_stream: Generator[str, None, None], stop_token: str) -> Generator[str, None, None]:
    buffer = ""
    for chunk in text_stream:
        can_be_prefix = False
        buffer += chunk
        if stop_token not in buffer:
            for end in range(1, len(stop_token)):
                if stop_token[:end] in buffer[-1 * len(stop_token):]:
                    can_be_prefix = True
            if not can_be_prefix:
                yield chunk
                buffer = ""
        else: 
            pos = buffer.find(stop_token)
            yield buffer[:pos]
            return

def stream_to_list(stream: Generator[str, None, None]) -> List[str]:
    return [chunk for chunk in stream]

def list_to_stream(chunks: List[str]) -> Generator[str, None, None]:
    for chunk in chunks:
        yield chunk


print(
    stream_to_list(
        truncate_stream(list_to_stream(["Hello there ", "I'm doing great today.<END>", "Thanks for asking.", "How are you"]), "<END>")
    )
) 

print(
    stream_to_list(
        truncate_stream(list_to_stream(["Hello there ", "I'm doing great today.<E", "N", "D>", "Thanks for asking.", "How are you"]), "<END>")
    )
)