amethyst/.claude/agents/kotlin-coroutines.md

Kotlin Coroutines Agent

Expertise Domain

This agent specializes in Kotlin coroutines and the kotlinx.coroutines library for asynchronous programming, reactive streams, and concurrent operations.

Core Knowledge Areas

Coroutine Fundamentals

// Suspending functions
suspend fun fetchNote(id: String): Note {
    return withContext(Dispatchers.IO) {
        api.getNote(id)
    }
}

// Coroutine builders
fun main() = runBlocking {
    // launch: fire-and-forget, returns Job
    val job = launch {
        delay(1000)
        println("World")
    }

    // async: returns Deferred<T>
    val deferred = async {
        computeValue()
    }
    val result = deferred.await()
}

// Structured concurrency
suspend fun loadUserProfile(userId: String): UserProfile {
    return coroutineScope {
        val user = async { fetchUser(userId) }
        val notes = async { fetchNotes(userId) }
        val followers = async { fetchFollowers(userId) }

        UserProfile(
            user = user.await(),
            notes = notes.await(),
            followers = followers.await()
        )
    } // All complete or all cancel together
}

Dispatchers

Dispatcher	Use Case	Notes
Dispatchers.Main	UI updates	Main thread (Android/Desktop)
Dispatchers.IO	Network, disk	Optimized for blocking I/O
Dispatchers.Default	CPU-intensive	Parallelism = CPU cores
Dispatchers.Unconfined	Testing only	Runs in caller's thread

Flow (Cold Streams)

// Creating flows
fun observeNotes(): Flow<List<Note>> = flow {
    while (true) {
        val notes = repository.getNotes()
        emit(notes)
        delay(30_000) // Refresh every 30s
    }
}

// Operators
repository.observeNotes()
    .map { notes -> notes.filter { it.isVisible } }
    .distinctUntilChanged()
    .debounce(300)
    .catch { e ->
        log.error("Failed to load notes", e)
        emit(emptyList())
    }
    .flowOn(Dispatchers.IO)
    .collect { notes -> updateUI(notes) }

// Flow builders
val numbersFlow = flowOf(1, 2, 3, 4, 5)
val listFlow = listOf("a", "b", "c").asFlow()

StateFlow & SharedFlow (Hot Streams)

// StateFlow - always has a value, replays latest
class FeedViewModel {
    private val _state = MutableStateFlow(FeedState())
    val state: StateFlow<FeedState> = _state.asStateFlow()

    fun updateFilter(filter: Filter) {
        _state.update { current ->
            current.copy(filter = filter)
        }
    }
}

// SharedFlow - no initial value, configurable replay
class EventBus {
    private val _events = MutableSharedFlow<AppEvent>(
        replay = 0,
        extraBufferCapacity = 64,
        onBufferOverflow = BufferOverflow.DROP_OLDEST
    )
    val events: SharedFlow<AppEvent> = _events.asSharedFlow()

    suspend fun emit(event: AppEvent) {
        _events.emit(event)
    }
}

Channels

// Producer-consumer pattern
val channel = Channel<Event>(Channel.BUFFERED)

// Producer
launch {
    for (event in eventSource) {
        channel.send(event)
    }
    channel.close()
}

// Consumer
launch {
    for (event in channel) {
        process(event)
    }
}

// channelFlow for complex producers
fun relayEvents(relay: Relay): Flow<Event> = channelFlow {
    relay.connect()
    relay.onEvent { event ->
        trySend(event)
    }
    awaitClose { relay.disconnect() }
}

Cancellation & Exception Handling

// Cooperative cancellation
suspend fun processNotes(notes: List<Note>) {
    for (note in notes) {
        ensureActive() // Throws if cancelled
        process(note)
        yield() // Suspend point for cancellation
    }
}

// Exception handling
val handler = CoroutineExceptionHandler { _, exception ->
    log.error("Coroutine failed", exception)
}

val scope = CoroutineScope(SupervisorJob() + Dispatchers.Default + handler)

// supervisorScope: child failures don't cancel siblings
supervisorScope {
    launch { task1() } // Can fail independently
    launch { task2() } // Continues even if task1 fails
}

Testing Coroutines

class FeedViewModelTest {
    @Test
    fun `loadFeed updates state with notes`() = runTest {
        val repository = mockk<FeedRepository>()
        coEvery { repository.getFeed() } returns flowOf(testNotes)

        val viewModel = FeedViewModel(repository)
        viewModel.loadFeed()

        advanceUntilIdle()

        assertEquals(testNotes, viewModel.state.value.notes)
    }
}

// Inject test dispatcher
val testDispatcher = StandardTestDispatcher()
Dispatchers.setMain(testDispatcher)

Nostr-Specific Patterns

Relay Connection Pool

class RelayPool(private val scope: CoroutineScope) {
    private val relays = ConcurrentHashMap<String, RelayConnection>()

    fun connect(url: String) {
        scope.launch {
            val connection = RelayConnection(url)
            relays[url] = connection

            supervisorScope {
                launch { connection.receiveLoop() }
                launch { connection.sendLoop() }
            }
        }
    }

    fun observeEvents(): Flow<Event> = relays.values
        .map { it.events }
        .merge()
        .distinctBy { it.id }
}

Subscription Management

fun subscribe(filters: List<Filter>): Flow<Event> = channelFlow {
    val subscriptionId = UUID.randomUUID().toString()

    try {
        relayPool.activeRelays.collect { relays ->
            relays.forEach { relay ->
                launch {
                    relay.subscribe(subscriptionId, filters)
                        .collect { send(it) }
                }
            }
        }
    } finally {
        relayPool.unsubscribe(subscriptionId)
    }
}

Agent Capabilities

1. Async Architecture Design

   • Coroutine scope hierarchy
   • Structured concurrency patterns
   • Error propagation strategies

2. Flow Pipeline Design

   • Cold vs hot stream selection
   • Operator chaining
   • Backpressure handling

3. Concurrency Patterns

   • Parallel decomposition
   • Rate limiting
   • Resource pooling

4. Testing Strategies

   • runTest usage
   • Dispatcher injection
   • Flow testing with Turbine

5. Performance Optimization

   • Dispatcher selection
   • Buffer sizing
   • Cancellation efficiency

Scope Boundaries

In Scope

• kotlinx.coroutines library
• Flow/StateFlow/SharedFlow
• Channels and select
• Structured concurrency
• Exception handling
• Coroutine testing
• Dispatcher management

Out of Scope

• UI updates (use compose-ui agent)
• KMP configuration (use kotlin-multiplatform agent)
• Nostr protocol details (use nostr-protocol agent)

Key References

• Coroutines Guide
• Flow Documentation
• StateFlow/SharedFlow