feat(#1747): wrap session — events flow from teacher and student

teacher wrap student (returning Pipeline<IN, OUT>) now populates
sessionExec so both agents stream events during pipeline.session(input).

Consolidation: invokeSuspendForSession gains an optional promptOverride
parameter. invokeSuspendWithPromptOverride becomes a thin shim that
delegates with emitter=null — same byte-for-byte non-streaming
behavior wrap had pre-streaming. runAgentInSession also takes
promptOverride and passes it through.

Wrap's sessionExec:
- runAgentInSession(teacher, input, emitter) → teacher's events stream,
  output is the prompt override
- runAgentInSession(student, input, emitter, promptOverride = override)
  → student's events stream under the override

Student receives the SAME input as teacher (wrap semantics — teacher's
String OUT becomes the system prompt for student, not the input).

TDD red-first: WrapSessionTest builds an implementedBy teacher and an
agentic student with a stub ModelClient. Asserts ordered events from
both agents PLUS captures the student's system prompt to verify the
override was actually active (the teacher's emitted text appears,
the student's baked-in prompt does not).

Full regression sweep green.

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

Author: Skobeltsyn

src/main/kotlin/agents_engine/composition/wrap/Wrap.kt

@@ -45,8 +45,23 @@ infix fun <IN, OUT> Agent<IN, String>.wrap(student: Agent<IN, OUT>): Pipeline<IN
     this.markPlaced("pipeline")
     student.markPlaced("pipeline")
     val teacher = this
-    return Pipeline(listOf(teacher, student)) { input ->
-        val promptOverride = teacher.invokeSuspend(input)
-        student.invokeSuspendWithPromptOverride(input, promptOverride)
-    }
+    return Pipeline(
+        agents = listOf(teacher, student),
+        // #1747: streaming path — teacher streams its events, then its
+        // typed `String` output becomes the student's prompt override
+        // for the wrapped run. Both agents' events flow with their own
+        // agentIds.
+        sessionExec = { input, emitter ->
+            val (override, _) = agents_engine.runtime.events.runAgentInSession(teacher, input, emitter)
+            val (out, _) = agents_engine.runtime.events.runAgentInSession(
+                student, input, emitter,
+                promptOverride = override,
+            )
+            out
+        },
+        execution = { input ->
+            val promptOverride = teacher.invokeSuspend(input)
+            student.invokeSuspendWithPromptOverride(input, promptOverride)
+        },
+    )
 }

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

@@ -268,6 +268,14 @@ class Agent<IN, OUT>(
     internal suspend fun invokeSuspendForSession(
         input: IN,
         emitter: agents_engine.model.AgentEventEmitter? = null,
+        /**
+         * #1747 — optional system-prompt override (used by the `wrap` operator).
+         * When non-null, replaces `this.prompt` as the effective system prompt
+         * for this invocation only. Consolidates the previous separate
+         * `invokeSuspendWithPromptOverride` entry point — that one now
+         * delegates here with `emitter = null`.
+         */
+        promptOverride: String? = null,
         onSkillCompleted: (agents_engine.model.TokenUsage?) -> Unit = { /* no-op */ },
         onSkillStarted: (String) -> Unit,
     ): OUT {
@@ -276,7 +284,11 @@ class Agent<IN, OUT>(
             skillChosenListener?.invoke(skill.name)
             onSkillStarted(skill.name)
             return if (skill.isAgentic) {
-                val result = executeAgentic(this, skill, input, emitter = emitter)
+                val result = executeAgentic(
+                    this, skill, input,
+                    effectivePrompt = promptOverride ?: this.prompt,
+                    emitter = emitter,
+                )
                 // #1740: surface cumulative usage on the way out. Non-agentic
                 // skills don't go through executeAgentic, so onSkillCompleted
                 // stays at its default null for the implementedBy path below.
@@ -320,22 +332,15 @@ class Agent<IN, OUT>(
      * differs (reads the override instead of `agent.prompt`).
      */
     internal suspend fun invokeSuspendWithPromptOverride(input: IN, promptOverride: String): OUT {
-        try {
-            val skill = resolveSkill(input)
-            skillChosenListener?.invoke(skill.name)
-            return if (skill.isAgentic) {
-                castOut(executeAgentic(this, skill, input, effectivePrompt = promptOverride).output)
-            } else {
-                // Non-agentic skills don't read prompt — implementedBy lambdas
-                // ignore the override. Same behavior as the legacy path.
-                castOut(executors[skill.name]!!(input))
-            }
-        } catch (t: Throwable) {
-            errorListener?.let { listener ->
-                try { listener(t) } catch (cb: Throwable) { t.addSuppressed(cb) }
-            }
-            throw t
-        }
+        // #1747 — consolidated into invokeSuspendForSession. emitter = null
+        // preserves the non-streaming behavior the wrap operator used pre-
+        // step 4; the streaming variant goes through runAgentInSession with
+        // the same promptOverride parameter.
+        return invokeSuspendForSession(
+            input = input,
+            emitter = null,
+            promptOverride = promptOverride,
+        ) { /* no-op onSkillStarted */ }
     }
 
     private suspend fun resolveSkill(input: IN): Skill<*, *> {

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

@@ -81,12 +81,20 @@ internal suspend fun <IN, OUT> runAgentInSession(
     agent: Agent<IN, OUT>,
     input: IN,
     emitter: AgentEventEmitter,
+    /**
+     * #1747 — when non-null, runs the agentic loop with this string as
+     * the effective system prompt instead of `agent.prompt`. Used by the
+     * `wrap` operator's session path (teacher's output becomes the
+     * student's per-call system prompt).
+     */
+    promptOverride: String? = null,
 ): Pair<OUT, TokenUsage?> {
     var capturedSkillName: String? = null
     var capturedUsage: TokenUsage? = null
     val output = agent.invokeSuspendForSession(
         input,
         emitter = emitter,
+        promptOverride = promptOverride,
         onSkillCompleted = { usage -> capturedUsage = usage },
     ) { skillName ->
         // SkillStarted fires BEFORE the skill body runs — emitting from

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

@@ -0,0 +1,93 @@
+package agents_engine.runtime.events
+
+import agents_engine.composition.pipeline.session
+import agents_engine.composition.wrap.wrap
+import agents_engine.core.agent
+import agents_engine.model.LlmMessage
+import agents_engine.model.LlmResponse
+import agents_engine.model.ModelClient
+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
+
+// #1747 — wrap (teacher wrap student) returns a Pipeline that runs
+// teacher first to produce a system-prompt String, then runs student
+// under that override. Both agents' events stream through to the
+// session's events Flow with their own agentIds.
+
+class WrapSessionTest {
+
+    @Test
+    fun `wrap session emits ordered events from teacher then student with student-prompt-override active`() = runTest {
+        // Teacher emits a system-prompt string for the student.
+        // Student is agentic — uses a stub ModelClient so the prompt-override
+        // path through executeAgentic is exercised under streaming.
+        val capturedSystemPrompt = mutableListOf<String>()
+        val stub = ModelClient { messages: List<LlmMessage> ->
+            messages.firstOrNull { it.role == "system" }?.content?.let { capturedSystemPrompt += it }
+            LlmResponse.Text("student-output")
+        }
+
+        val teacher = agent<String, String>("teacher") {
+            skills {
+                skill<String, String>("write-prompt", "Emits a system prompt for the student") {
+                    implementedBy { input -> "Teacher emitted prompt for: $input" }
+                }
+            }
+        }
+        val student = agent<String, String>("student") {
+            prompt("Baked-in student prompt that should NOT be used during wrap.")
+            model { ollama("llama3"); client = stub }
+            skills {
+                skill<String, String>("do-task", "Does the actual task via LLM") { tools() }
+            }
+        }
+        val pipeline = teacher wrap student
+
+        val session = pipeline.session("hello")
+        val events = session.events.toList()
+        val output = session.await()
+
+        assertEquals("student-output", output, "wrap pipeline output is student's output")
+
+        // Teacher's events come first.
+        val firstStarted = events.filterIsInstance<AgentEvent.SkillStarted>().first()
+        assertEquals("teacher", firstStarted.agentId)
+        assertEquals("write-prompt", firstStarted.skillName)
+
+        // Student's events follow.
+        val studentStarted = events.filterIsInstance<AgentEvent.SkillStarted>()
+            .firstOrNull { it.agentId == "student" }
+            ?: error("expected SkillStarted(student); got: $events")
+        assertEquals("do-task", studentStarted.skillName)
+
+        // Order: teacher.SkillStarted < teacher.SkillCompleted < student.SkillStarted
+        val teacherStarted = events.indexOfFirst { it is AgentEvent.SkillStarted && it.agentId == "teacher" }
+        val teacherCompleted = events.indexOfFirst { it is AgentEvent.SkillCompleted && it.agentId == "teacher" }
+        val studentStartedIdx = events.indexOfFirst { it is AgentEvent.SkillStarted && it.agentId == "student" }
+        assertTrue(teacherStarted < teacherCompleted, "teacher.SkillStarted < teacher.SkillCompleted")
+        assertTrue(teacherCompleted < studentStartedIdx, "teacher.SkillCompleted < student.SkillStarted")
+
+        // Terminal Completed uses the student's name.
+        val terminal = events.last()
+        assertIs<AgentEvent.Completed<String>>(terminal)
+        assertEquals("student", terminal.agentId, "Completed.agentId = student's name (last agent in chain)")
+        assertEquals("student-output", terminal.output)
+
+        // Prompt override was active: the system prompt the student's LLM saw must contain
+        // the teacher's emitted text, NOT the student's baked-in prompt.
+        assertTrue(capturedSystemPrompt.isNotEmpty(), "student's stub ModelClient must have received a system message")
+        val systemPrompt = capturedSystemPrompt.single()
+        assertTrue(
+            "Teacher emitted prompt for: hello" in systemPrompt,
+            "student's system prompt must carry the teacher's output verbatim; got: \"$systemPrompt\"",
+        )
+        assertTrue(
+            "Baked-in student prompt" !in systemPrompt,
+            "student's baked-in prompt must NOT appear when wrap is active; got: \"$systemPrompt\"",
+        )
+    }
+}