88"""
99from __future__ import annotations
1010import json
11+ import functools
12+ import hashlib
1113import os
1214from pathlib import Path
1315import re
16+ import stat
17+ import subprocess
1418import urllib .error
1519import urllib .request
1620
1923MAX_BODY = 2_000_000
2024# These are native inference server identities, not an installation mechanism.
2125NATIVE_EXECUTABLES = {"llama" : {"llama-server" }, "skippy" : {"skippy-server" , "skippy" }}
26+ BUZZ_EXECUTABLE = Path ("/usr/bin/buzz-desktop" )
27+ BUZZ_MANIFEST = Path ("/var/lib/dpkg/info/buzz.md5sums" )
28+ BUZZ_VERSION = "0.5.17"
2229
2330class ComputeError (ValueError ):
2431 pass
@@ -45,6 +52,41 @@ def request(url, payload=None, timeout=3):
4552def valid_name (value ):
4653 return isinstance (value , str ) and 0 < len (value .strip ()) <= 256 and not any (ord (char ) < 32 for char in value )
4754
55+ def protected_file (path ):
56+ metadata = path .lstat ()
57+ return stat .S_ISREG (metadata .st_mode ) and metadata .st_uid == 0 and not metadata .st_mode & 0o022
58+
59+ @functools .lru_cache (maxsize = 2 )
60+ def file_hashes (path , identity ):
61+ # Cache only within this controller process, keyed by inode/mtime/ctime/size.
62+ md5 , sha = hashlib .md5 (), hashlib .sha256 ()
63+ with Path (path ).open ("rb" ) as stream :
64+ while block := stream .read (1024 * 1024 ):
65+ md5 .update (block )
66+ sha .update (block )
67+ return md5 .hexdigest (), sha .hexdigest ()
68+
69+ def buzz_binary_proof (binary ):
70+ if binary != BUZZ_EXECUTABLE or not protected_file (binary ) or not protected_file (BUZZ_MANIFEST ):
71+ raise ComputeError ("Embedded Buzz inference requires the protected system package executable" )
72+ try :
73+ package = subprocess .run (["/usr/bin/dpkg-query" , "-W" , "-f=${Version} ${db:Status-Abbrev}" , "buzz" ], capture_output = True , text = True , timeout = 3 )
74+ except subprocess .SubprocessError as exc :
75+ raise ComputeError ("Could not verify the embedded Buzz package: " + str (exc ))
76+ if package .returncode or package .stdout .strip () != BUZZ_VERSION + " ii" :
77+ raise ComputeError ("Embedded native inference requires the verified Buzz " + BUZZ_VERSION + " package" )
78+ entries = [line .split (None , 1 ) for line in BUZZ_MANIFEST .read_text ().splitlines () if line .strip ()]
79+ expected = [checksum for checksum , name in entries if name == "usr/bin/buzz-desktop" ]
80+ if len (expected ) != 1 or not re .fullmatch (r"[a-f0-9]{32}" , expected [0 ]):
81+ raise ComputeError ("Buzz executable is missing from its protected package manifest" )
82+ before = binary .stat ()
83+ identity = (before .st_dev , before .st_ino , before .st_size , before .st_mtime_ns , before .st_ctime_ns )
84+ md5 , sha = file_hashes (str (binary ), identity )
85+ after = binary .stat ()
86+ if identity != (after .st_dev , after .st_ino , after .st_size , after .st_mtime_ns , after .st_ctime_ns ) or md5 != expected [0 ]:
87+ raise ComputeError ("Buzz executable does not match the installed package manifest" )
88+ return {"package" : "buzz" , "version" : BUZZ_VERSION , "sha256" : sha , "integrity" : "root-owned pinned package and matching dpkg file manifest" , "mode" : "embedded-native-skippy" }
89+
4890def process_proof (record , proc_root = Path ("/proc" )):
4991 if not isinstance (record , dict ) or not valid_name (record .get ("name" )):
5092 raise ComputeError ("Invalid native process record" )
@@ -58,7 +100,10 @@ def process_proof(record, proc_root=Path("/proc")):
58100 if proc .stat ().st_uid != os .getuid ():
59101 raise ComputeError ("Native model process belongs to a different user" )
60102 binary = Path (os .readlink (proc / "exe" ))
61- if binary .name not in NATIVE_EXECUTABLES [backend ]:
103+ package_proof = None
104+ if backend == "skippy" and binary == BUZZ_EXECUTABLE :
105+ package_proof = buzz_binary_proof (binary )
106+ elif binary .name not in NATIVE_EXECUTABLES [backend ]:
62107 raise ComputeError ("Model process executable does not match its native backend" )
63108 # Field 22 is process start time. The command can contain spaces/parens.
64109 started = (proc / "stat" ).read_text ().rsplit (")" , 1 )[1 ].split ()[19 ]
@@ -80,13 +125,15 @@ def process_proof(record, proc_root=Path("/proc")):
80125 raise ComputeError ("Native process does not own the advertised loopback listening port" )
81126 except OSError as exc :
82127 raise ComputeError ("Cannot prove native model process/socket ownership: " + str (exc ))
83- return {"name" : record ["name" ].strip (), "pid" : pid , "port" : port , "backend" : backend , "instance_id" : record .get ("instance_id" ), "process_start" : started , "executable" : str (binary ), "local_only_verified" : True , "endpoint" : f"http://127.0.0.1:{ port } , "proof" : "Buzz native process inventory + same-user executable + owned loopback socket" }
128+ return {"name" : record ["name" ].strip (), "pid" : pid , "port" : port , "backend" : backend , "instance_id" : record .get ("instance_id" ), "process_start" : started , "executable" : str (binary ), "package_identity" : package_proof , " local_only_verified"True , "endpoint" : f"http://127.0.0.1:{ port } , "proof" : "Buzz native process inventory + same-user executable + owned loopback socket" }
84129
85130def local_models ():
86131 try :
87132 records = request (MANAGEMENT + "/api/runtime/processes" ).get ("processes" )
88133 if not isinstance (records , list ):
89134 return []
135+ if not no_distributed_stages ():
136+ return []
90137 models = []
91138 for record in records [:64 ]:
92139 try :
@@ -98,6 +145,12 @@ def local_models():
98145 except (OSError , ValueError ):
99146 return []
100147
148+ def no_distributed_stages ():
149+ # Pinned MeshLLM exposes stage deployment separately from local processes.
150+ # Be conservative when any pipeline is present, even beside a local model.
151+ state = request (MANAGEMENT + "/api/runtime/stages" )
152+ return state .get ("stages" ) == [] and state .get ("topologies" ) == []
153+
101154def shared_models ():
102155 try :
103156 records = request (ROUTER + "/v1/models" ).get ("data" )
@@ -123,11 +176,26 @@ def target(model, allow_network=False):
123176def complete (payload , allow_network = False , timeout = 180 ):
124177 selected = target (payload .get ("model" , "" ), allow_network )
125178 payload = dict (payload , model = selected ["name" ])
179+ if selected ["local_only_verified" ]:
180+ # Native Skippy also supports automatic peer-consultation hooks. The
181+ # pinned vendor request switch must disable every hook for local work.
182+ payload ["mesh_hooks" ] = False
183+ selected ["mesh_hooks_enabled" ] = False
184+ if selected .get ("backend" ) == "skippy" :
185+ # Pinned Skippy hides unfinished reasoning; a short task can exhaust
186+ # its budget with no answer unless the supported template switch
187+ # explicitly requests direct output.
188+ payload ["reasoning_effort" ] = "none"
189+ payload ["chat_template_kwargs" ] = dict (payload .get ("chat_template_kwargs" ) or {}, enable_thinking = False )
190+ selected ["reasoning_mode" ] = "disabled for bounded artifact generation"
126191 response = request (selected ["endpoint" ] + "/v1/chat/completions" , payload , timeout = timeout )
127192 if selected ["local_only_verified" ]:
128193 # Detect a disappeared/replaced process before reporting a verified run.
129194 verified = process_proof (dict (selected , status = "ready" ))
130195 if verified ["process_start" ] != selected ["process_start" ]:
131196 raise ComputeError ("Native model process changed while inference was running" )
197+ if not no_distributed_stages ():
198+ raise ComputeError ("Distributed model stages appeared during native inference; locality cannot be verified" )
199+ selected ["distributed_stages" ] = "No stage or topology deployments observed before and after inference"
132200 response ["shadowfetch_compute" ] = selected
133201 return response
0 commit comments