apps\hub\src\main\java\cn\nfarms\hub\engine\Harness.kt

源码路径:apps\hub\src\main\java\cn\nfarms\hub\engine\Harness.kt · 162 行

package cn.nfarms.hub.engine

import cn.nfarms.hub.task.HubJob
import cn.nfarms.hub.task.JobCtl
import java.io.File
import java.util.concurrent.TimeUnit

/**
 * 「ds harness」调度:拉起一个本地进程、把它的输出留在内存里、可停。
 *
 * 为什么不并进 [cn.nfarms.hub.task.TaskScheduler]:那个调度器管的是
 * 「跑完就结束」的作业,而 harness 是**常驻**的(进程一直活着等命令)。
 * 两者的生命周期完全不同,塞进同一个状态机会让「已结束」这个字段永远为 null。
 *
 * # 这个进程是什么
 *
 * 桌面版的 nfarms 拉的是 `dsh`(DeepSeek Harness 的 CLI),它会在 stdout 打印
 * 带一次性 token 的访问地址 —— **地址必须从输出里抓,不能自己拼**
 * (拼出来的没 token,打开只会 401)。手机上默认没有 dsh,
 * 所以这里默认跑一个**演示进程**(见 [DemoProcess]):行为完全一样(打印根地址 + 持续输出),
 * 只是不做真事 —— 这样「启动/看日志/停止」这条链路能真的跑通、也真的能测。
 */
class HarnessRunner(private val workDir: File) {

    class Session(
        val process: Process,
        val port: Int,
        val startedAt: Long = System.currentTimeMillis(),
    ) {
        @Volatile var url: String? = null
        @Volatile var alive: Boolean = true
        val lines = ArrayDeque<String>()

        fun append(line: String) = synchronized(lines) {
            lines.addLast(line)
            if (lines.size > 400) lines.removeFirst()
        }

        fun snapshot(): List<String> = synchronized(lines) { lines.toList() }

        /** 从输出里抓根地址:带一次性 token 的那个。 */
        fun captureUrl(line: String) {
            val m = Regex("""(https?://[^\s"']+)""").find(line) ?: return
            val u = m.value
            if (u.contains("token=", ignoreCase = true) || u.contains("127.0.0.1")) url = u
        }
    }

    @Volatile var current: Session? = null
        private set

    fun start(port: Int, ctl: JobCtl): Session {
        stop()
        val cmd = resolveCommand(port)
        val pb = ProcessBuilder(cmd).directory(workDir).redirectErrorStream(true)
        val proc = pb.start()
        val session = Session(proc, port)
        current = session

        // 把 stdout 一路读进环形缓冲,并抓根地址
        Thread {
            proc.inputStream.bufferedReader().forEachLine { line ->
                session.append(line)
                session.captureUrl(line)
            }
            session.alive = false
        }.apply { isDaemon = true; name = "harness-out"; start() }

        // 等它打印出地址(最多 8 秒);没打印也照样返回 —— 界面上会显示「还没打印根地址」
        val deadline = System.currentTimeMillis() + 8_000
        while (System.currentTimeMillis() < deadline && session.url == null && proc.isAlive) {
            ctl.throwIfCancelled()
            Thread.sleep(120)
        }
        if (!proc.isAlive) {
            val tail = session.snapshot().joinToString("\n")
            session.alive = false
            throw IllegalStateException("进程一启动就退出了(退出码 ${proc.exitValue()})\n$tail")
        }
        return session
    }

    fun stop() {
        val s = current ?: return
        runCatching { s.process.destroy() }
        runCatching { s.process.waitFor(2, TimeUnit.SECONDS) }
        if (s.process.isAlive) runCatching { s.process.destroyForcibly() }
        s.alive = false
        current = null
    }

    /**
     * 有 `dsh` 就用真的,没有就走演示进程。
     *
     * 这个「降级」是**显式**的:界面会显示当前跑的是哪一个,
     * 免得用户以为自己在用真 harness。
     */
    fun resolveCommand(port: Int): List<String> {
        val dsh = sequenceOf(File(workDir, "bin/dsh"), File("/data/local/tmp/dsh"), File("/system/bin/dsh"))
            .firstOrNull { it.canExecute() }
        if (dsh != null) return listOf(dsh.absolutePath, "web", "--port", port.toString())
        // 演示脚本用系统 sh。`/system/bin/sh` 在所有 Android 上都存在;
        // 万一某个精简系统没有,退到 `/bin/sh`(模拟器与桌面 Linux 上有)。
        val sh = sequenceOf("/system/bin/sh", "/bin/sh", "sh").first { File(it).exists() || it == "sh" }
        return listOf(sh, "-c", demoScript(port))
    }

    fun isRealHarness(): Boolean =
        sequenceOf(File(workDir, "bin/dsh"), File("/data/local/tmp/dsh"), File("/system/bin/dsh"))
            .any { it.canExecute() }

    /**
     * 演示脚本:打印一行「带 token 的根地址」,然后每 2 秒吐一行心跳。
     *
     * # 为什么心跳用 `sleep` 循环拼成这样
     *
     * 这条命令是经 `sh -c` 执行的,而 `sh -c` 的参数**没有引号保护** ——
     * Kotlin 里的 `\$` 只是普通字符,不会被 shell 再转义一次。所以这里
     * 不要写 `while true; do ... done`(它依赖 `;` 与 `$` 的精确转义,
     * 少一个就整个脚本语法错),改用最朴素的 `: ...; sleep 2; ...` 串,
     * 只依赖分号 —— 那是 `sh` 里最不可能出意外的分隔符。
     */
    private fun demoScript(port: Int): String {
        val token = (100000..999999).random()
        // 每 2 秒一行,循环 5 次后继续用 `exec sleep` 挂住不退出太复杂,
        // 这里干脆只打 5 行心跳就退出 —— 进程活着与否由"启动时是否拿到地址"判定,
        // 而"持续运行"这件事在演示里并不需要真的常驻。
        return buildString {
            append("echo 'ds harness (demo) starting on port $port'; ")
            append("echo 'root: http://127.0.0.1:$port/?token=$token'; ")
            for (i in 1..5) {
                append("sleep 2; echo '[heartbeat] tick=$i port=$port'; ")
            }
            append("sleep 600")
        }
    }
}

/**
 * 「启动 harness」的 HUB 任务。
 *
 * 任务本身**秒结束**(它只负责把进程拉起来并把状态交出去),
 * 常驻进程由 [HarnessRunner] 持有 —— 这样的好处是任务历史里有一条
 * 「启动过」,而停止不需要一条任务。
 */
class HarnessStartJob(
    private val runner: HarnessRunner,
    private val port: Int,
    private val onStarted: (HarnessRunner.Session) -> Unit,
) : HubJob {
    override val kind: String = "harness"
    override val title: String = "启动 ds harness"
    override val detail: String = "端口 $port"

    override fun run(ctl: JobCtl) {
        ctl.progress(0.2f, "拉起进程")
        val session = runner.start(port, ctl)
        ctl.progress(0.9f, "已拿到根地址")
        onStarted(session)
    }
}