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 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, and simple lists. No multi-line scalars, no anchors, 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."""
    result = {}
    current_key = None

    for raw_line in text.splitlines():
        # Strip inline comments (but not inside quoted values)
        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 = _dequote(value.strip())
                current_key = key
                if value:
                    result[key] = 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 = _dequote(value.strip())
                result[current_key][key] = value

    return result


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'),
        '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 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(perms.get('commands', []))
    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.name}: command '{token}' not in permissions.yaml")

    return errors


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', {})

    # Default for optional placeholders that the template ships commented out
    # but skills reference unconditionally. Matches the documented default.
    project_config.setdefault('TICKET_LABEL_CREATION_ALLOWED', 'false')

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

    # Determine which skills to sync
    skills_dir = CODECANNON_DIR / 'skills'
    all_skill_files = sorted(skills_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.")

        perm_errors = validate_permissions(all_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.")

        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

    print("\nDone.")

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


if __name__ == '__main__':
    main()