Skill Runner

skill

Generic skill runner framework.

Provides SkillConfig and run_skill() — a reusable pipeline for running AI agent skills against tickets/issues in CI. All domain-specific behavior is injected via callable hooks on SkillConfig, making this framework agnostic to the issue tracker, git forge, and skill content.

Usage::

from agentic_ci.skill import SkillConfig, run_skill

config = SkillConfig(
    skill_name="my-resolve",
    prompt_builder=my_prompt_fn,
    verdict_loader=my_verdict_fn,
    label_applier=my_label_fn,
)
rc = run_skill(config, ticket_key="PROJ-123", work_dir=Path("/tmp/work"), ...)

SkillHook

Bases: _SkillHookRequired

Structured definition of an extra skill to run at a pipeline hook point.

SkillConfig(skill_name, skill_source='', skill_ref='main', prompt_builder=lambda **kw: '', context_writer=_noop, verdict_loader=_noop_verdict, verdict_path_fn=lambda wd: wd / 'verdict.json', label_applier=_noop, cost_formatter=lambda d: None, extension_config_writer=_noop, pre_gates=list(), post_gates=list(), extra_skills=list(), context_dir='.context', artifacts=list(), max_retries=1, retryable_modes=frozenset({'resolve'}), backend_name='podman', harness_name='claude-code', container_image=None, container_env=dict(), container_runner=None) dataclass

Configuration for a skill run. All domain-specific behavior via hooks.

run_skill(config, ticket_key, work_dir, config_dir, *, mode='resolve', ticket=None, dry_run=False, dry_run_verdict_path=None, **extra_kwargs)

Run a skill pipeline for a single ticket. Returns exit code.

Flow: 1. Run pre-gates (skip container if any gate returns a non-None message) 2. Write context via context_writer hook 3. Write extension config via extension_config_writer hook 4. Build prompt via prompt_builder hook 5. Launch container (or dry-run) 6. Read cost data (OTEL) 7. Run post-gates 8. Load verdict via verdict_loader hook 9. Format comment and apply labels via label_applier hook

Source code in src/agentic_ci/skill.py

def run_skill(
    config: SkillConfig,
    ticket_key: str,
    work_dir: Path,
    config_dir: Path,
    *,
    mode: str = "resolve",
    ticket: dict | None = None,
    dry_run: bool = False,
    dry_run_verdict_path: Path | None = None,
    **extra_kwargs,
) -> int:
    """Run a skill pipeline for a single ticket. Returns exit code.

    Flow:
    1. Run pre-gates (skip container if any gate returns a non-None message)
    2. Write context via context_writer hook
    3. Write extension config via extension_config_writer hook
    4. Build prompt via prompt_builder hook
    5. Launch container (or dry-run)
    6. Read cost data (OTEL)
    7. Run post-gates
    8. Load verdict via verdict_loader hook
    9. Format comment and apply labels via label_applier hook
    """
    log.info("[%s] Starting %s in %s mode", ticket_key, config.skill_name, mode)

    for gate in config.pre_gates:
        result = gate(
            ticket_key=ticket_key,
            ticket=ticket,
            mode=mode,
            work_dir=work_dir,
            **extra_kwargs,
        )
        if result is not None:
            log.info("[%s] Pre-gate blocked: %s", ticket_key, result)
            return 0

    config.context_writer(
        ticket_key=ticket_key,
        ticket=ticket,
        mode=mode,
        work_dir=work_dir,
        **extra_kwargs,
    )

    if config.extra_skills:
        raw_ctx_dir = work_dir / config.context_dir
        if raw_ctx_dir.is_symlink():
            raise ValueError(f"context_dir is a symlink: {raw_ctx_dir}")
        ctx_dir = raw_ctx_dir.resolve()
        try:
            ctx_dir.relative_to(work_dir.resolve())
        except ValueError as exc:
            raise ValueError(f"context_dir escapes work_dir: {config.context_dir!r}") from exc
        ctx_dir.mkdir(parents=True, exist_ok=True)
        config_path = ctx_dir / "config.json"
        if config_path.is_symlink():
            raise ValueError(f"config.json is a symlink: {config_path}")
        config_path.write_text(
            json.dumps(
                {"extra_skills": config.extra_skills},
                indent=2,
                ensure_ascii=False,
            ),
            encoding="utf-8",
        )

    config.extension_config_writer(
        ticket_key=ticket_key,
        ticket=ticket,
        config=config,
        work_dir=work_dir,
        **extra_kwargs,
    )

    prompt = config.prompt_builder(
        ticket_key=ticket_key,
        mode=mode,
        skill_name=config.skill_name,
        **extra_kwargs,
    )
    output_file = work_dir / "agent-output.txt"

    runner = config.container_runner or _default_run_container
    runner_kwargs: dict = {"image": config.container_image}
    if config.container_env:
        runner_kwargs["container_env"] = config.container_env
    if config.container_runner is None:
        runner_kwargs["verdict_path"] = config.verdict_path_fn(work_dir)
        runner_kwargs["backend_name"] = config.backend_name
        runner_kwargs["harness_name"] = config.harness_name

    if dry_run:
        if dry_run_verdict_path:
            verdict_dest = config.verdict_path_fn(work_dir)
            verdict_dest.parent.mkdir(parents=True, exist_ok=True)
            shutil.copy2(dry_run_verdict_path, verdict_dest)
        rc = 0
    else:
        rc = runner(work_dir, prompt, output_file, **runner_kwargs)

        attempt = 0
        while (
            rc != 0
            and mode in config.retryable_modes
            and rc in TRANSIENT_EXIT_CODES
            and attempt < config.max_retries
        ):
            attempt += 1
            log.warning(
                "[%s] Transient failure (exit %d), retry %d/%d",
                ticket_key,
                rc,
                attempt,
                config.max_retries,
            )
            rc = runner(work_dir, prompt, output_file, **runner_kwargs)

    if rc != 0:
        log.error("[%s] Container exited with code %d", ticket_key, rc)
        config.label_applier(
            ticket_key=ticket_key,
            verdict=None,
            rc=rc,
            mode=mode,
            work_dir=work_dir,
            **extra_kwargs,
        )
        return rc

    cost_data = _load_otel_cost(work_dir)

    gate_errors: list[str] = []
    verdict = None
    for gate in config.post_gates:
        v, errors = gate(work_dir=work_dir, ticket_key=ticket_key, **extra_kwargs)
        if v is not None:
            verdict = v
        gate_errors.extend(errors)

    if gate_errors:
        log.error("[%s] Post-gate failures: %s", ticket_key, gate_errors)
        config.label_applier(
            ticket_key=ticket_key,
            verdict=None,
            gate_errors=gate_errors,
            mode=mode,
            work_dir=work_dir,
            **extra_kwargs,
        )
        return 1

    if verdict is None:
        verdict_error: Exception | None = None
        try:
            verdict = config.verdict_loader(work_dir)
        except Exception as exc:
            verdict_error = exc
            if not dry_run and mode in config.retryable_modes and config.max_retries > 0:
                log.warning("[%s] Verdict missing (%s), retrying once", ticket_key, exc)
                rc = runner(work_dir, prompt, output_file, **runner_kwargs)
                if rc != 0:
                    log.error("[%s] Retry container also failed (exit %d)", ticket_key, rc)
                    config.label_applier(
                        ticket_key=ticket_key,
                        verdict=None,
                        rc=rc,
                        mode=mode,
                        work_dir=work_dir,
                        **extra_kwargs,
                    )
                    return rc
                try:
                    verdict = config.verdict_loader(work_dir)
                    verdict_error = None
                except Exception as retry_exc:
                    log.error("[%s] Verdict still missing after retry: %s", ticket_key, retry_exc)
                    verdict_error = retry_exc

            if verdict is None:
                log.error("[%s] Failed to load verdict: %s", ticket_key, verdict_error)
                config.label_applier(
                    ticket_key=ticket_key,
                    verdict=None,
                    gate_errors=[str(verdict_error)],
                    mode=mode,
                    work_dir=work_dir,
                    **extra_kwargs,
                )
                return 1

    cost_summary = config.cost_formatter(cost_data)
    if cost_summary:
        verdict["_cost_summary"] = cost_summary

    config.label_applier(
        ticket_key=ticket_key,
        verdict=verdict,
        mode=mode,
        work_dir=work_dir,
        **extra_kwargs,
    )

    log.info(
        "[%s] %s complete: verdict=%s",
        ticket_key,
        config.skill_name,
        verdict.get("verdict", "unknown"),
    )
    return 0