{"record":{"id":"05eb7a372d12c70c","repo":"zylon-ai/private-gpt","slug":"stream-with-correlation-id-correlation-id-not-fo","errorCode":null,"errorMessage":"Stream with correlation_id {correlation_id} not found","messagePattern":"Stream with correlation_id (.+?) not found","errorType":"exception","errorClass":"ValueError","httpStatus":null,"severity":"error","filePath":"private_gpt/components/streaming/providers/in_memory_stream_service.py","lineNumber":73,"sourceCode":"                    f\"Stream with correlation_id {correlation_id} already exists\"\n                )\n            self._metadata[correlation_id] = stream_metadata\n            self._events[correlation_id] = []\n            self._event_counters[correlation_id] = 0\n\n        return correlation_id\n\n    async def update_stream_status(\n        self,\n        correlation_id: str,\n        status: StreamStatus,\n        error_message: str | None = None,\n        metadata: dict[str, Any] | None = None,\n    ) -> None:\n        \"\"\"Update stream status and metadata.\"\"\"\n        async with self._lock:\n            if correlation_id not in self._metadata:\n                raise ValueError(\n                    f\"Stream with correlation_id {correlation_id} not found\"\n                )\n\n            stream_meta = self._metadata[correlation_id]\n            stream_meta.status = status\n            stream_meta.updated_at = datetime.now(UTC)\n\n            if error_message:\n                stream_meta.error_message = error_message\n\n            if status in [\n                StreamStatus.COMPLETED,\n                StreamStatus.CANCELLED,\n                StreamStatus.ERROR,\n            ]:\n                stream_meta.completed_at = datetime.now(UTC)\n\n            if metadata:","sourceCodeStart":55,"sourceCodeEnd":91,"githubUrl":"https://github.com/zylon-ai/private-gpt/blob/4a030776a31a901ad80b1bf4d7faa2c1a367efbb/private_gpt/components/streaming/providers/in_memory_stream_service.py#L55-L91","documentation":"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.","triggerScenarios":"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).","commonSituations":"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.","solutions":["Verify with await stream_service.stream_exists(correlation_id) (or get_stream_metadata) before updating","If the stream was deleted, re-create it with create_stream before pushing status updates","For multi-process deployments switch to the Redis stream provider so state is shared"],"exampleFix":"# before\nawait stream_service.update_stream_status(cid, StreamStatus.COMPLETED)\n# after\nmeta = await stream_service.get_stream_metadata(cid)\nif meta is None:\n    await stream_service.create_stream(correlation_id=cid, stream_type=\"ingestion\")\nawait stream_service.update_stream_status(cid, StreamStatus.COMPLETED)","handlingStrategy":"validation","validationCode":"meta = await stream_service.get_stream_metadata(correlation_id)\nif meta is None:\n    await stream_service.create_stream(\n        correlation_id=correlation_id, stream_type=stream_type\n    )\nawait stream_service.update_stream_status(correlation_id, StreamStatus.COMPLETED)","typeGuard":null,"tryCatchPattern":"try:\n    await stream_service.update_stream_status(cid, status)\nexcept ValueError as e:\n    if \"not found\" not in str(e):\n        raise\n    logger.warning(\"stale stream %s; recreating\", cid)\n    await stream_service.create_stream(correlation_id=cid, stream_type=\"ingestion\")","preventionTips":["Do not share in-memory streams across processes/workers","Re-create streams after worker restarts before pushing status updates"],"tags":["streaming","in-memory","not-found","lifecycle"],"backgroundTag":null,"analyzedSha":"4a030776a31a901ad80b1bf4d7faa2c1a367efbb","analyzedAt":"2026-08-15T03:51:26.951Z","schemaVersion":2},"datasetVersion":"2026-08-15T22:17:37.221Z"}