""" Backend bridge to delimit-gateway core engine. Adapter Boundary Contract v1.0: - Pure translation layer: no governance logic here - Deterministic error on failure (never swallow) - Zero state (stateless between calls) - No schema forking (gateway types are canonical) """ import os import re import sys import json import logging from pathlib import Path from typing import Any, Dict, List, Optional logger = logging.getLogger("delimit.ai.gateway_core") # LED-1265: identity-string redaction filter for changelog output. Patterns are # loaded from the DELIMIT_CHANGELOG_REDACT_PATTERNS env var (semicolon-separated # regex|replacement pairs, e.g. "FOO|[redacted];BAR|[redacted]"), NOT hardcoded. # Why env-var-driven: hardcoding the patterns inline would itself constitute the # leak the filter exists to prevent — the patterns must match the strings being # redacted, so committing them to source recreates the leak. When the env var is # unset (customer machines), this is a no-op pass-through. The internal gateway # sets the env var at process start with the exact patterns to scrub before any # auto-generated changelog reaches a public surface. def _load_redaction_patterns() -> List[tuple]: """Parse DELIMIT_CHANGELOG_REDACT_PATTERNS env var into compiled patterns. Format: semicolon-separated `regex|replacement` pairs. Empty / unset returns []. Invalid regexes are warned and skipped (filter is fail-open: better to skip a bad pattern than block all output). """ raw = os.environ.get("DELIMIT_CHANGELOG_REDACT_PATTERNS", "").strip() if not raw: return [] patterns = [] for entry in raw.split(";"): entry = entry.strip() if not entry or "|" not in entry: continue pat, _, repl = entry.partition("|") try: patterns.append((re.compile(pat), repl)) except re.error as exc: logger.warning("ignoring invalid changelog redaction pattern %r: %s", pat, exc) return patterns _IDENTITY_STRING_PATTERNS = _load_redaction_patterns() def _redact_identity_strings(text: str) -> str: """Apply env-var-loaded redaction patterns to text. No-op when env var unset. Defensive filter for changelog generation: immutable commit messages on public repos may contain identity strings the auto-generated CHANGELOG would re-leak to a fresh public surface. The filter is opt-in per gateway deployment via env var; customers don't need to set it. """ if not text or not _IDENTITY_STRING_PATTERNS: return text for pattern, replacement in _IDENTITY_STRING_PATTERNS: text = pattern.sub(replacement, text) return text # Add gateway root to path so we can import core modules GATEWAY_ROOT = Path(__file__).resolve().parent.parent.parent if str(GATEWAY_ROOT) not in sys.path: sys.path.insert(0, str(GATEWAY_ROOT)) def _load_specs(spec_path: str) -> Dict[str, Any]: """Load an API spec (OpenAPI or JSON Schema) from a file path. Performs a non-fatal version compatibility check (LED-290) so that unknown OpenAPI versions log a warning instead of silently parsing. JSON Schema documents skip the OpenAPI version assert. """ import yaml p = Path(spec_path) if not p.exists(): raise FileNotFoundError(f"Spec file not found: {spec_path}") content = p.read_text(encoding="utf-8") if p.suffix in (".yaml", ".yml"): spec = yaml.safe_load(content) else: spec = json.loads(content) # LED-290: warn (non-fatal) if version is outside the validated set. # Only applies to OpenAPI/Swagger documents — bare JSON Schema files # have no "openapi"/"swagger" key and would otherwise trip the assert. try: if isinstance(spec, dict) and ("openapi" in spec or "swagger" in spec): from core.openapi_version import assert_supported assert_supported(spec, strict=False) except Exception as exc: # pragma: no cover -- defensive only logger.debug("openapi version check skipped: %s", exc) return spec # --------------------------------------------------------------------------- # LED-713: JSON Schema spec-type dispatch helpers # --------------------------------------------------------------------------- def _spec_type(doc: Any) -> str: """Classify a loaded spec doc. 'openapi' or 'json_schema'.""" from core.spec_detector import detect_spec_type t = detect_spec_type(doc) # Fallback to openapi for unknown so we never break existing flows. return "json_schema" if t == "json_schema" else "openapi" def _json_schema_changes_to_dicts(changes: List[Any]) -> List[Dict[str, Any]]: return [ { "type": c.type.value, "path": c.path, "message": c.message, "is_breaking": c.is_breaking, "details": c.details, } for c in changes ] def _json_schema_semver(changes: List[Any]) -> Dict[str, Any]: """Build an OpenAPI-compatible semver result from JSON Schema changes. Mirrors core.semver_classifier.classify_detailed shape so downstream consumers (PR comment, CI formatter, ledger) don't need to branch. """ breaking = [c for c in changes if c.is_breaking] non_breaking = [c for c in changes if not c.is_breaking] if breaking: bump = "major" elif non_breaking: bump = "minor" else: bump = "none" return { "bump": bump, "is_breaking": bool(breaking), "counts": { "breaking": len(breaking), "non_breaking": len(non_breaking), "total": len(changes), }, } def _bump_semver_version(current: str, bump: str) -> Optional[str]: """Minimal semver bump for JSON Schema path (core.semver_classifier only understands OpenAPI ChangeType enums).""" if not current: return None try: parts = current.lstrip("v").split(".") major, minor, patch = (int(parts[0]), int(parts[1]), int(parts[2])) except Exception: return None if bump == "major": return f"{major + 1}.0.0" if bump == "minor": return f"{major}.{minor + 1}.0" if bump == "patch": return f"{major}.{minor}.{patch + 1}" return current def _run_json_schema_lint( old_doc: Dict[str, Any], new_doc: Dict[str, Any], current_version: Optional[str] = None, api_name: Optional[str] = None, ) -> Dict[str, Any]: """Build an evaluate_with_policy-compatible result for JSON Schema. Policy rules in Delimit are defined against OpenAPI ChangeType values, so they do not apply here. We return zero violations and rely on the breaking-change count + semver bump to drive the governance gate. """ from core.json_schema_diff import JSONSchemaDiffEngine engine = JSONSchemaDiffEngine() changes = engine.compare(old_doc, new_doc) semver = _json_schema_semver(changes) if current_version: semver["current_version"] = current_version semver["next_version"] = _bump_semver_version(current_version, semver["bump"]) breaking_count = semver["counts"]["breaking"] total = semver["counts"]["total"] decision = "pass" exit_code = 0 # No policy rules apply to JSON Schema, but breaking changes still # flag MAJOR semver and the downstream gate uses that to block. # Mirror the shape of evaluate_with_policy so the action/CLI renderers # need no JSON Schema-specific branch. result: Dict[str, Any] = { "spec_type": "json_schema", "api_name": api_name or new_doc.get("title") or old_doc.get("title") or "JSON Schema", "decision": decision, "exit_code": exit_code, "violations": [], "summary": { "total_changes": total, "breaking_changes": breaking_count, "violations": 0, "errors": 0, "warnings": 0, }, "all_changes": [ { "type": c.type.value, "path": c.path, "message": c.message, "is_breaking": c.is_breaking, } for c in changes ], "semver": semver, } return result def _read_jsonl(path: Path) -> List[Dict[str, Any]]: """Read JSONL entries from a file, skipping malformed lines.""" items: List[Dict[str, Any]] = [] if not path.exists(): return items try: with open(path, "r", encoding="utf-8") as handle: for line in handle: stripped = line.strip() if not stripped: continue try: payload = json.loads(stripped) except json.JSONDecodeError: continue if isinstance(payload, dict): items.append(payload) except OSError: return [] return items def _query_project_ledger_fallback(ledger_path: Path) -> Optional[Dict[str, Any]]: """Fallback for project-local ledgers that use operations/strategy jsonl files.""" if ledger_path.name != "events.jsonl": return None ledger_dir = ledger_path.parent operations = _read_jsonl(ledger_dir / "operations.jsonl") strategy = _read_jsonl(ledger_dir / "strategy.jsonl") combined = operations + strategy if not combined: return None latest = combined[-1] return { "path": str(ledger_path), "event_count": len(combined), "latest_event": latest, "storage_mode": "project_local_ledger", "ledger_files": [ str(p) for p in (ledger_dir / "operations.jsonl", ledger_dir / "strategy.jsonl") if p.exists() ], "chain_valid": True, } def run_spec_health(spec_path: str) -> Dict[str, Any]: """Score a single OpenAPI spec on quality dimensions. Returns overall score, letter grade, per-dimension scores, and actionable recommendations. """ from core.spec_health import score_spec spec = _load_specs(spec_path) return score_spec(spec) def run_lint(old_spec: str, new_spec: str, policy_file: Optional[str] = None) -> Dict[str, Any]: """Run the full lint pipeline: diff + policy evaluation. This is the Tier 1 primary tool — combines diff detection with policy enforcement into a single pass/fail decision. Auto-detects spec type (OpenAPI vs JSON Schema, LED-713) and dispatches to the matching engine. """ from core.policy_engine import evaluate_with_policy old = _load_specs(old_spec) new = _load_specs(new_spec) # LED-713: JSON Schema dispatch. Policy rules are OpenAPI-specific, # so JSON Schema takes the no-policy (breaking-count + semver) path. if _spec_type(new) == "json_schema" or _spec_type(old) == "json_schema": return _run_json_schema_lint(old, new) return evaluate_with_policy(old, new, policy_file) def run_diff(old_spec: str, new_spec: str) -> Dict[str, Any]: """Run diff engine only — no policy evaluation. Auto-detects OpenAPI vs JSON Schema and dispatches (LED-713). """ old = _load_specs(old_spec) new = _load_specs(new_spec) if _spec_type(new) == "json_schema" or _spec_type(old) == "json_schema": from core.json_schema_diff import JSONSchemaDiffEngine engine = JSONSchemaDiffEngine() changes = engine.compare(old, new) breaking = [c for c in changes if c.is_breaking] return { "spec_type": "json_schema", "total_changes": len(changes), "breaking_changes": len(breaking), "changes": _json_schema_changes_to_dicts(changes), } from core.diff_engine_v2 import OpenAPIDiffEngine engine = OpenAPIDiffEngine() changes = engine.compare(old, new) breaking = [c for c in changes if c.is_breaking] return { "spec_type": "openapi", "total_changes": len(changes), "breaking_changes": len(breaking), "changes": [ { "type": c.type.value, "path": c.path, "message": c.message, "is_breaking": c.is_breaking, "details": c.details, } for c in changes ], # LED-1588: structured side-channel for fail-open skips (unresolvable # refs, malformed nodes) so a clean `changes` list is not mistaken for # "proven safe". Additive — existing keys are unchanged. The JSON # Schema branch above intentionally omits this (engine does not yet # populate advisories; a misleading empty field would imply it checked). "advisories": engine.advisories, } def run_changelog( old_spec: str, new_spec: str, fmt: str = "markdown", version: str = "", ) -> Dict[str, Any]: """Generate a changelog from API spec changes. Uses the diff engine to detect changes, then formats them into a human-readable changelog grouped by category. """ from datetime import datetime, timezone old = _load_specs(old_spec) new = _load_specs(new_spec) # LED-713: dispatch on spec type. JSONSchemaChange / Change share the # (.type.value, .path, .message, .is_breaking) duck type. if _spec_type(new) == "json_schema" or _spec_type(old) == "json_schema": from core.json_schema_diff import JSONSchemaDiffEngine engine = JSONSchemaDiffEngine() else: from core.diff_engine_v2 import OpenAPIDiffEngine engine = OpenAPIDiffEngine() changes = engine.compare(old, new) # Categorize changes breaking = [] features = [] deprecations = [] fixes = [] for c in changes: entry = { "type": c.type.value, "path": c.path, "message": c.message, "is_breaking": c.is_breaking, } if c.type.value == "deprecated_added": deprecations.append(entry) elif c.is_breaking: breaking.append(entry) elif c.type.value in ( "endpoint_added", "method_added", "optional_param_added", "response_added", "optional_field_added", "enum_value_added", "security_added", ): features.append(entry) else: fixes.append(entry) date_str = datetime.now(timezone.utc).strftime("%Y-%m-%d") version_label = version or "Unreleased" if fmt == "json": return { "format": "json", "version": version_label, "date": date_str, "total_changes": len(changes), "sections": { "breaking_changes": breaking, "new_features": features, "deprecations": deprecations, "other_changes": fixes, }, } if fmt == "keepachangelog": lines = [f"## [{version_label}] - {date_str}", ""] if breaking: lines.append("### Removed / Breaking") for e in breaking: lines.append(f"- {e['message']} (`{e['path']}`)") lines.append("") if features: lines.append("### Added") for e in features: lines.append(f"- {e['message']} (`{e['path']}`)") lines.append("") if deprecations: lines.append("### Deprecated") for e in deprecations: lines.append(f"- {e['message']} (`{e['path']}`)") lines.append("") if fixes: lines.append("### Changed") for e in fixes: lines.append(f"- {e['message']} (`{e['path']}`)") lines.append("") return { "format": "keepachangelog", "version": version_label, "date": date_str, "total_changes": len(changes), "changelog": "\n".join(lines), } if fmt == "github-release": lines = [] if breaking: lines.append("## :warning: Breaking Changes") for e in breaking: lines.append(f"- {e['message']} (`{e['path']}`)") lines.append("") if features: lines.append("## :rocket: New Features") for e in features: lines.append(f"- {e['message']} (`{e['path']}`)") lines.append("") if deprecations: lines.append("## :no_entry_sign: Deprecations") for e in deprecations: lines.append(f"- {e['message']} (`{e['path']}`)") lines.append("") if fixes: lines.append("## :wrench: Other Changes") for e in fixes: lines.append(f"- {e['message']} (`{e['path']}`)") lines.append("") return { "format": "github-release", "version": version_label, "date": date_str, "total_changes": len(changes), "changelog": "\n".join(lines), } # Default: markdown lines = [f"# Changelog — {version_label} ({date_str})", ""] if breaking: lines.append("## Breaking Changes") for e in breaking: lines.append(f"- **{e['type']}**: {e['message']} (`{e['path']}`)") lines.append("") if features: lines.append("## New Features") for e in features: lines.append(f"- **{e['type']}**: {e['message']} (`{e['path']}`)") lines.append("") if deprecations: lines.append("## Deprecations") for e in deprecations: lines.append(f"- **{e['type']}**: {e['message']} (`{e['path']}`)") lines.append("") if fixes: lines.append("## Other Changes") for e in fixes: lines.append(f"- **{e['type']}**: {e['message']} (`{e['path']}`)") lines.append("") return { "format": "markdown", "version": version_label, "date": date_str, "total_changes": len(changes), "changelog": "\n".join(lines), } def run_changelog_from_git( repo_path: str = ".", version: str = "", fmt: str = "keepachangelog", since_tag: str = "", include_ledger: bool = True, output_file: str = "", ) -> Dict[str, Any]: """Generate a changelog from git commits and ledger items. Reads git log since the last tag (or a specified tag), categorizes commits by conventional-commit prefix, optionally pulls completed ledger items, and formats as Markdown. Works for ANY git repo, not just Delimit's own. """ import re import subprocess from datetime import datetime, timezone repo = Path(repo_path).resolve() if not (repo / ".git").exists(): return {"error": "not_a_git_repo", "message": f"{repo} is not a git repository."} # --- Resolve the base tag --- if since_tag: base_tag = since_tag else: try: result = subprocess.run( ["git", "describe", "--tags", "--abbrev=0"], cwd=str(repo), capture_output=True, text=True, timeout=10, ) base_tag = result.stdout.strip() if result.returncode == 0 else "" except Exception: base_tag = "" # --- Get git log --- git_log_cmd = ["git", "log", "--pretty=format:%H|%s|%an", "--no-merges"] if base_tag: git_log_cmd.append(f"{base_tag}..HEAD") try: result = subprocess.run( git_log_cmd, cwd=str(repo), capture_output=True, text=True, timeout=30, ) raw_lines = result.stdout.strip().split("\n") if result.stdout.strip() else [] except Exception as e: return {"error": "git_log_failed", "message": str(e)} # --- Get diff stats --- diff_stat_cmd = ["git", "diff", "--stat"] if base_tag: diff_stat_cmd.append(f"{base_tag}..HEAD") else: diff_stat_cmd.append("--cached") # fallback: staged changes try: stat_result = subprocess.run( diff_stat_cmd, cwd=str(repo), capture_output=True, text=True, timeout=15, ) stat_summary = stat_result.stdout.strip().split("\n")[-1] if stat_result.stdout.strip() else "" except Exception: stat_summary = "" # --- Parse and categorize commits --- # Conventional commit pattern: type(scope): message OR type: message cc_pattern = re.compile( r"^(?Pfeat|fix|refactor|docs|test|tests|ci|chore|perf|style|build|revert)" r"(?:\([^)]*\))?:\s*(?P.+)$", re.IGNORECASE, ) categories = { "feat": [], "fix": [], "refactor": [], "docs": [], "test": [], "ci": [], "chore": [], "other": [], } # Keyword fallback patterns for non-conventional commits keyword_map = [ (re.compile(r"\b(add|feature|implement|new)\b", re.I), "feat"), (re.compile(r"\b(fix|bug|patch|resolve|close)\b", re.I), "fix"), (re.compile(r"\b(refactor|restructure|clean|simplify)\b", re.I), "refactor"), (re.compile(r"\b(doc|readme|comment|jsdoc)\b", re.I), "docs"), (re.compile(r"\b(test|spec|coverage|assert)\b", re.I), "test"), (re.compile(r"\b(ci|workflow|action|pipeline|deploy)\b", re.I), "ci"), ] commits_parsed = [] for line in raw_lines: parts = line.split("|", 2) if len(parts) < 3: continue sha, subject, author = parts[0], parts[1], parts[2] m = cc_pattern.match(subject) if m: ctype = m.group("type").lower() if ctype in ("tests",): ctype = "test" msg = m.group("msg") else: # Keyword fallback ctype = "other" msg = subject for pattern, cat in keyword_map: if pattern.search(subject): ctype = cat break # LED-1265: redact founder-holdco identity strings from the commit # subject (msg) and author before they enter any output path. msg = _redact_identity_strings(msg) author = _redact_identity_strings(author) bucket = ctype if ctype in categories else "other" entry = {"sha": sha[:8], "message": msg, "author": author, "category": bucket} categories[bucket].append(entry) commits_parsed.append(entry) # --- Pull completed ledger items (if requested and ledger exists) --- ledger_items = [] if include_ledger: try: ledger_dir = Path.home() / ".delimit" / "ledger" ops_file = ledger_dir / "operations.jsonl" if ops_file.exists(): import json as _json items_raw = [] for ln in ops_file.read_text().splitlines(): ln = ln.strip() if ln: try: items_raw.append(_json.loads(ln)) except _json.JSONDecodeError: continue # Build current state by replaying events state = {} for item in items_raw: item_id = item.get("id", "") if item.get("type") == "update": if item_id in state: if "status" in item: state[item_id]["status"] = item["status"] elif item_id: state[item_id] = item # Find the timestamp of the base tag to filter ledger items tag_dt = None if base_tag: try: ts_result = subprocess.run( ["git", "log", "-1", "--format=%aI", base_tag], cwd=str(repo), capture_output=True, text=True, timeout=10, ) if ts_result.returncode == 0 and ts_result.stdout.strip(): tag_dt = datetime.fromisoformat(ts_result.stdout.strip()) except Exception: pass for item_id, item in state.items(): if item.get("status") == "done": created = item.get("created_at", "") # If we have a tag datetime, only include items created after it if tag_dt and created: try: # Normalize "Z" suffix to "+00:00" for fromisoformat created_norm = created.replace("Z", "+00:00") if created.endswith("Z") else created created_dt = datetime.fromisoformat(created_norm) if created_dt < tag_dt: continue except (ValueError, TypeError): pass # If parsing fails, include the item # LED-1265: ledger titles may contain identity strings # (e.g. LED items filed before the doctrine bound). ledger_items.append({ "id": item_id, "title": _redact_identity_strings(item.get("title", "")), "priority": item.get("priority", ""), }) except Exception: pass # Ledger is optional; failing silently is fine # --- Count stats --- test_commits = len(categories["test"]) total_commits = len(commits_parsed) # Parse files changed from stat summary files_changed = 0 insertions = 0 deletions = 0 if stat_summary: fc_match = re.search(r"(\d+) files? changed", stat_summary) ins_match = re.search(r"(\d+) insertions?\(\+\)", stat_summary) del_match = re.search(r"(\d+) deletions?\(-\)", stat_summary) if fc_match: files_changed = int(fc_match.group(1)) if ins_match: insertions = int(ins_match.group(1)) if del_match: deletions = int(del_match.group(1)) date_str = datetime.now(timezone.utc).strftime("%Y-%m-%d") version_label = version or "Unreleased" stats = { "total_commits": total_commits, "files_changed": files_changed, "insertions": insertions, "deletions": deletions, "tests_added": test_commits, "base_tag": base_tag or "(initial)", } # --- Category display names (keepachangelog style) --- section_map = { "feat": "Added", "fix": "Fixed", "refactor": "Changed", "docs": "Documentation", "test": "Tests", "ci": "CI/CD", "chore": "Chores", "other": "Other", } # --- JSON format --- if fmt == "json": return { "format": "json", "version": version_label, "date": date_str, "stats": stats, "categories": {k: v for k, v in categories.items() if v}, "ledger_items": ledger_items, } # --- Markdown formats --- lines = [] if fmt == "github-release": # No top-level header for GH release (title comes from the release itself) pass else: lines.append(f"## [{version_label}] - {date_str}") lines.append("") for cat_key in ("feat", "fix", "refactor", "docs", "test", "ci", "chore", "other"): entries = categories[cat_key] if not entries: continue section_name = section_map[cat_key] lines.append(f"### {section_name}") for e in entries: lines.append(f"- {e['message']} ({e['sha']})") lines.append("") if ledger_items: lines.append("### Completed Ledger Items") for item in ledger_items: priority_tag = f"[{item['priority']}] " if item.get("priority") else "" lines.append(f"- **{item['id']}**: {priority_tag}{item['title']}") lines.append("") # Stats footer lines.append("### Stats") lines.append(f"- **Commits**: {total_commits}") lines.append(f"- **Files changed**: {files_changed}") lines.append(f"- **Insertions**: {insertions}(+) / {deletions}(-)") if test_commits: lines.append(f"- **Test commits**: {test_commits}") if base_tag: lines.append(f"- **Since**: {base_tag}") lines.append("") changelog_text = "\n".join(lines) # --- Write to file if requested --- wrote_file = "" if output_file: out_path = Path(output_file) if out_path.name == "CHANGELOG.md" and out_path.exists(): # Prepend to existing CHANGELOG.md (keep old content) existing = out_path.read_text() # Insert after the first line if it starts with "# Changelog" if existing.startswith("# Changelog"): header_end = existing.index("\n") + 1 new_content = existing[:header_end] + "\n" + changelog_text + existing[header_end:] else: new_content = "# Changelog\n\n" + changelog_text + "\n" + existing out_path.write_text(new_content) else: out_path.parent.mkdir(parents=True, exist_ok=True) out_path.write_text("# Changelog\n\n" + changelog_text) wrote_file = str(out_path) return { "format": fmt, "version": version_label, "date": date_str, "stats": stats, "total_commits": total_commits, "ledger_items_count": len(ledger_items), "changelog": changelog_text, "wrote_file": wrote_file, } def run_policy(spec_files: List[str], policy_file: Optional[str] = None) -> Dict[str, Any]: """Evaluate specs against governance policy without diffing.""" from core.policy_engine import PolicyEngine engine = PolicyEngine(policy_file) return { "rules_loaded": len(engine.rules), "custom_rules": len(engine.custom_rules), "policy_file": policy_file, "template": engine.create_policy_template() if not policy_file else None, } def simulate_policy( old_spec: str, new_spec: str, policy_file: Optional[str] = None, ) -> Dict[str, Any]: """Run lint + policy across all presets in dry-run mode. Returns what would pass/fail under strict, default, and relaxed presets, plus the custom policy if provided. Nothing is enforced or recorded. """ from core.policy_engine import PolicyEngine, POLICY_PRESETS, evaluate_with_policy from core.diff_engine_v2 import OpenAPIDiffEngine old = _load_specs(old_spec) new = _load_specs(new_spec) # Run diff once (shared across all preset evaluations) diff_engine = OpenAPIDiffEngine() changes = diff_engine.compare(old, new) change_dicts = [ { "type": c.type.value, "path": c.path, "message": c.message, "is_breaking": c.is_breaking, } for c in changes ] # Evaluate each preset preset_results: Dict[str, Any] = {} for preset in POLICY_PRESETS: engine = PolicyEngine(preset) violations = engine.evaluate(changes) has_errors = any(v.severity == "error" for v in violations) has_warnings = any(v.severity == "warning" for v in violations) if has_errors: decision = "fail" elif has_warnings: decision = "warn" else: decision = "pass" preset_results[preset] = { "decision": decision, "rules_loaded": len(engine.rules), "violations": [ { "rule": v.rule_id, "name": v.rule_name, "severity": v.severity, "message": v.message, "path": v.change.path, } for v in violations ], "errors": len([v for v in violations if v.severity == "error"]), "warnings": len([v for v in violations if v.severity == "warning"]), } # Evaluate custom policy if provided (in addition to presets) custom_result = None if policy_file: custom_engine = PolicyEngine(policy_file) custom_violations = custom_engine.evaluate(changes) has_errors = any(v.severity == "error" for v in custom_violations) has_warnings = any(v.severity == "warning" for v in custom_violations) if has_errors: custom_decision = "fail" elif has_warnings: custom_decision = "warn" else: custom_decision = "pass" custom_result = { "decision": custom_decision, "policy_file": policy_file, "rules_loaded": len(custom_engine.rules), "violations": [ { "rule": v.rule_id, "name": v.rule_name, "severity": v.severity, "message": v.message, "path": v.change.path, } for v in custom_violations ], "errors": len([v for v in custom_violations if v.severity == "error"]), "warnings": len([v for v in custom_violations if v.severity == "warning"]), } # Build comparison matrix comparison = {} for preset in POLICY_PRESETS: comparison[preset] = preset_results[preset]["decision"] if custom_result: comparison["custom"] = custom_result["decision"] return { "simulated": True, "dry_run": True, "summary": { "total_changes": len(changes), "breaking_changes": len([c for c in changes if c.is_breaking]), }, "all_changes": change_dicts, "presets": preset_results, "custom_policy": custom_result, "comparison": comparison, } def query_ledger( ledger_path: str, api_name: Optional[str] = None, repository: Optional[str] = None, validate_chain: bool = False, ) -> Dict[str, Any]: """Query the contract ledger.""" from core.contract_ledger import ContractLedger ledger = ContractLedger(ledger_path) if not ledger.exists(): return {"error": "Ledger not found", "path": ledger_path} result: Dict[str, Any] = {"path": ledger_path, "event_count": ledger.get_event_count()} if result["event_count"] == 0: fallback = _query_project_ledger_fallback(Path(ledger_path)) if fallback: if api_name: fallback["events"] = [e for e in _read_jsonl(Path(ledger_path).parent / "operations.jsonl") + _read_jsonl(Path(ledger_path).parent / "strategy.jsonl") if e.get("api_name") == api_name] elif repository: fallback["events"] = [e for e in _read_jsonl(Path(ledger_path).parent / "operations.jsonl") + _read_jsonl(Path(ledger_path).parent / "strategy.jsonl") if e.get("repository") == repository] return fallback if validate_chain: try: ledger.validate_chain() result["chain_valid"] = True except Exception as e: result["chain_valid"] = False result["chain_error"] = str(e) if api_name: result["events"] = ledger.get_api_timeline(api_name) elif repository: result["events"] = ledger.get_events_by_repository(repository) else: latest = ledger.get_latest_event() result["latest_event"] = latest return result def run_impact(api_name: str, dependency_file: Optional[str] = None) -> Dict[str, Any]: """Analyze downstream impact of an API change.""" from core.dependency_graph import DependencyGraph from core.impact_analyzer import ImpactAnalyzer graph = DependencyGraph() if dependency_file: graph.load_from_file(dependency_file) analyzer = ImpactAnalyzer(graph) return analyzer.analyze(api_name) def run_semver( old_spec: str, new_spec: str, current_version: Optional[str] = None, ) -> Dict[str, Any]: """Classify the semver bump for a spec change. Returns detailed breakdown: bump level, per-category counts, and optionally the bumped version string. Auto-detects OpenAPI vs JSON Schema (LED-713). """ old = _load_specs(old_spec) new = _load_specs(new_spec) # LED-713: JSON Schema path if _spec_type(new) == "json_schema" or _spec_type(old) == "json_schema": from core.json_schema_diff import JSONSchemaDiffEngine engine = JSONSchemaDiffEngine() changes = engine.compare(old, new) result = _json_schema_semver(changes) if current_version: result["current_version"] = current_version result["next_version"] = _bump_semver_version(current_version, result["bump"]) return result from core.diff_engine_v2 import OpenAPIDiffEngine from core.semver_classifier import classify_detailed, bump_version, classify engine = OpenAPIDiffEngine() changes = engine.compare(old, new) result = classify_detailed(changes) if current_version: bump = classify(changes) result["current_version"] = current_version result["next_version"] = bump_version(current_version, bump) return result def run_explain( old_spec: str, new_spec: str, template: str = "developer", old_version: Optional[str] = None, new_version: Optional[str] = None, api_name: Optional[str] = None, ) -> Dict[str, Any]: """Generate a human-readable explanation of API changes. Supports 7 templates: developer, team_lead, product, migration, changelog, pr_comment, slack. """ from core.diff_engine_v2 import OpenAPIDiffEngine from core.explainer import explain, TEMPLATES old = _load_specs(old_spec) new = _load_specs(new_spec) engine = OpenAPIDiffEngine() changes = engine.compare(old, new) output = explain( changes, template=template, old_version=old_version, new_version=new_version, api_name=api_name, ) return { "template": template, "available_templates": TEMPLATES, "output": output, } def run_zero_spec( project_dir: str = ".", python_bin: Optional[str] = None, ) -> Dict[str, Any]: """Detect framework and extract OpenAPI spec from source code. Currently supports FastAPI. Returns the extracted spec or an error with guidance on how to fix it. """ from core.zero_spec.detector import detect_framework, Framework from core.zero_spec.express_extractor import extract_express_spec from core.zero_spec.fastapi_extractor import extract_fastapi_spec from core.zero_spec.nestjs_extractor import extract_nestjs_spec info = detect_framework(project_dir) result: Dict[str, Any] = { "framework": info.framework.value, "confidence": info.confidence, "message": info.message, } if info.framework == Framework.FASTAPI: extraction = extract_fastapi_spec( info, project_dir, python_bin=python_bin ) result.update(extraction) if extraction["success"] and info.app_locations: loc = info.app_locations[0] result["app_file"] = loc.file result["app_variable"] = loc.variable result["app_line"] = loc.line elif info.framework == Framework.NESTJS: extraction = extract_nestjs_spec(info, project_dir) result.update(extraction) if extraction["success"] and info.app_locations: loc = info.app_locations[0] result["app_file"] = loc.file result["app_variable"] = loc.variable result["app_line"] = loc.line elif info.framework == Framework.EXPRESS: extraction = extract_express_spec(info, project_dir) result.update(extraction) if extraction["success"] and info.app_locations: loc = info.app_locations[0] result["app_file"] = loc.file result["app_variable"] = loc.variable result["app_line"] = loc.line else: result["success"] = False result["error"] = "No supported API framework found. Provide an OpenAPI spec file." result["error_type"] = "no_framework" return result def run_diff_report( old_spec: str, new_spec: str, fmt: str = "html", output_file: Optional[str] = None, policy_file: Optional[str] = None, ) -> Dict[str, Any]: """Generate a rich comparison report for two API spec versions. Runs the full analysis pipeline (diff, policy, semver, health) and produces a self-contained HTML report or structured JSON suitable for sharing across teams. Args: old_spec: Path to the baseline OpenAPI spec. new_spec: Path to the proposed OpenAPI spec. fmt: Output format -- "html" or "json". output_file: Optional path to write the report file. policy_file: Optional custom policy YAML. Returns: Dict with report content, metadata, and analysis results. """ from datetime import datetime, timezone from core.policy_engine import PolicyEngine from core.semver_classifier import classify_detailed, classify from core.spec_health import score_spec from core.explainer import explain old = _load_specs(old_spec) new = _load_specs(new_spec) # LED-713: JSON Schema dispatch — short-circuit to a minimal report # shape compatible with the JSON renderer (HTML renderer remains # OpenAPI-only; JSON Schema callers should use fmt="json"). if _spec_type(new) == "json_schema" or _spec_type(old) == "json_schema": from core.json_schema_diff import JSONSchemaDiffEngine js_engine = JSONSchemaDiffEngine() js_changes = js_engine.compare(old, new) js_breaking = [c for c in js_changes if c.is_breaking] js_semver = _json_schema_semver(js_changes) now_js = datetime.now(timezone.utc) return { "format": fmt, "spec_type": "json_schema", "generated_at": now_js.isoformat(), "old_spec": old_spec, "new_spec": new_spec, "old_title": old.get("title", "") if isinstance(old, dict) else "", "new_title": new.get("title", "") if isinstance(new, dict) else "", "semver": js_semver, "changes": _json_schema_changes_to_dicts(js_changes), "breaking_count": len(js_breaking), "non_breaking_count": len(js_changes) - len(js_breaking), "total_changes": len(js_changes), "policy": { "decision": "pass", "violations": [], "errors": 0, "warnings": 0, }, "health": None, "migration": "", "output_file": output_file, "note": "JSON Schema report (policy rules and HTML report are OpenAPI-only in v1)", } from core.diff_engine_v2 import OpenAPIDiffEngine # -- Diff -- engine = OpenAPIDiffEngine() changes = engine.compare(old, new) breaking = [c for c in changes if c.is_breaking] non_breaking = [c for c in changes if not c.is_breaking] change_dicts = [ { "type": c.type.value, "path": c.path, "message": c.message, "is_breaking": c.is_breaking, "details": c.details, } for c in changes ] # -- Semver -- semver = classify_detailed(changes) bump = classify(changes) # -- Policy -- policy_engine = PolicyEngine(policy_file) violations = policy_engine.evaluate(changes) has_errors = any(v.severity == "error" for v in violations) has_warnings = any(v.severity == "warning" for v in violations) if has_errors: gate_decision = "fail" elif has_warnings: gate_decision = "warn" else: gate_decision = "pass" violation_dicts = [ { "rule": v.rule_id, "name": v.rule_name, "severity": v.severity, "message": v.message, "path": v.change.path, } for v in violations ] # -- Spec health -- old_health = score_spec(old) new_health = score_spec(new) # -- Migration guide (only if breaking changes exist) -- migration_text = "" if breaking: try: old_ver = old.get("info", {}).get("version") new_ver = new.get("info", {}).get("version") migration_text = explain( changes, template="migration", old_version=old_ver, new_version=new_ver, ) except Exception: migration_text = "" now = datetime.now(timezone.utc) report_data = { "generated_at": now.isoformat(), "old_spec": old_spec, "new_spec": new_spec, "old_version": old.get("info", {}).get("version", "unknown"), "new_version": new.get("info", {}).get("version", "unknown"), "old_title": old.get("info", {}).get("title", ""), "new_title": new.get("info", {}).get("title", ""), "semver": { "bump": semver["bump"], "is_breaking": semver["is_breaking"], "counts": semver["counts"], }, "changes": change_dicts, "breaking_count": len(breaking), "non_breaking_count": len(non_breaking), "total_changes": len(changes), "policy": { "decision": gate_decision, "violations": violation_dicts, "errors": len([v for v in violations if v.severity == "error"]), "warnings": len([v for v in violations if v.severity == "warning"]), }, "health": { "old": old_health, "new": new_health, }, "migration_guide": migration_text, } if fmt == "json": wrote_file = "" if output_file: p = Path(output_file) p.parent.mkdir(parents=True, exist_ok=True) p.write_text(json.dumps(report_data, indent=2, default=str), encoding="utf-8") wrote_file = str(p) return { "format": "json", "wrote_file": wrote_file, "report": report_data, } # -- HTML generation -- html = _render_diff_report_html(report_data) wrote_file = "" if output_file: p = Path(output_file) p.parent.mkdir(parents=True, exist_ok=True) p.write_text(html, encoding="utf-8") wrote_file = str(p) return { "format": "html", "wrote_file": wrote_file, "html": html, "summary": { "total_changes": report_data["total_changes"], "breaking_count": report_data["breaking_count"], "non_breaking_count": report_data["non_breaking_count"], "semver_bump": report_data["semver"]["bump"], "policy_decision": report_data["policy"]["decision"], "old_health_grade": old_health.get("grade", "?"), "new_health_grade": new_health.get("grade", "?"), }, } def _render_diff_report_html(data: Dict[str, Any]) -> str: """Render the diff report data as a self-contained HTML document.""" import html as html_mod esc = html_mod.escape semver = data["semver"] policy = data["policy"] changes = data["changes"] old_health = data["health"]["old"] new_health = data["health"]["new"] # Bump color bump_colors = { "major": "#dc2626", "minor": "#2563eb", "patch": "#16a34a", "none": "#6b7280", } bump_val = semver["bump"].lower() if isinstance(semver["bump"], str) else "none" bump_color = bump_colors.get(bump_val, "#6b7280") # Gate color gate_colors = {"fail": "#dc2626", "warn": "#d97706", "pass": "#16a34a"} gate_color = gate_colors.get(policy["decision"], "#6b7280") # Build change rows change_rows = [] for c in changes: severity_class = "breaking" if c["is_breaking"] else "non-breaking" severity_label = "Breaking" if c["is_breaking"] else "Compatible" severity_dot = "#dc2626" if c["is_breaking"] else "#16a34a" change_rows.append( f'' f' {esc(severity_label)}' f"{esc(c['type'])}" f"{esc(c['path'])}" f"{esc(c['message'])}" f"" ) change_rows_html = "\n".join(change_rows) if change_rows else 'No changes detected' # Build violation rows violation_rows = [] for v in policy["violations"]: sev_color = "#dc2626" if v["severity"] == "error" else "#d97706" violation_rows.append( f"" f' {esc(v["severity"].upper())}' f"{esc(v['rule'])}" f"{esc(v['message'])}" f"{esc(v['path'])}" f"" ) violation_rows_html = "\n".join(violation_rows) if violation_rows else 'No policy violations' # Health dimensions def _health_dimensions_html(health: Dict[str, Any], label: str) -> str: dims = health.get("dimensions", {}) if not dims: return f"

{label}: No dimensions available

" rows = [] for dim_name, dim_data in dims.items(): score = dim_data.get("score", 0) bar_color = "#16a34a" if score >= 70 else "#d97706" if score >= 40 else "#dc2626" rows.append( f'
' f'{esc(dim_name.replace("_", " ").title())}' f'
' f'{score}' f"
" ) return "\n".join(rows) old_dims_html = _health_dimensions_html(old_health, "Old") new_dims_html = _health_dimensions_html(new_health, "New") # Migration guide migration_html = "" if data.get("migration_guide"): lines = data["migration_guide"].split("\n") migration_parts = [] in_pre = False for line in lines: stripped = line.strip() if stripped.startswith("```"): if in_pre: migration_parts.append("") in_pre = False else: migration_parts.append("
")
                    in_pre = True
            elif in_pre:
                migration_parts.append(esc(line))
            elif stripped.startswith("# "):
                migration_parts.append(f"
{esc(stripped[2:])}
")
            elif stripped.startswith("## "):
                migration_parts.append(f"
{esc(stripped[3:])}
")
            elif stripped.startswith("### "):
                migration_parts.append(f"
{esc(stripped[4:])}
")
            elif stripped.startswith("- "):
                migration_parts.append(f"
  • {esc(stripped[2:])}
    • ")
            elif stripped:
                migration_parts.append(f"
    {esc(stripped)}
    ")
        if in_pre:
            migration_parts.append("
    ") migration_html = f"""

    Migration Guide

    {"".join(migration_parts)}
    """ html = f""" API Diff Report -- {esc(data['old_version'])} to {esc(data['new_version'])}

    {esc(data.get('new_title') or data.get('old_title') or 'API')} -- Diff Report

    {esc(data['old_version'])} → {esc(data['new_version'])} · Generated {esc(data['generated_at'][:19])} UTC
    {esc(bump_val.upper())} Gate: {esc(policy['decision'].upper())}
    Total Changes
    {data['total_changes']}
    Breaking
    {data['breaking_count']}
    Non-Breaking
    {data['non_breaking_count']}
    Semver Bump
    {esc(bump_val.upper())}
    {semver['counts'].get('breaking', 0)} breaking, {semver['counts'].get('additive', 0)} additive, {semver['counts'].get('patch', 0)} patch
    Policy Gate
    {esc(policy['decision'].upper())}
    {policy['errors']} errors, {policy['warnings']} warnings

    Changes

    {change_rows_html}
    Severity Type Path Description

    Policy Violations

    {violation_rows_html}
    Severity Rule Message Path
    {migration_html}

    Spec Health

    Baseline -- {esc(data['old_version'])} (Grade: {esc(str(old_health.get('grade', '?')))} Score: {old_health.get('overall_score', '?')})

    {old_dims_html}

    Proposed -- {esc(data['new_version'])} (Grade: {esc(str(new_health.get('grade', '?')))} Score: {new_health.get('overall_score', '?')})

    {new_dims_html}
    Generated by Delimit -- API governance for teams that ship. · delimit.ai
    """ return html