apunkt/rowboat
1
0
Fork 0
mirror of https://github.com/rowboatlabs/rowboat.git synced 2026-06-06 19:35:44 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions

added conext of triggers to the copilot along with being able to edit and delete triggers

Browse source
This commit is contained in:
tusharmagar 2025-09-29 10:48:07 +05:30
parent 4af86dc7ec
commit da6fa8597f
16 changed files with 571 additions and 86 deletions
Download patch file Download diff file Expand all files Collapse all files

4
apps/rowboat/app/actions/composio.actions.ts
View file

@ -157,6 +157,7 @@ export async function createComposioTriggerDeployment(request: {
export async function listComposioTriggerDeployments(request: {
projectId: string,
cursor?: string,
limit?: number,
}) {
const user = await authCheck();
@ -166,6 +167,7 @@ export async function listComposioTriggerDeployments(request: {
userId: user.id,
projectId: request.projectId,
cursor: request.cursor,
limit: request.limit,
});
}
@ -191,4 +193,4 @@ export async function fetchComposioTriggerDeployment(request: { deploymentId: st
userId: user.id,
deploymentId: request.deploymentId,
});
}
}

5
apps/rowboat/app/actions/copilot.actions.ts
View file

@ -3,6 +3,7 @@ import {
CopilotAPIRequest,
CopilotChatContext, CopilotMessage,
DataSourceSchemaForCopilot,
TriggerSchemaForCopilot,
} from "../../src/entities/models/copilot";
import {
Workflow} from "../lib/types/workflow_types";
@ -26,7 +27,8 @@ export async function getCopilotResponseStream(
messages: z.infer<typeof CopilotMessage>[],
current_workflow_config: z.infer<typeof Workflow>,
context: z.infer<typeof CopilotChatContext> | null,
dataSources?: z.infer<typeof DataSourceSchemaForCopilot>[]
dataSources?: z.infer<typeof DataSourceSchemaForCopilot>[],
triggers?: z.infer<typeof TriggerSchemaForCopilot>[]
): Promise<{
streamId: string;
} | { billingError: string }> {
@ -42,6 +44,7 @@ export async function getCopilotResponseStream(
workflow: current_workflow_config,
context,
dataSources,
triggers,
}
});
return {

17
apps/rowboat/app/projects/[projectId]/copilot/app.tsx
View file

@ -2,7 +2,7 @@
import { Button } from "@/components/ui/button";
import { Dropdown, DropdownItem, DropdownMenu, DropdownSection, DropdownTrigger, Spinner, Tooltip } from "@heroui/react";
import { useRef, useState, createContext, useContext, useCallback, forwardRef, useImperativeHandle, useEffect, Ref } from "react";
import { CopilotChatContext } from "../../../../src/entities/models/copilot";
import { CopilotChatContext, TriggerSchemaForCopilot } from "../../../../src/entities/models/copilot";
import { CopilotMessage } from "../../../../src/entities/models/copilot";
import { Workflow } from "@/app/lib/types/workflow_types";
import { DataSource } from "@/src/entities/models/data-source";
@ -36,6 +36,8 @@ interface AppProps {
onMessagesChange?: (messages: z.infer<typeof CopilotMessage>[]) => void;
isInitialState?: boolean;
dataSources?: z.infer<typeof DataSource>[];
triggers?: z.infer<typeof TriggerSchemaForCopilot>[];
onTriggersUpdated?: () => Promise<void> | void;
}
const App = forwardRef<{ handleCopyChat: () => void; handleUserMessage: (message: string) => void }, AppProps>(function App({
@ -47,6 +49,8 @@ const App = forwardRef<{ handleCopyChat: () => void; handleUserMessage: (message
onMessagesChange,
isInitialState = false,
dataSources,
triggers,
onTriggersUpdated,
}, ref) {
@ -85,7 +89,8 @@ const App = forwardRef<{ handleCopyChat: () => void; handleUserMessage: (message
projectId,
workflow: workflowRef.current,
context: effectiveContext,
dataSources: dataSources
dataSources: dataSources,
triggers: triggers
});
// Store latest start/cancel functions in refs
@ -264,6 +269,8 @@ const App = forwardRef<{ handleCopyChat: () => void; handleUserMessage: (message
onStatusBarChange={handleStatusBarChange}
toolCalling={toolCalling}
toolQuery={toolQuery}
triggers={triggers}
onTriggersUpdated={onTriggersUpdated}
/>
</div>
<div className="shrink-0 px-0 pb-10">
@ -319,8 +326,10 @@ export const Copilot = forwardRef<{ handleUserMessage: (message: string) => void
dispatch: (action: WorkflowDispatch) => void;
isInitialState?: boolean;
dataSources?: z.infer<typeof DataSource>[];
triggers?: z.infer<typeof TriggerSchemaForCopilot>[];
activePanel?: 'playground' | 'copilot';
onTogglePanel?: () => void;
onTriggersUpdated?: () => Promise<void> | void;
}>(({
projectId,
workflow,
@ -328,8 +337,10 @@ export const Copilot = forwardRef<{ handleUserMessage: (message: string) => void
dispatch,
isInitialState = false,
dataSources,
triggers,
activePanel,
onTogglePanel,
onTriggersUpdated,
}, ref) => {
console.log('🎪 Copilot wrapper component mounted:', {
projectId,
@ -415,6 +426,8 @@ export const Copilot = forwardRef<{ handleUserMessage: (message: string) => void
onMessagesChange={setMessages}
isInitialState={isInitialState}
dataSources={dataSources}
triggers={triggers}
onTriggersUpdated={onTriggersUpdated}
/>
</div>
</Panel>

6
apps/rowboat/app/projects/[projectId]/copilot/components/actions.tsx
View file

@ -211,7 +211,7 @@ export function Action({
{action.config_type === 'tool' && toolkitLogo ? (
<PictureImg src={toolkitLogo} alt={"Toolkit logo"} className="h-5 w-5 object-contain" />
) : (
action.config_type === 'agent' ? '🧑‍💼' : action.config_type === 'tool' ? '🛠️' : action.config_type === 'pipeline' ? '⚙️' : action.config_type === 'start_agent' ? '🏁' : action.config_type === 'prompt' ? '💬' : action.config_type === 'one_time_trigger' ? '⏰' : action.config_type === 'recurring_trigger' ? '🔄' : '💬'
action.config_type === 'agent' ? '🧑‍💼' : action.config_type === 'tool' ? '🛠️' : action.config_type === 'pipeline' ? '⚙️' : action.config_type === 'start_agent' ? '🏁' : action.config_type === 'prompt' ? '💬' : action.config_type === 'one_time_trigger' ? '⏰' : action.config_type === 'recurring_trigger' ? '🔄' : action.config_type === 'external_trigger' ? '🔗' : '💬'
)}
</span>
<span className="font-semibold text-sm text-zinc-800 dark:text-zinc-100 truncate flex-1">
@ -379,7 +379,7 @@ export function StreamingAction({
}: {
action: {
action?: 'create_new' | 'edit' | 'delete';
config_type?: 'tool' | 'agent' | 'prompt' | 'pipeline' | 'start_agent' | 'one_time_trigger' | 'recurring_trigger';
config_type?: 'tool' | 'agent' | 'prompt' | 'pipeline' | 'start_agent' | 'one_time_trigger' | 'recurring_trigger' | 'external_trigger';
name?: string;
};
loading: boolean;
@ -418,7 +418,7 @@ export function StreamingAction({
'bg-gray-200 text-gray-600': !action.action,
}
)}>
{action.config_type === 'agent' ? '🧑‍💼' : action.config_type === 'tool' ? '🛠️' : action.config_type === 'pipeline' ? '⚙️' : action.config_type === 'start_agent' ? '🏁' : action.config_type === 'one_time_trigger' ? '⏰' : action.config_type === 'recurring_trigger' ? '🔄' : '💬'}
{action.config_type === 'agent' ? '🧑‍💼' : action.config_type === 'tool' ? '🛠️' : action.config_type === 'pipeline' ? '⚙️' : action.config_type === 'start_agent' ? '🏁' : action.config_type === 'one_time_trigger' ? '⏰' : action.config_type === 'recurring_trigger' ? '🔄' : action.config_type === 'external_trigger' ? '🔗' : '💬'}
</span>
<span className="font-semibold text-sm text-zinc-800 dark:text-zinc-100 truncate flex-1">
{action.action === 'create_new' ? 'Add' : action.action === 'edit' ? 'Edit' : 'Delete'} {action.config_type}: {action.name}

281
apps/rowboat/app/projects/[projectId]/copilot/components/messages.tsx
View file

@ -5,7 +5,7 @@ import { z } from "zod";
import { Workflow} from "@/app/lib/types/workflow_types";
import MarkdownContent from "@/app/lib/components/markdown-content";
import { MessageSquareIcon, EllipsisIcon, XIcon, CheckCheckIcon, ChevronDown, ChevronUp } from "lucide-react";
import { CopilotMessage, CopilotAssistantMessage, CopilotAssistantMessageActionPart } from "@/src/entities/models/copilot";
import { CopilotMessage, CopilotAssistantMessage, CopilotAssistantMessageActionPart, TriggerSchemaForCopilot } from "@/src/entities/models/copilot";
import { Action, StreamingAction } from './actions';
import { useParsedBlocks } from "../use-parsed-blocks";
import { validateConfigChanges } from "@/app/lib/client_utils";
@ -13,9 +13,13 @@ import { PreviewModalProvider } from '../../workflow/preview-modal';
type ScheduledJobActionsModule = typeof import('@/app/actions/scheduled-job-rules.actions');
type RecurringJobActionsModule = typeof import('@/app/actions/recurring-job-rules.actions');
type ComposioActionsModule = typeof import('@/app/actions/composio.actions');
type CopilotTriggerType = z.infer<typeof TriggerSchemaForCopilot>;
let scheduledJobActionsPromise: Promise<ScheduledJobActionsModule> | null = null;
let recurringJobActionsPromise: Promise<RecurringJobActionsModule> | null = null;
let composioActionsPromise: Promise<ComposioActionsModule> | null = null;
function loadScheduledJobActions(): Promise<ScheduledJobActionsModule> {
if (!scheduledJobActionsPromise) {
@ -31,6 +35,13 @@ function loadRecurringJobActions(): Promise<RecurringJobActionsModule> {
return recurringJobActionsPromise;
}
function loadComposioActions(): Promise<ComposioActionsModule> {
if (!composioActionsPromise) {
composioActionsPromise = import('@/app/actions/composio.actions');
}
return composioActionsPromise;
}
const CopilotResponsePart = z.union([
z.object({
type: z.literal('text'),
@ -91,7 +102,7 @@ function enrich(response: string): z.infer<typeof CopilotResponsePart> {
type: 'action',
action: {
action: metadata.action as 'create_new' | 'edit' | 'delete',
config_type: metadata.config_type as 'tool' | 'agent' | 'prompt' | 'pipeline' | 'start_agent' | 'one_time_trigger' | 'recurring_trigger',
config_type: metadata.config_type as 'tool' | 'agent' | 'prompt' | 'pipeline' | 'start_agent' | 'one_time_trigger' | 'recurring_trigger' | 'external_trigger',
name: metadata.name,
change_description: jsonData.change_description || '',
config_changes: {},
@ -104,7 +115,7 @@ function enrich(response: string): z.infer<typeof CopilotResponsePart> {
type: 'action',
action: {
action: metadata.action as 'create_new' | 'edit' | 'delete',
config_type: metadata.config_type as 'tool' | 'agent' | 'prompt' | 'pipeline' | 'start_agent' | 'one_time_trigger' | 'recurring_trigger',
config_type: metadata.config_type as 'tool' | 'agent' | 'prompt' | 'pipeline' | 'start_agent' | 'one_time_trigger' | 'recurring_trigger' | 'external_trigger',
name: metadata.name,
change_description: jsonData.change_description || '',
config_changes: result.changes
@ -120,7 +131,7 @@ function enrich(response: string): z.infer<typeof CopilotResponsePart> {
type: 'streaming_action',
action: {
action: (metadata.action as 'create_new' | 'edit' | 'delete') || undefined,
config_type: (metadata.config_type as 'tool' | 'agent' | 'prompt' | 'pipeline' | 'start_agent' | 'one_time_trigger' | 'recurring_trigger') || undefined,
config_type: (metadata.config_type as 'tool' | 'agent' | 'prompt' | 'pipeline' | 'start_agent' | 'one_time_trigger' | 'recurring_trigger' | 'external_trigger') || undefined,
name: metadata.name
}
};
@ -193,6 +204,8 @@ function AssistantMessage({
loading,
onStatusBarChange,
projectId,
triggers,
onTriggersUpdated,
}: {
content: z.infer<typeof CopilotAssistantMessage>['content'],
workflow: z.infer<typeof Workflow>,
@ -201,11 +214,24 @@ function AssistantMessage({
loading: boolean,
onStatusBarChange?: (status: any) => void;
projectId: string;
triggers?: CopilotTriggerType[];
onTriggersUpdated?: () => Promise<void> | void;
}) {
const blocks = useParsedBlocks(content);
const [appliedActions, setAppliedActions] = useState<Set<number>>(new Set());
// Remove autoApplyEnabled and useEffect for auto-apply
const triggersRef = useRef<CopilotTriggerType[] | undefined>(triggers);
const triggerUpdateCallbackRef = useRef<typeof onTriggersUpdated>(onTriggersUpdated);
useEffect(() => {
triggersRef.current = triggers;
}, [triggers]);
useEffect(() => {
triggerUpdateCallbackRef.current = onTriggersUpdated;
}, [onTriggersUpdated]);
// parse actions from parts
const parsed = useMemo(() => {
const result: z.infer<typeof CopilotResponsePart>[] = [];
@ -353,54 +379,192 @@ function AssistantMessage({
}, [dispatch, workflow.agents, workflow.tools]);
const handleTriggerAction = useCallback(async (action: any): Promise<boolean> => {
if (action.action !== 'create_new') {
const configType = action.config_type;
const actionType = action.action;
const triggerList = triggersRef.current ?? [];
try {
if (configType === 'one_time_trigger') {
if (actionType === 'create_new') {
const { scheduledTime, input } = action.config_changes || {};
if (!scheduledTime || !input) {
console.error('Missing scheduledTime or input for one-time trigger', action);
return false;
}
const { createScheduledJobRule } = await loadScheduledJobActions();
await createScheduledJobRule({
projectId,
scheduledTime,
input,
});
return true;
}
const target = triggerList.find(
(trigger): trigger is Extract<z.infer<typeof TriggerSchemaForCopilot>, { type: 'one_time' }> =>
trigger.type === 'one_time' && trigger.name === action.name
);
if (!target) {
console.warn('Unable to resolve one-time trigger for action', action.name);
return false;
}
const { fetchScheduledJobRule, deleteScheduledJobRule, createScheduledJobRule } = await loadScheduledJobActions();
if (actionType === 'delete') {
await deleteScheduledJobRule({ projectId, ruleId: target.id });
return true;
}
if (actionType === 'edit') {
const existing = await fetchScheduledJobRule({ ruleId: target.id });
if (!existing) {
console.error('Failed to load existing one-time trigger for edit', action.name);
return false;
}
const scheduledTime = action.config_changes?.scheduledTime ?? existing.nextRunAt;
const input = action.config_changes?.input ?? existing.input;
if (!scheduledTime || !input) {
console.error('Missing data for one-time trigger edit', action);
return false;
}
const created = await createScheduledJobRule({
projectId,
scheduledTime,
input,
});
// Remove the previous rule only after successfully creating the updated one
await deleteScheduledJobRule({ projectId, ruleId: target.id });
return Boolean(created?.id);
}
}
if (configType === 'recurring_trigger') {
if (actionType === 'create_new') {
const { cron, input } = action.config_changes || {};
if (!cron || !input) {
console.error('Missing cron or input for recurring trigger', action);
return false;
}
const { createRecurringJobRule } = await loadRecurringJobActions();
await createRecurringJobRule({
projectId,
cron,
input,
});
return true;
}
const target = triggerList.find(
(trigger): trigger is Extract<z.infer<typeof TriggerSchemaForCopilot>, { type: 'recurring' }> =>
trigger.type === 'recurring' && trigger.name === action.name
);
if (!target) {
console.warn('Unable to resolve recurring trigger for action', action.name);
return false;
}
const {
fetchRecurringJobRule,
deleteRecurringJobRule,
createRecurringJobRule,
toggleRecurringJobRule,
} = await loadRecurringJobActions();
if (actionType === 'delete') {
await deleteRecurringJobRule({ projectId, ruleId: target.id });
return true;
}
if (actionType === 'edit') {
const existing = await fetchRecurringJobRule({ ruleId: target.id });
if (!existing) {
console.error('Failed to load existing recurring trigger for edit', action.name);
return false;
}
const desiredDisabled = typeof action.config_changes?.disabled === 'boolean'
? action.config_changes.disabled
: existing.disabled;
const hasCronChange = Object.prototype.hasOwnProperty.call(action.config_changes ?? {}, 'cron');
const hasInputChange = Object.prototype.hasOwnProperty.call(action.config_changes ?? {}, 'input');
const hasDisabledToggle = Object.prototype.hasOwnProperty.call(action.config_changes ?? {}, 'disabled');
if (!hasCronChange && !hasInputChange && hasDisabledToggle) {
if (desiredDisabled !== existing.disabled) {
await toggleRecurringJobRule({ ruleId: target.id, disabled: desiredDisabled });
}
return true;
}
const cron = action.config_changes?.cron ?? existing.cron;
const input = action.config_changes?.input ?? existing.input;
if (!cron || !input) {
console.error('Missing data for recurring trigger edit', action);
return false;
}
const created = await createRecurringJobRule({
projectId,
cron,
input,
});
await deleteRecurringJobRule({ projectId, ruleId: target.id });
if (desiredDisabled !== created.disabled) {
await toggleRecurringJobRule({ ruleId: created.id, disabled: desiredDisabled });
}
return true;
}
}
if (configType === 'external_trigger' && actionType === 'delete') {
const target = triggerList.find(
(trigger): trigger is Extract<z.infer<typeof TriggerSchemaForCopilot>, { type: 'external' }> =>
trigger.type === 'external' && trigger.triggerTypeName === action.name
);
if (!target) {
console.warn('Unable to resolve external trigger for action', action.name);
return false;
}
const { deleteComposioTriggerDeployment } = await loadComposioActions();
await deleteComposioTriggerDeployment({ projectId, deploymentId: target.id });
return true;
}
} catch (error) {
console.error('Failed to handle trigger action', action, error);
return false;
}
if (action.config_type === 'one_time_trigger') {
const { scheduledTime, input } = action.config_changes || {};
if (!scheduledTime || !input) {
console.error('Missing scheduledTime or input for one-time trigger', action);
return false;
}
try {
const { createScheduledJobRule } = await loadScheduledJobActions();
await createScheduledJobRule({
projectId,
scheduledTime,
input,
});
return true;
} catch (error) {
console.error('Failed to create one-time trigger', error);
return false;
}
}
if (action.config_type === 'recurring_trigger') {
const { cron, input } = action.config_changes || {};
if (!cron || !input) {
console.error('Missing cron or input for recurring trigger', action);
return false;
}
try {
const { createRecurringJobRule } = await loadRecurringJobActions();
await createRecurringJobRule({
projectId,
cron,
input,
});
return true;
} catch (error) {
console.error('Failed to create recurring trigger', error);
return false;
}
}
console.warn('Unhandled trigger action from Copilot applyAction', action);
return false;
}, [projectId]);
const refreshTriggers = useCallback(async () => {
const callback = triggerUpdateCallbackRef.current;
if (!callback) {
return;
}
try {
await callback();
} catch (error) {
console.error('Failed to refresh triggers after Copilot action', error);
}
}, []);
// Memoized handleApplyAll for useEffect dependencies
const handleApplyAll = useCallback(async () => {
const unapplied = parsed.reduce<Array<{ action: z.infer<typeof CopilotAssistantMessageActionPart>['content']; actionIndex: number }>>((acc, part, idx) => {
@ -411,16 +575,20 @@ function AssistantMessage({
}, []);
const newlyApplied: number[] = [];
let triggerMutated = false;
for (const { action, actionIndex } of unapplied) {
try {
const isTrigger = action.config_type === 'one_time_trigger' || action.config_type === 'recurring_trigger';
const isTrigger = action.config_type === 'one_time_trigger' || action.config_type === 'recurring_trigger' || action.config_type === 'external_trigger';
const success = isTrigger
? await handleTriggerAction(action)
: applyAction(action);
if (success) {
newlyApplied.push(actionIndex);
if (isTrigger) {
triggerMutated = true;
}
}
} catch (error) {
console.error('Failed to apply Copilot action', action, error);
@ -434,7 +602,11 @@ function AssistantMessage({
return next;
});
}
}, [parsed, appliedActions, applyAction, handleTriggerAction]);
if (triggerMutated) {
await refreshTriggers();
}
}, [parsed, appliedActions, applyAction, handleTriggerAction, refreshTriggers]);
// Manual single apply (from card)
const handleSingleApply = useCallback(async (action: z.infer<typeof CopilotAssistantMessageActionPart>['content'], actionIndex: number) => {
@ -443,18 +615,21 @@ function AssistantMessage({
}
try {
const isTrigger = action.config_type === 'one_time_trigger' || action.config_type === 'recurring_trigger';
const isTrigger = action.config_type === 'one_time_trigger' || action.config_type === 'recurring_trigger' || action.config_type === 'external_trigger';
const success = isTrigger
? await handleTriggerAction(action)
: applyAction(action);
if (success) {
setAppliedActions(prev => new Set([...prev, actionIndex]));
if (isTrigger) {
await refreshTriggers();
}
}
} catch (error) {
console.error('Failed to apply Copilot action', action, error);
}
}, [appliedActions, applyAction, handleTriggerAction]);
}, [appliedActions, applyAction, handleTriggerAction, refreshTriggers]);
useEffect(() => {
if (loading) {
@ -607,7 +782,9 @@ export function Messages({
dispatch,
onStatusBarChange,
toolCalling,
toolQuery
toolQuery,
triggers,
onTriggersUpdated,
}: {
projectId: string;
messages: z.infer<typeof CopilotMessage>[];
@ -618,6 +795,8 @@ export function Messages({
onStatusBarChange?: (status: any) => void;
toolCalling?: boolean;
toolQuery?: string | null;
triggers?: z.infer<typeof TriggerSchemaForCopilot>[];
onTriggersUpdated?: () => Promise<void> | void;
}) {
const messagesEndRef = useRef<HTMLDivElement>(null);
const [displayMessages, setDisplayMessages] = useState(messages);
@ -660,6 +839,8 @@ export function Messages({
messageIndex={messageIndex}
loading={loadingResponse}
projectId={projectId}
triggers={triggers}
onTriggersUpdated={onTriggersUpdated}
onStatusBarChange={status => {
// Only update for the last assistant message
if (messageIndex === displayMessages.length - 1) {

8
apps/rowboat/app/projects/[projectId]/copilot/use-copilot.tsx
View file

@ -3,6 +3,7 @@ import { getCopilotResponseStream } from "@/app/actions/copilot.actions";
import { CopilotMessage } from "@/src/entities/models/copilot";
import { Workflow } from "@/app/lib/types/workflow_types";
import { DataSource } from "@/src/entities/models/data-source";
import { TriggerSchemaForCopilot } from "@/src/entities/models/copilot";
import { z } from "zod";
import { WithStringId } from "@/app/lib/types/types";
@ -11,6 +12,7 @@ interface UseCopilotParams {
workflow: z.infer<typeof Workflow>;
context: any;
dataSources?: z.infer<typeof DataSource>[];
triggers?: z.infer<typeof TriggerSchemaForCopilot>[];
}
interface UseCopilotResult {
@ -29,7 +31,7 @@ interface UseCopilotResult {
cancel: () => void;
}
export function useCopilot({ projectId, workflow, context, dataSources }: UseCopilotParams): UseCopilotResult {
export function useCopilot({ projectId, workflow, context, dataSources, triggers }: UseCopilotParams): UseCopilotResult {
const [streamingResponse, setStreamingResponse] = useState('');
const [loading, setLoading] = useState(false);
const [toolCalling, setToolCalling] = useState(false);
@ -77,7 +79,7 @@ export function useCopilot({ projectId, workflow, context, dataSources }: UseCop
// Wait 2 rAF frames to let layout stabilize (avoids StrictMode/remount race on initial load)
await new Promise<void>((resolve) => requestAnimationFrame(() => requestAnimationFrame(() => resolve())));
const res = await getCopilotResponseStream(projectId, messages, workflow, context || null, dataSources);
const res = await getCopilotResponseStream(projectId, messages, workflow, context || null, dataSources, triggers);
// Check for billing error
@ -139,7 +141,7 @@ export function useCopilot({ projectId, workflow, context, dataSources }: UseCop
setLoading(false);
inFlightRef.current = false;
}
}, [projectId, workflow, context, dataSources]);
}, [projectId, workflow, context, dataSources, triggers]);
const cancel = useCallback(() => {
cancelRef.current?.();

61
apps/rowboat/app/projects/[projectId]/workflow/app.tsx
View file

@ -1,5 +1,6 @@
"use client";
import { DataSource } from "@/src/entities/models/data-source";
import { TriggerSchemaForCopilot } from "@/src/entities/models/copilot";
import { Project } from "@/src/entities/models/project";
import { z } from "zod";
import { useCallback, useEffect, useState } from "react";
@ -10,10 +11,15 @@ import { revertToLiveWorkflow } from "@/app/actions/project.actions";
import { fetchProject } from "@/app/actions/project.actions";
import { Workflow } from "@/app/lib/types/workflow_types";
import { ModelsResponse } from "@/app/lib/types/billing_types";
import { listScheduledJobRules } from "@/app/actions/scheduled-job-rules.actions";
import { listRecurringJobRules } from "@/app/actions/recurring-job-rules.actions";
import { listComposioTriggerDeployments } from "@/app/actions/composio.actions";
import { transformTriggersForCopilot, DEFAULT_TRIGGER_FETCH_LIMIT } from "./trigger-transform";
export function App({
initialProjectData,
initialDataSources,
initialTriggers,
eligibleModels,
useRag,
useRagUploads,
@ -24,6 +30,7 @@ export function App({
}: {
initialProjectData: z.infer<typeof Project>;
initialDataSources: z.infer<typeof DataSource>[];
initialTriggers: z.infer<typeof TriggerSchemaForCopilot>[];
eligibleModels: z.infer<typeof ModelsResponse> | "*";
useRag: boolean;
useRagUploads: boolean;
@ -44,6 +51,7 @@ export function App({
});
const [project, setProject] = useState<z.infer<typeof Project>>(initialProjectData);
const [dataSources, setDataSources] = useState<z.infer<typeof DataSource>[]>(initialDataSources);
const [triggers, setTriggers] = useState<z.infer<typeof TriggerSchemaForCopilot>[]>(initialTriggers);
const [loading, setLoading] = useState(false);
console.log('workflow app.tsx render');
@ -65,21 +73,42 @@ export function App({
workflow = mode === 'live' ? project?.liveWorkflow : project?.draftWorkflow;
}
const reloadData = useCallback(async () => {
setLoading(true);
const [
projectData,
sourcesData,
] = await Promise.all([
fetchProject(initialProjectData.id),
listDataSources(initialProjectData.id),
const fetchTriggers = useCallback(async () => {
const [scheduled, recurring, composio] = await Promise.all([
listScheduledJobRules({ projectId: initialProjectData.id, limit: DEFAULT_TRIGGER_FETCH_LIMIT }),
listRecurringJobRules({ projectId: initialProjectData.id, limit: DEFAULT_TRIGGER_FETCH_LIMIT }),
listComposioTriggerDeployments({ projectId: initialProjectData.id, limit: DEFAULT_TRIGGER_FETCH_LIMIT }),
]);
setProject(projectData);
setDataSources(sourcesData);
setLoading(false);
return transformTriggersForCopilot({
scheduled: scheduled.items ?? [],
recurring: recurring.items ?? [],
composio: composio.items ?? [],
});
}, [initialProjectData.id]);
const refreshTriggers = useCallback(async () => {
const nextTriggers = await fetchTriggers();
setTriggers(nextTriggers);
}, [fetchTriggers]);
const reloadData = useCallback(async () => {
setLoading(true);
try {
const [projectData, sourcesData, triggerData] = await Promise.all([
fetchProject(initialProjectData.id),
listDataSources(initialProjectData.id),
fetchTriggers(),
]);
setProject(projectData);
setDataSources(sourcesData);
setTriggers(triggerData);
} finally {
setLoading(false);
}
}, [fetchTriggers, initialProjectData.id]);
const handleProjectToolsUpdate = useCallback(async () => {
// Lightweight refresh for tool-only updates
const projectConfig = await fetchProject(initialProjectData.id);
@ -133,8 +162,12 @@ export function App({
async function handleRevertToLive() {
setLoading(true);
await revertToLiveWorkflow(initialProjectData.id);
reloadData();
try {
await revertToLiveWorkflow(initialProjectData.id);
await reloadData();
} finally {
setLoading(false);
}
}
// if workflow is null, show the selector
@ -152,6 +185,7 @@ export function App({
onToggleAutoPublish={handleToggleAutoPublish}
workflow={workflow}
dataSources={dataSources}
triggers={triggers}
projectConfig={project}
useRag={useRag}
useRagUploads={useRagUploads}
@ -164,6 +198,7 @@ export function App({
onProjectToolsUpdated={handleProjectToolsUpdate}
onDataSourcesUpdated={handleDataSourcesUpdate}
onProjectConfigUpdated={handleProjectConfigUpdate}
onTriggersUpdated={refreshTriggers}
chatWidgetHost={chatWidgetHost}
/>}
</>

44
apps/rowboat/app/projects/[projectId]/workflow/page.tsx
View file

@ -9,10 +9,17 @@ import { ModelsResponse } from "@/app/lib/types/billing_types";
import { requireAuth } from "@/app/lib/auth";
import { IFetchProjectController } from "@/src/interface-adapters/controllers/projects/fetch-project.controller";
import { IListDataSourcesController } from "@/src/interface-adapters/controllers/data-sources/list-data-sources.controller";
import { IListScheduledJobRulesController } from "@/src/interface-adapters/controllers/scheduled-job-rules/list-scheduled-job-rules.controller";
import { IListRecurringJobRulesController } from "@/src/interface-adapters/controllers/recurring-job-rules/list-recurring-job-rules.controller";
import { IListComposioTriggerDeploymentsController } from "@/src/interface-adapters/controllers/composio-trigger-deployments/list-composio-trigger-deployments.controller";
import { z } from "zod";
import { transformTriggersForCopilot, DEFAULT_TRIGGER_FETCH_LIMIT } from "./trigger-transform";
const fetchProjectController = container.resolve<IFetchProjectController>('fetchProjectController');
const listDataSourcesController = container.resolve<IListDataSourcesController>('listDataSourcesController');
const listScheduledJobRulesController = container.resolve<IListScheduledJobRulesController>('listScheduledJobRulesController');
const listRecurringJobRulesController = container.resolve<IListRecurringJobRulesController>('listRecurringJobRulesController');
const listComposioTriggerDeploymentsController = container.resolve<IListComposioTriggerDeploymentsController>('listComposioTriggerDeploymentsController');
const DEFAULT_MODEL = process.env.PROVIDER_DEFAULT_MODEL || "gpt-4.1";
@ -39,23 +46,50 @@ export default async function Page(
notFound();
}
const sources = await listDataSourcesController.execute({
caller: "user",
userId: user.id,
projectId: params.projectId,
});
const [sources, scheduledTriggers, recurringTriggers, composioTriggers] = await Promise.all([
listDataSourcesController.execute({
caller: "user",
userId: user.id,
projectId: params.projectId,
}),
listScheduledJobRulesController.execute({
caller: "user",
userId: user.id,
projectId: params.projectId,
limit: DEFAULT_TRIGGER_FETCH_LIMIT,
}),
listRecurringJobRulesController.execute({
caller: "user",
userId: user.id,
projectId: params.projectId,
limit: DEFAULT_TRIGGER_FETCH_LIMIT,
}),
listComposioTriggerDeploymentsController.execute({
caller: "user",
userId: user.id,
projectId: params.projectId,
limit: DEFAULT_TRIGGER_FETCH_LIMIT,
}),
]);
let eligibleModels: z.infer<typeof ModelsResponse> | "*" = '*';
if (USE_BILLING) {
eligibleModels = await getEligibleModels(customer.id);
}
const triggers = transformTriggersForCopilot({
scheduled: scheduledTriggers.items ?? [],
recurring: recurringTriggers.items ?? [],
composio: composioTriggers.items ?? [],
});
console.log('/workflow page.tsx serve');
return (
<App
initialProjectData={project}
initialDataSources={sources}
initialTriggers={triggers}
eligibleModels={eligibleModels}
useRag={USE_RAG}
useRagUploads={USE_RAG_UPLOADS}

77
apps/rowboat/app/projects/[projectId]/workflow/trigger-transform.ts Normal file
View file

@ -0,0 +1,77 @@
import { z } from "zod";
import { TriggerSchemaForCopilot } from "@/src/entities/models/copilot";
import { Message } from "@/app/lib/types/types";
const COPILOT_TRIGGER_LIMIT = 100;
export const DEFAULT_TRIGGER_FETCH_LIMIT = COPILOT_TRIGGER_LIMIT;
export type CopilotTrigger = z.infer<typeof TriggerSchemaForCopilot>;
interface TransformParams {
scheduled: Array<{
id: string;
nextRunAt: string;
status: 'pending' | 'processing' | 'triggered';
input?: { messages: Array<z.infer<typeof Message>> };
}>;
recurring: Array<{
id: string;
cron: string;
nextRunAt: string | null;
disabled: boolean;
input?: { messages: Array<z.infer<typeof Message>> };
}>;
composio: Array<{
id: string;
triggerTypeName: string;
toolkitSlug: string;
triggerTypeSlug: string;
triggerConfig: Record<string, unknown>;
}>;
}
export function transformTriggersForCopilot({
scheduled,
recurring,
composio,
}: TransformParams): CopilotTrigger[] {
const placeholderInput = {
messages: [
{
role: "user" as const,
content: "Trigger execution",
},
],
} satisfies { messages: Array<z.infer<typeof Message>> };
const oneTime = scheduled.map((trigger) => ({
type: "one_time" as const,
id: trigger.id,
name: `One-time trigger (${new Date(trigger.nextRunAt).toLocaleDateString('en-US')})`,
nextRunAt: trigger.nextRunAt,
status: trigger.status,
input: trigger.input ?? placeholderInput,
}));
const recurringTriggers = recurring.map((trigger) => ({
type: "recurring" as const,
id: trigger.id,
name: `Recurring trigger (${trigger.cron})`,
cron: trigger.cron,
nextRunAt: trigger.nextRunAt ?? '',
disabled: trigger.disabled,
input: trigger.input ?? placeholderInput,
}));
const external = composio.map((trigger) => ({
type: "external" as const,
id: trigger.id,
triggerTypeName: trigger.triggerTypeName,
toolkitSlug: trigger.toolkitSlug,
triggerTypeSlug: trigger.triggerTypeSlug,
triggerConfig: trigger.triggerConfig,
}));
return [...oneTime, ...recurringTriggers, ...external] as CopilotTrigger[];
}

7
apps/rowboat/app/projects/[projectId]/workflow/workflow_editor.tsx
View file

@ -3,6 +3,7 @@ import React, { useReducer, Reducer, useState, useCallback, useEffect, useRef, c
import { MCPServer, Message, WithStringId } from "../../../lib/types/types";
import { Workflow, WorkflowTool, WorkflowPrompt, WorkflowAgent, WorkflowPipeline } from "../../../lib/types/workflow_types";
import { DataSource } from "@/src/entities/models/data-source";
import { TriggerSchemaForCopilot } from "@/src/entities/models/copilot";
import { Project } from "@/src/entities/models/project";
import { produce, applyPatches, enablePatches, produceWithPatches, Patch } from 'immer';
import { AgentConfig } from "../entities/agent_config";
@ -962,6 +963,7 @@ export function useEntitySelection() {
export function WorkflowEditor({
projectId,
dataSources,
triggers,
workflow,
useRag,
useRagUploads,
@ -978,10 +980,12 @@ export function WorkflowEditor({
onProjectToolsUpdated,
onDataSourcesUpdated,
onProjectConfigUpdated,
onTriggersUpdated,
chatWidgetHost,
}: {
projectId: string;
dataSources: z.infer<typeof DataSource>[];
triggers: z.infer<typeof TriggerSchemaForCopilot>[];
workflow: z.infer<typeof Workflow>;
useRag: boolean;
useRagUploads: boolean;
@ -998,6 +1002,7 @@ export function WorkflowEditor({
onProjectToolsUpdated?: () => void;
onDataSourcesUpdated?: () => void;
onProjectConfigUpdated?: () => void;
onTriggersUpdated?: () => Promise<void> | void;
chatWidgetHost: string;
}) {
@ -2313,8 +2318,10 @@ export function WorkflowEditor({
}
isInitialState={isInitialState}
dataSources={dataSources}
triggers={triggers}
activePanel={activePanel}
onTogglePanel={handleTogglePanel}
onTriggersUpdated={onTriggersUpdated}
/>
{/* Config overlay above Copilot when agents + skipper layout is active */}
{state.present.selection && viewMode === 'two_agents_skipper' && (

64
apps/rowboat/src/application/lib/copilot/copilot.ts
View file

@ -2,7 +2,7 @@ import z from "zod";
import { createOpenAI } from "@ai-sdk/openai";
import { generateObject, streamText, tool } from "ai";
import { Workflow, WorkflowTool } from "@/app/lib/types/workflow_types";
import { CopilotChatContext, CopilotMessage, DataSourceSchemaForCopilot } from "../../../entities/models/copilot";
import { CopilotChatContext, CopilotMessage, DataSourceSchemaForCopilot, TriggerSchemaForCopilot } from "../../../entities/models/copilot";
import { PrefixLogger } from "@/app/lib/utils";
import zodToJsonSchema from "zod-to-json-schema";
import { COPILOT_INSTRUCTIONS_EDIT_AGENT } from "./copilot_edit_agent";
@ -100,6 +100,51 @@ function getCurrentTimePrompt(): string {
return `**CURRENT TIME**: ${new Date().toISOString()}`;
}
function getTriggersPrompt(triggers: z.infer<typeof TriggerSchemaForCopilot>[]): string {
if (!triggers || triggers.length === 0) {
return '';
}
const simplifiedTriggers = triggers.map(trigger => {
if (trigger.type === 'one_time') {
return {
id: trigger.id,
type: 'one_time',
name: trigger.name,
scheduledTime: trigger.nextRunAt,
input: trigger.input,
status: trigger.status,
};
} else if (trigger.type === 'recurring') {
return {
id: trigger.id,
type: 'recurring',
name: trigger.name,
cron: trigger.cron,
nextRunAt: trigger.nextRunAt,
disabled: trigger.disabled,
input: trigger.input,
};
} else {
return {
id: trigger.id,
type: 'external',
name: trigger.triggerTypeName,
toolkit: trigger.toolkitSlug,
triggerType: trigger.triggerTypeSlug,
config: trigger.triggerConfig,
};
}
});
return `**NOTE**:
The following triggers are currently configured:
\`\`\`json
${JSON.stringify(simplifiedTriggers)}
\`\`\`
`;
}
async function searchRelevantTools(usageTracker: UsageTracker, query: string): Promise<string> {
const logger = new PrefixLogger("copilot-search-tools");
console.log("🔧 TOOL CALL: searchRelevantTools", { query });
@ -189,10 +234,11 @@ function updateLastUserMessage(
contextPrompt: string,
dataSourcesPrompt: string = '',
timePrompt: string = '',
triggersPrompt: string = '',
): void {
const lastMessage = messages[messages.length - 1];
if (lastMessage.role === 'user') {
lastMessage.content = `${currentWorkflowPrompt}\n\n${contextPrompt}\n\n${dataSourcesPrompt}\n\n${timePrompt}\n\nUser: ${JSON.stringify(lastMessage.content)}`;
lastMessage.content = `${currentWorkflowPrompt}\n\n${contextPrompt}\n\n${dataSourcesPrompt}\n\n${timePrompt}\n\n${triggersPrompt}\n\nUser: ${JSON.stringify(lastMessage.content)}`;
}
}
@ -202,6 +248,7 @@ export async function getEditAgentInstructionsResponse(
context: z.infer<typeof CopilotChatContext> | null,
messages: z.infer<typeof CopilotMessage>[],
workflow: z.infer<typeof Workflow>,
triggers: z.infer<typeof TriggerSchemaForCopilot>[] = [],
): Promise<string> {
const logger = new PrefixLogger('copilot /getUpdatedAgentInstructions');
logger.log('context', context);
@ -216,8 +263,11 @@ export async function getEditAgentInstructionsResponse(
// set time prompt
let timePrompt = getCurrentTimePrompt();
// set triggers prompt
let triggersPrompt = getTriggersPrompt(triggers);
// add the above prompts to the last user message
updateLastUserMessage(messages, currentWorkflowPrompt, contextPrompt, '', timePrompt);
updateLastUserMessage(messages, currentWorkflowPrompt, contextPrompt, '', timePrompt, triggersPrompt);
// call model
console.log("calling model", JSON.stringify({
@ -257,7 +307,8 @@ export async function* streamMultiAgentResponse(
context: z.infer<typeof CopilotChatContext> | null,
messages: z.infer<typeof CopilotMessage>[],
workflow: z.infer<typeof Workflow>,
dataSources: z.infer<typeof DataSourceSchemaForCopilot>[]
dataSources: z.infer<typeof DataSourceSchemaForCopilot>[],
triggers: z.infer<typeof TriggerSchemaForCopilot>[] = []
): AsyncIterable<z.infer<typeof CopilotStreamEvent>> {
const logger = new PrefixLogger('copilot /stream');
logger.log('context', context);
@ -282,8 +333,11 @@ export async function* streamMultiAgentResponse(
// set time prompt
let timePrompt = getCurrentTimePrompt();
// set triggers prompt
let triggersPrompt = getTriggersPrompt(triggers);
// add the above prompts to the last user message
updateLastUserMessage(messages, currentWorkflowPrompt, contextPrompt, dataSourcesPrompt, timePrompt);
updateLastUserMessage(messages, currentWorkflowPrompt, contextPrompt, dataSourcesPrompt, timePrompt, triggersPrompt);
// call model
console.log("🤖 AI MODEL CALL STARTED", {

33
apps/rowboat/src/application/lib/copilot/copilot_multi_agent_build.ts
View file

@ -259,6 +259,39 @@ Recurring trigger example (COPY THIS EXACT FORMAT):
}
}
### Editing and Deleting Triggers
You can also edit or delete existing triggers that are shown in the current workflow context.
Edit trigger example:
// action: edit
// config_type: recurring_trigger
// name: Daily Status Check
{
"change_description": "Update the daily status check to run at 10 AM instead of 9 AM",
"config_changes": {
"cron": "0 10 * * *"
}
}
Delete trigger example:
// action: delete
// config_type: one_time_trigger
// name: Weekly Report - Dec 15
{
"change_description": "Remove the one-time trigger for weekly report as it's no longer needed"
}
### External Triggers
External triggers (from Composio integrations) can also be deleted:
// action: delete
// config_type: external_trigger
// name: Slack Message Received
{
"change_description": "Remove the Slack message trigger as we're switching to a different notification system"
}
</about_triggers>
<about_pipelines>

3
apps/rowboat/src/application/use-cases/copilot/create-copilot-cached-turn.use-case.ts
View file

@ -3,7 +3,7 @@ import { nanoid } from 'nanoid';
import { ICacheService } from '@/src/application/services/cache.service.interface';
import { IUsageQuotaPolicy } from '@/src/application/policies/usage-quota.policy.interface';
import { IProjectActionAuthorizationPolicy } from '@/src/application/policies/project-action-authorization.policy';
import { CopilotChatContext, CopilotMessage, DataSourceSchemaForCopilot } from '@/src/entities/models/copilot';
import { CopilotChatContext, CopilotMessage, DataSourceSchemaForCopilot, TriggerSchemaForCopilot } from '@/src/entities/models/copilot';
import { Workflow } from '@/app/lib/types/workflow_types';
import { USE_BILLING } from "@/app/lib/feature_flags";
import { authorize, getCustomerIdForProject } from "@/app/lib/billing";
@ -19,6 +19,7 @@ const inputSchema = z.object({
workflow: Workflow,
context: CopilotChatContext.nullable(),
dataSources: z.array(DataSourceSchemaForCopilot).optional(),
triggers: z.array(TriggerSchemaForCopilot).optional(),
}),
});

1
apps/rowboat/src/application/use-cases/copilot/run-copilot-cached-turn.use-case.ts
View file

@ -90,6 +90,7 @@ export class RunCopilotCachedTurnUseCase implements IRunCopilotCachedTurnUseCase
cachedTurn.messages,
cachedTurn.workflow,
cachedTurn.dataSources || [],
cachedTurn.triggers || [],
)) {
yield event;
}

43
apps/rowboat/src/entities/models/copilot.ts
View file

@ -2,6 +2,9 @@ import { z } from "zod";
import { Workflow } from "@/app/lib/types/workflow_types";
import { Message } from "@/app/lib/types/types";
import { DataSource } from "@/src/entities/models/data-source";
import { ScheduledJobRule } from "@/src/entities/models/scheduled-job-rule";
import { RecurringJobRule } from "@/src/entities/models/recurring-job-rule";
import { ComposioTriggerDeployment } from "@/src/entities/models/composio-trigger-deployment";
export const DataSourceSchemaForCopilot = DataSource.pick({
id: true,
@ -10,6 +13,43 @@ export const DataSourceSchemaForCopilot = DataSource.pick({
data: true,
});
export const ScheduledJobRuleSchemaForCopilot = ScheduledJobRule.pick({
id: true,
nextRunAt: true,
status: true,
input: true,
}).extend({
type: z.literal('one_time'),
name: z.string(),
});
export const RecurringJobRuleSchemaForCopilot = RecurringJobRule.pick({
id: true,
cron: true,
nextRunAt: true,
disabled: true,
input: true,
}).extend({
type: z.literal('recurring'),
name: z.string(),
});
export const ComposioTriggerDeploymentSchemaForCopilot = ComposioTriggerDeployment.pick({
id: true,
triggerTypeName: true,
toolkitSlug: true,
triggerTypeSlug: true,
triggerConfig: true,
}).extend({
type: z.literal('external'),
});
export const TriggerSchemaForCopilot = z.union([
ScheduledJobRuleSchemaForCopilot,
RecurringJobRuleSchemaForCopilot,
ComposioTriggerDeploymentSchemaForCopilot,
]);
export const CopilotUserMessage = z.object({
role: z.literal('user'),
content: z.string(),
@ -21,7 +61,7 @@ export const CopilotAssistantMessageTextPart = z.object({
export const CopilotAssistantMessageActionPart = z.object({
type: z.literal("action"),
content: z.object({
config_type: z.enum(['tool', 'agent', 'prompt', 'pipeline', 'start_agent', 'one_time_trigger', 'recurring_trigger']),
config_type: z.enum(['tool', 'agent', 'prompt', 'pipeline', 'start_agent', 'one_time_trigger', 'recurring_trigger', 'external_trigger']),
action: z.enum(['create_new', 'edit', 'delete']),
name: z.string(),
change_description: z.string(),
@ -60,6 +100,7 @@ export const CopilotAPIRequest = z.object({
workflow: Workflow,
context: CopilotChatContext.nullable(),
dataSources: z.array(DataSourceSchemaForCopilot).optional(),
triggers: z.array(TriggerSchemaForCopilot).optional(),
});
export const CopilotAPIResponse = z.union([
z.object({

3
apps/rowboat/src/interface-adapters/controllers/copilot/create-copilot-cached-turn.controller.ts
View file

@ -1,5 +1,5 @@
import { z } from "zod";
import { CopilotChatContext, CopilotMessage, DataSourceSchemaForCopilot } from '@/src/entities/models/copilot';
import { CopilotChatContext, CopilotMessage, DataSourceSchemaForCopilot, TriggerSchemaForCopilot } from '@/src/entities/models/copilot';
import { Workflow } from '@/app/lib/types/workflow_types';
import { ICreateCopilotCachedTurnUseCase } from "@/src/application/use-cases/copilot/create-copilot-cached-turn.use-case";
import { BadRequestError } from "@/src/entities/errors/common";
@ -14,6 +14,7 @@ const inputSchema = z.object({
workflow: Workflow,
context: CopilotChatContext.nullable(),
dataSources: z.array(DataSourceSchemaForCopilot).optional(),
triggers: z.array(TriggerSchemaForCopilot).optional(),
}),
});