[codex][AMPR-180 #508] Emission protocol foundation (#535)

* AMPR-180 #508 Add emission protocol foundation

Written by Codex.

* AMPR-180 #508 Fix flaky human escalation flow tests

Author: wow-miley

ampere-core/src/commonMain/kotlin/link/socket/ampere/agents/events/EventRepository.kt

@@ -5,6 +5,7 @@ import kotlinx.coroutines.withContext
 import kotlinx.datetime.Instant
 import kotlinx.serialization.SerializationException
 import kotlinx.serialization.json.Json
+import link.socket.ampere.agents.domain.event.EmissionEvent
 import link.socket.ampere.agents.domain.event.Event
 import link.socket.ampere.agents.domain.event.EventId
 import link.socket.ampere.agents.domain.event.EventType
@@ -255,5 +256,6 @@ private fun Event.runIdOrNull(): String? = when (this) {
     is MemoryEvent.MilestoneReached -> runId
     is link.socket.ampere.agents.domain.event.TaskEvent.TaskCompleted -> runId
     is link.socket.ampere.agents.domain.event.TaskEvent.TaskFailed -> runId
+    is EmissionEvent.Produced -> emission.provenance.runId
     else -> null
 }

ampere-core/src/commonMain/kotlin/link/socket/ampere/agents/events/messages/AgentMessageApi.kt

@@ -164,13 +164,17 @@ class AgentMessageApi(
      *    until the human replies.
      * 4. Transition thread back to [EventStatus.Open] and post the reply.
      *
+     * Set [awaitReply] to `false` for legacy fire-and-forget thread escalation
+     * where the caller only needs the durable status transition and bus events.
+     *
      * Thread state transitions are owned exclusively by this method — handlers must not
      * re-transition (CHI cell invariant).
      */
     suspend fun escalateToHuman(
         threadId: MessageThreadId,
         reason: String,
         context: Map<String, String> = emptyMap(),
+        awaitReply: Boolean = true,
     ) {
         val thread = messageRepository
             .findThreadById(threadId)
@@ -227,6 +231,10 @@ class AgentMessageApi(
             ),
         )
 
+        if (!awaitReply) {
+            return
+        }
+
         // Step 3: Produce the Emission and suspend for reply.
         val contextString = context.entries.joinToString("\n") { "${it.key}: ${it.value}" }
             .ifEmpty { null }

ampere-core/src/commonMain/kotlin/link/socket/ampere/agents/events/tickets/TicketOrchestrator.kt

@@ -436,6 +436,7 @@ class TicketOrchestrator(
                     "reportedBy" to reportedByAgentId,
                     "priority" to updatedTicket.priority.name,
                 ),
+                awaitReply = false,
             )
         }
 

ampere-core/src/jvmTest/kotlin/link/socket/ampere/agents/data/EventRepositoryTest.kt

@@ -15,7 +15,16 @@ import kotlinx.coroutines.test.TestScope
 import kotlinx.coroutines.test.UnconfinedTestDispatcher
 import kotlinx.datetime.Clock
 import kotlinx.datetime.Instant
+import kotlinx.serialization.json.JsonPrimitive
 import link.socket.ampere.agents.domain.Urgency
+import link.socket.ampere.agents.domain.emission.Affordance
+import link.socket.ampere.agents.domain.emission.DangerLevel
+import link.socket.ampere.agents.domain.emission.Emission
+import link.socket.ampere.agents.domain.emission.EmissionKind
+import link.socket.ampere.agents.domain.emission.EmissionPayload
+import link.socket.ampere.agents.domain.emission.EmissionProvenance
+import link.socket.ampere.agents.domain.emission.inputDigest
+import link.socket.ampere.agents.domain.event.EmissionEvent
 import link.socket.ampere.agents.domain.event.Event
 import link.socket.ampere.agents.domain.event.EventSource
 import link.socket.ampere.agents.events.EventRepository
@@ -32,13 +41,14 @@ class EventRepositoryTest {
     private val stubEventSourceB = EventSource.Agent("agent-B")
 
     private lateinit var driver: JdbcSqliteDriver
+    private lateinit var database: Database
     private lateinit var repo: EventRepository
 
     @BeforeTest
     fun setUp() {
         driver = JdbcSqliteDriver(JdbcSqliteDriver.IN_MEMORY)
         Database.Schema.create(driver)
-        val database = Database.Companion(driver)
+        database = Database.Companion(driver)
         repo = EventRepository(stubJson, testScope, database)
     }
 
@@ -85,6 +95,52 @@ class EventRepositoryTest {
         }
     }
 
+    @Test
+    fun `saveEvent stores emission produced provenance run id`() {
+        runBlocking {
+            val payload = EmissionPayload.Confirmation(
+                action = "deploy production",
+                preview = "Release v2",
+                dangerLevel = DangerLevel.HIGH,
+            )
+            val digest = inputDigest(payload)
+            val event = EmissionEvent.BaseProduced(
+                eventId = "evt-emission-produced",
+                timestamp = Instant.fromEpochSeconds(1_000),
+                eventSource = stubEventSourceA,
+                urgency = Urgency.HIGH,
+                emission = Emission(
+                    id = "emission-1",
+                    kind = EmissionKind.Confirmation,
+                    payload = payload,
+                    affordances = listOf(
+                        Affordance(
+                            id = "approve",
+                            label = "Approve",
+                            signalPayload = JsonPrimitive("approve"),
+                        ),
+                    ),
+                    provenance = EmissionProvenance(
+                        runId = "run-emission-1",
+                        workflowId = "workflow-1",
+                        sourceEventId = "source-event-1",
+                        toolInvocationId = "tool-1",
+                        pluginId = "plugin-1",
+                        modelId = "model-1",
+                        inputDigest = digest,
+                    ),
+                    dedupKey = digest,
+                    producedAt = Instant.fromEpochSeconds(1_000),
+                ),
+            )
+
+            repo.saveEvent(event).getOrThrow()
+
+            val row = database.eventStoreQueries.getEventById("evt-emission-produced").executeAsOne()
+            assertEquals("run-emission-1", row.run_id)
+        }
+    }
+
     @Test
     fun `query by type returns only matching`() {
         runBlocking {

ampere-core/src/jvmTest/kotlin/link/socket/ampere/agents/events/messages/AgentMessageApiTest.kt

@@ -8,7 +8,9 @@ import kotlin.test.assertEquals
 import kotlin.test.assertNotNull
 import kotlin.test.assertTrue
 import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.cancelAndJoin
 import kotlinx.coroutines.delay
+import kotlinx.coroutines.launch
 import kotlinx.coroutines.runBlocking
 import kotlinx.coroutines.test.TestScope
 import kotlinx.coroutines.test.UnconfinedTestDispatcher
@@ -102,42 +104,50 @@ class AgentMessageApiTest {
             assertEquals(2, fetchedThread2.messages.size)
 
             // Escalate -> WAITING_FOR_HUMAN
-            api.escalateToHuman(
-                threadId = thread.id,
-                reason = "Need approval",
-            )
-            val fetchedThread3 = api.getThread(thread.id).getOrNull()
-            assertNotNull(fetchedThread3)
-            assertEquals(EventStatus.WaitingForHuman, fetchedThread3.status)
-
-            // Posting now should fail, since the thread is waiting for human
-            var threw = false
-            try {
-                api.postMessage(
+            val escalationJob = launch {
+                api.escalateToHuman(
                     threadId = thread.id,
-                    content = "Should fail",
+                    reason = "Need approval",
                 )
-            } catch (e: IllegalArgumentException) {
-                threw = true
-            } catch (e: IllegalStateException) {
-                threw = true
             }
-            assertTrue(threw)
-
-            // Resolve
-            api.resolveThread(thread.id)
-            val fetched4 = api.getThread(thread.id).getOrNull()
-            assertNotNull(fetched4)
-            assertEquals(EventStatus.Resolved, fetched4.status)
-
-            // allow async event handlers to run
-            delay(200)
 
-            // Events were published (at least 1 create, 2 posts, 1 escalation, 2 status changes)
-            assertTrue(received.any { it is MessageEvent.ThreadCreated })
-            assertTrue(received.count { it is MessageEvent.MessagePosted } >= 2)
-            assertTrue(received.any { it is MessageEvent.EscalationRequested })
-            assertTrue(received.count { it is MessageEvent.ThreadStatusChanged } >= 2)
+            try {
+                delay(200)
+                val fetchedThread3 = api.getThread(thread.id).getOrNull()
+                assertNotNull(fetchedThread3)
+                assertEquals(EventStatus.WaitingForHuman, fetchedThread3.status)
+
+                // Posting now should fail, since the thread is waiting for human
+                var threw = false
+                try {
+                    api.postMessage(
+                        threadId = thread.id,
+                        content = "Should fail",
+                    )
+                } catch (e: IllegalArgumentException) {
+                    threw = true
+                } catch (e: IllegalStateException) {
+                    threw = true
+                }
+                assertTrue(threw)
+
+                // Resolve
+                api.resolveThread(thread.id)
+                val fetched4 = api.getThread(thread.id).getOrNull()
+                assertNotNull(fetched4)
+                assertEquals(EventStatus.Resolved, fetched4.status)
+
+                // allow async event handlers to run
+                delay(200)
+
+                // Events were published (at least 1 create, 2 posts, 1 escalation, 2 status changes)
+                assertTrue(received.any { it is MessageEvent.ThreadCreated })
+                assertTrue(received.count { it is MessageEvent.MessagePosted } >= 2)
+                assertTrue(received.any { it is MessageEvent.EscalationRequested })
+                assertTrue(received.count { it is MessageEvent.ThreadStatusChanged } >= 2)
+            } finally {
+                escalationJob.cancelAndJoin()
+            }
 
             // ** TODO: Test subscriptions can be unsubscribed from. */
         }
@@ -166,48 +176,56 @@ class AgentMessageApiTest {
             assertEquals(EventStatus.Open, fetchedThread1.status)
 
             // Escalate to WAITING_FOR_HUMAN
-            api.escalateToHuman(
-                threadId = thread.id,
-                reason = "Need human input",
-            )
-
-            val fetchedThread2 = api.getThread(thread.id).getOrNull()
-            assertNotNull(fetchedThread2)
-            assertEquals(EventStatus.WaitingForHuman, fetchedThread2.status)
-
-            // Verify posting is blocked
-            var blocked = false
-            try {
-                api.postMessage(thread.id, "Should fail")
-            } catch (e: IllegalArgumentException) {
-                blocked = true
+            val escalationJob = launch {
+                api.escalateToHuman(
+                    threadId = thread.id,
+                    reason = "Need human input",
+                )
             }
-            assertTrue(blocked, "Posting should be blocked when waiting for human")
-
-            // Reopen the thread
-            api.reopenThread(thread.id)
-
-            val fetchedThread3 = api.getThread(thread.id).getOrNull()
-            assertNotNull(fetchedThread3)
-            assertEquals(EventStatus.Open, fetchedThread3.status)
-
-            // Now posting should succeed
-            val newMessage = api.postMessage(thread.id, "Human intervention complete")
-            assertEquals("Human intervention complete", newMessage.content)
 
-            // Verify thread has the new message
-            val fetchedThread4 = api.getThread(thread.id).getOrNull()
-            assertNotNull(fetchedThread4)
-            assertEquals(2, fetchedThread4.messages.size)
-
-            // Allow async handlers to run
-            delay(200)
+            try {
+                delay(200)
+
+                val fetchedThread2 = api.getThread(thread.id).getOrNull()
+                assertNotNull(fetchedThread2)
+                assertEquals(EventStatus.WaitingForHuman, fetchedThread2.status)
+
+                // Verify posting is blocked
+                var blocked = false
+                try {
+                    api.postMessage(thread.id, "Should fail")
+                } catch (e: IllegalArgumentException) {
+                    blocked = true
+                }
+                assertTrue(blocked, "Posting should be blocked when waiting for human")
+
+                // Reopen the thread
+                api.reopenThread(thread.id)
 
-            // Verify status change events
-            val statusChanges = received.filterIsInstance<MessageEvent.ThreadStatusChanged>()
-            assertTrue(statusChanges.size >= 2)
-            assertTrue(statusChanges.any { it.newStatus == EventStatus.WaitingForHuman })
-            assertTrue(statusChanges.any { it.newStatus == EventStatus.Open })
+                val fetchedThread3 = api.getThread(thread.id).getOrNull()
+                assertNotNull(fetchedThread3)
+                assertEquals(EventStatus.Open, fetchedThread3.status)
+
+                // Now posting should succeed
+                val newMessage = api.postMessage(thread.id, "Human intervention complete")
+                assertEquals("Human intervention complete", newMessage.content)
+
+                // Verify thread has the new message
+                val fetchedThread4 = api.getThread(thread.id).getOrNull()
+                assertNotNull(fetchedThread4)
+                assertEquals(2, fetchedThread4.messages.size)
+
+                // Allow async handlers to run
+                delay(200)
+
+                // Verify status change events
+                val statusChanges = received.filterIsInstance<MessageEvent.ThreadStatusChanged>()
+                assertTrue(statusChanges.size >= 2)
+                assertTrue(statusChanges.any { it.newStatus == EventStatus.WaitingForHuman })
+                assertTrue(statusChanges.any { it.newStatus == EventStatus.Open })
+            } finally {
+                escalationJob.cancelAndJoin()
+            }
         }
     }
 
@@ -227,19 +245,25 @@ class AgentMessageApiTest {
                 initialMessageContent = "Need a decision",
             )
 
-            api.escalateToHuman(
-                threadId = thread.id,
-                reason = "Need human input on release timing",
-                context = mapOf("release" to "v1"),
-            )
-
-            delay(200)
+            val escalationJob = launch {
+                api.escalateToHuman(
+                    threadId = thread.id,
+                    reason = "Need human input on release timing",
+                    context = mapOf("release" to "v1"),
+                )
+            }
 
-            assertEquals(1, received.size)
-            val event = received.single()
-            assertEquals(thread.id, event.threadId)
-            assertEquals("Need human input on release timing", event.reason)
-            assertEquals(mapOf("release" to "v1"), event.context)
+            try {
+                delay(200)
+
+                assertEquals(1, received.size)
+                val event = received.single()
+                assertEquals(thread.id, event.threadId)
+                assertEquals("Need human input on release timing", event.reason)
+                assertEquals(mapOf("release" to "v1"), event.context)
+            } finally {
+                escalationJob.cancelAndJoin()
+            }
         }
     }