feat(jetbrains): show worktree activity badges

2026-09-21 14:07:20 +08:00 · 2026-07-30 17:18:59 -04:00
parent a678e549f5
commit 919d312fa4
25 changed files with 535 additions and 34 deletions
@@ -0,0 +1,5 @@
+---
+"@kilocode/kilo-jetbrains": minor
+---
+
+Show an activity badge on each Agent Manager worktree row, reflecting whether that worktree's sessions are running or waiting on a question or permission — even when the worktree's editor tab is not open.
@@ -0,0 +1,136 @@
+package ai.kilocode.backend.app
+
+import ai.kilocode.log.KiloLog
+import ai.kilocode.rpc.dto.ChatEventDto
+import ai.kilocode.rpc.dto.SessionActivityDto
+import ai.kilocode.rpc.dto.SessionActivityKindDto
+import ai.kilocode.rpc.dto.SessionStatusDto
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Job
+import kotlinx.coroutines.flow.MutableStateFlow
+import kotlinx.coroutines.flow.SharedFlow
+import kotlinx.coroutines.flow.StateFlow
+import kotlinx.coroutines.flow.asStateFlow
+import kotlinx.coroutines.launch
+
+/**
+ * Tracks live per-session activity (busy + pending question/permission)
+ * with each session's directory, exposed to the frontend so the Agent
+ * Manager worktree list can badge rows.
+ *
+ * **Not an IntelliJ service** — owned by [KiloBackendAppService] which
+ * calls [start] after [KiloAppState.Ready] and [stop] on disconnect.
+ *
+ * Baseline `busy` comes from the session status stream; question/permission
+ * overlays come from the global chat events. State is confined to a single
+ * lock so the two collectors don't race.
+ */
+class KiloBackendActivityManager(
+    private val cs: CoroutineScope,
+    private val log: KiloLog,
+) {
+    private val permissions = mutableMapOf<String, MutableSet<String>>()
+    private val questions = mutableMapOf<String, MutableMap<String, Boolean>>()
+    private val lock = Any()
+    private val _activity = MutableStateFlow<Map<String, SessionActivityDto>>(emptyMap())
+    val activity: StateFlow<Map<String, SessionActivityDto>> = _activity.asStateFlow()
+
+    private var statuses: StateFlow<Map<String, SessionStatusDto>>? = null
+    private var directory: (String) -> String? = { null }
+    private var status: Job? = null
+    private var events: Job? = null
+
+    fun start(
+        statuses: StateFlow<Map<String, SessionStatusDto>>,
+        directory: (String) -> String?,
+        chatEvents: SharedFlow<ChatEventDto>,
+    ) {
+        this.statuses = statuses
+        this.directory = directory
+        if (status?.isActive == true || events?.isActive == true) return
+        status = cs.launch {
+            statuses.collect { synchronized(lock) { recompute() } }
+        }
+        events = cs.launch {
+            chatEvents.collect { event ->
+                synchronized(lock) {
+                    handle(event)
+                    recompute()
+                }
+            }
+        }
+        log.info("Activity manager started")
+    }
+
+    fun stop() {
+        status?.cancel()
+        events?.cancel()
+        status = null
+        events = null
+        statuses = null
+        directory = { null }
+        synchronized(lock) {
+            permissions.clear()
+            questions.clear()
+        }
+        _activity.value = emptyMap()
+        log.info("Activity manager stopped")
+    }
+
+    private fun handle(event: ChatEventDto) {
+        when (event) {
+            is ChatEventDto.PermissionAsked -> permissions.getOrPut(event.sessionID) { mutableSetOf() }.add(event.request.id)
+            is ChatEventDto.PermissionReplied -> removeSet(permissions, event.sessionID, event.requestID)
+            is ChatEventDto.QuestionAsked -> questions.getOrPut(event.sessionID) { mutableMapOf() }[event.request.id] = plan(event)
+            is ChatEventDto.QuestionReplied -> removeMap(questions, event.sessionID, event.requestID)
+            is ChatEventDto.QuestionRejected -> removeMap(questions, event.sessionID, event.requestID)
+            is ChatEventDto.SessionIdle -> clear(event.sessionID)
+            is ChatEventDto.SessionStatusChanged -> if (event.status.type == "idle") clear(event.sessionID)
+            else -> Unit
+        }
+    }
+
+    private fun recompute() {
+        val current = statuses?.value ?: emptyMap()
+        val ids = LinkedHashSet<String>()
+        ids.addAll(current.filterValues { it.type == "busy" }.keys)
+        ids.addAll(permissions.keys)
+        ids.addAll(questions.keys)
+        _activity.value = ids.mapNotNull { id ->
+            val dir = directory(id) ?: return@mapNotNull null
+            val kind = kind(id, current[id]?.type == "busy") ?: return@mapNotNull null
+            id to SessionActivityDto(dir, kind)
+        }.toMap()
+    }
+
+    private fun kind(id: String, busy: Boolean): SessionActivityKindDto? {
+        if (permissions[id]?.isNotEmpty() == true) return SessionActivityKindDto.PERMISSION
+        val pending = questions[id]
+        if (pending?.isNotEmpty() == true) {
+            if (pending.values.any { it }) return SessionActivityKindDto.PLAN
+            return SessionActivityKindDto.QUESTION
+        }
+        if (busy) return SessionActivityKindDto.RUNNING
+        return null
+    }
+
+    private fun plan(event: ChatEventDto.QuestionAsked): Boolean =
+        event.request.questions.any { it.questionKey == "plan.followup.question" || it.headerKey == "plan.followup.header" }
+
+    private fun clear(id: String) {
+        permissions.remove(id)
+        questions.remove(id)
+    }
+
+    private fun <T> removeSet(map: MutableMap<String, MutableSet<T>>, id: String, value: T) {
+        val set = map[id] ?: return
+        set.remove(value)
+        if (set.isEmpty()) map.remove(id)
+    }
+
+    private fun <T> removeMap(map: MutableMap<String, MutableMap<T, Boolean>>, id: String, value: T) {
+        val items = map[id] ?: return
+        items.remove(value)
+        if (items.isEmpty()) map.remove(id)
+    }
+}
@@ -135,6 +135,7 @@ class KiloBackendAppService private constructor(

    val sessions = KiloBackendSessionManager(cs, log)
    val chat = KiloBackendChatManager(cs, log)
+    val activity = KiloBackendActivityManager(cs, log)
    val models = KiloBackendModelStateManager(log)
    val workspaces = KiloBackendWorkspaceManager(cs, sessions, log)
    @Volatile var profile: KiloProfile200Response? = null
@@ -458,6 +459,7 @@ class KiloBackendAppService private constructor(
                    models.start(connection.apiClient!!, connection.port)
                    sessions.start(connection.api!!, connection.apiClient!!, connection.port, connection.events)
                    chat.start(connection.apiClient!!, connection.port, connection.events)
+                    activity.start(sessions.statuses, sessions::sessionDirectory, chat.events)
                    workspaces.start(connection.api!!, connection.apiClient!!, connection.port, connection.events)
                    startWatchingGlobalSseEvents()
                    setTelemetry(true)
@@ -878,6 +880,7 @@ class KiloBackendAppService private constructor(
    private fun stopRuntime() {
        workspaces.stop()
        models.stop()
+        activity.stop()
        chat.stop()
        sessions.stop()
    }
@@ -50,6 +50,9 @@ class KiloBackendSessionManager(
    /** Per-session directory overrides (sessionId → worktree path). */
    private val directories = ConcurrentHashMap<String, String>()

+    /** Session directory cache populated while mapping CLI sessions. */
+    private val owned = ConcurrentHashMap<String, String>()
+
    private val _statuses = MutableStateFlow<Map<String, SessionStatusDto>>(emptyMap())
    val statuses: StateFlow<Map<String, SessionStatusDto>> = _statuses.asStateFlow()

@@ -95,6 +98,7 @@ class KiloBackendSessionManager(
        client = null
        http = null
        base = null
+        owned.clear()
        _statuses.value = emptyMap()
        log.info("Session manager stopped")
    }
@@ -154,6 +158,7 @@ class KiloBackendSessionManager(
            val dto = KiloCliDataParser.parseSession(raw!!)
            val meta = if (log.isDebugEnabled) ChatLogSummary.dir(dir) else "kind=session"
            log.info("${ChatLogSummary.sid(dto.id)} kind=session $meta created=true code=${response.code}")
+            owned[dto.id] = dto.directory
            return dto
        }
    }
@@ -168,6 +173,7 @@ class KiloBackendSessionManager(
    fun delete(id: String, dir: String) {
        requireClient().sessionDelete(sessionID = id, directory = dir)
        directories.remove(id)
+        owned.remove(id)
    }

    /**
@@ -196,7 +202,9 @@ class KiloBackendSessionManager(
                log.warn("Session rename failed: HTTP ${response.code}, body=$raw")
                throw RuntimeException("Session rename failed: HTTP ${response.code} — $raw")
            }
-            return KiloCliDataParser.parseSession(raw!!)
+            val dto = KiloCliDataParser.parseSession(raw!!)
+            owned[dto.id] = dto.directory
+            return dto
        }
    }

@@ -241,7 +249,9 @@ class KiloBackendSessionManager(
                log.warn("Cloud session import failed: HTTP ${response.code}, body=$raw")
                throw RuntimeException("Cloud session import failed: HTTP ${response.code} — $raw")
            }
-            return KiloCliDataParser.parseSession(raw)
+            val dto = KiloCliDataParser.parseSession(raw)
+            owned[dto.id] = dto.directory
+            return dto
        }
    }

@@ -266,6 +276,9 @@ class KiloBackendSessionManager(
    fun getDirectory(id: String, fallback: String): String =
        directories[id] ?: fallback

+    fun sessionDirectory(id: String): String? =
+        directories[id] ?: owned[id]
+
    // ------ mapping (generated API model → DTO) ------

    private fun dto(s: ai.kilocode.jetbrains.api.model.Session) = dto(
@@ -322,21 +335,24 @@ class KiloBackendSessionManager(
        archived: Double?,
        summary: SessionSummaryDto?,
        revert: SessionRevertDto?,
-    ) = SessionDto(
-        id = id,
-        projectID = project,
-        directory = dir,
-        parentID = parent,
-        title = title,
-        version = version,
-        time = SessionTimeDto(
-            created = time(id, "created", created),
-            updated = time(id, "updated", updated),
-            archived = archived,
-        ),
-        summary = summary,
-        revert = revert,
-    )
+    ): SessionDto {
+        owned[id] = dir
+        return SessionDto(
+            id = id,
+            projectID = project,
+            directory = dir,
+            parentID = parent,
+            title = title,
+            version = version,
+            time = SessionTimeDto(
+                created = time(id, "created", created),
+                updated = time(id, "updated", updated),
+                archived = archived,
+            ),
+            summary = summary,
+            revert = revert,
+        )
+    }

    private fun summary(add: Double?, del: Double?, files: Double?) = SessionSummaryDto(
        additions = count(add),
@@ -3,6 +3,7 @@
 package ai.kilocode.backend.rpc

 import ai.kilocode.backend.app.KiloBackendAppService
+import ai.kilocode.backend.app.KiloBackendActivityManager
 import ai.kilocode.backend.app.KiloBackendChatManager
 import ai.kilocode.backend.app.KiloBackendSessionManager
 import ai.kilocode.backend.workspace.KiloBackendWorkspaceManager
@@ -21,6 +22,7 @@ import ai.kilocode.rpc.dto.PromptDto
 import ai.kilocode.rpc.dto.QuestionReplyDto
 import ai.kilocode.rpc.dto.QuestionRequestDto
 import ai.kilocode.rpc.dto.SessionDto
+import ai.kilocode.rpc.dto.SessionActivityDto
 import ai.kilocode.rpc.dto.SessionListDto
 import ai.kilocode.rpc.dto.SessionStatusDto
 import com.intellij.openapi.components.service
@@ -58,6 +60,9 @@ class KiloSessionRpcApiImpl internal constructor(
    private val chat: KiloBackendChatManager
        get() = app.chat

+    private val activity: KiloBackendActivityManager
+        get() = app.activity
+
    private val app: KiloBackendAppService
        get() = appOverride ?: service()

@@ -103,6 +108,9 @@ class KiloSessionRpcApiImpl internal constructor(
    override suspend fun statuses(): Flow<Map<String, SessionStatusDto>> =
        sessions.statuses

+    override suspend fun activity(): Flow<Map<String, SessionActivityDto>> =
+        activity.activity
+
    override suspend fun setDirectory(id: String, directory: String) =
        sessions.setDirectory(id, directory)

@@ -0,0 +1,123 @@
+package ai.kilocode.backend.app
+
+import ai.kilocode.backend.testing.TestLog
+import ai.kilocode.rpc.dto.ChatEventDto
+import ai.kilocode.rpc.dto.PermissionRequestDto
+import ai.kilocode.rpc.dto.QuestionInfoDto
+import ai.kilocode.rpc.dto.QuestionRequestDto
+import ai.kilocode.rpc.dto.SessionActivityKindDto
+import ai.kilocode.rpc.dto.SessionStatusDto
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.SupervisorJob
+import kotlinx.coroutines.cancel
+import kotlinx.coroutines.flow.MutableSharedFlow
+import kotlinx.coroutines.flow.MutableStateFlow
+import kotlinx.coroutines.flow.first
+import kotlinx.coroutines.runBlocking
+import kotlinx.coroutines.withTimeout
+import kotlin.test.AfterTest
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertFalse
+
+class KiloBackendActivityManagerTest {
+    private val scope = CoroutineScope(SupervisorJob() + Dispatchers.Default)
+    private val events = MutableSharedFlow<ChatEventDto>(extraBufferCapacity = 16)
+    private val statuses = MutableStateFlow<Map<String, SessionStatusDto>>(emptyMap())
+    private val directories = mutableMapOf<String, String>()
+    private val manager = KiloBackendActivityManager(scope, TestLog())
+
+    @AfterTest
+    fun tearDown() {
+        manager.stop()
+        scope.cancel()
+    }
+
+    private fun start() = manager.start(statuses, { directories[it] }, events)
+
+    @Test
+    fun `busy status with known directory emits running activity`() = runBlocking {
+        directories["ses_1"] = "/repo/wt"
+        start()
+
+        statuses.value = mapOf("ses_1" to SessionStatusDto("busy"))
+
+        val snap = await("ses_1", SessionActivityKindDto.RUNNING)
+        assertEquals("/repo/wt", snap["ses_1"]?.directory)
+    }
+
+    @Test
+    fun `permission asked overlays running and reply reverts`() = runBlocking {
+        directories["ses_1"] = "/repo/wt"
+        statuses.value = mapOf("ses_1" to SessionStatusDto("busy"))
+        start()
+
+        events.emit(ChatEventDto.PermissionAsked("ses_1", PermissionRequestDto("perm_1", "ses_1", "edit", emptyList())))
+        await("ses_1", SessionActivityKindDto.PERMISSION)
+
+        events.emit(ChatEventDto.PermissionReplied("ses_1", "perm_1"))
+
+        await("ses_1", SessionActivityKindDto.RUNNING)
+    }
+
+    @Test
+    fun `question kinds distinguish plain and plan followup`() = runBlocking {
+        directories["ses_plain"] = "/repo/a"
+        directories["ses_plan"] = "/repo/b"
+        start()
+
+        events.emit(ChatEventDto.QuestionAsked("ses_plain", question("q_1", "ses_plain")))
+        events.emit(ChatEventDto.QuestionAsked("ses_plan", question("q_2", "ses_plan", plan = true)))
+
+        await("ses_plain", SessionActivityKindDto.QUESTION)
+        await("ses_plan", SessionActivityKindDto.PLAN)
+    }
+
+    @Test
+    fun `idle clears pending overlays and removes inactive entry`() = runBlocking {
+        directories["ses_1"] = "/repo/wt"
+        statuses.value = mapOf("ses_1" to SessionStatusDto("busy"))
+        start()
+        events.emit(ChatEventDto.QuestionAsked("ses_1", question("q_1", "ses_1")))
+        await("ses_1", SessionActivityKindDto.QUESTION)
+
+        statuses.value = mapOf("ses_1" to SessionStatusDto("idle"))
+        events.emit(ChatEventDto.SessionIdle("ses_1"))
+
+        withTimeout(5_000) { manager.activity.first { "ses_1" !in it } }
+        assertFalse("ses_1" in manager.activity.value)
+    }
+
+    @Test
+    fun `unknown session directory is omitted`() = runBlocking {
+        directories["ses_known"] = "/repo/wt"
+        start()
+
+        statuses.value = mapOf(
+            "ses_known" to SessionStatusDto("busy"),
+            "ses_unknown" to SessionStatusDto("busy"),
+        )
+
+        val snap = await("ses_known", SessionActivityKindDto.RUNNING)
+        assertFalse("ses_unknown" in snap)
+    }
+
+    private suspend fun await(id: String, kind: SessionActivityKindDto) = withTimeout(5_000) {
+        manager.activity.first { it[id]?.kind == kind }
+    }
+
+    private fun question(id: String, session: String, plan: Boolean = false): QuestionRequestDto {
+        val info = if (plan) {
+            QuestionInfoDto(
+                question = "Ready to implement?",
+                header = "Implement",
+                questionKey = "plan.followup.question",
+                headerKey = "plan.followup.header",
+            )
+        } else {
+            QuestionInfoDto(question = "Pick one", header = "Choice")
+        }
+        return QuestionRequestDto(id = id, sessionID = session, questions = listOf(info))
+    }
+}
@@ -2,6 +2,7 @@ package ai.kilocode.client

 import ai.kilocode.client.app.KiloWorkspaceService
 import ai.kilocode.client.app.Workspace
+import ai.kilocode.client.app.KiloSessionService
 import ai.kilocode.client.session.SessionManager
 import ai.kilocode.client.session.SessionSidePanelManager
 import ai.kilocode.client.telemetry.Telemetry
@@ -88,7 +89,12 @@ internal class KiloToolWindowSetupService(
        try {
            val manager = SessionSidePanelManager(project, workspace)

-            val worktrees = WorktreeController(service<KiloWorktreeService>(), workspace.directory, cs)
+            val worktrees = WorktreeController(
+                service<KiloWorktreeService>(),
+                workspace.directory,
+                cs,
+                activity = project.service<KiloSessionService>().activity,
+            )
            val agentManagerPanel = AgentManagerPanel(manager, worktrees, project)

            val chat = object : JPanel(BorderLayout()), DataProvider {
@@ -9,8 +9,10 @@ import ai.kilocode.client.agentManager.worktree.WorktreeSessionEditorKind
 import ai.kilocode.client.agentManager.worktree.ensureWorktreeSessionEditorKind
 import ai.kilocode.client.agentManager.worktree.worktreeSessionParams
 import ai.kilocode.client.plugin.KiloBundle
+import ai.kilocode.client.session.SessionActivityKind
 import ai.kilocode.client.ui.UiStyle
 import ai.kilocode.client.ui.list.ActiveList
+import ai.kilocode.client.ui.list.ActiveListBadge
 import ai.kilocode.client.ui.list.ActiveListCell
 import ai.kilocode.client.ui.list.ActiveListDeleteOptions
 import ai.kilocode.client.ui.list.ActiveListItem
@@ -90,6 +92,7 @@ class AgentManagerPanel(
        }
        controller.onCreateFailure = { err -> notifyCreateFailed(err) }
        controller.onRemoveSuccess = { item -> close(item) }
+        controller.onActivityChanged = { sync() }
        // Reflect names adopted or renamed in a worktree session editor tab in the list live.
        service<WorktreeNameCache>().addListener(this) { path, name -> controller.applyName(path, name) }
        ActionManager.getInstance().getAction("RenameElement")?.shortcutSet?.let { set ->
@@ -239,7 +242,7 @@ class AgentManagerPanel(
        list.update(
            (0 until controller.model.size).map {
                val item = controller.model.getElementAt(it)
-                WorktreeRow(item, controller.isPending(item.id), controller.isDeleting(item.id))
+                WorktreeRow(item, controller.isPending(item.id), controller.isDeleting(item.id), controller.kind(item.path))
            },
            ActiveListSelection.PreserveNoScroll,
        )
@@ -267,6 +270,7 @@ class AgentManagerPanel(
        controller.onSelect = null
        controller.onCreateFailure = null
        controller.onRemoveSuccess = null
+        controller.onActivityChanged = null
    }

    override fun uiDataSnapshot(sink: DataSink) {
@@ -306,13 +310,23 @@ class AgentManagerPanel(
        }
    }

-    private data class WorktreeRow(val dto: WorktreeDto, val pending: Boolean, override val deleting: Boolean) : ActiveListItem {
+    private data class WorktreeRow(
+        val dto: WorktreeDto,
+        val pending: Boolean,
+        override val deleting: Boolean,
+        val kind: SessionActivityKind?,
+    ) : ActiveListItem {
        override val key: String get() = dto.id
        override val title: String get() = dto.name
        override val description: String get() = dto.path.trimEnd('/').substringAfterLast('/')
        override val tooltip: String get() = dto.path
        override val icon = WorktreeIcons.forRow(dto.locked, pending)
        override val search: String get() = listOfNotNull(dto.name, dto.branch, dto.path, dto.lockReason).joinToString(" ")
+        override val badges: List<ActiveListBadge>
+            get() {
+                if (pending || deleting) return emptyList()
+                return listOfNotNull(kind?.let { ActiveListBadge(it.label(), it.style()) })
+            }
        override val cells: List<ActiveListCell>
            get() = if (dto.main || pending) emptyList() else listOf(
                ActiveListCell(
@@ -0,0 +1,30 @@
+package ai.kilocode.client.agentManager.worktree
+
+import ai.kilocode.client.session.SessionActivityKind
+import ai.kilocode.rpc.dto.SessionActivityDto
+import ai.kilocode.rpc.dto.SessionActivityKindDto
+
+internal fun aggregateWorktreeActivity(
+    activity: Map<String, SessionActivityDto>,
+): Map<String, SessionActivityKind> = activity.values
+    .groupBy { normalize(it.directory) }
+    .mapValues { (_, items) -> items.map { kind(it.kind) }.minBy(::rank) }
+
+internal fun normalizeWorktreePath(path: String): String = normalize(path)
+
+private fun normalize(path: String): String = path.trimEnd('/')
+
+private fun kind(kind: SessionActivityKindDto): SessionActivityKind = when (kind) {
+    SessionActivityKindDto.RUNNING -> SessionActivityKind.RUNNING
+    SessionActivityKindDto.QUESTION -> SessionActivityKind.QUESTION
+    SessionActivityKindDto.PLAN -> SessionActivityKind.PLAN
+    SessionActivityKindDto.PERMISSION -> SessionActivityKind.PERMISSION
+}
+
+private fun rank(kind: SessionActivityKind): Int = when (kind) {
+    SessionActivityKind.PERMISSION -> 0
+    SessionActivityKind.QUESTION -> 1
+    SessionActivityKind.PLAN -> 2
+    SessionActivityKind.RUNNING -> 3
+    SessionActivityKind.LOGIN_REQUIRED -> 4
+}
@@ -1,14 +1,18 @@
 package ai.kilocode.client.agentManager.worktree

 import ai.kilocode.client.telemetry.Telemetry
+import ai.kilocode.client.session.SessionActivityKind
 import ai.kilocode.rpc.dto.CreateWorktreeRequestDto
 import ai.kilocode.rpc.dto.RemoveWorktreeResultDto
+import ai.kilocode.rpc.dto.SessionActivityDto
 import ai.kilocode.rpc.dto.WorktreeDto
 import com.intellij.openapi.application.ApplicationManager
 import com.intellij.openapi.components.service
 import com.intellij.ui.CollectionListModel
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.async
+import kotlinx.coroutines.flow.MutableStateFlow
+import kotlinx.coroutines.flow.StateFlow
 import kotlinx.coroutines.launch
 import java.util.Collections

@@ -20,6 +24,7 @@ class WorktreeController(
    private val service: KiloWorktreeService,
    private val directory: String,
    private val cs: CoroutineScope,
+    activity: StateFlow<Map<String, SessionActivityDto>> = MutableStateFlow(emptyMap()),
    private val telemetry: (String, Map<String, String>) -> Unit = { event, props -> Telemetry.send(event, props) },
 ) {
    val model = CollectionListModel<WorktreeDto>()
@@ -28,6 +33,21 @@ class WorktreeController(
    var onSelect: ((String) -> Unit)? = null
    var onCreateFailure: ((String?) -> Unit)? = null
    var onRemoveSuccess: ((WorktreeDto) -> Unit)? = null
+    var onActivityChanged: (() -> Unit)? = null
+
+    @Volatile
+    private var kinds: Map<String, SessionActivityKind> = emptyMap()
+
+    init {
+        cs.launch {
+            activity.collect { snap ->
+                edt {
+                    kinds = aggregateWorktreeActivity(snap)
+                    onActivityChanged?.invoke()
+                }
+            }
+        }
+    }

    /** Branch checked out in the main worktree; used as the base for quick worktree creation. */
    @Volatile
@@ -47,6 +67,8 @@ class WorktreeController(

    fun isDeleting(id: String): Boolean = id in deleting

+    fun kind(path: String): SessionActivityKind? = kinds[normalizeWorktreePath(path)]
+
    fun reload() {
        cs.launch {
            val listing = async { service.list(directory) }
@@ -43,7 +43,7 @@ open class WorktreeSessionEditorManager(
            service<SessionUiFactory>().create(project, workspace, manager, ref, timers)
        },
    resolve: (String) -> Workspace = { dir -> service<KiloWorkspaceService>().workspace(dir) },
-    status: () -> Map<String, SessionActivityKind> = { project.service<KiloSessionService>().activity() },
+    status: () -> Map<String, SessionActivityKind> = { project.service<KiloSessionService>().activitySnapshot() },
    timers: UiTimerSource = UiTimers,
    request: (JComponent) -> Unit = { focus ->
        ApplicationManager.getApplication().invokeLater({
@@ -17,6 +17,7 @@ import ai.kilocode.rpc.dto.PartDto
 import ai.kilocode.rpc.dto.PromptDto
 import ai.kilocode.rpc.dto.QuestionReplyDto
 import ai.kilocode.rpc.dto.QuestionRequestDto
+import ai.kilocode.rpc.dto.SessionActivityDto
 import ai.kilocode.rpc.dto.SessionDto
 import ai.kilocode.rpc.dto.SessionListDto
 import ai.kilocode.rpc.dto.SessionStatusDto
@@ -65,6 +66,10 @@ class KiloSessionService internal constructor(
    val statuses: StateFlow<Map<String, SessionStatusDto>> =
        stream { statuses() }.stateIn(cs, SharingStarted.Eagerly, emptyMap())

+    /** Live session activity map from backend global events. */
+    val activity: StateFlow<Map<String, SessionActivityDto>> =
+        stream { activity() }.stateIn(cs, SharingStarted.Eagerly, emptyMap())
+
    // ------ RPC helpers ------

    private suspend fun <T> call(block: suspend KiloSessionRpcApi.() -> T): T {
@@ -91,7 +96,7 @@ class KiloSessionService internal constructor(
        }
    }

-    internal fun activity(): Map<String, SessionActivityKind> =
+    internal fun activitySnapshot(): Map<String, SessionActivityKind> =
        statuses.value
            .filterValues { it.type == "busy" }
            .mapValues { SessionActivityKind.RUNNING }
@@ -29,7 +29,7 @@ class SessionSidePanelManager(
            service<SessionUiFactory>().create(project, workspace, manager, ref, timers)
        },
    resolve: (String) -> Workspace = { dir -> service<KiloWorkspaceService>().workspace(dir) },
-    status: () -> Map<String, SessionActivityKind> = { project.service<KiloSessionService>().activity() },
+    status: () -> Map<String, SessionActivityKind> = { project.service<KiloSessionService>().activitySnapshot() },
    private val history: ((Disposable, (SessionRef) -> Unit, (String) -> Unit) -> JComponent)? = null,
    timers: UiTimerSource = UiTimers,
    request: (JComponent) -> Unit = { focus ->
@@ -509,7 +509,7 @@ class SessionUi(
                        controller,
                        event.recents,
                        history = { manager?.showHistory() },
-                        activity = { manager?.activity() ?: sessions.activity() },
+                        activity = { manager?.activity() ?: sessions.activitySnapshot() },
                        titles = { manager?.titles().orEmpty() },
                        timers = timers,
                    )
@@ -64,7 +64,7 @@ class HistoryController(
        reloadCloud()
    }

-    internal fun activity() = sessions.activity()
+    internal fun activity() = sessions.activitySnapshot()

    fun reloadLocal() {
        edt { local.start() }
@@ -8,6 +8,8 @@ import ai.kilocode.client.agentManager.worktree.worktreeSessionParams
 import ai.kilocode.client.testing.FakeWorktreeRpcApi
 import ai.kilocode.client.testing.TestCoroutines
 import ai.kilocode.client.testing.fire
+import ai.kilocode.client.session.SessionActivityKind
+import ai.kilocode.client.ui.list.ActiveListBadge
 import ai.kilocode.client.ui.list.ActiveListItem
 import ai.kilocode.client.ui.list.activeListToolWindowBackground
 import ai.kilocode.client.vfs.KiloPath
@@ -15,6 +17,8 @@ import ai.kilocode.client.vfs.KiloVfsManager
 import ai.kilocode.client.vfs.KiloVirtualFile
 import ai.kilocode.client.vfs.KiloVirtualFileSystem
 import ai.kilocode.rpc.dto.WorktreeDto
+import ai.kilocode.rpc.dto.SessionActivityDto
+import ai.kilocode.rpc.dto.SessionActivityKindDto
 import com.intellij.openapi.application.ApplicationManager
 import com.intellij.openapi.components.service
 import com.intellij.openapi.fileEditor.FileEditorManager
@@ -27,6 +31,7 @@ import java.awt.event.MouseEvent
 import javax.swing.JComponent
 import javax.swing.SwingUtilities
 import kotlinx.coroutines.CompletableDeferred
+import kotlinx.coroutines.flow.MutableStateFlow

@Suppress("UnstableApiUsage")
 class AgentManagerPanelTest : BasePlatformTestCase() {
@@ -207,6 +212,40 @@ class AgentManagerPanelTest : BasePlatformTestCase() {
        assertNull(KiloVirtualFileSystem.getInstance().cached(path))
    }

+    fun `test worktree row shows activity badge for matching directory`() {
+        val item = WorktreeDto("/repo/.kilo/worktrees/feature-x", "feature-x", "feature/x", "/repo/.kilo/worktrees/feature-x")
+        val activity = MutableStateFlow(mapOf(
+            "ses_1" to SessionActivityDto(item.path, SessionActivityKindDto.QUESTION),
+        ))
+        rpc.listed += item
+        val controller = WorktreeController(service, "/test", coroutines.scope, activity = activity)
+        val panel = edt { AgentManagerPanel(testRootDisposable, controller) }
+        edt { controller.reload() }
+        flush()
+
+        val row = row(panel, 0)
+        assertEquals(listOf(ActiveListBadge(SessionActivityKind.QUESTION.label(), SessionActivityKind.QUESTION.style())), row.badges)
+    }
+
+    fun `test worktree row hides badge while pending or deleting`() {
+        val path = "feature/y"
+        val activity = MutableStateFlow(mapOf(
+            "ses_1" to SessionActivityDto(path, SessionActivityKindDto.RUNNING),
+        ))
+        val gate = CompletableDeferred<Unit>()
+        rpc.beforeCreate = { gate.await() }
+        val controller = WorktreeController(service, "/test", coroutines.scope, activity = activity)
+        val panel = edt { AgentManagerPanel(testRootDisposable, controller) }
+
+        edt { controller.create("feature/y", null) }
+        flush()
+
+        val pending = row(panel, 0)
+        assertEquals(emptyList<ActiveListBadge>(), pending.badges)
+        gate.complete(Unit)
+        flush()
+    }
+
    private fun <T> edt(block: () -> T): T {
        val out = arrayOfNulls<Any?>(1)
        ApplicationManager.getApplication().invokeAndWait { out[0] = block() }
@@ -216,6 +255,11 @@ class AgentManagerPanelTest : BasePlatformTestCase() {

    private fun flush() = coroutines.drain(::pump)

+    private fun row(panel: AgentManagerPanel, idx: Int): ActiveListItem {
+        val list = edt { UIUtil.findComponentOfType(panel, JBList::class.java)!! }
+        return edt { list.model.getElementAt(idx) as ActiveListItem }
+    }
+
    private fun pump() {
        ApplicationManager.getApplication().invokeAndWait { UIUtil.dispatchAllInvocationEvents() }
    }
@@ -5,17 +5,21 @@ import ai.kilocode.client.agentManager.worktree.KiloWorktreeService
 import ai.kilocode.client.agentManager.worktree.WorktreeController
 import ai.kilocode.client.agentManager.worktree.WorktreeNameCache
 import ai.kilocode.client.agentManager.worktree.WorktreeNames
+import ai.kilocode.client.session.SessionActivityKind
 import ai.kilocode.client.testing.FakeWorktreeRpcApi
 import ai.kilocode.client.testing.TestCoroutines
 import ai.kilocode.rpc.dto.CreateWorktreeResultDto
 import ai.kilocode.rpc.dto.RemoveWorktreeResultDto
 import ai.kilocode.rpc.dto.RenameWorktreeResultDto
+import ai.kilocode.rpc.dto.SessionActivityDto
+import ai.kilocode.rpc.dto.SessionActivityKindDto
 import ai.kilocode.rpc.dto.WorktreeDto
 import com.intellij.openapi.application.ApplicationManager
 import com.intellij.openapi.components.service
 import com.intellij.testFramework.fixtures.BasePlatformTestCase
 import com.intellij.util.ui.UIUtil
 import kotlinx.coroutines.CompletableDeferred
+import kotlinx.coroutines.flow.MutableStateFlow

@Suppress("UnstableApiUsage")
 class WorktreeControllerTest : BasePlatformTestCase() {
@@ -331,6 +335,21 @@ class WorktreeControllerTest : BasePlatformTestCase() {
        assertSame(before, controller.model.getElementAt(0))
    }

+    fun `test activity flow updates worktree kind and notifies on EDT`() {
+        val activity = MutableStateFlow<Map<String, SessionActivityDto>>(emptyMap())
+        val controller = controller(activity)
+        val calls = mutableListOf<Boolean>()
+        controller.onActivityChanged = { calls.add(ApplicationManager.getApplication().isDispatchThread) }
+
+        activity.value = mapOf("ses_1" to SessionActivityDto("/repo/wt/", SessionActivityKindDto.RUNNING))
+        flush()
+
+        assertEquals(SessionActivityKind.RUNNING, controller.kind("/repo/wt"))
+        assertEquals(SessionActivityKind.RUNNING, controller.kind("/repo/wt/"))
+        assertTrue(calls.isNotEmpty())
+        assertTrue(calls.all { it })
+    }
+
    fun `test cache notifies on single put and remove but not on bulk sync`() {
        val cache = cache()
        val events = mutableListOf<Pair<String, String?>>()
@@ -347,8 +366,8 @@ class WorktreeControllerTest : BasePlatformTestCase() {
        assertEquals(listOf("/wt" to "Name", "/wt" to null), events)
    }

-    private fun controller() =
-        WorktreeController(service, "/test", coroutines.scope)
+    private fun controller(activity: MutableStateFlow<Map<String, SessionActivityDto>> = MutableStateFlow(emptyMap())) =
+        WorktreeController(service, "/test", coroutines.scope, activity = activity)

    private fun flush() = coroutines.drain(::pump)

@@ -0,0 +1,43 @@
+package ai.kilocode.client.agentManager.worktree
+
+import ai.kilocode.client.session.SessionActivityKind
+import ai.kilocode.rpc.dto.SessionActivityDto
+import ai.kilocode.rpc.dto.SessionActivityKindDto
+import kotlin.test.Test
+import kotlin.test.assertEquals
+
+class WorktreeActivityTest {
+    @Test
+    fun `aggregates multiple sessions by directory with deterministic precedence`() {
+        val result = aggregateWorktreeActivity(mapOf(
+            "ses_run" to SessionActivityDto("/repo/wt", SessionActivityKindDto.RUNNING),
+            "ses_plan" to SessionActivityDto("/repo/wt", SessionActivityKindDto.PLAN),
+            "ses_question" to SessionActivityDto("/repo/wt", SessionActivityKindDto.QUESTION),
+            "ses_permission" to SessionActivityDto("/repo/wt", SessionActivityKindDto.PERMISSION),
+        ))
+
+        assertEquals(SessionActivityKind.PERMISSION, result["/repo/wt"])
+    }
+
+    @Test
+    fun `question beats plan and running while running is used alone`() {
+        val result = aggregateWorktreeActivity(mapOf(
+            "ses_run" to SessionActivityDto("/repo/a", SessionActivityKindDto.RUNNING),
+            "ses_plan" to SessionActivityDto("/repo/b", SessionActivityKindDto.PLAN),
+            "ses_question" to SessionActivityDto("/repo/b", SessionActivityKindDto.QUESTION),
+        ))
+
+        assertEquals(SessionActivityKind.RUNNING, result["/repo/a"])
+        assertEquals(SessionActivityKind.QUESTION, result["/repo/b"])
+    }
+
+    @Test
+    fun `normalizes trailing slashes`() {
+        val result = aggregateWorktreeActivity(mapOf(
+            "ses_1" to SessionActivityDto("/repo/wt/", SessionActivityKindDto.RUNNING),
+        ))
+
+        assertEquals(mapOf("/repo/wt" to SessionActivityKind.RUNNING), result)
+        assertEquals("/repo/wt", normalizeWorktreePath("/repo/wt/"))
+    }
+}
@@ -320,7 +320,7 @@ class WorktreeSessionEditorManagerTest : BasePlatformTestCase() {
                }
            },
            resolve = { workspaces.workspace(it) },
-            status = { sessions.activity() },
+            status = { sessions.activitySnapshot() },
            timers = timers,
            request = { requested += it },
            notify = { title, content -> notified += title to content },
@@ -699,7 +699,7 @@ class SessionSidePanelManagerTest : BasePlatformTestCase() {
                }
            },
            resolve = { workspaces.workspace(it) },
-            status = { sessions.activity() },
+            status = { sessions.activitySnapshot() },
            history = history,
            timers = timers,
            request = request,
@@ -162,7 +162,7 @@ class HistoryControllerTest : BasePlatformTestCase() {
        )
        flush()

-        val activity = sessions.activity()
+        val activity = sessions.activitySnapshot()

        assertEquals(mapOf("ses_busy" to SessionActivityKind.RUNNING), activity)
    }
@@ -269,7 +269,7 @@ class HistoryControllerTest : BasePlatformTestCase() {
            override fun newSession() {}
            override fun showHistory() {}
            override fun openSession(ref: SessionRef) {}
-            override fun activity() = sessions.activity() + kind?.let { mapOf("ses_1" to it) }.orEmpty()
+            override fun activity() = sessions.activitySnapshot() + kind?.let { mapOf("ses_1" to it) }.orEmpty()
        })
        rpc.statuses.value = mapOf("ses_1" to SessionStatusDto("busy"))
        flush()
@@ -286,7 +286,7 @@ class EmptySessionPanelTest : BasePlatformTestCase() {
    fun `test renderer shows overlay badge for active recent session`() {
        val panel = panel(
            recents = listOf(session("ses_1")),
-            activity = { sessions.activity() + mapOf("ses_1" to SessionActivityKind.QUESTION) },
+            activity = { sessions.activitySnapshot() + mapOf("ses_1" to SessionActivityKind.QUESTION) },
        )
        rpc.statuses.value = mapOf("ses_1" to SessionStatusDto("busy"))
        flush()
@@ -301,7 +301,7 @@ class EmptySessionPanelTest : BasePlatformTestCase() {
        var kind: SessionActivityKind? = null
        val panel = panel(
            recents = listOf(session("ses_1")),
-            activity = { sessions.activity() + kind?.let { mapOf("ses_1" to it) }.orEmpty() },
+            activity = { sessions.activitySnapshot() + kind?.let { mapOf("ses_1" to it) }.orEmpty() },
        )
        rpc.statuses.value = mapOf("ses_1" to SessionStatusDto("busy"))
        flush()
@@ -350,7 +350,7 @@ class EmptySessionPanelTest : BasePlatformTestCase() {
    private fun panel(
        recents: List<SessionDto> = emptyList(),
        history: () -> Unit = {},
-        activity: () -> Map<String, SessionActivityKind> = { sessions.activity() },
+        activity: () -> Map<String, SessionActivityKind> = { sessions.activitySnapshot() },
        titles: () -> Map<String, String> = { emptyMap() },
    ) = EmptySessionPanel(testRootDisposable, controller, recents, history, activity, titles)

@@ -15,6 +15,7 @@ import ai.kilocode.rpc.dto.PromptDto
 import ai.kilocode.rpc.dto.QuestionReplyDto
 import ai.kilocode.rpc.dto.QuestionRequestDto
 import ai.kilocode.rpc.dto.SessionDto
+import ai.kilocode.rpc.dto.SessionActivityDto
 import ai.kilocode.rpc.dto.SessionListDto
 import ai.kilocode.rpc.dto.SessionStatusDto
 import ai.kilocode.rpc.dto.SessionTimeDto
@@ -70,6 +71,9 @@ class FakeSessionRpcApi : KiloSessionRpcApi {
    /** Push status updates here. */
    val statuses = MutableStateFlow<Map<String, SessionStatusDto>>(emptyMap())

+    /** Push activity updates here. */
+    val activity = MutableStateFlow<Map<String, SessionActivityDto>>(emptyMap())
+
    /** Pending permissions returned by [pendingPermissions]. */
    val pendingPermissionList = mutableListOf<PermissionRequestDto>()

@@ -189,6 +193,11 @@ class FakeSessionRpcApi : KiloSessionRpcApi {
        return statuses
    }

+    override suspend fun activity(): Flow<Map<String, SessionActivityDto>> {
+        assertNotEdt("activity")
+        return activity
+    }
+
    override suspend fun setDirectory(id: String, directory: String) {
        assertNotEdt("setDirectory")
    }
@@ -13,6 +13,7 @@ import ai.kilocode.rpc.dto.PromptDto
 import ai.kilocode.rpc.dto.QuestionReplyDto
 import ai.kilocode.rpc.dto.QuestionRequestDto
 import ai.kilocode.rpc.dto.SessionDto
+import ai.kilocode.rpc.dto.SessionActivityDto
 import ai.kilocode.rpc.dto.SessionListDto
 import ai.kilocode.rpc.dto.SessionStatusDto
 import com.intellij.platform.rpc.RemoteApiProviderService
@@ -64,6 +65,9 @@ interface KiloSessionRpcApi : RemoteApi<Unit> {
    /** Observe live session status changes. */
    suspend fun statuses(): Flow<Map<String, SessionStatusDto>>

+    /** Observe live per-session activity with the session's directory. */
+    suspend fun activity(): Flow<Map<String, SessionActivityDto>>
+
    /** Register a worktree directory override for a session. */
    suspend fun setDirectory(id: String, directory: String)

@@ -46,6 +46,20 @@ data class SessionStatusDto(
    val requestID: String? = null,
 )

+@Serializable
+enum class SessionActivityKindDto {
+    RUNNING,
+    QUESTION,
+    PLAN,
+    PERMISSION,
+}
+
+@Serializable
+data class SessionActivityDto(
+    val directory: String,
+    val kind: SessionActivityKindDto,
+)
+
@Serializable
 data class SessionListDto(
    val sessions: List<SessionDto>,