"""Redis-backed run records and per-run event streams. The store writes status-only run records as JSON strings and events as Redis streams. HTTP event cursors are Redis stream ids; ``0-0`` means replay from the beginning for polling and SSE. Records and streams share one retention window that is refreshed when status or event data is written. Execution is scheduled in-process by ``dify_agent.runtime.run_scheduler``; Redis is not a job queue, and create-run payloads are never persisted because layer config may include model credentials. """ from collections.abc import AsyncIterator from typing import cast from redis.asyncio import Redis from dify_agent.protocol.schemas import RUN_EVENT_ADAPTER, RunEvent, RunEventsResponse, RunStatus, utc_now from dify_agent.runtime.event_sink import RunEventSink from dify_agent.server.schemas import RunRecord, new_run_id from dify_agent.server.settings import DEFAULT_RUN_RETENTION_SECONDS from dify_agent.storage.redis_keys import run_events_key, run_record_key class RunNotFoundError(LookupError): """Raised when a requested run record does not exist.""" class RedisRunStore(RunEventSink): """Async Redis implementation for run records and event logs. ``run_retention_seconds`` is applied to both the run record key and the per-run Redis stream. Event writes run ``XADD`` and both TTL refreshes in one Redis transaction so a newly created stream is not left without expiration if the client is interrupted between commands. Event writes also refresh the record TTL so long-running runs that keep producing events do not lose their status record mid-run. """ redis: Redis prefix: str run_retention_seconds: int def __init__( self, redis: Redis, *, prefix: str = "dify-agent", run_retention_seconds: int = DEFAULT_RUN_RETENTION_SECONDS, ) -> None: if run_retention_seconds <= 0: raise ValueError("run_retention_seconds must be positive") self.redis = redis self.prefix = prefix self.run_retention_seconds = run_retention_seconds async def create_run(self) -> RunRecord: """Persist a running run record without storing the create request.""" run_id = new_run_id() record = RunRecord(run_id=run_id, status="running") await self.redis.set( run_record_key(self.prefix, run_id), record.model_dump_json(), ex=self.run_retention_seconds, ) return record async def get_run(self, run_id: str) -> RunRecord: """Return one run record or raise ``RunNotFoundError``.""" value = await self.redis.get(run_record_key(self.prefix, run_id)) if value is None: raise RunNotFoundError(run_id) if isinstance(value, bytes): value = value.decode() return RunRecord.model_validate_json(value) async def update_status(self, run_id: str, status: RunStatus, error: str | None = None) -> None: """Update the status fields of an existing run record.""" record = await self.get_run(run_id) updated = record.model_copy(update={"status": status, "updated_at": utc_now(), "error": error}) await self.redis.set( run_record_key(self.prefix, run_id), updated.model_dump_json(), ex=self.run_retention_seconds, ) async def append_event(self, event: RunEvent) -> str: """Append an event JSON payload to the run's Redis stream with TTLs.""" events_key = run_events_key(self.prefix, event.run_id) payload = RUN_EVENT_ADAPTER.dump_json(event, exclude={"id"}).decode() async with self.redis.pipeline(transaction=True) as pipeline: _ = pipeline.xadd( events_key, {"payload": payload}, ) _ = pipeline.expire(events_key, self.run_retention_seconds) _ = pipeline.expire(run_record_key(self.prefix, event.run_id), self.run_retention_seconds) results = cast(list[object], await pipeline.execute()) event_id = results[0] return event_id.decode() if isinstance(event_id, bytes) else str(event_id) async def get_events(self, run_id: str, *, after: str = "0-0", limit: int = 100) -> RunEventsResponse: """Read a bounded page of events after ``after`` cursor.""" await self.get_run(run_id) raw_events = await self.redis.xrange(run_events_key(self.prefix, run_id), min=f"({after}", count=limit) events = [self._decode_event(run_id, raw_id, fields) for raw_id, fields in raw_events] next_cursor = events[-1].id if events else after return RunEventsResponse(run_id=run_id, events=events, next_cursor=next_cursor) async def iter_events(self, run_id: str, *, after: str = "0-0") -> AsyncIterator[RunEvent]: """Yield replayed and future events for SSE clients.""" await self.get_run(run_id) cursor = after while True: page = await self.get_events(run_id, after=cursor, limit=100) for event in page.events: if event.id is not None: cursor = event.id yield event if not page.events: break while True: response = await self.redis.xread({run_events_key(self.prefix, run_id): cursor}, block=30_000, count=100) if not response: continue for _stream_name, entries in response: for raw_id, fields in entries: event = self._decode_event(run_id, raw_id, fields) if event.id is not None: cursor = event.id yield event @staticmethod def _decode_event(run_id: str, raw_id: object, fields: dict[object, object]) -> RunEvent: """Decode one Redis stream entry into a public event.""" payload = fields.get(b"payload") or fields.get("payload") if isinstance(payload, bytes): payload = payload.decode() event_id = raw_id.decode() if isinstance(raw_id, bytes) else str(raw_id) event = RUN_EVENT_ADAPTER.validate_json(cast(str, payload)) return event.model_copy(update={"id": event_id, "run_id": run_id}) __all__ = ["DEFAULT_RUN_RETENTION_SECONDS", "RedisRunStore", "RunNotFoundError"]