Implement API-first runtime evidence and fail-closed OpenCLI adapters

backend/api/v1/studio_lifecycle.py

@@ -32,6 +32,31 @@
 router = APIRouter()
 
 
+def _isolated_source_errors(
+    project: workflow_schemas.WorkflowProject,
+) -> list[workflow_schemas.WorkflowCompileError]:
+    """Reject Studio drafts whose root source cannot feed any downstream node.
+
+    The generic compiler intentionally permits standalone nodes for capability
+    previews. A Studio draft, however, is publishable and runnable, so accepting
+    an unconnected source would silently discard every record it collects.
+    """
+
+    connected_sources = {edge.source for edge in project.edges}
+    return [
+        workflow_schemas.WorkflowCompileError(
+            code="isolated_source_node",
+            message=(
+                f'Workflow source node "{node.id}" is not connected to a downstream node'
+            ),
+            node_id=node.id,
+            path=["nodes", node.id],
+        )
+        for node in project.nodes
+        if node.kind == "source" and node.id not in connected_sources
+    ]
+
+
 def _image_generation_nodes(
     nodes: object,
     *,
@@ -173,6 +198,7 @@ async def validate_draft(
         )
         valid = False
     else:
+        errors.extend(_isolated_source_errors(project))
         if errors:
             valid = False
         else:

backend/main.py

@@ -221,7 +221,13 @@ async def lifespan(app: FastAPI):
 def create_app() -> FastAPI:
     app = FastAPI(
         title="OpenCLI Admin",
-        description="Multi-channel data collection management system",
+        description=(
+            "Agent-driven workflow and data collection platform. Authenticate protected REST "
+            "and MCP calls with `Authorization: Bearer <API_AUTH_TOKEN>`. Agent workflow: "
+            "inspect `/api/v1/workflows/capabilities`, draft with "
+            "`/api/v1/workflows/demand-draft`, validate with `/api/v1/workflows/compile`, "
+            "then review before publishing or running."
+        ),
         version="0.4.0",
         docs_url="/docs",
         redoc_url="/redoc",
@@ -257,6 +263,75 @@ async def global_exception_handler(request: Request, exc: Exception) -> JSONResp
 
     # Routes
     app.include_router(v1_router)
+    default_openapi = app.openapi
+
+    def openapi_schema() -> dict:
+        if app.openapi_schema:
+            return app.openapi_schema
+        schema = default_openapi()
+        components = schema.setdefault("components", {})
+        security_schemes = components.setdefault("securitySchemes", {})
+        security_schemes["BearerAuth"] = {
+            "type": "http",
+            "scheme": "bearer",
+            "description": "Operator-provisioned OpenCLI Admin API token.",
+        }
+        for path, path_item in schema.get("paths", {}).items():
+            if not path.startswith("/api/"):
+                continue
+            for method, operation in path_item.items():
+                if method.lower() in {"get", "post", "put", "patch", "delete"}:
+                    operation.setdefault("security", [{"BearerAuth": []}])
+        schema["x-opencli-agent"] = {
+            "mcp": {
+                "url": "/mcp",
+                "transport": "streamable-http",
+                "authentication": "BearerAuth",
+            },
+            "workflow": [
+                "list_workflow_node_capabilities",
+                "draft_workflow_from_intent",
+                "preview_workflow_node_patch",
+                "compile_workflow_draft",
+                "run_published_workflow",
+            ],
+        }
+        app.openapi_schema = schema
+        return schema
+
+    app.openapi = openapi_schema
+
+    @app.get("/", include_in_schema=False)
+    async def discovery() -> dict:
+        """Return the stable public entrypoints a human or Agent needs to begin."""
+
+        return {
+            "name": "OpenCLI Admin",
+            "version": app.version,
+            "interfaces": {
+                "openapi": "/openapi.json",
+                "docs": "/docs",
+                "redoc": "/redoc",
+                "mcp": {
+                    "url": "/mcp",
+                    "transport": "streamable-http",
+                },
+            },
+            "authentication": {
+                "type": "http",
+                "scheme": "bearer",
+                "header": "Authorization: Bearer <API_AUTH_TOKEN>",
+                "provisioning": "operator-supplied",
+            },
+            "agentWorkflow": [
+                "discover capabilities",
+                "arrange a review-only node draft",
+                "compile and preflight",
+                "request operator review for effects",
+                "run an immutable published workflow",
+                "inspect trace and evidence",
+            ],
+        }
 
     @app.get("/health")
     async def health() -> dict:

backend/mcp_server.py

@@ -90,8 +90,10 @@ def _transport_security() -> TransportSecuritySettings:
     "opencli-admin",
     version="0.4.0",
     instructions=(
-        "Use project tools for immutable published workflow runs and their durable traces. "
-        "Use source tools for collection administration."
+        "For a new workflow, first inspect workflow node capabilities, then create a review-only "
+        "draft from the operator's intent or preview explicit node patches, then compile it. "
+        "Drafting and compilation never persist or execute work. Run only an immutable published "
+        "workflow after operator review. Use source tools for collection administration."
     ),
 )
 
@@ -241,6 +243,85 @@ async def list_project_workflows(workspace_id: str, project_id: str) -> dict[str
     )
 
 
+@mcp.tool(annotations=READ_ONLY_TOOL, structured_output=True)
+async def list_workflow_node_capabilities() -> dict[str, Any]:
+    """List typed Workflow nodes, runtime bindings, readiness, and input/output contracts."""
+
+    return await _request("GET", "/api/v1/workflows/capabilities")
+
+
+def _new_agent_workflow_project(intent: str, name: str, locale: str) -> dict[str, Any]:
+    """Create the canonical review-only seed used by intent-driven drafting."""
+
+    return {
+        "id": "agent-workflow-draft",
+        "name": name,
+        "profile": "intelligence",
+        "version": 1,
+        "nodes": [
+            {
+                "id": "collection-need",
+                "kind": "schedule",
+                "capability": "trigger",
+                "params": {"text": intent, "locale": locale, "mode": "demand-draft"},
+                "proposalState": "proposed",
+                "ui": {
+                    "catalogId": "intelligence.input.collection-need",
+                    "label": "Collection Need",
+                    "position": {"x": 160, "y": 180},
+                },
+            }
+        ],
+        "edges": [],
+        "adapters": [],
+        "agentPermissions": {
+            "canFetchNetwork": True,
+            "canSendNotifications": False,
+            "canWriteInbox": True,
+            "canMutateExternalSites": False,
+            "allowedDomains": [],
+        },
+    }
+
+
+@mcp.tool(annotations=READ_ONLY_TOOL, structured_output=True)
+async def draft_workflow_from_intent(
+    intent: str,
+    name: str = "Agent workflow draft",
+    locale: str = "zh-CN",
+    project: dict[str, Any] | None = None,
+) -> dict[str, Any]:
+    """Arrange existing nodes into a review-only draft; never persist, publish, or execute it."""
+
+    base_project = project or _new_agent_workflow_project(intent, name, locale)
+    return await _request(
+        "POST",
+        "/api/v1/workflows/demand-draft",
+        json={"project": base_project, "text": intent, "locale": locale},
+    )
+
+
+@mcp.tool(annotations=READ_ONLY_TOOL, structured_output=True)
+async def preview_workflow_node_patch(
+    project: dict[str, Any],
+    operations: list[dict[str, Any]],
+) -> dict[str, Any]:
+    """Preview explicit add/connect/update node operations without persisting the graph."""
+
+    return await _request(
+        "POST",
+        "/api/v1/workflows/patch",
+        json={"project": project, "operations": operations},
+    )
+
+
+@mcp.tool(annotations=READ_ONLY_TOOL, structured_output=True)
+async def compile_workflow_draft(project: dict[str, Any]) -> dict[str, Any]:
+    """Validate and compile a draft graph in memory without dispatching or persisting work."""
+
+    return await _request("POST", "/api/v1/workflows/compile", json={"project": project})
+
+
 @mcp.tool(annotations=IDEMPOTENT_WRITE_TOOL, structured_output=True)
 async def run_published_workflow(
     workspace_id: str,

backend/workflow/demand_assembler.py

@@ -305,6 +305,18 @@ def _native_first_loop_operations(
             ),
         ]
     )
+    if _requires_popularity_ranking(demand_text) and _keyword_from_need(demand_text) != "热门":
+        operations.append(
+            WorkflowPatchOperation(
+                op="request_missing_capability",
+                capability="collection.rank.popularity",
+                reason=(
+                    "The matched OpenCLI search adapters accept a query and limit but expose no "
+                    "popularity sort or threshold. Keep the topic search runnable, but require a "
+                    "governed popularity-ranking capability before claiming hot-post fidelity."
+                ),
+            )
+        )
     return operations
 
 
@@ -576,16 +588,35 @@ def _legacy_keyword_slots_for_need(text: str) -> list[dict[str, Any]]:
 
 def _keyword_from_need(text: str) -> str:
     value = text.strip()
+    value = re.sub(
+        r"[,，;；。]\s*(?:(?:再|然后|并|以及)\s*)?"
+        r"(?:清洗|去重|保存|存储|入库|发送|通知|汇总|分析).*$",
+        "",
+        value,
+        flags=re.IGNORECASE,
+    )
     for pattern in (
-        r"^(抓|采集|收集|监控|找|看)\s*",
+        r"^(?:请|帮我)?\s*(?:抓取?|采集|收集|监控|找|搜索|搜|看(?:下|一下)?)\s*",
         r"(小红书|xiaohongshu|xhs|哔哩哔哩|哔哩|bilibili|b站|bili)",
+        r"(相关的?|有关的?)",
         r"(热帖|热门帖子|热门内容|hot posts?)",
     ):
         value = re.sub(pattern, " ", value, flags=re.IGNORECASE)
+    value = re.sub(r"^\s*(?:(?:和|与|及|、|的)\s*)+", "", value)
     value = re.sub(r"\s+", " ", value).strip(" ，,。")
     return value or "热门"
 
 
+def _requires_popularity_ranking(text: str) -> bool:
+    return bool(
+        re.search(
+            r"(热帖|热门帖子|热门内容|爆款|高热度|hot posts?|trending)",
+            text,
+            flags=re.IGNORECASE,
+        )
+    )
+
+
 # --- OpenCLI adapter catalog matching --------------------------------------
 #
 # Chinese aliases for catalog sites that are commonly typed in Chinese rather

backend/workflow/node_registry.py

@@ -21,7 +21,9 @@
     "intelligence.input.collection-need",
     "intelligence.schedule.cron",
     "intelligence.source.jin10",
+    "intelligence.source.http",
     "intelligence.source.rss",
+    "intelligence.source.rss-bridge",
     "intelligence.source.rsshub",
     "intelligence.source.searxng",
     "intelligence.source.pool",

backend/workflow/opencli_adapter_nodes.py

@@ -33,6 +33,9 @@
 _OPENCLI_CATALOG_GENERATION = 0
 _OPENCLI_SOURCE_CATALOG_ID = "intelligence.source.opencli-slot"
 _EXTERNAL_TOOL_CATALOG_ID = "external.tool.capability"
+_KNOWN_UNAVAILABLE_COMMANDS = {
+    ("sse", "company-list"): "upstream_http_404",
+}
 _TOP_LEVEL_PARAM_KEYS = {
     "format",
     "mode",
@@ -172,6 +175,11 @@ def materialize_opencli_adapter_node(
             "unknown_opencli_adapter_node",
             f'OpenCLI adapter node "{adapter_node_id}" is not registered.',
         )
+    if adapter_node.status in {"preview_only", "design_only"}:
+        raise OpenCLIAdapterNodeMaterializationError(
+            "opencli_adapter_node_unavailable",
+            f'OpenCLI adapter node "{adapter_node_id}" is not currently runnable.',
+        )
     materialized_params = _materialized_params(adapter_node, params or {})
     missing = _missing_required_args(adapter_node, materialized_params)
     if missing:
@@ -287,7 +295,19 @@ def _build_adapter_node(entry: dict[str, Any]) -> WorkflowOpenCLIAdapterNode:
     args = [_adapter_arg(arg) for arg in _read_args(entry.get("args"))]
     required_args = [arg.name for arg in args if arg.required]
     is_read = access == "read"
-    status = "runnable" if is_read and not required_args else "blocked"
+    unavailable_reason = _unavailability_reason(
+        site=site,
+        command=command,
+        browser=browser,
+        strategy=_read_string(entry.get("strategy")),
+    )
+    status = (
+        "preview_only"
+        if is_read and unavailable_reason
+        else "runnable"
+        if is_read and not required_args
+        else "blocked"
+    )
     catalog_id = _OPENCLI_SOURCE_CATALOG_ID if is_read else _EXTERNAL_TOOL_CATALOG_ID
     kind = "source" if is_read else "action"
     capability = "fetch" if is_read else "store"
@@ -344,14 +364,18 @@ def _build_adapter_node(entry: dict[str, Any]) -> WorkflowOpenCLIAdapterNode:
             },
             "canvas": {
                 "node": True,
-                "runBlocked": runtime_readiness != "source_slot_ready",
+                "runBlocked": status != "runnable" or runtime_readiness != "source_slot_ready",
                 "catalogId": catalog_id,
                 "materialization": runtime_readiness,
                 "presetKind": preset_kind,
                 "requiredArgs": required_args,
                 "positionalRequiredArgs": positional_required,
                 "namedRequiredArgs": named_required,
             },
+            "availability": {
+                "available": unavailable_reason is None,
+                "reason": unavailable_reason,
+            },
             "runtime": {
                 "binding": OPENCLI_BINDING_ID if is_read else EXTERNAL_TOOL_BINDING_ID,
             },
@@ -368,6 +392,23 @@ def _build_adapter_node(entry: dict[str, Any]) -> WorkflowOpenCLIAdapterNode:
     )
 
 
+def _unavailability_reason(
+    *,
+    site: str,
+    command: str,
+    browser: bool,
+    strategy: str | None,
+) -> str | None:
+    known_reason = _KNOWN_UNAVAILABLE_COMMANDS.get((site.lower(), command.lower()))
+    if known_reason:
+        return known_reason
+    if browser:
+        return "browser_session_readiness_unverified"
+    if strategy == "cookie":
+        return "cookie_readiness_unverified"
+    return None
+
+
 def _adapter_arg(value: dict[str, Any]) -> WorkflowOpenCLIAdapterNodeArg:
     return WorkflowOpenCLIAdapterNodeArg(
         name=_read_string(value.get("name")) or "arg",