From 9debdacffbc73b4c67eeb21411a17a663f6b771a Mon Sep 17 00:00:00 2001 From: JNK234 Date: Tue, 8 Sep 2026 21:41:58 -0400 Subject: [PATCH 1/2] feat: named profiles so different agents call different models Provider and model were one global configuration shared by every agent, and swapping it per agent broke under async calls. A profile is a second config file loaded under a name with its own provider, key and model; any agent can be bound to one and its calls go through that profile's provider. Unbound agents use the global configuration exactly as before. New primitives: llm:load-profile name file (validated like load-config; a rejected reload leaves the old profile intact), llm:use-profile name (the reserved name "default" unbinds), llm:profile, llm:profiles. llm:active and llm:config report the calling agent's effective configuration. Each profile caches one provider instance, created on first use, so an async call keeps the provider it started with when the agent is rebound. Profiles survive clear-all like the global config; bindings do not. Setters (set-model, set-thinking, ...) keep acting on the default config. Tests: ProfileStoreSpec (11) covers load/replace/reserved names/caching; 14 tests.txt cases cover routing, async capture, reload, failed reload, clear-all, and that history and usage stay per agent, via a __TEST_ECHO_MODEL marker. T10 in the live harness binds a turtle to a second Groq model: 41/41 pass. Closes #68 --- demos/e2e-tests/e2e-tests.nlogox | 65 ++++++++ demos/test-profile-a | 4 + demos/test-profile-b | 4 + docs/API-REFERENCE.md | 90 +++++++++++ docs/CONFIGURATION.md | 20 +++ src/main/LLMExtension.scala | 198 +++++++++++++++++++---- src/main/config/ProfileStore.scala | 83 ++++++++++ src/test/DeterministicTestProvider.scala | 4 + src/test/ProfileStoreSpec.scala | 122 ++++++++++++++ tests.txt | 169 +++++++++++++++++++ 10 files changed, 726 insertions(+), 33 deletions(-) create mode 100644 demos/test-profile-a create mode 100644 demos/test-profile-b create mode 100644 src/main/config/ProfileStore.scala create mode 100644 src/test/ProfileStoreSpec.scala diff --git a/demos/e2e-tests/e2e-tests.nlogox b/demos/e2e-tests/e2e-tests.nlogox index 2473c7c..e17c639 100644 --- a/demos/e2e-tests/e2e-tests.nlogox +++ b/demos/e2e-tests/e2e-tests.nlogox @@ -303,6 +303,70 @@ to test-usage report-totals end +;; --------------------------------------------------------------------------- +;; T10 — per-agent profiles (#68) +;; +;; A second configuration is derived from the active one with a different +;; model, loaded as a profile, and bound to one turtle. That turtle must +;; report and call the alternate model while the observer stays on the +;; default, and the real reply must come back. +;; --------------------------------------------------------------------------- + +to test-profiles + log-line "T10 per-agent profiles" + let default-model item 1 llm:active + let alt alt-model + let alt-config (word "config-" provider-under-test "-alt.txt") + write-config-with-model active-config alt-config alt + carefully + [ llm:load-profile "alt" alt-config ] + [ log-line (word " load-profile failed: " error-message) ] + + crt 1 [ set label "profile-turtle" ] + let t one-of turtles with [label = "profile-turtle"] + ask t [ llm:use-profile "alt" ] + + let reply "" + carefully + [ set reply [llm:chat "Reply with exactly the word: OK"] of t ] + [ set reply (word "ERROR: " error-message) ] + log-line (word " turtle on " [llm:active] of t " -> " reply) + + assert "profile is listed" (member? "alt" llm:profiles) + assert "turtle reports its profile" ([llm:profile] of t = "alt") + assert "turtle's active model is the profile's model" ([item 1 llm:active] of t = alt) + assert "observer stays on the default model" (item 1 llm:active = default-model) + assert "observer reports the default profile" (llm:profile = "default") + assert "call through the profile returned" (is-string? reply and not member? "ERROR:" reply) + assert "usage was credited to the turtle" ([llm:get llm:usage "calls"] of t = 1) + + ask t [ die ] + carefully [ file-delete alt-config ] [ ] + report-totals +end + +;; A second model on the provider under test, so a profile can be told apart +;; from the default. Ollama has no guaranteed second model pulled, so it +;; reuses the active one and the test still proves routing and bookkeeping. +to-report alt-model + if provider-under-test = "groq" [ report "openai/gpt-oss-120b" ] + if provider-under-test = "anthropic" [ report "claude-3-5-haiku-latest" ] + report item 1 llm:active +end + +;; Copy a config file with its model line replaced. +to write-config-with-model [src dst model-name] + let lines [] + file-open src + while [ not file-at-end? ] [ set lines lput file-read-line lines ] + file-close + carefully [ file-delete dst ] [ ] + file-open dst + foreach filter [ l -> not (length l >= 6 and substring l 0 6 = "model=") ] lines [ l -> file-print l ] + file-print (word "model=" model-name) + file-close +end + ;; Run everything in sequence. to run-headless carefully [ file-delete "e2e-results.txt" ] [ ] @@ -326,6 +390,7 @@ to test-all test-choose test-throttling test-usage + test-profiles output-print "" log-line (word "TOTAL passed " pass-count " failed " fail-count) end diff --git a/demos/test-profile-a b/demos/test-profile-a new file mode 100644 index 0000000..a0a2dce --- /dev/null +++ b/demos/test-profile-a @@ -0,0 +1,4 @@ +# Test fixture: profile A for per-agent model selection tests +provider=openai +openai_api_key=test-key +model=model-a diff --git a/demos/test-profile-b b/demos/test-profile-b new file mode 100644 index 0000000..c1a502a --- /dev/null +++ b/demos/test-profile-b @@ -0,0 +1,4 @@ +# Test fixture: profile B for per-agent model selection tests +provider=openai +openai_api_key=test-key +model=model-b diff --git a/docs/API-REFERENCE.md b/docs/API-REFERENCE.md index c8c3b6f..8fd3ba5 100644 --- a/docs/API-REFERENCE.md +++ b/docs/API-REFERENCE.md @@ -26,6 +26,10 @@ The NetLogo Multi-LLM Extension provides a unified interface for multiple Large | `llm:set-history list` | History | Set conversation history for current agent | | `llm:clear-history` | History | Clear conversation history for current agent | | `llm:load-config filename` | Configuration | Load settings from file | +| `llm:load-profile name filename` | Profiles | Load a second configuration under a name | +| `llm:use-profile name` | Profiles | Route the calling agent's calls through that profile | +| `llm:profile` | Profiles | The calling agent's profile name (`"default"` if none) | +| `llm:profiles` | Profiles | Names of the loaded profiles | | `llm:set-provider name` | Configuration | Set active provider (openai, anthropic, gemini, ollama, openrouter, together) | | `llm:set-api-key key` | Configuration | Set API key for current provider | | `llm:set-model name` | Configuration | Set model to use for current provider | @@ -860,6 +864,92 @@ print llm:list-models ; Shows all providers, with Anthropic marked as ACTIVE - Custom models added via `models-override.yaml` are marked with `[custom]` - The currently active provider and model are marked with `[ACTIVE]` +## Profiles: different models for different agents + +By default every agent shares one configuration: one provider, one key, one +model. A profile is a second configuration loaded under a name. Any agent can +be bound to a profile, and from then on its calls go through that profile's +provider and model. Agents that are not bound keep using the default +configuration exactly as before. + +```netlogo +to setup + clear-all + llm:load-config "config.txt" ;; the default, as before + llm:load-profile "llama" "config-groq.txt" ;; a second provider and key + llm:load-profile "sonnet" "config-anthropic.txt" + llm:load-profile "local" "config-ollama.txt" + + create-turtles 30 + ask turtles with [who mod 3 = 0] [ llm:use-profile "llama" ] + ask turtles with [who mod 3 = 1] [ llm:use-profile "sonnet" ] + ask turtles with [who mod 3 = 2] [ llm:use-profile "local" ] +end + +to go + ask turtles [ + let reply llm:chat "One word: forage or rest?" ;; each goes to its own model + ] +end +``` + +Each turtle keeps its own history and its own `llm:usage`, so three models can +be compared inside one run under identical conditions. + +### llm:load-profile + +**Syntax**: `llm:load-profile name filename` + +**Description**: Loads a configuration file under a name. The file has the same +format and keys as `llm:load-config`, including thinking, retry and throttling +settings, and is validated the same way: an unknown provider, a missing key, or +an unreachable local server rejects the load. Loading a name that already +exists replaces it. A rejected reload leaves the existing profile untouched. + +**Parameters**: + +- `name` (string): Any name except `"default"`, which is reserved for the + global configuration. Names are case-insensitive. +- `filename` (string): Config file, resolved like `llm:load-config` + +**Notes**: + +- Profiles survive `clear-all`, like the global configuration does. Agent + bindings do not, since the agents themselves are gone. +- `llm:set-model`, `llm:set-provider`, `llm:set-api-key` and the thinking + setters act on the default configuration only. A profile is what its file + says. To change a profile, edit the file and load it again. +- Two profiles on the same provider and endpoint share one request throttle. + +### llm:use-profile + +**Syntax**: `llm:use-profile name` + +**Description**: Routes the calling agent's calls through the named profile. +`llm:use-profile "default"` returns the agent to the global configuration. +An unknown name is an error that lists the loaded profiles. + +**Notes**: + +- An async call already in flight keeps the provider it started with. Rebinding + an agent affects its next call, never a pending one. +- Any agent can be bound: turtles, patches, links, or the observer. + +### llm:profile + +**Syntax**: `llm:profile` + +**Returns**: String - the calling agent's profile name, or `"default"` + +### llm:profiles + +**Syntax**: `llm:profiles` + +**Returns**: List - the loaded profile names, sorted + +With a profile bound, `llm:active` and `llm:config` report the calling agent's +effective provider, model and configuration rather than the global ones. + ## Token Usage Every provider reports how many tokens a call consumed. The extension records diff --git a/docs/CONFIGURATION.md b/docs/CONFIGURATION.md index 210ef36..29dd18e 100644 --- a/docs/CONFIGURATION.md +++ b/docs/CONFIGURATION.md @@ -125,6 +125,26 @@ model=gpt-4o-mini # llm:set-model "claude-3-5-sonnet-20241022" ``` +### Profiles: one config file per model, different agents on different models + +`llm:load-config` sets the one configuration every agent shares. To run several +models in one simulation, load extra config files as named profiles and bind +agents to them. Each profile is an ordinary config file with its own provider, +key and model: + +``` +llm:load-config "config.txt" ;; default for everyone +llm:load-profile "llama" "config-groq.txt" +llm:load-profile "sonnet" "config-anthropic.txt" +ask turtles with [role = "scout"] [ llm:use-profile "llama" ] +ask turtles with [role = "leader"] [ llm:use-profile "sonnet" ] +``` + +A profile file is validated on load the same way as `config.txt`. Thinking, +retry and throttling keys in the file apply to that profile. See the API +reference for `llm:load-profile`, `llm:use-profile`, `llm:profile` and +`llm:profiles`. + ### Request Throttling (staying inside a rate limit) A model that calls the LLM once per agent per tick sends one request per agent simultaneously. On a free tier that exceeds the quota on the first tick. diff --git a/src/main/LLMExtension.scala b/src/main/LLMExtension.scala index bac7de5..c8237bf 100644 --- a/src/main/LLMExtension.scala +++ b/src/main/LLMExtension.scala @@ -2,7 +2,7 @@ package org.nlogo.extensions.llm import org.nlogo.api._ import org.nlogo.core.{LogoList, Syntax} -import org.nlogo.extensions.llm.config.{ConfigLoader, ConfigStore} +import org.nlogo.extensions.llm.config.{ConfigLoader, ConfigStore, ProfileStore} import org.nlogo.extensions.llm.providers.{LLMProvider, ProviderDescriptor, ProviderFactory, ProviderRegistry, ProviderRegistrations, ModelRegistry, OllamaProvider, ReadinessCheck, RetryPolicy} import org.nlogo.extensions.llm.models.{ChatMessage, ChatResponse, EnumFormat, JsonObjectFormat, ResponseFormat, Usage} import org.nlogo.extensions.llm.utils.JsonToNetLogo @@ -53,6 +53,13 @@ class LLMExtension extends DefaultClassManager { // Current provider instance private var currentProvider: Option[LLMProvider] = None + + // Named configurations loaded with llm:load-profile, each with its own + // provider, and which agent is bound to which. An agent with no binding + // uses the global configStore/currentProvider pair above, unchanged. + private val profiles: ProfileStore = new ProfileStore(cs => LLMExtension.createProvider(cs)) + private val agentProfile: WeakHashMap[Agent, String] = WeakHashMap() + private val profileLock = new Object // Per-agent conversation history private val messageHistory: WeakHashMap[Agent, ArrayBuffer[ChatMessage]] = WeakHashMap() @@ -140,6 +147,12 @@ class LLMExtension extends DefaultClassManager { // Token accounting primitives manager.addPrimitive("usage", UsageReporter) manager.addPrimitive("usage-total", UsageTotalReporter) + + // Per-agent profile primitives + manager.addPrimitive("load-profile", LoadProfileCommand) + manager.addPrimitive("use-profile", UseProfileCommand) + manager.addPrimitive("profile", ProfileReporter) + manager.addPrimitive("profiles", ProfilesReporter) } /** @@ -151,6 +164,8 @@ class LLMExtension extends DefaultClassManager { agentUsage.clear() runUsage = UsageTotals.empty } + // Bindings die with the agents; loaded profiles persist like the global config does. + profileLock.synchronized { agentProfile.clear() } } /** @@ -170,6 +185,34 @@ class LLMExtension extends DefaultClassManager { } } + /** The profile name an agent is bound to, or the reserved default name. */ + private def profileNameFor(agent: Agent): String = + profileLock.synchronized { agentProfile.getOrElse(agent, ProfileStore.DefaultName) } + + /** + * The provider a call from this agent goes through: the bound profile's + * provider, or the global one. The instance is resolved once per call and + * captured by the caller, so an async request keeps its provider even if + * the agent is rebound before the reply arrives. + */ + private def ensureProvider(agent: Agent): LLMProvider = + profileNameFor(agent) match { + case ProfileStore.DefaultName => ensureProvider() + case name => + profiles.provider(name) match { + case Success(provider) => provider + case Failure(e) => + throw new ExtensionException(s"Failed to initialize LLM provider for profile '$name': ${e.getMessage}") + } + } + + /** The configuration an agent's calls are shaped by. */ + private def effectiveConfig(agent: Agent): ConfigStore = + profileNameFor(agent) match { + case ProfileStore.DefaultName => configStore + case name => profiles.get(name).map(_.config).getOrElse(configStore) + } + /** * Get or create conversation history for an agent. * Callers must hold historyLock — the buffer must not escape a locked section. @@ -270,22 +313,49 @@ class LLMExtension extends DefaultClassManager { /** * Check if a provider has an API key configured */ - private def hasApiKey(providerName: String): Boolean = { + private def hasApiKey(providerName: String): Boolean = hasApiKey(providerName, configStore) + + private def hasApiKey(providerName: String, config: ConfigStore): Boolean = { val providerKeyName = ConfigStore.getProviderApiKeyName(providerName) - configStore.get(providerKeyName).orElse(configStore.get(ConfigStore.API_KEY)) match { + config.get(providerKeyName).orElse(config.get(ConfigStore.API_KEY)) match { case Some(key) => key.trim.nonEmpty case None => false } } + + /** + * The readiness check llm:load-config and llm:load-profile share: a cloud + * provider needs a key in this config, a local one needs a reachable server. + */ + private def checkReadiness(desc: ProviderDescriptor, providerName: String, config: ConfigStore): Unit = + desc.readinessCheck match { + case ReadinessCheck.ServerReachable => + if (!isOllamaReachable(config)) { + val baseUrl = config.get(desc.baseUrlConfigKey) + .orElse(config.get(ConfigStore.BASE_URL)) + .getOrElse(desc.defaultBaseUrl) + throw new ExtensionException( + s"Config loaded but ${desc.displayName} not reachable at $baseUrl. Please start the server or change ${desc.baseUrlConfigKey} in config. For help: print llm:provider-help \"${desc.name}\"" + ) + } + case ReadinessCheck.ApiKey => + if (!hasApiKey(providerName, config)) { + throw new ExtensionException( + s"Config loaded but ${desc.displayName} provider requires an API key. Set '${desc.apiKeyConfigKey}' in config. For help: print llm:provider-help \"${desc.name}\"" + ) + } + } /** * Check if Ollama is reachable (synchronous with short timeout) */ - private def isOllamaReachable: Boolean = { + private def isOllamaReachable: Boolean = isOllamaReachable(configStore) + + private def isOllamaReachable(config: ConfigStore): Boolean = { try { val provider = new OllamaProvider() - val baseUrl = configStore.get(ConfigStore.OLLAMA_BASE_URL) - .orElse(configStore.get(ConfigStore.BASE_URL)) + val baseUrl = config.get(ConfigStore.OLLAMA_BASE_URL) + .orElse(config.get(ConfigStore.BASE_URL)) .getOrElse(ConfigStore.DEFAULT_OLLAMA_BASE_URL) provider.setConfig(ConfigStore.BASE_URL, baseUrl) @@ -496,23 +566,7 @@ class LLMExtension extends DefaultClassManager { // Readiness check uses the now-loaded config (needs apiKeyConfigKey lookup). // Roll back on failure so llm:active and friends keep the prior state. try { - desc.readinessCheck match { - case ReadinessCheck.ServerReachable => - if (!isOllamaReachable) { - val baseUrl = configStore.get(desc.baseUrlConfigKey) - .orElse(configStore.get(ConfigStore.BASE_URL)) - .getOrElse(desc.defaultBaseUrl) - throw new ExtensionException( - s"Config loaded but ${desc.displayName} not reachable at $baseUrl. Please start the server or change ${desc.baseUrlConfigKey} in config. For help: print llm:provider-help \"${desc.name}\"" - ) - } - case ReadinessCheck.ApiKey => - if (!hasApiKey(providerName)) { - throw new ExtensionException( - s"Config loaded but ${desc.displayName} provider requires an API key. Set '${desc.apiKeyConfigKey}' in config. For help: print llm:provider-help \"${desc.name}\"" - ) - } - } + checkReadiness(desc, providerName, configStore) } catch { case e: ExtensionException => configStore.loadFromMap(previousConfig) @@ -539,7 +593,7 @@ class LLMExtension extends DefaultClassManager { val agent = context.getAgent try { - val provider = ensureProvider() + val provider = ensureProvider(agent) val userMessage = ChatMessage.user(inputText) @@ -572,7 +626,7 @@ class LLMExtension extends DefaultClassManager { val agent = context.getAgent try { - val provider = ensureProvider() + val provider = ensureProvider(agent) val userMessage = ChatMessage.user(inputText) @@ -607,7 +661,7 @@ class LLMExtension extends DefaultClassManager { val agent = context.getAgent try { - val provider = ensureProvider() + val provider = ensureProvider(agent) // Extract model directory from workspace val modelDir = Option(context.workspace.getModelPath).flatMap { path => @@ -670,7 +724,7 @@ class LLMExtension extends DefaultClassManager { val agent = context.getAgent try { - val provider = ensureProvider() + val provider = ensureProvider(agent) val choices = choicesList.map(_.toString).toList @@ -795,7 +849,7 @@ class LLMExtension extends DefaultClassManager { } try { - val provider = ensureProvider() + val provider = ensureProvider(agent) val userMessage = ChatMessage.user(inputText) @@ -846,7 +900,7 @@ class LLMExtension extends DefaultClassManager { val agent = context.getAgent try { - val provider = ensureProvider() + val provider = ensureProvider(agent) // Anthropic has no schemaless JSON mode and Gemini's mime type alone is // only a hint, so the instruction is also stated in the prompt. Providers @@ -938,7 +992,7 @@ class LLMExtension extends DefaultClassManager { val agent = context.getAgent try { - val provider = ensureProvider() + val provider = ensureProvider(agent) val userMessage = ChatMessage.user(inputText) @@ -1251,6 +1305,83 @@ class LLMExtension extends DefaultClassManager { } } + /** + * llm:load-profile name file — load a config file under a name so agents + * can be bound to it with llm:use-profile. Validated exactly like + * llm:load-config; a rejected load leaves any existing profile of that + * name untouched, because the store is only updated after every check. + */ + object LoadProfileCommand extends Command { + override def getSyntax: Syntax = Syntax.commandSyntax(right = List(Syntax.StringType, Syntax.StringType)) + + override def perform(args: Array[Argument], context: Context): Unit = { + val name = args(0).getString + val filename = args(1).getString + + val modelDir = Option(context.workspace.getModelPath).flatMap { path => + Option(new java.io.File(path).getParent) + } + + val config = ConfigLoader.loadFromFile(filename, modelDir) match { + case Success(c) => c + case Failure(e) => + throw new ExtensionException(s"llm:load-profile: Failed to load configuration from '$filename': ${e.getMessage}") + } + + val providerName = config.getOrElse(ConfigStore.PROVIDER, ConfigStore.DEFAULT_PROVIDER) + val desc = ProviderRegistry.get(providerName.toLowerCase.trim).getOrElse { + throw new ExtensionException( + s"llm:load-profile: Unknown provider '$providerName' in config. Supported: ${ProviderRegistry.allNames.toList.sorted.mkString(", ")}" + ) + } + + val candidate = ConfigStore.withDefaults() + candidate.updateFromMap(config) + checkReadiness(desc, providerName, candidate) + + try profiles.load(name, candidate.toMap) + catch { + case e: IllegalArgumentException => + throw new ExtensionException(s"llm:load-profile: ${e.getMessage}") + } + } + } + + /** + * llm:use-profile name — route the calling agent's calls through a loaded + * profile. The reserved default name unbinds it. + */ + object UseProfileCommand extends Command { + override def getSyntax: Syntax = Syntax.commandSyntax(right = List(Syntax.StringType)) + + override def perform(args: Array[Argument], context: Context): Unit = { + val name = args(0).getString.trim.toLowerCase + val agent = context.getAgent + if (name == ProfileStore.DefaultName) { + profileLock.synchronized { agentProfile.remove(agent) } + } else if (profiles.contains(name)) { + profileLock.synchronized { agentProfile.update(agent, name) } + } else { + throw new ExtensionException( + s"llm:use-profile: no profile named '${args(0).getString.trim}'. Loaded profiles: ${profiles.describeLoaded}" + ) + } + } + } + + /** llm:profile — the calling agent's profile name, or the default name. */ + object ProfileReporter extends Reporter { + override def getSyntax: Syntax = Syntax.reporterSyntax(ret = Syntax.StringType) + override def report(args: Array[Argument], context: Context): AnyRef = profileNameFor(context.getAgent) + } + + /** llm:profiles — the loaded profile names, sorted. */ + object ProfilesReporter extends Reporter { + override def getSyntax: Syntax = Syntax.reporterSyntax(ret = Syntax.ListType) + override def report(args: Array[Argument], context: Context): AnyRef = + LogoList.fromIterator(profiles.names.iterator.map(n => n: AnyRef)) + } + /** * Token accounting as a `[[key value] ...]` list, the same shape structured * output uses so `llm:get` reads it. Counters are numbers; `cost` is a @@ -1294,8 +1425,9 @@ class LLMExtension extends DefaultClassManager { override def report(args: Array[Argument], context: Context): AnyRef = { try { - val provider = configStore.getOrElse(ConfigStore.PROVIDER, ConfigStore.DEFAULT_PROVIDER) - val model = configStore.getOrElse(ConfigStore.MODEL, ModelRegistry.defaultModel(provider)) + val config = effectiveConfig(context.getAgent) + val provider = config.getOrElse(ConfigStore.PROVIDER, ConfigStore.DEFAULT_PROVIDER) + val model = config.getOrElse(ConfigStore.MODEL, ModelRegistry.defaultModel(provider)) LogoList(provider, model) } catch { case e: Exception => @@ -1309,7 +1441,7 @@ class LLMExtension extends DefaultClassManager { override def report(args: Array[Argument], context: Context): AnyRef = { try { - configStore.summary + effectiveConfig(context.getAgent).summary } catch { case e: Exception => throw new ExtensionException(s"Failed to get config summary: ${e.getMessage}") diff --git a/src/main/config/ProfileStore.scala b/src/main/config/ProfileStore.scala new file mode 100644 index 0000000..7c729dc --- /dev/null +++ b/src/main/config/ProfileStore.scala @@ -0,0 +1,83 @@ +// ABOUTME: Registry of named configurations, each with its own ConfigStore and one cached provider +// ABOUTME: Backs per-agent model selection: an agent bound to a profile calls through that profile's provider +package org.nlogo.extensions.llm.config + +import org.nlogo.extensions.llm.providers.LLMProvider +import scala.util.{Failure, Success, Try} + +/** + * One named configuration and the provider built from it. + * + * The provider is created on first use and then reused for the profile's + * lifetime, so an async call that captured it keeps talking to the same + * endpoint no matter what the modeler reassigns afterwards. A failed + * creation is not cached: the next call tries again, so a transient error + * (a server that was not up yet) does not poison the profile. + */ +final class Profile(val name: String, val config: ConfigStore, create: ConfigStore => Try[LLMProvider]) { + private var cached: Option[LLMProvider] = None + + def provider: Try[LLMProvider] = synchronized { + cached match { + case Some(p) => Success(p) + case None => + create(config).map { p => + cached = Some(p) + p + } + } + } +} + +/** + * Named configurations the modeler has loaded with llm:load-profile. + * + * Names are trimmed and lower-cased so "Fast" and "fast" are one profile. + * The name reserved for the global configuration is rejected here so no + * profile can shadow it. Loading a name that already exists replaces the + * whole entry, cached provider included. + */ +final class ProfileStore(create: ConfigStore => Try[LLMProvider]) { + private var profiles: Map[String, Profile] = Map.empty + + private def normalise(name: String): String = name.trim.toLowerCase + + def load(name: String, config: Map[String, String]): Unit = { + val key = normalise(name) + if (key.isEmpty) + throw new IllegalArgumentException("Profile name cannot be blank") + if (key == ProfileStore.DefaultName) + throw new IllegalArgumentException( + s"'${ProfileStore.DefaultName}' is reserved for the global configuration. Choose another name." + ) + val store = new ConfigStore() + store.loadFromMap(config) + val profile = new Profile(key, store, create) + synchronized { profiles = profiles.updated(key, profile) } + } + + def get(name: String): Option[Profile] = synchronized { profiles.get(normalise(name)) } + + def contains(name: String): Boolean = get(name).isDefined + + def names: Seq[String] = synchronized { profiles.keys.toSeq.sorted } + + def provider(name: String): Try[LLMProvider] = + get(name) match { + case Some(profile) => profile.provider + case None => + Failure(new IllegalArgumentException( + s"no profile named '${name.trim}'. Loaded profiles: ${describeLoaded}" + )) + } + + def describeLoaded: String = { + val loaded = names + if (loaded.isEmpty) "none" else loaded.mkString(", ") + } +} + +object ProfileStore { + /** The name that means "the global configuration", never a loaded profile. */ + val DefaultName: String = "default" +} diff --git a/src/test/DeterministicTestProvider.scala b/src/test/DeterministicTestProvider.scala index cc70b32..15358ea 100644 --- a/src/test/DeterministicTestProvider.scala +++ b/src/test/DeterministicTestProvider.scala @@ -195,6 +195,10 @@ class DeterministicTestProvider(implicit ec: ExecutionContext) extends LLMProvid } else if (lastUserMessage.contains("__TEST_EMPTY_CONTENT")) { // Return empty content (simulates thinking model with no content) "" + } else if (lastUserMessage.contains("__TEST_ECHO_MODEL")) { + // Reveal which configuration served this call, so profile routing can + // be asserted: the model is the one field every profile fixture sets. + configStore.get(ConfigStore.MODEL).getOrElse("__NO_MODEL_CONFIGURED__") } else if (lastUserMessage.contains("Options:\n") && lastUserMessage.contains("Your choice as {")) { // Choose prompt — extract and return the first option from the Options block val optionsIdx = lastUserMessage.indexOf("Options:\n") diff --git a/src/test/ProfileStoreSpec.scala b/src/test/ProfileStoreSpec.scala new file mode 100644 index 0000000..19c8a4d --- /dev/null +++ b/src/test/ProfileStoreSpec.scala @@ -0,0 +1,122 @@ +// ABOUTME: Deterministic tests for ProfileStore, the named-configuration registry behind per-agent profiles +// ABOUTME: Asserts load/replace semantics, reserved names, and that each profile caches exactly one provider +package org.nlogo.extensions.llm.config + +import org.nlogo.extensions.llm.providers.{DeterministicTestProvider, LLMProvider} +import org.scalatest.funsuite.AnyFunSuite +import java.util.concurrent.atomic.AtomicInteger +import scala.concurrent.ExecutionContext.Implicits.global +import scala.util.{Failure, Success, Try} + +class ProfileStoreSpec extends AnyFunSuite { + + /** A store whose provider factory counts calls and records the model it was built with. */ + private def storeWithCounter(): (ProfileStore, AtomicInteger) = { + val created = new AtomicInteger(0) + val store = new ProfileStore({ config => + created.incrementAndGet() + val provider = new DeterministicTestProvider() + config.toMap.foreach { case (k, v) => provider.setConfig(k, v) } + Success(provider) + }) + (store, created) + } + + private val configA = Map("provider" -> "openai", "openai_api_key" -> "k", "model" -> "model-a") + private val configB = Map("provider" -> "openai", "openai_api_key" -> "k", "model" -> "model-b") + + test("a freshly created store has no profiles") { + val (store, _) = storeWithCounter() + assert(store.names.isEmpty) + assert(store.get("a").isEmpty) + } + + test("load registers a profile under its name with its own config") { + val (store, _) = storeWithCounter() + store.load("a", configA) + assert(store.names == Seq("a")) + assert(store.get("a").map(_.config.get("model")).contains(Some("model-a"))) + } + + test("names are reported sorted") { + val (store, _) = storeWithCounter() + store.load("zeta", configA) + store.load("alpha", configB) + assert(store.names == Seq("alpha", "zeta")) + } + + test("the reserved default name cannot be loaded") { + val (store, _) = storeWithCounter() + val err = intercept[IllegalArgumentException](store.load(ProfileStore.DefaultName, configA)) + assert(err.getMessage.contains("reserved")) + assert(store.names.isEmpty) + } + + test("a blank name cannot be loaded") { + val (store, _) = storeWithCounter() + intercept[IllegalArgumentException](store.load(" ", configA)) + assert(store.names.isEmpty) + } + + test("names are matched case-insensitively and trimmed") { + val (store, _) = storeWithCounter() + store.load("Fast", configA) + assert(store.get(" fast ").isDefined) + assert(store.names == Seq("fast")) + } + + test("provider is created once per profile and then reused") { + val (store, created) = storeWithCounter() + store.load("a", configA) + val first = store.provider("a") + val second = store.provider("a") + assert(first.isSuccess) + assert(first.get eq second.get) + assert(created.get() == 1) + } + + test("reloading a name replaces its config and drops the cached provider") { + val (store, created) = storeWithCounter() + store.load("a", configA) + val before = store.provider("a").get + store.load("a", configB) + val after = store.provider("a").get + assert(!(before eq after)) + assert(created.get() == 2) + assert(after.getConfig("model").contains("model-b")) + assert(store.names == Seq("a")) + } + + test("provider for an unknown name is a failure that lists what is loaded") { + val (store, _) = storeWithCounter() + store.load("a", configA) + store.provider("nope") match { + case Failure(e) => + assert(e.getMessage.contains("nope")) + assert(e.getMessage.contains("a")) + case Success(_) => fail("expected a failure for an unknown profile") + } + } + + test("a provider factory failure is surfaced and not cached") { + var attempts = 0 + val store = new ProfileStore({ _ => + attempts += 1 + if (attempts == 1) Failure(new RuntimeException("boom")) else Success(new DeterministicTestProvider()) + }) + store.load("a", configA) + assert(store.provider("a").isFailure) + assert(store.provider("a").isSuccess, "a later attempt should retry rather than cache the failure") + } + + test("concurrent first use of a profile still creates exactly one provider") { + val (store, created) = storeWithCounter() + store.load("a", configA) + val threads = (1 to 8).map { _ => + new Thread(() => { store.provider("a"); () }) + } + threads.foreach(_.start()) + threads.foreach(_.join()) + assert(created.get() == 1) + } +} diff --git a/tests.txt b/tests.txt index 4faf5de..0130a0b 100644 --- a/tests.txt +++ b/tests.txt @@ -836,3 +836,172 @@ LLMUsageUnchangedWhenCallFails llm:get llm:usage "calls" => 0 llm:get llm:usage "input-tokens" => 0 O> clear-all + +LLMProfileDefaultsToGlobal + extensions [llm] + O> llm:set-api-key "test-key" + O> llm:set-provider "openai" + O> clear-all + llm:profile => "default" + llm:profiles => [] + +LLMLoadProfileRejectsBadProvider + extensions [llm] + globals [err] + O> carefully [llm:load-profile "bad" "demos/test-config-bad-provider"] [set err error-message] + member? "Unknown provider" err => true + llm:profiles => [] + +LLMLoadProfileRejectsMissingFile + extensions [llm] + globals [err] + O> carefully [llm:load-profile "nope" "demos/no-such-file"] [set err error-message] + member? "Failed to load" err => true + llm:profiles => [] + +LLMLoadProfileRejectsReservedName + extensions [llm] + O> llm:load-profile "default" "demos/test-profile-a" => ERROR Extension exception: llm:load-profile: 'default' is reserved for the global configuration. Choose another name. + +LLMUseProfileUnknownThrows + extensions [llm] + O> llm:load-profile "a" "demos/test-profile-a" + O> llm:use-profile "zzz" => ERROR Extension exception: llm:use-profile: no profile named 'zzz'. Loaded profiles: a + +LLMUseProfileRoutesToItsModel + extensions [llm] + O> llm:set-api-key "test-key" + O> llm:set-provider "openai" + O> llm:set-model "model-default" + O> clear-all + O> llm:load-profile "a" "demos/test-profile-a" + O> llm:load-profile "b" "demos/test-profile-b" + O> crt 2 + O> ask turtle 0 [ llm:use-profile "a" ] + O> ask turtle 1 [ llm:use-profile "b" ] + [llm:chat "__TEST_ECHO_MODEL"] of turtle 0 => "model-a" + [llm:chat "__TEST_ECHO_MODEL"] of turtle 1 => "model-b" + llm:chat "__TEST_ECHO_MODEL" => "model-default" + [llm:profile] of turtle 0 => "a" + [llm:profile] of turtle 1 => "b" + llm:profile => "default" + llm:profiles => ["a" "b"] + O> clear-all + +LLMActiveAndConfigReportCallingAgentProfile + extensions [llm] + O> llm:set-api-key "test-key" + O> llm:set-provider "openai" + O> llm:set-model "model-default" + O> clear-all + O> llm:load-profile "a" "demos/test-profile-a" + O> crt 1 + O> ask turtle 0 [ llm:use-profile "a" ] + [llm:active] of turtle 0 => ["openai" "model-a"] + llm:active => ["openai" "model-default"] + member? "model-a" [llm:config] of turtle 0 => true + member? "model-a" llm:config => false + O> clear-all + +LLMUseProfileDefaultReturnsToGlobal + extensions [llm] + O> llm:set-api-key "test-key" + O> llm:set-provider "openai" + O> llm:set-model "model-default" + O> clear-all + O> llm:load-profile "a" "demos/test-profile-a" + O> crt 1 + O> ask turtle 0 [ llm:use-profile "a" ] + [llm:chat "__TEST_ECHO_MODEL"] of turtle 0 => "model-a" + O> ask turtle 0 [ llm:use-profile "default" ] + [llm:chat "__TEST_ECHO_MODEL"] of turtle 0 => "model-default" + [llm:profile] of turtle 0 => "default" + O> clear-all + +LLMProfileAsyncKeepsProviderAfterReassignment + extensions [llm] + globals [p r] + O> llm:set-api-key "test-key" + O> llm:set-provider "openai" + O> llm:set-model "model-default" + O> clear-all + O> llm:load-profile "a" "demos/test-profile-a" + O> llm:load-profile "b" "demos/test-profile-b" + O> crt 1 + O> ask turtle 0 [ llm:use-profile "a" ] + O> ask turtle 0 [ set p llm:chat-async "__TEST_DELAY:300:__TEST_ECHO_MODEL" ] + O> ask turtle 0 [ llm:use-profile "b" ] + O> set r (runresult p) + r => "model-a" + [llm:chat "__TEST_ECHO_MODEL"] of turtle 0 => "model-b" + O> clear-all + +LLMSetModelLeavesProfilesAlone + extensions [llm] + O> llm:set-api-key "test-key" + O> llm:set-provider "openai" + O> llm:set-model "model-default" + O> clear-all + O> llm:load-profile "a" "demos/test-profile-a" + O> crt 1 + O> ask turtle 0 [ llm:use-profile "a" ] + O> llm:set-model "model-changed" + [llm:chat "__TEST_ECHO_MODEL"] of turtle 0 => "model-a" + llm:chat "__TEST_ECHO_MODEL" => "model-changed" + O> clear-all + +LLMReloadProfileReplacesIt + extensions [llm] + O> llm:set-api-key "test-key" + O> llm:set-provider "openai" + O> clear-all + O> llm:load-profile "a" "demos/test-profile-a" + O> crt 1 + O> ask turtle 0 [ llm:use-profile "a" ] + [llm:chat "__TEST_ECHO_MODEL"] of turtle 0 => "model-a" + O> llm:load-profile "a" "demos/test-profile-b" + [llm:chat "__TEST_ECHO_MODEL"] of turtle 0 => "model-b" + llm:profiles => ["a"] + O> clear-all + +LLMFailedReloadKeepsOldProfile + extensions [llm] + O> llm:set-api-key "test-key" + O> llm:set-provider "openai" + O> clear-all + O> llm:load-profile "a" "demos/test-profile-a" + O> crt 1 + O> ask turtle 0 [ llm:use-profile "a" ] + O> carefully [llm:load-profile "a" "demos/test-config-bad-provider"] [ ] + [llm:chat "__TEST_ECHO_MODEL"] of turtle 0 => "model-a" + O> clear-all + +LLMClearAllResetsAssignmentsKeepsProfiles + extensions [llm] + O> llm:set-api-key "test-key" + O> llm:set-provider "openai" + O> clear-all + O> llm:load-profile "a" "demos/test-profile-a" + O> llm:use-profile "a" + llm:profile => "a" + O> clear-all + llm:profile => "default" + llm:profiles => ["a"] + O> clear-all + +LLMProfileHistoryAndUsageStayPerAgent + extensions [llm] + globals [r] + O> llm:set-api-key "test-key" + O> llm:set-provider "openai" + O> clear-all + O> llm:load-profile "a" "demos/test-profile-a" + O> crt 2 + O> ask turtle 0 [ llm:use-profile "a" ] + O> ask turtles [ set r llm:chat "__TEST_USAGE:5,1 hi" ] + [length llm:history] of turtle 0 => 2 + [length llm:history] of turtle 1 => 2 + [llm:get llm:usage "input-tokens"] of turtle 0 => 5 + [llm:get llm:usage "input-tokens"] of turtle 1 => 5 + llm:get llm:usage-total "calls" => 2 + O> clear-all From 9dd9d26dab444ebbed65d9facccd25aef27fcbb2 Mon Sep 17 00:00:00 2001 From: JNK234 Date: Tue, 8 Sep 2026 22:48:06 -0400 Subject: [PATCH 2/2] fix: address review findings on profiles: shared throttle, defaults, wait budget Three problems a review found in how profiles interact with the rest of the extension. Throttle gates were replaced whenever the configured cap or interval changed. Two profiles on one endpoint with different settings alternate every tick, so each call got a fresh gate with a full set of permits and the cap meant nothing. The gate is now reconfigured in place: a smaller cap admits nothing until in-flight requests drain below it, a larger one hands the new permits to the oldest waiters at once. release() repays a shrunk-cap deficit before admitting anyone. A profile config was built on ConfigStore.withDefaults(), which carries model=gpt-4o-mini, so an Anthropic profile that named no model asked for an OpenAI model. Profiles now load the file as the whole configuration, the way llm:load-config does, and the provider's own default applies. Every Await read the global timeout and retry budget. Calls now use the calling agent's effective config, and an async call captures the budget at launch so a later config change or rebinding cannot alter it. Tests: throttle spec covers in-place reconfigure, shrink under load, and grow with waiters; tests.txt covers a model-less Anthropic profile, a profile's own one-second budget timing out while the observer succeeds, and an async call keeping its budget after the global config changes. --- demos/test-config-fast-timeout | 6 ++ demos/test-profile-anthropic | 3 + demos/test-profile-slow | 6 ++ docs/API-REFERENCE.md | 6 ++ src/main/LLMExtension.scala | 40 +++++++----- src/main/providers/RequestThrottle.scala | 79 ++++++++++++++++++----- src/test/RequestThrottleSpec.scala | 82 ++++++++++++++++++++---- tests.txt | 41 ++++++++++++ 8 files changed, 217 insertions(+), 46 deletions(-) create mode 100644 demos/test-config-fast-timeout create mode 100644 demos/test-profile-anthropic create mode 100644 demos/test-profile-slow diff --git a/demos/test-config-fast-timeout b/demos/test-config-fast-timeout new file mode 100644 index 0000000..20efb8d --- /dev/null +++ b/demos/test-config-fast-timeout @@ -0,0 +1,6 @@ +# Test fixture: global config with a one-second wait budget and no retry allowance +provider=openai +openai_api_key=test-key +model=model-fast-timeout +timeout_seconds=1 +retry_max_elapsed_seconds=0 diff --git a/demos/test-profile-anthropic b/demos/test-profile-anthropic new file mode 100644 index 0000000..e6ee957 --- /dev/null +++ b/demos/test-profile-anthropic @@ -0,0 +1,3 @@ +# Test fixture: Anthropic profile with no model, so the provider default must apply +provider=anthropic +anthropic_api_key=test-key diff --git a/demos/test-profile-slow b/demos/test-profile-slow new file mode 100644 index 0000000..68fa661 --- /dev/null +++ b/demos/test-profile-slow @@ -0,0 +1,6 @@ +# Test fixture: profile with a one-second wait budget and no retry allowance +provider=openai +openai_api_key=test-key +model=model-slow +timeout_seconds=1 +retry_max_elapsed_seconds=0 diff --git a/docs/API-REFERENCE.md b/docs/API-REFERENCE.md index 8fd3ba5..cbc7613 100644 --- a/docs/API-REFERENCE.md +++ b/docs/API-REFERENCE.md @@ -920,6 +920,12 @@ exists replaces it. A rejected reload leaves the existing profile untouched. setters act on the default configuration only. A profile is what its file says. To change a profile, edit the file and load it again. - Two profiles on the same provider and endpoint share one request throttle. + If their `max_concurrent_requests` or `min_request_interval_ms` differ, the + shared throttle takes whichever setting the most recent call carried, without + losing count of requests already in flight. Give profiles on one endpoint the + same throttling settings. +- A profile's `timeout_seconds` and retry settings govern calls made through + it. An async call keeps the budget in force when it was launched. ### llm:use-profile diff --git a/src/main/LLMExtension.scala b/src/main/LLMExtension.scala index c8237bf..2147ca9 100644 --- a/src/main/LLMExtension.scala +++ b/src/main/LLMExtension.scala @@ -82,13 +82,16 @@ class LLMExtension extends DefaultClassManager { * Create an AwaitableReporter that wraps a Future to provide truly async behavior * The Future starts immediately but execution defers until runresult is called */ - private def createAwaitableReporter(future: Future[String]): AnonymousReporter = { + private def createAwaitableReporter(future: Future[String], timeout: FiniteDuration): AnonymousReporter = { new AnonymousReporter { override def syntax: Syntax = Syntax.reporterSyntax(right = List(), ret = Syntax.StringType) override def report(context: Context, args: Array[AnyRef]): AnyRef = { try { - Await.result(future, getAwaitTimeout) + // The budget in force for the calling agent when the request was + // launched: a later config change or profile rebinding cannot alter + // how long an already-pending call may take. + Await.result(future, timeout) } catch { case e: Exception => throw new ExtensionException(s"Async LLM operation failed: ${e.getMessage}") @@ -271,8 +274,8 @@ class LLMExtension extends DefaultClassManager { /** * Get timeout from config, falling back to 30 seconds */ - private def getTimeoutSeconds: Int = - configStore.get(ConfigStore.TIMEOUT_SECONDS).map { s => + private def getTimeoutSeconds(config: ConfigStore): Int = + config.get(ConfigStore.TIMEOUT_SECONDS).map { s => scala.util.Try(s.toInt).getOrElse { System.err.println(s"WARNING: Invalid timeout_seconds value '$s' (not a valid integer), using default 30") 30 @@ -301,13 +304,13 @@ class LLMExtension extends DefaultClassManager { * bound is unchanged from before throttling existed, so an unthrottled model — the * default — behaves exactly as it did. */ - private def getAwaitTimeout: FiniteDuration = { - val retryBudget = configStore.get(RetryPolicy.MAX_ELAPSED_SECONDS) + private def awaitTimeoutFor(config: ConfigStore): FiniteDuration = { + val retryBudget = config.get(RetryPolicy.MAX_ELAPSED_SECONDS) .flatMap(s => scala.util.Try(s.trim.toDouble).toOption) .filter(d => d >= 0.0 && d.isFinite) .map(d => (d * 1000.0).toLong.millis) .getOrElse(RetryPolicy.DefaultMaxElapsed) - getTimeoutSeconds.seconds + retryBudget + getTimeoutSeconds(config).seconds + retryBudget } /** @@ -599,7 +602,7 @@ class LLMExtension extends DefaultClassManager { // Send chat request with user message included, but don't mutate history yet val responseFuture = provider.chatWithFullResponse(snapshotHistory(agent) :+ userMessage) - val response = Await.result(responseFuture, getAwaitTimeout) + val response = Await.result(responseFuture, awaitTimeoutFor(effectiveConfig(agent))) recordUsage(agent, response) val responseMessage = replyMessage(response) @@ -640,7 +643,7 @@ class LLMExtension extends DefaultClassManager { } // Return AnonymousReporter that wraps the Future - createAwaitableReporter(responseFuture) + createAwaitableReporter(responseFuture, awaitTimeoutFor(effectiveConfig(agent))) } catch { case e: Exception => @@ -692,7 +695,7 @@ class LLMExtension extends DefaultClassManager { // Send chat request val responseFuture = provider.chatWithFullResponse(tempHistory.toSeq) - val response = Await.result(responseFuture, getAwaitTimeout) + val response = Await.result(responseFuture, awaitTimeoutFor(effectiveConfig(agent))) recordUsage(agent, response) val responseMessage = replyMessage(response) @@ -754,7 +757,7 @@ class LLMExtension extends DefaultClassManager { // The prompt still spells out the options, so a provider that ignores // the constraint behaves exactly as it did before. val responseFuture = provider.chatWithFormat(tempHistory.toSeq, EnumFormat(choices)) - val response = Await.result(responseFuture, getAwaitTimeout) + val response = Await.result(responseFuture, awaitTimeoutFor(effectiveConfig(agent))) recordUsage(agent, response) // Extract text: prefer content, fall back to thinking field @@ -854,7 +857,7 @@ class LLMExtension extends DefaultClassManager { val userMessage = ChatMessage.user(inputText) val responseFuture = provider.chatWithFormat(snapshotHistory(agent) :+ userMessage, format) - val response = Await.result(responseFuture, getAwaitTimeout) + val response = Await.result(responseFuture, awaitTimeoutFor(effectiveConfig(agent))) recordUsage(agent, response) val content = response.firstContent.getOrElse("") @@ -915,7 +918,7 @@ class LLMExtension extends DefaultClassManager { tempHistory += userMessage val responseFuture = provider.chatWithFormat(tempHistory.toSeq, JsonObjectFormat) - val response = Await.result(responseFuture, getAwaitTimeout) + val response = Await.result(responseFuture, awaitTimeoutFor(effectiveConfig(agent))) recordUsage(agent, response) val content = response.firstContent.getOrElse("") @@ -998,7 +1001,7 @@ class LLMExtension extends DefaultClassManager { // Send with user message included, but don't mutate history yet val responseFuture = provider.chatWithFullResponse(snapshotHistory(agent) :+ userMessage) - val response = Await.result(responseFuture, getAwaitTimeout) + val response = Await.result(responseFuture, awaitTimeoutFor(effectiveConfig(agent))) recordUsage(agent, response) val answerText = response.firstContent.getOrElse("") @@ -1335,11 +1338,14 @@ class LLMExtension extends DefaultClassManager { ) } - val candidate = ConfigStore.withDefaults() - candidate.updateFromMap(config) + // Mirror llm:load-config: the file is the whole configuration, with no + // OpenAI-flavoured defaults layered underneath. A profile that names no + // model gets its own provider's default, not gpt-4o-mini. + val candidate = new ConfigStore() + candidate.loadFromMap(config) checkReadiness(desc, providerName, candidate) - try profiles.load(name, candidate.toMap) + try profiles.load(name, config) catch { case e: IllegalArgumentException => throw new ExtensionException(s"llm:load-profile: ${e.getMessage}") diff --git a/src/main/providers/RequestThrottle.scala b/src/main/providers/RequestThrottle.scala index 6de0e0b..6e31d78 100644 --- a/src/main/providers/RequestThrottle.scala +++ b/src/main/providers/RequestThrottle.scala @@ -67,16 +67,53 @@ object SystemThrottleClock extends ThrottleClock { * @param clock time source and delay scheduler */ class RequestThrottle( - val maxConcurrent: Int, - val minIntervalMs: Long, + initialMaxConcurrent: Int, + initialMinIntervalMs: Long, clock: ThrottleClock = SystemThrottleClock ) { - require(maxConcurrent > 0, "maxConcurrent must be positive") + require(initialMaxConcurrent > 0, "maxConcurrent must be positive") - // Guards `available`, `waiters` and `lastStartMs` together. Held only for - // queue arithmetic — never across a user callback or an HTTP send. + // Guards `available`, `waiters`, `lastStartMs` and the limits together. Held + // only for queue arithmetic — never across a user callback or an HTTP send. private val lock = new Object - private var available: Int = maxConcurrent + @volatile private var currentMax: Int = initialMaxConcurrent + @volatile private var currentInterval: Long = initialMinIntervalMs + private var available: Int = initialMaxConcurrent + + def maxConcurrent: Int = currentMax + def minIntervalMs: Long = currentInterval + + /** Requests holding a permit right now. Can exceed the cap briefly after it shrinks. */ + def inFlight: Int = lock.synchronized { currentMax - available } + + /** Requests queued for a permit. */ + def waiting: Int = lock.synchronized { waiters.size } + + /** + * Change the limits without replacing the gate. + * + * Two profiles on one endpoint can carry different settings, and calls + * alternate between them every tick. Swapping in a fresh gate per change + * would hand every call a full set of permits and the cap would mean + * nothing. Adjusting in place keeps the in-flight count and the queue: a + * smaller cap admits nothing until requests drain below it, a larger one + * hands the new permits to the oldest waiters at once. + */ + def reconfigure(newMaxConcurrent: Int, newMinIntervalMs: Long): Unit = { + require(newMaxConcurrent > 0, "maxConcurrent must be positive") + val admitted = lock.synchronized { + available += newMaxConcurrent - currentMax + currentMax = newMaxConcurrent + currentInterval = newMinIntervalMs + val handoffs = mutable.ListBuffer.empty[Promise[Unit]] + while (available > 0 && waiters.nonEmpty) { + available -= 1 + handoffs += waiters.dequeue() + } + handoffs.toList + } + admitted.foreach(_.success(())) + } // FIFO, so a waiter cannot be overtaken indefinitely — with agents calling // every tick, barging would pass one over for the whole run. Only ever @@ -155,7 +192,7 @@ class RequestThrottle( * exceed an RPM limit. */ private def paceThenProceed()(implicit ec: ExecutionContext): Future[Unit] = { - if (minIntervalMs <= 0L) return Future.unit + if (currentInterval <= 0L) return Future.unit val waitMs = lock.synchronized { val now = clock.nowMs @@ -164,7 +201,7 @@ class RequestThrottle( // timestamp and starting together. val earliest = if (lastStartMs == Long.MinValue) now - else math.max(now, lastStartMs + minIntervalMs) + else math.max(now, lastStartMs + currentInterval) lastStartMs = earliest earliest - now } @@ -193,7 +230,11 @@ class RequestThrottle( */ private def release(): Unit = { val handoff = lock.synchronized { - if (waiters.nonEmpty) Some(waiters.dequeue()) + // A negative `available` means the cap shrank while more than the new + // limit were in flight. Those releases repay the deficit; nobody is + // admitted until the count is back under the cap. + if (available < 0) { available += 1; None } + else if (waiters.nonEmpty) Some(waiters.dequeue()) else { available += 1; None } } handoff.foreach(_.success(())) @@ -304,23 +345,27 @@ object RequestThrottle { // or race two replacements past each other. The remapping function does // no I/O — it runs while the map holds a bin lock, so the notice below is // emitted afterwards rather than inside it. + // One gate per endpoint for the life of the process. A changed setting + // reconfigures it in place rather than replacing it, so requests already + // in flight stay counted — two profiles with different caps alternating + // on one endpoint would otherwise each get a fresh, empty gate. val throttle = throttles.compute(key, (_, existing) => { - if (existing != null && existing.maxConcurrent == limit && existing.minIntervalMs == interval) { - existing - } else { - if (existing != null) replaced.set(true) + if (existing == null) { new RequestThrottle(limit, interval) + } else { + if (existing.maxConcurrent != limit || existing.minIntervalMs != interval) { + existing.reconfigure(limit, interval) + replaced.set(true) + } + existing } }) if (replaced.get()) { - // Requests already running under the previous gate keep its permits, so - // in-flight work can briefly exceed the new cap. Reporting beats a stall - // or a silently changed limit. warnOnce( s"NOTE: request throttle for $providerName updated to " + s"$MAX_CONCURRENT_REQUESTS=$limit, $MIN_REQUEST_INTERVAL_MS=$interval. " + - "Requests already in flight finish under the previous limit." + "Requests already in flight count against the new limit." ) } throttle diff --git a/src/test/RequestThrottleSpec.scala b/src/test/RequestThrottleSpec.scala index ad081b9..cf1bbf5 100644 --- a/src/test/RequestThrottleSpec.scala +++ b/src/test/RequestThrottleSpec.scala @@ -78,7 +78,7 @@ class RequestThrottleSpec extends AnyFunSuite { test("permits beyond the cap wait, then every request completes") { // The core invariant: a burst wider than the cap is deferred, not dropped. - val throttle = new RequestThrottle(maxConcurrent = 2, minIntervalMs = 0) + val throttle = new RequestThrottle(2, 0) val gates = List.fill(5)(Promise[Unit]()) val running = new java.util.concurrent.atomic.AtomicInteger(0) val peak = new java.util.concurrent.atomic.AtomicInteger(0) @@ -104,7 +104,7 @@ class RequestThrottleSpec extends AnyFunSuite { // model stops issuing requests and looks like a hang. All three exit paths // must return the permit, so all three are exercised against a cap of 1 — // a leak in any of them blocks the requests that follow. - val throttle = new RequestThrottle(maxConcurrent = 1, minIntervalMs = 0) + val throttle = new RequestThrottle(1, 0) Await.result(throttle.withPermit(Future.successful("ok")), 5.seconds) @@ -126,7 +126,7 @@ class RequestThrottleSpec extends AnyFunSuite { test("queued work is admitted in arrival order") { // FIFO is what rules out starvation: with agents calling every tick, a // barging (non-fair) gate can pass over one waiter for an entire run. - val throttle = new RequestThrottle(maxConcurrent = 1, minIntervalMs = 0) + val throttle = new RequestThrottle(1, 0) val hold = Promise[Unit]() val admitted = new java.util.concurrent.ConcurrentLinkedQueue[Int]() @@ -147,7 +147,7 @@ class RequestThrottleSpec extends AnyFunSuite { // Blocking on a semaphore would park one thread per queued agent and can // deadlock the global pool. Queue far more work than the pool has threads, // then prove the pool still runs something else. - val throttle = new RequestThrottle(maxConcurrent = 1, minIntervalMs = 0) + val throttle = new RequestThrottle(1, 0) val hold = Promise[Unit]() val blocker = throttle.withPermit(hold.future) val queuedCalls = (1 to 64).map(_ => throttle.withPermit(Future.successful(()))) @@ -163,7 +163,7 @@ class RequestThrottleSpec extends AnyFunSuite { // Deterministic: the clock only moves when this test moves it, so the // assertion is about the delays the throttle asks for, not about timing. val clock = new ManualClock() - val throttle = new RequestThrottle(maxConcurrent = 4, minIntervalMs = 100, clock) + val throttle = new RequestThrottle(4, 100, clock) val calls = (1 to 3).map(_ => throttle.withPermit(Future.successful(()))) eventually("all three to reserve a start slot")(clock.scheduledAt.size == 2) @@ -181,7 +181,7 @@ class RequestThrottleSpec extends AnyFunSuite { // with a cap of 1 the gate is dead, and every later request waits forever // for a permit nobody holds. Losing a request is recoverable; losing the // gate is not. - val throttle = new RequestThrottle(maxConcurrent = 1, minIntervalMs = 100, FailingClock) + val throttle = new RequestThrottle(1, 100, FailingClock) intercept[IllegalStateException] { Await.result(throttle.withPermit(Future.successful("never runs")), 5.seconds) @@ -197,7 +197,7 @@ class RequestThrottleSpec extends AnyFunSuite { // SystemThrottleClock.sleep schedules on an executor, which throws // RejectedExecutionException rather than returning a failed Future if that // executor is shut down. That path must not lose the permit either. - val throttle = new RequestThrottle(maxConcurrent = 1, minIntervalMs = 100, ThrowingClock) + val throttle = new RequestThrottle(1, 100, ThrowingClock) intercept[IllegalStateException] { Await.result(throttle.withPermit(Future.successful("never runs")), 5.seconds) @@ -211,7 +211,7 @@ class RequestThrottleSpec extends AnyFunSuite { test("pacing is skipped entirely when the interval is zero") { // Pacing off must cost nothing — no scheduled delays at all. val clock = new ManualClock() - val throttle = new RequestThrottle(maxConcurrent = 4, minIntervalMs = 0, clock) + val throttle = new RequestThrottle(4, 0, clock) val calls = (1 to 3).map(_ => throttle.withPermit(Future.successful(()))) Await.result(Future.sequence(calls), 10.seconds) @@ -311,14 +311,72 @@ class RequestThrottleSpec extends AnyFunSuite { assert(!canonical.eq(gate("https://other.normalize.test/v1")), "a different host is a different endpoint") } - test("changing the configured cap replaces the gate") { + test("changing the configured cap reconfigures the same gate in place") { val url = "http://identity-change.local" - val first = RequestThrottle.forProvider("cfgchange", url, Some("2"), None).getOrElse(fail()) + val first = RequestThrottle.forProvider("cfgchange", url, Some("2"), Some("100")).getOrElse(fail()) assert(first.maxConcurrent == 2) - val second = RequestThrottle.forProvider("cfgchange", url, Some("5"), None).getOrElse(fail()) + val second = RequestThrottle.forProvider("cfgchange", url, Some("5"), Some("250")).getOrElse(fail()) assert(second.maxConcurrent == 5, "a changed cap must take effect") - assert(!first.eq(second)) + assert(second.minIntervalMs == 250L, "a changed interval must take effect") + assert(first.eq(second), "the endpoint keeps one gate so in-flight permits stay counted") + } + + /** Poll until `cond` holds or `limit` passes; release() runs on a callback thread. */ + private def eventually(limit: FiniteDuration = 2.seconds)(cond: => Boolean): Boolean = { + val deadline = System.nanoTime() + limit.toNanos + while (!cond && System.nanoTime() < deadline) Thread.sleep(5) + cond + } + + test("shrinking the cap under in-flight requests admits nothing until they drain below it") { + // Two profiles on one endpoint with caps 2 and 1, alternating calls: the + // gate must not be swapped for a fresh one with a full set of permits. + val url = "http://cap-shrink.local" + val gate = RequestThrottle.forProvider("capshrink", url, Some("2"), None).getOrElse(fail()) + val hold1 = Promise[Unit]() + val hold2 = Promise[Unit]() + val f1 = gate.withPermit(hold1.future) + val f2 = gate.withPermit(hold2.future) + assert(gate.inFlight == 2) + + val same = RequestThrottle.forProvider("capshrink", url, Some("1"), None).getOrElse(fail()) + assert(same.eq(gate)) + @volatile var thirdStarted = false + val f3 = same.withPermit { thirdStarted = true; Future.unit } + assert(same.waiting == 1, "the third request must queue: two are in flight above the new cap of 1") + assert(!thirdStarted) + + hold1.success(()) + Await.ready(f1, 1.second) + assert(eventually()(same.inFlight == 1)) + assert(!thirdStarted, "one still in flight fills a cap of 1") + + hold2.success(()) + Await.ready(f2, 1.second) + Await.ready(f3, 1.second) + assert(thirdStarted) + assert(eventually()(same.inFlight == 0 && same.waiting == 0)) + } + + test("growing the cap hands the new permits to waiters immediately") { + val url = "http://cap-grow.local" + val gate = RequestThrottle.forProvider("capgrow", url, Some("1"), None).getOrElse(fail()) + val hold = Promise[Unit]() + val f1 = gate.withPermit(hold.future) + @volatile var secondStarted = false + val f2 = gate.withPermit { secondStarted = true; Future.unit } + assert(gate.waiting == 1) + + val same = RequestThrottle.forProvider("capgrow", url, Some("2"), None).getOrElse(fail()) + assert(same.eq(gate)) + Await.ready(f2, 1.second) + assert(secondStarted, "raising the cap must admit the waiter without waiting for a release") + assert(same.waiting == 0) + + hold.success(()) + Await.ready(f1, 1.second) + assert(eventually()(same.inFlight == 0)) } /** Run `body`, returning whatever it wrote to stderr. */ diff --git a/tests.txt b/tests.txt index 0130a0b..b3545e9 100644 --- a/tests.txt +++ b/tests.txt @@ -1005,3 +1005,44 @@ LLMProfileHistoryAndUsageStayPerAgent [llm:get llm:usage "input-tokens"] of turtle 1 => 5 llm:get llm:usage-total "calls" => 2 O> clear-all + +LLMProfileWithoutModelUsesProviderDefault + extensions [llm] + O> llm:set-api-key "test-key" + O> llm:set-provider "openai" + O> clear-all + O> llm:load-profile "claude" "demos/test-profile-anthropic" + O> crt 1 + O> ask turtle 0 [ llm:use-profile "claude" ] + [llm:active] of turtle 0 => ["anthropic" "claude-haiku-4-5-20251001"] + O> clear-all + +LLMProfileUsesItsOwnWaitBudget + extensions [llm] + globals [err r] + O> llm:set-api-key "test-key" + O> llm:set-provider "openai" + O> clear-all + O> llm:load-profile "slow" "demos/test-profile-slow" + O> crt 1 + O> ask turtle 0 [ llm:use-profile "slow" ] + O> ask turtle 0 [ carefully [ set r llm:chat "__TEST_DELAY:1500:hi" ] [ set err error-message ] ] + member? "timed out" err => true + O> set r llm:chat "__TEST_DELAY:1500:hi" + r => "stub:hi" + O> clear-all + +LLMAsyncKeepsTheBudgetItStartedWith + extensions [llm] + globals [p r] + O> llm:set-api-key "test-key" + O> llm:set-provider "openai" + O> clear-all + O> llm:load-profile "a" "demos/test-profile-a" + O> crt 1 + O> ask turtle 0 [ llm:use-profile "a" ] + O> ask turtle 0 [ set p llm:chat-async "__TEST_DELAY:1500:hi" ] + O> llm:load-config "demos/test-config-fast-timeout" + O> set r (runresult p) + r => "stub:hi" + O> clear-all