task_manager.hpp

Go to the documentation of this file.

// Copyright (c) 2025 [caomengxuan666]
 
#pragma once
 
 
#include <chrono>
#include <cstdint>
#include <functional>
#include <mutex>
#include <optional>
#include <string>
#include <string_view>
#include <unordered_map>
#include <vector>
 
#include "cxxmcp/core/executor.hpp"
#include "cxxmcp/core/result.hpp"
#include "cxxmcp/protocol/task.hpp"
#include "cxxmcp/protocol/tool.hpp"
#include "cxxmcp/server/registry.hpp"
#include "cxxmcp/server/transport.hpp"
 
namespace mcp::server {
 
using TaskStatusNotificationHook = std::function<void(const protocol::Task&)>;
 
using TaskProgressNotificationHook =
    std::function<void(const protocol::ProgressNotificationParams&)>;
 
struct TaskOperationProcessorOptions {
  std::size_t worker_count = 4;
  std::size_t queue_size = 64;
  std::chrono::seconds default_timeout{300};
  std::optional<std::int64_t> poll_interval;
  std::optional<std::chrono::milliseconds> completed_task_ttl;
  std::size_t max_completed_tasks = 128;
  std::size_t list_page_size = 0;
  TaskStatusNotificationHook task_status_hook;
  TaskProgressNotificationHook task_progress_hook;
  bool emit_progress_for_task_state_changes = true;
};
 
struct TaskOperationDescriptor {
  std::string name;
  std::optional<protocol::TaskRequestParameters> task;
  std::optional<protocol::ProgressToken> progress_token;
};
 
using TaskOperationHandler =
    std::function<core::Result<protocol::Json>(const CancellationToken&)>;
 
class TaskOperationProcessor {
 public:
  explicit TaskOperationProcessor(TaskOperationProcessorOptions options = {});
 
  TaskOperationProcessor(const TaskOperationProcessor&) = delete;
  TaskOperationProcessor& operator=(const TaskOperationProcessor&) = delete;
 
  void stop() noexcept;
 
  core::Result<protocol::CreateTaskResult> submit_operation(
      TaskOperationDescriptor descriptor, TaskOperationHandler operation);
 
  core::Result<protocol::CreateTaskResult> submit_tool_call(
      const ToolRegistry& tools, protocol::ToolCall call,
      const SessionContext& context,
      const JsonSchemaValidator* schema_validator = nullptr);
 
  core::Result<protocol::TaskListResult> list_tasks(
      const protocol::TaskListParams& params = {});
 
  core::Result<protocol::Task> get_task(const protocol::TaskGetParams& params);
 
  core::Result<protocol::Task> cancel_task(
      const protocol::TaskCancelParams& params);
 
  core::Result<protocol::Json> task_result(
      const protocol::TaskResultParams& params);
 
 private:
  struct TaskRecord {
    protocol::Task task;
    std::optional<protocol::ProgressToken> progress_token;
    std::chrono::steady_clock::time_point started_at;
    std::optional<std::chrono::steady_clock::time_point> terminal_at;
    std::optional<std::chrono::seconds> timeout;
    CancellationSource cancellation;
    std::optional<protocol::Json> result;
    std::optional<core::Error> failure;
  };
 
  struct TaskUpdate {
    protocol::Task task;
    std::optional<protocol::ProgressToken> progress_token;
  };
 
  static std::string now_timestamp();
  static bool is_terminal(protocol::TaskStatus status) noexcept;
 
  std::string make_task_id();
  core::Error task_not_found_error(std::string_view task_id) const;
  TaskRecord* find_task_locked(std::string_view task_id);
  const TaskRecord* find_task_locked(std::string_view task_id) const;
  void finish_task(std::string task_id, core::Result<protocol::Json> result);
  void refresh_locked(std::vector<TaskUpdate>* updates = nullptr);
  void trim_completed_locked();
  void append_update_locked(std::vector<TaskUpdate>* updates,
                            const TaskRecord& record) const;
  void emit_updates(const std::vector<TaskUpdate>& updates) const;
 
  TaskOperationProcessorOptions options_;
  core::Executor executor_;
  mutable std::mutex mutex_;
  std::unordered_map<std::string, TaskRecord> tasks_;
  std::vector<std::string> order_;
  std::uint64_t next_task_index_ = 1;
};
 
}  // namespace mcp::server