feat(robomp): added incoming PR review feature with classify and submit

- Added `review_pr` task that checks out PR head in a detached worktree, classifies rank/type/area, and posts a batched GitHub review as `event=COMMENT`.
- Added four new host tools: `fetch_pr`, `classify_pr`, `pr_review_comment`, and `submit_pr_review`; review tools self-gate on `review_mode`, push/open-PR tools refuse when `review_mode` is set.
- Added sqlite staging table `pr_review_comments` with `stage_review_comment`, `list_staged_review_comments`, and `clear_staged_review_comments` DAOs.
- Routed `pull_request.opened/reopened/ready_for_review` to `review_pr` and extended `pull_request.closed` cleanup to any tracked PR regardless of author.

python/robomp/tests/test_worker.py

@@ -659,10 +659,74 @@ async def test_run_rpc_skips_reminder_when_unclassified(tmp_path: Path, settings
     assert len(fake.prompts) == 1
 
 
+@pytest.mark.asyncio
+async def test_run_rpc_review_pr_reminds_until_submit_pr_review(tmp_path: Path, settings: Settings) -> None:
+    inputs, bindings = _make_inputs(tmp_path, settings, session_has_jsonl=False)
+    loop = asyncio.new_event_loop()
+    try:
+        worker._run_rpc_blocking(
+            inputs,
+            task_kind="review_pr",
+            prompt="kickoff",
+            loop=loop,
+            bindings=bindings,  # type: ignore[arg-type]
+        )
+    finally:
+        loop.close()
+    fake = _FakeRpcClient.instances[0]
+    assert len(fake.prompts) == 1 + settings.task_completion_max_reminders
+    assert fake.prompts[0] == "kickoff"
+    assert all("submit_pr_review" in p for p in fake.prompts[1:])
+    assert all("gh_open_pr" not in p for p in fake.prompts[1:])
+
+
+@pytest.mark.asyncio
+async def test_run_rpc_review_pr_stops_after_submit_without_dirty_probe(
+    tmp_path: Path, settings: Settings, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    inputs, bindings = _make_inputs(tmp_path, settings, session_has_jsonl=False)
+
+    def _probe(_workspace, _slot_uid):  # type: ignore[no-untyped-def]
+        raise AssertionError("review_pr must not run dirty-state probes")
+
+    monkeypatch.setattr(worker, "_probe_workspace_dirty", _probe)
+    original_on_tool_end = _FakeRpcClient.on_tool_execution_end
+
+    def _record_tool_end(self, cb) -> None:
+        self._tool_end_callbacks = getattr(self, "_tool_end_callbacks", [])
+        self._tool_end_callbacks.append(cb)
+
+    def _on_prompt(client: _FakeRpcClient, _prompt: str) -> None:
+        for cb in client._tool_end_callbacks:
+            cb(SimpleNamespace(tool_name="submit_pr_review", result={}))
+
+    _FakeRpcClient.on_tool_execution_end = _record_tool_end  # type: ignore[assignment]
+    try:
+        _FakeRpcClient.on_prompt = staticmethod(_on_prompt)  # type: ignore[attr-defined]
+        loop = asyncio.new_event_loop()
+        try:
+            worker._run_rpc_blocking(
+                inputs,
+                task_kind="review_pr",
+                prompt="kickoff",
+                loop=loop,
+                bindings=bindings,  # type: ignore[arg-type]
+            )
+        finally:
+            loop.close()
+    finally:
+        _FakeRpcClient.on_tool_execution_end = original_on_tool_end  # type: ignore[assignment]
+        delattr(_FakeRpcClient, "on_prompt")
+
+    fake = _FakeRpcClient.instances[0]
+    assert fake.prompts == ["kickoff"]
+
+
 # ---------------------------------------------------------------------------
 # Dirty-state watchdog
 # ---------------------------------------------------------------------------
 
+
 @pytest.mark.asyncio
 async def test_run_rpc_sends_dirty_state_reminder_when_worktree_has_unpushed_work(
     tmp_path: Path, settings: Settings, monkeypatch: pytest.MonkeyPatch