fix(backend): filter graph-only blocks from CoPilot's find_block results

Exclude blocks that are unsuitable for standalone execution from CoPilot's
block search and execution:

- Filter by BlockType: INPUT, OUTPUT, WEBHOOK, WEBHOOK_MANUAL, NOTE,
  HUMAN_IN_THE_LOOP, AGENT
- Filter by ID: SmartDecisionMakerBlock (requires graph topology)

Changes:
- find_block.py: Skip excluded blocks in search results
- run_block.py: Prevent execution of excluded blocks with clear error
- content_handlers.py: Don't index excluded blocks for search

This does NOT affect the Builder UI which uses load_all_blocks() directly.

Resolves: SECRT-1831

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
Nicholas Tindle
2026-01-29 19:44:25 -06:00
parent e0dfae5732
commit 1780f8f1be
3 changed files with 129 additions and 56 deletions

View File

@@ -13,10 +13,28 @@ from backend.api.features.chat.tools.models import (
NoResultsResponse,
)
from backend.api.features.store.hybrid_search import unified_hybrid_search
from backend.data.block import get_block
from backend.data.block import BlockType, get_block
logger = logging.getLogger(__name__)
# Blocks excluded from CoPilot standalone execution
# NOTE: This does NOT affect the Builder UI which uses load_all_blocks() directly
EXCLUDED_BLOCK_TYPES = {
BlockType.INPUT, # Graph interface definition - data enters via chat, not graph inputs
BlockType.OUTPUT, # Graph interface definition - data exits via chat, not graph outputs
BlockType.WEBHOOK, # Wait for external events - would hang forever in CoPilot
BlockType.WEBHOOK_MANUAL, # Same as WEBHOOK
BlockType.NOTE, # Visual annotation only - no runtime behavior
BlockType.HUMAN_IN_THE_LOOP, # Pauses for human approval - CoPilot IS human-in-the-loop
BlockType.AGENT, # AgentExecutorBlock requires execution_context - use run_agent tool
}
# Blocks that have STANDARD/other types but still require graph context
EXCLUDED_BLOCK_IDS = {
# SmartDecisionMakerBlock - dynamically discovers downstream blocks via graph topology
"3b191d9f-356f-482d-8238-ba04b6d18381",
}
class FindBlockTool(BaseTool):
"""Tool for searching available blocks."""
@@ -108,61 +126,70 @@ class FindBlockTool(BaseTool):
block = get_block(block_id)
# Skip disabled blocks
if block and not block.disabled:
# Get input/output schemas
input_schema = {}
output_schema = {}
try:
input_schema = block.input_schema.jsonschema()
except Exception:
pass
try:
output_schema = block.output_schema.jsonschema()
except Exception:
pass
if not block or block.disabled:
continue
# Get categories from block instance
categories = []
if hasattr(block, "categories") and block.categories:
categories = [cat.value for cat in block.categories]
# Skip blocks excluded from CoPilot (graph-only blocks)
if (
block.block_type in EXCLUDED_BLOCK_TYPES
or block.id in EXCLUDED_BLOCK_IDS
):
continue
# Extract required inputs for easier use
required_inputs: list[BlockInputFieldInfo] = []
if input_schema:
properties = input_schema.get("properties", {})
required_fields = set(input_schema.get("required", []))
# Get credential field names to exclude from required inputs
credentials_fields = set(
block.input_schema.get_credentials_fields().keys()
)
# Get input/output schemas
input_schema = {}
output_schema = {}
try:
input_schema = block.input_schema.jsonschema()
except Exception:
pass
try:
output_schema = block.output_schema.jsonschema()
except Exception:
pass
for field_name, field_schema in properties.items():
# Skip credential fields - they're handled separately
if field_name in credentials_fields:
continue
# Get categories from block instance
categories = []
if hasattr(block, "categories") and block.categories:
categories = [cat.value for cat in block.categories]
required_inputs.append(
BlockInputFieldInfo(
name=field_name,
type=field_schema.get("type", "string"),
description=field_schema.get("description", ""),
required=field_name in required_fields,
default=field_schema.get("default"),
)
)
blocks.append(
BlockInfoSummary(
id=block_id,
name=block.name,
description=block.description or "",
categories=categories,
input_schema=input_schema,
output_schema=output_schema,
required_inputs=required_inputs,
)
# Extract required inputs for easier use
required_inputs: list[BlockInputFieldInfo] = []
if input_schema:
properties = input_schema.get("properties", {})
required_fields = set(input_schema.get("required", []))
# Get credential field names to exclude from required inputs
credentials_fields = set(
block.input_schema.get_credentials_fields().keys()
)
for field_name, field_schema in properties.items():
# Skip credential fields - they're handled separately
if field_name in credentials_fields:
continue
required_inputs.append(
BlockInputFieldInfo(
name=field_name,
type=field_schema.get("type", "string"),
description=field_schema.get("description", ""),
required=field_name in required_fields,
default=field_schema.get("default"),
)
)
blocks.append(
BlockInfoSummary(
id=block_id,
name=block.name,
description=block.description or "",
categories=categories,
input_schema=input_schema,
output_schema=output_schema,
required_inputs=required_inputs,
)
)
if not blocks:
return NoResultsResponse(
message=f"No blocks found for '{query}'",

View File

@@ -5,6 +5,10 @@ from collections import defaultdict
from typing import Any
from backend.api.features.chat.model import ChatSession
from backend.api.features.chat.tools.find_block import (
EXCLUDED_BLOCK_IDS,
EXCLUDED_BLOCK_TYPES,
)
from backend.data.block import get_block
from backend.data.execution import ExecutionContext
from backend.data.model import CredentialsMetaInput
@@ -182,6 +186,16 @@ class RunBlockTool(BaseTool):
session_id=session_id,
)
# Check if block is excluded from CoPilot (graph-only blocks)
if block.block_type in EXCLUDED_BLOCK_TYPES or block.id in EXCLUDED_BLOCK_IDS:
return ErrorResponse(
message=(
f"Block '{block.name}' cannot be run directly in CoPilot. "
"This block is designed for use within graphs only."
),
session_id=session_id,
)
logger.info(f"Executing block {block.name} ({block_id}) for user {user_id}")
# Check credentials

View File

@@ -13,10 +13,29 @@ from typing import Any
from prisma.enums import ContentType
from backend.data.block import BlockType
from backend.data.db import query_raw_with_schema
logger = logging.getLogger(__name__)
# Blocks excluded from CoPilot standalone execution and search indexing
# NOTE: This does NOT affect the Builder UI which uses load_all_blocks() directly
EXCLUDED_BLOCK_TYPES = {
BlockType.INPUT, # Graph interface definition - data enters via chat, not graph inputs
BlockType.OUTPUT, # Graph interface definition - data exits via chat, not graph outputs
BlockType.WEBHOOK, # Wait for external events - would hang forever in CoPilot
BlockType.WEBHOOK_MANUAL, # Same as WEBHOOK
BlockType.NOTE, # Visual annotation only - no runtime behavior
BlockType.HUMAN_IN_THE_LOOP, # Pauses for human approval - CoPilot IS human-in-the-loop
BlockType.AGENT, # AgentExecutorBlock requires execution_context - use run_agent tool
}
# Blocks that have STANDARD/other types but still require graph context
EXCLUDED_BLOCK_IDS = {
# SmartDecisionMakerBlock - dynamically discovers downstream blocks via graph topology
"3b191d9f-356f-482d-8238-ba04b6d18381",
}
@dataclass
class ContentItem:
@@ -192,6 +211,13 @@ class BlockHandler(ContentHandler):
if block_instance.disabled:
continue
# Skip blocks excluded from CoPilot (graph-only blocks)
if (
block_instance.block_type in EXCLUDED_BLOCK_TYPES
or block_id in EXCLUDED_BLOCK_IDS
):
continue
# Build searchable text from block metadata
parts = []
if hasattr(block_instance, "name") and block_instance.name:
@@ -253,12 +279,18 @@ class BlockHandler(ContentHandler):
all_blocks = get_blocks()
# Filter out disabled blocks - they're not indexed
enabled_block_ids = [
block_id
for block_id, block_cls in all_blocks.items()
if not block_cls().disabled
]
# Filter out disabled blocks and excluded blocks - they're not indexed
enabled_block_ids = []
for block_id, block_cls in all_blocks.items():
block_instance = block_cls()
if block_instance.disabled:
continue
if (
block_instance.block_type in EXCLUDED_BLOCK_TYPES
or block_id in EXCLUDED_BLOCK_IDS
):
continue
enabled_block_ids.append(block_id)
total_blocks = len(enabled_block_ids)
if total_blocks == 0: