diff --git a/backend/alembic/versions/00091_add_conversation_search_indexes.py b/backend/alembic/versions/00091_add_conversation_search_indexes.py index c5ed27ef0..e0f01d982 100644 --- a/backend/alembic/versions/00091_add_conversation_search_indexes.py +++ b/backend/alembic/versions/00091_add_conversation_search_indexes.py @@ -1,15 +1,19 @@ """add indexes to speed up conversation search/filter -Targets the slow conversation search queries (topic/summary ILIKE, correlated -EXISTS on conversation_analysis, ORDER BY created_at DESC pagination). +Targets the slow conversation search queries (correlated EXISTS on +conversation_analysis, ORDER BY created_at DESC pagination). Adds: - b-tree index on conversation_analysis.conversation_id (FK had no index -> the correlated EXISTS subquery was doing a seq scan per conversation row) -- pg_trgm GIN indexes on conversations.topic and conversation_analysis.summary so - leading-wildcard ILIKE '%x%' can use an index instead of a seq scan - partial b-tree on conversations.created_at DESC (live rows) for ORDER BY + LIMIT +Note: topic/summary ILIKE '%x%' searches are intentionally NOT indexed here. +A leading-wildcard ILIKE can only be index-backed via a pg_trgm GIN index, and +creating that extension requires CREATE privilege on the database which is not +available in all (e.g. managed prod) environments. Those searches continue to +run as sequential scans. + All indexes are built with CREATE INDEX CONCURRENTLY so no write locks are taken on prod tables. CONCURRENTLY cannot run inside a transaction, so we use an autocommit block (Alembic wraps each migration in a transaction by default). @@ -32,10 +36,6 @@ def upgrade() -> None: - # pg_trgm can be created inside the normal transaction; it enables the - # gin_trgm_ops operator class used by the topic/summary indexes below. - op.execute("CREATE EXTENSION IF NOT EXISTS pg_trgm") - # CREATE INDEX CONCURRENTLY must run outside a transaction. with op.get_context().autocommit_block(): # (1) Critical: index the FK column the correlated EXISTS subquery uses. @@ -47,23 +47,7 @@ def upgrade() -> None: """ ) - # (2) Trigram indexes so ILIKE '%term%' is index-backed. - op.execute( - """ - CREATE INDEX CONCURRENTLY IF NOT EXISTS - ix_conversations_topic_trgm - ON conversations USING gin (topic gin_trgm_ops) - """ - ) - op.execute( - """ - CREATE INDEX CONCURRENTLY IF NOT EXISTS - ix_conversation_analysis_summary_trgm - ON conversation_analysis USING gin (summary gin_trgm_ops) - """ - ) - - # (3) Support ORDER BY created_at DESC + pagination for live rows. + # (2) Support ORDER BY created_at DESC + pagination for live rows. op.execute( """ CREATE INDEX CONCURRENTLY IF NOT EXISTS @@ -80,13 +64,6 @@ def downgrade() -> None: op.execute( "DROP INDEX CONCURRENTLY IF EXISTS ix_conversations_created_at_desc" ) - op.execute( - "DROP INDEX CONCURRENTLY IF EXISTS ix_conversation_analysis_summary_trgm" - ) - op.execute( - "DROP INDEX CONCURRENTLY IF EXISTS ix_conversations_topic_trgm" - ) op.execute( "DROP INDEX CONCURRENTLY IF EXISTS ix_conversation_analysis_conversation_id" ) - # Leave the pg_trgm extension in place; other objects may depend on it. diff --git a/backend/alembic/versions/00095_add_conversation_to_test_cases.py b/backend/alembic/versions/00095_add_conversation_to_test_cases.py new file mode 100644 index 000000000..d2c18a81a --- /dev/null +++ b/backend/alembic/versions/00095_add_conversation_to_test_cases.py @@ -0,0 +1,55 @@ +"""add_conversation_to_test_cases + +Links each test case to the conversation it was imported from and its position +within that conversation, so evaluation runs can replay each conversation as an +isolated memory thread. + +Both columns are nullable: legacy and manually created cases have no source +conversation and are treated as independent single-turn conversations. + +Revision ID: 4e1c7a9d2b05 +Revises: 322883fb3a4d +Create Date: 2026-07-20 10:00:00.000000 + +""" + +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa +from sqlalchemy.dialects import postgresql + + +revision: str = "4e1c7a9d2b05" +down_revision: Union[str, None] = "322883fb3a4d" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.add_column( + "test_cases", + sa.Column("source_conversation_id", postgresql.UUID(as_uuid=True), nullable=True), + ) + op.add_column( + "test_cases", + sa.Column("turn_index", sa.Integer(), nullable=True), + ) + op.create_index( + "ix_test_cases_source_conversation_id", + "test_cases", + ["source_conversation_id"], + ) + # Execution groups by conversation and orders by turn within it. + op.create_index( + "ix_test_cases_conversation_turn", + "test_cases", + ["suite_id", "source_conversation_id", "turn_index"], + ) + + +def downgrade() -> None: + op.drop_index("ix_test_cases_conversation_turn", table_name="test_cases") + op.drop_index("ix_test_cases_source_conversation_id", table_name="test_cases") + op.drop_column("test_cases", "turn_index") + op.drop_column("test_cases", "source_conversation_id") diff --git a/backend/alembic/versions/00096_backfill_test_case_conversations.py b/backend/alembic/versions/00096_backfill_test_case_conversations.py new file mode 100644 index 000000000..60f4da7cc --- /dev/null +++ b/backend/alembic/versions/00096_backfill_test_case_conversations.py @@ -0,0 +1,169 @@ +"""backfill_test_case_conversations + +Recovers the source conversation of previously imported test cases. + +A suite's imported cases are treated as one cluster and linked only when exactly +one live conversation yields precisely the same ordered question/answer exchanges, +rebuilt with the same speaker roles and message order the importer used. Anything +ambiguous, partial or unmatched stays null and is logged for manual review; no +synthetic conversation id is ever stored. + +Revision ID: 8b2f5c31d740 +Revises: 4e1c7a9d2b05 +Create Date: 2026-07-20 10:30:00.000000 + +""" + +import logging +from typing import Any, List, Sequence, Tuple, Union + +from alembic import op +import sqlalchemy as sa + + +revision: str = "8b2f5c31d740" +down_revision: Union[str, None] = "4e1c7a9d2b05" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + +logger = logging.getLogger("alembic.runtime.migration") + +# Frozen copies of the importer's speaker roles: this migration must keep matching +# the data as it was written, even if the application constants later change. +_CUSTOMER_SPEAKERS = {"customer", "user"} +_AGENT_SPEAKERS = {"agent", "assistant", "bot"} + +_IMPORTED_CASES = sa.text( + """ + SELECT id, + input_data->>'message' AS message, + expected_output->>'value' AS answer + FROM test_cases + WHERE suite_id = :suite_id + AND is_deleted = 0 + AND tags @> '["imported"]'::jsonb + ORDER BY created_at, id + """ +) + +# Live conversations whose live messages include this text from a customer. +_CANDIDATE_CONVERSATIONS = sa.text( + """ + SELECT DISTINCT message.conversation_id + FROM transcript_messages AS message + JOIN conversations AS conversation + ON conversation.id = message.conversation_id + AND conversation.is_deleted = 0 + WHERE message.is_deleted = 0 + AND message.text = :question + AND lower(message.speaker) = ANY(:customer_speakers) + """ +) + +_CONVERSATION_MESSAGES = sa.text( + """ + SELECT speaker, text + FROM transcript_messages + WHERE conversation_id = :conversation_id + AND is_deleted = 0 + ORDER BY sequence_number + """ +) + + +def _exchanges_of(bind, conversation_id: Any) -> List[Tuple[str, str]]: + """Rebuild a conversation's ordered question/answer pairs, as the importer did.""" + rows = bind.execute( + _CONVERSATION_MESSAGES, {"conversation_id": conversation_id} + ).all() + + exchanges: List[Tuple[str, str]] = [] + pending_question: str | None = None + for row in rows: + speaker = (row.speaker or "").lower() + if speaker in _CUSTOMER_SPEAKERS: + pending_question = row.text + elif speaker in _AGENT_SPEAKERS and pending_question is not None: + exchanges.append((pending_question, row.text)) + pending_question = None + return exchanges + + +def upgrade() -> None: + bind = op.get_bind() + + suite_ids = bind.execute( + sa.text( + """ + SELECT DISTINCT suite_id + FROM test_cases + WHERE is_deleted = 0 + AND source_conversation_id IS NULL + AND tags @> '["imported"]'::jsonb + """ + ) + ).scalars().all() + + linked = 0 + skipped: list[tuple[str, str]] = [] + + for suite_id in suite_ids: + cases = bind.execute(_IMPORTED_CASES, {"suite_id": suite_id}).all() + cluster = [(case.message, case.answer) for case in cases] + + if not cluster or not all(question and answer for question, answer in cluster): + skipped.append((str(suite_id), "cluster has an incomplete exchange")) + continue + + candidate_ids = bind.execute( + _CANDIDATE_CONVERSATIONS, + {"question": cluster[0][0], "customer_speakers": list(_CUSTOMER_SPEAKERS)}, + ).scalars().all() + + matches = [ + candidate_id + for candidate_id in candidate_ids + if _exchanges_of(bind, candidate_id) == cluster + ] + + if len(matches) != 1: + skipped.append( + ( + str(suite_id), + f"{len(matches)} conversations reproduce the cluster exactly", + ) + ) + continue + + # Turn order comes from the transcript rebuild, not from row creation order. + for turn_index, case in enumerate(cases): + bind.execute( + sa.text( + """ + UPDATE test_cases + SET source_conversation_id = :conversation_id, turn_index = :turn_index + WHERE id = :case_id + """ + ), + { + "conversation_id": matches[0], + "turn_index": turn_index, + "case_id": case.id, + }, + ) + linked += len(cases) + + logger.info("Linked %s imported cases to their source conversation", linked) + for suite_id, reason in skipped: + logger.warning( + "Suite %s left unlinked for manual review: %s", suite_id, reason + ) + + +def downgrade() -> None: + """Deliberately a no-op. + + Clearing the columns cannot distinguish values recovered here from those the + importer set for real imports, and the latter are unrecoverable. The preceding + migration drops both columns, so nothing is left behind either way. + """ diff --git a/backend/alembic/versions/00097_unique_active_conversation_turns.py b/backend/alembic/versions/00097_unique_active_conversation_turns.py new file mode 100644 index 000000000..77ef4bb84 --- /dev/null +++ b/backend/alembic/versions/00097_unique_active_conversation_turns.py @@ -0,0 +1,39 @@ +"""unique_active_conversation_turns + +Guards against a conversation being imported twice into the same dataset, which +would produce duplicate turn positions and replay the conversation out of order. + +Only active rows are constrained. Manual and legacy cases keep a null +source_conversation_id and are unaffected, since nulls never conflict. + +Revision ID: c5d9e0a71b38 +Revises: 8b2f5c31d740 +Create Date: 2026-07-21 09:00:00.000000 + +""" + +from typing import Sequence, Union + +from alembic import op + + +revision: str = "c5d9e0a71b38" +down_revision: Union[str, None] = "8b2f5c31d740" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + +_INDEX_NAME = "uq_test_cases_active_conversation_turn" + + +def upgrade() -> None: + op.create_index( + _INDEX_NAME, + "test_cases", + ["suite_id", "source_conversation_id", "turn_index"], + unique=True, + postgresql_where="is_deleted = 0 AND source_conversation_id IS NOT NULL", + ) + + +def downgrade() -> None: + op.drop_index(_INDEX_NAME, table_name="test_cases") diff --git a/backend/alembic/versions/00098_add_status_to_test_results.py b/backend/alembic/versions/00098_add_status_to_test_results.py new file mode 100644 index 000000000..7d750b072 --- /dev/null +++ b/backend/alembic/versions/00098_add_status_to_test_results.py @@ -0,0 +1,47 @@ +"""add_status_to_test_results + +Records why a case did or did not produce metrics, so skipped cases, execution +failures and scoring failures can be told apart instead of all reading as +"no metrics". + +Legacy rows that carry metrics are marked scored. Legacy rows without metrics +keep a null status: the stored data cannot prove whether they were skipped, +execution failures or scoring failures, and the reader treats null as unscored. + +Revision ID: e7a4b0c95d61 +Revises: c5d9e0a71b38 +Create Date: 2026-07-21 16:30:00.000000 + +""" + +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa + + +revision: str = "e7a4b0c95d61" +down_revision: Union[str, None] = "c5d9e0a71b38" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.add_column( + "test_results", + sa.Column("status", sa.String(32), nullable=True), + ) + # Only rows holding real metrics can be proven to have scored. + op.execute( + """ + UPDATE test_results + SET status = 'scored' + WHERE metrics IS NOT NULL + AND jsonb_typeof(metrics) = 'object' + AND metrics <> '{}'::jsonb + """ + ) + + +def downgrade() -> None: + op.drop_column("test_results", "status") diff --git a/backend/alembic/versions/00099_add_template_marketplace.py b/backend/alembic/versions/00099_add_template_marketplace.py new file mode 100644 index 000000000..3e4df18eb --- /dev/null +++ b/backend/alembic/versions/00099_add_template_marketplace.py @@ -0,0 +1,76 @@ +"""add template marketplace (templates table + feature flag) + +Consolidated migration for the Template Marketplace feature: the full +``templates`` table (private + published/community lifecycle + install +tracking) plus a ``feature.templateMarketplace`` feature flag seeded hidden +(``val='false'``) so the feature is off until an admin enables it. + +Revision ID: d4b1f2a3c5e6 +Revises: e7a4b0c95d61 +Create Date: 2026-07-21 00:00:00.000000 + +""" +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa +from sqlalchemy.dialects import postgresql + +# revision identifiers, used by Alembic. +revision: str = 'd4b1f2a3c5e6' +down_revision: Union[str, None] = 'e7a4b0c95d61' +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + +FEATURE_FLAG_KEY = "feature.templateMarketplace" + + +def upgrade() -> None: + op.create_table( + 'templates', + sa.Column('id', sa.UUID(), nullable=False), + sa.Column('title', sa.String(length=120), nullable=False), + sa.Column('description', sa.String(length=500), nullable=True), + sa.Column('category', sa.String(length=60), nullable=True), + sa.Column('icon', sa.String(length=60), nullable=True), + sa.Column('tags', postgresql.JSONB(astext_type=sa.Text()), nullable=True), + sa.Column('node_types', postgresql.JSONB(astext_type=sa.Text()), nullable=True), + sa.Column('graph', postgresql.JSONB(astext_type=sa.Text()), nullable=False), + sa.Column('agent_config', postgresql.JSONB(astext_type=sa.Text()), nullable=True), + sa.Column('is_official', sa.Boolean(), server_default=sa.text('false'), nullable=False), + sa.Column('install_count', sa.Integer(), server_default='0', nullable=False), + sa.Column('status', sa.String(length=20), server_default='private', nullable=False), + sa.Column('source_tenant', sa.String(length=120), nullable=True), + sa.Column('published_by', sa.UUID(), nullable=True), + sa.Column('source_template_id', sa.UUID(), nullable=True), + sa.Column('approved_by', sa.UUID(), nullable=True), + sa.Column('approved_at', sa.DateTime(timezone=True), nullable=True), + sa.Column('rejection_reason', sa.String(length=500), nullable=True), + sa.Column('created_by', sa.UUID(), nullable=True), + sa.Column('updated_by', sa.UUID(), nullable=True), + sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('CURRENT_TIMESTAMP'), nullable=True), + sa.Column('updated_at', sa.DateTime(timezone=True), server_default=sa.text('CURRENT_TIMESTAMP'), nullable=True), + sa.Column('is_deleted', sa.Integer(), server_default=sa.text('0'), nullable=False), + sa.PrimaryKeyConstraint('id'), + ) + op.create_index(op.f('ix_templates_status'), 'templates', ['status'], unique=False) + op.create_index(op.f('ix_templates_source_template_id'), 'templates', ['source_template_id'], unique=False) + op.create_index(op.f('ix_templates_created_by'), 'templates', ['created_by'], unique=False) + + # Seed the feature flag hidden by default (idempotent on the unique key). + op.execute( + f""" + INSERT INTO feature_flags (id, key, val, description, is_active, is_deleted) + VALUES (gen_random_uuid(), '{FEATURE_FLAG_KEY}', 'false', + 'Template Marketplace (browse, publish, and install agent templates)', 1, 0) + ON CONFLICT (key) DO NOTHING + """ + ) + + +def downgrade() -> None: + op.execute(f"DELETE FROM feature_flags WHERE key = '{FEATURE_FLAG_KEY}'") + op.drop_index(op.f('ix_templates_created_by'), table_name='templates') + op.drop_index(op.f('ix_templates_source_template_id'), table_name='templates') + op.drop_index(op.f('ix_templates_status'), table_name='templates') + op.drop_table('templates') diff --git a/backend/app/api/v1/routes/__init__.py b/backend/app/api/v1/routes/__init__.py index e46f8af91..0e6bd237f 100644 --- a/backend/app/api/v1/routes/__init__.py +++ b/backend/app/api/v1/routes/__init__.py @@ -46,6 +46,7 @@ tenants, test_cases, test_evaluations, + templates, test_runs, test_suites, translations, @@ -115,6 +116,7 @@ agent_knowledge.router, prefix="/genagent/knowledge", tags=["Knowledge Base"] ) router.include_router(workflows.router, prefix="/genagent/workflow", tags=["Workflows"]) +router.include_router(templates.router, prefix="/genagent/templates", tags=["Templates"]) router.include_router( workflow_schedule.router, prefix="/genagent/workflow-schedules", diff --git a/backend/app/api/v1/routes/templates.py b/backend/app/api/v1/routes/templates.py new file mode 100644 index 000000000..8e2427cbc --- /dev/null +++ b/backend/app/api/v1/routes/templates.py @@ -0,0 +1,197 @@ +from typing import List +from uuid import UUID + +from fastapi import APIRouter, Body, Depends, Request, status +from fastapi_injector import Injected + +from app.auth.dependencies import auth, permissions +from app.core.exceptions.error_messages import ErrorKey +from app.core.exceptions.exception_classes import AppException +from app.core.permissions.constants import Permissions as P +from app.core.tenant_scope import get_tenant_context +from app.schemas.template import ( + TemplateCreateFromAgent, + TemplateInstallRequest, + TemplateInstallResponse, + TemplateListItem, + TemplateRead, + TemplateRejectRequest, +) +from app.services.template import TemplateService + +router = APIRouter() + + +async def require_master_scope() -> None: + """Approval actions run only in the master (control-plane) scope.""" + if get_tenant_context() != "master": + raise AppException( + status_code=403, error_key=ErrorKey.NOT_AUTHORIZED_ACCESS_RESOURCE + ) + + +@router.get( + "", + response_model=List[TemplateListItem], + dependencies=[Depends(auth), Depends(permissions(P.Template.READ))], +) +async def list_templates( + request: Request, + service: TemplateService = Injected(TemplateService), +): + """Gallery listing: official + approved community + the user's own templates.""" + return await service.list_templates(request.state.user.id) + + +# NOTE: declared before "/{template_id}" so the literal path wins over the UUID param. +@router.get( + "/review", + response_model=List[TemplateListItem], + dependencies=[ + Depends(auth), + Depends(permissions(P.Template.APPROVE)), + Depends(require_master_scope), + ], +) +async def list_review_queue( + service: TemplateService = Injected(TemplateService), +): + """Pending publish submissions awaiting master-admin review.""" + return await service.list_pending() + + +@router.get( + "/{template_id}", + response_model=TemplateRead, + dependencies=[Depends(auth), Depends(permissions(P.Template.READ))], +) +async def get_template( + template_id: UUID, + request: Request, + service: TemplateService = Injected(TemplateService), +): + return await service.get_template(template_id, request.state.user.id) + + +@router.post( + "/{template_id}/install", + response_model=TemplateInstallResponse, + dependencies=[Depends(auth), Depends(permissions(P.Template.INSTALL))], +) +async def install_template( + template_id: UUID, + request: Request, + body: TemplateInstallRequest = Body(default=TemplateInstallRequest()), + service: TemplateService = Injected(TemplateService), +): + """Instantiate a template into the current tenant as a new agent + workflow.""" + return await service.install(template_id, body.name, request.state.user.id) + + +@router.post( + "/{template_id}/publish", + response_model=TemplateRead, + dependencies=[Depends(auth), Depends(permissions(P.Template.PUBLISH))], +) +async def publish_template( + template_id: UUID, + request: Request, + service: TemplateService = Injected(TemplateService), +): + """Submit one of the user's own templates to the global review queue.""" + return await service.publish(template_id, request.state.user.id) + + +@router.post( + "/review/{template_id}/approve", + status_code=status.HTTP_204_NO_CONTENT, + dependencies=[ + Depends(auth), + Depends(permissions(P.Template.APPROVE)), + Depends(require_master_scope), + ], +) +async def approve_template( + template_id: UUID, + request: Request, + service: TemplateService = Injected(TemplateService), +): + await service.approve(template_id, request.state.user.id) + + +@router.post( + "/review/{template_id}/reject", + status_code=status.HTTP_204_NO_CONTENT, + dependencies=[ + Depends(auth), + Depends(permissions(P.Template.APPROVE)), + Depends(require_master_scope), + ], +) +async def reject_template( + template_id: UUID, + request: Request, + body: TemplateRejectRequest = Body(default=TemplateRejectRequest()), + service: TemplateService = Injected(TemplateService), +): + await service.reject(template_id, request.state.user.id, body.reason) + + +@router.delete( + "/review/{template_id}", + status_code=status.HTTP_204_NO_CONTENT, + dependencies=[ + Depends(auth), + Depends(permissions(P.Template.APPROVE)), + Depends(require_master_scope), + ], +) +async def remove_global_template( + template_id: UUID, + request: Request, + service: TemplateService = Injected(TemplateService), +): + """Master-admin removal of a published/community template from the library.""" + await service.remove_global(template_id, request.state.user.id) + + +@router.post( + "/{template_id}/unpublish", + status_code=status.HTTP_204_NO_CONTENT, + dependencies=[Depends(auth), Depends(permissions(P.Template.PUBLISH))], +) +async def unpublish_template( + template_id: UUID, + request: Request, + service: TemplateService = Injected(TemplateService), +): + """Withdraw the caller's published copy of their own template.""" + await service.unpublish(template_id, request.state.user.id) + + +@router.post( + "/from-agent", + response_model=TemplateRead, + status_code=status.HTTP_201_CREATED, + dependencies=[Depends(auth), Depends(permissions(P.Template.CREATE))], +) +async def create_template_from_agent( + request: Request, + data: TemplateCreateFromAgent = Body(...), + service: TemplateService = Injected(TemplateService), +): + """Save one of the user's own agents as a private, tenant-local template.""" + return await service.create_from_agent(data, request.state.user.id) + + +@router.delete( + "/{template_id}", + status_code=status.HTTP_204_NO_CONTENT, + dependencies=[Depends(auth), Depends(permissions(P.Template.DELETE))], +) +async def delete_template( + template_id: UUID, + request: Request, + service: TemplateService = Injected(TemplateService), +): + await service.delete_template(template_id, request.state.user.id) diff --git a/backend/app/api/v1/routes/test_cases.py b/backend/app/api/v1/routes/test_cases.py index 66b41500b..8dfd64173 100644 --- a/backend/app/api/v1/routes/test_cases.py +++ b/backend/app/api/v1/routes/test_cases.py @@ -64,6 +64,20 @@ async def import_cases_from_conversation( ) +@router.delete( + "/suites/{suite_id}/conversations/{conversation_id}", + status_code=status.HTTP_204_NO_CONTENT, + dependencies=[Depends(auth), Depends(permissions(P.Workflow.UPDATE))], +) +async def remove_conversation_from_suite( + suite_id: UUID, + conversation_id: UUID, + service: TestSuiteService = Injected(TestSuiteService), +): + """Remove every case imported from one conversation, leaving the rest intact.""" + await service.remove_conversation_from_suite(suite_id, conversation_id) + + @router.patch( "/cases/{case_id}", response_model=TestCase, diff --git a/backend/app/api/v1/routes/workflows.py b/backend/app/api/v1/routes/workflows.py index 24a86049b..c0837f664 100644 --- a/backend/app/api/v1/routes/workflows.py +++ b/backend/app/api/v1/routes/workflows.py @@ -8,16 +8,18 @@ from fastapi_injector import Injected from app.auth.dependencies import auth, permissions +from app.core.exceptions.exception_classes import AppException from app.core.permissions.constants import Permissions as P +from app.core.utils.string_utils import truncate_for_log from app.dependencies.injector import injector +from app.modules.workflow.agents.sub_agents.turn_router import SubAgentTurnRouter +from app.modules.workflow.engine.pii_anonymizer import PIIAnonymizer from app.modules.workflow.engine.workflow_engine import WorkflowEngine from app.modules.workflow.llm.provider import LLMProvider from app.modules.workflow.utils import generate_python_function_template from app.schemas.dynamic_form_schemas.nodes import NODE_DIALOG_SCHEMAS from app.schemas.workflow import Workflow, WorkflowCreate, WorkflowMinimal, WorkflowUpdate from app.services.llm_providers import LlmProviderService -from app.modules.workflow.engine.pii_anonymizer import PIIAnonymizer -from app.core.utils.string_utils import truncate_for_log from app.services.workflow import WorkflowService router = APIRouter() @@ -31,6 +33,7 @@ "chatOutputNode", "routerNode", "agentNode", + "subAgentNode", "apiToolNode", "openApiNode", "templateNode", @@ -329,17 +332,32 @@ async def test_workflow( thread_id = input_data.get("thread_id", str(uuid.uuid4())) start_node_id = input_data.get("human_in_the_loop_node_id") + turn_router = SubAgentTurnRouter(workflow_engine, owner_id=workflow_config["id"]) + supplied_thread = bool(input_data.get("thread_id")) + if turn_router.has_sub_agents(): + input_data.setdefault("agent_id", workflow_config["id"]) + if supplied_thread and not start_node_id: + routed = await turn_router.route_turn( + input_data.get("message", ""), thread_id, input_data, persist=True + ) + if routed is not None: + return routed + state = await workflow_engine.execute_from_node( start_node_id=start_node_id, input_data=input_data, thread_id=thread_id, + registry_managed=supplied_thread, ) - return state.format_state_as_response() + return turn_router.finalize(state.format_state_as_response()) except HTTPException: # Re-raise HTTP exceptions as-is raise + except AppException: + # Sub-agent session errors carry their own status/key + raise except Exception as e: logger.error(f"Error testing workflow with new engine: {e}") raise HTTPException(status_code=500, detail=str(e)) @@ -405,8 +423,7 @@ async def test_individual_node(test_data: Dict[str, Any]): } workflow_engine = WorkflowEngine(workflow_config) - # Execute the workflow - state = await workflow_engine.execute_from_node(input_data=input_data) + state = await workflow_engine.execute_from_node(start_node_id=test_id, input_data=input_data) return state.format_state_as_response() diff --git a/backend/app/core/exceptions/error_messages.py b/backend/app/core/exceptions/error_messages.py index 65b92f50f..39f782ef9 100644 --- a/backend/app/core/exceptions/error_messages.py +++ b/backend/app/core/exceptions/error_messages.py @@ -14,6 +14,8 @@ class ErrorKey(Enum): AUDIT_LOG_NOT_FOUND = "audit_log_not_found" SENTIMENT_OBJECT_STRUCTURE = "sentiment_object_structure" AGENT_NOT_FOUND = "AGENT_NOT_FOUND" + TEMPLATE_NOT_FOUND = "TEMPLATE_NOT_FOUND" + TEMPLATE_INVALID = "TEMPLATE_INVALID" OPERATOR_NOT_FOUND = "OPERATOR_NOT_FOUND" INVALID_FILE_FORMAT = "invalid_file_format" NO_SELECTED_FILE = "no_selected_file" @@ -85,6 +87,9 @@ class ErrorKey(Enum): APP_SETTINGS_NOT_FOUND = "APP_SETTINGS_NOT_FOUND" FEATURE_FLAG_NOT_FOUND = "FEATURE_FLAG_NOT_FOUND" WORKFLOW_NOT_FOUND = "WORKFLOW_NOT_FOUND" + SUB_AGENT_SESSION_STALE = "SUB_AGENT_SESSION_STALE" + SUB_AGENT_INVALID_TOPOLOGY = "SUB_AGENT_INVALID_TOPOLOGY" + SUB_AGENT_INVALID_CONFIG = "SUB_AGENT_INVALID_CONFIG" OPERATOR_ROLE_MISSING = "OPERATOR_ROLE_MISSING" CREATE_USER_TYPE_IN_MENU = "CREATE_USER_TYPE_IN_MENU" LOGIN_ERROR_CONSOLE_USER = "LOGIN_ERROR_CONSOLE_USER" @@ -174,6 +179,8 @@ class ErrorKey(Enum): ErrorKey.AUDIT_LOG_NOT_FOUND: "The requested log was not found.", ErrorKey.SENTIMENT_OBJECT_STRUCTURE: "Sentiment object must have 'positive', 'neutral', and 'negative' fields.", ErrorKey.AGENT_NOT_FOUND: "Agent not found.", + ErrorKey.TEMPLATE_NOT_FOUND: "Template not found.", + ErrorKey.TEMPLATE_INVALID: "The template is invalid or cannot be used.", ErrorKey.INVALID_FILE_FORMAT: "Invalid file format.", ErrorKey.NO_SELECTED_FILE: "No selected file", ErrorKey.INVALID_RECORDED_AT: "Invalid recorded_at format. Use ISO 8601: YYYY-MM-DDTHH:MM:SSZ", @@ -330,11 +337,17 @@ class ErrorKey(Enum): ErrorKey.SSO_MICROSOFT_USER_DENIED: "Your account is not provisioned for GenAssist. Contact an administrator.", ErrorKey.SSO_MICROSOFT_REDIRECT_NOT_ALLOWED: "SSO redirect target is not allowed by server configuration.", ErrorKey.SSO_MICROSOFT_NOT_CONFIGURED: "Microsoft SSO is enabled but required settings are missing.", + ErrorKey.SUB_AGENT_SESSION_STALE: "The workflow changed while a sub-agent conversation was in progress. Please start a new message.", + ErrorKey.SUB_AGENT_INVALID_TOPOLOGY: "The sub-agent connections in this workflow are invalid: {0}", + ErrorKey.SUB_AGENT_INVALID_CONFIG: "A sub-agent in this workflow is misconfigured: {0}", }, "fr": { ErrorKey.INTERNAL_ERROR: "Une erreur interne du serveur est survenue. Veuillez réessayer plus tard.", ErrorKey.FILE_MANAGER_INITIALIZATION_FAILED: "Échec de l'initialisation du service de gestion des fichiers.", ErrorKey.INTERNAL_SERVER_ERROR: "Une erreur interne du serveur est survenue. Veuillez réessayer plus tard.", + ErrorKey.SUB_AGENT_SESSION_STALE: "Le workflow a changé pendant une conversation avec un sous-agent. Veuillez démarrer un nouveau message.", + ErrorKey.SUB_AGENT_INVALID_TOPOLOGY: "Les connexions de sous-agents de ce workflow sont invalides : {0}", + ErrorKey.SUB_AGENT_INVALID_CONFIG: "Un sous-agent de ce workflow est mal configuré : {0}", }, } diff --git a/backend/app/core/permissions/constants.py b/backend/app/core/permissions/constants.py index 3ad0dc329..e68c42685 100644 --- a/backend/app/core/permissions/constants.py +++ b/backend/app/core/permissions/constants.py @@ -18,6 +18,16 @@ class AgentPermissions: SWITCH = "switch:agent" +class TemplatePermissions: + """Template Marketplace permissions""" + READ = "read:template" + CREATE = "create:template" + INSTALL = "install:template" + DELETE = "delete:template" + PUBLISH = "publish:template" + APPROVE = "approve:template" + + class ApiKeyPermissions: """API Key CRUD permissions""" CREATE = "create:api_key" @@ -248,6 +258,7 @@ class Permissions: ... """ Agent = AgentPermissions + Template = TemplatePermissions ApiKey = ApiKeyPermissions AppSettings = AppSettingsPermissions AuditLog = AuditLogPermissions @@ -294,7 +305,7 @@ def get_all_permission_constants() -> set[str]: # Get all permission classes permission_classes = [ - AgentPermissions, ApiKeyPermissions, AppSettingsPermissions, + AgentPermissions, TemplatePermissions, ApiKeyPermissions, AppSettingsPermissions, AuditLogPermissions, ConversationPermissions, DataSourcePermissions, FeatureFlagPermissions, KnowledgeBasePermissions, LlmAnalystPermissions, LlmProviderPermissions, MlModelPermissions, OperatorPermissions, diff --git a/backend/app/core/utils/enums/template_status_enum.py b/backend/app/core/utils/enums/template_status_enum.py new file mode 100644 index 000000000..7297ecd56 --- /dev/null +++ b/backend/app/core/utils/enums/template_status_enum.py @@ -0,0 +1,15 @@ +import enum + + +class TemplateStatus(str, enum.Enum): + """Lifecycle status of a marketplace template. + + ``PRIVATE`` rows live in a tenant DB (the owner's own template). The + ``PENDING``/``APPROVED``/``REJECTED`` states apply to published copies in the + master (control-plane) DB; only ``APPROVED`` ones are installable cross-tenant. + """ + + PRIVATE = "private" + PENDING = "pending" + APPROVED = "approved" + REJECTED = "rejected" diff --git a/backend/app/core/utils/llm_usage_utils.py b/backend/app/core/utils/llm_usage_utils.py index 0c875bb47..f79669701 100644 --- a/backend/app/core/utils/llm_usage_utils.py +++ b/backend/app/core/utils/llm_usage_utils.py @@ -1,8 +1,9 @@ """ Token usage extraction utilities for LLM responses. -Extracts input_tokens, output_tokens, total_tokens from LangChain AIMessage -response_metadata, handling provider-specific structures (OpenAI, Anthropic, etc.). +Extracts input_tokens, output_tokens, total_tokens from a LangChain AIMessage, +preferring the standardized usage_metadata attribute with a response_metadata +fallback for provider-specific structures (OpenAI, Anthropic, etc.). """ from typing import Any, Dict, Optional @@ -73,10 +74,7 @@ def extract_usage_from_response_metadata(metadata: Dict[str, Any]) -> Optional[D def extract_usage_from_aimessage(message: Any) -> Optional[Dict[str, int]]: """ - Extract token usage from a LangChain AIMessage. - - Args: - message: AIMessage instance (from llm.ainvoke) with response_metadata + Extract token usage from a LangChain AIMessage. Prefer ``usage_metadata`` first. Returns: Dict with input_tokens, output_tokens, total_tokens, or None if not found. @@ -84,25 +82,37 @@ def extract_usage_from_aimessage(message: Any) -> Optional[Dict[str, int]]: if message is None: return None - metadata = None - if hasattr(message, "response_metadata"): - metadata = getattr(message, "response_metadata", None) - elif hasattr(message, "usage_metadata"): - metadata = getattr(message, "usage_metadata", None) - if metadata and not isinstance(metadata, dict): - metadata = {"usage_metadata": metadata} if metadata else None - - if not metadata: + response_metadata = getattr(message, "response_metadata", None) + + usage = None + usage_metadata = getattr(message, "usage_metadata", None) + if isinstance(usage_metadata, dict): + input_tokens = usage_metadata.get("input_tokens") + output_tokens = usage_metadata.get("output_tokens") + if input_tokens is not None or output_tokens is not None: + input_tokens = input_tokens if input_tokens is not None else 0 + output_tokens = output_tokens if output_tokens is not None else 0 + total_tokens = usage_metadata.get("total_tokens") + if total_tokens is None: + total_tokens = input_tokens + output_tokens + usage = { + "input_tokens": input_tokens, + "output_tokens": output_tokens, + "total_tokens": total_tokens, + } + + if usage is None and response_metadata: + usage = extract_usage_from_response_metadata(response_metadata) + + if usage is None: return None - usage = extract_usage_from_response_metadata(metadata) - # If this response came from a FallbackChatModel, record which provider actually # answered so usage can be attributed correctly (the primary may have failed over). - if usage is not None and isinstance(metadata, dict): + if isinstance(response_metadata, dict): from app.modules.workflow.llm.fallback_exceptions import FALLBACK_PROVIDER_ID_KEY - responding_provider_id = metadata.get(FALLBACK_PROVIDER_ID_KEY) + responding_provider_id = response_metadata.get(FALLBACK_PROVIDER_ID_KEY) if responding_provider_id: usage["provider_id"] = responding_provider_id diff --git a/backend/app/db/models/__init__.py b/backend/app/db/models/__init__.py index bff100ed7..5c4413ecb 100644 --- a/backend/app/db/models/__init__.py +++ b/backend/app/db/models/__init__.py @@ -53,6 +53,7 @@ PipelineRunStatus, ) from .prompt_editor import PromptConfigModel, PromptVersionModel +from .template import TemplateModel from .tenant import TenantModel from .tool import ToolModel from .user_group import UserGroupModel @@ -108,6 +109,7 @@ "AgentModel", "AgentSecuritySettingsModel", "WorkflowModel", + "TemplateModel", "MLModel", "MLModelPipelineConfig", "MLModelPipelineRun", @@ -175,6 +177,7 @@ AgentModel, AgentSecuritySettingsModel, WorkflowModel, + TemplateModel, MLModel, MLModelPipelineConfig, MLModelPipelineRun, diff --git a/backend/app/db/models/template.py b/backend/app/db/models/template.py new file mode 100644 index 000000000..f26674e95 --- /dev/null +++ b/backend/app/db/models/template.py @@ -0,0 +1,60 @@ +from typing import List, Optional + +from sqlalchemy import Boolean, Column, DateTime, Integer, String +from sqlalchemy.dialects.postgresql import JSONB, UUID as PGUUID +from sqlalchemy.orm import Mapped, mapped_column + +from app.core.utils.enums.template_status_enum import TemplateStatus +from app.db.base import Base + + +class TemplateModel(Base): + """A reusable agent/workflow template for the Template Marketplace. + + The same model backs two physically separate stores (schema is shared across + the master and every tenant DB): + + * **Tenant DB** — a user's private template (``status="private"``, + ``created_by`` = owner). Never leaves that tenant. + * **Master (control-plane) DB** — a *published* copy submitted for global + sharing (``status`` in ``pending``/``approved``/``rejected``). Only rows + here are visible cross-tenant, and only ``approved`` ones are installable. + Global rows are always accessed via the ``"master"`` session factory. + + ``graph`` is an already-sanitized workflow graph (secrets stripped, per-tenant + references blanked). See ``app.services.template_sanitizer``. + """ + + __tablename__ = "templates" + + title = Column(String(120), nullable=False) + description = Column(String(500), nullable=True) + category = Column(String(60), nullable=True) + # Lucide icon name (e.g. "Headphones") or emoji, used by the gallery card. + icon = Column(String(60), nullable=True) + tags: Mapped[Optional[List[str]]] = mapped_column(JSONB, nullable=True) + # Distinct node types present in the graph, for preview chips + validation. + node_types: Mapped[Optional[List[str]]] = mapped_column(JSONB, nullable=True) + # Sanitized workflow graph: {"nodes": [...], "edges": [...], "testInput": {...}?} + graph: Mapped[dict] = mapped_column(JSONB, nullable=False) + # Default agent config applied on install (name/description/welcome_message/...). + agent_config: Mapped[Optional[dict]] = mapped_column(JSONB, nullable=True) + is_official = Column(Boolean, nullable=False, server_default="false") + # Number of times this template has been installed (drives "most used"). + install_count = Column(Integer, nullable=False, server_default="0") + + # ---- publishing / approval lifecycle ---- + # private (tenant-local) | pending | approved | rejected — see TemplateStatus. + # Kept as a plain String column (not sa.Enum) so no DB-level enum type is + # created; values are constrained in the app layer via TemplateStatus. + status = Column(String(20), nullable=False, server_default=TemplateStatus.PRIVATE.value) + # Slug of the tenant a published copy originated from. + source_tenant = Column(String(120), nullable=True) + # User who submitted the publish request. + published_by = Column(PGUUID(as_uuid=True), nullable=True) + # The originating tenant-local template id (for dedup / traceability). + source_template_id = Column(PGUUID(as_uuid=True), nullable=True) + # Master-admin who approved/rejected, and when. + approved_by = Column(PGUUID(as_uuid=True), nullable=True) + approved_at = Column(DateTime(timezone=True), nullable=True) + rejection_reason = Column(String(500), nullable=True) diff --git a/backend/app/db/models/test_suite.py b/backend/app/db/models/test_suite.py index 94bb7c49b..17af27878 100644 --- a/backend/app/db/models/test_suite.py +++ b/backend/app/db/models/test_suite.py @@ -1,8 +1,9 @@ from typing import List, Optional from uuid import UUID -from sqlalchemy import ForeignKey, String, Text +from sqlalchemy import ForeignKey, Index, String, Text from sqlalchemy.dialects.postgresql import JSONB +from sqlalchemy.dialects.postgresql import UUID as PGUUID from sqlalchemy.orm import Mapped, mapped_column, relationship from app.db.base import Base @@ -65,11 +66,27 @@ class TestSuiteModel(Base): class TestCaseModel(Base): __tablename__ = "test_cases" + __table_args__ = ( + Index( + "ix_test_cases_conversation_turn", + "suite_id", + "source_conversation_id", + "turn_index", + ), + ) suite_id: Mapped[UUID] = mapped_column( ForeignKey("test_suites.id"), nullable=False ) + # Source conversation for imported cases; null for manual and legacy cases + source_conversation_id: Mapped[Optional[UUID]] = mapped_column( + PGUUID(as_uuid=True), nullable=True, index=True + ) + + # Position of this turn within its source conversation + turn_index: Mapped[Optional[int]] = mapped_column(nullable=True) + # Free-form scenario tags, e.g. ["refund", "es-ES"] tags: Mapped[Optional[List[str]]] = mapped_column(JSONB, nullable=True) @@ -153,6 +170,9 @@ class TestResultModel(Base): # { technique_key: { "score": float|bool, "passed": bool, "comment": str|null } } metrics: Mapped[Optional[dict]] = mapped_column(JSONB, nullable=True) + # scored | execution_failed | scoring_failed | skipped; null for legacy rows + status: Mapped[Optional[str]] = mapped_column(String(32), nullable=True) + error: Mapped[Optional[str]] = mapped_column(Text, nullable=True) run = relationship("TestRunModel", back_populates="results") diff --git a/backend/app/db/seed/knowledge/generate_node_specs.py b/backend/app/db/seed/knowledge/generate_node_specs.py index 76ba29b39..faa1cbcc5 100644 --- a/backend/app/db/seed/knowledge/generate_node_specs.py +++ b/backend/app/db/seed/knowledge/generate_node_specs.py @@ -54,6 +54,28 @@ "Email classifier that routes to different actions", ], }, + "subAgentNode": { + "category": "AI", + "description": ( + "A specialist child agent a parent agentNode (or another subAgentNode) delegates to. " + "It never sits in the main flow: attach it to a parent by wiring its output_sub_agent " + "port to the parent's input_sub_agents port, and the parent gains a delegation tool for it. " + "Three collaboration modes: single_turn (answers once and returns to the parent, like an " + "agent-as-tool), task (does a bounded job, may ask the user one clarifying question, then " + "calls finish_task), and chat (owns the conversation until it calls return_to_parent)." + ), + "when_to_use": ( + "Use a sub-agent for stateful, context-dependent, multi-step work that may need to clarify " + "with the user (task/chat modes) or for isolating a reusable specialist (single_turn). Prefer " + "a plain toolBuilderNode tool for stateless, atomic operations instead. The canvas assistant " + "attaches one via the as_sub_agent_for action (target = the parent agent's id)." + ), + "example_use_cases": [ + "A flight-search specialist that asks 'Is a layover okay?' before booking (task mode)", + "A billing-support agent that takes over the chat until the issue is resolved (chat mode)", + "An isolated summarizer the parent calls and gets one answer back from (single_turn mode)", + ], + }, "llmModelNode": { "category": "AI", "description": "A standalone LLM call node. Sends a system prompt + user prompt to a language model and returns the response. Similar to agentNode but without tool calling or ReAct loops.", @@ -364,6 +386,8 @@ def generate_node_specs() -> str: lines.append("- `text`: can connect to `any` or `text` ports") lines.append("- `tools`: can ONLY connect to other `tools` ports") lines.append(" - Used for: toolBuilderNode.output_tool -> agentNode.input_tools") + lines.append("- `sub_agents`: can ONLY connect to other `sub_agents` ports") + lines.append(" - Used for: subAgentNode.output_sub_agent -> agentNode.input_sub_agents (delegation)") lines.append("") # ── Common Patterns ─────────────────────────────────────────────────── @@ -383,6 +407,11 @@ def generate_node_specs() -> str: lines.append("toolBuilderNode 'API' -> apiToolNode (tool)") lines.append("Both toolBuilderNodes connect output_tool -> agentNode.input_tools") lines.append("") + lines.append("### Chatbot with a Sub-Agent") + lines.append("chatInputNode -> agentNode -> chatOutputNode") + lines.append("subAgentNode 'flight_search' (task mode) attached to the agent") + lines.append("subAgentNode.output_sub_agent -> agentNode.input_sub_agents (delegation)") + lines.append("") lines.append("### Branching Workflow") lines.append("chatInputNode -> agentNode -> routerNode") lines.append("routerNode.output_true -> [urgent path] -> chatOutputNode") @@ -543,6 +572,7 @@ def generate_node_specs() -> str: lines.append("- `edges.from`/`edges.to`: reference uniqueId values") lines.append("- `edges.sourceHandle` defaults to \"output\", `edges.targetHandle` defaults to \"input\"") lines.append("- For tool connections: sourceHandle=\"output_tool\", targetHandle=\"input_tools\"") + lines.append("- For sub-agent delegation: sourceHandle=\"output_sub_agent\", targetHandle=\"input_sub_agents\" (from the subAgentNode to its parent)") lines.append("- For router branches: sourceHandle=\"output_true\" or \"output_false\"") lines.append("") diff --git a/backend/app/db/seed/knowledge/node_specs.md b/backend/app/db/seed/knowledge/node_specs.md index 1222e437c..6a0755742 100644 --- a/backend/app/db/seed/knowledge/node_specs.md +++ b/backend/app/db/seed/knowledge/node_specs.md @@ -9,14 +9,15 @@ This document is the authoritative reference for every node type available in Ge Nodes connect via **handlers** (ports). Each handler has: - **type**: `source` (output) or `target` (input) - **position**: left, right, top, bottom -- **compatibility**: `any`, `text`, `tools` +- **compatibility**: `any`, `text`, `tools`, `sub_agents` **Compatibility rules:** - `any` ↔ `any`: allowed - `any` ↔ `text`: allowed - `text` ↔ `text`: allowed - `tools` ↔ `tools`: allowed (ONLY tools-to-tools) -- `any` / `text` → `tools`: NOT allowed +- `sub_agents` ↔ `sub_agents`: allowed (ONLY sub_agents-to-sub_agents — used for delegation) +- `any` / `text` → `tools` / `sub_agents`: NOT allowed **CRITICAL — Single Input Rule:** Most nodes accept only ONE incoming edge to their `input` handler. You CANNOT connect two nodes to the same node's input (except `aggregatorNode`, which is specifically designed to receive multiple inputs). If a workflow has branches (e.g., after a routerNode), each branch MUST have its own separate `chatOutputNode`. Do NOT merge branches into a single chatOutputNode. @@ -28,6 +29,7 @@ Most nodes accept only ONE incoming edge to their `input` handler. You CANNOT co Defaults: `sourceHandle = "output"`, `targetHandle = "input"`. Override for special connections: - Tool: `sourceHandle: "output_tool"`, `targetHandle: "input_tools"` - Tool sub-flow: `sourceHandle: "starter_processor"`, `targetHandle: "input"` +- Sub-agent delegation: `sourceHandle: "output_sub_agent"`, `targetHandle: "input_sub_agents"` - Router true: `sourceHandle: "output_true"` - Router false: `sourceHandle: "output_false"` @@ -65,6 +67,16 @@ Edges: 6→7 (starter_processor → input) ``` +### Chatbot with a Sub-Agent +A `subAgentNode` is a specialist the agent delegates to; it hangs off the parent and is NOT in the main chain. +``` +chatInputNode(1) → agentNode(2) → chatOutputNode(3) +subAgentNode(4) "flight_search" (task mode) +Edges: + 1→2, 2→3 + 4→2 (output_sub_agent → input_sub_agents) +``` + ### Branching Workflow Each branch MUST have its own chatOutputNode — nodes only accept one input edge (except aggregatorNode). ``` @@ -148,6 +160,13 @@ These rules are **non-negotiable**. Violating any of them produces a broken work - `{"from": "", "to": "", "sourceHandle": "output_tool", "targetHandle": "input_tools"}` — registers the tool with the agent - `{"from": "", "to": "", "sourceHandle": "starter_processor", "targetHandle": "input"}` — connects the tool's processor +### Sub-Agent Delegation Rules +- A `subAgentNode` is a specialist child that a parent `agentNode` (or another `subAgentNode`) delegates to. It NEVER goes in the main chain and has no `input`/`output` handle. +- Attach it with a SINGLE edge from the child to the parent: `{"from": "", "to": "", "sourceHandle": "output_sub_agent", "targetHandle": "input_sub_agents"}`. The parent then gains a delegation tool named after the child. +- Each child has exactly ONE parent. Depth is limited (parent → child → grandchild, up to 3). No cycles or self-links. +- Choose a sub-agent over a tool for stateful, multi-step work that may need to clarify with the user; choose a plain `toolBuilderNode` tool for stateless, atomic operations. +- The `mode` decides how control returns: `single_turn` answers once, `task` may ask one clarifying question then finishes, `chat` owns the conversation until it hands back. A `task` child must be a leaf (no further children). + ### Integration Rules - Use dedicated nodes where available: `zendeskTicketNode` for Zendesk, `slackMessageNode` for Slack, `jiraNode` for Jira, `gmailNode` for Gmail, `knowledgeBaseNode` for document search, `calendarEventNode` for calendar, `readMailsNode` for reading emails, `whatsappToolNode` for WhatsApp. Only use `apiToolNode` for APIs without a dedicated node. - **ALL** integration nodes MUST be connected as tools of an `agentNode`. NEVER place them inline in the main flow. @@ -176,6 +195,11 @@ Tool connections — both edges required: {"from": "", "to": "", "sourceHandle": "starter_processor", "targetHandle": "input"} ``` +Sub-agent delegation — single edge from child to parent: +```json +{"from": "", "to": "", "sourceHandle": "output_sub_agent", "targetHandle": "input_sub_agents"} +``` + --- ## Node Reference @@ -254,6 +278,49 @@ Tool connections — both edges required: --- +### subAgentNode — Sub-Agent +**Category:** AI +**Purpose:** A specialist child agent that a parent `agentNode` (or another `subAgentNode`) delegates to. It never sits in the main flow — attach it by wiring its `output_sub_agent` port to the parent's `input_sub_agents` port, and the parent gains a delegation tool named after it. The `mode` controls how control returns: `single_turn` answers once and returns (like an agent-as-tool), `task` does a bounded job and may ask ONE clarifying question before calling `finish_task`, `chat` owns the conversation until it calls `return_to_parent`. +**Use cases:** Flight-search specialist that asks "Is a layover okay?" before booking (task), billing-support agent that takes over the chat until resolved (chat), isolated summarizer the parent calls for one answer (single_turn). Prefer a plain `toolBuilderNode` tool for stateless, atomic work. + +**Handlers:** +| ID | Type | Position | Compatibility | +|---|---|---|---| +| output_sub_agent | source | top | sub_agents | +| input_tools | target | bottom | tools | +| input_sub_agents | target | bottom | sub_agents | + +**Required config:** +| Field | Type | Default | Description | +|---|---|---|---| +| providerId | select | — | LLM provider to use | +| description | text | — | What this sub-agent handles — the parent reads this to decide when to delegate | +| mode | select | "single_turn" | Collaboration mode: single_turn, task, chat | + +**Optional config:** +| Field | Type | Default | Condition | +|---|---|---|---| +| name | text | — | Names the delegation tool the parent sees (e.g. "flight_search") | +| systemPrompt | text | "You are a helpful specialist sub-agent." | Always | +| type | select | "ToolSelector" | Agent pattern: ToolSelector, ReActAgent, ReActAgentLC | +| maxIterations | number | 7 | Max reasoning cycles | +| timeoutSeconds | number | 120 | Max seconds the parent waits for one delegated turn (5–300) | +| memory | boolean | true | Enable the child's own conversation memory | +| piiMasking | boolean | false | Mask PII before sending text to the LLM | +| memoryTrimmingMode | select | "message_count" | When memory=true. Options: message_count, token_budget, message_compacting, rag_retrieval | +| maxMessages | number | 20 | When memoryTrimmingMode=message_count | +| tokenBudget | number | 10000 | When memoryTrimmingMode=token_budget | +| conversationHistoryTokens | number | 5000 | When memoryTrimmingMode=token_budget | +| compactingThreshold | number | 20 | When memoryTrimmingMode=message_compacting | +| compactingKeepRecent | number | 10 | When memoryTrimmingMode=message_compacting | +| compactingModel | select | — | When memoryTrimmingMode=message_compacting | +| compactingImportantEntities | tags | — | When memoryTrimmingMode=message_compacting | +| ragTopK / ragRecentMessages / ragMaxHistoryHours / ragQueryContextMessages / ragGroupSize / ragGroupOverlap / ragPassthroughThreshold | number | — | When memoryTrimmingMode=rag_retrieval | + +Memory sub-settings match agentNode. **Attachment:** single edge `subAgentNode.output_sub_agent` → `agentNode.input_sub_agents` (or another subAgentNode). The canvas assistant attaches one via the `as_sub_agent_for` action (target = the parent agent's id). A sub-agent is NOT wired into the main `input`/`output` flow. + +--- + ### llmModelNode — LLM Model **Category:** AI **Purpose:** Simple LLM call — sends system prompt + user prompt to a model. No tool calling or reasoning loops. Use for text generation, summarization, classification. diff --git a/backend/app/db/seed/workflow_builder_wf_data.json b/backend/app/db/seed/workflow_builder_wf_data.json index c3892c782..5c8fc57ca 100644 --- a/backend/app/db/seed/workflow_builder_wf_data.json +++ b/backend/app/db/seed/workflow_builder_wf_data.json @@ -110,7 +110,7 @@ "providerId": "00000196-19d2-9c28-a2dd-561fff608fa0", "userPrompt": "{{session.message}}", "maxMessages": 10, - "systemPrompt": "You are a workflow editing assistant for GenAssist \u2014 a visual workflow builder platform. You help users modify their existing workflows on the canvas through natural language.\n\nYou operate in a fast, action-oriented style. Users describe what they want changed, and you make it happen.\n\n## CRITICAL: How You Work (Two Phases)\n\nYou work in two phases. Both are mandatory.\n\n**Phase 1 \u2014 Research (use your tools):**\nBefore making ANY change, call the appropriate knowledge base tools to look up node specs and example workflows. Use the tools provided to you via function calling. Do NOT skip this step.\n\n**Phase 2 \u2014 Act (output action tags in your final text response):**\nAfter researching, write your final response as plain text that MUST include literal XML action tags. These tags are parsed by the frontend to modify the canvas. If you describe an action without including the tag, NOTHING happens on the canvas.\n\nIMPORTANT RULES:\n- ALWAYS respond to the user's CURRENT message. Ignore previous conversation topics if the user has moved on.\n- Your final text response MUST contain the literal XML action tags shown below. They are NOT decorative \u2014 they are machine-parsed instructions.\n- NEVER describe actions without including the corresponding XML tags. A response like \"I'll add a Slack node\" without an tag does NOTHING.\n- Do NOT wrap your response in JSON. Output plain text with embedded XML action tags.\n- NEVER wrap action tags in markdown code blocks (no ```xml```, no ```). The tags must appear as raw text.\n- Keep responses SHORT \u2014 one brief confirmation sentence + action tags. No technical explanations.\n- ONLY use these exact action tag names: , , , . Do NOT invent other tags.\n\n## Your Knowledge Base Tools\n\nYou have access to TWO knowledge base tools (called via function calling). You MUST use them \u2014 do NOT rely on your own knowledge for node types, configs, or workflow patterns.\n\n### 1. Node Specs Tool\nYour source of truth for all available node types, their configurations, connection handlers, and usage guidelines.\n- ALWAYS call this tool before adding or updating ANY node \u2014 regardless of how common the node type seems.\n- Never guess or assume node configurations from memory.\n\n### 2. Workflow Examples Tool\nYour source of truth for proven workflow architectures and patterns.\n- ALWAYS call this tool when adding multiple nodes or restructuring a workflow.\n- When the user describes a use case (e.g., \"add a guardrail\", \"add routing\"), search for an example workflow first.\n\n### Mandatory Tool Usage Rules\n- You MUST call the Node Specs tool before ANY add or update action. No exceptions.\n- You MUST call the Workflow Examples tool when adding patterns that involve multiple nodes.\n- You MUST NOT invent node configurations, handler names, or connection patterns from memory.\n- If a tool returns no results, try rephrasing your query.\n\n## Canvas Context\n\nEvery message includes a [CANVAS_CONTEXT] block showing the current workflow state:\n\n[CANVAS_CONTEXT]\nNodes:\n- chatInputNode(id=\"node-1\", label=\"Chat Input\")\n- agentNode(id=\"node-2\", label=\"Support Agent\")\n- chatOutputNode(id=\"node-3\", label=\"Chat Output\")\nEdges: node-1:output->node-2:input, node-2:output->node-3:input\n[/CANVAS_CONTEXT]\n\nUse these exact IDs when modifying the workflow.\n\n## Action Tags (MUST appear literally in your response \u2014 never in code blocks)\n\nThese XML tags are machine-parsed. You MUST write them as raw text, NEVER inside markdown code blocks. Every action REQUIRES the corresponding tag.\n\n**Add a node (regular chain \u2014 not a tool):**\n\n{\"node_type\": \"slackMessageNode\", \"label\": \"Send Alert\", \"config\": {}, \"connect_to\": \"existing_node_id\"}\n\n\n**Add a node as a tool for an agent (use as_tool_for, NOT connect_to):**\n\n{\"node_type\": \"gmailNode\", \"label\": \"Send Email\", \"config\": {}, \"as_tool_for\": \"agent_node_id\"}\n\n\n**Update a node's config:**\n\n{\"node_id\": \"existing_node_id\", \"updates\": {\"systemPrompt\": \"New prompt.\", \"name\": \"New Name\"}}\n\n\n**Remove a node:**\n\n{\"node_id\": \"existing_node_id\"}\n\n\n**Remove a connection between two nodes:**\n\n{\"from_node_id\": \"source_node_id\", \"to_node_id\": \"target_node_id\"}\n\n\n**Insert a node between two existing nodes (connect_to = before, then_connect_to = after):**\n\n{\"node_type\": \"templateNode\", \"label\": \"Format Response\", \"config\": {}, \"connect_to\": \"agent_node_id\", \"then_connect_to\": \"chat_output_node_id\"}\n\n\n**Add a node connecting it to a node you just created in the same response (use connect_to_label when the target has no canvas ID yet):**\n\n{\"node_type\": \"chatOutputNode\", \"label\": \"Output\", \"config\": {}, \"connect_to_label\": \"Ticket Agent\"}\n\n\nYou can include MULTIPLE action tags in a single response.\n\n\n## Tool Pattern \u2014 Adding Nodes as Agent Tools\n\nWhen a user asks to add any integration node (gmailNode, slackMessageNode, jiraNode, whatsappToolNode, apiToolNode, calendarEventNode, readMailsNode, sqlNode, etc.) **as a tool for an agentNode**, you MUST use the `as_tool_for` field instead of `connect_to`.\n\nUsing `as_tool_for` automatically:\n1. Creates a `toolBuilderNode` that wraps the integration node\n2. Connects `toolBuilderNode.output_tool` \u2192 `agentNode.input_tools`\n3. Connects `toolBuilderNode.starter_processor` \u2192 `integrationNode.input`\n\n**NEVER use `connect_to` when adding an integration node as a tool for an agent. Use `as_tool_for`.**\n\nExample \u2014 user asks: \"Add Gmail as a tool for the support agent\" (agent id = \"node-2\"):\n\nDone! Added Gmail as a tool for the support agent.\n\n{\"node_type\": \"gmailNode\", \"label\": \"Send Email\", \"config\": {}, \"as_tool_for\": \"node-2\"}\n\n\nExample \u2014 user asks: \"Add Slack as a second tool for the agent\" (agent id = \"node-2\"):\n\nDone! Added Slack as a tool for the agent.\n\n{\"node_type\": \"slackMessageNode\", \"label\": \"Send Slack Message\", \"config\": {}, \"as_tool_for\": \"node-2\"}\n\n\nWRONG \u2014 never use connect_to for tool additions:\n\n{\"node_type\": \"gmailNode\", \"label\": \"Send Email\", \"config\": {}, \"connect_to\": \"node-2\"}\n\n(This creates a linear chain edge, NOT a tool connection \u2014 the agent can't call it.)\n\n## Understanding User Intent\n\n- \"Add X after Y\" -> find Y's ID in CANVAS_CONTEXT -> use it as connect_to\n- \"Add X before the output\" -> find the node before chatOutputNode in the edge chain -> use that as connect_to\n- \"Add X at the end\" -> find the last node before chatOutputNode -> use that as connect_to\n- \"Break/remove the connection between X and Y\" -> find both node IDs in CANVAS_CONTEXT -> use REMOVE_EDGE\n- \"Add X as a tool for the agent\" -> find the agent's ID in CANVAS_CONTEXT -> use as_tool_for\n- \"Insert X between A and B\" / \"Add X before output\" -> use connect_to=A, then_connect_to=B (also use REMOVE_EDGE to remove old A\u2192B connection)\n- Building a multi-node chain (e.g. agentNode + outputNode) -> first node: connect_to=; second node: connect_to_label=\"