Merge pull request #41 from devchat-ai/hanle-workflows-errors

pplam · web-flow · commit 031a3abbedfa · 2023-12-22T15:36:42.000+08:00
Hanle workflows errors
diff --git a/src/main/kotlin/ai/devchat/cli/DevChatWrapper.kt b/src/main/kotlin/ai/devchat/cli/DevChatWrapper.kt
@@ -10,6 +10,10 @@ import com.alibaba.fastjson.JSONArray
 import com.alibaba.fastjson.JSONObject
 import com.intellij.util.containers.addIfNotNull
 import kotlinx.coroutines.*
+import kotlinx.coroutines.channels.SendChannel
+import kotlinx.coroutines.channels.actor
+import kotlinx.coroutines.selects.select
+import kotlinx.coroutines.selects.whileSelect
 import java.io.File
 import java.io.IOException
 
@@ -33,16 +37,13 @@ suspend fun executeCommand(
     command: List<String>,
     workDir: String?,
     env: Map<String, String>,
-    onOutputLine: (String) -> Unit,
-    onErrorLine: (String) -> Unit
-): Int {
+): Process {
     val processBuilder = ProcessBuilder(command)
     workDir?.let {processBuilder.directory(File(workDir))}
     env.forEach { (key, value) -> processBuilder.environment()[key] = value}
-    val process = withContext(Dispatchers.IO) {
+    return withContext(Dispatchers.IO) {
         processBuilder.start()
     }
-    return process.await(onOutputLine, onErrorLine)
 }
 
 class Command(val cmd: MutableList<String> = mutableListOf()) {
@@ -62,34 +63,38 @@ class Command(val cmd: MutableList<String> = mutableListOf()) {
         return this
     }
 
-    fun exec(
-        flags: List<Pair<String, String?>> = listOf(),
-        callback: ((String) -> Unit)? = null,
-        onFinish: ((Int) -> Unit)? = null
-    ): String? {
+    private fun prepare(flags: List<Pair<String, String?>> = listOf()): List<String> {
         val args = flags.fold(mutableListOf<String>()) { acc, (name, value) ->
             acc.add("--$name")
             acc.addIfNotNull(value)
             acc
         }
-        return try {
-            callback?.let {
-                execAsync(cmd + args, callback, DevChatNotifier::stickyError, onFinish);
-                ""
-            } ?: exec(cmd + args)
-        } catch (e: Exception) {
-            Log.warn("Failed to run command $cmd: ${e.message}")
-            throw CommandExecutionException("Failed to run command $cmd: ${e.message}")
-        }
+        return cmd + args
     }
 
-    private fun exec(commands: List<String>): String {
-        Log.info("Executing command: ${commands.joinToString(" ")}}")
+    private fun toString(flags: List<Pair<String, String?>>): String {
+        val preparedCommand = prepare(flags)
+        return env.entries.joinToString(" ") { (k, v) ->
+            val masked = if (k == "OPENAI_API_KEY") v.mapIndexed { i, c ->
+                if (i in 7 until v.length - 7) '*' else c
+            }.joinToString("") else v
+            "$k=$masked"
+        } + " " + preparedCommand.joinToString(" ")
+    }
+
+    fun exec(flags: List<Pair<String, String?>> = listOf()): String {
+        val preparedCommand = prepare(flags)
+        val commandStr = toString(flags)
+        Log.info("Executing command: $commandStr")
         return try {
             val outputLines: MutableList<String> = mutableListOf()
             val errorLines: MutableList<String> = mutableListOf()
             val exitCode = runBlocking {
-                executeCommand(commands, ProjectUtils.project?.basePath, env, outputLines::add, errorLines::add)
+                executeCommand(
+                    preparedCommand,
+                    ProjectUtils.project?.basePath,
+                    env
+                ).await(outputLines::add, errorLines::add)
             }
             val errors = errorLines.joinToString("\n")
 
@@ -98,28 +103,47 @@ class Command(val cmd: MutableList<String> = mutableListOf()) {
             } else {
                 outputLines.joinToString("\n")
             }
-        } catch (e: IOException) {
-            Log.warn("Failed to execute command: $commands, Exception: $e")
-            throw e
+        } catch (e: Exception) {
+            val msg = "Failed to execute command `$commandStr`: $e"
+            Log.warn(msg)
+            throw CommandExecutionException(msg)
         }
     }
 
-    private fun execAsync(
-        commands: List<String>,
+    @OptIn(ObsoleteCoroutinesApi::class, ExperimentalCoroutinesApi::class)
+    fun execAsync(
+        flags: List<Pair<String, String?>>,
         onOutput: (String) -> Unit,
         onError: (String) -> Unit = Log::warn,
         onFinish: ((Int) -> Unit)? = null,
-    ): Job {
-        Log.info("Executing command: ${commands.joinToString(" ")}}")
+    ): SendChannel<String> {
+        val preparedCommand = prepare(flags)
+        val commandStr = toString(flags)
+        Log.info("Executing command: $commandStr")
         val exceptionHandler = CoroutineExceptionHandler { _, exception ->
-            val msg = "Failed to execute command: $commands, Exception: $exception"
+            val msg = "Failed to execute command `$commandStr`: $exception"
             Log.warn(msg)
             onError(msg)
         }
-        val cmdScope = CoroutineScope(SupervisorJob() + Dispatchers.Default)
-
-        return cmdScope.launch(exceptionHandler) {
-            val exitCode = executeCommand(commands, ProjectUtils.project?.basePath, env, onOutput, onError)
+        return CoroutineScope(
+            SupervisorJob() + Dispatchers.Default + exceptionHandler
+        ).actor {
+            val process = executeCommand(preparedCommand, ProjectUtils.project?.basePath, env)
+            val deferred = async {process.await(onOutput, onError)}
+            var exitCode = 0
+            whileSelect {
+                deferred.onAwait {
+                    exitCode = it
+                    false
+                }
+                channel.onReceive {msg ->
+                    process.outputStream.use {
+                        it.write(msg.toByteArray())
+                    }
+                    Log.info("Input wrote: $msg")
+                    true
+                }
+            }
             onFinish?.let { onFinish(exitCode) }
             if (exitCode != 0) {
                 throw CommandExecutionException("Command failure with exit Code: $exitCode")
@@ -156,49 +180,47 @@ class DevChatWrapper(
             Log.info("api_key: ${it.substring(0, 5)}...${it.substring(it.length - 4)}")
         }
         env["PYTHONPATH"] = PathUtils.pythonPath
+        env["command_python"] = PathUtils.pythonForWorkflows
         return env
     }
 
-    val runCmd = Command(baseCommand).subcommand("run")::exec
-    val logCmd = Command(baseCommand).subcommand("log")::exec
-    val topicCmd = Command(baseCommand).subcommand("topic")::exec
-    val routeCmd = Command(baseCommand).subcommand("route")::exec
-
-    val run get() = { flags: List<Pair<String, String?>>  ->  runCmd(flags, null, null)}
-    val log get() = { flags: List<Pair<String, String?>>  ->  logCmd(flags, null, null)}
-    val topic get() = { flags: List<Pair<String, String?>>  ->  topicCmd(flags, null, null)}
+    val run = Command(baseCommand).subcommand("run")::exec
+    val log = Command(baseCommand).subcommand("log")::exec
+    val topic = Command(baseCommand).subcommand("topic")::exec
+    val routeCmd = Command(baseCommand).subcommand("route")::execAsync
 
     fun route(
         flags: List<Pair<String, String?>>,
         message: String,
-        callback: ((String) -> Unit)?,
-        onFinish: ((Int) -> Unit)?
+        callback: (String) -> Unit,
+        onError: (String) -> Unit = DevChatNotifier::stickyError,
+        onFinish: ((Int) -> Unit)? = null
     ) {
         when {
             apiKey.isNullOrEmpty() -> DevChatNotifier.stickyError("Please config your API key first.")
             !apiKey!!.startsWith("DC.") -> DevChatNotifier.stickyError("Invalid API key format.")
-            else -> routeCmd(
+            else -> activeChannel = routeCmd(
                 flags
                         + (if (flags.any {
                             it.first == "model" && !it.second.isNullOrEmpty()
                         }) emptyList() else listOf("model" to defaultModel))
                         + listOf("" to message),
                 callback,
+                onError,
                 onFinish
             )
         }
     }
 
     val topicList: JSONArray get() = try {
-        val r = topic(mutableListOf("list" to null)) ?: "[]"
+        val r = topic(mutableListOf("list" to null))
         JSON.parseArray(r)
     } catch (e: Exception) {
         Log.warn("Error list topics: $e")
         JSONArray()
     }
     val commandList: JSONArray get() = try {
-        val r = run(mutableListOf("list" to null)) ?: "[]"
-        JSON.parseArray(r)
+        JSON.parseArray(run(mutableListOf("list" to null)))
     } catch (e: Exception) {
         Log.warn("Error list commands: $e")
         JSONArray()
@@ -207,11 +229,10 @@ class DevChatWrapper(
     val logTopic: (String, Int?) -> JSONArray get() = {topic: String, maxCount: Int? ->
         val num: Int = maxCount ?: DEFAULT_LOG_MAX_COUNT
         try {
-            val r = log(mutableListOf(
+            JSON.parseArray(log(mutableListOf(
                 "topic" to topic,
                 "max-count" to num.toString()
-            )) ?: "[]"
-            JSON.parseArray(r)
+            )))
         } catch (e: Exception) {
             Log.warn("Error log topic: $e")
             JSONArray()
@@ -221,22 +242,24 @@ class DevChatWrapper(
         try {
             log(listOf("insert" to item))
         } catch (e: Exception) {
-            val msg = "Error insert log: $e"
-            Log.warn(msg)
-            DevChatNotifier.error(msg)
+            Log.warn("Error insert log: $e")
         }
     }
 
     val logLast: () -> JSONObject? get() = {
         try {
             log(mutableListOf(
                 "max-count" to "1"
-            ))?.let {
+            )).let {
                 JSON.parseArray(it).getJSONObject(0)
             }
         } catch (e: Exception) {
             Log.warn("Error log topic: $e")
             null
         }
     }
+    companion object {
+        var activeChannel: SendChannel<String>? = null
+    }
+
 }
diff --git a/src/main/kotlin/ai/devchat/common/PathUtils.kt b/src/main/kotlin/ai/devchat/common/PathUtils.kt
@@ -7,6 +7,7 @@ import java.nio.file.attribute.BasicFileAttributes
 object PathUtils {
     val workPath: String = Paths.get(System.getProperty("user.home"), ".chat").toString()
     var pythonCommand: String = "python"
+    var pythonForWorkflows: String = "python"
     val pythonPath: String = Paths.get(workPath, "site-packages").toString()
 
     fun copyResourceDirToPath(resourceDir: String, outputPath: String) {
diff --git a/src/main/kotlin/ai/devchat/devchat/ActionHandlerFactory.kt b/src/main/kotlin/ai/devchat/devchat/ActionHandlerFactory.kt
@@ -25,6 +25,7 @@ class ActionHandlerFactory {
         DevChatActions.COMMIT_CODE_REQUEST to CommitCodeRequestHandler::class,
         DevChatActions.GET_SETTING_REQUEST to GetSettingRequestHandler::class,
         DevChatActions.UPDATE_SETTING_REQUEST to UpdateSettingRequestHandler::class,
+        DevChatActions.INPUT_REQUEST to InputRequestHandler::class,
         DevChatActions.SHOW_SETTING_DIALOG_REQUEST to ShowSettingDialogRequestHandler::class,
         DevChatActions.DELETE_LAST_CONVERSATION_REQUEST to DeleteLastConversationRequestHandler::class,
         DevChatActions.DELETE_TOPIC_REQUEST to DeleteTopicRequestHandler::class,
diff --git a/src/main/kotlin/ai/devchat/devchat/DevChatActions.kt b/src/main/kotlin/ai/devchat/devchat/DevChatActions.kt
@@ -38,6 +38,8 @@ object DevChatActions {
     const val GET_SETTING_RESPONSE = "getSetting/response"
     const val UPDATE_SETTING_REQUEST = "updateSetting/request"
     const val UPDATE_SETTING_RESPONSE = "updateSetting/response"
+    const val INPUT_REQUEST = "input/request"
+    const val INPUT_RESPONSE = "input/response"
     const val SHOW_SETTING_DIALOG_REQUEST = "showSettingDialog/request"
     const val DELETE_LAST_CONVERSATION_REQUEST = "deleteLastConversation/request"
     const val DELETE_LAST_CONVERSATION_RESPONSE = "deleteLastConversation/response"
diff --git a/src/main/kotlin/ai/devchat/devchat/handler/InputRequestHandler.kt b/src/main/kotlin/ai/devchat/devchat/handler/InputRequestHandler.kt
@@ -0,0 +1,18 @@
+package ai.devchat.devchat.handler
+
+import ai.devchat.cli.DevChatWrapper
+import ai.devchat.devchat.BaseActionHandler
+import ai.devchat.devchat.DevChatActions
+import com.alibaba.fastjson.JSONObject
+import kotlinx.coroutines.runBlocking
+
+class InputRequestHandler(metadata: JSONObject?, payload: JSONObject?) : BaseActionHandler(metadata, payload) {
+    override val actionName: String = DevChatActions.INPUT_RESPONSE
+
+    override fun action() {
+        runBlocking {
+            DevChatWrapper.activeChannel?.send(payload!!.getString("data"))
+        }
+        send()
+    }
+}
diff --git a/src/main/kotlin/ai/devchat/devchat/handler/SendMessageRequestHandler.kt b/src/main/kotlin/ai/devchat/devchat/handler/SendMessageRequestHandler.kt
@@ -64,20 +64,13 @@ class SendMessageRequestHandler(metadata: JSONObject?, payload: JSONObject?) : B
         wrapper.route(
             flags,
             message,
-            {line ->
+            callback = {line ->
                 response.update(line)
                 promptCallback(response)
             },
-            { _ ->
-                insertLog(
-                    contextJSONs,
-                    model.takeIf { it.isNotEmpty() } ?: DevChatSettingsState.instance.defaultModel,
-                    message,
-                    response.message ?: "",
-                    parent
-                )
-                val lastRecord = wrapper.logLast()
-                response.update("prompt ${lastRecord!!["hash"]}")
+            onFinish = { _ ->
+                val record = insertLog(contextJSONs, model, message, response.message, parent)
+                response.update("prompt ${record["hash"]}")
                 promptCallback(response)
 
                 val currentTopic = ActiveConversation.topic ?: response.promptHash!!
@@ -136,13 +129,14 @@ class SendMessageRequestHandler(metadata: JSONObject?, payload: JSONObject?) : B
 
     private fun insertLog(
         contexts: List<String>?,
-        model: String,
+        model: String?,
         request: String,
-        response: String,
+        response: String?,
         parent: String?
-    ) {
-        val item = mutableMapOf<String, Any?>(
-            "model" to model,
+    ): JSONObject {
+        val defaultModel = DevChatSettingsState.instance.defaultModel
+        val item = mutableMapOf(
+            "model" to if (model.isNullOrEmpty()) defaultModel else model,
             "messages" to listOf(
                 mutableMapOf(
                     "role" to "user",
@@ -163,6 +157,9 @@ class SendMessageRequestHandler(metadata: JSONObject?, payload: JSONObject?) : B
         )
         parent?.let {item.put("parent", parent)}
         wrapper.logInsert(JSONObject(item).toJSONString())
+        val lastRecord = wrapper.logLast()
+        Log.info("Log item inserted: ${lastRecord!!["hash"]}")
+        return lastRecord
     }
 
 }
diff --git a/src/main/kotlin/ai/devchat/idea/DevChatSetupThread.kt b/src/main/kotlin/ai/devchat/idea/DevChatSetupThread.kt
@@ -58,6 +58,7 @@ class DevChatSetupThread : Thread() {
             ?.let {
                 val workflowEnv = envManager.createEnv("devchat-commands", defaultPythonVersion)
                 workflowEnv.installRequirements(it)
+                PathUtils.pythonForWorkflows = workflowEnv.pythonCommand
             }
     }
 

Original file line number	Diff line number	Diff line change
`@@ -58,6 +58,7 @@ class DevChatSetupThread : Thread() {`
`58`	`58`	`?.let {`
`59`	`59`	`val workflowEnv = envManager.createEnv("devchat-commands", defaultPythonVersion)`
`60`	`60`	`workflowEnv.installRequirements(it)`
	`61`	`+ PathUtils.pythonForWorkflows = workflowEnv.pythonCommand`
`61`	`62`	`}`
`62`	`63`	`}`
`63`	`64`