feat(runtime): terminal-job retention + resume-token sweeper + configurable buffer

Three bookkeeping fixes that share an axis (memory hygiene + spec
§6.3/§7.6 buffer sizing) and motivate the same new options:

- `ArcpServerOptions.EventLogCapacity` (default 4096) sizes the
  session-scoped `EventLog` AND the per-job subscription-history
  buffer. Previously the per-job buffer was hard-coded at 1024, so
  subscribers with `history: true` saw a shorter window than resumers
  — inconsistent with spec §7.6 "same buffer window".
- `ArcpServerOptions.TerminalJobRetentionSec` (default 600) bounds how
  long terminal jobs hang around in `_jobs` / `_idempotency`.
  `JobManager.ScheduleTerminalCleanup` evicts them after the window.
  Previously these dictionaries grew without bound for the lifetime
  of the runtime.
- A background `PeriodicTimer`-driven sweeper in `ArcpServer` purges
  expired resume tokens, their reverse-index entries, and dormant
  session shells. Cancelled on `DisposeAsync`. Previously only
  cleaned opportunistically on lookup.

Plumb the three values through the `JobManager` constructor; bump the
public-API tracker accordingly.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: nficano

src/Arcp.Runtime/ArcpServer.cs

@@ -26,6 +26,8 @@ public sealed class ArcpServer : IAsyncDisposable
     private readonly ConcurrentDictionary<SessionId, string> _resumeTokenBySession = new();
     private readonly ConcurrentDictionary<SessionId, SessionState> _resumableSessions = new();
     private readonly ILoggerFactory _loggerFactory;
+    private readonly CancellationTokenSource _sweeperCts = new();
+    private readonly Task? _sweeperTask;
 
     private readonly record struct ResumeRegistration(SessionId SessionId, DateTimeOffset ExpiresAt);
 
@@ -71,11 +73,41 @@ public ArcpServer(ArcpServerOptions options, ILoggerFactory? loggerFactory = nul
             _loggerFactory,
             CredentialManager,
             options.IdempotencyWindowSec,
+            options.EventLogCapacity,
+            options.TerminalJobRetentionSec,
             options.FatalBudgetExhaustion);
         if (CredentialManager is not null)
         {
             _ = Task.Run(() => RevokeOutstandingCredentialsAsync(CancellationToken.None));
         }
+        _sweeperTask = Task.Run(() => RunResumeTokenSweeperAsync(_sweeperCts.Token));
+    }
+
+    /// <summary>Periodically purge expired resume tokens, their reverse indices, and dormant
+    /// session shells (spec §6.3). Without this, long-running runtimes accumulate one entry per
+    /// session forever.</summary>
+    private async Task RunResumeTokenSweeperAsync(CancellationToken cancellationToken)
+    {
+        // Sweep every minute, or sooner for short windows.
+        var interval = TimeSpan.FromSeconds(Math.Clamp(Options.ResumeWindowSec / 4, 30, 300));
+        try
+        {
+            using var timer = new PeriodicTimer(interval, Options.TimeProvider);
+            while (await timer.WaitForNextTickAsync(cancellationToken).ConfigureAwait(false))
+            {
+                var now = Options.TimeProvider.GetUtcNow();
+                foreach (var kv in _resumeTokens)
+                {
+                    if (kv.Value.ExpiresAt > now) continue;
+                    if (_resumeTokens.TryRemove(kv.Key, out _))
+                    {
+                        _resumeTokenBySession.TryRemove(kv.Value.SessionId, out _);
+                        _resumableSessions.TryRemove(kv.Value.SessionId, out _);
+                    }
+                }
+            }
+        }
+        catch (OperationCanceledException) { /* shutdown */ }
     }
 
     /// <summary>Register agent.</summary>
@@ -181,6 +213,14 @@ internal bool TryResume(string resumeToken, out SessionState? session)
     /// <summary>Dispose (asynchronous).</summary>
     public async ValueTask DisposeAsync()
     {
+        try { _sweeperCts.Cancel(); } catch (ObjectDisposedException) { /* already disposed */ }
+        if (_sweeperTask is not null)
+        {
+            try { await _sweeperTask.ConfigureAwait(false); }
+            catch (OperationCanceledException) { /* expected */ }
+        }
+        _sweeperCts.Dispose();
+
         foreach (var session in _sessions.Values)
         {
             try { await session.CloseAsync().ConfigureAwait(false); } catch { /* shutdown */ }

src/Arcp.Runtime/ArcpServerOptions.cs

@@ -31,6 +31,16 @@ public sealed class ArcpServerOptions
     /// (spec §7.2). Submissions with the same key after this window create a fresh job.</summary>
     public int IdempotencyWindowSec { get; init; } = 3600;
 
+    /// <summary>Capacity of the session-scoped event log (spec §6.3 resume window). Per-job
+    /// subscription-history buffers (spec §7.6) inherit this size so a subscriber with
+    /// <c>history: true</c> sees the same window as a resume client.</summary>
+    public int EventLogCapacity { get; init; } = 4096;
+
+    /// <summary>How long terminal jobs are retained in <c>session.list_jobs</c> and idempotency
+    /// records before being garbage-collected. Set to <c>0</c> or a negative value to retain
+    /// indefinitely (legacy behavior).</summary>
+    public int TerminalJobRetentionSec { get; init; } = 600;
+
     /// <summary>Whether a <c>cost.budget</c> exhaustion terminates the job with
     /// <c>BUDGET_EXHAUSTED</c> (legacy v1.0 behavior) or surfaces a non-fatal
     /// <c>tool_result.error</c> so the agent may emit a partial result and return normally

src/Arcp.Runtime/Job.cs

@@ -26,7 +26,7 @@ public sealed class Job
     private readonly List<IssuedCredential> _credentials = [];
     private readonly object _eventBufferGate = new();
     private readonly List<Envelope> _eventBuffer = [];
-    private const int EventBufferCapacity = 1024;
+    private readonly int _eventBufferCapacity;
 
     /// <summary>Gets the job id.</summary>
     public JobId JobId { get; }
@@ -117,7 +117,8 @@ internal Job(JobId jobId, SessionId sessionId, AgentRef agent, Lease lease, Leas
                JsonElement? input, string? idempotencyKey, TraceId? traceId, string? parentJobId,
                string? submitterPrincipal, int? maxRuntimeSec, DateTimeOffset createdAt,
                Func<Envelope, CancellationToken, ValueTask> emit, TimeProvider time,
-               CancellationToken parentCancellation)
+               CancellationToken parentCancellation,
+               int eventBufferCapacity = 4096)
     {
         JobId = jobId;
         SessionId = sessionId;
@@ -133,6 +134,7 @@ internal Job(JobId jobId, SessionId sessionId, AgentRef agent, Lease lease, Leas
         CreatedAt = createdAt;
         _emit = emit;
         _time = time;
+        _eventBufferCapacity = eventBufferCapacity > 0 ? eventBufferCapacity : 4096;
         CancellationSource = CancellationTokenSource.CreateLinkedTokenSource(parentCancellation);
         BudgetLedger.Initialize(lease);
     }
@@ -193,14 +195,14 @@ public async ValueTask EmitEventAsync(string kind, object body, CancellationToke
     private void BufferEvent(Envelope env)
     {
         // Spec §7.6: bounded per-job history so a later subscriber with `history: true`
-        // can receive prior events in order before live events. Bounded by EventBufferCapacity
-        // to keep server memory predictable.
+        // can receive prior events in order before live events. Sized from
+        // `ArcpServerOptions.EventLogCapacity` so subscribers see the same window resumers do.
         lock (_eventBufferGate)
         {
             _eventBuffer.Add(env);
-            if (_eventBuffer.Count > EventBufferCapacity)
+            if (_eventBuffer.Count > _eventBufferCapacity)
             {
-                _eventBuffer.RemoveRange(0, _eventBuffer.Count - EventBufferCapacity);
+                _eventBuffer.RemoveRange(0, _eventBuffer.Count - _eventBufferCapacity);
             }
         }
     }

src/Arcp.Runtime/JobManager.cs

@@ -35,6 +35,8 @@ public sealed partial class JobManager
     private readonly ILoggerFactory _loggers;
     private readonly CredentialManager? _credentials;
     private readonly int _idempotencyWindowSec;
+    private readonly int _eventBufferCapacity;
+    private readonly int _terminalRetentionSec;
     private readonly bool _fatalBudgetExhaustion;
 
     /// <summary>Stored record for an idempotency key: original submission fingerprint plus issue time.</summary>
@@ -48,6 +50,8 @@ public JobManager(
         ILoggerFactory loggers,
         CredentialManager? credentials = null,
         int idempotencyWindowSec = 3600,
+        int eventBufferCapacity = 4096,
+        int terminalRetentionSec = 600,
         bool fatalBudgetExhaustion = false)
     {
         _agents = agents;
@@ -56,6 +60,8 @@ public JobManager(
         _loggers = loggers;
         _credentials = credentials;
         _idempotencyWindowSec = idempotencyWindowSec > 0 ? idempotencyWindowSec : 3600;
+        _eventBufferCapacity = eventBufferCapacity > 0 ? eventBufferCapacity : 4096;
+        _terminalRetentionSec = terminalRetentionSec;
         _fatalBudgetExhaustion = fatalBudgetExhaustion;
     }
 
@@ -134,7 +140,8 @@ public bool TryGet(JobId id, out Job? job)
             jobId, sessionId, resolved, lease, submit.LeaseConstraints,
             submit.Input, submit.IdempotencyKey, traceId, submit.ParentJobId, submitterPrincipal,
             submit.MaxRuntimeSec,
-            _time.GetUtcNow(), emit, _time, parentCancellation);
+            _time.GetUtcNow(), emit, _time, parentCancellation,
+            eventBufferCapacity: _eventBufferCapacity);
         if (_credentials is not null)
         {
             await _credentials.IssueForJobAsync(job, cancellationToken).ConfigureAwait(false);
@@ -281,9 +288,38 @@ public async Task RunAsync(Job job, IAgent agent, Func<Envelope, CancellationTok
             try { watchdogCts.Cancel(); } catch (ObjectDisposedException) { /* race on dispose */ }
             await AwaitWatchdogAsync(watchdog).ConfigureAwait(false);
             await AwaitWatchdogAsync(runtimeWatchdog).ConfigureAwait(false);
+
+            // Spec §6.6 / hygiene: terminal jobs stay listable for the retention window so
+            // session.list_jobs surfaces them briefly, then are GC'd to bound memory.
+            ScheduleTerminalCleanup(job);
         }
     }
 
+    /// <summary>Schedule removal of a terminal job's bookkeeping after
+    /// <see cref="ArcpServerOptions.TerminalJobRetentionSec"/>. Negative / zero retention disables
+    /// cleanup (legacy "retain forever" behavior).</summary>
+    private void ScheduleTerminalCleanup(Job job)
+    {
+        if (_terminalRetentionSec <= 0) return;
+        _ = Task.Run(async () =>
+        {
+            try
+            {
+                await Task.Delay(TimeSpan.FromSeconds(_terminalRetentionSec), _time).ConfigureAwait(false);
+            }
+            catch (OperationCanceledException) { return; }
+            _jobs.TryRemove(job.JobId, out _);
+            if (job.IdempotencyKey is { } key)
+            {
+                var idemKey = $"{job.SubmitterPrincipal ?? "*"}|{key}";
+                if (_idempotency.TryGetValue(idemKey, out var rec) && rec.JobId.Equals(job.JobId))
+                {
+                    _idempotency.TryRemove(idemKey, out _);
+                }
+            }
+        });
+    }
+
     private Task? StartLeaseWatchdog(Job job, Func<Envelope, CancellationToken, ValueTask> emit, CancellationToken cancellationToken)
     {
         // Spec §9.5.

src/Arcp.Runtime/PublicAPI.Unshipped.txt

@@ -197,14 +197,18 @@ Arcp.Runtime.Credentials.InMemoryCredentialStore.RemoveAsync(Arcp.Core.Ids.JobId
 Arcp.Runtime.Job.Credentials.get -> System.Collections.Generic.IReadOnlyList<Arcp.Core.Messages.ProvisionedCredential!>!
 Arcp.Runtime.JobContext.Credentials.get -> System.Collections.Generic.IReadOnlyList<Arcp.Core.Messages.ProvisionedCredential!>!
 Arcp.Runtime.JobContext.RotateCredentialAsync(string! credentialId, Arcp.Core.Messages.ProvisionedCredential! next, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.ValueTask
-Arcp.Runtime.JobManager.JobManager(Arcp.Runtime.Agents.AgentRegistry! agents, Arcp.Runtime.Leases.LeaseManager! leases, System.TimeProvider! time, Microsoft.Extensions.Logging.ILoggerFactory! loggers, Arcp.Runtime.Credentials.CredentialManager? credentials = null, int idempotencyWindowSec = 3600, bool fatalBudgetExhaustion = false) -> void
+Arcp.Runtime.JobManager.JobManager(Arcp.Runtime.Agents.AgentRegistry! agents, Arcp.Runtime.Leases.LeaseManager! leases, System.TimeProvider! time, Microsoft.Extensions.Logging.ILoggerFactory! loggers, Arcp.Runtime.Credentials.CredentialManager? credentials = null, int idempotencyWindowSec = 3600, int eventBufferCapacity = 4096, int terminalRetentionSec = 600, bool fatalBudgetExhaustion = false) -> void
 Arcp.Runtime.Job.LeaseExpired.get -> bool
 Arcp.Runtime.Job.MaxRuntimeSec.get -> int?
 Arcp.Runtime.Job.RuntimeLimitExceeded.get -> bool
+Arcp.Runtime.ArcpServerOptions.EventLogCapacity.get -> int
+Arcp.Runtime.ArcpServerOptions.EventLogCapacity.init -> void
 Arcp.Runtime.ArcpServerOptions.FatalBudgetExhaustion.get -> bool
 Arcp.Runtime.ArcpServerOptions.FatalBudgetExhaustion.init -> void
 Arcp.Runtime.ArcpServerOptions.IdempotencyWindowSec.get -> int
 Arcp.Runtime.ArcpServerOptions.IdempotencyWindowSec.init -> void
+Arcp.Runtime.ArcpServerOptions.TerminalJobRetentionSec.get -> int
+Arcp.Runtime.ArcpServerOptions.TerminalJobRetentionSec.init -> void
 Arcp.Runtime.JobManager.SubmitAsync(Arcp.Core.Messages.JobSubmitPayload! submit, Arcp.Core.Ids.SessionId sessionId, string? submitterPrincipal, System.Func<Arcp.Core.Wire.Envelope!, System.Threading.CancellationToken, System.Threading.Tasks.ValueTask>! emit, Arcp.Core.Ids.TraceId? inboundTraceId, System.Threading.CancellationToken parentCancellation, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.Task<(Arcp.Runtime.Job! Job, Arcp.Core.Messages.JobAcceptedPayload! Accepted)>!
 Arcp.Runtime.Leases.LeaseManager.AuthorizeModelUse(Arcp.Core.Leases.Lease! lease, Arcp.Core.Leases.LeaseConstraints? constraints, string! modelId) -> void
 override Arcp.Runtime.Credentials.CredentialIssueContext.Equals(object? obj) -> bool

src/Arcp.Runtime/SessionState.cs

@@ -43,8 +43,8 @@ public sealed partial class SessionState : IAsyncDisposable
     /// <summary>Gets the effective features.</summary>
     public IReadOnlyList<string> EffectiveFeatures { get; private set; } = Array.Empty<string>();
 
-    /// <summary>Gets the event log.</summary>
-    public EventLog EventLog { get; private set; } = new();
+    /// <summary>Gets the event log. Sized from <see cref="ArcpServerOptions.EventLogCapacity"/>.</summary>
+    public EventLog EventLog { get; private set; }
 
     /// <summary>Adopt the durable resumable state from a prior session of the same id. Called
     /// when a client supplies a still-valid resume token (spec §6.3).</summary>
@@ -69,6 +69,7 @@ internal SessionState(ITransport transport, ArcpServer server, ArcpServerOptions
             SingleWriter = false,
             FullMode = BoundedChannelFullMode.Wait,
         });
+        EventLog = new EventLog(options.EventLogCapacity);
         _cts = CancellationTokenSource.CreateLinkedTokenSource(cancellation);
         _lastInboundAt = options.TimeProvider.GetUtcNow();
     }