diff --git a/pcc/k8s_plugin/__init__.py b/pcc/k8s_plugin/__init__.py
index 148dece..b2bed1f 100644
--- a/pcc/k8s_plugin/__init__.py
+++ b/pcc/k8s_plugin/__init__.py
@@ -174,3 +174,48 @@ async def cluster_status(env, cluster_id):
 
 import json
 from sqlor.dbpools import DBPools
+
+
+async def allocate_unit(env, cluster_id, unit_name, cpu, memory, gpu=0):
+    """K8s 算力单元分配: 创建 namespace + ResourceQuota"""
+    from pcc import ssh_exec as _ssh
+    node = await _find_control_node(env, cluster_id)
+    if not node:
+        return {'status': 'error', 'message': 'No control node found'}
+
+    quota_yaml = (
+        f"apiVersion: v1\\nkind: ResourceQuota\\nmetadata:\\n  name: {unit_name}-quota\\n  namespace: {unit_name}\\n"
+        f"spec:\\n  hard:\\n    requests.cpu: \\\"{cpu}\\\"\\n    requests.memory: \\\"{memory}Gi\\\"\\n    requests.nvidia.com/gpu: \\\"{gpu}\\\""
+    )
+    cmds = [
+        f"kubectl create namespace {unit_name}",
+        f"echo '{quota_yaml}' | kubectl apply -f -",
+    ]
+    for cmd in cmds:
+        rc, out, err = await _ssh(node.ip_address, node.ssh_port or 22, node.ssh_user or 'root', cmd)
+        if rc != 0:
+            return {'status': 'error', 'message': f'kubectl failed: {err[:200]}'}
+    return {'status': 'ok', 'message': f'Unit {unit_name}: ns + {cpu}C/{memory}G/{gpu}GPU'}
+
+
+async def release_unit(env, cluster_id, unit_name):
+    """K8s 算力单元回收: 删除 namespace"""
+    node = await _find_control_node(env, cluster_id)
+    if not node:
+        return {'status': 'error', 'message': 'No control node found'}
+    from pcc import ssh_exec as _ssh
+    rc, _, err = await _ssh(node.ip_address, node.ssh_port or 22, node.ssh_user or 'root',
+                              f"kubectl delete namespace {unit_name} --timeout=60s")
+    if rc != 0:
+        return {'status': 'error', 'message': f'delete ns failed: {err[:200]}'}
+    return {'status': 'ok', 'message': f'Unit {unit_name} released'}
+
+
+async def _find_control_node(env, cluster_id):
+    from sqlor.dbpools import DBPools
+    dbname = env.get_module_dbname('pccs')
+    async with DBPools().sqlorContext(dbname) as sor:
+        recs = await sor.R('cluster_node', {'cluster_id': cluster_id, 'role': 'control', 'status': 'active'})
+    if not recs:
+        return None
+    return await _node_info(env, recs[0].node_id)
diff --git a/pcc/ray_plugin/__init__.py b/pcc/ray_plugin/__init__.py
index 972f976..e0bae45 100644
--- a/pcc/ray_plugin/__init__.py
+++ b/pcc/ray_plugin/__init__.py
@@ -132,3 +132,43 @@ async def cluster_status(env, cluster_id):
 
 import json
 from sqlor.dbpools import DBPools
+
+
+async def allocate_unit(env, cluster_id, unit_name, cpu, memory, gpu=0):
+    """Ray 算力单元分配: 创建 detached actor / placement group 绑定资源标签"""
+    node = await _find_control_node(env, cluster_id)
+    if not node:
+        return {'status': 'error', 'message': 'No control node found'}
+    from pcc import ssh_exec as _ssh
+    # 用 ray submit 提交一个资源占位 job
+    script = (
+        "import ray; ray.init(address='auto'); "
+        "pg = ray.util.placement_group([{'CPU': " + str(cpu) + ", 'GPU': " + str(gpu) + "}], strategy='STRICT_SPREAD', name='" + unit_name + "'); "
+        "ray.get(pg.ready()); print('PG ready: " + unit_name + "')"
+    )
+    py_cmd = f"python3 -c '{script}'"
+    rc, out, err = await _ssh(node.ip_address, node.ssh_port or 22, node.ssh_user or 'root', py_cmd, timeout=60)
+    if rc != 0:
+        return {'status': 'error', 'message': f'ray pg failed: {err[:200]}'}
+    return {'status': 'ok', 'message': f'PlacementGroup {unit_name}: {cpu}C/{gpu}GPU'}
+
+
+async def release_unit(env, cluster_id, unit_name):
+    """Ray 算力单元回收: 删除 placement group"""
+    node = await _find_control_node(env, cluster_id)
+    if not node:
+        return {'status': 'error', 'message': 'No control node found'}
+    from pcc import ssh_exec as _ssh
+    py_cmd = f"python3 -c 'import ray; ray.init(address=\"auto\"); ray.util.remove_placement_group(ray.util.get_placement_group(\"{unit_name}\")); print(\"removed\")'"
+    await _ssh(node.ip_address, node.ssh_port or 22, node.ssh_user or 'root', py_cmd, timeout=30)
+    return {'status': 'ok', 'message': f'PlacementGroup {unit_name} released'}
+
+
+async def _find_control_node(env, cluster_id):
+    from sqlor.dbpools import DBPools
+    dbname = env.get_module_dbname('pccs')
+    async with DBPools().sqlorContext(dbname) as sor:
+        recs = await sor.R('cluster_node', {'cluster_id': cluster_id, 'role': 'control', 'status': 'active'})
+    if not recs:
+        return None
+    return await _node_info(env, recs[0].node_id)
diff --git a/pcc/slurm_plugin/__init__.py b/pcc/slurm_plugin/__init__.py
index ae487ad..989c5a5 100644
--- a/pcc/slurm_plugin/__init__.py
+++ b/pcc/slurm_plugin/__init__.py
@@ -176,3 +176,54 @@ async def cluster_status(env, cluster_id):
 
 import json
 from sqlor.dbpools import DBPools
+
+
+async def allocate_unit(env, cluster_id, unit_name, cpu, memory, gpu=0):
+    """Slurm 算力单元分配: 创建 partition + 关联节点"""
+    node = await _find_control_node(env, cluster_id)
+    if not node:
+        return {'status': 'error', 'message': 'No control node found'}
+    from pcc import ssh_exec as _ssh
+    # 创建 partition
+    cmd = f"scontrol create PartitionName={unit_name} MaxNodes=UNLIMITED Default=NO MaxTime=UNLIMITED State=UP"
+    rc, out, err = await _ssh(node.ip_address, node.ssh_port or 22, node.ssh_user or 'root', cmd)
+    if rc != 0:
+        return {'status': 'error', 'message': f'scontrol failed: {err[:200]}'}
+
+    # 分配 node 到 partition (通过 cluster_node 表找该集群下可用的 compute 节点)
+    dbname = env.get_module_dbname('pccs')
+    async with DBPools().sqlorContext(dbname) as sor:
+        assigned = await sor.sqlExe(
+            """SELECT cn.name FROM cluster_node cnn
+               JOIN compute_node cn ON cn.id = cnn.node_id
+               WHERE cnn.cluster_id=${cid}$ AND cnn.role='compute' AND cnn.status='active'
+               LIMIT 1""",
+            {'cid': cluster_id}
+        )
+        if assigned:
+            node_name = assigned[0].name
+            await _ssh(node.ip_address, node.ssh_port or 22, node.ssh_user or 'root',
+                       f"scontrol update NodeName={node_name} Partition={unit_name}")
+
+    return {'status': 'ok', 'message': f'Partition {unit_name} created'}
+
+
+async def release_unit(env, cluster_id, unit_name):
+    """Slurm 算力单元回收: 删除 partition"""
+    node = await _find_control_node(env, cluster_id)
+    if not node:
+        return {'status': 'error', 'message': 'No control node found'}
+    from pcc import ssh_exec as _ssh
+    await _ssh(node.ip_address, node.ssh_port or 22, node.ssh_user or 'root',
+               f"scontrol delete PartitionName={unit_name}")
+    return {'status': 'ok', 'message': f'Partition {unit_name} deleted'}
+
+
+async def _find_control_node(env, cluster_id):
+    from sqlor.dbpools import DBPools
+    dbname = env.get_module_dbname('pccs')
+    async with DBPools().sqlorContext(dbname) as sor:
+        recs = await sor.R('cluster_node', {'cluster_id': cluster_id, 'role': 'control', 'status': 'active'})
+    if not recs:
+        return None
+    return await _node_info(env, recs[0].node_id)
diff --git a/wwwroot/api/unit_allocate.dspy b/wwwroot/api/unit_allocate.dspy
new file mode 100644
index 0000000..2a03ffa
--- /dev/null
+++ b/wwwroot/api/unit_allocate.dspy
@@ -0,0 +1,22 @@
+# 算力单元分配 (集群内部资源: K8s ns+quota / Slurm partition / Ray PlacementGroup)
+# params: cluster_id, unit_name, cpu, memory, gpu(可选)
+cluster_id = params_kw.get('cluster_id', '')
+unit_name = params_kw.get('unit_name', '')
+cpu = int(params_kw.get('cpu', 1))
+memory = int(params_kw.get('memory', 4))
+gpu = int(params_kw.get('gpu', 0))
+if not cluster_id or not unit_name:
+    return {'widgettype': 'Error', 'options': {'title': '失败', 'message': 'Missing cluster_id or unit_name'}}
+
+env = request._run_ns
+dbname = env.get_module_dbname('pccs')
+async with DBPools().sqlorContext(dbname) as sor:
+    recs = await sor.R('cluster', {'id': cluster_id})
+if not recs:
+    return {'widgettype': 'Error', 'options': {'title': '失败', 'message': 'Cluster not found'}}
+
+plugin = _get_plugin(recs[0].cluster_type)
+result = await plugin.allocate_unit(env, cluster_id, unit_name, cpu, memory, gpu)
+if result.get('status') == 'ok':
+    return {'widgettype': 'Message', 'options': {'title': '分配成功', 'message': result['message'], 'type': 'success', 'timeout': 5}}
+return {'widgettype': 'Error', 'options': {'title': '分配失败', 'message': result.get('message', ''), 'cwidth': 16, 'cheight': 9, 'timeout': 3}}
diff --git a/wwwroot/api/unit_release.dspy b/wwwroot/api/unit_release.dspy
new file mode 100644
index 0000000..2f4a357
--- /dev/null
+++ b/wwwroot/api/unit_release.dspy
@@ -0,0 +1,19 @@
+# 算力单元回收
+# params: cluster_id, unit_name
+cluster_id = params_kw.get('cluster_id', '')
+unit_name = params_kw.get('unit_name', '')
+if not cluster_id or not unit_name:
+    return {'widgettype': 'Error', 'options': {'title': '失败', 'message': 'Missing cluster_id or unit_name'}}
+
+env = request._run_ns
+dbname = env.get_module_dbname('pccs')
+async with DBPools().sqlorContext(dbname) as sor:
+    recs = await sor.R('cluster', {'id': cluster_id})
+if not recs:
+    return {'widgettype': 'Error', 'options': {'title': '失败', 'message': 'Cluster not found'}}
+
+plugin = _get_plugin(recs[0].cluster_type)
+result = await plugin.release_unit(env, cluster_id, unit_name)
+if result.get('status') == 'ok':
+    return {'widgettype': 'Message', 'options': {'title': '回收成功', 'message': result['message'], 'type': 'success', 'timeout': 5}}
+return {'widgettype': 'Error', 'options': {'title': '回收失败', 'message': result.get('message', ''), 'cwidth': 16, 'cheight': 9, 'timeout': 3}}