fix: address CodeRabbit review feedback for PR #82

- Add UTF-8 encoding specification in project_context.py
- Fix TOCTOU race conditions by consolidating exists()/stat() calls
- Move re module import to top-level in prompts.py
- Remove duplicate IdeationConfig type, use shared types
- Add thinking level validation with warning logging
- Fix OAuth handler security: redact device codes from logs
- Remove redundant setTimeout in OAuth extraction flow
- Use explicit string replace instead of regex for clarity

Also adds test_thinking_level_validation.py for validation coverage.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <[email protected]>
This commit is contained in:
AndyMik90
2025-12-21 00:24:58 +01:00
co-authored by Claude Opus 4.5
parent f6601efc8a
commit 6ee5a731f4
9 changed files with 131 additions and 33 deletions
@@ -9,9 +9,9 @@ import { getClaudeProfileManager } from '../claude-profile-manager';
import {
SpecCreationMetadata,
TaskExecutionOptions,
IdeationConfig,
RoadmapConfig
} from './types';
import type { IdeationConfig } from '../../shared/types';
/**
* Main AgentManager - orchestrates agent process lifecycle
+2 -1
View File
@@ -5,7 +5,8 @@ import { EventEmitter } from 'events';
import { AgentState } from './agent-state';
import { AgentEvents } from './agent-events';
import { AgentProcessManager } from './agent-process';
import { IdeationConfig, RoadmapConfig } from './types';
import { RoadmapConfig } from './types';
import type { IdeationConfig } from '../../shared/types';
import { MODEL_ID_MAP } from '../../shared/constants';
import { detectRateLimit, createSDKRateLimitInfo, getProfileEnv } from '../rate-limit-detector';
import { debugLog, debugError } from '../../shared/utils/debug-logger';
+3 -1
View File
@@ -20,9 +20,11 @@ export type {
ExecutionProgressData,
ProcessType,
AgentManagerEvents,
IdeationConfig,
TaskExecutionOptions,
SpecCreationMetadata,
IdeationProgressData,
RoadmapProgressData
} from './types';
// Re-export IdeationConfig from shared types for consistency
export type { IdeationConfig } from '../../shared/types';
+2 -9
View File
@@ -1,4 +1,5 @@
import { ChildProcess } from 'child_process';
import type { IdeationConfig } from '../../shared/types';
/**
* Agent-specific types for process and state management
@@ -32,15 +33,7 @@ export interface AgentManagerEvents {
'execution-progress': (taskId: string, progress: ExecutionProgressData) => void;
}
export interface IdeationConfig {
enabledTypes: string[];
includeRoadmapContext: boolean;
includeKanbanContext: boolean;
maxIdeasPerType: number;
append?: boolean;
model?: string; // Model shorthand (opus, sonnet, haiku)
thinkingLevel?: string; // Thinking level (none, low, medium, high, ultrathink)
}
// IdeationConfig now imported from shared types to maintain consistency
export interface RoadmapConfig {
model?: string; // Model shorthand (opus, sonnet, haiku)
@@ -56,8 +56,8 @@ function parseDeviceCode(output: string): string | null {
const match = output.match(DEVICE_CODE_PATTERN);
if (match && match[1]) {
// Normalize: replace space with hyphen (GitHub expects XXXX-XXXX format)
const normalizedCode = match[1].replace(/\s/, '-');
debugLog('Parsed device code:', normalizedCode);
const normalizedCode = match[1].replace(' ', '-');
debugLog('Device code extracted successfully (code redacted for security)');
return normalizedCode;
}
return null;
@@ -236,7 +236,7 @@ export function registerStartGhAuth(): void {
extractedDeviceCode = deviceFlowInfo.deviceCode;
extractedAuthUrl = deviceFlowInfo.authUrl;
debugLog('Device code extracted:', extractedDeviceCode);
debugLog('Device code extracted successfully (code redacted for security)');
debugLog('Auth URL:', extractedAuthUrl);
// Open browser using Electron's shell.openExternal
@@ -251,10 +251,9 @@ export function registerStartGhAuth(): void {
// Don't fail here - we'll return the device code so user can manually navigate
}
// Reset extraction flag after browser open attempt to allow retry if needed
setTimeout(() => {
extractionInProgress = false;
}, 5000);
// Extraction complete - mutex flag stays true to prevent re-extraction
// The deviceCodeExtracted flag will prevent future attempts
extractionInProgress = false;
} else {
// No device code found yet, allow next data chunk to try again
extractionInProgress = false;
+11 -1
View File
@@ -118,7 +118,17 @@ def get_thinking_budget(thinking_level: str) -> int | None:
Returns:
Token budget or None for no extended thinking
"""
return THINKING_BUDGET_MAP.get(thinking_level, THINKING_BUDGET_MAP["medium"])
import logging
if thinking_level not in THINKING_BUDGET_MAP:
valid_levels = ", ".join(THINKING_BUDGET_MAP.keys())
logging.warning(
f"Invalid thinking_level '{thinking_level}'. Valid values: {valid_levels}. "
f"Defaulting to 'medium'."
)
return THINKING_BUDGET_MAP["medium"]
return THINKING_BUDGET_MAP[thinking_level]
def load_task_metadata(spec_dir: Path) -> TaskMetadataConfig | None:
+11 -11
View File
@@ -29,7 +29,7 @@ def load_project_index(project_dir: Path) -> dict:
return {}
try:
with open(index_file) as f:
with open(index_file, encoding="utf-8") as f:
return json.load(f)
except (json.JSONDecodeError, OSError):
return {}
@@ -200,10 +200,11 @@ def should_refresh_project_index(project_dir: Path) -> bool:
for dep_file in dep_files:
try:
if dep_file.exists() and dep_file.stat().st_mtime > index_mtime:
dep_mtime = dep_file.stat().st_mtime
if dep_mtime > index_mtime:
return True # Dependency file changed, refresh needed
except OSError:
continue # Skip files we can't stat
except (OSError, FileNotFoundError):
continue # Skip files we can't stat or don't exist
# Also check subdirectories for monorepos (first level only)
try:
@@ -222,19 +223,18 @@ def should_refresh_project_index(project_dir: Path) -> bool:
subdir_pkg = subdir / "package.json"
try:
if subdir_pkg.exists() and subdir_pkg.stat().st_mtime > index_mtime:
pkg_mtime = subdir_pkg.stat().st_mtime
if pkg_mtime > index_mtime:
return True
except OSError:
except (OSError, FileNotFoundError):
continue
subdir_pyproject = subdir / "pyproject.toml"
try:
if (
subdir_pyproject.exists()
and subdir_pyproject.stat().st_mtime > index_mtime
):
pyproject_mtime = subdir_pyproject.stat().st_mtime
if pyproject_mtime > index_mtime:
return True
except OSError:
except (OSError, FileNotFoundError):
continue
except OSError:
pass # Can't iterate dir, use cached index
+1 -2
View File
@@ -7,6 +7,7 @@ Supports dynamic prompt assembly based on project type for context optimization.
"""
import json
import re
from pathlib import Path
from .project_context import (
@@ -384,8 +385,6 @@ The project root is: `{project_dir}`
mcp_content += "\n\n---\n"
# Replace the multi-line marker comment block
import re
marker_pattern = r"<!-- PROJECT-SPECIFIC VALIDATION TOOLS WILL BE INJECTED HERE -->.*?<!-- - API validation \(for projects with API endpoints\) -->"
base_prompt = re.sub(marker_pattern, mcp_content, base_prompt, flags=re.DOTALL)
elif mcp_sections:
+94
View File
@@ -0,0 +1,94 @@
"""
Tests for thinking level validation in phase_config module.
Ensures that invalid thinking levels are caught with proper warnings
and default to 'medium' as expected.
"""
import logging
import sys
from pathlib import Path
import pytest
# Add auto-claude to path
sys.path.insert(0, str(Path(__file__).parent.parent / "auto-claude"))
from phase_config import THINKING_BUDGET_MAP, get_thinking_budget
class TestThinkingLevelValidation:
"""Test thinking level validation and error handling."""
def test_valid_thinking_levels(self):
"""Test that all valid thinking levels return correct budgets."""
valid_levels = ["none", "low", "medium", "high", "ultrathink"]
for level in valid_levels:
budget = get_thinking_budget(level)
expected = THINKING_BUDGET_MAP[level]
assert budget == expected, f"Expected {expected} for {level}, got {budget}"
def test_none_level_returns_none(self):
"""Test that 'none' thinking level returns None (no extended thinking)."""
assert get_thinking_budget("none") is None
def test_ultrathink_max_budget(self):
"""Test that 'ultrathink' returns maximum budget."""
assert get_thinking_budget("ultrathink") == 65536
def test_invalid_level_logs_warning(self, caplog):
"""Test that invalid thinking level logs a warning."""
with caplog.at_level(logging.WARNING):
budget = get_thinking_budget("invalid_level")
# Should default to medium
assert budget == THINKING_BUDGET_MAP["medium"]
# Should have logged a warning
assert len(caplog.records) == 1
assert "Invalid thinking_level 'invalid_level'" in caplog.text
assert "Valid values:" in caplog.text
assert "Defaulting to 'medium'" in caplog.text
def test_invalid_level_shows_valid_options(self, caplog):
"""Test that warning message includes all valid options."""
with caplog.at_level(logging.WARNING):
get_thinking_budget("bad_value")
# Check all valid levels are mentioned
for level in ["none", "low", "medium", "high", "ultrathink"]:
assert level in caplog.text
def test_empty_string_level(self, caplog):
"""Test that empty string is treated as invalid."""
with caplog.at_level(logging.WARNING):
budget = get_thinking_budget("")
assert budget == THINKING_BUDGET_MAP["medium"]
assert "Invalid thinking_level" in caplog.text
def test_case_sensitive(self, caplog):
"""Test that thinking level is case-sensitive."""
with caplog.at_level(logging.WARNING):
# "MEDIUM" should be invalid (not "medium")
budget = get_thinking_budget("MEDIUM")
assert budget == THINKING_BUDGET_MAP["medium"]
assert "Invalid thinking_level 'MEDIUM'" in caplog.text
def test_multiple_invalid_calls(self, caplog):
"""Test that each invalid call produces a warning."""
invalid_levels = ["bad1", "bad2", "bad3"]
with caplog.at_level(logging.WARNING):
for level in invalid_levels:
get_thinking_budget(level)
# Should have 3 warnings
assert len(caplog.records) == 3
def test_budget_values_match_expected(self):
"""Test that budget values match documented amounts."""
assert get_thinking_budget("low") == 1024
assert get_thinking_budget("medium") == 4096
assert get_thinking_budget("high") == 16384
assert get_thinking_budget("ultrathink") == 65536