feat(#1749): Loop.session(input) — bracket events per iteration

Loop runs the wrapped agent or pipeline repeatedly. The session
emits the inner SkillStarted/SkillCompleted pair per iteration —
all sharing the wrapped agent's agentId — terminated by a single
Completed with the final OUT.

Loop gains:
- sessionExec — populated by Agent.loop with runAgentInSession and
  by Pipeline.loop with effectiveSessionExec
- loopAgentId — wrapped agent's name (or pipeline's last agent's
  name). Used as agentId on the terminal Completed.
- execution exposed as `internal` so the session extension can call
  it directly when sessionExec is null (no-factory fallback)

Session iterates with the same control flow as invokeSuspend but
routes each iteration through sessionExec to stream inner events.

TDD red-first: 1 → 2 → 4 → 8 doubling-and-stop-at-8 loop produces
exactly 3× SkillStarted + 3× SkillCompleted (all agentId=doubler),
plus a terminal Completed(doubler, 8). Bracket pairs are contiguous
in arrival order.

Full regression sweep green.

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

Author: Skobeltsyn

src/main/kotlin/agents_engine/composition/loop/Loop.kt

@@ -13,9 +13,21 @@ private const val DEFAULT_MAX_ITERATIONS = 1_000
  * widen without breaking callers.
  */
 class Loop<IN, OUT>(
-    private val execution: suspend (IN) -> OUT,
-    private val next: (OUT) -> IN?,
-    private val maxIterations: Int = DEFAULT_MAX_ITERATIONS,
+    internal val execution: suspend (IN) -> OUT,
+    internal val next: (OUT) -> IN?,
+    internal val maxIterations: Int = DEFAULT_MAX_ITERATIONS,
+    /**
+     * #1749 — session-aware execution path. When non-null and called via
+     * `loop.session(input)`, each iteration's wrapped agent (or pipeline)
+     * streams events through the emitter with its own `agentId`.
+     */
+    internal val sessionExec: (suspend (input: IN, emitter: agents_engine.model.AgentEventEmitter) -> OUT)? = null,
+    /**
+     * #1749 — `agentId` for the terminal `AgentEvent.Completed` emitted
+     * by `loop.session(input)`. Null when constructed outside the factory
+     * functions; the session falls back to `"loop"` in that case.
+     */
+    internal val loopAgentId: String? = null,
 ) {
     init {
         require(maxIterations > 0) { "Loop maxIterations must be greater than 0." }
@@ -42,11 +54,30 @@ fun <A, B> Agent<A, B>.loop(
     next: (B) -> A?,
 ): Loop<A, B> {
     this.markPlaced("loop")
-    return Loop(execution = { input -> this.invokeSuspend(input) }, next = next, maxIterations = maxIterations)
+    val agent = this
+    return Loop(
+        execution = { input -> agent.invokeSuspend(input) },
+        next = next,
+        maxIterations = maxIterations,
+        // #1749: stream the wrapped agent's events per iteration.
+        sessionExec = { input, emitter ->
+            agents_engine.runtime.events.runAgentInSession(agent, input, emitter).first
+        },
+        loopAgentId = agent.name,
+    )
 }
 
 fun <A, B> Pipeline<A, B>.loop(
     maxIterations: Int = DEFAULT_MAX_ITERATIONS,
     next: (B) -> A?,
-): Loop<A, B> =
-    Loop(execution = { input -> this.invokeSuspend(input) }, next = next, maxIterations = maxIterations)
+): Loop<A, B> {
+    val inner = this
+    return Loop(
+        execution = { input -> inner.invokeSuspend(input) },
+        next = next,
+        maxIterations = maxIterations,
+        // #1749: pipeline's effectiveSessionExec streams every stage's events per iteration.
+        sessionExec = { input, emitter -> inner.effectiveSessionExec(input, emitter) },
+        loopAgentId = inner.agents.lastOrNull()?.name,
+    )
+}

src/main/kotlin/agents_engine/composition/loop/LoopSessionExtension.kt

@@ -0,0 +1,70 @@
+package agents_engine.composition.loop
+
+import agents_engine.model.AgentEventEmitter
+import agents_engine.runtime.events.AgentEvent
+import agents_engine.runtime.events.AgentSession
+import kotlinx.coroutines.CompletableDeferred
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.SupervisorJob
+import kotlinx.coroutines.channels.Channel
+import kotlinx.coroutines.flow.consumeAsFlow
+import kotlinx.coroutines.launch
+
+/**
+ * #1749 — start a streaming session against [this] loop.
+ *
+ * Each iteration runs the wrapped agent (or pipeline) under the same
+ * emitter, so the consumer sees bracket events repeated per iteration
+ * with the same `agentId`. The loop terminates when `next(out)` returns
+ * null OR when `maxIterations` is reached (the latter throws, surfacing
+ * as `AgentEvent.Failed`).
+ *
+ * Terminal `Completed` uses `loopAgentId` — the wrapped agent's name
+ * (or the pipeline's last agent's name). Falls back to `"loop"` when
+ * the Loop was constructed outside the factory functions.
+ */
+fun <IN, OUT> Loop<IN, OUT>.session(input: IN): AgentSession<OUT> {
+    val loop = this
+    val channel = Channel<AgentEvent<OUT>>(Channel.BUFFERED)
+    val result = CompletableDeferred<OUT>()
+    val scope = CoroutineScope(SupervisorJob() + Dispatchers.Unconfined)
+    val terminalAgentId = loop.loopAgentId ?: "loop"
+
+    scope.launch {
+        @Suppress("UNCHECKED_CAST")
+        val emitter: AgentEventEmitter = { event -> channel.trySend(event as AgentEvent<OUT>) }
+        try {
+            // sessionExec streams the wrapped run's inner events per
+            // iteration; falls back to plain execution (no events) when
+            // the Loop was constructed without the factory functions.
+            val streamingExec: suspend (IN) -> OUT = loop.sessionExec?.let { f -> { input -> f(input, emitter) } }
+                ?: loop.execution
+
+            var current = streamingExec(input)
+            var iterations = 1
+            while (true) {
+                val feedback = loop.next(current)
+                if (feedback == null) break
+                check(iterations < loop.maxIterations) {
+                    "Loop exceeded maxIterations=${loop.maxIterations} without termination."
+                }
+                current = streamingExec(feedback)
+                iterations++
+            }
+
+            channel.trySend(AgentEvent.Completed(terminalAgentId, current, null))
+            channel.close()
+            result.complete(current)
+        } catch (t: Throwable) {
+            channel.trySend(AgentEvent.Failed(terminalAgentId, t))
+            channel.close()
+            result.completeExceptionally(t)
+        }
+    }
+
+    return AgentSession(
+        events = channel.consumeAsFlow(),
+        resultDeferred = result,
+    )
+}

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

@@ -0,0 +1,60 @@
+package agents_engine.runtime.events
+
+import agents_engine.composition.loop.loop
+import agents_engine.composition.loop.session
+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
+
+// #1749 — Loop.session runs the wrapped agent (or pipeline) repeatedly,
+// streaming bracket events per iteration with the same agentId.
+// Terminal Completed carries the loop's final output.
+
+class LoopSessionTest {
+
+    @Test
+    fun `loop session emits bracket events per iteration plus a terminal Completed`() = runTest {
+        val doubler = agent<Int, Int>("doubler") {
+            skills {
+                skill<Int, Int>("double", "Doubles the input") {
+                    implementedBy { it * 2 }
+                }
+            }
+        }
+        // 1 → 2 → 4 → 8 (terminate, since 8 >= 8).
+        val loop = doubler.loop(maxIterations = 5) { if (it >= 8) null else it }
+
+        val session = loop.session(1)
+        val events = session.events.toList()
+        val output = session.await()
+
+        assertEquals(8, output, "loop output is 1 * 2 * 2 * 2 = 8 after three iterations")
+
+        val starts = events.filterIsInstance<AgentEvent.SkillStarted>()
+        val completedSkills = events.filterIsInstance<AgentEvent.SkillCompleted>()
+        assertEquals(3, starts.size, "expected exactly 3 SkillStarted events (one per iteration); got: $starts")
+        assertEquals(3, completedSkills.size, "expected exactly 3 SkillCompleted events; got: $completedSkills")
+
+        // All bracket events share the wrapped agent's agentId.
+        starts.forEach { assertEquals("doubler", it.agentId, "every iteration must carry the wrapped agent's name") }
+        completedSkills.forEach { assertEquals("doubler", it.agentId) }
+
+        // Order: each Started/Completed pair is contiguous, and iterations are sequential.
+        // events[0] = Started, events[1] = Completed, events[2] = Started, events[3] = Completed, ...
+        for (i in 0 until 3) {
+            val started = events[i * 2]
+            val completed = events[i * 2 + 1]
+            assertIs<AgentEvent.SkillStarted>(started, "event ${i * 2} should be SkillStarted")
+            assertIs<AgentEvent.SkillCompleted>(completed, "event ${i * 2 + 1} should be SkillCompleted")
+        }
+
+        // Terminal Completed.
+        val terminal = events.last()
+        assertIs<AgentEvent.Completed<Int>>(terminal)
+        assertEquals("doubler", terminal.agentId, "Loop's terminal Completed uses the wrapped agent's name")
+        assertEquals(8, terminal.output)
+    }
+}