Loading...
Loading...
L3 Worker. Audits layer boundaries + cross-layer consistency: I/O violations, transaction boundaries (commit ownership), session ownership (DI vs local), async consistency (sync I/O in async), fire-and-forget tasks.
npx skill4agent add levnikolaevich/claude-code-skills ln-642-layer-boundary-auditor- architecture_path: string # Path to docs/architecture.md
- codebase_root: string # Root directory to scan
- skip_violations: string[] # Files to skip (legacy)Read docs/architecture.md
Extract from Section 4.2 (Top-Level Decomposition):
- architecture_type: "Layered" | "Hexagonal" | "Clean" | "MVC" | etc.
- layers: [{name, directories[], purpose}]
Extract from Section 5.3 (Infrastructure Layer Components):
- infrastructure_components: [{name, responsibility}]
IF architecture.md not found:
Use fallback presets from common_patterns.md
Build ruleset:
FOR EACH layer:
allowed_deps = layers that can be imported
forbidden_deps = layers that cannot be importedFOR EACH violation_type IN common_patterns.md I/O Pattern Boundary Rules:
grep_pattern = violation_type.detection_grep
forbidden_dirs = violation_type.forbidden_in
matches = Grep(grep_pattern, codebase_root, include="*.py,*.ts,*.js")
FOR EACH match IN matches:
IF match.path NOT IN skip_violations:
IF any(forbidden IN match.path FOR forbidden IN forbidden_dirs):
violations.append({
type: "layer_violation",
severity: "HIGH",
pattern: violation_type.name,
file: match.path,
line: match.line,
code: match.context,
allowed_in: violation_type.allowed_in,
suggestion: f"Move to {violation_type.allowed_in}"
})repo_commits = Grep("\.commit\(\)|\.rollback\(\)", "**/repositories/**/*.py")
service_commits = Grep("\.commit\(\)|\.rollback\(\)", "**/services/**/*.py")
api_commits = Grep("\.commit\(\)|\.rollback\(\)", "**/api/**/*.py")
layers_with_commits = count([repo_commits, service_commits, api_commits].filter(len > 0))_callbacks.py# UoW boundary| Condition | Severity | Issue |
|---|---|---|
| layers_with_commits >= 3 | CRITICAL | Mixed UoW ownership across all layers |
| repo + api commits | HIGH | Transaction control bypasses service layer |
| repo + service commits | HIGH | Ambiguous UoW owner (repo vs service) |
| service + api commits | MEDIUM | Transaction control spans service + API |
di_session = Grep("Depends\(get_session\)|Depends\(get_db\)", "**/api/**/*.py")
local_session = Grep("AsyncSessionLocal\(\)|async_sessionmaker", "**/services/**/*.py")
local_in_repo = Grep("AsyncSessionLocal\(\)", "**/repositories/**/*.py")| Condition | Severity | Issue |
|---|---|---|
| di_session AND local_in_repo in same module | HIGH | Repo creates own session while API injects different |
| local_session in service calling DI-based repo | MEDIUM | Session mismatch in call chain |
# For each file with "async def":
sync_file_io = Grep("\.read_bytes\(\)|\.read_text\(\)|\.write_bytes\(\)|\.write_text\(\)", file)
sync_open = Grep("(?<!aiofiles\.)open\(", file) # open() not preceded by aiofiles.
# Safe patterns (not violations):
# - "asyncio.to_thread(" wrapping the call
# - "await aiofiles.open("
# - "run_in_executor(" wrapping the call| Pattern | Severity | Issue |
|---|---|---|
| Path.read_bytes() in async def | HIGH | Blocking file read in async context |
| open() without aiofiles in async def | HIGH | Blocking file operation |
| time.sleep() in async def | HIGH | Blocking sleep (use asyncio.sleep) |
asyncio.to_thread()aiofilesall_tasks = Grep("create_task\(", codebase)
# For each match, check context:
# - Has .add_done_callback() → OK
# - Assigned to variable with later await → OK
# - Has "# fire-and-forget" comment → OK (documented intent)
# - None of above → VIOLATION| Pattern | Severity | Issue |
|---|---|---|
| create_task() without handler or comment | MEDIUM | Unhandled task exception possible |
| create_task() in loop without error collection | HIGH | Multiple silent failures possible |
task.add_done_callback(handle_exception)# HTTP Client Coverage
all_http_calls = Grep("httpx\\.|aiohttp\\.|requests\\.", codebase_root)
abstracted_calls = Grep("client\\.(get|post|put|delete)", infrastructure_dirs)
IF len(all_http_calls) > 0:
coverage = len(abstracted_calls) / len(all_http_calls) * 100
IF coverage < 90%:
violations.append({
type: "low_coverage",
severity: "MEDIUM",
pattern: "HTTP Client Abstraction",
coverage: coverage,
uncovered_files: files with direct calls outside infrastructure
})
# Error Handling Duplication
http_error_handlers = Grep("except\\s+(httpx\\.|aiohttp\\.|requests\\.)", codebase_root)
unique_files = set(f.path for f in http_error_handlers)
IF len(unique_files) > 2:
violations.append({
type: "duplication",
severity: "MEDIUM",
pattern: "HTTP Error Handling",
files: list(unique_files),
suggestion: "Centralize in infrastructure layer"
})shared/references/audit_scoring.md{
"category": "Layer Boundary",
"score": 4.5,
"total_issues": 8,
"critical": 1,
"high": 3,
"medium": 4,
"low": 0,
"architecture": {
"type": "Layered",
"layers": ["api", "services", "domain", "infrastructure"]
},
"checks": [
{"id": "io_isolation", "name": "I/O Isolation", "status": "failed", "details": "HTTP client found in domain layer"},
{"id": "http_abstraction", "name": "HTTP Abstraction", "status": "warning", "details": "75% coverage, 3 direct calls outside infrastructure"},
{"id": "error_centralization", "name": "Error Centralization", "status": "failed", "details": "HTTP error handlers in 4 files, should be centralized"},
{"id": "transaction_boundary", "name": "Transaction Boundary", "status": "failed", "details": "commit() in repos (3), services (2), api (4) - mixed UoW ownership"},
{"id": "session_ownership", "name": "Session Ownership", "status": "passed", "details": "DI-based sessions used consistently"},
{"id": "async_consistency", "name": "Async Consistency", "status": "failed", "details": "Blocking I/O in async functions detected"},
{"id": "fire_and_forget", "name": "Fire-and-Forget Handling", "status": "warning", "details": "2 tasks without error handlers"}
],
"findings": [
{
"severity": "CRITICAL",
"location": "app/",
"issue": "Mixed UoW ownership: commit() found in repositories (3), services (2), api (4)",
"principle": "Layer Boundary / Transaction Control",
"recommendation": "Choose single UoW owner (service layer recommended), remove commit() from other layers",
"effort": "L"
},
{
"severity": "HIGH",
"location": "app/services/job/service.py:45",
"issue": "Blocking file I/O in async: Path.read_bytes() inside async def process_job()",
"principle": "Layer Boundary / Async Consistency",
"recommendation": "Use asyncio.to_thread(path.read_bytes) or aiofiles",
"effort": "S"
},
{
"severity": "HIGH",
"location": "app/domain/pdf/parser.py:45",
"issue": "Layer violation: HTTP client used in domain layer",
"principle": "Layer Boundary / I/O Isolation",
"recommendation": "Move httpx.AsyncClient to infrastructure/http/clients/",
"effort": "M"
},
{
"severity": "MEDIUM",
"location": "app/api/v1/jobs.py:78",
"issue": "Fire-and-forget task without error handler: create_task(notify_user())",
"principle": "Layer Boundary / Task Error Handling",
"recommendation": "Add task.add_done_callback(handle_exception) or document with # fire-and-forget comment",
"effort": "S"
}
],
"coverage": {
"http_abstraction": 75,
"error_centralization": false,
"transaction_boundary_consistent": false,
"session_ownership_consistent": true,
"async_io_consistent": false,
"fire_and_forget_handled": false
}
}../ln-640-pattern-evolution-auditor/references/common_patterns.md../ln-640-pattern-evolution-auditor/references/scoring_rules.md