Package at.aimon.session.web

Interface WebAgentSessionManager

All Superinterfaces:
AutoCloseable
All Known Implementing Classes:
DefaultWebAgentSessionManager
public interface WebAgentSessionManager extends AutoCloseable
Multi-instance-ready agent session manager facade.

Wraps AgentSession with the cross-node concerns required by a web deployment: distributed locking per conversation, cross-node signal fan-out, mailbox hand-off when the calling node is not the lock holder, and idempotency.

The manager is application-scoped: one instance per process, lifetime ≥ the AgentSessionFactory it owns. It is AutoCloseable so application shutdown can release local sessions and disconnect from any cross-node SPI backends.

Design reference: docs/design/implemented/web-agent-session-manager-design.md §5.1.

  • Method Summary

    Modifier and Type
    Method
    Description
    static WebAgentSessionManagerBuilder
    builder()
    Build a new manager.
    void
    close()
    Close the manager.
    default boolean
    closeGracefully(Duration timeout)
    Stop accepting new submits and wait up to timeout for in-flight turns to finish, then close the manager.
    default void
    deleteConversation(at.aimon.core.agent.conversation.ConversationId conversationId)
    Permanently delete the conversation, including the history persisted in ConversationRepository.
    Flow.Publisher<at.aimon.core.agent.stream.AgentExecutionEvent>
    events(at.aimon.core.agent.conversation.ConversationId conversationId)
    Subscribe to streaming progress events for conversationId.
    void
    interrupt(at.aimon.core.agent.conversation.ConversationId conversationId, at.aimon.core.agent.interrupt.InterruptReason reason)
    Trip an interrupt on the active turn, if any.
    void
    releaseConversation(at.aimon.core.agent.conversation.ConversationId conversationId)
    Release the local session cache entry and emit onComplete() on the corresponding events() publisher.
    WebSubmitOutcome
    submit(WebSubmitRequest request)
    Submit one turn of input to the conversation identified by request.

  • Method Details

    • builder

      static WebAgentSessionManagerBuilder builder()
      Build a new manager. See WebAgentSessionManagerBuilder for the required SPIs and operational tuning parameters.
      Returns:
      a fresh builder

    • submit

      WebSubmitOutcome submit(WebSubmitRequest request)
      Submit one turn of input to the conversation identified by request.

      The manager evaluates WebSubmitRequest.getAgentRef() against the conversation's current binding (design §3.6) and either runs the turn locally (lock acquired) or delivers the input to the cross-node inbox (lock held elsewhere).

      Parameters:
      request - the submit request (must not be null)
      Returns:
      outcome describing whether the turn ran here or was queued for another node
      Throws:
      at.aimon.core.agent.session.exception.ConflictingAgentException - when the request's agentRef differs from the conversation's existing binding

    • events

      Flow.Publisher<at.aimon.core.agent.stream.AgentExecutionEvent> events(at.aimon.core.agent.conversation.ConversationId conversationId)
      Subscribe to streaming progress events for conversationId.

      The publisher fan-outs both locally-emitted events (when this node owns the turn) and remote events arriving via the signal bus (when another node owns the turn). Multi-subscriber: several SSE clients may observe the same conversation simultaneously. Subscribers receive onComplete() on releaseConversation(ConversationId).

      Parameters:
      conversationId - the conversation (must not be null)
      Returns:
      a Flow.Publisher of events

    • interrupt

      void interrupt(at.aimon.core.agent.conversation.ConversationId conversationId, at.aimon.core.agent.interrupt.InterruptReason reason)
      Trip an interrupt on the active turn, if any.

      The manager broadcasts an INTERRUPT signal — every node trips its own active session for that conversation. Idempotent: when no turn is active anywhere, the call is a silent no-op.

      Parameters:
      conversationId - the conversation (must not be null)
      reason - the interrupt classification (must not be null)

    • releaseConversation

      void releaseConversation(at.aimon.core.agent.conversation.ConversationId conversationId)
      Release the local session cache entry and emit onComplete() on the corresponding events() publisher.

      History stored in ConversationRepository is preserved — submitting again with the same id resumes the conversation. For permanent deletion (history removal) call deleteConversation(ConversationId) instead; direct repository.delete(id) bypasses the manager's lock and races in-flight turns (design §13 issue #10).

      Parameters:
      conversationId - the conversation (must not be null)

    • deleteConversation

      default void deleteConversation(at.aimon.core.agent.conversation.ConversationId conversationId)
      Permanently delete the conversation, including the history persisted in ConversationRepository.

      Implementations MUST acquire the conversation lock first (broadcasting INTERRUPT(SESSION_RELEASED) on contention so the current holder yields) and only then call repository.delete(id). This closes design §13 issue #10: a direct repository.delete(id) from outside the manager would race the in-flight turn (the holder may still be writing history) and corrupt cluster state.

      After delete the manager performs the same teardown as releaseConversation(ConversationId) (cache eviction, inbox purge, terminal InterruptedAt, onComplete, EVICT broadcast).

      The default implementation throws UnsupportedOperationException so existing in-process tests and minimal implementations remain source-compatible.

      Parameters:
      conversationId - the conversation (must not be null)
      Throws:
      IllegalStateException - if the lock cannot be acquired within the implementation's bounded retry budget — the caller may retry later

    • close

      void close()
      Close the manager. Local sessions are closed; application-scoped collaborators (factory, executor, repository, scheduling components) are not.
      Specified by:
      close in interface AutoCloseable

    • closeGracefully

      default boolean closeGracefully(Duration timeout)
      Stop accepting new submits and wait up to timeout for in-flight turns to finish, then close the manager. Subsequent submit(WebSubmitRequest) calls during draining throw IllegalStateException.

      If the timeout elapses before all turns drain, surviving turns are interrupted with InterruptReason.SYSTEM_SHUTDOWN and the manager proceeds to a hard close. The lock release on each surviving turn flips the lease so other nodes can take over.

      Parameters:
      timeout - maximum time to wait for in-flight turns; Duration.ZERO disables draining and matches close() semantics (must not be null, must not be negative)
      Returns:
      true if all in-flight turns drained cleanly within timeout, false on timeout