feat: add Kotlin coroutine extension layer (Mono -> suspend, Flux -> Flow)

agentscope-core/src/main/java/io/agentscope/core/ReActAgent.java

@@ -680,12 +680,17 @@ public int priority() {
      */
     private class HookNotifier {
 
+        private List<Hook> getSortedHooks() {
+            return ReActAgent.this.getSortedHooks();
+        }
+
         Mono<List<Msg>> notifyPreReasoning(AgentBase agent, List<Msg> msgs) {
             PreReasoningEvent event =
                     new PreReasoningEvent(agent, model.getModelName(), null, msgs);
+
             Mono<PreReasoningEvent> result = Mono.just(event);
             for (Hook hook : getSortedHooks()) {
-                result = result.flatMap(e -> hook.onEvent(e));
+                result = result.flatMap(hook::onEvent);
             }
             return result.map(PreReasoningEvent::getInputMessages);
         }
@@ -694,9 +699,10 @@ Mono<Msg> notifyPostReasoning(Msg reasoningMsg) {
             PostReasoningEvent event =
                     new PostReasoningEvent(
                             ReActAgent.this, model.getModelName(), null, reasoningMsg);
+
             Mono<PostReasoningEvent> result = Mono.just(event);
             for (Hook hook : getSortedHooks()) {
-                result = result.flatMap(e -> hook.onEvent(e));
+                result = result.flatMap(hook::onEvent);
             }
             return result.map(PostReasoningEvent::getReasoningMessage);
         }
@@ -705,28 +711,41 @@ Mono<Void> notifyReasoningChunk(Msg chunk, Msg accumulated) {
             ReasoningChunkEvent event =
                     new ReasoningChunkEvent(
                             ReActAgent.this, model.getModelName(), null, chunk, accumulated);
-            return Flux.fromIterable(getSortedHooks()).flatMap(hook -> hook.onEvent(event)).then();
+
+            Mono<ReasoningChunkEvent> result = Mono.just(event);
+            for (Hook hook : getSortedHooks()) {
+                result = result.flatMap(hook::onEvent);
+            }
+            return result.then();
         }
 
         Mono<ToolUseBlock> notifyPreActing(ToolUseBlock toolUse) {
             PreActingEvent event = new PreActingEvent(ReActAgent.this, toolkit, toolUse);
+
             Mono<PreActingEvent> result = Mono.just(event);
             for (Hook hook : getSortedHooks()) {
-                result = result.flatMap(e -> hook.onEvent(e));
+                result = result.flatMap(hook::onEvent);
             }
             return result.map(PreActingEvent::getToolUse);
         }
 
         Mono<Void> notifyActingChunk(ToolUseBlock toolUse, ToolResultBlock chunk) {
             ActingChunkEvent event = new ActingChunkEvent(ReActAgent.this, toolkit, toolUse, chunk);
-            return Flux.fromIterable(getSortedHooks()).flatMap(hook -> hook.onEvent(event)).then();
+
+            Mono<ActingChunkEvent> result = Mono.just(event);
+            for (Hook hook : getSortedHooks()) {
+                result = result.flatMap(hook::onEvent);
+            }
+            return result.then();
         }
 
         Mono<ToolResultBlock> notifyPostActing(ToolUseBlock toolUse, ToolResultBlock toolResult) {
-            var event = new PostActingEvent(ReActAgent.this, toolkit, toolUse, toolResult);
+            PostActingEvent event =
+                    new PostActingEvent(ReActAgent.this, toolkit, toolUse, toolResult);
+
             Mono<PostActingEvent> result = Mono.just(event);
             for (Hook hook : getSortedHooks()) {
-                result = result.flatMap(e -> hook.onEvent(e));
+                result = result.flatMap(hook::onEvent);
             }
             return result.map(PostActingEvent::getToolResult);
         }

agentscope-core/src/main/java/io/agentscope/core/agent/AgentBase.java

@@ -25,6 +25,7 @@
 import io.agentscope.core.state.StateModuleBase;
 import io.agentscope.core.tracing.TracerRegistry;
 import java.util.ArrayList;
+import java.util.Comparator;
 import java.util.List;
 import java.util.Map;
 import java.util.UUID;
@@ -91,7 +92,12 @@ public abstract class AgentBase extends StateModuleBase implements Agent {
     private final AtomicBoolean running = new AtomicBoolean(false);
     private final boolean checkRunning;
     private final List<Hook> hooks;
+    // Cached sorted hooks (invalidated when hooks list changes)
+    private transient volatile List<Hook> cachedSortedHooks;
+    private final AtomicBoolean hooksDirty = new AtomicBoolean(true);
+
     private static final List<Hook> systemHooks = new CopyOnWriteArrayList<>();
+
     private final Map<String, List<AgentBase>> hubSubscribers = new ConcurrentHashMap<>();
 
     // Interrupt state management (available to all agents)
@@ -240,6 +246,8 @@ protected Mono<Msg> doCall(List<Msg> msgs, Class<?> structuredOutputClass) {
                         "Structured output not supported by " + getClass().getSimpleName()));
     }
 
+    // Note: system hooks are applied at agent construction time;
+    // dynamic system hook changes do not affect existing agents.
     public static void addSystemHook(Hook hook) {
         systemHooks.add(hook);
     }
@@ -395,22 +403,66 @@ protected Mono<Void> doObserve(Msg msg) {
 
     /**
      * Get the list of hooks for this agent.
-     * Protected to allow subclasses to access hooks for custom notification logic.
      *
-     * @return List of hooks
+     * <p>Returns an immutable snapshot of the internal hook list.
+     * Callers must not attempt to modify the returned list.
+     * To add or remove hooks, use {@link #addHook(Hook)} or
+     * {@link #removeHook(Hook)}.
+     *
+     * <p>This is a breaking change from previous behavior where
+     * callers could mutate the returned list directly.
+     *
+     * @return Immutable list of hooks
      */
     public List<Hook> getHooks() {
-        return hooks;
+        return List.copyOf(hooks);
+    }
+
+    /**
+     * Add a hook to this agent.
+     *
+     * <p>Hooks should generally be added during agent setup,
+     * before execution begins. Modifying hooks during execution
+     * is not thread-safe and may lead to undefined behavior.
+     *
+     * @param hook Hook to add
+     */
+    public void addHook(Hook hook) {
+        hooks.add(hook);
+        hooksDirty.set(true);
+    }
+
+    /**
+     * Remove a hook from this agent.
+     *
+     * <p>Hooks should generally be removed during agent setup.
+     * Modifying hooks during execution is not thread-safe.
+     *
+     * @param hook Hook to remove
+     */
+    public void removeHook(Hook hook) {
+        hooks.remove(hook);
+        hooksDirty.set(true);
     }
 
     /**
      * Get hooks sorted by priority (lower value = higher priority).
      * Hooks with the same priority maintain registration order.
      *
+     * <p>Results may be cached until the hook list changes.
+     *
      * @return Sorted list of hooks
      */
     protected List<Hook> getSortedHooks() {
-        return hooks.stream().sorted(java.util.Comparator.comparingInt(Hook::priority)).toList();
+        if (!hooksDirty.get() && cachedSortedHooks != null) {
+            return cachedSortedHooks;
+        }
+
+        List<Hook> sorted = hooks.stream().sorted(Comparator.comparingInt(Hook::priority)).toList();
+
+        cachedSortedHooks = sorted;
+        hooksDirty.set(false);
+        return sorted;
     }
 
     /**
@@ -460,7 +512,11 @@ private Mono<Msg> notifyPostCall(Msg finalMsg) {
      */
     private Mono<Void> notifyError(Throwable error) {
         ErrorEvent event = new ErrorEvent(this, error);
-        return Flux.fromIterable(getSortedHooks()).flatMap(hook -> hook.onEvent(event)).then();
+        Mono<ErrorEvent> result = Mono.just(event);
+        for (Hook hook : getSortedHooks()) {
+            result = result.flatMap(hook::onEvent);
+        }
+        return result.then();
     }
 
     /**
@@ -626,15 +682,15 @@ private Flux<Event> createEventStream(StreamOptions options, Supplier<Mono<Msg>>
                             StreamingHook streamingHook = new StreamingHook(sink, options);
 
                             // Add temporary hook
-                            hooks.add(streamingHook);
+                            addHook(streamingHook);
 
                             // Execute call and manage hook lifecycle
                             callSupplier
                                     .get()
                                     .doFinally(
                                             signalType -> {
                                                 // Remove temporary hook
-                                                hooks.remove(streamingHook);
+                                                removeHook(streamingHook);
                                             })
                                     .subscribe(
                                             finalMsg -> {

agentscope-extensions/agentscope-kotlin/pom.xml

@@ -0,0 +1,89 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  ~ Copyright 2024-2025 the original author or authors.
+  ~
+  ~ Licensed under the Apache License, Version 2.0 (the "License");
+  ~ You may not use this file except in compliance with the License.
+  ~ You may obtain a copy of the License at
+  ~
+  ~     https://www.apache.org/licenses/LICENSE-2.0
+  ~
+  ~ Unless required by applicable law or agreed to in writing, software
+  ~ distributed under the License is distributed on an "AS IS" BASIS,
+  ~ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+  ~ See the License for the specific language governing permissions and
+  ~ limitations under the License.
+  -->
+
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
+                             http://maven.apache.org/xsd/maven-4.0.0.xsd">
+
+    <modelVersion>4.0.0</modelVersion>
+
+    <parent>
+        <groupId>io.agentscope</groupId>
+        <artifactId>agentscope-extensions</artifactId>
+        <version>${revision}</version>
+        <relativePath>../pom.xml</relativePath>
+    </parent>
+
+    <artifactId>agentscope-kotlin</artifactId>
+    <name>AgentScope Kotlin Coroutine Extensions</name>
+    <packaging>jar</packaging>
+
+    <properties>
+        <kotlin.version>1.9.24</kotlin.version>
+        <kotlin.coroutines.version>1.8.1</kotlin.coroutines.version>
+    </properties>
+
+    <dependencies>
+        <!-- AgentScope core -->
+        <dependency>
+            <groupId>io.agentscope</groupId>
+            <artifactId>agentscope-core</artifactId>
+        </dependency>
+
+        <!-- Kotlin coroutines -->
+        <dependency>
+            <groupId>org.jetbrains.kotlinx</groupId>
+            <artifactId>kotlinx-coroutines-core</artifactId>
+            <version>${kotlin.coroutines.version}</version>
+        </dependency>
+
+        <!-- Reactor <-> Coroutine bridge -->
+        <dependency>
+            <groupId>org.jetbrains.kotlinx</groupId>
+            <artifactId>kotlinx-coroutines-reactor</artifactId>
+            <version>${kotlin.coroutines.version}</version>
+        </dependency>
+    </dependencies>
+
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.jetbrains.kotlin</groupId>
+                <artifactId>kotlin-maven-plugin</artifactId>
+                <version>${kotlin.version}</version>
+                <executions>
+                    <execution>
+                        <id>compile</id>
+                        <phase>compile</phase>
+                        <goals>
+                            <goal>compile</goal>
+                        </goals>
+                    </execution>
+                    <execution>
+                        <id>test-compile</id>
+                        <phase>test-compile</phase>
+                        <goals>
+                            <goal>test-compile</goal>
+                        </goals>
+                    </execution>
+                </executions>
+            </plugin>
+        </plugins>
+    </build>
+
+</project>

agentscope-extensions/agentscope-kotlin/src/main/kotlin/io/agentscope/kotlin/AgentExtensions.kt

@@ -0,0 +1,94 @@
+/*
+ * Copyright 2024-2025 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+
+
+package io.agentscope.kotlin
+
+import io.agentscope.core.agent.Agent
+import io.agentscope.core.agent.Event
+import io.agentscope.core.agent.StreamOptions
+import io.agentscope.core.message.Msg
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.reactor.awaitFirstOrNull
+import kotlinx.coroutines.reactor.awaitSingle
+import kotlinx.coroutines.reactive.asFlow
+
+/* ---------- call(...) -> suspend ---------- */
+
+suspend fun Agent.callSuspend(msg: Msg): Msg =
+    this.call(msg).awaitSingle()
+
+suspend fun Agent.callSuspend(msgs: List<Msg>): Msg =
+    this.call(msgs).awaitSingle()
+
+suspend fun Agent.callSuspend(): Msg =
+    this.call().awaitSingle()
+
+suspend fun Agent.callSuspend(
+    msg: Msg,
+    structuredModel: Class<*>
+): Msg =
+    this.call(msg, structuredModel).awaitSingle()
+
+suspend fun Agent.callSuspend(
+    msgs: List<Msg>,
+    structuredModel: Class<*>
+): Msg =
+    this.call(msgs, structuredModel).awaitSingle()
+
+suspend fun Agent.callSuspend(
+    structuredModel: Class<*>
+): Msg =
+    this.call(structuredModel).awaitSingle()
+
+/* ---------- observe(...) -> suspend ---------- */
+
+suspend fun Agent.observeSuspend(msg: Msg) {
+    this.observe(msg).awaitFirstOrNull()
+}
+
+suspend fun Agent.observeSuspend(msgs: List<Msg>) {
+    this.observe(msgs).awaitFirstOrNull()
+}
+
+/* ---------- stream(...) -> Flow ---------- */
+
+fun Agent.streamFlow(
+    msg: Msg,
+    options: StreamOptions = StreamOptions.defaults()
+): Flow<Event> =
+    this.stream(msg, options).asFlow()
+
+fun Agent.streamFlow(
+    msgs: List<Msg>,
+    options: StreamOptions = StreamOptions.defaults()
+): Flow<Event> =
+    this.stream(msgs, options).asFlow()
+
+fun Agent.streamFlow(
+    msg: Msg,
+    options: StreamOptions,
+    structuredModel: Class<*>
+): Flow<Event> =
+    this.stream(msg, options, structuredModel).asFlow()
+
+fun Agent.streamFlow(
+    msgs: List<Msg>,
+    options: StreamOptions,
+    structuredModel: Class<*>
+): Flow<Event> =
+    this.stream(msgs, options, structuredModel).asFlow()