opensource/kilocode
3
0
Fork 0
mirror of https://github.com/Kilo-Org/kilocode.git synced 2026-09-21 05:52:35 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix(jetbrains): address review feedback from #8886 and #8849

Browse Source 
- Add ensureActive() in monitorProcess() to prevent stale process
  monitor from racing fresh connections during intentional restarts
- Refactor load() to use local variables, assigning to shared fields
  only after ensureActive() to eliminate NPE race with clear()
- Refactor fetchProfile() to return FetchResult instead of throwing,
  making all fetchers use consistent result types
- Seed session statuses in list() so first load/reconnect gets current
  status instead of empty map
- Use stored worktree directory in session get/delete to fix lookups
  for sessions created in worktrees
- Make KiloBackendCliManager.process @Volatile and update KDoc to
  document that exited() is called off-mutex from IO dispatcher
This commit is contained in:
kirillk
2026-04-14 12:28:00 -04:00
parent b45967633c
commit 017ca5738c
5 changed files with 61 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/kilo-jetbrains/backend/src/main/kotlin/ai/kilocode/backend/app/KiloBackendAppService.kt
+46 -35
View File
@@ -177,31 +177,32 @@ class KiloBackendAppService private constructor(
_appState.value = KiloAppState.Loading(progress.get())
val errors = mutableListOf<LoadError>()
var cfg: Config? = null
var prof: KiloProfile200Response? = null
var notifs: List<KiloNotifications200ResponseInner> = emptyList()
try {
coroutineScope {
launch {
try {
val result = fetchProfile()
progress.updateAndGet { it.copy(profile = result) }
.also { _appState.value = KiloAppState.Loading(it) }
} catch (e: CancellationException) {
throw e
} catch (e: Exception) {
val err = LoadError(
resource = "profile",
status = (e as? ClientException)?.statusCode
?: (e as? ServerException)?.statusCode,
detail = e.message,
)
synchronized(errors) { errors.add(err) }
throw LoadFailure(err)
val result = fetchProfile()
val status = when {
result.error != null -> {
synchronized(errors) { errors.add(result.error) }
throw LoadFailure(result.error)
}
result.value != null -> {
prof = result.value
ProfileResult.LOADED
}
else -> ProfileResult.NOT_LOGGED_IN
}
progress.updateAndGet { it.copy(profile = status) }
.also { _appState.value = KiloAppState.Loading(it) }
}
launch {
val result = fetchWithRetry("config") { fetchConfig() }
if (result.value != null) {
config = result.value
cfg = result.value
progress.updateAndGet { it.copy(config = true) }
.also { _appState.value = KiloAppState.Loading(it) }
} else {
@@ -213,7 +214,7 @@ class KiloBackendAppService private constructor(
launch {
val result = fetchWithRetry("notifications") { fetchNotifications() }
if (result.value != null) {
notifications = result.value
notifs = result.value
progress.updateAndGet { it.copy(notifications = true) }
.also { _appState.value = KiloAppState.Loading(it) }
} else {
@@ -224,18 +225,21 @@ class KiloBackendAppService private constructor(
}
}
ensureActive()
sessions.start(connection.api!!, connection.events)
workspaces.start(connection.api!!, connection.events)
_appState.value = KiloAppState.Ready(
AppData(
profile = profile,
config = config!!,
notifications = notifications,
ensureActive()
profile = prof
config = cfg
notifications = notifs
sessions.start(connection.api!!, connection.events)
workspaces.start(connection.api!!, connection.events)
_appState.value = KiloAppState.Ready(
AppData(
profile = prof,
config = cfg!!,
notifications = notifs,
)
)
)
log.info("Application started — config, profile, notifications loaded")
startWatchingGlobalSseEvents()
log.info("Application started — config, profile, notifications loaded")
startWatchingGlobalSseEvents()
} catch (e: CancellationException) {
throw e
} catch (e: Exception) {
@@ -249,23 +253,30 @@ class KiloBackendAppService private constructor(
}
}
private suspend fun fetchProfile(): ProfileResult {
val client = connection.api ?: return ProfileResult.NOT_LOGGED_IN
/**
* Fetch the user profile. Returns [FetchResult.ok] with the response
* on success, [FetchResult.ok] with `null` when not logged in (401),
* or [FetchResult.fail] on other errors. Never throws.
*/
private suspend fun fetchProfile(): FetchResult<KiloProfile200Response?> {
val client = connection.api
?: return FetchResult.ok(null)
return try {
val response = client.kiloProfile()
profile = response
log.info("Profile: ${response.profile.email}")
ProfileResult.LOADED
FetchResult.ok(response)
} catch (e: ClientException) {
if (e.statusCode == 401) {
log.info("Profile: not logged in (401)")
return ProfileResult.NOT_LOGGED_IN
return FetchResult.ok(null)
}
log.warn("Profile fetch failed: HTTP ${e.statusCode}", e)
throw e
logResponseBody("profile", e)
FetchResult.fail("profile", e)
} catch (e: Exception) {
log.warn("Profile fetch failed: ${e.message}", e)
throw e
logResponseBody("profile", e)
FetchResult.fail("profile", e)
}
}
packages/kilo-jetbrains/backend/src/main/kotlin/ai/kilocode/backend/app/KiloBackendCliManager.kt
+4 -1
View File
@@ -25,7 +25,9 @@ import java.util.concurrent.TimeUnit
* spawns `kilo serve --port 0`, and exposes the result as [State].
*
* Concurrency is handled by the owning [KiloBackendAppService] — all public
* methods are called under its mutex so no internal synchronization is needed.
* methods except [exited] are called under its mutex. [exited] is called from
* [KiloConnectionService]'s IO dispatcher and is thread-safe via the stale-ref
* guard and volatile [process] field.
*/
class KiloBackendCliManager(
private val log: KiloLog = IntellijLog(KiloBackendCliManager::class.java),
@@ -37,6 +39,7 @@ class KiloBackendCliManager(
private val PORT_REGEX = Regex("""listening on http://[\w.]+:(\d+)""")
}
@Volatile
private var process: Process? = null
private var hook: Thread? = null
packages/kilo-jetbrains/backend/src/main/kotlin/ai/kilocode/backend/app/KiloBackendConnectionService.kt
+2
View File
@@ -6,6 +6,7 @@ import ai.kilocode.backend.util.KiloLog
import ai.kilocode.jetbrains.api.client.DefaultApi
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.ensureActive
import kotlinx.coroutines.Job
import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.MutableSharedFlow
@@ -304,6 +305,7 @@ class KiloConnectionService(
private fun monitorProcess(proc: Process) = cs.launch(Dispatchers.IO) {
proc.waitFor()
ensureActive()
server.exited(proc)
val code = proc.exitValue()
log.warn("CLI process exited with code $code")
packages/kilo-jetbrains/backend/src/main/kotlin/ai/kilocode/backend/app/KiloBackendSessionManager.kt
+1
View File
@@ -97,6 +97,7 @@ class KiloBackendSessionManager(
/** List root sessions for a directory and include current statuses. */
fun list(dir: String): SessionListDto {
seed(dir)
val raw = requireClient().sessionList(directory = dir, roots = true)
val mapped = raw.map(::dto)
val ids = mapped.map { it.id }.toSet()
packages/kilo-jetbrains/backend/src/main/kotlin/ai/kilocode/backend/rpc/KiloSessionRpcApiImpl.kt
+8 -4
View File
@@ -34,11 +34,15 @@ class KiloSessionRpcApiImpl : KiloSessionRpcApi {
override suspend fun create(directory: String): SessionDto =
workspaces.get(directory).createSession()
override suspend fun get(id: String, directory: String): SessionDto =
sessions.get(id, directory)
override suspend fun get(id: String, directory: String): SessionDto {
val dir = sessions.getDirectory(id, directory)
return sessions.get(id, dir)
}
override suspend fun delete(id: String, directory: String) =
workspaces.get(directory).deleteSession(id)
override suspend fun delete(id: String, directory: String) {
val dir = sessions.getDirectory(id, directory)
workspaces.get(dir).deleteSession(id)
}
override suspend fun statuses(): Flow<Map<String, SessionStatusDto>> =
sessions.statuses