fix(runtime): agent_versions gating, chunk/progress/metric validation, revocation surfacing

- reject name@version without agent_versions; strip suffix from non-negotiated sessions (#79)
- enforce chunk_seq monotonicity and single-stream completion (#84)
- progress current>=0 (reject) and current<=total (clamp) (#85)
- negative cost metrics rejected; negative non-cost metrics still emit (#86)
- permanent credential revocation failures logged + exposed via RevocationFailures (#87)

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

Author: Nick Ficano

src/Arcp.Runtime/Internal/CredentialRegistry.fs

@@ -13,6 +13,10 @@ type internal CredentialRegistry(provisioner: ICredentialProvisioner, store: ICr
     let perJob = ConcurrentDictionary<string, ConcurrentDictionary<string, unit>>()
     let retryDelays = [ 200; 1000; 5000 ]
 
+    // §9.8.2/§14: credentials whose revocation permanently failed, kept
+    // for operator inspection (`RevocationFailures`).
+    let revocationFailures = ConcurrentDictionary<string, string>()
+
     let remember (jobId: JobId) (credentialId: string) =
         let ids =
             perJob.GetOrAdd(jobId.Value, fun _ -> ConcurrentDictionary<string, unit>())
@@ -43,6 +47,8 @@ type internal CredentialRegistry(provisioner: ICredentialProvisioner, store: ICr
                     attempt <- attempt + 1
 
             if revoked then
+                revocationFailures.TryRemove credentialId |> ignore
+
                 match jobIdOpt with
                 | Some jobId ->
                     do! store.RecordRevokedAsync(jobId, credentialId)
@@ -54,8 +60,26 @@ type internal CredentialRegistry(provisioner: ICredentialProvisioner, store: ICr
                         if id = credentialId then
                             do! store.RecordRevokedAsync(jobId, credentialId)
                             forget jobId credentialId
+            else
+                // §9.8.2: permanent revocation failures MUST be logged; §14:
+                // surfaced to operators. Record for `RevocationFailures`.
+                let jobLabel =
+                    jobIdOpt |> Option.map (fun j -> j.Value) |> Option.defaultValue "<unknown>"
+
+                revocationFailures.[credentialId] <- jobLabel
+
+                eprintfn
+                    "[ARCP] WARN credential revocation failed permanently: credential_id=%s job_id=%s"
+                    credentialId
+                    jobLabel
         }
 
+    /// Credentials whose revocation permanently failed, as
+    /// `(credentialId, jobId)` pairs. Operators can poll this to find
+    /// dangling credentials (§9.8.2, §14).
+    member _.RevocationFailures: (string * string) list =
+        revocationFailures |> Seq.map (fun kv -> kv.Key, kv.Value) |> List.ofSeq
+
     member _.Track(jobId: JobId, cred: Credential) : Task =
         task {
             remember jobId cred.Id

src/Arcp.Runtime/Internal/JobSubmitFlow.fs

@@ -225,10 +225,30 @@ module internal JobSubmitFlow =
                         do! sendAccepted ctx.Transport ctx.SessionId requestId (JobId.ofString existing) accepted ct
                     | Error err -> do! EnvelopeOut.respondWithError ctx requestId err ct
             | _ ->
+                let agentVersionsOn = ctx.NegotiatedFeatures.Contains Features.AgentVersions
+
+                if not agentVersionsOn && submit.Agent.Contains '@' then
+                    // §6.2/§7.5: cannot pin a version without negotiating
+                    // agent_versions.
+                    do!
+                        EnvelopeOut.respondWithError
+                            ctx
+                            requestId
+                            (ARCPError.InvalidRequest("agent_versions not negotiated; bare agent name required", None))
+                            ct
+                else
+
                 match inventory.Resolve submit.Agent with
                 | Error err -> do! EnvelopeOut.respondWithError ctx requestId err ct
                 | Ok(name, version, _) ->
-                    let resolvedAgent = AgentRef.format name (Some version)
+                    // The handler is always keyed name@version; the agent
+                    // surfaced on the wire omits the version when the feature
+                    // wasn't negotiated (§6.2).
+                    let handlerKey = AgentRef.format name (Some version)
+
+                    let resolvedAgent =
+                        if agentVersionsOn then handlerKey else name
+
                     let lease = submit.LeaseRequest |> Option.defaultValue Lease.empty
 
                     match validateConstraints timeProvider submit.LeaseConstraints with
@@ -344,7 +364,7 @@ module internal JobSubmitFlow =
 
                                 do! sendAccepted ctx.Transport ctx.SessionId requestId jobId accepted ct
 
-                                match agentHandlers.TryGetValue resolvedAgent with
+                                match agentHandlers.TryGetValue handlerKey with
                                 | true, handler ->
                                     JobLauncher.launch jobs credentialRegistry timeProvider record handler
                                 | _ ->

src/Arcp.Runtime/JobContext.fs

@@ -32,6 +32,12 @@ type JobContext
         streamResultBegin: unit -> ResultId,
         onCostMetric: string * decimal -> unit
     ) =
+    // Per-result_id chunk ordering state (§8.4): next expected chunk_seq
+    // and whether the stream was closed by a `more=false` chunk.
+    let chunkNext = System.Collections.Generic.Dictionary<string, int64>()
+    let chunkClosed = System.Collections.Generic.HashSet<string>()
+    let chunkLock = obj ()
+
     member _.JobId: JobId = jobId
     member _.SessionId: SessionId = sessionId
     member _.ParentJobId: JobId option = parentJobId
@@ -63,10 +69,21 @@ type JobContext
     member _.RotateCredentialAsync(credentialId: string, newValue: string, ct: CancellationToken) : Task =
         rotateCredential (credentialId, newValue, ct)
 
+    /// Emit a `progress` event (§8.2.1). `current` MUST be non-negative
+    /// (rejected with INVALID_REQUEST otherwise); when `total` is present
+    /// `current` is clamped to `total` so the wire invariant holds.
     member _.EmitProgressAsync
         (current: decimal, total: decimal option, units: string option, message: string option, _ct: CancellationToken)
         : Task =
-        emit (JobEventBody.Progress(current, total, units, message))
+        if current < 0m then
+            raise (ArcpException(ARCPError.InvalidRequest("progress.current must be non-negative", None)))
+
+        let clamped =
+            match total with
+            | Some t when current > t -> t
+            | _ -> current
+
+        emit (JobEventBody.Progress(clamped, total, units, message))
 
     /// Emit a `metric` event. Names starting with `cost.` and a
     /// budgeted `unit` decrement the matching budget counter
@@ -80,7 +97,12 @@ type JobContext
             _ct: CancellationToken
         ) : Task =
         if value < 0m then
-            Task.CompletedTask
+            // §9.6: negative cost metrics are rejected; other negative
+            // metrics are not governed by §9.6 and still flow through.
+            if name.StartsWith("cost.") then
+                raise (ArcpException(ARCPError.InvalidRequest("cost metric value must be non-negative", None)))
+            else
+                emit (JobEventBody.Metric(name, value, unit, dimensions))
         else
             // §9.6: `cost.budget.*` is budget telemetry (e.g.
             // `cost.budget.remaining`), not a charge — it must not
@@ -135,6 +157,36 @@ type JobContext
             more: bool,
             _ct: CancellationToken
         ) : Task =
+        // §8.4: chunk_seq is 0-based monotonic per result_id and chunks
+        // MUST be emitted in order; nothing may follow a `more=false`
+        // chunk. Enforce both before anything reaches the wire.
+        lock chunkLock (fun () ->
+            if chunkClosed.Contains resultId.Value then
+                raise (
+                    ArcpException(
+                        ARCPError.InternalError(sprintf "result_id %s already completed; no further chunks allowed" resultId.Value)
+                    )
+                )
+
+            let expected =
+                match chunkNext.TryGetValue resultId.Value with
+                | true, n -> n
+                | _ -> 0L
+
+            if chunkSeq <> expected then
+                raise (
+                    ArcpException(
+                        ARCPError.InternalError(
+                            sprintf "out-of-order chunk_seq %d (expected %d) for result_id %s" chunkSeq expected resultId.Value
+                        )
+                    )
+                )
+
+            chunkNext.[resultId.Value] <- expected + 1L
+
+            if not more then
+                chunkClosed.Add resultId.Value |> ignore)
+
         let encoded =
             match encoding with
             | ChunkEncoding.Utf8 -> Encoding.UTF8.GetString(data.Span)

tests/Arcp.IntegrationTests/IdempotencyAndCancelTests.fs

@@ -57,10 +57,12 @@ let ``list_jobs filter by agent narrows the set`` () =
         let! _ = p.Client.SubmitAsync(mkRequest "b", CancellationToken.None)
         let! _ = p.Client.SubmitAsync(mkRequest "b", CancellationToken.None)
 
+        // agent_versions not negotiated -> agents are stored bare; the
+        // bare-name filter matches (§79, §108).
         let filter: JobListFilter =
             {
                 Status = None
-                Agent = Some "a@default"
+                Agent = Some "a"
                 CreatedAfter = None
             }
 

tests/Arcp.UnitTests/JobContextTests.fs

@@ -90,12 +90,24 @@ let ``EmitMetric with positive cost.* decrements budget tracking`` () =
     costs.[0] |> should equal ("USD", 0.25m)
 
 [<Fact>]
-let ``EmitMetric with negative value is silently dropped`` () =
+let ``EmitMetric negative cost value is rejected`` () =
+    // §86: negative cost.* metrics raise INVALID_REQUEST.
     let ctx, emitted, _, costs, _ = mkContext Lease.empty
-    ctx.EmitMetricAsync("anything", -1m, None, None, CancellationToken.None).Wait()
+
+    (fun () -> ctx.EmitMetricAsync("cost.inference", -0.01m, Some "USD", None, CancellationToken.None) |> ignore)
+    |> should throw typeof<ArcpException>
+
     emitted.Count |> should equal 0
     costs.Count |> should equal 0
 
+[<Fact>]
+let ``EmitMetric negative non-cost value still emits`` () =
+    // §86: §9.6 only governs cost metrics; negative non-cost metrics flow.
+    let ctx, emitted, _, costs, _ = mkContext Lease.empty
+    ctx.EmitMetricAsync("latency.ms", -5m, Some "ms", None, CancellationToken.None).Wait()
+    emitted.Count |> should equal 1
+    costs.Count |> should equal 0
+
 [<Fact>]
 let ``EmitMetric non-cost name does not touch budget`` () =
     let ctx, emitted, _, costs, _ = mkContext Lease.empty