laoxong/AstrBot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: add get_sandbox_capabilities API and structured logging to computer_client

Browse Source
This commit is contained in:
zenfun
2026-03-11 03:00:52 +08:00
parent e1d76117b4
commit a5a1ba72fd
1 changed files with 81 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files
astrbot/core/computer/computer_client.py
+81 -31
View File
@@ -78,22 +78,25 @@ def _discover_bay_credentials(endpoint: str) -> str:
and cred_endpoint.rstrip("/") != endpoint.rstrip("/")
):
logger.warning(
"[Computer] credentials.json endpoint mismatch: "
"file=%s, configured=%s — using key anyway",
"[Computer] bay_credentials_mismatch file_endpoint=%s configured_endpoint=%s action=use_key",
cred_endpoint,
endpoint,
)
masked_key = f"{api_key[:4]}..." if len(api_key) >= 6 else "redacted"
logger.info(
"[Computer] Auto-discovered Bay API key from %s (prefix=%s)",
"[Computer] bay_credentials_lookup status=found path=%s key_prefix=%s",
cred_path,
masked_key,
)
return api_key
except (json.JSONDecodeError, OSError) as exc:
logger.debug("[Computer] Failed to read %s: %s", cred_path, exc)
logger.debug(
"[Computer] bay_credentials_read_failed path=%s error=%s",
cred_path,
exc,
)
logger.debug("[Computer] No Bay credentials.json found in search paths")
logger.debug("[Computer] bay_credentials_lookup status=not_found")
return ""
@@ -346,29 +349,33 @@ async def _apply_skills_to_sandbox(booter: ComputerBooter) -> None:
This function is intentionally limited to file mutation. Metadata scanning is
executed in a separate phase to keep failure domains clear.
"""
logger.info("[Computer] Skill sync phase=apply start")
logger.info("[Computer] sandbox_sync phase=apply status=start")
apply_result = await booter.shell.exec(_build_apply_sync_command())
if not _shell_exec_succeeded(apply_result):
detail = _format_exec_error_detail(apply_result)
logger.error("[Computer] Skill sync phase=apply failed: %s", detail)
logger.error(
"[Computer] sandbox_sync phase=apply status=failed detail=%s", detail
)
raise RuntimeError(f"Failed to apply sandbox skill sync strategy: {detail}")
logger.info("[Computer] Skill sync phase=apply done")
logger.info("[Computer] sandbox_sync phase=apply status=done")
async def _scan_sandbox_skills(booter: ComputerBooter) -> dict | None:
"""Scan sandbox skills and return normalized payload for cache update."""
logger.info("[Computer] Skill sync phase=scan start")
logger.info("[Computer] sandbox_sync phase=scan status=start")
scan_result = await booter.shell.exec(_build_scan_command())
if not _shell_exec_succeeded(scan_result):
detail = _format_exec_error_detail(scan_result)
logger.error("[Computer] Skill sync phase=scan failed: %s", detail)
logger.error(
"[Computer] sandbox_sync phase=scan status=failed detail=%s", detail
)
raise RuntimeError(f"Failed to scan sandbox skills after sync: {detail}")
payload = _decode_sync_payload(str(scan_result.get("stdout", "") or ""))
if payload is None:
logger.warning("[Computer] Skill sync phase=scan returned empty payload")
logger.warning("[Computer] sandbox_sync phase=scan status=empty_payload")
else:
logger.info("[Computer] Skill sync phase=scan done")
logger.info("[Computer] sandbox_sync phase=scan status=done")
return payload
@@ -394,14 +401,16 @@ async def _sync_skills_to_sandbox(booter: ComputerBooter) -> None:
zip_path.unlink()
shutil.make_archive(str(zip_base), "zip", str(skills_root))
remote_zip = Path(SANDBOX_SKILLS_ROOT) / "skills.zip"
logger.info("Uploading skills bundle to sandbox...")
logger.info("[Computer] sandbox_sync phase=upload status=start")
await booter.shell.exec(f"mkdir -p {SANDBOX_SKILLS_ROOT}")
upload_result = await booter.upload_file(str(zip_path), str(remote_zip))
if not upload_result.get("success", False):
logger.error("[Computer] sandbox_sync phase=upload status=failed")
raise RuntimeError("Failed to upload skills bundle to sandbox.")
logger.info("[Computer] sandbox_sync phase=upload status=done")
else:
logger.info(
"No local skills found. Keeping sandbox built-ins and refreshing metadata."
"[Computer] sandbox_sync phase=upload status=skipped reason=no_local_skills"
)
await booter.shell.exec(f"rm -f {SANDBOX_SKILLS_ROOT}/skills.zip")
@@ -412,7 +421,7 @@ async def _sync_skills_to_sandbox(booter: ComputerBooter) -> None:
_update_sandbox_skills_cache(payload)
managed = payload.get("managed_skills", []) if isinstance(payload, dict) else []
logger.info(
"[Computer] Sandbox skill sync complete: managed=%d",
"[Computer] sandbox_sync phase=overall status=done managed=%d",
len(managed),
)
finally:
@@ -420,7 +429,10 @@ async def _sync_skills_to_sandbox(booter: ComputerBooter) -> None:
try:
zip_path.unlink()
except Exception:
logger.warning(f"Failed to remove temp skills zip: {zip_path}")
logger.warning(
"[Computer] sandbox_sync phase=cleanup status=failed path=%s",
zip_path,
)
async def get_booter(
@@ -440,7 +452,9 @@ async def get_booter(
if session_id not in session_booter:
uuid_str = uuid.uuid5(uuid.NAMESPACE_DNS, session_id).hex
logger.info(
f"[Computer] Initializing booter: type={booter_type}, session={session_id}"
"[Computer] booter_init booter=%s session=%s",
booter_type,
session_id,
)
if booter_type == "shipyard":
from .booters.shipyard import ShipyardBooter
@@ -484,12 +498,18 @@ async def get_booter(
try:
await client.boot(uuid_str)
logger.info(
f"[Computer] Sandbox booted successfully: type={booter_type}, session={session_id}"
"[Computer] booter_ready booter=%s session=%s",
booter_type,
session_id,
)
await _sync_skills_to_sandbox(client)
except Exception as e:
logger.error(f"Error booting sandbox for session {session_id}: {e}")
raise e
except Exception:
logger.exception(
"[Computer] booter_init_failed booter=%s session=%s",
booter_type,
session_id,
)
raise
session_booter[session_id] = client
return session_booter[session_id]
@@ -498,18 +518,19 @@ async def get_booter(
async def sync_skills_to_active_sandboxes() -> None:
"""Best-effort skills synchronization for all active sandbox sessions."""
logger.info(
"[Computer] Syncing skills to %d active sandbox(es)", len(session_booter)
"[Computer] sandbox_sync scope=active sessions=%d",
len(session_booter),
)
for session_id, booter in list(session_booter.items()):
try:
if not await booter.available():
continue
await _sync_skills_to_sandbox(booter)
except Exception as e:
logger.warning(
"Failed to sync skills to sandbox for session %s: %s",
except Exception:
logger.exception(
"[Computer] sandbox_sync_failed session=%s booter=%s",
session_id,
e,
booter.__class__.__name__,
)
@@ -539,7 +560,10 @@ def _get_booter_class(booter_type: str) -> type[ComputerBooter] | None:
from .booters.boxlite import BoxliteBooter
return BoxliteBooter
logger.warning("[Computer] booter_class_lookup booter=%s found=false", booter_type)
logger.warning(
"[Computer] booter_class_lookup booter=%s found=false",
booter_type,
)
return None
@@ -547,10 +571,19 @@ def get_sandbox_tools(session_id: str) -> list[FunctionTool]:
"""Return precise tool list from a booted session, or [] if not booted."""
booter = session_booter.get(session_id)
if booter is None:
logger.debug(
"[Computer] sandbox_tools source=booted session=%s booter=none tools=0 capabilities=none",
session_id,
)
return []
tools = booter.get_tools()
caps = getattr(booter, "capabilities", None)
logger.debug(
"[Computer] get_sandbox_tools: session=%s, tools=%d", session_id, len(tools)
"[Computer] sandbox_tools source=booted session=%s booter=%s tools=%d capabilities=%s",
session_id,
booter.__class__.__name__,
len(tools),
list(caps) if caps is not None else None,
)
return tools
@@ -559,8 +592,19 @@ def get_sandbox_capabilities(session_id: str) -> tuple[str, ...] | None:
"""Return capability tuple from a booted session, or None if unavailable."""
booter = session_booter.get(session_id)
if booter is None:
logger.debug(
"[Computer] sandbox_capabilities session=%s booter=none capabilities=none",
session_id,
)
return None
return getattr(booter, "capabilities", None)
caps = getattr(booter, "capabilities", None)
logger.debug(
"[Computer] sandbox_capabilities session=%s booter=%s capabilities=%s",
session_id,
booter.__class__.__name__,
list(caps) if caps is not None else None,
)
return caps
def get_default_sandbox_tools(sandbox_cfg: dict) -> list[FunctionTool]:
@@ -569,7 +613,7 @@ def get_default_sandbox_tools(sandbox_cfg: dict) -> list[FunctionTool]:
cls = _get_booter_class(booter_type)
tools = cls.get_default_tools() if cls else []
logger.debug(
"[Computer] get_default_sandbox_tools: booter=%s, tools=%d",
"[Computer] sandbox_tools source=default booter=%s tools=%d capabilities=unknown",
booter_type,
len(tools),
)
@@ -580,4 +624,10 @@ def get_sandbox_prompt_parts(sandbox_cfg: dict) -> list[str]:
"""Return booter-specific system prompt fragments based on config."""
booter_type = sandbox_cfg.get("booter", BOOTER_SHIPYARD_NEO)
cls = _get_booter_class(booter_type)
return cls.get_system_prompt_parts() if cls else []
prompt_parts = cls.get_system_prompt_parts() if cls else []
logger.debug(
"[Computer] sandbox_prompts booter=%s parts=%d",
booter_type,
len(prompt_parts),
)
return prompt_parts