evotraders/backend/api/openclaw.py

# -*- coding: utf-8 -*-
"""Read-only OpenClaw CLI API routes — typed with Pydantic models."""

from __future__ import annotations

from typing import Any

from fastapi import APIRouter, Depends, HTTPException, Query
from pydantic import BaseModel, Field

from backend.services.openclaw_cli import OpenClawCliError, OpenClawCliService
from shared.models.openclaw import OpenClawStatus


router = APIRouter(prefix="/api/openclaw", tags=["openclaw"])


def get_openclaw_cli_service() -> OpenClawCliService:
    """Build the OpenClaw CLI service dependency."""
    return OpenClawCliService()


def _raise_cli_http_error(exc: OpenClawCliError) -> None:
    detail = {
        "message": str(exc),
        "command": exc.command,
        "exit_code": exc.exit_code,
        "stdout": exc.stdout,
        "stderr": exc.stderr,
    }
    status_code = 503 if exc.exit_code is None else 502
    raise HTTPException(status_code=status_code, detail=detail) from exc


# ---------------------------------------------------------------------------
# Response wrappers
# ---------------------------------------------------------------------------

class StatusResponse(BaseModel):
    status: object


class SessionsResponse(BaseModel):
    sessions: list[object]


class SessionDetailResponse(BaseModel):
    session: object | None


class SessionHistoryResponse(BaseModel):
    session_key: str
    session_id: str | None
    events: list[object]
    history: list[object]
    raw_text: str | None


class CronResponse(BaseModel):
    cron: list[object]
    jobs: list[object]


class ApprovalsResponse(BaseModel):
    approvals: list[object]
    pending: list[object]


class AgentsResponse(BaseModel):
    agents: list[object]


class SkillsResponse(BaseModel):
    workspace_dir: str
    managed_skills_dir: str
    skills: list[object]


class ModelsResponse(BaseModel):
    models: list[object]


class HooksResponse(BaseModel):
    workspace_dir: str
    managed_hooks_dir: str
    hooks: list[object]


class PluginsResponse(BaseModel):
    workspace_dir: str
    plugins: list[object]
    diagnostics: list[object]


class SecretsAuditResponse(BaseModel):
    version: int
    status: str
    findings: list[object]


class SecurityAuditResponse2(BaseModel):
    report: object | None
    secret_diagnostics: list[str]


class DaemonStatusResponse(BaseModel):
    service: object | None
    port: object | None
    rpc: object | None
    health: object | None


class PairingListResponse2(BaseModel):
    channel: str
    requests: list[object]


class QrCodeResponse2(BaseModel):
    setup_code: str
    gateway_url: str
    auth: str
    url_source: str


class UpdateStatusResponse2(BaseModel):
    update: object | None
    channel: object | None


class ModelAliasesResponse(BaseModel):
    aliases: dict[str, str]


class ModelFallbacksResponse(BaseModel):
    key: str
    label: str
    items: list[object]


class SkillUpdateResponse(BaseModel):
    ok: bool
    slug: str
    version: str
    error: str | None


class ModelsStatusResponse(BaseModel):
    configPath: str | None = None
    agentId: str | None = None
    agentDir: str | None = None
    defaultModel: str | None = None
    resolvedDefault: str | None = None
    fallbacks: list[str] = Field(default_factory=list)
    imageModel: str | None = None
    imageFallbacks: list[str] = Field(default_factory=list)
    aliases: dict[str, str] = Field(default_factory=dict)
    allowed: list[str] = Field(default_factory=list)
    auth: dict[str, Any] = Field(default_factory=dict)


class ChannelsStatusResponse(BaseModel):
    reachable: bool | None = None
    channelAccounts: dict[str, Any] = Field(default_factory=dict)
    channels: list[str] = Field(default_factory=list)
    issues: list[dict[str, Any]] = Field(default_factory=list)


class ChannelsListResponse(BaseModel):
    chat: dict[str, list[str]] = Field(default_factory=dict)
    auth: list[dict[str, Any]] = Field(default_factory=list)
    usage: dict[str, Any] | None = None


class HookInfoResponse(BaseModel):
    name: str | None = None
    description: str | None = None
    source: str | None = None
    pluginId: str | None = None
    filePath: str | None = None
    handlerPath: str | None = None
    hookKey: str | None = None
    emoji: str | None = None
    homepage: str | None = None
    events: list[str] = Field(default_factory=list)
    enabledByConfig: bool | None = None
    loadable: bool | None = None
    requirementsSatisfied: bool | None = None
    requirements: dict[str, Any] = Field(default_factory=dict)
    error: str | None = None
    raw: str | None = None


class HooksCheckResponse(BaseModel):
    workspace_dir: str = ""
    managed_hooks_dir: str = ""
    hooks: list[dict[str, Any]] = Field(default_factory=list)
    eligible: bool | None = None
    verbose: bool | None = None


class PluginInspectEntry(BaseModel):
    plugin: dict[str, Any] = Field(default_factory=dict)
    shape: str | None = None
    capabilityMode: str | None = None
    capabilityCount: int = 0
    capabilities: list[dict[str, Any]] = Field(default_factory=list)
    typedHooks: list[dict[str, Any]] = Field(default_factory=list)
    customHooks: list[dict[str, Any]] = Field(default_factory=list)
    tools: list[dict[str, Any]] = Field(default_factory=list)
    commands: list[str] = Field(default_factory=list)
    cliCommands: list[str] = Field(default_factory=list)
    services: list[str] = Field(default_factory=list)
    gatewayMethods: list[str] = Field(default_factory=list)
    mcpServers: list[dict[str, Any]] = Field(default_factory=list)
    lspServers: list[dict[str, Any]] = Field(default_factory=list)
    httpRouteCount: int = 0
    bundleCapabilities: list[str] = Field(default_factory=list)


class PluginsInspectResponse(BaseModel):
    inspect: list[dict[str, Any]] = Field(default_factory=list)


class AgentBindingItem(BaseModel):
    agentId: str
    match: dict[str, Any]
    description: str


class AgentsBindingsResponse(BaseModel):
    bindings: list[AgentBindingItem]


# ---------------------------------------------------------------------------
# Routes — use typed model methods and return Pydantic models directly
# ---------------------------------------------------------------------------

@router.get("/status")
async def api_openclaw_status(
    service: OpenClawCliService = Depends(get_openclaw_cli_service),
) -> OpenClawStatus:
    """Read `openclaw status --json` and return a typed model."""
    try:
        return service.status_model()
    except OpenClawCliError as exc:
        _raise_cli_http_error(exc)


@router.get("/sessions")
async def api_openclaw_sessions(
    service: OpenClawCliService = Depends(get_openclaw_cli_service),
) -> SessionsResponse:
    """Read `openclaw sessions --json` and return a typed SessionsList."""
    try:
        result = service.list_sessions_model()
        return SessionsResponse(sessions=result.sessions)
    except OpenClawCliError as exc:
        _raise_cli_http_error(exc)


@router.get("/sessions/{session_key:path}/history")
async def api_openclaw_session_history(
    session_key: str,
    limit: int = Query(20, ge=1, le=200),
    service: OpenClawCliService = Depends(get_openclaw_cli_service),
) -> SessionHistoryResponse:
    """Read session history and return a typed SessionHistory."""
    try:
        result = service.get_session_history_model(session_key, limit=limit)
        return SessionHistoryResponse(
            session_key=result.session_key,
            session_id=result.session_id,
            events=result.events,
            history=result.events,  # alias for compat
            raw_text=result.raw_text,
        )
    except OpenClawCliError as exc:
        _raise_cli_http_error(exc)


@router.get("/sessions/{session_key:path}")
async def api_openclaw_session_detail(
    session_key: str,
    service: OpenClawCliService = Depends(get_openclaw_cli_service),
) -> SessionDetailResponse:
    """Resolve a single session and return it as a typed model."""
    try:
        session = service.get_session_model(session_key)
        return SessionDetailResponse(session=session)
    except KeyError as exc:
        raise HTTPException(status_code=404, detail=f"session '{session_key}' not found") from exc
    except OpenClawCliError as exc:
        _raise_cli_http_error(exc)


@router.get("/cron")
async def api_openclaw_cron(
    service: OpenClawCliService = Depends(get_openclaw_cli_service),
) -> CronResponse:
    """Read `openclaw cron list --json` and return a typed CronList."""
    try:
        result = service.list_cron_jobs_model()
        return CronResponse(cron=list(result.cron), jobs=list(result.jobs))
    except OpenClawCliError as exc:
        _raise_cli_http_error(exc)


@router.get("/approvals")
async def api_openclaw_approvals(
    service: OpenClawCliService = Depends(get_openclaw_cli_service),
) -> ApprovalsResponse:
    """Read `openclaw approvals get --json` and return a typed ApprovalsList."""
    try:
        result = service.list_approvals_model()
        return ApprovalsResponse(
            approvals=list(result.approvals),
            pending=list(result.pending),
        )
    except OpenClawCliError as exc:
        _raise_cli_http_error(exc)


@router.get("/agents")
async def api_openclaw_agents(
    service: OpenClawCliService = Depends(get_openclaw_cli_service),
) -> AgentsResponse:
    """Read `openclaw agents list --json` and return a typed AgentsList."""
    try:
        result = service.list_agents_model()
        return AgentsResponse(agents=list(result.agents))
    except OpenClawCliError as exc:
        _raise_cli_http_error(exc)


@router.get("/agents/presence")
async def api_openclaw_agents_presence(
    service: OpenClawCliService = Depends(get_openclaw_cli_service),
) -> dict[str, Any]:
    """Read runtime session presence for all agents from session files."""
    result = service.agents_presence()
    return result


# ---------------------------------------------------------------------------
# Write agents routes
# ---------------------------------------------------------------------------


class AgentAddResponse(BaseModel):
    agentId: str
    name: str
    workspace: str
    agentDir: str
    model: str | None = None
    bindings: dict[str, Any] = Field(default_factory=dict)


class AgentDeleteResponse(BaseModel):
    agentId: str
    workspace: str
    agentDir: str
    sessionsDir: str
    removedBindings: list[str] = Field(default_factory=list)
    removedAllow: list[str] = Field(default_factory=list)


class AgentBindResponse(BaseModel):
    agentId: str
    added: list[str] = Field(default_factory=list)
    updated: list[str] = Field(default_factory=list)
    skipped: list[str] = Field(default_factory=list)
    conflicts: list[str] = Field(default_factory=list)


class AgentUnbindResponse(BaseModel):
    agentId: str
    removed: list[str] = Field(default_factory=list)
    missing: list[str] = Field(default_factory=list)
    conflicts: list[str] = Field(default_factory=list)


class AgentIdentityResponse(BaseModel):
    agentId: str
    identity: dict[str, Any] = Field(default_factory=dict)
    workspace: str | None = None
    identityFile: str | None = None


@router.post("/agents/add")
async def api_openclaw_agents_add(
    name: str,
    *,
    workspace: str | None = None,
    model: str | None = None,
    agent_dir: str | None = None,
    bind: list[str] | None = None,
    non_interactive: bool = False,
    service: OpenClawCliService = Depends(get_openclaw_cli_service),
) -> AgentAddResponse:
    """Run `openclaw agents add <name>` and return JSON result."""
    try:
        result = service.agents_add(
            name,
            workspace=workspace,
            model=model,
            agent_dir=agent_dir,
            bind=bind,
            non_interactive=non_interactive,
        )
        return AgentAddResponse.model_validate(result, strict=False)
    except OpenClawCliError as exc:
        _raise_cli_http_error(exc)


@router.post("/agents/delete/{id}")
async def api_openclaw_agents_delete(
    id: str,
    force: bool = False,
    service: OpenClawCliService = Depends(get_openclaw_cli_service),
) -> AgentDeleteResponse:
    """Run `openclaw agents delete <id> [--force]` and return JSON result."""
    try:
        result = service.agents_delete(id, force=force)
        return AgentDeleteResponse.model_validate(result, strict=False)
    except OpenClawCliError as exc:
        _raise_cli_http_error(exc)


@router.post("/agents/bind")
async def api_openclaw_agents_bind(
    *,
    agent: str | None = None,
    bind: list[str] | None = None,
    service: OpenClawCliService = Depends(get_openclaw_cli_service),
) -> AgentBindResponse:
    """Run `openclaw agents bind [--agent <id>] [--bind <spec>]` and return JSON result."""
    try:
        result = service.agents_bind(agent=agent, bind=bind)
        return AgentBindResponse.model_validate(result, strict=False)
    except OpenClawCliError as exc:
        _raise_cli_http_error(exc)


@router.post("/agents/unbind")
async def api_openclaw_agents_unbind(
    *,
    agent: str | None = None,
    bind: list[str] | None = None,
    all: bool = False,
    service: OpenClawCliService = Depends(get_openclaw_cli_service),
) -> AgentUnbindResponse:
    """Run `openclaw agents unbind [--agent <id>] [--bind <spec>] [--all]` and return JSON result."""
    try:
        result = service.agents_unbind(agent=agent, bind=bind, all=all)
        return AgentUnbindResponse.model_validate(result, strict=False)
    except OpenClawCliError as exc:
        _raise_cli_http_error(exc)


@router.post("/agents/set-identity")
async def api_openclaw_agents_set_identity(
    *,
    agent: str | None = None,
    workspace: str | None = None,
    identity_file: str | None = None,
    name: str | None = None,
    emoji: str | None = None,
    theme: str | None = None,
    avatar: str | None = None,
    from_identity: bool = False,
    service: OpenClawCliService = Depends(get_openclaw_cli_service),
) -> AgentIdentityResponse:
    """Run `openclaw agents set-identity` and return JSON result."""
    try:
        result = service.agents_set_identity(
            agent=agent,
            workspace=workspace,
            identity_file=identity_file,
            name=name,
            emoji=emoji,
            theme=theme,
            avatar=avatar,
            from_identity=from_identity,
        )
        return AgentIdentityResponse.model_validate(result, strict=False)
    except OpenClawCliError as exc:
        _raise_cli_http_error(exc)


@router.get("/skills")
async def api_openclaw_skills(
    service: OpenClawCliService = Depends(get_openclaw_cli_service),
) -> SkillsResponse:
    """Read `openclaw skills list --json` and return a typed SkillStatusReport."""
    try:
        result = service.list_skills_model()
        return SkillsResponse(
            workspace_dir=result.workspace_dir,
            managed_skills_dir=result.managed_skills_dir,
            skills=list(result.skills),
        )
    except OpenClawCliError as exc:
        _raise_cli_http_error(exc)


@router.get("/models")
async def api_openclaw_models(
    service: OpenClawCliService = Depends(get_openclaw_cli_service),
) -> ModelsResponse:
    """Read `openclaw models list --json` and return a typed ModelsList."""
    try:
        result = service.list_models_model()
        return ModelsResponse(models=list(result.models))
    except OpenClawCliError as exc:
        _raise_cli_http_error(exc)


@router.get("/hooks")
async def api_openclaw_hooks(
    service: OpenClawCliService = Depends(get_openclaw_cli_service),
) -> HooksResponse:
    try:
        result = service.list_hooks_model()
        return HooksResponse(
            workspace_dir=result.workspace_dir,
            managed_hooks_dir=result.managed_hooks_dir,
            hooks=list(result.hooks),
        )
    except OpenClawCliError as exc:
        _raise_cli_http_error(exc)


@router.get("/plugins")
async def api_openclaw_plugins(
    service: OpenClawCliService = Depends(get_openclaw_cli_service),
) -> PluginsResponse:
    try:
        result = service.list_plugins_model()
        return PluginsResponse(
            workspace_dir=result.workspace_dir,
            plugins=list(result.plugins),
            diagnostics=list(result.diagnostics),
        )
    except OpenClawCliError as exc:
        _raise_cli_http_error(exc)


@router.get("/secrets-audit")
async def api_openclaw_secrets_audit(
    service: OpenClawCliService = Depends(get_openclaw_cli_service),
) -> SecretsAuditResponse:
    try:
        result = service.secrets_audit_model()
        return SecretsAuditResponse(
            version=result.version,
            status=result.status,
            findings=list(result.findings),
        )
    except OpenClawCliError as exc:
        _raise_cli_http_error(exc)


@router.get("/security-audit")
async def api_openclaw_security_audit(
    service: OpenClawCliService = Depends(get_openclaw_cli_service),
) -> SecurityAuditResponse2:
    try:
        result = service.security_audit_model()
        return SecurityAuditResponse2(
            report=result.report.model_dump() if result.report else None,
            secret_diagnostics=list(result.secret_diagnostics),
        )
    except OpenClawCliError as exc:
        _raise_cli_http_error(exc)


@router.get("/daemon-status")
async def api_openclaw_daemon_status(
    service: OpenClawCliService = Depends(get_openclaw_cli_service),
) -> DaemonStatusResponse:
    try:
        result = service.daemon_status_model()
        return DaemonStatusResponse(
            service=result.service.model_dump() if result.service else None,
            port=result.port.model_dump() if result.port else None,
            rpc=result.rpc.model_dump() if result.rpc else None,
            health=result.health.model_dump() if result.health else None,
        )
    except OpenClawCliError as exc:
        _raise_cli_http_error(exc)


@router.get("/pairing")
async def api_openclaw_pairing(
    service: OpenClawCliService = Depends(get_openclaw_cli_service),
) -> PairingListResponse2:
    try:
        result = service.pairing_list_model()
        return PairingListResponse2(
            channel=result.channel,
            requests=list(result.requests),
        )
    except OpenClawCliError as exc:
        _raise_cli_http_error(exc)


@router.get("/qr")
async def api_openclaw_qr(
    service: OpenClawCliService = Depends(get_openclaw_cli_service),
) -> QrCodeResponse2:
    try:
        result = service.qr_code_model()
        return QrCodeResponse2(
            setup_code=result.setup_code,
            gateway_url=result.gateway_url,
            auth=result.auth,
            url_source=result.url_source,
        )
    except OpenClawCliError as exc:
        _raise_cli_http_error(exc)


@router.get("/update-status")
async def api_openclaw_update_status(
    service: OpenClawCliService = Depends(get_openclaw_cli_service),
) -> UpdateStatusResponse2:
    try:
        result = service.update_status_model()
        return UpdateStatusResponse2(
            update=result.update.model_dump() if result.update else None,
            channel=result.channel.model_dump() if result.channel else None,
        )
    except OpenClawCliError as exc:
        _raise_cli_http_error(exc)


@router.get("/models-aliases")
async def api_openclaw_models_aliases(
    service: OpenClawCliService = Depends(get_openclaw_cli_service),
) -> ModelAliasesResponse:
    try:
        result = service.list_model_aliases_model()
        return ModelAliasesResponse(aliases=result.aliases)
    except OpenClawCliError as exc:
        _raise_cli_http_error(exc)


@router.get("/models-fallbacks")
async def api_openclaw_models_fallbacks(
    service: OpenClawCliService = Depends(get_openclaw_cli_service),
) -> ModelFallbacksResponse:
    try:
        result = service.list_model_fallbacks_model()
        return ModelFallbacksResponse(
            key=result.key,
            label=result.label,
            items=list(result.items),
        )
    except OpenClawCliError as exc:
        _raise_cli_http_error(exc)


@router.get("/models-image-fallbacks")
async def api_openclaw_models_image_fallbacks(
    service: OpenClawCliService = Depends(get_openclaw_cli_service),
) -> ModelFallbacksResponse:
    try:
        result = service.list_model_image_fallbacks_model()
        return ModelFallbacksResponse(
            key=result.key,
            label=result.label,
            items=list(result.items),
        )
    except OpenClawCliError as exc:
        _raise_cli_http_error(exc)


@router.get("/skill-update")
async def api_openclaw_skill_update(
    slug: str | None = None,
    all: bool = False,
    service: OpenClawCliService = Depends(get_openclaw_cli_service),
) -> SkillUpdateResponse:
    try:
        result = service.skill_update_model(slug=slug, all=all)
        return SkillUpdateResponse(
            ok=result.ok,
            slug=result.slug,
            version=result.version,
            error=result.error,
        )
    except OpenClawCliError as exc:
        _raise_cli_http_error(exc)


@router.get("/models-status")
async def api_openclaw_models_status(
    probe: bool = False,
    service: OpenClawCliService = Depends(get_openclaw_cli_service),
) -> ModelsStatusResponse:
    """Read `openclaw models status --json [--probe]` and return a typed dict."""
    try:
        result = service.models_status_model(probe=probe)
        return ModelsStatusResponse.model_validate(result, strict=False)
    except OpenClawCliError as exc:
        _raise_cli_http_error(exc)


@router.get("/channels-status")
async def api_openclaw_channels_status(
    probe: bool = False,
    service: OpenClawCliService = Depends(get_openclaw_cli_service),
) -> ChannelsStatusResponse:
    """Read `openclaw channels status --json [--probe]` and return a typed dict."""
    try:
        result = service.channels_status_model(probe=probe)
        return ChannelsStatusResponse.model_validate(result, strict=False)
    except OpenClawCliError as exc:
        _raise_cli_http_error(exc)


@router.get("/channels-list")
async def api_openclaw_channels_list(
    service: OpenClawCliService = Depends(get_openclaw_cli_service),
) -> ChannelsListResponse:
    """Read `openclaw channels list --json` and return a typed dict."""
    try:
        result = service.channels_list_model()
        return ChannelsListResponse.model_validate(result, strict=False)
    except OpenClawCliError as exc:
        _raise_cli_http_error(exc)


@router.get("/hooks/info/{name}")
async def api_openclaw_hook_info(
    name: str,
    service: OpenClawCliService = Depends(get_openclaw_cli_service),
) -> HookInfoResponse:
    """Read `openclaw hooks info <name> --json` and return a typed dict."""
    try:
        result = service.hook_info_model(name)
        return HookInfoResponse.model_validate(result, strict=False)
    except OpenClawCliError as exc:
        _raise_cli_http_error(exc)


@router.get("/hooks/check")
async def api_openclaw_hooks_check(
    service: OpenClawCliService = Depends(get_openclaw_cli_service),
) -> HooksCheckResponse:
    """Read `openclaw hooks check --json` and return a typed dict."""
    try:
        result = service.hooks_check_model()
        return HooksCheckResponse.model_validate(result, strict=False)
    except OpenClawCliError as exc:
        _raise_cli_http_error(exc)


@router.get("/plugins-inspect")
async def api_openclaw_plugins_inspect(
    plugin_id: str | None = None,
    all: bool = False,
    service: OpenClawCliService = Depends(get_openclaw_cli_service),
) -> PluginsInspectResponse:
    """Read `openclaw plugins inspect --json [--all]` and return a typed dict."""
    try:
        result = service.plugins_inspect_model(plugin_id=plugin_id, all=all)
        inspect = result if isinstance(result, list) else result.get("inspect", [])
        return PluginsInspectResponse(inspect=inspect)
    except OpenClawCliError as exc:
        _raise_cli_http_error(exc)


class AgentBindingItem(BaseModel):
    agentId: str
    match: dict[str, Any]
    description: str


class AgentsBindingsResponse(BaseModel):
    bindings: list[AgentBindingItem]


@router.get("/agents-bindings")
async def api_openclaw_agents_bindings(
    agent: str | None = None,
    service: OpenClawCliService = Depends(get_openclaw_cli_service),
) -> AgentsBindingsResponse:
    """Read `openclaw agents bindings --json [--agent <id>]` and return bindings list."""
    try:
        result = service.agents_bindings_model(agent=agent)
        bindings = result if isinstance(result, list) else []
        return AgentsBindingsResponse(bindings=bindings)
    except OpenClawCliError as exc:
        _raise_cli_http_error(exc)


@router.get("/gateway-status")
async def api_openclaw_gateway_status(
    url: str | None = None,
    token: str | None = None,
    service: OpenClawCliService = Depends(get_openclaw_cli_service),
) -> dict[str, Any]:
    """Read `openclaw gateway status --json [--url <url>] [--token <token>]`. Returns full gateway probe result."""
    try:
        result = service.gateway_status(url=url, token=token)
        return result
    except OpenClawCliError as exc:
        _raise_cli_http_error(exc)


@router.get("/memory-status")
async def api_openclaw_memory_status(
    agent: str | None = None,
    deep: bool = False,
    service: OpenClawCliService = Depends(get_openclaw_cli_service),
) -> list[dict[str, Any]]:
    """Read `openclaw memory status --json [--agent <id>] [--deep]`. Returns array of per-agent memory status."""
    try:
        result = service.memory_status(agent=agent, deep=deep)
        return result if isinstance(result, list) else []
    except OpenClawCliError as exc:
        _raise_cli_http_error(exc)


class WorkspaceFilesResponse(BaseModel):
    workspace: str
    files: list[dict[str, Any]]
    error: str | None = None


@router.get("/workspace-files")
async def api_openclaw_workspace_files(
    workspace: str = Query(..., description="Path to the agent workspace directory"),
    service: OpenClawCliService = Depends(get_openclaw_cli_service),
) -> WorkspaceFilesResponse:
    """List .md files in an OpenClaw agent workspace with their content previews."""
    result = service.list_workspace_files(workspace)
    return WorkspaceFilesResponse.model_validate(result, strict=False)