001/* 002 * Copyright (c) 2026 Singular 003 * SPDX-License-Identifier: MIT 004 */ 005package ai.singlr.runtime; 006 007import ai.singlr.session.AgentSession; 008import ai.singlr.session.SessionOptions; 009import java.time.Clock; 010import java.time.Duration; 011import java.time.Instant; 012import java.util.Collection; 013import java.util.Comparator; 014import java.util.Map; 015import java.util.Objects; 016import java.util.Optional; 017import java.util.Set; 018import java.util.concurrent.ConcurrentHashMap; 019import java.util.concurrent.atomic.AtomicReference; 020import java.util.function.Function; 021 022/** 023 * In-memory registry of live sessions. The HTTP service holds one per process; route handlers 024 * lookup sessions by id to dispatch send / interrupt / events / close. 025 * 026 * <p>Sessions are created via {@link #create(SessionOptions)} — a {@link Function} factory wired at 027 * construction (typically {@code AgentSession::create}) builds the impl. The factory is injectable 028 * so tests can substitute a stub session. 029 * 030 * <p>Sessions remain in the registry until {@link #close(String)} is called, even after they reach 031 * a terminal {@link ai.singlr.session.ResultMessage ResultMessage} — keeping them around lets late 032 * SSE subscribers fetch the final {@code LoopEnded} event after termination, and lets the {@code 033 * DELETE /sessions/{id}} route be the explicit cleanup boundary. 034 * 035 * <h2>Retention</h2> 036 * 037 * The registry can keep terminal sessions indefinitely; for long-running services that creates a 038 * slow leak. Two opt-in eviction surfaces address this: 039 * 040 * <ul> 041 * <li>{@link #purgeTerminalOlderThan(Duration)} — sweep terminal sessions older than the supplied 042 * age. Live sessions are never touched. Best called periodically (every minute or so) from 043 * the deployer's scheduler. 044 * <li>{@code SessionRegistry.newBuilder().withMaxSessions(int)} — cap on registered sessions. 045 * When {@link #create(SessionOptions)} would push the count over the cap, the registry evicts 046 * the oldest terminal session first; if none is available, the create call throws {@link 047 * IllegalStateException}. Live sessions are never evicted. 048 * </ul> 049 * 050 * <h2>Thread-safety</h2> 051 * 052 * Thread-safe. All routes share one registry; concurrent create / get / close are common. Backed by 053 * {@link ConcurrentHashMap}; {@link #create(SessionOptions)} rejects duplicate ids. The cap check 054 * is best-effort under contention — under a flood of concurrent creates the count may briefly 055 * exceed the cap before evictions catch up; the cap is an SLA hint, not a hard barrier. 056 */ 057public final class SessionRegistry { 058 059 private final ConcurrentHashMap<String, SessionEntry> sessions = new ConcurrentHashMap<>(); 060 private final Function<SessionOptions, AgentSession> factory; 061 private final Clock clock; 062 private final int maxSessions; 063 064 /** 065 * Registry that constructs sessions via {@link AgentSession#create(SessionOptions)} with system 066 * clock and no cap. 067 * 068 * @return a fresh registry 069 */ 070 public static SessionRegistry inMemory() { 071 return newBuilder().build(); 072 } 073 074 /** 075 * Registry that constructs sessions via a custom factory, system clock, no cap. Intended for 076 * tests; production sessions use {@link #inMemory()} or {@link #newBuilder()}. 077 * 078 * @param factory non-null function mapping options to a fresh session 079 * @return a fresh registry 080 * @throws NullPointerException if {@code factory} is null 081 */ 082 public static SessionRegistry withFactory(Function<SessionOptions, AgentSession> factory) { 083 return newBuilder().withFactory(factory).build(); 084 } 085 086 /** 087 * Start building a registry. Set any of factory / clock / maxSessions; defaults are {@code 088 * AgentSession::create}, {@link Clock#systemUTC()}, and no cap. 089 * 090 * @return a fresh builder 091 */ 092 public static Builder newBuilder() { 093 return new Builder(); 094 } 095 096 private SessionRegistry( 097 Function<SessionOptions, AgentSession> factory, Clock clock, int maxSessions) { 098 this.factory = Objects.requireNonNull(factory, "factory must not be null"); 099 this.clock = Objects.requireNonNull(clock, "clock must not be null"); 100 if (maxSessions <= 0) { 101 throw new IllegalArgumentException("maxSessions must be positive, got " + maxSessions); 102 } 103 this.maxSessions = maxSessions; 104 } 105 106 /** 107 * Create a new session from the given options and register it under its session id. 108 * 109 * @param options the composition record; non-null 110 * @return the freshly-created, unstarted session 111 * @throws NullPointerException if {@code options} is null 112 * @throws IllegalStateException if a session with the same id is already registered, or if the 113 * registry is at its configured {@code maxSessions} cap and no terminal session is available 114 * to evict 115 */ 116 public AgentSession create(SessionOptions options) { 117 Objects.requireNonNull(options, "options must not be null"); 118 if (sessions.size() >= maxSessions && !tryEvictOldestTerminal()) { 119 throw new IllegalStateException( 120 "registry at capacity " 121 + maxSessions 122 + " and no terminal sessions are available to evict"); 123 } 124 var session = factory.apply(options); 125 Objects.requireNonNull(session, "factory returned null session"); 126 var entry = new SessionEntry(session, new AtomicReference<>()); 127 var prev = sessions.putIfAbsent(options.sessionId(), entry); 128 if (prev != null) { 129 session.close(); 130 throw new IllegalStateException("session id already registered: " + options.sessionId()); 131 } 132 session.result().whenComplete((r, t) -> entry.terminatedAt().set(clock.instant())); 133 return session; 134 } 135 136 /** 137 * Look up a registered session by id. 138 * 139 * @param sessionId non-null id 140 * @return the session if present 141 * @throws NullPointerException if {@code sessionId} is null 142 */ 143 public Optional<AgentSession> get(String sessionId) { 144 Objects.requireNonNull(sessionId, "sessionId must not be null"); 145 var entry = sessions.get(sessionId); 146 return entry == null ? Optional.empty() : Optional.of(entry.session()); 147 } 148 149 /** 150 * Close and unregister the session. If no session is registered under {@code sessionId} this is a 151 * no-op. 152 * 153 * @param sessionId non-null id 154 * @return {@code true} if a session was found and closed; {@code false} if no session was 155 * registered 156 * @throws NullPointerException if {@code sessionId} is null 157 */ 158 public boolean close(String sessionId) { 159 Objects.requireNonNull(sessionId, "sessionId must not be null"); 160 var entry = sessions.remove(sessionId); 161 if (entry == null) { 162 return false; 163 } 164 entry.session().close(); 165 return true; 166 } 167 168 /** 169 * Snapshot of currently-registered session ids. Stable point-in-time view; mutations after this 170 * call are not reflected. 171 * 172 * @return defensive snapshot 173 */ 174 public Collection<String> sessionIds() { 175 return Set.copyOf(sessions.keySet()); 176 } 177 178 /** 179 * Number of registered sessions. 180 * 181 * @return non-negative count 182 */ 183 public int size() { 184 return sessions.size(); 185 } 186 187 /** Close and unregister every session. Idempotent. */ 188 public void closeAll() { 189 for (var id : Set.copyOf(sessions.keySet())) { 190 close(id); 191 } 192 } 193 194 /** 195 * Sweep every terminal session whose termination instant is older than {@code age} relative to 196 * the registry's {@link Clock}. Live sessions are not touched, even if the registry has held them 197 * far longer than {@code age}. Returns the count of sessions closed + unregistered. 198 * 199 * <p>A session is "terminal" once its {@link AgentSession#result()} future has completed — the 200 * registry captures the wall-clock instant of completion when the future settles, and this method 201 * compares that instant against {@code now - age}. 202 * 203 * @param age non-null, non-negative; sessions terminated at-or-before {@code now - age} are 204 * purged 205 * @return number of sessions purged 206 * @throws NullPointerException if {@code age} is null 207 * @throws IllegalArgumentException if {@code age} is negative 208 */ 209 public int purgeTerminalOlderThan(Duration age) { 210 Objects.requireNonNull(age, "age must not be null"); 211 if (age.isNegative()) { 212 throw new IllegalArgumentException("age must be non-negative, got " + age); 213 } 214 var cutoff = clock.instant().minus(age); 215 int purged = 0; 216 for (var entry : sessions.entrySet()) { 217 var terminated = entry.getValue().terminatedAt().get(); 218 if (terminated != null && !terminated.isAfter(cutoff)) { 219 if (close(entry.getKey())) { 220 purged++; 221 } 222 } 223 } 224 return purged; 225 } 226 227 /** 228 * Evict the single oldest terminal session, if any. Returns {@code true} when an entry was 229 * removed and closed; {@code false} when no terminal session was available. 230 */ 231 private boolean tryEvictOldestTerminal() { 232 var oldestId = 233 sessions.entrySet().stream() 234 .filter(e -> e.getValue().terminatedAt().get() != null) 235 .min( 236 Comparator.comparing( 237 e -> e.getValue().terminatedAt().get(), Comparator.naturalOrder())) 238 .map(Map.Entry::getKey); 239 return oldestId.map(this::close).orElse(false); 240 } 241 242 /** Per-session bookkeeping: the session itself + the instant its result-future completed. */ 243 private record SessionEntry(AgentSession session, AtomicReference<Instant> terminatedAt) {} 244 245 /** Fluent builder for {@link SessionRegistry}. */ 246 public static final class Builder { 247 248 private Function<SessionOptions, AgentSession> factory = AgentSession::create; 249 private Clock clock = Clock.systemUTC(); 250 private int maxSessions = Integer.MAX_VALUE; 251 252 private Builder() {} 253 254 /** 255 * Override the session factory. Production code keeps the default; tests supply a stub. 256 * 257 * @param factory non-null factory 258 * @return this builder 259 * @throws NullPointerException if {@code factory} is null 260 */ 261 public Builder withFactory(Function<SessionOptions, AgentSession> factory) { 262 this.factory = Objects.requireNonNull(factory, "factory must not be null"); 263 return this; 264 } 265 266 /** 267 * Override the clock used to stamp terminal sessions. Tests use a fixed clock for deterministic 268 * eviction; production keeps {@link Clock#systemUTC()}. 269 * 270 * @param clock non-null clock 271 * @return this builder 272 * @throws NullPointerException if {@code clock} is null 273 */ 274 public Builder withClock(Clock clock) { 275 this.clock = Objects.requireNonNull(clock, "clock must not be null"); 276 return this; 277 } 278 279 /** 280 * Cap registered sessions. When {@link #create(SessionOptions)} would push the count over this 281 * number, the registry evicts the oldest terminal session first; if no terminal session is 282 * available, the create call throws. 283 * 284 * @param maxSessions positive cap 285 * @return this builder 286 * @throws IllegalArgumentException if {@code maxSessions} is not positive 287 */ 288 public Builder withMaxSessions(int maxSessions) { 289 if (maxSessions <= 0) { 290 throw new IllegalArgumentException("maxSessions must be positive, got " + maxSessions); 291 } 292 this.maxSessions = maxSessions; 293 return this; 294 } 295 296 /** 297 * Build the immutable registry. 298 * 299 * @return a fresh registry 300 */ 301 public SessionRegistry build() { 302 return new SessionRegistry(factory, clock, maxSessions); 303 } 304 } 305}