Fix native MCP project search scope and bounded results (JVNAUTOSCI-2737)

docs/engineering/workflow_first_jira_to_von_migration_jvnautosci_1223.md

@@ -111,6 +111,10 @@ same canonical services, including `task_list_projects`, `task_get_project`,
 `task_get` accepts legacy Jira keys as well as Von task concept IDs.
 Project listings return navigation metadata and descriptions; retrieve source
 archives, retained-document references and their histories with `task_get_project`.
+Stdio `task_search` returns task summaries, short description previews and source
+identities; use `task_get` for the full task and `task_get_source_archive` for its
+retained original. Assignee and creator filters are optional query criteria,
+independent of the configured authenticated task actor.
 
 A local stdio connection can bind the trusted operator once in its server
 environment:

src/backend/mcp_server/mcp_stdio_server.py

@@ -1260,6 +1260,20 @@ async def call_tool(name: str, arguments: Any) -> list[TextContent]:  # type: ig
                     "author_concept_id",
                     "user_concept_id",
                 )
+                if name == "task_search":
+                    # These arguments select tasks; they do not select the
+                    # authenticated principal. Keep omitted filters omitted
+                    # and permit searching other people within canonical ACLs.
+                    actor_fields = tuple(
+                        key
+                        for key in actor_fields
+                        if key
+                        not in {
+                            "user_concept_id",
+                            "created_by_concept_id",
+                            "creator_concept_id",
+                        }
+                    )
                 if any(
                     parsed_arguments.get(key) not in (None, "", configured_actor)
                     for key in actor_fields
@@ -4816,6 +4830,51 @@ async def _handle_otter_collection_status(arguments: dict[str, Any]) -> list[Tex
     return [_json_text(call_collection("status", **arguments))]
 
 
+def _task_search_summary(task):
+    """Keep search navigation bounded; task_get retains the full task details."""
+    fields = (
+        "task_concept_id",
+        "title",
+        "status",
+        "priority",
+        "assignee_concept_id",
+        "created_by_concept_id",
+        "reporter_concept_id",
+        "report_to_concept_id",
+        "project_concept_id",
+        "collection_concept_ids",
+        "organisation_concept_id",
+        "reference_code",
+        "task_type_ids",
+        "primary_task_type_label",
+        "task_source_id",
+        "task_source_label",
+        "parent_task_concept_id",
+        "epic_task_concept_id",
+        "start_date",
+        "due_date",
+        "created_at",
+        "updated_at",
+        "comments_count",
+        "attachments_count",
+        "worklog_entries_count",
+        "is_imported_jira_task",
+    )
+    summary = {key: task[key] for key in fields if key in task}
+    description = task.get("description")
+    if isinstance(description, str):
+        summary["description_preview"] = description[:300]
+        summary["description_truncated"] = len(description) > 300
+    summary["external_references"] = {
+        source: {
+            key: ref[key] for key in ("external_id", "issue_id", "url") if key in ref
+        }
+        for source, ref in (task.get("external_references") or {}).items()
+        if isinstance(ref, dict)
+    }
+    return summary
+
+
 def _native_task_handler(name):
     async def handle(arguments):
         from src.backend.integrations.internal_mcp import catalogue
@@ -4824,6 +4883,15 @@ async def handle(arguments):
         # by call_tool before reaching this adapter; actor-bound callers retain
         # their context and ordinary continuation checks.
         result = getattr(catalogue, f"_{name}")(**arguments)
+        if name == "task_search" and result.get("success"):
+            result = {
+                **result,
+                "tasks": [
+                    _task_search_summary(task) for task in result.get("tasks", [])
+                ],
+                "projection": "task_search_summary",
+                "detail_tool": "task_get",
+            }
         return [_json_text(result)]
 
     return handle

tests/backend/test_mcp_stdio_task_actor.py

@@ -57,6 +57,116 @@ async def handler(arguments):
     assert json.loads(result[0].text)["error_code"] == "task_actor_mismatch"
 
 
+@pytest.mark.asyncio
+async def test_project_search_does_not_add_assignee_or_creator_filters(monkeypatch):
+    from src.backend.services import task_management_service as tasks
+
+    monkeypatch.setenv("VON_MCP_TASK_ACTOR_CONCEPT_ID", "#V#owner")
+    monkeypatch.setenv("VON_MCP_TASK_ORGANISATION_CONCEPT_ID", "#V#org")
+
+    def search(**kwargs):
+        assert get_effective_user_concept_id() == "#V#owner"
+        assert get_effective_organisation_concept_id() == "#V#org"
+        assert kwargs["project_concept_id"] == "#V#project"
+        assert kwargs["assignee_concept_id"] is None
+        assert kwargs["created_by_concept_id"] is None
+        return {"tasks": [], "total": 0}
+
+    monkeypatch.setattr(tasks, "search_tasks", search)
+    original = {"project_concept_id": "#V#project"}
+    result = await server.call_tool("task_search", original)
+    assert json.loads(result[0].text)["success"]
+    assert original == {"project_concept_id": "#V#project"}
+    assert get_effective_user_concept_id() is None
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ("argument", "filter_name"),
+    [
+        ("user_concept_id", "assignee_concept_id"),
+        ("assignee_id", "assignee_concept_id"),
+        ("created_by_concept_id", "created_by_concept_id"),
+        ("creator_concept_id", "created_by_concept_id"),
+    ],
+)
+async def test_task_person_filters_do_not_replace_the_actor(
+    monkeypatch, argument, filter_name
+):
+    from src.backend.services import task_management_service as tasks
+
+    monkeypatch.setenv("VON_MCP_TASK_ACTOR_CONCEPT_ID", "#V#owner")
+
+    def search(**kwargs):
+        assert get_effective_user_concept_id() == "#V#owner"
+        assert kwargs[filter_name] == "#V#colleague"
+        return {"tasks": [], "total": 0}
+
+    monkeypatch.setattr(tasks, "search_tasks", search)
+    result = await server.call_tool("task_search", {argument: "#V#colleague"})
+    assert json.loads(result[0].text)["success"]
+    assert get_effective_user_concept_id() is None
+
+
+@pytest.mark.asyncio
+async def test_task_search_still_rejects_a_conflicting_actor(monkeypatch):
+    from src.backend.services import task_management_service as tasks
+
+    monkeypatch.setenv("VON_MCP_TASK_ACTOR_CONCEPT_ID", "#V#owner")
+
+    def search(**kwargs):
+        pytest.fail("A conflicting actor must not reach canonical search")
+
+    monkeypatch.setattr(tasks, "search_tasks", search)
+    result = await server.call_tool(
+        "task_search",
+        {
+            "created_by_concept_id": "#V#colleague",
+            "acting_user_concept_id": "#V#colleague",
+        },
+    )
+    assert json.loads(result[0].text)["error_code"] == "task_actor_mismatch"
+
+
+@pytest.mark.asyncio
+async def test_project_search_returns_summaries_when_import_details_exceed_transport(
+    monkeypatch,
+):
+    from src.backend.integrations.internal_mcp import catalogue
+
+    monkeypatch.setenv("VON_MCP_TASK_ACTOR_CONCEPT_ID", "#V#owner")
+    monkeypatch.setenv("VON_MCP_STDIO_MAX_RESPONSE_CHARS", "10000")
+    task = {
+        "task_concept_id": "#V#task",
+        "title": "Imported task",
+        "description": "Retained description. " * 1000,
+        "project_concept_id": "#V#project",
+        "external_references": {
+            "jira": {
+                "external_id": "TEST-1",
+                "source_archive_history": [{"retained_source": "x" * 10000}],
+            }
+        },
+    }
+    monkeypatch.setattr(
+        catalogue,
+        "_task_search",
+        lambda **kwargs: {"success": True, "tasks": [task], "total": 1},
+    )
+    result = await server.call_tool("task_search", {"project_concept_id": "#V#project"})
+    payload = json.loads(result[0].text)
+    assert payload["success"] and payload["total"] == 1
+    assert payload["detail_tool"] == "task_get"
+    summary = payload["tasks"][0]
+    assert summary["task_concept_id"] == task["task_concept_id"]
+    assert summary["project_concept_id"] == "#V#project"
+    assert summary["external_references"]["jira"]["external_id"] == "TEST-1"
+    assert summary["description_truncated"]
+    assert task["description"].startswith(summary["description_preview"])
+    assert "source_archive_history" in task["external_references"]["jira"]
+    assert len(task["description"]) > 10000
+
+
 @pytest.mark.asyncio
 async def test_configured_operator_does_not_replace_an_inherited_task_context(
     monkeypatch,