Package at.aimon.session.web.internal

Class DefaultWebAgentSessionManager

java.lang.Object

at.aimon.session.web.internal.DefaultWebAgentSessionManager

All Implemented Interfaces:

WebAgentSessionManager, AutoCloseable

public final class DefaultWebAgentSessionManager
extends Object
implements WebAgentSessionManager

Default WebAgentSessionManager implementation that wires the in-memory and SPI components together to realize design §7.1's submit flow on a single node, with cross-node hooks for distributed deployments.

The manager is intentionally stateless beyond the per-conversation collaborators: locks, inbox, signal bus, and idempotency store are accessed exclusively through their SPIs. Local-only state — session cache, in-process event publisher, signal subscriptions, and per-conversation idle-sweep scheduler — is owned and closed at close().

Out of scope for close(): the AgentExecutionContext is agent-scoped and shared across conversations; closing this manager closes cached sessions but does not close those contexts (and must not — other agents/sessions may still hold references). Closing AgentExecutionContext (and the McpClientManager / other ContextScoped resources it owns) is the application bootstrap's responsibility, typically via OrcaAgentExecutionContextManager#destroyContext at shutdown. See AgentSessionOpener and docs/design/agent-execution-context-rescoping.md.

Constructor Summary

Constructors

Constructor	Description
DefaultWebAgentSessionManager(at.aimon.core.agent.session.AgentSessionFactory sessionFactory, at.aimon.core.agent.conversation.ConversationRepository repository, ConversationLock lock, ConversationSignalBus signalBus, ConversationInbox inbox, IdempotencyStore idempotencyStore, String nodeId, Duration idleTtl, int maxCachedSessions, Duration lockLease, Duration lockExtendInterval, Duration holderLossSweepInterval, Duration idempotencyPrimaryTtl, Duration idempotencySecondaryTtl, Duration releaseInterruptTimeout)
DefaultWebAgentSessionManager(at.aimon.core.agent.session.AgentSessionFactory sessionFactory, at.aimon.core.agent.conversation.ConversationRepository repository, ConversationLock lock, ConversationSignalBus signalBus, ConversationInbox inbox, IdempotencyStore idempotencyStore, String nodeId, Duration idleTtl, int maxCachedSessions, Duration lockLease, Duration lockExtendInterval, Duration holderLossSweepInterval, Duration idempotencyPrimaryTtl, Duration idempotencySecondaryTtl, Duration releaseInterruptTimeout, SessionMetrics metrics)
DefaultWebAgentSessionManager(AgentSessionOpener sessionOpener, at.aimon.core.agent.conversation.ConversationRepository repository, ConversationLock lock, ConversationSignalBus signalBus, ConversationInbox inbox, IdempotencyStore idempotencyStore, String nodeId, Duration idleTtl, int maxCachedSessions, Duration lockLease, Duration lockExtendInterval, Duration holderLossSweepInterval, Duration idempotencyPrimaryTtl, Duration idempotencySecondaryTtl, Duration releaseInterruptTimeout)
DefaultWebAgentSessionManager(AgentSessionOpener sessionOpener, at.aimon.core.agent.conversation.ConversationRepository repository, ConversationLock lock, ConversationSignalBus signalBus, ConversationInbox inbox, IdempotencyStore idempotencyStore, String nodeId, Duration idleTtl, int maxCachedSessions, Duration lockLease, Duration lockExtendInterval, Duration holderLossSweepInterval, Duration idempotencyPrimaryTtl, Duration idempotencySecondaryTtl, Duration releaseInterruptTimeout, SessionMetrics metrics)

Method Summary

Modifier and Type	Method	Description
void	close()	Close the manager.
boolean	closeGracefully(Duration timeout)	Stop accepting new submits and wait up to timeout for in-flight turns to finish, then close the manager.
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.

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Constructor Details

DefaultWebAgentSessionManager

public DefaultWebAgentSessionManager(at.aimon.core.agent.session.AgentSessionFactory sessionFactory,
 at.aimon.core.agent.conversation.ConversationRepository repository,
 ConversationLock lock,
 ConversationSignalBus signalBus,
 ConversationInbox inbox,
 IdempotencyStore idempotencyStore,
 String nodeId,
 Duration idleTtl,
 int maxCachedSessions,
 Duration lockLease,
 Duration lockExtendInterval,
 Duration holderLossSweepInterval,
 Duration idempotencyPrimaryTtl,
 Duration idempotencySecondaryTtl,
 Duration releaseInterruptTimeout)

DefaultWebAgentSessionManager

public DefaultWebAgentSessionManager(at.aimon.core.agent.session.AgentSessionFactory sessionFactory,
 at.aimon.core.agent.conversation.ConversationRepository repository,
 ConversationLock lock,
 ConversationSignalBus signalBus,
 ConversationInbox inbox,
 IdempotencyStore idempotencyStore,
 String nodeId,
 Duration idleTtl,
 int maxCachedSessions,
 Duration lockLease,
 Duration lockExtendInterval,
 Duration holderLossSweepInterval,
 Duration idempotencyPrimaryTtl,
 Duration idempotencySecondaryTtl,
 Duration releaseInterruptTimeout,
 SessionMetrics metrics)

DefaultWebAgentSessionManager

public DefaultWebAgentSessionManager(AgentSessionOpener sessionOpener,
 at.aimon.core.agent.conversation.ConversationRepository repository,
 ConversationLock lock,
 ConversationSignalBus signalBus,
 ConversationInbox inbox,
 IdempotencyStore idempotencyStore,
 String nodeId,
 Duration idleTtl,
 int maxCachedSessions,
 Duration lockLease,
 Duration lockExtendInterval,
 Duration holderLossSweepInterval,
 Duration idempotencyPrimaryTtl,
 Duration idempotencySecondaryTtl,
 Duration releaseInterruptTimeout)

DefaultWebAgentSessionManager

public DefaultWebAgentSessionManager(AgentSessionOpener sessionOpener,
 at.aimon.core.agent.conversation.ConversationRepository repository,
 ConversationLock lock,
 ConversationSignalBus signalBus,
 ConversationInbox inbox,
 IdempotencyStore idempotencyStore,
 String nodeId,
 Duration idleTtl,
 int maxCachedSessions,
 Duration lockLease,
 Duration lockExtendInterval,
 Duration holderLossSweepInterval,
 Duration idempotencyPrimaryTtl,
 Duration idempotencySecondaryTtl,
 Duration releaseInterruptTimeout,
 SessionMetrics metrics)

Method Details

submit

public WebSubmitOutcome submit(WebSubmitRequest request)

Description copied from interface: WebAgentSessionManager

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).

Specified by:

submit in interface WebAgentSessionManager

Parameters:

request - the submit request (must not be null)

Returns:

outcome describing whether the turn ran here or was queued for another node

events

public Flow.Publisher<at.aimon.core.agent.stream.AgentExecutionEvent> events(at.aimon.core.agent.conversation.ConversationId conversationId)

Description copied from interface: WebAgentSessionManager

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 WebAgentSessionManager.releaseConversation(ConversationId).

Specified by:

events in interface WebAgentSessionManager

Parameters:

conversationId - the conversation (must not be null)

Returns:

a Flow.Publisher of events

interrupt

public void interrupt(at.aimon.core.agent.conversation.ConversationId conversationId,
 at.aimon.core.agent.interrupt.InterruptReason reason)

Description copied from interface: WebAgentSessionManager

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.

Specified by:

interrupt in interface WebAgentSessionManager

Parameters:

conversationId - the conversation (must not be null)

reason - the interrupt classification (must not be null)

releaseConversation

public void releaseConversation(at.aimon.core.agent.conversation.ConversationId conversationId)

Description copied from interface: WebAgentSessionManager

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 WebAgentSessionManager.deleteConversation(ConversationId) instead; direct repository.delete(id) bypasses the manager's lock and races in-flight turns (design §13 issue #10).

Specified by:

releaseConversation in interface WebAgentSessionManager

Parameters:

conversationId - the conversation (must not be null)

deleteConversation

public void deleteConversation(at.aimon.core.agent.conversation.ConversationId conversationId)

Description copied from interface: WebAgentSessionManager

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 WebAgentSessionManager.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.

Specified by:

deleteConversation in interface WebAgentSessionManager

Parameters:

conversationId - the conversation (must not be null)

close

public void close()

Description copied from interface: WebAgentSessionManager

Close the manager. Local sessions are closed; application-scoped collaborators (factory, executor, repository, scheduling components) are not.

Specified by:

close in interface AutoCloseable

Specified by:

close in interface WebAgentSessionManager

closeGracefully

public boolean closeGracefully(Duration timeout)

Description copied from interface: WebAgentSessionManager

Stop accepting new submits and wait up to timeout for in-flight turns to finish, then close the manager. Subsequent WebAgentSessionManager.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.

Specified by:

closeGracefully in interface WebAgentSessionManager

Parameters:

timeout - maximum time to wait for in-flight turns; Duration.ZERO disables draining and matches WebAgentSessionManager.close() semantics (must not be null, must not be negative)

Returns:

true if all in-flight turns drained cleanly within timeout, false on timeout