feat(agents): 新增 mydsh 启动器 —— DeepSeek Harness CLI 包装

- launcher _BACKENDS 加 dsh:profile 透传、zstd 会话列表(session/title 或首条用户消息)、
  --resume 透传、DSH_HOME 环境变量与 expanduser
- entrypoints/cli/completion/打包/卸载全触点 + TestDsh 用例(裸跑/版本/透传/列表/缺 bin)
This commit is contained in:
Zhengshou Lai
2026-08-17 19:54:13 +08:00
parent f12cd5fcf1
commit f14c39d7c0
9 changed files with 301 additions and 7 deletions
+107 -1
View File
@@ -5,6 +5,7 @@ import sqlite3
from pathlib import Path
from unittest.mock import MagicMock, patch
import pytest
from click.testing import CliRunner
import myagents.launcher
@@ -29,7 +30,7 @@ class TestMyagentsHelp:
"""Tests for top-level myagents command."""
def test_help_shows_agent_subcommands(self) -> None:
"""--help should list claude, kimi, codex, hermes and cursor subcommands."""
"""--help should list claude, kimi, codex, hermes, cursor and dsh."""
runner = CliRunner()
result = runner.invoke(cli, ["--help"])
assert result.exit_code == 0
@@ -38,6 +39,7 @@ class TestMyagentsHelp:
assert "codex" in result.output
assert "hermes" in result.output
assert "cursor" in result.output
assert "dsh" in result.output
assert "update" in result.output
assert "upgrade" in result.output
@@ -692,6 +694,110 @@ class TestCursorSubcommand:
assert mock_run.call_args.kwargs.get("cwd") == str(test_dir.resolve())
class TestDshSubcommand:
"""Tests for ``myagents dsh``."""
def test_help_shows_options(self) -> None:
runner = CliRunner()
result = runner.invoke(cli, ["dsh", "--help"])
assert result.exit_code == 0
assert "--cwd" in result.output
assert "--list" in result.output
def test_runs_dsh(self) -> None:
runner = CliRunner()
with (
patch(
"myagents.launcher.shutil.which", side_effect=_backend_which("dsh")
),
patch("myagents.launcher.subprocess.run") as mock_run,
):
mock_run.return_value = MagicMock(returncode=0)
result = runner.invoke(cli, ["dsh"])
assert result.exit_code == 0
mock_run.assert_called_once()
# dsh is a profile launcher: no default args injected.
assert mock_run.call_args[0][0] == ["/usr/bin/dsh"]
def test_missing_binary_error(self) -> None:
runner = CliRunner()
with (
patch("myagents.launcher.shutil.which", return_value=None),
patch.dict("os.environ", {}, clear=False) as env,
):
env.pop("DSH_BIN", None)
result = runner.invoke(cli, ["dsh"])
assert result.exit_code == 127
assert "dsh CLI not found" in result.output
def test_cwd_option_passed(self, tmp_path: Path) -> None:
runner = CliRunner()
test_dir = tmp_path / "test_cwd"
test_dir.mkdir()
with (
patch(
"myagents.launcher.shutil.which", side_effect=_backend_which("dsh")
),
patch("myagents.launcher.subprocess.run") as mock_run,
):
mock_run.return_value = MagicMock(returncode=0)
result = runner.invoke(cli, ["dsh", "--cwd", str(test_dir)])
assert result.exit_code == 0
assert mock_run.call_args.kwargs.get("cwd") == str(test_dir.resolve())
class TestDshListSessions:
"""``mydsh -l`` reads zstd-compressed dsh session logs."""
def _write_session(self, dsh_home: Path, slug: str, sid: str, cwd: str) -> Path:
zstd = pytest.importorskip("zstandard")
log = dsh_home / "sessions" / slug / sid / "session.jsonl.zstd"
log.parent.mkdir(parents=True, exist_ok=True)
lines = [
json.dumps(
{"type": "session", "version": 0, "id": sid, "cwd": cwd}
),
json.dumps(
{
"type": "session/title",
"seq": 1,
"data": {"title": "My dsh test"},
}
),
]
with log.open("wb") as fh:
with zstd.ZstdCompressor().stream_writer(fh) as w:
w.write(("\n".join(lines) + "\n").encode("utf-8"))
return log
def test_list_shows_dsh_sessions(self, tmp_path: Path, monkeypatch) -> None:
workspace = tmp_path / "workspace"
workspace.mkdir()
dsh_home = tmp_path / "dshhome"
self._write_session(dsh_home, "slug", "sess-1", str(workspace))
monkeypatch.setenv("DSH_HOME", str(dsh_home))
runner = CliRunner()
with patch(
"myagents.launcher.shutil.which", return_value="/usr/bin/dsh"
):
result = runner.invoke(
cli, ["dsh", "-l", "--cwd", str(workspace)]
)
assert result.exit_code == 0
assert "sess-1" in result.output
assert "My dsh test" in result.output
def test_first_prompt_reads_zstd(self, tmp_path: Path) -> None:
from myagents.launcher import _first_prompt_dsh
workspace = tmp_path / "ws"
workspace.mkdir()
log = self._write_session(tmp_path / "home", "slug", "sid-1", str(workspace))
assert _first_prompt_dsh(log) == "My dsh test"
class TestCursorPassthrough:
"""Map myagents-style resume flags to Cursor Agent CLI flags."""
+44
View File
@@ -9,6 +9,7 @@ from myagents.entrypoints import (
claude_cli,
codex_cli,
cursor_cli,
dsh_cli,
hermes_cli,
kimi_cli,
)
@@ -204,6 +205,49 @@ class TestMycursorEntrypoint:
assert mock_run.call_args[0][0][-2:] == ["--resume", "abc123"]
class TestMydshEntrypoint:
"""``mydsh`` standalone entrypoint."""
def test_runs_dsh(self) -> None:
runner = CliRunner()
with (
patch("myagents.launcher.shutil.which", return_value="/usr/bin/dsh"),
patch("myagents.launcher.subprocess.run") as mock_run,
):
mock_run.return_value = MagicMock(returncode=0)
result = runner.invoke(dsh_cli, [])
assert result.exit_code == 0
# dsh is a profile launcher; no default args are injected.
assert mock_run.call_args[0][0] == ["/usr/bin/dsh"]
def test_version_shows_mydsh(self) -> None:
runner = CliRunner()
result = runner.invoke(dsh_cli, ["--version"])
assert result.exit_code == 0
assert "mydsh" in result.output
def test_passthrough(self, tmp_path: Path) -> None:
"""``--resume <id>`` passes through to the native dsh CLI."""
runner = CliRunner()
with (
patch("myagents.launcher.shutil.which", return_value="/usr/bin/dsh"),
patch("myagents.launcher.subprocess.run") as mock_run,
):
mock_run.return_value = MagicMock(returncode=0)
result = runner.invoke(
dsh_cli, ["--cwd", str(tmp_path), "--resume", "abc123"]
)
assert result.exit_code == 0
assert mock_run.call_args[0][0][-2:] == ["--resume", "abc123"]
def test_missing_binary_exits_127(self) -> None:
runner = CliRunner()
with patch("myagents.launcher.shutil.which", return_value=None):
result = runner.invoke(dsh_cli, [])
assert result.exit_code == 127
assert "dsh CLI not found" in result.output
class TestTmuxOption:
"""``--tmux`` / ``-t`` wraps the backend in an attachable tmux session."""