Package com.axelor.ai.chat.stream

Class ChatStreamRegistry

java.lang.Object
com.axelor.ai.chat.stream.ChatStreamRegistry
@Singleton public class ChatStreamRegistry extends Object
Registry of in-flight chat runs, backed by the unified cache API: in-memory under caffeine, distributed under Redisson/JCache — chosen by the app's cache provider config, nothing AI-specific.

Layout in the streams cache (single writer per run, all values JSON):

  • chat:<chatId> — claim {"streamId","deadline","done","initiatorUser","rebuildsTurn"}; staleness is decided by deadline, cache TTL is only garbage collection
  • run:<streamId>#<seq> — chunk of concatenated SSE frames, seq from 1
  • run:<streamId>#head — {"seq","done"}
  • run:<streamId>#turn — the assistant message id from the run's opening frame; with the claim's rebuildsTurn, which turn a resuming client must let the replay rebuild
  • tool:<chatId>#<callId> — pending client-tool call {"toolName","ownerUser", "claimedBy","state"}; written before the pending part is persisted. Execution eligibility comes from a claimed record only — never from the frame stream or from record absence (clients read absence as "not mine to run", the server as expired). The cache TTL is the only expiry.

A single shared topic carries {"kind","streamId"} envelopes as pure wake-up signals; subscribers always read data from the cache and dedupe by seq, so a lost publish never loses data. All topic access stays in this class — DistributedFactory.getTopic(java.lang.String) scopes topic names by caller class.

  • Constructor Details

    • ChatStreamRegistry

      @Inject public ChatStreamRegistry(com.fasterxml.jackson.databind.ObjectMapper mapper)

  • Method Details

    • begin

      public ActiveStream begin(String chatId, Duration timeout)
      Claims the chat for a new run. Throws AlreadyStreamingException if a live run holds the claim (not done, deadline not passed). A stale deadline means the owning node died — the claim is overwritten.

    • begin

      public ActiveStream begin(String chatId, Duration timeout, String initiatorUser)
      Claims the chat for a run initiated by initiatorUser.

    • begin

      public ActiveStream begin(String chatId, Duration timeout, String initiatorUser, boolean rebuildsTurn)
      Claims the chat, recording whether a replay rebuilds the turn (see ChatStreamRegistry.Claim).

    • subscribe

      public StreamSubscription subscribe(String chatId)
      Returns a subscription replaying the chat's latest run (live or recently finished and still cached), or null when there is nothing to replay.

    • rebuildingMessageId

      public String rebuildingMessageId(String chatId)
      The message a resuming client must drop and let the replay produce again, or null when there is nothing to hand over: no live run, a continuation, or no opening frame written yet.

    • isActive

      public boolean isActive(String chatId)
      True when a live run holds the chat's claim.

    • cancel

      public boolean cancel(String chatId)
      Requests cooperative cancellation. Returns true if a live run was signalled.

    • cancel

      public boolean cancel(String chatId, String user)
      Cancels only when user initiated the active run. This lets a shared-chat prompter stop their own work without allowing one driver to interrupt another driver's run.

    • putPendingCall

      public void putPendingCall(String chatId, String callId, String toolName, String ownerUser)
      Registers a pending client-tool call. Must precede persisting the pending part.

    • readPendingCall

      public ChatStreamRegistry.PendingCall readPendingCall(String chatId, String callId)
      The pending-call record, or null when absent (expired, cancelled, or never registered).

    • claimCall

      public boolean claimCall(String chatId, String callId, String user, String sessionId)
      Claims the call for one of the owner's sessions. CAS under the per-key lock: only a pending record owned by user is claimable; a repeat claim by the same session is idempotent (absorbs a lost response + retry). Everything else — absent record, other user, other session — loses.

    • resolveCall

      public boolean resolveCall(String chatId, String callId, String user, String sessionId)
      Resolves the call — one-time CAS: only the claiming session of the owning user wins, exactly once. Late duplicates and wrong-session submissions return false.

    • removePendingCall

      public void removePendingCall(String chatId, String callId)
      Drops the record (the call was cancelled/superseded) — absence means "not mine to run".

    • absenceMark

      public Long absenceMark(String chatId, String callId)
      When the initiator was first seen absent for this pending call, or null.

    • markAbsence

      public void markAbsence(String chatId, String callId, long epochMillis)

    • clearAbsenceMark

      public void clearAbsenceMark(String chatId, String callId)