fix(system): S61-S62 — API cleanup, drone path routing, naming checker, .backup→.recovery rename (#142)

- API: 18-item P0/P1/P2 cleanup (debug prints→logger, help gate fix, URL injection fix, key masking, unused imports)
- Drone: passport walk-up replaces hardcoded src/aipass/<branch> pattern, 13 naming bypasses removed
- Seedgo: naming checker fixed (__dunder__ skip + column-0 scope detection) — eliminates 71 false positives system-wide
- Spawn: .backup→.recovery rename (pre-change snapshots no longer conflict with backup branch namespace)
- .gitignore: .recovery/ added
- HERALD.md: Updated through S62 (backup audit, .recovery rename, night shift launch)
- README.md: Stats updated (141 PRs, 2900+ tests)

Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
AIPass
2026-03-29 01:55:21 -07:00
committed by GitHub
co-authored by Claude Opus 4.6
parent ae459853f0
commit 6772cacce2
27 changed files with 245 additions and 212 deletions
+5
View File
@@ -74,6 +74,11 @@
"file": "apps/modules/google_client.py",
"standard": "unused_function",
"reason": "Public API exports for cross-branch consumption (get_drive_service, authenticate_google, validate_google, reauth_google). Tested in test_google_client.py. Backup migration pending."
},
{
"file": "apps/modules/openrouter_client.py",
"standard": "modules",
"reason": "make_call() is a CLI orchestrator — parses args then delegates to client.get_response(). All business logic lives in handlers/openrouter/client.py."
}
],
"notes": {
+8 -1
View File
@@ -157,8 +157,15 @@ def print_help():
table.add_row("reauth google", "Re-authenticate Google OAuth2")
table.add_row("test", "Test OpenRouter connection status")
table.add_row("models", "List available models from provider")
table.add_row("status", "Check OpenRouter client status")
table.add_row("call", "Make API call to model")
table.add_row("list-providers", "List available API providers")
table.add_row("init", "Initialize .env template")
table.add_row("track", "Track API usage metrics")
table.add_row("stats", "Display API usage statistics")
table.add_row("session", "Show session usage data")
table.add_row("caller-usage", "Show usage by caller module")
table.add_row("cleanup", "Clean up old usage data")
console.print(table)
console.print()
@@ -193,7 +200,7 @@ def print_help():
console.print("─" * 70)
console.print()
console.print("[dim]Commands: get-key, validate, test, models, track, stats, help, --help[/dim]")
console.print("[dim]Commands: get-key, validate, test, models, status, call, list-providers, init, track, stats, session, caller-usage, cleanup[/dim]")
console.print()
+3 -3
View File
@@ -42,9 +42,9 @@ def _guard_branch_access():
import os
if os.environ.get("AIPASS_DEBUG_GUARD"):
import sys
print(f"[GUARD DEBUG] caller_file = {caller_file}", file=sys.stderr)
print(f"[GUARD DEBUG] import_line = {import_line}", file=sys.stderr)
from aipass.prax import logger as _guard_logger
_guard_logger.info(f"[GUARD] caller_file = {caller_file}")
_guard_logger.info(f"[GUARD] import_line = {import_line}")
if caller_file is None:
stack = inspect.stack()
+1 -2
View File
@@ -17,10 +17,9 @@ Functions:
# Infrastructure
from pathlib import Path
import sys
# Standard library
from typing import Optional, Dict
from typing import Optional
# Logging
from aipass.prax import logger
@@ -10,7 +10,6 @@ import json
from pathlib import Path
from datetime import datetime
from typing import Dict, Any, Optional
import sys
import inspect
# Logging
@@ -121,7 +120,7 @@ def ensure_json_exists(module_name: str, json_type: str) -> bool:
if validate_json_structure(data, json_type):
return True
else:
pass # Corrupted - regenerating
logger.warning(f"Corrupted JSON structure at {json_path}, regenerating")
except Exception as e:
logger.warning(f"Unreadable JSON at {json_path}, regenerating: {e}")
@@ -11,7 +11,7 @@
OpenRouter Client Handler
Business logic for OpenRouter API client creation and request execution.
Extracted from archive.temp/openrouter.py following AIPASS standards.
OpenRouter API client creation and request execution.
Functions:
- get_response() - Main API call with tracking integration
@@ -34,8 +34,6 @@ Standards:
"""
# INFRASTRUCTURE IMPORT PATTERN
import sys
from pathlib import Path
# Standard library imports
import time
@@ -193,6 +191,7 @@ def make_api_request(client: OpenAI, messages: List[Dict], model: str, retries:
OpenAI response object or None on failure
"""
if not client or not messages or not model:
logger.warning(f"make_api_request() called with missing params — client={bool(client)}, messages={bool(messages)}, model={bool(model)}")
return None
api_params = {
@@ -242,9 +241,11 @@ def extract_response(response: Any) -> Optional[Dict[str, Any]]:
try:
# Validate response structure
if not hasattr(response, 'choices') or not response.choices:
logger.warning("Response missing 'choices' or choices is empty")
return None
if not hasattr(response.choices[0], 'message'):
logger.warning("Response choice missing 'message' attribute")
return None
# Extract content
@@ -25,12 +25,11 @@ COMPLIANT STANDARDS:
- Under 300 lines
"""
import sys
from pathlib import Path
import json
from datetime import datetime
from typing import Dict, Any, Optional, Tuple
from typing import Dict, Any, Optional
from aipass.prax import logger
@@ -102,7 +101,7 @@ def get_default_caller_config() -> Dict[str, Any]:
Dict with default OpenRouter configuration (model must be set by caller)
"""
return {
"skill_name": "openrouter",
"module_name": "openrouter",
"timestamp": datetime.now().isoformat(),
"config": {
"ai_model": "", # Caller must set their own model
@@ -121,7 +120,7 @@ def get_default_caller_data() -> Dict[str, Any]:
Dict with initial usage tracking data
"""
return {
"skill_name": "openrouter",
"module_name": "openrouter",
"timestamp": datetime.now().isoformat(),
"data": {
"total_requests": 0,
@@ -141,7 +140,7 @@ def get_default_caller_log() -> Dict[str, Any]:
Dict with empty log structure
"""
return {
"skill_name": "openrouter",
"module_name": "openrouter",
"timestamp": datetime.now().isoformat(),
"logs": []
}
@@ -15,7 +15,6 @@ Business logic for aggregating usage statistics:
- Cost, token, and latency aggregation
- Model usage tracking and breakdown
Extracted from legacy archive (api_usage.py).
Functions: get_caller_usage(), get_session_summary()
"""
@@ -15,11 +15,9 @@ Business logic for tracking API usage from OpenRouter:
- Store generation tracking data with newest-first ordering
- Handle HTTP requests with proper error handling
Extracted from legacy archive (api_usage.py).
Functions: track_usage(), _get_generation_metrics(), _store_usage_data()
Functions: track_usage(), get_generation_metrics(), store_usage_data()
"""
import sys
from pathlib import Path
# Standard library imports
@@ -147,7 +145,8 @@ def get_generation_metrics(generation_id: str, api_key: str) -> Optional[Dict[st
# Query the generation endpoint
response = requests.get(
f"{GENERATION_ENDPOINT}?id={generation_id}",
GENERATION_ENDPOINT,
params={"id": generation_id},
headers=headers,
timeout=30
)
@@ -158,7 +157,7 @@ def get_generation_metrics(generation_id: str, api_key: str) -> Optional[Dict[st
# Validate response structure
if not data or "data" not in data:
# Invalid response structure from OpenRouter
logger.warning(f"[{MODULE_NAME}] Invalid response structure from OpenRouter for generation {generation_id}")
return None
# Extract metrics from response
@@ -176,7 +175,7 @@ def get_generation_metrics(generation_id: str, api_key: str) -> Optional[Dict[st
return result
else:
# OpenRouter API returned non-200 status
logger.warning(f"[{MODULE_NAME}] OpenRouter API returned status {response.status_code} for generation {generation_id}")
return None
except requests.exceptions.Timeout as e:
+3 -3
View File
@@ -20,10 +20,9 @@ from pathlib import Path
from typing import List
from aipass.prax.apps.modules.logger import system_logger as logger
from aipass.cli.apps.modules import console, header, success, error, warning, section
from aipass.cli.apps.modules import console, header, success, error
from aipass.api.apps.handlers.json import json_handler
from aipass.api.apps.handlers.auth import keys, env
from aipass.api.apps.handlers.config import provider
def print_introspection():
@@ -110,7 +109,8 @@ def get_key(args: List[str]):
if api_key:
success(f"API key retrieved for {provider_name}")
console.print(f" Key (first 20 chars): {api_key[:20]}...")
masked = api_key[:6] + "****" + api_key[-4:] if len(api_key) > 10 else "****"
console.print(f" Key: {masked}")
else:
error(f"Failed to retrieve API key for {provider_name}")
+6 -5
View File
@@ -124,11 +124,6 @@ def handle_command(command: str, args: List[str]) -> bool:
Returns:
True if command was handled, False to pass through.
"""
# Help gate
if args and args[0] in ("--help", "-h", "help"):
print_help()
return True
# NO-ARGS GATE (seedgo standard)
if not args:
if command == "google":
@@ -140,6 +135,12 @@ def handle_command(command: str, args: List[str]) -> bool:
if args[0] != "google":
return False
# Help gate — only fires after confirming this is a google command
remaining = args[1:] if len(args) > 1 else []
if remaining and remaining[0] in ("--help", "-h", "help"):
print_help()
return True
if command == "validate":
_cmd_validate()
return True
@@ -17,11 +17,10 @@ Orchestrates LLM API client operations:
"""
import sys
from pathlib import Path
from typing import List
from aipass.prax.apps.modules.logger import system_logger as logger
from aipass.cli.apps.modules import console, header, success, error, warning, section
from aipass.cli.apps.modules import console, header, success, error
from aipass.api.apps.handlers.json import json_handler
from aipass.api.apps.handlers.auth import keys
from aipass.api.apps.handlers.openrouter import client, models
@@ -188,7 +187,42 @@ def make_call(args: List[str]):
header("OpenRouter API Call")
console.print()
warning("API call workflow - TODO")
if not args:
error("Prompt required", suggestion="drone @api call \"your prompt\" --model MODEL")
return
# Parse args: first non-flag arg is prompt, --model MODEL is optional
prompt = None
model = None
i = 0
while i < len(args):
if args[i] == "--model" and i + 1 < len(args):
model = args[i + 1]
i += 2
elif prompt is None:
prompt = args[i]
i += 1
else:
i += 1
if not prompt:
error("Prompt required", suggestion="drone @api call \"your prompt\" --model MODEL")
return
if not model:
error("Model required", suggestion="drone @api call \"your prompt\" --model anthropic/claude-3.5-sonnet")
return
console.print(f"[dim]Calling {model}...[/dim]")
response = client.get_response(prompt, caller="cli", model=model)
if response:
success(f"Response received ({len(response['content'])} chars)")
console.print()
console.print(response["content"])
else:
error("API call failed")
def list_models(args: List[str] | None = None):
+17 -2
View File
@@ -21,7 +21,7 @@ from pathlib import Path
from typing import List
from aipass.prax.apps.modules.logger import system_logger as logger
from aipass.cli.apps.modules import console, header, success, error, warning, section
from aipass.cli.apps.modules import console, header, success, error, warning
from aipass.api.apps.handlers.json import json_handler
from aipass.api.apps.handlers.usage import tracking, aggregation, cleanup
@@ -174,7 +174,22 @@ def track_usage(args: List[str]):
header("Track API Usage")
console.print()
warning("Usage tracking workflow - TODO")
if not args:
error("Generation ID required", suggestion="drone @api track <generation_id> [caller]")
return
generation_id = args[0]
caller = args[1] if len(args) > 1 else "manual"
console.print(f"[dim]Tracking generation {generation_id}...[/dim]")
result = tracking.track_usage(generation_id, caller)
if result.get("success"):
metrics = result.get("metrics", {})
success(f"Tracked: {metrics.get('tokens_prompt', 0)} prompt + {metrics.get('tokens_completion', 0)} completion tokens, ${metrics.get('total_cost', 0):.6f}")
else:
error(f"Tracking failed: {result.get('error', 'unknown')}")
def show_stats():
+2 -2
View File
@@ -137,10 +137,10 @@ def test_handle_command_routes_reauth_google(
def test_handle_command_help_gate(
_warn, _err, _succ, _hdr, _json, _retry, _factory, _auth, mock_console
):
"""handle_command prints help when args=['--help'] and returns True."""
"""handle_command prints help when args=['google', '--help'] and returns True."""
from aipass.api.apps.modules import google_client
result = google_client.handle_command("validate", ["--help"])
result = google_client.handle_command("validate", ["google", "--help"])
assert result is True
# print_help calls console.print with the argparse output
@@ -596,17 +596,17 @@ def test_list_models_formats_free_pricing(mock_console, mock_header, mock_keys,
# =============================================
@patch(f"{_MOD}.warning")
@patch(f"{_MOD}.error")
@patch(f"{_MOD}.header")
@patch(f"{_MOD}.console")
def test_make_call_shows_todo_warning(mock_console, mock_header, mock_warning):
"""make_call currently shows a TODO warning."""
def test_make_call_no_model_shows_error(mock_console, mock_header, mock_error):
"""make_call without --model shows error."""
from aipass.api.apps.modules import openrouter_client
openrouter_client.make_call(["What is AI?"])
mock_warning.assert_called_once()
assert "TODO" in mock_warning.call_args[0][0]
mock_error.assert_called_once()
assert "Model required" in mock_error.call_args[0][0]
# =============================================
+3 -3
View File
@@ -60,7 +60,7 @@ def test_create_caller_config_defaults(tmp_path: Path):
assert config["config"]["ai_max_tokens"] == 4000
assert config["config"]["enabled"] is True
assert config["config"]["ai_model"] == ""
assert config["skill_name"] == "openrouter"
assert config["module_name"] == "openrouter"
def test_create_caller_config_data_defaults(tmp_path: Path):
@@ -87,7 +87,7 @@ def test_create_caller_config_log_defaults(tmp_path: Path):
log = read_json(json_folder / "openrouter_skill_log.json")
assert log is not None
assert log["logs"] == []
assert log["skill_name"] == "openrouter"
assert log["module_name"] == "openrouter"
def test_create_caller_config_creates_folder(tmp_path: Path):
@@ -187,7 +187,7 @@ def test_get_default_caller_config_structure():
"""Default config should have expected structure."""
config = get_default_caller_config()
assert "skill_name" in config
assert "module_name" in config
assert "timestamp" in config
assert "config" in config
assert isinstance(config["config"], dict)
+2 -80
View File
@@ -44,85 +44,6 @@
29
],
"reason": "Scanner needs discovery_handler's get_entry_point() and parse_help_for_commands() to discover commands from branch help output. Justified — scanner's core function depends on discovery's parsing."
},
{
"file": "apps/modules/router.py",
"standard": "naming",
"reason": "branch_path, branch_name, command are local variables inside route_command(), not module-level constants. False positive from naming checker."
},
{
"file": "apps/modules/scan.py",
"standard": "naming",
"pattern": "__all__",
"reason": "__all__ is a standard Python convention (PEP 8) — lowercase dunder is correct."
},
{
"file": "apps/modules/commands.py",
"standard": "naming",
"pattern": "__all__",
"reason": "__all__ is a standard Python convention (PEP 8) — lowercase dunder is correct."
},
{
"file": "apps/modules/config.py",
"standard": "naming",
"pattern": "__all__",
"reason": "__all__ is a standard Python convention (PEP 8) — lowercase dunder is correct."
},
{
"file": "apps/modules/module_registry.py",
"standard": "naming",
"pattern": "__all__",
"reason": "__all__ is a standard Python convention (PEP 8) — lowercase dunder is correct."
},
{
"file": "apps/modules/registry.py",
"standard": "naming",
"pattern": "__all__",
"reason": "__all__ is a standard Python convention (PEP 8) — lowercase dunder is correct."
},
{
"file": "apps/modules/__init__.py",
"standard": "naming",
"pattern": "__all__",
"reason": "__all__ is a standard Python convention (PEP 8) — lowercase dunder is correct."
},
{
"file": "apps/handlers/json/json_handler.py",
"standard": "naming",
"pattern": "__all__",
"reason": "__all__ is a standard Python convention (PEP 8) — lowercase dunder is correct."
},
{
"file": "apps/handlers/scanning/__init__.py",
"standard": "naming",
"pattern": "__all__",
"reason": "__all__ is a standard Python convention (PEP 8) — lowercase dunder is correct."
},
{
"file": "__init__.py",
"standard": "naming",
"pattern": "__all__",
"reason": "__all__ is a standard Python convention (PEP 8) — lowercase dunder is correct."
},
{
"file": "apps/handlers/router_handler.py",
"standard": "naming",
"reason": "cmd_args, executable, args are local variables inside execute_branch_command(), not module-level constants. False positive."
},
{
"file": "apps/handlers/executor.py",
"standard": "naming",
"reason": "run_env, cwd, shell are local variables inside run_command(), not module-level constants. False positive."
},
{
"file": "apps/handlers/registry_handler.py",
"standard": "naming",
"reason": "filtered is a local variable inside get_all_branches(), not a module-level constant. False positive."
},
{
"file": "apps/handlers/command_registry/ops.py",
"standard": "naming",
"reason": "module_name is a local variable, not a module-level constant. False positive."
}
],
"notes": {
@@ -138,6 +59,7 @@
"lines": "Optional - specific line numbers to bypass",
"pattern": "Optional - pattern to match (e.g. 'if __name__')",
"reason": "Required - why this bypass exists"
}
},
"removed_s61": "13 naming bypasses removed (9 __all__ + 4 local variable false positives). Seedgo dispatched to fix naming_check.py. If audit fails on naming after this, the checker fix didn't land."
}
}
+25 -8
View File
@@ -32,14 +32,31 @@ _COMMANDS = ("pr", "status", "sync", "lock", "unlock")
def _detect_branch_dir() -> tuple[str, Path] | None:
"""Detect caller's branch from CWD. Returns (branch_name, branch_dir) or None."""
cwd = Path.cwd()
parts = cwd.parts
for i, part in enumerate(parts):
if part == "aipass" and i > 0 and parts[i - 1] == "src" and i + 1 < len(parts):
branch_name = parts[i + 1]
branch_dir = Path(*parts[: i + 2])
return branch_name, branch_dir
"""Detect caller's branch from CWD via passport lookup.
Walks up from CWD looking for ``.trinity/passport.json`` and extracts
the branch name + directory. Works for any registered branch regardless
of where it lives on disk (commons, skills, aipass sub-dirs, etc.).
"""
current = Path.cwd().resolve()
for _ in range(10):
passport = current / ".trinity" / "passport.json"
if passport.exists():
try:
with open(passport, "r", encoding="utf-8") as fh:
data = json.load(fh)
name = data.get("branch_info", {}).get("branch_name")
if not name:
name = data.get("identity", {}).get("name")
if name:
return name, current
except Exception as exc:
logger.warning("Failed to read passport at %s: %s", passport, exc)
return None
parent = current.parent
if parent == current:
break
current = parent
return None
+37 -9
View File
@@ -585,24 +585,52 @@ class TestGitModuleRouting:
class TestDetectBranchDir:
"""Branch directory detection tests."""
def test_detects_branch_from_path(self, monkeypatch: pytest.MonkeyPatch) -> None:
"""Detects branch name and dir from a path like .../src/aipass/drone/..."""
# Simulate CWD being inside a branch
fake_path = Path("/home/user/Projects/AIPass/src/aipass/drone/apps/modules")
monkeypatch.chdir("/") # Just need a valid dir for the test
with patch("aipass.drone.apps.modules.git_module.Path.cwd", return_value=fake_path):
detected = _detect_branch_dir()
def test_detects_branch_from_passport(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
"""Detects branch name and dir by walking up to .trinity/passport.json."""
# Create a fake branch directory with passport
branch_dir = tmp_path / "mybranch"
trinity = branch_dir / ".trinity"
trinity.mkdir(parents=True)
passport = trinity / "passport.json"
passport.write_text(json.dumps({
"branch_info": {"branch_name": "mybranch"},
}))
# CWD is inside a subdirectory of the branch
sub_dir = branch_dir / "apps" / "modules"
sub_dir.mkdir(parents=True)
monkeypatch.chdir(sub_dir)
detected = _detect_branch_dir()
assert detected is not None
name, bdir = detected
assert name == "drone"
assert name == "mybranch"
assert bdir == branch_dir.resolve()
def test_returns_none_for_unrecognized_path(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
"""Returns None when CWD is not under src/aipass/."""
"""Returns None when CWD has no .trinity/passport.json above it."""
monkeypatch.chdir(tmp_path)
detected = _detect_branch_dir()
assert detected is None
def test_detects_non_aipass_branch(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
"""Detects branches outside src/aipass/ (e.g. commons, skills)."""
branch_dir = tmp_path / "src" / "commons"
trinity = branch_dir / ".trinity"
trinity.mkdir(parents=True)
passport = trinity / "passport.json"
passport.write_text(json.dumps({
"branch_info": {"branch_name": "commons"},
}))
monkeypatch.chdir(branch_dir)
detected = _detect_branch_dir()
assert detected is not None
name, bdir = detected
assert name == "commons"
assert bdir == branch_dir.resolve()
class TestGitModuleHelp:
"""Help and introspection tests."""
@@ -150,5 +150,6 @@
"DPLAN-0059_seedgo_test_quality_standard_deep_coverage_en_2026-03-24.md": "2026-03-28T19:52:31.066363",
"DPLAN-0060_custom_function_scanner_seedgo_module_for_bra_2026-03-24.md": "2026-03-28T19:52:49.501398",
"DPLAN-0061_expand_test_quality_checker_to_all_standard_c_2026-03-24.md": "2026-03-28T19:53:13.017860",
"FPLAN-0138_s59_walkthrough_fixes_2026-03-28.md": "2026-03-28T19:53:32.097935"
"FPLAN-0138_s59_walkthrough_fixes_2026-03-28.md": "2026-03-28T19:53:32.097935",
"FPLAN-0139_dplan_0029_branch_audit_cleanup_fix_all_remai_2026-03-28.md": "2026-03-28T22:57:07.425893"
}
+7
View File
@@ -41,6 +41,13 @@
"subject": "Monitor Phase 2: caller attribution via drone CALLER markers",
"date_closed": "2026-03-19",
"location": "prax"
},
{
"plan_id": "FPLAN-0071",
"type": "FPLAN",
"subject": "Self-audit compliance sprint",
"date_closed": "2026-03-28",
"location": "prax"
}
]
}
@@ -13,10 +13,9 @@ Validates module compliance with AIPass naming standards.
Checks file naming, function naming, variable naming, constant naming.
"""
import sys
import re
from pathlib import Path
from typing import Dict, List, Optional
from typing import Dict, Optional
from aipass.prax import logger
from aipass.seedgo.apps.handlers.json import json_handler
@@ -90,7 +89,6 @@ def check_module(module_path: str, bypass_rules: list | None = None) -> Dict:
try:
with open(path, 'r', encoding='utf-8') as f:
content = f.read()
lines = content.split('\n')
except Exception as e:
logger.info("Cannot read %s: %s", path, e)
return {
@@ -236,49 +234,6 @@ def check_function_naming(content: str) -> Optional[Dict]:
}
def _iter_module_level_lines(lines: List[str]) -> List[str]:
result = []
in_function_or_class = False
in_multiline_string = False
multiline_delimiter = None
current_indentation = 0
for line in lines:
stripped = line.strip()
if '"""' in stripped or "'''" in stripped:
delimiter = '"""' if '"""' in stripped else "'''"
count = stripped.count(delimiter)
if count % 2 == 1 and not in_multiline_string:
in_multiline_string = True
multiline_delimiter = delimiter
elif count % 2 == 1 and in_multiline_string and multiline_delimiter == delimiter:
in_multiline_string = False
multiline_delimiter = None
if in_multiline_string:
continue
if (stripped.startswith('def ') or
stripped.startswith('class ') or
stripped.startswith('async def ') or
stripped.startswith('if __name__')):
in_function_or_class = True
current_indentation = len(line) - len(line.lstrip())
continue
if in_function_or_class and not line.startswith(' ') and not line.startswith('\t') and stripped:
in_function_or_class = False
current_indentation = 0
if in_function_or_class:
continue
result.append(stripped)
return result
def check_constant_naming(content: str) -> Optional[Dict]:
"""
@@ -286,8 +241,9 @@ def check_constant_naming(content: str) -> Optional[Dict]:
Rules:
- UPPER_CASE for module-level constants
- Assigned outside of functions/classes
- Assigned outside of functions/classes (column 0 only)
- Excludes module imports (from X import Y as Z)
- Excludes __dunder__ variables (PEP 8 convention: always lowercase)
"""
# First pass: collect imported names to exclude from constant checking
imported_names = set()
@@ -306,13 +262,34 @@ def check_constant_naming(content: str) -> Optional[Dict]:
# Direct import like "console"
imported_names.add(item.strip())
# Second pass: find module-level assignments (not inside functions/classes)
module_lines = _iter_module_level_lines(content.split('\n'))
# Second pass: find TRUE module-level assignments
# Only check lines at column 0 (no indentation) — this eliminates local
# variables inside functions/classes, which are always indented.
# The old _iter_module_level_lines approach had a bug: multi-line function
# signatures with closing ) at column 0 tricked it into thinking the
# function body had ended.
constants = []
bad_constants = []
in_multiline_string = False
for stripped in module_lines:
# Find assignments at module level
for line in content.split('\n'):
stripped = line.strip()
# Track multiline strings
for delimiter in ('"""', "'''"):
if delimiter in stripped:
count = stripped.count(delimiter)
if count % 2 == 1:
in_multiline_string = not in_multiline_string
if in_multiline_string:
continue
# Only consider lines with zero indentation (true module-level)
if line and (line[0] == ' ' or line[0] == '\t'):
continue
# Find assignments
if '=' not in stripped or stripped.startswith('#'):
continue
@@ -324,6 +301,10 @@ def check_constant_naming(content: str) -> Optional[Dict]:
const_name = match.group(1)
assigned_value = match.group(2)
# Skip __dunder__ variables (__all__, __version__, etc.)
if const_name.startswith('__') and const_name.endswith('__'):
continue
# Skip if this is an imported name (like logger, console)
if const_name in imported_names:
continue
+2 -2
View File
@@ -97,7 +97,7 @@ def deep_merge(template_data: Any, existing_data: Any) -> Any:
def backup_json(file_path: Path) -> Path:
"""Create a timestamped backup of a JSON file before modifying.
The backup is placed in a ``.backup/`` directory alongside the file,
The backup is placed in a ``.recovery/`` directory alongside the file,
named with a timestamp suffix.
Args:
@@ -116,7 +116,7 @@ def backup_json(file_path: Path) -> Path:
raise FileNotFoundError(f"Cannot backup — file not found: {file_path}")
# Create backup directory alongside the file
backup_dir = file_path.parent / ".backup"
backup_dir = file_path.parent / ".recovery"
backup_dir.mkdir(parents=True, exist_ok=True)
# Timestamped backup name
+2 -2
View File
@@ -615,7 +615,7 @@ class TestBackupJson:
"""Tests for backup_json()."""
def test_creates_backup_file(self, tmp_path):
"""Backup should create a copy in .backup/ directory."""
"""Backup should create a copy in .recovery/ directory."""
from aipass.spawn.apps.handlers.json_ops import backup_json
source = tmp_path / "data.json"
@@ -624,7 +624,7 @@ class TestBackupJson:
backup_path = backup_json(source)
assert backup_path.exists()
assert ".backup" in str(backup_path)
assert ".recovery" in str(backup_path)
assert "data.json" in backup_path.name
assert ".backup" in backup_path.name