v2 connect streams stop after the current run and cannot follow future thread runs

[0xTomDaniel]

Summary

Current v2 connect behavior does not remain attached to a thread across future runs.

Instead, it appears to replay history plus the currently active run, then complete when that run ends, or complete immediately if no run is active. In Intelligence mode, the connect stream explicitly completes on RUN_FINISHED / RUN_ERROR.

That means connect cannot function as a durable thread-follow stream for later runs on the same thread.

Relevant commits

• Initial in-memory runner behavior: 7b83854
• Initial Intelligence runner behavior: 764ec32
• Later Intelligence support reference point: 5b3037a

Why this matters

There is an important class of agent systems where runs are not always initiated by the connected client.

Generically, these agents may:

• continue operating in the background
• emit periodic heartbeats
• run on cron/schedulers
• react to external triggers
• update thread state/messages from server-initiated runs long after the original client-triggered run has finished

For those systems, connect is not just "replay the latest run". It needs to mean "stay attached to this thread and keep surfacing AG-UI events/state for future runs until the client disconnects."

Without that behavior, the UI can connect successfully, observe one run, and then silently stop receiving updates even though the same thread continues evolving on the server.

Current behavior across implementations

This does not appear to be limited to one implementation:

• In-memory connect replays historic events, bridges only the currently active run, and completes otherwise.
• Intelligence connect completes when it receives RUN_FINISHED or RUN_ERROR.

So this looks like a shared contract assumption rather than a single isolated bug.

Expected behavior

A connect stream for a thread should be able to remain open across future runs on that same thread until the client disconnects.

Specifically, connect should support thread-follow semantics for:

• future runs started after the connect stream is opened
• runs initiated by background schedulers, heartbeats, cron jobs, or other server-side automation
• AG-UI state/message updates for runs that were never initiated directly from the current client session

Why this should be first-class

Long-lived/background agents are not an edge case for agent systems. They are a core pattern.

If CopilotKit wants AG-UI thread detail views to reflect real agent state over time, connect needs to support ongoing thread evolution rather than only the currently active run.

Request

Please make v2 connect support durable thread-follow semantics across future runs, instead of completing after the current run or after replay when no run is active.

[copilotkit-support-bot]

To make v2 connect support durable thread-follow semantics across future runs, the completion logic in both the IntelligenceAgentRunner and InMemoryAgentRunner must be modified. Currently, these implementations complete the stream once a run finishes or immediately after replaying history when no run is active.

Details

Proposed Solution

1. IntelligenceAgentRunner Implementation

In IntelligenceAgentRunner.connect, the observer should not be completed when receiving RUN_FINISHED or RUN_ERROR events. This allows the Phoenix channel to remain joined and continue surfacing events from future runs initiated on the same thread (e.g., via background jobs or other sessions).

Location: packages/v2/runtime/src/runner/intelligence.ts

// Update connect method in IntelligenceAgentRunner
channel.on("ag_ui_event", (payload: BaseEvent) => {
  observer.next(payload);

  // REMOVE: 
  // if (payload.type === EventType.RUN_FINISHED || payload.type === EventType.RUN_ERROR) {
  //   observer.complete();
  // }
});

2. InMemoryAgentRunner Implementation

The InMemoryAgentRunner needs a way to bridge events from subsequent runs to existing connect subscribers. This can be achieved by introducing a long-lived Subject per thread that does not complete when an individual run ends.

Location: packages/v2/runtime/src/runner/in-memory.ts

• Update InMemoryEventStore:
Add a threadSubject that survives across multiple runs.

class InMemoryEventStore {
  // ... existing fields
  /** A subject that receives events from all runs and stays open. */
  threadSubject = new Subject<BaseEvent>();
}

• Update run() method:
Push all events to the threadSubject in addition to the per-run nextSubject.

// Inside onEvent callback in run()
onEvent: ({ event }) => {
  // ... existing logic
  nextSubject.next(processedEvent);
  store.threadSubject.next(processedEvent); // NEW: push to long-lived subject
  currentRunEvents.push(processedEvent);
},

// Inside finalization logic in run()
const appendedEvents = finalizeRunEvents(currentRunEvents, { ... });
for (const event of appendedEvents) {
  nextSubject.next(event);
  store.threadSubject.next(event); // NEW: push finalization events
}

• Update connect() method:
Instead of completing after history or active run completion, subscribe to the threadSubject.

connect(request: AgentRunnerConnectRequest): Observable<BaseEvent> {
  const store = GLOBAL_STORE.get(request.threadId);
  const connectionSubject = new ReplaySubject<BaseEvent>(Infinity);

  // ... (Keep existing history replay logic) ...

  // Subscribe to the long-lived thread subject for current and future events
  if (store) {
    store.threadSubject.subscribe({
      next: (event) => {
        // Dedup if message was already in history
        if ("messageId" in event && typeof event.messageId === "string" && emittedMessageIds.has(event.messageId)) {
          return;
        }
        connectionSubject.next(event);
      },
      error: (err) => connectionSubject.error(err),
      // Do NOT call connectionSubject.complete() here
    });
  }

  return connectionSubject.asObservable();
}

Documentation & References

• Agent Runner Interface: packages/v2/runtime/src/runner/agent-runner.ts
• Intelligence Runner: packages/v2/runtime/src/runner/intelligence.ts
• In-Memory Runner: packages/v2/runtime/src/runner/in-memory.ts

Confidence Assessment: High. The requested behavior is clearly blocked by explicit observer.complete() calls and per-run subject lifecycle management identified in the analyzed source code. Changing these to long-lived subjects/observers directly addresses the "durable thread-follow" requirement.

---

Was this helpful?

If this solution worked for you, please click on the appropriate option below to help us improve:

✅ This was helpful | ❌ Need more help

[0xTomDaniel]

I opened a companion discussion to ask the higher-level contract question behind this bug report:

#3534

The issue here is the concrete behavior gap. The discussion asks whether connect is intended to be a durable thread-follow stream across future runs, which seems important for long-lived/background agents and thread detail views that need to stay accurate over time.

[vahe-templi]

Looking forward to this change in order to implement "Human Handoff" in a better way. Now I am forced to update the state on langgraph side and then make requests periodically to fetch latest state to update message. It could have been achieved by just receiving state updates via running thread.