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