codeant-ai-for-open-source[bot] commented on code in PR #44439: URL: https://github.com/apache/superset/pull/44439#discussion_r4052314354
########## automation/orchestrator/__main__.py: ########## @@ -0,0 +1,128 @@ +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You under the Apache License, Version 2.0 +# (the "License"); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""Command line entry point: ``python -m automation.orchestrator <command>``.""" + +from __future__ import annotations + +import argparse +import logging +import os +import sys +from datetime import datetime, timedelta, timezone + +from . import dispatch as ops +from .api import Devin, GitHub + +REPO = os.environ.get("REPO", "moliyadhaval/superset") +ORG_ID = os.environ.get("DEVIN_ORG_ID", "org-eef03b923043402190c037ffa8141840") +DASHBOARD_ISSUE = int(os.environ.get("DASHBOARD_ISSUE", "18")) + + +def _require(*names: str) -> None: + missing = [n for n in names if not os.environ.get(n)] + if missing: + sys.exit(f"missing environment variable(s): {', '.join(missing)}") + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(prog="automation.orchestrator") + parser.add_argument("-v", "--verbose", action="store_true") + sub = parser.add_subparsers(dest="cmd", required=True) + + p = sub.add_parser("dispatch", help="dispatch one fix session per eligible issue") + p.add_argument( + "--prompt-file", required=True, help="file holding the canonical fix prompt" + ) + p.add_argument("--cap", type=int, default=30) + p.add_argument("--dry-run", action="store_true") + + p = sub.add_parser( + "watch-session", help="poll a session; fail unless it produced a PR" + ) + p.add_argument("session_id") + p.add_argument("--issue", type=int, required=True) + p.add_argument("--timeout-minutes", type=int, default=90) + p.add_argument("--poll-seconds", type=float, default=60) + p.add_argument( + "--keep-branch", + action="store_true", + help="do not delete the orphan branch on failure", + ) + + p = sub.add_parser( + "stale-branches", help="list (or delete) fix branches without an open/merged PR" + ) + p.add_argument("--days", type=int, default=int(os.environ.get("STALE_DAYS", "3"))) + p.add_argument("--delete", action="store_true") + + args = parser.parse_args(argv) + logging.basicConfig( + level=logging.DEBUG if args.verbose else logging.INFO, + format="%(asctime)s %(levelname)s %(name)s: %(message)s", + stream=sys.stderr, + ) + gh = GitHub(REPO) + devin = Devin(ORG_ID) + today = datetime.now(timezone.utc).date().isoformat() + + if args.cmd == "dispatch": + _require("GH_TOKEN", *([] if args.dry_run else ["DEVIN_API_KEY"])) + with open(args.prompt_file, encoding="utf-8") as fh: + prompt = fh.read() + outcomes = ops.dispatch( + gh, devin, fix_prompt=prompt, cap=args.cap, dry_run=args.dry_run + ) + table = ops.dashboard_table(outcomes) + print(table) + if not args.dry_run: + ops.ensure_comment( + gh, + DASHBOARD_ISSUE, + f"dispatch:{today}", + f"### Dispatch run {today}\n\n{table}", + ) + return 1 if any(o.status == "failed" for o in outcomes) else 0 + + if args.cmd == "watch-session": + _require("GH_TOKEN", "DEVIN_API_KEY") + ok, detail = False, "watch aborted" + try: + ok, detail = ops.watch_session( + devin, + args.session_id, + timeout=timedelta(minutes=args.timeout_minutes), + poll=args.poll_seconds, + ) + finally: + verdict = "โ PR opened" if ok else "โ Failed" + line = f"[Issue #{args.issue}] {verdict} - {detail}" + logging.getLogger("automation").log( + logging.INFO if ok else logging.ERROR, line + ) + ops.ensure_comment(gh, DASHBOARD_ISSUE, f"session:{args.session_id}", line) + if not ok and not args.keep_branch: + ops.cleanup_branch(gh, args.issue) Review Comment: **Suggestion:** An exception from `ensure_comment` replaces the polling result and prevents `cleanup_branch`, leaving failed sessions' orphan branches behind. **Assessment:** ๐ `Major` ยท ๐ `Occurrence: Sometimes` ยท ๐ท๏ธ `Missing cleanup` [](https://docs.codeant.ai/cli/resolve-pr-comments-skill) [](https://app.codeant.ai/fix-in-ide?tool=cursor&prompt_id=ac2edd86564f49c6901ae2f550f16d0a&service=github&base_url=https%3A%2F%2Fgithub.com&org=apache&repo=apache%2Fsuperset) [](https://app.codeant.ai/fix-in-ide?tool=vscode-claude&prompt_id=ac2edd86564f49c6901ae2f550f16d0a&service=github&base_url=https%3A%2F%2Fgithub.com&org=apache&repo=apache%2Fsuperset) <details> <summary><b>Prompt for AI Agent ๐ค </b></summary> ```mdx This is a comment left during a code review. **Path:** automation/orchestrator/__main__.py **Line:** 108:116 **Comment:** *Missing Cleanup: An exception from `ensure_comment` replaces the polling result and prevents `cleanup_branch`, leaving failed sessions' orphan branches behind. Validate the correctness of the flagged issue. If correct, How can I resolve this? If you propose a fix, implement it and please make it concise. Once fix is implemented, also check other comments on the same PR, and ask user if the user wants to fix the rest of the comments as well. if said yes, then fetch all the comments validate the correctness and implement a minimal fix ``` </details> <a href='https://app.codeant.ai/feedback?pr_url=https%3A%2F%2Fgithub.com%2Fapache%2Fsuperset%2Fpull%2F44439&comment_hash=5846e7ed93adabd06f9e7e7592acea958bfa2b5922f210f2a0fb27ebe18b8792&reaction=like'>๐</a> | <a href='https://app.codeant.ai/feedback?pr_url=https%3A%2F%2Fgithub.com%2Fapache%2Fsuperset%2Fpull%2F44439&comment_hash=5846e7ed93adabd06f9e7e7592acea958bfa2b5922f210f2a0fb27ebe18b8792&reaction=dislike'>๐</a> ########## automation/orchestrator/api.py: ########## @@ -0,0 +1,342 @@ +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You under the Apache License, Version 2.0 +# (the "License"); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""HTTP clients for GitHub and the Devin API with a shared retry policy. + +Retry contract +-------------- +* ``429`` and rate-limited ``403`` (primary/secondary/abuse) are retried for + every method: the server rejected the request, nothing happened. +* ``5xx`` and transport errors (timeouts, resets) are retried for idempotent + methods (GET/HEAD/PUT/DELETE). For POST/PATCH they raise + :class:`AmbiguousWriteError` so the caller probes for the resource before + writing again -- this is what keeps retries idempotent. +* Other ``4xx`` raise :class:`PermanentError` immediately. +* Waits use exponential backoff with full jitter (tenacity), capped per + attempt and by a total budget, and honour ``Retry-After`` / + ``X-RateLimit-Reset`` when present. + +Secrets are only ever placed in request headers; they are never logged. +""" + +from __future__ import annotations + +import email.utils +import logging +import os +import time +from collections.abc import Callable, Iterator +from dataclasses import dataclass, field +from typing import Any + +import requests +from tenacity import ( + retry_if_exception_type, + RetryCallState, + Retrying, + stop_after_attempt, + stop_after_delay, + wait_exponential_jitter, +) + +log = logging.getLogger("automation.api") + +IDEMPOTENT_METHODS = frozenset({"GET", "PUT", "DELETE"}) +RATE_LIMIT_MARKERS = ("rate limit", "abuse detection", "too many requests") + + +class ApiError(Exception): + """Base class for API failures.""" + + def __init__(self, message: str, status: int | None = None) -> None: + super().__init__(message) + self.status = status + + +class TransientError(ApiError): + """Safe to retry: the request was rejected or the read failed.""" + + def __init__( + self, message: str, status: int | None = None, retry_after: float | None = None + ) -> None: + super().__init__(message, status) + self.retry_after = retry_after + + +class AmbiguousWriteError(ApiError): + """A write may or may not have landed; probe before repeating it.""" + + +class PermanentError(ApiError): + """Client error that will not succeed on retry (401/404/422...).""" + + +@dataclass(frozen=True) +class RetryPolicy: + max_attempts: int = int(os.environ.get("RETRY_MAX_ATTEMPTS", "6")) + initial: float = float(os.environ.get("RETRY_BASE_SECONDS", "2")) + max_sleep: float = float(os.environ.get("RETRY_MAX_SLEEP", "120")) + budget_seconds: float = float(os.environ.get("RETRY_BUDGET_SECONDS", "900")) + timeout: float = float(os.environ.get("HTTP_TIMEOUT", "60")) + + +def _wait(policy: RetryPolicy) -> Callable[[RetryCallState], float]: + """Exponential backoff with full jitter, overridden by a server hint.""" + jitter = wait_exponential_jitter(initial=policy.initial, max=policy.max_sleep) + + def wait(state: RetryCallState) -> float: + exc = state.outcome.exception() if state.outcome else None + if isinstance(exc, TransientError) and exc.retry_after is not None: + return min(max(exc.retry_after, 0.0), policy.max_sleep) + return float(jitter(state)) + + return wait + + +def _retry_after_seconds(headers: Any) -> float | None: + """Parse Retry-After (seconds or HTTP-date) or X-RateLimit-Reset (epoch).""" + if raw := headers.get("Retry-After"): + try: + return float(raw) + except ValueError: + parsed = email.utils.parsedate_to_datetime(raw) + return max(parsed.timestamp() - time.time(), 0.0) Review Comment: **Suggestion:** A malformed nonnumeric `Retry-After` header makes date parsing raise an uncaught `ValueError`, aborting an otherwise retryable request. **Assessment:** ๐ `Major` ยท ๐ `Occurrence: Rarely` ยท ๐ท๏ธ `Possible bug` [](https://docs.codeant.ai/cli/resolve-pr-comments-skill) [](https://app.codeant.ai/fix-in-ide?tool=cursor&prompt_id=f720560ca9814df08338fbdc4763bcd3&service=github&base_url=https%3A%2F%2Fgithub.com&org=apache&repo=apache%2Fsuperset) [](https://app.codeant.ai/fix-in-ide?tool=vscode-claude&prompt_id=f720560ca9814df08338fbdc4763bcd3&service=github&base_url=https%3A%2F%2Fgithub.com&org=apache&repo=apache%2Fsuperset) <details> <summary><b>Prompt for AI Agent ๐ค </b></summary> ```mdx This is a comment left during a code review. **Path:** automation/orchestrator/api.py **Line:** 109:114 **Comment:** *Possible Bug: A malformed nonnumeric `Retry-After` header makes date parsing raise an uncaught `ValueError`, aborting an otherwise retryable request. Validate the correctness of the flagged issue. If correct, How can I resolve this? If you propose a fix, implement it and please make it concise. Once fix is implemented, also check other comments on the same PR, and ask user if the user wants to fix the rest of the comments as well. if said yes, then fetch all the comments validate the correctness and implement a minimal fix ``` </details> <a href='https://app.codeant.ai/feedback?pr_url=https%3A%2F%2Fgithub.com%2Fapache%2Fsuperset%2Fpull%2F44439&comment_hash=7c7747d05624c6553271a93a1bd8bea9cbcd93c83b70ec0b0934468d441e4a7a&reaction=like'>๐</a> | <a href='https://app.codeant.ai/feedback?pr_url=https%3A%2F%2Fgithub.com%2Fapache%2Fsuperset%2Fpull%2F44439&comment_hash=7c7747d05624c6553271a93a1bd8bea9cbcd93c83b70ec0b0934468d441e4a7a&reaction=dislike'>๐</a> ########## automation/orchestrator/dispatch.py: ########## @@ -0,0 +1,290 @@ +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You under the Apache License, Version 2.0 +# (the "License"); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""Dispatch one Devin fix session per open nightly-scan issue, safely. + +Every step is idempotent: a session is looked up by title before it is +created, a dashboard comment is looked up by a hidden marker before it is +posted, and a failure for one issue is recorded and the loop moves on. +""" + +from __future__ import annotations + +import logging +import re +import time +from collections.abc import Callable, Iterable +from dataclasses import dataclass +from datetime import datetime, timedelta, timezone +from typing import Any + +from .api import AmbiguousWriteError, ApiError, Devin, GitHub, session_url + +log = logging.getLogger("automation.dispatch") + +BRANCH_PREFIX = "devin/nightly-fix-" +BRANCH_ISSUE = re.compile(r"devin/nightly-fix-(?:issue-)?(\d+)") +BODY_REF = re.compile(r"\b(?:Fixes|Refs|Closes)\s+#(\d+)", re.I) +DEAD_STATUSES = frozenset({"error", "terminated", "cancelled", "canceled"}) +FINISHED_STATUSES = frozenset({"finished", "completed", "blocked", "suspended"}) + + +@dataclass +class Outcome: + issue: int + status: str # dispatched | skipped | failed + detail: str = "" + + +def marker(key: str) -> str: + return f"<!-- devin:{key} -->" + + +def attempted_issues( + prs: Iterable[dict[str, Any]], sessions: Iterable[dict[str, Any]] +) -> set[int]: + """Issue numbers that already have a PR, a fix branch, or a live session.""" + seen: set[int] = set() + for pr in prs: + seen.update(int(n) for n in BODY_REF.findall(pr.get("body") or "")) + if m := BRANCH_ISSUE.search(pr["head"]["ref"]): + seen.add(int(m.group(1))) + for s in sessions: + if str(s.get("status") or s.get("status_enum") or "").lower() in DEAD_STATUSES: + continue + if m := re.search(r"#(\d+)", s.get("title") or ""): + seen.add(int(m.group(1))) + return seen + + +def ensure_session(devin: Devin, payload: dict[str, Any], tag: str) -> dict[str, Any]: + """Create a session unless one with the same title is already alive. + + An ambiguous failure (5xx/timeout on the POST) is resolved by probing + again, so a request that landed is never duplicated. + """ + title = payload["title"] + for attempt in (1, 2): + for s in devin.list_sessions(tag): + if ( + s.get("title") == title + and str(s.get("status") or s.get("status_enum") or "").lower() + not in DEAD_STATUSES + ): + log.info("%s already has session %s", title, session_url(s)) + return s + try: + return devin.create_session(payload) Review Comment: **Suggestion:** Concurrent dispatchers can both observe no matching session and create duplicate sessions because the lookup and creation are separate requests. **Assessment:** ๐ `Major` ยท ๐ `Occurrence: Rarely` ยท ๐ท๏ธ `Race condition` [](https://docs.codeant.ai/cli/resolve-pr-comments-skill) [](https://app.codeant.ai/fix-in-ide?tool=cursor&prompt_id=31699b41175040c6bdd05473dc3a3f19&service=github&base_url=https%3A%2F%2Fgithub.com&org=apache&repo=apache%2Fsuperset) [](https://app.codeant.ai/fix-in-ide?tool=vscode-claude&prompt_id=31699b41175040c6bdd05473dc3a3f19&service=github&base_url=https%3A%2F%2Fgithub.com&org=apache&repo=apache%2Fsuperset) <details> <summary><b>Prompt for AI Agent ๐ค </b></summary> ```mdx This is a comment left during a code review. **Path:** automation/orchestrator/dispatch.py **Line:** 79:88 **Comment:** *Race Condition: Concurrent dispatchers can both observe no matching session and create duplicate sessions because the lookup and creation are separate requests. Validate the correctness of the flagged issue. If correct, How can I resolve this? If you propose a fix, implement it and please make it concise. Once fix is implemented, also check other comments on the same PR, and ask user if the user wants to fix the rest of the comments as well. if said yes, then fetch all the comments validate the correctness and implement a minimal fix ``` </details> <a href='https://app.codeant.ai/feedback?pr_url=https%3A%2F%2Fgithub.com%2Fapache%2Fsuperset%2Fpull%2F44439&comment_hash=9b7625a45277bd852385448382c97745088822003e6d61a2cb3db422bcdd06bb&reaction=like'>๐</a> | <a href='https://app.codeant.ai/feedback?pr_url=https%3A%2F%2Fgithub.com%2Fapache%2Fsuperset%2Fpull%2F44439&comment_hash=9b7625a45277bd852385448382c97745088822003e6d61a2cb3db422bcdd06bb&reaction=dislike'>๐</a> ########## automation/orchestrator/dispatch.py: ########## @@ -0,0 +1,290 @@ +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You under the Apache License, Version 2.0 +# (the "License"); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""Dispatch one Devin fix session per open nightly-scan issue, safely. + +Every step is idempotent: a session is looked up by title before it is +created, a dashboard comment is looked up by a hidden marker before it is +posted, and a failure for one issue is recorded and the loop moves on. +""" + +from __future__ import annotations + +import logging +import re +import time +from collections.abc import Callable, Iterable +from dataclasses import dataclass +from datetime import datetime, timedelta, timezone +from typing import Any + +from .api import AmbiguousWriteError, ApiError, Devin, GitHub, session_url + +log = logging.getLogger("automation.dispatch") + +BRANCH_PREFIX = "devin/nightly-fix-" +BRANCH_ISSUE = re.compile(r"devin/nightly-fix-(?:issue-)?(\d+)") +BODY_REF = re.compile(r"\b(?:Fixes|Refs|Closes)\s+#(\d+)", re.I) +DEAD_STATUSES = frozenset({"error", "terminated", "cancelled", "canceled"}) +FINISHED_STATUSES = frozenset({"finished", "completed", "blocked", "suspended"}) + + +@dataclass +class Outcome: + issue: int + status: str # dispatched | skipped | failed + detail: str = "" + + +def marker(key: str) -> str: + return f"<!-- devin:{key} -->" + + +def attempted_issues( + prs: Iterable[dict[str, Any]], sessions: Iterable[dict[str, Any]] +) -> set[int]: + """Issue numbers that already have a PR, a fix branch, or a live session.""" + seen: set[int] = set() + for pr in prs: + seen.update(int(n) for n in BODY_REF.findall(pr.get("body") or "")) + if m := BRANCH_ISSUE.search(pr["head"]["ref"]): + seen.add(int(m.group(1))) + for s in sessions: + if str(s.get("status") or s.get("status_enum") or "").lower() in DEAD_STATUSES: + continue + if m := re.search(r"#(\d+)", s.get("title") or ""): + seen.add(int(m.group(1))) + return seen + + +def ensure_session(devin: Devin, payload: dict[str, Any], tag: str) -> dict[str, Any]: + """Create a session unless one with the same title is already alive. + + An ambiguous failure (5xx/timeout on the POST) is resolved by probing + again, so a request that landed is never duplicated. + """ + title = payload["title"] + for attempt in (1, 2): + for s in devin.list_sessions(tag): + if ( + s.get("title") == title + and str(s.get("status") or s.get("status_enum") or "").lower() + not in DEAD_STATUSES + ): + log.info("%s already has session %s", title, session_url(s)) + return s + try: + return devin.create_session(payload) + except AmbiguousWriteError as exc: + log.warning("%s: %s. Re-probing (attempt %d/2)", title, exc, attempt) + time.sleep(devin.policy.initial) + raise ApiError(f"{title}: session creation ambiguous after re-probe") + + +def ensure_comment(gh: GitHub, issue: int, key: str, body: str) -> dict[str, Any]: + """Post ``body`` on ``issue`` once per ``key``; update it if it exists.""" + text = f"{body}\n\n{marker(key)}" + for c in gh.comments(issue): + if marker(key) in (c.get("body") or ""): + return gh.update_comment(int(c["id"]), text) if c["body"] != text else c + try: + return gh.create_comment(issue, text) + except AmbiguousWriteError: + for c in gh.comments(issue): + if marker(key) in (c.get("body") or ""): + return c + raise + + +def watch_session( + devin: Devin, + session_id: str, + *, + timeout: timedelta, + poll: float = 60.0, + sleep: Callable[[float], None] = time.sleep, +) -> tuple[bool, str]: + """Poll a session until it finishes, with an orchestrator-side circuit breaker. + + Success requires *both* a finished status and a pull request on the + session; a "completed" session with no pull request is a failure. The v3 + API exposes ``pull_requests: [{pr_url, pr_state}]`` (older payloads used a + single ``pull_request`` object); both shapes are accepted. + """ + deadline = time.monotonic() + timeout.total_seconds() + attempt = 0 + while True: + attempt += 1 + s = devin.get_session(session_id) + status = str(s.get("status") or s.get("status_enum") or "").lower() + pr = session_pr(s) + log.info( + "[Session %s] poll %d: status=%s pr=%s", + session_id, + attempt, + status, + bool(pr), + ) + if pr: + return True, pr + if status in DEAD_STATUSES: + return False, f"session {status}" + if status in FINISHED_STATUSES: + return False, f"session {status} but pull_request is null" + if time.monotonic() >= deadline: + return False, f"timeout reached after {timeout} (last status: {status})" + sleep(poll) + + +def session_pr(session: dict[str, Any]) -> str | None: + """URL of the session's pull request, or None when it has not opened one.""" + prs = list(session.get("pull_requests") or []) + if session.get("pull_request"): + prs.append(session["pull_request"]) + for pr in prs: + if url := pr.get("pr_url") or pr.get("url"): + return str(url) + return None + + +def branch_for_issue(gh: GitHub, issue: int) -> str | None: + for b in gh.paginate("branches"): + if (m := BRANCH_ISSUE.match(b["name"])) and int(m.group(1)) == issue: + return str(b["name"]) + return None + + +def cleanup_branch(gh: GitHub, issue: int, dry_run: bool = False) -> str | None: + """Delete the fix branch for ``issue`` unless a PR (any state) uses it.""" + branch = branch_for_issue(gh, issue) + if not branch: + return None + if any(pr["head"]["ref"] == branch for pr in gh.pulls("all")): + log.info("[Issue #%d] keeping %s: referenced by a PR", issue, branch) + return None + if not dry_run: + gh.delete_branch(branch) Review Comment: **Suggestion:** The branch usage check and deletion are separate requests, so a newly opened pull request can be missed and its branch deleted. **Assessment:** ๐ด `Critical` ยท ๐ `Occurrence: Rarely` ยท ๐ท๏ธ `Race condition` [](https://docs.codeant.ai/cli/resolve-pr-comments-skill) [](https://app.codeant.ai/fix-in-ide?tool=cursor&prompt_id=53c3552ae800417ba3cee589bf0897aa&service=github&base_url=https%3A%2F%2Fgithub.com&org=apache&repo=apache%2Fsuperset) [](https://app.codeant.ai/fix-in-ide?tool=vscode-claude&prompt_id=53c3552ae800417ba3cee589bf0897aa&service=github&base_url=https%3A%2F%2Fgithub.com&org=apache&repo=apache%2Fsuperset) <details> <summary><b>Prompt for AI Agent ๐ค </b></summary> ```mdx This is a comment left during a code review. **Path:** automation/orchestrator/dispatch.py **Line:** 173:177 **Comment:** *Race Condition: The branch usage check and deletion are separate requests, so a newly opened pull request can be missed and its branch deleted. Validate the correctness of the flagged issue. If correct, How can I resolve this? If you propose a fix, implement it and please make it concise. Once fix is implemented, also check other comments on the same PR, and ask user if the user wants to fix the rest of the comments as well. if said yes, then fetch all the comments validate the correctness and implement a minimal fix ``` </details> <a href='https://app.codeant.ai/feedback?pr_url=https%3A%2F%2Fgithub.com%2Fapache%2Fsuperset%2Fpull%2F44439&comment_hash=12de1a73052bbbeb18ca9bae79da89322dd15d6f95e2053d00196c9929f0f434&reaction=like'>๐</a> | <a href='https://app.codeant.ai/feedback?pr_url=https%3A%2F%2Fgithub.com%2Fapache%2Fsuperset%2Fpull%2F44439&comment_hash=12de1a73052bbbeb18ca9bae79da89322dd15d6f95e2053d00196c9929f0f434&reaction=dislike'>๐</a> ########## automation/orchestrator/dispatch.py: ########## @@ -0,0 +1,290 @@ +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You under the Apache License, Version 2.0 +# (the "License"); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""Dispatch one Devin fix session per open nightly-scan issue, safely. + +Every step is idempotent: a session is looked up by title before it is +created, a dashboard comment is looked up by a hidden marker before it is +posted, and a failure for one issue is recorded and the loop moves on. +""" + +from __future__ import annotations + +import logging +import re +import time +from collections.abc import Callable, Iterable +from dataclasses import dataclass +from datetime import datetime, timedelta, timezone +from typing import Any + +from .api import AmbiguousWriteError, ApiError, Devin, GitHub, session_url + +log = logging.getLogger("automation.dispatch") + +BRANCH_PREFIX = "devin/nightly-fix-" +BRANCH_ISSUE = re.compile(r"devin/nightly-fix-(?:issue-)?(\d+)") +BODY_REF = re.compile(r"\b(?:Fixes|Refs|Closes)\s+#(\d+)", re.I) +DEAD_STATUSES = frozenset({"error", "terminated", "cancelled", "canceled"}) +FINISHED_STATUSES = frozenset({"finished", "completed", "blocked", "suspended"}) + + +@dataclass +class Outcome: + issue: int + status: str # dispatched | skipped | failed + detail: str = "" + + +def marker(key: str) -> str: + return f"<!-- devin:{key} -->" + + +def attempted_issues( + prs: Iterable[dict[str, Any]], sessions: Iterable[dict[str, Any]] +) -> set[int]: + """Issue numbers that already have a PR, a fix branch, or a live session.""" + seen: set[int] = set() + for pr in prs: + seen.update(int(n) for n in BODY_REF.findall(pr.get("body") or "")) + if m := BRANCH_ISSUE.search(pr["head"]["ref"]): + seen.add(int(m.group(1))) + for s in sessions: + if str(s.get("status") or s.get("status_enum") or "").lower() in DEAD_STATUSES: + continue + if m := re.search(r"#(\d+)", s.get("title") or ""): + seen.add(int(m.group(1))) + return seen + + +def ensure_session(devin: Devin, payload: dict[str, Any], tag: str) -> dict[str, Any]: + """Create a session unless one with the same title is already alive. + + An ambiguous failure (5xx/timeout on the POST) is resolved by probing + again, so a request that landed is never duplicated. + """ + title = payload["title"] + for attempt in (1, 2): + for s in devin.list_sessions(tag): + if ( + s.get("title") == title + and str(s.get("status") or s.get("status_enum") or "").lower() + not in DEAD_STATUSES + ): + log.info("%s already has session %s", title, session_url(s)) + return s + try: + return devin.create_session(payload) + except AmbiguousWriteError as exc: + log.warning("%s: %s. Re-probing (attempt %d/2)", title, exc, attempt) + time.sleep(devin.policy.initial) + raise ApiError(f"{title}: session creation ambiguous after re-probe") + + +def ensure_comment(gh: GitHub, issue: int, key: str, body: str) -> dict[str, Any]: + """Post ``body`` on ``issue`` once per ``key``; update it if it exists.""" + text = f"{body}\n\n{marker(key)}" + for c in gh.comments(issue): + if marker(key) in (c.get("body") or ""): + return gh.update_comment(int(c["id"]), text) if c["body"] != text else c + try: + return gh.create_comment(issue, text) + except AmbiguousWriteError: + for c in gh.comments(issue): + if marker(key) in (c.get("body") or ""): + return c + raise + + +def watch_session( + devin: Devin, + session_id: str, + *, + timeout: timedelta, + poll: float = 60.0, + sleep: Callable[[float], None] = time.sleep, +) -> tuple[bool, str]: + """Poll a session until it finishes, with an orchestrator-side circuit breaker. + + Success requires *both* a finished status and a pull request on the + session; a "completed" session with no pull request is a failure. The v3 + API exposes ``pull_requests: [{pr_url, pr_state}]`` (older payloads used a + single ``pull_request`` object); both shapes are accepted. + """ + deadline = time.monotonic() + timeout.total_seconds() + attempt = 0 + while True: + attempt += 1 + s = devin.get_session(session_id) + status = str(s.get("status") or s.get("status_enum") or "").lower() + pr = session_pr(s) + log.info( + "[Session %s] poll %d: status=%s pr=%s", + session_id, + attempt, + status, + bool(pr), + ) + if pr: + return True, pr + if status in DEAD_STATUSES: + return False, f"session {status}" + if status in FINISHED_STATUSES: + return False, f"session {status} but pull_request is null" + if time.monotonic() >= deadline: + return False, f"timeout reached after {timeout} (last status: {status})" + sleep(poll) + + +def session_pr(session: dict[str, Any]) -> str | None: + """URL of the session's pull request, or None when it has not opened one.""" + prs = list(session.get("pull_requests") or []) + if session.get("pull_request"): + prs.append(session["pull_request"]) + for pr in prs: + if url := pr.get("pr_url") or pr.get("url"): + return str(url) + return None + + +def branch_for_issue(gh: GitHub, issue: int) -> str | None: + for b in gh.paginate("branches"): + if (m := BRANCH_ISSUE.match(b["name"])) and int(m.group(1)) == issue: + return str(b["name"]) + return None + + +def cleanup_branch(gh: GitHub, issue: int, dry_run: bool = False) -> str | None: + """Delete the fix branch for ``issue`` unless a PR (any state) uses it.""" + branch = branch_for_issue(gh, issue) + if not branch: + return None + if any(pr["head"]["ref"] == branch for pr in gh.pulls("all")): + log.info("[Issue #%d] keeping %s: referenced by a PR", issue, branch) + return None + if not dry_run: + gh.delete_branch(branch) + log.info("[Issue #%d] deleted orphaned branch %s", issue, branch) + return branch + + +def stale_branches( + gh: GitHub, older_than: timedelta, *, delete: bool = False +) -> list[tuple[str, str]]: + """Fix branches with no open/merged PR and a last commit older than the cutoff.""" + prs: dict[str, str] = {} + for pr in gh.pulls("all"): + state = "MERGED" if pr.get("merged_at") else str(pr["state"]).upper() + prs[pr["head"]["ref"]] = ( + state if prs.get(pr["head"]["ref"]) != "OPEN" else "OPEN" + ) Review Comment: **Suggestion:** When a merged and later closed pull request share a branch, this assignment can overwrite `MERGED` with `CLOSED`, allowing stale cleanup to delete the branch. **Assessment:** ๐ `Major` ยท ๐ `Occurrence: Rarely` ยท ๐ท๏ธ `Incorrect condition logic` [](https://docs.codeant.ai/cli/resolve-pr-comments-skill) [](https://app.codeant.ai/fix-in-ide?tool=cursor&prompt_id=6ab6f43bf20b468d88db959af3708934&service=github&base_url=https%3A%2F%2Fgithub.com&org=apache&repo=apache%2Fsuperset) [](https://app.codeant.ai/fix-in-ide?tool=vscode-claude&prompt_id=6ab6f43bf20b468d88db959af3708934&service=github&base_url=https%3A%2F%2Fgithub.com&org=apache&repo=apache%2Fsuperset) <details> <summary><b>Prompt for AI Agent ๐ค </b></summary> ```mdx This is a comment left during a code review. **Path:** automation/orchestrator/dispatch.py **Line:** 187:191 **Comment:** *Incorrect Condition Logic: When a merged and later closed pull request share a branch, this assignment can overwrite `MERGED` with `CLOSED`, allowing stale cleanup to delete the branch. Validate the correctness of the flagged issue. If correct, How can I resolve this? If you propose a fix, implement it and please make it concise. Once fix is implemented, also check other comments on the same PR, and ask user if the user wants to fix the rest of the comments as well. if said yes, then fetch all the comments validate the correctness and implement a minimal fix ``` </details> <a href='https://app.codeant.ai/feedback?pr_url=https%3A%2F%2Fgithub.com%2Fapache%2Fsuperset%2Fpull%2F44439&comment_hash=209ea1991e5eb603a933b023be4169462c7b442ccd7f718b63c15002bc0527da&reaction=like'>๐</a> | <a href='https://app.codeant.ai/feedback?pr_url=https%3A%2F%2Fgithub.com%2Fapache%2Fsuperset%2Fpull%2F44439&comment_hash=209ea1991e5eb603a933b023be4169462c7b442ccd7f718b63c15002bc0527da&reaction=dislike'>๐</a> -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
