feat(#1736): AgentEvent sealed hierarchy + AgentSession + Agent.session(input)

Second v0.5.0 streaming step. Consumer-facing layer on top of #1722's
LlmChunk foundation. TDD red-first.

Adds:
- AgentEvent sealed interface (Token / ToolCallStarted /
  ToolCallArgumentsDelta / ToolCallFinished / SkillStarted /
  SkillCompleted / Completed<OUT> / Failed). agentId on every
  subtype for composition provenance.
- AgentSession<OUT> with cold `events: Flow<AgentEvent<OUT>>`
  and `suspend fun await()`.
- Free function `Agent<IN, OUT>.session(input)` driving a
  Channel-backed Flow + CompletableDeferred from a dedicated
  SupervisorJob scope. Closure-captured skill-name holder is
  per-session — concurrent sessions can't race.

Backward compat: existing Agent.invokeSuspend(input) now delegates
to a new internal invokeSuspendForSession(input, onSkillStarted)
with a no-op callback. Same logic, byte-for-byte behavior; only
the new entry point on Agent surfaces the skill name into the
event flow.

Step 2 scope (deliberate): emits ONLY SkillStarted → SkillCompleted →
Completed (or Failed). Token / ToolCall* subtypes are defined in
the hierarchy but NOT YET emitted — that's step 3, where the
agentic loop is rewired onto a FlowCollector<AgentEvent>.
tokensUsed stays null for the same reason.

Full root + KSP + no-reflect smoke suites green.

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

Author: Skobeltsyn

src/main/kotlin/agents_engine/core/Agent.kt

@@ -252,10 +252,20 @@ class Agent<IN, OUT>(
      * which lets parent-scope cancellation and `withTimeout` propagate cleanly into
      * the agentic loop. The blocking [invoke] is a thin shim over this.
      */
-    suspend fun invokeSuspend(input: IN): OUT {
+    suspend fun invokeSuspend(input: IN): OUT = invokeSuspendForSession(input) { /* no-op */ }
+
+    /**
+     * #1736 — session-aware sibling of [invokeSuspend]. Same logic, plus an
+     * extra [onSkillStarted] callback fired after skill resolution and before
+     * execution. Existing `invokeSuspend` delegates with a no-op callback, so
+     * backward-compat is byte-for-byte; this entry point is only called by
+     * `Agent.session(input)` to surface the skill name into the event flow.
+     */
+    internal suspend fun invokeSuspendForSession(input: IN, onSkillStarted: (String) -> Unit): OUT {
         try {
             val skill = resolveSkill(input)
             skillChosenListener?.invoke(skill.name)
+            onSkillStarted(skill.name)
             return if (skill.isAgentic) {
                 castOut(executeAgentic(this, skill, input))
             } else {

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

@@ -0,0 +1,121 @@
+package agents_engine.runtime.events
+
+import agents_engine.model.TokenUsage
+
+/**
+ * #1736 — typed event emitted from `Agent.session(input).events` while the
+ * agentic loop runs. See [the v0.5.0 streaming premortem](../../../../docs/premortem-0.5.0-streaming.md)
+ * for the full design rationale.
+ *
+ * The sealed hierarchy is complete here so consumers can write exhaustive
+ * `when` matches today. **Not every subtype is emitted yet.** v0.5.0 step 2
+ * surfaces only [SkillStarted], [SkillCompleted], [Completed], and [Failed]
+ * — enough for the consumer surface to be useful for `implementedBy`-style
+ * agents. [Token], [ToolCallStarted], [ToolCallArgumentsDelta], and
+ * [ToolCallFinished] land in step 3 when the agentic loop is rewired onto a
+ * `FlowCollector<AgentEvent>`.
+ *
+ * Every event carries [agentId] — the name of the agent that produced it.
+ * Composition operators (`then`, `Pipeline`, `Branch`, `wrap`, `Swarm`)
+ * preserve provenance via this field so a consumer collecting from a
+ * composed pipeline can still tell which agent emitted which event.
+ *
+ * Only [Completed] carries the typed `OUT` payload; every other subtype
+ * is `AgentEvent<Nothing>` so events flow through any `AgentSession<OUT>`
+ * regardless of OUT.
+ */
+sealed interface AgentEvent<out OUT> {
+    /** The agent that produced this event. For composed pipelines this is the inner agent's name, not the composition's. */
+    val agentId: String
+
+    /**
+     * A chunk of LLM-streamed text from a single skill turn. Providers chunk at
+     * their own granularity — [text] may be a single token or a multi-token
+     * chunk; the framework passes through as-is.
+     *
+     * Not yet emitted (step 3 — agentic loop rewire).
+     */
+    data class Token(
+        override val agentId: String,
+        val skillName: String,
+        val text: String,
+    ) : AgentEvent<Nothing>
+
+    /**
+     * A new tool call has begun streaming. [callId] is unique per call within a
+     * session; [ToolCallArgumentsDelta] and [ToolCallFinished] for the same call
+     * share this id.
+     *
+     * Not yet emitted (step 3).
+     */
+    data class ToolCallStarted(
+        override val agentId: String,
+        val skillName: String,
+        val callId: String,
+        val toolName: String,
+    ) : AgentEvent<Nothing>
+
+    /**
+     * Partial tool-call arguments JSON. Multiple deltas may arrive per call for
+     * providers that stream argument JSON (Anthropic, OpenAI). Non-streaming
+     * providers emit one delta with the full JSON.
+     *
+     * Not yet emitted (step 3).
+     */
+    data class ToolCallArgumentsDelta(
+        override val agentId: String,
+        val callId: String,
+        val deltaJson: String,
+    ) : AgentEvent<Nothing>
+
+    /**
+     * Tool call complete — arguments parsed, executor invoked, result captured.
+     * [isError] flags executor exceptions that `onError` swallowed (loop keeps
+     * going); when [isError] is true and `onError` rethrew, the session emits
+     * [Failed] instead.
+     *
+     * Not yet emitted (step 3).
+     */
+    data class ToolCallFinished(
+        override val agentId: String,
+        val callId: String,
+        val toolName: String,
+        val arguments: Map<String, Any?>,
+        val result: Any?,
+        val isError: Boolean,
+    ) : AgentEvent<Nothing>
+
+    /** Agent has resolved a skill and is about to execute it (typed-tool dispatch OR an `implementedBy` lambda). */
+    data class SkillStarted(
+        override val agentId: String,
+        val skillName: String,
+    ) : AgentEvent<Nothing>
+
+    /**
+     * Skill execution complete. [tokensUsed] reports the cumulative LLM token
+     * usage for this skill turn, or null for `implementedBy` skills (no LLM)
+     * and for v0.5.0 step 2 (token threading lands in step 3).
+     */
+    data class SkillCompleted(
+        override val agentId: String,
+        val skillName: String,
+        val tokensUsed: TokenUsage?,
+    ) : AgentEvent<Nothing>
+
+    /** Terminal success — carries the typed output of the agent invocation. Emitted exactly once on the happy path. */
+    data class Completed<out OUT>(
+        override val agentId: String,
+        val output: OUT,
+        val tokensUsed: TokenUsage?,
+    ) : AgentEvent<OUT>
+
+    /**
+     * Terminal failure — emitted exactly once on the error path, BEFORE the
+     * exception propagates. Consumers collecting `events.toList()` see this
+     * as the last element; `session.await()` rethrows [cause].
+     */
+    data class Failed(
+        override val agentId: String,
+        val cause: Throwable,
+    ) : AgentEvent<Nothing>
+}

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

@@ -0,0 +1,31 @@
+package agents_engine.runtime.events
+
+import kotlinx.coroutines.Deferred
+import kotlinx.coroutines.flow.Flow
+
+/**
+ * #1736 — handle to an in-flight agent invocation. Returned by
+ * `agent.session(input)`. Carries a cold [events] flow and a terminal
+ * [await] entry point.
+ *
+ * Cold flow: each call to `agent.session(...)` starts a fresh invocation,
+ * regardless of whether you've collected from a previous session's [events].
+ * To share one invocation's events across multiple collectors, use
+ * `events.shareIn(scope, ...)`.
+ *
+ * Cancellation: cancelling the coroutine collecting [events] cancels the
+ * agent invocation. Cancelling the coroutine calling [await] does the same.
+ * Either path propagates into the upstream LLM HTTP call (step 3 hardens
+ * this once native streaming adapters land).
+ */
+class AgentSession<OUT> internal constructor(
+    val events: Flow<AgentEvent<OUT>>,
+    private val resultDeferred: Deferred<OUT>,
+) {
+    /**
+     * Awaits the agent's typed output. Throws the original exception (NOT
+     * wrapped) if the invocation failed — the [AgentEvent.Failed] event
+     * still appears in [events] as the terminal element.
+     */
+    suspend fun await(): OUT = resultDeferred.await()
+}

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

@@ -0,0 +1,67 @@
+package agents_engine.runtime.events
+
+import agents_engine.core.Agent
+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
+
+/**
+ * #1736 — start a streaming session against [this] agent.
+ *
+ * Returns an [AgentSession] whose [AgentSession.events] is a cold flow of
+ * typed [AgentEvent]s and whose [AgentSession.await] surfaces the typed
+ * output (or rethrows the failure).
+ *
+ * **Step 2 scope (intentional):** for `implementedBy` skills the emitted
+ * sequence is `SkillStarted → SkillCompleted → Completed`. For agentic
+ * skills the same three events bracket the agentic loop, but [AgentEvent.Token]
+ * and `ToolCall*` events are NOT yet surfaced — that's step 3, where the
+ * agentic loop itself is rewired onto a `FlowCollector<AgentEvent>`.
+ *
+ * Cold flow: each call starts a fresh invocation. To replay events to
+ * multiple collectors, wrap with `events.shareIn(...)`.
+ */
+fun <IN, OUT> Agent<IN, OUT>.session(input: IN): AgentSession<OUT> {
+    val agent = this
+    // BUFFERED keeps event production decoupled from consumer pace; an
+    // implementedBy skill can complete and queue all four events before
+    // the collector starts pulling. Step 3 may tune this for the
+    // token-streaming case.
+    val channel = Channel<AgentEvent<OUT>>(Channel.BUFFERED)
+    val result = CompletableDeferred<OUT>()
+
+    // Dedicated scope per session so a cancelled collector doesn't leak
+    // the result coroutine. SupervisorJob keeps the session independent
+    // 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. Step 3's
+        // agentic-loop rewire moves skill-name tracking into the
+        // FlowCollector chain proper.
+        var capturedSkillName: String? = null
+        try {
+            val output = agent.invokeSuspendForSession(input) { skillName ->
+                capturedSkillName = skillName
+                channel.trySend(AgentEvent.SkillStarted(agent.name, skillName))
+            }
+            channel.trySend(AgentEvent.SkillCompleted(agent.name, capturedSkillName ?: "?", null))
+            channel.trySend(AgentEvent.Completed(agent.name, output, null))
+            channel.close()
+            result.complete(output)
+        } catch (t: Throwable) {
+            channel.trySend(AgentEvent.Failed(agent.name, t))
+            channel.close()
+            result.completeExceptionally(t)
+        }
+    }
+
+    return AgentSession(
+        events = channel.consumeAsFlow(),
+        resultDeferred = result,
+    )
+}

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

@@ -0,0 +1,60 @@
+package agents_engine.runtime.events
+
+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.assertNull
+
+// #1736 — TDD red-first for the v0.5.0 streaming consumer surface.
+//
+// The narrowest meaningful test: an `implementedBy` skill on a typed
+// agent, invoked via `agent.session(input)`, must produce ordered
+// `SkillStarted → SkillCompleted → Completed` events and the deferred
+// terminal must return the typed output. No LLM is involved — Token
+// and ToolCall* events are defined in the hierarchy but NOT emitted
+// in this step (step 3 will rewire the agentic loop to surface them).
+//
+// If the agentic-loop rewire later changes event semantics for
+// `implementedBy` skills, this test fires. The contract is: zero
+// Token / ToolCall events for non-agentic skills, by construction.
+
+class AgentSessionBasicEventsTest {
+
+    @Test
+    fun `session emits SkillStarted then SkillCompleted then Completed for an implementedBy skill`() = runTest {
+        val echoAgent = agent<String, String>("echo") {
+            skills {
+                skill<String, String>("uppercase", "Uppercases the input") {
+                    implementedBy { it.uppercase() }
+                }
+            }
+        }
+
+        val session = echoAgent.session("hello")
+        val events = session.events.toList()
+        val output = session.await()
+
+        assertEquals("HELLO", output, "session.await() must return the typed agent output")
+        assertEquals(3, events.size, "expected 3 events for the implementedBy happy path; got: $events")
+
+        val started = events[0]
+        assertIs<AgentEvent.SkillStarted>(started, "first event must be SkillStarted; got: $started")
+        assertEquals("echo", started.agentId)
+        assertEquals("uppercase", started.skillName)
+
+        val completed = events[1]
+        assertIs<AgentEvent.SkillCompleted>(completed, "second event must be SkillCompleted; got: $completed")
+        assertEquals("echo", completed.agentId)
+        assertEquals("uppercase", completed.skillName)
+        assertNull(completed.tokensUsed, "SkillCompleted.tokensUsed stays null until step 3 threads it through executeAgentic")
+
+        val terminal = events[2]
+        assertIs<AgentEvent.Completed<String>>(terminal, "third event must be Completed<String>; got: $terminal")
+        assertEquals("echo", terminal.agentId)
+        assertEquals("HELLO", terminal.output)
+        assertNull(terminal.tokensUsed)
+    }
+}