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

1"""Keel Canary & Automated Rollback Guard. 

2 

3Monitors post-merge health signals (CI on main branch, health probes, test gates) 

4and automatically executes an atomic git revert rollback if regression is detected. 

5""" 

6 

7from __future__ import annotations 

8 

9from dataclasses import dataclass 

10from pathlib import Path 

11from typing import Any 

12 

13from .config import load_config 

14from .swarm_runtime import SubprocessRunner, default_runner 

15 

16 

17@dataclass(frozen=True) 

18class CanaryResult: 

19 target: str 

20 passed: bool 

21 status: str 

22 health_output: str 

23 reverted: bool 

24 revert_commit: str | None = None 

25 details: str = "" 

26 

27 def to_dict(self) -> dict[str, Any]: 

28 return { 

29 "target": self.target, 

30 "passed": self.passed, 

31 "status": self.status, 

32 "health_output": self.health_output, 

33 "reverted": self.reverted, 

34 "revert_commit": self.revert_commit, 

35 "details": self.details, 

36 } 

37 

38 

39@dataclass(frozen=True) 

40class RollbackResult: 

41 target_sha: str 

42 success: bool 

43 revert_sha: str | None 

44 error: str | None = None 

45 

46 def to_dict(self) -> dict[str, Any]: 

47 return { 

48 "target_sha": self.target_sha, 

49 "success": self.success, 

50 "revert_sha": self.revert_sha, 

51 "error": self.error, 

52 } 

53 

54 

55def render_canary_result(result: CanaryResult) -> str: 

56 lines = [ 

57 f"keel canary-guard — target: {result.target}", 

58 f" status : {result.status} {'✓' if result.passed else '❌'}", 

59 ( 

60 f" reverted : yes ({result.revert_commit})" 

61 if result.reverted and result.revert_commit 

62 else f" reverted : {'yes' if result.reverted else 'no'}" 

63 ), 

64 ] 

65 if result.details: 

66 lines.append(f" details : {result.details}") 

67 if result.health_output: 

68 lines.append(f" health output : {result.health_output.strip()}") 

69 return "\n".join(lines) 

70 

71 

72def render_rollback_result(result: RollbackResult) -> str: 

73 lines = [ 

74 f"keel rollback — target: {result.target_sha}", 

75 f" status : {'success ✓' if result.success else 'failed ❌'}", 

76 ] 

77 if result.revert_sha: 

78 lines.append(f" revert commit : {result.revert_sha}") 

79 if result.error: 

80 lines.append(f" error : {result.error}") 

81 return "\n".join(lines) 

82 

83 

84def execute_rollback( 

85 target_sha: str, 

86 root: str | Path = ".", 

87 base_branch: str = "main", 

88 runner: SubprocessRunner | None = None, 

89) -> RollbackResult: 

90 """Execute an atomic revert commit for target merge commit.""" 

91 run = runner or default_runner 

92 root_path = Path(root).resolve() 

93 

94 # Attempt git revert --no-edit -m 1 <sha> or git revert --no-edit <sha> 

95 cmd = ["git", "revert", "--no-edit", "-m", "1", target_sha] 

96 res = run(cmd, root_path) 

97 if not res.ok: 

98 # Fallback to single-parent revert 

99 cmd_fallback = ["git", "revert", "--no-edit", target_sha] 

100 res_fallback = run(cmd_fallback, root_path) 

101 if not res_fallback.ok: 

102 run(["git", "revert", "--abort"], root_path) 

103 return RollbackResult( 

104 target_sha=target_sha, 

105 success=False, 

106 revert_sha=None, 

107 error=res_fallback.output.strip() or res.output.strip(), 

108 ) 

109 

110 # Get newly created commit SHA 

111 rev_res = run(["git", "rev-parse", "HEAD"], root_path) 

112 revert_sha = rev_res.output.strip() if rev_res.ok else None 

113 return RollbackResult( 

114 target_sha=target_sha, 

115 success=True, 

116 revert_sha=revert_sha, 

117 ) 

118 

119 

120def run_canary_guard( 

121 project_yaml: str, 

122 *, 

123 pr_number: int | None = None, 

124 commit_sha: str | None = None, 

125 root: str | Path = ".", 

126 duration_m: int = 1, 

127 health_cmd: str | None = None, 

128 auto_revert: bool = False, 

129 runner: SubprocessRunner | None = None, 

130) -> CanaryResult: 

131 """Run canary health checks and conditionally execute rollback.""" 

132 run = runner or default_runner 

133 root_path = Path(root).resolve() 

134 target_desc = f"PR #{pr_number}" if pr_number else (commit_sha or "HEAD") 

135 

136 # Load config to get default gates/knobs if needed 

137 cfg = load_config(project_yaml) 

138 cmd = health_cmd or getattr(cfg.knobs, "build_gate_cmd", None) or "make test" 

139 

140 # Run health command 

141 health_res = run(["sh", "-c", cmd], root_path) 

142 if health_res.ok: 

143 return CanaryResult( 

144 target=target_desc, 

145 passed=True, 

146 status="healthy", 

147 health_output=health_res.output, 

148 reverted=False, 

149 details="Canary health verification passed", 

150 ) 

151 

152 # Health check failed 

153 revert_performed = False 

154 revert_sha: str | None = None 

155 details = f"Canary failed: health check returned exit code {health_res.code}" 

156 

157 if auto_revert and commit_sha: 

158 rb_res = execute_rollback(commit_sha, root=root_path, runner=runner) 

159 if rb_res.success: 

160 revert_performed = True 

161 revert_sha = rb_res.revert_sha 

162 details += f"; automatically rolled back in {revert_sha}" 

163 else: 

164 details += f"; rollback attempt failed: {rb_res.error}" 

165 

166 return CanaryResult( 

167 target=target_desc, 

168 passed=False, 

169 status="regression_detected", 

170 health_output=health_res.output, 

171 reverted=revert_performed, 

172 revert_commit=revert_sha, 

173 details=details, 

174 )