feat(system): add serial arm and AGV action queue

cmvr-es/common/types/arm/arm_types.h

@@ -2,6 +2,7 @@
 #define CMVR_ES_ARM_TYPES_H
 
 #include <cstdint>
+#include <functional>
 #include <string>
 #include <vector>
 
@@ -162,6 +163,11 @@ struct MotionOptions {
     double jerk{5.0};
     std::vector<double> joint_velocity_limits;
     bool asynchronous{false};
+    // Framework-independent cancellation check used by queued synchronous
+    // motion. Cancellation after device acceptance retains a typed motion
+    // barrier; the owner must call stopMotion() to confirm physical idle.
+    // Drivers must not retain this callback after moveJ/moveL returns.
+    std::function<bool()> cancellation_requested;
 };
 
 struct ServoOptions {

cmvr-es/devices/agv/abstract_agv.h

@@ -15,6 +15,12 @@
 
 namespace cmvr::device {
 
+enum class AgvActionKind {
+    NavigateToPose,
+    NavigateToStation,
+    FollowPath,
+};
+
 /**
  * @brief AGV/移动底盘设备抽象基类。
  *
@@ -28,6 +34,16 @@ public:
 
     DeviceKind kind() const noexcept override { return DeviceKind::AGV; }
 
+    /**
+     * @brief Whether this backend provides terminal-state and stopped-motion
+     *        confirmation plus bounded cancellation suitable for synchronous
+     *        Action execution.
+     */
+    virtual bool supportsSynchronousAction(AgvActionKind) const noexcept
+    {
+        return false;
+    }
+
     /**
      * @brief 获取 AGV 运行状态快照。
      */
@@ -167,6 +183,19 @@ public:
         return setVelocity(AgvVelocity{});
     }
 
+    /**
+     * @brief 确认 AGV 已进入可安全释放控制权的停止状态。
+     *
+     * 该接口只在导航任务已终止且底盘速度经过连续采样确认为零后返回
+     * 成功；仅收到取消、停止或零速度命令的应答不构成成功。
+     */
+    virtual AgvResult confirmMotionStopped()
+    {
+        return AgvResult::failure(
+            AgvErrorCode::UnsupportedCommand,
+            "confirmMotionStopped not implemented");
+    }
+
     /**
      * @brief 查询 AGV 可用地图名称列表。
      */

cmvr-es/devices/agv/seer_robokit/include/seer_robokit_agv.h

@@ -27,6 +27,17 @@ public:
 
     std::string typeName() const override { return "SeerRobokitAgv"; }
 
+    bool supportsSynchronousAction(AgvActionKind kind) const noexcept override
+    {
+        switch (kind) {
+            case AgvActionKind::NavigateToPose:
+            case AgvActionKind::NavigateToStation:
+            case AgvActionKind::FollowPath:
+                return true;
+        }
+        return false;
+    }
+
     bool init() override;
     bool start() override;
     bool stop() override;
@@ -57,6 +68,7 @@ public:
     AgvResult cancelNavigation() override;
 
     AgvResult setVelocity(const AgvVelocity& velocity) override;
+    AgvResult confirmMotionStopped() override;
 
     AgvResult listMaps(std::vector<std::string>& maps) const override;
     AgvResult listStations(std::vector<AgvStation>& stations) const override;
@@ -166,7 +178,8 @@ private:
     AgvResult waitForCanceledTaskToStop_(
         const TrackedNavigationContext& context,
         const AgvMotionOptions& options,
-        const std::string& reason);
+        const std::string& reason,
+        bool require_global_stopped = false);
     AgvResult failAndCancelTrackedNavigation_(
         const TrackedNavigationContext& context,
         const AgvMotionOptions& options,

cmvr-es/devices/agv/seer_robokit/src/seer_robokit_navigation_wait.cpp

@@ -473,10 +473,71 @@ AgvResult SeerRobokitAgv::cancelTrackedNavigation_(
     return result;
 }
 
+AgvResult SeerRobokitAgv::confirmMotionStopped()
+{
+    TrackedNavigationContext tracked_navigation;
+    if (currentTrackedNavigation_(tracked_navigation)) {
+        return waitForCanceledTaskToStop_(
+            tracked_navigation,
+            AgvMotionOptions{},
+            "explicit motion-stop confirmation",
+            true);
+    }
+
+    const auto confirmation_window =
+        kNavigationCancelConfirmationTimeout;
+    const auto deadline =
+        std::chrono::steady_clock::now() + confirmation_window;
+    int stopped_samples = 0;
+    std::string last_detail = "no 1101 status received";
+    while (std::chrono::steady_clock::now() < deadline) {
+        NavigationSnapshot snapshot;
+        const auto snapshot_result = queryNavigationSnapshot_(snapshot);
+        if (!snapshot_result.ok()) {
+            return AgvResult::failure(
+                snapshot_result.code,
+                "SEER Robokit stopped state could not be confirmed with 1101: "
+                    + snapshot_result.message);
+        }
+
+        last_detail = snapshot.detail;
+        const bool no_active_navigation =
+            snapshot.task_status_present &&
+            globalTaskStateIsKnownTerminal(snapshot.task_status);
+        if (no_active_navigation && navigationStopped(snapshot)) {
+            ++stopped_samples;
+            if (stopped_samples >= kRequiredCompletedStopSamples) {
+                return {
+                    AgvErrorCode::OK,
+                    "no active navigation and two zero-velocity samples were "
+                    "confirmed with 1101"};
+            }
+        } else {
+            stopped_samples = 0;
+        }
+
+        const auto now = std::chrono::steady_clock::now();
+        if (now < deadline) {
+            std::this_thread::sleep_for(std::min(
+                kNavigationCancelPollInterval,
+                std::chrono::duration_cast<std::chrono::milliseconds>(
+                    deadline - now)));
+        }
+    }
+
+    return AgvResult::failure(
+        AgvErrorCode::Timeout,
+        "SEER Robokit did not confirm an inactive navigation task and two "
+        "zero-velocity samples within "
+            + std::to_string(confirmation_window.count())
+            + " ms; last_status=" + last_detail);
+}
+
 AgvResult SeerRobokitAgv::waitForCanceledTaskToStop_(
     const TrackedNavigationContext& context,
     const AgvMotionOptions& options,
-    const std::string& reason)
+    const std::string& reason,
+    const bool require_global_stopped)
 {
     if (context.task_ids.empty()) {
         return AgvResult::failure(
@@ -525,7 +586,8 @@ AgvResult SeerRobokitAgv::waitForCanceledTaskToStop_(
         const bool another_local_navigation_started =
             currentTrackedNavigation_(active_context)
             && active_context.token != context.token;
-        if (all_exact_tasks_terminal && another_local_navigation_started) {
+        if (all_exact_tasks_terminal && another_local_navigation_started &&
+            !require_global_stopped) {
             // A later navigation is allowed to move after this exact task has
             // reached a terminal state. Its velocity must not keep the older
             // waiter alive or make it cancel the newer task.
@@ -567,14 +629,17 @@ AgvResult SeerRobokitAgv::waitForCanceledTaskToStop_(
                    context.target_ids.end(),
                    snapshot.target_id)
                 != context.target_ids.end();
-        const bool global_terminal = snapshot.task_status == 0
+        const bool global_terminal = snapshot.task_status_present &&
+            (snapshot.task_status == 0
              || (exactTaskStateIsKnownTerminal(snapshot.task_status)
                  && snapshot.task_type == expected_global_type
-                && global_target_matches);
+                 && global_target_matches));
+        const bool task_termination_confirmed = require_global_stopped
+            ? (!any_exact_task_active && global_terminal)
+            : (all_exact_tasks_terminal
+               || (!any_exact_task_active && global_terminal));
 
-        if ((all_exact_tasks_terminal
-             || (!any_exact_task_active && global_terminal))
-            && navigationStopped(snapshot)) {
+        if (task_termination_confirmed && navigationStopped(snapshot)) {
             ++stopped_samples;
             if (stopped_samples >= kRequiredCompletedStopSamples) {
                 for (const auto& task_id : context.task_ids) {

cmvr-es/devices/agv/seer_robokit/tests/seer_robokit_control_authority_test.cpp

@@ -605,6 +605,144 @@ protected:
     std::unique_ptr<SeerRobokitAgv> agv_;
 };
 
+TEST_F(SeerRobokitControlAuthorityTest, DeclaresSynchronousActionSupport)
+{
+    EXPECT_TRUE(agv_->supportsSynchronousAction(
+        AgvActionKind::NavigateToPose));
+    EXPECT_TRUE(agv_->supportsSynchronousAction(
+        AgvActionKind::NavigateToStation));
+    EXPECT_TRUE(agv_->supportsSynchronousAction(
+        AgvActionKind::FollowPath));
+}
+
+TEST_F(
+    SeerRobokitControlAuthorityTest,
+    ConfirmMotionStoppedWithoutTrackedTaskUsesGlobalStatusOnly)
+{
+    controller_.clearRecords();
+
+    const auto result = agv_->confirmMotionStopped();
+
+    ASSERT_TRUE(result.ok()) << result.message;
+    const auto records = controller_.records();
+    EXPECT_EQ(
+        std::count_if(
+            records.begin(),
+            records.end(),
+            [](const CommandRecord& record) {
+                return record.command == kRobotStatusAll2;
+            }),
+        2);
+    EXPECT_EQ(
+        std::count_if(
+            records.begin(),
+            records.end(),
+            [](const CommandRecord& record) {
+                return record.command == kRobotStatusTaskPackage;
+            }),
+        0);
+}
+
+TEST_F(
+    SeerRobokitControlAuthorityTest,
+    ConfirmMotionStoppedForTrackedStationChecksExactTaskAndGlobalVelocity)
+{
+    const auto navigate_result = agv_->navigateToStation(
+        "station-confirm-stop",
+        asynchronousMotionOptions());
+    ASSERT_TRUE(navigate_result.ok()) << navigate_result.message;
+    controller_.setResponsePayload(
+        kRobotStatusTaskPackage,
+        R"({"ret_code":0,"task_status_package":{"task_status_list":[{"task_id":"${TASK_ID}","status":4,"type":2}]}})");
+    controller_.setResponsePayload(
+        kRobotStatusAll2,
+        R"({"ret_code":0,"task_status":4,"task_type":2,"target_id":"station-confirm-stop","blocked":false,"vx":0.0,"vy":0.0,"w":0.0,"fatals":[],"errors":[]})");
+    controller_.clearRecords();
+
+    const auto result = agv_->confirmMotionStopped();
+
+    ASSERT_TRUE(result.ok()) << result.message;
+    const auto records = controller_.records();
+    EXPECT_GE(
+        std::count_if(
+            records.begin(),
+            records.end(),
+            [](const CommandRecord& record) {
+                return record.command == kRobotStatusTaskPackage;
+            }),
+        2);
+    EXPECT_GE(
+        std::count_if(
+            records.begin(),
+            records.end(),
+            [](const CommandRecord& record) {
+                return record.command == kRobotStatusAll2;
+            }),
+        2);
+}
+
+TEST_F(
+    SeerRobokitControlAuthorityTest,
+    ConfirmMotionStoppedTimesOutWhenTrackedTaskIsTerminalButVelocityIsNonzero)
+{
+    const auto navigate_result = agv_->navigateToStation(
+        "station-still-moving",
+        asynchronousMotionOptions());
+    ASSERT_TRUE(navigate_result.ok()) << navigate_result.message;
+    controller_.setResponsePayload(
+        kRobotStatusTaskPackage,
+        R"({"ret_code":0,"task_status_package":{"task_status_list":[{"task_id":"${TASK_ID}","status":4,"type":2}]}})");
+    controller_.setResponsePayload(
+        kRobotStatusAll2,
+        R"({"ret_code":0,"task_status":4,"task_type":2,"target_id":"station-still-moving","blocked":false,"vx":0.1,"vy":0.0,"w":0.0,"fatals":[],"errors":[]})");
+    controller_.clearRecords();
+
+    const auto result = agv_->confirmMotionStopped();
+
+    EXPECT_EQ(result.code, AgvErrorCode::Timeout);
+    EXPECT_NE(
+        result.message.find("stopped velocity were not confirmed"),
+        std::string::npos);
+}
+
+TEST_F(
+    SeerRobokitControlAuthorityTest,
+    ConfirmMotionStoppedWaitsForGlobalTaskToBecomeTerminal)
+{
+    const auto navigate_result = agv_->navigateToStation(
+        "station-global-active",
+        asynchronousMotionOptions());
+    ASSERT_TRUE(navigate_result.ok()) << navigate_result.message;
+    controller_.setResponsePayload(
+        kRobotStatusTaskPackage,
+        R"({"ret_code":0,"task_status_package":{"task_status_list":[{"task_id":"${TASK_ID}","status":4,"type":2}]}})");
+    controller_.setResponsePayload(
+        kRobotStatusAll2,
+        R"({"ret_code":0,"task_status":2,"task_type":2,"target_id":"station-global-active","blocked":false,"vx":0.0,"vy":0.0,"w":0.0,"fatals":[],"errors":[]})");
+    controller_.clearRecords();
+    std::atomic<bool> finished{false};
+    AgvResult result;
+
+    std::thread confirmation([this, &finished, &result]() {
+        result = agv_->confirmMotionStopped();
+        finished.store(true, std::memory_order_release);
+    });
+    const bool zero_samples_observed = waitForCommandCount(
+        kRobotStatusAll2, 2, 1000);
+    std::this_thread::sleep_for(std::chrono::milliseconds(50));
+    const bool returned_while_global_active =
+        finished.load(std::memory_order_acquire);
+
+    controller_.setResponsePayload(
+        kRobotStatusAll2,
+        R"({"ret_code":0,"task_status":4,"task_type":2,"target_id":"station-global-active","blocked":false,"vx":0.0,"vy":0.0,"w":0.0,"fatals":[],"errors":[]})");
+    confirmation.join();
+
+    EXPECT_TRUE(zero_samples_observed);
+    EXPECT_FALSE(returned_while_global_active);
+    EXPECT_TRUE(result.ok()) << result.message;
+}
+
 TEST_F(SeerRobokitControlAuthorityTest, EveryImplementedMutatingOperationAcquiresAuthorityFirst)
 {
     controller_.clearRecords();

cmvr-es/devices/arm/aubo_arm/aubo_arm.cpp

@@ -23,6 +23,21 @@
 namespace cmvr::device {
 namespace {
 
+bool cancellationRequested(
+    const std::function<bool()>& cancellation_requested) noexcept
+{
+    if (!cancellation_requested) {
+        return false;
+    }
+    try {
+        return cancellation_requested();
+    } catch (...) {
+        // A broken cancellation source must never permit a queued motion to be
+        // submitted after its ownership can no longer be established.
+        return true;
+    }
+}
+
 class MotionOwnerGuard final {
 public:
     MotionOwnerGuard(
@@ -1818,11 +1833,12 @@ Result AuboArm::moveJ(const JointPositionCommand& target, const MotionOptions& o
         auto motion_control = robot_interface->getMotionControl();
         motion_control->setSpeedFraction(speed_scaling_);
         motion_owner.requireExplicitSettlement();
-        if (!validateSafetyPermit(safety_monitor, safety_permit)) {
+        if (cancellationRequested(options.cancellation_requested) ||
+            !validateSafetyPermit(safety_monitor, safety_permit)) {
             motion_owner.settle();
             return Result::failure(
                 ArmErrorCode::CommandRejected,
-                "[AuboArm] moveJ cancelled by hardware safety before submission");
+                "[AuboArm] moveJ cancelled before submission");
         }
         const int ret = motion_control->moveJoint(
             target.position,
@@ -1843,18 +1859,39 @@ Result AuboArm::moveJ(const JointPositionCommand& target, const MotionOptions& o
              motion_state,
              safety_monitor,
              safety_permit,
+             cancellation_requested = options.cancellation_requested,
              token = motion.token]() {
                 return waitArrival(
                     robot_interface,
                     [motion_state,
                      safety_monitor,
                      safety_permit,
+                     cancellation_requested,
                      token]() {
-                        return motion_state->cancelled(token) ||
+                        return cancellationRequested(
+                                   cancellation_requested) ||
+                               motion_state->cancelled(token) ||
                                !validateSafetyPermit(
                                    safety_monitor, safety_permit);
                     });
             });
+        if (cancellationRequested(options.cancellation_requested)) {
+            if (ret == arcs::common_interface::AUBO_OK) {
+                if (outcome ==
+                    aubo_internal::MotionCommandOutcome::CompletedAfterMotion) {
+                    motion_owner.clearOnFinish();
+                } else {
+                    // The request was accepted but its completion is no longer
+                    // owned by this caller. Preserve the typed motion state so
+                    // the cancellation owner can issue stopJoint/stopLine.
+                    motion_owner.retainKind();
+                }
+            }
+            motion_owner.settle();
+            return Result::failure(
+                ArmErrorCode::CommandRejected,
+                "[AuboArm] moveJ cancelled by its caller");
+        }
         if (!validateSafetyPermit(safety_monitor, safety_permit)) {
             motion_owner.settle();
             return Result::failure(
@@ -2019,11 +2056,12 @@ Result AuboArm::moveL(const CartesianPose& target, const MotionOptions& options,
         robot_interface->getRobotConfig()->setTcpOffset(tcp_offset);
         std::vector<double> pose{target.x, target.y, target.z, target.rx, target.ry, target.rz};
         motion_owner.requireExplicitSettlement();
-        if (!validateSafetyPermit(safety_monitor, safety_permit)) {
+        if (cancellationRequested(options.cancellation_requested) ||
+            !validateSafetyPermit(safety_monitor, safety_permit)) {
             motion_owner.settle();
             return Result::failure(
                 ArmErrorCode::CommandRejected,
-                "[AuboArm] moveL cancelled by hardware safety before submission");
+                "[AuboArm] moveL cancelled before submission");
         }
         const int ret = motion_control->moveLine(
             pose,
@@ -2044,18 +2082,38 @@ Result AuboArm::moveL(const CartesianPose& target, const MotionOptions& options,
              motion_state,
              safety_monitor,
              safety_permit,
+             cancellation_requested = options.cancellation_requested,
              token = motion.token]() {
                 return waitArrival(
                     robot_interface,
                     [motion_state,
                      safety_monitor,
                      safety_permit,
+                     cancellation_requested,
                      token]() {
-                        return motion_state->cancelled(token) ||
+                        return cancellationRequested(
+                                   cancellation_requested) ||
+                               motion_state->cancelled(token) ||
                                !validateSafetyPermit(
                                    safety_monitor, safety_permit);
                     });
             });
+        if (cancellationRequested(options.cancellation_requested)) {
+            if (ret == arcs::common_interface::AUBO_OK) {
+                if (outcome ==
+                    aubo_internal::MotionCommandOutcome::CompletedAfterMotion) {
+                    motion_owner.clearOnFinish();
+                } else {
+                    // Keep the accepted linear kind until a typed Stop confirms
+                    // that the controller is idle.
+                    motion_owner.retainKind();
+                }
+            }
+            motion_owner.settle();
+            return Result::failure(
+                ArmErrorCode::CommandRejected,
+                "[AuboArm] moveL cancelled by its caller");
+        }
         if (!validateSafetyPermit(safety_monitor, safety_permit)) {
             motion_owner.settle();
             return Result::failure(

cmvr-es/devices/arm/aubo_arm/aubo_arm.h

@@ -33,6 +33,7 @@ public:
     RobotMode getRobotMode() const override;
     SafetyMode getSafetyMode() const override;
     ControlMode getControlMode() const override;
+    bool supportsActionQueueMotion() const noexcept override { return true; }
 
     Result torqueOn() override;
     Result torqueOff() override;

cmvr-es/devices/arm/huayan_arm/huayan_arm.cpp

@@ -34,6 +34,21 @@ constexpr auto kControllerStopTimeout = std::chrono::milliseconds(3000);
 constexpr auto kOwnerExitTimeout = std::chrono::milliseconds(3000);
 constexpr auto kCompletionCorrelationGrace = std::chrono::milliseconds(250);
 
+bool cancellationRequested(
+    const std::function<bool()>& cancellation_requested) noexcept
+{
+    if (!cancellation_requested) {
+        return false;
+    }
+    try {
+        return cancellation_requested();
+    } catch (...) {
+        // Cancellation sources are part of the motion-admission safety gate.
+        // Treat an exception as cancellation instead of admitting new motion.
+        return true;
+    }
+}
+
 double radToDeg(const double value)
 {
     return value * 180.0 / kPi;
@@ -662,11 +677,12 @@ Result HuayanRobot::moveJ(
 
     if (targetReached_(&target.position, nullptr) &&
         controllerIdleStable_(runtime, std::chrono::milliseconds(300))) {
-        if (!runtime->safety.validate(permit)) {
+        if (cancellationRequested(options.cancellation_requested) ||
+            !runtime->safety.validate(permit)) {
             runtime->motion.finish(start.token, MotionFinishMode::Clear);
             return Result::failure(
                 ArmErrorCode::CommandRejected,
-                "[HuayanRobot] moveJ cancelled by a safety transition");
+                "[HuayanRobot] moveJ cancelled by its caller or a safety transition");
         }
         runtime->motion.finish(start.token, MotionFinishMode::Clear);
         return Result::success();
@@ -687,18 +703,20 @@ Result HuayanRobot::moveJ(
     const double blend = metersToMm(options.blend_radius);
     const auto command_id = nextCommandId_();
 
-    if (!runtime->safety.validate(permit)) {
+    if (cancellationRequested(options.cancellation_requested) ||
+        !runtime->safety.validate(permit)) {
         runtime->motion.finish(start.token, MotionFinishMode::Clear);
         return Result::failure(
             ArmErrorCode::CommandRejected,
-            "[HuayanRobot] moveJ cancelled before submission by a safety transition");
+            "[HuayanRobot] moveJ cancelled before submission");
     }
     int ret = 0;
     {
         std::lock_guard<std::recursive_mutex> submission_lock(
             runtime->submission_mutex);
         std::lock_guard<std::recursive_mutex> sdk_lock(sdk_mutex_);
-        if (runtime->motion.cancelled(start.token) ||
+        if (cancellationRequested(options.cancellation_requested) ||
+            runtime->motion.cancelled(start.token) ||
             !runtime->safety.validate(permit)) {
             runtime->motion.finish(start.token, MotionFinishMode::Clear);
             return Result::failure(
@@ -719,7 +737,8 @@ Result HuayanRobot::moveJ(
     admission_lock.unlock();
     return waitMotionDone_(
         "moveJ", runtime, start.token, permit, command_id,
-        &target.position, nullptr, 60000);
+        &target.position, nullptr, 60000,
+        options.cancellation_requested);
 }
 
 Result HuayanRobot::speedJ(
@@ -837,11 +856,12 @@ Result HuayanRobot::moveL(
 
     if (targetReached_(nullptr, &target) &&
         controllerIdleStable_(runtime, std::chrono::milliseconds(300))) {
-        if (!runtime->safety.validate(permit)) {
+        if (cancellationRequested(options.cancellation_requested) ||
+            !runtime->safety.validate(permit)) {
             runtime->motion.finish(start.token, MotionFinishMode::Clear);
             return Result::failure(
                 ArmErrorCode::CommandRejected,
-                "[HuayanRobot] moveL cancelled by a safety transition");
+                "[HuayanRobot] moveL cancelled by its caller or a safety transition");
         }
         runtime->motion.finish(start.token, MotionFinishMode::Clear);
         return Result::success();
@@ -868,18 +888,20 @@ Result HuayanRobot::moveL(
     const double blend = metersToMm(options.blend_radius);
     const auto command_id = nextCommandId_();
 
-    if (!runtime->safety.validate(permit)) {
+    if (cancellationRequested(options.cancellation_requested) ||
+        !runtime->safety.validate(permit)) {
         runtime->motion.finish(start.token, MotionFinishMode::Clear);
         return Result::failure(
             ArmErrorCode::CommandRejected,
-            "[HuayanRobot] moveL cancelled before submission by a safety transition");
+            "[HuayanRobot] moveL cancelled before submission");
     }
     int ret = 0;
     {
         std::lock_guard<std::recursive_mutex> submission_lock(
             runtime->submission_mutex);
         std::lock_guard<std::recursive_mutex> sdk_lock(sdk_mutex_);
-        if (runtime->motion.cancelled(start.token) ||
+        if (cancellationRequested(options.cancellation_requested) ||
+            runtime->motion.cancelled(start.token) ||
             !runtime->safety.validate(permit)) {
             runtime->motion.finish(start.token, MotionFinishMode::Clear);
             return Result::failure(
@@ -900,7 +922,8 @@ Result HuayanRobot::moveL(
     admission_lock.unlock();
     return waitMotionDone_(
         "moveL", runtime, start.token, permit, command_id,
-        nullptr, &target, 60000);
+        nullptr, &target, 60000,
+        options.cancellation_requested);
 }
 
 Result HuayanRobot::speedL(
@@ -2279,7 +2302,8 @@ Result HuayanRobot::waitMotionDone_(
     const std::string& command_id,
     const std::vector<double>* joint_target,
     const CartesianPose* tcp_target,
-    const int timeout_ms) const
+    const int timeout_ms,
+    const std::function<bool()>& cancellation_requested) const
 {
     const auto started_at = std::chrono::steady_clock::now();
     bool saw_motion = false;
@@ -2287,13 +2311,20 @@ Result HuayanRobot::waitMotionDone_(
     int stable_completion_samples = 0;
 
     while (runtime->monitor_running.load()) {
-        if (runtime->motion.cancelled(motion_token) ||
+        const bool caller_cancelled =
+            cancellationRequested(cancellation_requested);
+        if (caller_cancelled ||
+            runtime->motion.cancelled(motion_token) ||
             !runtime->safety.validate(safety_permit)) {
-            runtime->motion.finish(motion_token, MotionFinishMode::Clear);
+            runtime->motion.finish(
+                motion_token,
+                caller_cancelled
+                    ? MotionFinishMode::Retain
+                    : MotionFinishMode::Clear);
             return Result::failure(
                 ArmErrorCode::CommandRejected,
                 "[HuayanRobot] " + context +
-                    " cancelled by Stop or a safety transition");
+                    " cancelled by its caller, Stop, or a safety transition");
         }
 
         bool done = false;

cmvr-es/devices/arm/huayan_arm/huayan_arm.h

@@ -11,6 +11,7 @@
 #include <atomic>
 #include <chrono>
 #include <cstdint>
+#include <functional>
 #include <memory>
 #include <mutex>
 #include <optional>
@@ -41,6 +42,7 @@ public:
     RobotMode getRobotMode() const override;
     SafetyMode getSafetyMode() const override;
     ControlMode getControlMode() const override;
+    bool supportsActionQueueMotion() const noexcept override { return true; }
 
     Result torqueOn() override;
     Result torqueOff() override;
@@ -149,7 +151,8 @@ private:
         const std::string& command_id,
         const std::vector<double>* joint_target,
         const CartesianPose* tcp_target,
-        int timeout_ms) const;
+        int timeout_ms,
+        const std::function<bool()>& cancellation_requested = {}) const;
     bool targetReached_(
         const std::vector<double>* joint_target,
         const CartesianPose* tcp_target) const;

cmvr-es/devices/arm/huayan_arm/tests/huayan_arm_sdk_test.cpp

@@ -592,11 +592,51 @@ int main()
     MotionOptions options;
     options.velocity = 0.4;
     options.acceleration = 0.8;
+    CHECK_TRUE(arm.supportsActionQueueMotion());
+
+    JointPositionCommand joint_a{{0.10, -0.05, 0.08, 0.0, 0.02, -0.03}};
+    MotionOptions cancelled_options = options;
+    cancelled_options.cancellation_requested = []() { return true; };
+    CHECK_TRUE(!arm.moveJ(joint_a, cancelled_options).ok());
+    CartesianPose cancelled_pose;
+    cancelled_pose.x = 0.20;
+    cancelled_pose.z = 0.30;
+    CHECK_TRUE(!arm.moveL(cancelled_pose, cancelled_options).ok());
+    {
+        std::lock_guard lock(g_sdk.mutex);
+        CHECK_TRUE(g_sdk.move_j_calls == 0);
+        CHECK_TRUE(g_sdk.move_l_calls == 0);
+    }
+
+    // Once a command has been accepted, caller cancellation returns promptly
+    // but retains the typed motion barrier until Stop confirms controller idle.
+    std::atomic<bool> cancel_during_wait{false};
+    MotionOptions cancellable_options = options;
+    cancellable_options.cancellation_requested = [&cancel_during_wait]() {
+        return cancel_during_wait.load();
+    };
+    JointPositionCommand cancelled_in_wait{
+        {0.05, -0.02, 0.04, 0.01, 0.0, -0.01}};
+    holdNextMotion();
+    auto cancelled_motion = std::async(std::launch::async, [&]() {
+        return arm.moveJ(cancelled_in_wait, cancellable_options);
+    });
+    CHECK_TRUE(waitUntil([&]() {
+        std::lock_guard lock(g_sdk.mutex);
+        return g_sdk.move_j_calls > 0;
+    }));
+    cancel_during_wait.store(true);
+    CHECK_TRUE(cancelled_motion.wait_for(1s) == std::future_status::ready);
+    if (cancelled_motion.wait_for(0ms) == std::future_status::ready) {
+        CHECK_TRUE(!cancelled_motion.get().ok());
+    }
+    CHECK_TRUE(arm.busy());
+    CHECK_TRUE(arm.stopMotion().ok());
+    CHECK_TRUE(!arm.busy());
 
     // The first IsMotionDone read intentionally reports the preceding idle
     // state. Completion must be correlated with the command/target. Once the
     // target is reached, an identical command is an idempotent no-op.
-    JointPositionCommand joint_a{{0.10, -0.05, 0.08, 0.0, 0.02, -0.03}};
     CHECK_TRUE(arm.moveJ(joint_a, options).ok());
     int move_j_after_first = 0;
     {

cmvr-es/devices/arm/robot_arm.h

@@ -28,6 +28,12 @@ public:
     virtual SafetyMode getSafetyMode() const = 0;
     virtual ControlMode getControlMode() const = 0;
 
+    // Queued actions require synchronous completion, cooperative cancellation
+    // at the final device-submission boundary, and a bounded typed Stop which
+    // returns success only after controller idle is confirmed. Backends must
+    // opt in only after all of these semantics have been validated.
+    virtual bool supportsActionQueueMotion() const noexcept { return false; }
+
     // ArmTeleop requires an explicitly reviewed group-servo implementation.
     // Existing and vendor arms remain unavailable until their implementations
     // override this capability after timing and partial-write validation.

cmvr-es/manager/control_authority/include/control_authority_manager.h

@@ -49,6 +49,20 @@ public:
         const std::string& resource_id,
         const std::string& owner_id,
         Duration ttl);
+
+    // Converts the expected normal lease into a safety barrier only while it
+    // is still the current lease. A stale token never preempts a successor or
+    // joins an existing safety barrier.
+    ControlAcquireResult preemptAcquireIfCurrent(
+        const ControlLeaseToken& expected_token,
+        const std::string& owner_id,
+        Duration ttl);
+
+    // Permanently blocks the resource only if the expected normal lease is
+    // still current. Quarantine does not allocate and can only be removed by
+    // an explicit revoke/clear.
+    bool quarantineIfCurrent(
+        const ControlLeaseToken& expected_token) noexcept;
     bool renew(const ControlLeaseToken& token, Duration ttl);
     bool validate(const ControlLeaseToken& token);
     void release(const ControlLeaseToken& token) noexcept;
@@ -69,9 +83,11 @@ private:
         std::uint64_t generation{0};
         std::chrono::steady_clock::time_point deadline;
         bool preemptible{true};
+        bool quarantined{false};
         std::unordered_map<std::uint64_t, std::string> safety_holders;
     };
 
+    static void quarantine_(Entry& entry) noexcept;
     bool expired_(const Entry& entry) const noexcept;
 
     std::mutex mutex_;

cmvr-es/manager/control_authority/src/control_authority_manager.cpp

@@ -1,5 +1,6 @@
 #include "manager/control_authority/include/control_authority_manager.h"
 
+#include <type_traits>
 #include <utility>
 
 namespace cmvr::control {
@@ -44,6 +45,7 @@ ControlAcquireResult ControlAuthorityManager::tryAcquire(
             token.generation,
             std::chrono::steady_clock::now() + ttl,
             true,
+            false,
             {}});
     return {true, std::move(token), {}};
 }
@@ -71,22 +73,112 @@ ControlAcquireResult ControlAuthorityManager::preemptAcquire(
                 token.generation, token.owner_id);
             return {true, std::move(token), {}};
         }
-        entries_.erase(existing);
     }
 
+    const bool has_existing = existing != entries_.end();
+    const bool replacing_normal =
+        has_existing && existing->second.preemptible;
+    try {
         ControlLeaseToken token;
         token.resource_id = resource_id;
         token.owner_id = owner_id;
         token.generation = ++next_generation_;
-    entries_.emplace(
-        resource_id,
-        Entry{
+        Entry replacement{
             owner_id,
             token.generation,
             std::chrono::steady_clock::time_point::max(),
             false,
-            {{token.generation, owner_id}}});
+            false,
+            {{token.generation, owner_id}}};
+
+        if (has_existing) {
+            static_assert(
+                std::is_nothrow_move_assignable_v<Entry>,
+                "safety barrier replacement must not throw");
+            existing->second = std::move(replacement);
+        } else {
+            entries_.emplace(resource_id, std::move(replacement));
+        }
         return {true, std::move(token), {}};
+    } catch (...) {
+        if (replacing_normal && existing->second.preemptible) {
+            quarantine_(existing->second);
+        }
+        throw;
+    }
+}
+
+ControlAcquireResult ControlAuthorityManager::preemptAcquireIfCurrent(
+    const ControlLeaseToken& expected_token,
+    const std::string& owner_id,
+    const Duration ttl)
+{
+    if (!expected_token.valid() || owner_id.empty() ||
+        ttl <= Duration::zero()) {
+        return {false, {}, "invalid conditional control barrier request"};
+    }
+
+    std::lock_guard lock(mutex_);
+    const auto existing = entries_.find(expected_token.resource_id);
+    if (existing == entries_.end() ||
+        expired_(existing->second) ||
+        !existing->second.preemptible ||
+        existing->second.owner_id != expected_token.owner_id ||
+        existing->second.generation != expected_token.generation) {
+        return {
+            false,
+            {},
+            "expected control lease is no longer current"};
+    }
+
+    try {
+        ControlLeaseToken token;
+        token.resource_id = expected_token.resource_id;
+        token.owner_id = owner_id;
+        token.generation = ++next_generation_;
+        Entry replacement{
+            owner_id,
+            token.generation,
+            std::chrono::steady_clock::time_point::max(),
+            false,
+            false,
+            {{token.generation, owner_id}}};
+
+        static_assert(
+            std::is_nothrow_move_assignable_v<Entry>,
+            "safety barrier replacement must not throw");
+        existing->second = std::move(replacement);
+        return {true, std::move(token), {}};
+    } catch (...) {
+        if (existing->second.preemptible) {
+            quarantine_(existing->second);
+        }
+        throw;
+    }
+}
+
+bool ControlAuthorityManager::quarantineIfCurrent(
+    const ControlLeaseToken& expected_token) noexcept
+{
+    if (!expected_token.valid()) {
+        return false;
+    }
+    try {
+        std::lock_guard lock(mutex_);
+        const auto existing =
+            entries_.find(expected_token.resource_id);
+        if (existing == entries_.end() ||
+            expired_(existing->second) ||
+            !existing->second.preemptible ||
+            existing->second.owner_id != expected_token.owner_id ||
+            existing->second.generation != expected_token.generation) {
+            return false;
+        }
+        quarantine_(existing->second);
+        return true;
+    } catch (...) {
+        return false;
+    }
 }
 
 bool ControlAuthorityManager::renew(
@@ -164,7 +256,8 @@ void ControlAuthorityManager::release(
                 return;
             }
             found->second.safety_holders.erase(holder);
-            if (found->second.safety_holders.empty()) {
+            if (found->second.safety_holders.empty() &&
+                !found->second.quarantined) {
                 entries_.erase(found);
             }
         } else if (found->second.owner_id == token.owner_id &&
@@ -215,7 +308,16 @@ void ControlAuthorityManager::clear() noexcept
 bool ControlAuthorityManager::expired_(
     const Entry& entry) const noexcept
 {
-    return std::chrono::steady_clock::now() >= entry.deadline;
+    return !entry.quarantined &&
+           std::chrono::steady_clock::now() >= entry.deadline;
+}
+
+void ControlAuthorityManager::quarantine_(Entry& entry) noexcept
+{
+    entry.deadline =
+        std::chrono::steady_clock::time_point::max();
+    entry.preemptible = false;
+    entry.quarantined = true;
 }
 
 } // namespace cmvr::control

cmvr-es/manager/control_authority/tests/control_authority_manager_test.cpp

@@ -93,6 +93,139 @@ TEST_F(ControlAuthorityManagerTest,
     EXPECT_FALSE(manager.isLeased("right_arm"));
 }
 
+TEST_F(ControlAuthorityManagerTest,
+       ConditionalSafetyBarrierPreemptsMatchingCurrentLease)
+{
+    auto& manager = ControlAuthorityManager::instance();
+    const auto control =
+        manager.tryAcquire("right_arm", "move-session", 100ms);
+    ASSERT_TRUE(control.acquired);
+
+    const auto barrier = manager.preemptAcquireIfCurrent(
+        control.token, "timed-out-action", 100ms);
+    ASSERT_TRUE(barrier.acquired) << barrier.detail;
+    EXPECT_FALSE(manager.validate(control.token));
+    EXPECT_TRUE(manager.validate(barrier.token));
+    manager.release(control.token);
+    EXPECT_TRUE(manager.validate(barrier.token));
+    EXPECT_FALSE(
+        manager.tryAcquire("right_arm", "new-move", 100ms)
+            .acquired);
+
+    manager.release(barrier.token);
+    EXPECT_FALSE(manager.isLeased("right_arm"));
+}
+
+TEST_F(ControlAuthorityManagerTest,
+       ConditionalSafetyBarrierDoesNotPreemptSuccessorForStaleToken)
+{
+    auto& manager = ControlAuthorityManager::instance();
+    const auto old =
+        manager.tryAcquire("right_arm", "move-session", 100ms);
+    ASSERT_TRUE(old.acquired);
+    const auto direct_stop = manager.preemptAcquire(
+        "right_arm", "direct-stop", 100ms);
+    ASSERT_TRUE(direct_stop.acquired) << direct_stop.detail;
+    manager.release(direct_stop.token);
+    const auto successor =
+        manager.tryAcquire("right_arm", "move-session", 100ms);
+    ASSERT_TRUE(successor.acquired);
+    ASSERT_NE(old.token.generation, successor.token.generation);
+
+    const auto barrier = manager.preemptAcquireIfCurrent(
+        old.token, "delayed-stop", 100ms);
+    EXPECT_FALSE(barrier.acquired);
+    EXPECT_FALSE(barrier.token.valid());
+    EXPECT_TRUE(manager.validate(successor.token));
+    EXPECT_FALSE(
+        manager.tryAcquire("right_arm", "competing-move", 100ms)
+            .acquired);
+
+    manager.release(successor.token);
+    EXPECT_FALSE(manager.isLeased("right_arm"));
+}
+
+TEST_F(ControlAuthorityManagerTest,
+       ConditionalSafetyBarrierDoesNotJoinExistingSafetyBarrier)
+{
+    auto& manager = ControlAuthorityManager::instance();
+    const auto control =
+        manager.tryAcquire("right_arm", "move-session", 100ms);
+    ASSERT_TRUE(control.acquired);
+    const auto existing_barrier = manager.preemptAcquire(
+        "right_arm", "direct-stop", 100ms);
+    ASSERT_TRUE(existing_barrier.acquired) << existing_barrier.detail;
+
+    const auto delayed_barrier = manager.preemptAcquireIfCurrent(
+        control.token, "delayed-action-stop", 100ms);
+    EXPECT_FALSE(delayed_barrier.acquired);
+    EXPECT_FALSE(delayed_barrier.token.valid());
+    EXPECT_TRUE(manager.validate(existing_barrier.token));
+
+    manager.release(existing_barrier.token);
+    EXPECT_FALSE(manager.isLeased("right_arm"));
+    EXPECT_TRUE(
+        manager.tryAcquire("right_arm", "new-move", 100ms)
+            .acquired);
+}
+
+TEST_F(ControlAuthorityManagerTest,
+       QuarantineSurvivesNormalAndTemporarySafetyTokenRelease)
+{
+    auto& manager = ControlAuthorityManager::instance();
+    const auto control =
+        manager.tryAcquire("right_arm", "move-session", 20ms);
+    ASSERT_TRUE(control.acquired);
+
+    ASSERT_TRUE(manager.quarantineIfCurrent(control.token));
+    EXPECT_FALSE(manager.validate(control.token));
+    EXPECT_TRUE(manager.isLeased("right_arm"));
+
+    manager.release(control.token);
+    std::this_thread::sleep_for(30ms);
+    EXPECT_TRUE(manager.isLeased("right_arm"));
+    EXPECT_FALSE(
+        manager.tryAcquire("right_arm", "new-move", 100ms)
+            .acquired);
+
+    const auto temporary_stop = manager.preemptAcquire(
+        "right_arm", "temporary-stop", 100ms);
+    ASSERT_TRUE(temporary_stop.acquired) << temporary_stop.detail;
+    EXPECT_TRUE(manager.validate(temporary_stop.token));
+    manager.release(temporary_stop.token);
+
+    EXPECT_TRUE(manager.isLeased("right_arm"));
+    EXPECT_FALSE(
+        manager.tryAcquire("right_arm", "new-move", 100ms)
+            .acquired);
+
+    manager.revoke("right_arm");
+    EXPECT_FALSE(manager.isLeased("right_arm"));
+}
+
+TEST_F(ControlAuthorityManagerTest,
+       QuarantineWithStaleTokenDoesNotAffectSuccessor)
+{
+    auto& manager = ControlAuthorityManager::instance();
+    const auto old =
+        manager.tryAcquire("right_arm", "move-session", 100ms);
+    ASSERT_TRUE(old.acquired);
+    manager.release(old.token);
+    const auto successor =
+        manager.tryAcquire("right_arm", "move-session", 100ms);
+    ASSERT_TRUE(successor.acquired);
+    ASSERT_NE(old.token.generation, successor.token.generation);
+
+    EXPECT_FALSE(manager.quarantineIfCurrent(old.token));
+    EXPECT_TRUE(manager.validate(successor.token));
+    EXPECT_FALSE(
+        manager.tryAcquire("right_arm", "competing-move", 100ms)
+            .acquired);
+
+    manager.release(successor.token);
+    EXPECT_FALSE(manager.isLeased("right_arm"));
+}
+
 TEST_F(ControlAuthorityManagerTest, ExpiryAndRenewUseMonotonicLocalTime)
 {
     auto& manager = ControlAuthorityManager::instance();

cmvr-es/service/CMakeLists.txt

@@ -1,5 +1,6 @@
 
 add_library(service
+        action/src/action_queue_executor.cpp
         grpc/src/grpc_camera_service.cpp
         grpc/src/grpc_system_service.cpp
         grpc/src/grpc_speaker_service.cpp

cmvr-es/service/README.md

@@ -8,6 +8,7 @@
 
 | 目录 | 职责 |
 | --- | --- |
+| `action/` | SystemService ActionQueue 的校验、幂等账本和边缘端 FIFO 执行器 |
 | `grpc/` | 入站设备控制、状态查询和兼容流式接口 |
 | `quic_edge/` | 边缘端主动连接平台的 QUIC client、控制状态机和媒体 packetizer |
 | `quic_edge/tests/` | 已登记到 CTest 的 QUIC 协议测试 |
@@ -39,6 +40,56 @@ grpcurl -plaintext \
   cmvr.api.SystemService/GetDeviceList
 ```
 
+## SystemService ActionQueue
+
+`SystemService/ExecuteActionQueue` 接收一个完整的有限动作序列，在边缘端排队并
+逐步串行执行，所有步骤结束后返回最终结果。平台只需要提交一次请求，因此连续机械臂
+动作不会再受到每个单独 gRPC 往返和 Wi-Fi 抖动的影响。
+
+当前 v1 仅允许以下 `ActionStep.command`：
+
+- 机械臂同步 `MoveJ`、`MoveL`；
+- AGV 同步 `navigateToPose`、`navigateToStation`、`followPath`；
+- 边缘端本地 `delay`。
+
+机械臂和 AGV 请求复用各自已有的类型化 Request，目标设备仍由每一步的
+`header.device_id` 指定。所有运动步骤必须设置 `asynchronous=false`；`MoveL` v1 仅接受
+Base frame；AGV 后端还必须明确支持同步导航终态确认。`speedJ`、`speedL`、`servoJ`、
+AGV `translate`、速度控制、查询和流式 RPC 都不属于 ActionQueue v1。
+
+ActionQueue 遵循以下执行语义：
+
+- 平台先调用 `GetSystemInfo` 读取 `action_service_instance_id`，并在每次提交和重试中填入
+  `expected_service_instance_id`。ActionQueue 账本随服务实例重建；若断线期间边缘服务重启，
+  旧实例 ID 会被拒绝，平台必须先对账，不能用新 ID 自动重放不确定的动作；
+- `action_id` 是必填的全局唯一幂等键；同一服务实例内，相同内容的已受理请求不会重复下发
+  设备命令，相同 ID 但内容不同的请求必须拒绝。服务端缓存最近 4096 个完整结果，更早的
+  已执行 ID 由精确 retired-ID 账本 fail-closed 拒绝、不会重跑；单实例最多记录
+  262144 个已受理 ID，达到容量后仅拒绝新 ID，已有 ID 仍可查询；
+- 入队前校验全部步骤、设备、参数和同步能力，校验失败时不会执行任何步骤；
+- v1 每个请求最多 256 步、序列化大小最多 512 KiB、排队或执行中的 Action 最多 64 个、
+  同时提交或等待结果的 RPC 最多 256 个；Action 与单步超时上限均为 24 小时，
+  `total_timeout_ms=0` 使用 30 分钟默认值，AGV 路径最多 4096 段；
+- `total_timeout_ms` 包含排队与执行时间，单步 `timeout_ms=0` 时继承 Action 剩余时间
+  或服务端默认值；所有超时值均由服务端施加上限；
+- 任一步失败、取消或超时后立即停止序列，不再执行后续步骤；`completed_steps` 表示此前
+  成功完成的步骤数，`failed_step_index` 仅在存在对应失败步骤时出现；
+- Action 一旦受理，不因平台连接中断而自动取消；断线只结束该 RPC waiter，边缘动作继续。
+  平台可用相同 `action_id` 重试并取得仍在缓存中的同一次执行结果；
+- `StopAll`、机械臂 `stopMotion`、AGV `cancelNavigation` 和软件急停不进入 FIFO，必须
+  作为高优先级安全/抢占路径执行。它们仍不具备功能安全等级。
+- 机械臂步骤超时会立即走 typed `stopMotion` 并等待停车确认；若无法确认停车，设备控制权
+  保持隔离，不会继续后续步骤或接受新的普通控制命令；需先按设备安全流程确认状态，再
+  重启边缘服务恢复控制。
+- AGV 的取消 ACK、零速度 ACK 均不等于停稳；ActionQueue 和安全停止 RPC 只有在导航任务
+  终态且底盘连续零速度采样确认后才释放控制权，否则同样保留隔离。
+
+已知的执行完成、业务失败、取消、超时和预校验拒绝由 `ActionResultCode` 与
+`CommandHeader.Feedback` 表达。`ActionDeduplicationStatus` 结构化区分新受理、合并等待、
+缓存结果、已淘汰结果、账本耗尽、ID 冲突和服务实例不匹配；平台不得通过解析错误字符串
+判断动作是否执行过。
+`ACTION_RESULT_CODE_UNSPECIFIED` 不得作为服务端最终结果。
+
 ## MotorService
 
 `MotorService` 将 gRPC 电机命令适配到已经由 `DeviceManager` 创建的

cmvr-es/service/action/include/action_queue_executor.h

@@ -0,0 +1,65 @@
+#ifndef CMVR_ES_ACTION_QUEUE_EXECUTOR_H
+#define CMVR_ES_ACTION_QUEUE_EXECUTOR_H
+
+#include <chrono>
+#include <cstddef>
+#include <functional>
+#include <memory>
+#include <string>
+
+#include "cmvr/api/system_command.pb.h"
+
+namespace cmvr::device {
+class DeviceManager;
+}
+
+namespace cmvr::service {
+
+// Owns the process-local FIFO used by SystemService ActionQueue requests.
+// The executor intentionally has no grpc::ServerContext dependency: once a
+// request is accepted, loss of the platform connection must not cancel device
+// motion on the edge.
+class ActionQueueExecutor final {
+public:
+    static constexpr std::size_t kDefaultMaxAcceptedActionIds =
+        256U * 1024U;
+
+    enum class WaitResult {
+        Terminal,
+        CanceledBeforeAdmission,
+        CanceledAfterAdmission,
+    };
+
+    explicit ActionQueueExecutor(
+        device::DeviceManager& device_manager,
+        std::size_t max_accepted_action_ids =
+            kDefaultMaxAcceptedActionIds);
+    ~ActionQueueExecutor();
+
+    ActionQueueExecutor(const ActionQueueExecutor&) = delete;
+    ActionQueueExecutor& operator=(const ActionQueueExecutor&) = delete;
+
+    // Validates, idempotently enqueues, and waits for the terminal result.
+    // Protocol and execution outcomes are represented in Feedback.
+    WaitResult submitAndWait(
+        const api::ActionQueueCommand_Request& request,
+        api::ActionQueueCommand_Feedback& feedback,
+        const std::function<bool()>& waiter_canceled = {});
+
+    // StopAll uses this fail-closed transition. It rejects future submissions,
+    // cancels pending actions, and requests a typed stop for the active action.
+    // Returns true when every active Action device reported a confirmed stop.
+    // False means at least one resource remains fail-closed quarantined.
+    bool cancelAllAndDisable();
+
+    bool waitForIdle(std::chrono::milliseconds timeout);
+    const std::string& instanceId() const noexcept;
+
+private:
+    struct Impl;
+    std::unique_ptr<Impl> impl_;
+};
+
+} // namespace cmvr::service
+
+#endif // CMVR_ES_ACTION_QUEUE_EXECUTOR_H

cmvr-es/service/grpc/include/grpc_system_service.h

@@ -5,23 +5,33 @@
 #ifndef GRPC_SYSTEM_SERVICE_H
 #define GRPC_SYSTEM_SERVICE_H
 
+#include <memory>
+
 #include "cmvr/api/system_service.grpc.pb.h"
 #include "common/base/grpc_utils.h"
 #include "manager/device_manager/include/device_manager.h"
 
 namespace cmvr::service
 {
+    class ActionQueueExecutor;
+
     class gRPCSystemServiceImpl: public api::SystemService::Service {
     public:
         gRPCSystemServiceImpl();
-        ~gRPCSystemServiceImpl() override = default;
+        ~gRPCSystemServiceImpl() override;
+        // Called by GrpcServerTask before grpc::Server::Shutdown so accepted
+        // ActionQueue handlers can reach a terminal result and do not hold the
+        // synchronous server shutdown open indefinitely.
+        void prepareForShutdown();
         grpc::Status GetSystemInfo(grpc::ServerContext* context, const api::GetSystemInfoCommand_Request* request, api::GetSystemInfoCommand_Feedback* response) override;
         grpc::Status GetSystemStatus(grpc::ServerContext* context, const api::GetSystemStatusCommand_Request* request, api::GetSystemStatusCommand_Feedback* response) override;
         grpc::Status GetDeviceList(grpc::ServerContext* context, const api::GetDeviceListCommand_Request* request, api::GetDeviceListCommand_Feedback* response) override;
         grpc::Status UpdateParams(grpc::ServerContext* context, const cmvr::api::UpdateParamsCommand_Request* request, cmvr::api::UpdateParamsCommand_Feedback* response) override;
         grpc::Status StopAll(grpc::ServerContext* context, const cmvr::api::StopAllCommand_Request* request, cmvr::api::StopAllCommand_Feedback* response) override;
+        grpc::Status ExecuteActionQueue(grpc::ServerContext* context, const cmvr::api::ActionQueueCommand_Request* request, cmvr::api::ActionQueueCommand_Feedback* response) override;
     private:
         device::DeviceManager& dmgr_;
+        std::unique_ptr<ActionQueueExecutor> action_queue_;
     };
 }
 

cmvr-es/service/grpc/src/grpc_agv_service.cpp

@@ -1,12 +1,18 @@
 #include "service/grpc/include/grpc_agv_service.h"
 
+#include <atomic>
+#include <chrono>
 #include <cstdint>
 #include <exception>
 #include <string>
+#include <utility>
 #include <vector>
 
 #include <google/protobuf/util/time_util.h>
 
+#include "common/base/logging/logger.h"
+#include "manager/control_authority/include/control_authority_manager.h"
+
 using google::protobuf::util::TimeUtil;
 
 namespace cmvr::service {
@@ -75,6 +81,114 @@ grpc::Status setDeviceNotFound(api::CommandHeader_Feedback* response, const std:
     return grpc::Status(grpc::StatusCode::NOT_FOUND, message);
 }
 
+grpc::Status setControlLeaseConflict(
+    api::CommandHeader_Feedback* response,
+    const std::string& device_id,
+    const std::string& detail)
+{
+    const std::string message =
+        "AGV control is leased by another active control operation: " +
+        device_id;
+    if (!detail.empty()) {
+        CMVR_LOG(WARNING) << "[gRPCAgvServiceImpl] control lease conflict, id="
+                          << device_id << ", detail=" << detail;
+    }
+    fillFeedback(response, false, message);
+    return grpc::Status(
+        grpc::StatusCode::FAILED_PRECONDITION, message);
+}
+
+template <typename Response>
+grpc::Status setControlLeaseConflict(
+    Response* response,
+    const std::string& device_id,
+    const std::string& detail)
+{
+    return setControlLeaseConflict(
+        response->mutable_header(), device_id, detail);
+}
+
+class ScopedUnaryAgvControlLease final {
+public:
+    ScopedUnaryAgvControlLease(
+        const std::string& device_id,
+        const char* operation,
+        const bool preemptive = false)
+        : manager_(control::ControlAuthorityManager::instance())
+    {
+        static std::atomic<std::uint64_t> sequence{0};
+        const std::string owner =
+            std::string(preemptive
+                            ? "grpc-agv-safety:"
+                            : "grpc-agv-unary:") +
+            operation + ":" +
+            std::to_string(
+                sequence.fetch_add(
+                    1U, std::memory_order_relaxed) +
+                1U);
+        const auto ttl = std::chrono::duration_cast<
+            control::ControlAuthorityManager::Duration>(
+                std::chrono::hours(24));
+        auto acquired = preemptive
+            ? manager_.preemptAcquire(device_id, owner, ttl)
+            : manager_.tryAcquire(device_id, owner, ttl);
+        acquired_ = acquired.acquired;
+        token_ = std::move(acquired.token);
+        detail_ = std::move(acquired.detail);
+    }
+
+    ~ScopedUnaryAgvControlLease()
+    {
+        if (release_on_destroy_) {
+            manager_.release(token_);
+        }
+    }
+
+    bool acquired() const noexcept { return acquired_; }
+    const std::string& detail() const noexcept { return detail_; }
+
+    // Unknown physical outcomes stay fail-closed until an explicit device
+    // safety procedure or process restart clears the retained holder.
+    void quarantine() noexcept { release_on_destroy_ = false; }
+
+private:
+    control::ControlAuthorityManager& manager_;
+    control::ControlLeaseToken token_;
+    std::string detail_;
+    bool acquired_{false};
+    bool release_on_destroy_{true};
+};
+
+template <typename Response, typename Operation>
+grpc::Status executeConfirmedAgvStop(
+    Response* response,
+    const std::shared_ptr<device::AbstractAGV>& agv,
+    ScopedUnaryAgvControlLease& control_barrier,
+    const char* operation_name,
+    Operation&& operation)
+{
+    try {
+        const auto command_result = operation();
+        const auto stopped = agv->confirmMotionStopped();
+        if (!stopped.ok()) {
+            control_barrier.quarantine();
+            std::string message = std::string(operation_name) +
+                " did not reach a confirmed stopped state: " +
+                stopped.message;
+            if (!command_result.ok()) {
+                message += "; command_result=" + command_result.message;
+            }
+            return setResponseResult(
+                response,
+                device::AgvResult::failure(stopped.code, message));
+        }
+        return setResponseResult(response, command_result);
+    } catch (...) {
+        control_barrier.quarantine();
+        throw;
+    }
+}
+
 template <typename Response>
 grpc::Status setNavigationRequestCanceled(Response* response)
 {
@@ -409,11 +523,20 @@ grpc::Status gRPCAgvServiceImpl::emergencyStop(grpc::ServerContext*,
                                                api::CommandHeader_Feedback* response)
 {
     try {
-        auto agv = dmgr_.getDevice<device::AbstractAGV>(request->device_id());
+        const std::string device_id = request->device_id();
+        auto agv = dmgr_.getDevice<device::AbstractAGV>(device_id);
         if (!agv) {
-            return setDeviceNotFound(response, request->device_id());
+            return setDeviceNotFound(response, device_id);
         }
-        return setResponseResult(response, agv->emergencyStop());
+        ScopedUnaryAgvControlLease control_barrier(
+            device_id, "emergencyStop", true);
+        if (!control_barrier.acquired()) {
+            return setControlLeaseConflict(
+                response, device_id, control_barrier.detail());
+        }
+        return executeConfirmedAgvStop(
+            response, agv, control_barrier, "emergencyStop",
+            [&agv]() { return agv->emergencyStop(); });
     } catch (const std::exception& e) {
         fillFeedback(response, false, e.what());
         return grpc::Status(grpc::StatusCode::INTERNAL, e.what());
@@ -425,9 +548,16 @@ grpc::Status gRPCAgvServiceImpl::clearFault(grpc::ServerContext*,
                                             api::CommandHeader_Feedback* response)
 {
     try {
-        auto agv = dmgr_.getDevice<device::AbstractAGV>(request->device_id());
+        const std::string device_id = request->device_id();
+        auto agv = dmgr_.getDevice<device::AbstractAGV>(device_id);
         if (!agv) {
-            return setDeviceNotFound(response, request->device_id());
+            return setDeviceNotFound(response, device_id);
+        }
+        ScopedUnaryAgvControlLease control_lease(
+            device_id, "clearFault");
+        if (!control_lease.acquired()) {
+            return setControlLeaseConflict(
+                response, device_id, control_lease.detail());
         }
         return setResponseResult(response, agv->clearFault());
     } catch (const std::exception& e) {
@@ -449,6 +579,12 @@ grpc::Status gRPCAgvServiceImpl::navigateToPose(grpc::ServerContext* context,
         if (!agv) {
             return setDeviceNotFound(response, device_id);
         }
+        ScopedUnaryAgvControlLease control_lease(
+            device_id, "navigateToPose");
+        if (!control_lease.acquired()) {
+            return setControlLeaseConflict(
+                response, device_id, control_lease.detail());
+        }
         return setResponseResult(response, agv->navigateToPose(
             toPose2d(request->pose()),
             toMotionOptions(request->options(), context),
@@ -472,6 +608,12 @@ grpc::Status gRPCAgvServiceImpl::navigateToStation(grpc::ServerContext* context,
         if (!agv) {
             return setDeviceNotFound(response, device_id);
         }
+        ScopedUnaryAgvControlLease control_lease(
+            device_id, "navigateToStation");
+        if (!control_lease.acquired()) {
+            return setControlLeaseConflict(
+                response, device_id, control_lease.detail());
+        }
         return setResponseResult(response, agv->navigateToStation(
             request->station_id(),
             toMotionOptions(request->options(), context),
@@ -495,6 +637,12 @@ grpc::Status gRPCAgvServiceImpl::followPath(grpc::ServerContext* context,
         if (!agv) {
             return setDeviceNotFound(response, device_id);
         }
+        ScopedUnaryAgvControlLease control_lease(
+            device_id, "followPath");
+        if (!control_lease.acquired()) {
+            return setControlLeaseConflict(
+                response, device_id, control_lease.detail());
+        }
         std::vector<device::AgvPathSegment> path;
         path.reserve(static_cast<std::size_t>(request->path_size()));
         for (const auto& segment : request->path()) {
@@ -527,6 +675,12 @@ grpc::Status gRPCAgvServiceImpl::translate(
         if (!agv) {
             return setDeviceNotFound(response, device_id);
         }
+        ScopedUnaryAgvControlLease control_lease(
+            device_id, "translate");
+        if (!control_lease.acquired()) {
+            return setControlLeaseConflict(
+                response, device_id, control_lease.detail());
+        }
         return setResponseResult(
             response,
             agv->translate(toTranslation(request->translation())));
@@ -545,9 +699,16 @@ grpc::Status gRPCAgvServiceImpl::pauseNavigation(grpc::ServerContext*,
                                                  api::CommandHeader_Feedback* response)
 {
     try {
-        auto agv = dmgr_.getDevice<device::AbstractAGV>(request->device_id());
+        const std::string device_id = request->device_id();
+        auto agv = dmgr_.getDevice<device::AbstractAGV>(device_id);
         if (!agv) {
-            return setDeviceNotFound(response, request->device_id());
+            return setDeviceNotFound(response, device_id);
+        }
+        ScopedUnaryAgvControlLease control_lease(
+            device_id, "pauseNavigation");
+        if (!control_lease.acquired()) {
+            return setControlLeaseConflict(
+                response, device_id, control_lease.detail());
         }
         return setResponseResult(response, agv->pauseNavigation());
     } catch (const std::exception& e) {
@@ -561,9 +722,16 @@ grpc::Status gRPCAgvServiceImpl::resumeNavigation(grpc::ServerContext*,
                                                   api::CommandHeader_Feedback* response)
 {
     try {
-        auto agv = dmgr_.getDevice<device::AbstractAGV>(request->device_id());
+        const std::string device_id = request->device_id();
+        auto agv = dmgr_.getDevice<device::AbstractAGV>(device_id);
         if (!agv) {
-            return setDeviceNotFound(response, request->device_id());
+            return setDeviceNotFound(response, device_id);
+        }
+        ScopedUnaryAgvControlLease control_lease(
+            device_id, "resumeNavigation");
+        if (!control_lease.acquired()) {
+            return setControlLeaseConflict(
+                response, device_id, control_lease.detail());
         }
         return setResponseResult(response, agv->resumeNavigation());
     } catch (const std::exception& e) {
@@ -577,11 +745,20 @@ grpc::Status gRPCAgvServiceImpl::cancelNavigation(grpc::ServerContext*,
                                                   api::CommandHeader_Feedback* response)
 {
     try {
-        auto agv = dmgr_.getDevice<device::AbstractAGV>(request->device_id());
+        const std::string device_id = request->device_id();
+        auto agv = dmgr_.getDevice<device::AbstractAGV>(device_id);
         if (!agv) {
-            return setDeviceNotFound(response, request->device_id());
+            return setDeviceNotFound(response, device_id);
         }
-        return setResponseResult(response, agv->cancelNavigation());
+        ScopedUnaryAgvControlLease control_barrier(
+            device_id, "cancelNavigation", true);
+        if (!control_barrier.acquired()) {
+            return setControlLeaseConflict(
+                response, device_id, control_barrier.detail());
+        }
+        return executeConfirmedAgvStop(
+            response, agv, control_barrier, "cancelNavigation",
+            [&agv]() { return agv->cancelNavigation(); });
     } catch (const std::exception& e) {
         fillFeedback(response, false, e.what());
         return grpc::Status(grpc::StatusCode::INTERNAL, e.what());
@@ -598,6 +775,12 @@ grpc::Status gRPCAgvServiceImpl::setVelocity(grpc::ServerContext*,
         if (!agv) {
             return setDeviceNotFound(response, device_id);
         }
+        ScopedUnaryAgvControlLease control_lease(
+            device_id, "setVelocity");
+        if (!control_lease.acquired()) {
+            return setControlLeaseConflict(
+                response, device_id, control_lease.detail());
+        }
         return setResponseResult(response, agv->setVelocity(toVelocity(request->velocity())));
     } catch (const std::exception& e) {
         fillFeedback(response->mutable_header(), false, e.what());
@@ -610,11 +793,20 @@ grpc::Status gRPCAgvServiceImpl::stopVelocityControl(grpc::ServerContext*,
                                                      api::CommandHeader_Feedback* response)
 {
     try {
-        auto agv = dmgr_.getDevice<device::AbstractAGV>(request->device_id());
+        const std::string device_id = request->device_id();
+        auto agv = dmgr_.getDevice<device::AbstractAGV>(device_id);
         if (!agv) {
-            return setDeviceNotFound(response, request->device_id());
+            return setDeviceNotFound(response, device_id);
         }
-        return setResponseResult(response, agv->stopVelocityControl());
+        ScopedUnaryAgvControlLease control_barrier(
+            device_id, "stopVelocityControl", true);
+        if (!control_barrier.acquired()) {
+            return setControlLeaseConflict(
+                response, device_id, control_barrier.detail());
+        }
+        return executeConfirmedAgvStop(
+            response, agv, control_barrier, "stopVelocityControl",
+            [&agv]() { return agv->stopVelocityControl(); });
     } catch (const std::exception& e) {
         fillFeedback(response, false, e.what());
         return grpc::Status(grpc::StatusCode::INTERNAL, e.what());
@@ -679,6 +871,12 @@ grpc::Status gRPCAgvServiceImpl::switchMap(grpc::ServerContext*,
         if (!agv) {
             return setDeviceNotFound(response, device_id);
         }
+        ScopedUnaryAgvControlLease control_lease(
+            device_id, "switchMap");
+        if (!control_lease.acquired()) {
+            return setControlLeaseConflict(
+                response, device_id, control_lease.detail());
+        }
         return setResponseResult(response, agv->switchMap(request->map_name()));
     } catch (const std::exception& e) {
         fillFeedback(response->mutable_header(), false, e.what());
@@ -696,6 +894,12 @@ grpc::Status gRPCAgvServiceImpl::uploadMap(grpc::ServerContext*,
         if (!agv) {
             return setDeviceNotFound(response, device_id);
         }
+        ScopedUnaryAgvControlLease control_lease(
+            device_id, "uploadMap");
+        if (!control_lease.acquired()) {
+            return setControlLeaseConflict(
+                response, device_id, control_lease.detail());
+        }
         return setResponseResult(response, agv->uploadMap(request->map_name(), request->content()));
     } catch (const std::exception& e) {
         fillFeedback(response->mutable_header(), false, e.what());
@@ -735,6 +939,12 @@ grpc::Status gRPCAgvServiceImpl::startMapping(grpc::ServerContext*,
         if (!agv) {
             return setDeviceNotFound(response, device_id);
         }
+        ScopedUnaryAgvControlLease control_lease(
+            device_id, "startMapping");
+        if (!control_lease.acquired()) {
+            return setControlLeaseConflict(
+                response, device_id, control_lease.detail());
+        }
         device::AgvMappingOptions options;
         options.dimension = toMapDimension(request->dimension());
         options.map_name = request->map_name();
@@ -825,9 +1035,16 @@ grpc::Status gRPCAgvServiceImpl::stopMapping(grpc::ServerContext*,
                                              api::CommandHeader_Feedback* response)
 {
     try {
-        auto agv = dmgr_.getDevice<device::AbstractAGV>(request->device_id());
+        const std::string device_id = request->device_id();
+        auto agv = dmgr_.getDevice<device::AbstractAGV>(device_id);
         if (!agv) {
-            return setDeviceNotFound(response, request->device_id());
+            return setDeviceNotFound(response, device_id);
+        }
+        ScopedUnaryAgvControlLease control_lease(
+            device_id, "stopMapping");
+        if (!control_lease.acquired()) {
+            return setControlLeaseConflict(
+                response, device_id, control_lease.detail());
         }
         return setResponseResult(response, agv->stopMapping());
     } catch (const std::exception& e) {

cmvr-es/service/grpc/src/grpc_arm_service.cpp

@@ -175,21 +175,26 @@ public:
         acquired_ = acquired.acquired;
         token_ = std::move(acquired.token);
         detail_ = std::move(acquired.detail);
+        release_on_destroy_ = !preemptive;
     }
 
     ~ScopedUnaryControlLease()
     {
+        if (release_on_destroy_) {
             manager_.release(token_);
         }
+    }
 
     bool acquired() const noexcept { return acquired_; }
     const std::string& detail() const noexcept { return detail_; }
+    void confirmSafeToRelease() noexcept { release_on_destroy_ = true; }
 
 private:
     control::ControlAuthorityManager& manager_;
     control::ControlLeaseToken token_;
     std::string detail_;
     bool acquired_{false};
+    bool release_on_destroy_{true};
 };
 
 } // namespace
@@ -216,6 +221,9 @@ grpc::Status gRPCArmServiceImpl::torqueOff(grpc::ServerContext*,
                 response, device_id, control_barrier.detail());
         }
         const auto result = arm->torqueOff();
+        if (result.ok()) {
+            control_barrier.confirmSafeToRelease();
+        }
         fillFeedback(response, result.ok(), result.ok() ? "" : result.message);
         if (result.ok()) {
             logRpcSuccess("torqueOff", device_id);
@@ -424,6 +432,9 @@ grpc::Status gRPCArmServiceImpl::stopMotion(grpc::ServerContext*,
                 response, device_id, control_barrier.detail());
         }
         const auto result = arm->stopMotion();
+        if (result.ok()) {
+            control_barrier.confirmSafeToRelease();
+        }
         fillFeedback(response, result.ok(), result.ok() ? "" : result.message);
         if (result.ok()) {
             logRpcSuccess("stopMotion", device_id);

cmvr-es/service/grpc/src/grpc_system_service.cpp

@@ -12,7 +12,10 @@
 #include <vector>
 
 #include "common/base/logging/logger.h"
+#include "devices/agv/abstract_agv.h"
+#include "devices/arm/robot_arm.h"
 #include "manager/control_authority/include/control_authority_manager.h"
+#include "service/action/include/action_queue_executor.h"
 
 using namespace cmvr::device;
 using namespace cmvr::service;
@@ -25,8 +28,10 @@ public:
     {
         auto& authority =
             cmvr::control::ControlAuthorityManager::instance();
-        for (const auto& token : tokens_) {
-            authority.release(token);
+        for (const auto& barrier : barriers_) {
+            if (barrier.release_on_destroy) {
+                authority.release(barrier.token);
+            }
         }
     }
 
@@ -49,18 +54,44 @@ public:
             detail = result.detail;
             return false;
         }
-        try {
-            tokens_.push_back(result.token);
-        } catch (...) {
-            cmvr::control::ControlAuthorityManager::instance().release(
-                result.token);
-            throw;
-        }
+        // StopAll barriers default to fail-closed. If retaining the token in
+        // this local vector throws, deliberately leave the manager-side safety
+        // holder installed: releasing it would reopen control after StopAll
+        // already preempted an in-flight command.
+        barriers_.push_back({result.token, false});
         return true;
     }
 
+    void quarantine(const std::string& device_id)
+    {
+        for (auto& barrier : barriers_) {
+            if (barrier.token.resource_id == device_id) {
+                barrier.release_on_destroy = false;
+            }
+        }
+    }
+
+    void quarantineAll()
+    {
+        for (auto& barrier : barriers_) {
+            barrier.release_on_destroy = false;
+        }
+    }
+
+    void confirmSafeToReleaseAll()
+    {
+        for (auto& barrier : barriers_) {
+            barrier.release_on_destroy = true;
+        }
+    }
+
 private:
-    std::vector<cmvr::control::ControlLeaseToken> tokens_;
+    struct Barrier {
+        cmvr::control::ControlLeaseToken token;
+        bool release_on_destroy{false};
+    };
+
+    std::vector<Barrier> barriers_;
 };
 
 std::uint64_t unixTimeMs() noexcept
@@ -154,7 +185,18 @@ cmvr::api::SystemDeviceHealth toApiDeviceHealth(
 
 } // namespace
 
-gRPCSystemServiceImpl::gRPCSystemServiceImpl(): dmgr_(DeviceManager::getInstance()) {}
+gRPCSystemServiceImpl::gRPCSystemServiceImpl()
+    : dmgr_(DeviceManager::getInstance()),
+      action_queue_(std::make_unique<ActionQueueExecutor>(dmgr_))
+{
+}
+
+gRPCSystemServiceImpl::~gRPCSystemServiceImpl() = default;
+
+void gRPCSystemServiceImpl::prepareForShutdown()
+{
+    (void)action_queue_->cancelAllAndDisable();
+}
 
 grpc::Status gRPCSystemServiceImpl::GetSystemInfo(grpc::ServerContext* context,
     const api::GetSystemInfoCommand_Request* request, api::GetSystemInfoCommand_Feedback* response)
@@ -162,6 +204,8 @@ grpc::Status gRPCSystemServiceImpl::GetSystemInfo(grpc::ServerContext* context,
     try {
         response->set_version(dmgr_.version());
         response->set_system_name(dmgr_.name());
+        response->set_action_service_instance_id(
+            action_queue_->instanceId());
         response->mutable_header()->set_success(true);
         setCurrentTimestamp(response->mutable_header()->mutable_timestamp());
         CMVR_LOG(DEBUG) << "[gRPCSystemServiceImpl] (GetSystemInfo): success, name="
@@ -287,27 +331,132 @@ grpc::Status gRPCSystemServiceImpl::UpdateParams(grpc::ServerContext* context, c
 grpc::Status gRPCSystemServiceImpl::StopAll(grpc::ServerContext* context,
     const cmvr::api::StopAllCommand_Request* request, cmvr::api::StopAllCommand_Feedback* response)
 {
+    (void)context;
+    (void)request;
     try {
         const auto snapshot = dmgr_.snapshot();
         ScopedControlBarrierSet control_barriers;
         for (const auto& device : snapshot.devices) {
-            if (device.kind == cmvr::device::DeviceKind::Arm) {
+            if (device.kind == cmvr::device::DeviceKind::Arm ||
+                device.kind == cmvr::device::DeviceKind::AGV) {
                 std::string detail;
                 if (!control_barriers.acquire(device.id, detail)) {
                     CMVR_LOG(WARNING)
                         << "[gRPCSystemServiceImpl] (StopAll): failed to "
-                           "acquire arm safety barrier, id="
+                           "acquire device safety barrier, id="
                         << device.id << ", detail=" << detail;
                     throw std::runtime_error(
-                        "StopAll could not acquire the RobotArm safety "
+                        "StopAll could not acquire the device safety "
                         "barrier: " + device.id);
                 }
             }
         }
+        const bool action_stop_confirmed =
+            action_queue_->cancelAllAndDisable();
+        if (!action_stop_confirmed) {
+            control_barriers.quarantineAll();
+        }
+        std::vector<std::string> unconfirmed_devices;
+        for (const auto& device : snapshot.devices) {
+            if (device.kind != cmvr::device::DeviceKind::Arm) {
+                continue;
+            }
+            auto arm = dmgr_.getDevice<cmvr::device::RobotArm>(device.id);
+            if (!arm) {
+                continue;
+            }
+            try {
+                const auto stopped = arm->stopMotion();
+                if (!stopped.ok()) {
+                    control_barriers.quarantine(device.id);
+                    unconfirmed_devices.push_back(
+                        device.id + ": " + stopped.message);
+                }
+            } catch (const std::exception& error) {
+                control_barriers.quarantine(device.id);
+                unconfirmed_devices.push_back(
+                    device.id + ": stop threw: " + error.what());
+            } catch (...) {
+                control_barriers.quarantine(device.id);
+                unconfirmed_devices.push_back(
+                    device.id + ": stop threw an unknown exception");
+            }
+        }
+
+        // AbstractAGV::stop() is a lifecycle hook and some backends do not
+        // map it to a motion stop. Use the typed non-E-stop controls here;
+        // StopAll must not be silently upgraded to emergencyStop semantics.
+        for (const auto& device : snapshot.devices) {
+            if (device.kind != cmvr::device::DeviceKind::AGV) {
+                continue;
+            }
+            auto agv = dmgr_.getDevice<cmvr::device::AbstractAGV>(
+                device.id);
+            if (!agv) {
+                continue;
+            }
+            try {
+                const auto cancel_result = agv->cancelNavigation();
+                if (!cancel_result.ok() &&
+                    cancel_result.code !=
+                        cmvr::device::AgvErrorCode::UnsupportedCommand) {
+                    CMVR_LOG(WARNING)
+                        << "[gRPCSystemServiceImpl] (StopAll): AGV navigation "
+                           "cancel failed, id="
+                        << device.id << ", detail=" << cancel_result.message;
+                }
+                const auto velocity_stop = agv->stopVelocityControl();
+                if (!velocity_stop.ok() &&
+                    velocity_stop.code !=
+                        cmvr::device::AgvErrorCode::UnsupportedCommand) {
+                    CMVR_LOG(WARNING)
+                        << "[gRPCSystemServiceImpl] (StopAll): AGV velocity "
+                           "stop failed, id="
+                        << device.id << ", detail="
+                        << velocity_stop.message;
+                }
+                const auto stopped = agv->confirmMotionStopped();
+                if (!stopped.ok()) {
+                    control_barriers.quarantine(device.id);
+                    unconfirmed_devices.push_back(
+                        device.id + ": " + stopped.message);
+                }
+            } catch (const std::exception& error) {
+                control_barriers.quarantine(device.id);
+                unconfirmed_devices.push_back(
+                    device.id + ": stop confirmation threw: " +
+                    error.what());
+            } catch (...) {
+                control_barriers.quarantine(device.id);
+                unconfirmed_devices.push_back(
+                    device.id +
+                    ": stop confirmation threw an unknown exception");
+            }
+        }
+        if (!action_queue_->waitForIdle(std::chrono::seconds(15))) {
+            control_barriers.quarantineAll();
+            throw std::runtime_error(
+                "StopAll timed out waiting for ActionQueue to become idle");
+        }
+        if (!action_stop_confirmed) {
+            throw std::runtime_error(
+                "StopAll could not confirm that every active ActionQueue "
+                "device stopped; affected control resources remain "
+                "quarantined");
+        }
+        if (!unconfirmed_devices.empty()) {
+            throw std::runtime_error(
+                "StopAll could not confirm that every device stopped; "
+                "affected control resources remain quarantined: " +
+                unconfirmed_devices.front());
+        }
+        // Do not close device transports while the Action worker may still be
+        // unwinding a synchronous driver call.
         dmgr_.stop();
         response->mutable_header()->set_success(true);
         setCurrentTimestamp(response->mutable_header()->mutable_timestamp());
         CMVR_LOG(DEBUG) << "[gRPCSystemServiceImpl] (StopAll): success";
+        control_barriers.confirmSafeToReleaseAll();
         return grpc::Status::OK;
     }
     catch (std::exception& e) {
@@ -317,3 +466,49 @@ grpc::Status gRPCSystemServiceImpl::StopAll(grpc::ServerContext* context,
         return grpc::Status::OK;
     }
 }
+
+grpc::Status gRPCSystemServiceImpl::ExecuteActionQueue(
+    grpc::ServerContext* context,
+    const cmvr::api::ActionQueueCommand_Request* request,
+    cmvr::api::ActionQueueCommand_Feedback* response)
+{
+    if (!request || !response) {
+        return grpc::Status(
+            grpc::StatusCode::INVALID_ARGUMENT,
+            "ActionQueue request and response are required");
+    }
+    try {
+        // The callback is consumed only on this synchronous handler stack. It
+        // is never retained by the worker-owned action Record, so returning the
+        // RPC cannot leave a dangling ServerContext reference.
+        const auto wait_result = action_queue_->submitAndWait(
+            *request,
+            *response,
+            [context]() {
+                return context && context->IsCancelled();
+            });
+        if (wait_result ==
+            ActionQueueExecutor::WaitResult::CanceledBeforeAdmission) {
+            return grpc::Status(
+                grpc::StatusCode::CANCELLED,
+                "ActionQueue RPC was canceled before admission");
+        }
+        if (wait_result ==
+            ActionQueueExecutor::WaitResult::CanceledAfterAdmission) {
+            return grpc::Status(
+                grpc::StatusCode::CANCELLED,
+                "ActionQueue RPC waiter was canceled after admission; the edge action continues and its result can be retrieved with the same action_id");
+        }
+        return grpc::Status::OK;
+    } catch (const std::exception& error) {
+        response->Clear();
+        response->set_action_id(request->action_id());
+        response->set_service_instance_id(action_queue_->instanceId());
+        response->set_result(api::ACTION_RESULT_CODE_FAILED);
+        response->mutable_header()->set_success(false);
+        response->mutable_header()->set_error_message(error.what());
+        setCurrentTimestamp(
+            response->mutable_header()->mutable_timestamp());
+        return grpc::Status::OK;
+    }
+}

cmvr-es/service/grpc/tests/grpc_agv_service_test.cpp

@@ -1,5 +1,6 @@
 #include "service/grpc/include/grpc_agv_service.h"
 
+#include <chrono>
 #include <memory>
 #include <string>
 #include <vector>
@@ -8,6 +9,7 @@
 #include <gtest/gtest.h>
 
 #include "cmvr/config/device_manager_config/device_manager_config.pb.h"
+#include "manager/control_authority/include/control_authority_manager.h"
 #include "manager/device_manager/include/device_manager.h"
 
 namespace cmvr::service {
@@ -27,6 +29,14 @@ public:
 
     std::string typeName() const override { return "FakeAgv"; }
 
+    device::AgvResult emergencyStop() override
+    {
+        ++emergency_stop_calls_;
+        emergency_stop_barrier_observed_ = normalLeaseIsBlocked_(
+            "test-probe:emergencyStop");
+        return device::AgvResult::success();
+    }
+
     device::AgvResult navigateToPose(
         const math::Pose2d& pose,
         const device::AgvMotionOptions& options,
@@ -79,6 +89,41 @@ public:
             kNativeErrorMessage);
     }
 
+    device::AgvResult cancelNavigation() override
+    {
+        ++cancel_navigation_calls_;
+        cancel_navigation_barrier_observed_ = normalLeaseIsBlocked_(
+            "test-probe:cancelNavigation");
+        return device::AgvResult::success();
+    }
+
+    device::AgvResult stopVelocityControl() override
+    {
+        ++stop_velocity_calls_;
+        stop_velocity_barrier_observed_ = normalLeaseIsBlocked_(
+            "test-probe:stopVelocityControl");
+        return device::AgvResult::success();
+    }
+
+    device::AgvResult confirmMotionStopped() override
+    {
+        ++confirm_stopped_calls_;
+        confirm_stopped_barrier_observed_ = normalLeaseIsBlocked_(
+            "test-probe:confirmMotionStopped");
+        return confirm_stopped_result_;
+    }
+
+    bool normalLeaseIsBlocked_(const std::string& owner)
+    {
+        auto& authority = control::ControlAuthorityManager::instance();
+        const auto probe = authority.tryAcquire(
+            id_, owner, std::chrono::hours(1));
+        if (probe.acquired) {
+            authority.release(probe.token);
+        }
+        return !probe.acquired;
+    }
+
     math::Pose2d pose_;
     device::AgvMotionOptions pose_options_;
     device::AgvResult pose_result_{device::AgvResult::success()};
@@ -93,6 +138,16 @@ public:
     bool station_cancellation_requested_during_call_{false};
     bool path_cancellation_bound_{false};
     bool path_cancellation_requested_during_call_{false};
+    int emergency_stop_calls_{0};
+    int cancel_navigation_calls_{0};
+    int stop_velocity_calls_{0};
+    int confirm_stopped_calls_{0};
+    bool emergency_stop_barrier_observed_{false};
+    bool cancel_navigation_barrier_observed_{false};
+    bool stop_velocity_barrier_observed_{false};
+    bool confirm_stopped_barrier_observed_{false};
+    device::AgvResult confirm_stopped_result_{
+        device::AgvResult::success()};
 };
 
 class LegacyFollowPathAgv final : public device::AbstractAGV {
@@ -113,6 +168,7 @@ class GrpcAgvServiceTest : public ::testing::Test {
 protected:
     void SetUp() override
     {
+        control::ControlAuthorityManager::instance().clear();
         config::DeviceManagerConfig config;
         auto& manager = device::DeviceManager::getInstance(config);
         agv_ = std::make_shared<FakeAgv>();
@@ -125,6 +181,7 @@ protected:
         service_.reset();
         agv_.reset();
         device::DeviceManager::destroyInstance();
+        control::ControlAuthorityManager::instance().clear();
     }
 
     std::shared_ptr<FakeAgv> agv_;
@@ -293,6 +350,171 @@ TEST_F(GrpcAgvServiceTest, ExplicitAsynchronousNavigationIsForwarded)
     EXPECT_FALSE(agv_->station_cancellation_requested_during_call_);
 }
 
+TEST_F(GrpcAgvServiceTest, ActionLeaseBlocksOrdinaryMutatingRpcs)
+{
+    auto& authority = control::ControlAuthorityManager::instance();
+    const auto action_lease = authority.tryAcquire(
+        "test-agv",
+        "action-sequence:test-action",
+        std::chrono::hours(1));
+    ASSERT_TRUE(action_lease.acquired) << action_lease.detail;
+
+    api::AgvNavigateToPoseCommand_Request navigation_request;
+    navigation_request.mutable_header()->set_device_id("test-agv");
+    api::AgvNavigateToPoseCommand_Feedback navigation_response;
+    grpc::ServerContext navigation_context;
+    const auto navigation_status = service_->navigateToPose(
+        &navigation_context,
+        &navigation_request,
+        &navigation_response);
+    EXPECT_EQ(
+        navigation_status.error_code(),
+        grpc::StatusCode::FAILED_PRECONDITION);
+    EXPECT_FALSE(navigation_response.header().success());
+
+    api::CommandHeader_Request clear_fault_request;
+    clear_fault_request.set_device_id("test-agv");
+    api::CommandHeader_Feedback clear_fault_response;
+    grpc::ServerContext clear_fault_context;
+    const auto clear_fault_status = service_->clearFault(
+        &clear_fault_context,
+        &clear_fault_request,
+        &clear_fault_response);
+    EXPECT_EQ(
+        clear_fault_status.error_code(),
+        grpc::StatusCode::FAILED_PRECONDITION);
+    EXPECT_FALSE(clear_fault_response.success());
+
+    api::AgvMapCommand_Request switch_map_request;
+    switch_map_request.mutable_header()->set_device_id("test-agv");
+    switch_map_request.set_map_name("map-1");
+    api::AgvMapCommand_Feedback switch_map_response;
+    grpc::ServerContext switch_map_context;
+    const auto switch_map_status = service_->switchMap(
+        &switch_map_context,
+        &switch_map_request,
+        &switch_map_response);
+    EXPECT_EQ(
+        switch_map_status.error_code(),
+        grpc::StatusCode::FAILED_PRECONDITION);
+    EXPECT_FALSE(switch_map_response.header().success());
+
+    authority.release(action_lease.token);
+}
+
+TEST_F(GrpcAgvServiceTest, QueriesBypassAndSafetyStopsPreemptActionLease)
+{
+    auto& authority = control::ControlAuthorityManager::instance();
+    const auto action_lease = authority.tryAcquire(
+        "test-agv",
+        "action-sequence:test-action",
+        std::chrono::hours(1));
+    ASSERT_TRUE(action_lease.acquired) << action_lease.detail;
+
+    api::AgvRuntimeStateCommand_Request state_request;
+    state_request.mutable_header()->set_device_id("test-agv");
+    api::AgvRuntimeStateCommand_Feedback state_response;
+    grpc::ServerContext state_context;
+    const auto state_status = service_->getRuntimeState(
+        &state_context,
+        &state_request,
+        &state_response);
+    EXPECT_TRUE(state_status.ok()) << state_status.error_message();
+    EXPECT_TRUE(state_response.header().success());
+    EXPECT_TRUE(authority.validate(action_lease.token));
+
+    api::CommandHeader_Request stop_request;
+    stop_request.set_device_id("test-agv");
+
+    api::CommandHeader_Feedback emergency_response;
+    grpc::ServerContext emergency_context;
+    const auto emergency_status = service_->emergencyStop(
+        &emergency_context,
+        &stop_request,
+        &emergency_response);
+    EXPECT_TRUE(emergency_status.ok()) << emergency_status.error_message();
+    EXPECT_FALSE(authority.validate(action_lease.token));
+    EXPECT_TRUE(agv_->emergency_stop_barrier_observed_);
+
+    const auto lease_after_emergency = authority.tryAcquire(
+        "test-agv",
+        "action-sequence:after-emergency",
+        std::chrono::hours(1));
+    ASSERT_TRUE(lease_after_emergency.acquired)
+        << lease_after_emergency.detail;
+
+    api::CommandHeader_Feedback cancel_response;
+    grpc::ServerContext cancel_context;
+    const auto cancel_status = service_->cancelNavigation(
+        &cancel_context,
+        &stop_request,
+        &cancel_response);
+    EXPECT_TRUE(cancel_status.ok()) << cancel_status.error_message();
+    EXPECT_FALSE(authority.validate(lease_after_emergency.token));
+    EXPECT_TRUE(agv_->cancel_navigation_barrier_observed_);
+
+    const auto lease_after_cancel = authority.tryAcquire(
+        "test-agv",
+        "action-sequence:after-cancel",
+        std::chrono::hours(1));
+    ASSERT_TRUE(lease_after_cancel.acquired)
+        << lease_after_cancel.detail;
+
+    api::CommandHeader_Feedback velocity_response;
+    grpc::ServerContext velocity_context;
+    const auto velocity_status = service_->stopVelocityControl(
+        &velocity_context,
+        &stop_request,
+        &velocity_response);
+    EXPECT_TRUE(velocity_status.ok()) << velocity_status.error_message();
+    EXPECT_FALSE(authority.validate(lease_after_cancel.token));
+    EXPECT_TRUE(agv_->stop_velocity_barrier_observed_);
+
+    EXPECT_EQ(agv_->emergency_stop_calls_, 1);
+    EXPECT_EQ(agv_->cancel_navigation_calls_, 1);
+    EXPECT_EQ(agv_->stop_velocity_calls_, 1);
+    EXPECT_EQ(agv_->confirm_stopped_calls_, 3);
+    EXPECT_TRUE(agv_->confirm_stopped_barrier_observed_);
+
+    const auto lease_after_stops = authority.tryAcquire(
+        "test-agv",
+        "action-sequence:after-stops",
+        std::chrono::hours(1));
+    ASSERT_TRUE(lease_after_stops.acquired)
+        << lease_after_stops.detail;
+    authority.release(lease_after_stops.token);
+}
+
+TEST_F(GrpcAgvServiceTest,
+       SafetyStopQuarantinesControlWhenStoppedStateIsUnconfirmed)
+{
+    agv_->confirm_stopped_result_ = device::AgvResult::failure(
+        device::AgvErrorCode::Timeout,
+        "two zero-velocity samples were not observed");
+
+    api::CommandHeader_Request request;
+    request.set_device_id("test-agv");
+    api::CommandHeader_Feedback response;
+    grpc::ServerContext context;
+    const auto status = service_->cancelNavigation(
+        &context, &request, &response);
+
+    EXPECT_EQ(status.error_code(), grpc::StatusCode::DEADLINE_EXCEEDED);
+    EXPECT_FALSE(response.success());
+    EXPECT_NE(
+        response.error_message().find("confirmed stopped state"),
+        std::string::npos);
+    EXPECT_TRUE(agv_->cancel_navigation_barrier_observed_);
+    EXPECT_TRUE(agv_->confirm_stopped_barrier_observed_);
+
+    const auto lease =
+        control::ControlAuthorityManager::instance().tryAcquire(
+            "test-agv",
+            "normal-control-after-unconfirmed-stop",
+            std::chrono::hours(1));
+    EXPECT_FALSE(lease.acquired);
+}
+
 TEST_F(GrpcAgvServiceTest, NavigateToStationForwardsPgvAdapterParams)
 {
     api::AgvNavigateToStationCommand_Request request;
@@ -384,5 +606,17 @@ TEST(AbstractAgvCompatibilityTest, FollowPathOptionsDelegateToLegacyOverride)
     EXPECT_EQ(legacy.path_[0].target_station, "station-2");
 }
 
+TEST(AbstractAgvCompatibilityTest, SynchronousActionSupportDefaultsToFalse)
+{
+    LegacyFollowPathAgv legacy;
+
+    EXPECT_FALSE(legacy.supportsSynchronousAction(
+        device::AgvActionKind::NavigateToPose));
+    EXPECT_FALSE(legacy.supportsSynchronousAction(
+        device::AgvActionKind::NavigateToStation));
+    EXPECT_FALSE(legacy.supportsSynchronousAction(
+        device::AgvActionKind::FollowPath));
+}
+
 } // namespace
 } // namespace cmvr::service

cmvr-es/service/grpc/tests/grpc_arm_service_test.cpp

@@ -6,6 +6,7 @@
 #include <memory>
 #include <mutex>
 #include <optional>
+#include <stdexcept>
 #include <string>
 #include <utility>
 #include <vector>
@@ -135,6 +136,16 @@ public:
                 lock,
                 [this]() { return release_blocking_stop_; });
         }
+        if (throw_next_stop_) {
+            throw_next_stop_ = false;
+            throw std::runtime_error("simulated stopMotion exception");
+        }
+        if (fail_next_stop_) {
+            fail_next_stop_ = false;
+            return device::Result::failure(
+                device::ArmErrorCode::CommandFailed,
+                "simulated stopMotion failure");
+        }
         return device::Result::success();
     }
 
@@ -178,6 +189,18 @@ public:
         release_blocking_stop_ = false;
     }
 
+    void failNextStopMotion()
+    {
+        std::lock_guard lock(motion_mutex_);
+        fail_next_stop_ = true;
+    }
+
+    void throwNextStopMotion()
+    {
+        std::lock_guard lock(motion_mutex_);
+        throw_next_stop_ = true;
+    }
+
     bool waitForBlockingStop(const std::chrono::milliseconds timeout)
     {
         std::unique_lock lock(motion_mutex_);
@@ -350,6 +373,8 @@ private:
     bool block_next_stop_{false};
     bool blocking_stop_started_{false};
     bool release_blocking_stop_{false};
+    bool fail_next_stop_{false};
+    bool throw_next_stop_{false};
     std::string blocking_motion_name_;
     int move_j_calls_{0};
     int move_l_calls_{0};
@@ -688,5 +713,51 @@ TEST_F(GrpcArmServiceTest,
     EXPECT_EQ(aubo_arm_->stopMotionCalls(), 1);
 }
 
+TEST_F(GrpcArmServiceTest, StopMotionFailureRetainsSafetyBarrier)
+{
+    auto& authority = control::ControlAuthorityManager::instance();
+    const auto action_lease = authority.tryAcquire(
+        "aubo_arm", "action-queue:test", std::chrono::hours(1));
+    ASSERT_TRUE(action_lease.acquired) << action_lease.detail;
+    aubo_arm_->failNextStopMotion();
+
+    api::CommandHeader_Feedback stop_response;
+    const auto stop_status = stopMotion("aubo_arm", stop_response);
+    const auto rejected_move = moveJ("aubo_arm");
+
+    EXPECT_EQ(stop_status.error_code(), grpc::StatusCode::INTERNAL);
+    EXPECT_FALSE(stop_response.success());
+    EXPECT_FALSE(authority.validate(action_lease.token));
+    EXPECT_TRUE(authority.isLeased("aubo_arm"));
+    EXPECT_EQ(
+        rejected_move.status.error_code(),
+        grpc::StatusCode::FAILED_PRECONDITION);
+    EXPECT_EQ(aubo_arm_->moveJCalls(), 0);
+    EXPECT_EQ(aubo_arm_->stopMotionCalls(), 1);
+}
+
+TEST_F(GrpcArmServiceTest, StopMotionExceptionRetainsSafetyBarrier)
+{
+    auto& authority = control::ControlAuthorityManager::instance();
+    const auto action_lease = authority.tryAcquire(
+        "aubo_arm", "action-queue:test", std::chrono::hours(1));
+    ASSERT_TRUE(action_lease.acquired) << action_lease.detail;
+    aubo_arm_->throwNextStopMotion();
+
+    api::CommandHeader_Feedback stop_response;
+    const auto stop_status = stopMotion("aubo_arm", stop_response);
+    const auto rejected_move = moveL("aubo_arm");
+
+    EXPECT_EQ(stop_status.error_code(), grpc::StatusCode::INTERNAL);
+    EXPECT_FALSE(stop_response.success());
+    EXPECT_FALSE(authority.validate(action_lease.token));
+    EXPECT_TRUE(authority.isLeased("aubo_arm"));
+    EXPECT_EQ(
+        rejected_move.status.error_code(),
+        grpc::StatusCode::FAILED_PRECONDITION);
+    EXPECT_EQ(aubo_arm_->moveLCalls(), 0);
+    EXPECT_EQ(aubo_arm_->stopMotionCalls(), 1);
+}
+
 } // namespace
 } // namespace cmvr::service

cmvr-es/task/grpc_server_task/src/grpc_server_task.cpp

@@ -223,6 +223,11 @@ void GrpcServerTask::stop()
 {
     {
         std::lock_guard lock(mutex_);
+        if (auto* system_service =
+                dynamic_cast<service::gRPCSystemServiceImpl*>(
+                    system_service_.get())) {
+            system_service->prepareForShutdown();
+        }
         if (server_) {
             server_->Shutdown();
         }

protos/cmvr/api/system_command.proto

@@ -1,5 +1,7 @@
 syntax = "proto3";
 
+import "cmvr/api/agv_command.proto";
+import "cmvr/api/arm_command.proto";
 import "cmvr/api/common.proto";
 
 package cmvr.api;
@@ -103,6 +105,10 @@ message GetSystemInfoCommand {
         string os = 5;
         string kernel_version = 6;
         string architecture = 7;
+
+        // Changes whenever the in-process ActionQueue idempotency ledger is
+        // recreated. Clients bind submissions and retries to this value.
+        string action_service_instance_id = 8;
     }
 }
 
@@ -141,3 +147,97 @@ message StopAllCommand {
         CommandHeader.Feedback header = 1;
     }
 }
+
+// Final outcome of one ActionQueue execution.
+enum ActionResultCode {
+    ACTION_RESULT_CODE_UNSPECIFIED = 0;
+    ACTION_RESULT_CODE_COMPLETED = 1;
+    ACTION_RESULT_CODE_FAILED = 2;
+    ACTION_RESULT_CODE_CANCELED = 3;
+    ACTION_RESULT_CODE_TIMED_OUT = 4;
+    ACTION_RESULT_CODE_REJECTED = 5;
+}
+
+// Describes whether this RPC admitted a new action or observed an existing
+// idempotency record. Clients must not infer this from an error string.
+enum ActionDeduplicationStatus {
+    ACTION_DEDUPLICATION_STATUS_UNSPECIFIED = 0;
+    ACTION_DEDUPLICATION_STATUS_ACCEPTED_NEW = 1;
+    ACTION_DEDUPLICATION_STATUS_JOINED_IN_FLIGHT = 2;
+    ACTION_DEDUPLICATION_STATUS_CACHED_RESULT = 3;
+    ACTION_DEDUPLICATION_STATUS_RESULT_EVICTED = 4;
+    ACTION_DEDUPLICATION_STATUS_LEDGER_EXHAUSTED = 5;
+    ACTION_DEDUPLICATION_STATUS_ACTION_ID_CONFLICT = 6;
+    ACTION_DEDUPLICATION_STATUS_SERVICE_INSTANCE_MISMATCH = 7;
+}
+
+// Edge-local delay between two device commands.
+message DelayAction {
+    // Delay duration in milliseconds. The server applies a bounded maximum.
+    uint32 duration_ms = 1;
+}
+
+// One finite, synchronous command in an ActionQueue request.
+message ActionStep {
+    // Client-provided identifier used for diagnostics. It must be unique within
+    // one ActionQueue request.
+    string step_id = 1;
+
+    // Per-step timeout in milliseconds. Zero inherits the remaining action
+    // timeout or the server default.
+    uint32 timeout_ms = 2;
+
+    // Tags are grouped by domain so compatible commands can be added without
+    // renumbering existing alternatives: Arm 10-19, AGV 20-29, built-ins 90+.
+    oneof command {
+        MoveJ.Request arm_move_j = 10;
+        MoveL.Request arm_move_l = 11;
+
+        AgvNavigateToPoseCommand.Request agv_navigate_to_pose = 20;
+        AgvNavigateToStationCommand.Request agv_navigate_to_station = 21;
+        AgvFollowPathCommand.Request agv_follow_path = 22;
+
+        DelayAction delay = 90;
+    }
+}
+
+// Atomically submits a complete command sequence for edge-local serial
+// execution. Device motion alternatives must use synchronous execution.
+message ActionQueueCommand {
+    message Request {
+        // Client-generated globally unique idempotency key. During one Action
+        // service instance, retrying an identical accepted request with the
+        // same action_id does not dispatch its steps a second time. Recent
+        // terminal results can be returned; older accepted IDs are rejected
+        // fail-closed after their result is evicted. Deduplication is not
+        // persisted across an edge-service restart; the required instance
+        // epoch below prevents an old retry from being replayed after restart.
+        string action_id = 1;
+        repeated ActionStep steps = 2;
+
+        // Total queue-wait plus execution timeout in milliseconds. Zero uses a
+        // bounded server default.
+        uint32 total_timeout_ms = 3;
+
+        // Required instance epoch obtained from GetSystemInfo. A mismatch
+        // means the process-local deduplication ledger was recreated, so the
+        // server rejects the request instead of risking a replay.
+        string expected_service_instance_id = 4;
+    }
+
+    message Feedback {
+        CommandHeader.Feedback header = 1;
+        string action_id = 2;
+
+        // Number of steps which completed successfully before the final result.
+        uint32 completed_steps = 3;
+
+        // Present only when a particular step caused failure, cancellation,
+        // timeout, or rejection. Presence distinguishes index zero from no
+        // failed step.
+        optional uint32 failed_step_index = 4;
+        ActionResultCode result = 5;
+        string service_instance_id = 6;
+        ActionDeduplicationStatus deduplication_status = 7;
+    }
+}

protos/cmvr/api/system_service.proto

@@ -13,4 +13,6 @@ service SystemService {
     rpc UpdateParams(UpdateParamsCommand.Request) returns (UpdateParamsCommand.Feedback) {}
 
     rpc StopAll(StopAllCommand.Request) returns (StopAllCommand.Feedback) {}
+
+    rpc ExecuteActionQueue(ActionQueueCommand.Request) returns (ActionQueueCommand.Feedback) {}
 }