zylon-ai/private-gpt · error · ValueError

Stream with this message_id already exists

Error message

Stream with this message_id already exists

What it means

Raised by ChatAsyncService.initiate_chat_stream when the provided message_id already identifies an existing stream in the StreamManager. Message IDs are correlation keys for async streams, so duplicates are rejected with a plain ValueError before a new stream is created.

Source

Thrown at private_gpt/server/chat_async/chat_async_service.py:34

@singleton
class ChatAsyncService:
    @inject
    def __init__(
        self,
        stream_manager: StreamManager,
        chat_facade: ChatFacadeService,
    ):
        self.stream_manager = stream_manager
        self._chat_facade = chat_facade

    async def initiate_chat_stream(
        self, request: ChatRequest, message_id: str | None = None
    ) -> str:
        """Initiate a chat completion stream."""
        message_id = message_id or str(uuid4())
        if message_id and await self.stream_manager.stream_exists(message_id):
            raise ValueError("Stream with this message_id already exists")

        request = request.model_copy(
            update={
                "context": request.context.model_copy(
                    update={"correlation_id": message_id}
                )
            }
        )
        event_generator = await self._chat_facade.create_chat_event_generator(
            request=request
        )
        return await self.stream_manager.create_and_start_stream(
            event_handler=StreamingEventHandler(),
            stream_type="chat_completion",
            event_generator=event_generator,
            correlation_id=message_id,
            metadata={
                "message_count": len(request.messages),

View on GitHub (pinned to 4a030776a3)

Solutions

1. Let the server generate the id (omit message_id) unless you need idempotency.
2. On retry, first check stream_exists(message_id)/status endpoint; reuse or delete the existing stream before re-initiating.
3. Generate a fresh uuid4 per attempt if the previous stream was cancelled/deleted.

Example fix

# before
msg_id = 'fixed-client-id'
await chat_service.initiate_chat_stream(request, message_id=msg_id)  # second call raises

# after
if await chat_service.stream_manager.stream_exists(msg_id):
    await chat_service.stream_manager.clean_up_stream(msg_id)
await chat_service.initiate_chat_stream(request, message_id=msg_id)

if message_id and await stream_manager.stream_exists(message_id):
    raise RuntimeError('id in use — pick another or clean up first')
message_id = await chat_service.initiate_chat_stream(request, message_id=message_id)

try:
    await chat_service.initiate_chat_stream(request, message_id=msg_id)
except ValueError:
    meta = await client.get(f'/v1/messages/async/{msg_id}/status')
    if meta.status_code == 404:
        pass  # stale entry gone; safe to retry
    # else reuse the existing stream

When it happens

Trigger: Calling POST /v1/messages/async (or initiate_chat_stream) with an explicit message_id that is already active or retained; retrying a timed-out request with the same client-generated id while the first stream still exists.

Common situations: Idempotency-style retries after network timeouts where the first request actually succeeded; clients reusing uuids from persistence; testing with hardcoded message ids.

Related errors

• No items returned from astream_structured_predict
• Stream with correlation_id {correlation_id} already exists
• Invalid status: {value}. Valid options: {[s.value for s in c
• Stream with correlation_id {correlation_id} already exists
• Duplicate tool use ID found: {block.id}

AI-assisted analysis of zylon-ai/private-gpt@4a030776a3 (2026-08-15). Data as JSON: /api/errors/c9d0c127fdf032dc. Report an issue: GitHub.