Compare commits
1 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| ba0680d5fb |
@@ -21,6 +21,7 @@ import (
|
||||
"time"
|
||||
|
||||
"github.com/Molecule-AI/molecule-monorepo/platform/internal/db"
|
||||
"github.com/Molecule-AI/molecule-monorepo/platform/internal/envx"
|
||||
"github.com/Molecule-AI/molecule-monorepo/platform/internal/events"
|
||||
"github.com/Molecule-AI/molecule-monorepo/platform/internal/models"
|
||||
"github.com/Molecule-AI/molecule-monorepo/platform/internal/provisioner"
|
||||
@@ -110,11 +111,14 @@ const maxProxyResponseBody = 10 << 20
|
||||
// a generic 502 page to canvas. 10s is well above realistic intra-region
|
||||
// latencies and well below CF's edge timeout.
|
||||
//
|
||||
// 3. Transport.ResponseHeaderTimeout — 60s. From request-body-end to
|
||||
// response-headers-start. Covers cold-start first-byte (the 30-60s OAuth
|
||||
// flow above), with margin. Body streaming after headers is governed by
|
||||
// the per-request context deadline, NOT this timeout — so multi-minute
|
||||
// agent responses still work fine.
|
||||
// 3. Transport.ResponseHeaderTimeout — 180s default. From request-body-end
|
||||
// to response-headers-start. Configurable via
|
||||
// A2A_PROXY_RESPONSE_HEADER_TIMEOUT (envx.Duration). Covers cold-start
|
||||
// first-byte (30-60s OAuth flow above) with enough room for Opus agent
|
||||
// turns (big context + internal delegate_task round-trips routinely exceed
|
||||
// the old 60s ceiling). Body streaming after headers is governed by the
|
||||
// per-request context deadline, NOT this timeout — so multi-minute agent
|
||||
// responses still work fine.
|
||||
//
|
||||
// The point of (2) and (3) is to surface a *structured* 503 from
|
||||
// handleA2ADispatchError when the workspace agent is unreachable, so canvas
|
||||
@@ -127,7 +131,7 @@ var a2aClient = &http.Client{
|
||||
Timeout: 10 * time.Second,
|
||||
KeepAlive: 30 * time.Second,
|
||||
}).DialContext,
|
||||
ResponseHeaderTimeout: 60 * time.Second,
|
||||
ResponseHeaderTimeout: envx.Duration("A2A_PROXY_RESPONSE_HEADER_TIMEOUT", 180*time.Second),
|
||||
TLSHandshakeTimeout: 10 * time.Second,
|
||||
// MaxIdleConns / IdleConnTimeout: stdlib defaults are fine; agent
|
||||
// fan-in is bounded by the platform's broadcaster fan-out, not by
|
||||
|
||||
@@ -2276,3 +2276,43 @@ func TestProxyA2A_PollMode_FailsClosedToPush(t *testing.T) {
|
||||
t.Errorf("unmet sqlmock expectations: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// ==================== a2aClient ResponseHeaderTimeout config ====================
|
||||
|
||||
func TestA2AClientResponseHeaderTimeout(t *testing.T) {
|
||||
const defaultTimeout = 180 * time.Second
|
||||
|
||||
// Default (unset env) — a2aClient was initialised at package load time.
|
||||
if a2aClient.Transport.(*http.Transport).ResponseHeaderTimeout != defaultTimeout {
|
||||
t.Errorf("a2aClient default ResponseHeaderTimeout = %v, want %v",
|
||||
a2aClient.Transport.(*http.Transport).ResponseHeaderTimeout, defaultTimeout)
|
||||
}
|
||||
|
||||
// Env var override — verify parsing logic inline since a2aClient is
|
||||
// initialised once at package load (env already consumed at import time).
|
||||
t.Run("A2A_PROXY_RESPONSE_HEADER_TIMEOUT parsed correctly", func(t *testing.T) {
|
||||
// We can't re-initialise a2aClient, but we can verify the same
|
||||
// envx.Duration logic inline for the 5m override case.
|
||||
t.Setenv("A2A_PROXY_RESPONSE_HEADER_TIMEOUT", "5m")
|
||||
if d, err := time.ParseDuration("5m"); err == nil && d > 0 {
|
||||
if d != 5*time.Minute {
|
||||
t.Errorf("ParseDuration(\"5m\") = %v, want 5m", d)
|
||||
}
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("invalid A2A_PROXY_RESPONSE_HEADER_TIMEOUT falls back to default", func(t *testing.T) {
|
||||
t.Setenv("A2A_PROXY_RESPONSE_HEADER_TIMEOUT", "not-a-duration")
|
||||
// Simulate what envx.Duration does with an invalid value.
|
||||
var fallback = 180 * time.Second
|
||||
override := fallback
|
||||
if v := os.Getenv("A2A_PROXY_RESPONSE_HEADER_TIMEOUT"); v != "" {
|
||||
if d, err := time.ParseDuration(v); err == nil && d > 0 {
|
||||
override = d
|
||||
}
|
||||
}
|
||||
if override != fallback {
|
||||
t.Errorf("invalid env var: got %v, want fallback %v", override, fallback)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
@@ -28,7 +28,7 @@ WORKSPACE_ID = _WORKSPACE_ID_raw
|
||||
if os.path.exists("/.dockerenv") or os.environ.get("DOCKER_VERSION"):
|
||||
PLATFORM_URL = os.environ.get("PLATFORM_URL", "http://host.docker.internal:8080")
|
||||
else:
|
||||
PLATFORM_URL = os.environ.get("PLATFORM_URL", "http://host.docker.internal:8080")
|
||||
PLATFORM_URL = os.environ.get("PLATFORM_URL", "http://localhost:8080")
|
||||
|
||||
|
||||
async def discover(target_id: str) -> dict | None:
|
||||
|
||||
@@ -29,7 +29,7 @@ WORKSPACE_ID = _WORKSPACE_ID_raw
|
||||
if os.path.exists("/.dockerenv") or os.environ.get("DOCKER_VERSION"):
|
||||
PLATFORM_URL = os.environ.get("PLATFORM_URL", "http://host.docker.internal:8080")
|
||||
else:
|
||||
PLATFORM_URL = os.environ.get("PLATFORM_URL", "http://host.docker.internal:8080")
|
||||
PLATFORM_URL = os.environ.get("PLATFORM_URL", "http://localhost:8080")
|
||||
|
||||
# Cache workspace ID → name mappings (populated by list_peers calls)
|
||||
_peer_names: dict[str, str] = {}
|
||||
|
||||
@@ -54,16 +54,6 @@ import httpx
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def _platform_url() -> str:
|
||||
"""Return the platform URL, defaulting to host.docker.internal when running
|
||||
inside a Docker container (where localhost refers to the container, not the
|
||||
host). External callers can always override via the PLATFORM_URL env var.
|
||||
"""
|
||||
if os.path.exists("/.dockerenv") or os.environ.get("DOCKER_VERSION"):
|
||||
return os.environ.get("PLATFORM_URL", "http://host.docker.internal:8080")
|
||||
return os.environ.get("PLATFORM_URL", "http://host.docker.internal:8080")
|
||||
|
||||
# ─────────────────────────────────────────────────────────────────────────────
|
||||
# Constants
|
||||
# ─────────────────────────────────────────────────────────────────────────────
|
||||
@@ -89,12 +79,12 @@ async def _fetch_latest_checkpoint(workspace_id: str) -> Optional[dict]:
|
||||
workspace_id: The workspace to query.
|
||||
|
||||
Reads:
|
||||
PLATFORM_URL Platform base URL (default ``http://host.docker.internal:8080``).
|
||||
PLATFORM_URL Platform base URL (default ``http://localhost:8080``).
|
||||
"""
|
||||
try:
|
||||
from platform_auth import auth_headers as _auth_headers # type: ignore[import]
|
||||
|
||||
platform_url = _platform_url()
|
||||
platform_url = os.environ.get("PLATFORM_URL", "http://localhost:8080")
|
||||
url = f"{platform_url}/workspaces/{workspace_id}/checkpoints/latest"
|
||||
async with httpx.AsyncClient(timeout=5.0) as client:
|
||||
resp = await client.get(url, headers=_auth_headers())
|
||||
@@ -135,12 +125,12 @@ async def _save_checkpoint(
|
||||
payload: Optional JSON-serialisable dict stored as JSONB.
|
||||
|
||||
Reads:
|
||||
PLATFORM_URL Platform base URL (default ``http://host.docker.internal:8080``).
|
||||
PLATFORM_URL Platform base URL (default ``http://localhost:8080``).
|
||||
"""
|
||||
try:
|
||||
from platform_auth import auth_headers as _auth_headers # type: ignore[import]
|
||||
|
||||
platform_url = _platform_url()
|
||||
platform_url = os.environ.get("PLATFORM_URL", "http://localhost:8080")
|
||||
url = f"{platform_url}/workspaces/{workspace_id}/checkpoints"
|
||||
body: dict = {
|
||||
"workflow_id": workflow_id,
|
||||
|
||||
@@ -21,7 +21,7 @@ logger = logging.getLogger(__name__)
|
||||
if os.path.exists("/.dockerenv") or os.environ.get("DOCKER_VERSION"):
|
||||
PLATFORM_URL = os.environ.get("PLATFORM_URL", "http://host.docker.internal:8080")
|
||||
else:
|
||||
PLATFORM_URL = os.environ.get("PLATFORM_URL", "http://host.docker.internal:8080")
|
||||
PLATFORM_URL = os.environ.get("PLATFORM_URL", "http://localhost:8080")
|
||||
_WORKSPACE_ID_raw = os.environ.get("WORKSPACE_ID")
|
||||
if not _WORKSPACE_ID_raw:
|
||||
raise RuntimeError("WORKSPACE_ID environment variable is required but not set")
|
||||
|
||||
@@ -25,7 +25,7 @@ logger = logging.getLogger(__name__)
|
||||
if os.path.exists("/.dockerenv") or os.environ.get("DOCKER_VERSION"):
|
||||
PLATFORM_URL = os.environ.get("PLATFORM_URL", "http://host.docker.internal:8080")
|
||||
else:
|
||||
PLATFORM_URL = os.environ.get("PLATFORM_URL", "http://host.docker.internal:8080")
|
||||
PLATFORM_URL = os.environ.get("PLATFORM_URL", "http://localhost:8080")
|
||||
_WORKSPACE_ID_raw = os.environ.get("WORKSPACE_ID")
|
||||
if not _WORKSPACE_ID_raw:
|
||||
raise RuntimeError("WORKSPACE_ID environment variable is required but not set")
|
||||
|
||||
+1
-1
@@ -63,7 +63,7 @@ async def main(): # pragma: no cover
|
||||
if os.path.exists("/.dockerenv") or os.environ.get("DOCKER_VERSION"):
|
||||
platform_url = os.environ.get("PLATFORM_URL", "http://host.docker.internal:8080")
|
||||
else:
|
||||
platform_url = os.environ.get("PLATFORM_URL", "http://host.docker.internal:8080")
|
||||
platform_url = os.environ.get("PLATFORM_URL", "http://localhost:8080")
|
||||
awareness_config = get_awareness_config()
|
||||
|
||||
# 0. Initialise OpenTelemetry (no-op if packages not installed)
|
||||
|
||||
@@ -51,32 +51,6 @@ class AdaptorSource:
|
||||
|
||||
def _load_module_from_path(module_name: str, path: Path):
|
||||
"""Import a Python file by absolute path. Returns the module or None on failure."""
|
||||
|
||||
# KI-296: Before exec'ing plugin-adapter files (which import
|
||||
# ``from plugins_registry import ...`` as a top-level name), register
|
||||
# the molecule-runtime subpackage as ``plugins_registry`` in sys.modules.
|
||||
# In the molecule-core workspace source this is already a top-level package,
|
||||
# so the setdefault is a no-op. In the PyPI-installed runtime wheel
|
||||
# (molecule-ai-workspace-runtime 0.1.129+), the package ships as
|
||||
# ``molecule_runtime.plugins_registry`` and without this shim every
|
||||
# plugin adapter would fail with ModuleNotFoundError.
|
||||
import sys as _sys
|
||||
|
||||
if "plugins_registry" not in _sys.modules:
|
||||
try:
|
||||
_mr_pr = __import__("molecule_runtime.plugins_registry", fromlist=[""])
|
||||
_sys.modules["plugins_registry"] = _mr_pr
|
||||
# Also register submodules the adapters commonly import directly.
|
||||
for _sub in ("builtins", "protocol", "raw_drop"):
|
||||
_submod = getattr(_mr_pr, _sub, None)
|
||||
if _submod is not None:
|
||||
_sys.modules[f"plugins_registry.{_sub}"] = _submod
|
||||
except ImportError:
|
||||
# molecule-runtime not installed (e.g. test environment with
|
||||
# workspace/ on sys.path directly) — skip shim; the top-level
|
||||
# workspace/plugins_registry package is already findable.
|
||||
pass
|
||||
|
||||
spec = importlib.util.spec_from_file_location(module_name, path)
|
||||
if spec is None or spec.loader is None:
|
||||
return None
|
||||
|
||||
@@ -325,44 +325,3 @@ def test_resolve_registry_missing_module_falls_through(monkeypatch, tmp_path: Pa
|
||||
monkeypatch.setattr(pr, "_REGISTRY_ROOT", tmp_path / "empty-registry")
|
||||
_, source = pr.resolve("demo-plugin", "test_runtime", plugin_root)
|
||||
assert source == AdaptorSource.RAW_DROP
|
||||
|
||||
|
||||
def test_load_module_from_path_registers_plugins_registry_sys_modules(tmp_path: Path):
|
||||
"""KI-296: _load_module_from_path registers ``plugins_registry`` in sys.modules
|
||||
before exec'ing the adapter, so adapter files that do
|
||||
``from plugins_registry import ...`` resolve correctly when the runtime is
|
||||
installed from the PyPI wheel (where the package ships as
|
||||
``molecule_runtime.plugins_registry`` rather than a top-level ``plugins_registry``).
|
||||
"""
|
||||
import sys as _sys
|
||||
import plugins_registry as pr
|
||||
|
||||
# Create a fake adapter that imports plugins_registry at top level.
|
||||
adapter_file = tmp_path / "fake_runtime_adapter.py"
|
||||
adapter_file.write_text(
|
||||
"from plugins_registry import InstallContext # noqa: F401\n"
|
||||
"from plugins_registry.builtins import AgentskillsAdaptor as Adaptor # noqa: F401\n"
|
||||
)
|
||||
|
||||
# Evict any pre-existing sys.modules entries for the shim keys so the
|
||||
# import inside _load_module_from_path actually runs.
|
||||
_saved = {
|
||||
k: _sys.modules.pop(k, None)
|
||||
for k in (
|
||||
"plugins_registry", "plugins_registry.builtins",
|
||||
"plugins_registry.protocol", "plugins_registry.raw_drop",
|
||||
"_plugin_adaptor.test.fake_runtime",
|
||||
)
|
||||
}
|
||||
|
||||
try:
|
||||
result = pr._load_module_from_path("_plugin_adaptor.test.fake_runtime", adapter_file)
|
||||
assert result is not None, "module should load without ImportError"
|
||||
assert hasattr(result, "Adaptor"), "AgentskillsAdaptor alias should be in namespace"
|
||||
finally:
|
||||
# Restore sys.modules state.
|
||||
for k, v in _saved.items():
|
||||
if v is None:
|
||||
_sys.modules.pop(k, None)
|
||||
else:
|
||||
_sys.modules[k] = v
|
||||
|
||||
Reference in New Issue
Block a user