zylon-ai/private-gpt · error · ValueError

Stream with correlation_id {correlation_id} not found

Error message

Stream with correlation_id {correlation_id} not found

What it means

Raised by InMemoryStreamService.update_stream_status when the given correlation_id has no entry in the metadata map. Status updates are only valid on streams previously created with create_stream. The lookup happens under the lock so the stream cannot vanish mid-update, but a wrong or expired id fails immediately.

Source

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

                    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:
                raise ValueError(
                    f"Stream with correlation_id {correlation_id} not found"
                )

            stream_meta = self._metadata[correlation_id]
            stream_meta.status = status
            stream_meta.updated_at = datetime.now(UTC)

            if error_message:
                stream_meta.error_message = error_message

            if status in [
                StreamStatus.COMPLETED,
                StreamStatus.CANCELLED,
                StreamStatus.ERROR,
            ]:
                stream_meta.completed_at = datetime.now(UTC)

            if metadata:

View on GitHub (pinned to 4a030776a3)

Solutions

1. Verify with await stream_service.stream_exists(correlation_id) (or get_stream_metadata) before updating
2. If the stream was deleted, re-create it with create_stream before pushing status updates
3. For multi-process deployments switch to the Redis stream provider so state is shared

Example fix

# before
await stream_service.update_stream_status(cid, StreamStatus.COMPLETED)
# after
meta = await stream_service.get_stream_metadata(cid)
if meta is None:
    await stream_service.create_stream(correlation_id=cid, stream_type="ingestion")
await stream_service.update_stream_status(cid, StreamStatus.COMPLETED)

meta = await stream_service.get_stream_metadata(correlation_id)
if meta is None:
    await stream_service.create_stream(
        correlation_id=correlation_id, stream_type=stream_type
    )
await stream_service.update_stream_status(correlation_id, StreamStatus.COMPLETED)

try:
    await stream_service.update_stream_status(cid, status)
except ValueError as e:
    if "not found" not in str(e):
        raise
    logger.warning("stale stream %s; recreating", cid)
    await stream_service.create_stream(correlation_id=cid, stream_type="ingestion")

When it happens

Trigger: Calling update_stream_status with a correlation_id never created, already deleted via delete_stream, or from a different process (the in-memory store is per-process).

Common situations: Worker restarts wiping in-memory state while callers still hold old stream ids; using the in-memory provider in a multi-process deployment (uvicorn workers) where the creator and updater are different processes; processing a stale queue message.

Related errors

• Stream with correlation_id {event.correlation_id} not found
• Stream with correlation_id {correlation_id} already exists
• Unsupported content block type: {type(block)}
• Unknown message role {self.role}. Expected 'system', 'user',
• Invalid system item: {item}

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