mirror of
https://github.com/HKUDS/OpenSpace.git
synced 2026-08-28 05:15:00 +00:00
fix: pass max_iterations in no-skill path and unique workflow IDs
- Forward resolved max_iterations to grounding agent in no-skill execution path - Use hash-based workflow ID to prevent collision across roots and separator ambiguity Co-authored-by: wul48527-code <wul48527-code@users.noreply.github.com>
This commit is contained in:
commit
2fb8024ff6
4 changed files with 175 additions and 4 deletions
|
|
@ -419,6 +419,18 @@ def _build_lineage_payload(skill_id: str, store: SkillStore) -> Dict[str, Any]:
|
|||
}
|
||||
|
||||
|
||||
def _workflow_id(workflow_dir: Path) -> str:
|
||||
"""Stable short ID for a workflow directory, unique across roots.
|
||||
|
||||
Uses a hash suffix derived from the resolved path to avoid collisions
|
||||
when directory names contain the separator character.
|
||||
"""
|
||||
import hashlib
|
||||
resolved = str(workflow_dir.resolve())
|
||||
path_hash = hashlib.sha256(resolved.encode()).hexdigest()[:8]
|
||||
return f"{workflow_dir.name}_{path_hash}"
|
||||
|
||||
|
||||
def _discover_workflow_dirs() -> List[Path]:
|
||||
discovered: Dict[str, Path] = {}
|
||||
for root in WORKFLOW_ROOTS:
|
||||
|
|
@ -439,14 +451,14 @@ def _scan_workflow_tree(directory: Path, discovered: Dict[str, Path], *, _depth:
|
|||
if not child.is_dir():
|
||||
continue
|
||||
if (child / "metadata.json").exists() or (child / "traj.jsonl").exists():
|
||||
discovered.setdefault(child.name, child)
|
||||
discovered.setdefault(str(child.resolve()), child)
|
||||
else:
|
||||
_scan_workflow_tree(child, discovered, _depth=_depth + 1, _max_depth=_max_depth)
|
||||
|
||||
|
||||
def _get_workflow_dir(workflow_id: str) -> Optional[Path]:
|
||||
for path in _discover_workflow_dirs():
|
||||
if path.name == workflow_id:
|
||||
if _workflow_id(path) == workflow_id:
|
||||
return path
|
||||
return None
|
||||
|
||||
|
|
@ -464,7 +476,7 @@ def _build_workflow_summary(workflow_dir: Path) -> Dict[str, Any]:
|
|||
for candidate in video_candidates:
|
||||
if candidate.exists():
|
||||
rel = candidate.relative_to(workflow_dir).as_posix()
|
||||
video_url = url_for("workflow_artifact", workflow_id=workflow_dir.name, artifact_path=rel)
|
||||
video_url = url_for("workflow_artifact", workflow_id=_workflow_id(workflow_dir), artifact_path=rel)
|
||||
break
|
||||
|
||||
outcome = metadata.get("execution_outcome") or {}
|
||||
|
|
@ -514,7 +526,7 @@ def _build_workflow_summary(workflow_dir: Path) -> Dict[str, Any]:
|
|||
iterations = len(trajectory)
|
||||
|
||||
return {
|
||||
"id": workflow_dir.name,
|
||||
"id": _workflow_id(workflow_dir),
|
||||
"path": str(workflow_dir),
|
||||
"task_id": metadata.get("task_id") or metadata.get("task_name") or workflow_dir.name,
|
||||
"task_name": metadata.get("task_name") or metadata.get("task_id") or workflow_dir.name,
|
||||
|
|
|
|||
|
|
@ -507,6 +507,7 @@ class OpenSpace:
|
|||
f"Executing with GroundingAgent "
|
||||
f"(max {max_iterations} iterations, no skills)..."
|
||||
)
|
||||
execution_context["max_iterations"] = max_iterations
|
||||
result = await self._grounding_agent.process(execution_context)
|
||||
|
||||
execution_time = asyncio.get_event_loop().time() - start_time
|
||||
|
|
|
|||
64
tests/test_max_iterations.py
Normal file
64
tests/test_max_iterations.py
Normal file
|
|
@ -0,0 +1,64 @@
|
|||
"""Test that max_iterations is passed to grounding agent in no-skill path."""
|
||||
|
||||
import asyncio
|
||||
import types
|
||||
from unittest.mock import AsyncMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from openspace.tool_layer import OpenSpace, OpenSpaceConfig
|
||||
|
||||
|
||||
def _make_openspace(configured_max: int = 50) -> OpenSpace:
|
||||
config = OpenSpaceConfig()
|
||||
config.grounding_max_iterations = configured_max
|
||||
os_inst = OpenSpace(config)
|
||||
os_inst._initialized = True
|
||||
os_inst._running = False
|
||||
os_inst._task_done = asyncio.Event()
|
||||
os_inst._task_done.set()
|
||||
os_inst._grounding_client = types.SimpleNamespace(_registry={})
|
||||
os_inst._recording_manager = None
|
||||
os_inst._skill_registry = None
|
||||
os_inst._execution_analyzer = None
|
||||
os_inst._skill_evolver = None
|
||||
return os_inst
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_no_skill_path_passes_configured_max_iterations():
|
||||
"""When no skills match, the configured grounding_max_iterations
|
||||
must be forwarded to the agent, not silently dropped."""
|
||||
os_inst = _make_openspace(configured_max=50)
|
||||
recorded = {}
|
||||
|
||||
async def fake_process(context):
|
||||
recorded.update(context)
|
||||
return {"status": "success", "iterations": 1, "tool_executions": []}
|
||||
|
||||
os_inst._grounding_agent = types.SimpleNamespace(process=fake_process)
|
||||
os_inst._maybe_analyze_execution = AsyncMock()
|
||||
os_inst._maybe_evolve_quality = AsyncMock()
|
||||
|
||||
await os_inst.execute("do something")
|
||||
|
||||
assert recorded.get("max_iterations") == 50
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_no_skill_path_uses_caller_override_when_larger():
|
||||
"""Caller-provided max_iterations should win when larger than config."""
|
||||
os_inst = _make_openspace(configured_max=20)
|
||||
recorded = {}
|
||||
|
||||
async def fake_process(context):
|
||||
recorded.update(context)
|
||||
return {"status": "success", "iterations": 1, "tool_executions": []}
|
||||
|
||||
os_inst._grounding_agent = types.SimpleNamespace(process=fake_process)
|
||||
os_inst._maybe_analyze_execution = AsyncMock()
|
||||
os_inst._maybe_evolve_quality = AsyncMock()
|
||||
|
||||
await os_inst.execute("do something", max_iterations=100)
|
||||
|
||||
assert recorded.get("max_iterations") == 100
|
||||
94
tests/test_workflow_id.py
Normal file
94
tests/test_workflow_id.py
Normal file
|
|
@ -0,0 +1,94 @@
|
|||
"""Tests for workflow ID uniqueness and discovery."""
|
||||
|
||||
from pathlib import Path
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
|
||||
from openspace import dashboard_server
|
||||
|
||||
|
||||
def _make_workflow(path: Path):
|
||||
"""Create a minimal workflow directory with metadata.json."""
|
||||
path.mkdir(parents=True, exist_ok=True)
|
||||
(path / "metadata.json").write_text("{}", encoding="utf-8")
|
||||
|
||||
|
||||
class TestWorkflowId:
|
||||
def test_same_leaf_name_different_roots_get_unique_ids(self, tmp_path):
|
||||
root_a = tmp_path / "root_a"
|
||||
root_b = tmp_path / "root_b"
|
||||
wf_a = root_a / "task1"
|
||||
wf_b = root_b / "task1"
|
||||
_make_workflow(wf_a)
|
||||
_make_workflow(wf_b)
|
||||
|
||||
with patch.object(dashboard_server, "WORKFLOW_ROOTS", [root_a, root_b]):
|
||||
id_a = dashboard_server._workflow_id(wf_a)
|
||||
id_b = dashboard_server._workflow_id(wf_b)
|
||||
|
||||
assert id_a != id_b
|
||||
assert id_a.startswith("task1_")
|
||||
assert id_b.startswith("task1_")
|
||||
|
||||
def test_id_contains_dir_name_and_hash(self, tmp_path):
|
||||
root = tmp_path / "root"
|
||||
wf = root / "my-task"
|
||||
_make_workflow(wf)
|
||||
|
||||
with patch.object(dashboard_server, "WORKFLOW_ROOTS", [root]):
|
||||
wf_id = dashboard_server._workflow_id(wf)
|
||||
|
||||
assert wf_id.startswith("my-task_")
|
||||
assert len(wf_id) == len("my-task_") + 8 # 8-char hex hash
|
||||
|
||||
def test_separator_collision_produces_different_ids(self, tmp_path):
|
||||
"""Regression: a dir named 'a__b' must not collide with path a/b."""
|
||||
root = tmp_path / "root"
|
||||
flat = root / "a__b"
|
||||
nested = root / "a" / "b"
|
||||
_make_workflow(flat)
|
||||
_make_workflow(nested)
|
||||
|
||||
with patch.object(dashboard_server, "WORKFLOW_ROOTS", [root]):
|
||||
id_flat = dashboard_server._workflow_id(flat)
|
||||
id_nested = dashboard_server._workflow_id(nested)
|
||||
|
||||
assert id_flat != id_nested
|
||||
|
||||
def test_workflow_outside_roots_still_works(self, tmp_path):
|
||||
wf = tmp_path / "orphan_workflow"
|
||||
_make_workflow(wf)
|
||||
|
||||
with patch.object(dashboard_server, "WORKFLOW_ROOTS", []):
|
||||
wf_id = dashboard_server._workflow_id(wf)
|
||||
|
||||
assert wf_id.startswith("orphan_workflow_")
|
||||
|
||||
|
||||
class TestDiscoverWorkflowDirs:
|
||||
def test_same_name_workflows_both_discovered(self, tmp_path):
|
||||
root_a = tmp_path / "root_a"
|
||||
root_b = tmp_path / "root_b"
|
||||
_make_workflow(root_a / "task1")
|
||||
_make_workflow(root_b / "task1")
|
||||
|
||||
with patch.object(dashboard_server, "WORKFLOW_ROOTS", [root_a, root_b]):
|
||||
dirs = dashboard_server._discover_workflow_dirs()
|
||||
|
||||
assert len(dirs) == 2
|
||||
|
||||
def test_get_workflow_dir_resolves_correct_path(self, tmp_path):
|
||||
root_a = tmp_path / "root_a"
|
||||
root_b = tmp_path / "root_b"
|
||||
_make_workflow(root_a / "task1")
|
||||
_make_workflow(root_b / "task1")
|
||||
|
||||
with patch.object(dashboard_server, "WORKFLOW_ROOTS", [root_a, root_b]):
|
||||
id_a = dashboard_server._workflow_id(root_a / "task1")
|
||||
id_b = dashboard_server._workflow_id(root_b / "task1")
|
||||
found_a = dashboard_server._get_workflow_dir(id_a)
|
||||
found_b = dashboard_server._get_workflow_dir(id_b)
|
||||
|
||||
assert found_a.resolve() == (root_a / "task1").resolve()
|
||||
assert found_b.resolve() == (root_b / "task1").resolve()
|
||||
Loading…
Add table
Reference in a new issue