Codex/fix issues 119 183

backend/app/api/tools.py

@@ -4,6 +4,7 @@
 
 import httpx
 from fastapi import APIRouter, Depends, HTTPException, Query
+from pydantic import BaseModel, Field
 from sqlmodel import Session, select
 
 from app.agents.branching import (
@@ -22,6 +23,8 @@
 from app.capability_scope import normalize_capability_scope
 from app.db import get_session
 from app.db.models import (
+    A2ATaskEvent,
+    A2ATaskRun,
     AgentEvent,
     AgentProfile,
     AgentResourceBinding,
@@ -79,6 +82,43 @@
 MCP_APP_RESOURCE_MAX_BYTES = 10 * 1024 * 1024
 
 
+class A2ATaskEventRead(BaseModel):
+    sequence: int
+    event_type: str
+    data: dict[str, Any] = Field(default_factory=dict)
+    created_at: str
+
+
+class A2ATaskRunRead(BaseModel):
+    id: str
+    direction: str
+    remote_task_id: str | None = None
+    context_id: str | None = None
+    codex_session_id: str | None = None
+    status: str
+    endpoint_url: str
+    protocol_version: str
+    cancel_requested: bool
+    recovery_attempts: int
+    artifacts: list[dict[str, Any]] = Field(default_factory=list)
+    error: dict[str, Any] = Field(default_factory=dict)
+    created_at: str
+    started_at: str | None = None
+    finished_at: str | None = None
+    updated_at: str
+    events: list[A2ATaskEventRead] = Field(default_factory=list)
+
+
+class CodexA2AAdapterRead(BaseModel):
+    enabled: bool
+    endpoint_url: str
+    agent_card_url: str
+    command: str
+    workspace_root: str
+    timeout_seconds: float
+    token_configured: bool
+
+
 def tool_read(row: Tool, metadata: dict[str, Any] | None = None) -> ToolRead:
     config = dict(row.config_json or {})
     return ToolRead(
@@ -317,6 +357,97 @@ def probe_tool(
     )
 
 
+@router.get(
+    "/a2a/codex-adapter",
+    response_model=CodexA2AAdapterRead,
+    dependencies=[Depends(require_agent_scope_viewer)],
+)
+def get_codex_a2a_adapter() -> CodexA2AAdapterRead:
+    settings = get_settings()
+    return CodexA2AAdapterRead(
+        enabled=bool(settings.codex_a2a_enabled),
+        endpoint_url="/api/a2a/codex",
+        agent_card_url="/.well-known/agent-card.json",
+        command=settings.codex_a2a_command,
+        workspace_root=settings.codex_a2a_workspace_root,
+        timeout_seconds=settings.codex_a2a_timeout_seconds,
+        token_configured=bool(settings.codex_a2a_token),
+    )
+
+
+@router.get(
+    "/{tool_id}/a2a-runs",
+    response_model=list[A2ATaskRunRead],
+    dependencies=[Depends(require_agent_scope_viewer)],
+)
+def list_a2a_task_runs(
+    tool_id: str,
+    tenant_id: str = Query(...),
+    agent_id: str | None = Query(default=None),
+    limit: int = Query(default=20, ge=1, le=100),
+    db: Session = Depends(get_session),
+) -> list[A2ATaskRunRead]:
+    row = _get_tool(db, tenant_id, tool_id)
+    _ensure_tool_visible(db, tenant_id, row, agent_id)
+    if row.tool_type != "a2a":
+        raise HTTPException(status_code=400, detail="仅 A2A 工具包含持久化任务记录")
+    runs = list(
+        db.exec(
+            select(A2ATaskRun)
+            .where(
+                A2ATaskRun.tenant_id == tenant_id,
+                A2ATaskRun.tool_id == tool_id,
+            )
+            .order_by(A2ATaskRun.created_at.desc())
+            .limit(limit)
+        ).all()
+    )
+    if not runs:
+        return []
+    events_by_run: dict[str, list[A2ATaskEventRead]] = {run.id: [] for run in runs}
+    events = list(
+        db.exec(
+            select(A2ATaskEvent)
+            .where(A2ATaskEvent.run_id.in_([run.id for run in runs]))
+            .order_by(A2ATaskEvent.run_id, A2ATaskEvent.sequence)
+        ).all()
+    )
+    for event in events:
+        events_by_run.setdefault(event.run_id, []).append(
+            A2ATaskEventRead(
+                sequence=event.sequence,
+                event_type=event.event_type,
+                data=dict(event.data_json or {}),
+                created_at=event.created_at.isoformat(),
+            )
+        )
+    return [_a2a_task_run_read(run, events_by_run.get(run.id, [])) for run in runs]
+
+
+@router.post("/{tool_id}/a2a-runs/{run_id}:cancel", response_model=A2ATaskRunRead)
+def cancel_a2a_task_run(
+    tool_id: str,
+    run_id: str,
+    tenant_id: str = Query(...),
+    agent_id: str | None = Query(default=None),
+    db: Session = Depends(get_session),
+    current_user: User = Depends(get_current_user),
+) -> A2ATaskRunRead:
+    row = _get_tool(db, tenant_id, tool_id)
+    ensure_agent_scope_manager(db, tenant_id, agent_id, current_user)
+    _ensure_tool_visible(db, tenant_id, row, agent_id)
+    run = db.get(A2ATaskRun, run_id)
+    if run is None or run.tenant_id != tenant_id or run.tool_id != tool_id:
+        raise HTTPException(status_code=404, detail="A2A 任务不存在")
+    if run.status not in {"completed", "failed", "canceled", "cancelled", "rejected"}:
+        run.cancel_requested = True
+        run.updated_at = utc_now()
+        db.add(run)
+        db.commit()
+        db.refresh(run)
+    return _a2a_task_run_read(run, [])
+
+
 @router.get(
     "/{tool_id}", response_model=ToolRead, dependencies=[Depends(require_agent_scope_viewer)]
 )
@@ -470,6 +601,31 @@ def _get_tool(db: Session, tenant_id: str, tool_id: str) -> Tool:
     return row
 
 
+def _a2a_task_run_read(
+    run: A2ATaskRun,
+    events: list[A2ATaskEventRead],
+) -> A2ATaskRunRead:
+    return A2ATaskRunRead(
+        id=run.id,
+        direction=run.direction,
+        remote_task_id=run.remote_task_id,
+        context_id=run.context_id,
+        codex_session_id=run.codex_session_id,
+        status=run.status,
+        endpoint_url=run.endpoint_url,
+        protocol_version=run.protocol_version,
+        cancel_requested=run.cancel_requested,
+        recovery_attempts=run.recovery_attempts,
+        artifacts=list(run.artifacts_json or []),
+        error=dict(run.error_json or {}),
+        created_at=run.created_at.isoformat(),
+        started_at=run.started_at.isoformat() if run.started_at else None,
+        finished_at=run.finished_at.isoformat() if run.finished_at else None,
+        updated_at=run.updated_at.isoformat(),
+        events=events,
+    )
+
+
 def _visible_tool_rows(
     db: Session,
     tenant_id: str,

backend/app/tools/mcp_client.py

@@ -264,9 +264,43 @@ def list_tools(self) -> list[dict[str, Any]]:
     def list_tools_with_capabilities(self) -> tuple[list[dict[str, Any]], dict[str, Any]]:
         with self:
             self._initialize()
-            result = self._request("tools/list", {})
-            tools = result.get("tools") if isinstance(result, dict) else None
-            return tools if isinstance(tools, list) else [], dict(self.initialize_result)
+            tools = self._list_all_pages("tools/list", "tools")
+            return tools, dict(self.initialize_result)
+
+    def _list_all_pages(self, method: str, result_key: str) -> list[dict[str, Any]]:
+        """Collect a cursor-paginated MCP list result.
+
+        MCP servers may paginate ``tools/list`` even when the first version of a
+        server returned every tool in one response.  Discovery must therefore
+        follow ``nextCursor`` on every refresh; otherwise tools added later can
+        remain permanently invisible once they move beyond the first page.
+        """
+
+        items: list[dict[str, Any]] = []
+        cursor: str | None = None
+        seen_cursors: set[str] = set()
+        while True:
+            params = {"cursor": cursor} if cursor is not None else {}
+            result = self._request(method, params)
+            if not isinstance(result, dict):
+                raise MCPClientError(f"MCP {method} 返回内容不是 object。")
+            page = result.get(result_key)
+            if page is not None and not isinstance(page, list):
+                raise MCPClientError(f"MCP {method} 的 {result_key} 不是数组。")
+            items.extend(item for item in (page or []) if isinstance(item, dict))
+
+            raw_cursor = result.get("nextCursor")
+            if raw_cursor is None:
+                # Tolerate snake_case implementations while keeping the MCP
+                # specification's nextCursor as the canonical form.
+                raw_cursor = result.get("next_cursor")
+            next_cursor = str(raw_cursor).strip() if raw_cursor is not None else ""
+            if not next_cursor:
+                return items
+            if next_cursor in seen_cursors:
+                raise MCPClientError(f"MCP {method} 返回了重复分页游标：{next_cursor}")
+            seen_cursors.add(next_cursor)
+            cursor = next_cursor
 
     def read_resource(self, uri: str) -> dict[str, Any]:
         with self:

backend/tests/test_enterprise_auth_guards.py

@@ -42,7 +42,9 @@ def test_enterprise_read_endpoints_require_authentication() -> None:
         "/api/enterprise/memories?tenant_id=tenant_demo",
         "/api/enterprise/tools?tenant_id=tenant_demo",
         "/api/enterprise/tools/buckets?tenant_id=tenant_demo",
+        "/api/enterprise/tools/a2a/codex-adapter",
         "/api/enterprise/tools/tool_demo?tenant_id=tenant_demo",
+        "/api/enterprise/tools/tool_demo/a2a-runs?tenant_id=tenant_demo",
         "/api/enterprise/mcp-servers?tenant_id=tenant_demo",
         "/api/enterprise/mcp-servers/server_demo?tenant_id=tenant_demo",
         "/api/enterprise/general-skills?tenant_id=tenant_demo",

backend/tests/test_mcp_client_stdio.py

@@ -12,6 +12,7 @@
 
 from app.tools.mcp_client import (
     MCPClientError,
+    _MCPSession,
     _PipeReader,
     _read_response,
     _send_json,
@@ -22,11 +23,50 @@
 )
 
 
+class _PagedToolSession(_MCPSession):
+    def __init__(self, *, repeat_cursor: bool = False) -> None:
+        super().__init__({}, timeout_seconds=1)
+        self.repeat_cursor = repeat_cursor
+        self.requests: list[tuple[str, dict[str, object]]] = []
+
+    def _request(self, method: str, params: dict[str, object]):  # type: ignore[override]
+        self.requests.append((method, params))
+        if method == "initialize":
+            return {"capabilities": {"tools": {"listChanged": True}}}
+        if method != "tools/list":
+            raise AssertionError(f"unexpected method: {method}")
+        if params.get("cursor") is None:
+            return {"tools": [{"name": "existing"}], "nextCursor": "page-2"}
+        return {
+            "tools": [{"name": "newly_added"}],
+            "nextCursor": "page-2" if self.repeat_cursor else None,
+        }
+
+    def _notify(self, method: str, params: dict[str, object]) -> None:  # type: ignore[override]
+        self.requests.append((method, params))
+
+
 class _WindowsAnonymousPipe(io.StringIO):
     def fileno(self) -> int:
         raise OSError(10038, "在一个非套接字上尝试了一个操作")
 
 
+def test_tools_list_discovers_new_tools_from_all_cursor_pages() -> None:
+    session = _PagedToolSession()
+
+    tools, initialize_result = session.list_tools_with_capabilities()
+
+    assert [tool["name"] for tool in tools] == ["existing", "newly_added"]
+    assert initialize_result["capabilities"]["tools"]["listChanged"] is True
+    assert ("tools/list", {}) in session.requests
+    assert ("tools/list", {"cursor": "page-2"}) in session.requests
+
+
+def test_tools_list_rejects_repeated_cursor_instead_of_looping() -> None:
+    with pytest.raises(MCPClientError, match="重复分页游标"):
+        _PagedToolSession(repeat_cursor=True).list_tools_with_capabilities()
+
+
 class _FakeProcess:
     def __init__(self, exit_code: int | None = None) -> None:
         self.exit_code = exit_code