Skip to content

Commit 9dd9d26

Browse files
committed
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.
1 parent 9debdac commit 9dd9d26

8 files changed

Lines changed: 217 additions & 46 deletions

File tree

‎demos/test-config-fast-timeout‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,6 @@
1+
# Test fixture: global config with a one-second wait budget and no retry allowance
2+
provider=openai
3+
openai_api_key=test-key
4+
model=model-fast-timeout
5+
timeout_seconds=1
6+
retry_max_elapsed_seconds=0

‎demos/test-profile-anthropic‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,3 @@
1+
# Test fixture: Anthropic profile with no model, so the provider default must apply
2+
provider=anthropic
3+
anthropic_api_key=test-key

‎demos/test-profile-slow‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,6 @@
1+
# Test fixture: profile with a one-second wait budget and no retry allowance
2+
provider=openai
3+
openai_api_key=test-key
4+
model=model-slow
5+
timeout_seconds=1
6+
retry_max_elapsed_seconds=0

‎docs/API-REFERENCE.md‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -920,6 +920,12 @@ exists replaces it. A rejected reload leaves the existing profile untouched.
920920
setters act on the default configuration only. A profile is what its file
921921
says. To change a profile, edit the file and load it again.
922922
- Two profiles on the same provider and endpoint share one request throttle.
923+
If their `max_concurrent_requests` or `min_request_interval_ms` differ, the
924+
shared throttle takes whichever setting the most recent call carried, without
925+
losing count of requests already in flight. Give profiles on one endpoint the
926+
same throttling settings.
927+
- A profile's `timeout_seconds` and retry settings govern calls made through
928+
it. An async call keeps the budget in force when it was launched.
923929

924930
### llm:use-profile
925931

‎src/main/LLMExtension.scala‎

Lines changed: 23 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -82,13 +82,16 @@ class LLMExtension extends DefaultClassManager {
8282
* Create an AwaitableReporter that wraps a Future to provide truly async behavior
8383
* The Future starts immediately but execution defers until runresult is called
8484
*/
85-
private def createAwaitableReporter(future: Future[String]): AnonymousReporter = {
85+
private def createAwaitableReporter(future: Future[String], timeout: FiniteDuration): AnonymousReporter = {
8686
new AnonymousReporter {
8787
override def syntax: Syntax = Syntax.reporterSyntax(right = List(), ret = Syntax.StringType)
8888

8989
override def report(context: Context, args: Array[AnyRef]): AnyRef = {
9090
try {
91-
Await.result(future, getAwaitTimeout)
91+
// The budget in force for the calling agent when the request was
92+
// launched: a later config change or profile rebinding cannot alter
93+
// how long an already-pending call may take.
94+
Await.result(future, timeout)
9295
} catch {
9396
case e: Exception =>
9497
throw new ExtensionException(s"Async LLM operation failed: ${e.getMessage}")
@@ -271,8 +274,8 @@ class LLMExtension extends DefaultClassManager {
271274
/**
272275
* Get timeout from config, falling back to 30 seconds
273276
*/
274-
private def getTimeoutSeconds: Int =
275-
configStore.get(ConfigStore.TIMEOUT_SECONDS).map { s =>
277+
private def getTimeoutSeconds(config: ConfigStore): Int =
278+
config.get(ConfigStore.TIMEOUT_SECONDS).map { s =>
276279
scala.util.Try(s.toInt).getOrElse {
277280
System.err.println(s"WARNING: Invalid timeout_seconds value '$s' (not a valid integer), using default 30")
278281
30
@@ -301,13 +304,13 @@ class LLMExtension extends DefaultClassManager {
301304
* bound is unchanged from before throttling existed, so an unthrottled model — the
302305
* default — behaves exactly as it did.
303306
*/
304-
private def getAwaitTimeout: FiniteDuration = {
305-
val retryBudget = configStore.get(RetryPolicy.MAX_ELAPSED_SECONDS)
307+
private def awaitTimeoutFor(config: ConfigStore): FiniteDuration = {
308+
val retryBudget = config.get(RetryPolicy.MAX_ELAPSED_SECONDS)
306309
.flatMap(s => scala.util.Try(s.trim.toDouble).toOption)
307310
.filter(d => d >= 0.0 && d.isFinite)
308311
.map(d => (d * 1000.0).toLong.millis)
309312
.getOrElse(RetryPolicy.DefaultMaxElapsed)
310-
getTimeoutSeconds.seconds + retryBudget
313+
getTimeoutSeconds(config).seconds + retryBudget
311314
}
312315

313316
/**
@@ -599,7 +602,7 @@ class LLMExtension extends DefaultClassManager {
599602

600603
// Send chat request with user message included, but don't mutate history yet
601604
val responseFuture = provider.chatWithFullResponse(snapshotHistory(agent) :+ userMessage)
602-
val response = Await.result(responseFuture, getAwaitTimeout)
605+
val response = Await.result(responseFuture, awaitTimeoutFor(effectiveConfig(agent)))
603606
recordUsage(agent, response)
604607
val responseMessage = replyMessage(response)
605608

@@ -640,7 +643,7 @@ class LLMExtension extends DefaultClassManager {
640643
}
641644

642645
// Return AnonymousReporter that wraps the Future
643-
createAwaitableReporter(responseFuture)
646+
createAwaitableReporter(responseFuture, awaitTimeoutFor(effectiveConfig(agent)))
644647

645648
} catch {
646649
case e: Exception =>
@@ -692,7 +695,7 @@ class LLMExtension extends DefaultClassManager {
692695

693696
// Send chat request
694697
val responseFuture = provider.chatWithFullResponse(tempHistory.toSeq)
695-
val response = Await.result(responseFuture, getAwaitTimeout)
698+
val response = Await.result(responseFuture, awaitTimeoutFor(effectiveConfig(agent)))
696699
recordUsage(agent, response)
697700
val responseMessage = replyMessage(response)
698701

@@ -754,7 +757,7 @@ class LLMExtension extends DefaultClassManager {
754757
// The prompt still spells out the options, so a provider that ignores
755758
// the constraint behaves exactly as it did before.
756759
val responseFuture = provider.chatWithFormat(tempHistory.toSeq, EnumFormat(choices))
757-
val response = Await.result(responseFuture, getAwaitTimeout)
760+
val response = Await.result(responseFuture, awaitTimeoutFor(effectiveConfig(agent)))
758761
recordUsage(agent, response)
759762

760763
// Extract text: prefer content, fall back to thinking field
@@ -854,7 +857,7 @@ class LLMExtension extends DefaultClassManager {
854857
val userMessage = ChatMessage.user(inputText)
855858

856859
val responseFuture = provider.chatWithFormat(snapshotHistory(agent) :+ userMessage, format)
857-
val response = Await.result(responseFuture, getAwaitTimeout)
860+
val response = Await.result(responseFuture, awaitTimeoutFor(effectiveConfig(agent)))
858861
recordUsage(agent, response)
859862

860863
val content = response.firstContent.getOrElse("")
@@ -915,7 +918,7 @@ class LLMExtension extends DefaultClassManager {
915918
tempHistory += userMessage
916919

917920
val responseFuture = provider.chatWithFormat(tempHistory.toSeq, JsonObjectFormat)
918-
val response = Await.result(responseFuture, getAwaitTimeout)
921+
val response = Await.result(responseFuture, awaitTimeoutFor(effectiveConfig(agent)))
919922
recordUsage(agent, response)
920923

921924
val content = response.firstContent.getOrElse("")
@@ -998,7 +1001,7 @@ class LLMExtension extends DefaultClassManager {
9981001

9991002
// Send with user message included, but don't mutate history yet
10001003
val responseFuture = provider.chatWithFullResponse(snapshotHistory(agent) :+ userMessage)
1001-
val response = Await.result(responseFuture, getAwaitTimeout)
1004+
val response = Await.result(responseFuture, awaitTimeoutFor(effectiveConfig(agent)))
10021005
recordUsage(agent, response)
10031006

10041007
val answerText = response.firstContent.getOrElse("")
@@ -1335,11 +1338,14 @@ class LLMExtension extends DefaultClassManager {
13351338
)
13361339
}
13371340

1338-
val candidate = ConfigStore.withDefaults()
1339-
candidate.updateFromMap(config)
1341+
// Mirror llm:load-config: the file is the whole configuration, with no
1342+
// OpenAI-flavoured defaults layered underneath. A profile that names no
1343+
// model gets its own provider's default, not gpt-4o-mini.
1344+
val candidate = new ConfigStore()
1345+
candidate.loadFromMap(config)
13401346
checkReadiness(desc, providerName, candidate)
13411347

1342-
try profiles.load(name, candidate.toMap)
1348+
try profiles.load(name, config)
13431349
catch {
13441350
case e: IllegalArgumentException =>
13451351
throw new ExtensionException(s"llm:load-profile: ${e.getMessage}")

‎src/main/providers/RequestThrottle.scala‎

Lines changed: 62 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -67,16 +67,53 @@ object SystemThrottleClock extends ThrottleClock {
6767
* @param clock time source and delay scheduler
6868
*/
6969
class RequestThrottle(
70-
val maxConcurrent: Int,
71-
val minIntervalMs: Long,
70+
initialMaxConcurrent: Int,
71+
initialMinIntervalMs: Long,
7272
clock: ThrottleClock = SystemThrottleClock
7373
) {
74-
require(maxConcurrent > 0, "maxConcurrent must be positive")
74+
require(initialMaxConcurrent > 0, "maxConcurrent must be positive")
7575

76-
// Guards `available`, `waiters` and `lastStartMs` together. Held only for
77-
// queue arithmetic — never across a user callback or an HTTP send.
76+
// Guards `available`, `waiters`, `lastStartMs` and the limits together. Held
77+
// only for queue arithmetic — never across a user callback or an HTTP send.
7878
private val lock = new Object
79-
private var available: Int = maxConcurrent
79+
@volatile private var currentMax: Int = initialMaxConcurrent
80+
@volatile private var currentInterval: Long = initialMinIntervalMs
81+
private var available: Int = initialMaxConcurrent
82+
83+
def maxConcurrent: Int = currentMax
84+
def minIntervalMs: Long = currentInterval
85+
86+
/** Requests holding a permit right now. Can exceed the cap briefly after it shrinks. */
87+
def inFlight: Int = lock.synchronized { currentMax - available }
88+
89+
/** Requests queued for a permit. */
90+
def waiting: Int = lock.synchronized { waiters.size }
91+
92+
/**
93+
* Change the limits without replacing the gate.
94+
*
95+
* Two profiles on one endpoint can carry different settings, and calls
96+
* alternate between them every tick. Swapping in a fresh gate per change
97+
* would hand every call a full set of permits and the cap would mean
98+
* nothing. Adjusting in place keeps the in-flight count and the queue: a
99+
* smaller cap admits nothing until requests drain below it, a larger one
100+
* hands the new permits to the oldest waiters at once.
101+
*/
102+
def reconfigure(newMaxConcurrent: Int, newMinIntervalMs: Long): Unit = {
103+
require(newMaxConcurrent > 0, "maxConcurrent must be positive")
104+
val admitted = lock.synchronized {
105+
available += newMaxConcurrent - currentMax
106+
currentMax = newMaxConcurrent
107+
currentInterval = newMinIntervalMs
108+
val handoffs = mutable.ListBuffer.empty[Promise[Unit]]
109+
while (available > 0 && waiters.nonEmpty) {
110+
available -= 1
111+
handoffs += waiters.dequeue()
112+
}
113+
handoffs.toList
114+
}
115+
admitted.foreach(_.success(()))
116+
}
80117

81118
// FIFO, so a waiter cannot be overtaken indefinitely — with agents calling
82119
// every tick, barging would pass one over for the whole run. Only ever
@@ -155,7 +192,7 @@ class RequestThrottle(
155192
* exceed an RPM limit.
156193
*/
157194
private def paceThenProceed()(implicit ec: ExecutionContext): Future[Unit] = {
158-
if (minIntervalMs <= 0L) return Future.unit
195+
if (currentInterval <= 0L) return Future.unit
159196

160197
val waitMs = lock.synchronized {
161198
val now = clock.nowMs
@@ -164,7 +201,7 @@ class RequestThrottle(
164201
// timestamp and starting together.
165202
val earliest =
166203
if (lastStartMs == Long.MinValue) now
167-
else math.max(now, lastStartMs + minIntervalMs)
204+
else math.max(now, lastStartMs + currentInterval)
168205
lastStartMs = earliest
169206
earliest - now
170207
}
@@ -193,7 +230,11 @@ class RequestThrottle(
193230
*/
194231
private def release(): Unit = {
195232
val handoff = lock.synchronized {
196-
if (waiters.nonEmpty) Some(waiters.dequeue())
233+
// A negative `available` means the cap shrank while more than the new
234+
// limit were in flight. Those releases repay the deficit; nobody is
235+
// admitted until the count is back under the cap.
236+
if (available < 0) { available += 1; None }
237+
else if (waiters.nonEmpty) Some(waiters.dequeue())
197238
else { available += 1; None }
198239
}
199240
handoff.foreach(_.success(()))
@@ -304,23 +345,27 @@ object RequestThrottle {
304345
// or race two replacements past each other. The remapping function does
305346
// no I/O — it runs while the map holds a bin lock, so the notice below is
306347
// emitted afterwards rather than inside it.
348+
// One gate per endpoint for the life of the process. A changed setting
349+
// reconfigures it in place rather than replacing it, so requests already
350+
// in flight stay counted — two profiles with different caps alternating
351+
// on one endpoint would otherwise each get a fresh, empty gate.
307352
val throttle = throttles.compute(key, (_, existing) => {
308-
if (existing != null && existing.maxConcurrent == limit && existing.minIntervalMs == interval) {
309-
existing
310-
} else {
311-
if (existing != null) replaced.set(true)
353+
if (existing == null) {
312354
new RequestThrottle(limit, interval)
355+
} else {
356+
if (existing.maxConcurrent != limit || existing.minIntervalMs != interval) {
357+
existing.reconfigure(limit, interval)
358+
replaced.set(true)
359+
}
360+
existing
313361
}
314362
})
315363

316364
if (replaced.get()) {
317-
// Requests already running under the previous gate keep its permits, so
318-
// in-flight work can briefly exceed the new cap. Reporting beats a stall
319-
// or a silently changed limit.
320365
warnOnce(
321366
s"NOTE: request throttle for $providerName updated to " +
322367
s"$MAX_CONCURRENT_REQUESTS=$limit, $MIN_REQUEST_INTERVAL_MS=$interval. " +
323-
"Requests already in flight finish under the previous limit."
368+
"Requests already in flight count against the new limit."
324369
)
325370
}
326371
throttle

0 commit comments

Comments
 (0)