From 0c245926a1ef2e36e671f28fae30610eca614abf Mon Sep 17 00:00:00 2001 From: lunnt <2276214182@qq.com> Date: Mon, 3 Aug 2026 20:44:38 +0800 Subject: [PATCH] WIP: integrate API-first runtime evidence on latest main [gstack-context] --- backend/api/v1/studio_lifecycle.py | 26 ++++ backend/main.py | 77 +++++++++++- backend/mcp_server.py | 85 ++++++++++++- backend/workflow/demand_assembler.py | 33 ++++- backend/workflow/node_registry.py | 2 + backend/workflow/opencli_adapter_nodes.py | 45 ++++++- backend/workflow/opencli_hda_tracer.py | 28 +++++ frontend/components/flow/command-palette.tsx | 2 +- .../workflow/backend-opencli-adapter-nodes.ts | 1 + frontend/lib/workflow/schema.ts | 32 ++++- .../workflow/use-opencli-adapter-catalog.ts | 7 +- frontend/next-env.d.ts | 2 +- .../scripts/check-workflow-regressions.mjs | 114 ++++++++++++++++++ .../integration/test_studio_lifecycle_api.py | 51 ++++++++ .../test_workflow_capabilities_api.py | 6 +- .../integration/test_workflow_compile_api.py | 71 +++++++++++ .../test_workflow_opencli_hda_trace_api.py | 19 ++- tests/integration/test_workflow_patch_api.py | 48 +++++++- tests/unit/test_demand_assembler.py | 15 ++- tests/unit/test_main.py | 33 +++++ tests/unit/test_mcp_server.py | 41 +++++++ tests/unit/test_opencli_adapter_nodes.py | 67 ++++++++++ 22 files changed, 783 insertions(+), 22 deletions(-) diff --git a/backend/api/v1/studio_lifecycle.py b/backend/api/v1/studio_lifecycle.py index 7262e4d..951fca7 100644 --- a/backend/api/v1/studio_lifecycle.py +++ b/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: diff --git a/backend/main.py b/backend/main.py index 3e45dfe..a09b027 100644 --- a/backend/main.py +++ b/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 `. 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 ", + "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: diff --git a/backend/mcp_server.py b/backend/mcp_server.py index 0422d2a..2885c7f 100644 --- a/backend/mcp_server.py +++ b/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, diff --git a/backend/workflow/demand_assembler.py b/backend/workflow/demand_assembler.py index 5a37eff..1882cff 100644 --- a/backend/workflow/demand_assembler.py +++ b/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 diff --git a/backend/workflow/node_registry.py b/backend/workflow/node_registry.py index 94e7de2..afe7512 100644 --- a/backend/workflow/node_registry.py +++ b/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", diff --git a/backend/workflow/opencli_adapter_nodes.py b/backend/workflow/opencli_adapter_nodes.py index b850bbc..662eee1 100644 --- a/backend/workflow/opencli_adapter_nodes.py +++ b/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,7 +364,7 @@ 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, @@ -352,6 +372,10 @@ def _build_adapter_node(entry: dict[str, Any]) -> WorkflowOpenCLIAdapterNode: "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", diff --git a/backend/workflow/opencli_hda_tracer.py b/backend/workflow/opencli_hda_tracer.py index 55a9dd8..4fcedb0 100644 --- a/backend/workflow/opencli_hda_tracer.py +++ b/backend/workflow/opencli_hda_tracer.py @@ -1350,6 +1350,12 @@ async def start_workflow_run( fleet_match, **dispatch_kwargs, ) + if not is_write: + output_items, agent_dispatch_details = _bounded_opencli_dispatch_result( + output_items, + agent_dispatch_details, + max_items=body.project.settings.maxItemsPerRun, + ) output_items = _opencli_dispatch_source_items(node, dispatch, output_items) batch = _batch_reference(body.project.id, run_id, dispatch) if output_items: @@ -3007,6 +3013,28 @@ def _opencli_dispatch_source_items( ] +def _bounded_opencli_dispatch_result( + raw_items: list[dict[str, Any]], + details: dict[str, object] | None, + *, + max_items: int, +) -> tuple[list[dict[str, Any]], dict[str, object] | None]: + """Enforce the workflow-level item cap when an adapter cannot push it down.""" + + limit = max(1, max_items) + received_count = len(raw_items) + items = raw_items[:limit] + if details is None or received_count <= limit: + return items, details + return items, { + **details, + "itemCount": len(items), + "receivedItemCount": received_count, + "maxItemsPerRun": limit, + "truncated": True, + } + + def _is_local_opencli_dispatch(details: dict[str, object] | None) -> bool: return bool(details and details.get("protocol") == "local") diff --git a/frontend/components/flow/command-palette.tsx b/frontend/components/flow/command-palette.tsx index 6ab019d..e01b872 100644 --- a/frontend/components/flow/command-palette.tsx +++ b/frontend/components/flow/command-palette.tsx @@ -591,7 +591,7 @@ export function CommandPalette({ } addWorkflowNodeFromCatalog(workflowCatalogItemForOpenCLIAdapterNode(item, values), anchorPosition()) } else if (materialization === "source_slot_ready" && item.status === "runnable") { - addWorkflowNodeFromCatalog(openCLIAdapterNodeToCatalogItem(item), anchorPosition()) + addWorkflowNodeFromCatalog(workflowCatalogItemForOpenCLIAdapterNode(item), anchorPosition()) } else if (materialization === "tool_capability_review_required") { addWorkflowNodeFromCatalog(openCLIAdapterNodeToCatalogItem(item), anchorPosition()) } else { diff --git a/frontend/lib/workflow/backend-opencli-adapter-nodes.ts b/frontend/lib/workflow/backend-opencli-adapter-nodes.ts index f3cbe4f..9e6ddae 100644 --- a/frontend/lib/workflow/backend-opencli-adapter-nodes.ts +++ b/frontend/lib/workflow/backend-opencli-adapter-nodes.ts @@ -317,6 +317,7 @@ export function openCLIAdapterNodePresentation( export function openCLIAdapterNodeMaterialization( node: WorkflowOpenCLIAdapterNode, ): WorkflowOpenCLIAdapterMaterialization { + if (node.status === "preview_only" || node.status === "design_only") return "unavailable" if ( node.runtimeReadiness === "source_slot_ready" || node.runtimeReadiness === "source_slot_requires_params" || diff --git a/frontend/lib/workflow/schema.ts b/frontend/lib/workflow/schema.ts index 14242f3..64dd6fb 100644 --- a/frontend/lib/workflow/schema.ts +++ b/frontend/lib/workflow/schema.ts @@ -201,8 +201,38 @@ export type AdapterBinding = z.infer export type AgentPermissions = z.infer export type WorkflowProject = z.infer +const OPENCLI_SOURCE_SLOT_CATALOG_ID = "intelligence.source.opencli-slot" +const LEGACY_OPENCLI_ADAPTER_CATALOG_ID = /^opencli\.adapter\.[a-z0-9][a-z0-9._-]*$/i + +function normalizeLegacyOpenCLISourceCatalogIds(project: WorkflowProject): WorkflowProject { + const adapters = new Map(project.adapters.map((adapter) => [adapter.id, adapter])) + let changed = false + const nodes = project.nodes.map((node) => { + const catalogId = node.ui?.catalogId + const opencliAdapterNodeId = node.params.opencliAdapterNodeId + const adapter = node.adapter ? adapters.get(node.adapter) : undefined + if ( + node.kind !== "source" || + node.capability !== "fetch" || + typeof catalogId !== "string" || + !LEGACY_OPENCLI_ADAPTER_CATALOG_ID.test(catalogId) || + opencliAdapterNodeId !== catalogId || + adapter?.type !== "source" || + adapter.provider !== "opencli" + ) { + return node + } + changed = true + return { + ...node, + ui: { ...node.ui, catalogId: OPENCLI_SOURCE_SLOT_CATALOG_ID }, + } + }) + return changed ? { ...project, nodes } : project +} + export function parseWorkflowProject(input: unknown): WorkflowProject { - const project = workflowProjectSchema.parse(input) + const project = normalizeLegacyOpenCLISourceCatalogIds(workflowProjectSchema.parse(input)) validateWorkflowReferences(project) return project } diff --git a/frontend/lib/workflow/use-opencli-adapter-catalog.ts b/frontend/lib/workflow/use-opencli-adapter-catalog.ts index c5cbf40..92bd6d1 100644 --- a/frontend/lib/workflow/use-opencli-adapter-catalog.ts +++ b/frontend/lib/workflow/use-opencli-adapter-catalog.ts @@ -4,6 +4,7 @@ import { useCallback, useEffect, useState } from "react" import { fetchWorkflowOpenCLIAdapterNodes, + workflowCatalogItemForOpenCLIAdapterNode, type WorkflowOpenCLIAdapterNodesResponse, } from "./backend-opencli-adapter-nodes" import { @@ -39,7 +40,11 @@ export function useOpenCLIAdapterCatalog(enabled = true) { }) if (signal?.aborted) return setState({ - items: response.nodes.map(openCLIAdapterNodeToCatalogItem), + items: response.nodes.map((node) => + node.access === "read" + ? workflowCatalogItemForOpenCLIAdapterNode(node) + : openCLIAdapterNodeToCatalogItem(node), + ), response, error: null, loading: false, diff --git a/frontend/next-env.d.ts b/frontend/next-env.d.ts index 9edff1c..c4b7818 100644 --- a/frontend/next-env.d.ts +++ b/frontend/next-env.d.ts @@ -1,6 +1,6 @@ /// /// -import "./.next/types/routes.d.ts"; +import "./.next/dev/types/routes.d.ts"; // NOTE: This file should not be edited // see https://nextjs.org/docs/app/api-reference/config/typescript for more information. diff --git a/frontend/scripts/check-workflow-regressions.mjs b/frontend/scripts/check-workflow-regressions.mjs index 4f54c78..fa7b8fd 100644 --- a/frontend/scripts/check-workflow-regressions.mjs +++ b/frontend/scripts/check-workflow-regressions.mjs @@ -232,6 +232,120 @@ test('OpenCLI write commands become accepted action nodes with explicit mutation assert.ok(updated.adapters.some((adapter) => adapter.id === 'opencli-twitter')) }) +test('runnable OpenCLI source presets retain the stable backend catalog binding', async () => { + const [{ openCLIAdapterNodeMaterialization, workflowCatalogItemForOpenCLIAdapterNode }, { addCatalogNodeToWorkflowProject }, { parseWorkflowProject }, paletteSource, catalogHookSource] = await Promise.all([ + importTypeScript('lib/workflow/backend-opencli-adapter-nodes.ts'), + importTypeScript('lib/workflow/node-catalog.ts'), + importTypeScript('lib/workflow/schema.ts'), + readSource('components/flow/command-palette.tsx'), + readSource('lib/workflow/use-opencli-adapter-catalog.ts'), + ]) + const catalogItem = workflowCatalogItemForOpenCLIAdapterNode({ + id: 'opencli.adapter.statsgov.nbs', + label: 'statsgov · nbs', + description: 'National Bureau of Statistics releases', + status: 'runnable', + site: 'statsgov', + command: 'nbs', + access: 'read', + browser: false, + catalogId: 'intelligence.source.opencli-slot', + kind: 'source', + capability: 'fetch', + presetKind: 'source_slot', + runtimeReadiness: 'source_slot_ready', + requiredArgs: [], + args: [], + adapter: { + id: 'opencli-statsgov-nbs', + type: 'source', + provider: 'opencli', + mode: 'live', + config: { channel: 'opencli' }, + }, + params: { + site: 'statsgov', + command: 'nbs', + format: 'json', + args: {}, + positional_args: [], + }, + manifest: {}, + }) + const addOpenCLIAdapter = sourceSection( + paletteSource, + 'const addOpenCLIAdapter', + 'const generate', + ) + + assert.match( + addOpenCLIAdapter, + /materialization === "source_slot_ready"[\s\S]*?workflowCatalogItemForOpenCLIAdapterNode\(item\)/, + ) + assert.match( + catalogHookSource, + /node\.access === "read"[\s\S]*?workflowCatalogItemForOpenCLIAdapterNode\(node\)/, + ) + assert.equal(catalogItem.id, 'intelligence.source.opencli-slot') + assert.equal( + openCLIAdapterNodeMaterialization({ + status: 'preview_only', + runtimeReadiness: 'source_slot_ready', + manifest: { canvas: { materialization: 'source_slot_ready' } }, + }), + 'unavailable', + ) + + const project = parseWorkflowProject({ + id: 'opencli-statsgov-project', + name: 'OpenCLI statsgov project', + profile: 'intelligence', + nodes: [{ + id: 'manual-trigger', + kind: 'action', + capability: 'trigger', + params: {}, + }], + edges: [], + }) + const updated = addCatalogNodeToWorkflowProject( + project, + catalogItem, + 'source-opencli-statsgov-nbs', + { x: 320, y: 180 }, + ) + const sourceNode = updated.nodes.find((node) => node.id === 'source-opencli-statsgov-nbs') + + assert.equal(sourceNode?.ui?.catalogId, 'intelligence.source.opencli-slot') + assert.equal(sourceNode?.params.opencliAdapterNodeId, 'opencli.adapter.statsgov.nbs') + assert.ok(updated.adapters.some((adapter) => adapter.id === 'opencli-statsgov-nbs')) + + const persistedBadDraft = structuredClone(updated) + persistedBadDraft.nodes.find((node) => node.id === sourceNode.id).ui.catalogId = 'opencli.adapter.statsgov.nbs' + const rejected = compileWithBackend(persistedBadDraft) + assert.notEqual(rejected.status, 0) + assert.ok(rejected.report.errors.some((error) => error.code === 'unknown_node_library_binding')) + + const repaired = parseWorkflowProject(persistedBadDraft) + const repairedSource = repaired.nodes.find((node) => node.id === sourceNode.id) + assert.equal(repairedSource?.ui?.catalogId, 'intelligence.source.opencli-slot') + + const untrustedNearMiss = structuredClone(persistedBadDraft) + untrustedNearMiss.adapters.find((adapter) => adapter.id === 'opencli-statsgov-nbs').provider = 'http' + assert.equal( + parseWorkflowProject(untrustedNearMiss).nodes.find((node) => node.id === sourceNode.id)?.ui?.catalogId, + 'opencli.adapter.statsgov.nbs', + ) + + const compiled = compileWithBackend(repaired) + assert.equal(compiled.status, 0, `${compiled.stdout}\n${compiled.stderr}`) + assert.equal(compiled.report.valid, true) + assert.equal( + compiled.report.plan.runtime.nodes.find((node) => node.id === sourceNode.id).runtime.binding.binding_id, + 'iii.collector-opencli.snapshot', + ) +}) + test('OpenTabs tools become typed callable nodes and compile to the OpenTabs executor', async () => { const [{ workflowCatalogItemForOpenTabsToolNode }, { addCatalogNodeToWorkflowProject }, { parseWorkflowProject }] = await Promise.all([ importTypeScript('lib/workflow/backend-opentabs-tool-nodes.ts'), diff --git a/tests/integration/test_studio_lifecycle_api.py b/tests/integration/test_studio_lifecycle_api.py index 1bde089..cae4574 100644 --- a/tests/integration/test_studio_lifecycle_api.py +++ b/tests/integration/test_studio_lifecycle_api.py @@ -119,6 +119,57 @@ async def test_studio_workflow_draft_validation_run_is_persisted(client): assert run["runId"] +@pytest.mark.asyncio +async def test_studio_validation_rejects_an_isolated_source_node(client): + graph = workflow_conformance_project() + graph["nodes"].append( + { + "id": "isolated-http-source", + "kind": "source", + "capability": "fetch", + "adapter": "isolated-http-adapter", + "params": { + "channelType": "http", + "endpoint": "https://example.com/data", + "method": "GET", + }, + "ui": {"catalogId": "intelligence.source.http"}, + } + ) + graph["adapters"].append( + { + "id": "isolated-http-adapter", + "type": "source", + "provider": "http", + "mode": "live", + "config": {"channelType": "http"}, + } + ) + created = await _create_studio_workflow(client, graph=graph) + + response = await client.post( + f"{created['base_url']}/draft/validation-runs", + json={}, + ) + + assert response.status_code == 201, response.text + run = response.json()["data"] + assert run["status"] == "failed" + assert run["valid"] is False + assert run["errors"] == [ + { + "code": "isolated_source_node", + "message": ( + 'Workflow source node "isolated-http-source" is not connected ' + "to a downstream node" + ), + "node_id": "isolated-http-source", + "edge_id": None, + "path": ["nodes", "isolated-http-source"], + } + ] + + @pytest.mark.asyncio async def test_studio_workflow_current_validated_revision_can_be_published(client): created = await _create_studio_workflow(client) diff --git a/tests/integration/test_workflow_capabilities_api.py b/tests/integration/test_workflow_capabilities_api.py index e6f574a..bbac419 100644 --- a/tests/integration/test_workflow_capabilities_api.py +++ b/tests/integration/test_workflow_capabilities_api.py @@ -590,7 +590,7 @@ def test_opencli_adapter_nodes_classify_manifest_entries(monkeypatch): assert bbc.params == {"site": "bbc", "command": "news", "format": "json", "args": {}} twitter_search = nodes["opencli.adapter.twitter.search"] - assert twitter_search.status == "blocked" + assert twitter_search.status == "preview_only" assert twitter_search.catalogId == "intelligence.source.opencli-slot" assert twitter_search.requiredArgs == ["query"] assert twitter_search.presetKind == "source_slot" @@ -622,7 +622,7 @@ def test_opencli_adapter_nodes_classify_manifest_entries(monkeypatch): "capability": {"fetch": 2, "store": 1}, "access": {"read": 2, "write": 1}, "browser": {"non_browser": 1, "browser": 2}, - "status": {"runnable": 1, "blocked": 2}, + "status": {"runnable": 1, "blocked": 1, "preview_only": 1}, "presetKind": {"source_slot": 2, "tool_capability": 1}, "runtimeReadiness": { "source_slot_ready": 1, @@ -760,7 +760,7 @@ async def test_opencli_adapter_nodes_endpoint_filters_presets_and_returns_facets "capability": {"fetch": 1}, "access": {"read": 1}, "browser": {"browser": 1}, - "status": {"blocked": 1}, + "status": {"preview_only": 1}, "presetKind": {"source_slot": 1}, "runtimeReadiness": {"source_slot_requires_params": 1}, } diff --git a/tests/integration/test_workflow_compile_api.py b/tests/integration/test_workflow_compile_api.py index 5d063c5..054066b 100644 --- a/tests/integration/test_workflow_compile_api.py +++ b/tests/integration/test_workflow_compile_api.py @@ -944,6 +944,77 @@ async def test_compile_resolves_opencli_source_to_iii_runtime_binding(client): } +@pytest.mark.parametrize( + ("catalog_id", "provider", "channel_type", "params"), + [ + ( + "intelligence.source.http", + "http", + "http", + {"endpoint": "https://example.com/data", "method": "GET"}, + ), + ( + "intelligence.source.rss-bridge", + "rss", + "rss", + {"url": "https://rss-bridge.example/?action=display&format=Atom"}, + ), + ], +) +@pytest.mark.asyncio +async def test_compile_resolves_declared_source_presets_to_source_fetch_runtime( + client, + catalog_id: str, + provider: str, + channel_type: str, + params: dict, +): + project = _valid_workflow_project() + project["nodes"] = [ + { + "id": "source-preset", + "kind": "source", + "capability": "fetch", + "adapter": "source-preset-adapter", + "params": {**params, "channelType": channel_type}, + "ui": {"catalogId": catalog_id}, + } + ] + project["edges"] = [] + project["adapters"] = [ + { + "id": "source-preset-adapter", + "type": "source", + "provider": provider, + "mode": "live", + "config": {"channelType": channel_type}, + } + ] + + response = await client.post("/api/v1/workflows/compile", json={"project": project}) + + assert response.status_code == 200 + data = response.json()["data"] + assert data["valid"] is True, data["errors"] + source_node = data["plan"]["runtime"]["nodes"][0] + assert source_node["runtime"]["origin"] == { + "kind": "node_library", + "catalog_id": catalog_id, + "notes": [], + } + _assert_binding_includes( + source_node["runtime"]["binding"], + { + "status": "bound", + "binding_id": "workflow.source.fetch", + "runtime": "workflow", + "channel": "source", + }, + ) + assert source_node["runtime"]["binding"]["input"]["provider"] == provider + assert source_node["runtime"]["binding"]["input"]["channelType"] == channel_type + + @pytest.mark.asyncio async def test_compile_projects_native_first_loop_nodes_to_runtime_bindings(client): response = await client.post( diff --git a/tests/integration/test_workflow_opencli_hda_trace_api.py b/tests/integration/test_workflow_opencli_hda_trace_api.py index 80a3bef..5db0591 100644 --- a/tests/integration/test_workflow_opencli_hda_trace_api.py +++ b/tests/integration/test_workflow_opencli_hda_trace_api.py @@ -14,7 +14,7 @@ from backend.models.source import DataSource from backend.models.task import CollectionTask from backend.models.workflow_run import WorkflowRun, WorkflowRunEvent -from backend.workflow.opencli_hda_tracer import _RUNS +from backend.workflow.opencli_hda_tracer import _RUNS, _bounded_opencli_dispatch_result from tests.fixtures.workflow_conformance import workflow_conformance_project from tests.integration.test_workflow_compile_api import ( _nested_operator_project, @@ -162,6 +162,23 @@ def _multi_source_opencli_hda_project() -> dict: } +def test_opencli_dispatch_result_enforces_workflow_item_cap() -> None: + items, details = _bounded_opencli_dispatch_result( + [{"id": "1"}, {"id": "2"}, {"id": "3"}], + {"success": True, "itemCount": 3}, + max_items=1, + ) + + assert items == [{"id": "1"}] + assert details == { + "success": True, + "itemCount": 1, + "receivedItemCount": 3, + "maxItemsPerRun": 1, + "truncated": True, + } + + def _opencli_write_action_project( *, proposal_state: str, diff --git a/tests/integration/test_workflow_patch_api.py b/tests/integration/test_workflow_patch_api.py index 600b250..679d6f3 100644 --- a/tests/integration/test_workflow_patch_api.py +++ b/tests/integration/test_workflow_patch_api.py @@ -48,6 +48,29 @@ def _fixture_opencli_adapter_catalog() -> tuple[dict, ...]: ) +def _fixture_opencli_required_arg_catalog() -> tuple[dict, ...]: + return ( + { + "site": "example", + "name": "search", + "description": "Example search", + "access": "read", + "browser": False, + "strategy": "public", + "args": [ + { + "name": "query", + "type": "str", + "required": True, + "positional": True, + }, + {"name": "limit", "type": "int", "required": False}, + ], + "columns": ["id", "text"], + }, + ) + + @pytest.mark.asyncio async def test_patch_adds_existing_node_updates_params_connects_and_compiles(client): project = _valid_workflow_project() @@ -335,7 +358,7 @@ async def test_patch_materializes_opencli_required_arg_adapter_with_params( ): monkeypatch.setattr( "backend.workflow.opencli_adapter_nodes._load_opencli_catalog", - _fixture_opencli_adapter_catalog, + _fixture_opencli_required_arg_catalog, ) project = _valid_workflow_project() @@ -346,7 +369,7 @@ async def test_patch_materializes_opencli_required_arg_adapter_with_params( "operations": [ { "op": "materialize_opencli_adapter", - "adapterNodeId": "opencli.adapter.twitter.search", + "adapterNodeId": "opencli.adapter.example.search", "nodeId": "source-x-openai", "params": {"query": "openai", "limit": 1}, } @@ -359,7 +382,7 @@ async def test_patch_materializes_opencli_required_arg_adapter_with_params( assert data["valid"] is True nodes = {node["id"]: node for node in data["project"]["nodes"]} node = nodes["source-x-openai"] - assert node["adapter"] == "opencli-twitter" + assert node["adapter"] == "opencli-example" assert node["params"]["positional_args"] == ["openai"] assert node["params"]["args"] == {"limit": 1} @@ -374,7 +397,7 @@ async def test_patch_materialize_opencli_required_arg_reports_missing_params( ): monkeypatch.setattr( "backend.workflow.opencli_adapter_nodes._load_opencli_catalog", - _fixture_opencli_adapter_catalog, + _fixture_opencli_required_arg_catalog, ) project = _valid_workflow_project() @@ -385,7 +408,7 @@ async def test_patch_materialize_opencli_required_arg_reports_missing_params( "operations": [ { "op": "materialize_opencli_adapter", - "adapterNodeId": "opencli.adapter.twitter.search", + "adapterNodeId": "opencli.adapter.example.search", } ], }, @@ -401,7 +424,7 @@ async def test_patch_materialize_opencli_required_arg_reports_missing_params( { "capability": "opencli.adapter.params", "reason": "Missing OpenCLI adapter params: query", - "n8n_search_hint": "opencli.adapter.twitter.search", + "n8n_search_hint": "opencli.adapter.example.search", } ] @@ -582,6 +605,19 @@ async def test_demand_draft_assembles_multi_source_need_through_merge(client): nodes = {node["id"]: node for node in data["project"]["nodes"]} assert "source-xiaohongshu" in nodes assert "source-bilibili" in nodes + assert nodes["source-xiaohongshu"]["params"]["args"] == {"keyword": "AI"} + assert nodes["source-bilibili"]["params"]["args"] == {"keyword": "AI"} + assert data["missing_capabilities"] == [ + { + "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." + ), + "n8n_search_hint": "collection.rank.popularity", + } + ] edges = { (edge["source"], edge["target"], edge.get("targetPort")) for edge in data["project"]["edges"] diff --git a/tests/unit/test_demand_assembler.py b/tests/unit/test_demand_assembler.py index cbcb843..98c60ed 100644 --- a/tests/unit/test_demand_assembler.py +++ b/tests/unit/test_demand_assembler.py @@ -217,7 +217,7 @@ def test_legacy_keyword_slots_unchanged_for_both_known_sites(monkeypatch): "sourceGroup": "social", "site": "xiaohongshu", "command": "search", - "args": {"keyword": "和 AI"}, + "args": {"keyword": "AI"}, }, { "id": "bilibili", @@ -225,11 +225,22 @@ def test_legacy_keyword_slots_unchanged_for_both_known_sites(monkeypatch): "sourceGroup": "video", "site": "bilibili", "command": "search", - "args": {"keyword": "和 AI"}, + "args": {"keyword": "AI"}, }, ] +def test_keyword_excludes_platform_glue_and_downstream_processing(monkeypatch): + text = "抓小红书和B站的AI热帖，清洗去重后保存" + + slots = _legacy_keyword_slots_for_need(text) + + assert [slot["args"] for slot in slots] == [ + {"keyword": "AI"}, + {"keyword": "AI"}, + ] + + def test_known_site_aliases_ignore_unrelated_catalog_description_matches(monkeypatch): _patch_catalog( monkeypatch, diff --git a/tests/unit/test_main.py b/tests/unit/test_main.py index 78be5b8..b8455e4 100644 --- a/tests/unit/test_main.py +++ b/tests/unit/test_main.py @@ -113,3 +113,36 @@ async def _check(): schema = response.json() assert "openapi" in schema assert "paths" in schema + assert schema["components"]["securitySchemes"]["BearerAuth"] == { + "type": "http", + "scheme": "bearer", + "description": "Operator-provisioned OpenCLI Admin API token.", + } + assert schema["paths"]["/api/v1/workflows/capabilities"]["get"]["security"] == [ + {"BearerAuth": []} + ] + assert schema["x-opencli-agent"]["mcp"]["url"] == "/mcp" + + +def test_root_discovers_agent_interfaces(client): + """GET / gives a new Agent the authentication and workflow entry sequence.""" + import asyncio + + from httpx import ASGITransport, AsyncClient + + from backend.main import app + + async def _check(): + async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as ac: + return await ac.get("/") + + response = asyncio.run(_check()) + assert response.status_code == 200 + data = response.json() + assert data["interfaces"]["mcp"]["url"] == "/mcp" + assert data["authentication"]["scheme"] == "bearer" + assert data["agentWorkflow"][:3] == [ + "discover capabilities", + "arrange a review-only node draft", + "compile and preflight", + ] diff --git a/tests/unit/test_mcp_server.py b/tests/unit/test_mcp_server.py index ae43a94..c9ab849 100644 --- a/tests/unit/test_mcp_server.py +++ b/tests/unit/test_mcp_server.py @@ -17,6 +17,10 @@ async def test_modern_protocol_discovers_structured_project_tools(): assert result.result_type == "complete" assert result.cache_scope == "private" assert { + "list_workflow_node_capabilities", + "draft_workflow_from_intent", + "preview_workflow_node_patch", + "compile_workflow_draft", "list_project_workflows", "run_published_workflow", "get_project_runtime_summary", @@ -25,6 +29,8 @@ async def test_modern_protocol_discovers_structured_project_tools(): } <= tools.keys() assert tools["run_published_workflow"].output_schema["type"] == "object" assert tools["run_published_workflow"].annotations.idempotent_hint is True + assert tools["draft_workflow_from_intent"].annotations.read_only_hint is True + assert tools["compile_workflow_draft"].annotations.read_only_hint is True assert tools["list_project_runtime_logs"].annotations.read_only_hint is True @@ -53,3 +59,38 @@ async def test_published_workflow_tool_reuses_real_project_run_endpoint(monkeypa "user": "agent-1", }, ) + + +@pytest.mark.asyncio +async def test_agent_workflow_tools_reuse_stateless_rest_contracts(monkeypatch): + request = AsyncMock(return_value={"success": True, "data": {"valid": True}}) + monkeypatch.setattr(mcp_server, "_request", request) + + await mcp_server.list_workflow_node_capabilities() + request.assert_awaited_with("GET", "/api/v1/workflows/capabilities") + + request.reset_mock() + drafted = await mcp_server.draft_workflow_from_intent("抓小红书热帖") + assert drafted["data"]["valid"] is True + _, kwargs = request.await_args + assert request.await_args.args[:2] == ("POST", "/api/v1/workflows/demand-draft") + assert kwargs["json"]["text"] == "抓小红书热帖" + seed = kwargs["json"]["project"] + assert seed["nodes"][0]["ui"]["catalogId"] == "intelligence.input.collection-need" + assert seed["agentPermissions"]["canMutateExternalSites"] is False + + request.reset_mock() + await mcp_server.preview_workflow_node_patch(seed, [{"op": "add_node"}]) + request.assert_awaited_once_with( + "POST", + "/api/v1/workflows/patch", + json={"project": seed, "operations": [{"op": "add_node"}]}, + ) + + request.reset_mock() + await mcp_server.compile_workflow_draft(seed) + request.assert_awaited_once_with( + "POST", + "/api/v1/workflows/compile", + json={"project": seed}, + ) diff --git a/tests/unit/test_opencli_adapter_nodes.py b/tests/unit/test_opencli_adapter_nodes.py index b82b0ae..226b202 100644 --- a/tests/unit/test_opencli_adapter_nodes.py +++ b/tests/unit/test_opencli_adapter_nodes.py @@ -2,6 +2,8 @@ from subprocess import CompletedProcess +import pytest + from backend.workflow import opencli_adapter_nodes @@ -17,3 +19,68 @@ def test_opencli_catalog_fails_closed_when_decoding_produces_no_stdout(monkeypat assert opencli_adapter_nodes._load_opencli_catalog() == () finally: opencli_adapter_nodes._load_opencli_catalog.cache_clear() + + +def _catalog_entry(**overrides) -> dict: + return { + "site": "example", + "name": "list", + "description": "Example public reader", + "access": "read", + "strategy": "public", + "browser": False, + "args": [], + **overrides, + } + + +def test_public_adapter_without_runtime_dependencies_remains_runnable() -> None: + node = opencli_adapter_nodes._build_adapter_node(_catalog_entry()) + + assert node.status == "runnable" + assert node.runtimeReadiness == "source_slot_ready" + assert node.manifest["canvas"]["runBlocked"] is False + assert node.manifest["availability"] == {"available": True, "reason": None} + + +@pytest.mark.parametrize( + ("entry", "reason"), + [ + ( + _catalog_entry(site="sse", name="company-list"), + "upstream_http_404", + ), + ( + _catalog_entry(site="browser-site", name="feed", browser=True), + "browser_session_readiness_unverified", + ), + ( + _catalog_entry(site="cookie-site", name="feed", strategy="cookie"), + "cookie_readiness_unverified", + ), + ], +) +def test_unverified_adapter_dependencies_fail_closed(entry: dict, reason: str) -> None: + node = opencli_adapter_nodes._build_adapter_node(entry) + + assert node.status == "preview_only" + assert node.manifest["canvas"]["runBlocked"] is True + assert node.manifest["availability"] == {"available": False, "reason": reason} + + +def test_known_unavailable_adapter_cannot_materialize(monkeypatch) -> None: + monkeypatch.setattr( + opencli_adapter_nodes, + "get_opencli_adapter_catalog", + lambda **kwargs: (_catalog_entry(site="sse", name="company-list"),), + ) + + with pytest.raises( + opencli_adapter_nodes.OpenCLIAdapterNodeMaterializationError, + match="not currently runnable", + ) as error: + opencli_adapter_nodes.materialize_opencli_adapter_node( + "opencli.adapter.sse.company-list" + ) + + assert error.value.code == "opencli_adapter_node_unavailable"