Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
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
4 changes: 2 additions & 2 deletions .claude/tools/amplihack/hooks/power_steering_sdk.py
Original file line number Diff line number Diff line change
Expand Up @@ -40,14 +40,14 @@

# --- Launcher detection -------------------------------------------------------

_detector_cache: object | None = None
_detector_cache: str | None = None


def _detect_launcher(project_root: Path) -> str:
"""Detect launcher type, cached per process."""
global _detector_cache
if _detector_cache is not None:
return _detector_cache # type: ignore[return-value]
return _detector_cache
try:
sys.path.insert(0, str(Path(__file__).parents[3] / "src" / "amplihack"))
from amplihack.context.adaptive.detector import LauncherDetector
Expand Down
84 changes: 18 additions & 66 deletions .claude/tools/amplihack/hooks/tests/test_issue_1872_bug_fixes.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,7 @@
import tempfile
import unittest
from pathlib import Path
from unittest.mock import patch
from unittest.mock import AsyncMock, patch

sys.path.insert(0, str(Path(__file__).parent.parent))

Expand Down Expand Up @@ -361,21 +361,14 @@ def tearDown(self):
shutil.rmtree(self.temp_dir)

@patch("claude_power_steering.CLAUDE_SDK_AVAILABLE", True)
@patch("claude_power_steering.query")
@patch("claude_power_steering.query_llm", new_callable=AsyncMock)
def test_analyze_consideration_returns_tuple(self, mock_query):
"""Test that analyze_consideration returns Tuple[bool, Optional[str]]."""
# Import here to get patched version
from claude_power_steering import analyze_consideration

# Mock SDK response with NOT SATISFIED
async def mock_response(*args, **kwargs):
class MockMessage:
def __init__(self, text):
self.text = text

yield MockMessage("NOT SATISFIED: Missing tests")

mock_query.return_value = mock_response()
mock_query.return_value = "NOT SATISFIED: Missing tests"

consideration = {
"id": "test_check",
Expand All @@ -398,18 +391,12 @@ def __init__(self, text):
self.assertIsInstance(reason, (str, type(None)), "Second element should be str or None")

@patch("claude_power_steering.CLAUDE_SDK_AVAILABLE", True)
@patch("claude_power_steering.query")
@patch("claude_power_steering.query_llm", new_callable=AsyncMock)
def test_reason_extracted_when_check_fails(self, mock_query):
"""Test that reason is extracted when check fails."""
from claude_power_steering import analyze_consideration

async def mock_response(*args, **kwargs):
class MockMessage:
text = "NOT SATISFIED: TodoWrite shows 3 incomplete tasks"

yield MockMessage()

mock_query.return_value = mock_response()
mock_query.return_value = "NOT SATISFIED: TodoWrite shows 3 incomplete tasks"

consideration = {
"id": "todos_complete",
Expand All @@ -429,20 +416,13 @@ class MockMessage:
self.assertIn("incomplete", reason.lower(), "Reason should mention incomplete tasks")

@patch("claude_power_steering.CLAUDE_SDK_AVAILABLE", True)
@patch("claude_power_steering.query")
@patch("claude_power_steering.query_llm", new_callable=AsyncMock)
def test_reason_truncated_to_200_chars(self, mock_query):
"""Test that reason is truncated to 200 characters."""
from claude_power_steering import analyze_consideration

long_reason = "NOT SATISFIED: " + ("A" * 300) # 313 chars total

async def mock_response(*args, **kwargs):
class MockMessage:
text = long_reason

yield MockMessage()

mock_query.return_value = mock_response()
mock_query.return_value = long_reason

consideration = {
"id": "test_check",
Expand All @@ -461,18 +441,12 @@ class MockMessage:
self.assertLessEqual(len(reason), 200, "Reason should be truncated to 200 chars")

@patch("claude_power_steering.CLAUDE_SDK_AVAILABLE", True)
@patch("claude_power_steering.query")
@patch("claude_power_steering.query_llm", new_callable=AsyncMock)
def test_reason_none_when_check_passes(self, mock_query):
"""Test that reason is None when check passes."""
from claude_power_steering import analyze_consideration

async def mock_response(*args, **kwargs):
class MockMessage:
text = "SATISFIED: All tests passed successfully"

yield MockMessage()

mock_query.return_value = mock_response()
mock_query.return_value = "SATISFIED: All tests passed successfully"

consideration = {
"id": "local_testing",
Expand Down Expand Up @@ -520,19 +494,12 @@ def test_generate_final_guidance_function_exists(self):
self.fail("generate_final_guidance function should exist")

@patch("claude_power_steering.CLAUDE_SDK_AVAILABLE", True)
@patch("claude_power_steering.query")
@patch("claude_power_steering.query_llm", new_callable=AsyncMock)
def test_generate_final_guidance_calls_sdk(self, mock_query):
"""Test that generate_final_guidance calls SDK with failed checks and reasons."""
from claude_power_steering import generate_final_guidance

# Mock SDK response
async def mock_response(*args, **kwargs):
class MockMessage:
text = "Complete the remaining TODOs and run tests locally."

yield MockMessage()

mock_query.return_value = mock_response()
mock_query.return_value = "Complete the remaining TODOs and run tests locally."

failed_checks = [
("todos_complete", "3 tasks remain incomplete"),
Expand All @@ -551,18 +518,12 @@ class MockMessage:
self.assertGreater(len(guidance), 0, "Guidance should not be empty")

@patch("claude_power_steering.CLAUDE_SDK_AVAILABLE", True)
@patch("claude_power_steering.query")
@patch("claude_power_steering.query_llm", new_callable=AsyncMock)
def test_generate_final_guidance_includes_failure_context(self, mock_query):
"""Test that generate_final_guidance includes actual failure context in prompt."""
from claude_power_steering import generate_final_guidance

async def mock_response(*args, **kwargs):
class MockMessage:
text = "Fix the failing checks"

yield MockMessage()

mock_query.return_value = mock_response()
mock_query.return_value = "Fix the failing checks"

failed_checks = [
("ci_status", "CI checks failing on test_module.py"),
Expand All @@ -574,7 +535,7 @@ class MockMessage:

# Verify the prompt passed to SDK includes the failure info
call_args = mock_query.call_args
prompt = call_args[1]["prompt"] # Get keyword argument 'prompt'
prompt = call_args[0][0] # First positional arg to query_llm(prompt, project_root)

self.assertIn("ci_status", prompt, "Prompt should include check ID")
self.assertIn("failing", prompt.lower(), "Prompt should include failure reason")
Expand Down Expand Up @@ -604,18 +565,12 @@ def test_generate_final_guidance_fallback_when_sdk_unavailable(self):
self.assertIn("local_testing", guidance, "Should mention failed check")

@patch("claude_power_steering.CLAUDE_SDK_AVAILABLE", True)
@patch("claude_power_steering.query")
@patch("claude_power_steering.query_llm", new_callable=AsyncMock)
def test_generate_final_guidance_is_specific_not_generic(self, mock_query):
"""Test that guidance is specific to actual failures, not generic advice."""
from claude_power_steering import generate_final_guidance

async def mock_response(*args, **kwargs):
class MockMessage:
text = "You need to complete the 3 incomplete TODOs and run pytest locally."

yield MockMessage()

mock_query.return_value = mock_response()
mock_query.return_value = "You need to complete the 3 incomplete TODOs and run pytest locally."

failed_checks = [
("todos_complete", "3 incomplete tasks"),
Expand All @@ -633,16 +588,13 @@ class MockMessage:
self.assertIn("pytest", guidance.lower(), "Should mention specific tool from reason")

@patch("claude_power_steering.CLAUDE_SDK_AVAILABLE", True)
@patch("claude_power_steering.query")
@patch("claude_power_steering.query_llm", new_callable=AsyncMock)
def test_generate_final_guidance_sdk_failure_uses_template(self, mock_query):
"""Test that SDK failure falls back to template guidance."""
from claude_power_steering import generate_final_guidance

# Make SDK raise exception
async def failing_response(*args, **kwargs):
raise RuntimeError("SDK timeout")

mock_query.side_effect = failing_response
mock_query.side_effect = RuntimeError("SDK timeout")

failed_checks = [
("ci_status", "CI failing"),
Expand Down
Loading
Loading