Proposal: adapt durable stream for communication between API and workflow workers

**AI disclosure**: This issue was drafted with Codex. I have reviewed the report, and I am responsible for the content.

### Self Checks

- [x] I have read the [Contributing Guide](https://github.com/langgenius/dify/blob/main/CONTRIBUTING.md) and [Language Policy](https://github.com/langgenius/dify/issues/1542).
- [x] I have searched for existing issues [search for existing issues](https://github.com/langgenius/dify/issues), including closed ones.
- [x] I confirm that I am using English to submit this report, otherwise it will be closed.
- [x] Please do not modify this template :) and fill in all the required fields.

### 1. Is this request related to a challenge you're experiencing? Tell me about your story.

## Context

Communication between API processes and workflow workers needs a stable resume cursor.

When an API connection is interrupted, the caller should be able to save the last returned resume boundary, establish a new subscription from that boundary, and replay unprocessed events that remain inside the retention window.

The existing broadcast-channel abstraction does not provide this contract:

- A subscription does not explicitly declare its starting position.
- Received payloads do not carry a resume cursor.
- Using the Redis Streams tail (`$`) can lose records appended before the reader is fully established.
- Reading from the beginning (`0-0`) can replay stale workflow events from an earlier connection.
- Backend-specific fixes cannot provide reliable resumption without changing the public abstraction.

Related reports include:

- [#32518](https://github.com/langgenius/dify/issues/32518): a subscribe-before-publish race can drop `workflow_started`.
- [#34040](https://github.com/langgenius/dify/issues/34040): replaying from `0-0` can emit stale `workflow_paused` events and cause reconnect loops.
- [#40948](https://github.com/langgenius/dify/issues/40948): reading Redis Streams from `$` reintroduces early-event loss.

## Proposal

Add a durable-stream API alongside the existing broadcast-channel API.

The two abstractions should not inherit from each other. The durable-stream API retains the useful parts of the existing model—topic-bound producers, independent subscribers, and explicit subscription lifecycles—while adding:

- An explicit subscription starting position.
- A resume cursor on every returned record.
- Exclusive cursor-based resumption.
- Defined behavior when retention makes resumption impossible.
- Backend-independent failure semantics.

The public API should not expose Redis stream IDs, Kafka offsets or partitions or other broker specific identities.

## Proposed API stub

```python
from __future__ import annotations

import dataclasses
import enum
import types
from abc import abstractmethod
from contextlib import AbstractContextManager
from typing import NewType, Protocol, Self, override

# A cursor is an opaque continuation token that may be persisted or transmitted
# and then returned unchanged to the topic that produced it. Callers must not
# parse, modify, order, synthesize, or rely on equality between cursors.
TopicCursor = NewType("TopicCursor", str)

class Closed(enum.Enum):
"""The terminal receive state for a closed subscription."""

CLOSED = enum.auto()

CLOSED = Closed.CLOSED

class DurableStreamError(Exception):
"""Base class for errors exposed by the durable stream abstraction."""

class CursorUnavailableError(DurableStreamError):
"""The backend reported that a requested resume position is unavailable."""

class DurableStreamUnavailableError(DurableStreamError):
"""The operation could not complete according to the stream contract."""

@dataclasses.dataclass(slots=True, frozen=True)
class DurableStreamRecord:
payload: bytes
cursor: TopicCursor

class DurableStreamSubscription(
AbstractContextManager["DurableStreamSubscription"],
Protocol,
):
"""An independently positioned subscription to one durable stream topic."""

@override
@abstractmethod
def __enter__(self) -> Self:
"""Establish delivery from the starting position fixed when the subscription was created."""
...

@override
def __exit__(
self,
exc_type: type[BaseException] | None,
exc_value: BaseException | None,
traceback: types.TracebackType | None,
) -> bool | None:
self.close()
return None

@abstractmethod
def close(self) -> None:
"""Close the subscription and release its resources.

This method is idempotent and must not raise. It may be called from
another thread and must unblock a receive call that is waiting for a
record. The interrupted receive call returns CLOSED.
"""
...

@abstractmethod
def receive(self, timeout: float = 0.1) -> DurableStreamRecord | Closed | None:
"""Receive the next record.

Return None when the timeout expires. Return CLOSED when the
subscription is permanently closed.

Raises:
CursorUnavailableError: The backend reported that the current
resume position is unavailable.
DurableStreamUnavailableError: Delivery cannot continue without
violating the stream contract.
"""
...

class DurableStreamProducer(Protocol):
"""A write-only interface already bound to one topic.

Implementations must be thread-safe and support concurrent calls.
"""

@abstractmethod
def append(self, payload: bytes) -> None:
"""Append payload and return after the backend acknowledges the record."""
...

class DurableStreamTopic(DurableStreamProducer, Protocol):
"""A named, ordered durable stream with independent subscriptions.

Implementations must be thread-safe and support concurrent calls.
"""

@abstractmethod
def as_producer(self) -> DurableStreamProducer:
"""Return a write-only view of this topic."""
...

@abstractmethod
def subscribe_from_beginning(self) -> DurableStreamSubscription:
"""Create a subscription starting at the earliest retained record."""
...

@abstractmethod
def subscribe_from_cursor(
self,
cursor: TopicCursor,
) -> DurableStreamSubscription:
"""Establish and return a subscription starting strictly after cursor.

Raises:
CursorUnavailableError: The cursor cannot be resolved by this topic.
DurableStreamUnavailableError: The subscription cannot be established.
"""
...

@abstractmethod
def subscribe_from_tail(self) -> DurableStreamSubscription:
"""Create a subscription starting after the establishment-time tail."""
...

@abstractmethod
def seal(self) -> None:
"""Irreversibly prevent further appends to this topic.

This operation is idempotent. Sealing finalizes the logical record
sequence.
"""
```

## Why this API has a separate abstraction

A broadcast channel represents live delivery. A durable stream represents ordered records that remain available for a configured retention period.

Extending the existing broadcast-channel interface would leave important behavior implicit, including:

- Whether a new subscription starts from retained history or the current tail.
- When the tail position becomes fixed.
- How a subscriber resumes after a disconnection.
- How the caller distinguishes a timeout from permanent closure.
- What happens when the requested history has expired.

A parallel API lets both abstractions keep a narrow and testable contract.

## Cursor and record semantics

`TopicCursor` is an opaque resume boundary produced by one logical topic.

A caller may persist or transmit the cursor and later return it unchanged to the topic that produced it. A caller must not:

- Parse or modify the cursor.
- Construct a cursor.
- Compare cursors for ordering.
- Depend on cursor equality.
- Reuse a cursor with another topic.

`DurableStreamRecord.cursor` is a safe resume boundary returned together with the payload. Resuming from that cursor must not skip any record that the previous subscription had not returned.

The cursor represents subscription progress. It does not represent delivery identity, and the API does not provide deduplication or exactly-once delivery.

When a cursor crosses a trust boundary, the protocol adapter should wrap it in a versioned, AEAD-protected opaque token. Token encoding and key management are outside the durable-stream abstraction.

## Append semantics

`append()` returns successfully only after the backend has accepted the record and assigned it a stable stream position. Independent subscribers can subsequently read the record until the configured retention policy removes it.

The method returns `None` in the initial API. A producer-side cursor is not currently required because subscribers receive cursors through `DurableStreamRecord`.

The API does not define a replication factor, `fsync` policy, or survival guarantees after a broker failure. Those guarantees belong to the backend deployment policy.

If `append()` raises `DurableStreamUnavailableError`, the caller may not be able to determine whether the backend accepted the record. Retrying the operation may therefore append a duplicate record.

Multiple threads may call the same producer concurrently. The API does not guarantee call-entry order or return order. The record order established by the topic is authoritative.

## Subscription start positions

The three subscription methods have distinct behavior:

- `subscribe_from_beginning()` fixes the earliest retained record as its inclusive starting position before the method returns.
- `subscribe_from_cursor(cursor)` uses the supplied cursor as its exclusive starting boundary.
- `subscribe_from_tail()` fixes the current tail as its exclusive starting boundary before the method returns.

The starting boundary is immutable after the subscription method returns.

`__enter__()` initializes delivery from the previously fixed starting boundary. It may perform adapter-specific initialization, but it must not recalculate, replace, or advance that boundary.

Records appended after the starting boundary has been fixed must be available to the subscription after it is successfully entered, unless retention removes those records before they are delivered.

If the fixed starting boundary is malformed, belongs to another topic, or can no longer be resolved when delivery is initialized, `__enter__()` must raise `CursorUnavailableError`. The adapter must not silently fall back to the beginning or the tail.

If `__enter__()` acquires partial resources and then fails, it must release those resources before raising the error.

A subscription is one-shot. The caller may enter it at most once. If initialization fails or the context is exited, the subscription cannot be entered again.

### Retention

Retention is configured and enforced outside this abstraction. Deployments are responsible for sizing retention for the expected disconnection and processing lag.

The abstraction does not independently detect or repair gaps caused by retention. When the backend
reports that a requested or current resume position is unavailable, the adapter must raise
`CursorUnavailableError`. The adapter must not intentionally fall back to the beginning or tail
after receiving such a report.

## Receiving, closure, and failure behavior

`receive(timeout)` has three normal results:

- `DurableStreamRecord` means a record was received.
- `None` means only that this call reached its timeout.
- `CLOSED` means the subscription is permanently closed.

An explicit `receive()` method is used instead of `__iter__()` so callers can regularly process cancellation, heartbeat, and connection-disconnect checks.

`close()` must:

- Be idempotent.
- Return `None`.
- Never raise.
- Be callable from another thread.
- Unblock a waiting `receive()`.
- Cause the interrupted and all subsequent `receive()` calls to return `CLOSED`.

Except for cross-thread calls to `close()`, one subscription does not need to support concurrent method calls.

If the backend reports that the subscription can no longer continue from its current resume boundary because records were removed by retention, `receive()` must raise `CursorUnavailableError`. The adapter must not knowingly skip to a later retained record.

Other failures that prevent the adapter from continuing according to this contract must raise `DurableStreamUnavailableError`. Redis, Kafka, NATS, or other backend-specific exceptions must not escape the adapter.

After either error is raised, the subscription is permanently closed. Subsequent `receive()` calls return `CLOSED`.

## Ordering and subscriber independence

Each subscription reads the topic independently. An implementation must not use competing-consumer semantics in which one subscriber prevents another subscriber from receiving the same record.

Records from one topic are returned in stream order.

`DurableStreamTopic` represents a logical stream. It does not need to map one-to-one to a physical broker topic or stream. An adapter may multiplex several logical topics onto one backend resource if it preserves:

- Topic isolation.
- Per-topic ordering.
- Independent subscriptions.
- Cursor resumption semantics.

Topic construction, provisioning, retention configuration, physical topology, and deletion are outside this public API.

## Backpressure

Memory used to hold records that have not yet been returned must be bounded.

A slow subscription must not:

- Cause records to be silently discarded.
- Block producers indefinitely.

An adapter may pause its reader, use synchronous pulls, apply broker-native flow control, or permanently close a lagging subscription with the appropriate durable-stream error. The public API does not prescribe one backpressure implementation.

## Backend feasibility

The public contract is intended to be implementable over multiple durable backends:

| Backend | Native record position | Possible exclusive-resume implementation | Constraint hidden by the adapter |
| --- | --- | --- | --- |
| Redis Streams | Stream entry ID | Pass the last received ID to `XREAD`, which returns entries with larger IDs | The adapter must fix when `$` is resolved during subscription establishment |
| Kafka | Partition and offset | Pin each logical topic to one partition and seek to `offset + 1` | A logical topic spanning multiple partitions has no authoritative total order; a composite cursor alone is insufficient |
| NATS JetStream | Stream sequence | Create a consumer starting from `stream_sequence + 1` | Stream and consumer sequence numbers have different scopes |

These examples test whether the abstraction is implementable. They do not define the public API.

An adapter that cannot satisfy the ordering and resumption contract for its chosen topology must not claim to implement `DurableStreamTopic`.

For backends that support persistent subscriber state, the first adapter may use ephemeral subscribers. Subscriber identity and persistent subscriber lifecycle are not part of the initial API.

## Proposed API-to-worker flow

The durable stream can be applied to API and workflow-worker communication as follows:

1. A workflow worker appends serialized workflow events to a logical stream topic.
2. The API establishes a subscription using an explicit starting position.
3. Each returned `DurableStreamRecord` contains the workflow event payload and its safe resume cursor.
4. The API returns or persists the cursor through its protocol adapter.
5. After a connection interruption, the API creates a new subscription with `subscribe_from_cursor()`.
6. The new subscription starts strictly after the last record returned to the caller.

The workflow event payload schema remains separate from the transport cursor. Business code must not depend on the cursor encoding or the selected stream backend.

## Migration direction

1. Add the durable-stream interfaces, errors, and backend-independent contract tests.
2. Implement the first adapter without changing the existing broadcast-channel contract.
3. Migrate API-to-workflow-worker event delivery behind a configuration or rollout flag.
4. Propagate protected resume tokens through API protocols that support reconnection.
5. Verify reconnect behavior within the retention window.
6. Keep the existing broadcast channel for callers that need only live delivery.

The initial adapter may use Redis Streams because the project already operates that backend. This proposal does not select Redis Streams as the permanent default.

## Non-goals

The initial proposal does not provide:

- Unbounded historical replay.
- Replay pagination.
- Snapshotting or compaction.
- Cold storage.
- Optimized replay latency or memory usage for very long sessions.
- Exactly-once delivery.
- Delivery identity or deduplication.
- Cross-topic cursor reuse.
- Public access to broker-specific stream positions.
- Public topic provisioning, retention configuration, or deletion APIs.

The initial goal is correct cursor-based recovery for records that remain inside the configured retention window.

## Acceptance criteria

- Each subscription method fixes the subscription’s starting boundary before returning.
- `__enter__()` initializes delivery from the previously fixed starting boundary without recalculating, replacing, or advancing it.
- A retained record appended after `subscribe_from_tail()` returns but before `__enter__()` succeeds is delivered after the subscription is entered.
- `subscribe_from_beginning()` delivers records starting with the earliest record retained when the subscription method is called.
- Cursor-based resumption delivers only records strictly after the supplied cursor.
- Two independent subscriptions can receive the same records in topic stream order.
- A missing, malformed, wrong-topic, or backend-reported unavailable cursor raises `CursorUnavailableError` without falling back to the beginning or the tail.
- When the backend reports that delivery cannot continue from the current resume boundary because retention removed records, `receive()` raises `CursorUnavailableError` instead of skipping to a later retained record.
- Failures that prevent append, subscription initialization, or delivery from satisfying the contract are exposed as `DurableStreamUnavailableError`.
- Backend-specific positions, identifiers, topology, and exceptions do not appear in the public API.
- If subscription creation or `__enter__()` fails after acquiring partial resources, the adapter releases those resources before raising the error.
- A subscription cannot be entered again after initialization fails, it exits its context, or it is closed.
- `close()` is idempotent, does not raise, and can be called from another thread.
- A blocked `receive()` returns `CLOSED` after another thread calls `close()`.
- After a terminal `CursorUnavailableError` or `DurableStreamUnavailableError`, the subscription is permanently closed and subsequent `receive()` calls return `CLOSED`.
- Adapter buffering remains bounded and does not silently discard records.
- A slow subscription does not cause `append()` to block indefinitely. If the adapter can no longer preserve gap-free delivery under backpressure, it terminates the subscription with `DurableStreamUnavailableError`.
- An API integration test can disconnect after receiving a record and resume from the next record while the cursor remains within retention.
- Existing broadcast-channel behavior remains unchanged during migration.

## Open question

Which backend should be the default durable-stream implementation?

The current candidates are:

- Redis Streams.
- PostgreSQL, potentially combined with partition and retention management through `pg_partman` or TimescaleDB.

The selection should be based on operational constraints and the ability to satisfy the public contract. Broker-specific concepts should not change the abstraction.

### 2. Additional context or comments

_No response_

### 3. Can you help us with this feature?

- [ ] I am interested in contributing to this feature.