Coverage for src/keel/swarm_landing.py: 100%
180 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-18 12:05 +0000
1"""Keel Swarm Landing — Orthogonal batch landing and drift self-healing merge engine.
3Thin I/O execution layer for evaluating branch disjointness, merging orthogonal diff trees
4under atomic merge locks, and automatically rebasing / healing drifted sequential clusters.
5"""
7from __future__ import annotations
9from collections.abc import Callable
10from pathlib import Path
12from .lock import merge_lock
13from .swarm import (
14 SwarmLandingResult,
15 SwarmPlan,
16 SwarmWave,
17 evaluate_wave_landing_mode,
18 load_swarm_state,
19 save_swarm_state,
20 update_worker_state,
21)
22from .swarm_runtime import SubprocessRunner, default_runner
25def parse_conflict_hunks(text: str) -> list[dict[str, str]]:
26 """Parse standard git conflict markers (<<<<<<<, =======, >>>>>>>) into hunks."""
27 lines = text.splitlines(keepends=True)
28 hunks: list[dict[str, str]] = []
29 in_conflict = False
30 in_theirs = False
31 ours_lines: list[str] = []
32 theirs_lines: list[str] = []
34 for line in lines:
35 if line.startswith("<<<<<<<"):
36 in_conflict = True
37 in_theirs = False
38 ours_lines = []
39 theirs_lines = []
40 elif in_conflict and line.startswith("======="):
41 in_theirs = True
42 elif in_conflict and line.startswith(">>>>>>>"):
43 in_conflict = False
44 hunks.append({
45 "ours": "".join(ours_lines),
46 "theirs": "".join(theirs_lines),
47 })
48 elif in_conflict:
49 if in_theirs:
50 theirs_lines.append(line)
51 else:
52 ours_lines.append(line)
54 return hunks
57def is_safe_declarative_chunk(lines: list[str]) -> bool:
58 """Check if lines consist entirely of safe declarative items."""
59 for line in lines:
60 stripped = line.strip()
61 if not stripped:
62 continue
63 if stripped.startswith(("import ", "from ", "#", "//", "/*", "*")):
64 continue
65 if stripped.startswith(("- ", "* ")):
66 continue
67 if (stripped.startswith('"') or stripped.startswith("'")) and stripped.endswith(
68 (",", ";", '",', "',")
69 ):
70 continue
71 return False
72 return True
75def resolve_adjacent_conflict(ours: str, theirs: str) -> str | None:
76 """Smart resolution for adjacent non-conflicting additions (e.g. imports or lists).
78 Returns ``None`` when the hunk is not safe to resolve without a human. The
79 caller writes whatever this returns and stages it, so refusing is the only
80 way a person gets to look.
82 An empty side does **not** mean "take the other one". It is what git prints
83 for a delete-versus-modify conflict, so accepting the non-empty side there
84 silently restores something the other branch deleted (#798). Both sides go
85 through the same declarative gate, empty or not.
86 """
87 ours_lines = [line for line in ours.splitlines() if line.strip()]
88 theirs_lines = [line for line in theirs.splitlines() if line.strip()]
89 if not is_safe_declarative_chunk(ours_lines) or not is_safe_declarative_chunk(
90 theirs_lines
91 ):
92 return None
93 if not ours_lines:
94 return theirs
95 if not theirs_lines:
96 return ours
97 if set(ours_lines).isdisjoint(set(theirs_lines)):
98 combined = [*ours.splitlines(), *theirs.splitlines()]
99 trailing = "\n" if (ours.endswith("\n") or theirs.endswith("\n")) else ""
100 return "\n".join(combined) + trailing
101 return None
104def resolve_conflict_content(content: str) -> str | None:
105 """Attempt deterministic self-healing on conflict-marked text. Returns resolved text or None."""
106 if "<<<<<<<" not in content or ">>>>>>>" not in content:
107 return content
109 lines = content.splitlines(keepends=True)
110 resolved_lines: list[str] = []
111 in_conflict = False
112 in_theirs = False
113 ours_lines: list[str] = []
114 theirs_lines: list[str] = []
116 for line in lines:
117 if line.startswith("<<<<<<<"):
118 in_conflict = True
119 in_theirs = False
120 ours_lines = []
121 theirs_lines = []
122 elif in_conflict and line.startswith("======="):
123 in_theirs = True
124 elif in_conflict and line.startswith(">>>>>>>"):
125 in_conflict = False
126 ours_text = "".join(ours_lines)
127 theirs_text = "".join(theirs_lines)
128 resolved = resolve_adjacent_conflict(ours_text, theirs_text)
129 if resolved is None:
130 return None
131 resolved_lines.append(resolved)
132 elif in_conflict:
133 if in_theirs:
134 theirs_lines.append(line)
135 else:
136 ours_lines.append(line)
137 else:
138 resolved_lines.append(line)
140 return "".join(resolved_lines)
143def rebase_and_heal_cluster_branch(
144 repo_root: Path,
145 branch_name: str,
146 base_branch: str = "main",
147 runner: SubprocessRunner | None = None,
148 resolver: Callable[[str], str | None] | None = None,
149) -> tuple[bool, str]:
150 """Rebase a cluster branch onto base branch, with intelligent self-healing on conflict."""
151 run = runner or default_runner
152 # Checkout branch
153 run(["git", "checkout", branch_name], repo_root)
154 # Attempt rebase
155 res = run(["git", "rebase", f"origin/{base_branch}"], repo_root)
156 if res.ok:
157 return True, "clean_rebase"
159 # Inspect status for unmerged conflict paths
160 status_res = run(["git", "status", "--porcelain"], repo_root)
161 conflict_files: list[str] = []
162 raw_status = getattr(status_res, "output", getattr(status_res, "stdout", ""))
163 for line in str(raw_status).splitlines():
164 if line.startswith("UU ") or line.startswith("AA ") or line.startswith("UD "):
165 conflict_files.append(line[3:].strip())
167 if conflict_files:
168 healed_all = True
169 resolve_fn = resolver or resolve_conflict_content
170 for rel_path in conflict_files:
171 file_path = repo_root / rel_path
172 if file_path.exists():
173 try:
174 text = file_path.read_text(encoding="utf-8")
175 resolved = resolve_fn(text)
176 if resolved is not None:
177 file_path.write_text(resolved, encoding="utf-8")
178 run(["git", "add", rel_path], repo_root)
179 else:
180 healed_all = False
181 break
182 except (OSError, UnicodeDecodeError):
183 healed_all = False
184 break
185 else:
186 healed_all = False
187 break
189 if healed_all:
190 continue_res = run(
191 ["git", "-c", "core.editor=true", "rebase", "--continue"], repo_root
192 )
193 if continue_res.ok:
194 return True, "self_healed_rebase"
196 # Fail soft: abort rebase cleanly so git workspace remains in valid state
197 run(["git", "rebase", "--abort"], repo_root)
198 return False, "conflict_detected"
201def merge_cluster_branch(
202 repo_root: Path,
203 branch_name: str,
204 base_branch: str = "main",
205 runner: SubprocessRunner | None = None,
206) -> bool:
207 """Merge a cluster branch into base branch."""
208 run = runner or default_runner
209 run(["git", "checkout", base_branch], repo_root)
210 cmd = ["git", "merge", "--no-ff", branch_name, "-m", f"Merge branch {branch_name}"]
211 res = run(cmd, repo_root)
212 return res.ok
215def land_wave_clusters(
216 plan: SwarmPlan,
217 wave_index: int,
218 project_yaml: str,
219 *,
220 root: str | Path = ".",
221 dry_run: bool = True,
222 pr_diff_map: dict[str, list[str] | tuple[str, ...]] | None = None,
223 runner: SubprocessRunner | None = None,
224 resolver: Callable[[str], str | None] | None = None,
225) -> SwarmLandingResult:
226 """Execute orthogonal batch landing or adaptive sequential funneling for a wave."""
227 root_path = Path(root).resolve()
228 target_wave: SwarmWave | None = None
229 for w in plan.waves:
230 if w.wave_index == wave_index:
231 target_wave = w
232 break
234 if target_wave is None:
235 return SwarmLandingResult(
236 swarm_id=plan.swarm_id,
237 wave_index=wave_index,
238 mode="none",
239 landed_clusters=(),
240 healed_clusters=(),
241 failed_clusters=(),
242 status="failed",
243 )
245 # Evaluate actual diff files vs predicted
246 diff_map = pr_diff_map or {}
247 decision = evaluate_wave_landing_mode(target_wave, diff_map)
249 landed: list[str] = []
250 healed: list[str] = []
251 failed: list[str] = []
253 state = load_swarm_state(plan.swarm_id, root=root_path)
255 if dry_run:
256 # Dry run simulation
257 for c in target_wave.clusters:
258 landed.append(c.cluster_id)
259 return SwarmLandingResult(
260 swarm_id=plan.swarm_id,
261 wave_index=wave_index,
262 mode=decision.mode,
263 landed_clusters=tuple(landed),
264 healed_clusters=(),
265 failed_clusters=(),
266 status="success",
267 )
269 # Live landing protected by atomic merge lock
270 lock_path = root_path / ".keel" / "state" / "merge.lock"
271 with merge_lock(lock_path):
272 for c in target_wave.clusters:
273 branch_name = f"swarm/{plan.swarm_id}/{c.cluster_id}"
274 if decision.mode == "direct_batch":
275 ok = merge_cluster_branch(root_path, branch_name, runner=runner)
276 if ok:
277 landed.append(c.cluster_id)
278 if state:
279 state = update_worker_state(
280 state, c.cluster_id, step="s10", status="merged"
281 )
282 else:
283 failed.append(c.cluster_id)
284 if state:
285 state = update_worker_state(
286 state, c.cluster_id, step="s10", status="failed", details="merge failed"
287 )
288 else:
289 # Sequential funnel with rebase & heal
290 rebase_ok, reason = rebase_and_heal_cluster_branch(
291 root_path, branch_name, runner=runner, resolver=resolver
292 )
293 if rebase_ok:
294 healed.append(c.cluster_id)
295 merge_ok = merge_cluster_branch(root_path, branch_name, runner=runner)
296 if merge_ok:
297 landed.append(c.cluster_id)
298 if state:
299 state = update_worker_state(
300 state, c.cluster_id, step="s10", status="merged"
301 )
302 else:
303 failed.append(c.cluster_id)
304 if state:
305 state = update_worker_state(
306 state,
307 c.cluster_id,
308 step="s10",
309 status="failed",
310 details="post-rebase merge failed",
311 )
312 else:
313 failed.append(c.cluster_id)
314 if state:
315 state = update_worker_state(
316 state,
317 c.cluster_id,
318 step="s10",
319 status="failed",
320 details=f"rebase conflict: {reason}",
321 )
323 if state:
324 save_swarm_state(state, root=root_path)
326 overall_status = (
327 "success"
328 if len(failed) == 0 and len(landed) > 0
329 else ("partial_failure" if len(landed) > 0 else "failed")
330 )
332 return SwarmLandingResult(
333 swarm_id=plan.swarm_id,
334 wave_index=wave_index,
335 mode=decision.mode,
336 landed_clusters=tuple(landed),
337 healed_clusters=tuple(healed),
338 failed_clusters=tuple(failed),
339 status=overall_status,
340 )