Coverage for src/keel/project_commands.py: 100%

41 statements  

« prev     ^ index     » next       coverage.py v7.16.2, created at 2026-10-02 20:26 +0000

1"""Project-provided command declarations. 

2 

3Project commands are durable policy data: keel can discover, list, plan, and capability-check 

4them, but their implementation remains in the consumer repository. 

5""" 

6 

7from __future__ import annotations 

8 

9from dataclasses import dataclass 

10from typing import Any 

11 

12from .config import ProjectConfig 

13 

14 

15@dataclass(frozen=True) 

16class ProjectCommand: 

17 """One project-owned command exposed through ``policy_pack``.""" 

18 

19 name: str 

20 command: str | None = None 

21 description: str | None = None 

22 agent_role: str | None = None 

23 paths: tuple[str, ...] = () 

24 required_capabilities: tuple[str, ...] = () 

25 optional_capabilities: tuple[str, ...] = () 

26 side_effects: tuple[str, ...] = () 

27 dry_run_safe: bool = False 

28 source: str = "policy_pack.project_commands" 

29 

30 def as_dict(self) -> dict[str, Any]: 

31 """Render as JSON-compatible contract data.""" 

32 return { 

33 "name": self.name, 

34 "command": self.command, 

35 "description": self.description, 

36 "agent_role": self.agent_role, 

37 "paths": list(self.paths), 

38 "required_capabilities": list(self.required_capabilities), 

39 "optional_capabilities": list(self.optional_capabilities), 

40 "side_effects": list(self.side_effects), 

41 "dry_run_safe": self.dry_run_safe, 

42 "source": self.source, 

43 } 

44 

45 

46def list_project_commands(config: ProjectConfig) -> tuple[ProjectCommand, ...]: 

47 """Return project-owned commands declared by the policy pack.""" 

48 pack = config.policy_pack 

49 commands: list[ProjectCommand] = [] 

50 commands.extend( 

51 _commands_from_map(pack.get("project_commands", {}), source="policy_pack.project_commands") 

52 ) 

53 legacy_names = {cmd.name for cmd in commands} 

54 for command in _commands_from_map( 

55 pack.get("command_routing", {}), source="policy_pack.command_routing" 

56 ): 

57 if command.name not in legacy_names: 

58 commands.append(command) 

59 return tuple(sorted(commands, key=lambda cmd: cmd.name)) 

60 

61 

62def get_project_command(config: ProjectConfig, name: str) -> ProjectCommand | None: 

63 """Return one project command by name, or ``None`` when absent.""" 

64 for command in list_project_commands(config): 

65 if command.name == name: 

66 return command 

67 return None 

68 

69 

70def _commands_from_map(value: Any, *, source: str) -> list[ProjectCommand]: 

71 if not isinstance(value, dict): 

72 return [] 

73 commands: list[ProjectCommand] = [] 

74 for name, raw in value.items(): 

75 if not isinstance(raw, dict): 

76 continue 

77 commands.append( 

78 ProjectCommand( 

79 name=name, 

80 command=raw.get("command"), 

81 description=raw.get("description"), 

82 agent_role=raw.get("agent_role"), 

83 paths=tuple(raw.get("paths", [])), 

84 required_capabilities=tuple(raw.get("required_capabilities", [])), 

85 optional_capabilities=tuple(raw.get("optional_capabilities", [])), 

86 side_effects=tuple(raw.get("side_effects", [])), 

87 dry_run_safe=bool(raw.get("dry_run_safe", False)), 

88 source=source, 

89 ) 

90 ) 

91 return commands