feat(#1751): Forum.session(input) — participants concurrent, captain last

Forum's session runs participants concurrently (events interleave by
agentId, like Parallel) then runs the captain sequentially (events
stream next). Terminal Completed has the captain's agentId.

Preserves ForumReturnException short-circuit: if a participant or
captain calls forum_return, the captain doesn't run (or is interrupted);
terminal Completed carries the captured value cast through
castForumReturnInternal. The mention listener still fires per-agent —
the streaming session is purely additive.

Forum gains:
- outType, castOut, captainTakesTranscript exposed as `internal` so
  the session extension can read them (they remain private to the
  outside world).
- castForumReturnInternal — visible wrapper around the private
  castForumReturn.
- fireMentionListener — visible wrapper around the private mention
  callback.
- ForumReturnException is now internal (was private at file scope)
  so the extension can catch it.

TDD red-first: 2 participants + captain via the `*` builder. Asserts
all three agents emit SkillStarted, captain's bracket follows both
participants' SkillCompleted, terminal Completed has captain's
agentId + the verdict.

Full regression sweep green.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: Skobeltsyn

src/main/kotlin/agents_engine/composition/forum/Forum.kt

@@ -30,12 +30,18 @@ data class ForumTranscript<IN>(
 
 class Forum<IN, OUT>(
     val agents: List<Agent<IN, *>>,
-    private val outType: KClass<*>,
-    private val castOut: (Any?) -> OUT,
-    private val captainTakesTranscript: Boolean = false,
+    internal val outType: KClass<*>,
+    internal val castOut: (Any?) -> OUT,
+    internal val captainTakesTranscript: Boolean = false,
 ) {
     private var mentionListener: ((agentName: String, output: Any?) -> Unit)? = null
 
+    internal fun fireMentionListener(name: String, output: Any?) {
+        mentionListener?.invoke(name, output)
+    }
+
+    internal fun castForumReturnInternal(raw: Any?): OUT = castForumReturn(raw)
+
     fun onMentionEmitted(block: (agentName: String, output: Any?) -> Unit) {
         mentionListener = block
     }
@@ -98,7 +104,7 @@ class Forum<IN, OUT>(
     }
 }
 
-private class ForumReturnException(val value: Any?) : RuntimeException()
+internal class ForumReturnException(val value: Any?) : RuntimeException()
 
 private fun buildForumReturnTool(): agents_engine.model.ToolDef =
     agents_engine.model.ToolDef(

src/main/kotlin/agents_engine/composition/forum/ForumSessionExtension.kt

@@ -0,0 +1,101 @@
+package agents_engine.composition.forum
+
+import agents_engine.core.Agent
+import agents_engine.model.AgentEventEmitter
+import agents_engine.runtime.events.AgentEvent
+import agents_engine.runtime.events.AgentSession
+import agents_engine.runtime.events.runAgentInSession
+import kotlinx.coroutines.CompletableDeferred
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.SupervisorJob
+import kotlinx.coroutines.async
+import kotlinx.coroutines.awaitAll
+import kotlinx.coroutines.channels.Channel
+import kotlinx.coroutines.coroutineScope
+import kotlinx.coroutines.flow.consumeAsFlow
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.withContext
+
+/**
+ * #1751 — start a streaming session against [this] forum.
+ *
+ * Participants run concurrently — their events stream into the shared
+ * emitter and interleave by arrival order (like Parallel). After every
+ * participant completes, the captain runs sequentially; the captain's
+ * events stream next. Terminal `Completed(agentId=captain.name, output)`.
+ *
+ * Preserves the `ForumReturnException` short-circuit: if a participant
+ * (or, less commonly, the captain) calls `forum_return`, the captain
+ * doesn't run; terminal `Completed` carries the captured value cast
+ * through `castForumReturnInternal`.
+ *
+ * Mention listener still fires per-agent (forum.onMentionEmitted) — the
+ * streaming session is purely additive to the existing observability.
+ */
+fun <IN, OUT> Forum<IN, OUT>.session(input: IN): AgentSession<OUT> {
+    val forum = this
+    val channel = Channel<AgentEvent<OUT>>(Channel.BUFFERED)
+    val result = CompletableDeferred<OUT>()
+    val scope = CoroutineScope(SupervisorJob() + Dispatchers.Unconfined)
+    val captain = forum.agents.last()
+    val participants = forum.agents.dropLast(1)
+
+    scope.launch {
+        @Suppress("UNCHECKED_CAST")
+        val emitter: AgentEventEmitter = { event -> channel.trySend(event as AgentEvent<OUT>) }
+        try {
+            val verdict: OUT = try {
+                val contributions = withContext(Dispatchers.Default) {
+                    coroutineScope {
+                        participants.map { participant ->
+                            async {
+                                @Suppress("UNCHECKED_CAST")
+                                val out = runAgentInSession(
+                                    participant as Agent<IN, Any?>,
+                                    input,
+                                    emitter,
+                                ).first
+                                forum.fireMentionListener(participant.name, out)
+                                ParticipantContribution(participant.name, out)
+                            }
+                        }.awaitAll()
+                    }
+                }
+                val captainVerdict: OUT = if (forum.captainTakesTranscript) {
+                    val transcript = ForumTranscript(originalInput = input, contributions = contributions)
+                    @Suppress("UNCHECKED_CAST")
+                    runAgentInSession(
+                        captain as Agent<ForumTranscript<IN>, OUT>,
+                        transcript,
+                        emitter,
+                    ).first
+                } else {
+                    @Suppress("UNCHECKED_CAST")
+                    runAgentInSession(
+                        captain as Agent<IN, OUT>,
+                        input,
+                        emitter,
+                    ).first
+                }
+                forum.fireMentionListener(captain.name, captainVerdict)
+                captainVerdict
+            } catch (e: ForumReturnException) {
+                forum.castForumReturnInternal(e.value)
+            }
+
+            channel.trySend(AgentEvent.Completed(captain.name, verdict, null))
+            channel.close()
+            result.complete(verdict)
+        } catch (t: Throwable) {
+            channel.trySend(AgentEvent.Failed(captain.name, t))
+            channel.close()
+            result.completeExceptionally(t)
+        }
+    }
+
+    return AgentSession(
+        events = channel.consumeAsFlow(),
+        resultDeferred = result,
+    )
+}

src/test/kotlin/agents_engine/runtime/events/ForumSessionTest.kt

@@ -0,0 +1,77 @@
+package agents_engine.runtime.events
+
+import agents_engine.composition.forum.session
+import agents_engine.composition.forum.times
+import agents_engine.core.agent
+import kotlinx.coroutines.flow.toList
+import kotlinx.coroutines.test.runTest
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertIs
+import kotlin.test.assertTrue
+
+// #1751 — Forum session streams events from all participants
+// concurrently, then from the captain. Terminal Completed has the
+// captain's agentId.
+
+class ForumSessionTest {
+
+    @Test
+    fun `forum session emits participants' events then captain's events with captain as terminal agentId`() = runTest {
+        val analyst = agent<String, String>("analyst") {
+            skills {
+                skill<String, String>("analyze", "Analyzes") {
+                    implementedBy { "analysis: $it" }
+                }
+            }
+        }
+        val critic = agent<String, String>("critic") {
+            skills {
+                skill<String, String>("critique", "Critiques") {
+                    implementedBy { "critique: $it" }
+                }
+            }
+        }
+        val captain = agent<String, String>("captain") {
+            skills {
+                skill<String, String>("verdict", "Final verdict") {
+                    implementedBy { "verdict: $it" }
+                }
+            }
+        }
+        val forum = analyst * critic * captain
+
+        val session = forum.session("topic")
+        val events = session.events.toList()
+        val output = session.await()
+
+        assertEquals("verdict: topic", output)
+
+        // Both participants' SkillStarted/SkillCompleted appear.
+        val starts = events.filterIsInstance<AgentEvent.SkillStarted>()
+        assertEquals(
+            setOf("analyst", "critic", "captain"),
+            starts.map { it.agentId }.toSet(),
+            "every forum agent must emit SkillStarted; got: $starts",
+        )
+
+        // Captain's bracket comes AFTER both participants completed.
+        val captainStartIdx = events.indexOfFirst { it is AgentEvent.SkillStarted && it.agentId == "captain" }
+        val analystCompletedIdx = events.indexOfFirst { it is AgentEvent.SkillCompleted && it.agentId == "analyst" }
+        val criticCompletedIdx = events.indexOfFirst { it is AgentEvent.SkillCompleted && it.agentId == "critic" }
+        assertTrue(
+            analystCompletedIdx < captainStartIdx,
+            "analyst.SkillCompleted must precede captain.SkillStarted; got events: $events",
+        )
+        assertTrue(
+            criticCompletedIdx < captainStartIdx,
+            "critic.SkillCompleted must precede captain.SkillStarted; got events: $events",
+        )
+
+        // Terminal Completed.
+        val terminal = events.last()
+        assertIs<AgentEvent.Completed<String>>(terminal)
+        assertEquals("captain", terminal.agentId, "Forum's terminal Completed uses the captain's name")
+        assertEquals("verdict: topic", terminal.output)
+    }
+}