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

1"""The keel backbone: the fixed, ordered step machine and its extension slots. 

2 

3This module is the single source of truth for the step IDs, the named slots an 

4extension may register into, and the invariants the backbone always preserves. 

5It is pure data — no I/O, no config — so consumers (config, extensions, 

6orchestrator) and tests can all agree on one definition. 

7""" 

8 

9from __future__ import annotations 

10 

11from dataclasses import dataclass 

12 

13 

14@dataclass(frozen=True) 

15class Step: 

16 """One backbone step. 

17 

18 ``slot`` is the historical primary extension point, kept for compatibility. 

19 New hook-aware code should use :func:`slots_for_step`. 

20 """ 

21 

22 id: str 

23 name: str 

24 slot: str | None = None 

25 agentic: bool = False # True if the step dispatches to an agent (implement/review/classify) 

26 

27 

28@dataclass(frozen=True) 

29class Slot: 

30 """One named extension hook exposed by a backbone step.""" 

31 

32 name: str 

33 step_id: str 

34 execution_mode: str = "deterministic" 

35 may_block: bool = False 

36 adapter_required: bool = False 

37 

38 

39#: The fixed backbone, in execution order. Changing this is a keel-core change. 

40BACKBONE: tuple[Step, ...] = ( 

41 Step("s0", "config"), 

42 Step("s1", "select"), 

43 Step("s2", "branch"), 

44 Step("s3", "guard"), 

45 Step("s4", "implement", slot="after-implement", agentic=True), 

46 Step("s5", "classify", agentic=True), 

47 Step("s6", "ci"), 

48 Step("s7", "review", slot="reviewers", agentic=True), 

49 Step("s8", "test", slot="tester"), 

50 Step("s9", "fixloop"), 

51 Step("s10", "merge", slot="pre-merge"), 

52 Step("s11", "capture", slot="post-merge"), 

53 Step("s12", "close"), 

54) 

55 

56#: The named hooks, in backbone order. Extensions are add-only into these. 

57SLOT_DEFINITIONS: tuple[Slot, ...] = ( 

58 Slot("after:config", "s0"), 

59 Slot("before:select", "s1"), 

60 Slot("select", "s1", adapter_required=True), 

61 Slot("after:select", "s1"), 

62 Slot("before:branch", "s2"), 

63 Slot("after:branch", "s2"), 

64 Slot("guard", "s3", may_block=True), 

65 Slot("before:implement", "s4", adapter_required=True), 

66 Slot("after-implement", "s4", adapter_required=True), 

67 Slot("classify", "s5", adapter_required=True), 

68 Slot("after:classify", "s5"), 

69 Slot("before:ci", "s6"), 

70 Slot("after:ci", "s6"), 

71 Slot("reviewers", "s7", execution_mode="agentic", adapter_required=True), 

72 Slot("after:review", "s7", adapter_required=True), 

73 Slot("tester", "s8", execution_mode="hybrid", may_block=True, adapter_required=True), 

74 Slot("test", "s8", execution_mode="hybrid", may_block=True, adapter_required=True), 

75 Slot("after:test", "s8"), 

76 Slot("before:fixloop", "s9", adapter_required=True), 

77 Slot("fixloop", "s9", execution_mode="hybrid", adapter_required=True), 

78 Slot("after:fixloop", "s9"), 

79 Slot("pre-merge", "s10", may_block=True), 

80 Slot("after:merge", "s10"), 

81 Slot("capture", "s11", adapter_required=True), 

82 Slot("post-merge", "s11", adapter_required=True), 

83 Slot("before:close", "s12"), 

84 Slot("on-close", "s12", adapter_required=True), 

85 Slot("after:close", "s12"), 

86) 

87 

88#: The named slots, in backbone order. Extensions are add-only into these. 

89SLOTS: tuple[str, ...] = tuple(slot.name for slot in SLOT_DEFINITIONS) 

90 

91#: Invariants the backbone always preserves — no config or extension can override. 

92INVARIANTS: tuple[str, ...] = ( 

93 "merge_lock", # every merge goes through the mkdir-based lock 

94 "window_gate", # the night no-merge window is enforced 

95 "fail_soft", # a soft failure degrades to a no-op, never aborts 

96 "orchestrator_only_writes", # only the orchestrator writes to the PR 

97 "attribution", # implementer/reviewer vendor+model is recorded 

98) 

99 

100_BY_ID: dict[str, Step] = {s.id: s for s in BACKBONE} 

101_SLOT_META: dict[str, Slot] = {slot.name: slot for slot in SLOT_DEFINITIONS} 

102_BY_SLOT: dict[str, Step] = {slot: _BY_ID[_SLOT_META[slot].step_id] for slot in SLOTS} 

103 

104 

105def step_ids() -> tuple[str, ...]: 

106 """All backbone step IDs in order.""" 

107 return tuple(s.id for s in BACKBONE) 

108 

109 

110def get_step(step_id: str) -> Step: 

111 """Return the step with ``step_id`` (raises ``KeyError`` if unknown).""" 

112 return _BY_ID[step_id] 

113 

114 

115def step_for_slot(slot: str) -> Step: 

116 """Return the backbone step that exposes ``slot`` (raises ``KeyError``).""" 

117 return _BY_SLOT[slot] 

118 

119 

120def slot_meta(slot: str) -> Slot: 

121 """Return metadata for a named extension hook (raises ``KeyError``).""" 

122 return _SLOT_META[slot] 

123 

124 

125def slots_for_step(step_id: str) -> tuple[Slot, ...]: 

126 """Return extension hooks exposed by ``step_id`` in declared order.""" 

127 return tuple(slot for slot in SLOT_DEFINITIONS if slot.step_id == step_id)