A Deterministic Daemon That Turns a Labelled Issue Into a Pull Request

# Broken. Do not copy this.
agent.run(f"""
Read issue #{issue_number}, fix it, and open a PR.
You have git and gh available. Use them however you need to.
""")

# daemon.py
import subprocess
from dataclasses import dataclass
from pathlib import Path

import requests

GITHUB_API = "https://api.github.com"


@dataclass
class Issue:
    number: int
    title: str
    body: str
    labels: list[str]


class GitHubClient:
    def __init__(self, token: str, repo: str):
        self.repo = repo
        self.session = requests.Session()
        self.session.headers.update({
            "Authorization": f"Bearer {token}",
            "Accept": "application/vnd.github+json",
        })

    def issues_with_label(self, label: str) -> list[Issue]:
        resp = self.session.get(
            f"{GITHUB_API}/repos/{self.repo}/issues",
            params={"labels": label, "state": "open"},
        )
        resp.raise_for_status()
        return [
            Issue(i["number"], i["title"], i.get("body") or "", [l["name"] for l in i["labels"]])
            for i in resp.json()
            if "pull_request" not in i
        ]

    def set_labels(self, issue_number: int, labels: list[str]) -> None:
        resp = self.session.put(
            f"{GITHUB_API}/repos/{self.repo}/issues/{issue_number}/labels",
            json={"labels": labels},
        )
        resp.raise_for_status()

    def open_pull_request(self, title: str, head: str, base: str, body: str) -> str:
        resp = self.session.post(
            f"{GITHUB_API}/repos/{self.repo}/pulls",
            json={"title": title, "head": head, "base": base, "body": body},
        )
        resp.raise_for_status()
        return resp.json()["html_url"]


def generate_patch(issue: Issue, workspace: Path) -> str:
    """Pluggable patch-generation step.

    In production this calls out to a model. For this example, and for
    tests, it is a deterministic stub — the point of the split is that this
    function's output is the only thing the rest of the daemon trusts, and
    it can be swapped without touching commit/push/PR logic at all.
    """
    target = workspace / "NOTES.md"
    original = target.read_text() if target.exists() else ""
    target.write_text(original + f"\n- addressed issue #{issue.number}: {issue.title}\n")
    diff = subprocess.run(
        ["git", "diff"], cwd=workspace, capture_output=True, text=True, check=True
    )
    return diff.stdout


def create_worktree(repo_path: Path, branch: str) -> Path:
    worktree_path = repo_path.parent / f"worktree-{branch}"
    subprocess.run(
        ["git", "worktree", "add", "-b", branch, str(worktree_path)],
        cwd=repo_path, check=True,
    )
    return worktree_path


def remove_worktree(repo_path: Path, worktree_path: Path, branch: str) -> None:
    subprocess.run(
        ["git", "worktree", "remove", "--force", str(worktree_path)],
        cwd=repo_path, check=False,
    )
    subprocess.run(["git", "branch", "-D", branch], cwd=repo_path, check=False)


def patch_applies_cleanly(worktree_path: Path, patch: str) -> bool:
    result = subprocess.run(
        ["git", "apply", "--check", "-"],
        cwd=worktree_path, input=patch, text=True, capture_output=True,
    )
    return result.returncode == 0


def process_issue(gh: GitHubClient, repo_path: Path, issue: Issue) -> str | None:
    if "agent-ready" not in issue.labels:
        return None  # not our concern; also guards against a stale poll result

    gh.set_labels(issue.number, ["agent-in-progress"])

    branch = f"agent/issue-{issue.number}"
    worktree_path = create_worktree(repo_path, branch)
    try:
        patch = generate_patch(issue, worktree_path)
        if not patch.strip():
            gh.set_labels(issue.number, ["agent-ready"])  # nothing to do; retry later
            return None

        if not patch_applies_cleanly(worktree_path, patch):
            gh.set_labels(issue.number, ["agent-failed"])
            return None

        subprocess.run(["git", "add", "-A"], cwd=worktree_path, check=True)
        subprocess.run(
            ["git", "commit", "-m", f"Address issue #{issue.number}: {issue.title}"],
            cwd=worktree_path, check=True,
        )
        subprocess.run(["git", "push", "origin", branch], cwd=worktree_path, check=True)

        pr_url = gh.open_pull_request(
            title=f"Fix #{issue.number}: {issue.title}",
            head=branch,
            base="main",
            body=f"Closes #{issue.number}",
        )
        gh.set_labels(issue.number, ["agent-done"])
        return pr_url
    finally:
        remove_worktree(repo_path, worktree_path, branch)


def poll_once(gh: GitHubClient, repo_path: Path) -> list[str]:
    results = []
    for issue in gh.issues_with_label("agent-ready"):
        pr_url = process_issue(gh, repo_path, issue)
        if pr_url:
            results.append(pr_url)
    return results

# test_daemon.py
from pathlib import Path
from unittest.mock import MagicMock

from daemon import Issue, process_issue


def test_issue_without_agent_ready_label_is_never_processed(tmp_path):
    """The load-bearing test: an issue whose label already moved past
    agent-ready (in-progress, done, or removed entirely) must not be
    reprocessed, even if a stale poll result still lists it."""
    gh = MagicMock()
    issue = Issue(number=1, title="Already claimed", body="", labels=["agent-in-progress"])

    result = process_issue(gh, tmp_path, issue)

    assert result is None
    gh.set_labels.assert_not_called()


def test_issue_with_agent_ready_label_is_claimed_first(tmp_path, monkeypatch):
    gh = MagicMock()
    issue = Issue(number=2, title="Needs work", body="", labels=["agent-ready"])

    calls = []
    gh.set_labels.side_effect = lambda num, labels: calls.append((num, tuple(labels)))

    import daemon
    monkeypatch.setattr(daemon, "create_worktree", lambda repo, branch: tmp_path)
    monkeypatch.setattr(daemon, "generate_patch", lambda issue, ws: "")
    monkeypatch.setattr(daemon, "remove_worktree", lambda *a, **k: None)

    process_issue(gh, tmp_path, issue)

    # The very first label transition must be the claim, before any patch
    # work happens — this is what makes a second concurrent poll see
    # agent-in-progress rather than agent-ready.
    assert calls[0] == (2, ("agent-in-progress",))