mirror of
https://github.com/codeflash-ai/codeflash-agent.git
synced 2026-05-04 18:25:19 +00:00
81 lines
2 KiB
Python
81 lines
2 KiB
Python
"""Agent runner for CI mode.
|
|
|
|
Invokes the CLI backend (Claude, Codex, ...) with the codeflash-ci plugin
|
|
agent in the cloned repo directory. The agent reads
|
|
``.codeflash/ci-context.json`` and handles everything autonomously.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import logging
|
|
import os
|
|
from typing import TYPE_CHECKING
|
|
|
|
from .backends import get_backend
|
|
|
|
if TYPE_CHECKING:
|
|
from pathlib import Path
|
|
|
|
from .config import Config
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
|
|
async def run_agent(
|
|
config: Config,
|
|
repo_dir: Path,
|
|
token: str,
|
|
*,
|
|
agent: str = "codeflash-ci",
|
|
prompt: str = "CI: process .codeflash/ci-context.json",
|
|
timeout: int = 3600,
|
|
) -> str:
|
|
"""Run a plugin agent in the cloned repo.
|
|
|
|
Sets ``GITHUB_TOKEN`` and ``GH_TOKEN`` so the agent can use
|
|
``gh`` CLI for all GitHub interactions.
|
|
|
|
Returns the agent's stdout output.
|
|
"""
|
|
spec = get_backend(config.lead_backend)
|
|
cmd, cwd = spec.build_edit_cmd(
|
|
cli=config.cli_for_backend(config.lead_backend),
|
|
model=config.model_for_backend(config.lead_backend),
|
|
prompt=prompt,
|
|
repo_dir=repo_dir,
|
|
plugin_dir=config.plugin_dir,
|
|
agent=agent,
|
|
)
|
|
|
|
env = {**os.environ, "GITHUB_TOKEN": token, "GH_TOKEN": token}
|
|
|
|
log.info("Running agent in %s: %s", repo_dir, " ".join(cmd[:6]))
|
|
|
|
proc = await asyncio.create_subprocess_exec(
|
|
*cmd,
|
|
cwd=cwd,
|
|
stdout=asyncio.subprocess.PIPE,
|
|
stderr=asyncio.subprocess.PIPE,
|
|
env=env,
|
|
)
|
|
try:
|
|
stdout, stderr = await asyncio.wait_for(
|
|
proc.communicate(),
|
|
timeout=timeout,
|
|
)
|
|
except TimeoutError:
|
|
proc.kill()
|
|
msg = f"Agent timed out after {timeout}s"
|
|
raise TimeoutError(msg) from None
|
|
|
|
if proc.returncode != 0:
|
|
log.error(
|
|
"Agent failed (rc=%d): %s",
|
|
proc.returncode,
|
|
stderr.decode(),
|
|
)
|
|
msg = f"Agent exited with code {proc.returncode}"
|
|
raise RuntimeError(msg)
|
|
|
|
return stdout.decode()
|