"""muse code contract — infer the implicit behavioral contract of a symbol. Every function has two contracts: The **explicit** contract — type annotations, docstrings, parameter names. The **implicit** contract — what callers actually depend on, what tests actually assert, and how the contract has evolved under real pressure. ``muse code contract`` infers the implicit contract by mining four sources that only Muse has simultaneous access to: 1. **Current signature** — parameters, type annotations, defaults, and return type extracted from the AST. 2. **Call-site patterns** — how every direct caller in the production codebase actually invokes the symbol: which keyword args are used, whether the return value is stored, discarded, asserted, or returned, and what argument types appear in practice. 3. **Test assertions** — ``assert`` statements extracted from every test function that exercises this symbol, forming the ground-truth specification of expected behavior. 4. **Commit history** — stability fingerprint (MAJOR/MINOR/PATCH bumps, signature changes, body rewrites) plus commit messages that contain contract signals like ``breaking:``, ``invariant:``, ``deprecated:``. Combined, these four sources answer questions no single tool can answer alone: * "Is this parameter actually required, or do callers always rely on the default?" * "Do callers check the return value, or do they silently discard it?" * "What do tests actually guarantee about the return value?" * "Has this function been stable for 200 commits, or is it changing every week?" * "Does any commit message hint at an implicit invariant?" Usage:: muse code contract "billing.py::Invoice.compute_total" muse code contract "muse/core/store.py::resolve_commit_ref" muse code contract "billing.py::Invoice.compute_total" --max-commits 200 muse code contract "billing.py::Invoice.compute_total" --json Output:: Contract: billing.py::Invoice.compute_total Signature def compute_total(items, currency='USD') -> float Parameters items positional always provided · list/comprehension at 11/12 sites currency keyword omitted at 8/12 sites (default relied on) observed values: 'USD' (8×) · 'EUR' (4×) Return value → stored at 9/12 call sites (75%) → discarded at 2/12 call sites ⚠️ possible misuse → returned at 1/12 call sites Inferred preconditions • items is always a list or list-comprehension • currency is effectively optional — default 'USD' used by 67% of callers Inferred postconditions • Return value is expected to be meaningful (stored 75% of the time) • 2 callers silently discard the return — possible side-effect assumption Test assertions (3 test functions) assert result == expected_total assert result > 0 assert isinstance(result, (int, float)) Commit signals MINOR "feat: add optional currency parameter" (Feb 03) PATCH "perf: vectorise invoice computation" (Feb 14) Stability 97 commits · 2 signature changes · 3 body rewrites 0 MAJOR · 5 MINOR · 11 PATCH Est. 25% of original implementation remains Assessment: evolving — contract is extending actively ⚠️ Warnings • No return type annotation — return type inferred by callers • 2 call sites discard the return value — check for misuse JSON output (``--json``):: { "address": "billing.py::Invoice.compute_total", "name": "compute_total", "kind": "method", "signature": "def compute_total(items, currency='USD') -> float", "parameters": [ { "name": "items", "annotation": null, "has_default": false, "default_str": null }, { "name": "currency", "annotation": null, "has_default": true, "default_str": "'USD'" } ], "return_annotation": "float", "call_sites": 12, "caller_files": 4, "return_dispositions": { "stored": 9, "discarded": 2, "returned": 1, "asserted": 0, "compared": 0, "argument": 0, "unknown": 0 }, "arg_observations": [ { "param_index": 0, "categories": {"list": 8, "name": 3, "call": 1}, "none_count": 0, "always_provided": true }, { "param_index": 1, "categories": {"str": 4}, "none_count": 0, "always_provided": false, "omit_count": 8 } ], "test_assertions": [ "assert result == expected_total", "assert result > 0", "assert isinstance(result, (int, float))" ], "commit_signals": [ { "bump": "minor", "message": "feat: add optional currency parameter", "date": "2026-02-03" } ], "history": { "commits_analysed": 97, "truncated": false, "major_bumps": 0, "minor_bumps": 5, "patch_bumps": 11, "sig_changes": 2, "impl_changes": 3, "est_survival_pct": 25 }, "preconditions": [ "items is always a list or list-comprehension" ], "postconditions": [ "return value is stored at 75%% of call sites" ], "warnings": [ "no return type annotation", "2 call sites discard the return value" ], "stability": "evolving" } Security note ------------- Symbol addresses are validated before any store access. Commit messages are sanitised (control characters stripped) before display. All file access goes through the content-addressed object store — no user-supplied paths are opened directly. """ from __future__ import annotations import argparse import ast import datetime import json import logging import pathlib import re import sys from collections import Counter from dataclasses import dataclass from typing import Literal, TypedDict from muse.core.errors import ExitCode from muse.core.repo import read_repo_id, require_repo from muse.core.store import ( get_commit_snapshot_manifest, read_current_branch, resolve_commit_ref, ) from muse.domain import DomainOp from muse.plugins.code._callgraph import ( ReverseGraph, call_name, find_func_node, transitive_callers, ) from muse.plugins.code._query import flat_symbol_ops, symbols_for_snapshot, walk_commits_bfs from muse.plugins.code.ast_parser import SymbolRecord, parse_symbols from muse.core.validation import clamp_int, MAX_AST_BYTES, sanitize_display type _BlobMap = dict[str, bytes] type _SymbolIndex = dict[str, SymbolRecord] type _CounterMap = dict[str, int] logger = logging.getLogger(__name__) # ── Constants ────────────────────────────────────────────────────────────────── _DEFAULT_MAX_COMMITS = 2_000 _DEFAULT_MAX_CALL_SITES = 200 # cap to keep analysis bounded _PY_SUFFIXES: frozenset[str] = frozenset({".py", ".pyi"}) _TEST_PATTERNS: tuple[re.Pattern[str], ...] = ( re.compile(r"(^|/)test_"), re.compile(r"_test\.py$"), re.compile(r"(^|/)tests/"), re.compile(r"(^|/)spec/"), re.compile(r"(^|/)conftest\.py$"), ) # Keywords in commit messages that signal contract-relevant events. _CONTRACT_SIGNAL_WORDS: frozenset[str] = frozenset( { "breaking", "break", "invariant", "contract", "guarantee", "guarantee:", "deprecated", "deprecate", "must", "require", "require:", "assert", "precondition", "postcondition", "never", "always", "invariants", } ) _CTRL_RE = re.compile(r"[\x00-\x09\x0b-\x1f\x7f-\x9f\x1b]") # Op summary keywords for change classification (same as age / narrative). _IMPL_KW: frozenset[str] = frozenset({"implementation", "modified", "body", "reformatted"}) _SIG_KW: frozenset[str] = frozenset({"signature"}) # ── TypedDicts ───────────────────────────────────────────────────────────────── class _ParamInfo(TypedDict): name: str annotation: str | None has_default: bool default_str: str | None class _ArgObs(TypedDict): """Observed argument patterns at call sites for one positional slot.""" param_index: int categories: _CounterMap # ast category → count none_count: int always_provided: bool omit_count: int class _CommitSignal(TypedDict): bump: str message: str date: str class _HistorySummary(TypedDict): commits_analysed: int truncated: bool major_bumps: int minor_bumps: int patch_bumps: int sig_changes: int impl_changes: int est_survival_pct: int class _ContractJson(TypedDict): address: str name: str kind: str signature: str parameters: list[_ParamInfo] return_annotation: str | None call_sites: int caller_files: int return_dispositions: _CounterMap arg_observations: list[_ArgObs] test_assertions: list[str] commit_signals: list[_CommitSignal] history: _HistorySummary preconditions: list[str] postconditions: list[str] warnings: list[str] stability: str # ── Internal accumulators ────────────────────────────────────────────────────── @dataclass class _CallSiteRecord: """Data collected from a single call to the target symbol.""" caller_addr: str is_test: bool positional_count: int keyword_names: list[str] disposition: str # stored|returned|asserted|discarded|compared|argument|unknown arg_categories: list[str] # one entry per positional arg # ── Helpers ──────────────────────────────────────────────────────────────────── def _is_test_file(fp: str) -> bool: return any(p.search(fp) for p in _TEST_PATTERNS) def _sanitise(s: str, max_len: int = 80) -> str: clean = _CTRL_RE.sub("", s).strip() return clean[:max_len - 1] + "…" if len(clean) > max_len else clean def _arg_category(node: ast.expr) -> str: """Classify an AST argument expression into a coarse type category.""" if isinstance(node, ast.Constant): if node.value is None: return "None" return type(node.value).__name__ # "str", "int", "float", "bool" if isinstance(node, ast.List): return "list" if isinstance(node, ast.Dict): return "dict" if isinstance(node, ast.Tuple): return "tuple" if isinstance(node, ast.Set): return "set" if isinstance(node, (ast.ListComp, ast.GeneratorExp, ast.DictComp, ast.SetComp)): return "comprehension" if isinstance(node, ast.Call): return "call" if isinstance(node, ast.Name): return f"var" return "other" def _extract_direct_call(node: ast.expr | None, target_name: str) -> ast.Call | None: """If *node* is a direct ast.Call to *target_name*, return it; else None.""" if isinstance(node, ast.Call): n = call_name(node.func) if n == target_name: return node return None def _classify_disposition( stmts: list[ast.stmt], target_name: str, ) -> list[tuple[str, list[ast.expr], list[ast.keyword]]]: """Walk *stmts* and classify how each direct call to *target_name* is used. Returns a list of ``(disposition, positional_args, keywords)`` tuples. """ results: list[tuple[str, list[ast.expr], list[ast.keyword]]] = [] for stmt in stmts: if isinstance(stmt, ast.Expr): c = _extract_direct_call(stmt.value, target_name) if c: results.append(("discarded", list(c.args), list(c.keywords))) elif isinstance(stmt, (ast.Assign, ast.AnnAssign)): val = stmt.value if isinstance(stmt, ast.Assign) else stmt.value if val is not None: c = _extract_direct_call(val, target_name) if c: results.append(("stored", list(c.args), list(c.keywords))) elif isinstance(stmt, ast.Return) and stmt.value is not None: c = _extract_direct_call(stmt.value, target_name) if c: results.append(("returned", list(c.args), list(c.keywords))) elif isinstance(stmt, ast.Assert): c = _extract_direct_call(stmt.test, target_name) if c: results.append(("asserted", list(c.args), list(c.keywords))) elif isinstance(stmt, ast.If): c = _extract_direct_call(stmt.test, target_name) if c: results.append(("compared", list(c.args), list(c.keywords))) results.extend(_classify_disposition(stmt.body, target_name)) results.extend(_classify_disposition(stmt.orelse, target_name)) elif isinstance(stmt, (ast.For, ast.While)): results.extend(_classify_disposition(stmt.body, target_name)) elif isinstance(stmt, ast.With): results.extend(_classify_disposition(stmt.body, target_name)) elif isinstance(stmt, ast.Try): results.extend(_classify_disposition(stmt.body, target_name)) for h in stmt.handlers: results.extend(_classify_disposition(h.body, target_name)) results.extend(_classify_disposition(stmt.finalbody, target_name)) return results def _extract_assertions_from_test( func_node: ast.FunctionDef | ast.AsyncFunctionDef, target_bare_name: str, ) -> list[str]: """Return ``assert`` expressions from a test function that calls *target*. Only collects assertions from test functions that contain at least one call to *target_bare_name*. Caps at 8 assertions to keep output focused. """ # Check if this function ever calls target. has_call = any( isinstance(node, ast.Call) and call_name(node.func) == target_bare_name for node in ast.walk(func_node) ) if not has_call: return [] assertions: list[str] = [] for node in ast.walk(func_node): if isinstance(node, ast.Assert): try: assertions.append(ast.unparse(node)) except Exception: pass if len(assertions) >= 8: break return assertions # ── Signature extraction ─────────────────────────────────────────────────────── def _extract_signature( raw: bytes, address: str, ) -> tuple[str, list[_ParamInfo], str | None]: """Extract signature data for *address* from *raw* Python source. Returns: ``(signature_str, params, return_annotation)`` ``signature_str`` is a compact human-readable signature line. ``params`` is a list of parameter info dicts. ``return_annotation`` is the return type as a string, or ``None``. """ if "::" not in address: return ("", [], None) sym_name = address.split("::", 1)[1] try: if len(raw) > MAX_AST_BYTES: return ("", [], None) tree = ast.parse(raw) except SyntaxError: return ("", [], None) func_node = find_func_node(tree.body, sym_name.split(".")) if func_node is None: return ("", [], None) args = func_node.args all_args = list(args.args) # Include *args, **kwargs for completeness if present. num_defaults = len(args.defaults) num_params = len(all_args) # Defaults align to the last N params. default_offset = num_params - num_defaults params: list[_ParamInfo] = [] for i, arg in enumerate(all_args): ann = ast.unparse(arg.annotation) if arg.annotation else None has_default = i >= default_offset default_str: str | None = None if has_default: d = args.defaults[i - default_offset] try: default_str = ast.unparse(d) except Exception: default_str = "..." # Skip "self" and "cls" from the public contract. if arg.arg in ("self", "cls"): continue params.append( _ParamInfo( name=arg.arg, annotation=ann, has_default=has_default, default_str=default_str, ) ) ret_ann: str | None = None if func_node.returns: try: ret_ann = ast.unparse(func_node.returns) except Exception: pass # Build compact signature string. parts: list[str] = [] for i, arg in enumerate(all_args): if arg.arg in ("self", "cls"): continue ann_str = f": {ast.unparse(arg.annotation)}" if arg.annotation else "" has_default = (all_args.index(arg)) >= default_offset def_str = "" if has_default: d = args.defaults[all_args.index(arg) - default_offset] try: def_str = f"={ast.unparse(d)}" except Exception: def_str = "=..." parts.append(f"{arg.arg}{ann_str}{def_str}") bare_name = sym_name.split(".")[-1] ret_str = f" -> {ret_ann}" if ret_ann else "" signature_str = f"def {bare_name}({', '.join(parts)}){ret_str}" return signature_str, params, ret_ann # ── Call-site analysis ───────────────────────────────────────────────────────── def _analyze_call_sites( blobs: _BlobMap, direct_callers: list[str], target_bare_name: str, ) -> list[_CallSiteRecord]: """Analyse how each direct caller invokes the target symbol. Args: blobs: Pre-loaded Python blobs. direct_callers: List of caller symbol addresses (direct callers only). target_bare_name: Bare name of the target symbol. Returns: One ``_CallSiteRecord`` per call to the target found. """ records: list[_CallSiteRecord] = [] for caller_addr in direct_callers[:_DEFAULT_MAX_CALL_SITES]: if "::" not in caller_addr: continue file_path = caller_addr.split("::")[0] sym_name = caller_addr.split("::", 1)[1] raw = blobs.get(file_path) if raw is None: continue try: if len(raw) > MAX_AST_BYTES: continue tree = ast.parse(raw) except SyntaxError: continue func_node = find_func_node(tree.body, sym_name.split(".")) if func_node is None: continue is_test = _is_test_file(file_path) call_items = _classify_disposition(list(func_node.body), target_bare_name) for disposition, pos_args, kw_args in call_items: arg_cats = [_arg_category(a) for a in pos_args] kw_names = [k.arg for k in kw_args if k.arg is not None] records.append( _CallSiteRecord( caller_addr=caller_addr, is_test=is_test, positional_count=len(pos_args), keyword_names=kw_names, disposition=disposition, arg_categories=arg_cats, ) ) return records def _extract_all_test_assertions( blobs: _BlobMap, direct_callers: list[str], target_bare_name: str, ) -> list[str]: """Collect assert statements from test callers of the target. Deduplicates across all test functions and caps at 10 total. """ seen: set[str] = set() assertions: list[str] = [] for caller_addr in direct_callers: if "::" not in caller_addr: continue file_path = caller_addr.split("::")[0] if not _is_test_file(file_path): continue sym_name = caller_addr.split("::", 1)[1] raw = blobs.get(file_path) if raw is None: continue try: if len(raw) > MAX_AST_BYTES: continue tree = ast.parse(raw) except SyntaxError: continue func_node = find_func_node(tree.body, sym_name.split(".")) if func_node is None: continue for assertion in _extract_assertions_from_test(func_node, target_bare_name): if assertion not in seen: seen.add(assertion) assertions.append(assertion) if len(assertions) >= 10: return assertions return assertions # ── History analysis ─────────────────────────────────────────────────────────── def _collect_history( root: pathlib.Path, head_commit_id: str, address: str, max_commits: int, ) -> tuple[_HistorySummary, list[_CommitSignal]]: """Walk the commit DAG and build a history summary for *address*. Returns: ``(history_summary, commit_signals)`` """ commits, truncated = walk_commits_bfs(root, head_commit_id, max_commits) major = minor = patch = sig = impl = 0 signals: list[_CommitSignal] = [] for commit in commits: if commit.structured_delta is None: continue ops: list[DomainOp] = commit.structured_delta["ops"] touched = False for op in flat_symbol_ops(ops): if op["address"] != address: continue touched = True kind = op.get("op", "") if kind == "replace": new_sum = str(op.get("new_summary") or "").lower() old_sum = str(op.get("old_summary") or "").lower() if any(kw in new_sum for kw in _SIG_KW): sig += 1 elif any(kw in new_sum for kw in _IMPL_KW) or any( kw in old_sum for kw in _IMPL_KW ): impl += 1 else: impl += 1 # conservative if not touched: continue bump = (commit.sem_ver_bump if commit.sem_ver_bump else "none").lower() if bump == "major": major += 1 elif bump == "minor": minor += 1 elif bump == "patch": patch += 1 # Extract contract signals from commit message. msg = _sanitise(commit.message, max_len=100) msg_lower = msg.lower() if any(w in msg_lower for w in _CONTRACT_SIGNAL_WORDS): signals.append( _CommitSignal( bump=bump, message=msg, date=commit.committed_at.strftime("%Y-%m-%d"), ) ) est_survival = round(100 / (impl + 1)) return ( _HistorySummary( commits_analysed=len(commits), truncated=truncated, major_bumps=major, minor_bumps=minor, patch_bumps=patch, sig_changes=sig, impl_changes=impl, est_survival_pct=est_survival, ), signals[:20], # cap signals to keep output readable ) # ── Contract inference ───────────────────────────────────────────────────────── def _infer_contract( params: list[_ParamInfo], call_records: list[_CallSiteRecord], history: _HistorySummary, ) -> tuple[list[str], list[str], list[str], str]: """Infer preconditions, postconditions, warnings, and stability assessment. Returns: ``(preconditions, postconditions, warnings, stability)`` """ pre: list[str] = [] post: list[str] = [] warn: list[str] = [] n = len(call_records) # ── Return value patterns ────────────────────────────────────────────────── disp_counts: Counter[str] = Counter(r.disposition for r in call_records) discarded = disp_counts.get("discarded", 0) stored = disp_counts.get("stored", 0) returned = disp_counts.get("returned", 0) asserted = disp_counts.get("asserted", 0) compared = disp_counts.get("compared", 0) if n > 0: use_count = n - discarded if use_count > 0: post.append( f"Return value is used at {use_count}/{n} call sites" f" ({round(use_count / n * 100)}%)" ) if stored > 0: post.append( f"Return value is stored at {stored}/{n} call sites" f" — callers expect a meaningful result" ) if discarded > 0: warn.append( f"{discarded} call site{'s' if discarded > 1 else ''} discard" f" the return value — check for misuse or side-effect assumption" ) # ── Argument patterns ────────────────────────────────────────────────────── # For each positional param slot, gather observed arg categories. for slot, param in enumerate(params): slot_cats: Counter[str] = Counter() omit_count = 0 for r in call_records: if slot < len(r.arg_categories): slot_cats[r.arg_categories[slot]] += 1 elif param["has_default"]: omit_count += 1 if param["has_default"] and n > 0: omit_rate = omit_count / n if omit_rate > 0.5: pre.append( f"``{param['name']}`` defaults relied upon at" f" {omit_count}/{n} call sites — effectively optional" ) elif omit_rate == 0.0: pre.append( f"``{param['name']}`` always provided explicitly — callers" " never rely on the default" ) if slot_cats: top_cat, top_n = slot_cats.most_common(1)[0] if top_n == len(call_records) and len(call_records) > 1: pre.append( f"``{param['name']}`` is always a {top_cat}" f" ({top_n} call sites agree)" ) elif "None" in slot_cats: none_count = slot_cats["None"] pre.append( f"``{param['name']}`` is sometimes None" f" ({none_count}/{n} call sites)" ) # ── Annotation gaps ──────────────────────────────────────────────────────── unannotated = [p["name"] for p in params if not p["annotation"]] if unannotated: warn.append( f"No type annotations on: {', '.join(unannotated)}" " — contract is implicit" ) # ── Stability assessment ─────────────────────────────────────────────────── if history["major_bumps"] > 0: stability = "volatile" elif history["sig_changes"] > 2 or history["minor_bumps"] > 3: stability = "evolving" elif history["impl_changes"] == 0 and history["sig_changes"] == 0: stability = "dormant" elif history["impl_changes"] <= 1 and history["sig_changes"] == 0: stability = "stable" else: stability = "evolving" return pre, post, warn, stability # ── Formatters ───────────────────────────────────────────────────────────────── def _pct_bar(n: int, total: int, width: int = 20) -> str: if total <= 0: return "" filled = round(n / total * width) return "█" * filled + "░" * (width - filled) def _print_contract( address: str, name: str, kind: str, signature_str: str, params: list[_ParamInfo], ret_ann: str | None, call_records: list[_CallSiteRecord], test_assertions: list[str], commit_signals: list[_CommitSignal], history: _HistorySummary, pre: list[str], post: list[str], warn: list[str], stability: str, ) -> None: """Print the human-readable contract report.""" caller_files = len({r.caller_addr.split("::")[0] for r in call_records}) n = len(call_records) print(f"\n Contract: {sanitize_display(address)}\n") # ── Signature ────────────────────────────────────────────────────────────── print(" Signature") print(f" {signature_str}\n") # ── Parameters ──────────────────────────────────────────────────────────── if params: print(" Parameters") for slot, param in enumerate(params): ann = f": {param['annotation']}" if param["annotation"] else "" default = f" = {param['default_str']}" if param["has_default"] else "" print(f" {sanitize_display(param['name'])}{ann}{default}") # Arg usage patterns. slot_cats: Counter[str] = Counter() omit_count = 0 for r in call_records: if slot < len(r.arg_categories): slot_cats[r.arg_categories[slot]] += 1 elif param["has_default"]: omit_count += 1 lines: list[str] = [] if omit_count > 0 and n > 0: lines.append(f"omitted at {omit_count}/{n} sites (default relied on)") if slot_cats: cats = ", ".join( f"{cat} ({cnt}×)" for cat, cnt in slot_cats.most_common(3) ) lines.append(f"observed: {cats}") for line in lines: print(f" {line}") print() # ── Return value ────────────────────────────────────────────────────────── if n > 0: print(" Return value") disp: Counter[str] = Counter(r.disposition for r in call_records) for label, count in disp.most_common(): if count == 0: continue bar = _pct_bar(count, n, width=16) warn_flag = " ⚠️" if label == "discarded" and count > 1 else "" print(f" → {label:<12} {count:>3}/{n} [{bar}]{warn_flag}") print() # ── Preconditions ───────────────────────────────────────────────────────── if pre: print(" Inferred preconditions") for p in pre: print(f" • {p}") print() # ── Postconditions ──────────────────────────────────────────────────────── if post: print(" Inferred postconditions") for p in post: print(f" • {p}") print() # ── Test assertions ─────────────────────────────────────────────────────── test_sites = sum(1 for r in call_records if r.is_test) if test_assertions: print(f" Test assertions ({test_sites} test call site{'s' if test_sites != 1 else ''})") for a in test_assertions: print(f" {a}") print() # ── Commit signals ──────────────────────────────────────────────────────── if commit_signals: print(" Commit signals") for sig in commit_signals[:5]: print(f" {sig['bump'].upper():<8} \"{sanitize_display(sig['message'][:60])}\" ({sig['date']})") print() # ── Stability ───────────────────────────────────────────────────────────── print(" Stability") h = history print( f" {h['commits_analysed']} commits" f" · {h['sig_changes']} signature change{'s' if h['sig_changes'] != 1 else ''}" f" · {h['impl_changes']} body rewrite{'s' if h['impl_changes'] != 1 else ''}" ) print( f" {h['major_bumps']} MAJOR" f" · {h['minor_bumps']} MINOR" f" · {h['patch_bumps']} PATCH" ) print(f" Est. {h['est_survival_pct']}% of original implementation remains") stability_desc = { "stable": "stable — contract has been consistent", "evolving": "evolving — contract is extending actively", "volatile": "volatile — MAJOR contract breaks in history", "dormant": "dormant — rarely changed", }.get(stability, stability) print(f" Assessment: {stability_desc}\n") # ── Warnings ────────────────────────────────────────────────────────────── if warn: print(" ⚠️ Warnings") for w in warn: print(f" • {w}") print() # ── Entry point ──────────────────────────────────────────────────────────────── def run(args: argparse.Namespace) -> None: """Entry point for ``muse code contract``.""" root = require_repo() # ── Argument validation ──────────────────────────────────────────────────── address: str = args.address.strip() if "::" not in address: print( "❌ ADDRESS must be in ``file::Symbol`` format.", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) max_commits: int = clamp_int(args.max_commits, 1, 100000, 'max_commits') if max_commits < 1: print("❌ --max-commits must be >= 1.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) # ── Resolve HEAD ─────────────────────────────────────────────────────────── repo_id = read_repo_id(root) branch = read_current_branch(root) head = resolve_commit_ref(root, repo_id, branch, None) if head is None: print("❌ HEAD commit not found — is this an empty repository?", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) manifest = get_commit_snapshot_manifest(root, head.commit_id) or {} # ── Load all Python blobs in one pass ───────────────────────────────────── from muse.core.object_store import read_object blobs: _BlobMap = {} for fp, obj_id in manifest.items(): if pathlib.PurePosixPath(fp).suffix.lower() in _PY_SUFFIXES: raw = read_object(root, obj_id) if raw is not None: blobs[fp] = raw # ── Verify symbol exists at HEAD ─────────────────────────────────────────── all_trees = symbols_for_snapshot(root, manifest) addr_to_rec: _SymbolIndex = {} for sym_tree in all_trees.values(): addr_to_rec.update(sym_tree) if address not in addr_to_rec: print(f"❌ {sanitize_display(address)!r} not found in HEAD snapshot.", file=sys.stderr) print(" Use ``muse code symbols`` to list available addresses.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) rec = addr_to_rec[address] name = rec["name"] kind = rec["kind"] file_path = address.split("::")[0] # ── Extract current signature ────────────────────────────────────────────── raw_file = blobs.get(file_path) params: list[_ParamInfo] if raw_file is None: signature_str, params, ret_ann = "", [], None else: signature_str, params, ret_ann = _extract_signature(raw_file, address) # ── Build forward + reverse call graph from production blobs ────────────── from muse.plugins.code._callgraph import ForwardGraph forward: ForwardGraph = {} for fp, raw in blobs.items(): if _is_test_file(fp): continue try: if len(raw) > MAX_AST_BYTES: continue tree_ast = ast.parse(raw) except SyntaxError: continue sym_tree = parse_symbols(raw, fp) for addr2, rec2 in sym_tree.items(): if rec2["kind"] not in {"function", "async_function", "method", "async_method"}: continue fn = find_func_node(tree_ast.body, rec2["qualified_name"].split(".")) if fn is None: continue callees: set[str] = set() for node in ast.walk(fn): if isinstance(node, ast.Call): n2 = call_name(node.func) if n2: callees.add(n2) forward[addr2] = frozenset(callees) # Invert → reverse graph. reverse: ReverseGraph = {} for caller_addr, callee_names in forward.items(): for cn in callee_names: reverse.setdefault(cn, []).append(caller_addr) for cn in reverse: reverse[cn].sort() # Find direct callers only (depth=1). depth_map = transitive_callers(name, reverse, max_depth=1) direct_callers: list[str] = list(depth_map.get(1, [])) # Also include test-file callers for test assertion extraction. # Build a separate test reverse graph including test files. test_forward: ForwardGraph = {} for fp, raw in blobs.items(): if not _is_test_file(fp): continue try: if len(raw) > MAX_AST_BYTES: continue tree_ast = ast.parse(raw) except SyntaxError: continue sym_tree = parse_symbols(raw, fp) for addr2, rec2 in sym_tree.items(): if rec2["kind"] not in {"function", "async_function", "method", "async_method"}: continue fn = find_func_node(tree_ast.body, rec2["qualified_name"].split(".")) if fn is None: continue callees = set() for node in ast.walk(fn): if isinstance(node, ast.Call): n2 = call_name(node.func) if n2: callees.add(n2) test_forward[addr2] = frozenset(callees) test_reverse: ReverseGraph = {} for addr2, callee_names in test_forward.items(): for cn in callee_names: test_reverse.setdefault(cn, []).append(addr2) test_callers = list(transitive_callers(name, test_reverse, max_depth=1).get(1, [])) all_direct = direct_callers + test_callers # ── Analyse call sites ───────────────────────────────────────────────────── call_records = _analyze_call_sites(blobs, all_direct, name) # ── Extract test assertions ──────────────────────────────────────────────── test_assertions = _extract_all_test_assertions(blobs, test_callers, name) # ── Commit history ──────────────────────────────────────────────────────── history, commit_signals = _collect_history(root, head.commit_id, address, max_commits) # ── Infer contract ──────────────────────────────────────────────────────── pre, post, warn, stability = _infer_contract(params, call_records, history) # ── Output ──────────────────────────────────────────────────────────────── if args.json: disp_counts: _CounterMap = { "stored": 0, "returned": 0, "asserted": 0, "discarded": 0, "compared": 0, "argument": 0, "unknown": 0, } for r in call_records: disp_counts[r.disposition] = disp_counts.get(r.disposition, 0) + 1 # Arg observations per param slot. arg_obs: list[_ArgObs] = [] for slot, param in enumerate(params): cats: Counter[str] = Counter() omit = 0 for r in call_records: if slot < len(r.arg_categories): cats[r.arg_categories[slot]] += 1 elif param["has_default"]: omit += 1 arg_obs.append( _ArgObs( param_index=slot, categories=dict(cats), none_count=cats.get("None", 0), always_provided=omit == 0, omit_count=omit, ) ) caller_files = len({r.caller_addr.split("::")[0] for r in call_records}) out: _ContractJson = _ContractJson( address=address, name=name, kind=kind, signature=signature_str, parameters=params, return_annotation=ret_ann, call_sites=len(call_records), caller_files=caller_files, return_dispositions=disp_counts, arg_observations=arg_obs, test_assertions=test_assertions, commit_signals=commit_signals, history=history, preconditions=pre, postconditions=post, warnings=warn, stability=stability, ) print(json.dumps(out, indent=2)) return _print_contract( address, name, kind, signature_str, params, ret_ann, call_records, test_assertions, commit_signals, history, pre, post, warn, stability, ) # ── CLI registration ─────────────────────────────────────────────────────────── def register( sub: argparse._SubParsersAction[argparse.ArgumentParser], ) -> None: """Register ``contract`` under the ``code`` subcommand group.""" p = sub.add_parser( "contract", help=( "Infer the implicit behavioral contract of a symbol from its" " call sites, test assertions, and commit history." ), description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter, ) p.add_argument( "address", metavar="ADDRESS", help=( "Full symbol address" " (e.g. billing.py::Invoice.compute_total)." ), ) p.add_argument( "--max-commits", type=int, default=_DEFAULT_MAX_COMMITS, metavar="N", help=f"Maximum commits to walk for history analysis (default: {_DEFAULT_MAX_COMMITS}).", ) p.add_argument( "--json", action="store_true", help="Emit JSON instead of human-readable text.", ) p.set_defaults(func=run)