feat(workflow): add pause and resume functionality to team workflow

[anthonydevs17]

No description provided.

[darielnoel]

We must follow the Flux architecture as much as possible because:

1. KaibanJS is built on the idea that Stores are the brain, while Agents and Tasks are the actors
2. When state management is scattered across components, debugging becomes a nightmare
3. Having a single source of truth (Stores) makes the system predictable and reliable
4. As we add more features, centralized state management will keep the codebase maintainable

Current Issues:

1. Direct State Management

• The Team class is directly managing workflow state through store.setState
• State transition logic is mixed with the Team class responsibilities
• Violates Flux pattern where stores should manage state

2. Validation Logic Location

• State transition validations are in the Team class
• These validations are part of the workflow state machine and should be in the store

3. Error Handling

• Error messages are hardcoded in the Team class
• Error handling strategy should be consistent and managed by stores

Suggested Implementation:

class Team {
  /**
   * Pauses the team's workflow.
   * @returns {void}
   */
  pause() {
    this.store.getState().pauseWorkflow();
  }

  /**
   * Resumes the team's workflow.
   * @returns {void}
   */
  resume() {
    this.store.getState().resumeWorkflow();
  }

  /**
   * Stops the team's workflow.
   * @returns {void}
   */
  stop() {
    this.store.getState().stopWorkflow();
  }
}

And in the workflow store:

// In workflowStore.js
const workflowStore = {
  teamWorkflowStatus: WORKFLOW_STATUS_enum.INITIAL,

  // State transition validations
  allowedTransitions: {
    [WORKFLOW_STATUS_enum.RUNNING]: [WORKFLOW_STATUS_enum.PAUSED, WORKFLOW_STATUS_enum.STOPPING],
    [WORKFLOW_STATUS_enum.PAUSED]: [WORKFLOW_STATUS_enum.RUNNING, WORKFLOW_STATUS_enum.STOPPING],
    [WORKFLOW_STATUS_enum.STOPPING]: [WORKFLOW_STATUS_enum.STOPPED],
  },

  // Actions
  pauseWorkflow: () => {
    const currentStatus = get().teamWorkflowStatus;
    get().validateTransition(currentStatus, WORKFLOW_STATUS_enum.PAUSED);
    set({ teamWorkflowStatus: WORKFLOW_STATUS_enum.PAUSED });
  },

  resumeWorkflow: () => {
    const currentStatus = get().teamWorkflowStatus;
    get().validateTransition(currentStatus, WORKFLOW_STATUS_enum.RUNNING);
    set({ teamWorkflowStatus: WORKFLOW_STATUS_enum.RUNNING });
  },

  stopWorkflow: () => {
    const currentStatus = get().teamWorkflowStatus;
    get().validateTransition(currentStatus, WORKFLOW_STATUS_enum.STOPPING);
    set({ teamWorkflowStatus: WORKFLOW_STATUS_enum.STOPPING });
  },

  // Validation helper
  validateTransition: (from, to) => {
    const allowedNext = get().allowedTransitions[from] || [];
    if (!allowedNext.includes(to)) {
      throw new WorkflowError(
        'INVALID_TRANSITION',
        `Cannot transition from ${from} to ${to}`
      );
    }
  }
}

Benefits of this Approach:

1. Clear Separation of Concerns

   • Team class only dispatches actions
   • Store manages all state transitions and validations
   • Workflow logic is centralized

2. Better State Management

   • All state transitions are tracked in one place
   • Explicit state machine through allowedTransitions
   • Consistent error handling

3. More Maintainable

   • Easier to add new workflow states
   • Clearer validation rules
   • Better testability

4. Follows Flux Pattern

   • Unidirectional data flow
   • State changes only happen in stores
   • Components dispatch actions

Would you like me to elaborate on any part of this suggestion?

[darielnoel]

FEEDBACK: Agent Loop Control

What the code is trying to do:

1. Pause the agent's thinking process when workflow status is PAUSED
2. Stop the agent's thinking process when workflow status is STOPPED
3. Return the current results when stopped
4. Wait and check periodically when paused

Issues with current approach:

1. Using a polling loop (while + setTimeout) is inefficient
2. Agent is directly managing workflow control
3. Multiple redundant store checks in the loop
4. Blocking execution with a busy-wait pattern

The intention is correct (controlling agent execution based on workflow state), but the implementation should leverage Flux patterns and store subscriptions instead of direct polling.

[darielnoel]

Another Note... I think this could be a direct approach....

The main caveat here is to store the intermmidiate Agentic Loop States, and be able to recover it once needed. In a pause or resume.

class ReactChampionAgent {
constructor(config) {
// ... other constructor logic ...

// Subscribe to workflow status changes
this.store.subscribe(
  (state) => state.teamWorkflowStatus,
  async (status) => {
    switch (status) {
      case WORKFLOW_STATUS_enum.PAUSED:
        await this.pauseLoop();
        break;
      case WORKFLOW_STATUS_enum.RUNNING:
        await this.resumeLoop();
        break;
      case WORKFLOW_STATUS_enum.STOPPED:
        // Handle stop if needed
        break;
    }
  }
);

}

#loopState = {
currentIteration: 0,
lastThinkingResult: null,
isPaused: false
};

async pauseLoop() {
this.#loopState.isPaused = true;
}

async resumeLoop() {
this.#loopState.isPaused = false;
await this.executeAgentLoop(this.currentTask, this.#loopState);
}
}.

[darielnoel]

LEt's try to do this control handler in an outer level... to try to follow the Flux principles...

The main thing is having a list of promises that we can kill anytime we needed in a directly way. Instead to wire each part to an external workflow controller.

Example

1. Promise Tracking

class ReactChampionAgent {
  #thinkingPromises = new Set();
  #toolPromises = new Set();

  async executeAgentLoop(task) {
    while (iterations < maxAgentIterations && !loopCriticalError) {
      try {
        // 1. Thinking Phase
        const thinkingPromise = new Promise((resolve, reject) => {
          this.#executableAgent.invoke(
            { feedbackMessage },
            {
              configurable: { sessionId: 'foo-bar-baz' },
              callbacks: [
                {
                  handleChatModelStart: async (llm, messages) => {
                    await agent.handleThinkingStart({ agent, task, messages });
                  },
                  handleLLMEnd: async (output) => {
                    const result = await agent.handleThinkingEnd({ agent, task, output });
                    resolve(result);
                  }
                }
              ]
            }
          ).catch(error => reject(error));
        });

        // Track thinking promise
        this.#thinkingPromises.add({ promise: thinkingPromise, reject });
        const thinkingResult = await thinkingPromise;
        this.#thinkingPromises.delete({ promise: thinkingPromise, reject });

        // 2. Tool Phase (if needed)
        if (thinkingResult.action) {
          const toolPromise = new Promise((resolve, reject) => {
            const tool = this.getToolByName(thinkingResult.action);
            tool.call(thinkingResult.actionInput)
              .then(result => resolve(result))
              .catch(error => reject(error));
          });

          // Track tool promise
          this.#toolPromises.add({ promise: toolPromise, reject });
          const toolResult = await toolPromise;
          this.#toolPromises.delete({ promise: toolPromise, reject });
        }

      } catch (error) {
        if (error instanceof AbortError) {
          // Clean exit if we're stopping/pausing
          break;
        }
        // Handle other errors
      }
    }
  }

  // Abort handler remains the same
  abortAllPromises() {
    for (const { reject } of this.#thinkingPromises) {
      reject(new AbortError('Workflow stopped or paused'));
    }
    this.#thinkingPromises.clear();

    for (const { reject } of this.#toolPromises) {
      reject(new AbortError('Workflow stopped or paused'));
    }
    this.#toolPromises.clear();
  }

  constructor(config) {
    super(config);
    this.store.subscribe(
      (state) => state.teamWorkflowStatus,
      (status) => {
        if (status === WORKFLOW_STATUS_enum.STOPPED || 
            status === WORKFLOW_STATUS_enum.PAUSED) {
          this.abortAllPromises();
        }
      }
    );
  }
}

Implementation Flow

1. Track all promises (thinking and tools)
2. Subscribe to workflow status changes
3. When paused/stopped, abort all tracked promises
4. Promises auto-cleanup when completed
5. AbortError triggers clean exit from agentic loop

[darielnoel]

Why Blocked tasks are being reset here to the Doing state?