diff --git a/README.md b/README.md
index b24b097e4c..ce737d7c56 100644
--- a/README.md
+++ b/README.md
@@ -786,6 +786,7 @@ In plan mode, malformed TODO statuses return normal tool-validation errors witho
 Tool-produced paths and URLs can be retained as short artifact handles across context compaction (`tool_artifacts` in `config.yaml`). Handles distinguish separate tool-result occurrences, even when a provider reuses call IDs. Detected file URLs preserve their query strings and fragments. When PII redaction is enabled, model-visible artifact labels follow that policy; internal references stay intact for tool argument resolution. The configured registry limit retains the newest artifacts, while checkpointed processing identities prevent evicted results from being recaptured after restart. Resolution runs before authorization and write-safety checks; unknown or expired handles return an error without executing the tool. Small unknown structured results may be retained as complete JSON up to 4096 UTF-8 bytes; empty or oversized payloads are skipped. Handles are agent-local: task arguments resolve parent handles to concrete references, and delegated reports must return concrete references rather than child-local handles. A truncated model projection reports how many handles are omitted.
 
 DeerFlow supports configurable MCP servers and skills to extend its capabilities.
+When durable MCP background tasks are enabled, agents can use `list_background_tasks(status="failed")` or `status="input_required"` to find failed tasks or tasks awaiting input in the current chat. Other supported statuses are `submitted`, `working`, `completed`, and `cancelled`; omitting the status preserves existing behavior. The database applies the filter before limiting results to the 20 most recent matching tasks. When combined with `active_only=true`, both filters apply: active statuses are `submitted`, `working`, and `input_required`, so terminal statuses return an empty list. `GET /api/threads/{thread_id}/mcp-tasks` supports the same `status` and `active_only` filters before its `limit` (default 50, range 1–100); for example, `?status=failed&limit=20`. Unknown statuses return 422, and omitting the filters preserves the existing response.
 For HTTP/SSE MCP servers, OAuth token flows are supported (`client_credentials`, `refresh_token`).
 Missing, malformed, or out-of-range token response `expires_in` values use a one-hour default lifetime. This includes lifetimes that cannot be added to the current time without overflowing the expiry timestamp.
 Durable HTTP/SSE task status and cancellation calls select configured `user_auth` credentials using the persisted task owner, including after restart; per-request secrets are not retained for background calls. If a request-scoped credential overrides submit authentication, both credentials must authorize access to the same remote task.
diff --git a/backend/app/gateway/routers/AGENTS.md b/backend/app/gateway/routers/AGENTS.md
index bff540aa53..2e3f9dbaf0 100644
--- a/backend/app/gateway/routers/AGENTS.md
+++ b/backend/app/gateway/routers/AGENTS.md
@@ -1,5 +1,9 @@
 # Router contracts
 
+The MCP task list accepts optional `TaskStatus` and `active_only` query filters.
+Pass both to the repository before its limit; preserve `threads:read`, owner
+checks, current-thread incarnation scoping, and the unfiltered array response.
+
 Custom Agent packages are strict, versioned definition-only JSON documents.
 Export explicitly projects the caller's portable `AgentConfig` fields and SOUL;
 exclude `github` installation bindings, memory contents, conversations and
diff --git a/backend/app/gateway/routers/mcp_tasks.py b/backend/app/gateway/routers/mcp_tasks.py
index 5e88e3068a..9297d0d9b9 100644
--- a/backend/app/gateway/routers/mcp_tasks.py
+++ b/backend/app/gateway/routers/mcp_tasks.py
@@ -8,6 +8,7 @@ from fastapi import APIRouter, HTTPException, Query, Request
 
 from app.gateway.authz import require_permission
 from app.gateway.deps import get_current_user, get_mcp_task_repo, get_mcp_task_service, get_thread_store
+from deerflow.mcp.tasks import TaskStatus
 from deerflow.mcp_scope import is_valid_thread_incarnation
 from deerflow.utils.thread_id import ThreadId
 
@@ -95,6 +96,8 @@ async def list_mcp_tasks(
     thread_id: ThreadId,
     request: Request,
     limit: int = Query(default=50, ge=1, le=100),
+    status: TaskStatus | None = None,
+    active_only: bool = False,
 ) -> list[dict[str, Any]]:
     """List the durable MCP background tasks recorded for a thread.
 
@@ -102,6 +105,8 @@ async def list_mcp_tasks(
         thread_id: Thread whose tasks are listed.
         request: Incoming request, used for auth and repository access.
         limit: Maximum number of tasks to return (1-100).
+        status: Optional task status filter applied in SQL before the limit.
+        active_only: Restrict to submitted, working, and input_required tasks; intersects with status.
 
     Returns:
         Tasks in repository order, each projected onto the list view.
@@ -119,6 +124,8 @@ async def list_mcp_tasks(
         user_id=user_id,
         thread_incarnation=thread_incarnation,
         limit=limit,
+        status=status,
+        active_only=active_only,
     )
     threshold = service.tracking_degraded_after_errors
     return [_list_item(record, threshold=threshold) for record in records]
diff --git a/backend/app/mcp_tasks/service.py b/backend/app/mcp_tasks/service.py
index a44b384d52..728f512f6c 100644
--- a/backend/app/mcp_tasks/service.py
+++ b/backend/app/mcp_tasks/service.py
@@ -702,6 +702,7 @@ class McpTaskService:
         thread_incarnation: str | None,
         limit: int = 50,
         active_only: bool = False,
+        status: TaskStatus | None = None,
     ) -> list[dict[str, Any]]:
         return await self._repository.list_by_thread(
             thread_id,
@@ -709,6 +710,7 @@ class McpTaskService:
             thread_incarnation=thread_incarnation,
             limit=limit,
             active_only=active_only,
+            **({"status": TaskStatus(status)} if status is not None else {}),
         )
 
     async def cancel_task(
diff --git a/backend/packages/harness/deerflow/mcp/AGENTS.md b/backend/packages/harness/deerflow/mcp/AGENTS.md
index 61d9bc6bcf..f3051f1921 100644
--- a/backend/packages/harness/deerflow/mcp/AGENTS.md
+++ b/backend/packages/harness/deerflow/mcp/AGENTS.md
@@ -1,5 +1,7 @@
 ### MCP System (`packages/harness/deerflow/mcp/`)
 
+The optional background-task `status` filter flows through `McpTaskSubmitter.list_tasks` → `McpTaskService.list_tasks` → `McpTaskRepository.list_by_thread`. SQL applies status, active-state, user, thread, and current-incarnation predicates before ordering and LIMIT; never filter an already truncated list. Omitted statuses do not add a keyword argument to legacy implementations. Filtering uses existing states without changing polling, notifications, retries, or cancellation. Regression coverage: `tests/test_background_task_status_filter.py`.
+
 - Uses `langchain-mcp-adapters` `MultiServerMCPClient` for multi-server management
 - **OAuth expiry**: Missing, non-convertible or overflowing `expires_in` values use the one-hour default. Guard both `timedelta` construction and addition to one captured UTC `now`; avoid float-derived range ceilings. `tests/test_mcp_oauth.py` pins both overflow stages and the exact `datetime` boundary with a fixed clock.
 - **Long-running task foundation**: `mcp/tasks/` defines the protocol-neutral `McpTaskDriver` contract and normalized `TaskSnapshot` states (`submitted`, `working`, `input_required`, `completed`, `failed`, `cancelled`). A driver-supplied `poll_after_seconds` must be a finite positive number, validated at the `TaskSnapshot` boundary so every driver is held to the same invariant rather than each one guarding the consumer that turns the interval into a `timedelta`. `persistence/mcp_tasks/` owns the durable remote-handle mapping, poll schedule, notification state, lease owner, and separate consecutive poll/delivery error counters; `app/mcp_tasks/McpTaskService` performs status, cancellation, and notification work outside the Agent/LLM loop. Notification retries keep their idempotency attempt separate from the delivery-failure count, use capped exponential backoff, and stop after five failures; strict existing-thread admission dead-letters a deleted/mismatched target immediately. A status result is applied only when the worker still owns an unexpired lease, so a stale result cannot be written after expiry even before another worker reclaims the row. Every claim carries a fresh per-claim token, and every poll/cancel/notification release or snapshot apply is fenced by the owner plus that token, so a release that completes after the same service reclaims the row cannot clear the newer lease. Poll, cancel, and notification claims also have separate phase-level single-flight owners that protect only an ambiguous claim outcome: after an uncancelled timeout or caller cancellation, later scans do not start another claim while the outcome is unknown. Once the claim resolves, the phase owner is released immediately, and any returned rows are released through bounded, service-owned background work; per-claim token fencing prevents a late release from mutating a newer claim generation, so a stuck release does not stall the whole phase. While the claim outcome remains unresolved, later scans skip the phase and emit a warning rather than overlapping an ambiguous database claim. Once the claim resolves, the handoff may continue releasing returned rows in the background without blocking later scans. Lease expiry remains the cross-process crash fallback rather than the normal recovery for a live service's late claim. Poll timestamps and retry schedules are based on the remote call's completion time rather than the scan start. If submission succeeds but persistence fails or the caller is cancelled while persistence is in flight, the service best-effort cancels the remote task and preserves the original error or cancellation if that compensation also fails. The exact `uq_mcp_tasks_user_server_remote` conflict is different: an existing durable row already owns the remote handle, so the conflict surfaces without cancelling that tracked task. Unexpected per-task poll failures are isolated from sibling claims and remain recoverable through lease expiry; Gateway shutdown cancels the poller so a hung external status call cannot block process exit. Routine cancellation releases clear only the lease/token and preserve any pre-existing `last_cancel_error` or `notification_error`; cancellation itself is not persisted as a task failure diagnostic. Cancelling an in-flight poll releases only the owner- and per-claim-token-fenced lease and preserves its preclaim schedule and poll-failure state; real poll failures retain exponential backoff and tracking-degradation behavior. `input_required` and terminal states stop polling and become `notification_status=pending` for later Agent/UI delivery. Durable recovery requires a SQL database backend (`sqlite` or `postgres`); the in-memory backend leaves the repository/service unavailable. The runtime is startup-configured by `mcp_tasks` and disabled by default until a concrete driver is registered; this foundation does not alter ordinary MCP tool behavior on its own.
diff --git a/backend/packages/harness/deerflow/mcp/tasks/runtime.py b/backend/packages/harness/deerflow/mcp/tasks/runtime.py
index fc25e5e5a7..0900a34105 100644
--- a/backend/packages/harness/deerflow/mcp/tasks/runtime.py
+++ b/backend/packages/harness/deerflow/mcp/tasks/runtime.py
@@ -7,7 +7,7 @@ from typing import Any, Protocol
 from deerflow.config.extensions_config import ExtensionsConfig, McpServerConfig
 from deerflow.mcp.config_normalization import normalize_mcp_interceptor_paths, normalize_mcp_server_config
 from deerflow.mcp.oauth import OAuthTokenManager
-from deerflow.mcp.tasks.models import TaskSubmitRequest
+from deerflow.mcp.tasks.models import TaskStatus, TaskSubmitRequest
 
 
 class McpTaskConfigurationError(RuntimeError):
@@ -31,6 +31,7 @@ class McpTaskSubmitter(Protocol):
         thread_incarnation: str | None,
         limit: int = 50,
         active_only: bool = False,
+        status: TaskStatus | None = None,
     ) -> list[dict[str, Any]]: ...
 
     async def cancel_matching_task(
diff --git a/backend/packages/harness/deerflow/persistence/mcp_tasks/sql.py b/backend/packages/harness/deerflow/persistence/mcp_tasks/sql.py
index 32ace7f401..94a4c1b735 100644
--- a/backend/packages/harness/deerflow/persistence/mcp_tasks/sql.py
+++ b/backend/packages/harness/deerflow/persistence/mcp_tasks/sql.py
@@ -11,6 +11,7 @@ from sqlalchemy.exc import IntegrityError
 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
 
 from deerflow.mcp.tasks import ATTENTION_TASK_STATUSES, POLLABLE_TASK_STATUSES, TERMINAL_TASK_STATUSES
+from deerflow.mcp.tasks.models import TaskStatus
 from deerflow.persistence.mcp_tasks.model import McpTaskRow
 from deerflow.persistence.thread_meta.model import ThreadMetaRow
 from deerflow.utils.time import coerce_iso
@@ -275,6 +276,7 @@ class McpTaskRepository:
         thread_incarnation: str | None,
         limit: int = 50,
         active_only: bool = False,
+        status: TaskStatus | None = None,
     ) -> list[dict[str, Any]]:
         stmt = select(McpTaskRow).where(
             McpTaskRow.thread_id == thread_id,
@@ -288,6 +290,8 @@ class McpTaskRepository:
         )
         if active_only:
             stmt = stmt.where(McpTaskRow.status.in_(_POLLABLE_STATUS_VALUES))
+        if status is not None:
+            stmt = stmt.where(McpTaskRow.status == TaskStatus(status).value)
         stmt = stmt.order_by(McpTaskRow.created_at.desc(), McpTaskRow.id.desc()).limit(limit)
         async with self._sf() as session:
             result = await session.execute(stmt)
diff --git a/backend/packages/harness/deerflow/tools/AGENTS.md b/backend/packages/harness/deerflow/tools/AGENTS.md
index 20f96b8d59..faf19ab96b 100644
--- a/backend/packages/harness/deerflow/tools/AGENTS.md
+++ b/backend/packages/harness/deerflow/tools/AGENTS.md
@@ -60,6 +60,8 @@ Scheduled-task runtime note:
 
 Durable MCP task-management tools are added only while the process-local task submitter is installed. They expose bounded local task fields, including whether cancellation was requested, but never the remote handle. Cancellation records that request durably and returns immediately; the background service owns the remote call and retries. These remain ordinary business tools under an active skill's `allowed-tools` policy and must be declared explicitly.
 
+`list_background_tasks` accepts an optional `TaskStatus` filter and intersects it with `active_only`. Omitting it or passing `None` preserves existing calls. The tool returns the 20 most recent matching records; `count` is the number returned. The tool schema validates statuses, and internal callers must not silently ignore invalid values. Scope comes from the runtime, never model-supplied user, thread, or incarnation values.
+
 **Community tools** (`packages/harness/deerflow/community/`): optional integrations, each in its own subpackage and wired through `config.yaml`. Documented examples:
 - `tavily/` - Web search (5 results default) and web fetch (4KB limit)
 - `sofya/` - Web search (5 results default, per-result content capped at 2000 chars) and web fetch (4KB limit)
diff --git a/backend/packages/harness/deerflow/tools/builtins/background_tasks_tool.py b/backend/packages/harness/deerflow/tools/builtins/background_tasks_tool.py
index 3e5c6e0e7e..139b57e201 100644
--- a/backend/packages/harness/deerflow/tools/builtins/background_tasks_tool.py
+++ b/backend/packages/harness/deerflow/tools/builtins/background_tasks_tool.py
@@ -7,6 +7,7 @@ from typing import Annotated, Any
 from langchain.tools import tool
 
 from deerflow.agents.middlewares.input_sanitization_middleware import neutralize_untrusted_tags
+from deerflow.mcp.tasks.models import TaskStatus
 from deerflow.mcp.tasks.runtime import get_mcp_task_submitter
 from deerflow.mcp_scope import runtime_thread_incarnation
 from deerflow.tools.builtins.list_uploaded_files_tool import _resolve_thread_id, _resolve_user_id
@@ -30,6 +31,7 @@ async def _list_background_tasks_impl(
     *,
     active_only: bool = False,
     limit: int = 20,
+    status: TaskStatus | None = None,
 ) -> dict[str, Any]:
     thread_id = _resolve_thread_id(runtime)
     if thread_id is None:
@@ -40,6 +42,7 @@ async def _list_background_tasks_impl(
         thread_incarnation=runtime_thread_incarnation(runtime),
         limit=max(1, min(limit, 50)),
         active_only=active_only,
+        **({"status": TaskStatus(status)} if status is not None else {}),
     )
     tasks = [_public_task(record) for record in records]
     return {"tasks": tasks, "count": len(tasks)}
@@ -49,9 +52,10 @@ async def _list_background_tasks_impl(
 async def list_background_tasks(
     runtime: Runtime,
     active_only: Annotated[bool, "Return only tasks that are still active."] = False,
+    status: Annotated[TaskStatus | None, "Optional task status. Intersects with active_only and filters before the limit of 20 recent matching tasks."] = None,
 ) -> dict[str, Any]:
     """List current and recent durable background tasks for this chat."""
-    return await _list_background_tasks_impl(runtime, active_only=active_only)
+    return await _list_background_tasks_impl(runtime, active_only=active_only, status=status)
 
 
 @tool
diff --git a/backend/tests/test_background_task_status_filter.py b/backend/tests/test_background_task_status_filter.py
new file mode 100644
index 0000000000..c3f7ecedaf
--- /dev/null
+++ b/backend/tests/test_background_task_status_filter.py
@@ -0,0 +1,158 @@
+from datetime import UTC, datetime, timedelta
+from types import SimpleNamespace
+
+import pytest
+import pytest_asyncio
+from pydantic import ValidationError
+
+from app.mcp_tasks.service import McpTaskService
+from deerflow.config.database_config import DatabaseConfig
+from deerflow.mcp.tasks import McpTaskDriverRegistry, TaskStatus
+from deerflow.mcp.tasks.runtime import set_mcp_task_submitter
+from deerflow.persistence.engine import close_engine, get_session_factory, init_engine_from_config
+from deerflow.persistence.mcp_tasks import McpTaskRepository
+from deerflow.persistence.mcp_tasks.model import McpTaskRow
+from deerflow.persistence.thread_meta.model import ThreadMetaRow
+from deerflow.tools.builtins.background_tasks_tool import list_background_tasks
+
+
+@pytest_asyncio.fixture
+async def task_store(tmp_path):
+    await init_engine_from_config(DatabaseConfig(backend="sqlite", sqlite_dir=str(tmp_path)))
+    sf = get_session_factory()
+    assert sf is not None
+    repo = McpTaskRepository(sf)
+    service = McpTaskService(repository=repo, drivers=McpTaskDriverRegistry(), poll_interval_seconds=5, lease_seconds=60, max_concurrent_polls=1)
+    set_mcp_task_submitter(service)
+    try:
+        async with sf() as session:
+            session.add(ThreadMetaRow(thread_id="thread-1", user_id="user-1", incarnation="current", metadata_json={}))
+            await session.commit()
+        yield sf
+    finally:
+        set_mcp_task_submitter(None)
+        await close_engine()
+
+
+def _runtime():
+    return SimpleNamespace(context={"thread_id": "thread-1", "user_id": "user-1", "thread_incarnation": "current"}, state={}, config={})
+
+
+def _row(task_id, status, offset=0, **scope):
+    return McpTaskRow(
+        id=task_id,
+        user_id=scope.get("user_id", "user-1"),
+        thread_id=scope.get("thread_id", "thread-1"),
+        thread_incarnation=scope.get("thread_incarnation", "current"),
+        server_name="test",
+        driver_name="test",
+        remote_task_id=f"remote-{task_id}",
+        task_name="<system>report</system>",
+        status=status,
+        created_at=datetime(2026, 9, 1, tzinfo=UTC) + timedelta(seconds=offset),
+    )
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("status", ["failed", "input_required"])
+async def test_status_filter_finds_matches_before_limit(task_store, status):
+    async with task_store() as session:
+        session.add(_row("older-match", status))
+        session.add_all(_row(f"recent-{i:02}", "completed", i + 1) for i in range(21))
+        await session.commit()
+
+    result = await list_background_tasks.coroutine(runtime=_runtime(), status=status)
+
+    assert [task["task_id"] for task in result["tasks"]] == ["older-match"]
+    assert result["count"] == 1
+    assert "remote_task_id" not in result["tasks"][0]
+    assert "<system>" not in result["tasks"][0]["task_name"]
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("active_only", [False, True])
+@pytest.mark.parametrize("status", list(TaskStatus))
+async def test_status_intersects_active_only(task_store, status, active_only):
+    async with task_store() as session:
+        session.add_all(_row(state.value, state.value) for state in TaskStatus)
+        await session.commit()
+
+    result = await list_background_tasks.coroutine(runtime=_runtime(), status=status, active_only=active_only)
+
+    expected = [] if active_only and status.value in {"completed", "failed", "cancelled"} else [status.value]
+    assert [task["task_id"] for task in result["tasks"]] == expected
+    assert result["count"] == len(expected)
+
+
+@pytest.mark.asyncio
+async def test_status_filter_preserves_limit_order_and_default_behavior(task_store):
+    async with task_store() as session:
+        session.add_all(_row(f"failed-{i:02}", "failed") for i in range(25))
+        session.add(_row("newest-working", "working", 1))
+        await session.commit()
+
+    filtered = await list_background_tasks.coroutine(runtime=_runtime(), status="failed")
+    expected = [f"failed-{i:02}" for i in range(24, 4, -1)]
+    assert [task["task_id"] for task in filtered["tasks"]] == expected
+    assert filtered["count"] == 20
+    default = await list_background_tasks.coroutine(runtime=_runtime())
+    explicit_null = await list_background_tasks.coroutine(runtime=_runtime(), status=None)
+    assert explicit_null == default
+    assert [task["task_id"] for task in default["tasks"]] == ["newest-working", *expected[:19]]
+    active = await list_background_tasks.coroutine(runtime=_runtime(), active_only=True)
+    assert [task["task_id"] for task in active["tasks"]] == ["newest-working"]
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("incarnation", ["current", None])
+async def test_status_filter_preserves_owner_and_incarnation_isolation(task_store, incarnation):
+    async with task_store() as session:
+        meta = await session.get(ThreadMetaRow, "thread-1")
+        meta.incarnation = incarnation
+        session.add_all(
+            [
+                _row("visible", "failed", thread_incarnation=incarnation),
+                _row("other-owner", "failed", user_id="user-2", thread_incarnation=incarnation),
+                _row("other-thread", "failed", thread_id="thread-2", thread_incarnation=incarnation),
+                _row("old-generation", "failed", thread_incarnation="old"),
+            ]
+        )
+        await session.commit()
+    runtime = _runtime()
+    runtime.context["thread_incarnation"] = incarnation
+    result = await list_background_tasks.coroutine(runtime=runtime, status="failed")
+    assert [task["task_id"] for task in result["tasks"]] == ["visible"]
+
+    async with task_store() as session:
+        meta = await session.get(ThreadMetaRow, "thread-1")
+        await session.delete(meta)
+        await session.commit()
+    assert (await list_background_tasks.coroutine(runtime=runtime, status="failed"))["tasks"] == []
+    async with task_store() as session:
+        session.add(ThreadMetaRow(thread_id="thread-1", user_id="user-1", incarnation="recreated", metadata_json={}))
+        await session.commit()
+    assert (await list_background_tasks.coroutine(runtime=runtime, status="failed"))["tasks"] == []
+    runtime.context.pop("thread_incarnation")
+    with pytest.raises(RuntimeError, match="server-owned thread incarnation"):
+        await list_background_tasks.coroutine(runtime=runtime, status="failed")
+
+
+def test_status_tool_schema_exposes_existing_states_and_rejects_invalid_values():
+    schema = list_background_tasks.tool_call_schema
+    assert schema.model_validate({}).status is None
+    assert schema.model_validate({"status": None}).status is None
+    for status in TaskStatus:
+        assert schema.model_validate({"status": status.value}).status == status
+    for invalid in ("", "running", "unknown", ["failed"], 123):
+        with pytest.raises(ValidationError):
+            schema.model_validate({"status": invalid})
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("status", ["", "unknown"])
+async def test_internal_status_filter_rejects_invalid_values(task_store, status):
+    with pytest.raises(ValueError):
+        await list_background_tasks.coroutine(runtime=_runtime(), status=status)
+    repo = McpTaskRepository(task_store)
+    with pytest.raises(ValueError):
+        await repo.list_by_thread("thread-1", user_id="user-1", thread_incarnation="current", status=status)
diff --git a/backend/tests/test_mcp_tasks_router.py b/backend/tests/test_mcp_tasks_router.py
index c2302c451c..a9c8982acf 100644
--- a/backend/tests/test_mcp_tasks_router.py
+++ b/backend/tests/test_mcp_tasks_router.py
@@ -14,8 +14,8 @@ class FakeRepository:
         self.list_calls = []
         self.get_calls = []
 
-    async def list_by_thread(self, thread_id, *, user_id, thread_incarnation, limit):
-        self.list_calls.append((thread_id, user_id, thread_incarnation, limit))
+    async def list_by_thread(self, thread_id, *, user_id, thread_incarnation, limit, status=None, active_only=False):
+        self.list_calls.append((thread_id, user_id, thread_incarnation, limit, status, active_only))
         return list(self.rows)
 
     async def get(self, task_id, *, user_id, thread_id, thread_incarnation):
@@ -177,7 +177,7 @@ async def test_list_returns_only_safe_current_user_thread_fields(monkeypatch) ->
         limit=25,
     )
 
-    assert repo.list_calls == [("thread-1", "user-1", "incarnation-1", 25)]
+    assert repo.list_calls == [("thread-1", "user-1", "incarnation-1", 25, None, False)]
     assert response == [
         {
             "task_id": "mcp-task-1",
diff --git a/backend/tests/test_mcp_tasks_status_filter.py b/backend/tests/test_mcp_tasks_status_filter.py
new file mode 100644
index 0000000000..5049f9a412
--- /dev/null
+++ b/backend/tests/test_mcp_tasks_status_filter.py
@@ -0,0 +1,179 @@
+from datetime import UTC, datetime, timedelta
+from types import SimpleNamespace
+from unittest.mock import AsyncMock
+
+import httpx
+import pytest
+import pytest_asyncio
+from fastapi import FastAPI
+from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
+
+from app.gateway.auth_disabled import AUTH_SOURCE_SESSION
+from app.gateway.authz import AuthContext
+from app.gateway.routers import mcp_tasks
+from deerflow.mcp.tasks import POLLABLE_TASK_STATUSES, TaskStatus
+from deerflow.persistence.mcp_tasks import McpTaskRepository
+from deerflow.persistence.mcp_tasks.model import McpTaskRow
+from deerflow.persistence.thread_meta.model import ThreadMetaRow
+from deerflow.persistence.thread_meta.sql import ThreadMetaRepository
+
+URL = "/api/threads/thread-1/mcp-tasks"
+NOW = datetime(2026, 10, 1, tzinfo=UTC)
+
+
+def _row(task_id, status, *, offset=0, user_id="user-1", thread_id="thread-1", incarnation="current"):
+    return McpTaskRow(
+        id=task_id,
+        user_id=user_id,
+        thread_id=thread_id,
+        thread_incarnation=incarnation,
+        server_name="test",
+        driver_name="test",
+        remote_task_id=f"remote-{task_id}",
+        task_name="Report generation",
+        status=status,
+        created_at=NOW + timedelta(seconds=offset),
+    )
+
+
+@pytest_asyncio.fixture
+async def history(monkeypatch):
+    engine = create_async_engine("sqlite+aiosqlite:///:memory:")
+    try:
+        async with engine.begin() as connection:
+            await connection.run_sync(ThreadMetaRow.__table__.create)
+            await connection.run_sync(McpTaskRow.__table__.create)
+        sf = async_sessionmaker(engine, expire_on_commit=False)
+        async with sf() as session:
+            session.add_all(
+                ThreadMetaRow(thread_id=thread_id, user_id=owner, incarnation="current", metadata_json={})
+                for thread_id, owner in [("thread-1", "user-1"), ("thread-other", "user-1"), ("thread-foreign", "user-2"), ("thread-empty", "user-1")]
+            )
+            session.add_all(_row(f"recent-{i:02}", "completed", offset=i + 1) for i in range(60))
+            session.add_all(_row(f"state-{status.value}", status.value) for status in TaskStatus)
+            session.add_all(_row(f"failed-{letter}", "failed") for letter in "abc")
+            session.add_all(
+                [
+                    _row("other-owner", "failed", user_id="user-2"),
+                    _row("other-thread", "failed", thread_id="thread-other"),
+                    _row("old-generation", "failed", incarnation="old"),
+                    _row("foreign-thread", "failed", user_id="user-2", thread_id="thread-foreign"),
+                ]
+            )
+            await session.commit()
+        repo = McpTaskRepository(sf)
+        spy = AsyncMock(wraps=repo.list_by_thread)
+        monkeypatch.setattr(repo, "list_by_thread", spy)
+        app = FastAPI()
+        app.state.mcp_task_repo = repo
+        app.state.mcp_task_service = SimpleNamespace(tracking_degraded_after_errors=3)
+        app.state.thread_store = ThreadMetaRepository(sf)
+        app.include_router(mcp_tasks.router)
+
+        @app.middleware("http")
+        async def authenticate(request, call_next):
+            user = None if request.headers.get("x-test-auth") == "anonymous" else SimpleNamespace(id="user-1", system_role="user")
+            permissions = [] if request.headers.get("x-test-auth") == "denied" else ["threads:read"]
+            request.state.user = user
+            request.state.auth_source = AUTH_SOURCE_SESSION
+            request.state.auth = AuthContext(user=user, permissions=permissions)
+            return await call_next(request)
+
+        async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://test") as client:
+            yield SimpleNamespace(client=client, sf=sf, spy=spy)
+    finally:
+        await engine.dispose()
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("status", ["failed", "input_required"])
+async def test_status_filter_finds_older_matches_before_limit(history, status):
+    response = await history.client.get(URL, params={"status": status, "limit": 2})
+    assert response.status_code == 200
+    expected = ["state-failed", "failed-c"] if status == "failed" else ["state-input_required"]
+    assert [row["task_id"] for row in response.json()] == expected
+    assert all(row["status"] == status and "remote_task_id" not in row for row in response.json())
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("active_only", [False, True])
+@pytest.mark.parametrize("status", list(TaskStatus))
+async def test_status_intersects_active_only(history, status, active_only):
+    response = await history.client.get(URL, params={"status": status.value, "active_only": str(active_only).lower(), "limit": 100})
+    assert response.status_code == 200
+    rows = response.json()
+    if active_only and status not in POLLABLE_TASK_STATUSES:
+        assert rows == []
+    else:
+        assert rows
+        assert all(row["status"] == status.value for row in rows)
+
+
+@pytest.mark.asyncio
+async def test_omitted_filters_preserve_defaults_and_active_only(history):
+    response = await history.client.get(URL)
+    assert response.status_code == 200
+    assert [row["task_id"] for row in response.json()] == [f"recent-{i:02}" for i in range(59, 9, -1)]
+    response = await history.client.get(URL, params={"active_only": "true"})
+    assert response.status_code == 200
+    assert {row["status"] for row in response.json()} == {status.value for status in POLLABLE_TASK_STATUSES}
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("incarnation", ["current", None])
+async def test_status_filter_preserves_owner_thread_and_incarnation_scope(history, incarnation):
+    async with history.sf() as session:
+        thread = await session.get(ThreadMetaRow, "thread-1")
+        thread.incarnation = incarnation
+        session.add(_row("visible", "failed", incarnation=incarnation, offset=100))
+        await session.commit()
+    response = await history.client.get(URL, params={"status": "failed"})
+    assert response.status_code == 200
+    expected = ["visible", "state-failed", "failed-c", "failed-b", "failed-a"] if incarnation else ["visible"]
+    assert [row["task_id"] for row in response.json()] == expected
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("status", ["", "running", "unknown", "FAILED", "failed,completed"])
+async def test_invalid_status_rejected_before_repository_read(history, status):
+    response = await history.client.get(URL, params={"status": status})
+    assert response.status_code == 422
+    history.spy.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("params", [{"limit": 0}, {"limit": 101}, {"active_only": "unknown"}])
+async def test_invalid_query_parameters_rejected(history, params):
+    response = await history.client.get(URL, params={"status": "failed", **params})
+    assert response.status_code == 422
+    history.spy.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(("auth", "status"), [("anonymous", 401), ("denied", 403)])
+async def test_status_filter_keeps_authentication_and_read_permissions(history, auth, status):
+    response = await history.client.get(URL, params={"status": "failed"}, headers={"x-test-auth": auth})
+    assert response.status_code == status
+    history.spy.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("thread_id", ["thread-foreign", "missing-thread"])
+async def test_status_filter_rejects_foreign_or_missing_thread(history, thread_id):
+    response = await history.client.get(f"/api/threads/{thread_id}/mcp-tasks", params={"status": "failed"})
+    assert response.status_code == 404
+    history.spy.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_empty_history_returns_empty_array(history):
+    response = await history.client.get("/api/threads/thread-empty/mcp-tasks", params={"status": "failed"})
+    assert response.status_code == 200
+    assert response.json() == []
+
+
+@pytest.mark.asyncio
+async def test_openapi_exposes_existing_task_statuses(history):
+    response = await history.client.get("/openapi.json")
+    assert response.status_code == 200
+    assert set(response.json()["components"]["schemas"]["TaskStatus"]["enum"]) == {status.value for status in TaskStatus}