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}