fix(control): close the captcha governance loop (#65)

PR #63 added error_type='captcha_challenge' classification but nothing
consumed it: is_retryable() treated it as permanent (unknown types default
to permanent), so a Doubao captcha wall permanently failed the task with no
recovery path, while the control layer's pause/review machinery (actuator,
paused_until, review_required) had no scheduler consumer at all.

- error_taxonomy: add CAPTCHA_CHALLENGE + is_captcha() — a third category
  (needs human), neither retryable nor permanent
- error_kinds: map captcha_challenge -> ErrorKind.CAPTCHA for the control
  vocabulary
- actuator: add pause_source_for_captcha() — pause + require_review in one
  call, keeping the actuator the only mutator of DataSource
- pipeline: on captcha_challenge collect failure, pause the source (TTL from
  control_pause_ttl_seconds) + flag review instead of failing permanently;
  best-effort so actuator/DB errors never mask the original collect error
- scheduler: _get_enabled_schedules now also skips sources with
  review_required=True, so a paused/flagged source is never dispatched until
  a human clears it (control loop writes the state, scheduler honors it)

Tests: taxonomy (is_captcha/retryability), error_kinds mapping, actuator
pause+review (incl. idempotent re-pause), pipeline wiring (captcha pauses,
ordinary failures don't), scheduler review/disabled gating.

Co-authored-by: 1012839419a-alt <1012839419a-alt@users.noreply.github.com>

Author: 1012839419a-alt

backend/control/actuator.py

@@ -205,6 +205,30 @@ async def execute_action(
     )
 
 
+async def pause_source_for_captcha(
+    session: AsyncSession, *, source: DataSource, now: datetime, ttl_seconds: int
+) -> dict[str, Any]:
+    """Pause a source that hit a human-cleared challenge wall (captcha) and
+    flag it for review.
+
+    The pipeline calls this when a channel classifies a collect failure as
+    ``captcha_challenge`` (see ``backend.pipeline.error_taxonomy.is_captcha``):
+    automatic retry would burn budget on the same wall, so instead the source
+    is disabled for ``ttl_seconds`` (the normal pause TTL semantics — the
+    scheduler already skips disabled sources, and
+    :func:`auto_resume_expired_pauses` re-enables it when the wall should have
+    cooled down) and ``review_required`` is set so the UI surfaces it for a
+    human to confirm/clear.
+
+    Stays inside the actuator: this module remains the ONLY code allowed to
+    mutate a ``DataSource`` on the control system's behalf.
+    """
+    detail = await _apply_pause(session, source=source, now=now, ttl_seconds=ttl_seconds)
+    review_detail = await _apply_require_review(session, source=source)
+    detail["review_required"] = review_detail["already_flagged"] or True
+    return detail
+
+
 async def auto_resume_expired_pauses(
     session: AsyncSession, *, now: datetime
 ) -> list[tuple[DataSource, ExecutionResult]]:

backend/control/error_kinds.py

@@ -34,6 +34,9 @@ class ErrorKind(str, Enum):
     ODP_UNAVAILABLE = "odp_unavailable"
     STORE_FAILED = "store_failed"
     POISON_MESSAGE = "poison_message"
+    # Human-cleared challenge wall (Doubao captcha) — the controller should
+    # pause + require review, not backoff/retry on its own.
+    CAPTCHA = "captcha"
     UNKNOWN = "unknown"
 
 
@@ -95,6 +98,9 @@ class ErrorKind(str, Enum):
     # Poison message (DLQ-bound: a message that will never succeed no matter
     # how many times it's retried)
     "PoisonMessageError": ErrorKind.POISON_MESSAGE,
+    # Human-cleared challenge wall (doubao_research_channel's captcha
+    # classification — see error_taxonomy.CAPTCHA_CHALLENGE)
+    "captcha_challenge": ErrorKind.CAPTCHA,
 }
 
 

backend/pipeline/error_taxonomy.py

@@ -42,6 +42,24 @@
 })
 
 
+#: Explicit error_type for a site wall that only a human can clear (Doubao's
+#: captcha/人机验证, set by doubao_research_channel). Deliberately NOT in either
+#: set below: it is not a transient fault (retrying burns budget on the same
+#: wall) and not a permanent fault (the source is fine once a human clears it)
+#: — the pipeline treats it via :func:`is_captcha` (pause + require review).
+CAPTCHA_CHALLENGE = "captcha_challenge"
+
+
+def is_captcha(error_type: str | None) -> bool:
+    """True when the failure is a human-cleared challenge wall (captcha).
+
+    Distinct from retryability: a captcha is neither transient nor permanent —
+    the correct response is to pause the source and surface it for human
+    action, not to retry automatically.
+    """
+    return error_type == CAPTCHA_CHALLENGE
+
+
 def is_retryable(error_type: str | None) -> bool:
     """Classify a failure by its exception class name.
 

backend/pipeline/pipeline.py

@@ -13,7 +13,7 @@
 from backend.control.recorder import FreshnessInfo, record_run_measurement
 from backend.models.source import DataSource
 from backend.pipeline import events
-from backend.pipeline.error_taxonomy import effective_error_type, is_retryable
+from backend.pipeline.error_taxonomy import effective_error_type, is_captcha, is_retryable
 
 logger = logging.getLogger(__name__)
 
@@ -255,6 +255,48 @@ async def run_pipeline(
             )
         if is_retryable(channel_result.error_type):
             raise ChannelFetchError(channel_result.error or "collect failed")
+        if is_captcha(channel_result.error_type):
+            # Human-cleared challenge wall (Doubao captcha). Automatic retry
+            # would burn budget on the same wall and a permanent failure hides
+            # the recovery path, so instead pause the source (scheduler
+            # already skips disabled sources) and flag it for review — a human
+            # clears the wall, TTL expiry auto-resumes. Best-effort: a DB or
+            # actuator failure here must not mask the original collect error.
+            try:
+                from backend.config import get_settings
+                from backend.control.actuator import pause_source_for_captcha
+                from backend.database import AsyncSessionLocal
+
+                ttl = get_settings().control_pause_ttl_seconds
+                async with AsyncSessionLocal() as session:
+                    src = await session.get(DataSource, source.id)
+                    if src is not None:
+                        await pause_source_for_captcha(
+                            session,
+                            source=src,
+                            now=datetime.now(timezone.utc),
+                            ttl_seconds=ttl,
+                        )
+                        await session.commit()
+                        logger.warning(
+                            "[task:%s] captcha wall | paused source=%s (ttl=%ss, review_required)",
+                            task_id, source.id, ttl,
+                        )
+                        if run_id:
+                            await events.emit(
+                                run_id, "collect",
+                                "验证码拦截：数据源已暂停，等待人工处理",
+                                level="warning",
+                                detail={"captcha_paused": True, "pause_ttl_seconds": ttl},
+                            )
+            except Exception:
+                logger.exception("[task:%s] failed to pause source on captcha", task_id)
+            return PipelineResult(
+                success=False,
+                source_id=source.id,
+                error=channel_result.error,
+                metadata={"captcha_paused": True},
+            )
         if run_id:
             await _record_measurement_best_effort(
                 source_id=source.id, run_id=run_id,

backend/scheduler.py

@@ -29,6 +29,11 @@ async def _get_enabled_schedules() -> list[dict]:
             select(CronSchedule, DataSource)
             .join(DataSource, CronSchedule.source_id == DataSource.id)
             .where(CronSchedule.enabled.is_(True), DataSource.enabled.is_(True))
+            # A source flagged review_required (e.g. by a captcha pause — see
+            # backend.pipeline.pipeline's captcha branch) must not be
+            # dispatched until a human clears the flag; the control loop
+            # writes the state, the scheduler honors it.
+            .where(DataSource.review_required.is_(False))
         )
         return [
             {

tests/unit/control/test_actuator.py

@@ -264,3 +264,38 @@ async def test_dangerous_suggestion_downgrades_and_preserves_original(db_session
     # Source was NOT paused or otherwise mutated beyond the review flag —
     # the downgrade never performs the originally-suggested action.
     assert source.enabled is True
+
+
+# ── pause_source_for_captcha (captcha governance, PR-captcha-governance) ──
+
+
+@pytest.mark.asyncio
+async def test_pause_source_for_captcha_pauses_and_flags_review(db_session):
+    source = await _make_source(db_session)
+
+    detail = await actuator.pause_source_for_captcha(
+        db_session, source=source, now=NOW, ttl_seconds=900
+    )
+
+    assert source.enabled is False
+    assert source.paused_until == NOW + timedelta(seconds=900)
+    assert source.review_required is True
+    assert detail["paused_until"] == source.paused_until.isoformat()
+    assert detail["review_required"] is True
+    assert detail["was_enabled"] is True
+
+
+@pytest.mark.asyncio
+async def test_pause_source_for_captcha_refreshes_ttl_when_already_paused(db_session):
+    source = await _make_source(db_session, enabled=False)
+    source.paused_until = NOW - timedelta(seconds=1)
+    source.review_required = True
+    await db_session.flush()
+
+    detail = await actuator.pause_source_for_captcha(
+        db_session, source=source, now=NOW, ttl_seconds=1800
+    )
+
+    assert source.paused_until == NOW + timedelta(seconds=1800)
+    assert source.review_required is True
+    assert detail["was_enabled"] is False

tests/unit/control/test_error_kinds.py

@@ -45,6 +45,9 @@ def test_schema_drift(self):
     def test_store_failed(self):
         assert map_error_type("IntegrityError") is ErrorKind.STORE_FAILED
 
+    def test_captcha_challenge_maps_to_captcha(self):
+        assert map_error_type("captcha_challenge") is ErrorKind.CAPTCHA
+
 
 class TestMapException:
     def test_none_maps_to_unknown(self):

tests/unit/pipeline/test_error_taxonomy.py

@@ -2,7 +2,12 @@
 
 import pytest
 
-from backend.pipeline.error_taxonomy import is_retryable, is_retryable_http_status
+from backend.pipeline.error_taxonomy import (
+    CAPTCHA_CHALLENGE,
+    is_captcha,
+    is_retryable,
+    is_retryable_http_status,
+)
 
 
 @pytest.mark.parametrize("error_type", [
@@ -60,3 +65,33 @@ def test_request_timeout_408_is_retryable():
     """408 is a transient per-request timeout, not a durably broken request —
     it belongs with 429/5xx, not with the permanent 4xx family."""
     assert is_retryable_http_status(408) is True
+
+
+# ── captcha_challenge: third category (needs human, not retry/permanent) ─────
+
+
+def test_captcha_challenge_constant_value():
+    assert CAPTCHA_CHALLENGE == "captcha_challenge"
+
+
+def test_captcha_challenge_is_not_retryable():
+    """A captcha wall is not a transient fault — retrying immediately burns
+    retry budget on a wall that only a human can clear."""
+    assert is_retryable(CAPTCHA_CHALLENGE) is False
+
+
+def test_captcha_challenge_is_captcha():
+    assert is_captcha(CAPTCHA_CHALLENGE) is True
+
+
+def test_none_is_not_captcha():
+    assert is_captcha(None) is False
+
+
+def test_empty_string_is_not_captcha():
+    assert is_captcha("") is False
+
+
+def test_ordinary_errors_are_not_captcha():
+    for t in ("TimeoutException", "ValueError", "RetryableHTTPStatus", "SomeNewError"):
+        assert is_captcha(t) is False

tests/unit/pipeline/test_pipeline_errors.py

@@ -61,6 +61,89 @@ async def test_pipeline_collect_exception(db_session):
     assert "network down" in result.error
 
 
+# ── captcha governance wiring (PR-captcha-governance) ────────────────────
+
+
+@pytest.mark.asyncio
+async def test_pipeline_captcha_failure_pauses_source_for_review(db_session):
+    """A collect failure classified as captcha_challenge pauses the source
+    (enabled=False + review_required=True) instead of failing permanently or
+    retrying automatically."""
+    from backend.models.source import DataSource
+    from backend.models.task import CollectionTask
+
+    source = DataSource(
+        name="Captcha Source",
+        channel_type="doubao_research",
+        channel_config={"question": "x"},
+    )
+    db_session.add(source)
+    await db_session.flush()
+
+    task = CollectionTask(source_id=source.id, trigger_type="manual", parameters={})
+    db_session.add(task)
+    await db_session.flush()
+
+    channel_result = ChannelResult.fail("verification challenge", error_type="captcha_challenge")
+
+    mock_session = AsyncMock()
+    mock_session.get = AsyncMock(return_value=source)
+    mock_session.commit = AsyncMock()
+    mock_session_cm = AsyncMock()
+    mock_session_cm.__aenter__ = AsyncMock(return_value=mock_session)
+    mock_session_cm.__aexit__ = AsyncMock(return_value=False)
+
+    with (
+        patch("backend.pipeline.collector.collect", return_value=channel_result),
+        patch("backend.database.AsyncSessionLocal", return_value=mock_session_cm),
+        patch("backend.control.actuator.pause_source_for_captcha", new_callable=AsyncMock) as mock_pause,
+        patch("backend.config.get_settings") as mock_settings,
+    ):
+        mock_settings.return_value.control_pause_ttl_seconds = 900
+        result = await run_pipeline(task.id, source)
+
+    assert result.success is False
+    assert "verification challenge" in result.error
+    assert result.metadata.get("captcha_paused") is True
+    mock_pause.assert_awaited_once()
+    assert mock_pause.await_args.kwargs["source"].id == source.id
+    assert mock_pause.await_args.kwargs["ttl_seconds"] == 900
+    assert source.enabled is True  # the real pause happens in the actuator via the mocked call
+
+
+@pytest.mark.asyncio
+async def test_pipeline_ordinary_failure_does_not_pause_source(db_session):
+    """Non-captcha failures keep the existing permanent-failure path and never
+    touch the actuator."""
+    from backend.models.source import DataSource
+    from backend.models.task import CollectionTask
+
+    source = DataSource(
+        name="Plain Fail Source",
+        channel_type="rss",
+        channel_config={"feed_url": "https://ex.com/feed.xml"},
+    )
+    db_session.add(source)
+    await db_session.flush()
+
+    task = CollectionTask(source_id=source.id, trigger_type="manual", parameters={})
+    db_session.add(task)
+    await db_session.flush()
+
+    channel_result = ChannelResult.fail("feed malformed", error_type="JSONDecodeError")
+
+    with (
+        patch("backend.pipeline.collector.collect", return_value=channel_result),
+        patch("backend.control.actuator.pause_source_for_captcha", new_callable=AsyncMock) as mock_pause,
+    ):
+        result = await run_pipeline(db_session, source, task.id)
+
+    assert result.success is False
+    assert "feed malformed" in result.error
+    assert "captcha_paused" not in (result.metadata or {})
+    mock_pause.assert_not_awaited()
+
+
 @pytest.mark.asyncio
 async def test_pipeline_with_ai_failure_still_returns_success(db_session):
     from backend.models.source import DataSource