phy-proto-break-check#

Protobuf/gRPC breaking change detector — compares .proto schema versions and tells you exactly what will break client-generated stubs, serialized messages, and wire compatibility before you push.

Field numbers are sacred in protobuf. Reusing one corrupts serialized data in production. This tool catches that and 11 other breaking-change patterns before they reach users.

What It Detects#

IDChange TypeSeverityWhy It BreaksFN001Field number reused for different typeCRITICALCorrupts existing serialized messages silentlyFN002Field removed without [deprecated = true]BREAKINGOld clients send the field; new server ignores it (silent data loss)FT001Field type changed (e.g., int32 → string)BREAKINGWire encoding changes — deserialization failsFT002Field label changed ( repeated ↔ singular)BREAKINGCollection unpacking breaksSM001Service RPC method removedBREAKINGAll callers get NOT_FOUND; generated stubs fail to compileSM002RPC streaming mode changed (unary/server/client/bidi)BREAKINGGenerated client interface changes; existing calls failSM003Service name or package changedBREAKINGAll generated stubs must be recompiled and redeployedEN001Enum value removedBREAKINGClients may have serialized the removed value; decode failsEN002Enum value number changed for existing nameCRITICALCorrupts all previously serialized enum valuesOO001Field moved into or out of oneofBREAKINGOneof fields have different wire behaviorDP001[deprecated = true] removed from fieldNOTEField re-activated — may confuse clients that stopped sending itNB001New field added with unused numberNON-BREAKINGSafe — old clients skip unknown fieldsNB002New RPC method addedNON-BREAKINGSafe — old clients simply don't call itNB003Field marked [deprecated = true]NON-BREAKINGSoft signal — field still functional

How to Use#

text

/phy-proto-break-check
Compare proto files: before/ is HEAD~1, after/ is HEAD

text

/phy-proto-break-check
Check if my .proto changes in src/proto/ are backward-compatible using git diff

text

/phy-proto-break-check
Diff user.proto v1 vs user.proto v2 for breaking changes

The agent will:

1. Locate .proto files (from git diff, two paths, or current directory)
2. Parse both versions into structured message/service/enum maps
3. Run all 14 checks against each field, enum value, and RPC
4. Print findings grouped by CRITICAL → BREAKING → NON-BREAKING → NOTE
5. Generate a migration guide for each BREAKING change
6. Print a CI fail-gate command

---

Implementation#

When invoked, run the following Python analysis. Accept --before and --after directory paths, or use git diff to extract proto changes automatically.

python

#!/usr/bin/env python3
"""
phy-proto-break-check — Protobuf/gRPC backward-compatibility checker
Detects FN001, FN002, FT001, FT002, SM001, SM002, SM003,
        EN001, EN002, OO001, DP001, NB001, NB002, NB003
No external dependencies — zero protoc requirement.
"""
import os
import re
import sys
import subprocess
from dataclasses import dataclass, field
from pathlib import Path
from typing import Optional

# ─────────────────────────────────────────────────────
# Data models
# ─────────────────────────────────────────────────────

@dataclass
class ProtoField:
    number: int
    name: str
    type_: str          # int32, string, MyMessage, etc.
    label: str          # singular, repeated, optional, required
    deprecated: bool = False
    in_oneof: Optional[str] = None   # name of the oneof block, if any

@dataclass
class ProtoEnum:
    name: str
    values: dict[str, int]   # name → number
    deprecated_values: set[str] = field(default_factory=set)

@dataclass
class ProtoRPC:
    name: str
    request_type: str
    response_type: str
    client_streaming: bool = False
    server_streaming: bool = False

@dataclass
class ProtoMessage:
    name: str
    fields: dict[int, ProtoField]   # field_number → ProtoField
    oneofs: dict[str, list[int]]    # oneof_name → [field_numbers]
    nested: dict[str, 'ProtoMessage'] = field(default_factory=dict)

@dataclass
class ProtoService:
    name: str
    rpcs: dict[str, ProtoRPC]       # rpc_name → ProtoRPC

@dataclass
class ProtoFile:
    path: str
    package: str
    syntax: str                     # proto2 or proto3
    messages: dict[str, ProtoMessage]
    enums: dict[str, ProtoEnum]
    services: dict[str, ProtoService]

@dataclass
class BreakingChange:
    check_id: str
    severity: str          # CRITICAL / BREAKING / NON-BREAKING / NOTE
    proto_file: str
    element: str           # e.g. "MyMessage.user_id" or "UserService.GetUser"
    description: str
    migration: str

# ─────────────────────────────────────────────────────
# Proto parser (regex-based, no protoc required)
# ─────────────────────────────────────────────────────

_COMMENT_RE = re.compile(r'//[^\n]*|/\*.*?\*/', re.DOTALL)
_BLOCK_RE = re.compile(r'\{([^{}]*(?:\{[^{}]*\}[^{}]*)*)\}', re.DOTALL)

def strip_comments(text: str) -> str:
    return _COMMENT_RE.sub('', text)

def parse_field_line(line: str) -> Optional[tuple]:
    """Parse a single proto field line → (label, type, name, number, deprecated, oneof)"""
    line = line.strip().rstrip(';').strip()
    if not line or line.startswith('//') or line.startswith('option '):
        return None

    deprecated = '[deprecated = true]' in line.lower() or '[deprecated=true]' in line.lower()
    # Remove options block
    clean = re.sub(r'\[.*?\]', '', line).strip()

    # repeated/optional/required type name = number
    m = re.match(
        r'^(repeated|optional|required|map<[^>]+>)?\s*'
        r'(\S+)\s+(\w+)\s*=\s*(\d+)',
        clean
    )
    if m:
        label_raw = m.group(1) or 'singular'
        if label_raw.startswith('map<'):
            label_raw = 'map'
        type_ = m.group(2)
        name = m.group(3)
        number = int(m.group(4))
        return (label_raw, type_, name, number, deprecated)

    return None

def parse_oneof_block(block_text: str) -> tuple[str, list[str]]:
    """Extract oneof name and field names from a oneof { ... } block."""
    m = re.match(r'\s*oneof\s+(\w+)\s*\{(.*)\}', block_text, re.DOTALL)
    if not m:
        return ('', [])
    oneof_name = m.group(1)
    inner = m.group(2)
    field_names = re.findall(r'\w+\s+(\w+)\s*=\s*\d+', inner)
    return (oneof_name, field_names)

def parse_message(name: str, body: str) -> ProtoMessage:
    """Parse a message body into ProtoMessage."""
    fields_by_num: dict[int, ProtoField] = {}
    oneofs: dict[str, list[int]] = {}

    # Find oneof blocks first
    oneof_pattern = re.compile(r'\boneof\s+(\w+)\s*\{([^}]*)\}', re.DOTALL)
    oneof_field_names: dict[str, str] = {}  # field_name → oneof_name
    for m in oneof_pattern.finditer(body):
        oo_name = m.group(1)
        oo_body = m.group(2)
        oo_field_nums = []
        for line in oo_body.split('\n'):
            parsed = parse_field_line(line)
            if parsed:
                label, type_, fname, fnum, dep = parsed
                fields_by_num[fnum] = ProtoField(
                    number=fnum, name=fname, type_=type_,
                    label='oneof', deprecated=dep, in_oneof=oo_name
                )
                oo_field_nums.append(fnum)
                oneof_field_names[fname] = oo_name
        oneofs[oo_name] = oo_field_nums

    # Strip oneof blocks to avoid double-parsing
    cleaned_body = oneof_pattern.sub('', body)

    for line in cleaned_body.split('\n'):
        parsed = parse_field_line(line)
        if parsed:
            label, type_, fname, fnum, dep = parsed
            if fnum not in fields_by_num:
                fields_by_num[fnum] = ProtoField(
                    number=fnum, name=fname, type_=type_,
                    label=label, deprecated=dep,
                    in_oneof=oneof_field_names.get(fname)
                )

    return ProtoMessage(name=name, fields=fields_by_num, oneofs=oneofs)

def parse_enum(name: str, body: str) -> ProtoEnum:
    values: dict[str, int] = {}
    deprecated_values: set[str] = set()
    for line in body.split('\n'):
        line = line.strip().rstrip(';')
        if not line or line.startswith('//') or line.startswith('option'):
            continue
        deprecated = '[deprecated = true]' in line.lower()
        clean = re.sub(r'\[.*?\]', '', line).strip()
        m = re.match(r'(\w+)\s*=\s*(-?\d+)', clean)
        if m:
            vname = m.group(1)
            vnum = int(m.group(2))
            values[vname] = vnum
            if deprecated:
                deprecated_values.add(vname)
    return ProtoEnum(name=name, values=values, deprecated_values=deprecated_values)

def parse_service(name: str, body: str) -> ProtoService:
    rpcs: dict[str, ProtoRPC] = {}
    rpc_re = re.compile(
        r'rpc\s+(\w+)\s*\(\s*(stream\s+)?(\S+)\s*\)\s*returns\s*\(\s*(stream\s+)?(\S+)\s*\)',
        re.DOTALL
    )
    for m in rpc_re.finditer(body):
        rpc_name = m.group(1)
        client_stream = bool(m.group(2))
        req_type = m.group(3).strip(')').strip()
        server_stream = bool(m.group(4))
        resp_type = m.group(5).strip(')').strip()
        rpcs[rpc_name] = ProtoRPC(
            name=rpc_name,
            request_type=req_type,
            response_type=resp_type,
            client_streaming=client_stream,
            server_streaming=server_stream,
        )
    return ProtoService(name=name, rpcs=rpcs)

def parse_proto_file(path: Path) -> ProtoFile:
    """Parse a .proto file into a ProtoFile structure."""
    text = path.read_text(errors='ignore')
    text = strip_comments(text)

    package_m = re.search(r'^\s*package\s+([\w.]+)\s*;', text, re.MULTILINE)
    package = package_m.group(1) if package_m else ''

    syntax_m = re.search(r'^\s*syntax\s*=\s*"(proto[23])"\s*;', text, re.MULTILINE)
    syntax = syntax_m.group(1) if syntax_m else 'proto3'

    messages: dict[str, ProtoMessage] = {}
    enums: dict[str, ProtoEnum] = {}
    services: dict[str, ProtoService] = {}

    # Simple top-level block parser
    # Handles: message Name { ... }  enum Name { ... }  service Name { ... }
    block_re = re.compile(
        r'\b(message|enum|service)\s+(\w+)\s*\{',
        re.DOTALL
    )

    pos = 0
    while pos < len(text):
        m = block_re.search(text, pos)
        if not m:
            break
        keyword = m.group(1)
        name = m.group(2)
        start = m.end() - 1  # position of opening brace

        # Find matching closing brace
        depth = 0
        i = start
        while i < len(text):
            if text[i] == '{':
                depth += 1
            elif text[i] == '}':
                depth -= 1
                if depth == 0:
                    break
            i += 1

        body = text[start + 1:i]
        pos = i + 1

        if keyword == 'message':
            messages[name] = parse_message(name, body)
        elif keyword == 'enum':
            enums[name] = parse_enum(name, body)
        elif keyword == 'service':
            services[name] = parse_service(name, body)

    return ProtoFile(
        path=str(path),
        package=package,
        syntax=syntax,
        messages=messages,
        enums=enums,
        services=services,
    )

def parse_proto_dir(directory: Path) -> dict[str, ProtoFile]:
    """Parse all .proto files in a directory tree."""
    result = {}
    for proto_path in directory.rglob('*.proto'):
        try:
            pf = parse_proto_file(proto_path)
            # Key by relative path from directory root
            rel = str(proto_path.relative_to(directory))
            result[rel] = pf
        except Exception as e:
            print(f"  [WARN] Could not parse {proto_path}: {e}", file=sys.stderr)
    return result

# ─────────────────────────────────────────────────────
# Git diff extractor
# ─────────────────────────────────────────────────────

def extract_proto_from_git(git_ref_before: str = 'HEAD~1',
                            git_ref_after: str = 'HEAD',
                            repo_root: Path = Path('.')) -> tuple[dict, dict]:
    """
    Extract proto file content at two git refs.
    Returns (before_protos, after_protos) as dicts of {rel_path: ProtoFile}.
    """
    import tempfile

    def git_proto_files(ref: str) -> list[str]:
        result = subprocess.run(
            ['git', 'ls-tree', '-r', '--name-only', ref],
            capture_output=True, text=True, cwd=repo_root
        )
        return [f for f in result.stdout.split('\n') if f.endswith('.proto')]

    def git_show(ref: str, path: str) -> Optional[str]:
        result = subprocess.run(
            ['git', 'show', f'{ref}:{path}'],
            capture_output=True, text=True, cwd=repo_root
        )
        if result.returncode == 0:
            return result.stdout
        return None

    before_protos: dict[str, ProtoFile] = {}
    after_protos: dict[str, ProtoFile] = {}

    # Get changed proto files
    diff_result = subprocess.run(
        ['git', 'diff', '--name-only', git_ref_before, git_ref_after, '--', '*.proto'],
        capture_output=True, text=True, cwd=repo_root
    )
    changed_protos = [f for f in diff_result.stdout.split('\n') if f.endswith('.proto')]

    if not changed_protos:
        # If no explicit diff, check all proto files in HEAD
        changed_protos = git_proto_files(git_ref_after)

    with tempfile.TemporaryDirectory() as tmpdir:
        tmp = Path(tmpdir)
        for rel_path in changed_protos:
            if not rel_path.strip():
                continue
            # Write before version
            before_content = git_show(git_ref_before, rel_path)
            if before_content:
                before_file = tmp / 'before' / rel_path
                before_file.parent.mkdir(parents=True, exist_ok=True)
                before_file.write_text(before_content)
                try:
                    before_protos[rel_path] = parse_proto_file(before_file)
                except Exception:
                    pass

            # Write after version
            after_content = git_show(git_ref_after, rel_path)
            if after_content:
                after_file = tmp / 'after' / rel_path
                after_file.parent.mkdir(parents=True, exist_ok=True)
                after_file.write_text(after_content)
                try:
                    after_protos[rel_path] = parse_proto_file(after_file)
                except Exception:
                    pass

    return before_protos, after_protos

# ─────────────────────────────────────────────────────
# Diff engine
# ─────────────────────────────────────────────────────

def compare_proto_files(
    rel_path: str,
    before: ProtoFile,
    after: ProtoFile,
) -> list[BreakingChange]:
    changes: list[BreakingChange] = []

    # SM003 — Package or service names changed
    if before.package != after.package and before.package and after.package:
        changes.append(BreakingChange(
            check_id='SM003',
            severity='BREAKING',
            proto_file=rel_path,
            element=f'package {before.package}',
            description=f'Package renamed: `{before.package}` → `{after.package}`. '
                        f'All generated stub namespaces change.',
            migration=(
                f'If clients use old stubs, they must regenerate code with the new package. '
                f'Consider keeping the old package name and using an alias, or deprecate with a compatibility shim.'
            )
        ))

    # Compare messages
    for msg_name, before_msg in before.messages.items():
        if msg_name not in after.messages:
            # Entire message removed — check if any service references it
            changes.append(BreakingChange(
                check_id='SM001',
                severity='BREAKING',
                proto_file=rel_path,
                element=f'message {msg_name}',
                description=f'Message `{msg_name}` removed entirely.',
                migration=f'Mark as deprecated first: add a comment `// Deprecated: use NewMessage instead`. '
                          f'Keep for at least one release cycle before removal.'
            ))
            continue

        after_msg = after.messages[msg_name]
        changes.extend(_compare_message_fields(rel_path, msg_name, before_msg, after_msg))

    # Compare enums
    for enum_name, before_enum in before.enums.items():
        if enum_name not in after.enums:
            changes.append(BreakingChange(
                check_id='EN001',
                severity='BREAKING',
                proto_file=rel_path,
                element=f'enum {enum_name}',
                description=f'Enum `{enum_name}` removed.',
                migration=f'Mark as deprecated. Keep all enum values; remove only if usage tracking confirms zero clients.'
            ))
            continue
        after_enum = after.enums[enum_name]
        changes.extend(_compare_enum(rel_path, enum_name, before_enum, after_enum))

    # Compare services
    for svc_name, before_svc in before.services.items():
        if svc_name not in after.services:
            changes.append(BreakingChange(
                check_id='SM001',
                severity='BREAKING',
                proto_file=rel_path,
                element=f'service {svc_name}',
                description=f'Service `{svc_name}` removed. All RPCs unavailable.',
                migration=f'Deprecate the service before removal. Redirect traffic to new service.'
            ))
            continue
        after_svc = after.services[svc_name]
        changes.extend(_compare_service(rel_path, svc_name, before_svc, after_svc))

    # NB001 / NB002 — non-breaking additions
    for msg_name in after.messages:
        if msg_name not in before.messages:
            changes.append(BreakingChange(
                check_id='NB001',
                severity='NON-BREAKING',
                proto_file=rel_path,
                element=f'message {msg_name}',
                description=f'New message `{msg_name}` added.',
                migration='No action required.'
            ))
    for svc_name in after.services:
        if svc_name not in before.services:
            changes.append(BreakingChange(
                check_id='NB002',
                severity='NON-BREAKING',
                proto_file=rel_path,
                element=f'service {svc_name}',
                description=f'New service `{svc_name}` added.',
                migration='No action required.'
            ))

    return changes

def _compare_message_fields(
    rel_path: str,
    msg_name: str,
    before: ProtoMessage,
    after: ProtoMessage,
) -> list[BreakingChange]:
    changes: list[BreakingChange] = []

    # Map: field_number → field in before and after
    before_nums = set(before.fields.keys())
    after_nums = set(after.fields.keys())

    # FN001 — field number reused for different field
    for num in before_nums & after_nums:
        bf = before.fields[num]
        af = after.fields[num]
        if bf.name != af.name:
            changes.append(BreakingChange(
                check_id='FN001',
                severity='CRITICAL',
                proto_file=rel_path,
                element=f'{msg_name}.{bf.name} (field {num})',
                description=(
                    f'Field number {num} was `{bf.name}` ({bf.type_}), now `{af.name}` ({af.type_}). '
                    f'Reusing a field number CORRUPTS all existing serialized messages.'
                ),
                migration=(
                    f'NEVER reuse field numbers. Reserve the old number: '
                    f'`reserved {num};` and `reserved "{bf.name}";` '
                    f'Then assign a NEW number to `{af.name}`.'
                )
            ))

    # FN002 — field removed without deprecation
    for num in before_nums - after_nums:
        bf = before.fields[num]
        if not bf.deprecated:
            changes.append(BreakingChange(
                check_id='FN002',
                severity='BREAKING',
                proto_file=rel_path,
                element=f'{msg_name}.{bf.name} (field {num})',
                description=(
                    f'Field `{bf.name}` (number {num}, type {bf.type_}) removed without prior deprecation. '
                    f'Clients still sending this field will have data silently dropped.'
                ),
                migration=(
                    f'Before removing: add `[deprecated = true]` option and release. '
                    f'After all clients have migrated, add `reserved {num};` and `reserved "{bf.name}";` '
                    f'to prevent accidental reuse.'
                )
            ))

    # FT001 / FT002 — field type or label changed
    for num in before_nums & after_nums:
        bf = before.fields[num]
        af = after.fields[num]
        if bf.name != af.name:
            continue  # Already caught by FN001
        if bf.type_ != af.type_:
            changes.append(BreakingChange(
                check_id='FT001',
                severity='BREAKING',
                proto_file=rel_path,
                element=f'{msg_name}.{bf.name} (field {num})',
                description=(
                    f'Field type changed: `{bf.type_}` → `{af.type_}`. '
                    f'Wire encoding changes; old clients fail to deserialize.'
                ),
                migration=(
                    f'Add a new field with a new field number and the new type. '
                    f'Mark the old field `[deprecated = true]`. Migrate clients. '
                    f'Then reserve the old field number.'
                )
            ))
        if bf.label != af.label:
            changes.append(BreakingChange(
                check_id='FT002',
                severity='BREAKING',
                proto_file=rel_path,
                element=f'{msg_name}.{bf.name} (field {num})',
                description=(
                    f'Field label changed: `{bf.label}` → `{af.label}`. '
                    f'`repeated` ↔ singular conversion breaks collection unpacking in all generated stubs.'
                ),
                migration=(
                    f'Add a new field with the correct label and a new field number. '
                    f'Deprecate the old field and migrate clients.'
                )
            ))

    # OO001 — field moved into or out of oneof
    for num in before_nums & after_nums:
        bf = before.fields[num]
        af = after.fields[num]
        if bf.name != af.name:
            continue
        if (bf.in_oneof is None) != (af.in_oneof is None):
            direction = 'moved INTO oneof' if af.in_oneof else 'moved OUT OF oneof'
            changes.append(BreakingChange(
                check_id='OO001',
                severity='BREAKING',
                proto_file=rel_path,
                element=f'{msg_name}.{bf.name} (field {num})',
                description=(
                    f'Field `{bf.name}` {direction}. '
                    f'Oneof fields have different serialization behavior; existing clients break.'
                ),
                migration=(
                    f'Oneof membership is part of the wire contract. '
                    f'Add a new field in/out of the oneof with a new field number instead.'
                )
            ))

    # DP001 — deprecated flag removed
    for num in before_nums & after_nums:
        bf = before.fields[num]
        af = after.fields[num]
        if bf.name == af.name and bf.deprecated and not af.deprecated:
            changes.append(BreakingChange(
                check_id='DP001',
                severity='NOTE',
                proto_file=rel_path,
                element=f'{msg_name}.{bf.name} (field {num})',
                description=f'Deprecation annotation removed from `{bf.name}`. Field re-activated.',
                migration='Verify no clients have dropped sending this field during the deprecation period.'
            ))

    # NB001 — new fields added
    for num in after_nums - before_nums:
        af = after.fields[num]
        changes.append(BreakingChange(
            check_id='NB001',
            severity='NON-BREAKING',
            proto_file=rel_path,
            element=f'{msg_name}.{af.name} (field {num})',
            description=f'New field `{af.name}` ({af.type_}) added with number {num}.',
            migration='No action required. Old clients skip unknown fields (proto3).'
        ))

    return changes

def _compare_enum(
    rel_path: str,
    enum_name: str,
    before: ProtoEnum,
    after: ProtoEnum,
) -> list[BreakingChange]:
    changes: list[BreakingChange] = []

    for vname, vnum in before.values.items():
        if vname not in after.values:
            if vname not in before.deprecated_values:
                changes.append(BreakingChange(
                    check_id='EN001',
                    severity='BREAKING',
                    proto_file=rel_path,
                    element=f'{enum_name}.{vname} = {vnum}',
                    description=(
                        f'Enum value `{vname}` (= {vnum}) removed without prior deprecation. '
                        f'Clients with serialized data containing this value fail to decode.'
                    ),
                    migration=(
                        f'Mark as deprecated first: `{vname} = {vnum} [deprecated = true];` '
                        f'Then after migration: use `reserved "{vname}"; reserved {vnum};`'
                    )
                ))
        else:
            if after.values[vname] != vnum:
                changes.append(BreakingChange(
                    check_id='EN002',
                    severity='CRITICAL',
                    proto_file=rel_path,
                    element=f'{enum_name}.{vname}',
                    description=(
                        f'Enum value number changed: `{vname}` was {vnum}, now {after.values[vname]}. '
                        f'All existing serialized data with value {vnum} will decode to the WRONG enum constant.'
                    ),
                    migration=(
                        f'NEVER change enum value numbers. Revert immediately. '
                        f'If you need a rename, add a new alias: `{vname}_V2 = {after.values[vname]};` '
                        f'and deprecate the old name.'
                    )
                ))

    return changes

def _compare_service(
    rel_path: str,
    svc_name: str,
    before: ProtoService,
    after: ProtoService,
) -> list[BreakingChange]:
    changes: list[BreakingChange] = []

    for rpc_name, before_rpc in before.rpcs.items():
        if rpc_name not in after.rpcs:
            changes.append(BreakingChange(
                check_id='SM001',
                severity='BREAKING',
                proto_file=rel_path,
                element=f'{svc_name}.{rpc_name}',
                description=f'RPC method `{rpc_name}` removed. All callers receive NOT_FOUND / unimplemented.',
                migration=(
                    f'Mark the method as deprecated first with a comment. '
                    f'Monitor call volume to zero before removing. '
                    f'Keep the method and return UNIMPLEMENTED for a grace period.'
                )
            ))
            continue

        after_rpc = after.rpcs[rpc_name]
        if (before_rpc.client_streaming != after_rpc.client_streaming or
                before_rpc.server_streaming != after_rpc.server_streaming):
            before_mode = _stream_mode(before_rpc)
            after_mode = _stream_mode(after_rpc)
            changes.append(BreakingChange(
                check_id='SM002',
                severity='BREAKING',
                proto_file=rel_path,
                element=f'{svc_name}.{rpc_name}',
                description=(
                    f'RPC streaming mode changed: {before_mode} → {after_mode}. '
                    f'Generated client interface changes; existing calls fail.'
                ),
                migration=(
                    f'Add a new RPC method with the new streaming mode. '
                    f'Deprecate the old method. Migrate clients. Then remove.'
                )
            ))

    for rpc_name in after.rpcs:
        if rpc_name not in before.rpcs:
            changes.append(BreakingChange(
                check_id='NB002',
                severity='NON-BREAKING',
                proto_file=rel_path,
                element=f'{svc_name}.{rpc_name}',
                description=f'New RPC method `{rpc_name}` added to `{svc_name}`.',
                migration='No action required.'
            ))

    return changes

def _stream_mode(rpc: ProtoRPC) -> str:
    if rpc.client_streaming and rpc.server_streaming:
        return 'bidirectional streaming'
    if rpc.client_streaming:
        return 'client streaming'
    if rpc.server_streaming:
        return 'server streaming'
    return 'unary'

# ─────────────────────────────────────────────────────
# Main runner
# ─────────────────────────────────────────────────────

SEVERITY_ORDER = {'CRITICAL': 0, 'BREAKING': 1, 'NON-BREAKING': 2, 'NOTE': 3}
SEVERITY_EMOJI = {'CRITICAL': '💀', 'BREAKING': '🔴', 'NON-BREAKING': '🟢', 'NOTE': '🔵'}

def run_check(before_protos: dict, after_protos: dict, ci_mode: bool = False) -> int:
    all_changes: list[BreakingChange] = []

    for rel_path in set(before_protos) | set(after_protos):
        if rel_path in before_protos and rel_path in after_protos:
            changes = compare_proto_files(rel_path, before_protos[rel_path], after_protos[rel_path])
            all_changes.extend(changes)
        elif rel_path not in before_protos:
            # New proto file — no breaking changes, just note
            all_changes.append(BreakingChange(
                check_id='NB001',
                severity='NON-BREAKING',
                proto_file=rel_path,
                element='<new file>',
                description=f'New proto file added: {rel_path}',
                migration='No action required.'
            ))
        else:
            # Proto file removed — all messages/services/enums are BREAKING
            all_changes.append(BreakingChange(
                check_id='SM003',
                severity='BREAKING',
                proto_file=rel_path,
                element='<file removed>',
                description=f'Proto file `{rel_path}` removed. All types and services are gone.',
                migration='Never remove a proto file unless all clients have been migrated. '
                          'Keep the file and mark all messages/services as deprecated.'
            ))

    # Sort by severity
    all_changes.sort(key=lambda c: (SEVERITY_ORDER.get(c.severity, 4), c.proto_file, c.element))

    critical = [c for c in all_changes if c.severity == 'CRITICAL']
    breaking = [c for c in all_changes if c.severity == 'BREAKING']
    non_breaking = [c for c in all_changes if c.severity == 'NON-BREAKING']
    notes = [c for c in all_changes if c.severity == 'NOTE']

    print(f"\n🔍  phy-proto-break-check\n{'─'*60}")
    print(f"Scanned {len(before_protos)} proto file(s)  |  "
          f"💀 CRITICAL={len(critical)}  🔴 BREAKING={len(breaking)}  "
          f"🟢 NON-BREAKING={len(non_breaking)}  🔵 NOTE={len(notes)}\n")

    if not critical and not breaking:
        print("✅  No breaking changes detected.")
        if non_breaking:
            print(f"\n{len(non_breaking)} non-breaking addition(s):")
            for c in non_breaking:
                print(f"  🟢 [{c.check_id}] {c.proto_file} — {c.description}")
    else:
        # Print CRITICAL and BREAKING
        for severity in ('CRITICAL', 'BREAKING'):
            group = [c for c in all_changes if c.severity == severity]
            if not group:
                continue
            emoji = SEVERITY_EMOJI[severity]
            print(f"\n{emoji} {severity} CHANGES\n{'─'*50}")
            for c in group:
                print(f"\n[{c.check_id}] {c.proto_file}")
                print(f"  Element: {c.element}")
                print(f"  Issue:   {c.description}")
                print(f"  Fix:")
                for line in c.migration.split('\n'):
                    print(f"    {line}")

        if non_breaking or notes:
            print(f"\n{'─'*50}")
            print(f"Non-blocking: {len(non_breaking)} additions, {len(notes)} notes")

    # Migration summary
    if critical or breaking:
        print(f"\n{'═'*60}")
        print(f"MIGRATION REQUIRED: {len(critical)} CRITICAL + {len(breaking)} BREAKING changes")
        print(f"\nRequired steps before merging:")
        printed_fixes: set[str] = set()
        for c in critical + breaking:
            key = c.check_id + c.element
            if key not in printed_fixes:
                print(f"  • [{c.check_id}] {c.element}: {c.migration.split('.')[0]}")
                printed_fixes.add(key)
        print(f"\nCI fail-gate:")
        print(f"  python proto_break_check.py --before path/to/old --after path/to/new --ci")

    if ci_mode and (critical or breaking):
        print("\n[CI] Exit 1 — breaking changes detected.")
        return 1
    return 0

if __name__ == '__main__':
    import argparse

    parser = argparse.ArgumentParser(description='Protobuf/gRPC breaking change detector')
    parser.add_argument('--before', help='Directory of old .proto files (or git ref)')
    parser.add_argument('--after', help='Directory of new .proto files (or git ref)')
    parser.add_argument('--git-before', default='HEAD~1', help='Git ref for before state (default: HEAD~1)')
    parser.add_argument('--git-after', default='HEAD', help='Git ref for after state (default: HEAD)')
    parser.add_argument('--use-git', action='store_true', help='Compare using git history instead of directories')
    parser.add_argument('--ci', action='store_true', help='Exit 1 on any CRITICAL or BREAKING change')
    args = parser.parse_args()

    if args.before and args.after:
        before_protos = parse_proto_dir(Path(args.before))
        after_protos = parse_proto_dir(Path(args.after))
    elif args.use_git:
        print(f"Comparing git {args.git_before}..{args.git_after}")
        before_protos, after_protos = extract_proto_from_git(
            args.git_before, args.git_after
        )
    else:
        # Default: use git HEAD~1..HEAD
        print("No --before/--after specified. Using git HEAD~1..HEAD")
        before_protos, after_protos = extract_proto_from_git('HEAD~1', 'HEAD')

    sys.exit(run_check(before_protos, after_protos, ci_mode=args.ci))

---

Usage Examples#

1. Check current git diff (most common in CI)#

bash

python proto_break_check.py --use-git

2. Compare two versions of a proto directory#

bash

python proto_break_check.py --before proto/v1/ --after proto/v2/

3. CI pipeline (exits 1 on breaking changes)#

bash

python proto_break_check.py --use-git --ci
# In GitHub Actions:
# - name: Check proto compatibility
#   run: python proto_break_check.py --git-before origin/main --git-after HEAD --ci

4. Compare specific git refs#

bash

python proto_break_check.py --use-git --git-before v1.2.0 --git-after v2.0.0

---

Example Output#

text

🔍  phy-proto-break-check
────────────────────────────────────────────────────────────
Scanned 3 proto file(s)  |  💀 CRITICAL=1  🔴 BREAKING=3  🟢 NON-BREAKING=2  🔵 NOTE=0

💀 CRITICAL CHANGES
──────────────────────────────────────────────────────

[FN001] api/user.proto
  Element: UserProfile.email (field 3)
  Issue:   Field number 3 was `email` (string), now `phone_number` (string).
           Reusing a field number CORRUPTS all existing serialized messages.
  Fix:
    NEVER reuse field numbers. Reserve the old number:
    `reserved 3;` and `reserved "email";`
    Then assign a NEW number to `phone_number`.

🔴 BREAKING CHANGES
──────────────────────────────────────────────────────

[SM001] api/user.proto
  Element: UserService.DeleteUser
  Issue:   RPC method `DeleteUser` removed. All callers receive NOT_FOUND.
  Fix:
    Mark the method as deprecated first with a comment.
    Monitor call volume to zero before removing.

[FT001] api/payment.proto
  Element: PaymentRequest.amount (field 2)
  Issue:   Field type changed: `int64` → `double`. Wire encoding changes;
           old clients fail to deserialize.
  Fix:
    Add a new field `amount_v2 = 7 double;` and migrate clients.
    Deprecate `amount`. Reserve field 2 after migration.

[EN002] api/status.proto
  Element: OrderStatus.PENDING
  Issue:   Enum value number changed: `PENDING` was 1, now 0.
           All serialized data with value 1 will decode to the WRONG enum constant.
  Fix:
    NEVER change enum value numbers. Revert immediately.

════════════════════════════════════════════════════════════
MIGRATION REQUIRED: 1 CRITICAL + 3 BREAKING changes

Required steps before merging:
  • [FN001] UserProfile.email (field 3): NEVER reuse field numbers
  • [SM001] UserService.DeleteUser: Mark the method as deprecated first
  • [FT001] PaymentRequest.amount (field 2): Add a new field with the new type
  • [EN002] OrderStatus.PENDING: NEVER change enum value numbers

CI fail-gate:
  python proto_break_check.py --before path/to/old --after path/to/new --ci

---

Why Field Numbers Are Sacred#

In Protocol Buffers, the field number (not the field name) is the wire identifier. When you serialize a UserProfile with email = 3, the bytes on the wire say "field 3, type string". If a new client reads those bytes and field 3 is now phone_number, the client silently reads the old email value into phone_number. Data corruption is guaranteed, no error is thrown.

The protobuf spec solution:

protobuf

message UserProfile {
  // After removing email:
  reserved 3;           // prevents reuse of field number
  reserved "email";     // prevents reuse of field name
  string phone_number = 7;   // new field, new number
}

This tool enforces that discipline automatically.

---

Relationship to Other phy- Skills#

SkillProtocolWhen to Usephy-proto-break-check (this)gRPC / protobuf .protoBefore merging schema changesphy-api-changelog-genREST / OpenAPI 3.xBefore releasing a new API versionphy-api-version-auditRESTAuditing versioning consistency in source codephy-openapi-mock-serverREST / OpenAPIGenerating mock servers from specs

---

What It Does NOT Cover#

• Runtime reflection or actual message deserialization testing
• Proto3 default value semantics (use buf lint for that)
• Import cycle detection (use buf build)
• Code generation validation (use protoc --descriptor_set_out)

For full proto linting and style enforcement, combine with buf lint in CI. This tool focuses exclusively on backward-compatibility — the changes that will break existing clients.

---

Author#

Canlah AI — Run performance marketing without breaking your brand.

• GitHub: github.com/PHY041
• All Skills: clawhub.ai/PHY041