fix: bound outbound I/O and async waits

docs/content/docs/framework/tools/builtin.en.mdx

@@ -468,6 +468,7 @@ Environment variables:
 - `TOOL_VESPEECH_API_KEY`: VeSpeech service API key
 - `TOOL_VESPEECH_SPEAKER`: voice, defaults to `zh_female_vv_uranus_bigtts`
 - `TOOL_VESPEECH_AUDIO_OUTPUT_PATH`: audio output directory, defaults to the system temp directory
+- `VEADK_HTTP_STREAM_BUDGET`: total wall-clock ceiling for consuming the streamed synthesis response, defaults to 300 seconds; synthesis aborts with an error once it is exceeded. See [Environment variables · HTTP timeouts](/en/docs/references/configuration/environment-variables#http-timeouts)
 
 ## Code sandboxes
 

docs/content/docs/framework/tools/builtin.mdx

@@ -465,6 +465,7 @@ if __name__ == "__main__":
 - `TOOL_VESPEECH_API_KEY`：VeSpeech 服务的 API Key
 - `TOOL_VESPEECH_SPEAKER`：音色，默认为 `zh_female_vv_uranus_bigtts`
 - `TOOL_VESPEECH_AUDIO_OUTPUT_PATH`：音频输出目录，默认为系统临时目录
+- `VEADK_HTTP_STREAM_BUDGET`：消费流式合成响应的墙钟总时长上限，默认 300 秒；超时后合成中止并报错。详见[环境变量 · HTTP 超时](/cn/docs/references/configuration/environment-variables#http-超时)
 
 ## 代码沙箱
 

docs/content/docs/references/configuration/environment-variables.en.mdx

@@ -187,6 +187,30 @@ Prefix `OBSERVABILITY_`, configuring where traces and metrics are reported.
 | | `OBSERVABILITY_PROMETHEUS_USERNAME` | Username |
 | | `OBSERVABILITY_PROMETHEUS_PASSWORD` | Password |
 
+## HTTP timeouts
+
+The `VEADK_HTTP_*` variables bound outbound `requests` calls VeADK makes internally (built-in tools, knowledge base backends, deployment and integration SDKs). `VEADK_MAX_DOWNLOAD_BYTES` separately caps remote bodies downloaded by media and Skill tooling. `requests` has no default timeout of its own: a peer that accepts the connection and then goes silent blocks the call forever.
+
+| Variable | Default | Meaning |
+| :- | :- | :- |
+| `VEADK_HTTP_CONNECT_TIMEOUT` | `10` | Seconds allowed to establish a TCP/TLS connection. An unreachable peer fails fast instead of occupying a worker. |
+| `VEADK_HTTP_READ_TIMEOUT` | `60` | Maximum seconds allowed **between two consecutive socket reads**. It caps silence, not the total duration of the call, so a large body is not cut off merely for taking a long time to transfer. |
+| `VEADK_HTTP_STREAM_BUDGET` | `300` | **Total wall-clock ceiling** in seconds for consuming a streamed response end to end; used by TTS plus remote media and Skill downloads. This is a different quantity from the read timeout above: a server emitting an endless trickle of valid frames resets the read gap forever and is only caught by a total budget. |
+| `VEADK_MAX_DOWNLOAD_BYTES` | `268435456` | Maximum size of one remote media or Skill download in bytes (default 256 MiB). Oversized bodies are stopped before replacing a destination file. |
+
+Timeout values are clamped to a minimum of `1.0` second and the byte limit to at least one byte. Invalid values fall back to their defaults.
+
+<Callout type="warn">
+These four variables are evaluated at module import time, so they **must come from the real process environment** (`export`, or environment variables injected by a container or FaaS platform). Setting them in `.env` or `config.yaml` has no effect — VeADK loads those into `os.environ` later than these constants are evaluated, so the "every variable can also be expressed in `config.yaml`" rule stated at the top of this page does not apply here.
+</Callout>
+
+```bash
+export VEADK_HTTP_CONNECT_TIMEOUT=10
+export VEADK_HTTP_READ_TIMEOUT=60
+export VEADK_HTTP_STREAM_BUDGET=300
+export VEADK_MAX_DOWNLOAD_BYTES=268435456
+```
+
 ## Others
 
 | Variable | Meaning |

docs/content/docs/references/configuration/environment-variables.mdx

@@ -185,6 +185,30 @@ database:
 | | `OBSERVABILITY_PROMETHEUS_USERNAME` | 用户名 |
 | | `OBSERVABILITY_PROMETHEUS_PASSWORD` | 密码 |
 
+## HTTP 超时
+
+`VEADK_HTTP_*` 用于约束 VeADK 内部基于 `requests` 的出站调用（内置工具、知识库后端、部署与集成 SDK）；`VEADK_MAX_DOWNLOAD_BYTES` 另行限制媒体和 Skill 工具下载的远程响应体大小。`requests` 自身没有默认超时：对端接受连接后不再返回数据时，调用会永久挂起。
+
+| 环境变量 | 默认值 | 释义 |
+| :- | :- | :- |
+| `VEADK_HTTP_CONNECT_TIMEOUT` | `10` | 建立 TCP/TLS 连接的秒数上限。对端不可达时快速失败，而不是长时间占用一个 worker。 |
+| `VEADK_HTTP_READ_TIMEOUT` | `60` | **两次相邻 socket 读取之间**允许的最大间隔秒数。它约束的是「静默」而非响应总时长，因此大响应体不会仅因为传输耗时长而被中断。 |
+| `VEADK_HTTP_STREAM_BUDGET` | `300` | 完整消费一个流式响应的**墙钟总时长**上限（秒），用于 TTS、远程媒体及 Skill 下载。这与上面的读超时是两个不同的量：持续吐出合法分片的服务端会无限刷新读间隔，只有总预算能兜住这种情况。 |
+| `VEADK_MAX_DOWNLOAD_BYTES` | `268435456` | 单次远程媒体或 Skill 下载的最大字节数，默认 256 MiB。超限响应会在替换目标文件前终止。 |
+
+超时值最小为 `1.0` 秒，字节上限最小为 1 字节；无法解析时回退到各自默认值。
+
+<Callout type="warn">
+这四个变量在模块导入时求值，因此**必须来自真实的进程环境变量**（`export`、容器或函数计算注入的环境变量等）。写在 `.env` 或 `config.yaml` 中不会生效——VeADK 将它们写入 `os.environ` 的时机晚于这些常量的求值，此处不适用本页开头「环境变量与 `config.yaml` 等价」的约定。
+</Callout>
+
+```bash
+export VEADK_HTTP_CONNECT_TIMEOUT=10
+export VEADK_HTTP_READ_TIMEOUT=60
+export VEADK_HTTP_STREAM_BUDGET=300
+export VEADK_MAX_DOWNLOAD_BYTES=268435456
+```
+
 ## 其他
 
 | 环境变量 | 释义 |

tests/a2a/test_hub_client_timeouts.py

@@ -0,0 +1,134 @@
+# Copyright (c) 2025 Beijing Volcano Engine Technology Co., Ltd. and/or its affiliates.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Regression tests: A2A hub/remote-agent HTTP calls must be bounded.
+
+The hub liveness probe deliberately runs on a tighter read budget than the
+ordinary calls -- a hub that has gone silent should be reported as down
+quickly, not after a full minute.
+"""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock
+
+import pytest
+from a2a.types import AgentCapabilities, AgentCard
+
+from veadk.utils.http_defaults import DEFAULT_CONNECT_TIMEOUT, DEFAULT_HTTP_TIMEOUT
+
+
+def _agent_card() -> AgentCard:
+    return AgentCard(
+        name="weather-agent",
+        description="Weather agent",
+        url="http://127.0.0.1:8000",
+        version="1.0.0",
+        capabilities=AgentCapabilities(),
+        default_input_modes=["text/plain"],
+        default_output_modes=["text/plain"],
+        skills=[],
+    )
+
+
+def _response(payload: dict | None = None, status_code: int = 200) -> MagicMock:
+    response = MagicMock()
+    response.status_code = status_code
+    response.json.return_value = payload if payload is not None else {}
+    return response
+
+
+def test_health_check_read_budget_is_tighter_than_default() -> None:
+    from veadk.a2a.hub.a2a_hub_client import HEALTH_CHECK_TIMEOUT
+
+    assert HEALTH_CHECK_TIMEOUT[0] == DEFAULT_CONNECT_TIMEOUT
+    assert HEALTH_CHECK_TIMEOUT[1] < DEFAULT_HTTP_TIMEOUT[1]
+
+
+def test_health_check_passes_health_check_timeout(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    from veadk.a2a.hub import a2a_hub_client as module
+
+    get = MagicMock(return_value=_response())
+    monkeypatch.setattr(module.requests, "get", get)
+
+    module.A2AHubClient(server_host="127.0.0.1", server_port=8888)
+
+    assert get.call_count == 1
+    assert get.call_args.kwargs["timeout"] == module.HEALTH_CHECK_TIMEOUT
+
+
+def test_get_agent_cards_passes_default_http_timeout(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    from veadk.a2a.hub import a2a_hub_client as module
+
+    get = MagicMock(return_value=_response({"agent_infos": [{"agent_id": "a1"}]}))
+    monkeypatch.setattr(module.requests, "get", get)
+
+    client = module.A2AHubClient(server_host="127.0.0.1", server_port=8888)
+    assert client.get_agent_cards(group_id="g1") == [{"agent_id": "a1"}]
+
+    # First call is the constructor's health check.
+    assert get.call_count == 2
+    assert get.call_args.kwargs["timeout"] == DEFAULT_HTTP_TIMEOUT
+
+
+def test_register_agent_passes_default_http_timeout(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    from veadk.a2a.hub import a2a_hub_client as module
+
+    monkeypatch.setattr(module.requests, "get", MagicMock(return_value=_response()))
+    post = MagicMock(return_value=_response())
+    monkeypatch.setattr(module.requests, "post", post)
+
+    client = module.A2AHubClient(server_host="127.0.0.1", server_port=8888)
+    client.register_agent(group_id="g1", agent_id="a1", agent_card=_agent_card())
+
+    assert post.call_count == 1
+    assert post.call_args.kwargs["timeout"] == DEFAULT_HTTP_TIMEOUT
+
+
+def test_create_group_passes_default_http_timeout(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    from veadk.a2a.hub import a2a_hub_client as module
+
+    monkeypatch.setattr(module.requests, "get", MagicMock(return_value=_response()))
+    post = MagicMock(return_value=_response())
+    monkeypatch.setattr(module.requests, "post", post)
+
+    client = module.A2AHubClient(server_host="127.0.0.1", server_port=8888)
+    client.create_group(group_id="g1")
+
+    assert post.call_count == 1
+    assert post.call_args.kwargs["timeout"] == DEFAULT_HTTP_TIMEOUT
+
+
+def test_remote_ve_agent_card_fetch_passes_default_http_timeout(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    from veadk.a2a import remote_ve_agent as module
+
+    card = _agent_card().model_dump(mode="json", by_alias=True)
+    get = MagicMock(return_value=_response(card))
+    monkeypatch.setattr(module.requests, "get", get)
+
+    agent = module.RemoteVeAgent(name="weather_agent", url="http://127.0.0.1:8000")
+
+    assert agent.name == "weather_agent"
+    assert get.call_count == 1
+    assert get.call_args.kwargs["timeout"] == DEFAULT_HTTP_TIMEOUT

tests/a2a/test_ve_task_store.py

@@ -0,0 +1,153 @@
+# Copyright (c) 2025 Beijing Volcano Engine Technology Co., Ltd. and/or its affiliates.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Regression tests: `VeTaskStore` must honour the A2A `TaskStore` contract.
+
+Two things used to be wrong. The overrides dropped the `context` parameter,
+which the A2A request handlers pass *positionally*
+(`await self.task_store.get(params.id, context)`), so every call raised
+`TypeError`. And the bodies were `return None` stubs, so a store that did get
+called would have silently discarded every task.
+
+The signature expectations below are derived from `TaskStore` itself rather
+than hardcoded, so an upstream parameter change fails these tests instead of
+letting the override quietly drift out of sync again.
+"""
+
+from __future__ import annotations
+
+import inspect
+
+import pytest
+from a2a.server.context import ServerCallContext
+from a2a.server.tasks import TaskStore
+from a2a.types import Task, TaskState, TaskStatus
+
+from veadk.a2a.ve_task_store import VeTaskStore
+
+TASK_STORE_METHODS = ["save", "get", "delete"]
+
+
+def _call_contract(func) -> list[tuple]:
+    """The part of a signature a caller must satisfy: name, kind, default."""
+    return [
+        (param.name, param.kind, param.default)
+        for param in inspect.signature(func).parameters.values()
+    ]
+
+
+def _task(task_id: str, state: TaskState = TaskState.submitted) -> Task:
+    return Task(
+        id=task_id,
+        context_id="test-context",
+        status=TaskStatus(state=state),
+    )
+
+
+@pytest.mark.parametrize("method_name", TASK_STORE_METHODS)
+def test_override_matches_base_call_contract(method_name: str) -> None:
+    """Each override accepts exactly what the base class promises callers."""
+    expected = _call_contract(getattr(TaskStore, method_name))
+    actual = _call_contract(getattr(VeTaskStore, method_name))
+
+    assert actual == expected
+
+
+@pytest.mark.asyncio
+async def test_context_accepted_positionally() -> None:
+    """A2A's request handlers and task manager pass `context` positionally."""
+    store = VeTaskStore()
+    context = ServerCallContext()
+    task = _task("positional-task")
+
+    await store.save(task, context)
+    assert await store.get(task.id, context) == task
+
+    await store.delete(task.id, context)
+    assert await store.get(task.id, context) is None
+
+
+@pytest.mark.asyncio
+async def test_context_accepted_by_keyword() -> None:
+    store = VeTaskStore()
+    context = ServerCallContext()
+    task = _task("keyword-task")
+
+    await store.save(task, context=context)
+    assert await store.get(task.id, context=context) == task
+
+    await store.delete(task.id, context=context)
+    assert await store.get(task.id, context=context) is None
+
+
+@pytest.mark.asyncio
+async def test_context_is_optional() -> None:
+    """`context` defaults to None, so callers may omit it entirely."""
+    store = VeTaskStore()
+    task = _task("no-context-task")
+
+    await store.save(task)
+    assert await store.get(task.id) == task
+
+    await store.delete(task.id)
+    assert await store.get(task.id) is None
+
+
+@pytest.mark.asyncio
+async def test_save_updates_existing_task() -> None:
+    """Re-saving an id replaces the stored task rather than duplicating it."""
+    store = VeTaskStore()
+    await store.save(_task("task-1", TaskState.submitted))
+    await store.save(_task("task-1", TaskState.completed))
+
+    stored = await store.get("task-1")
+
+    assert stored is not None
+    assert stored.status.state == TaskState.completed
+
+
+@pytest.mark.asyncio
+async def test_get_returns_none_only_for_unknown_ids() -> None:
+    """A miss must mean "absent", not "this store never returns anything"."""
+    store = VeTaskStore()
+    task = _task("stored-task")
+    await store.save(task)
+
+    assert await store.get("never-saved") is None
+    assert await store.get(task.id) == task
+
+
+@pytest.mark.asyncio
+async def test_delete_of_unknown_id_leaves_other_tasks_intact() -> None:
+    """Deleting a nonexistent task is tolerated and touches nothing else."""
+    store = VeTaskStore()
+    task = _task("survivor-task")
+    await store.save(task)
+
+    await store.delete("never-saved")
+
+    assert await store.get(task.id) == task
+
+
+@pytest.mark.asyncio
+async def test_tasks_are_not_shared_between_instances() -> None:
+    """Each store owns its tasks; nothing leaks through class-level state."""
+    first = VeTaskStore()
+    second = VeTaskStore()
+    task = _task("instance-task")
+
+    await first.save(task)
+
+    assert await first.get(task.id) == task
+    assert await second.get(task.id) is None