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( 

60 config: ProjectConfig, 

61 loaded: dict[str, list[Extension]], 

62 *, 

63 implement_mode: str | None = None, 

64) -> tuple[PlanItem, ...]: 

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

66 

67 ``implement_mode`` is threaded to :func:`keel.gates.plan_gates` so a run selecting the 

68 test-first s4 profile with ``--tdd`` sees its ``tdd-order`` gate in the rendered plan, 

69 rather than a plan that disagrees with the gates the same run will execute. 

70 """ 

71 specs = gates.plan_gates(config, loaded, implement_mode=implement_mode) 

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

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

74 

75 items: list[PlanItem] = [] 

76 for step in model.BACKBONE: 

77 if step.name == "guard": 

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

79 elif step.name == "test": 

80 step_gates = test_gates 

81 elif step.name == "merge": 

82 step_gates = pre_merge_gates 

83 else: 

84 step_gates = () 

85 items.append( 

86 PlanItem( 

87 step.id, 

88 step.name, 

89 step.agentic, 

90 step_gates, 

91 _slots_for_step(step.id, loaded), 

92 _hooks_for_step(step.id, loaded), 

93 ) 

94 ) 

95 return tuple(items) 

96 

97 

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

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

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

101 lines = [ 

102 f"keel plan — {repo}", 

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

104 " backbone:", 

105 ] 

106 for item in plan: 

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

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

109 for gate in item.gates: 

110 # The one built-in gate whose command can be absent (#1328): say so here, 

111 # where an operator reads the plan, rather than first in a blocked run. 

112 unset = gate == "build" and not config.knobs.build_gate_cmd 

113 note = " (not configured: set knobs.build_gate_cmd)" if unset else "" 

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

115 for hook in item.hooks: 

116 caps = _capability_summary(hook) 

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

118 lines.append( 

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

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

121 ) 

122 return "\n".join(lines) 

123 

124 

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

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

127 return [ 

128 { 

129 "step_id": item.step_id, 

130 "step_name": item.step_name, 

131 "agentic": item.agentic, 

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

133 "extension_slots": [ 

134 { 

135 "name": slot.name, 

136 "step_id": slot.step_id, 

137 "execution_mode": slot.execution_mode, 

138 "may_block": slot.may_block, 

139 "adapter_required": slot.adapter_required, 

140 "hook_count": slot.hook_count, 

141 "customization": "add-only", 

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

143 } 

144 for slot in item.extension_slots 

145 ], 

146 "hooks": [ 

147 { 

148 "slot": hook.slot, 

149 "id": hook.id, 

150 "kind": hook.kind, 

151 "on_fail": hook.on_fail, 

152 "execution_mode": hook.execution_mode, 

153 "adapter_required": hook.adapter_required, 

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

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

156 } 

157 for hook in item.hooks 

158 ], 

159 } 

160 for item in plan 

161 ] 

162 

163 

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

165 return tuple( 

166 PlanSlot( 

167 name=slot.name, 

168 step_id=slot.step_id, 

169 execution_mode=slot.execution_mode, 

170 may_block=slot.may_block, 

171 adapter_required=slot.adapter_required, 

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

173 ) 

174 for slot in model.slots_for_step(step_id) 

175 ) 

176 

177 

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

179 hooks: list[PlanHook] = [] 

180 for slot in model.slots_for_step(step_id): 

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

182 hooks.append( 

183 PlanHook( 

184 slot=slot.name, 

185 id=ext.id, 

186 kind=ext.kind, 

187 on_fail=ext.on_fail, 

188 execution_mode=ext.mode, 

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

190 required_capabilities=ext.required_capabilities, 

191 optional_capabilities=ext.optional_capabilities, 

192 ) 

193 ) 

194 return tuple(hooks) 

195 

196 

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

198 parts = [] 

199 if hook.required_capabilities: 

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

201 if hook.optional_capabilities: 

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

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