feat(#1746): multi-stage Pipeline session — events flow through every stage

Flips two of the Pipeline-involving `then` overloads to populate
`sessionExec`, so 3+ stage chains now stream events from ALL stages
instead of only the final Agent:

- Pipeline<A,B>.then(Agent<B,C>) — chains the inner Pipeline's
  effectiveSessionExec into runAgentInSession for the trailing Agent
- Pipeline<A,B>.then(Pipeline<B,C>) — chains both pipelines'
  effectiveSessionExec

Out of scope: overloads involving Forum, Parallel, Branch, Loop —
those operators don't have session support yet. Their related `then`
overloads stay at the default non-streaming fallback until each
operator gains its own session impl.

TDD red-first: new test `three-stage pipeline emits events from all
three agents` in PipelineSessionTest. Before flip, the 3-stage chain
emitted only the terminal Completed (1 event); after, all 7 events
fire (3× SkillStarted+SkillCompleted + 1 terminal).

Full regression sweep green across root + KSP + no-reflect.

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

Author: Skobeltsyn

src/main/kotlin/agents_engine/composition/pipeline/Pipeline.kt

@@ -62,7 +62,20 @@ infix fun <A, B, C> Agent<A, B>.then(other: Agent<B, C>): Pipeline<A, C> {
 
 infix fun <A, B, C> Pipeline<A, B>.then(other: Agent<B, C>): Pipeline<A, C> {
     other.markPlaced("pipeline")
-    return Pipeline(agents + other) { input -> other.invokeSuspend(this.invokeSuspend(input)) }
+    val inner = this
+    return Pipeline(
+        agents = inner.agents + other,
+        // #1746: chain the inner Pipeline's streaming output into the new
+        // Agent's session run. Inner Pipeline's effectiveSessionExec
+        // forwards events from each of its stages; runAgentInSession then
+        // emits the trailing Agent's bracket events.
+        sessionExec = { input, emitter ->
+            val mid = inner.effectiveSessionExec(input, emitter)
+            val (out, _) = agents_engine.runtime.events.runAgentInSession(other, mid, emitter)
+            out
+        },
+        execution = { input -> other.invokeSuspend(inner.invokeSuspend(input)) },
+    )
 }
 
 infix fun <A, B, C> Agent<A, B>.then(other: Forum<B, C>): Pipeline<A, C> {
@@ -75,7 +88,16 @@ infix fun <A, B, C> Pipeline<A, B>.then(other: Forum<B, C>): Pipeline<A, C> {
 }
 
 infix fun <A, B, C> Pipeline<A, B>.then(other: Pipeline<B, C>): Pipeline<A, C> {
-    return Pipeline(agents + other.agents) { input -> other.invokeSuspend(this.invokeSuspend(input)) }
+    val left = this
+    return Pipeline(
+        agents = left.agents + other.agents,
+        // #1746: chain both pipelines' streaming exec paths.
+        sessionExec = { input, emitter ->
+            val mid = left.effectiveSessionExec(input, emitter)
+            other.effectiveSessionExec(mid, emitter)
+        },
+        execution = { input -> other.invokeSuspend(left.invokeSuspend(input)) },
+    )
 }
 
 infix fun <A, B, C> Agent<A, B>.then(other: Parallel<B, C>): Pipeline<A, List<C>> {

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

@@ -50,6 +50,39 @@ class PipelineSessionTest {
         assertEquals("n=5", e4.output)
     }
 
+    @Test
+    fun `three-stage pipeline emits events from all three agents — proves Pipeline then Agent overload flows events through`() = runTest {
+        // a then b then c is left-associative: (a then b) then c, which goes
+        // through the Pipeline<*, *>.then(Agent<*, *>) overload. Before #1746
+        // that overload's sessionExec fell back to the default (no events from
+        // a or b) — only c's events appeared. After: all three.
+        val a = agent<String, Int>("a") {
+            skills { skill<String, Int>("len", "Length") { implementedBy { it.length } } }
+        }
+        val b = agent<Int, Int>("b") {
+            skills { skill<Int, Int>("doubled", "Doubles") { implementedBy { it * 2 } } }
+        }
+        val c = agent<Int, String>("c") {
+            skills { skill<Int, String>("describe", "Describe") { implementedBy { "n=$it" } } }
+        }
+        val pipeline = a then b then c
+
+        val session = pipeline.session("hello")
+        val events = session.events.toList()
+        val output = session.await()
+
+        assertEquals("n=10", output, "pipeline output: 5 → 10 → \"n=10\"")
+        assertEquals(7, events.size, "expected 3× SkillStarted/SkillCompleted + 1 Completed; got: $events")
+
+        // Ordered: each agent's pair appears in chain order before the next agent runs.
+        val agentIds = events.filterIsInstance<AgentEvent.SkillStarted>().map { it.agentId }
+        assertEquals(listOf("a", "b", "c"), agentIds, "SkillStarted events must fire in chain order; got: $agentIds")
+
+        val completed = events.last()
+        assertIs<AgentEvent.Completed<String>>(completed)
+        assertEquals("c", completed.agentId, "terminal Completed uses the LAST agent's name")
+    }
+
     @Test
     fun `pipeline session terminates with Failed when the second agent throws — only first agent's events precede`() = runTest {
         val boom = IllegalStateException("middle blew up")