sync.py

#!/usr/bin/env python3
"""
CodeCannon/sync.py — generate agent-specific skill files from source skills + project config

Usage:
  ./CodeCannon/sync.py                      # use .codecannon.yaml in current dir
  ./CodeCannon/sync.py --config path/to/config.yaml
  ./CodeCannon/sync.py --force              # overwrite customized files without prompting
  ./CodeCannon/sync.py --dry-run            # preview what would be written, exits 1 if any writes pending
  ./CodeCannon/sync.py --validate           # check all placeholders are defined, exits 1 if any missing
  ./CodeCannon/sync.py --skill start        # sync only the named skill(s), comma-separated

Generated files carry a hash comment so sync.py can detect manual edits.
If a generated file has been customized (hash mismatch), sync.py warns and skips it.
Pass --force to overwrite, or delete the file to let sync.py regenerate it cleanly.
"""

import sys
import os
import re
import json
import hashlib
import argparse
import subprocess
from pathlib import Path

CODECANNON_DIR = Path(__file__).parent
MARKER = "generated by CodeCannon/sync.py"
LEGACY_MARKERS = [
    "generated by CodeCannon/sync.sh",
    "generated by agentgate/sync.sh",
]


def first_line_has_sync_marker(first_line):
    """True if the line was produced by this tool (current or legacy marker)."""
    return MARKER in first_line or any(m in first_line for m in LEGACY_MARKERS)


# ── YAML parsing ──────────────────────────────────────────────────────────────
# We parse a strict subset of YAML: top-level keys, one-level-deep key:value
# pairs, simple lists, and literal block scalars (| / |- / |+). No anchors,
# no folded scalars (>), no complex types.

def _dequote(value):
    """Strip a single matching pair of surrounding quotes.

    Double-quoted values decode the YAML escapes we actually use: \\" → " and \\\\ → \\.
    Single-quoted values are returned verbatim (no escape processing).
    Unquoted values are returned as-is.
    """
    if len(value) >= 2 and value[0] == '"' and value[-1] == '"':
        inner = value[1:-1]
        # Decode \\ first to a placeholder so \" decoding doesn't see escaped backslashes
        return inner.replace('\\\\', '\x00').replace('\\"', '"').replace('\x00', '\\')
    if len(value) >= 2 and value[0] == "'" and value[-1] == "'":
        return value[1:-1]
    return value


def parse_yaml_simple(text):
    """Parse a simple flat YAML structure into a dict.

    Supports literal block scalars (`|`, `|-`, `|+`) on top-level and nested
    keys (PLATFORM_COMPLIANCE_NOTES, SENSITIVE_AREAS_CATEGORIES, etc.) — lines
    indented deeper than the key are captured verbatim until indent drops to
    the key's level or below.
    """
    result = {}
    current_key = None

    # Literal block scalar state: set when a key's value is | / |- / |+.
    # (parent_key_or_None, key, key_indent, chomp_indicator, [raw lines])
    block = None

    def close_block():
        nonlocal block
        parent, key, _, chomp, lines = block
        content_indent = min(
            (len(l) - len(l.lstrip()) for l in lines if l.strip()), default=0)
        value = '\n'.join(l[content_indent:] if l.strip() else '' for l in lines)
        value = value.rstrip('\n')
        if value and chomp != '-':
            value += '\n'  # default "clip" chomping keeps one trailing newline
        target = result if parent is None else result[parent]
        target[key] = value
        block = None

    for raw_line in text.splitlines():
        if block is not None:
            # Inside a block scalar: blank lines and deeper-indented lines are
            # content (including lines starting with # or -); anything else
            # ends the block and falls through to normal parsing.
            if not raw_line.strip():
                block[4].append('')
                continue
            if (len(raw_line) - len(raw_line.lstrip())) > block[2]:
                block[4].append(raw_line)
                continue
            close_block()

        line = raw_line
        if not line.strip() or line.strip().startswith('#'):
            continue

        indent = len(line) - len(line.lstrip())
        stripped = line.strip()

        if indent == 0:
            if ':' in stripped:
                key, _, value = stripped.partition(':')
                key = key.strip()
                value = value.strip()
                current_key = key
                if value in ('|', '|-', '|+'):
                    block = (None, key, indent, value[1:], [])
                elif value:
                    result[key] = _dequote(value)
                else:
                    result[key] = {}
        elif indent >= 2 and current_key is not None:
            if stripped.startswith('- '):
                value = _dequote(stripped[2:].strip())
                if not isinstance(result.get(current_key), list):
                    result[current_key] = []
                result[current_key].append(value)
            elif ':' in stripped and isinstance(result.get(current_key), dict):
                key, _, value = stripped.partition(':')
                key = key.strip()
                value = value.strip()
                if value in ('|', '|-', '|+'):
                    block = (current_key, key, indent, value[1:], [])
                else:
                    result[current_key][key] = _dequote(value)

    if block is not None:
        close_block()

    return result


def _strip_inline_comment(s):
    """Strip a trailing ` #...` comment from a single-line YAML scalar, honoring quotes.

    Only a `#` outside any quoted span, preceded by whitespace or at position 0,
    starts a comment — matches the convention used elsewhere for full-line comments.
    """
    in_squote = in_dquote = False
    for idx, ch in enumerate(s):
        if ch == '"' and not in_squote:
            in_dquote = not in_dquote
        elif ch == "'" and not in_dquote:
            in_squote = not in_squote
        elif ch == '#' and not in_squote and not in_dquote and (idx == 0 or s[idx - 1].isspace()):
            return s[:idx].rstrip()
    return s


def load_schema_defaults(schema_path):
    """Parse config.schema.yaml's `placeholders:` section into {NAME: default}.

    Deliberately a dedicated parser rather than an extension of parse_yaml_simple:
    the schema nests three levels deep (placeholders: -> NAME: -> default:), one
    level deeper than parse_yaml_simple supports, and that parser is reused by
    project .codecannon.yaml loading where the flatter shape is intentional.
    Handles simple quoted values and literal block scalars (`default: |`).
    """
    defaults = {}
    lines = schema_path.read_text().splitlines()
    in_placeholders = False
    current_key = None
    i = 0
    while i < len(lines):
        line = lines[i]
        stripped = line.strip()
        if not stripped or stripped.startswith('#'):
            i += 1
            continue
        indent = len(line) - len(line.lstrip())

        if indent == 0:
            in_placeholders = stripped.rstrip(':') == 'placeholders'
            current_key = None
            i += 1
            continue

        if not in_placeholders:
            i += 1
            continue

        if indent == 2 and ':' in stripped:
            key_part, _, rest = stripped.partition(':')
            rest = rest.strip()
            if rest == '' or rest.startswith('#'):
                current_key = key_part.strip()
                i += 1
                continue

        if indent == 4 and stripped.startswith('default:') and current_key:
            value = _strip_inline_comment(stripped[len('default:'):].strip())
            if value in ('|', '|-', '|+'):
                block_lines = []
                i += 1
                while i < len(lines) and (not lines[i].strip() or (len(lines[i]) - len(lines[i].lstrip())) > 4):
                    block_lines.append(lines[i])
                    i += 1
                content_indent = min(
                    (len(l) - len(l.lstrip()) for l in block_lines if l.strip()), default=0)
                block_value = '\n'.join(
                    l[content_indent:] if l.strip() else '' for l in block_lines).rstrip('\n')
                if block_value and value != '|-':
                    block_value += '\n'
                defaults[current_key] = block_value
                continue
            defaults[current_key] = _dequote(value)
            i += 1
            continue

        i += 1

    return defaults


def apply_schema_defaults(project_config, schema_path):
    """Backfill project_config with config.schema.yaml's declared defaults, in place.

    Used by both main() and the golden-file snapshot test so the two never diverge
    on what "the real sync behavior" applies. No-op if schema_path doesn't exist.
    """
    if not schema_path.exists():
        return
    for key, default_value in load_schema_defaults(schema_path).items():
        project_config.setdefault(key, default_value)


def parse_frontmatter(text):
    """Extract YAML frontmatter between --- delimiters. Returns (fm_dict, body_str)."""
    match = re.match(r'^---\n(.*?)\n---\n(.*)', text, re.DOTALL)
    if not match:
        return {}, text.strip()
    fm_text, body = match.group(1), match.group(2)
    fm = {}
    for line in fm_text.splitlines():
        if ':' in line:
            key, _, value = line.partition(':')
            stripped_val = value.strip()
            # Handle YAML lists on a single line: [a, b, c]
            if stripped_val.startswith('[') and stripped_val.endswith(']'):
                fm[key.strip()] = [_dequote(v.strip()) for v in stripped_val[1:-1].split(',')]
            else:
                fm[key.strip()] = _dequote(stripped_val)
    return fm, body.strip()


# ── Conditional blocks ────────────────────────────────────────────────────────
# Supports {{#if KEY}} / {{/if}} and {{#if !KEY}} / {{/if}}.
# A key is "truthy" when it exists in values and is non-empty after stripping.
# The directive lines are always removed from the output.
# Nesting is supported (inner blocks are evaluated innermost-first).

_IF_OPEN = re.compile(r'^\s*\{\{#if\s+(!?)([A-Z_]+)\}\}\s*$')
_IF_CLOSE = re.compile(r'^\s*\{\{/if\}\}\s*$')


def apply_conditionals(text, values):
    """Process {{#if KEY}} / {{#if !KEY}} ... {{/if}} blocks (innermost-first)."""
    changed = True
    while changed:
        changed = False
        lines = text.split('\n')
        # Find the first {{/if}} and match it with the nearest preceding {{#if}}
        close_idx = None
        for i, line in enumerate(lines):
            if _IF_CLOSE.match(line):
                close_idx = i
                break
        if close_idx is None:
            break
        open_idx = None
        for i in range(close_idx - 1, -1, -1):
            if _IF_OPEN.match(lines[i]):
                open_idx = i
                break
        if open_idx is None:
            break  # malformed — stop processing

        m = _IF_OPEN.match(lines[open_idx])
        negated = m.group(1) == '!'
        key = m.group(2)
        raw_value = values.get(key, '').strip()
        # Empty string and boolean-like falsy strings ('false', 'no', '0')
        # are treated as falsy so config values like 'false' behave intuitively
        # in conditional blocks.
        truthy = bool(raw_value) and raw_value.lower() not in ('false', 'no', '0')
        keep = (not truthy) if negated else truthy

        if keep:
            lines = lines[:open_idx] + lines[open_idx + 1:close_idx] + lines[close_idx + 1:]
        else:
            lines = lines[:open_idx] + lines[close_idx + 1:]

        text = '\n'.join(lines)
        changed = True

    return text


# ── Placeholder substitution ──────────────────────────────────────────────────

def apply_placeholders(text, values):
    """Replace {{KEY}} tokens with configured values."""
    for key, value in values.items():
        text = text.replace('{{' + key + '}}', value)
    return text


def find_unresolved(text):
    """Return list of placeholder names that were not substituted."""
    return re.findall(r'\{\{([A-Z_]+)\}\}', text)


# ── Hash and change detection ─────────────────────────────────────────────────

def content_hash(text):
    return hashlib.md5(text.encode()).hexdigest()[:8]


def read_file_info(file_path):
    """
    Inspect an existing file for Code Cannon provenance (current or legacy sync marker).

    Returns (stored_hash, body_hash, is_generated, needs_migration) where:
      stored_hash — hash recorded in the marker line when the file was last written
      body_hash   — hash of the file's actual content (excluding the marker line)
      is_generated — True if the file was written by this sync tool (current or legacy marker)
      needs_migration — True if the marker needs rewriting (legacy position or legacy marker text)

    Comparing stored_hash vs body_hash tells us whether the file was edited
    after the last sync. Comparing body_hash vs the freshly-computed h tells
    us whether the file already matches what we'd generate today.
    """
    path = Path(file_path)
    if not path.exists():
        return None, None, False, False

    content = path.read_text()
    lines = content.rstrip('\n').split('\n')

    # Marker can be on the last line (current) or the first line (legacy).
    marker_line = None
    marker_pos = None
    if lines and first_line_has_sync_marker(lines[-1]):
        marker_line = lines[-1]
        marker_pos = 'end'
    elif lines and first_line_has_sync_marker(lines[0]):
        marker_line = lines[0]
        marker_pos = 'start'

    if marker_line is None:
        return None, None, False, False

    match = re.search(r'hash: ([a-f0-9]+)', marker_line)
    stored_hash = match.group(1) if match else None

    if marker_pos == 'end':
        body = '\n'.join(lines[:-1]) + '\n'
    else:
        # Legacy position (start) — extract body after marker line.
        # The stored_hash was computed over this same body, so an unedited
        # legacy file will pass the customization guard (body_hash == stored_hash)
        # and then fail the up-to-date check (body_hash != h) because h is
        # computed from the new full_content, triggering regeneration.
        body = '\n'.join(lines[1:]) + '\n'
    body_hash = content_hash(body)

    # Migration needed if marker is at the start (legacy position) or uses a legacy marker string
    has_legacy_marker_text = MARKER not in marker_line and any(m in marker_line for m in LEGACY_MARKERS)
    migrate = (marker_pos == 'start') or has_legacy_marker_text

    return stored_hash, body_hash, True, migrate


# ── Adapter loading ───────────────────────────────────────────────────────────

def load_adapter(adapter_name):
    """Load adapter config and header template."""
    adapter_dir = CODECANNON_DIR / 'adapters' / adapter_name
    config_path = adapter_dir / 'config.yaml'
    header_path = adapter_dir / 'header.md'

    if not config_path.exists():
        print(f"  Error: adapter '{adapter_name}' not found at {adapter_dir}/")
        return None

    config = parse_yaml_simple(config_path.read_text())
    header = header_path.read_text() if header_path.exists() else ''

    return {
        'name': adapter_name,
        'output_directory': config.get('output_directory', f'.{adapter_name}'),
        'output_extension': config.get('output_extension', '.md'),
        'permissions_file': config.get('permissions_file'),
        'header_template': header,
    }


def build_header(adapter, skill_name, fm):
    """Render the adapter's invocation header for a specific skill."""
    template = adapter['header_template']
    description = fm.get('description', skill_name)
    header = template.replace('{skill}', skill_name).replace('{description}', description)
    return header


# ── Main sync logic ───────────────────────────────────────────────────────────

def sync_skill(skill_path, adapter, project_config, project_root, args):
    """Sync a single skill file to the adapter's output directory."""
    raw = skill_path.read_text()
    fm, body = parse_frontmatter(raw)

    skill_name = fm.get('skill', skill_path.stem)
    no_header = fm.get('no_invocation_header', 'false').lower() == 'true'
    output_path_override = fm.get('output_path_override', '')

    # Evaluate conditional blocks, then substitute placeholders
    body = apply_conditionals(body, project_config)
    body = apply_placeholders(body, project_config)
    if fm.get('description'):
        fm['description'] = apply_placeholders(fm['description'], project_config)
    if output_path_override:
        output_path_override = apply_placeholders(output_path_override, project_config)

    # Warn about unresolved placeholders (check body and description)
    unresolved = find_unresolved(body)
    if fm.get('description'):
        unresolved += find_unresolved(fm['description'])
    if unresolved:
        print(f"  Warning: {skill_name} has unresolved placeholders: {', '.join(unresolved)}")

    # Build full content
    header = '' if no_header else build_header(adapter, skill_name, fm)
    full_content = header + body + '\n'

    # Determine output path
    if output_path_override:
        out_path = project_root / output_path_override
    else:
        ext = adapter['output_extension']
        out_dir = project_root / adapter['output_directory']
        out_path = out_dir / f"{skill_name}{ext}"

    # Compute hash (of content, before marker line)
    h = content_hash(full_content)
    marker_line = f"<!-- {MARKER} | skill: {skill_name} | adapter: {adapter['name']} | hash: {h} | DO NOT EDIT — run CodeCannon/sync.py to regenerate -->"
    final_content = full_content + marker_line + '\n'

    # Check existing file
    stored_hash, body_hash, is_generated, needs_migration = read_file_info(out_path)

    skill_type = fm.get('type', 'skill')
    type_tag = f" [{skill_type}]" if skill_type != 'skill' else ''

    # body_hash is computed over the file content excluding the marker line,
    # which is exactly what we'd write as full_content. If they match, we're done.
    # Exception: if the marker is in legacy position, we must rewrite to migrate it.
    if body_hash == h and not needs_migration:
        print(f"  ✓ {skill_name}{type_tag} (up to date)")
        return False

    if out_path.exists() and not is_generated and not args.force:
        print(f"  ⚠ {skill_name}{type_tag} (file exists but was not generated by Code Cannon sync — skipping, use --force to overwrite)")
        return False

    if is_generated and not args.force:
        if body_hash != stored_hash:
            # File was edited after the last sync (body no longer matches stored hash)
            print(f"  ⚠ {skill_name}{type_tag} (customized since last sync — skipping, use --force to overwrite)")
            return False
        # body_hash == stored_hash but != h: source skill changed, file unmodified → safe to regenerate

    # Write
    if args.dry_run:
        print(f"  → {skill_name}{type_tag} (would write to {out_path})")
        return True
    else:
        out_path.parent.mkdir(parents=True, exist_ok=True)
        out_path.write_text(final_content)
        action = "updated" if out_path.exists() else "written"
        print(f"  ✓ {skill_name}{type_tag} ({action} → {out_path})")
        return False


def validate_placeholders(skill_files, project_config):
    """Scan all skills for {{PLACEHOLDER}} tokens; error if any are undefined."""
    errors = []
    for skill_path in skill_files:
        raw = skill_path.read_text()
        fm, body = parse_frontmatter(raw)
        # Evaluate conditionals first — placeholders inside stripped blocks
        # are not part of the final output and should not be reported.
        text_to_check = apply_conditionals(body, project_config)
        if fm.get('description'):
            text_to_check += '\n' + apply_conditionals(fm['description'], project_config)
        missing = [p for p in find_unresolved(text_to_check) if p not in project_config]
        for p in missing:
            errors.append(f"  {skill_path.name}: {{{{{p}}}}} not defined in config")
    return errors


def _validated_commands(perms):
    """Commands permitted to appear in skill code blocks: those emitted as allow
    rules (`commands:`) plus validate-only commands (`validate_only:`, e.g. `cd`)
    that are intentionally never emitted. Both are legal in skills and must pass
    validation; only `commands:` becomes a harness allow rule."""
    return list(perms.get('commands', [])) + list(perms.get('validate_only', []))


def validate_permissions(skill_files):
    """Check that command prefixes in skill code blocks are listed in permissions.yaml."""
    perms_path = CODECANNON_DIR / 'permissions.yaml'
    if not perms_path.exists():
        return ["  permissions.yaml not found"]

    perms = parse_yaml_simple(perms_path.read_text())
    allowed = set(_validated_commands(perms))
    if not allowed:
        return ["  permissions.yaml has no commands listed"]

    code_block_re = re.compile(r'```bash\n(.*?)```', re.DOTALL)
    errors = []
    seen = set()

    for skill_path in skill_files:
        text = skill_path.read_text()
        for block in code_block_re.findall(text):
            for line in block.strip().splitlines():
                line = line.strip()
                if not line or line.startswith('#') or line.startswith('<'):
                    continue
                token = line.split()[0]
                # Only check tokens that look like shell commands
                if not (token[0].islower() or token.startswith('./')):
                    continue
                # Normalize ./CodeCannon/sync.py → CodeCannon/sync.py
                if token.startswith('./'):
                    token = token[2:]
                # For path-like commands (CodeCannon/sync.py), check the full path
                # For simple commands (git, make), check the prefix
                if token not in allowed and token not in seen:
                    seen.add(token)
                    errors.append(f"  {skill_path.parent.name}/{skill_path.name}: command '{token}' not in permissions.yaml")

    return errors


# Shell shapes the harness permission system cannot statically analyze. A
# command using any of these prompts on every run and can never be "always
# allowed" (it can't be reduced to a stable prefix rule), which is the root of
# the permission-prompt friction in issue #202. Skill Bash examples must avoid
# them — push any irreducible case into a scripts/ helper instead.
_BAD_SHAPES = [
    ('&&', 'command chaining (&&) — split into separate commands'),
    ('||', 'command chaining (||) — split, or move the logic into a scripts/ helper'),
    ('$(', 'command substitution $(...) — run the inner command separately or use a scripts/ helper'),
    ('`', 'command substitution (backticks) — avoid'),
    ('|', 'pipe (|) — move the piped logic into a scripts/ helper'),
    (';', 'statement separator (;) — split into separate commands'),
    ('2>', 'stderr redirection (2>) — drop it, or capture output in a scripts/ helper'),
    ('&>', 'output redirection (&>) — drop it, or capture output in a scripts/ helper'),
    ('>&', 'fd redirection (>&) — drop it, or capture output in a scripts/ helper'),
]


def validate_command_shapes(skill_files):
    """Flag command lines in skill code blocks that use un-allowlistable shell
    shapes (see _BAD_SHAPES). Only lines whose first token is a known command
    (per permissions.yaml) are checked, so prose, templates, and output
    examples inside code fences are not falsely flagged."""
    perms_path = CODECANNON_DIR / 'permissions.yaml'
    allowed = set()
    if perms_path.exists():
        allowed = set(_validated_commands(parse_yaml_simple(perms_path.read_text())))

    block_re = re.compile(r'```[a-z]*\n(.*?)```', re.DOTALL)
    errors = []
    for skill_path in skill_files:
        text = skill_path.read_text()
        for m in block_re.finditer(text):
            fence_line = text.count('\n', 0, m.start()) + 1
            for i, raw in enumerate(m.group(1).splitlines()):
                line = raw.strip()
                if not line or line.startswith('#') or line.startswith('<'):
                    continue
                # Recognized command line? (base command listed in permissions.yaml)
                base = line.split()[0].lstrip('./').split('/')[0]
                if base not in allowed:
                    continue
                for bad, msg in _BAD_SHAPES:
                    if bad in line:
                        lineno = fence_line + 1 + i
                        errors.append(
                            f"  {skill_path.parent.name}/{skill_path.name}:{lineno}: "
                            f"'{bad}' — {msg}\n      {line}")
                        break

    return errors


def _allow_rules_from_permissions():
    """Turn permissions.yaml's `commands:` list into harness allow rules. Commands
    under `validate_only:` (e.g. `cd`) are intentionally not emitted — blessing
    them would re-invite the compound `cd … && …` shape this work removes. That
    exclusion lives in the data (the `commands:` / `validate_only:` split), so no
    command name is special-cased here.

    The rules are broad prefixes (e.g. `Bash(git:*)`, `Bash(gh:*)`), which
    auto-approve destructive subcommands too (`git push --force`, etc.). That
    breadth is a deliberate, accepted prompt-reduction ↔ safety tradeoff — a
    project adopting Code Cannon is opting into its command surface. Do not
    narrow to per-subcommand allowlists without a decision to revisit it."""
    perms_path = CODECANNON_DIR / 'permissions.yaml'
    if not perms_path.exists():
        return []
    cmds = parse_yaml_simple(perms_path.read_text()).get('commands', [])
    return [f"Bash({c}:*)" for c in cmds]


def generate_permissions(adapter, project_root, args):
    """Merge Code Cannon's command allowlist into the adapter's committed
    permission settings so a project runs without per-command prompts out of
    the box. Non-destructive: preserves existing allow rules and every other
    setting; only appends CC rules that are missing. Idempotent. Skipped for
    adapters whose config declares no `permissions_file` (no regression)."""
    settings_rel = adapter.get('permissions_file')
    if not settings_rel:
        return False
    rules = _allow_rules_from_permissions()
    if not rules:
        return False

    settings_path = project_root / settings_rel
    existing = {}
    if settings_path.exists():
        try:
            existing = json.loads(settings_path.read_text())
        except (json.JSONDecodeError, OSError):
            existing = {}
    # Guard against a settings file that is valid JSON but not the expected
    # object shape (e.g. a top-level array, or non-dict/list at these keys) —
    # otherwise setdefault/extend below would raise on the wrong type.
    if not isinstance(existing, dict):
        existing = {}
    if not isinstance(existing.get('permissions'), dict):
        existing['permissions'] = {}
    perms = existing['permissions']
    if not isinstance(perms.get('allow'), list):
        perms['allow'] = []
    allow = perms['allow']
    added = [r for r in rules if r not in allow]
    if not added:
        return False

    allow.extend(added)
    if args.dry_run:
        print(f"  [would update] {settings_path} (+{len(added)} allow rule(s))")
        return True
    settings_path.parent.mkdir(parents=True, exist_ok=True)
    settings_path.write_text(json.dumps(existing, indent=2) + '\n')
    print(f"  ✓ permissions → {settings_path} (+{len(added)} allow rule(s))")
    return True


def self_update_or_exit():
    """Update the CodeCannon checkout via git pull --ff-only. Only runs on main."""
    try:
        result = subprocess.run(
            ['git', '-C', str(CODECANNON_DIR), 'rev-parse', '--abbrev-ref', 'HEAD'],
            capture_output=True, text=True, check=True,
        )
    except (subprocess.CalledProcessError, FileNotFoundError) as e:
        print(f"Error: could not determine CodeCannon branch ({CODECANNON_DIR}): {e}")
        sys.exit(1)

    branch = result.stdout.strip()
    if branch != 'main':
        print(f"CodeCannon is on branch '{branch}', not 'main'. Skipping auto-update.")
        print("Update manually if desired, then re-run without --update.")
        sys.exit(1)

    print(f"Updating CodeCannon ({CODECANNON_DIR})...")
    pull = subprocess.run(
        ['git', '-C', str(CODECANNON_DIR), 'pull', '--ff-only'],
        text=True,
    )
    if pull.returncode != 0:
        print("Error: git pull --ff-only failed. Resolve the issue in the CodeCannon checkout and try again.")
        sys.exit(1)


def main():
    parser = argparse.ArgumentParser(
        description='Generate agent-specific skill files from Code Cannon skills/')
    parser.add_argument('--config', default='.codecannon.yaml',
                        help='Path to project config (default: .codecannon.yaml)')
    parser.add_argument('--force', action='store_true',
                        help='Overwrite customized files without prompting')
    parser.add_argument('--dry-run', action='store_true',
                        help='Show what would be written without writing anything. Exits non-zero if any writes are pending.')
    parser.add_argument('--validate', action='store_true',
                        help='Pre-flight check: verify all {{PLACEHOLDERS}} in skills are defined in config. Exits non-zero if any are missing. Does not write files.')
    parser.add_argument('--skill', default='',
                        help='Sync only specific skill(s), comma-separated (e.g. start,submit-for-review)')
    parser.add_argument('--update', action='store_true',
                        help='Self-update the Code Cannon checkout (git pull --ff-only on main) before syncing. Stops if not on the main branch.')
    args = parser.parse_args()

    if args.update:
        self_update_or_exit()

    project_root = Path.cwd()

    # Load project config
    config_path = project_root / args.config
    if not config_path.exists():
        print(f"Error: config file not found: {config_path}")
        print(f"Copy {CODECANNON_DIR}/templates/codecannon.yaml to .codecannon.yaml and customize it.")
        sys.exit(1)

    raw_config = parse_yaml_simple(config_path.read_text())
    adapters_list = raw_config.get('adapters', [])
    project_config = raw_config.get('config', {})
    skill_group = raw_config.get('skill_group', '')

    # Apply schema-declared defaults for any placeholder the project config
    # omits (e.g. optional settings the template ships commented out but
    # skills reference unconditionally). config.schema.yaml is the single
    # source of truth for these — never hardcode a default here.
    apply_schema_defaults(project_config, CODECANNON_DIR / 'config.schema.yaml')

    if not adapters_list:
        print("Error: no adapters specified in config. Add 'adapters: [claude]' to .codecannon.yaml")
        sys.exit(1)

    if not skill_group or not isinstance(skill_group, str):
        print("Error: 'skill_group' is required in .codecannon.yaml.")
        print("  Add a top-level entry naming the skill group to enable, e.g.:")
        print("      skill_group: github-agile")
        print("  Available groups are sibling directories under CodeCannon/skills/.")
        sys.exit(1)

    # Determine which skills to sync
    skills_dir = CODECANNON_DIR / 'skills'
    group_dir = skills_dir / skill_group
    if not group_dir.is_dir():
        print(f"Error: skill group '{skill_group}' not found at {group_dir}")
        print("  Available groups:")
        for entry in sorted(skills_dir.iterdir()):
            if entry.is_dir():
                print(f"    - {entry.name}")
        sys.exit(1)
    all_skill_files = sorted(group_dir.glob('*.md'))

    if args.skill:
        requested = {s.strip() for s in args.skill.split(',')}
        skill_files = [f for f in all_skill_files if f.stem in requested]
        if not skill_files:
            print(f"Error: no matching skills found for: {args.skill}")
            sys.exit(1)
    else:
        skill_files = all_skill_files

    # --validate: pre-flight placeholder check + permissions check, no writes
    if args.validate:
        failed = False
        errors = validate_placeholders(skill_files, project_config)
        if errors:
            print("Placeholder validation failed — undefined placeholders:\n")
            for e in errors:
                print(e)
            failed = True
        else:
            print("Placeholder validation passed — all placeholders are defined.")

        # When sync.py runs from inside the CodeCannon repo itself (rather than
        # as a consumer submodule), validate permissions across every skill group
        # so a gap in a non-enabled group can't ship unnoticed.
        if CODECANNON_DIR == project_root:
            perm_skill_files = sorted(skills_dir.glob('*/*.md'))
        else:
            perm_skill_files = all_skill_files
        perm_errors = validate_permissions(perm_skill_files)
        if perm_errors:
            print("\nPermission validation failed — commands not in permissions.yaml:\n")
            for e in perm_errors:
                print(e)
            failed = True
        else:
            print("Permission validation passed — all command prefixes are listed.")

        shape_errors = validate_command_shapes(perm_skill_files)
        if shape_errors:
            print("\nCommand-shape validation failed — un-allowlistable shell shapes "
                  "(these prompt on every run and can't be 'always allowed'):\n")
            for e in shape_errors:
                print(e)
            failed = True
        else:
            print("Command-shape validation passed — all commands are allowlist-friendly.")

        if failed:
            sys.exit(1)
        return

    if args.dry_run:
        print("Dry run — no files will be written.\n")

    # Sync each adapter
    any_pending = False
    for adapter_name in adapters_list:
        adapter = load_adapter(adapter_name)
        if not adapter:
            continue

        print(f"\n[{adapter_name}] → {adapter['output_directory']}/")

        for skill_path in skill_files:
            would_write = sync_skill(skill_path, adapter, project_config, project_root, args)
            if would_write:
                any_pending = True

        # Emit/refresh the committed permission allowlist for adapters whose
        # harness supports one, so the project runs without per-command prompts.
        if generate_permissions(adapter, project_root, args):
            any_pending = True

    print("\nDone.")

    if args.dry_run and any_pending:
        sys.exit(1)


if __name__ == '__main__':
    main()