feat(tasker): 添加 MaaTaskerOverridePipeline 接口支持通过 task_id 动态修改 pipeline (#1079)

MistEO · web-flow · commit 6b2f2be9e893 · 2026-01-20T22:53:42.000+08:00
diff --git a/docs/en_us/2.2-IntegratedInterfaceOverview.md b/docs/en_us/2.2-IntegratedInterfaceOverview.md
@@ -645,6 +645,13 @@ Get bound controller
 
 Clear all queryable information
 
+### MaaTaskerOverridePipeline
+
+- `task_id`: Task ID
+- `pipeline_override`: JSON for overriding
+
+Override the pipeline of a specified task, dynamically modify pipeline configuration during task execution.
+
 ### MaaTaskerGetRecognitionDetail
 
 - `reco_id`: Recognition id
diff --git a/docs/zh_cn/2.2-集成接口一览.md b/docs/zh_cn/2.2-集成接口一览.md
@@ -645,6 +645,13 @@
 
 清理所有可查询的信息
 
+### MaaTaskerOverridePipeline
+
+- `task_id`: 任务 ID
+- `pipeline_override`: 用于覆盖的 JSON
+
+覆盖指定任务的 pipeline，在任务执行期间动态修改 pipeline 配置。
+
 ### MaaTaskerGetRecognitionDetail
 
 - `reco_id`: 识别号
diff --git a/include/MaaFramework/Instance/MaaTasker.h b/include/MaaFramework/Instance/MaaTasker.h
@@ -88,6 +88,8 @@ extern "C"
 
     MAA_FRAMEWORK_API MaaBool MaaTaskerClearCache(MaaTasker* tasker);
 
+    MAA_FRAMEWORK_API MaaBool MaaTaskerOverridePipeline(MaaTasker* tasker, MaaTaskId task_id, const char* pipeline_override);
+
     /**
      * @param[out] hit
      */
diff --git a/source/Common/MaaTasker.cpp b/source/Common/MaaTasker.cpp
@@ -309,6 +309,29 @@ MaaBool MaaTaskerClearCache(MaaTasker* tasker)
     return true;
 }
 
+MaaBool MaaTaskerOverridePipeline(MaaTasker* tasker, MaaTaskId task_id, const char* pipeline_override)
+{
+    LogFunc << VAR_VOIDP(tasker) << VAR(task_id) << VAR(pipeline_override);
+
+    if (!tasker) {
+        LogError << "handle is null";
+        return false;
+    }
+
+    if (!pipeline_override) {
+        LogError << "pipeline_override is null";
+        return false;
+    }
+
+    auto ov_opt = json::parse(pipeline_override);
+    if (!ov_opt) {
+        LogError << "failed to parse" << VAR(pipeline_override);
+        return false;
+    }
+
+    return tasker->override_pipeline(task_id, *ov_opt);
+}
+
 #define CheckNullAndWarn(var)                        \
     if (!var) {                                      \
         LogWarn << #var << "is null, no assignment"; \
diff --git a/source/MaaAgentClient/Client/AgentClient.cpp b/source/MaaAgentClient/Client/AgentClient.cpp
@@ -298,6 +298,9 @@ bool AgentClient::handle_inserted_request(const json::value& j)
     else if (handle_tasker_clear_cache(j)) {
         return true;
     }
+    else if (handle_tasker_override_pipeline(j)) {
+        return true;
+    }
     else if (handle_tasker_get_task_detail(j)) {
         return true;
     }
@@ -1106,6 +1109,29 @@ bool AgentClient::handle_tasker_clear_cache(const json::value& j)
     return true;
 }
 
+bool AgentClient::handle_tasker_override_pipeline(const json::value& j)
+{
+    if (!j.is<TaskerOverridePipelineReverseRequest>()) {
+        return false;
+    }
+    const TaskerOverridePipelineReverseRequest& req = j.as<TaskerOverridePipelineReverseRequest>();
+    LogFunc << VAR(req) << VAR(ipc_addr_);
+
+    MaaTasker* tasker = query_tasker(req.tasker_id);
+    if (!tasker) {
+        LogError << "tasker not found" << VAR(req.tasker_id);
+        return false;
+    }
+
+    bool ret = tasker->override_pipeline(req.task_id, req.pipeline_override);
+
+    TaskerOverridePipelineReverseResponse resp {
+        .ret = ret,
+    };
+    send(resp);
+    return true;
+}
+
 bool AgentClient::handle_tasker_get_task_detail(const json::value& j)
 {
     if (!j.is<TaskerGetTaskDetailReverseRequest>()) {
diff --git a/source/MaaAgentClient/Client/AgentClient.h b/source/MaaAgentClient/Client/AgentClient.h
@@ -67,6 +67,7 @@ class AgentClient
     bool handle_tasker_resource(const json::value& j);
     bool handle_tasker_controller(const json::value& j);
     bool handle_tasker_clear_cache(const json::value& j);
+    bool handle_tasker_override_pipeline(const json::value& j);
     bool handle_tasker_get_task_detail(const json::value& j);
     bool handle_tasker_get_node_detail(const json::value& j);
     bool handle_tasker_get_reco_result(const json::value& j);
diff --git a/source/MaaAgentServer/RemoteInstance/RemoteTasker.cpp b/source/MaaAgentServer/RemoteInstance/RemoteTasker.cpp
@@ -204,6 +204,22 @@ void RemoteTasker::clear_cache()
     server_.send_and_recv<TaskerClearCacheReverseResponse>(req);
 }
 
+bool RemoteTasker::override_pipeline(MaaTaskId task_id, const json::value& pipeline_override)
+{
+    TaskerOverridePipelineReverseRequest req {
+        .tasker_id = tasker_id_,
+        .task_id = task_id,
+        .pipeline_override = pipeline_override,
+    };
+
+    auto resp_opt = server_.send_and_recv<TaskerOverridePipelineReverseResponse>(req);
+    if (!resp_opt) {
+        return false;
+    }
+
+    return resp_opt->ret;
+}
+
 std::optional<MAA_TASK_NS::TaskDetail> RemoteTasker::get_task_detail(MaaTaskId task_id) const
 {
     TaskerGetTaskDetailReverseRequest req {
diff --git a/source/MaaAgentServer/RemoteInstance/RemoteTasker.h b/source/MaaAgentServer/RemoteInstance/RemoteTasker.h
@@ -29,6 +29,8 @@ class RemoteTasker : public MaaTasker
         post_action(const std::string& action_type, const json::value& action_param, const cv::Rect& box, const std::string& reco_detail)
             override;
 
+    virtual bool override_pipeline(MaaTaskId task_id, const json::value& pipeline_override) override;
+
     virtual MaaStatus status(MaaTaskId task_id) const override;
     virtual MaaStatus wait(MaaTaskId task_id) const override;
 
diff --git a/source/MaaFramework/Base/AsyncRunner.hpp b/source/MaaFramework/Base/AsyncRunner.hpp
@@ -4,6 +4,7 @@
 #include <functional>
 #include <list>
 #include <mutex>
+#include <optional>
 #include <shared_mutex>
 #include <thread>
 
@@ -39,6 +40,8 @@ class AsyncRunner : public NonCopyable
     void wait(Id id) const;
     void wait_all() const;
     Status status(Id id) const;
+    std::optional<Item> get(Id id) const;
+    std::optional<Item> get_running() const;
 
     void clear();
     bool running() const;
@@ -49,10 +52,13 @@ class AsyncRunner : public NonCopyable
     ProcessFunc process_;
 
     std::list<std::pair<Id, Item>> queue_;
-    std::mutex queue_mutex_;
+    mutable std::mutex queue_mutex_;
     std::condition_variable queue_cond_;
     std::atomic_bool running_ = false;
 
+    Id running_id_ = 0;
+    Item running_item_ {};
+
     mutable std::shared_mutex status_mutex_;
     std::map<Id, Status> status_map_;
 
@@ -128,6 +134,9 @@ inline void AsyncRunner<Item>::working()
 
         auto [id, item] = std::move(queue_.front());
         queue_.pop_front();
+
+        running_id_ = id;
+        running_item_ = item;
         queue_lock.unlock();
 
         std::unique_lock status_lock(status_mutex_);
@@ -136,6 +145,12 @@ inline void AsyncRunner<Item>::working()
 
         bool ret = process_(id, std::move(item));
 
+        {
+            std::unique_lock clear_lock(queue_mutex_);
+            running_id_ = 0;
+            running_item_ = {};
+        }
+
         status_lock.lock();
         status_map_[id] = ret ? Status::succeeded : Status::failed;
         status_lock.unlock();
@@ -219,6 +234,36 @@ inline AsyncRunner<Item>::Status AsyncRunner<Item>::status(Id id) const
     return iter->second;
 }
 
+template <typename Item>
+inline std::optional<Item> AsyncRunner<Item>::get(Id id) const
+{
+    std::unique_lock queue_lock(queue_mutex_);
+
+    if (running_id_ == id) {
+        return running_item_;
+    }
+
+    for (const auto& [item_id, item] : queue_) {
+        if (item_id == id) {
+            return item;
+        }
+    }
+
+    return std::nullopt;
+}
+
+template <typename Item>
+inline std::optional<Item> AsyncRunner<Item>::get_running() const
+{
+    std::unique_lock queue_lock(queue_mutex_);
+
+    if (running_id_ != 0) {
+        return running_item_;
+    }
+
+    return std::nullopt;
+}
+
 template <typename Item>
 inline void AsyncRunner<Item>::clear()
 {
diff --git a/source/MaaFramework/Tasker/Tasker.cpp b/source/MaaFramework/Tasker/Tasker.cpp
@@ -167,11 +167,13 @@ MaaTaskId Tasker::post_stop()
 
     need_to_stop_ = true;
 
-    if (task_runner_ && task_runner_->running()) {
-        task_runner_->clear();
-    }
-    if (running_task_) {
-        running_task_->post_stop();
+    if (task_runner_) {
+        if (task_runner_->running()) {
+            task_runner_->clear();
+        }
+        if (auto running_task = task_runner_->get_running()) {
+            (*running_task)->post_stop();
+        }
     }
     if (resource_) {
         resource_->post_stop();
@@ -323,11 +325,8 @@ bool Tasker::run_task(RunnerId runner_id, TaskPtr task_ptr)
         return false;
     }
 
-    running_task_ = task_ptr;
-    OnScopeLeave([&] { running_task_ = nullptr; });
-
     if (need_to_stop_) {
-        running_task_->post_stop();
+        task_ptr->post_stop();
     }
 
     MaaTaskId task_id = task_ptr->task_id();
@@ -360,11 +359,27 @@ bool Tasker::run_task(RunnerId runner_id, TaskPtr task_ptr)
     }
     notifier_.notify(this, ret ? MaaMsg_Tasker_Task_Succeeded : MaaMsg_Tasker_Task_Failed, cb_detail);
 
-    running_task_ = nullptr;
-
     return ret;
 }
 
+bool Tasker::override_pipeline(MaaTaskId task_id, const json::value& pipeline_override)
+{
+    LogFunc << VAR(task_id) << VAR(pipeline_override);
+
+    RunnerId runner_id = task_id_to_runner_id(task_id);
+    if (runner_id == MaaInvalidId) {
+        return false;
+    }
+
+    auto task_ptr_opt = task_runner_->get(runner_id);
+    if (!task_ptr_opt) {
+        LogError << "task not found or already completed" << VAR(task_id) << VAR(runner_id);
+        return false;
+    }
+
+    return (*task_ptr_opt)->override_pipeline(pipeline_override);
+}
+
 bool Tasker::check_stop()
 {
     if (!need_to_stop_) {
diff --git a/source/MaaFramework/Tasker/Tasker.h b/source/MaaFramework/Tasker/Tasker.h
@@ -36,6 +36,8 @@ class Tasker : public MaaTasker
         post_action(const std::string& action_type, const json::value& action_param, const cv::Rect& box, const std::string& reco_detail)
             override;
 
+    virtual bool override_pipeline(MaaTaskId task_id, const json::value& pipeline_override) override;
+
     virtual MaaStatus status(MaaTaskId task_id) const override;
     virtual MaaStatus wait(MaaTaskId task_id) const override;
 
@@ -90,8 +92,6 @@ class Tasker : public MaaTasker
     std::map<MaaTaskId, RunnerId> task_id_mapping_;
     mutable std::shared_mutex task_id_mapping_mutex_;
 
-    TaskPtr running_task_ = nullptr;
-
     RuntimeCache runtime_cache_;
 };
 
diff --git a/source/binding/NodeJS/src/apis/tasker.cpp b/source/binding/NodeJS/src/apis/tasker.cpp
@@ -259,6 +259,14 @@ void TaskerImpl::clear_cache()
     MaaTaskerClearCache(tasker);
 }
 
+void TaskerImpl::override_pipeline(MaaTaskId task_id, maajs::ValueType pipeline)
+{
+    auto str = maajs::JsonStringify(env, pipeline);
+    if (!MaaTaskerOverridePipeline(tasker, task_id, str.c_str())) {
+        throw maajs::MaaError { "Tasker override_pipeline failed" };
+    }
+}
+
 std::optional<maajs::ValueType> TaskerImpl::recognition_detail(MaaRecoId id)
 {
     StringBuffer node_name;
@@ -448,6 +456,7 @@ void TaskerImpl::init_proto(maajs::ObjectType proto, maajs::FunctionType)
     MAA_BIND_GETTER_SETTER(proto, "resource", TaskerImpl::get_resource, TaskerImpl::set_resource);
     MAA_BIND_GETTER_SETTER(proto, "controller", TaskerImpl::get_controller, TaskerImpl::set_controller);
     MAA_BIND_FUNC(proto, "clear_cache", TaskerImpl::clear_cache);
+    MAA_BIND_FUNC(proto, "override_pipeline", TaskerImpl::override_pipeline);
     MAA_BIND_FUNC(proto, "recognition_detail", TaskerImpl::recognition_detail);
     MAA_BIND_FUNC(proto, "action_detail", TaskerImpl::action_detail);
     MAA_BIND_FUNC(proto, "node_detail", TaskerImpl::node_detail);
diff --git a/source/binding/NodeJS/src/apis/tasker.d.ts b/source/binding/NodeJS/src/apis/tasker.d.ts
@@ -159,6 +159,7 @@ declare global {
             set controller(res: Controller | null)
             get controller(): Controller | null
             clear_cache(): void
+            override_pipeline(task_id: TaskId, pipeline: Record<string, unknown> | Record<string, unknown>[]): void
             recognition_detail(id: RecoId): RecoDetail | null
             action_detail(id: ActId): ActionDetail | null
             node_detail(id: NodeId): NodeDetail | null
diff --git a/source/binding/NodeJS/src/apis/tasker.h b/source/binding/NodeJS/src/apis/tasker.h
@@ -61,6 +61,7 @@ struct TaskerImpl : public maajs::NativeClassBase
     void set_controller(std::optional<maajs::NativeObject<ControllerImpl>> ctrl);
     std::optional<maajs::ValueType> get_controller();
     void clear_cache();
+    void override_pipeline(MaaTaskId task_id, maajs::ValueType pipeline);
     std::optional<maajs::ValueType> recognition_detail(MaaRecoId id);
     std::optional<maajs::ValueType> action_detail(MaaActId id);
     std::optional<maajs::ValueType> node_detail(MaaNodeId id);
diff --git a/source/binding/Python/maa/context.py b/source/binding/Python/maa/context.py
@@ -11,6 +11,7 @@
 from .define import *
 from .library import Library
 from .tasker import Tasker
+from .job import TaskJob
 from .pipeline import (
     JPipelineData,
     JPipelineParser,
@@ -20,7 +21,6 @@
     JRecognitionParam,
     JActionParam,
 )
-from .job import JobWithResult
 
 
 class Context:
@@ -331,11 +331,11 @@ def tasker(self) -> Tasker:
         """
         return self._tasker
 
-    def get_task_job(self) -> JobWithResult:
+    def get_task_job(self) -> TaskJob:
         """获取对应任务号的任务作业 / Get task job for corresponding task id
 
         Returns:
-            JobWithResult: 任务作业对象 / Task job object
+            TaskJob: 任务作业对象 / Task job object
 
         Raises:
             ValueError: 如果任务 id 为 None
diff --git a/source/binding/Python/maa/job.py b/source/binding/Python/maa/job.py
diff --git a/source/binding/Python/maa/tasker.py b/source/binding/Python/maa/tasker.py
diff --git a/source/include/Common/MaaTypes.h b/source/include/Common/MaaTypes.h
diff --git a/source/include/MaaAgent/Message.hpp b/source/include/MaaAgent/Message.hpp
diff --git a/source/modules/MaaFramework.cppm b/source/modules/MaaFramework.cppm
diff --git a/test/agent/agent_child_test.py b/test/agent/agent_child_test.py
diff --git a/test/python/binding_test.py b/test/python/binding_test.py
