Deploy OpenClaw PR API

README.md

@@ -31,5 +31,6 @@ CLI examples:
 ```bash
 pr-search repo status
 pr-search pr similar 67096
-pr-search pr probe 67096 --json
+pr-search pr clusters 67096
+pr-search --json pr similar 67096
 ```

src/slop_farmer/app/pr_search.py

@@ -8,7 +8,9 @@ from slop_farmer.reports import pr_search_service
 run_pr_search_refresh = pr_search_service.run_pr_search_refresh
 get_pr_search_status = pr_search_service.get_pr_search_status
 get_pr_search_similar = pr_search_service.get_pr_search_similar
+get_pr_search_similar_lookup = pr_search_service.get_pr_search_similar_lookup
 get_pr_search_candidate_clusters = pr_search_service.get_pr_search_candidate_clusters
+get_pr_search_clusters = pr_search_service.get_pr_search_clusters
 get_pr_search_cluster = pr_search_service.get_pr_search_cluster
 explain_pr_search_pair = pr_search_service.explain_pr_search_pair
 probe_pr_search_live = pr_search_service.probe_pr_search_live
@@ -44,6 +46,11 @@ def format_pr_search_similar(result: Mapping[str, Any]) -> str:
         f"Active snapshot: {result['snapshot_id']}",
         "",
     ]
+    query = result.get("query") or {}
+    if query.get("mode_used") == "live":
+        source = query.get("source") or "live"
+        lines.insert(3, f"Lookup mode: live via {source}")
+        lines.insert(4, "")
     if not result["similar_prs"]:
         lines.append("No similar PRs found in the active run.")
         return "\n".join(lines)
@@ -87,6 +94,46 @@ def format_pr_search_candidate_clusters(result: Mapping[str, Any]) -> str:
     return "\n".join(lines)
 
 
+def format_pr_search_clusters(result: Mapping[str, Any]) -> str:
+    lines = [
+        f"PR #{result['pr']['pr_number']}: cluster context",
+        "",
+    ]
+    query = result.get("query") or {}
+    if query.get("mode_used") == "live":
+        source = query.get("source") or "live"
+        lines.extend([f"Lookup mode: live via {source}", ""])
+    assigned_clusters = result.get("assigned_clusters") or []
+    lines.append("Assigned clusters:")
+    if not assigned_clusters:
+        lines.append("- none")
+    else:
+        for cluster in assigned_clusters:
+            lines.append(
+                f"- {cluster['cluster_id']}  representative=PR #{cluster['representative_pr_number']}  "
+                f"size={cluster['cluster_size']}"
+            )
+            if cluster.get("summary"):
+                lines.append(f"  {cluster['summary']}")
+    lines.extend(["", "Candidate clusters:"])
+    candidate_clusters = result.get("candidate_clusters") or []
+    if not candidate_clusters:
+        lines.append("- none")
+        return "\n".join(lines)
+    for index, row in enumerate(candidate_clusters, start=1):
+        lines.append(
+            f"{index}. {row['cluster_id']}  score={row['candidate_score']:.2f}  "
+            f"assigned={'yes' if row['assigned'] else 'no'}"
+        )
+        lines.append(f"   representative: PR #{row['representative_pr_number']}")
+        matched = row.get("matched_member_pr_numbers") or []
+        if matched:
+            lines.append(f"   matched members: {', '.join(f'#{number}' for number in matched)}")
+        if row.get("reason"):
+            lines.append(f"   reason: {row['reason']}")
+    return "\n".join(lines)
+
+
 def format_pr_search_cluster(result: Mapping[str, Any]) -> str:
     cluster = result["cluster"]
     lines = [

src/slop_farmer/app/pr_search_api.py

@@ -4,7 +4,7 @@ import os
 from contextlib import asynccontextmanager
 from dataclasses import dataclass
 from pathlib import Path
-from typing import Any
+from typing import Any, Literal
 
 from fastapi import FastAPI, HTTPException, Request
 from fastapi.responses import JSONResponse
@@ -12,11 +12,10 @@ from fastapi.responses import JSONResponse
 from slop_farmer.config import PrSearchRefreshOptions
 from slop_farmer.data.ghreplica_api import GhReplicaProbeUnavailableError, GhrProbeClient
 from slop_farmer.reports.pr_search_service import (
-    get_pr_search_candidate_clusters,
     get_pr_search_cluster,
-    get_pr_search_similar,
+    get_pr_search_clusters,
+    get_pr_search_similar_lookup,
     get_pr_search_status,
-    probe_pr_search_live,
     run_pr_search_refresh,
 )
 
@@ -131,29 +130,33 @@ def create_app(settings: PrSearchApiSettings | None = None) -> FastAPI:
         number: int,
         request: Request,
         limit: int | None = None,
+        mode: Literal["auto", "indexed", "live"] = "auto",
     ) -> dict[str, Any]:
         settings = request.app.state.settings
         repo_slug = _repo_slug(settings, owner, repo)
-        return get_pr_search_similar(
+        return get_pr_search_similar_lookup(
             settings.index_path,
             repo=repo_slug,
             pr_number=number,
             limit=_limit(
                 limit, default=settings.similar_limit_default, maximum=settings.similar_limit_max
             ),
+            mode=mode,
+            client=_probe_client(settings),
         )
 
-    @app.get("/v1/repos/{owner}/{repo}/pulls/{number}/candidate-clusters")
-    async def pr_candidate_clusters(
+    @app.get("/v1/repos/{owner}/{repo}/pulls/{number}/clusters")
+    async def pr_clusters(
         owner: str,
         repo: str,
         number: int,
         request: Request,
         limit: int | None = None,
+        mode: Literal["auto", "indexed", "live"] = "auto",
     ) -> dict[str, Any]:
         settings = request.app.state.settings
         repo_slug = _repo_slug(settings, owner, repo)
-        return get_pr_search_candidate_clusters(
+        return get_pr_search_clusters(
             settings.index_path,
             repo=repo_slug,
             pr_number=number,
@@ -162,6 +165,8 @@ def create_app(settings: PrSearchApiSettings | None = None) -> FastAPI:
                 default=settings.candidate_limit_default,
                 maximum=settings.candidate_limit_max,
             ),
+            mode=mode,
+            client=_probe_client(settings),
         )
 
     @app.get("/v1/repos/{owner}/{repo}/clusters/{cluster_id}")
@@ -175,26 +180,6 @@ def create_app(settings: PrSearchApiSettings | None = None) -> FastAPI:
         repo_slug = _repo_slug(settings, owner, repo)
         return get_pr_search_cluster(settings.index_path, repo=repo_slug, cluster_id=cluster_id)
 
-    @app.get("/v1/repos/{owner}/{repo}/pulls/{number}/probe")
-    async def pr_probe(
-        owner: str,
-        repo: str,
-        number: int,
-        request: Request,
-        limit: int | None = None,
-    ) -> dict[str, Any]:
-        settings = request.app.state.settings
-        repo_slug = _repo_slug(settings, owner, repo)
-        return probe_pr_search_live(
-            settings.index_path,
-            repo=repo_slug,
-            pr_number=number,
-            limit=_limit(
-                limit, default=settings.probe_limit_default, maximum=settings.probe_limit_max
-            ),
-            client=_probe_client(settings),
-        )
-
     return app
 
 

src/slop_farmer/app/pr_search_client.py

@@ -10,9 +10,8 @@ from collections.abc import Callable
 from typing import Any
 
 from slop_farmer.app.pr_search import (
-    format_pr_search_candidate_clusters,
     format_pr_search_cluster,
-    format_pr_search_probe,
+    format_pr_search_clusters,
     format_pr_search_similar,
     format_pr_search_status,
 )
@@ -29,8 +28,8 @@ def build_parser() -> argparse.ArgumentParser:
             "Examples:\n"
             "  pr-search repo status\n"
             "  pr-search pr similar 67096\n"
-            "  pr-search pr candidates 67096 --limit 5\n"
-            "  pr-search pr probe 67096 --json\n"
+            "  pr-search pr clusters 67096\n"
+            "  pr-search --json pr similar 67096 --mode live\n"
             "  pr-search cluster view pr-scope-123-4\n"
             "  pr-search -R openclaw/openclaw repo status"
         ),
@@ -80,27 +79,31 @@ def build_parser() -> argparse.ArgumentParser:
 
     similar = pr_subparsers.add_parser(
         "similar",
-        help="Show similar indexed PRs.",
-        description="Find the nearest indexed pull requests for one PR number.",
+        help="Show similar PRs.",
+        description="Find similar pull requests for one PR number.",
     )
     similar.add_argument("number", type=int)
     similar.add_argument("--limit", type=int, default=None, help="Maximum rows to return.")
-
-    candidates = pr_subparsers.add_parser(
-        "candidates",
-        help="Show candidate clusters.",
-        description="Show candidate scope clusters for one indexed pull request.",
+    similar.add_argument(
+        "--mode",
+        choices=("auto", "indexed", "live"),
+        default="auto",
+        help="Lookup mode. Defaults to auto.",
     )
-    candidates.add_argument("number", type=int)
-    candidates.add_argument("--limit", type=int, default=None, help="Maximum rows to return.")
 
-    probe = pr_subparsers.add_parser(
-        "probe",
-        help="Probe a live PR against the active index.",
-        description="Fetch one live pull request and compare it against the active index.",
+    clusters = pr_subparsers.add_parser(
+        "clusters",
+        help="Show cluster context for a PR.",
+        description="Show assigned and candidate clusters for one PR number.",
+    )
+    clusters.add_argument("number", type=int)
+    clusters.add_argument("--limit", type=int, default=None, help="Maximum rows to return.")
+    clusters.add_argument(
+        "--mode",
+        choices=("auto", "indexed", "live"),
+        default="auto",
+        help="Lookup mode. Defaults to auto.",
     )
-    probe.add_argument("number", type=int)
-    probe.add_argument("--limit", type=int, default=None, help="Maximum rows to return.")
 
     cluster_parser = subparsers.add_parser(
         "cluster",
@@ -133,42 +136,37 @@ class PrSearchApiClient:
         *,
         number: int,
         limit: int | None,
+        mode: str,
     ) -> dict[str, Any]:
         owner, name = _split_repo(repo)
         return self._get_json(
             f"/v1/repos/{owner}/{name}/pulls/{number}/similar",
-            params=_limit_params(limit),
+            params=_lookup_params(limit, mode=mode),
         )
 
-    def get_candidate_clusters(
+    def get_clusters(
         self,
         repo: str,
         *,
         number: int,
         limit: int | None,
+        mode: str,
     ) -> dict[str, Any]:
         owner, name = _split_repo(repo)
         return self._get_json(
-            f"/v1/repos/{owner}/{name}/pulls/{number}/candidate-clusters",
-            params=_limit_params(limit),
+            f"/v1/repos/{owner}/{name}/pulls/{number}/clusters",
+            params=_lookup_params(limit, mode=mode),
         )
 
     def get_cluster(self, repo: str, *, cluster_id: str) -> dict[str, Any]:
         owner, name = _split_repo(repo)
         return self._get_json(f"/v1/repos/{owner}/{name}/clusters/{cluster_id}")
 
-    def probe(self, repo: str, *, number: int, limit: int | None) -> dict[str, Any]:
-        owner, name = _split_repo(repo)
-        return self._get_json(
-            f"/v1/repos/{owner}/{name}/pulls/{number}/probe",
-            params=_limit_params(limit),
-        )
-
     def _get_json(
         self,
         path: str,
         *,
-        params: dict[str, int] | None = None,
+        params: dict[str, int | str] | None = None,
     ) -> dict[str, Any]:
         query = f"?{urllib.parse.urlencode(params)}" if params else ""
         request = urllib.request.Request(f"{self.base_url}{path}{query}")
@@ -199,18 +197,22 @@ def main(argv: list[str] | None = None) -> None:
 
         if args.command == "pr":
             if args.pr_command == "similar":
-                result = client.get_similar(args.repo, number=args.number, limit=args.limit)
+                result = client.get_similar(
+                    args.repo,
+                    number=args.number,
+                    limit=args.limit,
+                    mode=args.mode,
+                )
                 _emit(result, args.json, format_pr_search_similar)
                 return
-            if args.pr_command == "candidates":
-                result = client.get_candidate_clusters(
-                    args.repo, number=args.number, limit=args.limit
+            if args.pr_command == "clusters":
+                result = client.get_clusters(
+                    args.repo,
+                    number=args.number,
+                    limit=args.limit,
+                    mode=args.mode,
                 )
-                _emit(result, args.json, format_pr_search_candidate_clusters)
-                return
-            if args.pr_command == "probe":
-                result = client.probe(args.repo, number=args.number, limit=args.limit)
-                _emit(result, args.json, format_pr_search_probe)
+                _emit(result, args.json, format_pr_search_clusters)
                 return
 
         if args.command == "cluster" and args.cluster_command == "view":
@@ -239,8 +241,11 @@ def _split_repo(repo: str) -> tuple[str, str]:
     return owner, name
 
 
-def _limit_params(limit: int | None) -> dict[str, int] | None:
-    return None if limit is None else {"limit": limit}
+def _lookup_params(limit: int | None, *, mode: str) -> dict[str, int | str]:
+    params: dict[str, int | str] = {"mode": mode}
+    if limit is not None:
+        params["limit"] = limit
+    return params
 
 
 def _error_detail(detail: str, *, fallback: str) -> str:

src/slop_farmer/reports/pr_search_service.py

@@ -287,6 +287,108 @@ def get_pr_search_candidate_clusters(
         connection.close()
 
 
+def get_pr_search_similar_lookup(
+    db_path: Path,
+    *,
+    pr_number: int,
+    repo: str | None = None,
+    limit: int = 10,
+    mode: str = "auto",
+    client: ProbeClientLike | None = None,
+) -> dict[str, Any]:
+    resolved_mode = _normalize_lookup_mode(mode)
+    if resolved_mode != "live":
+        try:
+            result = get_pr_search_similar(db_path, pr_number=pr_number, repo=repo, limit=limit)
+        except ValueError as exc:
+            if resolved_mode == "indexed" or not _is_index_miss(exc):
+                raise
+        else:
+            result["query"] = {
+                "pr_number": pr_number,
+                "mode_requested": resolved_mode,
+                "mode_used": "indexed",
+                "source": "active_index",
+            }
+            return result
+
+    live_result = probe_pr_search_live(
+        db_path,
+        pr_number=pr_number,
+        repo=repo,
+        limit=limit,
+        client=client,
+    )
+    return {
+        "repo": live_result["repo"],
+        "snapshot_id": live_result["snapshot_id"],
+        "run_id": live_result["run_id"],
+        "query": {
+            "pr_number": pr_number,
+            "mode_requested": resolved_mode,
+            "mode_used": "live",
+            "source": live_result["probe_source"]["provider"],
+        },
+        "pr": live_result["probe_pr"],
+        "probe_source": live_result["probe_source"],
+        "similar_prs": live_result["similar_prs"],
+    }
+
+
+def get_pr_search_clusters(
+    db_path: Path,
+    *,
+    pr_number: int,
+    repo: str | None = None,
+    limit: int = 5,
+    mode: str = "auto",
+    client: ProbeClientLike | None = None,
+) -> dict[str, Any]:
+    resolved_mode = _normalize_lookup_mode(mode)
+    if resolved_mode != "live":
+        try:
+            result = _get_pr_search_clusters_indexed(
+                db_path,
+                pr_number=pr_number,
+                repo=repo,
+                limit=limit,
+            )
+        except ValueError as exc:
+            if resolved_mode == "indexed" or not _is_index_miss(exc):
+                raise
+        else:
+            result["query"] = {
+                "pr_number": pr_number,
+                "mode_requested": resolved_mode,
+                "mode_used": "indexed",
+                "source": "active_index",
+            }
+            return result
+
+    live_result = probe_pr_search_live(
+        db_path,
+        pr_number=pr_number,
+        repo=repo,
+        limit=limit,
+        client=client,
+    )
+    return {
+        "repo": live_result["repo"],
+        "snapshot_id": live_result["snapshot_id"],
+        "run_id": live_result["run_id"],
+        "query": {
+            "pr_number": pr_number,
+            "mode_requested": resolved_mode,
+            "mode_used": "live",
+            "source": live_result["probe_source"]["provider"],
+        },
+        "pr": live_result["probe_pr"],
+        "probe_source": live_result["probe_source"],
+        "assigned_clusters": [],
+        "candidate_clusters": live_result["candidate_clusters"],
+    }
+
+
 def get_pr_search_cluster(
     db_path: Path,
     *,
@@ -584,6 +686,60 @@ def _scoped_rows(rows: list[dict[str, Any]], **extra: Any) -> list[dict[str, Any
     return [{**extra, **row} for row in rows]
 
 
+def _get_pr_search_clusters_indexed(
+    db_path: Path,
+    *,
+    pr_number: int,
+    repo: str | None = None,
+    limit: int = 5,
+) -> dict[str, Any]:
+    connection = connect_pr_search_db(db_path, read_only=True)
+    try:
+        active_run = resolve_active_run(connection, repo=repo)
+        run_id = str(active_run["id"])
+        document = _require_document(connection, run_id=run_id, pr_number=pr_number)
+        candidate_rows = get_candidate_cluster_rows(
+            connection,
+            run_id=run_id,
+            pr_number=pr_number,
+            limit=limit,
+        )
+        assigned_cluster_ids = get_cluster_ids_for_prs(
+            connection,
+            run_id=run_id,
+            pr_numbers=[pr_number],
+        ).get(pr_number, [])
+        assigned_clusters = []
+        for cluster_id in assigned_cluster_ids:
+            cluster = get_cluster(connection, run_id=run_id, cluster_id=cluster_id)
+            if cluster is None:
+                continue
+            assigned_clusters.append(_cluster_summary(cluster))
+        candidates = []
+        for row in candidate_rows:
+            evidence = _json_dict(row.get("evidence_json"))
+            candidates.append(
+                {
+                    **row,
+                    "shared_filenames": _json_list(row.get("shared_filenames_json")),
+                    "shared_directories": _json_list(row.get("shared_directories_json")),
+                    "evidence": evidence,
+                    "matched_member_pr_numbers": evidence.get("matched_member_pr_numbers") or [],
+                    "reason": evidence.get("reason") or "",
+                }
+            )
+        return {
+            "repo": active_run["repo"],
+            "snapshot_id": active_run["snapshot_id"],
+            "run_id": run_id,
+            "pr": document,
+            "assigned_clusters": assigned_clusters,
+            "candidate_clusters": candidates,
+        }
+    finally:
+        connection.close()
+
+
 def _require_document(connection: Any, *, run_id: str, pr_number: int) -> dict[str, Any]:
     document = get_document(connection, run_id=run_id, pr_number=pr_number)
     if document is None:
@@ -623,6 +779,25 @@ def _json_float_dict(raw: Any) -> dict[str, float]:
     return {str(key): float(value) for key, value in payload.items()}
 
 
+def _cluster_summary(cluster: dict[str, Any]) -> dict[str, Any]:
+    return {
+        **cluster,
+        "shared_filenames": _json_list(cluster.get("shared_filenames_json")),
+        "shared_directories": _json_list(cluster.get("shared_directories_json")),
+    }
+
+
+def _normalize_lookup_mode(mode: str) -> str:
+    normalized = mode.strip().lower()
+    if normalized not in {"auto", "indexed", "live"}:
+        raise ValueError(f"Unsupported mode {mode!r}; expected auto, indexed, or live.")
+    return normalized
+
+
+def _is_index_miss(exc: ValueError) -> bool:
+    return "active indexed universe" in str(exc)
+
+
 def _probe_source_metadata(
     client: Any,
     *,