diff --git a/server-tools/persona-compute-pool/entry_service.py b/server-tools/persona-compute-pool/entry_service.py
new file mode 100644
index 0000000..f39ca7f
--- /dev/null
+++ b/server-tools/persona-compute-pool/entry_service.py
@@ -0,0 +1,49 @@
+#!/usr/bin/env python3
+"""Loopback-first compute-pool entry with authenticated worker heartbeats."""
+from __future__ import annotations
+import json, os, secrets, threading
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from pathlib import Path
+
+STATE=Path(os.environ.get("GH_POOL_STATE","/var/lib/guanghu-compute-pool/state.json"))
+TOKEN=Path(os.environ.get("GH_POOL_TOKEN","/etc/guanghu-compute-pool/token"))
+NODE=os.environ.get("GH_POOL_NODE","QY-LH-MAIN-PROD-01")
+LOCK=threading.Lock()
+
+def read_state():
+    if STATE.exists(): return json.loads(STATE.read_text())
+    return {"schema":"guanghu.compute-pool-entry-state/v1","node_id":NODE,"role":"LIGHTWEIGHT_ENTERPRISE_COORDINATOR","state":"READY_LOOPBACK_ONLY","workers":{},"leases":{},"authority_granted":False}
+
+def write_state(value):
+    STATE.parent.mkdir(parents=True, exist_ok=True); tmp=STATE.with_suffix('.tmp'); tmp.write_text(json.dumps(value,ensure_ascii=False,indent=2)+'\n'); tmp.replace(STATE)
+
+def authorized(handler):
+    if not TOKEN.exists(): return False
+    expected=TOKEN.read_text().strip(); actual=handler.headers.get('Authorization','')
+    return secrets.compare_digest(actual, 'Bearer '+expected) if expected else False
+
+class Handler(BaseHTTPRequestHandler):
+    def send_json(self, code, value):
+        body=json.dumps(value,ensure_ascii=False).encode(); self.send_response(code); self.send_header('Content-Type','application/json'); self.send_header('Content-Length',str(len(body))); self.end_headers(); self.wfile.write(body)
+    def do_GET(self):
+        if self.path=='/health': self.send_json(200,{'ok':True,'service':'guanghu-compute-pool-entry','node_id':NODE,'authority_granted':False}); return
+        if not authorized(self): self.send_json(401,{'error':'AUTH_REQUIRED'}); return
+        if self.path=='/v1/pool/status': self.send_json(200,read_state()); return
+        self.send_json(404,{'error':'NOT_FOUND'})
+    def do_POST(self):
+        if not authorized(self): self.send_json(401,{'error':'AUTH_REQUIRED'}); return
+        try: payload=json.loads(self.rfile.read(int(self.headers.get('Content-Length','0')) or 0) or b'{}')
+        except Exception: self.send_json(400,{'error':'JSON_INVALID'}); return
+        with LOCK:
+            state=read_state()
+            if self.path=='/v1/worker/heartbeat':
+                node=payload.get('node_id');
+                if not node or node=='JD-FD-PRIMARY': self.send_json(403,{'error':'MOTHER_NODE_FORBIDDEN'}); return
+                state['workers'][node]={**payload,'last_seen':payload.get('observed_at'),'authority_granted':False}; write_state(state); self.send_json(200,{'outcome':'HEARTBEAT_ACCEPTED','node_id':node,'authority_granted':False}); return
+            if self.path=='/v1/lease/request':
+                self.send_json(200,{'outcome':'LEASE_QUEUED','state':'PENDING_DETERMINISTIC_SELECTION','authority_granted':False}); return
+        self.send_json(404,{'error':'NOT_FOUND'})
+    def log_message(self,*args): pass
+
+if __name__=='__main__':
+    bind=os.environ.get('GH_POOL_BIND','127.0.0.1'); port=int(os.environ.get('GH_POOL_PORT','8787')); write_state(read_state()); ThreadingHTTPServer((bind,port),Handler).serve_forever()
diff --git a/server-tools/persona-compute-pool/test_entry_service.py b/server-tools/persona-compute-pool/test_entry_service.py
new file mode 100644
index 0000000..ab3b881
--- /dev/null
+++ b/server-tools/persona-compute-pool/test_entry_service.py
@@ -0,0 +1,14 @@
+#!/usr/bin/env python3
+import importlib.util, json, tempfile, unittest
+from pathlib import Path
+ROOT=Path(__file__).resolve().parents[2]
+spec=importlib.util.spec_from_file_location('entry',ROOT/'server-tools/persona-compute-pool/entry_service.py'); M=importlib.util.module_from_spec(spec); assert spec.loader; spec.loader.exec_module(M)
+
+class EntryServiceTests(unittest.TestCase):
+    def test_mother_node_is_forbidden(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            old=M.STATE; M.STATE=Path(tmp)/'state.json'; M.write_state(M.read_state()); self.assertFalse('JD-FD-PRIMARY' in M.read_state()['workers']); M.STATE=old
+    def test_state_has_authority_false(self):
+        value=M.read_state(); self.assertFalse(value['authority_granted'])
+
+if __name__=='__main__': unittest.main()