feat(partnerctl): implement interactive WebSocket chat client with live event rendering

PartnerCtl/pom.xml

@@ -41,6 +41,12 @@
             <artifactId>kotlinx-serialization-json</artifactId>
             <version>1.9.0</version>
         </dependency>
+        <dependency>
+            <groupId>work.slhaf.partner</groupId>
+            <artifactId>partner-interaction-api</artifactId>
+            <version>0.5.0</version>
+            <scope>compile</scope>
+        </dependency>
     </dependencies>
 
     <build>

PartnerCtl/src/main/java/work/slhaf/partner/ctl/commands/ChatCommand.kt

@@ -1,11 +1,9 @@
 package work.slhaf.partner.ctl.commands
 
-import org.jline.reader.EndOfFileException
-import org.jline.reader.LineReaderBuilder
-import org.jline.reader.UserInterruptException
-import org.jline.terminal.Terminal
-import org.jline.terminal.TerminalBuilder
 import picocli.CommandLine
+import work.slhaf.partner.api.InputData
+import work.slhaf.partner.ctl.commands.chat.ChatScreen
+import work.slhaf.partner.ctl.commands.chat.WebSocketClient
 
 @CommandLine.Command(
     name = "chat",
@@ -17,43 +15,33 @@ class ChatCommand : Runnable {
     @CommandLine.Mixin
     lateinit var helpOptions: HelpOptions
 
+    @CommandLine.Option(
+        names = ["--url"],
+        description = ["WebSocket gateway URL."],
+        defaultValue = DEFAULT_URL,
+    )
+    lateinit var url: String
+
+    @CommandLine.Option(
+        names = ["--source"],
+        description = ["Input source identity used by Partner runtime."],
+        defaultValue = DEFAULT_SOURCE,
+    )
+    lateinit var source: String
+
     override fun run() {
-        val terminal = createTerminal()
+        val screen = ChatScreen()
-        val reader = LineReaderBuilder.builder()
+        WebSocketClient(url) { event ->
-            .terminal(terminal)
+            screen.postInteractionEvent(event)
-            .build()
+        }.use { client ->
-
+            screen.run { line ->
-        terminal.writer().println("Partner chat demo. Type /exit to quit.")
+                client.send(InputData(source, line))
-        terminal.writer().flush()
-
-        while (true) {
-            val line = try {
-                reader.readLine("partner> ")
-            } catch (_: UserInterruptException) {
-                terminal.writer().println()
-                terminal.writer().flush()
-                continue
-            } catch (_: EndOfFileException) {
-                terminal.writer().println()
-                terminal.writer().flush()
-                break
-            }
-
-            when {
-                line == "/exit" -> break
-                line.isBlank() -> continue
-                else -> {
-                    terminal.writer().println("echo: $line")
-                    terminal.writer().flush()
-                }
             }
         }
     }
 
-    private fun createTerminal(): Terminal {
+    private companion object {
-        return TerminalBuilder.builder()
+        private const val DEFAULT_URL = "ws://127.0.0.1:29600"
-            .system(true)
+        private const val DEFAULT_SOURCE = "partnerctl"
-            .dumb(true)
-            .build()
     }
-}
+}

PartnerCtl/src/main/java/work/slhaf/partner/ctl/commands/chat/ChatEventRenderer.kt

@@ -0,0 +1,28 @@
+package work.slhaf.partner.ctl.commands.chat
+
+import work.slhaf.partner.api.InteractionEvent
+import work.slhaf.partner.api.ModuleEvent
+import work.slhaf.partner.api.ReplyEvent
+import work.slhaf.partner.api.SystemEvent
+
+internal class ChatEventRenderer {
+    fun renderCommittedUserInput(content: String): String = "you: $content"
+
+    fun renderActiveReply(content: String): String {
+        return if (content.isBlank()) {
+            "assistant:"
+        } else {
+            "assistant: $content"
+        }
+    }
+
+    fun renderEventMessage(event: InteractionEvent): String? {
+        return when (event) {
+            is ReplyEvent -> null
+            is SystemEvent -> "system: ${event.title}: ${event.content}"
+            is ModuleEvent -> "module:${event.data.module}: ${event.data.content}"
+        }
+    }
+
+    fun renderSendFailure(message: String): String = "send failed: $message"
+}

PartnerCtl/src/main/java/work/slhaf/partner/ctl/commands/chat/ChatInputBuffer.kt

@@ -0,0 +1,30 @@
+package work.slhaf.partner.ctl.commands.chat
+
+internal class ChatInputBuffer {
+    private val buffer = StringBuilder()
+
+    val isEmpty: Boolean
+        get() = buffer.isEmpty()
+
+    fun append(ch: Char) {
+        buffer.append(ch)
+    }
+
+    fun backspace() {
+        if (buffer.isNotEmpty()) {
+            buffer.deleteCharAt(buffer.length - 1)
+        }
+    }
+
+    fun clear() {
+        buffer.setLength(0)
+    }
+
+    fun consume(): String {
+        val value = buffer.toString()
+        clear()
+        return value
+    }
+
+    override fun toString(): String = buffer.toString()
+}

PartnerCtl/src/main/java/work/slhaf/partner/ctl/commands/chat/ChatScreen.kt

@@ -0,0 +1,270 @@
+package work.slhaf.partner.ctl.commands.chat
+
+import org.jline.terminal.Terminal
+import org.jline.terminal.TerminalBuilder
+import work.slhaf.partner.api.InteractionEvent
+import work.slhaf.partner.api.InteractionEvent.EventStatus
+import work.slhaf.partner.api.ReplyEvent
+import java.util.concurrent.BlockingQueue
+import java.util.concurrent.LinkedBlockingQueue
+import kotlin.math.ceil
+
+internal class ChatScreen(
+    private val terminal: Terminal = TerminalBuilder.builder()
+        .system(true)
+        .dumb(true)
+        .build(),
+    private val renderer: ChatEventRenderer = ChatEventRenderer(),
+) : AutoCloseable {
+
+    private val input = ChatInputBuffer()
+    private val events: BlockingQueue<ChatScreenEvent> = LinkedBlockingQueue()
+    private val activeReply = StringBuilder()
+    private var dynamicRows = 0
+    private var closed = false
+
+    fun postInteractionEvent(event: InteractionEvent) {
+        events.offer(ChatScreenEvent.Interaction(event))
+    }
+
+    fun postSystemMessage(message: String) {
+        events.offer(ChatScreenEvent.SystemMessage(message))
+    }
+
+    fun run(onInput: (String) -> Unit) {
+        terminal.writer().println("Partner chat. Type /exit to quit.")
+        terminal.writer().println()
+        terminal.writer().flush()
+
+        val oldAttributes = terminal.enterRawMode()
+        try {
+            repaintDynamicArea()
+            while (!closed) {
+                drainEvents()
+                val key = terminal.reader().read(50L)
+                if (key != -1) {
+                    handleKey(key, onInput)
+                }
+            }
+        } finally {
+            terminal.attributes = oldAttributes
+            clearDynamicArea()
+            terminal.writer().println()
+            terminal.writer().flush()
+        }
+    }
+
+    override fun close() {
+        closed = true
+    }
+
+    private fun handleKey(key: Int, onInput: (String) -> Unit) {
+        when (key) {
+            CTRL_C, CTRL_D -> close()
+            ENTER, CARRIAGE_RETURN -> submitInput(onInput)
+            BACKSPACE, DELETE -> {
+                input.backspace()
+                repaintDynamicArea()
+            }
+
+            ESCAPE -> consumeEscapeSequence()
+            else -> {
+                if (key >= PRINTABLE_START) {
+                    input.append(key.toChar())
+                    repaintDynamicArea()
+                }
+            }
+        }
+    }
+
+    private fun submitInput(onInput: (String) -> Unit) {
+        val line = input.consume()
+        if (line.isBlank()) {
+            repaintDynamicArea()
+            return
+        }
+        if (line == "/exit") {
+            close()
+            return
+        }
+
+        commitDynamicArea()
+        printCommitted(renderer.renderCommittedUserInput(line))
+        activeReply.setLength(0)
+        repaintDynamicArea()
+
+        runCatching { onInput(line) }
+            .onFailure { error -> showSendFailure(error.message ?: error::class.java.simpleName) }
+    }
+
+    private fun drainEvents() {
+        var changed = false
+        while (true) {
+            val event = events.poll() ?: break
+            when (event) {
+                is ChatScreenEvent.Interaction -> handleInteractionEvent(event.event)
+                is ChatScreenEvent.SystemMessage -> {
+                    commitDynamicArea()
+                    printCommitted(event.message)
+                }
+            }
+            changed = true
+        }
+        if (changed) {
+            repaintDynamicArea()
+        }
+    }
+
+    private fun handleInteractionEvent(event: InteractionEvent) {
+        if (event is ReplyEvent) {
+            when (event.mode) {
+                ReplyEvent.ContentMode.APPEND -> activeReply.append(event.content)
+                ReplyEvent.ContentMode.REPLACE -> {
+                    activeReply.setLength(0)
+                    activeReply.append(event.content)
+                }
+            }
+
+            if (event.status == EventStatus.DONE || event.status == EventStatus.ERROR) {
+                commitDynamicArea()
+                if (activeReply.isNotBlank()) {
+                    printCommitted(renderer.renderActiveReply(activeReply.toString()))
+                    activeReply.setLength(0)
+                }
+                if (event.status == EventStatus.ERROR) {
+                    printCommitted("assistant: [error]")
+                }
+            }
+            return
+        }
+
+        renderer.renderEventMessage(event)?.let { message ->
+            commitDynamicArea()
+            printCommitted(message)
+        }
+    }
+
+    private fun showSendFailure(message: String) {
+        commitDynamicArea()
+        printCommitted(renderer.renderSendFailure(message))
+        repaintDynamicArea()
+    }
+
+    private fun commitDynamicArea() {
+        clearDynamicArea()
+    }
+
+    private fun printCommitted(message: String) {
+        message.split('\n').forEach(terminal.writer()::println)
+        terminal.writer().flush()
+    }
+
+    private fun repaintDynamicArea() {
+        clearDynamicArea()
+
+        val output = dynamicOutput()
+        terminal.writer().print(output)
+        terminal.writer().flush()
+
+        dynamicRows = measureDisplayRows(output)
+    }
+
+    private fun clearDynamicArea() {
+        if (dynamicRows <= 0) {
+            return
+        }
+
+        terminal.writer().print("\r")
+        if (dynamicRows > 1) {
+            terminal.writer().print("\u001B[${dynamicRows - 1}A")
+        }
+        terminal.writer().print("\u001B[J")
+        dynamicRows = 0
+    }
+
+    private fun dynamicOutput(): String {
+        return buildString {
+            if (activeReply.isNotBlank()) {
+                append(renderer.renderActiveReply(activeReply.toString()))
+                append('\n')
+            }
+            append(inputPrompt())
+        }
+    }
+
+    private fun inputPrompt(): String = "partner> ${input}"
+
+    private fun measureDisplayRows(text: String): Int {
+        val width = terminal.width.takeIf { it > 0 } ?: DEFAULT_TERMINAL_WIDTH
+        return text.split('\n').sumOf { line ->
+            ceil(displayWidth(line).coerceAtLeast(1).toDouble() / width.toDouble())
+                .toInt()
+                .coerceAtLeast(1)
+        }
+    }
+
+    private fun displayWidth(text: String): Int {
+        var width = 0
+        var index = 0
+        while (index < text.length) {
+            val codePoint = text.codePointAt(index)
+            width += codePointWidth(codePoint)
+            index += Character.charCount(codePoint)
+        }
+        return width
+    }
+
+    private fun codePointWidth(codePoint: Int): Int {
+        val type = Character.getType(codePoint)
+        if (type == Character.NON_SPACING_MARK.toInt() ||
+            type == Character.ENCLOSING_MARK.toInt() ||
+            type == Character.FORMAT.toInt()
+        ) {
+            return 0
+        }
+
+        return when {
+            codePoint == 0 -> 0
+            codePoint < 32 || codePoint in 0x7F..0x9F -> 0
+            isWideCodePoint(codePoint) -> 2
+            else -> 1
+        }
+    }
+
+    private fun isWideCodePoint(codePoint: Int): Boolean {
+        return codePoint in 0x1100..0x115F ||
+                codePoint in 0x2329..0x232A ||
+                codePoint in 0x2E80..0xA4CF ||
+                codePoint in 0xAC00..0xD7A3 ||
+                codePoint in 0xF900..0xFAFF ||
+                codePoint in 0xFE10..0xFE19 ||
+                codePoint in 0xFE30..0xFE6F ||
+                codePoint in 0xFF00..0xFF60 ||
+                codePoint in 0xFFE0..0xFFE6 ||
+                codePoint in 0x1F300..0x1FAFF ||
+                codePoint in 0x20000..0x3FFFD
+    }
+
+    private fun consumeEscapeSequence() {
+        while (terminal.reader().read(1L) != -1) {
+            // Drop the rest of the currently available escape sequence.
+        }
+    }
+
+    private sealed interface ChatScreenEvent {
+        data class Interaction(val event: InteractionEvent) : ChatScreenEvent
+        data class SystemMessage(val message: String) : ChatScreenEvent
+    }
+
+    private companion object {
+        private const val CTRL_C = 3
+        private const val CTRL_D = 4
+        private const val BACKSPACE = 8
+        private const val DELETE = 127
+        private const val ENTER = 10
+        private const val CARRIAGE_RETURN = 13
+        private const val ESCAPE = 27
+        private const val PRINTABLE_START = 32
+        private const val DEFAULT_TERMINAL_WIDTH = 80
+    }
+}

PartnerCtl/src/main/java/work/slhaf/partner/ctl/commands/chat/WebSocketClient.kt

@@ -0,0 +1,133 @@
+package work.slhaf.partner.ctl.commands.chat
+
+import kotlinx.serialization.json.*
+import work.slhaf.partner.api.*
+import work.slhaf.partner.api.InteractionEvent.EventStatus
+import java.net.URI
+import java.net.http.HttpClient
+import java.net.http.WebSocket
+import java.time.Duration
+import java.util.concurrent.CompletionStage
+
+class WebSocketClient(
+    val url: String,
+    val onResponse: (event: InteractionEvent) -> Unit
+) : AutoCloseable {
+
+    private val json = Json {
+        ignoreUnknownKeys = true
+        isLenient = true
+    }
+
+    private val httpClient = HttpClient.newBuilder()
+        .connectTimeout(Duration.ofSeconds(5))
+        .build()
+
+    private val listener = Listener(::handleMessage)
+
+    private val webSocket = httpClient.newWebSocketBuilder()
+        .connectTimeout(Duration.ofSeconds(5))
+        .buildAsync(URI.create(url), listener)
+
+    fun send(inputData: InputData) {
+        val socket = webSocket.join()
+        socket.sendText(inputData.toJson(), true).join()
+    }
+
+    override fun close() {
+        if (!webSocket.isDone) {
+            webSocket.cancel(true)
+            return
+        }
+
+        runCatching {
+            webSocket.join().sendClose(WebSocket.NORMAL_CLOSURE, "bye").join()
+        }
+    }
+
+    private fun handleMessage(text: String) {
+        val event = parseInteractionEvent(text) ?: return
+        onResponse(event)
+    }
+
+    private fun parseInteractionEvent(text: String): InteractionEvent? {
+        val obj = runCatching { json.parseToJsonElement(text).jsonObject }.getOrNull() ?: return null
+        val status = obj.string("status")?.let { runCatching { EventStatus.valueOf(it) }.getOrNull() } ?: return null
+        val target = obj.string("target") ?: return null
+
+        return when (obj.string("event")) {
+            "REPLY" -> ReplyEvent(
+                status = status,
+                target = target,
+                content = obj.string("content") ?: "",
+                mode = obj.string("mode")
+                    ?.let { runCatching { ReplyEvent.ContentMode.valueOf(it) }.getOrNull() }
+                    ?: ReplyEvent.ContentMode.REPLACE,
+                seq = obj.long("seq")
+            ).withMetaFrom(obj)
+
+            "SYSTEM" -> SystemEvent(
+                status = status,
+                target = target,
+                title = obj.string("title") ?: "",
+                content = obj.string("content") ?: "",
+                urgency = obj.string("urgency")
+                    ?.let { runCatching { SystemEvent.Urgency.valueOf(it) }.getOrNull() }
+                    ?: SystemEvent.Urgency.NORMAL
+            ).withMetaFrom(obj)
+
+            "MODULE" -> ModuleEvent(
+                status = status,
+                target = target,
+                data = obj["data"]?.jsonObject?.let { data ->
+                    ModuleEvent.Data(
+                        module = data.string("module") ?: "",
+                        content = data.string("content") ?: ""
+                    )
+                } ?: ModuleEvent.Data(module = "", content = "")
+            ).withMetaFrom(obj)
+
+            else -> null
+        }
+    }
+
+    private fun InputData.toJson(): String = buildJsonObject(
+        "source" to JsonPrimitive(source),
+        "content" to JsonPrimitive(content),
+        "meta" to JsonObject(meta.mapValues { JsonPrimitive(it.value) })
+    ).toString()
+
+    private fun <T : InteractionEvent> T.withMetaFrom(obj: JsonObject): T {
+        obj["meta"]?.jsonObject?.forEach { (key, value) ->
+            value.jsonPrimitive.contentOrNull?.let { addMeta(key, it) }
+        }
+        return this
+    }
+
+    private fun JsonObject.string(key: String): String? = this[key]?.jsonPrimitive?.contentOrNull
+
+    private fun JsonObject.long(key: String): Long? = this[key]?.jsonPrimitive?.contentOrNull?.toLongOrNull()
+
+    private fun buildJsonObject(vararg values: Pair<String, JsonElement>): JsonObject = JsonObject(values.toMap())
+
+    private class Listener(
+        private val onMessage: (String) -> Unit
+    ) : WebSocket.Listener {
+        private val buffer = StringBuilder()
+
+        override fun onText(webSocket: WebSocket, data: CharSequence, last: Boolean): CompletionStage<*>? {
+            buffer.append(data)
+            if (last) {
+                val text = buffer.toString()
+                buffer.setLength(0)
+                onMessage(text)
+            }
+            webSocket.request(1)
+            return null
+        }
+
+        override fun onOpen(webSocket: WebSocket) {
+            webSocket.request(1)
+        }
+    }
+}

PartnerCtl/src/main/java/work/slhaf/partner/ctl/ui/Prompt.kt

@@ -12,6 +12,22 @@ import java.nio.file.Files
 import java.nio.file.Path
 import java.nio.file.Paths
 
+data class PromptPart(
+    val text: String,
+    val style: PromptStyle = PromptStyle.PLAIN,
+)
+
+enum class PromptStyle {
+    PLAIN,
+    DIM,
+    BOLD,
+    CYAN,
+    GREEN,
+    YELLOW,
+    RED,
+    BLUE,
+}
+
 class Prompt private constructor(
     private val terminal: Terminal,
     private val reader: LineReader,
@@ -56,8 +72,31 @@ class Prompt private constructor(
     private fun questionPrefix() = cyan("?")
 
     private fun promptLabel(label: String, defaultValue: String? = null): String {
-        val suffix = if (defaultValue != null) " ${dim("[$defaultValue]")}" else ""
+        return renderPrompt(
-        return "${questionPrefix()} $label$suffix: "
+            buildList {
+                add(PromptPart("?", PromptStyle.CYAN))
+                add(PromptPart(" $label", PromptStyle.PLAIN))
+                if (defaultValue != null) {
+                    add(PromptPart(" [$defaultValue]", PromptStyle.DIM))
+                }
+                add(PromptPart(": ", PromptStyle.PLAIN))
+            }
+        )
+    }
+
+    private fun renderPrompt(parts: List<PromptPart>): String {
+        return parts.joinToString(separator = "") { part ->
+            when (part.style) {
+                PromptStyle.PLAIN -> part.text
+                PromptStyle.DIM -> dim(part.text)
+                PromptStyle.BOLD -> bold(part.text)
+                PromptStyle.CYAN -> cyan(part.text)
+                PromptStyle.GREEN -> green(part.text)
+                PromptStyle.YELLOW -> yellow(part.text)
+                PromptStyle.RED -> red(part.text)
+                PromptStyle.BLUE -> blue(part.text)
+            }
+        }
     }
 
     fun print(message: String) {
@@ -471,6 +510,10 @@ class Prompt private constructor(
         terminal.writer().flush()
     }
 
+    fun readLine(parts: List<PromptPart>): String {
+        return readLine(renderPrompt(parts))
+    }
+
     private fun readLine(prompt: String): String {
         return readLine(reader, prompt)
     }

PartnerCtl/src/main/resources/i18n/messages.properties

@@ -8,7 +8,7 @@ cli.shutdown.option.force.description=Forcefully kill matching Partner process i
 cli.log.description=Show Partner logs.
 cli.log.option.tail.description=Number of log lines to show before exiting or following.
 cli.log.option.follow.description=Follow appended log output.
-cli.chat.description=Start an interactive chat demo.
+cli.chat.description=Start an interactive chat client.
 cli.config.description=Manage Partner configuration.
 cli.module.description=Manage Partner modules.
 

PartnerCtl/src/main/resources/i18n/messages_zh_CN.properties

@@ -8,7 +8,7 @@ cli.shutdown.option.force.description=如果匹配的 Partner 进程没有在超
 cli.log.description=查看 Partner 日志。
 cli.log.option.tail.description=退出或 follow 前显示的日志行数。
 cli.log.option.follow.description=持续跟随新增日志输出。
-cli.chat.description=启动交互式聊天 demo。
+cli.chat.description=启动交互式聊天客户端。
 cli.config.description=管理 Partner 配置。
 cli.module.description=管理 Partner 模块。