fix: spec-conformance, concurrency, and security audit fixes (#37, #38, #39, #40, #41, #42, #43, #45, #46, #47, #67, #68)

Runtime / spec-conformance:
- #37 root running jobs at a runtime-scoped token so session teardown (heartbeat loss,
  graceful close, transport drop) no longer terminates in-flight jobs (spec §6.4, §6.7)
- #41 add session.close/session.closed wire types; the runtime acks a graceful close with
  session.closed (session.bye kept as a deprecated alias) (spec §6.7)
- #40 dispatch now surfaces session.error{INTERNAL_ERROR} for unexpected exceptions (spec §12)
- #46 advertise model.use independently of credential provisioning (spec §9.7)

Event delivery / ordering:
- #39 serialize event_seq assignment with the outbound enqueue via a per-session emit gate so
  wire order is strictly monotonic under concurrent emitters (spec §8.3)
- #38 subscriber fan-out is back-pressure-aware: on a full channel the subscription is torn down
  deterministically instead of silently dropping an already-sequenced event (spec §8.3)
- #44 make the subscribe history/live-fan-out boundary exact via an atomic register+snapshot and a
  per-job event index, so a mid-stream subscriber sees each event exactly once (spec §7.6)

Authorization / security:
- #42 gate lease operations on remaining budget (BUDGET_EXHAUSTED) before the pattern check (spec §9.6)
- #43 deny-by-default for uncovered tool.call/agent.delegate, with an explicit
  PermissiveUnleasedOperations opt-in (spec §9.3)
- #45 list_jobs fails closed: an empty/absent principal sees only what the policy permits (spec §6.6, §14)

Performance / correctness:
- #67 keyset pagination over (created_at, job_id): stable ties and page work bounded to limit+1
- #68 AgentRegistry no longer exposes its mutable version dictionary; ToInventory snapshots under lock
- #47 client detects event_seq gaps and raises a broken-session signal (spec §8.3)

Co-authored-by: Cursor <cursoragent@cursor.com>

Author: nficano

src/Arcp.Client/ArcpClient.Dispatch.cs

@@ -18,7 +18,15 @@ private async Task ReaderLoop(CancellationToken cancellationToken)
         {
             await foreach (var env in _transport.ReceiveAsync(cancellationToken).ConfigureAwait(false))
             {
-                if (env.EventSeq is { } seq) Interlocked.Exchange(ref _lastReceivedSeq, seq);
+                if (env.EventSeq is { } seq)
+                {
+                    // Spec §8.3: event_seq is strictly monotonic and gap-free. If the new seq skips
+                    // the expected successor, surface a detectable broken-session signal instead of
+                    // silently accepting the gap.
+                    var prev = Interlocked.Read(ref _lastReceivedSeq);
+                    if (prev > 0 && seq > prev + 1) OnEventSeqGap(prev + 1, seq);
+                    if (seq > prev) Interlocked.Exchange(ref _lastReceivedSeq, seq);
+                }
                 await DispatchAsync(env, cancellationToken).ConfigureAwait(false);
             }
         }

src/Arcp.Client/ArcpClient.cs

@@ -53,6 +53,22 @@ public sealed partial class ArcpClient : IAsyncDisposable
     /// <summary>Gets the last received seq.</summary>
     public long LastReceivedSeq => Interlocked.Read(ref _lastReceivedSeq);
 
+    /// <summary>True once an inbound <c>event_seq</c> has skipped the expected next value, indicating
+    /// the session stream has a gap and SHOULD be treated as broken (and resumed once resume is
+    /// wired) per spec §8.3.</summary>
+    public bool IsSessionBroken { get; private set; }
+
+    /// <summary>Raised when an inbound <c>event_seq</c> skips the expected successor (spec §8.3). The
+    /// arguments are <c>(expectedSeq, receivedSeq)</c>. Handlers run on the reader loop; keep them
+    /// fast and non-throwing.</summary>
+    public event Action<long, long>? EventSeqGapDetected;
+
+    private void OnEventSeqGap(long expected, long received)
+    {
+        IsSessionBroken = true;
+        EventSeqGapDetected?.Invoke(expected, received);
+    }
+
     /// <summary>Initializes a new instance of the <see cref="ArcpClient"/> class.</summary>
     public ArcpClient(ITransport transport, ArcpClientOptions options)
     {

src/Arcp.Client/PublicAPI.Unshipped.txt

@@ -7,7 +7,9 @@ Arcp.Client.ArcpClient.CancelJobAsync(Arcp.Core.Ids.JobId jobId, string? reason
 Arcp.Client.ArcpClient.ConnectAsync(System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.Task!
 Arcp.Client.ArcpClient.DisposeAsync() -> System.Threading.Tasks.ValueTask
 Arcp.Client.ArcpClient.EffectiveFeatures.get -> System.Collections.Generic.IReadOnlyList<string!>!
+Arcp.Client.ArcpClient.EventSeqGapDetected -> System.Action<long, long>?
 Arcp.Client.ArcpClient.HeartbeatIntervalSec.get -> int?
+Arcp.Client.ArcpClient.IsSessionBroken.get -> bool
 Arcp.Client.ArcpClient.LastReceivedSeq.get -> long
 Arcp.Client.ArcpClient.ListJobsAsync(Arcp.Core.Messages.JobListFilter? filter = null, int? limit = null, string? cursor = null, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.Task<Arcp.Core.Messages.SessionJobsPayload!>!
 Arcp.Client.ArcpClient.ResumeToken.get -> string?

src/Arcp.Core/Envelope/Envelope.cs

@@ -60,4 +60,11 @@ public sealed record Envelope
     /// unknown fields") so they round-trip without loss.</summary>
     [JsonExtensionData]
     public IDictionary<string, JsonElement>? Extensions { get; init; }
+
+    /// <summary>Runtime-only, NON-serialized per-job monotonic index. Used by the runtime to make the
+    /// <c>job.subscribe</c> history/live-fan-out boundary exact (spec §7.6) — a subscriber drops a
+    /// fanned-out event whose index was already covered by its replayed history. Never transmitted
+    /// on the wire.</summary>
+    [JsonIgnore]
+    public long? JobEventIndex { get; init; }
 }

src/Arcp.Core/Envelope/MessageTypeRegistry.cs

@@ -41,6 +41,8 @@ public static MessageTypeRegistry CreateCoreCatalog()
         var r = new MessageTypeRegistry();
         r.Register(MessageTypeNames.SessionHello, typeof(SessionHelloPayload));
         r.Register(MessageTypeNames.SessionWelcome, typeof(SessionWelcomePayload));
+        r.Register(MessageTypeNames.SessionClose, typeof(SessionByePayload));
+        r.Register(MessageTypeNames.SessionClosed, typeof(SessionByePayload));
         r.Register(MessageTypeNames.SessionBye, typeof(SessionByePayload));
         r.Register(MessageTypeNames.SessionPing, typeof(SessionPingPayload));
         r.Register(MessageTypeNames.SessionPong, typeof(SessionPongPayload));

src/Arcp.Core/Messages/MessageTypeNames.cs

@@ -16,7 +16,12 @@ public static class MessageTypeNames
     public const string SessionHello = "session.hello";
     /// <summary>Gets the session welcome.</summary>
     public const string SessionWelcome = "session.welcome";
-    /// <summary>Gets the session bye.</summary>
+    /// <summary>Client-sent graceful session close (spec §6.7).</summary>
+    public const string SessionClose = "session.close";
+    /// <summary>Runtime-sent acknowledgement of <see cref="SessionClose"/> (spec §6.7).</summary>
+    public const string SessionClosed = "session.closed";
+    /// <summary>Gets the session bye. Deprecated alias for <see cref="SessionClose"/> kept for
+    /// back-compat with pre-1.1 peers; the runtime treats it like <c>session.close</c> (spec §6.7).</summary>
     public const string SessionBye = "session.bye";
     /// <summary>Gets the session ping.</summary>
     public const string SessionPing = "session.ping";
@@ -63,7 +68,7 @@ public static class MessageTypeNames
     /// <summary>Gets the all.</summary>
     public static readonly FrozenSet<string> All = new HashSet<string>(StringComparer.Ordinal)
     {
-        SessionHello, SessionWelcome, SessionBye, SessionPing, SessionPong, SessionAck,
+        SessionHello, SessionWelcome, SessionClose, SessionClosed, SessionBye, SessionPing, SessionPong, SessionAck,
         SessionListJobs, SessionJobs, SessionError, SessionResume,
         JobSubmit, JobAccepted, JobEvent, JobResult, JobError, JobCancel, JobCancelled,
         JobSubscribe, JobSubscribed, JobUnsubscribe,

src/Arcp.Core/PublicAPI.Unshipped.txt

@@ -651,6 +651,8 @@ Arcp.Core.Wire.Envelope.Extensions.get -> System.Collections.Generic.IDictionary
 Arcp.Core.Wire.Envelope.Extensions.init -> void
 Arcp.Core.Wire.Envelope.Id.get -> string!
 Arcp.Core.Wire.Envelope.Id.init -> void
+Arcp.Core.Wire.Envelope.JobEventIndex.get -> long?
+Arcp.Core.Wire.Envelope.JobEventIndex.init -> void
 Arcp.Core.Wire.Envelope.JobId.get -> string?
 Arcp.Core.Wire.Envelope.JobId.init -> void
 Arcp.Core.Wire.Envelope.ParentSpanId.get -> string?
@@ -730,6 +732,8 @@ const Arcp.Core.Messages.MessageTypeNames.JobSubscribed = "job.subscribed" -> st
 const Arcp.Core.Messages.MessageTypeNames.JobUnsubscribe = "job.unsubscribe" -> string!
 const Arcp.Core.Messages.MessageTypeNames.SessionAck = "session.ack" -> string!
 const Arcp.Core.Messages.MessageTypeNames.SessionBye = "session.bye" -> string!
+const Arcp.Core.Messages.MessageTypeNames.SessionClose = "session.close" -> string!
+const Arcp.Core.Messages.MessageTypeNames.SessionClosed = "session.closed" -> string!
 const Arcp.Core.Messages.MessageTypeNames.SessionError = "session.error" -> string!
 const Arcp.Core.Messages.MessageTypeNames.SessionHello = "session.hello" -> string!
 const Arcp.Core.Messages.MessageTypeNames.SessionJobs = "session.jobs" -> string!

src/Arcp.Runtime/Agents/AgentRegistry.cs

@@ -63,21 +63,24 @@ public void SetDefaultVersion(string name, string version)
 
     /// <summary>To inventory.</summary>
     public IReadOnlyList<AgentInventoryEntry> ToInventory() =>
-        _byName.Values.Select(e => new AgentInventoryEntry
+        _byName.Values.Select(e =>
         {
-            Name = e.Name,
-            Versions = e.Versions.Count > 0 ? e.Versions.Keys.ToArray() : null,
-            Default = e.DefaultVersion,
+            var (versions, defaultVersion) = e.SnapshotInventoryFields();
+            return new AgentInventoryEntry
+            {
+                Name = e.Name,
+                Versions = versions,
+                Default = defaultVersion,
+            };
         }).ToArray();
 
     private sealed class AgentEntry
     {
         private readonly object _gate = new();
+        private readonly Dictionary<string, IAgent> _versions = new(StringComparer.Ordinal);
 
         public string Name { get; }
 
-        public Dictionary<string, IAgent> Versions { get; } = new(StringComparer.Ordinal);
-
         public string? DefaultVersion { get; private set; }
 
         public IAgent? UnversionedAgent { get; private set; }
@@ -86,14 +89,14 @@ private sealed class AgentEntry
 
         public bool TryGetVersion(string v, out IAgent? agent)
         {
-            lock (_gate) return Versions.TryGetValue(v, out agent);
+            lock (_gate) return _versions.TryGetValue(v, out agent);
         }
 
         public void AddVersion(string version, IAgent agent)
         {
             lock (_gate)
             {
-                Versions[version] = agent;
+                _versions[version] = agent;
                 DefaultVersion ??= version;
             }
         }
@@ -107,7 +110,7 @@ public void SetDefault(string version)
         {
             lock (_gate)
             {
-                if (!Versions.ContainsKey(version))
+                if (!_versions.ContainsKey(version))
                     throw new AgentVersionNotAvailableException($"{Name}@{version} not registered");
                 DefaultVersion = version;
             }
@@ -119,10 +122,22 @@ public void SetDefault(string version)
             {
                 lock (_gate)
                 {
-                    foreach (var kv in Versions) return (kv.Key, kv.Value);
+                    foreach (var kv in _versions) return (kv.Key, kv.Value);
                     return null;
                 }
             }
         }
+
+        /// <summary>Atomically snapshot the inventory-visible fields under the same lock that guards
+        /// mutation, so <see cref="ToInventory"/> never enumerates the mutable dictionary while a
+        /// concurrent registration is writing to it.</summary>
+        public (IReadOnlyList<string>? Versions, string? Default) SnapshotInventoryFields()
+        {
+            lock (_gate)
+            {
+                var versions = _versions.Count > 0 ? _versions.Keys.ToArray() : null;
+                return (versions, DefaultVersion);
+            }
+        }
     }
 }

src/Arcp.Runtime/ArcpServer.cs

@@ -75,7 +75,8 @@ public ArcpServer(ArcpServerOptions options, ILoggerFactory? loggerFactory = nul
             options.IdempotencyWindowSec,
             options.EventLogCapacity,
             options.TerminalJobRetentionSec,
-            options.FatalBudgetExhaustion);
+            options.FatalBudgetExhaustion,
+            options.PermissiveUnleasedOperations);
         if (CredentialManager is not null)
         {
             _ = Task.Run(() => RevokeOutstandingCredentialsAsync(CancellationToken.None));
@@ -213,6 +214,9 @@ internal bool TryResume(string resumeToken, out SessionState? session)
     /// <summary>Dispose (asynchronous).</summary>
     public async ValueTask DisposeAsync()
     {
+        // Cancel in-flight jobs only at runtime shutdown — never on individual session teardown
+        // (spec §6.4, §6.7). Jobs are rooted at the JobManager's runtime token, not session _cts.
+        JobManager.Dispose();
         try { _sweeperCts.Cancel(); } catch (ObjectDisposedException) { /* already disposed */ }
         if (_sweeperTask is not null)
         {
@@ -240,9 +244,12 @@ private static IReadOnlyList<string> ComputeAdvertisedFeatures(ArcpServerOptions
                 nameof(options));
         }
 
+        // Spec §9.7: model.use enforcement is independent of credential provisioning —
+        // LeaseManager.AuthorizeModelUse enforces it directly, so it MUST stay advertisable even
+        // when no ICredentialProvisioner is configured. Only provisioned_credentials is coupled to
+        // having a provisioner.
         return requested
-            .Where(f => !string.Equals(f, FeatureFlags.ProvisionedCredentials, StringComparison.Ordinal) &&
-                        !string.Equals(f, FeatureFlags.ModelUse, StringComparison.Ordinal))
+            .Where(f => !string.Equals(f, FeatureFlags.ProvisionedCredentials, StringComparison.Ordinal))
             .ToArray();
     }
 

src/Arcp.Runtime/ArcpServerOptions.cs

@@ -50,6 +50,12 @@ public sealed class ArcpServerOptions
     /// <summary>Gets the back pressure threshold.</summary>
     public int BackPressureThreshold { get; init; } = 1000;
 
+    /// <summary>When <see langword="false"/> (the default, spec §9.1/§9.3 deny-by-default), an
+    /// authority-bearing operation (<c>tool.call</c>, <c>agent.delegate</c>) whose namespace the
+    /// job's lease does not declare fails with <c>PERMISSION_DENIED</c>. Set to <see langword="true"/>
+    /// to opt into the legacy permissive behavior where uncovered namespaces are allowed.</summary>
+    public bool PermissiveUnleasedOperations { get; init; }
+
     /// <summary>Gets the authorization policy.</summary>
     public IJobAuthorizationPolicy AuthorizationPolicy { get; init; } = new SamePrincipalPolicy();