zylon-ai/private-gpt · error · ValueError

Stream with correlation_id {event.correlation_id} not found

Error message

Stream with correlation_id {event.correlation_id} not found

What it means

Raised by InMemoryStreamService.push_event_batch during its pre-flight validation pass: it iterates all events and verifies every event.correlation_id exists in the metadata map before writing anything. This makes the batch operation fail atomically — no partial writes — if any single event references an unknown stream.

Source

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

        async with self._lock:
            if correlation_id not in self._metadata:
                raise ValueError(
                    f"Stream with correlation_id {correlation_id} not found"
                )

            # Generate sequential message ID like Redis
            self._event_counters[correlation_id] += 1
            message_id = f"{int(datetime.now(UTC).timestamp() * 1000)}-{self._event_counters[correlation_id]}"

            self._events[correlation_id].append((message_id, event_data))
            for waiter in list(self._waiters.get(correlation_id, set())):
                waiter.set()
            return message_id

    async def push_event_batch(self, events: list[Event]) -> dict[str, str]:
        for event in events:
            if event.correlation_id not in self._metadata:
                raise ValueError(
                    f"Stream with correlation_id {event.correlation_id} not found"
                )
        grouped: dict[str, list[str]] = defaultdict(list)
        for event in events:
            grouped[event.correlation_id].append(event.event_data)

        result: dict[str, str] = {}
        for correlation_id, event_datas in grouped.items():
            last_id = None
            for event_data in event_datas:
                last_id = await self.push_event(correlation_id, event_data)
            if last_id:
                result[correlation_id] = last_id
        return result

    async def read_events(
        self,
        correlation_id: str,

View on GitHub (pinned to 4a030776a3)

Solutions

Pre-validate all correlation ids: missing = {e.correlation_id for e in events} - existing ids; create missing streams before the batch call
Split the batch per correlation_id and handle failures per stream so one bad id does not block the rest
Wrap the call in try/except ValueError, log the offending events, and drop/requeue only the invalid ones

Example fix

# before
await stream_service.push_event_batch(events)
# after
missing = {
    e.correlation_id
    for e in events
    if not await stream_service.stream_exists(e.correlation_id)
}
for cid in missing:
    await stream_service.create_stream(correlation_id=cid, stream_type="ingestion")
await stream_service.push_event_batch(events)

Defensive patterns

Strategy: validation

Validate before calling

existing = {
    e.correlation_id
    for e in events
    if await stream_service.stream_exists(e.correlation_id)
}
missing = {e.correlation_id for e in events} - existing
for cid in missing:
    await stream_service.create_stream(correlation_id=cid, stream_type="ingestion")
await stream_service.push_event_batch(events)

Try / catch

try:
    await stream_service.push_event_batch(events)
except ValueError as e:
    if "not found" not in str(e):
        raise
    # split per correlation_id and handle the bad id individually

Prevention

Remember push_event_batch is atomic — one bad id blocks the whole batch
Pre-validate all correlation ids when mixing streams from multiple jobs

When it happens

Trigger: Calling push_event_batch with a list where at least one Event has a correlation_id that was never created (or was deleted). The remaining events in the batch are not pushed.

Common situations: Batching events for multiple jobs where one job's stream was cleaned up mid-batch; mixed creation paths where one subsystem forgot to create its stream; reprocessing old buffered events after a restart.

Related errors

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