feat(#1745): Pipeline.session(input) — events flow through from every node

First composition operator gets the session surface. `a then b` produces
a Pipeline<IN, OUT>; pipeline.session(input) now exposes a Flow<AgentEvent>
that surfaces events from BOTH inner agents with their own agentIds,
plus a single terminal Completed/Failed.

Sequential composition is enforced by the typed boundary — `a` runs to
completion (streaming its Token events) before `b` starts (with a's full
typed OUT as input, streaming its own Token events). The user sees a
continuous event stream demultiplexed by agentId.

Implementation:
- Extract runAgentInSession(agent, input, emitter) -> (OUT, TokenUsage?)
  helper from Agent.session(input). Emits SkillStarted/SkillCompleted
  around the agent run; does NOT emit Completed. Both Agent.session and
  Pipeline.session use it.
- AgentEventEmitter typealias changed from `suspend (...) -> Unit` to
  non-suspend `(...) -> Unit` so it can be called from the non-suspend
  onSkillStarted callback. Implementations forward to channel.trySend
  which is already non-blocking.
- Pipeline gains an optional sessionExec: (IN, AgentEventEmitter) -> OUT
  constructor parameter declared BEFORE execution so trailing-lambda
  syntax keeps binding to execution. Effective sessionExec falls back
  to execution-without-events when null (existing unconverted `then`
  overloads).
- Only the Agent<A, B>.then(Agent<B, C>) overload populates sessionExec
  in this commit. Multi-stage chains (a then b then c) built via the
  Pipeline then Agent overload fall back to default; only the FINAL
  Agent emits events. Documented gap, separate follow-up flips each
  remaining overload.

TDD red-first: two tests in PipelineSessionTest. Happy path asserts the
ordered 5-event sequence and Completed.agentId = last agent's name.
Failure path asserts terminal Failed with original cause when the
second agent throws.

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

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

Author: Skobeltsyn

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

@@ -15,8 +15,29 @@ import kotlinx.coroutines.runBlocking
  */
 class Pipeline<IN, OUT>(
     val agents: List<Agent<*, *>>,
+    /**
+     * #1745 — session-aware execution path. When `pipeline.session(input)`
+     * is called, this runs instead of [execution] with a non-null emitter,
+     * surfacing inner-agent events with their own `agentId`s. Defaults to
+     * the non-streaming `execution` so any `then` overload that hasn't been
+     * converted yet still works (Pipeline session emits only the terminal
+     * Completed, no inner events — known gap, see #1745 follow-ups).
+     *
+     * Declared BEFORE [execution] so the trailing-lambda construction
+     * `Pipeline(agents) { ... }` still binds the lambda to [execution].
+     */
+    internal val sessionExec: (suspend (
+        input: IN,
+        emitter: agents_engine.model.AgentEventEmitter,
+    ) -> OUT)? = null,
     private val execution: suspend (IN) -> OUT,
 ) {
+    /** Effective sessionExec: explicit when supplied, otherwise falls back to execution (no events). */
+    internal val effectiveSessionExec: suspend (
+        input: IN,
+        emitter: agents_engine.model.AgentEventEmitter,
+    ) -> OUT = sessionExec ?: { input, _ -> execution(input) }
+
     operator fun invoke(input: IN): OUT = runBlocking { execution(input) }
 
     suspend fun invokeSuspend(input: IN): OUT = execution(input)
@@ -25,7 +46,18 @@ class Pipeline<IN, OUT>(
 infix fun <A, B, C> Agent<A, B>.then(other: Agent<B, C>): Pipeline<A, C> {
     this.markPlaced("pipeline")
     other.markPlaced("pipeline")
-    return Pipeline(listOf(this, other)) { input -> other.invokeSuspend(this.invokeSuspend(input)) }
+    val first = this
+    return Pipeline(
+        agents = listOf(first, other),
+        // #1745: streaming path runs both agents through runAgentInSession
+        // so events from both flow into the emitter with their own agentIds.
+        sessionExec = { input, emitter ->
+            val (mid, _) = agents_engine.runtime.events.runAgentInSession(first, input, emitter)
+            val (out, _) = agents_engine.runtime.events.runAgentInSession(other, mid, emitter)
+            out
+        },
+        execution = { input -> other.invokeSuspend(first.invokeSuspend(input)) },
+    )
 }
 
 infix fun <A, B, C> Pipeline<A, B>.then(other: Agent<B, C>): Pipeline<A, C> {

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

@@ -0,0 +1,63 @@
+package agents_engine.composition.pipeline
+
+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
+
+/**
+ * #1745 — start a streaming session against [this] pipeline.
+ *
+ * Sequential composition: each inner agent runs to completion before the
+ * next starts (the typed boundary forces a complete `MID` value to flow
+ * from `a` to `b`). But WITHIN each agent, events stream incrementally:
+ * the consumer sees `SkillStarted` / `Token` / `ToolCall*` / `SkillCompleted`
+ * for `a`, then the same for `b`, terminated by exactly one `Completed`
+ * with the pipeline's final `OUT`.
+ *
+ * `agentId` on every inner event names the source agent — composition
+ * preserves provenance. The terminal `Completed.agentId` uses the LAST
+ * agent's name (its `OUT` type matches the pipeline's `OUT`).
+ *
+ * **Step-4 scope:** only the `Agent then Agent` overload populates
+ * `Pipeline.sessionExec` today. Multi-stage chains (`a then b then c`)
+ * built via the `Pipeline then Agent` overload fall back to the default
+ * (non-streaming) `sessionExec` — `pipeline.session(input)` will emit only
+ * the terminal `Completed`. Separate follow-ups flip each composing
+ * overload. Documented in the corresponding session test.
+ */
+fun <IN, OUT> Pipeline<IN, OUT>.session(input: IN): AgentSession<OUT> {
+    val pipeline = this
+    val channel = Channel<AgentEvent<OUT>>(Channel.BUFFERED)
+    val result = CompletableDeferred<OUT>()
+    val scope = CoroutineScope(SupervisorJob() + Dispatchers.Unconfined)
+    // agentId for the terminal Completed: last agent's name (its OUT
+    // matches Pipeline's OUT). Pipeline has no name of its own.
+    val terminalAgentId = pipeline.agents.lastOrNull()?.name ?: "pipeline"
+
+    scope.launch {
+        @Suppress("UNCHECKED_CAST")
+        val emitter: AgentEventEmitter = { event -> channel.trySend(event as AgentEvent<OUT>) }
+        try {
+            val output = pipeline.effectiveSessionExec(input, emitter)
+            channel.trySend(AgentEvent.Completed(terminalAgentId, output, null))
+            channel.close()
+            result.complete(output)
+        } catch (t: Throwable) {
+            channel.trySend(AgentEvent.Failed(terminalAgentId, t))
+            channel.close()
+            result.completeExceptionally(t)
+        }
+    }
+
+    return AgentSession(
+        events = channel.consumeAsFlow(),
+        resultDeferred = result,
+    )
+}

src/main/kotlin/agents_engine/model/StreamingAggregator.kt

@@ -10,7 +10,12 @@ import kotlinx.coroutines.withContext
 // subtypes (Token, ToolCall*, SkillStarted, SkillCompleted, Failed);
 // only `AgentEvent.Completed<OUT>` carries the typed payload and that's
 // emitted in `Agent.session(input)` after the loop returns.
-internal typealias AgentEventEmitter = suspend (AgentEvent<*>) -> Unit
+//
+// Non-suspend (#1745) so it can be called from non-suspend callbacks
+// like `Agent.invokeSuspendForSession`'s `onSkillStarted` lambda.
+// Implementations typically forward to `channel.trySend(event)`, which
+// is itself non-blocking — appropriate for BUFFERED-channel-backed Flows.
+internal typealias AgentEventEmitter = (AgentEvent<*>) -> Unit
 
 /**
  * #1739 — round-trip the model: either via the existing non-streaming

src/main/kotlin/agents_engine/runtime/events/AgentSessionExtension.kt

@@ -1,6 +1,8 @@
 package agents_engine.runtime.events
 
 import agents_engine.core.Agent
+import agents_engine.model.AgentEventEmitter
+import agents_engine.model.TokenUsage
 import kotlinx.coroutines.CompletableDeferred
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
@@ -39,33 +41,17 @@ fun <IN, OUT> Agent<IN, OUT>.session(input: IN): AgentSession<OUT> {
     // of any unrelated coroutine the consumer happens to be running in.
     val scope = CoroutineScope(SupervisorJob() + Dispatchers.Unconfined)
     scope.launch {
-        // Captured-on-the-stack: each session has its own holder, so
-        // concurrent sessions can't race on a shared field.
-        var capturedSkillName: String? = null
-        // #1740: per-session usage capture from the agentic loop's cumulative
-        // total. Stays null for implementedBy skills (no LLM round-trip).
-        var capturedUsage: agents_engine.model.TokenUsage? = null
         // #1739: emitter forwards AgentEvents from inside the agentic loop
         // (Token, ToolCallStarted, ToolCallArgumentsDelta, ToolCallFinished)
         // into the same channel as the bracket events. trySend is non-
         // suspending — appropriate for a BUFFERED channel; if the buffer
         // ever fills (it has high capacity), excess events would be
-        // dropped silently. Step 4 will tighten this for high-throughput
-        // streaming.
-        val streamingEmitter: agents_engine.model.AgentEventEmitter = { event ->
-            channel.trySend(event as AgentEvent<OUT>)
-        }
+        // dropped silently.
+        @Suppress("UNCHECKED_CAST")
+        val emitter: AgentEventEmitter = { event -> channel.trySend(event as AgentEvent<OUT>) }
         try {
-            val output = agent.invokeSuspendForSession(
-                input,
-                emitter = streamingEmitter,
-                onSkillCompleted = { usage -> capturedUsage = usage },
-            ) { skillName ->
-                capturedSkillName = skillName
-                channel.trySend(AgentEvent.SkillStarted(agent.name, skillName))
-            }
-            channel.trySend(AgentEvent.SkillCompleted(agent.name, capturedSkillName ?: "?", capturedUsage))
-            channel.trySend(AgentEvent.Completed(agent.name, output, capturedUsage))
+            val (output, usage) = runAgentInSession(agent, input, emitter)
+            channel.trySend(AgentEvent.Completed(agent.name, output, usage))
             channel.close()
             result.complete(output)
         } catch (t: Throwable) {
@@ -80,3 +66,36 @@ fun <IN, OUT> Agent<IN, OUT>.session(input: IN): AgentSession<OUT> {
         resultDeferred = result,
     )
 }
+
+/**
+ * #1745 — shared "run an agent and surface its bracket + inner events
+ * onto [emitter]" helper. Used by both `Agent.session(input)` and
+ * `Pipeline.session(input)`. Emits `SkillStarted` and `SkillCompleted`
+ * around the agent's execution; does NOT emit `Completed` (the composing
+ * caller emits exactly one terminal `Completed` after all stages run).
+ *
+ * Returns the typed output paired with the cumulative `TokenUsage` for
+ * this agent's skill (null for `implementedBy`).
+ */
+internal suspend fun <IN, OUT> runAgentInSession(
+    agent: Agent<IN, OUT>,
+    input: IN,
+    emitter: AgentEventEmitter,
+): Pair<OUT, TokenUsage?> {
+    var capturedSkillName: String? = null
+    var capturedUsage: TokenUsage? = null
+    val output = agent.invokeSuspendForSession(
+        input,
+        emitter = emitter,
+        onSkillCompleted = { usage -> capturedUsage = usage },
+    ) { skillName ->
+        // SkillStarted fires BEFORE the skill body runs — emitting from
+        // this onSkillStarted callback (non-suspend; emitter is also
+        // non-suspend per #1745) means the event reaches the consumer
+        // before any Token / ToolCall* events from this skill's loop.
+        capturedSkillName = skillName
+        emitter(AgentEvent.SkillStarted(agent.name, skillName))
+    }
+    emitter(AgentEvent.SkillCompleted(agent.name, capturedSkillName ?: "?", capturedUsage))
+    return output to capturedUsage
+}

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

@@ -0,0 +1,94 @@
+package agents_engine.runtime.events
+
+import agents_engine.composition.pipeline.session
+import agents_engine.composition.pipeline.then
+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.assertFailsWith
+import kotlin.test.assertIs
+
+// #1745 — Pipeline.session(input) flows events from every inner agent
+// with its own agentId. Sequential between nodes (typed boundary forces
+// a complete MID value to flow from a to b); streaming WITHIN a node.
+
+class PipelineSessionTest {
+
+    @Test
+    fun `pipeline session emits ordered events from both agents plus a terminal Completed`() = runTest {
+        val parse = agent<String, Int>("parse") {
+            skills {
+                skill<String, Int>("length", "Computes input length") {
+                    implementedBy { it.length }
+                }
+            }
+        }
+        val describe = agent<Int, String>("describe") {
+            skills {
+                skill<Int, String>("format", "Formats integer") {
+                    implementedBy { "n=$it" }
+                }
+            }
+        }
+        val pipeline = parse then describe
+
+        val session = pipeline.session("hello")
+        val events = session.events.toList()
+        val output = session.await()
+
+        assertEquals("n=5", output, "pipeline output threads parse→describe correctly")
+        assertEquals(5, events.size, "expected 5 events: 2× SkillStarted/SkillCompleted + 1 Completed; got: $events")
+
+        val e0 = events[0]; assertIs<AgentEvent.SkillStarted>(e0); assertEquals("parse", e0.agentId); assertEquals("length", e0.skillName)
+        val e1 = events[1]; assertIs<AgentEvent.SkillCompleted>(e1); assertEquals("parse", e1.agentId); assertEquals("length", e1.skillName)
+        val e2 = events[2]; assertIs<AgentEvent.SkillStarted>(e2); assertEquals("describe", e2.agentId); assertEquals("format", e2.skillName)
+        val e3 = events[3]; assertIs<AgentEvent.SkillCompleted>(e3); assertEquals("describe", e3.agentId); assertEquals("format", e3.skillName)
+        val e4 = events[4]; assertIs<AgentEvent.Completed<String>>(e4)
+        assertEquals("describe", e4.agentId, "Completed.agentId uses the last agent's name (its OUT matches Pipeline's OUT)")
+        assertEquals("n=5", e4.output)
+    }
+
+    @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")
+        val first = agent<String, Int>("first") {
+            skills {
+                skill<String, Int>("ok", "Returns length") {
+                    implementedBy { it.length }
+                }
+            }
+        }
+        val second = agent<Int, String>("second") {
+            skills {
+                skill<Int, String>("explode", "Throws unconditionally") {
+                    implementedBy { throw boom }
+                }
+            }
+        }
+        val pipeline = first then second
+
+        val session = pipeline.session("hello")
+        val events = session.events.toList()
+
+        // First agent should run cleanly — its events appear.
+        val firstStarted = events.filterIsInstance<AgentEvent.SkillStarted>().firstOrNull { it.agentId == "first" }
+            ?: error("expected SkillStarted(first) before the failure; got: $events")
+        val firstCompleted = events.filterIsInstance<AgentEvent.SkillCompleted>().firstOrNull { it.agentId == "first" }
+            ?: error("expected SkillCompleted(first) before the failure; got: $events")
+        assertEquals("first", firstStarted.agentId)
+        assertEquals("first", firstCompleted.agentId)
+
+        // Second agent's SkillStarted may or may not have fired before
+        // the implementedBy threw — but the terminal MUST be Failed,
+        // not Completed, and must carry the original exception.
+        val terminal = events.last()
+        assertIs<AgentEvent.Failed>(terminal, "last event must be Failed; got: $terminal")
+        // assertSame on cause: AgentEvent.Failed.cause carries the identity instance per #1737's contract.
+        assertEquals(boom.message, terminal.cause.message, "terminal cause must carry the original exception")
+
+        // await() rethrows.
+        assertFailsWith<IllegalStateException> { session.await() }
+    }
+}