mirror of
https://github.com/rookiestar28/ComfyUI-OpenClaw.git
synced 2026-08-14 00:48:07 +00:00
446 lines
15 KiB
Python
446 lines
15 KiB
Python
"""Owned observability and jobs handler implementations for the API facade."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import time
|
|
from collections.abc import Callable
|
|
from contextlib import suppress
|
|
from dataclasses import dataclass
|
|
from typing import Any
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class RouteHandlerDependencies:
|
|
web: Any
|
|
pack_name: Any
|
|
pack_version: Any
|
|
pack_start_time: Any
|
|
log_file: Any
|
|
metrics: Any
|
|
tail_log: Any
|
|
require_observability_access: Any
|
|
require_admin_token: Any
|
|
check_rate_limit: Any
|
|
build_rate_limit_response: Any
|
|
trace_store: Any
|
|
get_executor_diagnostics: Any
|
|
redact_text: Any
|
|
check_dependency: Callable[[str], bool]
|
|
resolve_token_info: Any
|
|
emit_audit_event: Any
|
|
jobs_request_tenant_scope: Any
|
|
normalize_jobs_query: Any
|
|
build_jobs_audit_details: Any
|
|
safe_job_audit_outcomes: Any
|
|
jobs_security_error: Any
|
|
tenant_boundary_error: Any
|
|
jobs_host_contract_unsupported: Any
|
|
jobs_backend_unavailable: Any
|
|
read_jobs: Any
|
|
ensure_observability_deps_ready: Any
|
|
|
|
|
|
def ensure_observability_deps_ready(
|
|
deps: RouteHandlerDependencies,
|
|
) -> tuple[bool, str | None]:
|
|
"""Reject partially initialized observability handlers deterministically."""
|
|
|
|
missing: list[str] = []
|
|
if not callable(deps.require_observability_access):
|
|
missing.append("require_observability_access")
|
|
if not callable(deps.check_rate_limit):
|
|
missing.append("check_rate_limit")
|
|
if not callable(deps.tail_log):
|
|
missing.append("tail_log")
|
|
if missing:
|
|
return (
|
|
False,
|
|
"Backend not fully initialized (missing route dependencies: "
|
|
+ ", ".join(missing)
|
|
+ ").",
|
|
)
|
|
return True, None
|
|
|
|
|
|
async def health_response(request: Any, deps: RouteHandlerDependencies) -> Any:
|
|
"""Build the existing partial-failure-tolerant health response."""
|
|
|
|
if deps.web is None:
|
|
raise RuntimeError("aiohttp not available")
|
|
try:
|
|
from ..services.llm_client import LLMClient
|
|
from ..services.providers.keys import requires_api_key
|
|
except ImportError:
|
|
from services.llm_client import LLMClient
|
|
from services.providers.keys import requires_api_key
|
|
|
|
uptime = time.time() - deps.pack_start_time
|
|
provider_info = {
|
|
"provider": "unknown",
|
|
"key_configured": False,
|
|
"model": "unknown",
|
|
"base_url": None,
|
|
"api_type": None,
|
|
}
|
|
key_required = True
|
|
try:
|
|
client = LLMClient()
|
|
provider_info = client.get_provider_summary()
|
|
key_required = requires_api_key(provider_info.get("provider", "unknown"))
|
|
except Exception:
|
|
provider_info = {
|
|
"provider": "unknown",
|
|
"key_configured": False,
|
|
"model": "unknown",
|
|
"base_url": None,
|
|
"api_type": None,
|
|
}
|
|
key_required = True
|
|
|
|
try:
|
|
from ..services.access_control import is_loopback
|
|
|
|
_ = is_loopback
|
|
|
|
token_val = (
|
|
os.environ.get("OPENCLAW_OBSERVABILITY_TOKEN")
|
|
or os.environ.get("MOLTBOT_OBSERVABILITY_TOKEN")
|
|
or ""
|
|
).strip()
|
|
token_configured = bool(token_val)
|
|
except ImportError:
|
|
from services.access_control import is_loopback
|
|
|
|
_ = is_loopback
|
|
|
|
token_val = (
|
|
os.environ.get("OPENCLAW_OBSERVABILITY_TOKEN")
|
|
or os.environ.get("MOLTBOT_OBSERVABILITY_TOKEN")
|
|
or ""
|
|
).strip()
|
|
token_configured = bool(token_val)
|
|
policy_mode = "token" if token_configured else "loopback_only"
|
|
|
|
try:
|
|
metrics_snapshot = deps.metrics.get_snapshot()
|
|
except Exception:
|
|
metrics_snapshot = {"errors_captured": 0, "logs_processed": 0}
|
|
try:
|
|
executor_snapshot = deps.get_executor_diagnostics() or {}
|
|
except Exception:
|
|
executor_snapshot = {}
|
|
try:
|
|
if __package__ and "." in __package__:
|
|
from ..services.startup_lifecycle import get_startup_diagnostics
|
|
else:
|
|
from services.startup_lifecycle import get_startup_diagnostics
|
|
startup_diagnostics = get_startup_diagnostics()
|
|
except Exception:
|
|
# SECURITY: keep the public fallback deterministic and content-free even when
|
|
# startup diagnostics cannot be imported.
|
|
startup_diagnostics = {
|
|
"schema_version": 1,
|
|
"phase": "package_import",
|
|
"state": "fatal",
|
|
"reason_code": "bootstrap_import_failed",
|
|
"ready": False,
|
|
"degraded": False,
|
|
"fatal": True,
|
|
"attempt": 0,
|
|
"max_attempts": 0,
|
|
"elapsed_ms": 0,
|
|
"phase_elapsed_ms": 0,
|
|
"ready_elapsed_ms": None,
|
|
"warmups": [],
|
|
}
|
|
|
|
job_stats = {}
|
|
try:
|
|
from ..services.job_events import get_job_event_store
|
|
|
|
job_stats = get_job_event_store().stats()
|
|
except Exception:
|
|
pass
|
|
|
|
control_plane_info = {}
|
|
runtime_profile = "minimal"
|
|
try:
|
|
try:
|
|
from ..services.capabilities import _get_control_plane_info
|
|
from ..services.runtime_profile import get_runtime_profile
|
|
except ImportError:
|
|
from services.capabilities import _get_control_plane_info
|
|
from services.runtime_profile import get_runtime_profile
|
|
control_plane_info = _get_control_plane_info()
|
|
runtime_profile = get_runtime_profile().value
|
|
except Exception:
|
|
pass
|
|
|
|
return deps.web.json_response(
|
|
{
|
|
"ok": True,
|
|
"pack": {
|
|
"name": deps.pack_name,
|
|
"version": deps.pack_version,
|
|
"dependencies": {
|
|
"aiohttp": deps.check_dependency("aiohttp"),
|
|
"watchdog": deps.check_dependency("watchdog"),
|
|
},
|
|
},
|
|
"uptime_sec": uptime,
|
|
"config": {
|
|
"provider": provider_info.get("provider"),
|
|
"model": provider_info.get("model"),
|
|
"base_url": provider_info.get("base_url"),
|
|
"api_type": provider_info.get("api_type"),
|
|
"llm_key_configured": provider_info.get("key_configured", False),
|
|
"llm_key_required": key_required,
|
|
},
|
|
"stats": {
|
|
"errors_captured": metrics_snapshot["errors_captured"],
|
|
"logs_processed": metrics_snapshot["logs_processed"],
|
|
"executors": executor_snapshot,
|
|
"observability": job_stats,
|
|
},
|
|
"startup": startup_diagnostics,
|
|
"access_policy": {
|
|
"observability": policy_mode,
|
|
"token_configured": token_configured,
|
|
},
|
|
"control_plane": control_plane_info,
|
|
"runtime_profile": runtime_profile,
|
|
}
|
|
)
|
|
|
|
|
|
async def logs_tail_response(request: Any, deps: RouteHandlerDependencies) -> Any:
|
|
"""Authorize, bound, filter, and redact the log-tail response."""
|
|
|
|
if deps.web is None:
|
|
raise RuntimeError("aiohttp not available")
|
|
ok, init_error = deps.ensure_observability_deps_ready()
|
|
if not ok:
|
|
return deps.web.json_response({"ok": False, "error": init_error}, status=500)
|
|
allowed, error = deps.require_admin_token(request)
|
|
if not allowed:
|
|
return deps.web.json_response({"ok": False, "error": error}, status=403)
|
|
if not deps.check_rate_limit(request, "logs"):
|
|
return deps.build_rate_limit_response(
|
|
request,
|
|
"logs",
|
|
web_module=deps.web,
|
|
error="Rate limit exceeded",
|
|
include_ok=True,
|
|
)
|
|
|
|
try:
|
|
line_count = 50
|
|
val_n = request.query.get("n")
|
|
val_lines = request.query.get("lines")
|
|
target_val = val_n if val_n is not None else val_lines
|
|
if target_val:
|
|
with suppress(ValueError):
|
|
line_count = int(target_val)
|
|
line_count = min(max(line_count, 1), 500)
|
|
trace_id_filter = request.query.get("trace_id")
|
|
prompt_id_filter = request.query.get("prompt_id")
|
|
content = deps.tail_log(deps.log_file, line_count)
|
|
if trace_id_filter or prompt_id_filter:
|
|
content = [
|
|
line
|
|
for line in content
|
|
if (trace_id_filter and trace_id_filter in line)
|
|
or (prompt_id_filter and prompt_id_filter in line)
|
|
]
|
|
if deps.redact_text:
|
|
content = [deps.redact_text(line) for line in content]
|
|
max_bytes = 100_000
|
|
if sum(len(line.encode("utf-8")) for line in content) > max_bytes:
|
|
truncated: list[str] = []
|
|
current_bytes = 0
|
|
for line in reversed(content):
|
|
line_bytes = len(line.encode("utf-8"))
|
|
if current_bytes + line_bytes > max_bytes:
|
|
break
|
|
truncated.insert(0, line)
|
|
current_bytes += line_bytes
|
|
content = truncated
|
|
return deps.web.json_response(
|
|
{
|
|
"ok": True,
|
|
"content": content,
|
|
"filtered": bool(trace_id_filter or prompt_id_filter),
|
|
}
|
|
)
|
|
except Exception as exc:
|
|
return deps.web.json_response({"ok": False, "error": str(exc)}, status=500)
|
|
|
|
|
|
def emit_jobs_list_audit(
|
|
deps: RouteHandlerDependencies,
|
|
*,
|
|
request: Any,
|
|
token_info: Any,
|
|
outcome: str,
|
|
status_code: int,
|
|
reason: str,
|
|
**counts: Any,
|
|
) -> None:
|
|
safe_outcome = outcome if outcome in deps.safe_job_audit_outcomes else "error"
|
|
deps.emit_audit_event(
|
|
action="jobs.list",
|
|
target="jobs",
|
|
outcome=safe_outcome,
|
|
token_info=token_info,
|
|
status_code=status_code,
|
|
details=deps.build_jobs_audit_details(reason, **counts),
|
|
request=request,
|
|
)
|
|
|
|
|
|
async def jobs_response(request: Any, deps: RouteHandlerDependencies) -> Any:
|
|
"""Serve the R213 bounded jobs read model behind its security transaction."""
|
|
|
|
if deps.web is None:
|
|
raise RuntimeError("aiohttp not available")
|
|
token_info = deps.resolve_token_info(request)
|
|
if not deps.check_rate_limit(request, "admin"):
|
|
emit_jobs_list_audit(
|
|
deps,
|
|
request=request,
|
|
token_info=token_info,
|
|
outcome="rate_limit",
|
|
status_code=429,
|
|
reason="jobs_rate_limited",
|
|
)
|
|
return deps.build_rate_limit_response(
|
|
request,
|
|
"admin",
|
|
web_module=deps.web,
|
|
error="jobs_rate_limited",
|
|
include_ok=True,
|
|
)
|
|
|
|
# CRITICAL: metadata is descriptive; this guard must precede queue/history access.
|
|
allowed, _error = deps.require_admin_token(request)
|
|
if not allowed:
|
|
emit_jobs_list_audit(
|
|
deps,
|
|
request=request,
|
|
token_info=token_info,
|
|
outcome="deny",
|
|
status_code=403,
|
|
reason="jobs_admin_required",
|
|
)
|
|
return deps.web.json_response(
|
|
{"ok": False, "error": "jobs_admin_required"}, status=403
|
|
)
|
|
try:
|
|
with deps.jobs_request_tenant_scope(request, token_info) as tenant_context:
|
|
query = deps.normalize_jobs_query(request.query)
|
|
body = deps.read_jobs(query, tenant_id=tenant_context.tenant_id)
|
|
scan = body["scan"]
|
|
emit_jobs_list_audit(
|
|
deps,
|
|
request=request,
|
|
token_info=token_info,
|
|
outcome="allow",
|
|
status_code=200,
|
|
reason="jobs_listed",
|
|
returned_count=len(body["jobs"]),
|
|
excluded_count=scan["excluded"],
|
|
malformed_count=scan["malformed"],
|
|
)
|
|
except deps.tenant_boundary_error as exc:
|
|
emit_jobs_list_audit(
|
|
deps,
|
|
request=request,
|
|
token_info=token_info,
|
|
outcome="deny",
|
|
status_code=403,
|
|
reason=exc.code,
|
|
)
|
|
return deps.web.json_response({"ok": False, "error": exc.code}, status=403)
|
|
except deps.jobs_security_error:
|
|
emit_jobs_list_audit(
|
|
deps,
|
|
request=request,
|
|
token_info=token_info,
|
|
outcome="error",
|
|
status_code=400,
|
|
reason="jobs_query_invalid",
|
|
)
|
|
return deps.web.json_response(
|
|
{"ok": False, "error": "jobs_query_invalid"}, status=400
|
|
)
|
|
except deps.jobs_host_contract_unsupported:
|
|
emit_jobs_list_audit(
|
|
deps,
|
|
request=request,
|
|
token_info=token_info,
|
|
outcome="unsupported",
|
|
status_code=501,
|
|
reason="jobs_host_contract_unsupported",
|
|
)
|
|
return deps.web.json_response(
|
|
{"ok": False, "error": "jobs_host_contract_unsupported"}, status=501
|
|
)
|
|
except deps.jobs_backend_unavailable:
|
|
emit_jobs_list_audit(
|
|
deps,
|
|
request=request,
|
|
token_info=token_info,
|
|
outcome="error",
|
|
status_code=503,
|
|
reason="jobs_backend_unavailable",
|
|
)
|
|
return deps.web.json_response(
|
|
{"ok": False, "error": "jobs_backend_unavailable"}, status=503
|
|
)
|
|
return deps.web.json_response(body)
|
|
|
|
|
|
async def trace_response(request: Any, deps: RouteHandlerDependencies) -> Any:
|
|
"""Authorize and return the redacted operator trace projection."""
|
|
|
|
if deps.web is None:
|
|
raise RuntimeError("aiohttp not available")
|
|
ok, init_error = deps.ensure_observability_deps_ready()
|
|
if not ok:
|
|
return deps.web.json_response({"ok": False, "error": init_error}, status=500)
|
|
allowed, error = deps.require_admin_token(request)
|
|
if not allowed:
|
|
return deps.web.json_response({"ok": False, "error": error}, status=403)
|
|
prompt_id = request.match_info.get("prompt_id")
|
|
if not prompt_id:
|
|
return deps.web.json_response(
|
|
{"ok": False, "error": "missing_prompt_id"}, status=400
|
|
)
|
|
record = deps.trace_store.get(prompt_id)
|
|
if not record:
|
|
return deps.web.json_response({"ok": False, "error": "not_found"}, status=404)
|
|
trace_data = record.to_dict()
|
|
try:
|
|
from ..services.reasoning_redaction import (
|
|
audit_reasoning_reveal,
|
|
resolve_reasoning_reveal,
|
|
sanitize_operator_payload,
|
|
)
|
|
from ..services.redaction import redact_json
|
|
except ImportError:
|
|
from services.reasoning_redaction import (
|
|
audit_reasoning_reveal,
|
|
resolve_reasoning_reveal,
|
|
sanitize_operator_payload,
|
|
)
|
|
from services.redaction import redact_json
|
|
if redact_json:
|
|
trace_data = redact_json(trace_data)
|
|
reveal = resolve_reasoning_reveal(request, admin_authorized=allowed)
|
|
audit_reasoning_reveal(request, target="trace.get", decision=reveal)
|
|
trace_data = sanitize_operator_payload(
|
|
trace_data, include_reasoning=reveal["allowed"]
|
|
)
|
|
return deps.web.json_response({"ok": True, "trace": trace_data})
|