Coverage for little_loops / config / orchestration.py: 88%

1"""Orchestration configuration dataclass. 

2 

3Covers host CLI selection and related orchestration settings. 

4""" 

5 

6from __future__ import annotations 

7 

8from dataclasses import dataclass, field 

9from typing import Any 

10 

11 

12@dataclass 

13class ComposerAdaptiveConfig: 

14 """Tuning knobs for the adaptive loop-composer-adaptive built-in loop.""" 

15 

16 enabled: bool = False 

17 max_replans: int = 2 

18 reassess_min_confidence: float = 0.6 

19 

20 @classmethod 

21 def from_dict(cls, data: dict[str, Any]) -> ComposerAdaptiveConfig: 

22 """Create ComposerAdaptiveConfig from dictionary.""" 

23 return cls( 

24 enabled=data.get("enabled", False), 

25 max_replans=data.get("max_replans", 2), 

26 reassess_min_confidence=data.get("reassess_min_confidence", 0.6), 

27 ) 

28 

29 

30@dataclass 

31class ClusterConfig: 

32 """Settings for the goal-cluster multi-goal orchestration loop.""" 

33 

34 max_batch_size: int = 5 

35 enable_dedup: bool = True 

36 propagate_context: bool = True 

37 

38 @classmethod 

39 def from_dict(cls, data: dict[str, Any]) -> ClusterConfig: 

40 """Create ClusterConfig from dictionary.""" 

41 return cls( 

42 max_batch_size=data.get("max_batch_size", 5), 

43 enable_dedup=data.get("enable_dedup", True), 

44 propagate_context=data.get("propagate_context", True), 

45 ) 

46 

47 

48@dataclass 

49class ComposerConfig: 

50 """Settings for the loop-composer built-in orchestration loop.""" 

51 

52 adaptive: ComposerAdaptiveConfig = field(default_factory=ComposerAdaptiveConfig) 

53 

54 @classmethod 

55 def from_dict(cls, data: dict[str, Any]) -> ComposerConfig: 

56 """Create ComposerConfig from dictionary.""" 

57 return cls( 

58 adaptive=ComposerAdaptiveConfig.from_dict(data.get("adaptive", {})), 

59 ) 

60 

61 

62@dataclass 

63class OrchestrationConfig: 

64 """Orchestration settings, primarily host CLI selection. 

65 

66 ``host_cli`` mirrors the ``LL_HOST_CLI`` environment variable and is used 

67 by :func:`~little_loops.host_runner.apply_host_cli_from_config` to export 

68 the config value into the environment before :func:`~little_loops.host_runner.resolve_host` 

69 runs. The env var takes precedence if already set. 

70 """ 

71 

72 host_cli: str | None = None 

73 composer: ComposerConfig = field(default_factory=ComposerConfig) 

74 cluster: ClusterConfig = field(default_factory=ClusterConfig) 

75 

76 @classmethod 

77 def from_dict(cls, data: dict[str, Any]) -> OrchestrationConfig: 

78 """Create OrchestrationConfig from dictionary.""" 

79 return cls( 

80 host_cli=data.get("host_cli"), 

81 composer=ComposerConfig.from_dict(data.get("composer", {})), 

82 cluster=ClusterConfig.from_dict(data.get("cluster", {})), 

83 )