Coverage for cli / commands / agent.py: 16%

1""" 

2/agent 命令处理模块 

3""" 

4import shutil 

5from rich.console import Console 

6from rich.table import Table 

7from qrclaw.workspace import Workspace, list_agents, AGENTS_ROOT 

8from qrclaw.memory.session import Session 

9from qrclaw.logger import setup_logger 

10from qrclaw.config import LOG_LEVEL, LOG_MAX_DAYS, LOG_TO_FILE, LOG_TO_CONSOLE, LOG_CONSOLE_LEVEL 

11 

12 

13def handle(args: str, current_workspace: Workspace, current_session: Session, console: Console) -> tuple[Workspace, Session]: 

14 """ 

15 处理 /agent 子命令。 

16 返回 (workspace, session)，切换 agent 时两者都会更新。 

17 

18 子命令： 

19 list 列出所有顶级 agent 

20 new <id> 新建并切换 

21 switch <id> 切换到已有 agent 

22 delete <id> 删除指定 agent（不能删当前） 

23 """ 

24 parts = args.strip().split(maxsplit=1) 

25 

26 if not parts or not parts[0]: 

27 _print_help(console) 

28 return current_workspace, current_session 

29 

30 subcommand = parts[0] 

31 sub_args = parts[1].strip() if len(parts) > 1 else "" 

32 

33 if subcommand == "list": 

34 return _cmd_list(current_workspace, console) 

35 elif subcommand == "new": 

36 return _cmd_new(sub_args, current_workspace, current_session, console) 

37 elif subcommand == "switch": 

38 return _cmd_switch(sub_args, current_workspace, current_session, console) 

39 elif subcommand == "delete": 

40 return _cmd_delete(sub_args, current_workspace, current_session, console) 

41 else: 

42 console.print(f"[red]未知子命令: {subcommand}[/red]") 

43 _print_help(console) 

44 return current_workspace, current_session 

45 

46 

47def _cmd_list(current_workspace: Workspace, console: Console) -> tuple[Workspace, Session]: 

48 agents = list_agents() 

49 if not agents: 

50 console.print("[yellow]没有已创建的 agent[/yellow]") 

51 return current_workspace, None 

52 

53 table = Table(title="Agent 列表") 

54 table.add_column("Agent ID", style="cyan") 

55 table.add_column("路径", style="dim") 

56 table.add_column("", style="bold yellow") 

57 

58 for agent_id in agents: 

59 marker = "← 当前" if agent_id == current_workspace.agent_id else "" 

60 table.add_row(agent_id, str(AGENTS_ROOT / agent_id), marker) 

61 

62 console.print(table) 

63 return current_workspace, None 

64 

65 

66def _cmd_new(agent_id: str, current_workspace: Workspace, current_session: Session, console: Console) -> tuple[Workspace, Session]: 

67 if not agent_id: 

68 console.print("[red]用法: /agent new <agentID>[/red]") 

69 return current_workspace, current_session 

70 

71 new_workspace = Workspace(agent_id=agent_id) 

72 new_session = _switch_to(new_workspace, console) 

73 console.print(f"[bold cyan]已新建并切换到 agent: {agent_id}[/bold cyan]") 

74 return new_workspace, new_session 

75 

76 

77def _cmd_switch(agent_id: str, current_workspace: Workspace, current_session: Session, console: Console) -> tuple[Workspace, Session]: 

78 if not agent_id: 

79 console.print("[red]用法: /agent switch <agentID>[/red]") 

80 return current_workspace, current_session 

81 

82 if agent_id == current_workspace.agent_id: 

83 console.print(f"[yellow]已经在 agent: {agent_id}[/yellow]") 

84 return current_workspace, current_session 

85 

86 target_workspace = Workspace(agent_id=agent_id) 

87 new_session = _switch_to(target_workspace, console) 

88 console.print(f"[bold cyan]已切换到 agent: {agent_id}[/bold cyan]") 

89 return target_workspace, new_session 

90 

91 

92def _cmd_delete(agent_id: str, current_workspace: Workspace, current_session: Session, console: Console) -> tuple[Workspace, Session]: 

93 if not agent_id: 

94 console.print("[red]用法: /agent delete <agentID>[/red]") 

95 return current_workspace, current_session 

96 

97 if agent_id == current_workspace.agent_id: 

98 console.print("[red]不能删除当前正在使用的 agent，请先切换到其他 agent[/red]") 

99 return current_workspace, current_session 

100 

101 agent_root = AGENTS_ROOT / agent_id 

102 if not agent_root.exists(): 

103 console.print(f"[yellow]agent {agent_id} 不存在[/yellow]") 

104 return current_workspace, current_session 

105 

106 shutil.rmtree(agent_root) 

107 console.print(f"[dim]agent {agent_id} 已删除[/dim]") 

108 return current_workspace, current_session 

109 

110 

111def _switch_to(workspace: Workspace, console: Console) -> Session: 

112 """切换到指定 workspace，重建日志和 session。""" 

113 setup_logger( 

114 session_id="init", 

115 log_level=LOG_LEVEL, 

116 log_to_file=LOG_TO_FILE, 

117 log_to_console=LOG_TO_CONSOLE, 

118 log_max_days=LOG_MAX_DAYS, 

119 console_level=LOG_CONSOLE_LEVEL, 

120 log_dir=workspace.logs_dir, 

121 ) 

122 new_session = Session(sessions_dir=workspace.sessions_dir) 

123 # 用真实 session_id 重建日志 

124 setup_logger( 

125 session_id=new_session.session_id, 

126 log_level=LOG_LEVEL, 

127 log_to_file=LOG_TO_FILE, 

128 log_to_console=LOG_TO_CONSOLE, 

129 log_max_days=LOG_MAX_DAYS, 

130 console_level=LOG_CONSOLE_LEVEL, 

131 log_dir=workspace.logs_dir, 

132 ) 

133 if new_session.messages: 

134 console.print(f"[dim]已加载历史会话，共 {len(new_session.messages)} 条消息[/dim]") 

135 return new_session 

136 

137 

138def _print_help(console: Console) -> None: 

139 console.print("[dim]用法: /agent <list|new|switch|delete> [agentID][/dim]") 

140 console.print("[dim] list 列出所有 agent[/dim]") 

141 console.print("[dim] new <id> 新建 agent[/dim]") 

142 console.print("[dim] switch <id> 切换 agent[/dim]") 

143 console.print("[dim] delete <id> 删除 agent[/dim]")