473 lines
22 KiB
Python
473 lines
22 KiB
Python
import json
|
|
import os
|
|
import re
|
|
import time
|
|
from .adapters import ADAPTERS # type: ignore[reportMissingImports]
|
|
from .detection import resolve_provider # type: ignore[reportMissingImports]
|
|
from .errors import TrackerError # type: ignore[reportMissingImports]
|
|
from .models import CompletedCommand, Envelope, ResourceRef, RetryPolicy # type: ignore[reportMissingImports]
|
|
from .runner import SubprocessRunner # type: ignore[reportMissingImports]
|
|
|
|
|
|
TRANSIENT_MARKERS = ("timeout", "timed out", "connection", "network", "temporarily", "try again", "rate limit", "429", "502", "503", "504")
|
|
AUTH_MARKERS = ("not logged", "authentication", "unauthorized", "forbidden", "login", "token")
|
|
NOT_FOUND_MARKERS = ("not found", "does not exist", "unknown issue", "unknown pull")
|
|
|
|
|
|
class Tracker:
|
|
"""Provider-neutral facade. Every method returns the CLI-compatible envelope dict."""
|
|
|
|
def __init__(self, provider=None, *, runner=None, repo=None, cwd=None, retry=None, env=None, external_associations=None):
|
|
self.runner = runner or SubprocessRunner()
|
|
self.repo = repo
|
|
self.cwd = cwd or os.getcwd()
|
|
self.env = env
|
|
self.external_associations = external_associations
|
|
self.retry = retry or RetryPolicy()
|
|
remote = None
|
|
if provider is None and not (env or os.environ).get("TRACKER_PROVIDER"):
|
|
remote = self._discover_remote()
|
|
self.provider = resolve_provider(provider, env=env, remote=remote)
|
|
self.adapter = ADAPTERS[self.provider](repo=repo)
|
|
|
|
def _discover_remote(self):
|
|
try:
|
|
result = self._run_runner(["git", "remote", "get-url", "origin"], retries=1)
|
|
except TrackerError:
|
|
return None
|
|
return result.stdout.strip() or None
|
|
|
|
def _run_runner(self, argv, *, retries=None) -> CompletedCommand:
|
|
attempts = retries or self.retry.attempts
|
|
last = None
|
|
for attempt in range(attempts):
|
|
try:
|
|
if hasattr(self.runner, "run"):
|
|
response = self.runner.run(argv, cwd=self.cwd)
|
|
else:
|
|
response = self.runner(argv)
|
|
except (OSError, TimeoutError) as exc:
|
|
last = CompletedCommand(stderr=str(exc), returncode=1)
|
|
else:
|
|
if isinstance(response, CompletedCommand):
|
|
last = response
|
|
elif isinstance(response, tuple):
|
|
last = CompletedCommand(*response)
|
|
elif isinstance(response, dict):
|
|
last = CompletedCommand(**response)
|
|
else:
|
|
raise TypeError("runner must return CompletedCommand, tuple, or dict")
|
|
if last.returncode == 0:
|
|
return last
|
|
if not self._is_transient(last.stderr) or attempt == attempts - 1:
|
|
return last
|
|
if self.retry.delay:
|
|
time.sleep(self.retry.delay)
|
|
return last or CompletedCommand(stderr="provider runner returned no result", returncode=1)
|
|
|
|
@staticmethod
|
|
def _number(value):
|
|
try:
|
|
number = int(value)
|
|
except (TypeError, ValueError) as error:
|
|
raise TrackerError("invalid_input", f"invalid resource number: {value}") from error
|
|
if number < 1:
|
|
raise TrackerError("invalid_input", "resource number must be positive")
|
|
return number
|
|
|
|
@staticmethod
|
|
def _is_transient(message):
|
|
text = (message or "").lower()
|
|
return any(marker in text for marker in TRANSIENT_MARKERS)
|
|
|
|
def _failure(self, operation, message, *, returncode=1, attempts=None, uncertain=False):
|
|
text = (message or "provider command failed").strip()
|
|
lower = text.lower()
|
|
if uncertain:
|
|
code, retryable = "uncertain_outcome", False
|
|
elif any(marker in lower for marker in AUTH_MARKERS):
|
|
code, retryable = "auth_required", False
|
|
elif "no such file" in lower or ("executable" in lower and "not found" in lower):
|
|
code, retryable = "provider_cli_missing", False
|
|
elif any(marker in lower for marker in NOT_FOUND_MARKERS):
|
|
code, retryable = "not_found", False
|
|
elif "rate limit" in lower or "429" in lower:
|
|
code, retryable = "rate_limited", True
|
|
elif self._is_transient(lower):
|
|
code, retryable = "transient_failure", True
|
|
else:
|
|
code, retryable = "provider_error", False
|
|
return TrackerError(
|
|
code,
|
|
text,
|
|
retryable=retryable,
|
|
provider=self.provider,
|
|
operation=operation,
|
|
details={"returncode": returncode, **({"attempts": attempts} if attempts else {})},
|
|
)
|
|
|
|
def _run(self, operation, argv, *, mutating=False, uncertain=False) -> CompletedCommand:
|
|
result = self._run_runner(argv)
|
|
if result.returncode:
|
|
error = self._failure(
|
|
operation,
|
|
result.stderr or result.stdout,
|
|
returncode=result.returncode,
|
|
attempts=self.retry.attempts,
|
|
uncertain=uncertain and self._is_transient(result.stderr),
|
|
)
|
|
raise error
|
|
return result
|
|
|
|
def _json(self, operation, argv, *, mutating=False, uncertain=False):
|
|
result = self._run(operation, argv, mutating=mutating, uncertain=uncertain)
|
|
text = result.stdout.strip()
|
|
if not text:
|
|
return {}
|
|
try:
|
|
return json.loads(text)
|
|
except (TypeError, ValueError):
|
|
return {"output": text}
|
|
|
|
def _safe(self, operation, function):
|
|
try:
|
|
result, details = function()
|
|
return Envelope(True, self.provider, operation, result, details).to_dict()
|
|
except TrackerError as error:
|
|
if error.provider is None:
|
|
error.provider = self.provider
|
|
if error.operation is None:
|
|
error.operation = operation
|
|
return Envelope(False, self.provider, operation, error=error.to_dict()).to_dict()
|
|
except (ValueError, TypeError) as error:
|
|
failure = TrackerError("invalid_input", str(error), provider=self.provider, operation=operation)
|
|
return Envelope(False, self.provider, operation, error=failure.to_dict()).to_dict()
|
|
|
|
def _call_json(self, operation, command, *, kind=None, mutating=False, uncertain=False):
|
|
data = self._json(operation, command, mutating=mutating, uncertain=uncertain)
|
|
if kind:
|
|
if isinstance(data, list):
|
|
data = [self.adapter.normalize(item, kind) for item in data]
|
|
elif isinstance(data, dict) and isinstance(data.get("items"), list):
|
|
data = {**data, "items": [self.adapter.normalize(item, kind) for item in data["items"]]}
|
|
else:
|
|
data = self.adapter.normalize(data, kind)
|
|
return data
|
|
|
|
@staticmethod
|
|
def _as_list(value):
|
|
if isinstance(value, list):
|
|
return value
|
|
if isinstance(value, dict):
|
|
for key in ("items", "labels", "data", "results"):
|
|
if isinstance(value.get(key), list):
|
|
return value[key]
|
|
return []
|
|
|
|
@staticmethod
|
|
def _label_names(labels):
|
|
return {item.get("name") if isinstance(item, dict) else item for item in labels}
|
|
|
|
def _ensure_label_impl(self, name, color="ededed", description=None):
|
|
labels = self._call_json("label.ensure", self.adapter.command("label.list"))
|
|
found = next((label for label in self._as_list(labels) if (label.get("name") if isinstance(label, dict) else label) == name), None)
|
|
if found is not None:
|
|
return found, {"created": False}
|
|
created = self._call_json(
|
|
"label.ensure",
|
|
self.adapter.command("label.create", name=name, color=color, description=description),
|
|
mutating=True,
|
|
)
|
|
return created or {"name": name, "color": color, "description": description}, {"created": True}
|
|
|
|
def ensure_label(self, name, *, color="ededed", description=None):
|
|
return self._safe("label.ensure", lambda: self._ensure_label_impl(name, color, description))
|
|
|
|
def _create(self, kind, title, body="", labels=(), assignees=(), head=None, base=None):
|
|
for label in labels:
|
|
self._ensure_label_impl(label)
|
|
command_args = {"title": title, "body": body, "labels": list(labels), "assignees": list(assignees)}
|
|
if kind == "pr":
|
|
command_args.update(head=head, base=base)
|
|
data = self._call_json(
|
|
f"{kind}.create",
|
|
self.adapter.command(f"{kind}.create", **command_args),
|
|
kind=kind,
|
|
mutating=True,
|
|
)
|
|
return data, {}
|
|
|
|
def create_issue(self, title, *, body="", labels=(), assignees=()):
|
|
return self._safe("issue.create", lambda: self._create("issue", title, body, labels, assignees))
|
|
|
|
def create_pr(self, title, *, body="", head=None, base=None, labels=(), assignees=()):
|
|
return self._safe("pr.create", lambda: self._create("pr", title, body, labels, assignees, head, base))
|
|
|
|
def _get(self, kind, number, comments=True):
|
|
ref = ResourceRef(kind, self._number(number))
|
|
data = self._call_json(f"{kind}.get", self.adapter.command(f"{kind}.get", number=ref.number, comments=comments), kind=kind)
|
|
return data, {}
|
|
|
|
def get_issue(self, number, *, comments=True):
|
|
return self._safe("issue.get", lambda: self._get("issue", number, comments))
|
|
|
|
def get_pr(self, number, *, comments=True, diff=False):
|
|
def operation():
|
|
data, details = self._get("pr", number, comments)
|
|
if diff:
|
|
if not isinstance(data, dict):
|
|
data = {"resource": data}
|
|
data["diff"] = self._run("pr.diff", self.adapter.command("pr.diff", number=self._number(number))).stdout
|
|
details["included_diff"] = True
|
|
return data, details
|
|
return self._safe("pr.get", operation)
|
|
|
|
def _list(self, kind, state="open", labels=(), limit=100):
|
|
command_args = {"state": state, "limit": limit}
|
|
if kind == "issue":
|
|
command_args["labels"] = list(labels)
|
|
data = self._call_json(f"{kind}.list", self.adapter.command(f"{kind}.list", **command_args), kind=kind)
|
|
return data, {}
|
|
|
|
def list_issues(self, *, state="open", labels=(), limit=100):
|
|
return self._safe("issue.list", lambda: self._list("issue", state, labels, limit))
|
|
|
|
def list_prs(self, *, state="open", limit=100, external_only=False):
|
|
def operation():
|
|
data, details = self._list("pr", state, (), limit)
|
|
if not external_only:
|
|
return data, details
|
|
resources = self._as_list(data)
|
|
external = []
|
|
for resource in resources:
|
|
association = resource.get("author_association") if isinstance(resource, dict) else None
|
|
if association is None:
|
|
raise TrackerError(
|
|
"unsupported_capability",
|
|
f"{self.provider} did not provide author membership metadata",
|
|
provider=self.provider,
|
|
operation="pr.list",
|
|
details={"capability": "author_membership"},
|
|
)
|
|
values = self.external_associations or {"owner", "member", "collaborator"}
|
|
if str(association).lower() not in {str(value).lower() for value in values}:
|
|
external.append(resource)
|
|
return external, {**details, "external_only": True}
|
|
return self._safe("pr.list", operation)
|
|
|
|
def _edit(self, kind, number, **kwargs):
|
|
data = self._call_json(f"{kind}.edit", self.adapter.command(f"{kind}.edit", number=self._number(number), **kwargs), kind=kind, mutating=True)
|
|
return data, {}
|
|
|
|
def edit_issue(self, number, *, title=None, body=None):
|
|
return self._safe("issue.edit", lambda: self._edit("issue", number, title=title, body=body))
|
|
|
|
def edit_pr(self, number, *, title=None, body=None):
|
|
return self._safe("pr.edit", lambda: self._edit("pr", number, title=title, body=body))
|
|
|
|
def comment(self, kind, number, body):
|
|
def operation():
|
|
result = self._call_json(f"{kind}.comment", self.adapter.command(f"{kind}.comment", number=self._number(number), body=body), mutating=True, uncertain=True)
|
|
return result, {}
|
|
return self._safe(f"{kind}.comment", operation)
|
|
|
|
def add_label(self, kind, number, name, *, color="ededed", description=None):
|
|
def operation():
|
|
_, ensure_details = self._ensure_label_impl(name, color, description)
|
|
result = self._call_json(
|
|
f"{kind}.label.add",
|
|
self.adapter.command(f"{kind}.edit", number=self._number(number), add_label=name),
|
|
kind=kind,
|
|
mutating=True,
|
|
)
|
|
return result, {"label": name, "ensured": ensure_details}
|
|
return self._safe(f"{kind}.label.add", operation)
|
|
|
|
def remove_label(self, kind, number, name):
|
|
return self._safe(
|
|
f"{kind}.label.remove",
|
|
lambda: (self._edit("issue" if kind == "issue" else "pr", number, remove_label=name)[0], {}),
|
|
)
|
|
|
|
def assign(self, kind, number, user):
|
|
return self._safe(
|
|
f"{kind}.assign",
|
|
lambda: (self._edit(kind, number, assignee=user)[0], {"assignee": user}),
|
|
)
|
|
|
|
def close(self, kind, number, *, explanation=None):
|
|
def operation():
|
|
steps = []
|
|
if explanation:
|
|
self._call_json(f"{kind}.comment", self.adapter.command(f"{kind}.comment", number=self._number(number), body=explanation), mutating=True, uncertain=True)
|
|
steps.append("comment")
|
|
self._call_json(f"{kind}.close", self.adapter.command(f"{kind}.close", number=self._number(number)), kind=kind, mutating=True)
|
|
steps.append("close")
|
|
return {"number": self._number(number), "closed": True}, {"completed": steps}
|
|
return self._safe(f"{kind}.close", operation)
|
|
|
|
def diff(self, number):
|
|
return self._safe("pr.diff", lambda: ({"diff": self._run("pr.diff", self.adapter.command("pr.diff", number=self._number(number))).stdout}, {}))
|
|
|
|
def resolve_reference(self, number):
|
|
"""Resolve a shared issue/PR number explicitly; GitLab keeps its spaces separate."""
|
|
def operation():
|
|
matches = []
|
|
for kind in ("issue", "pr"):
|
|
result = self._run_runner(self.adapter.command(f"{kind}.get", number=self._number(number), comments=False))
|
|
if result.returncode == 0:
|
|
matches.append(self.adapter.normalize(self.adapter.json_value(result.stdout), kind))
|
|
if len(matches) != 1:
|
|
code = "ambiguous_reference" if len(matches) > 1 else "not_found"
|
|
raise TrackerError(code, f"reference #{number} did not resolve to exactly one resource", provider=self.provider, details={"matches": matches})
|
|
return matches[0], {"matches": [matches[0]["kind"]]}
|
|
return self._safe("reference.resolve", operation)
|
|
|
|
def create_map(self, title, *, body="", labels=()):
|
|
map_labels = tuple(dict.fromkeys(["wayfinder:map", *labels]))
|
|
return self._safe("map.create", lambda: self._create("issue", title, body, map_labels, ()) )
|
|
|
|
@staticmethod
|
|
def _result_number(result):
|
|
if isinstance(result, dict):
|
|
if result.get("number"):
|
|
return result["number"]
|
|
match = re.search(r"/(?:issues|pulls)/(\d+)", str(result.get("output", "")))
|
|
if match:
|
|
try:
|
|
return int(match.group(1))
|
|
except (TypeError, ValueError):
|
|
return None
|
|
return None
|
|
|
|
def _native_child_link(self, map_number, child_number):
|
|
if self.provider == "github":
|
|
command = ["gh", "api", "--method", "POST", f"repos/{{owner}}/{{repo}}/issues/{map_number}/sub_issues", "-F", f"sub_issue_id={child_number}"]
|
|
elif self.provider == "gitea":
|
|
command = ["tea", "api", "--method", "POST", f"/repos/{{owner}}/{{repo}}/issues/{map_number}/sub-issues", "-F", f"child_issue_id={child_number}"]
|
|
else:
|
|
return False
|
|
try:
|
|
self._run("child.link", command, mutating=True)
|
|
except TrackerError:
|
|
return False
|
|
return True
|
|
|
|
def _append_map_child(self, map_number, child_number, title):
|
|
if child_number is None:
|
|
return False
|
|
try:
|
|
map_data, _ = self._get("issue", map_number)
|
|
body = map_data.get("body", "") if isinstance(map_data, dict) else ""
|
|
line = f"- [ ] #{child_number} {title}"
|
|
if line not in body:
|
|
body = f"{body.rstrip()}\n\n{line}".lstrip()
|
|
self._edit("issue", map_number, body=body)
|
|
return True
|
|
except TrackerError:
|
|
return False
|
|
|
|
def create_child(self, map_number, title, *, wayfinder_type="task", body="", labels=()):
|
|
def operation():
|
|
map_number_value = self._number(map_number)
|
|
child_body = f"Part of #{map_number_value}\n\n{body}".rstrip()
|
|
child_labels = tuple(dict.fromkeys([f"wayfinder:{wayfinder_type}", *labels]))
|
|
result, details = self._create("issue", title, child_body, child_labels, ())
|
|
child_number = self._result_number(result)
|
|
native = self._native_child_link(map_number_value, child_number)
|
|
fallback = self._append_map_child(map_number_value, child_number, title)
|
|
details.update({"relationship": "native" if native else "fallback_task_list", "map_updated": fallback})
|
|
return result, details
|
|
return self._safe("child.create", operation)
|
|
|
|
def capabilities(self):
|
|
native = {
|
|
"child_relationships": self.provider == "github",
|
|
"blocking_dependencies": self.provider in {"github", "gitea"},
|
|
"diff": True,
|
|
"author_membership": self.provider == "github",
|
|
}
|
|
return Envelope(True, self.provider, "capabilities", native).to_dict()
|
|
|
|
def list_external_prs(self, *, state="open", limit=100):
|
|
return self.list_prs(state=state, limit=limit, external_only=True)
|
|
|
|
def _special(self, operation, action, child, blocker):
|
|
if self.provider == "github":
|
|
command = ["gh", "api", "--method", "POST", f"repos/{{owner}}/{{repo}}/issues/{child}/dependencies/blocked_by", "-F", f"issue_id={blocker}"]
|
|
elif self.provider == "gitlab":
|
|
command = ["glab", "issue", "note", str(child), "--message", f"/blocked_by #{blocker}"]
|
|
else:
|
|
command = ["tea", "api", "--method", "POST", f"/repos/{{owner}}/{{repo}}/issues/{child}/dependencies", "-F", f"index={blocker}"]
|
|
return self._json(operation, command, mutating=True)
|
|
|
|
def add_dependency(self, child, blocker):
|
|
return self._safe("dependency.add", lambda: (self._special("dependency.add", "add", self._number(child), self._number(blocker)), {"fallback": self.provider == "gitlab"}))
|
|
|
|
@staticmethod
|
|
def _refs(body):
|
|
numbers = []
|
|
for value in re.findall(r"(?:Part of|\[[ xX]\].*?)?\s*#(\d+)", body or ""):
|
|
try:
|
|
numbers.append(int(value))
|
|
except (TypeError, ValueError):
|
|
continue
|
|
return numbers
|
|
|
|
def frontier(self, map_number):
|
|
def operation():
|
|
map_data, _ = self._get("issue", self._number(map_number))
|
|
body = map_data.get("body", "") if isinstance(map_data, dict) else ""
|
|
candidates = []
|
|
map_order = self._refs(body)
|
|
order = {str(number): index for index, number in enumerate(map_order)}
|
|
for number in map_order:
|
|
child_value = self._get("issue", self._number(number))[0]
|
|
if not isinstance(child_value, dict):
|
|
continue
|
|
child = child_value
|
|
if child.get("state") == "open" and not child.get("assignees") and not re.search(r"Blocked by:\s*#", child.get("body", ""), re.I):
|
|
candidates.append(child)
|
|
candidates.sort(key=lambda item: order.get(str(item.get("number", "")), 999999))
|
|
return candidates, {"map": self._number(map_number), "deterministic": True}
|
|
return self._safe("frontier.query", operation)
|
|
|
|
def _current_user(self):
|
|
if self.provider == "github":
|
|
result = self._json("auth.current_user", ["gh", "api", "user", "--jq", ".login"])
|
|
elif self.provider == "gitlab":
|
|
result = self._json("auth.current_user", ["glab", "api", "user"])
|
|
else:
|
|
result = self._json("auth.current_user", ["tea", "api", "/user"])
|
|
if isinstance(result, str):
|
|
return result
|
|
if isinstance(result, dict):
|
|
return result.get("login", result.get("username", result.get("name", result.get("output"))))
|
|
return None
|
|
|
|
def claim(self, kind, number, *, user=None):
|
|
def operation():
|
|
owner = user or self._current_user()
|
|
if not owner:
|
|
raise TrackerError("current_user_unavailable", "provider did not return the current user", provider=self.provider)
|
|
result = self._edit(kind, number, assignee=owner)[0]
|
|
return result, {"assignee": owner}
|
|
return self._safe(f"{kind}.claim", operation)
|
|
|
|
def resolve(self, kind, number, answer, *, map_number=None):
|
|
def operation():
|
|
completed = []
|
|
try:
|
|
self._call_json(f"{kind}.comment", self.adapter.command(f"{kind}.comment", number=self._number(number), body=answer), mutating=True, uncertain=True)
|
|
completed.append("comment")
|
|
self._call_json(f"{kind}.close", self.adapter.command(f"{kind}.close", number=self._number(number)), mutating=True)
|
|
completed.append("close")
|
|
if map_number:
|
|
pointer = f"Resolved #{self._number(number)}: {answer.splitlines()[0][:200]}"
|
|
self._call_json("map.pointer", self.adapter.command("issue.comment", number=self._number(map_number), body=pointer), mutating=True, uncertain=True)
|
|
completed.append("map_pointer")
|
|
except TrackerError as error:
|
|
raise TrackerError("partial_failure", str(error), retryable=False, provider=self.provider, details={"completed": completed, "recovery": "repeat only incomplete steps"})
|
|
return {"number": self._number(number), "resolved": True}, {"completed": completed}
|
|
return self._safe(f"{kind}.resolve", operation)
|