mirror of
https://github.com/langgenius/dify.git
synced 2026-11-11 05:41:09 +08:00
twwu
ffc2f267d0
Observe Builder commands and native tests on one session stream. Reconnect to the retained Run without submitting another test. Use Stop and Resume for task control during native tests and keep approval details available. Allow retained local sandbox bindings to be checked without reading restricted ancestor directories. Refs: ENG-1448
80 lines
3.1 KiB
Python
80 lines
3.1 KiB
Python
"""Session progress bus over the Redis broadcast channel.
|
|
|
|
The session SSE endpoint subscribes here before reading the session snapshot.
|
|
Merely constructing the Redis subscription is not sufficient because
|
|
subscriptions activate lazily.
|
|
|
|
Uses ``extensions.ext_redis.get_pubsub_broadcast_channel()`` rather than
|
|
constructing ``StreamsBroadcastChannel`` directly: that helper builds the
|
|
channel on the raw pubsub client (avoiding a double key-prefix bug when
|
|
``REDIS_KEY_PREFIX`` is set) and respects the deployment's configured
|
|
``PUBSUB_REDIS_CHANNEL_TYPE`` (streams / sharded / pubsub), so publishers
|
|
and subscribers always agree on channel type and keys. It asserts the
|
|
pubsub client is initialized, so it must be called lazily inside
|
|
``publish``/``subscribe`` rather than at module import time.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
from typing import Any
|
|
|
|
from extensions.ext_redis import get_pubsub_broadcast_channel
|
|
from libs.broadcast_channel.channel import Subscription, SupportsPreparedSubscription
|
|
|
|
_TOPIC_FMT = "dify_builder:{session_id}"
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def _topic(session_id: str) -> str:
|
|
return _TOPIC_FMT.format(session_id=session_id)
|
|
|
|
|
|
def publish(session_id: str, event: dict[str, Any]) -> None:
|
|
"""Publish ``event`` to ``session_id``'s progress topic, JSON-encoded."""
|
|
get_pubsub_broadcast_channel().topic(_topic(session_id)).publish(json.dumps(event).encode())
|
|
|
|
|
|
def publish_native_change(session_id: str) -> None:
|
|
"""Notify observers after commit; delivery failure never retries execution.
|
|
|
|
No receipt or private native state travels on this invalidation. A live
|
|
subscriber rereads through the authenticated owner; reconnect restores the
|
|
durable snapshot if notification delivery was interrupted.
|
|
"""
|
|
try:
|
|
publish(session_id, {"kind": "native_state_changed", "session_id": session_id})
|
|
except Exception:
|
|
logger.warning("Builder native observation notification unavailable")
|
|
|
|
|
|
def publish_session_change(session_id: str) -> None:
|
|
"""Invalidate the read model after a synchronous command has committed."""
|
|
try:
|
|
publish(session_id, {"kind": "session_state_changed", "session_id": session_id})
|
|
except Exception:
|
|
logger.warning("Builder session observation notification unavailable")
|
|
|
|
|
|
def subscribe(session_id: str) -> Subscription:
|
|
"""Establish a lossless delivery boundary for ``session_id`` progress.
|
|
|
|
Redis Streams can fix the boundary without starting their listener, so use
|
|
``prepare_subscription`` when the topic exposes that capability. Redis
|
|
Pub/Sub has no replay boundary and must be entered eagerly so its SUBSCRIBE
|
|
command completes before the Celery task can publish. The SSE generator
|
|
consumes the returned subscription as-is and owns closing it.
|
|
"""
|
|
subscriber = get_pubsub_broadcast_channel().topic(_topic(session_id)).as_subscriber()
|
|
if isinstance(subscriber, SupportsPreparedSubscription):
|
|
return subscriber.prepare_subscription()
|
|
|
|
subscription = subscriber.subscribe()
|
|
try:
|
|
subscription.__enter__()
|
|
except BaseException:
|
|
subscription.close()
|
|
raise
|
|
return subscription
|