🧩 テストもSASTも素通りするAI生成コードの構造破綻「Patchwork Problem」

# 実装例: Python Import Graph 構築 (ast)
import ast
from pathlib import Path


def extract_imports(file_path: Path) -> list[dict]:
    tree = ast.parse(file_path.read_text(), filename=str(file_path))
    edges = []
    for node in ast.walk(tree):
        if isinstance(node, ast.Import):
            for alias in node.names:
                edges.append({
                    "file": str(file_path), "line": node.lineno,
                    "module": alias.name, "symbol": None, "level": 0,
                })
        elif isinstance(node, ast.ImportFrom):
            for alias in node.names:
                edges.append({
                    "file": str(file_path), "line": node.lineno,
                    "module": node.module or "", "symbol": alias.name,
                    "level": node.level,  # 相対import解決に使用
                })
    return edges

// 実装例: TypeScript Import Graph 構築 (tsc Compiler API)
import * as ts from "typescript";

function buildImportGraph(tsconfigPath: string) {
  const configFile = ts.readConfigFile(tsconfigPath, ts.sys.readFile);
  const parsed = ts.parseJsonConfigFileContent(configFile.config, ts.sys, ".");
  const program = ts.createProgram({ rootNames: parsed.fileNames, options: parsed.options });
  const edges: { file: string; line: number; module: string }[] = [];

  for (const sourceFile of program.getSourceFiles()) {
    if (sourceFile.isDeclarationFile) continue;
    ts.forEachChild(sourceFile, (node) => {
      if (ts.isImportDeclaration(node) && ts.isStringLiteral(node.moduleSpecifier)) {
        const { line } = sourceFile.getLineAndCharacterOfPosition(node.getStart());
        edges.push({ file: sourceFile.fileName, line: line + 1, module: node.moduleSpecifier.text });
      }
    });
  }
  return edges;
}

# 実装例: PyCG で呼び出しグラフを生成
pip install pycg
python -m pycg --package myrepo $(find myrepo -name '*.py') -o callgraph.json

// 実装例: 呼び出しシグネチャ互換性チェック (tsc TypeChecker)
function checkCallCompatibility(program: ts.Program) {
  const checker = program.getTypeChecker();
  const findings: { file: string; line: number; message: string }[] = [];

  for (const sourceFile of program.getSourceFiles()) {
    if (sourceFile.isDeclarationFile) continue;
    ts.forEachChild(sourceFile, function visit(node) {
      if (ts.isCallExpression(node)) {
        const signature = checker.getResolvedSignature(node);
        if (!signature) {
          const { line } = sourceFile.getLineAndCharacterOfPosition(node.getStart());
          findings.push({ file: sourceFile.fileName, line: line + 1, message: "呼び出し先シグネチャを解決できない" });
        }
      }
      ts.forEachChild(node, visit);
    });
  }
  return findings;
}

# 実装例: Python 依存マニフェストの宣言パッケージ抽出
import tomllib
from pathlib import Path


def declared_python_deps(repo_root: Path) -> set[str]:
    deps: set[str] = set()
    pyproject = repo_root / "pyproject.toml"
    if pyproject.exists():
        data = tomllib.loads(pyproject.read_text())
        raw = data.get("project", {}).get("dependencies", [])
        deps |= {d.split(";")[0].split(">=")[0].split("==")[0].strip() for d in raw}
    requirements = repo_root / "requirements.txt"
    if requirements.exists():
        for line in requirements.read_text().splitlines():
            line = line.strip()
            if line and not line.startswith("#"):
                deps.add(line.split("==")[0].split(">=")[0].strip())
    return deps

# 実装例: Configuration Graph 構築
from pathlib import Path
import yaml


def extract_config_keys(repo_root: Path) -> set[str]:
    keys: set[str] = set()
    for env_file in (".env", ".env.example"):
        p = repo_root / env_file
        if p.exists():
            for line in p.read_text().splitlines():
                if "=" in line and not line.strip().startswith("#"):
                    keys.add(line.split("=", 1)[0].strip())
    compose = repo_root / "docker-compose.yml"
    if compose.exists():
        data = yaml.safe_load(compose.read_text()) or {}
        for svc in (data.get("services") or {}).values():
            env = svc.get("environment")
            if isinstance(env, dict):
                keys |= set(env.keys())
            elif isinstance(env, list):
                keys |= {e.split("=")[0] for e in env}
    return keys

# 実装例: Pydantic v2 モデルから JSON Schema を抽出
from myapp.models import UserResponse


def extract_pydantic_schema() -> dict:
    return UserResponse.model_json_schema()

# 実装例: SQLAlchemy 2.0 でテーブル定義を抽出
from sqlalchemy import inspect
from sqlalchemy.engine import Engine


def extract_sqlalchemy_schema(engine: Engine) -> dict:
    insp = inspect(engine)
    return {table: insp.get_columns(table) for table in insp.get_table_names()}

// 実装例: Zod スキーマを JSON Schema 相当に変換
// Zod 自体の公開 API は .shape / .keyof のみで JSON Schema を直接出さないため、
// zod-to-json-schema (別パッケージ) を併用する。
import { zodToJsonSchema } from "zod-to-json-schema";
import { UserResponseSchema } from "./schemas";

const schema = zodToJsonSchema(UserResponseSchema, "UserResponse");

# 実装例: ファイルシステムリソース参照の抽出
import ast


def extract_resource_refs(tree: ast.AST) -> list[dict]:
    refs = []
    for node in ast.walk(tree):
        if isinstance(node, ast.Call):
            name = getattr(node.func, "id", None) or getattr(node.func, "attr", None)
            if name in {"open", "Path"} and node.args:
                arg = node.args[0]
                if isinstance(arg, ast.Constant) and isinstance(arg.value, str):
                    refs.append({"line": node.lineno, "path": arg.value})
    return refs

# 実装例: CFG の深い解析を pylint / ESLint に委譲 (JSON出力で後続処理に渡す)
pylint --output-format=json --disable=all --enable=unreachable myrepo/ > pylint_result.json
npx eslint --format json --rule '{"no-unreachable":"error","no-constant-condition":"error"}' src/ > eslint_result.json

# 実装例: FastAPI Routing Graph 構築
from fastapi import FastAPI


def extract_fastapi_routes(app: FastAPI) -> list[dict]:
    routes = []
    for route in app.routes:
        routes.append({
            "path": getattr(route, "path", None),
            "methods": sorted(getattr(route, "methods", None) or []),
            "guards": [dep.dependency.__name__ for dep in getattr(route, "dependencies", [])],
        })
    return routes

// 実装例: Express Routing Graph 構築 (概略)
function extractExpressRoutes(app: import("express").Express) {
  const routes: { path: string; methods: string[]; guards: string[] }[] = [];
  // app._router はランタイム内部構造 (公開APIではない) のため、
  // バージョン差異に注意しつつミドルウェア名を layer.name から収集する。
  const stack = (app as any)._router?.stack ?? [];
  for (const layer of stack) {
    if (layer.route) {
      routes.push({
        path: layer.route.path,
        methods: Object.keys(layer.route.methods),
        guards: layer.route.stack.map((s: any) => s.name).filter((n: string) => n !== "<anonymous>"),
      });
    }
  }
  return routes;
}

# 実装例: DHI (Dependency Hallucination) 検証
import httpx

PYPI_JSON = "https://pypi.org/pypi/{name}/json"
NPM_REGISTRY = "https://registry.npmjs.org/{name}"


def verify_dependency_hallucination(
    import_edges: list[dict],
    declared_deps: set[str],
    ecosystem: str,  # "pypi" | "npm"
) -> list[dict]:
    findings: list[dict] = []
    registry_cache: dict[str, bool] = {}

    for edge in import_edges:
        pkg = edge["module"].split(".")[0]
        if pkg in declared_deps:
            continue  # 宣言済み → DHI対象外

        if pkg not in registry_cache:
            url = (PYPI_JSON if ecosystem == "pypi" else NPM_REGISTRY).format(name=pkg)
            resp = httpx.get(url, timeout=5.0)
            registry_cache[pkg] = resp.status_code == 200

        findings.append({
            "invariant": "外部importはマニフェスト宣言済み、またはレジストリ実在パッケージであること",
            "category": "DHI",
            "file": edge["file"],
            "line": edge["line"],
            "symbol": pkg,
            "status": "undeclared_but_exists" if registry_cache[pkg] else "phantom_package",
        })
    return findings

# 実装例: BCI (Build/Configuration Incoherence) 検証
import ast


class UnsafeEnvAccessVisitor(ast.NodeVisitor):
    """ガードされた os.environ アクセスを除外する。"""

    def __init__(self) -> None:
        self.guarded_lines: set[int] = set()
        self.findings: list[dict] = []

    def visit_Try(self, node: ast.Try) -> None:
        handles_keyerror = any(
            h.type is None or (isinstance(h.type, ast.Name) and h.type.id == "KeyError")
            for h in node.handlers
        )
        if handles_keyerror:
            for child in ast.walk(node):
                if hasattr(child, "lineno"):
                    self.guarded_lines.add(child.lineno)
        self.generic_visit(node)

    def visit_Subscript(self, node: ast.Subscript) -> None:
        is_environ_access = (
            isinstance(node.value, ast.Attribute)
            and node.value.attr == "environ"
            and isinstance(node.value.value, ast.Name)
            and node.value.value.id == "os"
        )
        if is_environ_access and node.lineno not in self.guarded_lines:
            key = node.slice.value if isinstance(node.slice, ast.Constant) else None
            self.findings.append({"line": node.lineno, "key": key})
        self.generic_visit(node)


def verify_config_space(
    source_files: list[tuple[str, str]],  # (file_path, source_text)
    declared_config_keys: set[str],
) -> list[dict]:
    findings: list[dict] = []
    for file_path, source in source_files:
        tree = ast.parse(source, filename=file_path)
        visitor = UnsafeEnvAccessVisitor()
        visitor.visit(tree)
        for f in visitor.findings:
            if f["key"] is not None and f["key"] not in declared_config_keys:
                findings.append({
                    "invariant": "設定アクセスは宣言済み config space C に存在すること",
                    "category": "BCI",
                    "file": file_path,
                    "line": f["line"],
                    "symbol": f["key"],
                })
    return findings

# 実装例: RCF 戻り値契約 (Return-path coverage)
import ast


def terminates_properly(stmts: list[ast.stmt]) -> bool:
    if not stmts:
        return False
    last = stmts[-1]
    if isinstance(last, ast.Return) and last.value is not None:
        return True
    if isinstance(last, ast.Raise):
        return True
    if isinstance(last, ast.If):
        body_ok = terminates_properly(last.body)
        orelse_ok = terminates_properly(last.orelse) if last.orelse else False
        return body_ok and orelse_ok
    if isinstance(last, ast.Try):
        body_ok = terminates_properly(last.body)
        handlers_ok = all(terminates_properly(h.body) for h in last.handlers)
        return body_ok and handlers_ok
    return False


def check_return_path_coverage(func: ast.FunctionDef, file_path: str, declared_return_type: str | None) -> list[dict]:
    if declared_return_type is None or terminates_properly(func.body):
        return []
    return [{
        "invariant": f"宣言された戻り値型 {declared_return_type} を持つ関数は全終了パスで値を返すか例外送出すること",
        "category": "RCF",
        "file": file_path,
        "line": func.lineno,
        "symbol": func.name,
    }]

# 実装例: RCF スキーマ完全性 (schema(P.output) ⊇ schema(C.input) の近似検証)
def check_schema_subset(producer_schema: dict, consumer_required_fields: set[str], file_path: str, line: int) -> list[dict]:
    producer_fields = set(producer_schema.get("properties", {}).keys())
    missing = consumer_required_fields - producer_fields
    return [
        {
            "invariant": "schema(P.output) ⊇ schema(C.input)",
            "category": "RCF",
            "file": file_path,
            "line": line,
            "symbol": field,
            "status": "missing_in_producer_schema",
        }
        for field in missing
    ]

# 実装例: CCV (Cross-File Contract Violations) 検証
def check_middleware_registration(registered_middlewares: list[str], route_import_edges: set[str]) -> list[dict]:
    findings: list[dict] = []
    seen: set[str] = set()
    for m in registered_middlewares:
        if m not in route_import_edges:
            findings.append({
                "invariant": "登録済み middleware は route module に import されていること",
                "category": "CCV", "symbol": m, "status": "not_imported",
            })
        if m in seen:
            findings.append({
                "invariant": "middleware 登録は重複しないこと",
                "category": "CCV", "symbol": m, "status": "duplicate_registration",
            })
        seen.add(m)
    return findings


def check_cross_file_contract(producer_fields: set[str], consumer_accessed_fields: set[str]) -> list[dict]:
    missing = consumer_accessed_fields - producer_fields
    return [
        {
            "invariant": "consumer が参照するフィールドは producer 出力に含まれること",
            "category": "CCV", "symbol": f, "status": "phantom_field",
        }
        for f in missing
    ]

# 実装例: SSR (Security Structural Regression) 検証
from collections import defaultdict

PUBLIC_WHITELIST_PATTERNS = ("health", "auth", "docs", "openapi", "static")  # 実装例: public 除外リスト
MUTATING_METHODS = {"POST", "PUT", "DELETE", "PATCH"}
MIN_CLUSTER_SIZE = 4    # 出典: arXiv HTML v1, Algorithm 7 (SSR) "at least 4 endpoints"
MAJORITY_RATIO = 0.9    # 出典: arXiv HTML v1, Algorithm 7 (SSR) "ratio >= 0.9"


def detect_security_regressions(routes: list[dict]) -> list[dict]:
    """routes: [{"path": str, "method": str, "guards": list[str], "file": str, "line": int}]"""
    clusters: dict[str, list[dict]] = defaultdict(list)
    for r in routes:
        segment = r["path"].split("/")[1] if "/" in r["path"] else r["path"]
        if any(w in segment for w in PUBLIC_WHITELIST_PATTERNS):
            continue
        clusters[segment].append(r)

    findings: list[dict] = []
    for segment, cluster in clusters.items():
        if len(cluster) < MIN_CLUSTER_SIZE:
            continue
        guard_counts: dict[str, int] = defaultdict(int)
        for r in cluster:
            for g in (r["guards"] or ["<none>"]):
                guard_counts[g] += 1
        majority_guard, majority_count = max(guard_counts.items(), key=lambda kv: kv[1])
        if majority_guard == "<none>" or majority_count / len(cluster) < MAJORITY_RATIO:
            continue
        for r in cluster:
            if r["method"] in MUTATING_METHODS and majority_guard not in (r["guards"] or []):
                findings.append({
                    "invariant": f"guarded_by(r, {majority_guard})",
                    "category": "SSR",
                    "file": r.get("file"),
                    "line": r.get("line"),
                    "symbol": f"{r['method']} {r['path']}",
                    "status": "missing_guard",
                })
    return findings

# 実装例: Evidence Trace の統一出力
from datetime import datetime, timezone


def to_evidence_trace(finding: dict, constraint: str) -> dict:
    return {
        "category": finding.get("category"),
        "invariant": finding.get("invariant"),
        "file": finding.get("file"),
        "line": finding.get("line"),
        "symbol": finding.get("symbol"),
        "constraint": constraint,               # 成立すべき制約 (修復のヒント)
        "status": finding.get("status", "violation"),
        "detected_at": datetime.now(timezone.utc).isoformat(),
    }

{
  "category": "SRF",
  "invariant": "resolve(r,f,M)→d が全シンボル参照で成立すること",
  "file": "src/services/user_service.py",
  "line": 42,
  "symbol": "UserRepository.find_by_email",
  "constraint": "UserRepository モジュールに find_by_email メソッドが定義されていること",
  "status": "unresolved_symbol",
  "detected_at": "2026-07-14T00:00:00+00:00"
}

# 実装案: GitHub Actions への統合イメージ
# (論文はCI実装を提供していないため、検出パイプラインを想定した独自ラッパーを前提にした構成例)
name: structural-consistency-check

on:
  pull_request:
    branches: [main]

jobs:
  patchwork-gate:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
        with:
          fetch-depth: 0

      - name: Build repo graphs and run detectors
        run: |
          patchwork-check \
            --changed-files "$(git diff --name-only origin/main...HEAD)" \
            --format sarif \
            --output findings.sarif

      - name: Upload SARIF for Code Scanning
        uses: github/codeql-action/upload-sarif@v3
        with:
          sarif_file: findings.sarif

      - name: Post PR comment with evidence trace
        uses: reviewdog/action-suggester@v1
        with:
          reporter: github-pr-review