| """v288: read-only, bounded, source-verified GitHub file intelligence. |
| |
| Deterministic parsing and graph edges complement the existing source specialist. |
| Neither repository text nor model output can authorize a tool call or mutation. |
| """ |
| from __future__ import annotations |
|
|
| import ast |
| import hashlib |
| import json |
| import posixpath |
| import re |
| from pathlib import PurePosixPath |
|
|
| SHA = re.compile(r"^[a-f0-9]{40}$") |
| SECRET = re.compile(r"(?i)(^|/)(?:secrets?|credentials?|tokens?|keys?|certificates?|id_rsa|id_ed25519)(/|$)") |
| EXTS = {".py", ".js", ".mjs", ".jsx", ".ts", ".tsx", ".kt", ".java", |
| ".go", ".rs", ".md", ".txt", ".json", ".yml", ".yaml", ".toml"} |
| MAX_FILES = 4 |
| MAX_TREE = 20000 |
| MAX_RELATED_BYTES = 32000 |
| TOTAL_RELATED_BYTES = 96000 |
|
|
| SCENARIOS = ( |
| ("security", r"\b(?:security|vulnerab|injection|permissions?|auth|secret|unsafe|exploit)\b", |
| ("inspect-boundaries", "trace-imports", "identify-unverified-risks")), |
| ("regression", r"\b(?:regression|broke|stopped working|since the update|before.?after)\b", |
| ("trace-imports", "identify-related-code", "request-commit-diff")), |
| ("impact", r"\b(?:impact|break if|dependenc|references?|rename|change this|affected)\b", |
| ("extract-symbols", "trace-imports", "map-potential-impact")), |
| ("debug", r"\b(?:bug|error|fail|crash|debug|diagnos|not working|exception)\b", |
| ("extract-symbols", "trace-imports", "identify-test-targets")), |
| ("architecture", r"\b(?:architect|how (?:it|this) works|connected|flow|relationship|integrat)\b", |
| ("extract-symbols", "trace-imports", "map-components")), |
| ) |
|
|
|
|
| def scenario_for(goal): |
| """Select reusable read-only skills, not arbitrary provider operations.""" |
| matches = [(name, ops) for name, pattern, ops in SCENARIOS |
| if re.search(pattern, str(goal or "")[:700], re.I)] |
| if not matches: |
| return {"primary": "explain", "secondary": [], |
| "capabilities": ["extract-symbols", "trace-imports", "explain-evidence"]} |
| ops = list(dict.fromkeys(x for _, steps in matches[:2] for x in steps)) |
| return {"primary": matches[0][0], "secondary": [x[0] for x in matches[1:2]], |
| "capabilities": ops[:6]} |
|
|
|
|
| def safe_source_path(path): |
| if not isinstance(path, str) or len(path) > 240 or path.startswith("/") or "\\" in path: |
| return False |
| if any(x in ("", ".", "..") or x.startswith(".") for x in path.split("/")): |
| return False |
| return not SECRET.search(path) and PurePosixPath(path).suffix.lower() in EXTS |
|
|
|
|
| def blob_sha(source): |
| data = source.encode("utf-8") |
| return hashlib.sha1(b"blob " + str(len(data)).encode() + b"\0" + data).hexdigest() |
|
|
|
|
| def parse_structure(path, source): |
| """Python AST is exact; JS/TS is best-effort; other formats disclose limits.""" |
| ext = PurePosixPath(path).suffix.lower() |
| rows = source.splitlines() |
| symbols, imports, limitations = [], [], [] |
| if ext == ".py": |
| try: |
| tree = ast.parse(source) |
| except (SyntaxError, ValueError) as exc: |
| return {"language": "Python", "lines": len(rows), "symbols": [], "imports": [], |
| "limitations": ["AST unavailable: " + type(exc).__name__]} |
| for node in ast.walk(tree): |
| if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)): |
| symbols.append({"name": node.name, "line": node.lineno, |
| "kind": "class" if isinstance(node, ast.ClassDef) else "function"}) |
| elif isinstance(node, ast.Import): |
| imports.extend({"ref": alias.name, "line": node.lineno} for alias in node.names) |
| elif isinstance(node, ast.ImportFrom): |
| ref = "." * node.level + (node.module or "") |
| imports.append({"ref": ref, "line": node.lineno}) |
| if node.level and not node.module: |
| imports.extend({"ref": ref + alias.name, "line": node.lineno} |
| for alias in node.names if alias.name != "*") |
| language = "Python" |
| elif ext in (".js", ".jsx", ".mjs", ".ts", ".tsx"): |
| language = "TypeScript" if ext in (".ts", ".tsx") else "JavaScript" |
| sym = re.compile(r"\b(?:function|class|interface|type|enum)\s+([A-Za-z_$][\w$]*)") |
| imp = re.compile(r"(?:\b(?:import|export)\s+(?:[^;]*?\s+from\s+)?|\brequire\s*\(|" |
| r"\bimport\s*\()\s*['\"]([^'\"\n]+)['\"]") |
| for n, row in enumerate(rows, 1): |
| if len(row) > 2000: |
| continue |
| symbols.extend({"name": m.group(1), "line": n, "kind": "declaration"} for m in sym.finditer(row)) |
| imports.extend({"ref": m.group(1), "line": n} for m in imp.finditer(row)) |
| limitations.append("JavaScript/TypeScript uses lexical extraction, not a full parser") |
| elif ext in (".md", ".txt"): |
| language = "Markdown" if ext == ".md" else "Text" |
| if ext == ".md": |
| for n, row in enumerate(rows, 1): |
| if re.match(r"^#{1,4}\s", row): |
| symbols.append({"name": row.lstrip("# ").strip()[:100], "line": n, "kind": "heading"}) |
| elif ext == ".json": |
| language = "JSON" |
| try: |
| obj = json.loads(source) |
| if isinstance(obj, dict): |
| symbols = [{"name": str(k)[:90], "line": None, "kind": "key"} |
| for k in list(obj)[:40]] |
| except (ValueError, TypeError): |
| limitations.append("Invalid JSON") |
| else: |
| language = ext.lstrip(".").upper() |
| limitations.append("No syntax parser for this language") |
| return {"language": language, "lines": len(rows), "symbols": symbols[:40], |
| "imports": imports[:35], "limitations": limitations} |
|
|
|
|
| def candidates(path, ref): |
| """Only exact, existing tree entries can become verified dependency edges.""" |
| folder = posixpath.dirname(path) |
| ext = PurePosixPath(path).suffix.lower() |
| if ext == ".py": |
| if ref.startswith("."): |
| level = len(ref) - len(ref.lstrip(".")) |
| parts = folder.split("/") if folder else [] |
| if level > len(parts) + 1: |
| return [] |
| base = "/".join(parts[:len(parts) - level + 1]) |
| names = [posixpath.join(base, ref[level:].replace(".", "/"))] |
| else: |
| stem = ref.replace(".", "/") |
| names = [stem, posixpath.join(folder, stem)] |
| paths = [p for n in names for p in (n + ".py", n + "/__init__.py")] |
| elif ext in (".js", ".jsx", ".mjs", ".ts", ".tsx") and ref.startswith("."): |
| stem = posixpath.normpath(posixpath.join(folder, ref)) |
| paths = [stem + suffix for suffix in ("", ".js", ".ts", ".jsx", ".tsx", ".mjs", |
| "/index.js", "/index.ts", "/index.tsx")] |
| else: |
| paths = [] |
| return list(dict.fromkeys(p for p in paths if safe_source_path(p))) |
|
|
|
|
| def inspect(snapshot, path, source, read_file, goal=""): |
| """Inspect a single verified file plus up to four verified direct dependencies.""" |
| if (not isinstance(snapshot, dict) or snapshot.get("complete") is not True |
| or not isinstance(snapshot.get("items"), list) |
| or len(snapshot["items"]) > MAX_TREE |
| or not SHA.fullmatch(str(snapshot.get("sourceSha") or ""))): |
| raise ValueError("Complete SHA-pinned Git tree required") |
| index = {row["path"]: row for row in snapshot["items"] if isinstance(row, dict) |
| and isinstance(row.get("path"), str) and row.get("type") == "file"} |
| if not safe_source_path(path) or path not in index or not isinstance(source, str): |
| raise ValueError("Invalid or unsupported selected file") |
| primary = index[path] |
| if (not SHA.fullmatch(str(primary.get("sha") or "")) |
| or len(source.encode("utf-8")) > 256000 or blob_sha(source) != primary["sha"]): |
| raise ValueError("Primary source blob mismatch or oversized") |
| structure = parse_structure(path, source) |
| links = [] |
| for imp in structure["imports"]: |
| targets = [p for p in candidates(path, imp["ref"]) if p in index and p != path] |
| links.append({"ref": imp["ref"], "line": imp["line"], |
| "status": "resolved" if len(targets) == 1 else |
| "ambiguous" if targets else "external-or-unresolved", |
| "targets": targets[:4]}) |
| related, visited, consumed = [], {path}, 0 |
| for link in links: |
| if link["status"] != "resolved": |
| continue |
| target = link["targets"][0] |
| if target in visited: |
| continue |
| visited.add(target) |
| row = index[target] |
| size = row.get("sizeBytes") |
| if (len(related) >= MAX_FILES or not isinstance(size, int) |
| or size > MAX_RELATED_BYTES or consumed + size > TOTAL_RELATED_BYTES |
| or not SHA.fullmatch(str(row.get("sha") or ""))): |
| continue |
| try: |
| doc = read_file(target, snapshot["sourceSha"], MAX_RELATED_BYTES) |
| except Exception: |
| continue |
| if not isinstance(doc, dict) or doc.get("sha") != row["sha"]: |
| continue |
| body = doc.get("body") |
| if (not isinstance(body, str) or len(body.encode("utf-8")) > MAX_RELATED_BYTES |
| or blob_sha(body) != row["sha"]): |
| continue |
| consumed += len(body.encode("utf-8")) |
| extra = parse_structure(target, body) |
| related.append({"path": target, "blobSha": row["sha"], "language": extra["language"], |
| "symbols": extra["symbols"][:12], "lines": extra["lines"]}) |
| return {"schema": "swrlz-file-intelligence-v1", "readOnly": True, |
| "sourceSha": snapshot["sourceSha"], "path": path, "blobSha": primary["sha"], |
| "scenario": scenario_for(goal), "primary": structure, |
| "imports": links, "related": related, |
| "coverage": {"filesInTree": len(index), "filesInspected": 1 + len(related), |
| "relatedBytes": consumed, "repositoryComplete": False}} |
|
|
|
|
| def markdown_report(data): |
| """Bounded display/model evidence; no raw dependency source is returned.""" |
| if data.get("schema") != "swrlz-file-intelligence-v1": |
| raise ValueError("Unexpected intelligence schema") |
| q = chr(96) |
| safe = lambda x: str(x).replace(q, "'").replace("\n", " ")[:130] |
| info = data["primary"] |
| plan = data["scenario"] |
| out = ["### File intelligence · static evidence", |
| "**Scenario:** " + safe(plan["primary"]) + " · **Capabilities:** " + |
| ", ".join(plan["capabilities"]), |
| "**File:** " + q + safe(data["path"]) + q + " · " + safe(info["language"]) + |
| " · " + str(info["lines"]) + " lines"] |
| if info["symbols"]: |
| out.append("**Structures:** " + ", ".join( |
| q + safe(x["name"]) + q + (" (line " + str(x["line"]) + ")" if x.get("line") else "") |
| for x in info["symbols"][:8])) |
| linked = [x for x in data["imports"] if x["status"] == "resolved"] |
| if linked: |
| out.append("**Exact tree dependencies:** " + "; ".join( |
| q + safe(x["ref"]) + q + " → " + q + safe(x["targets"][0]) + q + |
| " (source line " + str(x["line"]) + ")" for x in linked[:7])) |
| else: |
| out.append("**Exact tree dependencies:** none resolved") |
| if data["related"]: |
| out.append("**Related source inspected:** " + ", ".join(q + safe(x["path"]) + q for x in data["related"])) |
| unresolved = sum(x["status"] != "resolved" for x in data["imports"]) |
| if unresolved: |
| out.append("**Ambiguous/external/unresolved imports:** " + str(unresolved) + |
| " (not confirmed missing or broken)") |
| if info["limitations"]: |
| out.append("**Parser limit:** " + "; ".join(info["limitations"][:2])) |
| cover = data["coverage"] |
| out.append("**Coverage:** " + str(cover["filesInspected"]) + " inspected of " + |
| str(cover["filesInTree"]) + " paths in pinned tree. Not a full repo audit; " + |
| "no tests, execution, edits or deployments.") |
| return "\n\n".join(out) |
|
|
