git_ops.py

"""Git worktree management and Gitea PR creation.

One persistent worktree per repo. Tasks for the same repo run consecutively
(enforced by the repo lock in agent_runner). Each task resets the worktree
to the default branch and checks out a fresh task branch.
"""

import logging
import os
from urllib.parse import urlsplit

import httpx

from config import settings
from services import proc as proc_util

logger = logging.getLogger(__name__)


def _detect_provider(clone_url: str, provider: str | None) -> str:
    """Resolve the git-host provider for a repo.

    An explicit ``provider`` ('gitea' | 'github' | 'local') wins. When it
    is missing or unrecognised the host of the clone URL is sniffed so a
    github.com repo always behaves correctly even if the column was never
    set. 'local' is never sniffed — it must be explicit (local repos have
    no clone URL at all).
    """
    if provider:
        p = provider.strip().lower()
        if p in ("gitea", "github", "local"):
            return p
    host = (urlsplit(clone_url).hostname or "").lower()
    if host == "github.com" or host.endswith(".github.com"):
        return "github"
    return "gitea"


def _resolve_token(repo_token: str | None, provider: str) -> str:
    """Pick the token for a repo without leaking one provider's into another.

    A per-repo token always wins. The global fallback is provider-specific:
    GitHub never falls back to ``settings.gitea_token`` (GitHub would reject
    it with a misleading "Invalid username or token" error). Local repos
    need no auth at all.
    """
    if provider == "local":
        return ""
    if repo_token:
        return repo_token
    if provider == "github":
        return settings.github_token or ""
    return settings.gitea_token or ""


def _parse_owner_repo(clone_url: str) -> tuple[str, str]:
    """Best-effort (owner, repo) extraction from an HTTPS clone URL."""
    path = urlsplit(clone_url).path.strip("/")
    if path.endswith(".git"):
        path = path[:-4]
    parts = [p for p in path.split("/") if p]
    if len(parts) >= 2:
        return parts[-2], parts[-1]
    return "", (parts[-1] if parts else "")


async def _run(cmd: list[str], cwd: str | None = None) -> tuple[int, str, str]:
    """Run a subprocess, return (returncode, stdout, stderr)."""
    env = {**os.environ}
    if settings.git_ssl_no_verify:
        env["GIT_SSL_NO_VERIFY"] = "true"
    rc, stdout, stderr = await proc_util.run(cmd, cwd=cwd, env=env)
    return rc, stdout.decode(), stderr.decode()


def _auth_url(clone_url: str, token: str, provider: str = "gitea") -> str:
    """Inject a token into an HTTPS clone URL using the provider's scheme.

    Gitea accepts ``token:<pat>@host`` — any username works because the
    token is validated as the password. GitHub rejects that form
    ("Password authentication is not supported for Git operations") and
    expects the token under a recognised username. ``x-access-token`` is
    the username GitHub itself uses for App/installation tokens and it
    also works for classic and fine-grained PATs, so it is the single
    form that works for every GitHub token type.
    """
    if not token or not clone_url.startswith("https://"):
        return clone_url
    userinfo = f"x-access-token:{token}" if provider == "github" else f"token:{token}"
    return clone_url.replace("https://", f"https://{userinfo}@", 1)


def get_worktree_path(repo_name: str) -> str:
    """Return the persistent worktree path for a repo."""
    return os.path.join(settings.worktree_dir, repo_name)


async def setup_worktree(
    repo_name: str,
    clone_url: str,
    default_branch: str,
    task_key: str,
    gitea_token: str | None = None,
    *,
    provider: str | None = None,
    continuation: bool = False,
) -> tuple[str, str]:
    """Prepare the per-repo worktree for a new task.

    - Creates bare repo + worktree on first use.
    - On subsequent tasks: fetches latest, resets to default branch, creates task branch.

    When ``continuation`` is True the worktree is NOT reset to the default
    branch — the existing task branch (with all its commits and working-tree
    changes) is preserved so the next agent session can pick up exactly
    where the previous one stopped.

    Returns (worktree_path, branch_name).
    """
    resolved_provider = _detect_provider(clone_url, provider)
    token = _resolve_token(gitea_token, resolved_provider)
    auth_url = _auth_url(clone_url, token, resolved_provider)

    bare_repo = os.path.join(settings.bare_repo_dir, repo_name)
    # Sanitize: spaces and other invalid chars → hyphens, lowercase
    safe_key = task_key.replace(" ", "-").replace("/", "-").strip("-")
    branch_name = f"agent/{safe_key}"
    worktree_path = get_worktree_path(repo_name)

    # --- Ensure bare repo exists ---
    if not os.path.isdir(bare_repo):
        os.makedirs(os.path.dirname(bare_repo), exist_ok=True)
        logger.info("Cloning bare repo %s", repo_name)
        rc, out, err = await _run(["git", "clone", "--bare", auth_url, bare_repo])
        if rc != 0:
            raise RuntimeError(f"git clone --bare failed: {err}")

    # Update remote URL (token may have changed)
    await _run(["git", "-C", bare_repo, "remote", "set-url", "origin", auth_url])

    # Fetch latest from remote
    rc, out, err = await _run([
        "git", "-C", bare_repo, "fetch", "origin",
        "+refs/heads/*:refs/heads/*", "--prune",
    ])
    if rc != 0:
        logger.warning("git fetch warning: %s", err)

    # --- Ensure worktree exists ---
    if not os.path.isdir(worktree_path):
        logger.info("Creating worktree for %s at %s", repo_name, worktree_path)
        rc, out, err = await _run([
            "git", "-C", bare_repo, "worktree", "add",
            worktree_path, default_branch,
        ])
        if rc != 0:
            raise RuntimeError(f"git worktree add failed: {err}")
    else:
        logger.info("Reusing existing worktree for %s", repo_name)

    # Configure git identity
    await _run(["git", "-C", worktree_path, "config", "user.email", settings.git_user_email])
    await _run(["git", "-C", worktree_path, "config", "user.name", settings.git_user_name])

    # --- Continuation mode: preserve existing branch state ---
    if continuation:
        # Just ensure we're on the task branch. Do NOT reset or clean.
        rc_check, _, _ = await _run(
            ["git", "-C", worktree_path, "rev-parse", "--verify", branch_name],
        )
        if rc_check == 0:
            await _run(["git", "-C", worktree_path, "checkout", branch_name])
            logger.info("Continuation: worktree on branch %s (preserved)", branch_name)
        else:
            # Branch doesn't exist yet — fall through to normal setup
            logger.warning(
                "Continuation requested but branch %s not found, "
                "falling back to normal setup", branch_name,
            )
            return await setup_worktree(
                repo_name, clone_url, default_branch, task_key,
                gitea_token, provider=provider, continuation=False,
            )
        return worktree_path, branch_name

    # --- Reset worktree to a clean default branch state ---
    # Hard reset first to discard any modified tracked files (e.g. packages.lock.json),
    # then checkout. This avoids "please commit or stash" errors.
    await _run(["git", "-C", worktree_path, "reset", "--hard"])
    rc, out, err = await _run(
        ["git", "-C", worktree_path, "checkout", default_branch],
    )
    if rc != 0:
        logger.warning("Checkout %s failed (%s), forcing", default_branch, err.strip())
        await _run(["git", "-C", worktree_path, "checkout", "--force", default_branch])

    # Hard reset to remote state
    await _run([
        "git", "-C", worktree_path, "reset", "--hard", f"origin/{default_branch}",
    ])

    # Mark generated lock files as skip-worktree so dotnet restore doesn't dirty the index
    rc, out, _ = await _run(
        ["git", "-C", worktree_path, "ls-files", "--", "**/packages.lock.json", "packages.lock.json"],
    )
    lock_files = [f for f in out.splitlines() if f.strip()]
    if lock_files:
        await _run(
            ["git", "-C", worktree_path, "update-index", "--skip-worktree"] + lock_files
        )

    # Remove untracked files
    await _run(["git", "-C", worktree_path, "clean", "-fdx", "--exclude=.env"])

    # Check if task branch already exists with committed work
    rc_check, _, _ = await _run(
        ["git", "-C", worktree_path, "rev-parse", "--verify", branch_name],
    )
    if rc_check == 0:
        # Branch exists — check if it has commits ahead of default_branch
        rc_ahead, ahead_out, _ = await _run([
            "git", "-C", worktree_path, "rev-list", "--count",
            f"{default_branch}..{branch_name}",
        ])
        commits_ahead = int(ahead_out.strip() or "0") if rc_ahead == 0 else 0
        if commits_ahead > 0:
            # Previous run made progress — resume from existing branch
            logger.info(
                "Resuming task branch %s (%d commits ahead of %s)",
                branch_name, commits_ahead, default_branch,
            )
            rc, out, err = await _run(
                ["git", "-C", worktree_path, "checkout", branch_name],
            )
            if rc != 0:
                raise RuntimeError(f"git checkout {branch_name} failed: {err}")
            logger.info("Worktree ready (resumed): %s on branch %s", worktree_path, branch_name)
            return worktree_path, branch_name
        else:
            # Branch exists but empty — delete and start fresh
            await _run(["git", "-C", worktree_path, "branch", "-D", branch_name])
    # else: branch doesn't exist, nothing to delete

    # Create fresh task branch
    rc, out, err = await _run([
        "git", "-C", worktree_path, "checkout", "-b", branch_name,
    ])
    if rc != 0:
        raise RuntimeError(f"git checkout -b {branch_name} failed: {err}")

    logger.info("Worktree ready: %s on branch %s", worktree_path, branch_name)
    return worktree_path, branch_name


async def refresh_repo(
    repo_name: str,
    clone_url: str,
    default_branch: str,
    gitea_token: str | None = None,
    provider: str | None = None,
) -> dict:
    """Clone bare repo + worktree if missing, or fetch + pull if they exist.

    Returns a status dict with keys: ok, message, cloned, fetched.
    """
    resolved_provider = _detect_provider(clone_url, provider)
    token = _resolve_token(gitea_token, resolved_provider)
    auth_url = _auth_url(clone_url, token, resolved_provider)
    bare_repo = os.path.join(settings.bare_repo_dir, repo_name)
    worktree_path = get_worktree_path(repo_name)

    cloned = False
    fetched = False

    # --- Ensure bare repo exists ---
    if not os.path.isdir(bare_repo):
        os.makedirs(os.path.dirname(bare_repo), exist_ok=True)
        logger.info("Cloning bare repo %s", repo_name)
        rc, out, err = await _run(["git", "clone", "--bare", auth_url, bare_repo])
        if rc != 0:
            return {"ok": False, "message": f"git clone --bare failed: {err}"}
        cloned = True

    # Update remote URL (token may have changed)
    await _run(["git", "-C", bare_repo, "remote", "set-url", "origin", auth_url])

    # Fetch latest from remote.
    # The refspec +refs/heads/*:refs/heads/* updates local branch refs in-place
    # inside the bare repo.  Git refuses to update a ref whose branch is checked
    # out in a linked worktree (e.g. refs/heads/master when the worktree has
    # master checked out).  Detach the worktree HEAD first so no branch is
    # "checked out" and git allows the update.
    worktree_was_attached = False
    if os.path.isdir(worktree_path):
        rc_head, head_out, _ = await _run(
            ["git", "-C", worktree_path, "symbolic-ref", "--short", "HEAD"],
        )
        if rc_head == 0 and head_out.strip():
            worktree_was_attached = True
            await _run(["git", "-C", worktree_path, "checkout", "--detach"])

    rc, out, err = await _run([
        "git", "-C", bare_repo, "fetch", "origin",
        "+refs/heads/*:refs/heads/*", "--prune",
    ])
    if rc != 0:
        # Re-attach before returning the error
        if worktree_was_attached:
            await _run(["git", "-C", worktree_path, "checkout", default_branch])
        return {"ok": False, "message": f"git fetch failed: {err}"}
    fetched = True

    # --- Ensure worktree exists ---
    if not os.path.isdir(worktree_path):
        logger.info("Creating worktree for %s at %s", repo_name, worktree_path)
        rc, out, err = await _run([
            "git", "-C", bare_repo, "worktree", "add",
            worktree_path, default_branch,
        ])
        if rc != 0:
            return {"ok": False, "message": f"git worktree add failed: {err}"}
        cloned = True
    else:
        # Re-attach worktree to the (now-updated) default branch.
        # The bare repo has no remote-tracking refs — branches live directly
        # in refs/heads/* — so we reset to the branch name, not origin/*.
        await _run(["git", "-C", worktree_path, "reset", "--hard"])
        rc, out, err = await _run(
            ["git", "-C", worktree_path, "checkout", default_branch],
        )
        if rc != 0:
            await _run(["git", "-C", worktree_path, "checkout", "--force", default_branch])
        await _run([
            "git", "-C", worktree_path, "reset", "--hard", default_branch,
        ])

    # Configure git identity
    await _run(["git", "-C", worktree_path, "config", "user.email", settings.git_user_email])
    await _run(["git", "-C", worktree_path, "config", "user.name", settings.git_user_name])

    action = "cloned" if cloned else "fetched"
    logger.info("Refresh complete for %s (%s)", repo_name, action)
    return {"ok": True, "message": f"Repository {action} successfully", "cloned": cloned, "fetched": fetched}


async def reset_worktree(repo_name: str, default_branch: str) -> None:
    """Reset the worktree back to default branch after task completion.

    Called in the finally block so the next task always starts clean.
    """
    worktree_path = get_worktree_path(repo_name)
    if not os.path.isdir(worktree_path):
        return

    logger.info("Resetting worktree %s to %s", repo_name, default_branch)
    await _run(["git", "-C", worktree_path, "reset", "--hard"])
    await _run(["git", "-C", worktree_path, "checkout", "--force", default_branch])
    await _run(["git", "-C", worktree_path, "reset", "--hard", f"origin/{default_branch}"])
    await _run(["git", "-C", worktree_path, "clean", "-fdx", "--exclude=.env"])


# ── Local repos (provider='local') ────────────────────────────────────────────
#
# A local repo is defined by a folder on the worker host that already holds
# a git checkout (``repos.gitea_url`` stores the path — the UI labels the
# field "Local Root Folder"). There is no remote, no clone, no bare repo and
# no per-repo worktree copy: the agent runs git commands directly inside the
# operator's folder. Nothing is ever pushed, and the folder is never
# hard-reset or cleaned — it belongs to the operator.


def is_local_provider(provider: str | None) -> bool:
    """True when the repo is a local-folder repo (provider='local')."""
    return (provider or "").strip().lower() == "local"


def resolve_local_root(local_root: str | None) -> str:
    """Normalise the Local Root Folder path. Raises on an empty value."""
    path = os.path.expanduser((local_root or "").strip())
    if not path:
        raise RuntimeError(
            "Local repo has no Local Root Folder configured "
            "(set it on the repository form)"
        )
    return os.path.abspath(path)


async def _assert_local_git_repo(local_root: str) -> None:
    if not os.path.isdir(local_root):
        raise RuntimeError(f"Local Root Folder does not exist: {local_root}")
    rc, out, err = await _run(
        ["git", "-C", local_root, "rev-parse", "--is-inside-work-tree"],
    )
    if rc != 0 or out.strip() != "true":
        raise RuntimeError(
            f"Local Root Folder is not a git repository: {local_root}"
            + (f" ({err.strip()})" if err.strip() else "")
        )


async def get_current_branch(repo_path: str) -> str:
    """Name of the currently checked-out branch ('' on detached HEAD/error)."""
    rc, out, _ = await _run(
        ["git", "-C", repo_path, "rev-parse", "--abbrev-ref", "HEAD"],
    )
    branch = out.strip() if rc == 0 else ""
    return "" if branch == "HEAD" else branch


async def setup_local_repo(
    local_root: str,
    task_key: str,
    git_flow: str,
    *,
    continuation: bool = False,
) -> tuple[str, str]:
    """Prepare a local-folder repo for a task. Returns (root_path, branch_name).

    Unlike :func:`setup_worktree` this NEVER clones, fetches, resets or
    cleans — the folder is the operator's own checkout.

    - ``git_flow='untracked'``: leave the repo exactly as it is; the agent
      only edits files and nothing is ever committed. The returned branch
      name is the currently checked-out branch (informational only).
    - ``git_flow='patch'``: check out a local ``agent/{task_key}`` branch
      from the current HEAD; the agent commits there and patches are
      exported. Requires a clean working tree, otherwise the operator's own
      uncommitted changes would be swept into the agent's commits.
    """
    root = resolve_local_root(local_root)
    await _assert_local_git_repo(root)

    current_branch = await get_current_branch(root)

    if git_flow == "untracked":
        logger.info(
            "Local repo ready (untracked flow): %s on %s",
            root, current_branch or "detached HEAD",
        )
        return root, current_branch or "HEAD"

    safe_key = task_key.replace(" ", "-").replace("/", "-").strip("-")
    branch_name = f"agent/{safe_key}"

    rc_branch, _, _ = await _run(
        ["git", "-C", root, "rev-parse", "--verify", branch_name],
    )
    branch_exists = rc_branch == 0

    # Refuse to start on a dirty tree — `git add -A` after the run would
    # otherwise swallow the operator's own work into the agent's commit.
    # Continuation is exempt: leftover changes there are the agent's own.
    if not continuation and not (branch_exists and current_branch == branch_name):
        rc, status_out, _ = await _run(["git", "-C", root, "status", "--porcelain"])
        if status_out.strip():
            raise RuntimeError(
                f"Local repo {root} has uncommitted changes. Commit or stash "
                "them first, or use the 'Untracked changes' git flow."
            )

    if branch_exists:
        if current_branch != branch_name:
            rc, _, err = await _run(["git", "-C", root, "checkout", branch_name])
            if rc != 0:
                raise RuntimeError(
                    f"git checkout {branch_name} failed in {root}: {err.strip()}"
                )
        logger.info("Local repo ready (resumed branch): %s on %s", root, branch_name)
        return root, branch_name

    rc, _, err = await _run(["git", "-C", root, "checkout", "-b", branch_name])
    if rc != 0:
        raise RuntimeError(
            f"git checkout -b {branch_name} failed in {root}: {err.strip()}"
        )
    logger.info("Local repo ready: %s on new branch %s", root, branch_name)
    return root, branch_name


async def restore_local_branch(local_root: str, original_branch: str | None) -> None:
    """Best-effort: put the operator's checkout back on its original branch.

    Called from the task's finally block for local repos using the 'patch'
    flow. Skipped when the tree is dirty (a failed run may leave work the
    operator wants to inspect) — never force-checkouts, never resets.
    """
    if not original_branch:
        return
    try:
        root = resolve_local_root(local_root)
    except RuntimeError:
        return
    current = await get_current_branch(root)
    if not current or current == original_branch:
        return
    _, status_out, _ = await _run(["git", "-C", root, "status", "--porcelain"])
    if status_out.strip():
        logger.info(
            "Local repo %s left on %s (dirty tree, not restoring %s)",
            root, current, original_branch,
        )
        return
    rc, _, err = await _run(["git", "-C", root, "checkout", original_branch])
    if rc != 0:
        logger.warning(
            "Could not restore %s to branch %s: %s", root, original_branch, err.strip(),
        )
    else:
        logger.info("Restored local repo %s to branch %s", root, original_branch)


async def refresh_local_repo(local_root: str) -> dict:
    """'Refresh' for a local repo — validate only, never clone or fetch."""
    try:
        root = resolve_local_root(local_root)
        await _assert_local_git_repo(root)
    except RuntimeError as exc:
        return {"ok": False, "message": str(exc)}
    branch = await get_current_branch(root)
    return {
        "ok": True,
        "message": f"Local repository OK at {root} (on {branch or 'detached HEAD'})",
        "cloned": False,
        "fetched": False,
    }


async def ensure_committed(worktree_path: str, task_key: str, title: str) -> bool:
    """Commit any uncommitted changes. Returns True if a commit was made."""
    rc, stdout, _ = await _run(["git", "status", "--porcelain"], cwd=worktree_path)
    if not stdout.strip():
        return False

    logger.info("Uncommitted changes detected — committing")
    await _run(["git", "add", "-A"], cwd=worktree_path)
    msg = f"[{task_key}] {title}\n\nGenerated by DevServer autonomous agent"
    # Pass the agent identity inline: worktrees already have it in repo
    # config (no-op there), but local-folder repos deliberately keep the
    # operator's own config untouched and still need agent-attributed commits.
    await _run([
        "git",
        "-c", f"user.email={settings.git_user_email}",
        "-c", f"user.name={settings.git_user_name}",
        "commit", "-m", msg,
    ], cwd=worktree_path)
    return True


async def commit_to_default_branch(
    worktree_path: str,
    branch_name: str,
    default_branch: str,
    task_key: str,
    title: str,
) -> bool:
    """Squash-merge the task branch directly onto default_branch and push.

    Uses git plumbing (commit-tree) to create a squash commit on top of the
    latest default_branch WITHOUT checking it out.  This avoids the "branch
    already checked out" error that occurs in bare-repo linked worktrees
    when ``git checkout default_branch`` is attempted, and also sidesteps
    the missing ``refs/remotes/origin/*`` issue (bare repos fetched with
    ``+refs/heads/*:refs/heads/*`` have no remote-tracking refs).

    Returns True on success.
    """
    # Fetch latest default branch into the local ref.  The bare repo
    # uses +refs/heads/*:refs/heads/* so we update the local branch
    # directly.  Running fetch from the worktree is fine — it shares
    # the same git dir.
    rc, _, err = await _run(
        ["git", "fetch", "origin",
         f"+refs/heads/{default_branch}:refs/heads/{default_branch}"],
        cwd=worktree_path,
    )
    if rc != 0:
        logger.error("fetch %s failed: %s", default_branch, err)
        return False

    # Rebase the task branch onto the freshly fetched default branch so
    # any remote changes that landed while the agent was working are
    # incorporated into the squash commit's tree. Without this the
    # task branch's tree (based on an older default_branch) would
    # clobber concurrent remote changes on push.
    rc, _, err = await _run(
        ["git", "rebase", default_branch, branch_name], cwd=worktree_path,
    )
    if rc != 0:
        logger.error("rebase %s onto %s failed: %s", branch_name, default_branch, err)
        await _run(["git", "rebase", "--abort"], cwd=worktree_path)
        return False

    # Resolve the tree of the task branch (all file content) and the
    # commit of the default branch (parent for the new commit).
    rc, tree_sha, err = await _run(
        ["git", "rev-parse", f"{branch_name}^{{tree}}"], cwd=worktree_path,
    )
    if rc != 0:
        logger.error("rev-parse %s^{tree} failed: %s", branch_name, err)
        return False
    tree_sha = tree_sha.strip()

    rc, parent_sha, err = await _run(
        ["git", "rev-parse", default_branch], cwd=worktree_path,
    )
    if rc != 0:
        logger.error("rev-parse %s failed: %s", default_branch, err)
        return False
    parent_sha = parent_sha.strip()

    # Check if the tree is identical to the parent's tree (nothing changed)
    rc, parent_tree, _ = await _run(
        ["git", "rev-parse", f"{default_branch}^{{tree}}"], cwd=worktree_path,
    )
    if rc == 0 and parent_tree.strip() == tree_sha:
        logger.warning("git_flow=commit: task branch tree identical to %s — nothing to commit", default_branch)
        return True  # no-op is not an error

    # Create a squash commit whose tree is the task branch's snapshot
    # and whose parent is the tip of the default branch.
    msg = f"[{task_key}] {title}\n\nGenerated by DevServer autonomous agent"
    rc, commit_sha, err = await _run(
        ["git", "commit-tree", tree_sha, "-p", parent_sha, "-m", msg],
        cwd=worktree_path,
    )
    if rc != 0:
        logger.error("commit-tree failed: %s", err)
        return False
    commit_sha = commit_sha.strip()

    # Fast-forward the default branch ref to the new commit and push.
    rc, _, err = await _run(
        ["git", "push", "origin", f"{commit_sha}:refs/heads/{default_branch}"],
        cwd=worktree_path,
    )
    if rc != 0:
        logger.error("push %s failed: %s", default_branch, err)
        return False

    logger.info("git_flow=commit: squash-merged %s → %s (%s)", branch_name, default_branch, commit_sha[:10])
    return True


async def create_gitea_pr(
    worktree_path: str,
    branch_name: str,
    default_branch: str,
    title: str,
    body: str,
    gitea_url: str | None = None,
    gitea_owner: str | None = None,
    gitea_repo: str | None = None,
    gitea_token: str | None = None,
    provider: str | None = None,
    clone_url: str = "",
) -> str | None:
    """Push the branch and open a pull request. Returns the PR URL or None.

    Provider-aware despite the legacy ``gitea`` name: Gitea uses
    ``{base}/api/v1/repos/{owner}/{repo}/pulls`` with a ``token`` auth
    header, GitHub uses ``api.github.com`` (or ``{host}/api/v3`` for
    GitHub Enterprise Server) with a Bearer token. The branch push works
    for both because ``origin`` was rewritten to a provider-correct
    authenticated URL in ``setup_worktree``.
    """
    # Safety net: local-folder repos must never push anywhere. The agent
    # runner coerces their git_flow away from 'branch', so reaching this
    # indicates a misconfiguration — refuse rather than touch a remote.
    if is_local_provider(provider):
        logger.error("create_gitea_pr called for a local repo — refusing to push")
        return None

    # Push branch (origin already carries provider-correct auth from setup)
    rc, out, err = await _run(
        ["git", "push", "origin", branch_name, "--force-with-lease"],
        cwd=worktree_path,
    )
    if rc != 0:
        logger.error("git push failed: %s", err)
        return None

    resolved_provider = _detect_provider(clone_url or gitea_url or "", provider)
    token = _resolve_token(gitea_token, resolved_provider)

    owner = gitea_owner or ""
    repo = gitea_repo or ""
    if (not owner or not repo) and clone_url:
        parsed_owner, parsed_repo = _parse_owner_repo(clone_url)
        owner = owner or parsed_owner
        repo = repo or parsed_repo

    if resolved_provider == "github":
        host = (urlsplit(clone_url or gitea_url or "").hostname or "github.com").lower()
        if host == "github.com" or host.endswith(".github.com"):
            api_url = f"https://api.github.com/repos/{owner}/{repo}/pulls"
        else:
            # GitHub Enterprise Server
            api_url = f"https://{host}/api/v3/repos/{owner}/{repo}/pulls"
        headers = {
            "Authorization": f"Bearer {token}",
            "Accept": "application/vnd.github+json",
            "X-GitHub-Api-Version": "2022-11-28",
            "Content-Type": "application/json",
        }
        label = "GitHub"
    else:
        base_url = gitea_url or settings.gitea_url
        owner = owner or settings.gitea_owner
        api_url = f"{base_url}/api/v1/repos/{owner}/{repo}/pulls"
        headers = {
            "Authorization": f"token {token}",
            "Content-Type": "application/json",
        }
        label = "Gitea"

    try:
        async with httpx.AsyncClient(timeout=30, verify=not settings.git_ssl_no_verify) as client:
            resp = await client.post(
                api_url,
                headers=headers,
                json={
                    "title": title,
                    "body": body,
                    "head": branch_name,
                    "base": default_branch,
                },
            )
            data = resp.json()
            pr_url = data.get("html_url")
            if not pr_url:
                logger.error("%s PR creation failed (%s): %s", label, resp.status_code, data)
                return None
            logger.info("PR created: %s", pr_url)
            return pr_url
    except Exception:
        logger.exception("%s PR creation error", label)
        return None