fix: rework reentrant concurrency model for Java interop (#1945)

## Problem

For more stable Java interop, we need to properly handle the reentrant
scenario `Suspendable Koog code` → `Blocking user Java code` →
`Suspendable Koog code`. Naively wrapping Java-interacting components in
`runBlocking(context)` or `withContext(context)` introduces a risk of
deadlock or thread starvation.

Consider the following example:

```kotlin
val singleThreadDispatcher = Executors.newSingleThreadExecutor().asCoroutineDispatcher()

fun suspendableKotlinCodeOne() {
  runBlocking(singleThreadDispatcher) {
    blockingJavaCodeTwo()
  }
}

fun suspendableKotlinCodeThree() {
  runBlocking(singleThreadDispatcher) {
    // do something
  }
}
```

```java
public void blockingJavaCodeTwo() {
  suspendableKotlinCodeThree();
}
```

`suspendableKotlinCodeOne` dispatches a block onto
`singleThreadDispatcher`, which then calls into `blockingJavaCodeTwo`.
The Java method calls back into Kotlin via `suspendableKotlinCodeThree`,
which again tries to dispatch onto the same single-threaded executor.

Even though execution stays on the same thread throughout, the coroutine
context tracking this is lost when crossing the Kotlin → Java → Kotlin
boundary. As a result, `suspendableKotlinCodeThree` cannot tell that it
is already running on the correct context and attempts to redispatch —
causing a deadlock, since the only thread is already occupied. The
redispatch was never actually necessary; the inner block could have
executed directly in place.

## Current state

The solution is to store the coroutine context in a thread local, then
check it on dispatch to determine whether the requested context is
already active. Several previous attempts have tried to implement this
and fix related bugs, the most recent being #1716. However, the current
implementation still has major problems:

* After the last fix, `runBlockingIfRequired` ignores its `context`
argument, meaning it never dispatches to the requested dispatcher.
* Thread-local-based dispatcher tracking stores the context on the wrong
thread — the current thread running `runBlockingIfRequired`, rather than
the thread associated with the supplied `context`.
* There is a bunch of similar-looking helper functions implemented
separately, leading to duplication. The naming is also confusing.

## Solution

I've narrowed the scope down to two functions with (hopefully) clearer
names: `runBlockingReentrant` and `withContextReentrant`. They implement
the approach described above correctly, avoiding unnecessary dispatch. A
few tests have been added to verify the behavior. All existing utility
functions were deleted and all implementations have been migrated.

## Important note

The case where a user manually submits a task to the executor outside of
Koog — for example, running the agent itself on the same single-threaded
executor — is deliberately out of scope. In such cases, it is
technically impossible to reliably determine whether we are already on
the same executor/thread. This should be treated as user error rather
than something Koog tries to handle.

DEPRECATED:
* Constructors and methods in `AIAgentConfig` on the JVM that accept
`ExecutorService` parameters. Constructors and methods that accept more
generic `Executor` should be used instead.

closes [KG-750](https://youtrack.jetbrains.com/issue/KG-750)

Author: EugeneTheDev

agents/agents-core/src/jvmCommonMain/kotlin/ai/koog/agents/core/agent/AIAgent.kt

@@ -10,12 +10,13 @@ import ai.koog.agents.core.agent.entity.AIAgentGraphStrategy
 import ai.koog.agents.core.agent.session.AIAgentRunSession
 import ai.koog.agents.core.annotation.InternalAgentsApi
 import ai.koog.agents.core.tools.ToolRegistry
-import ai.koog.agents.core.utils.runOnStrategyDispatcher
+import ai.koog.agents.core.utils.runBlockingOnLLMDispatcher
 import ai.koog.agents.planner.AIAgentPlannerStrategy
 import ai.koog.agents.planner.PlannerAIAgent
 import ai.koog.prompt.executor.model.PromptExecutor
 import ai.koog.prompt.llm.LLModel
 import ai.koog.prompt.processor.ResponseProcessor
+import ai.koog.utils.annotations.InternalKoogUtils
 import ai.koog.utils.io.Closeable
 import ai.koog.utils.time.KoogClock
 import java.util.concurrent.ExecutorService
@@ -37,14 +38,17 @@ public actual abstract class AIAgent<Input, Output> : Closeable {
      *                        If not provided, the default coroutine context is used.
      * @return The output resulting from the execution of the AI agent with the given input.
      */
+    @OptIn(InternalKoogUtils::class)
     @JavaAPI
     @JvmOverloads
     @JvmName("run")
     public final fun javaNonSuspendRun(
         agentInput: Input,
         sessionId: String? = null,
         executorService: ExecutorService? = null
-    ): Output = agentConfig.runOnStrategyDispatcher(executorService) { run(agentInput, sessionId) }
+    ): Output = agentConfig.runBlockingOnLLMDispatcher(executorService) {
+        run(agentInput, sessionId)
+    }
 
     // Common (multiplatform) methods:
     public actual abstract suspend fun run(agentInput: Input, sessionId: String?): Output

agents/agents-core/src/jvmCommonMain/kotlin/ai/koog/agents/core/agent/AIAgentService.kt

@@ -1,5 +1,4 @@
 @file:Suppress("EXPECT_ACTUAL_CLASSIFIERS_ARE_IN_BETA_WARNING", "MissingKDocForPublicAPI")
-@file:OptIn(InternalAgentsApi::class)
 
 package ai.koog.agents.core.agent
 
@@ -8,13 +7,14 @@ import ai.koog.agents.core.agent.config.AIAgentConfig
 import ai.koog.agents.core.agent.entity.AIAgentGraphStrategy
 import ai.koog.agents.core.annotation.InternalAgentsApi
 import ai.koog.agents.core.tools.ToolRegistry
-import ai.koog.agents.core.utils.runOnStrategyDispatcher
+import ai.koog.agents.core.utils.runBlockingOnStrategyDispatcher
 import ai.koog.prompt.executor.model.PromptExecutor
 import ai.koog.prompt.llm.LLModel
 import ai.koog.prompt.processor.ResponseProcessor
 import ai.koog.utils.time.KoogClock
 import java.util.concurrent.ExecutorService
 
+@OptIn(InternalAgentsApi::class)
 public actual abstract class AIAgentService<Input, Output, TAgent : AIAgent<Input, Output>> {
     public actual abstract val promptExecutor: PromptExecutor
     public actual abstract val agentConfig: AIAgentConfig
@@ -56,7 +56,7 @@ public actual abstract class AIAgentService<Input, Output, TAgent : AIAgent<Inpu
         agentConfig: AIAgentConfig = this.agentConfig,
         executorService: ExecutorService? = null,
         clock: KoogClock = KoogClock.System
-    ): TAgent = agentConfig.runOnStrategyDispatcher(executorService) {
+    ): TAgent = agentConfig.runBlockingOnStrategyDispatcher(executorService) {
         createAgent(id, additionalToolRegistry, agentConfig, clock)
     }
 
@@ -98,7 +98,7 @@ public actual abstract class AIAgentService<Input, Output, TAgent : AIAgent<Inpu
     public fun removeAgent(
         agent: TAgent,
         executorService: ExecutorService? = null
-    ): Boolean = agentConfig.runOnStrategyDispatcher(executorService) {
+    ): Boolean = agentConfig.runBlockingOnStrategyDispatcher(executorService) {
         removeAgent(agent)
     }
 
@@ -114,7 +114,7 @@ public actual abstract class AIAgentService<Input, Output, TAgent : AIAgent<Inpu
     public fun removeAgentWithId(
         id: String,
         executorService: ExecutorService? = null
-    ): Boolean = agentConfig.runOnStrategyDispatcher(executorService) {
+    ): Boolean = agentConfig.runBlockingOnStrategyDispatcher(executorService) {
         removeAgentWithId(id)
     }
 
@@ -133,59 +133,10 @@ public actual abstract class AIAgentService<Input, Output, TAgent : AIAgent<Inpu
     public fun agentById(
         id: String,
         executorService: ExecutorService? = null
-    ): TAgent? = agentConfig.runOnStrategyDispatcher(executorService) {
+    ): TAgent? = agentConfig.runBlockingOnStrategyDispatcher(executorService) {
         agentById(id)
     }
 
-    /**
-     * Lists all currently active agents using the configured strategy dispatcher.
-     *
-     * If an optional `executorService` is provided, it will be used as the execution context
-     * for running the operation. Otherwise, a default executor service or dispatcher is utilized.
-     *
-     * @param executorService The optional `ExecutorService` to use for task execution. Defaults to `null`.
-     * @return A list of currently active agents of type `TAgent`.
-     */
-    @JavaAPI
-    @JvmOverloads
-    public fun listActiveAgents(
-        executorService: ExecutorService? = null
-    ): List<TAgent> = agentConfig.runOnStrategyDispatcher(executorService) {
-        listActiveAgents()
-    }
-
-    /**
-     * Retrieves a list of all inactive agents.
-     *
-     * This method utilizes the provided `ExecutorService` for dispatching, if specified.
-     * If no `ExecutorService` is provided, a default strategy-specific executor will be used.
-     *
-     * @param executorService An optional `ExecutorService` to execute the operation.
-     *                        If `null`, the default executor service of the agent's configuration is used.
-     * @return A list of `TAgent` objects representing the inactive agents.
-     */
-    @JavaAPI
-    @JvmOverloads
-    public fun listInactiveAgents(
-        executorService: ExecutorService? = null
-    ): List<TAgent> = agentConfig.runOnStrategyDispatcher(executorService) {
-        listInactiveAgents()
-    }
-
-    /**
-     * Retrieves a list of finished agents.
-     *
-     * @param executorService The optional executor service to be used for the operation. If no executor service is provided, a default one will be used.
-     * @return A list of finished agents.
-     */
-    @JavaAPI
-    @JvmOverloads
-    public fun listFinishedAgents(
-        executorService: ExecutorService? = null
-    ): List<TAgent> = agentConfig.runOnStrategyDispatcher(executorService) {
-        listFinishedAgents()
-    }
-
     public actual companion object {
         @JvmStatic
         public actual fun builder(): AIAgentServiceBuilder = AIAgentServiceBuilder()