Skip to content
Merged
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 0 additions & 3 deletions .claude/skills/sc-implement/scripts/loop_entry.py
Original file line number Diff line number Diff line change
Expand Up @@ -79,7 +79,6 @@ def build_config(context: dict[str, Any]) -> "LoopConfig":
return LoopConfig(
max_iterations=context.get("max_iterations", 3),
quality_threshold=context.get("quality_threshold", 70.0),
min_improvement=context.get("min_improvement", 5.0),
pal_review_enabled=context.get("pal_review", True),
pal_model=context.get("pal_model", "gpt-5"),
timeout_seconds=context.get("timeout_seconds"),
Expand Down Expand Up @@ -183,8 +182,6 @@ def create_signal_only_response(context: dict[str, Any]) -> dict[str, Any]:
),
"safety": {
"hard_max_iterations": 5,
"detect_oscillation": True,
"detect_stagnation": True,
},
}

Expand Down
77 changes: 0 additions & 77 deletions SuperClaude/Orchestrator/loop_runner.py
Original file line number Diff line number Diff line change
Expand Up @@ -41,8 +41,6 @@ class TerminationReason(Enum):

QUALITY_MET = "quality_threshold_met"
MAX_ITERATIONS = "max_iterations_reached"
OSCILLATION = "oscillation_detected"
STAGNATION = "stagnation_detected"
TIMEOUT = "timeout_exceeded"
USER_CANCELLED = "user_cancelled"
ERROR = "error"
Expand All @@ -58,11 +56,6 @@ class LoopConfig:

# Quality settings
quality_threshold: float = 70.0
min_improvement: float = 5.0 # Minimum score improvement to continue

# Termination detection
oscillation_window: int = 3
stagnation_threshold: float = 2.0

# Timeouts
timeout_seconds: float | None = None
Expand Down Expand Up @@ -257,20 +250,6 @@ def combined_callback(result: IterationResult) -> None:
logger.info("Quality threshold met!")
break

if len(score_history) >= config.oscillation_window:
if _is_oscillating(score_history[-config.oscillation_window :]):
termination_reason = TerminationReason.OSCILLATION
logger.warning("Oscillation detected, terminating loop")
break

if len(score_history) >= 2:
if _is_stagnating(
score_history[-2:], config.stagnation_threshold, config.min_improvement
):
termination_reason = TerminationReason.STAGNATION
logger.warning("Stagnation detected, terminating loop")
break

# Check timeout
if config.timeout_seconds:
elapsed = (datetime.now() - loop_start).total_seconds()
Expand Down Expand Up @@ -343,62 +322,6 @@ def _build_iteration_prompt(
return prompt


def _is_oscillating(scores: list[float], threshold: float = 5.0) -> bool:
"""
Detect if scores are oscillating (up/down/up pattern).

Args:
scores: Recent score history (at least 3 values)
threshold: Minimum delta to count as a direction change

Returns:
True if oscillating pattern detected
"""
if len(scores) < 3:
return False

deltas = [scores[i + 1] - scores[i] for i in range(len(scores) - 1)]

# Check for alternating positive/negative deltas
alternating = 0
for i in range(len(deltas) - 1):
if (deltas[i] > threshold and deltas[i + 1] < -threshold) or (
deltas[i] < -threshold and deltas[i + 1] > threshold
):
alternating += 1

return alternating >= 1


def _is_stagnating(
scores: list[float],
variance_threshold: float,
min_improvement: float,
) -> bool:
"""
Detect if scores are stagnating (no meaningful improvement).

Args:
scores: Recent score history (at least 2 values)
variance_threshold: Max variance to count as stagnant
min_improvement: Minimum improvement needed

Returns:
True if stagnating
"""
if len(scores) < 2:
return False

# Check if improvement is below threshold
delta = scores[-1] - scores[-2]
if delta < min_improvement:
return True

# Check variance
variance = max(scores) - min(scores)
return variance < variance_threshold


# Synchronous wrapper for non-async contexts
def run_agentic_loop_sync(
task: str,
Expand Down
7 changes: 0 additions & 7 deletions core/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,15 +13,11 @@

Safety guarantees:
- HARD_MAX_ITERATIONS = 5 (cannot be overridden)
- Oscillation detection (prevents infinite back-and-forth)
- Stagnation detection (stops when no progress)
- Minimum improvement threshold (stops if < 5 point gain)
"""

from .loop_orchestrator import LoopOrchestrator
from .pal_integration import PALReviewSignal
from .quality_assessment import QualityAssessor
from .termination import detect_oscillation, detect_stagnation
from .types import (
IterationResult,
LoopConfig,
Expand All @@ -37,9 +33,6 @@
"LoopResult",
"IterationResult",
"QualityAssessment",
# Functions
"detect_oscillation",
"detect_stagnation",
# Classes
"QualityAssessor",
"PALReviewSignal",
Expand Down
82 changes: 4 additions & 78 deletions core/loop_orchestrator.py
Original file line number Diff line number Diff line change
Expand Up @@ -24,11 +24,6 @@
from .metrics import MetricsEmitter, noop_emitter
from .pal_integration import PALReviewSignal, incorporate_pal_feedback
from .quality_assessment import QualityAssessor
from .termination import (
check_insufficient_improvement,
detect_oscillation,
detect_stagnation,
)
from .types import (
IterationResult,
LoopConfig,
Expand All @@ -45,7 +40,7 @@ class LoopOrchestrator:
This class implements the --loop functionality that was archived
in v5's QualityScorer.agentic_loop(). It provides:

- Safety mechanisms (hard max, oscillation/stagnation detection)
- Safety mechanisms (hard max, timeout)
- Quality-driven iteration (stop when threshold met)
- PAL MCP integration within loop (not just after)
- Signal-based skill invocation (Claude Code processes signals)
Expand Down Expand Up @@ -219,58 +214,7 @@ def run(
)
break

# 4. Check termination conditions
if detect_oscillation(
self.score_history,
self.config.oscillation_window,
):
self.logger.info("Oscillation detected.", extra=log_context)
termination_reason = TerminationReason.OSCILLATION
self._record_iteration(
iteration=iteration,
assessment=assessment,
time_taken=time.monotonic() - iter_start,
success=False,
termination="oscillation",
changed_files=changed_files,
)
break

if detect_stagnation(
self.score_history,
self.config.oscillation_window,
self.config.stagnation_threshold,
):
self.logger.info("Stagnation detected.", extra=log_context)
termination_reason = TerminationReason.STAGNATION
self._record_iteration(
iteration=iteration,
assessment=assessment,
time_taken=time.monotonic() - iter_start,
success=False,
termination="stagnation",
changed_files=changed_files,
)
break

if iteration > 0 and check_insufficient_improvement(
self.score_history[-1],
self.score_history[-2],
self.config.min_improvement,
):
self.logger.info("Insufficient improvement detected.", extra=log_context)
termination_reason = TerminationReason.INSUFFICIENT_IMPROVEMENT
self._record_iteration(
iteration=iteration,
assessment=assessment,
time_taken=time.monotonic() - iter_start,
success=False,
termination="insufficient_improvement",
changed_files=changed_files,
)
break

# 5. Generate PAL review signal (if enabled and not last iteration)
# 4. Generate PAL review signal (if enabled and not last iteration)
pal_signal = None
if self.config.pal_review_enabled and iteration < self.config.max_iterations - 1:
self.logger.debug("Generating PAL review signal.", extra=log_context)
Expand All @@ -281,7 +225,7 @@ def run(
model=self.config.pal_model,
)

# 6. Record iteration
# 5. Record iteration
self._record_iteration(
iteration=iteration,
assessment=assessment,
Expand All @@ -292,7 +236,7 @@ def run(
pal_signal=pal_signal,
)

# 7. Prepare next iteration context
# 6. Prepare next iteration context
current_context = self._prepare_next_iteration(
current_context,
assessment,
Expand All @@ -314,24 +258,6 @@ def run(
if self.iteration_history:
self.iteration_history[-1].pal_review = final_signal

elif termination_reason in (
TerminationReason.OSCILLATION,
TerminationReason.STAGNATION,
):
# Debug signal for stuck loops
self.logger.debug(
"Generating PAL debug signal for stuck loop.",
extra={"loop_id": self.loop_id, "reason": termination_reason.value},
)
debug_signal = PALReviewSignal.generate_debug_signal(
iteration=len(self.iteration_history) - 1,
termination_reason=termination_reason.value,
score_history=self.score_history,
model=self.config.pal_model,
)
if self.iteration_history:
self.iteration_history[-1].pal_review = debug_signal

total_time = time.monotonic() - self._start_time
final_tags = {"termination_reason": termination_reason.value}
self.metrics_emitter("loop.completed.count", 1, final_tags)
Expand Down
88 changes: 0 additions & 88 deletions core/pal_integration.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,6 @@

This enables:
- Per-iteration code review via mcp__pal__codereview
- Debugging assistance via mcp__pal__debug when stuck
- Consensus building via mcp__pal__consensus for architecture
"""

Expand Down Expand Up @@ -86,52 +85,6 @@ def generate_review_signal(
},
}

@staticmethod
def generate_debug_signal(
iteration: int,
termination_reason: str,
score_history: list[float],
model: str = "gpt-5",
) -> dict[str, Any]:
"""
Generate a PAL debug signal when loop is stuck.

Used when oscillation or stagnation is detected to
diagnose why improvements aren't converging.

Args:
iteration: Current iteration number
termination_reason: Why the loop is stopping
score_history: History of quality scores
model: Model to use for debugging

Returns:
Signal dict for Claude Code to process
"""
return {
"action_required": True,
"tool": PALReviewSignal.TOOL_DEBUG,
"iteration": iteration,
"instruction": (
f"Loop terminated due to {termination_reason}. "
f"Diagnose why improvements aren't converging."
),
"model": model,
"context": {
"termination_reason": termination_reason,
"score_history": score_history,
"pattern": _detect_pattern(score_history),
},
"parameters": {
"step": f"Diagnose {termination_reason} in improvement loop",
"step_number": 1,
"total_steps": 1,
"next_step_required": False,
"findings": "",
"hypothesis": f"Loop stuck due to {termination_reason}",
},
}

@staticmethod
def generate_final_validation_signal(
changed_files: list[str],
Expand Down Expand Up @@ -182,47 +135,6 @@ def generate_final_validation_signal(
}


def _detect_pattern(score_history: list[float]) -> str:
"""
Detect the pattern in score history for debugging.

Args:
score_history: List of quality scores

Returns:
Pattern description
"""
if len(score_history) < 2:
return "insufficient_data"

# Check for oscillation
directions = []
for i in range(1, len(score_history)):
diff = score_history[i] - score_history[i - 1]
if abs(diff) > 2.0:
directions.append("up" if diff > 0 else "down")

if len(directions) >= 2:
alternating = all(directions[i] != directions[i + 1] for i in range(len(directions) - 1))
if alternating:
return "oscillating"

# Check for stagnation
recent = score_history[-3:] if len(score_history) >= 3 else score_history
if max(recent) - min(recent) < 2.0:
return "stagnating"

# Check for declining
if len(score_history) >= 2 and score_history[-1] < score_history[0]:
return "declining"

# Check for improving
if len(score_history) >= 2 and score_history[-1] > score_history[0]:
return "improving"

return "mixed"


def incorporate_pal_feedback(
context: dict[str, Any],
pal_result: dict[str, Any],
Expand Down
Loading
Loading