diff --git a/omnibase_compat b/omnibase_compat deleted file mode 120000 index 2656f55aab..0000000000 --- a/omnibase_compat +++ /dev/null @@ -1 +0,0 @@ -/Volumes/PRO-G40/Code/omni_home/omnibase_compat \ No newline at end of file diff --git a/omnibase_core b/omnibase_core deleted file mode 120000 index 6268c3ae57..0000000000 --- a/omnibase_core +++ /dev/null @@ -1 +0,0 @@ -/Volumes/PRO-G40/Code/omni_home/omnibase_core \ No newline at end of file diff --git a/pyproject.toml b/pyproject.toml index a495677173..f2a35fe2aa 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -51,7 +51,7 @@ Repository = "https://github.com/OmniNode-ai/omnimarket" [project.entry-points."onex.nodes"] node_aislop_sweep = "omnimarket.nodes.node_aislop_sweep" node_authorize = "omnimarket.nodes.node_authorize" -node_merge_sweep = "omnimarket.nodes.node_merge_sweep" +node_merge_sweep_compute = "omnimarket.nodes.node_merge_sweep_compute" node_ledger_orchestrator = "omnimarket.nodes.node_ledger_orchestrator" node_ledger_append_effect = "omnimarket.nodes.node_ledger_append_effect" node_ledger_hash_compute = "omnimarket.nodes.node_ledger_hash_compute" diff --git a/src/omnimarket/nodes/node_merge_sweep/__init__.py b/src/omnimarket/nodes/node_merge_sweep/__init__.py deleted file mode 100644 index 63da0ccca4..0000000000 --- a/src/omnimarket/nodes/node_merge_sweep/__init__.py +++ /dev/null @@ -1,7 +0,0 @@ -"""node_merge_sweep — Org-wide PR sweep for auto-merge and polish.""" - -from omnimarket.nodes.node_merge_sweep.handlers.handler_merge_sweep import ( - NodeMergeSweep, -) - -__all__ = ["NodeMergeSweep"] diff --git a/src/omnimarket/nodes/node_merge_sweep/branch_protection.py b/src/omnimarket/nodes/node_merge_sweep/branch_protection.py deleted file mode 100644 index c69cbc93d5..0000000000 --- a/src/omnimarket/nodes/node_merge_sweep/branch_protection.py +++ /dev/null @@ -1,88 +0,0 @@ -# SPDX-FileCopyrightText: 2025 OmniNode.ai Inc. -# SPDX-License-Identifier: MIT -"""Per-sweep branch protection fetcher for merge-sweep [OMN-9106]. - -Resolves `required_approving_review_count` for a repo's default branch via -``gh api repos///branches/main/protection``. One call per repo -per sweep, cached in-memory for the sweep's lifetime. - -Returned semantics: - - int >= 0 : live value from GitHub - - None : no approving review required (either explicit `null` or missing - ``required_pull_request_reviews`` block), OR fetch failed - -Failed fetches intentionally collapse to ``None`` — merge-sweep then treats the -approval gate as cleared, matching the solo-dev repo configuration that is -the dominant OmniNode case. If that default ever becomes wrong for a repo -(i.e. protection actually requires approval), the gh API call will succeed -and return a non-zero count. -""" - -from __future__ import annotations - -import json -import logging -import subprocess - -_log = logging.getLogger(__name__) - - -class BranchProtectionCache: - """Per-sweep cache of required_approving_review_count per repo. - - One instance per sweep run; discard between runs so stale values don't - persist across sweeps. - """ - - def __init__(self) -> None: - self._cache: dict[str, int | None] = {} - - def required_approving_review_count(self, repo: str) -> int | None: - """Return required approving review count for ``repo``'s main branch. - - Lazily fetches once per repo per instance. Returns None on fetch failure - or when protection does not require approving reviews. - """ - if repo in self._cache: - return self._cache[repo] - count = _fetch_required_approving_review_count(repo) - self._cache[repo] = count - return count - - -def _fetch_required_approving_review_count(repo: str) -> int | None: - """Call gh API once; never raises. Returns None on any failure.""" - try: - result = subprocess.run( - [ - "gh", - "api", - f"repos/{repo}/branches/main/protection", - ], - capture_output=True, - text=True, - timeout=15, - ) - except (subprocess.TimeoutExpired, OSError) as exc: - _log.debug("branch protection fetch failed for %s: %s", repo, exc) - return None - if result.returncode != 0: - _log.debug( - "branch protection fetch returned rc=%s for %s: %s", - result.returncode, - repo, - result.stderr.strip(), - ) - return None - try: - data: dict[str, object] = json.loads(result.stdout) - except json.JSONDecodeError as exc: - _log.debug("branch protection JSON parse failed for %s: %s", repo, exc) - return None - reviews = data.get("required_pull_request_reviews") - if not isinstance(reviews, dict): - return None - raw = reviews.get("required_approving_review_count") - if isinstance(raw, int): - return raw - return None diff --git a/src/omnimarket/nodes/node_merge_sweep/metadata.yaml b/src/omnimarket/nodes/node_merge_sweep/metadata.yaml deleted file mode 100644 index 9f024dbcf8..0000000000 --- a/src/omnimarket/nodes/node_merge_sweep/metadata.yaml +++ /dev/null @@ -1,26 +0,0 @@ -name: node_merge_sweep -version: "1.0.0" -description: "Org-wide PR sweep — enables GitHub auto-merge on ready PRs and polishes fixable PRs" -omnibase_core_compat: ">=0.39.0,<1.0.0" -entry_points: - onex.nodes: - node_merge_sweep: "omnimarket.nodes.node_merge_sweep" -capabilities: - standalone: true - full_runtime: true - requires_network: true - requires_repo: true - requires_secrets: true - requires_docker: false - side_effect_class: mutating -dependencies: - - "omnibase_core>=0.39.0" -authors: ["OmniNode Platform Team"] -license: "MIT" -tags: ["sweep", "pr", "merge", "github", "pipeline"] -deprecated: true -deprecated_by: node_pr_lifecycle_orchestrator -deprecated_reason: "Superseded by node_pr_lifecycle_orchestrator which provides FSM-driven PR lifecycle management with inventory, triage, merge, fix, and reducer sub-handlers." -pack: "pr_lifecycle" -display_name: "Merge Sweep" -node_role: "effect" diff --git a/src/omnimarket/nodes/node_merge_sweep_compute/__init__.py b/src/omnimarket/nodes/node_merge_sweep_compute/__init__.py new file mode 100644 index 0000000000..fcf707160e --- /dev/null +++ b/src/omnimarket/nodes/node_merge_sweep_compute/__init__.py @@ -0,0 +1,7 @@ +"""node_merge_sweep_compute — Pure PR classification into Track A/B/Skip.""" + +from omnimarket.nodes.node_merge_sweep_compute.handlers.handler_merge_sweep import ( + NodeMergeSweep, +) + +__all__ = ["NodeMergeSweep"] diff --git a/src/omnimarket/nodes/node_merge_sweep/__main__.py b/src/omnimarket/nodes/node_merge_sweep_compute/__main__.py similarity index 72% rename from src/omnimarket/nodes/node_merge_sweep/__main__.py rename to src/omnimarket/nodes/node_merge_sweep_compute/__main__.py index 675ced7ad8..3192fd14a9 100644 --- a/src/omnimarket/nodes/node_merge_sweep/__main__.py +++ b/src/omnimarket/nodes/node_merge_sweep_compute/__main__.py @@ -1,9 +1,9 @@ -# SPDX-FileCopyrightText: 2025 OmniNode.ai Inc. +# SPDX-FileCopyrightText: 2026 OmniNode.ai Inc. # SPDX-License-Identifier: MIT """CLI entry point for node_merge_sweep. -Reads open PRs from GitHub (via gh CLI), classifies them into tracks, and -outputs a JSON classification report to stdout. +Reads open PRs from GitHub (via HTTP adapter), classifies them into tracks, +and outputs a JSON classification report to stdout. Usage: python -m omnimarket.nodes.node_merge_sweep \ @@ -13,6 +13,9 @@ --dry-run Outputs JSON to stdout: ModelMergeSweepResult model. + +Environment: + GH_PAT GitHub PAT (required — fail-fast if missing) """ from __future__ import annotations @@ -22,11 +25,16 @@ import logging import os import pathlib -import subprocess import sys +from typing import Any -from omnimarket.nodes.node_merge_sweep.branch_protection import BranchProtectionCache -from omnimarket.nodes.node_merge_sweep.handlers.handler_merge_sweep import ( +from omnimarket.nodes.node_merge_sweep_compute.adapter_github_http import ( + GitHubHttpClient, +) +from omnimarket.nodes.node_merge_sweep_compute.branch_protection import ( + BranchProtectionCache, +) +from omnimarket.nodes.node_merge_sweep_compute.handlers.handler_merge_sweep import ( ModelFailureHistoryEntry, ModelMergeSweepRequest, ModelPRInfo, @@ -50,52 +58,20 @@ "OmniNode-ai/omniweb", ] -_PR_FIELDS = ( - "number,title,mergeable,mergeStateStatus,statusCheckRollup," - "reviewDecision,headRefName,baseRefName,isDraft,labels,author,updatedAt" -) - - -def _fetch_prs(repo: str) -> list[dict]: # type: ignore[type-arg] - """Fetch open PRs for a repo via gh CLI.""" - result = subprocess.run( - [ - "gh", - "pr", - "list", - "--repo", - repo, - "--state", - "open", - "--json", - _PR_FIELDS, - "--limit", - "100", - ], - capture_output=True, - text=True, - ) - if result.returncode != 0: - _log.warning("gh pr list failed for %s: %s", repo, result.stderr.strip()) - return [] - try: - return json.loads(result.stdout) # type: ignore[no-any-return] - except json.JSONDecodeError as exc: - _log.warning("failed to parse gh output for %s: %s", repo, exc) - return [] - -def _is_green(pr: dict) -> bool: # type: ignore[type-arg] - """All required checks pass.""" +def _is_green(pr: dict[str, Any]) -> bool: rollup = pr.get("statusCheckRollup") or [] - required = [c for c in rollup if c.get("isRequired")] + required = [c for c in rollup if isinstance(c, dict) and c.get("isRequired")] if not required: return True - return all(c.get("conclusion") == "SUCCESS" for c in required) + return all( + isinstance(c, dict) and c.get("conclusion") == "SUCCESS" for c in required + ) -def _to_pr_info(pr: dict, repo: str, required_approving: int | None) -> ModelPRInfo: # type: ignore[type-arg] - # Normalize reviewDecision "" → None (GitHub returns "" on solo-dev repos). +def _to_pr_info( + pr: dict[str, Any], repo: str, required_approving: int | None +) -> ModelPRInfo: review_decision_raw = pr.get("reviewDecision") review_decision = review_decision_raw if review_decision_raw else None return ModelPRInfo( @@ -107,7 +83,9 @@ def _to_pr_info(pr: dict, repo: str, required_approving: int | None) -> ModelPRI is_draft=pr.get("isDraft", False), review_decision=review_decision, required_checks_pass=_is_green(pr), - labels=[lbl["name"] for lbl in (pr.get("labels") or [])], + labels=[ + lbl["name"] for lbl in (pr.get("labels") or []) if isinstance(lbl, dict) + ], required_approving_review_count=required_approving, ) @@ -181,14 +159,16 @@ def main() -> None: ) args = parser.parse_args() - repos = [r.strip() for r in args.repos.split(",") if r.strip()] or _DEFAULT_REPOS + # Fail-fast: GH_PAT must be present + github = GitHubHttpClient() + all_prs: list[ModelPRInfo] = [] - protection = BranchProtectionCache() + protection = BranchProtectionCache(github) for repo in repos: required_approving = protection.required_approving_review_count(repo) - for pr in _fetch_prs(repo): + for pr in github.fetch_open_prs(repo): all_prs.append(_to_pr_info(pr, repo, required_approving)) failure_history = _load_failure_history(onex_state_dir) diff --git a/src/omnimarket/nodes/node_merge_sweep_compute/adapter_github_http.py b/src/omnimarket/nodes/node_merge_sweep_compute/adapter_github_http.py new file mode 100644 index 0000000000..5ffdf09607 --- /dev/null +++ b/src/omnimarket/nodes/node_merge_sweep_compute/adapter_github_http.py @@ -0,0 +1,246 @@ +# SPDX-FileCopyrightText: 2026 OmniNode.ai Inc. +# SPDX-License-Identifier: MIT +"""GitHub HTTP adapter for node_merge_sweep — zero external dependencies. + +Uses GitHub GraphQL API (via urllib) to fetch open PRs with rich status fields, +and REST API for branch protection. Reads GH_PAT from env (fail-fast, no fallback). + +This is the ONLY file in node_merge_sweep that touches the network. +Everything else works against GitHubPrFetchProtocol. + +OMN-MERGE-SWEEP. +""" + +from __future__ import annotations + +import json +import logging +import os +import urllib.error +import urllib.request +from typing import Any + +from omnimarket.nodes.node_merge_sweep_compute.protocols import ( + GitHubPrFetchProtocol, + GitHubTransportError, +) + +_log = logging.getLogger(__name__) + +_GITHUB_GRAPHQL = "https://api.github.com/graphql" +_GITHUB_REST = "https://api.github.com" +_REQUEST_TIMEOUT = 30 + +# GraphQL query for open PRs with all the fields we need. +# Mirrors the output of `gh pr list --json number,title,mergeable,mergeStateStatus, +# statusCheckRollup,reviewDecision,headRefName,baseRefName,isDraft,labels,headRefOid` +_PR_GRAPHQL_QUERY = """ +query($owner: String!, $name: String!, $after: String) { + repository(owner: $owner, name: $name) { + pullRequests(states: [OPEN], first: 100, after: $after, orderBy: {field: UPDATED_AT, direction: DESC}) { + pageInfo { + hasNextPage + endCursor + } + nodes { + number + title + isDraft + mergeable + mergeStateStatus + reviewDecision + headRefName + baseRefName + headRefOid + labels(first: 20) { + nodes { name } + } + statusCheckRollup: commits(last: 1) { + nodes { + commit { + statusCheckRollup { + contexts { + ... on StatusContext { + context + state + description + } + ... on CheckRun { + name + status + conclusion + } + } + } + } + } + } + } + } + } +} +""" + + +def _split_repo(repo: str) -> tuple[str, str]: + """Split 'OmniNode-ai/omnimarket' into ('OmniNode-ai', 'omnimarket').""" + parts = repo.split("/", 1) + if len(parts) != 2: + raise ValueError(f"Invalid repo format: {repo!r} — expected 'org/name'") + return parts[0], parts[1] + + +class GitHubHttpClient(GitHubPrFetchProtocol): + """Real GitHub HTTP client using GraphQL for PRs + REST for branch protection. + + Reads GH_PAT from the environment at construction time (fail-fast). + """ + + def __init__(self) -> None: + token = os.environ.get("GH_PAT", "") + if not token: + raise RuntimeError( + "GH_PAT environment variable is not set. " + "Export it before running node_merge_sweep." + ) + self._token = token + + def _graphql(self, query: str, variables: dict[str, object]) -> dict[str, Any]: + """Execute a GraphQL query. Returns the data dict. + + Raises GitHubTransportError on network, auth, or decode failures so + callers can distinguish transport failure from an empty result. + """ + payload = json.dumps({"query": query, "variables": variables}).encode() + req = urllib.request.Request( + _GITHUB_GRAPHQL, + data=payload, + headers={ + "Authorization": f"bearer {self._token}", + "Content-Type": "application/json", + }, + ) + try: + with urllib.request.urlopen(req, timeout=_REQUEST_TIMEOUT) as resp: + body = json.loads(resp.read()) + except (urllib.error.URLError, json.JSONDecodeError, OSError) as exc: + raise GitHubTransportError(f"GraphQL request failed: {exc}") from exc + if "errors" in body: + _log.warning("GraphQL errors: %s", body["errors"]) + raise GitHubTransportError(f"GraphQL returned errors: {body['errors']}") + data = body.get("data") + return data if isinstance(data, dict) else {} + + def _rest_get(self, path: str) -> dict[str, Any] | None: + """Execute a REST API GET. Returns parsed JSON or None for 404. + + Raises GitHubTransportError on network, auth, or non-404 HTTP failures. + Returns None only for 404 (resource genuinely absent). + """ + url = f"{_GITHUB_REST}{path}" + req = urllib.request.Request( + url, + headers={ + "Authorization": f"bearer {self._token}", + "Accept": "application/vnd.github+json", + }, + ) + try: + with urllib.request.urlopen(req, timeout=_REQUEST_TIMEOUT) as resp: + return json.loads(resp.read()) # type: ignore[no-any-return] + except urllib.error.HTTPError as exc: + # 404 = no branch protection configured — that's a valid business state + if exc.code == 404: + return None + raise GitHubTransportError(f"REST API error for {path}: {exc}") from exc + except (urllib.error.URLError, json.JSONDecodeError, OSError) as exc: + raise GitHubTransportError( + f"REST API request failed for {path}: {exc}" + ) from exc + + def fetch_open_prs(self, repo: str) -> list[dict[str, Any]]: + """Fetch open PRs via GitHub GraphQL API.""" + owner, name = _split_repo(repo) + all_prs: list[dict[str, Any]] = [] + cursor: str | None = None + + while True: + variables: dict[str, object] = {"owner": owner, "name": name} + if cursor: + variables["after"] = cursor + + data = self._graphql(_PR_GRAPHQL_QUERY, variables) + repo_data = data.get("repository") + if not repo_data: + break + + pr_conn = repo_data.get("pullRequests", {}) + nodes = pr_conn.get("nodes", []) + for node in nodes: + # Flatten labels; skip nodes without a name key (malformed API response) + label_nodes = (node.get("labels") or {}).get("nodes", []) + node["labels"] = [ + {"name": ln["name"]} for ln in label_nodes if ln and "name" in ln + ] + + # Flatten statusCheckRollup from nested commit structure + rollup_nodes = ( + ((node.get("statusCheckRollup") or {}).get("nodes") or [{}])[0] + .get("commit", {}) + .get("statusCheckRollup", {}) + .get("contexts", []) + ) + # Normalize to the same shape as gh pr list --json + normalized_rollup: list[dict[str, Any]] = [] + for ctx in rollup_nodes: + if "conclusion" in ctx: + # CheckRun + normalized_rollup.append( + { + "name": ctx.get("name", ""), + "conclusion": (ctx.get("conclusion") or "").upper(), + "status": ctx.get("status", ""), + } + ) + else: + # StatusContext + state = (ctx.get("state") or "").upper() + normalized_rollup.append( + { + "context": ctx.get("context", ""), + "conclusion": "SUCCESS" + if state == "SUCCESS" + else state, + "state": ctx.get("state", ""), + } + ) + # Mark isRequired — GitHub GraphQL doesn't expose this directly + # in the statusCheckRollup. We treat all as required for safety; + # the consumer's _is_green() already handles the empty case. + for ctx in normalized_rollup: + ctx["isRequired"] = True + node["statusCheckRollup"] = normalized_rollup + all_prs.append(node) + + page_info = pr_conn.get("pageInfo", {}) + if not page_info.get("hasNextPage"): + break + cursor = page_info.get("endCursor") + + return all_prs + + def fetch_branch_protection(self, repo: str) -> int | None: + """Fetch required_approving_review_count via REST API.""" + data = self._rest_get(f"/repos/{repo}/branches/main/protection") + if data is None: + return None + reviews = data.get("required_pull_request_reviews") + if not isinstance(reviews, dict): + return None + raw = reviews.get("required_approving_review_count") + if isinstance(raw, int): + return raw + return None + + +__all__: list[str] = ["GitHubHttpClient"] diff --git a/src/omnimarket/nodes/node_merge_sweep_compute/branch_protection.py b/src/omnimarket/nodes/node_merge_sweep_compute/branch_protection.py new file mode 100644 index 0000000000..4bbc1bd017 --- /dev/null +++ b/src/omnimarket/nodes/node_merge_sweep_compute/branch_protection.py @@ -0,0 +1,53 @@ +# SPDX-FileCopyrightText: 2026 OmniNode.ai Inc. +# SPDX-License-Identifier: MIT +"""Per-sweep branch protection cache for merge-sweep [OMN-9106]. + +Caches ``required_approving_review_count`` per repo for the lifetime of a +sweep run. Delegates to GitHubPrFetchProtocol instead of shelling out to +``gh api`` — the adapter (real HTTP or test stub) is injected at construction. + +Returned semantics: + - int >= 0 : live value from GitHub + - None : no approving review required (either explicit ``null`` or missing + ``required_pull_request_reviews`` block, or 404 response) + - raises GitHubTransportError : network/auth failure — callers must handle +""" + +from __future__ import annotations + +import logging +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from omnimarket.nodes.node_merge_sweep_compute.protocols import ( + GitHubPrFetchProtocol, + ) + +_log = logging.getLogger(__name__) + + +class BranchProtectionCache: + """Per-sweep cache of required_approving_review_count per repo. + + One instance per sweep run; discard between runs so stale values don't + persist across sweeps. + """ + + def __init__(self, github: GitHubPrFetchProtocol) -> None: + self._github = github + self._cache: dict[str, int | None] = {} + + def required_approving_review_count(self, repo: str) -> int | None: + """Return required approving review count for ``repo``'s main branch. + + Lazily fetches once per repo per instance. Returns None when no branch + protection requires approving reviews (404 or missing block). + Raises GitHubTransportError on network/auth failures — callers must + handle this rather than treating it as "no protection required". + """ + if repo in self._cache: + return self._cache[repo] + count = self._github.fetch_branch_protection(repo) + self._cache[repo] = count + _log.debug("branch protection for %s: %s", repo, count) + return count diff --git a/src/omnimarket/nodes/node_merge_sweep/consumer.py b/src/omnimarket/nodes/node_merge_sweep_compute/consumer.py similarity index 83% rename from src/omnimarket/nodes/node_merge_sweep/consumer.py rename to src/omnimarket/nodes/node_merge_sweep_compute/consumer.py index 0536cfdbc8..0ab09fa253 100644 --- a/src/omnimarket/nodes/node_merge_sweep/consumer.py +++ b/src/omnimarket/nodes/node_merge_sweep_compute/consumer.py @@ -7,12 +7,13 @@ on completion. Environment variables (all resolved at startup — no hardcoded strings): + GH_PAT GitHub PAT (required — fail-fast if missing) KAFKA_BROKER Redpanda/Kafka bootstrap server (default: localhost:9092) ONEX_STATE_DIR Failure-history state dir (default: ~/.onex_state) MERGE_SWEEP_GROUP Consumer group ID (default: omnimarket.merge_sweep.consume.v1) Usage (standalone consumer loop): - python -m omnimarket.nodes.node_merge_sweep.consumer + python -m omnimarket.nodes.node_merge_sweep_compute.consumer The loop runs until SIGINT/SIGTERM. It is safe to run alongside the existing ``python -m omnimarket.nodes.node_merge_sweep`` CLI — they are independent. @@ -26,12 +27,16 @@ import os import pathlib import signal -import subprocess import sys from typing import Any -from omnimarket.nodes.node_merge_sweep.branch_protection import BranchProtectionCache -from omnimarket.nodes.node_merge_sweep.handlers.handler_merge_sweep import ( +from omnimarket.nodes.node_merge_sweep_compute.adapter_github_http import ( + GitHubHttpClient, +) +from omnimarket.nodes.node_merge_sweep_compute.branch_protection import ( + BranchProtectionCache, +) +from omnimarket.nodes.node_merge_sweep_compute.handlers.handler_merge_sweep import ( TOPIC_MERGE_SWEEP_COMPLETED, TOPIC_MERGE_SWEEP_START, ModelFailureHistoryEntry, @@ -39,6 +44,7 @@ ModelPRInfo, NodeMergeSweep, ) +from omnimarket.nodes.node_merge_sweep_compute.protocols import GitHubTransportError _log = logging.getLogger(__name__) @@ -57,40 +63,6 @@ "OmniNode-ai/omniweb", ] -_PR_FIELDS = ( - "number,title,mergeable,mergeStateStatus,statusCheckRollup," - "reviewDecision,headRefName,baseRefName,isDraft,labels,author,updatedAt" -) - - -def _fetch_prs(repo: str) -> list[dict[str, Any]]: - result = subprocess.run( - [ - "gh", - "pr", - "list", - "--repo", - repo, - "--state", - "open", - "--json", - _PR_FIELDS, - "--limit", - "100", - ], - capture_output=True, - text=True, - timeout=30, - ) - if result.returncode != 0: - _log.warning("gh pr list failed for %s: %s", repo, result.stderr.strip()) - return [] - try: - return json.loads(result.stdout) # type: ignore[no-any-return] - except json.JSONDecodeError as exc: - _log.warning("failed to parse gh output for %s: %s", repo, exc) - return [] - def _is_green(pr: dict[str, Any]) -> bool: rollup = pr.get("statusCheckRollup") or [] @@ -103,7 +75,6 @@ def _is_green(pr: dict[str, Any]) -> bool: def _to_pr_info( pr: dict[str, Any], repo: str, required_approving: int | None ) -> ModelPRInfo: - # Normalize reviewDecision "" → None (GitHub returns "" on solo-dev repos). review_decision_raw = pr.get("reviewDecision") review_decision = review_decision_raw if review_decision_raw else None return ModelPRInfo( @@ -136,7 +107,11 @@ def _load_failure_history(state_dir: str) -> dict[str, ModelFailureHistoryEntry] return {} -def _build_request(cmd: dict[str, Any], state_dir: str) -> ModelMergeSweepRequest: +def _build_request( + github: GitHubHttpClient, + cmd: dict[str, Any], + state_dir: str, +) -> ModelMergeSweepRequest: """Build a ModelMergeSweepRequest from a Kafka command payload.""" repos_raw: str | list[str] = cmd.get("repos", "") if isinstance(repos_raw, list): @@ -145,10 +120,18 @@ def _build_request(cmd: dict[str, Any], state_dir: str) -> ModelMergeSweepReques repos = [r.strip() for r in repos_raw.split(",") if r.strip()] or _DEFAULT_REPOS all_prs: list[ModelPRInfo] = [] - protection = BranchProtectionCache() + protection = BranchProtectionCache(github) for repo in repos: - required_approving = protection.required_approving_review_count(repo) - for pr in _fetch_prs(repo): + try: + required_approving = protection.required_approving_review_count(repo) + prs = github.fetch_open_prs(repo) + except ValueError as exc: + _log.warning("skipping repo %r — invalid format: %s", repo, exc) + continue + except GitHubTransportError as exc: + _log.error("skipping repo %r — GitHub transport error: %s", repo, exc) + continue + for pr in prs: all_prs.append(_to_pr_info(pr, repo, required_approving)) failure_history = _load_failure_history(state_dir) @@ -178,6 +161,9 @@ async def _run_consumer(broker: str, group_id: str, state_dir: str) -> None: ) sys.exit(1) + # Fail-fast: GH_PAT must be present for GitHubHttpClient + github = GitHubHttpClient() + consumer = AIOKafkaConsumer( TOPIC_MERGE_SWEEP_START, bootstrap_servers=broker, @@ -222,7 +208,7 @@ def _signal_handler(sig: int, _: Any) -> None: ) try: - request = _build_request(cmd, state_dir) + request = _build_request(github, cmd, state_dir) result = handler.handle(request) payload: dict[str, Any] = { "correlation_id": correlation_id, diff --git a/src/omnimarket/nodes/node_merge_sweep/contract.yaml b/src/omnimarket/nodes/node_merge_sweep_compute/contract.yaml similarity index 67% rename from src/omnimarket/nodes/node_merge_sweep/contract.yaml rename to src/omnimarket/nodes/node_merge_sweep_compute/contract.yaml index a2352fa2ca..c525fae526 100644 --- a/src/omnimarket/nodes/node_merge_sweep/contract.yaml +++ b/src/omnimarket/nodes/node_merge_sweep_compute/contract.yaml @@ -1,8 +1,8 @@ --- -name: merge_sweep -contract_version: {major: 1, minor: 1, patch: 0} -node_type: orchestrator -node_version: {major: 1, minor: 1, patch: 0} +name: merge_sweep_compute +contract_version: {major: 2, minor: 0, patch: 0} +node_type: compute +node_version: {major: 2, minor: 0, patch: 0} description: > Org-wide PR sweep. Classifies open PRs into merge-ready (Track A), @@ -12,18 +12,18 @@ description: > with escalation thresholds (STUCK, CHRONIC, RECIDIVIST). handler: - module: omnimarket.nodes.node_merge_sweep.handlers.handler_merge_sweep + module: omnimarket.nodes.node_merge_sweep_compute.handlers.handler_merge_sweep class: NodeMergeSweep - input_model: omnimarket.nodes.node_merge_sweep.handlers.handler_merge_sweep.ModelMergeSweepRequest + input_model: omnimarket.nodes.node_merge_sweep_compute.handlers.handler_merge_sweep.ModelMergeSweepRequest handler_routing: default_handler: handler:NodeMergeSweep descriptor: - node_archetype: orchestrator - purity: impure + node_archetype: compute + purity: pure idempotent: true - timeout_ms: 300000 + timeout_ms: 5000 terminal_event: onex.evt.omnimarket.merge-sweep-completed.v1 diff --git a/src/omnimarket/nodes/node_merge_sweep/handlers/__init__.py b/src/omnimarket/nodes/node_merge_sweep_compute/handlers/__init__.py similarity index 100% rename from src/omnimarket/nodes/node_merge_sweep/handlers/__init__.py rename to src/omnimarket/nodes/node_merge_sweep_compute/handlers/__init__.py diff --git a/src/omnimarket/nodes/node_merge_sweep/handlers/handler_merge_sweep.py b/src/omnimarket/nodes/node_merge_sweep_compute/handlers/handler_merge_sweep.py similarity index 100% rename from src/omnimarket/nodes/node_merge_sweep/handlers/handler_merge_sweep.py rename to src/omnimarket/nodes/node_merge_sweep_compute/handlers/handler_merge_sweep.py diff --git a/src/omnimarket/nodes/node_merge_sweep_compute/metadata.yaml b/src/omnimarket/nodes/node_merge_sweep_compute/metadata.yaml new file mode 100644 index 0000000000..e2b863af8b --- /dev/null +++ b/src/omnimarket/nodes/node_merge_sweep_compute/metadata.yaml @@ -0,0 +1,23 @@ +name: node_merge_sweep_compute +version: "2.0.0" +description: "Pure compute: classifies open PRs into Track A/B/Skip queues" +omnibase_core_compat: ">=0.39.0,<1.0.0" +entry_points: + onex.nodes: + node_merge_sweep_compute: "omnimarket.nodes.node_merge_sweep_compute" +capabilities: + standalone: true + full_runtime: true + requires_network: false + requires_repo: false + requires_secrets: false + requires_docker: false + side_effect_class: read_only +dependencies: + - "omnibase_core>=0.39.0" +authors: ["OmniNode Platform Team"] +license: "MIT" +tags: ["compute", "classification", "pr-lifecycle", "merge-sweep"] +pack: "pr-lifecycle" +display_name: "Merge Sweep Compute" +node_role: "compute" diff --git a/src/omnimarket/nodes/node_merge_sweep_compute/protocols.py b/src/omnimarket/nodes/node_merge_sweep_compute/protocols.py new file mode 100644 index 0000000000..661c054672 --- /dev/null +++ b/src/omnimarket/nodes/node_merge_sweep_compute/protocols.py @@ -0,0 +1,48 @@ +# SPDX-FileCopyrightText: 2026 OmniNode.ai Inc. +# SPDX-License-Identifier: MIT +"""Protocols for node_merge_sweep external dependencies. + +These protocols define the contracts that adapters must satisfy. +Real adapters make HTTP calls; test stubs return canned data. +The handler never imports these — only consumer.py and __main__.py do. +""" + +from __future__ import annotations + +from typing import Any, Protocol, runtime_checkable + + +class GitHubTransportError(Exception): + """Raised when a GitHub API call fails due to a transport or auth error. + + Distinct from an empty result (repo has no open PRs), so callers can + abort rather than silently treating an outage as an empty list. + """ + + +@runtime_checkable +class GitHubPrFetchProtocol(Protocol): + """Fetch open PRs with rich status fields from GitHub.""" + + def fetch_open_prs(self, repo: str) -> list[dict[str, Any]]: + """Return open PRs for ``repo`` (org/repo format). + + Each dict must have keys matching gh pr list --json output: + number, title, mergeable, mergeStateStatus, statusCheckRollup, + reviewDecision, isDraft, labels, headRefOid. + + Raises GitHubTransportError on network/auth failures so callers + can distinguish transport failure from an empty PR list. + Raises ValueError if ``repo`` is not in 'org/name' format. + """ + ... + + def fetch_branch_protection(self, repo: str) -> int | None: + """Return required_approving_review_count for repo's main branch. + + Returns None when no branch protection rule requires approving reviews + (404 or missing required_pull_request_reviews block). + Raises GitHubTransportError on network/auth failures so callers + can distinguish transport failure from 'no protection configured'. + """ + ... diff --git a/src/omnimarket/nodes/node_merge_sweep/tests/__init__.py b/src/omnimarket/nodes/node_merge_sweep_compute/tests/__init__.py similarity index 100% rename from src/omnimarket/nodes/node_merge_sweep/tests/__init__.py rename to src/omnimarket/nodes/node_merge_sweep_compute/tests/__init__.py diff --git a/src/omnimarket/nodes/node_merge_sweep_triage_orchestrator/handlers/handler_triage.py b/src/omnimarket/nodes/node_merge_sweep_triage_orchestrator/handlers/handler_triage.py index b63481d20d..fd36096bd7 100644 --- a/src/omnimarket/nodes/node_merge_sweep_triage_orchestrator/handlers/handler_triage.py +++ b/src/omnimarket/nodes/node_merge_sweep_triage_orchestrator/handlers/handler_triage.py @@ -34,7 +34,7 @@ from omnibase_core.models.dispatch.model_handler_output import ModelHandlerOutput -from omnimarket.nodes.node_merge_sweep.handlers.handler_merge_sweep import ( +from omnimarket.nodes.node_merge_sweep_compute.handlers.handler_merge_sweep import ( EnumPRTrack, ModelClassifiedPR, ) diff --git a/src/omnimarket/nodes/node_merge_sweep_triage_orchestrator/models/model_triage_request.py b/src/omnimarket/nodes/node_merge_sweep_triage_orchestrator/models/model_triage_request.py index 97d6a11eec..0326291680 100644 --- a/src/omnimarket/nodes/node_merge_sweep_triage_orchestrator/models/model_triage_request.py +++ b/src/omnimarket/nodes/node_merge_sweep_triage_orchestrator/models/model_triage_request.py @@ -19,7 +19,7 @@ from pydantic import BaseModel, ConfigDict, Field -from omnimarket.nodes.node_merge_sweep.handlers.handler_merge_sweep import ( +from omnimarket.nodes.node_merge_sweep_compute.handlers.handler_merge_sweep import ( ModelMergeSweepResult, ) diff --git a/src/omnimarket/nodes/node_overnight/handlers/handler_overnight.py b/src/omnimarket/nodes/node_overnight/handlers/handler_overnight.py index 87599acb07..0a5c63963a 100644 --- a/src/omnimarket/nodes/node_overnight/handlers/handler_overnight.py +++ b/src/omnimarket/nodes/node_overnight/handlers/handler_overnight.py @@ -1100,7 +1100,7 @@ def _dispatch_merge_sweep( list to prove the wiring; a follow-up ticket must add a PR-inventory adapter before this produces real output. """ - from omnimarket.nodes.node_merge_sweep.handlers.handler_merge_sweep import ( + from omnimarket.nodes.node_merge_sweep_compute.handlers.handler_merge_sweep import ( ModelMergeSweepRequest, NodeMergeSweep, ) diff --git a/src/omnimarket/nodes/node_pr_health_monitor/handlers/handler_pr_health_monitor.py b/src/omnimarket/nodes/node_pr_health_monitor/handlers/handler_pr_health_monitor.py index 997e6c6942..c2de13c223 100644 --- a/src/omnimarket/nodes/node_pr_health_monitor/handlers/handler_pr_health_monitor.py +++ b/src/omnimarket/nodes/node_pr_health_monitor/handlers/handler_pr_health_monitor.py @@ -24,7 +24,9 @@ from datetime import UTC, datetime from typing import Literal -from omnimarket.nodes.node_merge_sweep.handlers.handler_merge_sweep import ModelPRInfo +from omnimarket.nodes.node_merge_sweep_compute.handlers.handler_merge_sweep import ( + ModelPRInfo, +) from omnimarket.nodes.node_pr_health_monitor.models.model_pr_health_input import ( ModelPrHealthInput, ) diff --git a/src/omnimarket/nodes/node_pr_health_monitor/models/model_pr_health_input.py b/src/omnimarket/nodes/node_pr_health_monitor/models/model_pr_health_input.py index 3fd6ce21d8..66317f2884 100644 --- a/src/omnimarket/nodes/node_pr_health_monitor/models/model_pr_health_input.py +++ b/src/omnimarket/nodes/node_pr_health_monitor/models/model_pr_health_input.py @@ -6,7 +6,9 @@ from pydantic import BaseModel, ConfigDict, Field -from omnimarket.nodes.node_merge_sweep.handlers.handler_merge_sweep import ModelPRInfo +from omnimarket.nodes.node_merge_sweep_compute.handlers.handler_merge_sweep import ( + ModelPRInfo, +) class ModelPrHealthInput(BaseModel): diff --git a/src/omnimarket/nodes/node_pr_health_monitor/models/model_pr_health_result.py b/src/omnimarket/nodes/node_pr_health_monitor/models/model_pr_health_result.py index c79e5af00b..1bde403fc5 100644 --- a/src/omnimarket/nodes/node_pr_health_monitor/models/model_pr_health_result.py +++ b/src/omnimarket/nodes/node_pr_health_monitor/models/model_pr_health_result.py @@ -8,7 +8,9 @@ from pydantic import BaseModel, ConfigDict, Field -from omnimarket.nodes.node_merge_sweep.handlers.handler_merge_sweep import ModelPRInfo +from omnimarket.nodes.node_merge_sweep_compute.handlers.handler_merge_sweep import ( + ModelPRInfo, +) class EnumPrHealthStatus(StrEnum): diff --git a/src/omnimarket/nodes/node_pr_snapshot_effect/handlers/handler_pr_snapshot.py b/src/omnimarket/nodes/node_pr_snapshot_effect/handlers/handler_pr_snapshot.py index d0f8b42926..2e6f5cdfa5 100644 --- a/src/omnimarket/nodes/node_pr_snapshot_effect/handlers/handler_pr_snapshot.py +++ b/src/omnimarket/nodes/node_pr_snapshot_effect/handlers/handler_pr_snapshot.py @@ -17,7 +17,7 @@ from pathlib import Path from typing import Any, Literal -from omnimarket.nodes.node_merge_sweep.handlers.handler_merge_sweep import ( +from omnimarket.nodes.node_merge_sweep_compute.handlers.handler_merge_sweep import ( ModelPRInfo, ) from omnimarket.nodes.node_pr_snapshot_effect.models.model_pr_snapshot_input import ( diff --git a/src/omnimarket/nodes/node_pr_snapshot_effect/models/model_pr_snapshot_result.py b/src/omnimarket/nodes/node_pr_snapshot_effect/models/model_pr_snapshot_result.py index 0b01624a99..641fd56ffe 100644 --- a/src/omnimarket/nodes/node_pr_snapshot_effect/models/model_pr_snapshot_result.py +++ b/src/omnimarket/nodes/node_pr_snapshot_effect/models/model_pr_snapshot_result.py @@ -6,7 +6,7 @@ from pydantic import BaseModel, ConfigDict, Field -from omnimarket.nodes.node_merge_sweep.handlers.handler_merge_sweep import ( +from omnimarket.nodes.node_merge_sweep_compute.handlers.handler_merge_sweep import ( ModelPRInfo, ) from omnimarket.nodes.node_pr_snapshot_effect.models.model_pr_stall_event import ( diff --git a/src/omnimarket/nodes/node_pr_snapshot_effect/models/model_repo_scan_result.py b/src/omnimarket/nodes/node_pr_snapshot_effect/models/model_repo_scan_result.py index af09ba11fb..577eea0774 100644 --- a/src/omnimarket/nodes/node_pr_snapshot_effect/models/model_repo_scan_result.py +++ b/src/omnimarket/nodes/node_pr_snapshot_effect/models/model_repo_scan_result.py @@ -6,7 +6,7 @@ from pydantic import BaseModel, ConfigDict, Field -from omnimarket.nodes.node_merge_sweep.handlers.handler_merge_sweep import ( +from omnimarket.nodes.node_merge_sweep_compute.handlers.handler_merge_sweep import ( ModelPRInfo, ) diff --git a/tests/test_golden_chain_merge_sweep.py b/tests/test_golden_chain_merge_sweep.py index e25d22c6bb..2471fee30c 100644 --- a/tests/test_golden_chain_merge_sweep.py +++ b/tests/test_golden_chain_merge_sweep.py @@ -1,549 +1,250 @@ -"""Golden chain test for node_merge_sweep. +"""Golden chain tests for node_merge_sweep. -Verifies PR classification logic, failure history tracking, event bus wiring, -and lifecycle-ordered Track A merge sequencing. +The handler (NodeMergeSweep) is pure compute — all classification logic +is table-driven with zero network calls. Tests construct ModelPRInfo directly +and verify track assignment. """ from __future__ import annotations -import json - import pytest -from omnibase_core.event_bus.event_bus_inmemory import EventBusInmemory -from omnimarket.nodes.node_merge_sweep.handlers.handler_merge_sweep import ( - FailureCategory, - FailureHistoryEntry, - MergeSweepRequest, +from omnimarket.nodes.node_merge_sweep_compute.handlers.handler_merge_sweep import ( + EnumPRTrack, + ModelFailureHistoryEntry, + ModelMergeSweepRequest, + ModelPRInfo, NodeMergeSweep, - PRInfo, - PRTrack, ) -CMD_TOPIC = "onex.cmd.omnimarket.merge-sweep-start.v1" -EVT_TOPIC = "onex.evt.omnimarket.merge-sweep-completed.v1" + +def _pr(**overrides: object) -> ModelPRInfo: + defaults: dict[str, object] = { + "number": 1, + "title": "test PR", + "repo": "OmniNode-ai/omnimarket", + "mergeable": "MERGEABLE", + "merge_state_status": "CLEAN", + "is_draft": False, + "review_decision": "APPROVED", + "required_checks_pass": True, + "labels": [], + "required_approving_review_count": 1, + } + defaults.update(overrides) + return ModelPRInfo(**defaults) # type: ignore[arg-type] @pytest.mark.unit -class TestMergeSweepGoldenChain: - """Golden chain: command -> classify -> completion event.""" +class TestMergeSweepClassification: + """Core track classification logic.""" - async def test_merge_ready_pr_classified_track_a( - self, event_bus: EventBusInmemory - ) -> None: - """A mergeable, green, approved PR should go to Track A.""" + def test_draft_skipped(self) -> None: handler = NodeMergeSweep() - pr = PRInfo( - number=42, - title="feat: add feature", - repo="OmniNode-ai/omniclaude", - mergeable="MERGEABLE", - merge_state_status="CLEAN", - review_decision="APPROVED", - required_checks_pass=True, - ) - request = MergeSweepRequest(prs=[pr]) - result = handler.handle(request) + result = handler.handle(ModelMergeSweepRequest(prs=[_pr(is_draft=True)])) + assert result.classified[0].track == EnumPRTrack.SKIP + assert "Draft" in result.classified[0].reason - assert result.status == "queued" - assert len(result.track_a_merge) == 1 - assert result.track_a_merge[0].pr.number == 42 - assert result.track_a_merge[0].failure_categories == [] + def test_merge_ready(self) -> None: + handler = NodeMergeSweep() + result = handler.handle(ModelMergeSweepRequest(prs=[_pr()])) + assert result.classified[0].track == EnumPRTrack.A_MERGE + assert result.classified[0].reason == "Merge-ready" - async def test_behind_pr_classified_track_a_update( - self, event_bus: EventBusInmemory - ) -> None: - """A mergeable but BEHIND PR should go to Track A-update.""" + def test_behind_goes_to_a_update(self) -> None: handler = NodeMergeSweep() - pr = PRInfo( - number=10, - title="fix: typo", - repo="OmniNode-ai/omnibase_core", - mergeable="MERGEABLE", - merge_state_status="BEHIND", - review_decision="APPROVED", - required_checks_pass=True, + result = handler.handle( + ModelMergeSweepRequest(prs=[_pr(merge_state_status="BEHIND")]) ) - request = MergeSweepRequest(prs=[pr]) - result = handler.handle(request) + assert result.classified[0].track == EnumPRTrack.A_UPDATE + assert "stale" in result.classified[0].reason.lower() - assert len(result.track_a_update) == 1 - assert result.track_a_update[0].track == PRTrack.A_UPDATE - assert ( - FailureCategory.BRANCH_STALE.value - in result.track_a_update[0].failure_categories + def test_unknown_mergeable_goes_to_a_update(self) -> None: + handler = NodeMergeSweep() + result = handler.handle(ModelMergeSweepRequest(prs=[_pr(mergeable="UNKNOWN")])) + assert result.classified[0].track == EnumPRTrack.A_UPDATE + + def test_conflicting_goes_to_b_polish(self) -> None: + handler = NodeMergeSweep() + result = handler.handle( + ModelMergeSweepRequest(prs=[_pr(mergeable="CONFLICTING")]) ) + assert result.classified[0].track == EnumPRTrack.B_POLISH + assert "conflicts" in result.classified[0].reason.lower() - async def test_blocked_green_pr_classified_track_a_resolve( - self, event_bus: EventBusInmemory - ) -> None: - """A MERGEABLE + BLOCKED + GREEN PR goes to Track A-resolve.""" + def test_ci_failing_goes_to_b_polish(self) -> None: handler = NodeMergeSweep() - pr = PRInfo( - number=517, - title="feat: visibility toggles", - repo="OmniNode-ai/omnidash", - mergeable="MERGEABLE", - merge_state_status="BLOCKED", - review_decision=None, - required_checks_pass=True, + result = handler.handle( + ModelMergeSweepRequest(prs=[_pr(required_checks_pass=False)]) ) - request = MergeSweepRequest(prs=[pr]) - result = handler.handle(request) + assert result.classified[0].track == EnumPRTrack.B_POLISH + assert "CI" in result.classified[0].reason - assert len(result.track_a_resolve) == 1 - assert result.track_a_resolve[0].track == PRTrack.A_RESOLVE - assert ( - FailureCategory.THREADS_BLOCKED.value - in result.track_a_resolve[0].failure_categories + def test_changes_requested_goes_to_b_polish(self) -> None: + handler = NodeMergeSweep() + result = handler.handle( + ModelMergeSweepRequest(prs=[_pr(review_decision="CHANGES_REQUESTED")]) ) + assert result.classified[0].track == EnumPRTrack.B_POLISH - async def test_conflicting_pr_classified_track_b( - self, event_bus: EventBusInmemory - ) -> None: - """A conflicting PR should go to Track B for polish.""" + def test_blocked_by_threads_goes_to_a_resolve(self) -> None: + """MERGEABLE + BLOCKED + checks pass = thread resolution.""" handler = NodeMergeSweep() - pr = PRInfo( - number=99, - title="chore: update deps", - repo="OmniNode-ai/omnidash", - mergeable="CONFLICTING", - merge_state_status="DIRTY", - review_decision="APPROVED", - required_checks_pass=True, + result = handler.handle( + ModelMergeSweepRequest( + prs=[_pr(merge_state_status="BLOCKED", required_checks_pass=True)] + ) ) - request = MergeSweepRequest(prs=[pr]) - result = handler.handle(request) + assert result.classified[0].track == EnumPRTrack.A_RESOLVE + assert "thread" in result.classified[0].reason.lower() - assert len(result.track_b_polish) == 1 - assert "conflicts" in result.track_b_polish[0].reason - assert ( - FailureCategory.CONFLICT.value - in result.track_b_polish[0].failure_categories + def test_review_bot_gate_failed_goes_to_a_resolve(self) -> None: + handler = NodeMergeSweep() + result = handler.handle( + ModelMergeSweepRequest(prs=[_pr(review_bot_gate_passed=False)]) ) + assert result.classified[0].track == EnumPRTrack.A_RESOLVE - async def test_ci_failing_pr_has_failure_categories( - self, event_bus: EventBusInmemory - ) -> None: - """A PR with CI failures should have ci_test in failure_categories.""" + def test_review_bot_gate_failed_not_merge_ready(self) -> None: + """review_bot_gate_passed=False should NOT be Track A even if all else green.""" handler = NodeMergeSweep() - pr = PRInfo( - number=200, - title="feat: broken tests", - repo="OmniNode-ai/omniclaude", - mergeable="MERGEABLE", - merge_state_status="BLOCKED", - required_checks_pass=False, + result = handler.handle( + ModelMergeSweepRequest(prs=[_pr(review_bot_gate_passed=False)]) ) - request = MergeSweepRequest(prs=[pr]) - result = handler.handle(request) + assert result.classified[0].track != EnumPRTrack.A_MERGE - assert len(result.track_b_polish) == 1 - assert ( - FailureCategory.CI_TEST.value in result.track_b_polish[0].failure_categories + def test_no_approval_required_solo_dev(self) -> None: + """required_approving_review_count=0 should allow merge without APPROVED.""" + handler = NodeMergeSweep() + result = handler.handle( + ModelMergeSweepRequest( + prs=[_pr(review_decision=None, required_approving_review_count=0)] + ) ) + assert result.classified[0].track == EnumPRTrack.A_MERGE - async def test_draft_pr_skipped(self, event_bus: EventBusInmemory) -> None: - """Draft PRs should be skipped.""" + def test_no_approval_required_none(self) -> None: + """required_approving_review_count=None (no protection) allows merge.""" handler = NodeMergeSweep() - pr = PRInfo( - number=5, - title="wip: new feature", - repo="OmniNode-ai/omniclaude", - mergeable="MERGEABLE", - merge_state_status="CLEAN", - is_draft=True, + result = handler.handle( + ModelMergeSweepRequest( + prs=[_pr(review_decision=None, required_approving_review_count=None)] + ) ) - request = MergeSweepRequest(prs=[pr]) - result = handler.handle(request) + assert result.classified[0].track == EnumPRTrack.A_MERGE - assert result.status == "nothing_to_merge" - assert len(result.skipped) == 1 - async def test_max_total_merges_cap(self, event_bus: EventBusInmemory) -> None: - """max_total_merges should cap Track A candidates.""" +@pytest.mark.unit +class TestMergeSweepResultProperties: + """Result track accessors.""" + + def test_track_accessors(self) -> None: handler = NodeMergeSweep() prs = [ - PRInfo( - number=i, - title=f"PR #{i}", - repo="OmniNode-ai/test", - mergeable="MERGEABLE", - merge_state_status="CLEAN", - review_decision="APPROVED", - required_checks_pass=True, - ) - for i in range(5) + _pr(number=1), # A_MERGE + _pr(number=2, is_draft=True), # SKIP + _pr(number=3, mergeable="CONFLICTING"), # B_POLISH + _pr(number=4, merge_state_status="BEHIND"), # A_UPDATE + _pr( + number=5, merge_state_status="BLOCKED", required_checks_pass=True + ), # A_RESOLVE ] - request = MergeSweepRequest(prs=prs, max_total_merges=2) - result = handler.handle(request) + result = handler.handle(ModelMergeSweepRequest(prs=prs)) - assert len(result.track_a_merge) == 2 - assert len(result.skipped) == 3 + assert len(result.track_a_merge) == 1 + assert len(result.track_a_update) == 1 + assert len(result.track_a_resolve) == 1 + assert len(result.track_b_polish) == 1 + assert len(result.skipped) == 1 - async def test_skip_polish_flag(self, event_bus: EventBusInmemory) -> None: - """--skip-polish should move Track B PRs to skip.""" + def test_status_queued_when_actionable(self) -> None: handler = NodeMergeSweep() - pr = PRInfo( - number=77, - title="fix: broken", - repo="OmniNode-ai/test", - mergeable="CONFLICTING", - merge_state_status="DIRTY", - ) - request = MergeSweepRequest(prs=[pr], skip_polish=True) - result = handler.handle(request) + result = handler.handle(ModelMergeSweepRequest(prs=[_pr()])) + assert result.status == "queued" - assert len(result.track_b_polish) == 0 - assert len(result.skipped) == 1 + def test_status_nothing_when_empty(self) -> None: + handler = NodeMergeSweep() + result = handler.handle(ModelMergeSweepRequest(prs=[])) + assert result.status == "nothing_to_merge" @pytest.mark.unit -class TestFailureHistory: - """Cross-run failure history tracking and escalation.""" +class TestMergeSweepFailureHistory: + """Failure history escalation.""" - async def test_chronic_pr_skips_polish(self, event_bus: EventBusInmemory) -> None: - """A PR with >=5 consecutive failures should skip polish.""" + def test_chronic_skips_polish(self) -> None: handler = NodeMergeSweep() - pr = PRInfo( - number=99, - title="fix: always broken", - repo="OmniNode-ai/test", - mergeable="CONFLICTING", - merge_state_status="DIRTY", - ) history = { - "OmniNode-ai/test#99": FailureHistoryEntry( - first_seen="2026-04-01T00:00:00Z", - last_seen="2026-04-05T00:00:00Z", + "OmniNode-ai/omnimarket#3": ModelFailureHistoryEntry( + first_seen="2026-01-01", + last_seen="2026-01-05", consecutive_failures=5, - total_failures=5, - total_polishes=3, - last_failure_categories=["conflict"], - last_result="blocked", - last_run_id="ms-prev", - runs_seen=["ms-1", "ms-2", "ms-3", "ms-4", "ms-prev"], - ), + total_failures=10, + ) } - request = MergeSweepRequest(prs=[pr], failure_history=history) - result = handler.handle(request) - - assert len(result.track_b_polish) == 0 - assert len(result.skipped) == 1 - assert "CHRONIC" in result.skipped[0].reason + result = handler.handle( + ModelMergeSweepRequest( + prs=[_pr(number=3, mergeable="CONFLICTING")], + failure_history=history, + ) + ) + assert result.classified[0].track == EnumPRTrack.SKIP + assert "CHRONIC" in result.classified[0].reason - async def test_recidivist_pr_skips_polish( - self, event_bus: EventBusInmemory - ) -> None: - """A PR polished >=3 times but still failing should skip polish.""" + def test_recidivist_skips_polish(self) -> None: handler = NodeMergeSweep() - pr = PRInfo( - number=42, - title="fix: keeps breaking", - repo="OmniNode-ai/test", - mergeable="MERGEABLE", - merge_state_status="BLOCKED", - required_checks_pass=False, - ) history = { - "OmniNode-ai/test#42": FailureHistoryEntry( - first_seen="2026-04-01T00:00:00Z", - last_seen="2026-04-05T00:00:00Z", + "OmniNode-ai/omnimarket#3": ModelFailureHistoryEntry( + first_seen="2026-01-01", + last_seen="2026-01-05", consecutive_failures=1, - total_failures=4, + total_failures=3, total_polishes=3, - last_failure_categories=["ci_test"], - last_result="polished_and_queued", - last_run_id="ms-prev", - runs_seen=["ms-1", "ms-2", "ms-3", "ms-prev"], - ), + ) } - request = MergeSweepRequest(prs=[pr], failure_history=history) - result = handler.handle(request) - - assert len(result.track_b_polish) == 0 - assert len(result.skipped) == 1 - assert "RECIDIVIST" in result.skipped[0].reason - - async def test_stuck_pr_still_gets_polished( - self, event_bus: EventBusInmemory - ) -> None: - """A PR with 3-4 consecutive failures (STUCK) should still get polished.""" - handler = NodeMergeSweep() - pr = PRInfo( - number=10, - title="fix: struggling", - repo="OmniNode-ai/test", - mergeable="CONFLICTING", - merge_state_status="DIRTY", + result = handler.handle( + ModelMergeSweepRequest( + prs=[_pr(number=3, mergeable="CONFLICTING")], + failure_history=history, + ) ) - history = { - "OmniNode-ai/test#10": FailureHistoryEntry( - first_seen="2026-04-03T00:00:00Z", - last_seen="2026-04-05T00:00:00Z", - consecutive_failures=3, - total_failures=3, - total_polishes=1, - last_failure_categories=["conflict"], - last_result="blocked", - last_run_id="ms-prev", - runs_seen=["ms-1", "ms-2", "ms-prev"], - ), - } - request = MergeSweepRequest(prs=[pr], failure_history=history) - result = handler.handle(request) - - # STUCK but still below CHRONIC threshold — should still polish - assert len(result.track_b_polish) == 1 + assert result.classified[0].track == EnumPRTrack.SKIP + assert "RECIDIVIST" in result.classified[0].reason - async def test_failure_history_summary_computed( - self, event_bus: EventBusInmemory - ) -> None: - """Failure history summary should aggregate stats correctly.""" + def test_failure_summary(self) -> None: handler = NodeMergeSweep() - pr = PRInfo( - number=1, - title="test", - repo="test/repo", - mergeable="MERGEABLE", - merge_state_status="CLEAN", - review_decision="APPROVED", - required_checks_pass=True, - ) history = { - "test/repo#10": FailureHistoryEntry( - first_seen="2026-04-01T00:00:00Z", - last_seen="2026-04-05T00:00:00Z", - consecutive_failures=3, - total_failures=3, - ), - "test/repo#20": FailureHistoryEntry( - first_seen="2026-04-01T00:00:00Z", - last_seen="2026-04-05T00:00:00Z", - consecutive_failures=6, - total_failures=6, + "a#1": ModelFailureHistoryEntry( + first_seen="2026-01-01", last_seen="2026-01-02", consecutive_failures=3 ), - "test/repo#30": FailureHistoryEntry( - first_seen="2026-04-01T00:00:00Z", - last_seen="2026-04-05T00:00:00Z", - consecutive_failures=1, - total_failures=4, - total_polishes=3, + "a#2": ModelFailureHistoryEntry( + first_seen="2026-01-01", last_seen="2026-01-05", consecutive_failures=5 ), } - request = MergeSweepRequest(prs=[pr], failure_history=history) - result = handler.handle(request) - - assert result.failure_history_summary.total_tracked == 3 - assert result.failure_history_summary.stuck_prs == 1 # #10 (3 consecutive) - assert result.failure_history_summary.chronic_prs == 1 # #20 (6 consecutive) - assert ( - result.failure_history_summary.recidivist_prs == 1 - ) # #30 (3 polishes, still failing) - - async def test_no_failure_history_still_works( - self, event_bus: EventBusInmemory - ) -> None: - """Handler works correctly with empty failure history.""" - handler = NodeMergeSweep() - pr = PRInfo( - number=1, - title="feat: new", - repo="test/repo", - mergeable="CONFLICTING", - merge_state_status="DIRTY", - ) - request = MergeSweepRequest(prs=[pr]) - result = handler.handle(request) - - assert len(result.track_b_polish) == 1 - assert result.failure_history_summary.total_tracked == 0 + result = handler.handle(ModelMergeSweepRequest(prs=[], failure_history=history)) + assert result.failure_history_summary.total_tracked == 2 + assert result.failure_history_summary.stuck_prs == 1 + assert result.failure_history_summary.chronic_prs == 1 @pytest.mark.unit -class TestMergeSweepEventBus: - """Event bus wiring tests.""" +class TestMergeSweepCaps: + """max_total_merges and skip_polish.""" - async def test_event_bus_wiring(self, event_bus: EventBusInmemory) -> None: - """Handler can be wired to event bus for command/completion flow.""" + def test_max_total_merges_cap(self) -> None: handler = NodeMergeSweep() - completions: list[dict[str, object]] = [] - - async def on_command(message: object) -> None: - payload = json.loads(message.value) - prs = [PRInfo(**pr_data) for pr_data in payload.get("prs", [])] - request = MergeSweepRequest(prs=prs) - result = handler.handle(request) - completion = { - "status": result.status, - "track_a": len(result.track_a_merge), - "track_b": len(result.track_b_polish), - } - completions.append(completion) - await event_bus.publish( - EVT_TOPIC, - key=None, - value=json.dumps(completion).encode(), - ) - - await event_bus.start() - await event_bus.subscribe( - CMD_TOPIC, on_message=on_command, group_id="test-merge" - ) - - cmd_payload = json.dumps( - { - "prs": [ - { - "number": 1, - "title": "test", - "repo": "test/repo", - "mergeable": "MERGEABLE", - "merge_state_status": "CLEAN", - "review_decision": "APPROVED", - "required_checks_pass": True, - } - ] - } - ).encode() - await event_bus.publish(CMD_TOPIC, key=None, value=cmd_payload) - - assert len(completions) == 1 - assert completions[0]["status"] == "queued" - assert completions[0]["track_a"] == 1 - - history = await event_bus.get_event_history(topic=EVT_TOPIC) - assert len(history) == 1 - - await event_bus.close() - + prs = [_pr(number=i) for i in range(5)] + result = handler.handle(ModelMergeSweepRequest(prs=prs, max_total_merges=2)) + assert len(result.track_a_merge) == 2 + assert len(result.skipped) == 3 -@pytest.mark.unit -class TestLifecycleOrdering: - """Lifecycle-ordered Track A merge sequencing via pr_lifecycle_orchestrator.""" - - async def test_use_lifecycle_ordering_flag_accepted( - self, event_bus: EventBusInmemory - ) -> None: - """use_lifecycle_ordering=True is accepted by the request model.""" - pr = PRInfo( - number=1, - title="feat: single PR", - repo="OmniNode-ai/omnimarket", - mergeable="MERGEABLE", - merge_state_status="CLEAN", - review_decision="APPROVED", - required_checks_pass=True, - ) - request = MergeSweepRequest(prs=[pr], use_lifecycle_ordering=True) + def test_skip_polish(self) -> None: handler = NodeMergeSweep() - result = handler.handle(request) - - assert result.status == "queued" - assert len(result.track_a_merge) == 1 - - async def test_lifecycle_ordering_preserves_all_track_a( - self, event_bus: EventBusInmemory - ) -> None: - """All Track A PRs survive lifecycle ordering — none are dropped.""" - prs = [ - PRInfo( - number=i, - title=f"feat: PR {i}", - repo="OmniNode-ai/omnimarket", - mergeable="MERGEABLE", - merge_state_status="CLEAN", - review_decision="APPROVED", - required_checks_pass=True, + result = handler.handle( + ModelMergeSweepRequest( + prs=[_pr(number=1, mergeable="CONFLICTING")], + skip_polish=True, ) - for i in range(1, 5) - ] - request = MergeSweepRequest(prs=prs, use_lifecycle_ordering=True) - handler = NodeMergeSweep() - result = handler.handle(request) - - assert result.status == "queued" - assert len(result.track_a_merge) == 4 - # All original PR numbers present - numbers = {c.pr.number for c in result.track_a_merge} - assert numbers == {1, 2, 3, 4} - - async def test_lifecycle_ordering_mixed_tracks( - self, event_bus: EventBusInmemory - ) -> None: - """Non-Track-A PRs are not reordered; Track A PRs are lifecycle-ordered.""" - prs = [ - PRInfo( - number=10, - title="feat: green PR", - repo="OmniNode-ai/omniclaude", - mergeable="MERGEABLE", - merge_state_status="CLEAN", - review_decision="APPROVED", - required_checks_pass=True, - ), - PRInfo( - number=20, - title="fix: conflicted", - repo="OmniNode-ai/omniclaude", - mergeable="CONFLICTING", - merge_state_status="DIRTY", - ), - PRInfo( - number=30, - title="feat: another green", - repo="OmniNode-ai/omnimarket", - mergeable="MERGEABLE", - merge_state_status="CLEAN", - review_decision="APPROVED", - required_checks_pass=True, - ), - ] - request = MergeSweepRequest(prs=prs, use_lifecycle_ordering=True) - handler = NodeMergeSweep() - result = handler.handle(request) - - assert result.status == "queued" - assert len(result.track_a_merge) == 2 - assert len(result.track_b_polish) == 1 - # Track A PRs contain both green PRs - track_a_numbers = {c.pr.number for c in result.track_a_merge} - assert track_a_numbers == {10, 30} - - async def test_lifecycle_ordering_false_by_default( - self, event_bus: EventBusInmemory - ) -> None: - """use_lifecycle_ordering defaults to False — existing behaviour unchanged.""" - pr = PRInfo( - number=1, - title="feat: test", - repo="OmniNode-ai/omnimarket", - mergeable="MERGEABLE", - merge_state_status="CLEAN", - review_decision="APPROVED", - required_checks_pass=True, - ) - request = MergeSweepRequest(prs=[pr]) - assert request.use_lifecycle_ordering is False - handler = NodeMergeSweep() - result = handler.handle(request) - assert result.status == "queued" - - async def test_lifecycle_ordering_empty_track_a_no_error( - self, event_bus: EventBusInmemory - ) -> None: - """use_lifecycle_ordering=True with no Track A PRs does not error.""" - pr = PRInfo( - number=99, - title="wip: draft", - repo="OmniNode-ai/omnimarket", - mergeable="MERGEABLE", - merge_state_status="CLEAN", - is_draft=True, ) - request = MergeSweepRequest(prs=[pr], use_lifecycle_ordering=True) - handler = NodeMergeSweep() - result = handler.handle(request) - - assert result.status == "nothing_to_merge" - assert len(result.track_a_merge) == 0 + assert result.classified[0].track == EnumPRTrack.SKIP + assert "Polish skipped" in result.classified[0].reason diff --git a/tests/test_golden_chain_merge_sweep_executor.py b/tests/test_golden_chain_merge_sweep_executor.py index a900e844bc..5c112cf998 100644 --- a/tests/test_golden_chain_merge_sweep_executor.py +++ b/tests/test_golden_chain_merge_sweep_executor.py @@ -20,15 +20,15 @@ from omnimarket.nodes.node_ci_rerun_effect.handlers.handler_ci_rerun import ( HandlerCiRerunEffect, ) -from omnimarket.nodes.node_merge_sweep.handlers.handler_merge_sweep import ( +from omnimarket.nodes.node_merge_sweep_auto_merge_arm_effect.handlers.handler_auto_merge_arm import ( + HandlerAutoMergeArmEffect, +) +from omnimarket.nodes.node_merge_sweep_compute.handlers.handler_merge_sweep import ( EnumPRTrack, ModelClassifiedPR, ModelMergeSweepResult, ModelPRInfo, ) -from omnimarket.nodes.node_merge_sweep_auto_merge_arm_effect.handlers.handler_auto_merge_arm import ( - HandlerAutoMergeArmEffect, -) from omnimarket.nodes.node_merge_sweep_state_reducer.handlers.handler_sweep_state import ( HandlerMergeSweepStateReducer, ) diff --git a/tests/test_golden_chain_pr_health_monitor.py b/tests/test_golden_chain_pr_health_monitor.py index 16ac35a120..a517cdba60 100644 --- a/tests/test_golden_chain_pr_health_monitor.py +++ b/tests/test_golden_chain_pr_health_monitor.py @@ -10,7 +10,9 @@ import pytest -from omnimarket.nodes.node_merge_sweep.handlers.handler_merge_sweep import ModelPRInfo +from omnimarket.nodes.node_merge_sweep_compute.handlers.handler_merge_sweep import ( + ModelPRInfo, +) from omnimarket.nodes.node_pr_health_monitor.handlers.handler_pr_health_monitor import ( HandlerPrHealthMonitor, _classify_pr, diff --git a/tests/test_golden_chain_pr_snapshot_effect.py b/tests/test_golden_chain_pr_snapshot_effect.py index c2efa933f3..32ec0074ac 100644 --- a/tests/test_golden_chain_pr_snapshot_effect.py +++ b/tests/test_golden_chain_pr_snapshot_effect.py @@ -15,7 +15,7 @@ import pytest -from omnimarket.nodes.node_merge_sweep.handlers.handler_merge_sweep import ( +from omnimarket.nodes.node_merge_sweep_compute.handlers.handler_merge_sweep import ( ModelMergeSweepRequest, ModelPRInfo, ) diff --git a/tests/test_merge_sweep_consumer.py b/tests/test_merge_sweep_consumer.py index c623798824..7ba7131ffe 100644 --- a/tests/test_merge_sweep_consumer.py +++ b/tests/test_merge_sweep_consumer.py @@ -1,6 +1,6 @@ # SPDX-FileCopyrightText: 2025 OmniNode.ai Inc. # SPDX-License-Identifier: MIT -"""Unit tests for node_merge_sweep Kafka consumer wiring. +"""Unit tests for node_merge_sweep_compute Kafka consumer wiring. Tests the topics module constants and the _build_request helper that translates a Kafka command payload into a ModelMergeSweepRequest. @@ -9,11 +9,11 @@ from __future__ import annotations import pathlib -from unittest.mock import patch +from unittest.mock import MagicMock import pytest -from omnimarket.nodes.node_merge_sweep.handlers.handler_merge_sweep import ( +from omnimarket.nodes.node_merge_sweep_compute.handlers.handler_merge_sweep import ( TOPIC_MERGE_SWEEP_COMPLETED, TOPIC_MERGE_SWEEP_START, ) @@ -37,14 +37,17 @@ def test_completed_topic_matches_contract(self) -> None: class TestBuildRequest: """_build_request translates Kafka command payload → ModelMergeSweepRequest.""" + def _make_stub_client(self, prs: list[dict] | None = None) -> MagicMock: + client = MagicMock() + client.fetch_open_prs.return_value = prs or [] + client.fetch_branch_protection.return_value = None + return client + def test_defaults_used_when_command_is_empty(self, tmp_path: pathlib.Path) -> None: - from omnimarket.nodes.node_merge_sweep.consumer import _build_request + from omnimarket.nodes.node_merge_sweep_compute.consumer import _build_request - with patch( - "omnimarket.nodes.node_merge_sweep.consumer._fetch_prs", - return_value=[], - ): - req = _build_request({}, str(tmp_path)) + github = self._make_stub_client() + req = _build_request(github, {}, str(tmp_path)) assert req.require_approval is True assert req.merge_method == "squash" @@ -54,7 +57,7 @@ def test_defaults_used_when_command_is_empty(self, tmp_path: pathlib.Path) -> No assert req.prs == [] def test_command_overrides_defaults(self, tmp_path: pathlib.Path) -> None: - from omnimarket.nodes.node_merge_sweep.consumer import _build_request + from omnimarket.nodes.node_merge_sweep_compute.consumer import _build_request cmd = { "repos": "OmniNode-ai/omniclaude", @@ -66,11 +69,8 @@ def test_command_overrides_defaults(self, tmp_path: pathlib.Path) -> None: "correlation_id": "test-123", } - with patch( - "omnimarket.nodes.node_merge_sweep.consumer._fetch_prs", - return_value=[], - ): - req = _build_request(cmd, str(tmp_path)) + github = self._make_stub_client() + req = _build_request(github, cmd, str(tmp_path)) assert req.require_approval is False assert req.merge_method == "rebase" @@ -80,19 +80,37 @@ def test_command_overrides_defaults(self, tmp_path: pathlib.Path) -> None: assert req.run_id == "test-123" def test_repos_list_accepted(self, tmp_path: pathlib.Path) -> None: - from omnimarket.nodes.node_merge_sweep.consumer import _build_request + from omnimarket.nodes.node_merge_sweep_compute.consumer import _build_request cmd = {"repos": ["OmniNode-ai/omniclaude", "OmniNode-ai/omnidash"]} - calls: list[str] = [] + github = self._make_stub_client() + _build_request(github, cmd, str(tmp_path)) + + assert github.fetch_open_prs.call_count == 2 + + def test_transport_error_skips_repo(self, tmp_path: pathlib.Path) -> None: + from omnimarket.nodes.node_merge_sweep_compute.consumer import _build_request + from omnimarket.nodes.node_merge_sweep_compute.protocols import ( + GitHubTransportError, + ) + + github = MagicMock() + github.fetch_branch_protection.return_value = None + github.fetch_open_prs.side_effect = GitHubTransportError("network down") - def _fake_fetch(repo: str) -> list[dict]: - calls.append(repo) - return [] + cmd = {"repos": "OmniNode-ai/omniclaude"} + req = _build_request(github, cmd, str(tmp_path)) - with patch( - "omnimarket.nodes.node_merge_sweep.consumer._fetch_prs", - side_effect=_fake_fetch, - ): - _build_request(cmd, str(tmp_path)) + assert req.prs == [] + + def test_invalid_repo_format_skips_repo(self, tmp_path: pathlib.Path) -> None: + from omnimarket.nodes.node_merge_sweep_compute.consumer import _build_request + + github = MagicMock() + github.fetch_branch_protection.return_value = None + github.fetch_open_prs.side_effect = ValueError("Invalid repo format") - assert calls == ["OmniNode-ai/omniclaude", "OmniNode-ai/omnidash"] + cmd = {"repos": "not-a-valid-repo-format"} + req = _build_request(github, cmd, str(tmp_path)) + + assert req.prs == [] diff --git a/tests/test_merge_sweep_review_bot_gate.py b/tests/test_merge_sweep_review_bot_gate.py index f221e8a40b..66e89752e8 100644 --- a/tests/test_merge_sweep_review_bot_gate.py +++ b/tests/test_merge_sweep_review_bot_gate.py @@ -6,7 +6,7 @@ import pytest -from omnimarket.nodes.node_merge_sweep.handlers.handler_merge_sweep import ( +from omnimarket.nodes.node_merge_sweep_compute.handlers.handler_merge_sweep import ( ModelMergeSweepRequest, ModelPRInfo, NodeMergeSweep, diff --git a/tests/test_metadata_schema.py b/tests/test_metadata_schema.py index 43e5ad09c6..7c61c0830e 100644 --- a/tests/test_metadata_schema.py +++ b/tests/test_metadata_schema.py @@ -61,7 +61,7 @@ def test_deprecation_fields_default_to_not_deprecated(self) -> None: def test_deprecation_fields_parse_correctly(self) -> None: """A node marked deprecated parses all three deprecation fields.""" data = { - "name": "node_merge_sweep", + "name": "node_merge_sweep_compute", "version": "1.0.0", "description": "Old node", "deprecated": True, @@ -73,15 +73,14 @@ def test_deprecation_fields_parse_correctly(self) -> None: assert schema.deprecated_by == "node_pr_lifecycle_orchestrator" assert schema.deprecated_reason == "Superseded by orchestrator." - def test_node_merge_sweep_is_deprecated(self) -> None: - """node_merge_sweep metadata.yaml marks node as deprecated.""" - meta_path = _NODES_DIR / "node_merge_sweep" / "metadata.yaml" + def test_node_merge_sweep_compute_is_not_deprecated(self) -> None: + """node_merge_sweep_compute metadata.yaml — active compute node.""" + meta_path = _NODES_DIR / "node_merge_sweep_compute" / "metadata.yaml" with meta_path.open() as f: data = yaml.safe_load(f) schema = MetadataSchema(**data) - assert schema.deprecated is True - assert schema.deprecated_by == "node_pr_lifecycle_orchestrator" - assert schema.deprecated_reason + assert schema.deprecated is False + assert schema.node_role.value == "compute" def test_node_pr_snapshot_effect_is_deprecated(self) -> None: """node_pr_snapshot_effect metadata.yaml marks node as deprecated.""" diff --git a/tests/test_plugin_compat_handshake.py b/tests/test_plugin_compat_handshake.py index 065a147c21..c5aac4b616 100644 --- a/tests/test_plugin_compat_handshake.py +++ b/tests/test_plugin_compat_handshake.py @@ -52,7 +52,7 @@ def compat_yaml_content() -> dict: ], }, { - "node": "node_merge_sweep", + "node": "node_merge_sweep_compute", "topics": [ { "name": "onex.cmd.omnimarket.merge-sweep-start.v1", diff --git a/tests/test_proof_runtime_local_merge_sweep.py b/tests/test_proof_runtime_local_merge_sweep.py index 78a71a0a38..f6cd8206f8 100644 --- a/tests/test_proof_runtime_local_merge_sweep.py +++ b/tests/test_proof_runtime_local_merge_sweep.py @@ -11,7 +11,7 @@ CONTRACT_PATH = ( Path(__file__).resolve().parents[1] - / "src/omnimarket/nodes/node_merge_sweep/contract.yaml" + / "src/omnimarket/nodes/node_merge_sweep_compute/contract.yaml" ) @@ -36,7 +36,7 @@ def test_runtime_local_runs_merge_sweep_with_defaults(tmp_path: Path) -> None: data = json.loads(state_file.read_text()) assert data["result"] == "completed" assert data["exit_code"] == 0 - assert data["workflow"].endswith("node_merge_sweep/contract.yaml") + assert data["workflow"].endswith("node_merge_sweep_compute/contract.yaml") def test_runtime_local_runs_merge_sweep_with_real_payload(tmp_path: Path) -> None: @@ -70,7 +70,7 @@ def test_runtime_local_runs_merge_sweep_with_real_payload(tmp_path: Path) -> Non # EnumWorkflowResult.COMPLETED.value is the lowercase string "completed". assert state["result"] == "completed" assert state["exit_code"] == 0 - assert state["workflow"].endswith("node_merge_sweep/contract.yaml") + assert state["workflow"].endswith("node_merge_sweep_compute/contract.yaml") def test_runtime_local_publishes_terminal_event_to_bus(tmp_path: Path) -> None: diff --git a/tests/test_stall_detector.py b/tests/test_stall_detector.py index d49f644f22..e88e5971f4 100644 --- a/tests/test_stall_detector.py +++ b/tests/test_stall_detector.py @@ -20,7 +20,9 @@ import pytest -from omnimarket.nodes.node_merge_sweep.handlers.handler_merge_sweep import ModelPRInfo +from omnimarket.nodes.node_merge_sweep_compute.handlers.handler_merge_sweep import ( + ModelPRInfo, +) from omnimarket.nodes.node_pr_snapshot_effect.handlers.handler_pr_snapshot import ( HandlerPrSnapshot, _detect_stalls, diff --git a/tests/test_triage_orchestrator.py b/tests/test_triage_orchestrator.py index 62c149085f..294d69a647 100644 --- a/tests/test_triage_orchestrator.py +++ b/tests/test_triage_orchestrator.py @@ -14,7 +14,7 @@ import pytest -from omnimarket.nodes.node_merge_sweep.handlers.handler_merge_sweep import ( +from omnimarket.nodes.node_merge_sweep_compute.handlers.handler_merge_sweep import ( EnumPRTrack, ModelClassifiedPR, ModelMergeSweepResult, diff --git a/tests/test_triage_phase2_emit_rules.py b/tests/test_triage_phase2_emit_rules.py index 2bd3b4065f..bbfe0384ee 100644 --- a/tests/test_triage_phase2_emit_rules.py +++ b/tests/test_triage_phase2_emit_rules.py @@ -19,7 +19,7 @@ import pytest -from omnimarket.nodes.node_merge_sweep.handlers.handler_merge_sweep import ( +from omnimarket.nodes.node_merge_sweep_compute.handlers.handler_merge_sweep import ( EnumPRTrack, ModelClassifiedPR, ModelMergeSweepResult,