(refactor) Finish controller tool and base node updates

- Update benchmark tools with improved submission validation
- Update base node with stronger handover contract enforcement
- Add agent tools for manifest-aware submission routing

Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com>

Author: MRiabov

controller/agent/benchmark/tools.py

@@ -6,6 +6,7 @@
 from pathlib import Path
 
 import yaml
+from sqlalchemy import select
 
 from controller.agent.tools import (
     _invoke_cots_search_subagent,
@@ -15,6 +16,9 @@
     run_validate_and_price_script,
 )
 from controller.middleware.remote_fs import RemoteFilesystemMiddleware
+from controller.persistence.db import get_sessionmaker
+from controller.persistence.models import Asset
+from controller.utils import resolve_episode_id
 from shared.agents.config import DraftingMode, load_agents_config
 from shared.enums import AgentName
 from shared.git_utils import repo_revision
@@ -262,16 +266,37 @@ async def submit_plan() -> dict:
         artifacts: dict[str, str] = {}
         missing_files: list[str] = []
 
-        for rel_path in required_files:
+        async def _read_required_text(rel_path: str) -> str | None:
             content = await fs.client.read_file_optional(
                 rel_path, bypass_agent_permissions=True
             )
+            if isinstance(content, str) and content.strip():
+                return content
+
+            try:
+                episode_uuid = resolve_episode_id(fs.episode_id)
+            except Exception:
+                return None
+
+            session_factory = get_sessionmaker()
+            async with session_factory() as db:
+                result = await db.execute(
+                    select(Asset.content).where(
+                        Asset.episode_id == episode_uuid,
+                        Asset.s3_path == rel_path,
+                    )
+                )
+                persisted_content = result.scalar_one_or_none()
+
+            if isinstance(persisted_content, str) and persisted_content.strip():
+                return persisted_content
+            return None
+
+        for rel_path in required_files:
+            content = await _read_required_text(rel_path)
             if content is None:
                 missing_files.append(rel_path)
                 continue
-            if not content.strip():
-                missing_files.append(rel_path)
-                continue
             artifacts[rel_path] = content
 
         if missing_files:
@@ -283,8 +308,8 @@ async def submit_plan() -> dict:
             )
             return result.model_dump(mode="json")
 
-        manufacturing_config_text = await fs.client.read_file_optional(
-            "manufacturing_config.yaml", bypass_agent_permissions=True
+        manufacturing_config_text = await _read_required_text(
+            "manufacturing_config.yaml"
         )
         if manufacturing_config_text is None:
             result = PlannerSubmissionResult(

controller/agent/nodes/base.py

@@ -45,8 +45,9 @@
     report_usage_to_current_observation,
 )
 from controller.observability.tracing import record_worker_events
-from controller.persistence.models import Trace
-from controller.utils import EpisodeIdentity
+from controller.persistence.db import get_sessionmaker
+from controller.persistence.models import Asset, Trace
+from controller.utils import EpisodeIdentity, resolve_episode_id
 from shared.agents.config import load_agents_config
 from shared.enums import AgentName, TraceType
 from shared.git_utils import repo_revision
@@ -2544,14 +2545,42 @@ async def stream_lm_history_live(start_idx: int | None) -> None:
                             output_obj=prediction,
                         )
 
+                    async def _read_validation_artifact(path: str) -> str | None:
+                        content = await self.ctx.fs.read_file_optional(
+                            path, bypass_agent_permissions=True
+                        )
+                        if isinstance(content, str) and content.strip():
+                            return content
+
+                        try:
+                            episode_uuid = resolve_episode_id(episode_id)
+                        except Exception:
+                            return None
+
+                        session_factory = get_sessionmaker()
+                        async with session_factory() as db:
+                            result = await db.execute(
+                                select(Asset.content).where(
+                                    Asset.episode_id == episode_uuid,
+                                    Asset.s3_path == path,
+                                )
+                            )
+                            persisted_content = result.scalar_one_or_none()
+
+                        if (
+                            isinstance(persisted_content, str)
+                            and persisted_content.strip()
+                        ):
+                            return persisted_content
+                        return None
+
                     results = await asyncio.gather(
-                        *[self.ctx.fs.read_file(f) for f in validate_files],
-                        return_exceptions=True,
+                        *[_read_validation_artifact(f) for f in validate_files]
                     )
                     artifacts = {
                         f: res
                         for f, res in zip(validate_files, results, strict=False)
-                        if not isinstance(res, Exception)
+                        if res
                     }
 
                     is_valid, validation_errors = validate_node_output(

controller/agent/tools.py

@@ -49,7 +49,11 @@ def _tool_name(tool: Callable) -> str:
 
 
 def _runtime_skill_script_path(*relative_parts: str) -> Path:
-    return Path(__file__).resolve().parents[2].joinpath(*relative_parts)
+    repo_root = Path(__file__).resolve().parents[2]
+    skill_root = repo_root / ".agents" / "skills"
+    if skill_root.exists():
+        return skill_root.joinpath(*relative_parts)
+    return repo_root.joinpath(*relative_parts)
 
 
 def _engineer_planner_drafting_required() -> bool: