zylon-ai/private-gpt · error · ValueError

Stream with correlation_id {correlation_id} already exists

Error message

Stream with correlation_id {correlation_id} already exists

What it means

Raised by InMemoryStreamService.create_stream when a stream with the same correlation_id is registered twice. The service keys metadata, events and counters by correlation_id, so duplicate creation would overwrite state; the check runs under the async lock to guarantee uniqueness. The offending correlation_id is included in the message.

Source

Thrown at private_gpt/components/streaming/providers/in_memory_stream_service.py:54

        metadata: dict[str, Any] | None = None,
    ) -> str:
        """Create a new stream and return correlation ID."""
        if correlation_id is None:
            correlation_id = str(uuid.uuid4())

        now = datetime.now(UTC)
        stream_metadata = StreamMetadata(
            correlation_id=correlation_id,
            status=StreamStatus.PENDING,
            created_at=now,
            updated_at=now,
            stream_type=stream_type,
            metadata=metadata or {},
        )

        async with self._lock:
            if correlation_id in self._metadata:
                raise ValueError(
                    f"Stream with correlation_id {correlation_id} already exists"
                )
            self._metadata[correlation_id] = stream_metadata
            self._events[correlation_id] = []
            self._event_counters[correlation_id] = 0

        return correlation_id

    async def update_stream_status(
        self,
        correlation_id: str,
        status: StreamStatus,
        error_message: str | None = None,
        metadata: dict[str, Any] | None = None,
    ) -> None:
        """Update stream status and metadata."""
        async with self._lock:
            if correlation_id not in self._metadata:

View on GitHub (pinned to 4a030776a3)

Solutions

1. Before creating, check with await stream_service.stream_exists(correlation_id) and reuse or delete the existing stream
2. Use a fresh correlation id (e.g. uuid4) for each new stream instead of reusing a business-level id
3. On retry paths, catch the ValueError and resume consuming the existing stream rather than re-creating it

Example fix

# before
await stream_service.create_stream(correlation_id=job_id, stream_type="ingestion")
# after
if await stream_service.stream_exists(job_id):
    await stream_service.delete_stream(job_id)
await stream_service.create_stream(correlation_id=job_id, stream_type="ingestion")

if await stream_service.stream_exists(correlation_id):
    await stream_service.delete_stream(correlation_id)
await stream_service.create_stream(correlation_id=correlation_id, stream_type=stream_type)

try:
    await stream_service.create_stream(correlation_id=cid, stream_type=st)
except ValueError as e:
    if "already exists" not in str(e):
        raise
    # reuse the existing stream

When it happens

Trigger: Calling create_stream(correlation_id=X) twice without delete_stream(X) in between; retrying an ingestion/job request that reuse the same client-supplied correlation id; concurrent workers registering the same id.

Common situations: At-least-once job queues redelivering a task that already created its stream; client retries after a timeout where the first create actually succeeded; idempotency keys reused across runs.

Related errors

• Stream with correlation_id {correlation_id} already exists
• Stream with correlation_id {correlation_id} not found
• Stream with correlation_id {event.correlation_id} not found
• Stream with this message_id already exists
• Handler must define function handle(input, context).

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