feat(orchestrator): inbound external-PR discovery + review-task ingestion

Add the dormant inbound path for external-PR review (gated by external_pr_enabled,
off by default):

- GitService.list_open_prs lists a project's open PRs, normalized with fork /
  author-association classification (the inbound counterpart to the org's
  outbound, head-filtered PR calls).
- TaskService.ingest_external_pr + external_review_task_exists create one
  de-duped review task per newly-seen external PR (source='external_pr',
  confirmed_by_human=False) — a gate so no agent fetches or runs contributor
  code until a human confirms the PR.
- A poll loop in the orchestrator, mirroring the strategy-engine loop: only when
  enabled it lists each active project's open PRs, ingests the external ones, and
  wakes the dispatcher.

The trust-critical author/fork classifier is unit-tested; the GitHub-list and
DB-ingest paths are exercised by the integration gate.

roboco/runtime/orchestrator.py

@@ -21,7 +21,7 @@ import tempfile
 from dataclasses import dataclass
 from datetime import UTC, datetime, timedelta
 from pathlib import Path
-from typing import TYPE_CHECKING, Any, ClassVar
+from typing import TYPE_CHECKING, Any, ClassVar, cast
 
 import httpx
 
@@ -29,6 +29,8 @@ if TYPE_CHECKING:
     from collections.abc import Callable, Coroutine
     from uuid import UUID
 
+    from sqlalchemy.ext.asyncio import AsyncSession
+
     from roboco.services.llm import AgentRoute
     from roboco.services.task import TaskService
 import structlog
@@ -595,6 +597,7 @@ class AgentOrchestrator:
         # rate-limited providers and resolves waiting agents on success.
         self._rate_limit_probe_task: asyncio.Task | None = None
         self._strategy_engine_task: asyncio.Task | None = None
+        self._external_pr_poll_task: asyncio.Task | None = None
         # Tracks which providers have already received a CEO notification
         # during the current rate-limit episode.  Cleared when the probe
         # succeeds and the rate limit is lifted (tracker.clear() path).
@@ -673,6 +676,7 @@ class AgentOrchestrator:
         self._sweeper_task = asyncio.create_task(self._sweeper_loop())
         self._rate_limit_probe_task = asyncio.create_task(self._rate_limit_probe_loop())
         self._strategy_engine_task = asyncio.create_task(self._strategy_engine_loop())
+        self._external_pr_poll_task = asyncio.create_task(self._external_pr_poll_loop())
 
         logger.info(
             "Orchestrator started",
@@ -710,6 +714,11 @@ class AgentOrchestrator:
             with contextlib.suppress(asyncio.CancelledError):
                 await self._strategy_engine_task
 
+        if self._external_pr_poll_task:
+            self._external_pr_poll_task.cancel()
+            with contextlib.suppress(asyncio.CancelledError):
+                await self._external_pr_poll_task
+
         # Stop all agents
         for agent_id in list(self._instances.keys()):
             await self.stop_agent(agent_id)
@@ -4519,6 +4528,72 @@ Start by:
             except Exception:
                 logger.exception("strategy engine cycle failed")
 
+    async def _external_pr_poll_loop(self) -> None:
+        """Engine 3: discover inbound external PRs and open review tasks.
+
+        Dormant by default — returns immediately unless ``external_pr_enabled``,
+        so a standard deployment makes no inbound GitHub call. This only lists
+        open PRs and records a review task per newly-seen external one; it never
+        fetches or runs contributor code (that waits on a human confirmation
+        downstream). New review tasks wake the dispatcher.
+        """
+        if not settings.external_pr_enabled:
+            return
+        from roboco.db import get_db_context
+
+        interval = settings.external_pr_poll_interval_seconds
+        while self._running:
+            try:
+                await asyncio.sleep(interval)
+                async with get_db_context() as db:
+                    ingested = await self._poll_external_prs_once(db)
+                if ingested:
+                    self._dispatch_wake.set()
+            except asyncio.CancelledError:
+                break
+            except Exception:
+                logger.exception("external-PR poll cycle failed")
+
+    async def _poll_external_prs_once(self, db: "AsyncSession") -> int:
+        """One discovery pass across active projects; returns tasks ingested.
+
+        Lists each active project's open PRs, keeps the external ones, and
+        ingests a de-duped review task for each. Commits once at the end.
+        """
+        from roboco.services.git import GitService
+        from roboco.services.project import get_project_service
+        from roboco.services.task import get_task_service
+
+        git = GitService(db)
+        task_service = get_task_service(db)
+        projects = await get_project_service(db).list_all(active_only=True)
+        system_id = _foundation.AGENTS["system"].uuid
+        ingested = 0
+        for project in projects:
+            for pr in await git.list_open_prs(project.slug):
+                number = pr.get("number")
+                if number is None or not self._is_external_pr(pr):
+                    continue
+                created = await task_service.ingest_external_pr(
+                    project_id=cast("UUID", project.id),
+                    pr=pr,
+                    created_by=system_id,
+                    team=Team.BOARD,
+                )
+                if created is not None:
+                    ingested += 1
+        await db.commit()
+        return ingested
+
+    @staticmethod
+    def _is_external_pr(pr: dict[str, Any]) -> bool:
+        """A PR the org did not author: a fork head or a non-member author."""
+        if pr.get("is_fork"):
+            return True
+        trusted = {"OWNER", "MEMBER", "COLLABORATOR"}
+        assoc = (pr.get("author_association") or "").upper()
+        return assoc not in trusted
+
     async def _rate_limit_probe_loop(self) -> None:
         """Background loop: probe rate-limited providers every ~30 seconds.
 

roboco/services/git.py

@@ -1370,6 +1370,72 @@ class GitService(BaseService):
             return cast("dict[str, Any]", existing.json()[0])
         return None
 
+    async def list_open_prs(self, project_slug: str) -> list[dict[str, Any]]:
+        """List a project's open PRs, normalized with fork/author classification.
+
+        The inbound counterpart to the org's outbound PR calls: lists ALL open
+        PRs (no ``head=`` filter), so it sees external/fork contributions the org
+        did not create. Each record carries ``number``, ``url``, ``title``,
+        ``head_ref``, ``is_fork`` (head repo differs from base repo),
+        ``user_login`` and ``author_association`` so the caller can classify
+        trust. Returns ``[]`` on a missing token, unparseable remote, or any
+        GitHub error — it never raises into the poll loop.
+        """
+        project = await get_project_service(self.session).get_by_slug(project_slug)
+        if project is None or not project.git_url:
+            return []
+        try:
+            owner, repo = self._parse_git_url(project.git_url)
+        except GitError:
+            return []
+        git_token = await self._token_for_project(project_slug)
+        if not git_token:
+            return []
+        api_base = settings.github_api_base_url.rstrip("/")
+        try:
+            async with httpx.AsyncClient(timeout=_default_git_timeout()) as client:
+                resp = await client.get(
+                    f"{api_base}/repos/{owner}/{repo}/pulls",
+                    headers={
+                        "Authorization": f"Bearer {git_token}",
+                        "Accept": "application/vnd.github+json",
+                        "X-GitHub-Api-Version": "2022-11-28",
+                    },
+                    params={"state": "open", "per_page": 100},
+                )
+        except httpx.HTTPError as e:
+            self.log.warning(
+                "list_open_prs request failed",
+                project=project_slug,
+                error=str(e),
+            )
+            return []
+        if not resp.is_success:
+            self.log.warning(
+                "list_open_prs non-2xx",
+                project=project_slug,
+                status=resp.status_code,
+            )
+            return []
+        base_full = f"{owner}/{repo}"
+        prs: list[dict[str, Any]] = []
+        for pr in resp.json():
+            head = pr.get("head") or {}
+            head_repo = head.get("repo") or {}
+            head_full = head_repo.get("full_name")
+            prs.append(
+                {
+                    "number": pr.get("number"),
+                    "url": pr.get("html_url") or "",
+                    "title": pr.get("title") or "",
+                    "head_ref": head.get("ref"),
+                    "is_fork": bool(head_full and head_full != base_full),
+                    "user_login": (pr.get("user") or {}).get("login"),
+                    "author_association": pr.get("author_association"),
+                }
+            )
+        return prs
+
     async def _post_pr(
         self,
         owner: str,

roboco/services/task.py

@@ -36,6 +36,7 @@ from roboco.models.base import (
     AgentRole,
     AgentStatus,
     BlockerResolverType,
+    Complexity,
     JournalEntryType,
     TaskNature,
     TaskStatus,
@@ -602,6 +603,75 @@ class TaskService(BaseService):
         )
         return task
 
+    async def external_review_task_exists(
+        self, project_id: UUID, pr_number: int
+    ) -> bool:
+        """True if a review task already exists for this (project, external PR).
+
+        The de-dupe key for inbound external-PR ingestion: one review task per
+        ``(project_id, source='external_pr', pr_number)`` so re-polling an open
+        PR never creates a duplicate.
+        """
+        result = await self.session.execute(
+            select(TaskTable.id).where(
+                TaskTable.project_id == project_id,
+                TaskTable.source == "external_pr",
+                TaskTable.pr_number == pr_number,
+            )
+        )
+        return result.first() is not None
+
+    async def ingest_external_pr(
+        self,
+        *,
+        project_id: UUID,
+        pr: dict[str, Any],
+        created_by: UUID,
+        team: Team,
+    ) -> TaskTable | None:
+        """Create one review task for a newly-seen external PR; ``None`` if it exists.
+
+        ``pr`` is a normalized record from ``GitService.list_open_prs`` (number,
+        url, title). De-duped on ``(project_id, source='external_pr', pr_number)``.
+        The task is CODE-typed with ``source='external_pr'`` and
+        ``confirmed_by_human=False`` — a deliberate gate: no agent fetches, checks
+        out, or runs the contributor's code until a human confirms the PR. Caller
+        commits.
+        """
+        pr_number = int(pr["number"])
+        pr_url = str(pr.get("url") or "")
+        pr_title = str(pr.get("title") or "")
+        if await self.external_review_task_exists(project_id, pr_number):
+            return None
+        title = f"Review external PR #{pr_number}: {pr_title}".strip()
+        req = TaskCreateRequest(
+            title=title[:200],
+            description=(
+                f"An external contributor opened PR #{pr_number} ({pr_url}).\n\n"
+                "Review it adversarially and post a single, complete "
+                "change-request with per-criterion findings. Do not fetch, check "
+                "out, or run the contributor's code until a human has confirmed "
+                "this PR."
+            ),
+            acceptance_criteria=[
+                "Exactly one complete GitHub review is posted with per-criterion "
+                "findings",
+            ],
+            team=team,
+            created_by=created_by,
+            task_type=TaskType.CODE,
+            nature=TaskNature.TECHNICAL,
+            estimated_complexity=Complexity.MEDIUM,
+            project_id=project_id,
+            source="external_pr",
+            confirmed_by_human=False,
+        )
+        task = await self.create(req)
+        task.pr_number = pr_number
+        task.pr_url = pr_url
+        await self.session.flush()
+        return task
+
     async def _inherit_parent_session(
         self,
         task_id: UUID,

tests/unit/runtime/test_external_pr_classify.py

@@ -0,0 +1,34 @@
+"""External-PR author/fork classification — the inbound trust decision.
+
+``_is_external_pr`` decides whether an open PR was authored by the org itself
+or by an outside contributor. It must default to *external* (the cautious
+side) for anything it does not positively recognize as internal.
+"""
+
+from __future__ import annotations
+
+import pytest
+from roboco.runtime.orchestrator import AgentOrchestrator
+
+
+@pytest.mark.parametrize(
+    ("pr", "expected"),
+    [
+        # A fork head is always external, regardless of association.
+        ({"is_fork": True, "author_association": "OWNER"}, True),
+        # Same-repo branch from a trusted association is internal (the org).
+        ({"is_fork": False, "author_association": "OWNER"}, False),
+        ({"is_fork": False, "author_association": "MEMBER"}, False),
+        ({"is_fork": False, "author_association": "COLLABORATOR"}, False),
+        ({"is_fork": False, "author_association": "member"}, False),  # case-insensitive
+        # Outside associations are external even on a same-repo branch.
+        ({"is_fork": False, "author_association": "CONTRIBUTOR"}, True),
+        ({"is_fork": False, "author_association": "FIRST_TIME_CONTRIBUTOR"}, True),
+        ({"is_fork": False, "author_association": "NONE"}, True),
+        ({"is_fork": False, "author_association": None}, True),
+        # Unknown/empty shape defaults to external (cautious).
+        ({}, True),
+    ],
+)
+def test_is_external_pr(pr: dict[str, object], *, expected: bool) -> None:
+    assert AgentOrchestrator._is_external_pr(pr) is expected