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}