feat: 初始化 Android 客户端（M0 骨架 + M1 登录 + M2 基础模块）
M0：Gradle 9.7.1 / AGP 9.2.1 / Kotlin 2.4 / Miuix 0.9.3，minSdk 33，compileSdk 37。 M1：登录页（服务器地址可配、服务端/本地错误分流、EncryptedSharedPreferences、自动登录与静默重登、401 登出）。 M2 基础模块：消息内容解析器（<p>/<action>/<narration> + 流式增量）、WebSocket 客户端、REST 数据层与分页去重。 45 个单元测试全绿，真机验证通过（登录页 + 启动图标）。 devdocs/ 收录设计文档、前后端协同事项与细化方案。
This commit is contained in:
AskaEth committed
commit
84de51c7b7
67 files changed
+3960
No files matched your search
@@ -0,0 +1,81 @@
|
||||
package top.yeij.cyrene.data.api
|
||||
|
||||
import top.yeij.cyrene.BuildConfig
|
||||
import kotlinx.serialization.json.Json
|
||||
import okhttp3.Interceptor
|
||||
import okhttp3.MediaType.Companion.toMediaType
|
||||
import okhttp3.OkHttpClient
|
||||
import okhttp3.Response
|
||||
import okhttp3.logging.HttpLoggingInterceptor
|
||||
import retrofit2.Retrofit
|
||||
import retrofit2.converter.kotlinx.serialization.asConverterFactory
|
||||
import java.util.concurrent.TimeUnit
|
||||
|
||||
object ApiClientFactory {
|
||||
|
||||
val json: Json = Json {
|
||||
ignoreUnknownKeys = true
|
||||
explicitNulls = false
|
||||
coerceInputValues = true
|
||||
}
|
||||
|
||||
fun create(
|
||||
baseUrl: String,
|
||||
tokenProvider: () -> String? = { null },
|
||||
onUnauthorized: () -> Unit = {},
|
||||
): CyreneApi {
|
||||
val client = OkHttpClient.Builder()
|
||||
.connectTimeout(10, TimeUnit.SECONDS)
|
||||
.readTimeout(60, TimeUnit.SECONDS)
|
||||
.writeTimeout(60, TimeUnit.SECONDS)
|
||||
.addInterceptor(HeaderInterceptor(tokenProvider, onUnauthorized))
|
||||
.apply {
|
||||
if (BuildConfig.DEBUG) {
|
||||
addInterceptor(
|
||||
HttpLoggingInterceptor().apply {
|
||||
level = HttpLoggingInterceptor.Level.BASIC
|
||||
redactHeader("Authorization")
|
||||
},
|
||||
)
|
||||
}
|
||||
}
|
||||
.build()
|
||||
|
||||
return Retrofit.Builder()
|
||||
.baseUrl(normalizeBaseUrl(baseUrl))
|
||||
.client(client)
|
||||
.addConverterFactory(json.asConverterFactory("application/json".toMediaType()))
|
||||
.build()
|
||||
.create(CyreneApi::class.java)
|
||||
}
|
||||
|
||||
/** Retrofit 要求 baseUrl 以 `/` 结尾；用户没写 scheme 时补 `http://`。 */
|
||||
fun normalizeBaseUrl(raw: String): String {
|
||||
var url = raw.trim()
|
||||
if (url.isEmpty()) return url
|
||||
if (!url.startsWith("http://") && !url.startsWith("https://")) {
|
||||
url = "http://$url"
|
||||
}
|
||||
if (!url.endsWith("/")) url += "/"
|
||||
return url
|
||||
}
|
||||
|
||||
private class HeaderInterceptor(
|
||||
private val tokenProvider: () -> String?,
|
||||
private val onUnauthorized: () -> Unit,
|
||||
) : Interceptor {
|
||||
override fun intercept(chain: Interceptor.Chain): Response {
|
||||
val token = tokenProvider()
|
||||
val request = chain.request().newBuilder()
|
||||
.header("X-API-Version", "1")
|
||||
.apply { if (token != null) header("Authorization", "Bearer $token") }
|
||||
.build()
|
||||
val response = chain.proceed(request)
|
||||
// 只有「带了 token 仍 401」才算会话过期；登录请求不带 token，不受影响。
|
||||
if (response.code == 401 && token != null) {
|
||||
onUnauthorized()
|
||||
}
|
||||
return response
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,21 @@
|
||||
package top.yeij.cyrene.data.api
|
||||
|
||||
import kotlinx.serialization.Serializable
|
||||
|
||||
/** 后端统一信封：成功 `{data}`，失败 `{error:{code,message}}`。 */
|
||||
@Serializable
|
||||
data class ApiEnvelope<T>(
|
||||
val data: T? = null,
|
||||
val error: ApiError? = null,
|
||||
)
|
||||
|
||||
@Serializable
|
||||
data class ApiErrorEnvelope(
|
||||
val error: ApiError? = null,
|
||||
)
|
||||
|
||||
@Serializable
|
||||
data class ApiError(
|
||||
val code: String = "",
|
||||
val message: String = "",
|
||||
)
|
||||
@@ -0,0 +1,76 @@
|
||||
package top.yeij.cyrene.data.api
|
||||
|
||||
import top.yeij.cyrene.core.error.CyreneError
|
||||
import kotlinx.coroutines.CancellationException
|
||||
import kotlinx.serialization.SerializationException
|
||||
import retrofit2.Response
|
||||
import java.io.IOException
|
||||
import java.net.ConnectException
|
||||
import java.net.SocketTimeoutException
|
||||
import java.net.UnknownHostException
|
||||
import javax.net.ssl.SSLException
|
||||
|
||||
/**
|
||||
* 统一错误映射（S1.1）：已建连接用服务端 `error`，本地异常按类型分类。
|
||||
* 供所有 REST 仓库复用；不要在各个仓库里各写一份。
|
||||
*/
|
||||
object ApiErrorMapper {
|
||||
|
||||
/** 有 HTTP 响应：401 → [CyreneError.Unauthorized]，其余 → [CyreneError.Server]。 */
|
||||
fun fromHttp(httpStatus: Int, errorBody: String?): CyreneError {
|
||||
val error = parseErrorEnvelope(errorBody)
|
||||
val code = error?.code.orEmpty()
|
||||
val message = error?.message.orEmpty()
|
||||
return if (httpStatus == 401) {
|
||||
CyreneError.Unauthorized(code = code, message = message)
|
||||
} else {
|
||||
CyreneError.Server(code = code, message = message, httpStatus = httpStatus)
|
||||
}
|
||||
}
|
||||
|
||||
/** 本地/网络异常 → [CyreneError.Local]。 */
|
||||
fun fromThrowable(throwable: Throwable): CyreneError = when (throwable) {
|
||||
is UnknownHostException -> CyreneError.Local(CyreneError.LocalKind.Dns)
|
||||
is SSLException -> CyreneError.Local(CyreneError.LocalKind.Tls)
|
||||
is SocketTimeoutException -> CyreneError.Local(CyreneError.LocalKind.Timeout)
|
||||
is ConnectException -> CyreneError.Local(CyreneError.LocalKind.ConnectFailed)
|
||||
is SerializationException -> CyreneError.Local(CyreneError.LocalKind.Parse)
|
||||
is IOException -> CyreneError.Local(CyreneError.LocalKind.NoNetwork)
|
||||
else -> CyreneError.Local(CyreneError.LocalKind.Unknown)
|
||||
}
|
||||
|
||||
/** 解析 `{"error":{code,message}}`；非信封 / 空白 / 解析失败均返回 null。 */
|
||||
fun parseErrorEnvelope(errorBody: String?): ApiError? = errorBody
|
||||
?.takeIf { it.isNotBlank() }
|
||||
?.let {
|
||||
runCatching {
|
||||
ApiClientFactory.json.decodeFromString(ApiErrorEnvelope.serializer(), it).error
|
||||
}.getOrNull()
|
||||
}
|
||||
}
|
||||
|
||||
/** 把 [CyreneError] 塞进 Kotlin `Result` 的载体，便于 `Result<T>` 统一传递失败。 */
|
||||
class CyreneException(val error: CyreneError) : Exception(error.userMessage)
|
||||
|
||||
/**
|
||||
* 执行一次信封调用并把结果收敛成 `Result<T>`：
|
||||
* 成功且 `data != null` → success；成功但 `data == null` → Parse；否则走 [ApiErrorMapper]。
|
||||
*/
|
||||
suspend fun <T : Any> apiCall(block: suspend () -> Response<ApiEnvelope<T>>): Result<T> =
|
||||
try {
|
||||
val response = block()
|
||||
val data = response.body()?.data
|
||||
when {
|
||||
response.isSuccessful && data != null -> Result.success(data)
|
||||
response.isSuccessful -> Result.failure(CyreneException(CyreneError.Local(CyreneError.LocalKind.Parse)))
|
||||
else -> Result.failure(
|
||||
CyreneException(
|
||||
ApiErrorMapper.fromHttp(response.code(), response.errorBody()?.string()),
|
||||
),
|
||||
)
|
||||
}
|
||||
} catch (e: CancellationException) {
|
||||
throw e
|
||||
} catch (e: Exception) {
|
||||
Result.failure(CyreneException(ApiErrorMapper.fromThrowable(e)))
|
||||
}
|
||||
@@ -0,0 +1,13 @@
|
||||
package top.yeij.cyrene.data.api
|
||||
|
||||
import top.yeij.cyrene.data.api.dto.LoginRequest
|
||||
import top.yeij.cyrene.data.api.dto.LoginResponse
|
||||
import retrofit2.Response
|
||||
import retrofit2.http.Body
|
||||
import retrofit2.http.POST
|
||||
|
||||
interface CyreneApi {
|
||||
|
||||
@POST("api/v1/auth/login")
|
||||
suspend fun login(@Body body: LoginRequest): Response<ApiEnvelope<LoginResponse>>
|
||||
}
|
||||
@@ -0,0 +1,44 @@
|
||||
package top.yeij.cyrene.data.api
|
||||
|
||||
import top.yeij.cyrene.data.api.dto.CreateSessionResponse
|
||||
import top.yeij.cyrene.data.api.dto.DeleteSessionResponse
|
||||
import top.yeij.cyrene.data.api.dto.HistoryMessageDto
|
||||
import top.yeij.cyrene.data.api.dto.SessionSummaryDto
|
||||
import retrofit2.Response
|
||||
import retrofit2.http.DELETE
|
||||
import retrofit2.http.GET
|
||||
import retrofit2.http.POST
|
||||
import retrofit2.http.Path
|
||||
import retrofit2.http.Query
|
||||
|
||||
/**
|
||||
* 会话与消息相关端点（api-reference §3.2/§3.4/§3.5/§3.6）。
|
||||
* 与 M1 的 `CyreneApi` 分开，避免互相污染。
|
||||
*/
|
||||
interface SessionsApi {
|
||||
|
||||
@GET("api/v1/sessions")
|
||||
suspend fun listSessions(): Response<ApiEnvelope<List<SessionSummaryDto>>>
|
||||
|
||||
/** 无 body；服务端返回 `201 {data:{sessionId}}`。 */
|
||||
@POST("api/v1/sessions")
|
||||
suspend fun createSession(): Response<ApiEnvelope<CreateSessionResponse>>
|
||||
|
||||
/** 主会话 `web` 删除会返回 `403 forbidden`。 */
|
||||
@DELETE("api/v1/sessions/{id}")
|
||||
suspend fun deleteSession(@Path("id") sessionId: String): Response<ApiEnvelope<DeleteSessionResponse>>
|
||||
|
||||
/**
|
||||
* 消息热分页，服务端按 `createdAt` 倒序（newest first）返回。
|
||||
* @param limit 默认 50、上限 200（由 [top.yeij.cyrene.data.repository.MessageRepository] 钳制）
|
||||
* @param before 上一页最旧消息的 `createdAt`（Unix 毫秒）；null = 最新一页
|
||||
* @param platform 来源平台过滤；null = 全量
|
||||
*/
|
||||
@GET("api/v1/sessions/{id}/messages")
|
||||
suspend fun listMessages(
|
||||
@Path("id") sessionId: String,
|
||||
@Query("limit") limit: Int? = null,
|
||||
@Query("before") before: Long? = null,
|
||||
@Query("platform") platform: String? = null,
|
||||
): Response<ApiEnvelope<List<HistoryMessageDto>>>
|
||||
}
|
||||
@@ -0,0 +1,16 @@
|
||||
package top.yeij.cyrene.data.api.dto
|
||||
|
||||
import kotlinx.serialization.Serializable
|
||||
|
||||
@Serializable
|
||||
data class LoginRequest(
|
||||
val username: String,
|
||||
val password: String,
|
||||
)
|
||||
|
||||
@Serializable
|
||||
data class LoginResponse(
|
||||
val token: String,
|
||||
val expiresAt: Long,
|
||||
val role: String,
|
||||
)
|
||||
@@ -0,0 +1,24 @@
|
||||
package top.yeij.cyrene.data.api.dto
|
||||
|
||||
import kotlinx.serialization.Serializable
|
||||
|
||||
/**
|
||||
* `GET /api/v1/sessions/{id}/messages` 单条消息（api-reference §3.4）。
|
||||
* 服务端按 `createdAt` 倒序返回，渲染前需 reverse / merge。
|
||||
*/
|
||||
@Serializable
|
||||
data class HistoryMessageDto(
|
||||
val id: String,
|
||||
val sessionId: String,
|
||||
val role: String,
|
||||
val content: String = "",
|
||||
val mode: String? = null,
|
||||
val mood: String? = null,
|
||||
val image: String? = null,
|
||||
val thinking: String? = null,
|
||||
val platform: String? = null,
|
||||
val deviceId: String? = null,
|
||||
val senderName: String? = null,
|
||||
val adapterId: String? = null,
|
||||
val createdAt: Long,
|
||||
)
|
||||
@@ -0,0 +1,36 @@
|
||||
package top.yeij.cyrene.data.api.dto
|
||||
|
||||
import kotlinx.serialization.Serializable
|
||||
|
||||
/**
|
||||
* `GET /api/v1/sessions` 持久化会话摘要（api-reference §3.2 A）。
|
||||
* 主会话 `id` 恒为 `"web"`、`type = "main"`，且由服务端排在首位。
|
||||
*/
|
||||
@Serializable
|
||||
data class SessionSummaryDto(
|
||||
val id: String,
|
||||
val type: String,
|
||||
val title: String? = null,
|
||||
val createdAt: Long,
|
||||
) {
|
||||
val isMain: Boolean
|
||||
get() = type == TYPE_MAIN || id == MAIN_SESSION_ID
|
||||
|
||||
companion object {
|
||||
const val MAIN_SESSION_ID = "web"
|
||||
const val TYPE_MAIN = "main"
|
||||
const val TYPE_SUB = "sub"
|
||||
}
|
||||
}
|
||||
|
||||
/** `POST /api/v1/sessions` → `201 {"data":{"sessionId":"web:<uuid>"}}`。 */
|
||||
@Serializable
|
||||
data class CreateSessionResponse(
|
||||
val sessionId: String,
|
||||
)
|
||||
|
||||
/** `DELETE /api/v1/sessions/{id}` → `200 {"data":{"deleted":"<id>"}}`。 */
|
||||
@Serializable
|
||||
data class DeleteSessionResponse(
|
||||
val deleted: String,
|
||||
)
|
||||
@@ -0,0 +1,61 @@
|
||||
package top.yeij.cyrene.data.repository
|
||||
|
||||
import top.yeij.cyrene.data.api.SessionsApi
|
||||
import top.yeij.cyrene.data.api.apiCall
|
||||
import top.yeij.cyrene.data.api.dto.HistoryMessageDto
|
||||
import kotlinx.coroutines.CoroutineDispatcher
|
||||
import kotlinx.coroutines.Dispatchers
|
||||
import kotlinx.coroutines.withContext
|
||||
|
||||
/**
|
||||
* 会话消息分页获取层。服务端按倒序返回，这里统一转成升序的 [PagedMessages]。
|
||||
* 去重 / 合并请用 [mergeMessages]。
|
||||
*/
|
||||
class MessageRepository(
|
||||
private val api: SessionsApi,
|
||||
private val dispatcher: CoroutineDispatcher = Dispatchers.IO,
|
||||
) {
|
||||
|
||||
/**
|
||||
* 拉一页消息。
|
||||
*
|
||||
* @param before 上一页最旧消息的 `createdAt`（用 [PagedMessages.oldestCreatedAt]）；
|
||||
* null = 最新一页
|
||||
* @param platform 来源平台过滤；null = 全量（Web 端共享主会话语义）
|
||||
*/
|
||||
suspend fun loadPage(
|
||||
sessionId: String,
|
||||
limit: Int = DEFAULT_PAGE_SIZE,
|
||||
before: Long? = null,
|
||||
platform: String? = null,
|
||||
): Result<PagedMessages> = withContext(dispatcher) {
|
||||
val pageSize = limit.coerceIn(1, MAX_PAGE_SIZE)
|
||||
apiCall { api.listMessages(sessionId, pageSize, before, platform) }
|
||||
.map { raw -> toPagedMessages(raw, pageSize) }
|
||||
}
|
||||
|
||||
/** 首屏 / 刷新：最新一页。 */
|
||||
suspend fun loadLatest(
|
||||
sessionId: String,
|
||||
limit: Int = DEFAULT_PAGE_SIZE,
|
||||
platform: String? = null,
|
||||
): Result<PagedMessages> = loadPage(sessionId, limit, before = null, platform = platform)
|
||||
|
||||
companion object {
|
||||
const val DEFAULT_PAGE_SIZE = 50
|
||||
const val MAX_PAGE_SIZE = 200
|
||||
}
|
||||
}
|
||||
|
||||
private fun toPagedMessages(
|
||||
rawNewestFirst: List<HistoryMessageDto>,
|
||||
pageSize: Int,
|
||||
): PagedMessages {
|
||||
val ascending = rawNewestFirst.sortedWith(MessageOrder)
|
||||
return PagedMessages(
|
||||
messages = ascending,
|
||||
hasMore = rawNewestFirst.size >= pageSize,
|
||||
oldestCreatedAt = ascending.firstOrNull()?.createdAt,
|
||||
newestCreatedAt = ascending.lastOrNull()?.createdAt,
|
||||
)
|
||||
}
|
||||
@@ -0,0 +1,40 @@
|
||||
package top.yeij.cyrene.data.repository
|
||||
|
||||
import top.yeij.cyrene.data.api.dto.HistoryMessageDto
|
||||
|
||||
/**
|
||||
* 一页消息（已由仓库转成 **升序**，可直接追加渲染）。
|
||||
*
|
||||
* @param hasMore 该页是否可能还有更旧的消息（`原始页大小 == 请求页大小` 的保守判断）
|
||||
* @param oldestCreatedAt 本页最旧一条的 `createdAt`，作为下一页的 `before` 游标
|
||||
* @param newestCreatedAt 本页最新一条的 `createdAt`
|
||||
*/
|
||||
data class PagedMessages(
|
||||
val messages: List<HistoryMessageDto>,
|
||||
val hasMore: Boolean,
|
||||
val oldestCreatedAt: Long?,
|
||||
val newestCreatedAt: Long?,
|
||||
)
|
||||
|
||||
/** 消息排序：`createdAt` 升序，同毫秒用 `id` 兜底，保证稳定可复现。 */
|
||||
internal val MessageOrder: Comparator<HistoryMessageDto> = compareBy<HistoryMessageDto>({ it.createdAt }, { it.id })
|
||||
|
||||
/**
|
||||
* 纯函数合并：`existing`（本地已加载，任意序）+ `incoming`（服务端新到的一页，倒序）
|
||||
* → 按 `id` 去重、按 `createdAt` 升序（同毫秒用 `id` 兜底）的结果。
|
||||
*
|
||||
* 冲突时 **incoming 胜出**（§3.2「服务端优先」）。不修改入参，返回新列表。
|
||||
* 时间复杂度 O((n+m) log(n+m))，空间 O(n+m)。
|
||||
*/
|
||||
fun mergeMessages(
|
||||
existing: List<HistoryMessageDto>,
|
||||
incoming: List<HistoryMessageDto>,
|
||||
): List<HistoryMessageDto> {
|
||||
if (incoming.isEmpty()) return existing.sortedWith(MessageOrder)
|
||||
if (existing.isEmpty()) return incoming.sortedWith(MessageOrder)
|
||||
|
||||
val byId = LinkedHashMap<String, HistoryMessageDto>(existing.size + incoming.size)
|
||||
existing.forEach { byId[it.id] = it }
|
||||
incoming.forEach { byId[it.id] = it }
|
||||
return byId.values.sortedWith(MessageOrder)
|
||||
}
|
||||
@@ -0,0 +1,35 @@
|
||||
package top.yeij.cyrene.data.repository
|
||||
|
||||
import top.yeij.cyrene.data.api.SessionsApi
|
||||
import top.yeij.cyrene.data.api.apiCall
|
||||
import top.yeij.cyrene.data.api.dto.SessionSummaryDto
|
||||
import kotlinx.coroutines.CoroutineDispatcher
|
||||
import kotlinx.coroutines.Dispatchers
|
||||
import kotlinx.coroutines.withContext
|
||||
|
||||
/**
|
||||
* 会话列表 / 新建 / 删除的数据获取层。
|
||||
*
|
||||
* 列表策略（§3.1）：调用方拿到结果后**整体替换**本地快照，不做 merge；
|
||||
* 失败时调用方保留旧快照。仓库本身不持有任何持久化状态。
|
||||
*/
|
||||
class SessionRepository(
|
||||
private val api: SessionsApi,
|
||||
private val dispatcher: CoroutineDispatcher = Dispatchers.IO,
|
||||
) {
|
||||
|
||||
/** `GET /sessions`，原样返回服务端顺序（主会话恒首位）。 */
|
||||
suspend fun listSessions(): Result<List<SessionSummaryDto>> = withContext(dispatcher) {
|
||||
apiCall { api.listSessions() }
|
||||
}
|
||||
|
||||
/** `POST /sessions`，成功返回新建的 `sessionId`（形如 `web:<uuid>`）。 */
|
||||
suspend fun createSession(): Result<String> = withContext(dispatcher) {
|
||||
apiCall { api.createSession() }.map { it.sessionId }
|
||||
}
|
||||
|
||||
/** `DELETE /sessions/{id}`，成功返回被删除的会话 id；主会话会以 403 失败。 */
|
||||
suspend fun deleteSession(sessionId: String): Result<String> = withContext(dispatcher) {
|
||||
apiCall { api.deleteSession(sessionId) }.map { it.deleted }
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,254 @@
|
||||
package top.yeij.cyrene.data.ws
|
||||
|
||||
import top.yeij.cyrene.data.api.ApiClientFactory
|
||||
import kotlinx.coroutines.CancellationException
|
||||
import kotlinx.coroutines.CompletableDeferred
|
||||
import kotlinx.coroutines.CoroutineScope
|
||||
import kotlinx.coroutines.Dispatchers
|
||||
import kotlinx.coroutines.Job
|
||||
import kotlinx.coroutines.SupervisorJob
|
||||
import kotlinx.coroutines.channels.BufferOverflow
|
||||
import kotlinx.coroutines.cancel
|
||||
import kotlinx.coroutines.delay
|
||||
import kotlinx.coroutines.flow.MutableSharedFlow
|
||||
import kotlinx.coroutines.flow.MutableStateFlow
|
||||
import kotlinx.coroutines.flow.SharedFlow
|
||||
import kotlinx.coroutines.flow.StateFlow
|
||||
import kotlinx.coroutines.flow.asSharedFlow
|
||||
import kotlinx.coroutines.flow.asStateFlow
|
||||
import kotlinx.coroutines.launch
|
||||
import okhttp3.HttpUrl
|
||||
import okhttp3.HttpUrl.Companion.toHttpUrlOrNull
|
||||
import okhttp3.OkHttpClient
|
||||
import okhttp3.Request
|
||||
import okhttp3.Response
|
||||
import okhttp3.WebSocket
|
||||
import okhttp3.WebSocketListener
|
||||
import java.util.concurrent.TimeUnit
|
||||
import kotlin.random.Random
|
||||
|
||||
/**
|
||||
* Cyrene 实时通道客户端（`GET /ws/chat?token=<JWT>`，api-reference §6）。
|
||||
*
|
||||
* ## 用法
|
||||
* ```
|
||||
* val socket = ChatSocket(baseUrlProvider = { cfg.baseUrl }, tokenProvider = { auth.token },
|
||||
* onAuthFailed = { auth.logout() })
|
||||
* socket.connect()
|
||||
* // 收集 socket.events，观察 socket.state
|
||||
* socket.sendChat(sessionId = "web", text = "你好", deviceId = deviceId)
|
||||
* socket.close()
|
||||
* ```
|
||||
*
|
||||
* ## 会话归属约束
|
||||
* `message.*` / `thinking.*` / `tool_call.*` / `stream_chunk` 等流式帧**不带 `sessionId`**，
|
||||
* 本类不会替它们编造会话。上层必须把它们归属到「当前唯一的 active 流式会话」——
|
||||
* 也就是**同一时刻只允许一个会话在流式**（切会话时需先结束或 `interrupt`）。
|
||||
*
|
||||
* ## 心跳与重连
|
||||
* - OkHttp `pingInterval` 30s（服务端 ping 周期 54s / pong 等待 60s，客户端 30s 足够）。
|
||||
* - 服务端**没有 resume / 补发**：重连成功后由上层用新连接的 `history` 帧 + REST 补差去重。
|
||||
* - 退避 1s/2s/4s/8s/16s/30s（上限 30s，含抖动）。
|
||||
* - 握手返回 **HTTP 401** → [WsState.Offline]，触发 [onAuthFailed]，**不再重连**（需重新登录）。
|
||||
*
|
||||
* ## 坏帧
|
||||
* 每帧解析都走 [WsEvent.parse]，非 JSON / 未知 `type` / 结构不匹配一律丢弃，不会中断连接。
|
||||
*
|
||||
* @param baseUrlProvider 服务器根地址（如 `http://host:8080`），每次重连时重新读取。
|
||||
* @param tokenProvider 当前 JWT，每次重连时重新读取（支持静默重登后复用同一实例）。
|
||||
* @param onAuthFailed 握手 401 回调；在协程内调用，注意切主线程。
|
||||
* @param client 可注入自定义 OkHttpClient（会 `newBuilder` 派生，保留其拦截器）；缺省新建。
|
||||
*/
|
||||
class ChatSocket(
|
||||
private val baseUrlProvider: () -> String?,
|
||||
private val tokenProvider: () -> String?,
|
||||
private val onAuthFailed: () -> Unit = {},
|
||||
client: OkHttpClient? = null,
|
||||
) {
|
||||
|
||||
private val _state = MutableStateFlow(WsState.Idle)
|
||||
|
||||
/** 连接状态机；见 [WsState]。 */
|
||||
val state: StateFlow<WsState> = _state.asStateFlow()
|
||||
|
||||
private val _events = MutableSharedFlow<WsEvent>(
|
||||
replay = 0,
|
||||
extraBufferCapacity = EVENT_BUFFER,
|
||||
onBufferOverflow = BufferOverflow.DROP_OLDEST,
|
||||
)
|
||||
|
||||
/**
|
||||
* 下行事件流。
|
||||
*
|
||||
* `replay = 0`：重放旧 delta 会让新订阅者重复渲染，会话状态应由 `history` / REST 重建。
|
||||
* `extraBufferCapacity = 512 + DROP_OLDEST`：发送方是 OkHttp 读线程，不能阻塞；溢出时丢最旧的
|
||||
* 增量帧（服务端发送缓冲 32 本就会丢帧，`stream_end` / `history` 仍可兜底）。
|
||||
*/
|
||||
val events: SharedFlow<WsEvent> = _events.asSharedFlow()
|
||||
|
||||
private val wsClient: OkHttpClient = (client ?: OkHttpClient()).newBuilder()
|
||||
.pingInterval(PING_INTERVAL_SECONDS, TimeUnit.SECONDS)
|
||||
.readTimeout(0, TimeUnit.MILLISECONDS)
|
||||
.connectTimeout(CONNECT_TIMEOUT_SECONDS, TimeUnit.SECONDS)
|
||||
.build()
|
||||
|
||||
@Volatile
|
||||
private var wantConnection: Boolean = false
|
||||
|
||||
@Volatile
|
||||
private var webSocket: WebSocket? = null
|
||||
|
||||
@Volatile
|
||||
private var attempt: Int = 0
|
||||
|
||||
@Volatile
|
||||
private var sessionJob: Job? = null
|
||||
|
||||
private var scope: CoroutineScope? = null
|
||||
|
||||
/** 开始连接（幂等：已在连接 / 重连中时忽略）。 */
|
||||
fun connect() {
|
||||
if (wantConnection) return
|
||||
wantConnection = true
|
||||
attempt = 0
|
||||
sessionJob = ensureScope().launch { runConnectionLoop() }
|
||||
}
|
||||
|
||||
/** 主动断开并停掉重连；之后可再次 [connect]。 */
|
||||
fun close() {
|
||||
wantConnection = false
|
||||
val socket = webSocket
|
||||
webSocket = null
|
||||
socket?.close(CLOSE_CODE_NORMAL, CLOSE_REASON_CLIENT)
|
||||
sessionJob?.cancel()
|
||||
sessionJob = null
|
||||
_state.value = WsState.Idle
|
||||
scope?.cancel()
|
||||
scope = null
|
||||
}
|
||||
|
||||
/**
|
||||
* 发送 `chat`（api-reference §6.3）。
|
||||
* @return 未连接或写失败返回 `false`（本层不做离线队列，由上层标红重试）。
|
||||
*/
|
||||
fun sendChat(sessionId: String?, text: String, deviceId: String?): Boolean =
|
||||
send(WsOutbound.Chat(text = text, sessionId = sessionId, deviceId = deviceId))
|
||||
|
||||
/** 发送 `interrupt`（取消在跑 turn + 清空 pending）。 */
|
||||
fun sendInterrupt(sessionId: String): Boolean = send(WsOutbound.Interrupt(sessionId))
|
||||
|
||||
private fun send(outbound: WsOutbound): Boolean {
|
||||
if (_state.value != WsState.Connected) return false
|
||||
val socket = webSocket ?: return false
|
||||
return runCatching { socket.send(outbound.toJson()) }.getOrDefault(false)
|
||||
}
|
||||
|
||||
private suspend fun runConnectionLoop() {
|
||||
try {
|
||||
while (wantConnection) {
|
||||
when (val end = runSession()) {
|
||||
SessionEnd.ConfigMissing -> {
|
||||
wantConnection = false
|
||||
_state.value = WsState.Offline
|
||||
}
|
||||
|
||||
SessionEnd.AuthFailed -> {
|
||||
wantConnection = false
|
||||
_state.value = WsState.Offline
|
||||
onAuthFailed()
|
||||
}
|
||||
|
||||
SessionEnd.Closed -> {
|
||||
if (!wantConnection) break
|
||||
_state.value = WsState.Reconnecting
|
||||
delay(backoffDelayMs(attempt))
|
||||
attempt += 1
|
||||
}
|
||||
}
|
||||
}
|
||||
} catch (_: CancellationException) {
|
||||
// close() 主动取消，正常退出。
|
||||
}
|
||||
}
|
||||
|
||||
/** 建立一次连接并挂起，直到它关闭 / 失败 / 被取消。 */
|
||||
private suspend fun runSession(): SessionEnd {
|
||||
val url = buildUrl() ?: return SessionEnd.ConfigMissing
|
||||
_state.value = if (attempt == 0) WsState.Connecting else WsState.Reconnecting
|
||||
val finished = CompletableDeferred<SessionEnd>()
|
||||
val socket = wsClient.newWebSocket(
|
||||
Request.Builder().url(url).build(),
|
||||
object : WebSocketListener() {
|
||||
override fun onOpen(webSocket: WebSocket, response: Response) {
|
||||
attempt = 0
|
||||
_state.value = WsState.Connected
|
||||
}
|
||||
|
||||
override fun onMessage(webSocket: WebSocket, text: String) {
|
||||
// 未知 type / 坏帧由 parse 返回 null，静默丢弃。
|
||||
WsEvent.parse(text)?.let { _events.tryEmit(it) }
|
||||
}
|
||||
|
||||
override fun onClosed(webSocket: WebSocket, code: Int, reason: String) {
|
||||
finished.complete(SessionEnd.Closed)
|
||||
}
|
||||
|
||||
override fun onFailure(webSocket: WebSocket, t: Throwable, response: Response?) {
|
||||
val end = if (isAuthFailure(response, t)) SessionEnd.AuthFailed else SessionEnd.Closed
|
||||
finished.complete(end)
|
||||
}
|
||||
},
|
||||
)
|
||||
webSocket = socket
|
||||
return try {
|
||||
finished.await()
|
||||
} finally {
|
||||
if (webSocket === socket) webSocket = null
|
||||
socket.cancel()
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 握手鉴权失败是**握手前**的 HTTP 401（不是 WS close），OkHttp 会把它随 [onFailure] 的
|
||||
* `response` 一起回调；个别路径拿不到 response 时再退化为消息文本匹配。
|
||||
*/
|
||||
private fun isAuthFailure(response: Response?, t: Throwable): Boolean {
|
||||
if (response?.code == 401) return true
|
||||
return t.message?.contains("401") == true
|
||||
}
|
||||
|
||||
private fun buildUrl(): HttpUrl? {
|
||||
val token = tokenProvider()?.takeIf { it.isNotBlank() } ?: return null
|
||||
val base = baseUrlProvider()?.takeIf { it.isNotBlank() } ?: return null
|
||||
val httpUrl = ApiClientFactory.normalizeBaseUrl(base).toHttpUrlOrNull() ?: return null
|
||||
return httpUrl.newBuilder()
|
||||
.addPathSegments("ws/chat")
|
||||
.addQueryParameter("token", token)
|
||||
.build()
|
||||
}
|
||||
|
||||
/** 1s/2s/4s/8s/16s/30s，上限 30s；加 0~25% 抖动避免同时重连。 */
|
||||
private fun backoffDelayMs(attempt: Int): Long {
|
||||
val shift = attempt.coerceIn(0, MAX_BACKOFF_SHIFT)
|
||||
val base = (INITIAL_BACKOFF_MS shl shift).coerceAtMost(MAX_BACKOFF_MS)
|
||||
val jitter = Random.nextLong(base / JITTER_DIVISOR + 1)
|
||||
return (base + jitter).coerceAtMost(MAX_BACKOFF_MS)
|
||||
}
|
||||
|
||||
private fun ensureScope(): CoroutineScope =
|
||||
scope ?: CoroutineScope(SupervisorJob() + Dispatchers.IO).also { scope = it }
|
||||
|
||||
private enum class SessionEnd { Closed, AuthFailed, ConfigMissing }
|
||||
|
||||
private companion object {
|
||||
const val PING_INTERVAL_SECONDS = 30L
|
||||
const val CONNECT_TIMEOUT_SECONDS = 10L
|
||||
const val INITIAL_BACKOFF_MS = 1_000L
|
||||
const val MAX_BACKOFF_MS = 30_000L
|
||||
const val MAX_BACKOFF_SHIFT = 5
|
||||
const val JITTER_DIVISOR = 4L
|
||||
const val EVENT_BUFFER = 512
|
||||
const val CLOSE_CODE_NORMAL = 1000
|
||||
const val CLOSE_REASON_CLIENT = "client closed"
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,459 @@
|
||||
package top.yeij.cyrene.data.ws
|
||||
|
||||
import top.yeij.cyrene.data.api.ApiClientFactory
|
||||
import top.yeij.cyrene.data.api.dto.HistoryMessageDto
|
||||
import kotlinx.serialization.Serializable
|
||||
import kotlinx.serialization.json.JsonElement
|
||||
import kotlinx.serialization.json.JsonNull
|
||||
import kotlinx.serialization.json.JsonObject
|
||||
import kotlinx.serialization.json.JsonPrimitive
|
||||
import kotlinx.serialization.json.decodeFromJsonElement
|
||||
|
||||
/**
|
||||
* 下行事件（服务端 → 客户端）。
|
||||
*
|
||||
* 帧壳：`{type, turnId?, seq?, sessionId?, data?}`（api-reference §6.2），
|
||||
* 只有 `type` 必填，`data` 的形状由 `type` 决定。字段名逐字对齐 api-reference §6.4。
|
||||
*
|
||||
* ## 会话归属约束（重要）
|
||||
* `message.start/delta/end`、`thinking.*`、`tool_call.*`、`tool_execution.update`、
|
||||
* `stream_chunk` **不带 `sessionId`**。本层不会凭空补 `sessionId`，因此上层必须把这些事件
|
||||
* 归属到「当前唯一的 active 流式会话」——即**同一时刻只允许一个会话在流式**。
|
||||
* 带 `sessionId` 的是 `history` / `stream_end` / `message_persisted` / `channel_persisted`
|
||||
* 以及编排状态帧（`session_busy` / `pending_*` / `session_idle` / `session_interrupted`）。
|
||||
*
|
||||
* ## 未知事件与坏帧
|
||||
* [parse] 对未知 `type`、非 JSON、结构不匹配的帧一律返回 `null`（安全忽略），绝不抛异常。
|
||||
*/
|
||||
sealed class WsEvent {
|
||||
|
||||
abstract val type: String
|
||||
|
||||
/** 一轮交换的关联 id（`turn-<hex>`）；OnStream 结构化流路径不带。 */
|
||||
open val turnId: String? = null
|
||||
|
||||
/** hub 级单调递增序号，供排序关联流式突发。 */
|
||||
open val seq: Long? = null
|
||||
|
||||
/** 能确定会话的事件才有；流式正文/思维链/工具帧恒为 `null`。 */
|
||||
open val sessionId: String? = null
|
||||
|
||||
/** 连接首帧：主会话最近 50 条历史（`thinking` 已被服务端剥离）。 */
|
||||
data class History(
|
||||
val messages: List<HistoryMessageDto>,
|
||||
override val turnId: String? = null,
|
||||
override val seq: Long? = null,
|
||||
override val sessionId: String? = null,
|
||||
) : WsEvent() {
|
||||
override val type: String get() = "history"
|
||||
}
|
||||
|
||||
/** 正文段开始。 */
|
||||
data class MessageStart(
|
||||
val index: Int,
|
||||
override val turnId: String? = null,
|
||||
override val seq: Long? = null,
|
||||
) : WsEvent() {
|
||||
override val type: String get() = "message.start"
|
||||
}
|
||||
|
||||
/** 正文 token 增量（累加到当前正文段）。 */
|
||||
data class MessageDelta(
|
||||
val delta: String,
|
||||
val index: Int,
|
||||
override val turnId: String? = null,
|
||||
override val seq: Long? = null,
|
||||
) : WsEvent() {
|
||||
override val type: String get() = "message.delta"
|
||||
}
|
||||
|
||||
/** 正文段结束（不结束整轮，整轮由 [StreamEnd] 收尾）。 */
|
||||
data class MessageEnd(
|
||||
val index: Int,
|
||||
override val turnId: String? = null,
|
||||
override val seq: Long? = null,
|
||||
) : WsEvent() {
|
||||
override val type: String get() = "message.end"
|
||||
}
|
||||
|
||||
/** 思维链段开始。 */
|
||||
data class ThinkingStart(
|
||||
val index: Int,
|
||||
override val turnId: String? = null,
|
||||
override val seq: Long? = null,
|
||||
) : WsEvent() {
|
||||
override val type: String get() = "thinking.start"
|
||||
}
|
||||
|
||||
/** 思维链 token 增量。 */
|
||||
data class ThinkingDelta(
|
||||
val delta: String,
|
||||
val index: Int,
|
||||
override val turnId: String? = null,
|
||||
override val seq: Long? = null,
|
||||
) : WsEvent() {
|
||||
override val type: String get() = "thinking.delta"
|
||||
}
|
||||
|
||||
/** 思维链段结束。 */
|
||||
data class ThinkingEnd(
|
||||
val index: Int,
|
||||
override val turnId: String? = null,
|
||||
override val seq: Long? = null,
|
||||
) : WsEvent() {
|
||||
override val type: String get() = "thinking.end"
|
||||
}
|
||||
|
||||
/** 工具调用中间态开始。 */
|
||||
data class ToolCallStart(
|
||||
val id: String,
|
||||
val index: Int,
|
||||
override val turnId: String? = null,
|
||||
override val seq: Long? = null,
|
||||
) : WsEvent() {
|
||||
override val type: String get() = "tool_call.start"
|
||||
}
|
||||
|
||||
/** 工具调用参数增量（如 `input_json_delta`）。 */
|
||||
data class ToolCallDelta(
|
||||
val id: String,
|
||||
val index: Int,
|
||||
val delta: String,
|
||||
override val turnId: String? = null,
|
||||
override val seq: Long? = null,
|
||||
) : WsEvent() {
|
||||
override val type: String get() = "tool_call.delta"
|
||||
}
|
||||
|
||||
/** 工具调用中间态结束。 */
|
||||
data class ToolCallEnd(
|
||||
val id: String,
|
||||
val index: Int,
|
||||
override val turnId: String? = null,
|
||||
override val seq: Long? = null,
|
||||
) : WsEvent() {
|
||||
override val type: String get() = "tool_call.end"
|
||||
}
|
||||
|
||||
/** 工具执行进度 / 局部输出。 */
|
||||
data class ToolExecutionUpdate(
|
||||
val id: String,
|
||||
val index: Int,
|
||||
val delta: String,
|
||||
override val turnId: String? = null,
|
||||
override val seq: Long? = null,
|
||||
) : WsEvent() {
|
||||
override val type: String get() = "tool_execution.update"
|
||||
}
|
||||
|
||||
/** 向后兼容整段文本（仅非流式回退路径；流式路径不 emit）。 */
|
||||
data class StreamChunk(
|
||||
val delta: String,
|
||||
override val turnId: String? = null,
|
||||
override val seq: Long? = null,
|
||||
override val sessionId: String? = null,
|
||||
) : WsEvent() {
|
||||
override val type: String get() = "stream_chunk"
|
||||
}
|
||||
|
||||
/**
|
||||
* 一轮结束。`stopReason` 取值未穷举（`stop` / `idle` / `error` / …），用字符串承载；
|
||||
* `error` 形状未定义（字符串或对象都可能），原样以 [JsonElement] 暴露。
|
||||
*/
|
||||
data class StreamEnd(
|
||||
val stopReason: String,
|
||||
val steps: Int? = null,
|
||||
val error: JsonElement? = null,
|
||||
override val turnId: String? = null,
|
||||
override val seq: Long? = null,
|
||||
override val sessionId: String? = null,
|
||||
) : WsEvent() {
|
||||
override val type: String get() = "stream_end"
|
||||
}
|
||||
|
||||
/** 会话进入忙碌态。`mode` ∈ `idle` / `thinking` / `executing` / `subagent`。 */
|
||||
data class SessionBusy(
|
||||
override val sessionId: String,
|
||||
val mode: String,
|
||||
override val turnId: String? = null,
|
||||
override val seq: Long? = null,
|
||||
) : WsEvent() {
|
||||
override val type: String get() = "session_busy"
|
||||
}
|
||||
|
||||
/** 忙时新消息已入队（「收到，稍后回」）。 */
|
||||
data class PendingAck(
|
||||
override val sessionId: String,
|
||||
val mode: String,
|
||||
val queueDepth: Int,
|
||||
override val turnId: String? = null,
|
||||
override val seq: Long? = null,
|
||||
) : WsEvent() {
|
||||
override val type: String get() = "pending_ack"
|
||||
}
|
||||
|
||||
/** 忙完开始消费下一条待处理消息。 */
|
||||
data class PendingDrain(
|
||||
override val sessionId: String,
|
||||
val mode: String,
|
||||
val queueDepth: Int,
|
||||
override val turnId: String? = null,
|
||||
override val seq: Long? = null,
|
||||
) : WsEvent() {
|
||||
override val type: String get() = "pending_drain"
|
||||
}
|
||||
|
||||
/** 会话回到空闲、队列清空。 */
|
||||
data class SessionIdle(
|
||||
override val sessionId: String,
|
||||
val mode: String,
|
||||
override val turnId: String? = null,
|
||||
override val seq: Long? = null,
|
||||
) : WsEvent() {
|
||||
override val type: String get() = "session_idle"
|
||||
}
|
||||
|
||||
/** 用户打断：当前任务取消、pending 清空，`dropped` 为丢弃条数。 */
|
||||
data class SessionInterrupted(
|
||||
override val sessionId: String,
|
||||
val mode: String,
|
||||
val dropped: Int,
|
||||
override val turnId: String? = null,
|
||||
override val seq: Long? = null,
|
||||
) : WsEvent() {
|
||||
override val type: String get() = "session_interrupted"
|
||||
}
|
||||
|
||||
/** 单会话 pending 达上限（`max`，默认 5），拒绝入队。 */
|
||||
data class PendingFull(
|
||||
override val sessionId: String,
|
||||
val mode: String,
|
||||
val max: Int,
|
||||
override val turnId: String? = null,
|
||||
override val seq: Long? = null,
|
||||
) : WsEvent() {
|
||||
override val type: String get() = "pending_full"
|
||||
}
|
||||
|
||||
/** 主会话（`web` / `web:<uuid>`）消息落库后广播。 */
|
||||
data class MessagePersisted(
|
||||
val message: HistoryMessageDto?,
|
||||
override val turnId: String? = null,
|
||||
override val seq: Long? = null,
|
||||
override val sessionId: String? = null,
|
||||
) : WsEvent() {
|
||||
override val type: String get() = "message_persisted"
|
||||
}
|
||||
|
||||
/** 隔离会话（`sess-xxx`，群聊 / 陌生人）消息落库后广播，供 IM 适配器日志页。 */
|
||||
data class ChannelPersisted(
|
||||
val message: HistoryMessageDto?,
|
||||
override val turnId: String? = null,
|
||||
override val seq: Long? = null,
|
||||
override val sessionId: String? = null,
|
||||
) : WsEvent() {
|
||||
override val type: String get() = "channel_persisted"
|
||||
}
|
||||
|
||||
/** 语音合成回复（定向回请求连接）。 */
|
||||
data class VoiceOutput(
|
||||
val audio: String?,
|
||||
val format: String?,
|
||||
val text: String?,
|
||||
val reply: String?,
|
||||
override val turnId: String? = null,
|
||||
override val seq: Long? = null,
|
||||
override val sessionId: String? = null,
|
||||
) : WsEvent() {
|
||||
override val type: String get() = "voice_output"
|
||||
}
|
||||
|
||||
/** 自主主动推送（仅 owner/admin 连接收到）。 */
|
||||
data class ProactiveMessage(
|
||||
val platform: String?,
|
||||
val channelId: String?,
|
||||
val channelType: String?,
|
||||
val content: String,
|
||||
val contentType: String?,
|
||||
val sender: String?,
|
||||
val timestamp: Long?,
|
||||
val messageId: String?,
|
||||
override val turnId: String? = null,
|
||||
override val seq: Long? = null,
|
||||
override val sessionId: String? = null,
|
||||
) : WsEvent() {
|
||||
override val type: String get() = "proactive_message"
|
||||
}
|
||||
|
||||
/** 服务端错误（非法 JSON / 未知类型 / 语音·子代理·任务错误等）。 */
|
||||
data class Error(
|
||||
val message: String,
|
||||
override val turnId: String? = null,
|
||||
override val seq: Long? = null,
|
||||
override val sessionId: String? = null,
|
||||
) : WsEvent() {
|
||||
override val type: String get() = "error"
|
||||
}
|
||||
|
||||
companion object {
|
||||
|
||||
/**
|
||||
* 解析一帧文本。**永不抛异常**：非 JSON、缺 `type`、未知 `type`、`data` 结构不匹配
|
||||
* 都返回 `null`，由调用方安全忽略。
|
||||
*/
|
||||
fun parse(raw: String): WsEvent? {
|
||||
val root = runCatching { ApiClientFactory.json.parseToJsonElement(raw) }
|
||||
.getOrNull() as? JsonObject ?: return null
|
||||
val type = root["type"].asStringOrNull() ?: return null
|
||||
val turnId = root["turnId"].asStringOrNull()
|
||||
val seq = root["seq"].asStringOrNull()?.toLongOrNull()
|
||||
val sessionId = root["sessionId"].asStringOrNull()
|
||||
val data = root["data"]?.takeIf { it !is JsonNull }
|
||||
return runCatching { dispatch(type, turnId, seq, sessionId, data) }.getOrNull()
|
||||
}
|
||||
|
||||
private fun dispatch(
|
||||
type: String,
|
||||
turnId: String?,
|
||||
seq: Long?,
|
||||
sessionId: String?,
|
||||
data: JsonElement?,
|
||||
): WsEvent? = dispatchStreaming(type, turnId, seq, sessionId, data)
|
||||
?: dispatchSession(type, turnId, seq, sessionId, data)
|
||||
|
||||
private fun dispatchStreaming(
|
||||
type: String,
|
||||
turnId: String?,
|
||||
seq: Long?,
|
||||
sessionId: String?,
|
||||
data: JsonElement?,
|
||||
): WsEvent? = when (type) {
|
||||
"history" -> History(decode<HistoryData>(data).messages, turnId, seq, sessionId)
|
||||
"message.start" -> MessageStart(decode<IndexData>(data).index, turnId, seq)
|
||||
"message.delta" -> decode<DeltaIndexData>(data).let { MessageDelta(it.delta, it.index, turnId, seq) }
|
||||
"message.end" -> MessageEnd(decode<IndexData>(data).index, turnId, seq)
|
||||
"thinking.start" -> ThinkingStart(decode<IndexData>(data).index, turnId, seq)
|
||||
"thinking.delta" -> decode<DeltaIndexData>(data).let { ThinkingDelta(it.delta, it.index, turnId, seq) }
|
||||
"thinking.end" -> ThinkingEnd(decode<IndexData>(data).index, turnId, seq)
|
||||
"tool_call.start" -> decode<ToolCallIdData>(data).let { ToolCallStart(it.id, it.index, turnId, seq) }
|
||||
"tool_call.delta" -> decode<ToolCallData>(data).let { ToolCallDelta(it.id, it.index, it.delta, turnId, seq) }
|
||||
"tool_call.end" -> decode<ToolCallIdData>(data).let { ToolCallEnd(it.id, it.index, turnId, seq) }
|
||||
"tool_execution.update" -> decode<ToolCallData>(data)
|
||||
.let { ToolExecutionUpdate(it.id, it.index, it.delta, turnId, seq) }
|
||||
"stream_chunk" -> StreamChunk(decode<StreamChunkData>(data).delta, turnId, seq, sessionId)
|
||||
"stream_end" -> decode<StreamEndData>(data)
|
||||
.let { StreamEnd(it.stopReason, it.steps, it.error, turnId, seq, sessionId) }
|
||||
else -> null
|
||||
}
|
||||
|
||||
private fun dispatchSession(
|
||||
type: String,
|
||||
turnId: String?,
|
||||
seq: Long?,
|
||||
sessionId: String?,
|
||||
data: JsonElement?,
|
||||
): WsEvent? = when (type) {
|
||||
"session_busy" -> decode<SessionModeData>(data)
|
||||
.let { SessionBusy(resolveSessionId(sessionId, it.sessionId), it.mode, turnId, seq) }
|
||||
"pending_ack" -> decode<PendingData>(data)
|
||||
.let { PendingAck(resolveSessionId(sessionId, it.sessionId), it.mode, it.queueDepth, turnId, seq) }
|
||||
"pending_drain" -> decode<PendingData>(data)
|
||||
.let { PendingDrain(resolveSessionId(sessionId, it.sessionId), it.mode, it.queueDepth, turnId, seq) }
|
||||
"session_idle" -> decode<SessionModeData>(data)
|
||||
.let { SessionIdle(resolveSessionId(sessionId, it.sessionId), it.mode, turnId, seq) }
|
||||
"session_interrupted" -> decode<InterruptedData>(data)
|
||||
.let { SessionInterrupted(resolveSessionId(sessionId, it.sessionId), it.mode, it.dropped, turnId, seq) }
|
||||
"pending_full" -> decode<PendingFullData>(data)
|
||||
.let { PendingFull(resolveSessionId(sessionId, it.sessionId), it.mode, it.max, turnId, seq) }
|
||||
"message_persisted" -> decode<PersistedData>(data)
|
||||
.let { MessagePersisted(it.message, turnId, seq, sessionId ?: it.message?.sessionId) }
|
||||
"channel_persisted" -> decode<PersistedData>(data)
|
||||
.let { ChannelPersisted(it.message, turnId, seq, sessionId ?: it.message?.sessionId) }
|
||||
"voice_output" -> decode<VoiceOutputData>(data)
|
||||
.let { VoiceOutput(it.audio, it.format, it.text, it.reply, turnId, seq, sessionId) }
|
||||
"proactive_message" -> decode<ProactiveData>(data).let {
|
||||
ProactiveMessage(
|
||||
it.platform, it.channelId, it.channelType, it.content, it.contentType,
|
||||
it.sender, it.timestamp, it.messageId, turnId, seq, sessionId,
|
||||
)
|
||||
}
|
||||
"error" -> Error(decode<ErrorData>(data).message, turnId, seq, sessionId)
|
||||
else -> null
|
||||
}
|
||||
|
||||
private fun resolveSessionId(shell: String?, fromData: String): String =
|
||||
fromData.ifBlank { shell ?: "" }
|
||||
|
||||
private inline fun <reified T> decode(data: JsonElement?): T =
|
||||
ApiClientFactory.json.decodeFromJsonElement<T>(data ?: JsonObject(emptyMap()))
|
||||
}
|
||||
}
|
||||
|
||||
private fun JsonElement?.asStringOrNull(): String? {
|
||||
val primitive = this as? JsonPrimitive ?: return null
|
||||
if (primitive is JsonNull) return null
|
||||
return primitive.content
|
||||
}
|
||||
|
||||
@Serializable
|
||||
private data class IndexData(val index: Int = 0)
|
||||
|
||||
@Serializable
|
||||
private data class DeltaIndexData(val delta: String = "", val index: Int = 0)
|
||||
|
||||
@Serializable
|
||||
private data class ToolCallIdData(val id: String = "", val index: Int = 0)
|
||||
|
||||
@Serializable
|
||||
private data class ToolCallData(val id: String = "", val index: Int = 0, val delta: String = "")
|
||||
|
||||
@Serializable
|
||||
private data class HistoryData(val messages: List<HistoryMessageDto> = emptyList())
|
||||
|
||||
@Serializable
|
||||
private data class StreamChunkData(val delta: String = "")
|
||||
|
||||
@Serializable
|
||||
private data class StreamEndData(
|
||||
val stopReason: String = "",
|
||||
val steps: Int? = null,
|
||||
val error: JsonElement? = null,
|
||||
)
|
||||
|
||||
@Serializable
|
||||
private data class SessionModeData(val sessionId: String = "", val mode: String = "")
|
||||
|
||||
@Serializable
|
||||
private data class PendingData(val sessionId: String = "", val mode: String = "", val queueDepth: Int = 0)
|
||||
|
||||
@Serializable
|
||||
private data class InterruptedData(val sessionId: String = "", val mode: String = "", val dropped: Int = 0)
|
||||
|
||||
@Serializable
|
||||
private data class PendingFullData(val sessionId: String = "", val mode: String = "", val max: Int = 0)
|
||||
|
||||
@Serializable
|
||||
private data class PersistedData(val message: HistoryMessageDto? = null)
|
||||
|
||||
@Serializable
|
||||
private data class VoiceOutputData(
|
||||
val audio: String? = null,
|
||||
val format: String? = null,
|
||||
val text: String? = null,
|
||||
val reply: String? = null,
|
||||
)
|
||||
|
||||
@Serializable
|
||||
private data class ProactiveData(
|
||||
val platform: String? = null,
|
||||
val channelId: String? = null,
|
||||
val channelType: String? = null,
|
||||
val content: String = "",
|
||||
val contentType: String? = null,
|
||||
val sender: String? = null,
|
||||
val timestamp: Long? = null,
|
||||
val messageId: String? = null,
|
||||
)
|
||||
|
||||
@Serializable
|
||||
private data class ErrorData(val message: String = "")
|
||||
@@ -0,0 +1,118 @@
|
||||
package top.yeij.cyrene.data.ws
|
||||
|
||||
import kotlinx.serialization.json.JsonElement
|
||||
import kotlinx.serialization.json.buildJsonObject
|
||||
import kotlinx.serialization.json.put
|
||||
|
||||
/**
|
||||
* 上行事件（客户端 → 服务端）。
|
||||
*
|
||||
* 帧壳：`{type, reqId?, data?}`（api-reference §6.2）。字段名逐字对齐 §6.3：
|
||||
* `chat` / `interrupt` / `voice_input` / `mode_switch` / `subagent_ask` / `task_output_query`。
|
||||
*
|
||||
* 序列化用 [buildJsonObject] 显式拼装，而不是 `@Serializable` 数据类，原因：
|
||||
* 上行帧是「信封 + 变长 data」，这样能保证 `type` 与 `data` 的嵌套结构不随类字段变化，
|
||||
* 且可选字段为 `null` 时整键省略（与后端 `omitempty` 语义一致）。
|
||||
*/
|
||||
sealed interface WsOutbound {
|
||||
|
||||
/** 事件类型，即帧壳的 `type`。 */
|
||||
val type: String
|
||||
|
||||
/** 请求关联 id；服务端当前不回填（api-reference §6.3），预留。 */
|
||||
val reqId: String? get() = null
|
||||
|
||||
/** `data` 负载；返回 `null` 时省略该键。 */
|
||||
fun data(): JsonElement?
|
||||
|
||||
/** 编码成一条文本帧。 */
|
||||
fun toJson(): String = buildJsonObject {
|
||||
put("type", type)
|
||||
reqId?.let { put("reqId", it) }
|
||||
data()?.let { put("data", it) }
|
||||
}.toString()
|
||||
|
||||
/** `chat`：`{sessionId?, text, image?, deviceId?}`；text+image 至少一项。 */
|
||||
data class Chat(
|
||||
val text: String,
|
||||
val sessionId: String? = null,
|
||||
val deviceId: String? = null,
|
||||
val image: String? = null,
|
||||
override val reqId: String? = null,
|
||||
) : WsOutbound {
|
||||
override val type: String get() = "chat"
|
||||
|
||||
override fun data(): JsonElement = buildJsonObject {
|
||||
sessionId?.let { put("sessionId", it) }
|
||||
put("text", text)
|
||||
image?.let { put("image", it) }
|
||||
deviceId?.let { put("deviceId", it) }
|
||||
}
|
||||
}
|
||||
|
||||
/** `interrupt`：`{sessionId}`。命中空闲会话为 no-op。 */
|
||||
data class Interrupt(
|
||||
val sessionId: String,
|
||||
override val reqId: String? = null,
|
||||
) : WsOutbound {
|
||||
override val type: String get() = "interrupt"
|
||||
|
||||
override fun data(): JsonElement = buildJsonObject { put("sessionId", sessionId) }
|
||||
}
|
||||
|
||||
/** `voice_input`：`{audio?, format?, text?, sessionId?}`。 */
|
||||
data class VoiceInput(
|
||||
val audio: String? = null,
|
||||
val format: String? = null,
|
||||
val text: String? = null,
|
||||
val sessionId: String? = null,
|
||||
override val reqId: String? = null,
|
||||
) : WsOutbound {
|
||||
override val type: String get() = "voice_input"
|
||||
|
||||
override fun data(): JsonElement = buildJsonObject {
|
||||
audio?.let { put("audio", it) }
|
||||
format?.let { put("format", it) }
|
||||
text?.let { put("text", it) }
|
||||
sessionId?.let { put("sessionId", it) }
|
||||
}
|
||||
}
|
||||
|
||||
/** `mode_switch`：服务端原样广播回显，负载形状不限，这里发 `{mode}`。 */
|
||||
data class ModeSwitch(
|
||||
val mode: String,
|
||||
override val reqId: String? = null,
|
||||
) : WsOutbound {
|
||||
override val type: String get() = "mode_switch"
|
||||
|
||||
override fun data(): JsonElement = buildJsonObject { put("mode", mode) }
|
||||
}
|
||||
|
||||
/** `subagent_ask`：`{id, mode?, n?, query?}`。`mode` ∈ `""/full/tail/keyword`。 */
|
||||
data class SubagentAsk(
|
||||
val id: String,
|
||||
val mode: String? = null,
|
||||
val n: Int? = null,
|
||||
val query: String? = null,
|
||||
override val reqId: String? = null,
|
||||
) : WsOutbound {
|
||||
override val type: String get() = "subagent_ask"
|
||||
|
||||
override fun data(): JsonElement = buildJsonObject {
|
||||
put("id", id)
|
||||
mode?.let { put("mode", it) }
|
||||
n?.let { put("n", it) }
|
||||
query?.let { put("query", it) }
|
||||
}
|
||||
}
|
||||
|
||||
/** `task_output_query`：`{taskId}`。 */
|
||||
data class TaskOutputQuery(
|
||||
val taskId: String,
|
||||
override val reqId: String? = null,
|
||||
) : WsOutbound {
|
||||
override val type: String get() = "task_output_query"
|
||||
|
||||
override fun data(): JsonElement = buildJsonObject { put("taskId", taskId) }
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,25 @@
|
||||
package top.yeij.cyrene.data.ws
|
||||
|
||||
/**
|
||||
* 实时通道连接状态。
|
||||
*
|
||||
* 迁移：`Idle → Connecting → Connected → Reconnecting(退避) → Offline`。
|
||||
* `Offline` 只在**不可恢复**的情况下出现（握手 401 / 缺少 token 或服务器地址），
|
||||
* 需要上层重新登录或配置服务器后再调 [ChatSocket.connect]。
|
||||
*/
|
||||
enum class WsState {
|
||||
/** 未启动：初始态，或主动 [ChatSocket.close] 之后。 */
|
||||
Idle,
|
||||
|
||||
/** 首次握手进行中。 */
|
||||
Connecting,
|
||||
|
||||
/** 已连接，可收发帧。 */
|
||||
Connected,
|
||||
|
||||
/** 连接断开，正在按退避策略等待重连。 */
|
||||
Reconnecting,
|
||||
|
||||
/** 不可恢复（握手 401 / 缺少 token 或地址），不会自动重连。 */
|
||||
Offline,
|
||||
}
|
||||
Reference in new issue
Block a user