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

1"""The pure run planner: lay the project's gates/extensions onto the backbone. 

2 

3``build_plan`` is deterministic and I/O-free — it answers "what would run, in what 

4order, for this project?" by mapping the planned gates onto their backbone steps. 

5This is exactly what a dry-run / ``keel plan`` shows, and what config-injection 

6tests assert against (the right project values appear, no foreign ones leak). 

7 

8Actually *executing* the plan (git, gh, agent dispatch) is the thin I/O layer; the 

9ordering and composition live here, so they stay testable and reproducible. 

10""" 

11 

12from __future__ import annotations 

13 

14from dataclasses import dataclass 

15 

16from . import gates, model 

17from .config import ProjectConfig 

18from .extensions import Extension 

19 

20 

21@dataclass(frozen=True) 

22class PlanHook: 

23 """One resolved extension hook rendered in step order.""" 

24 

25 slot: str 

26 id: str 

27 kind: str 

28 on_fail: str 

29 execution_mode: str 

30 adapter_required: bool 

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

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

33 

34 

35@dataclass(frozen=True) 

36class PlanSlot: 

37 """One add-only extension slot exposed by a backbone step.""" 

38 

39 name: str 

40 step_id: str 

41 execution_mode: str 

42 may_block: bool 

43 adapter_required: bool 

44 hook_count: int 

45 

46 

47@dataclass(frozen=True) 

48class PlanItem: 

49 """One backbone step plus the gate ids that execute at it.""" 

50 

51 step_id: str 

52 step_name: str 

53 agentic: bool 

54 gates: tuple[str, ...] = () 

55 extension_slots: tuple[PlanSlot, ...] = () 

56 hooks: tuple[PlanHook, ...] = () 

57 

58 

59def build_plan(config: ProjectConfig, loaded: dict[str, list[Extension]]) -> tuple[PlanItem, ...]: 

60 """Map the planned gates onto the fixed backbone (deterministic).""" 

61 specs = gates.plan_gates(config, loaded) 

62 test_gates = tuple(s.id for s in specs if s.phase == "test") 

63 pre_merge_gates = tuple(s.id for s in specs if s.phase == "pre-merge") 

64 

65 items: list[PlanItem] = [] 

66 for step in model.BACKBONE: 

67 if step.name == "guard": 

68 step_gates = tuple(s.id for s in specs if s.phase == "guard") 

69 elif step.name == "test": 

70 step_gates = test_gates 

71 elif step.name == "merge": 

72 step_gates = pre_merge_gates 

73 else: 

74 step_gates = () 

75 items.append(PlanItem( 

76 step.id, 

77 step.name, 

78 step.agentic, 

79 step_gates, 

80 _slots_for_step(step.id, loaded), 

81 _hooks_for_step(step.id, loaded), 

82 )) 

83 return tuple(items) 

84 

85 

86def render_plan(config: ProjectConfig, plan: tuple[PlanItem, ...]) -> str: 

87 """Render a plan as a stable, human-readable tree (used by dry-run / CLI).""" 

88 repo = config.repo or "(repo)" 

89 lines = [ 

90 f"keel plan — {repo}", 

91 f" base_branch: {config.base_branch} core_version: {config.core_version}", 

92 " backbone:", 

93 ] 

94 for item in plan: 

95 marker = " [agent]" if item.agentic else "" 

96 lines.append(f" {item.step_id:>3} {item.step_name}{marker}") 

97 for gate in item.gates: 

98 lines.append(f" - gate: {gate}") 

99 for hook in item.hooks: 

100 caps = _capability_summary(hook) 

101 adapter = " adapter-required" if hook.adapter_required else "" 

102 lines.append( 

103 f" - hook: {hook.id} [{hook.slot}; {hook.kind}; " 

104 f"on_fail={hook.on_fail}; mode={hook.execution_mode}{adapter}{caps}]" 

105 ) 

106 return "\n".join(lines) 

107 

108 

109def plan_as_dict(plan: tuple[PlanItem, ...]) -> list[dict]: 

110 """Render a plan as plain data for JSON output.""" 

111 return [ 

112 { 

113 "step_id": item.step_id, 

114 "step_name": item.step_name, 

115 "agentic": item.agentic, 

116 "gates": list(item.gates), 

117 "extension_slots": [ 

118 { 

119 "name": slot.name, 

120 "step_id": slot.step_id, 

121 "execution_mode": slot.execution_mode, 

122 "may_block": slot.may_block, 

123 "adapter_required": slot.adapter_required, 

124 "hook_count": slot.hook_count, 

125 "customization": "add-only", 

126 "failure_mode": "blocking-capable" if slot.may_block else "fail-soft", 

127 } 

128 for slot in item.extension_slots 

129 ], 

130 "hooks": [ 

131 { 

132 "slot": hook.slot, 

133 "id": hook.id, 

134 "kind": hook.kind, 

135 "on_fail": hook.on_fail, 

136 "execution_mode": hook.execution_mode, 

137 "adapter_required": hook.adapter_required, 

138 "required_capabilities": list(hook.required_capabilities), 

139 "optional_capabilities": list(hook.optional_capabilities), 

140 } 

141 for hook in item.hooks 

142 ], 

143 } 

144 for item in plan 

145 ] 

146 

147 

148def _slots_for_step(step_id: str, loaded: dict[str, list[Extension]]) -> tuple[PlanSlot, ...]: 

149 return tuple( 

150 PlanSlot( 

151 name=slot.name, 

152 step_id=slot.step_id, 

153 execution_mode=slot.execution_mode, 

154 may_block=slot.may_block, 

155 adapter_required=slot.adapter_required, 

156 hook_count=len(loaded.get(slot.name, [])), 

157 ) 

158 for slot in model.slots_for_step(step_id) 

159 ) 

160 

161 

162def _hooks_for_step(step_id: str, loaded: dict[str, list[Extension]]) -> tuple[PlanHook, ...]: 

163 hooks: list[PlanHook] = [] 

164 for slot in model.slots_for_step(step_id): 

165 for ext in loaded.get(slot.name, []): 

166 hooks.append(PlanHook( 

167 slot=slot.name, 

168 id=ext.id, 

169 kind=ext.kind, 

170 on_fail=ext.on_fail, 

171 execution_mode=ext.mode, 

172 adapter_required=slot.adapter_required or ext.kind == "agentic", 

173 required_capabilities=ext.required_capabilities, 

174 optional_capabilities=ext.optional_capabilities, 

175 )) 

176 return tuple(hooks) 

177 

178 

179def _capability_summary(hook: PlanHook) -> str: 

180 parts = [] 

181 if hook.required_capabilities: 

182 parts.append("required=" + ",".join(hook.required_capabilities)) 

183 if hook.optional_capabilities: 

184 parts.append("optional=" + ",".join(hook.optional_capabilities)) 

185 return "; " + "; ".join(parts) if parts else ""