from datetime import datetime
from typing import Dict
from .schemas import FlowInput, LintResult, FlowMeta, QuickFix, Finding
from .utils import calculate_flow_hash
from .rules import collect_findings
# In-memory cache for idempotency
# Key: flow_hash, Value: LintResult
_RESULT_CACHE: Dict[str, LintResult] = {}
def lint_flow(flow: FlowInput) -> LintResult:
# 1. Hashing & Idempotency
flow_hash = calculate_flow_hash(flow)
if flow_hash in _RESULT_CACHE:
return _RESULT_CACHE[flow_hash]
# 2. Run Rules
errors, warnings = collect_findings(flow)
# 3. Calculate Score
# Start: 100
# Error: -15
# Warning: -5
# Min: 0, Max: 100
initial_score = 100
deduction = (len(errors) * 15) + (len(warnings) * 5)
score = max(0, min(100, initial_score - deduction))
# 4. Generate Meta
meta = FlowMeta(
nodes=len(flow.nodes),
edges=len(flow.edges),
unreachable_count=sum(1 for e in errors if e.code == "unreachable_node"),
dead_end_count=sum(1 for e in errors if e.code == "dead_end_not_end"),
created_at=datetime.utcnow().isoformat()
)
# 5. Generate Quick Fixes (Stub/Simple Logic)
quick_fixes = []
for err in errors:
if err.fix:
# Simple instruction patch
quick_fixes.append(QuickFix(
node_id=err.node_id or "global",
patch_type="instructions",
patch=err.fix
))
for warn in warnings:
if warn.fix:
quick_fixes.append(QuickFix(
node_id=warn.node_id or "global",
patch_type="instructions",
patch=warn.fix
))
# 6. Construct Result
result = LintResult(
flow_hash=flow_hash,
score=score,
errors=errors,
warnings=warnings,
quick_fixes=quick_fixes,
flow_meta=meta
)
# 7. Update Cache
_RESULT_CACHE[flow_hash] = result
return result
