From 5b264c052bb3419bf1414c2eb4f671737f083d7a Mon Sep 17 00:00:00 2001
From: "Matheus Henrique(Aka: TheusHen)" <mmmatheushenriqueee@gmail.com>
Date: Mon, 13 Jul 2026 14:06:04 -0300
Subject: [PATCH 1/8] Add torrent acquisition integrations

---
 .env.example                                  |   4 +
 .../versions/d4e5f6a7b8c9_add_acquisition.py  |  79 +++++
 papyrus/api/routes/__init__.py                |   2 +
 papyrus/api/routes/acquisition.py             | 209 +++++++++++
 papyrus/config.py                             |   1 +
 papyrus/main.py                               |  22 +-
 papyrus/models/__init__.py                    |   4 +
 papyrus/models/acquisition.py                 |  63 ++++
 papyrus/schemas/acquisition.py                | 116 +++++++
 papyrus/services/acquisition.py               | 327 ++++++++++++++++++
 tests/api/routes/test_acquisition.py          |  59 ++++
 11 files changed, 882 insertions(+), 4 deletions(-)
 create mode 100644 alembic/versions/d4e5f6a7b8c9_add_acquisition.py
 create mode 100644 papyrus/api/routes/acquisition.py
 create mode 100644 papyrus/models/acquisition.py
 create mode 100644 papyrus/schemas/acquisition.py
 create mode 100644 papyrus/services/acquisition.py
 create mode 100644 tests/api/routes/test_acquisition.py

diff --git a/.env.example b/.env.example
index 9ce5d9e..d60a99b 100644
--- a/.env.example
+++ b/.env.example
@@ -53,6 +53,10 @@ DEV_PAGES_USE_VITE=true
 DEV_PAGES_VITE_URL=http://localhost:5173
 DEV_PAGES_MANIFEST_PATH=frontend/dev-pages/dist/.vite/manifest.json
 
+# Execute enabled acquisition rules on this interval (seconds). Set a value that
+# respects each indexer's API limits.
+ACQUISITION_AUTOMATION_INTERVAL_SECONDS=300
+
 # Prefer file-based PowerSync keys for local development.
 # Generate them with: ./scripts/generate_dev_powersync_keys.sh
 POWERSYNC_JWT_PRIVATE_KEY_FILE=.local/powersync/private.pem
diff --git a/alembic/versions/d4e5f6a7b8c9_add_acquisition.py b/alembic/versions/d4e5f6a7b8c9_add_acquisition.py
new file mode 100644
index 0000000..9a785c2
--- /dev/null
+++ b/alembic/versions/d4e5f6a7b8c9_add_acquisition.py
@@ -0,0 +1,79 @@
+"""add acquisition endpoints, rules and jobs
+
+Revision ID: d4e5f6a7b8c9
+Revises: c3f8b2a9d1e4
+"""
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from sqlalchemy.dialects import postgresql
+
+from alembic import op
+
+revision: str = "d4e5f6a7b8c9"
+down_revision: str | Sequence[str] | None = "c3f8b2a9d1e4"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    """Create private acquisition integration tables."""
+    op.create_table(
+        "acquisition_endpoints",
+        sa.Column("endpoint_id", sa.Uuid(), nullable=False),
+        sa.Column("owner_user_id", sa.Uuid(), nullable=False),
+        sa.Column("name", sa.String(120), nullable=False),
+        sa.Column("kind", sa.String(32), nullable=False),
+        sa.Column("base_url", sa.String(2048), nullable=False),
+        sa.Column("credentials", postgresql.JSONB(), nullable=True),
+        sa.Column("settings", postgresql.JSONB(), nullable=True),
+        sa.Column("enabled", sa.Boolean(), server_default=sa.text("true"), nullable=False),
+        sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False),
+        sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False),
+        sa.ForeignKeyConstraint(["owner_user_id"], ["users.user_id"], ondelete="CASCADE"),
+        sa.PrimaryKeyConstraint("endpoint_id"),
+    )
+    op.create_index("ix_acquisition_endpoints_owner_user_id", "acquisition_endpoints", ["owner_user_id"])
+    op.create_table(
+        "acquisition_rules",
+        sa.Column("rule_id", sa.Uuid(), nullable=False),
+        sa.Column("owner_user_id", sa.Uuid(), nullable=False),
+        sa.Column("name", sa.String(120), nullable=False),
+        sa.Column("query", sa.String(500), nullable=False),
+        sa.Column("endpoint_ids", postgresql.JSONB(), nullable=True),
+        sa.Column("download_client_id", sa.Uuid(), nullable=True),
+        sa.Column("filters", postgresql.JSONB(), nullable=True),
+        sa.Column("enabled", sa.Boolean(), server_default=sa.text("true"), nullable=False),
+        sa.Column("last_run_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False),
+        sa.ForeignKeyConstraint(["download_client_id"], ["acquisition_endpoints.endpoint_id"]),
+        sa.ForeignKeyConstraint(["owner_user_id"], ["users.user_id"], ondelete="CASCADE"),
+        sa.PrimaryKeyConstraint("rule_id"),
+    )
+    op.create_index("ix_acquisition_rules_owner_user_id", "acquisition_rules", ["owner_user_id"])
+    op.create_table(
+        "acquisition_jobs",
+        sa.Column("job_id", sa.Uuid(), nullable=False),
+        sa.Column("owner_user_id", sa.Uuid(), nullable=False),
+        sa.Column("endpoint_id", sa.Uuid(), nullable=False),
+        sa.Column("rule_id", sa.Uuid(), nullable=True),
+        sa.Column("title", sa.String(500), nullable=False),
+        sa.Column("download_url", sa.Text(), nullable=False),
+        sa.Column("status", sa.String(32), server_default=sa.text("'queued'"), nullable=False),
+        sa.Column("client_reference", sa.String(255), nullable=True),
+        sa.Column("error", sa.Text(), nullable=True),
+        sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False),
+        sa.ForeignKeyConstraint(["endpoint_id"], ["acquisition_endpoints.endpoint_id"]),
+        sa.ForeignKeyConstraint(["owner_user_id"], ["users.user_id"], ondelete="CASCADE"),
+        sa.ForeignKeyConstraint(["rule_id"], ["acquisition_rules.rule_id"], ondelete="SET NULL"),
+        sa.PrimaryKeyConstraint("job_id"),
+    )
+    op.create_index("ix_acquisition_jobs_owner_user_id", "acquisition_jobs", ["owner_user_id"])
+
+
+def downgrade() -> None:
+    """Drop acquisition tables."""
+    op.drop_table("acquisition_jobs")
+    op.drop_table("acquisition_rules")
+    op.drop_table("acquisition_endpoints")
diff --git a/papyrus/api/routes/__init__.py b/papyrus/api/routes/__init__.py
index 7835593..19772d8 100644
--- a/papyrus/api/routes/__init__.py
+++ b/papyrus/api/routes/__init__.py
@@ -3,6 +3,7 @@
 from fastapi import APIRouter, FastAPI
 
 from papyrus.api.routes import (
+    acquisition,
     annotations,
     auth,
     bookmarks,
@@ -26,6 +27,7 @@
 
 api_router = APIRouter()
 
+api_router.include_router(acquisition.router, prefix="/acquisition", tags=["Acquisition"])
 api_router.include_router(auth.router, prefix="/auth", tags=["Auth"])
 api_router.include_router(users.router, prefix="/users", tags=["Users"])
 api_router.include_router(books.router, prefix="/books", tags=["Books"])
diff --git a/papyrus/api/routes/acquisition.py b/papyrus/api/routes/acquisition.py
new file mode 100644
index 0000000..a5a0ae2
--- /dev/null
+++ b/papyrus/api/routes/acquisition.py
@@ -0,0 +1,209 @@
+"""Private torrent and indexer acquisition endpoints."""
+
+from typing import Annotated
+from uuid import UUID
+
+from fastapi import APIRouter, Depends, HTTPException, status
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from papyrus.api.deps import CurrentUserId
+from papyrus.core.database import get_db
+from papyrus.models.acquisition import AcquisitionEndpoint as AcquisitionEndpointModel
+from papyrus.models.acquisition import AcquisitionJob as AcquisitionJobModel
+from papyrus.models.acquisition import AcquisitionRule as AcquisitionRuleModel
+from papyrus.schemas.acquisition import (
+    AcquisitionEndpoint,
+    AcquisitionEndpointCreate,
+    AcquisitionEndpointUpdate,
+    AcquisitionJob,
+    AcquisitionRule,
+    AcquisitionRuleCreate,
+    ArrCommandRequest,
+    Release,
+    SearchRequest,
+    SubmitRequest,
+)
+from papyrus.services.acquisition import (
+    dispatch_arr_command,
+    owned_endpoint,
+    run_rule,
+    search_endpoint,
+    submit_to_client,
+)
+
+router = APIRouter()
+DbSession = Annotated[AsyncSession, Depends(get_db)]
+
+
+def _credentials(api_key: str | None, username: str | None, password: str | None) -> dict[str, str] | None:
+    values = {
+        key: value for key, value in {"api_key": api_key, "username": username, "password": password}.items() if value
+    }
+    return values or None
+
+
+@router.get("/endpoints", response_model=list[AcquisitionEndpoint])
+async def list_endpoints(user_id: CurrentUserId, db: DbSession) -> list[AcquisitionEndpointModel]:
+    result = await db.execute(
+        select(AcquisitionEndpointModel)
+        .where(AcquisitionEndpointModel.owner_user_id == user_id)
+        .order_by(AcquisitionEndpointModel.name)
+    )
+    return list(result.scalars())
+
+
+@router.post("/endpoints", response_model=AcquisitionEndpoint, status_code=status.HTTP_201_CREATED)
+async def create_endpoint(
+    user_id: CurrentUserId, request: AcquisitionEndpointCreate, db: DbSession
+) -> AcquisitionEndpointModel:
+    endpoint = AcquisitionEndpointModel(
+        owner_user_id=user_id,
+        name=request.name,
+        kind=request.kind.value,
+        base_url=str(request.base_url),
+        credentials=_credentials(
+            request.api_key.get_secret_value() if request.api_key else None,
+            request.username.get_secret_value() if request.username else None,
+            request.password.get_secret_value() if request.password else None,
+        ),
+        settings=request.settings,
+    )
+    db.add(endpoint)
+    await db.commit()
+    await db.refresh(endpoint)
+    return endpoint
+
+
+@router.patch("/endpoints/{endpoint_id}", response_model=AcquisitionEndpoint)
+async def update_endpoint(
+    user_id: CurrentUserId, endpoint_id: UUID, request: AcquisitionEndpointUpdate, db: DbSession
+) -> AcquisitionEndpointModel:
+    endpoint = await owned_endpoint(db, user_id, endpoint_id)
+    updates = request.model_dump(exclude_unset=True, exclude={"api_key", "username", "password"})
+    for field, value in updates.items():
+        setattr(endpoint, field, str(value) if field == "base_url" else value)
+    replacement = _credentials(
+        request.api_key.get_secret_value() if request.api_key else None,
+        request.username.get_secret_value() if request.username else None,
+        request.password.get_secret_value() if request.password else None,
+    )
+    if replacement:
+        endpoint.credentials = {**(endpoint.credentials or {}), **replacement}
+    await db.commit()
+    await db.refresh(endpoint)
+    return endpoint
+
+
+@router.delete("/endpoints/{endpoint_id}", status_code=status.HTTP_204_NO_CONTENT)
+async def delete_endpoint(user_id: CurrentUserId, endpoint_id: UUID, db: DbSession) -> None:
+    endpoint = await owned_endpoint(db, user_id, endpoint_id)
+    await db.delete(endpoint)
+    await db.commit()
+
+
+@router.post("/search", response_model=list[Release])
+async def search_releases(user_id: CurrentUserId, request: SearchRequest, db: DbSession) -> list[Release]:
+    statement = select(AcquisitionEndpointModel).where(
+        AcquisitionEndpointModel.owner_user_id == user_id,
+        AcquisitionEndpointModel.enabled.is_(True),
+        AcquisitionEndpointModel.kind.in_(("prowlarr", "torznab", "newznab")),
+    )
+    if request.endpoint_ids:
+        statement = statement.where(AcquisitionEndpointModel.endpoint_id.in_(request.endpoint_ids))
+    result = await db.execute(statement)
+    releases: list[Release] = []
+    for endpoint in result.scalars():
+        releases.extend(await search_endpoint(endpoint, request.query))
+    return releases
+
+
+@router.post("/submissions", response_model=AcquisitionJob, status_code=status.HTTP_201_CREATED)
+async def submit_release(user_id: CurrentUserId, request: SubmitRequest, db: DbSession) -> AcquisitionJobModel:
+    endpoint = await owned_endpoint(db, user_id, request.endpoint_id)
+    job = AcquisitionJobModel(
+        owner_user_id=user_id, endpoint_id=endpoint.endpoint_id, title=request.title, download_url=request.download_url
+    )
+    db.add(job)
+    try:
+        job.client_reference = await submit_to_client(
+            endpoint, request.download_url, request.category, request.save_path
+        )
+        job.status = "submitted"
+    except HTTPException as exc:
+        job.status = "failed"
+        job.error = str(exc.detail)
+    await db.commit()
+    await db.refresh(job)
+    return job
+
+
+@router.post("/arr/{endpoint_id}/commands", response_model=AcquisitionJob, status_code=status.HTTP_201_CREATED)
+async def run_arr_command(
+    user_id: CurrentUserId, endpoint_id: UUID, request: ArrCommandRequest, db: DbSession
+) -> AcquisitionJobModel:
+    """Delegate a managed search/acquisition to Readarr or another Arr app."""
+    endpoint = await owned_endpoint(db, user_id, endpoint_id)
+    job = AcquisitionJobModel(
+        owner_user_id=user_id,
+        endpoint_id=endpoint.endpoint_id,
+        title=request.command,
+        download_url=f"arr-command:{request.command}",
+    )
+    db.add(job)
+    try:
+        job.client_reference = await dispatch_arr_command(endpoint, request.command, request.ids)
+        job.status = "submitted"
+    except HTTPException as exc:
+        job.status = "failed"
+        job.error = str(exc.detail)
+    await db.commit()
+    await db.refresh(job)
+    return job
+
+
+@router.get("/jobs", response_model=list[AcquisitionJob])
+async def list_jobs(user_id: CurrentUserId, db: DbSession) -> list[AcquisitionJobModel]:
+    result = await db.execute(
+        select(AcquisitionJobModel)
+        .where(AcquisitionJobModel.owner_user_id == user_id)
+        .order_by(AcquisitionJobModel.created_at.desc())
+    )
+    return list(result.scalars())
+
+
+@router.get("/rules", response_model=list[AcquisitionRule])
+async def list_rules(user_id: CurrentUserId, db: DbSession) -> list[AcquisitionRuleModel]:
+    result = await db.execute(select(AcquisitionRuleModel).where(AcquisitionRuleModel.owner_user_id == user_id))
+    return list(result.scalars())
+
+
+@router.post("/rules", response_model=AcquisitionRule, status_code=status.HTTP_201_CREATED)
+async def create_rule(user_id: CurrentUserId, request: AcquisitionRuleCreate, db: DbSession) -> AcquisitionRuleModel:
+    await owned_endpoint(db, user_id, request.download_client_id)
+    rule = AcquisitionRuleModel(
+        owner_user_id=user_id,
+        name=request.name,
+        query=request.query,
+        endpoint_ids=[str(value) for value in request.endpoint_ids] if request.endpoint_ids else None,
+        download_client_id=request.download_client_id,
+        filters=request.filters,
+        enabled=request.enabled,
+    )
+    db.add(rule)
+    await db.commit()
+    await db.refresh(rule)
+    return rule
+
+
+@router.post("/rules/{rule_id}/run", response_model=list[AcquisitionJob])
+async def run_acquisition_rule(user_id: CurrentUserId, rule_id: UUID, db: DbSession) -> list[AcquisitionJobModel]:
+    result = await db.execute(
+        select(AcquisitionRuleModel).where(
+            AcquisitionRuleModel.rule_id == rule_id, AcquisitionRuleModel.owner_user_id == user_id
+        )
+    )
+    rule = result.scalar_one_or_none()
+    if rule is None:
+        raise HTTPException(status_code=404, detail="Acquisition rule not found")
+    return await run_rule(db, rule)
diff --git a/papyrus/config.py b/papyrus/config.py
index 4208038..b6aa29d 100644
--- a/papyrus/config.py
+++ b/papyrus/config.py
@@ -71,6 +71,7 @@ class Settings(BaseSettings):
     dev_pages_use_vite: bool = False
     dev_pages_vite_url: str = "http://localhost:5173"
     dev_pages_manifest_path: str = "frontend/dev-pages/dist/.vite/manifest.json"
+    acquisition_automation_interval_seconds: int = 300
 
     @field_validator("debug", mode="before")
     @classmethod
diff --git a/papyrus/main.py b/papyrus/main.py
index f3db84d..9bec526 100644
--- a/papyrus/main.py
+++ b/papyrus/main.py
@@ -1,8 +1,9 @@
 """FastAPI application factory and configuration."""
 
+import asyncio
 import logging
 from collections.abc import AsyncGenerator
-from contextlib import asynccontextmanager
+from contextlib import asynccontextmanager, suppress
 
 import structlog
 from fastapi import FastAPI, Request, status
@@ -16,9 +17,11 @@
 
 from papyrus.api.routes import api_router, include_debug_routers
 from papyrus.config import get_settings
+from papyrus.core.database import async_session_maker
 from papyrus.core.dev_pages import DEV_PAGES_DIST_DIR, DEV_PAGES_STATIC_URL
 from papyrus.core.exceptions import AppError
 from papyrus.core.rate_limit import limiter
+from papyrus.services.acquisition import run_enabled_rules
 
 settings = get_settings()
 
@@ -40,9 +43,20 @@ def configure_logging() -> None:
 @asynccontextmanager
 async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
     """Application lifespan events."""
-    # Startup
-    yield
-    # Shutdown
+
+    async def acquisition_worker() -> None:
+        while True:
+            async with async_session_maker() as session:
+                await run_enabled_rules(session)
+            await asyncio.sleep(settings.acquisition_automation_interval_seconds)
+
+    worker = asyncio.create_task(acquisition_worker())
+    try:
+        yield
+    finally:
+        worker.cancel()
+        with suppress(asyncio.CancelledError):
+            await worker
 
 
 def create_app() -> FastAPI:
diff --git a/papyrus/models/__init__.py b/papyrus/models/__init__.py
index be17f15..a94957f 100644
--- a/papyrus/models/__init__.py
+++ b/papyrus/models/__init__.py
@@ -1,4 +1,5 @@
 from papyrus.core.database import Base
+from papyrus.models.acquisition import AcquisitionEndpoint, AcquisitionJob, AcquisitionRule
 from papyrus.models.auth import AuthExchangeCode, AuthSession, EmailActionToken, PasswordCredential, UserIdentity
 from papyrus.models.media import MediaAsset
 from papyrus.models.powersync_demo import PowerSyncDemoItem
@@ -6,6 +7,9 @@
 from papyrus.models.user import User
 
 __all__ = [
+    "AcquisitionEndpoint",
+    "AcquisitionJob",
+    "AcquisitionRule",
     "AuthExchangeCode",
     "AuthSession",
     "Base",
diff --git a/papyrus/models/acquisition.py b/papyrus/models/acquisition.py
new file mode 100644
index 0000000..66e2e9d
--- /dev/null
+++ b/papyrus/models/acquisition.py
@@ -0,0 +1,63 @@
+"""Acquisition sources, download clients and automatic acquisition rules."""
+
+from __future__ import annotations
+
+from datetime import datetime
+from uuid import UUID, uuid4
+
+from sqlalchemy import Boolean, DateTime, ForeignKey, String, Text, Uuid, func
+from sqlalchemy.dialects.postgresql import JSONB
+from sqlalchemy.orm import Mapped, mapped_column
+
+from papyrus.core.database import Base
+
+
+class AcquisitionEndpoint(Base):
+    """A private indexer, Prowlarr instance, or download client connection."""
+
+    __tablename__ = "acquisition_endpoints"
+
+    endpoint_id: Mapped[UUID] = mapped_column(Uuid, primary_key=True, default=uuid4)
+    owner_user_id: Mapped[UUID] = mapped_column(ForeignKey("users.user_id", ondelete="CASCADE"), index=True)
+    name: Mapped[str] = mapped_column(String(120), nullable=False)
+    kind: Mapped[str] = mapped_column(String(32), nullable=False)
+    base_url: Mapped[str] = mapped_column(String(2048), nullable=False)
+    credentials: Mapped[dict[str, str] | None] = mapped_column(JSONB, nullable=True)
+    settings: Mapped[dict[str, object] | None] = mapped_column(JSONB, nullable=True)
+    enabled: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True, server_default="true")
+    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False, server_default=func.now())
+    updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False, server_default=func.now())
+
+
+class AcquisitionRule(Base):
+    """A saved search that can automatically submit matching releases."""
+
+    __tablename__ = "acquisition_rules"
+
+    rule_id: Mapped[UUID] = mapped_column(Uuid, primary_key=True, default=uuid4)
+    owner_user_id: Mapped[UUID] = mapped_column(ForeignKey("users.user_id", ondelete="CASCADE"), index=True)
+    name: Mapped[str] = mapped_column(String(120), nullable=False)
+    query: Mapped[str] = mapped_column(String(500), nullable=False)
+    endpoint_ids: Mapped[list[str] | None] = mapped_column(JSONB, nullable=True)
+    download_client_id: Mapped[UUID | None] = mapped_column(Uuid, ForeignKey("acquisition_endpoints.endpoint_id"))
+    filters: Mapped[dict[str, object] | None] = mapped_column(JSONB, nullable=True)
+    enabled: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True, server_default="true")
+    last_run_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
+    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False, server_default=func.now())
+
+
+class AcquisitionJob(Base):
+    """An auditable submission to a download client."""
+
+    __tablename__ = "acquisition_jobs"
+
+    job_id: Mapped[UUID] = mapped_column(Uuid, primary_key=True, default=uuid4)
+    owner_user_id: Mapped[UUID] = mapped_column(ForeignKey("users.user_id", ondelete="CASCADE"), index=True)
+    endpoint_id: Mapped[UUID] = mapped_column(Uuid, ForeignKey("acquisition_endpoints.endpoint_id"), nullable=False)
+    rule_id: Mapped[UUID | None] = mapped_column(Uuid, ForeignKey("acquisition_rules.rule_id", ondelete="SET NULL"))
+    title: Mapped[str] = mapped_column(String(500), nullable=False)
+    download_url: Mapped[str] = mapped_column(Text, nullable=False)
+    status: Mapped[str] = mapped_column(String(32), nullable=False, default="queued", server_default="queued")
+    client_reference: Mapped[str | None] = mapped_column(String(255), nullable=True)
+    error: Mapped[str | None] = mapped_column(Text, nullable=True)
+    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False, server_default=func.now())
diff --git a/papyrus/schemas/acquisition.py b/papyrus/schemas/acquisition.py
new file mode 100644
index 0000000..d40d5e8
--- /dev/null
+++ b/papyrus/schemas/acquisition.py
@@ -0,0 +1,116 @@
+"""HTTP schemas for private acquisition integrations."""
+
+from datetime import datetime
+from enum import StrEnum
+from uuid import UUID
+
+from pydantic import BaseModel, ConfigDict, Field, HttpUrl, SecretStr
+
+
+class EndpointKind(StrEnum):
+    QBITTORRENT = "qbittorrent"
+    TRANSMISSION = "transmission"
+    DELUGE = "deluge"
+    PROWLARR = "prowlarr"
+    TORZNAB = "torznab"
+    NEWZNAB = "newznab"
+    READARR = "readarr"
+    SONARR = "sonarr"
+    RADARR = "radarr"
+    LIDARR = "lidarr"
+    WHISPARR = "whisparr"
+
+
+class AcquisitionEndpointCreate(BaseModel):
+    name: str = Field(min_length=1, max_length=120)
+    kind: EndpointKind
+    base_url: HttpUrl
+    api_key: SecretStr | None = None
+    username: SecretStr | None = None
+    password: SecretStr | None = None
+    settings: dict[str, object] | None = None
+
+
+class AcquisitionEndpointUpdate(BaseModel):
+    name: str | None = Field(None, min_length=1, max_length=120)
+    base_url: HttpUrl | None = None
+    api_key: SecretStr | None = None
+    username: SecretStr | None = None
+    password: SecretStr | None = None
+    settings: dict[str, object] | None = None
+    enabled: bool | None = None
+
+
+class AcquisitionEndpoint(BaseModel):
+    model_config = ConfigDict(from_attributes=True)
+    endpoint_id: UUID
+    name: str
+    kind: EndpointKind
+    base_url: str
+    enabled: bool
+    settings: dict[str, object] | None = None
+    created_at: datetime | None = None
+
+
+class Release(BaseModel):
+    title: str
+    download_url: str
+    protocol: str
+    indexer: str
+    size_bytes: int | None = None
+    seeders: int | None = None
+    publish_date: datetime | None = None
+
+
+class SearchRequest(BaseModel):
+    query: str = Field(min_length=1, max_length=500)
+    endpoint_ids: list[UUID] | None = None
+
+
+class SubmitRequest(BaseModel):
+    endpoint_id: UUID
+    title: str = Field(min_length=1, max_length=500)
+    download_url: str = Field(min_length=1, max_length=4096)
+    category: str | None = Field(None, max_length=100)
+    save_path: str | None = Field(None, max_length=1024)
+
+
+class ArrCommandRequest(BaseModel):
+    """A Servarr command and the IDs scoped to that application's library."""
+
+    command: str = Field(min_length=1, max_length=80)
+    ids: list[int] = Field(default_factory=list, max_length=100)
+
+
+class AcquisitionRuleCreate(BaseModel):
+    name: str = Field(min_length=1, max_length=120)
+    query: str = Field(min_length=1, max_length=500)
+    endpoint_ids: list[UUID] | None = None
+    download_client_id: UUID
+    filters: dict[str, object] | None = None
+    enabled: bool = True
+
+
+class AcquisitionRule(BaseModel):
+    model_config = ConfigDict(from_attributes=True)
+    rule_id: UUID
+    name: str
+    query: str
+    endpoint_ids: list[UUID] | None = None
+    download_client_id: UUID | None
+    filters: dict[str, object] | None = None
+    enabled: bool
+    last_run_at: datetime | None = None
+
+
+class AcquisitionJob(BaseModel):
+    model_config = ConfigDict(from_attributes=True)
+    job_id: UUID
+    endpoint_id: UUID
+    rule_id: UUID | None
+    title: str
+    download_url: str
+    status: str
+    client_reference: str | None = None
+    error: str | None = None
+    created_at: datetime | None = None
diff --git a/papyrus/services/acquisition.py b/papyrus/services/acquisition.py
new file mode 100644
index 0000000..c6e1d00
--- /dev/null
+++ b/papyrus/services/acquisition.py
@@ -0,0 +1,327 @@
+"""Adapters for indexer protocols and self-hosted download clients."""
+
+from __future__ import annotations
+
+import asyncio
+import base64
+import json
+from datetime import UTC, datetime
+from typing import Any
+from urllib.error import HTTPError, URLError
+from urllib.parse import urlencode, urljoin
+from urllib.request import Request, urlopen
+from xml.etree import ElementTree
+
+from fastapi import HTTPException, status
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from papyrus.models.acquisition import AcquisitionEndpoint, AcquisitionJob, AcquisitionRule
+from papyrus.schemas.acquisition import Release
+
+
+def _url(endpoint: AcquisitionEndpoint, path: str) -> str:
+    return urljoin(endpoint.base_url.rstrip("/") + "/", path.lstrip("/"))
+
+
+async def _request(
+    url: str, *, method: str = "GET", headers: dict[str, str] | None = None, body: bytes | None = None
+) -> tuple[int, dict[str, str], bytes]:
+    """Perform a bounded blocking HTTP request off the event loop."""
+
+    def send() -> tuple[int, dict[str, str], bytes]:
+        request = Request(url, data=body, headers=headers or {}, method=method)
+        try:
+            with urlopen(request, timeout=15) as response:  # noqa: S310 - user-owned self-hosted integrations
+                return response.status, dict(response.headers.items()), response.read(5_000_000)
+        except HTTPError as exc:
+            return exc.code, dict(exc.headers.items()), exc.read(1_000_000)
+        except URLError as exc:
+            raise HTTPException(status_code=502, detail=f"Integration request failed: {exc.reason}") from exc
+
+    return await asyncio.to_thread(send)
+
+
+def _credentials(endpoint: AcquisitionEndpoint) -> dict[str, str]:
+    return endpoint.credentials or {}
+
+
+async def search_endpoint(endpoint: AcquisitionEndpoint, query: str) -> list[Release]:
+    """Search Prowlarr or a Torznab/Newznab-compatible indexer."""
+    credentials = _credentials(endpoint)
+    if endpoint.kind == "prowlarr":
+        request_url = _url(endpoint, f"api/v1/search?{urlencode({'query': query})}")
+        response_status, _, payload = await _request(request_url, headers={"X-Api-Key": credentials.get("api_key", "")})
+        if response_status >= 400:
+            raise HTTPException(status_code=502, detail="Prowlarr search failed")
+        data = json.loads(payload)
+        return [
+            Release(
+                title=item.get("title", "Untitled"),
+                download_url=item.get("downloadUrl") or item.get("magnetUrl") or item.get("guid", ""),
+                protocol="torrent" if item.get("protocol", "torrent") == "torrent" else "usenet",
+                indexer=item.get("indexer", "Prowlarr"),
+                size_bytes=item.get("size"),
+                seeders=item.get("seeders"),
+            )
+            for item in data
+            if item.get("downloadUrl") or item.get("magnetUrl") or item.get("guid")
+        ]
+
+    params = urlencode({"t": "search", "q": query, "apikey": credentials.get("api_key", "")})
+    response_status, _, payload = await _request(_url(endpoint, f"api?{params}"))
+    if response_status >= 400:
+        raise HTTPException(status_code=502, detail=f"{endpoint.kind.title()} search failed")
+    return _parse_newznab(payload, endpoint.name)
+
+
+def _parse_newznab(payload: bytes, indexer: str) -> list[Release]:
+    try:
+        root = ElementTree.fromstring(payload)
+    except ElementTree.ParseError as exc:
+        raise HTTPException(status_code=502, detail="Indexer returned invalid XML") from exc
+    releases: list[Release] = []
+    for item in root.findall(".//item"):
+        enclosure = item.find("enclosure")
+        link = (enclosure.get("url") if enclosure is not None else None) or item.findtext("link")
+        if not link:
+            continue
+        attrs = {child.attrib.get("name"): child.attrib.get("value") for child in item if child.tag.endswith("attr")}
+        size = attrs.get("size")
+        seeders = attrs.get("seeders")
+        releases.append(
+            Release(
+                title=item.findtext("title") or "Untitled",
+                download_url=link,
+                protocol="torrent" if link.startswith("magnet:") or ".torrent" in link else "usenet",
+                indexer=indexer,
+                size_bytes=int(size) if size is not None and size.isdigit() else None,
+                seeders=int(seeders) if seeders is not None and seeders.isdigit() else None,
+            )
+        )
+    return releases
+
+
+async def submit_to_client(
+    endpoint: AcquisitionEndpoint, download_url: str, category: str | None, save_path: str | None
+) -> str | None:
+    """Submit a magnet or torrent URL to a supported BitTorrent client."""
+    if endpoint.kind == "qbittorrent":
+        return await _submit_qbittorrent(endpoint, download_url, category, save_path)
+    if endpoint.kind == "transmission":
+        return await _submit_transmission(endpoint, download_url, save_path)
+    if endpoint.kind == "deluge":
+        return await _submit_deluge(endpoint, download_url, save_path)
+    raise HTTPException(status_code=422, detail="Endpoint is not a download client")
+
+
+async def dispatch_arr_command(endpoint: AcquisitionEndpoint, command: str, ids: list[int]) -> str | None:
+    """Start an acquisition/search command in a Servarr application.
+
+    The Arr apps own their library and release-grab decisions, so they must be
+    driven through their command API rather than handed an arbitrary magnet.
+    """
+    if endpoint.kind not in {"readarr", "sonarr", "radarr", "lidarr", "whisparr"}:
+        raise HTTPException(status_code=422, detail="Endpoint is not a Servarr application")
+
+    allowed_commands = {
+        "readarr": {"AuthorSearch", "BookSearch"},
+        "sonarr": {"SeriesSearch", "EpisodeSearch", "MissingEpisodeSearch"},
+        "radarr": {"MoviesSearch", "MissingMoviesSearch"},
+        "lidarr": {"ArtistSearch", "AlbumSearch", "MissingAlbumSearch"},
+        "whisparr": {"SeriesSearch", "EpisodeSearch", "MissingEpisodeSearch"},
+    }
+    if command not in allowed_commands[endpoint.kind]:
+        raise HTTPException(status_code=422, detail="Command is not supported by this Servarr application")
+
+    id_field = {
+        "AuthorSearch": "authorIds",
+        "BookSearch": "bookIds",
+        "SeriesSearch": "seriesId",
+        "EpisodeSearch": "episodeIds",
+        "MissingEpisodeSearch": "seriesId",
+        "MoviesSearch": "movieIds",
+        "MissingMoviesSearch": "movieIds",
+        "ArtistSearch": "artistIds",
+        "AlbumSearch": "albumIds",
+        "MissingAlbumSearch": "artistIds",
+    }[command]
+    payload: dict[str, object] = {"name": command}
+    if ids:
+        payload[id_field] = ids[0] if id_field in {"seriesId"} else ids
+
+    response_status, _, response_payload = await _request(
+        _url(endpoint, "api/v1/command"),
+        method="POST",
+        headers={"Content-Type": "application/json", "X-Api-Key": _credentials(endpoint).get("api_key", "")},
+        body=json.dumps(payload).encode(),
+    )
+    if response_status >= 400:
+        raise HTTPException(status_code=502, detail=f"{endpoint.kind.title()} command failed")
+    response = json.loads(response_payload)
+    return str(response.get("id")) if response.get("id") is not None else None
+
+
+async def _submit_qbittorrent(
+    endpoint: AcquisitionEndpoint, download_url: str, category: str | None, save_path: str | None
+) -> str | None:
+    credentials = _credentials(endpoint)
+    login = urlencode(
+        {"username": credentials.get("username", ""), "password": credentials.get("password", "")}
+    ).encode()
+    response_status, headers, _ = await _request(
+        _url(endpoint, "api/v2/auth/login"),
+        method="POST",
+        headers={"Content-Type": "application/x-www-form-urlencoded"},
+        body=login,
+    )
+    if response_status >= 400:
+        raise HTTPException(status_code=502, detail="qBittorrent authentication failed")
+    payload = {"urls": download_url}
+    if category:
+        payload["category"] = category
+    if save_path:
+        payload["savepath"] = save_path
+    response_status, _, _ = await _request(
+        _url(endpoint, "api/v2/torrents/add"),
+        method="POST",
+        headers={
+            "Content-Type": "application/x-www-form-urlencoded",
+            "Cookie": headers.get("Set-Cookie", "").split(";", 1)[0],
+        },
+        body=urlencode(payload).encode(),
+    )
+    if response_status >= 400:
+        raise HTTPException(status_code=502, detail="qBittorrent rejected the release")
+    return None
+
+
+async def _submit_transmission(endpoint: AcquisitionEndpoint, download_url: str, save_path: str | None) -> str | None:
+    credentials = _credentials(endpoint)
+    arguments: dict[str, str] = {"filename": download_url}
+    if save_path:
+        arguments["download-dir"] = save_path
+    body = json.dumps({"method": "torrent-add", "arguments": arguments}).encode()
+    headers = {"Content-Type": "application/json"}
+    if credentials.get("username"):
+        token = base64.b64encode(f"{credentials['username']}:{credentials.get('password', '')}".encode()).decode()
+        headers["Authorization"] = f"Basic {token}"
+    response_status, response_headers, payload = await _request(
+        _url(endpoint, "transmission/rpc"), method="POST", headers=headers, body=body
+    )
+    if response_status == 409:
+        headers["X-Transmission-Session-Id"] = response_headers.get("X-Transmission-Session-Id", "")
+        response_status, _, payload = await _request(
+            _url(endpoint, "transmission/rpc"), method="POST", headers=headers, body=body
+        )
+    if response_status >= 400:
+        raise HTTPException(status_code=502, detail="Transmission rejected the release")
+    return str(json.loads(payload).get("arguments", {}).get("torrent-added", {}).get("hashString") or "") or None
+
+
+async def _submit_deluge(endpoint: AcquisitionEndpoint, download_url: str, save_path: str | None) -> str | None:
+    credentials = _credentials(endpoint)
+    headers = {"Content-Type": "application/json"}
+    login = json.dumps({"method": "auth.login", "params": [credentials.get("password", "")], "id": 1}).encode()
+    response_status, response_headers, _ = await _request(
+        _url(endpoint, "json"), method="POST", headers=headers, body=login
+    )
+    if response_status >= 400:
+        raise HTTPException(status_code=502, detail="Deluge authentication failed")
+    options = {"download_location": save_path} if save_path else {}
+    body = json.dumps({"method": "core.add_torrent_magnet", "params": [download_url, options], "id": 2}).encode()
+    headers["Cookie"] = response_headers.get("Set-Cookie", "").split(";", 1)[0]
+    response_status, _, payload = await _request(_url(endpoint, "json"), method="POST", headers=headers, body=body)
+    if response_status >= 400:
+        raise HTTPException(status_code=502, detail="Deluge rejected the release")
+    result = json.loads(payload).get("result")
+    return str(result) if result is not None else None
+
+
+async def owned_endpoint(session: AsyncSession, owner_user_id: Any, endpoint_id: Any) -> AcquisitionEndpoint:
+    result = await session.execute(
+        select(AcquisitionEndpoint).where(
+            AcquisitionEndpoint.endpoint_id == endpoint_id, AcquisitionEndpoint.owner_user_id == owner_user_id
+        )
+    )
+    endpoint = result.scalar_one_or_none()
+    if endpoint is None:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Acquisition endpoint not found")
+    return endpoint
+
+
+async def run_rule(session: AsyncSession, rule: AcquisitionRule) -> list[AcquisitionJob]:
+    """Run one rule once; callers may schedule this from their worker/cron service."""
+    client = await owned_endpoint(session, rule.owner_user_id, rule.download_client_id)
+    if client.kind in {"readarr", "sonarr", "radarr", "lidarr", "whisparr"}:
+        filters = rule.filters or {}
+        command = filters.get("arr_command")
+        ids = filters.get("arr_ids", [])
+        if (
+            not isinstance(command, str)
+            or not isinstance(ids, list)
+            or not all(isinstance(value, int) for value in ids)
+        ):
+            raise HTTPException(
+                status_code=422,
+                detail="Arr rules require filters.arr_command and filters.arr_ids",
+            )
+        job = AcquisitionJob(
+            owner_user_id=rule.owner_user_id,
+            endpoint_id=client.endpoint_id,
+            rule_id=rule.rule_id,
+            title=command,
+            download_url=f"arr-command:{command}",
+        )
+        session.add(job)
+        try:
+            job.client_reference = await dispatch_arr_command(client, command, ids)
+            job.status = "submitted"
+        except HTTPException as exc:
+            job.status = "failed"
+            job.error = str(exc.detail)
+        rule.last_run_at = datetime.now(UTC)
+        await session.commit()
+        return [job]
+
+    endpoint_ids = rule.endpoint_ids or []
+    result = await session.execute(
+        select(AcquisitionEndpoint).where(
+            AcquisitionEndpoint.owner_user_id == rule.owner_user_id,
+            AcquisitionEndpoint.endpoint_id.in_(endpoint_ids),
+            AcquisitionEndpoint.enabled.is_(True),
+        )
+    )
+    releases = [release for endpoint in result.scalars() for release in await search_endpoint(endpoint, rule.query)]
+    if not releases:
+        rule.last_run_at = datetime.now(UTC)
+        await session.commit()
+        return []
+    selected = sorted(releases, key=lambda release: (release.seeders or 0, release.size_bytes or 0), reverse=True)[0]
+    job = AcquisitionJob(
+        owner_user_id=rule.owner_user_id,
+        endpoint_id=client.endpoint_id,
+        rule_id=rule.rule_id,
+        title=selected.title,
+        download_url=selected.download_url,
+    )
+    session.add(job)
+    try:
+        job.client_reference = await submit_to_client(client, selected.download_url, None, None)
+        job.status = "submitted"
+    except HTTPException as exc:
+        job.status = "failed"
+        job.error = str(exc.detail)
+    rule.last_run_at = datetime.now(UTC)
+    await session.commit()
+    return [job]
+
+
+async def run_enabled_rules(session: AsyncSession) -> None:
+    """Run enabled rules, isolating a failed remote integration from the others."""
+    result = await session.execute(select(AcquisitionRule).where(AcquisitionRule.enabled.is_(True)))
+    for rule in result.scalars():
+        try:
+            await run_rule(session, rule)
+        except HTTPException:
+            await session.rollback()
diff --git a/tests/api/routes/test_acquisition.py b/tests/api/routes/test_acquisition.py
new file mode 100644
index 0000000..f4272a7
--- /dev/null
+++ b/tests/api/routes/test_acquisition.py
@@ -0,0 +1,59 @@
+"""Tests for private BitTorrent acquisition configuration."""
+
+from httpx import AsyncClient
+
+
+async def test_create_and_list_endpoint_hides_credentials(client: AsyncClient, auth_headers: dict[str, str]) -> None:
+    """Credentials can be configured but must never be returned to a client."""
+    response = await client.post(
+        "/v1/acquisition/endpoints",
+        headers=auth_headers,
+        json={
+            "name": "Home qBittorrent",
+            "kind": "qbittorrent",
+            "base_url": "http://127.0.0.1:8080",
+            "username": "admin",
+            "password": "secret",
+        },
+    )
+
+    assert response.status_code == 201
+    assert "password" not in response.text
+    assert "username" not in response.text
+    endpoint = response.json()
+
+    response = await client.get("/v1/acquisition/endpoints", headers=auth_headers)
+    assert response.status_code == 200
+    assert response.json() == [endpoint]
+
+
+async def test_create_rule_requires_owned_download_client(client: AsyncClient, auth_headers: dict[str, str]) -> None:
+    """Rules cannot target a client belonging to another user or a missing client."""
+    response = await client.post(
+        "/v1/acquisition/rules",
+        headers=auth_headers,
+        json={
+            "name": "Monthly reading",
+            "query": "example book",
+            "download_client_id": "00000000-0000-0000-0000-000000000001",
+        },
+    )
+
+    assert response.status_code == 404
+
+
+async def test_readarr_endpoint_is_a_supported_integration(client: AsyncClient, auth_headers: dict[str, str]) -> None:
+    """Readarr uses its own acquisition workflow and is accepted as an endpoint."""
+    response = await client.post(
+        "/v1/acquisition/endpoints",
+        headers=auth_headers,
+        json={
+            "name": "Reading automation",
+            "kind": "readarr",
+            "base_url": "http://readarr.local:8787",
+            "api_key": "private-key",
+        },
+    )
+
+    assert response.status_code == 201
+    assert response.json()["kind"] == "readarr"

From a7e6808625e141b41d92048da94c2d88c8dd1b7d Mon Sep 17 00:00:00 2001
From: "Matheus Henrique(Aka: TheusHen)" <mmmatheushenriqueee@gmail.com>
Date: Tue, 14 Jul 2026 17:30:10 -0300
Subject: [PATCH 2/8] Complete torrent acquisition contract

---
 papyrus/api/routes/acquisition.py    | 50 +++++++++++++++++++++--
 papyrus/core/security.py             | 27 +++++++++++++
 papyrus/schemas/acquisition.py       | 41 ++++++++++++++++++-
 papyrus/services/acquisition.py      | 23 +++++++----
 tests/api/routes/test_acquisition.py | 59 +++++++++++++++++++++++++++-
 5 files changed, 186 insertions(+), 14 deletions(-)

diff --git a/papyrus/api/routes/acquisition.py b/papyrus/api/routes/acquisition.py
index a5a0ae2..bcbbadf 100644
--- a/papyrus/api/routes/acquisition.py
+++ b/papyrus/api/routes/acquisition.py
@@ -9,10 +9,12 @@
 
 from papyrus.api.deps import CurrentUserId
 from papyrus.core.database import get_db
+from papyrus.core.security import decrypt_secret_payload, encrypt_secret_payload
 from papyrus.models.acquisition import AcquisitionEndpoint as AcquisitionEndpointModel
 from papyrus.models.acquisition import AcquisitionJob as AcquisitionJobModel
 from papyrus.models.acquisition import AcquisitionRule as AcquisitionRuleModel
 from papyrus.schemas.acquisition import (
+    AcquisitionCapabilities,
     AcquisitionEndpoint,
     AcquisitionEndpointCreate,
     AcquisitionEndpointUpdate,
@@ -40,7 +42,48 @@ def _credentials(api_key: str | None, username: str | None, password: str | None
     values = {
         key: value for key, value in {"api_key": api_key, "username": username, "password": password}.items() if value
     }
-    return values or None
+    if not values:
+        return None
+    return {"encrypted": encrypt_secret_payload(values)}
+
+
+def _merge_credentials(endpoint: AcquisitionEndpointModel, replacement: dict[str, str] | None) -> None:
+    if replacement is None:
+        return
+    current = endpoint.credentials or {}
+    if encrypted := current.get("encrypted"):
+        current = decrypt_secret_payload(encrypted)
+    endpoint.credentials = {
+        "encrypted": encrypt_secret_payload({**current, **decrypt_secret_payload(replacement["encrypted"])})
+    }
+
+
+@router.get("/capabilities", response_model=AcquisitionCapabilities)
+async def acquisition_capabilities() -> AcquisitionCapabilities:
+    return AcquisitionCapabilities(
+        endpoint_kinds=[
+            "qbittorrent",
+            "transmission",
+            "deluge",
+            "prowlarr",
+            "torznab",
+            "readarr",
+            "sonarr",
+            "radarr",
+            "lidarr",
+            "whisparr",
+        ],
+        indexer_kinds=["prowlarr", "torznab"],
+        download_client_kinds=["qbittorrent", "transmission", "deluge"],
+        arr_kinds=["readarr", "sonarr", "radarr", "lidarr", "whisparr"],
+        arr_commands={
+            "readarr": ["AuthorSearch", "BookSearch"],
+            "sonarr": ["SeriesSearch", "EpisodeSearch", "MissingEpisodeSearch"],
+            "radarr": ["MoviesSearch", "MissingMoviesSearch"],
+            "lidarr": ["ArtistSearch", "AlbumSearch", "MissingAlbumSearch"],
+            "whisparr": ["SeriesSearch", "EpisodeSearch", "MissingEpisodeSearch"],
+        },
+    )
 
 
 @router.get("/endpoints", response_model=list[AcquisitionEndpoint])
@@ -88,8 +131,7 @@ async def update_endpoint(
         request.username.get_secret_value() if request.username else None,
         request.password.get_secret_value() if request.password else None,
     )
-    if replacement:
-        endpoint.credentials = {**(endpoint.credentials or {}), **replacement}
+    _merge_credentials(endpoint, replacement)
     await db.commit()
     await db.refresh(endpoint)
     return endpoint
@@ -107,7 +149,7 @@ async def search_releases(user_id: CurrentUserId, request: SearchRequest, db: Db
     statement = select(AcquisitionEndpointModel).where(
         AcquisitionEndpointModel.owner_user_id == user_id,
         AcquisitionEndpointModel.enabled.is_(True),
-        AcquisitionEndpointModel.kind.in_(("prowlarr", "torznab", "newznab")),
+        AcquisitionEndpointModel.kind.in_(("prowlarr", "torznab")),
     )
     if request.endpoint_ids:
         statement = statement.where(AcquisitionEndpointModel.endpoint_id.in_(request.endpoint_ids))
diff --git a/papyrus/core/security.py b/papyrus/core/security.py
index 8334630..b35cd48 100644
--- a/papyrus/core/security.py
+++ b/papyrus/core/security.py
@@ -1,13 +1,17 @@
 """Authentication and token security helpers."""
 
+from base64 import urlsafe_b64encode
 from datetime import UTC, datetime, timedelta
 from functools import lru_cache
 from hashlib import sha256
+from json import dumps as json_dumps
+from json import loads as json_loads
 from pathlib import Path
 from secrets import token_urlsafe
 from typing import Any
 
 import jwt
+from cryptography.fernet import Fernet, InvalidToken
 from cryptography.hazmat.primitives import serialization
 from jwt.algorithms import RSAAlgorithm
 from pwdlib import PasswordHash
@@ -125,6 +129,29 @@ def generate_opaque_token() -> str:
     return token_urlsafe(48)
 
 
+@lru_cache
+def _get_credentials_cipher() -> Fernet:
+    key = urlsafe_b64encode(sha256(get_settings().secret_key.encode("utf-8")).digest())
+    return Fernet(key)
+
+
+def encrypt_secret_payload(payload: dict[str, str]) -> str:
+    return _get_credentials_cipher().encrypt(json_dumps(payload).encode("utf-8")).decode("utf-8")
+
+
+def decrypt_secret_payload(token: str) -> dict[str, str]:
+    try:
+        decoded = _get_credentials_cipher().decrypt(token.encode("utf-8")).decode("utf-8")
+    except InvalidToken as exc:
+        raise ValueError("Encrypted credential payload is invalid") from exc
+    payload = json_loads(decoded)
+    if not isinstance(payload, dict) or not all(
+        isinstance(key, str) and isinstance(value, str) for key, value in payload.items()
+    ):
+        raise ValueError("Encrypted credential payload has an invalid shape")
+    return payload
+
+
 def create_access_token(data: dict[str, Any], expires_delta: timedelta | None = None) -> str:
     settings = get_settings()
     ttl = expires_delta or timedelta(minutes=settings.access_token_expire_minutes)
diff --git a/papyrus/schemas/acquisition.py b/papyrus/schemas/acquisition.py
index d40d5e8..850f118 100644
--- a/papyrus/schemas/acquisition.py
+++ b/papyrus/schemas/acquisition.py
@@ -4,7 +4,7 @@
 from enum import StrEnum
 from uuid import UUID
 
-from pydantic import BaseModel, ConfigDict, Field, HttpUrl, SecretStr
+from pydantic import BaseModel, ConfigDict, Field, HttpUrl, SecretStr, field_validator
 
 
 class EndpointKind(StrEnum):
@@ -13,7 +13,6 @@ class EndpointKind(StrEnum):
     DELUGE = "deluge"
     PROWLARR = "prowlarr"
     TORZNAB = "torznab"
-    NEWZNAB = "newznab"
     READARR = "readarr"
     SONARR = "sonarr"
     RADARR = "radarr"
@@ -21,6 +20,15 @@ class EndpointKind(StrEnum):
     WHISPARR = "whisparr"
 
 
+class AcquisitionCapabilities(BaseModel):
+    enabled: bool = True
+    endpoint_kinds: list[EndpointKind]
+    indexer_kinds: list[EndpointKind]
+    download_client_kinds: list[EndpointKind]
+    arr_kinds: list[EndpointKind]
+    arr_commands: dict[EndpointKind, list[str]]
+
+
 class AcquisitionEndpointCreate(BaseModel):
     name: str = Field(min_length=1, max_length=120)
     kind: EndpointKind
@@ -30,6 +38,11 @@ class AcquisitionEndpointCreate(BaseModel):
     password: SecretStr | None = None
     settings: dict[str, object] | None = None
 
+    @field_validator("base_url")
+    @classmethod
+    def validate_endpoint_url(cls, value: HttpUrl) -> HttpUrl:
+        return validate_endpoint_url(value)
+
 
 class AcquisitionEndpointUpdate(BaseModel):
     name: str | None = Field(None, min_length=1, max_length=120)
@@ -40,6 +53,13 @@ class AcquisitionEndpointUpdate(BaseModel):
     settings: dict[str, object] | None = None
     enabled: bool | None = None
 
+    @field_validator("base_url")
+    @classmethod
+    def validate_endpoint_url(cls, value: HttpUrl | None) -> HttpUrl | None:
+        if value is None:
+            return None
+        return validate_endpoint_url(value)
+
 
 class AcquisitionEndpoint(BaseModel):
     model_config = ConfigDict(from_attributes=True)
@@ -74,6 +94,15 @@ class SubmitRequest(BaseModel):
     category: str | None = Field(None, max_length=100)
     save_path: str | None = Field(None, max_length=1024)
 
+    @field_validator("download_url")
+    @classmethod
+    def validate_torrent_download_url(cls, value: str) -> str:
+        if value.startswith("magnet:"):
+            return value
+        if value.startswith(("http://", "https://")):
+            return value
+        raise ValueError("download_url must be a magnet, HTTP, or HTTPS torrent URL")
+
 
 class ArrCommandRequest(BaseModel):
     """A Servarr command and the IDs scoped to that application's library."""
@@ -114,3 +143,11 @@ class AcquisitionJob(BaseModel):
     client_reference: str | None = None
     error: str | None = None
     created_at: datetime | None = None
+
+
+def validate_endpoint_url(value: HttpUrl) -> HttpUrl:
+    if value.username or value.password:
+        raise ValueError("Endpoint URL must not include credentials")
+    if value.scheme not in {"http", "https"}:
+        raise ValueError("Endpoint URL must use HTTP or HTTPS")
+    return value
diff --git a/papyrus/services/acquisition.py b/papyrus/services/acquisition.py
index c6e1d00..e6a3cfd 100644
--- a/papyrus/services/acquisition.py
+++ b/papyrus/services/acquisition.py
@@ -16,6 +16,7 @@
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
+from papyrus.core.security import decrypt_secret_payload
 from papyrus.models.acquisition import AcquisitionEndpoint, AcquisitionJob, AcquisitionRule
 from papyrus.schemas.acquisition import Release
 
@@ -43,11 +44,18 @@ def send() -> tuple[int, dict[str, str], bytes]:
 
 
 def _credentials(endpoint: AcquisitionEndpoint) -> dict[str, str]:
-    return endpoint.credentials or {}
+    credentials = endpoint.credentials or {}
+    encrypted = credentials.get("encrypted")
+    if encrypted is None:
+        return credentials
+    try:
+        return decrypt_secret_payload(encrypted)
+    except ValueError as exc:
+        raise HTTPException(status_code=500, detail="Stored integration credentials are invalid") from exc
 
 
 async def search_endpoint(endpoint: AcquisitionEndpoint, query: str) -> list[Release]:
-    """Search Prowlarr or a Torznab/Newznab-compatible indexer."""
+    """Search Prowlarr or a Torznab-compatible torrent indexer."""
     credentials = _credentials(endpoint)
     if endpoint.kind == "prowlarr":
         request_url = _url(endpoint, f"api/v1/search?{urlencode({'query': query})}")
@@ -59,23 +67,24 @@ async def search_endpoint(endpoint: AcquisitionEndpoint, query: str) -> list[Rel
             Release(
                 title=item.get("title", "Untitled"),
                 download_url=item.get("downloadUrl") or item.get("magnetUrl") or item.get("guid", ""),
-                protocol="torrent" if item.get("protocol", "torrent") == "torrent" else "usenet",
+                protocol="torrent",
                 indexer=item.get("indexer", "Prowlarr"),
                 size_bytes=item.get("size"),
                 seeders=item.get("seeders"),
             )
             for item in data
-            if item.get("downloadUrl") or item.get("magnetUrl") or item.get("guid")
+            if (item.get("downloadUrl") or item.get("magnetUrl") or item.get("guid"))
+            and item.get("protocol", "torrent") == "torrent"
         ]
 
     params = urlencode({"t": "search", "q": query, "apikey": credentials.get("api_key", "")})
     response_status, _, payload = await _request(_url(endpoint, f"api?{params}"))
     if response_status >= 400:
         raise HTTPException(status_code=502, detail=f"{endpoint.kind.title()} search failed")
-    return _parse_newznab(payload, endpoint.name)
+    return _parse_torznab(payload, endpoint.name)
 
 
-def _parse_newznab(payload: bytes, indexer: str) -> list[Release]:
+def _parse_torznab(payload: bytes, indexer: str) -> list[Release]:
     try:
         root = ElementTree.fromstring(payload)
     except ElementTree.ParseError as exc:
@@ -93,7 +102,7 @@ def _parse_newznab(payload: bytes, indexer: str) -> list[Release]:
             Release(
                 title=item.findtext("title") or "Untitled",
                 download_url=link,
-                protocol="torrent" if link.startswith("magnet:") or ".torrent" in link else "usenet",
+                protocol="torrent",
                 indexer=indexer,
                 size_bytes=int(size) if size is not None and size.isdigit() else None,
                 seeders=int(seeders) if seeders is not None and seeders.isdigit() else None,
diff --git a/tests/api/routes/test_acquisition.py b/tests/api/routes/test_acquisition.py
index f4272a7..d471034 100644
--- a/tests/api/routes/test_acquisition.py
+++ b/tests/api/routes/test_acquisition.py
@@ -1,9 +1,29 @@
 """Tests for private BitTorrent acquisition configuration."""
 
 from httpx import AsyncClient
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
 
+from papyrus.core.security import decrypt_secret_payload
+from papyrus.models.acquisition import AcquisitionEndpoint
 
-async def test_create_and_list_endpoint_hides_credentials(client: AsyncClient, auth_headers: dict[str, str]) -> None:
+
+async def test_capabilities_advertise_torrent_only_scope(client: AsyncClient, auth_headers: dict[str, str]) -> None:
+    response = await client.get("/v1/acquisition/capabilities", headers=auth_headers)
+
+    assert response.status_code == 200
+    body = response.json()
+    assert body["indexer_kinds"] == ["prowlarr", "torznab"]
+    assert body["download_client_kinds"] == ["qbittorrent", "transmission", "deluge"]
+    assert "newznab" not in body["endpoint_kinds"]
+    assert body["arr_commands"]["readarr"] == ["AuthorSearch", "BookSearch"]
+
+
+async def test_create_and_list_endpoint_hides_and_encrypts_credentials(
+    client: AsyncClient,
+    auth_headers: dict[str, str],
+    db_session: AsyncSession,
+) -> None:
     """Credentials can be configured but must never be returned to a client."""
     response = await client.post(
         "/v1/acquisition/endpoints",
@@ -26,6 +46,14 @@ async def test_create_and_list_endpoint_hides_credentials(client: AsyncClient, a
     assert response.status_code == 200
     assert response.json() == [endpoint]
 
+    stored = (await db_session.execute(select(AcquisitionEndpoint))).scalar_one()
+    assert stored.credentials is not None
+    assert "password" not in stored.credentials
+    assert decrypt_secret_payload(stored.credentials["encrypted"]) == {
+        "username": "admin",
+        "password": "secret",
+    }
+
 
 async def test_create_rule_requires_owned_download_client(client: AsyncClient, auth_headers: dict[str, str]) -> None:
     """Rules cannot target a client belonging to another user or a missing client."""
@@ -57,3 +85,32 @@ async def test_readarr_endpoint_is_a_supported_integration(client: AsyncClient,
 
     assert response.status_code == 201
     assert response.json()["kind"] == "readarr"
+
+
+async def test_newznab_endpoint_is_rejected(client: AsyncClient, auth_headers: dict[str, str]) -> None:
+    response = await client.post(
+        "/v1/acquisition/endpoints",
+        headers=auth_headers,
+        json={
+            "name": "Usenet",
+            "kind": "newznab",
+            "base_url": "http://newznab.local",
+            "api_key": "private-key",
+        },
+    )
+
+    assert response.status_code == 422
+
+
+async def test_endpoint_url_rejects_embedded_credentials(client: AsyncClient, auth_headers: dict[str, str]) -> None:
+    response = await client.post(
+        "/v1/acquisition/endpoints",
+        headers=auth_headers,
+        json={
+            "name": "Bad URL",
+            "kind": "qbittorrent",
+            "base_url": "http://user:secret@127.0.0.1:8080",
+        },
+    )
+
+    assert response.status_code == 422

From 56554924f657924875a0726ce91472e30235e1cc Mon Sep 17 00:00:00 2001
From: Eoic <karolis.strazdas@protonmail.com>
Date: Fri, 17 Jul 2026 22:50:34 +0300
Subject: [PATCH 3/8] docs: specify acquisition integration hardening

---
 ...cquisition-integration-hardening-design.md | 143 ++++++++++++++++++
 1 file changed, 143 insertions(+)
 create mode 100644 docs/superpowers/specs/2026-07-17-acquisition-integration-hardening-design.md

diff --git a/docs/superpowers/specs/2026-07-17-acquisition-integration-hardening-design.md b/docs/superpowers/specs/2026-07-17-acquisition-integration-hardening-design.md
new file mode 100644
index 0000000..b15f8e2
--- /dev/null
+++ b/docs/superpowers/specs/2026-07-17-acquisition-integration-hardening-design.md
@@ -0,0 +1,143 @@
+# Acquisition Integration Hardening Technical Specification
+
+- Status: Approved for implementation
+- Date: 2026-07-17
+- Audience: Papyrus server and Flutter client maintainers
+- Scope: `PapyrusReader/server#4` and `PapyrusReader/client#18`
+
+## Overview
+
+This specification defines the changes required to make the server-mediated torrent acquisition feature safe to enable, predictable across the HTTP boundary, and complete enough for client integration. It records the accepted decisions to disable automatic rule execution and require an explicit server-operator opt-in.
+
+## Problem
+
+The current pull requests expose useful integration primitives, but they do not yet provide a safe release boundary. Remote submission failures are returned as successful HTTP requests and ignored by the client, endpoint deletion conflicts with audit-history foreign keys, every API process starts an unsafe repeating worker, and arbitrary outbound integration requests are enabled on every server. The client also lacks capability-aware navigation and several required management states.
+
+## Goals
+
+- Require a server operator to enable acquisition explicitly.
+- Keep manual endpoint, search, submission, Arr-command, and rule execution workflows.
+- Remove automatic background rule execution from this release.
+- Preserve audit jobs while allowing integrations to be deleted safely.
+- Make failed remote operations visible to the Flutter client.
+- Support non-persisting connection tests for new and edited integrations.
+- Complete capability gating, integration forms, indexer selection, and per-release submission state.
+- Add regression coverage for each corrected behavior.
+
+## Non-goals
+
+- Scheduling or automatically claiming acquisition rules.
+- Cross-replica worker leadership or distributed job queues.
+- Usenet or Newznab support.
+- A general-purpose outbound proxy or administrator-managed hostname allowlist.
+- Redesigning unrelated authentication, profile, or synchronization code.
+
+## Constraints
+
+- Private and loopback integration URLs remain valid after an operator enables acquisition because LAN services are the primary use case.
+- Credentials stay encrypted at rest and are never returned by API schemas.
+- Existing PR dependencies and repository architecture remain unchanged.
+- Route handlers remain thin; protocol and transaction-aware behavior belongs in `papyrus/services`.
+- The Flutter feature remains disabled by default in local preferences.
+
+## Proposed design
+
+### Server activation boundary
+
+Add `acquisition_enabled: bool = False` to server settings and document `ACQUISITION_ENABLED=false` in `.env.example`. `GET /v1/acquisition/capabilities` remains available and returns `enabled=false` with empty supported-kind and command collections when disabled. Every other acquisition route uses a shared dependency that returns HTTP 404 while the feature is disabled.
+
+Enabling this setting is the operator's authorization for authenticated users to connect the server to private/LAN HTTP services. Deployments that do not trust their users must leave the setting disabled or enforce outbound restrictions outside the application.
+
+### Manual rules only
+
+Remove the acquisition worker from the FastAPI lifespan and remove the automation interval setting. Retain rule creation, listing, and `POST /rules/{rule_id}/run` for explicit execution. No code in this release executes rules on a timer.
+
+### Job outcome contract
+
+Submission and Arr-command endpoints continue returning HTTP 201 with an `AcquisitionJob` audit record. The domain outcome is the job's `status`:
+
+- `submitted`: the remote system accepted the operation.
+- `failed`: the remote adapter rejected or could not complete the operation; `error` contains a user-safe message.
+- `queued`: reserved for a future asynchronous execution model.
+
+The Flutter API client parses the job response. UI success feedback is shown only for `submitted`; `failed` becomes an operation error using the returned message.
+
+Transmission adapters validate the RPC `result` field. Deluge adapters validate JSON-RPC errors/results and choose `core.add_torrent_magnet` for magnets or `core.add_torrent_url` for HTTP(S) torrent URLs. Malformed remote payloads become controlled HTTP 502 errors instead of uncaught exceptions.
+
+### Safe endpoint deletion
+
+Audit jobs remain after integration deletion. Change `acquisition_jobs.endpoint_id` and `acquisition_rules.download_client_id` to nullable foreign keys with `ON DELETE SET NULL`. Before deleting an endpoint, remove its identifier from the owner's rule `endpoint_ids`; disable any rule that loses its download client or all configured indexers. The endpoint credentials are deleted with the endpoint.
+
+The unmerged Alembic revision is updated in place so model metadata and the migration remain aligned. Downgrade continues to drop the three acquisition tables.
+
+### Connection testing
+
+Add authenticated `POST /v1/acquisition/endpoints/test`. The request accepts either:
+
+- complete unsaved endpoint values (`kind`, `base_url`, and relevant credentials), or
+- an owned `endpoint_id` plus optional URL or credential overrides for an edit flow.
+
+The server builds an in-memory endpoint configuration, merges stored credentials only for an owned endpoint, performs a bounded protocol-specific health/authentication request, and returns `{ "ok": true }`. Failures use controlled 4xx/502 responses and never persist the supplied credentials.
+
+Protocol checks use the smallest supported request: system status/capabilities for indexers and Arr apps, authentication or session inspection for download clients.
+
+### Client availability and navigation
+
+Add a lifecycle-owned acquisition availability provider that caches capability state for the active server. It closes/replaces its HTTP client when the server changes and refreshes through the existing access-token refresh path.
+
+The Profile toggle remains visible and disabled by default. The management row appears only when the preference is enabled and the active server reports `enabled=true`. The router redirects `/acquisition` to Profile unless both conditions are true. Unknown/loading capability state is treated as unavailable.
+
+### Client management and action states
+
+The endpoint dialog:
+
+- shows API key fields for Prowlarr, Torznab, and Arr applications;
+- shows username/password for qBittorrent and Transmission;
+- shows password for Deluge;
+- validates HTTP(S) URLs without embedded credentials;
+- supports a non-persisting connection test;
+- shows inline test/save progress and error text;
+- prevents duplicate test or save requests.
+
+Search exposes enabled indexers as selectable controls. Search is enabled only when at least one indexer is selected and at least one enabled download client exists. A new search clears stale results and errors.
+
+Submission state is tracked by release and target client. An in-flight pair cannot be submitted twice, while unrelated releases remain actionable. Returned failed jobs display their server-provided error. Arr commands use the same job-outcome handling.
+
+## Interfaces and dependencies
+
+- Server configuration: `ACQUISITION_ENABLED`.
+- Existing authenticated acquisition routes retain their paths.
+- New route: `POST /v1/acquisition/endpoints/test`.
+- `AcquisitionCapabilities.enabled` becomes authoritative in the Flutter model.
+- `AcquisitionJob` becomes a typed Flutter response model.
+- No new third-party dependencies are introduced.
+
+## Testing strategy
+
+Server tests cover disabled capabilities/routes, connection-test ownership and non-persistence, failed job responses, Transmission/Deluge protocol errors, magnet versus URL dispatch, endpoint deletion with jobs/rules, and manual-only application lifespan. Migration metadata and a single Alembic head are verified alongside Ruff, Mypy, and Pytest.
+
+Client tests cover capability parsing, availability refresh and server switching, route/profile gating, conditional credential fields, connection-test serialization and error state, indexer selection/search prerequisites, per-release duplicate prevention, and submitted/failed job feedback. Dart formatting, Flutter analysis, and the full Flutter test suite must pass.
+
+## Risks and mitigations
+
+- Private-network access is intentionally powerful. It is mitigated by a disabled-by-default operator setting and documented trust requirement.
+- Making endpoint foreign keys nullable changes response schemas. The server schema and any client job model must accept missing endpoint IDs.
+- Capability checks can race server changes. Availability state is keyed to the active server URI and reset before each refresh.
+- Protocol health endpoints vary by version. Tests cover request shape, and failures are surfaced rather than treated as success.
+
+## Rollout and rollback
+
+1. Merge and deploy the server PR with the migration.
+2. Leave acquisition disabled until the operator explicitly sets `ACQUISITION_ENABLED=true`.
+3. Verify capabilities, connection testing, manual search, submission, and Arr commands against supported services.
+4. Merge and release the client PR after server verification.
+
+Rollback the client independently if UI behavior regresses. Roll back the server only before users depend on stored acquisition configuration; the downgrade deletes acquisition endpoints, encrypted credentials, rules, and job history.
+
+## Accepted decisions
+
+- Automatic background rule execution is excluded from this release.
+- Acquisition is disabled by default and enabled by server operators.
+- Enabled servers permit private/LAN integration URLs.
+- Audit jobs survive endpoint deletion through nullable foreign keys.
+- HTTP 201 plus typed job status is the submission outcome contract.

From 71ad147356ed620f77983d64eda6ea058f541c72 Mon Sep 17 00:00:00 2001
From: Eoic <karolis.strazdas@protonmail.com>
Date: Fri, 17 Jul 2026 22:56:09 +0300
Subject: [PATCH 4/8] docs: plan acquisition integration hardening

---
 ...07-17-acquisition-integration-hardening.md | 624 ++++++++++++++++++
 1 file changed, 624 insertions(+)
 create mode 100644 docs/superpowers/plans/2026-07-17-acquisition-integration-hardening.md

diff --git a/docs/superpowers/plans/2026-07-17-acquisition-integration-hardening.md b/docs/superpowers/plans/2026-07-17-acquisition-integration-hardening.md
new file mode 100644
index 0000000..b29ce7b
--- /dev/null
+++ b/docs/superpowers/plans/2026-07-17-acquisition-integration-hardening.md
@@ -0,0 +1,624 @@
+# Acquisition Integration Hardening Implementation Plan
+
+> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
+
+**Goal:** Make server PR #4 and client PR #18 safe to merge by adding an operator activation boundary, removing automatic execution, correcting persistence and remote-operation semantics, and completing the client management workflow.
+
+**Architecture:** The server remains the only component that stores credentials or contacts acquisition integrations. A disabled-by-default server capability gates every private route, protocol adapters return controlled job outcomes, and manual rules replace the unsafe worker. The Flutter client caches capability availability, consumes typed job outcomes, and owns presentation state for integration testing, search selection, and submissions.
+
+**Tech Stack:** FastAPI, Pydantic, async SQLAlchemy, PostgreSQL, Alembic, pytest, Flutter, Dart, Provider, go_router, package:http, flutter_test.
+
+---
+
+## Repository map
+
+Server checkout: `/tmp/papyrus-server-pr4-work.aXSYmc/repo`
+
+- `papyrus/config.py`: operator feature setting.
+- `papyrus/api/routes/acquisition.py`: HTTP dependencies and thin route wiring.
+- `papyrus/services/acquisition.py`: connection testing, protocol adapters, deletion cleanup, and job orchestration.
+- `papyrus/schemas/acquisition.py`: capability, connection-test, and nullable job contracts.
+- `papyrus/models/acquisition.py`: nullable foreign-key model metadata.
+- `alembic/versions/d4e5f6a7b8c9_add_acquisition.py`: new-table migration aligned with the models.
+- `papyrus/main.py`: application lifespan without acquisition automation.
+- `tests/api/routes/test_acquisition.py`: authenticated API and persistence regressions.
+- `tests/services/test_acquisition.py`: protocol adapter regressions.
+
+Client checkout: `/tmp/papyrus-client-pr18-work.hH4BiS/repo`
+
+- `app/lib/acquisition/acquisition_models.dart`: capability and job response types.
+- `app/lib/acquisition/acquisition_api_client.dart`: typed job and connection-test calls.
+- `app/lib/providers/acquisition_availability_provider.dart`: lifecycle-owned capability cache.
+- `app/lib/main.dart`: provider construction, replacement, disposal, and registration.
+- `app/lib/config/app_router.dart`: synchronous preference plus capability guard.
+- `app/lib/pages/profile_page.dart`: capability-aware management visibility.
+- `app/lib/pages/acquisition_page.dart`: integration dialog, search selection, and per-release state.
+- `app/test/acquisition/`: API, model, and availability-provider tests.
+- `app/test/config/app_router_test.dart`: route guard regression.
+- `app/test/pages/profile_storage_sync_test.dart`: Profile visibility regression.
+- `app/test/pages/acquisition_page_test.dart`: management and action-state widget regressions.
+
+### Task 1: Add the server activation boundary and remove automation
+
+**Files:**
+- Modify: `.env.example`
+- Modify: `papyrus/config.py`
+- Modify: `papyrus/api/routes/acquisition.py`
+- Modify: `papyrus/main.py`
+- Modify: `tests/api/routes/test_acquisition.py`
+
+- [ ] **Step 1: Write failing activation tests**
+
+Add tests that temporarily set the singleton application setting and restore it in `finally`:
+
+```python
+async def test_disabled_capabilities_are_empty(client: AsyncClient, monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(app_settings, "acquisition_enabled", False)
+
+    response = await client.get("/v1/acquisition/capabilities")
+
+    assert response.status_code == 200
+    assert response.json() == {
+        "enabled": False,
+        "endpoint_kinds": [],
+        "indexer_kinds": [],
+        "download_client_kinds": [],
+        "arr_kinds": [],
+        "arr_commands": {},
+    }
+
+
+async def test_disabled_acquisition_routes_are_not_available(
+    client: AsyncClient,
+    auth_headers: dict[str, str],
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setattr(app_settings, "acquisition_enabled", False)
+
+    response = await client.get("/v1/acquisition/endpoints", headers=auth_headers)
+
+    assert response.status_code == 404
+```
+
+- [ ] **Step 2: Verify the tests fail for the expected reason**
+
+Run: `uv run pytest tests/api/routes/test_acquisition.py -k "disabled" -q`
+
+Expected: capability assertion fails because `enabled` is true and endpoints remain accessible.
+
+- [ ] **Step 3: Implement the setting, dependency, and manual-only lifespan**
+
+Add the setting and route dependency:
+
+```python
+class Settings(BaseSettings):
+    acquisition_enabled: bool = False
+
+
+def require_acquisition_enabled() -> None:
+    if not get_settings().acquisition_enabled:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Acquisition is disabled")
+```
+
+Attach `Depends(require_acquisition_enabled)` to every acquisition route except `/capabilities`. Return empty capability collections when disabled. Remove `acquisition_worker`, its task lifecycle, the `run_enabled_rules` import, and `acquisition_automation_interval_seconds`. Document `ACQUISITION_ENABLED=false` in `.env.example`.
+
+- [ ] **Step 4: Verify activation tests pass**
+
+Run: `uv run pytest tests/api/routes/test_acquisition.py -k "capabilities or disabled" -q`
+
+Expected: all selected tests pass.
+
+- [ ] **Step 5: Commit the activation boundary**
+
+```bash
+git add .env.example papyrus/config.py papyrus/api/routes/acquisition.py papyrus/main.py tests/api/routes/test_acquisition.py
+git commit -m "fix: gate acquisition behind server opt-in"
+```
+
+### Task 2: Correct remote adapter and job outcome behavior
+
+**Files:**
+- Modify: `papyrus/services/acquisition.py`
+- Modify: `papyrus/api/routes/acquisition.py`
+- Create: `tests/services/test_acquisition.py`
+- Modify: `tests/api/routes/test_acquisition.py`
+
+- [ ] **Step 1: Write failing Transmission and Deluge tests**
+
+Use real adapter functions with only `_request` replaced:
+
+```python
+async def test_transmission_rejects_rpc_failure(monkeypatch: pytest.MonkeyPatch) -> None:
+    async def request(*args: object, **kwargs: object) -> tuple[int, dict[str, str], bytes]:
+        return 200, {}, b'{"result":"invalid or corrupt torrent file","arguments":{}}'
+
+    monkeypatch.setattr(acquisition, "_request", request)
+
+    with pytest.raises(HTTPException) as exc_info:
+        await acquisition.submit_to_client(_endpoint("transmission"), "magnet:?xt=urn:btih:test", None, None)
+
+    assert exc_info.value.status_code == 502
+
+
+async def test_deluge_uses_url_method_for_http_torrent(monkeypatch: pytest.MonkeyPatch) -> None:
+    bodies: list[dict[str, object]] = []
+
+    async def request(*args: object, **kwargs: object) -> tuple[int, dict[str, str], bytes]:
+        bodies.append(json.loads(cast(bytes, kwargs["body"])))
+        if len(bodies) == 1:
+            return 200, {"Set-Cookie": "_session_id=test"}, b'{"result":true,"error":null,"id":1}'
+        return 200, {}, b'{"result":"torrent-id","error":null,"id":2}'
+
+    monkeypatch.setattr(acquisition, "_request", request)
+
+    await acquisition.submit_to_client(_endpoint("deluge"), "https://indexer.test/release.torrent", None, None)
+
+    assert bodies[1]["method"] == "core.add_torrent_url"
+```
+
+Add a route regression asserting a rejected adapter produces a persisted `status="failed"` job with HTTP 201 and a safe `error` string.
+
+- [ ] **Step 2: Verify the adapter tests fail**
+
+Run: `uv run pytest tests/services/test_acquisition.py -q`
+
+Expected: Transmission does not raise and Deluge uses `core.add_torrent_magnet` for the URL.
+
+- [ ] **Step 3: Implement strict remote response parsing**
+
+Add focused helpers:
+
+```python
+def _json_object(payload: bytes, integration: str) -> dict[str, object]:
+    try:
+        value = json.loads(payload)
+    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
+        raise HTTPException(status_code=502, detail=f"{integration} returned invalid JSON") from exc
+    if not isinstance(value, dict):
+        raise HTTPException(status_code=502, detail=f"{integration} returned an invalid response")
+    return value
+
+
+def _require_deluge_result(payload: bytes) -> object:
+    response = _json_object(payload, "Deluge")
+    if response.get("error") is not None or response.get("result") in {None, False}:
+        raise HTTPException(status_code=502, detail="Deluge rejected the request")
+    return response["result"]
+```
+
+Require Transmission `result == "success"`. Validate Deluge login and add responses. Choose the Deluge method from `download_url.startswith("magnet:")`. Use `_json_object` for Prowlarr and Arr payloads so malformed responses remain controlled failures.
+
+- [ ] **Step 4: Verify service and route tests pass**
+
+Run: `uv run pytest tests/services/test_acquisition.py tests/api/routes/test_acquisition.py -q`
+
+Expected: all selected tests pass.
+
+- [ ] **Step 5: Commit adapter corrections**
+
+```bash
+git add papyrus/services/acquisition.py papyrus/api/routes/acquisition.py tests/services/test_acquisition.py tests/api/routes/test_acquisition.py
+git commit -m "fix: validate acquisition client responses"
+```
+
+### Task 3: Make endpoint deletion preserve audit history
+
+**Files:**
+- Modify: `papyrus/models/acquisition.py`
+- Modify: `papyrus/schemas/acquisition.py`
+- Modify: `papyrus/services/acquisition.py`
+- Modify: `papyrus/api/routes/acquisition.py`
+- Modify: `alembic/versions/d4e5f6a7b8c9_add_acquisition.py`
+- Modify: `tests/api/routes/test_acquisition.py`
+
+- [ ] **Step 1: Write the failing deletion regression**
+
+Create an endpoint, a job referencing it, and a rule that uses it as both an indexer and download client. Delete through the API and assert:
+
+```python
+assert response.status_code == 204
+assert persisted_job.endpoint_id is None
+assert persisted_rule.download_client_id is None
+assert persisted_rule.endpoint_ids == []
+assert persisted_rule.enabled is False
+```
+
+- [ ] **Step 2: Verify deletion fails with a foreign-key error**
+
+Run: `uv run pytest tests/api/routes/test_acquisition.py -k "delete_endpoint_preserves" -q`
+
+Expected: the request returns 500 or raises `IntegrityError` because dependent rows still reference the endpoint.
+
+- [ ] **Step 3: Implement nullable foreign keys and cleanup**
+
+Update model metadata and the unmerged migration:
+
+```python
+download_client_id: Mapped[UUID | None] = mapped_column(
+    Uuid,
+    ForeignKey("acquisition_endpoints.endpoint_id", ondelete="SET NULL"),
+    nullable=True,
+)
+endpoint_id: Mapped[UUID | None] = mapped_column(
+    Uuid,
+    ForeignKey("acquisition_endpoints.endpoint_id", ondelete="SET NULL"),
+    nullable=True,
+)
+```
+
+Make `AcquisitionJob.endpoint_id` optional in the response schema. Add a service that locks/loads the owner's rules, removes the deleted ID from JSON lists, clears matching download clients, disables unusable rules, deletes the endpoint, and commits once. Delegate the route to that service.
+
+- [ ] **Step 4: Verify deletion and migration metadata**
+
+Run: `uv run pytest tests/api/routes/test_acquisition.py -k "delete_endpoint_preserves" -q`
+
+Run: `uv run alembic heads`
+
+Expected: the regression passes and `d4e5f6a7b8c9 (head)` is the only head.
+
+- [ ] **Step 5: Commit persistence corrections**
+
+```bash
+git add papyrus/models/acquisition.py papyrus/schemas/acquisition.py papyrus/services/acquisition.py papyrus/api/routes/acquisition.py alembic/versions/d4e5f6a7b8c9_add_acquisition.py tests/api/routes/test_acquisition.py
+git commit -m "fix: preserve acquisition jobs on endpoint deletion"
+```
+
+### Task 4: Add non-persisting connection tests
+
+**Files:**
+- Modify: `papyrus/schemas/acquisition.py`
+- Modify: `papyrus/services/acquisition.py`
+- Modify: `papyrus/api/routes/acquisition.py`
+- Modify: `tests/api/routes/test_acquisition.py`
+- Modify: `tests/services/test_acquisition.py`
+
+- [ ] **Step 1: Write failing API tests for unsaved and edited connections**
+
+Add tests that call `/v1/acquisition/endpoints/test` with an unsaved Prowlarr payload and with an owned endpoint ID plus credential overrides. Replace `test_endpoint_connection` at the route boundary and capture the transient endpoint. Assert no new `AcquisitionEndpoint` row is persisted and another user's endpoint returns 404.
+
+```python
+assert response.status_code == 200
+assert response.json() == {"ok": True}
+assert captured.kind == "prowlarr"
+assert decrypt_secret_payload(captured.credentials["encrypted"])["api_key"] == "override"
+assert await endpoint_count(db_session) == before_count
+```
+
+- [ ] **Step 2: Verify the connection-test route is missing**
+
+Run: `uv run pytest tests/api/routes/test_acquisition.py -k "test_connection" -q`
+
+Expected: requests return 404.
+
+- [ ] **Step 3: Implement schemas, transient configuration, and protocol checks**
+
+Add:
+
+```python
+class AcquisitionEndpointTest(BaseModel):
+    endpoint_id: UUID | None = None
+    kind: EndpointKind | None = None
+    base_url: HttpUrl | None = None
+    api_key: SecretStr | None = None
+    username: SecretStr | None = None
+    password: SecretStr | None = None
+
+
+class AcquisitionEndpointTestResult(BaseModel):
+    ok: bool
+```
+
+Validate that `kind` and `base_url` are present without `endpoint_id`. For an edit, load only an owned endpoint and merge supplied values with decrypted stored credentials in memory. Implement protocol-specific bounded checks and return `AcquisitionEndpointTestResult(ok=True)` only after the remote accepts the request.
+
+- [ ] **Step 4: Verify connection-test coverage**
+
+Run: `uv run pytest tests/api/routes/test_acquisition.py tests/services/test_acquisition.py -q`
+
+Expected: all selected tests pass.
+
+- [ ] **Step 5: Commit connection testing**
+
+```bash
+git add papyrus/schemas/acquisition.py papyrus/services/acquisition.py papyrus/api/routes/acquisition.py tests/api/routes/test_acquisition.py tests/services/test_acquisition.py
+git commit -m "feat: test acquisition connections without saving"
+```
+
+### Task 5: Verify the complete server branch
+
+**Files:**
+- Review all server files changed by Tasks 1-4.
+
+- [ ] **Step 1: Run formatting and lint checks**
+
+Run: `uv run ruff format --check .`
+
+Run: `uv run ruff check .`
+
+Expected: both commands exit 0.
+
+- [ ] **Step 2: Run type checking**
+
+Run: `uv run mypy .`
+
+Expected: exit 0 with no type errors.
+
+- [ ] **Step 3: Run the full server test suite**
+
+Run: `uv run pytest`
+
+Expected: all tests pass.
+
+- [ ] **Step 4: Verify migration shape**
+
+Run: `uv run alembic heads`
+
+Run against a disposable test database: `uv run alembic upgrade head`
+
+Expected: one head and a successful upgrade creating nullable `SET NULL` acquisition foreign keys.
+
+### Task 6: Add typed client capability, job, and connection-test contracts
+
+**Files:**
+- Modify: `app/lib/acquisition/acquisition_models.dart`
+- Modify: `app/lib/acquisition/acquisition_api_client.dart`
+- Modify: `app/test/acquisition/acquisition_models_test.dart`
+- Modify: `app/test/acquisition/acquisition_api_client_test.dart`
+
+- [ ] **Step 1: Write failing model and API tests**
+
+Add assertions for `enabled`, nullable job endpoint IDs, failed jobs, and connection-test payloads:
+
+```dart
+expect(AcquisitionCapabilities.fromJson({'enabled': false}).enabled, isFalse);
+
+final job = AcquisitionJob.fromJson({
+  'job_id': 'job-1',
+  'endpoint_id': null,
+  'rule_id': null,
+  'title': 'Release',
+  'download_url': 'magnet:?xt=urn:btih:test',
+  'status': 'failed',
+  'error': 'Transmission rejected the release',
+});
+expect(job.isSubmitted, isFalse);
+expect(job.error, 'Transmission rejected the release');
+```
+
+Expect `submitRelease` and `runArrCommand` to return `AcquisitionJob`. Expect `testEndpoint` to POST `/v1/acquisition/endpoints/test` with either unsaved values or `endpoint_id` and overrides.
+
+- [ ] **Step 2: Verify the new contract tests fail**
+
+Run: `flutter test --no-pub test/acquisition/acquisition_models_test.dart test/acquisition/acquisition_api_client_test.dart`
+
+Expected: missing `enabled`, `AcquisitionJob`, and `testEndpoint` APIs fail compilation.
+
+- [ ] **Step 3: Implement typed contracts**
+
+Add immutable model fields and parsing:
+
+```dart
+class AcquisitionJob {
+  final String id;
+  final String? endpointId;
+  final String status;
+  final String? error;
+
+  bool get isSubmitted => status == 'submitted';
+}
+```
+
+Return parsed jobs from submission methods and add the connection-test call. Keep all authentication errors represented by `AuthApiException`.
+
+- [ ] **Step 4: Verify model and API tests pass**
+
+Run: `flutter test --no-pub test/acquisition/acquisition_models_test.dart test/acquisition/acquisition_api_client_test.dart`
+
+Expected: all selected tests pass.
+
+- [ ] **Step 5: Commit client contracts**
+
+```bash
+git add app/lib/acquisition/acquisition_models.dart app/lib/acquisition/acquisition_api_client.dart app/test/acquisition/acquisition_models_test.dart app/test/acquisition/acquisition_api_client_test.dart
+git commit -m "fix: consume acquisition capability and job outcomes"
+```
+
+### Task 7: Add lifecycle-owned capability availability and route gating
+
+**Files:**
+- Create: `app/lib/providers/acquisition_availability_provider.dart`
+- Modify: `app/lib/main.dart`
+- Modify: `app/lib/config/app_router.dart`
+- Modify: `app/lib/pages/profile_page.dart`
+- Create: `app/test/acquisition/acquisition_availability_provider_test.dart`
+- Modify: `app/test/config/app_router_test.dart`
+- Modify: `app/test/pages/profile_storage_sync_test.dart`
+
+- [ ] **Step 1: Write failing provider, router, and Profile tests**
+
+Define provider tests with an injected capability loader:
+
+```dart
+final provider = AcquisitionAvailabilityProvider(
+  loadCapabilities: (_) async => const AcquisitionCapabilities(
+    enabled: true,
+    endpointKinds: [],
+    indexerKinds: [],
+    downloadClientKinds: [],
+    arrKinds: [],
+    arrCommands: {},
+  ),
+);
+await provider.refresh(serverBaseUri);
+expect(provider.isAvailableFor(serverBaseUri), isTrue);
+```
+
+Extend router tests so `/acquisition` redirects when the preference is on but availability is false, then permits the route after availability becomes true. Extend Profile tests so the toggle remains visible while `Torrent & automation` appears only for enabled preference plus available server.
+
+- [ ] **Step 2: Verify availability tests fail**
+
+Run: `flutter test --no-pub test/acquisition/acquisition_availability_provider_test.dart test/config/app_router_test.dart test/pages/profile_storage_sync_test.dart`
+
+Expected: the provider type is missing and router/Profile only check local preferences.
+
+- [ ] **Step 3: Implement the availability provider and wiring**
+
+Create a `ChangeNotifier` keyed by active server URI with `unknown`, `loading`, `available`, and `unavailable` state. It owns one `AcquisitionApiClient`, closes it on server change/disposal, and loads through `AuthProvider.withFreshAccessToken`.
+
+Construct it in `_PapyrusState`, pass it to `AppRouter`, register it in `MultiProvider`, refresh it after auth/server changes, and dispose it. Include it in `Listenable.merge`. Guard with:
+
+```dart
+if (location == '/acquisition' &&
+    (!preferencesProvider.acquisitionEnabled ||
+     !acquisitionAvailabilityProvider.isAvailableFor(activeServerUri))) {
+  return '/profile';
+}
+```
+
+Profile watches the provider and gates only the management row/button, not the opt-in toggle.
+
+- [ ] **Step 4: Verify provider, router, and Profile tests pass**
+
+Run: `flutter test --no-pub test/acquisition/acquisition_availability_provider_test.dart test/config/app_router_test.dart test/pages/profile_storage_sync_test.dart`
+
+Expected: all selected tests pass.
+
+- [ ] **Step 5: Commit availability gating**
+
+```bash
+git add app/lib/providers/acquisition_availability_provider.dart app/lib/main.dart app/lib/config/app_router.dart app/lib/pages/profile_page.dart app/test/acquisition/acquisition_availability_provider_test.dart app/test/config/app_router_test.dart app/test/pages/profile_storage_sync_test.dart
+git commit -m "fix: gate acquisition UI by server capability"
+```
+
+### Task 8: Complete integration testing and form state
+
+**Files:**
+- Modify: `app/lib/pages/acquisition_page.dart`
+- Create: `app/test/pages/acquisition_page_test.dart`
+
+- [ ] **Step 1: Write failing dialog widget tests**
+
+Pump `AcquisitionPage` with injected/fake API behavior and assert:
+
+- Prowlarr shows API key but not username/password.
+- qBittorrent shows username/password but not API key.
+- Deluge shows password only.
+- Test connection displays progress, prevents a second request, and renders a returned error inside the dialog.
+- Save is disabled while a test/save request is active.
+
+- [ ] **Step 2: Verify dialog tests fail**
+
+Run: `flutter test --no-pub test/pages/acquisition_page_test.dart --plain-name "integration dialog"`
+
+Expected: all credential fields are currently unconditional and there is no connection-test action.
+
+- [ ] **Step 3: Implement conditional fields and inline operation state**
+
+Extract a focused stateful `_EndpointDialog` widget with injected callbacks. Derive fields from `AcquisitionEndpointKind` and maintain distinct `testing`, `saving`, and `error` state. Send current form values to `testEndpoint`; for edits include `endpointId` so blank credentials preserve stored secrets.
+
+- [ ] **Step 4: Verify dialog tests pass**
+
+Run: `flutter test --no-pub test/pages/acquisition_page_test.dart --plain-name "integration dialog"`
+
+Expected: all selected tests pass.
+
+- [ ] **Step 5: Commit integration form completion**
+
+```bash
+git add app/lib/pages/acquisition_page.dart app/test/pages/acquisition_page_test.dart
+git commit -m "feat: test acquisition integrations from the client"
+```
+
+### Task 9: Complete search selection and per-release outcomes
+
+**Files:**
+- Modify: `app/lib/pages/acquisition_page.dart`
+- Modify: `app/test/pages/acquisition_page_test.dart`
+
+- [ ] **Step 1: Write failing search and submission widget tests**
+
+Assert that search remains disabled without both a selected enabled indexer and an enabled download client. Select one of two indexer chips and verify only its ID is sent. Start one submission and assert only that release/client pair is disabled. Complete with a failed job and assert the backend error is displayed instead of success.
+
+- [ ] **Step 2: Verify state tests fail**
+
+Run: `flutter test --no-pub test/pages/acquisition_page_test.dart --plain-name "search and submission"`
+
+Expected: no indexer controls exist, search ignores client availability, and the page uses one global submission flag.
+
+- [ ] **Step 3: Implement selected indexers and pair-scoped submissions**
+
+Maintain `Set<String> _selectedIndexerIds` and initialize newly loaded enabled indexers only when no explicit selection exists. Render `FilterChip` controls. Compute `canSearch` from selected enabled indexers plus enabled clients.
+
+Maintain `Set<String> _submittingKeys` where the key combines release URL and client ID. Add before awaiting, remove in `finally`, and pass disabled state to each release tile. Inspect the returned job and show success only for `submitted`; otherwise show `job.error ?? 'Submission failed.'`. Apply the same outcome check to Arr commands.
+
+- [ ] **Step 4: Verify state tests pass**
+
+Run: `flutter test --no-pub test/pages/acquisition_page_test.dart`
+
+Expected: all acquisition page tests pass.
+
+- [ ] **Step 5: Commit search and submission completion**
+
+```bash
+git add app/lib/pages/acquisition_page.dart app/test/pages/acquisition_page_test.dart
+git commit -m "fix: complete acquisition search and submission states"
+```
+
+### Task 10: Verify the complete client branch
+
+**Files:**
+- Review all client files changed by Tasks 6-9.
+
+- [ ] **Step 1: Format and verify formatting**
+
+Run: `dart format app/lib app/test`
+
+Run: `dart format --output=none --set-exit-if-changed app/lib app/test`
+
+Expected: the second command exits 0 with zero changed files.
+
+- [ ] **Step 2: Run static analysis**
+
+Run from `app/`: `flutter analyze --no-pub`
+
+Expected: exit 0 with no issues.
+
+- [ ] **Step 3: Run the full Flutter test suite**
+
+Run from `app/`: `flutter test --no-pub`
+
+Expected: all tests pass; only explicitly skipped integration tests remain skipped.
+
+### Task 11: Final cross-repository verification and publication
+
+**Files:**
+- Review both repository diffs and commit histories.
+
+- [ ] **Step 1: Confirm clean worktrees and intended commits**
+
+Run in each checkout: `git status --short --branch`
+
+Run in each checkout: `git log --oneline --decorate -8`
+
+Expected: clean worktrees with focused commits on top of the original PR heads.
+
+- [ ] **Step 2: Re-run the narrow cross-contract tests**
+
+Server: `uv run pytest tests/api/routes/test_acquisition.py tests/services/test_acquisition.py -q`
+
+Client from `app/`: `flutter test --no-pub test/acquisition test/config/app_router_test.dart test/pages/acquisition_page_test.dart test/pages/profile_storage_sync_test.dart`
+
+Expected: both commands exit 0.
+
+- [ ] **Step 3: Push to the existing PR source branches**
+
+After confirming the source branch names and remote permissions:
+
+```bash
+git push github HEAD:feature/torrent-acquisition
+```
+
+Run once from each repository checkout. Do not force-push.
+
+- [ ] **Step 4: Verify GitHub heads and checks**
+
+Confirm server PR #4 and client PR #18 point at the new commits. Report CI as pending until GitHub Actions completes; do not claim remote success from local checks alone.

From 2a9577a80468ed53badd3ad778c641e76487eb4d Mon Sep 17 00:00:00 2001
From: Eoic <karolis.strazdas@protonmail.com>
Date: Fri, 17 Jul 2026 23:06:18 +0300
Subject: [PATCH 5/8] fix: gate acquisition behind server opt-in

---
 .env.example                         |  2 +-
 papyrus/api/routes/acquisition.py    | 46 +++++++++++++++++++++-------
 papyrus/config.py                    |  2 +-
 papyrus/main.py                      | 20 ++----------
 tests/api/routes/test_acquisition.py | 34 ++++++++++++++++++++
 5 files changed, 73 insertions(+), 31 deletions(-)

diff --git a/.env.example b/.env.example
index d60a99b..8d00aab 100644
--- a/.env.example
+++ b/.env.example
@@ -55,7 +55,7 @@ DEV_PAGES_MANIFEST_PATH=frontend/dev-pages/dist/.vite/manifest.json
 
 # Execute enabled acquisition rules on this interval (seconds). Set a value that
 # respects each indexer's API limits.
-ACQUISITION_AUTOMATION_INTERVAL_SECONDS=300
+ACQUISITION_ENABLED=false
 
 # Prefer file-based PowerSync keys for local development.
 # Generate them with: ./scripts/generate_dev_powersync_keys.sh
diff --git a/papyrus/api/routes/acquisition.py b/papyrus/api/routes/acquisition.py
index bcbbadf..0a333c7 100644
--- a/papyrus/api/routes/acquisition.py
+++ b/papyrus/api/routes/acquisition.py
@@ -8,6 +8,7 @@
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from papyrus.api.deps import CurrentUserId
+from papyrus.config import get_settings
 from papyrus.core.database import get_db
 from papyrus.core.security import decrypt_secret_payload, encrypt_secret_payload
 from papyrus.models.acquisition import AcquisitionEndpoint as AcquisitionEndpointModel
@@ -38,6 +39,14 @@
 DbSession = Annotated[AsyncSession, Depends(get_db)]
 
 
+def require_acquisition_enabled() -> None:
+    if not get_settings().acquisition_enabled:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Not found")
+
+
+protected_router = APIRouter(dependencies=[Depends(require_acquisition_enabled)])
+
+
 def _credentials(api_key: str | None, username: str | None, password: str | None) -> dict[str, str] | None:
     values = {
         key: value for key, value in {"api_key": api_key, "username": username, "password": password}.items() if value
@@ -60,6 +69,16 @@ def _merge_credentials(endpoint: AcquisitionEndpointModel, replacement: dict[str
 
 @router.get("/capabilities", response_model=AcquisitionCapabilities)
 async def acquisition_capabilities() -> AcquisitionCapabilities:
+    if not get_settings().acquisition_enabled:
+        return AcquisitionCapabilities(
+            enabled=False,
+            endpoint_kinds=[],
+            indexer_kinds=[],
+            download_client_kinds=[],
+            arr_kinds=[],
+            arr_commands={},
+        )
+
     return AcquisitionCapabilities(
         endpoint_kinds=[
             "qbittorrent",
@@ -86,7 +105,7 @@ async def acquisition_capabilities() -> AcquisitionCapabilities:
     )
 
 
-@router.get("/endpoints", response_model=list[AcquisitionEndpoint])
+@protected_router.get("/endpoints", response_model=list[AcquisitionEndpoint])
 async def list_endpoints(user_id: CurrentUserId, db: DbSession) -> list[AcquisitionEndpointModel]:
     result = await db.execute(
         select(AcquisitionEndpointModel)
@@ -96,7 +115,7 @@ async def list_endpoints(user_id: CurrentUserId, db: DbSession) -> list[Acquisit
     return list(result.scalars())
 
 
-@router.post("/endpoints", response_model=AcquisitionEndpoint, status_code=status.HTTP_201_CREATED)
+@protected_router.post("/endpoints", response_model=AcquisitionEndpoint, status_code=status.HTTP_201_CREATED)
 async def create_endpoint(
     user_id: CurrentUserId, request: AcquisitionEndpointCreate, db: DbSession
 ) -> AcquisitionEndpointModel:
@@ -118,7 +137,7 @@ async def create_endpoint(
     return endpoint
 
 
-@router.patch("/endpoints/{endpoint_id}", response_model=AcquisitionEndpoint)
+@protected_router.patch("/endpoints/{endpoint_id}", response_model=AcquisitionEndpoint)
 async def update_endpoint(
     user_id: CurrentUserId, endpoint_id: UUID, request: AcquisitionEndpointUpdate, db: DbSession
 ) -> AcquisitionEndpointModel:
@@ -137,14 +156,14 @@ async def update_endpoint(
     return endpoint
 
 
-@router.delete("/endpoints/{endpoint_id}", status_code=status.HTTP_204_NO_CONTENT)
+@protected_router.delete("/endpoints/{endpoint_id}", status_code=status.HTTP_204_NO_CONTENT)
 async def delete_endpoint(user_id: CurrentUserId, endpoint_id: UUID, db: DbSession) -> None:
     endpoint = await owned_endpoint(db, user_id, endpoint_id)
     await db.delete(endpoint)
     await db.commit()
 
 
-@router.post("/search", response_model=list[Release])
+@protected_router.post("/search", response_model=list[Release])
 async def search_releases(user_id: CurrentUserId, request: SearchRequest, db: DbSession) -> list[Release]:
     statement = select(AcquisitionEndpointModel).where(
         AcquisitionEndpointModel.owner_user_id == user_id,
@@ -160,7 +179,7 @@ async def search_releases(user_id: CurrentUserId, request: SearchRequest, db: Db
     return releases
 
 
-@router.post("/submissions", response_model=AcquisitionJob, status_code=status.HTTP_201_CREATED)
+@protected_router.post("/submissions", response_model=AcquisitionJob, status_code=status.HTTP_201_CREATED)
 async def submit_release(user_id: CurrentUserId, request: SubmitRequest, db: DbSession) -> AcquisitionJobModel:
     endpoint = await owned_endpoint(db, user_id, request.endpoint_id)
     job = AcquisitionJobModel(
@@ -180,7 +199,9 @@ async def submit_release(user_id: CurrentUserId, request: SubmitRequest, db: DbS
     return job
 
 
-@router.post("/arr/{endpoint_id}/commands", response_model=AcquisitionJob, status_code=status.HTTP_201_CREATED)
+@protected_router.post(
+    "/arr/{endpoint_id}/commands", response_model=AcquisitionJob, status_code=status.HTTP_201_CREATED
+)
 async def run_arr_command(
     user_id: CurrentUserId, endpoint_id: UUID, request: ArrCommandRequest, db: DbSession
 ) -> AcquisitionJobModel:
@@ -204,7 +225,7 @@ async def run_arr_command(
     return job
 
 
-@router.get("/jobs", response_model=list[AcquisitionJob])
+@protected_router.get("/jobs", response_model=list[AcquisitionJob])
 async def list_jobs(user_id: CurrentUserId, db: DbSession) -> list[AcquisitionJobModel]:
     result = await db.execute(
         select(AcquisitionJobModel)
@@ -214,13 +235,13 @@ async def list_jobs(user_id: CurrentUserId, db: DbSession) -> list[AcquisitionJo
     return list(result.scalars())
 
 
-@router.get("/rules", response_model=list[AcquisitionRule])
+@protected_router.get("/rules", response_model=list[AcquisitionRule])
 async def list_rules(user_id: CurrentUserId, db: DbSession) -> list[AcquisitionRuleModel]:
     result = await db.execute(select(AcquisitionRuleModel).where(AcquisitionRuleModel.owner_user_id == user_id))
     return list(result.scalars())
 
 
-@router.post("/rules", response_model=AcquisitionRule, status_code=status.HTTP_201_CREATED)
+@protected_router.post("/rules", response_model=AcquisitionRule, status_code=status.HTTP_201_CREATED)
 async def create_rule(user_id: CurrentUserId, request: AcquisitionRuleCreate, db: DbSession) -> AcquisitionRuleModel:
     await owned_endpoint(db, user_id, request.download_client_id)
     rule = AcquisitionRuleModel(
@@ -238,7 +259,7 @@ async def create_rule(user_id: CurrentUserId, request: AcquisitionRuleCreate, db
     return rule
 
 
-@router.post("/rules/{rule_id}/run", response_model=list[AcquisitionJob])
+@protected_router.post("/rules/{rule_id}/run", response_model=list[AcquisitionJob])
 async def run_acquisition_rule(user_id: CurrentUserId, rule_id: UUID, db: DbSession) -> list[AcquisitionJobModel]:
     result = await db.execute(
         select(AcquisitionRuleModel).where(
@@ -249,3 +270,6 @@ async def run_acquisition_rule(user_id: CurrentUserId, rule_id: UUID, db: DbSess
     if rule is None:
         raise HTTPException(status_code=404, detail="Acquisition rule not found")
     return await run_rule(db, rule)
+
+
+router.include_router(protected_router)
diff --git a/papyrus/config.py b/papyrus/config.py
index b6aa29d..2e58f92 100644
--- a/papyrus/config.py
+++ b/papyrus/config.py
@@ -71,7 +71,7 @@ class Settings(BaseSettings):
     dev_pages_use_vite: bool = False
     dev_pages_vite_url: str = "http://localhost:5173"
     dev_pages_manifest_path: str = "frontend/dev-pages/dist/.vite/manifest.json"
-    acquisition_automation_interval_seconds: int = 300
+    acquisition_enabled: bool = False
 
     @field_validator("debug", mode="before")
     @classmethod
diff --git a/papyrus/main.py b/papyrus/main.py
index 9bec526..1f3d84d 100644
--- a/papyrus/main.py
+++ b/papyrus/main.py
@@ -1,9 +1,8 @@
 """FastAPI application factory and configuration."""
 
-import asyncio
 import logging
 from collections.abc import AsyncGenerator
-from contextlib import asynccontextmanager, suppress
+from contextlib import asynccontextmanager
 
 import structlog
 from fastapi import FastAPI, Request, status
@@ -17,11 +16,9 @@
 
 from papyrus.api.routes import api_router, include_debug_routers
 from papyrus.config import get_settings
-from papyrus.core.database import async_session_maker
 from papyrus.core.dev_pages import DEV_PAGES_DIST_DIR, DEV_PAGES_STATIC_URL
 from papyrus.core.exceptions import AppError
 from papyrus.core.rate_limit import limiter
-from papyrus.services.acquisition import run_enabled_rules
 
 settings = get_settings()
 
@@ -43,20 +40,7 @@ def configure_logging() -> None:
 @asynccontextmanager
 async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
     """Application lifespan events."""
-
-    async def acquisition_worker() -> None:
-        while True:
-            async with async_session_maker() as session:
-                await run_enabled_rules(session)
-            await asyncio.sleep(settings.acquisition_automation_interval_seconds)
-
-    worker = asyncio.create_task(acquisition_worker())
-    try:
-        yield
-    finally:
-        worker.cancel()
-        with suppress(asyncio.CancelledError):
-            await worker
+    yield
 
 
 def create_app() -> FastAPI:
diff --git a/tests/api/routes/test_acquisition.py b/tests/api/routes/test_acquisition.py
index d471034..7a0aa54 100644
--- a/tests/api/routes/test_acquisition.py
+++ b/tests/api/routes/test_acquisition.py
@@ -1,13 +1,47 @@
 """Tests for private BitTorrent acquisition configuration."""
 
+import pytest
 from httpx import AsyncClient
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from papyrus.core.security import decrypt_secret_payload
+from papyrus.main import settings as app_settings
 from papyrus.models.acquisition import AcquisitionEndpoint
 
 
+@pytest.fixture(autouse=True)
+def enable_acquisition(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(app_settings, "acquisition_enabled", True, raising=False)
+
+
+async def test_disabled_capabilities_hide_acquisition_scope(client: AsyncClient) -> None:
+    app_settings.acquisition_enabled = False
+
+    response = await client.get("/v1/acquisition/capabilities")
+
+    assert response.status_code == 200
+    assert response.json() == {
+        "enabled": False,
+        "endpoint_kinds": [],
+        "indexer_kinds": [],
+        "download_client_kinds": [],
+        "arr_kinds": [],
+        "arr_commands": {},
+    }
+
+
+async def test_disabled_acquisition_routes_are_not_found(
+    client: AsyncClient,
+    auth_headers: dict[str, str],
+) -> None:
+    app_settings.acquisition_enabled = False
+
+    response = await client.get("/v1/acquisition/endpoints", headers=auth_headers)
+
+    assert response.status_code == 404
+
+
 async def test_capabilities_advertise_torrent_only_scope(client: AsyncClient, auth_headers: dict[str, str]) -> None:
     response = await client.get("/v1/acquisition/capabilities", headers=auth_headers)
 

From 1bdf4cf299c032af53b5cfdf83abb220bfb046a8 Mon Sep 17 00:00:00 2001
From: Eoic <karolis.strazdas@protonmail.com>
Date: Fri, 17 Jul 2026 23:08:45 +0300
Subject: [PATCH 6/8] fix: validate acquisition client responses

---
 papyrus/services/acquisition.py      | 69 ++++++++++++++++++--
 tests/api/routes/test_acquisition.py | 36 +++++++++++
 tests/services/test_acquisition.py   | 97 ++++++++++++++++++++++++++++
 3 files changed, 195 insertions(+), 7 deletions(-)
 create mode 100644 tests/services/test_acquisition.py

diff --git a/papyrus/services/acquisition.py b/papyrus/services/acquisition.py
index e6a3cfd..af0615f 100644
--- a/papyrus/services/acquisition.py
+++ b/papyrus/services/acquisition.py
@@ -54,6 +54,35 @@ def _credentials(endpoint: AcquisitionEndpoint) -> dict[str, str]:
         raise HTTPException(status_code=500, detail="Stored integration credentials are invalid") from exc
 
 
+def _json_value(payload: bytes, integration: str) -> object:
+    try:
+        return json.loads(payload)
+    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
+        raise HTTPException(status_code=502, detail=f"{integration} returned invalid JSON") from exc
+
+
+def _json_object(payload: bytes, integration: str) -> dict[str, object]:
+    value = _json_value(payload, integration)
+    if not isinstance(value, dict):
+        raise HTTPException(status_code=502, detail=f"{integration} returned an invalid response")
+    return value
+
+
+def _json_array(payload: bytes, integration: str) -> list[dict[str, object]]:
+    value = _json_value(payload, integration)
+    if not isinstance(value, list) or not all(isinstance(item, dict) for item in value):
+        raise HTTPException(status_code=502, detail=f"{integration} returned an invalid response")
+    return value
+
+
+def _require_deluge_result(payload: bytes) -> object:
+    response = _json_object(payload, "Deluge")
+    result = response.get("result")
+    if response.get("error") is not None or result is None or result is False:
+        raise HTTPException(status_code=502, detail="Deluge rejected the request")
+    return result
+
+
 async def search_endpoint(endpoint: AcquisitionEndpoint, query: str) -> list[Release]:
     """Search Prowlarr or a Torznab-compatible torrent indexer."""
     credentials = _credentials(endpoint)
@@ -62,7 +91,7 @@ async def search_endpoint(endpoint: AcquisitionEndpoint, query: str) -> list[Rel
         response_status, _, payload = await _request(request_url, headers={"X-Api-Key": credentials.get("api_key", "")})
         if response_status >= 400:
             raise HTTPException(status_code=502, detail="Prowlarr search failed")
-        data = json.loads(payload)
+        data = _json_array(payload, "Prowlarr")
         return [
             Release(
                 title=item.get("title", "Untitled"),
@@ -167,7 +196,7 @@ async def dispatch_arr_command(endpoint: AcquisitionEndpoint, command: str, ids:
     )
     if response_status >= 400:
         raise HTTPException(status_code=502, detail=f"{endpoint.kind.title()} command failed")
-    response = json.loads(response_payload)
+    response = _json_object(response_payload, endpoint.kind.title())
     return str(response.get("id")) if response.get("id") is not None else None
 
 
@@ -225,26 +254,52 @@ async def _submit_transmission(endpoint: AcquisitionEndpoint, download_url: str,
         )
     if response_status >= 400:
         raise HTTPException(status_code=502, detail="Transmission rejected the release")
-    return str(json.loads(payload).get("arguments", {}).get("torrent-added", {}).get("hashString") or "") or None
+
+    response = _json_object(payload, "Transmission")
+    if response.get("result") != "success":
+        raise HTTPException(status_code=502, detail="Transmission rejected the release")
+
+    response_arguments = response.get("arguments")
+    if not isinstance(response_arguments, dict):
+        raise HTTPException(status_code=502, detail="Transmission returned an invalid response")
+
+    torrent = response_arguments.get("torrent-added") or response_arguments.get("torrent-duplicate")
+    if not isinstance(torrent, dict):
+        return None
+
+    reference = torrent.get("hashString")
+    return str(reference) if reference is not None else None
 
 
 async def _submit_deluge(endpoint: AcquisitionEndpoint, download_url: str, save_path: str | None) -> str | None:
     credentials = _credentials(endpoint)
     headers = {"Content-Type": "application/json"}
     login = json.dumps({"method": "auth.login", "params": [credentials.get("password", "")], "id": 1}).encode()
-    response_status, response_headers, _ = await _request(
+    response_status, response_headers, login_payload = await _request(
         _url(endpoint, "json"), method="POST", headers=headers, body=login
     )
     if response_status >= 400:
         raise HTTPException(status_code=502, detail="Deluge authentication failed")
+
+    try:
+        _require_deluge_result(login_payload)
+    except HTTPException as exc:
+        raise HTTPException(status_code=502, detail="Deluge authentication failed") from exc
+
     options = {"download_location": save_path} if save_path else {}
-    body = json.dumps({"method": "core.add_torrent_magnet", "params": [download_url, options], "id": 2}).encode()
+    method = "core.add_torrent_magnet" if download_url.startswith("magnet:") else "core.add_torrent_url"
+    body = json.dumps({"method": method, "params": [download_url, options], "id": 2}).encode()
     headers["Cookie"] = response_headers.get("Set-Cookie", "").split(";", 1)[0]
     response_status, _, payload = await _request(_url(endpoint, "json"), method="POST", headers=headers, body=body)
     if response_status >= 400:
         raise HTTPException(status_code=502, detail="Deluge rejected the release")
-    result = json.loads(payload).get("result")
-    return str(result) if result is not None else None
+
+    try:
+        result = _require_deluge_result(payload)
+    except HTTPException as exc:
+        raise HTTPException(status_code=502, detail="Deluge rejected the release") from exc
+
+    return str(result)
 
 
 async def owned_endpoint(session: AsyncSession, owner_user_id: Any, endpoint_id: Any) -> AcquisitionEndpoint:
diff --git a/tests/api/routes/test_acquisition.py b/tests/api/routes/test_acquisition.py
index 7a0aa54..4802cd9 100644
--- a/tests/api/routes/test_acquisition.py
+++ b/tests/api/routes/test_acquisition.py
@@ -8,6 +8,7 @@
 from papyrus.core.security import decrypt_secret_payload
 from papyrus.main import settings as app_settings
 from papyrus.models.acquisition import AcquisitionEndpoint
+from papyrus.services import acquisition as acquisition_service
 
 
 @pytest.fixture(autouse=True)
@@ -148,3 +149,38 @@ async def test_endpoint_url_rejects_embedded_credentials(client: AsyncClient, au
     )
 
     assert response.status_code == 422
+
+
+async def test_rejected_submission_persists_failed_job(
+    client: AsyncClient,
+    auth_headers: dict[str, str],
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    async def request(*args: object, **kwargs: object) -> tuple[int, dict[str, str], bytes]:
+        return 200, {}, b'{"result":"invalid or corrupt torrent file","arguments":{}}'
+
+    monkeypatch.setattr(acquisition_service, "_request", request)
+    endpoint_response = await client.post(
+        "/v1/acquisition/endpoints",
+        headers=auth_headers,
+        json={
+            "name": "Transmission",
+            "kind": "transmission",
+            "base_url": "http://transmission.local:9091",
+        },
+    )
+    endpoint_id = endpoint_response.json()["endpoint_id"]
+
+    response = await client.post(
+        "/v1/acquisition/submissions",
+        headers=auth_headers,
+        json={
+            "endpoint_id": endpoint_id,
+            "title": "Rejected release",
+            "download_url": "magnet:?xt=urn:btih:test",
+        },
+    )
+
+    assert response.status_code == 201
+    assert response.json()["status"] == "failed"
+    assert response.json()["error"] == "Transmission rejected the release"
diff --git a/tests/services/test_acquisition.py b/tests/services/test_acquisition.py
new file mode 100644
index 0000000..4acdd49
--- /dev/null
+++ b/tests/services/test_acquisition.py
@@ -0,0 +1,97 @@
+"""Tests for acquisition integration adapters."""
+
+import json
+from typing import cast
+from uuid import uuid4
+
+import pytest
+from fastapi import HTTPException
+
+from papyrus.models.acquisition import AcquisitionEndpoint
+from papyrus.services import acquisition
+
+
+def _endpoint(kind: str) -> AcquisitionEndpoint:
+    return AcquisitionEndpoint(
+        owner_user_id=uuid4(),
+        name=f"Test {kind}",
+        kind=kind,
+        base_url="http://integration.test",
+    )
+
+
+async def test_transmission_rejects_rpc_failure(monkeypatch: pytest.MonkeyPatch) -> None:
+    async def request(*args: object, **kwargs: object) -> tuple[int, dict[str, str], bytes]:
+        return 200, {}, b'{"result":"invalid or corrupt torrent file","arguments":{}}'
+
+    monkeypatch.setattr(acquisition, "_request", request)
+
+    with pytest.raises(HTTPException) as exc_info:
+        await acquisition.submit_to_client(
+            _endpoint("transmission"),
+            "magnet:?xt=urn:btih:test",
+            None,
+            None,
+        )
+
+    assert exc_info.value.status_code == 502
+
+
+async def test_deluge_uses_url_method_for_http_torrent(monkeypatch: pytest.MonkeyPatch) -> None:
+    bodies: list[dict[str, object]] = []
+
+    async def request(*args: object, **kwargs: object) -> tuple[int, dict[str, str], bytes]:
+        bodies.append(json.loads(cast(bytes, kwargs["body"])))
+
+        if len(bodies) == 1:
+            return 200, {"Set-Cookie": "_session_id=test"}, b'{"result":true,"error":null,"id":1}'
+
+        return 200, {}, b'{"result":"torrent-id","error":null,"id":2}'
+
+    monkeypatch.setattr(acquisition, "_request", request)
+
+    await acquisition.submit_to_client(
+        _endpoint("deluge"),
+        "https://indexer.test/release.torrent",
+        None,
+        None,
+    )
+
+    assert bodies[1]["method"] == "core.add_torrent_url"
+
+
+async def test_deluge_rejects_json_rpc_error(monkeypatch: pytest.MonkeyPatch) -> None:
+    responses = iter(
+        [
+            (200, {"Set-Cookie": "_session_id=test"}, b'{"result":true,"error":null,"id":1}'),
+            (200, {}, b'{"result":null,"error":{"message":"invalid torrent"},"id":2}'),
+        ]
+    )
+
+    async def request(*args: object, **kwargs: object) -> tuple[int, dict[str, str], bytes]:
+        return next(responses)
+
+    monkeypatch.setattr(acquisition, "_request", request)
+
+    with pytest.raises(HTTPException) as exc_info:
+        await acquisition.submit_to_client(
+            _endpoint("deluge"),
+            "magnet:?xt=urn:btih:test",
+            None,
+            None,
+        )
+
+    assert exc_info.value.status_code == 502
+
+
+async def test_prowlarr_rejects_invalid_json(monkeypatch: pytest.MonkeyPatch) -> None:
+    async def request(*args: object, **kwargs: object) -> tuple[int, dict[str, str], bytes]:
+        return 200, {}, b"not-json"
+
+    monkeypatch.setattr(acquisition, "_request", request)
+
+    with pytest.raises(HTTPException) as exc_info:
+        await acquisition.search_endpoint(_endpoint("prowlarr"), "book")
+
+    assert exc_info.value.status_code == 502
+    assert exc_info.value.detail == "Prowlarr returned invalid JSON"

From b8d8817434475b6fe6a86bc2c41c05c94bcb35f2 Mon Sep 17 00:00:00 2001
From: Eoic <karolis.strazdas@protonmail.com>
Date: Fri, 17 Jul 2026 23:10:39 +0300
Subject: [PATCH 7/8] fix: preserve acquisition history on endpoint deletion

---
 .../versions/d4e5f6a7b8c9_add_acquisition.py  | 14 +++--
 papyrus/api/routes/acquisition.py             |  5 +-
 papyrus/models/acquisition.py                 | 12 ++++-
 papyrus/schemas/acquisition.py                |  2 +-
 papyrus/services/acquisition.py               | 27 ++++++++++
 tests/api/routes/test_acquisition.py          | 51 ++++++++++++++++++-
 6 files changed, 101 insertions(+), 10 deletions(-)

diff --git a/alembic/versions/d4e5f6a7b8c9_add_acquisition.py b/alembic/versions/d4e5f6a7b8c9_add_acquisition.py
index 9a785c2..88090d0 100644
--- a/alembic/versions/d4e5f6a7b8c9_add_acquisition.py
+++ b/alembic/versions/d4e5f6a7b8c9_add_acquisition.py
@@ -47,7 +47,11 @@ def upgrade() -> None:
         sa.Column("enabled", sa.Boolean(), server_default=sa.text("true"), nullable=False),
         sa.Column("last_run_at", sa.DateTime(timezone=True), nullable=True),
         sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False),
-        sa.ForeignKeyConstraint(["download_client_id"], ["acquisition_endpoints.endpoint_id"]),
+        sa.ForeignKeyConstraint(
+            ["download_client_id"],
+            ["acquisition_endpoints.endpoint_id"],
+            ondelete="SET NULL",
+        ),
         sa.ForeignKeyConstraint(["owner_user_id"], ["users.user_id"], ondelete="CASCADE"),
         sa.PrimaryKeyConstraint("rule_id"),
     )
@@ -56,7 +60,7 @@ def upgrade() -> None:
         "acquisition_jobs",
         sa.Column("job_id", sa.Uuid(), nullable=False),
         sa.Column("owner_user_id", sa.Uuid(), nullable=False),
-        sa.Column("endpoint_id", sa.Uuid(), nullable=False),
+        sa.Column("endpoint_id", sa.Uuid(), nullable=True),
         sa.Column("rule_id", sa.Uuid(), nullable=True),
         sa.Column("title", sa.String(500), nullable=False),
         sa.Column("download_url", sa.Text(), nullable=False),
@@ -64,7 +68,11 @@ def upgrade() -> None:
         sa.Column("client_reference", sa.String(255), nullable=True),
         sa.Column("error", sa.Text(), nullable=True),
         sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False),
-        sa.ForeignKeyConstraint(["endpoint_id"], ["acquisition_endpoints.endpoint_id"]),
+        sa.ForeignKeyConstraint(
+            ["endpoint_id"],
+            ["acquisition_endpoints.endpoint_id"],
+            ondelete="SET NULL",
+        ),
         sa.ForeignKeyConstraint(["owner_user_id"], ["users.user_id"], ondelete="CASCADE"),
         sa.ForeignKeyConstraint(["rule_id"], ["acquisition_rules.rule_id"], ondelete="SET NULL"),
         sa.PrimaryKeyConstraint("job_id"),
diff --git a/papyrus/api/routes/acquisition.py b/papyrus/api/routes/acquisition.py
index 0a333c7..281a744 100644
--- a/papyrus/api/routes/acquisition.py
+++ b/papyrus/api/routes/acquisition.py
@@ -28,6 +28,7 @@
     SubmitRequest,
 )
 from papyrus.services.acquisition import (
+    delete_acquisition_endpoint,
     dispatch_arr_command,
     owned_endpoint,
     run_rule,
@@ -158,9 +159,7 @@ async def update_endpoint(
 
 @protected_router.delete("/endpoints/{endpoint_id}", status_code=status.HTTP_204_NO_CONTENT)
 async def delete_endpoint(user_id: CurrentUserId, endpoint_id: UUID, db: DbSession) -> None:
-    endpoint = await owned_endpoint(db, user_id, endpoint_id)
-    await db.delete(endpoint)
-    await db.commit()
+    await delete_acquisition_endpoint(db, user_id, endpoint_id)
 
 
 @protected_router.post("/search", response_model=list[Release])
diff --git a/papyrus/models/acquisition.py b/papyrus/models/acquisition.py
index 66e2e9d..49e397a 100644
--- a/papyrus/models/acquisition.py
+++ b/papyrus/models/acquisition.py
@@ -39,7 +39,11 @@ class AcquisitionRule(Base):
     name: Mapped[str] = mapped_column(String(120), nullable=False)
     query: Mapped[str] = mapped_column(String(500), nullable=False)
     endpoint_ids: Mapped[list[str] | None] = mapped_column(JSONB, nullable=True)
-    download_client_id: Mapped[UUID | None] = mapped_column(Uuid, ForeignKey("acquisition_endpoints.endpoint_id"))
+    download_client_id: Mapped[UUID | None] = mapped_column(
+        Uuid,
+        ForeignKey("acquisition_endpoints.endpoint_id", ondelete="SET NULL"),
+        nullable=True,
+    )
     filters: Mapped[dict[str, object] | None] = mapped_column(JSONB, nullable=True)
     enabled: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True, server_default="true")
     last_run_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
@@ -53,7 +57,11 @@ class AcquisitionJob(Base):
 
     job_id: Mapped[UUID] = mapped_column(Uuid, primary_key=True, default=uuid4)
     owner_user_id: Mapped[UUID] = mapped_column(ForeignKey("users.user_id", ondelete="CASCADE"), index=True)
-    endpoint_id: Mapped[UUID] = mapped_column(Uuid, ForeignKey("acquisition_endpoints.endpoint_id"), nullable=False)
+    endpoint_id: Mapped[UUID | None] = mapped_column(
+        Uuid,
+        ForeignKey("acquisition_endpoints.endpoint_id", ondelete="SET NULL"),
+        nullable=True,
+    )
     rule_id: Mapped[UUID | None] = mapped_column(Uuid, ForeignKey("acquisition_rules.rule_id", ondelete="SET NULL"))
     title: Mapped[str] = mapped_column(String(500), nullable=False)
     download_url: Mapped[str] = mapped_column(Text, nullable=False)
diff --git a/papyrus/schemas/acquisition.py b/papyrus/schemas/acquisition.py
index 850f118..213baeb 100644
--- a/papyrus/schemas/acquisition.py
+++ b/papyrus/schemas/acquisition.py
@@ -135,7 +135,7 @@ class AcquisitionRule(BaseModel):
 class AcquisitionJob(BaseModel):
     model_config = ConfigDict(from_attributes=True)
     job_id: UUID
-    endpoint_id: UUID
+    endpoint_id: UUID | None
     rule_id: UUID | None
     title: str
     download_url: str
diff --git a/papyrus/services/acquisition.py b/papyrus/services/acquisition.py
index af0615f..9a3b740 100644
--- a/papyrus/services/acquisition.py
+++ b/papyrus/services/acquisition.py
@@ -314,6 +314,33 @@ async def owned_endpoint(session: AsyncSession, owner_user_id: Any, endpoint_id:
     return endpoint
 
 
+async def delete_acquisition_endpoint(session: AsyncSession, owner_user_id: Any, endpoint_id: Any) -> None:
+    endpoint = await owned_endpoint(session, owner_user_id, endpoint_id)
+    result = await session.execute(
+        select(AcquisitionRule).where(AcquisitionRule.owner_user_id == owner_user_id).with_for_update()
+    )
+
+    for rule in result.scalars():
+        endpoint_ids = rule.endpoint_ids or []
+        remaining_endpoint_ids = [value for value in endpoint_ids if value != str(endpoint_id)]
+        indexer_deleted = remaining_endpoint_ids != endpoint_ids
+        download_client_deleted = rule.download_client_id == endpoint_id
+
+        if indexer_deleted:
+            rule.endpoint_ids = remaining_endpoint_ids
+
+        if download_client_deleted:
+            rule.download_client_id = None
+
+        if download_client_deleted or (indexer_deleted and not remaining_endpoint_ids):
+            rule.enabled = False
+
+    await session.flush()
+
+    await session.delete(endpoint)
+    await session.commit()
+
+
 async def run_rule(session: AsyncSession, rule: AcquisitionRule) -> list[AcquisitionJob]:
     """Run one rule once; callers may schedule this from their worker/cron service."""
     client = await owned_endpoint(session, rule.owner_user_id, rule.download_client_id)
diff --git a/tests/api/routes/test_acquisition.py b/tests/api/routes/test_acquisition.py
index 4802cd9..cb87a37 100644
--- a/tests/api/routes/test_acquisition.py
+++ b/tests/api/routes/test_acquisition.py
@@ -1,5 +1,7 @@
 """Tests for private BitTorrent acquisition configuration."""
 
+from uuid import UUID
+
 import pytest
 from httpx import AsyncClient
 from sqlalchemy import select
@@ -7,7 +9,7 @@
 
 from papyrus.core.security import decrypt_secret_payload
 from papyrus.main import settings as app_settings
-from papyrus.models.acquisition import AcquisitionEndpoint
+from papyrus.models.acquisition import AcquisitionEndpoint, AcquisitionJob, AcquisitionRule
 from papyrus.services import acquisition as acquisition_service
 
 
@@ -184,3 +186,50 @@ async def request(*args: object, **kwargs: object) -> tuple[int, dict[str, str],
     assert response.status_code == 201
     assert response.json()["status"] == "failed"
     assert response.json()["error"] == "Transmission rejected the release"
+
+
+async def test_delete_endpoint_preserves_jobs_and_disables_rules(
+    client: AsyncClient,
+    auth_headers: dict[str, str],
+    auth_user: dict[str, str],
+    db_session: AsyncSession,
+) -> None:
+    endpoint_response = await client.post(
+        "/v1/acquisition/endpoints",
+        headers=auth_headers,
+        json={
+            "name": "Disposable client",
+            "kind": "qbittorrent",
+            "base_url": "http://qbittorrent.local:8080",
+        },
+    )
+    endpoint_id = UUID(endpoint_response.json()["endpoint_id"])
+    owner_user_id = UUID(auth_user["user_id"])
+    rule = AcquisitionRule(
+        owner_user_id=owner_user_id,
+        name="Affected rule",
+        query="book",
+        endpoint_ids=[str(endpoint_id)],
+        download_client_id=endpoint_id,
+        enabled=True,
+    )
+    job = AcquisitionJob(
+        owner_user_id=owner_user_id,
+        endpoint_id=endpoint_id,
+        title="Audited release",
+        download_url="magnet:?xt=urn:btih:test",
+        status="submitted",
+    )
+
+    db_session.add_all([rule, job])
+    await db_session.commit()
+
+    response = await client.delete(f"/v1/acquisition/endpoints/{endpoint_id}", headers=auth_headers)
+
+    assert response.status_code == 204
+    await db_session.refresh(job)
+    await db_session.refresh(rule)
+    assert job.endpoint_id is None
+    assert rule.download_client_id is None
+    assert rule.endpoint_ids == []
+    assert rule.enabled is False

From 3cfb9dd164c40f6b340c30291e1be362e2067c2a Mon Sep 17 00:00:00 2001
From: Eoic <karolis.strazdas@protonmail.com>
Date: Fri, 17 Jul 2026 23:14:24 +0300
Subject: [PATCH 8/8] feat: test acquisition connections without saving

---
 papyrus/api/routes/acquisition.py    |  17 +++
 papyrus/schemas/acquisition.py       |  29 +++++-
 papyrus/services/acquisition.py      | 150 ++++++++++++++++++++++++++-
 tests/api/routes/test_acquisition.py | 105 ++++++++++++++++++-
 tests/services/test_acquisition.py   |  32 ++++++
 5 files changed, 326 insertions(+), 7 deletions(-)

diff --git a/papyrus/api/routes/acquisition.py b/papyrus/api/routes/acquisition.py
index 281a744..fb61434 100644
--- a/papyrus/api/routes/acquisition.py
+++ b/papyrus/api/routes/acquisition.py
@@ -18,6 +18,8 @@
     AcquisitionCapabilities,
     AcquisitionEndpoint,
     AcquisitionEndpointCreate,
+    AcquisitionEndpointTest,
+    AcquisitionEndpointTestResult,
     AcquisitionEndpointUpdate,
     AcquisitionJob,
     AcquisitionRule,
@@ -28,12 +30,14 @@
     SubmitRequest,
 )
 from papyrus.services.acquisition import (
+    build_test_endpoint,
     delete_acquisition_endpoint,
     dispatch_arr_command,
     owned_endpoint,
     run_rule,
     search_endpoint,
     submit_to_client,
+    test_endpoint_connection,
 )
 
 router = APIRouter()
@@ -162,6 +166,19 @@ async def delete_endpoint(user_id: CurrentUserId, endpoint_id: UUID, db: DbSessi
     await delete_acquisition_endpoint(db, user_id, endpoint_id)
 
 
+@protected_router.post("/endpoints/test", response_model=AcquisitionEndpointTestResult)
+async def test_connection(
+    user_id: CurrentUserId,
+    request: AcquisitionEndpointTest,
+    db: DbSession,
+) -> AcquisitionEndpointTestResult:
+    endpoint = await build_test_endpoint(db, user_id, request)
+
+    await test_endpoint_connection(endpoint)
+
+    return AcquisitionEndpointTestResult(ok=True)
+
+
 @protected_router.post("/search", response_model=list[Release])
 async def search_releases(user_id: CurrentUserId, request: SearchRequest, db: DbSession) -> list[Release]:
     statement = select(AcquisitionEndpointModel).where(
diff --git a/papyrus/schemas/acquisition.py b/papyrus/schemas/acquisition.py
index 213baeb..3880b9e 100644
--- a/papyrus/schemas/acquisition.py
+++ b/papyrus/schemas/acquisition.py
@@ -2,9 +2,10 @@
 
 from datetime import datetime
 from enum import StrEnum
+from typing import Self
 from uuid import UUID
 
-from pydantic import BaseModel, ConfigDict, Field, HttpUrl, SecretStr, field_validator
+from pydantic import BaseModel, ConfigDict, Field, HttpUrl, SecretStr, field_validator, model_validator
 
 
 class EndpointKind(StrEnum):
@@ -61,6 +62,32 @@ def validate_endpoint_url(cls, value: HttpUrl | None) -> HttpUrl | None:
         return validate_endpoint_url(value)
 
 
+class AcquisitionEndpointTest(BaseModel):
+    endpoint_id: UUID | None = None
+    kind: EndpointKind | None = None
+    base_url: HttpUrl | None = None
+    api_key: SecretStr | None = None
+    username: SecretStr | None = None
+    password: SecretStr | None = None
+
+    @field_validator("base_url")
+    @classmethod
+    def validate_endpoint_url(cls, value: HttpUrl | None) -> HttpUrl | None:
+        if value is None:
+            return None
+        return validate_endpoint_url(value)
+
+    @model_validator(mode="after")
+    def validate_endpoint_target(self) -> Self:
+        if self.endpoint_id is None and (self.kind is None or self.base_url is None):
+            raise ValueError("kind and base_url are required for an unsaved endpoint")
+        return self
+
+
+class AcquisitionEndpointTestResult(BaseModel):
+    ok: bool
+
+
 class AcquisitionEndpoint(BaseModel):
     model_config = ConfigDict(from_attributes=True)
     endpoint_id: UUID
diff --git a/papyrus/services/acquisition.py b/papyrus/services/acquisition.py
index 9a3b740..d6c950f 100644
--- a/papyrus/services/acquisition.py
+++ b/papyrus/services/acquisition.py
@@ -16,9 +16,9 @@
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from papyrus.core.security import decrypt_secret_payload
+from papyrus.core.security import decrypt_secret_payload, encrypt_secret_payload
 from papyrus.models.acquisition import AcquisitionEndpoint, AcquisitionJob, AcquisitionRule
-from papyrus.schemas.acquisition import Release
+from papyrus.schemas.acquisition import AcquisitionEndpointTest, Release
 
 
 def _url(endpoint: AcquisitionEndpoint, path: str) -> str:
@@ -189,7 +189,7 @@ async def dispatch_arr_command(endpoint: AcquisitionEndpoint, command: str, ids:
         payload[id_field] = ids[0] if id_field in {"seriesId"} else ids
 
     response_status, _, response_payload = await _request(
-        _url(endpoint, "api/v1/command"),
+        _url(endpoint, "api/v3/command"),
         method="POST",
         headers={"Content-Type": "application/json", "X-Api-Key": _credentials(endpoint).get("api_key", "")},
         body=json.dumps(payload).encode(),
@@ -207,13 +207,13 @@ async def _submit_qbittorrent(
     login = urlencode(
         {"username": credentials.get("username", ""), "password": credentials.get("password", "")}
     ).encode()
-    response_status, headers, _ = await _request(
+    response_status, headers, response_payload = await _request(
         _url(endpoint, "api/v2/auth/login"),
         method="POST",
         headers={"Content-Type": "application/x-www-form-urlencoded"},
         body=login,
     )
-    if response_status >= 400:
+    if response_status >= 400 or response_payload.strip() != b"Ok.":
         raise HTTPException(status_code=502, detail="qBittorrent authentication failed")
     payload = {"urls": download_url}
     if category:
@@ -341,6 +341,146 @@ async def delete_acquisition_endpoint(session: AsyncSession, owner_user_id: Any,
     await session.commit()
 
 
+async def build_test_endpoint(
+    session: AsyncSession,
+    owner_user_id: Any,
+    request: AcquisitionEndpointTest,
+) -> AcquisitionEndpoint:
+    stored_endpoint = None
+    stored_credentials: dict[str, str] = {}
+
+    if request.endpoint_id is not None:
+        stored_endpoint = await owned_endpoint(session, owner_user_id, request.endpoint_id)
+        stored_credentials = _credentials(stored_endpoint)
+
+    credentials = dict(stored_credentials)
+    for field in ("api_key", "username", "password"):
+        value = getattr(request, field)
+        if value is not None:
+            credentials[field] = value.get_secret_value()
+
+    if request.kind is not None:
+        kind = request.kind.value
+    elif stored_endpoint is not None:
+        kind = stored_endpoint.kind
+    else:
+        raise HTTPException(status_code=422, detail="Endpoint kind is required")
+
+    if request.base_url is not None:
+        base_url = str(request.base_url)
+    elif stored_endpoint is not None:
+        base_url = stored_endpoint.base_url
+    else:
+        raise HTTPException(status_code=422, detail="Endpoint URL is required")
+
+    encrypted_credentials = {"encrypted": encrypt_secret_payload(credentials)} if credentials else None
+
+    return AcquisitionEndpoint(
+        owner_user_id=owner_user_id,
+        name=stored_endpoint.name if stored_endpoint is not None else "Connection test",
+        kind=kind,
+        base_url=base_url,
+        credentials=encrypted_credentials,
+        settings=stored_endpoint.settings if stored_endpoint is not None else None,
+    )
+
+
+async def test_endpoint_connection(endpoint: AcquisitionEndpoint) -> None:
+    credentials = _credentials(endpoint)
+
+    if endpoint.kind == "prowlarr":
+        response_status, _, payload = await _request(
+            _url(endpoint, "api/v1/system/status"),
+            headers={"X-Api-Key": credentials.get("api_key", "")},
+        )
+        if response_status >= 400:
+            raise HTTPException(status_code=502, detail="Prowlarr connection test failed")
+
+        _json_object(payload, "Prowlarr")
+        return
+
+    if endpoint.kind == "torznab":
+        params = urlencode({"t": "caps", "apikey": credentials.get("api_key", "")})
+        response_status, _, payload = await _request(_url(endpoint, f"api?{params}"))
+        if response_status >= 400:
+            raise HTTPException(status_code=502, detail="Torznab connection test failed")
+
+        try:
+            ElementTree.fromstring(payload)
+        except ElementTree.ParseError as exc:
+            raise HTTPException(status_code=502, detail="Torznab returned invalid XML") from exc
+        return
+
+    if endpoint.kind in {"readarr", "sonarr", "radarr", "lidarr", "whisparr"}:
+        response_status, _, payload = await _request(
+            _url(endpoint, "api/v3/system/status"),
+            headers={"X-Api-Key": credentials.get("api_key", "")},
+        )
+        if response_status >= 400:
+            raise HTTPException(status_code=502, detail=f"{endpoint.kind.title()} connection test failed")
+
+        _json_object(payload, endpoint.kind.title())
+        return
+
+    if endpoint.kind == "qbittorrent":
+        login = urlencode(
+            {"username": credentials.get("username", ""), "password": credentials.get("password", "")}
+        ).encode()
+        response_status, _, payload = await _request(
+            _url(endpoint, "api/v2/auth/login"),
+            method="POST",
+            headers={"Content-Type": "application/x-www-form-urlencoded"},
+            body=login,
+        )
+        if response_status >= 400 or payload.strip() != b"Ok.":
+            raise HTTPException(status_code=502, detail="qBittorrent authentication failed")
+        return
+
+    if endpoint.kind == "transmission":
+        body = json.dumps({"method": "session-get", "arguments": {}}).encode()
+        headers = {"Content-Type": "application/json"}
+        if credentials.get("username"):
+            token = base64.b64encode(f"{credentials['username']}:{credentials.get('password', '')}".encode()).decode()
+            headers["Authorization"] = f"Basic {token}"
+
+        response_status, response_headers, payload = await _request(
+            _url(endpoint, "transmission/rpc"),
+            method="POST",
+            headers=headers,
+            body=body,
+        )
+        if response_status == 409:
+            headers["X-Transmission-Session-Id"] = response_headers.get("X-Transmission-Session-Id", "")
+            response_status, _, payload = await _request(
+                _url(endpoint, "transmission/rpc"),
+                method="POST",
+                headers=headers,
+                body=body,
+            )
+        if response_status >= 400 or _json_object(payload, "Transmission").get("result") != "success":
+            raise HTTPException(status_code=502, detail="Transmission connection test failed")
+        return
+
+    if endpoint.kind == "deluge":
+        login = json.dumps({"method": "auth.login", "params": [credentials.get("password", "")], "id": 1}).encode()
+        response_status, _, payload = await _request(
+            _url(endpoint, "json"),
+            method="POST",
+            headers={"Content-Type": "application/json"},
+            body=login,
+        )
+        if response_status >= 400:
+            raise HTTPException(status_code=502, detail="Deluge authentication failed")
+
+        try:
+            _require_deluge_result(payload)
+        except HTTPException as exc:
+            raise HTTPException(status_code=502, detail="Deluge authentication failed") from exc
+        return
+
+    raise HTTPException(status_code=422, detail="Endpoint kind is not supported")
+
+
 async def run_rule(session: AsyncSession, rule: AcquisitionRule) -> list[AcquisitionJob]:
     """Run one rule once; callers may schedule this from their worker/cron service."""
     client = await owned_endpoint(session, rule.owner_user_id, rule.download_client_id)
diff --git a/tests/api/routes/test_acquisition.py b/tests/api/routes/test_acquisition.py
index cb87a37..a41a4f2 100644
--- a/tests/api/routes/test_acquisition.py
+++ b/tests/api/routes/test_acquisition.py
@@ -4,12 +4,14 @@
 
 import pytest
 from httpx import AsyncClient
-from sqlalchemy import select
+from sqlalchemy import func, select
 from sqlalchemy.ext.asyncio import AsyncSession
 
+from papyrus.api.routes import acquisition as acquisition_routes
 from papyrus.core.security import decrypt_secret_payload
 from papyrus.main import settings as app_settings
 from papyrus.models.acquisition import AcquisitionEndpoint, AcquisitionJob, AcquisitionRule
+from papyrus.models.user import User
 from papyrus.services import acquisition as acquisition_service
 
 
@@ -233,3 +235,104 @@ async def test_delete_endpoint_preserves_jobs_and_disables_rules(
     assert rule.download_client_id is None
     assert rule.endpoint_ids == []
     assert rule.enabled is False
+
+
+async def test_connection_checks_unsaved_endpoint_without_persisting(
+    client: AsyncClient,
+    auth_headers: dict[str, str],
+    db_session: AsyncSession,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    captured: list[AcquisitionEndpoint] = []
+
+    async def test_connection(endpoint: AcquisitionEndpoint) -> None:
+        captured.append(endpoint)
+
+    monkeypatch.setattr(acquisition_routes, "test_endpoint_connection", test_connection, raising=False)
+    before_count = await db_session.scalar(select(func.count()).select_from(AcquisitionEndpoint))
+
+    response = await client.post(
+        "/v1/acquisition/endpoints/test",
+        headers=auth_headers,
+        json={
+            "kind": "prowlarr",
+            "base_url": "http://prowlarr.local:9696",
+            "api_key": "unsaved-key",
+        },
+    )
+
+    assert response.status_code == 200
+    assert response.json() == {"ok": True}
+    assert captured[0].kind == "prowlarr"
+    assert decrypt_secret_payload(captured[0].credentials["encrypted"])["api_key"] == "unsaved-key"
+    after_count = await db_session.scalar(select(func.count()).select_from(AcquisitionEndpoint))
+    assert after_count == before_count
+
+
+async def test_connection_merges_owned_endpoint_overrides_without_persisting(
+    client: AsyncClient,
+    auth_headers: dict[str, str],
+    db_session: AsyncSession,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    endpoint_response = await client.post(
+        "/v1/acquisition/endpoints",
+        headers=auth_headers,
+        json={
+            "name": "Saved Prowlarr",
+            "kind": "prowlarr",
+            "base_url": "http://prowlarr.local:9696",
+            "api_key": "saved-key",
+        },
+    )
+    captured: list[AcquisitionEndpoint] = []
+
+    async def test_connection(endpoint: AcquisitionEndpoint) -> None:
+        captured.append(endpoint)
+
+    monkeypatch.setattr(acquisition_routes, "test_endpoint_connection", test_connection, raising=False)
+    before_count = await db_session.scalar(select(func.count()).select_from(AcquisitionEndpoint))
+
+    response = await client.post(
+        "/v1/acquisition/endpoints/test",
+        headers=auth_headers,
+        json={
+            "endpoint_id": endpoint_response.json()["endpoint_id"],
+            "base_url": "http://prowlarr-edited.local:9696",
+            "api_key": "override",
+        },
+    )
+
+    assert response.status_code == 200
+    assert response.json() == {"ok": True}
+    assert captured[0].base_url == "http://prowlarr-edited.local:9696/"
+    assert decrypt_secret_payload(captured[0].credentials["encrypted"])["api_key"] == "override"
+    after_count = await db_session.scalar(select(func.count()).select_from(AcquisitionEndpoint))
+    assert after_count == before_count
+
+
+async def test_connection_rejects_another_users_endpoint(
+    client: AsyncClient,
+    auth_headers: dict[str, str],
+    db_session: AsyncSession,
+) -> None:
+    other_user = User(display_name="Other User")
+    db_session.add(other_user)
+    await db_session.flush()
+
+    endpoint = AcquisitionEndpoint(
+        owner_user_id=other_user.user_id,
+        name="Other Prowlarr",
+        kind="prowlarr",
+        base_url="http://other-prowlarr.local:9696",
+    )
+    db_session.add(endpoint)
+    await db_session.commit()
+
+    response = await client.post(
+        "/v1/acquisition/endpoints/test",
+        headers=auth_headers,
+        json={"endpoint_id": str(endpoint.endpoint_id)},
+    )
+
+    assert response.status_code == 404
diff --git a/tests/services/test_acquisition.py b/tests/services/test_acquisition.py
index 4acdd49..0029261 100644
--- a/tests/services/test_acquisition.py
+++ b/tests/services/test_acquisition.py
@@ -95,3 +95,35 @@ async def request(*args: object, **kwargs: object) -> tuple[int, dict[str, str],
 
     assert exc_info.value.status_code == 502
     assert exc_info.value.detail == "Prowlarr returned invalid JSON"
+
+
+async def test_qbittorrent_rejects_failed_login_body(monkeypatch: pytest.MonkeyPatch) -> None:
+    async def request(*args: object, **kwargs: object) -> tuple[int, dict[str, str], bytes]:
+        return 200, {}, b"Fails."
+
+    monkeypatch.setattr(acquisition, "_request", request)
+
+    with pytest.raises(HTTPException) as exc_info:
+        await acquisition.submit_to_client(
+            _endpoint("qbittorrent"),
+            "magnet:?xt=urn:btih:test",
+            None,
+            None,
+        )
+
+    assert exc_info.value.status_code == 502
+    assert exc_info.value.detail == "qBittorrent authentication failed"
+
+
+async def test_arr_commands_use_v3_api(monkeypatch: pytest.MonkeyPatch) -> None:
+    urls: list[str] = []
+
+    async def request(url: str, **kwargs: object) -> tuple[int, dict[str, str], bytes]:
+        urls.append(url)
+        return 201, {}, b'{"id":1}'
+
+    monkeypatch.setattr(acquisition, "_request", request)
+
+    await acquisition.dispatch_arr_command(_endpoint("readarr"), "BookSearch", [1])
+
+    assert urls == ["http://integration.test/api/v3/command"]